Cache immutable whole-subtree support checks for sparse V2 scans

Sparse identity scans must validate complete list element and dictionary
value subtrees, even when the selected root rows touch few chunks. Repeating
these checks on each scan made the 64 KiB list layout disproportionately
expensive. Cache only whole-subtree answers within ScanFile preparation;
selection-dependent answers remain uncached. Retain keyed layout Arcs to
prevent address reuse, and bypass the cache when the root layout changes.

The prior 32 MiB estimated list-work admission cap and independently
scheduled projection pieces are unchanged. No segment data is retained by
this metadata cache. Errors are not cached; unsupported answers are.

On warm local disk, 100 selected rows, three seeds, two alternating rounds
of 64 queries (discard eight), full Arrow decode/drop:
- 64 KiB clustered cached, 30 cores: 0.2053 -> 0.1525 ms (-25.7%).
- 64 KiB clustered cached, one core: 0.2716 -> 0.2220 ms (-18.3%).
- 64 KiB uniform cached, 30 cores: 0.8233 -> 0.7558 ms (-8.2%).
- No aggregate regression above 3% across 48 cases versus previous V2.
- Twelve longer reruns used four rounds of 128 queries, discard sixteen;
  the largest confirmed slowdown was 2.6% (nested structs, one core).

This does not resolve all V1 gaps: 13/48 initial cases remain above +5%,
including reopened clustered lists and parallel uniform feature vectors.
Reopening receives little benefit because the preparation cache is fresh.
Experiments with smaller clustered split targets and a direct contiguous
list-element source did not improve end-to-end timing and were discarded.

Validation: 207 targeted scan::v2 unit tests; 60 real-file Arrow comparisons
against V1 full-scan-plus-take across empty, singleton, clustered, sparse,
and all-row selections at one and 30 cores; pinned rustfmt; diff --check.
No workspace-wide checks were run. Local experiment reports and rejected
patches are under the list-layout-experiment directory in the scan bench
workspace; the CI branch runs the comparable matrix on a dedicated runner.

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

Author: joseph-isaacs

vortex-layout/src/scan/v2/mod.rs

@@ -32,6 +32,7 @@ use futures::TryStreamExt;
 use futures::stream;
 use futures::stream::BoxStream;
 use once_cell::sync::OnceCell;
+use parking_lot::Mutex;
 use vortex_array::ArrayRef;
 use vortex_error::VortexResult;
 use vortex_error::vortex_err;
@@ -43,6 +44,7 @@ use vortex_scan::planning::selection::SelectedRows;
 use vortex_scan::planning::selection::SelectionPlanner;
 use vortex_scan::selection::Selection;
 use vortex_session::VortexSession;
+use vortex_utils::aliases::hash_map::HashMap;
 use vortex_utils::parallelism::get_available_parallelism;
 
 use crate::LayoutRef;
@@ -103,11 +105,15 @@ struct FilePreparation {
     segment_bytes: u64,
     projection_supported: OnceCell<bool>,
     identity_supported: OnceCell<bool>,
+    whole_identity_support: WholeSupport,
     root: OnceCell<PlanRef>,
     ranges: OnceCell<SegmentRanges>,
     values: OnceCell<ValueCache>,
 }
 
