[CARBONDATA-4291] Carbon hive table supports float datatype

Author: jack86596

core/src/main/java/org/apache/carbondata/core/datastore/page/ColumnPage.java

@@ -445,6 +445,8 @@ public Object getData(int rowId) {
       return getInt(rowId);
     } else if (dataType == LONG) {
       return getLong(rowId);
+    } else if (dataType == DataTypes.FLOAT) {
+      return getFloat(rowId);
     } else if (dataType == DataTypes.DOUBLE) {
       return getDouble(rowId);
     } else if (DataTypes.isDecimal(dataType)) {
@@ -560,6 +562,8 @@ private Object getNull(int rowId) {
       result = getInt(rowId);
     } else if (dataType == LONG) {
       result = getLong(rowId);
+    } else if (dataType == DataTypes.FLOAT) {
+      return getFloat(rowId);
     } else if (dataType == DataTypes.DOUBLE) {
       result = getDouble(rowId);
     } else if (DataTypes.isDecimal(dataType)) {

core/src/main/java/org/apache/carbondata/core/datastore/page/ColumnPageValueConverter.java

@@ -43,6 +43,12 @@ public interface ColumnPageValueConverter {
 
   long decodeLong(int value);
 
+  float decodeFloat(byte value);
+
+  float decodeFloat(short value);
+
+  float decodeFloat(int value);
+
   double decodeDouble(byte value);
 
   double decodeDouble(short value);

core/src/main/java/org/apache/carbondata/core/datastore/page/LazyColumnPage.java

@@ -94,7 +94,22 @@ public double getDouble(int rowId) {
 
   @Override
   public float getFloat(int rowId) {
-    return (float) getDouble(rowId);
+    DataType dataType = columnPage.getDataType();
+    if (dataType == DataTypes.BOOLEAN || dataType == DataTypes.BYTE) {
+      return converter.decodeFloat(columnPage.getByte(rowId));
+    } else if (dataType == DataTypes.SHORT) {
+      return converter.decodeFloat(columnPage.getShort(rowId));
+    } else if (dataType == DataTypes.SHORT_INT) {
+      return converter.decodeFloat(columnPage.getShortInt(rowId));
+    } else if (dataType == DataTypes.INT) {
+      return converter.decodeFloat(columnPage.getInt(rowId));
+    } else if (dataType == DataTypes.FLOAT) {
+      return columnPage.getFloat(rowId);
+    } else if (dataType == DataTypes.BINARY) {
+      return converter.decodeFloat(columnPage.getByte(rowId));
+    } else {
+      throw new RuntimeException("internal error: " + this);
+    }
   }
 
   @Override

core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/ColumnPageEncoderMeta.java

@@ -247,6 +247,11 @@ private byte[] getValueAsBytes(Object value, DataType dataType) {
       b.putLong((long) value);
       b.flip();
       return b.array();
+    } else if (dataType == DataTypes.FLOAT) {
+      b = ByteBuffer.allocate(8);
+      b.putFloat((float) value);
+      b.flip();
+      return b.array();
     } else if (dataType == DataTypes.DOUBLE) {
       b = ByteBuffer.allocate(8);
       b.putDouble((double) value);

core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/DefaultEncodingFactory.java

@@ -166,6 +166,8 @@ private static DataType fitMinMax(DataType dataType, Object max, Object min) {
       return fitLongMinMax((int) max, (int) min);
     } else if ((dataType == DataTypes.LONG) || (dataType == DataTypes.TIMESTAMP)) {
       return fitLongMinMax((long) max, (long) min);
+    } else if (dataType == DataTypes.FLOAT) {
+      return fitLongMinMax((long) (float) max, (long) (float) min);
     } else if (dataType == DataTypes.DOUBLE) {
       return fitLongMinMax((long) (double) max, (long) (double) min);
     } else {
@@ -212,6 +214,8 @@ private static DataType fitDelta(DataType dataType, Object max, Object min) {
       value = (long) (short) max - (long) (short) min;
     } else if (dataType == DataTypes.INT) {
       value = (long) (int) max - (long) (int) min;
+    } else if (dataType == DataTypes.FLOAT) {
+      value = (long) (float) max - (long) (float) min;
     } else if (dataType == DataTypes.LONG || dataType == DataTypes.TIMESTAMP) {
       value = (long) max - (long) min;
       // The subtraction overflowed iff the operands have opposing signs
@@ -329,6 +333,9 @@ private static ColumnPageCodec getColumnPageCodec(SimpleStatsResult stats,
     // If absMaxValue exceeds LONG.MAX_VALUE, then go for direct compression
     if ((Math.pow(10, decimalCount) * absMaxValue) > Long.MAX_VALUE) {
       return new DirectCompressCodec(DataTypes.DOUBLE);
+    } else if (srcDataType == DataTypes.FLOAT &&
+        (Math.pow(10, decimalCount) * absMaxValue) > Integer.MAX_VALUE) {
+      return new DirectCompressCodec(DataTypes.FLOAT);
     } else {
       long max = (long) (Math.pow(10, decimalCount) * absMaxValue);
       DataType adaptiveDataType = fitLongMinMax(max, 0);

core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/adaptive/AdaptiveDeltaFloatingCodec.java

@@ -53,6 +53,7 @@
 public class AdaptiveDeltaFloatingCodec extends AdaptiveCodec {
 
   private Double factor;
+  private float floatFactor;
   private long max;
 
   public static ColumnPageCodec newInstance(DataType srcDataType, DataType targetDataType,
@@ -65,6 +66,7 @@ public AdaptiveDeltaFloatingCodec(DataType srcDataType, DataType targetDataType,
       SimpleStatsResult stats, boolean isInvertedIndex) {
     super(srcDataType, targetDataType, stats, isInvertedIndex);
     this.factor = Math.pow(10, stats.getDecimalCount());
+    floatFactor = factor.floatValue();
     if (srcDataType == DataTypes.FLOAT) {
       this.max =
           (long) ((long) Math.pow(10, stats.getDecimalCount()) * ((float) stats.getMax()));
@@ -185,13 +187,15 @@ public void encode(int rowId, long value) {
     @Override
     public void encode(int rowId, float value) {
       if (targetDataType.equals(DataTypes.BYTE)) {
-        encodedPage.putByte(rowId, (byte) (max - (value * factor)));
+        encodedPage.putByte(rowId, (byte) (max - (value * floatFactor)));
       } else if (targetDataType.equals(DataTypes.SHORT)) {
-        encodedPage.putShort(rowId, (short) (max - (value * factor)));
+        encodedPage.putShort(rowId, (short) (max - (value * floatFactor)));
       } else if (targetDataType.equals(DataTypes.SHORT_INT)) {
-        encodedPage.putShortInt(rowId, (int) (max - (value * factor)));
+        encodedPage.putShortInt(rowId, (int) (max - (value * floatFactor)));
       } else if (targetDataType.equals(DataTypes.INT)) {
-        encodedPage.putInt(rowId, (int) (max - (value * factor)));
+        encodedPage.putInt(rowId, (int) (max - (value * floatFactor)));
+      } else if (targetDataType.equals(DataTypes.FLOAT)) {
+        encodedPage.putFloat(rowId, value);
       } else {
         throw new RuntimeException("internal error: " + debugInfo());
       }
@@ -229,6 +233,21 @@ public long decodeLong(int value) {
       throw new RuntimeException("internal error: " + debugInfo());
     }
 
+    @Override
+    public float decodeFloat(byte value) {
+      return (max - value) / floatFactor;
+    }
+
+    @Override
+    public float decodeFloat(short value) {
+      return (max - value) / floatFactor;
+    }
+
+    @Override
+    public float decodeFloat(int value) {
+      return (max - value) / floatFactor;
+    }
+
     @Override
     public double decodeDouble(byte value) {
       return (max - value) / factor;
@@ -265,7 +284,6 @@ public void decodeAndFillVector(byte[] pageData, ColumnVectorInfo vectorInfo, Bi
       int intSizeInBytes = DataTypes.INT.getSizeInBytes();
       int longSizeInBytes = DataTypes.LONG.getSizeInBytes();
       if (vectorDataType == DataTypes.FLOAT) {
-        float floatFactor = factor.floatValue();
         if (pageDataType == DataTypes.BOOLEAN || pageDataType == DataTypes.BYTE) {
           for (int i = 0; i < pageSize; i++) {
             vector.putFloat(i, (max - pageData[i]) / floatFactor);

core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/adaptive/AdaptiveDeltaIntegralCodec.java

@@ -69,6 +69,8 @@ public AdaptiveDeltaIntegralCodec(DataType srcDataType, DataType targetDataType,
       this.max = (int) stats.getMax();
     } else if (srcDataType == DataTypes.LONG || srcDataType == DataTypes.TIMESTAMP) {
       this.max = (long) stats.getMax();
+    } else if (srcDataType == DataTypes.FLOAT) {
+      this.max = (int) (float) stats.getMax();
     } else if (srcDataType == DataTypes.DOUBLE) {
       this.max = (long) (double) stats.getMax();
     } else if (DataTypes.isDecimal(srcDataType)) {
@@ -236,8 +238,6 @@ public void encode(int rowId, float value) {
         encodedPage.putShortInt(rowId, (int) (max - value));
       } else if (targetDataType == DataTypes.INT) {
         encodedPage.putInt(rowId, (int) (max - value));
-      } else if (targetDataType == DataTypes.LONG) {
-        encodedPage.putLong(rowId, (long) (max - value));
       } else {
         throw new RuntimeException("internal error");
       }
@@ -275,6 +275,21 @@ public long decodeLong(int value) {
       return max - value;
     }
 
+    @Override
+    public float decodeFloat(byte value) {
+      return max - value;
+    }
+
+    @Override
+    public float decodeFloat(short value) {
+      return max - value;
+    }
+
+    @Override
+    public float decodeFloat(int value) {
+      return max - value;
+    }
+
     @Override
     public double decodeDouble(byte value) {
       return max - value;

core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/adaptive/AdaptiveFloatingCodec.java

@@ -172,7 +172,19 @@ public void encode(int rowId, long value) {
 
     @Override
     public void encode(int rowId, float value) {
-      encode(rowId, (double) value);
+      if (targetDataType == DataTypes.BYTE) {
+        encodedPage.putByte(rowId, (byte) Math.round(value * floatFactor));
+      } else if (targetDataType == DataTypes.SHORT) {
+        encodedPage.putShort(rowId, (short) Math.round(value * floatFactor));
+      } else if (targetDataType == DataTypes.SHORT_INT) {
+        encodedPage.putShortInt(rowId, Math.round(value * floatFactor));
+      } else if (targetDataType == DataTypes.INT) {
+        encodedPage.putInt(rowId, Math.round(value * floatFactor));
+      } else if (targetDataType == DataTypes.FLOAT) {
+        encodedPage.putFloat(rowId, value);
+      } else {
+        throw new RuntimeException("internal error: " + debugInfo());
+      }
     }
 
     @Override
@@ -209,6 +221,21 @@ public long decodeLong(int value) {
       throw new RuntimeException("internal error: " + debugInfo());
     }
 
+    @Override
+    public float decodeFloat(byte value) {
+      return value / floatFactor;
+    }
+
+    @Override
+    public float decodeFloat(short value) {
+      return value / floatFactor;
+    }
+
+    @Override
+    public float decodeFloat(int value) {
+      return value / floatFactor;
+    }
+
     @Override
     public double decodeDouble(byte value) {
       return value / factor;

core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/adaptive/AdaptiveIntegralCodec.java

@@ -251,6 +251,21 @@ public long decodeLong(int value) {
       return value;
     }
 
+    @Override
+    public float decodeFloat(byte value) {
+      return value;
+    }
+
+    @Override
+    public float decodeFloat(short value) {
+      return value;
+    }
+
+    @Override
+    public float decodeFloat(int value) {
+      return value;
+    }
+
     @Override
     public double decodeDouble(byte value) {
       return value;

core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/compress/DirectCompressCodec.java

@@ -215,6 +215,21 @@ public long decodeLong(int value) {
       return value;
     }
 
+    @Override
+    public float decodeFloat(byte value) {
+      return value;
+    }
+
+    @Override
+    public float decodeFloat(short value) {
+      return value;
+    }
+
+    @Override
+    public float decodeFloat(int value) {
+      return value;
+    }
+
     @Override
     public double decodeDouble(byte value) {
       return value;