You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@carbondata.apache.org by ja...@apache.org on 2018/08/30 08:22:21 UTC
carbondata git commit: [CARBONDATA-2835] [MVDataMap] Block MV datamap
on streaming table
Repository: carbondata
Updated Branches:
refs/heads/master 7fc0c6b9f -> 612552e61
[CARBONDATA-2835] [MVDataMap] Block MV datamap on streaming table
This PR block creating MV datamap on streaming table and also block setting streaming property for table which has MV datamap.
This closes #2627
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/612552e6
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/612552e6
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/612552e6
Branch: refs/heads/master
Commit: 612552e61853cd31c45a00b2407d16be5d78a024
Parents: 7fc0c6b
Author: ndwangsen <lu...@huawei.com>
Authored: Fri Aug 10 09:32:59 2018 +0800
Committer: Jacky Li <ja...@qq.com>
Committed: Thu Aug 30 16:22:06 2018 +0800
----------------------------------------------------------------------
.../apache/carbondata/mv/datamap/MVHelper.scala | 15 ++++
.../mv/rewrite/MVCreateTestCase.scala | 80 ++++++++++++++++++++
.../datamap/CarbonCreateDataMapCommand.scala | 8 --
.../sql/execution/strategy/DDLStrategy.scala | 5 ++
4 files changed, 100 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/612552e6/datamap/mv/core/src/main/scala/org/apache/carbondata/mv/datamap/MVHelper.scala
----------------------------------------------------------------------
diff --git a/datamap/mv/core/src/main/scala/org/apache/carbondata/mv/datamap/MVHelper.scala b/datamap/mv/core/src/main/scala/org/apache/carbondata/mv/datamap/MVHelper.scala
index 8c5fa7b..701f63d 100644
--- a/datamap/mv/core/src/main/scala/org/apache/carbondata/mv/datamap/MVHelper.scala
+++ b/datamap/mv/core/src/main/scala/org/apache/carbondata/mv/datamap/MVHelper.scala
@@ -33,6 +33,7 @@ import org.apache.spark.sql.execution.command.table.CarbonCreateTableCommand
import org.apache.spark.sql.execution.datasources.LogicalRelation
import org.apache.spark.sql.parser.CarbonSpark2SqlParser
+import org.apache.carbondata.common.exceptions.sql.MalformedCarbonCommandException
import org.apache.carbondata.core.datamap.DataMapStoreManager
import org.apache.carbondata.core.metadata.schema.datamap.DataMapClassProvider
import org.apache.carbondata.core.metadata.schema.table.{DataMapSchema, RelationIdentifier}
@@ -80,6 +81,20 @@ object MVHelper {
dmProperties.foreach(t => tableProperties.put(t._1, t._2))
val selectTables = getTables(logicalPlan)
+ selectTables.foreach { selectTable =>
+ val mainCarbonTable = try {
+ Some(CarbonEnv.getCarbonTable(selectTable.identifier.database,
+ selectTable.identifier.table)(sparkSession))
+ } catch {
+ // Exception handling if it's not a CarbonTable
+ case ex : Exception => None
+ }
+
+ if (!mainCarbonTable.isEmpty && mainCarbonTable.get.isStreamingSink) {
+ throw new MalformedCarbonCommandException(
+ s"Streaming table does not support creating MV datamap")
+ }
+ }
// TODO inherit the table properties like sort order, sort scope and block size from parent
// tables to mv datamap table
http://git-wip-us.apache.org/repos/asf/carbondata/blob/612552e6/datamap/mv/core/src/test/scala/org/apache/carbondata/mv/rewrite/MVCreateTestCase.scala
----------------------------------------------------------------------
diff --git a/datamap/mv/core/src/test/scala/org/apache/carbondata/mv/rewrite/MVCreateTestCase.scala b/datamap/mv/core/src/test/scala/org/apache/carbondata/mv/rewrite/MVCreateTestCase.scala
index fac80eb..3084ded 100644
--- a/datamap/mv/core/src/test/scala/org/apache/carbondata/mv/rewrite/MVCreateTestCase.scala
+++ b/datamap/mv/core/src/test/scala/org/apache/carbondata/mv/rewrite/MVCreateTestCase.scala
@@ -917,6 +917,83 @@ class MVCreateTestCase extends QueryTest with BeforeAndAfterAll {
sql("drop table if exists mvtable2")
}
+ test("test create datamap with streaming table") {
+ sql("drop datamap if exists dm_stream_test1")
+ sql("drop datamap if exists dm_stream_bloom")
+ sql("drop datamap if exists dm_stream_PreAggMax")
+ sql("drop table if exists fact_streaming_table1")
+ sql(
+ """
+ | CREATE TABLE fact_streaming_table1 (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'
+ | tblproperties('streaming'='true')
+ """.stripMargin)
+ sql(
+ s"""
+ | CREATE DATAMAP dm_stream_bloom ON TABLE fact_streaming_table1
+ | USING 'bloomfilter'
+ | DMProperties('INDEX_COLUMNS'='empname,deptname', 'BLOOM_SIZE'='640000')
+ """.stripMargin)
+
+ sql("create datamap dm_stream_PreAggMax on table fact_streaming_table1 using 'preaggregate' " +
+ "as select empname,max(salary) as max from fact_streaming_table1 group by empname")
+
+ val exception_tb_mv: Exception = intercept[Exception] {
+ sql("create datamap dm_stream_test1 using 'mv' as select empname, sum(utilization) from " +
+ "fact_streaming_table1 group by empname")
+ }
+ assert(exception_tb_mv.getMessage
+ .contains("Streaming table does not support creating MV datamap"))
+ }
+
+ test("test create datamap with streaming table join carbon table and join non-carbon table ") {
+ sql("drop datamap if exists dm_stream_test2")
+ sql("drop table if exists fact_streaming_table2")
+ sql("drop table if exists fact_table_parquet")
+ sql(
+ """
+ | CREATE TABLE fact_streaming_table2 (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'
+ | tblproperties('streaming'='true')
+ """.stripMargin)
+ sql(
+ """
+ | CREATE TABLE fact_table_parquet (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 AS parquet
+ """.stripMargin)
+
+ val exception_tb_mv2: Exception = intercept[Exception] {
+ sql("create datamap dm_stream_test2 using 'mv' as select t1.empname as c1, t2.designation, " +
+ "t2.empname as c2 from (fact_table1 t1 inner join fact_streaming_table2 t2 " +
+ "on (t1.empname = t2.empname)) inner join fact_table_parquet t3 " +
+ "on (t1.empname = t3.empname)")
+ }
+ assert(exception_tb_mv2.getMessage
+ .contains("Streaming table does not support creating MV datamap"))
+ }
+
+ test("test set streaming property of the table which has MV datamap") {
+ sql("drop datamap if exists dm_stream_test3")
+ sql("create datamap dm_stream_test3 using 'mv' as select empname, sum(utilization) from " +
+ "fact_table1 group by empname")
+ sql("rebuild datamap dm_stream_test3")
+ val exception_tb_mv3: Exception = intercept[Exception] {
+ sql("alter table fact_table1 set tblproperties('streaming'='true')")
+ }
+ assert(exception_tb_mv3.getMessage
+ .contains("The table which has MV datamap does not support set streaming property"))
+ sql("drop datamap if exists dm_stream_test3")
+ }
+
def verifyMVDataMap(logicalPlan: LogicalPlan, dataMapName: String): Boolean = {
val tables = logicalPlan collect {
case l: LogicalRelation => l.catalogTable.get
@@ -931,6 +1008,9 @@ class MVCreateTestCase extends QueryTest with BeforeAndAfterAll {
sql("drop table IF EXISTS fact_table4")
sql("drop table IF EXISTS fact_table5")
sql("drop table IF EXISTS fact_table6")
+ sql("drop table IF EXISTS fact_streaming_table1")
+ sql("drop table IF EXISTS fact_streaming_table2")
+ sql("drop table IF EXISTS fact_table_parquet")
}
override def afterAll {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/612552e6/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 17376a9..66f9e47 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
@@ -73,14 +73,6 @@ case class CarbonCreateDataMapCommand(
}
}
- if (mainTable != null &&
- mainTable.isStreamingSink &&
- !(dmProviderName.equalsIgnoreCase(DataMapClassProvider.PREAGGREGATE.toString)
- || dmProviderName.equalsIgnoreCase(DataMapClassProvider.TIMESERIES.toString))) {
- throw new MalformedCarbonCommandException(s"Streaming table does not support creating " +
- s"$dmProviderName datamap")
- }
-
if (mainTable !=null && CarbonUtil.getFormatVersion(mainTable) != ColumnarFormatVersion.V3) {
throw new MalformedCarbonCommandException(s"Unsupported operation on table with " +
s"V1 or V2 format data")
http://git-wip-us.apache.org/repos/asf/carbondata/blob/612552e6/integration/spark2/src/main/scala/org/apache/spark/sql/execution/strategy/DDLStrategy.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/strategy/DDLStrategy.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/strategy/DDLStrategy.scala
index f4fb90a..0c9490d 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/strategy/DDLStrategy.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/strategy/DDLStrategy.scala
@@ -35,6 +35,7 @@ import org.apache.spark.util.{CarbonReflectionUtils, FileUtils}
import org.apache.carbondata.common.exceptions.sql.MalformedCarbonCommandException
import org.apache.carbondata.common.logging.{LogService, LogServiceFactory}
+import org.apache.carbondata.core.metadata.schema.table.CarbonTable
import org.apache.carbondata.core.util.{CarbonProperties, ThreadLocalSessionInfo}
/**
@@ -279,6 +280,10 @@ class DDLStrategy(sparkSession: SparkSession) extends SparkStrategy {
throw new MalformedCarbonCommandException(
"Streaming property value is incorrect")
}
+ if (CarbonTable.hasMVDataMap(carbonTable)) {
+ throw new MalformedCarbonCommandException(
+ "The table which has MV datamap does not support set streaming property")
+ }
}
}
ExecutedCommandExec(CarbonAlterTableSetCommand(tableName, properties, isView)) :: Nil