Github user manishgupta88 commented on a diff in the pull request:
https://github.com/apache/carbondata/pull/2819#discussion_r225836328
--- Diff:
core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/compress/DirectCompressCodec.java
---
@@ -178,6 +196,143 @@ public double decodeDouble(float value) {
public double decodeDouble(double value) {
return value;
}
+
+ @Override public void decodeAndFillVector(ColumnPage columnPage,
ColumnVectorInfo vectorInfo) {
+ CarbonColumnVector vector = vectorInfo.vector;
+ BitSet nullBits = columnPage.getNullBits();
+ DataType dataType = vector.getType();
+ DataType type = columnPage.getDataType();
+ int pageSize = columnPage.getPageSize();
+ BitSet deletedRows = vectorInfo.deletedRows;
+ fillVector(columnPage, vector, dataType, type, pageSize, vectorInfo);
+ if (deletedRows == null || deletedRows.isEmpty()) {
+ for (int i = nullBits.nextSetBit(0); i >= 0; i =
nullBits.nextSetBit(i + 1)) {
+ vector.putNull(i);
+ }
+ }
+ }
+
+ private void fillVector(ColumnPage columnPage, CarbonColumnVector
vector, DataType dataType,
+ DataType type, int pageSize, ColumnVectorInfo vectorInfo) {
+ if (type == DataTypes.BOOLEAN || type == DataTypes.BYTE) {
+ byte[] byteData = columnPage.getByteData();
+ if (dataType == DataTypes.SHORT) {
+ for (int i = 0; i < pageSize; i++) {
+ vector.putShort(i, (short) byteData[i]);
+ }
+ } else if (dataType == DataTypes.INT) {
+ for (int i = 0; i < pageSize; i++) {
+ vector.putInt(i, (int) byteData[i]);
+ }
+ } else if (dataType == DataTypes.LONG) {
+ for (int i = 0; i < pageSize; i++) {
+ vector.putLong(i, byteData[i]);
+ }
+ } else if (dataType == DataTypes.TIMESTAMP) {
+ for (int i = 0; i < pageSize; i++) {
+ vector.putLong(i, byteData[i] * 1000);
+ }
+ } else if (dataType == DataTypes.BOOLEAN || dataType ==
DataTypes.BYTE) {
+ vector.putBytes(0, pageSize, byteData, 0);
+ } else if (DataTypes.isDecimal(dataType)) {
+ DecimalConverterFactory.DecimalConverter decimalConverter =
vectorInfo.decimalConverter;
+ decimalConverter.fillVector(byteData, pageSize, vectorInfo,
columnPage.getNullBits());
+ } else {
+ for (int i = 0; i < pageSize; i++) {
+ vector.putDouble(i, byteData[i]);
+ }
+ }
+ } else if (type == DataTypes.SHORT) {
+ short[] shortData = columnPage.getShortData();
+ if (dataType == DataTypes.SHORT) {
+ vector.putShorts(0, pageSize, shortData, 0);
+ } else if (dataType == DataTypes.INT) {
+ for (int i = 0; i < pageSize; i++) {
+ vector.putInt(i, (int) shortData[i]);
+ }
+ } else if (dataType == DataTypes.LONG) {
+ for (int i = 0; i < pageSize; i++) {
+ vector.putLong(i, shortData[i]);
+ }
+ } else if (dataType == DataTypes.TIMESTAMP) {
+ for (int i = 0; i < pageSize; i++) {
+ vector.putLong(i, shortData[i] * 1000);
+ }
+ } else if (DataTypes.isDecimal(dataType)) {
+ DecimalConverterFactory.DecimalConverter decimalConverter =
vectorInfo.decimalConverter;
+ decimalConverter.fillVector(shortData, pageSize, vectorInfo,
columnPage.getNullBits());
+ } else {
+ for (int i = 0; i < pageSize; i++) {
+ vector.putDouble(i, shortData[i]);
+ }
+ }
+
+ } else if (type == DataTypes.SHORT_INT) {
+ int[] shortIntData = columnPage.getShortIntData();
+ if (dataType == DataTypes.INT) {
+ vector.putInts(0, pageSize, shortIntData, 0);
+ } else if (dataType == DataTypes.LONG) {
+ for (int i = 0; i < pageSize; i++) {
+ vector.putLong(i, shortIntData[i]);
+ }
+ } else if (dataType == DataTypes.TIMESTAMP) {
+ for (int i = 0; i < pageSize; i++) {
+ vector.putLong(i, shortIntData[i] * 1000);
+ }
+ } else if (DataTypes.isDecimal(dataType)) {
+ DecimalConverterFactory.DecimalConverter decimalConverter =
vectorInfo.decimalConverter;
+ decimalConverter.fillVector(shortIntData, pageSize, vectorInfo,
columnPage.getNullBits());
+ } else {
+ for (int i = 0; i < pageSize; i++) {
+ vector.putDouble(i, shortIntData[i]);
+ }
+ }
+ } else if (type == DataTypes.INT) {
+ int[] intData = columnPage.getIntData();
+ if (dataType == DataTypes.INT) {
+ vector.putInts(0, pageSize, intData, 0);
+ } else if (dataType == DataTypes.LONG) {
+ for (int i = 0; i < pageSize; i++) {
+ vector.putLong(i, intData[i]);
+ }
+ } else if (dataType == DataTypes.TIMESTAMP) {
+ for (int i = 0; i < pageSize; i++) {
+ vector.putLong(i, intData[i] * 1000);
+ }
+ } else if (DataTypes.isDecimal(dataType)) {
+ DecimalConverterFactory.DecimalConverter decimalConverter =
vectorInfo.decimalConverter;
+ decimalConverter.fillVector(intData, pageSize, vectorInfo,
columnPage.getNullBits());
+ } else {
+ for (int i = 0; i < pageSize; i++) {
+ vector.putDouble(i, intData[i]);
+ }
+ }
+ } else if (type == DataTypes.LONG) {
+ long[] longData = columnPage.getLongData();
+ if (dataType == DataTypes.LONG) {
+ for (int i = 0; i < pageSize; i++) {
+ vector.putLong(i, longData[i]);
+ }
+ vector.putLongs(0, pageSize, longData, 0);
+ } else if (dataType == DataTypes.TIMESTAMP) {
+ for (int i = 0; i < pageSize; i++) {
+ vector.putLong(i, longData[i] * 1000);
+ }
+ } else if (DataTypes.isDecimal(dataType)) {
+ DecimalConverterFactory.DecimalConverter decimalConverter =
vectorInfo.decimalConverter;
+ decimalConverter.fillVector(longData, pageSize, vectorInfo,
columnPage.getNullBits());
+ }
+ } else if (DataTypes.isDecimal(type)) {
+ if (DataTypes.isDecimal(dataType)) {
--- End diff --
Already else if block is checking for decimal type..no need to add one more
if check inside it for checking decimal type
---
