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

import java.util.UUID

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

import org.apache.spark.sql.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, CarbonLoadDataCommand}
import org.apache.spark.sql.execution.command.partition.CarbonAlterTableDropHivePartitionCommand
import org.apache.spark.sql.parser.CarbonSpark2SqlParser

import org.apache.carbondata.common.exceptions.MetadataProcessException
import org.apache.carbondata.common.logging.{LogService, LogServiceFactory}
import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.datamap.Segment
import org.apache.carbondata.core.datastore.filesystem.{CarbonFile, CarbonFileFilter}
import org.apache.carbondata.core.datastore.impl.FileFactory
import org.apache.carbondata.core.locks.{CarbonLockUtil, ICarbonLock}
import org.apache.carbondata.core.metadata.schema.table.{AggregationDataMapSchema, CarbonTable}
import org.apache.carbondata.core.statusmanager.{SegmentStatus, SegmentStatusManager}
import org.apache.carbondata.core.util.CarbonUtil
import org.apache.carbondata.core.util.path.CarbonTablePath
import org.apache.carbondata.events._
import org.apache.carbondata.processing.loading.events.LoadEvents.{LoadMetadataEvent, LoadTablePostExecutionEvent, LoadTablePostStatusUpdateEvent, LoadTablePreExecutionEvent, LoadTablePreStatusUpdateEvent}
import org.apache.carbondata.processing.merger.CompactionType

object AlterTableDropPartitionPreStatusListener 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 && carbonTable.hasAggregationDataMap) {
      val childCommands =
        childDropPartitionCommands.asInstanceOf[Seq[CarbonAlterTableDropHivePartitionCommand]]
      childCommands.foreach(_.processData(SparkSession.getActiveSession.get))
    }
  }
}

trait CommitHelper {

  val LOGGER = LogServiceFactory.getLogService(this.getClass.getName)

  protected def markInProgressSegmentAsDeleted(tableStatusFile: String,
      operationContext: OperationContext,
      carbonTable: CarbonTable): Unit = {
    val lockFile: ICarbonLock = new SegmentStatusManager(carbonTable
      .getAbsoluteTableIdentifier).getTableStatusLock
    val retryCount = CarbonLockUtil
      .getLockProperty(CarbonCommonConstants.NUMBER_OF_TRIES_FOR_CONCURRENT_LOCK,
        CarbonCommonConstants.NUMBER_OF_TRIES_FOR_CONCURRENT_LOCK_DEFAULT)
    val maxTimeout = CarbonLockUtil
      .getLockProperty(CarbonCommonConstants.MAX_TIMEOUT_FOR_CONCURRENT_LOCK,
        CarbonCommonConstants.MAX_TIMEOUT_FOR_CONCURRENT_LOCK_DEFAULT)
    try {
      if (lockFile.lockWithRetries(retryCount, maxTimeout)) {
        val loadMetaDataDetails = SegmentStatusManager.readTableStatusFile(tableStatusFile)
        val segmentBeingLoaded =
          operationContext.getProperty(carbonTable.getTableUniqueName + "_Segment").toString
        val newDetails = loadMetaDataDetails.collect {
          case detail if detail.getLoadName.equalsIgnoreCase(segmentBeingLoaded) =>
            detail.setSegmentStatus(SegmentStatus.MARKED_FOR_DELETE)
            detail
          case others => others
        }
        SegmentStatusManager.writeLoadDetailsIntoFile(tableStatusFile, newDetails)
      } else {
        throw new RuntimeException("Uable to update table status file")
      }
    } finally {
      lockFile.unlock()
    }
  }

  /**
   *  Used to rename table status files for commit operation.
   */
  protected def renameDataMapTableStatusFiles(sourceFileName: String,
      destinationFileName: String, uuid: String): Boolean = {
    val oldCarbonFile = FileFactory.getCarbonFile(sourceFileName)
    val newCarbonFile = FileFactory.getCarbonFile(destinationFileName)
    if (oldCarbonFile.exists() && newCarbonFile.exists()) {
      val backUpPostFix = if (uuid.nonEmpty) {
        "_backup_" + uuid
      } else {
        ""
      }
      LOGGER.info(s"Renaming $newCarbonFile to ${destinationFileName + backUpPostFix}")
      if (newCarbonFile.renameForce(destinationFileName + backUpPostFix)) {
        LOGGER.info(s"Renaming $oldCarbonFile to $destinationFileName")
        oldCarbonFile.renameForce(destinationFileName)
      } else {
        LOGGER.info(s"Renaming $newCarbonFile to ${destinationFileName + backUpPostFix} failed")
        false
      }
    } else {
     /**
      * Tablestatus_uuid will fail when Pre-Aggregate table is not valid for compaction.
      * Hence this should return true
      */
      true
    }
  }

