From c26897b2d8d37ba4e5361da7585dfe6942c42f5a Mon Sep 17 00:00:00 2001
From: Indhumathi27 <indhumathim27@gmail.com>
Date: Fri, 27 Jul 2018 12:22:25 +0530
Subject: [PATCH] [CARBONDATA-2791]Fix Adaptive Encoding for Double if exceeds
 LONG.Max_value

---
 .../page/encoding/DefaultEncodingFactory.java | 27 +++++++++++--------
 .../complexType/TestAdaptiveComplexType.scala | 17 ++++++++++++
 2 files changed, 33 insertions(+), 11 deletions(-)

diff --git a/core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/DefaultEncodingFactory.java b/core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/DefaultEncodingFactory.java
index fa8bfad99ad..1cc2ba84739 100644
--- a/core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/DefaultEncodingFactory.java
+++ b/core/src/main/java/org/apache/carbondata/core/datastore/page/encoding/DefaultEncodingFactory.java
@@ -300,18 +300,23 @@ static ColumnPageCodec selectCodecByAlgorithmForFloating(SimpleStatsResult stats
       return new DirectCompressCodec(DataTypes.DOUBLE);
     } else {
       // double
-      long max = (long) (Math.pow(10, decimalCount) * absMaxValue);
-      DataType adaptiveDataType = fitLongMinMax(max, 0);
-      DataType deltaDataType = compareMinMaxAndSelectDataType(
-          (long) (Math.pow(10, decimalCount) * (maxValue - minValue)));
-      if (adaptiveDataType.getSizeInBytes() > deltaDataType.getSizeInBytes()) {
-        return new AdaptiveDeltaFloatingCodec(srcDataType, deltaDataType, stats);
-      } else if (adaptiveDataType.getSizeInBytes() < DataTypes.DOUBLE.getSizeInBytes() || (
-          (isComplexPrimitive) && (adaptiveDataType.getSizeInBytes() == DataTypes.DOUBLE
-              .getSizeInBytes()))) {
-        return new AdaptiveFloatingCodec(srcDataType, adaptiveDataType, stats);
-      } else {
+      // 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 {
+        long max = (long) (Math.pow(10, decimalCount) * absMaxValue);
+        DataType adaptiveDataType = fitLongMinMax(max, 0);
+        DataType deltaDataType = compareMinMaxAndSelectDataType(
+            (long) (Math.pow(10, decimalCount) * (maxValue - minValue)));
+        if (adaptiveDataType.getSizeInBytes() > deltaDataType.getSizeInBytes()) {
+          return new AdaptiveDeltaFloatingCodec(srcDataType, deltaDataType, stats);
+        } else if (adaptiveDataType.getSizeInBytes() < DataTypes.DOUBLE.getSizeInBytes() || (
+            (isComplexPrimitive) && (adaptiveDataType.getSizeInBytes() == DataTypes.DOUBLE
+                .getSizeInBytes()))) {
+          return new AdaptiveFloatingCodec(srcDataType, adaptiveDataType, stats);
+        } else {
+          return new DirectCompressCodec(DataTypes.DOUBLE);
+        }
       }
     }
   }
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/complexType/TestAdaptiveComplexType.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/complexType/TestAdaptiveComplexType.scala
index 6b0a13f1a06..7fff15d3dca 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/complexType/TestAdaptiveComplexType.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/complexType/TestAdaptiveComplexType.scala
@@ -551,4 +551,21 @@ trait TestAdaptiveComplexType extends QueryTest {
       Seq(Row(1, Row(true, "abc", mutable.WrappedArray.make(Array(false, true, false))))))
   }
 
+  test("test Double with large decimalcount") {
+    sql("Drop table if exists adaptive")
+    sql(
+      "create table adaptive(array1 array<struct<double1:double,double2:double,double3:double>>) " +
+      "stored by 'carbondata'")
+    sql(
+      "insert into adaptive values('10.35:40000.35:1.7976931348623157$67890985.888:65.5656:200')," +
+      "('20.25:50000.25:4.945464565654656546546546324$10000000:300000:3000')")
+    checkExistence(sql("select * from adaptive"), true, "1.0E7,300000.0,3000.0")
+    sql("Drop table if exists adaptive")
+    sql("create table adaptive(struct_arr struct<array_db1:array<double>>) stored by 'carbondata'")
+    sql("insert into adaptive values('5555555.9559:12345678991234567:3444.999')")
+    checkExistence(sql("select * from adaptive"),
+      true,
+      "5555555.9559, 1.2345678991234568E16, 3444.999")
+  }
+
 }