You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@carbondata.apache.org by ra...@apache.org on 2017/11/30 15:22:51 UTC
carbondata git commit: [CARBONDATA-1757] [PreAgg] Fix for wrong avg
values after pre-agg table creation
Repository: carbondata
Updated Branches:
refs/heads/master 80329eb5f -> 95a395112
[CARBONDATA-1757] [PreAgg] Fix for wrong avg values after pre-agg table creation
when a sum/count aggregation function is applied on the same column along with avg. The plan that was getting transformed was adding 2 columns for sum/count which resulted in wrong data being inserted.
This closes #1542
Project: http://git-wip-us.apache.org/repos/asf/carbondata/repo
Commit: http://git-wip-us.apache.org/repos/asf/carbondata/commit/95a39511
Tree: http://git-wip-us.apache.org/repos/asf/carbondata/tree/95a39511
Diff: http://git-wip-us.apache.org/repos/asf/carbondata/diff/95a39511
Branch: refs/heads/master
Commit: 95a3951126c0a2c95bac42e909aa887a6537edf5
Parents: 80329eb
Author: kunal642 <ku...@gmail.com>
Authored: Tue Nov 21 12:55:25 2017 +0530
Committer: ravipesala <ra...@gmail.com>
Committed: Thu Nov 30 20:52:04 2017 +0530
----------------------------------------------------------------------
.../TestPreAggregateTableSelection.scala | 11 +++-
.../preaaggregate/PreAggregateUtil.scala | 3 +-
.../sql/hive/CarbonPreAggregateRules.scala | 68 +++++++++++++-------
3 files changed, 57 insertions(+), 25 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/carbondata/blob/95a39511/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 68518c4..1480ae3 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
@@ -14,11 +14,12 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.carbondata.integration.spark.testsuite.preaTable1regate
+
+package org.apache.carbondata.integration.spark.testsuite.preaggregate
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
import org.apache.spark.sql.execution.datasources.LogicalRelation
-import org.apache.spark.sql._
+import org.apache.spark.sql.{CarbonDatasourceHadoopRelation, Row}
import org.apache.spark.sql.test.util.QueryTest
import org.scalatest.BeforeAndAfterAll
@@ -44,11 +45,17 @@ class TestPreAggregateTableSelection extends QueryTest with BeforeAndAfterAll {
sql("create datamap agg5 on table mainTable using 'preaggregate' as select name,avg(age) from mainTable group by name")
sql("create datamap agg6 on table mainTable using 'preaggregate' as select name,min(age) from mainTable group by name")
sql("create datamap agg7 on table mainTable using 'preaggregate' as select name,max(age) from mainTable group by name")
+ sql("create datamap agg8 on table maintable using 'preaggregate' as select name, sum(id), avg(id) from maintable group by name")
sql(s"LOAD DATA LOCAL INPATH '$resourcesPath/measureinsertintotest.csv' into table mainTable")
sql("create table if not exists lineitem(L_SHIPDATE string,L_SHIPMODE string,L_SHIPINSTRUCT string,L_RETURNFLAG string,L_RECEIPTDATE string,L_ORDERKEY string,L_PARTKEY string,L_SUPPKEY string,L_LINENUMBER int,L_QUANTITY double,L_EXTENDEDPRICE double,L_DISCOUNT double,L_TAX double,L_LINESTATUS string,L_COMMITDATE string,L_COMMENT string) STORED BY 'org.apache.carbondata.format'TBLPROPERTIES ('table_blocksize'='128','NO_INVERTED_INDEX'='L_SHIPDATE,L_SHIPMODE,L_SHIPINSTRUCT,L_RETURNFLAG,L_RECEIPTDATE,L_ORDERKEY,L_PARTKEY,L_SUPPKEY','sort_columns'='')")
sql("create datamap agr_lineitem ON TABLE lineitem USING 'preaggregate' as select L_RETURNFLAG,L_LINESTATUS,sum (L_QUANTITY),sum(L_EXTENDEDPRICE) from lineitem group by L_RETURNFLAG, L_LINESTATUS")
}
+ test("test sum and avg on same column should give proper results") {
+ val df = sql("select name, sum(id), avg(id) from maintable group by name")
+ checkAnswer(df, Seq(Row("david",1,1.0), Row("jarry",6,3.0), Row("kunal",4,4.0), Row("eason",2,2.0),Row("vishal",4,4.0)))
+ }
+
test("test PreAggregate table selection 1") {
val df = sql("select name from mainTable group by name")
http://git-wip-us.apache.org/repos/asf/carbondata/blob/95a39511/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 a2809ed..c50e717 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
@@ -297,7 +297,7 @@ object PreAggregateUtil {
precision = precision,
scale = scale,
rawSchema = rawSchema), dataMapField)
-} else {
+ } else {
(Field(column = actualColumnName,
dataType = Some(dataType.typeName),
name = Some(actualColumnName),
@@ -484,4 +484,5 @@ object PreAggregateUtil {
}
updatedPlan
}
+
}
http://git-wip-us.apache.org/repos/asf/carbondata/blob/95a39511/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 1bb6ab0..baa9008 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
@@ -27,6 +27,7 @@ import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeRef
import org.apache.spark.sql.catalyst.expressions.aggregate._
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.LogicalRelation
import org.apache.spark.sql.types._
import org.apache.spark.sql.util.CarbonException
@@ -822,33 +823,56 @@ case class CarbonPreInsertionCasts(sparkSession: SparkSession) extends Rule[Logi
* @return
*/
private def transformAggregatePlan(logicalPlan: LogicalPlan): LogicalPlan = {
+ val validExpressionsMap = scala.collection.mutable.LinkedHashMap.empty[String, NamedExpression]
logicalPlan transform {
case aggregate@Aggregate(_, aExp, _) =>
- val newExpressions = aExp.flatMap {
- case alias@Alias(attrExpression: AggregateExpression, _) =>
- attrExpression.aggregateFunction match {
- case Average(attr: AttributeReference) =>
- Seq(Alias(attrExpression
- .copy(aggregateFunction = Sum(attr),
- resultId = NamedExpression.newExprId), attr.name + "_sum")(),
- Alias(attrExpression
- .copy(aggregateFunction = Count(attr),
- resultId = NamedExpression.newExprId), attr.name + "_count")())
- case Average(cast@MatchCast(attr: AttributeReference, _)) =>
- Seq(Alias(attrExpression
- .copy(aggregateFunction = Sum(cast),
- resultId = NamedExpression.newExprId),
- attr.name + "_sum")(),
- Alias(attrExpression
- .copy(aggregateFunction = Count(cast),
- resultId = NamedExpression.newExprId), attr.name + "_count")())
- case _ => Seq(alias)
- }
- case namedExpr: NamedExpression => Seq(namedExpr)
+ aExp.foreach {
+ case alias: Alias =>
+ validExpressionsMap ++= validateAggregateFunctionAndGetAlias(alias)
+ case namedExpr: NamedExpression => validExpressionsMap.put(namedExpr.name, namedExpr)
}
- aggregate.copy(aggregateExpressions = newExpressions.asInstanceOf[Seq[NamedExpression]])
+ aggregate.copy(aggregateExpressions = validExpressionsMap.values.toSeq)
case plan: LogicalPlan => plan
}
}
+
+ /**
+ * This method will split the avg column into sum and count and will return a sequence of tuple
+ * of unique name, alias
+ *
+ */
+ def validateAggregateFunctionAndGetAlias(alias: Alias): Seq[(String, NamedExpression)] = {
+ alias match {
+ case alias@Alias(attrExpression: AggregateExpression, _) =>
+ attrExpression.aggregateFunction match {
+ case Sum(attr: AttributeReference) =>
+ (attr.name + "_sum", alias) :: Nil
+ case Sum(Cast(attr: AttributeReference, _)) =>
+ (attr.name + "_sum", alias) :: Nil
+ case Count(Seq(attr: AttributeReference)) =>
+ (attr.name + "_count", alias) :: Nil
+ case Count(Seq(Cast(attr: AttributeReference, _))) =>
+ (attr.name + "_count", alias) :: Nil
+ case Average(attr: AttributeReference) =>
+ Seq((attr.name + "_sum", Alias(attrExpression
+ .copy(aggregateFunction = Sum(attr),
+ resultId = NamedExpression.newExprId), attr.name + "_sum")()),
+ (attr.name, Alias(attrExpression
+ .copy(aggregateFunction = Count(attr),
+ resultId = NamedExpression.newExprId), attr.name + "_count")()))
+ case Average(cast@Cast(attr: AttributeReference, _)) =>
+ Seq((attr.name + "_sum", Alias(attrExpression
+ .copy(aggregateFunction = Sum(cast),
+ resultId = NamedExpression.newExprId),
+ attr.name + "_sum")()),
+ (attr.name, Alias(attrExpression
+ .copy(aggregateFunction = Count(cast),
+ resultId = NamedExpression.newExprId), attr.name + "_count")()))
+ case _ => Seq(("", alias))
+ }
+
+ }
+ }
+
}