reimplement eliminate_outer_join

Author: jackwener

datafusion/expr/src/logical_plan/plan.rs

@@ -999,6 +999,12 @@ pub enum JoinType {
     RightAnti,
 }
 
+impl JoinType {
+    pub fn is_outer(self) -> bool {
+        self == JoinType::Left || self == JoinType::Right || self == JoinType::Full
+    }
+}
+
 impl Display for JoinType {
     fn fmt(&self, f: &mut Formatter) -> fmt::Result {
         let join_type = match self {

datafusion/optimizer/src/reduce_outer_join.rs renamed to datafusion/optimizer/src/eliminate_outer_join.rs

@@ -15,18 +15,17 @@
 // specific language governing permissions and limitations
 // under the License.
 
-//! Optimizer rule to reduce left/right/full join to inner join if possible.
-use crate::{OptimizerConfig, OptimizerRule};
+//! Optimizer rule to eliminate left/right/full join to inner join if possible.
+use crate::{utils, OptimizerConfig, OptimizerRule};
 use datafusion_common::{Column, DFSchema, Result};
 use datafusion_expr::{
     expr::BinaryExpr,
-    logical_plan::{Filter, Join, JoinType, LogicalPlan, Projection},
+    logical_plan::{Join, JoinType, LogicalPlan},
     utils::from_plan,
 };
 use datafusion_expr::{Expr, Operator};
 
 use datafusion_expr::expr::Cast;
-use std::collections::HashMap;
 use std::sync::Arc;
 
 #[derive(Default)]
@@ -39,191 +38,107 @@ impl ReduceOuterJoin {
     }
 }
 
+/// Attempt to eliminate outer joins to inner joins.
+/// for query: select ... from a left join b on ... where b.xx = 100;
+/// if b.xx is null, and b.xx = 100 returns false, filtered those null rows.
+/// Therefore, there is no need to produce null rows for output, we can use
+/// inner join instead of left join.
+///
+/// Generally, an outer join can be eliminated to inner join if equals from where
+/// return false while any inputs are null and columns of those equals are come from
+/// nullable side of outer join.
 impl OptimizerRule for ReduceOuterJoin {
     fn optimize(
         &self,
         plan: &LogicalPlan,
         optimizer_config: &mut OptimizerConfig,
     ) -> Result<LogicalPlan> {
-        let mut nonnullable_cols: Vec<Column> = vec![];
+        match plan {
+            LogicalPlan::Filter(filter) => match filter.input().as_ref() {
+                LogicalPlan::Join(join) => {
+                    let mut nonnullable_cols: Vec<Column> = vec![];
+
+                    extract_nonnullable_columns(
+                        filter.predicate(),
+                        &mut nonnullable_cols,
+                        join.left.schema(),
+                        join.right.schema(),
+                        true,
+                    )?;
 
-        reduce_outer_join(self, plan, &mut nonnullable_cols, optimizer_config)
+                    let new_join_type = if join.join_type.is_outer() {
+                        let mut left_nonnullable = false;
+                        let mut right_nonnullable = false;
+                        for col in nonnullable_cols.iter() {
+                            if join.left.schema().field_from_column(col).is_ok() {
+                                left_nonnullable = true;
+                            }
+                            if join.right.schema().field_from_column(col).is_ok() {
+                                right_nonnullable = true;
+                            }
+                        }
+                        eliminate_outer(
+                            join.join_type,
+                            left_nonnullable,
+                            right_nonnullable,
+                        )
+                    } else {
+                        join.join_type
+                    };
+
+                    let new_join = LogicalPlan::Join(Join {
+                        left: Arc::new((*join.left).clone()),
+                        right: Arc::new((*join.right).clone()),
+                        join_type: new_join_type,
+                        join_constraint: join.join_constraint,
+                        on: join.on.clone(),
+                        filter: join.filter.clone(),
+                        schema: join.schema.clone(),
+                        null_equals_null: join.null_equals_null,
+                    });
+                    let new_plan = from_plan(plan, &plan.expressions(), &[new_join])?;
+                    utils::optimize_children(self, &new_plan, optimizer_config)
+                }
+                _ => utils::optimize_children(self, plan, optimizer_config),
+            },
+            _ => utils::optimize_children(self, plan, optimizer_config),
+        }
     }
 
     fn name(&self) -> &str {
-        "reduce_outer_join"
+        "eliminate_outer_join"
     }
 }
 
-/// Attempt to reduce outer joins to inner joins.
-/// for query: select ... from a left join b on ... where b.xx = 100;
-/// if b.xx is null, and b.xx = 100 returns false, filterd those null rows.
-/// Therefore, there is no need to produce null rows for output, we can use
-/// inner join instead of left join.
-///
-/// Generally, an outer join can be reduced to inner join if quals from where
-/// return false while any inputs are null and columns of those quals are come from
-/// nullable side of outer join.
-fn reduce_outer_join(
-    _optimizer: &ReduceOuterJoin,
-    plan: &LogicalPlan,
-    nonnullable_cols: &mut Vec<Column>,
-    _optimizer_config: &OptimizerConfig,
-) -> Result<LogicalPlan> {
-    match plan {
-        LogicalPlan::Filter(filter) => match filter.input().as_ref() {
-            LogicalPlan::Join(join) => {
-                extract_nonnullable_columns(
-                    filter.predicate(),
-                    nonnullable_cols,
-                    join.left.schema(),
-                    join.right.schema(),
-                    true,
-                )?;
-                Ok(LogicalPlan::Filter(Filter::try_new(
-                    filter.predicate().clone(),
-                    Arc::new(reduce_outer_join(
-                        _optimizer,
-                        filter.input(),
-                        nonnullable_cols,
-                        _optimizer_config,
-                    )?),
-                )?))
+pub fn eliminate_outer(
+    join_type: JoinType,
+    left_nonnullable: bool,
+    right_nonnullable: bool,
+) -> JoinType {
+    let mut new_join_type = join_type;
+    match join_type {
+        JoinType::Left => {
+            if right_nonnullable {
+                new_join_type = JoinType::Inner;
             }
-            _ => Ok(LogicalPlan::Filter(Filter::try_new(
-                filter.predicate().clone(),
-                Arc::new(reduce_outer_join(
-                    _optimizer,
-                    filter.input(),
-                    nonnullable_cols,
-                    _optimizer_config,
-                )?),
-            )?)),
-        },
-        LogicalPlan::Join(join) => {
-            let mut new_join_type = join.join_type;
-
-            if join.join_type == JoinType::Left
-                || join.join_type == JoinType::Right
-                || join.join_type == JoinType::Full
-            {
-                let mut left_nonnullable = false;
-                let mut right_nonnullable = false;
-                for col in nonnullable_cols.iter() {
-                    if join.left.schema().field_from_column(col).is_ok() {
-                        left_nonnullable = true;
-                    }
-                    if join.right.schema().field_from_column(col).is_ok() {
-                        right_nonnullable = true;
-                    }
-                }
-
-                match join.join_type {
-                    JoinType::Left => {
-                        if right_nonnullable {
-                            new_join_type = JoinType::Inner;
-                        }
-                    }
-                    JoinType::Right => {
-                        if left_nonnullable {
-                            new_join_type = JoinType::Inner;
-                        }
-                    }
-                    JoinType::Full => {
-                        if left_nonnullable && right_nonnullable {
-                            new_join_type = JoinType::Inner;
-                        } else if left_nonnullable {
-                            new_join_type = JoinType::Left;
-                        } else if right_nonnullable {
-                            new_join_type = JoinType::Right;
-                        }
-                    }
-                    _ => {}
-                };
-            }
-
-            let left_plan = reduce_outer_join(
-                _optimizer,
-                &join.left,
-                &mut nonnullable_cols.clone(),
-                _optimizer_config,
-            )?;
-            let right_plan = reduce_outer_join(
-                _optimizer,
-                &join.right,
-                &mut nonnullable_cols.clone(),
-                _optimizer_config,
-            )?;
-
-            Ok(LogicalPlan::Join(Join {
-                left: Arc::new(left_plan),
-                right: Arc::new(right_plan),
-                join_type: new_join_type,
-                join_constraint: join.join_constraint,
-                on: join.on.clone(),
-                filter: join.filter.clone(),
-                schema: join.schema.clone(),
-                null_equals_null: join.null_equals_null,
-            }))
         }
-        LogicalPlan::Projection(Projection {
-            input,
-            expr,
-            schema,
-            alias: _,
-        }) => {
-            let projection = schema
-                .fields()
-                .iter()
-                .enumerate()
-                .map(|(i, field)| {
-                    // strip alias, as they should not be part of filters
-                    let expr = match &expr[i] {
-                        Expr::Alias(expr, _) => expr.as_ref().clone(),
-                        expr => expr.clone(),
-                    };
-
-                    (field.qualified_name(), expr)
-                })
-                .collect::<HashMap<_, _>>();
-
-            // re-write all Columns based on this projection
-            for col in nonnullable_cols.iter_mut() {
-                if let Some(Expr::Column(column)) = projection.get(&col.flat_name()) {
-                    *col = column.clone();
-                }
+        JoinType::Right => {
+            if left_nonnullable {
+                new_join_type = JoinType::Inner;
             }
-
-            // optimize inner
-            let new_input = reduce_outer_join(
-                _optimizer,
-                input,
-                nonnullable_cols,
-                _optimizer_config,
-            )?;
-
-            from_plan(plan, expr, &[new_input])
         }
-        _ => {
-            let expr = plan.expressions();
-
-            // apply the optimization to all inputs of the plan
-            let inputs = plan.inputs();
-            let new_inputs = inputs
-                .iter()
-                .map(|plan| {
-                    reduce_outer_join(
-                        _optimizer,
-                        plan,
-                        nonnullable_cols,
-                        _optimizer_config,
-                    )
-                })
-                .collect::<Result<Vec<_>>>()?;
-
-            from_plan(plan, &expr, &new_inputs)
+        JoinType::Full => {
+            if left_nonnullable && right_nonnullable {
+                new_join_type = JoinType::Inner;
+            } else if left_nonnullable {
+                new_join_type = JoinType::Left;
+            } else if right_nonnullable {
+                new_join_type = JoinType::Right;
+            }
         }
+        _ => {}
     }
+    new_join_type
 }
 
 /// Recursively traverses expr, if expr returns false when
@@ -309,9 +224,9 @@ fn extract_nonnullable_columns(
                 )?;
 
                 // for query: select *** from a left join b where b.c1 ... or b.c2 ...
-                // this can be reduced to inner join.
+                // this can be eliminated to inner join.
                 // for query: select *** from a left join b where a.c1 ... or b.c2 ...
-                // this can not be reduced.
+                // this can not be eliminated.
                 // If columns of relation exist in both sub exprs, any columns of this relation
                 // can be added to non nullable columns.
                 if !left_nonnullable_cols.is_empty() && !right_nonnullable_cols.is_empty()
@@ -387,11 +302,11 @@ mod tests {
     }
 
     #[test]
-    fn reduce_left_with_null() -> Result<()> {
+    fn eliminate_left_with_null() -> Result<()> {
         let t1 = test_table_scan_with_name("t1")?;
         let t2 = test_table_scan_with_name("t2")?;
 
-        // could not reduce to inner join
+        // could not eliminate to inner join
         let plan = LogicalPlanBuilder::from(t1)
             .join(
                 &t2,
@@ -412,11 +327,11 @@ mod tests {
     }
 
     #[test]
-    fn reduce_left_with_not_null() -> Result<()> {
+    fn eliminate_left_with_not_null() -> Result<()> {
         let t1 = test_table_scan_with_name("t1")?;
         let t2 = test_table_scan_with_name("t2")?;
 
-        // reduce to inner join
+        // eliminate to inner join
         let plan = LogicalPlanBuilder::from(t1)
             .join(
                 &t2,
@@ -437,11 +352,11 @@ mod tests {
     }
 
     #[test]
-    fn reduce_right_with_or() -> Result<()> {
+    fn eliminate_right_with_or() -> Result<()> {
         let t1 = test_table_scan_with_name("t1")?;
         let t2 = test_table_scan_with_name("t2")?;
 
-        // reduce to inner join
+        // eliminate to inner join
         let plan = LogicalPlanBuilder::from(t1)
             .join(
                 &t2,
@@ -466,11 +381,11 @@ mod tests {
     }
 
     #[test]
-    fn reduce_full_with_and() -> Result<()> {
+    fn eliminate_full_with_and() -> Result<()> {
         let t1 = test_table_scan_with_name("t1")?;
         let t2 = test_table_scan_with_name("t2")?;
 
-        // reduce to inner join
+        // eliminate to inner join
         let plan = LogicalPlanBuilder::from(t1)
             .join(
                 &t2,
@@ -495,11 +410,11 @@ mod tests {
     }
 
     #[test]
-    fn reduce_full_with_type_cast() -> Result<()> {
+    fn eliminate_full_with_type_cast() -> Result<()> {
         let t1 = test_table_scan_with_name("t1")?;
         let t2 = test_table_scan_with_name("t2")?;
 
-        // reduce to inner join
+        // eliminate to inner join
         let plan = LogicalPlanBuilder::from(t1)
             .join(
                 &t2,