init_fmt_check

Author: B Vadlamani

datafusion/physical-plan/src/joins/utils.rs

@@ -1789,18 +1789,27 @@ pub(super) fn equal_rows_arr(
 
         let mut is_equal = true;
         for (col_idx, cmp) in comparators.iter().enumerate() {
-            match (null_equality, cmp(left_idx, right_idx)) {
-                (NullEquality::NullEqualsNull, Ordering::Equal) => continue,
-                (NullEquality::NullEqualsNothing, Ordering::Equal) => {
-                    if left_arrays.get(col_idx).unwrap().is_null(left_idx) {
+            let left_arr = left_arrays.get(col_idx).unwrap();
+            let right_arr = right_arrays.get(col_idx).unwrap();
+            let left_null = left_arr.data_type() == &DataType::Null || left_arr.is_null(left_idx);
+            let right_null = right_arr.data_type() == &DataType::Null || right_arr.is_null(right_idx);
+
+            match (null_equality, left_null, right_null) {
+                (NullEquality::NullEqualsNull, true, true) => continue, // Nulls match
+                (NullEquality::NullEqualsNothing, true, _) | (NullEquality::NullEqualsNothing, _, true) => {
+                    is_equal = false; // Nulls never match
+                    break;
+                }
+                (_, true, false) | (_, false, true) => {
+                    is_equal = false; // Different null states
+                    break;
+                }
+                (_, false, false) => {
+                    if cmp(left_idx, right_idx) != Ordering::Equal {
                         is_equal = false;
                         break;
                     }
                 }
-                _ => {
-                    is_equal = false;
-                    break;
-                }
             }
         }
 

test_null_join.rs

@@ -0,0 +1,12 @@
+use datafusion::prelude::*;
+use datafusion_common::Result;
+use std::sync::Arc;
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    let ctx = SessionContext::new();
+    let sql = "SELECT * FROM (SELECT null AS id1) t1 INNER JOIN (SELECT null AS id2) t2 ON id1 = id2";
+    let df = ctx.sql(sql).await?;
+    df.show().await?;
+    Ok(())
+}