fix(rust, python): fix streaming joins where the join order has been … (pola-rs#6143)

Author: ritchie46

polars/polars-lazy/polars-pipe/src/executors/operators/dummy.rs

@@ -17,4 +17,8 @@ impl Operator for Dummy {
     fn split(&self, _thread_no: usize) -> Box<dyn Operator> {
         Box::new(Self {})
     }
+
+    fn fmt(&self) -> &str {
+        "dummy"
+    }
 }

polars/polars-lazy/polars-pipe/src/executors/operators/filter.rs

@@ -33,4 +33,7 @@ impl Operator for FilterOperator {
     fn split(&self, _thread_no: usize) -> Box<dyn Operator> {
         Box::new(self.clone())
     }
+    fn fmt(&self) -> &str {
+        "filter"
+    }
 }

polars/polars-lazy/polars-pipe/src/executors/operators/projection.rs

@@ -24,6 +24,9 @@ impl Operator for FastProjectionOperator {
     fn split(&self, _thread_no: usize) -> Box<dyn Operator> {
         Box::new(self.clone())
     }
+    fn fmt(&self) -> &str {
+        "fast_join_projection"
+    }
 }
 
 #[derive(Clone)]
@@ -49,6 +52,9 @@ impl Operator for ProjectionOperator {
     fn split(&self, _thread_no: usize) -> Box<dyn Operator> {
         Box::new(self.clone())
     }
+    fn fmt(&self) -> &str {
+        "projection"
+    }
 }
 
 #[derive(Clone)]
@@ -79,4 +85,7 @@ impl Operator for HstackOperator {
     fn split(&self, _thread_no: usize) -> Box<dyn Operator> {
         Box::new(self.clone())
     }
+    fn fmt(&self) -> &str {
+        "hstack"
+    }
 }

polars/polars-lazy/polars-pipe/src/executors/sinks/groupby/aggregates/convert.rs

@@ -5,6 +5,7 @@ use polars_core::datatypes::Field;
 use polars_core::error::PolarsResult;
 use polars_core::prelude::{DataType, SchemaRef, Series, IDX_DTYPE};
 use polars_core::schema::Schema;
+use polars_plan::dsl::Expr;
 use polars_plan::logical_plan::{ArenaExprIter, Context};
 use polars_plan::prelude::{AAggExpr, AExpr};
 use polars_utils::arena::{Arena, Node};
@@ -30,6 +31,10 @@ impl PhysicalPipedExpr for Count {
     fn field(&self, _input_schema: &Schema) -> PolarsResult<Field> {
         todo!()
     }
+
+    fn expression(&self) -> Expr {
+        Expr::Count
+    }
 }
 
 pub fn can_convert_to_hash_agg(

polars/polars-lazy/polars-pipe/src/executors/sinks/groupby/generic.rs

@@ -428,4 +428,7 @@ impl Sink for GenericGroupbySink {
     fn as_any(&mut self) -> &mut dyn Any {
         self
     }
+    fn fmt(&self) -> &str {
+        "generic_groupby"
+    }
 }

polars/polars-lazy/polars-pipe/src/executors/sinks/groupby/primitive.rs

@@ -360,6 +360,9 @@ where
     fn as_any(&mut self) -> &mut dyn Any {
         self
     }
+    fn fmt(&self) -> &str {
+        "primitive_groupby"
+    }
 }
 
 fn insert_and_get<T>(

polars/polars-lazy/polars-pipe/src/executors/sinks/groupby/string.rs

@@ -358,6 +358,9 @@ impl Sink for Utf8GroupbySink {
     fn as_any(&mut self) -> &mut dyn Any {
         self
     }
+    fn fmt(&self) -> &str {
+        "utf8_groupby"
+    }
 }
 
 // write agg_idx to the hashes buffer.

polars/polars-lazy/polars-pipe/src/executors/sinks/joins/cross.rs

@@ -62,6 +62,10 @@ impl Sink for CrossJoin {
     fn as_any(&mut self) -> &mut dyn Any {
         self
     }
+
+    fn fmt(&self) -> &str {
+        "cross_join_sink"
+    }
 }
 
 #[derive(Clone)]
@@ -152,4 +156,8 @@ impl Operator for CrossJoinProbe {
     fn split(&self, _thread_no: usize) -> Box<dyn Operator> {
         Box::new(self.clone())
     }
+
+    fn fmt(&self) -> &str {
+        "cross_join_probe"
+    }
 }

polars/polars-lazy/polars-pipe/src/executors/sinks/joins/generic_build.rs

@@ -151,6 +151,7 @@ impl GenericBuild {
         chunk: &DataChunk,
     ) -> PolarsResult<&[Series]> {
         self.join_series.clear();
+
         for phys_e in self.join_columns_left.iter() {
             let s = phys_e.evaluate(chunk, context.execution_state.as_any())?;
             let s = s.to_physical_repr();
@@ -337,7 +338,7 @@ impl Sink for GenericBuild {
                 let suffix = self.suffix.clone();
                 let hb = self.hb.clone();
                 let hash_tables = Arc::new(std::mem::take(&mut self.hash_tables));
-                let join_columns_left = self.join_columns_right.clone();
+                let join_columns_left = self.join_columns_left.clone();
                 let join_columns_right = self.join_columns_right.clone();
 
                 // take the buffers, this saves one allocation
@@ -369,6 +370,9 @@ impl Sink for GenericBuild {
     fn as_any(&mut self) -> &mut dyn Any {
         self
     }
+    fn fmt(&self) -> &str {
+        "generic_join_build"
+    }
 }
 
 pub(super) struct KeysIter<'a> {

polars/polars-lazy/polars-pipe/src/executors/sinks/joins/inner_left.rs

@@ -66,15 +66,12 @@ impl GenericJoinProbe {
         hash_tables: Arc<Vec<PlIdHashMap<Key, Vec<ChunkId>>>>,
         join_columns_left: Arc<Vec<Arc<dyn PhysicalPipedExpr>>>,
         join_columns_right: Arc<Vec<Arc<dyn PhysicalPipedExpr>>>,
-        mut swapped_or_left: bool,
+        swapped_or_left: bool,
         join_series: Vec<Series>,
         hashes: Vec<u64>,
         context: &PExecutionContext,
         how: JoinType,
     ) -> Self {
-        if matches!(how, JoinType::Left) {
-            swapped_or_left = true
-        }
         if swapped_or_left {
             let tmp = DataChunk {
                 data: df_a.slice(0, 1),
@@ -329,4 +326,7 @@ impl Operator for GenericJoinProbe {
         let new = self.clone();
         Box::new(new)
     }
+    fn fmt(&self) -> &str {
+        "generic_join_probe"
+    }
 }