[GitHub] [carbondata] akkio-97 opened a new pull request #3773: [CARBONDATA-3830]Presto complex columns read support

[GitBox]

ajantha-bhat commented on a change in pull request #3773:
URL: https://github.com/apache/carbondata/pull/3773#discussion_r469704722



##########
File path: 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 java.util.Stack;

Review comment:
       You have missed to handle adaptive **delta** flows
   AdaptiveDeltaIntegralCodec
   AdaptiveDeltaFloatingCodec
   




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
[hidden email]

[GitBox]

ajantha-bhat commented on a change in pull request #3773:
URL: https://github.com/apache/carbondata/pull/3773#discussion_r469722186



##########
File path: core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/FillVector.java
##########
@@ -0,0 +1,347 @@
+/*
+ * 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.encoding;

Review comment:
       please remove this class and handle in the original class. It's not common code to keep at one place




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
[hidden email]

[GitBox]

ajantha-bhat commented on a change in pull request #3773:
URL: https://github.com/apache/carbondata/pull/3773#discussion_r469858716



##########
File path: core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/FillVector.java
##########
@@ -0,0 +1,347 @@
+/*
+ * 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.encoding;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.BitSet;
+
+import org.apache.carbondata.core.metadata.datatype.DataType;
+import org.apache.carbondata.core.metadata.datatype.DataTypes;
+import org.apache.carbondata.core.metadata.datatype.DecimalConverterFactory;
+import org.apache.carbondata.core.scan.result.vector.CarbonColumnVector;
+import org.apache.carbondata.core.scan.result.vector.ColumnVectorInfo;
+import org.apache.carbondata.core.scan.result.vector.impl.CarbonColumnVectorImpl;
+import org.apache.carbondata.core.util.ByteUtil;
+
+public class FillVector {
+  private byte[] pageData;
+  private float floatFactor = 0;
+  private double factor = 0;
+  private ColumnVectorInfo vectorInfo;
+  private BitSet nullBits;
+
+  public FillVector(byte[] pageData, ColumnVectorInfo vectorInfo, BitSet nullBits) {
+    this.pageData = pageData;
+    this.vectorInfo = vectorInfo;
+    this.nullBits = nullBits;
+  }
+
+  public void setFactor(double factor) {
+    this.factor = factor;
+  }
+
+  public void setFloatFactor(float floatFactor) {
+    this.floatFactor = floatFactor;
+  }
+
+  public void basedOnType(CarbonColumnVector vector, DataType vectorDataType, int pageSize,
+      DataType pageDataType) {
+    if (vectorInfo.vector.getColumnVector() != null && ((CarbonColumnVectorImpl) vectorInfo.vector
+        .getColumnVector()).isComplex()) {
+      fillComplexType(vector.getColumnVector(), pageDataType);
+    } else {
+      fillPrimitiveType(vector, vectorDataType, pageSize, pageDataType);
+      vector.setIndex(0);
+    }
+  }
+
+  private void fillComplexType(CarbonColumnVector vector, DataType pageDataType) {
+    CarbonColumnVectorImpl vectorImpl = (CarbonColumnVectorImpl) vector;
+    if (vector != null && vector.getChildrenVector() != null) {
+      ArrayList<Integer> childElements = ((CarbonColumnVectorImpl) vector).getChildrenElements();
+      for (int i = 0; i < childElements.size(); i++) {
+        int count = childElements.get(i);
+        typeComplexObject(vectorImpl.getChildrenVector().get(0), count, pageDataType);
+        vector.putArrayObject();
+      }
+      vectorImpl.getChildrenVector().get(0).setIndex(0);
+    }
+  }
+
+  private void fillPrimitiveType(CarbonColumnVector vector, DataType vectorDataType, int pageSize,
+      DataType pageDataType) {
+    // offset which denotes the start index for pageData
+    int pageIndex = vector.getIndex();
+    int rowId = 0;
+
+    // Filling into vector is done based on page data type
+
+    if (vectorDataType == DataTypes.FLOAT && floatFactor != 0.0) {
+      if (pageDataType == DataTypes.BOOLEAN || pageDataType == DataTypes.BYTE) {
+        for (int i = 0; i < pageSize; i++) {
+          vector.putFloat(i, (pageData[pageIndex++] / floatFactor));
+        }
+      } else if (pageDataType == DataTypes.SHORT) {
+        int size = pageSize * DataTypes.SHORT.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.SHORT.getSizeInBytes()) {
+          vector.putFloat(rowId++,
+              (ByteUtil.toShortLittleEndian(pageData, pageIndex + i) / floatFactor));
+        }
+        pageIndex += size;
+      } else if (pageDataType == DataTypes.SHORT_INT) {
+        int size = pageSize * DataTypes.SHORT_INT.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.SHORT_INT.getSizeInBytes()) {
+          vector.putFloat(rowId++, (ByteUtil.valueOf3Bytes(pageData, pageIndex + i) / floatFactor));
+        }
+        pageIndex += size;
+      } else if (pageDataType == DataTypes.INT) {
+        int size = pageSize * DataTypes.INT.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.INT.getSizeInBytes()) {
+          vector.putFloat(rowId++,
+              (ByteUtil.toIntLittleEndian(pageData, pageIndex + i) / floatFactor));
+        }
+        pageIndex += size;
+      } else if (pageDataType == DataTypes.LONG) {
+        int size = pageSize * DataTypes.LONG.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.LONG.getSizeInBytes()) {
+          vector.putFloat(rowId++,
+              (float) (ByteUtil.toLongLittleEndian(pageData, pageIndex + i) / factor));
+        }
+        pageIndex += size;
+      } else {
+        throw new RuntimeException("internal error: " + this.toString());
+      }
+    } else if (vectorDataType == DataTypes.DOUBLE && factor != 0.0) {
+      if (pageDataType == DataTypes.BOOLEAN || pageDataType == DataTypes.BYTE) {
+        for (int i = 0; i < pageSize; i++) {
+          vector.putDouble(i, (pageData[pageIndex++] / factor));
+        }
+      } else if (pageDataType == DataTypes.SHORT) {
+        int size = pageSize * DataTypes.SHORT.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.SHORT.getSizeInBytes()) {
+          vector
+              .putDouble(rowId++, (ByteUtil.toShortLittleEndian(pageData, pageIndex + i) / factor));
+        }
+        pageIndex += size;
+      } else if (pageDataType == DataTypes.SHORT_INT) {
+        int size = pageSize * DataTypes.SHORT_INT.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.SHORT_INT.getSizeInBytes()) {
+          vector.putDouble(rowId++, (ByteUtil.valueOf3Bytes(pageData, pageIndex + i) / factor));
+        }
+        pageIndex += size;
+      } else if (pageDataType == DataTypes.INT) {
+        int size = pageSize * DataTypes.INT.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.INT.getSizeInBytes()) {
+          vector.putDouble(rowId++, (ByteUtil.toIntLittleEndian(pageData, pageIndex + i) / factor));
+        }
+        pageIndex += size;
+      } else if (pageDataType == DataTypes.LONG) {
+        int size = pageSize * DataTypes.LONG.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.LONG.getSizeInBytes()) {
+          vector
+              .putDouble(rowId++, (ByteUtil.toLongLittleEndian(pageData, pageIndex + i) / factor));
+        }
+        pageIndex += size;
+      } else {
+        throw new RuntimeException("Unsupported datatype : " + pageDataType);
+      }
+    } else if (pageDataType == DataTypes.BYTE_ARRAY) {
+      if (vectorDataType == DataTypes.STRING) {

Review comment:
       DirectCompressCodec & AdaptiveIntegralCodec had code related to decimal,
   I think you have removed it while extracting common code.
   This is impact primitive type functionality also. I guess no testcase present. so, you CI didn't catch.




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
[hidden email]

[GitBox]

ajantha-bhat commented on a change in pull request #3773:
URL: https://github.com/apache/carbondata/pull/3773#discussion_r469858716



##########
File path: core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/FillVector.java
##########
@@ -0,0 +1,347 @@
+/*
+ * 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.encoding;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.BitSet;
+
+import org.apache.carbondata.core.metadata.datatype.DataType;
+import org.apache.carbondata.core.metadata.datatype.DataTypes;
+import org.apache.carbondata.core.metadata.datatype.DecimalConverterFactory;
+import org.apache.carbondata.core.scan.result.vector.CarbonColumnVector;
+import org.apache.carbondata.core.scan.result.vector.ColumnVectorInfo;
+import org.apache.carbondata.core.scan.result.vector.impl.CarbonColumnVectorImpl;
+import org.apache.carbondata.core.util.ByteUtil;
+
+public class FillVector {
+  private byte[] pageData;
+  private float floatFactor = 0;
+  private double factor = 0;
+  private ColumnVectorInfo vectorInfo;
+  private BitSet nullBits;
+
+  public FillVector(byte[] pageData, ColumnVectorInfo vectorInfo, BitSet nullBits) {
+    this.pageData = pageData;
+    this.vectorInfo = vectorInfo;
+    this.nullBits = nullBits;
+  }
+
+  public void setFactor(double factor) {
+    this.factor = factor;
+  }
+
+  public void setFloatFactor(float floatFactor) {
+    this.floatFactor = floatFactor;
+  }
+
+  public void basedOnType(CarbonColumnVector vector, DataType vectorDataType, int pageSize,
+      DataType pageDataType) {
+    if (vectorInfo.vector.getColumnVector() != null && ((CarbonColumnVectorImpl) vectorInfo.vector
+        .getColumnVector()).isComplex()) {
+      fillComplexType(vector.getColumnVector(), pageDataType);
+    } else {
+      fillPrimitiveType(vector, vectorDataType, pageSize, pageDataType);
+      vector.setIndex(0);
+    }
+  }
+
+  private void fillComplexType(CarbonColumnVector vector, DataType pageDataType) {
+    CarbonColumnVectorImpl vectorImpl = (CarbonColumnVectorImpl) vector;
+    if (vector != null && vector.getChildrenVector() != null) {
+      ArrayList<Integer> childElements = ((CarbonColumnVectorImpl) vector).getChildrenElements();
+      for (int i = 0; i < childElements.size(); i++) {
+        int count = childElements.get(i);
+        typeComplexObject(vectorImpl.getChildrenVector().get(0), count, pageDataType);
+        vector.putArrayObject();
+      }
+      vectorImpl.getChildrenVector().get(0).setIndex(0);
+    }
+  }
+
+  private void fillPrimitiveType(CarbonColumnVector vector, DataType vectorDataType, int pageSize,
+      DataType pageDataType) {
+    // offset which denotes the start index for pageData
+    int pageIndex = vector.getIndex();
+    int rowId = 0;
+
+    // Filling into vector is done based on page data type
+
+    if (vectorDataType == DataTypes.FLOAT && floatFactor != 0.0) {
+      if (pageDataType == DataTypes.BOOLEAN || pageDataType == DataTypes.BYTE) {
+        for (int i = 0; i < pageSize; i++) {
+          vector.putFloat(i, (pageData[pageIndex++] / floatFactor));
+        }
+      } else if (pageDataType == DataTypes.SHORT) {
+        int size = pageSize * DataTypes.SHORT.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.SHORT.getSizeInBytes()) {
+          vector.putFloat(rowId++,
+              (ByteUtil.toShortLittleEndian(pageData, pageIndex + i) / floatFactor));
+        }
+        pageIndex += size;
+      } else if (pageDataType == DataTypes.SHORT_INT) {
+        int size = pageSize * DataTypes.SHORT_INT.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.SHORT_INT.getSizeInBytes()) {
+          vector.putFloat(rowId++, (ByteUtil.valueOf3Bytes(pageData, pageIndex + i) / floatFactor));
+        }
+        pageIndex += size;
+      } else if (pageDataType == DataTypes.INT) {
+        int size = pageSize * DataTypes.INT.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.INT.getSizeInBytes()) {
+          vector.putFloat(rowId++,
+              (ByteUtil.toIntLittleEndian(pageData, pageIndex + i) / floatFactor));
+        }
+        pageIndex += size;
+      } else if (pageDataType == DataTypes.LONG) {
+        int size = pageSize * DataTypes.LONG.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.LONG.getSizeInBytes()) {
+          vector.putFloat(rowId++,
+              (float) (ByteUtil.toLongLittleEndian(pageData, pageIndex + i) / factor));
+        }
+        pageIndex += size;
+      } else {
+        throw new RuntimeException("internal error: " + this.toString());
+      }
+    } else if (vectorDataType == DataTypes.DOUBLE && factor != 0.0) {
+      if (pageDataType == DataTypes.BOOLEAN || pageDataType == DataTypes.BYTE) {
+        for (int i = 0; i < pageSize; i++) {
+          vector.putDouble(i, (pageData[pageIndex++] / factor));
+        }
+      } else if (pageDataType == DataTypes.SHORT) {
+        int size = pageSize * DataTypes.SHORT.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.SHORT.getSizeInBytes()) {
+          vector
+              .putDouble(rowId++, (ByteUtil.toShortLittleEndian(pageData, pageIndex + i) / factor));
+        }
+        pageIndex += size;
+      } else if (pageDataType == DataTypes.SHORT_INT) {
+        int size = pageSize * DataTypes.SHORT_INT.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.SHORT_INT.getSizeInBytes()) {
+          vector.putDouble(rowId++, (ByteUtil.valueOf3Bytes(pageData, pageIndex + i) / factor));
+        }
+        pageIndex += size;
+      } else if (pageDataType == DataTypes.INT) {
+        int size = pageSize * DataTypes.INT.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.INT.getSizeInBytes()) {
+          vector.putDouble(rowId++, (ByteUtil.toIntLittleEndian(pageData, pageIndex + i) / factor));
+        }
+        pageIndex += size;
+      } else if (pageDataType == DataTypes.LONG) {
+        int size = pageSize * DataTypes.LONG.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.LONG.getSizeInBytes()) {
+          vector
+              .putDouble(rowId++, (ByteUtil.toLongLittleEndian(pageData, pageIndex + i) / factor));
+        }
+        pageIndex += size;
+      } else {
+        throw new RuntimeException("Unsupported datatype : " + pageDataType);
+      }
+    } else if (pageDataType == DataTypes.BYTE_ARRAY) {
+      if (vectorDataType == DataTypes.STRING) {

Review comment:
       DirectCompressCodec & AdaptiveIntegralCodec had code related to decimal,
   original code look for `DecimalConverterFactory.DecimalConverter`
   I think you have removed it while extracting common code.
   This is impact primitive type functionality also. I guess no testcase present. so, you CI didn't catch.




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
[hidden email]

[GitBox]

ajantha-bhat commented on a change in pull request #3773:
URL: https://github.com/apache/carbondata/pull/3773#discussion_r469858716



##########
File path: core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/FillVector.java
##########
@@ -0,0 +1,347 @@
+/*
+ * 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.encoding;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.BitSet;
+
+import org.apache.carbondata.core.metadata.datatype.DataType;
+import org.apache.carbondata.core.metadata.datatype.DataTypes;
+import org.apache.carbondata.core.metadata.datatype.DecimalConverterFactory;
+import org.apache.carbondata.core.scan.result.vector.CarbonColumnVector;
+import org.apache.carbondata.core.scan.result.vector.ColumnVectorInfo;
+import org.apache.carbondata.core.scan.result.vector.impl.CarbonColumnVectorImpl;
+import org.apache.carbondata.core.util.ByteUtil;
+
+public class FillVector {
+  private byte[] pageData;
+  private float floatFactor = 0;
+  private double factor = 0;
+  private ColumnVectorInfo vectorInfo;
+  private BitSet nullBits;
+
+  public FillVector(byte[] pageData, ColumnVectorInfo vectorInfo, BitSet nullBits) {
+    this.pageData = pageData;
+    this.vectorInfo = vectorInfo;
+    this.nullBits = nullBits;
+  }
+
+  public void setFactor(double factor) {
+    this.factor = factor;
+  }
+
+  public void setFloatFactor(float floatFactor) {
+    this.floatFactor = floatFactor;
+  }
+
+  public void basedOnType(CarbonColumnVector vector, DataType vectorDataType, int pageSize,
+      DataType pageDataType) {
+    if (vectorInfo.vector.getColumnVector() != null && ((CarbonColumnVectorImpl) vectorInfo.vector
+        .getColumnVector()).isComplex()) {
+      fillComplexType(vector.getColumnVector(), pageDataType);
+    } else {
+      fillPrimitiveType(vector, vectorDataType, pageSize, pageDataType);
+      vector.setIndex(0);
+    }
+  }
+
+  private void fillComplexType(CarbonColumnVector vector, DataType pageDataType) {
+    CarbonColumnVectorImpl vectorImpl = (CarbonColumnVectorImpl) vector;
+    if (vector != null && vector.getChildrenVector() != null) {
+      ArrayList<Integer> childElements = ((CarbonColumnVectorImpl) vector).getChildrenElements();
+      for (int i = 0; i < childElements.size(); i++) {
+        int count = childElements.get(i);
+        typeComplexObject(vectorImpl.getChildrenVector().get(0), count, pageDataType);
+        vector.putArrayObject();
+      }
+      vectorImpl.getChildrenVector().get(0).setIndex(0);
+    }
+  }
+
+  private void fillPrimitiveType(CarbonColumnVector vector, DataType vectorDataType, int pageSize,
+      DataType pageDataType) {
+    // offset which denotes the start index for pageData
+    int pageIndex = vector.getIndex();
+    int rowId = 0;
+
+    // Filling into vector is done based on page data type
+
+    if (vectorDataType == DataTypes.FLOAT && floatFactor != 0.0) {
+      if (pageDataType == DataTypes.BOOLEAN || pageDataType == DataTypes.BYTE) {
+        for (int i = 0; i < pageSize; i++) {
+          vector.putFloat(i, (pageData[pageIndex++] / floatFactor));
+        }
+      } else if (pageDataType == DataTypes.SHORT) {
+        int size = pageSize * DataTypes.SHORT.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.SHORT.getSizeInBytes()) {
+          vector.putFloat(rowId++,
+              (ByteUtil.toShortLittleEndian(pageData, pageIndex + i) / floatFactor));
+        }
+        pageIndex += size;
+      } else if (pageDataType == DataTypes.SHORT_INT) {
+        int size = pageSize * DataTypes.SHORT_INT.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.SHORT_INT.getSizeInBytes()) {
+          vector.putFloat(rowId++, (ByteUtil.valueOf3Bytes(pageData, pageIndex + i) / floatFactor));
+        }
+        pageIndex += size;
+      } else if (pageDataType == DataTypes.INT) {
+        int size = pageSize * DataTypes.INT.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.INT.getSizeInBytes()) {
+          vector.putFloat(rowId++,
+              (ByteUtil.toIntLittleEndian(pageData, pageIndex + i) / floatFactor));
+        }
+        pageIndex += size;
+      } else if (pageDataType == DataTypes.LONG) {
+        int size = pageSize * DataTypes.LONG.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.LONG.getSizeInBytes()) {
+          vector.putFloat(rowId++,
+              (float) (ByteUtil.toLongLittleEndian(pageData, pageIndex + i) / factor));
+        }
+        pageIndex += size;
+      } else {
+        throw new RuntimeException("internal error: " + this.toString());
+      }
+    } else if (vectorDataType == DataTypes.DOUBLE && factor != 0.0) {
+      if (pageDataType == DataTypes.BOOLEAN || pageDataType == DataTypes.BYTE) {
+        for (int i = 0; i < pageSize; i++) {
+          vector.putDouble(i, (pageData[pageIndex++] / factor));
+        }
+      } else if (pageDataType == DataTypes.SHORT) {
+        int size = pageSize * DataTypes.SHORT.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.SHORT.getSizeInBytes()) {
+          vector
+              .putDouble(rowId++, (ByteUtil.toShortLittleEndian(pageData, pageIndex + i) / factor));
+        }
+        pageIndex += size;
+      } else if (pageDataType == DataTypes.SHORT_INT) {
+        int size = pageSize * DataTypes.SHORT_INT.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.SHORT_INT.getSizeInBytes()) {
+          vector.putDouble(rowId++, (ByteUtil.valueOf3Bytes(pageData, pageIndex + i) / factor));
+        }
+        pageIndex += size;
+      } else if (pageDataType == DataTypes.INT) {
+        int size = pageSize * DataTypes.INT.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.INT.getSizeInBytes()) {
+          vector.putDouble(rowId++, (ByteUtil.toIntLittleEndian(pageData, pageIndex + i) / factor));
+        }
+        pageIndex += size;
+      } else if (pageDataType == DataTypes.LONG) {
+        int size = pageSize * DataTypes.LONG.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.LONG.getSizeInBytes()) {
+          vector
+              .putDouble(rowId++, (ByteUtil.toLongLittleEndian(pageData, pageIndex + i) / factor));
+        }
+        pageIndex += size;
+      } else {
+        throw new RuntimeException("Unsupported datatype : " + pageDataType);
+      }
+    } else if (pageDataType == DataTypes.BYTE_ARRAY) {
+      if (vectorDataType == DataTypes.STRING) {

Review comment:
       DirectCompressCodec & AdaptiveIntegralCodec had code related to decimal,
   original code look for `DecimalConverterFactory.DecimalConverter`
   I think you have removed it while extracting common code.
   This is impact primitive type functionality also. I guess no testcase present. so, CI didn't catch.




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
[hidden email]

[GitBox]

ajantha-bhat commented on a change in pull request #3773:
URL: https://github.com/apache/carbondata/pull/3773#discussion_r469858716



##########
File path: core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/FillVector.java
##########
@@ -0,0 +1,347 @@
+/*
+ * 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.encoding;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.BitSet;
+
+import org.apache.carbondata.core.metadata.datatype.DataType;
+import org.apache.carbondata.core.metadata.datatype.DataTypes;
+import org.apache.carbondata.core.metadata.datatype.DecimalConverterFactory;
+import org.apache.carbondata.core.scan.result.vector.CarbonColumnVector;
+import org.apache.carbondata.core.scan.result.vector.ColumnVectorInfo;
+import org.apache.carbondata.core.scan.result.vector.impl.CarbonColumnVectorImpl;
+import org.apache.carbondata.core.util.ByteUtil;
+
+public class FillVector {
+  private byte[] pageData;
+  private float floatFactor = 0;
+  private double factor = 0;
+  private ColumnVectorInfo vectorInfo;
+  private BitSet nullBits;
+
+  public FillVector(byte[] pageData, ColumnVectorInfo vectorInfo, BitSet nullBits) {
+    this.pageData = pageData;
+    this.vectorInfo = vectorInfo;
+    this.nullBits = nullBits;
+  }
+
+  public void setFactor(double factor) {
+    this.factor = factor;
+  }
+
+  public void setFloatFactor(float floatFactor) {
+    this.floatFactor = floatFactor;
+  }
+
+  public void basedOnType(CarbonColumnVector vector, DataType vectorDataType, int pageSize,
+      DataType pageDataType) {
+    if (vectorInfo.vector.getColumnVector() != null && ((CarbonColumnVectorImpl) vectorInfo.vector
+        .getColumnVector()).isComplex()) {
+      fillComplexType(vector.getColumnVector(), pageDataType);
+    } else {
+      fillPrimitiveType(vector, vectorDataType, pageSize, pageDataType);
+      vector.setIndex(0);
+    }
+  }
+
+  private void fillComplexType(CarbonColumnVector vector, DataType pageDataType) {
+    CarbonColumnVectorImpl vectorImpl = (CarbonColumnVectorImpl) vector;
+    if (vector != null && vector.getChildrenVector() != null) {
+      ArrayList<Integer> childElements = ((CarbonColumnVectorImpl) vector).getChildrenElements();
+      for (int i = 0; i < childElements.size(); i++) {
+        int count = childElements.get(i);
+        typeComplexObject(vectorImpl.getChildrenVector().get(0), count, pageDataType);
+        vector.putArrayObject();
+      }
+      vectorImpl.getChildrenVector().get(0).setIndex(0);
+    }
+  }
+
+  private void fillPrimitiveType(CarbonColumnVector vector, DataType vectorDataType, int pageSize,
+      DataType pageDataType) {
+    // offset which denotes the start index for pageData
+    int pageIndex = vector.getIndex();
+    int rowId = 0;
+
+    // Filling into vector is done based on page data type
+
+    if (vectorDataType == DataTypes.FLOAT && floatFactor != 0.0) {
+      if (pageDataType == DataTypes.BOOLEAN || pageDataType == DataTypes.BYTE) {
+        for (int i = 0; i < pageSize; i++) {
+          vector.putFloat(i, (pageData[pageIndex++] / floatFactor));
+        }
+      } else if (pageDataType == DataTypes.SHORT) {
+        int size = pageSize * DataTypes.SHORT.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.SHORT.getSizeInBytes()) {
+          vector.putFloat(rowId++,
+              (ByteUtil.toShortLittleEndian(pageData, pageIndex + i) / floatFactor));
+        }
+        pageIndex += size;
+      } else if (pageDataType == DataTypes.SHORT_INT) {
+        int size = pageSize * DataTypes.SHORT_INT.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.SHORT_INT.getSizeInBytes()) {
+          vector.putFloat(rowId++, (ByteUtil.valueOf3Bytes(pageData, pageIndex + i) / floatFactor));
+        }
+        pageIndex += size;
+      } else if (pageDataType == DataTypes.INT) {
+        int size = pageSize * DataTypes.INT.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.INT.getSizeInBytes()) {
+          vector.putFloat(rowId++,
+              (ByteUtil.toIntLittleEndian(pageData, pageIndex + i) / floatFactor));
+        }
+        pageIndex += size;
+      } else if (pageDataType == DataTypes.LONG) {
+        int size = pageSize * DataTypes.LONG.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.LONG.getSizeInBytes()) {
+          vector.putFloat(rowId++,
+              (float) (ByteUtil.toLongLittleEndian(pageData, pageIndex + i) / factor));
+        }
+        pageIndex += size;
+      } else {
+        throw new RuntimeException("internal error: " + this.toString());
+      }
+    } else if (vectorDataType == DataTypes.DOUBLE && factor != 0.0) {
+      if (pageDataType == DataTypes.BOOLEAN || pageDataType == DataTypes.BYTE) {
+        for (int i = 0; i < pageSize; i++) {
+          vector.putDouble(i, (pageData[pageIndex++] / factor));
+        }
+      } else if (pageDataType == DataTypes.SHORT) {
+        int size = pageSize * DataTypes.SHORT.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.SHORT.getSizeInBytes()) {
+          vector
+              .putDouble(rowId++, (ByteUtil.toShortLittleEndian(pageData, pageIndex + i) / factor));
+        }
+        pageIndex += size;
+      } else if (pageDataType == DataTypes.SHORT_INT) {
+        int size = pageSize * DataTypes.SHORT_INT.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.SHORT_INT.getSizeInBytes()) {
+          vector.putDouble(rowId++, (ByteUtil.valueOf3Bytes(pageData, pageIndex + i) / factor));
+        }
+        pageIndex += size;
+      } else if (pageDataType == DataTypes.INT) {
+        int size = pageSize * DataTypes.INT.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.INT.getSizeInBytes()) {
+          vector.putDouble(rowId++, (ByteUtil.toIntLittleEndian(pageData, pageIndex + i) / factor));
+        }
+        pageIndex += size;
+      } else if (pageDataType == DataTypes.LONG) {
+        int size = pageSize * DataTypes.LONG.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.LONG.getSizeInBytes()) {
+          vector
+              .putDouble(rowId++, (ByteUtil.toLongLittleEndian(pageData, pageIndex + i) / factor));
+        }
+        pageIndex += size;
+      } else {
+        throw new RuntimeException("Unsupported datatype : " + pageDataType);
+      }
+    } else if (pageDataType == DataTypes.BYTE_ARRAY) {
+      if (vectorDataType == DataTypes.STRING) {

Review comment:
       DirectCompressCodec & AdaptiveIntegralCodec had code related to decimal,
   original code look for `DecimalConverterFactory.DecimalConverter`
   I think you have removed it while extracting common code.
   This will impact primitive type functionality also. I guess no testcase present. so, CI didn't catch.




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
[hidden email]

[GitBox]

ajantha-bhat commented on a change in pull request #3773:
URL: https://github.com/apache/carbondata/pull/3773#discussion_r469863077



##########
File path: core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/FillVector.java
##########
@@ -0,0 +1,347 @@
+/*
+ * 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.encoding;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.BitSet;
+
+import org.apache.carbondata.core.metadata.datatype.DataType;
+import org.apache.carbondata.core.metadata.datatype.DataTypes;
+import org.apache.carbondata.core.metadata.datatype.DecimalConverterFactory;
+import org.apache.carbondata.core.scan.result.vector.CarbonColumnVector;
+import org.apache.carbondata.core.scan.result.vector.ColumnVectorInfo;
+import org.apache.carbondata.core.scan.result.vector.impl.CarbonColumnVectorImpl;
+import org.apache.carbondata.core.util.ByteUtil;
+
+public class FillVector {
+  private byte[] pageData;
+  private float floatFactor = 0;
+  private double factor = 0;
+  private ColumnVectorInfo vectorInfo;
+  private BitSet nullBits;
+
+  public FillVector(byte[] pageData, ColumnVectorInfo vectorInfo, BitSet nullBits) {
+    this.pageData = pageData;
+    this.vectorInfo = vectorInfo;
+    this.nullBits = nullBits;
+  }
+
+  public void setFactor(double factor) {
+    this.factor = factor;
+  }
+
+  public void setFloatFactor(float floatFactor) {
+    this.floatFactor = floatFactor;
+  }
+
+  public void basedOnType(CarbonColumnVector vector, DataType vectorDataType, int pageSize,
+      DataType pageDataType) {
+    if (vectorInfo.vector.getColumnVector() != null && ((CarbonColumnVectorImpl) vectorInfo.vector
+        .getColumnVector()).isComplex()) {
+      fillComplexType(vector.getColumnVector(), pageDataType);
+    } else {
+      fillPrimitiveType(vector, vectorDataType, pageSize, pageDataType);
+      vector.setIndex(0);
+    }
+  }
+
+  private void fillComplexType(CarbonColumnVector vector, DataType pageDataType) {
+    CarbonColumnVectorImpl vectorImpl = (CarbonColumnVectorImpl) vector;
+    if (vector != null && vector.getChildrenVector() != null) {
+      ArrayList<Integer> childElements = ((CarbonColumnVectorImpl) vector).getChildrenElements();
+      for (int i = 0; i < childElements.size(); i++) {
+        int count = childElements.get(i);
+        typeComplexObject(vectorImpl.getChildrenVector().get(0), count, pageDataType);
+        vector.putArrayObject();
+      }
+      vectorImpl.getChildrenVector().get(0).setIndex(0);
+    }
+  }
+
+  private void fillPrimitiveType(CarbonColumnVector vector, DataType vectorDataType, int pageSize,
+      DataType pageDataType) {
+    // offset which denotes the start index for pageData
+    int pageIndex = vector.getIndex();
+    int rowId = 0;
+
+    // Filling into vector is done based on page data type
+
+    if (vectorDataType == DataTypes.FLOAT && floatFactor != 0.0) {
+      if (pageDataType == DataTypes.BOOLEAN || pageDataType == DataTypes.BYTE) {
+        for (int i = 0; i < pageSize; i++) {
+          vector.putFloat(i, (pageData[pageIndex++] / floatFactor));
+        }
+      } else if (pageDataType == DataTypes.SHORT) {
+        int size = pageSize * DataTypes.SHORT.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.SHORT.getSizeInBytes()) {
+          vector.putFloat(rowId++,
+              (ByteUtil.toShortLittleEndian(pageData, pageIndex + i) / floatFactor));
+        }
+        pageIndex += size;
+      } else if (pageDataType == DataTypes.SHORT_INT) {
+        int size = pageSize * DataTypes.SHORT_INT.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.SHORT_INT.getSizeInBytes()) {
+          vector.putFloat(rowId++, (ByteUtil.valueOf3Bytes(pageData, pageIndex + i) / floatFactor));
+        }
+        pageIndex += size;
+      } else if (pageDataType == DataTypes.INT) {
+        int size = pageSize * DataTypes.INT.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.INT.getSizeInBytes()) {
+          vector.putFloat(rowId++,
+              (ByteUtil.toIntLittleEndian(pageData, pageIndex + i) / floatFactor));
+        }
+        pageIndex += size;
+      } else if (pageDataType == DataTypes.LONG) {
+        int size = pageSize * DataTypes.LONG.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.LONG.getSizeInBytes()) {
+          vector.putFloat(rowId++,
+              (float) (ByteUtil.toLongLittleEndian(pageData, pageIndex + i) / factor));
+        }
+        pageIndex += size;
+      } else {
+        throw new RuntimeException("internal error: " + this.toString());
+      }
+    } else if (vectorDataType == DataTypes.DOUBLE && factor != 0.0) {
+      if (pageDataType == DataTypes.BOOLEAN || pageDataType == DataTypes.BYTE) {
+        for (int i = 0; i < pageSize; i++) {
+          vector.putDouble(i, (pageData[pageIndex++] / factor));
+        }
+      } else if (pageDataType == DataTypes.SHORT) {
+        int size = pageSize * DataTypes.SHORT.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.SHORT.getSizeInBytes()) {
+          vector
+              .putDouble(rowId++, (ByteUtil.toShortLittleEndian(pageData, pageIndex + i) / factor));
+        }
+        pageIndex += size;
+      } else if (pageDataType == DataTypes.SHORT_INT) {
+        int size = pageSize * DataTypes.SHORT_INT.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.SHORT_INT.getSizeInBytes()) {
+          vector.putDouble(rowId++, (ByteUtil.valueOf3Bytes(pageData, pageIndex + i) / factor));
+        }
+        pageIndex += size;
+      } else if (pageDataType == DataTypes.INT) {
+        int size = pageSize * DataTypes.INT.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.INT.getSizeInBytes()) {
+          vector.putDouble(rowId++, (ByteUtil.toIntLittleEndian(pageData, pageIndex + i) / factor));
+        }
+        pageIndex += size;
+      } else if (pageDataType == DataTypes.LONG) {
+        int size = pageSize * DataTypes.LONG.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.LONG.getSizeInBytes()) {
+          vector
+              .putDouble(rowId++, (ByteUtil.toLongLittleEndian(pageData, pageIndex + i) / factor));
+        }
+        pageIndex += size;
+      } else {
+        throw new RuntimeException("Unsupported datatype : " + pageDataType);
+      }
+    } else if (pageDataType == DataTypes.BYTE_ARRAY) {
+      if (vectorDataType == DataTypes.STRING) {

Review comment:
       Also if more than one complex type children has decimal, this vector info has to be redesigned. it cannot contain all decimal column vectors now




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
[hidden email]

[GitBox]

ajantha-bhat commented on pull request #3773:
URL: https://github.com/apache/carbondata/pull/3773#issuecomment-674940989


   @akkio-97 : Thanks for working on this.
   
   Based on the above comments (array design is not vectorized, fill vector need to be dissolved and keep it as original, interface need to have a default implementation, all the data types not handled, delta flows were not handled) **it will be very hard to stabilize this based on the current design**.
   **I have analyzed and reworked on the new design in #3887,** **I will add you as co-Author to the same.**
   **you can close this PR** and later raise PR for local dict support, multi level array & struct support and map support.


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
[hidden email]

[GitBox]

ajantha-bhat edited a comment on pull request #3773:
URL: https://github.com/apache/carbondata/pull/3773#issuecomment-674940989


   @akkio-97 : Thanks for working on this.
   
   Based on the above comments (array design is not vectorized, fill vector need to be dissolved and keep it as original, interface need to have a default implementation, all the data types not handled, delta flows were not handled, null values not handled), **it will be very hard to stabilize this based on the current design**.
   **I have analyzed and reworked on the new design in #3887,** **I will add you as co-Author to the same.**
   **you can close this PR** and later raise PR for local dict support, multi level array & struct support and map support.


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
[hidden email]

[GitBox]

akkio-97 commented on a change in pull request #3773:
URL: https://github.com/apache/carbondata/pull/3773#discussion_r472038054



##########
File path: 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 java.util.Stack;

Review comment:
       okay

##########
File path: core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/FillVector.java
##########
@@ -0,0 +1,347 @@
+/*
+ * 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.encoding;

Review comment:
       okay

##########
File path: core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/FillVector.java
##########
@@ -0,0 +1,347 @@
+/*
+ * 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.encoding;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.BitSet;
+
+import org.apache.carbondata.core.metadata.datatype.DataType;
+import org.apache.carbondata.core.metadata.datatype.DataTypes;
+import org.apache.carbondata.core.metadata.datatype.DecimalConverterFactory;
+import org.apache.carbondata.core.scan.result.vector.CarbonColumnVector;
+import org.apache.carbondata.core.scan.result.vector.ColumnVectorInfo;
+import org.apache.carbondata.core.scan.result.vector.impl.CarbonColumnVectorImpl;
+import org.apache.carbondata.core.util.ByteUtil;
+
+public class FillVector {
+  private byte[] pageData;
+  private float floatFactor = 0;
+  private double factor = 0;
+  private ColumnVectorInfo vectorInfo;
+  private BitSet nullBits;
+
+  public FillVector(byte[] pageData, ColumnVectorInfo vectorInfo, BitSet nullBits) {
+    this.pageData = pageData;
+    this.vectorInfo = vectorInfo;
+    this.nullBits = nullBits;
+  }
+
+  public void setFactor(double factor) {
+    this.factor = factor;
+  }
+
+  public void setFloatFactor(float floatFactor) {
+    this.floatFactor = floatFactor;
+  }
+
+  public void basedOnType(CarbonColumnVector vector, DataType vectorDataType, int pageSize,
+      DataType pageDataType) {
+    if (vectorInfo.vector.getColumnVector() != null && ((CarbonColumnVectorImpl) vectorInfo.vector
+        .getColumnVector()).isComplex()) {
+      fillComplexType(vector.getColumnVector(), pageDataType);
+    } else {
+      fillPrimitiveType(vector, vectorDataType, pageSize, pageDataType);
+      vector.setIndex(0);
+    }
+  }
+
+  private void fillComplexType(CarbonColumnVector vector, DataType pageDataType) {
+    CarbonColumnVectorImpl vectorImpl = (CarbonColumnVectorImpl) vector;
+    if (vector != null && vector.getChildrenVector() != null) {
+      ArrayList<Integer> childElements = ((CarbonColumnVectorImpl) vector).getChildrenElements();
+      for (int i = 0; i < childElements.size(); i++) {
+        int count = childElements.get(i);
+        typeComplexObject(vectorImpl.getChildrenVector().get(0), count, pageDataType);
+        vector.putArrayObject();
+      }
+      vectorImpl.getChildrenVector().get(0).setIndex(0);
+    }
+  }
+
+  private void fillPrimitiveType(CarbonColumnVector vector, DataType vectorDataType, int pageSize,
+      DataType pageDataType) {
+    // offset which denotes the start index for pageData
+    int pageIndex = vector.getIndex();
+    int rowId = 0;
+
+    // Filling into vector is done based on page data type
+
+    if (vectorDataType == DataTypes.FLOAT && floatFactor != 0.0) {
+      if (pageDataType == DataTypes.BOOLEAN || pageDataType == DataTypes.BYTE) {
+        for (int i = 0; i < pageSize; i++) {
+          vector.putFloat(i, (pageData[pageIndex++] / floatFactor));
+        }
+      } else if (pageDataType == DataTypes.SHORT) {
+        int size = pageSize * DataTypes.SHORT.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.SHORT.getSizeInBytes()) {
+          vector.putFloat(rowId++,
+              (ByteUtil.toShortLittleEndian(pageData, pageIndex + i) / floatFactor));
+        }
+        pageIndex += size;
+      } else if (pageDataType == DataTypes.SHORT_INT) {
+        int size = pageSize * DataTypes.SHORT_INT.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.SHORT_INT.getSizeInBytes()) {
+          vector.putFloat(rowId++, (ByteUtil.valueOf3Bytes(pageData, pageIndex + i) / floatFactor));
+        }
+        pageIndex += size;
+      } else if (pageDataType == DataTypes.INT) {
+        int size = pageSize * DataTypes.INT.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.INT.getSizeInBytes()) {
+          vector.putFloat(rowId++,
+              (ByteUtil.toIntLittleEndian(pageData, pageIndex + i) / floatFactor));
+        }
+        pageIndex += size;
+      } else if (pageDataType == DataTypes.LONG) {
+        int size = pageSize * DataTypes.LONG.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.LONG.getSizeInBytes()) {
+          vector.putFloat(rowId++,
+              (float) (ByteUtil.toLongLittleEndian(pageData, pageIndex + i) / factor));
+        }
+        pageIndex += size;
+      } else {
+        throw new RuntimeException("internal error: " + this.toString());
+      }
+    } else if (vectorDataType == DataTypes.DOUBLE && factor != 0.0) {
+      if (pageDataType == DataTypes.BOOLEAN || pageDataType == DataTypes.BYTE) {
+        for (int i = 0; i < pageSize; i++) {
+          vector.putDouble(i, (pageData[pageIndex++] / factor));
+        }
+      } else if (pageDataType == DataTypes.SHORT) {
+        int size = pageSize * DataTypes.SHORT.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.SHORT.getSizeInBytes()) {
+          vector
+              .putDouble(rowId++, (ByteUtil.toShortLittleEndian(pageData, pageIndex + i) / factor));
+        }
+        pageIndex += size;
+      } else if (pageDataType == DataTypes.SHORT_INT) {
+        int size = pageSize * DataTypes.SHORT_INT.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.SHORT_INT.getSizeInBytes()) {
+          vector.putDouble(rowId++, (ByteUtil.valueOf3Bytes(pageData, pageIndex + i) / factor));
+        }
+        pageIndex += size;
+      } else if (pageDataType == DataTypes.INT) {
+        int size = pageSize * DataTypes.INT.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.INT.getSizeInBytes()) {
+          vector.putDouble(rowId++, (ByteUtil.toIntLittleEndian(pageData, pageIndex + i) / factor));
+        }
+        pageIndex += size;
+      } else if (pageDataType == DataTypes.LONG) {
+        int size = pageSize * DataTypes.LONG.getSizeInBytes();
+        for (int i = 0; i < size; i += DataTypes.LONG.getSizeInBytes()) {
+          vector
+              .putDouble(rowId++, (ByteUtil.toLongLittleEndian(pageData, pageIndex + i) / factor));
+        }
+        pageIndex += size;
+      } else {
+        throw new RuntimeException("Unsupported datatype : " + pageDataType);
+      }
+    } else if (pageDataType == DataTypes.BYTE_ARRAY) {
+      if (vectorDataType == DataTypes.STRING) {

Review comment:
       okay




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
[hidden email]

[GitBox]

akkio-97 commented on pull request #3773:
URL: https://github.com/apache/carbondata/pull/3773#issuecomment-675365410


   okay, thanks for all your suggestions.


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
[hidden email]

[GitBox]

akkio-97 closed pull request #3773:
URL: https://github.com/apache/carbondata/pull/3773


   


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
[hidden email]