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

import scala.collection.JavaConverters._

import org.apache.spark.SPARK_VERSION
import org.apache.spark.sql._
import org.apache.spark.sql.CarbonExpressions.CarbonUnresolvedRelation
import org.apache.spark.sql.catalyst.analysis.{UnresolvedAlias, UnresolvedAttribute, UnresolvedFunction, UnresolvedRelation, UnresolvedStar}
import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, Cast, NamedExpression}
import org.apache.spark.sql.catalyst.plans.Inner
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.rules._
import org.apache.spark.sql.execution.command.mutation.CarbonProjectForDeleteCommand
import org.apache.spark.sql.execution.datasources.{CatalogFileIndex, FileFormat, HadoopFsRelation, LogicalRelation, SparkCarbonTableFormat}
import org.apache.spark.sql.util.CarbonException
import org.apache.spark.util.{CarbonReflectionUtils, DataMapUtil, SparkUtil}

import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.datamap.DataMapStoreManager
import org.apache.carbondata.core.datamap.status.DataMapStatusManager
import org.apache.carbondata.core.util.CarbonUtil

case class CarbonIUDAnalysisRule(sparkSession: SparkSession) extends Rule[LogicalPlan] {

  private lazy val parser = sparkSession.sessionState.sqlParser
  private lazy val optimizer = sparkSession.sessionState.optimizer
  private lazy val analyzer = sparkSession.sessionState.analyzer

  private def processUpdateQuery(
      table: UnresolvedRelation,
      columns: List[String],
      selectStmt: String,
      alias: Option[String],
      filter: String): LogicalPlan = {
    var includedDestColumns = false
    var includedDestRelation = false
    var addedTupleId = false

    def prepareTargetReleation(relation: UnresolvedRelation): SubqueryAlias = {
      val tupleId = UnresolvedAlias(Alias(UnresolvedFunction("getTupleId",
        Seq.empty, isDistinct = false), "tupleId")())

      val projList = Seq(UnresolvedAlias(UnresolvedStar(alias.map(Seq(_)))), tupleId)
      val carbonTable = CarbonEnv.getCarbonTable(table.tableIdentifier)(sparkSession)
      if (carbonTable != null) {
        val indexSchemas = DataMapStoreManager.getInstance().getDataMapSchemasOfTable(carbonTable)
        if (DataMapUtil.hasMVDataMap(carbonTable)) {
          val allDataMapSchemas = DataMapStoreManager.getInstance
            .getDataMapSchemasOfTable(carbonTable).asScala
            .filter(dataMapSchema => null != dataMapSchema.getRelationIdentifier &&
                                     !dataMapSchema.isIndexDataMap).asJava
          allDataMapSchemas.asScala.foreach { dataMapSchema =>
            DataMapStatusManager.disableDataMap(dataMapSchema.getDataMapName)
          }
        } else if (!indexSchemas.isEmpty) {
          throw new UnsupportedOperationException(
            "Update operation is not supported for table which has index datamaps")
        }
        if (carbonTable.isChildTableForMV) {
          throw new UnsupportedOperationException(
            "Update operation is not supported for mv datamap table")
        }
      }
      val tableRelation = if (SparkUtil.isSparkVersionEqualTo("2.1")) {
        relation
      } else if (SparkUtil.isSparkVersionXandAbove("2.2")) {
        alias match {
          case Some(_) =>
            CarbonReflectionUtils.getSubqueryAlias(
              sparkSession,
              alias,
              relation,
              Some(table.tableIdentifier))
          case _ => relation
        }
      } else {
        throw new UnsupportedOperationException("Unsupported Spark version.")
      }

      CarbonReflectionUtils.getSubqueryAlias(
        sparkSession,
        alias,
        Project(projList, tableRelation),
        Some(table.tableIdentifier))
    }

    // get the un-analyzed logical plan
    val targetTable = prepareTargetReleation(table)
    val selectPlan = parser.parsePlan(selectStmt) transform {
      case Project(projectList, child) if !includedDestColumns =>
        includedDestColumns = true
        if (projectList.size != columns.size) {
          CarbonException.analysisException(
            "The number of columns in source table and destination table columns mismatch")
        }
        val renamedProjectList = projectList.zip(columns).map { case (attr, col) =>
          attr match {
            case UnresolvedAlias(child22, _) =>
              UnresolvedAlias(Alias(child22, col + CarbonCommonConstants.UPDATED_COL_EXTENSION)())
            case UnresolvedAttribute(_) =>
              UnresolvedAlias(Alias(attr, col + CarbonCommonConstants.UPDATED_COL_EXTENSION)())
            case _ => attr
          }
        }
        val tableName: Option[Seq[String]] = alias match {
          case Some(_) => Some(alias.toSeq)
          case _ => Some(Seq(child.asInstanceOf[UnresolvedRelation].tableIdentifier.table.toString))
        }
        val list = Seq(
          UnresolvedAlias(UnresolvedStar(tableName))) ++ renamedProjectList
        Project(list, child)
      case Filter(cond, child) if !includedDestRelation =>
        includedDestRelation = true
        Filter(cond, Join(child, targetTable, Inner, None))
      case r@CarbonUnresolvedRelation(t) if !includedDestRelation && t != table.tableIdentifier =>
        includedDestRelation = true
        Join(r, targetTable, Inner, None)
    }
    val updatedSelectPlan: LogicalPlan = if (!includedDestRelation) {
      // special case to handle self join queries
      // Eg. update tableName  SET (column1) = (column1+1)
      selectPlan transform {
        case relation: UnresolvedRelation
          if table.tableIdentifier == relation.tableIdentifier && !addedTupleId =>
          addedTupleId = true
          targetTable
      }
    } else {
      selectPlan
    }
    val finalPlan = if (filter.length > 0) {
      var transformed: Boolean = false
      // Create a dummy projection to include filter conditions
      var newPlan: LogicalPlan = null
      if (table.tableIdentifier.database.isDefined) {
        newPlan = parser.parsePlan("select * from  " +
           table.tableIdentifier.database.getOrElse("") + "." +
           table.tableIdentifier.table + " " + alias.getOrElse("") + " " + filter)
      }
      else {
        newPlan = parser.parsePlan("select * from  " +
           table.tableIdentifier.table + " " + alias.getOrElse("") + " " + filter)
      }
      newPlan transform {
        case CarbonUnresolvedRelation(t)
          if !transformed && t == table.tableIdentifier =>
          transformed = true

          CarbonReflectionUtils.getSubqueryAlias(
            sparkSession,
            alias,
            updatedSelectPlan,
            Some(table.tableIdentifier))
      }
    } else {
      updatedSelectPlan
    }
    val destinationTable = CarbonReflectionUtils.getUnresolvedRelation(table.tableIdentifier, alias)

    // In Spark 2.1 and 2.2, it uses Analyzer.execute method to transform LogicalPlan
    // but in Spark 2.3, it uses Analyzer.executeAndCheck method
    val analyzedPlan = CarbonReflectionUtils.invokeAnalyzerExecute(
        analyzer, ProjectForUpdate(destinationTable, columns, Seq(finalPlan)))
    // For all commands, they execute eagerly, and will be transformed to
    // logical plan 'LocalRelation' in analyze phase(please see the code in 'Dataset.logicalPlan'),
    // so it needs to return logical plan 'CarbonProjectForUpdateCommand' here
    // instead of 'ProjectForUpdate'
    optimizer.execute(analyzedPlan)
  }


  def processDeleteRecordsQuery(selectStmt: String,
      alias: Option[String],
      table: UnresolvedRelation): LogicalPlan = {
    var addedTupleId = false
    val parsePlan = parser.parsePlan(selectStmt)

    val selectPlan = parsePlan transform {
      case relation: UnresolvedRelation
        if table.tableIdentifier == relation.tableIdentifier && !addedTupleId =>
        addedTupleId = true
        val tupleId = UnresolvedAlias(Alias(UnresolvedFunction("getTupleId",
          Seq.empty, isDistinct = false), "tupleId")())

        val projList = Seq(UnresolvedAlias(UnresolvedStar(alias.map(Seq(_)))), tupleId)
        val carbonTable = CarbonEnv.getCarbonTable(table.tableIdentifier)(sparkSession)
        if (carbonTable != null) {
          if (carbonTable.isChildTableForMV) {
            throw new UnsupportedOperationException(
              "Delete operation is not supported for datamap table")
          }
          val indexSchemas = DataMapStoreManager.getInstance().getDataMapSchemasOfTable(carbonTable)
          if (DataMapUtil.hasMVDataMap(carbonTable)) {
            val allDataMapSchemas = DataMapStoreManager.getInstance
              .getDataMapSchemasOfTable(carbonTable).asScala
              .filter(dataMapSchema => null != dataMapSchema.getRelationIdentifier &&
                                       !dataMapSchema.isIndexDataMap).asJava
            allDataMapSchemas.asScala.foreach { dataMapSchema =>
              DataMapStatusManager.disableDataMap(dataMapSchema.getDataMapName)
            }
          } else if (!indexSchemas.isEmpty) {
            throw new UnsupportedOperationException(
              "Delete operation is not supported for table which has index datamaps")
          }
        }
        // include tuple id in subquery
        if (SparkUtil.isSparkVersionEqualTo("2.1")) {
          Project(projList, relation)
        } else if (SparkUtil.isSparkVersionXandAbove("2.2")) {
          alias match {
            case Some(_) =>
              val subqueryAlias = CarbonReflectionUtils.getSubqueryAlias(
                sparkSession,
                alias,
                relation,
                Some(table.tableIdentifier))
              Project(projList, subqueryAlias)
            case _ => Project(projList, relation)
          }
        } else {
          throw new UnsupportedOperationException("Unsupported Spark version.")
        }
    }
    CarbonProjectForDeleteCommand(
      selectPlan,
      table.tableIdentifier.database,
      table.tableIdentifier.table,
      System.currentTimeMillis().toString)
  }

  override def apply(logicalplan: LogicalPlan): LogicalPlan = {

    logicalplan transform {
      case UpdateTable(t, cols, sel, alias, where) => processUpdateQuery(t, cols, sel, alias, where)
      case DeleteRecords(statement, alias, table) =>
        processDeleteRecordsQuery(
          statement,
          alias,
          table)
    }
  }
}

