andygrove opened a new issue, #2046:
URL: https://github.com/apache/datafusion-ballista/issues/2046
## 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 <data> --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
<register the 24 TPC-DS tables as external parquet>
<q38 body> -- -> 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.
--
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]
