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

import java.io.{ByteArrayOutputStream, DataOutputStream, File}
import java.math.BigDecimal
import java.text.SimpleDateFormat
import java.util
import java.util.UUID
import java.util.regex.{Matcher, Pattern}

import scala.collection.JavaConverters._
import scala.collection.mutable
import scala.collection.mutable.Map
import scala.math.BigDecimal.RoundingMode

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat
import org.apache.spark.{SparkContext, SparkEnv}
import org.apache.spark.sql.catalyst.{InternalRow, TableIdentifier}
import org.apache.spark.sql.catalyst.expressions.{UnsafeArrayData, UnsafeMapData, UnsafeRow}
import org.apache.spark.sql.execution.command.{ColumnProperty, Field, PartitionerField}
import org.apache.spark.sql.types.{ArrayType, DataType, DateType, DecimalType, MapType, StringType, StructField, StructType, TimestampType}
import org.apache.spark.util.FileUtils

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.impl.FileFactory
import org.apache.carbondata.core.keygenerator.directdictionary.timestamp.DateDirectDictionaryGenerator
import org.apache.carbondata.core.memory.{UnsafeMemoryManager, UnsafeSortMemoryManager}
import org.apache.carbondata.core.metadata.CarbonMetadata
import org.apache.carbondata.core.metadata.datatype.DataTypes
import org.apache.carbondata.core.metadata.schema.table.CarbonTable
import org.apache.carbondata.core.statusmanager.SegmentStatusManager
import org.apache.carbondata.core.util.{CarbonProperties, CarbonUtil, DataTypeUtil, ThreadLocalTaskInfo}
import org.apache.carbondata.core.util.path.CarbonTablePath
import org.apache.carbondata.processing.datatypes.{ArrayDataType, GenericDataType, StructDataType}
import org.apache.carbondata.processing.loading.CarbonDataLoadConfiguration
import org.apache.carbondata.processing.loading.complexobjects.{ArrayObject, StructObject}
import org.apache.carbondata.processing.loading.constants.DataLoadProcessorConstants
import org.apache.carbondata.processing.loading.converter.BadRecordLogHolder
import org.apache.carbondata.processing.loading.converter.impl.FieldEncoderFactory
import org.apache.carbondata.processing.loading.csvinput.CSVInputFormat
import org.apache.carbondata.processing.loading.model.CarbonLoadModel
import org.apache.carbondata.processing.util.CarbonDataProcessorUtil

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

  val FIXED_DECIMAL = """decimal\(\s*(\d+)\s*,\s*(\-?\d+)\s*\)""".r
  val FIXED_DECIMALTYPE = """decimaltype\(\s*(\d+)\s*,\s*(\-?\d+)\s*\)""".r

  val ONE_KB: Long = 1024L
  val ONE_KB_BI: BigDecimal = BigDecimal.valueOf(ONE_KB)
  val ONE_MB: Long = ONE_KB * ONE_KB
  val ONE_MB_BI: BigDecimal = BigDecimal.valueOf(ONE_MB)
  val ONE_GB: Long = ONE_KB * ONE_MB
  val ONE_GB_BI: BigDecimal = BigDecimal.valueOf(ONE_GB)
  val ONE_TB: Long = ONE_KB * ONE_GB
  val ONE_TB_BI: BigDecimal = BigDecimal.valueOf(ONE_TB)
  val ONE_PB: Long = ONE_KB * ONE_TB
  val ONE_PB_BI: BigDecimal = BigDecimal.valueOf(ONE_PB)
  val ONE_EB: Long = ONE_KB * ONE_PB
  val ONE_EB_BI: BigDecimal = BigDecimal.valueOf(ONE_EB)

  def getColumnProperties(column: String,
      tableProperties: Map[String, String]): Option[util.List[ColumnProperty]] = {
    val fieldProps = new util.ArrayList[ColumnProperty]()
    val columnPropertiesStartKey = CarbonCommonConstants.COLUMN_PROPERTIES + "." + column + "."
    tableProperties.foreach {
      case (key, value) =>
        if (key.startsWith(columnPropertiesStartKey)) {
          fieldProps.add(ColumnProperty(key.substring(columnPropertiesStartKey.length(),
            key.length()), value))
        }
    }
    if (fieldProps.isEmpty) {
      None
    } else {
      Some(fieldProps)
    }
  }

  def validateTblProperties(tableProperties: Map[String, String], fields: Seq[Field]): Boolean = {
    var isValid: Boolean = true
    if (fields.nonEmpty) {
      tableProperties.foreach {
        case (key, value) =>
          if (!validateFields(key, fields)) {
            isValid = false
            throw new MalformedCarbonCommandException(s"Invalid table properties $key")
          }
      }
    }
    isValid
  }

  def validateTypeConvertForSpark2(partitionerField: PartitionerField, value: String): Boolean = {
    val result = partitionerField.dataType.get.toLowerCase match {
      case "integertype" =>
        scala.util.Try(value.toInt).isSuccess
      case "stringtype" =>
        scala.util.Try(value).isSuccess
      case "longtype" =>
        scala.util.Try(value.toLong).isSuccess
      case "floattype" =>
        scala.util.Try(value.toFloat).isSuccess
      case "doubletype" =>
        scala.util.Try(value.toDouble).isSuccess
      case "numerictype" =>
        scala.util.Try(value.toDouble).isSuccess
      case "smallinttype" =>
        scala.util.Try(value.toShort).isSuccess
      case "tinyinttype" =>
        scala.util.Try(value.toShort).isSuccess
      case "shorttype" =>
        scala.util.Try(value.toShort).isSuccess
      case FIXED_DECIMALTYPE(_, _) =>
        validateDecimal(partitionerField, value, 12)
      case "timestamptype" =>
        validateDateAndTimestamp(CarbonCommonConstants.CARBON_TIMESTAMP_FORMAT,
          value, CarbonCommonConstants.CARBON_TIMESTAMP_DEFAULT_FORMAT)
      case "datetype" =>
        validateDateAndTimestamp(CarbonCommonConstants.CARBON_DATE_FORMAT,
          value, CarbonCommonConstants.CARBON_DATE_DEFAULT_FORMAT)
      case others =>
       if (others != null && others.startsWith("char")) {
         scala.util.Try(value).isSuccess
        } else if (others != null && others.startsWith("varchar")) {
         scala.util.Try(value).isSuccess
        } else {
          throw new MalformedCarbonCommandException(
            "UnSupported partition type: " + partitionerField.dataType)
        }
    }
    result
  }

  def validateTypeConvert(partitionerField: PartitionerField, value: String): Boolean = {
    val result = partitionerField.dataType.get.toLowerCase() match {
      case "int" =>
        scala.util.Try(value.toInt).isSuccess
      case "string" =>
        scala.util.Try(value).isSuccess
      case "bigint" =>
        scala.util.Try(value.toLong).isSuccess
      case "long" =>
        scala.util.Try(value.toLong).isSuccess
      case "float" =>
        scala.util.Try(value.toFloat).isSuccess
      case "double" =>
        scala.util.Try(value.toDouble).isSuccess
      case "numeric" =>
        scala.util.Try(value.toDouble).isSuccess
      case "smallint" =>
        scala.util.Try(value.toShort).isSuccess
      case "tinyint" =>
        scala.util.Try(value.toShort).isSuccess
      case "boolean" =>
        scala.util.Try(value.toBoolean).isSuccess
      case FIXED_DECIMAL(_, _) =>
        validateDecimal(partitionerField, value, 8)
      case "timestamp" =>
        validateDateAndTimestamp(CarbonCommonConstants.CARBON_TIMESTAMP_FORMAT,
          value, CarbonCommonConstants.CARBON_TIMESTAMP_DEFAULT_FORMAT)
      case "date" =>
        validateDateAndTimestamp(CarbonCommonConstants.CARBON_DATE_FORMAT,
          value, CarbonCommonConstants.CARBON_DATE_DEFAULT_FORMAT)
      case _ =>
        validateTypeConvertForSpark2(partitionerField, value)
    }

    if(!result) {
      throw new MalformedCarbonCommandException(s"Invalid Partition Values for partition " +
        s"column: ${partitionerField.partitionColumn}")
    } else {
      result
    }
  }

  private def validateDecimal(partitionerField: PartitionerField,
      value: String, precisionIndex: Int) = {
    val parField = partitionerField.dataType.get.split(",")
    val precision = parField(0).substring(precisionIndex).toInt
    val scale = parField(1).substring(0, parField(1).length - 1).toInt
    val pattern = "^([-]?[0-9]{0," + (precision - scale) +
                  "})([.][0-9]{1," + scale + "})?$"
    value.matches(pattern)
  }

  private def validateDateAndTimestamp(key: String, value: String, defaultValue: String) = {
    val dateFormat = new SimpleDateFormat(CarbonProperties.getInstance()
      .getProperty(key, defaultValue))
    scala.util.Try(dateFormat.parse(value)).isSuccess
  }

  def validateFields(key: String, fields: Seq[Field]): Boolean = {
    var isValid: Boolean = false
    fields.foreach { field =>
      if (field.children.isDefined && field.children.get != null) {
        field.children.foreach(fields => {
          fields.foreach(complexfield => {
            val column = if ("val" == complexfield.column) {
              field.column
            } else {
              field.column + "." + complexfield.column
            }
            if (validateColumnProperty(key, column)) {
              isValid = true
            }
          }
          )
        }
        )
      } else {
        if (validateColumnProperty(key, field.column)) {
          isValid = true
        }
      }

    }
    isValid
  }

  def validateColumnProperty(key: String, column: String): Boolean = {
    if (!key.startsWith(CarbonCommonConstants.COLUMN_PROPERTIES)) {
      return true
    }
    val columnPropertyKey = CarbonCommonConstants.COLUMN_PROPERTIES + "." + column + "."
    if (key.startsWith(columnPropertyKey)) {
      true
    } else {
      false
    }
  }

  /**
   * validate table level properties for compaction
   *
   * @param tableProperties
   */
  def validateTableLevelCompactionProperties(tableProperties: Map[String, String]): Unit = {
    validateMajorCompactionSize(tableProperties)
    validateMinorCompactionSize(tableProperties)
    validateAutoLoadMerge(tableProperties)
    validateCompactionLevelThreshold(tableProperties)
    validateCompactionPreserveSegmentsOrAllowedDays(tableProperties,
      CarbonCommonConstants.TABLE_COMPACTION_PRESERVE_SEGMENTS)
    validateCompactionPreserveSegmentsOrAllowedDays(tableProperties,
      CarbonCommonConstants.TABLE_ALLOWED_COMPACTION_DAYS)
  }

  /**
   * This method will validate the major compaction size specified by the user
   * the property is used while doing major compaction
   *
   * @param tableProperties
   */
  def validateMajorCompactionSize(tableProperties: Map[String, String]): Unit = {
    var majorCompactionSize: Integer = 0
    val tblPropName = CarbonCommonConstants.TABLE_MAJOR_COMPACTION_SIZE
    if (tableProperties.get(tblPropName).isDefined) {
      val majorCompactionSizeStr: String =
        parsePropertyValueStringInMB(tableProperties(tblPropName))
      try {
        majorCompactionSize = Integer.parseInt(majorCompactionSizeStr)
      } catch {
        case e: NumberFormatException =>
          throw new MalformedCarbonCommandException(s"Invalid $tblPropName value found: " +
            s"$majorCompactionSizeStr, only int value greater than 0 is supported.")
      }
      if (majorCompactionSize < 0) {
        throw new MalformedCarbonCommandException(s"Invalid $tblPropName value found: " +
          s"$majorCompactionSizeStr, only int value greater than 0 is supported.")
      }
      tableProperties.put(tblPropName, majorCompactionSizeStr)
    }
  }

  def validateMinorCompactionSize(tableProperties: Map[String, String]): Unit = {
    var minorCompactionSize: Integer = 0
    val minorCompactionSizePropName = CarbonCommonConstants.TABLE_MINOR_COMPACTION_SIZE
    if (tableProperties.contains(minorCompactionSizePropName)) {
      val minorCompactionSizeStr: String =
        parsePropertyValueStringInMB(tableProperties(minorCompactionSizePropName))
      try {
        minorCompactionSize = Integer.parseInt(minorCompactionSizeStr)
      } catch {
        case e: NumberFormatException =>
          throw new MalformedCarbonCommandException(s"Invalid value $minorCompactionSizeStr" +
            s" configured for $minorCompactionSizePropName. Please consider configuring value" +
            s" greater than 0")
      }
      if (minorCompactionSize <= 0) {
        throw new MalformedCarbonCommandException(s"Invalid value $minorCompactionSizeStr" +
          s" configured for $minorCompactionSizePropName. Please consider configuring value" +
          s" greater than 0")
      }
      tableProperties.put(minorCompactionSizePropName, minorCompactionSizeStr)
    }
  }

  private def validateBoolean(tableProperties: Map[String, String], property: String): Unit = {
    if (tableProperties.get(property).isDefined) {
      val trimStr = tableProperties(property).trim
      if (!trimStr.equalsIgnoreCase("true") && !trimStr.equalsIgnoreCase("false")) {
        throw new MalformedCarbonCommandException(s"Invalid $property value found: " +
                                                  s"$trimStr, only true|false is supported.")
      }
      tableProperties.put(property, trimStr)
    }
  }

  /**
   * This method will validate the auto merge load property specified by the user
   * the property is used while doing minor compaction
   *
   * @param tableProperties
   */
  def validateAutoLoadMerge(tableProperties: Map[String, String]): Unit = {
    validateBoolean(tableProperties, CarbonCommonConstants.TABLE_AUTO_LOAD_MERGE)
  }

  /**
   * This method will validate the flat folder property specified by the user
   *
   * @param tableProperties
   */
  def validateFlatFolder(tableProperties: Map[String, String]): Unit = {
    validateBoolean(tableProperties, CarbonCommonConstants.FLAT_FOLDER)
  }

  /**
   * This method will validate the compaction level threshold property specified by the user
   * the property is used while doing minor compaction
   *
   * @param tableProperties
   */
  def validateCompactionLevelThreshold(tableProperties: Map[String, String]): Unit = {
    val tblPropName = CarbonCommonConstants.TABLE_COMPACTION_LEVEL_THRESHOLD
    if (tableProperties.get(tblPropName).isDefined) {
      val regularedStr = tableProperties(tblPropName).replace(" ", "")
      try {
        val levels: Array[String] = regularedStr.split(",")
        val thresholds = regularedStr.split(",").map(levelThresholdStr => levelThresholdStr.toInt)
        if (!thresholds.forall(t => t < 100 && t >= 0)) {
          throw new MalformedCarbonCommandException(s"Invalid $tblPropName value found: " +
            s"$regularedStr, only int values separated by comma and between 0 " +
            s"and 100 are supported.")
        }
      }
      catch {
        case e: NumberFormatException =>
          throw new MalformedCarbonCommandException(s"Invalid $tblPropName value found: " +
            s"$regularedStr, only int values separated by comma and between 0 " +
            s"and 100 are supported.")
      }
      tableProperties.put(tblPropName, regularedStr)
    }
  }

  /**
   * This method will validate the compaction preserve segments property
   * or compaction allowed days property
   *
   * @param tableProperties
   * @param tblPropName
   */
  def validateCompactionPreserveSegmentsOrAllowedDays(tableProperties: Map[String, String],
                                                      tblPropName: String): Unit = {
    if (tblPropName.equals(CarbonCommonConstants.TABLE_COMPACTION_PRESERVE_SEGMENTS) ||
      tblPropName.equals(CarbonCommonConstants.TABLE_ALLOWED_COMPACTION_DAYS)) {
      var propValue: Integer = 0
      if (tableProperties.get(tblPropName).isDefined) {
        val propValueStr = tableProperties(tblPropName).trim
        try {
          propValue = Integer.parseInt(propValueStr)
        } catch {
          case e: NumberFormatException =>
            throw new MalformedCarbonCommandException(s"Invalid $tblPropName value found: " +
              s"$propValueStr, only int value between 0 and 100 is supported.")
        }
        if (propValue < 0 || propValue > 100) {
          throw new MalformedCarbonCommandException(s"Invalid $tblPropName value found: " +
            s"$propValueStr, only int value between 0 and 100 is supported.")
        }
        tableProperties.put(tblPropName, propValueStr)
      }
    }
  }

  /**
   * This method will validate the table block size and blocklet size specified by the user
   *
   * @param tableProperties table property specified by user
   * @param propertyName property name
   */
  def validateSize(tableProperties: Map[String, String], propertyName: String): Unit = {
    var size: Integer = 0
    if (tableProperties.get(propertyName).isDefined) {
      val blockSizeStr: String =
        parsePropertyValueStringInMB(tableProperties(propertyName))
      try {
        size = Integer.parseInt(blockSizeStr)
      } catch {
        case e: NumberFormatException =>
          throw new MalformedCarbonCommandException(s"Invalid $propertyName value found: " +
                                                    s"$blockSizeStr, only int value from 1 MB to " +
                                                    s"2048 MB is supported.")
      }
      if (size < CarbonCommonConstants.BLOCK_SIZE_MIN_VAL ||
          size > CarbonCommonConstants.BLOCK_SIZE_MAX_VAL) {
        throw new MalformedCarbonCommandException(s"Invalid $propertyName value found: " +
                                                  s"$blockSizeStr, only int value from 1 MB to " +
                                                  s"2048 MB is supported.")
      }
      tableProperties.put(propertyName, blockSizeStr)
    }
  }

  /**
   * This method will validate the cache expiration time specified by the user
   *
   * @param tableProperties table property specified by user
   * @param propertyName property name
   */
  def validateCacheExpiration(tableProperties: Map[String, String], propertyName: String): Unit = {
    var expirationTime: java.lang.Integer = 0
    if (tableProperties.get(propertyName).isDefined) {
      val value = tableProperties(propertyName)
      val exceptionMsg = s"Invalid $propertyName value found: " +
                         s"$value, only duration from 1 second to INT_MAX_VALUE is supported."
      try {
        expirationTime = java.lang.Integer.parseInt(value)
      } catch {
        case n: NumberFormatException =>
          throw new MalformedCarbonCommandException(exceptionMsg)
      }
      if (expirationTime == 0L) {
        throw new MalformedCarbonCommandException(exceptionMsg)
      }
      tableProperties.put(propertyName, value)
    }
  }

  /**
   * This method will validate the table page size
   *
   * @param tableProperties table property specified by user
   * @param propertyName property name
   */
  def validatePageSizeInmb(tableProperties: Map[String, String], propertyName: String): Unit = {
    var size: Integer = 0
    if (tableProperties.get(propertyName).isDefined) {
      val pageSize: String =
        parsePropertyValueStringInMB(tableProperties(propertyName))
      val minPageSize = CarbonCommonConstants.TABLE_PAGE_SIZE_MIN_INMB
      val maxPageSize = CarbonCommonConstants.TABLE_PAGE_SIZE_MAX_INMB
      try {
        size = Integer.parseInt(pageSize)
      } catch {
        case e: NumberFormatException =>
          throw new MalformedCarbonCommandException(s"Invalid $propertyName value found: " +
                                                    s"$pageSize, only int value from $minPageSize" +
                                                    s" to " +
                                                    s"$maxPageSize is supported.")
      }
      if (size < minPageSize || size > maxPageSize) {
        throw new MalformedCarbonCommandException(s"Invalid $propertyName value found: " +
                                                  s"$pageSize, only int value from $minPageSize " +
                                                  s"to " +
                                                  s"$maxPageSize is supported.")
      }
      tableProperties.put(propertyName, pageSize)
    }
  }

  /**
   * This method will parse the configure string from 'XX MB/M' to 'XX'
   *
   * @param propertyValueString
   */
  def parsePropertyValueStringInMB(propertyValueString: String): String = {
    var parsedPropertyValueString: String = propertyValueString
    if (propertyValueString.trim.toLowerCase.endsWith("mb")) {
      parsedPropertyValueString = propertyValueString.trim.toLowerCase
        .substring(0, propertyValueString.trim.toLowerCase.lastIndexOf("mb")).trim
    }
    if (propertyValueString.trim.toLowerCase.endsWith("m")) {
      parsedPropertyValueString = propertyValueString.trim.toLowerCase
        .substring(0, propertyValueString.trim.toLowerCase.lastIndexOf("m")).trim
    }
    parsedPropertyValueString
  }

  def configureCSVInputFormat(configuration: Configuration,
      carbonLoadModel: CarbonLoadModel): Unit = {
    CSVInputFormat.setCommentCharacter(configuration, carbonLoadModel.getCommentChar)
    CSVInputFormat.setCSVDelimiter(configuration, carbonLoadModel.getCsvDelimiter)
    CSVInputFormat.setSkipEmptyLine(configuration, carbonLoadModel.getSkipEmptyLine)
    CSVInputFormat.setEscapeCharacter(configuration, carbonLoadModel.getEscapeChar)
    CSVInputFormat.setMaxColumns(configuration, carbonLoadModel.getMaxColumns)
    CSVInputFormat.setNumberOfColumns(configuration, carbonLoadModel.getCsvHeaderColumns.length
      .toString)
    CSVInputFormat.setHeaderExtractionEnabled(configuration,
      carbonLoadModel.getCsvHeader == null || carbonLoadModel.getCsvHeader.isEmpty)
    CSVInputFormat.setQuoteCharacter(configuration, carbonLoadModel.getQuoteChar)
    CSVInputFormat.setReadBufferSize(configuration, CarbonProperties.getInstance
      .getProperty(CarbonCommonConstants.CSV_READ_BUFFER_SIZE,
        CarbonCommonConstants.CSV_READ_BUFFER_SIZE_DEFAULT))
    val lineSeparator = carbonLoadModel.getLineSeparator
    if (lineSeparator != null) {
      CSVInputFormat.setLineSeparator(configuration, lineSeparator)
    }
  }

  def configSplitMaxSize(context: SparkContext, filePaths: String,
      hadoopConfiguration: Configuration): Unit = {
    val defaultParallelism = if (context.defaultParallelism < 1) {
      1
    } else {
      context.defaultParallelism
    }
    val spaceConsumed = FileUtils.getSpaceOccupied(filePaths, hadoopConfiguration)
    val blockSize =
      hadoopConfiguration.getLongBytes("dfs.blocksize", CarbonCommonConstants.CARBON_256MB)
    LOGGER.info("[Block Distribution]")
    // calculate new block size to allow use all the parallelism
    if (spaceConsumed < defaultParallelism * blockSize) {
      var newSplitSize: Long = spaceConsumed / defaultParallelism
      if (newSplitSize < CarbonCommonConstants.CARBON_16MB) {
        newSplitSize = CarbonCommonConstants.CARBON_16MB
      }
      hadoopConfiguration.set(FileInputFormat.SPLIT_MAXSIZE, newSplitSize.toString)
      LOGGER.info(s"totalInputSpaceConsumed: $spaceConsumed , " +
                  s"defaultParallelism: $defaultParallelism")
      LOGGER.info(s"mapreduce.input.fileinputformat.split.maxsize: ${ newSplitSize.toString }")
    }
  }

  /**
   * Method to clear the memory for a task
   * if present
   */
  def clearUnsafeMemory(taskId: String) {
    UnsafeMemoryManager.
      INSTANCE.freeMemoryAll(taskId)
    UnsafeSortMemoryManager.
      INSTANCE.freeMemoryAll(taskId)
    ThreadLocalTaskInfo.clearCarbonTaskInfo()
  }

  def getScaleAndPrecision(dataType: String): (Int, Int) = {
    val m: Matcher = Pattern.compile("^decimal\\(([^)]+)\\)").matcher(dataType)
    m.find()
    val matchedString: String = m.group(1)
    val scaleAndPrecision = matchedString.split(",")
    (Integer.parseInt(scaleAndPrecision(0).trim), Integer.parseInt(scaleAndPrecision(1).trim))
  }


  def setTempStoreLocation(
      index: Int,
      carbonLoadModel: CarbonLoadModel,
      isCompactionFlow: Boolean,
      isAltPartitionFlow: Boolean) : Unit = {
    val storeLocation = getTempStoreLocations(index.toString).mkString(File.pathSeparator)

    val tempLocationKey = CarbonDataProcessorUtil.getTempStoreLocationKey(
      carbonLoadModel.getDatabaseName,
      carbonLoadModel.getTableName,
      carbonLoadModel.getSegmentId,
      carbonLoadModel.getTaskNo,
      isCompactionFlow,
      isAltPartitionFlow)
    CarbonProperties.getInstance().addProperty(tempLocationKey, storeLocation)
  }

  /**
   * get the temp locations for each process thread
   *
   * @param index the id for each process thread
   * @return an array of temp locations
   */
  def getTempStoreLocations(index: String) : Array[String] = {
    var storeLocation: Array[String] = Array[String]()
    val isCarbonUseYarnLocalDir = CarbonProperties.getInstance().getProperty(
      CarbonCommonConstants.CARBON_LOADING_USE_YARN_LOCAL_DIR,
      CarbonCommonConstants.CARBON_LOADING_USE_YARN_LOCAL_DIR_DEFAULT).equalsIgnoreCase("true")
    val tmpLocationSuffix = s"${ File.separator }carbon${
      UUID.randomUUID().toString
        .replace("-", "")
    }${ CarbonCommonConstants.UNDERSCORE }$index"
    if (isCarbonUseYarnLocalDir) {
      val yarnStoreLocations = Util.getConfiguredLocalDirs(SparkEnv.get.conf)

      if (null != yarnStoreLocations && yarnStoreLocations.nonEmpty) {
        storeLocation = yarnStoreLocations.map(_ + tmpLocationSuffix)
      } else {
        LOGGER.warn("It seems that the we didn't configure local dirs for yarn," +
                    " so we are unable to use them for data loading." +
                    " Here we will fall back using the java tmp dir.")
        storeLocation = storeLocation :+ (System.getProperty("java.io.tmpdir") + tmpLocationSuffix)
      }
    } else {
      storeLocation = storeLocation :+ (System.getProperty("java.io.tmpdir") + tmpLocationSuffix)
    }
    storeLocation
  }

  /**
   * This method will validate for spatial column
   *
   * @param properties
   */
  def validateForSpatialTypeColumn(properties: Map[String, String]): Unit = {
    // Do not allow to set table properties on spatial column.
    // Spatial column is only allowed in sort columns and spatial index property
    val spatialProperty = properties.get(CarbonCommonConstants.SPATIAL_INDEX)
    if (spatialProperty.isDefined) {
      val spatialColumn = spatialProperty.get.trim
      properties.foreach { case (key, value) =>
        if (!key.startsWith(CarbonCommonConstants.SPATIAL_INDEX) &&
            !key.equalsIgnoreCase(CarbonCommonConstants.SORT_COLUMNS) &&
            value.contains(spatialColumn)) {
          val errorMessage =
            s"$spatialColumn is a spatial index column and is not allowed for " +
            s"the option(s): $key"
          throw new MalformedCarbonCommandException(errorMessage)
        }
      }
    }
  }

  /**
   * This method will validate the cache level
   *
   * @param cacheLevel
   * @param tableProperties
   */
  def validateCacheLevel(cacheLevel: String, tableProperties: Map[String, String]): Unit = {
    val supportedCacheLevel = Seq("BLOCK", "BLOCKLET")
    if (cacheLevel.trim.isEmpty) {
      val errorMessage = "Invalid value: Empty column names for the option(s): " +
                         CarbonCommonConstants.CACHE_LEVEL
      throw new MalformedCarbonCommandException(errorMessage)
    } else {
      val trimmedCacheLevel = cacheLevel.trim.toUpperCase
      if (!supportedCacheLevel.contains(trimmedCacheLevel)) {
        val errorMessage = s"Invalid value: Allowed values for ${
          CarbonCommonConstants.CACHE_LEVEL} are BLOCK AND BLOCKLET"
        throw new MalformedCarbonCommandException(errorMessage)
      }
      tableProperties.put(CarbonCommonConstants.CACHE_LEVEL, trimmedCacheLevel)
    }
  }

  /**
   * This will validate the column meta cache i.e the columns to be cached.
   * By default all dimensions will be cached.
   * If the property is already defined in create table DDL then validate it,
   * else add all the dimension columns as columns to be cached to table properties.
   * valid values for COLUMN_META_CACHE can either be empty or can have one or more comma
   * separated values
   */
  def validateColumnMetaCacheFields(dbName: String,
      tableName: String,
      tableColumns: Seq[String],
      cachedColumns: String,
      tableProperties: Map[String, String]): Unit = {
    val tableIdentifier = TableIdentifier(tableName, Some(dbName))
    // below check is added because empty value for column_meta_cache is allowed and in that
    // case there should not be any validation
    if (!cachedColumns.equals("")) {
      validateColumnMetaCacheOption(tableIdentifier, dbName, cachedColumns, tableColumns)
      val columnsToBeCached = cachedColumns.split(",").map(x => x.trim.toLowerCase).toSeq
      // make the columns in create table order and then add it to table properties
      val createOrder = tableColumns.filter(col => columnsToBeCached.contains(col))
      tableProperties.put(CarbonCommonConstants.COLUMN_META_CACHE, createOrder.mkString(","))
    }
  }

  /**
   * Validate the column_meta_cache option in table properties
   *
   * @param tableIdentifier
   * @param databaseName
   * @param columnsToBeCached
   * @param tableColumns
   */
  private def validateColumnMetaCacheOption(tableIdentifier: TableIdentifier,
      databaseName: String,
      columnsToBeCached: String,
      tableColumns: Seq[String]): Unit = {
    // check if only empty spaces are given in the property value
    if (columnsToBeCached.trim.isEmpty) {
      val errorMessage = "Invalid value: Empty column names for the option(s): " +
                         CarbonCommonConstants.COLUMN_META_CACHE
      throw new MalformedCarbonCommandException(errorMessage)
    } else {
      val columns: Array[String] = columnsToBeCached.split(',').map(x => x.toLowerCase.trim)
      // Check for duplicate column names
      columns.groupBy(col => col.toLowerCase).foreach(f => if (f._2.size > 1) {
        throw new MalformedCarbonCommandException(s"Duplicate column name found : ${ f._1 }")
      })
      columns.foreach(col => {
        // check if any intermediate column is empty
        if (null == col || col.trim.isEmpty) {
          val errorMessage = "Invalid value: Empty column names for the option(s): " +
                             CarbonCommonConstants.COLUMN_META_CACHE
          throw new MalformedCarbonCommandException(errorMessage)
        }
        // check if the column exists in the table
        if (!tableColumns.contains(col.toLowerCase)) {
          val errorMessage = s"Column $col does not exists in the table ${ tableIdentifier.table }"
          throw new MalformedCarbonCommandException(errorMessage)
        }
      })
    }
  }

  /**
   * This method will validate single node minimum load data volume of table specified by the user
   *
   * @param tableProperties table property specified by user
   * @param propertyName property name
   */
  def validateLoadMinSize(tableProperties: Map[String, String], propertyName: String): Unit = {
    var size: Integer = 0
    if (tableProperties.get(propertyName).isDefined) {
      val loadSizeStr: String =
        parsePropertyValueStringInMB(tableProperties(propertyName))
      try {
        size = Integer.parseInt(loadSizeStr)
      } catch {
        case e: NumberFormatException =>
          throw new MalformedCarbonCommandException(s"Invalid $propertyName value found: " +
                                                    s"$loadSizeStr, only int value greater " +
                                                    s"than 0 is supported.")
      }
      // if the value is negative, set the value is 0
      if(size > 0) {
        tableProperties.put(propertyName, loadSizeStr)
      } else {
        tableProperties.put(propertyName, CarbonCommonConstants.CARBON_LOAD_MIN_SIZE_INMB_DEFAULT)
      }
    }
  }

  def isDataTypeSupportedForSortColumn(columnDataType: String): Boolean = {
    val dataTypes = Array("array", "struct", "map", "double", "float", "decimal", "binary")
    dataTypes.exists(x => x.equalsIgnoreCase(columnDataType))
  }

  def validateSortScope(newProperties: Map[String, String]): Unit = {
    val sortScopeOption = newProperties.get(CarbonCommonConstants.SORT_SCOPE)
    if (sortScopeOption.isDefined) {
      if (!CarbonUtil.isValidSortOption(sortScopeOption.get)) {
        throw new MalformedCarbonCommandException(
          s"Invalid SORT_SCOPE ${ sortScopeOption.get }, " +
          s"valid SORT_SCOPE are 'NO_SORT', 'LOCAL_SORT' and 'GLOBAL_SORT'")
      }
    }
  }

  def validateGlobalSortPartitions(propertiesMap: mutable.Map[String, String]): Unit = {
    if (propertiesMap.contains("global_sort_partitions")) {
      val globalSortPartitionsProp = propertiesMap("global_sort_partitions")
      var pass = false
      try {
        val globalSortPartitions = Integer.parseInt(globalSortPartitionsProp)
        if (globalSortPartitions > 0) {
          pass = true
        }
      } catch {
        case _: Exception =>
      }
      if (!pass) {
        throw new MalformedCarbonCommandException(
          s"Table property global_sort_partitions : ${ globalSortPartitionsProp }" +
          s" is invalid")
      }
    }
  }

  def validateSortColumns(
      sortKey: Array[String],
      fields: Seq[(String, String)],
      varcharCols: Seq[String]
  ): Unit = {
    val sortKeyLowerCase = sortKey.map(_.toLowerCase())
    if ((sortKeyLowerCase).diff(sortKeyLowerCase.distinct).length > 0 ||
      (sortKeyLowerCase.length > 1 && sortKeyLowerCase.contains(""))) {
      throw new MalformedCarbonCommandException(
        "SORT_COLUMNS Either contains duplicate columns : " +
          sortKeyLowerCase.diff(sortKeyLowerCase.distinct).mkString(",") +
          " or it contains an illegal argument.")
    }

    sortKey.foreach { column =>
      if (!fields.exists(x => x._1.equalsIgnoreCase(column))) {
        val errorMsg = "sort_columns: " + column +
                       " does not exist in table. Please check the create table statement."
        throw new MalformedCarbonCommandException(errorMsg)
      } else {
        val dataType = fields.find(x =>
          x._1.equalsIgnoreCase(column)).get._2
        if (isDataTypeSupportedForSortColumn(dataType)) {
          val errorMsg = s"sort_columns is unsupported for $dataType datatype column: " + column
          throw new MalformedCarbonCommandException(errorMsg)
        }
        if (varcharCols.exists(x => x.equalsIgnoreCase(column))) {
          throw new MalformedCarbonCommandException(
            s"sort_columns is unsupported for long string datatype column: $column")
        }
      }
    }
  }

  def validateSortColumns(carbonTable: CarbonTable, newProperties: Map[String, String]): Unit = {
    val fields = carbonTable.getVisibleDimensions.asScala ++ carbonTable.getVisibleMeasures.asScala
    val tableProperties = carbonTable.getTableInfo.getFactTable.getTableProperties
    var sortKeyOption = newProperties.get(CarbonCommonConstants.SORT_COLUMNS)
    val varcharColsString = tableProperties.get(CarbonCommonConstants.LONG_STRING_COLUMNS)
    val varcharCols: Seq[String] = if (varcharColsString == null) {
      Seq.empty[String]
    } else {
      varcharColsString.split(",").map(_.trim)
    }

    if (!sortKeyOption.isDefined) {
      // default no columns are selected for sorting in no_sort scope
      sortKeyOption = Some("")
    }
    val sortKeyString = CarbonUtil.unquoteChar(sortKeyOption.get).trim
    if (!sortKeyString.isEmpty) {
      val sortKey = sortKeyString.split(',').map(_.trim)
      validateSortColumns(
        sortKey,
        fields.map { field => (field.getColName, field.getDataType.getName) },
        varcharCols
      )
    }
  }

  def bytesToDisplaySize(size: Long): String = {
    try {
      bytesToDisplaySize(BigDecimal.valueOf(size))
    } catch {
      case _: Exception =>
        size.toString
    }
  }

  // This method converts the bytes count to display size upto 2 decimal places
  def bytesToDisplaySize(size: BigDecimal): String = {
    var displaySize: String = null
    if (size.divideToIntegralValue(ONE_EB_BI).compareTo(BigDecimal.ZERO) > 0) {
      displaySize = size.divide(ONE_EB_BI).setScale(2, RoundingMode.HALF_DOWN).doubleValue() + " EB"
    } else if (size.divideToIntegralValue(ONE_PB_BI).compareTo(BigDecimal.ZERO) > 0) {
      displaySize = size.divide(ONE_PB_BI).setScale(2, RoundingMode.HALF_DOWN).doubleValue() + " PB"
    } else if (size.divideToIntegralValue(ONE_TB_BI).compareTo(BigDecimal.ZERO) > 0) {
      displaySize = size.divide(ONE_TB_BI).setScale(2, RoundingMode.HALF_DOWN).doubleValue() + " TB"
    } else if (size.divideToIntegralValue(ONE_GB_BI).compareTo(BigDecimal.ZERO) > 0) {
      displaySize = size.divide(ONE_GB_BI).setScale(2, RoundingMode.HALF_DOWN).doubleValue() + " GB"
    } else if (size.divideToIntegralValue(ONE_MB_BI).compareTo(BigDecimal.ZERO) > 0) {
      displaySize = size.divide(ONE_MB_BI).setScale(2, RoundingMode.HALF_DOWN).doubleValue() + " MB"
    } else if (size.divideToIntegralValue(ONE_KB_BI).compareTo(BigDecimal.ZERO) > 0) {
      displaySize = size.divide(ONE_KB_BI).setScale(2, RoundingMode.HALF_DOWN).doubleValue() + " KB"
    } else {
      displaySize = size + " B"
    }
    displaySize
  }

  def getObjectArrayFromInternalRowAndConvertComplexType(row: InternalRow,
      fieldTypes: Seq[DataType],
      outputArrayLength: Int): Array[AnyRef] = {
    val data = new Array[AnyRef](outputArrayLength)
    var i = 0
    val fieldTypesLen = fieldTypes.length
    while (i < fieldTypesLen) {
      if (!row.isNullAt(i)) {
        fieldTypes(i) match {
          case StringType =>
            // get normal (non-UTF8) string
            data(i) = row.getString(i)
          case d: DecimalType =>
            data(i) = row.getDecimal(i, d.precision, d.scale).toJavaBigDecimal
          case arrayType : ArrayType =>
            data(i) = convertSparkComplexTypeToCarbonObject(row.getArray(i), arrayType)
          case structType : StructType =>
            data(i) = convertSparkComplexTypeToCarbonObject(row.getStruct(i,
              structType.fields.length), structType)
          case mapType : MapType =>
            data(i) = convertSparkComplexTypeToCarbonObject(row.getMap(i), mapType)
          case other =>
            data(i) = row.get(i, other)
        }
      }
      i += 1
    }
    data
  }

  /**
   * After converting complex objects to carbon objects, need to convert to byte array
   *
   * @param row
   * @param fields
   * @param dataFieldsWithComplexDataType
   * @return
   */
  def getObjectArrayFromInternalRowAndConvertComplexTypeForGlobalSort(
      row: InternalRow,
      fields: Seq[StructField],
      dataFieldsWithComplexDataType: Map[String, GenericDataType[_]]): Array[AnyRef] = {
    val data = new Array[AnyRef](fields.size)
    val badRecordLogHolder = new BadRecordLogHolder();
    var i = 0
    val fieldTypesLen = fields.length
    while (i < fieldTypesLen) {
      fields(i).dataType match {
        case StringType =>
          data(i) = if (row.isNullAt(i)) {
            DataTypeUtil.getBytesDataDataTypeForNoDictionaryColumn(null,
              DataTypes.STRING)
          } else {
            DataTypeUtil.getBytesDataDataTypeForNoDictionaryColumn(row.getString(i),
              DataTypes.STRING)
          }
        case d: DecimalType =>
          val decimalValue = row.getDecimal(i, d.precision, d.scale)
          if (null == decimalValue) {
            data(i) = null
          } else {
            data(i) = decimalValue.toJavaBigDecimal
          }
        case dataType@(_: ArrayType | _: MapType) =>
          val result = convertSparkComplexTypeToCarbonObject(row.get(i, dataType), dataType)
          // convert carbon complex object to byte array
          val byteArray: ByteArrayOutputStream = new ByteArrayOutputStream()
          val dataOutputStream: DataOutputStream = new DataOutputStream(byteArray)
          dataFieldsWithComplexDataType(fields(i).name).asInstanceOf[ArrayDataType]
            .writeByteArray(result.asInstanceOf[ArrayObject],
              dataOutputStream,
              badRecordLogHolder,
              true)
          dataOutputStream.close()
          data(i) = byteArray.toByteArray.asInstanceOf[AnyRef]
        case structType: StructType =>
          val result = convertSparkComplexTypeToCarbonObject(row.get(i, structType), structType)
          // convert carbon complex object to byte array
          val byteArray: ByteArrayOutputStream = new ByteArrayOutputStream()
          val dataOutputStream: DataOutputStream = new DataOutputStream(byteArray)
          dataFieldsWithComplexDataType(fields(i).name).asInstanceOf[StructDataType]
            .writeByteArray(result.asInstanceOf[StructObject],
              dataOutputStream,
              badRecordLogHolder,
              true)
          dataOutputStream.close()
          data(i) = byteArray.toByteArray.asInstanceOf[AnyRef]
        case other =>
          data(i) = row.get(i, other)
      }
      i += 1
    }
    data
  }

  private def convertSparkComplexTypeToCarbonObject(data: AnyRef,
      objectDataType: DataType): AnyRef = {
    if (data == null && (objectDataType.isInstanceOf[ArrayType]
      || objectDataType.isInstanceOf[MapType] || objectDataType.isInstanceOf[StructType])) {
      return null
    }
    objectDataType match {
      case _: ArrayType =>
        val arrayDataType = objectDataType.asInstanceOf[ArrayType]
        val arrayData = data.asInstanceOf[UnsafeArrayData]
        val size = arrayData.numElements()
        val childDataType = arrayDataType.elementType
        val arrayChildObjects = new Array[AnyRef](size)
        var i = 0
        while (i < size) {
          arrayChildObjects(i) = convertSparkComplexTypeToCarbonObject(arrayData.get(i,
            childDataType), childDataType)
          i = i + 1
        }
        new ArrayObject(arrayChildObjects)
      case _: MapType =>
        val mapDataType = objectDataType.asInstanceOf[MapType]
        val keyDataType = mapDataType.keyType
        val valueDataType = mapDataType.valueType
        val mapData = data.asInstanceOf[UnsafeMapData]
        val size = mapData.numElements()
        val keys = mapData.keyArray()
        val values = mapData.valueArray()
        val arrayMapChildObjects = new Array[AnyRef](size)
        var i = 0
        while (i < size) {
          val structChildObjects = new Array[AnyRef](2)
          structChildObjects(0) = convertSparkComplexTypeToCarbonObject(keys.get(i, keyDataType),
            keyDataType)
          structChildObjects(1) = convertSparkComplexTypeToCarbonObject(values.get(i,
            valueDataType), valueDataType)
          arrayMapChildObjects(i) = new StructObject(structChildObjects)
          i = i + 1
        }
        new ArrayObject(arrayMapChildObjects)
      case _: StructType =>
        val structDataType = objectDataType.asInstanceOf[StructType]
        val structData = data.asInstanceOf[UnsafeRow]
        val size = structData.numFields()
        val structChildObjects = new Array[AnyRef](size)
        var i = 0
        val childrenSchema = structDataType.fields
        while (i < size) {
          structChildObjects(i) = convertSparkComplexTypeToCarbonObject(structData.get(i,
            childrenSchema(i).dataType), childrenSchema(i).dataType)
          i = i + 1
        }
        new StructObject(structChildObjects)
      case _: DateType =>
        if (data == null) {
          CarbonCommonConstants.DIRECT_DICT_VALUE_NULL.asInstanceOf[AnyRef]
        } else {
          (data.asInstanceOf[Int] + DateDirectDictionaryGenerator.cutOffDate).asInstanceOf[AnyRef]
        }
      case _: TimestampType =>
        if (data == null) {
          null
        } else {
          (data.asInstanceOf[Long] / 1000).asInstanceOf[AnyRef]
        }
      case _ => data
    }
  }

  def convertComplexDataType(dataFieldsWithComplexDataType: Map[String,
    GenericDataType[_]], configuration: CarbonDataLoadConfiguration): Unit = {
    val fields = configuration.getDataFields
    val nullFormat = configuration
      .getDataLoadProperty(DataLoadProcessorConstants.SERIALIZATION_NULL_FORMAT)
      .toString
    for (field <- fields) {
      if (field.getColumn.isComplex) {
        dataFieldsWithComplexDataType +=
        (field.getColumn.getColName ->
         FieldEncoderFactory.createComplexDataType(field, nullFormat, null))
      }
    }
  }
}
