insert_broadcast_execs reuses the input of a build-side CoalescePartitionsExec and
rebuilds a fresh one above the BroadcastExec:
|
let broadcast_input = build_child |
let broadcast_input = build_child
.downcast_ref::<CoalescePartitionsExec>()
.map_or_else(
|| Arc::clone(build_child),
|coalesce| Arc::clone(coalesce.input()),
);
let broadcast: Arc<dyn ExecutionPlan> = Arc::new(BroadcastExec::new(broadcast_input, 1));
let new_build_child: Arc<dyn ExecutionPlan> =
Arc::new(CoalescePartitionsExec::new(broadcast));
CoalescePartitionsExec carries a fetch, which is where LimitPushdown parks a LIMIT.
Neither coalesce.input() nor CoalescePartitionsExec::new() keeps it, so the limit is
dropped and the join runs against the whole build side. Silent wrong results, no error.
Repro: build_side = ids 0..100 over 4 files, probe_side = same ids x50, 4 workers,
target_partitions = 3, broadcast joins on.
SELECT b.id, p.id FROM (SELECT id FROM build_side LIMIT 50) b
JOIN probe_side p ON b.id = p.id
Single node has CoalescePartitionsExec: fetch=50 under the join and returns 2500 rows.
Distributed has a plain CoalescePartitionsExec above the BroadcastExec and returns 3750.
Pre-existing on main (9d51c04f). normalize_collect_joins::collect_left_to_partitioned
and the NestedLoopJoinExec arm next to it (added in #583) do the same strip, so a fix should
cover all three.
Keeping the coalesce below the BroadcastExec isn't enough — the fetch would then apply once
per producer task instead of once globally.
insert_broadcast_execsreuses the input of a build-sideCoalescePartitionsExecandrebuilds a fresh one above the
BroadcastExec:datafusion-distributed/src/distributed_planner/insert_broadcast.rs
Line 136 in 9dd018a
CoalescePartitionsExeccarries afetch, which is whereLimitPushdownparks aLIMIT.Neither
coalesce.input()norCoalescePartitionsExec::new()keeps it, so the limit isdropped and the join runs against the whole build side. Silent wrong results, no error.
Repro:
build_side= ids 0..100 over 4 files,probe_side= same ids x50, 4 workers,target_partitions = 3, broadcast joins on.Single node has
CoalescePartitionsExec: fetch=50under the join and returns 2500 rows.Distributed has a plain
CoalescePartitionsExecabove theBroadcastExecand returns 3750.Pre-existing on
main(9d51c04f).normalize_collect_joins::collect_left_to_partitionedand the
NestedLoopJoinExecarm next to it (added in #583) do the same strip, so a fix shouldcover all three.
Keeping the coalesce below the
BroadcastExecisn't enough — the fetch would then apply onceper producer task instead of once globally.