diff --git a/crates/asap-aware-mapping/src/cost_model.rs b/crates/asap-aware-mapping/src/cost_model.rs index b1780c8d..3967e50e 100644 --- a/crates/asap-aware-mapping/src/cost_model.rs +++ b/crates/asap-aware-mapping/src/cost_model.rs @@ -56,14 +56,219 @@ use asap_types::pre_asap::agg_intent::AggIntent; use asap_types::pre_asap::expr_ir::ColumnRef; use asap_types::pre_asap::query_expr::QueryExpr; +use crate::exact_composition::{CompositionPlacement, ExactComposition}; use crate::recurrence::{ - self, Horizon, RecurrenceCostExplanation, RecurrenceError, RecurrenceProfile, + self, CostRate, EvaluationRate, Horizon, RecurrenceCostExplanation, RecurrenceError, + RecurrenceProfile, }; use crate::replacement::{ realize_child, Implementation, Replacement, ReplacementProvenance, ReplacementSubDAG, TargetSubDAG, }; +// ── Recurring-cost vocabulary for mixed exact/summary plans (issue #171) ── + +/// The unit a recurring cost is expressed in. One variant today; an enum so +/// a JSON/DAG export names the unit explicitly instead of a consumer +/// assuming it, and so a future per-resource unit can be added without +/// changing every hook's signature. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum CostUnit { + /// Abstract cost units per wall-clock second — the common currency + /// every recurring alternative (maintain-and-read vs. recompute-per-eval) + /// is compared in. + CostUnitsPerSecond, +} + +impl CostUnit { + /// Stable name for export (`"cost_units_per_second"`). + pub fn as_str(self) -> &'static str { + match self { + Self::CostUnitsPerSecond => "cost_units_per_second", + } + } +} + +/// Who produced a set of [`ExactCompositionCostInputs`], and under which +/// model version — carried into every composed decision's explanation and +/// DAG export so a reviewer can tell a deployment's measured numbers from +/// a placeholder. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct CostProvenance { + /// The cost model's own name (e.g. `"DefaultCostModel"`). + pub model: String, + /// The model's own version string, whatever scheme it uses. + pub version: String, +} + +/// Which mixed-execution shapes the downstream runtime can actually +/// execute (issue #171). [`crate::exact_composition::ExactCompositionStrategy`] +/// proposes an `ExactPostProcess` candidate only when +/// `exact_post_process` is set, and an `ExactTransform` candidate only +/// when `exact_update_transform` is — a runtime that cannot run an exact +/// operator on the update path must never be handed one. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub struct MixedExecutionCapabilities { + /// The runtime can apply an exact operator to summary readouts at + /// query evaluation time. + pub exact_post_process: bool, + /// The runtime can apply an exact row transform on the update path, + /// feeding its output into maintained summary state. + pub exact_update_transform: bool, +} + +impl MixedExecutionCapabilities { + /// Neither shape supported. + pub const NONE: Self = Self { + exact_post_process: false, + exact_update_transform: false, + }; + /// Both shapes supported. + pub const ALL: Self = Self { + exact_post_process: true, + exact_update_transform: true, + }; + + pub fn supports(self, placement: CompositionPlacement) -> bool { + match placement { + CompositionPlacement::PostProcess => self.exact_post_process, + CompositionPlacement::Transform => self.exact_update_transform, + } + } +} + +/// What [`CostModel::exact_composition_cost_inputs`] is asked about: one +/// composed alternative at one site, paired with the concrete summary it +/// composes with. +#[derive(Debug, Clone, Copy)] +pub struct ExactCompositionCostRequest<'a> { + /// The pre-ASAP target the composed candidate replaces. + pub target: &'a QueryExpr, + /// The composition itself — placement, operator, child target. + pub composition: &'a ExactComposition, + /// For [`CompositionPlacement::PostProcess`]: the child target's *selected* + /// summary readout candidate the exact operator consumes. For + /// [`CompositionPlacement::Transform`]: the maintained summary *above* the + /// transform that consumes its output (the `SummaryAgg` this transform + /// feeds). Either way, the summary whose maintenance/read cost the + /// formula charges. + pub summary: &'a SummaryNode, + /// How many times this site actually runs once ancestors' own choices + /// are accounted for (see `PlanSpace::global_selection`). + pub effective_consumer_count: usize, +} + +/// Every input the issue #171 cost formulas need, each individually +/// optional: **an unknown stays `None` — never a zero** — so a formula +/// with a missing input yields no rate at all rather than a spuriously +/// cheap one, and global selection then keeps the conservative +/// `KeepPreAsap` behavior. A deployment model that wants defaults supplies +/// them explicitly by overriding [`CostModel::exact_composition_cost_inputs`]. +#[derive(Debug, Clone, PartialEq)] +pub struct ExactCompositionCostInputs { + /// Exact operator cost per row it processes — per readout row for a + /// post-process, per input row for an update-path transform. + pub exact_cost_per_row: Option, + /// Rows the exact operator consumes per evaluation (post-process) or + /// per update (transform). + pub expected_input_rows: Option, + /// Rows the exact operator emits per evaluation/update. + pub expected_output_rows: Option, + /// Cost of one update to the composed-with summary's maintained state. + pub summary_maintenance_cost_per_update: Option, + /// Cost of one readout of that summary at evaluation time. + pub summary_read_cost: Option, + /// Update (ingest) events per second reaching this site. + pub update_rate: Option, + /// Evaluations per second across every consumer of this site. + pub evaluation_rate: Option, + /// Cost of one full raw recompute of the target from pre-ASAP data — + /// the `KeepPreAsap` baseline's per-evaluation cost. + pub raw_recompute_cost: Option, + pub unit: CostUnit, + pub provenance: CostProvenance, +} + +impl ExactCompositionCostInputs { + /// Every input unknown, attributed to `provenance` — what a model that + /// has no statistics for a site returns. + pub fn unknown(provenance: CostProvenance) -> Self { + Self { + exact_cost_per_row: None, + expected_input_rows: None, + expected_output_rows: None, + summary_maintenance_cost_per_update: None, + summary_read_cost: None, + update_rate: None, + evaluation_rate: None, + raw_recompute_cost: None, + unit: CostUnit::CostUnitsPerSecond, + provenance, + } + } + + /// The rate for whichever composition placement is requested — + /// [`postprocess_plan_cost_rate`] or [`pretransform_plan_cost_rate`]. + pub fn composed_plan_cost_rate(&self, placement: CompositionPlacement) -> Option { + match placement { + CompositionPlacement::PostProcess => postprocess_plan_cost_rate(self), + CompositionPlacement::Transform => pretransform_plan_cost_rate(self), + } + } +} + +/// Outer exact post-process over a maintained summary: +/// +/// ```text +/// postprocess_plan_cost_rate = +/// update_rate * summary_maintenance_cost_per_update +/// + evaluation_rate * (summary_read_cost +/// + output_rows_per_eval * exact_postprocess_cost_per_row) +/// ``` +/// +/// `None` if any input is unknown — see [`ExactCompositionCostInputs`]. +pub fn postprocess_plan_cost_rate(inputs: &ExactCompositionCostInputs) -> Option { + let maintenance = inputs.update_rate? * inputs.summary_maintenance_cost_per_update?; + let per_eval = + inputs.summary_read_cost? + inputs.expected_output_rows? * inputs.exact_cost_per_row?; + let evaluation = inputs.evaluation_rate?.0 * per_eval; + finite_rate(maintenance + evaluation) +} + +/// Outer maintained summary over an exact update-time transform: +/// +/// ```text +/// pretransform_plan_cost_rate = +/// update_rate * (exact_transform_cost_per_input_row +/// + summary_maintenance_cost_per_update) +/// + evaluation_rate * summary_read_cost +/// ``` +/// +/// `None` if any input is unknown — see [`ExactCompositionCostInputs`]. +pub fn pretransform_plan_cost_rate(inputs: &ExactCompositionCostInputs) -> Option { + let per_update = inputs.exact_cost_per_row? + inputs.summary_maintenance_cost_per_update?; + let maintenance = inputs.update_rate? * per_update; + let evaluation = inputs.evaluation_rate?.0 * inputs.summary_read_cost?; + finite_rate(maintenance + evaluation) +} + +/// The raw/pre-ASAP fallback baseline: +/// +/// ```text +/// raw_recompute_cost_rate = evaluation_rate * raw_recompute_cost +/// ``` +/// +/// `None` if either input is unknown — see [`ExactCompositionCostInputs`]. +pub fn raw_recompute_cost_rate(inputs: &ExactCompositionCostInputs) -> Option { + finite_rate(inputs.evaluation_rate?.0 * inputs.raw_recompute_cost?) +} + +fn finite_rate(units_per_second: f64) -> Option { + units_per_second + .is_finite() + .then_some(CostRate(units_per_second)) +} + /// A CSE-detected, legality-gated shared subtree with two or more consumers /// — the unit [`CostModel::cse_share_decision`] decides over. Built by /// [`PlanSpace::cost_sorted`](crate::replacement::PlanSpace::cost_sorted) @@ -504,6 +709,47 @@ pub trait CostModel { let _ = (candidate, target); f64::NAN } + + /// Which mixed exact/summary execution shapes the downstream runtime + /// advertises (issue #171). Gates candidate *generation* in + /// [`crate::exact_composition::ExactCompositionStrategy`]: a shape the + /// runtime can't execute is never proposed, so it can't be selected + /// either. + /// + /// Default: [`MixedExecutionCapabilities::ALL`]. The built-in model + /// describes no particular runtime, and leaving both shapes *visible* + /// in `PlanSpace` (for explanations and the DAG viewer) is the more + /// informative default; selection is still gated separately by + /// [`Self::exact_composition_cost_inputs`], whose default supplies no + /// statistics, so nothing is ever *committed* to under the built-in + /// model. A deployment whose runtime lacks a shape narrows this. + fn mixed_execution_capabilities(&self) -> MixedExecutionCapabilities { + MixedExecutionCapabilities::ALL + } + + /// The statistics the issue #171 recurring-cost formulas need for one + /// composed alternative — see [`ExactCompositionCostInputs`] for each + /// input and [`postprocess_plan_cost_rate`]/ + /// [`pretransform_plan_cost_rate`]/[`raw_recompute_cost_rate`] for how + /// they combine. One structured hook rather than eight scalar ones, so + /// a deployment answers them all from one place (and can attach its own + /// [`CostProvenance`]). + /// + /// Default: every input unknown ([`ExactCompositionCostInputs::unknown`]) + /// — unknown is never zero, and with no rate derivable + /// `PlanSpace::global_selection` keeps the conservative `KeepPreAsap` + /// behavior for the site. A deployment that wants defaults must supply + /// them here explicitly. + fn exact_composition_cost_inputs( + &self, + request: &ExactCompositionCostRequest<'_>, + ) -> ExactCompositionCostInputs { + let _ = request; + ExactCompositionCostInputs::unknown(CostProvenance { + model: "CostModel::exact_composition_cost_inputs (default)".into(), + version: "unknown".into(), + }) + } } fn sketch_state( @@ -661,6 +907,12 @@ impl CostModel for DefaultCostModel { (self.cse_recompute_cost(&cse) * consumer_count).0 } } + // A composed candidate is costed in cost-units-per-second by + // `PlanSpace::global_selection` against the child decision it + // is committed with — a different unit from this structural + // estimate, and unknowable here without that child. `NaN` + // keeps it from ever out-ranking a real estimate by accident. + Replacement::ExactComposition(_) => f64::NAN, } } } @@ -805,6 +1057,73 @@ mod tests { ); } + // ── Recurring-cost formulas (issue #171) ───────────────────────────── + + fn known_inputs() -> ExactCompositionCostInputs { + ExactCompositionCostInputs { + exact_cost_per_row: Some(0.1), + expected_input_rows: Some(50.0), + expected_output_rows: Some(10.0), + summary_maintenance_cost_per_update: Some(0.01), + summary_read_cost: Some(1.0), + update_rate: Some(100.0), + evaluation_rate: Some(EvaluationRate(2.0)), + raw_recompute_cost: Some(100.0), + unit: CostUnit::CostUnitsPerSecond, + provenance: CostProvenance { + model: "test".into(), + version: "1".into(), + }, + } + } + + #[test] + fn composition_formulas_match_the_issue_definitions() { + let inputs = known_inputs(); + // 100 * 0.01 + 2 * (1 + 10 * 0.1) = 1 + 4 = 5 + assert_eq!(postprocess_plan_cost_rate(&inputs).unwrap().0, 5.0); + // 100 * (0.1 + 0.01) + 2 * 1 = 11 + 2 = 13 + assert!((pretransform_plan_cost_rate(&inputs).unwrap().0 - 13.0).abs() < 1e-9); + // 2 * 100 + assert_eq!(raw_recompute_cost_rate(&inputs).unwrap().0, 200.0); + assert_eq!( + crate::recurrence::total_cost(CostRate(5.0), Horizon(10.0), Cost(3.0)), + Cost(53.0) + ); + } + + #[test] + fn a_missing_input_yields_no_rate_not_zero() { + let mut inputs = known_inputs(); + inputs.summary_maintenance_cost_per_update = None; + assert_eq!(postprocess_plan_cost_rate(&inputs), None); + assert_eq!(pretransform_plan_cost_rate(&inputs), None); + // The baseline doesn't need maintenance and is still known. + assert!(raw_recompute_cost_rate(&inputs).is_some()); + let unknown = ExactCompositionCostInputs::unknown(known_inputs().provenance); + assert_eq!(raw_recompute_cost_rate(&unknown), None); + } + + #[test] + fn default_model_advertises_capabilities_but_no_statistics() { + assert_eq!( + DefaultCostModel.mixed_execution_capabilities(), + MixedExecutionCapabilities::ALL + ); + assert!(MixedExecutionCapabilities::NONE + .supports(CompositionPlacement::PostProcess) + .not()); + } + + trait Not { + fn not(self) -> bool; + } + impl Not for bool { + fn not(self) -> bool { + !self + } + } + // ── CSE sharing (issue #237, #223 stage 4) ────────────────────────── use asap_types::post_asap::{ diff --git a/crates/asap-aware-mapping/src/exact_composition.rs b/crates/asap-aware-mapping/src/exact_composition.rs new file mode 100644 index 00000000..5b23d854 --- /dev/null +++ b/crates/asap-aware-mapping/src/exact_composition.rs @@ -0,0 +1,673 @@ +//! [`ExactCompositionStrategy`] — composing an exact operator with a +//! summary plan across an explicit update/readout boundary (issue #171). +//! +//! ## The gap this closes +//! +//! `construct_summary_agg` already nests: a `SummaryAgg` recursively +//! realizes its child, so a KLL over an exact `Sum` accumulator, or a +//! `quantile(rate(...))` over a `Rate` accumulator, come out as one bound +//! DAG. What it cannot represent is an exact operator that is **not** +//! maintained summary state sitting next to a summary: +//! +//! - `max by (zone) (quantile_over_time(0.99, latency[5m]))` — the outer +//! `max` is an exact fold over the inner summary's *readout*. A `MinMax` +//! accumulator over that readout is data_state-illegal (a maintained summary +//! can't consume query-time values — see +//! `asap_types::post_asap::execution_data_state`), +//! and `avg` has no accumulator at all, so today either shape collapses +//! into one opaque `KeepPreAsap` that swallows the realizable inner +//! quantile. +//! - `quantile(0.99, deriv(m[5m]))` — the inner `deriv` is an exact, +//! per-sample transform with no accumulator form; the outer summary can +//! only consume it as an opaque raw `KeepPreAsap` blob today, with no +//! explicit "this row transform runs on the update path" node. +//! +//! [`SummaryExpr::ReadoutPostProcess`] and [`SummaryExpr::UpdateTransform`] +//! are the two data_state-explicit representations; this strategy is what +//! proposes them. +//! +//! ## Reference, don't select +//! +//! A composed candidate needs a child plan to compose *with* — the inner +//! quantile's own summary readout, say. This strategy deliberately does +//! **not** pick that child itself (the way `construct_summary_agg`'s +//! `realize_child` takes the head of the child's own ranking): a +//! [`Replacement::ExactComposition`] carries only the child *target* +//! (`ExactComposition::child_target`, the same `Rc` whose +//! `MemoGroup` in `PlanSpace` already holds every candidate for it). It is +//! [`PlanSpace::global_selection`](crate::replacement::PlanSpace::global_selection) +//! that commits the compatible parent/child pair — so the child's own +//! cost-model ranking, workload-wide effective consumer count, and shared +//! `Rc` identity (one inner summary serving two outer folds) all stay +//! correct, and a child that is also shared by an unrelated consumer is +//! maintained exactly once. `GlobalSelection::materialize` then links the +//! committed pair into one validated post-ASAP DAG. +//! +//! ## Proposal conditions +//! +//! A candidate is proposed only when all of these hold: +//! +//! - the target is a single-measure, `HAVING`-free exact aggregate; +//! - post-process: the child is a bindable aggregate that has at least one +//! readout-producing summary implementation (a sketch/sample/wavelet/ +//! model — the shapes a maintained accumulator can't sit above), and the +//! target's grouping keys resolve in the child's output schema; +//! transform: the target is a per-entity exact transform with no +//! accumulator form (its only implementation is `PassThrough`); +//! - the exact operator consumes only `Plain` values in its data_state — checked +//! again, structurally, when the pair is composed; +//! - the plugged-in [`CostModel`] advertises the matching +//! [`MixedExecutionCapabilities`](crate::cost_model::MixedExecutionCapabilities). +//! +//! `avg` gets a post-process candidate *and* keeps +//! [`crate::rewrite::AvgToSumOverCountStrategy`]'s rewrite in the same +//! group; the cost model picks between them, nothing here hard-codes one. +//! +//! ## What this strategy never does +//! +//! - Propose an `ExactPostProcess` for a position beneath a maintained +//! summary — data_state validation at composition rejects it as a typed +//! `ImplementError` regardless. +//! - Decide whether a composition is *worth it*: that is +//! `global_selection`'s job, using the issue's cost-units-per-second +//! formulas (see `crate::cost_model::postprocess_plan_cost_rate` and +//! siblings). Missing statistics keep the conservative `KeepPreAsap`. + +use std::rc::Rc; + +use asap_types::post_asap::execution_data_state::validate_execution_data_states_at; +use asap_types::post_asap::{ + exact_operator_output_schema, produced_data_state, CompositionOperator, ExactOperator, + ExecutionDataState, ExecutionDataStateError, SummaryExpr, SummaryNode, SummarySchema, + ValueOperator, +}; +use asap_types::pre_asap::agg_intent::AggIntent; +use asap_types::pre_asap::query_expr::{QueryExpr, Reduction}; +use asap_types::types::AccuracyTarget; + +use crate::cost_model::CostModel; +use crate::replacement::{ + bindable_intent, describe_intent, implementations_for_with, ImplementError, Implementation, + Replacement, ReplacementProvenance, ReplacementStrategy, ReplacementSubDAG, TargetSubDAG, +}; +use crate::{AccuracyModel, DefaultAccuracyModel, PropagationStats}; + +/// Which side of the update/readout boundary an [`ExactComposition`]'s +/// exact operator executes on — selects the `SummaryExpr` variant +/// [`ExactComposition::compose`] builds. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum CompositionPlacement { + /// [`SummaryExpr::ReadoutPostProcess`]: after the child's readout. + PostProcess, + /// [`SummaryExpr::UpdateTransform`]: on the update path, feeding + /// maintained state above. + Transform, +} + +impl CompositionPlacement { + /// The availability the composed operator consumes and produces. + pub fn data_state(self) -> ExecutionDataState { + match self { + Self::PostProcess => ExecutionDataState::READ_ROWS, + Self::Transform => ExecutionDataState::MAINTENANCE_ROWS, + } + } + + pub fn provenance(self) -> ReplacementProvenance { + match self { + Self::PostProcess => ReplacementProvenance::ExactPostProcess, + Self::Transform => ReplacementProvenance::ExactTransform, + } + } +} + +/// The payload of a [`Replacement::ExactComposition`] candidate: an exact +/// operator, the placement it runs at, and a *reference* to the child target +/// it composes over — never an already-selected child plan (see the module +/// docs' "Reference, don't select"). +#[derive(Debug, Clone)] +pub struct ExactComposition { + pub placement: CompositionPlacement, + pub op: ExactOperator, + /// The pre-ASAP child the operator consumes; its `MemoGroup` holds the + /// candidates `global_selection` may commit this composition with. + pub child_target: Rc, + /// The composed node's output schema — the target's own pre-ASAP + /// output schema, lifted with every column `Plain` (an exact operator + /// only ever produces plain values). + pub schema: SummarySchema, +} + +impl ExactComposition { + /// Can `child` legally be this composition's input? Phase legality + /// (the child's produced data_state — a `KeepPreAsap` leaf takes the + /// phase this edge assigns) plus the plain-operand rule, checked + /// through the same schema derivation [`Self::compose`] uses. + pub fn accepts_child(&self, child: &SummaryNode) -> bool { + let phase_ok = match produced_data_state(&child.expr) { + None => true, + Some(avail) => avail == self.placement.data_state(), + }; + phase_ok && exact_operator_output_schema(&self.op, &child.schema).is_ok() + } + + /// Build the composed, data_state-validated node over `child`. Every edge of + /// the result (including everything beneath `child`) is checked by + /// `asap_types::post_asap::validate_execution_data_states`; an illegal + /// placement is a typed [`ImplementError::ExecutionDataState`], never deferred to a + /// runtime. + pub fn compose(&self, child: Rc) -> Result, ImplementError> { + self.compose_with_accuracy(child, &DefaultAccuracyModel) + } + + /// Compose using the caller's accuracy algebra. Exact operators do not + /// erase an approximate child's error: supported folds propagate it; + /// unsupported folds fail closed with a typed accuracy error. + pub fn compose_with_accuracy( + &self, + child: Rc, + accuracy_model: &dyn AccuracyModel, + ) -> Result, ImplementError> { + if let Some(produced) = produced_data_state(&child.expr) { + if produced != self.placement.data_state() { + let edge = match self.placement { + CompositionPlacement::Transform => "UpdateTransform.child", + CompositionPlacement::PostProcess => "ReadoutPostProcess.child", + }; + return Err(ImplementError::ExecutionDataState( + ExecutionDataStateError::IllegalChildPhase { + edge, + child: produced, + }, + )); + } + } + let schema = exact_operator_output_schema(&self.op, &child.schema)?; + let guarantee = match &child.guarantee { + None => None, + Some(input) => { + let operator = match &self.op { + ExactOperator::Aggregate { measures, .. } => match measures.as_slice() { + [AggIntent::Sum { .. }] => CompositionOperator::ExactSum, + [AggIntent::Min { .. } | AggIntent::Max { .. } | AggIntent::Avg { .. }] => { + CompositionOperator::ExactExtremum + } + // This placeholder is only used by the model's exact-input + // fast path. Approximate inputs correctly fail closed. + [_] => CompositionOperator::Lipschitz { constant: 1.0 }, + _ => CompositionOperator::Lipschitz { constant: 1.0 }, + }, + _ => CompositionOperator::Lipschitz { constant: 1.0 }, + }; + Some(accuracy_model.propagate( + &operator, + std::slice::from_ref(input), + None, + &PropagationStats::default(), + )?) + } + }; + let expr = match self.placement { + CompositionPlacement::PostProcess => SummaryExpr::ReadoutPostProcess { + child, + op: ValueOperator::Exact(self.op.clone()), + }, + CompositionPlacement::Transform => SummaryExpr::UpdateTransform { + child, + op: ValueOperator::Exact(self.op.clone()), + }, + }; + let node = Rc::new(SummaryNode { + expr, + schema, + guarantee, + }); + validate_execution_data_states_at(&node, self.placement.data_state())?; + Ok(node) + } + + /// Structural identity for `MemoGroup` dedup: same placement, same + /// operator, same child `Rc`. + pub(crate) fn same_as(&self, other: &Self) -> bool { + self.placement == other.placement + && self.op == other.op + && Rc::ptr_eq(&self.child_target, &other.child_target) + } +} + +/// Which exact reducers may run as a query-time fold over readout rows. +/// `Count` only at `Exact` accuracy (an approximate count is a sketch +/// target, not an exact fold). +fn is_post_process_reducer(intent: &AggIntent) -> bool { + matches!( + intent, + AggIntent::Sum { .. } + | AggIntent::Min { .. } + | AggIntent::Max { .. } + | AggIntent::Avg { .. } + | AggIntent::StdDev { .. } + | AggIntent::Variance { .. } + | AggIntent::Count { + accuracy: AccuracyTarget::Exact + } + ) +} + +/// Does `implementation` need a `SummaryEstimate` readout to yield a value +/// — i.e. is it a shape a maintained accumulator can't legally sit above? +fn needs_readout(implementation: &Implementation) -> bool { + matches!( + implementation, + Implementation::Sketch(_) + | Implementation::Sample { .. } + | Implementation::Wavelet { .. } + | Implementation::StatModel { .. } + ) +} + +/// The `(op, child)` of a post-process-shaped target, or `None`. +fn post_process_shape( + root: &QueryExpr, + cost_model: &dyn CostModel, +) -> Option<(ExactOperator, Rc, AggIntent)> { + let QueryExpr::Aggregate { + reduction, + measures, + output_names, + having: None, + child, + } = root + else { + return None; + }; + let Reduction::Reduce(by) = reduction else { + return None; + }; + if by.is_without() { + return None; + } + let [intent] = measures.as_slice() else { + return None; + }; + if !is_post_process_reducer(intent) { + return None; + } + let child_intent = bindable_intent(child)?; + if !implementations_for_with(child_intent, cost_model) + .iter() + .any(needs_readout) + { + return None; + } + // Grouping keys must resolve in the child's output schema — the same + // derivation the composed node's own schema will use. + root.output_schema().ok()?; + Some(( + ExactOperator::Aggregate { + reduction: reduction.clone(), + measures: measures.clone(), + output_names: output_names.clone(), + having: None, + }, + Rc::clone(child), + intent.clone(), + )) +} + +/// The `(op, child)` of a transform-shaped target — a per-entity exact +/// transform with no accumulator form — or `None`. +fn transform_shape( + root: &QueryExpr, + cost_model: &dyn CostModel, +) -> Option<(ExactOperator, Rc, AggIntent)> { + let QueryExpr::Aggregate { + reduction: Reduction::PerEntity, + measures, + output_names, + having: None, + child, + } = root + else { + return None; + }; + let [intent] = measures.as_slice() else { + return None; + }; + if !intent.is_per_series() { + return None; + } + // Exact accumulators (`Rate`/`Increase`) are already directly nestable + // as `SummaryAgg(ExactAggregate)`; only a pass-through transform needs + // an explicit update-path node. + if implementations_for_with(intent, cost_model) + .iter() + .any(|i| *i != Implementation::PassThrough) + { + return None; + } + root.output_schema().ok()?; + Some(( + ExactOperator::Aggregate { + reduction: Reduction::PerEntity, + measures: measures.clone(), + output_names: output_names.clone(), + having: None, + }, + Rc::clone(child), + intent.clone(), + )) +} + +/// Proposes [`Replacement::ExactComposition`] candidates — see the module +/// docs. Holds a [`CostModel`] only to ask it which mixed-execution shapes +/// the runtime advertises and which implementations the child has; it +/// never uses it to *rank* anything. +pub struct ExactCompositionStrategy<'a> { + cost_model: &'a dyn CostModel, +} + +static DEFAULT_COST_MODEL: crate::cost_model::DefaultCostModel = + crate::cost_model::DefaultCostModel; + +impl ExactCompositionStrategy<'static> { + /// A strategy consulting the built-in [`DefaultCostModel`](crate::cost_model::DefaultCostModel). + pub fn default_cost_model() -> Self { + Self { + cost_model: &DEFAULT_COST_MODEL, + } + } +} + +impl<'a> ExactCompositionStrategy<'a> { + pub fn new(cost_model: &'a dyn CostModel) -> Self { + Self { cost_model } + } + + fn candidates(&self, target: &TargetSubDAG<'_>) -> Vec { + let capabilities = self.cost_model.mixed_execution_capabilities(); + let Ok(schema) = target.root.output_schema() else { + return Vec::new(); + }; + let schema = asap_types::post_asap::execution_data_state::lift_plain(&schema); + let mut out = Vec::new(); + + if capabilities.supports(CompositionPlacement::PostProcess) { + if let Some((op, child, intent)) = post_process_shape(target.root, self.cost_model) { + let child_desc = describe_intent( + bindable_intent(&child).expect("checked by post_process_shape"), + ); + out.push(ReplacementSubDAG { + strategy: "ExactCompositionStrategy", + replacement: Replacement::ExactComposition(ExactComposition { + placement: CompositionPlacement::PostProcess, + op, + child_target: child, + schema: schema.clone(), + }), + provenance: ReplacementProvenance::ExactPostProcess, + rationale: format!( + "{} is an exact fold whose input is the readout of {} — a maintained \ + accumulator cannot consume query-time values, so instead of collapsing \ + the whole tree into KeepPreAsap this applies the fold as an \ + ExactPostProcess over whichever summary readout global_selection \ + commits for the child target (asap_aware_mapping::exact_composition)", + describe_intent(&intent), + child_desc + ), + }); + } + } + + if capabilities.supports(CompositionPlacement::Transform) { + if let Some((op, child, intent)) = transform_shape(target.root, self.cost_model) { + out.push(ReplacementSubDAG { + strategy: "ExactCompositionStrategy", + replacement: Replacement::ExactComposition(ExactComposition { + placement: CompositionPlacement::Transform, + op, + child_target: child, + schema, + }), + provenance: ReplacementProvenance::ExactTransform, + rationale: format!( + "{} is an exact per-entity transform with no accumulator form; as an \ + explicit ExactTransform on the update path its output can feed a \ + maintained summary above it instead of being handed over as an opaque \ + raw KeepPreAsap blob (asap_aware_mapping::exact_composition)", + describe_intent(&intent) + ), + }); + } + } + out + } +} + +impl ReplacementStrategy for ExactCompositionStrategy<'_> { + fn matches(&self, target: &TargetSubDAG<'_>) -> bool { + !self.candidates(target).is_empty() + } + + fn replacements(&self, target: &TargetSubDAG<'_>) -> Vec { + self.candidates(target) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::cost_model::{DefaultCostModel, MixedExecutionCapabilities}; + use crate::replacement::keep_pre_asap; + use asap_types::post_asap::{ExecutionDataStateError, SketchAlgorithm, SummaryFamilyType}; + use asap_types::pre_asap::agg_intent::default_quantile; + use asap_types::pre_asap::query_expr::Source; + use asap_types::pre_asap::schema::{Column, DataType, Schema}; + + fn metric_scan(labels: &[&str]) -> QueryExpr { + let mut columns = vec![ + Column::new("ts", DataType::Timestamp, false), + Column::new("value", DataType::Float64, false), + ]; + columns.extend(labels.iter().map(|n| Column::new(*n, DataType::Utf8, true))); + QueryExpr::Scan { + source: Source::TimeSeries { metric: "m".into() }, + predicates: vec![], + schema: Schema::with_time_index(columns, 0, vec![]), + } + } + + fn agg(by: Vec, intent: AggIntent, child: QueryExpr) -> QueryExpr { + QueryExpr::Aggregate { + reduction: Reduction::by(by), + measures: vec![intent], + output_names: vec![], + having: None, + child: Rc::new(child), + } + } + + fn per_entity(intent: AggIntent, child: QueryExpr) -> QueryExpr { + QueryExpr::Aggregate { + reduction: Reduction::PerEntity, + measures: vec![intent], + output_names: vec![], + having: None, + child: Rc::new(child), + } + } + + /// `max by (zone) (quantile by (zone, host) (m))`. + fn max_over_quantile() -> Rc { + let inner = agg( + vec![2, 3], + default_quantile(0.99), + metric_scan(&["zone", "host"]), + ); + Rc::new(agg(vec![0], AggIntent::Max { col: None }, inner)) + } + + #[test] + fn proposes_post_process_for_max_over_quantile() { + let root = max_over_quantile(); + let target = TargetSubDAG::new(&root); + let strategy = ExactCompositionStrategy::default_cost_model(); + assert!(strategy.matches(&target)); + let candidates = strategy.replacements(&target); + assert_eq!(candidates.len(), 1); + let Replacement::ExactComposition(comp) = &candidates[0].replacement else { + panic!( + "expected a composition, got {:?}", + candidates[0].replacement + ); + }; + assert_eq!(comp.placement, CompositionPlacement::PostProcess); + assert_eq!( + candidates[0].provenance, + ReplacementProvenance::ExactPostProcess + ); + let QueryExpr::Aggregate { child, .. } = root.as_ref() else { + unreachable!() + }; + assert!( + Rc::ptr_eq(&comp.child_target, child), + "the candidate references the child target's own Rc — nothing selected" + ); + let names: Vec<_> = comp.schema.fields.iter().map(|f| f.name.as_str()).collect(); + assert_eq!(names, vec!["zone", "max"]); + } + + #[test] + fn proposes_post_process_for_avg_over_quantile_alongside_the_rewrite() { + let inner = agg(vec![2], default_quantile(0.99), metric_scan(&["zone"])); + let root = Rc::new(agg(vec![0], AggIntent::Avg { col: None }, inner)); + let target = TargetSubDAG::new(&root); + assert_eq!( + ExactCompositionStrategy::default_cost_model() + .replacements(&target) + .len(), + 1 + ); + // `avg` competes with AvgToSumOverCountStrategy in the same group. + assert!(crate::rewrite::AvgToSumOverCountStrategy.matches(&target)); + } + + #[test] + fn proposes_transform_for_a_per_entity_pass_through_over_raw_input() { + let root = Rc::new(per_entity(AggIntent::Deriv, metric_scan(&["zone"]))); + let target = TargetSubDAG::new(&root); + let candidates = ExactCompositionStrategy::default_cost_model().replacements(&target); + assert_eq!(candidates.len(), 1); + assert_eq!( + candidates[0].provenance, + ReplacementProvenance::ExactTransform + ); + } + + #[test] + fn does_not_propose_for_shapes_already_covered_by_accumulators() { + // sum by (zone) over an exact Sum child: the child has no readout, + // so SummaryAgg(Sum) over SummaryAgg(Sum) is already legal. + let inner = agg( + vec![2, 3], + AggIntent::Sum { col: None }, + metric_scan(&["zone", "host"]), + ); + let root = Rc::new(agg(vec![0], AggIntent::Sum { col: None }, inner)); + assert!(!ExactCompositionStrategy::default_cost_model().matches(&TargetSubDAG::new(&root))); + // rate is an exact accumulator — directly nestable, no transform. + let rate = Rc::new(per_entity(AggIntent::Rate, metric_scan(&[]))); + assert!(!ExactCompositionStrategy::default_cost_model().matches(&TargetSubDAG::new(&rate))); + // A sketch-capable outer intent is not an exact fold. + let inner = agg(vec![2], default_quantile(0.5), metric_scan(&["zone"])); + let root = Rc::new(agg(vec![0], default_quantile(0.99), inner)); + assert!(!ExactCompositionStrategy::default_cost_model().matches(&TargetSubDAG::new(&root))); + } + + struct NoMixedExecution; + impl CostModel for NoMixedExecution { + fn rank_candidates( + &self, + _intent: &AggIntent, + candidates: &[SketchAlgorithm], + ) -> Vec { + candidates.to_vec() + } + fn mixed_execution_capabilities(&self) -> MixedExecutionCapabilities { + MixedExecutionCapabilities::NONE + } + } + + #[test] + fn a_runtime_without_the_capability_gets_no_candidate() { + let root = max_over_quantile(); + let target = TargetSubDAG::new(&root); + let strategy = ExactCompositionStrategy::new(&NoMixedExecution); + assert!(!strategy.matches(&target)); + assert!(strategy.replacements(&target).is_empty()); + let deriv = Rc::new(per_entity(AggIntent::Deriv, metric_scan(&[]))); + assert!(!strategy.matches(&TargetSubDAG::new(&deriv))); + } + + #[test] + fn compose_rejects_a_maintained_state_child_for_a_post_process() { + let root = max_over_quantile(); + let target = TargetSubDAG::new(&root); + let candidates = ExactCompositionStrategy::default_cost_model().replacements(&target); + let Replacement::ExactComposition(comp) = &candidates[0].replacement else { + unreachable!() + }; + // A bare SummaryAgg (state, no readout) is not a legal post-process + // input — the operator would be consuming sketch state. + let state_child = + crate::replacement::realize_child(&comp.child_target, &DefaultCostModel).unwrap(); + let SummaryExpr::SummaryEstimate { summary_input, .. } = &state_child.expr else { + panic!("expected the child to realize to a readout"); + }; + assert!(!comp.accepts_child(summary_input)); + assert!(matches!( + comp.compose(Rc::clone(summary_input)), + Err(ImplementError::ExecutionDataState( + ExecutionDataStateError::IllegalChildPhase { .. } + )) + )); + // The readout itself is accepted and composes to a plain schema. + assert!(comp.accepts_child(&state_child)); + let composed = comp.compose(state_child).unwrap(); + assert!(matches!( + composed.expr, + SummaryExpr::ReadoutPostProcess { .. } + )); + assert!(composed + .schema + .fields + .iter() + .all(|f| matches!(f.dtype, SummaryFamilyType::Plain(_)))); + } + + #[test] + fn compose_rejects_a_readout_child_for_a_transform() { + let inner = agg(vec![2], default_quantile(0.99), metric_scan(&["zone"])); + let root = Rc::new(per_entity(AggIntent::Deriv, inner)); + let candidates = + ExactCompositionStrategy::default_cost_model().replacements(&TargetSubDAG::new(&root)); + let Replacement::ExactComposition(comp) = &candidates[0].replacement else { + unreachable!() + }; + let readout = + crate::replacement::realize_child(&comp.child_target, &DefaultCostModel).unwrap(); + assert!(!comp.accepts_child(&readout)); + assert!(matches!( + comp.compose(readout), + Err(ImplementError::ExecutionDataState( + ExecutionDataStateError::IllegalChildPhase { .. } + )) + )); + // Raw update input is fine. + let raw = keep_pre_asap(&comp.child_target).unwrap(); + assert!(comp.accepts_child(&raw)); + assert!(matches!( + comp.compose(raw).unwrap().expr, + SummaryExpr::UpdateTransform { .. } + )); + } +} diff --git a/crates/asap-aware-mapping/src/explanation.rs b/crates/asap-aware-mapping/src/explanation.rs index bcfee70a..3900c729 100644 --- a/crates/asap-aware-mapping/src/explanation.rs +++ b/crates/asap-aware-mapping/src/explanation.rs @@ -216,6 +216,13 @@ pub enum ExplanationKind { /// cross-subpopulation reuse entries, all the same underlying structural /// fact. CommonSubexpressionReuse, + /// A `TargetSubDAG`'s candidate list contains at least one + /// [`Replacement::ExactComposition`] — + /// [`crate::exact_composition::ExactCompositionStrategy`] found an exact + /// operator that can be composed with a summary plan across an explicit + /// update/readout boundary instead of collapsing the whole tree into + /// `KeepPreAsap` (issue #171). + ExactComposition, } /// Why a [`Replacement`] of `kind` exists at `location` (a human-readable @@ -319,6 +326,15 @@ fn findings_from_plan_space(space: &PlanSpace) -> Vec) -> Vec Option { + let reasons: Vec<&str> = group + .candidates + .iter() + .filter(|c| matches!(c.replacement, Replacement::ExactComposition(_))) + .map(|c| c.rationale.as_str()) + .collect(); + if reasons.is_empty() { + None + } else { + Some(reasons.join("; ")) + } +} + /// Does `group`'s candidate list contain a genuine sketch-family realization? /// If so, the finding's `reason` is every such candidate's own `rationale`, /// joined — this module does not invent new prose to restate why a candidate diff --git a/crates/asap-aware-mapping/src/lib.rs b/crates/asap-aware-mapping/src/lib.rs index 613d5061..1a9c7d1d 100644 --- a/crates/asap-aware-mapping/src/lib.rs +++ b/crates/asap-aware-mapping/src/lib.rs @@ -184,6 +184,7 @@ pub mod accuracy; pub mod accuracy_reconciliation; pub mod cost_model; +pub mod exact_composition; pub mod explanation; pub mod grouping; pub mod recurrence; @@ -198,7 +199,12 @@ pub use accuracy::{ PropagationStats, }; pub use accuracy_reconciliation::AccuracyReconciliationStrategy; -pub use cost_model::{CostModel, DefaultCostModel}; +pub use cost_model::{ + postprocess_plan_cost_rate, pretransform_plan_cost_rate, raw_recompute_cost_rate, CostModel, + CostProvenance, CostUnit, DefaultCostModel, ExactCompositionCostInputs, + ExactCompositionCostRequest, MixedExecutionCapabilities, +}; +pub use exact_composition::{CompositionPlacement, ExactComposition, ExactCompositionStrategy}; pub use explanation::{ explain_replacements, explain_replacements_with, ExplanationKind, ReplacementExplanation, }; @@ -210,11 +216,11 @@ pub use recurrence::{ }; pub use replacement::{ default_strategies, default_strategies_with, search_workload, search_workload_with, - search_workload_with_targets, summary_candidates, GlobalSelection, ImplementError, - Implementation, Matcher, MemoGroup, PlanSpace, Proposals, RankedGroup, RecurrenceProfileMap, - RejectedCandidate, Replacement, ReplacementProvenance, ReplacementStrategy, ReplacementSubDAG, - SelectedGroup, SharedSubtreeStrategy, SketchAlgorithmStrategy, TargetSubDAG, - MAX_SEARCH_ITERATIONS, + search_workload_with_targets, summary_candidates, CompositionDecision, GlobalSelection, + ImplementError, Implementation, Matcher, MemoGroup, PlanSpace, Proposals, RankedGroup, + RecurrenceProfileMap, RejectedCandidate, Replacement, ReplacementProvenance, + ReplacementStrategy, ReplacementSubDAG, SelectedGroup, SharedSubtreeStrategy, + SketchAlgorithmStrategy, TargetSubDAG, MAX_SEARCH_ITERATIONS, }; pub use rewrite::AvgToSumOverCountStrategy; pub use topk_reuse::TopKLimitReuseStrategy; diff --git a/crates/asap-aware-mapping/src/replacement.rs b/crates/asap-aware-mapping/src/replacement.rs index 5835adac..89860123 100644 --- a/crates/asap-aware-mapping/src/replacement.rs +++ b/crates/asap-aware-mapping/src/replacement.rs @@ -345,15 +345,17 @@ //! multi-group joint optimization beyond this per-site recurrence is left //! for whenever that changes. +use std::cell::RefCell; use std::collections::{HashMap, HashSet, VecDeque}; -use asap_types::post_asap::{AccuracyError, CompositionOperator, GuaranteeSource, ResultGuarantee}; use asap_types::post_asap::{ - ExactKind, ExactParams, GroupingStrategy, SamplingKind, SamplingParams, SketchAlgorithm, - SketchKind, SketchParams, SketchQuery as PostAsapSketchQuery, StatModelKind, StatModelParams, - SummaryExpr, SummaryFamilyType, SummaryField, SummaryNode, SummarySchema, WaveletKind, - WaveletParams, + validate_execution_data_states_at, ExactKind, ExactOperatorSchemaError, ExactParams, + ExecutionDataState, ExecutionDataStateError, GroupingStrategy, SamplingKind, SamplingParams, + SketchAlgorithm, SketchKind, SketchParams, SketchQuery as PostAsapSketchQuery, StatModelKind, + StatModelParams, SummaryExpr, SummaryFamilyType, SummaryField, SummaryNode, SummarySchema, + WaveletKind, WaveletParams, }; +use asap_types::post_asap::{AccuracyError, CompositionOperator, GuaranteeSource, ResultGuarantee}; use asap_types::pre_asap::agg_intent::{agg_is_mergeable, AggIntent}; use asap_types::pre_asap::cse::{share_common_subtrees, structural_hash, HashCache}; use asap_types::pre_asap::expr_ir::ColumnRef; @@ -370,8 +372,13 @@ use crate::accuracy::{ KLL_RANK_ERROR_EXPONENT_99, }; use crate::accuracy_reconciliation::AccuracyReconciliationStrategy; -use crate::cost_model::{CostModel, CseCandidate, DefaultCostModel, ShareDecision}; +use crate::cost_model::{ + raw_recompute_cost_rate, CostModel, CseCandidate, DefaultCostModel, ExactCompositionCostInputs, + ExactCompositionCostRequest, ShareDecision, +}; +use crate::exact_composition::{CompositionPlacement, ExactComposition, ExactCompositionStrategy}; use crate::grouping::HydraGroupingStrategy; +use crate::recurrence::CostRate; use crate::recurrence::{ evaluation_rate_of, Horizon, RecurrenceError, RecurrenceProfile, RootRecurrence, UpdateRate, }; @@ -396,6 +403,15 @@ pub enum ImplementError { /// records it as a [`RejectedCandidate`] instead of a candidate. #[error("accuracy-illegal candidate: {0}")] Accuracy(#[from] AccuracyError), + /// A constructed plan violates the update/readout phase contract + /// (issue #171) — e.g. a summary readout placed beneath a maintained + /// `SummaryAgg`. Detected at construction, never at runtime. + #[error("execution-data_state violation in post-ASAP plan: {0}")] + ExecutionDataState(#[from] ExecutionDataStateError), + /// An `ExactOperator`'s output schema could not be derived over its + /// child — the child carries summary state the operator can't read. + #[error("exact operator schema derivation failed: {0}")] + ExactOperatorSchema(#[from] ExactOperatorSchemaError), } /// A pre-ASAP sub-DAG a [`ReplacementStrategy`] knows how to replace. @@ -455,6 +471,15 @@ pub enum Replacement { /// different from the target's own `root` (e.g. sharing vs. not sharing /// a subtree) but semantically equivalent to it. Rewrite(Rc), + /// An exact operator composed over another target's *own* selected + /// decision across an explicit update/readout boundary (issue #171): + /// `ExactPostProcess` over a child's summary readout, or + /// `ExactTransform` feeding a maintained summary above. Carries only a + /// reference to the child target — [`PlanSpace::global_selection`] + /// commits the compatible parent/child pair and + /// [`GlobalSelection::materialize`] links it into one validated + /// `SummaryNode`. See [`crate::exact_composition`]. + ExactComposition(ExactComposition), } /// One candidate replacement for a [`TargetSubDAG`], plus a human-readable @@ -496,6 +521,12 @@ pub enum ReplacementProvenance { /// regardless, so pricing it like a full independent rebuild would be /// the wrong shape of cost, not just the wrong number. AccuracyReconciliation, + /// [`Replacement::ExactComposition`] with + /// [`CompositionPlacement::PostProcess`] (issue #171). + ExactPostProcess, + /// [`Replacement::ExactComposition`] with + /// [`CompositionPlacement::Transform`] (issue #171). + ExactTransform, } /// A candidate a strategy considered for a target but refused to propose on @@ -521,6 +552,7 @@ pub struct RejectedCandidate { pub struct Proposals { pub candidates: Vec, pub rejected: Vec, + domain_error: Option, } /// A replacement strategy: given a [`TargetSubDAG`], does this strategy have @@ -566,6 +598,7 @@ pub trait ReplacementStrategy { Proposals { candidates: self.replacements(target), rejected: Vec::new(), + domain_error: None, } } } @@ -1369,6 +1402,22 @@ impl<'a> SketchAlgorithmStrategy<'a> { ); } } + if proposals.candidates.is_empty() { + if let Some(error) = &proposals.domain_error { + if let Ok(node) = keep_pre_asap(root) { + proposals.candidates.push(ReplacementSubDAG { + strategy: "SketchAlgorithmStrategy", + replacement: Replacement::Summary(node), + provenance: ReplacementProvenance::SummaryImplementation, + rationale: format!( + "{} stays pre-ASAP because summary construction crosses an illegal \ + execution-data_state boundary ({error})", + describe_intent(intent) + ), + }); + } + } + } proposals } } @@ -1390,7 +1439,10 @@ impl Proposals { description: rationale, error, }), - Err(ImplementError::Schema(_)) => {} + Err(ImplementError::ExecutionDataState(error)) => { + self.domain_error.get_or_insert(error); + } + Err(ImplementError::Schema(_) | ImplementError::ExactOperatorSchema(_)) => {} } } } @@ -1538,10 +1590,10 @@ pub(crate) fn realize_child_with( .. }) => Ok(node), Some(ReplacementSubDAG { - replacement: Replacement::Rewrite(_), + replacement: Replacement::Rewrite(_) | Replacement::ExactComposition(_), .. }) => { - unreachable!("SketchAlgorithmStrategy never returns a Rewrite candidate") + unreachable!("SketchAlgorithmStrategy never returns a Rewrite/composition candidate") } // No candidate at all: `root` isn't `bindable_intent` shape (or its // intent has no realization `implementations_for_with` can't @@ -1762,6 +1814,10 @@ fn construct_summary_agg( // finalized value does. An exact accumulator's state is its value. guarantee: if estimate { None } else { guarantee.clone() }, }); + // Phase contract (issue #171): a maintained summary consumes update-path + // values or exact accumulator state — never a query-time readout. A + // typed error here, at construction; the caller decides the fallback. + validate_execution_data_states_at(&agg, ExecutionDataState::MAINTENANCE_SUMMARY)?; match query { // The readout: downstream of the estimate the schema is the plain // pre-ASAP row shape again (the summary-state type does not @@ -2079,10 +2135,13 @@ impl MemoGroup { (Replacement::Summary(existing_node), Replacement::Summary(node)) => { is_duplicate_summary(existing_node, node) } - // A `Rewrite` and a `Summary` are never the same candidate — - // they're different `Replacement` variants entirely. - (Replacement::Rewrite(_), Replacement::Summary(_)) - | (Replacement::Summary(_), Replacement::Rewrite(_)) => false, + ( + Replacement::ExactComposition(existing), + Replacement::ExactComposition(candidate), + ) => existing.same_as(candidate), + // Different `Replacement` variants are never the same + // candidate. + _ => false, } }); if is_duplicate { @@ -2618,7 +2677,7 @@ fn rank_group<'a>(group: &'a MemoGroup, cost_model: &dyn CostModel) -> Vec<&'a R .iter() .map(|c| match &c.replacement { Replacement::Summary(node) => sketch_kind_of(node), - Replacement::Rewrite(_) => None, + Replacement::Rewrite(_) | Replacement::ExactComposition(_) => None, }) .collect(); if let Some(kinds) = kinds { @@ -2626,7 +2685,7 @@ fn rank_group<'a>(group: &'a MemoGroup, cost_model: &dyn CostModel) -> Vec<&'a R ranked.sort_by_key(|c| { let kind = match &c.replacement { Replacement::Summary(node) => sketch_kind_of(node), - Replacement::Rewrite(_) => None, + Replacement::Rewrite(_) | Replacement::ExactComposition(_) => None, }; kind.and_then(|k| order.iter().position(|o| *o == k)) .unwrap_or(usize::MAX) @@ -2754,6 +2813,32 @@ pub struct SelectedGroup<'a> { /// registered strategy proposed anything for (mirrors /// [`MemoGroup::candidates`] being possibly empty). pub chosen: Option<&'a ReplacementSubDAG>, + /// When `chosen` is a [`Replacement::ExactComposition`]: the child + /// decision it was committed together with, and the cost comparison + /// that justified it — the explicit target-to-decision provenance + /// chain (issue #171). + pub composition: Option>, +} + +/// Why [`PlanSpace::global_selection`] committed an exact composition at a +/// site: which child candidate it composes with, and the +/// cost-units-per-second comparison against the raw fallback that it won. +#[derive(Debug)] +pub struct CompositionDecision<'a> { + /// The child target the composed operator consumes. + pub child_target: &'a Rc, + /// For a post-process: the child's own candidate committed alongside + /// (the summary readout the operator folds). `None` for an update-path + /// transform, whose input is raw update data — its cost is charged to + /// the maintained summary *above* it instead. + pub child_candidate: Option<&'a ReplacementSubDAG>, + /// The composed plan's recurring rate — `postprocess_plan_cost_rate` + /// or `pretransform_plan_cost_rate`. + pub cost_rate: CostRate, + /// `raw_recompute_cost_rate` — the `KeepPreAsap` baseline it beat. + pub baseline_rate: CostRate, + /// The statistics (and their provenance) both rates were computed from. + pub inputs: ExactCompositionCostInputs, } /// [`PlanSpace::global_selection`]'s result: one [`SelectedGroup`] per @@ -2763,6 +2848,10 @@ pub struct SelectedGroup<'a> { pub struct GlobalSelection<'a> { order: Vec<*const QueryExpr>, groups: HashMap<*const QueryExpr, SelectedGroup<'a>>, + /// [`Self::materialize`]'s memo — one bound node per target for the + /// life of this selection, so two parents composing over one shared + /// child get the *same* `Rc`. + materialized: RefCell>>, } impl<'a> GlobalSelection<'a> { @@ -2777,6 +2866,272 @@ impl<'a> GlobalSelection<'a> { pub fn for_target(&self, target: &Rc) -> Option<&SelectedGroup<'a>> { self.groups.get(&Rc::as_ptr(target)) } + + /// Link this selection's per-site decisions into one data_state-validated + /// post-ASAP DAG rooted at `target` — the one place a committed + /// composition's child *reference* becomes an actual `Rc` + /// edge (issue #171). `None` if `target` is not a discovered site. + /// + /// Per site: a [`Replacement::ExactComposition`] composes over its + /// child target's own materialization; a [`Replacement::Summary`] is + /// re-linked so its `SummaryAgg` child is the child target's own + /// materialization whenever that is phase-legal beneath maintenance + /// (so a child that chose an `ExactTransform` actually ends up under + /// the summary); a [`Replacement::Rewrite`] or an unmatched site stays + /// the conservative `KeepPreAsap`. Memoized by target identity, so a + /// shared inner summary is one `Rc` no matter how many roots reach it. + pub fn materialize( + &self, + target: &Rc, + ) -> Result>, ImplementError> { + if !self.groups.contains_key(&Rc::as_ptr(target)) { + return Ok(None); + } + self.materialize_inner(target).map(Some) + } + + fn materialize_inner(&self, target: &Rc) -> Result, ImplementError> { + let ptr = Rc::as_ptr(target); + if let Some(node) = self.materialized.borrow().get(&ptr) { + return Ok(Rc::clone(node)); + } + let node = match self + .groups + .get(&ptr) + .and_then(|sel| sel.chosen) + .map(|c| &c.replacement) + { + None => keep_pre_asap(target)?, + Some(Replacement::Rewrite(rewritten)) => keep_pre_asap(rewritten)?, + Some(Replacement::Summary(node)) => self.relink_summary(node, target)?, + Some(Replacement::ExactComposition(composition)) => { + let child = self.materialize_inner(&composition.child_target)?; + let child = if composition.accepts_child(&child) { + child + } else { + keep_pre_asap(&composition.child_target)? + }; + composition.compose(child)? + } + }; + self.materialized.borrow_mut().insert(ptr, Rc::clone(&node)); + Ok(node) + } + + /// Re-link a bound `Summary` candidate's `SummaryAgg` child to the + /// child target's own materialization when that is legal beneath + /// maintenance; otherwise keep the candidate exactly as constructed. + fn relink_summary( + &self, + node: &Rc, + target: &Rc, + ) -> Result, ImplementError> { + let QueryExpr::Aggregate { + child: pre_child, .. + } = target.as_ref() + else { + return Ok(Rc::clone(node)); + }; + if !self.groups.contains_key(&Rc::as_ptr(pre_child)) { + return Ok(Rc::clone(node)); + } + let new_child = self.materialize_inner(pre_child)?; + Ok(relink_agg_child(node, &new_child)) + } +} + +/// Rebuild `node` (a `SummaryAgg`, possibly under a `SummaryEstimate`) with +/// `new_child` as the `SummaryAgg`'s child, if the result still validates +/// as maintained state; otherwise return `node` unchanged. +fn relink_agg_child(node: &Rc, new_child: &Rc) -> Rc { + match &node.expr { + SummaryExpr::SummaryEstimate { + summary_input, + query, + } => { + let inner = relink_agg_child(summary_input, new_child); + if Rc::ptr_eq(&inner, summary_input) { + return Rc::clone(node); + } + Rc::new(SummaryNode { + expr: SummaryExpr::SummaryEstimate { + summary_input: inner, + query: query.clone(), + }, + schema: node.schema.clone(), + guarantee: node.guarantee.clone(), + }) + } + SummaryExpr::SummaryAgg { + child, + family, + col, + reduction, + grouping, + } => { + if Rc::ptr_eq(child, new_child) { + return Rc::clone(node); + } + let rebuilt = Rc::new(SummaryNode { + expr: SummaryExpr::SummaryAgg { + child: Rc::clone(new_child), + family: family.clone(), + col: col.clone(), + reduction: reduction.clone(), + grouping: grouping.clone(), + }, + schema: node.schema.clone(), + guarantee: node.guarantee.clone(), + }); + match validate_execution_data_states_at( + &rebuilt, + ExecutionDataState::MAINTENANCE_SUMMARY, + ) { + Ok(_) => rebuilt, + Err(_) => Rc::clone(node), + } + } + _ => Rc::clone(node), + } +} + +/// The maintained `SummaryAgg` a bound `Summary` candidate builds (under +/// its `SummaryEstimate` readout, if any) — the summary an `ExactTransform` +/// beneath it feeds, for `pretransform_plan_cost_rate`. +fn maintained_summary(node: &Rc) -> Option<&Rc> { + match &node.expr { + SummaryExpr::SummaryEstimate { summary_input, .. } => maintained_summary(summary_input), + SummaryExpr::SummaryAgg { .. } => Some(node), + _ => None, + } +} + +fn is_composition_candidate(candidate: &ReplacementSubDAG) -> bool { + matches!(candidate.replacement, Replacement::ExactComposition(_)) +} + +/// Everything [`PlanSpace::global_selection`] threads between sites for +/// exact compositions (issue #171): child candidates already committed by +/// an earlier parent, and the maintained summary above each site. +#[derive(Default)] +struct CompositionContext { + /// child target ptr → the child's candidate an ancestor's composition + /// already committed to (a later parent must compose with the *same* + /// one, and the child's own selection is forced to it). + committed_child: HashMap<*const QueryExpr, *const ReplacementSubDAG>, + /// site ptr → the maintained `SummaryAgg` directly above it, when its + /// parent chose a bound `Summary` — what an `ExactTransform` here feeds. + maintaining_parent: HashMap<*const QueryExpr, Rc>, +} + +/// One eligible composed alternative at a site, before the cheapest wins. +struct CompositionOption<'a> { + candidate: &'a ReplacementSubDAG, + decision: CompositionDecision<'a>, +} + +/// Every [`Replacement::ExactComposition`] candidate of `group` whose +/// composed-plan rate is *known* and beats the raw-recompute baseline — +/// costed against each compatible child candidate already in `PlanSpace` +/// (or the one an earlier parent committed). Unknown statistics yield no +/// option at all: the conservative `KeepPreAsap` path stays. +fn composition_options<'a>( + group: &'a MemoGroup, + groups: &'a HashMap<*const QueryExpr, MemoGroup>, + effective: usize, + cost_model: &dyn CostModel, + context: &CompositionContext, +) -> Vec> { + let mut options = Vec::new(); + for candidate in &group.candidates { + let Replacement::ExactComposition(composition) = &candidate.replacement else { + continue; + }; + let child_ptr = Rc::as_ptr(&composition.child_target); + let Some(child_group) = groups.get(&child_ptr) else { + continue; + }; + let already_committed = context.committed_child.get(&child_ptr).copied(); + let cost = |summary: &SummaryNode, shared: bool| { + let request = ExactCompositionCostRequest { + target: &group.target, + composition, + summary, + effective_consumer_count: effective, + }; + let mut inputs = cost_model.exact_composition_cost_inputs(&request); + if shared { + // Shared state is counted once: an earlier parent already + // pays this child's maintenance, so the marginal cost here + // is zero — a *known* zero, unlike an unknown input. + if let Some(maintenance) = inputs.summary_maintenance_cost_per_update.as_mut() { + *maintenance = 0.0; + } + } + let rate = inputs.composed_plan_cost_rate(composition.placement)?; + let baseline = raw_recompute_cost_rate(&inputs)?; + (rate < baseline).then_some((rate, baseline, inputs)) + }; + match composition.placement { + CompositionPlacement::PostProcess => { + let child_candidates: Vec<&'a ReplacementSubDAG> = match already_committed { + // SAFETY-free: the pointer was taken from `groups`'s own + // candidate storage, which outlives this borrow. + Some(ptr) => child_group + .candidates + .iter() + .filter(|c| std::ptr::eq(*c, ptr)) + .collect(), + None => child_group.candidates.iter().collect(), + }; + for child_candidate in child_candidates { + let Replacement::Summary(summary) = &child_candidate.replacement else { + continue; + }; + if !composition.accepts_child(summary) { + continue; + } + let Some((rate, baseline, inputs)) = cost(summary, already_committed.is_some()) + else { + continue; + }; + options.push(CompositionOption { + candidate, + decision: CompositionDecision { + child_target: &composition.child_target, + child_candidate: Some(child_candidate), + cost_rate: rate, + baseline_rate: baseline, + inputs, + }, + }); + } + } + CompositionPlacement::Transform => { + // An update-path transform only pays off beneath a + // maintained summary; with nothing above it, its output is + // never read and the raw fallback is the same computation. + let Some(parent) = context.maintaining_parent.get(&Rc::as_ptr(&group.target)) + else { + continue; + }; + let Some((rate, baseline, inputs)) = cost(parent, false) else { + continue; + }; + options.push(CompositionOption { + candidate, + decision: CompositionDecision { + child_target: &composition.child_target, + child_candidate: None, + cost_rate: rate, + baseline_rate: baseline, + inputs, + }, + }); + } + } + } + options } impl PlanSpace { @@ -2817,6 +3172,7 @@ impl PlanSpace { let mut effective_uses = graph.external_root_uses.clone(); let mut chosen_share: HashMap<*const QueryExpr, ShareDecision> = HashMap::new(); let mut groups: HashMap<*const QueryExpr, SelectedGroup<'_>> = HashMap::new(); + let mut context = CompositionContext::default(); for ptr in &topo { let group = &self.groups[ptr]; @@ -2824,7 +3180,50 @@ impl PlanSpace { let effective = effective_uses.get(ptr).copied().unwrap_or(0); effective_uses.insert(*ptr, effective); - let chosen = if effective >= 2 && cse_candidate_pair(group).is_some() { + // ── Exact compositions (issue #171) ───────────────────────── + // A child an earlier parent's composition committed to is + // forced to exactly that candidate — the parent/child pair is + // one decision. Otherwise, a composition here wins only when + // its cost-units-per-second rate is *known* and beats the raw + // recompute baseline; missing statistics keep the conservative + // path below. + let mut composition_decision = None; + let forced = context + .committed_child + .get(ptr) + .and_then(|&cptr| group.candidates.iter().find(|c| std::ptr::eq(*c, cptr))); + let composed = if forced.is_some() { + None + } else { + composition_options(group, &self.groups, effective, cost_model, &context) + .into_iter() + .min_by(|a, b| a.decision.cost_rate.0.total_cmp(&b.decision.cost_rate.0)) + }; + if let Some(option) = &composed { + if let Some(child_candidate) = option.decision.child_candidate { + context.committed_child.insert( + Rc::as_ptr(option.decision.child_target), + child_candidate as *const ReplacementSubDAG, + ); + } + if let Replacement::ExactComposition(composition) = &option.candidate.replacement { + if composition.placement == CompositionPlacement::Transform { + // A chain of transforms feeds the same summary. + if let Some(parent) = context.maintaining_parent.get(ptr).cloned() { + context + .maintaining_parent + .insert(Rc::as_ptr(&composition.child_target), parent); + } + } + } + } + + let chosen = if let Some(forced) = forced { + Some(forced) + } else if let Some(option) = composed { + composition_decision = Some(option.decision); + Some(option.candidate) + } else if effective >= 2 && cse_candidate_pair(group).is_some() { let decision = if let Some(profiles) = profiles { decide_group_with_recurrence( group, @@ -2844,7 +3243,9 @@ impl PlanSpace { let logical = group .candidates .iter() - .filter(|candidate| !is_cse_candidate(candidate)) + .filter(|candidate| { + !is_cse_candidate(candidate) && !is_composition_candidate(candidate) + }) .min_by(|a, b| { cost_model .estimate_cost(a, &effective_target) @@ -2875,15 +3276,32 @@ impl PlanSpace { // valid answer, just not a cross-group-aware one; this // group also contributes no Share collapse to its own // children (see `multiplier`'s `_ => effective` arm). - None => rank_group(group, cost_model).into_iter().next(), + None => rank_group(group, cost_model) + .into_iter() + .find(|candidate| !is_composition_candidate(candidate)), } } else { rank_group(group, cost_model) .into_iter() - .find(|candidate| !is_cse_candidate(candidate)) + .find(|candidate| { + !is_cse_candidate(candidate) && !is_composition_candidate(candidate) + }) .or_else(|| cse_candidate_pair(group).map(|(share, _)| share)) }; + // Record the maintained summary this site's bound candidate + // builds, for a child that may compose an `ExactTransform` + // beneath it. + if let (Some(Replacement::Summary(node)), QueryExpr::Aggregate { child, .. }) = + (chosen.map(|c| &c.replacement), group.target.as_ref()) + { + if let Some(summary) = maintained_summary(node) { + context + .maintaining_parent + .insert(Rc::as_ptr(child), Rc::clone(summary)); + } + } + let outgoing_multiplier = multiplier(*ptr, &effective_uses, &chosen_share); match chosen { Some(ReplacementSubDAG { @@ -2901,7 +3319,9 @@ impl PlanSpace { _ => { let selected_rewrite = match chosen.map(|candidate| &candidate.replacement) { Some(Replacement::Rewrite(rewrite)) => rewrite, - Some(Replacement::Summary(_)) | None => &group.target, + Some(Replacement::Summary(_) | Replacement::ExactComposition(_)) | None => { + &group.target + } }; for (child, edge_count) in direct_child_counts(selected_rewrite) { *effective_uses.entry(child).or_insert(0) += @@ -2917,6 +3337,7 @@ impl PlanSpace { consumer_count: group.consumer_count, effective_consumer_count: effective, chosen, + composition: composition_decision, }, ); } @@ -2924,6 +3345,7 @@ impl PlanSpace { Ok(GlobalSelection { order: self.order.clone(), groups, + materialized: RefCell::new(HashMap::new()), }) } } @@ -3299,6 +3721,7 @@ pub fn default_strategies() -> Vec> { Box::new(HydraGroupingStrategy::default_cost_model()), Box::new(SharedSubtreeStrategy), Box::new(crate::rewrite::AvgToSumOverCountStrategy), + Box::new(ExactCompositionStrategy::default_cost_model()), ] } @@ -3313,6 +3736,7 @@ pub fn default_strategies_with<'a>( Box::new(HydraGroupingStrategy::new(cost_model)), Box::new(SharedSubtreeStrategy), Box::new(crate::rewrite::AvgToSumOverCountStrategy), + Box::new(ExactCompositionStrategy::new(cost_model)), ] } @@ -3406,6 +3830,7 @@ pub fn search_workload_with_targets<'s, Id>( .as_ref() .is_some_and(|g| accuracy_model.satisfies(g, &target)), Replacement::Rewrite(_) => true, + Replacement::ExactComposition(_) => false, }); group.candidates = legal; group.rejected.extend(illegal.into_iter().map(|candidate| { @@ -3426,6 +3851,11 @@ pub fn search_workload_with_targets<'s, Id>( None, )), Replacement::Rewrite(_) => unreachable!("rewrites are never rejected here"), + Replacement::ExactComposition(_) => ( + asap_types::post_asap::ErrorMetric::AbsoluteValue, + None, + None, + ), }; RejectedCandidate { strategy: candidate.strategy, @@ -4379,7 +4809,9 @@ mod tests { .iter() .map(|r| match &r.replacement { Replacement::Summary(node) => summary_family_algorithm(node), - Replacement::Rewrite(_) => panic!("expected a Summary replacement"), + Replacement::Rewrite(_) | Replacement::ExactComposition(_) => { + panic!("expected a Summary replacement") + } }) .collect(); assert!(kinds.contains(&SketchAlgorithm::Kll), "{kinds:?}"); @@ -4399,7 +4831,9 @@ mod tests { .iter() .map(|r| match &r.replacement { Replacement::Summary(node) => summary_family_algorithm(node), - Replacement::Rewrite(_) => panic!("expected a Summary replacement"), + Replacement::Rewrite(_) | Replacement::ExactComposition(_) => { + panic!("expected a Summary replacement") + } }) .collect(); assert_eq!( @@ -4427,7 +4861,9 @@ mod tests { .iter() .map(|r| match &r.replacement { Replacement::Summary(node) => summary_family_algorithm(node), - Replacement::Rewrite(_) => panic!("expected a Summary replacement"), + Replacement::Rewrite(_) | Replacement::ExactComposition(_) => { + panic!("expected a Summary replacement") + } }) .collect(); assert_eq!(kinds, vec![SketchAlgorithm::Theta, SketchAlgorithm::Kmv]); @@ -4505,7 +4941,9 @@ mod tests { .iter() .map(|r| match &r.replacement { Replacement::Summary(node) => summary_family_algorithm(node), - Replacement::Rewrite(_) => panic!("expected a Summary replacement"), + Replacement::Rewrite(_) | Replacement::ExactComposition(_) => { + panic!("expected a Summary replacement") + } }) .collect(); assert!(kinds.contains(&SketchAlgorithm::Kll)); @@ -4513,13 +4951,13 @@ mod tests { assert_eq!(kinds.len(), 2); } - /// Enumerating candidates for the *target* node must only steer that - /// node's own decision — a nested aggregate underneath it still gets its - /// own independent (`cost_model`-ranked) enumeration, not whatever the - /// caller happened to pick for the outer target. This is the behavior - /// [`construct_summary`]'s recursion (via [`realize_child`]) - /// gets for free: only the top node's `Implementation` is ever forced - /// from outside; the child is always re-enumerated fresh. + /// Constructing the outer target's candidates never leaks the outer + /// choice into the nested aggregate — and, since issue #171's phase + /// contract, a maintained outer sketch can no longer sit above the + /// inner sketch's *readout* at all: the outer target degrades to the + /// conservative `KeepPreAsap` fallback (reported once, not once per + /// dropped family), while the inner quantile keeps its own, + /// independently cost-ranked candidates in its own `MemoGroup`. #[test] fn enumerating_the_targets_candidates_does_not_leak_into_a_nested_aggregate() { // outer: quantile(0.99, ...) over inner: quantile(0.5, m) — both @@ -4539,35 +4977,41 @@ mod tests { ) .replacements(&target); - let ddsketch = replacements - .iter() - .find(|r| { - matches!(&r.replacement, Replacement::Summary(node) - if summary_family_algorithm(node) == SketchAlgorithm::DDSketch) - }) - .expect("the outer target's DDSketch candidate must be present"); - let Replacement::Summary(node) = &ddsketch.replacement else { - unreachable!("filtered on Replacement::Summary above"); + assert_eq!(replacements.len(), 1, "{replacements:?}"); + let Replacement::Summary(node) = &replacements[0].replacement else { + unreachable!("SketchAlgorithmStrategy only returns Summary candidates"); }; - assert_eq!( - summary_family_algorithm(node), - SketchAlgorithm::DDSketch, - "the outer (target) node must be the DDSketch candidate" + assert!( + matches!(node.expr, SummaryExpr::KeepPreAsap(ref e) if Rc::ptr_eq(e, &outer)), + "a sketch over a sketch readout is data_state-illegal; expected the conservative \ + fallback, got {:?}", + node.expr ); - - let asap_types::post_asap::SummaryExpr::SummaryEstimate { summary_input, .. } = &node.expr - else { - panic!("expected SummaryEstimate root, got {:?}", node.expr); - }; - let asap_types::post_asap::SummaryExpr::SummaryAgg { child, .. } = &summary_input.expr - else { - panic!("expected SummaryAgg, got {:?}", summary_input.expr); + assert!(replacements[0].rationale.contains("readout")); + + // The inner target is still independently enumerated and ranked — + // a custom cost model that prefers DDSketch for it is honored, and + // nothing about the outer target's choice reaches it. + let space = search_workload_with( + vec![("q", Rc::clone(&outer))], + &default_strategies_with(&PreferDDSketchViaCostModel), + ); + let QueryExpr::Aggregate { child, .. } = space.roots[0].1.as_ref() else { + unreachable!() }; + let inner_group = space.group_for(child).expect("inner quantile is a target"); + let inner_kinds: Vec = inner_group + .candidates + .iter() + .filter_map(|c| match &c.replacement { + Replacement::Summary(node) => sketch_kind_of(node), + _ => None, + }) + .collect(); assert_eq!( - summary_family_algorithm(child), - SketchAlgorithm::Kll, - "the nested inner aggregate must still get the cost-model-ranked \ - default (Kll), not inherit the outer target's DDSketch candidate" + inner_kinds, + vec![SketchAlgorithm::DDSketch, SketchAlgorithm::Kll], + "the nested inner aggregate keeps its own cost-model-ranked candidates" ); } @@ -5094,7 +5538,7 @@ mod tests { assert_eq!(rewrites.len(), 2); let first_shares_target = match &rewrites[0].replacement { Replacement::Rewrite(rc) => Rc::ptr_eq(rc, &group.target), - Replacement::Summary(_) => false, + Replacement::Summary(_) | Replacement::ExactComposition(_) => false, }; assert!( first_shares_target, @@ -5130,7 +5574,7 @@ mod tests { assert_eq!(agg_group.candidates.len(), 2); let first_kind = match &agg_group.candidates[0].replacement { Replacement::Summary(node) => sketch_kind_of(node), - Replacement::Rewrite(_) => None, + Replacement::Rewrite(_) | Replacement::ExactComposition(_) => None, }; assert_eq!(first_kind, Some(SketchAlgorithm::DDSketch)); } @@ -5326,7 +5770,7 @@ mod tests { .unwrap(); let kind = match &agg_group.chosen.unwrap().replacement { Replacement::Summary(node) => sketch_kind_of(node), - Replacement::Rewrite(_) => None, + Replacement::Rewrite(_) | Replacement::ExactComposition(_) => None, }; assert_eq!(kind, Some(SketchAlgorithm::DDSketch)); } @@ -6741,6 +7185,7 @@ mod tests { DefaultAccuracyModel.satisfies(g, &AccuracyTarget::Epsilon(0.1)) }), Replacement::Rewrite(_) => false, + Replacement::ExactComposition(_) => false, })); let ranked = space.cost_sorted(&DefaultCostModel); let root_ranked = ranked.iter().find(|g| Rc::ptr_eq(g.target, root)).unwrap(); @@ -6808,6 +7253,7 @@ mod tests { .as_ref() .is_some_and(ResultGuarantee::is_exact), Replacement::Rewrite(_) => true, + Replacement::ExactComposition(_) => false, })); } diff --git a/crates/asap-aware-mapping/src/rollup.rs b/crates/asap-aware-mapping/src/rollup.rs index 5dc3c81e..a5abae8a 100644 --- a/crates/asap-aware-mapping/src/rollup.rs +++ b/crates/asap-aware-mapping/src/rollup.rs @@ -690,7 +690,7 @@ mod tests { .iter() .find_map(|candidate| match &candidate.replacement { Replacement::Rewrite(rewrite) => Some(rewrite), - Replacement::Summary(_) => None, + Replacement::Summary(_) | Replacement::ExactComposition(_) => None, }) .expect("default search must include the roll-up rewrite"); let QueryExpr::Aggregate { child, .. } = rewrite.as_ref() else { diff --git a/crates/devtools/src/bin/dag_export.rs b/crates/devtools/src/bin/dag_export.rs index 071c4a40..589e5d93 100644 --- a/crates/devtools/src/bin/dag_export.rs +++ b/crates/devtools/src/bin/dag_export.rs @@ -251,6 +251,56 @@ struct Winner<'a> { target: &'a Rc, candidate: &'a ReplacementSubDAG, cost: f64, + /// The unit `cost` is in — `None` for the legacy unitless structural + /// estimate, `Some("cost_units_per_second")` for a composed decision + /// `global_selection` costed as a recurring rate (issue #171). + cost_unit: Option<&'static str>, + /// For a `Replacement::ExactComposition` winner: the composed node + /// `GlobalSelection::materialize` linked over the child's own committed + /// decision — the shape actually exported. `None` otherwise. + materialized: Option>, + /// For a composed winner: the child target it was committed with. + child_target: Option<&'a Rc>, +} + +impl Winner<'_> { + /// The post-ASAP node to export for this winner, if it is a bound + /// (`Summary` or materialized composition) shape. + fn summary_node(&self) -> Option> { + match &self.candidate.replacement { + Replacement::Summary(node) => Some(Rc::clone(node)), + Replacement::ExactComposition(_) => self.materialized.clone(), + Replacement::Rewrite(_) => None, + } + } +} + +/// The `DagDecision` for winner `i`, with explicit provenance/unit and — +/// for a composed winner — the child decision it was committed with, so a +/// viewer never infers composition from graph shape (issue #171). +fn decision_for(i: usize, winners: &[Winner<'_>], role: &'static str) -> DagDecision { + let winner = &winners[i]; + let child_decisions = winner + .child_target + .into_iter() + .filter_map(|child| { + winners + .iter() + .position(|w| Rc::ptr_eq(w.target, child)) + .map(|j| j as u32) + }) + .collect(); + DagDecision { + id: i as u32, + strategy: winner.candidate.strategy.to_string(), + rationale: decision_rationale(winner), + rank: 0, + cost: winner.cost, + role, + provenance: Some(format!("{:?}", winner.candidate.provenance)), + cost_unit: winner.cost_unit.map(str::to_string), + child_decisions, + } } /// Short explanation intended for a selected winner in node-level UI. The @@ -280,6 +330,18 @@ fn decision_rationale(winner: &Winner<'_>) -> String { "Derives this smaller top-k from a compatible larger top-k result shared by the workload." .to_string() } + "ExactCompositionStrategy" => match winner.candidate.provenance { + asap_aware_mapping::replacement::ReplacementProvenance::ExactPostProcess => { + "Applies the exact fold at query time over the child's committed summary readout." + .to_string() + } + asap_aware_mapping::replacement::ReplacementProvenance::ExactTransform => { + "Runs the exact row transform on the update path, feeding the maintained summary above it." + .to_string() + } + _ => "Composes an exact operator with a summary plan across an explicit domain boundary." + .to_string(), + }, _ => { let summary = winner .candidate @@ -325,13 +387,14 @@ fn target_replacement( ) -> TargetReplacement { let strategy = winner.candidate.strategy.to_string(); let before = dag_export::export(winner.target); - let after = match &winner.candidate.replacement { - Replacement::Summary(node) => { - TargetReplacementAfter::Summary(dag_export::export_summary(node)) - } - Replacement::Rewrite(rewritten) => { + let after = match (&winner.candidate.replacement, winner.summary_node()) { + (Replacement::Rewrite(rewritten), _) => { TargetReplacementAfter::Rewrite(dag_export::export(rewritten)) } + (_, Some(node)) => TargetReplacementAfter::Summary(dag_export::export_summary(&node)), + (_, None) => { + unreachable!("a composed winner always carries its materialized node") + } }; TargetReplacement { decision_id, @@ -428,6 +491,10 @@ fn run_post_asap_with_progress( .collect(); let space = search_workload(roots); let ranked_groups = space.cost_sorted(&DefaultCostModel); + // Composed decisions (issue #171) are only ever committed by + // `global_selection`, together with the child decision they compose + // over — never by per-group `cost_sorted` ranking. + let selection = space.global_selection(&DefaultCostModel); // A group's top candidate can be `keep_pre_asap`'s own conservative // fallback — `Replacement::Summary(SummaryNode { expr: @@ -454,7 +521,26 @@ fn run_post_asap_with_progress( let winners: Vec> = ranked_groups .iter() .filter_map(|group| { - let candidate = group.candidates.first()?; + if let Some(selected) = selection.for_target(group.target) { + if let (Some(chosen), Some(decision)) = (selected.chosen, &selected.composition) { + let materialized = selection.materialize(group.target).ok().flatten()?; + return Some(Winner { + target: group.target, + candidate: chosen, + cost: decision.cost_rate.0, + cost_unit: Some(decision.inputs.unit.as_str()), + materialized: Some(materialized), + child_target: Some(decision.child_target), + }); + } + } + // A composition not committed by `global_selection` is never a + // winner on its own — it has no child to compose over. + let (i, candidate) = group + .candidates + .iter() + .enumerate() + .find(|(_, c)| !matches!(c.replacement, Replacement::ExactComposition(_)))?; if matches!( &candidate.replacement, Replacement::Summary(node) if matches!(node.expr, SummaryExpr::KeepPreAsap(_)) @@ -464,7 +550,10 @@ fn run_post_asap_with_progress( Some(Winner { target: group.target, candidate, - cost: group.costs[0], + cost: group.costs[i], + cost_unit: None, + materialized: None, + child_target: None, }) }) .collect(); @@ -507,24 +596,20 @@ fn run_post_asap_with_progress( let mut find_winner = |expr: &QueryExpr| -> Option { let i = lookup_winner(&by_hash, &winners, &mut post_graph_cache, expr)?; let winner = &winners[i]; - let decision = DagDecision { - id: i as u32, - strategy: winner.candidate.strategy.to_string(), - rationale: decision_rationale(winner), - rank: 0, - cost: winner.cost, - role: "replacement_region", - }; - Some(match &winners[i].candidate.replacement { - Replacement::Rewrite(rc) => PostAsapSubstitution::Rewrite { - replacement: Rc::clone(rc), - decision, + let decision = decision_for(i, &winners, "replacement_region"); + Some( + match (&winner.candidate.replacement, winner.summary_node()) { + (Replacement::Rewrite(rc), _) => PostAsapSubstitution::Rewrite { + replacement: Rc::clone(rc), + decision, + }, + (_, Some(node)) => PostAsapSubstitution::Summary { + replacement: node, + decision, + }, + (_, None) => unreachable!("a composed winner always carries its materialized node"), }, - Replacement::Summary(rc) => PostAsapSubstitution::Summary { - replacement: Rc::clone(rc), - decision, - }, - }) + ) }; let post_graphs: Vec<(String, DagGraph)> = lowered_queries .iter() diff --git a/crates/types/src/dag_export.rs b/crates/types/src/dag_export.rs index 5df2ed8c..3672c658 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 data_state 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 @@ -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_data_state, produced_data_state, AccuracyError, ExactOperator, + ExecutionDataState, 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_domain(node)); SummaryDagGraph { nodes, root } } +/// The explicit execution data_state of an exported plan's root — its own +/// produced data_state, or query-time readout for a bare `KeepPreAsap` +/// (the same convention `post_asap::execution_data_state::validate_execution_data_states` +/// uses for a root). +fn root_domain(node: &SummaryNode) -> ExecutionDataState { + produced_data_state(&node.expr).unwrap_or(ExecutionDataState::READ_ROWS) +} + +/// `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 { +/// +/// `data_state` is the node's explicit execution data_state (issue #171) — its own +/// [`produced_data_state`], or the edge-assigned data_state for a `KeepPreAsap` +/// — and is written into `detail.execution_data_state` on every post-ASAP node so a viewer +/// reads it rather than inferring it from the node's kind. +fn summary_shape( + expr: &SummaryExpr, + data_state: ExecutionDataState, +) -> (&'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,25 @@ 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( + "execution_data_state".into(), + serde_json::json!({ + "timing": data_state.timing.as_str(), + "primitive": data_state.primitive.as_str(), + }), + ); } + (kind, label, detail) } /// `expr`'s own `Rc` children, in the variant's field order @@ -455,6 +555,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 +566,22 @@ 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, + data_state: 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; 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, + "execution_data_state": { + "timing": data_state.timing.as_str(), + "primitive": data_state.primitive.as_str(), + }, + }); return push_summary_node( nodes, "KeepPreAsap", @@ -479,9 +593,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_data_state(&node.expr, child))) .collect(); - let (kind, label, detail) = summary_shape(&node.expr); + let (kind, label, detail) = summary_shape(&node.expr, data_state); push_summary_node(nodes, kind, label, detail, children, node.guarantee.clone()) } @@ -759,15 +873,24 @@ fn build_summary_hybrid( nodes: &mut Vec, cache: &mut HashCache, find_winner: &mut dyn FnMut(&QueryExpr) -> Option, + data_state: ExecutionDataState, ) -> 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_data_state(&node.expr, child), + ) + }) .collect(); - let (kind, label, mut detail) = summary_shape(&node.expr); + let (kind, label, mut detail) = summary_shape(&node.expr, data_state); // 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 +976,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_domain(&replacement), + ); for node in &mut nodes[first..] { if node.decision.is_none() { let mut node_decision = decision.clone(); @@ -1413,7 +1542,7 @@ mod tests { ); } - // ── Issue #223 stage 3: dag_export's hash literally *is* cse's hash ──── + // ── Issue #223 data_state 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/execution_data_state.rs b/crates/types/src/post_asap/execution_data_state.rs new file mode 100644 index 00000000..417a329f --- /dev/null +++ b/crates/types/src/post_asap/execution_data_state.rs @@ -0,0 +1,907 @@ +//! 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) +//! 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 data_state-ambiguous value operation. +//! +//! [`ExecutionDataState`] is what a node's output *is*, at which data_state; +//! [`validate_execution_data_states`] checks every edge of a DAG against the +//! rules below at plan construction, returning a typed [`ExecutionDataStateError`] rather +//! than deferring to a runtime failure. +//! +//! ## Edge rules +//! +//! | Parent | Accepts from `child` | +//! |---|---| +//! | `SummaryAgg.child` | `MAINTENANCE_ROWS`, or `MAINTENANCE_SUMMARY` of an **exact accumulator** family. Never a read-time data_state. | +//! | `SummaryEstimate.summary_input` | `MAINTENANCE_SUMMARY` (any family). Produces `READ_ROWS`. | +//! | `SummaryJoin.outer/inner` | `MAINTENANCE_ROWS` or `MAINTENANCE_SUMMARY`; never a read-time data_state. | +//! | `SummarySubtract`/`SummaryDelete`/`SummaryMerge` | `MAINTENANCE_SUMMARY`. | +//! | `UpdateTransform.child` | `MAINTENANCE_ROWS`. Produces `MAINTENANCE_ROWS`. | +//! | `ReadoutPostProcess.child` | `READ_ROWS`. Produces `READ_ROWS`. | +//! +//! ## `KeepPreAsap` declares its data_state through the derivation +//! +//! A [`SummaryExpr::KeepPreAsap`] leaf is a raw pre-ASAP computation that a +//! runtime can execute at either data_state: 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 data_state field of its own +//! — every existing consumer pattern-matches the one-field shape — so its +//! data_state is *assigned* by [`validate_execution_data_states`] from the edge that +//! reaches it and reported in the returned [`ExecutionDataStateAssignment`]. 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 +//! [`ExecutionDataStateError::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}; + +/// When a post-ASAP value is produced. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +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, +} + +impl DataPrimitive { + pub fn as_str(self) -> &'static str { + match self { + Self::Rows => "rows", + Self::SummaryState => "summary_state", + } + } +} + +/// The two-dimensional edge contract: when a value exists and which data +/// primitive it carries. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub struct ExecutionDataState { + pub timing: ExecutionTiming, + pub primitive: DataPrimitive, +} + +impl ExecutionDataState { + 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 ExecutionDataState { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{}/{}", self.timing.as_str(), self.primitive.as_str()) + } +} + +/// Which parent/edge a [`ExecutionDataStateError`] 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 ExecutionDataStateEdge { + SummaryAggChild, + SummaryEstimateInput, + SummaryJoinInput, + SummarySubtractInput, + SummaryDeleteInput, + SummaryMergeInput, + UpdateTransformChild, + ReadoutPostProcessChild, +} + +impl ExecutionDataStateEdge { + 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 data_state 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 ExecutionDataStateError { + /// A query-time value (`SummaryEstimate` / `ReadoutPostProcess` output) + /// placed beneath a maintained summary — the one shape issue #171's + /// data_state 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: ExecutionDataState, + }, + /// Any other edge whose child data_state 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: ExecutionDataState, + }, + /// 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 data_state-ambiguous: reached as {first} and as {second} in the same \ + plan" + )] + AmbiguousKeepPreAsap { + 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. + #[error("UpdateTransform cannot be a plan root: its update-path output feeds nothing")] + MaintenanceRowsAtRoot, + /// An `ExactOperator` whose input columns are not all `Plain` at its + /// declared data_state. + #[error("exact operator consumes non-plain column {column:?} ({dtype})")] + NonPlainOperand { column: String, dtype: String }, +} + +/// The data_state assigned to every node of a validated plan, keyed by +/// `Rc` pointer identity — the explicit per-node "execution_data_state" a +/// runtime or a DAG export reads instead of re-deriving it. For every +/// non-`KeepPreAsap` node this equals [`produced_data_state`]; for a +/// `KeepPreAsap` leaf it is the data_state the reaching edge assigned. +#[derive(Debug, Clone, Default)] +pub struct ExecutionDataStateAssignment { + domains: HashMap<*const SummaryNode, ExecutionDataState>, +} + +impl ExecutionDataStateAssignment { + /// The data_state assigned to `node`, if it was part of the validated plan. + pub fn data_state_of(&self, node: &Rc) -> Option { + self.domains.get(&Rc::as_ptr(node)).copied() + } + + /// The data_state assigned to the node at `ptr` — for callers walking a plan + /// by reference rather than by `Rc`. + pub fn data_state_of_ptr(&self, ptr: *const SummaryNode) -> Option { + self.domains.get(&ptr).copied() + } +} + +/// The data_state `expr` *produces*, independent of context — `None` for +/// [`SummaryExpr::KeepPreAsap`], whose data_state is assigned by the edge reaching +/// it (see the module docs). +pub fn produced_data_state(expr: &SummaryExpr) -> Option { + Some(match expr { + SummaryExpr::KeepPreAsap(_) => return None, + SummaryExpr::SummaryAgg { .. } + | SummaryExpr::SummaryJoin { .. } + | SummaryExpr::SummarySubtract { .. } + | SummaryExpr::SummaryDelete { .. } + | SummaryExpr::SummaryMerge { .. } => ExecutionDataState::MAINTENANCE_SUMMARY, + SummaryExpr::SummaryEstimate { .. } | SummaryExpr::ReadoutPostProcess { .. } => { + ExecutionDataState::READ_ROWS + } + SummaryExpr::UpdateTransform { .. } => ExecutionDataState::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<(), ExecutionDataStateError> { + for field in &schema.fields { + match &field.dtype { + SummaryFamilyType::Plain(_) | SummaryFamilyType::ExactAggregate(..) => {} + other => { + return Err(ExecutionDataStateError::UnsupportedStateComposition { + family: format!("{other:?}"), + }) + } + } + } + Ok(()) +} + +/// Validate every edge of the DAG rooted at `root` against the module-level +/// rules, returning each node's assigned data_state on success. Shared +/// `Rc`s are visited once per reaching edge (the assignment is +/// per node, so a conflict between two edges is what +/// [`ExecutionDataStateError::AmbiguousKeepPreAsap`] detects). +pub fn validate_execution_data_states( + 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_domain = match produced_data_state(&root.expr) { + None => ExecutionDataState::READ_ROWS, + Some(ExecutionDataState::MAINTENANCE_ROWS) => { + return Err(ExecutionDataStateError::MaintenanceRowsAtRoot) + } + Some(data_state) => data_state, + }; + validate_execution_data_states_at(root, root_domain) +} + +/// [`validate_execution_data_states`] for a *sub*-plan whose root is known to +/// sit at `data_state` — 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_data_states_at( + root: &Rc, + data_state: ExecutionDataState, +) -> Result { + let mut assignment = ExecutionDataStateAssignment::default(); + visit(root, data_state, &mut assignment)?; + Ok(assignment) +} + +/// Record `data_state` for `node` (detecting a conflicting earlier assignment +/// for a `KeepPreAsap`), then check and recurse into every child edge. +fn visit( + node: &Rc, + data_state: ExecutionDataState, + assignment: &mut ExecutionDataStateAssignment, +) -> Result<(), ExecutionDataStateError> { + let ptr = Rc::as_ptr(node); + if let Some(previous) = assignment.domains.get(&ptr) { + if *previous != data_state { + return Err(ExecutionDataStateError::AmbiguousKeepPreAsap { + first: *previous, + second: data_state, + }); + } + // Already validated through another edge with the same data_state. + return Ok(()); + } + assignment.domains.insert(ptr, data_state); + + match &node.expr { + SummaryExpr::KeepPreAsap(_) => Ok(()), + SummaryExpr::SummaryAgg { child, .. } => { + let child_domain = child_domain( + child, + ExecutionDataStateEdge::SummaryAggChild, + |avail| match avail { + ExecutionDataState::MAINTENANCE_ROWS => Ok(()), + ExecutionDataState::MAINTENANCE_SUMMARY => { + is_exact_accumulator_state(&child.schema) + } + other => Err(ExecutionDataStateError::ReadoutUnderMaintenance { + edge: ExecutionDataStateEdge::SummaryAggChild.describe(), + child: other, + }), + }, + )?; + visit(child, child_domain, assignment) + } + SummaryExpr::SummaryJoin { outer, inner, .. } => { + for input in [outer, inner] { + let s = child_domain(input, ExecutionDataStateEdge::SummaryJoinInput, |avail| { + match avail { + ExecutionDataState::MAINTENANCE_ROWS + | ExecutionDataState::MAINTENANCE_SUMMARY => Ok(()), + other => Err(ExecutionDataStateError::ReadoutUnderMaintenance { + edge: ExecutionDataStateEdge::SummaryJoinInput.describe(), + child: other, + }), + } + })?; + visit(input, s, assignment)?; + } + Ok(()) + } + SummaryExpr::SummarySubtract { left, right } => { + for input in [left, right] { + let s = state_only(input, ExecutionDataStateEdge::SummarySubtractInput)?; + visit(input, s, assignment)?; + } + Ok(()) + } + SummaryExpr::SummaryDelete { summary_input, .. } => { + let s = state_only(summary_input, ExecutionDataStateEdge::SummaryDeleteInput)?; + visit(summary_input, s, assignment) + } + SummaryExpr::SummaryMerge { children } => { + for input in children { + let s = state_only(input, ExecutionDataStateEdge::SummaryMergeInput)?; + visit(input, s, assignment)?; + } + Ok(()) + } + SummaryExpr::SummaryEstimate { summary_input, .. } => { + let s = state_only(summary_input, ExecutionDataStateEdge::SummaryEstimateInput)?; + visit(summary_input, s, assignment) + } + SummaryExpr::UpdateTransform { child, op } => { + let s = child_domain( + child, + ExecutionDataStateEdge::UpdateTransformChild, + |avail| match avail { + ExecutionDataState::MAINTENANCE_ROWS => Ok(()), + other => Err(ExecutionDataStateError::IllegalChildPhase { + edge: ExecutionDataStateEdge::UpdateTransformChild.describe(), + child: other, + }), + }, + )?; + check_plain_operands(op, &child.schema)?; + visit(child, s, assignment) + } + SummaryExpr::ReadoutPostProcess { child, op } => { + let s = child_domain( + child, + ExecutionDataStateEdge::ReadoutPostProcessChild, + |avail| match avail { + ExecutionDataState::READ_ROWS => Ok(()), + other => Err(ExecutionDataStateError::IllegalChildPhase { + edge: ExecutionDataStateEdge::ReadoutPostProcessChild.describe(), + child: other, + }), + }, + )?; + check_plain_operands(op, &child.schema)?; + visit(child, s, assignment) + } + } +} + +/// The data_state `child` takes as a direct input of `parent`, without +/// validating legality — `child`'s own produced data_state, or for a +/// `KeepPreAsap` leaf the data_state `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 — maintenance rows +/// under a state-only edge). For DAG export and other reporting that needs +/// an explicit per-node data_state even on a plan that +/// [`validate_execution_data_states`] would reject. +pub fn assigned_child_data_state(parent: &SummaryExpr, child: &SummaryNode) -> ExecutionDataState { + if let Some(avail) = produced_data_state(&child.expr) { + return avail; + } + match parent { + SummaryExpr::ReadoutPostProcess { .. } => ExecutionDataState::READ_ROWS, + SummaryExpr::KeepPreAsap(_) + | SummaryExpr::SummaryAgg { .. } + | SummaryExpr::SummaryJoin { .. } + | SummaryExpr::SummarySubtract { .. } + | SummaryExpr::SummaryDelete { .. } + | SummaryExpr::SummaryEstimate { .. } + | SummaryExpr::SummaryMerge { .. } + | SummaryExpr::UpdateTransform { .. } => ExecutionDataState::MAINTENANCE_ROWS, + } +} + +/// The data_state `child` takes on `edge`: its own produced data_state +/// (checked via `accept`), or — for a `KeepPreAsap` leaf — the data_state the +/// edge assigns it, derived from what that edge accepts. +fn child_domain( + child: &Rc, + edge: ExecutionDataStateEdge, + accept: impl Fn(ExecutionDataState) -> Result<(), ExecutionDataStateError>, +) -> Result { + match produced_data_state(&child.expr) { + Some(avail) => { + accept(avail)?; + Ok(avail) + } + None => { + // A raw pre-ASAP subtree executes at whichever data_state 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 { + ExecutionDataStateEdge::SummaryAggChild + | ExecutionDataStateEdge::SummaryJoinInput + | ExecutionDataStateEdge::UpdateTransformChild => { + ExecutionDataState::MAINTENANCE_ROWS + } + ExecutionDataStateEdge::ReadoutPostProcessChild => ExecutionDataState::READ_ROWS, + ExecutionDataStateEdge::SummaryEstimateInput + | ExecutionDataStateEdge::SummarySubtractInput + | ExecutionDataStateEdge::SummaryDeleteInput + | ExecutionDataStateEdge::SummaryMergeInput => { + return Err(ExecutionDataStateError::IllegalChildPhase { + edge: edge.describe(), + child: ExecutionDataState::MAINTENANCE_ROWS, + }) + } + }; + accept(assigned)?; + Ok(assigned) + } + } +} + +fn state_only( + child: &Rc, + edge: ExecutionDataStateEdge, +) -> Result { + child_domain(child, edge, |avail| match avail { + ExecutionDataState::MAINTENANCE_SUMMARY => Ok(()), + other => Err(ExecutionDataStateError::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<(), ExecutionDataStateError> { + 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(ExecutionDataStateError::NonPlainOperand { + column: field.name.clone(), + dtype: format!("{:?}", field.dtype), + }); + } + } + Ok(()) +} + +fn check_all_plain(input: &SummarySchema) -> Result<(), ExecutionDataStateError> { + for field in &input.fields { + if !matches!(field.dtype, SummaryFamilyType::Plain(_)) { + return Err(ExecutionDataStateError::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_data_states(&root).unwrap(); + assert_eq!( + assignment.data_state_of(&leaf), + Some(ExecutionDataState::MAINTENANCE_ROWS) + ); + assert_eq!( + assignment.data_state_of(&root), + Some(ExecutionDataState::MAINTENANCE_SUMMARY) + ); + } + + #[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_data_states(&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_data_states(&root), + Err(ExecutionDataStateError::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_data_states(&root).unwrap(); + assert_eq!( + assignment.data_state_of(&root), + Some(ExecutionDataState::READ_ROWS) + ); + } + + #[test] + fn non_exact_operator_uses_the_same_read_domain_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_data_states(&root).unwrap(); + assert_eq!( + assignment.data_state_of(&root), + Some(ExecutionDataState::READ_ROWS) + ); + } + + #[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_data_states(&root).err(), + Some(ExecutionDataStateError::ReadoutUnderMaintenance { + edge: "SummaryAgg.child", + child: ExecutionDataState::READ_ROWS, + }) + ); + } + + #[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_data_states(&transform).err(), + Some(ExecutionDataStateError::MaintenanceRowsAtRoot) + ); + let root = estimate(agg(Rc::clone(&transform), kll())); + let assignment = validate_execution_data_states(&root).unwrap(); + assert_eq!( + assignment.data_state_of(&transform), + Some(ExecutionDataState::MAINTENANCE_ROWS) + ); + } + + #[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_data_states(&root), + Err(ExecutionDataStateError::IllegalChildPhase { + edge: "UpdateTransform.child", + child: ExecutionDataState::READ_ROWS + }) + )); + } + + #[test] + 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. + 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 = ExecutionDataStateAssignment::default(); + visit( + &shared, + ExecutionDataState::MAINTENANCE_ROWS, + &mut assignment, + ) + .unwrap(); + assert_eq!( + visit(&shared, ExecutionDataState::READ_ROWS, &mut assignment), + Err(ExecutionDataStateError::AmbiguousKeepPreAsap { + first: ExecutionDataState::MAINTENANCE_ROWS, + second: ExecutionDataState::READ_ROWS, + }) + ); + assert!(validate_execution_data_states(&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) + )); + } +} diff --git a/crates/types/src/post_asap/expr.rs b/crates/types/src/post_asap/expr.rs index b93aa904..5441e832 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 data_state. Exact composition is one producer of the generic +/// [`ValueOperator`] data_state 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 data_state. +/// +/// 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::execution_data_state::ExecutionDataState`] 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::execution_data_state::ExecutionDataStateError`], 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..63a13b29 100644 --- a/crates/types/src/post_asap/mod.rs +++ b/crates/types/src/post_asap/mod.rs @@ -27,13 +27,20 @@ //! 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 use expr::{SummaryExpr, SummaryNode}; +pub use execution_data_state::{ + assigned_child_data_state, exact_operator_output_schema, produced_data_state, + validate_execution_data_states, validate_execution_data_states_at, DataPrimitive, + ExactOperatorSchemaError, ExecutionDataState, ExecutionDataStateAssignment, + ExecutionDataStateError, ExecutionTiming, +}; +pub use expr::{ExactOperator, SummaryExpr, SummaryNode, ValueOperator}; pub use guarantee::{ AccuracyError, BoundExpr, CompositionOperator, ErrorMetric, GuaranteeSource, ProbabilityExpr, ResultGuarantee,