andygrove opened a new pull request, #2070:
URL: https://github.com/apache/datafusion-ballista/pull/2070
# Which issue does this PR close?
Closes #2067.
# Rationale for this change
A `UNION ALL` over file scans silently returned **inflated** results — no
error,
just several times too much data:
```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);
-- expected 5492796521.50
-- got 49389026433.70 (--partitions 16, executor -c 8; varied run to
run)
```
A task executes one partition of its stage, so `restrict_scan_to_partition`
pins
each file scan beneath it to the single file group that partition reads.
Without that, DataFusion 54's shared work-queue lets a lone `execute(p)`
drain
the queue and scan the whole table (#1907).
The pinning applied the stage's partition number **directly to every scan**,
which is only correct when the partition passes straight through. A
`UnionExec`
concatenates its children's partitions: with N-partition children, stage
partition `N + k` is the right child's local partition `k`. So:
- for partitions inside the left child, the right child's scan was pinned to
the
wrong group (harmless — it is never polled), and
- for partitions `>= N`, `partition_id >= config.file_groups.len()` bailed
out
and returned `None`, leaving that scan **unrestricted** and free to read
its
entire table.
`restrict_scan_to_partition`'s own doc comment anticipated exactly 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).
Confirmed by disabling the trigger from the other side: the query was correct
whenever the union stage's tasks fitted in a single wave of executor slots
(so nothing was left half-polled), and wrong beyond it. Moving the executor
from
`-c 8` to `-c 32` moved the failure threshold from 8 to 32 tasks exactly.
# What changes are included in this PR?
- Add `restrict_scans_for_partition`, which walks the plan from the stage
root
and mirrors `UnionExec::execute`'s own partition arithmetic, so each scan
is
pinned to the group its **local** partition reads. Children a partition
never
reaches are left untouched — this task does not execute them. Every other
operator passes the partition straight through, as before.
- Split the scan restriction out of the `ShuffleReaderExec` rewrite, which is
partition-independent.
- Un-skip TPC-DS q2 and q5 in the correctness gate.
## Verification
The minimal case above is now correct at **every** partition count tried —
4, 5,
6, 8, 16, 20, 32 — where previously anything past one wave of task slots was
wrong.
Full SF1 TPC-DS gate: **88/88 verified** against single-process DataFusion,
both
on a single executor and across 2 executors. `cargo test --workspace`, `cargo
clippy --all-targets --workspace -- -D warnings`, and `cargo fmt --all` are
clean. New unit tests pin the union mapping, including the case that
previously
left a scan unrestricted.
## Note on q2 / q5
Both were skipped as "non-deterministic (ORDER BY ties; varies run-to-run)".
That diagnosis was wrong: with a total order imposed and `LIMIT` removed they
still mismatched, so they were real wrong results, not tie ambiguity — they
were
hitting this bug. Both now verify in their original form across repeated
runs.
q31 and q71 *are* genuinely ordering-only (both verify once a total order is
imposed) and stay skipped.
# Are there any user-facing changes?
Yes — `UNION ALL` over file scans now returns correct results instead of
silently inflated ones. No API changes.
--
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]
