Mailing-List: contact commits-help@carbondata.incubator.apache.org; run by ezmlm
Precedence: bulk
Reply-To: dev@carbondata.incubator.apache.org
Content-Type: text/plain; charset="us-ascii"
MIME-Version: 1.0
Content-Transfer-Encoding: 7bit
From: jackylk@apache.org
To: commits@carbondata.incubator.apache.org
Date: Thu, 19 Jan 2017 15:20:09 -0000
Message-Id: <bc56ec32cabf4d24b5537253989a5a77@git.apache.org>
Subject: [1/2] incubator-carbondata git commit: Fixed insert with select query
 when functions are used in query
archived-at: Thu, 19 Jan 2017 15:20:21 -0000

Repository: incubator-carbondata
Updated Branches:
  refs/heads/master db2c5f917 -> 68a16d2c9


Fixed insert with select query when functions are used in query

reverted example

Fixed testcase


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

Branch: refs/heads/master
Commit: 529c06d22d715249cf2e42db3420af698f0f917a
Parents: db2c5f9
Author: ravipesala <ravi.pesala@gmail.com>
Authored: Thu Jan 19 06:30:24 2017 +0530
Committer: jackylk <jacky.likun@huawei.com>
Committed: Thu Jan 19 23:19:26 2017 +0800

----------------------------------------------------------------------
 .../examples/CarbonSessionExample.scala         | 19 +++++-
 .../src/test/resources/data_with_all_types.csv  | 10 ++++
 .../InsertIntoCarbonTableTestCase.scala         | 62 ++++++++++++++++----
 .../execution/CarbonLateDecodeStrategy.scala    | 45 +++++++-------
 4 files changed, 102 insertions(+), 34 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/529c06d2/examples/spark2/src/main/scala/org/apache/carbondata/examples/CarbonSessionExample.scala
----------------------------------------------------------------------
diff --git a/examples/spark2/src/main/scala/org/apache/carbondata/examples/CarbonSessionExample.scala b/examples/spark2/src/main/scala/org/apache/carbondata/examples/CarbonSessionExample.scala
index 1d485cd..0d9c43f 100644
--- a/examples/spark2/src/main/scala/org/apache/carbondata/examples/CarbonSessionExample.scala
+++ b/examples/spark2/src/main/scala/org/apache/carbondata/examples/CarbonSessionExample.scala
@@ -19,6 +19,7 @@ package org.apache.carbondata.examples
 
 import java.io.File
 
+import org.apache.commons.io.FileUtils
 import org.apache.spark.sql.SparkSession
 
 import org.apache.carbondata.core.constants.CarbonCommonConstants
@@ -31,10 +32,19 @@ object CarbonSessionExample {
                             + "../../../..").getCanonicalPath
     val storeLocation = s"$rootPath/examples/spark2/target/store"
     val warehouse = s"$rootPath/examples/spark2/target/warehouse"
-    val metastoredb = s"$rootPath/examples/spark2/target"
+    val metastoredb = s"$rootPath/examples/spark2/target/metastore_db"
+
+    // clean data folder
+    if (true) {
+      val clean = (path: String) => FileUtils.deleteDirectory(new File(path))
+      clean(storeLocation)
+      clean(warehouse)
+      clean(metastoredb)
+    }
 
     CarbonProperties.getInstance()
       .addProperty("carbon.kettle.home", s"$rootPath/processing/carbonplugins")
+      .addProperty("carbon.storelocation", storeLocation)
       .addProperty(CarbonCommonConstants.CARBON_TIMESTAMP_FORMAT, "yyyy/MM/dd")
 
     import org.apache.spark.sql.CarbonSession._
