Skip to content

feat: Support Expanding OR Conditions in INNER JOIN into Multiple Mutually Exclusive Branches #22370

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
xiedeyantu wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat: Support Expanding OR Conditions in INNER JOIN into Multiple Mutually Exclusive Branches #22370

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 16 additions & 15 deletions datafusion/core/src/optimizer_rule_reference.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,21 +47,22 @@ Rule order matters. The default pipeline may change between releases.
| 8 | `scalar_subquery_to_join` | Rewrites eligible scalar subqueries into joins and adds schema-preserving projections. |
| 9 | `decorrelate_lateral_join` | Rewrites eligible lateral joins into regular joins. |
| 10 | `extract_equijoin_predicate` | Splits join filters into equijoin keys and residual predicates. |
| 11 | `eliminate_duplicated_expr` | Removes duplicate expressions from projections, aggregates, and similar operators. |
| 12 | `eliminate_filter` | Drops always-true filters and replaces always-false or NULL filters with empty relations. |
| 13 | `eliminate_cross_join` | Uses filter predicates to replace cross joins with inner joins when join keys can be found. |
| 14 | `eliminate_limit` | Removes no-op limits and simplifies trivial limit shapes. |
| 15 | `propagate_empty_relation` | Pushes empty-relation knowledge upward so operators fed by no rows collapse early. |
| 16 | `filter_null_join_keys` | Adds `IS NOT NULL` filters to nullable equijoin keys that can never match. |
| 17 | `eliminate_outer_join` | Rewrites outer joins to inner joins when later filters reject the NULL-extended rows. |
| 18 | `push_down_limit` | Moves literal limits closer to scans and unions and merges adjacent limits. |
| 19 | `push_down_filter` | Moves filters as early as possible through filter-commutative operators. |
| 20 | `single_distinct_aggregation_to_group_by` | Rewrites single-column `DISTINCT` aggregations into two-stage `GROUP BY` plans. |
| 21 | `eliminate_group_by_constant` | Removes constant or functionally redundant expressions from `GROUP BY`. |
| 22 | `common_sub_expression_eliminate` | Computes repeated subexpressions once and reuses the result. |
| 23 | `extract_leaf_expressions` | Pulls cheap leaf expressions closer to data sources so later pruning and filter rules can act earlier. |
| 24 | `push_down_leaf_projections` | Pushes the helper projections created by leaf extraction toward leaf inputs. |
| 25 | `optimize_projections` | Prunes unused columns and removes unnecessary logical projections. |
| 11 | `expand_join_or_predicate` | Rewrites inner joins whose OR branches are all hashjoin-capable equijoin predicates into a `UNION ALL` of inner joins. |
| 12 | `eliminate_duplicated_expr` | Removes duplicate expressions from projections, aggregates, and similar operators. |
| 13 | `eliminate_filter` | Drops always-true filters and replaces always-false or NULL filters with empty relations. |
| 14 | `eliminate_cross_join` | Uses filter predicates to replace cross joins with inner joins when join keys can be found. |
| 15 | `eliminate_limit` | Removes no-op limits and simplifies trivial limit shapes. |
| 16 | `propagate_empty_relation` | Pushes empty-relation knowledge upward so operators fed by no rows collapse early. |
| 17 | `filter_null_join_keys` | Adds `IS NOT NULL` filters to nullable equijoin keys that can never match. |
| 18 | `eliminate_outer_join` | Rewrites outer joins to inner joins when later filters reject the NULL-extended rows. |
| 19 | `push_down_limit` | Moves literal limits closer to scans and unions and merges adjacent limits. |
| 20 | `push_down_filter` | Moves filters as early as possible through filter-commutative operators. |
| 21 | `single_distinct_aggregation_to_group_by` | Rewrites single-column `DISTINCT` aggregations into two-stage `GROUP BY` plans. |
| 22 | `eliminate_group_by_constant` | Removes constant or functionally redundant expressions from `GROUP BY`. |
| 23 | `common_sub_expression_eliminate` | Computes repeated subexpressions once and reuses the result. |
| 24 | `extract_leaf_expressions` | Pulls cheap leaf expressions closer to data sources so later pruning and filter rules can act earlier. |
| 25 | `push_down_leaf_projections` | Pushes the helper projections created by leaf extraction toward leaf inputs. |
| 26 | `optimize_projections` | Prunes unused columns and removes unnecessary logical projections. |

### Physical Optimizer Rules

