fix: write empty nested list streams with structural layouts (#10533)

## Summary

Writing `list<list<int32>>` rows containing only empty or null outer
lists fails with `ListLayoutStrategy needs at least one chunk` when
structural list layouts are enabled. This reproduces in PyVortex 0.86.1
and 0.88.0 with `VORTEX_EXPERIMENTAL_LIST_LAYOUT=1`.

## Changes

Handle an empty list stream before spawning child writers by returning
the same typed zero-row chunked layout used by the flat writer. Pass the
first chunk explicitly into the nonempty transposition path, preserving
its sequence ID, bounded streaming, and existing layout behavior.

Add file write/read regression cases for empty lists, nullable empty
lists, all-null lists, and mixed empty/null rows. Replace the old
empty-stream rejection test with nullable/nonnullable dtype and
row-count checks.

## Validation

- All four new file-roundtrip cases fail on unmodified `develop`
(`18aebea`) with the reported error.
- After the fix, 76 selected tests pass, including all four regression
cases and both empty-stream nullability cases:
  ```sh
cargo nextest run --locked -p vortex-file -p vortex-layout -E
'test(nested_) | test(writer_empty_chunks) |
test(layouts::list::writer::tests)'
  ```
- Formatted both changed files with CI's pinned `nightly-2026-09-10`
rustfmt; `git diff --check` passes.

Implemented with OpenAI Codex assistance.

Signed-off-by: Daniel King <dan@spiraldb.com>

Author: danking

vortex-file/src/tests.rs

@@ -2037,6 +2037,30 @@ async fn nested_list_of_list_roundtrip() -> VortexResult<()> {
     Ok(())
 }
 
+#[rstest]
+#[case::empty_lists(Validity::NonNullable)]
+#[case::nullable_empty_lists(Validity::AllValid)]
+#[case::null_lists(Validity::AllInvalid)]
+#[case::mixed_empty_and_null(
+    Validity::Array(BoolArray::from_iter([true, false]).into_array())
+)]
+#[tokio::test]
+#[cfg_attr(miri, ignore)]
+async fn nested_empty_list_roundtrip(#[case] validity: Validity) -> VortexResult<()> {
+    let inner = ListArray::try_new(
+        PrimitiveArray::new::<i32>(buffer![], Validity::NonNullable).into_array(),
+        buffer![0u32].into_array(),
+        Validity::NonNullable,
+    )?
+    .into_array();
+    let outer = ListArray::try_new(inner, buffer![0u32, 0, 0].into_array(), validity)?.into_array();
+    let array = StructArray::from_fields(&[("nested", outer)])?.into_array();
+
+    let result = write_read_roundtrip(array.clone()).await?;
+    assert_arrays_eq!(result, array, &mut SESSION.create_execution_ctx());
+    Ok(())
+}
+
 type MapEntryFixture<'a> = (i32, Option<&'a str>);
 type MapRowFixture<'a> = Option<Vec<MapEntryFixture<'a>>>;
 

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

@@ -33,6 +33,8 @@ use vortex_session::VortexSession;
 use crate::LayoutRef;
 use crate::LayoutStrategy;
 use crate::LayoutWriterContext;
+use crate::children::OwnedLayoutChildren;
+use crate::layouts::chunked::ChunkedLayout;
 use crate::layouts::flat::writer::FlatLayoutStrategy;
 use crate::layouts::list::ListLayout;
 use crate::segments::SegmentSinkRef;
