10814

Build # 10814

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/131bcb51a0f376e82e1c15147c54914a131c0acc">131bcb51a<a href="https://github.com/apache/carbondata/commit/131bcb51a0f376e82e1c15147c54914a131c0acc">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/d995499ebef9104eb6bf838afb63f8905acc40f9">d995499eb</a>">d995499eb</a><a href="https://github.com/apache/carbondata/commit/131bcb51a0f376e82e1c15147c54914a131c0acc"> into 2ecf30ca0">2ecf30ca0</a>

Run Details

24 of 24 new or added lines in 3 files covered. (100.0%)

62460 of 78863 relevant lines covered (79.2%)

1.05 hits per line

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

93.75

/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/schema/CarbonAlterTableRenameCommand.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.schema

import java.util

import org.apache.spark.sql.{CarbonEnv, SparkSession}
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.catalog.CatalogTablePartition
import org.apache.spark.sql.execution.command.{AlterTableRenameModel, MetadataCommand}
import org.apache.spark.sql.hive.{CarbonRelation, CarbonSessionCatalog}
import org.apache.spark.util.AlterTableUtil

import org.apache.carbondata.common.exceptions.sql.MalformedCarbonCommandException
import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.datamap.DataMapStoreManager
import org.apache.carbondata.core.exception.ConcurrentOperationException
import org.apache.carbondata.core.features.TableOperation
import org.apache.carbondata.core.metadata.CarbonTableIdentifier
import org.apache.carbondata.core.metadata.schema.table.{CarbonTable, DataMapSchema}
import org.apache.carbondata.core.statusmanager.SegmentStatusManager
import org.apache.carbondata.events.{AlterTableRenamePostEvent, AlterTableRenamePreEvent, OperationContext, OperationListenerBus}
import org.apache.carbondata.format.SchemaEvolutionEntry

