/*
 * 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.carbondata.spark.rdd

import java.util
import java.util.UUID

import org.apache.hadoop.mapreduce.{Job, RecordReader, TaskAttemptID, TaskType}
import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl
import org.apache.spark.{Partition, SerializableWritable, TaskContext}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.expressions.GenericInternalRow

import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.converter.SparkDataTypeConverterImpl
import org.apache.carbondata.core.datastore.block.SegmentProperties
import org.apache.carbondata.core.datastore.impl.FileFactory
import org.apache.carbondata.core.index.Segment
import org.apache.carbondata.core.locks.{CarbonLockFactory, LockUsage}
import org.apache.carbondata.core.metadata.schema.table.CarbonTable
import org.apache.carbondata.core.scan.result.iterator.RawResultIterator
import org.apache.carbondata.core.statusmanager.{LoadMetadataDetails, SegmentStatus, SegmentStatusManager}
import org.apache.carbondata.core.util.{CarbonUtil, DataTypeUtil}
import org.apache.carbondata.core.util.path.CarbonTablePath
import org.apache.carbondata.events.{OperationContext, OperationListenerBus}
import org.apache.carbondata.hadoop.{CarbonInputSplit, CarbonProjection}
import org.apache.carbondata.hadoop.api.{CarbonInputFormat, CarbonTableInputFormat}
import org.apache.carbondata.hadoop.stream.CarbonStreamInputFormat
import org.apache.carbondata.hadoop.util.CarbonInputFormatUtil
import org.apache.carbondata.processing.loading.events.LoadEvents.{LoadTablePostStatusUpdateEvent, LoadTablePreStatusUpdateEvent}
import org.apache.carbondata.processing.loading.model.CarbonLoadModel
import org.apache.carbondata.processing.merger.{CompactionResultSortProcessor, CompactionType}
import org.apache.carbondata.processing.util.CarbonLoaderUtil
import org.apache.carbondata.spark.{HandoffResult, HandoffResultImpl}
import org.apache.carbondata.spark.util.{CarbonSparkUtil, CommonUtil}


/**
 * partition of the handoff segment
 */
class HandoffPartition(
    val rddId: Int,
    val idx: Int,
    @transient val inputSplit: CarbonInputSplit
) extends Partition {

  val split = new SerializableWritable[CarbonInputSplit](inputSplit)

  override val index: Int = idx

  override def hashCode(): Int = 41 * (41 + rddId) + idx
}

/**
 * package the record reader of the handoff segment to RawResultIterator
 * todo: actually we should not extends rawResultIterator if we don't use any method or variable
 * from it. We only use it to reduce duplicate code for compaction and handoff
 * and we can extract it later
 */
class StreamingRawResultIterator(
    recordReader: RecordReader[Void, Any]
) extends RawResultIterator(null, null, null, false) {

  override def hasNext: Boolean = {
    recordReader.nextKeyValue()
  }

  override def next(): Array[Object] = {
    val rowTmp = recordReader
      .getCurrentValue
      .asInstanceOf[GenericInternalRow]
      .values
      .asInstanceOf[Array[Object]]
    val row = new Array[Object](rowTmp.length)
    System.arraycopy(rowTmp, 0, row, 0, rowTmp.length)
    row
  }
}

/**
 * execute streaming segment handoff
 */
