Github user ravipesala commented on a diff in the pull request: https://github.com/apache/carbondata/pull/1464#discussion_r149004422 --- Diff: integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonPreAggregateQueryRules.scala --- @@ -0,0 +1,756 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql.hive + +import scala.collection.JavaConverters._ + +import org.apache.spark.sql.{AnalysisException, CarbonDatasourceHadoopRelation, SparkSession} +import org.apache.spark.sql.catalyst.TableIdentifier +import org.apache.spark.sql.catalyst.analysis.UnresolvedAlias +import org.apache.spark.sql.catalyst.expressions.{Alias, AttributeReference, Cast, Divide, +Expression, NamedExpression} +import org.apache.spark.sql.catalyst.expressions.aggregate._ +import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Filter, LogicalPlan, SubqueryAlias} +import org.apache.spark.sql.catalyst.rules.Rule +import org.apache.spark.sql.execution.datasources.LogicalRelation +import org.apache.spark.sql.types._ + +import org.apache.carbondata.core.metadata.schema.table.{CarbonTable, DataMapSchema} +import org.apache.carbondata.core.preagg.{AggregateTableSelector, QueryColumn, QueryPlan} +import org.apache.carbondata.spark.util.CarbonScalaUtil + +/** + * 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 Upadating 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 + * + * @param sparkSession + * spark session + */ +case class CarbonPreAggregateQueryRules(sparkSession: SparkSession) extends Rule[LogicalPlan] { + + 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(_, name) if name.equals("preAgg") => + 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 + } + // if plan is not valid for transformation then return same plan + if (!needAnalysis) { + plan + } else { + // create buffer to collect all the column and its metadata information + val list = scala.collection.mutable.ListBuffer.empty[QueryColumn] + var isValidPlan = true + val carbonTable = plan match { + // matching the plan based on supported plan + // if plan is matches with any case it will validate and get all + // information required for transforming the plan + + // When plan has grouping expression, aggregate expression + // subquery + case Aggregate(groupingExp, + aggregateExp, + SubqueryAlias(_, logicalRelation: LogicalRelation, _)) + // only carbon query plan is supported checking whether logical relation is + // is for carbon + if logicalRelation.relation.isInstanceOf[CarbonDatasourceHadoopRelation] + && + logicalRelation.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonTable + .hasPreAggDataMap => + val (carbonTable, tableName) = getCarbonTableAndTableName(logicalRelation) + // when table has child data map(pre aggregate table) then only plan will be transformed + if (!carbonTable.hasPreAggDataMap) { + isValidPlan = false + } + if (isValidPlan) { + // if it is valid plan then extract the query columns + isValidPlan = extractQueryColumnsFromAggExpression(groupingExp, + aggregateExp, + carbonTable, + tableName, + list) + } + carbonTable + + // below case for handling filter query + // When plan has grouping expression, aggregate expression + // filter expression + case Aggregate(groupingExp, aggregateExp, + Filter(filterExp, + SubqueryAlias(_, logicalRelation: LogicalRelation, _))) + // only carbon query plan is supported checking whether logical relation is + // is for carbon + if logicalRelation.relation.isInstanceOf[CarbonDatasourceHadoopRelation] + && --- End diff -- put && on the above line. Please do this for all case statements
---