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

import java.io.IOException
import java.util

import scala.collection.JavaConverters._

import org.apache.commons.lang3.StringUtils
import org.apache.hadoop.conf.Configuration
import org.apache.spark.scheduler.{SparkListener, SparkListenerApplicationEnd}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.execution.command.management.CommonLoadUtils
import org.apache.spark.sql.execution.streaming.{CarbonAppendableStreamSink, Sink}

import org.apache.carbondata.common.Maps
import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.constants.{CarbonCommonConstants, CarbonLoadOptionConstants}
import org.apache.carbondata.core.datastore.compression.CompressorFactory
import org.apache.carbondata.core.datastore.impl.FileFactory
import org.apache.carbondata.core.locks.{CarbonLockFactory, ICarbonLock, LockUsage}
import org.apache.carbondata.core.metadata.schema.table.CarbonTable
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.processing.loading.model.{CarbonLoadModel, CarbonLoadModelBuilder, LoadOption}
import org.apache.carbondata.processing.util.CarbonBadRecordUtil
import org.apache.carbondata.streaming.segment.StreamSegment

/**
 * Stream sink factory
 */
object StreamSinkFactory {

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

  val locks = new util.concurrent.ConcurrentHashMap[String, ICarbonLock]()

  def lock(carbonTable: CarbonTable): Unit = {
    val lock = CarbonLockFactory.getCarbonLockObj(carbonTable.getAbsoluteTableIdentifier,
      LockUsage.STREAMING_LOCK)
    if (lock.lockWithRetries()) {
      locks.put(carbonTable.getTableUniqueName, lock)
      LOGGER.info("Acquired the streaming lock for stream table: " + carbonTable.getDatabaseName +
                  "." + carbonTable.getTableName)
    } else {
      LOGGER.error("Not able to acquire the streaming lock for stream table:" +
        carbonTable.getDatabaseName + "." + carbonTable.getTableName)
      throw new IOException(
        "Not able to acquire the streaming lock for stream table: " +
        carbonTable.getDatabaseName + "." + carbonTable.getTableName)
    }
  }

  def unLock(tableUniqueName: String): Unit = {
    val lock = locks.remove(tableUniqueName)
    if (lock != null) {
      lock.unlock()
    }
  }

  def createStreamTableSink(
      sparkSession: SparkSession,
      hadoopConf: Configuration,
      carbonTable: CarbonTable,
      parameters: Map[String, String]): Sink = {

    lock(carbonTable)

    validateParameters(parameters)

    // build load model
    val carbonLoadModel = buildCarbonLoadModelForStream(
      sparkSession,
      hadoopConf,
      carbonTable,
      parameters,
      "")
    // fire pre event before streamin is started
    // in case of streaming options and optionsFinal can be same
    val operationContext = new OperationContext
    val (tableIndexes, indexOperationContext) = CommonLoadUtils.firePreLoadEvents(sparkSession,
      carbonLoadModel,
      "",
      carbonLoadModel.getFactFilePath,
      parameters.asJava,
      parameters.asJava,
      false,
      false,
      None,
      operationContext)
    // prepare the stream segment
    val segmentId = getStreamSegmentId(carbonTable)
    carbonLoadModel.setSegmentId(segmentId)

    // default is carbon appended stream sink
    val carbonAppendableStreamSink = new CarbonAppendableStreamSink(
      sparkSession,
      carbonTable,
      segmentId,
      parameters,
      carbonLoadModel,
      operationContext)

    // fire post event before streamin is started
    CommonLoadUtils.firePostLoadEvents(sparkSession,
      carbonLoadModel,
      tableIndexes,
      indexOperationContext,
      carbonTable,
      operationContext)
    carbonAppendableStreamSink
  }