@@ -42,9 +52,12 @@ object CarbonSessionExample {
     val spark = SparkSession
       .builder()
       .master("local")
-      .appName("CarbonSessionExample")
+      .appName("CarbonExample")
+      .enableHiveSupport()
       .config("spark.sql.warehouse.dir", warehouse)
-      .getOrCreateCarbonSession(storeLocation, metastoredb)
+      .config("javax.jdo.option.ConnectionURL",
+    s"jdbc:derby:;databaseName=$metastoredb;create=true")
+      .getOrCreateCarbonSession()
 
     spark.sparkContext.setLogLevel("WARN")
 

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/529c06d2/integration/spark-common-test/src/test/resources/data_with_all_types.csv
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/resources/data_with_all_types.csv b/integration/spark-common-test/src/test/resources/data_with_all_types.csv
new file mode 100644
index 0000000..5efff00
--- /dev/null
+++ b/integration/spark-common-test/src/test/resources/data_with_all_types.csv
@@ -0,0 +1,10 @@
+1,10,100,48.4,spark,2015/4/23 12:01:01,1.23,2015/4/23 11:01:01,aaa,2.5
+5,17,140,43.4,spark,2015/7/27 12:01:02,3.45,2015/7/27 11:01:02,bbb,2.5
+1,11,100,44.4,flink,2015/5/23 12:01:03,23.23,2015/5/23 11:01:03,ccc,2.5
+1,10,150,43.4,spark,2015/7/24 12:01:04,254.12,2015/7/24 11:01:04,ddd,2.5
+1,10,100,47.4,spark,2015/7/23 12:01:05,876.14,2015/7/23 11:01:05,eeee,3.5
+3,14,160,43.4,hive,2015/7/26 12:01:06,3454.32,2015/7/26 11:01:06,ff,2.5
+2,10,100,43.4,impala,2015/7/23 12:01:07,456.98,2015/7/23 11:01:07,ggg,2.5
+1,10,100,43.4,spark,2015/5/23 12:01:08,32.53,2015/5/23 11:01:08,hhh,2.5
+4,16,130,42.4,impala,2015/7/23 12:01:09,67.23,2015/7/23 11:01:09,iii,2.5
+1,10,100,43.4,spark,2015/7/23 12:01:10,832.23,2015/7/23 11:01:10,jjj,2.5

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/529c06d2/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/allqueries/InsertIntoCarbonTableTestCase.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/allqueries/InsertIntoCarbonTableTestCase.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/allqueries/InsertIntoCarbonTableTestCase.scala
index 3cb3520..db85393 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/allqueries/InsertIntoCarbonTableTestCase.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/allqueries/InsertIntoCarbonTableTestCase.scala
@@ -113,16 +113,58 @@ class InsertIntoCarbonTableTestCase extends QueryTest with BeforeAndAfterAll {
          sql("select imei,deviceInformationId,MAC from TCarbonLocal")
      )
   }
-//  test("insert->insert empty data -pass") {
-//     sql("drop table if exists TCarbon")
-//     sql("create table TCarbon (imei string,deviceInformationId int,MAC string) STORED BY 'org.apache.carbondata.format'")
-//     sql("insert into TCarbon select imei,deviceInformationId,MAC from THive where MAC='wrongdata'")
-//     val result = sql("select imei,deviceInformationId,MAC from TCarbon where MAC='wrongdata'").collect()
-//     checkAnswer(
-//         sql("select imei,deviceInformationId,MAC from THive where MAC='wrongdata'"),
-//         sql("select imei,deviceInformationId,MAC from TCarbon where MAC='wrongdata'")
-//     )
-//  }
+
+  test("insert->insert with functions") {
+    sql("DROP TABLE IF EXISTS carbon_table")
+    sql("DROP TABLE IF EXISTS carbon_table1")
+    // Create table
+    sql(
+      s"""
+         | CREATE TABLE carbon_table(
+         |    shortField smallint,
+         |    intField int,
+         |    bigintField bigint,
+         |    doubleField double,
+         |    stringField string,
+         |    timestampField timestamp,
+         |    decimalField decimal(18,2),
+         |    dateField date,
+         |    charField string,
+         |    floatField float
+         | )
+         | STORED BY 'carbondata'
+         | TBLPROPERTIES('DICTIONARY_INCLUDE'='dateField, charField')
+       """.stripMargin)
+
+    sql(
+      s"""
+         | CREATE TABLE carbon_table1(
+         |    shortField smallint,
+         |    intField int,
+         |    bigintField bigint,
+         |    doubleField double,
+         |    stringField string,
+         |    timestampField timestamp,
+         |    decimalField decimal(18,2),
+         |    dateField date,
+         |    charField string,
+         |    floatField float
+         | )
+         | STORED BY 'carbondata'
+         | TBLPROPERTIES('DICTIONARY_INCLUDE'='dateField, charField')
+       """.stripMargin)
+    sql(
+      s"""
+         | LOAD DATA LOCAL INPATH '${resourcesPath + "/data_with_all_types.csv"}'
+         | INTO TABLE carbon_table
+         | options('FILEHEADER'='shortField,intField,bigintField,doubleField,stringField,timestampField,decimalField,dateField,charField,floatField')
+       """.stripMargin)
+
+    sql("""insert into table carbon_table1 select shortField,intField,bigintField,doubleField,ASCII(stringField),
+                timestampField,decimalField,dateField,charField,floatField from carbon_table
+              """).show
+  }
+
   test("insert into existing load-pass") {
     val timeStampPropOrig = CarbonProperties.getInstance().getProperty(CarbonCommonConstants.CARBON_TIMESTAMP_FORMAT)
      CarbonProperties.getInstance()

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/529c06d2/integration/spark2/src/main/scala/org/apache/spark/sql/execution/CarbonLateDecodeStrategy.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/CarbonLateDecodeStrategy.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/CarbonLateDecodeStrategy.scala
index 6f913a8..11166bf 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/CarbonLateDecodeStrategy.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/CarbonLateDecodeStrategy.scala
@@ -223,7 +223,7 @@ private[sql] class CarbonLateDecodeStrategy extends SparkStrategy {
         pushedFilters,
         metadata,
         needDecoder,
-        updateRequestedColumns)
+        updateRequestedColumns.asInstanceOf[Seq[Attribute]])
       filterCondition.map(execution.FilterExec(_, scan)).getOrElse(scan)
     } else {
       // Don't request columns that are only referenced by pushed filters.
@@ -231,15 +231,16 @@ private[sql] class CarbonLateDecodeStrategy extends SparkStrategy {
       (projectSet ++ filterSet -- handledSet).map(relation.attributeMap).toSeq
       val updateRequestedColumns = updateRequestedColumnsFunc(requestedColumns, table, needDecoder)
       val scan = getDataSourceScan(relation,
-        updateRequestedColumns,
+        updateRequestedColumns.asInstanceOf[Seq[Attribute]],
         scanBuilder,
         candidatePredicates,
         pushedFilters,
         metadata,
         needDecoder,
-        updateRequestedColumns)
+        updateRequestedColumns.asInstanceOf[Seq[Attribute]])
       execution.ProjectExec(
-        projects, filterCondition.map(execution.FilterExec(_, scan)).getOrElse(scan))
+        updateRequestedColumnsFunc(projects, table, needDecoder).asInstanceOf[Seq[NamedExpression]],
+        filterCondition.map(execution.FilterExec(_, scan)).getOrElse(scan))
     }
   }
 