class StreamHandoffRDD[K, V](
    @transient private val ss: SparkSession,
    result: HandoffResult[K, V],
    carbonLoadModel: CarbonLoadModel,
    handOffSegmentId: String) extends CarbonRDD[(K, V)](ss, Nil) {

  private val jobTrackerId = CarbonInputFormatUtil.createJobTrackerID()

  override def internalCompute(
      split: Partition,
      context: TaskContext): Iterator[(K, V)] = {
    carbonLoadModel.setTaskNo("" + split.index)
    val carbonTable = carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable
    DataTypeUtil.setDataTypeConverter(new SparkDataTypeConverterImpl)
    // the input iterator is using raw row
    val iteratorList = prepareInputIterator(split, carbonTable)

    CommonUtil.setTempStoreLocation(split.index, carbonLoadModel, true, false)
    // use CompactionResultSortProcessor to sort data dan write to columnar files
    val processor = prepareHandoffProcessor(carbonTable)

    // The iterator list here is unsorted. The sorted iterators are null.
    val status = processor.execute(iteratorList, null)

    new Iterator[(K, V)] {
      private var finished = false

      override def hasNext: Boolean = {
        !finished
      }

      override def next(): (K, V) = {
        finished = true
        result.getKey("" + split.index, status)
      }
    }
  }

  /**
   * prepare input iterator by basing CarbonStreamRecordReader
   */
  private def prepareInputIterator(
      split: Partition,
      carbonTable: CarbonTable
  ): util.ArrayList[RawResultIterator] = {
    val inputSplit = split.asInstanceOf[HandoffPartition].split.value
    val attemptId = new TaskAttemptID(jobTrackerId, id, TaskType.MAP, split.index, 0)
    val hadoopConf = getConf
    CarbonInputFormat.setDatabaseName(hadoopConf, carbonTable.getDatabaseName)
    CarbonInputFormat.setTableName(hadoopConf, carbonTable.getTableName)
    CarbonInputFormat.setTablePath(hadoopConf, carbonTable.getTablePath)
    val projection = new CarbonProjection
    val dataFields = carbonTable.getStreamStorageOrderColumn()
    (0 until dataFields.size()).foreach { index =>
      projection.addColumn(dataFields.get(index).getColName)
    }
    CarbonInputFormat.setColumnProjection(hadoopConf, projection)
    CarbonInputFormat.setTableInfo(hadoopConf, carbonTable.getTableInfo)
    val attemptContext = new TaskAttemptContextImpl(hadoopConf, attemptId)
    val format = new CarbonTableInputFormat[Array[Object]]()
    val model = format.createQueryModel(inputSplit, attemptContext)
    val inputFormat = new CarbonStreamInputFormat
    inputFormat.setIsVectorReader(false)
    inputFormat.setModel(model)
    inputFormat.setUseRawRow(true)
    val streamReader = inputFormat.createRecordReader(inputSplit, attemptContext)
      .asInstanceOf[RecordReader[Void, Any]]
    streamReader.initialize(inputSplit, attemptContext)
    val iteratorList = new util.ArrayList[RawResultIterator](1)
    iteratorList.add(new StreamingRawResultIterator(streamReader))
    iteratorList
  }

  private def prepareHandoffProcessor(
      carbonTable: CarbonTable
  ): CompactionResultSortProcessor = {
    val wrapperColumnSchemaList = CarbonUtil.getColumnSchemaList(
      carbonTable.getVisibleDimensions, carbonTable.getVisibleMeasures)
    val segmentProperties = new SegmentProperties(wrapperColumnSchemaList)

    new CompactionResultSortProcessor(
      carbonLoadModel,
      carbonTable,
      segmentProperties,
      CompactionType.STREAMING,
      carbonTable.getTableName,
      null
    )
  }

  /**
   * get the partitions of the handoff segment
   */
  override protected def internalGetPartitions: Array[Partition] = {
    val job = CarbonSparkUtil.createHadoopJob()
    val inputFormat = new CarbonTableInputFormat[Array[Object]]()
    val segmentList = new util.ArrayList[Segment](1)
    segmentList.add(Segment.toSegment(handOffSegmentId, null))
    val splits = inputFormat.getSplitsOfStreaming(
      job,
      segmentList,
      carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable
    )

    (0 until splits.size()).map { index =>
      new HandoffPartition(id, index, splits.get(index).asInstanceOf[CarbonInputSplit])
    }.toArray[Partition]
  }
}

object StreamHandoffRDD {

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