  def mergeTableStatusContents(carbonTable: CarbonTable, uuidTableStatusPath: String,
      tableStatusPath: String): Boolean = {
    val lockFile: ICarbonLock = new SegmentStatusManager(carbonTable
      .getAbsoluteTableIdentifier).getTableStatusLock
    try {
      val retryCount = CarbonLockUtil
        .getLockProperty(CarbonCommonConstants.NUMBER_OF_TRIES_FOR_CONCURRENT_LOCK,
          CarbonCommonConstants.NUMBER_OF_TRIES_FOR_CONCURRENT_LOCK_DEFAULT)
      val maxTimeout = CarbonLockUtil
        .getLockProperty(CarbonCommonConstants.MAX_TIMEOUT_FOR_CONCURRENT_LOCK,
          CarbonCommonConstants.MAX_TIMEOUT_FOR_CONCURRENT_LOCK_DEFAULT)
      if (lockFile.lockWithRetries(retryCount, maxTimeout)) {
        val tableStatusContents = SegmentStatusManager.readTableStatusFile(tableStatusPath)
        val newLoadContent = SegmentStatusManager.readTableStatusFile(uuidTableStatusPath)
        val mergedContent = tableStatusContents.collect {
          case content =>
            val contentIndex = newLoadContent.indexOf(content)
            if (contentIndex == -1) {
              content
            } else {
              newLoadContent(contentIndex)
            }
        }
        SegmentStatusManager.writeLoadDetailsIntoFile(tableStatusPath, mergedContent)
        true
      } else {
        false
      }
    } catch {
      case ex: Exception =>
        LOGGER.error("Exception occurred while merging files", ex)
        false
    } finally {
      lockFile.unlock()
    }
  }

  /**
   * Used to remove table status files with UUID and segment folders.
   */
  protected def cleanUpStaleTableStatusFiles(
      childTables: Seq[CarbonTable],
      operationContext: OperationContext,
      uuid: String): Unit = {
    childTables.foreach { childTable =>
      val metaDataDir = FileFactory.getCarbonFile(
        CarbonTablePath.getMetadataPath(childTable.getTablePath))
      val tableStatusFiles = metaDataDir.listFiles(new CarbonFileFilter {
        override def accept(file: CarbonFile): Boolean = {
          file.getName.contains(uuid) || file.getName.contains("backup")
        }
      })
      tableStatusFiles.foreach(_.delete())
    }
  }
}

object AlterTableDropPartitionPostStatusListener extends OperationEventListener with CommitHelper {
  /**
   * Called on a specified event occurrence
   *
   * @param event
   * @param operationContext
   */
  override protected def onEvent(event: Event,
      operationContext: OperationContext) = {
    val postStatusListener = event.asInstanceOf[AlterTableDropPartitionPostStatusEvent]
    val carbonTable = postStatusListener.carbonTable
    val childDropPartitionCommands = operationContext.getProperty("dropPartitionCommands")
    val uuid = Option(operationContext.getProperty("uuid")).getOrElse("").toString
    if (childDropPartitionCommands != null && carbonTable.hasAggregationDataMap) {
      val childCommands =
        childDropPartitionCommands.asInstanceOf[Seq[CarbonAlterTableDropHivePartitionCommand]]
      val updateFailed = try {
        val renamedDataMaps = childCommands.takeWhile {
          childCommand =>
            val childCarbonTable = childCommand.table
            val oldTableSchemaPath = CarbonTablePath.getTableStatusFilePathWithUUID(
              childCarbonTable.getTablePath, uuid)
            // Generate table status file name without UUID, forExample: tablestatus
            val newTableSchemaPath = CarbonTablePath.getTableStatusFilePath(
              childCarbonTable.getTablePath)
            mergeTableStatusContents(childCarbonTable, oldTableSchemaPath, newTableSchemaPath)
        }
        // if true then the commit for one of the child tables has failed
        val commitFailed = renamedDataMaps.lengthCompare(childCommands.length) != 0
        if (commitFailed) {
          LOGGER.warn("Reverting table status file to original state")
          childCommands.foreach {
            childDropCommand =>
              val tableStatusPath = CarbonTablePath.getTableStatusFilePath(
                childDropCommand.table.getTablePath)
              markInProgressSegmentAsDeleted(tableStatusPath,
                operationContext,
                childDropCommand.table)
          }
        }
        commitFailed
      } finally {
        // after success/failure of commit delete all tablestatus files with UUID in their names.
        // if commit failed then remove the segment directory
        // TODO Need to handle deletion on tablestatus files with UUID in cleanup command.
        cleanUpStaleTableStatusFiles(childCommands.map(_.table),
          operationContext,
          uuid)
      }
      if (updateFailed) {
        sys.error("Failed to update table status for pre-aggregate table")
      }
    }
  }
}

