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

vortex-file/tests/test_write_table.rs

@@ -167,3 +167,46 @@ async fn test_dict_listview_validity_roundtrip() {
     vortex_array::assert_arrays_eq!(data, chunk);
     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,11 @@ impl LayoutStrategy for CollectStrategy {
                 chunks.push(chunk);
             }
 
-            let collected = ChunkedArray::try_new(chunks, _dtype)?.into_array();
-            yield (latest_sequence_id.vortex_expect("must have visited at least one chunk"), collected);
+            // an empty input yields no chunk; the child layout handles it.
+            if let Some(sequence_id) = latest_sequence_id {
+                let collected = ChunkedArray::try_new(chunks, _dtype)?.into_array();
+                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?;