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 2017/11/13 15:12:11 UTC
carbondata git commit: [CARBONDATA-1576] Added create datamap parser
and saved to schema file
Repository: carbondata
Updated Branches:
refs/heads/pre-aggregate df41cc66e -> c1eefee7c
[CARBONDATA-1576] Added create datamap parser and saved to schema file
User can create datamap using the following syntax.
CREATE DATAMAP agg_sales
ON TABLE sales
USING org.apache.carbondata.datamap.AggregateDataMapHandler
DMPROPERTIES (
'KEY’=’VALUE’
) AS
SELECT order_time, count(user_id) FROM sales GROUP BY order_time
In the above syntax DMPROPERTIES and AS QUERY are optional.
This closes #1481
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/c1eefee7
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/c1eefee7
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/c1eefee7
Branch: refs/heads/pre-aggregate
Commit: c1eefee7c848b1a020f77febcdac80c77aff6971
Parents: df41cc6
Author: ravipesala <ra...@gmail.com>
Authored: Sun Nov 12 17:08:09 2017 +0530
Committer: Jacky Li <ja...@qq.com>
Committed: Mon Nov 13 23:11:41 2017 +0800
----------------------------------------------------------------------
.../ThriftWrapperSchemaConverterImpl.java | 44 +++---
.../metadata/schema/table/DataMapSchema.java | 14 +-
.../core/metadata/schema/table/TableInfo.java | 2 +-
.../core/metadata/schema/table/TableSchema.java | 12 +-
format/src/main/thrift/schema.thrift | 10 +-
.../preaggregate/TestPreAggCreateCommand.scala | 112 +++++++--------
.../preaggregate/TestPreAggregateDrop.scala | 28 ++--
.../preaggregate/TestPreAggregateLoad.scala | 19 +--
.../testsuite/datamap/TestDataMapCommand.scala | 95 +++++++++++++
.../iud/DeleteCarbonTableTestCase.scala | 8 +-
.../iud/UpdateCarbonTableTestCase.scala | 8 +-
.../spark/sql/catalyst/CarbonDDLSqlParser.scala | 5 +-
.../datamap/CarbonCreateDataMapCommand.scala | 68 +++++++++
.../CreatePreAggregateTableCommand.scala | 141 ++++++++-----------
.../preaaggregate/PreAggregateListeners.scala | 8 +-
.../preaaggregate/PreAggregateUtil.scala | 8 +-
.../spark/sql/hive/CarbonFileMetastore.scala | 5 +-
.../sql/parser/CarbonSpark2SqlParser.scala | 24 +++-
.../spark/sql/parser/CarbonSparkSqlParser.scala | 31 +---
.../AlterTableValidationTestCase.scala | 6 +-
.../vectorreader/ChangeDataTypeTestCases.scala | 8 +-
.../vectorreader/DropColumnTestCases.scala | 6 +-
.../apache/spark/util/CarbonCommandSuite.scala | 6 +-
23 files changed, 411 insertions(+), 257 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/c1eefee7/core/src/main/java/org/apache/carbondata/core/metadata/converter/ThriftWrapperSchemaConverterImpl.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/metadata/converter/ThriftWrapperSchemaConverterImpl.java b/core/src/main/java/org/apache/carbondata/core/metadata/converter/ThriftWrapperSchemaConverterImpl.java
index 5ffc612..b914e06 100644
--- a/core/src/main/java/org/apache/carbondata/core/metadata/converter/ThriftWrapperSchemaConverterImpl.java
+++ b/core/src/main/java/org/apache/carbondata/core/metadata/converter/ThriftWrapperSchemaConverterImpl.java
@@ -322,17 +322,22 @@ public class ThriftWrapperSchemaConverterImpl implements SchemaConverter {
for (DataMapSchema wrapperChildSchema : wrapperChildSchemaList) {
org.apache.carbondata.format.DataMapSchema thriftChildSchema =
new org.apache.carbondata.format.DataMapSchema();
- org.apache.carbondata.format.RelationIdentifier relationIdentifier =
- new org.apache.carbondata.format.RelationIdentifier();
- relationIdentifier
- .setDatabaseName(wrapperChildSchema.getRelationIdentifier().getDatabaseName());
- relationIdentifier.setTableName(wrapperChildSchema.getRelationIdentifier().getTableName());
- relationIdentifier.setTableId(wrapperChildSchema.getRelationIdentifier().getTableId());
- thriftChildSchema.setRelationIdentifire(relationIdentifier);
+ if (wrapperChildSchema.getRelationIdentifier() != null) {
+ org.apache.carbondata.format.RelationIdentifier relationIdentifier =
+ new org.apache.carbondata.format.RelationIdentifier();
+ relationIdentifier.setDatabaseName(
+ wrapperChildSchema.getRelationIdentifier().getDatabaseName());
+ relationIdentifier.setTableName(wrapperChildSchema.getRelationIdentifier().getTableName());
+ relationIdentifier.setTableId(wrapperChildSchema.getRelationIdentifier().getTableId());
+ thriftChildSchema.setRelationIdentifire(relationIdentifier);
+ }
thriftChildSchema.setProperties(wrapperChildSchema.getProperties());
thriftChildSchema.setClassName(wrapperChildSchema.getClassName());
- thriftChildSchema.setChildTableSchema(
- fromWrapperToExternalTableSchema(wrapperChildSchema.getChildSchema()));
+ thriftChildSchema.setDataMapName(wrapperChildSchema.getDataMapName());
+ if (wrapperChildSchema.getChildSchema() != null) {
+ thriftChildSchema.setChildTableSchema(
+ fromWrapperToExternalTableSchema(wrapperChildSchema.getChildSchema()));
+ }
thriftChildSchemas.add(thriftChildSchema);
}
return thriftChildSchemas;
@@ -623,16 +628,19 @@ public class ThriftWrapperSchemaConverterImpl implements SchemaConverter {
@Override public DataMapSchema fromExternalToWrapperDataMapSchema(
org.apache.carbondata.format.DataMapSchema thriftDataMapSchema) {
- RelationIdentifier relationIdentifier =
- new RelationIdentifier(thriftDataMapSchema.getRelationIdentifire().getDatabaseName(),
- thriftDataMapSchema.getRelationIdentifire().getTableName(),
- thriftDataMapSchema.getRelationIdentifire().getTableId());
- DataMapSchema childSchema = new DataMapSchema(thriftDataMapSchema.getClassName());
+ DataMapSchema childSchema =
+ new DataMapSchema(thriftDataMapSchema.getDataMapName(), thriftDataMapSchema.getClassName());
childSchema.setProperties(thriftDataMapSchema.getProperties());
- childSchema.setChildSchema(
- fromExternalToWrapperTableSchema(thriftDataMapSchema.getChildTableSchema(),
- relationIdentifier.getTableName()));
- childSchema.setRelationIdentifier(relationIdentifier);
+ if (thriftDataMapSchema.getRelationIdentifire() != null) {
+ RelationIdentifier relationIdentifier =
+ new RelationIdentifier(thriftDataMapSchema.getRelationIdentifire().getDatabaseName(),
+ thriftDataMapSchema.getRelationIdentifire().getTableName(),
+ thriftDataMapSchema.getRelationIdentifire().getTableId());
+ childSchema.setRelationIdentifier(relationIdentifier);
+ childSchema.setChildSchema(
+ fromExternalToWrapperTableSchema(thriftDataMapSchema.getChildTableSchema(),
+ relationIdentifier.getTableName()));
+ }
return childSchema;
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/c1eefee7/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/DataMapSchema.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/DataMapSchema.java b/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/DataMapSchema.java
index 9c71e37..e0632d9 100644
--- a/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/DataMapSchema.java
+++ b/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/DataMapSchema.java
@@ -30,6 +30,8 @@ public class DataMapSchema implements Serializable, Writable {
private static final long serialVersionUID = 6577149126264181553L;
+ private String dataMapName;
+
private String className;
private RelationIdentifier relationIdentifier;
@@ -43,7 +45,11 @@ public class DataMapSchema implements Serializable, Writable {
*/
private Map<String, String> properties;
- public DataMapSchema(String className) {
+ public DataMapSchema() {
+ }
+
+ public DataMapSchema(String dataMapName, String className) {
+ this.dataMapName = dataMapName;
this.className = className;
}
@@ -75,7 +81,12 @@ public class DataMapSchema implements Serializable, Writable {
this.properties = properties;
}
+ public String getDataMapName() {
+ return dataMapName;
+ }
+
@Override public void write(DataOutput out) throws IOException {
+ out.writeUTF(dataMapName);
out.writeUTF(className);
boolean isRelationIdentifierExists = null != relationIdentifier;
out.writeBoolean(isRelationIdentifierExists);
@@ -99,6 +110,7 @@ public class DataMapSchema implements Serializable, Writable {
}
@Override public void readFields(DataInput in) throws IOException {
+ this.dataMapName = in.readUTF();
this.className = in.readUTF();
boolean isRelationIdnentifierExists = in.readBoolean();
if (isRelationIdnentifierExists) {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/c1eefee7/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/TableInfo.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/TableInfo.java b/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/TableInfo.java
index 44d8126..1d9e2ec 100644
--- a/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/TableInfo.java
+++ b/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/TableInfo.java
@@ -293,7 +293,7 @@ public class TableInfo implements Serializable, Writable {
if (isChildSchemaExists) {
short numberOfChildTable = in.readShort();
for (int i = 0; i < numberOfChildTable; i++) {
- DataMapSchema childSchema = new DataMapSchema(null);
+ DataMapSchema childSchema = new DataMapSchema();
childSchema.readFields(in);
dataMapSchemaList.add(childSchema);
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/c1eefee7/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/TableSchema.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/TableSchema.java b/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/TableSchema.java
index 5b013d1..714e0d8 100644
--- a/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/TableSchema.java
+++ b/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/TableSchema.java
@@ -255,21 +255,15 @@ public class TableSchema implements Serializable, Writable {
* Below method will be used to build child schema object which will be stored in
* parent table
*
- * @param className
- * @param databaseName
- * @param queryString
- * @param queryType
- *
- * @return datamap schema
*/
- public DataMapSchema buildChildSchema(String className, String databaseName, String queryString,
- String queryType) {
+ public DataMapSchema buildChildSchema(String dataMapName, String className, String databaseName,
+ String queryString, String queryType) {
RelationIdentifier relationIdentifier =
new RelationIdentifier(databaseName, tableName, tableId);
Map<String, String> properties = new HashMap<>();
properties.put("CHILD_SELECT QUERY", queryString);
properties.put("QUERYTYPE", queryType);
- DataMapSchema dataMapSchema = new DataMapSchema(className);
+ DataMapSchema dataMapSchema = new DataMapSchema(dataMapName, className);
dataMapSchema.setChildSchema(this);
dataMapSchema.setProperties(properties);
dataMapSchema.setRelationIdentifier(relationIdentifier);
http://git-wip-us.apache.org/repos/asf/carbondata/blob/c1eefee7/format/src/main/thrift/schema.thrift
----------------------------------------------------------------------
diff --git a/format/src/main/thrift/schema.thrift b/format/src/main/thrift/schema.thrift
index 4e9bb21..fc73cfb 100644
--- a/format/src/main/thrift/schema.thrift
+++ b/format/src/main/thrift/schema.thrift
@@ -187,15 +187,17 @@ struct ParentColumnTableRelation {
}
struct DataMapSchema {
+ // DataMap name
+ 1: required string dataMapName;
// class name
- 1: required string className;
+ 2: required string className;
// relation indentifier
- 2: optional RelationIdentifier relationIdentifire;
+ 3: optional RelationIdentifier relationIdentifire;
// in case of preaggregate it will be used to maintain the child schema
// which will be usefull in case of query and data load
- 3: optional TableSchema childTableSchema;
+ 4: optional TableSchema childTableSchema;
// to maintain properties like select query, query type like groupby, join
- 4: optional map<string, string> properties;
+ 5: optional map<string, string> properties;
}
struct TableInfo{
http://git-wip-us.apache.org/repos/asf/carbondata/blob/c1eefee7/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 6120e88..ce2d1d2 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
@@ -16,100 +16,100 @@ class TestPreAggCreateCommand extends QueryTest with BeforeAndAfterAll {
test("test pre agg create table One") {
- sql("create table preagg1 stored BY 'carbondata' tblproperties('parent'='PreAggMain') as select a,sum(b) from PreAggMain group by a")
- checkExistence(sql("DESCRIBE FORMATTED preagg1"), true, "preaggmain_a")
- checkExistence(sql("DESCRIBE FORMATTED preagg1"), true, "preaggmain_b_sum")
- sql("drop table preagg1")
+ sql("create datamap preagg1 on table PreAggMain using 'preaggregate' as select a,sum(b) from PreAggMain group by a")
+ checkExistence(sql("DESCRIBE FORMATTED PreAggMain_preagg1"), true, "preaggmain_a")
+ checkExistence(sql("DESCRIBE FORMATTED PreAggMain_preagg1"), true, "preaggmain_b_sum")
+ sql("drop table PreAggMain_preagg1")
}
test("test pre agg create table Two") {
- sql("create table preagg2 stored BY 'carbondata' tblproperties('parent'='PreAggMain') as select a as a1,sum(b) from PreAggMain group by a")
- checkExistence(sql("DESCRIBE FORMATTED preagg2"), true, "preaggmain_a")
- checkExistence(sql("DESCRIBE FORMATTED preagg2"), true, "preaggmain_b_sum")
- sql("drop table preagg2")
+ sql("create datamap preagg2 on table PreAggMain using 'preaggregate' as select a as a1,sum(b) from PreAggMain group by a")
+ checkExistence(sql("DESCRIBE FORMATTED PreAggMain_preagg2"), true, "preaggmain_a")
+ checkExistence(sql("DESCRIBE FORMATTED PreAggMain_preagg2"), true, "preaggmain_b_sum")
+ sql("drop table PreAggMain_preagg2")
}
test("test pre agg create table Three") {
- sql("create table preagg3 stored BY 'carbondata' tblproperties('parent'='PreAggMain') as select a,sum(b) as sum from PreAggMain group by a")
- checkExistence(sql("DESCRIBE FORMATTED preagg3"), true, "preaggmain_a")
- checkExistence(sql("DESCRIBE FORMATTED preagg3"), true, "preaggmain_b_sum")
- sql("drop table preagg3")
+ sql("create datamap preagg3 on table PreAggMain using 'preaggregate' as select a,sum(b) as sum from PreAggMain group by a")
+ checkExistence(sql("DESCRIBE FORMATTED PreAggMain_preagg3"), true, "preaggmain_a")
+ checkExistence(sql("DESCRIBE FORMATTED PreAggMain_preagg3"), true, "preaggmain_b_sum")
+ sql("drop table PreAggMain_preagg3")
}
test("test pre agg create table four") {
- sql("create table preagg4 stored BY 'carbondata' tblproperties('parent'='PreAggMain') as select a as a1,sum(b) as sum from PreAggMain group by a")
- checkExistence(sql("DESCRIBE FORMATTED preagg4"), true, "preaggmain_a")
- checkExistence(sql("DESCRIBE FORMATTED preagg4"), true, "preaggmain_b_sum")
- sql("drop table preagg4")
+ sql("create datamap preagg4 on table PreAggMain using 'preaggregate' as select a as a1,sum(b) as sum from PreAggMain group by a")
+ checkExistence(sql("DESCRIBE FORMATTED PreAggMain_preagg4"), true, "preaggmain_a")
+ checkExistence(sql("DESCRIBE FORMATTED PreAggMain_preagg4"), true, "preaggmain_b_sum")
+ sql("drop table PreAggMain_preagg4")
}
test("test pre agg create table five") {
- sql("create table preagg11 stored BY 'carbondata' tblproperties('parent'='PreAggMain1') as select a,sum(b) from PreAggMain1 group by a")
- checkExistence(sql("DESCRIBE FORMATTED preagg11"), true, "preaggmain1_a")
- checkExistence(sql("DESCRIBE FORMATTED preagg11"), true, "preaggmain1_b_sum")
- checkExistence(sql("DESCRIBE FORMATTED preagg11"), true, "DICTIONARY")
- sql("drop table preagg11")
+ sql("create datamap preagg11 on table PreAggMain1 using 'preaggregate'as select a,sum(b) from PreAggMain1 group by a")
+ checkExistence(sql("DESCRIBE FORMATTED PreAggMain1_preagg11"), true, "preaggmain1_a")
+ checkExistence(sql("DESCRIBE FORMATTED PreAggMain1_preagg11"), true, "preaggmain1_b_sum")
+ checkExistence(sql("DESCRIBE FORMATTED PreAggMain1_preagg11"), true, "DICTIONARY")
+ sql("drop table PreAggMain1_preagg11")
}
test("test pre agg create table six") {
- sql("create table preagg12 stored BY 'carbondata' tblproperties('parent'='PreAggMain1') as select a as a1,sum(b) from PreAggMain1 group by a")
- checkExistence(sql("DESCRIBE FORMATTED preagg12"), true, "preaggmain1_a")
- checkExistence(sql("DESCRIBE FORMATTED preagg12"), true, "preaggmain1_b_sum")
- checkExistence(sql("DESCRIBE FORMATTED preagg12"), true, "DICTIONARY")
- sql("drop table preagg12")
+ sql("create datamap preagg12 on table PreAggMain1 using 'preaggregate' as select a as a1,sum(b) from PreAggMain1 group by a")
+ checkExistence(sql("DESCRIBE FORMATTED PreAggMain1_preagg12"), true, "preaggmain1_a")
+ checkExistence(sql("DESCRIBE FORMATTED PreAggMain1_preagg12"), true, "preaggmain1_b_sum")
+ checkExistence(sql("DESCRIBE FORMATTED PreAggMain1_preagg12"), true, "DICTIONARY")
+ sql("drop table PreAggMain1_preagg12")
}
test("test pre agg create table seven") {
- sql("create table preagg13 stored BY 'carbondata' tblproperties('parent'='PreAggMain1') as select a,sum(b) as sum from PreAggMain1 group by a")
- checkExistence(sql("DESCRIBE FORMATTED preagg13"), true, "preaggmain1_a")
- checkExistence(sql("DESCRIBE FORMATTED preagg13"), true, "preaggmain1_b_sum")
- checkExistence(sql("DESCRIBE FORMATTED preagg13"), true, "DICTIONARY")
- sql("drop table preagg13")
+ sql("create datamap preagg13 on table PreAggMain1 using 'preaggregate' as select a,sum(b) as sum from PreAggMain1 group by a")
+ checkExistence(sql("DESCRIBE FORMATTED PreAggMain1_preagg13"), true, "preaggmain1_a")
+ checkExistence(sql("DESCRIBE FORMATTED PreAggMain1_preagg13"), true, "preaggmain1_b_sum")
+ checkExistence(sql("DESCRIBE FORMATTED PreAggMain1_preagg13"), true, "DICTIONARY")
+ sql("drop table PreAggMain1_preagg13")
}
test("test pre agg create table eight") {
- sql("create table preagg14 stored BY 'carbondata' tblproperties('parent'='PreAggMain1') as select a as a1,sum(b) as sum from PreAggMain1 group by a")
- checkExistence(sql("DESCRIBE FORMATTED preagg14"), true, "preaggmain1_a")
- checkExistence(sql("DESCRIBE FORMATTED preagg14"), true, "preaggmain1_b_sum")
- checkExistence(sql("DESCRIBE FORMATTED preagg14"), true, "DICTIONARY")
- sql("drop table preagg14")
+ sql("create datamap preagg14 on table PreAggMain1 using 'preaggregate' as select a as a1,sum(b) as sum from PreAggMain1 group by a")
+ checkExistence(sql("DESCRIBE FORMATTED PreAggMain1_preagg14"), true, "preaggmain1_a")
+ checkExistence(sql("DESCRIBE FORMATTED PreAggMain1_preagg14"), true, "preaggmain1_b_sum")
+ checkExistence(sql("DESCRIBE FORMATTED PreAggMain1_preagg14"), true, "DICTIONARY")
+ sql("drop table PreAggMain1_preagg14")
}
test("test pre agg create table nine") {
- sql("create table preagg15 stored BY 'carbondata' tblproperties('parent'='PreAggMain2') as select a,avg(b) from PreAggMain2 group by a")
- checkExistence(sql("DESCRIBE FORMATTED preagg15"), true, "preaggmain2_a")
- checkExistence(sql("DESCRIBE FORMATTED preagg15"), true, "preaggmain2_b_sum")
- checkExistence(sql("DESCRIBE FORMATTED preagg15"), true, "preaggmain2_b_count")
- sql("drop table preagg15")
+ sql("create datamap preagg15 on table PreAggMain2 using 'preaggregate' as select a,avg(b) from PreAggMain2 group by a")
+ checkExistence(sql("DESCRIBE FORMATTED PreAggMain2_preagg15"), true, "preaggmain2_a")
+ checkExistence(sql("DESCRIBE FORMATTED PreAggMain2_preagg15"), true, "preaggmain2_b_sum")
+ checkExistence(sql("DESCRIBE FORMATTED PreAggMain2_preagg15"), true, "preaggmain2_b_count")
+ sql("drop table PreAggMain2_preagg15")
}
test("test pre agg create table ten") {
- sql("create table preagg16 stored BY 'carbondata' tblproperties('parent'='PreAggMain2') as select a as a1,max(b) from PreAggMain2 group by a")
- checkExistence(sql("DESCRIBE FORMATTED preagg16"), true, "preaggmain2_a")
- checkExistence(sql("DESCRIBE FORMATTED preagg16"), true, "preaggmain2_b_max")
- sql("drop table preagg16")
+ sql("create datamap preagg16 on table PreAggMain2 using 'preaggregate' as select a as a1,max(b) from PreAggMain2 group by a")
+ checkExistence(sql("DESCRIBE FORMATTED PreAggMain2_preagg16"), true, "preaggmain2_a")
+ checkExistence(sql("DESCRIBE FORMATTED PreAggMain2_preagg16"), true, "preaggmain2_b_max")
+ sql("drop table PreAggMain2_preagg16")
}
test("test pre agg create table eleven") {
- sql("create table preagg17 stored BY 'carbondata' tblproperties('parent'='PreAggMain2') as select a,min(b) from PreAggMain2 group by a")
- checkExistence(sql("DESCRIBE FORMATTED preagg17"), true, "preaggmain2_a")
- checkExistence(sql("DESCRIBE FORMATTED preagg17"), true, "preaggmain2_b_min")
- sql("drop table preagg17")
+ sql("create datamap preagg17 on table PreAggMain2 using 'preaggregate' as select a,min(b) from PreAggMain2 group by a")
+ checkExistence(sql("DESCRIBE FORMATTED PreAggMain2_preagg17"), true, "preaggmain2_a")
+ checkExistence(sql("DESCRIBE FORMATTED PreAggMain2_preagg17"), true, "preaggmain2_b_min")
+ sql("drop table PreAggMain2_preagg17")
}
test("test pre agg create table twelve") {
- sql("create table preagg18 stored BY 'carbondata' tblproperties('parent'='PreAggMain2') as select a as a1,count(b) from PreAggMain2 group by a")
- checkExistence(sql("DESCRIBE FORMATTED preagg18"), true, "preaggmain2_a")
- checkExistence(sql("DESCRIBE FORMATTED preagg18"), true, "preaggmain2_b_count")
- sql("drop table preagg18")
+ sql("create datamap preagg18 on table PreAggMain2 using 'preaggregate' as select a as a1,count(b) from PreAggMain2 group by a")
+ checkExistence(sql("DESCRIBE FORMATTED PreAggMain2_preagg18"), true, "preaggmain2_a")
+ checkExistence(sql("DESCRIBE FORMATTED PreAggMain2_preagg18"), true, "preaggmain2_b_count")
+ sql("drop table PreAggMain2_preagg18")
}
test("test pre agg create table thirteen") {
try {
sql(
- "create table preagg19 stored BY 'carbondata' tblproperties('parent'='PreAggMain2') as select a as a1,count(distinct b) from PreAggMain2 group by a")
+ "create datamap preagg19 on table PreAggMain2 using 'preaggregate' as select a as a1,count(distinct b) from PreAggMain2 group by a")
assert(false)
} catch {
case _: Exception =>
@@ -120,7 +120,7 @@ class TestPreAggCreateCommand extends QueryTest with BeforeAndAfterAll {
test("test pre agg create table fourteen") {
try {
sql(
- "create table preagg20 stored BY 'carbondata' tblproperties('parent'='PreAggMain2') as select a as a1,sum(distinct b) from PreAggMain2 group by a")
+ "create datamap preagg20 on table PreAggMain2 using 'preaggregate' as select a as a1,sum(distinct b) from PreAggMain2 group by a")
assert(false)
} catch {
case _: Exception =>
@@ -131,7 +131,7 @@ class TestPreAggCreateCommand extends QueryTest with BeforeAndAfterAll {
test("test pre agg create table fifteen") {
try {
sql(
- "create table preagg21 stored BY 'carbondata' tblproperties('parent'='PreAggMain2') as select a as a1,sum(b) from PreAggMain2 where a='vishal' group by a")
+ "create datamap preagg21 on table PreAggMain2 using 'preaggregate' as select a as a1,sum(b) from PreAggMain2 where a='vishal' group by a")
assert(false)
} catch {
case _: Exception =>
http://git-wip-us.apache.org/repos/asf/carbondata/blob/c1eefee7/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateDrop.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateDrop.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateDrop.scala
index 4dad3e1..fee73a7 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateDrop.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateDrop.scala
@@ -24,44 +24,44 @@ class TestPreAggregateDrop extends QueryTest with BeforeAndAfterAll {
override def beforeAll {
sql("drop table if exists maintable")
- sql("drop table if exists preagg1")
- sql("drop table if exists preagg2")
+ sql("drop table if exists maintable_preagg1")
+ sql("drop table if exists maintable_preagg2")
sql("create table maintable (a string, b string, c string) stored by 'carbondata'")
}
test("create and drop preaggregate table") {
sql(
- "create table preagg1 stored BY 'carbondata' tblproperties('parent'='maintable') as select" +
+ "create datamap preagg1 on table maintable using 'preaggregate' as select" +
" a,sum(b) from maintable group by a")
- sql("drop table if exists preagg1")
- checkExistence(sql("show tables"), false, "preagg1")
+ sql("drop table if exists maintable_preagg1")
+ checkExistence(sql("show tables"), false, "maintable_preagg1")
}
test("dropping 1 aggregate table should not drop others") {
sql(
- "create table preagg1 stored BY 'carbondata' tblproperties('parent'='maintable') as select" +
+ "create datamap preagg1 on table maintable using 'preaggregate' as select" +
" a,sum(b) from maintable group by a")
sql(
- "create table preagg2 stored BY 'carbondata' tblproperties('parent'='maintable') as select" +
+ "create datamap preagg2 on table maintable using 'preaggregate' as select" +
" a,sum(c) from maintable group by a")
- sql("drop table if exists preagg2")
+ sql("drop table if exists maintable_preagg2")
val showTables = sql("show tables")
- checkExistence(showTables, false, "preagg2")
- checkExistence(showTables, true, "preagg1")
+ checkExistence(showTables, false, "maintable_preagg2")
+ checkExistence(showTables, true, "maintable_preagg1")
}
test("drop main table and check if preaggreagte is deleted") {
sql(
- "create table preagg2 stored BY 'carbondata' tblproperties('parent'='maintable') as select" +
+ "create datamap preagg2 on table maintable using 'preaggregate' as select" +
" a,sum(c) from maintable group by a")
sql("drop table if exists maintable")
- checkExistence(sql("show tables"), false, "preagg1", "maintable", "preagg2")
+ checkExistence(sql("show tables"), false, "maintable_preagg1", "maintable", "maintable_preagg2")
}
override def afterAll() {
sql("drop table if exists maintable")
- sql("drop table if exists preagg1")
- sql("drop table if exists preagg2")
+ sql("drop table if exists maintable_preagg1")
+ sql("drop table if exists maintable_preagg2")
}
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/c1eefee7/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateLoad.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateLoad.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateLoad.scala
index 1f576c5..5ac3534 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateLoad.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateLoad.scala
@@ -23,33 +23,28 @@ import org.scalatest.{BeforeAndAfterAll, Ignore}
@Ignore
class TestPreAggregateLoad extends QueryTest with BeforeAndAfterAll {
-
+
val testData = s"$resourcesPath/sample.csv"
-
+
override def beforeAll(): Unit = {
sql("DROP TABLE IF EXISTS maintable")
}
private def createAllAggregateTables(parentTableName: String): Unit = {
sql(
- s"""create table ${ parentTableName }_preagg_sum stored BY 'carbondata' tblproperties
- |('parent'='$parentTableName') as select id,sum(age) from $parentTableName group by id"""
+ s"""create datamap preagg_sum on table $parentTableName using 'preaggregate' as select id,sum(age) from $parentTableName group by id"""
.stripMargin)
sql(
- s"""create table ${ parentTableName }_preagg_avg stored BY 'carbondata' tblproperties
- |('parent'='$parentTableName') as select id,avg(age) from $parentTableName group by id"""
+ s"""create datamap preagg_avg on table $parentTableName using 'preaggregate' as select id,avg(age) from $parentTableName group by id"""
.stripMargin)
sql(
- s"""create table ${ parentTableName }_preagg_count stored BY 'carbondata' tblproperties
- |('parent'='$parentTableName') as select id,count(age) from $parentTableName group by id"""
+ s"""create datamap preagg_count on table $parentTableName using 'preaggregate' as select id,count(age) from $parentTableName group by id"""
.stripMargin)
sql(
- s"""create table ${ parentTableName }_preagg_min stored BY 'carbondata' tblproperties
- |('parent'='$parentTableName') as select id,min(age) from $parentTableName group by id"""
+ s"""create datamap preagg_min on table $parentTableName using 'preaggregate' as select id,min(age) from $parentTableName group by id"""
.stripMargin)
sql(
- s"""create table ${ parentTableName }_preagg_max stored BY 'carbondata' tblproperties
- |('parent'='$parentTableName') as select id,max(age) from $parentTableName group by id"""
+ s"""create datamap preagg_max on table $parentTableName using 'preaggregate' as select id,max(age) from $parentTableName group by id"""
.stripMargin)
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/c1eefee7/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
new file mode 100644
index 0000000..2512def
--- /dev/null
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/datamap/TestDataMapCommand.scala
@@ -0,0 +1,95 @@
+/*
+ * 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.testsuite.datamap
+
+import org.apache.spark.sql.test.util.QueryTest
+import org.scalatest.BeforeAndAfterAll
+
+import org.apache.carbondata.core.metadata.CarbonMetadata
+
+class TestDataMapCommand extends QueryTest with BeforeAndAfterAll {
+
+ override def beforeAll {
+ sql("drop table if exists datamaptest")
+ 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"))
+ }
+
+ 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 existing name") {
+ intercept[Exception] {
+ 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)
+ }
+
+ test("test datamap create with preagg") {
+ sql("drop table if exists datamap3")
+ sql(
+ "create datamap datamap3 on table datamaptest using 'preaggregate' dmproperties('key'='value') as select count(a) from datamaptest")
+ 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"))
+ }
+
+ test("test datamap create with preagg with duplicate name") {
+ intercept[Exception] {
+ sql("drop table if exists datamap2")
+ sql(
+ "create datamap datamap2 on table datamaptest using 'preaggregate' dmproperties('key'='value') as select count(a) from datamaptest")
+
+ }
+ val table = CarbonMetadata.getInstance().getCarbonTable("default_datamaptest")
+ assert(table != null)
+ val dataMapSchemaList = table.getTableInfo.getDataMapSchemaList
+ assert(dataMapSchemaList.size() == 3)
+ }
+
+
+ override def afterAll {
+ sql("drop table if exists datamaptest")
+ }
+}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/c1eefee7/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/DeleteCarbonTableTestCase.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/DeleteCarbonTableTestCase.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/DeleteCarbonTableTestCase.scala
index a2bd6aa..f7e93af 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/DeleteCarbonTableTestCase.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/DeleteCarbonTableTestCase.scala
@@ -131,17 +131,17 @@ class DeleteCarbonTableTestCase extends QueryTest with BeforeAndAfterAll {
test("test if delete is unsupported for pre-aggregate tables") {
sql("drop table if exists preaggMain")
- sql("drop table if exists preagg1")
+ sql("drop table if exists preaggmain_preagg1")
sql("create table preaggMain (a string, b string, c string) stored by 'carbondata'")
- sql("create table preagg1 stored BY 'carbondata' tblproperties('parent'='PreAggMain') as select a,sum(b) from PreAggMain group by a")
+ sql("create datamap preagg1 on table PreAggMain USING 'preaggregate' as select a,sum(b) from PreAggMain group by a")
intercept[RuntimeException] {
sql("delete from preaggmain where a = 'abc'").show()
}.getMessage.contains("Delete operation is not supported for tables")
intercept[RuntimeException] {
- sql("delete from preagg1 where preaggmain_a = 'abc'").show()
+ sql("delete from preaggmain_preagg1 where preaggmain_a = 'abc'").show()
}.getMessage.contains("Delete operation is not supported for pre-aggregate table")
sql("drop table if exists preaggMain")
- sql("drop table if exists preagg1")
+ sql("drop table if exists preaggmain_preagg1")
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/c1eefee7/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/UpdateCarbonTableTestCase.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/UpdateCarbonTableTestCase.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/UpdateCarbonTableTestCase.scala
index 4c43ec0..fd5f144 100644
--- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/UpdateCarbonTableTestCase.scala
+++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/spark/testsuite/iud/UpdateCarbonTableTestCase.scala
@@ -515,17 +515,17 @@ class UpdateCarbonTableTestCase extends QueryTest with BeforeAndAfterAll {
test("test if update is unsupported for pre-aggregate tables") {
sql("drop table if exists preaggMain")
- sql("drop table if exists preagg1")
+ sql("drop table if exists preaggMain_preagg1")
sql("create table preaggMain (a string, b string, c string) stored by 'carbondata'")
- sql("create table preagg1 stored BY 'carbondata' tblproperties('parent'='PreAggMain') as select a,sum(b) from PreAggMain group by a")
+ sql("create datamap preagg1 on table PreAggMain using 'preaggregate' as select a,sum(b) from PreAggMain group by a")
intercept[RuntimeException] {
sql("update preaggmain set (a)=('a')").show
}.getMessage.contains("Update operation is not supported for tables")
intercept[RuntimeException] {
- sql("update preagg1 set (a)=('a')").show
+ sql("update preaggMain_preagg1 set (a)=('a')").show
}.getMessage.contains("Update operation is not supported for pre-aggregate table")
sql("drop table if exists preaggMain")
- sql("drop table if exists preagg1")
+ sql("drop table if exists preaggMain_preagg1")
}
override def afterAll {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/c1eefee7/integration/spark-common/src/main/scala/org/apache/spark/sql/catalyst/CarbonDDLSqlParser.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/spark/sql/catalyst/CarbonDDLSqlParser.scala b/integration/spark-common/src/main/scala/org/apache/spark/sql/catalyst/CarbonDDLSqlParser.scala
index bb80bce..42447da 100644
--- a/integration/spark-common/src/main/scala/org/apache/spark/sql/catalyst/CarbonDDLSqlParser.scala
+++ b/integration/spark-common/src/main/scala/org/apache/spark/sql/catalyst/CarbonDDLSqlParser.scala
@@ -170,6 +170,9 @@ abstract class CarbonDDLSqlParser extends AbstractCarbonSparkSQLParser {
protected val CHANGE = carbonKeyWord("CHANGE")
protected val TBLPROPERTIES = carbonKeyWord("TBLPROPERTIES")
protected val ID = carbonKeyWord("ID")
+ protected val DATAMAP = carbonKeyWord("DATAMAP")
+ protected val ON = carbonKeyWord("ON")
+ protected val DMPROPERTIES = carbonKeyWord("DMPROPERTIES")
protected val doubleQuotedString = "\"([^\"]+)\"".r
protected val singleQuotedString = "'([^']+)'".r
@@ -750,7 +753,7 @@ abstract class CarbonDDLSqlParser extends AbstractCarbonSparkSQLParser {
* @param dbName
* @return Option of String
*/
- protected def convertDbNameToLowerCase(dbName: Option[String]): Option[String] = {
+ def convertDbNameToLowerCase(dbName: Option[String]): Option[String] = {
dbName match {
case Some(databaseName) => Some(convertDbNameToLowerCase(databaseName))
case None => dbName
http://git-wip-us.apache.org/repos/asf/carbondata/blob/c1eefee7/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
new file mode 100644
index 0000000..8b6e3b9
--- /dev/null
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/datamap/CarbonCreateDataMapCommand.scala
@@ -0,0 +1,68 @@
+/*
+ * 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.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.carbondata.common.logging.LogServiceFactory
+import org.apache.carbondata.core.metadata.schema.table.DataMapSchema
+
+/**
+ * Below command class will be used to create datamap on table
+ * and updating the parent table about the datamap information
+ *
+ * @param queryString
+ */
+case class CarbonCreateDataMapCommand(
+ dataMapName: String,
+ tableIdentifier: TableIdentifier,
+ dmClassName: String,
+ dmproperties: Map[String, String],
+ queryString: Option[String])
+ extends RunnableCommand with SchemaProcessCommand {
+
+ override def run(sparkSession: SparkSession): Seq[Row] = {
+ processSchema(sparkSession)
+ }
+
+ override def processSchema(sparkSession: SparkSession): Seq[Row] = {
+ val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)
+ if (dmClassName.equals("org.apache.carbondata.datamap.AggregateDataMapHandler") ||
+ dmClassName.equalsIgnoreCase("preaggregate")) {
+ CreatePreAggregateTableCommand(dataMapName,
+ tableIdentifier,
+ dmClassName,
+ dmproperties,
+ queryString.get).run(sparkSession)
+ } else {
+ val dataMapSchema = new DataMapSchema(dataMapName, dmClassName)
+ dataMapSchema.setProperties(new java.util.HashMap[String, String](dmproperties.asJava))
+ val dbName = GetDB.getDatabaseName(tableIdentifier.database, sparkSession)
+ // upadting the parent table about dataschema
+ PreAggregateUtil.updateMainTable(dbName, tableIdentifier.table, dataMapSchema, sparkSession)
+ }
+ LOGGER.audit(s"DataMap ${dataMapName} successfully added to Table ${tableIdentifier.table}")
+ Seq.empty
+ }
+}
+
+
http://git-wip-us.apache.org/repos/asf/carbondata/blob/c1eefee7/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 e12cbb9..ebf6273 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
@@ -16,16 +16,14 @@
*/
package org.apache.spark.sql.execution.command.preaaggregate
+import scala.collection.mutable
+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.carbondata.common.logging.LogServiceFactory
-import org.apache.carbondata.core.constants.CarbonCommonConstants
-import org.apache.carbondata.core.exception.InvalidConfigurationException
-import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier
-import org.apache.carbondata.core.metadata.schema.table.TableInfo
-import org.apache.carbondata.core.util.CarbonUtil
+import org.apache.spark.sql.hive.CarbonRelation
+import org.apache.spark.sql.parser.CarbonSpark2SqlParser
/**
* Below command class will be used to create pre-aggregate table
@@ -34,17 +32,14 @@ import org.apache.carbondata.core.util.CarbonUtil
* 1. failed to create pre aggregate table.
* 2. failed to update main table
*
- * @param cm
- * @param dataFrame
- * @param createDSTable
* @param queryString
*/
case class CreatePreAggregateTableCommand(
- cm: TableModel,
- dataFrame: DataFrame,
- createDSTable: Boolean = true,
- queryString: String,
- fieldRelationMap: scala.collection.mutable.LinkedHashMap[Field, DataMapField])
+ dataMapName: String,
+ parentTableIdentifier: TableIdentifier,
+ dmClassName: String,
+ dmproperties: Map[String, String],
+ queryString: String)
extends RunnableCommand with SchemaProcessCommand {
override def run(sparkSession: SparkSession): Seq[Row] = {
@@ -52,87 +47,61 @@ case class CreatePreAggregateTableCommand(
}
override def processSchema(sparkSession: SparkSession): Seq[Row] = {
- val storePath = CarbonEnv.getInstance(sparkSession).storePath
- CarbonEnv.getInstance(sparkSession).carbonMetastore.
- checkSchemasModifiedTimeAndReloadTables(storePath)
- val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)
- cm.databaseName = GetDB.getDatabaseName(cm.databaseNameOp, sparkSession)
- val tbName = cm.tableName
- val dbName = cm.databaseName
- LOGGER.audit(s"Creating Table with Database name [$dbName] and Table name [$tbName]")
+ val df = sparkSession.sql(queryString)
+ val fieldRelationMap = PreAggregateUtil
+ .validateActualSelectPlanAndGetAttrubites(df.logicalPlan, queryString)
+ val fields = fieldRelationMap.keySet.toSeq
+ val tableProperties = mutable.Map[String, String]()
+ dmproperties.foreach(t => tableProperties.put(t._1, t._2))
+ // Create the aggregation table name with parent table name prefix
+ val tableIdentifier = TableIdentifier(
+ parentTableIdentifier.table +"_" + dataMapName, parentTableIdentifier.database)
+ // prepare table model of the collected tokens
+ val tableModel: TableModel = new CarbonSpark2SqlParser().prepareTableModel(false,
+ new CarbonSpark2SqlParser().convertDbNameToLowerCase(tableIdentifier.database),
+ tableIdentifier.table.toLowerCase,
+ fields,
+ Seq(),
+ tableProperties,
+ None,
+ false,
+ None)
+
// getting the parent table
- val parentTable = PreAggregateUtil.getParentCarbonTable(dataFrame.logicalPlan)
+ val parentTable = PreAggregateUtil.getParentCarbonTable(df.logicalPlan)
// getting the table name
val parentTableName = parentTable.getFactTableName
// getting the db name of parent table
val parentDbName = parentTable.getDatabaseName
+
+ assert(parentTableName.equalsIgnoreCase(parentTableIdentifier.table))
// updating the relation identifier, this will be stored in child table
// which can be used during dropping of pre-aggreate table as parent table will
// also get updated
- cm.parentTable = Some(parentTable)
- cm.dataMapRelation = Some(fieldRelationMap)
- val tableInfo: TableInfo = TableNewProcessor(cm)
- // Add validation for sort scope when create table
- val sortScope = tableInfo.getFactTable.getTableProperties
- .getOrDefault("sort_scope", CarbonCommonConstants.LOAD_SORT_SCOPE_DEFAULT)
- if (!CarbonUtil.isValidSortOption(sortScope)) {
- throw new InvalidConfigurationException(
- s"Passing invalid SORT_SCOPE '$sortScope', valid SORT_SCOPE are 'NO_SORT', 'BATCH_SORT'," +
- s" 'LOCAL_SORT' and 'GLOBAL_SORT' ")
- }
-
- if (tableInfo.getFactTable.getListOfColumns.size <= 0) {
- sys.error("No Dimensions found. Table should have at least one dimesnion !")
- }
-
- if (sparkSession.sessionState.catalog.listTables(dbName)
- .exists(_.table.equalsIgnoreCase(tbName))) {
- if (!cm.ifNotExistsSet) {
- LOGGER.audit(
- s"Table creation with Database name [$dbName] and Table name [$tbName] failed. " +
- s"Table [$tbName] already exists under database [$dbName]")
- sys.error(s"Table [$tbName] already exists under database [$dbName]")
- }
- } else {
- val tableIdentifier = AbsoluteTableIdentifier.from(storePath, dbName, tbName)
- // Add Database to catalog and persist
- val catalog = CarbonEnv.getInstance(sparkSession).carbonMetastore
- val tablePath = tableIdentifier.getTablePath
- val carbonSchemaString = catalog.generateTableSchemaString(tableInfo, tablePath)
- if (createDSTable) {
- try {
- val fields = new Array[Field](cm.dimCols.size + cm.msrCols.size)
- cm.dimCols.foreach(f => fields(f.schemaOrdinal) = f)
- cm.msrCols.foreach(f => fields(f.schemaOrdinal) = f)
- sparkSession.sql(
- s"""CREATE TABLE $dbName.$tbName
- |(${ fields.map(f => f.rawSchema).mkString(",") })
- |USING org.apache.spark.sql.CarbonSource""".stripMargin +
- s""" OPTIONS (tableName "$tbName", dbName "$dbName", tablePath """.stripMargin +
- s""""$tablePath"$carbonSchemaString) """)
- // child schema object which will be updated on parent table about the
- val childSchema = tableInfo.getFactTable
- .buildChildSchema("", tableInfo.getDatabaseName, queryString, "AGGREGATION")
- // upadting the parent table about child table
- PreAggregateUtil.updateMainTable(parentDbName, parentTableName, childSchema, sparkSession)
- val loadAvailable = PreAggregateUtil
- .checkMainTableLoad(parentTable)
- if (loadAvailable) {
- sparkSession.sql(s"insert into ${ cm.databaseName }.${ cm.tableName } $queryString")
- }
- } catch {
- case e: Exception =>
- val identifier: TableIdentifier = TableIdentifier(tbName, Some(dbName))
- // call the drop table to delete the created table.
- CarbonEnv.getInstance(sparkSession).carbonMetastore
- .dropTable(tablePath, identifier)(sparkSession)
- LOGGER.audit(s"Table creation with Database name [$dbName] " +
- s"and Table name [$tbName] failed")
- throw e
- }
+ tableModel.parentTable = Some(parentTable)
+ tableModel.dataMapRelation = Some(fieldRelationMap)
+ CarbonCreateTableCommand(tableModel).run(sparkSession)
+ try {
+ val relation = CarbonEnv.getInstance(sparkSession).carbonMetastore.
+ lookupRelation(tableIdentifier)(sparkSession).asInstanceOf[CarbonRelation]
+ val tableInfo = relation.tableMeta.carbonTable.getTableInfo
+ // child schema object which will be updated on parent table about the
+ val childSchema = tableInfo.getFactTable.buildChildSchema(
+ dataMapName, "", tableInfo.getDatabaseName, queryString, "AGGREGATION")
+ dmproperties.foreach(f => childSchema.getProperties.put(f._1, f._2))
+ // updating the parent table about child table
+ PreAggregateUtil.updateMainTable(parentDbName, parentTableName, childSchema, sparkSession)
+ val loadAvailable = PreAggregateUtil.checkMainTableLoad(parentTable)
+ if (loadAvailable) {
+ sparkSession.sql(
+ s"insert into ${ tableModel.databaseName }.${ tableModel.tableName } $queryString")
}
+ } catch {
+ case e: Exception =>
+ sparkSession.sql(
+ s"""DROP TABLE IF EXISTS ${ tableModel.databaseName }.${ tableModel.tableName }""")
+ throw e
- LOGGER.audit(s"Table created with Database name [$dbName] and Table name [$tbName]")
}
Seq.empty
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/c1eefee7/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateListeners.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateListeners.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateListeners.scala
index 2ce97fe..8271e57 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateListeners.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateListeners.scala
@@ -41,9 +41,11 @@ object DropPreAggregateTablePostListener extends OperationEventListener {
!carbonTable.get.getTableInfo.getDataMapSchemaList.isEmpty) {
val childSchemas = carbonTable.get.getTableInfo.getDataMapSchemaList
for (childSchema: DataMapSchema <- childSchemas.asScala) {
- CarbonDropTableCommand(ifExistsSet = true,
- Some(childSchema.getRelationIdentifier.getDatabaseName),
- childSchema.getRelationIdentifier.getTableName).run(sparkSession)
+ if (childSchema.getRelationIdentifier != null) {
+ CarbonDropTableCommand(ifExistsSet = true,
+ Some(childSchema.getRelationIdentifier.getDatabaseName),
+ childSchema.getRelationIdentifier.getTableName).run(sparkSession)
+ }
}
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/c1eefee7/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateUtil.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateUtil.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateUtil.scala
index b35b525..b926705 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateUtil.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateUtil.scala
@@ -17,6 +17,7 @@
package org.apache.spark.sql.execution.command.preaaggregate
import scala.collection.mutable.ListBuffer
+import scala.collection.JavaConverters._
import org.apache.spark.SparkConf
import org.apache.spark.sql.{CarbonDatasourceHadoopRelation, CarbonEnv, SparkSession}
@@ -303,6 +304,10 @@ object PreAggregateUtil {
tableName,
carbonTable.getStorePath)
numberOfCurrentChild = wrapperTableInfo.getDataMapSchemaList.size
+ if (wrapperTableInfo.getDataMapSchemaList.asScala.
+ exists(f => f.getDataMapName.equalsIgnoreCase(childSchema.getDataMapName))) {
+ throw new Exception("Duplicate datamap")
+ }
wrapperTableInfo.getDataMapSchemaList.add(childSchema)
val thriftTable = schemaConverter
.fromWrapperToExternalTableInfo(wrapperTableInfo, dbName, tableName)
@@ -446,7 +451,6 @@ object PreAggregateUtil {
}
def checkMainTableLoad(carbonTable: CarbonTable): Boolean = {
- SegmentStatusManager.readLoadMetadata(
- carbonTable.getMetaDataFilepath).nonEmpty
+ SegmentStatusManager.readLoadMetadata(carbonTable.getMetaDataFilepath).nonEmpty
}
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/c1eefee7/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonFileMetastore.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonFileMetastore.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonFileMetastore.scala
index ac75fa7..a5e1ec4 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonFileMetastore.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonFileMetastore.scala
@@ -465,8 +465,9 @@ class CarbonFileMetastore extends CarbonMetaStore {
}
val childSchemaIterator = childSchemas.iterator()
while (childSchemaIterator.hasNext) {
- val childSchema = childSchemaIterator.next()
- if (childSchema.getChildSchema.equals(childCarbonTable.getTableInfo.getFactTable)) {
+ val childSchema = childSchemaIterator.next().getChildSchema
+ if (childSchema != null &&
+ childSchema.equals(childCarbonTable.getTableInfo.getFactTable)) {
childSchemaIterator.remove()
}
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/c1eefee7/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSpark2SqlParser.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSpark2SqlParser.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSpark2SqlParser.scala
index bf21bc8..46a2515 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSpark2SqlParser.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSpark2SqlParser.scala
@@ -21,11 +21,12 @@ import scala.collection.mutable
import scala.language.implicitConversions
import org.apache.spark.sql.{DeleteRecords, ShowLoadsCommand, UpdateTable}
-import org.apache.spark.sql.catalyst.CarbonDDLSqlParser
+import org.apache.spark.sql.catalyst.{CarbonDDLSqlParser, TableIdentifier}
import org.apache.spark.sql.catalyst.CarbonTableIdentifierImplicit._
import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.execution.command._
+import org.apache.spark.sql.execution.command.datamap.CarbonCreateDataMapCommand
import org.apache.spark.sql.execution.command.management.{AlterTableCompactionCommand, CleanFilesCommand, DeleteLoadByIdCommand, DeleteLoadByLoadDateCommand, LoadTableCommand}
import org.apache.spark.sql.execution.command.partition.{AlterTableDropCarbonPartitionCommand, AlterTableSplitCarbonPartitionCommand}
import org.apache.spark.sql.execution.command.schema.{CarbonAlterTableAddColumnCommand, CarbonAlterTableDataTypeChangeCommand, CarbonAlterTableDropColumnCommand}
@@ -66,7 +67,8 @@ class CarbonSpark2SqlParser extends CarbonDDLSqlParser {
protected lazy val start: Parser[LogicalPlan] = explainPlan | startCommand
protected lazy val startCommand: Parser[LogicalPlan] =
- loadManagement|showLoads|alterTable|restructure|updateTable|deleteRecords|alterPartition
+ loadManagement|showLoads|alterTable|restructure|updateTable|deleteRecords|
+ alterPartition|datamapManagement
protected lazy val loadManagement: Parser[LogicalPlan] =
deleteLoadsByID | deleteLoadsByLoadDate | cleanFiles | loadDataNew
@@ -77,6 +79,9 @@ class CarbonSpark2SqlParser extends CarbonDDLSqlParser {
protected lazy val alterPartition: Parser[LogicalPlan] =
alterAddPartition | alterSplitPartition | alterDropPartition
+ protected lazy val datamapManagement: Parser[LogicalPlan] =
+ createDataMap
+
protected lazy val alterAddPartition: Parser[LogicalPlan] =
ALTER ~> TABLE ~> (ident <~ ".").? ~ ident ~ (ADD ~> PARTITION ~>
"(" ~> repsep(stringLit, ",") <~ ")") <~ opt(";") ^^ {
@@ -121,6 +126,21 @@ class CarbonSpark2SqlParser extends CarbonDDLSqlParser {
AlterTableCompactionCommand(altertablemodel)
}
+ /**
+ * The syntax of datamap creation is as follows.
+ * CREATE DATAMAP datamapName ON TABLE tableName USING 'DataMapClassName'
+ * DMPROPERTIES('KEY'='VALUE') AS SELECT COUNT(COL1) FROM tableName
+ */
+ protected lazy val createDataMap: Parser[LogicalPlan] =
+ CREATE ~> DATAMAP ~> ident ~ (ON ~ TABLE) ~ (ident <~ ".").? ~ ident ~
+ (USING ~> stringLit) ~ (DMPROPERTIES ~> "(" ~> repsep(loadOptions, ",") <~ ")").? ~
+ (AS ~> restInput).? <~ opt(";") ^^ {
+ case dmname ~ ontable ~ dbName ~ tableName ~ className ~ dmprops ~ query =>
+ val map = dmprops.getOrElse(List[(String, String)]()).toMap[String, String]
+ CarbonCreateDataMapCommand(dmname,
+ TableIdentifier(tableName, dbName), className, map, query)
+ }
+
protected lazy val deleteRecords: Parser[LogicalPlan] =
(DELETE ~> FROM ~> table) ~ restInput.? <~ opt(";") ^^ {
case table ~ rest =>
http://git-wip-us.apache.org/repos/asf/carbondata/blob/c1eefee7/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSparkSqlParser.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSparkSqlParser.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSparkSqlParser.scala
index 256477e..5c51156 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSparkSqlParser.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/parser/CarbonSparkSqlParser.scala
@@ -25,8 +25,7 @@ import org.apache.spark.sql.catalyst.parser.ParserUtils._
import org.apache.spark.sql.catalyst.parser.SqlBaseParser.{CreateTableContext, TablePropertyListContext}
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
import org.apache.spark.sql.execution.SparkSqlAstBuilder
-import org.apache.spark.sql.execution.command._
-import org.apache.spark.sql.execution.command.preaaggregate.{CreatePreAggregateTableCommand, PreAggregateUtil}
+import org.apache.spark.sql.execution.command.{BucketFields, CarbonCreateTableCommand, PartitionerField, TableModel}
import org.apache.spark.sql.internal.{SQLConf, VariableSubstitution}
import org.apache.spark.sql.types.StructField
@@ -40,7 +39,7 @@ import org.apache.carbondata.spark.util.CommonUtil
*/
class CarbonSparkSqlParser(conf: SQLConf, sparkSession: SparkSession) extends AbstractSqlParser {
- val astBuilder = new CarbonSqlAstBuilder(conf, sparkSession: SparkSession)
+ val astBuilder = new CarbonSqlAstBuilder(conf)
private val substitutor = new VariableSubstitution(conf)
@@ -70,8 +69,7 @@ class CarbonSparkSqlParser(conf: SQLConf, sparkSession: SparkSession) extends Ab
}
}
-class CarbonSqlAstBuilder(conf: SQLConf, sparkSession: SparkSession)
- extends SparkSqlAstBuilder(conf) {
+class CarbonSqlAstBuilder(conf: SQLConf) extends SparkSqlAstBuilder(conf) {
val parser = new CarbonSpark2SqlParser
@@ -119,18 +117,8 @@ class CarbonSqlAstBuilder(conf: SQLConf, sparkSession: SparkSession)
val (partitionByStructFields, partitionFields) =
validateParitionFields(ctx, colNames, tableProperties)
- val isAggTable = tableProperties.get("parent").isDefined
- var fields = parser.getFields(colsStructFields ++ partitionByStructFields)
- val dfAndFieldRelationTuple = if (isAggTable) {
- val selectQuery = Option(ctx.query).map(plan).get
- val df = Dataset.ofRows(sparkSession, selectQuery)
- val fieldRelationMap = PreAggregateUtil
- .validateActualSelectPlanAndGetAttrubites(df.logicalPlan, source(ctx.query()))
- fields = fieldRelationMap.keySet.toSeq
- Some(df, fieldRelationMap)
- } else {
- None
- }
+ val fields = parser.getFields(colsStructFields ++ partitionByStructFields)
+
// validate bucket fields
val bucketFields: Option[BucketFields] =
parser.getBucketFields(tableProperties, fields, options)
@@ -149,14 +137,7 @@ class CarbonSqlAstBuilder(conf: SQLConf, sparkSession: SparkSession)
isAlterFlow = false,
tableComment)
- if(!isAggTable) {
- CarbonCreateTableCommand(tableModel)
- } else {
- CreatePreAggregateTableCommand(tableModel,
- dfAndFieldRelationTuple.get._1,
- queryString = source(ctx.query).toString,
- fieldRelationMap = dfAndFieldRelationTuple.get._2)
- }
+ CarbonCreateTableCommand(tableModel)
} else {
super.visitCreateTable(ctx)
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/c1eefee7/integration/spark2/src/test/scala/org/apache/spark/carbondata/restructure/AlterTableValidationTestCase.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/test/scala/org/apache/spark/carbondata/restructure/AlterTableValidationTestCase.scala b/integration/spark2/src/test/scala/org/apache/spark/carbondata/restructure/AlterTableValidationTestCase.scala
index 0274605..b798379 100644
--- a/integration/spark2/src/test/scala/org/apache/spark/carbondata/restructure/AlterTableValidationTestCase.scala
+++ b/integration/spark2/src/test/scala/org/apache/spark/carbondata/restructure/AlterTableValidationTestCase.scala
@@ -493,13 +493,13 @@ class AlterTableValidationTestCase extends Spark2QueryTest with BeforeAndAfterAl
test("test to check if new parent table name is reflected in pre-aggregate tables") {
sql("drop table if exists preaggMain")
sql("drop table if exists preaggmain_new")
- sql("drop table if exists preagg1")
+ sql("drop table if exists preaggMain_preagg1")
sql("create table preaggMain (a string, b string, c string) stored by 'carbondata'")
sql(
- "create table preagg1 stored BY 'carbondata' tblproperties('parent'='PreAggMain') as select" +
+ "create datamap preagg1 on table PreAggMain using 'preaggregate' as select" +
" a,sum(b) from PreAggMain group by a")
intercept[RuntimeException] {
- sql("alter table preagg1 rename to preagg2")
+ sql("alter table PreAggMain_preagg1 rename to preagg2")
}.getMessage.contains("Rename operation for pre-aggregate table is not supported.")
intercept[RuntimeException] {
sql("alter table preaggmain rename to preaggmain_new")
http://git-wip-us.apache.org/repos/asf/carbondata/blob/c1eefee7/integration/spark2/src/test/scala/org/apache/spark/carbondata/restructure/vectorreader/ChangeDataTypeTestCases.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/test/scala/org/apache/spark/carbondata/restructure/vectorreader/ChangeDataTypeTestCases.scala b/integration/spark2/src/test/scala/org/apache/spark/carbondata/restructure/vectorreader/ChangeDataTypeTestCases.scala
index 67ea21e..31a8b6c 100644
--- a/integration/spark2/src/test/scala/org/apache/spark/carbondata/restructure/vectorreader/ChangeDataTypeTestCases.scala
+++ b/integration/spark2/src/test/scala/org/apache/spark/carbondata/restructure/vectorreader/ChangeDataTypeTestCases.scala
@@ -149,19 +149,19 @@ class ChangeDataTypeTestCases extends Spark2QueryTest with BeforeAndAfterAll {
test("test data type change for with pre-aggregate table should throw exception") {
sql("drop table if exists preaggMain")
- sql("drop table if exists preagg1")
+ sql("drop table if exists PreAggMain_preagg1")
sql("create table preaggMain (a string, b string, c string) stored by 'carbondata'")
sql(
- "create table preagg1 stored BY 'carbondata' tblproperties('parent'='PreAggMain') as select" +
+ "create datamap preagg1 on table PreAggMain using 'preaggregate' as select" +
" a,sum(b) from PreAggMain group by a")
intercept[RuntimeException] {
sql("alter table preaggmain drop columns(a)").show
}.getMessage.contains("exists in pre-aggregate table")
intercept[RuntimeException] {
- sql("alter table preagg1 drop columns(a)").show
+ sql("alter table PreAggMain_preagg1 drop columns(a)").show
}.getMessage.contains("cannot be dropped")
sql("drop table if exists preaggMain")
- sql("drop table if exists preagg1")
+ sql("drop table if exists PreAggMain_preagg1")
}
override def afterAll {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/c1eefee7/integration/spark2/src/test/scala/org/apache/spark/carbondata/restructure/vectorreader/DropColumnTestCases.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/test/scala/org/apache/spark/carbondata/restructure/vectorreader/DropColumnTestCases.scala b/integration/spark2/src/test/scala/org/apache/spark/carbondata/restructure/vectorreader/DropColumnTestCases.scala
index 1a1d5d5..b887771 100644
--- a/integration/spark2/src/test/scala/org/apache/spark/carbondata/restructure/vectorreader/DropColumnTestCases.scala
+++ b/integration/spark2/src/test/scala/org/apache/spark/carbondata/restructure/vectorreader/DropColumnTestCases.scala
@@ -100,10 +100,10 @@ class DropColumnTestCases extends Spark2QueryTest with BeforeAndAfterAll {
test("test dropping of column in pre-aggregate should throw exception") {
sql("drop table if exists preaggMain")
- sql("drop table if exists preagg1")
+ sql("drop table if exists preaggMain_preagg1")
sql("create table preaggMain (a string, b string, c string) stored by 'carbondata'")
sql(
- "create table preagg1 stored BY 'carbondata' tblproperties('parent'='PreAggMain') as select" +
+ "create datamap preagg1 on table PreAggMain using 'preaggregate' as select" +
" a,sum(b) from PreAggMain group by a")
sql("alter table preaggmain drop columns(c)").show
checkExistence(sql("desc table preaggmain"), false, "c")
@@ -111,7 +111,7 @@ class DropColumnTestCases extends Spark2QueryTest with BeforeAndAfterAll {
sql("alter table preaggmain drop columns(a)").show
}.getMessage.contains("cannot be dropped")
sql("drop table if exists preaggMain")
- sql("drop table if exists preagg1")
+ sql("drop table if exists preaggMain_preagg1")
}
override def afterAll {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/c1eefee7/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 c65bcc4..f79ea2d 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
@@ -143,14 +143,14 @@ class CarbonCommandSuite extends Spark2QueryTest with BeforeAndAfterAll {
test("test if delete segments by id is unsupported for pre-aggregate tables") {
dropTable("preaggMain")
- dropTable("preagg1")
+ dropTable("preaggMain_preagg1")
sql("create table preaggMain (a string, b string, c string) stored by 'carbondata'")
- sql("create table preagg1 stored BY 'carbondata' tblproperties('parent'='PreAggMain') as select a,sum(b) from PreAggMain group by a")
+ sql("create datamap preagg1 on table PreAggMain using 'preaggregate' as select a,sum(b) from PreAggMain group by a")
intercept[UnsupportedOperationException] {
sql("delete from table preaggMain where segment.id in (1,2)")
}.getMessage.contains("Delete segment operation is not supported on tables")
intercept[UnsupportedOperationException] {
- sql("delete from table preagg1 where segment.id in (1,2)")
+ sql("delete from table preaggMain_preagg1 where segment.id in (1,2)")
}.getMessage.contains("Delete segment operation is not supported on pre-aggregate tables")
dropTable("preaggMain")
dropTable("preagg1")