Github user ravipesala commented on a diff in the pull request: https://github.com/apache/carbondata/pull/1728#discussion_r159459370 --- Diff: integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonPreAggregateRules.scala --- @@ -98,238 +143,499 @@ case class CarbonPreAggregateQueryRules(sparkSession: SparkSession) extends Rule needAnalysis = false attr } + if(needAnalysis) { + needAnalysis = isValidPlan(plan) + } // 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.HashSet.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 + val updatedPlan = transformPreAggQueryPlan(plan) + val newPlan = updatePlan(updatedPlan) + print(newPlan.toString()) + newPlan + } + } - // When plan has grouping expression, aggregate expression - // subquery - case Aggregate(groupingExp, - aggregateExp, - CarbonSubqueryAlias(_, 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].carbonRelation. - metaData.hasAggregateDataMapSchema => - val (carbonTable, tableName) = getCarbonTableAndTableName(logicalRelation) - // if it is valid plan then extract the query columns - isValidPlan = extractQueryColumnsFromAggExpression(groupingExp, - aggregateExp, - carbonTable, - tableName, - list) - carbonTable + /** + * 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), updateNamedExpression(aggExp), child) + case Filter(filterExp, child) => + Filter(updateConditionExpression(Some(filterExp)).get, child) + case Project(projectList, child) => + Project(updateNamedExpression(projectList), child) + case Sort(sortOrders, global, child) => + Sort(updateSortExpression(sortOrders), global, child) + case Join(left, right, joinType, condition) => + Join(left, right, joinType, updateConditionExpression(condition)) + } + updatedPlan + } - // below case for handling filter query - // When plan has grouping expression, aggregate expression - // filter expression - case Aggregate(groupingExp, - aggregateExp, - Filter(filterExp, - CarbonSubqueryAlias(_, 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].carbonRelation. - metaData.hasAggregateDataMapSchema => - val (carbonTable, tableName) = getCarbonTableAndTableName(logicalRelation) - // if it is valid plan then extract the query columns - isValidPlan = extractQueryColumnsFromAggExpression(groupingExp, - aggregateExp, - carbonTable, - tableName, - list) - if(isValidPlan) { - isValidPlan = !CarbonReflectionUtils.hasPredicateSubquery(filterExp) - } - // getting the columns from filter expression - if(isValidPlan) { - isValidPlan = extractQueryColumnFromFilterExp(filterExp, list, carbonTable, tableName) + /** + * 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 => + order.child match { + case attr: AttributeReference => + val childExp = updatedExpression.find { p => p._1.sameRef(attr) } + if (childExp.isDefined) { + val newExpression = AttributeReference( + childExp.get._2.name, + childExp.get._2.dataType, + childExp.get._2.nullable, + childExp.get._2.metadata)(childExp.get._2.exprId, attr.qualifier, attr.isGenerated) + SortOrder(newExpression, order.direction) + } else { + SortOrder(attr, order.direction) } - carbonTable + } + } + } - // When plan has grouping expression, aggregate expression - // logical relation - case Aggregate(groupingExp, aggregateExp, 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].carbonRelation. - metaData.hasAggregateDataMapSchema => - val (carbonTable, tableName) = getCarbonTableAndTableName(logicalRelation) - // if it is valid plan then extract the query columns - isValidPlan = extractQueryColumnsFromAggExpression(groupingExp, - aggregateExp, - carbonTable, - tableName, - list) - carbonTable - // case for handling aggregation, order by - case Project(projectList, - Sort(sortOrders, - _, - Aggregate(groupingExp, - aggregateExp, - CarbonSubqueryAlias(_, logicalRelation: LogicalRelation)))) - if logicalRelation.relation.isInstanceOf[CarbonDatasourceHadoopRelation] && - logicalRelation.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonRelation. - metaData.hasAggregateDataMapSchema => - val (carbonTable, tableName) = getCarbonTableAndTableName(logicalRelation) - isValidPlan = extractQueryColumnsFromAggExpression(groupingExp, - aggregateExp, - carbonTable, - tableName, - list) - if(isValidPlan) { - list ++ - extractQueryColumnForOrderBy(Some(projectList), sortOrders, carbonTable, tableName) - } - carbonTable - // case for handling aggregation, order by and filter - case Project(projectList, - Sort(sortOrders, - _, - Aggregate(groupingExp, - aggregateExp, - Filter(filterExp, CarbonSubqueryAlias(_, logicalRelation: LogicalRelation))))) - if logicalRelation.relation.isInstanceOf[CarbonDatasourceHadoopRelation] && - logicalRelation.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonRelation. - metaData.hasAggregateDataMapSchema => - val (carbonTable, tableName) = getCarbonTableAndTableName(logicalRelation) - isValidPlan = extractQueryColumnsFromAggExpression(groupingExp, - aggregateExp, - carbonTable, - tableName, - list) - if(isValidPlan) { - isValidPlan = !CarbonReflectionUtils.hasPredicateSubquery(filterExp) + /** + * 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] = { + val newExp = expressions map { expression => + expression transform { + case attr: AttributeReference => + val childExp = updatedExpression.find { p => p._1.sameRef(attr) } + if (childExp.isDefined) { + val newExpression = AttributeReference( + childExp.get._2.name, + childExp.get._2.dataType, + childExp.get._2.nullable, + childExp.get._2.metadata)(childExp.get._2.exprId, attr.qualifier, attr.isGenerated) + newExpression + } else { + attr } - if (isValidPlan) { - list ++ - extractQueryColumnForOrderBy(Some(projectList), sortOrders, carbonTable, tableName) - isValidPlan = extractQueryColumnFromFilterExp(filterExp, list, carbonTable, tableName) + } + } + newExp + } + + /** + * Below method will be used to updated the named expression like aggregate expression + * @param namedExpression + * any named expression like aggregate expression + * @return updated named expression + */ + def updateNamedExpression(namedExpression: Seq[NamedExpression]) : Seq[NamedExpression] = { + namedExpression map { + case attr: AttributeReference => + val childExp = updatedExpression.find(p => p._1.sameRef(attr)) + if(childExp.isDefined) { + val newExp = AttributeReference( + childExp.get._2.name, + childExp.get._2.dataType, + childExp.get._2.nullable, + childExp.get._2.metadata)(childExp.get._2.exprId, attr.qualifier, attr.isGenerated) + newExp + } else { + attr + } + case alias@Alias(exp, name) => + val newExp = exp.transform { + case attr: AttributeReference => + val childExp = updatedExpression.find(p => p._1.sameRef(attr)) + if (childExp.isDefined) { + val newExp = AttributeReference( + childExp.get._2.name, + childExp.get._2.dataType, + childExp.get._2.nullable, + childExp.get._2.metadata)(childExp.get._2.exprId, attr.qualifier, attr.isGenerated) + newExp + } else { + attr + } + } + Alias(newExp, name)(alias.exprId, alias.qualifier, Some(alias.metadata), alias.isGenerated) + } + } + + /** + * Below method will be used to updated condition expression + * @param conditionExp + * any condition expression join condition or filter condition + * @return updated condition expression + */ + def updateConditionExpression(conditionExp: Option[Expression]): Option[Expression] = { + if (conditionExp.isDefined) { + val filterExp = conditionExp.get + Some(filterExp.transform { + case attr: AttributeReference => + val childExp = updatedExpression.find(p => p._1.sameRef(attr)) + if(childExp.isDefined) { + childExp.get._2 + } else { + attr } - carbonTable - // case for handling aggregation with order by when only projection column exits - case Sort(sortOrders, - _, - Aggregate(groupingExp, - aggregateExp, - CarbonSubqueryAlias(_, logicalRelation: LogicalRelation))) - if logicalRelation.relation.isInstanceOf[CarbonDatasourceHadoopRelation] && - logicalRelation.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonRelation. - metaData.hasAggregateDataMapSchema => - val (carbonTable, tableName) = getCarbonTableAndTableName(logicalRelation) - isValidPlan = extractQueryColumnsFromAggExpression(groupingExp, - aggregateExp, + }) + } else { + conditionExp + } + } + + /** + * 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 + * + * @param logicalPlan + * parent logical plan + * @return transformed plan + */ + def transformPreAggQueryPlan(logicalPlan: LogicalPlan): LogicalPlan = { + val updatedPlan = logicalPlan.transform { + // case for aggregation query + case agg@Aggregate(grExp, + aggExp, + child@CarbonSubqueryAlias(_, logicalRelation: LogicalRelation)) + if logicalRelation.relation.isInstanceOf[CarbonDatasourceHadoopRelation] && + logicalRelation.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonRelation. + metaData.hasAggregateDataMapSchema => + val carbonTable = getCarbonTable(logicalRelation) + 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, - tableName, - list) - if(isValidPlan) { - list ++ extractQueryColumnForOrderBy(sortOrders = sortOrders, - carbonTable = carbonTable, - tableName = tableName) + logicalRelation) + 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, + logicalRelation) + Aggregate(updatedGroupExp, + updatedAggExp, + newChild) + } else { + agg } - carbonTable - // case for handling aggregation with order by and filter when only projection column exits - case Sort(sortOrders, - _, - Aggregate(groupingExp, - aggregateExp, - Filter(filterExp, CarbonSubqueryAlias(_, logicalRelation: LogicalRelation)))) - if logicalRelation.relation.isInstanceOf[CarbonDatasourceHadoopRelation] && - logicalRelation.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonRelation. - metaData.hasAggregateDataMapSchema => - val (carbonTable, tableName) = getCarbonTableAndTableName(logicalRelation) - isValidPlan = extractQueryColumnsFromAggExpression(groupingExp, - aggregateExp, + } else { + agg + } + // case of handling aggregation query with filter + case agg@Aggregate(grExp, + aggExp, + Filter(expression, child@CarbonSubqueryAlias(_, logicalRelation: LogicalRelation))) + if logicalRelation.relation.isInstanceOf[CarbonDatasourceHadoopRelation] && + logicalRelation.relation.asInstanceOf[CarbonDatasourceHadoopRelation].carbonRelation. + metaData.hasAggregateDataMapSchema => + val carbonTable = getCarbonTable(logicalRelation) + val list = scala.collection.mutable.HashSet.empty[QueryColumn] + val aggregateExpressions = scala.collection.mutable.HashSet.empty[AggregateExpression] + var isValidPlan = extractQueryColumnsFromAggExpression( + grExp, + aggExp, + carbonTable, + list, + aggregateExpressions) + // getting the columns from filter expression + isValidPlan = !CarbonReflectionUtils.hasPredicateSubquery(expression) + if (isValidPlan) { + isValidPlan = extractQueryColumnFromFilterExp(expression, list, carbonTable) + } + if(isValidPlan) { + val (aggDataMapSchema, childPlan) = getChildDataMapForTransformation(list, + aggregateExpressions, carbonTable, - tableName, - list) - if(isValidPlan) { - isValidPlan = !CarbonReflectionUtils.hasPredicateSubquery(filterExp) - } - if(isValidPlan) { - list ++ extractQueryColumnForOrderBy(sortOrders = sortOrders, - carbonTable = carbonTable, - tableName = tableName) - isValidPlan = extractQueryColumnFromFilterExp(filterExp, list, carbonTable, tableName) - } - carbonTable - case _ => - isValidPlan = false - null - } - if (isValidPlan && null != carbonTable) { - isValidPlan = isSpecificSegmentPresent(carbonTable) - } - // if plan is valid then update the plan with child attributes - if (isValidPlan) { - // getting all the projection columns - val listProjectionColumn = list - .filter(queryColumn => queryColumn.getAggFunction.isEmpty && !queryColumn.isFilterColumn) - .toList - // getting all the filter columns - val listFilterColumn = list - .filter(queryColumn => queryColumn.getAggFunction.isEmpty && queryColumn.isFilterColumn) - .toList - // getting all the aggregation columns - val listAggregationColumn = list.filter(queryColumn => !queryColumn.getAggFunction.isEmpty) - .toList - // create a query plan object which will be used to select the list of pre aggregate tables - // matches with this plan - val queryPlan = new QueryPlan(listProjectionColumn.asJava, - listAggregationColumn.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() - // if it does not match with any pre aggregate table return the same plan - if (!selectedDataMapSchemas.isEmpty) { - // filter the selected child schema based on size to select the pre-aggregate tables - // that are nonEmpty - val catalog = CarbonEnv.getInstance(sparkSession).carbonMetastore - val relationBuffer = selectedDataMapSchemas.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) - if (relationBuffer.isEmpty) { - // If the size of relation Buffer is 0 then it means that none of the pre-aggregate - // tables have date yet. - // In this case we would return the original plan so that the query hits the parent - // table. - plan + logicalRelation) + 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, + logicalRelation) + Aggregate(updatedGroupExp, + updatedAggExp, + Filter(updatedFilterExpression.get, + newChild)) } else { - // If the relationBuffer is nonEmpty then find the table with the minimum size. - val (aggDataMapSchema, _, relation) = relationBuffer.minBy(_._2.sizeInBytes) - val newRelation = new FindDataSourceTable(sparkSession).apply(relation) - // transform the query plan based on selected child schema - transformPreAggQueryPlan(plan, aggDataMapSchema, newRelation) + agg } } else { - plan + agg } + } + updatedPlan + } + + /** + * 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 logicalRelation + * 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, + logicalRelation: LogicalRelation): (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 QueryPlan(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 + } + val aggExpLogicalPlans = aggregateExpressions.map { queryAggExp => + PreAggregateUtil.getLogicalPlanFromAggExp(queryAggExp, + carbonTable.getTableName, + carbonTable.getDatabaseName, + logicalRelation, + sparkSession, + parser) + }.toSeq + // if query does not have any aggregate function no need to validate the same + if (aggregateExpressions.size > 0 && selectedAggMaps.size > 0) { + selectedAggMaps = validateAggregateExpression(selectedAggMaps.asScala.toSeq, + carbonTable, + logicalRelation, + aggExpLogicalPlans).asJava + } + // 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 nonEmpty + 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) + if (relationBuffer.isEmpty) { + // If the size of relation Buffer is 0 then it means that none of the pre-aggregate + // tables have date yet. + // In this case we would return the original plan so that the query hits the parent + // table. + (null, null) } else { - plan + // If the relationBuffer is nonEmpty then find the table with the minimum size. + val (aggDataMapSchema, _, relation) = relationBuffer.minBy(_._2.sizeInBytes) + val newRelation = new FindDataSourceTable(sparkSession).apply(relation) + (aggDataMapSchema.asInstanceOf[AggregationDataMapSchema], newRelation) } + } 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 logicalRelation + * parent logical relation + * @param queryAggExpLogicalPlans + * query agg expression logical plan + * @return valid data map + */ + def validateAggregateExpression(selectedDataMap: Seq[DataMapSchema], + carbonTable: CarbonTable, + logicalRelation: LogicalRelation, + queryAggExpLogicalPlans: Seq[LogicalPlan]): Seq[DataMapSchema] = { + def validateDataMap(dataMap: DataMapSchema, + aggExpLogicalPlans: Seq[LogicalPlan]): Boolean = { + val mappingModel = getLogicalPlanForAggregateExpression(dataMap, + carbonTable, + logicalRelation) + aggExpLogicalPlans.forall{ + p => mappingModel.exists(m => p.sameResult(m.logicalPlan)) --- End diff -- move p => to up
---