11623

Build # 11623

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/9ad11a87e331c27863ad93af7cb0de656d7135c7">9ad11a87e<a href="https://github.com/apache/carbondata/commit/9ad11a87e331c27863ad93af7cb0de656d7135c7">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/0d7638b7afdebd33596be07df42942e734380825">0d7638b7a</a>">0d7638b7a</a><a href="https://github.com/apache/carbondata/commit/9ad11a87e331c27863ad93af7cb0de656d7135c7"> into 2a28dba04">2a28dba04</a>

Run Details

241 of 241 new or added lines in 12 files covered. (100.0%)

64255 of 82297 relevant lines covered (78.08%)

1.03 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

41.29

/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/mv/DataMapListeners.scala

/*
* 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.mv

import scala.collection.JavaConverters._
import scala.collection.mutable

import org.apache.spark.sql.{CarbonEnv, SparkSession}
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.execution.command.AlterTableModel
import org.apache.spark.sql.execution.command.management.CarbonAlterTableCompactionCommand
import org.apache.spark.sql.execution.command.partition.CarbonAlterTableDropHivePartitionCommand
import org.apache.spark.util.DataMapUtil

import org.apache.carbondata.common.exceptions.MetadataProcessException
import org.apache.carbondata.core.datamap.DataMapStoreManager
import org.apache.carbondata.core.datamap.status.DataMapStatusManager
import org.apache.carbondata.core.metadata.schema.table.{CarbonTable, DataMapSchema}
import org.apache.carbondata.datamap.DataMapManager
import org.apache.carbondata.events._
import org.apache.carbondata.processing.loading.events.LoadEvents.LoadTablePostExecutionEvent
import org.apache.carbondata.processing.merger.CompactionType


object DataMapListeners {
  def getDataMapTableColumns(dataMapSchema: DataMapSchema,
      carbonTable: CarbonTable): mutable.Buffer[String] = {
    val listOfColumns: mutable.Buffer[String] = new mutable.ArrayBuffer[String]()
    listOfColumns.asJava
      .addAll(dataMapSchema.getMainTableColumnList.get(carbonTable.getTableName))
    listOfColumns
  }
}

/**
 * Listener to trigger compaction on mv datamap after main table compaction
 */
object AlterDataMaptableCompactionPostListener extends OperationEventListener {
  /**
   * Called on a specified event occurrence
   *
   * @param event
   * @param operationContext
   */
  override def onEvent(event: Event, operationContext: OperationContext): Unit = {
    val compactionEvent = event.asInstanceOf[AlterTableCompactionPreStatusUpdateEvent]
    val carbonTable = compactionEvent.carbonTable
    val compactionType = compactionEvent.carbonMergerMapping.campactionType
    if (compactionType == CompactionType.CUSTOM) {
      return
    }
    val carbonLoadModel = compactionEvent.carbonLoadModel
    val sparkSession = compactionEvent.sparkSession
    val allDataMapSchemas = DataMapStoreManager.getInstance
      .getDataMapSchemasOfTable(carbonTable).asScala
      .filter(dataMapSchema => null != dataMapSchema.getRelationIdentifier &&
                               !dataMapSchema.isIndexDataMap)
    if (!allDataMapSchemas.asJava.isEmpty) {
      allDataMapSchemas.foreach { dataMapSchema =>
        val childRelationIdentifier = dataMapSchema.getRelationIdentifier
        val alterTableModel = AlterTableModel(Some(childRelationIdentifier.getDatabaseName),
          childRelationIdentifier.getTableName,
          None,
          compactionType.toString,
          Some(System.currentTimeMillis()),
          "")
        operationContext.setProperty(
          dataMapSchema.getRelationIdentifier.getDatabaseName + "_" +
          dataMapSchema.getRelationIdentifier.getTableName + "_Segment",
          carbonLoadModel.getSegmentId)
        CarbonAlterTableCompactionCommand(alterTableModel, operationContext = operationContext)
          .run(sparkSession)
      }
    }
  }
}

/**
 * Listener to trigger data load on mv datamap after main table data load
 */
object LoadPostDataMapListener extends OperationEventListener {
  /**
   * Called on a specified event occurrence
   *
   * @param event
   * @param operationContext
   */
  override def onEvent(event: Event, operationContext: OperationContext): Unit = {
    val sparkSession = SparkSession.getActiveSession.get
    val carbonTable: CarbonTable =
      event match {
        case event: LoadTablePostExecutionEvent =>
          val carbonLoadModelOption = Some(event.getCarbonLoadModel)
          if (carbonLoadModelOption.isDefined) {
            val carbonLoadModel = carbonLoadModelOption.get
            carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable
          } else {
            null
          }
        case event: UpdateTablePostEvent =>
          val table = Some(event.carbonTable)
          if (table.isDefined) {
            table.get
          } else {
            null
          }
        case event: DeleteFromTablePostEvent =>
          val table = Some(event.carbonTable)
          if (table.isDefined) {
            table.get
          } else {
            null
          }
        case _ => null
      }
    if (null != carbonTable) {
      val allDataMapSchemas = DataMapStoreManager.getInstance
        .getDataMapSchemasOfTable(carbonTable).asScala
        .filter(dataMapSchema => null != dataMapSchema.getRelationIdentifier &&
                                 !dataMapSchema.isIndexDataMap)
      if (!allDataMapSchemas.asJava.isEmpty) {
        allDataMapSchemas.foreach { dataMapSchema =>
          if (!dataMapSchema.isLazy) {
            val provider = DataMapManager.get()
              .getDataMapProvider(carbonTable, dataMapSchema, sparkSession)
            try {
              provider.rebuild()
              DataMapStatusManager.enableDataMap(dataMapSchema.getDataMapName)
            } catch {
              case ex: Exception =>
                DataMapStatusManager.disableDataMap(dataMapSchema.getDataMapName)
            }
          }
        }
      }
    }
  }
}

/**
 * Listeners to block operations like delete segment on id or by date on tables
 * having an mv datamap or on mv datamap tables
 */
object DataMapDeleteSegmentPreListener extends OperationEventListener {
  /**
   * Called on a specified event occurrence
   *
   * @param event
   * @param operationContext
   */
  override def onEvent(event: Event, operationContext: OperationContext): Unit = {
    val carbonTable = event match {
      case e: DeleteSegmentByIdPreEvent =>
        e.asInstanceOf[DeleteSegmentByIdPreEvent].carbonTable
      case e: DeleteSegmentByDatePreEvent =>
        e.asInstanceOf[DeleteSegmentByDatePreEvent].carbonTable
    }
    if (null != carbonTable) {
      if (DataMapUtil.hasMVDataMap(carbonTable)) {
        throw new UnsupportedOperationException(
          "Delete segment operation is not supported on tables having child datamap")
      }
      if (carbonTable.isChildTable) {
        throw new UnsupportedOperationException(
          "Delete segment operation is not supported on datamap table")
      }
    }
  }
}

object DataMapAddColumnsPreListener 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.isChildTable) {
      throw new UnsupportedOperationException(
        s"Cannot add columns in a DataMap table " +
        s"${ carbonTable.getDatabaseName }.${ carbonTable.getTableName }")
    }
  }
}


object DataMapDropColumnPreListener extends OperationEventListener {
  /**
   * Called on a specified event occurrence
   *
   * @param event
   * @param operationContext
   */
  override def onEvent(event: Event, operationContext: OperationContext): Unit = {
    val dropColumnChangePreListener = event.asInstanceOf[AlterTableDropColumnPreEvent]
    val carbonTable = dropColumnChangePreListener.carbonTable
    val alterTableDropColumnModel = dropColumnChangePreListener.alterTableDropColumnModel
    val columnsToBeDropped = alterTableDropColumnModel.columns
    if (DataMapUtil.hasMVDataMap(carbonTable)) {
      val dataMapSchemaList = DataMapStoreManager.getInstance
        .getDataMapSchemasOfTable(carbonTable).asScala
      for (dataMapSchema <- dataMapSchemaList) {
        if (null != dataMapSchema && !dataMapSchema.isIndexDataMap) {
          val listOfColumns = DataMapListeners.getDataMapTableColumns(dataMapSchema, carbonTable)
          val columnExistsInChild = listOfColumns.collectFirst {
            case parentColumnName if columnsToBeDropped.contains(parentColumnName) =>
              parentColumnName
          }
          if (columnExistsInChild.isDefined) {
            throw new UnsupportedOperationException(
              s"Column ${ columnExistsInChild.head } cannot be dropped because it exists " +
              s"in " + dataMapSchema.getProviderName + " datamap:" +
              s"${ dataMapSchema.getDataMapName }")
          }
        }
      }
    }
    if (carbonTable.isChildTable) {
      throw new UnsupportedOperationException(
        s"Cannot drop columns present in a datamap table ${ carbonTable.getDatabaseName }." +
        s"${ carbonTable.getTableName }")
    }
  }
}

