/*
 * 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.{InputStreamReader, IOException}
import java.util
import java.util.Collections
import java.util.concurrent.{Callable, Executors, ExecutorService}

import scala.collection.JavaConverters._

import com.google.gson.Gson
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.mapreduce.InputSplit
import org.apache.spark.sql.{CarbonEnv, Row, SparkSession}
import org.apache.spark.sql.execution.command.{Checker, DataCommand}
import org.apache.spark.sql.util.SparkSQLUtil

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.datastore.filesystem.CarbonFile
import org.apache.carbondata.core.datastore.impl.FileFactory
import org.apache.carbondata.core.locks.{CarbonLockFactory, CarbonLockUtil, ICarbonLock, LockUsage}
import org.apache.carbondata.core.metadata.{ColumnarFormatVersion, SegmentFileStore}
import org.apache.carbondata.core.metadata.schema.table.CarbonTable
import org.apache.carbondata.core.segmentmeta.SegmentMetaDataInfo
import org.apache.carbondata.core.statusmanager.{SegmentStatus, SegmentStatusManager, StageInput}
import org.apache.carbondata.core.util.CarbonProperties
import org.apache.carbondata.core.util.path.CarbonTablePath
import org.apache.carbondata.events.{OperationContext, OperationListenerBus}
import org.apache.carbondata.hadoop.CarbonInputSplit
import org.apache.carbondata.processing.loading.FailureCauses
import org.apache.carbondata.processing.loading.events.LoadEvents.{LoadTablePostExecutionEvent, LoadTablePostStatusUpdateEvent, LoadTablePreStatusUpdateEvent}
import org.apache.carbondata.processing.loading.model.CarbonLoadModel
import org.apache.carbondata.processing.util.CarbonLoaderUtil
import org.apache.carbondata.spark.load.DataLoadProcessBuilderOnSpark
import org.apache.carbondata.spark.rdd.CarbonDataRDDFactory

/**
 * Collect stage input files and trigger a loading into carbon table.
 *
 * @param databaseNameOp database name
 * @param tableName table name
 */
