[CARBONDATA-2606]Fix Complex array Pushdown and block auto merge comp…

…action
apache · Jul 24, 2018 · d48eef0 · d48eef0
1 parent bea277f
commit d48eef0
Show file tree

Hide file tree

Showing 9 changed files with 115 additions and 25 deletions.
diff --git a/...a/org/apache/carbondata/integration/spark/testsuite/complexType/TestComplexDataType.scala b/...a/org/apache/carbondata/integration/spark/testsuite/complexType/TestComplexDataType.scala
@@ -27,6 +27,8 @@ class TestComplexDataType extends QueryTest with BeforeAndAfterAll {
   override def afterAll(): Unit = {
     sql("DROP TABLE IF EXISTS table1")
     sql("DROP TABLE IF EXISTS test")
+    CarbonProperties.getInstance()
+      .addProperty(CarbonCommonConstants.ENABLE_AUTO_LOAD_MERGE, "false")
   }
 
   test("test Projection PushDown for Struct - Integer type") {
@@ -885,4 +887,62 @@ class TestComplexDataType extends QueryTest with BeforeAndAfterAll {
     checkExistence(sql("select * from table1"),true,"1.0E9")
   }
 
+  test("test block compaction - auto merge") {
+    sql("DROP TABLE IF EXISTS table1")
+    CarbonProperties.getInstance()
+      .addProperty(CarbonCommonConstants.ENABLE_AUTO_LOAD_MERGE, "true")
+    sql(
+      "create table table1 (roll int,person Struct<detail:int,age:string,height:double>) stored " +
+      "by 'carbondata'")
+    sql(
+      "load data inpath '" + resourcesPath +
+      "/Struct.csv' into table table1 options('delimiter'=','," +
+      "'quotechar'='\"','fileheader'='roll,person','complex_delimiter_level_1'='$'," +
+      "'complex_delimiter_level_2'='&')")
+    sql(
+      "load data inpath '" + resourcesPath +
+      "/Struct.csv' into table table1 options('delimiter'=','," +
+      "'quotechar'='\"','fileheader'='roll,person','complex_delimiter_level_1'='$'," +
+      "'complex_delimiter_level_2'='&')")
+    sql(
+      "load data inpath '" + resourcesPath +
+      "/Struct.csv' into table table1 options('delimiter'=','," +
+      "'quotechar'='\"','fileheader'='roll,person','complex_delimiter_level_1'='$'," +
+      "'complex_delimiter_level_2'='&')")
+    sql(
+      "load data inpath '" + resourcesPath +
+      "/Struct.csv' into table table1 options('delimiter'=','," +
+      "'quotechar'='\"','fileheader'='roll,person','complex_delimiter_level_1'='$'," +
+      "'complex_delimiter_level_2'='&')")
+    checkExistence(sql("show segments for table table1"),false, "Compacted")
+    CarbonProperties.getInstance()
+      .addProperty(CarbonCommonConstants.ENABLE_AUTO_LOAD_MERGE, "false")
+  }
+
+  test("decimal with two level struct type") {
+    sql("DROP TABLE IF EXISTS test")
+    sql(
+      "create table test(id int,a struct<c:struct<d:decimal(20,10)>>) stored by 'carbondata' " +
+      "tblproperties('dictionary_include'='a')")
+    checkExistence(sql("desc test"),true,"struct<c:struct<d:decimal(20,10)>>")
+    checkExistence(sql("describe formatted test"),true,"struct<c:struct<d:decimal(20,10)>>")
+    sql("insert into test values(1,'3999.999')")
+    checkExistence(sql("select * from test"),true,"3999.9990000000")
+  }
+
+  test("test dictionary include for second struct and array column") {
+    sql("DROP TABLE IF EXISTS test")
+    sql(
+      "create table test(id int,a struct<b:int,c:int>, d struct<e:int,f:int>, d1 struct<e1:int," +
+      "f1:int>) stored by 'carbondata' tblproperties('dictionary_include'='d1')")
+    sql("insert into test values(1,'2$3','4$5','6$7')")
+    checkAnswer(sql("select * from test"),Seq(Row(1,Row(2,3),Row(4,5),Row(6,7))))
+    sql("DROP TABLE IF EXISTS test")
+    sql(
+      "create table test(a array<int>, b array<int>) stored by 'carbondata' tblproperties" +
+      "('dictionary_include'='b')")
+    sql("insert into test values(1,2) ")
+    checkAnswer(sql("select b[0] from test"),Seq(Row(2)))
+  }
+
 }
diff --git a/integration/spark-common/src/main/scala/org/apache/spark/util/SparkTypeConverter.scala b/integration/spark-common/src/main/scala/org/apache/spark/util/SparkTypeConverter.scala
@@ -128,7 +128,10 @@ private[spark] object SparkTypeConverter {
       case "struct" => s"${
         childDim.getColName.substring(dimName.length + 1)
       }:struct<${ getStructChildren(table, childDim.getColName) }>"
-      case dType => s"${ childDim.getColName.substring(dimName.length + 1) }:${ dType }"
+      case dType => s"${
+        childDim.getColName
+          .substring(dimName.length + 1)
+      }:${ addDecimalScaleAndPrecision(childDim, dType) }"
     }
   }
 }
