Skip materializing splitless chunk children during split collection (#9271)

## What changes are included in this PR?

Split collection previously built a full Layout + LayoutReader for every
chunk of every column just so flat chunks could re-register the
chunk-end boundary the parent already knows from its chunk offsets.

- `VTable::registers_interior_splits()` (default true; flat returns
false) lets a layout declare that its readers only ever push the end of
the requested range, forwarded through LayoutVTablePlugin and DynLayout.
- `LayoutChildren::child_has_no_interior_splits(idx)` answers that
without materializing the child: owned children ask the vtable, viewed
children resolve only the flatbuffer encoding tag against the registry,
with a single-slot memo since siblings almost always share one encoding.
- `ChunkedReader` caches a lazy per-chunk classification; when no chunk
has interior splits it bulk-extends boundaries straight from
chunk_offsets.
- `RowSplits` drops consecutive identical ascending runs (columns with
aligned chunk boundaries) and skips the final sort when a single run
survives. StructReader pushes its end boundary after the field walk so
the common case stays one ascending run.
- `LazyReaderChildren::new_uniform` avoids a DType and name clone per
chunk at reader construction.

Split-collection bench medians (64 columns x 256 chunks): cold 916us ->
181us, warm 247us -> 13us; with fully misaligned columns cold 1223us ->
321us, warm 387us -> 146us.

---------

Signed-off-by: Adam Gutglick <adam@spiraldb.com>
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

Author: AdamGS

.github/workflows/codspeed.yml

@@ -49,7 +49,7 @@ jobs:
     strategy:
       matrix:
         include:
-          - { shard: 1, name: "Core foundation", packages: "vortex-buffer vortex-error vortex-mask vortex-compute" }
+          - { shard: 1, name: "Core foundation", packages: "vortex-buffer vortex-error vortex-mask vortex-compute vortex-file" }
           - { shard: 2, name: "Arrays", packages: "vortex-array", features: "--features _test-harness" }
           - { shard: 3, name: "Main library", packages: "vortex" }
           - { shard: 4, name: "Encodings 1", packages: "vortex-alp vortex-bytebool vortex-datetime-parts" }

Cargo.lock

@@ -61,6 +61,7 @@ vortex-zigzag = { workspace = true }
 vortex-zstd = { workspace = true, optional = true }
 
 [dev-dependencies]
+divan = { workspace = true }
 rstest = { workspace = true }
 tokio = { workspace = true, features = ["full"] }
 vortex-array = { workspace = true, features = ["_test-harness"] }
@@ -71,6 +72,10 @@ vortex-scan = { workspace = true }
 [lints]
 workspace = true
 
+[[bench]]
+name = "split_collection"
+harness = false
+
 [features]
 object_store = ["dep:object_store", "vortex-io/object_store", "tokio"]
 tokio = [

vortex-file/Cargo.toml

@@ -0,0 +1,257 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+//! Benchmarks scan split collection (`SplitBy::Layout`) over written files.
+//!
+//! The default write strategy produces `struct -> zoned -> chunked -> flat` per column, so
+//! split collection walks every chunk of every column. `cold` builds a fresh reader tree per
+//! iteration (as a first scan over a file would); `warm` reuses the reader tree so lazily
+//! cached child readers persist across iterations. `cold_misaligned` scans a file whose
+//! columns share no interior chunk boundaries, so the split set cannot be collapsed by run
+//! deduplication.
+
+#![expect(clippy::unwrap_used)]
+
+use std::sync::Arc;
+use std::sync::LazyLock;
+
+use divan::Bencher;
+use vortex_array::IntoArray;
+use vortex_array::arrays::ChunkedArray;
+use vortex_array::arrays::StructArray;
+use vortex_array::dtype::Field;
+use vortex_array::dtype::FieldMask;
+use vortex_array::session::ArraySessionExt;
+use vortex_buffer::Buffer;
+use vortex_buffer::ByteBufferMut;
+use vortex_edition::Edition;
+use vortex_edition::EditionId;
+use vortex_edition::EditionInclusion;
+use vortex_edition::EditionSessionExt;
+use vortex_file::OpenOptionsSessionExt;
+use vortex_file::VortexFile;
+use vortex_file::WriteOptionsSessionExt;
+use vortex_io::session::RuntimeSession;
+use vortex_io::session::RuntimeSessionExt;
+use vortex_layout::layouts::chunked::writer::ChunkedLayoutStrategy;
+use vortex_layout::layouts::flat::writer::FlatLayoutStrategy;
+use vortex_layout::layouts::repartition::RepartitionStrategy;
+use vortex_layout::layouts::repartition::RepartitionWriterOptions;
+use vortex_layout::scan::split_by::SplitBy;
+use vortex_layout::session::LayoutSession;
+use vortex_session::VortexSession;
+use vortex_utils::aliases::hash_map::HashMap;
+
+fn main() {
+    divan::main();
+}
+
+const ROWS_PER_CHUNK: usize = 1024;
+
+/// (columns, chunks) configurations.
+const CONFIGS: &[(usize, usize)] = &[(64, 256)];
+
+static RUNTIME: LazyLock<tokio::runtime::Runtime> = LazyLock::new(|| {
+    tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()
+        .unwrap()
+});
+
+static SESSION: LazyLock<VortexSession> = LazyLock::new(|| {
+    let _guard = RUNTIME.enter();
+    let session = vortex_array::array_session()
+        .with::<LayoutSession>()
+        .with::<RuntimeSession>()
+        .with_tokio();
+    vortex_file::register_default_encodings(&session);
+    enable_all_registered_array_encodings(&session);
+    session
+});
+
+const BENCH_EDITION: EditionId = EditionId::new("bench", 2026, 8, 0);
+
+fn enable_all_registered_array_encodings(session: &VortexSession) {
+    let editions = session.editions();
+    editions
+        .declare_edition(Edition {
+            id: BENCH_EDITION,
+            min_vortex_version: None,
+        })
+        .unwrap();
+    let ids = session
+        .arrays()
+        .registry()
+        .read(|map| map.keys().copied().collect::<Vec<_>>());
+    for id in ids {
+        editions
+            .declare_inclusion(EditionInclusion::new(&id, BENCH_EDITION))
+            .unwrap();
+    }
+    session.enable_edition(BENCH_EDITION).unwrap();
+}
+
+fn make_file(columns: usize, chunks: usize) -> VortexFile {
+    let field_names = (0..columns).map(|c| format!("col_{c}")).collect::<Vec<_>>();
+    let struct_chunks = (0..chunks)
+        .map(|chunk| {
+            let fields = field_names
+                .iter()
+                .map(|name| {
+                    let start = (chunk * ROWS_PER_CHUNK) as i64;
+                    let values =
+                        Buffer::from_iter(start..start + ROWS_PER_CHUNK as i64).into_array();
+                    (name.as_str(), values)
+                })
+                .collect::<Vec<_>>();
+            StructArray::from_fields(&fields).unwrap().into_array()
+        })
+        .collect::<Vec<_>>();
+    let array = ChunkedArray::from_iter(struct_chunks).into_array();
+
+    let strategy = vortex_file::WriteStrategyBuilder::default()
+        .with_row_block_size(ROWS_PER_CHUNK)
+        .with_data_block_target_bytes(None)
+        .build();
+
+    let mut buf = ByteBufferMut::empty();
+    RUNTIME
+        .block_on(
+            SESSION
+                .write_options()
+                .with_strategy(strategy)
+                .write(&mut buf, array.to_array_stream()),
+        )
+        .unwrap();
+
+    SESSION.open_options().open_buffer(buf).unwrap()
+}
+
+static FILES: LazyLock<HashMap<(usize, usize), VortexFile>> = LazyLock::new(|| {
+    CONFIGS
+        .iter()
+        .map(|&(columns, chunks)| ((columns, chunks), make_file(columns, chunks)))
+        .collect()
+});
+
+/// (columns, average chunks per column) for the misaligned files. A single column cannot be
+/// misaligned, so only multi-column configs are used.
+const MISALIGNED_CONFIGS: &[(usize, usize)] = &[(64, 256)];
+
+/// Per-column repartition block length: all distinct, so no two columns share interior chunk
+/// boundaries. Mirrors real files where byte-size coalescing gives each column its own chunking.
+fn misaligned_block_len(column: usize) -> usize {
+    384 + column * 8
+}
+
+/// Like [`make_file`], but each column is chunked at a different row granularity so chunk
+/// boundaries never align across columns: run deduplication in split collection cannot collapse
+/// them, exercising the sort fallback.
+fn make_misaligned_file(columns: usize, chunks: usize) -> VortexFile {
+    let mean_block_len = (0..columns).map(misaligned_block_len).sum::<usize>() / columns.max(1);
+    let rows = chunks * mean_block_len;
+
+    let fields = (0..columns)
+        .map(|c| {
+            let values = Buffer::from_iter(0..rows as i64).into_array();
+            (format!("col_{c}"), values)
+        })
+        .collect::<Vec<_>>();
+    let array = StructArray::from_fields(
+        &fields
+            .iter()
+            .map(|(name, values)| (name.as_str(), values.clone()))
+            .collect::<Vec<_>>(),
+    )
+    .unwrap()
+    .into_array();
+
+    let mut strategy = vortex_file::WriteStrategyBuilder::default();
+    for (c, (name, _)) in fields.iter().enumerate() {
+        let field_strategy = RepartitionStrategy::new(
+            ChunkedLayoutStrategy::new(FlatLayoutStrategy::default()),
+            RepartitionWriterOptions {
+                block_size_minimum: 0,
+                block_len_multiple: misaligned_block_len(c),
+                block_size_target: None,
+                canonicalize: false,
+            },
+        );
+        strategy = strategy.with_field_writer(Field::from(name.as_str()), Arc::new(field_strategy));
+    }
+
+    let mut buf = ByteBufferMut::empty();
+    RUNTIME
+        .block_on(
+            SESSION
+                .write_options()
+                .with_strategy(strategy.build())
+                .write(&mut buf, array.to_array_stream()),
+        )
+        .unwrap();
+
+    SESSION.open_options().open_buffer(buf).unwrap()
+}
+
+static MISALIGNED_FILES: LazyLock<HashMap<(usize, usize), VortexFile>> = LazyLock::new(|| {
+    MISALIGNED_CONFIGS
+        .iter()
+        .map(|&(columns, chunks)| ((columns, chunks), make_misaligned_file(columns, chunks)))
+        .collect()
+});
+
+fn collect_splits(file: &VortexFile) -> Vec<u64> {
+    let reader = file.layout_reader().unwrap();
+    SplitBy::Layout
+        .splits(reader.as_ref(), &(0..file.row_count()), &[FieldMask::All])
+        .unwrap()
+}
+
+/// Builds a fresh reader tree per iteration, so split collection pays child reader
+/// construction for every chunk of every column, like the first scan over a file.
+#[divan::bench(args = CONFIGS)]
+fn cold(bencher: Bencher, config: &(usize, usize)) {
+    let file = &FILES[config];
+    // Sanity-check the written chunk structure once per config.
+    let n_splits = collect_splits(file).len();
+    assert!(
+        n_splits > config.1 / 2,
+        "expected roughly one split per chunk, got {n_splits} splits for {} chunks",
+        config.1
+    );
+
+    bencher.bench(|| collect_splits(file));
+}
+
+/// Reuses the reader tree across iterations, so lazily constructed child readers are cached
+/// and split collection measures only the layout walk.
+#[divan::bench(args = CONFIGS)]
+fn warm(bencher: Bencher, config: &(usize, usize)) {
+    let file = &FILES[config];
+    let reader = file.layout_reader().unwrap();
+    let row_count = file.row_count();
+
+    bencher.bench(|| {
+        SplitBy::Layout
+            .splits(reader.as_ref(), &(0..row_count), &[FieldMask::All])
+            .unwrap()
+    });
+}
+
+/// Like `cold`, but over a file whose columns share no interior chunk boundaries, so the split
+/// set cannot be collapsed by run deduplication and must be sorted.
+#[divan::bench(args = MISALIGNED_CONFIGS)]
+fn cold_misaligned(bencher: Bencher, config: &(usize, usize)) {
+    let file = &MISALIGNED_FILES[config];
+    // Sanity-check the misalignment once per config: boundaries should be mostly distinct
+    // across columns, i.e. roughly columns × chunks in total.
+    let n_splits = collect_splits(file).len();
+    assert!(
+        n_splits > config.0 * config.1 / 2,
+        "expected mostly-distinct boundaries, got {n_splits} splits for {} columns x {} chunks",
+        config.0,
+        config.1,
+    );
+
+    bencher.bench(|| collect_splits(file));
+}

vortex-file/benches/split_collection.rs

@@ -35,6 +35,16 @@ pub trait LayoutChildren: 'static + Send + Sync {
     fn child_row_count(&self, idx: usize) -> u64;
 
     fn nchildren(&self) -> usize;
+
+    /// Returns `true` if the child at `idx` is known — without materializing it — to be
+    /// indivisible: it registers no split boundaries strictly inside its row range (see
+    /// [`VTable::is_indivisible`](crate::VTable::is_indivisible)).
+    ///
+    /// Implementations must conservatively return `false` when answering would require
+    /// materializing the child.
+    fn child_is_indivisible(&self, _idx: usize) -> bool {
+        false
+    }
 }
 
 impl Debug for dyn LayoutChildren {
@@ -61,6 +71,10 @@ impl LayoutChildren for Arc<dyn LayoutChildren> {
     fn nchildren(&self) -> usize {
         self.as_ref().nchildren()
     }
+
+    fn child_is_indivisible(&self, idx: usize) -> bool {
+        self.as_ref().child_is_indivisible(idx)
+    }
 }
 
 /// An implementation of [`LayoutChildren`] for in-memory owned children.
@@ -102,6 +116,10 @@ impl LayoutChildren for OwnedLayoutChildren {
     fn nchildren(&self) -> usize {
         self.0.len()
     }
+
+    fn child_is_indivisible(&self, idx: usize) -> bool {
+        self.0[idx].dyn_is_indivisible()
+    }
 }
 
 #[derive(Clone)]
@@ -114,6 +132,9 @@ pub(crate) struct ViewedLayoutChildren {
     allow_unknown: bool,
     session: VortexSession,
     cache: Arc<[OnceCell<LayoutRef>]>,
+    /// Per-child answers to [`LayoutChildren::child_is_indivisible`], precomputed at
+    /// construction from the flatbuffer encoding tags alone.
+    indivisible: Arc<[bool]>,
 }
 
 impl ViewedLayoutChildren {
@@ -132,11 +153,23 @@ impl ViewedLayoutChildren {
         session: VortexSession,
     ) -> Self {
         // SAFETY: guaranteed by caller
-        let nchildren = unsafe { fbl::Layout::follow(flatbuffer.as_ref(), flatbuffer_loc) }
+        let fb_children = unsafe { fbl::Layout::follow(flatbuffer.as_ref(), flatbuffer_loc) }
             .children()
-            .unwrap_or_default()
-            .len();
-        let cache = vec![OnceCell::new(); nchildren].into_boxed_slice().into();
+            .unwrap_or_default();
+        let cache = vec![OnceCell::new(); fb_children.len()]
+            .into_boxed_slice()
+            .into();
+        // Unknown encodings are conservatively not indivisible so callers fall back to
+        // materializing the child.
+        let indivisible = fb_children
+            .iter()
+            .map(|child| {
+                layout_read_ctx
+                    .resolve(child.encoding())
+                    .and_then(|encoding_id| layouts.get(&encoding_id))
+                    .is_some_and(|encoding| encoding.is_indivisible())
+            })
+            .collect::<Arc<[bool]>>();
         Self {
             flatbuffer,
             flatbuffer_loc,
@@ -146,6 +179,7 @@ impl ViewedLayoutChildren {
             allow_unknown,
             session,
             cache,
+            indivisible,
         }
     }
 
@@ -270,4 +304,8 @@ impl LayoutChildren for ViewedLayoutChildren {
     fn nchildren(&self) -> usize {
         self.cache.len()
     }
+
+    fn child_is_indivisible(&self, idx: usize) -> bool {
+        self.indivisible.get(idx).copied().unwrap_or(false)
+    }
 }