diff --git a/crates/asap-aware-mapping/src/accuracy/estimators/cms.rs b/crates/asap-aware-mapping/src/accuracy/estimators/cms.rs index 9482c539a..a3767fe52 100644 --- a/crates/asap-aware-mapping/src/accuracy/estimators/cms.rs +++ b/crates/asap-aware-mapping/src/accuracy/estimators/cms.rs @@ -41,7 +41,7 @@ mod tests { #[test] fn local_guarantee_inverts_frequency_sizing() { - use crate::replacement::default_size_params; + use crate::realization::default_size_params; use asap_types::post_asap::{GroupingStrategy, SketchKind}; let c = asap_types::pre_asap::agg_intent::default_cardinality(); let params = default_size_params(SketchAlgorithm::Cms, &c, 0.01, 0.001); diff --git a/crates/asap-aware-mapping/src/accuracy/estimators/count_sketch.rs b/crates/asap-aware-mapping/src/accuracy/estimators/count_sketch.rs index c95059637..a031a6f8e 100644 --- a/crates/asap-aware-mapping/src/accuracy/estimators/count_sketch.rs +++ b/crates/asap-aware-mapping/src/accuracy/estimators/count_sketch.rs @@ -55,7 +55,7 @@ mod tests { #[test] fn count_sketch_uses_an_l2_guarantee() { - use crate::replacement::default_size_params; + use crate::realization::default_size_params; use asap_types::post_asap::{GroupingStrategy, SketchKind}; use asap_types::pre_asap::agg_intent::default_cardinality; let intent = default_cardinality(); diff --git a/crates/asap-aware-mapping/src/accuracy/estimators/hll.rs b/crates/asap-aware-mapping/src/accuracy/estimators/hll.rs index f4556446c..a4a0ffd5c 100644 --- a/crates/asap-aware-mapping/src/accuracy/estimators/hll.rs +++ b/crates/asap-aware-mapping/src/accuracy/estimators/hll.rs @@ -230,7 +230,7 @@ mod tests { } #[test] fn generic_rse_sizing_does_not_certify_confidence() { - use crate::replacement::default_size_params; + use crate::realization::default_size_params; use asap_types::post_asap::{GroupingStrategy, SketchKind}; use asap_types::pre_asap::agg_intent::default_cardinality; let c = default_cardinality(); diff --git a/crates/asap-aware-mapping/src/accuracy/estimators/kll.rs b/crates/asap-aware-mapping/src/accuracy/estimators/kll.rs index 4f880894d..7e06ef63c 100644 --- a/crates/asap-aware-mapping/src/accuracy/estimators/kll.rs +++ b/crates/asap-aware-mapping/src/accuracy/estimators/kll.rs @@ -42,7 +42,7 @@ mod tests { #[test] fn local_guarantee_inverts_rank_sizing() { - use crate::replacement::default_size_params; + use crate::realization::default_size_params; use asap_types::post_asap::{GroupingStrategy, SketchKind}; use asap_types::pre_asap::agg_intent::default_quantile; let q = default_quantile(0.99); diff --git a/crates/asap-aware-mapping/src/accuracy/estimators/mod.rs b/crates/asap-aware-mapping/src/accuracy/estimators/mod.rs index 0cd5fcb4f..115f550c0 100644 --- a/crates/asap-aware-mapping/src/accuracy/estimators/mod.rs +++ b/crates/asap-aware-mapping/src/accuracy/estimators/mod.rs @@ -157,7 +157,7 @@ impl<'a> EstimatorAccuracy<'a> { target: Option<&AccuracyTarget>, ) -> Self { let (epsilon, delta) = target - .map(crate::replacement::accuracy_budget) + .map(crate::realization::accuracy_budget) .unwrap_or((0.0, 0.0)); Self { base, diff --git a/crates/asap-aware-mapping/src/accuracy/mod.rs b/crates/asap-aware-mapping/src/accuracy/mod.rs index a2f3354c0..47ae62b1d 100644 --- a/crates/asap-aware-mapping/src/accuracy/mod.rs +++ b/crates/asap-aware-mapping/src/accuracy/mod.rs @@ -45,7 +45,7 @@ pub trait AccuracyModel { /// The guarantee of reading `query` out of a summary of family `family` /// built over an **exact** input — derived from the family's committed /// parameters by inverting the same sizing formulas - /// [`crate::replacement::default_size_params`] uses. `None` when this + /// [`crate::realization::default_size_params`] uses. `None` when this /// model has no error model for the family (the default has none for /// `Sample`/`Wavelet`/`StatModel`). fn local_guarantee( diff --git a/crates/asap-aware-mapping/src/accuracy/reconciliation.rs b/crates/asap-aware-mapping/src/accuracy/reconciliation.rs index 9d4722a6c..bbeaddf24 100644 --- a/crates/asap-aware-mapping/src/accuracy/reconciliation.rs +++ b/crates/asap-aware-mapping/src/accuracy/reconciliation.rs @@ -34,7 +34,7 @@ //! 1. Both are the same bindable shape [`crate::replacement::ASAPStrategies`] //! itself targets — a single measure, no `HAVING` (`bindable_intent`'s own //! scope) — **and** that one measure is one of the four accuracy-bearing -//! [`AggIntent`] variants ([`crate::replacement::accuracy_target`]'s own +//! [`AggIntent`] variants ([`crate::realization::accuracy_target`]'s own //! scope: `Count` / `Quantile` / `Cardinality` / `TopK`). Every other //! intent has no `AccuracyTarget` to reconcile in the first place. //! 2. Same `reduction` (grouping), same `output_names`, and the same shared @@ -61,7 +61,7 @@ //! //! ## Safety of tightening: why reading the tighter build is always sound //! -//! [`crate::replacement::accuracy_budget`] resolves *every* `AccuracyTarget` +//! [`crate::realization::accuracy_budget`] resolves *every* `AccuracyTarget` //! (`Epsilon`/`EpsilonDelta`) to the literal `(eps, delta)` pair //! `realizations_for_intent`'s `sketch_realizations` feeds into //! `CostModel::size_params` — the same numbers `default_size_params`' @@ -158,9 +158,9 @@ use asap_types::ir::{NonASAPOp, OperatorNode}; use asap_types::pre_asap::agg_intent::AggIntent; use asap_types::types::AccuracyTarget; +use crate::realization::{accuracy_budget, accuracy_target}; use crate::replacement::{ - accuracy_budget, accuracy_target, Replacement, ReplacementProvenance, ReplacementStrategy, - ReplacementSubDAG, TargetSubDAG, + Replacement, ReplacementProvenance, ReplacementStrategy, ReplacementSubDAG, TargetSubDAG, }; /// `bindable_accuracy_aggregate`'s return shape: `(reduction, intent, @@ -179,7 +179,7 @@ type BindableAccuracyAggregate<'a> = ( /// [`crate::replacement::ASAPStrategies`] targets (see that /// module's private `bindable_intent`), further narrowed to a measure whose /// intent actually carries an [`AccuracyTarget`] -/// ([`crate::replacement::accuracy_target`]'s own scope: `Count` / +/// ([`crate::realization::accuracy_target`]'s own scope: `Count` / /// `Quantile` / `Cardinality` / `TopK`). `None` for anything else, including /// a multi-measure or `HAVING` aggregate, a non-`Aggregate` node, or an /// accuracy-free intent (`Sum`, `Avg`, …). @@ -534,7 +534,7 @@ mod tests { let epsilon_only = AccuracyTarget::Epsilon(0.01); let equivalent_epsilon_delta = AccuracyTarget::EpsilonDelta { epsilon: 0.01, - delta: crate::replacement::DEFAULT_DELTA, + delta: crate::realization::DEFAULT_DELTA, }; assert!(dominates(&epsilon_only, &equivalent_epsilon_delta)); assert!(dominates(&equivalent_epsilon_delta, &epsilon_only)); diff --git a/crates/asap-aware-mapping/src/cost_model.rs b/crates/asap-aware-mapping/src/cost_model.rs index 9589d1382..d46b2c0cd 100644 --- a/crates/asap-aware-mapping/src/cost_model.rs +++ b/crates/asap-aware-mapping/src/cost_model.rs @@ -377,7 +377,7 @@ pub fn default_cse_shared_maintenance_cost(family: &FieldDataType) -> Cost { /// Ranks the candidate sketch algorithms for one [`AggIntent`], best choice /// first. /// -/// [`replacement::summary_candidates`] returns every algorithm that *can* answer an +/// [`realization::summary_candidates`] returns every algorithm that *can* answer an /// intent, in an arbitrary static preference order (issue #98's "one home" /// for the candidate set). A `CostModel` re-orders that list under real, /// deployment-specific cost knowledge this crate has no way to know about — @@ -416,7 +416,7 @@ pub trait CostModel { } /// Rank `candidates` (as returned by - /// [`summary_candidates`](crate::replacement::summary_candidates)) for + /// [`summary_candidates`](crate::realization::summary_candidates)) for /// `intent`, best choice first. /// /// Implementations MAY reorder freely, but MUST return exactly the input @@ -442,7 +442,7 @@ pub trait CostModel { /// discrete rungs required by a downstream catalog) without forking /// `replacement::realizations_for_intent` — the same "one extension /// point" rationale as `rank_candidates`, one level deeper. Default: - /// [`replacement::default_size_params`], `asap-plan`'s built-in formulas + /// [`realization::default_size_params`], `asap-plan`'s built-in formulas /// (unchanged) — a deployment that only needs to reorder candidates, /// not resize them, can leave this method unimplemented. /// @@ -463,7 +463,7 @@ pub trait CostModel { eps: f64, delta: f64, ) -> SketchParams { - crate::replacement::default_size_params(kind, intent, eps, delta) + crate::realization::default_size_params(kind, intent, eps, delta) } /// Estimated number of distinct subpopulations produced by `target`'s @@ -867,9 +867,9 @@ pub(crate) fn validated_candidate_ranking( } /// The default cost model: preserves [`summary_candidates`]'s built-in static -/// order and [`replacement::default_size_params`]'s built-in sizing unchanged. +/// order and [`realization::default_size_params`]'s built-in sizing unchanged. /// -/// [`summary_candidates`]: crate::replacement::summary_candidates +/// [`summary_candidates`]: crate::realization::summary_candidates pub struct DefaultCostModel; impl CostModel for DefaultCostModel { @@ -978,7 +978,7 @@ impl CostModel for DefaultCostModel { #[cfg(test)] mod tests { use super::*; - use crate::replacement::summary_candidates; + use crate::realization::summary_candidates; use asap_types::pre_asap::agg_intent::default_cardinality; #[test] @@ -1063,7 +1063,7 @@ mod tests { let intent = default_cardinality(); assert_eq!( AlwaysPreferLast.size_params(SketchAlgorithm::Hll, &intent, 0.01, 0.01), - crate::replacement::default_size_params(SketchAlgorithm::Hll, &intent, 0.01, 0.01), + crate::realization::default_size_params(SketchAlgorithm::Hll, &intent, 0.01, 0.01), ); } @@ -1094,7 +1094,7 @@ mod tests { let k = if eps >= 0.01 { 200 } else { 2048 }; SketchParams::Kll { k } } - other => crate::replacement::default_size_params(other, intent, eps, delta), + other => crate::realization::default_size_params(other, intent, eps, delta), } } } @@ -1111,7 +1111,7 @@ mod tests { // Untouched kinds still fall through to the default formula. assert_eq!( DiscreteKllRungs.size_params(SketchAlgorithm::Hll, &intent, 0.01, 0.01), - crate::replacement::default_size_params(SketchAlgorithm::Hll, &intent, 0.01, 0.01), + crate::realization::default_size_params(SketchAlgorithm::Hll, &intent, 0.01, 0.01), ); } diff --git a/crates/asap-aware-mapping/src/empirical_comparison.rs b/crates/asap-aware-mapping/src/empirical_comparison.rs deleted file mode 100644 index 17e3e9a82..000000000 --- a/crates/asap-aware-mapping/src/empirical_comparison.rs +++ /dev/null @@ -1,915 +0,0 @@ -//! Query-matched, fixed-snapshot offline recommendations. Observed error is an -//! explicit acceptance criterion, never a replacement for formal guarantees. - -use asap_types::post_asap::{SketchAlgorithm, SketchParams}; -use serde::{Deserialize, Serialize}; - -use crate::empirical_cost::{ - DistributionDescriptor, EmpiricalEvidenceProvider, EnvironmentDescriptor, EvidenceArtifact, - EvidenceContext, Measurement, MeasurementProvenance, OfflineMeasurement, -}; - -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct OfflineQueryDescriptor { - pub kind: String, - pub value_type: String, - /// Identifies the exact probe population used for timing and observed error. - pub probe_set: String, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct MeasurementQueryBinding { - pub record_id: String, - pub query: OfflineQueryDescriptor, -} - -pub use crate::empirical_resources::ExactResourceMeasurements; - -#[derive(Debug, Clone, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct OfflineExactMeasurement { - pub id: String, - pub distribution: DistributionDescriptor, - pub environment: EnvironmentDescriptor, - pub query: OfflineQueryDescriptor, - pub measured_at_unix_seconds: u64, - pub valid_until_unix_seconds: u64, - pub provenance: MeasurementProvenance, - pub metrics: ExactResourceMeasurements, -} - -/// The companion format binds otherwise query-agnostic sketch primitives to -/// their measured evaluation and exact reference. Bindings describe state after -/// ingestion, without merges or intervening updates during the read sequence. -#[derive(Debug, Clone, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct OfflineComparisonEvidence { - pub schema_version: u32, - /// Every CPU phase excludes other phases and destruction of retained state. - pub timing_contract: String, - pub sketch_evidence: EvidenceArtifact, - pub query_bindings: Vec, - pub exact_records: Vec, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct EmpiricalAccuracyRequirement { - pub metric: String, - pub max_observed_mean: f64, - pub minimum_trials: u32, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct OfflineWorkload { - pub input_items_per_state: u64, - /// Reads of the fixed, fully ingested snapshot; no updates between reads. - pub reads_per_state: u64, - pub merges_per_state: u64, - pub state_instances: u64, - pub horizon_seconds: f64, -} - -/// Explicit scalarization: CPU ns × cpu_ns_weight + byte-seconds × -/// retained_byte_seconds_weight. Weights must be nonnegative and not both zero. -#[derive(Debug, Clone, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct ResourceWeights { - pub cpu_ns_weight: f64, - pub retained_byte_seconds_weight: f64, -} - -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct SketchConfiguration { - pub algorithm: SketchAlgorithm, - pub params: SketchParams, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct OfflineComparisonRequest { - pub context: EvidenceContext, - pub exact_environment: EnvironmentDescriptor, - pub query: OfflineQueryDescriptor, - pub accuracy: EmpiricalAccuracyRequirement, - pub workload: OfflineWorkload, - pub weights: ResourceWeights, - /// `Some` restricts selection to these algorithms and configurations at - /// least as large as their formally legal deployment parameters. `None` - /// requests a purely offline recommendation, unsuitable for formal binding. - pub formal_minimums: Option>, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct OfflineResourceEstimate { - pub cpu_ns: f64, - pub retained_bytes: Option, - pub retained_byte_seconds: Option, - /// Conservative sum of per-state construction/ingestion peaks. - pub peak_bytes_upper_bound: Option, - /// Per-state snapshot sizes, not charged as writes in this in-memory model. - pub serialized_bytes_per_state: Option, - pub disk_bytes_per_state: Option, - pub objective_cost: f64, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct OfflineCandidateEstimate { - pub record_id: String, - pub configuration: Option, - pub observed_error_mean: Option, - pub resources: Option, - pub rejection: Option, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct OfflineRecommendation { - pub query: OfflineQueryDescriptor, - pub selected: OfflineCandidateEstimate, - pub exact_baseline: OfflineCandidateEstimate, - pub candidates: Vec, - pub estimated_cpu_savings_ns: f64, - pub estimated_retained_bytes_savings: Option, - /// True only if the request supplied and every selected sketch passed - /// explicit formal minima; observed error alone cannot authorize binding. - pub checked_formal_minimums: bool, - pub assumptions: Vec, -} - -impl OfflineRecommendation { - /// An exact selection deliberately returns no sketch. The caller must - /// preserve its exact path; do not silently substitute a default sketch. - pub fn selected_sketch(&self) -> Option<&SketchConfiguration> { - self.selected.configuration.as_ref() - } -} - -/// Compare complete CPU components for a fixed snapshot and retain every -/// unavailable alternative with its reason. Without an applicable exact -/// baseline no benefit can be established, so this returns an error. -pub fn recommend_offline( - evidence: &OfflineComparisonEvidence, - request: &OfflineComparisonRequest, -) -> Result { - validate_request(request)?; - if evidence.schema_version != 1 { - return Err("unsupported offline comparison schema version".into()); - } - if evidence.timing_contract != "disjoint_live_state_v1" { - return Err( - "comparison requires disjoint CPU phases measured with retained state alive".into(), - ); - } - let provider = - EmpiricalEvidenceProvider::new(evidence.sketch_evidence.clone(), request.context.clone()) - .map_err(|e| e.to_string())?; - let mut bindings = std::collections::HashMap::new(); - for binding in &evidence.query_bindings { - if bindings - .insert(&binding.record_id, &binding.query) - .is_some() - { - return Err("duplicate query binding".into()); - } - if !evidence - .sketch_evidence - .records - .iter() - .any(|r| r.id == binding.record_id) - { - return Err("query binding refers to missing record".into()); - } - } - let exact: Vec<_> = evidence - .exact_records - .iter() - .filter(|r| { - r.distribution == request.context.distribution - && r.environment == request.exact_environment - && r.query == request.query - && r.measured_at_unix_seconds <= request.context.now_unix_seconds - && request.context.now_unix_seconds <= r.valid_until_unix_seconds - }) - .collect(); - if exact.len() != 1 { - return Err("missing, stale, incompatible or ambiguous exact baseline".into()); - } - let exact = exact[0]; - validate_exact(exact)?; - let baseline_resources = estimate_exact(exact, request)?; - let exact_baseline = OfflineCandidateEstimate { - record_id: exact.id.clone(), - configuration: None, - observed_error_mean: Some(0.0), - resources: Some(baseline_resources), - rejection: None, - }; - let mut candidates = Vec::new(); - for row in &evidence.sketch_evidence.records { - let mut candidate = OfflineCandidateEstimate { - record_id: row.id.clone(), - configuration: Some(SketchConfiguration { - algorithm: row.algorithm.clone(), - params: row.params.clone(), - }), - observed_error_mean: row.error.as_ref().and_then(|e| e.mean), - resources: None, - rejection: None, - }; - let result = (|| { - let matched = provider - .lookup(&row.algorithm, &row.params) - .map_err(|e| e.to_string())?; - if matched.id != row.id { - return Err("record belongs to another applicability context".into()); - } - if bindings.get(&row.id).copied() != Some(&request.query) { - return Err("missing or incompatible measured query binding".into()); - } - if let Some(minimums) = &request.formal_minimums { - if !minimums.iter().any(|m| { - m.algorithm == row.algorithm && parameters_at_least(&row.params, &m.params) - }) { - return Err( - "configuration does not meet the deployment's formal minimum".into(), - ); - } - } - let error = row - .error - .as_ref() - .ok_or("missing offline error observation")?; - if error.query.get("kind").and_then(|v| v.as_str()) != Some(request.query.kind.as_str()) - || error.query.get("value_type").and_then(|v| v.as_str()) - != Some(request.query.value_type.as_str()) - { - return Err( - "offline error observation has incompatible evaluation semantics".into(), - ); - } - if error.metric != request.accuracy.metric - || error.trials < request.accuracy.minimum_trials - { - return Err("incompatible error metric or insufficient offline trials".into()); - } - let mean = error.mean.ok_or("missing observed mean error")?; - if mean > request.accuracy.max_observed_mean { - return Err("observed error exceeds explicit offline acceptance budget".into()); - } - estimate_sketch(row, request) - })(); - match result { - Ok(resources) => candidate.resources = Some(resources), - Err(reason) => candidate.rejection = Some(reason), - } - candidates.push(candidate); - } - let mut selected = exact_baseline.clone(); - for candidate in &candidates { - if let Some(resources) = &candidate.resources { - if resources.objective_cost < selected.resources.as_ref().unwrap().objective_cost { - selected = candidate.clone(); - } - } - } - let baseline = exact_baseline.resources.as_ref().unwrap(); - let chosen = selected.resources.as_ref().unwrap(); - Ok(OfflineRecommendation { query: request.query.clone(), - estimated_cpu_savings_ns: baseline.cpu_ns - chosen.cpu_ns, - estimated_retained_bytes_savings: baseline.retained_bytes.zip(chosen.retained_bytes).map(|(a,b)| a-b), - selected, exact_baseline, candidates, checked_formal_minimums: request.formal_minimums.is_some(), - assumptions: vec![ - "Fixed snapshot: construct, ingest all measured input, prepare exact index once, then read without updates or merges".into(), - "Read CPU is the measured average over all distinct keys; individual key latency and error can differ".into(), - "CPU sums disjoint measured construction/update/prepare/read phases while state remains alive; the comparison ends with retained state and excludes retirement".into(), - "No serialization or disk-write CPU is modeled; reported disk/serialized bytes describe one optional persisted snapshot only".into(), - "Offline mean-error acceptance is restricted to the measured input and probe population; it is not a formal or runtime error guarantee".into(), - ] }) -} - -fn validate_request(request: &OfflineComparisonRequest) -> Result<(), String> { - let w = &request.workload; - if request.query.kind != "point_frequency" - || request.query.value_type != "i64" - || request.query.probe_set != "all_distinct_keys" - { - return Err("unsupported offline query contract".into()); - } - if request.accuracy.metric.trim().is_empty() - || !nonnegative(request.accuracy.max_observed_mean) - || request.accuracy.minimum_trials == 0 - { - return Err("invalid empirical accuracy requirement".into()); - } - if w.input_items_per_state != request.context.distribution.sample_count - || w.state_instances == 0 - || !w.horizon_seconds.is_finite() - || w.horizon_seconds <= 0.0 - { - return Err( - "workload must match the measured snapshot and have positive states/horizon".into(), - ); - } - if w.merges_per_state != 0 { - return Err("no post-merge error or exact merge baseline was measured".into()); - } - if !nonnegative(request.weights.cpu_ns_weight) - || !nonnegative(request.weights.retained_byte_seconds_weight) - || request.weights.cpu_ns_weight == 0.0 - && request.weights.retained_byte_seconds_weight == 0.0 - { - return Err("invalid resource objective weights".into()); - } - let a = &request.context.environment; - let b = &request.exact_environment; - if a.cpu != b.cpu || a.os != b.os || a.runtime != b.runtime { - return Err( - "sketch and exact measurements must share hardware, OS and benchmark runtime".into(), - ); - } - Ok(()) -} - -fn validate_exact(row: &OfflineExactMeasurement) -> Result<(), String> { - let p = &row.provenance; - if row.id.trim().is_empty() - || [ - &p.command, - &p.dataset, - &p.source_revision, - &row.environment.id, - &row.environment.implementation, - &row.environment.implementation_version, - ] - .iter() - .any(|s| s.trim().is_empty()) - || p.repetitions == 0 - || row.measured_at_unix_seconds > row.valid_until_unix_seconds - { - return Err("invalid exact baseline provenance".into()); - } - let m = &row.metrics.resources; - for measurement in [ - &m.cpu.build_cpu_ns, - &m.cpu.update_cpu_ns, - &m.cpu.merge_cpu_ns, - &m.cpu.prepare_cpu_ns, - &m.cpu.read_cpu_ns, - &m.retained_memory_bytes, - &m.peak_memory_bytes, - &m.serialized_bytes, - &m.disk_bytes, - &m.scan_bytes, - ] - .into_iter() - .flatten() - { - if !nonnegative(measurement.value) - || measurement.samples == 0 - || measurement.stddev.is_some_and(|s| !nonnegative(s)) - { - return Err("invalid exact baseline measurement".into()); - } - } - Ok(()) -} - -fn charge(measurement: &Option, count: u64, name: &str) -> Result { - if count == 0 { - return Ok(0.0); - } - let value = measurement - .as_ref() - .ok_or_else(|| format!("missing {name}"))? - .value - * count as f64; - if !nonnegative(value) { - return Err(format!("invalid or overflowing {name}")); - } - Ok(value) -} - -fn estimate_sketch( - row: &OfflineMeasurement, - request: &OfflineComparisonRequest, -) -> Result { - let m = &row.metrics.resources; - let w = &request.workload; - let cpu = charge(&m.cpu.build_cpu_ns, 1, "empty sketch construction CPU")? - + charge( - &m.cpu.update_cpu_ns, - w.input_items_per_state, - "sketch update CPU", - )? - + charge( - &m.cpu.read_cpu_ns, - w.reads_per_state, - "query-matched sketch read CPU", - )? - + charge(&m.cpu.merge_cpu_ns, w.merges_per_state, "sketch merge CPU")? - + crate::empirical_cost::snapshot_prepare_cpu(row) - .ok_or("missing or invalid sketch snapshot preparation CPU")?; - estimate_resources( - cpu, - &m.retained_memory_bytes, - &m.peak_memory_bytes, - m.serialized_bytes.as_ref().map(|m| m.value), - m.disk_bytes.as_ref().map(|m| m.value), - request, - ) -} - -fn estimate_exact( - row: &OfflineExactMeasurement, - request: &OfflineComparisonRequest, -) -> Result { - let m = &row.metrics.resources; - let w = &request.workload; - let cpu = charge(&m.cpu.build_cpu_ns, 1, "exact empty construction CPU")? - + charge( - &m.cpu.update_cpu_ns, - w.input_items_per_state, - "exact update CPU", - )? - + charge(&m.cpu.prepare_cpu_ns, 1, "exact snapshot preparation CPU")? - + charge(&m.cpu.read_cpu_ns, w.reads_per_state, "exact read CPU")?; - estimate_resources( - cpu, - &m.retained_memory_bytes, - &m.peak_memory_bytes, - m.serialized_bytes.as_ref().map(|m| m.value), - m.disk_bytes.as_ref().map(|m| m.value), - request, - ) -} - -fn estimate_resources( - per_state_cpu: f64, - retained: &Option, - peak: &Option, - serialized: Option, - disk: Option, - request: &OfflineComparisonRequest, -) -> Result { - let count = request.workload.state_instances as f64; - let cpu_ns = per_state_cpu * count; - let retained_bytes = retained.as_ref().map(|m| m.value * count); - let retained_byte_seconds = retained_bytes.map(|v| v * request.workload.horizon_seconds); - let peak_bytes_upper_bound = peak.as_ref().map(|m| m.value * count); - let memory_cost = if request.weights.retained_byte_seconds_weight == 0.0 { - 0.0 - } else { - retained_byte_seconds.ok_or("missing retained memory for weighted resource objective")? - * request.weights.retained_byte_seconds_weight - }; - let objective_cost = cpu_ns * request.weights.cpu_ns_weight + memory_cost; - if [ - Some(cpu_ns), - retained_bytes, - retained_byte_seconds, - peak_bytes_upper_bound, - Some(objective_cost), - ] - .into_iter() - .flatten() - .any(|v| !nonnegative(v)) - { - return Err("overflowing resource estimate".into()); - } - Ok(OfflineResourceEstimate { - cpu_ns, - retained_bytes, - retained_byte_seconds, - peak_bytes_upper_bound, - serialized_bytes_per_state: serialized, - disk_bytes_per_state: disk, - objective_cost, - }) -} - -fn nonnegative(value: f64) -> bool { - value.is_finite() && value >= 0.0 -} - -/// Conservative componentwise dominance for known planner sizing families. -/// A deployment still checks its own catalog/layout constraints before binding. -pub fn parameters_at_least(candidate: &SketchParams, minimum: &SketchParams) -> bool { - match (candidate, minimum) { - (SketchParams::Cms { width: a, depth: b }, SketchParams::Cms { width: c, depth: d }) - | ( - SketchParams::CountSketch { width: a, depth: b }, - SketchParams::CountSketch { width: c, depth: d }, - ) => a >= c && b >= d, - (SketchParams::Kll { k: a }, SketchParams::Kll { k: b }) - | (SketchParams::Kmv { k: a }, SketchParams::Kmv { k: b }) - | (SketchParams::Theta { k: a }, SketchParams::Theta { k: b }) => a >= b, - (SketchParams::Hll { precision: a }, SketchParams::Hll { precision: b }) => a >= b, - (SketchParams::DDSketch { alpha: a }, SketchParams::DDSketch { alpha: b }) => { - a.is_finite() && *a > 0.0 && a <= b - } - ( - SketchParams::CmsWithHeap { - width: a, - depth: b, - heap_size: c, - }, - SketchParams::CmsWithHeap { - width: d, - depth: e, - heap_size: f, - }, - ) - | ( - SketchParams::CountSketchWithHeap { - width: a, - depth: b, - heap_size: c, - }, - SketchParams::CountSketchWithHeap { - width: d, - depth: e, - heap_size: f, - }, - ) => a >= d && b >= e && c >= f, - _ => false, - } -} - -#[cfg(test)] -mod tests { - use super::*; - - fn m(value: f64) -> Option { - Some(Measurement { - value, - stddev: None, - samples: 3, - method: Some("synthetic test fixture, not measured".into()), - }) - } - - fn fixture() -> (OfflineComparisonEvidence, OfflineComparisonRequest) { - let mut sketches: EvidenceArtifact = serde_json::from_str(include_str!( - "../tests/data/offline-evidence-synthetic.json" - )) - .unwrap(); - let query = OfflineQueryDescriptor { - kind: "point_frequency".into(), - value_type: "i64".into(), - probe_set: "all_distinct_keys".into(), - }; - let row = &mut sketches.records[0]; - row.metrics.resources.cpu.build_cpu_ns = m(1.0); - row.metrics.resources.cpu.update_cpu_ns = m(1.0); - row.metrics.resources.cpu.read_cpu_ns = m(1.0); - row.metrics.resources.retained_memory_bytes = m(100.0); - row.error.as_mut().unwrap().mean = Some(0.5); - row.error.as_mut().unwrap().query = - serde_json::json!({"kind":"point_frequency","value_type":"i64"}); - let mut wide = row.clone(); - wide.id = "synthetic-wide-cms".into(); - wide.params = SketchParams::Cms { - width: 544, - depth: 5, - }; - wide.metrics.resources.cpu.update_cpu_ns = m(2.0); - wide.metrics.resources.retained_memory_bytes = m(200.0); - wide.error.as_mut().unwrap().mean = Some(0.001); - let context = EvidenceContext { - distribution: row.distribution.clone(), - environment: row.environment.clone(), - now_unix_seconds: 150, - }; - let mut exact_environment = row.environment.clone(); - exact_environment.id = "synthetic-exact".into(); - exact_environment.implementation = "synthetic exact baseline".into(); - let exact = OfflineExactMeasurement { - id: "exact".into(), - distribution: row.distribution.clone(), - environment: exact_environment.clone(), - query: query.clone(), - measured_at_unix_seconds: 100, - valid_until_unix_seconds: 200, - provenance: row.provenance.clone(), - metrics: ExactResourceMeasurements { - resources: asap_types::resources::MeasuredResources { - cpu: asap_types::resources::MeasuredCpu { - build_cpu_ns: m(1.0), - update_cpu_ns: m(5.0), - prepare_cpu_ns: m(1000.0), - read_cpu_ns: m(5.0), - ..Default::default() - }, - retained_memory_bytes: m(1000.0), - ..Default::default() - }, - }, - }; - let metric = row.error.as_ref().unwrap().metric.clone(); - sketches.records.push(wide); - let bindings = sketches - .records - .iter() - .map(|r| MeasurementQueryBinding { - record_id: r.id.clone(), - query: query.clone(), - }) - .collect(); - let request = OfflineComparisonRequest { - context, - exact_environment, - query, - accuracy: EmpiricalAccuracyRequirement { - metric, - max_observed_mean: 0.01, - minimum_trials: 1, - }, - workload: OfflineWorkload { - input_items_per_state: 1000, - reads_per_state: 1000, - merges_per_state: 0, - state_instances: 1, - horizon_seconds: 10.0, - }, - weights: ResourceWeights { - cpu_ns_weight: 1.0, - retained_byte_seconds_weight: 0.0, - }, - formal_minimums: None, - }; - ( - OfflineComparisonEvidence { - schema_version: 1, - timing_contract: "disjoint_live_state_v1".into(), - sketch_evidence: sketches, - query_bindings: bindings, - exact_records: vec![exact], - }, - request, - ) - } - - /// Observed acceptance rejects a cheap inaccurate rung and chooses a larger - /// measured configuration, with every CPU component and state counted. - #[test] - fn accuracy_requirement_changes_selected_configuration_and_cost() { - let (evidence, mut request) = fixture(); - request.formal_minimums = Some(vec![SketchConfiguration { - algorithm: SketchAlgorithm::Cms, - params: SketchParams::Cms { - width: 512, - depth: 5, - }, - }]); - let chosen = recommend_offline(&evidence, &request).unwrap(); - assert_eq!(chosen.selected.record_id, "synthetic-wide-cms"); - assert_eq!(chosen.selected.resources.as_ref().unwrap().cpu_ns, 3001.0); - assert_eq!( - chosen.exact_baseline.resources.as_ref().unwrap().cpu_ns, - 11001.0 - ); - assert_eq!(chosen.estimated_cpu_savings_ns, 8000.0); - assert_eq!(chosen.estimated_retained_bytes_savings, Some(800.0)); - assert!(chosen.checked_formal_minimums); - request.formal_minimums = None; - request.accuracy.max_observed_mean = 1.0; - assert_eq!( - recommend_offline(&evidence, &request) - .unwrap() - .selected - .record_id, - "synthetic-test-cms" - ); - } - - /// Optional sketch preparation is charged once; exact preparation never - /// borrows the independent merge measurement, and snapshot bytes survive. - #[test] - fn preparation_and_exact_resource_dimensions_keep_their_meaning() { - let (mut evidence, request) = fixture(); - let old = recommend_offline(&evidence, &request).unwrap(); - evidence.sketch_evidence.records[1] - .metrics - .resources - .cpu - .prepare_cpu_ns = m(37.0); - let exact = &mut evidence.exact_records[0].metrics.resources; - exact.cpu.merge_cpu_ns = m(1e9); - exact.serialized_bytes = m(256.0); - exact.disk_bytes = m(4096.0); - exact.scan_bytes = m(8000.0); - let changed = recommend_offline(&evidence, &request).unwrap(); - assert_eq!( - changed.selected.resources.as_ref().unwrap().cpu_ns, - old.selected.resources.as_ref().unwrap().cpu_ns + 37.0 - ); - let baseline = changed.exact_baseline.resources.unwrap(); - assert_eq!( - baseline.cpu_ns, - old.exact_baseline.resources.unwrap().cpu_ns - ); - assert_eq!(baseline.serialized_bytes_per_state, Some(256.0)); - assert_eq!(baseline.disk_bytes_per_state, Some(4096.0)); - } - - /// Exact observations validate optional dimensions even when the comparison - /// does not execute the corresponding operation. - #[test] - fn optional_exact_dimensions_cannot_hide_invalid_measurements() { - let selectors: [fn( - &mut asap_types::resources::MeasuredResources, - ) -> &mut Option; 4] = [ - |r| &mut r.cpu.merge_cpu_ns, - |r| &mut r.serialized_bytes, - |r| &mut r.disk_bytes, - |r| &mut r.scan_bytes, - ]; - for select in selectors { - let (mut evidence, request) = fixture(); - *select(&mut evidence.exact_records[0].metrics.resources) = m(-1.0); - assert!(recommend_offline(&evidence, &request).is_err()); - } - } - - /// A family outside the established CMS/CountSketch contract cannot gain - /// an apparently cheap comparison by omitting its preparation measurement. - #[test] - fn sketch_comparison_requires_unknown_preparation_phase() { - let (evidence, request) = fixture(); - let mut row = evidence.sketch_evidence.records[1].clone(); - let original_cpu = estimate_sketch(&row, &request).unwrap().cpu_ns; - row.algorithm = SketchAlgorithm::Kll; - row.params = SketchParams::Kll { k: 269 }; - assert!(estimate_sketch(&row, &request) - .unwrap_err() - .contains("preparation CPU")); - row.metrics.resources.cpu.prepare_cpu_ns = m(37.0); - assert_eq!( - estimate_sketch(&row, &request).unwrap().cpu_ns, - original_cpu + 37.0 - ); - } - - /// Missing required measurements and failing observed-error budgets return - /// the applicable exact baseline, never an optimistically free sketch. - #[test] - fn missing_cost_or_failed_error_acceptance_selects_exact() { - let (mut evidence, mut request) = fixture(); - request.accuracy.max_observed_mean = 0.0; - assert!(recommend_offline(&evidence, &request) - .unwrap() - .selected_sketch() - .is_none()); - request.accuracy.max_observed_mean = 0.01; - evidence.sketch_evidence.records[1] - .metrics - .resources - .cpu - .build_cpu_ns = None; - let chosen = recommend_offline(&evidence, &request).unwrap(); - assert!(chosen.selected_sketch().is_none()); - assert!(chosen.candidates[1] - .rejection - .as_ref() - .unwrap() - .contains("construction")); - evidence.exact_records[0] - .metrics - .resources - .cpu - .prepare_cpu_ns = None; - assert!(recommend_offline(&evidence, &request) - .unwrap_err() - .contains("preparation")); - } - - /// Query, environment, snapshot cardinality, metric, trial count and - /// validity are required independently; nearby evidence is not extrapolated. - #[test] - fn applicability_is_checked_before_recommendation() { - let (evidence, request) = fixture(); - for case in ["query", "environment", "cardinality", "expired", "merges"] { - let mut request = request.clone(); - match case { - "query" => request.query.kind = "total_count".into(), - "environment" => request.exact_environment.cpu = "other CPU".into(), - "cardinality" => request.workload.input_items_per_state = 1001, - "expired" => request.context.now_unix_seconds = 201, - "merges" => request.workload.merges_per_state = 1, - _ => unreachable!(), - } - assert!(recommend_offline(&evidence, &request).is_err(), "{case}"); - } - for case in ["metric", "trials", "binding", "evaluation"] { - let mut evidence = evidence.clone(); - let mut request = request.clone(); - match case { - "metric" => request.accuracy.metric = "rank_error".into(), - "trials" => request.accuracy.minimum_trials = 100, - "binding" => evidence.query_bindings.clear(), - "evaluation" => { - for row in &mut evidence.sketch_evidence.records { - row.error.as_mut().unwrap().query["kind"] = - serde_json::json!("total_count"); - } - } - _ => unreachable!(), - } - assert!( - recommend_offline(&evidence, &request) - .unwrap() - .selected_sketch() - .is_none(), - "{case}" - ); - } - } - - /// Resource weights have explicit dimensions; missing memory blocks a - /// memory-weighted objective but does not become a zero-memory estimate. - #[test] - fn resource_objective_and_unknown_memory_are_explicit() { - let (mut evidence, mut request) = fixture(); - evidence.sketch_evidence.records[1] - .metrics - .resources - .retained_memory_bytes = None; - let chosen = recommend_offline(&evidence, &request).unwrap(); - assert!(chosen.selected.resources.unwrap().retained_bytes.is_none()); - request.weights.retained_byte_seconds_weight = 1.0; - assert!(recommend_offline(&evidence, &request) - .unwrap() - .selected_sketch() - .is_none()); - evidence.sketch_evidence.records[1] - .metrics - .resources - .retained_memory_bytes = m(10000.0); - assert!(recommend_offline(&evidence, &request) - .unwrap() - .selected_sketch() - .is_none()); - request.weights.cpu_ns_weight = f64::NAN; - assert!(recommend_offline(&evidence, &request).is_err()); - } - - /// A measured rung below a deployment's formal sizing floor cannot be - /// selected even if its error happened to be zero on the offline input. - #[test] - fn formal_minimums_cannot_be_relaxed_by_observed_accuracy() { - let (evidence, mut request) = fixture(); - request.formal_minimums = Some(vec![SketchConfiguration { - algorithm: SketchAlgorithm::Cms, - params: SketchParams::Cms { - width: 1024, - depth: 5, - }, - }]); - assert!(recommend_offline(&evidence, &request) - .unwrap() - .selected_sketch() - .is_none()); - assert!(!parameters_at_least( - &SketchParams::Cms { - width: 1024, - depth: 4 - }, - &SketchParams::Cms { - width: 512, - depth: 5 - } - )); - assert!(!parameters_at_least( - &SketchParams::CountSketch { - width: 1024, - depth: 5 - }, - &SketchParams::Cms { - width: 512, - depth: 5 - } - )); - } - - /// Consuming-wrapper CPU phases and ambiguous exact generations cannot - /// establish a complete fixed-snapshot comparison. - #[test] - fn comparison_requires_disjoint_phases_and_one_exact_generation() { - let (mut evidence, request) = fixture(); - evidence.timing_contract = "consuming_upstream_wrappers".into(); - assert!(recommend_offline(&evidence, &request) - .unwrap_err() - .contains("disjoint")); - evidence.timing_contract = "disjoint_live_state_v1".into(); - evidence - .exact_records - .push(evidence.exact_records[0].clone()); - assert!(recommend_offline(&evidence, &request) - .unwrap_err() - .contains("ambiguous")); - } -} diff --git a/crates/asap-aware-mapping/src/empirical_cost.rs b/crates/asap-aware-mapping/src/empirical_cost.rs index eb9904f77..a63240355 100644 --- a/crates/asap-aware-mapping/src/empirical_cost.rs +++ b/crates/asap-aware-mapping/src/empirical_cost.rs @@ -7,9 +7,8 @@ use asap_types::pre_asap::AggIntent; use serde::{Deserialize, Serialize}; use crate::cost_model::{CostModel, DefaultCostModel}; -use crate::replacement::{ - accuracy_budget, accuracy_target, default_size_params, ReplacementSubDAG, TargetSubDAG, -}; +use crate::realization::{accuracy_budget, accuracy_target, default_size_params}; +use crate::replacement::{ReplacementSubDAG, TargetSubDAG}; pub const EVIDENCE_SCHEMA_VERSION: u32 = 1; pub const EVIDENCE_MODEL_VERSION: &str = "empirical-update-cpu-v1"; @@ -317,23 +316,6 @@ fn nonnegative(value: f64) -> bool { value.is_finite() && value >= 0.0 } -/// The existing fixed-snapshot CMS/CountSketch contract needs no separate -/// preparation. Other families must measure that phase, including an explicit -/// zero when no preparation is necessary; absence is not free work. -pub(crate) fn snapshot_prepare_cpu(row: &OfflineMeasurement) -> Option { - match &row.metrics.resources.cpu.prepare_cpu_ns { - Some(measurement) => nonnegative(measurement.value).then_some(measurement.value), - None if matches!( - row.algorithm, - SketchAlgorithm::Cms | SketchAlgorithm::CountSketch - ) => - { - Some(0.0) - } - None => None, - } -} - fn validate_context( distribution: &DistributionDescriptor, environment: &EnvironmentDescriptor, @@ -460,25 +442,6 @@ mod tests { } } - /// Only the established frequency-sketch contract can omit preparation. - #[test] - fn unmeasured_preparation_for_other_families_stays_unknown() { - let (mut artifact, _, _) = fixture(); - let row = &mut artifact.records[0]; - row.algorithm = SketchAlgorithm::CountSketch; - assert_eq!(snapshot_prepare_cpu(row), Some(0.0)); - row.algorithm = SketchAlgorithm::Kll; - row.params = SketchParams::Kll { k: 269 }; - assert_eq!(snapshot_prepare_cpu(row), None); - row.metrics.resources.cpu.prepare_cpu_ns = Some(Measurement { - value: 17.0, - stddev: None, - samples: 1, - method: None, - }); - assert_eq!(snapshot_prepare_cpu(row), Some(17.0)); - } - fn fixture() -> (EvidenceArtifact, EvidenceContext, AggIntent) { let distribution = DistributionDescriptor { id: "unit-test-uniform".into(), diff --git a/crates/asap-aware-mapping/src/erp.rs b/crates/asap-aware-mapping/src/erp.rs deleted file mode 100644 index 6b4ad6737..000000000 --- a/crates/asap-aware-mapping/src/erp.rs +++ /dev/null @@ -1,859 +0,0 @@ -//! Distribution-conditioned Error–Resource Profiles (ERP). -//! -//! This is a discrete, auditable profile selector. It does not infer a formal -//! sketch guarantee from benchmark observations and it does not interpolate -//! across distributions, implementations, or parameter points. - -use std::collections::{BTreeMap, HashSet}; - -use serde::{Deserialize, Serialize}; - -pub const ERP_SCHEMA_VERSION: u32 = 1; - -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct ErpResourceProfile { - pub memory_bytes: f64, - pub update_cpu_seconds: f64, - pub merge_cpu_seconds: f64, - pub query_cpu_seconds: f64, -} - -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct ErpRecord { - pub id: String, - pub sketch: String, - pub implementation: String, - pub parameters: serde_json::Value, - /// Opaque but equality-matched sketch-bench workload descriptor. - pub distribution: serde_json::Value, - pub trials: u32, - pub error_metrics: BTreeMap, - pub resources: ErpResourceProfile, -} - -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct ErpArtifact { - pub schema_version: u32, - pub producer_version: String, - pub records: Vec, -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum AccuracyMode { - /// ERP may select a smaller empirical configuration. The result is not a - /// formal guarantee and must be labelled accordingly by the deployment. - Empirical, - /// Same selection as empirical mode, but absence of applicable evidence is - /// reported so the caller can fall back to formal sizing or exact execution. - Hybrid, -} - -#[derive(Debug, Clone)] -pub struct ErpSelectionRequest { - pub distribution: serde_json::Value, - pub implementation: Option, - pub allowed_sketches: Vec, - pub error_metric: String, - pub max_error: f64, - pub min_trials: u32, - pub expected_updates: f64, - pub expected_queries: f64, - pub expected_merges: f64, - pub retention_seconds: f64, - pub cpu_weight: f64, - pub byte_second_weight: f64, - pub mode: AccuracyMode, -} - -#[derive(Debug, Clone, PartialEq)] -pub struct ErpSelection<'a> { - pub record: &'a ErpRecord, - pub observed_error: f64, - pub estimated_cost: f64, - pub accuracy_mode: AccuracyMode, -} - -/// Runtime-observable input shape used to match a benchmark scenario. Input -/// volume is a sufficiency gate, not a distance axis: once the benchmark has -/// enough samples, repeating the same stationary distribution adds little -/// information about sketch error. -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct ErpDataShape { - pub cardinality: u64, - /// Stable distribution family, for example `uniform`, `zipf`, - /// `power_law`, or `empirical`. Different families are never interpolated. - pub family: String, - /// Family-specific numeric parameters. Zipf uses `exponent`; a continuous - /// power law may use `alpha` and `minimum`. Uniform has no parameters. - #[serde(default)] - pub parameters: BTreeMap, - pub benchmark_events: u64, -} - -/// One hypothesis fitted to the same bounded runtime observation. Lower -/// goodness-of-fit is better; confidence is in [0, 1]. Keeping all plausible -/// fits avoids prematurely classifying unknown data as one named family. -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct ErpShapeFit { - pub family: String, - #[serde(default)] - pub parameters: BTreeMap, - pub goodness_of_fit: f64, - pub confidence: f64, -} - -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct ErpShapeObservation { - pub cardinality: u64, - pub observed_events: u64, - pub fits: Vec, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub empirical_fingerprint: Option, -} - -#[derive(Debug, Clone)] -pub struct ErpMultiFitSelectionRequest { - pub selection: ErpSelectionRequest, - pub observed: ErpShapeObservation, - pub minimum_benchmark_events: u64, - pub max_log2_cardinality_distance: f64, - pub max_parameter_distance: f64, - pub max_goodness_of_fit: f64, - pub minimum_confidence: f64, - /// Required confidence separation between the best and second-best fit. - pub minimum_confidence_margin: f64, -} - -#[derive(Debug, Clone)] -pub struct ErpNearestSelectionRequest { - pub selection: ErpSelectionRequest, - pub observed: ErpDataShape, - pub minimum_benchmark_events: u64, - pub max_log2_cardinality_distance: f64, - /// Maximum normalized distance for every common distribution parameter. - pub max_parameter_distance: f64, -} - -#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct ErpWindowWorkload { - pub input_updates: u64, - pub query_executions: u64, - pub panes_per_query: u64, - pub retained_panes: u64, - pub materializations: u64, -} - -#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct ErpOperationCounts { - pub updates: u64, - pub merges: u64, - pub queries: u64, - pub retained_sketches: u64, -} - -#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] -pub enum ErpError { - #[error("unsupported ERP schema version {0}")] - UnsupportedVersion(u32), - #[error("invalid ERP artifact: {0}")] - Invalid(&'static str), - #[error("no applicable ERP configuration satisfies the empirical accuracy requirement")] - NoApplicableConfiguration, -} - -impl ErpArtifact { - pub fn validate(&self) -> Result<(), ErpError> { - if self.schema_version != ERP_SCHEMA_VERSION { - return Err(ErpError::UnsupportedVersion(self.schema_version)); - } - if self.producer_version.trim().is_empty() { - return Err(ErpError::Invalid("missing producer version")); - } - let mut ids = HashSet::new(); - for row in &self.records { - if row.id.trim().is_empty() || !ids.insert(row.id.as_str()) { - return Err(ErpError::Invalid("empty or duplicate record id")); - } - if row.sketch.trim().is_empty() - || row.implementation.trim().is_empty() - || row.trials == 0 - || !row.resources.valid() - || row - .error_metrics - .values() - .any(|value| !value.is_finite() || *value < 0.0) - { - return Err(ErpError::Invalid("invalid profile record")); - } - } - Ok(()) - } - - /// Select the least-cost measured parameter point satisfying the requested - /// empirical error. Distribution and implementation matching are exact by - /// design; a later model may add conservative interpolation explicitly. - pub fn select(&self, request: &ErpSelectionRequest) -> Result, ErpError> { - self.validate()?; - if !request.valid() { - return Err(ErpError::Invalid("invalid selection request")); - } - self.records - .iter() - .filter(|row| row.distribution == request.distribution) - .filter(|row| { - request - .implementation - .as_ref() - .is_none_or(|wanted| &row.implementation == wanted) - }) - .filter(|row| { - request.allowed_sketches.is_empty() - || request - .allowed_sketches - .iter() - .any(|name| name == &row.sketch) - }) - .filter(|row| row.trials >= request.min_trials) - .filter_map(|row| { - let observed_error = *row.error_metrics.get(&request.error_metric)?; - (observed_error <= request.max_error).then(|| ErpSelection { - record: row, - observed_error, - estimated_cost: request.cost(&row.resources), - accuracy_mode: request.mode, - }) - }) - .min_by(|left, right| { - left.estimated_cost - .total_cmp(&right.estimated_cost) - .then_with(|| left.record.id.cmp(&right.record.id)) - }) - .ok_or(ErpError::NoApplicableConfiguration) - } - - /// Select against the nearest compatible measured shape. Shape metadata is - /// read from `distribution.erp_shape`, keeping ERP v1 wire compatibility. - pub fn select_nearest( - &self, - request: &ErpNearestSelectionRequest, - ) -> Result, ErpError> { - self.validate()?; - if !request.selection.valid() || !request.valid() { - return Err(ErpError::Invalid("invalid nearest-shape request")); - } - self.records - .iter() - .filter_map(|row| Some((row, data_shape(&row.distribution)?))) - .filter(|(_, shape)| shape.benchmark_events >= request.minimum_benchmark_events) - .filter_map(|(row, shape)| Some((row, request.distance(shape)?))) - .filter(|(_, distance)| *distance <= 1.0) - .filter(|(row, _)| { - request - .selection - .implementation - .as_ref() - .is_none_or(|wanted| &row.implementation == wanted) - && (request.selection.allowed_sketches.is_empty() - || request - .selection - .allowed_sketches - .iter() - .any(|name| name == &row.sketch)) - && row.trials >= request.selection.min_trials - }) - .filter_map(|(row, distance)| { - let observed_error = *row.error_metrics.get(&request.selection.error_metric)?; - (observed_error <= request.selection.max_error).then_some(( - distance, - ErpSelection { - record: row, - observed_error, - estimated_cost: request.selection.cost(&row.resources), - accuracy_mode: request.selection.mode, - }, - )) - }) - .min_by(|(left_distance, left), (right_distance, right)| { - left_distance - .total_cmp(right_distance) - .then_with(|| left.estimated_cost.total_cmp(&right.estimated_cost)) - .then_with(|| left.record.id.cmp(&right.record.id)) - }) - .map(|(_, selection)| selection) - .ok_or(ErpError::NoApplicableConfiguration) - } - - /// Select using every statistically plausible fit for one observation. - /// Ambiguous and poor fits fail closed so callers can use their - /// theoretical-then-exact fallback policy. - pub fn select_multi_fit( - &self, - request: &ErpMultiFitSelectionRequest, - ) -> Result, ErpError> { - self.validate()?; - if !request.selection.valid() || !request.valid() { - return Err(ErpError::Invalid("invalid multi-fit request")); - } - if let Some(selected) = - request - .observed - .empirical_fingerprint - .as_deref() - .and_then(|wanted| { - self.records - .iter() - .filter(|row| benchmark_fingerprint(&row.distribution) == Some(wanted)) - .filter_map(|row| request.selection.evaluate(row).map(|value| (row, value))) - .min_by(|(left_row, left), (right_row, right)| { - left.estimated_cost - .total_cmp(&right.estimated_cost) - .then_with(|| left_row.id.cmp(&right_row.id)) - }) - .map(|(_, selected)| selected) - }) - { - return Ok(selected); - } - let mut fits = request - .observed - .fits - .iter() - .filter(|fit| { - fit.confidence >= request.minimum_confidence - && fit.goodness_of_fit <= request.max_goodness_of_fit - }) - .collect::>(); - fits.sort_by(|left, right| right.confidence.total_cmp(&left.confidence)); - let Some(best) = fits.first() else { - return Err(ErpError::NoApplicableConfiguration); - }; - if fits.get(1).is_some_and(|second| { - best.confidence - second.confidence < request.minimum_confidence_margin - }) { - return Err(ErpError::NoApplicableConfiguration); - } - // Compare every plausible fit with every compatible benchmark shape. - // Confidence and fit quality contribute to the joint score; neither - // field first collapses the observation to one family. - self.records - .iter() - .filter_map(|row| Some((row, data_shape(&row.distribution)?))) - .filter(|(_, shape)| shape.benchmark_events >= request.minimum_benchmark_events) - .flat_map(|(row, shape)| { - fits.iter().filter_map(move |fit| { - let nearest = ErpNearestSelectionRequest { - selection: request.selection.clone(), - observed: ErpDataShape { - cardinality: request.observed.cardinality, - family: fit.family.clone(), - parameters: fit.parameters.clone(), - benchmark_events: request.observed.observed_events, - }, - minimum_benchmark_events: request.minimum_benchmark_events, - max_log2_cardinality_distance: request.max_log2_cardinality_distance, - max_parameter_distance: request.max_parameter_distance, - }; - let shape_distance = nearest.distance(shape.clone())?; - (shape_distance <= 1.0).then_some((row, fit, shape_distance)) - }) - }) - .filter_map(|(row, fit, shape_distance)| { - let selected = request.selection.evaluate(row)?; - let fit_distance = - fit.goodness_of_fit / request.max_goodness_of_fit.max(f64::EPSILON); - let confidence_distance = 1.0 - fit.confidence; - Some(( - shape_distance.max(fit_distance).max(confidence_distance), - selected, - )) - }) - .min_by(|(left_distance, left), (right_distance, right)| { - left_distance - .total_cmp(right_distance) - .then_with(|| left.estimated_cost.total_cmp(&right.estimated_cost)) - .then_with(|| left.record.id.cmp(&right.record.id)) - }) - .map(|(_, selected)| selected) - .ok_or(ErpError::NoApplicableConfiguration) - } -} - -fn benchmark_fingerprint(distribution: &serde_json::Value) -> Option<&str> { - distribution - .pointer("/erp_shape/empirical_fingerprint") - .or_else(|| distribution.pointer("/workload/external/fingerprint")) - .and_then(serde_json::Value::as_str) -} - -fn data_shape(distribution: &serde_json::Value) -> Option { - serde_json::from_value(distribution.get("erp_shape")?.clone()).ok() -} - -impl ErpNearestSelectionRequest { - fn valid(&self) -> bool { - self.minimum_benchmark_events > 0 - && self.observed.cardinality > 0 - && self.max_log2_cardinality_distance.is_finite() - && self.max_log2_cardinality_distance > 0.0 - && !self.observed.family.trim().is_empty() - && self - .observed - .parameters - .values() - .all(|value| value.is_finite()) - && self.max_parameter_distance.is_finite() - && self.max_parameter_distance > 0.0 - } - - fn distance(&self, candidate: ErpDataShape) -> Option { - if candidate.cardinality == 0 - || candidate.family != self.observed.family - || candidate - .parameters - .keys() - .ne(self.observed.parameters.keys()) - { - return None; - } - let cardinality = ((candidate.cardinality as f64).log2() - - (self.observed.cardinality as f64).log2()) - .abs() - / self.max_log2_cardinality_distance; - let parameters = candidate - .parameters - .iter() - .map(|(name, value)| { - let observed = self.observed.parameters.get(name)?; - (value.is_finite() && observed.is_finite()) - .then_some((value - observed).abs() / self.max_parameter_distance) - }) - .collect::>>()? - .into_iter() - .fold(0.0_f64, f64::max); - Some(cardinality.max(parameters)) - } -} - -impl ErpMultiFitSelectionRequest { - fn valid(&self) -> bool { - self.observed.cardinality > 0 - && self.observed.observed_events > 0 - && (!self.observed.fits.is_empty() || self.observed.empirical_fingerprint.is_some()) - && self.max_goodness_of_fit.is_finite() - && self.max_goodness_of_fit >= 0.0 - && self.minimum_confidence.is_finite() - && (0.0..=1.0).contains(&self.minimum_confidence) - && self.minimum_confidence_margin.is_finite() - && (0.0..=1.0).contains(&self.minimum_confidence_margin) - && self.observed.fits.iter().all(|fit| { - !fit.family.trim().is_empty() - && fit.goodness_of_fit.is_finite() - && fit.goodness_of_fit >= 0.0 - && fit.confidence.is_finite() - && (0.0..=1.0).contains(&fit.confidence) - && fit.parameters.values().all(|value| value.is_finite()) - }) - } -} - -impl ErpOperationCounts { - /// Compose atomic benchmark costs with a pane-based window plan. A query - /// over one pane needs no merge; N panes need N-1 merges. - pub fn from_window(workload: ErpWindowWorkload) -> Self { - Self { - updates: workload - .input_updates - .saturating_mul(workload.materializations), - merges: workload - .query_executions - .saturating_mul(workload.panes_per_query.saturating_sub(1)), - queries: workload.query_executions, - retained_sketches: workload - .retained_panes - .saturating_mul(workload.materializations), - } - } - - pub fn cpu_seconds(self, resources: &ErpResourceProfile) -> f64 { - self.updates as f64 * resources.update_cpu_seconds - + self.merges as f64 * resources.merge_cpu_seconds - + self.queries as f64 * resources.query_cpu_seconds - } -} - -impl ErpResourceProfile { - fn valid(&self) -> bool { - [ - self.memory_bytes, - self.update_cpu_seconds, - self.merge_cpu_seconds, - self.query_cpu_seconds, - ] - .into_iter() - .all(|value| value.is_finite() && value >= 0.0) - } -} - -impl ErpSelectionRequest { - fn evaluate<'a>(&self, row: &'a ErpRecord) -> Option> { - if self - .implementation - .as_ref() - .is_some_and(|wanted| &row.implementation != wanted) - || (!self.allowed_sketches.is_empty() - && !self.allowed_sketches.iter().any(|name| name == &row.sketch)) - || row.trials < self.min_trials - { - return None; - } - let observed_error = *row.error_metrics.get(&self.error_metric)?; - (observed_error <= self.max_error).then(|| ErpSelection { - record: row, - observed_error, - estimated_cost: self.cost(&row.resources), - accuracy_mode: self.mode, - }) - } - - fn valid(&self) -> bool { - !self.error_metric.trim().is_empty() - && self.max_error.is_finite() - && self.max_error >= 0.0 - && self.min_trials > 0 - && [ - self.expected_updates, - self.expected_queries, - self.expected_merges, - self.retention_seconds, - self.cpu_weight, - self.byte_second_weight, - ] - .into_iter() - .all(|value| value.is_finite() && value >= 0.0) - } - - fn cost(&self, resources: &ErpResourceProfile) -> f64 { - self.cpu_weight - * (self.expected_updates * resources.update_cpu_seconds - + self.expected_queries * resources.query_cpu_seconds - + self.expected_merges * resources.merge_cpu_seconds) - + self.byte_second_weight * self.retention_seconds * resources.memory_bytes - } -} - -#[cfg(test)] -mod tests { - use super::*; - - fn row(id: &str, width: u64, error: f64, memory: f64) -> ErpRecord { - ErpRecord { - id: id.into(), - sketch: "cms".into(), - implementation: "oxide".into(), - parameters: serde_json::json!({"width": width, "depth": 3}), - distribution: serde_json::json!({"synthetic":{"description":{"kind":"zipf"}}}), - trials: 20, - error_metrics: BTreeMap::from([("relative_error".into(), error)]), - resources: ErpResourceProfile { - memory_bytes: memory, - update_cpu_seconds: 1e-7, - merge_cpu_seconds: 1e-5, - query_cpu_seconds: 1e-6, - }, - } - } - - fn request() -> ErpSelectionRequest { - ErpSelectionRequest { - distribution: serde_json::json!({"synthetic":{"description":{"kind":"zipf"}}}), - implementation: Some("oxide".into()), - allowed_sketches: vec!["cms".into()], - error_metric: "relative_error".into(), - max_error: 0.01, - min_trials: 10, - expected_updates: 1_000.0, - expected_queries: 100.0, - expected_merges: 0.0, - retention_seconds: 60.0, - cpu_weight: 1.0, - byte_second_weight: 1e-9, - mode: AccuracyMode::Hybrid, - } - } - - #[test] - fn selects_cheapest_applicable_accurate_configuration() { - let artifact = ErpArtifact { - schema_version: ERP_SCHEMA_VERSION, - producer_version: "bench-1".into(), - records: vec![ - row("too-small", 256, 0.02, 6_144.0), - row("winner", 512, 0.009, 12_288.0), - row("overprovisioned", 2048, 0.001, 49_152.0), - ], - }; - let selected = artifact.select(&request()).unwrap(); - assert_eq!(selected.record.id, "winner"); - } - - #[test] - fn distribution_mismatch_fails_closed() { - let artifact = ErpArtifact { - schema_version: ERP_SCHEMA_VERSION, - producer_version: "bench-1".into(), - records: vec![row("cms", 512, 0.009, 12_288.0)], - }; - let mut request = request(); - request.distribution = serde_json::json!({"external":{"dataset":"production"}}); - assert_eq!( - artifact.select(&request), - Err(ErpError::NoApplicableConfiguration) - ); - } - - fn multi_fit_request() -> ErpMultiFitSelectionRequest { - ErpMultiFitSelectionRequest { - selection: request(), - observed: ErpShapeObservation { - cardinality: 1_000, - observed_events: 20_000, - fits: vec![ErpShapeFit { - family: "zipf".into(), - parameters: BTreeMap::from([("exponent".into(), 1.2)]), - goodness_of_fit: 0.03, - confidence: 0.95, - }], - empirical_fingerprint: None, - }, - minimum_benchmark_events: 10_000, - max_log2_cardinality_distance: 1.0, - max_parameter_distance: 0.2, - max_goodness_of_fit: 0.1, - minimum_confidence: 0.8, - minimum_confidence_margin: 0.1, - } - } - - #[test] - fn multi_fit_selects_only_confident_well_fitting_family() { - let mut measured = row("zipf", 512, 0.009, 12_288.0); - measured.distribution = serde_json::json!({"erp_shape": { - "cardinality": 1000, "family": "zipf", - "parameters": {"exponent": 1.22}, "benchmark_events": 20000 - }}); - let artifact = ErpArtifact { - schema_version: ERP_SCHEMA_VERSION, - producer_version: "bench-1".into(), - records: vec![measured], - }; - assert_eq!( - artifact - .select_multi_fit(&multi_fit_request()) - .unwrap() - .record - .id, - "zipf" - ); - } - - #[test] - fn multi_fit_rejects_ambiguous_or_poor_observations() { - let mut measured = row("zipf", 512, 0.009, 12_288.0); - measured.distribution = serde_json::json!({"erp_shape": { - "cardinality": 1000, "family": "zipf", - "parameters": {"exponent": 1.2}, "benchmark_events": 20000 - }}); - let artifact = ErpArtifact { - schema_version: ERP_SCHEMA_VERSION, - producer_version: "bench-1".into(), - records: vec![measured], - }; - let mut ambiguous = multi_fit_request(); - ambiguous.observed.fits.push(ErpShapeFit { - family: "normal".into(), - parameters: BTreeMap::from([("mean".into(), 0.0), ("stddev".into(), 1.0)]), - goodness_of_fit: 0.04, - confidence: 0.90, - }); - assert_eq!( - artifact.select_multi_fit(&ambiguous), - Err(ErpError::NoApplicableConfiguration) - ); - ambiguous.observed.fits.truncate(1); - ambiguous.observed.fits[0].goodness_of_fit = 0.5; - assert_eq!( - artifact.select_multi_fit(&ambiguous), - Err(ErpError::NoApplicableConfiguration) - ); - } - - #[test] - fn multi_fit_jointly_ranks_all_plausible_families() { - let mut distant_high_confidence = row("zipf-distant", 512, 0.009, 1.0); - distant_high_confidence.distribution = serde_json::json!({"erp_shape": { - "cardinality": 1000, "family": "zipf", - "parameters": {"exponent": 1.39}, "benchmark_events": 20000 - }}); - let mut close_lower_confidence = row("normal-close", 512, 0.009, 100.0); - close_lower_confidence.distribution = serde_json::json!({"erp_shape": { - "cardinality": 1000, "family": "normal", - "parameters": {"mean": 4.0, "stddev": 1.0}, "benchmark_events": 20000 - }}); - let artifact = ErpArtifact { - schema_version: ERP_SCHEMA_VERSION, - producer_version: "bench-1".into(), - records: vec![distant_high_confidence, close_lower_confidence], - }; - let mut request = multi_fit_request(); - request.observed.fits.push(ErpShapeFit { - family: "normal".into(), - parameters: BTreeMap::from([("mean".into(), 4.0), ("stddev".into(), 1.0)]), - goodness_of_fit: 0.01, - confidence: 0.82, - }); - request.minimum_confidence_margin = 0.1; - assert_eq!( - artifact.select_multi_fit(&request).unwrap().record.id, - "normal-close" - ); - } - - #[test] - fn exact_empirical_fingerprint_precedes_fits_and_requires_identity() { - let mut exact = row("exact-trace", 512, 0.009, 100.0); - exact.distribution = serde_json::json!({"workload": {"external": { - "dataset": "trace-a", "fingerprint": "sha256:abc" - }}}); - let artifact = ErpArtifact { - schema_version: ERP_SCHEMA_VERSION, - producer_version: "bench-1".into(), - records: vec![exact], - }; - let mut request = multi_fit_request(); - request.observed.empirical_fingerprint = Some("sha256:abc".into()); - request.observed.fits.clear(); - assert_eq!( - artifact.select_multi_fit(&request).unwrap().record.id, - "exact-trace" - ); - request.observed.empirical_fingerprint = Some("sha256:different".into()); - assert_eq!( - artifact.select_multi_fit(&request), - Err(ErpError::NoApplicableConfiguration) - ); - } - - /// Verifies the JSON contract emitted by sketch-bench is directly - /// consumable without a backend-specific translation layer. - #[test] - fn deserializes_sketch_bench_wire_format() { - let json = serde_json::json!({ - "schema_version": 1, - "producer_version": "sketch-bench-rev", - "records": [{ - "id": "erp-0", - "sketch": "cms", - "implementation": "oxide", - "parameters": {"width": 512, "depth": 3}, - "distribution": {"synthetic": {"description": {"kind": "zipf"}}}, - "trials": 20, - "error_metrics": {"relative_error": 0.009}, - "resources": { - "memory_bytes": 12288.0, - "update_cpu_seconds": 1e-7, - "merge_cpu_seconds": 1e-5, - "query_cpu_seconds": 1e-6 - } - }] - }); - let artifact: ErpArtifact = serde_json::from_value(json).unwrap(); - let selected = artifact.select(&request()).unwrap(); - assert_eq!(selected.record.id, "erp-0"); - assert_eq!(selected.accuracy_mode, AccuracyMode::Hybrid); - } - - #[test] - fn nearest_shape_prefers_cardinality_and_skew_then_cost() { - let mut close = row("close", 512, 0.009, 12_288.0); - close.distribution = serde_json::json!({"erp_shape": { - "cardinality": 1000, "family": "zipf", "parameters": {"exponent": 1.2}, "benchmark_events": 100000 - }}); - let mut cheap_but_far = row("far", 256, 0.009, 1.0); - cheap_but_far.distribution = serde_json::json!({"erp_shape": { - "cardinality": 8000, "family": "zipf", "parameters": {"exponent": 1.2}, "benchmark_events": 100000 - }}); - let artifact = ErpArtifact { - schema_version: ERP_SCHEMA_VERSION, - producer_version: "bench-1".into(), - records: vec![cheap_but_far, close], - }; - let selected = artifact - .select_nearest(&ErpNearestSelectionRequest { - selection: request(), - observed: ErpDataShape { - cardinality: 1200, - family: "zipf".into(), - parameters: BTreeMap::from([("exponent".into(), 1.1)]), - benchmark_events: 0, - }, - minimum_benchmark_events: 10_000, - max_log2_cardinality_distance: 4.0, - max_parameter_distance: 0.5, - }) - .unwrap(); - assert_eq!(selected.record.id, "close"); - } - - #[test] - fn nearest_shape_rejects_distribution_family_and_small_benchmarks() { - let mut row = row("uniform", 512, 0.009, 12_288.0); - row.distribution = serde_json::json!({"erp_shape": { - "cardinality": 1000, "family": "uniform", "parameters": {}, "benchmark_events": 999 - }}); - let artifact = ErpArtifact { - schema_version: ERP_SCHEMA_VERSION, - producer_version: "bench-1".into(), - records: vec![row], - }; - let nearest = ErpNearestSelectionRequest { - selection: request(), - observed: ErpDataShape { - cardinality: 1000, - family: "zipf".into(), - parameters: BTreeMap::from([("exponent".into(), 1.0)]), - benchmark_events: 0, - }, - minimum_benchmark_events: 1_000, - max_log2_cardinality_distance: 1.0, - max_parameter_distance: 0.5, - }; - assert_eq!( - artifact.select_nearest(&nearest), - Err(ErpError::NoApplicableConfiguration) - ); - } - - #[test] - fn pane_window_composes_atomic_operation_costs() { - let counts = ErpOperationCounts::from_window(ErpWindowWorkload { - input_updates: 1_000, - query_executions: 10, - panes_per_query: 12, - retained_panes: 24, - materializations: 2, - }); - assert_eq!(counts.updates, 2_000); - assert_eq!(counts.merges, 110); - assert_eq!(counts.queries, 10); - assert_eq!(counts.retained_sketches, 48); - assert!((counts.cpu_seconds(&row("cost", 1, 0.0, 0.0).resources) - 0.00131).abs() < 1e-12); - } -} diff --git a/crates/asap-aware-mapping/src/grouping.rs b/crates/asap-aware-mapping/src/grouping.rs index 497bfa783..e866925ad 100644 --- a/crates/asap-aware-mapping/src/grouping.rs +++ b/crates/asap-aware-mapping/src/grouping.rs @@ -84,10 +84,11 @@ use crate::accuracy::{ AccuracyBudgetAllocator, AccuracyEvidenceProvider, AccuracyModel, PropagationStats, }; use crate::cost_model::{CostModel, DefaultCostModel}; +use crate::realization::{accuracy_target, summary_candidates}; use crate::replacement::{ - accuracy_target, bindable_intent, construct_summary_with, describe_intent, - realizations_for_intent, summary_candidates, CandidatePlanningInputs, Proposals, Realization, - RejectedCandidate, Replacement, ReplacementStrategy, ReplacementSubDAG, TargetSubDAG, + bindable_intent, construct_summary_with, describe_intent, realizations_for_intent, + CandidatePlanningInputs, Proposals, Realization, RejectedCandidate, Replacement, + ReplacementStrategy, ReplacementSubDAG, TargetSubDAG, }; /// Whether `reduction` has a genuine subpopulation concept for diff --git a/crates/asap-aware-mapping/src/lib.rs b/crates/asap-aware-mapping/src/lib.rs index e7f7dabcf..5aec33898 100644 --- a/crates/asap-aware-mapping/src/lib.rs +++ b/crates/asap-aware-mapping/src/lib.rs @@ -148,10 +148,8 @@ pub mod accuracy; pub mod analytical_cost; pub mod cost_model; -pub mod empirical_comparison; pub mod empirical_cost; pub mod empirical_resources; -pub mod erp; pub mod exact_composition; pub mod explanation; mod function_rules; @@ -162,6 +160,7 @@ pub mod physical_handoff_cost; pub mod physical_operator_statistics; pub mod physical_plan_cost_model; pub mod query_physical_lowering; +pub mod realization; pub mod recurrence; pub mod replacement; pub mod rewrite; @@ -192,6 +191,7 @@ pub use pass::{ OptimizeError, PassNameConflict, PassRegistry, PlanOutput, PlanningModels, QueryPlan, Stage1ToStage3Pass, Stage1ToStage3Run, }; +pub use realization::summary_candidates; pub use recurrence::{ evaluation_rate_of, total_cost, update_rate_from_data_workload, CostRate, EvaluationRate, Horizon, RecurrenceCostExplanation, RecurrenceError, RecurrenceProfile, RootRecurrence, @@ -199,12 +199,11 @@ pub use recurrence::{ }; pub use replacement::{ default_strategies, default_strategies_with, search_workload, search_workload_with, - search_workload_with_targets, summary_candidates, ASAPStrategies, CandidateLogicalASAPDAGs, - CompositionDecision, GlobalSelection, Matcher, Proposals, RankedTargetSubDAGCandidates, - Realization, RealizationError, RecurrenceProfileMap, RejectedCandidate, Replacement, - ReplacementProvenance, ReplacementStrategy, ReplacementSubDAG, SearchError, - SharedSubDAGStrategy, TargetSubDAG, TargetSubDAGCandidates, TargetSubDAGSelection, - MAX_SEARCH_ITERATIONS, + search_workload_with_targets, ASAPStrategies, CandidateLogicalASAPDAGs, CompositionDecision, + GlobalSelection, Matcher, Proposals, RankedTargetSubDAGCandidates, Realization, + RealizationError, RecurrenceProfileMap, RejectedCandidate, Replacement, ReplacementProvenance, + ReplacementStrategy, ReplacementSubDAG, SearchError, SharedSubDAGStrategy, TargetSubDAG, + TargetSubDAGCandidates, TargetSubDAGSelection, MAX_SEARCH_ITERATIONS, }; pub use rewrite::{AvgToSumOverCountStrategy, SemanticEquivalentRewriteStrategy}; pub use topk_reuse::TopKLimitReuseStrategy; diff --git a/crates/asap-aware-mapping/src/per_computation_candidates.rs b/crates/asap-aware-mapping/src/per_computation_candidates.rs index c62a0f849..3a83a57b3 100644 --- a/crates/asap-aware-mapping/src/per_computation_candidates.rs +++ b/crates/asap-aware-mapping/src/per_computation_candidates.rs @@ -86,7 +86,7 @@ use asap_types::types::AccuracyTarget; use thiserror::Error; use crate::accuracy::{AccuracyModel, DefaultAccuracyModel}; -use crate::replacement::{ +use crate::realization::{ accuracy_budget, accuracy_target, default_size_params, summary_candidates, }; @@ -1068,7 +1068,7 @@ fn measure_input( AggIntent::Count { .. } if input.result_kind == OperatorResultKind::Relation => { SummaryInputExpr::Column(counted_column(schema, filter)?) } - _ => crate::replacement::summarised_input(intent, schema).map_err(|_| { + _ => crate::realization::summarised_input(intent, schema).map_err(|_| { PerComputationCandidateError::Unsupported("input column outside child schema") })?, }; @@ -1109,7 +1109,7 @@ fn counted_column( .then_some(id) }) }; - let reference = |id: usize| schema.fields.get(id).map(crate::replacement::column_ref); + let reference = |id: usize| schema.fields.get(id).map(crate::realization::column_ref); filter .map_or_else(Vec::new, |filter| not_null(&filter.0)) .into_iter() @@ -1173,7 +1173,7 @@ fn summary_update( .enumerate() .filter(|(index, _)| *index != value && !groups.contains(index)) .map(|(_, field)| { - SummaryInputExpr::Column(crate::replacement::column_ref(field)) + SummaryInputExpr::Column(crate::realization::column_ref(field)) }) .collect(); if items.is_empty() { @@ -1186,7 +1186,7 @@ fn summary_update( let excluding = groups .iter() .filter_map(|&index| child.fields.get(index)) - .map(crate::replacement::column_ref) + .map(crate::realization::column_ref) .collect(); SummaryInputExpr::EntityIdentity(EntityIdentity::PromqlLabelSet { excluding }) }; diff --git a/crates/asap-aware-mapping/src/realization.rs b/crates/asap-aware-mapping/src/realization.rs new file mode 100644 index 000000000..128625d77 --- /dev/null +++ b/crates/asap-aware-mapping/src/realization.rs @@ -0,0 +1,202 @@ +//! Summary realization helpers shared by Pass 1 (`per_computation_candidates`) +//! and the legacy replacement search: which sketch kinds can serve an intent, +//! the intent's accuracy target and `(ε, δ)` budget, default sketch sizing, +//! and the column/tuple a summary consumes. + +use asap_types::ir::SchemaDerivationError; +use asap_types::post_asap::{AccuracyError, ExactOperationSchemaError, ExecutionDataStateError}; +use asap_types::post_asap::{Field, Schema, SketchAlgorithm, SketchParams, SummaryInputExpr}; +use asap_types::pre_asap::agg_intent::AggIntent; +use asap_types::pre_asap::expr_ir::ColumnRef; +use asap_types::types::AccuracyTarget; +use thiserror::Error; + +/// Confidence δ assumed when the target carries only an ε +/// (`AccuracyTarget::Epsilon`): the (ε, δ)-parameterised sketches (CMS) need +/// one. `ln(1/0.01) → depth 5`, matching the conventional CMS sizing. +pub const DEFAULT_DELTA: f64 = 0.01; + +/// The sketch kinds that can serve an intent, most-preferred first. +/// This is the `AggIntent → SketchAlgorithm` map of issue #98; +/// [`realizations_for_intent`] sizes and ranks every entry via `cost_model`. +/// Listed here so the candidate set has one home. +pub fn summary_candidates(intent: &AggIntent) -> &'static [SketchAlgorithm] { + match intent { + AggIntent::Quantile { .. } => &[SketchAlgorithm::Kll, SketchAlgorithm::DDSketch], + // A distinct-tuple count hashes the whole tuple as one item + // (`SummaryInputExpr::Tuple`), which the distinct-count sketches take + // unchanged. UnivMon is dropped there: it estimates frequency moments + // over a single value stream, and `realize_value_frequency_summary_input` + // would feed it one column of the tuple. + AggIntent::Cardinality { cols, .. } if cols.len() > 1 => &[ + SketchAlgorithm::Hll, + SketchAlgorithm::Theta, + SketchAlgorithm::Kmv, + ], + AggIntent::Cardinality { .. } => &[ + SketchAlgorithm::Hll, + SketchAlgorithm::Theta, + SketchAlgorithm::Kmv, + SketchAlgorithm::UnivMon, + ], + AggIntent::FrequencyL2 { .. } | AggIntent::FrequencyEntropy { .. } => { + &[SketchAlgorithm::UnivMon] + } + // Count-Sketch-with-heap is CMS-with-heap's balanced/zero-mean-error + // alternative for the same heavy-hitter shape. + AggIntent::TopK { .. } => &[ + SketchAlgorithm::CmsWithHeap, + SketchAlgorithm::CountSketchWithHeap, + ], + AggIntent::Count { .. } => &[ + SketchAlgorithm::Cms, + SketchAlgorithm::CountSketch, + SketchAlgorithm::UnivMon, + ], + _ => &[], + } +} + +/// The [`AccuracyTarget`] threaded onto an approximate-capable intent +/// (`Quantile`/`Cardinality`/`Count`/`TopK`), or `None` for every other +/// intent (no sketch candidate applies — [`realizations_for_intent`]'s own +/// match routes those elsewhere). Exposed so callers resolve the exact same +/// accuracy target [`realizations_for_intent`] does, without re-deriving it +/// from scratch. +pub fn accuracy_target(intent: &AggIntent) -> Option<&AccuracyTarget> { + match intent { + AggIntent::Quantile { accuracy, .. } + | AggIntent::Cardinality { accuracy, .. } + | AggIntent::FrequencyL2 { accuracy, .. } + | AggIntent::FrequencyEntropy { accuracy, .. } + | AggIntent::Count { accuracy } + | AggIntent::TopK { accuracy, .. } => Some(accuracy), + _ => None, + } +} + +/// Resolve an [`AccuracyTarget`] into the `(eps, delta)` budget +/// [`CostModel::size_params`] needs. Shared by [`sketch_realizations`] and +/// this crate's own sizing — one place this resolution happens, so nothing +/// can drift apart on it. +/// +/// `Exact` is unreachable via [`realizations_for_intent`] (which routes +/// `Exact` to [`exact_realization`] instead); degrades to the tightest +/// parameters for a caller that resolves it directly anyway. +pub fn accuracy_budget(accuracy: &AccuracyTarget) -> (f64, f64) { + match accuracy { + AccuracyTarget::Exact => (f64::MIN_POSITIVE, DEFAULT_DELTA), + AccuracyTarget::Epsilon(e) => (*e, DEFAULT_DELTA), + AccuracyTarget::EpsilonDelta { epsilon, delta } => (*epsilon, *delta), + } +} + +/// `asap-plan`'s built-in `SketchParams` sizing, keyed off the resolved +/// `(eps, delta)` accuracy budget. [`CostModel::size_params`]'s default +/// body — factored out to a free function so a deployment's own +/// `CostModel` impl can still delegate to it for the candidates it +/// doesn't want to resize itself. +/// +/// Each formula inverts the sketch family's standard error bound to the +/// smallest parameter satisfying the target, clamped to the family's sane +/// range. A non-positive ε saturates to the clamp maximum (tightest +/// allowed). +pub fn default_size_params( + kind: SketchAlgorithm, + intent: &AggIntent, + eps: f64, + delta: f64, +) -> SketchParams { + crate::accuracy::estimators::size_params(kind, intent, eps, delta) +} + +/// The column fed into a *single-column* summary: the intent's leading +/// positional input resolved to a name against the child schema, or the PromQL +/// sample value when it reads none. Callers are responsible for only reaching +/// here with a one-column intent — [`summarised_input`] is the general form. +pub(crate) fn summarised_column(intent: &AggIntent, child_schema: &Schema) -> ColumnRef { + match intent + .input_cols() + .first() + .and_then(|id| child_schema.fields.get(*id)) + { + Some(c) => column_ref(c), + None => ColumnRef::SampleValue, + } +} + +pub(crate) fn column_ref(column: &Field) -> ColumnRef { + match &column.table { + Some(t) => ColumnRef::Qualified { + table: t.clone(), + name: column.name.clone(), + }, + None => ColumnRef::Named(column.name.clone()), + } +} + +/// What the summary consumes per input row. An intent that reads one column (or +/// none) feeds that column; `COUNT(DISTINCT a, b)` feeds the whole tuple as one +/// item, so the distinct-count sketch hashes `(a, b)` rather than `a` — the +/// difference between tuple cardinality and single-column cardinality. +/// +/// A tuple leg outside the child schema is an error rather than +/// [`summarised_column`]'s sample-value fallback: a leg has no sample-value +/// reading, and silently dropping one would under-count. +pub(crate) fn summarised_input( + intent: &AggIntent, + child_schema: &Schema, +) -> Result { + let cols = intent.input_cols(); + if cols.len() < 2 { + return Ok(SummaryInputExpr::Column(summarised_column( + intent, + child_schema, + ))); + } + let legs = cols + .iter() + .map(|id| child_schema.fields.get(*id).map(column_ref)) + .collect::>>() + .ok_or(RealizationError::PhysicalRealization( + "a tuple column is outside the input schema", + ))?; + Ok(SummaryInputExpr::Tuple( + legs.into_iter().map(SummaryInputExpr::Column).collect(), + )) +} + +/// Errors from the pre-ASAP → post-ASAP replacement/construction path +/// (`realize_child` and `retain_exact`). Moved here from the former +/// `bind.rs` (issue #251): this is what a `ReplacementStrategy` +/// implementor's own construction path can realistically fail with — +/// schema derivation over a pre-ASAP [`OperatorNode`] sub-DAG — not +/// something specific to workload-wide orchestration. +#[derive(Debug, Error)] +pub enum RealizationError { + /// Schema derivation failed while lifting an edge to `Schema`. + #[error("schema derivation failed during pre-ASAP → post-ASAP binding: {0}")] + Schema(#[from] SchemaDerivationError), + /// The candidate is accuracy-illegal (issue #172): its composed + /// guarantee has no sound propagation rule, or misses the applicable + /// `AccuracyTarget`. Fail-closed — the candidate is never constructed + /// with the child "treated as exact". [`ASAPStrategies::propose`] + /// records it as a [`RejectedCandidate`] instead of a candidate. + #[error("accuracy-illegal candidate: {0}")] + Accuracy(#[from] AccuracyError), + /// The selected summary family has a physical realization rule for this + /// logical shape, but the rule cannot represent the complete input. The + /// candidate must not fall back to ordinary one-node binding because that + /// would change its semantics. + #[error("unsupported physical summary realization: {0}")] + PhysicalRealization(&'static str), + /// A constructed plan violates the update/evaluation phase contract + /// (issue #171) — e.g. a summary evaluation placed beneath a maintained + /// `SummaryAgg`. Detected at construction, never at runtime. + #[error("execution-data_state violation in post-ASAP plan: {0}")] + ExecutionDataState(#[from] ExecutionDataStateError), + /// An `ExactOperator`'s output schema could not be derived over its + /// child — the child carries summary state the operator can't read. + #[error("exact operator schema derivation failed: {0}")] + ExactOperationSchema(#[from] ExactOperationSchemaError), +} diff --git a/crates/asap-aware-mapping/src/replacement.rs b/crates/asap-aware-mapping/src/replacement.rs index 9c26d8d9e..aa580f39e 100644 --- a/crates/asap-aware-mapping/src/replacement.rs +++ b/crates/asap-aware-mapping/src/replacement.rs @@ -351,18 +351,17 @@ use std::collections::{HashMap, HashSet, VecDeque}; use asap_types::ir::cse::{share_common_sub_dags, structural_hash, HashCache}; use asap_types::ir::operator_properties::{BinaryOpKind, JoinKind, Reduction}; use asap_types::ir::timing::validate_at_ingestion; -use asap_types::ir::SchemaDerivationError; use asap_types::ir::{ ASAPOp, BinaryOperator, NonASAPOp, Operator, OperatorNode, Predicate, ProjectItem, ScalarExpr, SortKey, }; use asap_types::post_asap::{AccuracyError, CompositionOperator, GuaranteeSource, ResultGuarantee}; use asap_types::post_asap::{ - EntityIdentity, ExactKind, ExactOperationSchemaError, ExactParams, ExecutionDataStateError, - ExecutionTiming, Field, FieldDataType, GroupingStrategy, NonNegativeWeightProof, SamplingKind, - SamplingParams, Schema, SketchAlgorithm, SketchKind, SketchParams, - SketchStatistic as PostAsapSketchStatistic, StatModelKind, StatModelParams, SummaryInputExpr, - SummaryUpdate, WaveletKind, WaveletParams, WeightDomain, + EntityIdentity, ExactKind, ExactParams, ExecutionDataStateError, ExecutionTiming, Field, + FieldDataType, GroupingStrategy, NonNegativeWeightProof, SamplingKind, SamplingParams, Schema, + SketchAlgorithm, SketchKind, SketchParams, SketchStatistic as PostAsapSketchStatistic, + StatModelKind, StatModelParams, SummaryInputExpr, SummaryUpdate, WaveletKind, WaveletParams, + WeightDomain, }; use asap_types::pre_asap::agg_intent::{agg_is_mergeable, AggIntent}; use asap_types::pre_asap::expr_ir::{ArithmeticOpKind, ColumnRef}; @@ -370,7 +369,6 @@ use asap_types::pre_asap::schema::ColumnId; use asap_types::types::AccuracyTarget; use asap_types::workload::{DataWorkload, QueryRecurrence, QueryWorkload, RepeatedDemand}; use std::rc::{Rc, Weak}; -use thiserror::Error; use crate::accuracy::reconciliation::AccuracyReconciliationStrategy; use crate::accuracy::{ @@ -383,6 +381,10 @@ use crate::cost_model::{ }; use crate::exact_composition::{ExactComposition, ExactCompositionStrategy, OperationPlacement}; use crate::grouping::HydraGroupingStrategy; +use crate::realization::{ + accuracy_budget, accuracy_target, column_ref, default_size_params, summarised_column, + summarised_input, summary_candidates, DEFAULT_DELTA, +}; use crate::recurrence::CostRate; use crate::recurrence::{ evaluation_rate_of, Horizon, RecurrenceError, RecurrenceProfile, RootRecurrence, UpdateRate, @@ -390,40 +392,7 @@ use crate::recurrence::{ use crate::rollup::RollupStrategy; use crate::topk_reuse::TopKLimitReuseStrategy; -/// Errors from the pre-ASAP → post-ASAP replacement/construction path -/// ([`realize_child`] and [`retain_exact`]). Moved here from the former -/// `bind.rs` (issue #251): this is what a [`ReplacementStrategy`] -/// implementor's own construction path can realistically fail with — -/// schema derivation over a pre-ASAP [`OperatorNode`] sub-DAG — not -/// something specific to workload-wide orchestration. -#[derive(Debug, Error)] -pub enum RealizationError { - /// Schema derivation failed while lifting an edge to `Schema`. - #[error("schema derivation failed during pre-ASAP → post-ASAP binding: {0}")] - Schema(#[from] SchemaDerivationError), - /// The candidate is accuracy-illegal (issue #172): its composed - /// guarantee has no sound propagation rule, or misses the applicable - /// `AccuracyTarget`. Fail-closed — the candidate is never constructed - /// with the child "treated as exact". [`ASAPStrategies::propose`] - /// records it as a [`RejectedCandidate`] instead of a candidate. - #[error("accuracy-illegal candidate: {0}")] - Accuracy(#[from] AccuracyError), - /// The selected summary family has a physical realization rule for this - /// logical shape, but the rule cannot represent the complete input. The - /// candidate must not fall back to ordinary one-node binding because that - /// would change its semantics. - #[error("unsupported physical summary realization: {0}")] - PhysicalRealization(&'static str), - /// A constructed plan violates the update/evaluation phase contract - /// (issue #171) — e.g. a summary evaluation placed beneath a maintained - /// `SummaryAgg`. Detected at construction, never at runtime. - #[error("execution-data_state violation in post-ASAP plan: {0}")] - ExecutionDataState(#[from] ExecutionDataStateError), - /// An `ExactOperator`'s output schema could not be derived over its - /// child — the child carries summary state the operator can't read. - #[error("exact operator schema derivation failed: {0}")] - ExactOperationSchema(#[from] ExactOperationSchemaError), -} +pub use crate::realization::RealizationError; /// A pre-ASAP sub-DAG a [`ReplacementStrategy`] knows how to replace. /// @@ -789,70 +758,6 @@ pub trait Matcher { fn is_satisfied_by(&self, required: &Realization, available: &Realization) -> bool; } -/// Confidence δ assumed when the target carries only an ε -/// (`AccuracyTarget::Epsilon`): the (ε, δ)-parameterised sketches (CMS) need -/// one. `ln(1/0.01) → depth 5`, matching the conventional CMS sizing. -pub const DEFAULT_DELTA: f64 = 0.01; - -/// The sketch kinds that can serve an intent, most-preferred first. -/// This is the `AggIntent → SketchAlgorithm` map of issue #98; -/// [`realizations_for_intent`] sizes and ranks every entry via `cost_model`. -/// Listed here so the candidate set has one home. -pub fn summary_candidates(intent: &AggIntent) -> &'static [SketchAlgorithm] { - match intent { - AggIntent::Quantile { .. } => &[SketchAlgorithm::Kll, SketchAlgorithm::DDSketch], - // A distinct-tuple count hashes the whole tuple as one item - // (`SummaryInputExpr::Tuple`), which the distinct-count sketches take - // unchanged. UnivMon is dropped there: it estimates frequency moments - // over a single value stream, and `realize_value_frequency_summary_input` - // would feed it one column of the tuple. - AggIntent::Cardinality { cols, .. } if cols.len() > 1 => &[ - SketchAlgorithm::Hll, - SketchAlgorithm::Theta, - SketchAlgorithm::Kmv, - ], - AggIntent::Cardinality { .. } => &[ - SketchAlgorithm::Hll, - SketchAlgorithm::Theta, - SketchAlgorithm::Kmv, - SketchAlgorithm::UnivMon, - ], - AggIntent::FrequencyL2 { .. } | AggIntent::FrequencyEntropy { .. } => { - &[SketchAlgorithm::UnivMon] - } - // Count-Sketch-with-heap is CMS-with-heap's balanced/zero-mean-error - // alternative for the same heavy-hitter shape. - AggIntent::TopK { .. } => &[ - SketchAlgorithm::CmsWithHeap, - SketchAlgorithm::CountSketchWithHeap, - ], - AggIntent::Count { .. } => &[ - SketchAlgorithm::Cms, - SketchAlgorithm::CountSketch, - SketchAlgorithm::UnivMon, - ], - _ => &[], - } -} - -/// The [`AccuracyTarget`] threaded onto an approximate-capable intent -/// (`Quantile`/`Cardinality`/`Count`/`TopK`), or `None` for every other -/// intent (no sketch candidate applies — [`realizations_for_intent`]'s own -/// match routes those elsewhere). Exposed so callers resolve the exact same -/// accuracy target [`realizations_for_intent`] does, without re-deriving it -/// from scratch. -pub fn accuracy_target(intent: &AggIntent) -> Option<&AccuracyTarget> { - match intent { - AggIntent::Quantile { accuracy, .. } - | AggIntent::Cardinality { accuracy, .. } - | AggIntent::FrequencyL2 { accuracy, .. } - | AggIntent::FrequencyEntropy { accuracy, .. } - | AggIntent::Count { accuracy } - | AggIntent::TopK { accuracy, .. } => Some(accuracy), - _ => None, - } -} - /// Every valid [`Realization`] for `intent`, exhaustive and ranked /// (most-preferred first via `cost_model`) — the *only* place this crate /// decides what an `AggIntent` may become. Nothing in this crate computes @@ -996,22 +901,6 @@ fn exact_accumulator(intent: &AggIntent, kind: ExactKind, params: ExactParams) - Realization::ExactAggregate { kind, params } } -/// Resolve an [`AccuracyTarget`] into the `(eps, delta)` budget -/// [`CostModel::size_params`] needs. Shared by [`sketch_realizations`] and -/// this crate's own sizing — one place this resolution happens, so nothing -/// can drift apart on it. -/// -/// `Exact` is unreachable via [`realizations_for_intent`] (which routes -/// `Exact` to [`exact_realization`] instead); degrades to the tightest -/// parameters for a caller that resolves it directly anyway. -pub fn accuracy_budget(accuracy: &AccuracyTarget) -> (f64, f64) { - match accuracy { - AccuracyTarget::Exact => (f64::MIN_POSITIVE, DEFAULT_DELTA), - AccuracyTarget::Epsilon(e) => (*e, DEFAULT_DELTA), - AccuracyTarget::EpsilonDelta { epsilon, delta } => (*epsilon, *delta), - } -} - /// Every candidate sketch [`Realization`] for an approximate-capable /// intent, sized to `accuracy` and ranked via `cost_model.rank_candidates` /// (most-preferred first) — [`realizations_for_intent`]'s Sketch branch. @@ -1081,25 +970,6 @@ pub fn sketch_state_bytes(params: &SketchParams) -> Option { .checked_add(u64::from(heap_size).checked_mul(64)?) } -/// `asap-plan`'s built-in `SketchParams` sizing, keyed off the resolved -/// `(eps, delta)` accuracy budget. [`CostModel::size_params`]'s default -/// body — factored out to a free function so a deployment's own -/// `CostModel` impl can still delegate to it for the candidates it -/// doesn't want to resize itself. -/// -/// Each formula inverts the sketch family's standard error bound to the -/// smallest parameter satisfying the target, clamped to the family's sane -/// range. A non-positive ε saturates to the clamp maximum (tightest -/// allowed). -pub fn default_size_params( - kind: SketchAlgorithm, - intent: &AggIntent, - eps: f64, - delta: f64, -) -> SketchParams { - crate::accuracy::estimators::size_params(kind, intent, eps, delta) -} - /// A deployment's explicit bet about how "typical" (non-adversarial) its /// workload's collision pattern is expected to be, consumed only by /// [`posterior_aware_size_params`]. @@ -3823,62 +3693,6 @@ fn summary_col_index(out_schema: &Schema, reduction: &Reduction, measures: usize } } -/// The column fed into a *single-column* summary: the intent's leading -/// positional input resolved to a name against the child schema, or the PromQL -/// sample value when it reads none. Callers are responsible for only reaching -/// here with a one-column intent — [`summarised_input`] is the general form. -fn summarised_column(intent: &AggIntent, child_schema: &Schema) -> ColumnRef { - match intent - .input_cols() - .first() - .and_then(|id| child_schema.fields.get(*id)) - { - Some(c) => column_ref(c), - None => ColumnRef::SampleValue, - } -} - -pub(crate) fn column_ref(column: &Field) -> ColumnRef { - match &column.table { - Some(t) => ColumnRef::Qualified { - table: t.clone(), - name: column.name.clone(), - }, - None => ColumnRef::Named(column.name.clone()), - } -} - -/// What the summary consumes per input row. An intent that reads one column (or -/// none) feeds that column; `COUNT(DISTINCT a, b)` feeds the whole tuple as one -/// item, so the distinct-count sketch hashes `(a, b)` rather than `a` — the -/// difference between tuple cardinality and single-column cardinality. -/// -/// A tuple leg outside the child schema is an error rather than -/// [`summarised_column`]'s sample-value fallback: a leg has no sample-value -/// reading, and silently dropping one would under-count. -pub(crate) fn summarised_input( - intent: &AggIntent, - child_schema: &Schema, -) -> Result { - let cols = intent.input_cols(); - if cols.len() < 2 { - return Ok(SummaryInputExpr::Column(summarised_column( - intent, - child_schema, - ))); - } - let legs = cols - .iter() - .map(|id| child_schema.fields.get(*id).map(column_ref)) - .collect::>>() - .ok_or(RealizationError::PhysicalRealization( - "a tuple column is outside the input schema", - ))?; - Ok(SummaryInputExpr::Tuple( - legs.into_iter().map(SummaryInputExpr::Column).collect(), - )) -} - /// The `SummaryEstimate` evaluation for a summary-bound intent. fn evaluation( intent: &AggIntent, diff --git a/crates/asap-aware-mapping/tests/per_computation_candidates.rs b/crates/asap-aware-mapping/tests/per_computation_candidates.rs index 09ce034df..d63889f3b 100644 --- a/crates/asap-aware-mapping/tests/per_computation_candidates.rs +++ b/crates/asap-aware-mapping/tests/per_computation_candidates.rs @@ -937,7 +937,7 @@ fn summary_agg(node: &OperatorNode) -> &ASAPOp { /// measure's accuracy requirement. #[test] fn composed_summary_is_sized_for_its_measure() { - use asap_aware_mapping::replacement::default_size_params; + use asap_aware_mapping::realization::default_size_params; let intent = AggIntent::Count { accuracy: approximate(), }; diff --git a/crates/devtools/src/bin/show_pre_asap_ir.rs b/crates/devtools/src/bin/show_logical_dag.rs similarity index 91% rename from crates/devtools/src/bin/show_pre_asap_ir.rs rename to crates/devtools/src/bin/show_logical_dag.rs index bde7cfb3c..a37ceb966 100644 --- a/crates/devtools/src/bin/show_pre_asap_ir.rs +++ b/crates/devtools/src/bin/show_logical_dag.rs @@ -1,10 +1,10 @@ -// cargo run -p asap-devtools --bin show_pre_asap_ir -- queries.txt -// (or pipe via stdin: cargo run -p asap-devtools --bin show_pre_asap_ir < queries.txt) +// cargo run -p asap-devtools --bin show_logical_dag -- queries.txt +// (or pipe via stdin: cargo run -p asap-devtools --bin show_logical_dag < queries.txt) // // Lowers a batch of ad-hoc SQL/PromQL queries to **pre-ASAP IR** (the // sketch-agnostic intent algebra: an `OperatorNode` DAG of `NonASAPOp` // operators with `AggIntent` measures) and prints them. -// See `show_post_asap_ir` for the post-ASAP sketch-bound IR one layer +// See `stage_pipeline` for the planner stages one layer // downstream — this tool never picks a sketch, it only shows what a query // means. // @@ -49,7 +49,7 @@ async fn main() { assert_eq!( args.next().as_deref(), Some("--data-ingestion-interval-ms"), - "usage: show_pre_asap_ir --data-ingestion-interval-ms [queries.txt]" + "usage: show_logical_dag --data-ingestion-interval-ms [queries.txt]" ); let interval_ms = args .next() diff --git a/crates/devtools/src/bin/show_post_asap_ir.rs b/crates/devtools/src/bin/show_post_asap_ir.rs deleted file mode 100644 index bc5d5092a..000000000 --- a/crates/devtools/src/bin/show_post_asap_ir.rs +++ /dev/null @@ -1,205 +0,0 @@ -// cargo run -p asap-devtools --bin show_post_asap_ir -- queries.txt -// (or pipe via stdin: cargo run -p asap-devtools --bin show_post_asap_ir < queries.txt) -// -// Lowers a batch of ad-hoc SQL/PromQL queries to pre-ASAP IR, then runs the -// `asap-aware-mapping` pre-ASAP → post-ASAP binding pass and prints the -// resulting **post-ASAP IR** (the sketch-bound IR: an `OperatorNode` DAG in -// which `ASAPOp` operators — the concrete summary family/params committed per -// aggregate — replace the bound aggregates, while whatever the pass left -// untouched stays a plain `NonASAPOp` sub-DAG carrying an exact guarantee). -// See `show_pre_asap_ir` for the sketch-agnostic IR one layer upstream. -// -// File format: one query per line, prefixed with "sql>" or "promql>". -// Blank lines and lines starting with '#' are ignored. -// -// sql> SELECT service, COUNT(*) FROM metrics GROUP BY service -// promql> topk(5, rate(http_requests_total[5m])) -// -// Every query lowers at ACCURACY (ε = 0.01 below) rather than `Exact` — an -// exact target only ever exercises the mergeable-accumulator arm of the -// boundary decision, never a real sketch. SQL queries run against a fixed -// `metrics(ts, service, region, latency, bytes)` catalog — the same table -// used in cross_language.rs and topk_ir.rs. - -use asap_aware_mapping::replacement::retain_exact; -use asap_aware_mapping::{ - ASAPStrategies, Replacement, ReplacementStrategy, ReplacementSubDAG, TargetSubDAG, -}; -use asap_devtools::{lower_promql_with_data_ingestion_interval, lower_sql, SqlCatalog}; -use asap_types::ir::OperatorNode; -use asap_types::pre_asap::schema::{DataType, Field, Schema}; -use asap_types::types::AccuracyTarget; -use std::io::Read; -use std::rc::Rc; - -const ACCURACY: AccuracyTarget = AccuracyTarget::Epsilon(0.01); - -/// `ASAPStrategies::replacements` returns every candidate. This -/// debug tool prints all of them so callers can inspect the planner's choices. -/// If the strategy has none, preserve the single pre-ASAP fallback output. -fn bind_all(root: &Rc) -> Result>, String> { - let target = TargetSubDAG::new(root); - let candidates = ASAPStrategies::default_cost_model() - .replacements(&target) - .into_iter() - .filter_map(|candidate| match candidate { - ReplacementSubDAG { - replacement: Replacement::SubDAG(node), - .. - } => Some(node), - _ => None, - }) - .collect::>(); - - if candidates.is_empty() { - Ok(vec![retain_exact(root).map_err(|e| e.to_string())?]) - } else { - Ok(candidates) - } -} - -fn col(name: &str, dtype: DataType) -> Field { - Field::plain(name, dtype, false) -} - -fn catalog() -> SqlCatalog { - SqlCatalog::new().with_table( - "metrics", - Schema::with_time_index( - vec![ - col("ts", DataType::Timestamp), - col("service", DataType::Utf8), - col("region", DataType::Utf8), - col("latency", DataType::Float64), - col("bytes", DataType::Int64), - ], - 0, - vec![], - ), - ) -} - -#[tokio::main] -async fn main() { - let mut args = std::env::args().skip(1); - assert_eq!( - args.next().as_deref(), - Some("--data-ingestion-interval-ms"), - "usage: show_post_asap_ir --data-ingestion-interval-ms [queries.txt]" - ); - let interval_ms = args - .next() - .expect("missing interval") - .parse() - .expect("interval must be an unsigned integer"); - let input = match args.next() { - Some(path) => { - std::fs::read_to_string(&path).unwrap_or_else(|e| panic!("failed to read {path}: {e}")) - } - None => { - let mut buf = String::new(); - std::io::stdin() - .read_to_string(&mut buf) - .expect("failed to read stdin"); - buf - } - }; - - let catalog = catalog(); - for line in input.lines() { - let line = line.trim(); - if line.is_empty() || line.starts_with('#') { - continue; - } - println!("━━━ {line} ━━━"); - let l3 = if let Some(q) = line.strip_prefix("sql>") { - lower_sql(q.trim(), &catalog, ACCURACY.clone()) - .await - .map_err(|e| e.to_string()) - } else if let Some(q) = line.strip_prefix("promql>") { - lower_promql_with_data_ingestion_interval(q.trim(), ACCURACY.clone(), interval_ms) - .map_err(|e| e.to_string()) - } else { - println!("ERR: line must start with 'sql>' or 'promql>'"); - println!(); - continue; - }; - match l3.and_then(|expr| bind_all(&expr)) { - Ok(candidates) => { - for (index, candidate) in candidates.iter().enumerate() { - println!("--- candidate {} ---", index + 1); - println!("{:#?}", candidate.operator); - } - } - Err(e) => println!("ERR: {e}"), - } - println!(); - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn bind_all_returns_every_sketch_candidate() { - let expr = lower_promql_with_data_ingestion_interval( - "quantile(0.99, rate(http_requests_total[5m]))", - ACCURACY.clone(), - 1_000, - ) - .expect("query lowers to pre-ASAP IR"); - let expected = ASAPStrategies::default_cost_model() - .replacements(&TargetSubDAG::new(&expr)) - .len(); - - assert!(expected > 1, "fixture exposes alternative bindings"); - assert_eq!(bind_all(&expr).expect("binding succeeds").len(), expected); - } - - #[test] - fn bind_all_exposes_an_uncertified_ddsketch_ratio_by_default() { - let expr = lower_promql_with_data_ingestion_interval( - "quantile_over_time(0.9,data[5m])/quantile_over_time(0.5,data[5m])", - ACCURACY.clone(), - 1_000, - ) - .expect("query lowers to pre-ASAP IR"); - - let candidates = bind_all(&expr).expect("binding succeeds"); - assert_eq!(candidates.len(), 1); - assert!(matches!( - candidates[0].non_asap(), - Some(asap_types::ir::NonASAPOp::BinaryOp { .. }) - )); - assert!( - candidates[0].guarantee.is_none(), - "missing evidence must not claim a certified ratio bound" - ); - let timed = asap_types::ir::timing::apply_materialization_timings( - &candidates[0], - &asap_types::ir::timing::MaterializationAssignment::default(), - &mut asap_types::ir::timing::TimingMemo::new(), - ) - .expect("the demo candidate has a legal default timing"); - asap_types::ir::physical_export::compile_physical_asap_dag(&timed) - .expect("the demo candidate remains executable"); - } - - #[test] - fn default_ratio_candidate_does_not_relax_other_approximate_divisions() { - let expr = lower_promql_with_data_ingestion_interval( - "avg_over_time(data[5m])/quantile_over_time(0.5,data[5m])", - ACCURACY.clone(), - 1_000, - ) - .expect("query lowers to pre-ASAP IR"); - - let candidates = bind_all(&expr).expect("binding succeeds"); - assert_eq!(candidates.len(), 1); - assert!( - !candidates[0].contains_asap(), - "the whole query is kept pre-ASAP (no summary bound anywhere)" - ); - } -} diff --git a/crates/frontend-common/src/resolve.rs b/crates/frontend-common/src/resolve.rs index dc204527f..e528aa9c5 100644 --- a/crates/frontend-common/src/resolve.rs +++ b/crates/frontend-common/src/resolve.rs @@ -57,8 +57,8 @@ fn resolve_root_with_inherited( tree: &UnresolvedOp, inherited: &[String], ) -> Result, ResolveDAGError> { - let fallback = SchemaResolver::new().resolve_schema_with_inherited(tree, inherited); - let root = resolve(tree, &fallback)?; + let usage_derived_schema = SchemaResolver::new().resolve_schema_with_inherited(tree, inherited); + let root = resolve(tree, &usage_derived_schema)?; let root = canonicalize(root)?; root.validate_structure()?; Ok(root) @@ -84,11 +84,15 @@ fn node(op: NonASAPOp) -> Result, ResolveDAGError> { )?) } -/// The generic substitution walk. `fallback` is the usage-derived schema a +/// The generic substitution walk. `usage_derived_schema` is the schema a /// schemaless `Scan` in this scope binds to. -fn resolve(tree: &UnresolvedOp, fallback: &Schema) -> Result, ResolveDAGError> { +fn resolve( + tree: &UnresolvedOp, + usage_derived_schema: &Schema, +) -> Result, ResolveDAGError> { use UnresolvedOp as U; - let expr = |e: &UnresolvedScalar, schema: &Schema| resolve_expr_in(e, schema, fallback); + let expr = + |e: &UnresolvedScalar, schema: &Schema| resolve_expr_in(e, schema, usage_derived_schema); let pred = |p: &UnresolvedScalar, schema: &Schema| { Ok::<_, ResolveDAGError>(Predicate(expr(p, schema)?)) }; @@ -109,7 +113,9 @@ fn resolve(tree: &UnresolvedOp, fallback: &Schema) -> Result, R predicates, schema, } => { - let schema = schema.clone().unwrap_or_else(|| fallback.clone()); + let schema = schema + .clone() + .unwrap_or_else(|| usage_derived_schema.clone()); let predicates = predicates .iter() .map(|p| pred(&p.0, &schema)) @@ -139,7 +145,7 @@ fn resolve(tree: &UnresolvedOp, fallback: &Schema) -> Result, R } // A scalar at an operator position has no child scope; in practice a - // literal, so `fallback` is never consulted for a column here. + // literal, so `usage_derived_schema` is never consulted for a column here. U::PromqlScalarOp { child, scalar, @@ -147,7 +153,7 @@ fn resolve(tree: &UnresolvedOp, fallback: &Schema) -> Result, R scalar_left, return_bool, } => { - let child = resolve(child, fallback)?; + let child = resolve(child, usage_derived_schema)?; let child = if child.schema.closed { child } else { @@ -162,7 +168,7 @@ fn resolve(tree: &UnresolvedOp, fallback: &Schema) -> Result, R sample, drop_metric_name, } => { - let child = resolve(child, fallback)?; + let child = resolve(child, usage_derived_schema)?; let child = if child.schema.closed { child } else { @@ -172,12 +178,13 @@ fn resolve(tree: &UnresolvedOp, fallback: &Schema) -> Result, R let sample = resolve_expr(sample, &child.schema)?; project_sample(child, sample, *drop_metric_name) } - U::PromqlVectorFromScalar(inner) => { - node(NonASAPOp::PromqlVectorFromScalar(expr(inner, fallback)?)) - } + U::PromqlVectorFromScalar(inner) => node(NonASAPOp::PromqlVectorFromScalar(expr( + inner, + usage_derived_schema, + )?)), U::PromqlRelabel { dst, value, child } => { - let child = resolve(child, fallback)?; + let child = resolve(child, usage_derived_schema)?; let value = expr(value, &child.schema)?; node(NonASAPOp::PromqlRelabel { dst: dst.clone(), @@ -188,11 +195,11 @@ fn resolve(tree: &UnresolvedOp, fallback: &Schema) -> Result, R U::PromqlInfoEnrich { selector, child } => node(NonASAPOp::PromqlInfoEnrich { selector: selector.clone(), - child: resolve(child, fallback)?, + child: resolve(child, usage_derived_schema)?, }), U::PromqlSeriesSample { by, kind, child } => { - let child = resolve(child, fallback)?; + let child = resolve(child, usage_derived_schema)?; let by = resolve_group_keys(by, &child.schema)?; node(NonASAPOp::PromqlSeriesSample { by, @@ -202,7 +209,7 @@ fn resolve(tree: &UnresolvedOp, fallback: &Schema) -> Result, R } U::Filter { pred: p, child } => { - let child = resolve(child, fallback)?; + let child = resolve(child, usage_derived_schema)?; let pred = pred(&p.0, &child.schema)?; node(NonASAPOp::Filter { pred, child }) } @@ -212,7 +219,7 @@ fn resolve(tree: &UnresolvedOp, fallback: &Schema) -> Result, R qualifier, child, } => { - let child = resolve(child, fallback)?; + let child = resolve(child, usage_derived_schema)?; let cols = cols .iter() .map(|item| { @@ -237,7 +244,7 @@ fn resolve(tree: &UnresolvedOp, fallback: &Schema) -> Result, R having, child, } => { - let child = resolve(child, fallback)?; + let child = resolve(child, usage_derived_schema)?; let reduction = resolve_reduction(reduction, &child.schema)?; let measures = measures .iter() @@ -271,7 +278,7 @@ fn resolve(tree: &UnresolvedOp, fallback: &Schema) -> Result, R } U::Dedup { cols, child } => { - let child = resolve(child, fallback)?; + let child = resolve(child, usage_derived_schema)?; let cols = resolve_column_refs(cols, &child.schema)?; node(NonASAPOp::Dedup { cols, child }) } @@ -282,7 +289,7 @@ fn resolve(tree: &UnresolvedOp, fallback: &Schema) -> Result, R } => { let children = children .iter() - .map(|c| resolve(c, fallback)) + .map(|c| resolve(c, usage_derived_schema)) .collect::, _>>()?; // Resolved against the first branch's own output schema — the one // `output_schema`'s `Concat` arm derives the merged schema from. @@ -343,7 +350,7 @@ fn resolve(tree: &UnresolvedOp, fallback: &Schema) -> Result, R partition_by, child, } => { - let child = resolve(child, fallback)?; + let child = resolve(child, usage_derived_schema)?; let keys = sort_keys(keys, &child.schema)?; let partition_by = resolve_group_keys(partition_by, &child.schema)?; node(NonASAPOp::Sort { @@ -359,7 +366,7 @@ fn resolve(tree: &UnresolvedOp, fallback: &Schema) -> Result, R partition_by, child, } => { - let child = resolve(child, fallback)?; + let child = resolve(child, usage_derived_schema)?; let partition_by = resolve_group_keys(partition_by, &child.schema)?; node(NonASAPOp::Limit { n: *n, @@ -376,18 +383,18 @@ fn resolve(tree: &UnresolvedOp, fallback: &Schema) -> Result, R } => node(NonASAPOp::PromqlSubquery { range: *range, resolution: *resolution, - child: resolve(child, fallback)?, + child: resolve(child, usage_derived_schema)?, }), U::TimeRange { range, kind, child } => node(NonASAPOp::TimeRange { range: *range, kind: *kind, - child: resolve(child, fallback)?, + child: resolve(child, usage_derived_schema)?, }), U::TimeShift { shift, child } => node(NonASAPOp::TimeShift { shift: *shift, - child: resolve(child, fallback)?, + child: resolve(child, usage_derived_schema)?, }), U::SQLWindowFunc { @@ -399,7 +406,7 @@ fn resolve(tree: &UnresolvedOp, fallback: &Schema) -> Result, R output_name, child, } => { - let child = resolve(child, fallback)?; + let child = resolve(child, usage_derived_schema)?; let args = args .iter() .map(|a| expr(a, &child.schema)) @@ -429,7 +436,7 @@ fn resolve(tree: &UnresolvedOp, fallback: &Schema) -> Result, R // The inherited set is computed over the whole `BinaryOp`, so one // side's own labels are not conjured into the other. let own = collect_referenced_columns(tree); - let inherited: Vec = inherited_names(fallback) + let inherited: Vec = inherited_names(usage_derived_schema) .into_iter() .filter(|n| !own.contains(n)) .collect(); @@ -465,7 +472,7 @@ pub fn resolve_expr( } /// [`resolve_expr`] where the operators the expression reads inherit from -/// `enclosing` (the owning root's fallback schema) rather than from `schema`. +/// `enclosing` (the owning root's usage-derived schema) rather than from `schema`. fn resolve_expr_in( expr: &UnresolvedScalar, schema: &Schema, diff --git a/crates/frontend-common/src/schema_resolver.rs b/crates/frontend-common/src/schema_resolver.rs index ad7605fe6..07037766f 100644 --- a/crates/frontend-common/src/schema_resolver.rs +++ b/crates/frontend-common/src/schema_resolver.rs @@ -291,7 +291,7 @@ pub fn collect_referenced_columns(tree: &UnresolvedOp) -> Vec { } => { // An own-field `ColumnRef` must be seeded like `Dedup.cols`, or // a discriminator column referenced nowhere else in the tree is - // absent from the fallback schema and fails `NotFound` later. + // absent from the usage-derived schema and fails `NotFound` later. if let Some(key) = discriminator_unique_key { push_ref_name(key.discriminator(), out); key.inner_key().iter().for_each(|c| push_ref_name(c, out)); diff --git a/crates/integration-tests/tests/sql_to_post_asap.rs b/crates/integration-tests/tests/sql_to_post_asap.rs index 2b0b9e02a..27cda7c4a 100644 --- a/crates/integration-tests/tests/sql_to_post_asap.rs +++ b/crates/integration-tests/tests/sql_to_post_asap.rs @@ -721,7 +721,7 @@ async fn sql_quantile_binds_kll_sketch_over_named_column() { /// `SELECT COUNT(DISTINCT service) FROM metrics` at ε = 0.01 lowers to /// `AggIntent::Cardinality` (`sql_lowering.rs::count_distinct_is_cardinality`) /// — unlike `Quantile`/`Count`/`TopK`, its preferred candidate is HLL, not -/// KLL/CMS (`replacement::summary_candidates`), so this exercises a +/// KLL/CMS (`realization::summary_candidates`), so this exercises a /// distinct branch of the sketch-vs-exact decision than the quantile test /// above. #[tokio::test] diff --git a/crates/planner/tests/summary_sharing.rs b/crates/planner/tests/summary_sharing.rs index 3fb9c19cc..14ea937a0 100644 --- a/crates/planner/tests/summary_sharing.rs +++ b/crates/planner/tests/summary_sharing.rs @@ -10,7 +10,7 @@ use std::rc::Rc; use asap_aware_mapping::pass::{ run_stage1_to_stage3, PlanOutput, PlanningModels, QueryPlan, Stage1ToStage3Run, }; -use asap_aware_mapping::replacement::{default_size_params, DEFAULT_DELTA}; +use asap_aware_mapping::realization::{default_size_params, DEFAULT_DELTA}; use asap_frontend_promql::lower_promql_query_workload; use asap_frontend_sql::SqlCatalog; use asap_planner::{e2e_plan, FrontendInput, UserInput};