/*
    * Licensed to the Apache Software Foundation (ASF) under one
    * or more contributor license agreements.  See the NOTICE file
    * distributed with this work for additional information
    * regarding copyright ownership.  The ASF licenses this file
    * to you under the Apache License, Version 2.0 (the
    * "License"); you may not use this file except in compliance
    * with the License.  You may obtain a copy of the License at
    *
   *     http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  
  package org.apache.hbase.codec.prefixtree;
  
  import java.nio.ByteBuffer;
  
  import org.apache.hadoop.classification.InterfaceAudience;
  import org.apache.hadoop.hbase.KeyValue;
  import org.apache.hadoop.hbase.KeyValueTool;
  import org.apache.hadoop.hbase.io.encoding.DataBlockEncoder.EncodedSeeker;
  import org.apache.hbase.Cell;
  import org.apache.hbase.cell.CellScannerPosition;
  import org.apache.hbase.cell.CellTool;
  import org.apache.hbase.codec.prefixtree.decode.DecoderFactory;
  import org.apache.hbase.codec.prefixtree.decode.PrefixTreeArraySearcher;
  
  /**
   * These methods have the same definition as any implementation of the EncodedSeeker.
   *
   * In the future, the EncodedSeeker could be modified to work with the Cell interface directly.  It
   * currently returns a new KeyValue object each time getKeyValue is called.  This is not horrible,
   * but in order to create a new KeyValue object, we must first allocate a new byte[] and copy in
   * the data from the PrefixTreeCell.  It is somewhat heavyweight right now.
   */
  @InterfaceAudience.Private
  public class PrefixTreeSeeker implements EncodedSeeker {
  
    protected ByteBuffer block;
    protected boolean includeMvccVersion;
    protected PrefixTreeArraySearcher ptSearcher;
  
    public PrefixTreeSeeker(boolean includeMvccVersion) {
      this.includeMvccVersion = includeMvccVersion;
    }
  
    @Override
    public void setCurrentBuffer(ByteBuffer fullBlockBuffer) {
      block = fullBlockBuffer;
      ptSearcher = DecoderFactory.checkOut(block, includeMvccVersion);
      rewind();
    }
  
    /**
     * Currently unused.
     * <p/>
     * TODO performance leak. should reuse the searchers. hbase does not currently have a hook where
     * this can be called
     */
    public void releaseCurrentSearcher(){
      DecoderFactory.checkIn(ptSearcher);
    }
  
  
    @Override
    public ByteBuffer getKeyDeepCopy() {
      return KeyValueTool.copyKeyToNewByteBuffer(ptSearcher.getCurrent());
    }
  
  
    @Override
    public ByteBuffer getValueShallowCopy() {
      return CellTool.getValueBufferShallowCopy(ptSearcher.getCurrent());
    }
  
    /**
     * currently must do deep copy into new array
     */
    @Override
    public ByteBuffer getKeyValueBuffer() {
      return KeyValueTool.copyToNewByteBuffer(ptSearcher.getCurrent());
    }
  
    /**
     * currently must do deep copy into new array
     */
    @Override
    public KeyValue getKeyValue() {
      return KeyValueTool.copyToNewKeyValue(ptSearcher.getCurrent());
    }
  
    /**
     * Currently unused.
     * <p/>
    * A nice, lightweight reference, though the underlying cell is transient.  This method may return
    * the same reference to the backing PrefixTreeCell repeatedly, while other implementations may
    * return a different reference for each Cell.
    * <p/>
    * The goal will be to transition the upper layers of HBase, like Filters and KeyValueHeap, to use
    * this method instead of the getKeyValue() methods above.
    */
 //  @Override
   public Cell getCurrent() {
     return ptSearcher.getCurrent();
   }
 
   @Override
   public void rewind() {
     ptSearcher.positionAtFirstCell();
   }
 
   @Override
   public boolean next() {
     return ptSearcher.next();
   }
 
 //  @Override
   public boolean advance() {
     return ptSearcher.next();
   }
 
 
   private static final boolean USE_POSITION_BEFORE = false;
 
   /**
    * Seek forward only (should be called reseekToKeyInBlock?).
    * <p/>
    * If the exact key is found look at the seekBefore variable and:<br/>
    * - if true: go to the previous key if it's true<br/>
    * - if false: stay on the exact key
    * <p/>
    * If the exact key is not found, then go to the previous key *if possible*, but remember to leave
    * the scanner in a valid state if possible.
    * <p/>
    * @param keyOnlyBytes KeyValue format of a Cell's key at which to position the seeker
    * @param offset offset into the keyOnlyBytes array
    * @param length number of bytes of the keyOnlyBytes array to use
    * @param forceBeforeOnExactMatch if an exact match is found and seekBefore=true, back up one Cell
    * @return 0 if the seeker is on the exact key<br/>
    *         1 if the seeker is not on the key for any reason, including seekBefore being true
    */
   @Override
   public int seekToKeyInBlock(byte[] keyOnlyBytes, int offset, int length,
       boolean forceBeforeOnExactMatch) {
     if (USE_POSITION_BEFORE) {
       return seekToOrBeforeUsingPositionAtOrBefore(keyOnlyBytes, offset, length,
         forceBeforeOnExactMatch);
     }else{
       return seekToOrBeforeUsingPositionAtOrAfter(keyOnlyBytes, offset, length,
         forceBeforeOnExactMatch);
     }
   }
 
 
 
   /*
    * Support both of these options since the underlying PrefixTree supports both.  Possibly
    * expand the EncodedSeeker to utilize them both.
    */
 
   protected int seekToOrBeforeUsingPositionAtOrBefore(byte[] keyOnlyBytes, int offset, int length,
       boolean forceBeforeOnExactMatch){
     // this does a deep copy of the key byte[] because the CellSearcher interface wants a Cell
     KeyValue kv = KeyValue.createKeyValueFromKey(keyOnlyBytes, offset, length);
 
     CellScannerPosition position = ptSearcher.seekForwardToOrBefore(kv);
 
     if(CellScannerPosition.AT == position){
       if (forceBeforeOnExactMatch) {
         ptSearcher.previous();
         return 1;
       }
       return 0;
     }
 
     return 1;
   }
 
 
   protected int seekToOrBeforeUsingPositionAtOrAfter(byte[] keyOnlyBytes, int offset, int length,
       boolean forceBeforeOnExactMatch){
     // this does a deep copy of the key byte[] because the CellSearcher interface wants a Cell
     KeyValue kv = KeyValue.createKeyValueFromKey(keyOnlyBytes, offset, length);
 
     //should probably switch this to use the seekForwardToOrBefore method
     CellScannerPosition position = ptSearcher.seekForwardToOrAfter(kv);
 
     if(CellScannerPosition.AT == position){
       if (forceBeforeOnExactMatch) {
         ptSearcher.previous();
         return 1;
       }
       return 0;
 
     }
 
     if(CellScannerPosition.AFTER == position){
       if(!ptSearcher.isBeforeFirst()){
         ptSearcher.previous();
       }
       return 1;
     }
 
     if(position == CellScannerPosition.AFTER_LAST){
       return 1;
     }
 
     throw new RuntimeException("unexpected CellScannerPosition:"+position);
   }
 
 }