You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@carbondata.apache.org by ja...@apache.org on 2018/01/31 05:22:58 UTC
[38/50] [abbrv] carbondata git commit:
[CARBONDATA-2088][CARBONDATA-1516] Optimize syntax for creating timeseries
pre-aggregate table
[CARBONDATA-2088][CARBONDATA-1516] Optimize syntax for creating timeseries pre-aggregate table
change using 'timeseries' instead of using preaggregate for creating timeseries pre-aggregate table
change timeseries.eventTime and hour_granularity and so on
granularity only support one
It should throw UnsupportDataMapException if don't use timeseries or preaggregate to create datamap
This closes #1865
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/181c280b
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/181c280b
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/181c280b
Branch: refs/heads/carbonstore
Commit: 181c280b7d33ac5e4029bd935d6260b0fe79a2bf
Parents: b421c24
Author: xubo245 <60...@qq.com>
Authored: Fri Jan 26 17:47:46 2018 +0800
Committer: ravipesala <ra...@gmail.com>
Committed: Tue Jan 30 17:25:19 2018 +0530
----------------------------------------------------------------------
.../core/constants/CarbonCommonConstants.java | 4 -
.../schema/datamap/DataMapProvider.java | 32 +++
.../metadata/schema/datamap/Granularity.java | 46 +++
.../preaggregate/TestPreAggCreateCommand.scala | 60 +++-
.../TestPreAggregateTableSelection.scala | 26 +-
.../timeseries/TestTimeSeriesCreateTable.scala | 280 ++++++++++++++++---
.../timeseries/TestTimeseriesCompaction.scala | 67 ++++-
.../timeseries/TestTimeseriesDataLoad.scala | 142 +++++++++-
.../TestTimeseriesTableSelection.scala | 114 +++++++-
.../testsuite/datamap/TestDataMapCommand.scala | 155 +++++-----
.../MalformedDataMapCommandException.java | 32 +++
.../datamap/CarbonCreateDataMapCommand.scala | 72 ++---
.../CreatePreAggregateTableCommand.scala | 4 +-
.../command/timeseries/TimeSeriesUtil.scala | 83 +++++-
.../carbondata/CarbonDataSourceSuite.scala | 8 +
15 files changed, 943 insertions(+), 182 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/181c280b/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java b/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
index f46feef..cf95dd9 100644
--- a/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
+++ b/core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java
@@ -1543,10 +1543,6 @@ public final class CarbonCommonConstants {
*/
public static final long HANDOFF_SIZE_DEFAULT = 1024L * 1024 * 1024;
- public static final String TIMESERIES_EVENTTIME = "timeseries.eventtime";
-
- public static final String TIMESERIES_HIERARCHY = "timeseries.hierarchy";
-
/**
* It allows queries on hive metastore directly along with filter information, otherwise first
* fetches all partitions from hive and apply filters on it.
http://git-wip-us.apache.org/repos/asf/carbondata/blob/181c280b/core/src/main/java/org/apache/carbondata/core/metadata/schema/datamap/DataMapProvider.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/metadata/schema/datamap/DataMapProvider.java b/core/src/main/java/org/apache/carbondata/core/metadata/schema/datamap/DataMapProvider.java
new file mode 100644
index 0000000..65578b1
--- /dev/null
+++ b/core/src/main/java/org/apache/carbondata/core/metadata/schema/datamap/DataMapProvider.java
@@ -0,0 +1,32 @@
+/*
+ * 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.core.metadata.schema.datamap;
+
+/**
+ * type for create datamap
+ * The syntax of datamap creation is as follows.
+ * CREATE DATAMAP IF NOT EXISTS dataMapName ON TABLE tableName USING 'DataMapProvider'
+ * DMPROPERTIES('KEY'='VALUE') AS SELECT COUNT(COL1) FROM tableName
+ *
+ * Please refer {{org.apache.spark.sql.parser.CarbonSpark2SqlParser}}
+ */
+
+public enum DataMapProvider {
+ PREAGGREGATE,
+ TIMESERIES;
+}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/181c280b/core/src/main/java/org/apache/carbondata/core/metadata/schema/datamap/Granularity.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/metadata/schema/datamap/Granularity.java b/core/src/main/java/org/apache/carbondata/core/metadata/schema/datamap/Granularity.java
new file mode 100644
index 0000000..d6aefb6
--- /dev/null
+++ b/core/src/main/java/org/apache/carbondata/core/metadata/schema/datamap/Granularity.java
@@ -0,0 +1,46 @@
+/*
+ * 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.core.metadata.schema.datamap;
+
+/**
+ * type for create datamap
+ * The syntax of datamap creation is as follows.
+ * CREATE DATAMAP IF NOT EXISTS dataMapName ON TABLE tableName USING 'DataMapProvider'
+ * DMPROPERTIES('KEY'='VALUE') AS SELECT COUNT(COL1) FROM tableName
+ *
+ * Please refer {{org.apache.spark.sql.parser.CarbonSpark2SqlParser}}
+ */
+
+public enum Granularity {
+ YEAR("year_granularity"),
+ MONTH("month_granularity"),
+ DAY("day_granularity"),
+ HOUR("hour_granularity"),
+ MINUTE("minute_granularity"),
+ SECOND("second_granularity");
+ private String name;
+
+ Granularity(String name) {
+ this.name = name;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/181c280b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggCreateCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggCreateCommand.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggCreateCommand.scala
index 755a449..d3f0ff8 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggCreateCommand.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggCreateCommand.scala
@@ -1,15 +1,18 @@
package org.apache.carbondata.integration.spark.testsuite.preaggregate
+import scala.collection.JavaConverters._
+
import org.apache.spark.sql.CarbonDatasourceHadoopRelation
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
import org.apache.spark.sql.execution.datasources.LogicalRelation
import org.apache.spark.sql.hive.CarbonRelation
import org.apache.spark.sql.test.util.QueryTest
import org.scalatest.BeforeAndAfterAll
-import scala.collection.JavaConverters._
import org.apache.carbondata.core.metadata.encoder.Encoding
import org.apache.carbondata.core.metadata.schema.table.CarbonTable
+import org.apache.carbondata.core.metadata.schema.datamap.DataMapProvider.TIMESERIES
+import org.apache.carbondata.spark.exception.MalformedCarbonCommandException
class TestPreAggCreateCommand extends QueryTest with BeforeAndAfterAll {
@@ -212,6 +215,60 @@ class TestPreAggCreateCommand extends QueryTest with BeforeAndAfterAll {
sql("drop datamap agg0 on table maintable")
}
+ val timeSeries = TIMESERIES.toString
+
+ test("test PreAggregate table selection: create with preaggregate and hierarchy") {
+ sql("DROP TABLE IF EXISTS maintabletime")
+ sql(
+ """
+ | CREATE TABLE maintabletime(year INT,month INT,name STRING,salary INT,dob STRING)
+ | STORED BY 'carbondata'
+ | TBLPROPERTIES(
+ | 'SORT_SCOPE'='Global_sort',
+ | 'TABLE_BLOCKSIZE'='23',
+ | 'SORT_COLUMNS'='month,year,name')
+ """.stripMargin)
+ sql("INSERT INTO maintabletime SELECT 10,11,'x',12,'2014-01-01 00:00:00'")
+ sql(
+ s"""
+ | CREATE DATAMAP agg0 ON TABLE maintabletime
+ | USING 'preaggregate'
+ | AS SELECT dob,name FROM maintabletime
+ | GROUP BY dob,name
+ """.stripMargin)
+ val e = intercept[MalformedCarbonCommandException] {
+ sql(
+ s"""
+ | CREATE DATAMAP agg1 ON TABLE maintabletime
+ | USING 'preaggregate'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='dob',
+ | 'SECOND_GRANULARITY'='1')
+ | AS SELECT dob,name FROM maintabletime
+ | GROUP BY dob,name
+ """.stripMargin)
+ }
+ assert(e.getMessage.contains(s"$timeSeries keyword missing"))
+ sql("DROP TABLE IF EXISTS maintabletime")
+ }
+
+ test("test pre agg create table 21: using") {
+ sql("DROP DATAMAP agg0 ON TABLE maintable")
+
+ val e: Exception = intercept[Exception] {
+ sql(
+ """
+ | CREATE DATAMAP agg0 ON TABLE mainTable
+ | USING 'abc'
+ | AS SELECT column3, SUM(column3),column5, SUM(column5)
+ | FROM maintable
+ | GROUP BY column3,column5,column2
+ """.stripMargin)
+ }
+ assert(e.getMessage.contains(
+ s"Unknown data map type abc"))
+ sql("DROP DATAMAP agg0 ON TABLE maintable")
+ }
def getCarbontable(plan: LogicalPlan) : CarbonTable ={
var carbonTable : CarbonTable = null
@@ -239,5 +296,6 @@ class TestPreAggCreateCommand extends QueryTest with BeforeAndAfterAll {
sql("drop table if exists PreAggMain")
sql("drop table if exists PreAggMain1")
sql("drop table if exists PreAggMain2")
+ sql("drop table if exists maintabletime")
}
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/181c280b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateTableSelection.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateTableSelection.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateTableSelection.scala
index 17d95ef..f9ac354 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateTableSelection.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateTableSelection.scala
@@ -23,6 +23,8 @@ import org.apache.spark.sql.{CarbonDatasourceHadoopRelation, Row}
import org.apache.spark.sql.test.util.QueryTest
import org.scalatest.BeforeAndAfterAll
+import org.apache.carbondata.core.metadata.schema.datamap.DataMapProvider.TIMESERIES
+
class TestPreAggregateTableSelection extends QueryTest with BeforeAndAfterAll {
override def beforeAll: Unit = {
@@ -267,6 +269,8 @@ class TestPreAggregateTableSelection extends QueryTest with BeforeAndAfterAll {
preAggTableValidator(df.queryExecution.analyzed, "maintable")
}
+ val timeSeries = TIMESERIES.toString
+
test("test PreAggregate table selection with timeseries and normal together") {
sql("drop table if exists maintabletime")
sql(
@@ -277,17 +281,26 @@ test("test PreAggregate table selection with timeseries and normal together") {
sql(
"create datamap agg0 on table maintabletime using 'preaggregate' as select dob,name from " +
"maintabletime group by dob,name")
- sql(
- "create datamap agg1 on table maintabletime using 'preaggregate' DMPROPERTIES ('timeseries" +
- ".eventTime'='dob', 'timeseries.hierarchy'='hour=1,day=1,month=1,year=1') as select dob," +
- "name from maintabletime group by dob,name")
- val df = sql("select timeseries(dob,'year') from maintabletime group by timeseries(dob,'year')")
+
+ sql(
+ s"""
+ | CREATE DATAMAP agg1_year ON TABLE maintabletime
+ | USING '$timeSeries'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='dob',
+ | 'YEAR_GRANULARITY'='1')
+ | AS SELECT dob, name FROM maintabletime
+ | GROUP BY dob,name
+ """.stripMargin)
+
+ val df = sql("SELECT timeseries(dob,'year') FROM maintabletime GROUP BY timeseries(dob,'year')")
preAggTableValidator(df.queryExecution.analyzed, "maintabletime_agg1_year")
+ sql("DROP TABLE IF EXISTS maintabletime")
}
test("test table selection when unsupported aggregate function is present") {
- sql("drop table if exists maintabletime")
+ sql("DROP TABLE IF EXISTS maintabletime")
sql(
"create table maintabletime(year int,month int,name string,salary int,dob string) stored" +
" by 'carbondata' tblproperties('sort_scope'='Global_sort','table_blocksize'='23'," +
@@ -302,6 +315,7 @@ test("test PreAggregate table selection with timeseries and normal together") {
override def afterAll: Unit = {
sql("drop table if exists mainTable")
sql("drop table if exists lineitem")
+ sql("DROP TABLE IF EXISTS maintabletime")
}
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/181c280b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/timeseries/TestTimeSeriesCreateTable.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/timeseries/TestTimeSeriesCreateTable.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/timeseries/TestTimeSeriesCreateTable.scala
index c9041fa..0ca7cb9 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/timeseries/TestTimeSeriesCreateTable.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/timeseries/TestTimeSeriesCreateTable.scala
@@ -17,14 +17,68 @@
package org.apache.carbondata.integration.spark.testsuite.timeseries
import org.apache.spark.sql.test.util.QueryTest
-import org.scalatest.{BeforeAndAfterAll, Ignore}
+import org.scalatest.BeforeAndAfterAll
+
+import org.apache.carbondata.core.metadata.schema.datamap.DataMapProvider.TIMESERIES
+import org.apache.carbondata.spark.exception.{MalformedDataMapCommandException, MalformedCarbonCommandException}
class TestTimeSeriesCreateTable extends QueryTest with BeforeAndAfterAll {
+ val timeSeries = TIMESERIES.toString
+
override def beforeAll: Unit = {
- sql("drop table if exists mainTable")
+ sql("DROP TABLE IF EXISTS mainTable")
sql("CREATE TABLE mainTable(dataTime timestamp, name string, city string, age int) STORED BY 'org.apache.carbondata.format'")
- sql("create datamap agg0 on table mainTable using 'preaggregate' DMPROPERTIES ('timeseries.eventTime'='dataTime', 'timeseries.hierarchy'='second=1,hour=1,day=1,month=1,year=1') as select dataTime, sum(age) from mainTable group by dataTime")
+ sql(
+ s"""
+ | CREATE DATAMAP agg0_second ON TABLE mainTable
+ | USING '$timeSeries'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='dataTime',
+ | 'SECOND_GRANULARITY'='1')
+ | AS SELECT dataTime, SUM(age) FROM mainTable
+ | GROUP BY dataTime
+ """.stripMargin)
+ sql(
+ s"""
+ | CREATE DATAMAP agg0_hour ON TABLE mainTable
+ | USING '$timeSeries'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='dataTime',
+ | 'HOUR_GRANULARITY'='1')
+ | AS SELECT dataTime, SUM(age) FROM mainTable
+ | GROUP BY dataTime
+ """.stripMargin)
+ sql(
+ s"""
+ | CREATE DATAMAP agg0_day ON TABLE mainTable
+ | USING '$timeSeries'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='dataTime',
+ | 'day_granularity'='1')
+ | AS SELECT dataTime, SUM(age) FROM mainTable
+ | GROUP BY dataTime
+ """.stripMargin)
+ sql(
+ s"""
+ | CREATE DATAMAP agg0_month ON TABLE mainTable
+ | USING '$timeSeries'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='dataTime',
+ | 'month_granularity'='1')
+ | AS SELECT dataTime, SUM(age) FROM mainTable
+ | GROUP BY dataTime
+ """.stripMargin)
+ sql(
+ s"""
+ | CREATE DATAMAP agg0_year ON TABLE mainTable
+ | USING '$timeSeries'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='dataTime',
+ | 'year_granularity'='1')
+ | AS SELECT dataTime, SUM(age) FROM mainTable
+ | GROUP BY dataTime
+ """.stripMargin)
}
test("test timeseries create table Zero") {
@@ -53,28 +107,29 @@ class TestTimeSeriesCreateTable extends QueryTest with BeforeAndAfterAll {
intercept[Exception] {
sql(
s"""
- | create datamap agg0 on table mainTable
- | using 'preaggregate'
+ | CREATE DATAMAP agg0_second ON TABLE mainTable
+ | USING '$timeSeries'
| DMPROPERTIES (
- | 'timeseries.eventTime'='dataTime',
- | 'timeseries.hierarchy'='sec=1,hour=1,day=1,month=1,year=1')
- | as select dataTime, sum(age) from mainTable
- | group by dataTime
- """.stripMargin)
+ | 'EVENT_TIME'='dataTime',
+ | 'SEC_GRANULARITY'='1')
+ | AS SELECT dataTime, SUM(age) FROM mainTable
+ | GROUP BY dataTime
+ """.stripMargin)
}
}
test("test timeseries create table Six") {
intercept[Exception] {
sql(
- """
- | create datamap agg0 on table mainTable
- | using 'preaggregate'
- | DMPROPERTIES ('timeseries.eventTime'='dataTime', 'timeseries.hierarchy'='hour=2')
- | as select dataTime, sum(age) from mainTable
- | group by dataTime
- """.stripMargin)
-
+ s"""
+ | CREATE DATAMAP agg0_second ON TABLE mainTable
+ | USING '$timeSeries'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='dataTime',
+ | 'SECOND_GRANULARITY'='2')
+ | AS SELECT dataTime, SUM(age) FROM mainTable
+ | GROUP BY dataTime
+ """.stripMargin)
}
}
@@ -82,24 +137,24 @@ class TestTimeSeriesCreateTable extends QueryTest with BeforeAndAfterAll {
intercept[Exception] {
sql(
s"""
- | create datamap agg0 on table mainTable
- | using 'preaggregate'
+ | CREATE DATAMAP agg0_second ON TABLE mainTable
+ | USING '$timeSeries'
| DMPROPERTIES (
- | 'timeseries.eventTime'='dataTime',
- | 'timeseries.hierarchy'='hour=1,day=1,year=1,month=1')
- | as select dataTime, sum(age) from mainTable
- | group by dataTime
- """.stripMargin)
+ | 'EVENT_TIME'='dataTime',
+ | 'SECOND_GRANULARITY'='1')
+ | AS SELECT dataTime, SUM(age) FROM mainTable
+ | GROUP BY dataTime
+ """.stripMargin)
sql(
s"""
- | create datamap agg0 on table mainTable
- | using 'preaggregate'
+ | CREATE DATAMAP agg0_second ON TABLE mainTable
+ | USING '$timeSeries'
| DMPROPERTIES (
- | 'timeseries.eventTime'='dataTime',
- | 'timeseries.hierarchy'='hour=1,day=1,year=1,month=1')
- | as select dataTime, sum(age) from mainTable
- | group by dataTime
- """.stripMargin)
+ | 'EVENT_TIME'='dataTime',
+ | 'SECOND_GRANULARITY'='1')
+ | AS SELECT dataTime, SUM(age) FROM mainTable
+ | GROUP BY dataTime
+ """.stripMargin)
}
}
@@ -107,12 +162,14 @@ class TestTimeSeriesCreateTable extends QueryTest with BeforeAndAfterAll {
intercept[Exception] {
sql(
s"""
- | create datamap agg0 on table mainTable
- | using 'preaggregate'
- | DMPROPERTIES ('timeseries.eventTime'='name', 'timeseries.hierarchy'='hour=1,day=1,year=1,month=1')
- | as select name, sum(age) from mainTable
- | group by name
- """.stripMargin)
+ | CREATE DATAMAP agg0_second ON TABLE mainTable
+ | USING '$timeSeries'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='name',
+ | 'SECOND_GRANULARITY'='1')
+ | AS SELECT dataTime, SUM(age) FROM mainTable
+ | GROUP BY dataTime
+ """.stripMargin)
}
}
@@ -120,18 +177,149 @@ class TestTimeSeriesCreateTable extends QueryTest with BeforeAndAfterAll {
intercept[Exception] {
sql(
s"""
- | create datamap agg0 on table mainTable
- | using 'preaggregate'
+ | CREATE DATAMAP agg0_second ON TABLE mainTable
+ | USING '$timeSeries'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='name',
+ | 'SECOND_GRANULARITY'='1')
+ | AS SELECT dataTime, SUM(age) FROM mainTable
+ | GROUP BY dataTime
+ """.stripMargin)
+ }
+ }
+
+ test("test timeseries create table: USING") {
+ val e: Exception = intercept[MalformedDataMapCommandException] {
+ sql(
+ """CREATE DATAMAP agg1 ON TABLE mainTable
+ | USING 'abc'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='dataTime',
+ | 'SECOND_GRANULARITY'='1')
+ | AS SELECT dataTime, SUM(age) FROM mainTable
+ | GROUP BY dataTime
+ """.stripMargin)
+ }
+ assert(e.getMessage.equals("Unknown data map type abc"))
+ }
+
+ test("test timeseries create table: USING and catch MalformedCarbonCommandException") {
+ val e: Exception = intercept[MalformedCarbonCommandException] {
+ sql(
+ """CREATE DATAMAP agg1 ON TABLE mainTable
+ | USING 'abc'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='dataTime',
+ | 'SECOND_GRANULARITY'='1')
+ | AS SELECT dataTime, SUM(age) FROM mainTable
+ | GROUP BY dataTime
+ """.stripMargin)
+ }
+ assert(e.getMessage.equals("Unknown data map type abc"))
+ }
+
+ test("test timeseries create table: Only one granularity level can be defined 1") {
+ val e: Exception = intercept[MalformedCarbonCommandException] {
+ sql(
+ s"""
+ | CREATE DATAMAP agg0_second ON TABLE mainTable
+ | USING '$timeSeries'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='dataTime',
+ | 'SECOND_GRANULARITY'='1',
+ | 'HOUR_GRANULARITY'='1',
+ | 'DAY_GRANULARITY'='1',
+ | 'MONTH_GRANULARITY'='1',
+ | 'YEAR_GRANULARITY'='1')
+ | AS SELECT dataTime, SUM(age) FROM mainTable
+ | GROUP BY dataTime
+ """.stripMargin)
+ }
+ assert(e.getMessage.equals("Only one granularity level can be defined"))
+ }
+
+ test("test timeseries create table: Only one granularity level can be defined 2") {
+ val e: Exception = intercept[MalformedDataMapCommandException] {
+ sql(
+ s"""
+ | CREATE DATAMAP agg0_second ON TABLE mainTable
+ | USING '$timeSeries'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='dataTime',
+ | 'SECOND_GRANULARITY'='1',
+ | 'HOUR_GRANULARITY'='1')
+ | AS SELECT dataTime, SUM(age) FROM mainTable
+ | GROUP BY dataTime
+ """.stripMargin)
+ }
+ assert(e.getMessage.equals("Only one granularity level can be defined"))
+ }
+
+ test("test timeseries create table: Only one granularity level can be defined 3") {
+ val e: Exception = intercept[MalformedDataMapCommandException] {
+ sql(
+ s"""
+ | CREATE DATAMAP agg0_second ON TABLE mainTable
+ | USING '$timeSeries'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='dataTime',
+ | 'DAY_GRANULARITY'='1',
+ | 'HOUR_GRANULARITY'='1')
+ | AS SELECT dataTime, SUM(age) FROM mainTable
+ | GROUP BY dataTime
+ """.stripMargin)
+ }
+ assert(e.getMessage.equals("Only one granularity level can be defined"))
+ }
+
+ test("test timeseries create table: Granularity only support 1") {
+ val e = intercept[MalformedDataMapCommandException] {
+ sql(
+ s"""
+ | CREATE DATAMAP agg0_second ON TABLE mainTable
+ | USING '$timeSeries'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='dataTime',
+ | 'DAY_GRANULARITY'='2')
+ | AS SELECT dataTime, SUM(age) FROM mainTable
+ | GROUP BY dataTime
+ """.stripMargin)
+ }
+ assert(e.getMessage.equals("Granularity only support 1"))
+ }
+
+ test("test timeseries create table: Granularity only support 1 and throw Exception") {
+ val e = intercept[MalformedCarbonCommandException] {
+ sql(
+ s"""
+ | CREATE DATAMAP agg0_second ON TABLE mainTable
+ | USING '$timeSeries'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='dataTime',
+ | 'HOUR_GRANULARITY'='2')
+ | AS SELECT dataTime, SUM(age) FROM mainTable
+ | GROUP BY dataTime
+ """.stripMargin)
+ }
+ assert(e.getMessage.equals("Granularity only support 1"))
+ }
+
+ test("test timeseries create table: timeSeries should define time granularity") {
+ val e = intercept[MalformedDataMapCommandException] {
+ sql(
+ s"""
+ | CREATE DATAMAP agg0_second ON TABLE mainTable
+ | USING '$timeSeries'
| DMPROPERTIES (
- | 'timeseries.eventTime'='dataTime',
- | 'timeseries.hierarchy'='hour=1,day=1,year=1,month=1')
- | as select name, sum(age) from mainTable
- | group by name
- """.stripMargin)
+ | 'EVENT_TIME'='dataTime')
+ | AS SELECT dataTime, SUM(age) FROM mainTable
+ | GROUP BY dataTime
+ """.stripMargin)
}
+ assert(e.getMessage.equals(s"$timeSeries should define time granularity"))
}
override def afterAll: Unit = {
- sql("drop table if exists mainTable")
+ sql("DROP TABLE IF EXISTS mainTable")
}
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/181c280b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/timeseries/TestTimeseriesCompaction.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/timeseries/TestTimeseriesCompaction.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/timeseries/TestTimeseriesCompaction.scala
index a410fe4..d66c402 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/timeseries/TestTimeseriesCompaction.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/timeseries/TestTimeseriesCompaction.scala
@@ -18,15 +18,18 @@ package org.apache.carbondata.integration.spark.testsuite.timeseries
import org.apache.spark.sql.test.util.QueryTest
import org.apache.spark.util.SparkUtil4Test
-import org.scalatest.{BeforeAndAfterAll, Ignore}
+import org.scalatest.BeforeAndAfterAll
import org.scalatest.Matchers._
import org.apache.carbondata.core.constants.CarbonCommonConstants
+import org.apache.carbondata.core.metadata.schema.datamap.DataMapProvider.TIMESERIES
import org.apache.carbondata.core.util.CarbonProperties
class TestTimeseriesCompaction extends QueryTest with BeforeAndAfterAll {
var isCompactionEnabled = false
+ val timeSeries = TIMESERIES.toString
+
override def beforeAll: Unit = {
SparkUtil4Test.createTaskMockUp(sqlContext)
CarbonProperties.getInstance()
@@ -38,7 +41,67 @@ class TestTimeseriesCompaction extends QueryTest with BeforeAndAfterAll {
.addProperty(CarbonCommonConstants.ENABLE_AUTO_LOAD_MERGE, "true")
sql("drop table if exists mainTable")
sql("CREATE TABLE mainTable(mytime timestamp, name string, age int) STORED BY 'org.apache.carbondata.format'")
- sql("create datamap agg0 on table mainTable using 'preaggregate' DMPROPERTIES ('timeseries.eventTime'='mytime', 'timeseries.hierarchy'='second=1,minute=1,hour=1,day=1,month=1,year=1') as select mytime, sum(age) from mainTable group by mytime")
+ sql(
+ s"""
+ | CREATE DATAMAP agg0_second ON TABLE mainTable
+ | USING '$timeSeries'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='mytime',
+ | 'SECOND_GRANULARITY'='1')
+ | AS SELECT mytime, SUM(age) FROM mainTable
+ | GROUP BY mytime
+ """.stripMargin)
+ sql(
+ s"""
+ | CREATE DATAMAP agg0_minute ON TABLE mainTable
+ | USING '$timeSeries'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='mytime',
+ | 'MINUTE_GRANULARITY'='1')
+ | AS SELECT mytime, SUM(age) FROM mainTable
+ | GROUP BY mytime
+ """.stripMargin)
+ sql(
+ s"""
+ | CREATE DATAMAP agg0_hour ON TABLE mainTable
+ | USING '$timeSeries'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='mytime',
+ | 'HOUR_GRANULARITY'='1')
+ | AS SELECT mytime, SUM(age) FROM mainTable
+ | GROUP BY mytime
+ """.stripMargin)
+ sql(
+ s"""
+ | CREATE DATAMAP agg0_day ON TABLE mainTable
+ | USING '$timeSeries'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='mytime',
+ | 'DAY_GRANULARITY'='1')
+ | AS SELECT mytime, SUM(age) FROM mainTable
+ | GROUP BY mytime
+ """.stripMargin)
+ sql(
+ s"""
+ | CREATE DATAMAP agg0_month ON TABLE mainTable
+ | USING '$timeSeries'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='mytime',
+ | 'MONTH_GRANULARITY'='1')
+ | AS SELECT mytime, SUM(age) FROM mainTable
+ | GROUP BY mytime
+ """.stripMargin)
+ sql(
+ s"""
+ | CREATE DATAMAP agg0_year ON TABLE mainTable
+ | USING '$timeSeries'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='mytime',
+ | 'YEAR_GRANULARITY'='1')
+ | AS SELECT mytime, SUM(age) FROM mainTable
+ | GROUP BY mytime
+ """.stripMargin)
+
sql(s"LOAD DATA LOCAL INPATH '$resourcesPath/timeseriestest.csv' into table mainTable")
sql(s"LOAD DATA LOCAL INPATH '$resourcesPath/timeseriestest.csv' into table mainTable")
sql(s"LOAD DATA LOCAL INPATH '$resourcesPath/timeseriestest.csv' into table mainTable")
http://git-wip-us.apache.org/repos/asf/carbondata/blob/181c280b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/timeseries/TestTimeseriesDataLoad.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/timeseries/TestTimeseriesDataLoad.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/timeseries/TestTimeseriesDataLoad.scala
index d25710c..8bcdfc9 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/timeseries/TestTimeseriesDataLoad.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/timeseries/TestTimeseriesDataLoad.scala
@@ -24,10 +24,13 @@ import org.apache.spark.util.SparkUtil4Test
import org.scalatest.{BeforeAndAfterAll, Ignore}
import org.apache.carbondata.core.constants.CarbonCommonConstants
+import org.apache.carbondata.core.metadata.schema.datamap.DataMapProvider.TIMESERIES
import org.apache.carbondata.core.util.CarbonProperties
class TestTimeseriesDataLoad extends QueryTest with BeforeAndAfterAll {
+ val timeSeries = TIMESERIES.toString
+
override def beforeAll: Unit = {
SparkUtil4Test.createTaskMockUp(sqlContext)
CarbonProperties.getInstance()
@@ -35,11 +38,131 @@ class TestTimeseriesDataLoad extends QueryTest with BeforeAndAfterAll {
sql("drop table if exists mainTable")
sql("drop table if exists table_03")
sql("CREATE TABLE mainTable(mytime timestamp, name string, age int) STORED BY 'org.apache.carbondata.format'")
- sql("create datamap agg0 on table mainTable using 'preaggregate' DMPROPERTIES ('timeseries.eventTime'='mytime', 'timeseries.hierarchy'='second=1,minute=1,hour=1,day=1,month=1,year=1') as select mytime, sum(age) from mainTable group by mytime")
+ sql(
+ s"""
+ | CREATE DATAMAP agg0_second ON TABLE mainTable
+ | USING '$timeSeries'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='mytime',
+ | 'SECOND_GRANULARITY'='1')
+ | AS SELECT mytime, SUM(age) FROM mainTable
+ | GROUP BY mytime
+ """.stripMargin)
+ sql(
+ s"""
+ | CREATE DATAMAP agg0_minute ON TABLE mainTable
+ | USING '$timeSeries'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='mytime',
+ | 'minute_granularity'='1')
+ | AS SELECT mytime, SUM(age) FROM mainTable
+ | GROUP BY mytime
+ """.stripMargin)
+ sql(
+ s"""
+ | CREATE DATAMAP agg0_hour ON TABLE mainTable
+ | USING '$timeSeries'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='mytime',
+ | 'HOUR_GRANULARITY'='1')
+ | AS SELECT mytime, SUM(age) FROM mainTable
+ | GROUP BY mytime
+ """.stripMargin)
+ sql(
+ s"""
+ | CREATE DATAMAP agg0_day ON TABLE mainTable
+ | USING '$timeSeries'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='mytime',
+ | 'DAY_GRANULARITY'='1')
+ | AS SELECT mytime, SUM(age) FROM mainTable
+ | GROUP BY mytime
+ """.stripMargin)
+ sql(
+ s"""
+ | CREATE DATAMAP agg0_month ON TABLE mainTable
+ | USING '$timeSeries'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='mytime',
+ | 'MONTH_GRANULARITY'='1')
+ | AS SELECT mytime, SUM(age) FROM mainTable
+ | GROUP BY mytime
+ """.stripMargin)
+ sql(
+ s"""
+ | CREATE DATAMAP agg0_year ON TABLE mainTable
+ | USING '$timeSeries'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='mytime',
+ | 'year_granularity'='1')
+ | AS SELECT mytime, SUM(age) FROM mainTable
+ | GROUP BY mytime
+ """.stripMargin)
+
sql(s"LOAD DATA LOCAL INPATH '$resourcesPath/timeseriestest.csv' into table mainTable")
sql("CREATE TABLE table_03 (imei string,age int,mac string,productdate timestamp,updatedate timestamp,gamePointId double,contractid double ) STORED BY 'org.apache.carbondata.format'")
sql(s"LOAD DATA inpath '$resourcesPath/data_sort.csv' INTO table table_03 options ('DELIMITER'=',', 'QUOTECHAR'='','FILEHEADER'='imei,age,mac,productdate,updatedate,gamePointId,contractid')")
- sql("create datamap ag1 on table table_03 using 'preaggregate' DMPROPERTIES ( 'timeseries.eventtime'='productdate','timeseries.hierarchy'='second=1,minute=1,hour=1,day=1,month=1,year=1')as select productdate,mac,sum(age) from table_03 group by productdate,mac")
+
+ sql(
+ s"""
+ | CREATE DATAMAP ag1_second ON TABLE table_03
+ | USING '$timeSeries'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='productdate',
+ | 'SECOND_GRANULARITY'='1')
+ | AS SELECT productdate,mac,SUM(age) FROM table_03
+ | GROUP BY productdate,mac
+ """.stripMargin)
+ sql(
+ s"""
+ | CREATE DATAMAP ag1_minute ON TABLE table_03
+ | USING '$timeSeries'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='productdate',
+ | 'minute_granularity'='1')
+ | AS SELECT productdate,mac,SUM(age) FROM table_03
+ | GROUP BY productdate,mac
+ """.stripMargin)
+ sql(
+ s"""
+ | CREATE DATAMAP ag1_hour ON TABLE table_03
+ | USING '$timeSeries'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='productdate',
+ | 'HOUR_GRANULARITY'='1')
+ | AS SELECT productdate,mac,SUM(age) FROM table_03
+ | GROUP BY productdate,mac
+ """.stripMargin)
+ sql(
+ s"""
+ | CREATE DATAMAP ag1_day ON TABLE table_03
+ | USING '$timeSeries'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='productdate',
+ | 'DAY_GRANULARITY'='1')
+ | AS SELECT productdate,mac,SUM(age) FROM table_03
+ | GROUP BY productdate,mac
+ """.stripMargin)
+ sql(
+ s"""
+ | CREATE DATAMAP ag1_month ON TABLE table_03
+ | USING '$timeSeries'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='productdate',
+ | 'month_granularity'='1')
+ | AS SELECT productdate,mac,SUM(age) FROM table_03
+ | GROUP BY productdate,mac
+ """.stripMargin)
+ sql(
+ s"""
+ | CREATE DATAMAP ag1_year ON TABLE table_03
+ | USING '$timeSeries'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='productdate',
+ | 'year_granularity'='1')
+ | AS SELECT productdate,mac,SUM(age) FROM table_03
+ | GROUP BY productdate,mac
+ """.stripMargin)
}
test("test Year level timeseries data validation1 ") {
@@ -93,12 +216,21 @@ class TestTimeseriesDataLoad extends QueryTest with BeforeAndAfterAll {
Row(Timestamp.valueOf("2016-02-23 01:02:50.0"),50)))
}
- test("test if timeseries load is successful on table creation") {
+ test("test if timeseries load is successful ON TABLE creation") {
sql("drop table if exists mainTable")
sql("CREATE TABLE mainTable(mytime timestamp, name string, age int) STORED BY 'org.apache.carbondata.format'")
sql(s"LOAD DATA LOCAL INPATH '$resourcesPath/timeseriestest.csv' into table mainTable")
- sql("create datamap agg0 on table mainTable using 'preaggregate' DMPROPERTIES ('timeseries.eventTime'='mytime', 'timeseries.hierarchy'='second=1,minute=1,hour=1,day=1,month=1,year=1') as select mytime, sum(age) from mainTable group by mytime")
- checkAnswer( sql("select * from maintable_agg0_second"),
+ sql(
+ s"""
+ | CREATE DATAMAP agg0_second ON TABLE mainTable
+ | USING '$timeSeries'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='mytime',
+ | 'SECOND_GRANULARITY'='1')
+ | AS SELECT mytime, SUM(age) FROM mainTable
+ | GROUP BY mytime
+ """.stripMargin)
+ checkAnswer( sql("select * FROM maintable_agg0_second"),
Seq(Row(Timestamp.valueOf("2016-02-23 01:01:30.0"),10),
Row(Timestamp.valueOf("2016-02-23 01:01:40.0"),20),
Row(Timestamp.valueOf("2016-02-23 01:01:50.0"),30),
http://git-wip-us.apache.org/repos/asf/carbondata/blob/181c280b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/timeseries/TestTimeseriesTableSelection.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/timeseries/TestTimeseriesTableSelection.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/timeseries/TestTimeseriesTableSelection.scala
index a9d3965..3065952 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/timeseries/TestTimeseriesTableSelection.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/timeseries/TestTimeseriesTableSelection.scala
@@ -24,14 +24,78 @@ import org.apache.spark.sql.test.util.QueryTest
import org.apache.spark.util.SparkUtil4Test
import org.scalatest.BeforeAndAfterAll
+import org.apache.carbondata.core.metadata.schema.datamap.DataMapProvider.TIMESERIES
+import org.apache.carbondata.spark.exception.MalformedCarbonCommandException
class TestTimeseriesTableSelection extends QueryTest with BeforeAndAfterAll {
+ val timeSeries = TIMESERIES.toString
+
override def beforeAll: Unit = {
SparkUtil4Test.createTaskMockUp(sqlContext)
sql("drop table if exists mainTable")
sql("CREATE TABLE mainTable(mytime timestamp, name string, age int) STORED BY 'org.apache.carbondata.format'")
- sql("create datamap agg0 on table mainTable using 'preaggregate' DMPROPERTIES ('timeseries.eventTime'='mytime', 'timeseries.hierarchy'='second=1,minute=1,hour=1,day=1,month=1,year=1') as select mytime, sum(age) from mainTable group by mytime")
+ sql(
+ s"""
+ | CREATE DATAMAP agg0_second ON TABLE mainTable
+ | USING '$timeSeries'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='mytime',
+ | 'SECOND_GRANULARITY'='1')
+ | AS SELECT mytime, SUM(age) FROM mainTable
+ | GROUP BY mytime
+ """.stripMargin)
+ sql(
+ s"""
+ | CREATE DATAMAP agg0_minute ON TABLE mainTable
+ | USING '$timeSeries'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='mytime',
+ | 'minute_granularity'='1')
+ | AS SELECT mytime, SUM(age) FROM mainTable
+ | GROUP BY mytime
+ """.stripMargin)
+ sql(
+ s"""
+ | CREATE DATAMAP agg0_hour ON TABLE mainTable
+ | USING '$timeSeries'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='mytime',
+ | 'HOUR_GRANULARITY'='1')
+ | AS SELECT mytime, SUM(age) FROM mainTable
+ | GROUP BY mytime
+ """.stripMargin)
+ sql(
+ s"""
+ | CREATE DATAMAP agg0_day ON TABLE mainTable
+ | USING '$timeSeries'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='mytime',
+ | 'DAY_GRANULARITY'='1')
+ | AS SELECT mytime, SUM(age) FROM mainTable
+ | GROUP BY mytime
+ """.stripMargin)
+ sql(
+ s"""
+ | CREATE DATAMAP agg0_month ON TABLE mainTable
+ | USING '$timeSeries'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='mytime',
+ | 'MONTH_GRANULARITY'='1')
+ | AS SELECT mytime, SUM(age) FROM mainTable
+ | GROUP BY mytime
+ """.stripMargin)
+ sql(
+ s"""
+ | CREATE DATAMAP agg0_year ON TABLE mainTable
+ | USING '$timeSeries'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='mytime',
+ | 'YEAR_GRANULARITY'='1')
+ | AS SELECT mytime, SUM(age) FROM mainTable
+ | GROUP BY mytime
+ """.stripMargin)
+
sql(s"LOAD DATA LOCAL INPATH '$resourcesPath/timeseriestest.csv' into table mainTable")
}
@@ -100,6 +164,54 @@ class TestTimeseriesTableSelection extends QueryTest with BeforeAndAfterAll {
preAggTableValidator(df.queryExecution.analyzed,"maintable")
}
+ test("test timeseries table selection 14: Granularity only support 1 and throw Exception") {
+ val e = intercept[MalformedCarbonCommandException] {
+ sql(
+ s"""
+ | CREATE DATAMAP agg3_second ON TABLE mainTable
+ | USING '$timeSeries'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='dataTime',
+ | 'HOUR_GRANULARITY'='2')
+ | AS SELECT dataTime, SUM(age) FROM mainTable
+ | GROUP BY dataTime
+ """.stripMargin)
+ }
+ assert(e.getMessage.contains("Granularity only support 1"))
+ }
+
+ test("test timeseries table selection 15: Granularity only support 1 and throw Exception") {
+ val e = intercept[MalformedCarbonCommandException] {
+ sql(
+ s"""
+ | CREATE DATAMAP agg3_second ON TABLE mainTable
+ | USING '$timeSeries'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='dataTime',
+ | 'HOUR_GRANULARITY'='1.5')
+ | AS SELECT dataTime, SUM(age) FROM mainTable
+ | GROUP BY dataTime
+ """.stripMargin)
+ }
+ assert(e.getMessage.contains("Granularity only support 1"))
+ }
+
+ test("test timeseries table selection 16: Granularity only support 1 and throw Exception") {
+ val e = intercept[MalformedCarbonCommandException] {
+ sql(
+ s"""
+ | CREATE DATAMAP agg3_second ON TABLE mainTable
+ | USING '$timeSeries'
+ | DMPROPERTIES (
+ | 'EVENT_TIME'='dataTime',
+ | 'HOUR_GRANULARITY'='-1')
+ | AS SELECT dataTime, SUM(age) FROM mainTable
+ | GROUP BY dataTime
+ """.stripMargin)
+ }
+ assert(e.getMessage.contains("Granularity only support 1"))
+ }
+
def preAggTableValidator(plan: LogicalPlan, actualTableName: String) : Unit ={
var isValidPlan = false
plan.transform {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/181c280b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/TestDataMapCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/TestDataMapCommand.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/TestDataMapCommand.scala
index a0ea317..0c38239 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/TestDataMapCommand.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/TestDataMapCommand.scala
@@ -27,6 +27,7 @@ import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.metadata.CarbonMetadata
import org.apache.carbondata.core.util.CarbonProperties
import org.apache.carbondata.core.util.path.CarbonTablePath
+import org.apache.carbondata.spark.exception.MalformedDataMapCommandException
class TestDataMapCommand extends QueryTest with BeforeAndAfterAll {
@@ -39,37 +40,42 @@ class TestDataMapCommand extends QueryTest with BeforeAndAfterAll {
sql("create table datamaptest (a string, b string, c string) stored by 'carbondata'")
}
-
- test("test datamap create") {
- sql("create datamap datamap1 on table datamaptest using 'new.class'")
- val table = CarbonMetadata.getInstance().getCarbonTable("default", "datamaptest")
- assert(table != null)
- val dataMapSchemaList = table.getTableInfo.getDataMapSchemaList
- assert(dataMapSchemaList.size() == 1)
- assert(dataMapSchemaList.get(0).getDataMapName.equals("datamap1"))
- assert(dataMapSchemaList.get(0).getClassName.equals("new.class"))
+ val newClass = "org.apache.spark.sql.CarbonSource"
+
+ test("test datamap create: don't support using class, only support short name") {
+ intercept[MalformedDataMapCommandException] {
+ sql(s"CREATE DATAMAP datamap1 ON TABLE datamaptest USING '$newClass'")
+ val table = CarbonMetadata.getInstance().getCarbonTable("default", "datamaptest")
+ assert(table != null)
+ val dataMapSchemaList = table.getTableInfo.getDataMapSchemaList
+ assert(dataMapSchemaList.size() == 1)
+ assert(dataMapSchemaList.get(0).getDataMapName.equals("datamap1"))
+ assert(dataMapSchemaList.get(0).getClassName.equals(newClass))
+ }
}
- test("test datamap create with dmproperties") {
- sql("create datamap datamap2 on table datamaptest using 'new.class' dmproperties('key'='value')")
- val table = CarbonMetadata.getInstance().getCarbonTable("default", "datamaptest")
- assert(table != null)
- val dataMapSchemaList = table.getTableInfo.getDataMapSchemaList
- assert(dataMapSchemaList.size() == 2)
- assert(dataMapSchemaList.get(1).getDataMapName.equals("datamap2"))
- assert(dataMapSchemaList.get(1).getClassName.equals("new.class"))
- assert(dataMapSchemaList.get(1).getProperties.get("key").equals("value"))
+ test("test datamap create with dmproperties: don't support using class") {
+ intercept[MalformedDataMapCommandException] {
+ sql(s"CREATE DATAMAP datamap2 ON TABLE datamaptest USING '$newClass' DMPROPERTIES('key'='value')")
+ val table = CarbonMetadata.getInstance().getCarbonTable("default", "datamaptest")
+ assert(table != null)
+ val dataMapSchemaList = table.getTableInfo.getDataMapSchemaList
+ assert(dataMapSchemaList.size() == 2)
+ assert(dataMapSchemaList.get(1).getDataMapName.equals("datamap2"))
+ assert(dataMapSchemaList.get(1).getClassName.equals(newClass))
+ assert(dataMapSchemaList.get(1).getProperties.get("key").equals("value"))
+ }
}
- test("test datamap create with existing name") {
- intercept[Exception] {
+ test("test datamap create with existing name: don't support using class") {
+ intercept[MalformedDataMapCommandException] {
sql(
- "create datamap datamap2 on table datamaptest using 'new.class' dmproperties('key'='value')")
+ s"CREATE DATAMAP datamap2 ON TABLE datamaptest USING '$newClass' DMPROPERTIES('key'='value')")
+ val table = CarbonMetadata.getInstance().getCarbonTable("default", "datamaptest")
+ assert(table != null)
+ val dataMapSchemaList = table.getTableInfo.getDataMapSchemaList
+ assert(dataMapSchemaList.size() == 2)
}
- val table = CarbonMetadata.getInstance().getCarbonTable("default", "datamaptest")
- assert(table != null)
- val dataMapSchemaList = table.getTableInfo.getDataMapSchemaList
- assert(dataMapSchemaList.size() == 2)
}
test("test datamap create with preagg") {
@@ -79,10 +85,10 @@ class TestDataMapCommand extends QueryTest with BeforeAndAfterAll {
val table = CarbonMetadata.getInstance().getCarbonTable("default", "datamaptest")
assert(table != null)
val dataMapSchemaList = table.getTableInfo.getDataMapSchemaList
- assert(dataMapSchemaList.size() == 3)
- assert(dataMapSchemaList.get(2).getDataMapName.equals("datamap3"))
- assert(dataMapSchemaList.get(2).getProperties.get("key").equals("value"))
- assert(dataMapSchemaList.get(2).getChildSchema.getTableName.equals("datamaptest_datamap3"))
+ assert(dataMapSchemaList.size() == 1)
+ assert(dataMapSchemaList.get(0).getDataMapName.equals("datamap3"))
+ assert(dataMapSchemaList.get(0).getProperties.get("key").equals("value"))
+ assert(dataMapSchemaList.get(0).getChildSchema.getTableName.equals("datamaptest_datamap3"))
}
test("check hivemetastore after drop datamap") {
@@ -110,7 +116,7 @@ class TestDataMapCommand extends QueryTest with BeforeAndAfterAll {
}
}
- test("drop the table having pre-aggregate"){
+ test("drop the table having pre-aggregate") {
try {
CarbonProperties.getInstance()
.addProperty(CarbonCommonConstants.ENABLE_HIVE_SCHEMA_META_STORE,
@@ -140,13 +146,24 @@ class TestDataMapCommand extends QueryTest with BeforeAndAfterAll {
test("test datamap create with preagg with duplicate name") {
intercept[Exception] {
sql(
- "create datamap datamap2 on table datamaptest using 'preaggregate' dmproperties('key'='value') as select count(a) from datamaptest")
-
+ s"""
+ | CREATE DATAMAP datamap2 ON TABLE datamaptest
+ | USING 'preaggregate'
+ | DMPROPERTIES('key'='value')
+ | AS SELECT COUNT(a) FROM datamaptest
+ """.stripMargin)
+ sql(
+ s"""
+ | CREATE DATAMAP datamap2 ON TABLE datamaptest
+ | USING 'preaggregate'
+ | DMPROPERTIES('key'='value')
+ | AS SELECT COUNT(a) FROM datamaptest
+ """.stripMargin)
}
val table = CarbonMetadata.getInstance().getCarbonTable("default", "datamaptest")
assert(table != null)
val dataMapSchemaList = table.getTableInfo.getDataMapSchemaList
- assert(dataMapSchemaList.size() == 3)
+ assert(dataMapSchemaList.size() == 2)
}
test("test datamap drop with preagg") {
@@ -157,25 +174,29 @@ class TestDataMapCommand extends QueryTest with BeforeAndAfterAll {
val table = CarbonMetadata.getInstance().getCarbonTable("default", "datamaptest")
assert(table != null)
val dataMapSchemaList = table.getTableInfo.getDataMapSchemaList
- assert(dataMapSchemaList.size() == 3)
+ assert(dataMapSchemaList.size() == 2)
}
- test("test show datamap without preaggregate") {
- sql("drop table if exists datamapshowtest")
- sql("create table datamapshowtest (a string, b string, c string) stored by 'carbondata'")
- sql("create datamap datamap1 on table datamapshowtest using 'new.class' dmproperties('key'='value')")
- sql("create datamap datamap2 on table datamapshowtest using 'new.class' dmproperties('key'='value')")
- checkExistence(sql("show datamap on table datamapshowtest"), true, "datamap1", "datamap2", "(NA)", "new.class")
+ test("test show datamap without preaggregate: don't support using class") {
+ intercept[MalformedDataMapCommandException] {
+ sql("drop table if exists datamapshowtest")
+ sql("create table datamapshowtest (a string, b string, c string) stored by 'carbondata'")
+ sql(s"CREATE DATAMAP datamap1 ON TABLE datamapshowtest USING '$newClass' DMPROPERTIES('key'='value')")
+ sql(s"CREATE DATAMAP datamap2 ON TABLE datamapshowtest USING '$newClass' DMPROPERTIES('key'='value')")
+ checkExistence(sql("SHOW DATAMAP ON TABLE datamapshowtest"), true, "datamap1", "datamap2", "(NA)", newClass)
+ }
}
- test("test show datamap with preaggregate") {
- sql("drop table if exists datamapshowtest")
- sql("create table datamapshowtest (a string, b string, c string) stored by 'carbondata'")
- sql("create datamap datamap1 on table datamapshowtest using 'preaggregate' as select count(a) from datamapshowtest")
- sql("create datamap datamap2 on table datamapshowtest using 'new.class' dmproperties('key'='value')")
- val frame = sql("show datamap on table datamapshowtest")
- assert(frame.collect().length == 2)
- checkExistence(frame, true, "datamap1", "datamap2", "(NA)", "new.class", "default.datamapshowtest_datamap1")
+ test("test show datamap with preaggregate: don't support using class") {
+ intercept[MalformedDataMapCommandException] {
+ sql("drop table if exists datamapshowtest")
+ sql("create table datamapshowtest (a string, b string, c string) stored by 'carbondata'")
+ sql("create datamap datamap1 on table datamapshowtest using 'preaggregate' as select count(a) from datamapshowtest")
+ sql(s"CREATE DATAMAP datamap2 ON TABLE datamapshowtest USING '$newClass' DMPROPERTIES('key'='value')")
+ val frame = sql("show datamap on table datamapshowtest")
+ assert(frame.collect().length == 2)
+ checkExistence(frame, true, "datamap1", "datamap2", "(NA)", newClass, "default.datamapshowtest_datamap1")
+ }
}
test("test show datamap with no datamap") {
@@ -184,15 +205,17 @@ class TestDataMapCommand extends QueryTest with BeforeAndAfterAll {
assert(sql("show datamap on table datamapshowtest").collect().length == 0)
}
- test("test show datamap after dropping datamap") {
- sql("drop table if exists datamapshowtest")
- sql("create table datamapshowtest (a string, b string, c string) stored by 'carbondata'")
- sql("create datamap datamap1 on table datamapshowtest using 'preaggregate' as select count(a) from datamapshowtest")
- sql("create datamap datamap2 on table datamapshowtest using 'new.class' dmproperties('key'='value')")
- sql("drop datamap datamap1 on table datamapshowtest")
- val frame = sql("show datamap on table datamapshowtest")
- assert(frame.collect().length == 1)
- checkExistence(frame, true, "datamap2", "(NA)", "new.class")
+ test("test show datamap after dropping datamap: don't support using class") {
+ intercept[MalformedDataMapCommandException] {
+ sql("drop table if exists datamapshowtest")
+ sql("create table datamapshowtest (a string, b string, c string) stored by 'carbondata'")
+ sql("create datamap datamap1 on table datamapshowtest using 'preaggregate' as select count(a) from datamapshowtest")
+ sql(s"CREATE DATAMAP datamap2 ON TABLE datamapshowtest USING '$newClass' DMPROPERTIES('key'='value')")
+ sql("drop datamap datamap1 on table datamapshowtest")
+ val frame = sql("show datamap on table datamapshowtest")
+ assert(frame.collect().length == 1)
+ checkExistence(frame, true, "datamap2", "(NA)", newClass)
+ }
}
test("test if preaggregate load is successfull for hivemetastore") {
@@ -217,7 +240,7 @@ class TestDataMapCommand extends QueryTest with BeforeAndAfterAll {
sql("CREATE TABLE uniqdata(CUST_ID int,CUST_NAME String,ACTIVE_EMUI_VERSION string,DOB timestamp,DOJ timestamp, BIGINT_COLUMN1 bigint,BIGINT_COLUMN2 bigint,DECIMAL_COLUMN1 decimal(30,10),DECIMAL_COLUMN2 decimal(36,10),Double_COLUMN1 double, Double_COLUMN2 double,INTEGER_COLUMN1 int) STORED BY 'org.apache.carbondata.format'")
sql("insert into uniqdata select 9000,'CUST_NAME_00000','ACTIVE_EMUI_VERSION_00000','1970-01-01 01:00:03','1970-01-01 02:00:03',123372036854,-223372036854,12345678901.1234000000,22345678901.1234000000,11234567489.7976000000,-11234567489.7976000000,1")
sql("create datamap uniqdata_agg on table uniqdata using 'preaggregate' as select min(DECIMAL_COLUMN1) from uniqdata group by DECIMAL_COLUMN1")
- checkAnswer(sql("select * from uniqdata_uniqdata_agg"),Seq(Row(12345678901.1234000000, 12345678901.1234000000)))
+ checkAnswer(sql("select * from uniqdata_uniqdata_agg"), Seq(Row(12345678901.1234000000, 12345678901.1234000000)))
sql("drop datamap if exists uniqdata_agg on table uniqdata")
}
@@ -229,16 +252,16 @@ class TestDataMapCommand extends QueryTest with BeforeAndAfterAll {
sql("insert into main select 10,11,'amy',12")
sql("insert into main select 10,11,'amy',14")
sql("create datamap preagg on table main " +
- "using 'preaggregate' " +
- s"dmproperties ('path'='$path') " +
- "as select name,avg(salary) from main group by name")
+ "using 'preaggregate' " +
+ s"dmproperties ('path'='$path') " +
+ "as select name,avg(salary) from main group by name")
assertResult(true)(new File(path).exists())
assertResult(true)(new File(s"${CarbonTablePath.getSegmentPath(path, "0")}")
- .list(new FilenameFilter {
- override def accept(dir: File, name: String): Boolean = {
- name.contains(CarbonCommonConstants.FACT_FILE_EXT)
- }
- }).length > 0)
+ .list(new FilenameFilter {
+ override def accept(dir: File, name: String): Boolean = {
+ name.contains(CarbonCommonConstants.FACT_FILE_EXT)
+ }
+ }).length > 0)
checkAnswer(sql("select name,avg(salary) from main group by name"), Row("amy", 13.0))
checkAnswer(sql("select * from main_preagg"), Row("amy", 26, 2))
sql("drop datamap preagg on table main")
http://git-wip-us.apache.org/repos/asf/carbondata/blob/181c280b/integration/spark-common/src/main/java/org/apache/carbondata/spark/exception/MalformedDataMapCommandException.java
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/java/org/apache/carbondata/spark/exception/MalformedDataMapCommandException.java b/integration/spark-common/src/main/java/org/apache/carbondata/spark/exception/MalformedDataMapCommandException.java
new file mode 100644
index 0000000..a05d8e6
--- /dev/null
+++ b/integration/spark-common/src/main/java/org/apache/carbondata/spark/exception/MalformedDataMapCommandException.java
@@ -0,0 +1,32 @@
+/*
+ * 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.spark.exception;
+
+/**
+ * Throw exception when using illegal argument
+ */
+public class MalformedDataMapCommandException extends MalformedCarbonCommandException {
+ /**
+ * default serial version ID.
+ */
+ private static final long serialVersionUID = 1L;
+
+ public MalformedDataMapCommandException(String msg) {
+ super(msg);
+ }
+}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/181c280b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonCreateDataMapCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonCreateDataMapCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonCreateDataMapCommand.scala
index 8e00635..c4d32b4 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonCreateDataMapCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonCreateDataMapCommand.scala
@@ -16,18 +16,15 @@
*/
package org.apache.spark.sql.execution.command.datamap
-import scala.collection.JavaConverters._
-
import org.apache.spark.sql._
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.execution.command._
-import org.apache.spark.sql.execution.command.preaaggregate.{CreatePreAggregateTableCommand, PreAggregateUtil}
+import org.apache.spark.sql.execution.command.preaaggregate.CreatePreAggregateTableCommand
import org.apache.spark.sql.execution.command.timeseries.TimeSeriesUtil
import org.apache.carbondata.common.logging.LogServiceFactory
-import org.apache.carbondata.core.constants.CarbonCommonConstants
-import org.apache.carbondata.core.metadata.schema.table.DataMapSchema
-import org.apache.carbondata.spark.exception.MalformedCarbonCommandException
+import org.apache.carbondata.core.metadata.schema.datamap.DataMapProvider._
+import org.apache.carbondata.spark.exception.{MalformedCarbonCommandException, MalformedDataMapCommandException}
/**
* Below command class will be used to create datamap on table
@@ -41,70 +38,63 @@ case class CarbonCreateDataMapCommand(
queryString: Option[String])
extends AtomicRunnableCommand {
- var createPreAggregateTableCommands: Seq[CreatePreAggregateTableCommand] = _
+ var createPreAggregateTableCommands: CreatePreAggregateTableCommand = _
override def processMetadata(sparkSession: SparkSession): Seq[Row] = {
// since streaming segment does not support building index and pre-aggregate yet,
// so streaming table does not support create datamap
val carbonTable =
- CarbonEnv.getCarbonTable(tableIdentifier.database, tableIdentifier.table)(sparkSession)
+ CarbonEnv.getCarbonTable(tableIdentifier.database, tableIdentifier.table)(sparkSession)
if (carbonTable.isStreamingTable) {
throw new MalformedCarbonCommandException("Streaming table does not support creating datamap")
}
val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)
- if (dmClassName.equals("org.apache.carbondata.datamap.AggregateDataMapHandler") ||
- dmClassName.equalsIgnoreCase("preaggregate")) {
- val timeHierarchyString = dmproperties.get(CarbonCommonConstants.TIMESERIES_HIERARCHY)
- createPreAggregateTableCommands = if (timeHierarchyString.isDefined) {
+
+ if (dmClassName.equalsIgnoreCase(PREAGGREGATE.toString) ||
+ dmClassName.equalsIgnoreCase(TIMESERIES.toString)) {
+ TimeSeriesUtil.validateTimeSeriesGranularity(dmproperties, dmClassName)
+ createPreAggregateTableCommands = if (dmClassName.equalsIgnoreCase(TIMESERIES.toString)) {
val details = TimeSeriesUtil
- .validateAndGetTimeSeriesHierarchyDetails(
- timeHierarchyString.get)
- val updatedDmProperties = dmproperties - CarbonCommonConstants.TIMESERIES_HIERARCHY
- details.map { f =>
- CreatePreAggregateTableCommand(dataMapName + '_' + f._1,
- tableIdentifier,
- dmClassName,
- updatedDmProperties,
- queryString.get,
- Some(f._1))
- }.toSeq
+ .getTimeSeriesGranularityDetails(dmproperties, dmClassName)
+ val updatedDmProperties = dmproperties - details._1
+ CreatePreAggregateTableCommand(dataMapName,
+ tableIdentifier,
+ dmClassName,
+ updatedDmProperties,
+ queryString.get,
+ Some(details._1))
} else {
- Seq(CreatePreAggregateTableCommand(
+ CreatePreAggregateTableCommand(
dataMapName,
tableIdentifier,
dmClassName,
dmproperties,
queryString.get
- ))
+ )
}
- createPreAggregateTableCommands.flatMap(_.processMetadata(sparkSession))
+ createPreAggregateTableCommands.processMetadata(sparkSession)
} else {
- val dataMapSchema = new DataMapSchema(dataMapName, dmClassName)
- dataMapSchema.setProperties(new java.util.HashMap[String, String](dmproperties.asJava))
- val dbName = CarbonEnv.getDatabaseName(tableIdentifier.database)(sparkSession)
- // upadting the parent table about dataschema
- PreAggregateUtil.updateMainTable(dbName, tableIdentifier.table, dataMapSchema, sparkSession)
+ throw new MalformedDataMapCommandException("Unknown data map type " + dmClassName)
}
- LOGGER.audit(s"DataMap $dataMapName successfully added to Table ${ tableIdentifier.table }")
+ LOGGER.audit(s"DataMap $dataMapName successfully added to Table ${tableIdentifier.table}")
Seq.empty
}
override def processData(sparkSession: SparkSession): Seq[Row] = {
- if (dmClassName.equals("org.apache.carbondata.datamap.AggregateDataMapHandler") ||
- dmClassName.equalsIgnoreCase("preaggregate")) {
- createPreAggregateTableCommands.flatMap(_.processData(sparkSession))
+ if (dmClassName.equalsIgnoreCase(PREAGGREGATE.toString) ||
+ dmClassName.equalsIgnoreCase(TIMESERIES.toString)) {
+ createPreAggregateTableCommands.processData(sparkSession)
} else {
- Seq.empty
+ throw new MalformedDataMapCommandException("Unknown data map type " + dmClassName)
}
}
override def undoMetadata(sparkSession: SparkSession, exception: Exception): Seq[Row] = {
- if (dmClassName.equals("org.apache.carbondata.datamap.AggregateDataMapHandler") ||
- dmClassName.equalsIgnoreCase("preaggregate")) {
- val timeHierarchyString = dmproperties.get(CarbonCommonConstants.TIMESERIES_HIERARCHY)
- createPreAggregateTableCommands.flatMap(_.undoMetadata(sparkSession, exception))
+ if (dmClassName.equalsIgnoreCase(PREAGGREGATE.toString) ||
+ dmClassName.equalsIgnoreCase(TIMESERIES.toString)) {
+ createPreAggregateTableCommands.undoMetadata(sparkSession, exception)
} else {
- Seq.empty
+ throw new MalformedDataMapCommandException("Unknown data map type " + dmClassName)
}
}
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/181c280b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/CreatePreAggregateTableCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/CreatePreAggregateTableCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/CreatePreAggregateTableCommand.scala
index a75a06f..dbbf90c 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/CreatePreAggregateTableCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/CreatePreAggregateTableCommand.scala
@@ -104,9 +104,9 @@ case class CreatePreAggregateTableCommand(
TimeSeriesUtil.validateTimeSeriesEventTime(dmProperties, parentTable)
TimeSeriesUtil.validateEventTimeColumnExitsInSelect(
fieldRelationMap,
- dmProperties.get(CarbonCommonConstants.TIMESERIES_EVENTTIME).get)
+ dmProperties.get(TimeSeriesUtil.TIMESERIES_EVENTTIME).get)
TimeSeriesUtil.updateTimeColumnSelect(fieldRelationMap,
- dmProperties.get(CarbonCommonConstants.TIMESERIES_EVENTTIME).get,
+ dmProperties.get(TimeSeriesUtil.TIMESERIES_EVENTTIME).get,
timeSeriesFunction.get)
}
tableModel.parentTable = Some(parentTable)
http://git-wip-us.apache.org/repos/asf/carbondata/blob/181c280b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/timeseries/TimeSeriesUtil.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/timeseries/TimeSeriesUtil.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/timeseries/TimeSeriesUtil.scala
index 4fe9df0..987d4fe 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/timeseries/TimeSeriesUtil.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/timeseries/TimeSeriesUtil.scala
@@ -18,32 +18,33 @@ package org.apache.spark.sql.execution.command.timeseries
import org.apache.spark.sql.execution.command.{DataMapField, Field}
-import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.metadata.datatype.DataTypes
+import org.apache.carbondata.core.metadata.schema.datamap.DataMapProvider.TIMESERIES
+import org.apache.carbondata.core.metadata.schema.datamap.Granularity
import org.apache.carbondata.core.metadata.schema.table.CarbonTable
import org.apache.carbondata.core.preagg.TimeSeriesUDF
-import org.apache.carbondata.spark.exception.MalformedCarbonCommandException
+import org.apache.carbondata.spark.exception.{MalformedCarbonCommandException, MalformedDataMapCommandException}
/**
* Utility class for time series to keep
*/
object TimeSeriesUtil {
+ val TIMESERIES_EVENTTIME = "event_time"
+
/**
* Below method will be used to validate whether column mentioned in time series
* is timestamp column or not
*
- * @param dmproperties
- * data map properties
- * @param parentTable
- * parent table
+ * @param dmproperties data map properties
+ * @param parentTable parent table
* @return whether time stamp column
*/
def validateTimeSeriesEventTime(dmproperties: Map[String, String],
parentTable: CarbonTable) {
- val eventTime = dmproperties.get(CarbonCommonConstants.TIMESERIES_EVENTTIME)
+ val eventTime = dmproperties.get(TIMESERIES_EVENTTIME)
if (!eventTime.isDefined) {
- throw new MalformedCarbonCommandException("Eventtime not defined in time series")
+ throw new MalformedCarbonCommandException("event_time not defined in time series")
} else {
val carbonColumn = parentTable.getColumnByName(parentTable.getTableName, eventTime.get)
if (carbonColumn.getDataType != DataTypes.TIMESTAMP) {
@@ -55,13 +56,79 @@ object TimeSeriesUtil {
}
/**
+ * validate TimeSeries Granularity
+ *
+ * @param dmProperties datamap properties
+ * @param dmClassName datamap class name
+ * @return whether find only one granularity
+ */
+ def validateTimeSeriesGranularity(
+ dmProperties: Map[String, String],
+ dmClassName: String): Boolean = {
+ var isFound = false
+
+ // 1. granularity only support one
+ for (granularity <- Granularity.values()) {
+ if (dmProperties.get(granularity.getName).isDefined) {
+ if (isFound) {
+ throw new MalformedDataMapCommandException(
+ s"Only one granularity level can be defined")
+ } else {
+ isFound = true
+ }
+ }
+ }
+
+ // 2. check whether timeseries and granularity match
+ if (isFound && !dmClassName.equalsIgnoreCase(TIMESERIES.toString)) {
+ throw new MalformedDataMapCommandException(
+ s"${TIMESERIES.toString} keyword missing")
+ } else if (!isFound && dmClassName.equalsIgnoreCase(TIMESERIES.toString)) {
+ throw new MalformedDataMapCommandException(
+ s"${TIMESERIES.toString} should define time granularity")
+ } else if (isFound) {
+ true
+ } else {
+ false
+ }
+ }
+
+ /**
+ * get TimeSeries Granularity key and value
+ * check the value
+ *
+ * TODO:we will support value not only equal to 1 in the future
+ *
+ * @param dmProperties datamap properties
+ * @param dmClassName datamap class name
+ * @return key and value tuple
+ */
+ def getTimeSeriesGranularityDetails(
+ dmProperties: Map[String, String],
+ dmClassName: String): (String, String) = {
+
+ val defaultValue = "1"
+ for (granularity <- Granularity.values()) {
+ if (dmProperties.get(granularity.getName).isDefined &&
+ dmProperties.get(granularity.getName).get.equalsIgnoreCase(defaultValue)) {
+ return (granularity.toString.toLowerCase, dmProperties.get(granularity.getName).get)
+ }
+ }
+
+ throw new MalformedDataMapCommandException(
+ s"Granularity only support $defaultValue")
+ }
+
+ /**
* Below method will be used to validate the hierarchy of time series and its value
* validation will be done whether hierarchy order is proper or not and hierarchy level
* value
+ * TODO: we should remove this method
*
* @param timeSeriesHierarchyDetails
* time series hierarchy string
*/
+ @deprecated
def validateAndGetTimeSeriesHierarchyDetails(timeSeriesHierarchyDetails: String): Array[
(String, String)] = {
val updatedtimeSeriesHierarchyDetails = timeSeriesHierarchyDetails.toLowerCase
http://git-wip-us.apache.org/repos/asf/carbondata/blob/181c280b/integration/spark2/src/test/scala/org/apache/spark/carbondata/CarbonDataSourceSuite.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/test/scala/org/apache/spark/carbondata/CarbonDataSourceSuite.scala b/integration/spark2/src/test/scala/org/apache/spark/carbondata/CarbonDataSourceSuite.scala
index 0f934cb..eb52910 100644
--- a/integration/spark2/src/test/scala/org/apache/spark/carbondata/CarbonDataSourceSuite.scala
+++ b/integration/spark2/src/test/scala/org/apache/spark/carbondata/CarbonDataSourceSuite.scala
@@ -260,4 +260,12 @@ class CarbonDataSourceSuite extends Spark2QueryTest with BeforeAndAfterAll {
sql("drop table if exists carbon_test")
assert(exception.contains("Table creation failed. Table name cannot contain blank space"))
}
+
+ test("test create table: using") {
+ sql("DROP TABLE IF EXISTS usingTable")
+ val e: Exception = intercept[ClassNotFoundException] {
+ sql("CREATE TABLE usingTable(name STRING) USING abc")
+ }
+ assert(e.getMessage.contains("Failed to find data source: abc"))
+ }
}