fix(rust, python): fix join suffix collision

polars/polars-lazy/polars-plan/src/logical_plan/optimizer/projection_pushdown/joins.rs

@@ -346,7 +346,7 @@ fn process_projection(
     // Thus joining two tables with both a foo column leads to ["foo", "foo_right"]
 
     // try to push down projection in either of two tables
-    if !proj_pd.join_push_down(
+    let (pushed_at_least_once, already_projected) = proj_pd.join_push_down(
         schema_left,
         schema_right,
         proj,
@@ -355,7 +355,9 @@ fn process_projection(
         names_left,
         names_right,
         expr_arena,
-    )
+    );
+
+    if !(pushed_at_least_once || already_projected)
     // did not succeed push down in any tables.,
     // this might be due to the suffix in the projection name
     // this branch tries to pushdown the column without suffix
@@ -381,7 +383,7 @@ fn process_projection(
     }
     // did succeed pushdown at least in any of the two tables
     // if not already added locally we ensure we project local as well
-    else if add_local {
+    else if add_local && pushed_at_least_once {
         // always also do the projection locally, because the join columns may not be
         // included in the projection.
         // for instance:

polars/polars-lazy/polars-plan/src/logical_plan/optimizer/projection_pushdown/mod.rs

@@ -212,27 +212,29 @@ impl ProjectionPushDown {
         names_left: &mut PlHashSet<Arc<str>>,
         names_right: &mut PlHashSet<Arc<str>>,
         expr_arena: &mut Arena<AExpr>,
-    ) -> bool {
+    ) -> (bool, bool) {
         let mut pushed_at_least_one = false;
+        let mut already_projected = false;
         let names = aexpr_to_leaf_names(proj, expr_arena);
         let root_projections = aexpr_to_leaf_nodes(proj, expr_arena);
 
         for (name, root_projection) in names.into_iter().zip(root_projections) {
-            if check_input_node(root_projection, schema_left, expr_arena)
-                && names_left.insert(name.clone())
-            {
+            let was_not_in_left = names_left.insert(name.clone());
+            let was_not_in_right = names_right.insert(name.clone());
+            already_projected |= !was_not_in_left;
+            already_projected |= !was_not_in_right;
+
+            if check_input_node(root_projection, schema_left, expr_arena) && was_not_in_left {
                 pushdown_left.push(proj);
                 pushed_at_least_one = true;
             }
-            if check_input_node(root_projection, schema_right, expr_arena)
-                && names_right.insert(name)
-            {
+            if check_input_node(root_projection, schema_right, expr_arena) && was_not_in_right {
                 pushdown_right.push(proj);
                 pushed_at_least_one = true;
             }
         }
 
-        pushed_at_least_one
+        (pushed_at_least_one, already_projected)
     }
 
     /// This pushes down current node and assigns the result to this node.

polars/polars-lazy/polars-plan/src/logical_plan/optimizer/projection_pushdown/semi_anti_join.rs

@@ -47,7 +47,7 @@ pub(super) fn process_semi_anti_join(
         for proj in acc_projections {
             let add_local = process_alias(proj, &mut local_projection, expr_arena, true);
 
-            proj_pd.join_push_down(
+            let _ = proj_pd.join_push_down(
                 &schema_left,
                 &schema_right,
                 proj,

py-polars/tests/unit/test_projections.py

@@ -271,3 +271,23 @@ def test_distinct_projection_pd_7578() -> None:
         "bar": ["a", "b"],
         "count": [3, 2],
     }
+
+
+def test_join_suffix_collision_9562() -> None:
+    df = pl.DataFrame(
+        {
+            "foo": [1, 2, 3],
+            "bar": [6.0, 7.0, 8.0],
+            "ham": ["a", "b", "c"],
+        }
+    )
+    other_df = pl.DataFrame(
+        {
+            "apple": ["x", "y", "z"],
+            "ham": ["a", "b", "d"],
+        }
+    )
+    df.join(other_df, on="ham")
+    assert df.lazy().join(
+        other_df.lazy(), how="inner", left_on="ham", right_on="ham", suffix="m"
+    ).select("ham").collect().to_dict(False) == {"ham": ["a", "b"]}