@@ -117,7 +119,7 @@ impl LayoutStrategy for ListLayoutStrategy {
         &self,
         ctx: LayoutWriterContext,
         segment_sink: SegmentSinkRef,
-        stream: SendableSequentialStream,
+        mut stream: SendableSequentialStream,
         mut eof: SequencePointer,
         session: &VortexSession,
     ) -> VortexResult<LayoutRef> {
@@ -129,6 +131,20 @@ impl LayoutStrategy for ListLayoutStrategy {
                 .await;
         }
 
+        // Empty nested element streams are valid below empty or all-null lists.
+        // As in the flat writer, no input chunks require no child segments.
+        let first_chunk = match stream.next().await {
+            Some(chunk) => chunk?,
+            None => {
+                return Ok(ChunkedLayout::new(
+                    0,
+                    dtype,
+                    OwnedLayoutChildren::layout_children(vec![]),
+                )
+                .into_layout());
+            }
+        };
+
         let is_nullable = dtype.is_nullable();
         let element_dtype = dtype
             .as_list_element_opt()
@@ -153,6 +169,7 @@ impl LayoutStrategy for ListLayoutStrategy {
         // positions. Kept joined with the child writers below so producer errors surface rather
         // than being hidden as an early channel close.
         let fanout_fut = transpose_list_column(
+            first_chunk,
             stream,
             session.clone(),
             elements_tx,
@@ -214,10 +231,12 @@ impl LayoutStrategy for ListLayoutStrategy {
 /// sub-streams, rebasing each chunk's local `offsets` to global `u64` positions so the single
 /// `offsets` child indexes into the concatenated `elements` child.
 ///
+/// `first_chunk` establishes nonempty input; `stream` contains the remaining chunks.
 /// `validity_tx` is `Some` exactly when the list is nullable. Errors surface to the caller, which
 /// joins this against the child writers, rather than being hidden as an early channel close.
 async fn transpose_list_column(
-    mut stream: SendableSequentialStream,
+    first_chunk: (SequenceId, ArrayRef),
+    stream: SendableSequentialStream,
     session: VortexSession,
     elements_tx: kanal::AsyncSender<ChildChunk>,
     offsets_tx: kanal::AsyncSender<ChildChunk>,
@@ -226,10 +245,9 @@ async fn transpose_list_column(
     let mut exec_ctx = session.create_execution_ctx();
     let mut element_base: u64 = 0;
     let mut first = true;
-    let mut saw_chunk = false;
-    while let Some(chunk) = stream.next().await {
+    let mut chunks = futures::stream::iter([Ok(first_chunk)]).chain(stream);
+    while let Some(chunk) = chunks.next().await {
         let (sequence_id, array) = chunk?;
-        saw_chunk = true;
         let mut sp = sequence_id.descend();
         let ListDataParts {
             elements,
@@ -264,9 +282,6 @@ async fn transpose_list_column(
             }
         }
     }
-    if !saw_chunk {
-        vortex_bail!("ListLayoutStrategy needs at least one chunk");
-    }
     Ok(())
 }
 
@@ -326,6 +341,7 @@ impl Matcher for AnyList {
 #[cfg(test)]
 mod tests {
     use futures::stream;
+    use rstest::rstest;
     use vortex_array::ArrayContext;
     use vortex_array::VortexSessionExecute;
     use vortex_array::array_session;
@@ -440,24 +456,31 @@ mod tests {
         Ok(())
     }
 
+    #[rstest]
+    #[case::non_nullable(false)]
+    #[case::nullable(true)]
     #[tokio::test]
-    async fn empty_stream_errors() {
+    async fn empty_stream_preserves_dtype_and_row_count(
+        #[case] nullable: bool,
+    ) -> VortexResult<()> {
         let segments = Arc::new(TestSegments::default());
         let (_, eof) = SequenceId::root().split();
         let empty = stream::empty::<VortexResult<(SequenceId, ArrayRef)>>().boxed();
-        let stream = SequentialStreamAdapter::new(i32_list_dtype(false), empty).sendable();
+        let stream = SequentialStreamAdapter::new(i32_list_dtype(nullable), empty).sendable();
         let session = layout_test_session();
 
-        let res = flat_list_strategy()
+        let layout = flat_list_strategy()
             .write_stream(
                 ArrayContext::empty().into(),
                 segments,
                 stream,
                 eof,
                 &session,
             )
-            .await;
-        assert!(res.is_err())
+            .await?;
+        assert_eq!(layout.dtype(), &i32_list_dtype(nullable));
+        assert_eq!(layout.row_count(), 0);
+        Ok(())
     }
 
     #[tokio::test]