apache / apache/datafusion-ballista
Distributed execution returns incorrect results for several TPC-DS queries (static planner)
- Dominant language
- Rust
- Stars
- 2.1k
- Forks
- 320
- Avg merge
- 1d 22h
- Merged PRs (30d)
- 66
Description
## Describe the bug
Under the **default (static) planner**, Ballista returns results for several TPC-DS queries that differ from single-process DataFusion executing the same SQL over the same data. The differences are stable across repeated runs (not query non-determinism), and single-process DataFusion returns the same answer under both `prefer_hash_join=true` and `prefer_hash_join=false`, so this is not a join-strategy artifact — it points to a distributed-execution correctness problem.
Surfaced by a TPC-DS correctness cross-check (SF1) that runs each query on a Ballista cluster and diffs the result against a single-process DataFusion oracle over the same Parquet data (the mechanism added by the TPC-H `--verify` gate, extended to TPC-DS).
## Confirmed cases (SF1, `target_partitions=16`, `prefer_hash_join=false`)
| Query | Shape | Single-process DataFusion | Ballista (distributed) |
|-------|-------|---------------------------|------------------------|
| q38 | `count(*)` over `INTERSECT` of 3 channel customer sets | **104** | **0** |
| q87 | `count(*)` over `EXCEPT` | **46753** | **47011** |
| q78 | `… ORDER BY … LIMIT 100` | **100 rows** | **73 rows** (27 dropped) |
| q4 | year-over-year CTE ratios, `LIMIT 100` | **8 rows** | **0 rows** |
- **Stable, not ties:** q38/q78/q87 reproduce identically across repeated Ballista runs, and single-process DataFusion is stable too. (Queries whose results genuinely vary run-to-run due to `LIMIT`/`ORDER BY` ties are a separate matter and are excluded from this report.)
- **Not a join-config artifact:** single-process DataFusion returns the correct answer under **both** `prefer_hash_join=true` and `false` at `target_partitions=16` (verified with `datafusion-cli`). Only distributed vs single-process differs.
The `INTERSECT`/`EXCEPT` (q38, q87) and `LIMIT` row-drop (q78) symptoms suggest the set-operation and limit lowering across shuffle stages as the area to look at.
## To Reproduce
Generate SF1 TPC-DS Parquet, start a scheduler + executor, then for the distributed result:
```
tpcds --host localhost --port 50050 --path --partitions 16 --verify \
--query 38 -c datafusion.optimizer.prefer_hash_join=false
# -> Query 38 VERIFY MISMATCH: expected 104, got 0
```
Single-process baseline (both join modes give 104):
```
datafusion-cli:
SET datafusion.execution.target_partitions = 16;
SET datafusion.optimizer.prefer_hash_join = false; -- and = true
-- -> 104
```
## Expected behavior
Distributed execution returns the same results as single-process DataFusion for the same query and data.
## Additional context
- Reproduced at SF1 on a 1-scheduler / 1-executor native cluster (8 slots, 16 partitions), base `apache/main` @ f45996cf.
- Related: #330 (older, general "some benchmark queries produce incorrect results"); #1055 (CollectLeft/broadcast outer-join correctness — likely distinct, as this reproduces on the static planner with `prefer_hash_join=false`, i.e. sort-merge joins, no broadcast).
- The TPC-DS correctness harness that surfaced this is proposed in the PR for #1845.
Contributor guide
Research direction
Start with the TPC-DS --verify path and reproduce query 38 at target_partitions=16, comparing the Ballista result with the datafusion-cli baseline. Trace set-operation and LIMIT lowering across the shuffle stages, using q38, q87, q78, and q4 as correctness cases. Done means distributed execution matches single-process DataFusion for the confirmed queries and the verification run passes.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- rust, sql
- Domain
- databases, distributed-systems, testing-qa
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 48/100