http://git-wip-us.apache.org/repos/asf/carbondata/blob/bed51ba7/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 fd94344..7b7c0b6 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,12 +23,13 @@ import java.util.BitSet; import java.util.List; import java.util.Map; +import org.apache.carbondata.core.datastore.TableSpec; import org.apache.carbondata.core.datastore.compression.Compressor; import org.apache.carbondata.core.datastore.compression.CompressorFactory; import org.apache.carbondata.core.datastore.page.ColumnPage; import org.apache.carbondata.core.datastore.page.ColumnPageValueConverter; -import org.apache.carbondata.core.datastore.page.DecimalColumnPage; import org.apache.carbondata.core.datastore.page.LazyColumnPage; +import org.apache.carbondata.core.datastore.page.VarLengthColumnPageBase; import org.apache.carbondata.core.datastore.page.encoding.ColumnPageCodec; import org.apache.carbondata.core.datastore.page.encoding.ColumnPageDecoder; import org.apache.carbondata.core.datastore.page.encoding.ColumnPageEncoder; @@ -105,17 +106,32 @@ public class DirectCompressCodec implements ColumnPageCodec { @Override public void decodeAndFillVector(byte[] input, int offset, int length, - ColumnVectorInfo vectorInfo, BitSet nullBits, boolean isLVEncoded) + ColumnVectorInfo vectorInfo, BitSet nullBits, boolean isLVEncoded, int pageSize) throws MemoryException, IOException { - ColumnPage decodedPage; + Compressor compressor = + CompressorFactory.getInstance().getCompressor(meta.getCompressorName()); + byte[] unCompressData = compressor.unCompressByte(input, offset, length); if (DataTypes.isDecimal(dataType)) { - decodedPage = ColumnPage.decompressDecimalPage(meta, input, offset, length); - vectorInfo.decimalConverter = ((DecimalColumnPage) decodedPage).getDecimalConverter(); + TableSpec.ColumnSpec columnSpec = meta.getColumnSpec(); + DecimalConverterFactory.DecimalConverter decimalConverter = + DecimalConverterFactory.INSTANCE + .getDecimalConverter(columnSpec.getPrecision(), columnSpec.getScale()); + vectorInfo.decimalConverter = decimalConverter; + if (DataTypes.isDecimal(meta.getStoreDataType())) { + ColumnPage decimalColumnPage = + VarLengthColumnPageBase.newDecimalColumnPage(meta, unCompressData); + decimalConverter.fillVector(decimalColumnPage.getByteArrayPage(), pageSize, vectorInfo, + nullBits, meta.getStoreDataType()); + } else { + converter + .decodeAndFillVector(unCompressData, vectorInfo, nullBits, meta.getStoreDataType(), + pageSize); + } } else { - decodedPage = ColumnPage.decompress(meta, input, offset, length, isLVEncoded); + converter + .decodeAndFillVector(unCompressData, vectorInfo, nullBits, meta.getStoreDataType(), + pageSize); } - decodedPage.setNullBits(nullBits); - converter.decodeAndFillVector(decodedPage, vectorInfo); } @Override public ColumnPage decode(byte[] input, int offset, int length, boolean isLVEncoded) @@ -203,17 +219,15 @@ public class DirectCompressCodec implements ColumnPageCodec { } @Override - public void decodeAndFillVector(ColumnPage columnPage, ColumnVectorInfo vectorInfo) { + public void decodeAndFillVector(byte[] pageData, ColumnVectorInfo vectorInfo, BitSet nullBits, + DataType pageDataType, int pageSize) { CarbonColumnVector vector = vectorInfo.vector; - BitSet nullBits = columnPage.getNullBits(); DataType vectorDataType = vector.getType(); - DataType pageDataType = columnPage.getDataType(); - int pageSize = columnPage.getPageSize(); BitSet deletedRows = vectorInfo.deletedRows; vector = ColumnarVectorWrapperDirectFactory .getDirectVectorWrapperFactory(vector, vectorInfo.invertedIndex, nullBits, deletedRows, true, false); - fillVector(columnPage, vector, vectorDataType, pageDataType, pageSize, vectorInfo); + fillVector(pageData, vector, vectorDataType, pageDataType, pageSize, vectorInfo, nullBits); if (deletedRows == null || deletedRows.isEmpty()) { for (int i = nullBits.nextSetBit(0); i >= 0; i = nullBits.nextSetBit(i + 1)) { vector.putNull(i); @@ -224,130 +238,134 @@ public class DirectCompressCodec implements ColumnPageCodec { } } - private void fillVector(ColumnPage columnPage, CarbonColumnVector vector, - DataType vectorDataType, DataType pageDataType, int pageSize, ColumnVectorInfo vectorInfo) { + private void fillVector(byte[] pageData, CarbonColumnVector vector, DataType vectorDataType, + DataType pageDataType, int pageSize, ColumnVectorInfo vectorInfo, BitSet nullBits) { + int rowId = 0; if (pageDataType == DataTypes.BOOLEAN || pageDataType == DataTypes.BYTE) { - byte[] byteData = columnPage.getBytePage(); if (vectorDataType == DataTypes.SHORT) { for (int i = 0; i < pageSize; i++) { - vector.putShort(i, (short) byteData[i]); + vector.putShort(i, (short) pageData[i]); } } else if (vectorDataType == DataTypes.INT) { for (int i = 0; i < pageSize; i++) { - vector.putInt(i, (int) byteData[i]); + vector.putInt(i, (int) pageData[i]); } } else if (vectorDataType == DataTypes.LONG) { for (int i = 0; i < pageSize; i++) { - vector.putLong(i, byteData[i]); + vector.putLong(i, pageData[i]); } } else if (vectorDataType == DataTypes.TIMESTAMP) { for (int i = 0; i < pageSize; i++) { - vector.putLong(i, (long) byteData[i] * 1000); + vector.putLong(i, (long) pageData[i] * 1000); } } else if (vectorDataType == DataTypes.BOOLEAN || vectorDataType == DataTypes.BYTE) { - vector.putBytes(0, pageSize, byteData, 0); + vector.putBytes(0, pageSize, pageData, 0); } else if (DataTypes.isDecimal(vectorDataType)) { DecimalConverterFactory.DecimalConverter decimalConverter = vectorInfo.decimalConverter; - decimalConverter.fillVector(byteData, pageSize, vectorInfo, columnPage.getNullBits()); + decimalConverter.fillVector(pageData, pageSize, vectorInfo, nullBits, pageDataType); } else { for (int i = 0; i < pageSize; i++) { - vector.putDouble(i, byteData[i]); + vector.putDouble(i, pageData[i]); } } } else if (pageDataType == DataTypes.SHORT) { - short[] shortData = columnPage.getShortPage(); + int size = pageSize * DataTypes.SHORT.getSizeInBytes(); if (vectorDataType == DataTypes.SHORT) { - vector.putShorts(0, pageSize, shortData, 0); + for (int i = 0; i < size; i += DataTypes.SHORT.getSizeInBytes()) { + vector.putShort(rowId++, (ByteUtil.toShortLittleEndian(pageData, i))); + } } else if (vectorDataType == DataTypes.INT) { - for (int i = 0; i < pageSize; i++) { - vector.putInt(i, (int) shortData[i]); + for (int i = 0; i < size; i += DataTypes.SHORT.getSizeInBytes()) { + vector.putInt(rowId++, ByteUtil.toShortLittleEndian(pageData, i)); } } else if (vectorDataType == DataTypes.LONG) { - for (int i = 0; i < pageSize; i++) { - vector.putLong(i, shortData[i]); + for (int i = 0; i < size; i += DataTypes.SHORT.getSizeInBytes()) { + vector.putLong(rowId++, ByteUtil.toShortLittleEndian(pageData, i)); } } else if (vectorDataType == DataTypes.TIMESTAMP) { - for (int i = 0; i < pageSize; i++) { - vector.putLong(i, (long) shortData[i] * 1000); + for (int i = 0; i < size; i += DataTypes.SHORT.getSizeInBytes()) { + vector.putLong(rowId++, (long) ByteUtil.toShortLittleEndian(pageData, i) * 1000); } } else if (DataTypes.isDecimal(vectorDataType)) { DecimalConverterFactory.DecimalConverter decimalConverter = vectorInfo.decimalConverter; - decimalConverter.fillVector(shortData, pageSize, vectorInfo, columnPage.getNullBits()); + decimalConverter.fillVector(pageData, pageSize, vectorInfo, nullBits, pageDataType); } else { - for (int i = 0; i < pageSize; i++) { - vector.putDouble(i, shortData[i]); + for (int i = 0; i < size; i += DataTypes.SHORT.getSizeInBytes()) { + vector.putDouble(rowId++, ByteUtil.toShortLittleEndian(pageData, i)); } } } else if (pageDataType == DataTypes.SHORT_INT) { - byte[] shortIntPage = columnPage.getShortIntPage(); if (vectorDataType == DataTypes.INT) { for (int i = 0; i < pageSize; i++) { - int shortInt = ByteUtil.valueOf3Bytes(shortIntPage, i * 3); + int shortInt = ByteUtil.valueOf3Bytes(pageData, i * 3); vector.putInt(i, shortInt); } } else if (vectorDataType == DataTypes.LONG) { for (int i = 0; i < pageSize; i++) { - int shortInt = ByteUtil.valueOf3Bytes(shortIntPage, i * 3); + int shortInt = ByteUtil.valueOf3Bytes(pageData, i * 3); vector.putLong(i, shortInt); } } else if (vectorDataType == DataTypes.TIMESTAMP) { for (int i = 0; i < pageSize; i++) { - int shortInt = ByteUtil.valueOf3Bytes(shortIntPage, i * 3); + int shortInt = ByteUtil.valueOf3Bytes(pageData, i * 3); vector.putLong(i, (long) shortInt * 1000); } } else if (DataTypes.isDecimal(vectorDataType)) { DecimalConverterFactory.DecimalConverter decimalConverter = vectorInfo.decimalConverter; - int[] shortIntData = ByteUtil.toIntArrayFrom3Bytes(shortIntPage, pageSize); - decimalConverter.fillVector(shortIntData, pageSize, vectorInfo, columnPage.getNullBits()); + decimalConverter.fillVector(pageData, pageSize, vectorInfo, nullBits, pageDataType); } else { for (int i = 0; i < pageSize; i++) { - int shortInt = ByteUtil.valueOf3Bytes(shortIntPage, i * 3); + int shortInt = ByteUtil.valueOf3Bytes(pageData, i * 3); vector.putDouble(i, shortInt); } } } else if (pageDataType == DataTypes.INT) { - int[] intData = columnPage.getIntPage(); + int size = pageSize * DataTypes.INT.getSizeInBytes(); if (vectorDataType == DataTypes.INT) { - vector.putInts(0, pageSize, intData, 0); + for (int i = 0; i < size; i += DataTypes.INT.getSizeInBytes()) { + vector.putInt(rowId++, ByteUtil.toIntLittleEndian(pageData, i)); + } } else if (vectorDataType == DataTypes.LONG) { - for (int i = 0; i < pageSize; i++) { - vector.putLong(i, intData[i]); + for (int i = 0; i < size; i += DataTypes.INT.getSizeInBytes()) { + vector.putLong(rowId++, ByteUtil.toIntLittleEndian(pageData, i)); } } else if (vectorDataType == DataTypes.TIMESTAMP) { - for (int i = 0; i < pageSize; i++) { - vector.putLong(i, (long) intData[i] * 1000); + for (int i = 0; i < size; i += DataTypes.INT.getSizeInBytes()) { + vector.putLong(rowId++, (long) ByteUtil.toIntLittleEndian(pageData, i) * 1000); } } else if (DataTypes.isDecimal(vectorDataType)) { DecimalConverterFactory.DecimalConverter decimalConverter = vectorInfo.decimalConverter; - decimalConverter.fillVector(intData, pageSize, vectorInfo, columnPage.getNullBits()); + decimalConverter.fillVector(pageData, pageSize, vectorInfo, nullBits, pageDataType); } else { - for (int i = 0; i < pageSize; i++) { - vector.putDouble(i, intData[i]); + for (int i = 0; i < size; i += DataTypes.INT.getSizeInBytes()) { + vector.putDouble(rowId++, ByteUtil.toIntLittleEndian(pageData, i)); } } - } else if (pageDataType == DataTypes.LONG) { - long[] longData = columnPage.getLongPage(); + } else if (pageDataType == DataTypes.LONG) { + int size = pageSize * DataTypes.LONG.getSizeInBytes(); if (vectorDataType == DataTypes.LONG) { - vector.putLongs(0, pageSize, longData, 0); + for (int i = 0; i < size; i += DataTypes.LONG.getSizeInBytes()) { + vector.putLong(rowId++, ByteUtil.toLongLittleEndian(pageData, i)); + } } else if (vectorDataType == DataTypes.TIMESTAMP) { - for (int i = 0; i < pageSize; i++) { - vector.putLong(i, longData[i] * 1000); + for (int i = 0; i < size; i += DataTypes.LONG.getSizeInBytes()) { + vector.putLong(rowId++, ByteUtil.toLongLittleEndian(pageData, i) * 1000); } } else if (DataTypes.isDecimal(vectorDataType)) { DecimalConverterFactory.DecimalConverter decimalConverter = vectorInfo.decimalConverter; - decimalConverter.fillVector(longData, pageSize, vectorInfo, columnPage.getNullBits()); + decimalConverter.fillVector(pageData, pageSize, vectorInfo, nullBits, pageDataType); } - } else if (DataTypes.isDecimal(pageDataType)) { - DecimalConverterFactory.DecimalConverter decimalConverter = vectorInfo.decimalConverter; - decimalConverter.fillVector(columnPage.getByteArrayPage(), pageSize, vectorInfo, - columnPage.getNullBits()); } else if (vectorDataType == DataTypes.FLOAT) { - float[] floatPage = columnPage.getFloatPage(); - vector.putFloats(0, pageSize, floatPage, 0); + int size = pageSize * DataTypes.FLOAT.getSizeInBytes(); + for (int i = 0; i < size; i += DataTypes.FLOAT.getSizeInBytes()) { + vector.putFloat(rowId++, ByteUtil.toFloatLittleEndian(pageData, i)); + } } else { - double[] doubleData = columnPage.getDoublePage(); - vector.putDoubles(0, pageSize, doubleData, 0); + int size = pageSize * DataTypes.DOUBLE.getSizeInBytes(); + for (int i = 0; i < size; i += DataTypes.DOUBLE.getSizeInBytes()) { + vector.putDouble(rowId++, ByteUtil.toDoubleLittleEndian(pageData, i)); + } } } };
http://git-wip-us.apache.org/repos/asf/carbondata/blob/bed51ba7/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 c9b47db..2c940bb 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 @@ -318,7 +318,7 @@ public class RLECodec implements ColumnPageCodec { @Override public void decodeAndFillVector(byte[] input, int offset, int length, - ColumnVectorInfo vectorInfo, BitSet nullBits, boolean isLVEncoded) + ColumnVectorInfo vectorInfo, BitSet nullBits, boolean isLVEncoded, int pageSize) throws MemoryException, IOException { throw new UnsupportedOperationException("Not supposed to be called here"); } http://git-wip-us.apache.org/repos/asf/carbondata/blob/bed51ba7/core/src/main/java/org/apache/carbondata/core/datastore/page/statistics/PrimitivePageStatsCollector.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/page/statistics/PrimitivePageStatsCollector.java b/core/src/main/java/org/apache/carbondata/core/datastore/page/statistics/PrimitivePageStatsCollector.java index e604057..babf144 100644 --- a/core/src/main/java/org/apache/carbondata/core/datastore/page/statistics/PrimitivePageStatsCollector.java +++ b/core/src/main/java/org/apache/carbondata/core/datastore/page/statistics/PrimitivePageStatsCollector.java @@ -30,6 +30,7 @@ import static org.apache.carbondata.core.datastore.page.encoding.bool.BooleanCon /** statics for primitive column page */ public class PrimitivePageStatsCollector implements ColumnPageStatsCollector, SimpleStatsResult { + private static final String ZERO_STRING = "0"; private DataType dataType; private byte minByte, maxByte; private short minShort, maxShort; @@ -243,6 +244,12 @@ public class PrimitivePageStatsCollector implements ColumnPageStatsCollector, Si int integerPlaces = strValue.indexOf('.'); if (-1 != integerPlaces) { decimalPlaces = strValue.length() - integerPlaces - 1; + // If decimal places are one and it is just zero then treat the decimal count a zero. + if (decimalPlaces == 1) { + if (strValue.substring(integerPlaces + 1, strValue.length()).equals(ZERO_STRING)) { + decimalPlaces = 0; + } + } } } catch (NumberFormatException e) { if (!Double.isInfinite(value)) { http://git-wip-us.apache.org/repos/asf/carbondata/blob/bed51ba7/core/src/main/java/org/apache/carbondata/core/datastore/page/statistics/StatisticsCollector.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/page/statistics/StatisticsCollector.java b/core/src/main/java/org/apache/carbondata/core/datastore/page/statistics/StatisticsCollector.java deleted file mode 100644 index f8b336c..0000000 --- a/core/src/main/java/org/apache/carbondata/core/datastore/page/statistics/StatisticsCollector.java +++ /dev/null @@ -1,66 +0,0 @@ -/* - * 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.page.statistics; - -import org.apache.carbondata.core.datastore.page.ColumnPage; - -/** - * Calculate the statistics for a column page and blocklet - */ -public interface StatisticsCollector { - - /** - * name will be stored in Header - */ - String getName(); - - void startPage(int pageID); - - void endPage(int pageID); - - void startBlocklet(int blockletID); - - void endBlocklet(int blockletID); - - void startBlock(int blocklID); - - void endBlock(int blockID); - - /** - * Update the stats for the input batch - */ - void update(ColumnPage batch); - - /** - * Ouput will be written to DataChunk2 (page header) - */ - byte[] getPageStatistisc(); - - /** - * Output will be written to DataChunk3 (blocklet header) - */ - byte[] getBlockletStatistics(); - - /** - * Output will be written to Footer - */ - byte[] getBlockStatistics(); -} - - - http://git-wip-us.apache.org/repos/asf/carbondata/blob/bed51ba7/core/src/main/java/org/apache/carbondata/core/metadata/datatype/DecimalConverterFactory.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/carbondata/core/metadata/datatype/DecimalConverterFactory.java b/core/src/main/java/org/apache/carbondata/core/metadata/datatype/DecimalConverterFactory.java index 5231cb9..9793c38 100644 --- a/core/src/main/java/org/apache/carbondata/core/metadata/datatype/DecimalConverterFactory.java +++ b/core/src/main/java/org/apache/carbondata/core/metadata/datatype/DecimalConverterFactory.java @@ -23,6 +23,7 @@ import java.util.BitSet; import org.apache.carbondata.core.scan.result.vector.CarbonColumnVector; import org.apache.carbondata.core.scan.result.vector.ColumnVectorInfo; +import org.apache.carbondata.core.util.ByteUtil; import org.apache.carbondata.core.util.DataTypeUtil; /** @@ -75,7 +76,8 @@ public final class DecimalConverterFactory { BigDecimal getDecimal(Object valueToBeConverted); - void fillVector(Object valuesToBeConverted, int size, ColumnVectorInfo info, BitSet nullBitset); + void fillVector(Object valuesToBeConverted, int size, ColumnVectorInfo info, BitSet nullBitset, + DataType pageType); int getSize(); @@ -101,14 +103,18 @@ public final class DecimalConverterFactory { } @Override public void fillVector(Object valuesToBeConverted, int size, ColumnVectorInfo info, - BitSet nullBitset) { + BitSet nullBitset, DataType pageType) { // TODO we need to find way to directly set to vector with out conversion. This way is very // inefficient. CarbonColumnVector vector = info.vector; int precision = info.measure.getMeasure().getPrecision(); int newMeasureScale = info.measure.getMeasure().getScale(); - if (valuesToBeConverted instanceof byte[]) { - byte[] data = (byte[]) valuesToBeConverted; + if (!(valuesToBeConverted instanceof byte[])) { + throw new UnsupportedOperationException("This object type " + valuesToBeConverted.getClass() + + " is not supported in this method"); + } + byte[] data = (byte[]) valuesToBeConverted; + if (pageType == DataTypes.BYTE) { for (int i = 0; i < size; i++) { if (nullBitset.get(i)) { vector.putNull(i); @@ -120,39 +126,56 @@ public final class DecimalConverterFactory { vector.putDecimal(i, value, precision); } } - } else if (valuesToBeConverted instanceof short[]) { - short[] data = (short[]) valuesToBeConverted; + } else if (pageType == DataTypes.SHORT) { for (int i = 0; i < size; i++) { if (nullBitset.get(i)) { vector.putNull(i); } else { - BigDecimal value = BigDecimal.valueOf(data[i], scale); + BigDecimal value = BigDecimal + .valueOf(ByteUtil.toShortLittleEndian(data, i * DataTypes.SHORT.getSizeInBytes()), + scale); if (value.scale() < newMeasureScale) { value = value.setScale(newMeasureScale); } vector.putDecimal(i, value, precision); } } - } else if (valuesToBeConverted instanceof int[]) { - int[] data = (int[]) valuesToBeConverted; + } else if (pageType == DataTypes.SHORT_INT) { for (int i = 0; i < size; i++) { if (nullBitset.get(i)) { vector.putNull(i); } else { - BigDecimal value = BigDecimal.valueOf(data[i], scale); + BigDecimal value = BigDecimal + .valueOf(ByteUtil.valueOf3Bytes(data, i * DataTypes.SHORT_INT.getSizeInBytes()), + scale); if (value.scale() < newMeasureScale) { value = value.setScale(newMeasureScale); } vector.putDecimal(i, value, precision); } } - } else if (valuesToBeConverted instanceof long[]) { - long[] data = (long[]) valuesToBeConverted; + } else if (pageType == DataTypes.INT) { for (int i = 0; i < size; i++) { if (nullBitset.get(i)) { vector.putNull(i); } else { - BigDecimal value = BigDecimal.valueOf(data[i], scale); + BigDecimal value = BigDecimal + .valueOf(ByteUtil.toIntLittleEndian(data, i * DataTypes.INT.getSizeInBytes()), + scale); + if (value.scale() < newMeasureScale) { + value = value.setScale(newMeasureScale); + } + vector.putDecimal(i, value, precision); + } + } + } else if (pageType == DataTypes.LONG) { + for (int i = 0; i < size; i++) { + if (nullBitset.get(i)) { + vector.putNull(i); + } else { + BigDecimal value = BigDecimal + .valueOf(ByteUtil.toLongLittleEndian(data, i * DataTypes.LONG.getSizeInBytes()), + scale); if (value.scale() < newMeasureScale) { value = value.setScale(newMeasureScale); } @@ -239,7 +262,7 @@ public final class DecimalConverterFactory { } @Override public void fillVector(Object valuesToBeConverted, int size, ColumnVectorInfo info, - BitSet nullBitset) { + BitSet nullBitset, DataType pageType) { CarbonColumnVector vector = info.vector; int precision = info.measure.getMeasure().getPrecision(); int newMeasureScale = info.measure.getMeasure().getScale(); @@ -285,7 +308,7 @@ public final class DecimalConverterFactory { } @Override public void fillVector(Object valuesToBeConverted, int size, ColumnVectorInfo info, - BitSet nullBitset) { + BitSet nullBitset, DataType pageType) { CarbonColumnVector vector = info.vector; int precision = info.measure.getMeasure().getPrecision(); int newMeasureScale = info.measure.getMeasure().getScale(); http://git-wip-us.apache.org/repos/asf/carbondata/blob/bed51ba7/core/src/main/java/org/apache/carbondata/core/scan/result/vector/CarbonColumnVector.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/carbondata/core/scan/result/vector/CarbonColumnVector.java b/core/src/main/java/org/apache/carbondata/core/scan/result/vector/CarbonColumnVector.java index 6b8455f..25a2eb1 100644 --- a/core/src/main/java/org/apache/carbondata/core/scan/result/vector/CarbonColumnVector.java +++ b/core/src/main/java/org/apache/carbondata/core/scan/result/vector/CarbonColumnVector.java @@ -62,6 +62,10 @@ public interface CarbonColumnVector { void putByteArray(int rowId, int offset, int length, byte[] value); + void putArray(int rowId, int offset, int length); + + void putAllByteArray(byte[] data, int offset, int length); + void putByte(int rowId, byte value); void putBytes(int rowId, int count, byte[] value); http://git-wip-us.apache.org/repos/asf/carbondata/blob/bed51ba7/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 882a365..84d52a6 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 @@ -25,4 +25,6 @@ public interface CarbonDictionary { byte[] getDictionaryValue(int index); byte[][] getAllDictionaryValues(); + + byte[] getAllDictionaryValuesInSingleArray(); } http://git-wip-us.apache.org/repos/asf/carbondata/blob/bed51ba7/core/src/main/java/org/apache/carbondata/core/scan/result/vector/impl/CarbonColumnVectorImpl.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/carbondata/core/scan/result/vector/impl/CarbonColumnVectorImpl.java b/core/src/main/java/org/apache/carbondata/core/scan/result/vector/impl/CarbonColumnVectorImpl.java index f89ad9d..98536f6 100644 --- a/core/src/main/java/org/apache/carbondata/core/scan/result/vector/impl/CarbonColumnVectorImpl.java +++ b/core/src/main/java/org/apache/carbondata/core/scan/result/vector/impl/CarbonColumnVectorImpl.java @@ -53,6 +53,12 @@ public class CarbonColumnVectorImpl implements CarbonColumnVector { private DataType blockDataType; + private int[] lengths; + + private int[] offsets; + + private int batchSize; + /** * True if there is at least one NULL byte set. This is an optimization for the writer, to skip * having to clear NULL bits. @@ -64,6 +70,7 @@ public class CarbonColumnVectorImpl implements CarbonColumnVector { private CarbonColumnVector dictionaryVector; public CarbonColumnVectorImpl(int batchSize, DataType dataType) { + this.batchSize = batchSize; nullBytes = new BitSet(batchSize); this.dataType = dataType; if (dataType == DataTypes.BOOLEAN || dataType == DataTypes.BYTE) { @@ -223,8 +230,13 @@ public class CarbonColumnVectorImpl implements CarbonColumnVector { if (null != carbonDictionary) { int dictKey = (Integer) dictionaryVector.getData(rowId); return carbonDictionary.getDictionaryValue(dictKey); + } else if (byteArr != null) { + byte[] bytes = new byte[lengths[rowId]]; + System.arraycopy(byteArr, offsets[rowId], bytes, 0, bytes.length); + return bytes; + } else { + return bytes[rowId]; } - return bytes[rowId]; } else { return data[rowId]; } @@ -357,4 +369,25 @@ public class CarbonColumnVectorImpl implements CarbonColumnVector { @Override public void setLazyPage(LazyPageLoader lazyPage) { lazyPage.loadPage(); } + + @Override public void putArray(int rowId, int offset, int length) { + if (offsets == null) { + offsets = new int[batchSize]; + lengths = new int[batchSize]; + } + offsets[rowId] = offset; + lengths[rowId] = length; + } + + @Override public void putAllByteArray(byte[] data, int offset, int length) { + byteArr = data; + } + + public int[] getLengths() { + return lengths; + } + + public int[] getOffsets() { + return offsets; + } } http://git-wip-us.apache.org/repos/asf/carbondata/blob/bed51ba7/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 20e6171..962a772 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 @@ -22,6 +22,12 @@ public class CarbonDictionaryImpl implements CarbonDictionary { private byte[][] dictionary; + private byte[] singleArrayDictValues; + + private int[] dictLens; + + private int[] dictOffsets; + private int actualSize; public CarbonDictionaryImpl(byte[][] dictionary, int actualSize) { @@ -44,4 +50,35 @@ public class CarbonDictionaryImpl implements CarbonDictionary { @Override public byte[][] getAllDictionaryValues() { return dictionary; } + + @Override public byte[] getAllDictionaryValuesInSingleArray() { + if (singleArrayDictValues == null) { + dictLens = new int[dictionary.length]; + dictOffsets = new int[dictionary.length]; + int size = 0; + for (int i = 0; i < dictionary.length; i++) { + if (dictionary[i] != null) { + dictOffsets[i] = size; + size += dictionary[i].length; + dictLens[i] = dictionary[i].length; + } + } + singleArrayDictValues = new byte[size]; + for (int i = 0; i < dictionary.length; i++) { + if (dictionary[i] != null) { + System.arraycopy(dictionary[i], 0, singleArrayDictValues, dictOffsets[i], dictLens[i]); + } + } + dictionary = null; + } + return singleArrayDictValues; + } + + public int[] getDictLens() { + return dictLens; + } + + public int[] getDictOffsets() { + return dictOffsets; + } } http://git-wip-us.apache.org/repos/asf/carbondata/blob/bed51ba7/core/src/main/java/org/apache/carbondata/core/scan/result/vector/impl/directread/AbstractCarbonColumnarVector.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/carbondata/core/scan/result/vector/impl/directread/AbstractCarbonColumnarVector.java b/core/src/main/java/org/apache/carbondata/core/scan/result/vector/impl/directread/AbstractCarbonColumnarVector.java index 7a1f317..4c783b4 100644 --- a/core/src/main/java/org/apache/carbondata/core/scan/result/vector/impl/directread/AbstractCarbonColumnarVector.java +++ b/core/src/main/java/org/apache/carbondata/core/scan/result/vector/impl/directread/AbstractCarbonColumnarVector.java @@ -27,6 +27,12 @@ import org.apache.carbondata.core.scan.scanner.LazyPageLoader; public abstract class AbstractCarbonColumnarVector implements CarbonColumnVector, ConvertableVector { + protected CarbonColumnVector columnVector; + + public AbstractCarbonColumnarVector(CarbonColumnVector columnVector) { + this.columnVector = columnVector; + } + @Override public void putShorts(int rowId, int count, short value) { throw new UnsupportedOperationException("Not allowed from here " + getClass().getName()); @@ -136,4 +142,8 @@ public abstract class AbstractCarbonColumnarVector public void setLazyPage(LazyPageLoader lazyPage) { throw new UnsupportedOperationException("Not allowed from here " + getClass().getName()); } + + @Override public void putAllByteArray(byte[] data, int offset, int length) { + columnVector.putAllByteArray(data, offset, length); + } } http://git-wip-us.apache.org/repos/asf/carbondata/blob/bed51ba7/core/src/main/java/org/apache/carbondata/core/scan/result/vector/impl/directread/ColumnarVectorWrapperDirectWithDeleteDelta.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/carbondata/core/scan/result/vector/impl/directread/ColumnarVectorWrapperDirectWithDeleteDelta.java b/core/src/main/java/org/apache/carbondata/core/scan/result/vector/impl/directread/ColumnarVectorWrapperDirectWithDeleteDelta.java index ccde63e..67eb6ec 100644 --- a/core/src/main/java/org/apache/carbondata/core/scan/result/vector/impl/directread/ColumnarVectorWrapperDirectWithDeleteDelta.java +++ b/core/src/main/java/org/apache/carbondata/core/scan/result/vector/impl/directread/ColumnarVectorWrapperDirectWithDeleteDelta.java @@ -34,13 +34,11 @@ class ColumnarVectorWrapperDirectWithDeleteDelta extends AbstractCarbonColumnarV private int counter; - private CarbonColumnVector columnVector; - public ColumnarVectorWrapperDirectWithDeleteDelta(CarbonColumnVector vectorWrapper, BitSet deletedRows, BitSet nullBits) { + super(vectorWrapper); this.deletedRows = deletedRows; this.nullBits = nullBits; - this.columnVector = vectorWrapper; } @Override @@ -213,4 +211,10 @@ class ColumnarVectorWrapperDirectWithDeleteDelta extends AbstractCarbonColumnarV } } } + + @Override public void putArray(int rowId, int offset, int length) { + if (!deletedRows.get(rowId)) { + columnVector.putArray(counter++, offset, length); + } + } } http://git-wip-us.apache.org/repos/asf/carbondata/blob/bed51ba7/core/src/main/java/org/apache/carbondata/core/scan/result/vector/impl/directread/ColumnarVectorWrapperDirectWithDeleteDeltaAndInvertedIndex.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/carbondata/core/scan/result/vector/impl/directread/ColumnarVectorWrapperDirectWithDeleteDeltaAndInvertedIndex.java b/core/src/main/java/org/apache/carbondata/core/scan/result/vector/impl/directread/ColumnarVectorWrapperDirectWithDeleteDeltaAndInvertedIndex.java index 46b2041..2dd70bc 100644 --- a/core/src/main/java/org/apache/carbondata/core/scan/result/vector/impl/directread/ColumnarVectorWrapperDirectWithDeleteDeltaAndInvertedIndex.java +++ b/core/src/main/java/org/apache/carbondata/core/scan/result/vector/impl/directread/ColumnarVectorWrapperDirectWithDeleteDeltaAndInvertedIndex.java @@ -30,7 +30,7 @@ import org.apache.carbondata.core.scan.result.vector.impl.CarbonColumnVectorImpl * Column vector for column pages which has delete delta and inverted index, so it uses delta biset * to filter out data and use inverted index before filling to actual vector */ -class ColumnarVectorWrapperDirectWithDeleteDeltaAndInvertedIndex +public class ColumnarVectorWrapperDirectWithDeleteDeltaAndInvertedIndex extends ColumnarVectorWrapperDirectWithInvertedIndex { private BitSet deletedRows; @@ -78,6 +78,10 @@ class ColumnarVectorWrapperDirectWithDeleteDeltaAndInvertedIndex } } + @Override public void putAllByteArray(byte[] data, int offset, int length) { + carbonColumnVector.putAllByteArray(data, offset, length); + } + @Override public void convert() { if (columnVector instanceof CarbonColumnVectorImpl) { @@ -163,13 +167,27 @@ class ColumnarVectorWrapperDirectWithDeleteDeltaAndInvertedIndex } } } else if (dataType == DataTypes.STRING || dataType == DataTypes.BYTE_ARRAY) { - byte[][] dataArray = (byte[][]) localVector.getDataArray(); - for (int i = 0; i < length; i++) { - if (!deletedRows.get(i)) { - if (nullBits.get(i)) { - carbonColumnVector.putNull(counter++); - } else { - carbonColumnVector.putByteArray(counter++, dataArray[i]); + int[] offsets = localVector.getOffsets(); + int[] lengths = localVector.getLengths(); + if (offsets != null && lengths != null) { + for (int i = 0; i < length; i++) { + if (!deletedRows.get(i)) { + if (nullBits.get(i)) { + carbonColumnVector.putNull(counter++); + } else { + carbonColumnVector.putArray(counter++, offsets[i], lengths[i]); + } + } + } + } else { + byte[][] dataArray = (byte[][]) localVector.getDataArray(); + for (int i = 0; i < length; i++) { + if (!deletedRows.get(i)) { + if (nullBits.get(i)) { + carbonColumnVector.putNull(counter++); + } else { + carbonColumnVector.putByteArray(counter++, dataArray[i]); + } } } } http://git-wip-us.apache.org/repos/asf/carbondata/blob/bed51ba7/core/src/main/java/org/apache/carbondata/core/scan/result/vector/impl/directread/ColumnarVectorWrapperDirectWithInvertedIndex.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/carbondata/core/scan/result/vector/impl/directread/ColumnarVectorWrapperDirectWithInvertedIndex.java b/core/src/main/java/org/apache/carbondata/core/scan/result/vector/impl/directread/ColumnarVectorWrapperDirectWithInvertedIndex.java index f190d7d..aa816ef 100644 --- a/core/src/main/java/org/apache/carbondata/core/scan/result/vector/impl/directread/ColumnarVectorWrapperDirectWithInvertedIndex.java +++ b/core/src/main/java/org/apache/carbondata/core/scan/result/vector/impl/directread/ColumnarVectorWrapperDirectWithInvertedIndex.java @@ -26,18 +26,17 @@ import org.apache.carbondata.core.scan.result.vector.CarbonColumnVector; * Column vector for column pages which has inverted index, so it uses inverted index * before filling to actual vector */ -class ColumnarVectorWrapperDirectWithInvertedIndex extends AbstractCarbonColumnarVector { +public class ColumnarVectorWrapperDirectWithInvertedIndex extends AbstractCarbonColumnarVector { protected int[] invertedIndex; - protected CarbonColumnVector columnVector; protected boolean isnullBitsExists; public ColumnarVectorWrapperDirectWithInvertedIndex(CarbonColumnVector columnVector, int[] invertedIndex, boolean isnullBitsExists) { + super(columnVector); this.invertedIndex = invertedIndex; - this.columnVector = columnVector; this.isnullBitsExists = isnullBitsExists; } @@ -147,4 +146,8 @@ class ColumnarVectorWrapperDirectWithInvertedIndex extends AbstractCarbonColumna public DataType getBlockDataType() { return columnVector.getBlockDataType(); } + + @Override public void putArray(int rowId, int offset, int length) { + columnVector.putArray(invertedIndex[rowId], offset, length); + } } http://git-wip-us.apache.org/repos/asf/carbondata/blob/bed51ba7/core/src/main/java/org/apache/carbondata/core/util/ByteUtil.java ---------------------------------------------------------------------- diff --git a/core/src/main/java/org/apache/carbondata/core/util/ByteUtil.java b/core/src/main/java/org/apache/carbondata/core/util/ByteUtil.java index 8e6e15b..fd81610 100644 --- a/core/src/main/java/org/apache/carbondata/core/util/ByteUtil.java +++ b/core/src/main/java/org/apache/carbondata/core/util/ByteUtil.java @@ -734,11 +734,27 @@ public final class ByteUtil { return Float.intBitsToFloat(toXorInt(value, offset, length)); } - public static int[] toIntArrayFrom3Bytes(byte[] data, int size) { - int[] ints = new int[size]; - for (int i = 0; i < ints.length; i++) { - ints[i] = ByteUtil.valueOf3Bytes(data, i * 3); - } - return ints; + public static int toIntLittleEndian(byte[] bytes, int offset) { + return (((int) bytes[offset + 3] & 0xff) << 24) + (((int) bytes[offset + 2] & 0xff) << 16) + ( + ((int) bytes[offset + 1] & 0xff) << 8) + ((int) bytes[offset] & 0xff); + } + + public static short toShortLittleEndian(byte[] bytes, int offset) { + return (short) ((((int) bytes[offset + 1] & 0xff) << 8) + ((int) bytes[offset] & 0xff)); + } + + public static double toDoubleLittleEndian(byte[] bytes, int offset) { + return Double.longBitsToDouble(toLongLittleEndian(bytes, offset)); + } + + public static float toFloatLittleEndian(byte[] bytes, int offset) { + return Float.intBitsToFloat(toIntLittleEndian(bytes, offset)); + } + + public static long toLongLittleEndian(byte[] bytes, int offset) { + return ((((long) bytes[offset + 7]) << 56) | (((long) bytes[offset + 6] & 0xff) << 48) | ( + ((long) bytes[offset + 5] & 0xff) << 40) | (((long) bytes[offset + 4] & 0xff) << 32) | ( + ((long) bytes[offset + 3] & 0xff) << 24) | (((long) bytes[offset + 2] & 0xff) << 16) | ( + ((long) bytes[offset + 1] & 0xff) << 8) | (((long) bytes[offset] & 0xff))); } } http://git-wip-us.apache.org/repos/asf/carbondata/blob/bed51ba7/integration/presto/src/main/java/org/apache/carbondata/presto/CarbonColumnVectorWrapper.java ---------------------------------------------------------------------- diff --git a/integration/presto/src/main/java/org/apache/carbondata/presto/CarbonColumnVectorWrapper.java b/integration/presto/src/main/java/org/apache/carbondata/presto/CarbonColumnVectorWrapper.java index 765643a..a80751f 100644 --- a/integration/presto/src/main/java/org/apache/carbondata/presto/CarbonColumnVectorWrapper.java +++ b/integration/presto/src/main/java/org/apache/carbondata/presto/CarbonColumnVectorWrapper.java @@ -305,4 +305,13 @@ public class CarbonColumnVectorWrapper implements CarbonColumnVector { lazyPage.loadPage(); } + @Override public void putArray(int rowId, int offset, int length) { + if (!filteredRows[rowId]) { + columnVector.putArray(counter++, offset, length); + } + } + + @Override public void putAllByteArray(byte[] data, int offset, int length) { + columnVector.putAllByteArray(data, offset, length); + } } http://git-wip-us.apache.org/repos/asf/carbondata/blob/bed51ba7/integration/spark-common-test/src/test/resources/IUD/negativevalue.csv ---------------------------------------------------------------------- diff --git a/integration/spark-common-test/src/test/resources/IUD/negativevalue.csv b/integration/spark-common-test/src/test/resources/IUD/negativevalue.csv new file mode 100644 index 0000000..71bf4e5 --- /dev/null +++ b/integration/spark-common-test/src/test/resources/IUD/negativevalue.csv @@ -0,0 +1,7 @@ +-30000,aaa,-300 +0,ddd,0 +-20000,bbb,-200 +70000,ggg,700 +10000,eee,100, +-10000,ccc,-100, +null,null,null \ No newline at end of file http://git-wip-us.apache.org/repos/asf/carbondata/blob/bed51ba7/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/UpdateCarbonTableTestCase.scala ---------------------------------------------------------------------- diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/UpdateCarbonTableTestCase.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/UpdateCarbonTableTestCase.scala index 50fdd0c..744a310 100644 --- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/UpdateCarbonTableTestCase.scala +++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/UpdateCarbonTableTestCase.scala @@ -772,7 +772,22 @@ class UpdateCarbonTableTestCase extends QueryTest with BeforeAndAfterAll { sql("""drop table if exists iud.dest33_part""") } - override def afterAll { + test("check data after update with row.filter pushdown as false") { + sql("""drop table if exists iud.dest33_flat""") + sql( + """create table iud.dest33_part (c1 int,c2 string, c3 short) STORED BY 'carbondata'""" + .stripMargin) + sql( + s"""LOAD DATA LOCAL INPATH '$resourcesPath/IUD/negativevalue.csv' INTO table iud + |.dest33_part options('header'='false')""".stripMargin) + sql( + """update iud.dest33_part d set (c1) = (5) where d.c1 = 0""".stripMargin).show() + checkAnswer(sql("select c3 from iud.dest33_part"), Seq(Row(-300), Row(0), Row(-200), Row(700) + , Row(100), Row(-100), Row(null))) + sql("""drop table if exists iud.dest33_part""") + } + + override def afterAll { sql("use default") sql("drop database if exists iud cascade") CarbonProperties.getInstance() http://git-wip-us.apache.org/repos/asf/carbondata/blob/bed51ba7/integration/spark-datasource/src/main/scala/org/apache/carbondata/spark/vectorreader/ColumnarVectorWrapper.java ---------------------------------------------------------------------- diff --git a/integration/spark-datasource/src/main/scala/org/apache/carbondata/spark/vectorreader/ColumnarVectorWrapper.java b/integration/spark-datasource/src/main/scala/org/apache/carbondata/spark/vectorreader/ColumnarVectorWrapper.java index 22188c4..7520ef6 100644 --- a/integration/spark-datasource/src/main/scala/org/apache/carbondata/spark/vectorreader/ColumnarVectorWrapper.java +++ b/integration/spark-datasource/src/main/scala/org/apache/carbondata/spark/vectorreader/ColumnarVectorWrapper.java @@ -346,4 +346,14 @@ class ColumnarVectorWrapper implements CarbonColumnVector { @Override public void setLazyPage(LazyPageLoader lazyPage) { lazyPage.loadPage(); } + + @Override public void putArray(int rowId, int offset, int length) { + if (!filteredRows[rowId]) { + sparkColumnVectorProxy.putArray(rowId, offset, length); + } + } + + @Override public void putAllByteArray(byte[] data, int offset, int length) { + sparkColumnVectorProxy.putAllByteArray(data, offset, length); + } } http://git-wip-us.apache.org/repos/asf/carbondata/blob/bed51ba7/integration/spark-datasource/src/main/scala/org/apache/carbondata/spark/vectorreader/ColumnarVectorWrapperDirect.java ---------------------------------------------------------------------- diff --git a/integration/spark-datasource/src/main/scala/org/apache/carbondata/spark/vectorreader/ColumnarVectorWrapperDirect.java b/integration/spark-datasource/src/main/scala/org/apache/carbondata/spark/vectorreader/ColumnarVectorWrapperDirect.java index 7c5902e..c50d060 100644 --- a/integration/spark-datasource/src/main/scala/org/apache/carbondata/spark/vectorreader/ColumnarVectorWrapperDirect.java +++ b/integration/spark-datasource/src/main/scala/org/apache/carbondata/spark/vectorreader/ColumnarVectorWrapperDirect.java @@ -231,4 +231,12 @@ class ColumnarVectorWrapperDirect implements CarbonColumnVector { @Override public void setLazyPage(LazyPageLoader lazyPage) { sparkColumnVectorProxy.setLazyPage(lazyPage); } + + @Override public void putArray(int rowId, int offset, int length) { + sparkColumnVectorProxy.putArray(rowId, offset, length); + } + + @Override public void putAllByteArray(byte[] data, int offset, int length) { + sparkColumnVectorProxy.putAllByteArray(data, offset, length); + } } http://git-wip-us.apache.org/repos/asf/carbondata/blob/bed51ba7/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 1bde17d..8082511 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 @@ -36,6 +36,8 @@ import org.apache.carbondata.core.metadata.encoder.Encoding; import org.apache.carbondata.core.scan.executor.QueryExecutor; import org.apache.carbondata.core.scan.executor.QueryExecutorFactory; import org.apache.carbondata.core.scan.executor.exception.QueryExecutionException; +import org.apache.carbondata.core.scan.expression.Expression; +import org.apache.carbondata.core.scan.expression.conditional.NotEqualsExpression; import org.apache.carbondata.core.scan.model.ProjectionDimension; import org.apache.carbondata.core.scan.model.ProjectionMeasure; import org.apache.carbondata.core.scan.model.QueryModel; @@ -282,11 +284,13 @@ public class VectorizedCarbonRecordReader extends AbstractRecordReader<Object> { schema = schema.add(field); } } + boolean useLazyLoad = false; short batchSize = DEFAULT_BATCH_SIZE; if (queryModel.isDirectVectorFill()) { batchSize = CarbonV3DataFormatConstants.NUMBER_OF_ROWS_PER_BLOCKLET_COLUMN_PAGE_DEFAULT; + useLazyLoad = isUseLazyLoad(); } - vectorProxy = new CarbonVectorProxy(DEFAULT_MEMORY_MODE, schema, batchSize); + vectorProxy = new CarbonVectorProxy(DEFAULT_MEMORY_MODE, schema, batchSize, useLazyLoad); if (partitionColumns != null) { int partitionIdx = fields.length; @@ -318,6 +322,31 @@ public class VectorizedCarbonRecordReader extends AbstractRecordReader<Object> { carbonColumnarBatch = new CarbonColumnarBatch(vectors, vectorProxy.numRows(), filteredRows); } + /** + * Whether to use lazy load in vector or not. + * @return + */ + private boolean isUseLazyLoad() { + boolean useLazyLoad = false; + if (queryModel.getFilterExpressionResolverTree() != null) { + Expression expression = + queryModel.getFilterExpressionResolverTree().getFilterExpression(); + useLazyLoad = true; + // In case of join queries only not null filter would e pushed down so check and disable the + // lazy load in that case. + if (expression instanceof NotEqualsExpression) { + try { + if (((NotEqualsExpression) expression).getRight().evaluate(null).isNull()) { + useLazyLoad = false; + } + } catch (Exception e) { + throw new RuntimeException(e); + } + } + } + return useLazyLoad; + } + private void initBatch() { initBatch(DEFAULT_MEMORY_MODE, new StructType(), InternalRow.empty()); } http://git-wip-us.apache.org/repos/asf/carbondata/blob/bed51ba7/integration/spark-datasource/src/main/scala/org/apache/spark/sql/carbondata/execution/datasources/SparkCarbonFileFormat.scala ---------------------------------------------------------------------- diff --git a/integration/spark-datasource/src/main/scala/org/apache/spark/sql/carbondata/execution/datasources/SparkCarbonFileFormat.scala b/integration/spark-datasource/src/main/scala/org/apache/spark/sql/carbondata/execution/datasources/SparkCarbonFileFormat.scala index 8c2f200..8cb2ca4 100644 --- a/integration/spark-datasource/src/main/scala/org/apache/spark/sql/carbondata/execution/datasources/SparkCarbonFileFormat.scala +++ b/integration/spark-datasource/src/main/scala/org/apache/spark/sql/carbondata/execution/datasources/SparkCarbonFileFormat.scala @@ -117,9 +117,13 @@ class SparkCarbonFileFormat extends FileFormat /** * Add our own protocol to control the commit. */ - SparkSession.getActiveSession.get.sessionState.conf.setConfString( - "spark.sql.sources.commitProtocolClass", - "org.apache.spark.sql.carbondata.execution.datasources.CarbonSQLHadoopMapReduceCommitProtocol") + SparkSession.getActiveSession match { + case Some(session) => session.sessionState.conf.setConfString( + "spark.sql.sources.commitProtocolClass", + "org.apache.spark.sql.carbondata.execution." + + "datasources.CarbonSQLHadoopMapReduceCommitProtocol") + case _ => + } /** * Prepares a write job and returns an [[OutputWriterFactory]]. Client side job preparation is http://git-wip-us.apache.org/repos/asf/carbondata/blob/bed51ba7/integration/spark-datasource/src/main/spark2.1andspark2.2/org/apache/spark/sql/CarbonVectorProxy.java ---------------------------------------------------------------------- diff --git a/integration/spark-datasource/src/main/spark2.1andspark2.2/org/apache/spark/sql/CarbonVectorProxy.java b/integration/spark-datasource/src/main/spark2.1andspark2.2/org/apache/spark/sql/CarbonVectorProxy.java index 90e2cc5..edb1020 100644 --- a/integration/spark-datasource/src/main/spark2.1andspark2.2/org/apache/spark/sql/CarbonVectorProxy.java +++ b/integration/spark-datasource/src/main/spark2.1andspark2.2/org/apache/spark/sql/CarbonVectorProxy.java @@ -31,7 +31,6 @@ import org.apache.spark.sql.types.CalendarIntervalType; import org.apache.spark.sql.types.DataType; import org.apache.spark.sql.types.Decimal; import org.apache.spark.sql.types.DecimalType; -import org.apache.spark.sql.types.StructField; import org.apache.spark.sql.types.StructType; import org.apache.spark.unsafe.types.CalendarInterval; import org.apache.spark.unsafe.types.UTF8String; @@ -47,25 +46,6 @@ public class CarbonVectorProxy { private ColumnarBatch columnarBatch; private ColumnVectorProxy[] columnVectorProxies; - /** - * Adapter class which handles the columnar vector reading of the carbondata - * based on the spark ColumnVector and ColumnarBatch API. This proxy class - * handles the complexity of spark 2.3 version related api changes since - * spark ColumnVector and ColumnarBatch interfaces are still evolving. - * - * @param memMode which represent the type onheap or offheap vector. - * @param rowNum rows number for vector reading - * @param structFileds, metadata related to current schema of table. - */ - public CarbonVectorProxy(MemoryMode memMode, int rowNum, StructField[] structFileds) { - columnarBatch = ColumnarBatch.allocate(new StructType(structFileds), memMode, rowNum); - columnVectorProxies = new ColumnVectorProxy[columnarBatch.numCols()]; - for (int i = 0; i < columnVectorProxies.length; i++) { - columnVectorProxies[i] = new ColumnVectorProxy(columnarBatch.column(i), rowNum, memMode); - } - updateColumnVectors(); - - } private void updateColumnVectors() { try { @@ -77,11 +57,28 @@ public class CarbonVectorProxy { } } - public CarbonVectorProxy(MemoryMode memMode, StructType outputSchema, int rowNum) { + /** + * Adapter class which handles the columnar vector reading of the carbondata + * based on the spark ColumnVector and ColumnarBatch API. This proxy class + * handles the complexity of spark 2.3 version related api changes since + * spark ColumnVector and ColumnarBatch interfaces are still evolving. + * + * @param memMode which represent the type onheap or offheap vector. + * @param outputSchema, metadata related to current schema of table. + * @param rowNum rows number for vector reading + * @param useLazyLoad Whether to use lazy load while getting the data. + */ + public CarbonVectorProxy(MemoryMode memMode, StructType outputSchema, int rowNum, + boolean useLazyLoad) { columnarBatch = ColumnarBatch.allocate(outputSchema, memMode, rowNum); columnVectorProxies = new ColumnVectorProxy[columnarBatch.numCols()]; for (int i = 0; i < columnVectorProxies.length; i++) { - columnVectorProxies[i] = new ColumnVectorProxy(columnarBatch.column(i), rowNum, memMode); + if (useLazyLoad) { + columnVectorProxies[i] = + new ColumnVectorProxyWithLazyLoad(columnarBatch.column(i), rowNum, memMode); + } else { + columnVectorProxies[i] = new ColumnVectorProxy(columnarBatch.column(i), rowNum, memMode); + } } updateColumnVectors(); } @@ -158,10 +155,6 @@ public class CarbonVectorProxy { private ColumnVector vector; - private LazyPageLoader pageLoad; - - private boolean isLoaded; - public ColumnVectorProxy(ColumnVector columnVector, int capacity, MemoryMode mode) { super(capacity, columnVector.dataType(), mode); try { @@ -170,16 +163,14 @@ public class CarbonVectorProxy { childColumns.setAccessible(true); Object o = childColumns.get(columnVector); childColumns.set(this, o); - Field childColumns1 = + Field resultArray = columnVector.getClass().getSuperclass().getDeclaredField("resultArray"); - childColumns1.setAccessible(true); - Object o1 = childColumns1.get(columnVector); - childColumns1.set(this, o1); - + resultArray.setAccessible(true); + Object o1 = resultArray.get(columnVector); + resultArray.set(this, o1); } catch (Exception e) { throw new RuntimeException(e); } - vector = columnVector; } @@ -342,42 +333,34 @@ public class CarbonVectorProxy { } @Override public boolean isNullAt(int i) { - checkPageLoaded(); return vector.isNullAt(i); } @Override public boolean getBoolean(int i) { - checkPageLoaded(); return vector.getBoolean(i); } @Override public byte getByte(int i) { - checkPageLoaded(); return vector.getByte(i); } @Override public short getShort(int i) { - checkPageLoaded(); return vector.getShort(i); } @Override public int getInt(int i) { - checkPageLoaded(); return vector.getInt(i); } @Override public long getLong(int i) { - checkPageLoaded(); return vector.getLong(i); } @Override public float getFloat(int i) { - checkPageLoaded(); return vector.getFloat(i); } @Override public double getDouble(int i) { - checkPageLoaded(); return vector.getDouble(i); } @@ -433,12 +416,10 @@ public class CarbonVectorProxy { } @Override public int getArrayLength(int rowId) { - checkPageLoaded(); return vector.getArrayLength(rowId); } @Override public int getArrayOffset(int rowId) { - checkPageLoaded(); return vector.getArrayOffset(rowId); } @@ -450,10 +431,98 @@ public class CarbonVectorProxy { return vector.putByteArray(rowId, value, offset, count); } + /** + * It keeps all binary data of all rows to it. + * Should use along with @{putArray(int rowId, int offset, int length)} to keep lengths + * and offset. + */ + public void putAllByteArray(byte[] data, int offset, int length) { + vector.arrayData().appendBytes(length, data, offset); + } + @Override public void close() { vector.close(); } + public void reset() { + if (isConstant) { + return; + } + vector.reset(); + } + + public void setLazyPage(LazyPageLoader lazyPage) { + lazyPage.loadPage(); + } + + public ColumnVector getVector() { + return vector; + } + } + + public static class ColumnVectorProxyWithLazyLoad extends ColumnVectorProxy { + + private ColumnVector vector; + + private LazyPageLoader pageLoad; + + private boolean isLoaded; + + public ColumnVectorProxyWithLazyLoad(ColumnVector columnVector, int capacity, MemoryMode mode) { + super(columnVector, capacity, mode); + vector = columnVector; + } + + @Override public boolean isNullAt(int i) { + checkPageLoaded(); + return vector.isNullAt(i); + } + + @Override public boolean getBoolean(int i) { + checkPageLoaded(); + return vector.getBoolean(i); + } + + @Override public byte getByte(int i) { + checkPageLoaded(); + return vector.getByte(i); + } + + @Override public short getShort(int i) { + checkPageLoaded(); + return vector.getShort(i); + } + + @Override public int getInt(int i) { + checkPageLoaded(); + return vector.getInt(i); + } + + @Override public long getLong(int i) { + checkPageLoaded(); + return vector.getLong(i); + } + + @Override public float getFloat(int i) { + checkPageLoaded(); + return vector.getFloat(i); + } + + @Override public double getDouble(int i) { + checkPageLoaded(); + return vector.getDouble(i); + } + + @Override public int getArrayLength(int rowId) { + checkPageLoaded(); + return vector.getArrayLength(rowId); + } + + @Override public int getArrayOffset(int rowId) { + checkPageLoaded(); + return vector.getArrayOffset(rowId); + } + private void checkPageLoaded() { if (!isLoaded) { if (pageLoad != null) { @@ -475,8 +544,5 @@ public class CarbonVectorProxy { this.pageLoad = lazyPage; } - public ColumnVector getVector() { - return vector; - } } } http://git-wip-us.apache.org/repos/asf/carbondata/blob/bed51ba7/integration/spark-datasource/src/main/spark2.3plus/org/apache/spark/sql/CarbonVectorProxy.java ---------------------------------------------------------------------- diff --git a/integration/spark-datasource/src/main/spark2.3plus/org/apache/spark/sql/CarbonVectorProxy.java b/integration/spark-datasource/src/main/spark2.3plus/org/apache/spark/sql/CarbonVectorProxy.java index c8c4e2c..76a97ee 100644 --- a/integration/spark-datasource/src/main/spark2.3plus/org/apache/spark/sql/CarbonVectorProxy.java +++ b/integration/spark-datasource/src/main/spark2.3plus/org/apache/spark/sql/CarbonVectorProxy.java @@ -43,33 +43,29 @@ public class CarbonVectorProxy { private ColumnarBatch columnarBatch; private ColumnVectorProxy[] columnVectorProxies; - /** - * Adapter class which handles the columnar vector reading of the carbondata - * based on the spark ColumnVector and ColumnarBatch API. This proxy class - * handles the complexity of spark 2.3 version related api changes since - * spark ColumnVector and ColumnarBatch interfaces are still evolving. - * - * @param memMode which represent the type onheap or offheap vector. - * @param rowNum rows number for vector reading - * @param structFileds, metadata related to current schema of table. - */ - public CarbonVectorProxy(MemoryMode memMode, int rowNum, StructField[] structFileds) { - WritableColumnVector[] columnVectors = - ColumnVectorFactory.getColumnVector(memMode, new StructType(structFileds), rowNum); - columnVectorProxies = new ColumnVectorProxy[columnVectors.length]; - for (int i = 0; i < columnVectorProxies.length; i++) { - columnVectorProxies[i] = new ColumnVectorProxy(columnVectors[i]); - } - columnarBatch = new ColumnarBatch(columnVectorProxies); - columnarBatch.setNumRows(rowNum); - } - public CarbonVectorProxy(MemoryMode memMode, StructType outputSchema, int rowNum) { + /** + * Adapter class which handles the columnar vector reading of the carbondata + * based on the spark ColumnVector and ColumnarBatch API. This proxy class + * handles the complexity of spark 2.3 version related api changes since + * spark ColumnVector and ColumnarBatch interfaces are still evolving. + * + * @param memMode which represent the type onheap or offheap vector. + * @param outputSchema, metadata related to current schema of table. + * @param rowNum rows number for vector reading + * @param useLazyLoad Whether to use lazy load while getting the data. + */ + public CarbonVectorProxy(MemoryMode memMode, StructType outputSchema, int rowNum, + boolean useLazyLoad) { WritableColumnVector[] columnVectors = ColumnVectorFactory .getColumnVector(memMode, outputSchema, rowNum); columnVectorProxies = new ColumnVectorProxy[columnVectors.length]; for (int i = 0; i < columnVectorProxies.length; i++) { + if (useLazyLoad) { + columnVectorProxies[i] = new ColumnVectorProxyWithLazyLoad(columnVectors[i]); + } else { columnVectorProxies[i] = new ColumnVectorProxy(columnVectors[i]); + } } columnarBatch = new ColumnarBatch(columnVectorProxies); columnarBatch.setNumRows(rowNum); @@ -148,10 +144,6 @@ public class CarbonVectorProxy { private WritableColumnVector vector; - private LazyPageLoader pageLoad; - - private boolean isLoaded; - public ColumnVectorProxy(ColumnVector columnVector) { super(columnVector.dataType()); vector = (WritableColumnVector) columnVector; @@ -321,42 +313,34 @@ public class CarbonVectorProxy { } @Override public boolean isNullAt(int i) { - checkPageLoaded(); return vector.isNullAt(i); } @Override public boolean getBoolean(int i) { - checkPageLoaded(); return vector.getBoolean(i); } @Override public byte getByte(int i) { - checkPageLoaded(); return vector.getByte(i); } @Override public short getShort(int i) { - checkPageLoaded(); return vector.getShort(i); } @Override public int getInt(int i) { - checkPageLoaded(); return vector.getInt(i); } @Override public long getLong(int i) { - checkPageLoaded(); return vector.getLong(i); } @Override public float getFloat(int i) { - checkPageLoaded(); return vector.getFloat(i); } @Override public double getDouble(int i) { - checkPageLoaded(); return vector.getDouble(i); } @@ -365,66 +349,174 @@ public class CarbonVectorProxy { } @Override public boolean hasNull() { - checkPageLoaded(); return vector.hasNull(); } @Override public int numNulls() { - checkPageLoaded(); return vector.numNulls(); } @Override public ColumnarArray getArray(int i) { - checkPageLoaded(); return vector.getArray(i); } @Override public ColumnarMap getMap(int i) { - checkPageLoaded(); return vector.getMap(i); } @Override public Decimal getDecimal(int i, int i1, int i2) { - checkPageLoaded(); return vector.getDecimal(i, i1, i2); } @Override public UTF8String getUTF8String(int i) { - checkPageLoaded(); return vector.getUTF8String(i); } @Override public byte[] getBinary(int i) { - checkPageLoaded(); return vector.getBinary(i); } @Override protected ColumnVector getChild(int i) { - checkPageLoaded(); return vector.getChild(i); } - private void checkPageLoaded() { - if (!isLoaded) { - if (pageLoad != null) { - pageLoad.loadPage(); - } - isLoaded = true; - } - } - public void reset() { - isLoaded = false; - pageLoad = null; vector.reset(); } public void setLazyPage(LazyPageLoader lazyPage) { - this.pageLoad = lazyPage; + lazyPage.loadPage(); } - public WritableColumnVector getVector() { + /** + * It keeps all binary data of all rows to it. + * Should use along with @{putArray(int rowId, int offset, int length)} to keep lengths + * and offset. + */ + public void putAllByteArray(byte[] data, int offset, int length) { + vector.arrayData().appendBytes(length, data, offset); + } + + public void putArray(int rowId, int offset, int length) { + vector.putArray(rowId, offset, length); + } + + public WritableColumnVector getVector() { return vector; } } + + public static class ColumnVectorProxyWithLazyLoad extends ColumnVectorProxy { + + private WritableColumnVector vector; + + private LazyPageLoader pageLoad; + + private boolean isLoaded; + + public ColumnVectorProxyWithLazyLoad(ColumnVector columnVector) { + super(columnVector); + vector = (WritableColumnVector) columnVector; + } + + @Override public boolean isNullAt(int i) { + checkPageLoaded(); + return vector.isNullAt(i); + } + + @Override public boolean getBoolean(int i) { + checkPageLoaded(); + return vector.getBoolean(i); + } + + @Override public byte getByte(int i) { + checkPageLoaded(); + return vector.getByte(i); + } + + @Override public short getShort(int i) { + checkPageLoaded(); + return vector.getShort(i); + } + + @Override public int getInt(int i) { + checkPageLoaded(); + return vector.getInt(i); + } + + @Override public long getLong(int i) { + checkPageLoaded(); + return vector.getLong(i); + } + + @Override public float getFloat(int i) { + checkPageLoaded(); + return vector.getFloat(i); + } + + @Override public double getDouble(int i) { + checkPageLoaded(); + return vector.getDouble(i); + } + + @Override public boolean hasNull() { + checkPageLoaded(); + return vector.hasNull(); + } + + @Override public int numNulls() { + checkPageLoaded(); + return vector.numNulls(); + } + + @Override public ColumnarArray getArray(int i) { + checkPageLoaded(); + return vector.getArray(i); + } + + @Override public ColumnarMap getMap(int i) { + checkPageLoaded(); + return vector.getMap(i); + } + + @Override public Decimal getDecimal(int i, int i1, int i2) { + checkPageLoaded(); + return vector.getDecimal(i, i1, i2); + } + + @Override public UTF8String getUTF8String(int i) { + checkPageLoaded(); + return vector.getUTF8String(i); + } + + @Override public byte[] getBinary(int i) { + checkPageLoaded(); + return vector.getBinary(i); + } + + @Override protected ColumnVector getChild(int i) { + checkPageLoaded(); + return vector.getChild(i); + } + + public void reset() { + isLoaded = false; + pageLoad = null; + vector.reset(); + } + + private void checkPageLoaded() { + if (!isLoaded) { + if (pageLoad != null) { + pageLoad.loadPage(); + } + isLoaded = true; + } + } + + public void setLazyPage(LazyPageLoader lazyPage) { + this.pageLoad = lazyPage; + } + + } } http://git-wip-us.apache.org/repos/asf/carbondata/blob/bed51ba7/integration/spark2/src/main/scala/org/apache/carbondata/stream/CarbonStreamRecordReader.java ---------------------------------------------------------------------- diff --git a/integration/spark2/src/main/scala/org/apache/carbondata/stream/CarbonStreamRecordReader.java b/integration/spark2/src/main/scala/org/apache/carbondata/stream/CarbonStreamRecordReader.java index b66e446..50d6c46 100644 --- a/integration/spark2/src/main/scala/org/apache/carbondata/stream/CarbonStreamRecordReader.java +++ b/integration/spark2/src/main/scala/org/apache/carbondata/stream/CarbonStreamRecordReader.java @@ -438,14 +438,15 @@ public class CarbonStreamRecordReader extends RecordReader<Void, Object> { // if filter is null and output projection is empty, use the row number of blocklet header if (skipScanData) { int rowNums = header.getBlocklet_info().getNum_rows(); - vectorProxy= new CarbonVectorProxy(MemoryMode.OFF_HEAP,outputSchema,rowNums); + vectorProxy = new CarbonVectorProxy(MemoryMode.OFF_HEAP, outputSchema, rowNums, false); vectorProxy.setNumRows(rowNums); input.skipBlockletData(true); return rowNums > 0; } input.readBlockletData(header); - vectorProxy= new CarbonVectorProxy(MemoryMode.OFF_HEAP,outputSchema,input.getRowNums()); + vectorProxy = + new CarbonVectorProxy(MemoryMode.OFF_HEAP, outputSchema, input.getRowNums(), false); int rowNum = 0; if (null == filter) { while (input.hasNext()) { http://git-wip-us.apache.org/repos/asf/carbondata/blob/bed51ba7/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDictionaryDecoder.scala ---------------------------------------------------------------------- diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDictionaryDecoder.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDictionaryDecoder.scala index c9434a1..95ab29d 100644 --- a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDictionaryDecoder.scala +++ b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDictionaryDecoder.scala @@ -37,7 +37,7 @@ import org.apache.spark.util.SerializableConfiguration import org.apache.carbondata.core.cache.{Cache, CacheProvider, CacheType} import org.apache.carbondata.core.cache.dictionary.{Dictionary, DictionaryColumnUniqueIdentifier} import org.apache.carbondata.core.constants.CarbonCommonConstants -import org.apache.carbondata.core.metadata.{CarbonMetadata, ColumnIdentifier} +import org.apache.carbondata.core.metadata.ColumnIdentifier import org.apache.carbondata.core.metadata.datatype.{DataTypes => CarbonDataTypes} import org.apache.carbondata.core.metadata.encoder.Encoding import org.apache.carbondata.core.metadata.schema.table.CarbonTable @@ -145,79 +145,141 @@ case class CarbonDictionaryDecoder( ExpressionCanonicalizer.execute(BindReferences.bindReference(exp, child.output)) } ctx.currentVars = input + val dictTuple = exprs.map(e => new DictTuple(null, false)) + val decodeDictionary = ctx.freshName("deDict") + ctx.addNewFunction(decodeDictionary, + s""" + |private org.apache.spark.sql.DictTuple $decodeDictionary( + | org.apache.spark.sql.ForwardDictionaryWrapper dict, int surg) + | throws java.io.IOException { + | boolean isNull = false; + | byte[] valueIntern = dict.getDictionaryValueForKeyInBytes(surg); + | if (valueIntern == null || + | java.util.Arrays.equals(org.apache.carbondata.core.constants + | .CarbonCommonConstants.MEMBER_DEFAULT_VAL_ARRAY, valueIntern)) { + | isNull = true; + | valueIntern = org.apache.carbondata.core.constants + | .CarbonCommonConstants.ZERO_BYTE_ARRAY; + | } + | return new org.apache.spark.sql.DictTuple(valueIntern, isNull); + |}""".stripMargin) + + val decodeDecimal = ctx.freshName("deDictDec") + ctx.addNewFunction(decodeDecimal, + s""" + |private org.apache.spark.sql.DictTuple $decodeDecimal( + | org.apache.spark.sql.ForwardDictionaryWrapper dict, int surg) + | throws java.io.IOException { + | org.apache.spark.sql.DictTuple tuple = $decodeDictionary(dict, surg); + | tuple.setValue(org.apache.spark.sql.types.Decimal.apply(new java.math.BigDecimal( + | new String((byte[])tuple.getValue(), + | org.apache.carbondata.core.constants.CarbonCommonConstants.DEFAULT_CHARSET_CLASS)))); + | return tuple; + |}""".stripMargin) + + val decodeInt = ctx.freshName("deDictInt") + ctx.addNewFunction(decodeInt, + s""" + |private org.apache.spark.sql.DictTuple $decodeInt( + | org.apache.spark.sql.ForwardDictionaryWrapper dict, int surg) + | throws java.io.IOException { + | org.apache.spark.sql.DictTuple tuple = $decodeDictionary(dict, surg); + | tuple.setValue(Integer.parseInt(new String((byte[])tuple.getValue(), + | org.apache.carbondata.core.constants.CarbonCommonConstants.DEFAULT_CHARSET_CLASS))); + | return tuple; + |}""".stripMargin) + val decodeShort = ctx.freshName("deDictShort") + ctx.addNewFunction(decodeShort, + s""" + |private org.apache.spark.sql.DictTuple $decodeShort( + | org.apache.spark.sql.ForwardDictionaryWrapper dict, int surg) + | throws java.io.IOException { + | org.apache.spark.sql.DictTuple tuple = $decodeDictionary(dict, surg); + | tuple.setValue(Short.parseShort(new String((byte[])tuple.getValue(), + | org.apache.carbondata.core.constants.CarbonCommonConstants.DEFAULT_CHARSET_CLASS))); + | return tuple; + |}""".stripMargin) + val decodeDouble = ctx.freshName("deDictDoub") + ctx.addNewFunction(decodeDouble, + s""" + |private org.apache.spark.sql.DictTuple $decodeDouble( + | org.apache.spark.sql.ForwardDictionaryWrapper dict, int surg) + | throws java.io.IOException { + | org.apache.spark.sql.DictTuple tuple = $decodeDictionary(dict, surg); + | tuple.setValue(Double.parseDouble(new String((byte[])tuple.getValue(), + | org.apache.carbondata.core.constants.CarbonCommonConstants.DEFAULT_CHARSET_CLASS))); + | return tuple; + |}""".stripMargin) + val decodeLong = ctx.freshName("deDictLong") + ctx.addNewFunction(decodeLong, + s""" + |private org.apache.spark.sql.DictTuple $decodeLong( + | org.apache.spark.sql.ForwardDictionaryWrapper dict, int surg) + | throws java.io.IOException { + | org.apache.spark.sql.DictTuple tuple = $decodeDictionary(dict, surg); + | tuple.setValue(Long.parseLong(new String((byte[])tuple.getValue(), + | org.apache.carbondata.core.constants.CarbonCommonConstants.DEFAULT_CHARSET_CLASS))); + | return tuple; + |}""".stripMargin) + val decodeStr = ctx.freshName("deDictStr") + ctx.addNewFunction(decodeStr, + s""" + |private org.apache.spark.sql.DictTuple $decodeStr( + | org.apache.spark.sql.ForwardDictionaryWrapper dict, int surg) + | throws java.io.IOException { + | org.apache.spark.sql.DictTuple tuple = $decodeDictionary(dict, surg); + | tuple.setValue(UTF8String.fromBytes((byte[])tuple.getValue())); + | return tuple; + |}""".stripMargin) + + val resultVars = exprs.zipWithIndex.map { case (expr, index) => if (dicts(index) != null) { val ev = expr.genCode(ctx) - val value = ctx.freshName("value") - val valueIntern = ctx.freshName("valueIntern") - val isNull = ctx.freshName("isNull") - val dictsRef = ctx.addReferenceObj("dictsRef", dicts(index)) + val dictRef = ctx.addReferenceObj("df", dicts(index)) + val value = ctx.freshName("v") var code = s""" |${ev.code} """.stripMargin - code += + if (CarbonDataTypes.isDecimal(getDictionaryColumnIds(index)._3.getDataType)) { + code += s""" - |boolean $isNull = false; - |byte[] $valueIntern = $dictsRef.getDictionaryValueForKeyInBytes(${ ev.value }); - |if ($valueIntern == null || - | java.util.Arrays.equals(org.apache.carbondata.core.constants - |.CarbonCommonConstants.MEMBER_DEFAULT_VAL_ARRAY, $valueIntern)) { - | $isNull = true; - | $valueIntern = org.apache.carbondata.core.constants - | .CarbonCommonConstants.ZERO_BYTE_ARRAY; - |} - """.stripMargin - - val caseCode = - if (CarbonDataTypes.isDecimal(getDictionaryColumnIds(index)._3.getDataType)) { - s""" - |org.apache.spark.sql.types.Decimal $value = - |Decimal.apply(new java.math.BigDecimal( - |new String($valueIntern, org.apache.carbondata.core.constants - |.CarbonCommonConstants.DEFAULT_CHARSET_CLASS))); + |org.apache.spark.sql.DictTuple $value = $decodeDecimal($dictRef, ${ev.value}); """.stripMargin - } else { - getDictionaryColumnIds(index)._3.getDataType match { - case CarbonDataTypes.INT => - s""" - |int $value = Integer.parseInt(new String($valueIntern, - |org.apache.carbondata.core.constants.CarbonCommonConstants - |.DEFAULT_CHARSET_CLASS)); + ExprCode(code, s"$value.getIsNull()", + s"(org.apache.spark.sql.types.Decimal)$value.getValue()") + } else { + getDictionaryColumnIds(index)._3.getDataType match { + case CarbonDataTypes.INT => code += + s""" + |org.apache.spark.sql.DictTuple $value = $decodeInt($dictRef, ${ ev.value }); """.stripMargin - case CarbonDataTypes.SHORT => - s""" - |short $value = - |Short.parseShort(new String($valueIntern, - |org.apache.carbondata.core.constants.CarbonCommonConstants - |.DEFAULT_CHARSET_CLASS)); + ExprCode(code, s"$value.getIsNull()", s"(Integer)$value.getValue()") + case CarbonDataTypes.SHORT => code += + s""" + |org.apache.spark.sql.DictTuple $value = $decodeShort($dictRef, ${ ev.value }); """.stripMargin - case CarbonDataTypes.DOUBLE => - s""" - |double $value = - |Double.parseDouble(new String($valueIntern, - |org.apache.carbondata.core.constants.CarbonCommonConstants - |.DEFAULT_CHARSET_CLASS)); + ExprCode(code, s"$value.getIsNull()", s"(Short)$value.getValue()") + case CarbonDataTypes.DOUBLE => code += + s""" + |org.apache.spark.sql.DictTuple $value = $decodeDouble($dictRef, ${ ev.value }); """.stripMargin - case CarbonDataTypes.LONG => - s""" - |long $value = - |Long.parseLong(new String($valueIntern, - |org.apache.carbondata.core.constants.CarbonCommonConstants - |.DEFAULT_CHARSET_CLASS)); + ExprCode(code, s"$value.getIsNull()", s"(Double)$value.getValue()") + case CarbonDataTypes.LONG => code += + s""" + |org.apache.spark.sql.DictTuple $value = $decodeLong($dictRef, ${ ev.value }); """.stripMargin - case _ => - s""" - | UTF8String $value = UTF8String.fromBytes($valueIntern); + ExprCode(code, s"$value.getIsNull()", s"(Long)$value.getValue()") + case _ => code += + s""" + |org.apache.spark.sql.DictTuple $value = $decodeStr($dictRef, ${ev.value}); """.stripMargin - } - } - code += - s""" - |$caseCode - """.stripMargin + ExprCode(code, s"$value.getIsNull()", s"(UTF8String)$value.getValue()") + + } + } - ExprCode(code, isNull, value) } else { expr.genCode(ctx) } @@ -588,6 +650,21 @@ class ForwardDictionaryWrapper( } } +class DictTuple(var value: AnyRef, var isNull: Boolean) extends Serializable { + + def getValue: AnyRef = value + + def getIsNull: Boolean = isNull + + def setValue(value: AnyRef): Unit = { + this.value = value + } + + def setIsNull(isNull: Boolean): Unit = { + this.isNull = isNull + } +} + /** * It is Dictionary Loader class to load all dictionaries at a time instead of one by one. */
