/*
 * 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 scala.collection.mutable
import scala.collection.mutable.ArrayBuffer

import org.apache.spark.sql._
import org.apache.spark.sql.catalyst.TableIdentifier
import org.apache.spark.sql.CarbonExpressions.{CarbonScalaUDF, CarbonSubqueryAlias, MatchCastExpression}
import org.apache.spark.sql.catalyst.analysis.{UnresolvedAlias, UnresolvedAttribute}
import org.apache.spark.sql.catalyst.expressions.{Alias, AttributeReference, Cast, Divide, Expression, Literal, NamedExpression, ScalaUDF, SortOrder}
import org.apache.spark.sql.catalyst.expressions.aggregate.{Count, _}
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.execution.command.preaaggregate.PreAggregateUtil
import org.apache.spark.sql.execution.datasources.{FindDataSourceTable, LogicalRelation}
import org.apache.spark.sql.parser.CarbonSpark2SqlParser
import org.apache.spark.sql.types._
import org.apache.spark.util.CarbonReflectionUtils

import org.apache.carbondata.core.constants.{CarbonCommonConstants, CarbonCommonConstantsInternal}
import org.apache.carbondata.core.metadata.schema.datamap.DataMapClassProvider
import org.apache.carbondata.core.metadata.schema.table.{AggregationDataMapSchema, CarbonTable, DataMapSchema}
import org.apache.carbondata.core.metadata.schema.table.column.ColumnSchema
import org.apache.carbondata.core.preagg.{AggregateQueryPlan, AggregateTableSelector, QueryColumn}
import org.apache.carbondata.core.profiler.ExplainCollector
import org.apache.carbondata.core.statusmanager.SegmentStatusManager
import org.apache.carbondata.core.util.{CarbonUtil, ThreadLocalSessionInfo}

/**
 * model class to store aggregate expression logical plan
 * and its column schema mapping
 * @param expression aggregate expression
 * @param columnSchema list of column schema from table
 */
case class AggExpToColumnMappingModel(
    expression: Expression,
    var columnSchema: Option[Object] = None) {
  override def equals(o: Any) : Boolean = o match {
    case that: AggExpToColumnMappingModel =>
      that.expression==this.expression
    case _ => false
  }
  // TODO need to update the hash code generation code
  override def hashCode : Int = 1
}
/**
 * Class for applying Pre Aggregate rules
 * Responsibility.
 * 1. Check plan is valid plan for updating the parent table plan with child table
 * 2. Updated the plan based on child schema
 *
 * Rules for Updating the plan
 * 1. Grouping expression rules
 *    1.1 Change the parent attribute reference for of group expression
 * to child attribute reference
 *
 * 2. Aggregate expression rules
 *    2.1 Change the parent attribute reference for of group expression to
 * child attribute reference
 *    2.2 Change the count AggregateExpression to Sum as count
 * is already calculated so in case of aggregate table
 * we need to apply sum to get the count
 *    2.2 In case of average aggregate function select 2 columns from aggregate table with
 * aggregation
 * sum and count. Then add divide(sum(column with sum), sum(column with count)).
 * Note: During aggregate table creation for average table will be created with two columns
 * one for sum(column) and count(column) to support rollup
 *
 * 3. Filter Expression rules.
 *    3.1 Updated filter expression attributes with child table attributes
 * 4. Update the Parent Logical relation with child Logical relation
 * 5. Order By Query rules.
 *    5.1 Update project list based on updated aggregate expression
 *    5.2 Update sort order attributes based on pre aggregate table
 * 6. timeseries function
 *    6.1 validate maintable has timeseries datamap
 *    6.2 timeseries function is valid function or not
 * 7. Streaming
 * Examples1:
 * Query:
 *   SELECT name, sum(Salary) as totalSalary
 *   FROM maintable.
 * UpdatedQuery:
 *   SELECT name, sum(totalSalary) FROM(
 *          SELECT name, sum(Salary) as totalSalary
 *          FROM maintable
 *          GROUP BY name
 *          UNION ALL
 *          SELECT maintable_name,sum(maintable_salary) as totalSalary
 *          FROM maintable_agg
 *          GROUP BY maintable_name)
 *   GROUP BY name)
 * Example2:
 * Query:
 *   SELECT name, AVG(Salary) as avgSalary
 *        FROM maintable.
 * UpdatedQuery:
 *   SELECT name, Divide(sum(sumSalary)/sum(countsalary))
 *   FROM(
 *    SELECT name, sum(Salary) as sumSalary,count(salary) countsalary
 *      FROM maintable
 *      GROUP BY name
 *    UNION ALL
 *    SELECT maintable_name,sum(maintable_salary) as sumSalary, count(maintable_salary) countsalary
 *      FROM maintable_agg
 *      GROUP BY maintable_name)
 *   GROUP BY name)
 *
 * Rules for updating plan in case of streaming table:
 * In case of streaming data will be fetched from both fact and aggregate as aggregate table
 * will be updated only after each hand-off, so current streamed data won't be available on
 * aggregate table.
 * 7.1 Add one union node to add both fact and aggregate table plan to get the data from both table
 * 7.2 On top of Union Node add one Aggregate node to aggregate both table results
 * 7.3 In case of average(avg(column)) special handling is required for streaming
 *     7.3.1 Fact Plan will updated to return sum(column) and count(column) to do rollup
 *     7.3.2 Aggregate Plan will updated to return sum(column) and count(column) to do rollup
 * 7.4 In newly added Aggregate node all the aggregate expression must have same expression id as
 *     fact and fact plan will updated with new expression id. As query like order by this can be
 *     referred. In example1 sum(totalSalary) as totalSalary will have same expression id
 *     as in fact and fact plan sum(salary) will be updated with new expression id
 *
 * @param sparkSession spark session
 */
