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

import java.io.File
import java.util

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

import org.apache.hadoop.fs.FileStatus
import org.apache.spark.sql.{AnalysisException, CarbonEnv, Row, SparkSession}
import org.apache.spark.sql.carbondata.execution.datasources.CarbonSparkDataSourceUtil.convertSparkToCarbonDataType
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.execution.command.{Checker, MetadataCommand}
import org.apache.spark.sql.execution.strategy.MixedFormatHandler
import org.apache.spark.sql.hive.CarbonRelation
import org.apache.spark.sql.types.StructType

import org.apache.carbondata.common.exceptions.sql.MalformedCarbonCommandException
import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.datamap.{DataMapStoreManager, Segment}
import org.apache.carbondata.core.datamap.status.DataMapStatusManager
import org.apache.carbondata.core.datastore.impl.FileFactory
import org.apache.carbondata.core.exception.ConcurrentOperationException
import org.apache.carbondata.core.indexstore.{PartitionSpec => CarbonPartitionSpec}
import org.apache.carbondata.core.metadata.SegmentFileStore
import org.apache.carbondata.core.metadata.encoder.Encoding
import org.apache.carbondata.core.metadata.schema.table.CarbonTable
import org.apache.carbondata.core.mutate.CarbonUpdateUtil
import org.apache.carbondata.core.statusmanager.{FileFormat, LoadMetadataDetails, SegmentStatus, SegmentStatusManager}
import org.apache.carbondata.core.util.path.CarbonTablePath
import org.apache.carbondata.events.{BuildDataMapPostExecutionEvent, BuildDataMapPreExecutionEvent, OperationContext, OperationListenerBus}
import org.apache.carbondata.processing.loading.events.LoadEvents.{LoadTablePostExecutionEvent, LoadTablePostStatusUpdateEvent, LoadTablePreExecutionEvent, LoadTablePreStatusUpdateEvent}
import org.apache.carbondata.processing.loading.model.{CarbonDataLoadSchema, CarbonLoadModel}
import org.apache.carbondata.processing.util.CarbonLoaderUtil
import org.apache.carbondata.sdk.file.{Field, Schema}
import org.apache.carbondata.spark.rdd.CarbonDataRDDFactory.clearDataMapFiles


/**
 * User can add external data folder as a segment to a transactional table.
 * In case of external carbon data folder user no need to specify the format in options. But for
 * other formats like parquet user must specify the format=parquet in options.
 */
case class CarbonAddLoadCommand(
    databaseNameOp: Option[String],
    tableName: String,
    options: Map[String, String])
  extends MetadataCommand {

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

  override def processMetadata(sparkSession: SparkSession): Seq[Row] = {
    Checker.validateTableExists(databaseNameOp, tableName, sparkSession)
    val relation = CarbonEnv
      .getInstance(sparkSession)
      .carbonMetaStore
      .lookupRelation(databaseNameOp, tableName)(sparkSession)
      .asInstanceOf[CarbonRelation]
    val tableSchema = StructType.fromAttributes(relation.output)
    val carbonTable = relation.carbonTable
    setAuditTable(carbonTable)
    if (!carbonTable.getTableInfo.isTransactionalTable) {
      throw new MalformedCarbonCommandException("Unsupported operation on non transactional table")
    }

    if (carbonTable.getTableInfo.getFactTable.getListOfColumns.asScala.exists(
      c => c.hasEncoding(Encoding.DICTIONARY) && !c.hasEncoding(Encoding.DIRECT_DICTIONARY))) {
      throw new MalformedCarbonCommandException(
        "Unsupported operation on global dictionary columns table")
    }
    if (carbonTable.isChildTableForMV) {
      throw new MalformedCarbonCommandException("Unsupported operation on MV table")
    }
    // if insert overwrite in progress, do not allow add segment
    if (SegmentStatusManager.isOverwriteInProgressInTable(carbonTable)) {
      throw new ConcurrentOperationException(carbonTable, "insert overwrite", "delete segment")
    }

    val inputPath = options.getOrElse(
      "path", throw new UnsupportedOperationException("PATH is mandatory"))

    // If a path is already added then we should block the adding of the same path again.
    val allSegments = SegmentStatusManager.readLoadMetadata(carbonTable.getMetadataPath)
    if (allSegments.exists(a => a.getPath != null && a.getPath.equalsIgnoreCase(inputPath))) {
      throw new AnalysisException(s"path already exists in table status file, can not add same " +
                                  s"segment path repeatedly: $inputPath")
    }

    val format = options.getOrElse("format", "carbondata")
    val isCarbonFormat = format.equalsIgnoreCase("carbondata") || format.equalsIgnoreCase("carbon")

    // If in the given location no carbon index files are found then we should throw an exception
    if (isCarbonFormat && SegmentFileStore.getListOfCarbonIndexFiles(inputPath).isEmpty) {
      throw new AnalysisException("CarbonIndex files not present in the location")
    }

    // infer schema and collect FileStatus for all partitions
    val (inputPathSchema, lastLevelDirFileMap) =
      MixedFormatHandler.collectInfo(sparkSession, options, inputPath)
    val inputPathCarbonFields = inputPathSchema.fields.map { field =>
      val dataType = convertSparkToCarbonDataType(field.dataType)
      new Field(field.name, dataType)
    }
    val carbonTableSchema = new Schema(tableSchema.fields.map { field =>
      val dataType = convertSparkToCarbonDataType(field.dataType)
      new Field(field.name, dataType)
    })

    // update schema if has partition
    val inputPathTableFields = if (carbonTable.isHivePartitionTable) {
      val partitions = options.getOrElse("partition",
        throw new AnalysisException(
          "partition option is required when adding segment to partition table")
      )
      // extract partition given by user, partition option should be form of "a:int, b:string"
      val partitionFields = partitions
        .split(",")
        .map { input =>
          if (input.nonEmpty) {
            val nameAndDataType = input.trim.toLowerCase.split(":")
            if (nameAndDataType.size == 2) {
              new Field(nameAndDataType(0), nameAndDataType(1))
            } else {
              throw new AnalysisException(s"invalid partition option: ${ options.toString() }")
            }
          }
        }
      // validate against the partition in carbon table
      val carbonTablePartition = getCarbonTablePartition(sparkSession)
      if (!partitionFields.sameElements(carbonTablePartition)) {
        throw new AnalysisException(
          s"""
             |Partition is not same. Carbon table partition is :
             |${carbonTablePartition.mkString(",")} and input segment partition is :
             |${partitionFields.mkString(",")}
             |""".stripMargin)
      }
      inputPathCarbonFields ++ partitionFields
    } else {
      if (options.contains("partition")) {
        throw new AnalysisException(
          s"Invalid option: partition, $tableName is not a partitioned table")
      }
      inputPathCarbonFields
    }

    // validate the schema including partition columns
    val schemaMatched = carbonTableSchema.getFields.forall { field =>
      inputPathTableFields.exists(_.equals(field))
    }
    if (!schemaMatched) {
      throw new AnalysisException(s"Schema is not same. Table schema is : " +
                                  s"${tableSchema} and segment schema is : ${inputPathSchema}")
    }

    // all validation is done, update the metadata accordingly
    if (carbonTable.isHivePartitionTable) {
      // for each partition in input path, create a new segment in carbon table
      val partitionSpecs = collectPartitionSpecList(
        sparkSession, carbonTable.getTablePath, inputPath, lastLevelDirFileMap.keys.toSeq)
      // check the collected partition from input segment path should comply to
      // partitions in carbon table
      val carbonTablePartition = getCarbonTablePartition(sparkSession)
      if (partitionSpecs.head.getPartitions.size() != carbonTablePartition.length) {
        throw new AnalysisException(
          s"""
             |input segment path does not comply to partitions in carbon table:
             |${carbonTablePartition.mkString(",")}
             |""".stripMargin)
      }
      partitionSpecs.foreach { partitionSpec =>
        val dataFiles = lastLevelDirFileMap.getOrElse(partitionSpec.getLocation.toString,
          throw new RuntimeException(s"partition folder not found: ${partitionSpec.getLocation}"))
        writeMetaForSegment(sparkSession, carbonTable, inputPath, Some(partitionSpec), dataFiles)
      }
    } else {
      writeMetaForSegment(sparkSession, carbonTable, inputPath)
    }

    Seq.empty
  }

  private def getCarbonTablePartition(sparkSession: SparkSession): Array[Field] = {
    sparkSession.sessionState.catalog
      .getTableMetadata(TableIdentifier(tableName, databaseNameOp))
      .partitionSchema
      .fields
      .map(f => new Field(f.name, convertSparkToCarbonDataType(f.dataType)))
  }

  /**
   * Write metadata for external segment, including table status file and segment file
   *
   * @param sparkSession spark session
   * @param carbonTable carbon table
   * @param segmentPath external segment path specified by user
   * @param partitionSpecOp partition info extracted from the path
   * @param partitionDataFiles all data files in the partition
   */
  private def writeMetaForSegment(
      sparkSession: SparkSession,
      carbonTable: CarbonTable,
      segmentPath: String,
      partitionSpecOp: Option[CarbonPartitionSpec] = None,
      partitionDataFiles: Seq[FileStatus] = Seq.empty
  ): Unit = {
    val model = new CarbonLoadModel
    model.setCarbonTransactionalTable(true)
    model.setCarbonDataLoadSchema(new CarbonDataLoadSchema(carbonTable))
    model.setDatabaseName(carbonTable.getDatabaseName)
    model.setTableName(carbonTable.getTableName)
    val operationContext = new OperationContext
    operationContext.setProperty("isLoadOrCompaction", false)
    val loadTablePreExecutionEvent: LoadTablePreExecutionEvent =
      new LoadTablePreExecutionEvent(
        carbonTable.getCarbonTableIdentifier,
        model)
    operationContext.setProperty("isOverwrite", false)
    OperationListenerBus.getInstance.fireEvent(loadTablePreExecutionEvent, operationContext)
    // Add pre event listener for index datamap
    val tableDataMaps = DataMapStoreManager.getInstance().getAllDataMap(carbonTable)
    val dataMapOperationContext = new OperationContext()
    if (tableDataMaps.size() > 0) {
      val dataMapNames: mutable.Buffer[String] =
        tableDataMaps.asScala.map(dataMap => dataMap.getDataMapSchema.getDataMapName)
      val buildDataMapPreExecutionEvent: BuildDataMapPreExecutionEvent =
        BuildDataMapPreExecutionEvent(
          sparkSession, carbonTable.getAbsoluteTableIdentifier, dataMapNames)
      OperationListenerBus.getInstance().fireEvent(buildDataMapPreExecutionEvent,
        dataMapOperationContext)
    }

    val newLoadMetaEntry = new LoadMetadataDetails
    model.setFactTimeStamp(CarbonUpdateUtil.readCurrentTime)
    CarbonLoaderUtil.populateNewLoadMetaEntry(newLoadMetaEntry,
      SegmentStatus.INSERT_IN_PROGRESS,
      model.getFactTimeStamp,
      false)
    newLoadMetaEntry.setPath(segmentPath)
    val format = options.getOrElse("format", "carbondata")
    val isCarbonFormat = format.equalsIgnoreCase("carbondata") ||
                         format.equalsIgnoreCase("carbon")
    if (!isCarbonFormat) {
      newLoadMetaEntry.setFileFormat(new FileFormat(format))
    }

    CarbonLoaderUtil.recordNewLoadMetadata(newLoadMetaEntry, model, true, false)
    val segment = new Segment(
      model.getSegmentId,
      SegmentFileStore.genSegmentFileName(
        model.getSegmentId,
        System.nanoTime().toString) + CarbonTablePath.SEGMENT_EXT,
      segmentPath,
      new util.HashMap[String, String](options.asJava))
    val writeSegment =
      if (isCarbonFormat) {
        SegmentFileStore.writeSegmentFile(carbonTable, segment)
      } else {
        SegmentFileStore.writeSegmentFileForOthers(
          carbonTable, segment, partitionSpecOp.orNull, partitionDataFiles.asJava)
      }

    // This event will trigger merge index job, only trigger it if it is carbon file
    if (isCarbonFormat) {
      operationContext.setProperty(
        carbonTable.getTableUniqueName + "_Segment",
        model.getSegmentId)
      val loadTablePreStatusUpdateEvent: LoadTablePreStatusUpdateEvent =
        new LoadTablePreStatusUpdateEvent(
          carbonTable.getCarbonTableIdentifier,
          model)
      OperationListenerBus.getInstance().fireEvent(loadTablePreStatusUpdateEvent, operationContext)
    }

    val success = if (writeSegment) {
      SegmentFileStore.updateTableStatusFile(
        carbonTable,
        model.getSegmentId,
        segment.getSegmentFileName,
        carbonTable.getCarbonTableIdentifier.getTableId,
        new SegmentFileStore(carbonTable.getTablePath, segment.getSegmentFileName),
        SegmentStatus.SUCCESS)
    } else {
      false
    }

    val postExecutionEvent = if (success) {
      val loadTablePostStatusUpdateEvent: LoadTablePostStatusUpdateEvent =
        new LoadTablePostStatusUpdateEvent(model)
      val commitComplete = try {
        OperationListenerBus.getInstance()
          .fireEvent(loadTablePostStatusUpdateEvent, operationContext)
        true
      } catch {
        case ex: Exception =>
          LOGGER.error("Problem while committing data maps", ex)
          false
      }
      commitComplete
    } else {
      success
    }

    if (!postExecutionEvent || !success) {
      CarbonLoaderUtil.updateTableStatusForFailure(model, "uniqueTableStatusId")
      LOGGER.info("********starting clean up**********")
      // delete segment is applicable for transactional table
      CarbonLoaderUtil.deleteSegment(model, model.getSegmentId.toInt)
      // delete corresponding segment file from metadata
      val segmentFile = CarbonTablePath.getSegmentFilesLocation(carbonTable.getTablePath) +
                        File.separator + segment.getSegmentFileName
      FileFactory.deleteFile(segmentFile)
      clearDataMapFiles(carbonTable, model.getSegmentId)
      LOGGER.info("********clean up done**********")
      LOGGER.error("Data load failed due to failure in table status updation.")
      throw new Exception("Data load failed due to failure in table status updation.")
    }
    DataMapStatusManager.disableAllLazyDataMaps(carbonTable)
    val loadTablePostExecutionEvent: LoadTablePostExecutionEvent =
      new LoadTablePostExecutionEvent(
        carbonTable.getCarbonTableIdentifier,
        model)
    OperationListenerBus.getInstance.fireEvent(loadTablePostExecutionEvent, operationContext)
    if (tableDataMaps.size() > 0) {
      val buildDataMapPostExecutionEvent = BuildDataMapPostExecutionEvent(sparkSession,
        carbonTable.getAbsoluteTableIdentifier, null, Seq(model.getSegmentId), false)
      OperationListenerBus.getInstance()
        .fireEvent(buildDataMapPostExecutionEvent, dataMapOperationContext)
    }
  }

  // extract partition column and value, for example, given
  // path1 = path/to/partition/a=1/b=earth
  // path2 = path/to/partition/a=2/b=moon
  // will extract a list of CarbonPartitionSpec:
  //   CarbonPartitionSpec {("a=1","b=earth"), "path/to/partition"}
  //   CarbonPartitionSpec {("a=2","b=moon"), "path/to/partition"}
  def collectPartitionSpecList(
      sparkSession: SparkSession,
      tablePath: String,
      inputPath: String,
      partitionPaths: Seq[String]
  ): Seq[CarbonPartitionSpec] = {
    partitionPaths.map { path =>
      try {
        val partitionOnlyPath = path.substring(inputPath.length + 1)
        val partitionColumnAndValue = partitionOnlyPath.split("/").toList.asJava
        new CarbonPartitionSpec(partitionColumnAndValue, path)
      } catch {
        case t: Throwable => throw new RuntimeException(s"invalid partition path: $path")
      }
    }
  }

  override protected def opName: String = "ADD SEGMENT WITH PATH"
}