fix(layout): don't panic collecting an empty stream (#8472)

## Summary

`CollectStrategy` collects its whole input into a single chunk for a
child that requires exactly one chunk. When the input stream is empty,
e.g. writing a zero-row table with a nullable struct column whose
validity substream is empty, no chunk supplied a sequence id and it
panicked with `must have visited at least one chunk`. `CollectStrategy`
now yields nothing on empty input, and `FlatLayoutStrategy` returns an
empty `ChunkedLayout` instead of requiring a single chunk, so no segment
is written for an empty array.

The issue mentions the fuzzer's `assume()` guard could be dropped once
this is fixed. I left it in place here: reading a nullable struct nested
in a struct back is a separate bug (#8348), so the round-trip only works
once both are fixed.

Closes: #8347

## Testing

Added a regression test in `vortex-file/tests/test_write_table.rs` that
writes a zero-row nullable struct column; it panicked before this change
and now writes and reads back as zero rows. The issue's Python repro
(`vx.io.write` of an empty `struct` column) no longer panics. `cargo
nextest run -p vortex-layout -p vortex-file` passes (177 tests,
including the segment-ordering tests). `fmt --all` + `clippy
--all-targets --all-features` clean.

---

I'm Korean, so sorry if any wording reads a little awkward.

---------

Signed-off-by: Han Damin <miniex@daminstudio.net>

Author: miniex

vortex-file/tests/test_write_table.rs

@@ -166,3 +166,46 @@ async fn test_dict_listview_validity_roundtrip() {
     vortex_array::assert_arrays_eq!(data, chunk, &mut SESSION.create_execution_ctx());
     assert!(stream.next().await.is_none(), "expected a single chunk");
 }
+
+/// Regression test: writing a zero-row table with a nullable struct column used to
+/// panic in `CollectStrategy` ("must have visited at least one chunk").
+#[tokio::test]
+async fn test_write_empty_nullable_struct_column() {
+    let inner = StructArray::new(
+        FieldNames::from(["a"]),
+        vec![PrimitiveArray::from_iter(Vec::<i32>::new()).into_array()],
+        0,
+        Validity::AllValid,
+    )
+    .into_array();
+
+    let data = StructArray::new(
+        FieldNames::from(["c0"]),
+        vec![inner],
+        0,
+        Validity::NonNullable,
+    )
+    .into_array();
+
+    let mut bytes = Vec::new();
+    SESSION
+        .write_options()
+        .write(&mut bytes, data.to_array_stream())
+        .await
+        .expect("writing an empty nullable struct column should not panic");
+
+    let bytes = ByteBuffer::from(bytes);
+    let vxf = SESSION.open_options().open_buffer(bytes).expect("open");
+    let stream = vxf
+        .scan()
+        .expect("scan")
+        .into_stream()
+        .expect("into_stream");
+    pin_mut!(stream);
+
+    let mut rows = 0usize;
+    while let Some(next) = stream.next().await {
+        rows += next.expect("read back should succeed").len();
+    }
+    assert_eq!(rows, 0);
+}

vortex-layout/src/layouts/collect.rs

@@ -10,7 +10,6 @@ use futures::pin_mut;
 use vortex_array::ArrayContext;
 use vortex_array::IntoArray;
 use vortex_array::arrays::ChunkedArray;
-use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
 use vortex_session::VortexSession;
 
@@ -61,8 +60,13 @@ impl LayoutStrategy for CollectStrategy {
                 chunks.push(chunk);
             }
 
+            // an empty input yields no chunk; the child layout handles it.
+            let Some(sequence_id) = latest_sequence_id else {
+                return;
+            };
+
             let collected = ChunkedArray::try_new(chunks, _dtype)?.into_array();
-            yield (latest_sequence_id.vortex_expect("must have visited at least one chunk"), collected);
+            yield (sequence_id, collected);
         };
 
         let adapted = Box::pin(SequentialStreamAdapter::new(dtype, collected_stream));

vortex-layout/src/layouts/flat/writer.rs

@@ -29,6 +29,8 @@ use vortex_utils::aliases::hash_set::HashSet;
 use crate::IntoLayout;
 use crate::LayoutRef;
 use crate::LayoutStrategy;
+use crate::children::OwnedLayoutChildren;
+use crate::layouts::chunked::ChunkedLayout;
 use crate::layouts::flat::FlatLayout;
 use crate::layouts::flat::flat_layout_inline_array_node;
 use crate::segments::SegmentSinkRef;
@@ -104,7 +106,13 @@ impl LayoutStrategy for FlatLayoutStrategy {
     ) -> VortexResult<LayoutRef> {
         let ctx = ctx.clone();
         let Some(chunk) = stream.next().await else {
-            vortex_bail!("flat layout needs a single chunk");
+            // an empty input has no segment to write.
+            return Ok(ChunkedLayout::new(
+                0,
+                stream.dtype().clone(),
+                OwnedLayoutChildren::layout_children(vec![]),
+            )
+            .into_layout());
         };
         let (sequence_id, chunk) = chunk?;