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

import java.util

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

import org.apache.commons.lang3.StringUtils
import org.apache.spark.SparkConf
import org.apache.spark.sql.{CarbonEnv, SparkSession}
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.hive.{CarbonRelation, CarbonSessionCatalog}
import org.apache.spark.sql.hive.HiveExternalCatalog._

import org.apache.carbondata.common.exceptions.sql.MalformedCarbonCommandException
import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.constants.SortScopeOptions.SortScope
import org.apache.carbondata.core.datamap.DataMapStoreManager
import org.apache.carbondata.core.datastore.block.SegmentPropertiesAndSchemaHolder
import org.apache.carbondata.core.datastore.impl.FileFactory
import org.apache.carbondata.core.exception.InvalidConfigurationException
import org.apache.carbondata.core.locks.{CarbonLockUtil, ICarbonLock, LockUsage}
import org.apache.carbondata.core.metadata.{AbsoluteTableIdentifier, CarbonTableIdentifier}
import org.apache.carbondata.core.metadata.converter.{SchemaConverter, ThriftWrapperSchemaConverterImpl}
import org.apache.carbondata.core.metadata.datatype.DataTypes
import org.apache.carbondata.core.metadata.schema.table.CarbonTable
import org.apache.carbondata.core.metadata.schema.table.column.ColumnSchema
import org.apache.carbondata.core.util.CarbonUtil
import org.apache.carbondata.format.{SchemaEvolutionEntry, TableInfo}
import org.apache.carbondata.spark.util.{CarbonScalaUtil, CommonUtil}


object AlterTableUtil {