/**
 * Insert into carbon table from other source
 */
case class CarbonPreInsertionCasts(sparkSession: SparkSession) extends Rule[LogicalPlan] {
  def apply(plan: LogicalPlan): LogicalPlan = {
    plan.transform {
      // Wait until children are resolved.
      case p: LogicalPlan if !p.childrenResolved => p

      case p@InsertIntoTable(relation: LogicalRelation, _, child, _, _)
        if relation.relation.isInstanceOf[CarbonDatasourceHadoopRelation] =>
        castChildOutput(p, relation, child)
    }
  }

  def castChildOutput(p: InsertIntoTable,
      relation: LogicalRelation,
      child: LogicalPlan): LogicalPlan = {
    val carbonDSRelation = relation.relation.asInstanceOf[CarbonDatasourceHadoopRelation]
    if (carbonDSRelation.carbonRelation.output.size > CarbonCommonConstants
      .DEFAULT_MAX_NUMBER_OF_COLUMNS) {
      CarbonException.analysisException(
        s"Maximum number of columns supported: " +
          s"${CarbonCommonConstants.DEFAULT_MAX_NUMBER_OF_COLUMNS}")
    }
    // In spark, PreprocessTableInsertion rule has below cast logic.
    // It was missed in carbon when implemented insert into rules.
    var newChildOutput = if (child.output.size >= carbonDSRelation.carbonRelation.output.size) {
      val expectedOutput = carbonDSRelation.carbonRelation.output
      child.output.zip(expectedOutput).map {
        case (actual, expected) =>
          if (expected.dataType.sameType(actual.dataType) &&
              expected.name == actual.name &&
              expected.metadata == actual.metadata) {
            actual
          } else {
            // Renaming is needed for handling the following cases like
            // 1) Column names/types do not match, e.g., INSERT INTO TABLE tab1 SELECT 1, 2
            // 2) Target tables have column metadata
            Alias(Cast(actual, expected.dataType), expected.name)(
              explicitMetadata = Option(expected.metadata))
          }
      }
    } else {
      child.output
    }
    if (newChildOutput.size >= carbonDSRelation.carbonRelation.output.size ||
        carbonDSRelation.carbonTable.isHivePartitionTable) {
      newChildOutput = newChildOutput.zipWithIndex.map { columnWithIndex =>
        columnWithIndex._1 match {
          case attr: Attribute =>
            Alias(attr, s"col${ columnWithIndex._2 }")(NamedExpression.newExprId)
          case attr => attr
        }
      }
      val newChild: LogicalPlan = if (newChildOutput == child.output) {
        if (SparkUtil.isSparkVersionEqualTo("2.1")) {
          CarbonReflectionUtils.getField("child", p).asInstanceOf[LogicalPlan]
        } else if (SparkUtil.isSparkVersionEqualTo("2.2")) {
          CarbonReflectionUtils.getField("query", p).asInstanceOf[LogicalPlan]
        } else {
          throw new UnsupportedOperationException(s"Spark version $SPARK_VERSION is not supported")
        }
      } else {
        Project(newChildOutput, child)
      }

      val overwrite = CarbonReflectionUtils.getOverWriteOption("overwrite", p)

      InsertIntoCarbonTable(carbonDSRelation, p.partition, newChild, overwrite, true)
    } else {
      CarbonException.analysisException(
        "Cannot insert into target table because number of columns mismatch")
    }
  }
}