object AlterTableDropPartitionMetaListener extends OperationEventListener{
  /**
   * Called on a specified event occurrence
   *
   * @param event
   * @param operationContext
   */
  override protected def onEvent(event: Event,
      operationContext: OperationContext) = {
    val dropPartitionEvent = event.asInstanceOf[AlterTableDropPartitionMetaEvent]
    val parentCarbonTable = dropPartitionEvent.parentCarbonTable
    val partitionsToBeDropped = dropPartitionEvent.specs.flatMap(_.keys)
    if (parentCarbonTable.hasAggregationDataMap) {
      // used as a flag to block direct drop partition on aggregate tables fired by the user
      operationContext.setProperty("isInternalDropCall", "true")
      // Filter out all the tables which don't have the partition being dropped.
      val childTablesWithoutPartitionColumns =
        parentCarbonTable.getTableInfo.getDataMapSchemaList.asScala.filter { dataMapSchema =>
          val childColumns = dataMapSchema.getChildSchema.getListOfColumns.asScala
          val partitionColExists = partitionsToBeDropped.forall {
            partition =>
              childColumns.exists { childColumn =>
                  childColumn.getAggFunction.isEmpty &&
                    childColumn.getParentColumnTableRelations != null &&
                  childColumn.getParentColumnTableRelations.asScala.head.getColumnName.
                    equals(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(_.getChildSchema.getTableName)
        }. Please drop the specified aggregate tables to continue")
      } else {
        val childDropPartitionCommands =
          parentCarbonTable.getTableInfo.getDataMapSchemaList.asScala.map { dataMapSchema =>
            val tableIdentifier = TableIdentifier(dataMapSchema.getChildSchema.getTableName,
              Some(parentCarbonTable.getDatabaseName))
            // as the aggregate 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.isChildDataMap) {
      if (operationContext.getProperty("isInternalDropCall") == null) {
        throw new UnsupportedOperationException("Cannot drop partition directly on aggregate table")
      }
    }
  }
}

/**
 * below class will be used to create load command for compaction
 * for all the pre agregate child data map
 */
object CompactionProcessMetaListener extends OperationEventListener {
  /**
   * Called on a specified event occurrence
   *
   * @param event
   * @param operationContext
   */
  override protected def onEvent(event: Event,
      operationContext: OperationContext): Unit = {
    val sparkSession = SparkSession.getActiveSession.get
    val tableEvent = event.asInstanceOf[LoadMetadataEvent]
    val table = tableEvent.getCarbonTable
    if (!table.isChildDataMap && CarbonUtil.hasAggregationDataMap(table)) {
      val aggregationDataMapList = table.getTableInfo.getDataMapSchemaList.asScala
        .filter(_.isInstanceOf[AggregationDataMapSchema])
        .asInstanceOf[mutable.ArrayBuffer[AggregationDataMapSchema]]
      for (dataMapSchema: AggregationDataMapSchema <- aggregationDataMapList) {
        val childTableName = dataMapSchema.getRelationIdentifier.getTableName
        val childDatabaseName = dataMapSchema.getRelationIdentifier.getDatabaseName
        // Creating a new query string to insert data into pre-aggregate table from that same table.
        // For example: To compact preaggtable1 we can fire a query like insert into preaggtable1
        // select * from preaggtable1
        // The following code will generate the select query with a load UDF that will be used to
        // apply DataLoadingRules
        val childDataFrame = sparkSession.sql(new CarbonSpark2SqlParser()
          // adding the aggregation load UDF
          .addPreAggLoadFunction(
          // creating the select query on the bases on table schema
          PreAggregateUtil.createChildSelectQuery(
            dataMapSchema.getChildSchema, table.getDatabaseName))).drop("preAggLoad")
        val loadCommand = PreAggregateUtil.createLoadCommandForChild(
          dataMapSchema.getChildSchema.getListOfColumns,
          TableIdentifier(childTableName, Some(childDatabaseName)),
          childDataFrame,
          false,
          sparkSession,
          mutable.Map.empty[String, String])
        val uuid = Option(operationContext.getProperty("uuid")).
          getOrElse(UUID.randomUUID()).toString
        operationContext.setProperty("uuid", uuid)
        loadCommand.processMetadata(sparkSession)
        operationContext
          .setProperty(dataMapSchema.getChildSchema.getTableName + "_Compaction", loadCommand)
        loadCommand.operationContext = operationContext
      }
    } else if (table.isChildDataMap) {
      val childTableName = table.getTableName
      val childDatabaseName = table.getDatabaseName
      // Creating a new query string to insert data into pre-aggregate table from that same table.
      // For example: To compact preaggtable1 we can fire a query like insert into preaggtable1
      // select * from preaggtable1
      // The following code will generate the select query with a load UDF that will be used to
      // apply DataLoadingRules
      val childDataFrame = sparkSession.sql(new CarbonSpark2SqlParser()
        // adding the aggregation load UDF
        .addPreAggLoadFunction(
        // creating the select query on the bases on table schema
        PreAggregateUtil.createChildSelectQuery(
          table.getTableInfo.getFactTable, table.getDatabaseName))).drop("preAggLoad")
      val loadCommand = PreAggregateUtil.createLoadCommandForChild(
        table.getTableInfo.getFactTable.getListOfColumns,
        TableIdentifier(childTableName, Some(childDatabaseName)),
        childDataFrame,
        false,
        sparkSession,
        mutable.Map.empty[String, String])
      val uuid = Option(operationContext.getProperty("uuid")).getOrElse(UUID.randomUUID()).toString
      loadCommand.processMetadata(sparkSession)
      operationContext.setProperty(table.getTableName + "_Compaction", loadCommand)
      operationContext.setProperty("uuid", uuid)
      loadCommand.operationContext = operationContext
    }

  }
}

/**
 * Below class to is to create LoadCommand for loading the
 * the data of pre aggregate data map
 */
object LoadProcessMetaListener extends OperationEventListener {
  /**
   * Called on a specified event occurrence
   *
   * @param event
   * @param operationContext
   */
  override protected def onEvent(event: Event,
      operationContext: OperationContext): Unit = {
    val sparkSession = SparkSession.getActiveSession.get
    val tableEvent = event.asInstanceOf[LoadMetadataEvent]
    if (!tableEvent.isCompaction) {
      val table = tableEvent.getCarbonTable
      if (CarbonUtil.hasAggregationDataMap(table)) {
        // getting all the aggergate datamap schema
        val aggregationDataMapList = table.getTableInfo.getDataMapSchemaList.asScala
          .filter(_.isInstanceOf[AggregationDataMapSchema])
          .asInstanceOf[mutable.ArrayBuffer[AggregationDataMapSchema]]
        // sorting the datamap for timeseries rollup
        val sortedList = aggregationDataMapList.sortBy(_.getOrdinal)
        val parentTableName = table.getTableName
        val databaseName = table.getDatabaseName
        // if the table is child then extract the uuid from the operation context and the parent
        // would already generated UUID.
        // if parent table then generate a new UUID else use empty.
        val uuid =
          Option(operationContext.getProperty("uuid")).getOrElse(UUID.randomUUID()).toString
        val list = scala.collection.mutable.ListBuffer.empty[AggregationDataMapSchema]
        for (dataMapSchema: AggregationDataMapSchema <- sortedList) {
          val childTableName = dataMapSchema.getRelationIdentifier.getTableName
          val childDatabaseName = dataMapSchema.getRelationIdentifier.getDatabaseName
          val childSelectQuery = if (!dataMapSchema.isTimeseriesDataMap) {
            (PreAggregateUtil.getChildQuery(dataMapSchema), "")
          } else {
            // for timeseries rollup policy
            val tableSelectedForRollup = PreAggregateUtil.getRollupDataMapNameForTimeSeries(list,
              dataMapSchema)
            list += dataMapSchema
            // if non of the rollup data map is selected hit the maintable and prepare query
            if (tableSelectedForRollup.isEmpty) {
              (PreAggregateUtil.createTimeSeriesSelectQueryFromMain(dataMapSchema.getChildSchema,
                parentTableName,
                databaseName), "")
            } else {
              // otherwise hit the select rollup datamap schema
              (PreAggregateUtil.createTimeseriesSelectQueryForRollup(dataMapSchema.getChildSchema,
                tableSelectedForRollup.get,
                databaseName),
                s"$databaseName.${tableSelectedForRollup.get.getChildSchema.getTableName}")
            }
          }
          val childDataFrame = sparkSession.sql(new CarbonSpark2SqlParser().addPreAggLoadFunction(
            childSelectQuery._1)).drop("preAggLoad")
          val isOverwrite =
            operationContext.getProperty("isOverwrite").asInstanceOf[Boolean]
          val loadCommand = PreAggregateUtil.createLoadCommandForChild(
            dataMapSchema.getChildSchema.getListOfColumns,
            TableIdentifier(childTableName, Some(childDatabaseName)),
            childDataFrame,
            isOverwrite,
            sparkSession,
            tableEvent.getOptions.asScala,
            timeseriesParentTableName = childSelectQuery._2)
          operationContext.setProperty("uuid", uuid)
          loadCommand.operationContext.setProperty("uuid", uuid)
          loadCommand.processMetadata(sparkSession)
          operationContext.setProperty(dataMapSchema.getChildSchema.getTableName, loadCommand)
        }
      }
    }
  }
}
object LoadPostAggregateListener extends OperationEventListener {
  /**
   * Called on a specified event occurrence
   *
   * @param event
   */
  override def onEvent(event: Event, operationContext: OperationContext): Unit = {
    val carbonLoadModelOption =
      event match {
        case e: LoadTablePreStatusUpdateEvent => Some(e.getCarbonLoadModel)
        case e: LoadTablePostExecutionEvent => Some(e.getCarbonLoadModel)
        case _ => None
      }
    val sparkSession = SparkSession.getActiveSession.get
    if (carbonLoadModelOption.isDefined) {
      val carbonLoadModel = carbonLoadModelOption.get
      val table = carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable
      if (CarbonUtil.hasAggregationDataMap(table)) {
        val isOverwrite =
          operationContext.getProperty("isOverwrite").asInstanceOf[Boolean]
        if (isOverwrite && table.isHivePartitionTable) {
          val parentPartitionColumns = table.getPartitionInfo.getColumnSchemaList.asScala
            .map(_.getColumnName)
          val childTablesWithoutPartitionColumns =
            table.getTableInfo.getDataMapSchemaList.asScala.filter { dataMapSchema =>
              val childColumns = dataMapSchema.getChildSchema.getListOfColumns.asScala
              val partitionColExists = parentPartitionColumns.forall {
                partition =>
                  childColumns.exists { childColumn =>
                    childColumn.getAggFunction.isEmpty &&
                    childColumn.getParentColumnTableRelations.asScala.head.getColumnName.
                      equals(partition)
                  }
              }
              !partitionColExists
            }
          if (childTablesWithoutPartitionColumns.nonEmpty) {
            throw new MetadataProcessException(
              "Cannot execute load overwrite or insert overwrite as the following aggregate tables"
              + s" ${
                childTablesWithoutPartitionColumns.toList.map(_.getChildSchema.getTableName)
              } are not partitioned on all the partition column. Drop these to continue")
          }
        }
        // getting all the aggergate datamap schema
        val aggregationDataMapList = table.getTableInfo.getDataMapSchemaList.asScala
          .filter(_.isInstanceOf[AggregationDataMapSchema])
          .asInstanceOf[mutable.ArrayBuffer[AggregationDataMapSchema]]
        // sorting the datamap for timeseries rollup
        val sortedList = aggregationDataMapList.sortBy(_.getOrdinal)
        val successDataMaps = sortedList.takeWhile { dataMapSchema =>
          val childLoadCommand = operationContext
            .getProperty(dataMapSchema.getChildSchema.getTableName)
            .asInstanceOf[CarbonLoadDataCommand]
          childLoadCommand.dataFrame = Some(PreAggregateUtil
            .getDataFrame(sparkSession, childLoadCommand.logicalPlan.get))
          childLoadCommand.operationContext = operationContext
          val timeseriesParent = childLoadCommand.internalOptions.get("timeseriesParent")
          val (parentTableIdentifier, segmentToLoad) =
            if (timeseriesParent.isDefined && timeseriesParent.get.nonEmpty) {
              val (parentTableDatabase, parentTableName) =
                (timeseriesParent.get.split('.')(0), timeseriesParent.get.split('.')(1))
              (TableIdentifier(parentTableName, Some(parentTableDatabase)),
                operationContext.getProperty(
                  s"${ parentTableDatabase }_${ parentTableName }_Segment").toString)
            } else {
              val currentSegmentFile = operationContext.getProperty("current.segmentfile")
              val segment = if (currentSegmentFile != null) {
                new Segment(carbonLoadModel.getSegmentId, currentSegmentFile.toString)
              } else {
                Segment.toSegment(carbonLoadModel.getSegmentId, null)
              }
              (TableIdentifier(table.getTableName, Some(table.getDatabaseName)), segment.toString)
            }

        PreAggregateUtil.startDataLoadForDataMap(
        parentTableIdentifier,
            segmentToLoad,
            validateSegments = false,
            childLoadCommand,
            isOverwrite,
            sparkSession)
        }
        val loadFailed = successDataMaps.lengthCompare(sortedList.length) != 0
        if (loadFailed) {
          successDataMaps.foreach(dataMapSchema => markSuccessSegmentsAsFailed(operationContext
            .getProperty(dataMapSchema.getChildSchema.getTableName)
            .asInstanceOf[CarbonLoadDataCommand]))
          throw new RuntimeException(
            "Data Load failed for DataMap. Please check logs for the failure")
        }
      }
    }
  }

  private def markSuccessSegmentsAsFailed(childLoadCommand: CarbonLoadDataCommand) {
    val segmentToRevert = childLoadCommand.operationContext
      .getProperty(childLoadCommand.table.getTableUniqueName + "_Segment")
    val tableStatusPath = CarbonTablePath.getTableStatusFilePath(
      childLoadCommand.table.getTablePath)
    val tableStatusContents = SegmentStatusManager.readTableStatusFile(tableStatusPath)
    val updatedLoadDetails = tableStatusContents.collect {
      case content if content.getLoadName == segmentToRevert =>
        content.setSegmentStatus(SegmentStatus.MARKED_FOR_DELETE)
        content
      case others => others
    }
    SegmentStatusManager.writeLoadDetailsIntoFile(tableStatusPath, updatedLoadDetails)
  }
}

/**
 * This listener is used to commit all the child data aggregate tables in one transaction. If one
 * failes all will be reverted to original state.
 */
object CommitPreAggregateListener extends OperationEventListener with CommitHelper {

  override protected def onEvent(event: Event,
      operationContext: OperationContext): Unit = {
    // The same listener is called for both compaction and load therefore getting the
    // carbonLoadModel from the appropriate event.
    val carbonLoadModel = event match {
      case loadEvent: LoadTablePostStatusUpdateEvent =>
        loadEvent.getCarbonLoadModel
      case compactionEvent: AlterTableCompactionPostStatusUpdateEvent =>
        // Once main table compaction is done and 0.1, 4.1, 8.1 is created commit will happen for
        // all the tables. The commit listener will compact the child tables until no more segments
        // are left. But 2nd level compaction is yet to happen on the main table therefore again the
        // compaction flow will try to commit the child tables which is wrong. This check tell the
        // 2nd level compaction flow that the commit for datamaps is already done.
        if (null != operationContext.getProperty("commitComplete")) {
          return
        }
        compactionEvent.carbonLoadModel
    }
    val isCompactionFlow = Option(
      operationContext.getProperty("isCompaction")).getOrElse("false").toString.toBoolean
    val dataMapSchemas =
      carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable.getTableInfo.getDataMapSchemaList
      .asScala.filter(_.getChildSchema != null)
    // extract all child LoadCommands
    val childLoadCommands = if (!isCompactionFlow) {
      // If not compaction flow then the key for load commands will be tableName
        dataMapSchemas.map { dataMapSchema =>
          operationContext.getProperty(dataMapSchema.getChildSchema.getTableName)
            .asInstanceOf[CarbonLoadDataCommand]
        }
      } else {
      // If not compaction flow then the key for load commands will be tableName_Compaction
        dataMapSchemas.map { dataMapSchema =>
          operationContext.getProperty(dataMapSchema.getChildSchema.getTableName + "_Compaction")
            .asInstanceOf[CarbonLoadDataCommand]
        }
    }
    var commitFailed = false
    try {
      if (dataMapSchemas.nonEmpty) {
        val uuid = operationContext.getProperty("uuid").toString
        // keep committing until one fails
        val renamedDataMaps = childLoadCommands.takeWhile { childLoadCommand =>
          val childCarbonTable = childLoadCommand.table
          // Generate table status file name with UUID, forExample: tablestatus_1
          val oldTableSchemaPath = CarbonTablePath.getTableStatusFilePathWithUUID(
            childCarbonTable.getTablePath, uuid)
          // Generate table status file name without UUID, forExample: tablestatus
          val newTableSchemaPath = CarbonTablePath.getTableStatusFilePath(
            childCarbonTable.getTablePath)
          mergeTableStatusContents(childCarbonTable, oldTableSchemaPath, newTableSchemaPath)
        }
        // if true then the commit for one of the child tables has failed
        commitFailed = renamedDataMaps.lengthCompare(dataMapSchemas.length) != 0
        if (commitFailed) {
          LOGGER.warn("Reverting table status file to original state")
          childLoadCommands.foreach {
            childLoadCommand =>
              val tableStatusPath = CarbonTablePath.getTableStatusFilePath(
                childLoadCommand.table.getTablePath)
              markInProgressSegmentAsDeleted(tableStatusPath,
                operationContext,
                childLoadCommand.table)
          }
        }
        // after success/failure of commit delete all tablestatus files with UUID in their names.
        // if commit failed then remove the segment directory
        cleanUpStaleTableStatusFiles(childLoadCommands.map(_.table),
          operationContext,
          uuid)
        operationContext.setProperty("commitComplete", !commitFailed)
        if (commitFailed) {
          sys.error("Failed to update table status for pre-aggregate table")
        }
      }
    } catch {
      case e: Exception =>
        operationContext.setProperty("commitComplete", false)
        LOGGER.error("Problem while committing data maps", e)
    }
  }
}

/**
 * Listener to handle the operations that have to be done after compaction for a table has finished.
 */
object AlterPreAggregateTableCompactionPostListener 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
    if (CarbonUtil.hasAggregationDataMap(carbonTable)) {
      carbonTable.getTableInfo.getDataMapSchemaList.asScala.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)
      }
    }
  }
}

