Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -151,15 +151,24 @@ object PartialAggregation {

// Replace aggregations with a new expression that computes the result from the already
// computed partial evaluations and grouping values.
val rewrittenAggregateExpressions = aggregateExpressions.map(_.transformUp {
// transformDown is needed at here because we want to match aggregate function first.
// Otherwise, if a grouping expression is used as an argument of an aggregate function,
// we will match grouping expression first and have a wrong plan.
val rewrittenAggregateExpressions = aggregateExpressions.map(_.transformDown {
case e: Expression if partialEvaluations.contains(new TreeNodeRef(e)) =>
partialEvaluations(new TreeNodeRef(e)).finalEvaluation

case e: Expression =>
// Should trim aliases around `GetField`s. These aliases are introduced while
// resolving struct field accesses, because `GetField` is not a `NamedExpression`.
// (Should we just turn `GetField` into a `NamedExpression`?)
val trimmed = e.transform { case Alias(g: ExtractValue, _) => g }
def trimAliases(e: Expression): Expression =
e.transform { case Alias(g: ExtractValue, _) => g }
val trimmed = e match {
// Don't trim the top level Alias.
case Alias(child, name) => Alias(trimAliases(child), name)()
case _ => trimAliases(e)
}
namedGroupingExpressions
.find { case (k, v) => k semanticEquals trimmed }
.map(_._2.toAttribute)
Expand Down
22 changes: 22 additions & 0 deletions sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
Original file line number Diff line number Diff line change
Expand Up @@ -1335,4 +1335,26 @@ class SQLQuerySuite extends QueryTest with BeforeAndAfterAll {

checkAnswer(sql("SELECT a.`c.b`, `b.$q`[0].`a@!.q`, `q.w`.`w.i&`[0] FROM t"), Row(1, 1, 1))
}

test("SPARK-10169: grouping expressions used as arguments of aggregate functions.") {
sqlCtx.sparkContext
.parallelize((1 to 1000), 50)
.map(i => Tuple1(i))
.toDF("i")
.registerTempTable("t")

val query = sqlCtx.sql(
"""
|select i % 10, sum(if(i % 10 = 5, 1, 0)), count(i)
|from t
|where i % 10 = 5
|group by i % 10
""".stripMargin)

checkAnswer(
query,
Row(5, 100, 100))

dropTempTable("t")
}
}