private[sql] case class CarbonAlterTableRenameCommand(
    alterTableRenameModel: AlterTableRenameModel)
  extends MetadataCommand {

  override def processMetadata(sparkSession: SparkSession): Seq[Nothing] = {
    val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)
    val oldTableName = alterTableRenameModel.oldTableIdentifier.table.toLowerCase
    val newTableName = alterTableRenameModel.newTableIdentifier.table.toLowerCase
    val oldDatabaseName = alterTableRenameModel.oldTableIdentifier.database
      .getOrElse(sparkSession.catalog.currentDatabase)
    val oldTableIdentifier = TableIdentifier(oldTableName, Some(oldDatabaseName))
    val newTableIdentifier = TableIdentifier(newTableName, Some(oldDatabaseName))
    setAuditTable(oldDatabaseName, oldTableIdentifier.table)
    setAuditInfo(Map("newName" -> alterTableRenameModel.newTableIdentifier.table))
    val newDatabaseName = newTableIdentifier.database
      .getOrElse(sparkSession.catalog.currentDatabase)
    if (!oldDatabaseName.equalsIgnoreCase(newDatabaseName)) {
      throw new MalformedCarbonCommandException("Database name should be same for both tables")
    }
    val tableExists = sparkSession.catalog.tableExists(oldDatabaseName, newTableIdentifier.table)
    if (tableExists) {
      throw new MalformedCarbonCommandException(s"Table with name $newTableIdentifier " +
                                                s"already exists")
    }
    LOGGER.info(s"Rename table request has been received for $oldDatabaseName.$oldTableName")
    val metastore = CarbonEnv.getInstance(sparkSession).carbonMetaStore
    val relation: CarbonRelation =
      metastore.lookupRelation(oldTableIdentifier.database, oldTableName)(sparkSession)
        .asInstanceOf[CarbonRelation]
    if (relation == null) {
      throwMetadataException(oldDatabaseName, oldTableName, "Table does not exist")
    }

    var oldCarbonTable: CarbonTable = null
    oldCarbonTable = metastore.lookupRelation(Some(oldDatabaseName), oldTableName)(sparkSession)
      .asInstanceOf[CarbonRelation].carbonTable
    if (!oldCarbonTable.getTableInfo.isTransactionalTable) {
      throw new MalformedCarbonCommandException("Unsupported operation on non transactional table")
    }

    if (!oldCarbonTable.canAllow(oldCarbonTable, TableOperation.ALTER_RENAME)) {
      throw new MalformedCarbonCommandException("alter rename is not supported for index datamap")
    }
    // if table have create mv datamap, not support table rename
    if (CarbonTable.hasMVDataMap(oldCarbonTable)) {
      throw new MalformedCarbonCommandException("alter rename is not supported for mv datamap")
    }

    var timeStamp = 0L
    var carbonTable: CarbonTable = null
    var hiveRenameSuccess = false
    // lock file path to release locks after operation
    var carbonTableLockFilePath: String = null
    try {
      carbonTable = metastore.lookupRelation(Some(oldDatabaseName), oldTableName)(sparkSession)
        .asInstanceOf[CarbonRelation].carbonTable
      carbonTableLockFilePath = carbonTable.getTablePath
      // if any load is in progress for table, do not allow rename table
      if (SegmentStatusManager.isLoadInProgressInTable(carbonTable)) {
        throw new ConcurrentOperationException(carbonTable, "loading", "alter table rename")
      }
      // get the old table all data map schema
      val dataMapSchemaList: util.List[DataMapSchema] = new util.ArrayList[DataMapSchema]()
      if (carbonTable.hasDataMapSchema) {
        dataMapSchemaList.addAll(carbonTable.getTableInfo.getDataMapSchemaList)
      }
      val indexSchemas = DataMapStoreManager.getInstance().getDataMapSchemasOfTable(carbonTable)
      if (!indexSchemas.isEmpty) {
        dataMapSchemaList.addAll(indexSchemas)
      }
      // invalid data map for the old table, see CARBON-1690
      val oldAbsoluteTableIdentifier = carbonTable.getAbsoluteTableIdentifier
      DataMapStoreManager.getInstance().clearDataMaps(oldAbsoluteTableIdentifier)
      // get the latest carbon table and check for column existence
      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(carbonTable)
      val schemaEvolutionEntry = new SchemaEvolutionEntry(System.currentTimeMillis)
      schemaEvolutionEntry.setTableName(newTableName)
      timeStamp = System.currentTimeMillis()
      schemaEvolutionEntry.setTime_stamp(timeStamp)
      val newCarbonTableIdentifier = new CarbonTableIdentifier(oldDatabaseName,
        newTableName, carbonTable.getCarbonTableIdentifier.getTableId)
      metastore.removeTableFromMetadata(oldDatabaseName, oldTableName)
      var partitions: Seq[CatalogTablePartition] = Seq.empty
      if (carbonTable.isHivePartitionTable) {
        partitions =
          sparkSession.sessionState.catalog.listPartitions(oldTableIdentifier)
      }
      sparkSession.catalog.refreshTable(oldTableIdentifier.quotedString)
      sparkSession.sessionState.catalog.asInstanceOf[CarbonSessionCatalog].alterTableRename(
          oldTableIdentifier,
          newTableIdentifier,
        oldAbsoluteTableIdentifier.getTablePath)
      hiveRenameSuccess = true

      metastore.updateTableSchemaForAlter(
        newCarbonTableIdentifier,
        carbonTable.getCarbonTableIdentifier,
        tableInfo,
        schemaEvolutionEntry,
        carbonTable.getTablePath)(sparkSession)

      // Update the storage location with datamap schema
      if (!dataMapSchemaList.isEmpty) {
        DataMapStoreManager.getInstance().
          updateDataMapSchema(dataMapSchemaList, newTableName)
      }
      val alterTableRenamePostEvent: AlterTableRenamePostEvent = AlterTableRenamePostEvent(
        carbonTable,
        alterTableRenameModel,
        oldAbsoluteTableIdentifier.getTablePath,
        sparkSession)
      OperationListenerBus.getInstance().fireEvent(alterTableRenamePostEvent, operationContext)

      sparkSession.catalog.refreshTable(newTableIdentifier.quotedString)
      LOGGER.info(s"Table $oldTableName has been successfully renamed to $newTableName")
    } catch {
      case e: ConcurrentOperationException =>
        throw e
      case e: Exception =>
        if (hiveRenameSuccess) {
          sparkSession.sessionState.catalog.asInstanceOf[CarbonSessionCatalog].alterTableRename(
            newTableIdentifier,
            oldTableIdentifier,
            carbonTable.getAbsoluteTableIdentifier.getTablePath)
        }
        if (carbonTable != null) {
          AlterTableUtil.revertRenameTableChanges(
            newTableName,
            carbonTable,
            timeStamp)(
            sparkSession)
        }
        throwMetadataException(oldDatabaseName, oldTableName,
          opName + " operation failed: " + e.getMessage)
    }
    Seq.empty
  }

  override protected def opName: String = "ALTER TABLE RENAME TABLE"
}

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.schema
19
20	import java.util
21
22	import org.apache.spark.sql.{CarbonEnv, SparkSession}
23	import org.apache.spark.sql.catalyst.TableIdentifier
24	import org.apache.spark.sql.catalyst.catalog.CatalogTablePartition
25	import org.apache.spark.sql.execution.command.{AlterTableRenameModel, MetadataCommand}
26	import org.apache.spark.sql.hive.{CarbonRelation, CarbonSessionCatalog}
27	import org.apache.spark.util.AlterTableUtil
28
29	import org.apache.carbondata.common.exceptions.sql.MalformedCarbonCommandException
30	import org.apache.carbondata.common.logging.LogServiceFactory
31	import org.apache.carbondata.core.datamap.DataMapStoreManager
32	import org.apache.carbondata.core.exception.ConcurrentOperationException
33	import org.apache.carbondata.core.features.TableOperation
34	import org.apache.carbondata.core.metadata.CarbonTableIdentifier
35	import org.apache.carbondata.core.metadata.schema.table.{CarbonTable, DataMapSchema}
36	import org.apache.carbondata.core.statusmanager.SegmentStatusManager
37	import org.apache.carbondata.events.{AlterTableRenamePostEvent, AlterTableRenamePreEvent, OperationContext, OperationListenerBus}
38	import org.apache.carbondata.format.SchemaEvolutionEntry
39
40	private[sql] case class CarbonAlterTableRenameCommand(	2✔
41	alterTableRenameModel: AlterTableRenameModel)	2✔
42	extends MetadataCommand {	2✔
43
44	override def processMetadata(sparkSession: SparkSession): Seq[Nothing] = {
45	val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)	2✔
46	val oldTableName = alterTableRenameModel.oldTableIdentifier.table.toLowerCase	2✔
47	val newTableName = alterTableRenameModel.newTableIdentifier.table.toLowerCase	2✔
48	val oldDatabaseName = alterTableRenameModel.oldTableIdentifier.database	2✔
49	.getOrElse(sparkSession.catalog.currentDatabase)	2✔
50	val oldTableIdentifier = TableIdentifier(oldTableName, Some(oldDatabaseName))	2✔
51	val newTableIdentifier = TableIdentifier(newTableName, Some(oldDatabaseName))	2✔
52	setAuditTable(oldDatabaseName, oldTableIdentifier.table)	2✔
53	setAuditInfo(Map("newName" -> alterTableRenameModel.newTableIdentifier.table))	2✔
54	val newDatabaseName = newTableIdentifier.database	2✔
55	.getOrElse(sparkSession.catalog.currentDatabase)	2✔
56	if (!oldDatabaseName.equalsIgnoreCase(newDatabaseName)) {	2✔
57	throw new MalformedCarbonCommandException("Database name should be same for both tables")	×
58	}
59	val tableExists = sparkSession.catalog.tableExists(oldDatabaseName, newTableIdentifier.table)	2✔
60	if (tableExists) {	2✔
61	throw new MalformedCarbonCommandException(s"Table with name $newTableIdentifier " +	2✔
62	s"already exists")	2✔
63	}
64	LOGGER.info(s"Rename table request has been received for $oldDatabaseName.$oldTableName")	2✔
65	val metastore = CarbonEnv.getInstance(sparkSession).carbonMetaStore	2✔
66	val relation: CarbonRelation =	2✔
67	metastore.lookupRelation(oldTableIdentifier.database, oldTableName)(sparkSession)	2✔
68	.asInstanceOf[CarbonRelation]
69	if (relation == null) {	2✔
70	throwMetadataException(oldDatabaseName, oldTableName, "Table does not exist")	×
71	}
72
73	var oldCarbonTable: CarbonTable = null	2✔
74	oldCarbonTable = metastore.lookupRelation(Some(oldDatabaseName), oldTableName)(sparkSession)	2✔
75	.asInstanceOf[CarbonRelation].carbonTable	2✔
76	if (!oldCarbonTable.getTableInfo.isTransactionalTable) {	2✔
77	throw new MalformedCarbonCommandException("Unsupported operation on non transactional table")	1✔
78	}
79
80	if (!oldCarbonTable.canAllow(oldCarbonTable, TableOperation.ALTER_RENAME)) {	2✔
81	throw new MalformedCarbonCommandException("alter rename is not supported for index datamap")	1✔
82	}
83	// if table have create mv datamap, not support table rename
84	if (CarbonTable.hasMVDataMap(oldCarbonTable)) {	2✔
85	throw new MalformedCarbonCommandException("alter rename is not supported for mv datamap")	×
86	}
87
88	var timeStamp = 0L	2✔
89	var carbonTable: CarbonTable = null	2✔
90	var hiveRenameSuccess = false	2✔
91	// lock file path to release locks after operation
92	var carbonTableLockFilePath: String = null	2✔
93	try {	2✔
94	carbonTable = metastore.lookupRelation(Some(oldDatabaseName), oldTableName)(sparkSession)	2✔
95	.asInstanceOf[CarbonRelation].carbonTable	2✔
96	carbonTableLockFilePath = carbonTable.getTablePath	2✔
97	// if any load is in progress for table, do not allow rename table
98	if (SegmentStatusManager.isLoadInProgressInTable(carbonTable)) {	2✔
99	throw new ConcurrentOperationException(carbonTable, "loading", "alter table rename")	1✔
100	}
101	// get the old table all data map schema
102	val dataMapSchemaList: util.List[DataMapSchema] = new util.ArrayList[DataMapSchema]()	2✔
103	if (carbonTable.hasDataMapSchema) {	2✔
104	dataMapSchemaList.addAll(carbonTable.getTableInfo.getDataMapSchemaList)	2✔
105	}
106	val indexSchemas = DataMapStoreManager.getInstance().getDataMapSchemasOfTable(carbonTable)	2✔
107	if (!indexSchemas.isEmpty) {	2✔
108	dataMapSchemaList.addAll(indexSchemas)	1✔
109	}
110	// invalid data map for the old table, see CARBON-1690
111	val oldAbsoluteTableIdentifier = carbonTable.getAbsoluteTableIdentifier	2✔
112	DataMapStoreManager.getInstance().clearDataMaps(oldAbsoluteTableIdentifier)	2✔
113	// get the latest carbon table and check for column existence
114	val operationContext = new OperationContext	2✔
115	// TODO: Pass new Table Path in pre-event.
116	val alterTableRenamePreEvent: AlterTableRenamePreEvent = AlterTableRenamePreEvent(	2✔
117	carbonTable,	2✔
118	alterTableRenameModel,	2✔
119	"",	2✔
120	sparkSession)	2✔
121	OperationListenerBus.getInstance().fireEvent(alterTableRenamePreEvent, operationContext)	2✔
122	val tableInfo: org.apache.carbondata.format.TableInfo =	2✔
123	metastore.getThriftTableInfo(carbonTable)	2✔
124	val schemaEvolutionEntry = new SchemaEvolutionEntry(System.currentTimeMillis)	2✔
125	schemaEvolutionEntry.setTableName(newTableName)	2✔
126	timeStamp = System.currentTimeMillis()	2✔
127	schemaEvolutionEntry.setTime_stamp(timeStamp)	2✔
128	val newCarbonTableIdentifier = new CarbonTableIdentifier(oldDatabaseName,	2✔
129	newTableName, carbonTable.getCarbonTableIdentifier.getTableId)	2✔
130	metastore.removeTableFromMetadata(oldDatabaseName, oldTableName)	2✔
131	var partitions: Seq[CatalogTablePartition] = Seq.empty	2✔
132	if (carbonTable.isHivePartitionTable) {	2✔
133	partitions =	1✔
134	sparkSession.sessionState.catalog.listPartitions(oldTableIdentifier)	1✔
135	}
136	sparkSession.catalog.refreshTable(oldTableIdentifier.quotedString)	2✔
137	sparkSession.sessionState.catalog.asInstanceOf[CarbonSessionCatalog].alterTableRename(	2✔
138	oldTableIdentifier,	2✔
139	newTableIdentifier,	2✔
140	oldAbsoluteTableIdentifier.getTablePath)	2✔
141	hiveRenameSuccess = true	2✔
142
143	metastore.updateTableSchemaForAlter(	2✔
144	newCarbonTableIdentifier,	2✔
145	carbonTable.getCarbonTableIdentifier,	2✔
146	tableInfo,	2✔
147	schemaEvolutionEntry,	2✔
148	carbonTable.getTablePath)(sparkSession)	2✔
149
150	// Update the storage location with datamap schema
151	if (!dataMapSchemaList.isEmpty) {	2✔
152	DataMapStoreManager.getInstance().	1✔
153	updateDataMapSchema(dataMapSchemaList, newTableName)	1✔
154	}
155	val alterTableRenamePostEvent: AlterTableRenamePostEvent = AlterTableRenamePostEvent(	2✔
156	carbonTable,	2✔
157	alterTableRenameModel,	2✔
158	oldAbsoluteTableIdentifier.getTablePath,	2✔
159	sparkSession)	2✔
160	OperationListenerBus.getInstance().fireEvent(alterTableRenamePostEvent, operationContext)	2✔
161
162	sparkSession.catalog.refreshTable(newTableIdentifier.quotedString)	2✔
163	LOGGER.info(s"Table $oldTableName has been successfully renamed to $newTableName")	2✔
164	} catch {
165	case e: ConcurrentOperationException =>
166	throw e	1✔
167	case e: Exception =>
168	if (hiveRenameSuccess) {	2✔
169	sparkSession.sessionState.catalog.asInstanceOf[CarbonSessionCatalog].alterTableRename(	×
170	newTableIdentifier,	×
171	oldTableIdentifier,	×
172	carbonTable.getAbsoluteTableIdentifier.getTablePath)	×
173	}
174	if (carbonTable != null) {	2✔
175	AlterTableUtil.revertRenameTableChanges(	2✔
176	newTableName,	2✔
177	carbonTable,	2✔
178	timeStamp)(	2✔
179	sparkSession)	2✔
180	}
181	throwMetadataException(oldDatabaseName, oldTableName,	2✔
182	opName + " operation failed: " + e.getMessage)	2✔
183	}
184	Seq.empty	2✔
185	}
186
187	override protected def opName: String = "ALTER TABLE RENAME TABLE"	2✔
188	}

apache / carbondata / 10814

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