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

import org.apache.spark.sql.CarbonToSparkAdapter
import org.apache.spark.sql.catalyst.analysis._
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.optimizer._
import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, _}
import org.apache.spark.sql.catalyst.rules.{RuleExecutor, _}
import org.apache.spark.sql.SQLConf
import org.apache.spark.sql.util.SparkSQLUtil

object BirdcageOptimizer extends RuleExecutor[LogicalPlan] {

  val conf = new SQLConf()

  protected val fixedPoint = FixedPoint(conf.optimizerMaxIterations)

  def batches: Seq[Batch] = {
    // Technically some of the rules in Finish Analysis are not optimizer rules and belong more
    // in the analyzer, because they are needed for correctness (e.g. ComputeCurrentTime).
    // However, because we also use the analyzer to canonicalized queries (for view definition),
    // we do not eliminate subqueries or compute current time in the analyzer.
    Batch(
      "Finish Analysis", Once,
      EliminateSubqueryAliases,
      SparkSQLUtil.getEliminateViewObj(),
      ReplaceExpressions,
      ComputeCurrentTime,
      //      GetCurrentDatabase(sessionCatalog),
      RewriteDistinctAggregates,
      ReplaceDeduplicateWithAggregate) ::
    //////////////////////////////////////////////////////////////////////////////////////////
    // Optimizer rules start here
    //////////////////////////////////////////////////////////////////////////////////////////
    // - Do the first call of CombineUnions before starting the major Optimizer rules,
    //   since it can reduce the number of iteration and the other rules could add/move
    //   extra operators between two adjacent Union operators.
    // - Call CombineUnions again in Batch("Operator Optimizations"),
    //   since the other rules might make two separate Unions operators adjacent.
    Batch(
      "Union", Once,
      CombineUnions) ::
    Batch(
      "Pullup Correlated Expressions", Once,
      SparkSQLUtil.getPullupCorrelatedPredicatesObj()) ::
    Batch(
      "Subquery", Once,
      OptimizeSubqueries) ::
    Batch(
      "Replace Operators", fixedPoint,
      ReplaceIntersectWithSemiJoin,
      ReplaceExceptWithAntiJoin,
      ReplaceDistinctWithAggregate) ::
    Batch(
      "Aggregate", fixedPoint,
      RemoveLiteralFromGroupExpressions,
      RemoveRepetitionFromGroupExpressions) ::
    Batch(
      "Operator Optimizations", fixedPoint, Seq(
        // Operator push down
        PushProjectionThroughUnion,
        SparkSQLUtil.getReorderJoinObj(conf),
        SparkSQLUtil.getEliminateOuterJoinObj(conf),
        PushPredicateThroughJoin,
        PushDownPredicate,
        //      LimitPushDown(conf),
        ColumnPruning,
        //      InferFiltersFromConstraints(conf),
        // Operator combine
        CollapseRepartition,
        CollapseProject,
        CollapseWindow,
        CombineFilters,
        CombineLimits,
        CombineUnions,
        // Constant folding and strength reduction
        SparkSQLUtil.getNullPropagationObj(conf),
        FoldablePropagation,
        //      OptimizeIn(conf),
        ConstantFolding,
        ReorderAssociativeOperator,
        // No need to apply LikeSimplification rule while creating datamap
        // as modular plan asCompactSql will be set in datamapschema
        //        LikeSimplification,
        BooleanSimplification,
        SimplifyConditionals,
        RemoveDispensableExpressions,
        SimplifyBinaryComparison,
        //      PruneFilters(conf),
        EliminateSorts,
        SimplifyCasts,
        SimplifyCaseConversionExpressions,
        RewriteCorrelatedScalarSubquery,
        EliminateSerialization,
        SparkSQLUtil.getRemoveRedundantAliasesObj(),
        RemoveRedundantProject) ++
                                            extendedOperatorOptimizationRules: _*) ::
    Batch(
      "Check Cartesian Products", Once,
      SparkSQLUtil.getCheckCartesianProductsObj(conf)) ::
    //    Batch("Join Reorder", Once,
    //      CostBasedJoinReorder(conf)) ::
    //    Batch("Decimal Optimizations", fixedPoint,
    //      DecimalAggregates(conf)) ::
    Batch(
      "Object Expressions Optimization", fixedPoint,
      EliminateMapObjects,
      CombineTypedFilters) ::
    //    Batch("LocalRelation", fixedPoint,
    //      ConvertToLocalRelation,
    //      PropagateEmptyRelation) ::
    Batch(
      "OptimizeCodegen", Once, CarbonToSparkAdapter.getOptimizeCodegenRule(conf): _*) ::
    Batch(
      "RewriteSubquery", Once,
      RewritePredicateSubquery,
      CollapseProject) :: Nil
  }

  /**
   * Optimize all the subqueries inside expression.
   */
  object OptimizeSubqueries extends Rule[LogicalPlan] {
    def apply(plan: LogicalPlan): LogicalPlan = {
      plan transformAllExpressions {
        case s: SubqueryExpression =>
          val Subquery(newPlan) = BirdcageOptimizer.this.execute(Subquery(s.plan))
          s.withNewPlan(newPlan)
      }
    }
  }

  /**
   * Override to provide additional rules for the operator optimization batch.
   */
  def extendedOperatorOptimizationRules: Seq[Rule[LogicalPlan]] = {
    Nil
  }
}

/**
 * Push Aggregate through join to fact table.
 * Pushes down [[Aggregate]] operators where the `grouping` and `aggregate` expressions can
 * be evaluated using only the attributes of the fact table, the left or right side of a
 * star-join.
 * Other [[Aggregate]] expressions stay in the original [[Aggregate]].
 *
 * Check 'Aggregate Pushdown Over Join: Design & Preliminary Results' by LiTao for more details
 */
// case class PushAggregateThroughJoin(conf: SQLConf) extends Rule[LogicalPlan] with
// PredicateHelper {
//
//  val tableCluster = {
//    val mapper = new ObjectMapper().registerModule(DefaultScalaModule)
//    val tableClusterString = conf.getConfString("spark.mv.tableCluster")
//    mapper.readValue(tableClusterString, classOf[TableCluster])
//  }
//
//  def apply(plan: LogicalPlan): LogicalPlan = plan transform {
//
//    // Push down aggregate expressions through Join
//    case a @ Aggregate(grouping, aggregate, Project(projectList, Join(left, right, jt, cond)))
//        if (left.isInstanceOf[LeafNode] && => {
//      val fTables: Set[String] = tableCluster.getFact
//      val dTables: Set[String] = tableCluster.getDimension
// //      if canPushThrough(left,a)
//
//      if (fTables.contains(s"${left.databaseName}.${left.tableName}")
//          Aggregate(newGrouping, newAggregate, Project(projectList, Join(Aggregate(_,_,Project
// (projectList1, left)), right, jt, cond)))
//      }
//    }
//
//  private def canPushThrough(join: Join): Boolean = join match {
//    case Join(left : LeafNode, right: LeafNode, Inner, EqualTo(l: AttributeReference,
// r: AttributeReference)) => true
//
//
//  }
//
//
// }