block equality/ordering based predicates on null producing joins (#3939)

polars/polars-lazy/src/logical_plan/optimizer/predicate_pushdown/mod.rs

@@ -402,7 +402,10 @@ impl PredicatePushDown {
                         |e: &AExpr| matches!(e, AExpr::IsUnique(_) | AExpr::Duplicated(_));
 
                     let checks_nulls =
-                        |e: &AExpr| matches!(e, AExpr::IsNull(_) | AExpr::IsNotNull(_));
+                        |e: &AExpr| matches!(e, AExpr::IsNull(_) | AExpr::IsNotNull(_)) ||
+                            // any operation that checks for equality or ordering can be wrong because
+                            // the join can produce null values
+                            matches!(e, AExpr::BinaryExpr {op, ..} if !matches!(op, Operator::NotEq));
                     if has_aexpr(predicate, expr_arena, matches)
                         // join might create null values.
                         || has_aexpr(predicate, expr_arena, checks_nulls)

polars/tests/it/lazy/predicate_queries.rs

@@ -164,3 +164,25 @@ fn test_is_in_categorical_3420() -> Result<()> {
     assert!(out?.frame_equal(&expected));
     Ok(())
 }
+
+#[test]
+fn test_predicate_pushdown_blocked_by_outer_join() -> Result<()> {
+    let df1 = df! {
+        "a" => ["a1", "a2"],
+        "b" => ["b1", "b2"]
+    }?;
+    let df2 = df! {
+        "b" => ["b2", "b3"],
+        "c" => ["c2", "c3"]
+    }?;
+    let df = df1.lazy().outer_join(df2.lazy(), col("b"), col("b"));
+    let out = df.clone().filter(col("a").eq(lit("a1"))).collect()?;
+    let null: Option<&str> = None;
+    let expected = df![
+        "a" => ["a1"],
+        "b" => ["b1"],
+        "c" => [null],
+    ]?;
+    assert!(out.frame_equal_missing(&expected));
+    Ok(())
+}