[GitHub] [arrow-datafusion] alamb commented on a diff in pull request #4272: reimplement `eliminate_outer_join`

[GitBox <gi...@apache.org>]

alamb commented on code in PR #4272:
URL: https://github.com/apache/arrow-datafusion/pull/4272#discussion_r1030716347


##########
datafusion/optimizer/src/eliminate_outer_join.rs:
##########
@@ -15,235 +15,141 @@
 // 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)]
-pub struct ReduceOuterJoin;
+pub struct EliminateOuterJoin;
 
-impl ReduceOuterJoin {
+impl EliminateOuterJoin {
     #[allow(missing_docs)]
     pub fn new() -> Self {
         Self {}
     }
 }
 
-impl OptimizerRule for ReduceOuterJoin {
+/// Attempt to eliminate outer joins.
+impl OptimizerRule for EliminateOuterJoin {
     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 non_nullable_cols: Vec<Column> = vec![];
+
+                    extract_non_nullable_columns(
+                        filter.predicate(),
+                        &mut non_nullable_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_non_nullable = false;
+                        let mut right_non_nullable = false;
+                        for col in non_nullable_cols.iter() {
+                            if join.left.schema().field_from_column(col).is_ok() {
+                                left_non_nullable = true;
+                            }
+                            if join.right.schema().field_from_column(col).is_ok() {
+                                right_non_nullable = true;
+                            }
+                        }
+                        eliminate_outer(
+                            join.join_type,
+                            left_non_nullable,
+                            right_non_nullable,
+                        )
+                    } 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(

Review Comment:
   Thank you for eliminating the duplication



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org