carbondata-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ravipes...@apache.org
Subject carbondata git commit: [CARBONDATA-2088][CARBONDATA-1516] Optimize syntax for creating timeseries pre-aggregate table
Date Tue, 30 Jan 2018 11:55:34 GMT
Repository: carbondata
Updated Branches:
  refs/heads/master b421c2468 -> 181c280b7


[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/master
Commit: 181c280b7d33ac5e4029bd935d6260b0fe79a2bf
Parents: b421c24
Author: xubo245 <601450868@qq.com>
Authored: Fri Jan 26 17:47:46 2018 +0800
Committer: ravipesala <ravi.pesala@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"))
+  }
 }


Mime
View raw message