carbondata-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jack...@apache.org
Subject [02/14] incubator-carbondata git commit: add partition validate & example
Date Mon, 08 May 2017 09:48:14 GMT
add partition validate & example


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

Branch: refs/heads/12-dev
Commit: 28e7981d3dea7570532d50c614e15fafbf98e650
Parents: c95b311
Author: lionelcao <whucaolu@gmail.com>
Authored: Tue Apr 25 17:49:18 2017 +0800
Committer: lionelcao <whucaolu@gmail.com>
Committed: Thu May 4 14:53:09 2017 +0800

----------------------------------------------------------------------
 .../src/main/resources/partition_data.csv       | 25 ++++++
 .../examples/CarbonPartitionExample.scala       | 91 ++++++++++++++++++++
 .../carbondata/spark/util/CommonUtil.scala      | 34 ++++++++
 .../spark/sql/parser/CarbonSparkSqlParser.scala | 20 +++--
 4 files changed, 163 insertions(+), 7 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/28e7981d/examples/spark2/src/main/resources/partition_data.csv
----------------------------------------------------------------------
diff --git a/examples/spark2/src/main/resources/partition_data.csv b/examples/spark2/src/main/resources/partition_data.csv
new file mode 100644
index 0000000..2d521b7
--- /dev/null
+++ b/examples/spark2/src/main/resources/partition_data.csv
@@ -0,0 +1,25 @@
+vin,logdate,phonenumber,country,area
+A42158424831,2016/02/12,125371341,China,Asia
+A42158473831,2016/01/12,125371342,China,Asia
+A42152474832,2016/02/12,125371343,US,America
+A42151477823,2016/12/12,125371344,China,Asia
+A42158474135,2016/02/15,125371345,Japan,Asia
+A42258434831,2016/12/12,125371346,China,Asia
+A42158475831,2016/05/12,125371347,UK,Europe
+A41158494830,2015/07/12,225371348,China,Asia
+A42158424831,2015/02/12,225371349,China,Asia
+A42158473830,2014/01/12,225371310,China,Asia
+A42152474830,2013/02/12,325371311,US,America
+A42151477823,2012/12/12,425371312,China,Asia
+A42158474133,2012/02/15,325371313,Japan,Asia
+A42258434835,2013/12/12,525371314,China,Asia
+A42158475836,2014/05/12,625371315,UK,Europe
+A41158494838,2015/07/12,525371316,China,Asia
+A42158424833,2016/02/12,425371317,China,Asia
+A42158473832,2017/01/12,325371318,China,Asia
+A42152474834,2011/02/12,225371319,US,America
+A42151477824,2012/12/12,225371320,China,Asia
+A42158474137,2013/02/15,325371321,Japan,Asia
+A42258434837,2014/12/12,25371322,China,Asia
+A42158475838,2014/05/12,425371323,UK,Europe
+A41158494839,2016/07/12,625371324,China,Asia

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/28e7981d/examples/spark2/src/main/scala/org/apache/carbondata/examples/CarbonPartitionExample.scala
----------------------------------------------------------------------
diff --git a/examples/spark2/src/main/scala/org/apache/carbondata/examples/CarbonPartitionExample.scala
b/examples/spark2/src/main/scala/org/apache/carbondata/examples/CarbonPartitionExample.scala
new file mode 100644
index 0000000..96c1c72
--- /dev/null
+++ b/examples/spark2/src/main/scala/org/apache/carbondata/examples/CarbonPartitionExample.scala
@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.carbondata.examples
+
+import java.io.File
+
+import org.apache.spark.sql.SparkSession
+
+import org.apache.carbondata.core.constants.CarbonCommonConstants
+import org.apache.carbondata.core.util.CarbonProperties
+
+object CarbonPartitionExample {
+
+  def main(args: Array[String]) {
+    val rootPath = new File(this.getClass.getResource("/").getPath
+                            + "../../../..").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 testData = s"$rootPath/examples/spark2/src/main/resources/partition_data.csv"
+
+    CarbonProperties.getInstance()
+      .addProperty(CarbonCommonConstants.CARBON_TIMESTAMP_FORMAT, "yyyy/MM/dd")
+
+    import org.apache.spark.sql.CarbonSession._
+
+    val spark = SparkSession
+      .builder()
+      .master("local")
+      .appName("CarbonPartitionExample")
+      .config("spark.sql.warehouse.dir", warehouse)
+      .getOrCreateCarbonSession(storeLocation, metastoredb)
+
+    spark.sparkContext.setLogLevel("WARN")
+
+    spark.sql("DROP TABLE IF EXISTS non_partition_table")
+
+    spark.sql("""
+                | CREATE TABLE IF NOT EXISTS non_partition_table
+                | (
+                | vin String,
+                | logdate Timestamp,
+                | phonenumber Long,
+                | country String,
+                | area String
+                | )
+                | STORED BY 'carbondata'
+              """.stripMargin)
+
+    spark.sql("DROP TABLE IF EXISTS partition_table")
+
+    spark.sql("""
+       | CREATE TABLE IF NOT EXISTS partition_table
+       | (
+       | vin String,
+       | logdate Timestamp,
+       | phonenumber Long,
+       | country String,
+       | area String
+       | )
+       | PARTITIONED BY (vin String)
+       | STORED BY 'carbondata'
+       | TBLPROPERTIES('PARTITIONING'='HASH','PARTITIONCOUNT'='5')
+       """.stripMargin)
+
+    //spark.sql(s"""
+    //   LOAD DATA LOCAL INPATH '$testData' into table rx5_tbox_parquet_all options('BAD_RECORDS_ACTION'='FORCE')
+    //   """)
+
+    //spark.sql("select vin, count(*) from rx5_tbox_parquet_all group by vin order by count(*)
desc").show(50)
+
+    // Drop table
+    //spark.sql("DROP TABLE IF EXISTS carbon_table")
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/28e7981d/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/CommonUtil.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/CommonUtil.scala
b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/CommonUtil.scala
index 679a4e7..774261d 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/CommonUtil.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/spark/util/CommonUtil.scala
@@ -27,6 +27,7 @@ import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat
 import org.apache.spark.SparkContext
 import org.apache.spark.sql.execution.command.{ColumnProperty, Field}
+import org.apache.spark.sql.types.StructField
 import org.apache.spark.util.FileUtils
 
 import org.apache.carbondata.common.logging.LogServiceFactory
@@ -144,6 +145,39 @@ object CommonUtil {
     isValid
   }
 
+  /**
+   * 1. If partition by clause exists, then partitioning should be defined
+   * 2. If partitioning is Hash, then partitioncount should be defined
+   * 3. If partitioning is List, then value list should be defined
+   * 4. If partitioning is Range, then boundary value list should be defined
+   * 5. Only support single level partition for now
+   * @param tableProperties
+   * @param partitionCols
+   * @return
+   */
+  def validatePartitionColumns(tableProperties: Map[String, String],
+      partitionCols: Seq[StructField]): Boolean = {
+    var isValid: Boolean = true
+    val partitioning = tableProperties.get(CarbonCommonConstants.PARTITIONING)
+    val partitioncount = tableProperties.get(CarbonCommonConstants.PARTITIONCOUNT)
+
+    //partition column and partitioning should be both exist or not exist
+    if (partitionCols.isEmpty ^ partitioning.isEmpty) {
+      isValid = false
+    } else if (partitionCols.nonEmpty) {
+      partitioning.get.toUpperCase() match {
+        case "HASH" => if (!partitioncount.isDefined) isValid = false
+        case "LIST" => isValid = false
+        case "RANGE" => isValid = false
+        case "RANGE_INTERVAL" => isValid = false
+        case _ => isValid = false
+      }
+    }
+    //only support one partition column for now
+    if (partitionCols.length > 1) isValid = false
+    isValid
+  }
+
   def validateFields(key: String, fields: Seq[Field]): Boolean = {
     var isValid: Boolean = false
     fields.foreach { field =>

http://git-wip-us.apache.org/repos/asf/incubator-carbondata/blob/28e7981d/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSparkSqlParser.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSparkSqlParser.scala
b/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSparkSqlParser.scala
index b22422b..9aacc7f 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSparkSqlParser.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSparkSqlParser.scala
@@ -114,11 +114,19 @@ class CarbonSqlAstBuilder(conf: SQLConf) extends SparkSqlAstBuilder(conf)
{
                             duplicateColumns.mkString("[", ",", "]"), ctx)
       }
 
-      // partition columns must be part of the schema
-      val badPartCols = partitionCols.map(_.name).toSet.intersect(colNames.toSet)
-      if (badPartCols.isEmpty) {
-        operationNotAllowed(s"Partition columns must be specified in the schema: " +
-                            badPartCols.map("\"" + _ + "\"").mkString("[", ",", "]"), ctx)
+      val tableProperties = mutable.Map[String, String]()
+      properties.foreach(f => tableProperties.put(f._1, f._2.toLowerCase))
+
+      // validate partition clause
+      if (!CommonUtil.validatePartitionColumns(tableProperties, partitionCols)) {
+        operationNotAllowed("Invalid Partition definition", ctx)
+      } else if (partitionCols.nonEmpty) {
+        // partition columns must be part of the schema
+        val badPartCols = partitionCols.map(_.name).toSet.intersect(colNames.toSet)
+        if (badPartCols.isEmpty) {
+          operationNotAllowed(s"Partition columns must be specified in the schema: " +
+                              badPartCols.map("\"" + _ + "\"").mkString("[", ",", "]"), ctx)
+        }
       }
 
       val fields = cols.map { col =>
@@ -171,8 +179,6 @@ class CarbonSqlAstBuilder(conf: SQLConf) extends SparkSqlAstBuilder(conf)
{
         None
       }
 
-      val tableProperties = mutable.Map[String, String]()
-      properties.foreach(f => tableProperties.put(f._1, f._2.toLowerCase))
       // prepare table model of the collected tokens
       val tableModel: TableModel = parser.prepareTableModel(ifNotExists,
         convertDbNameToLowerCase(name.database),


Mime
View raw message