object LoadPreAggregateTablePreListener extends OperationEventListener {
  /**
   * Called on a specified event occurrence
   *
   * @param event
   * @param operationContext
   */
  override def onEvent(event: Event, operationContext: OperationContext): Unit = {
    val loadEvent = event.asInstanceOf[LoadTablePreExecutionEvent]
    val carbonLoadModel = loadEvent.getCarbonLoadModel
    val table = carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable
    val isInternalLoadCall = carbonLoadModel.isAggLoadRequest
    if ((table.isChildDataMap || table.isChildTable) && !isInternalLoadCall) {
      throw new UnsupportedOperationException(
        "Cannot insert/load data directly into pre-aggregate/child table")
    }
  }
}

object PreAggregateDataTypeChangePreListener 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 (CarbonUtil.hasAggregationDataMap(carbonTable)) {
      val dataMapSchemas = carbonTable.getTableInfo.getDataMapSchemaList
      dataMapSchemas.asScala.foreach { dataMapSchema =>
        val childColumns = dataMapSchema.getChildSchema.getListOfColumns
        val parentColumnNames = childColumns.asScala
          .flatMap(_.getParentColumnTableRelations.asScala.map(_.getColumnName))
        if (parentColumnNames.contains(columnToBeAltered)) {
          throw new UnsupportedOperationException(
            s"Column $columnToBeAltered exists in a pre-aggregate table. Drop pre-aggregate table" +
            "to continue")
        }
      }
    }
    if (carbonTable.isChildDataMap) {
      throw new UnsupportedOperationException(
        s"Cannot change data type or rename column for columns in pre-aggregate table ${
          carbonTable.getDatabaseName
        }.${ carbonTable.getTableName }")
    }
  }
}

