core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/unsafe/UnsafeVariableLengthDimensionDataChunkStore.java - carbondata - Git at Google

 /*
  * 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;
   }

 }
	/*
	* 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;
	}

	}