Repository: carbondata Updated Branches: refs/heads/master 5f2a748f6 -> 8034949e5
[CARBONDATA-2328][PreAggregate] Fixed table With alias Issue Issue: Query with table alias is not fetching data from pre-aggregate Problem: when table has alias all the attribute reference's qualifiers will have alias name, but as data map is created without alias so expression comparison is failing. Solution: While comparing alias remove qualifiers and then compare expressions This closes #2153 Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/8034949e Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/8034949e Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/8034949e Branch: refs/heads/master Commit: 8034949e593ce555cb609811af66a2835c9fd212 Parents: 5f2a748 Author: kumarvishal <[email protected]> Authored: Mon Apr 9 19:23:04 2018 +0800 Committer: ravipesala <[email protected]> Committed: Thu Apr 19 08:50:02 2018 +0530 ---------------------------------------------------------------------- .../preaggregate/TestPreAggStreaming.scala | 29 +++ .../preaggregate/TestPreAggregateLoad.scala | 2 +- .../TestPreAggregateTableSelection.scala | 5 + .../preaaggregate/PreAggregateUtil.scala | 17 +- .../sql/hive/CarbonPreAggregateRules.scala | 197 +++++++++++++++++-- 5 files changed, 226 insertions(+), 24 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/carbondata/blob/8034949e/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggStreaming.scala ---------------------------------------------------------------------- diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggStreaming.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggStreaming.scala index 0b644f5..9377108 100644 --- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggStreaming.scala +++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggStreaming.scala @@ -28,12 +28,16 @@ class TestPreAggStreaming extends QueryTest with BeforeAndAfterAll { override def beforeAll: Unit = { sql("drop table if exists mainTable") + sql("drop table if exists mainTableStreamingOne") sql("CREATE TABLE mainTable(id int, name string, city string, age string) STORED BY 'org.apache.carbondata.format' tblproperties('streaming'='true')") sql("create datamap agg0 on table mainTable using 'preaggregate' as select name from mainTable group by name") sql("create datamap agg1 on table mainTable using 'preaggregate' as select name,sum(age) from mainTable group by name") sql("create datamap agg2 on table mainTable using 'preaggregate' as select name,avg(age) from mainTable group by name") sql("create datamap agg3 on table mainTable using 'preaggregate' as select name,sum(CASE WHEN age=35 THEN id ELSE 0 END) from mainTable group by name") + sql("CREATE TABLE mainTableStreamingOne(id int, name string, city string, age smallint) STORED BY 'org.apache.carbondata.format' tblproperties('streaming'='true')") + sql("create datamap aggStreamingAvg on table mainTableStreamingOne using 'preaggregate' as select name,avg(age) from mainTableStreamingOne group by name") sql(s"LOAD DATA LOCAL INPATH '$resourcesPath/measureinsertintotest.csv' into table mainTable") + sql(s"LOAD DATA LOCAL INPATH '$resourcesPath/measureinsertintotest.csv' into table mainTableStreamingOne") } test("Test Pre Agg Streaming with project column and group by") { @@ -48,6 +52,18 @@ class TestPreAggStreaming extends QueryTest with BeforeAndAfterAll { assert(validateStreamingTablePlan(df.queryExecution.analyzed)) } + test("Test Pre Agg Streaming table with UDF") { + val df = sql("select substring(name,1,1), sum(age) from maintable group by substring(name,1,1)") + df.collect() + assert(validateStreamingTablePlan(df.queryExecution.analyzed)) + } + + test("Test Pre Agg Streaming table with UDF Only in group by") { + val df = sql("select sum(age) from maintable group by substring(name,1,1)") + df.collect() + assert(validateStreamingTablePlan(df.queryExecution.analyzed)) + } + test("Test Pre Agg Streaming table With Sum Aggregation And Order by") { val df = sql("select name, sum(age) from maintable group by name order by name") df.collect() @@ -66,6 +82,18 @@ class TestPreAggStreaming extends QueryTest with BeforeAndAfterAll { assert(validateStreamingTablePlan(df.queryExecution.analyzed)) } + test("Test Pre Agg Streaming table With only aggregate expression and group by") { + val df = sql("select sum(age) from maintable group by name") + df.collect() + assert(validateStreamingTablePlan(df.queryExecution.analyzed)) + } + + test("Test Pre Agg Streaming table With small int and avg") { + val df = sql("select name, avg(age) from mainTableStreamingOne group by name") + df.collect() + assert(validateStreamingTablePlan(df.queryExecution.analyzed)) + } + /** * 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 @@ -93,5 +121,6 @@ class TestPreAggStreaming extends QueryTest with BeforeAndAfterAll { override def afterAll: Unit = { sql("drop table if exists mainTable") + sql("drop table if exists mainTableStreamingOne") } } http://git-wip-us.apache.org/repos/asf/carbondata/blob/8034949e/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateLoad.scala ---------------------------------------------------------------------- diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateLoad.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateLoad.scala index 959da7e..55994e8 100644 --- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateLoad.scala +++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateLoad.scala @@ -542,7 +542,7 @@ test("check load and select for avg double datatype") { checkAnswer(sql(s"SELECT * FROM main_table_preagg_sum"), Seq(Row(1, null), Row(2, null), Row(3, null), Row(4, null))) checkAnswer(sql(s"SELECT * FROM main_table_preagg_avg"), - Seq(Row(1, null, 0), Row(2, null, 0), Row(3, null, 0), Row(4, null, 0))) + Seq(Row(1, null, 1.0), Row(2, null, 1.0), Row(3, null, 2.0), Row(4, null, 2.0))) checkAnswer(sql(s"SELECT * FROM main_table_preagg_count"), Seq(Row(1, 1), Row(2, 1), Row(3, 2), Row(4, 2))) checkAnswer(sql(s"SELECT * FROM main_table_preagg_min"), http://git-wip-us.apache.org/repos/asf/carbondata/blob/8034949e/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateTableSelection.scala ---------------------------------------------------------------------- diff --git a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateTableSelection.scala b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateTableSelection.scala index 8b98f17..95a524d 100644 --- a/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateTableSelection.scala +++ b/integration/spark-common-test/src/test/scala/org/apache/carbondata/integration/spark/testsuite/preaggregate/TestPreAggregateTableSelection.scala @@ -88,6 +88,11 @@ class TestPreAggregateTableSelection extends SparkQueryTest with BeforeAndAfterA preAggTableValidator(df.queryExecution.analyzed, "maintable_agg1") } + test("test PreAggregate table selection with table alias") { + val df = sql("select name, sum(age) from mainTable as t1 group by name") + preAggTableValidator(df.queryExecution.analyzed, "maintable_agg1") + } + test("test PreAggregate table selection 6") { val df = sql("select sum(age) from mainTable group by name") preAggTableValidator(df.queryExecution.analyzed, "maintable_agg1") http://git-wip-us.apache.org/repos/asf/carbondata/blob/8034949e/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateUtil.scala ---------------------------------------------------------------------- diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateUtil.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateUtil.scala index 04e2135..9b1f238 100644 --- a/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateUtil.scala +++ b/integration/spark2/src/main/scala/org/apache/spark/sql/execution/command/preaaggregate/PreAggregateUtil.scala @@ -745,8 +745,8 @@ object PreAggregateUtil { * @param aggExp aggregate expression * @return list of fields */ - def validateAggregateFunctionAndGetFields(aggExp: AggregateExpression): - Seq[AggregateExpression] = { + def validateAggregateFunctionAndGetFields(aggExp: AggregateExpression, + addCastForCount: Boolean = true): Seq[AggregateExpression] = { aggExp.aggregateFunction match { case Sum(MatchCastExpression(exp: Expression, changeDataType: DataType)) => Seq(AggregateExpression(Sum(Cast( @@ -783,16 +783,23 @@ object PreAggregateUtil { // in case of average need to return two columns // sum and count of the column to added during table creation to support rollup case Average(MatchCastExpression(exp: Expression, changeDataType: DataType)) => - Seq(AggregateExpression(Sum(Cast( + val sum = AggregateExpression(Sum(Cast( exp, changeDataType)), aggExp.mode, - aggExp.isDistinct), + aggExp.isDistinct) + val count = if (!addCastForCount) { + AggregateExpression(Count(exp), + aggExp.mode, + aggExp.isDistinct) + } else { AggregateExpression(Count(Cast( exp, changeDataType)), aggExp.mode, - aggExp.isDistinct)) + aggExp.isDistinct) + } + Seq(sum, count) // in case of average need to return two columns // sum and count of the column to added during table creation to support rollup case Average(exp: Expression) => http://git-wip-us.apache.org/repos/asf/carbondata/blob/8034949e/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonPreAggregateRules.scala ---------------------------------------------------------------------- diff --git a/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonPreAggregateRules.scala b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonPreAggregateRules.scala index 5c553b9..ab8ec30 100644 --- a/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonPreAggregateRules.scala +++ b/integration/spark2/src/main/scala/org/apache/spark/sql/hive/CarbonPreAggregateRules.scala @@ -19,6 +19,7 @@ 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 @@ -657,23 +658,46 @@ case class CarbonPreAggregateQueryRules(sparkSession: SparkSession) extends Rule // 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( - grExp, + createNewAggGroupBy(grExp, factAggPlan), streamingNodeExp.asInstanceOf[Seq[NamedExpression]], // add union node to get the result from both Union( factExp, - aggPlan)) + 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 @@ -722,6 +746,9 @@ case class CarbonPreAggregateQueryRules(sparkSession: SparkSession) extends Rule */ 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 @@ -773,22 +800,112 @@ case class CarbonPreAggregateQueryRules(sparkSession: SparkSession) extends Rule private def updateFactTablePlanForStreaming(logicalPlan: LogicalPlan) : LogicalPlan = { // only aggregate expression needs to be updated logicalPlan.transform{ - case agg@Aggregate(_, aggExp, _) => + case agg@Aggregate(grpExp, aggExp, _) => agg - .copy(aggregateExpressions = updateAggExpInFactForStreaming(aggExp) + .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 += Alias( + exp, + "dummy_" + counter)(NamedExpression.newExprId, None, None, false) + 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]) : Seq[Expression] = { - val updatedExp = namedExp.flatMap { + 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) => @@ -796,18 +913,28 @@ case class CarbonPreAggregateQueryRules(sparkSession: SparkSession) extends Rule val newAggExp = getAggFunctionForFactStreaming(aggExp) val updatedExp = newAggExp.map { exp => Alias(exp, - name)( - NamedExpression.newExprId, - alias.qualifier, + name)( + NamedExpression.newExprId, + alias.qualifier, Some(alias.metadata), - alias.isGenerated) + alias.isGenerated) } // 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(exp: Expression, _) => - Seq(exp) + case alias@Alias(exp: Expression, name) => + val newAlias = Seq(alias) + val attr = AttributeReference(name, + alias.dataType, + alias.nullable, + alias.metadata) (alias.exprId, alias.qualifier, alias.isGenerated) + factPlanGrpExpForStreaming.put( + AggExpToColumnMappingModel( + removeQualifiers(PreAggregateUtil.normalizeExprId(exp, plan.allAttributes))), + attr) + factPlanExpForStreaming.put(name, newAlias) + newAlias } updatedExp } @@ -823,13 +950,13 @@ case class CarbonPreAggregateQueryRules(sparkSession: SparkSession) extends Rule def getAggFunctionForFactStreaming(aggExp: AggregateExpression): Seq[Expression] = { aggExp.aggregateFunction match { case Average(MatchCastExpression(exp: Expression, changeDataType: DataType)) => - val newExp = Seq(AggregateExpression(Sum(Cast(exp, changeDataType)), + 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 newExp = Seq(AggregateExpression(Sum(exp), aggExp.mode, false), + val newExp = Seq(AggregateExpression(Sum(Cast(exp, DoubleType)), aggExp.mode, false), Cast(AggregateExpression(Count(exp), aggExp.mode, false), DoubleType)) newExp case _ => @@ -939,11 +1066,45 @@ case class CarbonPreAggregateQueryRules(sparkSession: SparkSession) extends Rule parentLogicalPlan) queryAggExpLogicalPlans.forall{p => mappingModel.exists{m => - PreAggregateUtil.normalizeExprId(p, parentLogicalPlan.allAttributes) == m.expression} + 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 => + AttributeReference( + attr.name, + attr.dataType, + attr.nullable, + attr.metadata)(attr.exprId, None, attr.isGenerated) } } /** + * 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 @@ -1357,8 +1518,8 @@ case class CarbonPreAggregateQueryRules(sparkSession: SparkSession) extends Rule case (schemaAggExpModel) if updatedAggExp .exists(p => - schemaAggExpModel.expression == - PreAggregateUtil.normalizeExprId(p, parentLogicalPlan.allAttributes)) => + matchExpression(schemaAggExpModel.expression, + PreAggregateUtil.normalizeExprId(p, parentLogicalPlan.allAttributes))) => attributes filter (_.name.equalsIgnoreCase( schemaAggExpModel.columnSchema.get.asInstanceOf[ColumnSchema].getColumnName)) }.flatten @@ -1601,7 +1762,7 @@ case class CarbonPreAggregateDataLoadingRules(sparkSession: SparkSession) 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) + val expressions = PreAggregateUtil.validateAggregateFunctionAndGetFields(aggExp, false) // if size is more than one then it was for average if(expressions.size > 1) { val sumExp = PreAggregateUtil.normalizeExprId(
