andygrove opened a new issue, #2067:
URL: https://github.com/apache/datafusion-ballista/issues/2067
## Describe the bug
A `UNION ALL` over two file scans returns **massively inflated results**
whenever
the union's stage has more tasks than the executor has concurrent task slots.
The query does not error — it silently over-counts, so aggregates come back
several times too large.
## To Reproduce
SF1 TPC-DS parquet, 1 scheduler + 1 executor, release build:
```sql
select round(sum(p),2) from
(select ws_ext_sales_price p from web_sales
union all
select cs_ext_sales_price p from catalog_sales);
```
With `--partitions 16` and an executor started `-c 8`:
```
expected 5492796521.50 (single-process DataFusion, same data)
got 42072988115.00 (~8x inflated; varies run to run)
```
Summing each table separately is correct, and the same aggregate over a
single
table (no `UNION ALL`) is correct — only the union is wrong.
Note `select count(*)` over the union is **not** a reproducer: it is answered
from parquet statistics without reading data. The aggregate must touch a
column.
## The rule: it breaks exactly when the stage exceeds one wave of task slots
A `UnionExec` has as many partitions as its children combined, so an
N-partition
union stage runs `2N` tasks. The result is correct **iff those tasks fit in a
single concurrent wave** on the executor:
Executor `-c 8`:
| `--partitions` | union stage tasks | result |
|---:|---:|---|
| 3 | 6 | correct |
| 4 | 8 | correct |
| 5 | 10 | **wrong** (12808834840.20) |
| 6 | 12 | **wrong** (16466853999.55) |
| 8 | 16 | **wrong** (16466853999.55) |
Same binary, same data, executor `-c 32` — the threshold moves with the slot
count, which isolates the cause:
| `--partitions` | union stage tasks | result |
|---:|---:|---|
| 5 | 10 | correct |
| 8 | 16 | correct |
| 16 | 32 | correct |
| 20 | 40 | **wrong** (20124873158.90) |
So: **correct iff `2 * partitions <= executor slots`.**
## Suspected cause
`restrict_scan_to_partition` (`ballista/executor/src/execution_engine.rs`)
exists
precisely because DF54's `DataSourceExec` hands file groups out from a shared
work-queue that is only divided correctly when all partitions of a plan
instance
are polled concurrently (#1907). It pins each task's scan to its own file
group —
but bails out early:
```rust
if partition_id >= config.file_groups.len() {
return None; // no restriction -> scan left free to drain the whole
queue
}
```
Its own doc comment anticipates this case:
> Returns `None` ... for a `partition_id` outside the source's file groups
(e.g.
> when an operator between the scan and the stage output changed the
partition
> count).
A `UnionExec` is exactly such an operator: with `target_partitions=16` each
scan
has 16 file groups but the union stage has 32 partitions, so tasks 16..31
skip
the restriction and their scans are left unrestricted. An unrestricted scan
only
behaves when the whole stage happens to be polled in one concurrent wave,
which
is precisely the boundary measured above.
Fixing it presumably means mapping the stage's output partition to each
scan's
*local* partition index (a union offsets its right child's partitions by the
left
child's count) rather than using the stage `partition_id` directly, so every
scan is pinned to the right file group.
## Impact
- **16 of the 99 TPC-DS queries use `UNION ALL`** (q2, q4, q5, q11, q14, q23,
q33, q54, q56, q60, q66, q71, q74, q76, q77, q80), so this is broadly
reachable
on any realistic partitions/slots ratio.
- It accounts for the q2 and q5 failures currently skipped in the TPC-DS
correctness gate as "non-deterministic (ORDER BY ties)". That label is
wrong:
with a total order imposed and `LIMIT` removed, both still mismatch, so
they
are real wrong results, not tie ambiguity. (q31 and q71 *are* genuinely
ordering-only — both verify OK once a total order is imposed.)
- Silent: nothing errors, results are just too large.
## Expected behavior
`UNION ALL` returns the same result as single-process DataFusion,
independent of
`target_partitions` and executor slot count.
## Additional context
- Reproduced at SF1 on a 1-scheduler / 1-executor native cluster,
`upstream/main`
(e843c8e8), default static planner, under both `prefer_hash_join=true` and
`false`.
- Regression surface introduced with the DataFusion 54 upgrade; closely
related to
#1907, whose fix covers the non-union case (a plain scan verifies
correctly at
any partition/slot ratio) but leaves partitions beyond `file_groups.len()`
unrestricted.
--
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]
