Repository navigation
[SPARK-58387][SQL] Do not pre-aggregate under an Expand when a duplicate-sensitive aggregate is present - #58102
Vivek1106-04 wants to merge 1 commit into
Conversation
…ate-sensitive aggregate is present OptimizeExpand inserts a de-duplicating Aggregate beneath the Expand produced by RewriteDistinctAggregates. That is sound only for pure distinct aggregates, and the rule's guard implements the precondition by requiring every Expand-produced attribute to be consumed by the inner GROUP BY. An aggregate that references no attribute, such as count(1) from COUNT(1) or COUNT(*), gets no dedicated Expand slot and therefore always passes that check. The inserted pre-aggregate then makes it count distinct rows instead of base rows, returning a wrong result. Reject duplicate-sensitive aggregates directly by checking the inner aggregate's aggregate expressions, using EliminateDistinct.isDuplicateAgnostic, the same soundness check RemoveRedundantAggregates uses.
a02c2fa to
40eb7cd
Compare
|
Thank you @Vivek1106-04! |
|
@uros-b thanks for the reviews. One process question, since conventions differ across ASF projects: once a PR is approved and CI is green, is there anything expected from the contributor's side, or is it purely waiting for a committer's merge window? Just want to make sure I'm not sitting on something. |
|
The fix is targeted and well scoped, reuses the established isDuplicateAgnostic predicate consistently with RemoveRedundantAggregates, and is verified against the JIRA reproduction case as well as the count(*) variant. The three new tests cover the reported scenarios; the broader duplicate-sensitive aggregate class (e.g., non-distinct SUM over a grouping key) also benefits silently from the fix. This should be ready to merge. Adding @cloud-fan who is familiar with Optimizer and EliminateDistinct, to validate reuse of that predicate here and review this PR as well. |
What changes were proposed in this pull request?
OptimizeExpand(added by SPARK-56315, gated by the internal confspark.sql.optimizer.optimizeExpandRatio, default-1= disabled) inserts a de-duplicatingAggregatebeneath theExpandproduced byRewriteDistinctAggregates. That de-duplication is sound only for pure distinct aggregates, and the conf's own doc states the precondition: "Only applies to pure distinct aggregates without non-distinct aggregates or FILTER clauses."The guard implementing that precondition tested attributes rather than aggregate functions:
The reasoning is that a non-distinct aggregate forces an
Expandoutput column outside the innerGROUP BY. That only holds for aggregates to whichRewriteDistinctAggregatesassigns a dedicatedExpandslot. An aggregate that reaches the innerAggregateascount(1)— fromCOUNT(1)orCOUNT(*)— references no attribute at all, so it always passes the guard.This PR rejects duplicate-sensitive aggregates directly, by checking the inner aggregate's aggregate expressions with
EliminateDistinct.isDuplicateAgnostic— the same soundness checkRemoveRedundantAggregatesuses for the same question:Why are the changes needed?
It is a correctness bug: the query returns the count of distinct rows where it must return the count of base rows.
COUNT(*)behaves identically. The optimized plan before the fix shows the inserted pre-aggregate feeding theExpandwhose downstreamcount(1)was supposed to count base rows:Note on the JIRA: it also lists
COUNT(a)on a non-nullable column as affected, on the grounds that the rewrite normalizes it tocount(1). That does not reproduce on master — no such normalization happens,RewriteDistinctAggregatesgives it its ownExpandslot, and the existing attribute check already rejects it. A test for that case was written and passed before the fix, so it was dropped rather than added as a non-regression test.Does this PR introduce any user-facing change?
Yes, it fixes wrong results for the queries above. Only when
spark.sql.optimizer.optimizeExpandRatiois explicitly set, since the rule is disabled by default.How was this patch tested?
New tests, verified failing before the fix and passing after:
OptimizeExpandSuite— "SPARK-58387: skips when a non-distinct count(1) is present": asserts no pre-aggregate is inserted.OptimizeExpandQuerySuite— "SPARK-58387: correctness: count distinct with a non-distinct count(1)": the JIRA repro, checked against the expected answer.OptimizeExpandQuerySuite— "SPARK-58387: correctness: count distinct with a non-distinct count(*)": checked against the rule-disabled result.Full suites pass:
OptimizeExpandSuite(9 tests) andOptimizeExpandQuerySuite(10 tests).Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 5)