list: top-level streaming decomposition behind VORTEX_EXPERIMENTAL_LIST_LAYOUT

Rework ListLayoutStrategy into a streaming transpose driven by the TableStrategy
dispatcher (global u64 offsets, memory-bounded), add bounded element reads and
is_null pruning routing in ListReader, and gate the whole thing off by default
behind the VORTEX_EXPERIMENTAL_LIST_LAYOUT env var / with_list_layout().

Signed-off-by: Matt Katz <mhkatz97@gmail.com>

Author: mhk197

vortex-file/src/strategy.rs

@@ -50,10 +50,10 @@ use vortex_layout::layouts::compressed::CompressingStrategy;
 use vortex_layout::layouts::compressed::CompressorPlugin;
 use vortex_layout::layouts::dict::writer::DictStrategy;
 use vortex_layout::layouts::flat::writer::FlatLayoutStrategy;
-use vortex_layout::layouts::list::writer::ListLayoutStrategy;
 use vortex_layout::layouts::repartition::RepartitionStrategy;
 use vortex_layout::layouts::repartition::RepartitionWriterOptions;
 use vortex_layout::layouts::table::TableStrategy;
+use vortex_layout::layouts::table::use_experimental_list_layout;
 use vortex_layout::layouts::zoned::writer::ZonedLayoutOptions;
 use vortex_layout::layouts::zoned::writer::ZonedStrategy;
 #[cfg(feature = "unstable_encodings")]
@@ -158,6 +158,10 @@ pub struct WriteStrategyBuilder {
     allow_encodings: Option<HashSet<ArrayId>>,
     flat_strategy: Option<Arc<dyn LayoutStrategy>>,
     probe_compressor: Option<Arc<dyn CompressorPlugin>>,
+    /// Force list-column decomposition on, overriding the
+    /// [`use_experimental_list_layout`](vortex_layout::layouts::table::use_experimental_list_layout)
+    /// env-var default of off.
+    list_layout: bool,
 }
 
 impl Default for WriteStrategyBuilder {
@@ -171,6 +175,7 @@ impl Default for WriteStrategyBuilder {
             allow_encodings: Some(ALLOWED_ENCODINGS.clone()),
             flat_strategy: None,
             probe_compressor: None,
+            list_layout: false,
         }
     }
 }
@@ -185,6 +190,14 @@ impl WriteStrategyBuilder {
         self
     }
 
+    /// Enable list-column decomposition, overriding the env-var default of off. When enabled, list
+    /// columns are written as a `ListLayout` (independently compressed/chunked
+    /// elements/offsets/validity) rather than as flat arrays.
+    pub fn with_list_layout(mut self) -> Self {
+        self.list_layout = true;
+        self
+    }
+
     /// Override the write layout for a specific field somewhere in the nested schema tree.
     ///
     /// The field path is matched after the root struct is split into columns. This is useful when a
@@ -251,20 +264,10 @@ impl WriteStrategyBuilder {
             Arc::new(FlatLayoutStrategy::default())
         };
 
-        // 7. for each chunk create a layout. List-typed chunks route through
-        // `ListLayoutStrategy` (separately-addressable elements/offsets/validity sub-layouts;
-        // non-list chunks fall through its built-in fallback to `flat`).
-        let leaf: Arc<dyn LayoutStrategy> = Arc::new(
-            // Thread the configured `flat` (which carries `allow_encodings` / any custom flat
-            // override) through every child.
-            ListLayoutStrategy::default()
-                .with_elements(Arc::clone(&flat))
-                .with_offsets(Arc::clone(&flat))
-                .with_validity(Arc::clone(&flat))
-                .with_fallback(Arc::clone(&flat)),
-        );
-
-        let chunked = ChunkedLayoutStrategy::new(leaf);
+        // 7. for each chunk create a flat layout. List columns are decomposed above this point by
+        // the `TableStrategy` dispatcher (into independently-compressed elements/offsets/validity
+        // sub-columns), so the per-chunk leaf only ever sees flat, non-list chunks.
+        let chunked = ChunkedLayoutStrategy::new(Arc::clone(&flat));
         // 6. buffer chunks so they end up with closer segment ids physically
         let buffered = BufferedStrategy::new(chunked, 2 * ONE_MEG); // 2MB
 
