content: /*
   * 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.carbondata.core.datastore.chunk.store.impl.unsafe;
  
  import java.nio.ByteBuffer;
  
  import org.apache.carbondata.core.constants.CarbonCommonConstants;
  import org.apache.carbondata.core.memory.CarbonUnsafe;
  import org.apache.carbondata.core.scan.executor.util.QueryUtil;
  import org.apache.carbondata.core.scan.result.vector.CarbonColumnVector;
  
  /**
   * Below class is responsible to store variable length dimension data chunk in
   * memory Memory occupied can be on heap or off-heap using unsafe interface
   */
  public abstract class UnsafeVariableLengthDimensionDataChunkStore
      extends UnsafeAbstractDimensionDataChunkStore {
  
    /**
     * total number of rows
     */
    private int numberOfRows;
  
    /**
     * pointers offsets
     */
    private long dataPointersOffsets;
  
    /**
     * Reusable data array
     * this will be useful for vector scenario, as it will be created once and filled every time
     * if new data length is bigger than exiting data length then create new data with bigger length
     * and assign to value
     */
    private byte[] value;
  
    public UnsafeVariableLengthDimensionDataChunkStore(long totalSize, boolean isInvertedIndex,
        int numberOfRows, int dataLength) {
      super(totalSize, isInvertedIndex, numberOfRows, dataLength);
      this.numberOfRows = numberOfRows;
      // initials size assigning to some random value
      this.value = new byte[20];
    }
  
    /**
     * Below method will be used to put the rows and its metadata in off-heap
     *
     * @param invertedIndex        inverted index to be stored
     * @param invertedIndexReverse inverted index reverse to be stored
     * @param data                 data to be stored
     */
    @Override
    public void putArray(final int[] invertedIndex, final int[] invertedIndexReverse,
        byte[] data) {
      // first put the data, inverted index and reverse inverted index to memory
      super.putArray(invertedIndex, invertedIndexReverse, data);
      // position from where offsets will start
      this.dataPointersOffsets = this.invertedIndexReverseOffset;
      if (isExplicitSorted) {
        this.dataPointersOffsets += (long) numberOfRows * CarbonCommonConstants.INT_SIZE_IN_BYTE;
      }
      // As data is of variable length and data format is
      // <length in short><data><length in short/int><data>
      // we need to store offset of each data so data can be accessed directly
      // for example:
      //data = {0,5,1,2,3,4,5,0,6,0,1,2,3,4,5,0,2,8,9}
      //so value stored in offset will be position of actual data
      // [2,9,17]
      // to store this value we need to get the actual data length + 2/4 bytes used for storing the
      // length
  
      // start position will be used to store the current data position
      int startOffset = 0;
      // as first position will be start from 2/4 byte as data is stored first in the memory block
      // we need to skip first two bytes this is because first two bytes will be length of the data
      // which we have to skip
      int [] dataOffsets = new int[numberOfRows];
      dataOffsets[0] = getLengthSize();
      // creating a byte buffer which will wrap the length of the row
      ByteBuffer buffer = ByteBuffer.wrap(data);
      for (int i = 1; i < numberOfRows; i++) {
        buffer.position(startOffset);
        // so current row position will be
        // previous row length + 2/4 bytes used for storing previous row data
        startOffset += getLengthFromBuffer(buffer) + getLengthSize();
        // as same byte buffer is used to avoid creating many byte buffer for each row
        // we need to clear the byte buffer
        dataOffsets[i] = startOffset + getLengthSize();
      }
      CarbonUnsafe.getUnsafe().copyMemory(dataOffsets, CarbonUnsafe.INT_ARRAY_OFFSET,
          dataPageMemoryBlock.getBaseObject(),
          dataPageMemoryBlock.getBaseOffset() + this.dataPointersOffsets,
          dataOffsets.length * CarbonCommonConstants.INT_SIZE_IN_BYTE);
    }
  
    protected abstract int getLengthSize();
  
    protected abstract int getLengthFromBuffer(ByteBuffer byteBuffer);
  
    /**
     * Below method will be used to get the row based on row id passed
     * Getting the row from unsafe works in below logic
     * 1. if inverted index is present then get the row id based on reverse inverted index
     * 2. get the current row id data offset
     * 3. if it's not a last row- get the next row offset
     * Subtract the current row offset + 2 bytes(to skip the data length) with next row offset
     * 4. if it's last row
     * subtract the current row offset + 2 bytes(to skip the data length) with complete data length
     * @param rowId
     * @return row
     */
    @Override
    public byte[] getRow(int rowId) {
      // get the actual row id
      rowId = getRowId(rowId);
      // get offset of data in unsafe
      int currentDataOffset = getOffSet(rowId);
      // get the data length
      int length = getLength(rowId, currentDataOffset);
      // create data array
      byte[] data = new byte[length];
      // fill the row data
      fillRowInternal(length, data, currentDataOffset);
      return data;
    }
  
    /**
     * Returns the actual row id for data
     * if inverted index is present then get the row id based on reverse inverted index
     * otherwise return the same row id
     * @param rowId row id
     * @return actual row id
     */
    private int getRowId(int rowId) {
      // if column was explicitly sorted we need to get the row id based inverted index reverse
      if (isExplicitSorted) {
        rowId = CarbonUnsafe.getUnsafe().getInt(dataPageMemoryBlock.getBaseObject(),
            dataPageMemoryBlock.getBaseOffset() + this.invertedIndexReverseOffset + ((long)rowId
                * CarbonCommonConstants.INT_SIZE_IN_BYTE));
      }
      return rowId;
    }
  
    /**
     * get data offset based on current row id
     * @param rowId row id
     * @return data offset
     */
    private int getOffSet(int rowId) {
      return CarbonUnsafe.getUnsafe().getInt(dataPageMemoryBlock.getBaseObject(),
          dataPageMemoryBlock.getBaseOffset() + this.dataPointersOffsets + ((long)rowId
              * CarbonCommonConstants.INT_SIZE_IN_BYTE));
    }
  
    /**
     * To get the length of data for row id
     * if it's not a last row- get the next row offset
     * Subtract the current row offset + 2/4 bytes(to skip the data length) with next row offset
     * if it's last row
     * subtract the current row offset + 2/4 bytes(to skip the data length) with complete data length
     * @param rowId rowId
     * @param currentDataOffset current data offset
     * @return length of row
     */
    private int getLength(int rowId, int currentDataOffset) {
      int length = 0;
      // calculating the length of data
      if (rowId < numberOfRows - 1) {
        int OffsetOfNextData = CarbonUnsafe.getUnsafe().getInt(dataPageMemoryBlock.getBaseObject(),
            dataPageMemoryBlock.getBaseOffset() + this.dataPointersOffsets + ((rowId + 1)
                * CarbonCommonConstants.INT_SIZE_IN_BYTE));
        length = OffsetOfNextData - (currentDataOffset + getLengthSize());
      } else {
        // for last record we need to subtract with data length
        length = this.dataLength - currentDataOffset;
      }
      return length;
    }
  
    /**
     * Return the row from unsafe
     * @param length length of the data
     * @param data data array
     * @param currentDataOffset current data offset
     */
    private void fillRowInternal(int length, byte[] data, int currentDataOffset) {
      CarbonUnsafe.getUnsafe().copyMemory(dataPageMemoryBlock.getBaseObject(),
          dataPageMemoryBlock.getBaseOffset() + currentDataOffset, data,
          CarbonUnsafe.BYTE_ARRAY_OFFSET, length);
    }
  
    /**
     *
     * Below method will be used to put the row in vector based on row id passed
     * Getting the row from unsafe works in below logic
     * 1. if inverted index is present then get the row id based on reverse inverted index
     * 2. get the current row id data offset
     * 3. if it's not a last row- get the next row offset
     * Subtract the current row offset + 2 bytes(to skip the data length) with next row offset
     * 4. if it's last row
     * subtract the current row offset + 2 bytes(to skip the data length) with complete data length
     * @param rowId row id
     * @param vector vector to be filled
     * @param vectorRow vector row id
     *
     */
    @Override
    public void fillRow(int rowId, CarbonColumnVector vector, int vectorRow) {
      vector.setDictionary(null);
      // get the row id from reverse inverted index based on row id
      rowId = getRowId(rowId);
      // get the current row offset
      int currentDataOffset = getOffSet(rowId);
      // get the row data length
      int length = getLength(rowId, currentDataOffset);
      // check if value length is less the current data length
      // then create a new array else use the same
      if (length > value.length) {
        value = new byte[length];
      }
      // get the row from unsafe
      fillRowInternal(length, value, currentDataOffset);
      QueryUtil.putDataToVector(vector, value, vectorRow, length);
    }
  
    /**
     * to compare the two byte array
     *
     * @param rowId index of first byte array
     * @param compareValue value of to be compared
     * @return compare result
     */
    @Override
    public int compareTo(int rowId, byte[] compareValue) {
      int currentDataOffset = getOffSet(rowId);
      int length = getLength(rowId, currentDataOffset);
      // as this class handles this variable length data, so filter value can be
      // smaller or bigger than than actual data, so we need to take the smaller length
      int compareResult;
      int compareLength = Math.min(length , compareValue.length);
      for (int i = 0; i < compareLength; i++) {
        compareResult = (CarbonUnsafe.getUnsafe().getByte(dataPageMemoryBlock.getBaseObject(),
            dataPageMemoryBlock.getBaseOffset() + currentDataOffset) & 0xff) - (compareValue[i]
            & 0xff);
        // if compare result is not equal we can break
        if (compareResult != 0) {
          return compareResult;
        }
        // increment the offset by one as comparison is done byte by byte
        currentDataOffset++;
      }
      return length - compareValue.length;
    }
  
  }
  