Repository: pola-rs/polars (language: python)

Issue:
join_where returns different results depending on order of equality expressions
### Checks

- [X] I have checked that this issue has not already been reported.
- [X] I have confirmed this bug exists on the [latest version](https://pypi.org/project/polars/) of Polars.

### Reproducible example

```python
import polars as pl

from polars.testing import assert_frame_equal

left = pl.LazyFrame(
    {
        "a": [1, 2, 3, 1, None],
        "b": [1, 2, 3, 4, 5],
        "c": [2, 3, 4, 5, 6],
    }
)

right = pl.LazyFrame(
    {
        "a": [1, 4, 3, 8, None, None, 1],
        "c": [2, 3, 4, 5, 6, 7, 8],
        "d": [6, None, 3, 8, -1, 2, 4],
    }
)


q_good = left.join_where(
    right,
    pl.col("a") < pl.col("d"),
    pl.col("b") < pl.col("c_right"),
    pl.col("a") < pl.col("a_right"),
)

q_bad = left.join_where(
    right,
    pl.col("b") < pl.col("c_right"),
    pl.col("a") < pl.col("a_right"),
    pl.col("a") < pl.col("d"),
)

q_expect = (
    left.join(right, how="cross")
    .collect()
    .filter(
        (pl.col("a") < pl.col("d"))
        & (pl.col("b") < pl.col("c_right"))
        & (pl.col("a") < pl.col("a_right"))
    )
)

assert_frame_equal(q_expect, q_good.collect(), check_row_order=False)
assert_frame_equal(q_expect, q_bad.collect(), check_row_order=False)
```


### Log output

```shell
join parallel: true
IEJOIN join dataframes finished
dataframe filtered
join parallel: true
IEJOIN join dataframes finished
join parallel: true
CROSS join dataframes finished
dataframe filtered
dataframe filtered
dataframe filtered
```


### Issue description

My understanding of `join_where` is that the conditions are ANDed together to produce the filter on the output table. 

When more than two conditions are applied the third and subsequent conditions are not applied in the `ie_join` implementation itself, but are instead done as postfilters on the result.

However, it looks like the normalisation rule for these filters does different things depending on the exact schemas of the left and right tables.

```
print(q_good.explain())
FILTER [(col("a")) < (col("a_right"))] FROM
  IEJOIN JOIN:
  LEFT PLAN ON: [col("a"), col("b")]
    DF ["a", "b", "c"]; PROJECT */3 COLUMNS; SELECTION: None
  RIGHT PLAN ON: [col("d"), col("c")]
    DF ["a", "c", "d"]; PROJECT */3 COLUMNS; SELECTION: None
  END IEJOIN JOIN
```

```
print(q_bad.explain())
IEJOIN JOIN:
LEFT PLAN ON: [col("b"), col("a")]
  DF ["a", "b", "c"]; PROJECT */3 COLUMNS; SELECTION: None
RIGHT PLAN ON: [col("c"), col("a")]
  DF ["a", "c", "d"]; PROJECT */3 COLUMNS; SELECTION: [(col("a")) < (col("d"))]
END IEJOIN JOIN
```

Notice how in the bad case, the condition `col("a") < col("d")` has been pushed into a selection on the right dataframe. However, this is not right, by the semantics of `join_where`, `col("a")` refers to the column in the left table, whereas `col("d")` refers to the column in the right table.

### Expected behavior

I expect `q_good` and `q_bad` to produce the same result, matching `q_expect`.

### Installed versions

<details>

```
--------Version info---------
Polars:              1.12.0
Index type:          UInt32
Platform:            Linux-6.8.0-48-generic-x86_64-with-glibc2.35
Python:              3.12.7 | packaged by conda-forge | (main, Oct  4 2024, 16:05:46) [GCC 13.3.0]
LTS CPU:             False

----Optional dependencies----
adbc_driver_manager  <not installed>
altair               <not installed>
cloudpickle          3.1.0
connectorx           <not installed>
deltalake            <not installed>
fastexcel            <not installed>
fsspec               2024.10.0
gevent               <not installed>
great_tables         <not installed>
matplotlib           <not installed>
nest_asyncio         1.6.0
numpy                2.0.2
openpyxl             3.1.5
pandas               2.2.3
pyarrow              17.0.0
pydantic             2.9.2
pyiceberg            <not installed>
sqlalchemy           2.0.36
torch                2.4.1.post302
xlsx2csv             <not installed>
xlsxwriter           <not installed>
```

</details>



Discussion / hints:
This appears to be an issue with predicate pushdown. If I run `q_bad.collect(predicate_pushdown=False)` then I get the expected result.
```
print(q_bad.explain(predicate_pushdown=False))
FILTER [(col("a")) < (col("d"))] FROM
  IEJOIN JOIN:
  LEFT PLAN ON: [col("b"), col("a")]
    DF ["a", "b", "c"]; PROJECT */3 COLUMNS; SELECTION: None
  RIGHT PLAN ON: [col("c"), col("a")]
    DF ["a", "c", "d"]; PROJECT */3 COLUMNS; SELECTION: None
  END IEJOIN JOIN
```

Fix patch:
diff --git a/crates/polars-ops/src/frame/join/args.rs b/crates/polars-ops/src/frame/join/args.rs
index 4c845a2ba541..d34c37e7ff67 100644
--- a/crates/polars-ops/src/frame/join/args.rs
+++ b/crates/polars-ops/src/frame/join/args.rs
@@ -174,6 +174,10 @@ impl JoinType {
         }
     }
 
