fix(rust, python): compute correct offset for streaming join on multi… (

#5479)

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

@@ -159,9 +159,12 @@ impl GenericBuild {
     ) {
         buf.clear();
         // get the right columns from the linearly packed buffer
+        let n_keys = self.number_of_keys();
+        let chunk_offset = chunk_idx as usize * n_keys;
+        let chunk_end = chunk_offset + n_keys;
         let join_cols = self
             .materialized_join_cols
-            .get_unchecked_release(chunk_idx as usize..chunk_idx as usize + self.number_of_keys());
+            .get_unchecked_release(chunk_offset..chunk_end);
         buf.extend(
             join_cols
                 .iter()

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

@@ -310,3 +310,24 @@ fn test_streaming_aggregate_join() -> PolarsResult<()> {
     assert_eq!(out_streaming.shape(), (3, 3));
     Ok(())
 }
+
+#[test]
+fn test_streaming_inner_join4() -> PolarsResult<()> {
+    let lfa =
+        df!["a" => [1, 3, 4, 8, 9, 12, 13, 18, 18, 21, 22, 24, 28, 28, 32, 35, 35, 36, 39, 39],
+    "b"=> [10, 0, 15, 5, 4, 18, 17, 14, 19, 9, 2, 12, 7, 8, 11, 6, 16, 3, 1, 13]]?
+        .lazy();
+
+    let lfb = df!["a"=> [16, 8, 27, 26, 1, 4, 30, 8, 18, 19, 2, 14, 30, 33, 19, 1, 31, 36, 37, 21],
+        "b" => [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19]]?
+    .lazy();
+
+    let on = [col("a"), col("b")];
+    let out = lfa
+        .join(lfb, on.clone(), on.clone(), JoinType::Inner)
+        .with_streaming(true)
+        .collect()
+        .unwrap();
+
+    Ok(())
+}