Add layout scan physical plan model

docs/developer-guide/index.md

@@ -23,6 +23,7 @@ internals/session
 internals/async-runtime
 internals/vtables
 internals/execution
+internals/scan-planning
 internals/stats-pruning
 internals/io
 internals/serialization

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

@@ -0,0 +1,63 @@
+# Scan Plans
+
+A scan plan is the physical plan for satisfying one scan query. It is a tree of physical operators
+over a row domain, describing the reads and derived work needed to produce that query's result.
+
+## Operators, not layout mirrors
+
+Plan operators describe *what work happens*, not *which layout produced it*. Their identity and
+operator-specific state are independent of the source layout kind. The complete plan node is not:
+its common lazy-child container can own hidden source state used to materialize individual children
+on demand.
+
+| Operator | Work |
+| --- | --- |
+| `SegmentScan` | read one segment and decode it to an array |
+| `Concat` | concatenate its children row-wise |
+| `Pack` | assemble a struct from one child per field, plus optional validity |
+| `Take` | index `values` by `codes` |
+| `ListPack` | assemble a list from elements and offsets, plus optional validity |
+| `Eval` | apply an expression to its child |
+| `RowIdx` | offset row numbers into the file's row domain |
+
+Naming operators for what they compute is what lets one rule cover every case. `Concat` of
+`Concat` flattens on shape alone, and `Take` over `SegmentScan` is the dictionary pushdown,
+regardless of the source layout.
+
+The stored layout tree describes all physical data in a file. A plan is query-specific: it is built
+from that tree for one projection, filter, and row domain. Different queries over the same file can
+therefore produce different plans.
+
+## Optimization
+
+Child replacement is implemented by the common plan container rather than by every operator. It
+replaces the external child container, clones `PlanData`, then invokes the operator's
+`PlanVTable::with_children` callback to validate the new children and refresh derived caches such
+as `Concat` row offsets. Rules therefore rewrite the generic tree without reconstructing common
+plan fields inside each operator.
+
+Optimization rewrites the initial tree so that each expression is evaluated as close as possible to
+the physical data that can satisfy it. Every rewrite must preserve the query result, including its
+dtype, row domain, row order, row identity, null behavior, and observable errors.
+
+Planning does not read segment data. It constructs and optimizes a description of the work that a
+later execution stage will perform.
+
+## Vtables
+
+Each operator is a small vtable type implementing `PlanVTable`, paired with a `Plan<V>` container
+over a shared `PlanRef`. `PlanRef` points to one allocation whose ordinary fields hold the operator
+ID, dtype, row count, and lazy children. Only the unsized tail containing the vtable and
+`V::PlanData` is erased behind `dyn DynPlan`, so common-field reads do not use dynamic dispatch.
+`Plan<V>` provides typed access to that operator data through `Deref`.
+
+`PlanVTable` also carries `id` and a `Metadata` codec. Operators with no unrecoverable state
+already serialize their metadata; the ones holding a read context or a bound expression return
+`None` until those codecs exist.
+
+## 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.

vortex-array/src/expr/bound_expression.rs

@@ -227,6 +227,22 @@ impl BoundExpression {
         matches!(self.kind, BoundKind::Root)
     }
 
