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) + )); + } +}