Restore pluggable left-join accumulator seam (CollectLeftAccumulator trait + recreate_with_accumulator) on DF53.1#163
Open
lukekim wants to merge 2 commits into
Conversation
…53.1 Restore the extension seam dropped during the DF53.1 reconciliation so downstream code can plug a custom build-side (left) accumulator. Re-publishes the public API of the 52.x PR #116/#117 seam, adapted by hand to the substantially restructured DF53.1 hash join (HashJoinExecBuilder, PushdownStrategy/SharedBuildAccumulator, and the new ArrayMap perfect-hash path) since the original commits no longer cherry-pick cleanly: - pub trait CollectLeftAccumulator (try_new/update_batch/evaluate/name/static_name) - pub trait ColumnBounds (dyn-able, returns a bounds physical_expr); native MinMaxColumnBounds implements it - HashJoinExec<A: CollectLeftAccumulator = MinMaxLeftAccumulator> + matching generic HashJoinExecBuilder<A> (default preserves native min/max + capped InList behavior exactly) - HashJoinExec::recreate_with_accumulator::<B>() - ConfigOptions::optimizer.exact_join_filter_max_bytes (default usize::MAX, no behavior change) The native perfect-hash-join path (try_create_array_map) recovers concrete min/max by downcasting dyn ColumnBounds to MinMaxColumnBounds; custom bounds types simply skip the ArrayMap optimization. cargo check -p datafusion is green and all 371 hash_join unit tests pass.
There was a problem hiding this comment.
Pull request overview
This PR restores the “pluggable left-join accumulator seam” on the DataFusion 53.1 line by making HashJoinExec generic over a new CollectLeftAccumulator trait (defaulting to the existing min/max behavior), reintroducing recreate_with_accumulator for post-construction swapping, and adding a new optimizer config budget (exact_join_filter_max_bytes) intended for custom exact-membership accumulators.
Changes:
- Introduces
CollectLeftAccumulator+ColumnBoundstraits and refactors the native min/max implementation intoMinMaxLeftAccumulator/MinMaxColumnBounds. - Makes
HashJoinExec/HashJoinExecBuildergeneric over the accumulator type (defaulting to the native min/max accumulator) and addsHashJoinExec::recreate_with_accumulator. - Adds
datafusion.optimizer.exact_join_filter_max_bytesconfig option and documents it.
Reviewed changes
Copilot reviewed 6 out of 6 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
| docs/source/user-guide/configs.md | Documents the new exact_join_filter_max_bytes optimizer setting. |
| datafusion/common/src/config.rs | Adds the exact_join_filter_max_bytes config option with rustdoc. |
| datafusion/physical-plan/src/joins/mod.rs | Re-exports the new accumulator/bounds traits and native implementations. |
| datafusion/physical-plan/src/joins/hash_join/mod.rs | Re-exports accumulator/bounds API from the hash join module. |
| datafusion/physical-plan/src/joins/hash_join/shared_bounds.rs | Introduces ColumnBounds trait and MinMaxColumnBounds implementation; updates bounds predicate construction. |
| datafusion/physical-plan/src/joins/hash_join/exec.rs | Makes HashJoinExec generic over accumulator type, adds recreate_with_accumulator, and threads the accumulator through build-side collection. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+86
to
97
| pub struct MinMaxColumnBounds { | ||
| /// The minimum value observed for this column | ||
| pub(crate) min: ScalarValue, | ||
| /// The maximum value observed for this column | ||
| /// The maximum value observed for this column | ||
| pub(crate) max: ScalarValue, | ||
| } | ||
|
|
||
| impl ColumnBounds { | ||
| impl MinMaxColumnBounds { | ||
| pub(crate) fn new(min: ScalarValue, max: ScalarValue) -> Self { | ||
| Self { min, max } | ||
| } | ||
| } |
Comment on lines
+871
to
+891
| HashJoinExec { | ||
| left: Arc::clone(&self.left), | ||
| right: Arc::clone(&self.right), | ||
| on: self.on.clone(), | ||
| filter: self.filter.clone(), | ||
| join_type: self.join_type, | ||
| join_schema: Arc::clone(&self.join_schema), | ||
| left_fut: Arc::clone(&self.left_fut), | ||
| random_state: self.random_state.clone(), | ||
| mode: self.mode, | ||
| metrics: self.metrics.clone(), | ||
| projection: self.projection.clone(), | ||
| column_indices: self.column_indices.clone(), | ||
| null_equality: self.null_equality, | ||
| null_aware: self.null_aware, | ||
| cache: Arc::clone(&self.cache), | ||
| dynamic_filter: self.dynamic_filter.clone(), | ||
| fetch: self.fetch, | ||
| _phantom_accumulator: PhantomData, | ||
| } | ||
| } |
Comment on lines
1956
to
1960
| let should_collect_min_max_for_phj = | ||
| should_collect_min_max_for_perfect_hash(&on_left, &schema)?; | ||
|
|
||
| let initial = BuildSideState::try_new( | ||
| let initial = BuildSideState::<A>::try_new( | ||
| metrics, |
…eyond input stats
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Re-ports the forked pluggable left-join accumulator seam onto the DF53.1 line. The DF53.1 reconciliation dropped it; this restores the ability to plug a custom
CollectLeftAccumulatorintoHashJoinExec— used downstream by Spice'sExactLeftAccumulatorfor exact-membership scan pruning on large analytical joins.What changed — additive, native behavior unchanged
HashJoinExec<A: CollectLeftAccumulator = MinMaxLeftAccumulator>+HashJoinExecBuilder<A = MinMaxLeftAccumulator>— generic over the accumulator, defaulting to the existing nativeMinMaxLeftAccumulator, so every bareHashJoinExecreference anddowncast_ref::<HashJoinExec>()across physical-optimizer / planner / proto / tests resolves unchanged.pub trait CollectLeftAccumulator(name/static_name/try_new/update_batch/evaluate) +pub trait ColumnBounds(as_any+physical_expr); the old concrete min/max accumulator →pub struct MinMaxLeftAccumulator/pub struct MinMaxColumnBoundsimplementing them.HashJoinExec::recreate_with_accumulator::<B>()to swap the accumulator post-construction (threaded throughwith_new_children/reset_state/with_projection/swap_inputs).optimizer.exact_join_filter_max_bytesConfigOptions field (defaultusize::MAX) — the budget bridge for a custom accumulator; not consulted by the native min/max path.try_create_array_mappath recovers concrete bounds viaas_any().downcast_ref::<MinMaxColumnBounds>()— identical for the default accumulator; a custom accumulator's bounds simply skip ArrayMap (correct).Why hand-ported, not cherry-picked
The 52.x seam commits (#116 trait-ify
CollectLeftAccumulator, #117recreate_with_accumulator) don't cherry-pick — DF53 rewrote the hash-join (SharedBuildAccumulator/report_build_data/PushdownStrategy/try_create_array_mapvs 52.xSharedBoundsAccumulator/report_partition_bounds). Hand-applied, keeping the native default path byte-for-byte equivalent.Validation
cargo check -p datafusion+-p datafusion-physical-plan(lib +--tests) +-p datafusion-common: green. Clippy clean on physical-plan + common.joins::hash_joinunit tests pass (incl. perfect-hash + dynamic-filter cases).