+// Retain each keyed layout so its address cannot be reused for a different support answer.
+type WholeSupport = Mutex<HashMap<usize, (LayoutRef, bool)>>;
+
 impl ScanFile {
     /// Creates a file description without retaining preparation between scans.
     pub fn new(
@@ -123,9 +129,9 @@ impl ScanFile {
         }
     }
 
-    /// Shares immutable layout plans, segment locations, and whole dictionary values between
-    /// this file's scans. Only identity projections retain dictionary values; selections and
-    /// expression results remain owned by each scan.
+    /// Shares immutable layout support checks and plans, segment locations, and whole dictionary
+    /// values between this file's scans. Only identity projections retain dictionary values;
+    /// selections and expression results remain owned by each scan.
     pub fn with_caching(mut self) -> Self {
         self.preparation = Some(Arc::new(FilePreparation {
             layout: Arc::clone(&self.layout),
@@ -138,6 +144,7 @@ impl ScanFile {
                 .sum(),
             projection_supported: OnceCell::new(),
             identity_supported: OnceCell::new(),
+            whole_identity_support: Mutex::default(),
             root: OnceCell::new(),
             ranges: OnceCell::new(),
             values: OnceCell::new(),
@@ -182,7 +189,13 @@ impl ScanFile {
             let rows = rows.cloned().unwrap_or(0..self.layout.row_count());
             if let Some(indices) = sparse_indices(selection, &rows) {
                 // A selection-specific answer cannot be cached as support for the whole file.
-                return supports_selected_layout(&self.layout, &indices, 0);
+                return supports_selected_layout(
+                    &self.layout,
+                    &indices,
+                    0,
+                    self.layout_preparation()
+                        .map(|cache| &cache.whole_identity_support),
+                );
             }
         }
         self.supports(identity)
@@ -327,6 +340,7 @@ fn supports_selected_layout(
     layout: &LayoutRef,
     indices: &[u64],
     offset: u64,
+    whole: Option<&WholeSupport>,
 ) -> VortexResult<bool> {
     if layout.is::<Flat>() {
         return Ok(true);
@@ -346,7 +360,12 @@ fn supports_selected_layout(
             let child = layout
                 .slot(chunk)?
                 .ok_or_else(|| vortex_err!("Missing selected chunk"))?;
-            if !supports_selected_layout(&child, &indices[first..last], offset + offsets[chunk])? {
+            if !supports_selected_layout(
+                &child,
+                &indices[first..last],
+                offset + offsets[chunk],
+                whole,
+            )? {
                 return Ok(false);
             }
             first = last;
@@ -355,7 +374,7 @@ fn supports_selected_layout(
     }
     if layout.is::<Zoned>() || layout.is::<LegacyStats>() {
         return match layout.slot(0)? {
-            Some(data) => supports_selected_layout(&data, indices, offset),
+            Some(data) => supports_selected_layout(&data, indices, offset, whole),
             None => Ok(false),
         };
     }
@@ -366,16 +385,15 @@ fn supports_selected_layout(
         let codes = layout
             .slot(1)?
             .ok_or_else(|| vortex_err!("Missing dictionary codes"))?;
-        return Ok(
-            supports_layout(&values, true)? && supports_selected_layout(&codes, indices, offset)?
-        );
+        return Ok(supports_whole_identity(&values, whole)?
+            && supports_selected_layout(&codes, indices, offset, whole)?);
     }
     if layout.is::<List>() {
-        return supports_layout(layout, true);
+        return supports_whole_identity(layout, whole);
     }
     if layout.is::<Struct>() {
         for child in layout.children()? {
-            if !supports_selected_layout(&child, indices, offset)? {
+            if !supports_selected_layout(&child, indices, offset, whole)? {
                 return Ok(false);
             }
         }
@@ -384,6 +402,20 @@ fn supports_selected_layout(
     Ok(false)
 }
 
+/// Unlike a selected-chunk check, this answer is independent of the selected rows.
+fn supports_whole_identity(layout: &LayoutRef, cache: Option<&WholeSupport>) -> VortexResult<bool> {
+    let Some(cache) = cache else {
+        return supports_layout(layout, true);
+    };
+    let key = Arc::as_ptr(layout).cast::<()>() as usize;
+    if let Some((_, supported)) = cache.lock().get(&key) {
+        return Ok(*supported);
+    }
+    let supported = supports_layout(layout, true)?;
+    cache.lock().insert(key, (Arc::clone(layout), supported));
+    Ok(supported)
+}
+
 /// A scan planned over its file, ready to run its splits.
 pub(crate) struct PreparedScan<A: 'static + Send> {
     session: VortexSession,

vortex-layout/src/scan/v2/tests.rs

@@ -59,6 +59,7 @@ use crate::layouts::chunked::writer::ChunkedLayoutStrategy;
 use crate::layouts::dict::DictLayout;
 use crate::layouts::flat::writer::FlatLayoutStrategy;
 use crate::layouts::foreign::new_foreign_layout;
+use crate::layouts::list::ListLayout;
 use crate::layouts::list::writer::ListLayoutStrategy;
 use crate::layouts::row_idx::row_idx;
 use crate::layouts::struct_::StructLayout;
@@ -488,6 +489,75 @@ impl LayoutChildren for CountedLayoutChildren {
     }
 }
 
+#[rstest]
+#[tokio::test]
+async fn caches_whole_subtree_support_without_reusing_replaced_layouts(
+    #[values(false, true)] caching: bool,
+    #[values(false, true)] unsupported: bool,
+    #[values(false, true)] list: bool,
+) -> VortexResult<()> {
+    let session = new_session().with_tokio();
+    let (segments, layout) = write(&session, table().as_ref()).await?;
+    let ints = layout.slot(1)?.ok_or_else(|| vortex_err!("missing ints"))?;
+    let mut children = ints.children()?;
+    if unsupported {
+        children[1] = new_foreign_layout(
+            LayoutId::new("test.unsupported"),
+            ints.dtype().clone(),
+            CHUNK_ROWS as u64,
+            Vec::new(),
+            Vec::new(),
+            Vec::new(),
+        );
+    }
+    let values = ints.as_::<Chunked>().with_children(children);
+    let visited = Arc::new(AtomicUsize::new(0));
+    let values = ChunkedLayout::new(
+        values.row_count(),
+        values.dtype().clone(),
+        Arc::new(CountedLayoutChildren {
+            inner: Arc::clone(values.children()),
+            materialized: Arc::clone(&visited),
+            encoding_hints: false,
+        }),
+    )
+    .into_layout();
+    let make_layout = |values: LayoutRef, ints: LayoutRef| {
+        if list {
+            ListLayout::new(
+                DType::List(Arc::new(values.dtype().clone()), false.into()),
+                values,
+                ints,
+                None,
+            )
+            .into_layout()
+        } else {
+            DictLayout::new(values, ints).into_layout()
+        }
+    };
+    let layout = make_layout(Arc::clone(&values), Arc::clone(&ints));
+    let mut file = scan_file(&segments, &layout)?;
+    if caching {
+        file = file.with_caching();
+    }
+    visited.store(0, Ordering::Relaxed);
+    let selection = Selection::IncludeRoaring([0, 3000].into_iter().collect());
+    assert_eq!(file.supports_scan(true, &selection, None)?, !unsupported);
+    let first = visited.load(Ordering::Relaxed);
+    assert!(first > 0);
+    assert_eq!(file.supports_scan(true, &selection, None)?, !unsupported);
+    assert_eq!(
+        visited.load(Ordering::Relaxed),
+        first * if caching { 1 } else { 2 }
+    );
+
+    let before = visited.load(Ordering::Relaxed);
+    file.layout = make_layout(values, ints);
+    assert_eq!(file.supports_scan(true, &selection, None)?, !unsupported);
+    assert_eq!(visited.load(Ordering::Relaxed), before + first);
+    Ok(())
+}
+
 #[rstest]
 #[tokio::test]
 async fn support_checks_flat_tags_without_materializing_chunks(