@@ -348,8 +351,14 @@ impl WriteStrategyBuilder {
         let validity_strategy = CollectStrategy::new(compress_then_flat);
 
         // Take any field overrides from the builder and apply them to the final strategy.
-        let table_strategy = TableStrategy::new(Arc::new(validity_strategy), Arc::new(repartition))
-            .with_field_writers(self.field_writers);
+        let mut table_strategy =
+            TableStrategy::new(Arc::new(validity_strategy), Arc::new(repartition))
+                .with_field_writers(self.field_writers);
+        // List decomposition is experimental: enabled by an explicit builder opt-in or the
+        // `VORTEX_EXPERIMENTAL_LIST_LAYOUT` env var; off otherwise.
+        if self.list_layout || use_experimental_list_layout() {
+            table_strategy = table_strategy.with_list_layout();
+        }
 
         Arc::new(table_strategy)
     }

vortex-file/src/tests.rs

@@ -15,6 +15,7 @@ use vortex_array::ArrayRef;
 use vortex_array::IntoArray;
 use vortex_array::VortexSessionExecute;
 use vortex_array::array_session;
+use vortex_array::arrays::BoolArray;
 use vortex_array::arrays::ChunkedArray;
 use vortex_array::arrays::ConstantArray;
 use vortex_array::arrays::DecimalArray;
@@ -1671,6 +1672,75 @@ async fn test_writer_with_complex_types() -> VortexResult<()> {
     Ok(())
 }
 
+/// Write `array` with list decomposition forced on (through the full compress/zone pipeline) and
+/// read the whole thing back.
+async fn write_read_roundtrip(array: ArrayRef) -> VortexResult<ArrayRef> {
+    let strategy = crate::strategy::WriteStrategyBuilder::default()
+        .with_list_layout()
+        .build();
+    let mut buf = ByteBufferMut::empty();
+    SESSION
+        .write_options()
+        .with_strategy(strategy)
+        .write(&mut buf, array.to_array_stream())
+        .await?;
+    SESSION
+        .open_options()
+        .open_buffer(buf)?
+        .scan()?
+        .into_array_stream()?
+        .read_all()
+        .await
+}
+
+/// A `list<list<i32>>` column round-trips through the `TableStrategy` dispatcher, exercising list
+/// decomposition recursing into itself (the outer list's `elements` are themselves lists).
+#[tokio::test]
+#[cfg_attr(miri, ignore)]
+async fn nested_list_of_list_roundtrip() -> VortexResult<()> {
+    let inner = ListArray::try_new(
+        buffer![1i32, 2, 3, 4, 5, 6].into_array(),
+        buffer![0u32, 2, 5, 5, 6].into_array(),
+        Validity::NonNullable,
+    )?
+    .into_array();
+    let outer = ListArray::try_new(
+        inner,
+        buffer![0u32, 2, 4].into_array(),
+        Validity::NonNullable,
+    )?
+    .into_array();
+    let st = StructArray::from_fields(&[("nested", outer)])?.into_array();
+
+    let result = write_read_roundtrip(st.clone()).await?;
+    assert_arrays_eq!(result, st, &mut SESSION.create_execution_ctx());
+    Ok(())
+}
+
+/// A `struct<{ items: list<struct<{a,b}>>? }>` column round-trips, exercising list decomposition
+/// recursing into struct decomposition (list `elements` are structs) plus a nullable list validity
+/// child.
+#[tokio::test]
+#[cfg_attr(miri, ignore)]
+async fn nested_struct_list_struct_roundtrip() -> VortexResult<()> {
+    let inner_struct = StructArray::from_fields(&[
+        ("a", buffer![1i32, 2, 3, 4, 5].into_array()),
+        ("b", buffer![10i32, 20, 30, 40, 50].into_array()),
+    ])?
+    .into_array();
+    let items = ListArray::try_new(
+        inner_struct,
+        buffer![0u32, 2, 5, 5].into_array(),
+        Validity::Array(BoolArray::from_iter([true, false, true]).into_array()),
+    )?
+    .into_array();
+    let st = StructArray::from_fields(&[("items", items)])?.into_array();
+
+    let result = write_read_roundtrip(st.clone()).await?;
+    assert_arrays_eq!(result, st, &mut SESSION.create_execution_ctx());
+    Ok(())
+}
+
 #[tokio::test]
 async fn test_writer_with_statistics() -> VortexResult<()> {
     let array = StructArray::from_fields(&[("numbers", buffer![1u32, 2, 3, 4, 5].into_array())])?

vortex-layout/src/layouts/chunked/mod.rs

@@ -1,7 +1,7 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
-mod reader;
+pub(crate) mod reader;
 pub mod writer;
 
 use std::sync::Arc;