Skip to content

fix: [branch-54] map a task's partition through UnionExec when restricting file scans (backport of #2070) - #2131

Merged
andygrove merged 1 commit into
apache:branch-54from
andygrove:fix/union-partition-scan-branch-54
Jul 22, 2026
Merged

fix: [branch-54] map a task's partition through UnionExec when restricting file scans (backport of #2070)#2131
andygrove merged 1 commit into
apache:branch-54from
andygrove:fix/union-partition-scan-branch-54

Conversation

@andygrove

Copy link
Copy Markdown
Member

Which issue does this PR close?

Backport of #2070 to branch-54. The issue it fixes is #2067.

Rationale for this change

A UNION ALL over file scans silently returns inflated results on 54.x — no error, just several times too much data:

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 >= N, partition_id >= config.file_groups.len() bailed out and returned None, leaving that scan unrestricted and free to read its entire table.

Silent wrong results are the worst failure mode to leave in a released line, which is why this is worth backporting.

What changes are included in this PR?

A clean cherry-pick of 449e7b5, confined to ballista/executor/src/execution_engine.rs. The plan is walked from the stage root, mirroring UnionExec::execute's own partition arithmetic, so each scan is pinned to the group its local partition actually reads. Children a partition never reaches are left untouched, since this task does not execute them.

The upstream commit also removed two entries from the TPC-DS correctness gate's SKIP list (benchmarks/src/bin/tpcds.rs). That harness arrived with #2048 and does not exist on branch-54, so that hunk is dropped. It has no bearing on the fix.

All four regression tests from the original PR come across and pass on the branch-54 base:

  • union_partition_maps_to_the_right_childs_local_group
  • every_union_partition_pins_exactly_one_group
  • scan_without_union_is_pinned_to_its_own_partition
  • restrict_scan_partition_out_of_range_is_left_untouched

Are there any user-facing changes?

Queries with a UNION ALL over file scans now return correct results instead of inflated ones. No public API, config, or proto/wire changes, so this is not a breaking change.

cargo fmt --check, clippy on ballista-executor with --all-targets --all-features, and the full ballista-executor test suite are clean.

…cans (apache#2070)

A task executes one partition of its stage, so each file scan beneath it is
pinned to the single file group that partition reads — otherwise DataFusion 54's
shared work-queue lets a lone `execute(p)` drain the queue and scan the whole
table.

That pinning applied the stage's partition number directly to every scan, which
is only right 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`. Applying the stage
number there addressed the wrong group, and for partitions past a scan's group
count it addressed none at all, leaving that scan unrestricted and free to read
its entire table. The result was a silently inflated answer — no error, just
several times too much data.

Walk the plan from the stage root instead, mirroring `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.

Minimal case, SF1, executor `-c 8`:

    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
    -- was      49389026433.70 at --partitions 16 (~9x, varying run to run)

Correct now at every partition count tried (4, 5, 6, 8, 16, 20, 32); previously
wrong whenever the union stage exceeded one wave of executor task slots.

TPC-DS q2 and q5 now match single-process DataFusion and join the correctness
gate. Both were skipped as "non-deterministic (ORDER BY ties)"; that diagnosis
was wrong — with a total order imposed and LIMIT removed they still mismatched,
because they were hitting this bug. q31 and q71 are genuinely ordering-only and
stay skipped.
@andygrove andygrove changed the title fix: map a task's partition through UnionExec when restricting file scans (backport of #2070) fix: [branch-54] map a task's partition through UnionExec when restricting file scans (backport of #2070) Jul 22, 2026
@andygrove
andygrove marked this pull request as ready for review July 22, 2026 02:35
@andygrove
andygrove merged commit aa0b626 into apache:branch-54 Jul 22, 2026
18 checks passed
@andygrove
andygrove deleted the fix/union-partition-scan-branch-54 branch July 22, 2026 12:40
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants