fix: Incorrect length check on streaming zip

crates/polars-stream/src/nodes/zip.rs

@@ -172,47 +172,28 @@ impl ComputeNode for ZipNode {
 
         let mut all_broadcast = true;
         let mut all_done_or_broadcast = true;
-        let mut at_least_one_non_broadcast_done = false;
-        let mut at_least_one_non_broadcast_nonempty = false;
+        let mut nonbroadcast_len = None;
+        let mut all_nonbroadcast_match_len = true;
         for (recv_idx, recv_state) in recv.iter().enumerate() {
             let input_head = &mut self.input_heads[recv_idx];
             if *recv_state == PortState::Done {
                 input_head.notify_no_more_morsels();
-
                 all_done_or_broadcast &=
                     input_head.is_broadcast == Some(true) || input_head.total_len == 0;
-                at_least_one_non_broadcast_done |=
-                    input_head.is_broadcast == Some(false) && input_head.total_len == 0;
+                if input_head.is_broadcast != Some(true) {
+                    all_nonbroadcast_match_len &=
+                        nonbroadcast_len.is_none_or(|l| l == input_head.total_len);
+                    nonbroadcast_len = Some(input_head.total_len);
+                }
             } else {
                 all_done_or_broadcast = false;
             }
 
             all_broadcast &= input_head.is_broadcast == Some(true);
-            at_least_one_non_broadcast_nonempty |=
-                input_head.is_broadcast == Some(false) && input_head.total_len > 0;
         }
 
-        match self.zip_behavior {
-            ZipBehavior::Broadcast => {
-                polars_ensure!(
-                    !(at_least_one_non_broadcast_done && at_least_one_non_broadcast_nonempty),
-                    ShapeMismatch: "zip node received non-equal length inputs"
-                );
-            },
-            ZipBehavior::Strict => {
-                if let Some(first_len) = self.input_heads.first().map(|h| h.total_len) {
-                    let all_len_equal = self
-                        .input_heads
-                        .iter()
-                        .filter(|h| h.is_broadcast == Some(false))
-                        .all(|h| h.total_len == first_len);
-                    polars_ensure!(
-                        all_len_equal,
-                        ShapeMismatch: "zip node received non-equal length inputs"
-                    );
-                }
-            },
-            ZipBehavior::NullExtend => {},
+        if !matches!(self.zip_behavior, ZipBehavior::NullExtend) {
+            polars_ensure!(all_nonbroadcast_match_len, ShapeMismatch: "zip node received non-equal length inputs");
         }
 
         let all_output_sent = all_done_or_broadcast && !all_broadcast;

py-polars/tests/unit/streaming/test_streaming.py

@@ -456,3 +456,22 @@ def test_streaming_strptime_infer_leading_nulls(
     result = df.lazy().select(pl.col("s").str.to_date()).collect(engine="streaming")
     expected = df.lazy().select(pl.col("s").str.to_date()).collect()
     assert_frame_equal(result, expected)
+
+
+def test_streaming_hconcat_strict_27372() -> None:
+    data = pl.LazyFrame({"ct": [1, 2, 3]}, schema={"ct": pl.UInt8})
+    lf = pl.concat(
+        [
+            data.select(
+                x=pl.col.ct
+                ^ pl.lit(pl.Series("LUT", [[0]], pl.List(pl.UInt8))).list.get(0)
+            ),
+            data,
+        ],
+        how="horizontal",
+        strict=True,
+    )
+
+    result = lf.collect(engine="streaming")
+    expected = lf.collect(engine="in-memory")
+    assert_frame_equal(result, expected)