fix(rust, python): fix streaming empty join panic (#5534)

polars/Cargo.toml

@@ -288,6 +288,9 @@ ahash = "0.7"
 criterion = "0.3"
 rand = "0.8"
 
+[build-dependencies]
+version_check = "0.9.4"
+
 # see: https://bheisler.github.io/criterion.rs/book/faq.html
 [lib]
 bench = false

polars/build.rs

@@ -0,0 +1,7 @@
+fn main() {
+    println!("cargo:rerun-if-changed=build.rs");
+    let channel = version_check::Channel::read().unwrap();
+    if channel.is_nightly() {
+        println!("cargo:rustc-cfg=feature=\"nightly\"");
+    }
+}

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

@@ -131,6 +131,14 @@ pub(super) fn compare_fn(
 }
 
 impl GenericBuild {
+    fn is_empty(&self) -> bool {
+        match self.chunks.len() {
+            0 => true,
+            1 => self.chunks[0].data.height() == 0,
+            _ => false,
+        }
+    }
+
     #[inline]
     fn number_of_keys(&self) -> usize {
         self.join_columns_left.len()
@@ -175,7 +183,17 @@ impl GenericBuild {
 
 impl Sink for GenericBuild {
     fn sink(&mut self, context: &PExecutionContext, chunk: DataChunk) -> PolarsResult<SinkResult> {
+        // we do some juggling here so that we don't
+        // end up with empty chunks
+        // But we always want one empty chunk if all is empty as we need
+        // to finish the join
+        if self.chunks.len() == 1 && self.chunks[0].data.height() == 0 {
+            self.chunks.pop().unwrap();
+        }
         if chunk.is_empty() {
+            if self.chunks.is_empty() {
+                self.chunks.push(chunk)
+            }
             return Ok(SinkResult::CanHaveMoreInput);
         }
         let mut hashes = std::mem::take(&mut self.hashes);
@@ -225,7 +243,15 @@ impl Sink for GenericBuild {
     }
 
     fn combine(&mut self, mut other: Box<dyn Sink>) {
+        if self.is_empty() {
+            let other = other.as_any().downcast_mut::<Self>().unwrap();
+            std::mem::swap(self, other);
+            return;
+        }
         let other = other.as_any().downcast_ref::<Self>().unwrap();
+        if other.is_empty() {
+            return;
+        }
         let mut tuple_buf = Vec::with_capacity(self.number_of_keys());
 
         let chunks_offset = self.chunks.len() as IdxSize;
@@ -300,7 +326,9 @@ impl Sink for GenericBuild {
                         .map(|chunk| chunk.data),
                 );
                 if let Ok(n_chunks) = left_df.n_chunks() {
-                    assert_eq!(n_chunks, chunks_len);
+                    if left_df.height() > 0 {
+                        assert_eq!(n_chunks, chunks_len);
+                    }
                 }
                 let materialized_join_cols =
                     Arc::new(std::mem::take(&mut self.materialized_join_cols));

py-polars/tests/unit/test_predicates.py

@@ -93,4 +93,4 @@ def test_streaming_empty_df() -> None:
 
     assert df.lazy().join(df.lazy(), on="a", how="inner").filter(
         2 == 1  # noqa: SIM300
-    ).collect().to_dict(False) == {"a": [], "b": [], "b_right": []}
+    ).collect(allow_streaming=True).to_dict(False) == {"a": [], "b": [], "b_right": []}