/*
 * 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.mv.datamap

import scala.collection.JavaConverters._

import org.apache.spark.sql.{CarbonEnv, SparkSession}
import org.apache.spark.sql.catalyst.analysis.{UnresolvedAlias, UnresolvedAttribute}
import org.apache.spark.sql.catalyst.catalog.CatalogTable
import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, ScalaUDF}
import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Command, DeserializeToObject, LogicalPlan}
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.execution.datasources.LogicalRelation

import org.apache.carbondata.common.logging.LogServiceFactory
import org.apache.carbondata.core.constants.CarbonCommonConstants
import org.apache.carbondata.core.datamap.DataMapStoreManager
import org.apache.carbondata.core.metadata.schema.datamap.DataMapClassProvider
import org.apache.carbondata.core.metadata.schema.table.DataMapSchema
import org.apache.carbondata.core.util.ThreadLocalSessionInfo
import org.apache.carbondata.datamap.DataMapManager
import org.apache.carbondata.mv.plans.modular.{ModularPlan, Select}
import org.apache.carbondata.mv.rewrite.{SummaryDataset, SummaryDatasetCatalog}

/**
 * Analyzer rule to rewrite the query for MV datamap
 *
 * @param sparkSession
 */
class MVAnalyzerRule(sparkSession: SparkSession) extends Rule[LogicalPlan] {

  // TODO Find way better way to get the provider.
  private val dataMapProvider =
    DataMapManager.get().getDataMapProvider(null,
      new DataMapSchema("", DataMapClassProvider.MV.getShortName), sparkSession)

  private val LOGGER = LogServiceFactory.getLogService(classOf[MVAnalyzerRule].getName)

  override def apply(plan: LogicalPlan): LogicalPlan = {
    var needAnalysis = true
    plan.transformAllExpressions {
      // first check if any mv UDF is applied it is present is in plan
      // then call is from create MV so no need to transform the query plan
      // TODO Add different UDF name
      case al@Alias(udf: ScalaUDF, name) if name.equalsIgnoreCase(CarbonEnv.MV_SKIP_RULE_UDF) =>
        needAnalysis = false
        al
      // in case of query if any unresolve alias is present then wait for plan to be resolved
      // return the same plan as we can tranform the plan only when everything is resolved
      case unresolveAlias@UnresolvedAlias(_, _) =>
        needAnalysis = false
        unresolveAlias
      case attr@UnresolvedAttribute(_) =>
        needAnalysis = false
        attr
    }
    plan.transform {
      case aggregate@Aggregate(grp, aExp, child) =>
        // check for if plan is for dataload for preaggregate table, then skip applying mv
        val isPreAggLoad = aExp.exists { p =>
          if (p.isInstanceOf[UnresolvedAlias]) {
            false
          } else {
            p.name.equals(CarbonEnv.MV_SKIP_RULE_UDF)
          }
        }
        if (isPreAggLoad) {
          needAnalysis = false
        }
        Aggregate(grp, aExp, child)
    }
    val catalog = DataMapStoreManager.getInstance().getDataMapCatalog(dataMapProvider,
      DataMapClassProvider.MV.getShortName).asInstanceOf[SummaryDatasetCatalog]
    if (needAnalysis && catalog != null && isValidPlan(plan, catalog)) {
      val modularPlan = catalog.mvSession.sessionState.rewritePlan(plan).withMVTable
      if (modularPlan.find(_.rewritten).isDefined) {
        var compactSQL = modularPlan.asCompactSQL
        compactSQL = reWriteTheUDFInSQLWithQualifierName(modularPlan, compactSQL)
        val analyzed = sparkSession.sql(compactSQL).queryExecution.analyzed
        analyzed
      } else {
        plan
      }
    } else {
      plan
    }
  }

