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

import scala.collection.JavaConverters._
import scala.collection.mutable

import org.apache.spark.sql._
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.execution.command._
import org.apache.spark.sql.execution.command.datamap.CarbonDropDataMapCommand
import org.apache.spark.sql.execution.command.management.CarbonLoadDataCommand
import org.apache.spark.sql.execution.command.table.CarbonCreateTableCommand
import org.apache.spark.sql.execution.command.timeseries.TimeSeriesUtil
import org.apache.spark.sql.optimizer.CarbonFilters
import org.apache.spark.sql.parser.CarbonSpark2SqlParser
import org.apache.spark.util.{DataMapUtil, PartitionUtils}

import org.apache.carbondata.common.exceptions.MetadataProcessException
import org.apache.carbondata.common.exceptions.sql.MalformedDataMapCommandException
import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.datamap.Segment
import org.apache.carbondata.core.metadata.schema.partition.PartitionType
import org.apache.carbondata.core.metadata.schema.table.CarbonTable
import org.apache.carbondata.core.statusmanager.{SegmentStatus, SegmentStatusManager}

/**
 * Below helper class will be used to create pre-aggregate table
 * and updating the parent table about the child table information
 * It will be either success or nothing happen in case of failure:
 * 1. failed to create pre aggregate table.
 * 2. failed to update main table
 *
 */