case class CarbonPreAggregateQueryRules(sparkSession: SparkSession) extends Rule[LogicalPlan] {

  /**
   * map for keeping parent attribute reference to child attribute reference
   * this will be used to updated the plan in case of join or order by
   */
  val updatedExpression = mutable.HashMap[AttributeReference, AttributeReference]()

  /**
   * parser
   */
  lazy val parser = new CarbonSpark2SqlParser

  /**
   * Below method will be used to validate the logical plan
   * @param logicalPlan query logical plan
   * @return isvalid or not
   */
  private def isValidPlan(logicalPlan: LogicalPlan) : Boolean = {
    var isValidPlan = true
    logicalPlan.transform {
      case aggregate@Aggregate(grp, aExp, child) =>
        isValidPlan = !aExp.exists { p =>
          if (p.isInstanceOf[UnresolvedAlias]) return false
          p.name.equals("preAggLoad") || p.name.equals("preAgg")
        }
        val updatedAggExp = aExp.filterNot(_.name.equalsIgnoreCase("preAggLoad"))
        Aggregate(grp, updatedAggExp, child)
    }
    isValidPlan
  }
  override def apply(plan: LogicalPlan): LogicalPlan = {
    var needAnalysis = true
    plan.transformExpressions {
      // first check if any preAgg scala function is applied it is present is in plan
      // then call is from create preaggregate table class so no need to transform the query plan
      case al@Alias(udf: ScalaUDF, name) if name.equalsIgnoreCase("preAgg") =>
        needAnalysis = false
        al
      case al@Alias(udf: ScalaUDF, name) if name.equalsIgnoreCase("preAggLoad") =>
        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
    }
    if(needAnalysis) {
      needAnalysis = isValidPlan(plan)
      if(needAnalysis) {
        needAnalysis = validateStreamingTablePlan(plan)
      }
    }
    // if plan is not valid for transformation then return same plan
    if (!needAnalysis) {
      plan
    } else {
      val updatedPlan = transformPreAggQueryPlan(plan)
      val newPlan = updatePlan(updatedPlan)
      newPlan
    }

  }

  /**
   * Below method will be used validate whether plan is already updated in case of streaming table
   * In case of streaming table it will add UnionNode to get the data from fact and aggregate both
   * as aggregate table will be updated after each handoff.
   * So if plan is already updated no need to transform the plan again
   * @param logicalPlan
   * query plan
   * @return whether need to update the query plan or not
   */
  def validateStreamingTablePlan(logicalPlan: LogicalPlan) : Boolean = {
    var needTransformation: Boolean = true
    logicalPlan.transform {
      case union @ Union(Seq(plan1, plan2)) =>
        plan2.collect{
          case logicalRelation: LogicalRelation if
          logicalRelation.relation.isInstanceOf[CarbonDatasourceHadoopRelation] &&
          logicalRelation.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonTable
            .isChildDataMap =>
            needTransformation = false
        }
        union
    }
    needTransformation
  }

  /**
   * Below method will be used to update the child plan
   * This will be used for updating expression like join condition,
   * order by, project list etc
   * @param plan child plan
   * @return updated plan
   */
  def updatePlan(plan: LogicalPlan) : LogicalPlan = {
    val updatedPlan = plan transform {
      case Aggregate(grp, aggExp, child) =>
        Aggregate(
          updateExpression(grp),
          updateExpression(aggExp.asInstanceOf[Seq[Expression]]).asInstanceOf[Seq[NamedExpression]],
          child)
      case Filter(filterExp, child) =>
        Filter(updateExpression(Seq(filterExp)).head, child)
      case Project(pList, child) =>
        Project(
          updateExpression(pList.asInstanceOf[Seq[Expression]]).asInstanceOf[Seq[NamedExpression]],
          child)
      case Sort(sortOrders, global, child) =>
        Sort(updateSortExpression(sortOrders), global, child)
      case Join(left, right, joinType, condition) =>
        val updatedCondition = condition match {
          case Some(expr) => Some(updateExpression(Seq(expr)).head)
          case _ => condition
        }
        Join(left, right, joinType, updatedCondition)
    }
    updatedPlan
  }

  /**
   * Below method will be used to update the sort expression
   * @param sortExp sort order expression in query
   * @return updated sort expression
   */
  def updateSortExpression(sortExp : Seq[SortOrder]) : Seq[SortOrder] = {
    sortExp map { order =>
      SortOrder(order.child transform  {
        case attr: AttributeReference =>
          updatedExpression.find { p => p._1.sameRef(attr) } match {
            case Some((_, childAttr)) =>
              CarbonToSparkAdapter.createAttributeReference(
                childAttr.name,
                childAttr.dataType,
                childAttr.nullable,
                childAttr.metadata,
                childAttr.exprId,
                attr.qualifier,
                attr)
            case None =>
              attr
          }
      }, order.direction )
    }
  }

  /**
   * Below method will be used to update the expression like group by expression
   * @param expressions sequence of expression like group by
   * @return updated expressions
   */
  def updateExpression(expressions : Seq[Expression]) : Seq[Expression] = {
    expressions map { expression =>
      expression transform {
        case attr: AttributeReference =>
          updatedExpression.find { p => p._1.sameRef(attr) } match {
            case Some((_, childAttr)) =>
              CarbonToSparkAdapter.createAttributeReference(
                childAttr.name,
                childAttr.dataType,
                childAttr.nullable,
                childAttr.metadata,
                childAttr.exprId,
                attr.qualifier,
                attr)
            case None =>
              attr
          }
      }
    }
  }

  /**
   * Below method will be used to validate and transform the main table plan to child table plan
   * rules for transforming is as below.
   * 1. Grouping expression rules
   *    1.1 Change the parent attribute reference for of group expression
   * to child attribute reference
   *
   * 2. Aggregate expression rules
   *    2.1 Change the parent attribute reference for of group expression to
   * child attribute reference
   *    2.2 Change the count AggregateExpression to Sum as count
   * is already calculated so in case of aggregate table
   * we need to apply sum to get the count
   *    2.2 In case of average aggregate function select 2 columns from aggregate table with
   * aggregation sum and count. Then add divide(sum(column with sum), sum(column with count)).
   * Note: During aggregate table creation for average table will be created with two columns
   * one for sum(column) and count(column) to support rollup
   * 3. Filter Expression rules.
   *    3.1 Updated filter expression attributes with child table attributes
   * 4. Update the Parent Logical relation with child Logical relation
   * 5. timeseries function
   *    5.1 validate parent table has timeseries datamap
   *    5.2 timeseries function is valid function or not
   * 6. Streaming
   * Rules for updating plan in case of streaming table:
   * In case of streaming data will be fetched from both fact and aggregate as aggregate table
   * will be updated only after each hand-off, so current streamed data won't be available on
   * aggregate table.
   * 6.1 Add one union node to add both fact and aggregate table plan to
   *     get the data from both table
   * 6.2 On top of Union Node add one Aggregate node to aggregate both table results
   * 6.3 In case of average(avg(column)) special handling is required for streaming
   *     7.3.1 Fact Plan will updated to return sum(column) and count(column) to do rollup
   *     7.3.2 Aggregate Plan will updated to return sum(column) and count(column) to do rollup
   * 6.4 In newly added Aggregate node all the aggregate expression must have same expression id as
   *     fact and fact plan will updated with new expression id. As query like order by this can be
   *     referred. In example1 sum(totalSalary) as totalSalary will have same expression id
   *     as in fact and fact plan sum(salary) will be updated with new expression id
   *
   * @param logicalPlan parent logical plan
   * @return transformed plan
   */
  def transformPreAggQueryPlan(logicalPlan: LogicalPlan): LogicalPlan = {
    var isPlanUpdated = false
    val updatedPlan = logicalPlan.transform {
      case agg@Aggregate(
        grExp,
        aggExp,
        CarbonSubqueryAlias(alias1, child@CarbonSubqueryAlias(alias2, l: LogicalRelation)))
        if l.relation.isInstanceOf[CarbonDatasourceHadoopRelation] &&
           l.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonRelation.
             metaData.hasAggregateDataMapSchema && !isPlanUpdated =>
        val carbonTable = getCarbonTable(l)
        if (isSpecificSegmentNotPresent(carbonTable)) {
          val list = scala.collection.mutable.HashSet.empty[QueryColumn]
          val aggregateExpressions = scala.collection.mutable.HashSet.empty[AggregateExpression]
          val isValidPlan = extractQueryColumnsFromAggExpression(
            grExp,
            aggExp,
            carbonTable,
            list,
            aggregateExpressions)
          if (isValidPlan) {
            val (aggDataMapSchema, childPlan) = getChildDataMapForTransformation(list,
              aggregateExpressions,
              carbonTable,
              agg)
            if (null != aggDataMapSchema && null != childPlan) {
              val attributes = childPlan.output.asInstanceOf[Seq[AttributeReference]]
              val (updatedGroupExp, updatedAggExp, newChild, None) =
                getUpdatedExpressions(grExp,
                  aggExp,
                  child,
                  None,
                  aggDataMapSchema,
                  attributes,
                  childPlan,
                  carbonTable,
                  agg)
              isPlanUpdated = true
              setExplain(aggDataMapSchema)
              val updateAggPlan =
                Aggregate(
                updatedGroupExp,
                updatedAggExp,
                CarbonReflectionUtils.getSubqueryAlias(
                  sparkSession,
                  Some(alias1),
                  CarbonReflectionUtils.getSubqueryAlias(
                    sparkSession,
                    Some(alias2),
                    newChild,
                    None),
                  None))
              getAggregateQueryPlan(
                updateAggPlan,
                grExp,
                aggExp,
                carbonTable,
                aggDataMapSchema,
                agg)
            } else {
              agg
            }
          } else {
            agg
          }
        } else {
          agg
        }
      // case for aggregation query
      case agg@Aggregate(
      grExp,
      aggExp,
      child@CarbonSubqueryAlias(alias, l: LogicalRelation))
        if l.relation.isInstanceOf[CarbonDatasourceHadoopRelation] &&
           l.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonRelation.
             metaData.hasAggregateDataMapSchema && !isPlanUpdated =>
        val carbonTable = getCarbonTable(l)
        if(isSpecificSegmentNotPresent(carbonTable)) {
          val list = scala.collection.mutable.HashSet.empty[QueryColumn]
          val aggregateExpressions = scala.collection.mutable.HashSet.empty[AggregateExpression]
          val isValidPlan = extractQueryColumnsFromAggExpression(
            grExp,
            aggExp,
            carbonTable,
            list,
            aggregateExpressions)
          if (isValidPlan) {
            val (aggDataMapSchema, childPlan) = getChildDataMapForTransformation(list,
              aggregateExpressions,
              carbonTable,
              agg)
            if (null != aggDataMapSchema && null != childPlan) {
              val attributes = childPlan.output.asInstanceOf[Seq[AttributeReference]]
              val (updatedGroupExp, updatedAggExp, newChild, None) =
                getUpdatedExpressions(grExp,
                  aggExp,
                  child,
                  None,
                  aggDataMapSchema,
                  attributes,
                  childPlan,
                  carbonTable,
                  agg)
              isPlanUpdated = true
              setExplain(aggDataMapSchema)
              val updateAggPlan =
                Aggregate(
                updatedGroupExp,
                updatedAggExp,
                CarbonReflectionUtils.getSubqueryAlias(
                  sparkSession,
                  Some(alias),
                  newChild,
                  None))
              getAggregateQueryPlan(
                updateAggPlan,
                grExp,
                aggExp,
                carbonTable,
                aggDataMapSchema,
                agg)
            } else {
              agg
            }
          } else {
            agg
          }
        } else {
          agg
        }
      // case of handling aggregation query with filter
      case agg@Aggregate(
      grExp,
      aggExp,
      Filter(expression, child@CarbonSubqueryAlias(alias, l: LogicalRelation)))
        if l.relation.isInstanceOf[CarbonDatasourceHadoopRelation] &&
           l.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonRelation.
             metaData.hasAggregateDataMapSchema && !isPlanUpdated =>
        val carbonTable = getCarbonTable(l)
        if(isSpecificSegmentNotPresent(carbonTable)) {
          val list = scala.collection.mutable.HashSet.empty[QueryColumn]
          val aggregateExpressions = scala.collection.mutable.HashSet.empty[AggregateExpression]
          var isValidPlan = extractQueryColumnsFromAggExpression(
            grExp,
            aggExp,
            carbonTable,
            list,
            aggregateExpressions)
          if (isValidPlan) {
            isValidPlan = !CarbonReflectionUtils.hasPredicateSubquery(expression)
          }
          // getting the columns from filter expression
          if (isValidPlan) {
            extractColumnFromExpression(expression, list, carbonTable, true)
          }
          if (isValidPlan) {
            val (aggDataMapSchema, childPlan) = getChildDataMapForTransformation(list,
              aggregateExpressions,
              carbonTable,
              agg)
            if (null != aggDataMapSchema && null != childPlan) {
              val attributes = childPlan.output.asInstanceOf[Seq[AttributeReference]]
              val (updatedGroupExp, updatedAggExp, newChild, updatedFilterExpression) =
                getUpdatedExpressions(grExp,
                  aggExp,
                  child,
                  Some(expression),
                  aggDataMapSchema,
                  attributes,
                  childPlan,
                  carbonTable,
                  agg)
              isPlanUpdated = true
              setExplain(aggDataMapSchema)
              val updateAggPlan =
                Aggregate(
                updatedGroupExp,
                updatedAggExp,
                Filter(
                  updatedFilterExpression.get,
                  CarbonReflectionUtils.getSubqueryAlias(
                    sparkSession,
                    Some(alias),
                    newChild,
                    None)))
              getAggregateQueryPlan(
                updateAggPlan,
                grExp,
                aggExp,
                carbonTable,
                aggDataMapSchema,
                agg)
            } else {
              agg
            }
          } else {
            agg
          }
        } else {
          agg
        }
      case agg@Aggregate(
      grExp,
      aggExp,
      Filter(
      expression,
      CarbonSubqueryAlias(alias1, child@CarbonSubqueryAlias(alias2, l: LogicalRelation))))
        if l.relation.isInstanceOf[CarbonDatasourceHadoopRelation] &&
           l.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonRelation.
             metaData.hasAggregateDataMapSchema && !isPlanUpdated =>
        val carbonTable = getCarbonTable(l)
        if(isSpecificSegmentNotPresent(carbonTable)) {
          val list = scala.collection.mutable.HashSet.empty[QueryColumn]
          val aggregateExpressions = scala.collection.mutable.HashSet.empty[AggregateExpression]
          var isValidPlan = extractQueryColumnsFromAggExpression(
            grExp,
            aggExp,
            carbonTable,
            list,
            aggregateExpressions)
          if (isValidPlan) {
            isValidPlan = !CarbonReflectionUtils.hasPredicateSubquery(expression)
          }
          // getting the columns from filter expression
          if (isValidPlan) {
            extractColumnFromExpression(expression, list, carbonTable, true)
          }
          if (isValidPlan) {
            val (aggDataMapSchema, childPlan) = getChildDataMapForTransformation(list,
              aggregateExpressions,
              carbonTable,
              agg)
            if (null != aggDataMapSchema && null != childPlan) {
              val attributes = childPlan.output.asInstanceOf[Seq[AttributeReference]]
              val (updatedGroupExp, updatedAggExp, newChild, updatedFilterExpression) =
                getUpdatedExpressions(grExp,
                  aggExp,
                  child,
                  Some(expression),
                  aggDataMapSchema,
                  attributes,
                  childPlan,
                  carbonTable,
                  agg)
              isPlanUpdated = true
              setExplain(aggDataMapSchema)
              val updateAggPlan =
                Aggregate(
                  updatedGroupExp,
                  updatedAggExp,
                  Filter(
                    updatedFilterExpression.get,
                    CarbonReflectionUtils.getSubqueryAlias(
                      sparkSession,
                      Some(alias1),
                      CarbonReflectionUtils.getSubqueryAlias(
                        sparkSession,
                        Some(alias2),
                        newChild,
                        None),
                      None)))
              getAggregateQueryPlan(
                updateAggPlan,
                grExp,
                aggExp,
                carbonTable,
                aggDataMapSchema,
                agg)
            } else {
              agg
            }
          } else {
            agg
          }
        } else {
          agg
        }

    }
    if(isPlanUpdated) {
      CarbonSession.threadSet(CarbonCommonConstants.SUPPORT_DIRECT_QUERY_ON_DATAMAP, "true")
    }
    updatedPlan
  }

  // set datamap match information for EXPLAIN command
  private def setExplain(dataMapSchema: AggregationDataMapSchema): Unit = {
    ExplainCollector.recordMatchedOlapDataMap(
      dataMapSchema.getProvider.getShortName, dataMapSchema.getDataMapName)
  }

  /**
   * Method to get the aggregate query plan
   * @param aggPlan
   * aggregate table query plan
   * @param grExp
   * fact group by expression
   * @param aggExp
   * fact aggregate expression
   * @param carbonTable
   * fact table
   * @param aggregationDataMapSchema
   * selected aggregation data map
   * @param factAggPlan
   * fact aggregate query plan
   * @return updated plan
   */
  def getAggregateQueryPlan(aggPlan: LogicalPlan,
      grExp: Seq[Expression],
      aggExp: Seq[NamedExpression],
      carbonTable: CarbonTable,
      aggregationDataMapSchema: DataMapSchema,
      factAggPlan: LogicalPlan): LogicalPlan = {
    // to handle streaming table with pre aggregate
    if (carbonTable.isStreamingSink) {
      setSegmentsForStreaming(carbonTable, aggregationDataMapSchema)
      // get new fact expression
      val factExp = updateFactTablePlanForStreaming(factAggPlan)
      // get new Aggregate node expression
      val aggPlanNew = updateAggTablePlanForStreaming(aggPlan)
      val streamingNodeExp = getExpressionsForStreaming(aggExp)
      // clear the expression as in case of streaming it is not required
      updatedExpression.clear
      // Add Aggregate node to aggregate data from fact and aggregate
      Aggregate(
        createNewAggGroupBy(grExp, factAggPlan),
        streamingNodeExp.asInstanceOf[Seq[NamedExpression]],
        // add union node to get the result from both
        Union(
          factExp,
      aggPlanNew))
    } else {
      aggPlan
    }
  }

  /**
   * create group by expression for newly Added Aggregate node
   * @param grExp fact group by expression
   * @param plan fact query plan
   * @return group by expression
   */
  private def createNewAggGroupBy(grExp: Seq[Expression], plan: LogicalPlan): Seq[Expression] = {
    grExp.map {
      case attr: AttributeReference =>
        val aggModel = AggExpToColumnMappingModel(
          removeQualifiers(PreAggregateUtil.normalizeExprId(attr, plan.allAttributes)))
        if (factPlanGrpExpForStreaming.get(aggModel).isDefined) {
          factPlanGrpExpForStreaming.get(aggModel).get
        } else {
          attr
        }
      case exp: Expression =>
        val aggModel = AggExpToColumnMappingModel(
          removeQualifiers(PreAggregateUtil.normalizeExprId(exp, plan.allAttributes)))
        factPlanGrpExpForStreaming.get(aggModel).get
    }
  }
  /**
   * Method to set the segments when query is fired on streaming table with pre aggregate
   * Adding a property streaming_seg so while removing from session params we can differentiate
   * it was set from CarbonPreAggregateRules
   * @param parentTable
   * parent arbon table
   * @param dataMapSchema
   * child datamap schema
   */
  def setSegmentsForStreaming(parentTable: CarbonTable, dataMapSchema: DataMapSchema): Unit = {
    val mainTableKey = parentTable.getDatabaseName + '.' + parentTable.getTableName
    val factManager = new SegmentStatusManager(parentTable.getAbsoluteTableIdentifier)
    CarbonSession
      .threadSet(CarbonCommonConstantsInternal.QUERY_ON_PRE_AGG_STREAMING + mainTableKey, "true")
    CarbonSession
      .threadSet(
        CarbonCommonConstants.CARBON_INPUT_SEGMENTS + mainTableKey,
        factManager.getValidAndInvalidSegments.getValidSegments.asScala.mkString(","))
    CarbonSession
      .threadSet(CarbonCommonConstants.VALIDATE_CARBON_INPUT_SEGMENTS + mainTableKey, "true")
    // below code is for aggregate table
    val identifier = TableIdentifier(
      dataMapSchema.getChildSchema.getTableName,
      Some(parentTable.getDatabaseName))
    val catalog = CarbonEnv.getInstance(sparkSession).carbonMetaStore
    val carbonRelation =
      catalog.lookupRelation(identifier)(sparkSession).asInstanceOf[CarbonRelation]
    val segmentStatusManager = new SegmentStatusManager(carbonRelation.carbonTable
      .getAbsoluteTableIdentifier)
    val validSegments = segmentStatusManager.getValidAndInvalidSegments.getValidSegments.asScala
      .mkString(",")
    val childTableKey = carbonRelation.carbonTable.getDatabaseName + '.' +
                   carbonRelation.carbonTable.getTableName
    CarbonSession
      .threadSet(CarbonCommonConstantsInternal.QUERY_ON_PRE_AGG_STREAMING + childTableKey, "true")
    CarbonSession
      .threadSet(CarbonCommonConstants.CARBON_INPUT_SEGMENTS + childTableKey, validSegments)
    CarbonSession
      .threadSet(CarbonCommonConstants.VALIDATE_CARBON_INPUT_SEGMENTS + childTableKey, "false")
  }

  /**
   * Map to keep expression name to its alias mapping. This will be used while adding a node when
   * plan for streaming table is updated.
   * Note: In case of average fact table plan will have two alias as sum(column) and count(column)
   * to support rollup
   */
  private val factPlanExpForStreaming = mutable.HashMap[String, Seq[NamedExpression]]()

  private val factPlanGrpExpForStreaming = mutable
    .HashMap[AggExpToColumnMappingModel, AttributeReference]()

  /**
   * Below method will be used to get the expression for Aggregate node added for streaming
   * Expression id will be same as fact plan as it can be referred in query
   *
   * @param aggExp
   * main table aggregate expression
   * @return updated aggregate expression
   */
  private def getExpressionsForStreaming(aggExp: Seq[Expression]): Seq[Expression] = {
    val updatedExp = aggExp map {
      case attr: AttributeReference =>
        attr
      case alias@Alias(aggExp: AggregateExpression, name) =>
        // in case of aggregate expression get the fact alias based on expression name
        val factAlias = factPlanExpForStreaming(name)
        // create attribute reference object for each expression
        val attrs = factAlias.map { factAlias =>
          CarbonToSparkAdapter.createAttributeReference(
            name,
            alias.dataType,
            alias.nullable,
            Metadata.empty,
            factAlias.exprId,
            alias.qualifier,
            alias)
        }
        // add aggregate function in Aggregate node added for handling streaming
        // to aggregate results from fact and aggregate table
        val updatedAggExp = getAggregateExpressionForAggregation(aggExp, attrs)
        // same reference id will be used as it can be used by above nodes in the plan like
        // sort, project, join
        CarbonToSparkAdapter.createAliasRef(
          updatedAggExp.head,
          name,
          alias.exprId,
          alias.qualifier,
          Option(alias.metadata),
          Some(alias))
      case alias@Alias(expression, name) =>
        CarbonToSparkAdapter.createAttributeReference(
          name,
          alias.dataType,
          alias.nullable,
          Metadata.empty,
          alias.exprId,
          alias.qualifier,
          alias)
    }
    updatedExp
  }

  /**
   * Below method will be used to update the fact plan in case of streaming table
   * This is required to handle average aggregte function as in case of average we need to return
   * two columns data sum(column) and count(column) to get the correct result
   *
   * @param logicalPlan
   * fact table Aggregate plan
   * @return updated aggregate plan for fact
   */
  private def updateFactTablePlanForStreaming(logicalPlan: LogicalPlan) : LogicalPlan = {
    // only aggregate expression needs to be updated
    logicalPlan.transform{
      case agg@Aggregate(grpExp, aggExp, _) =>
        agg
          .copy(aggregateExpressions = updateAggExpInFactForStreaming(aggExp, grpExp, agg)
            .asInstanceOf[Seq[NamedExpression]])
    }
  }

  /**
   * Below method will be used to update the aggregate table plan for streaming
   * @param logicalPlan
   * aggergate table logical plan
   * @return updated logical plan
   */
  private def updateAggTablePlanForStreaming(logicalPlan: LogicalPlan) : LogicalPlan = {
    // only aggregate expression needs to be updated
    logicalPlan.transform{
      case agg@Aggregate(grpExp, aggExp, _) =>
        agg
          .copy(aggregateExpressions = updateAggExpInAggForStreaming(aggExp, grpExp, agg)
            .asInstanceOf[Seq[NamedExpression]])
    }
  }

  /**
   * Below method will be used to update the aggregate plan for streaming
   * @param namedExp
   * aggregate expression
   * @param grpExp
   * group by expression
   * @param plan
   * aggregate query plan
   * @return updated aggregate expression
   */
  private def updateAggExpInAggForStreaming(namedExp : Seq[NamedExpression],
      grpExp: Seq[Expression], plan: LogicalPlan) : Seq[Expression] = {
    // removing alias from expression to compare with grouping expression
    // as in case of alias all the projection column will be updated with alias
    val updatedExp = namedExp.map {
      case Alias(attr: AttributeReference, name) =>
        attr
      case exp: Expression =>
        exp
    }
    addGrpExpToAggExp(grpExp, updatedExp, plan)
  }

  /**
   * below method will be used to updated the aggregate expression with missing
   * group by expression, when only aggregate expression is selected in query
   *
   * @param grpExp
   * group by expressions
   * @param aggExp
   * aggregate expressions
   * @param plan
   * logical plan
   * @return updated aggregate expression
   */
  private def addGrpExpToAggExp(grpExp: Seq[Expression],
      aggExp: Seq[Expression],
      plan: LogicalPlan): Seq[Expression] = {
    // set to add all the current aggregate expression
    val expressions = mutable.LinkedHashSet.empty[AggExpToColumnMappingModel]
    aggExp.foreach {
      case Alias(exp, _) =>
      expressions +=
      AggExpToColumnMappingModel(
        PreAggregateUtil.normalizeExprId(exp, plan.allAttributes), None)
      case attr: AttributeReference =>
        expressions +=
        AggExpToColumnMappingModel(
          PreAggregateUtil.normalizeExprId(attr, plan.allAttributes), None)
    }
    val newAggExp = new ArrayBuffer[Expression]
    newAggExp ++= aggExp
    // for each group by expression check if already present in set if it is present
    // then no need to add otherwise add
    var counter = 0
    grpExp.foreach{gExp =>
      val normalizedExp = AggExpToColumnMappingModel(
        PreAggregateUtil.normalizeExprId(gExp, plan.allAttributes), None)
      if(!expressions.contains(normalizedExp)) {
        gExp match {
          case attr: AttributeReference =>
            newAggExp += attr
          case exp: Expression =>
            newAggExp += CarbonToSparkAdapter.createAliasRef(
              exp,
              "dummy_" + counter,
              NamedExpression.newExprId)
            counter = counter + 1
        }
      }
    }
    newAggExp
  }
  /**
   * Below method will be used to update the aggregate expression for streaming fact table plan
   * @param namedExp
   * streaming Fact plan aggregate expression
   * @return
   * Updated streaming fact plan aggregate expression
   */
  private def updateAggExpInFactForStreaming(namedExp : Seq[NamedExpression],
  grpExp: Seq[Expression], plan: LogicalPlan) : Seq[Expression] = {
    val addedExp = addGrpExpToAggExp(grpExp, namedExp, plan)
    val updatedExp = addedExp.flatMap {
      case attr: AttributeReference =>
        Seq(attr)
      case alias@Alias(aggExp: AggregateExpression, name) =>
        // get the new aggregate expression
        val newAggExp = getAggFunctionForFactStreaming(aggExp)
        val updatedExp = newAggExp.map { exp =>
          CarbonToSparkAdapter.createAliasRef(exp,
            name,
            NamedExpression.newExprId,
            alias.qualifier,
            Some(alias.metadata),
            Some(alias))
        }
        // adding to map which will be used while Adding an Aggregate node for handling streaming
        // table plan change
        factPlanExpForStreaming.put(name, updatedExp)
        updatedExp
      case alias@Alias(exp: Expression, name) =>
        val newAlias = Seq(alias)
        val attr = CarbonToSparkAdapter.createAttributeReference(name,
          alias.dataType,
          alias.nullable,
          alias.metadata,
          alias.exprId,
          alias.qualifier,
          alias)
        factPlanGrpExpForStreaming.put(
          AggExpToColumnMappingModel(
            removeQualifiers(PreAggregateUtil.normalizeExprId(exp, plan.allAttributes))),
            attr)
        factPlanExpForStreaming.put(name, newAlias)
        newAlias
    }
    updatedExp
  }
  /**
   * Below method will be used to update the fact table query aggregate function expression
   * Rules for updating the expression.
   * In case of average return sum(expression), count(expression) to get the correct result
   * @param aggExp
   * actual query aggregate expression
   * @return seq of expression as in case of average we need to return two sum and count
   *
   */
  def getAggFunctionForFactStreaming(aggExp: AggregateExpression): Seq[Expression] = {
    aggExp.aggregateFunction match {
      case Average(MatchCastExpression(exp: Expression, changeDataType: DataType)) =>
        val newExp = Seq(AggregateExpression(Sum(Cast(exp, DoubleType)),
          aggExp.mode,
          isDistinct = false),
          Cast(AggregateExpression(Count(exp), aggExp.mode, false), DoubleType))
        newExp
      case Average(exp: Expression) =>
        val dataType =
          if (exp.dataType.isInstanceOf[DecimalType]) {
            // decimal must not go as double precision.
            exp.dataType.asInstanceOf[DecimalType]
          } else {
            DoubleType
          }
        val newExp = Seq(AggregateExpression(Sum(Cast(exp, dataType)), aggExp.mode, false),
          Cast(AggregateExpression(Count(exp), aggExp.mode, false), dataType))
        newExp
      case _ =>
        val newExp = Seq(aggExp)
        newExp
    }
  }

  /**
   * Below method will be used to validate query plan and get the proper aggregation data map schema
   * and child relation plan object if plan is valid for transformation
   * @param queryColumns list of query columns from projection and filter
   * @param aggregateExpressions list of aggregate expression (aggregate function)
   * @param carbonTable parent carbon table
   * @param parentLogicalPlan parent logical relation
   * @return if plan is valid then aggregation data map schema and its relation plan
   */
  def getChildDataMapForTransformation(queryColumns: scala.collection.mutable.HashSet[QueryColumn],
      aggregateExpressions: scala.collection.mutable.HashSet[AggregateExpression],
      carbonTable: CarbonTable,
      parentLogicalPlan: LogicalPlan): (AggregationDataMapSchema, LogicalPlan) = {
    // getting all the projection columns
    val listProjectionColumn = queryColumns
      .filter(queryColumn => !queryColumn.isFilterColumn)
      .toList
    // getting all the filter columns
    val listFilterColumn = queryColumns
      .filter(queryColumn => queryColumn.isFilterColumn)
      .toList
    val isProjectionColumnPresent = (listProjectionColumn.size + listFilterColumn.size) > 0
    // create a query plan object which will be used to select the list of pre aggregate tables
    // matches with this plan
    val queryPlan = new AggregateQueryPlan(listProjectionColumn.asJava, listFilterColumn.asJava)
    // create aggregate table selector object
    val aggregateTableSelector = new AggregateTableSelector(queryPlan, carbonTable)
    // select the list of valid child tables
    val selectedDataMapSchemas = aggregateTableSelector.selectPreAggDataMapSchema()
    // query has only aggregate expression then selected data map will be empty
    // the validate all the child data map otherwise validate selected data map
    var selectedAggMaps = if (isProjectionColumnPresent) {
      selectedDataMapSchemas
    } else {
      carbonTable.getTableInfo.getDataMapSchemaList
    }
    // if it does not match with any pre aggregate table return the same plan
    if (!selectedAggMaps.isEmpty) {
      // filter the selected child schema based on size to select the pre-aggregate tables
      // that are enabled
      val catalog = CarbonEnv.getInstance(sparkSession).carbonMetaStore
      val relationBuffer = selectedAggMaps.asScala.map { selectedDataMapSchema =>
        val identifier = TableIdentifier(
          selectedDataMapSchema.getRelationIdentifier.getTableName,
          Some(selectedDataMapSchema.getRelationIdentifier.getDatabaseName))
        val carbonRelation =
          catalog.lookupRelation(identifier)(sparkSession).asInstanceOf[CarbonRelation]
        val relation = sparkSession.sessionState.catalog.lookupRelation(identifier)
        (selectedDataMapSchema, carbonRelation, relation)
      }.filter(_._2.sizeInBytes != 0L).sortBy(_._2.sizeInBytes)
      if (relationBuffer.isEmpty) {
        // If the size of relation Buffer is 0 then it means that none of the pre-aggregate
        // tables have data yet.
        // In this case we would return the original plan so that the query hits the parent
        // table.
        (null, null)
      } else {
        // if query does not have any aggregate function no need to validate the same
        val tuple = if (aggregateExpressions.nonEmpty && !selectedAggMaps.isEmpty) {
          relationBuffer.collectFirst {
            case a@(datamapSchema, _, _)
              if validateAggregateExpression(datamapSchema,
                carbonTable,
                parentLogicalPlan,
                aggregateExpressions.toSeq) =>
              a
          }
        } else {
          Some(relationBuffer.head)
        }
        tuple match {
          case Some((dataMapSchema, _, logicalPlan)) => (dataMapSchema
            .asInstanceOf[AggregationDataMapSchema], new FindDataSourceTable(sparkSession)
            .apply(logicalPlan))
          case None => (null, null)
        }
        // If the relationBuffer is enabled then find the table with the minimum size.
      }
    } else {
      (null, null)
    }
  }

  /**
   * Below method will be used to validate aggregate expression with the data map
   * and will return the selected valid data maps
   * @param selectedDataMap list of data maps
   * @param carbonTable parent carbon table
   * @param parentLogicalPlan parent logical plan
   * @param queryAggExpLogicalPlans query agg expression logical plan
   * @return valid data map
   */
  def validateAggregateExpression(selectedDataMap: DataMapSchema,
      carbonTable: CarbonTable,
      parentLogicalPlan: LogicalPlan,
      queryAggExpLogicalPlans: Seq[AggregateExpression]): Boolean = {
    val mappingModel = getExpressionToColumnMapping(selectedDataMap,
      carbonTable,
      parentLogicalPlan)
    queryAggExpLogicalPlans.forall{p =>
      mappingModel.exists{m =>
        matchExpression(
          PreAggregateUtil.normalizeExprId(p, parentLogicalPlan.allAttributes),
          m.expression)}
    }
  }

  /**
   * Below method will be used to update the expression
   * It will remove the qualifiers
   * @param expression
   * expression
   * @return updated expressions
   */
  private def removeQualifiers(expression: Expression) : Expression = {
    expression.transform {
      case attr: AttributeReference =>
        CarbonToSparkAdapter.createAttributeReference(
          attr.name,
          attr.dataType,
          attr.nullable,
          attr.metadata,
          attr.exprId,
          None,
          attr)
    }
  }

  /**
   * Below method will be used to match two expressions
   * @param firstExp
   * first expression
   * @param secondExp
   * second expressios
   * @return is similare
   */
  private def matchExpression(firstExp: Expression, secondExp: Expression) : Boolean = {
    val first = removeQualifiers(firstExp)
    val second = removeQualifiers(secondExp)
    first == second
  }

  /**
   * Below method will be used to to get the logical plan for each aggregate expression in
   * child data map and its column schema mapping if mapping is already present
   * then it will use the same otherwise it will generate and stored in aggregation data map
   * @param selectedDataMap child data map
   * @param carbonTable parent table
   * @param parentLogicalPlan logical relation of actual plan
   * @return map of logical plan for each aggregate expression in child query and its column mapping
   */
  def getExpressionToColumnMapping(selectedDataMap: DataMapSchema,
      carbonTable: CarbonTable,
      parentLogicalPlan: LogicalPlan): mutable.Set[AggExpToColumnMappingModel] = {
    val aggDataMapSchema = selectedDataMap.asInstanceOf[AggregationDataMapSchema]
    if(null == aggDataMapSchema.getAggExpToColumnMapping) {
      // add preAGG UDF to avoid all the PreAggregate rule
      val childDataMapQueryString = parser.addPreAggFunction(
        PreAggregateUtil.getChildQuery(aggDataMapSchema))
      // get the logical plan
      val aggPlan = sparkSession.sql(childDataMapQueryString).logicalPlan
      // getting all aggregate expression from query
      val dataMapAggExp = getAggregateExpFromChildDataMap(aggPlan)
      // in case of average child table will have two columns which will be stored in sequence
      // so for average expression we need to get two columns for mapping
      var counter = 0
      // sorting the columns based on schema ordinal so search will give proper result
      val sortedColumnList = aggDataMapSchema.getChildSchema.getListOfColumns.asScala
        .sortBy(_.getSchemaOrdinal)
      val expressionToColumnMapping = mutable.LinkedHashSet.empty[AggExpToColumnMappingModel]
      dataMapAggExp.foreach { aggExp =>
        val updatedExp = PreAggregateUtil.normalizeExprId(aggExp, aggPlan.allAttributes)
        val model = AggExpToColumnMappingModel(updatedExp, None)
        if (!expressionToColumnMapping.contains(model)) {
          // check if aggregate expression is of type avg
          // get the columns
          val columnSchema = aggDataMapSchema
            .getAggColumnBasedOnIndex(counter, sortedColumnList.asJava)
          // increment the counter so when for next expression above code will be
          // executed it will search from that schema ordinal
          counter = columnSchema.getSchemaOrdinal + 1
          model.columnSchema = Some(columnSchema)
          expressionToColumnMapping += model
        }
      }
      aggDataMapSchema.setAggExpToColumnMapping(expressionToColumnMapping.asJava)
      // return the mapping
      expressionToColumnMapping
    } else {
      aggDataMapSchema.getAggExpToColumnMapping
        .asInstanceOf[java.util.Set[AggExpToColumnMappingModel]].asScala
        .asInstanceOf[mutable.LinkedHashSet[AggExpToColumnMappingModel]]
    }
  }

  /**
   * Below method will be used to get aggregate expression
   * @param logicalPlan logical plan
   * @return list of aggregate expression
   */
  def getAggregateExpFromChildDataMap(logicalPlan: LogicalPlan): Seq[AggregateExpression] = {
    val list = scala.collection.mutable.ListBuffer.empty[AggregateExpression]
    logicalPlan match {
      case _@Aggregate(_, aggExp, _) =>
        aggExp map {
          case Alias(attr: AggregateExpression, _) =>
            list ++= PreAggregateUtil.validateAggregateFunctionAndGetFields(attr)
          case _ =>
        }
    }
    list
  }

  /**
   * Below method will be used to check whether specific segment is set for maintable
   * if it is present then no need to transform the plan and query will be executed on
   * maintable
   * @param carbonTable parent table
   * @return is specific segment is present in session params
   */
  def isSpecificSegmentNotPresent(carbonTable: CarbonTable) : Boolean = {
    val carbonSessionInfo = ThreadLocalSessionInfo.getCarbonSessionInfo
    if (carbonSessionInfo != null) {
      carbonSessionInfo.getSessionParams
        .getProperty(CarbonCommonConstants.CARBON_INPUT_SEGMENTS +
                     carbonTable.getAbsoluteTableIdentifier.getCarbonTableIdentifier
                       .getDatabaseName + "." + carbonTable.getTableName, "").isEmpty
    } else {
      true
    }
  }

  /**
   * Below method will be used to extract the query columns from
   * filter expression
   * @param expression filter expression
   * @param queryColumns query column set
   * @param carbonTable parent table
   * @return isvalid filter expression for aggregate
   */
  def extractColumnFromExpression(expression: Expression,
      queryColumns: scala.collection.mutable.HashSet[QueryColumn],
      carbonTable: CarbonTable,
      isFilterColumn: Boolean = false) {
    // map to maintain attribute reference present in the filter to timeseries function
    // if applied this is added to avoid duplicate column
    val mapOfColumnSeriesFun = scala.collection.mutable.HashMap.empty[AttributeReference, String]
    expression.transform {
      case attr: AttributeReference =>
        if (mapOfColumnSeriesFun.get(attr).isEmpty) {
          mapOfColumnSeriesFun.put(attr, null)
        }
        attr
      case udf@CarbonScalaUDF(_) =>
        // for handling timeseries function
        if (udf.asInstanceOf[ScalaUDF].function.getClass.getName.equalsIgnoreCase(
          "org.apache.spark.sql.execution.command.timeseries.TimeseriesFunction") &&
            CarbonUtil.hasTimeSeriesDataMap(carbonTable)) {
          mapOfColumnSeriesFun.put(udf.children.head.asInstanceOf[AttributeReference],
            udf.children.last.asInstanceOf[Literal].value.toString)
        } else {
          // for any other scala udf
          udf.transform {
            case attr: AttributeReference =>
              if (mapOfColumnSeriesFun.get(attr).isEmpty) {
                mapOfColumnSeriesFun.put(attr, null)
              }
              attr
          }
        }
        udf
    }
    mapOfColumnSeriesFun.foreach { f =>
      if (f._2 == null) {
        queryColumns +=
        getQueryColumn(f._1.name, carbonTable, isFilterColumn)
      } else {
        queryColumns += getQueryColumn(f._1.name,
          carbonTable,
          isFilterColumn,
          timeseriesFunction = f._2)
      }
    }
  }

  /**
   * Below method will be used to get the child attribute reference
   * based on parent name
   *
   * @param dataMapSchema child schema
   * @param attributeReference parent attribute reference
   * @param attributes child logical relation
   * @param canBeNull this is added for strict validation in which case child attribute can be
   * null and when it cannot be null
   * @return child attribute reference
   */
  def getChildAttributeReference(dataMapSchema: DataMapSchema,
      attributeReference: AttributeReference,
      attributes: Seq[AttributeReference],
      canBeNull: Boolean = false,
      timeseriesFunction: String = ""): AttributeReference = {
    val aggregationDataMapSchema = dataMapSchema.asInstanceOf[AggregationDataMapSchema]
    val columnSchema = if (timeseriesFunction.isEmpty) {
      aggregationDataMapSchema.getChildColByParentColName(attributeReference.name.toLowerCase)
    } else {
      aggregationDataMapSchema
        .getTimeseriesChildColByParent(attributeReference.name.toLowerCase,
          timeseriesFunction)
    }
    // here column schema cannot be null, if it is null then aggregate table selection
    // logic has some problem
    if (!canBeNull && null == columnSchema) {
      throw new AnalysisException("Column does not exists in Pre Aggregate table")
    }
    if(null == columnSchema && canBeNull) {
      null
    } else {
      // finding the child attribute from child logical relation
      attributes.find(p => p.name.equals(columnSchema.getColumnName)).get
    }
  }

  /**
   * Below method will be used to get the updated expression for pre aggregated table.
   * It will replace the attribute of actual plan with child table attributes.
   * Updation will be done for below expression.
   * 1. Grouping expression
   * 2. aggregate expression
   * 3. child logical plan
   * 4. filter expression if present
   *
   * @param groupingExpressions actual plan grouping expression
   * @param aggregateExpressions actual plan aggregate expression
   * @param child child logical plan
   * @param filterExpression filter expression
   * @param aggDataMapSchema pre aggregate table schema
   * @param attributes pre aggregate table logical relation
   * @param aggPlan aggregate logical plan
   * @return tuple of(updated grouping expression,
   * updated aggregate expression,
   * updated child logical plan,
   * updated filter expression if present in actual plan)
   */
  def getUpdatedExpressions(groupingExpressions: Seq[Expression],
      aggregateExpressions: Seq[NamedExpression],
      child: LogicalPlan, filterExpression: Option[Expression] = None,
      aggDataMapSchema: AggregationDataMapSchema,
      attributes: Seq[AttributeReference],
      aggPlan: LogicalPlan,
      parentTable: CarbonTable,
      parentLogicalPlan: LogicalPlan): (Seq[Expression], Seq[NamedExpression], LogicalPlan,
    Option[Expression]) = {
    val aggExpColumnMapping = if (null != aggDataMapSchema.getAggExpToColumnMapping) {
      Some(aggDataMapSchema.getAggExpToColumnMapping
        .asInstanceOf[java.util.Set[AggExpToColumnMappingModel]].asScala
        .asInstanceOf[mutable.LinkedHashSet[AggExpToColumnMappingModel]])
    } else {
      None
    }

    // transforming the group by expression attributes with child attributes
    val updatedGroupExp = groupingExpressions.map { exp =>
      exp.transform {
        case attr: AttributeReference =>
          val childAttr = getChildAttributeReference(aggDataMapSchema, attr, attributes)
          childAttr
      }
    }
    // below code is for updating the aggregate expression.
    // Note: In case of aggregate expression updation we need to return alias as
    //       while showing the final result we need to show based on actual query
    //       for example: If query is "select name from table group by name"
    //       if we only update the attributes it will show child table column name in final output
    //       so for handling this if attributes does not have alias we need to return alias of
    // parent
    //       table column name
    // Rules for updating aggregate expression.
    // 1. If it matches with attribute reference return alias of child attribute reference
    // 2. If it matches with alias return same alias with child attribute reference
    // 3. If it matches with alias of any supported aggregate function return aggregate function
    // with child attribute reference. Please check class level documentation how when aggregate
    // function will be updated

    val updatedAggExp = aggregateExpressions.flatMap {
      // case for attribute reference
      case attr: AttributeReference =>
        val childAttr = getChildAttributeReference(aggDataMapSchema,
          attr,
          attributes)
        val newExpressionId = NamedExpression.newExprId
        val childTableAttr = CarbonToSparkAdapter.createAttributeReference(attr.name,
          childAttr.dataType,
          childAttr.nullable,
          childAttr.metadata,
          newExpressionId,
          childAttr.qualifier,
          attr)
        updatedExpression.put(attr, childTableAttr)
        // returning the alias to show proper column name in output
        Seq(Alias(childAttr,
          attr.name)(newExpressionId,
          childAttr.qualifier).asInstanceOf[NamedExpression])
      // case for alias
      case alias@Alias(attr: AttributeReference, name) =>
        val childAttr = getChildAttributeReference(aggDataMapSchema,
          attr,
          attributes)
        val newExpressionId = NamedExpression.newExprId
        val parentTableAttr = CarbonToSparkAdapter.createAttributeReference(name,
          alias.dataType,
          alias.nullable,
          Metadata.empty,
          alias.exprId,
          alias.qualifier,
          alias)
        val childTableAttr = CarbonToSparkAdapter.createAttributeReference(name,
          alias.dataType,
          alias.nullable,
          Metadata.empty,
          newExpressionId,
          alias.qualifier,
          alias)
        updatedExpression.put(parentTableAttr, childTableAttr)
        // returning alias with child attribute reference
        Seq(Alias(childAttr,
          name)(newExpressionId,
          childAttr.qualifier).asInstanceOf[NamedExpression])
      // for aggregate function case
      case alias@Alias(attr: AggregateExpression, name) =>
        // get the updated aggregate aggregate function
        val aggExp = if (aggExpColumnMapping.isDefined) {
          getUpdatedAggregateExpressionForChild(attr,
            aggDataMapSchema,
            attributes,
            parentTable,
            parentLogicalPlan,
            aggExpColumnMapping.get,
            parentTable.isStreamingSink)
        } else {
          Seq(attr)
        }
        if(!parentTable.isStreamingSink) {
          // for normal table
          // generate new expression id for child
          val newExpressionId = NamedExpression.newExprId
          // create a parent attribute reference which will be replced on node which may be referred
          // by node like sort join
          val parentTableAttr = CarbonToSparkAdapter.createAttributeReference(name,
            alias.dataType,
            alias.nullable,
            Metadata.empty,
            alias.exprId,
            alias.qualifier,
            alias)
          // creating a child attribute reference which will be replced
          val childTableAttr = CarbonToSparkAdapter.createAttributeReference(name,
            alias.dataType,
            alias.nullable,
            Metadata.empty,
            newExpressionId,
            alias.qualifier,
            alias)
          // adding to map, will be used during other node updation like sort, join, project
          updatedExpression.put(parentTableAttr, childTableAttr)
          // returning alias with child attribute reference
          Seq(Alias(aggExp.head,
            name)(newExpressionId,
            alias.qualifier).asInstanceOf[NamedExpression])
        } else {
          // for streaming table
          // create alias for aggregate table
          val aggExpForStreaming = aggExp.map{ exp =>
            CarbonToSparkAdapter.createAliasRef(exp,
              name,
              NamedExpression.newExprId,
              alias.qualifier,
              Some(alias.metadata),
              Some(alias)).asInstanceOf[NamedExpression]
          }
          aggExpForStreaming
        }
      case alias@Alias(expression: Expression, name) =>
        val updatedExp =
          if (expression.isInstanceOf[ScalaUDF] &&
              expression.asInstanceOf[ScalaUDF].function.getClass.getName.equalsIgnoreCase(
                "org.apache.spark.sql.execution.command.timeseries.TimeseriesFunction")) {
            expression.asInstanceOf[ScalaUDF].transform {
              case attr: AttributeReference =>
                val childAttributeReference = getChildAttributeReference(aggDataMapSchema,
                  attr,
                  attributes,
                  timeseriesFunction =
                    expression.asInstanceOf[ScalaUDF].children(1).asInstanceOf[Literal].value
                      .toString)
                childAttributeReference
            }
          } else {
            expression.transform{
              case attr: AttributeReference =>
                val childAttributeReference = getChildAttributeReference(aggDataMapSchema,
                  attr,
                  attributes)
                childAttributeReference
            }
          }
        val newExpressionId = NamedExpression.newExprId
        val parentTableAttr = CarbonToSparkAdapter.createAttributeReference(name,
          alias.dataType,
          alias.nullable,
          Metadata.empty,
          alias.exprId,
          alias.qualifier,
          alias)
        val childTableAttr = CarbonToSparkAdapter.createAttributeReference(name,
          alias.dataType,
          alias.nullable,
          Metadata.empty,
          newExpressionId,
          alias.qualifier,
          alias)
        updatedExpression.put(parentTableAttr, childTableAttr)
        Seq(Alias(updatedExp, name)(newExpressionId,
          alias.qualifier).asInstanceOf[NamedExpression])
    }
    // transforming the logical relation
    val newChild = child.transform {
      case _: LogicalRelation =>
        aggPlan
      case _: SubqueryAlias =>
        aggPlan match {
          case s: SubqueryAlias => s.child
          case others => others
        }
    }
    // updating the filter expression if present
    val updatedFilterExpression = if (filterExpression.isDefined) {
      val filterExp = filterExpression.get
      Some(filterExp.transform {
        case attr: AttributeReference =>
          getChildAttributeReference(aggDataMapSchema, attr, attributes)
      })
    } else {
      None
    }
    (updatedGroupExp, updatedAggExp, newChild, updatedFilterExpression)
  }

  /**
   * Below method will be used to get the aggregate expression based on match
   * Aggregate expression updation rules
   * 1 Change the count AggregateExpression to Sum as count
   * is already calculated so in case of aggregate table
   * we need to apply sum to get the count
   * 2 In case of average aggregate function select 2 columns from aggregate table
   * with aggregation sum and count.
   * Then add divide(sum(column with sum), sum(column with count)).
   * Note: During aggregate table creation for average aggregation function
   * table will be created with two columns one for sum(column) and count(column)
   * to support rollup
   *
   * @param aggExp aggregate expression
   * @param dataMapSchema child data map schema
   * @param attributes child logical relation
   * @param parentTable parent carbon table
   * @param parentLogicalPlan logical relation
   * @return updated expression
   */
  def getUpdatedAggregateExpressionForChild(aggExp: AggregateExpression,
      dataMapSchema: AggregationDataMapSchema,
      attributes: Seq[AttributeReference],
      parentTable: CarbonTable,
      parentLogicalPlan: LogicalPlan,
      aggExpColumnMapping: mutable.LinkedHashSet[AggExpToColumnMappingModel],
      isStreamingTable: Boolean):
  Seq[Expression] = {
    // get the updated aggregate expression, in case of average column
    // it will be divided in two aggergate expression
    val updatedAggExp = PreAggregateUtil.validateAggregateFunctionAndGetFields(aggExp)
    // get the attributes to be updated for child table
    val attrs = aggExpColumnMapping.collect {
      case (schemaAggExpModel)
        if updatedAggExp
          .exists(p =>
            matchExpression(schemaAggExpModel.expression,
            PreAggregateUtil.normalizeExprId(p, parentLogicalPlan.allAttributes))) =>
        attributes filter (_.name.equalsIgnoreCase(
          schemaAggExpModel.columnSchema.get.asInstanceOf[ColumnSchema].getColumnName))
    }.flatten
    // getting aggregate table aggregate expressions
    getAggregateExpressionForAggregation(aggExp, attrs.toSeq, isStreamingTable)
  }

  /**
   * Below method will be used to update the aggregate expression.
   * 1.In case of average below expression will be returned.
   * 1.1 Streaming table
   *    1.1.1 Aggregate table
   *        It will return sum(expression) and count(expression)
   *    1.2.1 Aggregate node added for streaming
   *        It will return Divide(sum(expression)/count(expression))
   * 2.1 Normal table
   *    2.1.1 Aggregate table
   *      It will return Divide(sum(expression)/count(expression))
   * 2. In case of count for aggregate table and aggregate node added for streaming
   *    table count will be aggregated to sum
   *
   * @param aggExp
   * aggregate expression
   * @param attrs
   * aggregate function Attribute, in case of average it will be two to support rollup
   * @return
   * aggregate expression
   */
  def getAggregateExpressionForAggregation(aggExp: AggregateExpression,
      attrs: Seq[AttributeReference],
      isStreamingTable: Boolean = false): Seq[Expression] = {
    aggExp.aggregateFunction match {
      case Sum(MatchCastExpression(_, changeDataType: DataType)) =>
        Seq(AggregateExpression(Sum(Cast(attrs.head, changeDataType)), aggExp.mode, false))
      case Sum(_) =>
        Seq(AggregateExpression(Sum(attrs.head), aggExp.mode, false))
      case Max(MatchCastExpression(_, changeDataType: DataType)) =>
        Seq(AggregateExpression(Max(Cast(attrs.head, changeDataType)), aggExp.mode, false))
      case Max(_) =>
        Seq(AggregateExpression(Max(attrs.head), aggExp.mode, false))
      case Min(MatchCastExpression(_, changeDataType: DataType)) =>
        Seq(AggregateExpression(Min(Cast(attrs.head, changeDataType)), aggExp.mode, false))
      case Min(_) =>
        Seq(AggregateExpression(Min(attrs.head), aggExp.mode, false))
      // Change the count AggregateExpression to Sum as count
      // is already calculated so in case of aggregate table
      // we need to apply sum to get the count
      case Count(Seq(expression: Expression)) =>
        Seq(AggregateExpression(Sum(Cast(attrs.head, LongType)), aggExp.mode, false))

      case Average(MatchCastExpression(exp: Expression, changeDataType: DataType)) =>
        // for handling Normal table case/Aggregate node added in case of streaming table
        if (!isStreamingTable) {
          // In case of average aggregate function select 2 columns from aggregate table
          // with aggregation sum and count.
          // Then add divide(sum(column with sum), sum(column with count)).
          Seq(Divide(AggregateExpression(Sum(Cast(
            attrs.head,
            DoubleType)),
            aggExp.mode,
            false),
            AggregateExpression(Sum(Cast(
              attrs.last,
              DoubleType)),
              aggExp.mode,
              false)))
        } else {
          // in case of streaming aggregate table return two aggregate function sum and count
          Seq(AggregateExpression(Sum(Cast(
            attrs.head,
            DoubleType)),
            aggExp.mode,
            false),
            AggregateExpression(Sum(Cast(
              attrs.last,
              DoubleType)),
              aggExp.mode,
              false))
        }
      case Average(exp: Expression) =>
        val dataType =
          if (exp.dataType.isInstanceOf[DecimalType]) {
            // decimal must not go as double precision.
            exp.dataType.asInstanceOf[DecimalType]
          } else {
            DoubleType
          }
        // for handling Normal table case/Aggregate node added in case of streaming table
        if (!isStreamingTable) {
          // In case of average aggregate function select 2 columns from aggregate table
          // with aggregation sum and count.
          // Then add divide(sum(column with sum), sum(column with count)).
          Seq(Divide(AggregateExpression(Sum(Cast(
            attrs.head,
            dataType)),
            aggExp.mode,
            false),
            AggregateExpression(Sum(Cast(
              attrs.last,
              dataType)),
              aggExp.mode,
              false)))
        } else {
          // in case of streaming aggregate table return two aggregate function sum and count
          Seq(AggregateExpression(Sum(Cast(
            attrs.head,
            dataType)),
            aggExp.mode,
            false),
            AggregateExpression(Sum(Cast(
              attrs.last,
              dataType)),
              aggExp.mode,
              false))
        }
    }
  }
  /**
   * Method to get the carbon table and table name
   * @param parentLogicalRelation parent table relation
   * @return tuple of carbon table
   */
  def getCarbonTable(parentLogicalRelation: LogicalRelation): CarbonTable = {
    val carbonTable = parentLogicalRelation.relation.asInstanceOf[CarbonDatasourceHadoopRelation]
      .carbonRelation
      .metaData.carbonTable
    carbonTable
  }

  /**
   * Below method will be used to get the query columns from plan
   * @param groupByExpression group by expression
   * @param aggregateExpressions aggregate expression
   * @param carbonTable parent carbon table
   * @param queryColumns list of attributes
   * @return plan is valid
   */
  def extractQueryColumnsFromAggExpression(groupByExpression: Seq[Expression],
      aggregateExpressions: Seq[NamedExpression],
      carbonTable: CarbonTable,
      queryColumns: scala.collection.mutable.HashSet[QueryColumn],
      aggreagteExps: scala.collection.mutable.HashSet[AggregateExpression]): Boolean = {
    var isValid = true
    groupByExpression foreach  { expression =>
      extractColumnFromExpression(expression, queryColumns, carbonTable)
    }
    aggregateExpressions.map {
      case attr: AttributeReference =>
        queryColumns += getQueryColumn(attr.name,
          carbonTable)
      case Alias(attr: AttributeReference, _) =>
        queryColumns += getQueryColumn(attr.name,
          carbonTable);
      case Alias(attr: AggregateExpression, _) =>
        if (attr.isDistinct) {
          isValid = false
        }
        val aggExp = PreAggregateUtil.validateAggregateFunctionAndGetFields(attr)
        if (aggExp.nonEmpty) {
          aggreagteExps ++= aggExp
        } else {
          isValid = false
        }
      case Alias(expression: Expression, _) =>
        if (expression.isInstanceOf[ScalaUDF] &&
            expression.asInstanceOf[ScalaUDF].function.getClass.getName.equalsIgnoreCase(
              "org.apache.spark.sql.execution.command.timeseries.TimeseriesFunction") &&
            CarbonUtil.hasTimeSeriesDataMap(carbonTable)) {
          queryColumns += getQueryColumn(expression.asInstanceOf[ScalaUDF].children(0)
            .asInstanceOf[AttributeReference].name,
            carbonTable,
            timeseriesFunction = expression.asInstanceOf[ScalaUDF].children(1).asInstanceOf[Literal]
              .value.toString)
        } else {
          expression.transform {
            case attr: AttributeReference =>
              queryColumns += getQueryColumn(attr.name,
                carbonTable)
              attr
            case attr: AggregateExpression =>
              if (attr.isDistinct) {
                isValid = false
              }
              val aggExp = PreAggregateUtil.validateAggregateFunctionAndGetFields(attr)
              if (aggExp.nonEmpty) {
                aggreagteExps ++= aggExp
              } else {
                isValid = false
              }
              attr

          }
        }
    }
    isValid
  }

  /**
   * Below method will be used to get the query column object which
   * will have details of the column and its property
   *
   * @param columnName parent column name
   * @param carbonTable parent carbon table
   * @param isFilterColumn is filter is applied on column
   * @return query column
   */
  def getQueryColumn(columnName: String,
      carbonTable: CarbonTable,
      isFilterColumn: Boolean = false,
      timeseriesFunction: String = ""): QueryColumn = {
    val columnSchema = carbonTable.getColumnByName(columnName.toLowerCase)
    if(null == columnSchema) {
      null
    } else {
        new QueryColumn(
          columnSchema.getColumnSchema,
        isFilterColumn,
        timeseriesFunction.toLowerCase)
    }
  }
}

