feat(layout): planner-driven V2 scan over physical plans

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

@@ -96,6 +96,11 @@ impl VTable for Chunked {
 }
 
 impl Layout<Chunked> {
+    /// Cached chunk boundaries, including the final row count.
+    pub(crate) fn chunk_offsets(&self) -> &[u64] {
+        &self.chunk_offsets
+    }
+
     /// Construct a chunked layout.
     pub fn new(row_count: u64, dtype: DType, children: Arc<dyn LayoutChildren>) -> Self {
         let offsets = chunk_offsets(children.as_ref()).vortex_expect("chunk row counts overflow");

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

@@ -99,6 +99,14 @@ impl PlanRef {
             .ok_or_else(|| vortex_err!("Missing plan child {index}"))
     }
 
+    /// Borrows the child at `index`, or returns an error if it is out of bounds.
+    pub(crate) fn child_ref_required(&self, index: usize) -> VortexResult<&PlanRef> {
+        self.0
+            .children
+            .get_ref(index)?
+            .ok_or_else(|| vortex_err!("Missing plan child {index}"))
+    }
+
     /// Rebuilds this plan with `children` stored outside its erased operator data.
     pub fn with_children(&self, children: impl Into<PlanChildren>) -> VortexResult<PlanRef> {
         self.dyn_plan().dyn_with_children(self, children.into())

vortex-layout/src/scan/filter.rs

@@ -55,7 +55,11 @@ fn bound_conjuncts(expr: &BoundExpression) -> Vec<BoundExpression> {
 
 impl FilterExpr {
     pub fn new(expr: BoundExpression) -> Self {
-        let conjuncts = bound_conjuncts(&expr);
+        Self::from_conjuncts(bound_conjuncts(&expr))
+    }
+
+    /// Builds a filter from conjuncts that are already split, without splitting them further.
+    pub(crate) fn from_conjuncts(conjuncts: Vec<BoundExpression>) -> Self {
         let num_conjuncts = conjuncts.len();
 
         let dynamic_conjuncts = conjuncts.iter().map(DynamicExprUpdates::new).collect_vec();

vortex-layout/src/scan/mod.rs

@@ -6,13 +6,15 @@ mod filter;
 pub mod layout;
 mod limit;
 pub mod multi;
+pub mod planning;
 pub mod repeated_scan;
 pub mod scan_builder;
 pub mod split_by;
 mod splits;
 mod tasks;
 #[cfg(test)]
 mod test;
+pub mod v2;
 
 /// A heuristic for an ideal split size.
 ///