From 2a393990d5cfe01ece568f69127ef58c1554011d Mon Sep 17 00:00:00 2001
From: Omer Banay <omer@epsio.io>
Date: Tue, 12 Sep 2023 15:03:04 +0300
Subject: [PATCH] Skip EliminateCrossJoin rule if inner join with filter is
 found - check recursively

---
 .../optimizer/src/eliminate_cross_join.rs     | 29 ++++++++++++++-----
 1 file changed, 21 insertions(+), 8 deletions(-)

diff --git a/datafusion/optimizer/src/eliminate_cross_join.rs b/datafusion/optimizer/src/eliminate_cross_join.rs
index ec4d8a2cbf1d1..c1623483fed62 100644
--- a/datafusion/optimizer/src/eliminate_cross_join.rs
+++ b/datafusion/optimizer/src/eliminate_cross_join.rs
@@ -68,18 +68,22 @@ impl OptimizerRule for EliminateCrossJoin {
                             return Ok(None);
                         }
 
-                        flatten_join_inputs(
+                        if !flatten_join_inputs(
                             &input,
                             &mut possible_join_keys,
                             &mut all_inputs,
-                        )?;
+                        )? {
+                            return Ok(None);
+                        }
                     }
                     LogicalPlan::CrossJoin(_) => {
-                        flatten_join_inputs(
+                        if !flatten_join_inputs(
                             &input,
                             &mut possible_join_keys,
                             &mut all_inputs,
-                        )?;
+                        )? {
+                            return Ok(None);
+                        }
                     }
                     _ => {
                         return utils::optimize_children(self, plan, config);
@@ -137,11 +141,13 @@ impl OptimizerRule for EliminateCrossJoin {
     }
 }
 
+/// Find inner join from all_inputs and possible_join_keys
+/// Returns a boolean indicating whether the plan can be flattened.
 fn flatten_join_inputs(
     plan: &LogicalPlan,
     possible_join_keys: &mut Vec<(Expr, Expr)>,
     all_inputs: &mut Vec<LogicalPlan>,
-) -> Result<()> {
+) -> Result<bool> {
     let children = match plan {
         LogicalPlan::Join(join) => {
             possible_join_keys.extend(join.on.clone());
@@ -163,18 +169,25 @@ fn flatten_join_inputs(
         match *child {
             LogicalPlan::Join(left_join) => {
                 if left_join.join_type == JoinType::Inner {
-                    flatten_join_inputs(child, possible_join_keys, all_inputs)?;
+                    if left_join.filter.is_some() {
+                        return Ok(false);
+                    }
+                    if !flatten_join_inputs(child, possible_join_keys, all_inputs)? {
+                        return Ok(false);
+                    }
                 } else {
                     all_inputs.push((*child).clone());
                 }
             }
             LogicalPlan::CrossJoin(_) => {
-                flatten_join_inputs(child, possible_join_keys, all_inputs)?;
+                if !flatten_join_inputs(child, possible_join_keys, all_inputs)? {
+                    return Ok(false);
+                }
             }
             _ => all_inputs.push((*child).clone()),
         }
     }
-    Ok(())
+    Ok(true)
 }
 
 fn find_inner_join(