Reuse validated chunk boundaries during lowering

Copy the already validated chunk start offsets into the existing Concat
Arc instead of repeatedly asking the layout for each child's row count.
Keep V1's offset storage and the base planner/scheduler unchanged. This is
a measured deviation from PR #10540, isolated for comparison with 5231b70.

Four regression cases cover regular, empty, absent, and zero-row chunks;
three fail before the change. No child layouts are materialized to lower
the boundaries. Production change is +7/-10 lines in two files.

Local four-round, five-second standard comparisons show clustered feature
vectors improving 8.0% cached / 3.5% reopen on one core and 17.1% / 4.9%
on 30 cores. The initial taxi cached +6.4% reduced to +0.8% in the longer
run. Nested-struct reopen is still +4.9%; identical-binary controls vary
up to 7.8%, so that result is unresolved rather than dismissed as noise.

Sixteen full-Arrow list cases initially show at most +2.01% slowdown;
longer list confirmation and isolated parent CI remain pending. Repeated
memory checks show no increase in taxi maximum live allocation, and median
requested read counts/bytes are unchanged. No all-cases V1 win is claimed.

Validation: 618 layout tests, all-target/all-feature layout Clippy with
-D warnings, pinned formatting, whitespace, release benchmark builds,
and 72 real-file V1 full-scan-plus-take correctness comparisons passed.
No workspace-wide checks or doctests were run. Keep separate from the
draft PR until the exact-parent CI comparison is assessed.

Signed-off-by: Joe Isaacs <joe.isaacs@live.co.uk>

Author: joseph-isaacs

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

@@ -125,6 +125,11 @@ impl Layout<Chunked> {
             OwnedLayoutChildren::layout_children(children),
         )
     }
+
+    /// Validated chunk boundaries, including the terminal row count.
+    pub(crate) fn chunk_offsets(&self) -> &[u64] {
+        &self.chunk_offsets
+    }
 }
 
 fn chunk_offsets(children: &dyn LayoutChildren) -> VortexResult<Vec<u64>> {

vortex-layout/src/plan/physical/lower.rs

@@ -83,21 +83,13 @@ fn lower_flat(layout: &FlatLayout) -> SegmentScanPlan {
 }
 
 fn lower_chunked(layout: &ChunkedLayout) -> VortexResult<ConcatPlan> {
-    let mut row_offsets = Vec::with_capacity(layout.nchildren());
-    let mut row_count = 0u64;
-    for index in 0..layout.nchildren() {
-        row_offsets.push(row_count);
-        row_count = row_count
-            .checked_add(layout.child_row_count(index))
-            .ok_or_else(|| vortex_err!("Chunked row count overflow"))?;
-    }
     // SAFETY: Chunked layout construction validates that every child has the parent dtype and
-    // that their row counts sum to the parent; offsets were computed with checked addition above.
+    // that these checked cumulative offsets end at the parent's row count.
     Ok(unsafe {
         ConcatPlan::from_children_unchecked(
             layout.dtype().clone(),
             layout.row_count(),
-            row_offsets.into(),
+            layout.chunk_offsets()[..layout.nchildren()].into(),
             lazy_children(layout.to_layout(), (0..layout.nchildren()).collect()),
         )
     })

vortex-layout/src/plan/physical/tests.rs

@@ -4,7 +4,10 @@
 use std::fmt;
 use std::num::NonZeroUsize;
 use std::sync::Arc;
+use std::sync::atomic::AtomicUsize;
+use std::sync::atomic::Ordering;
 
+use rstest::rstest;
 use vortex_array::IntoArray;
 use vortex_array::VortexSessionExecute;
 use vortex_array::aggregate_fn::AggregateFnRef;
@@ -43,6 +46,7 @@ use vortex_session::registry::ReadContext;
 
 use super::*;
 use crate::LayoutBuildContext;
+use crate::LayoutChildren;
 use crate::LayoutEncoding;
 use crate::LayoutRef;
 use crate::OwnedLayoutChildren;
@@ -150,6 +154,60 @@ fn chunked_plan_exposes_chunks() -> VortexResult<()> {
     Ok(())
 }
 
+#[derive(Clone)]
+struct CountedChunkRows {
+    rows: Arc<[u64]>,
+    calls: Arc<AtomicUsize>,
+}
+
+impl LayoutChildren for CountedChunkRows {
+    fn to_arc(&self) -> Arc<dyn LayoutChildren> {
+        Arc::new(self.clone())
+    }
+
+    fn child(&self, _idx: usize, _dtype: &DType) -> VortexResult<LayoutRef> {
+        Err(vortex_err!("Lowering must leave chunk children lazy"))
+    }
+
+    fn child_row_count(&self, idx: usize) -> u64 {
+        self.calls.fetch_add(1, Ordering::Relaxed);
+        self.rows[idx]
+    }
+
+    fn nchildren(&self) -> usize {
+        self.rows.len()
+    }
+}
+
+#[rstest]
+#[case::normal(vec![3, 2], vec![0, 3])]
+#[case::empty_chunks(vec![0, 2, 0, 3, 0], vec![0, 0, 2, 2, 5])]
+#[case::empty(vec![], vec![])]
+#[case::zero_rows(vec![0], vec![0])]
+fn lowering_reuses_validated_chunk_boundaries(
+    #[case] rows: Vec<u64>,
+    #[case] offsets: Vec<u64>,
+) -> VortexResult<()> {
+    let calls = Arc::new(AtomicUsize::new(0));
+    let row_count = rows.iter().sum();
+    let layout = ChunkedLayout::new(
+        row_count,
+        DType::Null,
+        Arc::new(CountedChunkRows {
+            rows: rows.into(),
+            calls: Arc::clone(&calls),
+        }),
+    )
+    .into_layout();
+    calls.store(0, Ordering::Relaxed);
+    let plan = lower(&layout)?;
+    assert_eq!(plan.as_::<Concat>().row_offsets(), offsets);
+    assert_eq!(plan.row_count(), row_count);
+    assert_eq!(plan.child_count(), offsets.len());
+    assert_eq!(calls.load(Ordering::Relaxed), 0);
+    Ok(())
+}
+
 #[test]
 fn chunked_plan_lowers_each_chunk_on_access() -> VortexResult<()> {
     let dtype = primitive(PType::I32, Nullability::NonNullable);