+    /// Return whether every scope root in this expression has `dtype`.
+    ///
+    /// Expressions without a scope root, such as literals, match every dtype.
+    pub fn is_root_bound_to(&self, dtype: &DType) -> bool {
+        let mut is_bound_to = true;
+        pre_order_visit_down(self, |node| {
+            if node.is_root() && node.dtype() != dtype {
+                is_bound_to = false;
+                return Ok(TraversalOrder::Stop);
+            }
+            Ok(TraversalOrder::Continue)
+        })
+        .vortex_expect("bound expression traversal cannot not fail");
+        is_bound_to
+    }
+
     /// Return an expression that proves this predicate is definitely false from statistics.
     pub fn falsify(&self, session: &VortexSession) -> VortexResult<Option<BoundExpression>> {
         StatsRewriteCtx::new(session).falsify(self)
@@ -361,6 +377,20 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn bound_to_checks_every_root() -> VortexResult<()> {
+        let dtype = struct_dtype();
+        let bound = eq(col("a"), col("a")).bind(&dtype)?;
+        assert!(bound.is_root_bound_to(&dtype));
+        assert!(!bound.is_root_bound_to(&DType::Bool(Nullability::NonNullable)));
+        assert!(
+            lit(true)
+                .bind(&dtype)?
+                .is_root_bound_to(&DType::Bool(Nullability::NonNullable))
+        );
+        Ok(())
+    }
+
     #[test]
     fn bound_display_matches_unbound() -> VortexResult<()> {
         for expr in [root(), col("a"), eq(col("a"), lit(1_i32)), lit(true)] {

vortex-layout/src/lib.rs

@@ -15,6 +15,7 @@
 //! optional bound filter, optional row range, [`Selection`](vortex_scan::selection::Selection),
 //! split strategy, and task concurrency settings, then produces array streams or iterators.
 pub mod layouts;
+pub mod plan;
 
 pub use children::*;
 pub use encoding::*;

vortex-layout/src/plan/children.rs

@@ -0,0 +1,118 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+use std::fmt;
+use std::sync::Arc;
+
+use once_cell::sync::OnceCell;
+use vortex_error::VortexResult;
+use vortex_error::vortex_err;
+
+use crate::plan::PlanRef;
+
+type ChildInitializer = dyn Fn(usize) -> VortexResult<PlanRef> + 'static + Send + Sync;
+
+/// Ordered plan children that may be initialized one slot at a time.
+///
+/// Eagerly constructed operators store already-filled slots. Layout lowering instead installs an
+/// initializer that owns the source layout and lowers each child on first access.
+#[derive(Clone)]
+pub struct PlanChildren {
+    initializer: Option<Arc<ChildInitializer>>,
+    cache: Arc<[OnceCell<PlanRef>]>,
+}
+
+impl PlanChildren {
+    /// Creates lazy child slots backed by `initializer`.
+    pub(crate) fn lazy(
+        len: usize,
+        initializer: impl Fn(usize) -> VortexResult<PlanRef> + 'static + Send + Sync,
+    ) -> Self {
+        Self {
+            initializer: Some(Arc::new(initializer)),
+            cache: (0..len).map(|_| OnceCell::new()).collect::<Vec<_>>().into(),
+        }
+    }
+
+    /// Returns the number of children without initializing any slot.
+    pub fn len(&self) -> usize {
+        self.cache.len()
+    }
+
+    /// Returns whether there are no children.
+    pub fn is_empty(&self) -> bool {
+        self.cache.is_empty()
+    }
+
+    /// Returns a child, initializing and caching its slot on first access.
+    pub fn get(&self, index: usize) -> VortexResult<Option<PlanRef>> {
+        let Some(cell) = self.cache.get(index) else {
+            return Ok(None);
+        };
+        if let Some(child) = cell.get() {
+            return Ok(Some(child.clone()));
+        }
+
+        let initializer = self
+            .initializer
+            .as_ref()
+            .ok_or_else(|| vortex_err!("Plan child {index} was not initialized"))?;
+        Ok(Some(cell.get_or_try_init(|| initializer(index))?.clone()))
+    }
+
+    /// Iterates over the children in logical order, initializing slots as they are visited.
+    pub fn iter(&self) -> impl ExactSizeIterator<Item = VortexResult<PlanRef>> + '_ {
+        (0..self.len()).map(|index| {
+            self.get(index)?
+                .ok_or_else(|| vortex_err!("Plan child {index} is absent"))
+        })
+    }
+
+    /// Materializes all children into an eager vector.
+    pub fn to_vec(&self) -> VortexResult<Vec<PlanRef>> {
+        self.iter().collect()
+    }
+}
+
+impl From<Vec<PlanRef>> for PlanChildren {
+    fn from(children: Vec<PlanRef>) -> Self {
+        let cache = children
+            .into_iter()
+            .map(OnceCell::with_value)
+            .collect::<Vec<_>>()
+            .into();
+        Self {
+            initializer: None,
+            cache,
+        }
+    }
+}
+
+impl<const N: usize> From<[PlanRef; N]> for PlanChildren {
+    fn from(children: [PlanRef; N]) -> Self {
+        Vec::from(children).into()
+    }
+}
+
+impl Default for PlanChildren {
+    fn default() -> Self {
+        Vec::new().into()
+    }
+}
+
+impl fmt::Debug for PlanChildren {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter
+            .debug_struct("PlanChildren")
+            .field("len", &self.len())
+            .field(
+                "initialized",
+                &self
+                    .cache
+                    .iter()
+                    .filter(|slot| slot.get().is_some())
+                    .count(),
+            )
+            .finish()
+    }
+}

vortex-layout/src/plan/display.rs

@@ -0,0 +1,131 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+use std::fmt;
+
+pub use vortex_utils::tree::DepthContext as PlanTreeContext;
+pub use vortex_utils::tree::IndentedFormatter as PlanIndentedFormatter;
+use vortex_utils::tree::TreeDisplayAdapter;
+pub use vortex_utils::tree::TreeDisplayExtractor as PlanTreeExtractor;
+use vortex_utils::tree::write_indented_tree;
+
+use super::PlanRef;
+
+/// Adds the plan's display representation to a tree node's header.
+pub struct PlanSummaryExtractor;
+
+impl PlanSummaryExtractor {
+    /// Writes a plan directly to `formatter`.
+    pub fn write(plan: &PlanRef, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(formatter, "{plan}")
+    }
+}
+
+impl PlanTreeExtractor<PlanRef, PlanTreeContext> for PlanSummaryExtractor {
+    fn write_header(
+        &self,
+        plan: &PlanRef,
+        _context: &PlanTreeContext,
+        formatter: &mut fmt::Formatter<'_>,
+    ) -> fmt::Result {
+        write!(formatter, " ")?;
+        Self::write(plan, formatter)
+    }
+}
+
+/// Composable display builder for a physical plan tree.
+///
+/// Call `plan.tree_display()` for the default extractors. Use `plan.tree_display_builder()` to
+/// start with only node and child names, then add extractors with [`Self::with`].
+pub struct PlanTreeDisplay<'a> {
+    plan: &'a PlanRef,
+    extractors: Vec<Box<dyn PlanTreeExtractor<PlanRef, PlanTreeContext>>>,
+}
+
+impl<'a> PlanTreeDisplay<'a> {
+    /// Creates a tree display for `plan` with no extractors.
+    pub fn new(plan: &'a PlanRef) -> Self {
+        Self {
+            plan,
+            extractors: Vec::new(),
+        }
+    }
+
+    /// Creates a tree display using each plan's display representation.
+    pub fn default_display(plan: &'a PlanRef) -> Self {
+        Self::new(plan).with(PlanSummaryExtractor)
+    }
+
+    /// Adds an extractor to the display pipeline.
+    pub fn with<E: PlanTreeExtractor<PlanRef, PlanTreeContext> + 'static>(
+        mut self,
+        extractor: E,
+    ) -> Self {
+        self.extractors.push(Box::new(extractor));
+        self
+    }
+
+    /// Adds a pre-boxed extractor to the display pipeline.
+    pub fn with_boxed(
+        mut self,
+        extractor: Box<dyn PlanTreeExtractor<PlanRef, PlanTreeContext>>,
+    ) -> Self {
+        self.extractors.push(extractor);
+        self
+    }
+}
+
+impl TreeDisplayAdapter for PlanTreeDisplay<'_> {
+    type Context = PlanTreeContext;
+    type Node = PlanRef;
+
+    fn write_node(
+        &self,
+        plan: &PlanRef,
+        context: &PlanTreeContext,
+        formatter: &mut fmt::Formatter<'_>,
+    ) -> fmt::Result {
+        for extractor in &self.extractors {
+            extractor.write_header(plan, context, formatter)?;
+        }
+        Ok(())
+    }
+
+    fn write_details(
+        &self,
+        plan: &PlanRef,
+        context: &PlanTreeContext,
+        formatter: &mut PlanIndentedFormatter<'_, '_>,
+    ) -> fmt::Result {
+        for extractor in &self.extractors {
+            extractor.write_details(plan, context, formatter)?;
+        }
+        Ok(())
+    }
+
+    fn visit_children(
+        &self,
+        plan: &PlanRef,
+        visit: &mut dyn FnMut(&str, &PlanRef, bool) -> fmt::Result,
+    ) -> fmt::Result {
+        let children = plan.children();
+        for index in 0..children.len() {
+            let child = plan.child_required(index).map_err(|_| fmt::Error)?;
+            let child_name = plan.child_name(index);
+            visit(child_name.as_ref(), &child, index + 1 == children.len())?;
+        }
+        Ok(())
+    }
+}
+
+impl fmt::Display for PlanTreeDisplay<'_> {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write_indented_tree(
+            self,
+            "root",
+            self.plan,
+            &mut PlanTreeContext::default(),
+            formatter,
+        )
+    }
+}