/*
    * 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.hadoop.hbase.codec.prefixtree.decode;
  
  import org.apache.hadoop.hbase.classification.InterfaceAudience;
  import org.apache.hadoop.hbase.Cell;
  import org.apache.hadoop.hbase.CellComparator;
  import org.apache.hadoop.hbase.CellScanner;
  import org.apache.hadoop.hbase.codec.prefixtree.PrefixTreeBlockMeta;
  import org.apache.hadoop.hbase.codec.prefixtree.decode.column.ColumnReader;
  import org.apache.hadoop.hbase.codec.prefixtree.decode.row.RowNodeReader;
  import org.apache.hadoop.hbase.codec.prefixtree.decode.timestamp.MvccVersionDecoder;
  import org.apache.hadoop.hbase.codec.prefixtree.decode.timestamp.TimestampDecoder;
  import org.apache.hadoop.hbase.codec.prefixtree.encode.other.ColumnNodeType;
  
  /**
   * Extends PtCell and manipulates its protected fields.  Could alternatively contain a PtCell and
   * call get/set methods.
   *
   * This is an "Array" scanner to distinguish from a future "ByteBuffer" scanner.  This
   * implementation requires that the bytes be in a normal java byte[] for performance.  The
   * alternative ByteBuffer implementation would allow for accessing data in an off-heap ByteBuffer
   * without copying the whole buffer on-heap.
   */
  @InterfaceAudience.Private
  public class PrefixTreeArrayScanner extends PrefixTreeCell implements CellScanner {
  
    /***************** fields ********************************/
  
    protected PrefixTreeBlockMeta blockMeta;
  
    protected boolean beforeFirst;
    protected boolean afterLast;
  
    protected RowNodeReader[] rowNodes;
    protected int rowNodeStackIndex;
  
    protected RowNodeReader currentRowNode;
    protected ColumnReader familyReader;
    protected ColumnReader qualifierReader;
    protected ColumnReader tagsReader;
    protected TimestampDecoder timestampDecoder;
    protected MvccVersionDecoder mvccVersionDecoder;
  
    protected boolean nubCellsRemain;
    protected int currentCellIndex;
  
  
    /*********************** construct ******************************/
  
    // pass in blockMeta so we can initialize buffers big enough for all cells in the block
    public PrefixTreeArrayScanner(PrefixTreeBlockMeta blockMeta, int rowTreeDepth,
        int rowBufferLength, int qualifierBufferLength, int tagsBufferLength) {
      this.rowNodes = new RowNodeReader[rowTreeDepth];
      for (int i = 0; i < rowNodes.length; ++i) {
        rowNodes[i] = new RowNodeReader();
      }
      this.rowBuffer = new byte[rowBufferLength];
      this.familyBuffer = new byte[PrefixTreeBlockMeta.MAX_FAMILY_LENGTH];
      this.familyReader = new ColumnReader(familyBuffer, ColumnNodeType.FAMILY);
      this.qualifierBuffer = new byte[qualifierBufferLength];
      this.tagsBuffer = new byte[tagsBufferLength];
      this.qualifierReader = new ColumnReader(qualifierBuffer, ColumnNodeType.QUALIFIER);
      this.tagsReader = new ColumnReader(tagsBuffer, ColumnNodeType.TAGS);
      this.timestampDecoder = new TimestampDecoder();
      this.mvccVersionDecoder = new MvccVersionDecoder();
    }
  
  
    /**************** init helpers ***************************************/
  
    /**
     * Call when first accessing a block.
     * @return entirely new scanner if false
     */
    public boolean areBuffersBigEnough() {
      if (rowNodes.length < blockMeta.getRowTreeDepth()) {
        return false;
      }
      if (rowBuffer.length < blockMeta.getMaxRowLength()) {
        return false;
      }
      if (qualifierBuffer.length < blockMeta.getMaxQualifierLength()) {
       return false;
     }
     if(tagsBuffer.length < blockMeta.getMaxTagsLength()) {
       return false;
     }
     return true;
   }
 
   public void initOnBlock(PrefixTreeBlockMeta blockMeta, byte[] block,
       boolean includeMvccVersion) {
     this.block = block;
     this.blockMeta = blockMeta;
     this.familyOffset = familyBuffer.length;
     this.familyReader.initOnBlock(blockMeta, block);
     this.qualifierOffset = qualifierBuffer.length;
     this.qualifierReader.initOnBlock(blockMeta, block);
     this.tagsOffset = tagsBuffer.length;
     this.tagsReader.initOnBlock(blockMeta, block);
     this.timestampDecoder.initOnBlock(blockMeta, block);
     this.mvccVersionDecoder.initOnBlock(blockMeta, block);
     this.includeMvccVersion = includeMvccVersion;
     resetToBeforeFirstEntry();
   }
 
   // Does this have to be in the CellScanner Interface?  TODO
   public void resetToBeforeFirstEntry() {
     beforeFirst = true;
     afterLast = false;
     rowNodeStackIndex = -1;
     currentRowNode = null;
     rowLength = 0;
     familyOffset = familyBuffer.length;
     familyLength = 0;
     qualifierOffset = blockMeta.getMaxQualifierLength();
     qualifierLength = 0;
     nubCellsRemain = false;
     currentCellIndex = -1;
     timestamp = -1L;
     type = DEFAULT_TYPE;
     absoluteValueOffset = 0;//use 0 vs -1 so the cell is valid when value hasn't been initialized
     valueLength = 0;// had it at -1, but that causes null Cell to add up to the wrong length
     tagsOffset = blockMeta.getMaxTagsLength();
     tagsLength = 0;
   }
 
   /**
    * Call this before putting the scanner back into a pool so it doesn't hold the last used block
    * in memory.
    */
   public void releaseBlockReference(){
     block = null;
   }
 
 
   /********************** CellScanner **********************/
 
   @Override
   public Cell current() {
     if(isOutOfBounds()){
       return null;
     }
     return (Cell)this;
   }
 
   /******************* Object methods ************************/
 
   @Override
   public boolean equals(Object obj) {
     //trivial override to confirm intent (findbugs)
     return super.equals(obj);
   }
 
   @Override
   public int hashCode() {
     return super.hashCode();
   }
 
   /**
    * Override PrefixTreeCell.toString() with a check to see if the current cell is valid.
    */
   @Override
   public String toString() {
     Cell currentCell = current();
     if(currentCell==null){
       return "null";
     }
     return ((PrefixTreeCell)currentCell).getKeyValueString();
   }
 
 
   /******************* advance ***************************/
 
   public boolean positionAtFirstCell() {
     reInitFirstNode();
     return advance();
   }
 
   @Override
   public boolean advance() {
     if (afterLast) {
       return false;
     }
     if (!hasOccurrences()) {
       resetToBeforeFirstEntry();
     }
     if (beforeFirst || isLastCellInRow()) {
       nextRow();
       if (afterLast) {
         return false;
       }
     } else {
       ++currentCellIndex;
     }
 
     populateNonRowFields(currentCellIndex);
     return true;
   }
 
 
   public boolean nextRow() {
     nextRowInternal();
     if (afterLast) {
       return false;
     }
     populateNonRowFields(currentCellIndex);
     return true;
   }
 
 
   /**
    * This method is safe to call when the scanner is not on a fully valid row node, as in the case
    * of a row token miss in the Searcher
    * @return true if we are positioned on a valid row, false if past end of block
    */
   protected boolean nextRowInternal() {
     if (afterLast) {
       return false;
     }
     if (beforeFirst) {
       initFirstNode();
       if (currentRowNode.hasOccurrences()) {
         if (currentRowNode.isNub()) {
           nubCellsRemain = true;
         }
         currentCellIndex = 0;
         return true;
       }
     }
     if (currentRowNode.isLeaf()) {
       discardCurrentRowNode(true);
     }
     while (!afterLast) {
       if (nubCellsRemain) {
         nubCellsRemain = false;
       }
       if (currentRowNode.hasMoreFanNodes()) {
         followNextFan();
         if (currentRowNode.hasOccurrences()) {
           // found some values
           currentCellIndex = 0;
           return true;
         }
       } else {
         discardCurrentRowNode(true);
       }
     }
     return false;// went past the end
   }
 
 
   /**************** secondary traversal methods ******************************/
 
   protected void reInitFirstNode() {
     resetToBeforeFirstEntry();
     initFirstNode();
   }
 
   protected void initFirstNode() {
     int offsetIntoUnderlyingStructure = blockMeta.getAbsoluteRowOffset();
     rowNodeStackIndex = 0;
     currentRowNode = rowNodes[0];
     currentRowNode.initOnBlock(blockMeta, block, offsetIntoUnderlyingStructure);
     appendCurrentTokenToRowBuffer();
     beforeFirst = false;
   }
 
   protected void followFirstFan() {
     followFan(0);
   }
 
   protected void followPreviousFan() {
     int nextFanPosition = currentRowNode.getFanIndex() - 1;
     followFan(nextFanPosition);
   }
 
   protected void followCurrentFan() {
     int currentFanPosition = currentRowNode.getFanIndex();
     followFan(currentFanPosition);
   }
 
   protected void followNextFan() {
     int nextFanPosition = currentRowNode.getFanIndex() + 1;
     followFan(nextFanPosition);
   }
 
   protected void followLastFan() {
     followFan(currentRowNode.getLastFanIndex());
   }
 
   protected void followFan(int fanIndex) {
     currentRowNode.setFanIndex(fanIndex);
     appendToRowBuffer(currentRowNode.getFanByte(fanIndex));
 
     int nextOffsetIntoUnderlyingStructure = currentRowNode.getOffset()
         + currentRowNode.getNextNodeOffset(fanIndex, blockMeta);
     ++rowNodeStackIndex;
 
     currentRowNode = rowNodes[rowNodeStackIndex];
     currentRowNode.initOnBlock(blockMeta, block, nextOffsetIntoUnderlyingStructure);
 
     //TODO getToken is spewing garbage
     appendCurrentTokenToRowBuffer();
     if (currentRowNode.isNub()) {
       nubCellsRemain = true;
     }
     currentCellIndex = 0;
   }
 
   /**
    * @param forwards which marker to set if we overflow
    */
   protected void discardCurrentRowNode(boolean forwards) {
     RowNodeReader rowNodeBeingPopped = currentRowNode;
     --rowNodeStackIndex;// pop it off the stack
     if (rowNodeStackIndex < 0) {
       currentRowNode = null;
       if (forwards) {
         markAfterLast();
       } else {
         markBeforeFirst();
       }
       return;
     }
     popFromRowBuffer(rowNodeBeingPopped);
     currentRowNode = rowNodes[rowNodeStackIndex];
   }
 
   protected void markBeforeFirst() {
     beforeFirst = true;
     afterLast = false;
     currentRowNode = null;
   }
 
   protected void markAfterLast() {
     beforeFirst = false;
     afterLast = true;
     currentRowNode = null;
   }
 
 
   /***************** helper methods **************************/
 
   protected void appendCurrentTokenToRowBuffer() {
     System.arraycopy(block, currentRowNode.getTokenArrayOffset(), rowBuffer, rowLength,
       currentRowNode.getTokenLength());
     rowLength += currentRowNode.getTokenLength();
   }
 
   protected void appendToRowBuffer(byte b) {
     rowBuffer[rowLength] = b;
     ++rowLength;
   }
 
   protected void popFromRowBuffer(RowNodeReader rowNodeBeingPopped) {
     rowLength -= rowNodeBeingPopped.getTokenLength();
     --rowLength; // pop the parent's fan byte
   }
 
   protected boolean hasOccurrences() {
     return currentRowNode != null && currentRowNode.hasOccurrences();
   }
 
   protected boolean isBranch() {
     return currentRowNode != null && !currentRowNode.hasOccurrences()
         && currentRowNode.hasChildren();
   }
 
   protected boolean isNub() {
     return currentRowNode != null && currentRowNode.hasOccurrences()
         && currentRowNode.hasChildren();
   }
 
   protected boolean isLeaf() {
     return currentRowNode != null && currentRowNode.hasOccurrences()
         && !currentRowNode.hasChildren();
   }
 
   //TODO expose this in a PrefixTreeScanner interface
   public boolean isBeforeFirst(){
     return beforeFirst;
   }
 
   public boolean isAfterLast(){
     return afterLast;
   }
 
   protected boolean isOutOfBounds(){
     return beforeFirst || afterLast;
   }
 
   protected boolean isFirstCellInRow() {
     return currentCellIndex == 0;
   }
 
   protected boolean isLastCellInRow() {
     return currentCellIndex == currentRowNode.getLastCellIndex();
   }
 
 
   /********************* fill in family/qualifier/ts/type/value ************/
 
   protected int populateNonRowFieldsAndCompareTo(int cellNum, Cell key) {
     populateNonRowFields(cellNum);
     return CellComparator.compare(this, key, true);
   }
 
   protected void populateFirstNonRowFields() {
     populateNonRowFields(0);
   }
 
   protected void populatePreviousNonRowFields() {
     populateNonRowFields(currentCellIndex - 1);
   }
 
   protected void populateLastNonRowFields() {
     populateNonRowFields(currentRowNode.getLastCellIndex());
   }
 
   protected void populateNonRowFields(int cellIndex) {
     currentCellIndex = cellIndex;
     populateFamily();
     populateQualifier();
     // Read tags only if there are tags in the meta
     if(blockMeta.getNumTagsBytes() != 0) {
       populateTag();
     }
     populateTimestamp();
     populateMvccVersion();
     populateType();
     populateValueOffsets();
   }
 
   protected void populateFamily() {
     int familyTreeIndex = currentRowNode.getFamilyOffset(currentCellIndex, blockMeta);
     familyOffset = familyReader.populateBuffer(familyTreeIndex).getColumnOffset();
     familyLength = familyReader.getColumnLength();
   }
 
   protected void populateQualifier() {
     int qualifierTreeIndex = currentRowNode.getColumnOffset(currentCellIndex, blockMeta);
     qualifierOffset = qualifierReader.populateBuffer(qualifierTreeIndex).getColumnOffset();
     qualifierLength = qualifierReader.getColumnLength();
   }
 
   protected void populateTag() {
     int tagTreeIndex = currentRowNode.getTagOffset(currentCellIndex, blockMeta);
     tagsOffset = tagsReader.populateBuffer(tagTreeIndex).getColumnOffset();
     tagsLength = tagsReader.getColumnLength();
   }
 
   protected void populateTimestamp() {
     if (blockMeta.isAllSameTimestamp()) {
       timestamp = blockMeta.getMinTimestamp();
     } else {
       int timestampIndex = currentRowNode.getTimestampIndex(currentCellIndex, blockMeta);
       timestamp = timestampDecoder.getLong(timestampIndex);
     }
   }
 
   protected void populateMvccVersion() {
     if (blockMeta.isAllSameMvccVersion()) {
       mvccVersion = blockMeta.getMinMvccVersion();
     } else {
       int mvccVersionIndex = currentRowNode.getMvccVersionIndex(currentCellIndex,
         blockMeta);
       mvccVersion = mvccVersionDecoder.getMvccVersion(mvccVersionIndex);
     }
   }
 
   protected void populateType() {
     int typeInt;
     if (blockMeta.isAllSameType()) {
       typeInt = blockMeta.getAllTypes();
     } else {
       typeInt = currentRowNode.getType(currentCellIndex, blockMeta);
     }
     type = PrefixTreeCell.TYPES[typeInt];
   }
 
   protected void populateValueOffsets() {
     int offsetIntoValueSection = currentRowNode.getValueOffset(currentCellIndex, blockMeta);
     absoluteValueOffset = blockMeta.getAbsoluteValueOffset() + offsetIntoValueSection;
     valueLength = currentRowNode.getValueLength(currentCellIndex, blockMeta);
   }
 
   /**************** getters ***************************/
 
   public byte[] getTreeBytes() {
     return block;
   }
 
   public PrefixTreeBlockMeta getBlockMeta() {
     return blockMeta;
   }
 
   public int getMaxRowTreeStackNodes() {
     return rowNodes.length;
   }
 
   public int getRowBufferLength() {
     return rowBuffer.length;
   }
 
   public int getQualifierBufferLength() {
     return qualifierBuffer.length;
   }
 
   public int getTagBufferLength() {
     return tagsBuffer.length;
   }
 }