Expand Down
13 changes: 7 additions & 6 deletions datafusion/optimizer/src/eliminate_cross_join.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -310,6 +310,11 @@ fn find_inner_join(
) -> Result<LogicalPlan> {
for (i, right_input) in rights.iter().enumerate() {
let mut join_keys = vec![];
let candidate_join_schema = Arc::new(build_join_schema(
left_input.schema(),
right_input.schema(),
&JoinType::Inner,
)?);

for (l, r) in possible_join_keys.iter() {
let key_pair = find_valid_equijoin_key_pair(
Expand All @@ -321,7 +326,7 @@ fn find_inner_join(

// Save join keys
if let Some((valid_l, valid_r)) = key_pair
&& can_hash(&valid_l.get_type(left_input.schema())?)
&& can_hash(&valid_l.get_type(candidate_join_schema.as_ref())?)
{
join_keys.push((valid_l, valid_r));
}
Expand All @@ -331,11 +336,7 @@ fn find_inner_join(
if !join_keys.is_empty() {
all_join_keys.insert_all(join_keys.iter());
let right_input = rights.remove(i);
let join_schema = Arc::new(build_join_schema(
left_input.schema(),
right_input.schema(),
&JoinType::Inner,
)?);
let join_schema = candidate_join_schema;

return Ok(LogicalPlan::Join(Join {
left: Arc::new(left_input),
Expand Down
174 changes: 174 additions & 0 deletions datafusion/optimizer/src/expand_join_or_predicate.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,174 @@
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.

//! [`ExpandJoinOrPredicate`] rewrites inner joins with OR filters into a UNION ALL
//! of mutually exclusive hashjoin-capable inner joins.

use crate::optimizer::ApplyOrder;
use crate::{OptimizerConfig, OptimizerRule};
use std::sync::Arc;

use datafusion_common::tree_node::Transformed;
use datafusion_common::Result;
use datafusion_expr::logical_plan::{Join, LogicalPlan, Projection, Union};
use datafusion_expr::utils::{can_hash, find_valid_equijoin_key_pair, split_binary_owned, split_conjunction_owned};
use datafusion_expr::{Expr, ExprSchemable, JoinType, Operator};

#[derive(Default, Debug)]
pub struct ExpandJoinOrPredicate;

impl ExpandJoinOrPredicate {
Copy link
Copy Markdown
Contributor

@Dandandan Dandandan May 19, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we avoid a new rule for this?

Copy link
Copy Markdown
Member Author

@xiedeyantu xiedeyantu May 19, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I see this as a relatively standalone piece of logic, and I haven’t yet found a clean way to hook into it.

Another concern is that I’m not sure whether DataFusion can reliably obtain the row counts from both sides of a join—during testing, I noticed that when both tables are very small (say, ~10 rows), NestedLoopJoin can actually be faster than HashJoin + Union. However, once both tables grow larger (e.g., 1000+ rows), the rewrite shows very significant gains.

Could you walk me through your concerns in more detail? That would help me figure out how to improve it.

Copy link
Copy Markdown
Contributor

@Dandandan Dandandan May 20, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The concern mostly is makung the optimizer list longer and writing more passes than necessary. If it would fit in an existing planning / optimization pass it would be good.

Copy link
Copy Markdown
Member Author

@xiedeyantu xiedeyantu May 20, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The only solution I can think of right now is to add this capability to the existing extract_equijoin_predicate rules. However, strictly speaking, they do completely different things, and the code logic isn't reusable. But if you agree, I can modify it this way. Or please feel free to let me know if you have any better suggestions.

Copy link
Copy Markdown
Contributor

@Dandandan Dandandan May 21, 2026 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I am wonder if instead using union + duplicating scan / subquery we can try one of the following approaches:

  • Disjoint hash join operator (supporting multiple branches) this would also work for non-mutually exclusive ones => This would probably be a big/complex feature.

  • Add a new operator that "expands" the input data for multiple output operators (two hash joins) and unions them together later (e.g. ExpandExec -> [HashJoinExec(a==b), HashJoinExec(c==d)] -> UnionExec). This avoids the duplicated scan / subquery which might be (much) slower in some cases. I think this will work pretty well in general and we probably could enable it by default?

Copy link
Copy Markdown
Contributor

@2010YOUY01 2010YOUY01 May 21, 2026 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think approach 1 means adding a new specialized hash join operator for ORed equality conditions:

SELECT *
FROM t1
JOIN t2
ON (t1.v1 = t2.v1) OR (t1.v2 = t2.v2)

We could add a specialized DisjointHashJoinExec operator. The core logic would look like:

  1. Buffer the build side and build hash tables for t1.v1 and t1.v2.
  2. Probe the other side. For each probe row, probe both hash tables, union and deduplicate the matched build-side row indices, and then materialize the joined rows.

The benefits are:

  • No repeated scans, which can be expensive for Parquet.
  • Can also support outer/semi/mark joins
  • Simpler planning.

I think this approach would be relatively straightforward to implement after:

I'm looking for help reviewing this feature 🎣 , would also be happy to help with follow-up work, such as extending this idea into a disjoint equi-join implementation.

Copy link
Copy Markdown
Contributor

@Dandandan Dandandan May 21, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah that is exactly what I mean, thanks @2010YOUY01 for explaining fully.

Copy link
Copy Markdown
Member Author

@xiedeyantu xiedeyantu May 21, 2026 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@2010YOUY01 Thank you so much for your detailed explanation of this logic! I think it's a very good idea, and for inner joins, this implementation is optimal, completing the entire logic directly at the physical execution layer. Regarding my current proposal, I support this implementation. However, since I'm not entirely clear on the execution-level logic, implementing it using DisjointHashJoinExec might take some time.

Actually, there's 2 PRs(apache/calcite#4300, apache/calcite#4315) I submitted to Calcite, This rule file (https://github.com/apache/calcite/blob/main/core/src/main/java/org/apache/calcite/rel/rules/JoinExpandOrToUnionRule.java) will be more intuitive. where I implemented inner/left/right/full/anti joins (I didn't implement semi-join because its semantics are not easily split into multiple mutually exclusive joins). Their methods for splitting multiple join branches differ (refer to the comments in the connection code above). This isn't easily implemented using DisjointHashJoinExec; for example, left joins would be split into inner and anti joins. I limited the PR to inner joins because I wanted to implement the first step first, as the performance improvement was significant when testing joins of two tables (1000+ rows). I can't construct SQL to test other scenarios yet, as they all involve anti joins. If everyone accepts this solution, I will expand it to support more join types later. This is why I implemented it as a separate rule. There's another reason, which I also mentioned above: when the table only has 10 rows of data, the overhead becomes apparent, slowing down this optimization. Therefore, parameters or statistical information can help decide whether to rewrite it.

Regarding your first question, I think we can achieve scan reuse at either the logical or physical layer; we don't need to worry too much about rewriting the logic.

Regarding the second question, I think it might be difficult for us to do, perhaps because my understanding of DataFusion's execution layer is still limited.

Regarding the third question, similar to the first, the complexity of the plan is not necessarily directly related to the actual execution logic or performance.

This is just my personal opinion; please correct me if I'm wrong. Thank you very much for participating in the discussion.

If we're only considering inner joins (without expanding to other join types), I personally like the second solution @Dandandan mentioned. @2010YOUY01 If you've already implemented it, then I'll close this PR. If you're interested in further expansions of this PR, I can implement these capabilities through multiple PRs. Looking forward to your reply!

Copy link
Copy Markdown
Contributor

@2010YOUY01 2010YOUY01 May 22, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@2010YOUY01 Thank you so much for your detailed explanation of this logic! I think it's a very good idea, and for inner joins, this implementation is optimal, completing the entire logic directly at the physical execution layer. Regarding my current proposal, I support this implementation. However, since I'm not entirely clear on the execution-level logic, implementing it using DisjointHashJoinExec might take some time.

This might not be obvious to everyone yet, but with #21983, extending this into a new specialized join operator should be easy, and it would automatically support outer/semi/... join types.

I can show this with a PoC later. Once we have that version, we can decide which approach to take.

Copy link
Copy Markdown
Member Author

@xiedeyantu xiedeyantu May 22, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@2010YOUY01 Thank you for your great idea. I look forward to your solution and will also study your existing solutions.

#[expect(missing_docs)]
pub fn new() -> Self {
Self {}
}
}

impl OptimizerRule for ExpandJoinOrPredicate {
fn name(&self) -> &str {
"expand_join_or_predicate"
}

fn supports_rewrite(&self) -> bool {
true
}

fn apply_order(&self) -> Option<ApplyOrder> {
Some(ApplyOrder::BottomUp)
}

fn rewrite(
&self,
plan: LogicalPlan,
_config: &dyn OptimizerConfig,
) -> Result<Transformed<LogicalPlan>> {
match plan {
LogicalPlan::Join(join) => rewrite_join(join),
_ => Ok(Transformed::no(plan)),
}
}
}

fn rewrite_join(join: Join) -> Result<Transformed<LogicalPlan>> {
let original_schema = Arc::clone(&join.schema);

if join.join_type != JoinType::Inner || join.null_aware {
return Ok(Transformed::no(LogicalPlan::Join(join)));
}

let Some(filter) = join.filter.clone() else {
return Ok(Transformed::no(LogicalPlan::Join(join)));
};

if filter.is_volatile() {
return Ok(Transformed::no(LogicalPlan::Join(join)));
}

let disjuncts = split_binary_owned(filter, Operator::Or);
if disjuncts.len() < 2 {
return Ok(Transformed::no(LogicalPlan::Join(join)));
}

let left_schema = join.left.schema();
let right_schema = join.right.schema();

let Some(branch_keys) = disjuncts
.iter()
.map(|expr| extract_hashjoin_keys(expr, left_schema, right_schema))
.collect::<Result<Option<Vec<Vec<(Expr, Expr)>>>>>()?
else {
return Ok(Transformed::no(LogicalPlan::Join(join)));
};

let mut guards = Vec::with_capacity(disjuncts.len().saturating_sub(1));
let mut branches = Vec::with_capacity(disjuncts.len());

for (disjunct, keys) in disjuncts.into_iter().zip(branch_keys.into_iter()) {
let branch_filter = guards
.iter()
.cloned()
.reduce(Expr::and);

let mut on = join.on.clone();
on.extend(keys);

let branch = LogicalPlan::Join(Join::try_new(
Arc::clone(&join.left),
Arc::clone(&join.right),
on,
branch_filter,
join.join_type,
join.join_constraint,
join.null_equality,
join.null_aware,
)?);

let branch = LogicalPlan::Projection(Projection::new_from_schema(
Arc::new(branch),
Arc::clone(&original_schema),
));

branches.push(Arc::new(branch));

guards.push(disjunct.is_not_true());
}

let rewritten = LogicalPlan::Union(Union {
inputs: branches,
schema: original_schema,
});

Ok(Transformed::yes(rewritten))
}

fn extract_hashjoin_keys(
expr: &Expr,
left_schema: &datafusion_common::DFSchema,
right_schema: &datafusion_common::DFSchema,
) -> Result<Option<Vec<(Expr, Expr)>>> {
let conjuncts = split_conjunction_owned(expr.clone());
let mut keys = Vec::with_capacity(conjuncts.len());

for conjunct in conjuncts {
let Expr::BinaryExpr(binary) = conjunct else {
return Ok(None);
};

if binary.op != Operator::Eq {
return Ok(None);
}

let Some((left, right)) = find_valid_equijoin_key_pair(
&binary.left,
&binary.right,
left_schema,
right_schema,
)? else {
return Ok(None);
};

let left_type = left.get_type(left_schema)?;
let right_type = right.get_type(right_schema)?;
if !can_hash(&left_type) || !can_hash(&right_type) {
return Ok(None);
}

keys.push((left, right));
}

Ok(Some(keys))
}
1 change: 1 addition & 0 deletions datafusion/optimizer/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,6 +56,7 @@ pub mod eliminate_nested_union {
pub type EliminateNestedUnion = OptimizeUnions;
}
pub mod eliminate_outer_join;
pub mod expand_join_or_predicate;
pub mod extract_equijoin_predicate;
pub mod extract_leaf_expressions;
pub mod filter_null_join_keys;
Expand Down
2 changes: 2 additions & 0 deletions datafusion/optimizer/src/optimizer.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@ use crate::eliminate_group_by_constant::EliminateGroupByConstant;
use crate::eliminate_join::EliminateJoin;
use crate::eliminate_limit::EliminateLimit;
use crate::eliminate_outer_join::EliminateOuterJoin;
use crate::expand_join_or_predicate::ExpandJoinOrPredicate;
use crate::extract_equijoin_predicate::ExtractEquijoinPredicate;
use crate::extract_leaf_expressions::{ExtractLeafExpressions, PushDownLeafProjections};
use crate::filter_null_join_keys::FilterNullJoinKeys;
Expand Down Expand Up @@ -289,6 +290,7 @@ impl Optimizer {
Arc::new(ScalarSubqueryToJoin::new()),
Arc::new(DecorrelateLateralJoin::new()),
Arc::new(ExtractEquijoinPredicate::new()),
Arc::new(ExpandJoinOrPredicate::new()),
Arc::new(EliminateDuplicatedExpr::new()),
Arc::new(EliminateFilter::new()),
Arc::new(EliminateCrossJoin::new()),
Expand Down
4 changes: 4 additions & 0 deletions datafusion/sqllogictest/test_files/explain.slt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -186,6 +186,7 @@ logical_plan after decorrelate_predicate_subquery SAME TEXT AS ABOVE
logical_plan after scalar_subquery_to_join SAME TEXT AS ABOVE
logical_plan after decorrelate_lateral_join SAME TEXT AS ABOVE
logical_plan after extract_equijoin_predicate SAME TEXT AS ABOVE
logical_plan after expand_join_or_predicate SAME TEXT AS ABOVE
logical_plan after eliminate_duplicated_expr SAME TEXT AS ABOVE
logical_plan after eliminate_filter SAME TEXT AS ABOVE
logical_plan after eliminate_cross_join SAME TEXT AS ABOVE
Expand All @@ -211,6 +212,7 @@ logical_plan after decorrelate_predicate_subquery SAME TEXT AS ABOVE
logical_plan after scalar_subquery_to_join SAME TEXT AS ABOVE
logical_plan after decorrelate_lateral_join SAME TEXT AS ABOVE
logical_plan after extract_equijoin_predicate SAME TEXT AS ABOVE
logical_plan after expand_join_or_predicate SAME TEXT AS ABOVE
logical_plan after eliminate_duplicated_expr SAME TEXT AS ABOVE
logical_plan after eliminate_filter SAME TEXT AS ABOVE
logical_plan after eliminate_cross_join SAME TEXT AS ABOVE
Expand Down Expand Up @@ -563,6 +565,7 @@ logical_plan after decorrelate_predicate_subquery SAME TEXT AS ABOVE
logical_plan after scalar_subquery_to_join SAME TEXT AS ABOVE
logical_plan after decorrelate_lateral_join SAME TEXT AS ABOVE
logical_plan after extract_equijoin_predicate SAME TEXT AS ABOVE
logical_plan after expand_join_or_predicate SAME TEXT AS ABOVE
logical_plan after eliminate_duplicated_expr SAME TEXT AS ABOVE
logical_plan after eliminate_filter SAME TEXT AS ABOVE
logical_plan after eliminate_cross_join SAME TEXT AS ABOVE
Expand All @@ -588,6 +591,7 @@ logical_plan after decorrelate_predicate_subquery SAME TEXT AS ABOVE
logical_plan after scalar_subquery_to_join SAME TEXT AS ABOVE
logical_plan after decorrelate_lateral_join SAME TEXT AS ABOVE
logical_plan after extract_equijoin_predicate SAME TEXT AS ABOVE
logical_plan after expand_join_or_predicate SAME TEXT AS ABOVE
logical_plan after eliminate_duplicated_expr SAME TEXT AS ABOVE
logical_plan after eliminate_filter SAME TEXT AS ABOVE
logical_plan after eliminate_cross_join SAME TEXT AS ABOVE
Expand Down
53 changes: 53 additions & 0 deletions datafusion/sqllogictest/test_files/join_only.slt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,3 +20,56 @@ include ./join.slt.part
# Config reset
statement ok
RESET datafusion.optimizer.repartition_joins;

# expand_join_or_predicate: rows that satisfy multiple OR branches should appear once
statement ok
CREATE TABLE IF NOT EXISTS join_or_left(lid INT, lalt INT) AS VALUES
(1, 10),
(2, 20),
(3, NULL);

statement ok
CREATE TABLE IF NOT EXISTS join_or_right(rid INT, ralt INT) AS VALUES
(1, 10),
(4, 20),
(3, 30);

query IIII rowsort
SELECT *
FROM join_or_left
INNER JOIN join_or_right
ON lid = rid OR lalt = ralt;
----
1 10 1 10
2 20 4 20
3 NULL 3 30

# expand_join_or_predicate: if the previous equality is NULL and the later branch is true,
# the row must still match
statement ok
CREATE TABLE IF NOT EXISTS join_or_null_left(la INT, lb INT) AS VALUES
(NULL, 1);

statement ok
CREATE TABLE IF NOT EXISTS join_or_null_right(ra INT, rb INT) AS VALUES
(NULL, 1);

query IIII rowsort
SELECT *
FROM join_or_null_left
INNER JOIN join_or_null_right
ON la = ra OR lb = rb;
----
NULL 1 NULL 1

statement ok
DROP TABLE IF EXISTS join_or_left;

statement ok
DROP TABLE IF EXISTS join_or_right;

statement ok
DROP TABLE IF EXISTS join_or_null_left;

statement ok
DROP TABLE IF EXISTS join_or_null_right;
Loading