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

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

import org.apache.spark.sql.{CarbonEnv, Row, SparkSession}
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.catalyst.analysis.NoSuchTableException
import org.apache.spark.sql.execution.command.AtomicRunnableCommand
import org.apache.spark.sql.execution.command.datamap.CarbonDropDataMapCommand
import org.apache.spark.sql.hive.CarbonFileMetastore

import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.cache.dictionary.ManageDictionaryAndBTree
import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.datamap.DataMapStoreManager
import org.apache.carbondata.core.datastore.impl.FileFactory
import org.apache.carbondata.core.exception.ConcurrentOperationException
import org.apache.carbondata.core.locks.{CarbonLockFactory, CarbonLockUtil, ICarbonLock, LockUsage}
import org.apache.carbondata.core.metadata.schema.table.CarbonTable
import org.apache.carbondata.core.statusmanager.SegmentStatusManager
import org.apache.carbondata.core.util.{CarbonProperties, CarbonUtil}
import org.apache.carbondata.events._

case class CarbonDropTableCommand(
    ifExistsSet: Boolean,
    databaseNameOp: Option[String],
    tableName: String,
    dropChildTable: Boolean = false)
  extends AtomicRunnableCommand {

  var carbonTable: CarbonTable = _
  var childDropCommands : Seq[CarbonDropTableCommand] = Seq.empty
  var childDropDataMapCommands : Seq[CarbonDropDataMapCommand] = Seq.empty

  override def processMetadata(sparkSession: SparkSession): Seq[Row] = {
    val LOGGER = LogServiceFactory.getLogService(this.getClass.getCanonicalName)

    val dbName = databaseNameOp.getOrElse(sparkSession.catalog.currentDatabase)
    setAuditTable(dbName, tableName)
    val carbonLocks: scala.collection.mutable.ListBuffer[ICarbonLock] = ListBuffer()
    try {
      carbonTable = CarbonEnv.getCarbonTable(databaseNameOp, tableName)(sparkSession)
      val locksToBeAcquired: List[String] = if (carbonTable.isTransactionalTable) {
        List(LockUsage.METADATA_LOCK, LockUsage.DROP_TABLE_LOCK)
      } else {
        List.empty
      }
      val identifier = carbonTable.getAbsoluteTableIdentifier
      locksToBeAcquired foreach {
        lock => carbonLocks +=
                CarbonLockUtil.getLockObject(identifier, lock)
      }
      // check for directly drop datamap table
      if (carbonTable.isChildTableForMV && !dropChildTable) {
        if (!ifExistsSet) {
          throwMetadataException(dbName, tableName,
            "Child table which is associated with datamap cannot be dropped, " +
            "use DROP DATAMAP command to drop")
        } else {
          LOGGER.info("Skipping Drop table " + tableName +
                      " because Child table which is associated with datamap cannot be dropped")
          return Seq.empty
        }
      }

      if (SegmentStatusManager.isLoadInProgressInTable(carbonTable)) {
        throw new ConcurrentOperationException(carbonTable, "loading", "drop table")
      }
      if (carbonTable.isStreamingSink) {
        // streaming table should acquire streaming.lock
        carbonLocks += CarbonLockUtil.getLockObject(identifier, LockUsage.STREAMING_LOCK)
      }
      val relationIdentifiers = carbonTable.getTableInfo.getParentRelationIdentifiers
      if (relationIdentifiers != null && !relationIdentifiers.isEmpty) {
        var ignoreParentTableCheck = false
        if (!ignoreParentTableCheck && !dropChildTable) {
          if (!ifExistsSet) {
            throwMetadataException(dbName, tableName,
              "Child table which is associated with datamap cannot be dropped, " +
              "use DROP DATAMAP command to drop")
          } else {
            return Seq.empty
          }
        }
      }
      val operationContext = new OperationContext
      val dropTablePreEvent: DropTablePreEvent =
        DropTablePreEvent(
          carbonTable,
          ifExistsSet,
          sparkSession)
      OperationListenerBus.getInstance.fireEvent(dropTablePreEvent, operationContext)

      CarbonEnv.getInstance(sparkSession).carbonMetaStore.dropTable(identifier)(sparkSession)

      if (carbonTable.hasDataMapSchema) {
        // drop all child tables
       val childSchemas = carbonTable.getTableInfo.getDataMapSchemaList

        childDropCommands = childSchemas.asScala
          .filter(_.getRelationIdentifier != null)
          .map { childSchema =>
            val childTable =
              CarbonEnv.getCarbonTable(
                TableIdentifier(childSchema.getRelationIdentifier.getTableName,
                  Some(childSchema.getRelationIdentifier.getDatabaseName)))(sparkSession)
            val dropCommand = CarbonDropTableCommand(
              ifExistsSet = true,
              Some(childSchema.getRelationIdentifier.getDatabaseName),
              childSchema.getRelationIdentifier.getTableName,
              dropChildTable = true
            )
            dropCommand.carbonTable = childTable
            dropCommand
          }
        childDropCommands.foreach(_.processMetadata(sparkSession))
      }
      val indexDatamapSchemas =
        DataMapStoreManager.getInstance().getDataMapSchemasOfTable(carbonTable)
      if (!indexDatamapSchemas.isEmpty) {
        childDropDataMapCommands = indexDatamapSchemas.asScala.map { schema =>
          val command = CarbonDropDataMapCommand(schema.getDataMapName,
            ifExistsSet,
            Some(TableIdentifier(tableName, Some(dbName))),
            forceDrop = true)
          command.dataMapSchema = schema
          command.mainTable = carbonTable
          command
        }
        childDropDataMapCommands.foreach(_.processMetadata(sparkSession))
      }

      // fires the event after dropping main table
      val dropTablePostEvent: DropTablePostEvent =
        DropTablePostEvent(
          carbonTable,
          ifExistsSet,
          sparkSession)
      OperationListenerBus.getInstance.fireEvent(dropTablePostEvent, operationContext)
      // Remove all invalid entries of carbonTable and corresponding updated timestamp
      // values from the cache. This case is valid when there are 2 JDBCServer and one of them
      // drops the table, the other server would not be able to clear its cache.
      try {
        CarbonEnv.getInstance(sparkSession).carbonMetaStore match {
          case metastore: CarbonFileMetastore => metastore.removeStaleTimeStampEntries(sparkSession)
          case _ =>
        }
      } catch {
        case _: Exception =>
          // Do nothing
      }
    } catch {
      case ex: NoSuchTableException =>
        if (!ifExistsSet) {
          throw ex
        } else {
          LOGGER.info("Masking error: " + ex.getLocalizedMessage)
        }
      case ex: ConcurrentOperationException =>
        LOGGER.error(ex.getLocalizedMessage, ex)
        throw ex
      case ex: Exception =>
        val msg = s"Dropping table $dbName.$tableName failed: ${ex.getMessage}"
        LOGGER.error(msg, ex)
        throwMetadataException(dbName, tableName, msg)
    } finally {
      if (carbonLocks.nonEmpty) {
        val unlocked = carbonLocks.forall(_.unlock())
        if (unlocked) {
          LOGGER.info("Table MetaData Unlocked Successfully")
        }
      }
    }
    Seq.empty
  }

  override def processData(sparkSession: SparkSession): Seq[Row] = {
    // clear driver side index and dictionary cache
    if (carbonTable != null && !(carbonTable.isChildTableForMV && !dropChildTable)) {
      ManageDictionaryAndBTree.clearBTreeAndDictionaryLRUCache(carbonTable)
      // delete the table folder
      val tablePath = carbonTable.getTablePath
      // delete table data only if it is not external table
      if (FileFactory.isFileExist(tablePath) &&
          !(carbonTable.isExternalTable || carbonTable.isFileLevelFormat)) {
        val file = FileFactory.getCarbonFile(tablePath)
        CarbonUtil.deleteFoldersAndFilesSilent(file)
      }
      // Delete lock directory if external lock path is specified.
      if (CarbonProperties.getInstance.getProperty(CarbonCommonConstants.LOCK_PATH,
        CarbonCommonConstants.LOCK_PATH_DEFAULT).toLowerCase
        .nonEmpty) {
        val tableLockPath = CarbonLockFactory
          .getLockpath(carbonTable.getCarbonTableIdentifier.getTableId)
        val file = FileFactory.getCarbonFile(tableLockPath)
        CarbonUtil.deleteFoldersAndFilesSilent(file)
      }
      if (carbonTable.hasDataMapSchema && childDropCommands.nonEmpty) {
        // drop all child tables
        childDropCommands.foreach(_.processData(sparkSession))
      }
      childDropDataMapCommands.foreach(_.processData(sparkSession))
    }
    Seq.empty
  }

  override protected def opName: String = "DROP TABLE"
}