You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/09/24 15:59:56 UTC
(datafusion) branch gh-readonly-queue/main/pr-25685-e2ca7f38051744b2010cf09b80db8e9dfa4b5d38 created (now 39916a3232)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-25685-e2ca7f38051744b2010cf09b80db8e9dfa4b5d38
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at 39916a3232 refactor: remove redundant join-flattening precheck (#25685)
This branch includes the following new commits:
new 39916a3232 refactor: remove redundant join-flattening precheck (#25685)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion) 01/01: refactor: remove redundant join-flattening precheck (#25685)
Posted by gi...@apache.org.
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-e2ca7f38051744b2010cf09b80db8e9dfa4b5d38
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 39916a3232b77d46a2c9ba530562e1c26d6a2510
Author: Malingshu <69...@users.noreply.github.com>
AuthorDate: Thu Sep 24 15:59:30 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.
Co-authored-by: Neil Conway <ne...@gmail.com>
---
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: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org