  def iterateStreamingHandoff(
      carbonLoadModel: CarbonLoadModel,
      operationContext: OperationContext,
      sparkSession: SparkSession): Unit = {
    val carbonTable = carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable
    val identifier = carbonTable.getAbsoluteTableIdentifier
    var continueHandoff = false
    // require handoff lock on table
    val lock = CarbonLockFactory.getCarbonLockObj(identifier, LockUsage.HANDOFF_LOCK)
    try {
      if (lock.lockWithRetries()) {
        LOGGER.info("Acquired the handoff lock for table" +
                    s" ${ carbonTable.getDatabaseName }.${ carbonTable.getTableName }")
        // handoff streaming segment one by one
        do {
          val segmentStatusManager = new SegmentStatusManager(identifier)
          var loadMetadataDetails: Array[LoadMetadataDetails] = null
          // lock table to read table status file
          val statusLock = segmentStatusManager.getTableStatusLock
          try {
            if (statusLock.lockWithRetries()) {
              loadMetadataDetails = SegmentStatusManager.readLoadMetadata(
                CarbonTablePath.getMetadataPath(identifier.getTablePath))
            }
          } finally {
            if (null != statusLock) {
              statusLock.unlock()
            }
          }
          if (null != loadMetadataDetails) {
            val streamSegments =
              loadMetadataDetails.filter(_.getSegmentStatus == SegmentStatus.STREAMING_FINISH)

            continueHandoff = streamSegments.length > 0
            if (continueHandoff) {
              // handoff a streaming segment
              val loadMetadataDetail = streamSegments(0)
              carbonLoadModel.setSegmentId(loadMetadataDetail.getLoadName)
              executeStreamingHandoff(
                carbonLoadModel,
                sparkSession,
                operationContext,
                loadMetadataDetail.getLoadName
              )
            }
          } else {
            continueHandoff = false
          }
        } while (continueHandoff)
      }
    } finally {
      if (null != lock) {
        lock.unlock()
      }
    }
  }

  /**
   * start new thread to execute stream segment handoff
   */
  def startStreamingHandoffThread(
      carbonLoadModel: CarbonLoadModel,
      operationContext: OperationContext,
      sparkSession: SparkSession,
      isDDL: Boolean
  ): Unit = {
    if (isDDL) {
      iterateStreamingHandoff(carbonLoadModel, operationContext, sparkSession)
    } else {
      // start a new thread to execute streaming segment handoff
      val handoffThread = new Thread() {
        override def run(): Unit = {
          iterateStreamingHandoff(carbonLoadModel, operationContext, sparkSession)
        }
      }
      handoffThread.start()
    }
  }

  /**
   * invoke StreamHandoffRDD to handoff a streaming segment to a columnar segment
   */
  def executeStreamingHandoff(
      carbonLoadModel: CarbonLoadModel,
      sparkSession: SparkSession,
      operationContext: OperationContext,
      handoffSegmentId: String): Unit = {
    var loadStatus = SegmentStatus.SUCCESS
    var errorMessage: String = "Handoff failure"
    try {
      // generate new columnar segment
      val newMetaEntry = new LoadMetadataDetails
      carbonLoadModel.setFactTimeStamp(System.currentTimeMillis())
      CarbonLoaderUtil.populateNewLoadMetaEntry(
        newMetaEntry,
        SegmentStatus.INSERT_IN_PROGRESS,
        carbonLoadModel.getFactTimeStamp,
        false)
      CarbonLoaderUtil.recordNewLoadMetadata(newMetaEntry, carbonLoadModel, true, false)
      // convert a streaming segment to columnar segment

      val status = new StreamHandoffRDD(
        sparkSession,
        new HandoffResultImpl(),
        carbonLoadModel,
        handoffSegmentId).collect()

      status.foreach { x =>
        if (!x._2) {
          loadStatus = SegmentStatus.LOAD_FAILURE
        }
      }
    } catch {
      case ex: Exception =>
        loadStatus = SegmentStatus.LOAD_FAILURE
        LOGGER.error(s"Handoff failed on streaming segment $handoffSegmentId", ex)
        errorMessage = errorMessage + ": " + ex.getCause.getMessage
        LOGGER.error(errorMessage)
    }

    if (loadStatus == SegmentStatus.LOAD_FAILURE) {
      CarbonLoaderUtil.updateTableStatusForFailure(carbonLoadModel)
      LOGGER.info("********starting clean up**********")
      CarbonLoaderUtil.deleteSegmentForFailure(carbonLoadModel)
      LOGGER.info("********clean up done**********")
      LOGGER.error("Cannot write load metadata file as handoff failed")
      throw new Exception(errorMessage)
    }

    if (loadStatus == SegmentStatus.SUCCESS) {
      operationContext.setProperty("uuid", UUID.randomUUID().toString)
      val loadTablePreStatusUpdateEvent: LoadTablePreStatusUpdateEvent =
        new LoadTablePreStatusUpdateEvent(
          carbonLoadModel.getCarbonDataLoadSchema.getCarbonTable.getCarbonTableIdentifier,
          carbonLoadModel)
      OperationListenerBus.getInstance().fireEvent(loadTablePreStatusUpdateEvent, operationContext)

      val done = updateLoadMetadata(handoffSegmentId, carbonLoadModel)

      val loadTablePostStatusUpdateEvent: LoadTablePostStatusUpdateEvent =
        new LoadTablePostStatusUpdateEvent(carbonLoadModel)
      OperationListenerBus.getInstance()
        .fireEvent(loadTablePostStatusUpdateEvent, operationContext)
      if (!done) {
        LOGGER.error("Handoff failed due to failure in table status update.")
        throw new Exception(errorMessage)
      }
    }

  }