@@ -251,7 +252,7 @@ private[sql] class CarbonLateDecodeStrategy extends SparkStrategy {
       pushedFilters: Seq[Filter],
       metadata: Map[String, String],
       needDecoder: ArrayBuffer[AttributeReference],
-      updateRequestedColumns: Seq[AttributeReference]): DataSourceScanExec = {
+      updateRequestedColumns: Seq[Attribute]): DataSourceScanExec = {
     val table = relation.relation.asInstanceOf[CarbonDatasourceHadoopRelation]
     if (supportBatchedDataSource(relation.relation.sqlContext, updateRequestedColumns) &&
         needDecoder.isEmpty) {
@@ -272,29 +273,31 @@ private[sql] class CarbonLateDecodeStrategy extends SparkStrategy {
     }
   }
 
-  def updateRequestedColumnsFunc(requestedColumns: Seq[AttributeReference],
+  def updateRequestedColumnsFunc(requestedColumns: Seq[Expression],
       relation: CarbonDatasourceHadoopRelation,
-      needDecoder: ArrayBuffer[AttributeReference]): Seq[AttributeReference] = {
+      needDecoder: ArrayBuffer[AttributeReference]): Seq[Expression] = {
     val map = relation.carbonRelation.metaData.dictionaryMap
-    requestedColumns.map { attr =>
-      if (needDecoder.exists(_.name.equalsIgnoreCase(attr.name))) {
-        attr
-      } else {
-        val dict = map.get(attr.name)
-        if (dict.isDefined && dict.get) {
-          AttributeReference(attr.name,
-            IntegerType,
-            attr.nullable,
-            attr.metadata)(attr.exprId, attr.qualifier)
-        } else {
+    requestedColumns.map {
+      case attr: AttributeReference =>
+        if (needDecoder.exists(_.name.equalsIgnoreCase(attr.name))) {
           attr
+        } else {
+          val dict = map.get(attr.name)
+          if (dict.isDefined && dict.get) {
+            AttributeReference(attr.name,
+              IntegerType,
+              attr.nullable,
+              attr.metadata)(attr.exprId, attr.qualifier)
+          } else {
+            attr
+          }
         }
-      }
+      case others => others
     }
   }
 
   private def getPartitioning(carbonTable: CarbonTable,
-      output: Seq[AttributeReference]): Partitioning = {
+      output: Seq[Attribute]): Partitioning = {
     val info: BucketingInfo = carbonTable.getBucketingInfo(carbonTable.getFactTableName)
     if (info != null) {
       val cols = info.getListOfColumns.asScala
@@ -304,7 +307,7 @@ private[sql] class CarbonLateDecodeStrategy extends SparkStrategy {
       val bucketColumns = cols.flatMap { n =>
         val attrRef = output.find(_.name.equalsIgnoreCase(n.getColumnName))
         attrRef match {
-          case Some(attr) =>
+          case Some(attr: AttributeReference) =>
             Some(AttributeReference(attr.name,
               CarbonScalaUtil.convertCarbonToSparkDataType(n.getDataType),
               attr.nullable,