You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@carbondata.apache.org by ra...@apache.org on 2017/11/13 22:18:30 UTC
carbondata git commit: [CARBONDATA-1528] [PreAgg] Restrict alter for
pre-aggregate tables
Repository: carbondata
Updated Branches:
refs/heads/master e4f2843c6 -> fab102b89
[CARBONDATA-1528] [PreAgg] Restrict alter for pre-aggregate tables
This closes #1493
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/fab102b8
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/fab102b8
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/fab102b8
Branch: refs/heads/master
Commit: fab102b89a41fe08ad4d583e35dcbaa6ab6177b9
Parents: e4f2843
Author: kunal642 <ku...@gmail.com>
Authored: Tue Nov 14 00:23:11 2017 +0530
Committer: ravipesala <ra...@gmail.com>
Committed: Tue Nov 14 03:48:19 2017 +0530
----------------------------------------------------------------------
.../schema/table/RelationIdentifier.java | 4 +
.../carbondata/events/AlterTableEvents.scala | 11 ++-
.../org/apache/carbondata/events/Events.scala | 8 ++
.../org/apache/spark/sql/CarbonSession.scala | 21 +++++
.../preaaggregate/PreAggregateListeners.scala | 85 ++++++++++++--------
.../CarbonAlterTableAddColumnCommand.scala | 4 +
.../CarbonAlterTableDataTypeChangeCommand.scala | 4 +
.../schema/CarbonAlterTableRenameCommand.scala | 15 ++--
.../spark/sql/hive/CarbonSessionState.scala | 19 -----
.../AlterTableValidationTestCase.scala | 10 +--
.../vectorreader/AddColumnTestCases.scala | 14 ++++
.../vectorreader/ChangeDataTypeTestCases.scala | 14 ++--
.../vectorreader/DropColumnTestCases.scala | 8 +-
13 files changed, 140 insertions(+), 77 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/fab102b8/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/RelationIdentifier.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/RelationIdentifier.java b/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/RelationIdentifier.java
index c9c44bf..20e539b 100644
--- a/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/RelationIdentifier.java
+++ b/core/src/main/java/org/apache/carbondata/core/metadata/schema/table/RelationIdentifier.java
@@ -101,4 +101,8 @@ public class RelationIdentifier implements Serializable, Writable {
result = 31 * result + (tableId != null ? tableId.hashCode() : 0);
return result;
}
+
+ @Override public String toString() {
+ return databaseName + "." + tableName;
+ }
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/fab102b8/integration/spark-common/src/main/scala/org/apache/carbondata/events/AlterTableEvents.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/events/AlterTableEvents.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/events/AlterTableEvents.scala
index 2f7cf63..5a818fc 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/events/AlterTableEvents.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/events/AlterTableEvents.scala
@@ -18,7 +18,7 @@ package org.apache.carbondata.events
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.SQLContext
-import org.apache.spark.sql.execution.command.{AlterTableDataTypeChangeModel, AlterTableDropColumnModel, AlterTableRenameModel}
+import org.apache.spark.sql.execution.command.{AlterTableAddColumnsModel, AlterTableDataTypeChangeModel, AlterTableDropColumnModel, AlterTableRenameModel}
import org.apache.carbondata.core.metadata.schema.table.CarbonTable
import org.apache.carbondata.processing.loading.model.CarbonLoadModel
@@ -77,6 +77,15 @@ case class AlterTableRenamePreEvent(carbonTable: CarbonTable,
alterTableRenameModel: AlterTableRenameModel, newTablePath: String,
sparkSession: SparkSession) extends Event with AlterTableRenameEventInfo
+/**
+ *
+ * @param carbonTable
+ * @param alterTableAddColumnsModel
+ */
+case class AlterTableAddColumnPreEvent(carbonTable: CarbonTable,
+ alterTableAddColumnsModel: AlterTableAddColumnsModel)
+ extends Event with AlterTableAddColumnEventInfo
+
/**
*
http://git-wip-us.apache.org/repos/asf/carbondata/blob/fab102b8/integration/spark-common/src/main/scala/org/apache/carbondata/events/Events.scala
----------------------------------------------------------------------
diff --git a/integration/spark-common/src/main/scala/org/apache/carbondata/events/Events.scala b/integration/spark-common/src/main/scala/org/apache/carbondata/events/Events.scala
index 9796dea..9c542b6 100644
--- a/integration/spark-common/src/main/scala/org/apache/carbondata/events/Events.scala
+++ b/integration/spark-common/src/main/scala/org/apache/carbondata/events/Events.scala
@@ -83,6 +83,14 @@ trait AlterTableRenameEventInfo {
}
/**
+ * event for alter_add_column
+ */
+trait AlterTableAddColumnEventInfo {
+ val carbonTable: CarbonTable
+ val alterTableAddColumnsModel: AlterTableAddColumnsModel
+}
+
+/**
* event for alter_table_rename
*/
trait AlterTableCompactionEventInfo {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/fab102b8/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonSession.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonSession.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonSession.scala
index 33e34bb..c0393d3 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonSession.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/CarbonSession.scala
@@ -24,6 +24,7 @@ import org.apache.hadoop.conf.Configuration
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.scheduler.{SparkListener, SparkListenerApplicationEnd}
import org.apache.spark.sql.SparkSession.Builder
+import org.apache.spark.sql.execution.command.preaaggregate._
import org.apache.spark.sql.execution.streaming.CarbonStreamingQueryListener
import org.apache.spark.sql.hive.CarbonSessionState
import org.apache.spark.sql.hive.execution.command.CarbonSetCommand
@@ -32,6 +33,7 @@ import org.apache.spark.util.Utils
import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.util.{CarbonProperties, CarbonSessionInfo, ThreadLocalSessionInfo}
+import org.apache.carbondata.events._
import org.apache.carbondata.spark.util.CommonUtil
/**
@@ -62,6 +64,10 @@ class CarbonSession(@transient val sc: SparkContext,
new CarbonSession(sparkContext, Some(sharedState))
}
+ if (existingSharedState.isEmpty) {
+ CarbonSession.initListeners
+ }
+
}
object CarbonSession {
@@ -218,4 +224,19 @@ object CarbonSession {
// preserve thread parameters across call
ThreadLocalSessionInfo.setCarbonSessionInfo(carbonSessionInfo)
}
+
+ def initListeners(): Unit = {
+ OperationListenerBus.getInstance()
+ .addListener(classOf[DropTablePostEvent], DropPreAggregateTablePostListener)
+ .addListener(classOf[LoadTablePostExecutionEvent], LoadPostAggregateListener)
+ .addListener(classOf[DeleteSegmentByIdPreEvent], PreAggregateDeleteSegmentByIdPreListener)
+ .addListener(classOf[DeleteSegmentByDatePreEvent], PreAggregateDeleteSegmentByDatePreListener)
+ .addListener(classOf[UpdateTablePreEvent], UpdatePreAggregatePreListener)
+ .addListener(classOf[DeleteFromTablePreEvent], DeletePreAggregatePreListener)
+ .addListener(classOf[DeleteFromTablePreEvent], DeletePreAggregatePreListener)
+ .addListener(classOf[AlterTableDropColumnPreEvent], PreAggregateDropColumnPreListener)
+ .addListener(classOf[AlterTableRenamePreEvent], PreAggregateRenameTablePreListener)
+ .addListener(classOf[AlterTableDataTypeChangePreEvent], PreAggregateDataTypeChangePreListener)
+ .addListener(classOf[AlterTableAddColumnPreEvent], PreAggregateAddColumnsPreListener)
+ }
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/fab102b8/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 7a66e88..6785f73 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
@@ -37,8 +37,7 @@ object DropPreAggregateTablePostListener extends OperationEventListener {
val dropPostEvent = event.asInstanceOf[DropTablePostEvent]
val carbonTable = dropPostEvent.carbonTable
val sparkSession = dropPostEvent.sparkSession
- if (carbonTable.isDefined && carbonTable.get.getTableInfo.getDataMapSchemaList != null &&
- !carbonTable.get.getTableInfo.getDataMapSchemaList.isEmpty) {
+ if (carbonTable.isDefined && carbonTable.get.hasDataMapSchema) {
val childSchemas = carbonTable.get.getTableInfo.getDataMapSchemaList
for (childSchema: DataMapSchema <- childSchemas.asScala) {
if (childSchema.getRelationIdentifier != null) {
@@ -63,7 +62,7 @@ object LoadPostAggregateListener extends OperationEventListener {
val sparkSession = loadEvent.sparkSession
val carbonLoadModel = loadEvent.carbonLoadModel
val table = carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable
- if (!table.getTableInfo.getDataMapSchemaList.isEmpty) {
+ if (table.hasDataMapSchema) {
for (dataMapSchema: DataMapSchema <- table.getTableInfo.getDataMapSchemaList.asScala) {
CarbonSession
.threadSet(CarbonCommonConstants.CARBON_INPUT_SEGMENTS +
@@ -94,23 +93,41 @@ object PreAggregateDataTypeChangePreListener extends OperationEventListener {
val carbonTable = dataTypeChangePreListener.carbonTable
val alterTableDataTypeChangeModel = dataTypeChangePreListener.alterTableDataTypeChangeModel
val columnToBeAltered: String = alterTableDataTypeChangeModel.columnName
- val dataMapSchemas = carbonTable.getTableInfo.getDataMapSchemaList
- if (dataMapSchemas != null && !dataMapSchemas.isEmpty) {
+ if (carbonTable.hasDataMapSchema) {
+ val dataMapSchemas = carbonTable.getTableInfo.getDataMapSchemaList
dataMapSchemas.asScala.foreach { dataMapSchema =>
- val childColumns = dataMapSchema.getChildSchema.getListOfColumns
- if (childColumns.asScala.exists(_.getColumnName.equalsIgnoreCase(columnToBeAltered))) {
- throw new UnsupportedOperationException(s"Column $columnToBeAltered exists in a " +
- s"pre-aggregate table ${ dataMapSchema.toString
- }. Cannot change datatype")
- }
+ val childColumns = dataMapSchema.getChildSchema.getListOfColumns
+ val parentColumnNames = childColumns.asScala
+ .flatMap(_.getParentColumnTableRelations.asScala.map(_.getColumnName))
+ if (parentColumnNames.contains(columnToBeAltered)) {
+ throw new UnsupportedOperationException(
+ s"Column $columnToBeAltered exists in a pre-aggregate table. Drop pre-aggregate table" +
+ "to continue")
+ }
}
+ }
+ if (carbonTable.isChildDataMap) {
+ throw new UnsupportedOperationException(
+ s"Cannot change data type for columns in pre-aggregate table ${ carbonTable.getDatabaseName
+ }.${ carbonTable.getFactTableName }")
+ }
+ }
+}
- if (carbonTable.isChildDataMap) {
- throw new UnsupportedOperationException(s"Cannot change data type for columns in " +
- s"pre-aggreagate table ${
- carbonTable.getDatabaseName
- }.${ carbonTable.getFactTableName }")
- }
+object PreAggregateAddColumnsPreListener extends OperationEventListener {
+ /**
+ * Called on a specified event occurrence
+ *
+ * @param event
+ * @param operationContext
+ */
+ override def onEvent(event: Event, operationContext: OperationContext): Unit = {
+ val dataTypeChangePreListener = event.asInstanceOf[AlterTableAddColumnPreEvent]
+ val carbonTable = dataTypeChangePreListener.carbonTable
+ if (carbonTable.isChildDataMap) {
+ throw new UnsupportedOperationException(
+ s"Cannot add columns in pre-aggreagate table ${ carbonTable.getDatabaseName
+ }.${ carbonTable.getFactTableName }")
}
}
}
@@ -175,26 +192,26 @@ object PreAggregateDropColumnPreListener extends OperationEventListener {
val carbonTable = dataTypeChangePreListener.carbonTable
val alterTableDropColumnModel = dataTypeChangePreListener.alterTableDropColumnModel
val columnsToBeDropped = alterTableDropColumnModel.columns
- val dataMapSchemas = carbonTable.getTableInfo.getDataMapSchemaList
- if (dataMapSchemas != null && !dataMapSchemas.isEmpty) {
+ if (carbonTable.hasDataMapSchema) {
+ val dataMapSchemas = carbonTable.getTableInfo.getDataMapSchemaList
dataMapSchemas.asScala.foreach { dataMapSchema =>
- val parentColumnNames = dataMapSchema.getChildSchema.getListOfColumns.asScala
- .flatMap(_.getParentColumnTableRelations.asScala.map(_.getColumnName))
- val columnExistsInChild = parentColumnNames.collectFirst {
- case parentColumnName if columnsToBeDropped.contains(parentColumnName) =>
- parentColumnName
- }
- if (columnExistsInChild.isDefined) {
- throw new UnsupportedOperationException(
- s"Column ${ columnExistsInChild.head } cannot be dropped because it exists in a " +
- s"pre-aggregate table ${ dataMapSchema.getRelationIdentifier.toString}")
- }
- }
- if (carbonTable.isChildDataMap) {
- throw new UnsupportedOperationException(s"Cannot drop columns in pre-aggreagate table ${
- carbonTable.getDatabaseName}.${ carbonTable.getFactTableName }")
+ val parentColumnNames = dataMapSchema.getChildSchema.getListOfColumns.asScala
+ .flatMap(_.getParentColumnTableRelations.asScala.map(_.getColumnName))
+ val columnExistsInChild = parentColumnNames.collectFirst {
+ case parentColumnName if columnsToBeDropped.contains(parentColumnName) =>
+ parentColumnName
+ }
+ if (columnExistsInChild.isDefined) {
+ throw new UnsupportedOperationException(
+ s"Column ${ columnExistsInChild.head } cannot be dropped because it exists in a " +
+ s"pre-aggregate table ${ dataMapSchema.getRelationIdentifier.toString }")
+ }
}
}
+ if (carbonTable.isChildDataMap) {
+ throw new UnsupportedOperationException(s"Cannot drop columns in pre-aggreagate table ${
+ carbonTable.getDatabaseName}.${ carbonTable.getFactTableName }")
+ }
}
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/fab102b8/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/schema/CarbonAlterTableAddColumnCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/schema/CarbonAlterTableAddColumnCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/schema/CarbonAlterTableAddColumnCommand.scala
index 8737464..7cc43d2 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/schema/CarbonAlterTableAddColumnCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/schema/CarbonAlterTableAddColumnCommand.scala
@@ -29,6 +29,7 @@ import org.apache.carbondata.core.locks.{ICarbonLock, LockUsage}
import org.apache.carbondata.core.metadata.converter.ThriftWrapperSchemaConverterImpl
import org.apache.carbondata.core.metadata.schema.table.CarbonTable
import org.apache.carbondata.core.util.path.CarbonStorePath
+import org.apache.carbondata.events.{AlterTableAddColumnPreEvent, OperationListenerBus}
import org.apache.carbondata.format.TableInfo
import org.apache.carbondata.spark.rdd.{AlterTableAddColumnRDD, AlterTableDropColumnRDD}
@@ -59,6 +60,9 @@ private[sql] case class CarbonAlterTableAddColumnCommand(
carbonTable = metastore
.lookupRelation(Some(dbName), tableName)(sparkSession).asInstanceOf[CarbonRelation]
.tableMeta.carbonTable
+ val alterTableAddColumnListener = AlterTableAddColumnPreEvent(carbonTable,
+ alterTableAddColumnsModel)
+ OperationListenerBus.getInstance().fireEvent(alterTableAddColumnListener)
// get the latest carbon table and check for column existence
// read the latest schema file
val carbonTablePath = CarbonStorePath.getCarbonTablePath(carbonTable.getStorePath,
http://git-wip-us.apache.org/repos/asf/carbondata/blob/fab102b8/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/schema/CarbonAlterTableDataTypeChangeCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/schema/CarbonAlterTableDataTypeChangeCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/schema/CarbonAlterTableDataTypeChangeCommand.scala
index 4e180c8..023e061 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/schema/CarbonAlterTableDataTypeChangeCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/schema/CarbonAlterTableDataTypeChangeCommand.scala
@@ -28,6 +28,7 @@ import org.apache.carbondata.common.logging.{LogService, LogServiceFactory}
import org.apache.carbondata.core.locks.{ICarbonLock, LockUsage}
import org.apache.carbondata.core.metadata.schema.table.CarbonTable
import org.apache.carbondata.core.util.path.CarbonStorePath
+import org.apache.carbondata.events.{AlterTableAddColumnPreEvent, AlterTableDataTypeChangePreEvent, OperationListenerBus}
import org.apache.carbondata.format.{ColumnSchema, SchemaEvolutionEntry, TableInfo}
import org.apache.carbondata.spark.util.{CarbonScalaUtil, DataTypeConverterUtil}
@@ -53,6 +54,9 @@ private[sql] case class CarbonAlterTableDataTypeChangeCommand(
carbonTable = metastore
.lookupRelation(Some(dbName), tableName)(sparkSession).asInstanceOf[CarbonRelation]
.tableMeta.carbonTable
+ val alterTableDataTypeChangeListener = AlterTableDataTypeChangePreEvent(carbonTable,
+ alterTableDataTypeChangeModel)
+ OperationListenerBus.getInstance().fireEvent(alterTableDataTypeChangeListener)
val columnName = alterTableDataTypeChangeModel.columnName
val carbonColumns = carbonTable.getCreateOrderColumn(tableName).asScala.filter(!_.isInvisible)
if (!carbonColumns.exists(_.getColName.equalsIgnoreCase(columnName))) {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/fab102b8/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/schema/CarbonAlterTableRenameCommand.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/schema/CarbonAlterTableRenameCommand.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/schema/CarbonAlterTableRenameCommand.scala
index b96baff..874bdfb 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/schema/CarbonAlterTableRenameCommand.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/schema/CarbonAlterTableRenameCommand.scala
@@ -90,6 +90,14 @@ private[sql] case class CarbonAlterTableRenameCommand(
// get the latest carbon table and check for column existence
val carbonTablePath = CarbonStorePath.getCarbonTablePath(oldTableIdentifier)
val tableMetadataFile = carbonTablePath.getPath
+ val operationContext = new OperationContext
+ // TODO: Pass new Table Path in pre-event.
+ val alterTableRenamePreEvent: AlterTableRenamePreEvent = AlterTableRenamePreEvent(
+ carbonTable,
+ alterTableRenameModel,
+ "",
+ sparkSession)
+ OperationListenerBus.getInstance().fireEvent(alterTableRenamePreEvent, operationContext)
val tableInfo: org.apache.carbondata.format.TableInfo =
metastore.getThriftTableInfo(carbonTablePath)(sparkSession)
val schemaEvolutionEntry = new SchemaEvolutionEntry(System.currentTimeMillis)
@@ -114,13 +122,6 @@ private[sql] case class CarbonAlterTableRenameCommand(
tableInfo,
schemaEvolutionEntry,
tableMeta.tablePath)(sparkSession)
- val operationContext = new OperationContext
- val alterTableRenamePreEvent: AlterTableRenamePreEvent = AlterTableRenamePreEvent(
- carbonTable,
- alterTableRenameModel,
- newTablePath,
- sparkSession)
- OperationListenerBus.getInstance().fireEvent(alterTableRenamePreEvent, operationContext)
metastore.removeTableFromMetadata(oldDatabaseName, oldTableName)
sparkSession.sessionState.asInstanceOf[CarbonSessionState].metadataHive
http://git-wip-us.apache.org/repos/asf/carbondata/blob/fab102b8/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonSessionState.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonSessionState.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonSessionState.scala
index 205b716..939f627 100644
--- a/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonSessionState.scala
+++ b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonSessionState.scala
@@ -126,23 +126,6 @@ class CarbonSessionCatalog(
}
}
-object CarbonSessionState {
-
- def init(): Unit = {
- OperationListenerBus.getInstance()
- .addListener(classOf[DropTablePostEvent], DropPreAggregateTablePostListener)
- .addListener(classOf[LoadTablePostExecutionEvent], LoadPostAggregateListener)
- .addListener(classOf[DeleteSegmentByIdPreEvent], PreAggregateDeleteSegmentByIdPreListener)
- .addListener(classOf[DeleteSegmentByDatePreEvent], PreAggregateDeleteSegmentByDatePreListener)
- .addListener(classOf[UpdateTablePreEvent], UpdatePreAggregatePreListener)
- .addListener(classOf[DeleteFromTablePreEvent], DeletePreAggregatePreListener)
- .addListener(classOf[DeleteFromTablePreEvent], DeletePreAggregatePreListener)
- .addListener(classOf[AlterTableDropColumnPreEvent], PreAggregateDropColumnPreListener)
- .addListener(classOf[AlterTableRenamePreEvent], PreAggregateRenameTablePreListener)
- .addListener(classOf[AlterTableDataTypeChangePreEvent], PreAggregateDataTypeChangePreListener)
- }
-}
-
/**
* Session state implementation to override sql parser and adding strategies
* @param sparkSession
@@ -159,8 +142,6 @@ class CarbonSessionState(sparkSession: SparkSession) extends HiveSessionState(sp
)
experimentalMethods.extraOptimizations = Seq(new CarbonLateDecodeRule)
- CarbonSessionState.init()
-
override lazy val optimizer: Optimizer = new CarbonOptimizer(catalog, conf, experimentalMethods)
override lazy val analyzer: Analyzer = {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/fab102b8/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 b798379..e574d7f 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
@@ -498,12 +498,12 @@ class AlterTableValidationTestCase extends Spark2QueryTest with BeforeAndAfterAl
sql(
"create datamap preagg1 on table PreAggMain using 'preaggregate' as select" +
" a,sum(b) from PreAggMain group by a")
- intercept[RuntimeException] {
- sql("alter table PreAggMain_preagg1 rename to preagg2")
- }.getMessage.contains("Rename operation for pre-aggregate table is not supported.")
- intercept[RuntimeException] {
+ assert(intercept[RuntimeException] {
+ sql("alter table preAggmain_preagg1 rename to preagg2")
+ }.getMessage.contains("Rename operation for pre-aggregate table is not supported."))
+ assert(intercept[RuntimeException] {
sql("alter table preaggmain rename to preaggmain_new")
- }.getMessage.contains("Rename operation is not supported for table with pre-aggregate tables")
+ }.getMessage.contains("Rename operation is not supported for table with pre-aggregate tables"))
}
override def afterAll {
http://git-wip-us.apache.org/repos/asf/carbondata/blob/fab102b8/integration/spark2/src/test/scala/org/apache/spark/carbondata/restructure/vectorreader/AddColumnTestCases.scala
----------------------------------------------------------------------
diff --git a/integration/spark2/src/test/scala/org/apache/spark/carbondata/restructure/vectorreader/AddColumnTestCases.scala b/integration/spark2/src/test/scala/org/apache/spark/carbondata/restructure/vectorreader/AddColumnTestCases.scala
index f5ff2e3..e78e49b 100644
--- a/integration/spark2/src/test/scala/org/apache/spark/carbondata/restructure/vectorreader/AddColumnTestCases.scala
+++ b/integration/spark2/src/test/scala/org/apache/spark/carbondata/restructure/vectorreader/AddColumnTestCases.scala
@@ -620,6 +620,20 @@ class AddColumnTestCases extends Spark2QueryTest with BeforeAndAfterAll {
}
+ test("test if adding column in pre-aggregate table throws exception") {
+ sql("drop table if exists preaggMain")
+ sql("drop table if exists preagg1")
+ sql("create table preaggMain (a string, b string, c string) stored by 'carbondata'")
+ sql(
+ "create datamap preagg1 on table PreAggMain using 'preaggregate' as select" +
+ " a,sum(b) from PreAggMain group by a")
+ assert(intercept[RuntimeException] {
+ sql("alter table preaggmain_preagg1 add columns(d string)")
+ }.getMessage.contains("Cannot add columns"))
+ sql("drop table if exists preaggMain")
+ sql("drop table if exists preagg1")
+ }
+
override def afterAll {
sql("DROP TABLE IF EXISTS addcolumntest")
sql("DROP TABLE IF EXISTS hivetable")
http://git-wip-us.apache.org/repos/asf/carbondata/blob/fab102b8/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 31a8b6c..16c0895 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
@@ -150,16 +150,16 @@ 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 PreAggMain_preagg1")
- sql("create table preaggMain (a string, b string, c string) stored by 'carbondata'")
+ sql("create table preaggMain (a int, b string, c string) stored by 'carbondata'")
sql(
"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 PreAggMain_preagg1 drop columns(a)").show
- }.getMessage.contains("cannot be dropped")
+ assert(intercept[RuntimeException] {
+ sql("alter table preaggmain change a a long").show
+ }.getMessage.contains("exists in a pre-aggregate table"))
+ assert(intercept[RuntimeException] {
+ sql("alter table preaggmain_preagg1 change a a long").show
+ }.getMessage.contains("Cannot change data type for columns in pre-aggregate table"))
sql("drop table if exists preaggMain")
sql("drop table if exists PreAggMain_preagg1")
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/fab102b8/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 b887771..8de9e9a 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
@@ -105,11 +105,11 @@ class DropColumnTestCases extends Spark2QueryTest with BeforeAndAfterAll {
sql(
"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
+ sql("alter table preaggmain drop columns(c)")
checkExistence(sql("desc table preaggmain"), false, "c")
- intercept[RuntimeException] {
- sql("alter table preaggmain drop columns(a)").show
- }.getMessage.contains("cannot be dropped")
+ assert(intercept[RuntimeException] {
+ sql("alter table preaggmain_preagg1 drop columns(preaggmain_b_sum)").show
+ }.getMessage.contains("Cannot drop columns in pre-aggreagate table"))
sql("drop table if exists preaggMain")
sql("drop table if exists preaggMain_preagg1")
}