You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@carbondata.apache.org by ma...@apache.org on 2018/11/21 09:10:42 UTC
[1/2] carbondata git commit: [CARBONDATA-3113] Fixed Local Dictionary
Query Performance and Added reusable buffer for direct flow
Repository: carbondata
Updated Branches:
refs/heads/master 647bfbaea -> d79ba999f
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/datastore/page/ColumnPage.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/page/ColumnPage.java b/core/src/main/java/org/apache/carbondata/core/datastore/page/ColumnPage.java
index e5312f3..51dfbf2 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/page/ColumnPage.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/page/ColumnPage.java
@@ -308,7 +308,8 @@ public abstract class ColumnPage {
private static ColumnPage newDecimalPage(ColumnPageEncoderMeta meta,
byte[] lvEncodedByteArray) throws MemoryException {
- return VarLengthColumnPageBase.newDecimalColumnPage(meta, lvEncodedByteArray);
+ return VarLengthColumnPageBase
+ .newDecimalColumnPage(meta, lvEncodedByteArray, lvEncodedByteArray.length);
}
private static ColumnPage newLVBytesPage(TableSpec.ColumnSpec columnSpec,
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/datastore/page/DecoderBasedFallbackEncoder.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/page/DecoderBasedFallbackEncoder.java b/core/src/main/java/org/apache/carbondata/core/datastore/page/DecoderBasedFallbackEncoder.java
index 9bed89f..1de8201 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/page/DecoderBasedFallbackEncoder.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/page/DecoderBasedFallbackEncoder.java
@@ -88,7 +88,8 @@ public class DecoderBasedFallbackEncoder implements Callable<FallbackEncodedColu
CarbonUtil.getIntArray(data, offset, encodedColumnPage.getPageMetadata().rle_page_length);
// uncompress the data with rle indexes
bytes = UnBlockIndexer
- .uncompressData(bytes, rlePage, CarbonCommonConstants.LOCAL_DICT_ENCODED_BYTEARRAY_SIZE);
+ .uncompressData(bytes, rlePage, CarbonCommonConstants.LOCAL_DICT_ENCODED_BYTEARRAY_SIZE,
+ bytes.length);
}
// disable encoding using local dictionary
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/datastore/page/VarLengthColumnPageBase.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/page/VarLengthColumnPageBase.java b/core/src/main/java/org/apache/carbondata/core/datastore/page/VarLengthColumnPageBase.java
index 81bb1b5..0f409f6 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/page/VarLengthColumnPageBase.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/page/VarLengthColumnPageBase.java
@@ -125,7 +125,7 @@ public abstract class VarLengthColumnPageBase extends ColumnPage {
* Create a new column page for decimal page
*/
public static ColumnPage newDecimalColumnPage(ColumnPageEncoderMeta meta,
- byte[] lvEncodedBytes) throws MemoryException {
+ byte[] lvEncodedBytes, int actualDataLength) throws MemoryException {
TableSpec.ColumnSpec columnSpec = meta.getColumnSpec();
DecimalConverterFactory.DecimalConverter decimalConverter =
DecimalConverterFactory.INSTANCE.getDecimalConverter(columnSpec.getPrecision(),
@@ -137,7 +137,7 @@ public abstract class VarLengthColumnPageBase extends ColumnPage {
CarbonCommonConstants.INT_SIZE_IN_BYTE, meta.getCompressorName());
} else {
// Here the size is always fixed.
- return getDecimalColumnPage(meta, lvEncodedBytes, size);
+ return getDecimalColumnPage(meta, lvEncodedBytes, size, actualDataLength);
}
}
@@ -160,7 +160,7 @@ public abstract class VarLengthColumnPageBase extends ColumnPage {
}
private static ColumnPage getDecimalColumnPage(ColumnPageEncoderMeta meta,
- byte[] lvEncodedBytes, int size) throws MemoryException {
+ byte[] lvEncodedBytes, int size, int actualDataLength) throws MemoryException {
TableSpec.ColumnSpec columnSpec = meta.getColumnSpec();
String compressorName = meta.getCompressorName();
TableSpec.ColumnSpec spec = TableSpec.ColumnSpec
@@ -171,7 +171,7 @@ public abstract class VarLengthColumnPageBase extends ColumnPage {
int offset;
int rowId = 0;
int counter = 0;
- for (offset = 0; offset < lvEncodedBytes.length; offset += size) {
+ for (offset = 0; offset < actualDataLength; offset += size) {
rowOffset.putInt(counter, offset);
rowId++;
counter++;
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/ColumnPageDecoder.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/ColumnPageDecoder.java b/core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/ColumnPageDecoder.java
index 6f36c08..b5dc502 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/ColumnPageDecoder.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/ColumnPageDecoder.java
@@ -20,6 +20,7 @@ package org.apache.carbondata.core.datastore.page.encoding;
import java.io.IOException;
import java.util.BitSet;
+import org.apache.carbondata.core.datastore.ReusableDataBuffer;
import org.apache.carbondata.core.datastore.page.ColumnPage;
import org.apache.carbondata.core.memory.MemoryException;
import org.apache.carbondata.core.scan.result.vector.ColumnVectorInfo;
@@ -29,13 +30,15 @@ public interface ColumnPageDecoder {
/**
* Apply decoding algorithm on input byte array and return decoded column page
*/
- ColumnPage decode(byte[] input, int offset, int length) throws MemoryException, IOException;
+ ColumnPage decode(byte[] input, int offset, int length)
+ throws MemoryException, IOException;
/**
* Apply decoding algorithm on input byte array and fill the vector here.
*/
void decodeAndFillVector(byte[] input, int offset, int length, ColumnVectorInfo vectorInfo,
- BitSet nullBits, boolean isLVEncoded, int pageSize) throws MemoryException, IOException;
+ BitSet nullBits, boolean isLVEncoded, int pageSize, ReusableDataBuffer reusableDataBuffer)
+ throws MemoryException, IOException;
ColumnPage decode(byte[] input, int offset, int length, boolean isLVEncoded)
throws MemoryException, IOException;
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/adaptive/AdaptiveDeltaFloatingCodec.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/adaptive/AdaptiveDeltaFloatingCodec.java b/core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/adaptive/AdaptiveDeltaFloatingCodec.java
index 735847e..fb53dba 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/adaptive/AdaptiveDeltaFloatingCodec.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/adaptive/AdaptiveDeltaFloatingCodec.java
@@ -23,6 +23,7 @@ import java.util.BitSet;
import java.util.List;
import java.util.Map;
+import org.apache.carbondata.core.datastore.ReusableDataBuffer;
import org.apache.carbondata.core.datastore.compression.Compressor;
import org.apache.carbondata.core.datastore.compression.CompressorFactory;
import org.apache.carbondata.core.datastore.page.ColumnPage;
@@ -123,20 +124,26 @@ public class AdaptiveDeltaFloatingCodec extends AdaptiveCodec {
@Override
public ColumnPageDecoder createDecoder(final ColumnPageEncoderMeta meta) {
return new ColumnPageDecoder() {
- @Override
- public ColumnPage decode(byte[] input, int offset, int length)
+ @Override public ColumnPage decode(byte[] input, int offset, int length)
throws MemoryException, IOException {
ColumnPage page = ColumnPage.decompress(meta, input, offset, length, false);
return LazyColumnPage.newPage(page, converter);
}
- @Override
- public void decodeAndFillVector(byte[] input, int offset, int length,
- ColumnVectorInfo vectorInfo, BitSet nullBits, boolean isLVEncoded, int pageSize)
+ @Override public void decodeAndFillVector(byte[] input, int offset, int length,
+ ColumnVectorInfo vectorInfo, BitSet nullBits, boolean isLVEncoded, int pageSize,
+ ReusableDataBuffer reusableDataBuffer)
throws MemoryException, IOException {
Compressor compressor =
CompressorFactory.getInstance().getCompressor(meta.getCompressorName());
- byte[] unCompressData = compressor.unCompressByte(input, offset, length);
+ byte[] unCompressData;
+ if (null != reusableDataBuffer && compressor.supportReusableBuffer()) {
+ int uncompressedLength = compressor.unCompressedLength(input, offset, length);
+ unCompressData = reusableDataBuffer.getDataBuffer(uncompressedLength);
+ compressor.rawUncompress(input, offset, length, unCompressData);
+ } else {
+ unCompressData = compressor.unCompressByte(input, offset, length);
+ }
converter.decodeAndFillVector(unCompressData, vectorInfo, nullBits, meta.getStoreDataType(),
pageSize);
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/adaptive/AdaptiveDeltaIntegralCodec.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/adaptive/AdaptiveDeltaIntegralCodec.java b/core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/adaptive/AdaptiveDeltaIntegralCodec.java
index 578945b..9ed21f4 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/adaptive/AdaptiveDeltaIntegralCodec.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/adaptive/AdaptiveDeltaIntegralCodec.java
@@ -24,6 +24,7 @@ import java.util.BitSet;
import java.util.List;
import java.util.Map;
+import org.apache.carbondata.core.datastore.ReusableDataBuffer;
import org.apache.carbondata.core.datastore.TableSpec;
import org.apache.carbondata.core.datastore.compression.Compressor;
import org.apache.carbondata.core.datastore.compression.CompressorFactory;
@@ -130,8 +131,7 @@ public class AdaptiveDeltaIntegralCodec extends AdaptiveCodec {
@Override
public ColumnPageDecoder createDecoder(final ColumnPageEncoderMeta meta) {
return new ColumnPageDecoder() {
- @Override
- public ColumnPage decode(byte[] input, int offset, int length)
+ @Override public ColumnPage decode(byte[] input, int offset, int length)
throws MemoryException, IOException {
ColumnPage page = null;
if (DataTypes.isDecimal(meta.getSchemaDataType())) {
@@ -142,13 +142,20 @@ public class AdaptiveDeltaIntegralCodec extends AdaptiveCodec {
return LazyColumnPage.newPage(page, converter);
}
- @Override
- public void decodeAndFillVector(byte[] input, int offset, int length,
- ColumnVectorInfo vectorInfo, BitSet nullBits, boolean isLVEncoded, int pageSize)
+ @Override public void decodeAndFillVector(byte[] input, int offset, int length,
+ ColumnVectorInfo vectorInfo, BitSet nullBits, boolean isLVEncoded, int pageSize,
+ ReusableDataBuffer reusableDataBuffer)
throws MemoryException, IOException {
Compressor compressor =
CompressorFactory.getInstance().getCompressor(meta.getCompressorName());
- byte[] unCompressData = compressor.unCompressByte(input, offset, length);
+ byte[] unCompressData;
+ if (null != reusableDataBuffer && compressor.supportReusableBuffer()) {
+ int uncompressedLength = compressor.unCompressedLength(input, offset, length);
+ unCompressData = reusableDataBuffer.getDataBuffer(uncompressedLength);
+ compressor.rawUncompress(input, offset, length, unCompressData);
+ } else {
+ unCompressData = compressor.unCompressByte(input, offset, length);
+ }
if (DataTypes.isDecimal(meta.getSchemaDataType())) {
TableSpec.ColumnSpec columnSpec = meta.getColumnSpec();
DecimalConverterFactory.DecimalConverter decimalConverter =
@@ -160,8 +167,7 @@ public class AdaptiveDeltaIntegralCodec extends AdaptiveCodec {
pageSize);
}
- @Override
- public ColumnPage decode(byte[] input, int offset, int length, boolean isLVEncoded)
+ @Override public ColumnPage decode(byte[] input, int offset, int length, boolean isLVEncoded)
throws MemoryException, IOException {
return decode(input, offset, length);
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/adaptive/AdaptiveFloatingCodec.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/adaptive/AdaptiveFloatingCodec.java b/core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/adaptive/AdaptiveFloatingCodec.java
index c66c065..b16e57d 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/adaptive/AdaptiveFloatingCodec.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/adaptive/AdaptiveFloatingCodec.java
@@ -23,6 +23,7 @@ import java.util.BitSet;
import java.util.List;
import java.util.Map;
+import org.apache.carbondata.core.datastore.ReusableDataBuffer;
import org.apache.carbondata.core.datastore.compression.Compressor;
import org.apache.carbondata.core.datastore.compression.CompressorFactory;
import org.apache.carbondata.core.datastore.page.ColumnPage;
@@ -111,26 +112,31 @@ public class AdaptiveFloatingCodec extends AdaptiveCodec {
@Override
public ColumnPageDecoder createDecoder(final ColumnPageEncoderMeta meta) {
return new ColumnPageDecoder() {
- @Override
- public ColumnPage decode(byte[] input, int offset, int length)
+ @Override public ColumnPage decode(byte[] input, int offset, int length)
throws MemoryException, IOException {
ColumnPage page = ColumnPage.decompress(meta, input, offset, length, false);
return LazyColumnPage.newPage(page, converter);
}
- @Override
- public void decodeAndFillVector(byte[] input, int offset, int length,
- ColumnVectorInfo vectorInfo, BitSet nullBits, boolean isLVEncoded, int pageSize)
+ @Override public void decodeAndFillVector(byte[] input, int offset, int length,
+ ColumnVectorInfo vectorInfo, BitSet nullBits, boolean isLVEncoded, int pageSize,
+ ReusableDataBuffer reusableDataBuffer)
throws MemoryException, IOException {
Compressor compressor =
CompressorFactory.getInstance().getCompressor(meta.getCompressorName());
- byte[] unCompressData = compressor.unCompressByte(input, offset, length);
+ byte[] unCompressData;
+ if (null != reusableDataBuffer && compressor.supportReusableBuffer()) {
+ int uncompressedLength = compressor.unCompressedLength(input, offset, length);
+ unCompressData = reusableDataBuffer.getDataBuffer(uncompressedLength);
+ compressor.rawUncompress(input, offset, length, unCompressData);
+ } else {
+ unCompressData = compressor.unCompressByte(input, offset, length);
+ }
converter.decodeAndFillVector(unCompressData, vectorInfo, nullBits, meta.getStoreDataType(),
pageSize);
}
- @Override
- public ColumnPage decode(byte[] input, int offset, int length, boolean isLVEncoded)
+ @Override public ColumnPage decode(byte[] input, int offset, int length, boolean isLVEncoded)
throws MemoryException, IOException {
return decode(input, offset, length);
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/adaptive/AdaptiveIntegralCodec.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/adaptive/AdaptiveIntegralCodec.java b/core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/adaptive/AdaptiveIntegralCodec.java
index d9db437..66639fe 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/adaptive/AdaptiveIntegralCodec.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/adaptive/AdaptiveIntegralCodec.java
@@ -23,6 +23,7 @@ import java.util.BitSet;
import java.util.List;
import java.util.Map;
+import org.apache.carbondata.core.datastore.ReusableDataBuffer;
import org.apache.carbondata.core.datastore.TableSpec;
import org.apache.carbondata.core.datastore.compression.Compressor;
import org.apache.carbondata.core.datastore.compression.CompressorFactory;
@@ -107,8 +108,7 @@ public class AdaptiveIntegralCodec extends AdaptiveCodec {
@Override
public ColumnPageDecoder createDecoder(final ColumnPageEncoderMeta meta) {
return new ColumnPageDecoder() {
- @Override
- public ColumnPage decode(byte[] input, int offset, int length)
+ @Override public ColumnPage decode(byte[] input, int offset, int length)
throws MemoryException, IOException {
ColumnPage page = null;
if (DataTypes.isDecimal(meta.getSchemaDataType())) {
@@ -119,13 +119,19 @@ public class AdaptiveIntegralCodec extends AdaptiveCodec {
return LazyColumnPage.newPage(page, converter);
}
- @Override
- public void decodeAndFillVector(byte[] input, int offset, int length,
- ColumnVectorInfo vectorInfo, BitSet nullBits, boolean isLVEncoded, int pageSize)
- throws MemoryException, IOException {
+ @Override public void decodeAndFillVector(byte[] input, int offset, int length,
+ ColumnVectorInfo vectorInfo, BitSet nullBits, boolean isLVEncoded, int pageSize,
+ ReusableDataBuffer reusableDataBuffer) throws MemoryException, IOException {
Compressor compressor =
CompressorFactory.getInstance().getCompressor(meta.getCompressorName());
- byte[] unCompressData = compressor.unCompressByte(input, offset, length);
+ byte[] unCompressData;
+ if (null != reusableDataBuffer && compressor.supportReusableBuffer()) {
+ int uncompressedLength = compressor.unCompressedLength(input, offset, length);
+ unCompressData = reusableDataBuffer.getDataBuffer(uncompressedLength);
+ compressor.rawUncompress(input, offset, length, unCompressData);
+ } else {
+ unCompressData = compressor.unCompressByte(input, offset, length);
+ }
if (DataTypes.isDecimal(meta.getSchemaDataType())) {
TableSpec.ColumnSpec columnSpec = meta.getColumnSpec();
DecimalConverterFactory.DecimalConverter decimalConverter =
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/compress/DirectCompressCodec.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/compress/DirectCompressCodec.java b/core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/compress/DirectCompressCodec.java
index 7b7c0b6..60344d6 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/compress/DirectCompressCodec.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/compress/DirectCompressCodec.java
@@ -23,6 +23,7 @@ import java.util.BitSet;
import java.util.List;
import java.util.Map;
+import org.apache.carbondata.core.datastore.ReusableDataBuffer;
import org.apache.carbondata.core.datastore.TableSpec;
import org.apache.carbondata.core.datastore.compression.Compressor;
import org.apache.carbondata.core.datastore.compression.CompressorFactory;
@@ -93,8 +94,8 @@ public class DirectCompressCodec implements ColumnPageCodec {
public ColumnPageDecoder createDecoder(final ColumnPageEncoderMeta meta) {
return new ColumnPageDecoder() {
- @Override
- public ColumnPage decode(byte[] input, int offset, int length) throws MemoryException {
+ @Override public ColumnPage decode(byte[] input, int offset, int length)
+ throws MemoryException {
ColumnPage decodedPage;
if (DataTypes.isDecimal(dataType)) {
decodedPage = ColumnPage.decompressDecimalPage(meta, input, offset, length);
@@ -106,11 +107,21 @@ public class DirectCompressCodec implements ColumnPageCodec {
@Override
public void decodeAndFillVector(byte[] input, int offset, int length,
- ColumnVectorInfo vectorInfo, BitSet nullBits, boolean isLVEncoded, int pageSize)
+ ColumnVectorInfo vectorInfo, BitSet nullBits, boolean isLVEncoded, int pageSize,
+ ReusableDataBuffer reusableDataBuffer)
throws MemoryException, IOException {
Compressor compressor =
CompressorFactory.getInstance().getCompressor(meta.getCompressorName());
- byte[] unCompressData = compressor.unCompressByte(input, offset, length);
+ int uncompressedLength;
+ byte[] unCompressData;
+ if (null != reusableDataBuffer && compressor.supportReusableBuffer()) {
+ uncompressedLength = compressor.unCompressedLength(input, offset, length);
+ unCompressData = reusableDataBuffer.getDataBuffer(uncompressedLength);
+ compressor.rawUncompress(input, offset, length, unCompressData);
+ } else {
+ unCompressData = compressor.unCompressByte(input, offset, length);
+ uncompressedLength = unCompressData.length;
+ }
if (DataTypes.isDecimal(dataType)) {
TableSpec.ColumnSpec columnSpec = meta.getColumnSpec();
DecimalConverterFactory.DecimalConverter decimalConverter =
@@ -118,8 +129,8 @@ public class DirectCompressCodec implements ColumnPageCodec {
.getDecimalConverter(columnSpec.getPrecision(), columnSpec.getScale());
vectorInfo.decimalConverter = decimalConverter;
if (DataTypes.isDecimal(meta.getStoreDataType())) {
- ColumnPage decimalColumnPage =
- VarLengthColumnPageBase.newDecimalColumnPage(meta, unCompressData);
+ ColumnPage decimalColumnPage = VarLengthColumnPageBase
+ .newDecimalColumnPage(meta, unCompressData, uncompressedLength);
decimalConverter.fillVector(decimalColumnPage.getByteArrayPage(), pageSize, vectorInfo,
nullBits, meta.getStoreDataType());
} else {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/rle/RLECodec.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/rle/RLECodec.java b/core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/rle/RLECodec.java
index 2c940bb..836a8b1 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/rle/RLECodec.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/rle/RLECodec.java
@@ -27,6 +27,7 @@ import java.util.BitSet;
import java.util.List;
import java.util.Map;
+import org.apache.carbondata.core.datastore.ReusableDataBuffer;
import org.apache.carbondata.core.datastore.TableSpec;
import org.apache.carbondata.core.datastore.page.ColumnPage;
import org.apache.carbondata.core.datastore.page.encoding.ColumnPageCodec;
@@ -295,8 +296,7 @@ public class RLECodec implements ColumnPageCodec {
this.compressorName = compressorName;
}
- @Override
- public ColumnPage decode(byte[] input, int offset, int length)
+ @Override public ColumnPage decode(byte[] input, int offset, int length)
throws MemoryException, IOException {
DataType dataType = columnSpec.getSchemaDataType();
DataInputStream in = new DataInputStream(new ByteArrayInputStream(input, offset, length));
@@ -316,9 +316,9 @@ public class RLECodec implements ColumnPageCodec {
return resultPage;
}
- @Override
- public void decodeAndFillVector(byte[] input, int offset, int length,
- ColumnVectorInfo vectorInfo, BitSet nullBits, boolean isLVEncoded, int pageSize)
+ @Override public void decodeAndFillVector(byte[] input, int offset, int length,
+ ColumnVectorInfo vectorInfo, BitSet nullBits, boolean isLVEncoded, int pageSize,
+ ReusableDataBuffer reusableDataBuffer)
throws MemoryException, IOException {
throw new UnsupportedOperationException("Not supposed to be called here");
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/scan/executor/impl/AbstractQueryExecutor.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/executor/impl/AbstractQueryExecutor.java b/core/src/main/java/org/apache/carbondata/core/scan/executor/impl/AbstractQueryExecutor.java
index 710fbe3..40d04e2 100644
--- a/core/src/main/java/org/apache/carbondata/core/scan/executor/impl/AbstractQueryExecutor.java
+++ b/core/src/main/java/org/apache/carbondata/core/scan/executor/impl/AbstractQueryExecutor.java
@@ -35,6 +35,7 @@ import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.constants.CarbonV3DataFormatConstants;
import org.apache.carbondata.core.datamap.Segment;
import org.apache.carbondata.core.datastore.IndexKey;
+import org.apache.carbondata.core.datastore.ReusableDataBuffer;
import org.apache.carbondata.core.datastore.block.AbstractIndex;
import org.apache.carbondata.core.datastore.block.SegmentProperties;
import org.apache.carbondata.core.datastore.block.TableBlockInfo;
@@ -404,19 +405,33 @@ public abstract class AbstractQueryExecutor<E> implements QueryExecutor<E> {
// fill all the block execution infos for all the blocks selected in
// query
// and query will be executed based on that infos
+ ReusableDataBuffer[] dimensionReusableDataBuffers = null;
+ ReusableDataBuffer[] measureReusableDataBuffers = null;
+
for (int i = 0; i < queryProperties.dataBlocks.size(); i++) {
AbstractIndex abstractIndex = queryProperties.dataBlocks.get(i);
BlockletDataRefNode dataRefNode =
(BlockletDataRefNode) abstractIndex.getDataRefNode();
- blockExecutionInfoList.add(
- getBlockExecutionInfoForBlock(
- queryModel,
- abstractIndex,
+ final BlockExecutionInfo blockExecutionInfoForBlock =
+ getBlockExecutionInfoForBlock(queryModel, abstractIndex,
dataRefNode.getBlockInfos().get(0).getBlockletInfos().getStartBlockletNumber(),
- dataRefNode.numberOfNodes(),
- dataRefNode.getBlockInfos().get(0).getFilePath(),
+ dataRefNode.numberOfNodes(), dataRefNode.getBlockInfos().get(0).getFilePath(),
dataRefNode.getBlockInfos().get(0).getDeletedDeltaFilePath(),
- dataRefNode.getBlockInfos().get(0).getSegment()));
+ dataRefNode.getBlockInfos().get(0).getSegment());
+ if (null == dimensionReusableDataBuffers || null == measureReusableDataBuffers) {
+ dimensionReusableDataBuffers = blockExecutionInfoForBlock.getDimensionResusableDataBuffer();
+ measureReusableDataBuffers = blockExecutionInfoForBlock.getMeasureResusableDataBuffer();
+ } else {
+ if (dimensionReusableDataBuffers.length == blockExecutionInfoForBlock
+ .getDimensionResusableDataBuffer().length) {
+ blockExecutionInfoForBlock.setDimensionResusableDataBuffer(dimensionReusableDataBuffers);
+ }
+ if (measureReusableDataBuffers.length == blockExecutionInfoForBlock
+ .getMeasureResusableDataBuffer().length) {
+ blockExecutionInfoForBlock.setMeasureResusableDataBuffer(measureReusableDataBuffers);
+ }
+ }
+ blockExecutionInfoList.add(blockExecutionInfoForBlock);
}
if (null != queryModel.getStatisticsRecorder()) {
QueryStatistic queryStatistic = new QueryStatistic();
@@ -534,6 +549,16 @@ public abstract class AbstractQueryExecutor<E> implements QueryExecutor<E> {
int[] dimensionChunkIndexes = QueryUtil.getDimensionChunkIndexes(projectDimensions,
segmentProperties.getDimensionOrdinalToChunkMapping(),
currentBlockFilterDimensions, allProjectionListDimensionIdexes);
+ int reusableBufferSize = segmentProperties.getDimensionOrdinalToChunkMapping().size()
+ < projectDimensions.size() ?
+ projectDimensions.size() :
+ segmentProperties.getDimensionOrdinalToChunkMapping().size();
+ ReusableDataBuffer[] dimensionBuffer =
+ new ReusableDataBuffer[reusableBufferSize];
+ for (int i = 0; i < dimensionBuffer.length; i++) {
+ dimensionBuffer[i] = new ReusableDataBuffer();
+ }
+ blockExecutionInfo.setDimensionResusableDataBuffer(dimensionBuffer);
int numberOfColumnToBeReadInOneIO = Integer.parseInt(CarbonProperties.getInstance()
.getProperty(CarbonV3DataFormatConstants.NUMBER_OF_COLUMN_TO_READ_IN_IO,
CarbonV3DataFormatConstants.NUMBER_OF_COLUMN_TO_READ_IN_IO_DEFAULTVALUE));
@@ -558,6 +583,12 @@ public abstract class AbstractQueryExecutor<E> implements QueryExecutor<E> {
currentBlockQueryMeasures, expressionMeasures,
segmentProperties.getMeasuresOrdinalToChunkMapping(), filterMeasures,
allProjectionListMeasureIndexes);
+ ReusableDataBuffer[] measureBuffer =
+ new ReusableDataBuffer[segmentProperties.getMeasuresOrdinalToChunkMapping().size()];
+ for (int i = 0; i < measureBuffer.length; i++) {
+ measureBuffer[i] = new ReusableDataBuffer();
+ }
+ blockExecutionInfo.setMeasureResusableDataBuffer(measureBuffer);
if (measureChunkIndexes.length > 0) {
numberOfElementToConsider = measureChunkIndexes[measureChunkIndexes.length - 1]
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/scan/executor/infos/BlockExecutionInfo.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/executor/infos/BlockExecutionInfo.java b/core/src/main/java/org/apache/carbondata/core/scan/executor/infos/BlockExecutionInfo.java
index f0ef23b..f19b96b 100644
--- a/core/src/main/java/org/apache/carbondata/core/scan/executor/infos/BlockExecutionInfo.java
+++ b/core/src/main/java/org/apache/carbondata/core/scan/executor/infos/BlockExecutionInfo.java
@@ -20,6 +20,7 @@ import java.util.Map;
import org.apache.carbondata.core.datastore.DataRefNode;
import org.apache.carbondata.core.datastore.IndexKey;
+import org.apache.carbondata.core.datastore.ReusableDataBuffer;
import org.apache.carbondata.core.datastore.block.AbstractIndex;
import org.apache.carbondata.core.mutate.DeleteDeltaVo;
import org.apache.carbondata.core.scan.filter.GenericQueryType;
@@ -221,6 +222,10 @@ public class BlockExecutionInfo {
*/
private boolean isDirectVectorFill;
+ private ReusableDataBuffer[] dimensionResusableDataBuffer;
+
+ private ReusableDataBuffer[] measureResusableDataBuffer;
+
/**
* @param blockIndex the tableBlock to set
*/
@@ -638,4 +643,20 @@ public class BlockExecutionInfo {
public void setDirectVectorFill(boolean directVectorFill) {
isDirectVectorFill = directVectorFill;
}
+
+ public ReusableDataBuffer[] getDimensionResusableDataBuffer() {
+ return dimensionResusableDataBuffer;
+ }
+
+ public void setDimensionResusableDataBuffer(ReusableDataBuffer[] dimensionResusableDataBuffer) {
+ this.dimensionResusableDataBuffer = dimensionResusableDataBuffer;
+ }
+
+ public ReusableDataBuffer[] getMeasureResusableDataBuffer() {
+ return measureResusableDataBuffer;
+ }
+
+ public void setMeasureResusableDataBuffer(ReusableDataBuffer[] measureResusableDataBuffer) {
+ this.measureResusableDataBuffer = measureResusableDataBuffer;
+ }
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/scan/result/BlockletScannedResult.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/result/BlockletScannedResult.java b/core/src/main/java/org/apache/carbondata/core/scan/result/BlockletScannedResult.java
index 8217487..bb373eb 100644
--- a/core/src/main/java/org/apache/carbondata/core/scan/result/BlockletScannedResult.java
+++ b/core/src/main/java/org/apache/carbondata/core/scan/result/BlockletScannedResult.java
@@ -27,6 +27,7 @@ import java.util.Map;
import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
+import org.apache.carbondata.core.datastore.ReusableDataBuffer;
import org.apache.carbondata.core.datastore.chunk.DimensionColumnPage;
import org.apache.carbondata.core.datastore.chunk.impl.DimensionRawColumnChunk;
import org.apache.carbondata.core.datastore.chunk.impl.MeasureRawColumnChunk;
@@ -149,8 +150,14 @@ public abstract class BlockletScannedResult {
protected LazyBlockletLoader lazyBlockletLoader;
+ private ReusableDataBuffer[] dimensionReusableBuffer;
+
+ private ReusableDataBuffer[] measureReusableBuffer;
+
public BlockletScannedResult(BlockExecutionInfo blockExecutionInfo,
QueryStatisticsModel queryStatisticsModel) {
+ this.dimensionReusableBuffer = blockExecutionInfo.getDimensionResusableDataBuffer();
+ this.measureReusableBuffer = blockExecutionInfo.getMeasureResusableDataBuffer();
this.fixedLengthKeySize = blockExecutionInfo.getFixedLengthKeySize();
this.noDictionaryColumnChunkIndexes = blockExecutionInfo.getNoDictionaryColumnChunkIndexes();
this.dictionaryColumnChunkIndexes = blockExecutionInfo.getDictionaryColumnChunkIndex();
@@ -382,15 +389,15 @@ public abstract class BlockletScannedResult {
long startTime = System.currentTimeMillis();
for (int i = 0; i < dimensionColumnPages.length; i++) {
if (dimensionColumnPages[i][pageCounter] == null && dimRawColumnChunks[i] != null) {
- dimensionColumnPages[i][pageCounter] =
- dimRawColumnChunks[i].convertToDimColDataChunkWithOutCache(pageCounter);
+ dimensionColumnPages[i][pageCounter] = dimRawColumnChunks[i]
+ .convertToDimColDataChunkWithOutCache(pageCounter, null);
}
}
for (int i = 0; i < measureColumnPages.length; i++) {
if (measureColumnPages[i][pageCounter] == null && msrRawColumnChunks[i] != null) {
- measureColumnPages[i][pageCounter] =
- msrRawColumnChunks[i].convertToColumnPageWithOutCache(pageCounter);
+ measureColumnPages[i][pageCounter] = msrRawColumnChunks[i]
+ .convertToColumnPageWithOutCache(pageCounter, null);
}
}
QueryStatistic pageUncompressTime = queryStatisticsModel.getStatisticsTypeAndObjMap()
@@ -412,18 +419,20 @@ public abstract class BlockletScannedResult {
for (int i = 0; i < this.dictionaryColumnChunkIndexes.length; i++) {
dictionaryInfo[i].vector.setLazyPage(
new LazyPageLoader(lazyBlockletLoader, dictionaryColumnChunkIndexes[i], false,
- pageIdFiltered[pageCounter], dictionaryInfo[i]));
+ pageIdFiltered[pageCounter], dictionaryInfo[i], dimensionReusableBuffer[i]));
}
+ int startIndex = dictionaryColumnChunkIndexes.length;
for (int i = 0; i < this.noDictionaryColumnChunkIndexes.length; i++) {
noDictionaryInfo[i].vector.setLazyPage(
new LazyPageLoader(lazyBlockletLoader, noDictionaryColumnChunkIndexes[i], false,
- pageIdFiltered[pageCounter], noDictionaryInfo[i]));
+ pageIdFiltered[pageCounter], noDictionaryInfo[i],
+ dimensionReusableBuffer[startIndex++]));
}
for (int i = 0; i < measuresOrdinal.length; i++) {
msrVectorInfo[i].vector.setLazyPage(
new LazyPageLoader(lazyBlockletLoader, measuresOrdinal[i], true,
- pageIdFiltered[pageCounter], msrVectorInfo[i]));
+ pageIdFiltered[pageCounter], msrVectorInfo[i], measureReusableBuffer[i]));
}
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/scan/result/vector/CarbonDictionary.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/result/vector/CarbonDictionary.java b/core/src/main/java/org/apache/carbondata/core/scan/result/vector/CarbonDictionary.java
index 84d52a6..298eadc 100644
--- a/core/src/main/java/org/apache/carbondata/core/scan/result/vector/CarbonDictionary.java
+++ b/core/src/main/java/org/apache/carbondata/core/scan/result/vector/CarbonDictionary.java
@@ -22,6 +22,10 @@ public interface CarbonDictionary {
int getDictionarySize();
+ boolean isDictionaryUsed();
+
+ void setDictionaryUsed();
+
byte[] getDictionaryValue(int index);
byte[][] getAllDictionaryValues();
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/scan/result/vector/impl/CarbonDictionaryImpl.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/result/vector/impl/CarbonDictionaryImpl.java b/core/src/main/java/org/apache/carbondata/core/scan/result/vector/impl/CarbonDictionaryImpl.java
index 962a772..135ec3f 100644
--- a/core/src/main/java/org/apache/carbondata/core/scan/result/vector/impl/CarbonDictionaryImpl.java
+++ b/core/src/main/java/org/apache/carbondata/core/scan/result/vector/impl/CarbonDictionaryImpl.java
@@ -30,6 +30,8 @@ public class CarbonDictionaryImpl implements CarbonDictionary {
private int actualSize;
+ private boolean isDictUsed;
+
public CarbonDictionaryImpl(byte[][] dictionary, int actualSize) {
this.dictionary = dictionary;
this.actualSize = actualSize;
@@ -43,6 +45,14 @@ public class CarbonDictionaryImpl implements CarbonDictionary {
return this.dictionary.length;
}
+ @Override public boolean isDictionaryUsed() {
+ return this.isDictUsed;
+ }
+
+ @Override public void setDictionaryUsed() {
+ this.isDictUsed = true;
+ }
+
@Override public byte[] getDictionaryValue(int index) {
return dictionary[index];
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/scan/scanner/LazyPageLoader.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/scan/scanner/LazyPageLoader.java b/core/src/main/java/org/apache/carbondata/core/scan/scanner/LazyPageLoader.java
index 42af6c3..0aca79a 100644
--- a/core/src/main/java/org/apache/carbondata/core/scan/scanner/LazyPageLoader.java
+++ b/core/src/main/java/org/apache/carbondata/core/scan/scanner/LazyPageLoader.java
@@ -18,6 +18,7 @@ package org.apache.carbondata.core.scan.scanner;
import java.io.IOException;
+import org.apache.carbondata.core.datastore.ReusableDataBuffer;
import org.apache.carbondata.core.datastore.chunk.impl.DimensionRawColumnChunk;
import org.apache.carbondata.core.datastore.chunk.impl.MeasureRawColumnChunk;
import org.apache.carbondata.core.scan.result.vector.ColumnVectorInfo;
@@ -43,14 +44,17 @@ public class LazyPageLoader {
private QueryStatisticsModel queryStatisticsModel;
+ private ReusableDataBuffer reusableDataBuffer;
+
public LazyPageLoader(LazyBlockletLoader lazyBlockletLoader, int index, boolean isMeasure,
- int pageNumber, ColumnVectorInfo vectorInfo) {
+ int pageNumber, ColumnVectorInfo vectorInfo, ReusableDataBuffer reusableDataBuffer) {
this.lazyBlockletLoader = lazyBlockletLoader;
this.lazyChunkWrapper = lazyBlockletLoader.getLazyChunkWrapper(index, isMeasure);
this.isMeasure = isMeasure;
this.pageNumber = pageNumber;
this.vectorInfo = vectorInfo;
this.queryStatisticsModel = lazyBlockletLoader.getQueryStatisticsModel();
+ this.reusableDataBuffer = reusableDataBuffer;
}
public void loadPage() {
@@ -64,10 +68,10 @@ public class LazyPageLoader {
long startTime = System.currentTimeMillis();
if (isMeasure) {
((MeasureRawColumnChunk) lazyChunkWrapper.getRawColumnChunk())
- .convertToColumnPageAndFillVector(pageNumber, vectorInfo);
+ .convertToColumnPageAndFillVector(pageNumber, vectorInfo, reusableDataBuffer);
} else {
((DimensionRawColumnChunk) lazyChunkWrapper.getRawColumnChunk())
- .convertToDimColDataChunkAndFillVector(pageNumber, vectorInfo);
+ .convertToDimColDataChunkAndFillVector(pageNumber, vectorInfo, reusableDataBuffer);
}
if (queryStatisticsModel.isEnabled()) {
QueryStatistic pageUncompressTime = queryStatisticsModel.getStatisticsTypeAndObjMap()
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/test/java/org/apache/carbondata/core/datastore/chunk/impl/FixedLengthDimensionDataChunkTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/carbondata/core/datastore/chunk/impl/FixedLengthDimensionDataChunkTest.java b/core/src/test/java/org/apache/carbondata/core/datastore/chunk/impl/FixedLengthDimensionDataChunkTest.java
index 8dabe4c..e6f3d9a 100644
--- a/core/src/test/java/org/apache/carbondata/core/datastore/chunk/impl/FixedLengthDimensionDataChunkTest.java
+++ b/core/src/test/java/org/apache/carbondata/core/datastore/chunk/impl/FixedLengthDimensionDataChunkTest.java
@@ -38,7 +38,8 @@ public class FixedLengthDimensionDataChunkTest {
int invertedIndexReverse[] = { 1, 0, 5, 7, 8 };
fixedLengthDimensionDataChunk =
- new FixedLengthDimensionColumnPage(data, invertedIndex, invertedIndexReverse, 5, 4);
+ new FixedLengthDimensionColumnPage(data, invertedIndex, invertedIndexReverse, 5, 4,
+ data.length);
}
@Test public void fillChunkDataTest() {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/test/java/org/apache/carbondata/core/scan/filter/executer/IncludeFilterExecuterImplTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/carbondata/core/scan/filter/executer/IncludeFilterExecuterImplTest.java b/core/src/test/java/org/apache/carbondata/core/scan/filter/executer/IncludeFilterExecuterImplTest.java
index 2662cee..ee3bc16 100644
--- a/core/src/test/java/org/apache/carbondata/core/scan/filter/executer/IncludeFilterExecuterImplTest.java
+++ b/core/src/test/java/org/apache/carbondata/core/scan/filter/executer/IncludeFilterExecuterImplTest.java
@@ -184,7 +184,7 @@ public class IncludeFilterExecuterImplTest extends TestCase {
dim.setFilterKeys(filterKeys);
dimensionColumnDataChunk = new FixedLengthDimensionColumnPage(dataChunk, null, null,
- dataChunkSize, dimColumnSize);
+ dataChunkSize, dimColumnSize, dataChunk.length);
// repeat query and compare 2 result between old code and new optimized code
for (int j = 0; j < queryTimes; j++) {
@@ -304,7 +304,7 @@ public class IncludeFilterExecuterImplTest extends TestCase {
dim.setFilterKeys(filterKeys);
dimensionColumnDataChunk = new FixedLengthDimensionColumnPage(dataChunk, null, null,
- dataChunk.length / dimColumnSize, dimColumnSize);
+ dataChunk.length / dimColumnSize, dimColumnSize, dataChunk.length);
// initial to run
BitSet bitOld = this.setFilterdIndexToBitSetWithColumnIndexOld(dimensionColumnDataChunk, dataChunkSize, filterKeys);
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/test/java/org/apache/carbondata/core/util/CarbonUtilTest.java
----------------------------------------------------------------------
diff --git a/core/src/test/java/org/apache/carbondata/core/util/CarbonUtilTest.java b/core/src/test/java/org/apache/carbondata/core/util/CarbonUtilTest.java
index 4a5d1b3..a82a8aa 100644
--- a/core/src/test/java/org/apache/carbondata/core/util/CarbonUtilTest.java
+++ b/core/src/test/java/org/apache/carbondata/core/util/CarbonUtilTest.java
@@ -247,7 +247,7 @@ public class CarbonUtilTest {
byte[] dataChunks = { 5, 6, 7, 8, 9 };
byte[] compareValues = { 7 };
FixedLengthDimensionColumnPage fixedLengthDataChunk =
- new FixedLengthDimensionColumnPage(dataChunks, null, null, 5, 1);
+ new FixedLengthDimensionColumnPage(dataChunks, null, null, 5, 1, dataChunks.length);
int result = CarbonUtil.nextLesserValueToTarget(2, fixedLengthDataChunk, compareValues);
assertEquals(result, 1);
}
@@ -256,7 +256,7 @@ public class CarbonUtilTest {
byte[] dataChunks = { 7, 7, 7, 8, 9 };
byte[] compareValues = { 7 };
FixedLengthDimensionColumnPage fixedLengthDataChunk =
- new FixedLengthDimensionColumnPage(dataChunks, null, null, 5, 1);
+ new FixedLengthDimensionColumnPage(dataChunks, null, null, 5, 1, dataChunks.length);
int result = CarbonUtil.nextLesserValueToTarget(2, fixedLengthDataChunk, compareValues);
assertEquals(result, -1);
}
@@ -265,7 +265,7 @@ public class CarbonUtilTest {
byte[] dataChunks = { 5, 6, 7, 8, 9 };
byte[] compareValues = { 7 };
FixedLengthDimensionColumnPage fixedLengthDataChunk =
- new FixedLengthDimensionColumnPage(dataChunks, null, null, 5, 1);
+ new FixedLengthDimensionColumnPage(dataChunks, null, null, 5, 1, dataChunks.length);
int result = CarbonUtil.nextGreaterValueToTarget(2, fixedLengthDataChunk, compareValues, 5);
assertEquals(result, 3);
}
@@ -282,7 +282,7 @@ public class CarbonUtilTest {
byte[] dataChunks = { 5, 6, 7, 7, 7 };
byte[] compareValues = { 7 };
FixedLengthDimensionColumnPage fixedLengthDataChunk =
- new FixedLengthDimensionColumnPage(dataChunks, null, null, 5, 1);
+ new FixedLengthDimensionColumnPage(dataChunks, null, null, 5, 1, dataChunks.length);
int result = CarbonUtil.nextGreaterValueToTarget(2, fixedLengthDataChunk, compareValues, 5);
assertEquals(result, 5);
}
@@ -772,7 +772,7 @@ public class CarbonUtilTest {
byte[] dataChunks = { 10, 20, 30, 40, 50, 60 };
byte[] compareValue = { 5 };
FixedLengthDimensionColumnPage fixedLengthDimensionDataChunk =
- new FixedLengthDimensionColumnPage(dataChunks, null, null, 6, 1);
+ new FixedLengthDimensionColumnPage(dataChunks, null, null, 6, 1, dataChunks.length);
int result = CarbonUtil
.getFirstIndexUsingBinarySearch(fixedLengthDimensionDataChunk, 1, 3, compareValue, false);
assertEquals(-2, result);
@@ -782,7 +782,7 @@ public class CarbonUtilTest {
byte[] dataChunks = { 10, 20, 30, 40, 50, 60 };
byte[] compareValue = { 30 };
FixedLengthDimensionColumnPage fixedLengthDimensionDataChunk =
- new FixedLengthDimensionColumnPage(dataChunks, null, null, 6, 1);
+ new FixedLengthDimensionColumnPage(dataChunks, null, null, 6, 1, dataChunks.length);
int result = CarbonUtil
.getFirstIndexUsingBinarySearch(fixedLengthDimensionDataChunk, 1, 3, compareValue, false);
assertEquals(2, result);
@@ -792,7 +792,7 @@ public class CarbonUtilTest {
byte[] dataChunks = { 10, 10, 10, 40, 50, 60 };
byte[] compareValue = { 10 };
FixedLengthDimensionColumnPage fixedLengthDimensionDataChunk =
- new FixedLengthDimensionColumnPage(dataChunks, null, null, 6, 1);
+ new FixedLengthDimensionColumnPage(dataChunks, null, null, 6, 1, dataChunks.length);
int result = CarbonUtil
.getFirstIndexUsingBinarySearch(fixedLengthDimensionDataChunk, 1, 3, compareValue, false);
assertEquals(0, result);
@@ -802,7 +802,7 @@ public class CarbonUtilTest {
byte[] dataChunks = { 10, 10, 10, 40, 50, 60 };
byte[] compareValue = { 10 };
FixedLengthDimensionColumnPage fixedLengthDimensionDataChunk =
- new FixedLengthDimensionColumnPage(dataChunks, null, null, 6, 1);
+ new FixedLengthDimensionColumnPage(dataChunks, null, null, 6, 1, dataChunks.length);
int result = CarbonUtil
.getFirstIndexUsingBinarySearch(fixedLengthDimensionDataChunk, 1, 3, compareValue, true);
assertEquals(2, result);
@@ -819,7 +819,7 @@ public class CarbonUtilTest {
dataChunk = "abbcccddddeffgggh".getBytes();
byte[][] dataArr = new byte[dataChunk.length / keyWord.length][keyWord.length];
fixedLengthDimensionDataChunk = new FixedLengthDimensionColumnPage(dataChunk, null, null,
- dataChunk.length / keyWord.length, keyWord.length);
+ dataChunk.length / keyWord.length, keyWord.length, dataChunk.length);
for (int ii = 0; ii < dataChunk.length / keyWord.length; ii++) {
dataArr[ii] = fixedLengthDimensionDataChunk.getChunkData(ii);
@@ -851,7 +851,7 @@ public class CarbonUtilTest {
dataChunk = "ab".getBytes();
fixedLengthDimensionDataChunk = new FixedLengthDimensionColumnPage(dataChunk, null, null,
- dataChunk.length / keyWord.length, keyWord.length);
+ dataChunk.length / keyWord.length, keyWord.length, dataChunk.length);
keyWord[0] = Byte.valueOf("97");
range = CarbonUtil.getRangeIndexUsingBinarySearch(fixedLengthDimensionDataChunk, 0, dataChunk.length - 1, keyWord);
@@ -865,7 +865,7 @@ public class CarbonUtilTest {
dataChunk = "aabb".getBytes();
fixedLengthDimensionDataChunk = new FixedLengthDimensionColumnPage(dataChunk, null, null,
- dataChunk.length / keyWord.length, keyWord.length);
+ dataChunk.length / keyWord.length, keyWord.length, dataChunk.length);
keyWord[0] = Byte.valueOf("97");
range = CarbonUtil.getRangeIndexUsingBinarySearch(fixedLengthDimensionDataChunk, 0, dataChunk.length - 1, keyWord);
@@ -879,7 +879,7 @@ public class CarbonUtilTest {
dataChunk = "a".getBytes();
fixedLengthDimensionDataChunk = new FixedLengthDimensionColumnPage(dataChunk, null, null,
- dataChunk.length / keyWord.length, keyWord.length);
+ dataChunk.length / keyWord.length, keyWord.length, dataChunk.length);
keyWord[0] = Byte.valueOf("97");
range = CarbonUtil.getRangeIndexUsingBinarySearch(fixedLengthDimensionDataChunk, 0, dataChunk.length - 1, keyWord);
@@ -888,7 +888,7 @@ public class CarbonUtilTest {
dataChunk = "aa".getBytes();
fixedLengthDimensionDataChunk = new FixedLengthDimensionColumnPage(dataChunk, null, null,
- dataChunk.length / keyWord.length, keyWord.length);
+ dataChunk.length / keyWord.length, keyWord.length, dataChunk.length);
keyWord[0] = Byte.valueOf("97");
range = CarbonUtil.getRangeIndexUsingBinarySearch(fixedLengthDimensionDataChunk, 0, dataChunk.length - 1, keyWord);
@@ -897,7 +897,7 @@ public class CarbonUtilTest {
dataChunk = "aabbbbbbbbbbcc".getBytes();
fixedLengthDimensionDataChunk = new FixedLengthDimensionColumnPage(dataChunk, null, null,
- dataChunk.length / keyWord.length, keyWord.length);
+ dataChunk.length / keyWord.length, keyWord.length, dataChunk.length);
keyWord[0] = Byte.valueOf("98");
range = CarbonUtil.getRangeIndexUsingBinarySearch(fixedLengthDimensionDataChunk, 0, dataChunk.length - 1, keyWord);
assertEquals(2, range[0]);
@@ -917,7 +917,7 @@ public class CarbonUtilTest {
byte[][] dataArr = new byte[dataChunk.length / keyWord.length][keyWord.length];
fixedLengthDimensionDataChunk = new FixedLengthDimensionColumnPage(dataChunk, null, null,
- dataChunk.length / keyWord.length, keyWord.length);
+ dataChunk.length / keyWord.length, keyWord.length, dataChunk.length);
for (int ii = 0; ii < dataChunk.length / keyWord.length; ii++) {
dataArr[ii] = fixedLengthDimensionDataChunk.getChunkData(ii);
@@ -959,7 +959,7 @@ public class CarbonUtilTest {
byte[][] dataArr = new byte[dataChunk.length / keyWord.length][keyWord.length];
fixedLengthDimensionDataChunk = new FixedLengthDimensionColumnPage(dataChunk, null, null,
- dataChunk.length / keyWord.length, keyWord.length);
+ dataChunk.length / keyWord.length, keyWord.length, dataChunk.length);
for (int ii = 0; ii < dataChunk.length / keyWord.length; ii++) {
dataArr[ii] = fixedLengthDimensionDataChunk.getChunkData(ii);
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/dataload/TestLoadDataWithCompression.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/dataload/TestLoadDataWithCompression.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/dataload/TestLoadDataWithCompression.scala
index f5153fc..6ec44e7 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/dataload/TestLoadDataWithCompression.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/dataload/TestLoadDataWithCompression.scala
@@ -147,6 +147,21 @@ class CustomizeCompressor extends Compressor {
override def supportUnsafe(): Boolean = {
false
}
+
+ override def unCompressedLength(data: Array[Byte],
+ offset: Int,
+ length: Int): Int = {
+ throw new RuntimeException("Unsupported operation Exception")
+ }
+
+ override def rawUncompress(data: Array[Byte],
+ offset: Int,
+ length: Int,
+ output: Array[Byte]): Int = {
+ throw new RuntimeException("Unsupported operation Exception")
+ }
+
+ override def supportReusableBuffer(): Boolean = false
}
class TestLoadDataWithCompression extends QueryTest with BeforeAndAfterEach with BeforeAndAfterAll {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/integration/spark-datasource/src/main/scala/org/apache/carbondata/spark/vectorreader/VectorizedCarbonRecordReader.java
----------------------------------------------------------------------
diff --git a/integration/spark-datasource/src/main/scala/org/apache/carbondata/spark/vectorreader/VectorizedCarbonRecordReader.java b/integration/spark-datasource/src/main/scala/org/apache/carbondata/spark/vectorreader/VectorizedCarbonRecordReader.java
index 8082511..1f28b8c 100644
--- a/integration/spark-datasource/src/main/scala/org/apache/carbondata/spark/vectorreader/VectorizedCarbonRecordReader.java
+++ b/integration/spark-datasource/src/main/scala/org/apache/carbondata/spark/vectorreader/VectorizedCarbonRecordReader.java
@@ -365,7 +365,6 @@ public class VectorizedCarbonRecordReader extends AbstractRecordReader<Object> {
for (int i = 0; i < isNoDictStringField.length; i++) {
if (isNoDictStringField[i]) {
vectorProxy.resetDictionaryIds(i);
- vectorProxy.column(i).setDictionary(null);
}
}
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/tools/cli/src/main/java/org/apache/carbondata/tool/ScanBenchmark.java
----------------------------------------------------------------------
diff --git a/tools/cli/src/main/java/org/apache/carbondata/tool/ScanBenchmark.java b/tools/cli/src/main/java/org/apache/carbondata/tool/ScanBenchmark.java
index 12384fc..07682ed 100644
--- a/tools/cli/src/main/java/org/apache/carbondata/tool/ScanBenchmark.java
+++ b/tools/cli/src/main/java/org/apache/carbondata/tool/ScanBenchmark.java
@@ -205,7 +205,7 @@ class ScanBenchmark implements Command {
DimensionColumnPage[] pages = new DimensionColumnPage[numPages];
for (int i = 0; i < pages.length; i++) {
pages[i] = dimensionColumnChunkReader.decodeColumnPage(
- (DimensionRawColumnChunk) rawColumnChunk, i);
+ (DimensionRawColumnChunk) rawColumnChunk, i, null);
}
return pages;
}
@@ -215,7 +215,7 @@ class ScanBenchmark implements Command {
ColumnPage[] pages = new ColumnPage[numPages];
for (int i = 0; i < pages.length; i++) {
pages[i] = measureColumnChunkReader.decodeColumnPage(
- (MeasureRawColumnChunk) rawColumnChunk, i);
+ (MeasureRawColumnChunk) rawColumnChunk, i, null);
}
return pages;
}
[2/2] carbondata git commit: [CARBONDATA-3113] Fixed Local Dictionary
Query Performance and Added reusable buffer for direct flow
Posted by ma...@apache.org.
[CARBONDATA-3113] Fixed Local Dictionary Query Performance and Added reusable buffer for direct flow
Following optimizations done in the PR.
1. Added reusable buffer for direct flow In query for each page each column it is creating a byte array, when number of columns are high it is
causing lots of minor gc and degrading query performance, as each page is getting uncompressed one by one we can
use same buffer for all the columns and based on requested size it will resize.
2. Fixed Local Dictionary performance issue.Reverted back #2895 and fixed NPE issue by setting null for local
dictionary to vector In safe and Unsafe VariableLengthDataChunkStore
This closes #2872
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/d79ba999
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/d79ba999
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/d79ba999
Branch: refs/heads/master
Commit: d79ba999f2f683da4be7554a8aba67b9dac01975
Parents: 647bfba
Author: kumarvishal09 <ku...@gmail.com>
Authored: Sun Oct 28 20:39:05 2018 +0530
Committer: manishgupta88 <to...@gmail.com>
Committed: Wed Nov 21 14:45:00 2018 +0530
----------------------------------------------------------------------
.../core/datastore/ReusableDataBuffer.java | 55 ++++++++++++++++++++
.../chunk/impl/DimensionRawColumnChunk.java | 15 +++---
.../impl/FixedLengthDimensionColumnPage.java | 16 +++---
.../chunk/impl/MeasureRawColumnChunk.java | 15 +++---
.../impl/VariableLengthDimensionColumnPage.java | 16 +++---
.../reader/DimensionColumnChunkReader.java | 6 ++-
.../chunk/reader/MeasureColumnChunkReader.java | 10 ++--
.../reader/dimension/AbstractChunkReader.java | 4 +-
...mpressedDimensionChunkFileBasedReaderV1.java | 28 ++++++----
...mpressedDimensionChunkFileBasedReaderV2.java | 26 ++++++---
...essedDimChunkFileBasedPageLevelReaderV3.java | 7 ++-
...mpressedDimensionChunkFileBasedReaderV3.java | 54 ++++++++++++-------
.../measure/AbstractMeasureChunkReader.java | 7 +--
...CompressedMeasureChunkFileBasedReaderV1.java | 5 +-
...CompressedMeasureChunkFileBasedReaderV2.java | 12 +++--
...CompressedMeasureChunkFileBasedReaderV3.java | 34 ++++++------
...essedMsrChunkFileBasedPageLevelReaderV3.java | 6 ++-
.../chunk/store/DimensionChunkStoreFactory.java | 18 ++++---
.../impl/LocalDictDimensionDataChunkStore.java | 17 ++++--
.../safe/AbstractNonDictionaryVectorFiller.java | 23 +++++---
...ariableIntLengthDimensionDataChunkStore.java | 5 +-
...feVariableLengthDimensionDataChunkStore.java | 19 ++++---
...iableShortLengthDimensionDataChunkStore.java | 5 +-
.../UnsafeAbstractDimensionDataChunkStore.java | 4 +-
...nsafeFixedLengthDimensionDataChunkStore.java | 4 +-
...ariableIntLengthDimensionDataChunkStore.java | 4 +-
...feVariableLengthDimensionDataChunkStore.java | 5 +-
...iableShortLengthDimensionDataChunkStore.java | 4 +-
.../core/datastore/columnar/UnBlockIndexer.java | 4 +-
.../compression/AbstractCompressor.java | 3 ++
.../core/datastore/compression/Compressor.java | 6 +++
.../datastore/compression/SnappyCompressor.java | 20 +++++++
.../datastore/compression/ZstdCompressor.java | 8 +++
.../core/datastore/page/ColumnPage.java | 3 +-
.../page/DecoderBasedFallbackEncoder.java | 3 +-
.../datastore/page/VarLengthColumnPageBase.java | 8 +--
.../page/encoding/ColumnPageDecoder.java | 7 ++-
.../adaptive/AdaptiveDeltaFloatingCodec.java | 19 ++++---
.../adaptive/AdaptiveDeltaIntegralCodec.java | 22 +++++---
.../adaptive/AdaptiveFloatingCodec.java | 22 +++++---
.../adaptive/AdaptiveIntegralCodec.java | 20 ++++---
.../encoding/compress/DirectCompressCodec.java | 23 +++++---
.../datastore/page/encoding/rle/RLECodec.java | 10 ++--
.../executor/impl/AbstractQueryExecutor.java | 45 +++++++++++++---
.../scan/executor/infos/BlockExecutionInfo.java | 21 ++++++++
.../core/scan/result/BlockletScannedResult.java | 23 +++++---
.../scan/result/vector/CarbonDictionary.java | 4 ++
.../vector/impl/CarbonDictionaryImpl.java | 10 ++++
.../core/scan/scanner/LazyPageLoader.java | 10 ++--
.../impl/FixedLengthDimensionDataChunkTest.java | 3 +-
.../executer/IncludeFilterExecuterImplTest.java | 4 +-
.../carbondata/core/util/CarbonUtilTest.java | 32 ++++++------
.../dataload/TestLoadDataWithCompression.scala | 15 ++++++
.../VectorizedCarbonRecordReader.java | 1 -
.../apache/carbondata/tool/ScanBenchmark.java | 4 +-
55 files changed, 546 insertions(+), 228 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/datastore/ReusableDataBuffer.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/ReusableDataBuffer.java b/core/src/main/java/org/apache/carbondata/core/datastore/ReusableDataBuffer.java
new file mode 100644
index 0000000..e0c234a
--- /dev/null
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/ReusableDataBuffer.java
@@ -0,0 +1,55 @@
+/*
+ * 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;
+
+import org.apache.carbondata.common.annotations.InterfaceAudience;
+import org.apache.carbondata.common.annotations.InterfaceStability;
+
+/**
+ * class holds the reusable data buffer based on request it will resize.
+ * If request size is less it will return the same buffer, in case it is more
+ * it will resize and update the buffer
+ */
+@InterfaceAudience.Internal
+@InterfaceStability.Evolving
+public class ReusableDataBuffer {
+ /**
+ * reusable byte array
+ */
+ private byte[] dataBuffer;
+
+ /**
+ * current size of data buffer
+ */
+ private int size;
+
+ /**
+ * below method will be used to get the data buffer based on size
+ * If requested size is less it will return same buffer, if size is more
+ * it resize the buffer and return
+ * @param requestedSize
+ * @return databuffer
+ */
+ public byte[] getDataBuffer(int requestedSize) {
+ if (dataBuffer == null || requestedSize > size) {
+ this.size = requestedSize + ((requestedSize * 30) / 100);
+ dataBuffer = new byte[size];
+ }
+ return dataBuffer;
+ }
+}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/datastore/chunk/impl/DimensionRawColumnChunk.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/impl/DimensionRawColumnChunk.java b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/impl/DimensionRawColumnChunk.java
index d84434e..9d798bd 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/impl/DimensionRawColumnChunk.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/impl/DimensionRawColumnChunk.java
@@ -23,6 +23,7 @@ import java.util.List;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.datastore.FileReader;
+import org.apache.carbondata.core.datastore.ReusableDataBuffer;
import org.apache.carbondata.core.datastore.chunk.AbstractRawColumnChunk;
import org.apache.carbondata.core.datastore.chunk.DimensionColumnPage;
import org.apache.carbondata.core.datastore.chunk.reader.DimensionColumnChunkReader;
@@ -72,7 +73,7 @@ public class DimensionRawColumnChunk extends AbstractRawColumnChunk {
for (int i = 0; i < pagesCount; i++) {
try {
if (dataChunks[i] == null) {
- dataChunks[i] = chunkReader.decodeColumnPage(this, i);
+ dataChunks[i] = chunkReader.decodeColumnPage(this, i, null);
}
} catch (IOException | MemoryException e) {
throw new RuntimeException(e);
@@ -93,7 +94,7 @@ public class DimensionRawColumnChunk extends AbstractRawColumnChunk {
}
if (dataChunks[pageNumber] == null) {
try {
- dataChunks[pageNumber] = chunkReader.decodeColumnPage(this, pageNumber);
+ dataChunks[pageNumber] = chunkReader.decodeColumnPage(this, pageNumber, null);
} catch (IOException | MemoryException e) {
throw new RuntimeException(e);
}
@@ -108,7 +109,8 @@ public class DimensionRawColumnChunk extends AbstractRawColumnChunk {
* @param index
* @return
*/
- public DimensionColumnPage convertToDimColDataChunkWithOutCache(int index) {
+ public DimensionColumnPage convertToDimColDataChunkWithOutCache(int index,
+ ReusableDataBuffer reusableDataBuffer) {
assert index < pagesCount;
// in case of filter query filter column if filter column is decoded and stored.
// then return the same
@@ -116,7 +118,7 @@ public class DimensionRawColumnChunk extends AbstractRawColumnChunk {
return dataChunks[index];
}
try {
- return chunkReader.decodeColumnPage(this, index);
+ return chunkReader.decodeColumnPage(this, index, reusableDataBuffer);
} catch (Exception e) {
throw new RuntimeException(e);
}
@@ -129,10 +131,11 @@ public class DimensionRawColumnChunk extends AbstractRawColumnChunk {
* @param pageNumber page number to decode and fill the vector
* @param vectorInfo vector to be filled with column page
*/
- public void convertToDimColDataChunkAndFillVector(int pageNumber, ColumnVectorInfo vectorInfo) {
+ public void convertToDimColDataChunkAndFillVector(int pageNumber, ColumnVectorInfo vectorInfo,
+ ReusableDataBuffer reusableDataBuffer) {
assert pageNumber < pagesCount;
try {
- chunkReader.decodeColumnPageAndFillVector(this, pageNumber, vectorInfo);
+ chunkReader.decodeColumnPageAndFillVector(this, pageNumber, vectorInfo, reusableDataBuffer);
} catch (Exception e) {
throw new RuntimeException(e);
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/datastore/chunk/impl/FixedLengthDimensionColumnPage.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/impl/FixedLengthDimensionColumnPage.java b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/impl/FixedLengthDimensionColumnPage.java
index e650e0e..9639e9b 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/impl/FixedLengthDimensionColumnPage.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/impl/FixedLengthDimensionColumnPage.java
@@ -39,14 +39,14 @@ public class FixedLengthDimensionColumnPage extends AbstractDimensionColumnPage
* @param columnValueSize size of each column value
*/
public FixedLengthDimensionColumnPage(byte[] dataChunk, int[] invertedIndex,
- int[] invertedIndexReverse, int numberOfRows, int columnValueSize) {
+ int[] invertedIndexReverse, int numberOfRows, int columnValueSize, int dataLength) {
boolean isExplicitSorted = isExplicitSorted(invertedIndex);
long totalSize = isExplicitSorted ?
- dataChunk.length + (2 * numberOfRows * CarbonCommonConstants.INT_SIZE_IN_BYTE) :
- dataChunk.length;
+ dataLength + (2 * numberOfRows * CarbonCommonConstants.INT_SIZE_IN_BYTE) :
+ dataLength;
dataChunkStore = DimensionChunkStoreFactory.INSTANCE
.getDimensionChunkStore(columnValueSize, isExplicitSorted, numberOfRows, totalSize,
- DimensionStoreType.FIXED_LENGTH, null, false);
+ DimensionStoreType.FIXED_LENGTH, null, false, dataLength);
dataChunkStore.putArray(invertedIndex, invertedIndexReverse, dataChunk);
}
@@ -62,14 +62,14 @@ public class FixedLengthDimensionColumnPage extends AbstractDimensionColumnPage
*/
public FixedLengthDimensionColumnPage(byte[] dataChunk, int[] invertedIndex,
int[] invertedIndexReverse, int numberOfRows, int columnValueSize,
- ColumnVectorInfo vectorInfo) {
+ ColumnVectorInfo vectorInfo, int dataLength) {
boolean isExplicitSorted = isExplicitSorted(invertedIndex);
long totalSize = isExplicitSorted ?
- dataChunk.length + (2 * numberOfRows * CarbonCommonConstants.INT_SIZE_IN_BYTE) :
- dataChunk.length;
+ dataLength + (2 * numberOfRows * CarbonCommonConstants.INT_SIZE_IN_BYTE) :
+ dataLength;
dataChunkStore = DimensionChunkStoreFactory.INSTANCE
.getDimensionChunkStore(columnValueSize, isExplicitSorted, numberOfRows, totalSize,
- DimensionStoreType.FIXED_LENGTH, null, vectorInfo != null);
+ DimensionStoreType.FIXED_LENGTH, null, vectorInfo != null, dataLength);
if (vectorInfo == null) {
dataChunkStore.putArray(invertedIndex, invertedIndexReverse, dataChunk);
} else {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/datastore/chunk/impl/MeasureRawColumnChunk.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/impl/MeasureRawColumnChunk.java b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/impl/MeasureRawColumnChunk.java
index 6a90569..86b5535 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/impl/MeasureRawColumnChunk.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/impl/MeasureRawColumnChunk.java
@@ -20,6 +20,7 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import org.apache.carbondata.core.datastore.FileReader;
+import org.apache.carbondata.core.datastore.ReusableDataBuffer;
import org.apache.carbondata.core.datastore.chunk.AbstractRawColumnChunk;
import org.apache.carbondata.core.datastore.chunk.reader.MeasureColumnChunkReader;
import org.apache.carbondata.core.datastore.page.ColumnPage;
@@ -56,7 +57,7 @@ public class MeasureRawColumnChunk extends AbstractRawColumnChunk {
for (int i = 0; i < pagesCount; i++) {
try {
if (columnPages[i] == null) {
- columnPages[i] = chunkReader.decodeColumnPage(this, i);
+ columnPages[i] = chunkReader.decodeColumnPage(this, i, null);
}
} catch (Exception e) {
throw new RuntimeException(e);
@@ -77,7 +78,7 @@ public class MeasureRawColumnChunk extends AbstractRawColumnChunk {
try {
if (columnPages[pageNumber] == null) {
- columnPages[pageNumber] = chunkReader.decodeColumnPage(this, pageNumber);
+ columnPages[pageNumber] = chunkReader.decodeColumnPage(this, pageNumber, null);
}
} catch (IOException | MemoryException e) {
throw new RuntimeException(e);
@@ -92,7 +93,8 @@ public class MeasureRawColumnChunk extends AbstractRawColumnChunk {
* @param index
* @return
*/
- public ColumnPage convertToColumnPageWithOutCache(int index) {
+ public ColumnPage convertToColumnPageWithOutCache(int index,
+ ReusableDataBuffer reusableDataBuffer) {
assert index < pagesCount;
// in case of filter query filter columns blocklet pages will uncompressed
// so no need to decode again
@@ -100,7 +102,7 @@ public class MeasureRawColumnChunk extends AbstractRawColumnChunk {
return columnPages[index];
}
try {
- return chunkReader.decodeColumnPage(this, index);
+ return chunkReader.decodeColumnPage(this, index, reusableDataBuffer);
} catch (IOException | MemoryException e) {
throw new RuntimeException(e);
}
@@ -113,10 +115,11 @@ public class MeasureRawColumnChunk extends AbstractRawColumnChunk {
* @param pageNumber page number to decode and fill the vector
* @param vectorInfo vector to be filled with column page
*/
- public void convertToColumnPageAndFillVector(int pageNumber, ColumnVectorInfo vectorInfo) {
+ public void convertToColumnPageAndFillVector(int pageNumber, ColumnVectorInfo vectorInfo,
+ ReusableDataBuffer reusableDataBuffer) {
assert pageNumber < pagesCount;
try {
- chunkReader.decodeColumnPageAndFillVector(this, pageNumber, vectorInfo);
+ chunkReader.decodeColumnPageAndFillVector(this, pageNumber, vectorInfo, reusableDataBuffer);
} catch (IOException | MemoryException e) {
throw new RuntimeException(e);
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/datastore/chunk/impl/VariableLengthDimensionColumnPage.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/impl/VariableLengthDimensionColumnPage.java b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/impl/VariableLengthDimensionColumnPage.java
index 6cb8174..8e6e83f 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/impl/VariableLengthDimensionColumnPage.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/impl/VariableLengthDimensionColumnPage.java
@@ -38,9 +38,9 @@ public class VariableLengthDimensionColumnPage extends AbstractDimensionColumnPa
*/
public VariableLengthDimensionColumnPage(byte[] dataChunks, int[] invertedIndex,
int[] invertedIndexReverse, int numberOfRows, DimensionStoreType dimStoreType,
- CarbonDictionary dictionary) {
+ CarbonDictionary dictionary, int dataLength) {
this(dataChunks, invertedIndex, invertedIndexReverse, numberOfRows, dimStoreType, dictionary,
- null);
+ null, dataLength);
}
/**
@@ -54,28 +54,28 @@ public class VariableLengthDimensionColumnPage extends AbstractDimensionColumnPa
*/
public VariableLengthDimensionColumnPage(byte[] dataChunks, int[] invertedIndex,
int[] invertedIndexReverse, int numberOfRows, DimensionStoreType dimStoreType,
- CarbonDictionary dictionary, ColumnVectorInfo vectorInfo) {
+ CarbonDictionary dictionary, ColumnVectorInfo vectorInfo, int dataLength) {
boolean isExplicitSorted = isExplicitSorted(invertedIndex);
long totalSize = 0;
switch (dimStoreType) {
case LOCAL_DICT:
totalSize = null != invertedIndex ?
- (dataChunks.length + (2 * numberOfRows * CarbonCommonConstants.INT_SIZE_IN_BYTE)) :
- dataChunks.length;
+ (dataLength + (2 * numberOfRows * CarbonCommonConstants.INT_SIZE_IN_BYTE)) :
+ dataLength;
break;
case VARIABLE_INT_LENGTH:
case VARIABLE_SHORT_LENGTH:
totalSize = null != invertedIndex ?
- (dataChunks.length + (2 * numberOfRows * CarbonCommonConstants.INT_SIZE_IN_BYTE) + (
+ (dataLength + (2 * numberOfRows * CarbonCommonConstants.INT_SIZE_IN_BYTE) + (
numberOfRows * CarbonCommonConstants.INT_SIZE_IN_BYTE)) :
- (dataChunks.length + (numberOfRows * CarbonCommonConstants.INT_SIZE_IN_BYTE));
+ (dataLength + (numberOfRows * CarbonCommonConstants.INT_SIZE_IN_BYTE));
break;
default:
throw new UnsupportedOperationException("Invalidate dimension store type");
}
dataChunkStore = DimensionChunkStoreFactory.INSTANCE
.getDimensionChunkStore(0, isExplicitSorted, numberOfRows, totalSize, dimStoreType,
- dictionary, vectorInfo != null);
+ dictionary, vectorInfo != null, dataLength);
if (vectorInfo != null) {
dataChunkStore.fillVector(invertedIndex, invertedIndexReverse, dataChunks, vectorInfo);
} else {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/DimensionColumnChunkReader.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/DimensionColumnChunkReader.java b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/DimensionColumnChunkReader.java
index e2d6be7..6ead57b 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/DimensionColumnChunkReader.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/DimensionColumnChunkReader.java
@@ -19,6 +19,7 @@ package org.apache.carbondata.core.datastore.chunk.reader;
import java.io.IOException;
import org.apache.carbondata.core.datastore.FileReader;
+import org.apache.carbondata.core.datastore.ReusableDataBuffer;
import org.apache.carbondata.core.datastore.chunk.DimensionColumnPage;
import org.apache.carbondata.core.datastore.chunk.impl.DimensionRawColumnChunk;
import org.apache.carbondata.core.memory.MemoryException;
@@ -60,11 +61,12 @@ public interface DimensionColumnChunkReader {
* @throws IOException
*/
DimensionColumnPage decodeColumnPage(DimensionRawColumnChunk dimensionRawColumnChunk,
- int pageNumber) throws IOException, MemoryException;
+ int pageNumber, ReusableDataBuffer reusableDataBuffer) throws IOException, MemoryException;
/**
* Decodes the raw data chunk of given page number and fill the vector with decoded data.
*/
void decodeColumnPageAndFillVector(DimensionRawColumnChunk dimensionRawColumnChunk,
- int pageNumber, ColumnVectorInfo vectorInfo) throws IOException, MemoryException;
+ int pageNumber, ColumnVectorInfo vectorInfo, ReusableDataBuffer reusableDataBuffer)
+ throws IOException, MemoryException;
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/MeasureColumnChunkReader.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/MeasureColumnChunkReader.java b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/MeasureColumnChunkReader.java
index 0fbbe6b..46db969 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/MeasureColumnChunkReader.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/MeasureColumnChunkReader.java
@@ -19,6 +19,7 @@ package org.apache.carbondata.core.datastore.chunk.reader;
import java.io.IOException;
import org.apache.carbondata.core.datastore.FileReader;
+import org.apache.carbondata.core.datastore.ReusableDataBuffer;
import org.apache.carbondata.core.datastore.chunk.impl.MeasureRawColumnChunk;
import org.apache.carbondata.core.datastore.page.ColumnPage;
import org.apache.carbondata.core.memory.MemoryException;
@@ -56,13 +57,14 @@ public interface MeasureColumnChunkReader {
* @return
* @throws IOException
*/
- ColumnPage decodeColumnPage(MeasureRawColumnChunk measureRawColumnChunk,
- int pageNumber) throws IOException, MemoryException;
+ ColumnPage decodeColumnPage(MeasureRawColumnChunk measureRawColumnChunk, int pageNumber,
+ ReusableDataBuffer reusableDataBuffer) throws IOException, MemoryException;
/**
* Decode raw data and fill the vector
*/
- void decodeColumnPageAndFillVector(MeasureRawColumnChunk measureRawColumnChunk,
- int pageNumber, ColumnVectorInfo vectorInfo) throws IOException, MemoryException;
+ void decodeColumnPageAndFillVector(MeasureRawColumnChunk measureRawColumnChunk, int pageNumber,
+ ColumnVectorInfo vectorInfo, ReusableDataBuffer reusableDataBuffer)
+ throws IOException, MemoryException;
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/dimension/AbstractChunkReader.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/dimension/AbstractChunkReader.java b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/dimension/AbstractChunkReader.java
index fc225bd..77ef5f5 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/dimension/AbstractChunkReader.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/dimension/AbstractChunkReader.java
@@ -19,6 +19,7 @@ package org.apache.carbondata.core.datastore.chunk.reader.dimension;
import java.io.IOException;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
+import org.apache.carbondata.core.datastore.ReusableDataBuffer;
import org.apache.carbondata.core.datastore.chunk.impl.DimensionRawColumnChunk;
import org.apache.carbondata.core.datastore.chunk.reader.DimensionColumnChunkReader;
import org.apache.carbondata.core.datastore.compression.Compressor;
@@ -86,7 +87,8 @@ public abstract class AbstractChunkReader implements DimensionColumnChunkReader
@Override
public void decodeColumnPageAndFillVector(DimensionRawColumnChunk dimensionRawColumnChunk,
- int pageNumber, ColumnVectorInfo vectorInfo) throws IOException, MemoryException {
+ int pageNumber, ColumnVectorInfo vectorInfo, ReusableDataBuffer reusableDataBuffer)
+ throws IOException, MemoryException {
throw new UnsupportedOperationException(
"This operation is not supported in this reader " + this.getClass().getName());
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/dimension/v1/CompressedDimensionChunkFileBasedReaderV1.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/dimension/v1/CompressedDimensionChunkFileBasedReaderV1.java b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/dimension/v1/CompressedDimensionChunkFileBasedReaderV1.java
index 225b867..d681b54 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/dimension/v1/CompressedDimensionChunkFileBasedReaderV1.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/dimension/v1/CompressedDimensionChunkFileBasedReaderV1.java
@@ -21,6 +21,7 @@ import java.nio.ByteBuffer;
import java.util.List;
import org.apache.carbondata.core.datastore.FileReader;
+import org.apache.carbondata.core.datastore.ReusableDataBuffer;
import org.apache.carbondata.core.datastore.chunk.DimensionColumnPage;
import org.apache.carbondata.core.datastore.chunk.impl.DimensionRawColumnChunk;
import org.apache.carbondata.core.datastore.chunk.impl.FixedLengthDimensionColumnPage;
@@ -101,19 +102,26 @@ public class CompressedDimensionChunkFileBasedReaderV1 extends AbstractChunkRead
return rawColumnChunk;
}
- @Override public DimensionColumnPage decodeColumnPage(
- DimensionRawColumnChunk dimensionRawColumnChunk, int pageNumber) throws IOException {
+ @Override
+ public DimensionColumnPage decodeColumnPage(DimensionRawColumnChunk dimensionRawColumnChunk,
+ int pageNumber, ReusableDataBuffer reusableDataBuffer) throws IOException {
int blockIndex = dimensionRawColumnChunk.getColumnIndex();
byte[] dataPage = null;
int[] invertedIndexes = new int[0];
int[] invertedIndexesReverse = new int[0];
int[] rlePage = null;
FileReader fileReader = dimensionRawColumnChunk.getFileReader();
-
ByteBuffer rawData = dimensionRawColumnChunk.getRawData();
- dataPage = compressor.unCompressByte(rawData.array(), (int) dimensionRawColumnChunk.getOffSet(),
- dimensionRawColumnChunk.getLength());
-
+ int uncompressedSize = compressor
+ .unCompressedLength(rawData.array(), (int) dimensionRawColumnChunk.getOffSet(),
+ dimensionRawColumnChunk.getLength());
+ if (null != reusableDataBuffer) {
+ dataPage = reusableDataBuffer.getDataBuffer(uncompressedSize);
+ } else {
+ dataPage = new byte[uncompressedSize];
+ }
+ compressor.rawUncompress(rawData.array(), (int) dimensionRawColumnChunk.getOffSet(),
+ dimensionRawColumnChunk.getLength(), dataPage);
// if row id block is present then read the row id chunk and uncompress it
DataChunk dataChunk = dimensionColumnChunk.get(blockIndex);
if (CarbonUtil.hasEncoding(dataChunk.getEncodingList(),
@@ -144,7 +152,9 @@ public class CompressedDimensionChunkFileBasedReaderV1 extends AbstractChunkRead
rlePage = numberCompressor
.unCompress(key, 0, dataChunk.getRlePageLength());
// uncompress the data with rle indexes
- dataPage = UnBlockIndexer.uncompressData(dataPage, rlePage, eachColumnValueSize[blockIndex]);
+ dataPage = UnBlockIndexer
+ .uncompressData(dataPage, rlePage, eachColumnValueSize[blockIndex], uncompressedSize);
+ uncompressedSize = dataPage.length;
rlePage = null;
}
// fill chunk attributes
@@ -156,12 +166,12 @@ public class CompressedDimensionChunkFileBasedReaderV1 extends AbstractChunkRead
columnDataChunk =
new VariableLengthDimensionColumnPage(dataPage, invertedIndexes, invertedIndexesReverse,
numberOfRows, DimensionChunkStoreFactory.DimensionStoreType.VARIABLE_SHORT_LENGTH,
- null);
+ null, uncompressedSize);
} else {
// to store fixed length column chunk values
columnDataChunk =
new FixedLengthDimensionColumnPage(dataPage, invertedIndexes, invertedIndexesReverse,
- numberOfRows, eachColumnValueSize[blockIndex]);
+ numberOfRows, eachColumnValueSize[blockIndex], uncompressedSize);
}
return columnDataChunk;
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/dimension/v2/CompressedDimensionChunkFileBasedReaderV2.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/dimension/v2/CompressedDimensionChunkFileBasedReaderV2.java b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/dimension/v2/CompressedDimensionChunkFileBasedReaderV2.java
index 4e5285b..de48fc6 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/dimension/v2/CompressedDimensionChunkFileBasedReaderV2.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/dimension/v2/CompressedDimensionChunkFileBasedReaderV2.java
@@ -20,6 +20,7 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import org.apache.carbondata.core.datastore.FileReader;
+import org.apache.carbondata.core.datastore.ReusableDataBuffer;
import org.apache.carbondata.core.datastore.chunk.DimensionColumnPage;
import org.apache.carbondata.core.datastore.chunk.impl.DimensionRawColumnChunk;
import org.apache.carbondata.core.datastore.chunk.impl.FixedLengthDimensionColumnPage;
@@ -118,8 +119,8 @@ public class CompressedDimensionChunkFileBasedReaderV2 extends AbstractChunkRead
return dataChunks;
}
- public DimensionColumnPage decodeColumnPage(
- DimensionRawColumnChunk dimensionRawColumnChunk, int pageNumber) throws IOException {
+ public DimensionColumnPage decodeColumnPage(DimensionRawColumnChunk dimensionRawColumnChunk,
+ int pageNumber, ReusableDataBuffer reusableDataBuffer) throws IOException {
byte[] dataPage = null;
int[] invertedIndexes = new int[0];
int[] invertedIndexesReverse = new int[0];
@@ -144,10 +145,17 @@ public class CompressedDimensionChunkFileBasedReaderV2 extends AbstractChunkRead
CarbonUtil.readDataChunk(rawData, copySourcePoint, dimensionChunksLength.get(blockIndex));
copySourcePoint += dimensionChunksLength.get(blockIndex);
}
-
// first read the data and uncompressed it
- dataPage = compressor
- .unCompressByte(rawData.array(), copySourcePoint, dimensionColumnChunk.data_page_length);
+ int uncompressedSize = compressor.unCompressedLength(rawData.array(), copySourcePoint,
+ dimensionColumnChunk.data_page_length);
+ if (null != reusableDataBuffer) {
+ dataPage = reusableDataBuffer.getDataBuffer(uncompressedSize);
+ } else {
+ dataPage = new byte[uncompressedSize];
+ }
+ compressor
+ .rawUncompress(rawData.array(), copySourcePoint, dimensionColumnChunk.data_page_length,
+ dataPage);
copySourcePoint += dimensionColumnChunk.data_page_length;
// if row id block is present then read the row id chunk and uncompress it
if (CarbonUtil.hasEncoding(dimensionColumnChunk.encoders, Encoding.INVERTED_INDEX)) {
@@ -169,7 +177,9 @@ public class CompressedDimensionChunkFileBasedReaderV2 extends AbstractChunkRead
rawData.get(dataRle);
rlePage = numberCompressor.unCompress(dataRle, 0, dimensionColumnChunk.rle_page_length);
// uncompress the data with rle indexes
- dataPage = UnBlockIndexer.uncompressData(dataPage, rlePage, eachColumnValueSize[blockIndex]);
+ dataPage = UnBlockIndexer
+ .uncompressData(dataPage, rlePage, eachColumnValueSize[blockIndex], uncompressedSize);
+ uncompressedSize = dataPage.length;
}
// fill chunk attributes
DimensionColumnPage columnDataChunk = null;
@@ -180,12 +190,12 @@ public class CompressedDimensionChunkFileBasedReaderV2 extends AbstractChunkRead
columnDataChunk =
new VariableLengthDimensionColumnPage(dataPage, invertedIndexes, invertedIndexesReverse,
numberOfRows, DimensionChunkStoreFactory.DimensionStoreType.VARIABLE_SHORT_LENGTH,
- null);
+ null, uncompressedSize);
} else {
// to store fixed length column chunk values
columnDataChunk =
new FixedLengthDimensionColumnPage(dataPage, invertedIndexes, invertedIndexesReverse,
- numberOfRows, eachColumnValueSize[blockIndex]);
+ numberOfRows, eachColumnValueSize[blockIndex], uncompressedSize);
}
return columnDataChunk;
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/dimension/v3/CompressedDimChunkFileBasedPageLevelReaderV3.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/dimension/v3/CompressedDimChunkFileBasedPageLevelReaderV3.java b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/dimension/v3/CompressedDimChunkFileBasedPageLevelReaderV3.java
index 86a4334..d5ca9ae 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/dimension/v3/CompressedDimChunkFileBasedPageLevelReaderV3.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/dimension/v3/CompressedDimChunkFileBasedPageLevelReaderV3.java
@@ -21,6 +21,7 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import org.apache.carbondata.core.datastore.FileReader;
+import org.apache.carbondata.core.datastore.ReusableDataBuffer;
import org.apache.carbondata.core.datastore.chunk.DimensionColumnPage;
import org.apache.carbondata.core.datastore.chunk.impl.DimensionRawColumnChunk;
import org.apache.carbondata.core.datastore.compression.CompressorFactory;
@@ -140,7 +141,8 @@ public class CompressedDimChunkFileBasedPageLevelReaderV3
* @return DimensionColumnDataChunk
*/
@Override public DimensionColumnPage decodeColumnPage(
- DimensionRawColumnChunk dimensionRawColumnChunk, int pageNumber)
+ DimensionRawColumnChunk dimensionRawColumnChunk, int pageNumber,
+ ReusableDataBuffer reusableDataBuffer)
throws IOException, MemoryException {
// data chunk of page
DataChunk2 pageMetadata = null;
@@ -171,6 +173,7 @@ public class CompressedDimChunkFileBasedPageLevelReaderV3
ByteBuffer rawData = dimensionRawColumnChunk.getFileReader()
.readByteBuffer(filePath, offset, length);
- return decodeDimension(dimensionRawColumnChunk, rawData, pageMetadata, 0, null);
+ return decodeDimension(dimensionRawColumnChunk, rawData, pageMetadata, 0, null,
+ reusableDataBuffer);
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/dimension/v3/CompressedDimensionChunkFileBasedReaderV3.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/dimension/v3/CompressedDimensionChunkFileBasedReaderV3.java b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/dimension/v3/CompressedDimensionChunkFileBasedReaderV3.java
index c85c9ee..5a673fb 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/dimension/v3/CompressedDimensionChunkFileBasedReaderV3.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/dimension/v3/CompressedDimensionChunkFileBasedReaderV3.java
@@ -24,6 +24,7 @@ import java.util.List;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.datastore.FileReader;
+import org.apache.carbondata.core.datastore.ReusableDataBuffer;
import org.apache.carbondata.core.datastore.chunk.DimensionColumnPage;
import org.apache.carbondata.core.datastore.chunk.impl.DimensionRawColumnChunk;
import org.apache.carbondata.core.datastore.chunk.impl.FixedLengthDimensionColumnPage;
@@ -207,14 +208,14 @@ public class CompressedDimensionChunkFileBasedReaderV3 extends AbstractChunkRead
* @param pageNumber number
* @return DimensionColumnPage
*/
- @Override public DimensionColumnPage decodeColumnPage(
- DimensionRawColumnChunk rawColumnPage, int pageNumber) throws IOException, MemoryException {
- return decodeColumnPage(rawColumnPage, pageNumber, null);
+ @Override public DimensionColumnPage decodeColumnPage(DimensionRawColumnChunk rawColumnPage,
+ int pageNumber, ReusableDataBuffer reusableDataBuffer) throws IOException, MemoryException {
+ return decodeColumnPage(rawColumnPage, pageNumber, null, reusableDataBuffer);
}
private DimensionColumnPage decodeColumnPage(
- DimensionRawColumnChunk rawColumnPage, int pageNumber,
- ColumnVectorInfo vectorInfo) throws IOException, MemoryException {
+ DimensionRawColumnChunk rawColumnPage, int pageNumber, ColumnVectorInfo vectorInfo,
+ ReusableDataBuffer reusableDataBuffer) throws IOException, MemoryException {
// data chunk of blocklet column
DataChunk3 dataChunk3 = rawColumnPage.getDataChunkV3();
// get the data buffer
@@ -229,19 +230,22 @@ public class CompressedDimensionChunkFileBasedReaderV3 extends AbstractChunkRead
int offset = (int) rawColumnPage.getOffSet() + dimensionChunksLength
.get(rawColumnPage.getColumnIndex()) + dataChunk3.getPage_offset().get(pageNumber);
// first read the data and uncompressed it
- return decodeDimension(rawColumnPage, rawData, pageMetadata, offset, vectorInfo);
+ return decodeDimension(rawColumnPage, rawData, pageMetadata, offset, vectorInfo,
+ reusableDataBuffer);
}
@Override
public void decodeColumnPageAndFillVector(DimensionRawColumnChunk dimensionRawColumnChunk,
- int pageNumber, ColumnVectorInfo vectorInfo) throws IOException, MemoryException {
+ int pageNumber, ColumnVectorInfo vectorInfo, ReusableDataBuffer reusableDataBuffer)
+ throws IOException, MemoryException {
DimensionColumnPage columnPage =
- decodeColumnPage(dimensionRawColumnChunk, pageNumber, vectorInfo);
+ decodeColumnPage(dimensionRawColumnChunk, pageNumber, vectorInfo, reusableDataBuffer);
columnPage.freeMemory();
}
private ColumnPage decodeDimensionByMeta(DataChunk2 pageMetadata, ByteBuffer pageData, int offset,
- boolean isLocalDictEncodedPage, ColumnVectorInfo vectorInfo, BitSet nullBitSet)
+ boolean isLocalDictEncodedPage, ColumnVectorInfo vectorInfo, BitSet nullBitSet,
+ ReusableDataBuffer reusableDataBuffer)
throws IOException, MemoryException {
List<Encoding> encodings = pageMetadata.getEncoders();
List<ByteBuffer> encoderMetas = pageMetadata.getEncoder_meta();
@@ -252,7 +256,8 @@ public class CompressedDimensionChunkFileBasedReaderV3 extends AbstractChunkRead
if (vectorInfo != null) {
decoder
.decodeAndFillVector(pageData.array(), offset, pageMetadata.data_page_length, vectorInfo,
- nullBitSet, isLocalDictEncodedPage, pageMetadata.numberOfRowsInpage);
+ nullBitSet, isLocalDictEncodedPage, pageMetadata.numberOfRowsInpage,
+ reusableDataBuffer);
return null;
} else {
return decoder
@@ -261,7 +266,8 @@ public class CompressedDimensionChunkFileBasedReaderV3 extends AbstractChunkRead
}
protected DimensionColumnPage decodeDimension(DimensionRawColumnChunk rawColumnPage,
- ByteBuffer pageData, DataChunk2 pageMetadata, int offset, ColumnVectorInfo vectorInfo)
+ ByteBuffer pageData, DataChunk2 pageMetadata, int offset, ColumnVectorInfo vectorInfo,
+ ReusableDataBuffer reusableDataBuffer)
throws IOException, MemoryException {
List<Encoding> encodings = pageMetadata.getEncoders();
org.apache.carbondata.core.metadata.encoder.Encoding.validateEncodingTypes(encodings);
@@ -286,7 +292,7 @@ public class CompressedDimensionChunkFileBasedReaderV3 extends AbstractChunkRead
}
BitSet nullBitSet = QueryUtil.getNullBitSet(pageMetadata.presence, this.compressor);
ColumnPage decodedPage = decodeDimensionByMeta(pageMetadata, pageData, dataOffset,
- null != rawColumnPage.getLocalDictionary(), vectorInfo, nullBitSet);
+ null != rawColumnPage.getLocalDictionary(), vectorInfo, nullBitSet, reusableDataBuffer);
if (decodedPage != null) {
decodedPage.setNullBits(nullBitSet);
}
@@ -294,7 +300,8 @@ public class CompressedDimensionChunkFileBasedReaderV3 extends AbstractChunkRead
invertedIndexesReverse, isEncodedWithAdaptiveMeta(pageMetadata), isExplicitSorted);
} else {
// following code is for backward compatibility
- return decodeDimensionLegacy(rawColumnPage, pageData, pageMetadata, offset, vectorInfo);
+ return decodeDimensionLegacy(rawColumnPage, pageData, pageMetadata, offset, vectorInfo,
+ reusableDataBuffer);
}
}
@@ -314,13 +321,23 @@ public class CompressedDimensionChunkFileBasedReaderV3 extends AbstractChunkRead
}
private DimensionColumnPage decodeDimensionLegacy(DimensionRawColumnChunk rawColumnPage,
- ByteBuffer pageData, DataChunk2 pageMetadata, int offset, ColumnVectorInfo vectorInfo)
+ ByteBuffer pageData, DataChunk2 pageMetadata, int offset, ColumnVectorInfo vectorInfo,
+ ReusableDataBuffer reusableDataBuffer)
throws IOException, MemoryException {
byte[] dataPage;
int[] rlePage;
int[] invertedIndexes = new int[0];
int[] invertedIndexesReverse = new int[0];
- dataPage = compressor.unCompressByte(pageData.array(), offset, pageMetadata.data_page_length);
+ int uncompressedSize = 0;
+ if (null != reusableDataBuffer && compressor.supportReusableBuffer()) {
+ uncompressedSize =
+ compressor.unCompressedLength(pageData.array(), offset, pageMetadata.data_page_length);
+ dataPage = reusableDataBuffer.getDataBuffer(uncompressedSize);
+ compressor.rawUncompress(pageData.array(), offset, pageMetadata.data_page_length, dataPage);
+ } else {
+ dataPage = compressor.unCompressByte(pageData.array(), offset, pageMetadata.data_page_length);
+ uncompressedSize = dataPage.length;
+ }
offset += pageMetadata.data_page_length;
// if row id block is present then read the row id chunk and uncompress it
if (CarbonUtil.hasEncoding(pageMetadata.encoders, Encoding.INVERTED_INDEX)) {
@@ -341,7 +358,8 @@ public class CompressedDimensionChunkFileBasedReaderV3 extends AbstractChunkRead
dataPage = UnBlockIndexer.uncompressData(dataPage, rlePage,
null == rawColumnPage.getLocalDictionary() ?
eachColumnValueSize[rawColumnPage.getColumnIndex()] :
- CarbonCommonConstants.LOCAL_DICT_ENCODED_BYTEARRAY_SIZE);
+ CarbonCommonConstants.LOCAL_DICT_ENCODED_BYTEARRAY_SIZE, uncompressedSize);
+ uncompressedSize = dataPage.length;
}
DimensionColumnPage columnDataChunk = null;
@@ -357,13 +375,13 @@ public class CompressedDimensionChunkFileBasedReaderV3 extends AbstractChunkRead
columnDataChunk =
new VariableLengthDimensionColumnPage(dataPage, invertedIndexes, invertedIndexesReverse,
pageMetadata.getNumberOfRowsInpage(), dimStoreType,
- rawColumnPage.getLocalDictionary(), vectorInfo);
+ rawColumnPage.getLocalDictionary(), vectorInfo, uncompressedSize);
} else {
// to store fixed length column chunk values
columnDataChunk =
new FixedLengthDimensionColumnPage(dataPage, invertedIndexes, invertedIndexesReverse,
pageMetadata.getNumberOfRowsInpage(),
- eachColumnValueSize[rawColumnPage.getColumnIndex()], vectorInfo);
+ eachColumnValueSize[rawColumnPage.getColumnIndex()], vectorInfo, uncompressedSize);
}
return columnDataChunk;
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/measure/AbstractMeasureChunkReader.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/measure/AbstractMeasureChunkReader.java b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/measure/AbstractMeasureChunkReader.java
index cd233d2..91a779e 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/measure/AbstractMeasureChunkReader.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/measure/AbstractMeasureChunkReader.java
@@ -18,6 +18,7 @@ package org.apache.carbondata.core.datastore.chunk.reader.measure;
import java.io.IOException;
+import org.apache.carbondata.core.datastore.ReusableDataBuffer;
import org.apache.carbondata.core.datastore.chunk.impl.MeasureRawColumnChunk;
import org.apache.carbondata.core.datastore.chunk.reader.MeasureColumnChunkReader;
import org.apache.carbondata.core.datastore.compression.Compressor;
@@ -54,9 +55,9 @@ public abstract class AbstractMeasureChunkReader implements MeasureColumnChunkRe
this.numberOfRows = numberOfRows;
}
- @Override
- public void decodeColumnPageAndFillVector(MeasureRawColumnChunk measureRawColumnChunk,
- int pageNumber, ColumnVectorInfo vectorInfo) throws IOException, MemoryException {
+ @Override public void decodeColumnPageAndFillVector(MeasureRawColumnChunk measureRawColumnChunk,
+ int pageNumber, ColumnVectorInfo vectorInfo, ReusableDataBuffer reusableDataBuffer)
+ throws IOException, MemoryException {
throw new UnsupportedOperationException(
"This operation is not supported in this class " + getClass().getName());
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/measure/v1/CompressedMeasureChunkFileBasedReaderV1.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/measure/v1/CompressedMeasureChunkFileBasedReaderV1.java b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/measure/v1/CompressedMeasureChunkFileBasedReaderV1.java
index de9ffec..429c797 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/measure/v1/CompressedMeasureChunkFileBasedReaderV1.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/measure/v1/CompressedMeasureChunkFileBasedReaderV1.java
@@ -21,6 +21,7 @@ import java.nio.ByteBuffer;
import java.util.List;
import org.apache.carbondata.core.datastore.FileReader;
+import org.apache.carbondata.core.datastore.ReusableDataBuffer;
import org.apache.carbondata.core.datastore.chunk.impl.MeasureRawColumnChunk;
import org.apache.carbondata.core.datastore.chunk.reader.measure.AbstractMeasureChunkReader;
import org.apache.carbondata.core.datastore.compression.CompressorFactory;
@@ -92,8 +93,8 @@ public class CompressedMeasureChunkFileBasedReaderV1 extends AbstractMeasureChun
}
@Override
- public ColumnPage decodeColumnPage(MeasureRawColumnChunk measureRawColumnChunk,
- int pageNumber) throws IOException, MemoryException {
+ public ColumnPage decodeColumnPage(MeasureRawColumnChunk measureRawColumnChunk, int pageNumber,
+ ReusableDataBuffer reusableDataBuffer) throws IOException, MemoryException {
int blockIndex = measureRawColumnChunk.getColumnIndex();
DataChunk dataChunk = measureColumnChunks.get(blockIndex);
ValueEncoderMeta meta = dataChunk.getValueEncoderMeta().get(0);
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/measure/v2/CompressedMeasureChunkFileBasedReaderV2.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/measure/v2/CompressedMeasureChunkFileBasedReaderV2.java b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/measure/v2/CompressedMeasureChunkFileBasedReaderV2.java
index 80813e9..d08bfaf 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/measure/v2/CompressedMeasureChunkFileBasedReaderV2.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/measure/v2/CompressedMeasureChunkFileBasedReaderV2.java
@@ -21,6 +21,7 @@ import java.nio.ByteBuffer;
import java.util.List;
import org.apache.carbondata.core.datastore.FileReader;
+import org.apache.carbondata.core.datastore.ReusableDataBuffer;
import org.apache.carbondata.core.datastore.chunk.impl.MeasureRawColumnChunk;
import org.apache.carbondata.core.datastore.chunk.reader.measure.AbstractMeasureChunkReaderV2V3Format;
import org.apache.carbondata.core.datastore.compression.CompressorFactory;
@@ -119,7 +120,7 @@ public class CompressedMeasureChunkFileBasedReaderV2 extends AbstractMeasureChun
}
public ColumnPage decodeColumnPage(MeasureRawColumnChunk measureRawColumnChunk,
- int pageNumber) throws IOException, MemoryException {
+ int pageNumber, ReusableDataBuffer reusableDataBuffer) throws IOException, MemoryException {
int copyPoint = (int) measureRawColumnChunk.getOffSet();
int blockIndex = measureRawColumnChunk.getColumnIndex();
ByteBuffer rawData = measureRawColumnChunk.getRawData();
@@ -127,13 +128,15 @@ public class CompressedMeasureChunkFileBasedReaderV2 extends AbstractMeasureChun
measureColumnChunkLength.get(blockIndex));
copyPoint += measureColumnChunkLength.get(blockIndex);
- ColumnPage page = decodeMeasure(measureRawColumnChunk, measureColumnChunk, copyPoint);
+ ColumnPage page =
+ decodeMeasure(measureRawColumnChunk, measureColumnChunk, copyPoint, reusableDataBuffer);
page.setNullBits(QueryUtil.getNullBitSet(measureColumnChunk.presence, this.compressor));
return page;
}
protected ColumnPage decodeMeasure(MeasureRawColumnChunk measureRawColumnChunk,
- DataChunk2 measureColumnChunk, int copyPoint) throws MemoryException, IOException {
+ DataChunk2 measureColumnChunk, int copyPoint, ReusableDataBuffer reusableDataBuffer)
+ throws MemoryException, IOException {
assert (measureColumnChunk.getEncoder_meta().size() > 0);
List<ByteBuffer> encoder_meta = measureColumnChunk.getEncoder_meta();
byte[] encodedMeta = encoder_meta.get(0).array();
@@ -142,6 +145,7 @@ public class CompressedMeasureChunkFileBasedReaderV2 extends AbstractMeasureChun
ColumnPageDecoder codec = encodingFactory.createDecoderLegacy(meta,
CompressorFactory.NativeSupportedCompressor.SNAPPY.getName());
byte[] rawData = measureRawColumnChunk.getRawData().array();
- return codec.decode(rawData, copyPoint, measureColumnChunk.data_page_length);
+ return codec
+ .decode(rawData, copyPoint, measureColumnChunk.data_page_length);
}
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/measure/v3/CompressedMeasureChunkFileBasedReaderV3.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/measure/v3/CompressedMeasureChunkFileBasedReaderV3.java b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/measure/v3/CompressedMeasureChunkFileBasedReaderV3.java
index 2d3979a..f81c608 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/measure/v3/CompressedMeasureChunkFileBasedReaderV3.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/measure/v3/CompressedMeasureChunkFileBasedReaderV3.java
@@ -22,6 +22,7 @@ import java.util.BitSet;
import java.util.List;
import org.apache.carbondata.core.datastore.FileReader;
+import org.apache.carbondata.core.datastore.ReusableDataBuffer;
import org.apache.carbondata.core.datastore.chunk.impl.MeasureRawColumnChunk;
import org.apache.carbondata.core.datastore.chunk.reader.measure.AbstractMeasureChunkReaderV2V3Format;
import org.apache.carbondata.core.datastore.compression.CompressorFactory;
@@ -188,21 +189,20 @@ public class CompressedMeasureChunkFileBasedReaderV3 extends AbstractMeasureChun
* @param pageNumber number
* @return DimensionColumnPage
*/
- @Override
- public ColumnPage decodeColumnPage(
- MeasureRawColumnChunk rawColumnChunk, int pageNumber)
+ @Override public ColumnPage decodeColumnPage(MeasureRawColumnChunk rawColumnChunk, int pageNumber,
+ ReusableDataBuffer reusableDataBuffer)
throws IOException, MemoryException {
- return decodeColumnPage(rawColumnChunk, pageNumber, null);
+ return decodeColumnPage(rawColumnChunk, pageNumber, null, reusableDataBuffer);
}
- @Override
- public void decodeColumnPageAndFillVector(MeasureRawColumnChunk measureRawColumnChunk,
- int pageNumber, ColumnVectorInfo vectorInfo) throws IOException, MemoryException {
- decodeColumnPage(measureRawColumnChunk, pageNumber, vectorInfo);
+ @Override public void decodeColumnPageAndFillVector(MeasureRawColumnChunk measureRawColumnChunk,
+ int pageNumber, ColumnVectorInfo vectorInfo, ReusableDataBuffer reusableDataBuffer)
+ throws IOException, MemoryException {
+ decodeColumnPage(measureRawColumnChunk, pageNumber, vectorInfo, reusableDataBuffer);
}
- private ColumnPage decodeColumnPage(
- MeasureRawColumnChunk rawColumnChunk, int pageNumber, ColumnVectorInfo vectorInfo)
+ private ColumnPage decodeColumnPage(MeasureRawColumnChunk rawColumnChunk, int pageNumber,
+ ColumnVectorInfo vectorInfo, ReusableDataBuffer reusableDataBuffer)
throws IOException, MemoryException {
// data chunk of blocklet column
DataChunk3 dataChunk3 = rawColumnChunk.getDataChunkV3();
@@ -219,7 +219,8 @@ public class CompressedMeasureChunkFileBasedReaderV3 extends AbstractMeasureChun
dataChunk3.getPage_offset().get(pageNumber);
BitSet nullBitSet = QueryUtil.getNullBitSet(pageMetadata.presence, this.compressor);
ColumnPage decodedPage =
- decodeMeasure(pageMetadata, rawColumnChunk.getRawData(), offset, vectorInfo, nullBitSet);
+ decodeMeasure(pageMetadata, rawColumnChunk.getRawData(), offset, vectorInfo, nullBitSet,
+ reusableDataBuffer);
if (decodedPage == null) {
return null;
}
@@ -231,7 +232,8 @@ public class CompressedMeasureChunkFileBasedReaderV3 extends AbstractMeasureChun
* Decode measure column page with page header and raw data starting from offset
*/
protected ColumnPage decodeMeasure(DataChunk2 pageMetadata, ByteBuffer pageData, int offset,
- ColumnVectorInfo vectorInfo, BitSet nullBitSet) throws MemoryException, IOException {
+ ColumnVectorInfo vectorInfo, BitSet nullBitSet, ReusableDataBuffer reusableDataBuffer)
+ throws MemoryException, IOException {
List<Encoding> encodings = pageMetadata.getEncoders();
org.apache.carbondata.core.metadata.encoder.Encoding.validateEncodingTypes(encodings);
List<ByteBuffer> encoderMetas = pageMetadata.getEncoder_meta();
@@ -240,12 +242,12 @@ public class CompressedMeasureChunkFileBasedReaderV3 extends AbstractMeasureChun
ColumnPageDecoder codec =
encodingFactory.createDecoder(encodings, encoderMetas, compressorName, vectorInfo != null);
if (vectorInfo != null) {
- codec
- .decodeAndFillVector(pageData.array(), offset, pageMetadata.data_page_length, vectorInfo,
- nullBitSet, false, pageMetadata.numberOfRowsInpage);
+ codec.decodeAndFillVector(pageData.array(), offset, pageMetadata.data_page_length, vectorInfo,
+ nullBitSet, false, pageMetadata.numberOfRowsInpage, reusableDataBuffer);
return null;
} else {
- return codec.decode(pageData.array(), offset, pageMetadata.data_page_length);
+ return codec
+ .decode(pageData.array(), offset, pageMetadata.data_page_length);
}
}
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/measure/v3/CompressedMsrChunkFileBasedPageLevelReaderV3.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/measure/v3/CompressedMsrChunkFileBasedPageLevelReaderV3.java b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/measure/v3/CompressedMsrChunkFileBasedPageLevelReaderV3.java
index b092350..b063024 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/measure/v3/CompressedMsrChunkFileBasedPageLevelReaderV3.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/reader/measure/v3/CompressedMsrChunkFileBasedPageLevelReaderV3.java
@@ -22,6 +22,7 @@ import java.nio.ByteBuffer;
import java.util.BitSet;
import org.apache.carbondata.core.datastore.FileReader;
+import org.apache.carbondata.core.datastore.ReusableDataBuffer;
import org.apache.carbondata.core.datastore.chunk.impl.MeasureRawColumnChunk;
import org.apache.carbondata.core.datastore.compression.CompressorFactory;
import org.apache.carbondata.core.datastore.page.ColumnPage;
@@ -135,7 +136,7 @@ public class CompressedMsrChunkFileBasedPageLevelReaderV3
* @return DimensionColumnDataChunk
*/
@Override public ColumnPage decodeColumnPage(
- MeasureRawColumnChunk rawColumnPage, int pageNumber)
+ MeasureRawColumnChunk rawColumnPage, int pageNumber, ReusableDataBuffer reusableDataBuffer)
throws IOException, MemoryException {
// data chunk of blocklet column
DataChunk3 dataChunk3 = rawColumnPage.getDataChunkV3();
@@ -153,7 +154,8 @@ public class CompressedMsrChunkFileBasedPageLevelReaderV3
.readByteBuffer(filePath, offset, pageMetadata.data_page_length);
BitSet nullBitSet = QueryUtil.getNullBitSet(pageMetadata.presence, this.compressor);
- ColumnPage decodedPage = decodeMeasure(pageMetadata, buffer, 0, null, nullBitSet);
+ ColumnPage decodedPage =
+ decodeMeasure(pageMetadata, buffer, 0, null, nullBitSet, reusableDataBuffer);
decodedPage.setNullBits(nullBitSet);
return decodedPage;
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/DimensionChunkStoreFactory.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/DimensionChunkStoreFactory.java b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/DimensionChunkStoreFactory.java
index 5346f35..849a1ee 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/DimensionChunkStoreFactory.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/DimensionChunkStoreFactory.java
@@ -65,22 +65,22 @@ public class DimensionChunkStoreFactory {
*/
public DimensionDataChunkStore getDimensionChunkStore(int columnValueSize,
boolean isInvertedIndex, int numberOfRows, long totalSize, DimensionStoreType storeType,
- CarbonDictionary dictionary, boolean fillDirectVector) {
+ CarbonDictionary dictionary, boolean fillDirectVector, int dataLength) {
if (isUnsafe && !fillDirectVector) {
switch (storeType) {
case FIXED_LENGTH:
return new UnsafeFixedLengthDimensionDataChunkStore(totalSize, columnValueSize,
- isInvertedIndex, numberOfRows);
+ isInvertedIndex, numberOfRows, dataLength);
case VARIABLE_SHORT_LENGTH:
return new UnsafeVariableShortLengthDimensionDataChunkStore(totalSize, isInvertedIndex,
- numberOfRows);
+ numberOfRows, dataLength);
case VARIABLE_INT_LENGTH:
return new UnsafeVariableIntLengthDimensionDataChunkStore(totalSize, isInvertedIndex,
- numberOfRows);
+ numberOfRows, dataLength);
case LOCAL_DICT:
return new LocalDictDimensionDataChunkStore(
new UnsafeFixedLengthDimensionDataChunkStore(totalSize, 3, isInvertedIndex,
- numberOfRows), dictionary);
+ numberOfRows, dataLength), dictionary, dataLength);
default:
throw new UnsupportedOperationException("Invalid dimension store type");
}
@@ -90,13 +90,15 @@ public class DimensionChunkStoreFactory {
return new SafeFixedLengthDimensionDataChunkStore(isInvertedIndex, columnValueSize,
numberOfRows);
case VARIABLE_SHORT_LENGTH:
- return new SafeVariableShortLengthDimensionDataChunkStore(isInvertedIndex, numberOfRows);
+ return new SafeVariableShortLengthDimensionDataChunkStore(isInvertedIndex, numberOfRows,
+ dataLength);
case VARIABLE_INT_LENGTH:
- return new SafeVariableIntLengthDimensionDataChunkStore(isInvertedIndex, numberOfRows);
+ return new SafeVariableIntLengthDimensionDataChunkStore(isInvertedIndex, numberOfRows,
+ dataLength);
case LOCAL_DICT:
return new LocalDictDimensionDataChunkStore(
new SafeFixedLengthDimensionDataChunkStore(isInvertedIndex, 3, numberOfRows),
- dictionary);
+ dictionary, dataLength);
default:
throw new UnsupportedOperationException("Invalid dimension store type");
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/LocalDictDimensionDataChunkStore.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/LocalDictDimensionDataChunkStore.java b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/LocalDictDimensionDataChunkStore.java
index 0eb6d65..c116a24 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/LocalDictDimensionDataChunkStore.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/LocalDictDimensionDataChunkStore.java
@@ -38,10 +38,13 @@ public class LocalDictDimensionDataChunkStore implements DimensionDataChunkStore
private CarbonDictionary dictionary;
+ private int dataLength;
+
public LocalDictDimensionDataChunkStore(DimensionDataChunkStore dimensionDataChunkStore,
- CarbonDictionary dictionary) {
+ CarbonDictionary dictionary, int dataLength) {
this.dimensionDataChunkStore = dimensionDataChunkStore;
this.dictionary = dictionary;
+ this.dataLength = dataLength;
}
/**
@@ -59,9 +62,12 @@ public class LocalDictDimensionDataChunkStore implements DimensionDataChunkStore
public void fillVector(int[] invertedIndex, int[] invertedIndexReverse, byte[] data,
ColumnVectorInfo vectorInfo) {
int columnValueSize = dimensionDataChunkStore.getColumnValueSize();
- int rowsNum = data.length / columnValueSize;
+ int rowsNum = dataLength / columnValueSize;
CarbonColumnVector vector = vectorInfo.vector;
- vector.setDictionary(dictionary);
+ if (!dictionary.isDictionaryUsed()) {
+ vector.setDictionary(dictionary);
+ dictionary.setDictionaryUsed();
+ }
BitSet nullBitset = new BitSet();
CarbonColumnVector dictionaryVector = ColumnarVectorWrapperDirectFactory
.getDirectVectorWrapperFactory(vector.getDictionaryVector(), invertedIndex, nullBitset,
@@ -88,7 +94,10 @@ public class LocalDictDimensionDataChunkStore implements DimensionDataChunkStore
}
@Override public void fillRow(int rowId, CarbonColumnVector vector, int vectorRow) {
- vector.setDictionary(dictionary);
+ if (!dictionary.isDictionaryUsed()) {
+ vector.setDictionary(dictionary);
+ dictionary.setDictionaryUsed();
+ }
int surrogate = dimensionDataChunkStore.getSurrogate(rowId);
if (surrogate == CarbonCommonConstants.MEMBER_DEFAULT_VAL_SURROGATE_KEY) {
vector.putNull(vectorRow);
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/safe/AbstractNonDictionaryVectorFiller.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/safe/AbstractNonDictionaryVectorFiller.java b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/safe/AbstractNonDictionaryVectorFiller.java
index 5e0dfdf..23ac4a9 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/safe/AbstractNonDictionaryVectorFiller.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/safe/AbstractNonDictionaryVectorFiller.java
@@ -44,15 +44,15 @@ public abstract class AbstractNonDictionaryVectorFiller {
class NonDictionaryVectorFillerFactory {
public static AbstractNonDictionaryVectorFiller getVectorFiller(int length, DataType type,
- int numberOfRows) {
+ int numberOfRows, int actualDataLength) {
if (type == DataTypes.STRING) {
if (length > DataTypes.SHORT.getSizeInBytes()) {
- return new LongStringVectorFiller(numberOfRows);
+ return new LongStringVectorFiller(numberOfRows, actualDataLength);
} else {
- return new StringVectorFiller(numberOfRows);
+ return new StringVectorFiller(numberOfRows, actualDataLength);
}
} else if (type == DataTypes.VARCHAR) {
- return new LongStringVectorFiller(numberOfRows);
+ return new LongStringVectorFiller(numberOfRows, actualDataLength);
} else if (type == DataTypes.TIMESTAMP) {
return new TimeStampVectorFiller(numberOfRows);
} else if (type == DataTypes.BOOLEAN) {
@@ -73,8 +73,11 @@ class NonDictionaryVectorFillerFactory {
class StringVectorFiller extends AbstractNonDictionaryVectorFiller {
- public StringVectorFiller(int numberOfRows) {
+ private int actualDataLength;
+
+ public StringVectorFiller(int numberOfRows, int actualDataLength) {
super(numberOfRows);
+ this.actualDataLength = actualDataLength;
}
@Override
@@ -93,13 +96,17 @@ class StringVectorFiller extends AbstractNonDictionaryVectorFiller {
}
localOffset += length;
}
- vector.putAllByteArray(data, 0, data.length);
+ vector.putAllByteArray(data, 0, actualDataLength);
}
}
class LongStringVectorFiller extends AbstractNonDictionaryVectorFiller {
- public LongStringVectorFiller(int numberOfRows) {
+
+ private int actualDataLength;
+
+ public LongStringVectorFiller(int numberOfRows, int actualDataLength) {
super(numberOfRows);
+ this.actualDataLength = actualDataLength;
}
@Override public void fillVector(byte[] data, CarbonColumnVector vector) {
@@ -135,7 +142,7 @@ class LongStringVectorFiller extends AbstractNonDictionaryVectorFiller {
}
localOffset += length;
}
- vector.putAllByteArray(data, 0, data.length);
+ vector.putAllByteArray(data, 0, actualDataLength);
}
}
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/safe/SafeVariableIntLengthDimensionDataChunkStore.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/safe/SafeVariableIntLengthDimensionDataChunkStore.java b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/safe/SafeVariableIntLengthDimensionDataChunkStore.java
index 773f078..8dc4c0b 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/safe/SafeVariableIntLengthDimensionDataChunkStore.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/safe/SafeVariableIntLengthDimensionDataChunkStore.java
@@ -27,8 +27,9 @@ import org.apache.carbondata.core.constants.CarbonCommonConstants;
*/
public class SafeVariableIntLengthDimensionDataChunkStore
extends SafeVariableLengthDimensionDataChunkStore {
- public SafeVariableIntLengthDimensionDataChunkStore(boolean isInvertedIndex, int numberOfRows) {
- super(isInvertedIndex, numberOfRows);
+ public SafeVariableIntLengthDimensionDataChunkStore(boolean isInvertedIndex, int numberOfRows,
+ int dataLength) {
+ super(isInvertedIndex, numberOfRows, dataLength);
}
@Override
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/safe/SafeVariableLengthDimensionDataChunkStore.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/safe/SafeVariableLengthDimensionDataChunkStore.java b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/safe/SafeVariableLengthDimensionDataChunkStore.java
index 3c00fd8..219d8c9 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/safe/SafeVariableLengthDimensionDataChunkStore.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/safe/SafeVariableLengthDimensionDataChunkStore.java
@@ -48,9 +48,14 @@ public abstract class SafeVariableLengthDimensionDataChunkStore
*/
private int[] dataOffsets;
- public SafeVariableLengthDimensionDataChunkStore(boolean isInvertedIndex, int numberOfRows) {
+ private int dataLength;
+
+ public SafeVariableLengthDimensionDataChunkStore(boolean isInvertedIndex, int numberOfRows,
+ int dataLength) {
super(isInvertedIndex);
this.numberOfRows = numberOfRows;
+ this.dataOffsets = new int[numberOfRows];
+ this.dataLength = dataLength;
}
/**
@@ -99,9 +104,10 @@ public abstract class SafeVariableLengthDimensionDataChunkStore
public void fillVector(int[] invertedIndex, int[] invertedIndexReverse, byte[] data,
ColumnVectorInfo vectorInfo) {
CarbonColumnVector vector = vectorInfo.vector;
+ vector.setDictionary(null);
DataType dt = vector.getType();
- AbstractNonDictionaryVectorFiller vectorFiller =
- NonDictionaryVectorFillerFactory.getVectorFiller(getLengthSize(), dt, numberOfRows);
+ AbstractNonDictionaryVectorFiller vectorFiller = NonDictionaryVectorFillerFactory
+ .getVectorFiller(getLengthSize(), dt, numberOfRows, dataLength);
vector = ColumnarVectorWrapperDirectFactory
.getDirectVectorWrapperFactory(vector, invertedIndex, new BitSet(), vectorInfo.deletedRows,
false, false);
@@ -133,7 +139,7 @@ public abstract class SafeVariableLengthDimensionDataChunkStore
length = dataOffsets[rowId + 1] - (currentDataOffset + getLengthSize());
} else {
// for last record
- length = this.data.length - currentDataOffset;
+ length = this.dataLength - currentDataOffset;
}
byte[] currentRowData = new byte[length];
System.arraycopy(data, currentDataOffset, currentRowData, 0, length);
@@ -142,6 +148,7 @@ public abstract class SafeVariableLengthDimensionDataChunkStore
@Override
public void fillRow(int rowId, CarbonColumnVector vector, int vectorRow) {
+ vector.setDictionary(null);
// if column was explicitly sorted we need to get the rowid based inverted index reverse
if (isExplictSorted) {
rowId = invertedIndexReverse[rowId];
@@ -159,7 +166,7 @@ public abstract class SafeVariableLengthDimensionDataChunkStore
length = dataOffsets[rowId + 1] - (currentDataOffset + getLengthSize());
} else {
// for last record
- length = this.data.length - currentDataOffset;
+ length = this.dataLength - currentDataOffset;
}
DataType dt = vector.getType();
@@ -205,7 +212,7 @@ public abstract class SafeVariableLengthDimensionDataChunkStore
length = dataOffsets[rowId + 1] - (currentDataOffset + getLengthSize());
} else {
// for last record
- length = this.data.length - currentDataOffset;
+ length = this.dataLength - currentDataOffset;
}
return ByteUtil.UnsafeComparer.INSTANCE
.compareTo(data, currentDataOffset, length, compareValue, 0, compareValue.length);
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/safe/SafeVariableShortLengthDimensionDataChunkStore.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/safe/SafeVariableShortLengthDimensionDataChunkStore.java b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/safe/SafeVariableShortLengthDimensionDataChunkStore.java
index beccf86..639973a 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/safe/SafeVariableShortLengthDimensionDataChunkStore.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/safe/SafeVariableShortLengthDimensionDataChunkStore.java
@@ -27,8 +27,9 @@ import org.apache.carbondata.core.constants.CarbonCommonConstants;
*/
public class SafeVariableShortLengthDimensionDataChunkStore
extends SafeVariableLengthDimensionDataChunkStore {
- public SafeVariableShortLengthDimensionDataChunkStore(boolean isInvertedIndex, int numberOfRows) {
- super(isInvertedIndex, numberOfRows);
+ public SafeVariableShortLengthDimensionDataChunkStore(boolean isInvertedIndex, int numberOfRows,
+ int dataLength) {
+ super(isInvertedIndex, numberOfRows, dataLength);
}
@Override protected int getLengthSize() {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/unsafe/UnsafeAbstractDimensionDataChunkStore.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/unsafe/UnsafeAbstractDimensionDataChunkStore.java b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/unsafe/UnsafeAbstractDimensionDataChunkStore.java
index 57e9de5..0150179 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/unsafe/UnsafeAbstractDimensionDataChunkStore.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/unsafe/UnsafeAbstractDimensionDataChunkStore.java
@@ -73,13 +73,14 @@ public abstract class UnsafeAbstractDimensionDataChunkStore implements Dimension
* @param numberOfRows total number of rows
*/
public UnsafeAbstractDimensionDataChunkStore(long totalSize, boolean isInvertedIdex,
- int numberOfRows) {
+ int numberOfRows, int dataLength) {
try {
// allocating the data page
this.dataPageMemoryBlock = UnsafeMemoryManager.allocateMemoryWithRetry(taskId, totalSize);
} catch (MemoryException e) {
throw new RuntimeException(e);
}
+ this.dataLength = dataLength;
this.isExplicitSorted = isInvertedIdex;
}
@@ -93,7 +94,6 @@ public abstract class UnsafeAbstractDimensionDataChunkStore implements Dimension
@Override public void putArray(final int[] invertedIndex, final int[] invertedIndexReverse,
final byte[] data) {
assert (!isMemoryOccupied);
- this.dataLength = data.length;
this.invertedIndexReverseOffset = dataLength;
if (isExplicitSorted) {
this.invertedIndexReverseOffset +=
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/unsafe/UnsafeFixedLengthDimensionDataChunkStore.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/unsafe/UnsafeFixedLengthDimensionDataChunkStore.java b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/unsafe/UnsafeFixedLengthDimensionDataChunkStore.java
index a689d8e..b0027bf 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/unsafe/UnsafeFixedLengthDimensionDataChunkStore.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/unsafe/UnsafeFixedLengthDimensionDataChunkStore.java
@@ -40,8 +40,8 @@ public class UnsafeFixedLengthDimensionDataChunkStore
* @param numberOfRows total number of rows
*/
public UnsafeFixedLengthDimensionDataChunkStore(long totalDataSize, int columnValueSize,
- boolean isInvertedIdex, int numberOfRows) {
- super(totalDataSize, isInvertedIdex, numberOfRows);
+ boolean isInvertedIdex, int numberOfRows, int dataLength) {
+ super(totalDataSize, isInvertedIdex, numberOfRows, dataLength);
this.columnValueSize = columnValueSize;
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/unsafe/UnsafeVariableIntLengthDimensionDataChunkStore.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/unsafe/UnsafeVariableIntLengthDimensionDataChunkStore.java b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/unsafe/UnsafeVariableIntLengthDimensionDataChunkStore.java
index 851fff6..80a7482 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/unsafe/UnsafeVariableIntLengthDimensionDataChunkStore.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/unsafe/UnsafeVariableIntLengthDimensionDataChunkStore.java
@@ -28,8 +28,8 @@ import org.apache.carbondata.core.constants.CarbonCommonConstants;
public class UnsafeVariableIntLengthDimensionDataChunkStore
extends UnsafeVariableLengthDimensionDataChunkStore {
public UnsafeVariableIntLengthDimensionDataChunkStore(long totalSize, boolean isInvertedIdex,
- int numberOfRows) {
- super(totalSize, isInvertedIdex, numberOfRows);
+ int numberOfRows, int dataLength) {
+ super(totalSize, isInvertedIdex, numberOfRows, dataLength);
}
@Override
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/unsafe/UnsafeVariableLengthDimensionDataChunkStore.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/unsafe/UnsafeVariableLengthDimensionDataChunkStore.java b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/unsafe/UnsafeVariableLengthDimensionDataChunkStore.java
index 15217b8..8d2f623 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/unsafe/UnsafeVariableLengthDimensionDataChunkStore.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/unsafe/UnsafeVariableLengthDimensionDataChunkStore.java
@@ -50,8 +50,8 @@ public abstract class UnsafeVariableLengthDimensionDataChunkStore
private byte[] value;
public UnsafeVariableLengthDimensionDataChunkStore(long totalSize, boolean isInvertedIdex,
- int numberOfRows) {
- super(totalSize, isInvertedIdex, numberOfRows);
+ int numberOfRows, int dataLength) {
+ super(totalSize, isInvertedIdex, numberOfRows, dataLength);
this.numberOfRows = numberOfRows;
// initials size assigning to some random value
this.value = new byte[20];
@@ -220,6 +220,7 @@ public abstract class UnsafeVariableLengthDimensionDataChunkStore
*/
@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
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/unsafe/UnsafeVariableShortLengthDimensionDataChunkStore.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/unsafe/UnsafeVariableShortLengthDimensionDataChunkStore.java b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/unsafe/UnsafeVariableShortLengthDimensionDataChunkStore.java
index 995f5ba..502fc48 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/unsafe/UnsafeVariableShortLengthDimensionDataChunkStore.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/chunk/store/impl/unsafe/UnsafeVariableShortLengthDimensionDataChunkStore.java
@@ -28,8 +28,8 @@ import org.apache.carbondata.core.constants.CarbonCommonConstants;
public class UnsafeVariableShortLengthDimensionDataChunkStore
extends UnsafeVariableLengthDimensionDataChunkStore {
public UnsafeVariableShortLengthDimensionDataChunkStore(long totalSize, boolean isInvertedIdex,
- int numberOfRows) {
- super(totalSize, isInvertedIdex, numberOfRows);
+ int numberOfRows, int dataLength) {
+ super(totalSize, isInvertedIdex, numberOfRows, dataLength);
}
@Override
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/datastore/columnar/UnBlockIndexer.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/columnar/UnBlockIndexer.java b/core/src/main/java/org/apache/carbondata/core/datastore/columnar/UnBlockIndexer.java
index 48484ce..d57dfad 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/columnar/UnBlockIndexer.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/columnar/UnBlockIndexer.java
@@ -54,7 +54,7 @@ public final class UnBlockIndexer {
return indexes;
}
- public static byte[] uncompressData(byte[] data, int[] index, int keyLen) {
+ public static byte[] uncompressData(byte[] data, int[] index, int keyLen, int dataLength) {
if (index.length < 1) {
return data;
}
@@ -67,7 +67,7 @@ public final class UnBlockIndexer {
}
byte[] uncompressedData = new byte[actualSize * keyLen];
int picIndex = 0;
- for (int i = 0; i < data.length; i += keyLen) {
+ for (int i = 0; i < dataLength; i += keyLen) {
numberOfCopy = index[picIndex * 2 + 1];
picIndex++;
for (int j = 0; j < numberOfCopy; j++) {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/datastore/compression/AbstractCompressor.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/compression/AbstractCompressor.java b/core/src/main/java/org/apache/carbondata/core/datastore/compression/AbstractCompressor.java
index 5123cc6..0724bdc 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/compression/AbstractCompressor.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/compression/AbstractCompressor.java
@@ -119,5 +119,8 @@ public abstract class AbstractCompressor implements Compressor {
throw new RuntimeException("Not implemented rawCompress for " + this.getName());
}
+ @Override public boolean supportReusableBuffer() {
+ return false;
+ }
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/datastore/compression/Compressor.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/compression/Compressor.java b/core/src/main/java/org/apache/carbondata/core/datastore/compression/Compressor.java
index 282e12c..a48034f 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/compression/Compressor.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/compression/Compressor.java
@@ -65,4 +65,10 @@ public interface Compressor {
* @return true if it supports, otherwise return false
*/
boolean supportUnsafe();
+
+ int unCompressedLength(byte[] data, int offset, int length);
+
+ int rawUncompress(byte[] data, int offset, int length, byte[] output);
+
+ boolean supportReusableBuffer();
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/datastore/compression/SnappyCompressor.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/compression/SnappyCompressor.java b/core/src/main/java/org/apache/carbondata/core/datastore/compression/SnappyCompressor.java
index 6f4a9c6..79e4f4f 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/compression/SnappyCompressor.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/compression/SnappyCompressor.java
@@ -210,4 +210,24 @@ public class SnappyCompressor extends AbstractCompressor {
public boolean supportUnsafe() {
return true;
}
+
+ @Override public int unCompressedLength(byte[] data, int offset, int length) {
+ try {
+ return Snappy.uncompressedLength(data, offset, length);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override public int rawUncompress(byte[] data, int offset, int length, byte[] output) {
+ try {
+ return Snappy.rawUncompress(data, offset, length, output, 0);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override public boolean supportReusableBuffer() {
+ return true;
+ }
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/d79ba999/core/src/main/java/org/apache/carbondata/core/datastore/compression/ZstdCompressor.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/compression/ZstdCompressor.java b/core/src/main/java/org/apache/carbondata/core/datastore/compression/ZstdCompressor.java
index 3e6a11b..8523c46 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/compression/ZstdCompressor.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/compression/ZstdCompressor.java
@@ -74,4 +74,12 @@ public class ZstdCompressor extends AbstractCompressor {
public boolean supportUnsafe() {
return false;
}
+
+ @Override public int unCompressedLength(byte[] data, int offset, int length) {
+ throw new RuntimeException("Unsupported operation Exception");
+ }
+
+ @Override public int rawUncompress(byte[] data, int offset, int length, byte[] output) {
+ throw new RuntimeException("Unsupported operation Exception");
+ }
}