You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@carbondata.apache.org by ra...@apache.org on 2017/05/15 05:06:21 UTC
[08/19] carbondata git commit: [CARBONDATA-888] Added options to
include and exclude dictionary columns in dataframe
[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/f2a51767
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/f2a51767
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/f2a51767
Branch: refs/heads/master
Commit: f2a517679e753b63325f8d8fcc0f33f3d76d0e79
Parents: f5c6f28
Author: Sanoj MG <sa...@gmail.com>
Authored: Tue Apr 18 17:22:14 2017 +0800
Committer: jackylk <ja...@huawei.com>
Committed: Sun May 14 20:40:12 2017 +0800
----------------------------------------------------------------------
.../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/f2a51767/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/f2a51767/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/f2a51767/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(", ") })