object DataMapChangeDataTypeorRenameColumnPreListener
  extends OperationEventListener {
  /**
   * Called on a specified event occurrence
   *
   * @param event
   * @param operationContext
   */
  override def onEvent(event: Event, operationContext: OperationContext): Unit = {
    val colRenameDataTypeChangePreListener = event
      .asInstanceOf[AlterTableColRenameAndDataTypeChangePreEvent]
    val carbonTable = colRenameDataTypeChangePreListener.carbonTable
    val alterTableDataTypeChangeModel = colRenameDataTypeChangePreListener
      .alterTableDataTypeChangeModel
    val columnToBeAltered: String = alterTableDataTypeChangeModel.columnName
    if (DataMapUtil.hasMVDataMap(carbonTable)) {
      val dataMapSchemaList = DataMapStoreManager.getInstance
        .getDataMapSchemasOfTable(carbonTable).asScala
      for (dataMapSchema <- dataMapSchemaList) {
        if (null != dataMapSchema && !dataMapSchema.isIndexDataMap) {
          val listOfColumns = DataMapListeners.getDataMapTableColumns(dataMapSchema, carbonTable)
          if (listOfColumns.contains(columnToBeAltered)) {
            throw new UnsupportedOperationException(
              s"Column $columnToBeAltered exists in a " + dataMapSchema.getProviderName +
              " datamap. Drop " + dataMapSchema.getProviderName + "  datamap to continue")
          }
        }
      }
    }
    if (carbonTable.isChildTable) {
      throw new UnsupportedOperationException(
        s"Cannot change data type or rename column for columns present in mv datamap table " +
        s"${ carbonTable.getDatabaseName }.${ carbonTable.getTableName }")
    }
  }
}

object DataMapAlterTableDropPartitionMetaListener extends OperationEventListener {
  /**
   * Called on a specified event occurrence
   *
   * @param event
   * @param operationContext
   */
  override def onEvent(event: Event, operationContext: OperationContext): Unit = {
    val dropPartitionEvent = event.asInstanceOf[AlterTableDropPartitionMetaEvent]
    val parentCarbonTable = dropPartitionEvent.parentCarbonTable
    val partitionsToBeDropped = dropPartitionEvent.specs.flatMap(_.keys)
    if (DataMapUtil.hasMVDataMap(parentCarbonTable)) {
      // used as a flag to block direct drop partition on datamap tables fired by the user
      operationContext.setProperty("isInternalDropCall", "true")
      // Filter out all the tables which don't have the partition being dropped.
      val dataMapSchemaList = DataMapStoreManager.getInstance
        .getDataMapSchemasOfTable(parentCarbonTable).asScala
      val childTablesWithoutPartitionColumns =
        dataMapSchemaList.filter { dataMapSchema =>
          val childColumns = dataMapSchema.getMainTableColumnList
            .get(parentCarbonTable.getTableName).asScala
          val partitionColExists =
            partitionsToBeDropped.forall {
              partition =>
                childColumns.exists { childColumn =>
                  childColumn.equalsIgnoreCase(partition)
                }
            }
          !partitionColExists
        }
      if (childTablesWithoutPartitionColumns.nonEmpty) {
        throw new MetadataProcessException(s"Cannot drop partition as one of the partition is not" +
                                           s" participating in the following datamaps ${
                                             childTablesWithoutPartitionColumns.toList
                                               .map(_.getRelationIdentifier.getTableName)
                                           }. Please drop the specified child tables to " +
                                           s"continue")
      } else {
        // blocked drop partition for child tables having more than one parent table
        val nonPartitionChildTables = dataMapSchemaList.filter(_.getParentTables.size() >= 2)
        if (nonPartitionChildTables.nonEmpty) {
          throw new MetadataProcessException(
            s"Cannot drop partition if child Table is mapped to more than one parent table. Drop " +
            s"datamaps ${ nonPartitionChildTables.toList.map(_.getDataMapName) }  to continue")
        }
        val childDropPartitionCommands =
          dataMapSchemaList.map { dataMapSchema =>
            val tableIdentifier = TableIdentifier(dataMapSchema.getRelationIdentifier.getTableName,
              Some(dataMapSchema.getRelationIdentifier.getDatabaseName))
            if (!CarbonEnv.getCarbonTable(tableIdentifier)(SparkSession.getActiveSession.get)
              .isHivePartitionTable) {
              throw new MetadataProcessException(
                "Cannot drop partition as one of the partition is not participating in the " +
                "following datamap " + dataMapSchema.getDataMapName +
                ". Please drop the specified datamap to continue")
            }
            // as the datamap table columns start with parent table name therefore the
            // partition column also has to be updated with parent table name to generate
            // partitionSpecs for the child table.
            val childSpecs = dropPartitionEvent.specs.map {
              spec =>
                spec.map {
                  case (key, value) => (s"${ parentCarbonTable.getTableName }_$key", value)
                }
            }
            CarbonAlterTableDropHivePartitionCommand(
              tableIdentifier,
              childSpecs,
              dropPartitionEvent.ifExists,
              dropPartitionEvent.purge,
              dropPartitionEvent.retainData,
              operationContext)
          }
        operationContext.setProperty("dropPartitionCommands", childDropPartitionCommands)
        childDropPartitionCommands.foreach(_.processMetadata(SparkSession.getActiveSession.get))
      }
    } else if (parentCarbonTable.isChildTable) {
      if (operationContext.getProperty("isInternalDropCall") == null) {
        throw new UnsupportedOperationException("Cannot drop partition directly on child table")
      }
    }
  }
}