  private val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)

  /**
   * Validates that the table exists and acquires meta lock on it.
   *
   * @param dbName
   * @param tableName
   * @param sparkSession
   * @return
   */
  def validateTableAndAcquireLock(dbName: String,
      tableName: String,
      locksToBeAcquired: List[String])
    (sparkSession: SparkSession): List[ICarbonLock] = {
    val relation =
      CarbonEnv.getInstance(sparkSession).carbonMetaStore
        .lookupRelation(Option(dbName), tableName)(sparkSession)
        .asInstanceOf[CarbonRelation]
    if (relation == null) {
      sys.error(s"Table $dbName.$tableName does not exist")
    }
    // acquire the lock first
    val table = relation.carbonTable
    val acquiredLocks = ListBuffer[ICarbonLock]()
    try {
      locksToBeAcquired.foreach { lock =>
        acquiredLocks += CarbonLockUtil.getLockObject(table.getAbsoluteTableIdentifier, lock)
      }
      acquiredLocks.toList
    } catch {
      case e: Exception =>
        releaseLocks(acquiredLocks.toList)
        throw e
    }
  }

  /**
   * This method will release the locks acquired for an operation
   *
   * @param locks
   */
  def releaseLocks(locks: List[ICarbonLock]): Unit = {
    locks.foreach { carbonLock =>
      if (carbonLock.unlock()) {
        LOGGER.info("Alter table lock released successfully")
      } else {
        LOGGER.error("Unable to release lock during alter table operation")
      }
    }
  }

  /**
   * update schema when SORT_COLUMNS are be changed
   */
  private def updateSchemaForSortColumns(
      thriftTable: TableInfo,
      lowerCasePropertiesMap: mutable.Map[String, String],
      schemaConverter: SchemaConverter
  ) = {
    val sortColumnsOption = lowerCasePropertiesMap.get(CarbonCommonConstants.SORT_COLUMNS)
    if (sortColumnsOption.isDefined) {
      val sortColumnsString = CarbonUtil.unquoteChar(sortColumnsOption.get).trim
      val columns = thriftTable.getFact_table.getTable_columns
      // remove old sort_columns property from ColumnSchema
      val columnSeq =
        columns
          .asScala
          .map { column =>
            val columnProperties = column.getColumnProperties
            if (columnProperties != null) {
              columnProperties.remove(CarbonCommonConstants.SORT_COLUMNS)
            }
            column
          }
          .zipWithIndex
      if (!sortColumnsString.isEmpty) {
        val newSortColumns = sortColumnsString.split(',').map(_.trim)
        // map sort_columns index in column list
        val sortColumnsIndexMap = newSortColumns
          .zipWithIndex
          .map { entry =>
            val column = columnSeq.find(_._1.getColumn_name.equalsIgnoreCase(entry._1)).get
            var columnProperties = column._1.getColumnProperties
            if (columnProperties == null) {
              columnProperties = new util.HashMap[String, String]()
              column._1.setColumnProperties(columnProperties)
            }
            // change sort_columns to dimension
            if (!column._1.isDimension) {
              column._1.setDimension(true)
              columnProperties.put(CarbonCommonConstants.COLUMN_DRIFT, "true")
            }
            // add sort_columns property
            columnProperties.put(CarbonCommonConstants.SORT_COLUMNS, "true")
            (column._2, entry._2)
          }
          .toMap
        var index = newSortColumns.length
        // re-order all columns, move sort_columns to the head of column list
        val newColumns = columnSeq
          .map { entry =>
            val sortColumnIndexOption = sortColumnsIndexMap.get(entry._2)
            val newIndex = if (sortColumnIndexOption.isDefined) {
              sortColumnIndexOption.get
            } else {
              val tempIndex = index
              index += 1
              tempIndex
            }
            (newIndex, entry._1)
          }
          .sortWith(_._1 < _._1)
          .map(_._2)
          .asJava
        // use new columns
        columns.clear()
        columns.addAll(newColumns)
      }
    }
  }

  /**
   * @param carbonTable
   * @param schemaEvolutionEntry
   * @param thriftTable
   * @param sparkSession
   */
  def updateSchemaInfo(carbonTable: CarbonTable,
      schemaEvolutionEntry: SchemaEvolutionEntry = null,
      thriftTable: TableInfo)
    (sparkSession: SparkSession): (TableIdentifier, String) = {
    val dbName = carbonTable.getDatabaseName
    val tableName = carbonTable.getTableName
    CarbonEnv.getInstance(sparkSession).carbonMetaStore
      .updateTableSchemaForAlter(carbonTable.getCarbonTableIdentifier,
        carbonTable.getCarbonTableIdentifier,
        thriftTable,
        schemaEvolutionEntry,
        carbonTable.getAbsoluteTableIdentifier.getTablePath)(sparkSession)
    val tableIdentifier = TableIdentifier(tableName, Some(dbName))
    sparkSession.catalog.refreshTable(tableIdentifier.quotedString)
    val schema = CarbonEnv.getInstance(sparkSession).carbonMetaStore
      .lookupRelation(tableIdentifier)(sparkSession).schema.json
    val schemaParts = prepareSchemaJsonForAlterTable(sparkSession.sparkContext.getConf, schema)
    (tableIdentifier, schemaParts)
  }

  /**
   * This method will split schema string into multiple parts of configured size and
   * registers the parts as keys in tableProperties which will be read by spark to prepare
   * Carbon Table fields
   *
   * @param sparkConf
   * @param schemaJsonString
   * @return
   */
  private def prepareSchemaJsonForAlterTable(sparkConf: SparkConf,
      schemaJsonString: String): String = {
    val threshold = sparkConf
      .getInt(CarbonCommonConstants.SPARK_SCHEMA_STRING_LENGTH_THRESHOLD,
        CarbonCommonConstants.SPARK_SCHEMA_STRING_LENGTH_THRESHOLD_DEFAULT)
    // Split the JSON string.
    val parts = schemaJsonString.grouped(threshold).toSeq
    var schemaParts: Seq[String] = Seq.empty
    schemaParts = schemaParts :+ s"'$DATASOURCE_SCHEMA_NUMPARTS'='${ parts.size }'"
    parts.zipWithIndex.foreach { case (part, index) =>
      schemaParts = schemaParts :+ s"'$DATASOURCE_SCHEMA_PART_PREFIX$index'='$part'"
    }
    schemaParts.mkString(",")
  }

  /**
   * This method reverts the changes to the schema if the rename table command fails.
   */
  def revertRenameTableChanges(
      newTableName: String,
      oldCarbonTable: CarbonTable,
      timeStamp: Long)
    (sparkSession: SparkSession): Unit = {
    val tablePath = oldCarbonTable.getTablePath
    val tableId = oldCarbonTable.getCarbonTableIdentifier.getTableId
    val oldCarbonTableIdentifier = oldCarbonTable.getCarbonTableIdentifier
    val database = oldCarbonTable.getDatabaseName
    val newCarbonTableIdentifier = new CarbonTableIdentifier(database, newTableName, tableId)
    val metastore = CarbonEnv.getInstance(sparkSession).carbonMetaStore
    if (FileFactory.isFileExist(tablePath)) {
      val tableInfo = metastore.getThriftTableInfo(oldCarbonTable)
      val evolutionEntryList = tableInfo.fact_table.schema_evolution.schema_evolution_history
      val updatedTime = evolutionEntryList.get(evolutionEntryList.size() - 1).time_stamp
      if (updatedTime == timeStamp) {
        LOGGER.error(s"Reverting changes for $database.${oldCarbonTable.getTableName}")
        val absoluteTableIdentifier = AbsoluteTableIdentifier.from(
          tablePath,
          newCarbonTableIdentifier)
        metastore.revertTableSchemaInAlterFailure(oldCarbonTableIdentifier,
          tableInfo, absoluteTableIdentifier)(sparkSession)
        metastore.removeTableFromMetadata(database, newTableName)
      }
    }
  }

  /**
   * This method reverts the changes to the schema if add column command fails.
   *
   * @param dbName
   * @param tableName
   * @param timeStamp
   * @param sparkSession
   */
  def revertAddColumnChanges(dbName: String, tableName: String, timeStamp: Long)
    (sparkSession: SparkSession): Unit = {
    val metastore = CarbonEnv.getInstance(sparkSession).carbonMetaStore
    val carbonTable = CarbonEnv.getCarbonTable(Some(dbName), tableName)(sparkSession)
    val thriftTable: TableInfo = metastore.getThriftTableInfo(carbonTable)
    val evolutionEntryList = thriftTable.fact_table.schema_evolution.schema_evolution_history
    val updatedTime = evolutionEntryList.get(evolutionEntryList.size() - 1).time_stamp
    if (updatedTime == timeStamp) {
      LOGGER.info(s"Reverting changes for $dbName.$tableName")
      val addedSchemas = evolutionEntryList.get(evolutionEntryList.size() - 1).added
      thriftTable.fact_table.table_columns.removeAll(addedSchemas)
      metastore
        .revertTableSchemaInAlterFailure(carbonTable.getCarbonTableIdentifier,
          thriftTable, carbonTable.getAbsoluteTableIdentifier)(sparkSession)
    }
  }

  /**
   * This method reverts the schema changes if drop table command fails.
   *
   * @param dbName
   * @param tableName
   * @param timeStamp
   * @param sparkSession
   */
  def revertDropColumnChanges(dbName: String, tableName: String, timeStamp: Long)
    (sparkSession: SparkSession): Unit = {
    val metastore = CarbonEnv.getInstance(sparkSession).carbonMetaStore
    val carbonTable = CarbonEnv.getCarbonTable(Some(dbName), tableName)(sparkSession)
    val thriftTable: TableInfo = metastore.getThriftTableInfo(carbonTable)
    val evolutionEntryList = thriftTable.fact_table.schema_evolution.schema_evolution_history
    val updatedTime = evolutionEntryList.get(evolutionEntryList.size() - 1).time_stamp
    if (updatedTime == timeStamp) {
      LOGGER.error(s"Reverting changes for $dbName.$tableName")
      val removedSchemas = evolutionEntryList.get(evolutionEntryList.size() - 1).removed
      thriftTable.fact_table.table_columns.asScala.foreach { columnSchema =>
        removedSchemas.asScala.foreach { removedSchemas =>
          if (columnSchema.invisible && removedSchemas.column_id == columnSchema.column_id) {
            columnSchema.setInvisible(false)
          }
        }
      }
      metastore
        .revertTableSchemaInAlterFailure(carbonTable.getCarbonTableIdentifier,
          thriftTable, carbonTable.getAbsoluteTableIdentifier)(sparkSession)
    }
  }

  /**
   * This method reverts the changes to schema if the data type change command fails.
   *
   * @param dbName
   * @param tableName
   * @param timeStamp
   * @param sparkSession
   */
  def revertColumnRenameAndDataTypeChanges(dbName: String, tableName: String, timeStamp: Long)
    (sparkSession: SparkSession): Unit = {
    val metaStore = CarbonEnv.getInstance(sparkSession).carbonMetaStore
    val carbonTable = CarbonEnv.getCarbonTable(Some(dbName), tableName)(sparkSession)
    val thriftTable: TableInfo = metaStore.getThriftTableInfo(carbonTable)
    val evolutionEntryList = thriftTable.fact_table.schema_evolution.schema_evolution_history
    val updatedTime = evolutionEntryList.get(evolutionEntryList.size() - 1).time_stamp
    if (updatedTime == timeStamp) {
      LOGGER.error(s"Reverting changes for $dbName.$tableName")
      val removedColumns = evolutionEntryList.get(evolutionEntryList.size() - 1).removed
      thriftTable.fact_table.table_columns.asScala.foreach { columnSchema =>
        removedColumns.asScala.foreach { removedColumn =>
          if (columnSchema.column_id.equalsIgnoreCase(removedColumn.column_id) &&
              !columnSchema.isInvisible) {
            columnSchema.setData_type(removedColumn.data_type)
            columnSchema.setPrecision(removedColumn.precision)
            columnSchema.setScale(removedColumn.scale)
          }
        }
      }
      metaStore
        .revertTableSchemaInAlterFailure(carbonTable.getCarbonTableIdentifier,
          thriftTable, carbonTable.getAbsoluteTableIdentifier)(sparkSession)
    }
  }

  /**
   * This method modifies the table properties if column rename happened
   * @param tableProperties tableProperties of the table
   * @param oldColumnName old COlumnname before rename
   * @param newColumnName new column name to rename
   */
  def modifyTablePropertiesAfterColumnRename(
      tableProperties: mutable.Map[String, String],
      oldColumnName: String,
      newColumnName: String): Unit = {
    tableProperties.foreach { tableProperty =>
      if (tableProperty._2.contains(oldColumnName)) {
        val tablePropertyKey = tableProperty._1
        val tablePropertyValue = tableProperty._2
        tableProperties
          .put(tablePropertyKey, tablePropertyValue.replace(oldColumnName, newColumnName))
      }
    }
  }

  /**
   * This method create a new SchemaEvolutionEntry and adds to SchemaEvolutionEntry List
   *
   * @param schemaEvolutionEntry List to add new SchemaEvolutionEntry
   * @param addColumnSchema          added new column schema
   * @param deletedColumnSchema      old column schema which is deleted
   * @return
   */
  def addNewSchemaEvolutionEntry(
      schemaEvolutionEntry: SchemaEvolutionEntry,
      timeStamp: Long,
      addColumnSchema: org.apache.carbondata.format.ColumnSchema,
      deletedColumnSchema: org.apache.carbondata.format.ColumnSchema): SchemaEvolutionEntry = {
    var schemaEvolutionEntry = new SchemaEvolutionEntry(timeStamp)
    schemaEvolutionEntry.setAdded(List(addColumnSchema).asJava)
    schemaEvolutionEntry.setRemoved(List(deletedColumnSchema).asJava)
    schemaEvolutionEntry
  }

  /**
   * This method add/modify the table comments.
   *
   * @param tableIdentifier
   * @param properties
   * @param propKeys
   * @param set
   * @param sparkSession
   */
  def modifyTableProperties(tableIdentifier: TableIdentifier, properties: Map[String, String],
      propKeys: Seq[String], set: Boolean)
    (sparkSession: SparkSession, catalog: CarbonSessionCatalog): Unit = {
    val tableName = tableIdentifier.table
    val dbName = tableIdentifier.database.getOrElse(sparkSession.catalog.currentDatabase)
    val locksToBeAcquired = List(LockUsage.METADATA_LOCK, LockUsage.COMPACTION_LOCK)
    var locks = List.empty[ICarbonLock]
    try {
      locks = AlterTableUtil
        .validateTableAndAcquireLock(dbName, tableName, locksToBeAcquired)(sparkSession)
      val metastore = CarbonEnv.getInstance(sparkSession).carbonMetaStore
      val carbonTable = CarbonEnv.getCarbonTable(Some(dbName), tableName)(sparkSession)
      val lowerCasePropertiesMap: mutable.Map[String, String] = mutable.Map.empty
      // convert all the keys to lower case
      properties.foreach { entry =>
        lowerCasePropertiesMap.put(entry._1.toLowerCase, entry._2)
      }
      // validate the required cache level properties
      validateColumnMetaCacheAndCacheLevel(carbonTable, lowerCasePropertiesMap)
      // get the latest carbon table
      // read the latest schema file
      val thriftTableInfo: TableInfo = metastore.getThriftTableInfo(carbonTable)
      val schemaConverter = new ThriftWrapperSchemaConverterImpl()
      val wrapperTableInfo = schemaConverter.fromExternalToWrapperTableInfo(
        thriftTableInfo,
        dbName,
        tableName,
        carbonTable.getTablePath)
      val thriftTable = schemaConverter.fromWrapperToExternalTableInfo(
        wrapperTableInfo, dbName, tableName)
      val tblPropertiesMap: mutable.Map[String, String] =
        thriftTable.fact_table.getTableProperties.asScala

      // validate the local dictionary properties
      validateLocalDictionaryProperties(lowerCasePropertiesMap, tblPropertiesMap, carbonTable)

      // validate the load min size properties
      validateLoadMinSizeProperties(carbonTable, lowerCasePropertiesMap)

      // validate the range column properties
      validateRangeColumnProperties(carbonTable, lowerCasePropertiesMap)

      // validate the Sort Scope and Sort Columns
      validateSortScopeAndSortColumnsProperties(carbonTable, lowerCasePropertiesMap)
      // if SORT_COLUMN is changed, it will move them to the head of column list
      updateSchemaForSortColumns(thriftTable, lowerCasePropertiesMap, schemaConverter)
      // below map will be used for cache invalidation. As tblProperties map is getting modified
      // in the next few steps the original map need to be retained for any decision making
      val existingTablePropertiesMap = mutable.Map(tblPropertiesMap.toSeq: _*)
      if (set) {
        // This overrides old newProperties and update the comment parameter of thriftTable
        // with the newly added/modified comment since thriftTable also holds comment as its
        // direct property.
        lowerCasePropertiesMap.foreach { property =>
          if (validateTableProperties(property._1)) {
            tblPropertiesMap.put(property._1, property._2)
          } else {
            val errorMessage = "Error: Invalid option(s): " + property._1.toString()
            throw new MalformedCarbonCommandException(errorMessage)
          }
        }
        // check if duplicate columns are present in both local dictionary include and exclude
        CarbonScalaUtil.validateDuplicateLocalDictIncludeExcludeColmns(tblPropertiesMap)
      } else {
        // This removes the comment parameter from thriftTable
        // since thriftTable also holds comment as its property.
        propKeys.foreach { propKey =>
          if (validateTableProperties(propKey)) {
            // This check is required because for old tables we need to keep same behavior for it,
            // meaning, local dictionary should be disabled. To enable we can use set command for
            // older tables. So no need to remove from table properties map for unset just to ensure
            // for older table behavior. So in case of unset, if enable property is already present
            // in map, then just set it to default value of local dictionary which is true.
            if (propKey.equalsIgnoreCase(CarbonCommonConstants.LOCAL_DICTIONARY_ENABLE)) {
              tblPropertiesMap
                .put(propKey.toLowerCase, CarbonCommonConstants.LOCAL_DICTIONARY_ENABLE_DEFAULT)
            } else if (propKey.equalsIgnoreCase(CarbonCommonConstants.SORT_SCOPE)) {
              tblPropertiesMap
                .put(propKey.toLowerCase, CarbonCommonConstants.LOAD_SORT_SCOPE_DEFAULT)
            } else if (propKey.equalsIgnoreCase(CarbonCommonConstants.SORT_COLUMNS)) {
              val errorMessage = "Error: Invalid option(s): " + propKey +
                                 ", please set SORT_COLUMNS as empty instead of unset"
              throw new MalformedCarbonCommandException(errorMessage)
            } else {
              tblPropertiesMap.remove(propKey.toLowerCase)
            }
          } else {
            val errorMessage = "Error: Invalid option(s): " + propKey
            throw new MalformedCarbonCommandException(errorMessage)
          }
        }
        // check if duplicate columns are present in both local dictionary include and exclude
        CarbonScalaUtil.validateDuplicateLocalDictIncludeExcludeColmns(tblPropertiesMap)
      }
      val (tableIdentifier, schemParts) = updateSchemaInfo(
        carbonTable = carbonTable,
        thriftTable = thriftTable)(sparkSession)
      catalog.alterTable(tableIdentifier, schemParts, None)
      sparkSession.catalog.refreshTable(tableIdentifier.quotedString)
      // check and clear the block/blocklet cache
      checkAndClearBlockletCache(carbonTable,
        existingTablePropertiesMap,
        lowerCasePropertiesMap,
        propKeys,
        set)
      LOGGER.info(s"Alter table newProperties is successful for table $dbName.$tableName")
    } catch {
      case e: Exception =>
        sys.error(s"Alter table newProperties operation failed: ${e.getMessage}")
    } finally {
      // release lock after command execution completion
      AlterTableUtil.releaseLocks(locks)
    }
  }

  private def validateTableProperties(propKey: String): Boolean = {
    val supportedOptions = Seq("STREAMING",
      "COMMENT",
      "COLUMN_META_CACHE",
      "CACHE_LEVEL",
      "LOCAL_DICTIONARY_ENABLE",
      "LOCAL_DICTIONARY_THRESHOLD",
      "LOCAL_DICTIONARY_INCLUDE",
      "LOCAL_DICTIONARY_EXCLUDE",
      "LOAD_MIN_SIZE_INMB",
      "RANGE_COLUMN",
      "SORT_SCOPE",
      "SORT_COLUMNS",
      "GLOBAL_SORT_PARTITIONS")
    supportedOptions.contains(propKey.toUpperCase)
  }

  /**
   * this method validates the local dictioanry properties for alter set
   *
   * @param lowerCasePropertiesMap
   * @param tblPropertiesMap
   * @param carbonTable
   */
  private def validateLocalDictionaryProperties(lowerCasePropertiesMap: mutable.Map[String, String],
      tblPropertiesMap: mutable.Map[String, String],
      carbonTable: CarbonTable): Unit = {
    lowerCasePropertiesMap.foreach { property =>
      if (property._1.equalsIgnoreCase(CarbonCommonConstants.LOCAL_DICTIONARY_ENABLE)) {
        if (!CarbonScalaUtil.validateLocalDictionaryEnable(property._2)) {
          lowerCasePropertiesMap
            .put(property._1.toLowerCase, CarbonCommonConstants.LOCAL_DICTIONARY_ENABLE_DEFAULT)
        } else {
          lowerCasePropertiesMap.put(property._1, property._2)
        }
      }
      if (property._1.equalsIgnoreCase(CarbonCommonConstants.LOCAL_DICTIONARY_INCLUDE) ||
          property._1.equalsIgnoreCase(CarbonCommonConstants.LOCAL_DICTIONARY_EXCLUDE)) {
        ValidateSetTablePropertiesForLocalDict(tblPropertiesMap, carbonTable, property)
      }

      if (property._1
        .equalsIgnoreCase(CarbonCommonConstants.LOCAL_DICTIONARY_THRESHOLD)) {
        if (!CarbonScalaUtil.validateLocalDictionaryThreshold(property._2)) {
          lowerCasePropertiesMap
            .put(property._1,
              CarbonCommonConstants.LOCAL_DICTIONARY_THRESHOLD_DEFAULT)
        } else {
          lowerCasePropertiesMap.put(property._1, property._2)
        }
      }
    }
  }

  /**
   * validate column meta cache and cache level properties if configured by the user
   *
   * @param carbonTable
   * @param propertiesMap
   */
  private def validateColumnMetaCacheAndCacheLevel(carbonTable: CarbonTable,
      propertiesMap: mutable.Map[String, String]): Unit = {
    // validate column meta cache property
    if (propertiesMap.get(CarbonCommonConstants.COLUMN_META_CACHE).isDefined) {
      val schemaList: util.List[ColumnSchema] = CarbonUtil
        .getColumnSchemaList(carbonTable.getVisibleDimensions, carbonTable.getVisibleMeasures)
      val tableColumns: Seq[String] = schemaList.asScala
        .map(columnSchema => columnSchema.getColumnName)
      CommonUtil
        .validateColumnMetaCacheFields(carbonTable.getDatabaseName,
          carbonTable.getTableName,
          tableColumns,
          propertiesMap.get(CarbonCommonConstants.COLUMN_META_CACHE).get,
          propertiesMap)
      val columnsToBeCached = propertiesMap.get(CarbonCommonConstants.COLUMN_META_CACHE).get
      validateForComplexTypeColumn(carbonTable, columnsToBeCached)
    }
    // validate cache level property
    if (propertiesMap.get(CarbonCommonConstants.CACHE_LEVEL).isDefined) {
      CommonUtil.validateCacheLevel(
        propertiesMap.get(CarbonCommonConstants.CACHE_LEVEL).get,
        propertiesMap)
    }
  }

  def validateRangeColumnProperties(carbonTable: CarbonTable,
      propertiesMap: mutable.Map[String, String]): Unit = {
    if (propertiesMap.get(CarbonCommonConstants.RANGE_COLUMN).isDefined) {
      val rangeColumnProp = propertiesMap.get(CarbonCommonConstants.RANGE_COLUMN).get
      if (rangeColumnProp.contains(",")) {
        val errorMsg = "range_column not support multiple columns"
        throw new MalformedCarbonCommandException(errorMsg)
      }
      val rangeColumn = carbonTable.getColumnByName(rangeColumnProp)
      if (rangeColumn == null) {
        throw new MalformedCarbonCommandException(
          s"Table property ${ CarbonCommonConstants.RANGE_COLUMN }: ${ rangeColumnProp }" +
          s" is not exists in the table")
      } else {
        propertiesMap.put(CarbonCommonConstants.RANGE_COLUMN, rangeColumn.getColName)
      }
    }
  }

  def validateSortScopeAndSortColumnsProperties(carbonTable: CarbonTable,
      propertiesMap: mutable.Map[String, String]): Unit = {
    CommonUtil.validateSortScope(propertiesMap)
    CommonUtil.validateSortColumns(carbonTable, propertiesMap)
    // match SORT_SCOPE and SORT_COLUMNS
    val newSortScope = propertiesMap.get(CarbonCommonConstants.SORT_SCOPE)
    val newSortColumns = propertiesMap.get(CarbonCommonConstants.SORT_COLUMNS)
    if (newSortScope.isDefined) {
      // 1. check SORT_COLUMNS when SORT_SCOPE is not changed to NO_SORT
      if (!SortScope.NO_SORT.name().equalsIgnoreCase(newSortScope.get)) {
        if (newSortColumns.isDefined) {
          if (StringUtils.isBlank(CarbonUtil.unquoteChar(newSortColumns.get))) {
            throw new InvalidConfigurationException(
              s"Cannot set SORT_COLUMNS as empty when setting SORT_SCOPE as ${newSortScope.get} ")
          }
        } else {
          if (carbonTable.getNumberOfSortColumns == 0) {
            throw new InvalidConfigurationException(
              s"Cannot set SORT_SCOPE as ${newSortScope.get} when table has no SORT_COLUMNS")
          }
        }
      }
    } else if (newSortColumns.isDefined) {
      // 2. check SORT_SCOPE when SORT_COLUMNS is changed to empty
      if (StringUtils.isBlank(CarbonUtil.unquoteChar(newSortColumns.get))) {
        if (!SortScope.NO_SORT.equals(carbonTable.getSortScope)) {
          throw new InvalidConfigurationException(
            s"Cannot set SORT_COLUMNS as empty when SORT_SCOPE is ${carbonTable.getSortScope} ")
        }
      }
    }
  }

  /**
   * This method will validate if there is any complex type column in the columns to be cached
   *
   * @param carbonTable
   * @param cachedColumns
   */
  private def validateForComplexTypeColumn(carbonTable: CarbonTable,
      cachedColumns: String): Unit = {
    if (cachedColumns.nonEmpty) {
      cachedColumns.split(",").foreach { column =>
        val dimension = carbonTable.getDimensionByName(column)
        if (null != dimension && dimension.isComplex) {
          val errorMessage =
            s"$column is a complex type column and complex type is not allowed for " +
            s"the option(s): ${ CarbonCommonConstants.COLUMN_META_CACHE }"
          throw new MalformedCarbonCommandException(errorMessage)
        }
      }
    }
  }

  /**
   * This method will check and clear the driver block/blocklet cache
   *
   * @param carbonTable
   * @param existingTableProperties
   * @param newProperties
   * @param propKeys
   * @param set
   */
  private def checkAndClearBlockletCache(carbonTable: CarbonTable,
      existingTableProperties: scala.collection.mutable.Map[String, String],
      newProperties: mutable.Map[String, String],
      propKeys: Seq[String],
      set: Boolean): Unit = {
    if (set) {
      clearBlockletCacheForCachingProperties(carbonTable, existingTableProperties, newProperties)
    } else {
      // convert all the unset keys to lower case
      val propertiesToBeRemoved = propKeys.map(key => key.toLowerCase)
      // This is unset scenario and the cache needs to be cleaned only when
      // 1. column_meta_cache property is getting unset and existing table properties contains
      // this property
      // 2. cache_level property is being unset and existing table properties contains this property
      // and the existing value is not equal to default value because after un-setting the property
      // the cache should be loaded again with default value
      if (propertiesToBeRemoved.contains(CarbonCommonConstants.COLUMN_META_CACHE) &&
          existingTableProperties.get(CarbonCommonConstants.COLUMN_META_CACHE).isDefined) {
        clearCache(carbonTable)
      } else if (propertiesToBeRemoved.contains(CarbonCommonConstants.CACHE_LEVEL)) {
        val cacheLevel = existingTableProperties.get(CarbonCommonConstants.CACHE_LEVEL)
        if (cacheLevel.isDefined &&
            !cacheLevel.equals(CarbonCommonConstants.CACHE_LEVEL_DEFAULT_VALUE)) {
          clearCache(carbonTable)
        }
      }
    }
  }

  private def clearCache(carbonTable: CarbonTable): Unit = {
    // clear dataMap cache
    DataMapStoreManager.getInstance().clearDataMaps(carbonTable.getAbsoluteTableIdentifier)
    // clear segmentProperties Cache
    SegmentPropertiesAndSchemaHolder.getInstance()
      .invalidate(carbonTable.getAbsoluteTableIdentifier)
  }

  /**
   * This method will validate the column_meta_cache and cache_level properties and clear the
   * driver block/blocklet cache
   *
   * @param carbonTable
   * @param tblPropertiesMap
   * @param newProperties
   */
  private def clearBlockletCacheForCachingProperties(
      carbonTable: CarbonTable,
      tblPropertiesMap: scala.collection.mutable.Map[String, String],
      newProperties: mutable.Map[String, String]): Unit = {
    // check if column meta cache is defined. if defined then validate and clear the BTree cache
    // if required
    val columnMetaCacheProperty = newProperties.get(CarbonCommonConstants.COLUMN_META_CACHE)
    columnMetaCacheProperty match {
      case Some(newColumnsToBeCached) =>
        if (!checkIfColumnsAreAlreadyCached(carbonTable, tblPropertiesMap
          .get(CarbonCommonConstants.COLUMN_META_CACHE), newColumnsToBeCached)) {
          clearCache(carbonTable)
        }
      case None =>
      // don't do anything
    }
    // check if column meta cache is defined. if defined then validate and clear the BTree cache
    // if required
    val cacheLevelProperty = newProperties.get(CarbonCommonConstants.CACHE_LEVEL)
    cacheLevelProperty match {
      case Some(newCacheLevelValue) =>
        if (!isCacheLevelValid(tblPropertiesMap.get(CarbonCommonConstants.CACHE_LEVEL),
          newCacheLevelValue)) {
          clearCache(carbonTable)
        }
      case None =>
      // don't do anything
    }
  }

  /**
   * Method to verify if the existing cache level is same as the new cache level
   *
   * @param existingCacheLevelValue
   * @param newCacheLevelValue
   * @return
   */
  private def isCacheLevelValid(existingCacheLevelValue: Option[String],
      newCacheLevelValue: String): Boolean = {
    existingCacheLevelValue match {
      case Some(existingValue) =>
        existingValue.equals(newCacheLevelValue)
      case None =>
        false
    }
  }

  /**
   * Check the new columns to be cached with the already cached columns. If count of new columns
   * and already cached columns is same and all the new columns are already cached then
   * false will be returned else true
   *
   * @param carbonTable
   * @param existingCacheColumns
   * @param newColumnsToBeCached
   * @return
   */
  private def checkIfColumnsAreAlreadyCached(
      carbonTable: CarbonTable,
      existingCacheColumns: Option[String],
      newColumnsToBeCached: String): Boolean = {
    val newColumns = newColumnsToBeCached.split(",").map(x => x.trim.toLowerCase)
    val isCached = existingCacheColumns match {
      case Some(value) =>
        val existingProperty = value.split(",").map(x => x.trim.toLowerCase)
        compareColumns(existingProperty, newColumns)
      case None =>
        // By default all the columns in the table will be cached. This case is to compare all the
        // table columns already cached to the newly specified cached columns
        val schemaList: util.List[ColumnSchema] = CarbonUtil
          .getColumnSchemaList(carbonTable.getVisibleDimensions, carbonTable.getVisibleMeasures)
        val tableColumns: Array[String] = schemaList.asScala
          .map(columnSchema => columnSchema.getColumnName).toArray
        compareColumns(tableColumns, newColumns)
    }
    isCached
  }

  /**
   * compare the existing cache columns and the new columns to be cached
   *
   * @param existingCachedColumns
   * @param newColumnsToBeCached
   * @return
   */
  private def compareColumns(existingCachedColumns: Array[String],
      newColumnsToBeCached: Array[String]): Boolean = {
    val allColumnsMatch = if (existingCachedColumns.length == newColumnsToBeCached.length) {
      existingCachedColumns.filter(col => !newColumnsToBeCached.contains(col)).length == 0
    } else {
      false
    }
    allColumnsMatch
  }

  /**
   * Validate LOCAL_DICT_COLUMNS property specified in Alter command
   * @param tblPropertiesMap
   * @param carbonTable
   * @param property
   */
  def ValidateSetTablePropertiesForLocalDict(tblPropertiesMap: mutable.Map[String, String],
      carbonTable: CarbonTable,
      property: (String, String)): Unit = {
    var primitiveComplexChildColumns = new mutable.HashSet[String]
    var localDictColumns: Seq[String] = Seq[String]()
    var dictIncludeColumns: Seq[String] = Seq[String]()

    val allColumns = carbonTable.getTableInfo.getFactTable.getListOfColumns.asScala
    localDictColumns = property._2.toString.toLowerCase.split(",").map(_.trim)

    CarbonScalaUtil.validateLocalDictionaryColumns(tblPropertiesMap, localDictColumns)

    // check if the column specified exists in table schema
    localDictColumns.foreach { distCol =>
      if (!allColumns.exists(x => x.getColumnName.equalsIgnoreCase(distCol.trim))) {
        val errorMsg = "LOCAL_DICTIONARY_INCLUDE/LOCAL_DICTIONARY_EXCLUDE column: " + distCol.trim +
                       " does not exist in table. Please check the DDL."
        throw new MalformedCarbonCommandException(errorMsg)
      }
    }

    /**
     * Verify if specified column is of no-dictionary string or varchar dataType
     */
    localDictColumns.foreach { dictCol =>
      if (allColumns.exists(col => col.getColumnName.equalsIgnoreCase(dictCol) &&
                                   !col.getDataType.toString
                                     .equalsIgnoreCase("STRING") &&
                                   !col.getDataType.toString
                                     .equalsIgnoreCase("VARCHAR") &&
                                   !col.getDataType.toString
                                     .equalsIgnoreCase("STRUCT") &&
                                   !col.getDataType.toString
                                     .equalsIgnoreCase("ARRAY"))) {
        val errMsg = "LOCAL_DICTIONARY_INCLUDE/LOCAL_DICTIONARY_EXCLUDE column: " + dictCol.trim +
                     " is not a string/complex/varchar datatype column. LOCAL_DICTIONARY_INCLUDE" +
                     "/LOCAL_DICTIONARY_EXCLUDE should be no " +
                     "dictionary string/complex/varchar datatype column."
        throw new MalformedCarbonCommandException(errMsg)
      }
    }
    var countOfDictCols = 0
    var trav = 0
    // Validate whether any of the child columns of complex dataType column is a string or
    // varchar dataType column
    if (property._1.equalsIgnoreCase(CarbonCommonConstants.LOCAL_DICTIONARY_INCLUDE)) {
      // Validate whether any of the child columns of complex dataType column is a string column
      localDictColumns.foreach { dictColm =>
        for (elem <- allColumns.indices) {
          var column = allColumns(elem)
          if (column.getColumnName.equalsIgnoreCase(dictColm) && column.getNumberOfChild > 0 &&
              !validateChildColumns(allColumns, column.getNumberOfChild, elem. +(1))) {
            val errMsg =
              "None of the child columns specified in the complex dataType column(s) in " +
              "local_dictionary_include are not of string dataType."
            throw new MalformedCarbonCommandException(errMsg)
          }
        }
      }
    }

    /**
     * check whether any child column present in complex type column is string or varchar type
     *
     * @param schemas
     * @return
     */
    def validateChildColumns(schemas: mutable.Buffer[ColumnSchema],
        colCount: Int, traverse: Int): Boolean = {
      trav = traverse
      var column: ColumnSchema = null
      for (i <- 0 until colCount) {
        column = schemas(trav)
        if (column.getNumberOfChild > 0) {
          validateChildColumns(schemas, column.getNumberOfChild, trav. +(1))
        } else {
          if (column.isDimensionColumn && (column.getDataType.equals(DataTypes.STRING) ||
                                           column.getDataType.equals(DataTypes.VARCHAR))) {
            countOfDictCols += 1
          }
          trav = trav + 1
        }
      }
      if (countOfDictCols > 0) {
        return true
      }
      false
    }
  }

  private def validateLoadMinSizeProperties(carbonTable: CarbonTable,
      propertiesMap: mutable.Map[String, String]): Unit = {
    // validate load min size property
    if (propertiesMap.get(CarbonCommonConstants.CARBON_LOAD_MIN_SIZE_INMB).isDefined) {
      CommonUtil.validateLoadMinSize(propertiesMap, CarbonCommonConstants.CARBON_LOAD_MIN_SIZE_INMB)
    }
  }
}