  /**
   * update streaming segment and new columnar segment
   */
  private def updateLoadMetadata(
      handoffSegmentId: String,
      loadModel: CarbonLoadModel
  ): Boolean = {
    var status = false
    val metaDataFilepath = loadModel.getCarbonDataLoadSchema.getCarbonTable.getMetadataPath
    val identifier = loadModel.getCarbonDataLoadSchema.getCarbonTable.getAbsoluteTableIdentifier
    val metadataPath = CarbonTablePath.getMetadataPath(identifier.getTablePath)
    if (!FileFactory.isFileExist(metadataPath)) {
      FileFactory.mkdirs(metadataPath)
    }
    val tableStatusPath = CarbonTablePath.getTableStatusFilePath(identifier.getTablePath)
    val segmentStatusManager = new SegmentStatusManager(identifier)
    val carbonLock = segmentStatusManager.getTableStatusLock
    try {
      if (carbonLock.lockWithRetries()) {
        LOGGER.info(
          "Acquired lock for table" + loadModel.getDatabaseName() + "." + loadModel.getTableName()
          + " for table status update")
        val listOfLoadFolderDetailsArray =
          SegmentStatusManager.readLoadMetadata(metaDataFilepath)

        // update new columnar segment to success status
        val newSegment =
          listOfLoadFolderDetailsArray.find(_.getLoadName.equals(loadModel.getSegmentId))
        if (newSegment.isEmpty) {
          throw new Exception("Failed to update table status for new segment")
        } else {
          newSegment.get.setSegmentStatus(SegmentStatus.SUCCESS)
          newSegment.get.setLoadEndTime(System.currentTimeMillis())
          CarbonLoaderUtil.addDataIndexSizeIntoMetaEntry(newSegment.get, loadModel.getSegmentId,
            loadModel.getCarbonDataLoadSchema.getCarbonTable)
        }

        // update streaming segment to compacted status
        val streamSegment =
          listOfLoadFolderDetailsArray.find(_.getLoadName.equals(handoffSegmentId))
        if (streamSegment.isEmpty) {
          throw new Exception("Failed to update table status for streaming segment")
        } else {
          streamSegment.get.setSegmentStatus(SegmentStatus.COMPACTED)
          streamSegment.get.setMergedLoadName(loadModel.getSegmentId)
        }

        // refresh table status file
        SegmentStatusManager.writeLoadDetailsIntoFile(tableStatusPath, listOfLoadFolderDetailsArray)
        status = true
      } else {
        LOGGER.error("Not able to acquire the lock for Table status update for table " + loadModel
          .getDatabaseName() + "." + loadModel.getTableName())
      }
    } finally {
      if (carbonLock.unlock()) {
        LOGGER.info("Table unlocked successfully after table status update" +
                    loadModel.getDatabaseName() + "." + loadModel.getTableName())
      } else {
        LOGGER.error("Unable to unlock Table lock for table" + loadModel.getDatabaseName() +
                     "." + loadModel.getTableName() + " during table status update")
      }
    }
    status
  }
}
