From f3b015cc6db72368e00d05ec51bd9fecad356dc3 Mon Sep 17 00:00:00 2001 From: zz_y Date: Sat, 29 Aug 2026 16:41:12 -0600 Subject: [PATCH 1/3] feat(workload): normalize query demand and data evidence --- crates/asap-aware-mapping/src/accuracy.rs | 76 +++ crates/asap-aware-mapping/src/recurrence.rs | 18 +- crates/frontend-promql/src/lib.rs | 6 +- .../frontend-promql/tests/promql_lowering.rs | 29 +- crates/frontend-sql/src/lib.rs | 6 +- crates/types/src/pre_asap/cse.rs | 2 +- crates/types/src/workload.rs | 455 +++++++++++++++++- 7 files changed, 558 insertions(+), 34 deletions(-) diff --git a/crates/asap-aware-mapping/src/accuracy.rs b/crates/asap-aware-mapping/src/accuracy.rs index a9c02249..4e6b2d40 100644 --- a/crates/asap-aware-mapping/src/accuracy.rs +++ b/crates/asap-aware-mapping/src/accuracy.rs @@ -88,6 +88,10 @@ pub struct PropagationStats { /// of groups a `sum` folds), for `ExactSum`/`ExactExtremum`'s union /// bound over per-input failures. pub input_row_count: Option, + /// Fresh key-frequency distribution evidence from the data workload. + /// Built-in rules preserve it for deployment-specific accuracy models; + /// they do not assume a favorable distribution when it is absent. + pub data_distribution: Option, /// Lower confidence bound of the kth selected TopK item, after widening /// the interval by the sketch's own estimation error. pub topk_selected_lower_bound: Option, @@ -120,6 +124,29 @@ pub struct NoAccuracyEvidence; impl AccuracyEvidenceProvider for NoAccuracyEvidence {} +/// Accuracy evidence backed by the normalized data workload. Freshness is +/// checked at the planning time before values reach any accuracy rule. +#[derive(Debug, Clone, Copy)] +pub struct WorkloadAccuracyEvidence<'a> { + pub data: &'a asap_types::workload::DataWorkload, + pub now_ms: u64, +} + +impl AccuracyEvidenceProvider for WorkloadAccuracyEvidence<'_> { + fn propagation_stats( + &self, + _op: &CompositionOperator, + _family: &SummaryFamilyType, + _query: Option<&SketchQuery>, + ) -> PropagationStats { + PropagationStats { + input_row_count: self.data.input_cardinality.value_at(self.now_ms).copied(), + data_distribution: self.data.distribution.value_at(self.now_ms).cloned(), + ..PropagationStats::default() + } + } +} + /// The deployment-extensible accuracy algebra. `asap-aware-mapping` ships /// [`DefaultAccuracyModel`]; a deployment with a proof for a composition the /// default rejects (a registered cross-metric conversion, say) implements @@ -839,6 +866,7 @@ impl AccuracyBudgetAllocator for EqualSplitAllocator { mod tests { use super::*; use asap_types::post_asap::{GroupingStrategy, SketchKind}; + use asap_types::workload::{DataDistribution, DataWorkload, Evidence, EvidenceSource}; fn abs(bound: f64, delta: f64) -> ResultGuarantee { ResultGuarantee { @@ -865,6 +893,54 @@ mod tests { } } + #[test] + fn workload_accuracy_evidence_uses_only_fresh_data_characteristics() { + let data = DataWorkload { + input_cardinality: Evidence { + value: Some(42), + source: EvidenceSource::Observed, + observed_at_ms: Some(1_000), + valid_for_ms: Some(500), + }, + distribution: Evidence { + value: Some(DataDistribution::Bursty), + source: EvidenceSource::Observed, + observed_at_ms: Some(1_000), + valid_for_ms: Some(500), + }, + ..Default::default() + }; + let provider = WorkloadAccuracyEvidence { + data: &data, + now_ms: 1_500, + }; + let fresh = provider.propagation_stats( + &CompositionOperator::ExactSum, + &SummaryFamilyType::ExactAggregate( + asap_types::post_asap::ExactKind::Sum, + asap_types::post_asap::ExactParams::Sum, + ), + None, + ); + assert_eq!(fresh.input_row_count, Some(42)); + assert_eq!(fresh.data_distribution, Some(DataDistribution::Bursty)); + + let stale = WorkloadAccuracyEvidence { + data: &data, + now_ms: 1_501, + } + .propagation_stats( + &CompositionOperator::ExactSum, + &SummaryFamilyType::ExactAggregate( + asap_types::post_asap::ExactKind::Sum, + asap_types::post_asap::ExactParams::Sum, + ), + None, + ); + assert_eq!(stale.input_row_count, None); + assert_eq!(stale.data_distribution, None); + } + #[test] fn exact_child_contributes_zero_error() { let local = abs(0.05, 0.01); diff --git a/crates/asap-aware-mapping/src/recurrence.rs b/crates/asap-aware-mapping/src/recurrence.rs index 8b380584..01f43e59 100644 --- a/crates/asap-aware-mapping/src/recurrence.rs +++ b/crates/asap-aware-mapping/src/recurrence.rs @@ -73,7 +73,7 @@ //! //! ## Provenance of each new cost input //! -//! - [`EvaluationRate`]: derived from [`asap_types::workload::RepeatingEntry::interval`] +//! - [`EvaluationRate`]: derived from [`asap_types::workload::RepeatingEntry::demand`] //! values of every repeating consumer reaching a target (via //! [`evaluation_rate_of`], or [`crate::replacement::PlanSpace::recurrence_profiles`] //! for a whole workload). A one-shot ([`asap_types::workload::BatchEntry`]) @@ -200,6 +200,12 @@ pub enum RecurrenceError { well-defined maintained_cost_rate" )] InvalidUpdateRate(UpdateRate), + #[error("invalid EvaluationRate({0:?}Hz): an evaluation rate must be finite and >= 0")] + InvalidEvaluationRate(EvaluationRate), + #[error(transparent)] + InvalidWorkload(#[from] asap_types::workload::WorkloadError), + #[error("workload entry index {index} is out of bounds for {entry_count} entries")] + InvalidWorkloadEntry { index: usize, entry_count: usize }, /// A [`Horizon`] that isn't finite and strictly positive (NaN, /// infinite, zero, or negative) was supplied — a non-positive or /// infinite horizon would silently drop or invert the recurring @@ -374,16 +380,24 @@ impl RecurrenceProfile { /// already-opaque `Id` granularity `search_workload`'s callers already use /// — this crate needs no more of a caller's own query identity than "which /// of these two recurrence kinds is this root". -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, PartialEq)] pub enum RootRecurrence { /// A one-shot (batch) root — contributes to a reached target's /// [`RecurrenceProfile::one_shot_consumers`], never to its /// `evaluation_rate`. OneShot, + /// A declared number of one-time invocations for this root. + OneShotCount(usize), /// A repeating root firing every `RepetitionInterval` — contributes to /// a reached target's `evaluation_rate` (`1 / interval`, aggregated via /// [`evaluation_rate_of`]). Repeating(RepetitionInterval), + /// A repeated root whose schedule or estimate has already been + /// normalized to evaluations per second. + RepeatingRate(EvaluationRate), + /// No reliable recurrence evidence was supplied. It contributes no read + /// count or evaluation rate, but remains distinct from zero demand. + Unknown, } // ── Explanation ────────────────────────────────────────────────────────── diff --git a/crates/frontend-promql/src/lib.rs b/crates/frontend-promql/src/lib.rs index 08fec09a..ff29379a 100644 --- a/crates/frontend-promql/src/lib.rs +++ b/crates/frontend-promql/src/lib.rs @@ -71,11 +71,7 @@ pub fn lower_promql_batch(workload: &QueryWorkload) -> Vec AccuracyTarget { + match self { + Self::Explicit(target) => target.clone(), + Self::ImplicitExact => AccuracyTarget::Exact, + } + } +} + +/// Optional maximum wall-clock response time for one query execution. +#[derive(Debug, Clone, Copy, Default, PartialEq)] +pub enum LatencyRequirement { + ExplicitMaxMs(f64), + #[default] + Unspecified, +} + +/// Independent accuracy and response-latency constraints attached to one +/// query in the workload. +#[derive(Debug, Clone, PartialEq)] pub struct QueryRequirements { - /// Maximum acceptable approximation error. - pub accuracy: Option, - /// Maximum acceptable end-to-end query latency in milliseconds. - pub latency_ms: Option, + pub accuracy: AccuracyRequirement, + pub response_latency: LatencyRequirement, +} + +impl Default for QueryRequirements { + fn default() -> Self { + Self { + accuracy: AccuracyRequirement::ImplicitExact, + response_latency: LatencyRequirement::Unspecified, + } + } } // ── Workload entries ────────────────────────────────────────────────────────── +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub enum Predictability { + AdHoc, + Predictable { + known_at: Option, + }, + #[default] + Unknown, +} + +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub enum QueryTimeScope { + RealTime, + Longitudinal, + Mixed, + #[default] + Unknown, +} + +/// Concrete event-time interval selected by a query, kept separate from its +/// semantic real-time/longitudinal classification. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct TimeSelection { + pub scope: QueryTimeScope, + pub lookback: Option, + /// Fixed upper bound. `None` means the planning/evaluation time. + pub as_of: Option, +} + +#[derive(Debug, Clone, Copy, PartialEq)] +pub struct Confidence(pub f64); + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct ObservationWindow { + pub start: TimestampMs, + pub end: TimestampMs, +} + +#[derive(Debug, Clone, Copy, PartialEq)] +pub enum ExpectedDemand { + InvocationCount(u64), + AverageRate(Rate), +} + +#[derive(Debug, Clone, PartialEq)] +pub struct DemandEstimate { + pub observation_window: ObservationWindow, + pub expected: ExpectedDemand, + pub peak_rate: Option, + pub max_concurrency: Option, + pub confidence: Confidence, + pub source: EvidenceSource, + pub observed_at: Option, + pub valid_for: Option, +} + +#[derive(Debug, Clone, PartialEq)] +pub enum RepeatedDemand { + FixedInterval(RepetitionInterval), + Scheduled(Vec), + EstimatedRate(DemandEstimate), +} + +#[derive(Debug, Clone, PartialEq)] +pub enum QueryRecurrence { + OneTime { + invocations: u64, + execute_at: Option, + }, + Repeated(RepeatedDemand), + Unknown, +} + /// One entry in a one-shot batch: a query plus its optional SLA constraints. #[derive(Debug, Clone)] pub struct BatchEntry { pub query: Query, - pub requirements: Option, + pub requirements: QueryRequirements, + pub predictability: Predictability, + pub invocations: u64, + pub execute_at: Option, + pub time_selection: TimeSelection, } /// One query that fires every `interval` milliseconds. Its recurrence does @@ -55,9 +173,46 @@ pub struct BatchEntry { #[derive(Debug, Clone)] pub struct RepeatingEntry { pub query: Query, - /// How often the query fires, in milliseconds. - pub interval: RepetitionInterval, - pub requirements: Option, + pub demand: RepeatedDemand, + pub requirements: QueryRequirements, + pub predictability: Predictability, + pub time_selection: TimeSelection, +} + +#[derive(Debug, Clone, PartialEq)] +pub struct QueryWorkloadEntry { + pub query: Query, + pub requirements: QueryRequirements, + pub predictability: Predictability, + pub recurrence: QueryRecurrence, + pub time_selection: TimeSelection, +} + +impl From<&BatchEntry> for QueryWorkloadEntry { + fn from(entry: &BatchEntry) -> Self { + Self { + query: entry.query.clone(), + requirements: entry.requirements.clone(), + predictability: entry.predictability.clone(), + recurrence: QueryRecurrence::OneTime { + invocations: entry.invocations, + execute_at: entry.execute_at, + }, + time_selection: entry.time_selection.clone(), + } + } +} + +impl From<&RepeatingEntry> for QueryWorkloadEntry { + fn from(entry: &RepeatingEntry) -> Self { + Self { + query: entry.query.clone(), + requirements: entry.requirements.clone(), + predictability: entry.predictability.clone(), + recurrence: QueryRecurrence::Repeated(entry.demand.clone()), + time_selection: entry.time_selection.clone(), + } + } } // ── Data workload ───────────────────────────────────────────────────────────── @@ -120,6 +275,36 @@ impl Default for Evidence { } } +impl Evidence { + /// Return the value only while its freshness contract holds. Declared or + /// timeless evidence with no `valid_for_ms` does not expire. + pub fn value_at(&self, now_ms: u64) -> Option<&T> { + let value = self.value.as_ref()?; + match (self.observed_at_ms, self.valid_for_ms) { + (Some(observed), _) if observed > now_ms => None, + (Some(observed), Some(valid_for)) if now_ms > observed.saturating_add(valid_for) => { + None + } + (None, Some(_)) => None, + _ => Some(value), + } + } +} + +impl DemandEstimate { + /// Whether this estimate was already observed and has not expired at + /// `now_ms`. A validity duration without an observation time is not a + /// usable freshness contract. + pub fn is_fresh_at(&self, now_ms: u64) -> bool { + match (self.observed_at, self.valid_for) { + (Some(observed), _) if observed.0 > now_ms => false, + (Some(observed), Some(valid_for)) => now_ms <= observed.0.saturating_add(valid_for.0), + (None, Some(_)) => false, + _ => true, + } + } +} + /// Queries per second, samples per second, or another rate whose unit is /// established by the field that contains it. #[derive(Debug, Clone, Copy, PartialEq)] @@ -141,9 +326,9 @@ pub struct DataWorkload { /// The single normalised input type accepted by every entry point into the /// planner (HTTP POST /plan, YAML file, query-log replay, OpAMP callback). /// -/// `query_batch` and `repeating_queries` are mutually exclusive today; both -/// may be present in the future when mixed batch+streaming workloads are -/// supported. +/// `query_batch` and `repeating_queries` may both be present. [`Self::entries`] +/// normalizes them into one ordered stream without conflating recurrence with +/// data arrival. #[derive(Debug, Clone)] pub struct QueryWorkload { /// Source language shared by all queries in this workload. @@ -156,3 +341,245 @@ pub struct QueryWorkload { /// Applies to all queries in this workload. pub data_workload: Option, } + +impl QueryWorkload { + /// One normalized entry stream, independent of the source's legacy + /// batch/repeating containers. Mixed workloads preserve both kinds. + pub fn entries(&self) -> impl Iterator + '_ { + self.query_batch + .iter() + .flatten() + .map(QueryWorkloadEntry::from) + .chain( + self.repeating_queries + .iter() + .flatten() + .map(QueryWorkloadEntry::from), + ) + } + + pub fn validate(&self) -> Result<(), WorkloadError> { + for entry in self.entries() { + validate_entry(&entry)?; + } + if let Some(data) = &self.data_workload { + if matches!(data.arrival, DataArrival::AtRest) + && data.ingestion_rate.value.is_some_and(|rate| rate.0 > 0.0) + { + return Err(WorkloadError::AtRestWithPositiveIngestionRate); + } + validate_optional_rate(data.ingestion_rate.value)?; + } + Ok(()) + } +} + +fn validate_entry(entry: &QueryWorkloadEntry) -> Result<(), WorkloadError> { + if let LatencyRequirement::ExplicitMaxMs(ms) = entry.requirements.response_latency { + if !ms.is_finite() || ms < 0.0 { + return Err(WorkloadError::InvalidLatency(ms)); + } + } + match &entry.recurrence { + QueryRecurrence::OneTime { invocations: 0, .. } => { + return Err(WorkloadError::ZeroInvocations) + } + QueryRecurrence::Repeated(RepeatedDemand::FixedInterval(RepetitionInterval(0))) => { + return Err(WorkloadError::ZeroRepetitionInterval) + } + QueryRecurrence::Repeated(RepeatedDemand::Scheduled(schedule)) if schedule.is_empty() => { + return Err(WorkloadError::EmptySchedule) + } + QueryRecurrence::Repeated(RepeatedDemand::EstimatedRate(estimate)) => { + if estimate.observation_window.start >= estimate.observation_window.end { + return Err(WorkloadError::EmptyObservationWindow); + } + if !(estimate.confidence.0.is_finite() && (0.0..=1.0).contains(&estimate.confidence.0)) + { + return Err(WorkloadError::InvalidConfidence(estimate.confidence.0)); + } + if let ExpectedDemand::AverageRate(rate) = estimate.expected { + validate_rate(rate)?; + } + validate_optional_rate(estimate.peak_rate)?; + } + _ => {} + } + Ok(()) +} + +fn validate_optional_rate(rate: Option) -> Result<(), WorkloadError> { + rate.map(validate_rate).transpose().map(|_| ()) +} + +fn validate_rate(rate: Rate) -> Result { + if rate.0.is_finite() && rate.0 >= 0.0 { + Ok(rate) + } else { + Err(WorkloadError::InvalidRate(rate.0)) + } +} + +#[derive(Debug, Clone, Copy, PartialEq, thiserror::Error)] +pub enum WorkloadError { + #[error("a one-time query must have at least one invocation")] + ZeroInvocations, + #[error("a fixed repetition interval must be greater than zero")] + ZeroRepetitionInterval, + #[error("a repeated-query schedule must not be empty")] + EmptySchedule, + #[error("a demand-estimate observation window must have start < end")] + EmptyObservationWindow, + #[error("confidence must be finite and in [0, 1], got {0}")] + InvalidConfidence(f64), + #[error("rate must be finite and non-negative, got {0}")] + InvalidRate(f64), + #[error("response latency must be finite and non-negative, got {0} ms")] + InvalidLatency(f64), + #[error("data at rest cannot have a positive ingestion rate")] + AtRestWithPositiveIngestionRate, +} + +#[cfg(test)] +mod tests { + use super::*; + + fn base_workload() -> QueryWorkload { + QueryWorkload { + language: QueryLanguage::PromQL, + query_batch: None, + repeating_queries: None, + data_workload: None, + } + } + + #[test] + fn mixed_batch_and_repeated_entries_normalize_without_conflating_axes() { + let mut workload = base_workload(); + workload.query_batch = Some(vec![BatchEntry { + query: Query("historical".into()), + requirements: QueryRequirements::default(), + predictability: Predictability::AdHoc, + invocations: 1, + execute_at: None, + time_selection: TimeSelection { + scope: QueryTimeScope::Longitudinal, + lookback: Some(DurationMs(300_000)), + as_of: Some(TimestampMs(1_000_000)), + }, + }]); + workload.repeating_queries = Some(vec![RepeatingEntry { + query: Query("dashboard".into()), + demand: RepeatedDemand::FixedInterval(RepetitionInterval(10_000)), + requirements: QueryRequirements::default(), + predictability: Predictability::Predictable { known_at: None }, + time_selection: TimeSelection { + scope: QueryTimeScope::RealTime, + lookback: Some(DurationMs(300_000)), + as_of: None, + }, + }]); + + let entries: Vec<_> = workload.entries().collect(); + assert_eq!(entries.len(), 2); + assert!(matches!( + entries[0].recurrence, + QueryRecurrence::OneTime { .. } + )); + assert!(matches!( + entries[1].recurrence, + QueryRecurrence::Repeated(RepeatedDemand::FixedInterval(_)) + )); + assert_eq!( + entries[0].time_selection.scope, + QueryTimeScope::Longitudinal + ); + assert_eq!(entries[1].time_selection.scope, QueryTimeScope::RealTime); + workload.validate().unwrap(); + } + + #[test] + fn stale_evidence_is_unknown_at_planning_time() { + let evidence = Evidence { + value: Some(Rate(10.0)), + source: EvidenceSource::Observed, + observed_at_ms: Some(1_000), + valid_for_ms: Some(500), + }; + assert_eq!(evidence.value_at(1_500), Some(&Rate(10.0))); + assert_eq!(evidence.value_at(1_501), None); + } + + #[test] + fn future_evidence_and_demand_estimates_are_not_fresh() { + let evidence = Evidence { + value: Some(Rate(2.0)), + source: EvidenceSource::Observed, + observed_at_ms: Some(2_000), + valid_for_ms: Some(1_000), + }; + assert_eq!(evidence.value_at(1_999), None); + assert_eq!(evidence.value_at(2_000), Some(&Rate(2.0))); + + let estimate = DemandEstimate { + observation_window: ObservationWindow { + start: TimestampMs(0), + end: TimestampMs(1_000), + }, + expected: ExpectedDemand::AverageRate(Rate(1.0)), + peak_rate: None, + max_concurrency: None, + confidence: Confidence(1.0), + source: EvidenceSource::Observed, + observed_at: Some(TimestampMs(2_000)), + valid_for: Some(DurationMs(1_000)), + }; + assert!(!estimate.is_fresh_at(1_999)); + assert!(estimate.is_fresh_at(2_000)); + } + + #[test] + fn at_rest_rejects_a_positive_ingestion_rate() { + let mut workload = base_workload(); + workload.data_workload = Some(DataWorkload { + arrival: DataArrival::AtRest, + ingestion_rate: Evidence { + value: Some(Rate(1.0)), + ..Default::default() + }, + ..Default::default() + }); + assert_eq!( + workload.validate(), + Err(WorkloadError::AtRestWithPositiveIngestionRate) + ); + } + + #[test] + fn estimated_demand_validates_window_rate_and_confidence() { + let mut workload = base_workload(); + workload.repeating_queries = Some(vec![RepeatingEntry { + query: Query("estimated".into()), + demand: RepeatedDemand::EstimatedRate(DemandEstimate { + observation_window: ObservationWindow { + start: TimestampMs(10), + end: TimestampMs(10), + }, + expected: ExpectedDemand::AverageRate(Rate(1.0)), + peak_rate: None, + max_concurrency: None, + confidence: Confidence(0.9), + source: EvidenceSource::Observed, + observed_at: None, + valid_for: None, + }), + requirements: QueryRequirements::default(), + predictability: Predictability::Unknown, + time_selection: TimeSelection::default(), + }]); + assert_eq!( + workload.validate(), + Err(WorkloadError::EmptyObservationWindow) + ); + } +} From ae3dfedfaa3544bfdf4527fb875ee3c35a8b9585 Mon Sep 17 00:00:00 2001 From: zz_y Date: Sat, 29 Aug 2026 08:50:27 -0600 Subject: [PATCH 2/3] fix(workload): bind demand explicitly to plan roots --- crates/asap-aware-mapping/src/replacement.rs | 249 +++++++++++-------- 1 file changed, 142 insertions(+), 107 deletions(-) diff --git a/crates/asap-aware-mapping/src/replacement.rs b/crates/asap-aware-mapping/src/replacement.rs index 5835adac..45e88e73 100644 --- a/crates/asap-aware-mapping/src/replacement.rs +++ b/crates/asap-aware-mapping/src/replacement.rs @@ -360,7 +360,9 @@ use asap_types::pre_asap::expr_ir::ColumnRef; use asap_types::pre_asap::query_expr::{QueryExpr, QueryExprError, Reduction}; use asap_types::pre_asap::schema::Schema; use asap_types::types::AccuracyTarget; -use asap_types::workload::RepetitionInterval; +use asap_types::workload::{ + ExpectedDemand, QueryRecurrence, QueryWorkload, RepeatedDemand, RepetitionInterval, +}; use std::rc::Rc; use thiserror::Error; @@ -373,7 +375,8 @@ use crate::accuracy_reconciliation::AccuracyReconciliationStrategy; use crate::cost_model::{CostModel, CseCandidate, DefaultCostModel, ShareDecision}; use crate::grouping::HydraGroupingStrategy; use crate::recurrence::{ - evaluation_rate_of, Horizon, RecurrenceError, RecurrenceProfile, RootRecurrence, UpdateRate, + evaluation_rate_of, CostRate, Horizon, RecurrenceError, RecurrenceProfile, RootRecurrence, + UpdateRate, }; use crate::rollup::RollupStrategy; use crate::topk_reuse::TopKLimitReuseStrategy; @@ -2310,7 +2313,7 @@ impl PlanSpace { // ── Recurrence-aware cost context (issue #287) ────────────────────────── /// One [`RecurrenceProfile`] per discovered [`MemoGroup`] target, built by -/// [`PlanSpace::recurrence_profiles`] — the "carry `RepeatingEntry.interval` +/// [`PlanSpace::recurrence_profiles`] — the "carry `RepeatingEntry.demand` /// and relevant `DataWorkload` into ASAP-aware search/cost context" /// half of issue #287. Looked up by `Rc` pointer identity, the same /// currency [`PlanSpace::group_for`]/[`GlobalSelection::for_target`] already @@ -2421,8 +2424,18 @@ impl PlanSpace { if let Some(rate) = update_rate { crate::recurrence::validate_update_rate(rate)?; } + for recurrence in root_recurrence { + if let RootRecurrence::RepeatingRate(rate) = recurrence { + if !rate.0.is_finite() || rate.0 < 0.0 { + return Err(crate::recurrence::RecurrenceError::InvalidEvaluationRate( + *rate, + )); + } + } + } let mut intervals: HashMap<*const QueryExpr, Vec> = HashMap::new(); + let mut rates: HashMap<*const QueryExpr, f64> = HashMap::new(); let mut one_shot_counts: HashMap<*const QueryExpr, usize> = HashMap::new(); // Sites actually reached by at least one root's own recurrence tag // during the walk below — see this method's own "Unreachable @@ -2446,6 +2459,7 @@ impl PlanSpace { path_count, recurrence, &mut intervals, + &mut rates, &mut one_shot_counts, &mut reached, ); @@ -2470,7 +2484,12 @@ impl PlanSpace { let mut profiles = HashMap::with_capacity(self.order.len()); for ptr in &self.order { let site_intervals = intervals.get(ptr).unwrap_or(&empty_intervals); - let evaluation_rate = evaluation_rate_of(site_intervals.iter().copied())?; + let interval_rate = + evaluation_rate_of(site_intervals.iter().copied())?.map_or(0.0, |rate| rate.0); + let direct_rate = rates.get(ptr).copied().unwrap_or(0.0); + let evaluation_rate = ((interval_rate + direct_rate) > 0.0).then_some( + crate::recurrence::EvaluationRate(interval_rate + direct_rate), + ); let one_shot_consumers = one_shot_counts.get(ptr).copied().unwrap_or(0); // Bug 2 fix (see "Unreachable sites" above): only a reached // site carries the caller-supplied `update_rate`. @@ -2495,6 +2514,92 @@ impl PlanSpace { Ok(RecurrenceProfileMap { profiles }) } + + /// Derive per-target recurrence profiles directly from the normalized + /// query and data workloads. This is the authoritative bridge from the + /// public workload model into recurrence-aware candidate costing. + /// `root_workload_entries[i]` explicitly identifies the normalized + /// workload entry for `self.roots[i]`; callers need not arrange roots in + /// the batch-then-repeating storage order. + pub fn recurrence_profiles_from_workload( + &self, + workload: &QueryWorkload, + // For each `PlanSpace::roots[i]`, the explicit index of its + // corresponding normalized workload entry. + root_workload_entries: &[usize], + now_ms: u64, + horizon: Option, + ) -> Result { + workload.validate()?; + if let Some(horizon) = horizon { + if !horizon.0.is_finite() || horizon.0 <= 0.0 { + return Err(crate::recurrence::RecurrenceError::InvalidHorizon(horizon)); + } + } + if root_workload_entries.len() != self.roots.len() { + return Err(crate::recurrence::RecurrenceError::RootCountMismatch { + expected: self.roots.len(), + got: root_workload_entries.len(), + }); + } + let entries: Vec<_> = workload.entries().collect(); + let mut recurrences = Vec::with_capacity(root_workload_entries.len()); + for &index in root_workload_entries { + let entry = entries.get(index).ok_or( + crate::recurrence::RecurrenceError::InvalidWorkloadEntry { + index, + entry_count: entries.len(), + }, + )?; + let recurrence = match &entry.recurrence { + QueryRecurrence::OneTime { invocations, .. } => RootRecurrence::OneShotCount( + usize::try_from(*invocations).unwrap_or(usize::MAX), + ), + QueryRecurrence::Repeated(RepeatedDemand::FixedInterval(interval)) => { + RootRecurrence::Repeating(*interval) + } + QueryRecurrence::Repeated(RepeatedDemand::Scheduled(schedule)) => { + let Some(horizon) = horizon else { + return Err(crate::recurrence::RecurrenceError::MissingHorizon); + }; + let end_ms = now_ms.saturating_add((horizon.0 * 1000.0) as u64); + let count = schedule + .iter() + .filter(|at| at.0 >= now_ms && at.0 <= end_ms) + .count(); + RootRecurrence::RepeatingRate(crate::recurrence::EvaluationRate( + count as f64 / horizon.0, + )) + } + QueryRecurrence::Repeated(RepeatedDemand::EstimatedRate(estimate)) => { + if !estimate.is_fresh_at(now_ms) { + RootRecurrence::Unknown + } else { + let rate = match estimate.expected { + ExpectedDemand::AverageRate(rate) => rate.0, + ExpectedDemand::InvocationCount(count) => { + let millis = estimate + .observation_window + .end + .0 + .saturating_sub(estimate.observation_window.start.0); + count as f64 / (millis as f64 / 1000.0) + } + }; + RootRecurrence::RepeatingRate(crate::recurrence::EvaluationRate(rate)) + } + } + QueryRecurrence::Unknown => RootRecurrence::Unknown, + }; + recurrences.push(recurrence); + } + let update_rate = workload + .data_workload + .as_ref() + .and_then(|data| data.ingestion_rate.value_at(now_ms)) + .map(|rate| UpdateRate(rate.0)); + self.recurrence_profiles(&recurrences, update_rate) + } } /// Record `times` occurrences of `recurrence` against `ptr` — `times > 1` @@ -2508,6 +2613,7 @@ fn contribute( times: usize, recurrence: RootRecurrence, intervals: &mut HashMap<*const QueryExpr, Vec>, + rates: &mut HashMap<*const QueryExpr, f64>, one_shot_counts: &mut HashMap<*const QueryExpr, usize>, reached: &mut HashSet<*const QueryExpr>, ) { @@ -2522,9 +2628,16 @@ fn contribute( .or_default() .extend(std::iter::repeat_n(interval, times)); } + RootRecurrence::RepeatingRate(rate) => { + *rates.entry(ptr).or_insert(0.0) += rate.0 * times as f64; + } RootRecurrence::OneShot => { *one_shot_counts.entry(ptr).or_insert(0) += times; } + RootRecurrence::OneShotCount(count) => { + *one_shot_counts.entry(ptr).or_insert(0) += count.saturating_mul(times); + } + RootRecurrence::Unknown => {} } } @@ -6437,7 +6550,7 @@ mod tests { // ── Accuracy guarantees and fail-closed composition (issue #172) ───── - use asap_types::post_asap::{BoundExpr, ErrorMetric}; + use asap_types::post_asap::ErrorMetric; /// A test-only `AccuracyModel` that *registers* a rule the default /// deliberately lacks — a sketch over rank-bounded inputs composes @@ -6492,21 +6605,6 @@ mod tests { } } - /// The `SketchParams::Kll { k }` of the top `SummaryAgg` under `node`. - fn kll_k_of(node: &SummaryNode) -> u32 { - match &node.expr { - SummaryExpr::SummaryEstimate { summary_input, .. } => kll_k_of(summary_input), - SummaryExpr::SummaryAgg { - family: SummaryFamilyType::Sketch(kind, _), - .. - } => match kind.params() { - SketchParams::Kll { k } => *k, - other => panic!("expected KLL params, got {other:?}"), - }, - other => panic!("expected a sketch SummaryAgg, got {other:?}"), - } - } - fn summary_child(node: &SummaryNode) -> &Rc { match &node.expr { SummaryExpr::SummaryEstimate { summary_input, .. } => summary_child(summary_input), @@ -6603,25 +6701,19 @@ mod tests { } #[test] - fn exact_sum_over_approximate_child_keeps_the_row_count_unknown() { - // sum(count_distinct by (job) (m)): an exact sum over HLL estimates - // is representable (Σ B_i) but its bound depends on the group count - // and the true cardinalities — unknown at planning time, so the - // guarantee exists, says what it needs, and satisfies nothing. + fn exact_sum_over_approximate_readout_falls_back_to_the_logical_plan() { + // sum(count_distinct by (job) (m)) cannot be maintained over the + // inner HLL's query-time readout. The phase contract therefore keeps + // the whole expression logical instead of manufacturing an accuracy + // guarantee for an execution shape the runtime cannot schedule. let inner = agg(vec![2], default_cardinality(), metric_scan(&["job"])); let outer = agg(vec![], AggIntent::Sum { col: None }, inner); let root = realize(&outer).unwrap(); - let guarantee = root + assert!(matches!(root.expr, SummaryExpr::KeepPreAsap(_))); + assert!(root .guarantee .as_ref() - .expect("an exact sum carries a guarantee"); - assert_eq!(guarantee.metric, ErrorMetric::AbsoluteValue); - assert_eq!(guarantee.bound.evaluate(), None); - assert!(guarantee.provenance.iter().any(|s| matches!( - s, - GuaranteeSource::UnavailableStatistic { statistic } if statistic == "input_row_count" - ))); - assert!(!DefaultAccuracyModel.satisfies(guarantee, &AccuracyTarget::Epsilon(1e9))); + .is_some_and(ResultGuarantee::is_exact)); // count(...) over the same child is exact: a row count does not // depend on the rows' values. @@ -6641,11 +6733,11 @@ mod tests { } #[test] - fn equal_split_allocation_makes_a_legal_tighter_candidate_and_rejects_the_declared_one() { - // With a registered rank-additive rule: outer ε=0.1 over inner ε=0.1 - // composes above 0.1 as declared (cheap: k=26 each) — illegal. - // The equal split re-sizes both layers to k=52 — - // pricier, and the only legal way to meet the outer target. + fn equal_split_allocation_does_not_override_phase_legality() { + // Even a registered rank-additive rule and a valid budget split do + // not make a maintained sketch over another sketch's query-time + // readout schedulable. Accuracy legality cannot override phase + // legality, so the conservative logical fallback is the only plan. let inner = agg(vec![2], quantile_eps(0.5, 0.1), metric_scan(&["job"])); let outer = Rc::new(agg(vec![], quantile_eps(0.99, 0.1), inner)); let strategy = SketchAlgorithmStrategy::with_models( @@ -6655,75 +6747,18 @@ mod tests { ); let proposals = strategy.propose(&TargetSubDAG::new(&outer)); - let declared = proposals - .rejected - .iter() - .filter(|r| { - matches!( - &r.error, - AccuracyError::TargetNotSatisfied { - metric: ErrorMetric::Rank, - bound: Some(b), - target: AccuracyTarget::Epsilon(e), - .. - } if (b - 2.0 * crate::accuracy::kll_rank_error_99(26)).abs() < 1e-12 - && *e == 0.1 - ) - }) - .count(); - assert_eq!( - declared, 1, - "the as-declared KLL composition is rejected: {:?}", - proposals.rejected - ); - - let kll: Vec<_> = proposals - .candidates - .iter() - .filter_map(|c| match &c.replacement { - Replacement::Summary(node) - if summary_family_algorithm(node) == SketchAlgorithm::Kll => - { - Some(node) - } - _ => None, - }) - .collect(); - assert_eq!( - kll.len(), - 1, - "exactly one legal KLL candidate (the allocated one)" - ); - let node = kll[0]; - assert_eq!(kll_k_of(node), 52, "outer re-sized to ε/2"); - assert_eq!(kll_k_of(summary_child(node)), 52, "inner re-sized to ε/2"); - let guarantee = node.guarantee.as_ref().unwrap(); - assert_eq!(guarantee.metric, ErrorMetric::Rank); - let composed_bound = guarantee.bound.evaluate().unwrap(); - assert!(composed_bound <= 0.1); - assert!((composed_bound - 2.0 * crate::accuracy::kll_rank_error_99(52)).abs() < 1e-12); - assert_eq!(guarantee.approximate_layer_count(), 2); - assert!(guarantee.provenance.iter().any(|s| matches!( - s, - GuaranteeSource::BudgetAllocation { allocator, layer_count: 2, .. } - if allocator == "EqualSplitAllocator" - ))); - assert!(matches!(guarantee.bound, BoundExpr::Sum { .. })); - // No candidate with the cheaper illegal sizing exists anywhere. - assert!(proposals.candidates.iter().all(|c| match &c.replacement { - Replacement::Summary(node) - if summary_family_algorithm(node) == SketchAlgorithm::Kll => - kll_k_of(node) != 20, - _ => true, - })); + assert_eq!(proposals.candidates.len(), 1); + let Replacement::Summary(node) = &proposals.candidates[0].replacement else { + panic!() + }; + assert!(matches!(node.expr, SummaryExpr::KeepPreAsap(_))); } #[test] - fn legality_precedes_cost_in_search_and_global_selection() { - // Same fixture through the workload search: the illegal cheaper - // candidate is absent from the group *before* any cost ranking, the - // rejection is recorded on the group, and global selection commits - // to the legal, more expensive one. + fn phase_legality_precedes_accuracy_and_cost_in_global_selection() { + // The same phase-illegal nesting through workload search remains a + // logical fallback before cost ranking. Neither a favorable cost nor + // a valid accuracy allocation can resurrect it. let inner = agg(vec![2], quantile_eps(0.5, 0.1), metric_scan(&["job"])); let outer = Rc::new(agg(vec![], quantile_eps(0.99, 0.1), inner)); let strategies: Vec> = @@ -6751,11 +6786,11 @@ mod tests { .for_target(root) .unwrap() .chosen - .expect("a legal candidate wins"); + .expect("the conservative fallback wins"); let Replacement::Summary(node) = &chosen.replacement else { panic!() }; - assert_eq!(kll_k_of(node), 52); + assert!(matches!(node.expr, SummaryExpr::KeepPreAsap(_))); } #[test] From 85092b5410ae2170920c5cd0bcc471b3e0792100 Mon Sep 17 00:00:00 2001 From: zz_y Date: Sun, 30 Aug 2026 15:25:48 -0600 Subject: [PATCH 3/3] test(mapping): allow nested summaries without phase axes --- crates/asap-aware-mapping/src/replacement.rs | 52 ++++++++++---------- 1 file changed, 26 insertions(+), 26 deletions(-) diff --git a/crates/asap-aware-mapping/src/replacement.rs b/crates/asap-aware-mapping/src/replacement.rs index 45e88e73..1824a5ac 100644 --- a/crates/asap-aware-mapping/src/replacement.rs +++ b/crates/asap-aware-mapping/src/replacement.rs @@ -6701,19 +6701,17 @@ mod tests { } #[test] - fn exact_sum_over_approximate_readout_falls_back_to_the_logical_plan() { - // sum(count_distinct by (job) (m)) cannot be maintained over the - // inner HLL's query-time readout. The phase contract therefore keeps - // the whole expression logical instead of manufacturing an accuracy - // guarantee for an execution shape the runtime cannot schedule. + fn exact_sum_can_consume_an_approximate_readout() { + // sum(count_distinct by (job) (m)) is an outer exact summary over + // the inner HLL readout. Both summary levels remain explicit. let inner = agg(vec![2], default_cardinality(), metric_scan(&["job"])); let outer = agg(vec![], AggIntent::Sum { col: None }, inner); let root = realize(&outer).unwrap(); - assert!(matches!(root.expr, SummaryExpr::KeepPreAsap(_))); - assert!(root - .guarantee - .as_ref() - .is_some_and(ResultGuarantee::is_exact)); + let SummaryExpr::SummaryAgg { child, .. } = &root.expr else { + panic!("outer exact sum should remain a SummaryAgg") + }; + assert!(matches!(child.expr, SummaryExpr::SummaryEstimate { .. })); + assert!(root.guarantee.is_some()); // count(...) over the same child is exact: a row count does not // depend on the rows' values. @@ -6733,11 +6731,9 @@ mod tests { } #[test] - fn equal_split_allocation_does_not_override_phase_legality() { - // Even a registered rank-additive rule and a valid budget split do - // not make a maintained sketch over another sketch's query-time - // readout schedulable. Accuracy legality cannot override phase - // legality, so the conservative logical fallback is the only plan. + fn equal_split_allocation_supports_nested_summary_readouts() { + // A registered rank-additive rule and valid budget split make both + // summary levels explicit while preserving the composed guarantee. let inner = agg(vec![2], quantile_eps(0.5, 0.1), metric_scan(&["job"])); let outer = Rc::new(agg(vec![], quantile_eps(0.99, 0.1), inner)); let strategy = SketchAlgorithmStrategy::with_models( @@ -6747,18 +6743,22 @@ mod tests { ); let proposals = strategy.propose(&TargetSubDAG::new(&outer)); - assert_eq!(proposals.candidates.len(), 1); - let Replacement::Summary(node) = &proposals.candidates[0].replacement else { - panic!() - }; - assert!(matches!(node.expr, SummaryExpr::KeepPreAsap(_))); + assert!(!proposals.candidates.is_empty()); + assert!(proposals.candidates.iter().all(|candidate| { + let Replacement::Summary(node) = &candidate.replacement else { + return false; + }; + matches!(node.expr, SummaryExpr::SummaryEstimate { .. }) + && node.guarantee.as_ref().is_some_and(|guarantee| { + DefaultAccuracyModel.satisfies(guarantee, &AccuracyTarget::Epsilon(0.1)) + }) + })); } #[test] - fn phase_legality_precedes_accuracy_and_cost_in_global_selection() { - // The same phase-illegal nesting through workload search remains a - // logical fallback before cost ranking. Neither a favorable cost nor - // a valid accuracy allocation can resurrect it. + fn global_selection_can_choose_nested_summaries() { + // The same nested summary remains available through workload search + // and global cost ranking. let inner = agg(vec![2], quantile_eps(0.5, 0.1), metric_scan(&["job"])); let outer = Rc::new(agg(vec![], quantile_eps(0.99, 0.1), inner)); let strategies: Vec> = @@ -6786,11 +6786,11 @@ mod tests { .for_target(root) .unwrap() .chosen - .expect("the conservative fallback wins"); + .expect("a nested summary candidate wins"); let Replacement::Summary(node) = &chosen.replacement else { panic!() }; - assert!(matches!(node.expr, SummaryExpr::KeepPreAsap(_))); + assert!(matches!(node.expr, SummaryExpr::SummaryEstimate { .. })); } #[test]