| /* |
| * 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.optimizer |
| |
| import java.util |
| import java.util.concurrent.atomic.AtomicLong |
| |
| import scala.collection.JavaConverters._ |
| import scala.collection.mutable.ArrayBuffer |
| import scala.util.control.Breaks.{break, breakable} |
| |
| import org.apache.spark.sql.{CarbonToSparkAdapter, SparkSession} |
| import org.apache.spark.sql.catalyst.catalog.HiveTableRelation |
| import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeMap, AttributeReference, Expression, Literal, NamedExpression, ScalaUDF, SortOrder} |
| import org.apache.spark.sql.catalyst.expressions.aggregate._ |
| import org.apache.spark.sql.catalyst.plans.logical.{Filter, Join, LogicalPlan} |
| import org.apache.spark.sql.execution.datasources.LogicalRelation |
| import org.apache.spark.sql.types.DataTypes |
| import org.apache.spark.unsafe.types.UTF8String |
| |
| import org.apache.carbondata.core.preagg.TimeSeriesFunctionEnum |
| import org.apache.carbondata.mv.expressions.modular.{ModularSubquery, ScalarModularSubquery} |
| import org.apache.carbondata.mv.plans.modular.{ExpressionHelper, GroupBy, HarmonizedRelation, LeafNode, Matchable, ModularPlan, ModularRelation, Select, SimpleModularizer} |
| import org.apache.carbondata.mv.plans.util.BirdcageOptimizer |
| import org.apache.carbondata.view.{MVCatalogInSpark, MVPlanWrapper, MVTimeGranularity, TimeSeriesFunction} |
| |
| /** |
| * The primary workflow for rewriting relational queries using Spark libraries. |
| * Designed to allow easy access to the intermediate phases of query rewrite for developers. |
| * |
| * While this is not a public class, we should avoid changing the function names for the sake of |
| * changing them, because a lot of developers use the feature for debugging. |
| */ |
| class MVRewrite(catalog: MVCatalogInSpark, logicalPlan: LogicalPlan, |
| session: SparkSession) { |
| |
| private def getAliasName(expression: NamedExpression): String = { |
| expression match { |
| case Alias(_, name) => name |
| case reference: AttributeReference => reference.name |
| } |
| } |
| |
| private def getAliasMap( |
| outputList1: Seq[NamedExpression], |
| outputList2: Seq[NamedExpression]): Map[AttributeKey, NamedExpression] = { |
| // when mv is created with duplicate columns like select sum(age),sum(age) from table, |
| // the outputList2 will have duplicate, so handle that case here |
| if (outputList1.length == outputList2.groupBy(_.name).size) { |
| outputList2.zip(outputList1).flatMap { |
| case (output1, output2) => |
| var aliasMapping = output1 collect { |
| case reference: AttributeReference => |
| (AttributeKey(reference), Alias(output2, reference.name)(exprId = output2.exprId)) |
| } |
| output1 match { |
| case alias: Alias => |
| aliasMapping = Seq((AttributeKey(alias.child), |
| Alias(output2, alias.name)(exprId = output2.exprId))) ++ aliasMapping |
| case _ => |
| } |
| Seq((AttributeKey(output1), |
| Alias(output2, output1.name)(exprId = output2.exprId))) ++ aliasMapping |
| }.toMap |
| } else { |
| throw new UnsupportedOperationException("Cannot create mapping with unequal sizes") |
| } |
| } |
| |
| private def getUpdatedOutputList(outputList: Seq[NamedExpression], |
| modularPlan: Option[ModularPlan]): Seq[NamedExpression] = { |
| modularPlan.collect { |
| case planWrapper: MVPlanWrapper => |
| val viewSchema = planWrapper.viewSchema |
| val viewColumnsOrderMap = viewSchema.getColumnsOrderMap |
| if (null != viewColumnsOrderMap && !viewColumnsOrderMap.isEmpty) { |
| val updatedOutputList = new util.ArrayList[NamedExpression]() |
| var i = 0 |
| while (i < viewColumnsOrderMap.size()) { |
| val updatedOutput = outputList.filter( |
| output => output.name.equalsIgnoreCase(viewColumnsOrderMap.get(i)) |
| ).head |
| updatedOutputList.add(updatedOutput) |
| i = i + 1 |
| } |
| updatedOutputList.asScala |
| } else { |
| outputList |
| } |
| case _ => outputList |
| }.get |
| } |
| |
| /** |
| * Check if modular plan can be rolled up by rewriting and matching the modular plan |
| * with existing mv datasets. |
| * @param modularPlan to check if can be rolled up |
| * @return new modular plan |
| */ |
| private def getRolledUpModularPlan(modularPlan: ModularPlan): Option[ModularPlan] = { |
| // check if modular plan contains timeseries udf |
| val timeSeriesFunction = modularPlan match { |
| case select: Select => |
| getTimeSeriesFunction(select.outputList) |
| case groupBy: GroupBy => |
| getTimeSeriesFunction(groupBy.outputList) |
| case _ => (null, null) |
| } |
| // set canDoRollUp to false, in case of Join queries and if filter has timeseries udf function |
| // TODO: support rollUp for join queries |
| var canRollUp = true |
| optimizedPlan.transformDown { |
| case join@Join(_, _, _, _) => |
| canRollUp = false |
| join |
| case filter@Filter(condition: Expression, _) => |
| condition.collect { |
| case function: ScalaUDF if function.function.isInstanceOf[TimeSeriesFunction] => |
| canRollUp = false |
| } |
| filter |
| } |
| if (null != timeSeriesFunction._2 && canRollUp) { |
| // check for rollup and rewrite the plan |
| // collect all the datasets which contains timeseries MV's |
| val viewSchemaWrappers = catalog.lookupFeasibleSchemas(modularPlan) |
| var granularityList: java.util.List[TimeSeriesFunctionEnum] = |
| new java.util.ArrayList[TimeSeriesFunctionEnum]() |
| // Get all the lower granularities for the query from datasets |
| viewSchemaWrappers.foreach { viewSchemaWrapper => |
| if (viewSchemaWrapper.viewSchema.isTimeSeries) { |
| viewSchemaWrapper.logicalPlan.transformExpressions { |
| case alias@Alias(function: ScalaUDF, _) => |
| if (function.function.isInstanceOf[TimeSeriesFunction]) { |
| val granularity = |
| function.children.last.asInstanceOf[Literal].toString().toUpperCase() |
| if (MVTimeGranularity.get(timeSeriesFunction._2.toUpperCase).seconds > |
| MVTimeGranularity.get(granularity).seconds) { |
| granularityList.add(TimeSeriesFunctionEnum.valueOf(granularity)) |
| } |
| } |
| alias |
| } |
| } |
| } |
| if (!granularityList.isEmpty) { |
| granularityList = granularityList.asScala.sortBy(_.getOrdinal)(Ordering[Int].reverse).asJava |
| var originalTable: String = null |
| // get query Table |
| modularPlan.collect { |
| case relation: ModularRelation => |
| originalTable = relation.tableName |
| } |
| var queryGranularity: String = null |
| var rewrittenPlan = modularPlan |
| // replace granularities in the plan and check if plan can be changed |
| breakable { |
| granularityList.asScala.foreach { func => |
| rewrittenPlan = rewriteGranularityInPlan(modularPlan, func.getName) |
| rewrittenPlans.add(rewrittenPlan) |
| val logicalPlan = session.sql(rewrittenPlan.asCompactSQL).queryExecution.optimizedPlan |
| rewrittenPlans.clear() |
| var rolledUpTable: String = null |
| logicalPlan.collect { |
| case relation: LogicalRelation => |
| rolledUpTable = relation.catalogTable.get.identifier.table |
| case relation: HiveTableRelation => |
| rolledUpTable = relation.tableMeta.identifier.table |
| } |
| if (!rolledUpTable.equalsIgnoreCase(originalTable)) { |
| queryGranularity = func.getName |
| break() |
| } |
| } |
| } |
| if (null != queryGranularity) { |
| // rewrite the plan and set it as rolled up plan |
| val rewrittenPlans = rewriteWithSchemaWrapper0(rewrittenPlan) |
| if (rewrittenPlans.isDefined) { |
| rewrittenPlans.get.map(_.setRolledUp()) |
| rewrittenPlans |
| } else { |
| None |
| } |
| } else { |
| None |
| } |
| } else { |
| None |
| } |
| } else { |
| None |
| } |
| } |
| |
| private def getTimeSeriesFunction(outputList: scala.Seq[NamedExpression]): (String, String) = { |
| outputList.collect { |
| case Alias(function: ScalaUDF, name) => |
| if (function.function.isInstanceOf[TimeSeriesFunction]) { |
| function.children.collect { |
| case literal: Literal => |
| return (name, literal.value.toString) |
| } |
| } |
| } |
| (null, null) |
| } |
| |
| /** |
| * Rewrite the updated mv query with corresponding MV table. |
| */ |
| private def rewrite(modularPlan: ModularPlan): ModularPlan = { |
| if (modularPlan.find(_.rewritten).isDefined) { |
| var updatedPlan = modularPlan transform { |
| case select: Select => |
| updatePlan(select) |
| case groupBy: GroupBy => |
| updatePlan(groupBy) |
| } |
| if (modularPlan.isRolledUp) { |
| // If the rewritten query is rolled up, then rewrite the query based on the original modular |
| // plan. Make a new outputList based on original modular plan and wrap rewritten plan with |
| // select & group-by nodes with new outputList. |
| |
| // For example: |
| // Given User query: |
| // SELECT timeseries(col,'day') from maintable group by timeseries(col,'day') |
| // If plan is rewritten as per 'hour' granularity of mv1, |
| // then rewritten query will be like, |
| // SELECT mv1.`UDF:timeseries_projectjoindate_hour` AS `UDF:timeseries |
| // (projectjoindate, hour)` |
| // FROM |
| // default.mv1 |
| // GROUP BY mv1.`UDF:timeseries_projectjoindate_hour` |
| // |
| // Now, rewrite the rewritten plan as per the 'day' granularity |
| // SELECT timeseries(gen_subsumer_0.`UDF:timeseries(projectjoindate, hour)`,'day' ) AS |
| // `UDF:timeseries(projectjoindate, day)` |
| // FROM |
| // (SELECT mv2.`UDF:timeseries_projectjoindate_hour` AS `UDF:timeseries |
| // (projectjoindate, hour)` |
| // FROM |
| // default.mv2 |
| // GROUP BY mv2.`UDF:timeseries_projectjoindate_hour`) gen_subsumer_0 |
| // GROUP BY timeseries(gen_subsumer_0.`UDF:timeseries(projectjoindate, hour)`,'day' ) |
| harmonizedPlan match { |
| case select: Select => |
| val outputList = select.outputList |
| val rolledUpOutputList = updatedPlan.asInstanceOf[Select].outputList |
| var finalOutputList: Seq[NamedExpression] = Seq.empty |
| val mapping = outputList zip rolledUpOutputList |
| val newSubsume = subqueryNameGenerator.newSubsumerName() |
| |
| mapping.foreach { outputLists => |
| val name: String = getAliasName(outputLists._2) |
| outputLists._1 match { |
| case a@Alias(scalaUdf: ScalaUDF, aliasName) => |
| if (scalaUdf.function.isInstanceOf[TimeSeriesFunction]) { |
| val newName = newSubsume + ".`" + name + "`" |
| val transformedUdf = updateTimeSeriesFunction(scalaUdf, newName) |
| finalOutputList = finalOutputList.:+(Alias(transformedUdf, aliasName)(a.exprId, |
| a.qualifier).asInstanceOf[NamedExpression]) |
| } |
| case Alias(attr: AttributeReference, _) => |
| finalOutputList = finalOutputList.:+( |
| CarbonToSparkAdapter.createAttributeReference(attr, name, newSubsume)) |
| case attr: AttributeReference => |
| finalOutputList = finalOutputList.:+( |
| CarbonToSparkAdapter.createAttributeReference(attr, name, newSubsume)) |
| } |
| } |
| val newChildren = new collection.mutable.ArrayBuffer[ModularPlan]() |
| val newAliasMap = new collection.mutable.HashMap[Int, String]() |
| |
| val sel_plan = select.copy(outputList = finalOutputList, |
| inputList = finalOutputList, |
| predicateList = Seq.empty) |
| newChildren += sel_plan |
| newAliasMap += (newChildren.indexOf(sel_plan) -> newSubsume) |
| updatedPlan = select.copy(outputList = finalOutputList, |
| inputList = finalOutputList, |
| aliasMap = newAliasMap.toMap, |
| predicateList = Seq.empty, |
| children = Seq(updatedPlan)).setRewritten() |
| |
| case groupBy: GroupBy => |
| updatedPlan match { |
| case select: Select => |
| val selectOutputList = groupBy.outputList |
| val rolledUpOutputList = updatedPlan.asInstanceOf[Select].outputList |
| var finalOutputList: Seq[NamedExpression] = Seq.empty |
| var predicateList: Seq[Expression] = Seq.empty |
| val mapping = selectOutputList zip rolledUpOutputList |
| val newSubsume = subqueryNameGenerator.newSubsumerName() |
| |
| mapping.foreach { outputLists => |
| val aliasName: String = getAliasName(outputLists._2) |
| outputLists._1 match { |
| case a@Alias(scalaUdf: ScalaUDF, _) => |
| if (scalaUdf.function.isInstanceOf[TimeSeriesFunction]) { |
| val newName = newSubsume + ".`" + aliasName + "`" |
| val transformedUdf = updateTimeSeriesFunction(scalaUdf, newName) |
| groupBy.predicateList.foreach { |
| case udf: ScalaUDF if udf.isInstanceOf[ScalaUDF] => |
| predicateList = predicateList.:+(transformedUdf) |
| case attr: AttributeReference => |
| predicateList = predicateList.:+( |
| CarbonToSparkAdapter.createAttributeReference(attr, |
| attr.name, |
| newSubsume)) |
| } |
| finalOutputList = finalOutputList.:+(Alias(transformedUdf, a.name)(a |
| .exprId, a.qualifier).asInstanceOf[NamedExpression]) |
| } |
| case attr: AttributeReference => |
| finalOutputList = finalOutputList.:+( |
| CarbonToSparkAdapter.createAttributeReference(attr, aliasName, newSubsume)) |
| case Alias(attr: AttributeReference, _) => |
| finalOutputList = finalOutputList.:+( |
| CarbonToSparkAdapter.createAttributeReference(attr, aliasName, newSubsume)) |
| case a@Alias(agg: AggregateExpression, name) => |
| val newAgg = agg.transform { |
| case attr: AttributeReference => |
| CarbonToSparkAdapter.createAttributeReference(attr, name, newSubsume) |
| } |
| finalOutputList = finalOutputList.:+(Alias(newAgg, name)(a.exprId, |
| a.qualifier).asInstanceOf[NamedExpression]) |
| case other => other |
| } |
| } |
| val newChildren = new collection.mutable.ArrayBuffer[ModularPlan]() |
| val newAliasMap = new collection.mutable.HashMap[Int, String]() |
| |
| val sel_plan = select.copy(outputList = finalOutputList, |
| inputList = finalOutputList, |
| predicateList = Seq.empty) |
| newChildren += sel_plan |
| newAliasMap += (newChildren.indexOf(sel_plan) -> newSubsume) |
| updatedPlan = select.copy(outputList = finalOutputList, |
| inputList = finalOutputList, |
| aliasMap = newAliasMap.toMap, |
| children = Seq(updatedPlan)).setRewritten() |
| updatedPlan = groupBy.copy(outputList = finalOutputList, |
| inputList = finalOutputList, |
| predicateList = predicateList, |
| alias = Some(newAliasMap.mkString), |
| child = updatedPlan).setRewritten() |
| updatedPlan = select.copy(outputList = finalOutputList, |
| inputList = finalOutputList, |
| children = Seq(updatedPlan)).setRewritten() |
| } |
| } |
| } |
| updatedPlan |
| } else { |
| modularPlan |
| } |
| } |
| |
| private def rewriteWithSchemaWrapper(modularPlan: ModularPlan): ModularPlan = { |
| val rewrittenPlan = modularPlan.transformAllExpressions { |
| case subquery: ModularSubquery => |
| if (subquery.children.isEmpty) { |
| rewriteWithSchemaWrapper0(subquery.plan) match { |
| case Some(rewrittenPlan) => |
| ScalarModularSubquery(rewrittenPlan, subquery.children, subquery.exprId) |
| case None => |
| subquery |
| } |
| } |
| else { |
| throw new UnsupportedOperationException(s"Rewrite expression $subquery isn't supported") |
| } |
| case other => other |
| } |
| rewriteWithSchemaWrapper0(rewrittenPlan).getOrElse(rewrittenPlan) |
| } |
| |
| private def rewriteWithSchemaWrapper0(modularPlan: ModularPlan): Option[ModularPlan] = { |
| val rewrittenPlan = modularPlan transformDown { |
| case plan => |
| if (plan.rewritten || !plan.isSPJGH) { |
| plan |
| } else { |
| val rewrittenPlans = |
| for {schemaWrapper <- catalog.lookupFeasibleSchemas(plan).toStream |
| subsumer <- SimpleModularizer.modularize( |
| BirdcageOptimizer.execute(schemaWrapper.logicalPlan)).map(_.semiHarmonized) |
| subsumee <- unifySubsumee(plan) |
| rewrittenPlan <- rewriteWithSchemaWrapper0( |
| unifySubsumer2( |
| unifySubsumer1( |
| subsumer, |
| subsumee, |
| schemaWrapper.modularPlan), |
| subsumee), |
| subsumee)} yield { |
| rewrittenPlan |
| } |
| rewrittenPlans.headOption.map(_.setRewritten()).getOrElse(plan) |
| } |
| } |
| if (rewrittenPlan.fastEquals(modularPlan)) { |
| if (rewrittenPlans.asScala.exists(plan => plan.sameResult(rewrittenPlan))) { |
| return None |
| } |
| getRolledUpModularPlan(rewrittenPlan) |
| } else { |
| Some(rewrittenPlan) |
| } |
| } |
| |
| private def rewriteWithSchemaWrapper0( |
| subsumer: ModularPlan, |
| subsumee: ModularPlan): Option[ModularPlan] = { |
| if (subsumer.getClass == subsumee.getClass) { |
| (subsumer.children, subsumee.children) match { |
| case (Nil, Nil) => None |
| case (r, e) if r.forall(_.isInstanceOf[LeafNode]) && e.forall(_.isInstanceOf[LeafNode]) => |
| val matchesPlans = |
| MVMatchMaker.execute(subsumer, subsumee, None, subqueryNameGenerator) |
| if (matchesPlans.hasNext) { |
| Some(matchesPlans.next) |
| } else { |
| None |
| } |
| case (subsumerChild :: Nil, subsumeeChild :: Nil) => |
| val compensation = rewriteWithSchemaWrapper0(subsumerChild, subsumeeChild) |
| val matchesPlans = compensation.map { |
| case comp if comp.eq(subsumerChild) => |
| MVMatchMaker.execute(subsumer, subsumee, None, subqueryNameGenerator) |
| case _ => |
| MVMatchMaker.execute(subsumer, subsumee, compensation, |
| subqueryNameGenerator) |
| } |
| if (matchesPlans.isDefined && matchesPlans.get.hasNext) { |
| Some(matchesPlans.get.next) |
| } else { |
| None |
| } |
| case _ => None |
| } |
| } else { |
| None |
| } |
| } |
| |
| /** |
| * Replace the identified immediate lower level granularity in the modular plan |
| * to perform rollup |
| * |
| * @param plan to be re-written |
| * @param queryGranularity to be replaced |
| * @return plan with granularity changed |
| */ |
| private def rewriteGranularityInPlan(plan: ModularPlan, queryGranularity: String) = { |
| val newPlan = plan.transformDown { |
| case p => p.transformAllExpressions { |
| case udf: ScalaUDF => |
| if (udf.function.isInstanceOf[TimeSeriesFunction]) { |
| val transformedUdf = udf.transformDown { |
| case _: Literal => |
| new Literal(UTF8String.fromString(queryGranularity.toLowerCase), |
| DataTypes.StringType) |
| } |
| transformedUdf |
| } else { |
| udf |
| } |
| case alias@Alias(udf: ScalaUDF, name) => |
| if (udf.function.isInstanceOf[TimeSeriesFunction]) { |
| var literal: String = null |
| val transformedUdf = udf.transformDown { |
| case l: Literal => |
| literal = l.value.toString |
| new Literal(UTF8String.fromString(queryGranularity.toLowerCase), |
| DataTypes.StringType) |
| } |
| Alias(transformedUdf, |
| name.replace(", " + literal, ", " + queryGranularity))(alias.exprId, |
| alias.qualifier).asInstanceOf[NamedExpression] |
| } else { |
| alias |
| } |
| } |
| } |
| newPlan |
| } |
| |
| // match the join table of subsumer and subsumee |
| // when the table names are the same |
| private def isJoinMatched( |
| subsumer: ModularPlan, |
| subsumee: ModularPlan, |
| subsumerTable: LeafNode, |
| subsumeeTable: LeafNode, |
| subsumerIndex: Int, |
| subsumeeIndex: Int): Boolean = { |
| (subsumerTable, subsumeeTable) match { |
| case _: (ModularRelation, ModularRelation) => |
| val subsumerTableParent = subsumer.find(plan => plan.children.contains(subsumerTable)).get |
| val subsumeeTableParent = subsumee.find(plan => plan.children.contains(subsumeeTable)).get |
| (subsumerTableParent, subsumeeTableParent) match { |
| case (subsumerSelect: Select, subsumeeSelect: Select) => |
| val intersectJoinEdges = subsumeeSelect.joinEdges intersect subsumerSelect.joinEdges |
| if (intersectJoinEdges.nonEmpty) { |
| return intersectJoinEdges.exists( |
| join => |
| join.left == subsumerIndex && |
| join.left == subsumeeIndex || join.right == subsumerIndex && |
| join.right == subsumeeIndex |
| ) |
| } |
| case _ => return false |
| } |
| } |
| true |
| } |
| |
| // add Select operator as placeholder on top of subsumee to facilitate matching |
| private def unifySubsumee(subsumee: ModularPlan): Option[ModularPlan] = { |
| subsumee match { |
| case groupBy@GroupBy(_, _, _, _, Select(_, _, _, _, _, _, _, _, _, _), _, _, _) => |
| Some( |
| Select( |
| groupBy.outputList, |
| groupBy.outputList, |
| Nil, |
| Map.empty, |
| Nil, |
| groupBy :: Nil, |
| groupBy.flags, |
| groupBy.flagSpec, |
| Seq.empty) |
| ) |
| case other => Some(other) |
| } |
| } |
| |
| // add Select operator as placeholder on top of subsumer to facilitate matching |
| private def unifySubsumer1( |
| subsumer: ModularPlan, |
| subsumee: ModularPlan, |
| modularPlan: ModularPlan): ModularPlan = { |
| // Update MV table relation to the subsumer modular plan |
| val updatedSubsumer = subsumer match { |
| // In case of order by it adds extra select but that can be ignored while doing selection. |
| case select@Select(_, _, _, _, _, Seq(g: GroupBy), _, _, _, _) => |
| select.copy( |
| children = Seq(g.copy(modularPlan = Some(modularPlan))), |
| outputList = updateDuplicateColumns(select.outputList)) |
| case select: Select => |
| select.copy( |
| modularPlan = Some(modularPlan), |
| outputList = updateDuplicateColumns(select.outputList)) |
| case groupBy: GroupBy => |
| groupBy.copy( |
| modularPlan = Some(modularPlan), |
| outputList = updateDuplicateColumns(groupBy.outputList)) |
| case other => other |
| } |
| (updatedSubsumer, subsumee) match { |
| case ( |
| groupBy@GroupBy(_, _, _, _, Select(_, _, _, _, _, _, _, _, _, _), _, _, _), |
| Select(_, _, _, _, _, |
| Seq(GroupBy(_, _, _, _, Select(_, _, _, _, _, _, _, _, _, _), _, _, _)), _, _, _, _) |
| ) => |
| Select( |
| groupBy.outputList, |
| groupBy.outputList, |
| Nil, |
| Map.empty, |
| Nil, |
| groupBy :: Nil, |
| groupBy.flags, |
| groupBy.flagSpec, |
| Seq.empty).setSkip() |
| case _ => updatedSubsumer.setSkip() |
| } |
| } |
| |
| private def unifySubsumer2(subsumer: ModularPlan, subsumee: ModularPlan): ModularPlan = { |
| val subsumerTables = subsumer.collect { case node: LeafNode => node } |
| val subsumeeTables = subsumee.collect { case node: LeafNode => node } |
| val tableMappings = for { |
| subsumerTableIndex <- subsumerTables.indices |
| subsumeeTableIndex <- subsumeeTables.indices |
| if subsumerTables(subsumerTableIndex) == subsumeeTables(subsumeeTableIndex) && |
| isJoinMatched( |
| subsumer, |
| subsumee, |
| subsumerTables(subsumerTableIndex), |
| subsumeeTables(subsumeeTableIndex), |
| subsumerTableIndex, |
| subsumeeTableIndex |
| ) |
| } yield { |
| (subsumerTables(subsumerTableIndex), subsumeeTables(subsumeeTableIndex)) |
| } |
| tableMappings.foldLeft(subsumer) { |
| case (currentSubsumer, tableMapping) => |
| val mappedOperator = |
| tableMapping._1 match { |
| case relation: HarmonizedRelation if relation.hasTag => |
| tableMapping._2.asInstanceOf[HarmonizedRelation].addTag |
| case _ => |
| tableMapping._2 |
| } |
| val nextSubsumer = currentSubsumer.transform { |
| case node: ModularRelation if node.fineEquals(tableMapping._1) => mappedOperator |
| case tableMapping._1 if !tableMapping._1.isInstanceOf[ModularRelation] => mappedOperator |
| } |
| // reverse first due to possible tag for left join |
| val rewrites = AttributeMap(tableMapping._1.output.zip(mappedOperator.output)) |
| nextSubsumer.transformUp { |
| case plan => plan.transformExpressions { |
| case attribute: Attribute if rewrites contains attribute => |
| rewrites(attribute).withQualifier(attribute.qualifier) |
| } |
| } |
| } |
| } |
| |
| private def updateTimeSeriesFunction(function: ScalaUDF, newName: String): Expression = { |
| function.transformDown { |
| case reference: AttributeReference => |
| AttributeReference(newName, reference.dataType)( |
| exprId = reference.exprId, |
| qualifier = reference.qualifier) |
| case literal: Literal => |
| Literal(UTF8String.fromString("'" + literal.toString() + "'"), |
| org.apache.spark.sql.types.DataTypes.StringType) |
| } |
| } |
| |
| private def updateDuplicateColumns(outputList: Seq[NamedExpression]): Seq[NamedExpression] = { |
| val duplicateColNames = outputList.groupBy(_.name).filter(_._2.length > 1).flatMap(_._2).toList |
| val updatedOutputList = outputList.map { |
| output => |
| val duplicateColumn = duplicateColNames.find(name => name.semanticEquals(output)) |
| if (duplicateColumn.isDefined) { |
| val attributesOfDuplicateCol = duplicateColumn.get.collect { |
| case reference: AttributeReference => reference |
| } |
| val attributesOfCol = output.collect { |
| case reference: AttributeReference => reference |
| } |
| // here need to check the whether the duplicate columns is of same tables, |
| // since query with duplicate columns is valid, we need to make sure, not to change their |
| // names with above defined qualifier name, for example in case of some expression like |
| // cast((FLOOR((cast(col_name) as double))).., upper layer even exprid will be same, |
| // we need to find the attribute ref(col_name) at lower level and check where expid is |
| // same or of same tables, so doin the semantic equals |
| val qualifiedName = |
| output.qualifier.headOption.getOrElse(s"${output.exprId.id}_${output.name}") |
| if (!attributesOfDuplicateCol.forall(attribute => |
| attributesOfCol.exists(a => a.semanticEquals(attribute)))) { |
| Alias(output, qualifiedName)(exprId = output.exprId) |
| } else if (output.qualifier.nonEmpty) { |
| Alias(output, qualifiedName)(exprId = output.exprId) |
| // this check is added in scenario where the column is direct Attribute reference and |
| // since duplicate columns select is allowed, we should just put alias for those columns |
| // and update, for this also above isStrictDuplicate will be true so, it will not be |
| // updated above |
| } else if (duplicateColumn.get.isInstanceOf[AttributeReference] && |
| output.isInstanceOf[AttributeReference]) { |
| Alias(output, qualifiedName)(exprId = output.exprId) |
| } else { |
| output |
| } |
| } else { |
| output |
| } |
| } |
| updatedOutputList |
| } |
| |
| /** |
| * Update the modular plan as per the mv table relation inside it. |
| * |
| * @param modularPlan plan to be updated |
| * @return Updated modular plan. |
| */ |
| private def updatePlan(modularPlan: ModularPlan): ModularPlan = { |
| modularPlan match { |
| case select: Select if select.modularPlan.isDefined => |
| val planWrapper = select.modularPlan.get.asInstanceOf[MVPlanWrapper] |
| val plan = planWrapper.modularPlan.asInstanceOf[Select] |
| // when the output list contains multiple projection of same column, but relation |
| // contains distinct columns, mapping may go wrong with columns, so select distinct |
| val updatedPlanOutputList = getUpdatedOutputList(plan.outputList, select.modularPlan) |
| val outputList = |
| for ((output1, output2) <- select.outputList.distinct zip updatedPlanOutputList) yield { |
| if (output1.name != output2.name) { |
| Alias(output2, output1.name)(exprId = output1.exprId) |
| } else { |
| output2 |
| } |
| } |
| plan.copy(outputList = outputList).setRewritten() |
| case select: Select => select.children match { |
| case Seq(groupBy: GroupBy) if groupBy.modularPlan.isDefined => |
| val planWrapper = groupBy.modularPlan.get.asInstanceOf[MVPlanWrapper] |
| val plan = planWrapper.modularPlan.asInstanceOf[Select] |
| val aliasMap = getAliasMap(plan.outputList, groupBy.outputList) |
| // Update the flagSpec as per the mv table attributes. |
| val updatedFlagSpec = updateFlagSpec(select, plan, aliasMap, keepAlias = false) |
| if (!planWrapper.viewSchema.isRefreshIncremental) { |
| val updatedPlanOutputList = getUpdatedOutputList(plan.outputList, groupBy.modularPlan) |
| val outputList = |
| for ((output1, output2) <- groupBy.outputList zip updatedPlanOutputList) yield { |
| if (output1.name != output2.name) { |
| Alias(output2, output1.name)(exprId = output1.exprId) |
| } else { |
| output2 |
| } |
| } |
| // Directly keep the relation as child. |
| select.copy( |
| outputList = select.outputList.map { |
| output => outputList.find(_.name.equals(output.name)).get |
| }, |
| children = Seq(plan), |
| aliasMap = plan.aliasMap, |
| flagSpec = updatedFlagSpec).setRewritten() |
| } else { |
| val child = updatePlan(groupBy).asInstanceOf[Matchable] |
| // First find the indices from the child output list. |
| val outputIndices = select.outputList.map { |
| output => |
| groupBy.outputList.indexWhere { |
| case alias: Alias if output.isInstanceOf[Alias] => |
| alias.child.semanticEquals(output.asInstanceOf[Alias].child) |
| case alias: Alias if alias.child.semanticEquals(output) => |
| true |
| case other if output.isInstanceOf[Alias] => |
| other.semanticEquals(output.asInstanceOf[Alias].child) |
| case other => |
| other.semanticEquals(output) || other.toAttribute.semanticEquals(output) |
| } |
| } |
| // Get the outList from converted child output list using already selected indices |
| val outputList = |
| outputIndices.map(child.outputList(_)).zip(select.outputList).map { |
| case (output1, output2) => |
| output1 match { |
| case alias: Alias if output2.isInstanceOf[Alias] => |
| Alias(alias.child, output2.name)(exprId = output2.exprId) |
| case alias: Alias => |
| alias |
| case other if output2.isInstanceOf[Alias] => |
| Alias(other, output2.name)(exprId = output2.exprId) |
| case other => |
| other |
| } |
| } |
| // TODO Remove the unnecessary columns from selection. |
| // Only keep columns which are required by parent. |
| select.copy( |
| outputList = outputList, |
| inputList = child.outputList, |
| flagSpec = updatedFlagSpec, |
| children = Seq(child)).setRewritten() |
| } |
| case _ => select |
| } |
| case groupBy: GroupBy if groupBy.modularPlan.isDefined => |
| val planWrapper = groupBy.modularPlan.get.asInstanceOf[MVPlanWrapper] |
| val plan = planWrapper.modularPlan.asInstanceOf[Select] |
| val updatedPlanOutputList = getUpdatedOutputList(plan.outputList, groupBy.modularPlan) |
| val outputListMapping = groupBy.outputList zip updatedPlanOutputList |
| val outputList = for ((output1, output2) <- outputListMapping) yield { |
| output1 match { |
| case Alias(aggregate@AggregateExpression(function@Sum(_), _, _, _), _) => |
| val uFun = function.copy(child = output2) |
| Alias(aggregate.copy(aggregateFunction = uFun), output1.name)(exprId = output1.exprId) |
| case Alias(aggregate@AggregateExpression(function@Max(_), _, _, _), _) => |
| val uFun = function.copy(child = output2) |
| Alias(aggregate.copy(aggregateFunction = uFun), output1.name)(exprId = output1.exprId) |
| case Alias(aggregate@AggregateExpression(function@Min(_), _, _, _), _) => |
| val uFun = function.copy(child = output2) |
| Alias(aggregate.copy(aggregateFunction = uFun), output1.name)(exprId = output1.exprId) |
| case Alias(aggregate@AggregateExpression(_@Count(Seq(_)), _, _, _), _) => |
| val uFun = Sum(output2) |
| Alias(aggregate.copy(aggregateFunction = uFun), output1.name)(exprId = output1.exprId) |
| case Alias(agg@AggregateExpression(_@Corr(_, _), _, _, _), _) => |
| val uFun = Sum(output2) |
| Alias(agg.copy(aggregateFunction = uFun), output1.name)(exprId = output1.exprId) |
| case Alias(aggregate@AggregateExpression(_@VariancePop(_), _, _, _), _) => |
| val uFun = Sum(output2) |
| Alias(aggregate.copy(aggregateFunction = uFun), output1.name)(exprId = output1.exprId) |
| case Alias(aggregate@AggregateExpression(_@VarianceSamp(_), _, _, _), _) => |
| val uFun = Sum(output2) |
| Alias(aggregate.copy(aggregateFunction = uFun), output1.name)(exprId = output1.exprId) |
| case Alias(aggregate@AggregateExpression(_@StddevSamp(_), _, _, _), _) => |
| val uFun = Sum(output2) |
| Alias(aggregate.copy(aggregateFunction = uFun), output1.name)(exprId = output1.exprId) |
| case Alias(aggregate@AggregateExpression(_@StddevPop(_), _, _, _), _) => |
| val uFun = Sum(output2) |
| Alias(aggregate.copy(aggregateFunction = uFun), output1.name)(exprId = output1.exprId) |
| case Alias(aggregate@AggregateExpression(_@CovPopulation(_, _), _, _, _), _) => |
| val uFun = Sum(output2) |
| Alias(aggregate.copy(aggregateFunction = uFun), output1.name)(exprId = output1.exprId) |
| case Alias(aggregate@AggregateExpression(_@CovSample(_, _), _, _, _), _) => |
| val uFun = Sum(output2) |
| Alias(aggregate.copy(aggregateFunction = uFun), output1.name)(exprId = output1.exprId) |
| case Alias(aggregate@AggregateExpression(_@Skewness(_), _, _, _), _) => |
| val uFun = Sum(output2) |
| Alias(aggregate.copy(aggregateFunction = uFun), output1.name)(exprId = output1.exprId) |
| case Alias(aggregate@AggregateExpression(_@Kurtosis(_), _, _, _), _) => |
| val uFun = Sum(output2) |
| Alias(aggregate.copy(aggregateFunction = uFun), output1.name)(exprId = output1.exprId) |
| case _ => |
| if (output1.name != output2.name) { |
| Alias(output2, output1.name)(exprId = output1.exprId) |
| } else { |
| output2 |
| } |
| } |
| } |
| val updatedPredicates = groupBy.predicateList.map { |
| predicate => |
| outputListMapping.find { |
| case (output1, _) => |
| output1 match { |
| case alias: Alias if predicate.isInstanceOf[Alias] => |
| alias.child.semanticEquals(predicate.children.head) |
| case alias: Alias => |
| alias.child.semanticEquals(predicate) |
| case other => |
| other.semanticEquals(predicate) |
| } |
| } match { |
| case Some((_, output2)) => output2 |
| case _ => predicate |
| } |
| } |
| groupBy.copy( |
| outputList = outputList, |
| inputList = plan.outputList, |
| predicateList = updatedPredicates, |
| child = plan, |
| modularPlan = None).setRewritten() |
| case other => other |
| } |
| } |
| |
| /** |
| * Updates the flagspec of given select plan with attributes of relation select plan |
| */ |
| private def updateFlagSpec( |
| select: Select, |
| relation: Select, |
| aliasMap: Map[AttributeKey, NamedExpression], |
| keepAlias: Boolean): Seq[Seq[Any]] = { |
| val updatedFlagSpec = select.flagSpec.map { |
| flagSpec => |
| flagSpec.map { |
| case list: ArrayBuffer[_] => |
| list.map { |
| case sortOrder: SortOrder => |
| val expressions = |
| updateOutputList( |
| Seq(sortOrder.child.asInstanceOf[Attribute]), |
| relation, |
| aliasMap, |
| keepAlias = false) |
| SortOrder(expressions.head, sortOrder.direction) |
| } |
| // In case of limit it goes to other. |
| case other => other |
| } |
| } |
| updatedFlagSpec |
| } |
| |
| private def updateOutputList( |
| outputList: Seq[NamedExpression], |
| select: Select, |
| aliasMap: Map[AttributeKey, NamedExpression], |
| keepAlias: Boolean): Seq[NamedExpression] = { |
| val updatedOutputList = updateSubsumeAttributes( |
| outputList, |
| aliasMap, |
| Some(select.aliasMap.values.head), |
| keepAlias).asInstanceOf[Seq[NamedExpression]] |
| updatedOutputList.zip(outputList).map { |
| case (updatedOutput, output) => |
| updatedOutput match { |
| case reference: AttributeReference => |
| Alias(reference, output.name)(output.exprId) |
| case Alias(reference: AttributeReference, _) => |
| Alias(reference, output.name)(output.exprId) |
| case other => other |
| } |
| } |
| } |
| |
| /** |
| * Updates the expressions as per the subsumer output expressions. It is needed to update the |
| * expressions as per the mv table relation |
| * |
| * @param expressions expressions which are needed to update |
| * @param aliasName table alias name |
| * @return Updated expressions |
| */ |
| private def updateSubsumeAttributes( |
| expressions: Seq[Expression], |
| attributeMap: Map[AttributeKey, NamedExpression], |
| aliasName: Option[String], |
| keepAlias: Boolean = false): Seq[Expression] = { |
| def getAttribute(expression: Expression) = { |
| expression match { |
| case Alias(aggregate: AggregateExpression, _) => |
| aggregate.aggregateFunction.collect { |
| case reference: AttributeReference => |
| ExpressionHelper.createReference(reference.name, |
| reference.dataType, |
| reference.nullable, |
| reference.metadata, |
| reference.exprId, |
| aliasName) |
| }.head |
| case Alias(child, _) => child |
| case other => other |
| } |
| } |
| |
| expressions.map { |
| case alias@Alias(agg: AggregateExpression, name) => |
| attributeMap.get(AttributeKey(agg)).map { |
| expression => |
| Alias(getAttribute(expression), name)( |
| alias.exprId, |
| alias.qualifier, |
| alias.explicitMetadata) |
| }.getOrElse(alias) |
| case reference: AttributeReference => |
| attributeMap.get(AttributeKey(reference)).map { |
| expression => |
| if (keepAlias) { |
| AttributeReference( |
| name = expression.name, |
| dataType = expression.dataType, |
| nullable = expression.nullable, |
| metadata = expression.metadata)( |
| exprId = expression.exprId, |
| qualifier = reference.qualifier) |
| } else { |
| expression |
| } |
| }.getOrElse(reference) |
| case alias@Alias(expression: Expression, name) => |
| attributeMap.get(AttributeKey(expression)).map { |
| expression => |
| Alias(getAttribute(expression), name)(alias.exprId, alias.qualifier, |
| alias.explicitMetadata) |
| }.getOrElse(alias) |
| case expression: Expression => |
| attributeMap.getOrElse(AttributeKey(expression), expression) |
| } |
| } |
| |
| private val subqueryNameGenerator: SubqueryNameGenerator = new SubqueryNameGenerator() |
| |
| private val rewrittenPlans: java.util.Set[ModularPlan] = new java.util.HashSet[ModularPlan]() |
| |
| lazy val optimizedPlan: LogicalPlan = BirdcageOptimizer.execute(logicalPlan) |
| |
| lazy val modularPlan: ModularPlan = SimpleModularizer.modularize(optimizedPlan).next() |
| |
| lazy val harmonizedPlan: ModularPlan = modularPlan.harmonized |
| |
| lazy val rewrittenPlan: ModularPlan = rewrite(rewriteWithSchemaWrapper(harmonizedPlan)) |
| |
| lazy val toCompactSQL: String = rewriteWithSchemaWrapper(modularPlan).asCompactSQL |
| |
| lazy val toOneLineSQL: String = rewriteWithSchemaWrapper(modularPlan).asOneLineSQL |
| |
| private case class AttributeKey(expression: Expression) { |
| |
| override def equals(other: Any): Boolean = { |
| other match { |
| case that: AttributeKey => |
| expression.semanticEquals(that.expression) |
| case _ => false |
| } |
| } |
| |
| // Basically we want to use it as simple linked list so hashcode is hardcoded. |
| override def hashCode: Int = { |
| 1 |
| } |
| |
| } |
| |
| } |
| |
| private class SubqueryNameGenerator { |
| |
| private val nextSubqueryId = new AtomicLong(0) |
| |
| def newSubsumerName(): String = s"gen_subsumer_${nextSubqueryId.getAndIncrement()}" |
| |
| } |