case class PreAggregateTableHelper(
    var parentTable: CarbonTable,
    dataMapName: String,
    dataMapClassName: String,
    dataMapProperties: java.util.Map[String, String],
    queryString: String,
    timeSeriesFunction: Option[String] = None,
    ifNotExistsSet: Boolean = false) {

  var loadCommand: CarbonLoadDataCommand = _

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

  def initMeta(sparkSession: SparkSession): Seq[Row] = {
    val dmProperties = dataMapProperties.asScala
    val updatedQuery = new CarbonSpark2SqlParser().addPreAggFunction(queryString)
    val df = sparkSession.sql(updatedQuery)
    val fieldRelationMap = PreAggregateUtil.validateActualSelectPlanAndGetAttributes(
      df.logicalPlan, queryString)

    val partitionInfo = parentTable.getPartitionInfo
    val fields = fieldRelationMap.keySet.toSeq
    val tableProperties = mutable.Map[String, String]()
    val usePartitioning = dataMapProperties.getOrDefault("partitioning", "true").toBoolean
    val parentPartitionColumns = if (!usePartitioning) {
      Seq.empty
    } else if (parentTable.isHivePartitionTable) {
      partitionInfo.getColumnSchemaList.asScala.map(_.getColumnName)
    } else {
      Seq()
    }
    // Generate child table partition columns in the same order as the parent table.
    val partitionerFields =
      PartitionUtils.getPartitionerFields(parentPartitionColumns, fieldRelationMap)

    dmProperties.foreach(t => tableProperties.put(t._1, t._2))

    val selectTable = PreAggregateUtil.getParentCarbonTable(df.logicalPlan)
    if (!parentTable.getTableName.equalsIgnoreCase(selectTable.getTableName)) {
      throw new MalformedDataMapCommandException(
        "Parent table name is different in select and create")
    }
    DataMapUtil
      .inheritTablePropertiesFromMainTable(parentTable, fields, fieldRelationMap, tableProperties)
    val tableIdentifier =
      TableIdentifier(parentTable.getTableName + "_" + dataMapName,
        Some(parentTable.getDatabaseName))
    // prepare table model of the collected tokens
    val tableModel: TableModel = new CarbonSpark2SqlParser().prepareTableModel(
      ifNotExistPresent = ifNotExistsSet,
      new CarbonSpark2SqlParser().convertDbNameToLowerCase(tableIdentifier.database),
      tableIdentifier.table.toLowerCase,
      fields,
      partitionerFields,
      tableProperties,
      None,
      isAlterFlow = false,
      true,
      None)

    // updating the relation identifier, this will be stored in child table
    // which can be used during dropping of pre-aggreate table as parent table will
    // also get updated
    if(timeSeriesFunction != null) {
      TimeSeriesUtil.validateTimeSeriesEventTime(dmProperties.toMap, parentTable)
      TimeSeriesUtil.validateEventTimeColumnExitsInSelect(
        fieldRelationMap,
        dmProperties(TimeSeriesUtil.TIMESERIES_EVENTTIME))
      TimeSeriesUtil.updateTimeColumnSelect(
        fieldRelationMap,
        dmProperties(TimeSeriesUtil.TIMESERIES_EVENTTIME),
        timeSeriesFunction.get)
    }
    tableModel.parentTable = Some(parentTable)
    tableModel.dataMapRelation = Some(fieldRelationMap)
    val tablePath = if (dmProperties.contains("path")) {
      dmProperties("path")
    } else {
      CarbonEnv.getTablePath(tableModel.databaseNameOp, tableModel.tableName)(sparkSession)
    }
    CarbonCreateTableCommand(TableNewProcessor(tableModel),
      tableModel.ifNotExistsSet, Some(tablePath), isVisible = false).run(sparkSession)

    val table = CarbonEnv.getCarbonTable(tableIdentifier)(sparkSession)
    val tableInfo = table.getTableInfo

    // child schema object will be saved on parent table schema
    val childSchema = tableInfo.getFactTable.buildChildSchema(
      dataMapName,
      dataMapClassName,
      tableInfo.getDatabaseName,
      queryString,
      "AGGREGATION")
    dmProperties.foreach(f => childSchema.getProperties.put(f._1, f._2))

    try {
      // updating the parent table about child table
      PreAggregateUtil.updateMainTable(parentTable, childSchema, sparkSession)
    } catch {
      case e: MetadataProcessException =>
        throw e
      case ex: Exception =>
        // If updation failed then forcefully remove datamap from metastore.
        val dropTableCommand = CarbonDropDataMapCommand(childSchema.getDataMapName,
          ifExistsSet = true,
          Some(TableIdentifier
            .apply(parentTable.getTableName, Some(parentTable.getDatabaseName))),
          forceDrop = true)
        dropTableCommand.processMetadata(sparkSession)
        throw ex
    }
    // After updating the parent carbon table with data map entry extract the latest table object
    // to be used in further create process.
    parentTable = CarbonEnv.getCarbonTable(Some(parentTable.getDatabaseName),
      parentTable.getTableName)(sparkSession)

    val updatedLoadQuery = if (timeSeriesFunction != null) {
      PreAggregateUtil.createTimeSeriesSelectQueryFromMain(
        childSchema.getChildSchema,
        parentTable.getTableName,
        parentTable.getDatabaseName)
    } else {
      queryString
    }
    val dataFrame = sparkSession.sql(new CarbonSpark2SqlParser().addPreAggLoadFunction(
      updatedLoadQuery)).drop("preAggLoad")
    loadCommand = PreAggregateUtil.createLoadCommandForChild(
      childSchema.getChildSchema.getListOfColumns,
      tableIdentifier,
      dataFrame,
      isOverwrite = false,
      sparkSession = sparkSession,
      mutable.Map.empty[String, String])
    loadCommand.processMetadata(sparkSession)
    Seq.empty
  }

  def initData(sparkSession: SparkSession): Seq[Row] = {
    // load child table if parent table has existing segments
    // This will be used to check if the parent table has any segments or not. If not then no
    // need to fire load for pre-aggregate table. Therefore reading the load details for PARENT
    // table.
    SegmentStatusManager.deleteLoadsAndUpdateMetadata(
      parentTable,
      false,
      CarbonFilters.getCurrentPartitions(
        sparkSession,
        TableIdentifier(parentTable.getTableName,
          Some(parentTable.getDatabaseName))
      ).map(_.asJava).orNull)

    if (SegmentStatusManager.isLoadInProgressInTable(parentTable)) {
      throw new UnsupportedOperationException(
        "Cannot create pre-aggregate table when insert is in progress on parent table")
    }
    // check if any segment if available for load in the parent table
    val loadAvailable = SegmentStatusManager.readLoadMetadata(parentTable.getMetadataPath)
      .collect {
        case segment if segment.getSegmentStatus == SegmentStatus.SUCCESS ||
                        segment.getSegmentStatus == SegmentStatus.LOAD_PARTIAL_SUCCESS =>
          new Segment(segment.getLoadName, segment.getSegmentFile).toString
      }
    if (loadAvailable.nonEmpty) {
      // Passing segmentToLoad as * because we want to load all the segments into the
      // pre-aggregate table even if the user has set some segments on the parent table.
      loadCommand.dataFrame = Some(PreAggregateUtil
        .getDataFrame(sparkSession, loadCommand.logicalPlan.get))
      PreAggregateUtil.startDataLoadForDataMap(
        TableIdentifier(parentTable.getTableName, Some(parentTable.getDatabaseName)),
        segmentToLoad = loadAvailable.mkString(","),
        validateSegments = false,
        loadCommand,
        isOverwrite = false,
        sparkSession)
    } else {
      LOGGER.info(s"No segment available for load in table ${parentTable.getTableUniqueName}")
    }
    Seq.empty
  }
}