Repository: carbondata Updated Branches: refs/heads/master 8bda43b05 -> 0e6fe6cae
http://git-wip-us.apache.org/repos/asf/carbondata/blob/0e6fe6ca/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 8cb6a81..ddacdb6 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 @@ -16,25 +16,39 @@ */ package org.apache.carbondata.integration.spark.testsuite.timeseries -import org.apache.spark.sql.CarbonDatasourceHadoopRelation +import java.sql.Timestamp + 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.apache.spark.sql.{CarbonDatasourceHadoopRelation, Row} import org.apache.spark.util.SparkUtil4Test import org.scalatest.BeforeAndAfterAll -import org.apache.carbondata.common.exceptions.sql.MalformedCarbonCommandException import org.apache.carbondata.core.metadata.schema.datamap.DataMapClassProvider.TIMESERIES +import org.apache.carbondata.core.constants.CarbonCommonConstants +import org.apache.carbondata.core.util.CarbonProperties class TestTimeseriesTableSelection extends QueryTest with BeforeAndAfterAll { val timeSeries = TIMESERIES.toString + var timestampFormat: String = _ override def beforeAll: Unit = { + timestampFormat = CarbonProperties.getInstance() + .getProperty(CarbonCommonConstants.CARBON_TIMESTAMP_FORMAT, + CarbonCommonConstants.CARBON_TIMESTAMP_DEFAULT_FORMAT) SparkUtil4Test.createTaskMockUp(sqlContext) + CarbonProperties.getInstance() + .addProperty(CarbonCommonConstants.CARBON_TIMESTAMP_FORMAT, + CarbonCommonConstants.CARBON_TIMESTAMP_DEFAULT_FORMAT) sql("drop table if exists mainTable") - sql("CREATE TABLE mainTable(mytime timestamp, name string, age int) STORED BY 'org.apache.carbondata.format'") + sql( + """ + | CREATE TABLE mainTable(mytime timestamp, name string, age int) + | STORED BY 'org.apache.carbondata.format' + """.stripMargin) sql( s""" | CREATE DATAMAP agg0_second ON TABLE mainTable @@ -99,124 +113,741 @@ class TestTimeseriesTableSelection extends QueryTest with BeforeAndAfterAll { sql(s"LOAD DATA LOCAL INPATH '$resourcesPath/timeseriestest.csv' into table mainTable") } - test("test PreAggregate table selection 1") { - val df = sql("select mytime from mainTable group by mytime") + test("test timeseries table selection 1") { + val df = sql("SELECT mytime FROM mainTable GROUP BY mytime") preAggTableValidator(df.queryExecution.analyzed, "maintable") } - test("test PreAggregate table selection 2") { - val df = sql("select timeseries(mytime,'hour') from mainTable group by timeseries(mytime,'hour')") + test("test timeseries table selection 2") { + val df = sql("SELECT TIMESERIES(mytime,'hour') FROM mainTable GROUP BY TIMESERIES(mytime,'hour')") preAggTableValidator(df.queryExecution.analyzed, "maintable_agg0_hour") } - test("test PreAggregate table selection 3") { - val df = sql("select timeseries(mytime,'milli') from mainTable group by timeseries(mytime,'milli')") - preAggTableValidator(df.queryExecution.analyzed, "maintable") + test("test timeseries table selection 3: No enum constant MILLI") { + val e = intercept[Exception] { + val df = sql( + """ + | SELECT TIMESERIES(mytime,'milli') + | FROM mainTable + | GROUP BY TIMESERIES(mytime,'milli') + """.stripMargin) + preAggTableValidator(df.queryExecution.analyzed, "maintable") + df.show() + } + assert(e.getMessage.contains( + "No enum constant org.apache.carbondata.core.preagg.TimeSeriesFunctionEnum.MILLI")) } - test("test PreAggregate table selection 4") { - val df = sql("select timeseries(mytime,'year') from mainTable group by timeseries(mytime,'year')") + test("test timeseries table selection 4") { + val df = sql("SELECT TIMESERIES(mytime,'year') FROM mainTable GROUP BY TIMESERIES(mytime,'year')") preAggTableValidator(df.queryExecution.analyzed,"maintable_agg0_year") } - test("test PreAggregate table selection 5") { - val df = sql("select timeseries(mytime,'day') from mainTable group by timeseries(mytime,'day')") + test("test timeseries table selection 5") { + val df = sql("SELECT TIMESERIES(mytime,'day') FROM mainTable GROUP BY TIMESERIES(mytime,'day')") preAggTableValidator(df.queryExecution.analyzed,"maintable_agg0_day") } - test("test PreAggregate table selection 6") { - val df = sql("select timeseries(mytime,'month') from mainTable group by timeseries(mytime,'month')") + test("test timeseries table selection 6") { + val df = sql("SELECT TIMESERIES(mytime,'month') FROM mainTable GROUP BY TIMESERIES(mytime,'month')") preAggTableValidator(df.queryExecution.analyzed,"maintable_agg0_month") } - test("test PreAggregate table selection 7") { - val df = sql("select timeseries(mytime,'minute') from mainTable group by timeseries(mytime,'minute')") + test("test timeseries table selection 7") { + val df = sql("SELECT TIMESERIES(mytime,'minute') FROM mainTable GROUP BY TIMESERIES(mytime,'minute')") preAggTableValidator(df.queryExecution.analyzed,"maintable_agg0_minute") } - test("test PreAggregate table selection 8") { - val df = sql("select timeseries(mytime,'second') from mainTable group by timeseries(mytime,'second')") + test("test timeseries table selection 8") { + val df = sql("SELECT TIMESERIES(mytime,'second') FROM mainTable GROUP BY TIMESERIES(mytime,'second')") preAggTableValidator(df.queryExecution.analyzed,"maintable_agg0_second") } - test("test PreAggregate table selection 9") { - val df = sql("select timeseries(mytime,'hour') from mainTable where timeseries(mytime,'hour')='x' group by timeseries(mytime,'hour')") - preAggTableValidator(df.queryExecution.analyzed,"maintable_agg0_hour") + test("test timeseries table selection 9") { + val df = sql( + """ + | SELECT TIMESERIES(mytime,'hour') + | FROM mainTable + | WHERE TIMESERIES(mytime,'hour')='x' + | GROUP BY TIMESERIES(mytime,'hour') + """.stripMargin) + preAggTableValidator(df.queryExecution.analyzed, "maintable_agg0_hour") } - test("test PreAggregate table selection 10") { - val df = sql("select timeseries(mytime,'hour') from mainTable where timeseries(mytime,'hour')='x' group by timeseries(mytime,'hour') order by timeseries(mytime,'hour')") - preAggTableValidator(df.queryExecution.analyzed,"maintable_agg0_hour") + test("test timeseries table selection 10") { + val df = sql( + """ + | SELECT TIMESERIES(mytime,'hour') + | FROM mainTable + | WHERE TIMESERIES(mytime,'hour')='x' + | GROUP BY TIMESERIES(mytime,'hour') + | ORDER BY TIMESERIES(mytime,'hour') + """.stripMargin) + preAggTableValidator(df.queryExecution.analyzed, "maintable_agg0_hour") + } + + test("test timeseries table selection 11") { + val df = sql( + """ + | SELECT TIMESERIES(mytime,'hour'),SUM(age) + | FROM mainTable + | WHERE TIMESERIES(mytime,'hour')='x' + | GROUP BY TIMESERIES(mytime,'hour') + | ORDER BY TIMESERIES(mytime,'hour') + """.stripMargin) + preAggTableValidator(df.queryExecution.analyzed, "maintable_agg0_hour") } - test("test PreAggregate table selection 11") { - val df = sql("select timeseries(mytime,'hour'),sum(age) from mainTable where timeseries(mytime,'hour')='x' group by timeseries(mytime,'hour') order by timeseries(mytime,'hour')") - preAggTableValidator(df.queryExecution.analyzed,"maintable_agg0_hour") + test("test timeseries table selection 12") { + val df = sql( + """ + | SELECT TIMESERIES(mytime,'hour') AS hourlevel,SUM(age) AS SUM + | FROM mainTable + | WHERE TIMESERIES(mytime,'hour')='x' + | GROUP BY TIMESERIES(mytime,'hour') + | ORDER BY TIMESERIES(mytime,'hour') + """.stripMargin) + preAggTableValidator(df.queryExecution.analyzed, "maintable_agg0_hour") } - test("test PreAggregate table selection 12") { - val df = sql("select timeseries(mytime,'hour')as hourlevel,sum(age) as sum from mainTable where timeseries(mytime,'hour')='x' group by timeseries(mytime,'hour') order by timeseries(mytime,'hour')") - preAggTableValidator(df.queryExecution.analyzed,"maintable_agg0_hour") + test("test timeseries table selection 13") { + val df = sql( + """ + | SELECT TIMESERIES(mytime,'hour')as hourlevel,SUM(age) AS SUM + | FROM mainTable + | WHERE TIMESERIES(mytime,'hour')='x' AND name='vishal' + | GROUP BY TIMESERIES(mytime,'hour') + | ORDER BY TIMESERIES(mytime,'hour') + """.stripMargin) + preAggTableValidator(df.queryExecution.analyzed, "maintable") } - test("test PreAggregate table selection 13") { - val df = sql("select timeseries(mytime,'hour')as hourlevel,sum(age) as sum from mainTable where timeseries(mytime,'hour')='x' and name='vishal' group by timeseries(mytime,'hour') order by timeseries(mytime,'hour')") - preAggTableValidator(df.queryExecution.analyzed,"maintable") + test("test timeseries table selection 14: TIMESERIES(mytime,'hour') match") { + val df = sql( + """ + | SELECT TIMESERIES(mytime,'hour') + | FROM mainTable + | WHERE TIMESERIES(mytime,'hour')='2016-02-23 09:00:00' + | GROUP BY TIMESERIES(mytime,'hour') + """.stripMargin) + preAggTableValidator(df.queryExecution.analyzed, "maintable_agg0_hour") + checkAnswer(df, Row(Timestamp.valueOf("2016-02-23 09:00:00.0"))) } - 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")) + test("test timeseries table selection 15: TIMESERIES(mytime,'hour') not match") { + val df = sql( + """ + | SELECT TIMESERIES(mytime,'hour') + | FROM mainTable + | WHERE TIMESERIES(mytime,'hour')='2016-02-23 09:01:00' + | GROUP BY TIMESERIES(mytime,'hour') + """.stripMargin) + preAggTableValidator(df.queryExecution.analyzed, "maintable_agg0_hour") + checkExistence(df, false, "2016-02-23 09:00:00", "2016-02-23 09:01:00") + } + + test("test timeseries table selection 16: TIMESERIES(mytime,'minute') match") { + checkExistence(sql("SELECT * FROM mainTable"), true, + "2016-02-23 09:01:30", "2016-02-23 09:02:40") + checkExistence(sql("SELECT * FROM mainTable"), false, + "2016-02-23 09:02:00", "2016-02-23 09:01:00") + val df = sql( + """ + |SELECT TIMESERIES(mytime,'minute') + |FROM mainTable + |GROUP BY TIMESERIES(mytime,'minute') + """.stripMargin) + preAggTableValidator(df.queryExecution.analyzed, "maintable_agg0_minute") + checkExistence(df, true, "2016-02-23 09:02:00", "2016-02-23 09:01:00") + checkAnswer(df, + Seq(Row(Timestamp.valueOf("2016-02-23 09:02:00.0")), + Row(Timestamp.valueOf("2016-02-23 09:01:00.0")))) + + val df2 = sql( + """ + | SELECT + | TIMESERIES(mytime,'minute')as minutelevel, + | SUM(age) AS SUM + | FROM mainTable + | WHERE TIMESERIES(mytime,'minute')='2016-02-23 09:01:00' + | GROUP BY TIMESERIES(mytime,'minute') + | ORDER BY TIMESERIES(mytime,'minute') + """.stripMargin) + preAggTableValidator(df2.queryExecution.analyzed, "maintable_agg0_minute") + checkAnswer(df2, Seq(Row(Timestamp.valueOf("2016-02-23 09:01:00"), 60))) + } + + test("test timeseries table selection 17: TIMESERIES(mytime,'minute') not match pre agg") { + val df = sql( + """ + | SELECT + | TIMESERIES(mytime,'minute')as minutelevel, + | SUM(age) AS SUM + | FROM mainTable + | WHERE TIMESERIES(mytime,'minute')='2016-02-23 09:01:00' AND name='vishal' + | GROUP BY TIMESERIES(mytime,'minute') + | ORDER BY TIMESERIES(mytime,'minute') + """.stripMargin) + checkAnswer(df, Seq(Row(Timestamp.valueOf("2016-02-23 09:01:00"), 10))) + preAggTableValidator(df.queryExecution.analyzed, "maintable") + } + + test("test timeseries table selection 18: select with many GROUP BY AND one filter") { + val df = sql( + """ + | SELECT + | TIMESERIES(mytime,'year') AS yearLevel, + | TIMESERIES(mytime,'month') AS monthLevel, + | TIMESERIES(mytime,'day') AS dayLevel, + | TIMESERIES(mytime,'hour') AS hourLevel, + | TIMESERIES(mytime,'minute') AS minuteLevel, + | TIMESERIES(mytime,'second') AS secondLevel, + | SUM(age) AS SUM + | FROM mainTable + | WHERE TIMESERIES(mytime,'minute')='2016-02-23 09:01:00' + | GROUP BY + | TIMESERIES(mytime,'year'), + | TIMESERIES(mytime,'month'), + | TIMESERIES(mytime,'day'), + | TIMESERIES(mytime,'hour'), + | TIMESERIES(mytime,'minute'), + | TIMESERIES(mytime,'second') + | ORDER BY + | TIMESERIES(mytime,'year'), + | TIMESERIES(mytime,'month'), + | TIMESERIES(mytime,'day'), + | TIMESERIES(mytime,'hour'), + | TIMESERIES(mytime,'minute'), + | TIMESERIES(mytime,'second') + """.stripMargin) + + checkExistence(df, true, + "2016-01-01 00:00:00", + "2016-02-01 00:00:00", + "2016-02-23 09:00:00", + "2016-02-23 09:01:00", + "2016-02-23 09:01:50", + "30" + ) + } + + test("test timeseries table selection 19: select with many GROUP BY AND many filter") { + val df = sql( + """ + | SELECT + | TIMESERIES(mytime,'year') AS yearLevel, + | TIMESERIES(mytime,'month') AS monthLevel, + | TIMESERIES(mytime,'day') AS dayLevel, + | TIMESERIES(mytime,'hour') AS hourLevel, + | TIMESERIES(mytime,'minute') AS minuteLevel, + | TIMESERIES(mytime,'second') AS secondLevel, + | SUM(age) AS SUM + | FROM mainTable + | WHERE + | TIMESERIES(mytime,'second')='2016-02-23 09:01:50' and + | TIMESERIES(mytime,'minute')='2016-02-23 09:01:00' and + | TIMESERIES(mytime,'hour')='2016-02-23 09:00:00' and + | TIMESERIES(mytime,'month')='2016-02-01 00:00:00' and + | TIMESERIES(mytime,'year')='2016-01-01 00:00:00' + | GROUP BY + | TIMESERIES(mytime,'year'), + | TIMESERIES(mytime,'month'), + | TIMESERIES(mytime,'day'), + | TIMESERIES(mytime,'hour'), + | TIMESERIES(mytime,'minute'), + | TIMESERIES(mytime,'second') + | ORDER BY + | TIMESERIES(mytime,'year'), + | TIMESERIES(mytime,'month'), + | TIMESERIES(mytime,'day'), + | TIMESERIES(mytime,'hour'), + | TIMESERIES(mytime,'minute'), + | TIMESERIES(mytime,'second') + """.stripMargin) + + checkExistence(df, true, + "2016-01-01 00:00:00", + "2016-02-01 00:00:00", + "2016-02-23 09:00:00", + "2016-02-23 09:01:00", + "2016-02-23 09:01:50", + "30" + ) + } + + test("test timeseries table selection 21: filter < AND >") { + val df = sql( + """ + | SELECT + | TIMESERIES(mytime,'minute') AS minuteLevel, + | SUM(age) AS SUM + | FROM mainTable + | WHERE TIMESERIES(mytime,'minute')<'2016-02-23 09:04:00' + | AND TIMESERIES(mytime,'minute')>'2016-02-23 09:01:00' + | GROUP BY + | TIMESERIES(mytime,'minute') + | ORDER BY + | TIMESERIES(mytime,'minute') + """.stripMargin) + + checkAnswer(df, Seq(Row(Timestamp.valueOf("2016-02-23 09:02:00"), 140))) + + preAggTableValidator(df.queryExecution.analyzed, "maintable_agg0_minute") + } + + test("test timeseries table selection 22: filter <= AND >=") { + val df = sql( + """ + | SELECT + | TIMESERIES(mytime,'minute') AS minuteLevel, + | SUM(age) AS SUM + | FROM mainTable + | WHERE TIMESERIES(mytime,'minute')<='2016-02-23 09:02:00' + | AND TIMESERIES(mytime,'minute')>='2016-02-23 09:01:00' + | GROUP BY + | TIMESERIES(mytime,'minute') + | ORDER BY + | TIMESERIES(mytime,'minute') + """.stripMargin) + + checkAnswer(df, + Seq(Row(Timestamp.valueOf("2016-02-23 09:01:00"), 60), + Row(Timestamp.valueOf("2016-02-23 09:02:00"), 140))) + + preAggTableValidator(df.queryExecution.analyzed, "maintable_agg0_minute") + } + + test("test timeseries table selection 23: filter < AND >=") { + val df = sql( + """ + | SELECT + | TIMESERIES(mytime,'minute') AS minuteLevel, + | SUM(age) AS SUM + | FROM mainTable + | WHERE TIMESERIES(mytime,'minute')<'2016-02-23 09:02:00' + | AND TIMESERIES(mytime,'minute')>='2016-02-23 09:01:00' + | GROUP BY + | TIMESERIES(mytime,'minute') + | ORDER BY + | TIMESERIES(mytime,'minute') + """.stripMargin) + + checkAnswer(df, Seq(Row(Timestamp.valueOf("2016-02-23 09:01:00"), 60))) + + preAggTableValidator(df.queryExecution.analyzed, "maintable_agg0_minute") + } + + test("test timeseries table selection 24: filter < AND >=") { + val df = sql( + """ + | SELECT + | TIMESERIES(mytime,'minute') AS minuteLevel, + | SUM(age) AS SUM + | FROM mainTable + | WHERE TIMESERIES(mytime,'minute')<'2016-02-23 09:01:00' + | AND TIMESERIES(mytime,'minute')>='2016-02-23 09:01:00' + | GROUP BY + | TIMESERIES(mytime,'minute') + | ORDER BY + | TIMESERIES(mytime,'minute') + """.stripMargin) + + checkAnswer(df, Seq.empty) + + preAggTableValidator(df.queryExecution.analyzed, "maintable_agg0_minute") + } + + test("test timeseries table selection 25: filter many column") { + val df = sql( + """ + | SELECT + | TIMESERIES(mytime,'minute') AS minuteLevel, + | SUM(age) AS SUM + | FROM mainTable + | WHERE + | TIMESERIES(mytime,'minute')<'2016-02-23 09:02:00' and + | TIMESERIES(mytime,'hour')>='2016-02-23 09:00:00' and + | name='vishal' + | GROUP BY + | TIMESERIES(mytime,'minute') + | ORDER BY + | TIMESERIES(mytime,'minute') + """.stripMargin) + + checkAnswer(df, Seq(Row(Timestamp.valueOf("2016-02-23 09:01:00"), 10))) + } + + test("test timeseries table selection 26: filter < AND >=, avg") { + val df = sql( + """ + | SELECT + | TIMESERIES(mytime,'minute') AS minuteLevel, + | avg(age) AS avg + | FROM mainTable + | WHERE TIMESERIES(mytime,'minute')<'2016-02-23 09:02:00' + | AND TIMESERIES(mytime,'minute')>='2016-02-23 09:01:00' + | GROUP BY + | TIMESERIES(mytime,'minute') + | ORDER BY + | TIMESERIES(mytime,'minute') + """.stripMargin) + + checkAnswer(df, Seq(Row(Timestamp.valueOf("2016-02-23 09:01:00"), 20.0))) + + } + + test("test timeseries table selection 27: filter < AND >=, max") { + val df = sql( + """ + | SELECT + | TIMESERIES(mytime,'second') AS secondLevel, + | max(age) AS maxValue + | FROM mainTable + | WHERE TIMESERIES(mytime,'second')<'2016-02-23 09:02:00' + | AND TIMESERIES(mytime,'second')>='2016-02-23 09:01:00' + | GROUP BY + | TIMESERIES(mytime,'second') + | ORDER BY + | TIMESERIES(mytime,'second') + """.stripMargin) + + checkAnswer(df, Seq(Row(Timestamp.valueOf("2016-02-23 09:01:30"), 10), + Row(Timestamp.valueOf("2016-02-23 09:01:40"), 20), + Row(Timestamp.valueOf("2016-02-23 09:01:50"), 30))) + } + + test("test timeseries table selection 28: filter < AND >=, min") { + val df = sql( + """ + | SELECT + | TIMESERIES(mytime,'second') AS secondLevel, + | min(age) AS minValue + | FROM mainTable + | WHERE TIMESERIES(mytime,'second')<'2016-02-23 09:02:00' + | AND TIMESERIES(mytime,'second')>='2016-02-23 09:01:00' + | GROUP BY + | TIMESERIES(mytime,'second') + | ORDER BY + | TIMESERIES(mytime,'second') + """.stripMargin) + + checkAnswer(df, Seq(Row(Timestamp.valueOf("2016-02-23 09:01:30"), 10), + Row(Timestamp.valueOf("2016-02-23 09:01:40"), 20), + Row(Timestamp.valueOf("2016-02-23 09:01:50"), 30))) + } + + test("test timeseries table selection 29: count, max, min, sum") { + dropDataMaps("maintable", "agg1_second", "agg1_minute") + sql( + s""" + | CREATE DATAMAP agg1_minute ON TABLE mainTable + | USING '$timeSeries' + | DMPROPERTIES ( + | 'event_time'='mytime', + | 'MINUTE_GRANULARITY'='1') + | AS SELECT mytime, SUM(age), count(age), max(age), min(age) + | FROM mainTable + | GROUP BY mytime + """.stripMargin) + + val df = sql( + """ + | SELECT + | TIMESERIES(mytime,'minute') AS minuteLevel, + | count(age) AS count + | FROM mainTable + | WHERE TIMESERIES(mytime,'minute')<'2016-02-23 09:02:00' + | AND TIMESERIES(mytime,'minute')>='2016-02-23 09:01:00' + | GROUP BY + | TIMESERIES(mytime,'minute') + | ORDER BY + | TIMESERIES(mytime,'minute') + """.stripMargin) + + checkAnswer(df, Seq(Row(Timestamp.valueOf("2016-02-23 09:01:00"), 3))) + + preAggTableValidator(df.queryExecution.analyzed, "maintable_agg1_minute") + + val df1 = sql( + """ + | SELECT + | TIMESERIES(mytime,'minute') AS minuteLevel, + | max(age) AS maxValue + | FROM mainTable + | WHERE TIMESERIES(mytime,'minute')<'2016-02-23 09:02:00' + | AND TIMESERIES(mytime,'minute')>='2016-02-23 09:01:00' + | GROUP BY + | TIMESERIES(mytime,'minute') + | ORDER BY + | TIMESERIES(mytime,'minute') + """.stripMargin) + + checkAnswer(df1, Seq(Row(Timestamp.valueOf("2016-02-23 09:01:00"), 30))) + + preAggTableValidator(df1.queryExecution.analyzed, "maintable_agg1_minute") + + val df2 = sql( + """ + | SELECT + | TIMESERIES(mytime,'minute') AS minuteLevel, + | min(age) AS minValue + | FROM mainTable + | WHERE TIMESERIES(mytime,'minute')<'2016-02-23 09:02:00' + | AND TIMESERIES(mytime,'minute')>='2016-02-23 09:01:00' + | GROUP BY + | TIMESERIES(mytime,'minute') + | ORDER BY + | TIMESERIES(mytime,'minute') + """.stripMargin) + + checkAnswer(df2, Seq(Row(Timestamp.valueOf("2016-02-23 09:01:00"), 10))) + + preAggTableValidator(df2.queryExecution.analyzed, "maintable_agg1_minute") + + dropDataMaps("maintable", "agg1_second", "agg1_minute") + + val df4 = sql( + """ + | SELECT + | TIMESERIES(mytime,'minute') AS minuteLevel, + | SUM(age) AS SUM + | FROM mainTable + | WHERE TIMESERIES(mytime,'minute')<'2016-02-23 09:02:00' + | AND TIMESERIES(mytime,'minute')>='2016-02-23 09:01:00' + | GROUP BY + | TIMESERIES(mytime,'minute') + | ORDER BY + | TIMESERIES(mytime,'minute') + """.stripMargin) + + checkAnswer(df4, Seq(Row(Timestamp.valueOf("2016-02-23 09:01:00"), 60))) + } + + test("test timeseries table selection 30: max, no create") { + val df1 = sql( + """ + | SELECT + | TIMESERIES(mytime,'minute') AS minuteLevel, + | max(age) AS maxValue + | FROM mainTable + | WHERE TIMESERIES(mytime,'minute')<'2016-02-23 09:02:00' + | AND TIMESERIES(mytime,'minute')>='2016-02-23 09:01:00' + | GROUP BY + | TIMESERIES(mytime,'minute') + | ORDER BY + | TIMESERIES(mytime,'minute') + """.stripMargin) + + checkAnswer(df1, Seq(Row(Timestamp.valueOf("2016-02-23 09:01:00"), 30))) + } + + test("test timeseries table selection 31: min, no create") { + val df1 = sql( + """ + | SELECT + | TIMESERIES(mytime,'minute') AS minuteLevel, + | min(age) AS minValue + | FROM mainTable + | WHERE TIMESERIES(mytime,'minute')<'2016-02-23 09:02:00' + | AND TIMESERIES(mytime,'minute')>='2016-02-23 09:01:00' + | GROUP BY + | TIMESERIES(mytime,'minute') + | ORDER BY + | TIMESERIES(mytime,'minute') + """.stripMargin) + + checkAnswer(df1, Seq(Row(Timestamp.valueOf("2016-02-23 09:01:00"), 10))) + } + + test("test timeseries table selection 32: filter < AND >=, min") { + val df = sql( + """ + | SELECT + | TIMESERIES(mytime,'second') AS secondLevel, + | min(age) AS minValue + | FROM mainTable + | WHERE TIMESERIES(mytime,'second')<'2016-02-23 09:02:00' + | AND TIMESERIES(mytime,'second')>='2016-02-23 09:01:00' + | GROUP BY + | TIMESERIES(mytime,'second') + | ORDER BY + | TIMESERIES(mytime,'second') + """.stripMargin) + + checkAnswer(df, Seq(Row(Timestamp.valueOf("2016-02-23 09:01:30"), 10), + Row(Timestamp.valueOf("2016-02-23 09:01:40"), 20), + Row(Timestamp.valueOf("2016-02-23 09:01:50"), 30))) + } + + test("test timeseries table selection 33: max") { + dropDataMaps("maintable", "agg1_minute") + sql( + s""" + | CREATE DATAMAP agg1_minute ON TABLE mainTable + | USING '$timeSeries' + | DMPROPERTIES ( + | 'event_time'='mytime', + | 'MINUTE_GRANULARITY'='1') + | AS SELECT mytime, SUM(age), count(age), max(age), min(age) + | FROM mainTable + | GROUP BY mytime + """.stripMargin) + + val df1 = sql( + """ + | SELECT + | TIMESERIES(mytime,'minute') AS minuteLevel, + | max(age) AS maxValue + | FROM mainTable + | WHERE TIMESERIES(mytime,'minute')<'2016-02-23 09:02:00' + | AND TIMESERIES(mytime,'minute')>='2016-02-23 09:01:00' + | GROUP BY + | TIMESERIES(mytime,'minute') + | ORDER BY + | TIMESERIES(mytime,'minute') + """.stripMargin) + + checkAnswer(df1, Seq(Row(Timestamp.valueOf("2016-02-23 09:01:00"), 30))) + + preAggTableValidator(df1.queryExecution.analyzed, "maintable_agg1_minute") + } + + test("test timeseries table selection 34: min") { + dropDataMaps("maintable", "agg1_second", "agg1_minute") + sql( + s""" + | CREATE DATAMAP agg1_minute ON TABLE mainTable + | USING '$timeSeries' + | DMPROPERTIES ( + | 'event_time'='mytime', + | 'MINUTE_GRANULARITY'='1') + | AS SELECT mytime, SUM(age), count(age), max(age), min(age) + | FROM mainTable + | GROUP BY mytime + """.stripMargin) + + val df1 = sql( + """ + | SELECT + | TIMESERIES(mytime,'minute') AS minuteLevel, + | min(age) AS minValue + | FROM mainTable + | WHERE TIMESERIES(mytime,'minute')<'2016-02-23 09:02:00' + | AND TIMESERIES(mytime,'minute')>='2016-02-23 09:01:00' + | GROUP BY + | TIMESERIES(mytime,'minute') + | ORDER BY + | TIMESERIES(mytime,'minute') + """.stripMargin) + + checkAnswer(df1, Seq(Row(Timestamp.valueOf("2016-02-23 09:01:00"), 10))) + + preAggTableValidator(df1.queryExecution.analyzed, "maintable_agg1_minute") + } + + test("test timeseries table selection 35: sum") { + dropDataMaps("maintable", "agg1_second", "agg1_minute") + sql( + s""" + | CREATE DATAMAP agg1_minute ON TABLE mainTable + | USING '$timeSeries' + | DMPROPERTIES ( + | 'event_time'='mytime', + | 'MINUTE_GRANULARITY'='1') + | AS SELECT mytime, SUM(age), count(age), max(age), min(age) + | FROM mainTable + | GROUP BY mytime + """.stripMargin) + + val df1 = sql( + """ + | SELECT + | TIMESERIES(mytime,'minute') AS minuteLevel, + | SUM(age) AS SUMValue + | FROM mainTable + | WHERE TIMESERIES(mytime,'minute')<'2016-02-23 09:02:00' + | AND TIMESERIES(mytime,'minute')>='2016-02-23 09:01:00' + | GROUP BY + | TIMESERIES(mytime,'minute') + | ORDER BY + | TIMESERIES(mytime,'minute') + """.stripMargin) + + checkAnswer(df1, Seq(Row(Timestamp.valueOf("2016-02-23 09:01:00"), 60))) + } + + test("test timeseries table selection 36: count") { + dropDataMaps("maintable", "agg1_second", "agg1_minute") + sql( + s""" + | CREATE DATAMAP agg1_minute ON TABLE mainTable + | USING '$timeSeries' + | DMPROPERTIES ( + | 'event_time'='mytime', + | 'MINUTE_GRANULARITY'='1') + | AS SELECT mytime, SUM(age), count(age), max(age), min(age) + | FROM mainTable + | GROUP BY mytime + """.stripMargin) + + val df1 = sql( + """ + | SELECT + | TIMESERIES(mytime,'minute') AS minuteLevel, + | count(age) AS countValue + | FROM mainTable + | WHERE TIMESERIES(mytime,'minute')<'2016-02-23 09:02:00' + | AND TIMESERIES(mytime,'minute')>='2016-02-23 09:01:00' + | GROUP BY + | TIMESERIES(mytime,'minute') + | ORDER BY + | TIMESERIES(mytime,'minute') + """.stripMargin) + + checkAnswer(df1, Seq(Row(Timestamp.valueOf("2016-02-23 09:01:00"), 3))) + + preAggTableValidator(df1.queryExecution.analyzed, "maintable_agg1_minute") + + } + + test("test timeseries table selection 37: avg") { + dropDataMaps("maintable", "agg1_second", "agg1_minute") + sql( + s""" + | CREATE DATAMAP agg1_minute ON TABLE mainTable + | USING '$timeSeries' + | DMPROPERTIES ( + | 'event_time'='mytime', + | 'MINUTE_GRANULARITY'='1') + | AS SELECT mytime, avg(age) + | FROM mainTable + | GROUP BY mytime + """.stripMargin) + + val df1 = sql( + """ + | SELECT + | TIMESERIES(mytime,'minute') AS minuteLevel, + | avg(age) AS avgValue + | FROM mainTable + | WHERE TIMESERIES(mytime,'minute')<'2016-02-23 09:02:00' + | AND TIMESERIES(mytime,'minute')>='2016-02-23 09:01:00' + | GROUP BY + | TIMESERIES(mytime,'minute') + | ORDER BY + | TIMESERIES(mytime,'minute') + """.stripMargin) + + checkAnswer(df1, Seq(Row(Timestamp.valueOf("2016-02-23 09:01:00"), 20))) + + preAggTableValidator(df1.queryExecution.analyzed, "maintable_agg1_minute") } def preAggTableValidator(plan: LogicalPlan, actualTableName: String) : Unit ={ var isValidPlan = false plan.transform { // first check if any preaTable1 scala function is applied it is present is in plan - // then call is from create preaTable1regate table class so no need to transform the query plan + // then call is FROM create preaTable1regate table class so no need to transform the query plan case ca:CarbonRelation => if (ca.isInstanceOf[CarbonDatasourceHadoopRelation]) { val relation = ca.asInstanceOf[CarbonDatasourceHadoopRelation] @@ -242,6 +873,9 @@ class TestTimeseriesTableSelection extends QueryTest with BeforeAndAfterAll { } override def afterAll: Unit = { + dropDataMaps("maintable", "agg0_second", "agg0_hour", "agg0_day", "agg0_month", "agg0_year") sql("drop table if exists mainTable") + CarbonProperties.getInstance() + .addProperty(CarbonCommonConstants.CARBON_TIMESTAMP_FORMAT, timestampFormat) } } http://git-wip-us.apache.org/repos/asf/carbondata/blob/0e6fe6ca/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/sortcolumns/TestSortColumns.scala ---------------------------------------------------------------------- diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/sortcolumns/TestSortColumns.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/sortcolumns/TestSortColumns.scala index adf8423..d9cb0e0 100644 --- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/sortcolumns/TestSortColumns.scala +++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/sortcolumns/TestSortColumns.scala @@ -33,16 +33,15 @@ class TestSortColumns extends QueryTest with BeforeAndAfterAll { CarbonCommonConstants.CARBON_TIMESTAMP_FORMAT, "dd-MM-yyyy") SparkUtil4Test.createTaskMockUp(sqlContext) - dropTable CarbonProperties.getInstance() .addProperty(CarbonCommonConstants.CARBON_TIMESTAMP_FORMAT, "dd-MM-yyyy") + dropTestTables sql("CREATE TABLE origintable1 (empno int, empname String, designation String, doj Timestamp, workgroupcategory int, workgroupcategoryname String, deptno int, deptname String, projectcode int, projectjoindate Timestamp, projectenddate Timestamp,attendance int,utilization int,salary int) STORED BY 'org.apache.carbondata.format'") sql(s"""LOAD DATA local inpath '$resourcesPath/data.csv' INTO TABLE origintable1 OPTIONS('DELIMITER'= ',', 'QUOTECHAR'= '\"', 'TIMESTAMPFORMAT'='dd-MM-yyyy')""") sql("CREATE TABLE tableOne(id int, name string, city string, age int) STORED BY 'org.apache.carbondata.format'") sql("CREATE TABLE tableTwo(id int, age int) STORED BY 'org.apache.carbondata.format'") sql(s"LOAD DATA LOCAL INPATH '$resourcesPath/measureinsertintotest.csv' into table tableOne") - } test("create table sort columns dictionary include - int") { @@ -382,12 +381,12 @@ class TestSortColumns extends QueryTest with BeforeAndAfterAll { } override def afterAll = { - dropTable + dropTestTables CarbonProperties.getInstance().addProperty( CarbonCommonConstants.CARBON_TIMESTAMP_FORMAT, CarbonCommonConstants.CARBON_TIMESTAMP_DEFAULT_FORMAT) } - def dropTable = { + def dropTestTables = { sql("drop table if exists sortint") sql("drop table if exists sortint1") sql("drop table if exists sortlong") http://git-wip-us.apache.org/repos/asf/carbondata/blob/0e6fe6ca/integration/spark-common/src/main/scala/org/apache/spark/sql/test/util/QueryTest.scala ---------------------------------------------------------------------- diff --git a/integration/spark-common/src/main/scala/org/apache/spark/sql/test/util/QueryTest.scala b/integration/spark-common/src/main/scala/org/apache/spark/sql/test/util/QueryTest.scala index 6e5630a..8f550fb 100644 --- a/integration/spark-common/src/main/scala/org/apache/spark/sql/test/util/QueryTest.scala +++ b/integration/spark-common/src/main/scala/org/apache/spark/sql/test/util/QueryTest.scala @@ -100,6 +100,16 @@ class QueryTest extends PlanTest { checkAnswer(df, expectedAnswer.collect()) } + protected def dropTable(tableName: String): Unit = { + sql(s"DROP TABLE IF EXISTS $tableName") + } + + protected def dropDataMaps(tableName: String, dataMapNames: String*): Unit = { + for (dataMapName <- dataMapNames) { + sql(s"DROP DATAMAP IF EXISTS $dataMapName ON TABLE $tableName") + } + } + def sql(sqlText: String): DataFrame = TestQueryExecutor.INSTANCE.sql(sqlText) val sqlContext: SQLContext = TestQueryExecutor.INSTANCE.sqlContext http://git-wip-us.apache.org/repos/asf/carbondata/blob/0e6fe6ca/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 1b621c1..e80579d 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 @@ -140,7 +140,7 @@ object TimeSeriesUtil { val splits = f.split("=") // checking hierarchy name is valid or not if (!TimeSeriesUDF.INSTANCE.TIMESERIES_FUNCTION.contains(splits(0).toLowerCase)) { - throw new MalformedCarbonCommandException(s"Not supported heirarchy type: ${ splits(0) }") + throw new MalformedCarbonCommandException(s"Not supported hierarchy type: ${ splits(0) }") } // validating hierarchy level is valid or not if (!splits(1).equals("1")) { @@ -154,7 +154,6 @@ object TimeSeriesUtil { // get the index of first hierarchy val indexOfFirstHierarchy = TimeSeriesUDF.INSTANCE.TIMESERIES_FUNCTION .indexOf(hierBuffer(0)._1.toLowerCase) - val index = 0 // now iterating through complete hierarchy to check any of the hierarchy index // is less than first one for (index <- 1 to hierBuffer.size - 1) { http://git-wip-us.apache.org/repos/asf/carbondata/blob/0e6fe6ca/integration/spark2/src/test/scala/org/apache/spark/util/CarbonCommandSuite.scala ---------------------------------------------------------------------- diff --git a/integration/spark2/src/test/scala/org/apache/spark/util/CarbonCommandSuite.scala b/integration/spark2/src/test/scala/org/apache/spark/util/CarbonCommandSuite.scala index 7dd6c7e..93777e9 100644 --- a/integration/spark2/src/test/scala/org/apache/spark/util/CarbonCommandSuite.scala +++ b/integration/spark2/src/test/scala/org/apache/spark/util/CarbonCommandSuite.scala @@ -190,7 +190,4 @@ class CarbonCommandSuite extends Spark2QueryTest with BeforeAndAfterAll { dropTable(tableName) } - protected def dropTable(tableName: String): Unit ={ - sql(s"DROP TABLE IF EXISTS $tableName") - } }
