Skip to content

[WIP][Experimental] feat: use DataFusion's isolated_partitions option for broadcast-safe join selection - #2612

Draft
andygrove wants to merge 1 commit into
apache:mainfrom
andygrove:isolated-partitions
Draft

andygrove wants to merge 1 commit into
apache:mainfrom
andygrove:isolated-partitions

Conversation

@andygrove

Copy link
Copy Markdown
Member

Warning

Work in progress / experimental. Do not merge. This is a companion to a DataFusion draft PR: apache/datafusion#26330.

It temporarily patches DataFusion to a 55.2.0 backport of that change (andygrove/datafusion@448ed53).

Which issue does this PR close?

No issue is closed: this is an experimental companion to the DataFusion PR. Related: #1900, #2315.

Rationale for this change

DataFusion's JoinSelection promotes a small build side to CollectLeft for every join type. Ballista runs each partition as an isolated task, and in that setting CollectLeft is wrong for joins that emit build-side rows (Left, LeftSemi, LeftAnti, LeftMark, Full): every task emits those rows. TPC-DS q77, for example, came back 16x inflated (#2315).

Ballista works around this with two pieces of its own code (#1900):

  • DemoteUnsafeBroadcastJoinRule, which the adaptive planner splices in right after DataFusion's join_selection.
  • collect_left_broadcast_safe, used by the static planner and the adaptive join resolver.

The companion DataFusion PR adds the option datafusion.execution.isolated_partitions. With it set, JoinSelection makes broadcast-safe choices itself. This PR shows what Ballista gains: it turns the option on and deletes its own workaround.

What changes are included in this PR?

  • Set datafusion.execution.isolated_partitions = true in Ballista's session defaults (ballista/core/src/extension.rs).
  • Delete DemoteUnsafeBroadcastJoinRule, and stop splicing it in after DataFusion's join_selection in the adaptive planner.
  • Replace collect_left_broadcast_safe with DataFusion's JoinType::emits_build_side_rows() in the static planner, the adaptive join resolver and their test oracles, then delete the helper.
  • Temporary: [patch.crates-io] pins DataFusion to the 55.2.0 backport above.

Size: 173 lines removed and 17 added across Ballista's source, not counting the Cargo patch.

Tests (run locally with this change):

  • cargo test --workspace: 1,599 passed, 0 failed, 8 ignored.
  • ballista-scheduler: 482 passed.
  • TPC-H plan stability: all 22 approved distributed plans are unchanged.

Ballista's static-path copy of JoinSelection stays. It also carries the swap guard from #1691 and the zero-statistics check, which the DataFusion option deliberately does not cover.

Are there any user-facing changes?

No. Plans are unchanged; the DataFusion option replaces Ballista's own rule.

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.

1 participant