datafusion_optimizer/
eliminate_cross_join.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18//! [`EliminateCrossJoin`] converts `CROSS JOIN` to `INNER JOIN` if join predicates are available.
19use crate::{OptimizerConfig, OptimizerRule};
20use std::sync::Arc;
21
22use crate::join_key_set::JoinKeySet;
23use datafusion_common::tree_node::{Transformed, TreeNode, TreeNodeRecursion};
24use datafusion_common::{NullEquality, Result};
25use datafusion_expr::expr::{BinaryExpr, Expr};
26use datafusion_expr::logical_plan::{
27    Filter, Join, JoinConstraint, JoinType, LogicalPlan, Projection,
28};
29use datafusion_expr::utils::{can_hash, find_valid_equijoin_key_pair};
30use datafusion_expr::{ExprSchemable, Operator, and, build_join_schema};
31
32#[derive(Default, Debug)]
33pub struct EliminateCrossJoin;
34
35impl EliminateCrossJoin {
36    #[expect(missing_docs)]
37    pub fn new() -> Self {
38        Self {}
39    }
40}
41
42/// Eliminate cross joins by rewriting them to inner joins when possible.
43///
44/// # Example
45/// The initial plan for this query:
46/// ```sql
47/// select ... from a, b where a.x = b.y and b.xx = 100;
48/// ```
49///
50/// Looks like this:
51/// ```text
52/// Filter(a.x = b.y AND b.xx = 100)
53///  Cross Join
54///   TableScan a
55///   TableScan b
56/// ```
57///
58/// After the rule is applied, the plan will look like this:
59/// ```text
60/// Filter(b.xx = 100)
61///   InnerJoin(a.x = b.y)
62///     TableScan a
63///     TableScan b
64/// ```
65///
66/// # Other Examples
67/// * 'select ... from a, b where a.x = b.y and b.xx = 100;'
68/// * 'select ... from a, b where (a.x = b.y and b.xx = 100) or (a.x = b.y and b.xx = 200);'
69/// * 'select ... from a, b, c where (a.x = b.y and b.xx = 100 and a.z = c.z)
70/// * or (a.x = b.y and b.xx = 200 and a.z=c.z);'
71/// * 'select ... from a, b where a.x > b.y'
72///
73/// For above queries, the join predicate is available in filters and they are moved to
74/// join nodes appropriately
75///
76/// This fix helps to improve the performance of TPCH Q19. issue#78
77impl OptimizerRule for EliminateCrossJoin {
78    fn supports_rewrite(&self) -> bool {
79        true
80    }
81
82    #[cfg_attr(feature = "recursive_protection", recursive::recursive)]
83    fn rewrite(
84        &self,
85        plan: LogicalPlan,
86        config: &dyn OptimizerConfig,
87    ) -> Result<Transformed<LogicalPlan>> {
88        // Fast path: nothing to do if the plan contains no `Join` nodes.
89        // Without this guard the rule still falls through to
90        // `rewrite_children`, which walks the entire plan, processes
91        // uncorrelated subqueries, and rewrites every direct child via
92        // `map_children` (clone-on-write) — paid by every query in the
93        // logical optimizer pipeline. Same shape as the
94        // `plan_has_subqueries` fast-path landed in #22298.
95        if !plan_has_joins(&plan) {
96            return Ok(Transformed::no(plan));
97        }
98
99        let plan_schema = Arc::clone(plan.schema());
100        let mut possible_join_keys = JoinKeySet::new();
101        let mut all_inputs: Vec<LogicalPlan> = vec![];
102        let mut all_filters: Vec<Expr> = vec![];
103        let mut null_equality = NullEquality::NullEqualsNothing;
104
105        let parent_predicate = if let LogicalPlan::Filter(filter) = plan {
106            // if input isn't a join that can potentially be rewritten
107            // avoid unwrapping the input
108            let rewritable = matches!(
109                filter.input.as_ref(),
110                LogicalPlan::Join(Join {
111                    join_type: JoinType::Inner,
112                    ..
113                })
114            );
115
116            if !rewritable {
117                // recursively try to rewrite children
118                return rewrite_children(self, LogicalPlan::Filter(filter), config);
119            }
120
121            if !can_flatten_join_inputs(&filter.input) {
122                return Ok(Transformed::no(LogicalPlan::Filter(filter)));
123            }
124
125            let Filter {
126                input, predicate, ..
127            } = filter;
128
129            // Extract null_equality setting from the input join
130            if let LogicalPlan::Join(join) = input.as_ref() {
131                null_equality = join.null_equality;
132            }
133
134            flatten_join_inputs(
135                Arc::unwrap_or_clone(input),
136                &mut possible_join_keys,
137                &mut all_inputs,
138                &mut all_filters,
139            )?;
140
141            extract_possible_join_keys(&predicate, &mut possible_join_keys);
142            Some(predicate)
143        } else {
144            match plan {
145                LogicalPlan::Join(Join {
146                    join_type: JoinType::Inner,
147                    null_equality: original_null_equality,
148                    ..
149                }) => {
150                    if !can_flatten_join_inputs(&plan) {
151                        return Ok(Transformed::no(plan));
152                    }
153                    flatten_join_inputs(
154                        plan,
155                        &mut possible_join_keys,
156                        &mut all_inputs,
157                        &mut all_filters,
158                    )?;
159                    null_equality = original_null_equality;
160                    None
161                }
162                _ => {
163                    // recursively try to rewrite children
164                    return rewrite_children(self, plan, config);
165                }
166            }
167        };
168
169        // Join keys are handled locally:
170        let mut all_join_keys = JoinKeySet::new();
171        let mut left = all_inputs.remove(0);
172        while !all_inputs.is_empty() {
173            left = find_inner_join(
174                left,
175                &mut all_inputs,
176                &possible_join_keys,
177                &mut all_join_keys,
178                null_equality,
179            )?;
180        }
181
182        left = rewrite_children(self, left, config)?.data;
183
184        if &plan_schema != left.schema() {
185            left = LogicalPlan::Projection(Projection::new_from_schema(
186                Arc::new(left),
187                Arc::clone(&plan_schema),
188            ));
189        }
190
191        if !all_filters.is_empty() {
192            // Add any filters on top - PushDownFilter can push filters down to applicable join
193            let first = all_filters.swap_remove(0);
194            let predicate = all_filters.into_iter().fold(first, and);
195            left = LogicalPlan::Filter(Filter::try_new(predicate, Arc::new(left))?);
196        }
197
198        let Some(predicate) = parent_predicate else {
199            return Ok(Transformed::yes(left));
200        };
201
202        // If there are no join keys then do nothing:
203        if all_join_keys.is_empty() {
204            Filter::try_new(predicate, Arc::new(left))
205                .map(|filter| Transformed::yes(LogicalPlan::Filter(filter)))
206        } else {
207            // Remove join expressions from filter:
208            match remove_join_expressions(predicate, &all_join_keys) {
209                Some(filter_expr) => Filter::try_new(filter_expr, Arc::new(left))
210                    .map(|filter| Transformed::yes(LogicalPlan::Filter(filter))),
211                _ => Ok(Transformed::yes(left)),
212            }
213        }
214    }
215
216    fn name(&self) -> &str {
217        "eliminate_cross_join"
218    }
219}
220
221/// Returns `true` if `plan` contains at least one [`LogicalPlan::Join`]
222/// node, either directly in its tree *or* inside an embedded subquery
223/// plan reachable through `Expr::ScalarSubquery` / `Expr::InSubquery`
224/// / `Expr::Exists` / `Expr::SetComparison`.
225///
226/// Used as a fast-path gate at the top of [`EliminateCrossJoin::rewrite`]
227/// so that join-free plans skip the full recursive rewrite. Subquery
228/// traversal matters because `rewrite_children` also dives into
229/// uncorrelated subqueries via `map_uncorrelated_subqueries`; ignoring
230/// them here would skip optimizing a `CROSS JOIN` that sits only inside
231/// an `IN (SELECT ... FROM a, b)`-style predicate.
232///
233/// `LogicalPlan::apply_with_subqueries` already implements the
234/// "walk this node + every child + every subquery plan" traversal we
235/// need, so the helper is a thin wrapper around it.
236fn plan_has_joins(plan: &LogicalPlan) -> bool {
237    let mut found = false;
238    let _ = plan.apply_with_subqueries(|node| {
239        if matches!(node, LogicalPlan::Join(_)) {
240            found = true;
241            Ok(TreeNodeRecursion::Stop)
242        } else {
243            Ok(TreeNodeRecursion::Continue)
244        }
245    });
246    found
247}
248
249fn rewrite_children(
250    optimizer: &impl OptimizerRule,
251    plan: LogicalPlan,
252    config: &dyn OptimizerConfig,
253) -> Result<Transformed<LogicalPlan>> {
254    // Process uncorrelated subqueries in expressions, then direct children.
255    let transformed_plan = plan
256        .map_uncorrelated_subqueries(|input| optimizer.rewrite(input, config))?
257        .transform_sibling(|plan| {
258            plan.map_children(|input| optimizer.rewrite(input, config))
259        })?;
260
261    // recompute schema if the plan was transformed
262    if transformed_plan.transformed {
263        transformed_plan.map_data(|plan| plan.recompute_schema())
264    } else {
265        Ok(transformed_plan)
266    }
267}
268
269/// Recursively accumulate possible_join_keys and inputs from inner joins
270/// (including cross joins).
271///
272/// Assumes can_flatten_join_inputs has returned true and thus the plan can be
273/// flattened. Adds all leaf inputs to `all_inputs` and join_keys to
274/// possible_join_keys
275fn flatten_join_inputs(
276    plan: LogicalPlan,
277    possible_join_keys: &mut JoinKeySet,
278    all_inputs: &mut Vec<LogicalPlan>,
279    all_filters: &mut Vec<Expr>,
280) -> Result<()> {
281    match plan {
282        LogicalPlan::Join(join) if join.join_type == JoinType::Inner => {
283            if let Some(filter) = join.filter {
284                all_filters.push(filter);
285            }
286            possible_join_keys.insert_all_owned(join.on);
287            flatten_join_inputs(
288                Arc::unwrap_or_clone(join.left),
289                possible_join_keys,
290                all_inputs,
291                all_filters,
292            )?;
293            flatten_join_inputs(
294                Arc::unwrap_or_clone(join.right),
295                possible_join_keys,
296                all_inputs,
297                all_filters,
298            )?;
299        }
300        _ => {
301            all_inputs.push(plan);
302        }
303    };
304    Ok(())
305}
306
307/// Returns true if the plan is a Join or Cross join could be flattened with
308/// `flatten_join_inputs`
309///
310/// Must stay in sync with `flatten_join_inputs`
311fn can_flatten_join_inputs(plan: &LogicalPlan) -> bool {
312    // can only flatten inner / cross joins
313    match plan {
314        LogicalPlan::Join(join) if join.join_type == JoinType::Inner => {}
315        _ => return false,
316    };
317
318    for child in plan.inputs() {
319        if let LogicalPlan::Join(Join {
320            join_type: JoinType::Inner,
321            ..
322        }) = child
323            && !can_flatten_join_inputs(child)
324        {
325            return false;
326        }
327    }
328    true
329}
330
331/// Finds the next to join with the left input plan,
332///
333/// Finds the next `right` from `rights` that can be joined with `left_input`
334/// plan based on the join keys in `possible_join_keys`.
335///
336/// If such a matching `right` is found:
337/// 1. Adds the matching join keys to `all_join_keys`.
338/// 2. Returns `left_input JOIN right ON (all join keys)`.
339///
340/// If no matching `right` is found:
341/// 1. Removes the first plan from `rights`
342/// 2. Returns `left_input CROSS JOIN right`.
343fn find_inner_join(
344    left_input: LogicalPlan,
345    rights: &mut Vec<LogicalPlan>,
346    possible_join_keys: &JoinKeySet,
347    all_join_keys: &mut JoinKeySet,
348    null_equality: NullEquality,
349) -> Result<LogicalPlan> {
350    for (i, right_input) in rights.iter().enumerate() {
351        let mut join_keys = vec![];
352
353        for (l, r) in possible_join_keys.iter() {
354            let key_pair = find_valid_equijoin_key_pair(
355                l,
356                r,
357                left_input.schema(),
358                right_input.schema(),
359            )?;
360
361            // Save join keys
362            if let Some((valid_l, valid_r)) = key_pair
363                && can_hash(&valid_l.get_type(left_input.schema())?)
364            {
365                join_keys.push((valid_l, valid_r));
366            }
367        }
368
369        // Found one or more matching join keys
370        if !join_keys.is_empty() {
371            all_join_keys.insert_all(join_keys.iter());
372            let right_input = rights.remove(i);
373            let join_schema = Arc::new(build_join_schema(
374                left_input.schema(),
375                right_input.schema(),
376                &JoinType::Inner,
377            )?);
378
379            return Ok(LogicalPlan::Join(Join {
380                left: Arc::new(left_input),
381                right: Arc::new(right_input),
382                join_type: JoinType::Inner,
383                join_constraint: JoinConstraint::On,
384                on: join_keys,
385                filter: None,
386                schema: join_schema,
387                null_equality,
388                null_aware: false,
389            }));
390        }
391    }
392
393    // no matching right plan had any join keys, cross join with the first right
394    // plan
395    let right = rights.remove(0);
396    let join_schema = Arc::new(build_join_schema(
397        left_input.schema(),
398        right.schema(),
399        &JoinType::Inner,
400    )?);
401
402    Ok(LogicalPlan::Join(Join {
403        left: Arc::new(left_input),
404        right: Arc::new(right),
405        schema: join_schema,
406        on: vec![],
407        filter: None,
408        join_type: JoinType::Inner,
409        join_constraint: JoinConstraint::On,
410        null_equality,
411        null_aware: false,
412    }))
413}
414
415/// Extract join keys from a WHERE clause
416fn extract_possible_join_keys(expr: &Expr, join_keys: &mut JoinKeySet) {
417    if let Expr::BinaryExpr(BinaryExpr { left, op, right }) = expr {
418        match op {
419            Operator::Eq => {
420                // insert handles ensuring  we don't add the same Join keys multiple times
421                join_keys.insert(left, right);
422            }
423            Operator::And => {
424                extract_possible_join_keys(left, join_keys);
425                extract_possible_join_keys(right, join_keys)
426            }
427            // Fix for issue#78 join predicates from inside of OR expr also pulled up properly.
428            Operator::Or => {
429                let mut left_join_keys = JoinKeySet::new();
430                let mut right_join_keys = JoinKeySet::new();
431
432                extract_possible_join_keys(left, &mut left_join_keys);
433                extract_possible_join_keys(right, &mut right_join_keys);
434
435                join_keys.insert_intersection(&left_join_keys, &right_join_keys)
436            }
437            _ => (),
438        };
439    }
440}
441
442/// Remove join expressions from a filter expression
443///
444/// # Returns
445/// * `Some()` when there are few remaining predicates in filter_expr
446/// * `None` otherwise
447fn remove_join_expressions(expr: Expr, join_keys: &JoinKeySet) -> Option<Expr> {
448    match expr {
449        Expr::BinaryExpr(BinaryExpr {
450            left,
451            op: Operator::Eq,
452            right,
453        }) if join_keys.contains(&left, &right) => {
454            // was a join key, so remove it
455            None
456        }
457        // Fix for issue#78 join predicates from inside of OR expr also pulled up properly.
458        Expr::BinaryExpr(BinaryExpr { left, op, right }) if op == Operator::And => {
459            let l = remove_join_expressions(*left, join_keys);
460            let r = remove_join_expressions(*right, join_keys);
461            match (l, r) {
462                (Some(ll), Some(rr)) => Some(Expr::BinaryExpr(BinaryExpr::new(
463                    Box::new(ll),
464                    op,
465                    Box::new(rr),
466                ))),
467                (Some(ll), _) => Some(ll),
468                (_, Some(rr)) => Some(rr),
469                _ => None,
470            }
471        }
472        Expr::BinaryExpr(BinaryExpr { left, op, right }) if op == Operator::Or => {
473            let l = remove_join_expressions(*left, join_keys);
474            let r = remove_join_expressions(*right, join_keys);
475            match (l, r) {
476                (Some(ll), Some(rr)) => Some(Expr::BinaryExpr(BinaryExpr::new(
477                    Box::new(ll),
478                    op,
479                    Box::new(rr),
480                ))),
481                // When either `left` or `right` is empty, it means they are `true`
482                // so OR'ing anything with them will also be true
483                _ => None,
484            }
485        }
486        _ => Some(expr),
487    }
488}
489
490#[cfg(test)]
491mod tests {
492    use super::*;
493    use crate::optimizer::OptimizerContext;
494    use crate::test::*;
495
496    use datafusion_expr::{
497        Operator::{And, Or},
498        binary_expr, col, lit,
499        logical_plan::builder::LogicalPlanBuilder,
500    };
501    use insta::assert_snapshot;
502
503    macro_rules! assert_optimized_plan_equal {
504        (
505            $plan:expr,
506            @ $expected:literal $(,)?
507        ) => {{
508            let starting_schema = Arc::clone($plan.schema());
509            let rule = EliminateCrossJoin::new();
510            let Transformed {transformed: is_plan_transformed, data: optimized_plan, ..} = rule.rewrite($plan, &OptimizerContext::new()).unwrap();
511            let formatted_plan = optimized_plan.display_indent_schema();
512            // Ensure the rule was actually applied
513            assert!(is_plan_transformed, "failed to optimize plan");
514            // Verify the schema remains unchanged
515            assert_eq!(&starting_schema, optimized_plan.schema());
516            assert_snapshot!(
517                formatted_plan,
518                @ $expected,
519            );
520
521            Ok(())
522        }};
523    }
524
525    #[test]
526    fn eliminate_cross_with_simple_and() -> Result<()> {
527        let t1 = test_table_scan_with_name("t1")?;
528        let t2 = test_table_scan_with_name("t2")?;
529
530        // could eliminate to inner join since filter has Join predicates
531        let plan = LogicalPlanBuilder::from(t1)
532            .cross_join(t2)?
533            .filter(binary_expr(
534                col("t1.a").eq(col("t2.a")),
535                And,
536                col("t2.c").lt(lit(20u32)),
537            ))?
538            .build()?;
539
540        assert_optimized_plan_equal!(
541            plan,
542            @ r"
543        Filter: t2.c < UInt32(20) [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
544          Inner Join: t1.a = t2.a [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
545            TableScan: t1 [a:UInt32, b:UInt32, c:UInt32]
546            TableScan: t2 [a:UInt32, b:UInt32, c:UInt32]
547        "
548        )
549    }
550
551    #[test]
552    fn eliminate_cross_with_simple_or() -> Result<()> {
553        let t1 = test_table_scan_with_name("t1")?;
554        let t2 = test_table_scan_with_name("t2")?;
555
556        // could not eliminate to inner join since filter OR expression and there is no common
557        // Join predicates in left and right of OR expr.
558        let plan = LogicalPlanBuilder::from(t1)
559            .cross_join(t2)?
560            .filter(binary_expr(
561                col("t1.a").eq(col("t2.a")),
562                Or,
563                col("t2.b").eq(col("t1.a")),
564            ))?
565            .build()?;
566
567        assert_optimized_plan_equal!(
568            plan,
569            @ r"
570        Filter: t1.a = t2.a OR t2.b = t1.a [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
571          Cross Join: [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
572            TableScan: t1 [a:UInt32, b:UInt32, c:UInt32]
573            TableScan: t2 [a:UInt32, b:UInt32, c:UInt32]
574        "
575        )
576    }
577
578    #[test]
579    fn eliminate_cross_with_and() -> Result<()> {
580        let t1 = test_table_scan_with_name("t1")?;
581        let t2 = test_table_scan_with_name("t2")?;
582
583        // could eliminate to inner join
584        let plan = LogicalPlanBuilder::from(t1)
585            .cross_join(t2)?
586            .filter(binary_expr(
587                binary_expr(col("t1.a").eq(col("t2.a")), And, col("t2.c").lt(lit(20u32))),
588                And,
589                binary_expr(col("t1.a").eq(col("t2.a")), And, col("t2.c").eq(lit(10u32))),
590            ))?
591            .build()?;
592
593        assert_optimized_plan_equal!(
594            plan,
595            @ r"
596        Filter: t2.c < UInt32(20) AND t2.c = UInt32(10) [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
597          Inner Join: t1.a = t2.a [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
598            TableScan: t1 [a:UInt32, b:UInt32, c:UInt32]
599            TableScan: t2 [a:UInt32, b:UInt32, c:UInt32]
600        "
601        )
602    }
603
604    #[test]
605    fn eliminate_cross_with_or() -> Result<()> {
606        let t1 = test_table_scan_with_name("t1")?;
607        let t2 = test_table_scan_with_name("t2")?;
608
609        // could eliminate to inner join since Or predicates have common Join predicates
610        let plan = LogicalPlanBuilder::from(t1)
611            .cross_join(t2)?
612            .filter(binary_expr(
613                binary_expr(col("t1.a").eq(col("t2.a")), And, col("t2.c").lt(lit(15u32))),
614                Or,
615                binary_expr(
616                    col("t1.a").eq(col("t2.a")),
617                    And,
618                    col("t2.c").eq(lit(688u32)),
619                ),
620            ))?
621            .build()?;
622
623        assert_optimized_plan_equal!(
624            plan,
625            @ r"
626        Filter: t2.c < UInt32(15) OR t2.c = UInt32(688) [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
627          Inner Join: t1.a = t2.a [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
628            TableScan: t1 [a:UInt32, b:UInt32, c:UInt32]
629            TableScan: t2 [a:UInt32, b:UInt32, c:UInt32]
630        "
631        )
632    }
633
634    #[test]
635    fn eliminate_cross_not_possible_simple() -> Result<()> {
636        let t1 = test_table_scan_with_name("t1")?;
637        let t2 = test_table_scan_with_name("t2")?;
638
639        // could not eliminate to inner join
640        let plan = LogicalPlanBuilder::from(t1)
641            .cross_join(t2)?
642            .filter(binary_expr(
643                binary_expr(col("t1.a").eq(col("t2.a")), And, col("t2.c").lt(lit(15u32))),
644                Or,
645                binary_expr(
646                    col("t1.b").eq(col("t2.b")),
647                    And,
648                    col("t2.c").eq(lit(688u32)),
649                ),
650            ))?
651            .build()?;
652
653        assert_optimized_plan_equal!(
654            plan,
655            @ r"
656        Filter: t1.a = t2.a AND t2.c < UInt32(15) OR t1.b = t2.b AND t2.c = UInt32(688) [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
657          Cross Join: [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
658            TableScan: t1 [a:UInt32, b:UInt32, c:UInt32]
659            TableScan: t2 [a:UInt32, b:UInt32, c:UInt32]
660        "
661        )
662    }
663
664    #[test]
665    fn eliminate_cross_not_possible() -> Result<()> {
666        let t1 = test_table_scan_with_name("t1")?;
667        let t2 = test_table_scan_with_name("t2")?;
668
669        // could not eliminate to inner join
670        let plan = LogicalPlanBuilder::from(t1)
671            .cross_join(t2)?
672            .filter(binary_expr(
673                binary_expr(col("t1.a").eq(col("t2.a")), And, col("t2.c").lt(lit(15u32))),
674                Or,
675                binary_expr(col("t1.a").eq(col("t2.a")), Or, col("t2.c").eq(lit(688u32))),
676            ))?
677            .build()?;
678
679        assert_optimized_plan_equal!(
680            plan,
681            @ r"
682        Filter: t1.a = t2.a AND t2.c < UInt32(15) OR t1.a = t2.a OR t2.c = UInt32(688) [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
683          Cross Join: [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
684            TableScan: t1 [a:UInt32, b:UInt32, c:UInt32]
685            TableScan: t2 [a:UInt32, b:UInt32, c:UInt32]
686        "
687        )
688    }
689
690    #[test]
691    fn eliminate_cross_possible_nested_inner_join_with_filter() -> Result<()> {
692        let t1 = test_table_scan_with_name("t1")?;
693        let t2 = test_table_scan_with_name("t2")?;
694        let t3 = test_table_scan_with_name("t3")?;
695
696        // could not eliminate to inner join with filter
697        let plan = LogicalPlanBuilder::from(t1)
698            .join(
699                t3,
700                JoinType::Inner,
701                (vec!["t1.a"], vec!["t3.a"]),
702                Some(col("t1.a").gt(lit(20u32))),
703            )?
704            .join(t2, JoinType::Inner, (vec!["t1.a"], vec!["t2.a"]), None)?
705            .filter(col("t1.a").gt(lit(15u32)))?
706            .build()?;
707
708        assert_optimized_plan_equal!(
709            plan,
710            @ r"
711        Filter: t1.a > UInt32(15) [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
712          Filter: t1.a > UInt32(20) [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
713            Inner Join: t1.a = t2.a [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
714              Inner Join: t1.a = t3.a [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
715                TableScan: t1 [a:UInt32, b:UInt32, c:UInt32]
716                TableScan: t3 [a:UInt32, b:UInt32, c:UInt32]
717              TableScan: t2 [a:UInt32, b:UInt32, c:UInt32]
718        "
719        )
720    }
721
722    #[test]
723    /// ```txt
724    /// filter: a.id = b.id and a.id = c.id
725    ///   cross_join a (bc)
726    ///     cross_join b c
727    /// ```
728    /// Without reorder, it will be
729    /// ```txt
730    ///   inner_join a (bc) on a.id = b.id and a.id = c.id
731    ///     cross_join b c
732    /// ```
733    /// Reorder it to be
734    /// ```txt
735    ///   inner_join (ab)c and a.id = c.id
736    ///     inner_join a b on a.id = b.id
737    /// ```
738    fn reorder_join_to_eliminate_cross_join_multi_tables() -> Result<()> {
739        let t1 = test_table_scan_with_name("t1")?;
740        let t2 = test_table_scan_with_name("t2")?;
741        let t3 = test_table_scan_with_name("t3")?;
742
743        // could eliminate to inner join
744        let plan = LogicalPlanBuilder::from(t1)
745            .cross_join(t2)?
746            .cross_join(t3)?
747            .filter(binary_expr(
748                binary_expr(col("t3.a").eq(col("t1.a")), And, col("t3.c").lt(lit(15u32))),
749                And,
750                binary_expr(col("t3.a").eq(col("t2.a")), And, col("t3.b").lt(lit(15u32))),
751            ))?
752            .build()?;
753
754        assert_optimized_plan_equal!(
755            plan,
756            @ r"
757        Filter: t3.c < UInt32(15) AND t3.b < UInt32(15) [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
758          Projection: t1.a, t1.b, t1.c, t2.a, t2.b, t2.c, t3.a, t3.b, t3.c [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
759            Inner Join: t3.a = t2.a [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
760              Inner Join: t1.a = t3.a [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
761                TableScan: t1 [a:UInt32, b:UInt32, c:UInt32]
762                TableScan: t3 [a:UInt32, b:UInt32, c:UInt32]
763              TableScan: t2 [a:UInt32, b:UInt32, c:UInt32]
764        "
765        )
766    }
767
768    #[test]
769    fn eliminate_cross_join_multi_tables() -> Result<()> {
770        let t1 = test_table_scan_with_name("t1")?;
771        let t2 = test_table_scan_with_name("t2")?;
772        let t3 = test_table_scan_with_name("t3")?;
773        let t4 = test_table_scan_with_name("t4")?;
774
775        // could eliminate to inner join
776        let plan1 = LogicalPlanBuilder::from(t1)
777            .cross_join(t2)?
778            .filter(binary_expr(
779                binary_expr(col("t1.a").eq(col("t2.a")), And, col("t2.c").lt(lit(15u32))),
780                Or,
781                binary_expr(
782                    col("t1.a").eq(col("t2.a")),
783                    And,
784                    col("t2.c").eq(lit(688u32)),
785                ),
786            ))?
787            .build()?;
788
789        let plan2 = LogicalPlanBuilder::from(t3)
790            .cross_join(t4)?
791            .filter(binary_expr(
792                binary_expr(
793                    binary_expr(
794                        col("t3.a").eq(col("t4.a")),
795                        And,
796                        col("t4.c").lt(lit(15u32)),
797                    ),
798                    Or,
799                    binary_expr(
800                        col("t3.a").eq(col("t4.a")),
801                        And,
802                        col("t3.c").eq(lit(688u32)),
803                    ),
804                ),
805                Or,
806                binary_expr(
807                    col("t3.a").eq(col("t4.a")),
808                    And,
809                    col("t3.b").eq(col("t4.b")),
810                ),
811            ))?
812            .build()?;
813
814        let plan = LogicalPlanBuilder::from(plan1)
815            .cross_join(plan2)?
816            .filter(binary_expr(
817                binary_expr(col("t3.a").eq(col("t1.a")), And, col("t4.c").lt(lit(15u32))),
818                Or,
819                binary_expr(
820                    col("t3.a").eq(col("t1.a")),
821                    And,
822                    col("t4.c").eq(lit(688u32)),
823                ),
824            ))?
825            .build()?;
826
827        assert_optimized_plan_equal!(
828            plan,
829            @ r"
830        Filter: t4.c < UInt32(15) OR t4.c = UInt32(688) [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
831          Inner Join: t1.a = t3.a [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
832            Filter: t2.c < UInt32(15) OR t2.c = UInt32(688) [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
833              Inner Join: t1.a = t2.a [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
834                TableScan: t1 [a:UInt32, b:UInt32, c:UInt32]
835                TableScan: t2 [a:UInt32, b:UInt32, c:UInt32]
836            Filter: t4.c < UInt32(15) OR t3.c = UInt32(688) OR t3.b = t4.b [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
837              Inner Join: t3.a = t4.a [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
838                TableScan: t3 [a:UInt32, b:UInt32, c:UInt32]
839                TableScan: t4 [a:UInt32, b:UInt32, c:UInt32]
840        "
841        )
842    }
843
844    #[test]
845    fn eliminate_cross_join_multi_tables_1() -> Result<()> {
846        let t1 = test_table_scan_with_name("t1")?;
847        let t2 = test_table_scan_with_name("t2")?;
848        let t3 = test_table_scan_with_name("t3")?;
849        let t4 = test_table_scan_with_name("t4")?;
850
851        // could eliminate to inner join
852        let plan1 = LogicalPlanBuilder::from(t1)
853            .cross_join(t2)?
854            .filter(binary_expr(
855                binary_expr(col("t1.a").eq(col("t2.a")), And, col("t2.c").lt(lit(15u32))),
856                Or,
857                binary_expr(
858                    col("t1.a").eq(col("t2.a")),
859                    And,
860                    col("t2.c").eq(lit(688u32)),
861                ),
862            ))?
863            .build()?;
864
865        // could eliminate to inner join
866        let plan2 = LogicalPlanBuilder::from(t3)
867            .cross_join(t4)?
868            .filter(binary_expr(
869                binary_expr(
870                    binary_expr(
871                        col("t3.a").eq(col("t4.a")),
872                        And,
873                        col("t4.c").lt(lit(15u32)),
874                    ),
875                    Or,
876                    binary_expr(
877                        col("t3.a").eq(col("t4.a")),
878                        And,
879                        col("t3.c").eq(lit(688u32)),
880                    ),
881                ),
882                Or,
883                binary_expr(
884                    col("t3.a").eq(col("t4.a")),
885                    And,
886                    col("t3.b").eq(col("t4.b")),
887                ),
888            ))?
889            .build()?;
890
891        // could not eliminate to inner join
892        let plan = LogicalPlanBuilder::from(plan1)
893            .cross_join(plan2)?
894            .filter(binary_expr(
895                binary_expr(col("t3.a").eq(col("t1.a")), And, col("t4.c").lt(lit(15u32))),
896                Or,
897                binary_expr(col("t3.a").eq(col("t1.a")), Or, col("t4.c").eq(lit(688u32))),
898            ))?
899            .build()?;
900
901        assert_optimized_plan_equal!(
902            plan,
903            @ r"
904        Filter: t3.a = t1.a AND t4.c < UInt32(15) OR t3.a = t1.a OR t4.c = UInt32(688) [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
905          Cross Join: [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
906            Filter: t2.c < UInt32(15) OR t2.c = UInt32(688) [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
907              Inner Join: t1.a = t2.a [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
908                TableScan: t1 [a:UInt32, b:UInt32, c:UInt32]
909                TableScan: t2 [a:UInt32, b:UInt32, c:UInt32]
910            Filter: t4.c < UInt32(15) OR t3.c = UInt32(688) OR t3.b = t4.b [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
911              Inner Join: t3.a = t4.a [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
912                TableScan: t3 [a:UInt32, b:UInt32, c:UInt32]
913                TableScan: t4 [a:UInt32, b:UInt32, c:UInt32]
914        "
915        )
916    }
917
918    #[test]
919    fn eliminate_cross_join_multi_tables_2() -> Result<()> {
920        let t1 = test_table_scan_with_name("t1")?;
921        let t2 = test_table_scan_with_name("t2")?;
922        let t3 = test_table_scan_with_name("t3")?;
923        let t4 = test_table_scan_with_name("t4")?;
924
925        // could eliminate to inner join
926        let plan1 = LogicalPlanBuilder::from(t1)
927            .cross_join(t2)?
928            .filter(binary_expr(
929                binary_expr(col("t1.a").eq(col("t2.a")), And, col("t2.c").lt(lit(15u32))),
930                Or,
931                binary_expr(
932                    col("t1.a").eq(col("t2.a")),
933                    And,
934                    col("t2.c").eq(lit(688u32)),
935                ),
936            ))?
937            .build()?;
938
939        // could not eliminate to inner join
940        let plan2 = LogicalPlanBuilder::from(t3)
941            .cross_join(t4)?
942            .filter(binary_expr(
943                binary_expr(
944                    binary_expr(
945                        col("t3.a").eq(col("t4.a")),
946                        And,
947                        col("t4.c").lt(lit(15u32)),
948                    ),
949                    Or,
950                    binary_expr(
951                        col("t3.a").eq(col("t4.a")),
952                        And,
953                        col("t3.c").eq(lit(688u32)),
954                    ),
955                ),
956                Or,
957                binary_expr(col("t3.a").eq(col("t4.a")), Or, col("t3.b").eq(col("t4.b"))),
958            ))?
959            .build()?;
960
961        // could eliminate to inner join
962        let plan = LogicalPlanBuilder::from(plan1)
963            .cross_join(plan2)?
964            .filter(binary_expr(
965                binary_expr(col("t3.a").eq(col("t1.a")), And, col("t4.c").lt(lit(15u32))),
966                Or,
967                binary_expr(
968                    col("t3.a").eq(col("t1.a")),
969                    And,
970                    col("t4.c").eq(lit(688u32)),
971                ),
972            ))?
973            .build()?;
974
975        assert_optimized_plan_equal!(
976            plan,
977            @ r"
978        Filter: t4.c < UInt32(15) OR t4.c = UInt32(688) [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
979          Inner Join: t1.a = t3.a [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
980            Filter: t2.c < UInt32(15) OR t2.c = UInt32(688) [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
981              Inner Join: t1.a = t2.a [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
982                TableScan: t1 [a:UInt32, b:UInt32, c:UInt32]
983                TableScan: t2 [a:UInt32, b:UInt32, c:UInt32]
984            Filter: t3.a = t4.a AND t4.c < UInt32(15) OR t3.a = t4.a AND t3.c = UInt32(688) OR t3.a = t4.a OR t3.b = t4.b [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
985              Cross Join: [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
986                TableScan: t3 [a:UInt32, b:UInt32, c:UInt32]
987                TableScan: t4 [a:UInt32, b:UInt32, c:UInt32]
988        "
989        )
990    }
991
992    #[test]
993    fn eliminate_cross_join_multi_tables_3() -> Result<()> {
994        let t1 = test_table_scan_with_name("t1")?;
995        let t2 = test_table_scan_with_name("t2")?;
996        let t3 = test_table_scan_with_name("t3")?;
997        let t4 = test_table_scan_with_name("t4")?;
998
999        // could not eliminate to inner join
1000        let plan1 = LogicalPlanBuilder::from(t1)
1001            .cross_join(t2)?
1002            .filter(binary_expr(
1003                binary_expr(col("t1.a").eq(col("t2.a")), Or, col("t2.c").lt(lit(15u32))),
1004                Or,
1005                binary_expr(
1006                    col("t1.a").eq(col("t2.a")),
1007                    And,
1008                    col("t2.c").eq(lit(688u32)),
1009                ),
1010            ))?
1011            .build()?;
1012
1013        // could eliminate to inner join
1014        let plan2 = LogicalPlanBuilder::from(t3)
1015            .cross_join(t4)?
1016            .filter(binary_expr(
1017                binary_expr(
1018                    binary_expr(
1019                        col("t3.a").eq(col("t4.a")),
1020                        And,
1021                        col("t4.c").lt(lit(15u32)),
1022                    ),
1023                    Or,
1024                    binary_expr(
1025                        col("t3.a").eq(col("t4.a")),
1026                        And,
1027                        col("t3.c").eq(lit(688u32)),
1028                    ),
1029                ),
1030                Or,
1031                binary_expr(
1032                    col("t3.a").eq(col("t4.a")),
1033                    And,
1034                    col("t3.b").eq(col("t4.b")),
1035                ),
1036            ))?
1037            .build()?;
1038
1039        // could eliminate to inner join
1040        let plan = LogicalPlanBuilder::from(plan1)
1041            .cross_join(plan2)?
1042            .filter(binary_expr(
1043                binary_expr(col("t3.a").eq(col("t1.a")), And, col("t4.c").lt(lit(15u32))),
1044                Or,
1045                binary_expr(
1046                    col("t3.a").eq(col("t1.a")),
1047                    And,
1048                    col("t4.c").eq(lit(688u32)),
1049                ),
1050            ))?
1051            .build()?;
1052
1053        assert_optimized_plan_equal!(
1054            plan,
1055            @ r"
1056        Filter: t4.c < UInt32(15) OR t4.c = UInt32(688) [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
1057          Inner Join: t1.a = t3.a [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
1058            Filter: t1.a = t2.a OR t2.c < UInt32(15) OR t1.a = t2.a AND t2.c = UInt32(688) [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
1059              Cross Join: [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
1060                TableScan: t1 [a:UInt32, b:UInt32, c:UInt32]
1061                TableScan: t2 [a:UInt32, b:UInt32, c:UInt32]
1062            Filter: t4.c < UInt32(15) OR t3.c = UInt32(688) OR t3.b = t4.b [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
1063              Inner Join: t3.a = t4.a [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
1064                TableScan: t3 [a:UInt32, b:UInt32, c:UInt32]
1065                TableScan: t4 [a:UInt32, b:UInt32, c:UInt32]
1066        "
1067        )
1068    }
1069
1070    #[test]
1071    fn eliminate_cross_join_multi_tables_4() -> Result<()> {
1072        let t1 = test_table_scan_with_name("t1")?;
1073        let t2 = test_table_scan_with_name("t2")?;
1074        let t3 = test_table_scan_with_name("t3")?;
1075        let t4 = test_table_scan_with_name("t4")?;
1076
1077        // could eliminate to inner join
1078        // filter: (t1.a = t2.a OR t2.c < 15) AND (t1.a = t2.a AND tc.2 = 688)
1079        let plan1 = LogicalPlanBuilder::from(t1)
1080            .cross_join(t2)?
1081            .filter(binary_expr(
1082                binary_expr(col("t1.a").eq(col("t2.a")), Or, col("t2.c").lt(lit(15u32))),
1083                And,
1084                binary_expr(
1085                    col("t1.a").eq(col("t2.a")),
1086                    And,
1087                    col("t2.c").eq(lit(688u32)),
1088                ),
1089            ))?
1090            .build()?;
1091
1092        // could eliminate to inner join
1093        let plan2 = LogicalPlanBuilder::from(t3).cross_join(t4)?.build()?;
1094
1095        // could eliminate to inner join
1096        // filter:
1097        //   ((t3.a = t1.a AND t4.c < 15) OR (t3.a = t1.a AND t4.c = 688))
1098        //     AND
1099        //   ((t3.a = t4.a AND t4.c < 15) OR (t3.a = t4.a AND t3.c = 688) OR (t3.a = t4.a AND t3.b = t4.b))
1100        let plan = LogicalPlanBuilder::from(plan1)
1101            .cross_join(plan2)?
1102            .filter(binary_expr(
1103                binary_expr(
1104                    binary_expr(
1105                        col("t3.a").eq(col("t1.a")),
1106                        And,
1107                        col("t4.c").lt(lit(15u32)),
1108                    ),
1109                    Or,
1110                    binary_expr(
1111                        col("t3.a").eq(col("t1.a")),
1112                        And,
1113                        col("t4.c").eq(lit(688u32)),
1114                    ),
1115                ),
1116                And,
1117                binary_expr(
1118                    binary_expr(
1119                        binary_expr(
1120                            col("t3.a").eq(col("t4.a")),
1121                            And,
1122                            col("t4.c").lt(lit(15u32)),
1123                        ),
1124                        Or,
1125                        binary_expr(
1126                            col("t3.a").eq(col("t4.a")),
1127                            And,
1128                            col("t3.c").eq(lit(688u32)),
1129                        ),
1130                    ),
1131                    Or,
1132                    binary_expr(
1133                        col("t3.a").eq(col("t4.a")),
1134                        And,
1135                        col("t3.b").eq(col("t4.b")),
1136                    ),
1137                ),
1138            ))?
1139            .build()?;
1140
1141        assert_optimized_plan_equal!(
1142            plan,
1143            @ r"
1144        Filter: (t4.c < UInt32(15) OR t4.c = UInt32(688)) AND (t4.c < UInt32(15) OR t3.c = UInt32(688) OR t3.b = t4.b) [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
1145          Inner Join: t3.a = t4.a [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
1146            Inner Join: t1.a = t3.a [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
1147              Filter: t2.c = UInt32(688) [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
1148                Inner Join: t1.a = t2.a [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
1149                  TableScan: t1 [a:UInt32, b:UInt32, c:UInt32]
1150                  TableScan: t2 [a:UInt32, b:UInt32, c:UInt32]
1151              TableScan: t3 [a:UInt32, b:UInt32, c:UInt32]
1152            TableScan: t4 [a:UInt32, b:UInt32, c:UInt32]
1153        "
1154        )
1155    }
1156
1157    #[test]
1158    fn eliminate_cross_join_multi_tables_5() -> Result<()> {
1159        let t1 = test_table_scan_with_name("t1")?;
1160        let t2 = test_table_scan_with_name("t2")?;
1161        let t3 = test_table_scan_with_name("t3")?;
1162        let t4 = test_table_scan_with_name("t4")?;
1163
1164        // could eliminate to inner join
1165        let plan1 = LogicalPlanBuilder::from(t1).cross_join(t2)?.build()?;
1166
1167        // could eliminate to inner join
1168        let plan2 = LogicalPlanBuilder::from(t3).cross_join(t4)?.build()?;
1169
1170        // could eliminate to inner join
1171        // Filter:
1172        //  ((t3.a = t1.a AND t4.c < 15) OR (t3.a = t1.a AND t4.c = 688))
1173        //      AND
1174        //  ((t3.a = t4.a AND t4.c < 15) OR (t3.a = t4.a AND t3.c = 688) OR (t3.a = t4.a AND t3.b = t4.b))
1175        //      AND
1176        //  ((t1.a = t2.a OR t2.c < 15) AND (t1.a = t2.a AND t2.c = 688))
1177        let plan = LogicalPlanBuilder::from(plan1)
1178            .cross_join(plan2)?
1179            .filter(binary_expr(
1180                binary_expr(
1181                    binary_expr(
1182                        binary_expr(
1183                            col("t3.a").eq(col("t1.a")),
1184                            And,
1185                            col("t4.c").lt(lit(15u32)),
1186                        ),
1187                        Or,
1188                        binary_expr(
1189                            col("t3.a").eq(col("t1.a")),
1190                            And,
1191                            col("t4.c").eq(lit(688u32)),
1192                        ),
1193                    ),
1194                    And,
1195                    binary_expr(
1196                        binary_expr(
1197                            binary_expr(
1198                                col("t3.a").eq(col("t4.a")),
1199                                And,
1200                                col("t4.c").lt(lit(15u32)),
1201                            ),
1202                            Or,
1203                            binary_expr(
1204                                col("t3.a").eq(col("t4.a")),
1205                                And,
1206                                col("t3.c").eq(lit(688u32)),
1207                            ),
1208                        ),
1209                        Or,
1210                        binary_expr(
1211                            col("t3.a").eq(col("t4.a")),
1212                            And,
1213                            col("t3.b").eq(col("t4.b")),
1214                        ),
1215                    ),
1216                ),
1217                And,
1218                binary_expr(
1219                    binary_expr(
1220                        col("t1.a").eq(col("t2.a")),
1221                        Or,
1222                        col("t2.c").lt(lit(15u32)),
1223                    ),
1224                    And,
1225                    binary_expr(
1226                        col("t1.a").eq(col("t2.a")),
1227                        And,
1228                        col("t2.c").eq(lit(688u32)),
1229                    ),
1230                ),
1231            ))?
1232            .build()?;
1233
1234        assert_optimized_plan_equal!(
1235            plan,
1236            @ r"
1237        Filter: (t4.c < UInt32(15) OR t4.c = UInt32(688)) AND (t4.c < UInt32(15) OR t3.c = UInt32(688) OR t3.b = t4.b) AND t2.c = UInt32(688) [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
1238          Inner Join: t3.a = t4.a [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
1239            Inner Join: t1.a = t3.a [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
1240              Inner Join: t1.a = t2.a [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
1241                TableScan: t1 [a:UInt32, b:UInt32, c:UInt32]
1242                TableScan: t2 [a:UInt32, b:UInt32, c:UInt32]
1243              TableScan: t3 [a:UInt32, b:UInt32, c:UInt32]
1244            TableScan: t4 [a:UInt32, b:UInt32, c:UInt32]
1245        "
1246        )
1247    }
1248
1249    #[test]
1250    fn eliminate_cross_join_with_expr_and() -> Result<()> {
1251        let t1 = test_table_scan_with_name("t1")?;
1252        let t2 = test_table_scan_with_name("t2")?;
1253
1254        // could eliminate to inner join since filter has Join predicates
1255        let plan = LogicalPlanBuilder::from(t1)
1256            .cross_join(t2)?
1257            .filter(binary_expr(
1258                (col("t1.a") + lit(100u32)).eq(col("t2.a") * lit(2u32)),
1259                And,
1260                col("t2.c").lt(lit(20u32)),
1261            ))?
1262            .build()?;
1263
1264        assert_optimized_plan_equal!(
1265            plan,
1266            @ r"
1267        Filter: t2.c < UInt32(20) [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
1268          Inner Join: t1.a + UInt32(100) = t2.a * UInt32(2) [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
1269            TableScan: t1 [a:UInt32, b:UInt32, c:UInt32]
1270            TableScan: t2 [a:UInt32, b:UInt32, c:UInt32]
1271        "
1272        )
1273    }
1274
1275    #[test]
1276    fn eliminate_cross_with_expr_or() -> Result<()> {
1277        let t1 = test_table_scan_with_name("t1")?;
1278        let t2 = test_table_scan_with_name("t2")?;
1279
1280        // could not eliminate to inner join since filter OR expression and there is no common
1281        // Join predicates in left and right of OR expr.
1282        let plan = LogicalPlanBuilder::from(t1)
1283            .cross_join(t2)?
1284            .filter(binary_expr(
1285                (col("t1.a") + lit(100u32)).eq(col("t2.a") * lit(2u32)),
1286                Or,
1287                col("t2.b").eq(col("t1.a")),
1288            ))?
1289            .build()?;
1290
1291        assert_optimized_plan_equal!(
1292            plan,
1293            @ r"
1294        Filter: t1.a + UInt32(100) = t2.a * UInt32(2) OR t2.b = t1.a [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
1295          Cross Join: [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
1296            TableScan: t1 [a:UInt32, b:UInt32, c:UInt32]
1297            TableScan: t2 [a:UInt32, b:UInt32, c:UInt32]
1298        "
1299        )
1300    }
1301
1302    #[test]
1303    fn eliminate_cross_with_common_expr_and() -> Result<()> {
1304        let t1 = test_table_scan_with_name("t1")?;
1305        let t2 = test_table_scan_with_name("t2")?;
1306
1307        // could eliminate to inner join
1308        let common_join_key = (col("t1.a") + lit(100u32)).eq(col("t2.a") * lit(2u32));
1309        let plan = LogicalPlanBuilder::from(t1)
1310            .cross_join(t2)?
1311            .filter(binary_expr(
1312                binary_expr(common_join_key.clone(), And, col("t2.c").lt(lit(20u32))),
1313                And,
1314                binary_expr(common_join_key, And, col("t2.c").eq(lit(10u32))),
1315            ))?
1316            .build()?;
1317
1318        assert_optimized_plan_equal!(
1319            plan,
1320            @ r"
1321        Filter: t2.c < UInt32(20) AND t2.c = UInt32(10) [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
1322          Inner Join: t1.a + UInt32(100) = t2.a * UInt32(2) [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
1323            TableScan: t1 [a:UInt32, b:UInt32, c:UInt32]
1324            TableScan: t2 [a:UInt32, b:UInt32, c:UInt32]
1325        "
1326        )
1327    }
1328
1329    #[test]
1330    fn eliminate_cross_with_common_expr_or() -> Result<()> {
1331        let t1 = test_table_scan_with_name("t1")?;
1332        let t2 = test_table_scan_with_name("t2")?;
1333
1334        // could eliminate to inner join since Or predicates have common Join predicates
1335        let common_join_key = (col("t1.a") + lit(100u32)).eq(col("t2.a") * lit(2u32));
1336        let plan = LogicalPlanBuilder::from(t1)
1337            .cross_join(t2)?
1338            .filter(binary_expr(
1339                binary_expr(common_join_key.clone(), And, col("t2.c").lt(lit(15u32))),
1340                Or,
1341                binary_expr(common_join_key, And, col("t2.c").eq(lit(688u32))),
1342            ))?
1343            .build()?;
1344
1345        assert_optimized_plan_equal!(
1346            plan,
1347            @ r"
1348        Filter: t2.c < UInt32(15) OR t2.c = UInt32(688) [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
1349          Inner Join: t1.a + UInt32(100) = t2.a * UInt32(2) [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
1350            TableScan: t1 [a:UInt32, b:UInt32, c:UInt32]
1351            TableScan: t2 [a:UInt32, b:UInt32, c:UInt32]
1352        "
1353        )
1354    }
1355
1356    #[test]
1357    fn reorder_join_with_expr_key_multi_tables() -> Result<()> {
1358        let t1 = test_table_scan_with_name("t1")?;
1359        let t2 = test_table_scan_with_name("t2")?;
1360        let t3 = test_table_scan_with_name("t3")?;
1361
1362        // could eliminate to inner join
1363        let plan = LogicalPlanBuilder::from(t1)
1364            .cross_join(t2)?
1365            .cross_join(t3)?
1366            .filter(binary_expr(
1367                binary_expr(
1368                    (col("t3.a") + lit(100u32)).eq(col("t1.a") * lit(2u32)),
1369                    And,
1370                    col("t3.c").lt(lit(15u32)),
1371                ),
1372                And,
1373                binary_expr(
1374                    (col("t3.a") + lit(100u32)).eq(col("t2.a") * lit(2u32)),
1375                    And,
1376                    col("t3.b").lt(lit(15u32)),
1377                ),
1378            ))?
1379            .build()?;
1380
1381        assert_optimized_plan_equal!(
1382            plan,
1383            @ r"
1384        Filter: t3.c < UInt32(15) AND t3.b < UInt32(15) [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
1385          Projection: t1.a, t1.b, t1.c, t2.a, t2.b, t2.c, t3.a, t3.b, t3.c [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
1386            Inner Join: t3.a + UInt32(100) = t2.a * UInt32(2) [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
1387              Inner Join: t1.a * UInt32(2) = t3.a + UInt32(100) [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]
1388                TableScan: t1 [a:UInt32, b:UInt32, c:UInt32]
1389                TableScan: t3 [a:UInt32, b:UInt32, c:UInt32]
1390              TableScan: t2 [a:UInt32, b:UInt32, c:UInt32]
1391        "
1392        )
1393    }
1394
1395    #[test]
1396    fn preserve_null_equality_setting() -> Result<()> {
1397        let t1 = test_table_scan_with_name("t1")?;
1398        let t2 = test_table_scan_with_name("t2")?;
1399
1400        // Create an inner join with NullEquality::NullEqualsNull
1401        let join_schema = Arc::new(build_join_schema(
1402            t1.schema(),
1403            t2.schema(),
1404            &JoinType::Inner,
1405        )?);
1406
1407        let inner_join = LogicalPlan::Join(Join {
1408            left: Arc::new(t1),
1409            right: Arc::new(t2),
1410            join_type: JoinType::Inner,
1411            join_constraint: JoinConstraint::On,
1412            on: vec![],
1413            filter: None,
1414            schema: join_schema,
1415            null_equality: NullEquality::NullEqualsNull, // Test preservation
1416            null_aware: false,
1417        });
1418
1419        // Apply filter that can create join conditions
1420        let plan = LogicalPlanBuilder::from(inner_join)
1421            .filter(binary_expr(
1422                col("t1.a").eq(col("t2.a")),
1423                And,
1424                col("t2.c").lt(lit(20u32)),
1425            ))?
1426            .build()?;
1427
1428        let rule = EliminateCrossJoin::new();
1429        let optimized_plan = rule.rewrite(plan, &OptimizerContext::new())?.data;
1430
1431        // Verify that null_equality is preserved in the optimized plan
1432        fn check_null_equality_preserved(plan: &LogicalPlan) -> bool {
1433            match plan {
1434                LogicalPlan::Join(join) => {
1435                    // All joins in the optimized plan should preserve null equality
1436                    if join.null_equality == NullEquality::NullEqualsNothing {
1437                        return false;
1438                    }
1439                    // Recursively check child plans
1440                    plan.inputs()
1441                        .iter()
1442                        .all(|input| check_null_equality_preserved(input))
1443                }
1444                _ => {
1445                    // Recursively check child plans for non-join nodes
1446                    plan.inputs()
1447                        .iter()
1448                        .all(|input| check_null_equality_preserved(input))
1449                }
1450            }
1451        }
1452
1453        assert!(
1454            check_null_equality_preserved(&optimized_plan),
1455            "null_equality setting should be preserved after optimization"
1456        );
1457
1458        Ok(())
1459    }
1460
1461    // ---------------- fast-path tests ----------------
1462
1463    /// `plan_has_joins` detects a `Join` at the root of the plan.
1464    #[test]
1465    fn plan_has_joins_detects_root_join() -> Result<()> {
1466        let plan = LogicalPlanBuilder::from(test_table_scan_with_name("t1")?)
1467            .cross_join(test_table_scan_with_name("t2")?)?
1468            .build()?;
1469        assert!(plan_has_joins(&plan));
1470        Ok(())
1471    }
1472
1473    /// `plan_has_joins` detects a `Join` nested under other operators.
1474    #[test]
1475    fn plan_has_joins_detects_nested_join() -> Result<()> {
1476        let plan = LogicalPlanBuilder::from(test_table_scan_with_name("t1")?)
1477            .cross_join(test_table_scan_with_name("t2")?)?
1478            .filter(col("t1.a").eq(col("t2.a")))?
1479            .project(vec![col("t1.a")])?
1480            .build()?;
1481        assert!(plan_has_joins(&plan));
1482        Ok(())
1483    }
1484
1485    /// Join-free plans return `false` so the fast-path in `rewrite` can
1486    /// bail out before doing any recursion.
1487    #[test]
1488    fn plan_has_joins_returns_false_for_join_free_plan() -> Result<()> {
1489        let plan = LogicalPlanBuilder::from(test_table_scan_with_name("t1")?)
1490            .filter(col("a").gt(lit(0_i32)))?
1491            .project(vec![col("a"), col("b")])?
1492            .build()?;
1493        assert!(!plan_has_joins(&plan));
1494        Ok(())
1495    }
1496
1497    /// `plan_has_joins` walks into embedded subquery plans — e.g. an
1498    /// outer `Filter` whose predicate is `IN (SELECT ... FROM a, b)`
1499    /// where the inner plan contains a `CROSS JOIN`. Without this the
1500    /// fast-path would silently skip optimizing joins-in-subqueries
1501    /// because `LogicalPlan::apply` doesn't descend into subquery
1502    /// plan trees.
1503    #[test]
1504    fn plan_has_joins_detects_join_inside_subquery() -> Result<()> {
1505        use datafusion_expr::in_subquery;
1506
1507        // Subquery plan that itself contains a join.
1508        let subquery_plan =
1509            LogicalPlanBuilder::from(test_table_scan_with_name("sub_t1")?)
1510                .cross_join(test_table_scan_with_name("sub_t2")?)?
1511                .project(vec![col("sub_t1.a")])?
1512                .build()?;
1513
1514        // Outer plan with NO direct Join — only the IN subquery reaches one.
1515        let outer = LogicalPlanBuilder::from(test_table_scan_with_name("t1")?)
1516            .filter(in_subquery(col("a"), Arc::new(subquery_plan)))?
1517            .project(vec![col("a")])?
1518            .build()?;
1519
1520        assert!(
1521            plan_has_joins(&outer),
1522            "plan_has_joins must descend into subquery plans"
1523        );
1524        Ok(())
1525    }
1526
1527    /// `EliminateCrossJoin::rewrite` short-circuits on join-free plans:
1528    /// no recursion into `rewrite_children`, no `Transformed::yes`,
1529    /// the plan comes back identical.
1530    #[test]
1531    fn rewrite_short_circuits_when_plan_has_no_joins() -> Result<()> {
1532        let plan = LogicalPlanBuilder::from(test_table_scan_with_name("t1")?)
1533            .filter(col("a").gt(lit(0_i32)))?
1534            .project(vec![col("a"), col("b")])?
1535            .build()?;
1536
1537        let starting_display = plan.display_indent_schema().to_string();
1538        let starting_schema = Arc::clone(plan.schema());
1539
1540        let rule = EliminateCrossJoin::new();
1541        let Transformed {
1542            transformed,
1543            data: optimized_plan,
1544            ..
1545        } = rule.rewrite(plan, &OptimizerContext::new())?;
1546
1547        assert!(
1548            !transformed,
1549            "join-free plan should not be marked as transformed"
1550        );
1551        assert_eq!(&starting_schema, optimized_plan.schema());
1552        assert_eq!(
1553            starting_display,
1554            optimized_plan.display_indent_schema().to_string()
1555        );
1556        Ok(())
1557    }
1558}
datafusion_optimizer/eliminate_cross_join.rs

datafusion_optimizer/
eliminate_cross_join.rs
