/*
 * 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.presto.util

import java.io._
import java.nio.charset.Charset
import java.text.SimpleDateFormat
import java.util
import java.util.concurrent.atomic.AtomicInteger
import java.util.{ArrayList, Date, UUID}

import scala.collection.JavaConversions._
import scala.collection.mutable

import com.google.gson.Gson
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.Path
import org.apache.hadoop.io.NullWritable
import org.apache.hadoop.mapred.TaskAttemptID
import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl
import org.apache.hadoop.mapreduce.{RecordReader, TaskType}

import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.datastore.compression.CompressorFactory
import org.apache.carbondata.core.datastore.impl.FileFactory
import org.apache.carbondata.core.fileoperations.{AtomicFileOperationFactory, AtomicFileOperations, FileWriteOperation}
import org.apache.carbondata.core.metadata.converter.{SchemaConverter, ThriftWrapperSchemaConverterImpl}
import org.apache.carbondata.core.metadata.datatype.{DataTypes, StructField}
import org.apache.carbondata.core.metadata.encoder.Encoding
import org.apache.carbondata.core.metadata.schema.table.column.CarbonDimension
import org.apache.carbondata.core.metadata.schema.table.{CarbonTable, CarbonTableBuilder, TableSchemaBuilder}
import org.apache.carbondata.core.metadata.{AbsoluteTableIdentifier, CarbonMetadata, CarbonTableIdentifier}
import org.apache.carbondata.core.statusmanager.{LoadMetadataDetails, SegmentStatus}
import org.apache.carbondata.core.util.path.CarbonTablePath
import org.apache.carbondata.core.util.{CarbonProperties, CarbonUtil}
import org.apache.carbondata.core.writer.ThriftWriter
import org.apache.carbondata.processing.loading.DataLoadExecutor
import org.apache.carbondata.processing.loading.constants.DataLoadProcessorConstants
import org.apache.carbondata.processing.loading.csvinput.{BlockDetails, CSVInputFormat, CSVRecordReaderIterator, StringArrayWritable}
import org.apache.carbondata.processing.loading.exception.CarbonDataLoadingException
import org.apache.carbondata.processing.loading.model.{CarbonDataLoadSchema, CarbonLoadModel}
import org.apache.carbondata.processing.util.TableOptionConstant

object CarbonDataStoreCreator {

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

  /**
   * Create store without any restructure
   */
  def createCarbonStore(storePath: String, dataFilePath: String,
      useLocalDict: Boolean = false): Unit = {
    try {
      logger.info("Creating The Carbon Store")
      val dbName: String = "testdb"
      val tableName: String = "testtable"
      val absoluteTableIdentifier = AbsoluteTableIdentifier.from(
        storePath + "/" + dbName + "/" + tableName,
        new CarbonTableIdentifier(dbName,
          tableName,
          UUID.randomUUID().toString))
      val table: CarbonTable = createTable(absoluteTableIdentifier, useLocalDict)
      val schema: CarbonDataLoadSchema = new CarbonDataLoadSchema(table)
      val loadModel: CarbonLoadModel = new CarbonLoadModel()
      import scala.collection.JavaConverters._
      val columnCompressor = table.getTableInfo.getFactTable.getTableProperties.asScala
        .getOrElse(CarbonCommonConstants.COMPRESSOR,
          CompressorFactory.getInstance().getCompressor().getName())
      loadModel.setColumnCompressor(columnCompressor)
      loadModel.setCarbonDataLoadSchema(schema)
      loadModel.setDatabaseName(
        absoluteTableIdentifier.getCarbonTableIdentifier.getDatabaseName)
      loadModel.setTableName(
        absoluteTableIdentifier.getCarbonTableIdentifier.getTableName)
      loadModel.setTableName(
        absoluteTableIdentifier.getCarbonTableIdentifier.getTableName)
      loadModel.setFactFilePath(dataFilePath)
      loadModel.setCarbonTransactionalTable(table.isTransactionalTable)
      loadModel.setLoadMetadataDetails(new ArrayList[LoadMetadataDetails]())
      loadModel.setTablePath(absoluteTableIdentifier.getTablePath)
      CarbonProperties.getInstance
        .addProperty(CarbonCommonConstants.ENABLE_UNSAFE_COLUMN_PAGE, "true")

      loadModel.setDefaultTimestampFormat(
        CarbonProperties.getInstance.getProperty(
          CarbonCommonConstants.CARBON_TIMESTAMP_FORMAT,
          CarbonCommonConstants.CARBON_TIMESTAMP_MILLIS))
      loadModel.setDefaultDateFormat(
        CarbonProperties.getInstance.getProperty(
          CarbonCommonConstants.CARBON_DATE_FORMAT,
          CarbonCommonConstants.CARBON_DATE_DEFAULT_FORMAT))
      loadModel.setSerializationNullFormat(
        TableOptionConstant.SERIALIZATION_NULL_FORMAT.getName +
        "," +
        "\\N")
      loadModel.setBadRecordsLoggerEnable(
        TableOptionConstant.BAD_RECORDS_LOGGER_ENABLE.getName +
        "," +
        "false")
      loadModel.setBadRecordsAction(
        TableOptionConstant.BAD_RECORDS_ACTION.getName + "," +
        "force")
      loadModel.setIsEmptyDataBadRecord(
        DataLoadProcessorConstants.IS_EMPTY_DATA_BAD_RECORD +
        "," +
        "true")
      loadModel.setMaxColumns("15")
      loadModel.setCsvHeader(
        "ID,date,country,name,phonetype,serialname,salary,bonus,monthlyBonus,dob,shortField,isCurrentEmployee")
      loadModel.setCsvHeaderColumns(loadModel.getCsvHeader.split(","))
      loadModel.setTaskNo("0")
      loadModel.setSegmentId("0")
      loadModel.setFactTimeStamp(System.currentTimeMillis())
      loadModel.setMaxColumns("15")
      executeGraph(loadModel, storePath)
    } catch {
      case e: Exception =>
        throw e
    }
  }

  private def createTable(absoluteTableIdentifier: AbsoluteTableIdentifier,
      useLocalDict: Boolean): CarbonTable = {

    val integer = new AtomicInteger(0)
    val schemaBuilder = new TableSchemaBuilder
    schemaBuilder.addColumn(new StructField("ID", DataTypes.INT), integer, false, false)
    schemaBuilder.addColumn(new StructField("date", DataTypes.DATE), integer, false, false)
    schemaBuilder.addColumn(new StructField("country", DataTypes.STRING), integer, false, false)
    schemaBuilder.addColumn(new StructField("name", DataTypes.STRING), integer, false, false)
    schemaBuilder.addColumn(new StructField("phonetype", DataTypes.STRING), integer, false, false)
    schemaBuilder.addColumn(new StructField("serialname", DataTypes.STRING), integer, false, false)
    schemaBuilder.addColumn(new StructField("salary", DataTypes.DOUBLE), integer, false, false)
    schemaBuilder.addColumn(new StructField("bonus", DataTypes.createDecimalType(10, 4)), integer, false, true)
    schemaBuilder.addColumn(new StructField("monthlyBonus", DataTypes.createDecimalType(18, 4)), integer, false, true)
    schemaBuilder.addColumn(new StructField("dob", DataTypes.TIMESTAMP), integer, false, true)
    schemaBuilder.addColumn(new StructField("shortField", DataTypes.SHORT), integer, false, false)
    schemaBuilder.addColumn(new StructField("isCurrentEmployee", DataTypes.BOOLEAN), integer, false, true)
    schemaBuilder.tableName(absoluteTableIdentifier.getTableName)
    val schema = schemaBuilder.build()

    val builder = new CarbonTableBuilder
    builder.databaseName(absoluteTableIdentifier.getDatabaseName)
      .tableName(absoluteTableIdentifier.getTableName)
      .tablePath(absoluteTableIdentifier.getTablePath)
      .isTransactionalTable(true)
      .tableSchema(schema)
    val carbonTable = builder.build()

    val tableInfo = carbonTable.getTableInfo
    val schemaFilePath: String = CarbonTablePath.getSchemaFilePath(
      absoluteTableIdentifier.getTablePath)
    val schemaMetadataPath: String =
      CarbonTablePath.getFolderContainingFile(schemaFilePath)
    CarbonMetadata.getInstance.loadTableMetadata(tableInfo)
    val schemaConverter: SchemaConverter =
      new ThriftWrapperSchemaConverterImpl()
    val thriftTableInfo: org.apache.carbondata.format.TableInfo =
      schemaConverter.fromWrapperToExternalTableInfo(
        tableInfo,
        tableInfo.getDatabaseName,
        tableInfo.getFactTable.getTableName)
    val schemaEvolutionEntry: org.apache.carbondata.format.SchemaEvolutionEntry =
      new org.apache.carbondata.format.SchemaEvolutionEntry(
        tableInfo.getLastUpdatedTime)
    thriftTableInfo.getFact_table.getSchema_evolution.getSchema_evolution_history
      .add(schemaEvolutionEntry)
    if (!FileFactory.isFileExist(schemaMetadataPath)) {
      FileFactory.mkdirs(schemaMetadataPath)
    }
    val thriftWriter: ThriftWriter = new ThriftWriter(schemaFilePath, false)
    thriftWriter.open()
    thriftWriter.write(thriftTableInfo)
    thriftWriter.close()
    CarbonMetadata.getInstance.getCarbonTable(tableInfo.getTableUniqueName)
  }

  private def addDictionaryValuesToDimensionSet(dims: util.List[CarbonDimension],
      dimensionIndex: mutable.Buffer[Int],
      dimensionSet: Array[util.List[String]],
      data: Array[String],
      index: Int) = {
    if (isDictionaryDefaultMember(dims, dimensionSet, index)) {
      dimensionSet(index).add(CarbonCommonConstants.MEMBER_DEFAULT_VAL)
      dimensionSet(index).add(data(dimensionIndex(index)))
    }
    else {
      if (data.length == 1) {
        dimensionSet(index).add("""\N""")
      } else {
        dimensionSet(index).add(data(dimensionIndex(index)))
      }
    }
  }

  private def isDictionaryDefaultMember(dims: util.List[CarbonDimension],
      dimensionSet: Array[util.List[String]],
      index: Int) = {
    dimensionSet(index).isEmpty && dims(index).hasEncoding(Encoding.DICTIONARY) &&
    !dims(index).hasEncoding(Encoding.DIRECT_DICTIONARY)
  }

  /**
   * Execute graph which will further load data
   *
   * @param loadModel     Carbon load model
   * @param storeLocation store location directory
   * @throws Exception
   */
  private def executeGraph(loadModel: CarbonLoadModel, storeLocation: String): Unit = {
    new File(storeLocation).mkdirs()
    val outPutLoc: String = storeLocation + "/etl"
    val databaseName: String = loadModel.getDatabaseName
    val tableName: String = loadModel.getTableName
    val tempLocationKey: String = databaseName + '_' + tableName + "_1"
    CarbonProperties.getInstance.addProperty(tempLocationKey, storeLocation)
    CarbonProperties.getInstance
      .addProperty("store_output_location", outPutLoc)
    CarbonProperties.getInstance.addProperty("send.signal.load", "false")
    CarbonProperties.getInstance
      .addProperty("carbon.is.columnar.storage", "true")
    CarbonProperties.getInstance
      .addProperty("carbon.dimension.split.value.in.columnar", "1")
    CarbonProperties.getInstance
      .addProperty("carbon.is.fullyfilled.bits", "true")
    CarbonProperties.getInstance.addProperty("is.int.based.indexer", "true")
    CarbonProperties.getInstance
      .addProperty("aggregate.columnar.keyblock", "true")
    CarbonProperties.getInstance.addProperty("is.compressed.keyblock", "false")
    CarbonProperties.getInstance
      .addProperty("carbon.direct.dictionary", "true")
    val graphPath: String = outPutLoc + File.separator + loadModel.getDatabaseName +
                            File.separator +
                            tableName +
                            File.separator +
                            0 +
                            File.separator +
                            1 +
                            File.separator +
                            tableName +
                            ".ktr"
    val path: File = new File(graphPath)
    if (path.exists()) {
      path.delete()
    }
    val blockDetails: BlockDetails = new BlockDetails(
      new Path(loadModel.getFactFilePath),
      0,
      new File(loadModel.getFactFilePath).length,
      Array("localhost"))
    val configuration: Configuration = new Configuration()
    CSVInputFormat.setCommentCharacter(configuration, loadModel.getCommentChar)
    CSVInputFormat.setCSVDelimiter(configuration, loadModel.getCsvDelimiter)
    CSVInputFormat.setEscapeCharacter(configuration, loadModel.getEscapeChar)
    CSVInputFormat.setHeaderExtractionEnabled(configuration, true)
    CSVInputFormat.setQuoteCharacter(configuration, loadModel.getQuoteChar)
    CSVInputFormat.setReadBufferSize(
      configuration,
      CarbonProperties.getInstance.getProperty(
        CarbonCommonConstants.CSV_READ_BUFFER_SIZE,
        CarbonCommonConstants.CSV_READ_BUFFER_SIZE_DEFAULT))
    CSVInputFormat.setNumberOfColumns(
      configuration,
      String.valueOf(loadModel.getCsvHeaderColumns.length))
    CSVInputFormat.setMaxColumns(configuration, "15")
    val hadoopAttemptContext: TaskAttemptContextImpl =
      new TaskAttemptContextImpl(configuration,
        new TaskAttemptID("", 1, TaskType.MAP, 0, 0))
    val format: CSVInputFormat = new CSVInputFormat()
    val recordReader: RecordReader[NullWritable, StringArrayWritable] =
      format.createRecordReader(blockDetails, hadoopAttemptContext)
    val readerIterator: CSVRecordReaderIterator = new CSVRecordReaderIterator(
      recordReader,
      blockDetails,
      hadoopAttemptContext)
    new DataLoadExecutor()
      .execute(loadModel, Array(storeLocation), Array(readerIterator))
    writeLoadMetadata(loadModel.getCarbonDataLoadSchema,
      loadModel.getTableName,
      loadModel.getTableName,
      new ArrayList[LoadMetadataDetails]())
  }

  private def writeLoadMetadata(
      schema: CarbonDataLoadSchema,
      databaseName: String,
      tableName: String,
      listOfLoadFolderDetails: util.List[LoadMetadataDetails]): Unit = {
    try {
      val loadMetadataDetails: LoadMetadataDetails = new LoadMetadataDetails()
      loadMetadataDetails.setLoadEndTime(System.currentTimeMillis())
      loadMetadataDetails.setSegmentStatus(SegmentStatus.SUCCESS)
      loadMetadataDetails.setLoadName(String.valueOf(0))
      loadMetadataDetails.setLoadStartTime(
        loadMetadataDetails.getTimeStamp(readCurrentTime()))
      listOfLoadFolderDetails.add(loadMetadataDetails)
      val dataLoadLocation: String = schema.getCarbonTable.getMetadataPath + File.separator +
                                     CarbonTablePath.TABLE_STATUS_FILE
      val gsonObjectToWrite: Gson = new Gson()
      val writeOperation: AtomicFileOperations = AtomicFileOperationFactory
        .getAtomicFileOperations(dataLoadLocation)
      val dataOutputStream =
        writeOperation.openForWrite(FileWriteOperation.OVERWRITE)
      val brWriter = new BufferedWriter(
        new OutputStreamWriter(
          dataOutputStream,
          Charset.forName(CarbonCommonConstants.DEFAULT_CHARSET)))
      val metadataInstance: String =
        gsonObjectToWrite.toJson(listOfLoadFolderDetails.toArray())
      brWriter.write(metadataInstance)
      if (Option(brWriter).isDefined) {
        brWriter.flush()
      }
      CarbonUtil.closeStreams(brWriter)
      writeOperation.close()
    }
    catch {
      case exception: Exception => logger.error(s"exception occurs $exception")
        throw new CarbonDataLoadingException("Data Loading Failed")
    }
  }

  private def readCurrentTime(): String = {
    val sdf: SimpleDateFormat = new SimpleDateFormat(
      CarbonCommonConstants.CARBON_TIMESTAMP_MILLIS)
    sdf.format(new Date())
  }

}