object PreAggregateAddColumnsPreListener extends OperationEventListener {
  /**
   * Called on a specified event occurrence
   *
   * @param event
   * @param operationContext
   */
  override def onEvent(event: Event, operationContext: OperationContext): Unit = {
    val dataTypeChangePreListener = event.asInstanceOf[AlterTableAddColumnPreEvent]
    val carbonTable = dataTypeChangePreListener.carbonTable
    if (carbonTable.isChildDataMap) {
      throw new UnsupportedOperationException(
        s"Cannot add columns in pre-aggreagate table ${ carbonTable.getDatabaseName
        }.${ carbonTable.getTableName }")
    }
  }
}

object PreAggregateDeleteSegmentByDatePreListener extends OperationEventListener {
  /**
   * Called on a specified event occurrence
   *
   * @param event
   * @param operationContext
   */
  override def onEvent(event: Event, operationContext: OperationContext): Unit = {
    val deleteSegmentByDatePreEvent = event.asInstanceOf[DeleteSegmentByDatePreEvent]
    val carbonTable = deleteSegmentByDatePreEvent.carbonTable
    if (carbonTable != null) {
      if (CarbonUtil.hasAggregationDataMap(carbonTable)) {
        throw new UnsupportedOperationException(
          "Delete segment operation is not supported on tables which have a pre-aggregate table. " +
          "Drop pre-aggregation table to continue")
      }
      if (carbonTable.isChildDataMap) {
        throw new UnsupportedOperationException(
          "Delete segment operation is not supported on pre-aggregate table")
      }
    }
  }
}