+    pub fn is_cross(&self) -> bool {
+        matches!(self, JoinType::Cross)
+    }
+
     pub fn is_ie(&self) -> bool {
         #[cfg(feature = "iejoin")]
         {
diff --git a/crates/polars-plan/src/plans/conversion/join.rs b/crates/polars-plan/src/plans/conversion/join.rs
index a81c36bef1f8..5222ca73cda1 100644
--- a/crates/polars-plan/src/plans/conversion/join.rs
+++ b/crates/polars-plan/src/plans/conversion/join.rs
@@ -41,7 +41,7 @@ pub fn resolve_join(
     }
 
     let owned = Arc::unwrap_or_clone;
-    if matches!(options.args.how, JoinType::Cross) {
+    if options.args.how.is_cross() {
         polars_ensure!(left_on.len() + right_on.len() == 0, InvalidOperation: "a 'cross' join doesn't expect any join keys");
     } else {
         polars_ensure!(left_on.len() + right_on.len() > 0, InvalidOperation: "expected join keys/predicates");
diff --git a/crates/polars-plan/src/plans/optimizer/collapse_joins.rs b/crates/polars-plan/src/plans/optimizer/collapse_joins.rs
index 778efee6aa9b..07a4a45948b8 100644
--- a/crates/polars-plan/src/plans/optimizer/collapse_joins.rs
+++ b/crates/polars-plan/src/plans/optimizer/collapse_joins.rs
@@ -192,7 +192,7 @@ pub fn optimize(root: Node, lp_arena: &mut Arena<IR>, expr_arena: &mut Arena<AEx
                 left_on,
                 right_on,
                 options,
-            } if matches!(options.args.how, JoinType::Cross) => {
+            } if options.args.how.is_cross() => {
                 if predicates.is_empty() {
                     continue;
                 }
diff --git a/crates/polars-plan/src/plans/optimizer/collect_members.rs b/crates/polars-plan/src/plans/optimizer/collect_members.rs
index 3157821c7696..f829b3362473 100644
--- a/crates/polars-plan/src/plans/optimizer/collect_members.rs
+++ b/crates/polars-plan/src/plans/optimizer/collect_members.rs
@@ -48,7 +48,7 @@ impl MemberCollector {
             match alp {
                 Join { .. } | Union { .. } => self.has_joins_or_unions = true,
                 Filter { input, .. } => {
-                    self.has_filter_with_join_input |= matches!(lp_arena.get(*input), Join { options, .. } if options.args.how == JoinType::Cross)
+                    self.has_filter_with_join_input |= matches!(lp_arena.get(*input), Join { options, .. } if options.args.how.is_cross())
                 },
                 Cache { .. } => self.has_cache = true,
                 ExtContext { .. } => self.has_ext_context = true,
diff --git a/crates/polars-plan/src/plans/optimizer/predicate_pushdown/join.rs b/crates/polars-plan/src/plans/optimizer/predicate_pushdown/join.rs
index 3b23faef8e04..b79de0e2b959 100644
--- a/crates/polars-plan/src/plans/optimizer/predicate_pushdown/join.rs
+++ b/crates/polars-plan/src/plans/optimizer/predicate_pushdown/join.rs
@@ -149,7 +149,8 @@ pub(super) fn process_join(
 
     for (_, predicate) in acc_predicates {
         // Cross joins produce a cartesian product, so if a predicate combines columns from both tables, we should not push down.
-        if matches!(options.args.how, JoinType::Cross)
+        // Inequality joins logically produce a cartesian product, so the same logic applies.
+        if (options.args.how.is_cross() || options.args.how.is_ie())
             && predicate_applies_to_both_tables(
                 predicate.node(),
                 expr_arena,


Test patch:
diff --git a/py-polars/tests/unit/operations/test_inequality_join.py b/py-polars/tests/unit/operations/test_inequality_join.py
index 89eec6bfd224..d306dd173431 100644
--- a/py-polars/tests/unit/operations/test_inequality_join.py
+++ b/py-polars/tests/unit/operations/test_inequality_join.py
@@ -624,3 +624,34 @@ def test_join_partial_column_name_overlap_19119() -> None:
         "a_right": [2],
         "d": [0],
     }
+
+
+def test_join_predicate_pushdown_19580() -> None:
+    left = pl.LazyFrame(
+        {
+            "a": [1, 2, 3, 1],
+            "b": [1, 2, 3, 4],
+            "c": [2, 3, 4, 5],
+        }
+    )
+
+    right = pl.LazyFrame({"a": [1, 3], "c": [2, 4], "d": [6, 3]})
+
+    q = left.join_where(
+        right,
+        pl.col("b") < pl.col("c_right"),
+        pl.col("a") < pl.col("a_right"),
+        pl.col("a") < pl.col("d"),
+    )
+
+    expect = (
+        left.join(right, how="cross")
+        .collect()
+        .filter(
+            (pl.col("a") < pl.col("d"))
+            & (pl.col("b") < pl.col("c_right"))
+            & (pl.col("a") < pl.col("a_right"))
+        )
+    )
+
+    assert_frame_equal(expect, q.collect(), check_row_order=False)