This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-25685-21a3215b66a8620cb932899de419e6b43d1848e6 in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 6ce67df8c18b8d2d08688191adae9d6986fb7ad4 Author: Malingshu <[email protected]> AuthorDate: Thu Sep 24 13:34:43 2026 +0000 refactor: remove redundant join-flattening precheck (#25685) ## Which issue does this PR close? Closes #25680. ## Rationale for this change `EliminateCrossJoin` walks the inner-join subtree twice before rebuilding it. Both callers of `can_flatten_join_inputs` already establish an inner-join root, and the helper only recurses into inner-join children, so the precheck always succeeds at these call sites. Removing it avoids the redundant traversal and the need to keep two traversal implementations in sync, while preserving the existing optimized plans. ## What changes are included in this PR? - Remove `can_flatten_join_inputs` and its two call-site guards. - Make the private `flatten_join_inputs` helper infallible and update its documentation. Keep filter collection, non-inner join boundaries, and child/subquery rewriting unchanged. - Add a focused plan test for preserving a left-join subtree and its filter beneath an inner join, including the inner join's residual filter and output schema. - Fix two existing Clippy warnings in `multi_group_by.rs` to pass the required local checks. ## What is the testing strategy for this PR? - All 26 `eliminate_cross_join` tests passed, including `preserve_outer_join_boundary`. - Extended workspace tests passed: 12,112 tests passed, 8 ignored, and all 523 sqllogictest files completed successfully. - `cargo fmt --all`, `cargo clippy --all-targets --all-features -- -D warnings`, and `uv run ./dev/rust_lint.sh` passed. - Compared the existing six-table self-join and no-join SQL planning workloads against `main` using a temporary Criterion harness. Across three alternating rounds, physical plans matched and no consistent end-to-end speedup was observed; six-table planning stayed around 4.9 ms. This is a maintainability refactor without a measured performance improvement claim. ## Are there any user-facing changes? No SQL behavior or public API changes. --- datafusion/optimizer/src/eliminate_cross_join.rs | 81 ++++++++++------------ datafusion/physical-plan/benches/multi_group_by.rs | 5 +- 2 files changed, 41 insertions(+), 45 deletions(-) diff --git a/datafusion/optimizer/src/eliminate_cross_join.rs b/datafusion/optimizer/src/eliminate_cross_join.rs index 5614948c9e..bdc70da593 100644 --- a/datafusion/optimizer/src/eliminate_cross_join.rs +++ b/datafusion/optimizer/src/eliminate_cross_join.rs @@ -118,10 +118,6 @@ impl OptimizerRule for EliminateCrossJoin { return rewrite_children(self, LogicalPlan::Filter(filter), config); } - if !can_flatten_join_inputs(&filter.input) { - return Ok(Transformed::no(LogicalPlan::Filter(filter))); - } - let Filter { input, predicate, .. } = filter; @@ -136,7 +132,7 @@ impl OptimizerRule for EliminateCrossJoin { &mut possible_join_keys, &mut all_inputs, &mut all_filters, - )?; + ); extract_possible_join_keys(&predicate, &mut possible_join_keys); Some(predicate) @@ -147,15 +143,12 @@ impl OptimizerRule for EliminateCrossJoin { null_equality: original_null_equality, .. }) => { - if !can_flatten_join_inputs(&plan) { - return Ok(Transformed::no(plan)); - } flatten_join_inputs( plan, &mut possible_join_keys, &mut all_inputs, &mut all_filters, - )?; + ); null_equality = original_null_equality; None } @@ -266,18 +259,15 @@ fn rewrite_children( } } -/// Recursively accumulate possible_join_keys and inputs from inner joins -/// (including cross joins). -/// -/// Assumes can_flatten_join_inputs has returned true and thus the plan can be -/// flattened. Adds all leaf inputs to `all_inputs` and join_keys to -/// possible_join_keys +/// Recursively collect inputs, join keys, and filters from inner joins +/// (including cross joins). Other nodes are retained as inputs without +/// flattening them. fn flatten_join_inputs( plan: LogicalPlan, possible_join_keys: &mut JoinKeySet, all_inputs: &mut Vec<LogicalPlan>, all_filters: &mut Vec<Expr>, -) -> Result<()> { +) { match plan { LogicalPlan::Join(join) if join.join_type == JoinType::Inner => { if let Some(filter) = join.filter { @@ -289,43 +279,18 @@ fn flatten_join_inputs( possible_join_keys, all_inputs, all_filters, - )?; + ); flatten_join_inputs( Arc::unwrap_or_clone(join.right), possible_join_keys, all_inputs, all_filters, - )?; + ); } _ => { all_inputs.push(plan); } } - Ok(()) -} - -/// Returns true if the plan is a Join or Cross join could be flattened with -/// `flatten_join_inputs` -/// -/// Must stay in sync with `flatten_join_inputs` -fn can_flatten_join_inputs(plan: &LogicalPlan) -> bool { - // can only flatten inner / cross joins - match plan { - LogicalPlan::Join(join) if join.join_type == JoinType::Inner => {} - _ => return false, - } - - for child in plan.inputs() { - if let LogicalPlan::Join(Join { - join_type: JoinType::Inner, - .. - }) = child - && !can_flatten_join_inputs(child) - { - return false; - } - } - true } /// Finds the next to join with the left input plan, @@ -1392,6 +1357,36 @@ mod tests { ) } + #[test] + fn preserve_outer_join_boundary() -> Result<()> { + let plan = LogicalPlanBuilder::from(test_table_scan_with_name("t1")?) + .join( + test_table_scan_with_name("t2")?, + JoinType::Left, + (vec!["t1.a"], vec!["t2.a"]), + Some(col("t1.b").gt(col("t2.b"))), + )? + .join( + test_table_scan_with_name("t3")?, + JoinType::Inner, + (vec!["t1.a"], vec!["t3.a"]), + Some(col("t3.c").gt(lit(7u32))), + )? + .build()?; + + assert_optimized_plan_equal!( + plan, + @ r" + Filter: t3.c > UInt32(7) [a:UInt32, b:UInt32, c:UInt32, a:UInt32;N, b:UInt32;N, c:UInt32;N, a:UInt32, b:UInt32, c:UInt32] + Inner Join: t1.a = t3.a [a:UInt32, b:UInt32, c:UInt32, a:UInt32;N, b:UInt32;N, c:UInt32;N, a:UInt32, b:UInt32, c:UInt32] + Left Join: t1.a = t2.a Filter: t1.b > t2.b [a:UInt32, b:UInt32, c:UInt32, a:UInt32;N, b:UInt32;N, c:UInt32;N] + TableScan: t1 [a:UInt32, b:UInt32, c:UInt32] + TableScan: t2 [a:UInt32, b:UInt32, c:UInt32] + TableScan: t3 [a:UInt32, b:UInt32, c:UInt32] + " + ) + } + #[test] fn preserve_null_equality_setting() -> Result<()> { let t1 = test_table_scan_with_name("t1")?; diff --git a/datafusion/physical-plan/benches/multi_group_by.rs b/datafusion/physical-plan/benches/multi_group_by.rs index 4753fe516f..260d3aacbb 100644 --- a/datafusion/physical-plan/benches/multi_group_by.rs +++ b/datafusion/physical-plan/benches/multi_group_by.rs @@ -1026,7 +1026,7 @@ fn generate_narrow_decimal_batches( ) -> Vec<Vec<ArrayRef>> { let num_full_batches = num_rows / batch_size; let remainder = num_rows % batch_size; - let num_batches = num_full_batches + if remainder > 0 { 1 } else { 0 }; + let num_batches = num_full_batches + usize::from(remainder > 0); (0..num_batches) .map(|batch_idx| { @@ -1055,7 +1055,8 @@ fn generate_narrow_decimal_batches( /// widths (4-byte `i32` and 8-byte `i64` native) on the multi-column path /// (previously such a schema fell back to `GroupValuesRows`). fn bench_narrow_decimals(c: &mut Criterion) { - let cases: [(&str, DataType, fn(&[usize]) -> ArrayRef); 2] = [ + type MakeKeys = fn(&[usize]) -> ArrayRef; + let cases: [(&str, DataType, MakeKeys); 2] = [ ("decimal32", DataType::Decimal32(9, 0), |group_ids| { Arc::new( Decimal32Array::from_iter_values(group_ids.iter().map(|&g| g as i32)) --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