object PreAggregateDeleteSegmentByIdPreListener extends OperationEventListener {
  /**
   * Called on a specified event occurrence
   *
   * @param event
   * @param operationContext
   */
  override def onEvent(event: Event, operationContext: OperationContext): Unit = {
    val tableEvent = event.asInstanceOf[DeleteSegmentByIdPreEvent]
    val carbonTable = tableEvent.carbonTable
    if (carbonTable != null) {
      if (CarbonUtil.hasAggregationDataMap(carbonTable)) {
        throw new UnsupportedOperationException(
          "Delete segment operation is not supported on tables which have a pre-aggregate table")
      }
      if (carbonTable.isChildDataMap) {
        throw new UnsupportedOperationException(
          "Delete segment operation is not supported on pre-aggregate table")
      }
    }
  }

}

object PreAggregateDropColumnPreListener extends OperationEventListener {
  /**
   * Called on a specified event occurrence
   *
   * @param event
   * @param operationContext
   */
  override def onEvent(event: Event, operationContext: OperationContext): Unit = {
    val dataTypeChangePreListener = event.asInstanceOf[AlterTableDropColumnPreEvent]
    val carbonTable = dataTypeChangePreListener.carbonTable
    val alterTableDropColumnModel = dataTypeChangePreListener.alterTableDropColumnModel
    val columnsToBeDropped = alterTableDropColumnModel.columns
    if (CarbonUtil.hasAggregationDataMap(carbonTable)) {
      val dataMapSchemas = carbonTable.getTableInfo.getDataMapSchemaList
      dataMapSchemas.asScala.foreach { dataMapSchema =>
        val parentColumnNames = dataMapSchema.getChildSchema.getListOfColumns.asScala
          .flatMap(_.getParentColumnTableRelations.asScala.map(_.getColumnName))
        val columnExistsInChild = parentColumnNames.collectFirst {
          case parentColumnName if columnsToBeDropped.contains(parentColumnName) =>
            parentColumnName
        }
        if (columnExistsInChild.isDefined) {
          throw new UnsupportedOperationException(
            s"Column ${ columnExistsInChild.head } cannot be dropped because it exists in a " +
            s"pre-aggregate table ${ dataMapSchema.getRelationIdentifier.toString }")
        }
      }
    }
    if (carbonTable.isChildDataMap) {
      throw new UnsupportedOperationException(s"Cannot drop columns in pre-aggreagate table ${
        carbonTable.getDatabaseName}.${ carbonTable.getTableName }")
    }
  }
}

