diff --git a/crates/asap-aware-mapping/src/accuracy/estimators/mod.rs b/crates/asap-aware-mapping/src/accuracy/estimators/mod.rs index 056f097f..df9ef4d9 100644 --- a/crates/asap-aware-mapping/src/accuracy/estimators/mod.rs +++ b/crates/asap-aware-mapping/src/accuracy/estimators/mod.rs @@ -99,7 +99,7 @@ pub(crate) fn size_params( SketchParams::CmsWithHeap { width: cms::cms_width(eps), depth: cms::cms_depth(delta), - heap_size: k as u32, + heap_size: topk_capacity(k, eps), } } // Non-preferred candidates (DDSketch / Theta / Kmv / CountSketch / @@ -124,11 +124,19 @@ pub(crate) fn size_params( SketchParams::CountSketchWithHeap { width: count_sketch::count_sketch_width(eps), depth: count_sketch::count_sketch_depth(delta), - heap_size: k as u32, + heap_size: topk_capacity(k, eps), } } } } +/// Accuracy-dependent candidate budget, not a completeness theorem. The +/// membership model must still certify the selected set independently. +pub(crate) fn topk_capacity(k: usize, eps: f64) -> u32 { + u32::try_from(k) + .unwrap_or(u32::MAX) + .max(saturating_ceil(1.0 / eps, 1, 1 << 26)) +} + /// `⌈x⌉` clamped to `[lo, hi]`; NaN / non-positive x saturate to `hi` /// (a degenerate ε means "as accurate as this family goes"). pub(crate) fn saturating_ceil(x: f64, lo: u32, hi: u32) -> u32 { diff --git a/crates/asap-aware-mapping/src/accuracy/evidence.rs b/crates/asap-aware-mapping/src/accuracy/evidence.rs index a915bf25..e73a1fc8 100644 --- a/crates/asap-aware-mapping/src/accuracy/evidence.rs +++ b/crates/asap-aware-mapping/src/accuracy/evidence.rs @@ -96,6 +96,16 @@ pub trait AccuracyEvidenceProvider { None } + /// Enforced upper bound on distinct (partition, item) identities across a + /// complete TopK readout. Used to union-bound score errors for adaptively + /// selected candidates. Observed cardinality is not sufficient evidence. + fn topk_max_distinct_items( + &self, + _expression: &asap_types::pre_asap::QueryExpr, + ) -> Option { + None + } + /// Proof scoped to this complete quantile expression, including its source, /// filters, grouping and window. `None` means unknown, including emptiness. fn quantile_input_domain( diff --git a/crates/asap-aware-mapping/src/accuracy/mod.rs b/crates/asap-aware-mapping/src/accuracy/mod.rs index 0c90ed91..9eef3315 100644 --- a/crates/asap-aware-mapping/src/accuracy/mod.rs +++ b/crates/asap-aware-mapping/src/accuracy/mod.rs @@ -123,6 +123,8 @@ impl AccuracyModel for DefaultAccuracyModel { } } +pub(crate) use estimators::topk_capacity; + #[cfg(test)] mod tests { use super::*; diff --git a/crates/asap-aware-mapping/src/cost_model.rs b/crates/asap-aware-mapping/src/cost_model.rs index 957b2d6a..366f0289 100644 --- a/crates/asap-aware-mapping/src/cost_model.rs +++ b/crates/asap-aware-mapping/src/cost_model.rs @@ -88,36 +88,36 @@ pub struct CostProvenance { /// Which mixed-execution shapes the downstream runtime can actually /// execute (issue #171). [`crate::exact_composition::ExactCompositionStrategy`] -/// proposes an `ValueOperationAtReadTime` candidate only when -/// `read_time` is set, and an `ValueOperationAtMaintenanceTime` candidate only -/// when `maintenance_time` is — a runtime that cannot run an exact +/// proposes an `ValueOperationAtQueryTime` candidate only when +/// `query_time` is set, and an `ValueOperationAtIngestionTime` candidate only +/// when `ingestion_time` 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 ValueOperationCapabilities { /// The runtime can apply an exact operator to summary readouts at /// query evaluation time. - pub read_time: bool, + pub query_time: bool, /// The runtime can apply an exact row transform on the update path, /// feeding its output into maintained summary state. - pub maintenance_time: bool, + pub ingestion_time: bool, } impl ValueOperationCapabilities { /// Neither shape supported. pub const NONE: Self = Self { - read_time: false, - maintenance_time: false, + query_time: false, + ingestion_time: false, }; /// Both shapes supported. pub const ALL: Self = Self { - read_time: true, - maintenance_time: true, + query_time: true, + ingestion_time: true, }; pub fn supports(self, placement: OperationPlacement) -> bool { match placement { - OperationPlacement::Read => self.read_time, - OperationPlacement::Maintenance => self.maintenance_time, + OperationPlacement::Read => self.query_time, + OperationPlacement::Maintenance => self.ingestion_time, } } } diff --git a/crates/asap-aware-mapping/src/exact_composition.rs b/crates/asap-aware-mapping/src/exact_composition.rs index 3568ba1b..58c77e39 100644 --- a/crates/asap-aware-mapping/src/exact_composition.rs +++ b/crates/asap-aware-mapping/src/exact_composition.rs @@ -97,15 +97,15 @@ impl OperationPlacement { /// The availability the composed operator consumes and produces. pub fn data_state(self) -> ExecutionDataState { match self { - Self::Read => ExecutionDataState::READ_ROWS, - Self::Maintenance => ExecutionDataState::MAINTENANCE_ROWS, + Self::Read => ExecutionDataState::QUERY_ROWS, + Self::Maintenance => ExecutionDataState::INGESTION_ROWS, } } pub fn provenance(self) -> ReplacementProvenance { match self { - Self::Read => ReplacementProvenance::ValueOperationAtReadTime, - Self::Maintenance => ReplacementProvenance::ValueOperationAtMaintenanceTime, + Self::Read => ReplacementProvenance::ValueOperationAtQueryTime, + Self::Maintenance => ReplacementProvenance::ValueOperationAtIngestionTime, } } } @@ -199,9 +199,9 @@ impl ExactComposition { }, }; let timing = match self.placement { - OperationPlacement::Read => asap_types::post_asap::ExecutionTiming::ReadTime, + OperationPlacement::Read => asap_types::post_asap::ExecutionTiming::QueryTime, OperationPlacement::Maintenance => { - asap_types::post_asap::ExecutionTiming::MaintenanceTime + asap_types::post_asap::ExecutionTiming::IngestionTime } }; let expr = SummaryExpr::ValueOperation { @@ -230,7 +230,7 @@ impl ExactComposition { /// 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_read_time_reducer(intent: &AggIntent) -> bool { +fn is_query_time_reducer(intent: &AggIntent) -> bool { matches!( intent, AggIntent::Sum { .. } @@ -258,7 +258,7 @@ fn needs_readout(implementation: &Realization) -> bool { } /// The `(op, child)` of a read-time operation-shaped target, or `None`. -fn read_time_shape( +fn query_time_shape( root: &QueryExpr, cost_model: &dyn CostModel, ) -> Option<(ExactOperation, Rc, AggIntent)> { @@ -281,7 +281,7 @@ fn read_time_shape( let [intent] = measures.as_slice() else { return None; }; - if !is_read_time_reducer(intent) { + if !is_query_time_reducer(intent) { return None; } let child_intent = bindable_intent(child)?; @@ -308,7 +308,7 @@ fn read_time_shape( /// The `(op, child)` of a function-shaped target — a per-entity exact /// transform with no accumulator form — or `None`. -fn maintenance_time_shape( +fn ingestion_time_shape( root: &QueryExpr, cost_model: &dyn CostModel, ) -> Option<(ExactOperation, Rc, AggIntent)> { @@ -382,14 +382,14 @@ impl<'a> ExactCompositionStrategy<'a> { let schema = asap_types::post_asap::execution_data_state::lift_plain(&schema); let mut out = Vec::new(); - if let Some((op, child, intent)) = read_time_shape(target.root, self.cost_model) { + if let Some((op, child, intent)) = query_time_shape(target.root, self.cost_model) { if self .cost_model .value_operation_support_evidence(&op, OperationPlacement::Read) != Some(false) { let child_desc = - describe_intent(bindable_intent(&child).expect("checked by read_time_shape")); + describe_intent(bindable_intent(&child).expect("checked by query_time_shape")); out.push(ReplacementSubDAG { strategy: "ExactCompositionStrategy", replacement: Replacement::ExactComposition(ExactComposition { @@ -398,7 +398,7 @@ impl<'a> ExactCompositionStrategy<'a> { child_target: child, schema: schema.clone(), }), - provenance: ReplacementProvenance::ValueOperationAtReadTime, + provenance: ReplacementProvenance::ValueOperationAtQueryTime, rationale: format!( "{} is an exact fold whose input is the readout of {} — a maintained \ accumulator cannot consume query-time values, so instead of collapsing \ @@ -412,7 +412,7 @@ impl<'a> ExactCompositionStrategy<'a> { } } - if let Some((op, child, intent)) = maintenance_time_shape(target.root, self.cost_model) { + if let Some((op, child, intent)) = ingestion_time_shape(target.root, self.cost_model) { if self .cost_model .value_operation_support_evidence(&op, OperationPlacement::Maintenance) @@ -426,7 +426,7 @@ impl<'a> ExactCompositionStrategy<'a> { child_target: child, schema, }), - provenance: ReplacementProvenance::ValueOperationAtMaintenanceTime, + provenance: ReplacementProvenance::ValueOperationAtIngestionTime, rationale: format!( "{} is an exact per-entity function with no accumulator form; as an \ explicit ExactMaintenance on the update path its output can feed a \ @@ -505,7 +505,7 @@ mod tests { } #[test] - fn proposes_read_time_operation_for_max_over_quantile() { + fn proposes_query_time_operation_for_max_over_quantile() { let root = max_over_quantile(); let target = TargetSubDAG::new(&root); let strategy = ExactCompositionStrategy::default_cost_model(); @@ -521,7 +521,7 @@ mod tests { assert_eq!(comp.placement, OperationPlacement::Read); assert_eq!( candidates[0].provenance, - ReplacementProvenance::ValueOperationAtReadTime + ReplacementProvenance::ValueOperationAtQueryTime ); let QueryExpr::Aggregate { child, .. } = root.as_ref() else { unreachable!() @@ -535,7 +535,7 @@ mod tests { } #[test] - fn proposes_read_time_operation_for_avg_over_quantile_alongside_the_rewrite() { + fn proposes_query_time_operation_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); @@ -550,14 +550,14 @@ mod tests { } #[test] - fn proposes_maintenance_time_operation_for_a_per_entity_pass_through_over_raw_input() { + fn proposes_ingestion_time_operation_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::ValueOperationAtMaintenanceTime + ReplacementProvenance::ValueOperationAtIngestionTime ); } @@ -607,7 +607,7 @@ mod tests { } #[test] - fn compose_rejects_a_maintained_state_child_for_a_read_time_operation() { + fn compose_rejects_a_maintained_state_child_for_a_query_time_operation() { let root = max_over_quantile(); let target = TargetSubDAG::new(&root); let candidates = ExactCompositionStrategy::default_cost_model().replacements(&target); @@ -638,7 +638,7 @@ mod tests { assert!(matches!( composed.expr, SummaryExpr::ValueOperation { - timing: ExecutionTiming::ReadTime, + timing: ExecutionTiming::QueryTime, .. } )); @@ -650,7 +650,7 @@ mod tests { } #[test] - fn compose_rejects_a_readout_child_for_a_maintenance_time_operation() { + fn compose_rejects_a_readout_child_for_a_ingestion_time_operation() { let inner = agg(vec![2], default_quantile(0.99), metric_scan(&["zone"])); let root = Rc::new(per_entity(AggIntent::Deriv, inner)); let candidates = @@ -673,7 +673,7 @@ mod tests { assert!(matches!( comp.compose(raw).unwrap().expr, SummaryExpr::ValueOperation { - timing: ExecutionTiming::MaintenanceTime, + timing: ExecutionTiming::IngestionTime, .. } )); diff --git a/crates/asap-aware-mapping/src/maintained_population.rs b/crates/asap-aware-mapping/src/maintained_population.rs index d6b3061b..687a0f7f 100644 --- a/crates/asap-aware-mapping/src/maintained_population.rs +++ b/crates/asap-aware-mapping/src/maintained_population.rs @@ -229,7 +229,7 @@ impl MaintainedPopulationStrategy { cols: cols.clone(), qualifier: qualifier.clone(), }, - timing: ExecutionTiming::ReadTime, + timing: ExecutionTiming::QueryTime, }, })); } @@ -258,7 +258,7 @@ impl MaintainedPopulationStrategy { expr: SummaryExpr::ValueOperation { child: scan, operation: ValueOperation::MaintainPopulation { population }, - timing: ExecutionTiming::MaintenanceTime, + timing: ExecutionTiming::IngestionTime, }, schema: input_schema, guarantee: Some(ResultGuarantee::exact( @@ -269,7 +269,7 @@ impl MaintainedPopulationStrategy { expr: SummaryExpr::ValueOperation { child: maintained, operation: ValueOperation::ReadPopulation { readout }, - timing: ExecutionTiming::ReadTime, + timing: ExecutionTiming::QueryTime, }, schema: plain(root.output_schema().ok()?), guarantee: Some(ResultGuarantee::exact("exact current-population readout")), diff --git a/crates/asap-aware-mapping/src/replacement.rs b/crates/asap-aware-mapping/src/replacement.rs index 7235f7bf..e2ce61d1 100644 --- a/crates/asap-aware-mapping/src/replacement.rs +++ b/crates/asap-aware-mapping/src/replacement.rs @@ -349,13 +349,12 @@ use std::cell::RefCell; use std::collections::{HashMap, HashSet, VecDeque}; use asap_types::post_asap::{ - validate_execution_data_states_at, CandidateCompleteness, EntityIdentity, ErrorMetric, - ExactKind, ExactOperation, ExactOperationSchemaError, ExactParams, ExecutionDataState, - ExecutionDataStateError, ExecutionTiming, GroupingStrategy, NonNegativeWeightProof, - SamplingKind, SamplingParams, SketchAlgorithm, SketchKind, SketchParams, - SketchQuery as PostAsapSketchQuery, StatModelKind, StatModelParams, SummaryExpr, - SummaryFamilyType, SummaryField, SummaryInputExpr, SummaryNode, SummarySchema, SummaryUpdate, - ValueOperation, WaveletKind, WaveletParams, WeightDomain, + validate_execution_data_states_at, EntityIdentity, ExactKind, ExactOperation, + ExactOperationSchemaError, ExactParams, ExecutionDataState, ExecutionDataStateError, + ExecutionTiming, GroupingStrategy, NonNegativeWeightProof, SamplingKind, SamplingParams, + SketchAlgorithm, SketchKind, SketchParams, SketchQuery as PostAsapSketchQuery, StatModelKind, + StatModelParams, SummaryExpr, SummaryFamilyType, SummaryField, SummaryInputExpr, SummaryNode, + SummarySchema, SummaryUpdate, ValueOperation, WaveletKind, WaveletParams, WeightDomain, }; use asap_types::post_asap::{AccuracyError, CompositionOperator, GuaranteeSource, ResultGuarantee}; use asap_types::pre_asap::agg_intent::{agg_is_mergeable, AggIntent}; @@ -482,8 +481,8 @@ pub enum Replacement { Rewrite(Rc), /// An exact operator composed over another target's *own* selected /// decision across an explicit update/readout boundary (issue #171): - /// `ValueOperationAtReadTime` over a child's summary readout, or - /// `ValueOperationAtMaintenanceTime` feeding a maintained summary above. Carries only a + /// `ValueOperationAtQueryTime` over a child's summary readout, or + /// `ValueOperationAtIngestionTime` feeding a maintained summary above. Carries only a /// reference to the child target — [`PlanSpace::global_selection`] /// commits the compatible parent/child pair and /// [`GlobalSelection::assemble_selected_dag`] links it into one validated @@ -557,10 +556,10 @@ pub enum ReplacementProvenance { AccuracyReconciliation, /// [`Replacement::ExactComposition`] with /// [`OperationPlacement::Read`] (issue #171). - ValueOperationAtReadTime, + ValueOperationAtQueryTime, /// [`Replacement::ExactComposition`] with /// [`OperationPlacement::Maintenance`] (issue #171). - ValueOperationAtMaintenanceTime, + ValueOperationAtIngestionTime, } /// A candidate a strategy considered for a target but refused to propose on @@ -1132,7 +1131,7 @@ pub fn posterior_aware_size_params( SketchParams::CmsWithHeap { width: relaxed_width(eps), depth: cms_depth(delta), - heap_size: k as u32, + heap_size: crate::accuracy::topk_capacity(k, eps), } } SketchAlgorithm::CountSketch => { @@ -1632,22 +1631,20 @@ fn exact_topk_over_temporal_values( let QueryExpr::Aggregate { reduction, measures, - output_names, + output_names: _, having: None, child, } = root.as_ref() else { return Ok(None); }; - if !matches!( - measures.as_slice(), - [AggIntent::TopK { - accuracy: AccuracyTarget::Exact, - .. - }] - ) { + let [AggIntent::TopK { + k, + accuracy: AccuracyTarget::Exact, + }] = measures.as_slice() + else { return Ok(None); - } + }; let QueryExpr::Aggregate { reduction: Reduction::PerEntity, child: input, @@ -1669,21 +1666,43 @@ fn exact_topk_over_temporal_values( return Ok(None); } let values = finalize_exact_accumulator(values, child)?; - let node = Rc::new(SummaryNode { + let partition_by = reduction + .group_keys() + .ok_or(RealizationError::PhysicalRealization( + "temporal ranking requires explicit grouping", + ))? + .clone(); + let score = ranking_score_index(child, &values.schema)?; + let sorted = Rc::new(SummaryNode { guarantee: values.guarantee.clone(), - schema: lift(&root.output_schema()?), + schema: values.schema.clone(), expr: SummaryExpr::ValueOperation { child: values, - operation: ValueOperation::Exact(ExactOperation::Aggregate { - reduction: reduction.clone(), - measures: measures.clone(), - output_names: output_names.clone(), - having: None, - }), - timing: ExecutionTiming::ReadTime, + operation: ValueOperation::Sort { + keys: vec![asap_types::pre_asap::SortKey { + expr: QueryExpr::Column(score), + ascending: false, + nulls_first: false, + }], + partition_by: partition_by.clone(), + }, + timing: ExecutionTiming::QueryTime, + }, + }); + let node = Rc::new(SummaryNode { + guarantee: sorted.guarantee.clone(), + schema: sorted.schema.clone(), + expr: SummaryExpr::ValueOperation { + child: sorted, + operation: ValueOperation::Limit { + n: *k, + offset: 0, + partition_by, + }, + timing: ExecutionTiming::QueryTime, }, }); - validate_execution_data_states_at(&node, ExecutionDataState::READ_ROWS)?; + validate_execution_data_states_at(&node, ExecutionDataState::QUERY_ROWS)?; Ok(Some(node)) } @@ -1700,7 +1719,7 @@ fn realize_temporal_average( return Ok(None); }; operator.checked_finite_division = true; - validate_execution_data_states_at(&node, ExecutionDataState::READ_ROWS)?; + validate_execution_data_states_at(&node, ExecutionDataState::QUERY_ROWS)?; Ok(Some(node)) } @@ -1921,7 +1940,7 @@ fn realize_binary( Ok(Some(Rc::new(SummaryNode { expr: SummaryExpr::BinaryOp { - timing: ExecutionTiming::ReadTime, + timing: ExecutionTiming::QueryTime, lhs: lhs_node, rhs: rhs_node, operator: asap_types::post_asap::BinaryOperator { @@ -1946,7 +1965,7 @@ fn finalize_exact_accumulator( node: Rc, logical_output: &QueryExpr, ) -> Result, RealizationError> { - finalize_exact_accumulator_at(node, logical_output, ExecutionTiming::ReadTime) + finalize_exact_accumulator_at(node, logical_output, ExecutionTiming::QueryTime) } fn finalize_exact_accumulator_at( @@ -2226,59 +2245,7 @@ pub(crate) fn construct_summary_with( allocation, )?; if is_counter_weighted_topk(intent, child) { - let values = finalize_exact_accumulator( - realize_child_with(child, planning_inputs, Some(&AccuracyTarget::Exact))?, - child, - )?; - if !values - .guarantee - .as_ref() - .is_some_and(ResultGuarantee::is_exact) - { - return Err(RealizationError::PhysicalRealization( - "CandidateTopK exact rerank input is not exact", - )); - } - let completeness = match candidate.guarantee.clone() { - Some(guarantee) - if guarantee.metric == ErrorMetric::TopKMembership - && !guarantee.has_unknown() => - { - CandidateCompleteness::Certified { guarantee } - } - guarantee => CandidateCompleteness::BestEffort { guarantee }, - }; - let AggIntent::TopK { k, accuracy } = intent else { - unreachable!() - }; - if matches!(accuracy, AccuracyTarget::Exact) - && !matches!(completeness, CandidateCompleteness::Certified { .. }) - { - return Err(RealizationError::PhysicalRealization( - "exact CandidateTopK requires certified candidate completeness", - )); - } - let grouping = reduction.group_keys().cloned().unwrap_or_default(); - let guarantee = match &completeness { - CandidateCompleteness::Certified { guarantee } - | CandidateCompleteness::BestEffort { - guarantee: Some(guarantee), - } => Some(guarantee.clone()), - CandidateCompleteness::BestEffort { guarantee: None } => None, - }; - let node = Rc::new(SummaryNode { - expr: SummaryExpr::CandidateTopK { - candidates: candidate, - values, - k: *k, - grouping, - completeness, - }, - schema: lift(&expr.output_schema()?), - guarantee, - }); - validate_execution_data_states_at(&node, ExecutionDataState::READ_ROWS)?; - return Ok(node); + return finish_weighted_topk(candidate, expr, intent); } return Ok(candidate); } @@ -2287,6 +2254,103 @@ pub(crate) fn construct_summary_with( keep_pre_asap_rc(Rc::new(expr.clone())) } +fn finish_weighted_topk( + candidate: Rc, + logical: &QueryExpr, + intent: &AggIntent, +) -> Result, RealizationError> { + let AggIntent::TopK { k, .. } = intent else { + unreachable!() + }; + let QueryExpr::Aggregate { + reduction: Reduction::Reduce(groups), + child, + .. + } = logical + else { + return Err(RealizationError::PhysicalRealization( + "TopK requires explicit grouping", + )); + }; + let schema = lift(&child.output_schema()?); + let score = ranking_score_index(child, &schema)?; + let cols = schema + .fields + .iter() + .enumerate() + .map(|(i, field)| { + let source = if i == score { + candidate.schema.fields.len() - 1 + } else { + let matches = candidate + .schema + .fields + .iter() + .enumerate() + .filter(|(_, f)| f.name == field.name && f.dtype == field.dtype) + .map(|(i, _)| i) + .collect::>(); + match matches.as_slice() { + [i] => *i, + _ => { + return Err(RealizationError::PhysicalRealization( + "ambiguous TopK output identity", + )) + } + } + }; + Ok(asap_types::pre_asap::query_expr::ProjectItem { + alias: Some(field.name.clone()), + expr: QueryExpr::Column(source), + }) + }) + .collect::, _>>()?; + let guarantee = candidate.guarantee.clone(); + let projected = Rc::new(SummaryNode { + expr: SummaryExpr::ValueOperation { + child: candidate, + operation: ValueOperation::Project { + cols, + qualifier: None, + }, + timing: ExecutionTiming::QueryTime, + }, + schema: schema.clone(), + guarantee: guarantee.clone(), + }); + let sorted = Rc::new(SummaryNode { + expr: SummaryExpr::ValueOperation { + child: projected, + operation: ValueOperation::Sort { + keys: vec![asap_types::pre_asap::SortKey { + expr: QueryExpr::Column(score), + ascending: false, + nulls_first: false, + }], + partition_by: groups.clone(), + }, + timing: ExecutionTiming::QueryTime, + }, + schema: schema.clone(), + guarantee: guarantee.clone(), + }); + let result = Rc::new(SummaryNode { + expr: SummaryExpr::ValueOperation { + child: sorted, + operation: ValueOperation::Limit { + n: *k, + offset: 0, + partition_by: groups.clone(), + }, + timing: ExecutionTiming::QueryTime, + }, + schema, + guarantee, + }); + validate_execution_data_states_at(&result, ExecutionDataState::QUERY_ROWS)?; + Ok(result) +} + fn is_counter_weighted_topk(intent: &AggIntent, child: &QueryExpr) -> bool { matches!(intent, AggIntent::TopK { .. }) && matches!(child, @@ -2453,7 +2517,7 @@ fn maintenance_exact_values(node: Rc) -> Option> { lhs: maintenance_exact_values(lhs.clone())?, rhs: maintenance_exact_values(rhs.clone())?, operator: operator.clone(), - timing: ExecutionTiming::MaintenanceTime, + timing: ExecutionTiming::IngestionTime, } } SummaryExpr::ValueOperation { @@ -2471,7 +2535,7 @@ fn maintenance_exact_values(node: Rc) -> Option> { SummaryExpr::ValueOperation { child: child.clone(), operation: ValueOperation::FinalizeExactAccumulator, - timing: ExecutionTiming::MaintenanceTime, + timing: ExecutionTiming::IngestionTime, } } _ => return Some(node), @@ -2504,10 +2568,71 @@ fn construct_summary_agg( SummaryFamilyType::Sketch(kind, _) if matches!(kind.algorithm(), SketchAlgorithm::CmsWithHeap | SketchAlgorithm::CountSketchWithHeap) ); - let physical_reduction = if keyed_heap && matches!(reduction, Reduction::PerEntity) { - // A heavy-hitter sidecar is one keyed state. `item` identifies the - // ranked series/group inside that state; retaining the logical - // PerEntity reduction here would allocate one full sketch per item. + let rate_weighted = matches!(node, QueryExpr::Aggregate { child, .. } + if is_counter_weighted_topk(intent, child)); + let mut family = family; + let score_population = if rate_weighted { + let bound = planning_inputs.evidence.topk_max_distinct_items(node); + if bound.is_some_and(|n| n == 0 || n > (1u64 << 53)) { + return Err(RealizationError::PhysicalRealization( + "invalid weighted TopK distinct-item bound", + )); + } + if let (Some(n), SummaryFamilyType::Sketch(kind, grouping)) = (bound, &family) { + let (eps, delta) = accuracy_budget(accuracy_target(intent).expect("TopK target")); + let params = default_size_params( + kind.algorithm().clone(), + intent, + eps, + delta / (2.0 * n as f64), + ); + family = SummaryFamilyType::Sketch( + SketchKind::new(kind.algorithm().clone(), params), + grouping.clone(), + ); + } + bound + } else { + None + }; + let physical_reduction = if rate_weighted { + let QueryExpr::Aggregate { child, .. } = node else { + unreachable!() + }; + let source = input.child.output_schema()?; + let Reduction::Reduce(keys) = reduction else { + return Err(RealizationError::PhysicalRealization( + "TopK requires explicit partitions", + )); + }; + if keys.is_without() { + return Err(RealizationError::PhysicalRealization( + "TopK requires explicit partitions", + )); + } + let mapped = keys + .iter() + .map(|index| { + let reference = schema_column_ref(child, *index).ok_or( + RealizationError::PhysicalRealization("invalid TopK partition key"), + )?; + let matches = source + .columns + .iter() + .enumerate() + .filter(|(_, column)| column_ref(column) == reference) + .map(|(index, _)| index) + .collect::>(); + match matches.as_slice() { + [index] => Ok(*index), + _ => Err(RealizationError::PhysicalRealization( + "ambiguous TopK partition key", + )), + } + }) + .collect::, _>>()?; + Reduction::by(mapped) + } else if keyed_heap && matches!(reduction, Reduction::PerEntity) { Reduction::by(vec![]) } else { reduction.clone() @@ -2520,15 +2645,43 @@ fn construct_summary_agg( let out_schema = node.output_schema()?; let state_idx = summary_col_index(&out_schema, &by, per_series); + let readout_schema = if keyed_heap + && matches!(node, QueryExpr::Aggregate { child, .. } if is_counter_weighted_topk(intent, child)) + { + keyed_heap_readout_schema(&input, node)? + } else { + lift(&out_schema) + }; + let summary_input = input.input; - let query = estimate.then(|| readout(intent, &summary_input, planning_inputs.cost)); + let query = estimate.then(|| { + if rate_weighted { + if let SummaryFamilyType::Sketch(kind, _) = &family { + let capacity = match kind.params() { + SketchParams::CmsWithHeap { heap_size, .. } + | SketchParams::CountSketchWithHeap { heap_size, .. } => *heap_size, + _ => unreachable!(), + }; + return PostAsapSketchQuery::TopK { + k: capacity as usize, + }; + } + } + readout(intent, &summary_input, planning_inputs.cost) + }); let mut state_schema = lift(&out_schema); if keyed_heap { let mut state = state_schema.fields[state_idx].clone(); state.dtype = family.clone(); + let mut fields = if rate_weighted { + readout_schema.fields[..reduction.group_keys().map_or(0, |keys| keys.len())].to_vec() + } else { + Vec::new() + }; + fields.push(state); state_schema = SummarySchema { - fields: vec![state], + fields, time_index: None, }; } else if let Some(field) = state_schema.fields.get_mut(state_idx) { @@ -2540,15 +2693,26 @@ fn construct_summary_agg( } } - let bound_child = realize_child_with(&input.child, planning_inputs, child_target)?; - // A maintained parent consumes finalized values, never the child's - // accumulator representation. Keep the read boundary explicit even when - // an exact scalar accumulator currently stores its value directly. - let bound_child = - finalize_exact_accumulator_at(bound_child, &input.child, ExecutionTiming::MaintenanceTime)?; - let bound_child = match maintenance_exact_values(bound_child) { - Some(child) => child, - None => keep_pre_asap(&input.child)?, + let bound_child = realize_child_with( + &input.child, + planning_inputs, + if rate_weighted { + Some(&AccuracyTarget::Exact) + } else { + child_target + }, + )?; + let bound_child = if rate_weighted { + // A fresh query-time summary consumes this evaluation's finalized rates. + // Moving rate snapshots must never accumulate across evaluations. + finalize_exact_accumulator(bound_child, &input.child)? + } else { + let child = finalize_exact_accumulator_at( + bound_child, + &input.child, + ExecutionTiming::IngestionTime, + )?; + maintenance_exact_values(child).unwrap_or(keep_pre_asap(&input.child)?) }; // ── Guarantee (issue #172) ────────────────────────────────────────── @@ -2565,9 +2729,14 @@ fn construct_summary_agg( planning_inputs.evidence.estimator_contract(node), local_target, ); - let guarantee = compose_guarantee( + let membership_query = if rate_weighted { + Some(readout(intent, &summary_input, planning_inputs.cost)) + } else { + query.clone() + }; + let mut guarantee = compose_guarantee( &family, - query.as_ref(), + membership_query.as_ref(), &bound_child, intent, &estimator, @@ -2575,6 +2744,65 @@ fn construct_summary_agg( allocation, )?; + if rate_weighted { + use asap_types::post_asap::{BoundExpr, ProbabilityExpr}; + let target = accuracy_target(intent).expect("TopK target"); + guarantee = if let Some(mut score) = + estimator.local_guarantee(&family, query.as_ref().unwrap()) + { + let count = match score_population { + Some(n) => BoundExpr::Constant { value: n as f64 }, + None => { + score + .provenance + .push(GuaranteeSource::UnavailableStatistic { + statistic: "topk_max_distinct_items".into(), + }); + BoundExpr::Unknown { + statistic: "topk_max_distinct_items".into(), + } + } + }; + score.provenance.push(GuaranteeSource::CompositionStep { + operator: CompositionOperator::ApproximateAggregate, + rule: "simultaneous_score_bounds_over_distinct_partition_item_identities".into(), + }); + score.failure_probability = ProbabilityExpr::Scaled { + count, + inner: Box::new(score.failure_probability), + }; + // #455: missing evidence preserves a logical candidate. Only known + // contributions that already violate the target reject it here. + if !estimator.satisfies(&score.optimistic_floor(), target) { + return Err(RealizationError::PhysicalRealization( + "weighted TopK scores miss accuracy target", + )); + } + let stats = planning_inputs.evidence.propagation_stats( + &CompositionOperator::TopKSelection, + &family, + membership_query.as_ref(), + ); + let mut joint = estimator.propagate( + &CompositionOperator::TopKSelection, + std::slice::from_ref(&score), + None, + &stats, + )?; + joint.failure_probability = ProbabilityExpr::UnionBound { + terms: vec![joint.failure_probability, score.failure_probability], + }; + if !estimator.satisfies(&joint.optimistic_floor(), target) { + return Err(RealizationError::PhysicalRealization( + "weighted TopK joint guarantee misses target", + )); + } + Some(joint) + } else { + None + }; + } + // `reduction` is carried onto `SummaryAgg` verbatim — not flattened to a // bare `Vec` — so `SummaryExecutor::find_candidates` can tell // a genuine empty-`by` reduction apart from a per-entity shape with no @@ -2602,13 +2830,191 @@ fn construct_summary_agg( summary_input: agg, query, }, - schema: lift(&out_schema), + schema: readout_schema, guarantee, })), None => Ok(agg), } } +// Heap readout rows contain the encoded item identity, subpopulation keys, +// and an estimated score. They never inherit the exact-value producer's schema. +fn keyed_heap_readout_schema( + input: &PhysicalSummaryInput, + node: &QueryExpr, +) -> Result { + let source = input.child.output_schema()?; + let mut refs = Vec::new(); + let QueryExpr::Aggregate { + reduction, child, .. + } = node + else { + return Err(RealizationError::PhysicalRealization( + "heap readout requires an aggregate", + )); + }; + if let Reduction::Reduce(groups) = reduction { + if groups.is_without() { + return Err(RealizationError::PhysicalRealization( + "heap readout requires explicit grouping", + )); + } + for index in groups.iter() { + refs.push(schema_column_ref(child, *index).ok_or( + RealizationError::PhysicalRealization("invalid heap partition key"), + )?); + } + } + fn item_refs( + item: &SummaryInputExpr, + schema: &Schema, + refs: &mut Vec, + ) -> Result<(), RealizationError> { + match item { + SummaryInputExpr::Column(column) => refs.push(column.clone()), + SummaryInputExpr::Tuple(items) => { + for item in items { + item_refs(item, schema, refs)?; + } + } + SummaryInputExpr::EntityIdentity(EntityIdentity::PromqlLabelSet { excluding }) => { + if !schema.closed { + return Err(RealizationError::PhysicalRealization( + "dynamic label identity requires an explicit row representation", + )); + } + for (index, column) in schema.columns.iter().enumerate() { + if Some(index) != schema.time_index && column.name != "value" { + let reference = match &column.table { + Some(table) => ColumnRef::Qualified { + table: table.clone(), + name: column.name.clone(), + }, + None => ColumnRef::Named(column.name.clone()), + }; + if !excluding.contains(&reference) { + refs.push(reference); + } + } + } + } + _ => { + return Err(RealizationError::PhysicalRealization( + "unsupported heap item identity", + )) + } + } + Ok(()) + } + item_refs( + input + .input + .item + .as_ref() + .ok_or(RealizationError::PhysicalRealization( + "heap item identity is missing", + ))?, + &source, + &mut refs, + )?; + let mut fields = Vec::::new(); + for reference in refs { + let matches: Vec<_> = source + .columns + .iter() + .filter(|column| match &reference { + ColumnRef::Named(name) => &column.name == name, + ColumnRef::Qualified { table, name } => { + column.table.as_ref() == Some(table) && &column.name == name + } + ColumnRef::SampleValue => column.name == "value", + ColumnRef::Wildcard => false, + }) + .collect(); + let [column] = matches.as_slice() else { + return Err(RealizationError::PhysicalRealization( + "heap key must resolve to exactly one source column", + )); + }; + if fields.iter().any(|field| field.name == column.name) || column.name == "__asap_estimate" + { + return Err(RealizationError::PhysicalRealization( + "heap keys must have distinct output names", + )); + } + fields.push(asap_types::post_asap::SummaryField { + name: column.name.clone(), + dtype: SummaryFamilyType::Plain(column.dtype.clone()), + nullable: column.nullable, + }); + } + if fields.is_empty() { + return Err(RealizationError::PhysicalRealization( + "heap readout has no identity columns", + )); + } + fields.push(asap_types::post_asap::SummaryField { + name: "__asap_estimate".into(), + dtype: SummaryFamilyType::Plain(asap_types::pre_asap::DataType::Float64), + nullable: false, + }); + Ok(SummarySchema { + fields, + time_index: None, + }) +} + +fn ranking_score_index( + logical: &QueryExpr, + values: &SummarySchema, +) -> Result { + let QueryExpr::Aggregate { + reduction, + measures, + .. + } = logical + else { + return Err(RealizationError::PhysicalRealization( + "ranking requires an explicit aggregate score", + )); + }; + if measures.len() != 1 { + return Err(RealizationError::PhysicalRealization( + "ranking requires exactly one score", + )); + } + let index = match reduction { + Reduction::Reduce(groups) if !groups.is_without() => groups.len(), + Reduction::PerEntity => values + .fields + .iter() + .position(|field| field.name == "value") + .ok_or(RealizationError::PhysicalRealization( + "ranking requires the sample value column", + ))?, + _ => { + return Err(RealizationError::PhysicalRealization( + "ranking requires explicit grouping", + )) + } + }; + if Some(index) == values.time_index + || !values.fields.get(index).is_some_and(|field| { + matches!( + field.dtype, + SummaryFamilyType::Plain( + asap_types::pre_asap::DataType::Int64 | asap_types::pre_asap::DataType::Float64 + ) + ) + }) + { + return Err(RealizationError::PhysicalRealization( + "ranking score must be numeric", + )); + } + Ok(index) +} + /// Realize the composite heavy-hitter realization for /// `TopK(Count GROUP BY key)`. The heap sketch consumes the raw keyed stream; /// it does not consume an independently materialized Count result. @@ -2641,24 +3047,13 @@ fn realize_keyed_additive_summary_input( else { return PhysicalSummaryInputRuleResult::NotApplicable; }; - let counter_input = match measures.as_slice() { - [AggIntent::Sum { .. }] => match raw_child.as_ref() { - QueryExpr::Aggregate { - measures, child, .. - } if matches!(measures.as_slice(), [AggIntent::Rate | AggIntent::Increase]) => { - Some(Rc::clone(child)) - } - _ => None, - }, - _ => None, - }; + let counter_input = matches!(measures.as_slice(), [AggIntent::Sum { .. }]) + && matches!(raw_child.as_ref(), QueryExpr::Aggregate { measures, .. } + if matches!(measures.as_slice(), [AggIntent::Rate | AggIntent::Increase])); let weight = match measures.as_slice() { [AggIntent::Count { .. }] => SummaryInputExpr::Constant(1.0), - [AggIntent::Sum { .. }] if counter_input.is_some() => { - SummaryInputExpr::ResetAwareCounterDelta { - value: ColumnRef::SampleValue, - series: EntityIdentity::PromqlLabelSet { excluding: vec![] }, - } + [AggIntent::Sum { .. }] if counter_input => { + SummaryInputExpr::Column(ColumnRef::SampleValue) } [AggIntent::Sum { col }] => SummaryInputExpr::Column(match col { None => ColumnRef::SampleValue, @@ -2677,7 +3072,7 @@ fn realize_keyed_additive_summary_input( [AggIntent::Count { .. }] => WeightDomain::NonNegative { proof: NonNegativeWeightProof::UnitCount, }, - [AggIntent::Sum { .. }] if counter_input.is_some() => WeightDomain::NonNegative { + [AggIntent::Sum { .. }] if counter_input => WeightDomain::NonNegative { proof: NonNegativeWeightProof::ResetAwareCounterDerivative, }, _ => WeightDomain::UnknownOrSigned, @@ -2736,7 +3131,7 @@ fn realize_keyed_additive_summary_input( } }; PhysicalSummaryInputRuleResult::Realized(PhysicalSummaryInput { - child: counter_input.unwrap_or_else(|| Rc::clone(raw_child)), + child: Rc::clone(raw_child), input: SummaryUpdate { item: Some(item), weight, @@ -4144,7 +4539,7 @@ impl<'a> GlobalSelection<'a> { /// a [`Replacement::Summary`] is /// re-linked so its `SummaryAgg` child is the child target's own /// DAG assembly whenever that is phase-legal beneath maintenance - /// (so a child that chose an `ValueOperationAtMaintenanceTime` actually ends up under + /// (so a child that chose an `ValueOperationAtIngestionTime` 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. @@ -4163,7 +4558,7 @@ impl<'a> GlobalSelection<'a> { if let Some(node) = self.assembled_nodes.borrow().get(&ptr) { return Ok(Rc::clone(node)); } - let node = if read_time_nested_sum(target) { + let node = if query_time_nested_sum(target) { self.assemble_residual(target)? } else { match self @@ -4214,8 +4609,8 @@ impl<'a> GlobalSelection<'a> { let Some(pred) = normalized_pred else { return keep_pre_asap(target); }; - let left = self.assemble_target(left)?; - let right = self.assemble_target(right)?; + let left = finalize_exact_accumulator(self.assemble_target(left)?, left)?; + let right = finalize_exact_accumulator(self.assemble_target(right)?, right)?; let guarantee = relational_join_guarantee(left.guarantee.as_ref(), right.guarantee.as_ref()); let node = Rc::new(SummaryNode { @@ -4224,11 +4619,12 @@ impl<'a> GlobalSelection<'a> { right, kind: kind.clone(), pred, + pruning: None, }, schema: lift(&target.output_schema()?), guarantee, }); - validate_execution_data_states_at(&node, ExecutionDataState::READ_ROWS)?; + validate_execution_data_states_at(&node, ExecutionDataState::QUERY_ROWS)?; return Ok(node); } let (child_target, operation) = match target.as_ref() { @@ -4262,6 +4658,10 @@ impl<'a> GlobalSelection<'a> { ValueOperation::Limit { n: *n, offset: *offset, + partition_by: match child.as_ref() { + QueryExpr::Sort { partition_by, .. } => partition_by.clone(), + _ => Default::default(), + }, }, ), QueryExpr::Aggregate { @@ -4270,7 +4670,7 @@ impl<'a> GlobalSelection<'a> { output_names, having, child, - } if read_time_nested_sum(target) => ( + } if query_time_nested_sum(target) => ( child, ValueOperation::Exact(ExactOperation::Aggregate { reduction: reduction.clone(), @@ -4281,23 +4681,18 @@ impl<'a> GlobalSelection<'a> { ), _ => return keep_pre_asap(target), }; - let child = self.assemble_target(child_target)?; - let child = if matches!(operation, ValueOperation::Exact(_)) { - finalize_exact_accumulator(child, child_target)? - } else { - child - }; + let child = finalize_exact_accumulator(self.assemble_target(child_target)?, child_target)?; let guarantee = child.guarantee.clone(); let node = Rc::new(SummaryNode { expr: SummaryExpr::ValueOperation { child, operation, - timing: ExecutionTiming::ReadTime, + timing: ExecutionTiming::QueryTime, }, schema: lift(&target.output_schema()?), guarantee, }); - validate_execution_data_states_at(&node, ExecutionDataState::READ_ROWS)?; + validate_execution_data_states_at(&node, ExecutionDataState::QUERY_ROWS)?; Ok(node) } @@ -4338,7 +4733,7 @@ impl<'a> GlobalSelection<'a> { /// reduction of the inner summary values. Maintaining the outer SUM directly /// would hide that inner temporal aggregate inside `KeepPreAsap` and lose its /// independently selected summary. -fn read_time_nested_sum(target: &QueryExpr) -> bool { +fn query_time_nested_sum(target: &QueryExpr) -> bool { let QueryExpr::Aggregate { measures, having: None, @@ -4405,10 +4800,8 @@ fn relink_agg_child(node: &Rc, new_child: &Rc) -> Rc rebuilt, Err(_) => Rc::clone(node), } @@ -4418,7 +4811,7 @@ fn relink_agg_child(node: &Rc, new_child: &Rc) -> Rc) -> Option<&Rc> { match &node.expr { @@ -4442,7 +4835,7 @@ struct CompositionContext { /// 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 `ValueOperationAtMaintenanceTime` here feeds. + /// parent chose a bound `Summary` — what an `ValueOperationAtIngestionTime` here feeds. maintaining_parent: HashMap<*const QueryExpr, Rc>, } @@ -4852,7 +5245,7 @@ impl PlanSpace { }; // Record the maintained summary this site's bound candidate - // builds, for a child that may compose an `ValueOperationAtMaintenanceTime` + // builds, for a child that may compose an `ValueOperationAtIngestionTime` // beneath it. if let (Some(Replacement::Summary(node)), QueryExpr::Aggregate { child, .. }) = (chosen.map(|c| &c.replacement), group.target.as_ref()) @@ -5887,13 +6280,40 @@ mod tests { .unwrap() .expect("exact Top-K candidate"); assert!(node.guarantee.as_ref().unwrap().is_exact()); - assert!(matches!( - node.expr, - SummaryExpr::ValueOperation { - operation: ValueOperation::Exact(ExactOperation::Aggregate { .. }), - .. - } - )); + let SummaryExpr::ValueOperation { + child: sorted, + operation: + ValueOperation::Limit { + n, + offset, + partition_by, + }, + .. + } = &node.expr + else { + panic!("temporal TopK must compose Sort and Limit"); + }; + assert_eq!((*n, *offset), (5, 0)); + let SummaryExpr::ValueOperation { + operation: + ValueOperation::Sort { + keys, + partition_by: sort_groups, + }, + child: values, + .. + } = &sorted.expr + else { + panic!("Limit must consume sorted temporal values"); + }; + assert_eq!(sort_groups, partition_by); + assert_eq!( + partition_by.keys().len(), + usize::from(query.contains("by(job)")) + ); + assert_eq!(keys.len(), 1); + assert!(!keys[0].ascending); + assert_eq!(node.schema, values.schema); asap_types::post_asap::compile_executable_dag(&node).unwrap(); } } @@ -6229,7 +6649,7 @@ mod tests { } #[test] - fn topk_heap_size_tracks_k() { + fn topk_heap_capacity_respects_accuracy_and_output_count() { let intent = AggIntent::TopK { k: 25, accuracy: eps(0.01), @@ -6244,7 +6664,7 @@ mod tests { else { unreachable!("SketchKind validates CmsWithHeap params") }; - assert_eq!(*heap_size, 25); + assert_eq!(*heap_size, 100); assert_eq!(*width, 272); // ⌈e/0.01⌉ assert_eq!(*depth, 5); } @@ -6441,7 +6861,7 @@ mod tests { } => { assert_eq!(width, 68); assert_eq!(depth, 5); - assert_eq!(heap_size, 7); + assert_eq!(heap_size, 100); } other => panic!("expected CmsWithHeap, got {other:?}"), } @@ -8602,7 +9022,7 @@ mod tests { let SummaryExpr::ValueOperation { child, operation: ValueOperation::FinalizeExactAccumulator, - timing: ExecutionTiming::MaintenanceTime, + timing: ExecutionTiming::IngestionTime, } = &child.expr else { panic!("expected explicit maintenance readout"); @@ -9626,4 +10046,97 @@ mod tests { if summary_family_algorithm(node) == SketchAlgorithm::Hll && node.guarantee.as_ref().is_some_and(|g| DefaultAccuracyModel.satisfies(g, &target))))); } } + + // A value projection cannot consume an opaque exact accumulator edge. + #[test] + fn residual_projection_finalizes_selected_exact_state() { + let inner = Rc::new(agg(vec![], AggIntent::Sum { col: None }, metric_scan(&[]))); + let root = Rc::new(QueryExpr::Project { + cols: vec![asap_types::pre_asap::ProjectItem { + expr: QueryExpr::Column(0), + alias: Some("result".into()), + }], + qualifier: None, + child: inner.clone(), + }); + let space = search_workload_with_targets( + vec![("q", root.clone(), Some(AccuracyTarget::Exact))], + &default_strategies(), + &DefaultAccuracyModel, + ); + let selected = space.global_selection(&DefaultCostModel); + selected + .assembled_nodes + .borrow_mut() + .insert(Rc::as_ptr(&inner), realize(inner.as_ref()).unwrap()); + let node = selected.assemble_target(&root).unwrap(); + let SummaryExpr::ValueOperation { + child, + operation: ValueOperation::Project { .. }, + .. + } = &node.expr + else { + panic!("expected Project"); + }; + assert!(matches!( + child.expr, + SummaryExpr::ValueOperation { + operation: ValueOperation::FinalizeExactAccumulator, + .. + } + )); + assert!(child + .schema + .fields + .iter() + .all(|field| matches!(field.dtype, SummaryFamilyType::Plain(_)))); + } + // A numeric group key must not be mistaken for the ranked aggregate score. + #[test] + fn ranking_uses_aggregate_output_position_not_first_numeric_column() { + let logical = agg(vec![2], AggIntent::Sum { col: None }, metric_scan(&["id"])); + let mut values = lift(&logical.output_schema().unwrap()); + values.fields[0].dtype = SummaryFamilyType::Plain(DataType::Int64); + assert_eq!(ranking_score_index(&logical, &values).unwrap(), 1); + } + // A heap's key schema is derived from its encoded item, not all label columns. + #[test] + fn heap_readout_preserves_numeric_item_identity() { + let mut raw = metric_scan(&["id", "description"]); + let QueryExpr::Scan { schema, .. } = &mut raw else { + unreachable!() + }; + schema.columns[2].dtype = DataType::Int64; + let node = agg( + vec![], + AggIntent::TopK { + k: 2, + accuracy: AccuracyTarget::Epsilon(0.01), + }, + agg(vec![2], AggIntent::Sum { col: None }, raw.clone()), + ); + let input = PhysicalSummaryInput { + child: Rc::new(raw), + input: SummaryUpdate { + item: Some(SummaryInputExpr::Column(ColumnRef::Named("id".into()))), + weight: SummaryInputExpr::Constant(1.0), + weight_domain: WeightDomain::NonNegative { + proof: NonNegativeWeightProof::UnitCount, + }, + }, + }; + let schema = keyed_heap_readout_schema(&input, &node).unwrap(); + assert_eq!( + schema + .fields + .iter() + .map(|f| f.name.as_str()) + .collect::>(), + vec!["id", "__asap_estimate"] + ); + assert_eq!( + schema.fields[0].dtype, + SummaryFamilyType::Plain(DataType::Int64) + ); + } } diff --git a/crates/asap-aware-mapping/src/summary_maintenance_cost/estimator.rs b/crates/asap-aware-mapping/src/summary_maintenance_cost/estimator.rs index eb714a1d..4af4dabd 100644 --- a/crates/asap-aware-mapping/src/summary_maintenance_cost/estimator.rs +++ b/crates/asap-aware-mapping/src/summary_maintenance_cost/estimator.rs @@ -32,7 +32,7 @@ pub(super) fn estimate_heterogeneous_summary( SummaryExpr::SummaryAgg { child, .. } | SummaryExpr::ValueOperation { child, .. } => { summary_source_selections(child, seen, out)? } - SummaryExpr::SummaryMerge { children } => { + SummaryExpr::SummaryMerge { children, .. } => { for child in children { summary_source_selections(child, seen, out)?; } @@ -48,11 +48,6 @@ pub(super) fn estimate_heterogeneous_summary( outer: left, inner: right, .. - } - | SummaryExpr::CandidateTopK { - candidates: left, - values: right, - .. } => { summary_source_selections(left, seen, out)?; summary_source_selections(right, seen, out)?; @@ -303,27 +298,7 @@ pub(super) fn estimate_heterogeneous_summary( io_bytes, )?; } - SummaryExpr::CandidateTopK { - candidates, values, .. - } => { - let operation = summary_operation_evidence(node, evidence)?.resource(); - *cpu_ops += evaluation_count as f64 - * validated_operator_executions("candidate_topk", operation)? as f64 - * validated_operator_cpu("candidate_topk", operation.cpu_ops)?; - add_operator_io(io_bytes, operation, evaluation_count)?; - for input in [candidates, values] { - visit_ops( - input, - seen, - by_node, - evidence, - scope, - evaluation_count, - cpu_ops, - io_bytes, - )?; - } - } + SummaryExpr::ValueOperation { child, .. } => { let operation = summary_operation_evidence(node, evidence)?.resource(); *cpu_ops += evaluation_count as f64 @@ -368,7 +343,7 @@ pub(super) fn estimate_heterogeneous_summary( io_bytes, )?; } - SummaryExpr::SummaryMerge { children } => { + SummaryExpr::SummaryMerge { children, .. } => { let operation = summary_operation_evidence(node, evidence)?.resource(); let merge = validated_operator_cpu("summary_merge", operation.cpu_ops)?; *cpu_ops += evaluation_count as f64 @@ -443,7 +418,7 @@ pub(super) fn estimate_heterogeneous_summary( collect_aggs(child, seen, out); } SummaryExpr::ValueOperation { child, .. } => collect_aggs(child, seen, out), - SummaryExpr::SummaryMerge { children } => { + SummaryExpr::SummaryMerge { children, .. } => { children .iter() .for_each(|child| collect_aggs(child, seen, out)); @@ -459,11 +434,6 @@ pub(super) fn estimate_heterogeneous_summary( outer: left, inner: right, .. - } - | SummaryExpr::CandidateTopK { - candidates: left, - values: right, - .. } => { collect_aggs(left, seen, out); collect_aggs(right, seen, out); @@ -652,7 +622,7 @@ fn validate_summary_edges_and_physical_ids( SummaryExpr::SummaryAgg { child, .. } | SummaryExpr::ValueOperation { child, .. } => { vec![child] } - SummaryExpr::SummaryMerge { children } => { + SummaryExpr::SummaryMerge { children, .. } => { children.iter().map(|child| child.as_ref()).collect() } SummaryExpr::SummarySubtract { left, right } @@ -666,11 +636,6 @@ fn validate_summary_edges_and_physical_ids( outer: left, inner: right, .. - } - | SummaryExpr::CandidateTopK { - candidates: left, - values: right, - .. } => vec![left, right], SummaryExpr::SummaryDelete { summary_input, .. } | SummaryExpr::SummaryEstimate { summary_input, .. } => vec![summary_input], @@ -829,7 +794,7 @@ pub(super) fn estimate_transient_liveness( SummaryExpr::SummaryAgg { child, .. } | SummaryExpr::ValueOperation { child, .. } => { vec![child] } - SummaryExpr::SummaryMerge { children } => { + SummaryExpr::SummaryMerge { children, .. } => { children.iter().map(|child| child.as_ref()).collect() } SummaryExpr::SummarySubtract { left, right } @@ -843,11 +808,6 @@ pub(super) fn estimate_transient_liveness( outer: left, inner: right, .. - } - | SummaryExpr::CandidateTopK { - candidates: left, - values: right, - .. } => vec![left, right], SummaryExpr::SummaryDelete { summary_input, .. } | SummaryExpr::SummaryEstimate { summary_input, .. } => vec![summary_input], @@ -891,7 +851,6 @@ pub(super) fn estimate_transient_liveness( SummaryExpr::SummaryMerge { .. } | SummaryExpr::BinaryOp { .. } | SummaryExpr::RelationalJoin { .. } - | SummaryExpr::CandidateTopK { .. } | SummaryExpr::ValueOperation { .. } | SummaryExpr::SummarySubtract { .. } | SummaryExpr::SummaryDelete { .. } @@ -958,7 +917,7 @@ pub(super) fn evidence_nodes(root: &SummaryNode) -> (Vec<&SummaryNode>, Vec<&Sum visit(child, seen, aggregations, joins); } SummaryExpr::ValueOperation { child, .. } => visit(child, seen, aggregations, joins), - SummaryExpr::SummaryMerge { children } => { + SummaryExpr::SummaryMerge { children, .. } => { for child in children { visit(child, seen, aggregations, joins); } @@ -974,11 +933,6 @@ pub(super) fn evidence_nodes(root: &SummaryNode) -> (Vec<&SummaryNode>, Vec<&Sum outer: left, inner: right, .. - } - | SummaryExpr::CandidateTopK { - candidates: left, - values: right, - .. } => { if matches!(&node.expr, SummaryExpr::SummaryJoin { .. }) { joins.push(node); @@ -1300,7 +1254,7 @@ fn count_operations(root: &SummaryNode) -> Result { + SummaryExpr::SummaryMerge { children, .. } => { if children.is_empty() { return Err(AnalyticalCostError::InvalidPhysicalDag( "summary merge has no children", @@ -1330,12 +1284,7 @@ fn count_operations(root: &SummaryNode) -> Result { - visit(candidates, seen, counts)?; - visit(values, seen, counts)?; - } + SummaryExpr::ValueOperation { child, .. } => visit(child, seen, counts)?, SummaryExpr::SummaryDelete { summary_input, .. } => { counts.deletes_per_update = counts diff --git a/crates/asap-aware-mapping/src/summary_maintenance_cost/model.rs b/crates/asap-aware-mapping/src/summary_maintenance_cost/model.rs index afa91c75..e339994a 100644 --- a/crates/asap-aware-mapping/src/summary_maintenance_cost/model.rs +++ b/crates/asap-aware-mapping/src/summary_maintenance_cost/model.rs @@ -2898,6 +2898,7 @@ mod tests { if merge { root = Rc::new(SummaryNode { expr: SummaryExpr::SummaryMerge { + timing: asap_types::post_asap::ExecutionTiming::IngestionTime, children: vec![Rc::clone(&agg), Rc::clone(&agg)], }, schema: schema.clone(), @@ -2982,7 +2983,7 @@ mod tests { let operand = summary_with_operations(false, false, false); Rc::new(SummaryNode { expr: SummaryExpr::BinaryOp { - timing: asap_types::post_asap::ExecutionTiming::ReadTime, + timing: asap_types::post_asap::ExecutionTiming::QueryTime, lhs: Rc::clone(&operand), rhs: operand, operator: asap_types::post_asap::BinaryOperator { @@ -3184,7 +3185,7 @@ mod tests { } SummaryExpr::SummaryAgg { child, .. } | SummaryExpr::ValueOperation { child, .. } => retained(model, child, seen), - SummaryExpr::SummaryMerge { children } => { + SummaryExpr::SummaryMerge { children, .. } => { for child in children { retained(model, child, seen); } @@ -3200,11 +3201,6 @@ mod tests { outer: left, inner: right, .. - } - | SummaryExpr::CandidateTopK { - candidates: left, - values: right, - .. } => { retained(model, left, seen); retained(model, right, seen); @@ -3322,7 +3318,7 @@ mod tests { StreamingSummaryOperatorEvidence::Merge(SummaryOperatorResourceEvidence { physical_id: format!("merge-{node:p}"), inputs: match &node.expr { - SummaryExpr::SummaryMerge { children } => { + SummaryExpr::SummaryMerge { children, .. } => { vec![test_edge(); children.len()] } _ => unreachable!(), @@ -3399,7 +3395,7 @@ mod tests { SummaryExpr::ValueOperation { child, .. } => { owning_aggs(child, seen, owners) } - SummaryExpr::SummaryMerge { children } => { + SummaryExpr::SummaryMerge { children, .. } => { for child in children { owning_aggs(child, seen, owners); } @@ -3415,11 +3411,6 @@ mod tests { outer: left, inner: right, .. - } - | SummaryExpr::CandidateTopK { - candidates: left, - values: right, - .. } => { owning_aggs(left, seen, owners); owning_aggs(right, seen, owners); @@ -3448,7 +3439,7 @@ mod tests { | SummaryExpr::ValueOperation { child, .. } => { bind_ops(model, child, seen, inputs, cpu) } - SummaryExpr::SummaryMerge { children } => { + SummaryExpr::SummaryMerge { children, .. } => { for child in children { bind_ops(model, child, seen, inputs, cpu); } @@ -3464,11 +3455,6 @@ mod tests { outer: left, inner: right, .. - } - | SummaryExpr::CandidateTopK { - candidates: left, - values: right, - .. } => { bind_ops(model, left, seen, inputs, cpu); bind_ops(model, right, seen, inputs, cpu); diff --git a/crates/asap-aware-mapping/src/summary_maintenance_cost/window.rs b/crates/asap-aware-mapping/src/summary_maintenance_cost/window.rs index d4ec85d6..eff12cf3 100644 --- a/crates/asap-aware-mapping/src/summary_maintenance_cost/window.rs +++ b/crates/asap-aware-mapping/src/summary_maintenance_cost/window.rs @@ -45,7 +45,7 @@ pub(super) fn summary_aggregation_identities(root: &SummaryNode) -> HashSet<*con visit(child, seen, out); } SummaryExpr::ValueOperation { child, .. } => visit(child, seen, out), - SummaryExpr::SummaryMerge { children } => { + SummaryExpr::SummaryMerge { children, .. } => { for child in children { visit(child, seen, out); } @@ -61,11 +61,6 @@ pub(super) fn summary_aggregation_identities(root: &SummaryNode) -> HashSet<*con outer: left, inner: right, .. - } - | SummaryExpr::CandidateTopK { - candidates: left, - values: right, - .. } => { visit(left, seen, out); visit(right, seen, out); diff --git a/crates/asap-aware-mapping/src/summary_maintenance_dag_export.rs b/crates/asap-aware-mapping/src/summary_maintenance_dag_export.rs index bbb0e2c8..efc8cf13 100644 --- a/crates/asap-aware-mapping/src/summary_maintenance_dag_export.rs +++ b/crates/asap-aware-mapping/src/summary_maintenance_dag_export.rs @@ -154,14 +154,9 @@ fn summary_children(expr: &SummaryExpr) -> Vec<&Rc> { | SummaryExpr::SummarySubtract { left: outer, right: inner, - } - | SummaryExpr::CandidateTopK { - candidates: outer, - values: inner, - .. } => vec![outer, inner], SummaryExpr::SummaryDelete { summary_input, .. } | SummaryExpr::SummaryEstimate { summary_input, .. } => vec![summary_input], - SummaryExpr::SummaryMerge { children } => children.iter().collect(), + SummaryExpr::SummaryMerge { children, .. } => children.iter().collect(), } } diff --git a/crates/asap-aware-mapping/src/summary_maintenance_lifecycle.rs b/crates/asap-aware-mapping/src/summary_maintenance_lifecycle.rs index f9d9d17f..87ef8d03 100644 --- a/crates/asap-aware-mapping/src/summary_maintenance_lifecycle.rs +++ b/crates/asap-aware-mapping/src/summary_maintenance_lifecycle.rs @@ -1021,11 +1021,6 @@ fn collect_summary_aggs( | SummaryExpr::SummarySubtract { left: outer, right: inner, - } - | SummaryExpr::CandidateTopK { - candidates: outer, - values: inner, - .. } => { collect_summary_aggs(outer, seen, output); collect_summary_aggs(inner, seen, output); @@ -1034,7 +1029,7 @@ fn collect_summary_aggs( | SummaryExpr::SummaryEstimate { summary_input, .. } => { collect_summary_aggs(summary_input, seen, output) } - SummaryExpr::SummaryMerge { children } => { + SummaryExpr::SummaryMerge { children, .. } => { for child in children { collect_summary_aggs(child, seen, output); } @@ -2334,6 +2329,7 @@ mod tests { let shared = summary(); let root = Rc::new(SummaryNode { expr: SummaryExpr::SummaryMerge { + timing: asap_types::post_asap::ExecutionTiming::IngestionTime, children: vec![Rc::clone(&shared), Rc::clone(&shared)], }, schema: shared.schema.clone(), diff --git a/crates/integration-tests/tests/exact_composition.rs b/crates/integration-tests/tests/exact_composition.rs index 45a6f34c..40b593a1 100644 --- a/crates/integration-tests/tests/exact_composition.rs +++ b/crates/integration-tests/tests/exact_composition.rs @@ -5,7 +5,7 @@ //! //! Covers the issue's integration matrix: both nesting directions, grouped //! fine-to-coarse and identity folds, one inner summary shared by several -//! queries, illegal readout-under-maintenance rejection, a runtime without +//! queries, phase-aware summary construction, a runtime without //! the capability, a cost model without statistics, and pre/post-ASAP //! schemas plus shared `Rc` identity — along with pins for every //! already-supported exact-accumulator nesting. @@ -17,8 +17,8 @@ use asap_aware_mapping::cost_model::{ ValueOperationCapabilities, }; use asap_aware_mapping::replacement::{ - default_strategies_with, search_workload_with, RealizationError, Replacement, - ReplacementProvenance, ReplacementStrategy, SketchAlgorithmStrategy, TargetSubDAG, + default_strategies_with, search_workload_with, Replacement, ReplacementProvenance, + ReplacementStrategy, SketchAlgorithmStrategy, TargetSubDAG, }; use asap_aware_mapping::{ CostModel, DefaultCostModel, EvaluationRate, ExplanationKind, OperationPlacement, @@ -26,9 +26,8 @@ use asap_aware_mapping::{ use asap_integration_tests::fixtures::lower_promql; use asap_types::dag_export; use asap_types::post_asap::{ - validate_execution_data_states, ExactKind, ExactOperation, ExecutionDataState, - ExecutionDataStateError, ExecutionTiming, SketchAlgorithm, SummaryExpr, SummaryFamilyType, - SummaryNode, SummaryUpdate, + validate_execution_data_states, ExactKind, ExactOperation, ExecutionDataState, ExecutionTiming, + SketchAlgorithm, SummaryExpr, SummaryFamilyType, SummaryNode, SummaryUpdate, }; use asap_types::pre_asap::agg_intent::{default_quantile, AggIntent}; use asap_types::pre_asap::query_expr::{QueryExpr, Reduction, Source}; @@ -367,7 +366,7 @@ fn every_exact_accumulator_is_finalized_before_an_outer_sketch() { let SummaryExpr::ValueOperation { child, operation: asap_types::post_asap::ValueOperation::FinalizeExactAccumulator, - timing: ExecutionTiming::MaintenanceTime, + timing: ExecutionTiming::IngestionTime, } = &child.expr else { panic!("{kind:?}: missing maintenance finalization"); @@ -387,12 +386,12 @@ fn every_exact_accumulator_is_finalized_before_an_outer_sketch() { // ── direction 1: outer exact fold over an inner summary readout ──────── /// Before this PR both `max`/`avg` over a quantile collapsed into one -/// opaque `KeepPreAsap`. Now: the outer group holds an `ValueOperationAtReadTime` +/// opaque `KeepPreAsap`. Now: the outer group holds an `ValueOperationAtQueryTime` /// candidate referencing the inner target, the inner group keeps its own /// sketch candidates, and with statistics the pair is committed and -/// materializes as `ValueOperationAtReadTime → SummaryEstimate → SummaryAgg`. +/// materializes as `ValueOperationAtQueryTime → SummaryEstimate → SummaryAgg`. #[test] -fn max_and_avg_over_quantile_compose_at_read_time_with_statistics() { +fn max_and_avg_over_quantile_compose_at_query_time_with_statistics() { for intent in [AggIntent::Max { col: None }, AggIntent::Avg { col: None }] { let root = agg(vec![0], intent.clone(), fine_quantile()); let space = plan(vec![("q", Rc::clone(&root))], &StatsModel); @@ -406,8 +405,8 @@ fn max_and_avg_over_quantile_compose_at_read_time_with_statistics() { outer_group .candidates .iter() - .any(|c| c.provenance == ReplacementProvenance::ValueOperationAtReadTime), - "{intent:?}: outer group must hold an ValueOperationAtReadTime candidate" + .any(|c| c.provenance == ReplacementProvenance::ValueOperationAtQueryTime), + "{intent:?}: outer group must hold an ValueOperationAtQueryTime candidate" ); let inner_group = space.candidates_for_target(inner).unwrap(); assert!( @@ -424,7 +423,7 @@ fn max_and_avg_over_quantile_compose_at_read_time_with_statistics() { let chosen = selected.chosen.expect("a decision"); assert_eq!( chosen.provenance, - ReplacementProvenance::ValueOperationAtReadTime + ReplacementProvenance::ValueOperationAtQueryTime ); let decision = selected .composition @@ -446,12 +445,12 @@ fn max_and_avg_over_quantile_compose_at_read_time_with_statistics() { let composed = selection.assemble_selected_dag(&root).unwrap().unwrap(); let SummaryExpr::ValueOperation { child, - timing: ExecutionTiming::ReadTime, + timing: ExecutionTiming::QueryTime, .. } = &composed.expr else { panic!( - "{intent:?}: expected ValueOperationAtReadTime root, got {:?}", + "{intent:?}: expected ValueOperationAtQueryTime root, got {:?}", composed.expr ); }; @@ -495,7 +494,7 @@ fn avg_over_quantile_keeps_the_sum_over_count_rewrite_as_a_competitor() { let group = space.candidates_for_target(&space.roots[0].1).unwrap(); let provenances: Vec<_> = group.candidates.iter().map(|c| c.provenance).collect(); assert!(provenances.contains(&ReplacementProvenance::LogicalRewrite)); - assert!(provenances.contains(&ReplacementProvenance::ValueOperationAtReadTime)); + assert!(provenances.contains(&ReplacementProvenance::ValueOperationAtQueryTime)); } /// Grouped fine-to-coarse fold (`by (zone)` over `by (zone, host)`) and the @@ -524,7 +523,7 @@ fn identity_and_genuine_multi_row_folds_both_compose() { matches!( composed.expr, SummaryExpr::ValueOperation { - timing: ExecutionTiming::ReadTime, + timing: ExecutionTiming::QueryTime, .. } ), @@ -587,10 +586,10 @@ fn a_shared_inner_summary_is_materialized_once_for_several_outer_folds() { let child_of = |n: &Rc| match &n.expr { SummaryExpr::ValueOperation { child, - timing: ExecutionTiming::ReadTime, + timing: ExecutionTiming::QueryTime, .. } => Rc::clone(child), - other => panic!("expected ValueOperationAtReadTime, got {other:?}"), + other => panic!("expected ValueOperationAtQueryTime, got {other:?}"), }; assert!( Rc::ptr_eq(&child_of(&composed[0]), &child_of(&composed[1])), @@ -601,11 +600,11 @@ fn a_shared_inner_summary_is_materialized_once_for_several_outer_folds() { // ── direction 2: outer summary over an inner exact maintenance-time operation ─ /// `quantile(0.99, deriv(latency[5m]))`: `deriv` has no accumulator form. -/// The function target gets an `ValueOperationAtMaintenanceTime` candidate; with a +/// The function target gets an `ValueOperationAtIngestionTime` candidate; with a /// maintained summary above it and statistics, it is committed, and the /// outer summary's materialization is re-linked over it. #[test] -fn outer_summary_over_an_exact_function_composes_at_maintenance_time() { +fn outer_summary_over_an_exact_function_composes_at_ingestion_time() { use std::time::Duration; let deriv = per_entity( AggIntent::Deriv, @@ -625,13 +624,13 @@ fn outer_summary_over_an_exact_function_composes_at_maintenance_time() { .unwrap() .candidates .iter() - .any(|c| c.provenance == ReplacementProvenance::ValueOperationAtMaintenanceTime)); + .any(|c| c.provenance == ReplacementProvenance::ValueOperationAtIngestionTime)); let selection = space.global_selection(&StatsModel); let deriv_sel = selection.for_target(deriv).unwrap(); assert_eq!( deriv_sel.chosen.unwrap().provenance, - ReplacementProvenance::ValueOperationAtMaintenanceTime + ReplacementProvenance::ValueOperationAtIngestionTime ); let decision = deriv_sel.composition.as_ref().unwrap(); assert!(decision.child_candidate.is_none(), "function input is raw"); @@ -646,12 +645,12 @@ fn outer_summary_over_an_exact_function_composes_at_maintenance_time() { }; let SummaryExpr::ValueOperation { child: raw, - timing: ExecutionTiming::MaintenanceTime, + timing: ExecutionTiming::IngestionTime, .. } = &child.expr else { panic!( - "expected ValueOperationAtMaintenanceTime under the maintained summary, got {:?}", + "expected ValueOperationAtIngestionTime under the maintained summary, got {:?}", child.expr ); }; @@ -659,23 +658,21 @@ fn outer_summary_over_an_exact_function_composes_at_maintenance_time() { let assignment = validate_execution_data_states(&composed).unwrap(); assert_eq!( assignment.data_state_of(child), - Some(ExecutionDataState::MAINTENANCE_ROWS) + Some(ExecutionDataState::INGESTION_ROWS) ); assert_eq!( assignment.data_state_of(raw), - Some(ExecutionDataState::MAINTENANCE_ROWS) + Some(ExecutionDataState::INGESTION_ROWS) ); } // ── rejection, capability, statistics ─────────────────────────────────── -/// A maintained summary above a query-time readout is a typed plan-time -/// error, both for the construction path and for a hand-built plan. +/// Summary construction can consume query-time values without pretending +/// they are available to an ingestion-time consumer. #[test] -fn readout_under_maintenance_is_rejected_at_construction() { +fn summary_construction_follows_its_value_input_phase() { let root = agg(vec![0], AggIntent::Max { col: None }, fine_quantile()); - // A read-time ValueOperation can never be placed under a SummaryAgg: compose a - // read-time operation, then try to maintain a summary over it. let space = plan(vec![("q", Rc::clone(&root))], &StatsModel); let post = space .global_selection(&StatsModel) @@ -699,12 +696,9 @@ fn readout_under_maintenance_is_rejected_at_construction() { }, guarantee: None, }); - assert!(matches!( - validate_execution_data_states(&illegal), - Err(ExecutionDataStateError::ReadoutUnderMaintenance { .. }) - )); - let err: RealizationError = validate_execution_data_states(&illegal).unwrap_err().into(); - assert!(matches!(err, RealizationError::ExecutionDataState(_))); + let state = asap_types::post_asap::produced_data_state(&illegal.expr).unwrap(); + assert_eq!(state.timing, ExecutionTiming::QueryTime); + asap_types::post_asap::validate_execution_data_states_at(&illegal, state).unwrap(); } #[test] @@ -742,7 +736,7 @@ fn missing_cost_statistics_preserve_the_conservative_keep_pre_asap() { .unwrap() .candidates .iter() - .any(|c| c.provenance == ReplacementProvenance::ValueOperationAtReadTime)); + .any(|c| c.provenance == ReplacementProvenance::ValueOperationAtQueryTime)); let selection = space.global_selection(&DefaultCostModel); let selected = selection.for_target(&root).unwrap(); assert!(selected.composition.is_none()); @@ -774,7 +768,7 @@ fn dag_export_carries_explicit_stage_and_plain_schema_for_a_composed_plan() { let graph = dag_export::export_summary(&composed); let node = &graph.nodes[graph.root as usize]; assert_eq!(node.kind, "ValueOperation"); - assert_eq!(node.detail["timing"], "read_time"); + assert_eq!(node.detail["timing"], "query_time"); assert!(node.detail["operation"] .as_str() .unwrap() @@ -806,7 +800,7 @@ fn promql_max_by_zone_over_quantile_over_time_composes() { let selected = selection.for_target(root).unwrap(); assert_eq!( selected.chosen.map(|c| c.provenance), - Some(ReplacementProvenance::ValueOperationAtReadTime), + Some(ReplacementProvenance::ValueOperationAtQueryTime), "{:?}", space .candidates_for_target(root) @@ -820,7 +814,7 @@ fn promql_max_by_zone_over_quantile_over_time_composes() { assert!(matches!( composed.expr, SummaryExpr::ValueOperation { - timing: ExecutionTiming::ReadTime, + timing: ExecutionTiming::QueryTime, .. } )); diff --git a/crates/integration-tests/tests/promql_to_post_asap.rs b/crates/integration-tests/tests/promql_to_post_asap.rs index c71fc87f..5f763fe4 100644 --- a/crates/integration-tests/tests/promql_to_post_asap.rs +++ b/crates/integration-tests/tests/promql_to_post_asap.rs @@ -20,10 +20,9 @@ use asap_aware_mapping::{ }; use asap_integration_tests::fixtures::lower_promql; use asap_types::post_asap::{ - compile_executable_dag, CandidateCompleteness, CompositionOperator, EdgeRole, EntityIdentity, - ExactKind, ExactParams, GroupingStrategy, NonNegativeWeightProof, SketchAlgorithm, SketchKind, - SketchParams, SketchQuery, SummaryExpr, SummaryFamilyType, SummaryInputExpr, SummaryNode, - SummarySchema, SummaryUpdate, ValueOperation, WeightDomain, + compile_executable_dag, CompositionOperator, EntityIdentity, ExactKind, ExactParams, + GroupingStrategy, SketchAlgorithm, SketchKind, SketchParams, SketchQuery, SummaryExpr, + SummaryFamilyType, SummaryInputExpr, SummaryNode, SummarySchema, SummaryUpdate, ValueOperation, }; use asap_types::pre_asap::expr_ir::ColumnRef; use asap_types::pre_asap::query_expr::{QueryExpr, Reduction}; @@ -91,7 +90,7 @@ fn value_ranked_topk_preserves_summary_children_in_post_asap_dag() { ] { let root = lower_search_and_materialize(query); let SummaryExpr::ValueOperation { - operation: ValueOperation::Limit { n, offset }, + operation: ValueOperation::Limit { n, offset, .. }, child: sort, .. } = &root.expr @@ -107,11 +106,23 @@ fn value_ranked_topk_preserves_summary_children_in_post_asap_dag() { else { panic!("expected query-time Sort under Limit for {query}"); }; - assert!( - matches!(child.expr, SummaryExpr::SummaryAgg { .. }), - "the materializable child must remain visible for {query}: {:?}", - child.expr - ); + let SummaryExpr::ValueOperation { + operation: ValueOperation::FinalizeExactAccumulator, + child: state, + .. + } = &child.expr + else { + panic!( + "Sort must consume finalized values for {query}: {:?}", + child.expr + ); + }; + assert!(matches!(state.expr, SummaryExpr::SummaryAgg { .. })); + assert!(child + .schema + .fields + .iter() + .all(|field| matches!(field.dtype, SummaryFamilyType::Plain(_)))); } } @@ -197,7 +208,9 @@ fn value_ranked_topk_over_binary_ratio_finalizes_both_summary_operands() { let query = "topk(1, sum by(job)(increase(a[6h])) / sum by(job)(increase(b[6h])))"; let root = lower_search_and_materialize(query); let SummaryExpr::ValueOperation { - operation: ValueOperation::Limit { n: 1, offset: 0 }, + operation: ValueOperation::Limit { + n: 1, offset: 0, .. + }, child: sort, .. } = &root.expr @@ -234,6 +247,10 @@ fn value_ranked_topk_over_binary_ratio_finalizes_both_summary_operands() { struct SeparatedTopK; impl AccuracyEvidenceProvider for SeparatedTopK { + fn topk_max_distinct_items(&self, _: &QueryExpr) -> Option { + Some(1000) + } + fn propagation_stats( &self, op: &CompositionOperator, @@ -251,15 +268,163 @@ impl AccuracyEvidenceProvider for SeparatedTopK { } } +// Rate-weighted summaries must consume finalized rates, never raw counter deltas. +#[test] +fn grouped_rate_topk_consumes_finalized_rate_values() { + let root = Rc::new( + lower_promql( + "topk by(job)(2, sum by(service, job)(rate(m[1m])))", + AccuracyTarget::Epsilon(0.01), + ) + .unwrap(), + ); + let strategy = SketchAlgorithmStrategy::new_with_planning_inputs_and_evidence( + &DefaultCostModel, + &DefaultAccuracyModel, + &EqualSplitAllocator, + &SeparatedTopK, + ); + let plan = strategy + .replacements(&TargetSubDAG::new(&root)) + .into_iter() + .find_map(|candidate| match candidate.replacement { + Replacement::Summary(node) if candidate.rationale.contains("CmsWithHeap") => Some(node), + _ => None, + }) + .expect("rate-weighted CMS plan"); + let dag = compile_executable_dag(&plan).unwrap(); + assert!(!dag.nodes.iter().any(|node| matches!( + node.payload, + asap_types::post_asap::ExecutableOperatorPayload::RelationalJoin { .. } + ))); + let node = dag.nodes.iter().find(|node| matches!(&node.payload, + asap_types::post_asap::ExecutableOperatorPayload::SummaryAgg { family: SummaryFamilyType::Sketch(kind, _), .. } + if kind.algorithm() == &SketchAlgorithm::CmsWithHeap)).unwrap(); + assert_eq!( + node.output_state.timing, + asap_types::post_asap::ExecutionTiming::QueryTime + ); + let asap_types::post_asap::ExecutableOperatorPayload::SummaryAgg { input, .. } = &node.payload + else { + unreachable!() + }; + assert_eq!( + input.weight, + SummaryInputExpr::Column(ColumnRef::SampleValue) + ); + assert_eq!( + input.item, + Some(SummaryInputExpr::Column(ColumnRef::Named("service".into()))) + ); +} + +// Selection is adaptive: a per-key score bound alone cannot certify all returned rows. +#[test] +fn weighted_topk_keeps_candidates_with_missing_population_evidence() { + struct NoPopulationBound; + impl AccuracyEvidenceProvider for NoPopulationBound { + fn propagation_stats( + &self, + op: &CompositionOperator, + family: &SummaryFamilyType, + query: Option<&SketchQuery>, + ) -> PropagationStats { + SeparatedTopK.propagation_stats(op, family, query) + } + } + let root = Rc::new( + lower_promql( + "topk by(job)(2, sum by(service, job)(rate(m[1m])))", + AccuracyTarget::Epsilon(0.01), + ) + .unwrap(), + ); + let strategy = SketchAlgorithmStrategy::new_with_planning_inputs_and_evidence( + &DefaultCostModel, + &DefaultAccuracyModel, + &EqualSplitAllocator, + &NoPopulationBound, + ); + let candidates = strategy.replacements(&TargetSubDAG::new(&root)); + assert!(candidates + .iter() + .any(|candidate| candidate.rationale.contains("CmsWithHeap") + && candidate.has_missing_accuracy_evidence())); +} + +// Unknown requirements must survive physical export for deployment to inspect. #[test] -fn counter_weighted_topk_uses_candidates_only_for_membership_and_exact_values_for_rerank() { - for (query, expected_k) in [ - ("topk(2, sum by(job)(rate(m[1m])))", 2), - ("topk(3, sum by(job)(rate(cpu_seconds_total[1h])))", 3), - ("topk(3, sum by(job)(increase(requests_total[6h])))", 3), +fn weighted_topk_exports_symbolic_evidence_requirements() { + let root = Rc::new( + lower_promql( + "topk by(job)(2, sum by(service, job)(rate(m[1m])))", + AccuracyTarget::EpsilonDelta { + epsilon: 0.01, + delta: 0.01, + }, + ) + .unwrap(), + ); + let candidates = + SketchAlgorithmStrategy::default_cost_model().replacements(&TargetSubDAG::new(&root)); + let candidate = candidates + .iter() + .find(|candidate| candidate.rationale.contains("CmsWithHeap")) + .unwrap(); + assert!(candidate.has_missing_accuracy_evidence()); + let Replacement::Summary(node) = &candidate.replacement else { + panic!("summary candidate") + }; + let dag = compile_executable_dag(node).unwrap(); + let exported = serde_json::to_string(&dag).unwrap(); + assert!(exported.contains("topk_max_distinct_items")); + assert!(exported.contains("topk_membership_margin")); + assert!(node.guarantee.as_ref().unwrap().has_unknown()); +} + +// Supplied invalid facts are distinct from absent evidence. +#[test] +fn weighted_topk_rejects_invalid_population_evidence() { + struct InvalidPopulation; + impl AccuracyEvidenceProvider for InvalidPopulation { + fn topk_max_distinct_items(&self, _: &QueryExpr) -> Option { + Some(0) + } + } + let root = Rc::new( + lower_promql( + "topk by(job)(2, sum by(service, job)(rate(m[1m])))", + AccuracyTarget::Epsilon(0.01), + ) + .unwrap(), + ); + let strategy = SketchAlgorithmStrategy::new_with_planning_inputs_and_evidence( + &DefaultCostModel, + &DefaultAccuracyModel, + &EqualSplitAllocator, + &InvalidPopulation, + ); + assert!(strategy.replacements(&TargetSubDAG::new(&root)).is_empty()); +} + +// The summary's estimate is projected back to logical service/job score rows. +#[test] +fn rate_and_increase_topk_use_summary_scores_and_grouped_limits() { + for query in [ + "topk(2, sum by(job)(rate(m[1m])))", + "topk by(job)(2, sum by(service, job)(rate(m[1m])))", + "topk(2, sum by(job)(increase(m[6h])))", ] { - let root = - Rc::new(lower_promql(query, AccuracyTarget::Epsilon(0.01)).expect("lowering failed")); + let root = Rc::new( + lower_promql( + query, + AccuracyTarget::EpsilonDelta { + epsilon: 0.01, + delta: 0.01, + }, + ) + .unwrap(), + ); let strategy = SketchAlgorithmStrategy::new_with_planning_inputs_and_evidence( &DefaultCostModel, &DefaultAccuracyModel, @@ -270,109 +435,91 @@ fn counter_weighted_topk_uses_candidates_only_for_membership_and_exact_values_fo .replacements(&TargetSubDAG::new(&root)) .into_iter() .find_map(|candidate| match candidate.replacement { - Replacement::Summary(node) - if candidate.rationale.contains("CmsWithHeap") - && matches!(node.expr, SummaryExpr::CandidateTopK { .. }) => - { + Replacement::Summary(node) if candidate.rationale.contains("CmsWithHeap") => { Some(node) } _ => None, }) - .unwrap_or_else(|| panic!("missing CandidateTopK for {query}")); - let SummaryExpr::CandidateTopK { - candidates, - values, - k, - completeness: CandidateCompleteness::Certified { .. }, + .expect("weighted summary"); + let SummaryExpr::ValueOperation { + child: sorted, + operation: + ValueOperation::Limit { + n: 2, + offset: 0, + partition_by, + }, .. } = &plan.expr else { - panic!("unexpected candidate plan for {query}: {:?}", plan.expr) + panic!("grouped limit") }; - assert_eq!(*k, expected_k); - let SummaryExpr::SummaryEstimate { summary_input, .. } = &candidates.expr else { - panic!("candidate membership must be a summary readout") + let SummaryExpr::ValueOperation { + child: projected, + operation: + ValueOperation::Sort { + partition_by: sort_groups, + .. + }, + .. + } = &sorted.expr + else { + panic!("grouped sort") }; + assert_eq!(partition_by, sort_groups); + assert_eq!(partition_by.len(), usize::from(query.contains("topk by"))); + let SummaryExpr::ValueOperation { + child: readout, + operation: ValueOperation::Project { .. }, + .. + } = &projected.expr + else { + panic!("logical output projection") + }; + let SummaryExpr::SummaryEstimate { + summary_input, + query: SketchQuery::TopK { k }, + } = &readout.expr + else { + panic!("heap readout") + }; + assert!(*k > 2, "candidate capacity is independent of output count"); let SummaryExpr::SummaryAgg { - child, - family, + child: rates, input, - reduction, .. } = &summary_input.expr else { - panic!("candidate membership must read a summary aggregate") + panic!("weighted summary") }; - assert!(matches!(family, SummaryFamilyType::Sketch(kind, _) - if kind.algorithm() == &SketchAlgorithm::CmsWithHeap)); - let SummaryFamilyType::Sketch(kind, _) = family else { - unreachable!() - }; - assert!( - asap_aware_mapping::replacement::sketch_state_bytes(kind.params()) - .is_some_and(|bytes| bytes - <= asap_aware_mapping::replacement::DEFAULT_MAX_SKETCH_STATE_BYTES) - ); - assert!(matches!(reduction, Reduction::Reduce(keys) if keys.is_empty())); - assert_eq!(summary_input.schema.fields.len(), 1); assert_eq!( - input.weight_domain, - WeightDomain::NonNegative { - proof: NonNegativeWeightProof::ResetAwareCounterDerivative, - } - ); - assert!(matches!( input.weight, - SummaryInputExpr::ResetAwareCounterDelta { - value: ColumnRef::SampleValue, - series: EntityIdentity::PromqlLabelSet { .. }, - } - )); - let executable = compile_executable_dag(&plan).expect("typed executable DAG"); - assert!(executable.nodes.iter().any(|node| matches!( - &node.payload, - asap_types::post_asap::ExecutableOperatorPayload::CandidateTopK { - k, - grouping, - completeness: CandidateCompleteness::Certified { .. }, - } if *k == expected_k as u64 && grouping.is_empty() && !grouping.is_without() - ))); - assert!(executable.nodes.iter().any(|node| matches!( - &node.payload, - asap_types::post_asap::ExecutableOperatorPayload::SummaryAgg { - input: SummaryUpdate { - weight: SummaryInputExpr::ResetAwareCounterDelta { .. }, - .. - }, - .. - } - ))); - assert!( - executable.edges.iter().all(|edge| edge.grouping - != asap_types::post_asap::GroupingEdgeCompatibility::Incompatible), - "unexpected incompatible edge: {:#?}", - executable.edges - ); - assert!(executable - .edges - .iter() - .any(|edge| edge.role == EdgeRole::CandidateMembership)); - assert!(executable - .edges - .iter() - .any(|edge| edge.role == EdgeRole::AuthoritativeValues)); - assert!( - !matches!(child.expr, SummaryExpr::SummaryAgg { .. }), - "membership materialization must bind ingest rows, not another summary" + SummaryInputExpr::Column(ColumnRef::SampleValue) ); - assert!(values.guarantee.as_ref().is_some_and(|g| g.is_exact())); assert!(matches!( - values.expr, + rates.expr, SummaryExpr::ValueOperation { operation: ValueOperation::FinalizeExactAccumulator, .. } )); + let dag = compile_executable_dag(&plan).unwrap(); + for phase in [ + asap_types::post_asap::ExecutionTiming::IngestionTime, + asap_types::post_asap::ExecutionTiming::QueryTime, + ] { + let phases = dag.nodes.iter().map(|node| (node.id, phase)).collect(); + let placed = dag.with_execution_phases(&phases).unwrap(); + assert!(placed + .nodes + .iter() + .all(|node| node.output_state.timing == phase)); + } + let guarantee = plan.guarantee.as_ref().unwrap(); + assert!(guarantee.failure_probability.evaluate().unwrap() <= 0.01); + assert!(guarantee.provenance.iter().any(|source| matches!(source, + asap_types::post_asap::GuaranteeSource::ChildGuarantee { guarantee, .. } + if guarantee.metric == asap_types::post_asap::ErrorMetric::Frequency))); } } @@ -585,7 +732,7 @@ fn planner_only_e2e_temporal_topk_preserves_query_update_and_readout_contract() | SketchParams::CountSketchWithHeap { heap_size, .. } => *heap_size, params => panic!("expected heap-bearing Top-K parameters, got {params:?}"), }; - assert_eq!(heap_size, *k as u32); + assert_eq!(heap_size, 100u32.max(*k as u32)); assert_eq!( state_input.item.as_ref(), Some(&SummaryInputExpr::EntityIdentity( @@ -808,7 +955,7 @@ fn promql_quantile_of_rate_binds_kll_over_rate_accumulator() { let SummaryExpr::ValueOperation { child, operation: ValueOperation::FinalizeExactAccumulator, - timing: asap_types::post_asap::ExecutionTiming::MaintenanceTime, + timing: asap_types::post_asap::ExecutionTiming::IngestionTime, } = &child.expr else { panic!("rate needs a maintenance readout"); @@ -947,7 +1094,7 @@ fn promql_sum_of_count_over_time_is_composed_by_default_search() { } #[test] -fn nested_summary_explicitly_finalizes_exact_child_at_maintenance_time() { +fn nested_summary_explicitly_finalizes_exact_child_at_ingestion_time() { // Real workload selection must expose the state-to-value edge; an outer // sketch must not interpret exact accumulator bytes as input samples. let pre = Rc::new( @@ -986,7 +1133,7 @@ fn nested_summary_explicitly_finalizes_exact_child_at_maintenance_time() { )); assert_eq!( *timing, - asap_types::post_asap::ExecutionTiming::MaintenanceTime + asap_types::post_asap::ExecutionTiming::IngestionTime ); assert!(matches!( source.expr, @@ -1009,16 +1156,16 @@ fn nested_summary_explicitly_finalizes_exact_child_at_maintenance_time() { } #[test] -fn exact_binary_maintenance_has_explicit_timing_and_legacy_wire_default() { +fn physical_node_owns_phase_independently_of_binary_payload() { use asap_types::post_asap::{ExecutableOperatorPayload, ExecutionTiming}; for (query, expected) in [ ( "quantile(0.9, sum_over_time(m[1m]) + sum_over_time(n[1m]))", - ExecutionTiming::MaintenanceTime, + ExecutionTiming::IngestionTime, ), ( "sum_over_time(m[1m]) + sum_over_time(n[1m])", - ExecutionTiming::ReadTime, + ExecutionTiming::QueryTime, ), ] { let input = lower_promql(query, AccuracyTarget::Epsilon(0.05)).unwrap(); @@ -1029,23 +1176,19 @@ fn exact_binary_maintenance_has_explicit_timing_and_legacy_wire_default() { .unwrap() .unwrap(); let dag = compile_executable_dag(&plan).unwrap(); - let payload = dag + let node = dag .nodes .iter() - .find_map(|node| { - matches!(node.payload, ExecutableOperatorPayload::Binary { .. }) - .then_some(&node.payload) - }) + .find(|node| matches!(node.payload, ExecutableOperatorPayload::Binary { .. })) .unwrap(); - assert!( - matches!(payload, ExecutableOperatorPayload::Binary { timing, .. } if *timing == expected) - ); - let wire = serde_json::to_value(payload).unwrap(); - if expected == ExecutionTiming::ReadTime { - assert!(wire.get("timing").is_none()); - } + assert_eq!(node.output_state.timing, expected); + let wire = serde_json::to_value(&node.payload).unwrap(); + assert!(wire.get("timing").is_none()); + let mut obsolete = wire.clone(); + obsolete["timing"] = serde_json::json!(expected.as_str()); + assert!(serde_json::from_value::(obsolete).is_err()); let restored: ExecutableOperatorPayload = serde_json::from_value(wire).unwrap(); - assert_eq!(&restored, payload); + assert_eq!(restored, node.payload); } } diff --git a/crates/integration-tests/tests/sql_to_post_asap.rs b/crates/integration-tests/tests/sql_to_post_asap.rs index ab4f9375..28de387b 100644 --- a/crates/integration-tests/tests/sql_to_post_asap.rs +++ b/crates/integration-tests/tests/sql_to_post_asap.rs @@ -309,6 +309,7 @@ async fn sql_join_recursively_binds_both_temporal_aggregate_children() { right, kind, pred, + pruning: None, } = &join.expr else { panic!("expected read-time relational join, got {:?}", join.expr); @@ -340,6 +341,14 @@ async fn sql_join_recursively_binds_both_temporal_aggregate_children() { else { panic!("derived table Project was not retained: {:?}", child.expr); }; + let SummaryExpr::ValueOperation { + child: aggregate, + operation: ValueOperation::FinalizeExactAccumulator, + .. + } = &aggregate.expr + else { + panic!("derived table Project must consume finalized exact values"); + }; assert!(matches!( aggregate.expr, SummaryExpr::SummaryAgg { @@ -442,7 +451,7 @@ async fn sql_relational_parents_retain_summary_bound_aggregate() { asap_types::post_asap::ValueOperation::Project { .. } => saw_project = true, asap_types::post_asap::ValueOperation::Filter { .. } => saw_filter = true, asap_types::post_asap::ValueOperation::Sort { .. } => saw_sort = true, - asap_types::post_asap::ValueOperation::Limit { n, offset } => { + asap_types::post_asap::ValueOperation::Limit { n, offset, .. } => { assert_eq!((*n, *offset), (5, 0)); saw_limit = true; } diff --git a/crates/types/src/dag_export.rs b/crates/types/src/dag_export.rs index 3a983fba..70c04846 100644 --- a/crates/types/src/dag_export.rs +++ b/crates/types/src/dag_export.rs @@ -475,7 +475,7 @@ macro_rules! define_summary_kind_tags { define_summary_kind_tags! { SummaryExpr::BinaryOp { .. } => "SummaryBinaryOp", - SummaryExpr::CandidateTopK { .. } => "CandidateTopK", + SummaryExpr::ValueOperation { .. } => "ValueOperation", SummaryExpr::RelationalJoin { .. } => "RelationalJoin", SummaryExpr::SummaryAgg { .. } => "SummaryAgg", @@ -500,16 +500,7 @@ fn summary_shape(expr: &SummaryExpr) -> (&'static str, String, serde_json::Value }); (kind, label, detail) } - SummaryExpr::CandidateTopK { - k, - grouping, - completeness, - .. - } => ( - kind, - format!("CandidateTopK(k={k})"), - serde_json::json!({ "k": k, "grouping": grouping, "completeness": completeness }), - ), + SummaryExpr::ValueOperation { operation, timing, .. } => ( @@ -565,7 +556,7 @@ fn summary_shape(expr: &SummaryExpr) -> (&'static str, String, serde_json::Value let detail = serde_json::json!({ "query": format!("{query:?}") }); (kind, label, detail) } - SummaryExpr::SummaryMerge { children } => { + SummaryExpr::SummaryMerge { children, .. } => { let label = format!("SummaryMerge({} children)", children.len()); (kind, label, serde_json::json!({})) } @@ -581,9 +572,7 @@ fn summary_children(expr: &SummaryExpr) -> Vec<&Rc> { match expr { SummaryExpr::KeepPreAsap(_) => vec![], SummaryExpr::BinaryOp { lhs, rhs, .. } => vec![lhs, rhs], - SummaryExpr::CandidateTopK { - candidates, values, .. - } => vec![candidates, values], + SummaryExpr::ValueOperation { child, .. } => vec![child], SummaryExpr::RelationalJoin { left, right, .. } => vec![left, right], SummaryExpr::SummaryAgg { child, .. } => vec![child], @@ -591,7 +580,7 @@ fn summary_children(expr: &SummaryExpr) -> Vec<&Rc> { SummaryExpr::SummarySubtract { left, right } => vec![left, right], SummaryExpr::SummaryDelete { summary_input, .. } => vec![summary_input], SummaryExpr::SummaryEstimate { summary_input, .. } => vec![summary_input], - SummaryExpr::SummaryMerge { children } => children.iter().collect(), + SummaryExpr::SummaryMerge { children, .. } => children.iter().collect(), } } diff --git a/crates/types/src/post_asap/cse.rs b/crates/types/src/post_asap/cse.rs index a8077f74..6d1d0dc9 100644 --- a/crates/types/src/post_asap/cse.rs +++ b/crates/types/src/post_asap/cse.rs @@ -40,22 +40,6 @@ fn same_node(left: &SummaryNode, right: &SummaryNode) -> bool { timing: bt, }, ) => Rc::ptr_eq(al, bl) && Rc::ptr_eq(ar, br) && ao == bo && at == bt, - ( - CandidateTopK { - candidates: ac, - values: av, - k: ak, - grouping: ag, - completeness: ax, - }, - CandidateTopK { - candidates: bc, - values: bv, - k: bk, - grouping: bg, - completeness: bx, - }, - ) => Rc::ptr_eq(ac, bc) && Rc::ptr_eq(av, bv) && ak == bk && ag == bg && same_value(ax, bx), ( ValueOperation { child: ac, @@ -74,14 +58,22 @@ fn same_node(left: &SummaryNode, right: &SummaryNode) -> bool { right: ar, kind: ak, pred: ap, + pruning: ax, }, RelationalJoin { left: bl, right: br, kind: bk, pred: bp, + pruning: bx, }, - ) => Rc::ptr_eq(al, bl) && Rc::ptr_eq(ar, br) && ak == bk && same_value(ap, bp), + ) => { + Rc::ptr_eq(al, bl) + && Rc::ptr_eq(ar, br) + && ak == bk + && same_value(ap, bp) + && same_value(ax, bx) + } ( SummaryAgg { child: ac, @@ -142,14 +134,20 @@ fn same_node(left: &SummaryNode, right: &SummaryNode) -> bool { key: bk, }, ) => Rc::ptr_eq(ai, bi) && ak == bk, - (SummaryMerge { children: a }, SummaryMerge { children: b }) => { - a.len() == b.len() && a.iter().zip(b).all(|(a, b)| Rc::ptr_eq(a, b)) - } + ( + SummaryMerge { + children: a, + timing: at, + }, + SummaryMerge { + children: b, + timing: bt, + }, + ) => at == bt && a.len() == b.len() && a.iter().zip(b).all(|(a, b)| Rc::ptr_eq(a, b)), // Keep this exhaustive on the left: new variants require a sharing rule. ( KeepPreAsap(_) | BinaryOp { .. } - | CandidateTopK { .. } | ValueOperation { .. } | RelationalJoin { .. } | SummaryAgg { .. } @@ -190,12 +188,7 @@ pub fn share_common_summary_subtrees( *lhs = visit(lhs, seen, pool); *rhs = visit(rhs, seen, pool); } - SummaryExpr::CandidateTopK { - candidates, values, .. - } => { - *candidates = visit(candidates, seen, pool); - *values = visit(values, seen, pool); - } + SummaryExpr::ValueOperation { child, .. } => *child = visit(child, seen, pool), SummaryExpr::RelationalJoin { left, right, .. } => { *left = visit(left, seen, pool); @@ -213,7 +206,7 @@ pub fn share_common_summary_subtrees( | SummaryExpr::SummaryDelete { summary_input, .. } => { *summary_input = visit(summary_input, seen, pool); } - SummaryExpr::SummaryMerge { children } => { + SummaryExpr::SummaryMerge { children, .. } => { for child in children { *child = visit(child, seen, pool); } @@ -271,6 +264,7 @@ mod tests { fn shares_children_across_distinct_roots() { let merge = Rc::new(SummaryNode { expr: SummaryExpr::SummaryMerge { + timing: crate::post_asap::ExecutionTiming::IngestionTime, children: vec![leaf(1.0), leaf(2.0)], }, schema: SummarySchema { @@ -280,7 +274,7 @@ mod tests { guarantee: None, }); let roots = share_common_summary_subtrees(vec![(0, leaf(1.0)), (1, merge)]); - let SummaryExpr::SummaryMerge { children } = &roots[1].1.expr else { + let SummaryExpr::SummaryMerge { children, .. } = &roots[1].1.expr else { panic!() }; assert!(Rc::ptr_eq(&roots[0].1, &children[0])); @@ -413,7 +407,7 @@ mod tests { for _ in 0..24 { current = Rc::new(SummaryNode { expr: SummaryExpr::BinaryOp { - timing: super::super::ExecutionTiming::ReadTime, + timing: super::super::ExecutionTiming::QueryTime, lhs: Rc::clone(¤t), rhs: current, operator: super::super::BinaryOperator { diff --git a/crates/types/src/post_asap/executable_dag.rs b/crates/types/src/post_asap/executable_dag.rs index 28b7aa78..c156aa97 100644 --- a/crates/types/src/post_asap/executable_dag.rs +++ b/crates/types/src/post_asap/executable_dag.rs @@ -11,18 +11,16 @@ use super::{ BinaryOperator, CandidateCompleteness, ExecutionTiming, GroupingStrategy, SketchQuery, SummaryFamilyType, SummaryUpdate, ValueOperation, }; -use crate::pre_asap::{ColumnRef, GroupKeys, JoinKind, Predicate, QueryExpr, Reduction}; +use crate::pre_asap::{ColumnRef, JoinKind, Predicate, QueryExpr, Reduction}; use thiserror::Error; -pub const POST_ASAP_DAG_WIRE_VERSION: u32 = 2; +pub const POST_ASAP_DAG_WIRE_VERSION: u32 = 5; #[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)] pub enum EdgeRole { Input, Left, Right, - CandidateMembership, - AuthoritativeValues, } #[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)] @@ -50,30 +48,21 @@ pub enum WindowEdgeCompatibility { pub struct PostAsapNodeId(pub u32); #[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)] -#[serde(tag = "kind", rename_all = "snake_case")] +#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] pub enum ExecutableOperatorPayload { Fallback { expression: QueryExpr, }, Binary { - #[serde(default, skip_serializing_if = "ExecutionTiming::is_read_time")] - timing: ExecutionTiming, operator: BinaryOperator, }, - CandidateTopK { - /// Fixed-width transport value; runtimes validate conversion to their - /// local collection index type at installation. - k: u64, - grouping: GroupKeys, - completeness: CandidateCompleteness, - }, Value { operation: ValueOperation, - timing: ExecutionTiming, }, RelationalJoin { join_kind: JoinKind, pred: Predicate, + pruning: Option, }, SummaryAgg { family: SummaryFamilyType, @@ -101,6 +90,7 @@ pub struct ExecutableDagNode { pub id: PostAsapNodeId, /// The payload variant is the sole operator identity (`payload.kind` in JSON). pub payload: ExecutableOperatorPayload, + /// Phase is a placement choice for every operator, independent of payload kind. pub output_state: ExecutionDataState, pub output_schema: SummarySchema, pub guarantee: Option, @@ -130,8 +120,7 @@ pub struct ExecutableDag { /// Versioned transport envelope for a post-ASAP semantic DAG. /// -/// `ExecutableDag` remains serializable as a legacy in-process adapter. New -/// process boundaries should exchange this envelope and call [`Self::validate`]. +/// Process boundaries exchange this envelope and call [`Self::validate`]. #[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)] #[serde(deny_unknown_fields)] pub struct PostAsapDagDocument { @@ -141,6 +130,13 @@ pub struct PostAsapDagDocument { #[derive(Debug, Clone, PartialEq, Eq, Error)] pub enum ExecutableDagValidationError { + #[error("phase assignment must name every DAG node exactly once")] + IncompletePhaseAssignment, + #[error("ingestion node {consumer:?} depends on query node {producer:?}")] + QueryDependencyInIngestion { + producer: PostAsapNodeId, + consumer: PostAsapNodeId, + }, #[error("unsupported post-ASAP DAG schema version {0}")] UnsupportedVersion(u32), #[error("duplicate post-ASAP node id {0:?}")] @@ -190,6 +186,31 @@ impl PostAsapDagDocument { } impl ExecutableDag { + /// Assign execution phases without changing operator semantics. Phase choices + /// do not prove deployment support: callers must bind concrete implementations + /// and storage boundaries before installing this plan. + pub fn with_execution_phases( + &self, + phases: &std::collections::BTreeMap, + ) -> Result { + self.validate()?; + if phases.len() != self.nodes.len() + || self.nodes.iter().any(|node| !phases.contains_key(&node.id)) + { + return Err(ExecutableDagValidationError::IncompletePhaseAssignment); + } + let mut dag = self.clone(); + for node in &mut dag.nodes { + node.output_state.timing = phases[&node.id]; + } + let states: HashMap<_, _> = dag.nodes.iter().map(|n| (n.id, n.output_state)).collect(); + for edge in &mut dag.edges { + edge.data_state = states[&edge.producer]; + } + dag.validate()?; + Ok(dag) + } + pub fn validate(&self) -> Result<(), ExecutableDagValidationError> { use std::collections::{HashMap, HashSet}; let mut nodes = HashMap::new(); @@ -234,6 +255,14 @@ impl ExecutableDag { edge.consumer, )); } + if producer.output_state.timing == ExecutionTiming::QueryTime + && nodes[&edge.consumer].output_state.timing == ExecutionTiming::IngestionTime + { + return Err(ExecutableDagValidationError::QueryDependencyInIngestion { + producer: edge.producer, + consumer: edge.consumer, + }); + } if edge.intermediate_schema != producer.output_schema { return Err(ExecutableDagValidationError::EdgeSchemaMismatch { producer: edge.producer, @@ -362,12 +391,7 @@ pub fn compile_executable_dag_with_node_ids( SummaryExpr::BinaryOp { lhs, rhs, .. } => { vec![(lhs, EdgeRole::Left), (rhs, EdgeRole::Right)] } - SummaryExpr::CandidateTopK { - candidates, values, .. - } => vec![ - (candidates, EdgeRole::CandidateMembership), - (values, EdgeRole::AuthoritativeValues), - ], + SummaryExpr::ValueOperation { child, .. } | SummaryExpr::SummaryAgg { child, .. } => { vec![(child, EdgeRole::Input)] } @@ -384,7 +408,7 @@ pub fn compile_executable_dag_with_node_ids( | SummaryExpr::SummaryEstimate { summary_input, .. } => { vec![(summary_input, EdgeRole::Input)] } - SummaryExpr::SummaryMerge { children } => { + SummaryExpr::SummaryMerge { children, .. } => { children.iter().map(|c| (c, EdgeRole::Input)).collect() } }; @@ -400,34 +424,23 @@ pub fn compile_executable_dag_with_node_ids( SummaryExpr::KeepPreAsap(expression) => ExecutableOperatorPayload::Fallback { expression: (**expression).clone(), }, - SummaryExpr::BinaryOp { - operator, timing, .. - } => ExecutableOperatorPayload::Binary { - timing: *timing, + SummaryExpr::BinaryOp { operator, .. } => ExecutableOperatorPayload::Binary { operator: operator.clone(), }, - SummaryExpr::CandidateTopK { - k, - grouping, - completeness, - .. - } => ExecutableOperatorPayload::CandidateTopK { - k: u64::try_from(*k).expect("usize always fits into the u64 wire count"), - grouping: grouping.clone(), - completeness: completeness.clone(), - }, - SummaryExpr::ValueOperation { - operation, timing, .. - } => ExecutableOperatorPayload::Value { + + SummaryExpr::ValueOperation { operation, .. } => ExecutableOperatorPayload::Value { operation: operation.clone(), - timing: *timing, }, - SummaryExpr::RelationalJoin { kind, pred, .. } => { - ExecutableOperatorPayload::RelationalJoin { - join_kind: kind.clone(), - pred: pred.clone(), - } - } + SummaryExpr::RelationalJoin { + kind, + pred, + pruning, + .. + } => ExecutableOperatorPayload::RelationalJoin { + join_kind: kind.clone(), + pred: pred.clone(), + pruning: pruning.clone(), + }, SummaryExpr::SummaryAgg { family, input, @@ -468,8 +481,8 @@ pub fn compile_executable_dag_with_node_ids( ids.insert(Rc::as_ptr(node), id); for (producer, child, role) in child_ids { let maintenance_dependency = nodes[producer.0 as usize].output_state.timing - == ExecutionTiming::MaintenanceTime - && nodes[id.0 as usize].output_state.timing == ExecutionTiming::MaintenanceTime; + == ExecutionTiming::IngestionTime + && nodes[id.0 as usize].output_state.timing == ExecutionTiming::IngestionTime; let grouping = match (&child.expr, &node.expr) { ( SummaryExpr::SummaryAgg { @@ -562,6 +575,159 @@ mod tests { use crate::pre_asap::{ColumnRef, DataType, QueryExpr, Reduction, Source}; use std::collections::BTreeMap; + #[test] + fn every_physical_payload_can_be_assigned_either_phase() { + use crate::post_asap::DataPrimitive; + use crate::pre_asap::{ArithmeticOpKind, BinaryOpKind, JoinKind, Predicate, ScalarValue}; + let family = SummaryFamilyType::ExactAggregate(ExactKind::Sum, ExactParams::Sum); + let predicate = Predicate(Rc::new(QueryExpr::Literal(ScalarValue::Boolean(true)))); + let payloads = vec![ + ExecutableOperatorPayload::Fallback { + expression: QueryExpr::Literal(ScalarValue::Int64(1)), + }, + ExecutableOperatorPayload::Binary { + operator: BinaryOperator { + checked_relative_division: false, + checked_finite_division: false, + kind: BinaryOpKind::Arithmetic(ArithmeticOpKind::Add), + vector_match: None, + }, + }, + ExecutableOperatorPayload::Value { + operation: ValueOperation::Limit { + n: 1, + offset: 0, + partition_by: Default::default(), + }, + }, + ExecutableOperatorPayload::RelationalJoin { + join_kind: JoinKind::Semi, + pred: predicate, + pruning: None, + }, + ExecutableOperatorPayload::SummaryAgg { + family: family.clone(), + input: SummaryUpdate::column(ColumnRef::SampleValue), + reduction: Reduction::by(vec![]), + grouping: GroupingStrategy::default(), + }, + ExecutableOperatorPayload::SummaryJoin { + key: ColumnRef::SampleValue, + family: family.clone(), + }, + ExecutableOperatorPayload::SummarySubtract, + ExecutableOperatorPayload::SummaryDelete { + key: ColumnRef::SampleValue, + }, + ExecutableOperatorPayload::SummaryEstimate { + query: SketchQuery::Cardinality, + }, + ExecutableOperatorPayload::SummaryMerge, + ]; + for payload in payloads { + // This checks physical identity and placement, not kernel availability. + let primitive = match &payload { + ExecutableOperatorPayload::Fallback { .. } + | ExecutableOperatorPayload::Binary { .. } + | ExecutableOperatorPayload::Value { .. } + | ExecutableOperatorPayload::RelationalJoin { .. } + | ExecutableOperatorPayload::SummaryEstimate { .. } => DataPrimitive::Raw, + ExecutableOperatorPayload::SummaryAgg { .. } + | ExecutableOperatorPayload::SummaryJoin { .. } + | ExecutableOperatorPayload::SummarySubtract + | ExecutableOperatorPayload::SummaryDelete { .. } + | ExecutableOperatorPayload::SummaryMerge => DataPrimitive::SummaryState, + }; + let dag = ExecutableDag { + root: PostAsapNodeId(0), + edges: vec![], + nodes: vec![ExecutableDagNode { + id: PostAsapNodeId(0), + payload: payload.clone(), + output_state: ExecutionDataState { + timing: ExecutionTiming::QueryTime, + primitive, + }, + output_schema: SummarySchema { + fields: vec![SummaryField { + name: "value".into(), + dtype: family.clone(), + nullable: false, + }], + time_index: None, + }, + guarantee: None, + }], + }; + for phase in [ExecutionTiming::IngestionTime, ExecutionTiming::QueryTime] { + let placed = dag + .with_execution_phases(&BTreeMap::from([(dag.root, phase)])) + .unwrap(); + assert_eq!(placed.nodes[0].payload, payload); + assert_eq!(placed.nodes[0].output_state.timing, phase); + let wire = serde_json::to_value(&placed).unwrap(); + assert!(wire["nodes"][0]["payload"].get("timing").is_none()); + assert_eq!( + serde_json::from_value::(wire).unwrap(), + placed + ); + } + assert!(dag.with_execution_phases(&BTreeMap::new()).is_err()); + } + } + + #[test] + fn phase_assignment_updates_edges_and_rejects_query_dependencies_in_ingestion() { + use crate::pre_asap::ScalarValue; + let schema = SummarySchema { + fields: vec![], + time_index: None, + }; + let nodes = [0, 1] + .into_iter() + .map(|id| ExecutableDagNode { + id: PostAsapNodeId(id), + payload: ExecutableOperatorPayload::Fallback { + expression: QueryExpr::Literal(ScalarValue::Int64(1)), + }, + output_state: ExecutionDataState::QUERY_ROWS, + output_schema: schema.clone(), + guarantee: None, + }) + .collect(); + let dag = ExecutableDag { + nodes, + root: PostAsapNodeId(1), + edges: vec![ExecutableDagEdge { + producer: PostAsapNodeId(0), + consumer: PostAsapNodeId(1), + role: EdgeRole::Input, + intermediate_schema: schema, + data_state: ExecutionDataState::QUERY_ROWS, + grouping: GroupingEdgeCompatibility::NotApplicable, + window: WindowEdgeCompatibility::NotApplicable, + }], + }; + let placed = dag + .with_execution_phases(&BTreeMap::from([ + (PostAsapNodeId(0), ExecutionTiming::IngestionTime), + (PostAsapNodeId(1), ExecutionTiming::QueryTime), + ])) + .unwrap(); + assert_eq!( + placed.edges[0].data_state.timing, + ExecutionTiming::IngestionTime + ); + assert_eq!(dag.edges[0].data_state.timing, ExecutionTiming::QueryTime); + assert!(matches!( + dag.with_execution_phases(&BTreeMap::from([ + (PostAsapNodeId(0), ExecutionTiming::QueryTime), + (PostAsapNodeId(1), ExecutionTiming::IngestionTime), + ])), + Err(ExecutableDagValidationError::QueryDependencyInIngestion { .. }) + )); + } + #[test] fn exports_summary_over_summary_as_typed_precompute_edges() { let scan = Rc::new(QueryExpr::Scan { @@ -608,7 +774,7 @@ mod tests { expr: SummaryExpr::ValueOperation { child: outer, operation: ValueOperation::FinalizeExactAccumulator, - timing: ExecutionTiming::ReadTime, + timing: ExecutionTiming::QueryTime, }, schema: SummarySchema { fields: vec![SummaryField { @@ -631,21 +797,18 @@ mod tests { assert_eq!(dag.root, PostAsapNodeId(3)); assert_eq!( dag.nodes[1].output_state, - ExecutionDataState::MAINTENANCE_SUMMARY + ExecutionDataState::INGESTION_SUMMARY ); assert_eq!( dag.nodes[2].output_state, - ExecutionDataState::MAINTENANCE_SUMMARY + ExecutionDataState::INGESTION_SUMMARY ); let dependency = dag .edges .iter() .find(|e| e.producer == PostAsapNodeId(1) && e.consumer == PostAsapNodeId(2)) .unwrap(); - assert_eq!( - dependency.data_state, - ExecutionDataState::MAINTENANCE_SUMMARY - ); + assert_eq!(dependency.data_state, ExecutionDataState::INGESTION_SUMMARY); assert_eq!(dependency.grouping, GroupingEdgeCompatibility::Identical); assert_eq!( dependency.window, diff --git a/crates/types/src/post_asap/execution_data_state.rs b/crates/types/src/post_asap/execution_data_state.rs index 26b30e63..e61cafe3 100644 --- a/crates/types/src/post_asap/execution_data_state.rs +++ b/crates/types/src/post_asap/execution_data_state.rs @@ -19,12 +19,13 @@ //! //! | 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`. | -//! | `ValueOperation.child` with `MaintenanceTime` | `MAINTENANCE_ROWS`; explicit `FinalizeExactAccumulator` also accepts exact accumulator state. Produces `MAINTENANCE_ROWS`. | -//! | `ValueOperation.child` with `ReadTime` | `READ_ROWS`. Produces `READ_ROWS`. | +//! | `SummaryAgg.child` | Rows or exact accumulator state at either phase. The initial construction phase follows the input; deployment assigns final phases. | +//! | `SummaryEstimate.summary_input` | Summary state at either phase (any family). Initial readout produces `QUERY_ROWS`. | +//! | `SummaryJoin.outer/inner` | `INGESTION_ROWS` or `INGESTION_SUMMARY`; never a read-time data_state. | +//! | `SummarySubtract`/`SummaryDelete` | `INGESTION_SUMMARY`. | +//! | `SummaryMerge` | Summary state at its explicit ingestion or read timing. | +//! | `ValueOperation.child` with `IngestionTime` | `INGESTION_ROWS`; explicit `FinalizeExactAccumulator` also accepts exact accumulator state. Produces `INGESTION_ROWS`. | +//! | `ValueOperation.child` with `QueryTime` | `QUERY_ROWS`. Produces `QUERY_ROWS`. | //! //! ## `KeepPreAsap` declares its data_state through the derivation //! @@ -55,20 +56,21 @@ use crate::pre_asap::schema::{Column, Schema}; #[derive( Default, Debug, Clone, Copy, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize, )] +#[serde(rename_all = "snake_case")] pub enum ExecutionTiming { - MaintenanceTime, + IngestionTime, #[default] - ReadTime, + QueryTime, } impl ExecutionTiming { - pub fn is_read_time(&self) -> bool { - *self == Self::ReadTime + pub fn is_query_time(&self) -> bool { + *self == Self::QueryTime } pub fn as_str(self) -> &'static str { match self { - Self::MaintenanceTime => "maintenance_time", - Self::ReadTime => "read_time", + Self::IngestionTime => "ingestion_time", + Self::QueryTime => "query_time", } } } @@ -100,16 +102,16 @@ pub struct ExecutionDataState { } impl ExecutionDataState { - pub const MAINTENANCE_ROWS: Self = Self { - timing: ExecutionTiming::MaintenanceTime, + pub const INGESTION_ROWS: Self = Self { + timing: ExecutionTiming::IngestionTime, primitive: DataPrimitive::Raw, }; - pub const MAINTENANCE_SUMMARY: Self = Self { - timing: ExecutionTiming::MaintenanceTime, + pub const INGESTION_SUMMARY: Self = Self { + timing: ExecutionTiming::IngestionTime, primitive: DataPrimitive::SummaryState, }; - pub const READ_ROWS: Self = Self { - timing: ExecutionTiming::ReadTime, + pub const QUERY_ROWS: Self = Self { + timing: ExecutionTiming::QueryTime, primitive: DataPrimitive::Raw, }; } @@ -229,7 +231,9 @@ impl ExecutionDataStateAssignment { } } -/// The data_state `expr` *produces*, independent of context — `None` for +/// Initial layout proposed by semantic realization, not a restriction on physical +/// operator placement. `ExecutableDag::with_execution_phases` assigns the final +/// phase independently of payload kind. Returns `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 { @@ -239,18 +243,23 @@ pub fn produced_data_state(expr: &SummaryExpr) -> Option { timing: *timing, primitive: DataPrimitive::Raw, }, - SummaryExpr::CandidateTopK { .. } | SummaryExpr::RelationalJoin { .. } => { - ExecutionDataState::READ_ROWS - } - SummaryExpr::SummaryAgg { .. } - | SummaryExpr::SummaryJoin { .. } + SummaryExpr::RelationalJoin { .. } => ExecutionDataState::QUERY_ROWS, + SummaryExpr::SummaryAgg { child, .. } => ExecutionDataState { + timing: produced_data_state(&child.expr) + .map_or(ExecutionTiming::IngestionTime, |state| state.timing), + primitive: DataPrimitive::SummaryState, + }, + SummaryExpr::SummaryJoin { .. } | SummaryExpr::SummarySubtract { .. } - | SummaryExpr::SummaryDelete { .. } - | SummaryExpr::SummaryMerge { .. } => ExecutionDataState::MAINTENANCE_SUMMARY, - SummaryExpr::SummaryEstimate { .. } => ExecutionDataState::READ_ROWS, + | SummaryExpr::SummaryDelete { .. } => ExecutionDataState::INGESTION_SUMMARY, + SummaryExpr::SummaryMerge { timing, .. } => ExecutionDataState { + timing: *timing, + primitive: DataPrimitive::SummaryState, + }, + SummaryExpr::SummaryEstimate { .. } => ExecutionDataState::QUERY_ROWS, SummaryExpr::ValueOperation { timing, .. } => match timing { - ExecutionTiming::MaintenanceTime => ExecutionDataState::MAINTENANCE_ROWS, - ExecutionTiming::ReadTime => ExecutionDataState::READ_ROWS, + ExecutionTiming::IngestionTime => ExecutionDataState::INGESTION_ROWS, + ExecutionTiming::QueryTime => ExecutionDataState::QUERY_ROWS, }, }) } @@ -283,8 +292,8 @@ pub fn validate_execution_data_states( // 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) => { + None => ExecutionDataState::QUERY_ROWS, + Some(ExecutionDataState::INGESTION_ROWS) => { return Err(ExecutionDataStateError::MaintenanceRowsAtRoot) } Some(data_state) => data_state, @@ -336,7 +345,7 @@ fn visit( } => { if (operator.checked_relative_division && operator.checked_finite_division) || (operator.checked_relative_division || operator.checked_finite_division) - && (*timing != ExecutionTiming::ReadTime + && (*timing != ExecutionTiming::QueryTime || !matches!( operator.kind, crate::pre_asap::BinaryOpKind::Arithmetic( @@ -346,7 +355,7 @@ fn visit( { return Err(ExecutionDataStateError::InvalidCheckedDivision); } - if *timing == ExecutionTiming::MaintenanceTime { + if *timing == ExecutionTiming::IngestionTime { use crate::pre_asap::{BinaryOpKind, DataType}; if operator.vector_match.is_some() || !matches!(operator.kind, BinaryOpKind::Arithmetic(_)) @@ -388,27 +397,12 @@ fn visit( } Ok(()) } - SummaryExpr::CandidateTopK { - candidates, values, .. - } => { - for input in [candidates, values] { - let state = - produced_data_state(&input.expr).unwrap_or(ExecutionDataState::READ_ROWS); - if state != ExecutionDataState::READ_ROWS { - return Err(ExecutionDataStateError::IllegalChildDataState { - edge: "CandidateTopK input", - child: state, - }); - } - visit(input, state, assignment)?; - } - Ok(()) - } + SummaryExpr::RelationalJoin { left, right, .. } => { for input in [left, right] { let state = - produced_data_state(&input.expr).unwrap_or(ExecutionDataState::READ_ROWS); - if state != ExecutionDataState::READ_ROWS { + produced_data_state(&input.expr).unwrap_or(ExecutionDataState::QUERY_ROWS); + if state != ExecutionDataState::QUERY_ROWS { return Err(ExecutionDataStateError::IllegalChildDataState { edge: "RelationalJoin input", child: state, @@ -423,8 +417,8 @@ fn visit( child, ExecutionDataStateEdge::SummaryAggChild, |avail| match avail { - ExecutionDataState::MAINTENANCE_ROWS => Ok(()), - ExecutionDataState::MAINTENANCE_SUMMARY => { + ExecutionDataState::INGESTION_ROWS | ExecutionDataState::QUERY_ROWS => Ok(()), + state if state.primitive == DataPrimitive::SummaryState => { is_exact_accumulator_state(&child.schema) } other => Err(ExecutionDataStateError::ReadoutUnderMaintenance { @@ -439,8 +433,8 @@ fn visit( for input in [outer, inner] { let s = child_domain(input, ExecutionDataStateEdge::SummaryJoinInput, |avail| { match avail { - ExecutionDataState::MAINTENANCE_ROWS - | ExecutionDataState::MAINTENANCE_SUMMARY => Ok(()), + ExecutionDataState::INGESTION_ROWS + | ExecutionDataState::INGESTION_SUMMARY => Ok(()), other => Err(ExecutionDataStateError::ReadoutUnderMaintenance { edge: ExecutionDataStateEdge::SummaryJoinInput.describe(), child: other, @@ -462,9 +456,20 @@ fn visit( let s = state_only(summary_input, ExecutionDataStateEdge::SummaryDeleteInput)?; visit(summary_input, s, assignment) } - SummaryExpr::SummaryMerge { children } => { + SummaryExpr::SummaryMerge { children, timing } => { for input in children { - let s = state_only(input, ExecutionDataStateEdge::SummaryMergeInput)?; + let s = child_domain(input, ExecutionDataStateEdge::SummaryMergeInput, |state| { + if state.primitive == DataPrimitive::SummaryState + && (*timing == ExecutionTiming::QueryTime || state.timing == *timing) + { + Ok(()) + } else { + Err(ExecutionDataStateError::IllegalChildDataState { + edge: ExecutionDataStateEdge::SummaryMergeInput.describe(), + child: state, + }) + } + })?; visit(input, s, assignment)?; } Ok(()) @@ -480,12 +485,12 @@ fn visit( } => { let valid_population = match operation { ValueOperation::MaintainPopulation { population } => { - *timing == ExecutionTiming::MaintenanceTime + *timing == ExecutionTiming::IngestionTime && matches!(&child.expr, SummaryExpr::KeepPreAsap(input) if population.matches_input(input)) } ValueOperation::ReadPopulation { readout } => { - *timing == ExecutionTiming::ReadTime - && matches!(&child.expr, SummaryExpr::ValueOperation { operation: ValueOperation::MaintainPopulation { population }, timing: ExecutionTiming::MaintenanceTime, .. } if population.supports(readout)) + *timing == ExecutionTiming::QueryTime + && matches!(&child.expr, SummaryExpr::ValueOperation { operation: ValueOperation::MaintainPopulation { population }, timing: ExecutionTiming::IngestionTime, .. } if population.supports(readout)) } _ => true, }; @@ -493,21 +498,22 @@ fn visit( return Err(ExecutionDataStateError::InvalidMaintainedPopulation); } let required = match timing { - ExecutionTiming::MaintenanceTime => ExecutionDataState::MAINTENANCE_ROWS, - ExecutionTiming::ReadTime => ExecutionDataState::READ_ROWS, + ExecutionTiming::IngestionTime => ExecutionDataState::INGESTION_ROWS, + ExecutionTiming::QueryTime => ExecutionDataState::QUERY_ROWS, }; let s = produced_data_state(&child.expr).unwrap_or(required); - let exact_readout = (*timing == ExecutionTiming::ReadTime + let exact_readout = (*timing == ExecutionTiming::QueryTime || matches!(operation, ValueOperation::FinalizeExactAccumulator)) - && s == ExecutionDataState::MAINTENANCE_SUMMARY + && s.primitive == DataPrimitive::SummaryState + && (*timing == ExecutionTiming::QueryTime || s.timing == *timing) && is_exact_accumulator_state(&child.schema).is_ok(); let population_readout = matches!(operation, ValueOperation::ReadPopulation { .. }) - && *timing == ExecutionTiming::ReadTime + && *timing == ExecutionTiming::QueryTime && matches!( &child.expr, SummaryExpr::ValueOperation { operation: ValueOperation::MaintainPopulation { .. }, - timing: ExecutionTiming::MaintenanceTime, + timing: ExecutionTiming::IngestionTime, .. } ); @@ -537,19 +543,18 @@ pub fn assigned_child_data_state(parent: &SummaryExpr, child: &SummaryNode) -> E } match parent { SummaryExpr::ValueOperation { - timing: ExecutionTiming::ReadTime, + timing: ExecutionTiming::QueryTime, .. } | SummaryExpr::BinaryOp { - timing: ExecutionTiming::ReadTime, + timing: ExecutionTiming::QueryTime, .. - } => ExecutionDataState::READ_ROWS, + } => ExecutionDataState::QUERY_ROWS, SummaryExpr::KeepPreAsap(_) | SummaryExpr::BinaryOp { - timing: ExecutionTiming::MaintenanceTime, + timing: ExecutionTiming::IngestionTime, .. } - | SummaryExpr::CandidateTopK { .. } | SummaryExpr::RelationalJoin { .. } | SummaryExpr::SummaryAgg { .. } | SummaryExpr::SummaryJoin { .. } @@ -558,9 +563,9 @@ pub fn assigned_child_data_state(parent: &SummaryExpr, child: &SummaryNode) -> E | SummaryExpr::SummaryEstimate { .. } | SummaryExpr::SummaryMerge { .. } | SummaryExpr::ValueOperation { - timing: ExecutionTiming::MaintenanceTime, + timing: ExecutionTiming::IngestionTime, .. - } => ExecutionDataState::MAINTENANCE_ROWS, + } => ExecutionDataState::INGESTION_ROWS, } } @@ -585,16 +590,14 @@ fn child_domain( let assigned = match edge { ExecutionDataStateEdge::SummaryAggChild | ExecutionDataStateEdge::SummaryJoinInput - | ExecutionDataStateEdge::ValueOperationChild => { - ExecutionDataState::MAINTENANCE_ROWS - } + | ExecutionDataStateEdge::ValueOperationChild => ExecutionDataState::INGESTION_ROWS, ExecutionDataStateEdge::SummaryEstimateInput | ExecutionDataStateEdge::SummarySubtractInput | ExecutionDataStateEdge::SummaryDeleteInput | ExecutionDataStateEdge::SummaryMergeInput => { return Err(ExecutionDataStateError::IllegalChildDataState { edge: edge.describe(), - child: ExecutionDataState::MAINTENANCE_ROWS, + child: ExecutionDataState::INGESTION_ROWS, }) } }; @@ -609,7 +612,13 @@ fn state_only( edge: ExecutionDataStateEdge, ) -> Result { child_domain(child, edge, |avail| match avail { - ExecutionDataState::MAINTENANCE_SUMMARY => Ok(()), + state + if state.primitive == DataPrimitive::SummaryState + && (state.timing == ExecutionTiming::IngestionTime + || matches!(edge, ExecutionDataStateEdge::SummaryEstimateInput)) => + { + Ok(()) + } other => Err(ExecutionDataStateError::IllegalChildDataState { edge: edge.describe(), child: other, @@ -770,15 +779,15 @@ mod tests { #[test] fn raw_primitive_labels() { assert_eq!( - ExecutionDataState::MAINTENANCE_ROWS.primitive, + ExecutionDataState::INGESTION_ROWS.primitive, DataPrimitive::Raw ); - assert_eq!(ExecutionDataState::READ_ROWS.primitive, DataPrimitive::Raw); + assert_eq!(ExecutionDataState::QUERY_ROWS.primitive, DataPrimitive::Raw); assert_eq!( - ExecutionDataState::MAINTENANCE_ROWS.to_string(), - "maintenance_time/raw" + ExecutionDataState::INGESTION_ROWS.to_string(), + "ingestion_time/raw" ); - assert_eq!(ExecutionDataState::READ_ROWS.to_string(), "read_time/raw"); + assert_eq!(ExecutionDataState::QUERY_ROWS.to_string(), "query_time/raw"); assert_eq!(DataPrimitive::SummaryState.as_str(), "summary_state"); } @@ -878,11 +887,11 @@ mod tests { let assignment = validate_execution_data_states(&root).unwrap(); assert_eq!( assignment.data_state_of(&leaf), - Some(ExecutionDataState::MAINTENANCE_ROWS) + Some(ExecutionDataState::INGESTION_ROWS) ); assert_eq!( assignment.data_state_of(&root), - Some(ExecutionDataState::MAINTENANCE_SUMMARY) + Some(ExecutionDataState::INGESTION_SUMMARY) ); } @@ -897,23 +906,25 @@ mod tests { } #[test] - fn readout_under_summary_agg_is_rejected() { + fn readout_can_feed_summary_construction_at_query_time() { let inner = estimate(agg(keep(), kll())); - let root = agg(inner, kll()); - assert!(matches!( - validate_execution_data_states(&root), - Err(ExecutionDataStateError::ReadoutUnderMaintenance { .. }) - )); + let summary = agg(inner, kll()); + let root = estimate(summary.clone()); + let assignment = validate_execution_data_states(&root).unwrap(); + assert_eq!( + assignment.data_state_of(&summary).unwrap().timing, + ExecutionTiming::QueryTime + ); } #[test] - fn read_time_operation_over_readout_is_legal_and_root_is_readout() { + fn query_time_operation_over_readout_is_legal_and_root_is_readout() { let inner = estimate(agg(keep(), kll())); let root = Rc::new(SummaryNode { expr: SummaryExpr::ValueOperation { child: inner, operation: ValueOperation::Exact(max_op()), - timing: ExecutionTiming::ReadTime, + timing: ExecutionTiming::QueryTime, }, schema: plain(&["max"]), guarantee: None, @@ -921,7 +932,7 @@ mod tests { let assignment = validate_execution_data_states(&root).unwrap(); assert_eq!( assignment.data_state_of(&root), - Some(ExecutionDataState::READ_ROWS) + Some(ExecutionDataState::QUERY_ROWS) ); } @@ -934,7 +945,7 @@ mod tests { operation: ValueOperation::Extension { name: "approximate_calibration".into(), }, - timing: ExecutionTiming::ReadTime, + timing: ExecutionTiming::QueryTime, }, schema: plain(&["calibrated"]), guarantee: None, @@ -943,30 +954,25 @@ mod tests { let assignment = validate_execution_data_states(&root).unwrap(); assert_eq!( assignment.data_state_of(&root), - Some(ExecutionDataState::READ_ROWS) + Some(ExecutionDataState::QUERY_ROWS) ); } #[test] - fn read_time_operation_under_summary_agg_is_rejected() { + fn query_time_values_can_feed_query_time_summary_construction() { let inner = estimate(agg(keep(), kll())); let post = Rc::new(SummaryNode { expr: SummaryExpr::ValueOperation { child: inner, operation: ValueOperation::Exact(max_op()), - timing: ExecutionTiming::ReadTime, + timing: ExecutionTiming::QueryTime, }, 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, - }) - ); + let root = estimate(root); + validate_execution_data_states(&root).unwrap(); } #[test] @@ -975,7 +981,7 @@ mod tests { expr: SummaryExpr::ValueOperation { child: keep(), operation: ValueOperation::Exact(max_op()), - timing: ExecutionTiming::MaintenanceTime, + timing: ExecutionTiming::IngestionTime, }, schema: plain(&["max"]), guarantee: None, @@ -988,7 +994,7 @@ mod tests { let assignment = validate_execution_data_states(&root).unwrap(); assert_eq!( assignment.data_state_of(&operation), - Some(ExecutionDataState::MAINTENANCE_ROWS) + Some(ExecutionDataState::INGESTION_ROWS) ); } @@ -999,7 +1005,7 @@ mod tests { expr: SummaryExpr::ValueOperation { child: inner, operation: ValueOperation::Exact(max_op()), - timing: ExecutionTiming::MaintenanceTime, + timing: ExecutionTiming::IngestionTime, }, schema: plain(&["max"]), guarantee: None, @@ -1009,11 +1015,78 @@ mod tests { validate_execution_data_states(&root), Err(ExecutionDataStateError::IllegalChildDataState { edge: "ValueOperation.child", - child: ExecutionDataState::READ_ROWS + child: ExecutionDataState::QUERY_ROWS }) )); } + #[test] + fn execution_phase_wire_names_are_ingestion_and_query_time() { + for (phase, name) in [ + (ExecutionTiming::IngestionTime, "ingestion_time"), + (ExecutionTiming::QueryTime, "query_time"), + ] { + assert_eq!(phase.as_str(), name); + assert_eq!(serde_json::to_value(phase).unwrap(), name); + assert_eq!( + serde_json::from_value::(serde_json::json!(name)).unwrap(), + phase + ); + } + assert!(serde_json::from_str::("\"maintenance_time\"").is_err()); + assert!(serde_json::from_str::("\"MaintenanceTime\"").is_err()); + } + + #[test] + fn summary_merge_runs_at_ingestion_or_query_time() { + for timing in [ExecutionTiming::IngestionTime, ExecutionTiming::QueryTime] { + let input = agg(keep(), kll()); + let merged = Rc::new(SummaryNode { + expr: SummaryExpr::SummaryMerge { + children: vec![input.clone()], + timing, + }, + schema: input.schema.clone(), + guarantee: None, + }); + let root = estimate(merged.clone()); + let assignment = validate_execution_data_states(&root).unwrap(); + assert_eq!( + assignment.data_state_of(&merged), + Some(ExecutionDataState { + timing, + primitive: DataPrimitive::SummaryState, + }) + ); + let exported = crate::post_asap::compile_executable_dag(&root).unwrap(); + assert!(exported.nodes.iter().any(|node| matches!(node.payload, + crate::post_asap::ExecutableOperatorPayload::SummaryMerge + if node.output_state.timing == timing))); + } + } + + #[test] + fn ingestion_merge_cannot_depend_on_query_execution() { + let input = agg(keep(), kll()); + let query_merge = Rc::new(SummaryNode { + expr: SummaryExpr::SummaryMerge { + children: vec![input.clone()], + timing: ExecutionTiming::QueryTime, + }, + schema: input.schema.clone(), + guarantee: None, + }); + let ingestion_merge = Rc::new(SummaryNode { + expr: SummaryExpr::SummaryMerge { + children: vec![query_merge], + timing: ExecutionTiming::IngestionTime, + }, + schema: input.schema.clone(), + guarantee: None, + }); + assert!(validate_execution_data_states(&ingestion_merge).is_err()); + } + #[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 @@ -1025,19 +1098,20 @@ mod tests { expr: SummaryExpr::ValueOperation { child: Rc::clone(&shared), operation: ValueOperation::Exact(max_op()), - timing: ExecutionTiming::ReadTime, + timing: ExecutionTiming::QueryTime, }, schema: plain(&["max"]), guarantee: None, }); let root = Rc::new(SummaryNode { expr: SummaryExpr::SummaryMerge { + timing: ExecutionTiming::IngestionTime, children: vec![ Rc::new(SummaryNode { expr: SummaryExpr::ValueOperation { child: maintained, operation: ValueOperation::Exact(max_op()), - timing: ExecutionTiming::ReadTime, + timing: ExecutionTiming::QueryTime, }, schema: plain(&["max"]), guarantee: None, @@ -1051,17 +1125,12 @@ mod tests { // 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(); + visit(&shared, ExecutionDataState::INGESTION_ROWS, &mut assignment).unwrap(); assert_eq!( - visit(&shared, ExecutionDataState::READ_ROWS, &mut assignment), + visit(&shared, ExecutionDataState::QUERY_ROWS, &mut assignment), Err(ExecutionDataStateError::AmbiguousKeepPreAsap { - first: ExecutionDataState::MAINTENANCE_ROWS, - second: ExecutionDataState::READ_ROWS, + first: ExecutionDataState::INGESTION_ROWS, + second: ExecutionDataState::QUERY_ROWS, }) ); assert!(validate_execution_data_states(&root).is_err()); diff --git a/crates/types/src/post_asap/expr.rs b/crates/types/src/post_asap/expr.rs index c5c878f9..12e2c21c 100644 --- a/crates/types/src/post_asap/expr.rs +++ b/crates/types/src/post_asap/expr.rs @@ -67,6 +67,8 @@ pub enum ValueOperation { Limit { n: usize, offset: usize, + /// Apply the offset and limit independently to each group. + partition_by: GroupKeys, }, Extension { name: String, @@ -136,17 +138,6 @@ pub enum SummaryExpr { operator: BinaryOperator, }, - /// Use an approximate keyed summary only to propose members, then rank - /// those members by authoritative exact values. `candidates` never - /// supplies caller-visible values. - CandidateTopK { - candidates: Rc, - values: Rc, - k: usize, - grouping: GroupKeys, - completeness: CandidateCompleteness, - }, - /// Plain-row semantics composed with a post-ASAP child. Timing is an /// independent physical choice, not part of the operation's identity. ValueOperation { @@ -163,6 +154,8 @@ pub enum SummaryExpr { right: Rc, kind: JoinKind, pred: Predicate, + /// Optional proof for candidate pruning; ranking remains a separate operation. + pruning: Option, }, /// Summary aggregation. Post-ASAP binding chose `family` — which @@ -252,7 +245,10 @@ pub enum SummaryExpr { /// `mergeable` must be true. Inserted by a deployment's own stage /// allocator (not modeled in this crate) on cut edges. /// Output schema: one field (same family + params as inputs). - SummaryMerge { children: Vec> }, + SummaryMerge { + children: Vec>, + timing: ExecutionTiming, + }, } /// All semantics owned by a post-ASAP binary operator. diff --git a/crates/types/src/post_asap/schema.rs b/crates/types/src/post_asap/schema.rs index ffef2a05..d67e249d 100644 --- a/crates/types/src/post_asap/schema.rs +++ b/crates/types/src/post_asap/schema.rs @@ -25,7 +25,7 @@ pub enum SummaryFamilyType { /// passed through unchanged from a pre-ASAP edge. Plain(DataType), /// Exact, mergeable accumulator state (`Sum`/`Count`/`Min`/`Max`/`Rate`/ - /// `Increase`) — the partial state *is* the value; no readout needed. + /// `Increase`). Value consumers require an explicit finalization boundary. ExactAggregate(ExactKind, ExactParams), /// Approximate sketch state (KLL/CMS/HLL/…), read out via a /// `SummaryEstimate`. A [`SketchKind`] already carries the concrete diff --git a/crates/types/src/post_asap/sketch.rs b/crates/types/src/post_asap/sketch.rs index 6eac4b75..391d12f0 100644 --- a/crates/types/src/post_asap/sketch.rs +++ b/crates/types/src/post_asap/sketch.rs @@ -588,14 +588,6 @@ pub enum SummaryInputExpr { Column(ColumnRef), Tuple(Vec), EntityIdentity(EntityIdentity), - /// Reset-aware non-negative increment derived at ingest from the current - /// counter sample and the previous sample for the same series. This is an - /// update expression, not a query-time rate estimate; CandidateTopK uses - /// it only for membership and reranks against an exact counter SDS. - ResetAwareCounterDelta { - value: ColumnRef, - series: EntityIdentity, - }, } /// What to extract from a built summary. Carried by `SummaryEstimate`. diff --git a/docs/design_docs/architecture/evidence-dependent-candidates.md b/docs/design_docs/architecture/evidence-dependent-candidates.md index 71a18e42..a897a630 100644 --- a/docs/design_docs/architecture/evidence-dependent-candidates.md +++ b/docs/design_docs/architecture/evidence-dependent-candidates.md @@ -54,6 +54,7 @@ not emit a `RejectedCandidate` for that case. | Direct DDSketch quantile ratio | Candidate with no root guarantee | A supplied domain incompatible with DDSketch causes this ratio candidate to be omitted; no `RejectedCandidate` is recorded. | | Hydra grouping | Symbolic shared-grid collision/failure terms | Reject with typed accuracy reason. | | Count-ranked TopK | Symbolic interval margin or failure probability | Reject overlapping/non-finite supplied intervals. | +| Rate/increase-weighted grouped TopK | Symbolic distinct-item count in the score union bound and symbolic membership margin/failure terms; retain the CMS/heap candidate | Reject invalid supplied population bounds, invalid intervals, or known contributions that already violate the target. | | HLL confidence | Symbolic failure probability | Reject a fully known unmet root target. | | Relative-value composition | Symbolic bound when input sign is unknown | Reject known signed input for this rule. | | Exact sum/average/extremum | Symbolic row-count probability term | Reject unsupported metric combinations. | diff --git a/docs/design_docs/architecture/physical-plan-integration.md b/docs/design_docs/architecture/physical-plan-integration.md index dbde8024..934bd958 100644 --- a/docs/design_docs/architecture/physical-plan-integration.md +++ b/docs/design_docs/architecture/physical-plan-integration.md @@ -112,7 +112,7 @@ Every `SummaryExpr` operation also needs explicit physical realization: | `BinaryOp` | binary evaluation preserving operand order, execution timing and any typed finite/relative-division guard | | `ValueOperation` | concrete realization of the value operation with its required execution timing and data state | | `RelationalJoin` | concrete row-join algorithm preserving join kind and predicate | -| `CandidateTopK` | candidate generation and authoritative value ranking that preserve the membership completeness contract | +| `RelationalJoin` with `JoinKind::Semi` | retain left rows matching explicit right-side keys; candidate pruning carries completeness evidence and ordinary TopK ranks the result | This table is a completeness requirement, not a claim that every realization already exists. Until lowering introduces an explicit physical operator, @@ -424,3 +424,39 @@ distinct from `checked_relative_division`, whose relative-error certificate also requires a normal result; setting both guards or attaching a guard to a non-division operator is invalid. Compilers must preserve this typed condition rather than recovering average semantics from query text. + +### Candidate pruning is a subgraph + +Candidate-based TopK uses a summary key readout, a general semi-join over +explicit matching key columns, grouped Sort by the authoritative score, and +grouped Limit. Sort and Limit carry the same partition keys. +The join preserves authoritative left-side values and does not rank or limit +rows. Completeness evidence controls whether pruning is valid. A plain Limit(k) +cannot replace either key matching or value ranking. + +Every physical operator can be placed at ingestion time or query time. The +physical node carries that choice; operator payloads do not contain phase +fields. The phase assignment API updates producer edge states and rejects an +ingestion computation that depends on query-time work. Deployment capability, +storage readiness, schemas and approximation guarantees remain separate checks. + +Executable DAG wire version 4 removes the special membership operator, its edge +roles and the duplicate operator phase fields without compatibility aliases. + +The candidate row producer derives its key columns from the summary update's +item identity and subpopulation keys. Its last column, `__asap_estimate`, is the +summary score; it is never substituted for an authoritative value. The join +maps keys by unambiguous name and matching type, including numeric keys, and +matches two NULL grouping keys. It does not infer identity from string types +or overwrite the right input's schema with the left input's schema. Unresolved, +ambiguous or incompatible identities are rejected during realization. Dynamic +label sets without an explicit row representation cannot use this row join. + +Ranking selects the aggregate output column, not the first numeric column: +a numeric entity key is not a score. Exact accumulator inputs are explicitly +finalized before row operators consume them. None of these operations proves +candidate completeness; that evidence belongs to the semi-join's pruning step. + +The semantic `SummaryExpr` constructors still propose an initial execution +layout. Uniform phase assignment applies to the exported executable DAG; +it is not a claim that every deployment has implemented every placement. diff --git a/docs/design_docs/concepts/post-asap-ir.md b/docs/design_docs/concepts/post-asap-ir.md index 9d72e4f8..b1543772 100644 --- a/docs/design_docs/concepts/post-asap-ir.md +++ b/docs/design_docs/concepts/post-asap-ir.md @@ -43,9 +43,10 @@ summary family supports incremental maintenance. filter, sort, limit or extension semantics with explicit execution timing. - `RelationalJoin`: join row-producing children using the specified join kind and predicate. -- `CandidateTopK`: propose candidate members and rank them by authoritative exact - values; the completeness contract distinguishes certified from best-effort - membership. +- Candidate pruning uses `RelationalJoin` with `JoinKind::Semi` and an explicit + equality predicate on key columns. The left input supplies authoritative + values; the right input supplies keys. Grouped Sort followed by grouped Limit ranks + and selects the joined rows. Completeness evidence belongs to pruning, not ranking. A `SummaryNode` carries its expression, schema and optional result guarantee. State and query values have different contracts. Exact operations over @@ -53,3 +54,63 @@ approximate readouts still require composed accuracy guarantees. See the [accuracy implementation companion](../../develop_docs/end-to-end-accuracy-guarantees.md) and [physical-plan integration](../architecture/physical-plan-integration.md) for the corresponding correctness and realization requirements. + +## Execution phase + +A physical operator defines what computation happens. The plan decides when it +happens: **ingestion time** or **query time**. Operator identity must not imply +one of these phases. Backend capability restrictions are implementation gaps, +not definitions of the operator. + +Every executable physical payload supports both phase assignments. Phase is +stored on the physical node, independently of its operator payload. +`ExecutableDag::with_execution_phases` assigns a phase to every node and updates +its edges. Ingestion work cannot depend on a future query result. Default +semantic realization still proposes an initial layout; it does not restrict +which phase a physical operator may use. Deployments must separately check that +they have an implementation and a valid data source for the chosen placement. + +## Weighted grouped TopK + +For `topk by(job)(2, sum by(service, job)(rate(m[1m])))`, the summary +realization consumes the complete per-series rate results. Each job owns a +separate CMS and candidate heap. Inside that partition, the item is service and +the update weight is the series rate. Summing updates for one item implements +the logical grouped sum without first constructing all exact grouped sums. + +The DAG is per-series rate → finalized values → partitioned summary construction +→ typed candidate/score readout → output projection → grouped Sort → grouped +Limit. The output count is two per job. The candidate capacity is a separate +parameter, provisionally `max(k, ceil(1 / epsilon))`; this sizing choice is not a +membership theorem. Missing evidence retains a logical candidate with symbolic unknown guarantees; +default selection does not certify or choose it. +The row readout restores job and service identities and returns estimated sums. +There is no mandatory exact scoring branch or candidate semi-join in this path. +The old raw counter-delta update expression is removed rather than retained as a +compatibility option: counter increments are not complete windowed rate results. + +The direct readout represents both score error and membership. A source provider +supplies an enforced upper bound on distinct partition/item identities for the +complete readout. Planner uses this bound to size confidence and union-bound +score errors over adaptively selected items. Membership evidence is evaluated +for the query's output count, not the candidate capacity. Score and membership +failure probabilities are combined, and the score guarantee remains in the +membership guarantee's child provenance. An exact request does not accept this +approximate output path merely because its selected identities are certified. + +Deployment chooses ingestion time or query time for these operators. The +semantic constructor proposes a layout; `with_execution_phases` assigns the +executable placement. Either deployment must give each evaluation a complete +rate window and an isolated summary state, or maintain an equivalent replacement +strategy. Appending successive rate snapshots to one cumulative state is invalid. +An ingestion execution can compute a window before the query and store its state; +a query execution can construct the same state on demand. These are placements +of the same computation, not separate summary semantics. + +This follows the evidence-dependent candidate contract from #455. Missing +population or margin evidence is exported as symbolic unknown terms, rather than +erasing a constructible summary. Backend/runtime/deployment inspects these +requirements and supplies applicable evidence before selection and installation. +Re-running planning with that provider resolves guarantees and may resize the +candidate. Known-invalid evidence or known bounds that already miss the target +are rejected; an optimistic floor is never exported as a certificate. diff --git a/tools/dag-viewer/node-style.js b/tools/dag-viewer/node-style.js index a13c51d8..447f0880 100644 --- a/tools/dag-viewer/node-style.js +++ b/tools/dag-viewer/node-style.js @@ -33,7 +33,6 @@ const KIND_CATEGORY_JSON = `{ "SummaryJoin": "summary", "SummarySubtract": "summary", "SummaryBinaryOp": "summary", - "CandidateTopK": "summary", "ValueOperation": "summary", "SummaryDelete": "summary", "SummaryEstimate": "summary", @@ -106,7 +105,7 @@ const CATEGORIES = { // Post-ASAP nodes use a neutral palette; KeepPreAsap has a muted override. summary: { label: 'Summary', - description: 'KeepPreAsap, SummaryBinaryOp, CandidateTopK, ValueOperation, SummaryAgg, SummaryJoin, SummarySubtract, SummaryDelete, SummaryEstimate, SummaryMerge — post-ASAP materialized structures', + description: 'KeepPreAsap, SummaryBinaryOp, ValueOperation, SummaryAgg, SummaryJoin, SummarySubtract, SummaryDelete, SummaryEstimate, SummaryMerge — post-ASAP materialized structures', light: { bg: '#f1f2f4', border: '#4b5563' }, dark: { bg: '#20242b', border: '#9ca3af' }, },