fix invalid predicate pushdown in left join

polars/polars-lazy/src/frame.rs

@@ -895,16 +895,18 @@ impl LazyFrame {
     ///
     /// fn example(ldf: LazyFrame, other: LazyFrame) -> LazyFrame {
     ///         ldf
-    ///         .join(other, vec![col("foo"), col("bar")], vec![col("foo"), col("bar")], JoinType::Inner)
+    ///         .join(other, [col("foo"), col("bar")], [col("foo"), col("bar")], JoinType::Inner)
     /// }
     /// ```
-    pub fn join(
+    pub fn join<E: AsRef<[Expr]>>(
         self,
         other: LazyFrame,
-        left_on: Vec<Expr>,
-        right_on: Vec<Expr>,
+        left_on: E,
+        right_on: E,
         how: JoinType,
     ) -> LazyFrame {
+        let left_on = left_on.as_ref().to_vec();
+        let right_on = right_on.as_ref().to_vec();
         self.join_builder()
             .with(other)
             .left_on(left_on)

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

@@ -330,6 +330,7 @@ impl PredicatePushDown {
                         local_predicates.push(predicate);
                         continue;
                     }
+                    // these indicate to which tables we are going to push down the predicate
                     let mut filter_left = false;
                     let mut filter_right = false;
 
@@ -354,10 +355,20 @@ impl PredicatePushDown {
                         );
                         filter_right = true;
                     }
-                    // if not pushed down on of the tables we have to do it locally.
-                    if !(filter_left | filter_right) {
-                        local_predicates.push(predicate);
-                        continue;
+                    match (filter_left, filter_right, options.how) {
+                        // if not pushed down on of the tables we have to do it locally.
+                        (false, false, _) |
+                        // if left join and predicate only available in right table,
+                        // 'we should not filter right, because that would lead to
+                        // invalid results.
+                        // see: #2057
+                        (false, true, JoinType::Left)
+                        => {
+                            local_predicates.push(predicate);
+                            continue;
+                        },
+                        // business as usual
+                        _ => {}
                     }
                     // An outer join or left join may create null values.
                     // we also do it local

polars/polars-lazy/src/tests/predicate_pushdown.rs

@@ -36,3 +36,24 @@ fn test_pred_pd_1() -> Result<()> {
 
     Ok(())
 }
+
+#[test]
+fn test_no_left_join_pass() -> Result<()> {
+    let df1 = df![
+        "foo" => ["abc", "def", "ghi"],
+        "idx1" => [0, 0, 1],
+    ]?;
+    let df2 = df![
+        "bar" => [5, 6],
+        "idx2" => [0, 1],
+    ]?;
+
+    let out = df1
+        .lazy()
+        .join(df2.lazy(), [col("idx1")], [col("idx2")], JoinType::Left)
+        .filter(col("bar").eq(lit(5i32)))
+        .collect()?;
+
+    dbg!(out);
+    Ok(())
+}