[GitHub] carbondata pull request #2990: [CARBONDATA-3149]Support alter table column r...

[qiuchenjian-2]

Github user jackylk commented on a diff in the pull request:

    https://github.com/apache/carbondata/pull/2990#discussion_r243250514
 
    --- Diff: integration/spark2/src/test/scala/org/apache/spark/carbondata/restructure/vectorreader/AlterTableColumnRenameTestCase.scala ---
    @@ -0,0 +1,344 @@
    +/*
    + * 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.carbondata.restructure.vectorreader
    +
    +import org.apache.spark.sql.common.util.Spark2QueryTest
    +import org.scalatest.BeforeAndAfterAll
    +
    +import org.apache.carbondata.core.metadata.CarbonMetadata
    +import org.apache.carbondata.spark.exception.ProcessMetaDataException
    +
    +class AlterTableColumnRenameTestCase extends Spark2QueryTest with BeforeAndAfterAll {
    +
    +
    --- End diff --
   
    remove extra empty line


---

[qiuchenjian-2]

Github user akashrn5 commented on a diff in the pull request:

    https://github.com/apache/carbondata/pull/2990#discussion_r243260747
 
    --- Diff: integration/spark-common/src/main/scala/org/apache/spark/sql/catalyst/CarbonDDLSqlParser.scala ---
    @@ -1511,7 +1514,16 @@ abstract class CarbonDDLSqlParser extends AbstractCarbonSparkSQLParser {
             }
             DataTypeInfo("decimal", precision, scale)
           case _ =>
    -        throw new MalformedCarbonCommandException("Data type provided is invalid.")
    +        if (isColumnRename) {
    +          dataType match {
    --- End diff --
   
    ok, refactored, if rename operation then just return the datatypeInfo object, else throw invalid datatype exception as before


---

[qiuchenjian-2]

Github user akashrn5 commented on a diff in the pull request:

    https://github.com/apache/carbondata/pull/2990#discussion_r243261978
 
    --- Diff: integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/schema/CarbonAlterTableColRenameDataTypeChangeCommand.scala ---
    @@ -0,0 +1,324 @@
    +/*
    + * 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 scala.collection.JavaConverters._
    +import scala.collection.mutable
    +
    +import org.apache.spark.sql.{CarbonEnv, Row, SparkSession}
    +import org.apache.spark.sql.execution.command.{AlterTableDataTypeChangeModel, DataTypeInfo,
    +  MetadataCommand}
    +import org.apache.spark.sql.hive.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.features.TableOperation
    +import org.apache.carbondata.core.locks.{ICarbonLock, LockUsage}
    +import org.apache.carbondata.core.metadata.converter.ThriftWrapperSchemaConverterImpl
    +import org.apache.carbondata.core.metadata.datatype.DecimalType
    +import org.apache.carbondata.core.metadata.schema.table.CarbonTable
    +import org.apache.carbondata.core.metadata.schema.table.column.CarbonColumn
    +import org.apache.carbondata.events.{AlterTableColRenameAndDataTypeChangePostEvent,
    +  AlterTableColRenameAndDataTypeChangePreEvent, OperationContext, OperationListenerBus}
    +import org.apache.carbondata.format.{ColumnSchema, SchemaEvolutionEntry, TableInfo}
    +import org.apache.carbondata.spark.util.DataTypeConverterUtil
    +
    +abstract class CarbonAlterTableColumnRenameCommand(oldColumnName: String, newColumnName: String)
    +  extends MetadataCommand {
    +
    +  protected def validColumnsForRenaming(carbonColumns: mutable.Buffer[CarbonColumn],
    +      oldCarbonColumn: CarbonColumn,
    +      carbonTable: CarbonTable): Unit = {
    +    // check whether new column name is already an existing column name
    +    if (carbonColumns.exists(_.getColName.equalsIgnoreCase(newColumnName))) {
    +      throw new MalformedCarbonCommandException(s"Column Rename Operation failed. New " +
    +                                                s"column name $newColumnName already exists" +
    +                                                s" in table ${ carbonTable.getTableName }")
    +    }
    +
    +    // if the column rename is for complex column, block the operation
    +    if (oldCarbonColumn.isComplex) {
    +      throw new MalformedCarbonCommandException(s"Column Rename Operation failed. Rename " +
    +                                                s"column is unsupported for complex datatype " +
    +                                                s"column ${ oldCarbonColumn.getColName }")
    +    }
    +
    +    // if column rename operation is on partition column, then fail the rename operation
    +    if (null != carbonTable.getPartitionInfo) {
    +      val partitionColumns = carbonTable.getPartitionInfo.getColumnSchemaList
    +      partitionColumns.asScala.foreach {
    +        col =>
    +          if (col.getColumnName.equalsIgnoreCase(oldColumnName)) {
    +            throw new MalformedCarbonCommandException(
    +              s"Column Rename Operation failed. Renaming " +
    +              s"the partition column $newColumnName is not " +
    +              s"allowed")
    +          }
    +      }
    +    }
    +
    +  }
    +}
    +
    +private[sql] case class CarbonAlterTableColRenameDataTypeChangeCommand(
    +    alterTableColRenameAndDataTypeChangeModel: AlterTableDataTypeChangeModel,
    +    childTableColumnRename: Boolean = false)
    +  extends CarbonAlterTableColumnRenameCommand(alterTableColRenameAndDataTypeChangeModel.columnName,
    +    alterTableColRenameAndDataTypeChangeModel.newColumnName) {
    +
    +  override def processMetadata(sparkSession: SparkSession): Seq[Row] = {
    +    val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)
    +    val tableName = alterTableColRenameAndDataTypeChangeModel.tableName
    +    val dbName = alterTableColRenameAndDataTypeChangeModel.databaseName
    +      .getOrElse(sparkSession.catalog.currentDatabase)
    +    var isDataTypeChange = false
    +    setAuditTable(dbName, tableName)
    +    setAuditInfo(Map(
    +      "column" -> alterTableColRenameAndDataTypeChangeModel.columnName,
    +      "newColumn" -> alterTableColRenameAndDataTypeChangeModel.newColumnName,
    +      "newType" -> alterTableColRenameAndDataTypeChangeModel.dataTypeInfo.dataType))
    +    val locksToBeAcquired = List(LockUsage.METADATA_LOCK, LockUsage.COMPACTION_LOCK)
    +    var locks = List.empty[ICarbonLock]
    +    // get the latest carbon table and check for column existence
    +    var carbonTable: CarbonTable = null
    +    var timeStamp = 0L
    +    try {
    +      locks = AlterTableUtil
    +        .validateTableAndAcquireLock(dbName, tableName, locksToBeAcquired)(sparkSession)
    +      val metaStore = CarbonEnv.getInstance(sparkSession).carbonMetaStore
    +      carbonTable = CarbonEnv.getCarbonTable(Some(dbName), tableName)(sparkSession)
    +      if (!alterTableColRenameAndDataTypeChangeModel.isColumnRename &&
    +          !carbonTable.canAllow(carbonTable, TableOperation.ALTER_CHANGE_DATATYPE,
    +            alterTableColRenameAndDataTypeChangeModel.columnName)) {
    +        throw new MalformedCarbonCommandException(
    +          "alter table change datatype is not supported for index datamap")
    +      }
    +      if (alterTableColRenameAndDataTypeChangeModel.isColumnRename &&
    +          !carbonTable.canAllow(carbonTable, TableOperation.ALTER_COLUMN_RENAME,
    +            alterTableColRenameAndDataTypeChangeModel.columnName)) {
    +        throw new MalformedCarbonCommandException(
    +          "alter table column rename is not supported for index datamap")
    +      }
    +      val operationContext = new OperationContext
    +      operationContext.setProperty("childTableColumnRename", childTableColumnRename)
    +      val alterTableColRenameAndDataTypeChangePreEvent =
    +        AlterTableColRenameAndDataTypeChangePreEvent(sparkSession, carbonTable,
    +          alterTableColRenameAndDataTypeChangeModel)
    +      OperationListenerBus.getInstance()
    +        .fireEvent(alterTableColRenameAndDataTypeChangePreEvent, operationContext)
    +      val newColumnName = alterTableColRenameAndDataTypeChangeModel.newColumnName.toLowerCase
    +      val oldColumnName = alterTableColRenameAndDataTypeChangeModel.columnName.toLowerCase
    +      val carbonColumns = carbonTable.getCreateOrderColumn(tableName).asScala.filter(!_.isInvisible)
    +      if (!carbonColumns.exists(_.getColName.equalsIgnoreCase(oldColumnName))) {
    +        throwMetadataException(dbName, tableName, s"Column does not exist: $oldColumnName")
    +      }
    +
    +      val oldCarbonColumn = carbonColumns.filter(_.getColName.equalsIgnoreCase(oldColumnName))
    +      if (oldCarbonColumn.size != 1) {
    +        throwMetadataException(dbName, tableName, s"Invalid Column: $oldColumnName")
    +      }
    +      val newColumnPrecision = alterTableColRenameAndDataTypeChangeModel.dataTypeInfo.precision
    +      val newColumnScale = alterTableColRenameAndDataTypeChangeModel.dataTypeInfo.scale
    +      if (alterTableColRenameAndDataTypeChangeModel.isColumnRename) {
    +        // validate the columns to be renamed
    +        validColumnsForRenaming(carbonColumns, oldCarbonColumn.head, carbonTable)
    +        // if the datatype is source datatype, then it is just a column rename operation, else set
    +        // the isDataTypeChange flag to true
    +        if (oldCarbonColumn.head.getDataType.getName
    +          .equalsIgnoreCase(alterTableColRenameAndDataTypeChangeModel.dataTypeInfo.dataType)) {
    +          val newColumnPrecision =
    +            alterTableColRenameAndDataTypeChangeModel.dataTypeInfo.precision
    +          val newColumnScale = alterTableColRenameAndDataTypeChangeModel.dataTypeInfo.scale
    +          // if the source datatype is decimal and there is change in precision and scale, then
    +          // along with rename, datatype change is also required for the command, so set the
    +          // isDataTypeChange flag to true in this case
    +          if (oldCarbonColumn.head.getDataType.getName.equalsIgnoreCase("decimal") &&
    +              (oldCarbonColumn.head.getDataType.asInstanceOf[DecimalType].getPrecision !=
    +               newColumnPrecision ||
    +               oldCarbonColumn.head.getDataType.asInstanceOf[DecimalType].getScale !=
    +               newColumnScale)) {
    +            isDataTypeChange = true
    +          }
    +        } else {
    +          isDataTypeChange = true
    +        }
    +      } else {
    +        isDataTypeChange = true
    +      }
    +      if (isDataTypeChange) {
    +        validateColumnDataType(alterTableColRenameAndDataTypeChangeModel.dataTypeInfo,
    +          oldCarbonColumn.head)
    +      }
    +      // read the latest schema file
    +      val tableInfo: TableInfo =
    +        metaStore.getThriftTableInfo(carbonTable)
    +      // maintain the added column for schema evolution history
    +      var addColumnSchema: ColumnSchema = null
    +      var deletedColumnSchema: ColumnSchema = null
    +      val schemaEvolutionEntry: SchemaEvolutionEntry = null
    +      val columnSchemaList = tableInfo.fact_table.table_columns.asScala.filter(!_.isInvisible)
    +
    +      columnSchemaList.foreach { columnSchema =>
    +        if (columnSchema.column_name.equalsIgnoreCase(oldColumnName)) {
    +          deletedColumnSchema = columnSchema.deepCopy()
    +          if (alterTableColRenameAndDataTypeChangeModel.isColumnRename) {
    +            // if only column rename, just get the column schema and rename, make a
    +            // schemaEvolutionEntry
    +            columnSchema.setColumn_name(newColumnName)
    +          }
    +          // if the column rename is false,it will be just datatype change only, then change the
    +          // datatype and make an evolution entry, If both the operations are happening, then rename
    +          // change datatype and make an evolution entry
    +          if (isDataTypeChange) {
    +            // if only datatype change,  just get the column schema and change datatype, make a
    +            // schemaEvolutionEntry
    +            columnSchema.setData_type(
    +              DataTypeConverterUtil.convertToThriftDataType(
    +                alterTableColRenameAndDataTypeChangeModel.dataTypeInfo.dataType))
    +            columnSchema
    +              .setPrecision(newColumnPrecision)
    +            columnSchema.setScale(newColumnScale)
    +          }
    +          addColumnSchema = columnSchema
    +          timeStamp = System.currentTimeMillis()
    +          // make a new schema evolution entry after column rename or datatype change
    +          AlterTableUtil
    +            .addNewSchemaEvolutionEntry(schemaEvolutionEntry, timeStamp, addColumnSchema,
    +              deletedColumnSchema)
    +        }
    +      }
    +
    +      // modify the table Properties with new column name if column rename happened
    +      if (alterTableColRenameAndDataTypeChangeModel.isColumnRename) {
    +        AlterTableUtil
    +          .modifyTablePropertiesAfterColumnRename(tableInfo.fact_table.tableProperties.asScala,
    +            oldColumnName, newColumnName)
    +      }
    +      updateSchemaAndRefreshTable(sparkSession,
    +        carbonTable,
    +        tableInfo,
    +        addColumnSchema,
    +        schemaEvolutionEntry)
    +      val alterTableColRenameAndDataTypeChangePostEvent
    +      : AlterTableColRenameAndDataTypeChangePostEvent =
    +        AlterTableColRenameAndDataTypeChangePostEvent(sparkSession, carbonTable,
    +          alterTableColRenameAndDataTypeChangeModel)
    +      OperationListenerBus.getInstance
    +        .fireEvent(alterTableColRenameAndDataTypeChangePostEvent, operationContext)
    +      if (isDataTypeChange) {
    +        LOGGER
    +          .info(s"Alter table for column rename or data type change is successful for table " +
    +                s"$dbName.$tableName")
    +      }
    +      if (alterTableColRenameAndDataTypeChangeModel.isColumnRename) {
    +        LOGGER.info(s"Alter table for column rename is successful for table $dbName.$tableName")
    +      }
    +    } catch {
    +      case e: Exception =>
    +        if (carbonTable != null) {
    +          AlterTableUtil
    +            .revertColumnRenameAndDataTypeChanges(dbName, tableName, timeStamp)(sparkSession)
    +        }
    +        if (isDataTypeChange) {
    +          throwMetadataException(dbName, tableName,
    +            s"Alter table data type change operation failed: ${ e.getMessage }")
    +        } else {
    +          throwMetadataException(dbName, tableName,
    +            s"Alter table data type change or column rename operation failed: ${ e.getMessage }")
    +        }
    +    } finally {
    +      // release lock after command execution completion
    +      AlterTableUtil.releaseLocks(locks)
    +    }
    +    Seq.empty
    +  }
    +
    +  /**
    +   * This method update the schema info and refresh the table
    +   *
    +   * @param sparkSession
    +   * @param carbonTable              carbonTable
    +   * @param tableInfo                tableInfo
    +   * @param addColumnSchema          added column schema
    +   * @param schemaEvolutionEntryList new SchemaEvolutionEntry
    +   */
    +  private def updateSchemaAndRefreshTable(sparkSession: SparkSession,
    +      carbonTable: CarbonTable,
    +      tableInfo: TableInfo,
    +      addColumnSchema: ColumnSchema,
    +      schemaEvolutionEntry: SchemaEvolutionEntry): Unit = {
    +    val schemaConverter = new ThriftWrapperSchemaConverterImpl
    +    val a = List(schemaConverter.fromExternalToWrapperColumnSchema(addColumnSchema))
    +    val (tableIdentifier, schemaParts, cols) = AlterTableUtil.updateSchemaInfo(
    +      carbonTable, schemaEvolutionEntry, tableInfo, Some(a))(sparkSession)
    +    sparkSession.sessionState.catalog.asInstanceOf[CarbonSessionCatalog]
    +      .alterColumnChangeDataType(tableIdentifier, schemaParts, cols)
    +    sparkSession.catalog.refreshTable(tableIdentifier.quotedString)
    +  }
    +
    +  /**
    +   * This method will validate a column for its data type and check whether the column data type
    +   * can be modified and update if conditions are met.
    +   */
    +  private def validateColumnDataType(
    +      dataTypeInfo: DataTypeInfo,
    +      carbonColumn: CarbonColumn): Unit = {
    +    carbonColumn.getDataType.getName match {
    --- End diff --
   
    this logic is kept same as old , as it was for datatype change command, `carbonColumn.getDataType` gives object and with that also we need to compare the string from datatypeinfo object as it has just a string, so this is ok i think


---

[qiuchenjian-2]

Github user akashrn5 commented on the issue:

    https://github.com/apache/carbondata/pull/2990
 
    @jackylk handled comments, please review


---

[qiuchenjian-2]

Github user CarbonDataQA commented on the issue:

    https://github.com/apache/carbondata/pull/2990
 
    Build Success with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder2.1/1879/



---

[qiuchenjian-2]

Github user CarbonDataQA commented on the issue:

    https://github.com/apache/carbondata/pull/2990
 
    Build Success with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder2.1/1880/



---

[qiuchenjian-2]

Github user CarbonDataQA commented on the issue:

    https://github.com/apache/carbondata/pull/2990
 
    Build Failed  with Spark 2.3.2, Please check CI http://136.243.101.176:8080/job/carbondataprbuilder2.3/10135/



---

[qiuchenjian-2]

Github user CarbonDataQA commented on the issue:

    https://github.com/apache/carbondata/pull/2990
 
    Build Failed with Spark 2.2.1, Please check CI http://95.216.28.178:8080/job/ApacheCarbonPRBuilder1/2090/



---

[qiuchenjian-2]

Github user CarbonDataQA commented on the issue:

    https://github.com/apache/carbondata/pull/2990
 
    Build Success with Spark 2.1.0, Please check CI http://136.243.101.176:8080/job/ApacheCarbonPRBuilder2.1/1883/



---

[qiuchenjian-2]

Github user CarbonDataQA commented on the issue:

    https://github.com/apache/carbondata/pull/2990
 
    Build Success with Spark 2.2.1, Please check CI http://95.216.28.178:8080/job/ApacheCarbonPRBuilder1/2092/



---

[qiuchenjian-2]

Github user CarbonDataQA commented on the issue:

    https://github.com/apache/carbondata/pull/2990
 
    Build Success with Spark 2.3.2, Please check CI http://136.243.101.176:8080/job/carbondataprbuilder2.3/10138/



---

[qiuchenjian-2]

Github user jackylk commented on the issue:

    https://github.com/apache/carbondata/pull/2990
 
    LGTM. Thanks for working on this


---

[qiuchenjian-2]

Github user akashrn5 commented on the issue:

    https://github.com/apache/carbondata/pull/2990
 
    @jackylk thank you


---

[qiuchenjian-2]

Github user asfgit closed the pull request at:

    https://github.com/apache/carbondata/pull/2990


---