diff --git a/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala b/integration/spark2/src/main/scala/org/apache/carbondata/spark/rdd/CarbonDataRDDFactory.scala
@@ -578,13 +578,19 @@ object CarbonDataRDDFactory {
         if (carbonTable.isHivePartitionTable) {
           carbonLoadModel.setFactTimeStamp(System.currentTimeMillis())
         }
-        val compactedSegments = new util.ArrayList[String]()
-        handleSegmentMerging(sqlContext,
-          carbonLoadModel,
-          carbonTable,
-          compactedSegments,
-          operationContext)
-        carbonLoadModel.setMergedSegmentIds(compactedSegments)
+        // Block compaction for table containing complex datatype
+        if (carbonTable.getTableInfo.getFactTable.getListOfColumns.asScala
+          .exists(m => m.getDataType.isComplexType)) {
+          LOGGER.warn("Compaction is skipped as table contains complex columns")
+        } else {
+          val compactedSegments = new util.ArrayList[String]()
+          handleSegmentMerging(sqlContext,
+            carbonLoadModel,
+            carbonTable,
+            compactedSegments,
+            operationContext)
+          carbonLoadModel.setMergedSegmentIds(compactedSegments)
+        }
       } catch {
         case e: Exception =>
           throw new Exception(

diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDatasourceHadoopRelation.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDatasourceHadoopRelation.scala
@@ -97,7 +97,7 @@ case class CarbonDatasourceHadoopRelation(
           breakable({
             while (ifGetArrayItemExists.containsChild != null) {
               if (ifGetArrayItemExists.childSchema.toString().contains("ArrayType")) {
-                arrayTypeExists = s.childSchema.toString().contains("ArrayType")
+                arrayTypeExists = ifGetArrayItemExists.childSchema.toString().contains("ArrayType")
                 break
               }
               if (ifGetArrayItemExists.child.isInstanceOf[AttributeReference]) {

diff --git a/.../main/scala/org/apache/spark/sql/execution/command/management/CarbonLoadDataCommand.scala b/.../main/scala/org/apache/spark/sql/execution/command/management/CarbonLoadDataCommand.scala
@@ -77,6 +77,7 @@ import org.apache.carbondata.spark.dictionary.provider.SecureDictionaryServicePr
 import org.apache.carbondata.spark.dictionary.server.SecureDictionaryServer
 import org.apache.carbondata.spark.load.{CsvRDDHelper, DataLoadProcessorStepOnSpark}
 import org.apache.carbondata.spark.rdd.CarbonDataRDDFactory
+import org.apache.carbondata.spark.rdd.CarbonDataRDDFactory.LOGGER
 import org.apache.carbondata.spark.util.{CarbonScalaUtil, CommonUtil, GlobalDictionaryUtil, SparkDataTypeConverterImpl}
 
 case class CarbonLoadDataCommand(
@@ -823,15 +824,21 @@ case class CarbonLoadDataCommand(
     }
     try {
       carbonLoadModel.setFactTimeStamp(System.currentTimeMillis())
-      val compactedSegments = new util.ArrayList[String]()
-      // Trigger auto compaction
-      CarbonDataRDDFactory.handleSegmentMerging(
-        sparkSession.sqlContext,
-        carbonLoadModel,
-        table,
-        compactedSegments,
-        operationContext)
-      carbonLoadModel.setMergedSegmentIds(compactedSegments)
+      // Block compaction for table containing complex datatype
+      if (table.getTableInfo.getFactTable.getListOfColumns.asScala
+        .exists(m => m.getDataType.isComplexType)) {
+        LOGGER.warn("Compaction is skipped as table contains complex columns")
+      } else {
+        val compactedSegments = new util.ArrayList[String]()
+        // Trigger auto compaction
+        CarbonDataRDDFactory.handleSegmentMerging(
+          sparkSession.sqlContext,
+          carbonLoadModel,
+          table,
+          compactedSegments,
+          operationContext)
+        carbonLoadModel.setMergedSegmentIds(compactedSegments)
+      }
     } catch {
       case e: Exception =>
         throw new Exception(

diff --git a/processing/src/main/java/org/apache/carbondata/processing/datatypes/ArrayDataType.java b/processing/src/main/java/org/apache/carbondata/processing/datatypes/ArrayDataType.java
@@ -172,8 +172,10 @@ public void setSurrogateIndex(int surrIndex) {
 
   @Override
   public void fillCardinality(List<Integer> dimCardWithComplex) {
-    dimCardWithComplex.add(0);
-    children.fillCardinality(dimCardWithComplex);
+    if (children.getIsColumnDictionary()) {
+      dimCardWithComplex.add(0);
+      children.fillCardinality(dimCardWithComplex);
+    }
   }
 
   @Override

diff --git a/processing/src/main/java/org/apache/carbondata/processing/datatypes/StructDataType.java b/processing/src/main/java/org/apache/carbondata/processing/datatypes/StructDataType.java
@@ -178,9 +178,17 @@ public void setSurrogateIndex(int surrIndex) {
 
   @Override
   public void fillCardinality(List<Integer> dimCardWithComplex) {
-    dimCardWithComplex.add(0);
-    for (int i = 0; i < children.size(); i++) {
-      children.get(i).fillCardinality(dimCardWithComplex);
+    boolean isDictionaryColumn = false;
+    for (GenericDataType child : children) {
+      if (child.getIsColumnDictionary()) {
+        isDictionaryColumn = true;
+      }
+    }
+    if (isDictionaryColumn) {
+      dimCardWithComplex.add(0);
+      for (int i = 0; i < children.size(); i++) {
+        children.get(i).fillCardinality(dimCardWithComplex);
+      }
     }
   }
 

diff --git a/...g/src/main/java/org/apache/carbondata/processing/store/CarbonFactDataHandlerColumnar.java b/...g/src/main/java/org/apache/carbondata/processing/store/CarbonFactDataHandlerColumnar.java
@@ -165,7 +165,9 @@ private void setComplexMapSurrogateIndex(int dimensionCount) {
         List<GenericDataType> primitiveTypes = new ArrayList<GenericDataType>();
         complexDataType.getAllPrimitiveChildren(primitiveTypes);
         for (GenericDataType eachPrimitive : primitiveTypes) {
-          eachPrimitive.setSurrogateIndex(surrIndex++);
+          if (eachPrimitive.getIsColumnDictionary()) {
+            eachPrimitive.setSurrogateIndex(surrIndex++);
+          }
         }
       } else {
         surrIndex++;

diff --git a/...sing/src/main/java/org/apache/carbondata/processing/store/CarbonFactDataHandlerModel.java b/...sing/src/main/java/org/apache/carbondata/processing/store/CarbonFactDataHandlerModel.java
@@ -248,7 +248,9 @@ public static CarbonFactDataHandlerModel createCarbonFactDataHandlerModel(
       List<GenericDataType> primitiveTypes = new ArrayList<GenericDataType>();
       complexDataType.getValue().getAllPrimitiveChildren(primitiveTypes);
       for (GenericDataType eachPrimitive : primitiveTypes) {
-        eachPrimitive.setSurrogateIndex(surrIndex++);
+        if (eachPrimitive.getIsColumnDictionary()) {
+          eachPrimitive.setSurrogateIndex(surrIndex++);
+        }
       }
     }