Add plan-native scan execution

Give `PlanVTable` an `execute` hook taking a row range and a selection
mask, and implement it for every operator: `SegmentScan` reads and decodes
its segment, the structural operators combine their children, and `Eval`
applies the expression to its child's output.

Add `vortex-scan-v2`, which copies the existing scan orchestration around
this API so the `LayoutReader` scanner is untouched while the plan-native
path is developed.

`Take` now records whether every dictionary value is referenced by some
code. That fact previously came from the dict layout; since operators no
longer hold a layout, lowering carries it into the operator's data.

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

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012obBhJ8oPZoBbKyeS79yMv

Author: joseph-isaacs

Cargo.lock

@@ -23,6 +23,7 @@ members = [
     "vortex-btrblocks",
     "vortex-layout",
     "vortex-scan",
+    "vortex-scan-v2",
     "vortex-file",
     "vortex-ipc",
     "vortex",
@@ -322,6 +323,7 @@ vortex-proto = { version = "0.1.0", path = "./vortex-proto", default-features =
 vortex-row = { version = "0.1.0", path = "./vortex-row", default-features = false }
 vortex-runend = { version = "0.1.0", path = "./encodings/runend", default-features = false }
 vortex-scan = { version = "0.1.0", path = "./vortex-scan", default-features = false }
+vortex-scan-v2 = { version = "0.1.0", path = "./vortex-scan-v2", default-features = false }
 vortex-sequence = { version = "0.1.0", path = "encodings/sequence", default-features = false }
 vortex-session = { version = "0.1.0", path = "./vortex-session", default-features = false }
 vortex-sparse = { version = "0.1.0", path = "./encodings/sparse", default-features = false }

Cargo.toml

@@ -57,12 +57,17 @@ holding the common fields — dtype, row count, and children. Operator-specific
 already serialize their metadata; the ones holding a read context or a bound expression return
 `None` until those codecs exist.
 
+## Execution
+
+Each operator executes over a row range and selection mask. `SegmentScan` reads its segment,
+structural operators combine their children, and `Eval` applies the remaining derived work.
+`vortex-scan-v2` copies the existing scan orchestration around this API, so the original
+`LayoutReader` scanner is untouched while the plan-native path is developed.
+
 ## Future work
 
-Plans currently stop at construction and optimization. Still to come: a plan registry and foreign
-operator placeholder so third-party operators survive a round trip, a serialization envelope, and
-an execution stage that walks an optimized plan, reads the referenced segments, and returns the
-query result.
+Still to come: a plan registry and foreign operator placeholder so third-party operators survive
+a round trip, and a serialization envelope.
 
 Lowering does not yet take a projection or row range, so it lowers the whole layout tree. Once it
 does, an unsupported layout in a column the query never reads will no longer fail the scan.

docs/developer-guide/internals/scan-planning.md

@@ -297,7 +297,7 @@ fn row_idx_dtype() -> DType {
 }
 
 // Returns a SequenceArray representing the row indices for the given row range,
-fn idx_array(row_offset: u64, row_range: &Range<u64>) -> SequenceArray {
+pub(crate) fn idx_array(row_offset: u64, row_range: &Range<u64>) -> SequenceArray {
     Sequence::try_new(
         PValue::U64(row_offset + row_range.start),
         PValue::U64(1),

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

@@ -0,0 +1,41 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+use std::sync::Arc;
+
+use futures::future::BoxFuture;
+use vortex_array::ArrayRef;
+use vortex_error::VortexResult;
+use vortex_session::VortexSession;
+
+use crate::segments::SegmentSource;
+
+/// Future resolving to the array produced by a physical plan.
+pub type PlanArrayFuture = BoxFuture<'static, VortexResult<ArrayRef>>;
+
+/// Runtime dependencies shared by every node in a plan execution.
+#[derive(Clone)]
+pub struct PlanExecutionContext {
+    segment_source: Arc<dyn SegmentSource>,
+    session: VortexSession,
+}
+
+impl PlanExecutionContext {
+    /// Creates an execution context over a segment source and Vortex session.
+    pub fn new(segment_source: Arc<dyn SegmentSource>, session: VortexSession) -> Self {
+        Self {
+            segment_source,
+            session,
+        }
+    }
+
+    /// Returns the segment source used to satisfy leaf reads.
+    pub fn segment_source(&self) -> &Arc<dyn SegmentSource> {
+        &self.segment_source
+    }
+
+    /// Returns the Vortex session used for array decoding and expression execution.
+    pub fn session(&self) -> &VortexSession {
+        &self.session
+    }
+}

vortex-layout/src/plan/execution.rs

@@ -123,7 +123,11 @@ fn lower_dict(layout: &DictLayout) -> VortexResult<TakePlan> {
             .slot(0)?
             .ok_or_else(|| vortex_err!("Dictionary values child is absent"))?,
     )?;
-    Ok(TakePlan::new(codes, values))
+    Ok(TakePlan::new_with_all_values_referenced(
+        codes,
+        values,
+        layout.has_all_values_referenced(),
+    ))
 }
 
 fn lower_list(layout: &ListLayout) -> VortexResult<ListPackPlan> {

vortex-layout/src/plan/lower.rs

@@ -8,6 +8,7 @@
 //! can reason about a plan's shape alone. [`lower`] is the one-way bridge from a stored layout.
 
 mod display;
+mod execution;
 mod lower;
 mod optimize;
 pub mod optimizer;
@@ -22,6 +23,8 @@ pub use display::PlanSummaryExtractor;
 pub use display::PlanTreeContext;
 pub use display::PlanTreeDisplay;
 pub use display::PlanTreeExtractor;
+pub use execution::PlanArrayFuture;
+pub use execution::PlanExecutionContext;
 pub use lower::lower;
 pub use optimize::optimize;
 pub use plans::Concat;
@@ -50,6 +53,7 @@ pub use plans::SegmentScan;
 pub use plans::SegmentScanData;
 pub use plans::SegmentScanPlan;
 pub use plans::Take;
+pub use plans::TakeData;
 pub use plans::TakePlan;
 pub use plans::Zoned;
 pub use plans::ZonedPlan;

vortex-layout/src/plan/mod.rs

@@ -2,20 +2,34 @@
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
 use std::borrow::Cow;
+use std::future;
+use std::ops::Range;
 use std::sync::Arc;
 
+use futures::FutureExt;
+use futures::TryStreamExt;
+use futures::stream::FuturesOrdered;
+use vortex_array::Canonical;
 use vortex_array::EmptyMetadata;
+use vortex_array::IntoArray;
+use vortex_array::MaskFuture;
+use vortex_array::arrays::ChunkedArray;
 use vortex_array::dtype::DType;
 use vortex_array::expr::ExactBoundExpr;
 use vortex_array::expr::label_bound_tree;
+use vortex_error::VortexExpect;
 use vortex_error::VortexResult;
 use vortex_error::vortex_bail;
+use vortex_error::vortex_ensure;
+use vortex_error::vortex_err;
 use vortex_session::registry::CachedId;
 
 use crate::layouts::row_idx::RowIdx as RowIdxFn;
 use crate::plan::Eval;
 use crate::plan::EvalPlan;
 use crate::plan::Plan;
+use crate::plan::PlanArrayFuture;
+use crate::plan::PlanExecutionContext;
 use crate::plan::PlanId;
 use crate::plan::PlanParts;
 use crate::plan::PlanRef;
@@ -95,6 +109,56 @@ impl PlanVTable for Concat {
         ConcatPlan::try_new(plan.dtype().clone(), children)
     }
 
+    fn execute(
+        plan: &Plan<Self>,
+        ctx: &PlanExecutionContext,
+        row_range: &Range<u64>,
+        mask: MaskFuture,
+    ) -> VortexResult<PlanArrayFuture> {
+        vortex_ensure!(
+            row_range.start <= row_range.end && row_range.end <= plan.row_count(),
+            "Concat row range {:?} is outside 0..{}",
+            row_range,
+            plan.row_count()
+        );
+        vortex_ensure!(
+            mask.len() == usize::try_from(row_range.end - row_range.start)?,
+            "Concat mask length mismatch"
+        );
+        if row_range.is_empty() {
+            let empty = Canonical::empty(plan.dtype()).into_array();
+            return Ok(future::ready(Ok(empty)).boxed());
+        }
+
+        let mut chunk_futures = Vec::new();
+        for (chunk, &chunk_offset) in plan.children().iter().zip(plan.row_offsets()) {
+            let chunk_end = chunk_offset
+                .checked_add(chunk.row_count())
+                .ok_or_else(|| vortex_err!("Chunk row offset overflow"))?;
+            let start = row_range.start.max(chunk_offset);
+            let end = row_range.end.min(chunk_end);
+            if start < end {
+                let child_range = start - chunk_offset..end - chunk_offset;
+                let mask_range = usize::try_from(start - row_range.start)?
+                    ..usize::try_from(end - row_range.start)?;
+                chunk_futures.push(chunk.execute(ctx, &child_range, mask.slice(mask_range))?);
+            }
+        }
+
+        Ok(async move {
+            let chunks: Vec<_> = FuturesOrdered::from_iter(chunk_futures)
+                .try_collect()
+                .await?;
+            vortex_ensure!(!chunks.is_empty(), "Non-empty row range selected no chunks");
+            if chunks.len() == 1 {
+                return Ok(chunks.into_iter().next().vortex_expect("one chunk"));
+            }
+            let dtype = chunks[0].dtype().clone();
+            Ok(ChunkedArray::try_new(chunks, dtype)?.into_array())
+        }
+        .boxed())
+    }
+
     fn child_name(_plan: &Plan<Self>, index: usize) -> Cow<'_, str> {
         Cow::Owned(format!("chunks[{index}]"))
     }

vortex-layout/src/plan/plans/concat.rs

@@ -2,8 +2,11 @@
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
 use std::borrow::Cow;
+use std::ops::Range;
 
+use futures::FutureExt;
 use vortex_array::EmptyMetadata;
+use vortex_array::MaskFuture;
 use vortex_array::dtype::DType;
 use vortex_array::dtype::FieldName;
 use vortex_array::expr::BoundExpression;
@@ -16,6 +19,8 @@ use vortex_error::VortexResult;
 use vortex_session::registry::CachedId;
 
 use crate::plan::Plan;
+use crate::plan::PlanArrayFuture;
+use crate::plan::PlanExecutionContext;
 use crate::plan::PlanId;
 use crate::plan::PlanParts;
 use crate::plan::PlanRef;
@@ -83,6 +88,17 @@ impl PlanVTable for Eval {
         ))
     }
 
+    fn execute(
+        plan: &Plan<Self>,
+        ctx: &PlanExecutionContext,
+        row_range: &Range<u64>,
+        mask: MaskFuture,
+    ) -> VortexResult<PlanArrayFuture> {
+        let child = plan.child_plan().execute(ctx, row_range, mask)?;
+        let expression = plan.expression().clone();
+        Ok(async move { child.await?.apply_bound(&expression) }.boxed())
+    }
+
     fn child_name(_plan: &Plan<Self>, index: usize) -> Cow<'_, str> {
         if index == 0 {
             Cow::Borrowed("child")