Add planner-owned selection and bounded ListLayout scans to V2

benchmarks/random-access-bench/README.md

@@ -28,6 +28,8 @@ work in each timed iteration. CI drives the full matrix via
 cargo run -p random-access-bench --profile release_debug --features lance
 ```
 
+Vortex measurements always use the planner-driven V2 scan builder.
+
 ## Running against S3
 
 The same benchmark can read its data from an object store instead of local disk. The remote

vortex-bench/src/random_access/take.rs

@@ -29,10 +29,12 @@ use tokio::fs::File;
 use vortex::array::Canonical;
 use vortex::array::IntoArray;
 use vortex::array::VortexSessionExecute;
+use vortex::array::stream::ArrayStreamAdapter;
 use vortex::array::stream::ArrayStreamExt;
 use vortex::buffer::Buffer;
 use vortex::file::OpenOptionsSessionExt;
 use vortex::file::VortexFile;
+use vortex::layout::scan::v2::ScanBuilder;
 use vortex::scan::strict_sorted_buffer::StrictSortedBuffer;
 use vortex::utils::aliases::hash_map::HashMap;
 
@@ -142,11 +144,7 @@ impl VortexRandomAccessor {
         name: impl Into<String>,
         format: Format,
     ) -> anyhow::Result<Self> {
-        let file = SESSION
-            .open_options()
-            .with_layout_reader_cache()
-            .open_path(path.as_ref())
-            .await?;
+        let file = SESSION.open_options().open_path(path.as_ref()).await?;
         Ok(Self {
             name: name.into(),
             format,
@@ -163,7 +161,6 @@ impl VortexRandomAccessor {
     ) -> anyhow::Result<Self> {
         let file = SESSION
             .open_options()
-            .with_layout_reader_cache()
             .open_object_store(remote.store(), remote.key(path)?)
             .await?;
         Ok(Self {
@@ -186,11 +183,10 @@ impl RandomAccessor for VortexRandomAccessor {
 
     async fn take(&self, indices: &[u64]) -> anyhow::Result<RandomAccessorRet> {
         let indices_buf: Buffer<u64> = Buffer::from(indices.to_vec());
-        let array = self
-            .file
-            .scan()?
+        let stream = ScanBuilder::new(SESSION.clone(), self.file.scan_file())
             .with_row_indices(StrictSortedBuffer::try_new(indices_buf)?)
-            .into_array_stream()?
+            .into_stream()?;
+        let array = ArrayStreamAdapter::new(self.file.dtype().clone(), stream)
             .read_all()
             .await?;
 

vortex-layout/benches/pipeline_scan.rs

@@ -10,14 +10,19 @@ use std::hint::black_box;
 use divan::Bencher;
 use mimalloc::MiMalloc;
 use vortex_array::ArrayContext;
+use vortex_array::ArrayRef;
 use vortex_array::IntoArray;
 use vortex_array::arrays::PrimitiveArray;
 use vortex_array::buffer::BufferHandle;
 use vortex_array::dtype::Nullability;
 use vortex_array::dtype::StructFields;
 use vortex_array::serde::SerializeOptions;
 use vortex_buffer::Alignment;
+use vortex_buffer::Buffer;
 use vortex_buffer::ByteBufferMut;
+use vortex_layout::plan::physical::ConcatPlan;
+use vortex_layout::plan::physical::FilterPlan;
+use vortex_layout::plan::physical::ListPackPlan;
 use vortex_layout::plan::physical::PackPlan;
 use vortex_layout::plan::physical::PlanRef;
 use vortex_layout::plan::physical::SegmentScanPlan;
@@ -26,6 +31,7 @@ use vortex_layout::plan::physical::pipeline::Split;
 use vortex_layout::plan::physical::pipeline::Turn;
 use vortex_layout::segments::SegmentId;
 use vortex_layout::session::LayoutSession;
+use vortex_mask::Mask;
 use vortex_session::VortexSession;
 use vortex_session::registry::ReadContext;
 
@@ -124,9 +130,140 @@ fn scan(bencher: Bencher, columns: u32) {
                 Turn::Output(_, array) => {
                     black_box(array);
                 }
+                Turn::Child(_) => unreachable!("benchmark has no asynchronous child work"),
                 Turn::Waiting => unreachable!("every read is delivered immediately"),
                 Turn::Done => break,
             }
         }
     });
 }
+
+struct ListFixture {
+    session: VortexSession,
+    plan: PlanRef,
+    segments: Vec<BufferHandle>,
+    mask: Mask,
+}
+
+impl ListFixture {
+    fn new(chunks: u32, uniform: bool) -> Self {
+        const LISTS: usize = 65_536;
+        const ELEMENTS_PER_LIST: usize = 8;
+        let session = vortex_array::array_session().with::<LayoutSession>();
+        let chunk_len = LISTS * ELEMENTS_PER_LIST / chunks as usize;
+        let elements = Buffer::from_iter(0..chunk_len as u64).into_array();
+        let offsets = Buffer::from_iter((0..=LISTS as u64).map(|row| row * 8)).into_array();
+        let serialize = |array: &ArrayRef| {
+            let context = ArrayContext::empty();
+            let buffers = array
+                .serialize(
+                    &context,
+                    &session,
+                    &SerializeOptions {
+                        offset: 0,
+                        include_padding: true,
+                    },
+                )
+                .unwrap();
+            let mut bytes = ByteBufferMut::empty_aligned(Alignment::new(64));
+            for buffer in buffers {
+                bytes.extend_from_slice(buffer.as_ref());
+            }
+            (BufferHandle::new_host(bytes.freeze()), context)
+        };
+        let (element_bytes, element_context) = serialize(&elements);
+        let (offset_bytes, offset_context) = serialize(&offsets);
+        let children = (0..chunks)
+            .map(|id| {
+                FilterPlan::new(
+                    SegmentScanPlan::new(
+                        elements.dtype().clone(),
+                        chunk_len as u64,
+                        SegmentId::from(id),
+                        ReadContext::new(element_context.to_ids()),
+                        None,
+                    )
+                    .into_plan(),
+                )
+                .into_plan()
+            })
+            .collect();
+        let elements = ConcatPlan::try_new(elements.dtype().clone(), children).unwrap();
+        let offsets = FilterPlan::new(
+            SegmentScanPlan::new(
+                offsets.dtype().clone(),
+                LISTS as u64 + 1,
+                SegmentId::from(chunks),
+                ReadContext::new(offset_context.to_ids()),
+                None,
+            )
+            .into_plan(),
+        );
+        let plan = ListPackPlan::try_new(
+            Nullability::NonNullable,
+            LISTS as u64,
+            elements.into_plan(),
+            offsets.into_plan(),
+            None,
+        )
+        .unwrap()
+        .into_plan();
+        let mut segments = vec![element_bytes; chunks as usize];
+        segments.push(offset_bytes);
+        let indices = (0..100).map(|index| {
+            if uniform {
+                index * (LISTS / 100)
+            } else {
+                index
+            }
+        });
+        Self {
+            session,
+            plan,
+            segments,
+            mask: Mask::from_indices(LISTS, indices),
+        }
+    }
+
+    fn scan(&self) {
+        let split = Split {
+            rows: 0..self.mask.len() as u64,
+            mask: self.mask.clone(),
+        };
+        let scan = Scan::try_new(self.session.clone(), self.plan.clone(), vec![split]).unwrap();
+        let arrays = self.execute(scan);
+        assert_eq!(arrays.iter().map(|array| array.len()).sum::<usize>(), 100);
+        black_box(arrays);
+    }
+
+    /// A minimal in-memory planner driving each selected child graph through the same protocol.
+    fn execute(&self, mut scan: Scan) -> Vec<ArrayRef> {
+        let mut output = Vec::new();
+        loop {
+            match scan.step().unwrap() {
+                Turn::Read(read) => {
+                    scan.deliver(read.id, self.segments[*read.segment_id as usize].clone())
+                        .unwrap();
+                }
+                Turn::Output(split, array) => output.push((split, array)),
+                Turn::Child(child) => {
+                    let (id, child_scan) = child.into_scan(self.session.clone()).unwrap();
+                    let arrays = self.execute(child_scan);
+                    scan.complete_child(id, Some(arrays)).unwrap();
+                }
+                Turn::Waiting => unreachable!("every read is delivered immediately"),
+                Turn::Done => break,
+            }
+        }
+        output.sort_by_key(|(split, _)| *split);
+        output.into_iter().map(|(_, array)| array).collect()
+    }
+}
+
+/// Fixed selected row count, varying physical chunk count. Compare in separate processes with
+/// `VORTEX_LIST_PIPELINE` unset and set to `1`; no disk IO or Arrow conversion is timed.
+#[divan::bench(args = [8, 128, 1024], consts = [false, true])]
+fn list_scan<const UNIFORM: bool>(bencher: Bencher, chunks: u32) {
+    let fixture = ListFixture::new(chunks, UNIFORM);
+    bencher.bench_local(|| fixture.scan());
+}

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/operator.rs

@@ -15,6 +15,7 @@ use vortex_session::VortexSession;
 use super::pipeline::Arena;
 use super::pipeline::Inlet;
 use super::pipeline::PortId;
+use crate::plan::physical::pipeline::ChildProjection;
 use crate::segments::SegmentId;
 
 /// The output of one non-blocking compute call.
@@ -43,6 +44,8 @@ pub enum SourceReady {
 pub enum SourceBlocked {
     /// Wait for bytes from newly queued or outstanding reads.
     Io,
+    /// Wait for a child projection, possibly alongside pending reads.
+    Child,
     /// The inlet needs more batches or its sink to close it. It must be open and have room
     /// for its sink to push; debug builds check that the sink can still wake this source.
     Inlet(usize),
@@ -157,9 +160,28 @@ pub struct Cx<'a> {
     pub(super) exec: &'a mut ExecutionCtx,
     pub(super) touched: &'a mut InletSet,
     pub(super) changed: &'a mut InletSet,
+    pub(super) child_requests: &'a mut Vec<ChildProjection>,
+    pub(super) child_inbox: &'a mut Option<Box<ChildInbox>>,
+}
+
+#[derive(Default)]
+pub(super) struct ChildInbox {
+    pub(super) pending: usize,
+    pub(super) completed: VecDeque<(u64, Option<Vec<ArrayRef>>)>,
 }
 
 impl Cx<'_> {
+    /// Describes child work for the enclosing planner. Return [`SourceBlocked::Child`] while awaiting its completion.
+    /// Identifiers need only be unique among this source's outstanding child projections.
+    pub(crate) fn request_child(&mut self, child: ChildProjection) {
+        self.child_requests.push(child);
+    }
+
+    /// Takes a completed child projection, oldest completion first.
+    pub(crate) fn take_child(&mut self) -> Option<(u64, Option<Vec<ArrayRef>>)> {
+        self.child_inbox.as_mut()?.completed.pop_front()
+    }
+
     /// Takes an inlet index whose sink appended data or closed since the last readiness
     /// check. Each index is reported once, and all inlets are reported on the first check.
     /// Remaining notifications are cleared after [`Source::ready`] returns. A source must