Dhruv-meesho opened a new issue, #12959:
URL: https://github.com/apache/gluten/issues/12959
### Backend
VL (Velox)
### Bug description
**Expected behavior:** a `LEFT JOIN` whose build side is unique on the join
key emits exactly one output row per probe row.
**Actual behavior:** the join emits extra rows, and the number of extra rows
differs between runs of the identical query on the identical configuration.
#### Root cause
`FlushableHashAggregateRule` decides whether an aggregate is an intermediate
one with:
```scala
case agg: RegularHashAggregateExecTransformer
if !agg.aggregateExpressions.forall(p => p.mode == Partial || p.mode ==
PartialMerge) =>
// Not an intermediate agg. Skip.
agg
```
For a **grouping-only aggregate** — `SELECT DISTINCT`, or a `GROUP BY` with
no aggregate functions — `aggregateExpressions` is empty
(`ReplaceDistinctWithAggregate` produces `Aggregate(child.output, child.output,
child)`), so `forall` is vacuously `true` and the guard never fires. With no
modes to inspect, the rule cannot distinguish the partial dedup from the final
dedup, and the **final** aggregate falls through to the catch-all and is
converted to `FlushableHashAggregateExecTransformer`.
That is then serialized as `allowFlush=1`
(`HashAggregateExecTransformer.formatExtOptimizationString`) and mapped to
`AggregationNode::Step::kPartial` by
`SubstraitToVeloxPlanConverter::toAggregationStep`. Velox gates
`abandonPartialAggregationEarly` on `isPartialOutput_`, so it is now free to
abandon aggregation and pass rows through — correct for a genuine partial
stage, but here no further aggregate follows, so the duplicate grouping keys
reach the consumer.
The rule only visits aggregates below a shuffle, so a plain `SELECT
DISTINCT` that is simply collected is unaffected — its final aggregate has no
exchange above it. The bug appears when the distinct output is **repartitioned
again**, e.g. when it feeds a join on a strict subset of the distinct keys. The
duplicates then become extra join output rows.
#### This is a regression
The case was previously covered by
`isAggInputAlreadyDistributedWithAggKeys`, added in #4443 to fix #4421
("Flushable distinct agg caused correctness issue"):
```scala
val distribution = ClusteredDistribution(agg.groupingExpressions)
agg.child.outputPartitioning.satisfies(distribution)
```
A distinct's final aggregate groups by the distinct keys and sits on a
shuffle partitioned by exactly those keys, so `satisfies` returned true and it
stayed regular.
That guard was removed in **#12098** (`d175c6322`, 2026-05-27) when the rule
was narrowed to protect only `AggUtils.planAggregateWithOneDistinct`. The
replacement `protectedAggs` machinery covers Spark's one-distinct
(`count(distinct …)`) pipeline but not a plain `SELECT DISTINCT`.
Note that the general principle was already articulated on #4421:
> **@Yohahaha:** "When agg function is empty, we may not use flushable agg
which may produce duplicate rows."
#### Minimal repro shape
```sql
SELECT count(*)
FROM t1
LEFT JOIN (SELECT DISTINCT a, b FROM t2) d ON t1.a = d.a
```
The join key must be a **strict subset** of the distinct keys, so a
repartition is required above the distinct's final aggregate. Needs enough data
(and low `abandonPartialAggregationMinPct` /
`abandonPartialAggregationMinRows`) for abandonment to actually trigger.
#### Evidence
Production query joining a probe side to `SELECT DISTINCT c1..c8 FROM dim`
on `c1`.
The build side is provably unique on the join key — measured in the same
query, via a separate `GROUP BY c1` over the same CTE:
| 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 |
Join output rows:
| run | output | excess |
|---|---:|---:|
| 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 |
Runs 1–3 used the same configuration. Rows are only ever duplicated, never
lost. Max per-key output multiplicity reached 18 against a build side with
multiplicity 1.
The same plan position, vanilla vs Gluten:
```
vanilla:
Sort
AQEShuffleRead / ShuffleQueryStage
Exchange [written 1,074,306,526]
HashAggregate [output 1,074,306,526] <-
final distinct
gluten:
SortExecTransformer [1,075,205,167]
AQEShuffleRead / ShuffleQueryStage
ColumnarExchange [1,075,205,167]
ProjectExecTransformer [1,075,205,167]
FlushableHashAggregateExecTransformer [1,075,205,167] <-
final distinct, flushed
```
The offending node, from the event log:
```
FlushableHashAggregateTransformer(
keys=[c1, c2, c3, c4, c5, c6, c7, c8], functions=[], isStreamingAgg=false)
```
`functions=[]` — the empty `aggregateExpressions` that makes the mode check
vacuous. It emitted 1,075,205,167 rows (and 1,075,589,830 on another run) where
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, stayed
`RegularHashAggregateExecTransformer` and produced the correct 1,074,306,526 —
which is why the same query can observe both the correct and the corrupted
count.
Not spill (all spill metrics zero) and not task retry (zero failed/retried
tasks, zero stage retries, zero speculation). Reproduces with
`spark.gluten.sql.columnar.forceShuffledHashJoin` both true and false, and with
`spark.sql.adaptive.skewJoin.enabled` both true and false.
#### Workaround
```
spark.gluten.sql.columnar.backend.velox.flushablePartialAggregation=false
```
With this set and everything else at its original value, output is
byte-identical to vanilla Spark.
#### Proposed fix
Skip only the case where the existing mode check is uninformative, rather
than restoring the broader guard removed by #12098 (which would give back its
performance gains):
```scala
private def isGroupingOnlyFinalAgg(agg: HashAggregateExecTransformer):
Boolean = {
agg.aggregateExpressions.isEmpty &&
agg.requiredChildDistributionExpressions.isDefined
}
```
`requiredChildDistributionExpressions` is `None` for a partial aggregate and
`Some(groupingAttributes)` for the final one
(`AggUtils.planAggregateWithoutDistinct`), and it survives when the modes
vanish. Spark makes the same check in
`HashAggregateExec.adaptivePartialAggEnabled`, its guard for the equivalent
runtime bypass (SPARK-58511). Aggregates that do have aggregate functions keep
the behavior introduced by #12098.
Happy to open a PR with this plus a regression test in
`VeloxAggregateFunctionsFlushSuite`.
Related: #4421, #4443, #12098, tracker #4652.
_This issue was written with the assistance of AI (Claude Code); all plan
output, row counts and code references were taken from actual runs and the
repository at `main`._
### 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
```
Spark 3.5.5, Delta 3.3.2.
### Relevant logs
_Plan trees and metrics above were extracted from the Spark event logs of
the affected runs._
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