case class CarbonInsertFromStageCommand(
    databaseNameOp: Option[String],
    tableName: String,
    options: Map[String, String]
) extends DataCommand {

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

  override def processData(spark: SparkSession): Seq[Row] = {
    Checker.validateTableExists(databaseNameOp, tableName, spark)
    val table = CarbonEnv.getCarbonTable(databaseNameOp, tableName)(spark)
    val hadoopConf = spark.sessionState.newHadoopConf()
    FileFactory.getConfiguration.addResource(hadoopConf)
    setAuditTable(table)

    if (!table.getTableInfo.isTransactionalTable) {
      throw new MalformedCarbonCommandException("Unsupported operation on non transactional table")
    }
    if (table.isMV) {
      throw new MalformedCarbonCommandException("Unsupported operation on MV table")
    }

    val tablePath = table.getTablePath
    val stagePath = CarbonTablePath.getStageDir(tablePath)
    val snapshotFilePath = CarbonTablePath.getStageSnapshotFile(tablePath)
    var stageFiles: Array[(CarbonFile, CarbonFile)] = Array.empty
    var executorService: ExecutorService = null
    val lock = acquireIngestLock(table)

    try {
      // 1. Check whether we need to recover from previous failure
      // We use a snapshot file to indicate whether there was failure in previous
      // ingest operation. A Snapshot file will be created when an ingest operation
      // starts and will be deleted only after the whole ingest operation is finished,
      // which includes two actions:
      //   1) action1: changing segment status to SUCCESS and
      //   2) action2: deleting all involved stage files.
      //
      // If one of these two actions is failed, the snapshot file will be exist, so
      // that recovery is needed.
      //
      // To do the recovery, do following steps:
      //   1) Check if corresponding segment in table status is SUCCESS,
      //      means deleting stage files had failed. So need to read the stage
      //      file list from the snapshot file and delete them again.
      //   2) Check if corresponding segment in table status is INSERT_IN_PROGRESS,
      //      means data loading had failed. So need to read the stage file list
      //      from the snapshot file and load again.
      recoverIfRequired(snapshotFilePath, table, hadoopConf)

      // 2. Start ingesting, steps:
      //   1) read all existing stage files
      //   2) read all stage files to collect input files for data loading
      //   3) add a new segment entry in table status as INSERT_IN_PROGRESS,
      //   4) write all existing stage file names into a new snapshot file
      //   5) do actual loading
      //   6) write segment file and update segment state to SUCCESS in table status
      //   7) delete stage files used for loading
      //   8) delete the snapshot file

      // 1) read all existing stage files
      val batchSize = try {
        Integer.valueOf(
          options.getOrElse(CarbonInsertFromStageCommand.BATCH_FILE_COUNT_KEY,
            CarbonInsertFromStageCommand.BATCH_FILE_COUNT_DEFAULT))
      } catch {
        case _: NumberFormatException =>
          throw new MalformedCarbonCommandException("Option [" +
            CarbonInsertFromStageCommand.BATCH_FILE_COUNT_KEY + "] is not a number.")
      }
      if (batchSize < 1) {
        throw new MalformedCarbonCommandException("Option [" +
            CarbonInsertFromStageCommand.BATCH_FILE_COUNT_KEY + "] is less than 1.")
      }
      val orderType = options.getOrElse(CarbonInsertFromStageCommand.BATCH_FILE_ORDER_KEY,
        CarbonInsertFromStageCommand.BATCH_FILE_ORDER_DEFAULT)
      if (!orderType.equalsIgnoreCase(CarbonInsertFromStageCommand.BATCH_FILE_ORDER_ASC) &&
        !orderType.equalsIgnoreCase(CarbonInsertFromStageCommand.BATCH_FILE_ORDER_DESC)) {
        throw new MalformedCarbonCommandException("Option [" +
            CarbonInsertFromStageCommand.BATCH_FILE_ORDER_KEY + "] is invalid, should be " +
            CarbonInsertFromStageCommand.BATCH_FILE_ORDER_ASC + " or " +
            CarbonInsertFromStageCommand.BATCH_FILE_ORDER_DESC + ".")
      }
      LOGGER.info("Option [" + CarbonInsertFromStageCommand.BATCH_FILE_ORDER_KEY +
                  "] value is " + orderType)
      stageFiles = listStageFiles(stagePath, hadoopConf, batchSize,
        orderType.equalsIgnoreCase(CarbonInsertFromStageCommand.BATCH_FILE_ORDER_ASC))
      if (stageFiles.isEmpty) {
        // no stage files, so do nothing
        LOGGER.warn("files not found under stage metadata folder")
        return Seq.empty
      }

      // We add a tag 'loading' to the stages in process.
      // different insertstage processes can load different data separately
      // by choose the stages without 'loading' tag or stages loaded timeout.
      // which avoid loading the same data between concurrent insertstage processes.
      // The 'loading' tag is actually an empty file with
      // '.loading' suffix filename
      val numThreads = Math.min(Math.max(stageFiles.length, 1), 10)
      executorService = Executors.newFixedThreadPool(numThreads)
      createStageLoadingFilesWithRetry(executorService, stageFiles)
    } catch {
      case ex: Throwable =>
        LOGGER.error(s"failed to insert ${table.getDatabaseName}.${table.getTableName}", ex)
        throw ex
    } finally {
      lock.unlock()
    }

    try{
      // 2) read all stage files to collect input files for data loading
      // create a thread pool to read them
      val stageInputs = collectStageInputs(executorService, stageFiles)

      // 3) perform data loading
      if (table.isHivePartitionTable) {
        startLoadingWithPartition(spark, table, stageInputs, stageFiles, snapshotFilePath)
      } else {
        startLoading(spark, table, stageInputs, stageFiles, snapshotFilePath)
      }

      // 4) delete stage files
      deleteStageFilesWithRetry(executorService, stageFiles)

      // 5) delete the snapshot file
      deleteSnapShotFileWithRetry(table, snapshotFilePath)
    } catch {
      case ex: Throwable =>
        LOGGER.error(s"failed to insert ${table.getDatabaseName}.${table.getTableName}", ex)
        throw ex
    }
    Seq.empty
  }

  /**
   * Check whether there was failure in previous ingest process and try to recover
   */
  private def recoverIfRequired(
      snapshotFilePath: String,
      table: CarbonTable,
      conf: Configuration): Unit = {
    if (!FileFactory.isFileExist(snapshotFilePath) || table.isHivePartitionTable) {
      // everything is fine
      return
    }

    // something wrong, read the snapshot file and do recover steps
    // 1. check segment state in table status file
    // 2. If in SUCCESS state, delete all stage files read inn snapshot file
    // 3. If in IN_PROGRESS state, delete the entry in table status and load again
    LOGGER.info(s"snapshot file found ($snapshotFilePath), start recovery process")
    val lines = FileFactory.readLinesInFile(snapshotFilePath, conf)
    if (lines.size() < 2) {
      throw new RuntimeException("Invalid snapshot file, " + lines.size() + " lines")
    }

    val segmentId = lines.get(0)
    val stageFileNames = lines.remove(0)
    LOGGER.info(s"Segment $segmentId need recovery, ${stageFileNames.length} stage files")

    // lock the table status
    var lock = CarbonLockFactory.getCarbonLockObj(
      table.getAbsoluteTableIdentifier, LockUsage.TABLE_STATUS_LOCK)
    if (!lock.lockWithRetries()) {
      throw new RuntimeException(s"Failed to lock table status for " +
        s"${table.getDatabaseName}.${table.getTableName}")
    }
    try {
      val segments = SegmentStatusManager.readTableStatusFile(
        CarbonTablePath.getTableStatusFilePath(table.getTablePath)
      )
      val matchedSegment = segments.filter(_.getLoadName.equals(segmentId))
      if (matchedSegment.length != 1) {
        throw new RuntimeException("unexpected " + matchedSegment.length + " segment found")
      }
      matchedSegment(0).getSegmentStatus match {
        case SegmentStatus.SUCCESS =>
          // delete all stage files
          lock.unlock()
          lock = null
          LOGGER.info(s"Segment $segmentId is in SUCCESS state, about to delete " +
            s"${stageFileNames.length} stage files")
          val numThreads = Math.min(Math.max(stageFileNames.length, 1), 10)
          val executorService = Executors.newFixedThreadPool(numThreads)
          stageFileNames.map { fileName =>
            executorService.submit(new Runnable {
              override def run(): Unit = {
                FileFactory.getCarbonFile(
                  CarbonTablePath.getStageDir(table.getTablePath) +
                    CarbonCommonConstants.FILE_SEPARATOR + fileName
                ).delete()
              }
            })
          }.map { future =>
            future.get()
          }
        case other =>
          // delete entry in table status and load again
          LOGGER.warn(s"Segment $segmentId is in $other state, about to delete the " +
            s"segment entry and load again")
          val segmentToWrite = segments.filterNot(_.getLoadName.equals(segmentId))
          SegmentStatusManager.writeLoadDetailsIntoFile(
            CarbonTablePath.getTableStatusFilePath(table.getTablePath),
            segmentToWrite)
      }
    } finally {
      if (lock != null) {
        lock.unlock()
      }
    }
    LOGGER.info(s"Finish recovery, delete snapshot file: $snapshotFilePath")
    FileFactory.getCarbonFile(snapshotFilePath).delete()
  }

  /**
   * Start global sort loading
   */
  private def startLoading(
      spark: SparkSession,
      table: CarbonTable,
      stageInput: Seq[StageInput],
      stageFiles: Array[(CarbonFile, CarbonFile)],
      snapshotFilePath: String
  ): Unit = {
    var loadModel: CarbonLoadModel = null
    try {
      // 1) add new segment with INSERT_IN_PROGRESS into table status
      loadModel = DataLoadProcessBuilderOnSpark.createLoadModelForGlobalSort(spark, table)
      CarbonLoaderUtil.recordNewLoadMetadata(loadModel)

      // 2) write all existing stage file names and segmentId into a new snapshot file
      // The content of snapshot file is: first line is segmentId, followed by each line is
      // one stage file name
      val content =
      (Seq(loadModel.getSegmentId) ++ stageFiles.map(_._1.getAbsolutePath)).mkString("\n")
      FileFactory.writeFile(content, snapshotFilePath)

      // 3) do loading.
      val splits = stageInput.flatMap(_.createSplits().asScala)
      LOGGER.info(s"start to load ${splits.size} files into " +
                  s"${table.getDatabaseName}.${table.getTableName}")
      val start = System.currentTimeMillis()
      val dataFrame = DataLoadProcessBuilderOnSpark.createInputDataFrame(spark, table, splits)
      // accumulator to collect segment metadata info such as columnId and it's minMax values
      val segmentMetaDataAccumulator = spark.sqlContext
        .sparkContext
        .collectionAccumulator[Map[String, SegmentMetaDataInfo]]
      if (table.getBucketingInfo == null) {
        DataLoadProcessBuilderOnSpark.loadDataUsingGlobalSort(
          spark,
          Option(dataFrame),
          loadModel,
          SparkSQLUtil.sessionState(spark).newHadoopConf(),
          segmentMetaDataAccumulator
        ).map { row =>
          (row._1, FailureCauses.NONE == row._2._2.failureCauses)
        }
      } else {
        CarbonDataRDDFactory.loadDataFrame(
          spark.sqlContext,
          Option(dataFrame),
          None,
          loadModel,
          segmentMetaDataAccumulator)
      }
      LOGGER.info(s"finish data loading, time taken ${System.currentTimeMillis() - start}ms")

      // 4) write segment file and update the segment entry to SUCCESS
      val segmentFileName = SegmentFileStore.writeSegmentFile(
        table, loadModel.getSegmentId, loadModel.getFactTimeStamp.toString)
      // create operationContext to fire load events
      val operationContext: OperationContext = new OperationContext
      val (tableIndexes, indexOperationContext) = CommonLoadUtils.firePreLoadEvents(
        sparkSession = spark,
        carbonLoadModel = loadModel,
        uuid = "",
        factPath = "",
        optionsFinal = options.asJava,
        options = options.asJava,
        isOverwriteTable = false,
        isDataFrame = true,
        updateModel = None,
        operationContext = operationContext)
      // in case of insert stage files, added the below property to avoid merge index and
      // fire event to load data to secondary index
      operationContext.setProperty(CarbonCommonConstants.IS_INSERT_STAGE, "true")
      val loadTablePreStatusUpdateEvent: LoadTablePreStatusUpdateEvent =
        new LoadTablePreStatusUpdateEvent(
          table.getCarbonTableIdentifier,
          loadModel)
      OperationListenerBus.getInstance().fireEvent(loadTablePreStatusUpdateEvent, operationContext)

      val status = SegmentFileStore.updateTableStatusFile(
        table, loadModel.getSegmentId, segmentFileName,
        table.getCarbonTableIdentifier.getTableId,
        new SegmentFileStore(table.getTablePath, segmentFileName),
        SegmentStatus.SUCCESS)

      // trigger load post events
      if (status) {
        val loadTablePostStatusUpdateEvent: LoadTablePostStatusUpdateEvent =
          new LoadTablePostStatusUpdateEvent(loadModel)
        try {
          OperationListenerBus.getInstance()
            .fireEvent(loadTablePostStatusUpdateEvent, operationContext)
        } catch {
          case ex: Exception =>
            LOGGER.error("Problem while committing indexes", ex)
        }
      }
      // fire event to load data to materialized views and merge bloom index files
      CommonLoadUtils.firePostLoadEvents(spark,
        loadModel,
        tableIndexes,
        indexOperationContext,
        table,
        operationContext)
    } catch {
      case ex: Throwable =>
        LOGGER.error(s"failed to insert ${table.getDatabaseName}.${table.getTableName}", ex)
        if (loadModel != null) {
          CarbonLoaderUtil.updateTableStatusForFailure(loadModel)
        }
        throw ex
    }
  }

  /**
   * Start global sort loading of partition table
   */
  private def startLoadingWithPartition(
      spark: SparkSession,
      table: CarbonTable,
      stageInput: Seq[StageInput],
      stageFiles: Array[(CarbonFile, CarbonFile)],
      snapshotFilePath: String
    ): Unit = {
    val partitionDataList = listPartitionFiles(stageInput)

    val start = System.currentTimeMillis()
    partitionDataList.map {
      case (partition, splits) =>
        LOGGER.info(s"start to load ${splits.size} files into " +
          s"${table.getDatabaseName}.${table.getTableName}. " +
          s"Partition information: ${partition.mkString(",")}")
        val dataFrame =
          DataLoadProcessBuilderOnSpark.createInputDataFrame(spark, table, splits)
        val columns = dataFrame.columns
        val header = columns.mkString(",")
        val selectColumns = columns.filter(!partition.contains(_))
        val selectedDataFrame = dataFrame.select(selectColumns.head, selectColumns.tail: _*)
        CarbonInsertIntoCommand(
          databaseNameOp = Option(table.getDatabaseName),
          tableName = table.getTableName,
          options = scala.collection.immutable.Map("fileheader" -> header,
            "binary_decoder" -> "base64"),
          isOverwriteTable = false,
          logicalPlan = selectedDataFrame.queryExecution.analyzed,
          tableInfo = table.getTableInfo,
          partition = partition
        ).run(spark)
    }
    LOGGER.info(s"finish data loading, time taken ${ System.currentTimeMillis() - start }ms")
  }

  /**
   * @return return a (partitionMap, InputSplits) pair list.
   *         the partitionMap contains all partition column name and value.
   *         the InputSplits is all data file information of current partition.
   */
  private def listPartitionFiles(
      stageInputs : Seq[StageInput]
    ): Seq[(Map[String, Option[String]], Seq[InputSplit])] = {
    val partitionMap = new util.HashMap[Map[String, Option[String]], util.List[InputSplit]]()
    stageInputs.foreach (
      stageInput => {
        val locations = stageInput.getLocations.asScala
        locations.foreach (
          location => {
            val partition = location.getPartitions.asScala.map(t => (t._1, Option(t._2))).toMap
            var splits = partitionMap.get(partition)
            if (splits == null) {
              partitionMap.put(partition, new util.ArrayList[InputSplit]())
              splits = partitionMap.get(partition)
            }
            splits.addAll (
              location.getFiles.asScala
              .filter(_._1.endsWith(CarbonCommonConstants.FACT_FILE_EXT))
              .map(
                file => {
                  CarbonInputSplit.from(
                    "-1", "0",
                    stageInput.getBase + CarbonCommonConstants.FILE_SEPARATOR + file._1, 0,
                    file._2, ColumnarFormatVersion.V3, null
                  )
                }
              ).toList.asJava
            )
          }
        )
      }
    )
    partitionMap.asScala.map(entry => (entry._1, entry._2.asScala)).toSeq
  }

  /**
   * Read stage files and return input files
   */
  private def collectStageInputs(
      executorService: ExecutorService,
      stageFiles: Array[(CarbonFile, CarbonFile)]
  ): Seq[StageInput] = {
    val startTime = System.currentTimeMillis()
    val output = Collections.synchronizedList(new util.ArrayList[StageInput]())
    val gson = new Gson()
    stageFiles.map { stage =>
      executorService.submit(new Runnable {
        override def run(): Unit = {
          val filePath = stage._1.getAbsolutePath
          val stream = FileFactory.getDataInputStream(filePath)
          try {
            val stageInput = gson.fromJson(new InputStreamReader(stream), classOf[StageInput])
            output.add(stageInput)
          } finally {
            stream.close()
          }
        }
      })
    }.map { future =>
      future.get()
    }
    LOGGER.info(s"read stage files taken ${System.currentTimeMillis() - startTime}ms")
    output.asScala
  }

  /**
   * create '.loading' file to tag the stage in process
   * return the loading files failed to create
   */
  private def createStageLoadingFiles(
      executorService: ExecutorService,
      stageFiles: Array[(CarbonFile, CarbonFile)]): Array[(CarbonFile, CarbonFile)] = {
    stageFiles.map { files =>
      executorService.submit(new Callable[(CarbonFile, CarbonFile, Boolean)] {
        override def call(): (CarbonFile, CarbonFile, Boolean) = {
          // Get the loading files path
          val stageLoadingFile =
            FileFactory.getCarbonFile(files._1.getAbsolutePath +
              CarbonTablePath.LOADING_FILE_SUFFIX);
          // Try to create loading files
          // make isFailed to be true if createNewFile return false.
          // the reason can be file exists or exceptions.
          var isFailed = !stageLoadingFile.createNewFile()
          // if file exists, modify the lastmodifiedtime of the file.
          if (isFailed) {
            // make isFailed to be true if setLastModifiedTime return false.
            isFailed = !stageLoadingFile.setLastModifiedTime(System.currentTimeMillis());
          }
          (files._1, files._2, isFailed)
        }
      })
    }.map { future =>
      future.get()
    }.filter { files =>
      // keep the files when isFailed is ture. so we can retry on these files.
      files._3
    }.map { files =>
      (files._1, files._2)
    }
  }

  /**
   * create '.loading' file with retry
   */
  private def createStageLoadingFilesWithRetry(
      executorService: ExecutorService,
      stageFiles: Array[(CarbonFile, CarbonFile)]): Unit = {
    val startTime = System.currentTimeMillis()
    var retry = CarbonInsertFromStageCommand.DELETE_FILES_RETRY_TIMES
    var needToCreateStageLoadingFiles = stageFiles
    while (retry > 0 && needToCreateStageLoadingFiles.nonEmpty) {
      needToCreateStageLoadingFiles =
        createStageLoadingFiles(executorService, needToCreateStageLoadingFiles)
      retry -= 1
    }
    LOGGER.info(s"finished to create stage loading files, time taken: " +
      s"${System.currentTimeMillis() - startTime}ms")
    if (needToCreateStageLoadingFiles.nonEmpty) {
      LOGGER.warn(s"failed to create loading files:" +
        needToCreateStageLoadingFiles.map(_._1.getName).mkString(","))
    }
  }

  /**
   * Delete stage files and success files and loading files
   * Return the files failed to delete
   */
  private def deleteStageFiles(
      executorService: ExecutorService,
      stageFiles: Array[(CarbonFile, CarbonFile)]): Array[(CarbonFile, CarbonFile)] = {
    stageFiles.map { files =>
      executorService.submit(new Callable[(CarbonFile, CarbonFile, Boolean)] {
        override def call(): (CarbonFile, CarbonFile, Boolean) = {
          // Delete three types of file: stage|.success|.loading
          val stageLoadingFile =
            FileFactory.getCarbonFile(files._1.getAbsolutePath
              + CarbonTablePath.LOADING_FILE_SUFFIX);
          var isFailed = false
          // If delete() return false, maybe the reason is FileNotFount or FileFailedClean.
          // Considering FileNotFound means FileCleanSucessfully.
          // We need double check the file exists or not when delete() return false.
          if (!(files._1.delete() && files._2.delete() && stageLoadingFile.delete())) {
            // If the file still exists,  make isFailed to be true
            // So we can retry to delete this file.
            isFailed = files._1.exists() || files._1.exists() || stageLoadingFile.exists()
          }
          (files._1, files._2, isFailed)
        }
      })
    }.map { future =>
      future.get()
    }.filter { files =>
      // keep the files when isFailed is ture. so we can retry on these files.
      files._3
    }.map { files =>
      (files._1, files._2)
    }
  }

  /**
   * Delete stage file and success file with retry
   */
  private def deleteStageFilesWithRetry(
      executorService: ExecutorService,
      stageFiles: Array[(CarbonFile, CarbonFile)]): Unit = {
    val startTime = System.currentTimeMillis()
    var retry = CarbonInsertFromStageCommand.DELETE_FILES_RETRY_TIMES
    var needToDeleteStageFiles = stageFiles
    while (retry > 0 && needToDeleteStageFiles.nonEmpty) {
      needToDeleteStageFiles =
        deleteStageFiles(executorService, needToDeleteStageFiles)
      retry -= 1
    }
    LOGGER.info(s"finished to delete stage files, time taken: " +
      s"${System.currentTimeMillis() - startTime}ms")
    // if there are still stage files failed to clean, print log.
    if (needToDeleteStageFiles.nonEmpty) {
      LOGGER.warn(s"failed to clean up stage files:" +
        needToDeleteStageFiles.map(_._1.getName).mkString(","))
    }
  }

  /**
   * Delete snapshot file with retry
   * Return false means the snapshot file was cleaned successfully
   * While return true means the snapshot file was failed to clean
   */
  private def deleteSnapShotFile(
      snapshotFilePath: String): Boolean = {
    val snapshotFile = FileFactory.getCarbonFile(snapshotFilePath)
    // If delete() return false, maybe the reason is FileNotFount or FileFailedClean.
    // Considering FileNotFound means FileCleanSucessfully.
    // We need double check the file exists or not when delete() return false.
    if (!snapshotFile.delete()) {
      return snapshotFile.exists()
    }
    true
  }

  /**
   * Delete snapshot file with retry
   */
  private def deleteSnapShotFileWithRetry(
      table: CarbonTable,
      snapshotFilePath: String): Unit = {
    if (table.isHivePartitionTable) {
      return
    }
    var retries = CarbonInsertFromStageCommand.DELETE_FILES_RETRY_TIMES
    while(deleteSnapShotFile(snapshotFilePath) && retries > 0) {
      retries -= 1
    }
  }

  /*
   * Collect all stage files and matched success files.
   * A stage file without success file will not be collected
   */
  private def listStageFiles(
      loadDetailsDir: String,
      hadoopConf: Configuration,
      batchSize: Int,
      ascendingSort: Boolean
  ): Array[(CarbonFile, CarbonFile)] = {
    val dir = FileFactory.getCarbonFile(loadDetailsDir, hadoopConf)
    if (dir.exists()) {
      val allFiles = dir.listFiles()
      val successFiles = allFiles.filter { file =>
        file.getName.endsWith(CarbonTablePath.SUCCESS_FILE_SUFFIX)
      }.map { file =>
        (file.getName.substring(0, file.getName.indexOf(".")), file)
      }.toMap

      // different insertstage processes can load different data separately
      // by choose the stages without 'loading' tag or stages loaded timeout.
      // which avoid loading the same data between concurrent insertstage processes.
      // Overall, There are two conditions to choose stages to process:
      // 1) stages never loaded, choose the stages without '.loading' tag.
      // 2) stages loaded timeout, the timeout threshold depends on INSERT_STAGE_TIMEOUT
      val loadingFiles = allFiles.filter { file =>
        file.getName.endsWith(CarbonTablePath.LOADING_FILE_SUFFIX)
      }.filter { file =>
        (System.currentTimeMillis() - file.getLastModifiedTime) <
          CarbonInsertFromStageCommand.INSERT_STAGE_TIMEOUT
      }.map { file =>
        (file.getName.substring(0, file.getName.indexOf(".")), file)
      }.toMap

      val stageFiles = allFiles.filter { file =>
        !file.getName.endsWith(CarbonTablePath.SUCCESS_FILE_SUFFIX)
      }.filter { file =>
        !file.getName.endsWith(CarbonTablePath.LOADING_FILE_SUFFIX)
      }.filter { file =>
        successFiles.contains(file.getName)
      }.filterNot { file =>
        loadingFiles.contains(file.getName)
      }.sortWith {
        (file1, file2) =>
          if (ascendingSort) {
            file1.getLastModifiedTime < file2.getLastModifiedTime
          } else {
            file1.getLastModifiedTime > file2.getLastModifiedTime
          }
      }.map { file =>
        (file, successFiles(file.getName))
      }
      if (stageFiles.length <= batchSize) {
        stageFiles
      } else {
        stageFiles.dropRight(stageFiles.length - batchSize)
      }
    } else {
      Array.empty
    }
  }

  /**
   * INGEST operation does not support concurrent, so it is one lock for one table
   */
  private def acquireIngestLock(table: CarbonTable): ICarbonLock = {
    val tableIdentifier = table.getAbsoluteTableIdentifier
    val lock = CarbonLockFactory.getCarbonLockObj(tableIdentifier, LockUsage.INGEST_LOCK)
    val retryCount = CarbonLockUtil.getLockProperty(
      CarbonCommonConstants.NUMBER_OF_TRIES_FOR_CARBON_LOCK,
      CarbonCommonConstants.NUMBER_OF_TRIES_FOR_CARBON_LOCK_DEFAULT
    )
    val maxTimeout = CarbonLockUtil.getLockProperty(
      CarbonCommonConstants.MAX_TIMEOUT_FOR_CONCURRENT_LOCK,
      CarbonCommonConstants.MAX_TIMEOUT_FOR_CONCURRENT_LOCK_DEFAULT
    )
    if (lock.lockWithRetries(retryCount, maxTimeout)) {
      lock
    } else {
      throw new IOException(
        s"Not able to acquire the lock for table status file for $tableIdentifier")
    }
  }

  override protected def opName: String = "INSERT STAGE"
}

object CarbonInsertFromStageCommand {

  val DELETE_FILES_RETRY_TIMES = 3

  val BATCH_FILE_COUNT_KEY = "batch_file_count"

  val BATCH_FILE_COUNT_DEFAULT: String = Integer.MAX_VALUE.toString

  val BATCH_FILE_ORDER_KEY = "batch_file_order"

  /**
   * Use this option will insert the earliest stage files into the table.
   */
  val BATCH_FILE_ORDER_ASC = "ASC"

  /**
   * Use this option will insert the latest stage files into the table.
   */
  val BATCH_FILE_ORDER_DESC = "DESC"

  /*
  * Keep default ascending order. (Earliest first)
  */
  val BATCH_FILE_ORDER_DEFAULT: String = BATCH_FILE_ORDER_ASC

  val INSERT_STAGE_TIMEOUT = CarbonProperties.getInsertStageTimeout
}