  private def validateParameters(parameters: Map[String, String]): Unit = {
    val segmentSize = parameters.get(CarbonCommonConstants.HANDOFF_SIZE)
    if (segmentSize.isDefined) {
      try {
        val value = java.lang.Long.parseLong(segmentSize.get)
        if (value < CarbonCommonConstants.HANDOFF_SIZE_MIN) {
          new CarbonStreamException(CarbonCommonConstants.HANDOFF_SIZE +
                                    "should be bigger than or equal " +
                                    CarbonCommonConstants.HANDOFF_SIZE_MIN)
        }
      } catch {
        case _: NumberFormatException =>
          new CarbonStreamException(CarbonCommonConstants.HANDOFF_SIZE +
                                    s" $segmentSize is an illegal number")
      }
    }
  }

  /**
   * get current stream segment id
   * @return
   */
  private def getStreamSegmentId(carbonTable: CarbonTable): String = {
    val segmentId = StreamSegment.open(carbonTable)
    val segmentDir = CarbonTablePath.getSegmentPath(carbonTable.getTablePath, segmentId)
    val metadataPath = CarbonTablePath.getMetadataPath(carbonTable.getTablePath)
    if (!FileFactory.isFileExist(metadataPath)) {
      // Create table directory path, in case of enabling hive metastore first load may not have
      // table folder created.
      FileFactory.mkdirs(metadataPath)
    }
    if (FileFactory.isFileExist(segmentDir)) {
      // recover fault
      StreamSegment.recoverSegmentIfRequired(segmentDir)
    } else {
      FileFactory.mkdirs(segmentDir)
    }
    segmentId
  }

  private def buildCarbonLoadModelForStream(
      sparkSession: SparkSession,
      hadoopConf: Configuration,
      carbonTable: CarbonTable,
      parameters: Map[String, String],
      segmentId: String): CarbonLoadModel = {
    val carbonProperty: CarbonProperties = CarbonProperties.getInstance()
    val tableProperties = carbonTable.getTableInfo.getFactTable.getTableProperties
    carbonProperty.addProperty("zookeeper.enable.lock", "false")
    val optionsFinal = LoadOption.fillOptionWithDefaultValue(parameters.asJava)
    optionsFinal.put("sort_scope", "no_sort")
    if (parameters.get("fileheader").isEmpty) {
      optionsFinal.put("fileheader", carbonTable.getCreateOrderColumn()
        .asScala.map(_.getColName).mkString(","))
    }
    optionsFinal
      .put("bad_record_path", CarbonBadRecordUtil.getBadRecordsPath(parameters.asJava, carbonTable))
    // If DATEFORMAT is not present in load options, check from table properties.
    if (optionsFinal.get("dateformat").isEmpty) {
      optionsFinal.put("dateformat", Maps.getOrDefault(tableProperties,
        "dateformat", CarbonProperties.getInstance
          .getProperty(CarbonCommonConstants.CARBON_DATE_FORMAT,
            CarbonCommonConstants.CARBON_DATE_DEFAULT_FORMAT)))
    }
    // If TIMESTAMPFORMAT is not present in load options, check from table properties.
    if (optionsFinal.get("timestampformat").isEmpty) {
      optionsFinal.put("timestampformat", Maps.getOrDefault(tableProperties,
        "timestampformat", CarbonProperties.getInstance
          .getProperty(CarbonCommonConstants.CARBON_TIMESTAMP_FORMAT,
            CarbonCommonConstants.CARBON_TIMESTAMP_DEFAULT_FORMAT)))
    }
    val carbonLoadModel = new CarbonLoadModel()
    new CarbonLoadModelBuilder(carbonTable).build(
      parameters.asJava,
      optionsFinal,
      carbonLoadModel,
      hadoopConf)
    carbonLoadModel.setSegmentId(segmentId)
    val columnCompressor = carbonTable.getTableInfo.getFactTable.getTableProperties.asScala
      .getOrElse(CarbonCommonConstants.COMPRESSOR,
        CompressorFactory.getInstance().getCompressor.getName)
    carbonLoadModel.setColumnCompressor(columnCompressor)
    carbonLoadModel
  }
}