Skip to content

[SPARK-36339][SQL] References to grouping that not part of aggregation should be replaced #33574

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
gaoyajun02 wants to merge 2 commits into from
Closed

[SPARK-36339][SQL] References to grouping that not part of aggregation should be replaced #33574

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
f369a62
[SPARK-36339] References to grouping that not part of aggregation sho…
Jul 29, 2021
321374d
Remove irrelevant queries in unit tests
gaoyajun02 Aug 5, 2021
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
4 changes: 2 additions & 2 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -580,15 +580,15 @@ class Analyzer(override val catalogManager: CatalogManager)
aggregations: Seq[NamedExpression],
groupByAliases: Seq[Alias],
groupingAttrs: Seq[Expression],
gid: Attribute): Seq[NamedExpression] = aggregations.map {
gid: Attribute): Seq[NamedExpression] = aggregations.map { agg =>
// collect all the found AggregateExpression, so we can check an expression is part of
// any AggregateExpression or not.
val aggsBuffer = ArrayBuffer[Expression]()
// Returns whether the expression belongs to any expressions in `aggsBuffer` or not.
def isPartOfAggregation(e: Expression): Boolean = {
aggsBuffer.exists(a => a.find(_ eq e).isDefined)
}
replaceGroupingFunc(_, groupByExprs, gid).transformDown {
replaceGroupingFunc(agg, groupByExprs, gid).transformDown {
Copy link
Contributor

@cfmcgrady cfmcgrady Jul 29, 2021 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What's the difference between underscore?

Copy link
Contributor Author

@gaoyajun02 gaoyajun02 Jul 29, 2021 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Using the underscore, the aggsBuffer is outside the scope of the map function at runtime and it will save the results of all elements.
using normal parameters, the aggsBuffer will only be recreated each time inside the map function loop.

I suspect that Scala syntactic sugar in the conversion of the code made changes to cause, I also debugged this code many times before I found this difference, here is a simplified code to test separately.

    def testMap(seq: Seq[Int]): Seq[Int] = {
      seq.map {
        val buf = ArrayBuffer[Int]()
        _ match {
          case e: Int if e < 1 =>
            val r = e + 1
            println(s"add to buf: $r")
            buf += r
            r
          case e: Int if buf.contains(e) =>
            println("already in buf")
            0
          case e =>
            println("not in buf")
            e
        }
      }
    }

    testMap(Seq(0, 1))

Copy link
Contributor

@cfmcgrady cfmcgrady Jul 29, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thank you for the detailed explanation.

// AggregateExpression should be computed on the unmodified value of its argument
// expressions, so we should not replace any references to grouping expression
// inside it.
Expand Down
13 changes: 13 additions & 0 deletions sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3405,6 +3405,19 @@ class SQLQuerySuite extends QueryTest with SharedSparkSession with AdaptiveSpark
}
}

test("SPARK-36339: References to grouping attributes should be replaced") {
withTempView("t") {
Seq("a", "a", "b").toDF("x").createOrReplaceTempView("t")
checkAnswer(
sql(
"""
|select count(x) c, x from t
|group by x grouping sets(x)
""".stripMargin),
Seq(Row(2, "a"), Row(1, "b")))
}
}

test("SPARK-31166: UNION map<null, null> and other maps should not fail") {
checkAnswer(
sql("(SELECT map()) UNION ALL (SELECT map(1, 2))"),
Expand Down