object DataMapAlterTableDropPartitionPreStatusListener extends OperationEventListener {
  /**
   * Called on a specified event occurrence
   *
   * @param event
   * @param operationContext
   */
  override protected def onEvent(event: Event,
      operationContext: OperationContext) = {
    val preStatusListener = event.asInstanceOf[AlterTableDropPartitionPreStatusEvent]
    val carbonTable = preStatusListener.carbonTable
    val childDropPartitionCommands = operationContext.getProperty("dropPartitionCommands")
    if (childDropPartitionCommands != null && DataMapUtil.hasMVDataMap(carbonTable)) {
      val childCommands =
        childDropPartitionCommands.asInstanceOf[Seq[CarbonAlterTableDropHivePartitionCommand]]
      childCommands.foreach(_.processData(SparkSession.getActiveSession.get))
    }
  }
}


1	/*
2	* Licensed to the Apache Software Foundation (ASF) under one or more
3	* contributor license agreements. See the NOTICE file distributed with
4	* this work for additional information regarding copyright ownership.
5	* The ASF licenses this file to You under the Apache License, Version 2.0
6	* (the "License"); you may not use this file except in compliance with
7	* the License. You may obtain a copy of the License at
8	*
9	* http://www.apache.org/licenses/LICENSE-2.0
10	*
11	* Unless required by applicable law or agreed to in writing, software
12	* distributed under the License is distributed on an "AS IS" BASIS,
13	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14	* See the License for the specific language governing permissions and
15	* limitations under the License.
16	*/
17
18	package org.apache.spark.sql.execution.command.mv
19
20	import scala.collection.JavaConverters._
21	import scala.collection.mutable
22
23	import org.apache.spark.sql.{CarbonEnv, SparkSession}
24	import org.apache.spark.sql.catalyst.TableIdentifier
25	import org.apache.spark.sql.execution.command.AlterTableModel
26	import org.apache.spark.sql.execution.command.management.CarbonAlterTableCompactionCommand
27	import org.apache.spark.sql.execution.command.partition.CarbonAlterTableDropHivePartitionCommand
28	import org.apache.spark.util.DataMapUtil
29
30	import org.apache.carbondata.common.exceptions.MetadataProcessException
31	import org.apache.carbondata.core.datamap.DataMapStoreManager
32	import org.apache.carbondata.core.datamap.status.DataMapStatusManager
33	import org.apache.carbondata.core.metadata.schema.table.{CarbonTable, DataMapSchema}
34	import org.apache.carbondata.datamap.DataMapManager
35	import org.apache.carbondata.events._
36	import org.apache.carbondata.processing.loading.events.LoadEvents.LoadTablePostExecutionEvent
37	import org.apache.carbondata.processing.merger.CompactionType
38
39
40	object DataMapListeners {
41	def getDataMapTableColumns(dataMapSchema: DataMapSchema,
42	carbonTable: CarbonTable): mutable.Buffer[String] = {
43	val listOfColumns: mutable.Buffer[String] = new mutable.ArrayBuffer[String]()	×
44	listOfColumns.asJava	×
45	.addAll(dataMapSchema.getMainTableColumnList.get(carbonTable.getTableName))	×
46	listOfColumns	×
47	}
48	}
49
50	/**
51	* Listener to trigger compaction on mv datamap after main table compaction
52	*/
53	object AlterDataMaptableCompactionPostListener extends OperationEventListener {	2✔
54	/**
55	* Called on a specified event occurrence
56	*
57	* @param event
58	* @param operationContext
59	*/
60	override def onEvent(event: Event, operationContext: OperationContext): Unit = {	2✔
61	val compactionEvent = event.asInstanceOf[AlterTableCompactionPreStatusUpdateEvent]	2✔
62	val carbonTable = compactionEvent.carbonTable	2✔
63	val compactionType = compactionEvent.carbonMergerMapping.campactionType	2✔
64	if (compactionType == CompactionType.CUSTOM) {	2✔
65	return	1✔
66	}
67	val carbonLoadModel = compactionEvent.carbonLoadModel	2✔
68	val sparkSession = compactionEvent.sparkSession	2✔
69	val allDataMapSchemas = DataMapStoreManager.getInstance	2✔
70	.getDataMapSchemasOfTable(carbonTable).asScala	2✔
71	.filter(dataMapSchema => null != dataMapSchema.getRelationIdentifier &&	2✔
72	!dataMapSchema.isIndexDataMap)	1✔
73	if (!allDataMapSchemas.asJava.isEmpty) {	2✔
74	allDataMapSchemas.foreach { dataMapSchema =>	×
75	val childRelationIdentifier = dataMapSchema.getRelationIdentifier	×
76	val alterTableModel = AlterTableModel(Some(childRelationIdentifier.getDatabaseName),	×
77	childRelationIdentifier.getTableName,	×
78	None,	×
79	compactionType.toString,	×
80	Some(System.currentTimeMillis()),	×
81	"")	×
82	operationContext.setProperty(	×
83	dataMapSchema.getRelationIdentifier.getDatabaseName + "_" +	×
84	dataMapSchema.getRelationIdentifier.getTableName + "_Segment",	×
85	carbonLoadModel.getSegmentId)	×
86	CarbonAlterTableCompactionCommand(alterTableModel, operationContext = operationContext)	×
87	.run(sparkSession)	×
88	}
89	}
90	}
91	}
92
93	/**
94	* Listener to trigger data load on mv datamap after main table data load
95	*/
96	object LoadPostDataMapListener extends OperationEventListener {	2✔
97	/**
98	* Called on a specified event occurrence
99	*
100	* @param event
101	* @param operationContext
102	*/
103	override def onEvent(event: Event, operationContext: OperationContext): Unit = {	2✔
104	val sparkSession = SparkSession.getActiveSession.get	2✔
105	val carbonTable: CarbonTable =	2✔
106	event match {	2✔
107	case event: LoadTablePostExecutionEvent =>	2✔
108	val carbonLoadModelOption = Some(event.getCarbonLoadModel)	2✔
109	if (carbonLoadModelOption.isDefined) {	2✔
110	val carbonLoadModel = carbonLoadModelOption.get	2✔
111	carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable	2✔
112	} else {
113	null	×
114	}
115	case event: UpdateTablePostEvent =>	2✔
116	val table = Some(event.carbonTable)	2✔
117	if (table.isDefined) {	2✔
118	table.get	2✔
119	} else {
120	null	×
121	}
122	case event: DeleteFromTablePostEvent =>	2✔
123	val table = Some(event.carbonTable)	2✔
124	if (table.isDefined) {	2✔
125	table.get	2✔
126	} else {
127	null	×
128	}
129	case _ => null	×
130	}
131	if (null != carbonTable) {	2✔
132	val allDataMapSchemas = DataMapStoreManager.getInstance	2✔
133	.getDataMapSchemasOfTable(carbonTable).asScala	2✔
134	.filter(dataMapSchema => null != dataMapSchema.getRelationIdentifier &&	2✔
135	!dataMapSchema.isIndexDataMap)	2✔
136	if (!allDataMapSchemas.asJava.isEmpty) {	2✔
137	allDataMapSchemas.foreach { dataMapSchema =>	×
138	if (!dataMapSchema.isLazy) {	×
139	val provider = DataMapManager.get()	×
140	.getDataMapProvider(carbonTable, dataMapSchema, sparkSession)	×
141	try {	×
142	provider.rebuild()	×
143	DataMapStatusManager.enableDataMap(dataMapSchema.getDataMapName)	×
144	} catch {
145	case ex: Exception =>
146	DataMapStatusManager.disableDataMap(dataMapSchema.getDataMapName)	×
147	}
148	}
149	}
150	}
151	}
152	}
153	}
154
155	/**
156	* Listeners to block operations like delete segment on id or by date on tables
157	* having an mv datamap or on mv datamap tables
158	*/
159	object DataMapDeleteSegmentPreListener extends OperationEventListener {	2✔
160	/**
161	* Called on a specified event occurrence
162	*
163	* @param event
164	* @param operationContext
165	*/
166	override def onEvent(event: Event, operationContext: OperationContext): Unit = {	2✔
167	val carbonTable = event match {	2✔
168	case e: DeleteSegmentByIdPreEvent =>	2✔
169	e.asInstanceOf[DeleteSegmentByIdPreEvent].carbonTable	2✔
170	case e: DeleteSegmentByDatePreEvent =>	2✔
171	e.asInstanceOf[DeleteSegmentByDatePreEvent].carbonTable	2✔
172	}
173	if (null != carbonTable) {	2✔
174	if (DataMapUtil.hasMVDataMap(carbonTable)) {	2✔
175	throw new UnsupportedOperationException(	×
176	"Delete segment operation is not supported on tables having child datamap")	×
177	}
178	if (carbonTable.isChildTable) {	2✔
179	throw new UnsupportedOperationException(	×
180	"Delete segment operation is not supported on datamap table")	×
181	}
182	}
183	}
184	}
185
186	object DataMapAddColumnsPreListener extends OperationEventListener {	2✔
187	/**
188	* Called on a specified event occurrence
189	*
190	* @param event
191	* @param operationContext
192	*/
193	override def onEvent(event: Event, operationContext: OperationContext): Unit = {	2✔
194	val dataTypeChangePreListener = event.asInstanceOf[AlterTableAddColumnPreEvent]	2✔
195	val carbonTable = dataTypeChangePreListener.carbonTable	2✔
196	if (carbonTable.isChildTable) {	2✔
197	throw new UnsupportedOperationException(	×
198	s"Cannot add columns in a DataMap table " +	×
199	s"${ carbonTable.getDatabaseName }.${ carbonTable.getTableName }")	×
200	}
201	}
202	}
203
204
205	object DataMapDropColumnPreListener extends OperationEventListener {	2✔
206	/**
207	* Called on a specified event occurrence
208	*
209	* @param event
210	* @param operationContext
211	*/
212	override def onEvent(event: Event, operationContext: OperationContext): Unit = {	2✔
213	val dropColumnChangePreListener = event.asInstanceOf[AlterTableDropColumnPreEvent]	2✔
214	val carbonTable = dropColumnChangePreListener.carbonTable	2✔
215	val alterTableDropColumnModel = dropColumnChangePreListener.alterTableDropColumnModel	2✔
216	val columnsToBeDropped = alterTableDropColumnModel.columns	2✔
217	if (DataMapUtil.hasMVDataMap(carbonTable)) {	2✔
218	val dataMapSchemaList = DataMapStoreManager.getInstance	×
219	.getDataMapSchemasOfTable(carbonTable).asScala	×
220	for (dataMapSchema <- dataMapSchemaList) {	×
221	if (null != dataMapSchema && !dataMapSchema.isIndexDataMap) {	×
222	val listOfColumns = DataMapListeners.getDataMapTableColumns(dataMapSchema, carbonTable)	×
223	val columnExistsInChild = listOfColumns.collectFirst {	×
224	case parentColumnName if columnsToBeDropped.contains(parentColumnName) =>	×
225	parentColumnName	×
226	}
227	if (columnExistsInChild.isDefined) {	×
228	throw new UnsupportedOperationException(	×
229	s"Column ${ columnExistsInChild.head } cannot be dropped because it exists " +	×
230	s"in " + dataMapSchema.getProviderName + " datamap:" +	×
231	s"${ dataMapSchema.getDataMapName }")	×
232	}
233	}
234	}
235	}
236	if (carbonTable.isChildTable) {	2✔
237	throw new UnsupportedOperationException(	×
238	s"Cannot drop columns present in a datamap table ${ carbonTable.getDatabaseName }." +	×
239	s"${ carbonTable.getTableName }")	×
240	}
241	}
242	}
243
244	object DataMapChangeDataTypeorRenameColumnPreListener
245	extends OperationEventListener {	2✔
246	/**
247	* Called on a specified event occurrence
248	*
249	* @param event
250	* @param operationContext
251	*/
252	override def onEvent(event: Event, operationContext: OperationContext): Unit = {	2✔
253	val colRenameDataTypeChangePreListener = event	2✔
254	.asInstanceOf[AlterTableColRenameAndDataTypeChangePreEvent]
255	val carbonTable = colRenameDataTypeChangePreListener.carbonTable	2✔
256	val alterTableDataTypeChangeModel = colRenameDataTypeChangePreListener	2✔
257	.alterTableDataTypeChangeModel	2✔
258	val columnToBeAltered: String = alterTableDataTypeChangeModel.columnName	2✔
259	if (DataMapUtil.hasMVDataMap(carbonTable)) {	2✔
260	val dataMapSchemaList = DataMapStoreManager.getInstance	×
261	.getDataMapSchemasOfTable(carbonTable).asScala	×
262	for (dataMapSchema <- dataMapSchemaList) {	×
263	if (null != dataMapSchema && !dataMapSchema.isIndexDataMap) {	×
264	val listOfColumns = DataMapListeners.getDataMapTableColumns(dataMapSchema, carbonTable)	×
265	if (listOfColumns.contains(columnToBeAltered)) {	×
266	throw new UnsupportedOperationException(	×
267	s"Column $columnToBeAltered exists in a " + dataMapSchema.getProviderName +	×
268	" datamap. Drop " + dataMapSchema.getProviderName + " datamap to continue")	×
269	}
270	}
271	}
272	}
273	if (carbonTable.isChildTable) {	2✔
274	throw new UnsupportedOperationException(	×
275	s"Cannot change data type or rename column for columns present in mv datamap table " +	×
276	s"${ carbonTable.getDatabaseName }.${ carbonTable.getTableName }")	×
277	}
278	}
279	}
280
281	object DataMapAlterTableDropPartitionMetaListener extends OperationEventListener {	2✔
282	/**
283	* Called on a specified event occurrence
284	*
285	* @param event
286	* @param operationContext
287	*/
288	override def onEvent(event: Event, operationContext: OperationContext): Unit = {	1✔
289	val dropPartitionEvent = event.asInstanceOf[AlterTableDropPartitionMetaEvent]	1✔
290	val parentCarbonTable = dropPartitionEvent.parentCarbonTable	1✔
291	val partitionsToBeDropped = dropPartitionEvent.specs.flatMap(_.keys)	1✔
292	if (DataMapUtil.hasMVDataMap(parentCarbonTable)) {	1✔
293	// used as a flag to block direct drop partition on datamap tables fired by the user
294	operationContext.setProperty("isInternalDropCall", "true")	×
295	// Filter out all the tables which don't have the partition being dropped.
296	val dataMapSchemaList = DataMapStoreManager.getInstance	×
297	.getDataMapSchemasOfTable(parentCarbonTable).asScala	×
298	val childTablesWithoutPartitionColumns =	×
299	dataMapSchemaList.filter { dataMapSchema =>	×
300	val childColumns = dataMapSchema.getMainTableColumnList	×
301	.get(parentCarbonTable.getTableName).asScala	×
302	val partitionColExists =	×
303	partitionsToBeDropped.forall {	×
304	partition =>	×
305	childColumns.exists { childColumn =>	×
306	childColumn.equalsIgnoreCase(partition)	×
307	}
308	}
309	!partitionColExists	×
310	}
311	if (childTablesWithoutPartitionColumns.nonEmpty) {	×
312	throw new MetadataProcessException(s"Cannot drop partition as one of the partition is not" +	×
313	s" participating in the following datamaps ${	×
314	childTablesWithoutPartitionColumns.toList	×
315	.map(_.getRelationIdentifier.getTableName)	×
316	}. Please drop the specified child tables to " +	×
317	s"continue")	×
318	} else {
319	// blocked drop partition for child tables having more than one parent table
320	val nonPartitionChildTables = dataMapSchemaList.filter(_.getParentTables.size() >= 2)	×
321	if (nonPartitionChildTables.nonEmpty) {	×
322	throw new MetadataProcessException(	×
323	s"Cannot drop partition if child Table is mapped to more than one parent table. Drop " +	×
324	s"datamaps ${ nonPartitionChildTables.toList.map(_.getDataMapName) } to continue")	×
325	}
326	val childDropPartitionCommands =	×
327	dataMapSchemaList.map { dataMapSchema =>	×
328	val tableIdentifier = TableIdentifier(dataMapSchema.getRelationIdentifier.getTableName,	×
329	Some(dataMapSchema.getRelationIdentifier.getDatabaseName))	×
330	if (!CarbonEnv.getCarbonTable(tableIdentifier)(SparkSession.getActiveSession.get)	×
331	.isHivePartitionTable) {	×
332	throw new MetadataProcessException(	×
333	"Cannot drop partition as one of the partition is not participating in the " +	×
334	"following datamap " + dataMapSchema.getDataMapName +	×
335	". Please drop the specified datamap to continue")	×
336	}
337	// as the datamap table columns start with parent table name therefore the
338	// partition column also has to be updated with parent table name to generate
339	// partitionSpecs for the child table.
340	val childSpecs = dropPartitionEvent.specs.map {	×
341	spec =>	×
342	spec.map {	×
343	case (key, value) => (s"${ parentCarbonTable.getTableName }_$key", value)	×
344	}
345	}
346	CarbonAlterTableDropHivePartitionCommand(	×
347	tableIdentifier,	×
348	childSpecs,	×
349	dropPartitionEvent.ifExists,	×
350	dropPartitionEvent.purge,	×
351	dropPartitionEvent.retainData,	×
352	operationContext)	×
353	}
354	operationContext.setProperty("dropPartitionCommands", childDropPartitionCommands)	×
355	childDropPartitionCommands.foreach(_.processMetadata(SparkSession.getActiveSession.get))	×
356	}
357	} else if (parentCarbonTable.isChildTable) {	1✔
358	if (operationContext.getProperty("isInternalDropCall") == null) {	×
359	throw new UnsupportedOperationException("Cannot drop partition directly on child table")	×
360	}
361	}
362	}
363	}
364
365	object DataMapAlterTableDropPartitionPreStatusListener extends OperationEventListener {	2✔
366	/**
367	* Called on a specified event occurrence
368	*
369	* @param event
370	* @param operationContext
371	*/
372	override protected def onEvent(event: Event,
373	operationContext: OperationContext) = {	1✔
374	val preStatusListener = event.asInstanceOf[AlterTableDropPartitionPreStatusEvent]	1✔
375	val carbonTable = preStatusListener.carbonTable	1✔
376	val childDropPartitionCommands = operationContext.getProperty("dropPartitionCommands")	1✔
377	if (childDropPartitionCommands != null && DataMapUtil.hasMVDataMap(carbonTable)) {	1✔
378	val childCommands =	×
379	childDropPartitionCommands.asInstanceOf[Seq[CarbonAlterTableDropHivePartitionCommand]]	×
380	childCommands.foreach(_.processData(SparkSession.getActiveSession.get))	×
381	}
382	}
383	}
384

apache / carbondata / 11623

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous