[VL] Flushable aggregate rule converts the final stage of a grouping-only aggregate, producing duplicate rows

[Dhruv-meesho]

Backend

VL (Velox)

Bug description

We are seeing incorrect and non-deterministic query results with Gluten + Velox for a specific query pattern involving a SELECT DISTINCT followed by a repartition and JOIN.

The issue appears to be that the final aggregation stage of a grouping-only aggregate (e.g. SELECT DISTINCT) is being converted to FlushableHashAggregateExecTransformer.

When Velox abandons the aggregation early under memory pressure, duplicate grouping keys can be emitted. This is safe for a genuine partial aggregation because another aggregation follows downstream, but in this case the aggregation is the final DISTINCT, so there is no subsequent aggregation to remove the duplicates.

As a result, the duplicates reach the JOIN and produce extra output rows.

Simple example

SELECT count(*)
FROM t1
LEFT JOIN (
    SELECT DISTINCT a, b
    FROM t2
) d
ON t1.a = d.a

Here, (a, b) is the DISTINCT key, while the JOIN is only on a.

The expected DISTINCT result might be:

a	b
1	X
1	Y

If the final DISTINCT aggregation incorrectly emits:

a	b
1	X
1	X
1	Y

the row from t1 with a = 1 matches the duplicate row multiple times, resulting in an inflated and incorrect JOIN output.

---

Production evidence

We reproduced this with a query where the build side is:

SELECT DISTINCT c1, c2, c3, c4, c5, c6, c7, c8
FROM dim

and the JOIN is performed on c1.

The build side is provably unique on the JOIN key:

Metric	Value
Build-side distinct keys	1,074,306,526
Build-side rows	1,074,306,526
Build-side max multiplicity	1
Probe-side rows	33,161,001

Vanilla Spark consistently produces:

• 33,161,001 rows

However, Gluten + Velox produces different results across identical runs:

Run	Join output	Excess rows
Vanilla Spark 3.5.5	33,161,001	0
Gluten, run 1	33,264,832	+103,831
Gluten, run 2	33,264,992	+103,991
Gluten, run 3	33,314,214	+153,213

(All three Gluten runs used the exact same dataset and cluster configuration).

The extra rows are duplicates rather than missing rows. The maximum per-key output multiplicity reached 18, while the corresponding build-side multiplicity was 1.

This demonstrates both:

1. Incorrect results compared with vanilla Spark.
2. Non-deterministic results across identical Gluten runs.

---

Root cause

FlushableHashAggregateRule currently determines whether an aggregate is an intermediate aggregate using:

agg.aggregateExpressions.forall(
  p => p.mode == Partial || p.mode == PartialMerge
)

For a grouping-only aggregate such as SELECT DISTINCT a, b, there are no aggregate functions, so aggregateExpressions is empty (Seq.empty).

Therefore, the forall check evaluates vacuously to true, and the rule cannot distinguish the partial deduplication stage from the final deduplication stage.

The final aggregate is consequently converted to FlushableHashAggregateExecTransformer, which is serialized with allowFlush=1 and mapped to AggregationNode::Step::kPartial in Velox.

Velox can then abandon the aggregation early based on its runtime heuristics and emit the remaining rows without aggregation. Because this is the final aggregation, the duplicate grouping keys are passed directly to the consumer.

---

Why the issue only appears in certain query shapes

FlushableHashAggregateRule only visits aggregates below a shuffle.

Therefore, a simple SELECT DISTINCT ... that is consumed directly does not reproduce the issue because its final aggregate has no exchange above it.

The problem appears when the DISTINCT output is repartitioned again:

DISTINCT(a, b, c)
       |
       v
    shuffle
       |
       v
   JOIN ON a

where the JOIN key is a strict subset of the DISTINCT keys. The duplicates produced by the final DISTINCT can then become additional JOIN matches.

---

Plan evidence

For the affected query, vanilla Spark has:

Sort
  AQEShuffleRead / ShuffleQueryStage
    Exchange
      HashAggregate                 <- final DISTINCT

while Gluten produces:

SortExecTransformer
  AQEShuffleRead / ShuffleQueryStage
    ColumnarExchange
      ProjectExecTransformer
        FlushableHashAggregateExecTransformer   <- final DISTINCT

The offending node from the event log is:

FlushableHashAggregateExecTransformer(
  keys=[c1, c2, c3, c4, c5, c6, c7, c8],
  functions=[],
  isStreamingAgg=false
)

Notice functions=[], which corresponds to the grouping-only aggregate.

The affected Gluten aggregate emitted:

• 1,075,205,167 rows in run 1
• 1,075,589,830 rows in run 3

while the correct DISTINCT count is 1,074,306,526.

A sibling instance of the same logical CTE, reached through a branch where the rule stopped at a different aggregate first, remained RegularHashAggregateExecTransformer and produced the correct 1,074,306,526 rows.

---

Additional validation

We verified that the behavior is not caused by:

• Spill (all spill metrics were 0)
• Task retries / stage retries / speculation (all 0)
• Join strategy (forceShuffledHashJoin true/false both reproduce)
• Skew join (skewJoin.enabled true/false both reproduce)

---

Regression

This appears to be a regression from #12098.

The affected case was previously covered by isAggInputAlreadyDistributedWithAggKeys, introduced in #4443 to fix #4421 (Flushable distinct agg caused correctness issue).

That protection was removed in #12098 when FlushableHashAggregateRule was narrowed to protect only the AggUtils.planAggregateWithOneDistinct path. The replacement protectedAggs logic covers Spark's COUNT(DISTINCT ...) pipeline, but does not cover a plain SELECT DISTINCT.

---

Workaround

Disabling flushable partial aggregation avoids the issue:

spark.gluten.sql.columnar.backend.velox.flushablePartialAggregation=false

With this configuration, the output is byte-identical to vanilla Spark.

---

Proposed fix

Rather than restoring the broader protection removed by #12098, we can specifically identify the final grouping-only aggregate using requiredChildDistributionExpressions.isDefined (which Spark sets to None for partial aggregates and Some(groupingAttributes) for final aggregates in AggUtils.planAggregateWithoutDistinct, matching Spark's own check in HashAggregateExec.adaptivePartialAggEnabled):

private def isGroupingOnlyFinalAgg(
    agg: HashAggregateExecTransformer
): Boolean = {
  agg.aggregateExpressions.isEmpty &&
  agg.requiredChildDistributionExpressions.isDefined
}

This allows the flushable aggregation optimization to remain enabled for aggregates with actual aggregate functions, while preventing it from being applied to the final grouping-only aggregation.

A fix and regression test have been submitted in PR #12960.

Related issues / PRs

• [VL] Flushable distinct agg caused correctness issue #4421 — Flushable distinct agg caused correctness issue
• [GLUTEN-4421][VL] Disable flushable aggregate when input is already partitioned by grouping keys #4443 — Disable flushable aggregate when input is already partitioned by grouping keys
• [VL] Narrow down FlushableHashAggregateRule to only protect against Spark code AggUtils.planAggregateWithOneDistinct #12098 — Narrow down FlushableHashAggregateRule
• [GLUTEN-12959][VL] Keep the final stage of a grouping-only aggregate non-flushable #12960 — Proposed fix for this issue

Gluten version

main branch

Spark version

Spark-3.5.x

Spark configurations

spark.plugins=org.apache.gluten.GlutenPlugin
spark.shuffle.manager=org.apache.spark.shuffle.sort.ColumnarShuffleManager
spark.memory.offHeap.enabled=true
spark.memory.offHeap.size=280g
spark.executor.cores=64
spark.sql.shuffle.partitions=10000
spark.sql.adaptive.enabled=true
spark.sql.autoBroadcastJoinThreshold=-1
spark.sql.join.preferSortMergeJoin=true
spark.gluten.memory.isolation=true
spark.gluten.sql.columnar.backend.velox.memoryCapRatio=0.8

Relevant logs

FlushableHashAggregateExecTransformer(keys=[c1, c2, c3, c4, c5, c6, c7, c8], functions=[], isStreamingAgg=false)