From 5b3394989241d25296b4a73f517eab6f516f745e Mon Sep 17 00:00:00 2001 From: zz_y Date: Sat, 29 Aug 2026 16:38:37 -0600 Subject: [PATCH 1/4] feat(post-asap): model and validate execution phases --- crates/types/src/dag_export.rs | 145 ++++- crates/types/src/post_asap/expr.rs | 74 +++ crates/types/src/post_asap/mod.rs | 8 +- crates/types/src/post_asap/phase.rs | 867 ++++++++++++++++++++++++++++ 4 files changed, 1082 insertions(+), 12 deletions(-) create mode 100644 crates/types/src/post_asap/phase.rs diff --git a/crates/types/src/dag_export.rs b/crates/types/src/dag_export.rs index 5df2ed8c..bda2196f 100644 --- a/crates/types/src/dag_export.rs +++ b/crates/types/src/dag_export.rs @@ -46,7 +46,10 @@ use std::rc::Rc; use serde::Serialize; -use crate::post_asap::{AccuracyError, ResultGuarantee, SummaryExpr, SummaryNode}; +use crate::post_asap::{ + assigned_child_stage, produced_availability, AccuracyError, ExactOperator, + ExecutionAvailability, ResultGuarantee, SummaryExpr, SummaryNode, ValueOperator, +}; use crate::pre_asap::cse::{structural_hash, HashCache}; use crate::pre_asap::query_expr::{QueryExpr, Source}; @@ -152,6 +155,24 @@ pub struct DagDecision { /// `replacement_root` for the node replacing the pre-ASAP target; /// `replacement_region` for its generated or carried descendants. pub role: &'static str, + /// Machine-readable origin of the winning candidate (a `Debug`-formatted + /// `asap_aware_mapping::ReplacementProvenance`, e.g. + /// `"ExactPostProcess"`), so a viewer never infers *how* a node was + /// composed from its label or shape (issue #171). Omitted when the + /// producing layer predates this field. + #[serde(skip_serializing_if = "Option::is_none")] + pub provenance: Option, + /// The unit `cost` is expressed in — e.g. `"cost_units_per_second"` for + /// a recurring-rate comparison, or absent for the legacy unitless + /// structural estimate. Additive; consumers must not assume one unit. + #[serde(skip_serializing_if = "Option::is_none")] + pub cost_unit: Option, + /// For a composed decision (an exact operator over another target's + /// own selected decision), the `id`s of the child decisions this one + /// was committed together with — the explicit target-to-decision + /// provenance chain, never reconstructed from graph adjacency. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub child_decisions: Vec, } /// One query's exported graph. `nodes[root as usize]` is the tree's root. @@ -338,10 +359,63 @@ pub struct SummaryDagGraph { /// top-level `DagNode::kind`. pub fn export_summary(node: &SummaryNode) -> SummaryDagGraph { let mut nodes = Vec::new(); - let root = build_summary(node, &mut nodes); + let root = build_summary(node, &mut nodes, root_stage(node)); SummaryDagGraph { nodes, root } } +/// The explicit execution stage of an exported plan's root — its own +/// produced availability, or query-time readout for a bare `KeepPreAsap` +/// (the same convention `post_asap::phase::validate_execution_phases` +/// uses for a root). +fn root_stage(node: &SummaryNode) -> ExecutionAvailability { + produced_availability(&node.expr).unwrap_or(ExecutionAvailability::ReadoutValue) +} + +/// `detail` for an [`ExactOperator`] payload — its own fields, rendered the +/// same way the pre-ASAP `Aggregate` node renders them. +fn exact_operator_detail(op: &ExactOperator) -> serde_json::Value { + match op { + ExactOperator::Aggregate { + reduction, + measures, + output_names, + having, + } => serde_json::json!({ + "op": "Aggregate", + "reduction": reduction, + "measures": measures, + "output_names": output_names, + "having": having.as_ref().map(|p| export(&p.0)), + }), + } +} + +fn exact_operator_label(op: &ExactOperator) -> String { + match op { + ExactOperator::Aggregate { measures, .. } => { + let funcs: Vec = measures.iter().map(|m| format!("{m:?}")).collect(); + format!("Aggregate[{}]", funcs.join(", ")) + } + } +} + +fn value_operator_detail(op: &ValueOperator) -> serde_json::Value { + match op { + ValueOperator::Exact(op) => exact_operator_detail(op), + ValueOperator::Extension { name } => serde_json::json!({ + "op": "Extension", + "name": name, + }), + } +} + +fn value_operator_label(op: &ValueOperator) -> String { + match op { + ValueOperator::Exact(op) => exact_operator_label(op), + ValueOperator::Extension { name } => name.clone(), + } +} + fn push_summary_node( nodes: &mut Vec, kind: &'static str, @@ -391,8 +465,16 @@ fn family_label(family: &crate::post_asap::SummaryFamilyType) -> String { /// shared [`DagGraph`] node list — see [`export_post_asap`]) can't drift /// apart on how every *other* variant's own shape is described, since /// nothing about that description differs between the two. -fn summary_shape(expr: &SummaryExpr) -> (&'static str, String, serde_json::Value) { - match expr { +/// +/// `stage` is the node's explicit execution phase (issue #171) — its own +/// [`produced_availability`], or the edge-assigned phase for a `KeepPreAsap` +/// — and is written into `detail.stage` on every post-ASAP node so a viewer +/// reads it rather than inferring it from the node's kind. +fn summary_shape( + expr: &SummaryExpr, + stage: ExecutionAvailability, +) -> (&'static str, String, serde_json::Value) { + let (kind, label, mut detail) = match expr { SummaryExpr::KeepPreAsap(_) => { unreachable!("summary_shape's callers special-case KeepPreAsap before calling it") } @@ -438,7 +520,22 @@ fn summary_shape(expr: &SummaryExpr) -> (&'static str, String, serde_json::Value let label = format!("SummaryMerge({} children)", children.len()); ("SummaryMerge", label, serde_json::json!({})) } + SummaryExpr::UpdateTransform { op, .. } => { + let label = format!("UpdateTransform({})", value_operator_label(op)); + ("UpdateTransform", label, value_operator_detail(op)) + } + SummaryExpr::ReadoutPostProcess { op, .. } => { + let label = format!("ReadoutPostProcess({})", value_operator_label(op)); + ("ReadoutPostProcess", label, value_operator_detail(op)) + } + }; + if let serde_json::Value::Object(map) = &mut detail { + map.insert( + "stage".into(), + serde_json::Value::String(stage.as_str().into()), + ); } + (kind, label, detail) } /// `expr`'s own `Rc` children, in the variant's field order @@ -455,6 +552,10 @@ fn summary_children(expr: &SummaryExpr) -> Vec<&Rc> { SummaryExpr::SummaryDelete { summary_input, .. } => vec![summary_input], SummaryExpr::SummaryEstimate { summary_input, .. } => vec![summary_input], SummaryExpr::SummaryMerge { children } => children.iter().collect(), + SummaryExpr::UpdateTransform { child, .. } + | SummaryExpr::ReadoutPostProcess { child, .. } => { + vec![child] + } } } @@ -462,12 +563,19 @@ fn summary_children(expr: &SummaryExpr) -> Vec<&Rc> { /// post-order (children pushed before their parent), and return the pushed /// root's id. Exhaustive over every [`SummaryExpr`] variant, matching this /// file's own exhaustive style for `QueryExpr` in [`build`]. -fn build_summary(node: &SummaryNode, nodes: &mut Vec) -> u32 { +fn build_summary( + node: &SummaryNode, + nodes: &mut Vec, + stage: ExecutionAvailability, +) -> u32 { if let SummaryExpr::KeepPreAsap(inner) = &node.expr { let pre_asap_subgraph = export(inner); let inner_kind = pre_asap_subgraph.nodes[pre_asap_subgraph.root as usize].kind; let label = format!("KeepPreAsap({inner_kind})"); - let detail = serde_json::json!({ "pre_asap_subgraph": pre_asap_subgraph }); + let detail = serde_json::json!({ + "pre_asap_subgraph": pre_asap_subgraph, + "stage": stage.as_str(), + }); return push_summary_node( nodes, "KeepPreAsap", @@ -479,9 +587,9 @@ fn build_summary(node: &SummaryNode, nodes: &mut Vec) -> u32 { } let children: Vec = summary_children(&node.expr) .into_iter() - .map(|child| build_summary(child, nodes)) + .map(|child| build_summary(child, nodes, assigned_child_stage(&node.expr, child))) .collect(); - let (kind, label, detail) = summary_shape(&node.expr); + let (kind, label, detail) = summary_shape(&node.expr, stage); push_summary_node(nodes, kind, label, detail, children, node.guarantee.clone()) } @@ -759,15 +867,24 @@ fn build_summary_hybrid( nodes: &mut Vec, cache: &mut HashCache, find_winner: &mut dyn FnMut(&QueryExpr) -> Option, + stage: ExecutionAvailability, ) -> u32 { if let SummaryExpr::KeepPreAsap(inner) = &node.expr { return build(inner, nodes, cache, find_winner); } let children: Vec = summary_children(&node.expr) .into_iter() - .map(|child| build_summary_hybrid(child, nodes, cache, find_winner)) + .map(|child| { + build_summary_hybrid( + child, + nodes, + cache, + find_winner, + assigned_child_stage(&node.expr, child), + ) + }) .collect(); - let (kind, label, mut detail) = summary_shape(&node.expr); + let (kind, label, mut detail) = summary_shape(&node.expr, stage); // The merged graph's `DagNode` has no dedicated guarantee field (it is // the pre-ASAP node shape); the guarantee rides in `detail` under the // same key/shape `SummaryDagNode::guarantee` uses, additively. @@ -853,7 +970,13 @@ fn build( decision, }) => { let first = nodes.len(); - let root = build_summary_hybrid(&replacement, nodes, cache, find_winner); + let root = build_summary_hybrid( + &replacement, + nodes, + cache, + find_winner, + root_stage(&replacement), + ); for node in &mut nodes[first..] { if node.decision.is_none() { let mut node_decision = decision.clone(); diff --git a/crates/types/src/post_asap/expr.rs b/crates/types/src/post_asap/expr.rs index b93aa904..3eda8f82 100644 --- a/crates/types/src/post_asap/expr.rs +++ b/crates/types/src/post_asap/expr.rs @@ -3,8 +3,60 @@ use std::rc::Rc; use super::guarantee::ResultGuarantee; use super::schema::{SummaryFamilyType, SummarySchema}; use super::sketch::{GroupingStrategy, SketchQuery}; +use crate::pre_asap::agg_intent::AggIntent; +use crate::pre_asap::query_expr::Predicate; use crate::pre_asap::{ColumnRef, QueryExpr, Reduction}; +// ── Exact operators composed with summary plans (issue #171) ──────────────── + +/// An exact, plain-row operator that a mixed exact/summary plan executes at +/// an explicit phase. Exact composition is one producer of the generic +/// [`ValueOperator`] phase payload. +/// +/// Deliberately **not** an intact pre-ASAP [`QueryExpr`] subtree: a +/// `QueryExpr`'s children are always `Rc`, so embedding one here +/// would point back at pre-ASAP nodes and recreate exactly the opaque +/// boundary [`SummaryExpr::KeepPreAsap`] already has (a logical parent +/// swallowing an otherwise-realizable descendant). Instead this carries only +/// the operator's *own* fields; its input is the post-ASAP `child` of the +/// enclosing `SummaryExpr` variant. +/// +/// `#[non_exhaustive]`: starts with the one payload issue #171 needs. Future +/// PRs add `Filter`/`Project`/`BinaryOp`/`Sort`/`Limit` payloads when a +/// concrete mixed plan needs them — never every relational `QueryExpr` +/// variant at once. +#[derive(Debug, Clone, PartialEq)] +#[non_exhaustive] +pub enum ExactOperator { + /// The same fields a pre-ASAP `QueryExpr::Aggregate` carries, applied + /// exactly (no summary family) over the enclosing node's post-ASAP + /// `child`. Output schema follows + /// `pre_asap::query_expr::aggregate_output_schema` over the child's + /// plain schema. + Aggregate { + reduction: Reduction, + measures: Vec, + output_names: Vec, + having: Option, + }, +} + +/// An operation over values at a declared execution phase. +/// +/// Phase placement is independent of whether the operation is exact or +/// approximate: [`SummaryExpr::UpdateTransform`] and +/// [`SummaryExpr::ReadoutPostProcess`] describe when their input is +/// available, while this payload describes what is computed. The extension +/// form lets summary families and approximate strategies name operations +/// whose output schema and guarantee are carried by the enclosing +/// [`SummaryNode`]. +#[derive(Debug, Clone, PartialEq)] +#[non_exhaustive] +pub enum ValueOperator { + Exact(ExactOperator), + Extension { name: String }, +} + // ── Post-ASAP DAG node ─────────────────────────────────────────────────────── /// A node in the post-ASAP DAG: wraps the expression and its derived output @@ -134,4 +186,26 @@ pub enum SummaryExpr { /// allocator (not modeled in this crate) on cut edges. /// Output schema: one field (same family + params as inputs). SummaryMerge { children: Vec> }, + + /// Value transformation executed on the **update/ingest + /// path** (issue #171). Consumes `child`'s plain update values and + /// produces plain update values, so its output may feed a downstream + /// [`SummaryAgg`](SummaryExpr::SummaryAgg)'s maintenance — the "outer + /// summary over an inner non-accumulator exact transform" direction. + /// See [`super::phase::ExecutionAvailability`] for the edge contract. + UpdateTransform { + child: Rc, + op: ValueOperator, + }, + + /// Operation executed **after** `child`'s summary has been read + /// out (issue #171). Consumes plain readout values and produces the + /// final plain query result — the "outer exact fold over an inner + /// summary readout" direction. Can never feed maintained state: a + /// `SummaryAgg` above one of these is a plan-time + /// [`super::phase::PhaseError`], never a runtime failure. + ReadoutPostProcess { + child: Rc, + op: ValueOperator, + }, } diff --git a/crates/types/src/post_asap/mod.rs b/crates/types/src/post_asap/mod.rs index 5ce45387..503c5ca1 100644 --- a/crates/types/src/post_asap/mod.rs +++ b/crates/types/src/post_asap/mod.rs @@ -29,15 +29,21 @@ pub mod expr; pub mod guarantee; +pub mod phase; pub mod query_time; pub mod schema; pub mod sketch; -pub use expr::{SummaryExpr, SummaryNode}; +pub use expr::{ExactOperator, SummaryExpr, SummaryNode, ValueOperator}; pub use guarantee::{ AccuracyError, BoundExpr, CompositionOperator, ErrorMetric, GuaranteeSource, ProbabilityExpr, ResultGuarantee, }; +pub use phase::{ + assigned_child_stage, exact_operator_output_schema, produced_availability, + validate_execution_phases, validate_execution_phases_at, ExactOperatorSchemaError, + ExecutionAvailability, PhaseAssignment, PhaseError, +}; pub use query_time::{ classic_cms_sizing, cms_posterior_error_bound, count_sketch_posterior_error_bound, cu_sketch_posterior_error_bound, traditional_a_priori_bound, diff --git a/crates/types/src/post_asap/phase.rs b/crates/types/src/post_asap/phase.rs new file mode 100644 index 00000000..45c68ec3 --- /dev/null +++ b/crates/types/src/post_asap/phase.rs @@ -0,0 +1,867 @@ +//! Execution-phase contract for mixed exact/summary plans (issue #171). +//! +//! A post-ASAP DAG mixes two very different moments of execution: the +//! **update/ingest path** (rows arrive, maintained summary state is updated) +//! and **query evaluation** (maintained state is read out and a final result +//! is produced). A plan that places a query-time residual *underneath* a +//! maintained summary is not merely expensive — it is unexecutable, because +//! the maintenance loop has no readout values to feed into that summary. +//! [`SummaryExpr::ReadoutPostProcess`] is exactly such a residual, which is +//! why it and [`SummaryExpr::UpdateTransform`] are two separate variants +//! rather than one phase-ambiguous value operation. +//! +//! [`ExecutionAvailability`] is what a node's output *is*, at which phase; +//! [`validate_execution_phases`] checks every edge of a DAG against the +//! rules below at plan construction, returning a typed [`PhaseError`] rather +//! than deferring to a runtime failure. +//! +//! ## Edge rules +//! +//! | Parent | Accepts from `child` | +//! |---|---| +//! | `SummaryAgg.child` | `UpdateValue`, or `SummaryState` of an **exact accumulator** family (the one explicitly supported state-composition input — `ExactAggregate` state *is* the value, so it can be re-accumulated on the update path). Never `ReadoutValue`. | +//! | `SummaryEstimate.summary_input` | `SummaryState` (any family). Produces `ReadoutValue`. | +//! | `SummaryJoin.outer/inner` | `UpdateValue` or `SummaryState`; never `ReadoutValue`. | +//! | `SummarySubtract`/`SummaryDelete`/`SummaryMerge` | `SummaryState`. | +//! | `UpdateTransform.child` | `UpdateValue`. Produces `UpdateValue`. | +//! | `ReadoutPostProcess.child` | `ReadoutValue`. Produces `ReadoutValue`. | +//! +//! ## `KeepPreAsap` declares its phase through the derivation +//! +//! A [`SummaryExpr::KeepPreAsap`] leaf is a raw pre-ASAP computation that a +//! runtime can execute at either phase: as update-path raw input beneath a +//! `SummaryAgg`/`UpdateTransform`, or as a query-time fallback beneath a +//! `ReadoutPostProcess` (or at the root). It carries no phase field of its own +//! — every existing consumer pattern-matches the one-field shape — so its +//! phase is *assigned* by [`validate_execution_phases`] from the edge that +//! reaches it and reported in the returned [`PhaseAssignment`]. What it may +//! not do is stay ambiguous inside one mixed plan: the same `Rc` +//! reached once as update input and once as query-time fallback is +//! [`PhaseError::AmbiguousKeepPreAsap`], because no single execution of that +//! subtree can serve both roles. + +use std::collections::HashMap; +use std::rc::Rc; + +use thiserror::Error; + +use super::expr::{ExactOperator, SummaryExpr, SummaryNode, ValueOperator}; +use super::schema::{SummaryFamilyType, SummaryField, SummarySchema}; +use crate::pre_asap::query_expr::{aggregate_output_schema, QueryExprError}; +use crate::pre_asap::schema::{Column, Schema}; + +/// What a post-ASAP node's output is, and at which execution phase it +/// exists — the edge-level contract [`validate_execution_phases`] enforces. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum ExecutionAvailability { + /// Plain rows available on the update/ingest path, while maintaining + /// downstream state. + UpdateValue, + /// Partial, mergeable summary state — not directly readable as a plain + /// value (except for exact accumulators, whose state *is* the value). + SummaryState, + /// Plain values available at query evaluation, after a readout. + ReadoutValue, +} + +impl ExecutionAvailability { + /// Stable lower-case name for JSON/DAG export (`"update_value"`, …). + pub fn as_str(self) -> &'static str { + match self { + Self::UpdateValue => "update_value", + Self::SummaryState => "summary_state", + Self::ReadoutValue => "readout_value", + } + } +} + +impl std::fmt::Display for ExecutionAvailability { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(self.as_str()) + } +} + +/// Which parent/edge a [`PhaseError`] is about — the variant name of the +/// parent `SummaryExpr` plus its field, for a message a plan author can act +/// on. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum PhaseEdge { + SummaryAggChild, + SummaryEstimateInput, + SummaryJoinInput, + SummarySubtractInput, + SummaryDeleteInput, + SummaryMergeInput, + UpdateTransformChild, + ReadoutPostProcessChild, +} + +impl PhaseEdge { + fn describe(self) -> &'static str { + match self { + Self::SummaryAggChild => "SummaryAgg.child", + Self::SummaryEstimateInput => "SummaryEstimate.summary_input", + Self::SummaryJoinInput => "SummaryJoin.{outer,inner}", + Self::SummarySubtractInput => "SummarySubtract.{left,right}", + Self::SummaryDeleteInput => "SummaryDelete.summary_input", + Self::SummaryMergeInput => "SummaryMerge.children[]", + Self::UpdateTransformChild => "UpdateTransform.child", + Self::ReadoutPostProcessChild => "ReadoutPostProcess.child", + } + } +} + +/// A plan-construction-time phase violation. Typed (not a string) so a +/// strategy can degrade to a conservative fallback on the specific variant +/// it expects, and so tests can assert the *reason* a plan was rejected. +#[derive(Debug, Clone, PartialEq, Eq, Error)] +pub enum PhaseError { + /// A query-time value (`SummaryEstimate` / `ReadoutPostProcess` output) + /// placed beneath a maintained summary — the one shape issue #171's + /// phase split exists to make unrepresentable. + #[error( + "readout value under maintenance: {edge} received a {child} input, but a maintained \ + summary can only consume update-path values (or exact accumulator state)" + )] + ReadoutUnderMaintenance { + edge: &'static str, + child: ExecutionAvailability, + }, + /// Any other edge whose child availability the parent does not accept + /// (e.g. plain update rows fed straight into a `SummaryEstimate`, or a + /// sketch's opaque state fed into a `ReadoutPostProcess`). + #[error("{edge} does not accept a {child} input")] + IllegalChildPhase { + edge: &'static str, + child: ExecutionAvailability, + }, + /// A `SummaryAgg` whose child is summary state of a family other than an + /// exact accumulator — re-accumulating opaque sketch/sample/… state on + /// the update path has no defined semantics here. + #[error( + "SummaryAgg.child carries {family} summary state; only exact accumulator state can be \ + composed into another maintained summary" + )] + UnsupportedStateComposition { family: String }, + /// One shared `KeepPreAsap` node reached both as update-path raw input + /// and as a query-time fallback — see the module docs. + #[error( + "KeepPreAsap subtree is phase-ambiguous: reached as {first} and as {second} in the same \ + plan" + )] + AmbiguousKeepPreAsap { + first: ExecutionAvailability, + second: ExecutionAvailability, + }, + /// An update-path-only node (`UpdateTransform`) at the root of a plan: + /// nothing maintains state above it, so its output is never read. + #[error("UpdateTransform cannot be a plan root: its update-path output feeds nothing")] + UpdateValueAtRoot, + /// An `ExactOperator` whose input columns are not all `Plain` at its + /// declared phase. + #[error("exact operator consumes non-plain column {column:?} ({dtype})")] + NonPlainOperand { column: String, dtype: String }, +} + +/// The phase assigned to every node of a validated plan, keyed by +/// `Rc` pointer identity — the explicit per-node "stage" a +/// runtime or a DAG export reads instead of re-deriving it. For every +/// non-`KeepPreAsap` node this equals [`produced_availability`]; for a +/// `KeepPreAsap` leaf it is the phase the reaching edge assigned. +#[derive(Debug, Clone, Default)] +pub struct PhaseAssignment { + stages: HashMap<*const SummaryNode, ExecutionAvailability>, +} + +impl PhaseAssignment { + /// The stage assigned to `node`, if it was part of the validated plan. + pub fn stage_of(&self, node: &Rc) -> Option { + self.stages.get(&Rc::as_ptr(node)).copied() + } + + /// The stage assigned to the node at `ptr` — for callers walking a plan + /// by reference rather than by `Rc`. + pub fn stage_of_ptr(&self, ptr: *const SummaryNode) -> Option { + self.stages.get(&ptr).copied() + } +} + +/// The availability `expr` *produces*, independent of context — `None` for +/// [`SummaryExpr::KeepPreAsap`], whose phase is assigned by the edge reaching +/// it (see the module docs). +pub fn produced_availability(expr: &SummaryExpr) -> Option { + Some(match expr { + SummaryExpr::KeepPreAsap(_) => return None, + SummaryExpr::SummaryAgg { .. } + | SummaryExpr::SummaryJoin { .. } + | SummaryExpr::SummarySubtract { .. } + | SummaryExpr::SummaryDelete { .. } + | SummaryExpr::SummaryMerge { .. } => ExecutionAvailability::SummaryState, + SummaryExpr::SummaryEstimate { .. } | SummaryExpr::ReadoutPostProcess { .. } => { + ExecutionAvailability::ReadoutValue + } + SummaryExpr::UpdateTransform { .. } => ExecutionAvailability::UpdateValue, + }) +} + +/// Is `family` the exact-accumulator family whose partial state *is* the +/// value — the one summary state a `SummaryAgg` may re-accumulate? +fn is_exact_accumulator_state(schema: &SummarySchema) -> Result<(), PhaseError> { + for field in &schema.fields { + match &field.dtype { + SummaryFamilyType::Plain(_) | SummaryFamilyType::ExactAggregate(..) => {} + other => { + return Err(PhaseError::UnsupportedStateComposition { + family: format!("{other:?}"), + }) + } + } + } + Ok(()) +} + +/// Validate every edge of the DAG rooted at `root` against the module-level +/// rules, returning each node's assigned stage on success. Shared +/// `Rc`s are visited once per reaching edge (the assignment is +/// per node, so a conflict between two edges is what +/// [`PhaseError::AmbiguousKeepPreAsap`] detects). +pub fn validate_execution_phases(root: &Rc) -> Result { + // The root may be a readable value or bare maintained state (a + // deployment may hand an `ExactAggregate` accumulator straight to a + // consumer) — only an update-path-only root is meaningless. + let root_stage = match produced_availability(&root.expr) { + None => ExecutionAvailability::ReadoutValue, + Some(ExecutionAvailability::UpdateValue) => return Err(PhaseError::UpdateValueAtRoot), + Some(stage) => stage, + }; + validate_execution_phases_at(root, root_stage) +} + +/// [`validate_execution_phases`] for a *sub*-plan whose root is known to +/// sit at `stage` — e.g. an `UpdateTransform` about to be placed beneath a +/// `SummaryAgg`, which would be rejected as a whole-plan root but is a +/// legal update-path input. Validates every edge beneath `root` exactly +/// as the whole-plan entry point does. +pub fn validate_execution_phases_at( + root: &Rc, + stage: ExecutionAvailability, +) -> Result { + let mut assignment = PhaseAssignment::default(); + visit(root, stage, &mut assignment)?; + Ok(assignment) +} + +/// Record `stage` for `node` (detecting a conflicting earlier assignment +/// for a `KeepPreAsap`), then check and recurse into every child edge. +fn visit( + node: &Rc, + stage: ExecutionAvailability, + assignment: &mut PhaseAssignment, +) -> Result<(), PhaseError> { + let ptr = Rc::as_ptr(node); + if let Some(previous) = assignment.stages.get(&ptr) { + if *previous != stage { + return Err(PhaseError::AmbiguousKeepPreAsap { + first: *previous, + second: stage, + }); + } + // Already validated through another edge with the same stage. + return Ok(()); + } + assignment.stages.insert(ptr, stage); + + match &node.expr { + SummaryExpr::KeepPreAsap(_) => Ok(()), + SummaryExpr::SummaryAgg { child, .. } => { + let child_stage = + child_stage(child, PhaseEdge::SummaryAggChild, |avail| match avail { + ExecutionAvailability::UpdateValue => Ok(()), + ExecutionAvailability::SummaryState => { + is_exact_accumulator_state(&child.schema) + } + ExecutionAvailability::ReadoutValue => { + Err(PhaseError::ReadoutUnderMaintenance { + edge: PhaseEdge::SummaryAggChild.describe(), + child: avail, + }) + } + })?; + visit(child, child_stage, assignment) + } + SummaryExpr::SummaryJoin { outer, inner, .. } => { + for input in [outer, inner] { + let s = child_stage(input, PhaseEdge::SummaryJoinInput, |avail| match avail { + ExecutionAvailability::UpdateValue | ExecutionAvailability::SummaryState => { + Ok(()) + } + ExecutionAvailability::ReadoutValue => { + Err(PhaseError::ReadoutUnderMaintenance { + edge: PhaseEdge::SummaryJoinInput.describe(), + child: avail, + }) + } + })?; + visit(input, s, assignment)?; + } + Ok(()) + } + SummaryExpr::SummarySubtract { left, right } => { + for input in [left, right] { + let s = state_only(input, PhaseEdge::SummarySubtractInput)?; + visit(input, s, assignment)?; + } + Ok(()) + } + SummaryExpr::SummaryDelete { summary_input, .. } => { + let s = state_only(summary_input, PhaseEdge::SummaryDeleteInput)?; + visit(summary_input, s, assignment) + } + SummaryExpr::SummaryMerge { children } => { + for input in children { + let s = state_only(input, PhaseEdge::SummaryMergeInput)?; + visit(input, s, assignment)?; + } + Ok(()) + } + SummaryExpr::SummaryEstimate { summary_input, .. } => { + let s = state_only(summary_input, PhaseEdge::SummaryEstimateInput)?; + visit(summary_input, s, assignment) + } + SummaryExpr::UpdateTransform { child, op } => { + let s = child_stage( + child, + PhaseEdge::UpdateTransformChild, + |avail| match avail { + ExecutionAvailability::UpdateValue => Ok(()), + other => Err(PhaseError::IllegalChildPhase { + edge: PhaseEdge::UpdateTransformChild.describe(), + child: other, + }), + }, + )?; + check_plain_operands(op, &child.schema)?; + visit(child, s, assignment) + } + SummaryExpr::ReadoutPostProcess { child, op } => { + let s = child_stage( + child, + PhaseEdge::ReadoutPostProcessChild, + |avail| match avail { + ExecutionAvailability::ReadoutValue => Ok(()), + other => Err(PhaseError::IllegalChildPhase { + edge: PhaseEdge::ReadoutPostProcessChild.describe(), + child: other, + }), + }, + )?; + check_plain_operands(op, &child.schema)?; + visit(child, s, assignment) + } + } +} + +/// The stage `child` takes as a direct input of `parent`, without +/// validating legality — `child`'s own produced availability, or for a +/// `KeepPreAsap` leaf the phase `parent`'s edge assigns it (update-path raw +/// input under maintenance/transform edges, query-time fallback under a +/// post-process, and — meaninglessly, but for a stable answer — `UpdateValue` +/// under a state-only edge). For DAG export and other reporting that needs +/// an explicit per-node stage even on a plan that +/// [`validate_execution_phases`] would reject. +pub fn assigned_child_stage(parent: &SummaryExpr, child: &SummaryNode) -> ExecutionAvailability { + if let Some(avail) = produced_availability(&child.expr) { + return avail; + } + match parent { + SummaryExpr::ReadoutPostProcess { .. } => ExecutionAvailability::ReadoutValue, + SummaryExpr::KeepPreAsap(_) + | SummaryExpr::SummaryAgg { .. } + | SummaryExpr::SummaryJoin { .. } + | SummaryExpr::SummarySubtract { .. } + | SummaryExpr::SummaryDelete { .. } + | SummaryExpr::SummaryEstimate { .. } + | SummaryExpr::SummaryMerge { .. } + | SummaryExpr::UpdateTransform { .. } => ExecutionAvailability::UpdateValue, + } +} + +/// The stage `child` takes on `edge`: its own produced availability +/// (checked via `accept`), or — for a `KeepPreAsap` leaf — the phase the +/// edge assigns it, derived from what that edge accepts. +fn child_stage( + child: &Rc, + edge: PhaseEdge, + accept: impl Fn(ExecutionAvailability) -> Result<(), PhaseError>, +) -> Result { + match produced_availability(&child.expr) { + Some(avail) => { + accept(avail)?; + Ok(avail) + } + None => { + // A raw pre-ASAP subtree executes at whichever phase its consumer + // needs: update-path input for maintenance/transform edges, + // query-time fallback for a post-process edge. State-only edges + // can't consume plain rows at all. + let assigned = match edge { + PhaseEdge::SummaryAggChild + | PhaseEdge::SummaryJoinInput + | PhaseEdge::UpdateTransformChild => ExecutionAvailability::UpdateValue, + PhaseEdge::ReadoutPostProcessChild => ExecutionAvailability::ReadoutValue, + PhaseEdge::SummaryEstimateInput + | PhaseEdge::SummarySubtractInput + | PhaseEdge::SummaryDeleteInput + | PhaseEdge::SummaryMergeInput => { + return Err(PhaseError::IllegalChildPhase { + edge: edge.describe(), + child: ExecutionAvailability::UpdateValue, + }) + } + }; + accept(assigned)?; + Ok(assigned) + } + } +} + +fn state_only( + child: &Rc, + edge: PhaseEdge, +) -> Result { + child_stage(child, edge, |avail| match avail { + ExecutionAvailability::SummaryState => Ok(()), + other => Err(PhaseError::IllegalChildPhase { + edge: edge.describe(), + child: other, + }), + }) +} + +/// The exact operator must consume only `Plain` columns of its input: for +/// an `Aggregate` payload, every grouping key and every measure's input +/// column. +fn check_plain_operands(op: &ValueOperator, input: &SummarySchema) -> Result<(), PhaseError> { + let ValueOperator::Exact(op) = op else { + return check_all_plain(input); + }; + let ExactOperator::Aggregate { + reduction, + measures, + .. + } = op; + let mut referenced: Vec = reduction + .group_keys() + .map(|keys| keys.keys().to_vec()) + .unwrap_or_default(); + for m in measures { + if let Some(col) = m.input_col() { + referenced.push(col); + } + } + // With no explicit input column (the PromQL sample-value convention) + // the operator reads every non-key column, so all must be plain. + let implicit = measures.iter().any(|m| m.input_col().is_none()); + for (i, field) in input.fields.iter().enumerate() { + if !(implicit || referenced.contains(&i)) { + continue; + } + if !matches!(field.dtype, SummaryFamilyType::Plain(_)) { + return Err(PhaseError::NonPlainOperand { + column: field.name.clone(), + dtype: format!("{:?}", field.dtype), + }); + } + } + Ok(()) +} + +fn check_all_plain(input: &SummarySchema) -> Result<(), PhaseError> { + for field in &input.fields { + if !matches!(field.dtype, SummaryFamilyType::Plain(_)) { + return Err(PhaseError::NonPlainOperand { + column: field.name.clone(), + dtype: format!("{:?}", field.dtype), + }); + } + } + Ok(()) +} + +/// The plain pre-ASAP `Schema` underlying an all-`Plain` `SummarySchema`, or +/// `None` if any column carries summary state. +pub fn plain_schema(schema: &SummarySchema) -> Option { + let mut columns = Vec::with_capacity(schema.fields.len()); + for field in &schema.fields { + let SummaryFamilyType::Plain(dtype) = &field.dtype else { + return None; + }; + columns.push(Column::new(&field.name, dtype.clone(), field.nullable)); + } + Some(Schema { + columns, + time_index: schema.time_index, + unique_keys: Vec::new(), + closed: true, + }) +} + +/// Lift a plain pre-ASAP schema to a `SummarySchema` with every column +/// `Plain` — the output of every exact operator. +pub fn lift_plain(schema: &Schema) -> SummarySchema { + SummarySchema { + fields: schema + .columns + .iter() + .map(|c| SummaryField { + name: c.name.clone(), + dtype: SummaryFamilyType::Plain(c.dtype.clone()), + nullable: c.nullable, + }) + .collect(), + time_index: schema.time_index, + } +} + +/// Output schema of `op` applied to a child whose edge carries `input` — +/// the same canonical derivation the pre-ASAP `Aggregate` node uses, so an +/// exact `ReadoutPostProcess`/`UpdateTransform` never disagrees with the pre-ASAP +/// target it was lowered from. `Err` when the child carries non-plain +/// state the operator cannot read. +pub fn exact_operator_output_schema( + op: &ExactOperator, + input: &SummarySchema, +) -> Result { + let plain = plain_schema(input).ok_or(ExactOperatorSchemaError::NonPlainInput)?; + let ExactOperator::Aggregate { + reduction, + measures, + output_names, + .. + } = op; + let out = aggregate_output_schema(&plain, reduction, measures, output_names)?; + Ok(lift_plain(&out)) +} + +/// Why [`exact_operator_output_schema`] could not derive a schema. +#[derive(Debug, Error)] +pub enum ExactOperatorSchemaError { + #[error("exact operator input carries summary state, not plain columns")] + NonPlainInput, + #[error("schema derivation failed: {0}")] + Schema(#[from] QueryExprError), +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::post_asap::{ExactKind, ExactParams, GroupingStrategy, SketchQuery}; + use crate::pre_asap::agg_intent::AggIntent; + use crate::pre_asap::expr_ir::ColumnRef; + use crate::pre_asap::query_expr::{QueryExpr, Reduction, Source}; + use crate::pre_asap::schema::DataType; + + fn scan() -> Rc { + Rc::new(QueryExpr::Scan { + source: Source::TimeSeries { metric: "m".into() }, + predicates: vec![], + schema: Schema::with_time_index( + vec![ + Column::new("ts", DataType::Timestamp, false), + Column::new("value", DataType::Float64, false), + Column::new("zone", DataType::Utf8, true), + ], + 0, + vec![], + ), + }) + } + + fn keep() -> Rc { + let s = scan(); + let schema = lift_plain(&s.output_schema().unwrap()); + Rc::new(SummaryNode { + expr: SummaryExpr::KeepPreAsap(s), + schema, + guarantee: None, + }) + } + + fn plain(names: &[&str]) -> SummarySchema { + SummarySchema { + fields: names + .iter() + .map(|n| SummaryField { + name: (*n).into(), + dtype: SummaryFamilyType::Plain(DataType::Float64), + nullable: false, + }) + .collect(), + time_index: None, + } + } + + fn agg(child: Rc, family: SummaryFamilyType) -> Rc { + Rc::new(SummaryNode { + expr: SummaryExpr::SummaryAgg { + child, + family: family.clone(), + col: ColumnRef::SampleValue, + reduction: Reduction::by(vec![]), + grouping: GroupingStrategy::default(), + }, + schema: SummarySchema { + fields: vec![SummaryField { + name: "state".into(), + dtype: family, + nullable: false, + }], + time_index: None, + }, + guarantee: None, + }) + } + + fn kll() -> SummaryFamilyType { + use crate::post_asap::{SketchAlgorithm, SketchKind, SketchParams}; + SummaryFamilyType::Sketch( + SketchKind::new(SketchAlgorithm::Kll, SketchParams::Kll { k: 200 }), + GroupingStrategy::default(), + ) + } + + fn estimate(child: Rc) -> Rc { + Rc::new(SummaryNode { + expr: SummaryExpr::SummaryEstimate { + summary_input: child, + query: SketchQuery::Quantile { q: 0.99 }, + }, + schema: plain(&["quantile_0_99"]), + guarantee: None, + }) + } + + fn max_op() -> ExactOperator { + ExactOperator::Aggregate { + reduction: Reduction::by(vec![]), + measures: vec![AggIntent::Max { col: None }], + output_names: vec![], + having: None, + } + } + + #[test] + fn keep_pre_asap_under_summary_agg_is_update_input() { + let leaf = keep(); + let root = agg(Rc::clone(&leaf), kll()); + let assignment = validate_execution_phases(&root).unwrap(); + assert_eq!( + assignment.stage_of(&leaf), + Some(ExecutionAvailability::UpdateValue) + ); + assert_eq!( + assignment.stage_of(&root), + Some(ExecutionAvailability::SummaryState) + ); + } + + #[test] + fn exact_accumulator_state_may_feed_another_summary_agg() { + let inner = agg( + keep(), + SummaryFamilyType::ExactAggregate(ExactKind::Sum, ExactParams::Sum), + ); + let root = estimate(agg(inner, kll())); + assert!(validate_execution_phases(&root).is_ok()); + } + + #[test] + fn readout_under_summary_agg_is_rejected() { + let inner = estimate(agg(keep(), kll())); + let root = agg(inner, kll()); + assert!(matches!( + validate_execution_phases(&root), + Err(PhaseError::ReadoutUnderMaintenance { .. }) + )); + } + + #[test] + fn post_process_over_readout_is_legal_and_root_is_readout() { + let inner = estimate(agg(keep(), kll())); + let root = Rc::new(SummaryNode { + expr: SummaryExpr::ReadoutPostProcess { + child: inner, + op: ValueOperator::Exact(max_op()), + }, + schema: plain(&["max"]), + guarantee: None, + }); + let assignment = validate_execution_phases(&root).unwrap(); + assert_eq!( + assignment.stage_of(&root), + Some(ExecutionAvailability::ReadoutValue) + ); + } + + #[test] + fn non_exact_operator_uses_the_same_readout_phase_contract() { + let inner = estimate(agg(keep(), kll())); + let root = Rc::new(SummaryNode { + expr: SummaryExpr::ReadoutPostProcess { + child: inner, + op: ValueOperator::Extension { + name: "approximate_calibration".into(), + }, + }, + schema: plain(&["calibrated"]), + guarantee: None, + }); + + let assignment = validate_execution_phases(&root).unwrap(); + assert_eq!( + assignment.stage_of(&root), + Some(ExecutionAvailability::ReadoutValue) + ); + } + + #[test] + fn post_process_under_summary_agg_is_rejected() { + let inner = estimate(agg(keep(), kll())); + let post = Rc::new(SummaryNode { + expr: SummaryExpr::ReadoutPostProcess { + child: inner, + op: ValueOperator::Exact(max_op()), + }, + schema: plain(&["max"]), + guarantee: None, + }); + let root = agg(post, kll()); + assert_eq!( + validate_execution_phases(&root).err(), + Some(PhaseError::ReadoutUnderMaintenance { + edge: "SummaryAgg.child", + child: ExecutionAvailability::ReadoutValue, + }) + ); + } + + #[test] + fn transform_under_summary_agg_is_legal_but_not_at_root() { + let transform = Rc::new(SummaryNode { + expr: SummaryExpr::UpdateTransform { + child: keep(), + op: ValueOperator::Exact(max_op()), + }, + schema: plain(&["max"]), + guarantee: None, + }); + assert_eq!( + validate_execution_phases(&transform).err(), + Some(PhaseError::UpdateValueAtRoot) + ); + let root = estimate(agg(Rc::clone(&transform), kll())); + let assignment = validate_execution_phases(&root).unwrap(); + assert_eq!( + assignment.stage_of(&transform), + Some(ExecutionAvailability::UpdateValue) + ); + } + + #[test] + fn transform_over_readout_is_rejected() { + let inner = estimate(agg(keep(), kll())); + let transform = Rc::new(SummaryNode { + expr: SummaryExpr::UpdateTransform { + child: inner, + op: ValueOperator::Exact(max_op()), + }, + schema: plain(&["max"]), + guarantee: None, + }); + let root = agg(transform, kll()); + assert!(matches!( + validate_execution_phases(&root), + Err(PhaseError::IllegalChildPhase { + edge: "UpdateTransform.child", + child: ExecutionAvailability::ReadoutValue + }) + )); + } + + #[test] + fn a_shared_keep_pre_asap_reached_at_two_phases_is_ambiguous() { + // One raw subtree used both as update input (under a SummaryAgg) and + // as a query-time fallback (under an ExactPostProcess) — no single + // execution can serve both, so the plan is rejected. + let shared = keep(); + let maintained = estimate(agg(Rc::clone(&shared), kll())); + let post_over_raw = Rc::new(SummaryNode { + expr: SummaryExpr::ReadoutPostProcess { + child: Rc::clone(&shared), + op: ValueOperator::Exact(max_op()), + }, + schema: plain(&["max"]), + guarantee: None, + }); + let root = Rc::new(SummaryNode { + expr: SummaryExpr::SummaryMerge { + children: vec![ + Rc::new(SummaryNode { + expr: SummaryExpr::ReadoutPostProcess { + child: maintained, + op: ValueOperator::Exact(max_op()), + }, + schema: plain(&["max"]), + guarantee: None, + }), + post_over_raw, + ], + }, + schema: plain(&["max"]), + guarantee: None, + }); + // SummaryMerge only accepts state, so this fails earlier for a + // different reason; probe the ambiguity through a direct visit. + let mut assignment = PhaseAssignment::default(); + visit(&shared, ExecutionAvailability::UpdateValue, &mut assignment).unwrap(); + assert_eq!( + visit( + &shared, + ExecutionAvailability::ReadoutValue, + &mut assignment + ), + Err(PhaseError::AmbiguousKeepPreAsap { + first: ExecutionAvailability::UpdateValue, + second: ExecutionAvailability::ReadoutValue, + }) + ); + assert!(validate_execution_phases(&root).is_err()); + } + + #[test] + fn exact_operator_schema_matches_pre_asap_aggregate_derivation() { + let child_schema = lift_plain(&scan().output_schema().unwrap()); + let op = ExactOperator::Aggregate { + reduction: Reduction::by(vec![2]), + measures: vec![AggIntent::Max { col: None }], + output_names: vec![], + having: None, + }; + let out = exact_operator_output_schema(&op, &child_schema).unwrap(); + let names: Vec<_> = out.fields.iter().map(|f| f.name.as_str()).collect(); + assert_eq!(names, vec!["zone", "max"]); + assert!(out + .fields + .iter() + .all(|f| matches!(f.dtype, SummaryFamilyType::Plain(_)))); + } + + #[test] + fn exact_operator_rejects_non_plain_input() { + let state = agg(keep(), kll()); + assert!(matches!( + exact_operator_output_schema(&max_op(), &state.schema), + Err(ExactOperatorSchemaError::NonPlainInput) + )); + } +} From 22e89981c401bf50725feb363f71d7c799c6c327 Mon Sep 17 00:00:00 2001 From: zz_y Date: Sun, 30 Aug 2026 12:09:01 -0600 Subject: [PATCH 2/4] refactor(post-asap): split value domain into timing and primitive --- crates/types/src/dag_export.rs | 58 +-- crates/types/src/post_asap/expr.rs | 10 +- crates/types/src/post_asap/mod.rs | 12 +- .../post_asap/{phase.rs => value_domain.rs} | 431 +++++++++--------- 4 files changed, 261 insertions(+), 250 deletions(-) rename crates/types/src/post_asap/{phase.rs => value_domain.rs} (65%) diff --git a/crates/types/src/dag_export.rs b/crates/types/src/dag_export.rs index bda2196f..d2a7b24d 100644 --- a/crates/types/src/dag_export.rs +++ b/crates/types/src/dag_export.rs @@ -14,7 +14,7 @@ //! //! This is literally the same hashing //! [`share_common_subtrees`](crate::pre_asap::cse::share_common_subtrees) -//! uses to bucket candidates in its `InternTable` (issue #223 stage 3) — not +//! uses to bucket candidates in its `InternTable` (issue #223 domain 3) — not //! a parallel reimplementation. `tools/dag-viewer`'s "shared subtree" //! highlighting is still a *proxy* for real CSE, though: a hash match here //! only means two nodes are legal `InternTable` bucket-mates (same coarse @@ -47,8 +47,8 @@ use std::rc::Rc; use serde::Serialize; use crate::post_asap::{ - assigned_child_stage, produced_availability, AccuracyError, ExactOperator, - ExecutionAvailability, ResultGuarantee, SummaryExpr, SummaryNode, ValueOperator, + assigned_child_domain, produced_domain, AccuracyError, ExactOperator, ResultGuarantee, + SummaryExpr, SummaryNode, ValueDomain, ValueOperator, }; use crate::pre_asap::cse::{structural_hash, HashCache}; use crate::pre_asap::query_expr::{QueryExpr, Source}; @@ -359,16 +359,16 @@ pub struct SummaryDagGraph { /// top-level `DagNode::kind`. pub fn export_summary(node: &SummaryNode) -> SummaryDagGraph { let mut nodes = Vec::new(); - let root = build_summary(node, &mut nodes, root_stage(node)); + let root = build_summary(node, &mut nodes, root_domain(node)); SummaryDagGraph { nodes, root } } -/// The explicit execution stage of an exported plan's root — its own -/// produced availability, or query-time readout for a bare `KeepPreAsap` -/// (the same convention `post_asap::phase::validate_execution_phases` +/// The explicit execution domain of an exported plan's root — its own +/// produced domain, or query-time readout for a bare `KeepPreAsap` +/// (the same convention `post_asap::value_domain::validate_execution_domains` /// uses for a root). -fn root_stage(node: &SummaryNode) -> ExecutionAvailability { - produced_availability(&node.expr).unwrap_or(ExecutionAvailability::ReadoutValue) +fn root_domain(node: &SummaryNode) -> ValueDomain { + produced_domain(&node.expr).unwrap_or(ValueDomain::READ_ROWS) } /// `detail` for an [`ExactOperator`] payload — its own fields, rendered the @@ -466,13 +466,13 @@ fn family_label(family: &crate::post_asap::SummaryFamilyType) -> String { /// apart on how every *other* variant's own shape is described, since /// nothing about that description differs between the two. /// -/// `stage` is the node's explicit execution phase (issue #171) — its own -/// [`produced_availability`], or the edge-assigned phase for a `KeepPreAsap` -/// — and is written into `detail.stage` on every post-ASAP node so a viewer +/// `domain` is the node's explicit execution domain (issue #171) — its own +/// [`produced_domain`], or the edge-assigned domain for a `KeepPreAsap` +/// — and is written into `detail.domain` on every post-ASAP node so a viewer /// reads it rather than inferring it from the node's kind. fn summary_shape( expr: &SummaryExpr, - stage: ExecutionAvailability, + domain: ValueDomain, ) -> (&'static str, String, serde_json::Value) { let (kind, label, mut detail) = match expr { SummaryExpr::KeepPreAsap(_) => { @@ -531,8 +531,11 @@ fn summary_shape( }; if let serde_json::Value::Object(map) = &mut detail { map.insert( - "stage".into(), - serde_json::Value::String(stage.as_str().into()), + "domain".into(), + serde_json::json!({ + "timing": domain.timing.as_str(), + "primitive": domain.primitive.as_str(), + }), ); } (kind, label, detail) @@ -563,18 +566,17 @@ fn summary_children(expr: &SummaryExpr) -> Vec<&Rc> { /// post-order (children pushed before their parent), and return the pushed /// root's id. Exhaustive over every [`SummaryExpr`] variant, matching this /// file's own exhaustive style for `QueryExpr` in [`build`]. -fn build_summary( - node: &SummaryNode, - nodes: &mut Vec, - stage: ExecutionAvailability, -) -> u32 { +fn build_summary(node: &SummaryNode, nodes: &mut Vec, domain: ValueDomain) -> u32 { if let SummaryExpr::KeepPreAsap(inner) = &node.expr { let pre_asap_subgraph = export(inner); let inner_kind = pre_asap_subgraph.nodes[pre_asap_subgraph.root as usize].kind; let label = format!("KeepPreAsap({inner_kind})"); let detail = serde_json::json!({ "pre_asap_subgraph": pre_asap_subgraph, - "stage": stage.as_str(), + "domain": { + "timing": domain.timing.as_str(), + "primitive": domain.primitive.as_str(), + }, }); return push_summary_node( nodes, @@ -587,9 +589,9 @@ fn build_summary( } let children: Vec = summary_children(&node.expr) .into_iter() - .map(|child| build_summary(child, nodes, assigned_child_stage(&node.expr, child))) + .map(|child| build_summary(child, nodes, assigned_child_domain(&node.expr, child))) .collect(); - let (kind, label, detail) = summary_shape(&node.expr, stage); + let (kind, label, detail) = summary_shape(&node.expr, domain); push_summary_node(nodes, kind, label, detail, children, node.guarantee.clone()) } @@ -867,7 +869,7 @@ fn build_summary_hybrid( nodes: &mut Vec, cache: &mut HashCache, find_winner: &mut dyn FnMut(&QueryExpr) -> Option, - stage: ExecutionAvailability, + domain: ValueDomain, ) -> u32 { if let SummaryExpr::KeepPreAsap(inner) = &node.expr { return build(inner, nodes, cache, find_winner); @@ -880,11 +882,11 @@ fn build_summary_hybrid( nodes, cache, find_winner, - assigned_child_stage(&node.expr, child), + assigned_child_domain(&node.expr, child), ) }) .collect(); - let (kind, label, mut detail) = summary_shape(&node.expr, stage); + let (kind, label, mut detail) = summary_shape(&node.expr, domain); // The merged graph's `DagNode` has no dedicated guarantee field (it is // the pre-ASAP node shape); the guarantee rides in `detail` under the // same key/shape `SummaryDagNode::guarantee` uses, additively. @@ -975,7 +977,7 @@ fn build( nodes, cache, find_winner, - root_stage(&replacement), + root_domain(&replacement), ); for node in &mut nodes[first..] { if node.decision.is_none() { @@ -1536,7 +1538,7 @@ mod tests { ); } - // ── Issue #223 stage 3: dag_export's hash literally *is* cse's hash ──── + // ── Issue #223 domain 3: dag_export's hash literally *is* cse's hash ──── #[test] fn root_hash_matches_cse_structural_hash_for_the_same_node() { diff --git a/crates/types/src/post_asap/expr.rs b/crates/types/src/post_asap/expr.rs index 3eda8f82..aedd0ba8 100644 --- a/crates/types/src/post_asap/expr.rs +++ b/crates/types/src/post_asap/expr.rs @@ -10,8 +10,8 @@ use crate::pre_asap::{ColumnRef, QueryExpr, Reduction}; // ── Exact operators composed with summary plans (issue #171) ──────────────── /// An exact, plain-row operator that a mixed exact/summary plan executes at -/// an explicit phase. Exact composition is one producer of the generic -/// [`ValueOperator`] phase payload. +/// an explicit domain. Exact composition is one producer of the generic +/// [`ValueOperator`] domain payload. /// /// Deliberately **not** an intact pre-ASAP [`QueryExpr`] subtree: a /// `QueryExpr`'s children are always `Rc`, so embedding one here @@ -41,7 +41,7 @@ pub enum ExactOperator { }, } -/// An operation over values at a declared execution phase. +/// An operation over values at a declared execution domain. /// /// Phase placement is independent of whether the operation is exact or /// approximate: [`SummaryExpr::UpdateTransform`] and @@ -192,7 +192,7 @@ pub enum SummaryExpr { /// produces plain update values, so its output may feed a downstream /// [`SummaryAgg`](SummaryExpr::SummaryAgg)'s maintenance — the "outer /// summary over an inner non-accumulator exact transform" direction. - /// See [`super::phase::ExecutionAvailability`] for the edge contract. + /// See [`super::value_domain::ValueDomain`] for the edge contract. UpdateTransform { child: Rc, op: ValueOperator, @@ -203,7 +203,7 @@ pub enum SummaryExpr { /// final plain query result — the "outer exact fold over an inner /// summary readout" direction. Can never feed maintained state: a /// `SummaryAgg` above one of these is a plan-time - /// [`super::phase::PhaseError`], never a runtime failure. + /// [`super::value_domain::DomainError`], never a runtime failure. ReadoutPostProcess { child: Rc, op: ValueOperator, diff --git a/crates/types/src/post_asap/mod.rs b/crates/types/src/post_asap/mod.rs index 503c5ca1..5ac2864b 100644 --- a/crates/types/src/post_asap/mod.rs +++ b/crates/types/src/post_asap/mod.rs @@ -29,21 +29,16 @@ pub mod expr; pub mod guarantee; -pub mod phase; pub mod query_time; pub mod schema; pub mod sketch; +pub mod value_domain; pub use expr::{ExactOperator, SummaryExpr, SummaryNode, ValueOperator}; pub use guarantee::{ AccuracyError, BoundExpr, CompositionOperator, ErrorMetric, GuaranteeSource, ProbabilityExpr, ResultGuarantee, }; -pub use phase::{ - assigned_child_stage, exact_operator_output_schema, produced_availability, - validate_execution_phases, validate_execution_phases_at, ExactOperatorSchemaError, - ExecutionAvailability, PhaseAssignment, PhaseError, -}; pub use query_time::{ classic_cms_sizing, cms_posterior_error_bound, count_sketch_posterior_error_bound, cu_sketch_posterior_error_bound, traditional_a_priori_bound, @@ -54,3 +49,8 @@ pub use sketch::{ HydraParams, SamplingKind, SamplingParams, SketchAlgorithm, SketchCategory, SketchKind, SketchParams, SketchQuery, StatModelKind, StatModelParams, WaveletKind, WaveletParams, }; +pub use value_domain::{ + assigned_child_domain, exact_operator_output_schema, produced_domain, + validate_execution_domains, validate_execution_domains_at, DataPrimitive, DomainAssignment, + DomainError, ExactOperatorSchemaError, ExecutionTiming, ValueDomain, +}; diff --git a/crates/types/src/post_asap/phase.rs b/crates/types/src/post_asap/value_domain.rs similarity index 65% rename from crates/types/src/post_asap/phase.rs rename to crates/types/src/post_asap/value_domain.rs index 45c68ec3..c65d705b 100644 --- a/crates/types/src/post_asap/phase.rs +++ b/crates/types/src/post_asap/value_domain.rs @@ -1,4 +1,4 @@ -//! Execution-phase contract for mixed exact/summary plans (issue #171). +//! Execution-domain contract for mixed exact/summary plans (issue #171). //! //! A post-ASAP DAG mixes two very different moments of execution: the //! **update/ingest path** (rows arrive, maintained summary state is updated) @@ -8,36 +8,36 @@ //! the maintenance loop has no readout values to feed into that summary. //! [`SummaryExpr::ReadoutPostProcess`] is exactly such a residual, which is //! why it and [`SummaryExpr::UpdateTransform`] are two separate variants -//! rather than one phase-ambiguous value operation. +//! rather than one domain-ambiguous value operation. //! -//! [`ExecutionAvailability`] is what a node's output *is*, at which phase; -//! [`validate_execution_phases`] checks every edge of a DAG against the -//! rules below at plan construction, returning a typed [`PhaseError`] rather +//! [`ValueDomain`] is what a node's output *is*, at which domain; +//! [`validate_execution_domains`] checks every edge of a DAG against the +//! rules below at plan construction, returning a typed [`DomainError`] rather //! than deferring to a runtime failure. //! //! ## Edge rules //! //! | Parent | Accepts from `child` | //! |---|---| -//! | `SummaryAgg.child` | `UpdateValue`, or `SummaryState` of an **exact accumulator** family (the one explicitly supported state-composition input — `ExactAggregate` state *is* the value, so it can be re-accumulated on the update path). Never `ReadoutValue`. | -//! | `SummaryEstimate.summary_input` | `SummaryState` (any family). Produces `ReadoutValue`. | -//! | `SummaryJoin.outer/inner` | `UpdateValue` or `SummaryState`; never `ReadoutValue`. | -//! | `SummarySubtract`/`SummaryDelete`/`SummaryMerge` | `SummaryState`. | -//! | `UpdateTransform.child` | `UpdateValue`. Produces `UpdateValue`. | -//! | `ReadoutPostProcess.child` | `ReadoutValue`. Produces `ReadoutValue`. | +//! | `SummaryAgg.child` | `MAINTENANCE_ROWS`, or `MAINTENANCE_SUMMARY` of an **exact accumulator** family. Never a read-time domain. | +//! | `SummaryEstimate.summary_input` | `MAINTENANCE_SUMMARY` (any family). Produces `READ_ROWS`. | +//! | `SummaryJoin.outer/inner` | `MAINTENANCE_ROWS` or `MAINTENANCE_SUMMARY`; never a read-time domain. | +//! | `SummarySubtract`/`SummaryDelete`/`SummaryMerge` | `MAINTENANCE_SUMMARY`. | +//! | `UpdateTransform.child` | `MAINTENANCE_ROWS`. Produces `MAINTENANCE_ROWS`. | +//! | `ReadoutPostProcess.child` | `READ_ROWS`. Produces `READ_ROWS`. | //! -//! ## `KeepPreAsap` declares its phase through the derivation +//! ## `KeepPreAsap` declares its domain through the derivation //! //! A [`SummaryExpr::KeepPreAsap`] leaf is a raw pre-ASAP computation that a -//! runtime can execute at either phase: as update-path raw input beneath a +//! runtime can execute at either domain: as update-path raw input beneath a //! `SummaryAgg`/`UpdateTransform`, or as a query-time fallback beneath a -//! `ReadoutPostProcess` (or at the root). It carries no phase field of its own +//! `ReadoutPostProcess` (or at the root). It carries no domain field of its own //! — every existing consumer pattern-matches the one-field shape — so its -//! phase is *assigned* by [`validate_execution_phases`] from the edge that -//! reaches it and reported in the returned [`PhaseAssignment`]. What it may +//! domain is *assigned* by [`validate_execution_domains`] from the edge that +//! reaches it and reported in the returned [`DomainAssignment`]. What it may //! not do is stay ambiguous inside one mixed plan: the same `Rc` //! reached once as update input and once as query-time fallback is -//! [`PhaseError::AmbiguousKeepPreAsap`], because no single execution of that +//! [`DomainError::AmbiguousKeepPreAsap`], because no single execution of that //! subtree can serve both roles. use std::collections::HashMap; @@ -50,42 +50,72 @@ use super::schema::{SummaryFamilyType, SummaryField, SummarySchema}; use crate::pre_asap::query_expr::{aggregate_output_schema, QueryExprError}; use crate::pre_asap::schema::{Column, Schema}; -/// What a post-ASAP node's output is, and at which execution phase it -/// exists — the edge-level contract [`validate_execution_phases`] enforces. +/// When a post-ASAP value is produced. #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] -pub enum ExecutionAvailability { - /// Plain rows available on the update/ingest path, while maintaining - /// downstream state. - UpdateValue, - /// Partial, mergeable summary state — not directly readable as a plain - /// value (except for exact accumulators, whose state *is* the value). +pub enum ExecutionTiming { + MaintenanceTime, + ReadTime, +} + +impl ExecutionTiming { + pub fn as_str(self) -> &'static str { + match self { + Self::MaintenanceTime => "maintenance_time", + Self::ReadTime => "read_time", + } + } +} + +/// The primitive representation carried by a post-ASAP edge. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum DataPrimitive { + Rows, SummaryState, - /// Plain values available at query evaluation, after a readout. - ReadoutValue, } -impl ExecutionAvailability { - /// Stable lower-case name for JSON/DAG export (`"update_value"`, …). +impl DataPrimitive { pub fn as_str(self) -> &'static str { match self { - Self::UpdateValue => "update_value", + Self::Rows => "rows", Self::SummaryState => "summary_state", - Self::ReadoutValue => "readout_value", } } } -impl std::fmt::Display for ExecutionAvailability { +/// The two-dimensional edge contract: when a value exists and which data +/// primitive it carries. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub struct ValueDomain { + pub timing: ExecutionTiming, + pub primitive: DataPrimitive, +} + +impl ValueDomain { + pub const MAINTENANCE_ROWS: Self = Self { + timing: ExecutionTiming::MaintenanceTime, + primitive: DataPrimitive::Rows, + }; + pub const MAINTENANCE_SUMMARY: Self = Self { + timing: ExecutionTiming::MaintenanceTime, + primitive: DataPrimitive::SummaryState, + }; + pub const READ_ROWS: Self = Self { + timing: ExecutionTiming::ReadTime, + primitive: DataPrimitive::Rows, + }; +} + +impl std::fmt::Display for ValueDomain { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.write_str(self.as_str()) + write!(f, "{}/{}", self.timing.as_str(), self.primitive.as_str()) } } -/// Which parent/edge a [`PhaseError`] is about — the variant name of the +/// Which parent/edge a [`DomainError`] is about — the variant name of the /// parent `SummaryExpr` plus its field, for a message a plan author can act /// on. #[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum PhaseEdge { +pub enum DomainEdge { SummaryAggChild, SummaryEstimateInput, SummaryJoinInput, @@ -96,7 +126,7 @@ pub enum PhaseEdge { ReadoutPostProcessChild, } -impl PhaseEdge { +impl DomainEdge { fn describe(self) -> &'static str { match self { Self::SummaryAggChild => "SummaryAgg.child", @@ -111,29 +141,29 @@ impl PhaseEdge { } } -/// A plan-construction-time phase violation. Typed (not a string) so a +/// A plan-construction-time domain violation. Typed (not a string) so a /// strategy can degrade to a conservative fallback on the specific variant /// it expects, and so tests can assert the *reason* a plan was rejected. #[derive(Debug, Clone, PartialEq, Eq, Error)] -pub enum PhaseError { +pub enum DomainError { /// A query-time value (`SummaryEstimate` / `ReadoutPostProcess` output) /// placed beneath a maintained summary — the one shape issue #171's - /// phase split exists to make unrepresentable. + /// domain split exists to make unrepresentable. #[error( "readout value under maintenance: {edge} received a {child} input, but a maintained \ summary can only consume update-path values (or exact accumulator state)" )] ReadoutUnderMaintenance { edge: &'static str, - child: ExecutionAvailability, + child: ValueDomain, }, - /// Any other edge whose child availability the parent does not accept + /// Any other edge whose child domain the parent does not accept /// (e.g. plain update rows fed straight into a `SummaryEstimate`, or a /// sketch's opaque state fed into a `ReadoutPostProcess`). #[error("{edge} does not accept a {child} input")] IllegalChildPhase { edge: &'static str, - child: ExecutionAvailability, + child: ValueDomain, }, /// A `SummaryAgg` whose child is summary state of a family other than an /// exact accumulator — re-accumulating opaque sketch/sample/… state on @@ -146,72 +176,72 @@ pub enum PhaseError { /// One shared `KeepPreAsap` node reached both as update-path raw input /// and as a query-time fallback — see the module docs. #[error( - "KeepPreAsap subtree is phase-ambiguous: reached as {first} and as {second} in the same \ + "KeepPreAsap subtree is domain-ambiguous: reached as {first} and as {second} in the same \ plan" )] AmbiguousKeepPreAsap { - first: ExecutionAvailability, - second: ExecutionAvailability, + first: ValueDomain, + second: ValueDomain, }, /// An update-path-only node (`UpdateTransform`) at the root of a plan: /// nothing maintains state above it, so its output is never read. #[error("UpdateTransform cannot be a plan root: its update-path output feeds nothing")] - UpdateValueAtRoot, + MaintenanceRowsAtRoot, /// An `ExactOperator` whose input columns are not all `Plain` at its - /// declared phase. + /// declared domain. #[error("exact operator consumes non-plain column {column:?} ({dtype})")] NonPlainOperand { column: String, dtype: String }, } -/// The phase assigned to every node of a validated plan, keyed by -/// `Rc` pointer identity — the explicit per-node "stage" a +/// The domain assigned to every node of a validated plan, keyed by +/// `Rc` pointer identity — the explicit per-node "domain" a /// runtime or a DAG export reads instead of re-deriving it. For every -/// non-`KeepPreAsap` node this equals [`produced_availability`]; for a -/// `KeepPreAsap` leaf it is the phase the reaching edge assigned. +/// non-`KeepPreAsap` node this equals [`produced_domain`]; for a +/// `KeepPreAsap` leaf it is the domain the reaching edge assigned. #[derive(Debug, Clone, Default)] -pub struct PhaseAssignment { - stages: HashMap<*const SummaryNode, ExecutionAvailability>, +pub struct DomainAssignment { + domains: HashMap<*const SummaryNode, ValueDomain>, } -impl PhaseAssignment { - /// The stage assigned to `node`, if it was part of the validated plan. - pub fn stage_of(&self, node: &Rc) -> Option { - self.stages.get(&Rc::as_ptr(node)).copied() +impl DomainAssignment { + /// The domain assigned to `node`, if it was part of the validated plan. + pub fn domain_of(&self, node: &Rc) -> Option { + self.domains.get(&Rc::as_ptr(node)).copied() } - /// The stage assigned to the node at `ptr` — for callers walking a plan + /// The domain assigned to the node at `ptr` — for callers walking a plan /// by reference rather than by `Rc`. - pub fn stage_of_ptr(&self, ptr: *const SummaryNode) -> Option { - self.stages.get(&ptr).copied() + pub fn domain_of_ptr(&self, ptr: *const SummaryNode) -> Option { + self.domains.get(&ptr).copied() } } -/// The availability `expr` *produces*, independent of context — `None` for -/// [`SummaryExpr::KeepPreAsap`], whose phase is assigned by the edge reaching +/// The domain `expr` *produces*, independent of context — `None` for +/// [`SummaryExpr::KeepPreAsap`], whose domain is assigned by the edge reaching /// it (see the module docs). -pub fn produced_availability(expr: &SummaryExpr) -> Option { +pub fn produced_domain(expr: &SummaryExpr) -> Option { Some(match expr { SummaryExpr::KeepPreAsap(_) => return None, SummaryExpr::SummaryAgg { .. } | SummaryExpr::SummaryJoin { .. } | SummaryExpr::SummarySubtract { .. } | SummaryExpr::SummaryDelete { .. } - | SummaryExpr::SummaryMerge { .. } => ExecutionAvailability::SummaryState, + | SummaryExpr::SummaryMerge { .. } => ValueDomain::MAINTENANCE_SUMMARY, SummaryExpr::SummaryEstimate { .. } | SummaryExpr::ReadoutPostProcess { .. } => { - ExecutionAvailability::ReadoutValue + ValueDomain::READ_ROWS } - SummaryExpr::UpdateTransform { .. } => ExecutionAvailability::UpdateValue, + SummaryExpr::UpdateTransform { .. } => ValueDomain::MAINTENANCE_ROWS, }) } /// Is `family` the exact-accumulator family whose partial state *is* the /// value — the one summary state a `SummaryAgg` may re-accumulate? -fn is_exact_accumulator_state(schema: &SummarySchema) -> Result<(), PhaseError> { +fn is_exact_accumulator_state(schema: &SummarySchema) -> Result<(), DomainError> { for field in &schema.fields { match &field.dtype { SummaryFamilyType::Plain(_) | SummaryFamilyType::ExactAggregate(..) => {} other => { - return Err(PhaseError::UnsupportedStateComposition { + return Err(DomainError::UnsupportedStateComposition { family: format!("{other:?}"), }) } @@ -221,86 +251,78 @@ fn is_exact_accumulator_state(schema: &SummarySchema) -> Result<(), PhaseError> } /// Validate every edge of the DAG rooted at `root` against the module-level -/// rules, returning each node's assigned stage on success. Shared +/// rules, returning each node's assigned domain on success. Shared /// `Rc`s are visited once per reaching edge (the assignment is /// per node, so a conflict between two edges is what -/// [`PhaseError::AmbiguousKeepPreAsap`] detects). -pub fn validate_execution_phases(root: &Rc) -> Result { +/// [`DomainError::AmbiguousKeepPreAsap`] detects). +pub fn validate_execution_domains(root: &Rc) -> Result { // The root may be a readable value or bare maintained state (a // deployment may hand an `ExactAggregate` accumulator straight to a // consumer) — only an update-path-only root is meaningless. - let root_stage = match produced_availability(&root.expr) { - None => ExecutionAvailability::ReadoutValue, - Some(ExecutionAvailability::UpdateValue) => return Err(PhaseError::UpdateValueAtRoot), - Some(stage) => stage, + let root_domain = match produced_domain(&root.expr) { + None => ValueDomain::READ_ROWS, + Some(ValueDomain::MAINTENANCE_ROWS) => return Err(DomainError::MaintenanceRowsAtRoot), + Some(domain) => domain, }; - validate_execution_phases_at(root, root_stage) + validate_execution_domains_at(root, root_domain) } -/// [`validate_execution_phases`] for a *sub*-plan whose root is known to -/// sit at `stage` — e.g. an `UpdateTransform` about to be placed beneath a +/// [`validate_execution_domains`] for a *sub*-plan whose root is known to +/// sit at `domain` — e.g. an `UpdateTransform` about to be placed beneath a /// `SummaryAgg`, which would be rejected as a whole-plan root but is a /// legal update-path input. Validates every edge beneath `root` exactly /// as the whole-plan entry point does. -pub fn validate_execution_phases_at( +pub fn validate_execution_domains_at( root: &Rc, - stage: ExecutionAvailability, -) -> Result { - let mut assignment = PhaseAssignment::default(); - visit(root, stage, &mut assignment)?; + domain: ValueDomain, +) -> Result { + let mut assignment = DomainAssignment::default(); + visit(root, domain, &mut assignment)?; Ok(assignment) } -/// Record `stage` for `node` (detecting a conflicting earlier assignment +/// Record `domain` for `node` (detecting a conflicting earlier assignment /// for a `KeepPreAsap`), then check and recurse into every child edge. fn visit( node: &Rc, - stage: ExecutionAvailability, - assignment: &mut PhaseAssignment, -) -> Result<(), PhaseError> { + domain: ValueDomain, + assignment: &mut DomainAssignment, +) -> Result<(), DomainError> { let ptr = Rc::as_ptr(node); - if let Some(previous) = assignment.stages.get(&ptr) { - if *previous != stage { - return Err(PhaseError::AmbiguousKeepPreAsap { + if let Some(previous) = assignment.domains.get(&ptr) { + if *previous != domain { + return Err(DomainError::AmbiguousKeepPreAsap { first: *previous, - second: stage, + second: domain, }); } - // Already validated through another edge with the same stage. + // Already validated through another edge with the same domain. return Ok(()); } - assignment.stages.insert(ptr, stage); + assignment.domains.insert(ptr, domain); match &node.expr { SummaryExpr::KeepPreAsap(_) => Ok(()), SummaryExpr::SummaryAgg { child, .. } => { - let child_stage = - child_stage(child, PhaseEdge::SummaryAggChild, |avail| match avail { - ExecutionAvailability::UpdateValue => Ok(()), - ExecutionAvailability::SummaryState => { - is_exact_accumulator_state(&child.schema) - } - ExecutionAvailability::ReadoutValue => { - Err(PhaseError::ReadoutUnderMaintenance { - edge: PhaseEdge::SummaryAggChild.describe(), - child: avail, - }) - } + let child_domain = + child_domain(child, DomainEdge::SummaryAggChild, |avail| match avail { + ValueDomain::MAINTENANCE_ROWS => Ok(()), + ValueDomain::MAINTENANCE_SUMMARY => is_exact_accumulator_state(&child.schema), + other => Err(DomainError::ReadoutUnderMaintenance { + edge: DomainEdge::SummaryAggChild.describe(), + child: other, + }), })?; - visit(child, child_stage, assignment) + visit(child, child_domain, assignment) } SummaryExpr::SummaryJoin { outer, inner, .. } => { for input in [outer, inner] { - let s = child_stage(input, PhaseEdge::SummaryJoinInput, |avail| match avail { - ExecutionAvailability::UpdateValue | ExecutionAvailability::SummaryState => { - Ok(()) - } - ExecutionAvailability::ReadoutValue => { - Err(PhaseError::ReadoutUnderMaintenance { - edge: PhaseEdge::SummaryJoinInput.describe(), - child: avail, - }) - } + let s = child_domain(input, DomainEdge::SummaryJoinInput, |avail| match avail { + ValueDomain::MAINTENANCE_ROWS | ValueDomain::MAINTENANCE_SUMMARY => Ok(()), + other => Err(DomainError::ReadoutUnderMaintenance { + edge: DomainEdge::SummaryJoinInput.describe(), + child: other, + }), })?; visit(input, s, assignment)?; } @@ -308,34 +330,34 @@ fn visit( } SummaryExpr::SummarySubtract { left, right } => { for input in [left, right] { - let s = state_only(input, PhaseEdge::SummarySubtractInput)?; + let s = state_only(input, DomainEdge::SummarySubtractInput)?; visit(input, s, assignment)?; } Ok(()) } SummaryExpr::SummaryDelete { summary_input, .. } => { - let s = state_only(summary_input, PhaseEdge::SummaryDeleteInput)?; + let s = state_only(summary_input, DomainEdge::SummaryDeleteInput)?; visit(summary_input, s, assignment) } SummaryExpr::SummaryMerge { children } => { for input in children { - let s = state_only(input, PhaseEdge::SummaryMergeInput)?; + let s = state_only(input, DomainEdge::SummaryMergeInput)?; visit(input, s, assignment)?; } Ok(()) } SummaryExpr::SummaryEstimate { summary_input, .. } => { - let s = state_only(summary_input, PhaseEdge::SummaryEstimateInput)?; + let s = state_only(summary_input, DomainEdge::SummaryEstimateInput)?; visit(summary_input, s, assignment) } SummaryExpr::UpdateTransform { child, op } => { - let s = child_stage( + let s = child_domain( child, - PhaseEdge::UpdateTransformChild, + DomainEdge::UpdateTransformChild, |avail| match avail { - ExecutionAvailability::UpdateValue => Ok(()), - other => Err(PhaseError::IllegalChildPhase { - edge: PhaseEdge::UpdateTransformChild.describe(), + ValueDomain::MAINTENANCE_ROWS => Ok(()), + other => Err(DomainError::IllegalChildPhase { + edge: DomainEdge::UpdateTransformChild.describe(), child: other, }), }, @@ -344,13 +366,13 @@ fn visit( visit(child, s, assignment) } SummaryExpr::ReadoutPostProcess { child, op } => { - let s = child_stage( + let s = child_domain( child, - PhaseEdge::ReadoutPostProcessChild, + DomainEdge::ReadoutPostProcessChild, |avail| match avail { - ExecutionAvailability::ReadoutValue => Ok(()), - other => Err(PhaseError::IllegalChildPhase { - edge: PhaseEdge::ReadoutPostProcessChild.describe(), + ValueDomain::READ_ROWS => Ok(()), + other => Err(DomainError::IllegalChildPhase { + edge: DomainEdge::ReadoutPostProcessChild.describe(), child: other, }), }, @@ -361,20 +383,20 @@ fn visit( } } -/// The stage `child` takes as a direct input of `parent`, without -/// validating legality — `child`'s own produced availability, or for a -/// `KeepPreAsap` leaf the phase `parent`'s edge assigns it (update-path raw +/// The domain `child` takes as a direct input of `parent`, without +/// validating legality — `child`'s own produced domain, or for a +/// `KeepPreAsap` leaf the domain `parent`'s edge assigns it (update-path raw /// input under maintenance/transform edges, query-time fallback under a -/// post-process, and — meaninglessly, but for a stable answer — `UpdateValue` +/// post-process, and — meaninglessly, but for a stable answer — maintenance rows /// under a state-only edge). For DAG export and other reporting that needs -/// an explicit per-node stage even on a plan that -/// [`validate_execution_phases`] would reject. -pub fn assigned_child_stage(parent: &SummaryExpr, child: &SummaryNode) -> ExecutionAvailability { - if let Some(avail) = produced_availability(&child.expr) { +/// an explicit per-node domain even on a plan that +/// [`validate_execution_domains`] would reject. +pub fn assigned_child_domain(parent: &SummaryExpr, child: &SummaryNode) -> ValueDomain { + if let Some(avail) = produced_domain(&child.expr) { return avail; } match parent { - SummaryExpr::ReadoutPostProcess { .. } => ExecutionAvailability::ReadoutValue, + SummaryExpr::ReadoutPostProcess { .. } => ValueDomain::READ_ROWS, SummaryExpr::KeepPreAsap(_) | SummaryExpr::SummaryAgg { .. } | SummaryExpr::SummaryJoin { .. } @@ -382,40 +404,40 @@ pub fn assigned_child_stage(parent: &SummaryExpr, child: &SummaryNode) -> Execut | SummaryExpr::SummaryDelete { .. } | SummaryExpr::SummaryEstimate { .. } | SummaryExpr::SummaryMerge { .. } - | SummaryExpr::UpdateTransform { .. } => ExecutionAvailability::UpdateValue, + | SummaryExpr::UpdateTransform { .. } => ValueDomain::MAINTENANCE_ROWS, } } -/// The stage `child` takes on `edge`: its own produced availability -/// (checked via `accept`), or — for a `KeepPreAsap` leaf — the phase the +/// The domain `child` takes on `edge`: its own produced domain +/// (checked via `accept`), or — for a `KeepPreAsap` leaf — the domain the /// edge assigns it, derived from what that edge accepts. -fn child_stage( +fn child_domain( child: &Rc, - edge: PhaseEdge, - accept: impl Fn(ExecutionAvailability) -> Result<(), PhaseError>, -) -> Result { - match produced_availability(&child.expr) { + edge: DomainEdge, + accept: impl Fn(ValueDomain) -> Result<(), DomainError>, +) -> Result { + match produced_domain(&child.expr) { Some(avail) => { accept(avail)?; Ok(avail) } None => { - // A raw pre-ASAP subtree executes at whichever phase its consumer + // A raw pre-ASAP subtree executes at whichever domain its consumer // needs: update-path input for maintenance/transform edges, // query-time fallback for a post-process edge. State-only edges // can't consume plain rows at all. let assigned = match edge { - PhaseEdge::SummaryAggChild - | PhaseEdge::SummaryJoinInput - | PhaseEdge::UpdateTransformChild => ExecutionAvailability::UpdateValue, - PhaseEdge::ReadoutPostProcessChild => ExecutionAvailability::ReadoutValue, - PhaseEdge::SummaryEstimateInput - | PhaseEdge::SummarySubtractInput - | PhaseEdge::SummaryDeleteInput - | PhaseEdge::SummaryMergeInput => { - return Err(PhaseError::IllegalChildPhase { + DomainEdge::SummaryAggChild + | DomainEdge::SummaryJoinInput + | DomainEdge::UpdateTransformChild => ValueDomain::MAINTENANCE_ROWS, + DomainEdge::ReadoutPostProcessChild => ValueDomain::READ_ROWS, + DomainEdge::SummaryEstimateInput + | DomainEdge::SummarySubtractInput + | DomainEdge::SummaryDeleteInput + | DomainEdge::SummaryMergeInput => { + return Err(DomainError::IllegalChildPhase { edge: edge.describe(), - child: ExecutionAvailability::UpdateValue, + child: ValueDomain::MAINTENANCE_ROWS, }) } }; @@ -425,13 +447,10 @@ fn child_stage( } } -fn state_only( - child: &Rc, - edge: PhaseEdge, -) -> Result { - child_stage(child, edge, |avail| match avail { - ExecutionAvailability::SummaryState => Ok(()), - other => Err(PhaseError::IllegalChildPhase { +fn state_only(child: &Rc, edge: DomainEdge) -> Result { + child_domain(child, edge, |avail| match avail { + ValueDomain::MAINTENANCE_SUMMARY => Ok(()), + other => Err(DomainError::IllegalChildPhase { edge: edge.describe(), child: other, }), @@ -441,7 +460,7 @@ fn state_only( /// The exact operator must consume only `Plain` columns of its input: for /// an `Aggregate` payload, every grouping key and every measure's input /// column. -fn check_plain_operands(op: &ValueOperator, input: &SummarySchema) -> Result<(), PhaseError> { +fn check_plain_operands(op: &ValueOperator, input: &SummarySchema) -> Result<(), DomainError> { let ValueOperator::Exact(op) = op else { return check_all_plain(input); }; @@ -467,7 +486,7 @@ fn check_plain_operands(op: &ValueOperator, input: &SummarySchema) -> Result<(), continue; } if !matches!(field.dtype, SummaryFamilyType::Plain(_)) { - return Err(PhaseError::NonPlainOperand { + return Err(DomainError::NonPlainOperand { column: field.name.clone(), dtype: format!("{:?}", field.dtype), }); @@ -476,10 +495,10 @@ fn check_plain_operands(op: &ValueOperator, input: &SummarySchema) -> Result<(), Ok(()) } -fn check_all_plain(input: &SummarySchema) -> Result<(), PhaseError> { +fn check_all_plain(input: &SummarySchema) -> Result<(), DomainError> { for field in &input.fields { if !matches!(field.dtype, SummaryFamilyType::Plain(_)) { - return Err(PhaseError::NonPlainOperand { + return Err(DomainError::NonPlainOperand { column: field.name.clone(), dtype: format!("{:?}", field.dtype), }); @@ -654,14 +673,14 @@ mod tests { fn keep_pre_asap_under_summary_agg_is_update_input() { let leaf = keep(); let root = agg(Rc::clone(&leaf), kll()); - let assignment = validate_execution_phases(&root).unwrap(); + let assignment = validate_execution_domains(&root).unwrap(); assert_eq!( - assignment.stage_of(&leaf), - Some(ExecutionAvailability::UpdateValue) + assignment.domain_of(&leaf), + Some(ValueDomain::MAINTENANCE_ROWS) ); assert_eq!( - assignment.stage_of(&root), - Some(ExecutionAvailability::SummaryState) + assignment.domain_of(&root), + Some(ValueDomain::MAINTENANCE_SUMMARY) ); } @@ -672,7 +691,7 @@ mod tests { SummaryFamilyType::ExactAggregate(ExactKind::Sum, ExactParams::Sum), ); let root = estimate(agg(inner, kll())); - assert!(validate_execution_phases(&root).is_ok()); + assert!(validate_execution_domains(&root).is_ok()); } #[test] @@ -680,8 +699,8 @@ mod tests { let inner = estimate(agg(keep(), kll())); let root = agg(inner, kll()); assert!(matches!( - validate_execution_phases(&root), - Err(PhaseError::ReadoutUnderMaintenance { .. }) + validate_execution_domains(&root), + Err(DomainError::ReadoutUnderMaintenance { .. }) )); } @@ -696,15 +715,12 @@ mod tests { schema: plain(&["max"]), guarantee: None, }); - let assignment = validate_execution_phases(&root).unwrap(); - assert_eq!( - assignment.stage_of(&root), - Some(ExecutionAvailability::ReadoutValue) - ); + let assignment = validate_execution_domains(&root).unwrap(); + assert_eq!(assignment.domain_of(&root), Some(ValueDomain::READ_ROWS)); } #[test] - fn non_exact_operator_uses_the_same_readout_phase_contract() { + fn non_exact_operator_uses_the_same_read_domain_contract() { let inner = estimate(agg(keep(), kll())); let root = Rc::new(SummaryNode { expr: SummaryExpr::ReadoutPostProcess { @@ -717,11 +733,8 @@ mod tests { guarantee: None, }); - let assignment = validate_execution_phases(&root).unwrap(); - assert_eq!( - assignment.stage_of(&root), - Some(ExecutionAvailability::ReadoutValue) - ); + let assignment = validate_execution_domains(&root).unwrap(); + assert_eq!(assignment.domain_of(&root), Some(ValueDomain::READ_ROWS)); } #[test] @@ -737,10 +750,10 @@ mod tests { }); let root = agg(post, kll()); assert_eq!( - validate_execution_phases(&root).err(), - Some(PhaseError::ReadoutUnderMaintenance { + validate_execution_domains(&root).err(), + Some(DomainError::ReadoutUnderMaintenance { edge: "SummaryAgg.child", - child: ExecutionAvailability::ReadoutValue, + child: ValueDomain::READ_ROWS, }) ); } @@ -756,14 +769,14 @@ mod tests { guarantee: None, }); assert_eq!( - validate_execution_phases(&transform).err(), - Some(PhaseError::UpdateValueAtRoot) + validate_execution_domains(&transform).err(), + Some(DomainError::MaintenanceRowsAtRoot) ); let root = estimate(agg(Rc::clone(&transform), kll())); - let assignment = validate_execution_phases(&root).unwrap(); + let assignment = validate_execution_domains(&root).unwrap(); assert_eq!( - assignment.stage_of(&transform), - Some(ExecutionAvailability::UpdateValue) + assignment.domain_of(&transform), + Some(ValueDomain::MAINTENANCE_ROWS) ); } @@ -780,16 +793,16 @@ mod tests { }); let root = agg(transform, kll()); assert!(matches!( - validate_execution_phases(&root), - Err(PhaseError::IllegalChildPhase { + validate_execution_domains(&root), + Err(DomainError::IllegalChildPhase { edge: "UpdateTransform.child", - child: ExecutionAvailability::ReadoutValue + child: ValueDomain::READ_ROWS }) )); } #[test] - fn a_shared_keep_pre_asap_reached_at_two_phases_is_ambiguous() { + fn a_shared_keep_pre_asap_reached_in_two_domains_is_ambiguous() { // One raw subtree used both as update input (under a SummaryAgg) and // as a query-time fallback (under an ExactPostProcess) — no single // execution can serve both, so the plan is rejected. @@ -822,20 +835,16 @@ mod tests { }); // SummaryMerge only accepts state, so this fails earlier for a // different reason; probe the ambiguity through a direct visit. - let mut assignment = PhaseAssignment::default(); - visit(&shared, ExecutionAvailability::UpdateValue, &mut assignment).unwrap(); + let mut assignment = DomainAssignment::default(); + visit(&shared, ValueDomain::MAINTENANCE_ROWS, &mut assignment).unwrap(); assert_eq!( - visit( - &shared, - ExecutionAvailability::ReadoutValue, - &mut assignment - ), - Err(PhaseError::AmbiguousKeepPreAsap { - first: ExecutionAvailability::UpdateValue, - second: ExecutionAvailability::ReadoutValue, + visit(&shared, ValueDomain::READ_ROWS, &mut assignment), + Err(DomainError::AmbiguousKeepPreAsap { + first: ValueDomain::MAINTENANCE_ROWS, + second: ValueDomain::READ_ROWS, }) ); - assert!(validate_execution_phases(&root).is_err()); + assert!(validate_execution_domains(&root).is_err()); } #[test] From 9db14098e92d3284bc18c10dd0536e3fdcc74775 Mon Sep 17 00:00:00 2001 From: zz_y Date: Sun, 30 Aug 2026 12:28:26 -0600 Subject: [PATCH 3/4] refactor(post-asap): rename value domain state type --- crates/types/src/dag_export.rs | 18 ++-- crates/types/src/post_asap/expr.rs | 2 +- crates/types/src/post_asap/mod.rs | 2 +- crates/types/src/post_asap/value_domain.rs | 109 ++++++++++++--------- 4 files changed, 77 insertions(+), 54 deletions(-) diff --git a/crates/types/src/dag_export.rs b/crates/types/src/dag_export.rs index d2a7b24d..b256fec3 100644 --- a/crates/types/src/dag_export.rs +++ b/crates/types/src/dag_export.rs @@ -47,8 +47,8 @@ use std::rc::Rc; use serde::Serialize; use crate::post_asap::{ - assigned_child_domain, produced_domain, AccuracyError, ExactOperator, ResultGuarantee, - SummaryExpr, SummaryNode, ValueDomain, ValueOperator, + assigned_child_domain, produced_domain, AccuracyError, ExactOperator, ExecutionDataState, + ResultGuarantee, SummaryExpr, SummaryNode, ValueOperator, }; use crate::pre_asap::cse::{structural_hash, HashCache}; use crate::pre_asap::query_expr::{QueryExpr, Source}; @@ -367,8 +367,8 @@ pub fn export_summary(node: &SummaryNode) -> SummaryDagGraph { /// produced domain, or query-time readout for a bare `KeepPreAsap` /// (the same convention `post_asap::value_domain::validate_execution_domains` /// uses for a root). -fn root_domain(node: &SummaryNode) -> ValueDomain { - produced_domain(&node.expr).unwrap_or(ValueDomain::READ_ROWS) +fn root_domain(node: &SummaryNode) -> ExecutionDataState { + produced_domain(&node.expr).unwrap_or(ExecutionDataState::READ_ROWS) } /// `detail` for an [`ExactOperator`] payload — its own fields, rendered the @@ -472,7 +472,7 @@ fn family_label(family: &crate::post_asap::SummaryFamilyType) -> String { /// reads it rather than inferring it from the node's kind. fn summary_shape( expr: &SummaryExpr, - domain: ValueDomain, + domain: ExecutionDataState, ) -> (&'static str, String, serde_json::Value) { let (kind, label, mut detail) = match expr { SummaryExpr::KeepPreAsap(_) => { @@ -566,7 +566,11 @@ fn summary_children(expr: &SummaryExpr) -> Vec<&Rc> { /// post-order (children pushed before their parent), and return the pushed /// root's id. Exhaustive over every [`SummaryExpr`] variant, matching this /// file's own exhaustive style for `QueryExpr` in [`build`]. -fn build_summary(node: &SummaryNode, nodes: &mut Vec, domain: ValueDomain) -> u32 { +fn build_summary( + node: &SummaryNode, + nodes: &mut Vec, + domain: ExecutionDataState, +) -> u32 { if let SummaryExpr::KeepPreAsap(inner) = &node.expr { let pre_asap_subgraph = export(inner); let inner_kind = pre_asap_subgraph.nodes[pre_asap_subgraph.root as usize].kind; @@ -869,7 +873,7 @@ fn build_summary_hybrid( nodes: &mut Vec, cache: &mut HashCache, find_winner: &mut dyn FnMut(&QueryExpr) -> Option, - domain: ValueDomain, + domain: ExecutionDataState, ) -> u32 { if let SummaryExpr::KeepPreAsap(inner) = &node.expr { return build(inner, nodes, cache, find_winner); diff --git a/crates/types/src/post_asap/expr.rs b/crates/types/src/post_asap/expr.rs index aedd0ba8..1f8dc920 100644 --- a/crates/types/src/post_asap/expr.rs +++ b/crates/types/src/post_asap/expr.rs @@ -192,7 +192,7 @@ pub enum SummaryExpr { /// produces plain update values, so its output may feed a downstream /// [`SummaryAgg`](SummaryExpr::SummaryAgg)'s maintenance — the "outer /// summary over an inner non-accumulator exact transform" direction. - /// See [`super::value_domain::ValueDomain`] for the edge contract. + /// See [`super::value_domain::ExecutionDataState`] for the edge contract. UpdateTransform { child: Rc, op: ValueOperator, diff --git a/crates/types/src/post_asap/mod.rs b/crates/types/src/post_asap/mod.rs index 5ac2864b..443e69b4 100644 --- a/crates/types/src/post_asap/mod.rs +++ b/crates/types/src/post_asap/mod.rs @@ -52,5 +52,5 @@ pub use sketch::{ pub use value_domain::{ assigned_child_domain, exact_operator_output_schema, produced_domain, validate_execution_domains, validate_execution_domains_at, DataPrimitive, DomainAssignment, - DomainError, ExactOperatorSchemaError, ExecutionTiming, ValueDomain, + DomainError, ExactOperatorSchemaError, ExecutionDataState, ExecutionTiming, }; diff --git a/crates/types/src/post_asap/value_domain.rs b/crates/types/src/post_asap/value_domain.rs index c65d705b..a6363ba7 100644 --- a/crates/types/src/post_asap/value_domain.rs +++ b/crates/types/src/post_asap/value_domain.rs @@ -10,7 +10,7 @@ //! why it and [`SummaryExpr::UpdateTransform`] are two separate variants //! rather than one domain-ambiguous value operation. //! -//! [`ValueDomain`] is what a node's output *is*, at which domain; +//! [`ExecutionDataState`] is what a node's output *is*, at which domain; //! [`validate_execution_domains`] checks every edge of a DAG against the //! rules below at plan construction, returning a typed [`DomainError`] rather //! than deferring to a runtime failure. @@ -85,12 +85,12 @@ impl DataPrimitive { /// The two-dimensional edge contract: when a value exists and which data /// primitive it carries. #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] -pub struct ValueDomain { +pub struct ExecutionDataState { pub timing: ExecutionTiming, pub primitive: DataPrimitive, } -impl ValueDomain { +impl ExecutionDataState { pub const MAINTENANCE_ROWS: Self = Self { timing: ExecutionTiming::MaintenanceTime, primitive: DataPrimitive::Rows, @@ -105,7 +105,7 @@ impl ValueDomain { }; } -impl std::fmt::Display for ValueDomain { +impl std::fmt::Display for ExecutionDataState { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { write!(f, "{}/{}", self.timing.as_str(), self.primitive.as_str()) } @@ -155,7 +155,7 @@ pub enum DomainError { )] ReadoutUnderMaintenance { edge: &'static str, - child: ValueDomain, + child: ExecutionDataState, }, /// Any other edge whose child domain the parent does not accept /// (e.g. plain update rows fed straight into a `SummaryEstimate`, or a @@ -163,7 +163,7 @@ pub enum DomainError { #[error("{edge} does not accept a {child} input")] IllegalChildPhase { edge: &'static str, - child: ValueDomain, + child: ExecutionDataState, }, /// A `SummaryAgg` whose child is summary state of a family other than an /// exact accumulator — re-accumulating opaque sketch/sample/… state on @@ -180,8 +180,8 @@ pub enum DomainError { plan" )] AmbiguousKeepPreAsap { - first: ValueDomain, - second: ValueDomain, + first: ExecutionDataState, + second: ExecutionDataState, }, /// An update-path-only node (`UpdateTransform`) at the root of a plan: /// nothing maintains state above it, so its output is never read. @@ -200,18 +200,18 @@ pub enum DomainError { /// `KeepPreAsap` leaf it is the domain the reaching edge assigned. #[derive(Debug, Clone, Default)] pub struct DomainAssignment { - domains: HashMap<*const SummaryNode, ValueDomain>, + domains: HashMap<*const SummaryNode, ExecutionDataState>, } impl DomainAssignment { /// The domain assigned to `node`, if it was part of the validated plan. - pub fn domain_of(&self, node: &Rc) -> Option { + pub fn domain_of(&self, node: &Rc) -> Option { self.domains.get(&Rc::as_ptr(node)).copied() } /// The domain assigned to the node at `ptr` — for callers walking a plan /// by reference rather than by `Rc`. - pub fn domain_of_ptr(&self, ptr: *const SummaryNode) -> Option { + pub fn domain_of_ptr(&self, ptr: *const SummaryNode) -> Option { self.domains.get(&ptr).copied() } } @@ -219,18 +219,18 @@ impl DomainAssignment { /// The domain `expr` *produces*, independent of context — `None` for /// [`SummaryExpr::KeepPreAsap`], whose domain is assigned by the edge reaching /// it (see the module docs). -pub fn produced_domain(expr: &SummaryExpr) -> Option { +pub fn produced_domain(expr: &SummaryExpr) -> Option { Some(match expr { SummaryExpr::KeepPreAsap(_) => return None, SummaryExpr::SummaryAgg { .. } | SummaryExpr::SummaryJoin { .. } | SummaryExpr::SummarySubtract { .. } | SummaryExpr::SummaryDelete { .. } - | SummaryExpr::SummaryMerge { .. } => ValueDomain::MAINTENANCE_SUMMARY, + | SummaryExpr::SummaryMerge { .. } => ExecutionDataState::MAINTENANCE_SUMMARY, SummaryExpr::SummaryEstimate { .. } | SummaryExpr::ReadoutPostProcess { .. } => { - ValueDomain::READ_ROWS + ExecutionDataState::READ_ROWS } - SummaryExpr::UpdateTransform { .. } => ValueDomain::MAINTENANCE_ROWS, + SummaryExpr::UpdateTransform { .. } => ExecutionDataState::MAINTENANCE_ROWS, }) } @@ -260,8 +260,10 @@ pub fn validate_execution_domains(root: &Rc) -> Result ValueDomain::READ_ROWS, - Some(ValueDomain::MAINTENANCE_ROWS) => return Err(DomainError::MaintenanceRowsAtRoot), + None => ExecutionDataState::READ_ROWS, + Some(ExecutionDataState::MAINTENANCE_ROWS) => { + return Err(DomainError::MaintenanceRowsAtRoot) + } Some(domain) => domain, }; validate_execution_domains_at(root, root_domain) @@ -274,7 +276,7 @@ pub fn validate_execution_domains(root: &Rc) -> Result, - domain: ValueDomain, + domain: ExecutionDataState, ) -> Result { let mut assignment = DomainAssignment::default(); visit(root, domain, &mut assignment)?; @@ -285,7 +287,7 @@ pub fn validate_execution_domains_at( /// for a `KeepPreAsap`), then check and recurse into every child edge. fn visit( node: &Rc, - domain: ValueDomain, + domain: ExecutionDataState, assignment: &mut DomainAssignment, ) -> Result<(), DomainError> { let ptr = Rc::as_ptr(node); @@ -306,8 +308,10 @@ fn visit( SummaryExpr::SummaryAgg { child, .. } => { let child_domain = child_domain(child, DomainEdge::SummaryAggChild, |avail| match avail { - ValueDomain::MAINTENANCE_ROWS => Ok(()), - ValueDomain::MAINTENANCE_SUMMARY => is_exact_accumulator_state(&child.schema), + ExecutionDataState::MAINTENANCE_ROWS => Ok(()), + ExecutionDataState::MAINTENANCE_SUMMARY => { + is_exact_accumulator_state(&child.schema) + } other => Err(DomainError::ReadoutUnderMaintenance { edge: DomainEdge::SummaryAggChild.describe(), child: other, @@ -318,7 +322,8 @@ fn visit( SummaryExpr::SummaryJoin { outer, inner, .. } => { for input in [outer, inner] { let s = child_domain(input, DomainEdge::SummaryJoinInput, |avail| match avail { - ValueDomain::MAINTENANCE_ROWS | ValueDomain::MAINTENANCE_SUMMARY => Ok(()), + ExecutionDataState::MAINTENANCE_ROWS + | ExecutionDataState::MAINTENANCE_SUMMARY => Ok(()), other => Err(DomainError::ReadoutUnderMaintenance { edge: DomainEdge::SummaryJoinInput.describe(), child: other, @@ -355,7 +360,7 @@ fn visit( child, DomainEdge::UpdateTransformChild, |avail| match avail { - ValueDomain::MAINTENANCE_ROWS => Ok(()), + ExecutionDataState::MAINTENANCE_ROWS => Ok(()), other => Err(DomainError::IllegalChildPhase { edge: DomainEdge::UpdateTransformChild.describe(), child: other, @@ -370,7 +375,7 @@ fn visit( child, DomainEdge::ReadoutPostProcessChild, |avail| match avail { - ValueDomain::READ_ROWS => Ok(()), + ExecutionDataState::READ_ROWS => Ok(()), other => Err(DomainError::IllegalChildPhase { edge: DomainEdge::ReadoutPostProcessChild.describe(), child: other, @@ -391,12 +396,12 @@ fn visit( /// under a state-only edge). For DAG export and other reporting that needs /// an explicit per-node domain even on a plan that /// [`validate_execution_domains`] would reject. -pub fn assigned_child_domain(parent: &SummaryExpr, child: &SummaryNode) -> ValueDomain { +pub fn assigned_child_domain(parent: &SummaryExpr, child: &SummaryNode) -> ExecutionDataState { if let Some(avail) = produced_domain(&child.expr) { return avail; } match parent { - SummaryExpr::ReadoutPostProcess { .. } => ValueDomain::READ_ROWS, + SummaryExpr::ReadoutPostProcess { .. } => ExecutionDataState::READ_ROWS, SummaryExpr::KeepPreAsap(_) | SummaryExpr::SummaryAgg { .. } | SummaryExpr::SummaryJoin { .. } @@ -404,7 +409,7 @@ pub fn assigned_child_domain(parent: &SummaryExpr, child: &SummaryNode) -> Value | SummaryExpr::SummaryDelete { .. } | SummaryExpr::SummaryEstimate { .. } | SummaryExpr::SummaryMerge { .. } - | SummaryExpr::UpdateTransform { .. } => ValueDomain::MAINTENANCE_ROWS, + | SummaryExpr::UpdateTransform { .. } => ExecutionDataState::MAINTENANCE_ROWS, } } @@ -414,8 +419,8 @@ pub fn assigned_child_domain(parent: &SummaryExpr, child: &SummaryNode) -> Value fn child_domain( child: &Rc, edge: DomainEdge, - accept: impl Fn(ValueDomain) -> Result<(), DomainError>, -) -> Result { + accept: impl Fn(ExecutionDataState) -> Result<(), DomainError>, +) -> Result { match produced_domain(&child.expr) { Some(avail) => { accept(avail)?; @@ -429,15 +434,15 @@ fn child_domain( let assigned = match edge { DomainEdge::SummaryAggChild | DomainEdge::SummaryJoinInput - | DomainEdge::UpdateTransformChild => ValueDomain::MAINTENANCE_ROWS, - DomainEdge::ReadoutPostProcessChild => ValueDomain::READ_ROWS, + | DomainEdge::UpdateTransformChild => ExecutionDataState::MAINTENANCE_ROWS, + DomainEdge::ReadoutPostProcessChild => ExecutionDataState::READ_ROWS, DomainEdge::SummaryEstimateInput | DomainEdge::SummarySubtractInput | DomainEdge::SummaryDeleteInput | DomainEdge::SummaryMergeInput => { return Err(DomainError::IllegalChildPhase { edge: edge.describe(), - child: ValueDomain::MAINTENANCE_ROWS, + child: ExecutionDataState::MAINTENANCE_ROWS, }) } }; @@ -447,9 +452,12 @@ fn child_domain( } } -fn state_only(child: &Rc, edge: DomainEdge) -> Result { +fn state_only( + child: &Rc, + edge: DomainEdge, +) -> Result { child_domain(child, edge, |avail| match avail { - ValueDomain::MAINTENANCE_SUMMARY => Ok(()), + ExecutionDataState::MAINTENANCE_SUMMARY => Ok(()), other => Err(DomainError::IllegalChildPhase { edge: edge.describe(), child: other, @@ -676,11 +684,11 @@ mod tests { let assignment = validate_execution_domains(&root).unwrap(); assert_eq!( assignment.domain_of(&leaf), - Some(ValueDomain::MAINTENANCE_ROWS) + Some(ExecutionDataState::MAINTENANCE_ROWS) ); assert_eq!( assignment.domain_of(&root), - Some(ValueDomain::MAINTENANCE_SUMMARY) + Some(ExecutionDataState::MAINTENANCE_SUMMARY) ); } @@ -716,7 +724,10 @@ mod tests { guarantee: None, }); let assignment = validate_execution_domains(&root).unwrap(); - assert_eq!(assignment.domain_of(&root), Some(ValueDomain::READ_ROWS)); + assert_eq!( + assignment.domain_of(&root), + Some(ExecutionDataState::READ_ROWS) + ); } #[test] @@ -734,7 +745,10 @@ mod tests { }); let assignment = validate_execution_domains(&root).unwrap(); - assert_eq!(assignment.domain_of(&root), Some(ValueDomain::READ_ROWS)); + assert_eq!( + assignment.domain_of(&root), + Some(ExecutionDataState::READ_ROWS) + ); } #[test] @@ -753,7 +767,7 @@ mod tests { validate_execution_domains(&root).err(), Some(DomainError::ReadoutUnderMaintenance { edge: "SummaryAgg.child", - child: ValueDomain::READ_ROWS, + child: ExecutionDataState::READ_ROWS, }) ); } @@ -776,7 +790,7 @@ mod tests { let assignment = validate_execution_domains(&root).unwrap(); assert_eq!( assignment.domain_of(&transform), - Some(ValueDomain::MAINTENANCE_ROWS) + Some(ExecutionDataState::MAINTENANCE_ROWS) ); } @@ -796,7 +810,7 @@ mod tests { validate_execution_domains(&root), Err(DomainError::IllegalChildPhase { edge: "UpdateTransform.child", - child: ValueDomain::READ_ROWS + child: ExecutionDataState::READ_ROWS }) )); } @@ -836,12 +850,17 @@ mod tests { // SummaryMerge only accepts state, so this fails earlier for a // different reason; probe the ambiguity through a direct visit. let mut assignment = DomainAssignment::default(); - visit(&shared, ValueDomain::MAINTENANCE_ROWS, &mut assignment).unwrap(); + visit( + &shared, + ExecutionDataState::MAINTENANCE_ROWS, + &mut assignment, + ) + .unwrap(); assert_eq!( - visit(&shared, ValueDomain::READ_ROWS, &mut assignment), + visit(&shared, ExecutionDataState::READ_ROWS, &mut assignment), Err(DomainError::AmbiguousKeepPreAsap { - first: ValueDomain::MAINTENANCE_ROWS, - second: ValueDomain::READ_ROWS, + first: ExecutionDataState::MAINTENANCE_ROWS, + second: ExecutionDataState::READ_ROWS, }) ); assert!(validate_execution_domains(&root).is_err()); From aa95a1aa4f575d5d5fc7d2c1cb8e280a95ae788a Mon Sep 17 00:00:00 2001 From: zz_y Date: Sun, 30 Aug 2026 12:33:41 -0600 Subject: [PATCH 4/4] refactor(post-asap): align module with execution data state --- crates/types/src/dag_export.rs | 2 +- .../{value_domain.rs => execution_data_state.rs} | 2 +- crates/types/src/post_asap/expr.rs | 4 ++-- crates/types/src/post_asap/mod.rs | 12 ++++++------ 4 files changed, 10 insertions(+), 10 deletions(-) rename crates/types/src/post_asap/{value_domain.rs => execution_data_state.rs} (99%) diff --git a/crates/types/src/dag_export.rs b/crates/types/src/dag_export.rs index b256fec3..5c1ee4ad 100644 --- a/crates/types/src/dag_export.rs +++ b/crates/types/src/dag_export.rs @@ -365,7 +365,7 @@ pub fn export_summary(node: &SummaryNode) -> SummaryDagGraph { /// The explicit execution domain of an exported plan's root — its own /// produced domain, or query-time readout for a bare `KeepPreAsap` -/// (the same convention `post_asap::value_domain::validate_execution_domains` +/// (the same convention `post_asap::execution_data_state::validate_execution_domains` /// uses for a root). fn root_domain(node: &SummaryNode) -> ExecutionDataState { produced_domain(&node.expr).unwrap_or(ExecutionDataState::READ_ROWS) diff --git a/crates/types/src/post_asap/value_domain.rs b/crates/types/src/post_asap/execution_data_state.rs similarity index 99% rename from crates/types/src/post_asap/value_domain.rs rename to crates/types/src/post_asap/execution_data_state.rs index a6363ba7..d0ca6fcd 100644 --- a/crates/types/src/post_asap/value_domain.rs +++ b/crates/types/src/post_asap/execution_data_state.rs @@ -1,4 +1,4 @@ -//! Execution-domain contract for mixed exact/summary plans (issue #171). +//! Execution-data-state contract for mixed exact/summary plans (issue #171). //! //! A post-ASAP DAG mixes two very different moments of execution: the //! **update/ingest path** (rows arrive, maintained summary state is updated) diff --git a/crates/types/src/post_asap/expr.rs b/crates/types/src/post_asap/expr.rs index 1f8dc920..7f95901f 100644 --- a/crates/types/src/post_asap/expr.rs +++ b/crates/types/src/post_asap/expr.rs @@ -192,7 +192,7 @@ pub enum SummaryExpr { /// produces plain update values, so its output may feed a downstream /// [`SummaryAgg`](SummaryExpr::SummaryAgg)'s maintenance — the "outer /// summary over an inner non-accumulator exact transform" direction. - /// See [`super::value_domain::ExecutionDataState`] for the edge contract. + /// See [`super::execution_data_state::ExecutionDataState`] for the edge contract. UpdateTransform { child: Rc, op: ValueOperator, @@ -203,7 +203,7 @@ pub enum SummaryExpr { /// final plain query result — the "outer exact fold over an inner /// summary readout" direction. Can never feed maintained state: a /// `SummaryAgg` above one of these is a plan-time - /// [`super::value_domain::DomainError`], never a runtime failure. + /// [`super::execution_data_state::DomainError`], never a runtime failure. ReadoutPostProcess { child: Rc, op: ValueOperator, diff --git a/crates/types/src/post_asap/mod.rs b/crates/types/src/post_asap/mod.rs index 443e69b4..49730550 100644 --- a/crates/types/src/post_asap/mod.rs +++ b/crates/types/src/post_asap/mod.rs @@ -27,13 +27,18 @@ //! alongside `reduction` and on sketch-valued edge types //! — see `asap_aware_mapping::grouping`'s module docs for why. +pub mod execution_data_state; pub mod expr; pub mod guarantee; pub mod query_time; pub mod schema; pub mod sketch; -pub mod value_domain; +pub use execution_data_state::{ + assigned_child_domain, exact_operator_output_schema, produced_domain, + validate_execution_domains, validate_execution_domains_at, DataPrimitive, DomainAssignment, + DomainError, ExactOperatorSchemaError, ExecutionDataState, ExecutionTiming, +}; pub use expr::{ExactOperator, SummaryExpr, SummaryNode, ValueOperator}; pub use guarantee::{ AccuracyError, BoundExpr, CompositionOperator, ErrorMetric, GuaranteeSource, ProbabilityExpr, @@ -49,8 +54,3 @@ pub use sketch::{ HydraParams, SamplingKind, SamplingParams, SketchAlgorithm, SketchCategory, SketchKind, SketchParams, SketchQuery, StatModelKind, StatModelParams, WaveletKind, WaveletParams, }; -pub use value_domain::{ - assigned_child_domain, exact_operator_output_schema, produced_domain, - validate_execution_domains, validate_execution_domains_at, DataPrimitive, DomainAssignment, - DomainError, ExactOperatorSchemaError, ExecutionDataState, ExecutionTiming, -};