carbondata-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ravipes...@apache.org
Subject [02/42] carbondata git commit: [CARBONDATA-888] Added options to include and exclude dictionary columns in dataframe
Date Thu, 15 Jun 2017 11:50:07 GMT
[CARBONDATA-888] Added options to include and exclude dictionary columns in dataframe


Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/01048f86
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/01048f86
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/01048f86

Branch: refs/heads/branch-1.1
Commit: 01048f86f3725df0618b52615a7ba0992d61d412
Parents: f9fb244
Author: Sanoj MG <sanoj.george.dev@gmail.com>
Authored: Tue Apr 18 17:22:14 2017 +0800
Committer: ravipesala <ravi.pesala@gmail.com>
Committed: Thu Jun 15 12:43:48 2017 +0530

----------------------------------------------------------------------
 .../testsuite/dataload/TestLoadDataFrame.scala  | 47 +++++++++++++++++++-
 .../spark/CarbonDataFrameWriter.scala           |  5 +++
 .../spark/sql/CarbonDataFrameWriter.scala       | 29 +++---------
 3 files changed, 56 insertions(+), 25 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/carbondata/blob/01048f86/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/dataload/TestLoadDataFrame.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/dataload/TestLoadDataFrame.scala
b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/dataload/TestLoadDataFrame.scala
index 2d86497..f50620f 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/dataload/TestLoadDataFrame.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/dataload/TestLoadDataFrame.scala
@@ -27,6 +27,7 @@ import org.scalatest.BeforeAndAfterAll
 class TestLoadDataFrame extends QueryTest with BeforeAndAfterAll {
   var df: DataFrame = _
   var dataFrame: DataFrame = _
+  var df2: DataFrame = _
 
 
   def buildTestData() = {
@@ -45,6 +46,9 @@ class TestLoadDataFrame extends QueryTest with BeforeAndAfterAll {
       StructField("string", StringType, nullable = false) :: Nil)
 
     dataFrame = sqlContext.createDataFrame(rdd, schema)
+    df2 = sqlContext.sparkContext.parallelize(1 to 1000)
+      .map(x => ("key_" + x, "str_" + x, x, x * 2, x * 3))
+      .toDF("c1", "c2", "c3", "c4", "c5")
   }
 
   def dropTable() = {
@@ -52,7 +56,9 @@ class TestLoadDataFrame extends QueryTest with BeforeAndAfterAll {
     sql("DROP TABLE IF EXISTS carbon2")
     sql("DROP TABLE IF EXISTS carbon3")
     sql("DROP TABLE IF EXISTS carbon4")
-
+    sql("DROP TABLE IF EXISTS carbon5")
+    sql("DROP TABLE IF EXISTS carbon6")
+    sql("DROP TABLE IF EXISTS carbon7")
   }
 
 
@@ -121,6 +127,45 @@ class TestLoadDataFrame extends QueryTest with BeforeAndAfterAll {
       sql("SELECT count(*) FROM carbon2"),Seq(Row(32000)))
   }
 
+  test("test load dataframe with integer columns included in the dictionary"){
+    df2.write
+      .format("carbondata")
+      .option("tableName", "carbon5")
+      .option("compress", "true")
+      .option("dictionary_include","c3,c4")
+      .mode(SaveMode.Overwrite)
+      .save()
+    checkAnswer(
+      sql("select count(*) from carbon5 where c3 > 300"), Row(700)
+    )
+  }
+
+  test("test load dataframe with string column excluded from the dictionary"){
+    df2.write
+      .format("carbondata")
+      .option("tableName", "carbon6")
+      .option("compress", "true")
+      .option("dictionary_exclude","c2")
+      .mode(SaveMode.Overwrite)
+      .save()
+    checkAnswer(
+      sql("select count(*) from carbon6 where c3 > 300"), Row(700)
+    )
+  }
+
+  test("test load dataframe with both dictionary include and exclude specified"){
+    df2.write
+      .format("carbondata")
+      .option("tableName", "carbon7")
+      .option("compress", "true")
+      .option("dictionary_include","c3,c4")
+      .option("dictionary_exclude","c2")
+      .mode(SaveMode.Overwrite)
+      .save()
+    checkAnswer(
+      sql("select count(*) from carbon7 where c3 > 300"), Row(700)
+    )
+  }
 
   override def afterAll {
     dropTable

http://git-wip-us.apache.org/repos/asf/carbondata/blob/01048f86/integration/spark/src/main/scala/org/apache/carbondata/spark/CarbonDataFrameWriter.scala
----------------------------------------------------------------------
diff --git a/integration/spark/src/main/scala/org/apache/carbondata/spark/CarbonDataFrameWriter.scala
b/integration/spark/src/main/scala/org/apache/carbondata/spark/CarbonDataFrameWriter.scala
index 917e2ed..36e2440 100644
--- a/integration/spark/src/main/scala/org/apache/carbondata/spark/CarbonDataFrameWriter.scala
+++ b/integration/spark/src/main/scala/org/apache/carbondata/spark/CarbonDataFrameWriter.scala
@@ -174,6 +174,10 @@ class CarbonDataFrameWriter(val dataFrame: DataFrame) {
   }
 
   private def makeCreateTableString(schema: StructType, options: CarbonOption): String =
{
+    val properties = Map(
+      "DICTIONARY_INCLUDE" -> options.dictionaryInclude,
+      "DICTIONARY_EXCLUDE" -> options.dictionaryExclude
+    ).filter(_._2.isDefined).map(p => s"'${p._1}' = '${p._2.get}'").mkString(",")
     val carbonSchema = schema.map { field =>
       s"${ field.name } ${ convertToCarbonType(field.dataType) }"
     }
@@ -181,6 +185,7 @@ class CarbonDataFrameWriter(val dataFrame: DataFrame) {
           CREATE TABLE IF NOT EXISTS ${options.dbName}.${options.tableName}
           (${ carbonSchema.mkString(", ") })
           STORED BY '${ CarbonContext.datasourceName }'
+          ${ if (properties.nonEmpty) " TBLPROPERTIES (" + properties + ")" else ""}
       """
   }
 

http://git-wip-us.apache.org/repos/asf/carbondata/blob/01048f86/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDataFrameWriter.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDataFrameWriter.scala
b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDataFrameWriter.scala
index 4de644a..1054c62 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDataFrameWriter.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonDataFrameWriter.scala
@@ -161,30 +161,11 @@ class CarbonDataFrameWriter(sqlContext: SQLContext, val dataFrame: DataFrame)
{
     val carbonSchema = schema.map { field =>
       s"${ field.name } ${ convertToCarbonType(field.dataType) }"
     }
-    var property = new StringBuilder
-    property.append(
-      if (options.dictionaryInclude.isDefined) {
-        s"'DICTIONARY_INCLUDE' = '${options.dictionaryInclude.get}' ,"
-      } else {
-        ""
-      }
-    ).append(
-      if (options.dictionaryExclude.isDefined) {
-        s"'DICTIONARY_EXCLUDE' = '${options.dictionaryExclude.get}' ,"
-      } else {
-        ""
-      }
-    ).append(
-      if (options.tableBlockSize.isDefined) {
-        s"'table_blocksize' = '${options.tableBlockSize.get}'"
-      } else {
-        ""
-      }
-    )
-    if (property.nonEmpty && property.charAt(property.length-1) == ',') {
-      property = property.replace(property.length - 1, property.length, "")
-    }
-
+    val property = Map(
+      "DICTIONARY_INCLUDE" -> options.dictionaryInclude,
+      "DICTIONARY_EXCLUDE" -> options.dictionaryExclude,
+      "TABLE_BLOCKSIZE" -> options.tableBlockSize
+    ).filter(_._2.isDefined).map(p => s"'${p._1}' = '${p._2.get}'").mkString(",")
     s"""
        | CREATE TABLE IF NOT EXISTS ${options.dbName}.${options.tableName}
        | (${ carbonSchema.mkString(", ") })


Mime
View raw message