/**
 * Data loading rule class to validate and update the data loading query plan
 * Validation rule:
 * 1. update the avg aggregate expression with two columns sum and count
 * 2. Remove duplicate sum and count expression if already there in plan
 * @param sparkSession spark session
 */
case class CarbonPreAggregateDataLoadingRules(sparkSession: SparkSession)
  extends Rule[LogicalPlan] {
  lazy val parser = new CarbonSpark2SqlParser
  override def apply(plan: LogicalPlan): LogicalPlan = {
    val validExpressionsMap = scala.collection.mutable.HashSet.empty[AggExpToColumnMappingModel]
    val namedExpressionList = scala.collection.mutable.LinkedHashSet.empty[NamedExpression]
    plan transform {
      case aggregate@Aggregate(groupingExpressions,
      aExp,
      CarbonSubqueryAlias(_, logicalRelation: LogicalRelation))
        if validateAggregateExpressions(aExp) &&
           logicalRelation.relation.isInstanceOf[CarbonDatasourceHadoopRelation] =>
        aExp.foreach {
          case attr: AttributeReference =>
            namedExpressionList += attr
          case alias@Alias(_: AttributeReference, _) =>
            namedExpressionList += alias
          case alias@Alias(aggExp: AggregateExpression, name) =>
            // get the updated expression for avg convert it to two expression
            // sum and count
            val expressions = PreAggregateUtil.validateAggregateFunctionAndGetFields(aggExp)
            // if size is more than one then it was for average
            if(expressions.size > 1) {
              val sumExp = PreAggregateUtil.normalizeExprId(
                expressions.head,
                aggregate.allAttributes)
              // get the logical plan fro count expression
              val countExp = PreAggregateUtil.normalizeExprId(
                expressions.last,
                aggregate.allAttributes)
              // check with same expression already sum is present then do not add to
              // named expression list otherwise update the list and add it to set
              if (!validExpressionsMap.contains(AggExpToColumnMappingModel(sumExp))) {
                namedExpressionList +=
                CarbonToSparkAdapter.createAliasRef(expressions.head,
                  name + "_ sum",
                  NamedExpression.newExprId,
                  alias.qualifier,
                  Some(alias.metadata),
                  Some(alias))
                validExpressionsMap += AggExpToColumnMappingModel(sumExp)
              }
              // check with same expression already count is present then do not add to
              // named expression list otherwise update the list and add it to set
              if (!validExpressionsMap.contains(AggExpToColumnMappingModel(countExp))) {
                namedExpressionList +=
                CarbonToSparkAdapter.createAliasRef(expressions.last, name + "_ count",
                    NamedExpression.newExprId,
                    alias.qualifier,
                    Some(alias.metadata),
                    Some(alias))
                validExpressionsMap += AggExpToColumnMappingModel(countExp)
              }
            } else {
              // get the logical plan for expression
              val exp = PreAggregateUtil.normalizeExprId(
                expressions.head,
                aggregate.allAttributes)
              // check with same expression already  present then do not add to
              // named expression list otherwise update the list and add it to set
              if (!validExpressionsMap.contains(AggExpToColumnMappingModel(exp))) {
                namedExpressionList+=alias
                validExpressionsMap += AggExpToColumnMappingModel(exp)
              }
            }
          case alias@Alias(_: Expression, _) =>
            namedExpressionList += alias
        }
        groupingExpressions foreach {
          case namedExpr: NamedExpression => namedExpressionList += namedExpr
          case _ => namedExpressionList
        }
        aggregate.copy(aggregateExpressions = namedExpressionList.toSeq)
      case plan: LogicalPlan => plan
    }
  }

  /**
   * Called by PreAggregateLoadingRules to validate if plan is valid for applying rules or not.
   * If the plan has PreAggLoad i.e Loading UDF and does not have PreAgg i.e Query UDF then it is
   * valid.
   * @param namedExpression named expressions
   * @return valid or not
   */
  private def validateAggregateExpressions(namedExpression: Seq[NamedExpression]): Boolean = {
    val filteredExpressions = namedExpression.filterNot(_.isInstanceOf[UnresolvedAlias])
    filteredExpressions.exists { expr =>
      !expr.name.equalsIgnoreCase("PreAgg") && expr.name.equalsIgnoreCase("preAggLoad")
    }
  }
}