  /**
   * This method is specially handled for timeseries on MV, because when we use timeseries UDF which
   * is a scala UDF, so after plan matching when query is made. We get as below query for example
   *
   * SELECT gen_subsumer_0.`UDF:timeseries(projectjoindate, hour)` AS `UDF:timeseries(projectjoi...
   * FROM
   * (SELECT datamap1_table.`UDF:timeseries_projectjoindate_hour` AS `UDF:timeseries(projectjoin...
   * FROM
   *     default.datamap1_table
   * GROUP BY datamap1_table.`UDF:timeseries_projectjoindate_hour`) gen_subsumer_0
   * WHERE
   * (UDF:timeseries(projectjoindate, hour) = TIMESTAMP('2016-02-23 09:00:00.0'))
   *
   * Here for Where filter expression is of type ScalaUDF, so when we do .sql() to prepare SQL, we
   * get without qualifier name(Refer org.apache.spark.sql.catalyst.expressions.NonSQLExpression)
   * which is 'gen_subsumer_0', so this funtion rewrites with qualifier name and returns, so that
   * parsing does not fail in spark, for rewritten MV query.
   * @param plan Modular Plan
   * @param compactSQL compactSQL generated from Modular plan
   * @return Rewritten plan with the qualifier names for where clauses in query.
   */
  private def reWriteTheUDFInSQLWithQualifierName(plan: ModularPlan, compactSQL: String): String = {
    var outPutUDFColumn = ""
    var reWrittenQuery = compactSQL
    plan match {
      case select: Select =>
        select.outputList.collect {
          case a: Alias if a.child.isInstanceOf[Attribute] =>
            val childName = a.child.asInstanceOf[Attribute].name
            if (childName.startsWith("UDF:timeseries")) {
              outPutUDFColumn = childName
            }
        }
        var queryArray: Array[String] = Array.empty
        if (!outPutUDFColumn.equalsIgnoreCase("") && compactSQL.contains("WHERE")) {
          queryArray = compactSQL.split("\n")
          queryArray(queryArray.indexOf("WHERE") + 1) = queryArray(
            queryArray.indexOf("WHERE") + 1).replace(outPutUDFColumn,
            s"gen_subsumer_0.`$outPutUDFColumn`")
          reWrittenQuery = queryArray.mkString("\n")
        }
        reWrittenQuery
      case _ =>
        compactSQL
    }
  }

  /**
   * Whether the plan is valid for doing modular plan matching and datamap replacing.
   */
  def isValidPlan(plan: LogicalPlan, catalog: SummaryDatasetCatalog): Boolean = {
    if (!plan.isInstanceOf[Command]  && !plan.isInstanceOf[DeserializeToObject]) {
      val catalogs = extractCatalogs(plan)
      !isDataMapReplaced(catalog.listAllValidSchema(), catalogs) &&
      isDataMapExists(catalog.listAllValidSchema(), catalogs) &&
      !isSegmentSetForMainTable(catalogs)
    } else {
      false
    }

  }
  /**
   * Check whether datamap table already updated in the query.
   *
   * @param mvdataSetArray Array of available mvdataset which include modular plans
   * @return Boolean whether already datamap replaced in the plan or not
   */
  def isDataMapReplaced(
      mvdataSetArray: Array[SummaryDataset],
      catalogs: Seq[Option[CatalogTable]]): Boolean = {
    catalogs.exists { c =>
      mvdataSetArray.exists { mv =>
        val identifier = mv.dataMapSchema.getRelationIdentifier
        identifier.getTableName.equals(c.get.identifier.table) &&
        identifier.getDatabaseName.equals(c.get.database)
      }
    }
  }

  /**
   * Check whether any suitable datamaps(like datamap which parent tables are present in the plan)
   * exists for this plan.
   *
   * @param mvs
   * @return
   */
  def isDataMapExists(mvs: Array[SummaryDataset], catalogs: Seq[Option[CatalogTable]]): Boolean = {
    catalogs.exists { c =>
      mvs.exists { mv =>
        mv.dataMapSchema.getParentTables.asScala.exists { identifier =>
          identifier.getTableName.equals(c.get.identifier.table) &&
          identifier.getDatabaseName.equals(c.get.database)
        }
      }
    }
  }

  private def extractCatalogs(plan: LogicalPlan): Seq[Option[CatalogTable]] = {
    val catalogs = plan collect {
      case l: LogicalRelation => l.catalogTable
    }
    catalogs
  }

  /**
   * Check if any segments are set for main table for Query. If any segments are set, then
   * skip mv datamap table for query
   */
  def isSegmentSetForMainTable(catalogs: Seq[Option[CatalogTable]]): Boolean = {
    catalogs.foreach { c =>
      val carbonSessionInfo = ThreadLocalSessionInfo.getCarbonSessionInfo
      if (carbonSessionInfo != null) {
        val segmentsToQuery = carbonSessionInfo.getSessionParams
          .getProperty(CarbonCommonConstants.CARBON_INPUT_SEGMENTS +
                       c.get.identifier.database.get + "." +
                       c.get.identifier.table, "")
        if (segmentsToQuery.isEmpty || segmentsToQuery.equalsIgnoreCase("*")) {
          return false
        } else {
          return true
        }
      } else {
        return false
      }
    }
    false
  }

}