andygrove opened a new issue, #2184:
URL: https://github.com/apache/datafusion-ballista/issues/2184
## Describe the bug
At `target_partitions=1`, the per-task file-scan restriction misbehaves for
plans containing `UnionExec`, producing two related symptoms: silently inflated
results, and a hard failure.
Surfaced by extending the TPC-DS SF1 correctness gate to run at 1 partition
as well as 16 (#2182).
## Confirmed cases (SF1, `--partitions 1`, `prefer_hash_join=false`, static
planner)
| Query | Single-process DataFusion | Ballista (distributed) | Ratio |
|---|---|---|---|
| q5 | `73724.99` | `294899.96` | exactly **4x** |
| q80 | `11343.60` | `22687.20` | exactly **2x** |
q49 does not return a wrong answer, it fails outright:
```
Job failed due to stage 7 failed: Task failed due to runtime execution error:
Internal("FileStreamBuilder invalid partition index: 0")
```
q5, q80 and q49 are exactly the three queries in the affected set that
contain `UNION`. All three pass at `--partitions 16`.
## Why this looks like a gap in #2070
#2070 ("map a task's partition through UnionExec when restricting file
scans") fixed this exact inflation symptom — its commit message describes "a
silently inflated answer — no error, just several times too much data" — and
states the fix was verified "correct now at every partition count tried (4, 5,
6, 8, 16, 20, 32)". **1 was not among them.**
## Candidate mechanism
In `restrict_plan_to_partitions`
(`ballista/scheduler/src/state/task_builder.rs`):
- The walker passes the parent's partition indices unchanged through
interior nodes.
- At leaf file scans it selects file groups with `config.file_groups.get(i)`
inside a `filter_map`, so an index beyond the scan's file-group count is
**silently dropped** rather than reported.
That yields both observed symptoms:
- Dropping every index leaves `file_groups` empty, and executing partition 0
of an empty `FileStream` is precisely `FileStreamBuilder invalid partition
index: 0`.
- A restriction that fails to apply leaves the scan free to drain the shared
work queue and read its entire table, which is the inflated-answer path #2070
describes.
The comment at that call site already flags the assumption:
> currently safe because the only shape emitted is `partitions == 1` and
UnionExec routing gives sub-slices of `[0]`
which is the case being stressed here.
## To Reproduce
Generate TPC-DS SF1 Parquet with `tpcgen-cli`, start a scheduler and
executor, then:
```
tpcds --host localhost --port 50050 --path <data> --partitions 1 --verify \
--query 5 -c datafusion.optimizer.prefer_hash_join=false
# -> Query 5 VERIFY MISMATCH: expected `73724.99`, got `294899.96`
tpcds --host localhost --port 50050 --path <data> --partitions 1 --verify \
--query 49 -c datafusion.optimizer.prefer_hash_join=false
# -> Internal("FileStreamBuilder invalid partition index: 0")
```
Both pass with `--partitions 16`.
## Expected behavior
Results at `target_partitions=1` should match single-process DataFusion, as
they do at every other partition count.
--
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]