object RenameTablePreListener extends OperationEventListener {
  /**
   * Called on a specified event occurrence
   *
   * @param event
   * @param operationContext
   */
  override def onEvent(event: Event,
      operationContext: OperationContext): Unit = {
    val renameTablePostListener = event.asInstanceOf[AlterTableRenamePreEvent]
    val carbonTable = renameTablePostListener.carbonTable
    if (carbonTable.isChildDataMap) {
      throw new UnsupportedOperationException(
        "Rename operation for datamaps is not supported.")
    }
    if (carbonTable.hasAggregationDataMap) {
      throw new UnsupportedOperationException(
        "Rename operation is not supported for table with pre-aggregate tables")
    }
  }
}

object UpdatePreAggregatePreListener extends OperationEventListener {
  /**
   * Called on a specified event occurrence
   *
   * @param event
   * @param operationContext
   */
  override def onEvent(event: Event, operationContext: OperationContext): Unit = {
    val tableEvent = event.asInstanceOf[UpdateTablePreEvent]
    val carbonTable = tableEvent.carbonTable
    if (carbonTable != null) {
      if (CarbonUtil.hasAggregationDataMap(carbonTable)) {
        throw new UnsupportedOperationException(
          "Update operation is not supported for tables which have a pre-aggregate table. Drop " +
          "pre-aggregate tables to continue.")
      }
      if (carbonTable.isChildDataMap) {
        throw new UnsupportedOperationException(
          "Update operation is not supported for pre-aggregate table")
      }
    }
  }
}

object DeletePreAggregatePreListener extends OperationEventListener {
  /**
   * Called on a specified event occurrence
   *
   * @param event
   * @param operationContext
   */
  override def onEvent(event: Event, operationContext: OperationContext): Unit = {
    val tableEvent = event.asInstanceOf[DeleteFromTablePreEvent]
    val carbonTable = tableEvent.carbonTable
    if (carbonTable != null) {
      if (CarbonUtil.hasAggregationDataMap(carbonTable)) {
        throw new UnsupportedOperationException(
          "Delete operation is not supported for tables which have a pre-aggregate table. Drop " +
          "pre-aggregate tables to continue.")
      }
      if (carbonTable.isChildDataMap) {
        throw new UnsupportedOperationException(
          "Delete operation is not supported for pre-aggregate table")
      }
    }
  }
}