Skip to content

[SPARK-58387][SQL] Do not pre-aggregate under an Expand when a duplicate-sensitive aggregate is present - #58102

Open
Vivek1106-04 wants to merge 1 commit into
apache:masterfrom
Vivek1106-04:SPARK-58387-optimize-expand
Open

Vivek1106-04 wants to merge 1 commit into
apache:masterfrom
Vivek1106-04:SPARK-58387-optimize-expand

Conversation

@Vivek1106-04

Copy link
Copy Markdown
Contributor

What changes were proposed in this pull request?

OptimizeExpand (added by SPARK-56315, gated by the internal conf spark.sql.optimizer.optimizeExpandRatio, default -1 = disabled) inserts a de-duplicating Aggregate beneath the Expand produced by RewriteDistinctAggregates. 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:

val innerGroupByAttrs = AttributeSet(innerAgg.groupingExpressions.flatMap(_.references))
if (!expand.producedAttributes.subsetOf(innerGroupByAttrs)) return false

The reasoning is that a non-distinct aggregate forces an Expand output column outside the inner GROUP BY. That only holds for aggregates to which RewriteDistinctAggregates assigns a dedicated Expand slot. An aggregate that reaches the inner Aggregate as count(1) — from COUNT(1) or COUNT(*) — 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 check RemoveRedundantAggregates uses for the same question:

val hasDuplicateSensitiveAgg = innerAgg.aggregateExpressions.exists(_.exists {
  case ae: AggregateExpression =>
    !ae.isDistinct && !EliminateDistinct.isDuplicateAgnostic(ae.aggregateFunction)
  case _ => false
})
if (hasDuplicateSensitiveAgg) return false

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.

CREATE TABLE oe USING parquet AS SELECT * FROM VALUES (1,5,7),(1,5,7),(1,5,7),(1,6,8),(2,9,9) AS t(k,a,b);

SET spark.sql.optimizer.optimizeExpandRatio=2;

SELECT k, COUNT(DISTINCT a), COUNT(DISTINCT b), COUNT(1) FROM oe GROUP BY k ORDER BY k;
-- before: [1,2,2,2], [2,1,1,1]   WRONG
-- after:  [1,2,2,4], [2,1,1,1]   k=1 has four rows

COUNT(*) behaves identically. The optimized plan before the fix shows the inserted pre-aggregate feeding the Expand whose downstream count(1) was supposed to count base rows:

Aggregate [k], [k, count(a) FILTER (gid=1), count(b) FILTER (gid=2),
                coalesce(first(count(1)) FILTER (gid=0), 0)]
+- Aggregate [k, a, b, gid], [k, a, b, gid, count(1)]
   +- Expand [[k,null,null,0], [k,a,null,1], [k,null,b,2]], [k, a, b, gid]
      +- Aggregate [k, a, b], [k, a, b]          <- inserted by OptimizeExpand
         +- Relation oe[k,a,b] parquet

Note on the JIRA: it also lists COUNT(a) on a non-nullable column as affected, on the grounds that the rewrite normalizes it to count(1). That does not reproduce on master — no such normalization happens, RewriteDistinctAggregates gives it its own Expand slot, 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.optimizeExpandRatio is 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) and OptimizeExpandQuerySuite (10 tests).

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Code (Claude Opus 5)

…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.
@Vivek1106-04
Vivek1106-04 force-pushed the SPARK-58387-optimize-expand branch from a02c2fa to 40eb7cd Compare August 19, 2026 05:08
@uros-b

uros-b commented Aug 19, 2026

Copy link
Copy Markdown
Member

Thank you @Vivek1106-04!

@Vivek1106-04

Copy link
Copy Markdown
Contributor Author

@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.

@uros-b

uros-b commented Sep 1, 2026

Copy link
Copy Markdown
Member

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.

@uros-b
uros-b requested a review from cloud-fan September 1, 2026 10:57

This branch has not been deployed

No deployments
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants