fix: proper null propagation

datafusion/expr/src/expr.rs

@@ -1981,6 +1981,39 @@ impl Expr {
         .map(|data| (data, has_placeholder))
     }
 
+    /// Only applicable to composite expr, return whether the null values of the
+    /// children expr should be propagated to the current expr
+    /// Example: is_null,count(*) function should not propagate null values
+    /// while sum,min,max aggregate functions, and other binary operator can
+    pub fn propagate_null_values(&self) -> Result<bool> {
+        match self {
+            Expr::IsNotNull(_) | Expr::IsNull(_) | Expr::Case(_) => {
+                return Ok(false);
+            }
+            Expr::AggregateFunction(fun) => return Ok(fun.func.is_nullable()),
+            Expr::BinaryExpr(BinaryExpr {
+                op:
+                    Operator::And
+                    | Operator::Or
+                    | Operator::IsDistinctFrom
+                    | Operator::IsNotDistinctFrom,
+                ..
+            }) => {
+                return Ok(false);
+            }
+            _ => {}
+        };
+        let mut propagate_null_values = true;
+        self.apply_children(|e| {
+            if !e.propagate_null_values()? {
+                propagate_null_values = false;
+                return Ok(TreeNodeRecursion::Stop);
+            }
+            Ok(TreeNodeRecursion::Continue)
+        })?;
+        Ok(propagate_null_values)
+    }
+
     /// Returns true if some of this `exprs` subexpressions may not be evaluated
     /// and thus any side effects (like divide by zero) may not be encountered
     pub fn short_circuits(&self) -> bool {

datafusion/expr/src/logical_plan/builder.rs

@@ -55,9 +55,10 @@ use arrow::datatypes::{DataType, Field, Fields, Schema, SchemaRef};
 use datafusion_common::display::ToStringifiedPlan;
 use datafusion_common::file_options::file_type::FileType;
 use datafusion_common::{
-    exec_err, get_target_functional_dependencies, not_impl_err, plan_datafusion_err,
-    plan_err, Column, Constraints, DFSchema, DFSchemaRef, DataFusionError, NullEquality,
-    Result, ScalarValue, TableReference, ToDFSchema, UnnestOptions,
+    exec_err, get_target_functional_dependencies, internal_err, not_impl_err,
+    plan_datafusion_err, plan_err, Column, Constraints, DFSchema, DFSchemaRef,
+    DataFusionError, NullEquality, Result, ScalarValue, TableReference, ToDFSchema,
+    UnnestOptions,
 };
 use datafusion_expr_common::type_coercion::binary::type_union_resolution;
 
@@ -918,12 +919,25 @@ impl LogicalPlanBuilder {
             .collect();
         let metadata = schema.metadata().clone();
         let dfschema = DFSchema::new_with_metadata(qualified_fields, metadata)?;
+        let any_join = match subquery_expr {
+            Some(ref expr) => match expr {
+                Expr::Exists(_) | Expr::InSubquery(_) => true,
+                _ => false,
+            },
+            None => match lateral_join_condition {
+                None => {
+                    return internal_err!("at least lateral join or subquery expr must be set to build dependent join");
+                }
+                Some(_) => false,
+            },
+        };
 
         Ok(Self::new(LogicalPlan::DependentJoin(DependentJoin {
             schema: DFSchemaRef::new(dfschema),
             left: Arc::new(left),
             right: Arc::new(right),
             correlated_columns,
+            any_join,
             subquery_expr,
             subquery_name,
             subquery_depth,

datafusion/expr/src/logical_plan/plan.rs

@@ -425,6 +425,7 @@ pub struct DependentJoin {
     pub subquery_name: String,
 
     pub lateral_join_condition: Option<(JoinType, Expr)>,
+    pub any_join: bool,
 }
 
 impl Display for DependentJoin {
@@ -758,7 +759,11 @@ impl LogicalPlan {
                 join_type,
                 ..
             }) => match join_type {
-                JoinType::Inner | JoinType::Left | JoinType::Right | JoinType::Full | JoinType::LeftSingle => {
+                JoinType::Inner
+                | JoinType::Left
+                | JoinType::Right
+                | JoinType::Full
+                | JoinType::LeftSingle => {
                     if left.schema().fields().is_empty() {
                         right.head_output_expr()
                     } else {
@@ -1554,7 +1559,10 @@ impl LogicalPlan {
                 ..
             }) => match join_type {
                 JoinType::Inner => Some(left.max_rows()? * right.max_rows()?),
-                JoinType::Left | JoinType::Right | JoinType::Full | JoinType::LeftSingle => {
+                JoinType::Left
+                | JoinType::Right
+                | JoinType::Full
+                | JoinType::LeftSingle => {
                     match (left.max_rows()?, right.max_rows()?, join_type) {
                         (0, 0, _) => Some(0),
                         (max_rows, 0, JoinType::Left | JoinType::Full) => Some(max_rows),

datafusion/expr/src/logical_plan/tree_node.rs

@@ -362,6 +362,7 @@ impl TreeNode for LogicalPlan {
                 lateral_join_condition,
                 left,
                 right,
+                any_join,
             }) => (left, right).map_elements(f)?.update_data(|(left, right)| {
                 LogicalPlan::DependentJoin(DependentJoin {
                     schema,
@@ -372,6 +373,7 @@ impl TreeNode for LogicalPlan {
                     lateral_join_condition,
                     left,
                     right,
+                    any_join,
                 })
             }),
         })