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

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

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

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

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

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

object LoadMVTablePreListener extends OperationEventListener {
  /**
   * Called on LoadTablePreExecutionEvent event occurrence
   */
  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.isChildTableForMV && !isInternalLoadCall) {
      throw new UnsupportedOperationException("Cannot insert data directly into MV table")
    }
  }
}

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

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

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


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

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

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

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