feat(layout): add push-based pipelines and the scan that runs them (#10391)

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

Author: joseph-isaacs

Cargo.lock

@@ -35,6 +35,7 @@ pin-project-lite = { workspace = true }
 prost = { workspace = true }
 rustc-hash = { workspace = true }
 sketches-ddsketch = { workspace = true }
+smallvec = { workspace = true }
 termtree = { workspace = true }
 tokio = { workspace = true, features = ["rt"], optional = true }
 tracing = { workspace = true }

vortex-layout/Cargo.toml

@@ -1,74 +1,6 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright the Vortex contributors
 
-//! Physical plans for scans.
-//!
-//! A plan is a tree of physical operators over a row domain. Operator identity and
-//! operator-specific state do not depend on the source layout kind, so rewrites can reason about a
-//! plan's shape alone. The common child container can initialize individual slots lazily.
+//! Scan planning.
 
-mod children;
-mod display;
-mod lower;
-mod optimize;
-pub mod optimizer;
-mod plans;
-mod typed;
-mod vtable;
-
-pub use children::PlanChildren;
-pub use display::PlanIndentedFormatter;
-pub use display::PlanSummaryExtractor;
-pub use display::PlanTreeContext;
-pub use display::PlanTreeDisplay;
-pub use display::PlanTreeExtractor;
-pub use lower::lower;
-pub use optimize::optimize;
-pub use plans::Concat;
-pub use plans::ConcatData;
-pub use plans::ConcatPlan;
-pub use plans::Eval;
-pub use plans::EvalData;
-pub use plans::EvalPlan;
-pub use plans::ListPack;
-pub use plans::ListPackData;
-pub use plans::ListPackPlan;
-pub use plans::Pack;
-pub use plans::PackData;
-pub use plans::PackPlan;
-pub use plans::RowIdx;
-pub use plans::RowIdxData;
-pub use plans::RowIdxPlan;
-pub use plans::SegmentScan;
-pub use plans::SegmentScanData;
-pub use plans::SegmentScanPlan;
-pub use plans::Take;
-pub use plans::TakePlan;
-pub use plans::plan_row_idx_expression;
-pub use plans::row_idx_dtype;
-pub use typed::DynPlan;
-pub use typed::Plan;
-pub use typed::PlanParts;
-pub use typed::PlanRef;
-use vortex_error::VortexResult;
-use vortex_error::vortex_bail;
-pub use vtable::PlanId;
-pub use vtable::PlanVTable;
-
-/// Returns an error when `children` does not have exactly `expected` entries.
-pub(crate) fn check_child_count(
-    name: &str,
-    children: &PlanChildren,
-    expected: usize,
-) -> VortexResult<()> {
-    if children.len() != expected {
-        vortex_bail!(
-            "{name} expects {expected} children but got {}",
-            children.len()
-        );
-    }
-    Ok(())
-}
-
-#[cfg(test)]
-mod tests;
+pub mod physical;

vortex-layout/src/plan/mod.rs

@@ -8,7 +8,7 @@ use once_cell::sync::OnceCell;
 use vortex_error::VortexResult;
 use vortex_error::vortex_err;
 
-use crate::plan::PlanRef;
+use crate::plan::physical::PlanRef;
 
 type ChildInitializer = dyn Fn(usize) -> VortexResult<PlanRef> + 'static + Send + Sync;
 

vortex-layout/src/plan/children.rs …tex-layout/src/plan/physical/children.rsvortex-layout/src/plan/children.rs renamed to vortex-layout/src/plan/physical/children.rs vortex-layout/src/plan/children.rs renamed to vortex-layout/src/plan/physical/children.rs

@@ -24,13 +24,13 @@ use crate::layouts::list::OFFSETS_CHILD_INDEX;
 use crate::layouts::list::VALIDITY_CHILD_INDEX;
 use crate::layouts::struct_::Struct;
 use crate::layouts::struct_::StructLayout;
-use crate::plan::ConcatPlan;
-use crate::plan::ListPackPlan;
-use crate::plan::PackPlan;
-use crate::plan::PlanChildren;
-use crate::plan::PlanRef;
-use crate::plan::SegmentScanPlan;
-use crate::plan::TakePlan;
+use crate::plan::physical::ConcatPlan;
+use crate::plan::physical::ListPackPlan;
+use crate::plan::physical::PackPlan;
+use crate::plan::physical::PlanChildren;
+use crate::plan::physical::PlanRef;
+use crate::plan::physical::SegmentScanPlan;
+use crate::plan::physical::TakePlan;
 
 /// Constructs a physical-plan fixture from `layout` for tests.
 ///

vortex-layout/src/plan/display.rs …rtex-layout/src/plan/physical/display.rsvortex-layout/src/plan/display.rs renamed to vortex-layout/src/plan/physical/display.rs vortex-layout/src/plan/display.rs renamed to vortex-layout/src/plan/physical/display.rs

@@ -0,0 +1,80 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+//! Physical plans for scans.
+//!
+//! A plan is a tree of physical operators over a row domain. Operator identity and
+//! operator-specific state do not depend on the source layout kind, so rewrites can reason about a
+//! plan's shape alone. The common child container can initialize individual slots lazily.
+//!
+//! [`operator`] defines the execution contracts, `operators` implements the concrete sources
+//! and stages, and [`pipeline`] compiles and runs them.
+
+mod children;
+mod display;
+mod lower;
+pub mod operator;
+pub(crate) mod operators;
+mod optimize;
+pub mod optimizer;
+pub mod pipeline;
+mod plans;
+mod typed;
+mod vtable;
+
+pub use children::PlanChildren;
+pub use display::PlanIndentedFormatter;
+pub use display::PlanSummaryExtractor;
+pub use display::PlanTreeContext;
+pub use display::PlanTreeDisplay;
+pub use display::PlanTreeExtractor;
+pub use lower::lower;
+pub use optimize::optimize;
+pub use plans::Concat;
+pub use plans::ConcatData;
+pub use plans::ConcatPlan;
+pub use plans::Eval;
+pub use plans::EvalData;
+pub use plans::EvalPlan;
+pub use plans::ListPack;
+pub use plans::ListPackData;
+pub use plans::ListPackPlan;
+pub use plans::Pack;
+pub use plans::PackData;
+pub use plans::PackPlan;
+pub use plans::RowIdx;
+pub use plans::RowIdxData;
+pub use plans::RowIdxPlan;
+pub use plans::SegmentScan;
+pub use plans::SegmentScanData;
+pub use plans::SegmentScanPlan;
+pub use plans::Take;
+pub use plans::TakePlan;
+pub use plans::plan_row_idx_expression;
+pub use plans::row_idx_dtype;
+pub use typed::DynPlan;
+pub use typed::Plan;
+pub use typed::PlanParts;
+pub use typed::PlanRef;
+use vortex_error::VortexResult;
+use vortex_error::vortex_bail;
+pub use vtable::PlanId;
+pub use vtable::PlanVTable;
+
+/// Returns an error when `children` does not have exactly `expected` entries.
+pub(crate) fn check_child_count(
+    name: &str,
+    children: &PlanChildren,
+    expected: usize,
+) -> VortexResult<()> {
+    if children.len() != expected {
+        vortex_bail!(
+            "{name} expects {expected} children but got {}",
+            children.len()
+        );
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests;