diff --git a/crates/asap-aware-mapping/src/analytical_cost.rs b/crates/asap-aware-mapping/src/analytical_cost.rs index 072cf414..63b8edf3 100644 --- a/crates/asap-aware-mapping/src/analytical_cost.rs +++ b/crates/asap-aware-mapping/src/analytical_cost.rs @@ -8,6 +8,7 @@ use std::collections::{HashMap, HashSet}; use asap_types::post_asap::{SketchAlgorithm, SketchParams}; +pub use asap_types::resources::ModeledCpu; use asap_types::workload::DataArrival; use serde::{Deserialize, Serialize}; @@ -54,11 +55,72 @@ impl ResourceCalibration { } } +/// An analytical estimate requires CPU work, peak memory, and scanned bytes. +/// The shared resource container keeps other dimensions explicitly unknown. #[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)] +#[serde(from = "ResourceEstimateWire", into = "ResourceEstimateWire")] pub struct ResourceEstimate { - pub cpu_ops: f64, - pub peak_memory_bytes: u64, - pub scan_bytes: u64, + resources: asap_types::resources::PhysicalResources, +} + +// Keep the established three-field JSON format. Required fields make a missing +// analytical quantity an import error instead of silently supplying zero. +#[derive(Serialize, Deserialize)] +struct ResourceEstimateWire { + cpu_ops: f64, + peak_memory_bytes: u64, + scan_bytes: u64, +} + +impl From for ResourceEstimate { + fn from(wire: ResourceEstimateWire) -> Self { + Self::new(wire.cpu_ops, wire.peak_memory_bytes, wire.scan_bytes) + } +} + +impl From for ResourceEstimateWire { + fn from(estimate: ResourceEstimate) -> Self { + Self { + cpu_ops: estimate.cpu_ops(), + peak_memory_bytes: estimate.peak_memory_bytes(), + scan_bytes: estimate.scan_bytes(), + } + } +} + +impl ResourceEstimate { + pub const fn new(cpu_ops: f64, peak_memory_bytes: u64, scan_bytes: u64) -> Self { + Self { + resources: asap_types::resources::PhysicalResources { + cpu: ModeledCpu { cpu_ops }, + peak_memory_bytes: Some(peak_memory_bytes), + retained_memory_bytes: None, + scan_bytes: Some(scan_bytes), + serialized_bytes: None, + disk_bytes: None, + }, + } + } + + pub fn cpu_ops(&self) -> f64 { + self.resources.cpu.cpu_ops + } + + pub fn peak_memory_bytes(&self) -> u64 { + self.resources + .peak_memory_bytes + .expect("analytical peak memory is required by construction") + } + + pub fn scan_bytes(&self) -> u64 { + self.resources + .scan_bytes + .expect("analytical scan bytes are required by construction") + } + + pub fn resources(&self) -> &asap_types::resources::PhysicalResources { + &self.resources + } } /// Physical operator classes used to expose CPU, memory, and disk formulas @@ -435,18 +497,18 @@ pub fn estimate_physical_dag( ExecutionMultiplicity::Once => 1, ExecutionMultiplicity::PerEvaluation => evaluation_count, }; - cpu_ops += local.cpu_ops * executions as f64; + cpu_ops += local.cpu_ops() * executions as f64; scan_bytes = scan_bytes .checked_add( local - .scan_bytes + .scan_bytes() .checked_mul(executions) .ok_or(AnalyticalCostError::Overflow)?, ) .ok_or(AnalyticalCostError::Overflow)?; peak_memory_bytes = peak_memory_bytes.max( live_bytes - .checked_add(local.peak_memory_bytes) + .checked_add(local.peak_memory_bytes()) .and_then(|bytes| bytes.checked_add(node.output_buffer_bytes)) .ok_or(AnalyticalCostError::Overflow)?, ); @@ -481,11 +543,11 @@ pub fn estimate_physical_dag( if !cpu_ops.is_finite() { return Err(AnalyticalCostError::Overflow); } - Ok(ResourceEstimate { + Ok(ResourceEstimate::new( cpu_ops, peak_memory_bytes, scan_bytes, - }) + )) } fn validate_operator_statistics( @@ -746,11 +808,7 @@ fn partitioned_order_estimate( if !cpu_ops.is_finite() { return Err(AnalyticalCostError::Overflow); } - Ok(ResourceEstimate { - cpu_ops, - peak_memory_bytes, - scan_bytes: 0, - }) + Ok(ResourceEstimate::new(cpu_ops, peak_memory_bytes, 0)) } fn validate_partitioning( @@ -835,11 +893,11 @@ pub fn estimate_operator( "PromQL scalar leaf must emit one scalar row per evaluation step", )); } - return Ok(ResourceEstimate { - cpu_ops: output.rows as f64, - peak_memory_bytes: per_row_width(output.rows, output.bytes)?, - scan_bytes: 0, - }); + return Ok(ResourceEstimate::new( + output.rows as f64, + per_row_width(output.rows, output.bytes)?, + 0, + )); } let left = input(0)?; let estimate = match (operator, &statistics) { @@ -848,36 +906,36 @@ pub fn estimate_operator( OperatorStatistics::Scan { source_read_bytes, .. }, - ) => ResourceEstimate { - cpu_ops: left.rows as f64, - peak_memory_bytes: per_row_width(left.rows, left.bytes)?, - scan_bytes: *source_read_bytes, - }, + ) => ResourceEstimate::new( + left.rows as f64, + per_row_width(left.rows, left.bytes)?, + *source_read_bytes, + ), ( PhysicalOperator::Filter { predicate_operations_per_row, }, _, - ) => ResourceEstimate { - cpu_ops: checked_cpu_product(left.rows, predicate_operations_per_row)?, - peak_memory_bytes: per_row_width(output.rows, output.bytes)?, - scan_bytes: 0, - }, + ) => ResourceEstimate::new( + checked_cpu_product(left.rows, predicate_operations_per_row)?, + per_row_width(output.rows, output.bytes)?, + 0, + ), ( PhysicalOperator::Project { expression_operations_per_row, }, _, - ) => ResourceEstimate { - cpu_ops: checked_cpu_product(left.rows, expression_operations_per_row)?, - peak_memory_bytes: per_row_width(output.rows, output.bytes)?, - scan_bytes: 0, - }, - (PhysicalOperator::PassThrough, _) => ResourceEstimate { - cpu_ops: left.rows as f64, - peak_memory_bytes: per_row_width(output.rows, output.bytes)?, - scan_bytes: 0, - }, + ) => ResourceEstimate::new( + checked_cpu_product(left.rows, expression_operations_per_row)?, + per_row_width(output.rows, output.bytes)?, + 0, + ), + (PhysicalOperator::PassThrough, _) => ResourceEstimate::new( + left.rows as f64, + per_row_width(output.rows, output.bytes)?, + 0, + ), ( PhysicalOperator::HashAggregate { grouping_key_count, @@ -889,22 +947,22 @@ pub fn estimate_operator( accumulator_bytes_per_group, .. }, - ) => ResourceEstimate { - cpu_ops: checked_cpu_product( + ) => ResourceEstimate::new( + checked_cpu_product( left.rows, grouping_key_count .checked_add(accumulator_count) .ok_or(AnalyticalCostError::Overflow)?, )?, - peak_memory_bytes: checked_bytes(&[ + checked_bytes(&[ *group_count, key_bytes .checked_add(*accumulator_bytes_per_group) .and_then(|bytes| bytes.checked_add(16)) .ok_or(AnalyticalCostError::Overflow)?, ])?, - scan_bytes: 0, - }, + 0, + ), ( PhysicalOperator::HashDeduplicate { key_count }, OperatorStatistics::HashDeduplicate { @@ -912,16 +970,16 @@ pub fn estimate_operator( key_bytes, .. }, - ) => ResourceEstimate { - cpu_ops: checked_cpu_product(left.rows, key_count)?, - peak_memory_bytes: checked_bytes(&[ + ) => ResourceEstimate::new( + checked_cpu_product(left.rows, key_count)?, + checked_bytes(&[ *distinct_key_count, key_bytes .checked_add(16) .ok_or(AnalyticalCostError::Overflow)?, ])?, - scan_bytes: 0, - }, + 0, + ), ( PhysicalOperator::InMemoryComparisonSort { ordering_key_count, .. @@ -958,16 +1016,13 @@ pub fn estimate_operator( .checked_add(offset) .ok_or(AnalyticalCostError::Overflow)?; let heap_rows = heap_capacity.min(left.rows); - ResourceEstimate { - cpu_ops: left.rows as f64 + ResourceEstimate::new( + left.rows as f64 * (heap_rows.max(2) as f64).log2().ceil() * ordering_key_count as f64, - peak_memory_bytes: checked_bytes(&[ - heap_rows, - per_row_width(left.rows, left.bytes)?, - ])?, - scan_bytes: 0, - } + checked_bytes(&[heap_rows, per_row_width(left.rows, left.bytes)?])?, + 0, + ) } ( PhysicalOperator::HashJoin { @@ -977,14 +1032,14 @@ pub fn estimate_operator( _, ) => { let right = input(1)?; - ResourceEstimate { - cpu_ops: checked_cpu_product( + ResourceEstimate::new( + checked_cpu_product( left.rows .checked_add(right.rows) .ok_or(AnalyticalCostError::Overflow)?, equality_key_count, )? + output.rows as f64, - peak_memory_bytes: match build_side { + match build_side { HashJoinBuildSide::Left => left .rows .checked_mul(16) @@ -995,14 +1050,14 @@ pub fn estimate_operator( .and_then(|metadata| right.bytes.checked_add(metadata)), } .ok_or(AnalyticalCostError::Overflow)?, - scan_bytes: 0, - } + 0, + ) } - (PhysicalOperator::Concat, _) => ResourceEstimate { - cpu_ops: output.rows as f64, - peak_memory_bytes: per_row_width(output.rows, output.bytes)?, - scan_bytes: 0, - }, + (PhysicalOperator::Concat, _) => ResourceEstimate::new( + output.rows as f64, + per_row_width(output.rows, output.bytes)?, + 0, + ), (PhysicalOperator::Limit { limit, offset }, _) => { let consumed = if limit == 0 { 0 @@ -1013,11 +1068,11 @@ pub fn estimate_operator( .ok_or(AnalyticalCostError::Overflow)?, ) }; - ResourceEstimate { - cpu_ops: consumed as f64, - peak_memory_bytes: per_row_width(output.rows, output.bytes)?, - scan_bytes: 0, - } + ResourceEstimate::new( + consumed as f64, + per_row_width(output.rows, output.bytes)?, + 0, + ) } ( PhysicalOperator::PromqlRange { .. }, @@ -1027,24 +1082,20 @@ pub fn estimate_operator( }, ) => { let promql = require_promql_unary(edges)?; - ResourceEstimate { - cpu_ops: left.rows as f64, - peak_memory_bytes: checked_bytes(&[ + ResourceEstimate::new( + left.rows as f64, + checked_bytes(&[ promql.input.series, *max_window_samples_per_series, per_row_width(left.rows, left.bytes)?, ])?, - scan_bytes: 0, - } + 0, + ) } ( PhysicalOperator::PromqlSubquery { .. }, OperatorStatistics::PromqlSubquery { edges, .. }, - ) => ResourceEstimate { - cpu_ops: left.rows as f64 + output.rows as f64, - peak_memory_bytes: edges.input.bytes, - scan_bytes: 0, - }, + ) => ResourceEstimate::new(left.rows as f64 + output.rows as f64, edges.input.bytes, 0), ( PhysicalOperator::PromqlBinary { operand_mode, @@ -1077,22 +1128,22 @@ pub fn estimate_operator( ])? } }; - ResourceEstimate { - cpu_ops: left.rows as f64 + right.rows as f64 + output.rows as f64, - peak_memory_bytes: matching_bytes, - scan_bytes: 0, - } + ResourceEstimate::new( + left.rows as f64 + right.rows as f64 + output.rows as f64, + matching_bytes, + 0, + ) } ( PhysicalOperator::PromqlRelabel { expression_operations_per_row, }, OperatorStatistics::PromqlRelabel { .. }, - ) => ResourceEstimate { - cpu_ops: checked_cpu_product(left.rows, expression_operations_per_row)?, - peak_memory_bytes: per_row_width(output.rows, output.bytes)?, - scan_bytes: 0, - }, + ) => ResourceEstimate::new( + checked_cpu_product(left.rows, expression_operations_per_row)?, + per_row_width(output.rows, output.bytes)?, + 0, + ), ( PhysicalOperator::PromqlInfoEnrich { matcher_operations_per_info_row, @@ -1104,19 +1155,19 @@ pub fn estimate_operator( ) => { let promql = require_promql_binary(edges)?; let right = input(1)?; - ResourceEstimate { - cpu_ops: left.rows as f64 + ResourceEstimate::new( + left.rows as f64 + right.rows as f64 + output.rows as f64 + checked_cpu_product(right.rows, matcher_operations_per_info_row)?, - peak_memory_bytes: checked_bytes(&[ + checked_bytes(&[ promql.inputs[1].series, matching_key_bytes .checked_add(16) .ok_or(AnalyticalCostError::Overflow)?, ])?, - scan_bytes: 0, - } + 0, + ) } ( PhysicalOperator::PromqlSeriesSample { .. }, @@ -1125,23 +1176,23 @@ pub fn estimate_operator( }, ) => { let promql = require_promql_unary(edges)?; - ResourceEstimate { - cpu_ops: left.rows as f64 + promql.input.series as f64, - peak_memory_bytes: checked_bytes(&[ + ResourceEstimate::new( + left.rows as f64 + promql.input.series as f64, + checked_bytes(&[ promql.output.series, key_bytes .checked_add(16) .ok_or(AnalyticalCostError::Overflow)?, ])?, - scan_bytes: 0, - } + 0, + ) } (PhysicalOperator::PromqlScalarToVector | PhysicalOperator::PromqlVectorToScalar, _) => { - ResourceEstimate { - cpu_ops: left.rows as f64 + output.rows as f64, - peak_memory_bytes: per_row_width(output.rows, output.bytes)?, - scan_bytes: 0, - } + ResourceEstimate::new( + left.rows as f64 + output.rows as f64, + per_row_width(output.rows, output.bytes)?, + 0, + ) } ( PhysicalOperator::PromqlPerSeries { @@ -1153,25 +1204,22 @@ pub fn estimate_operator( }, ) => { let promql = require_promql_unary(edges)?; - ResourceEstimate { - cpu_ops: checked_cpu_product(left.rows, operations_per_row)?, - peak_memory_bytes: checked_bytes(&[ - promql.input.series, - *accumulator_bytes_per_series, - ])?, - scan_bytes: 0, - } + ResourceEstimate::new( + checked_cpu_product(left.rows, operations_per_row)?, + checked_bytes(&[promql.input.series, *accumulator_bytes_per_series])?, + 0, + ) } ( PhysicalOperator::PromqlPresence { operations_per_row, .. }, OperatorStatistics::PromqlPresence { .. }, - ) => ResourceEstimate { - cpu_ops: checked_cpu_product(left.rows, operations_per_row)? + output.rows as f64, - peak_memory_bytes: per_row_width(output.rows, output.bytes)?, - scan_bytes: 0, - }, + ) => ResourceEstimate::new( + checked_cpu_product(left.rows, operations_per_row)? + output.rows as f64, + per_row_width(output.rows, output.bytes)?, + 0, + ), (PhysicalOperator::PromqlScalarLeaf, _) => unreachable!(), _ => { return Err(AnalyticalCostError::InconsistentOperatorStatistics( @@ -1179,7 +1227,7 @@ pub fn estimate_operator( )); } }; - if estimate.cpu_ops.is_finite() { + if estimate.cpu_ops().is_finite() { Ok(estimate) } else { Err(AnalyticalCostError::Overflow) @@ -1927,9 +1975,9 @@ impl ResourceEstimate { calibration: &ResourceCalibration, ) -> Result { calibration.validate()?; - let value = self.cpu_ops * calibration.cost_per_cpu_op - + self.scan_bytes as f64 * calibration.cost_per_scan_byte - + self.peak_memory_bytes as f64 * calibration.cost_per_retained_byte; + let value = self.cpu_ops() * calibration.cost_per_cpu_op + + self.scan_bytes() as f64 * calibration.cost_per_scan_byte + + self.peak_memory_bytes() as f64 * calibration.cost_per_retained_byte; if value.is_finite() { Ok(value) } else { @@ -2016,6 +2064,42 @@ mod tests { PromqlUnaryEdgeStatistics, PromqlValueKind, SourceCoverage, UnaryEdgeStatistics, }; + /// Analytical estimates reuse the shared dimensions while preserving exact + /// integer bytes and keeping unmodeled storage quantities unavailable. + #[test] + fn analytical_estimate_uses_shared_resources() { + let estimate = ResourceEstimate::new(12.5, u64::MAX, 0); + assert_eq!(estimate.cpu_ops(), 12.5); + assert_eq!(estimate.peak_memory_bytes(), u64::MAX); + assert_eq!(estimate.scan_bytes(), 0); + let shared: &asap_types::resources::PhysicalResources = + estimate.resources(); + assert_eq!(shared.cpu.cpu_ops, 12.5); + assert_eq!(shared.peak_memory_bytes, Some(u64::MAX)); + assert_eq!(shared.scan_bytes, Some(0)); + assert_eq!(shared.retained_memory_bytes, None); + assert_eq!(shared.serialized_bytes, None); + assert_eq!(shared.disk_bytes, None); + } + + /// Existing JSON retains its three required fields; omitted or null + /// analytical quantities must not silently become a zero estimate. + #[test] + fn analytical_resource_wire_format_remains_compatible() { + let wire = serde_json::json!({"cpu_ops": 12.5, "peak_memory_bytes": 9007199254740993_u64, "scan_bytes": 0}); + let estimate: ResourceEstimate = serde_json::from_value(wire.clone()).unwrap(); + assert_eq!(estimate.peak_memory_bytes(), 9007199254740993); + assert_eq!(serde_json::to_value(estimate).unwrap(), wire); + for field in ["cpu_ops", "peak_memory_bytes", "scan_bytes"] { + let mut missing = wire.clone(); + missing.as_object_mut().unwrap().remove(field); + assert!(serde_json::from_value::(missing).is_err()); + let mut unknown = wire.clone(); + unknown[field] = serde_json::Value::Null; + assert!(serde_json::from_value::(unknown).is_err()); + } + } + fn filter_operator() -> PhysicalOperator { PhysicalOperator::Filter { predicate_operations_per_row: 1, @@ -2202,8 +2286,8 @@ mod tests { }; let estimate = estimate_operator(PhysicalOperator::PromqlScalarLeaf, statistics).unwrap(); - assert_eq!(estimate.cpu_ops, 10.0); - assert_eq!(estimate.peak_memory_bytes, 8); + assert_eq!(estimate.cpu_ops(), 10.0); + assert_eq!(estimate.peak_memory_bytes(), 8); } #[test] @@ -2410,7 +2494,7 @@ mod tests { }, ) .unwrap(); - assert_eq!(scan.scan_bytes, 64_000); + assert_eq!(scan.scan_bytes(), 64_000); let topk = estimate_operator( PhysicalOperator::TopK { @@ -2433,9 +2517,9 @@ mod tests { }, ) .unwrap(); - assert_eq!(topk.scan_bytes, 0); - assert_eq!(topk.cpu_ops, 4_000.0); - assert_eq!(topk.peak_memory_bytes, 400); + assert_eq!(topk.scan_bytes(), 0); + assert_eq!(topk.cpu_ops(), 4_000.0); + assert_eq!(topk.peak_memory_bytes(), 400); let mismatched_join_statistics = estimate_operator( PhysicalOperator::HashJoin { @@ -2479,7 +2563,7 @@ mod tests { }, ) .unwrap(); - assert_eq!(build_left.peak_memory_bytes, 80_000); + assert_eq!(build_left.peak_memory_bytes(), 80_000); let aggregate = estimate_operator( aggregate_operator(), @@ -2501,7 +2585,7 @@ mod tests { }, ) .unwrap(); - assert_eq!(aggregate.peak_memory_bytes, 5_600); + assert_eq!(aggregate.peak_memory_bytes(), 5_600); let oversized_topk = estimate_operator( PhysicalOperator::TopK { @@ -2524,7 +2608,7 @@ mod tests { }, ) .unwrap(); - assert_eq!(oversized_topk.cpu_ops, 8.0); + assert_eq!(oversized_topk.cpu_ops(), 8.0); let offset_limit = estimate_operator( PhysicalOperator::Limit { @@ -2546,7 +2630,7 @@ mod tests { }, ) .unwrap(); - assert_eq!(offset_limit.cpu_ops, 900_010.0); + assert_eq!(offset_limit.cpu_ops(), 900_010.0); } #[test] @@ -2633,11 +2717,11 @@ mod tests { let mut scope = comparison_scope(); scope.horizon.0 = 20_000; let estimate = estimate_physical_dag(&nodes, "root", &scope, &provided).unwrap(); - assert_eq!(estimate.cpu_ops, 760.0); - assert_eq!(estimate.scan_bytes, 2_000); + assert_eq!(estimate.cpu_ops(), 760.0); + assert_eq!(estimate.scan_bytes(), 2_000); // This is neither the sum of every node's memory nor just the largest // node: it is the maximum state simultaneously live at the fan-out. - assert_eq!(estimate.peak_memory_bytes, 28); + assert_eq!(estimate.peak_memory_bytes(), 28); } #[test] @@ -2707,8 +2791,8 @@ mod tests { let mut scope = comparison_scope(); scope.horizon.0 = 100_000; let estimate = estimate_physical_dag(&nodes, "read", &scope, &provided).unwrap(); - assert_eq!(estimate.cpu_ops, 310.0); - assert_eq!(estimate.scan_bytes, 1_000); + assert_eq!(estimate.cpu_ops(), 310.0); + assert_eq!(estimate.scan_bytes(), 1_000); } fn comparison_scope() -> ComparisonScope { @@ -2928,8 +3012,8 @@ mod tests { let estimate = estimate_physical_dag(&nodes, "filter", &comparison_scope(), &provided).unwrap(); - assert_eq!(estimate.cpu_ops, 1_200.0); - assert_eq!(estimate.scan_bytes, 6_000); + assert_eq!(estimate.cpu_ops(), 1_200.0); + assert_eq!(estimate.scan_bytes(), 6_000); } #[test] @@ -2977,9 +3061,9 @@ mod tests { let estimate = estimate_physical_dag(&nodes, "filter", &comparison_scope(), &provided).unwrap(); - assert_eq!(estimate.cpu_ops, 1_200.0); - assert_eq!(estimate.peak_memory_bytes, 20); - assert_eq!(estimate.scan_bytes, 6_000); + assert_eq!(estimate.cpu_ops(), 1_200.0); + assert_eq!(estimate.peak_memory_bytes(), 20); + assert_eq!(estimate.scan_bytes(), 6_000); } #[test] @@ -3208,8 +3292,8 @@ mod tests { ) .unwrap(); - assert_eq!(estimate.scan_bytes, 2_500); - assert_eq!(estimate.peak_memory_bytes, 100); + assert_eq!(estimate.scan_bytes(), 2_500); + assert_eq!(estimate.peak_memory_bytes(), 100); } #[test] @@ -3225,7 +3309,7 @@ mod tests { assert_eq!( estimate_operator(filter_operator(), filter) .unwrap() - .scan_bytes, + .scan_bytes(), 0 ); } @@ -3266,8 +3350,8 @@ mod tests { }, ) .unwrap(); - assert_eq!(ungrouped.cpu_ops, 0.0); - assert_eq!(ungrouped.peak_memory_bytes, 24); + assert_eq!(ungrouped.cpu_ops(), 0.0); + assert_eq!(ungrouped.peak_memory_bytes(), 24); let grouped = estimate_operator( PhysicalOperator::HashAggregate { @@ -3282,7 +3366,7 @@ mod tests { }, ) .unwrap(); - assert_eq!(grouped.peak_memory_bytes, 0); + assert_eq!(grouped.peak_memory_bytes(), 0); } #[test] @@ -3302,7 +3386,7 @@ mod tests { filter, ) .unwrap() - .cpu_ops, + .cpu_ops(), 300.0 ); @@ -3328,8 +3412,8 @@ mod tests { }, ) .unwrap(); - assert_eq!(sort.cpu_ops, 600.0); - assert_eq!(sort.peak_memory_bytes, 400); + assert_eq!(sort.cpu_ops(), 600.0); + assert_eq!(sort.peak_memory_bytes(), 400); } #[test] diff --git a/crates/asap-aware-mapping/src/empirical_comparison.rs b/crates/asap-aware-mapping/src/empirical_comparison.rs new file mode 100644 index 00000000..ecd2a6cd --- /dev/null +++ b/crates/asap-aware-mapping/src/empirical_comparison.rs @@ -0,0 +1,913 @@ +//! 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 readout 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 readout 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", "readout"] { + 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(), + "readout" => { + 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 new file mode 100644 index 00000000..2e195bd3 --- /dev/null +++ b/crates/asap-aware-mapping/src/empirical_cost.rs @@ -0,0 +1,774 @@ +//! Offline sketch-bench evidence. Measurements describe a particular dataset, +//! configuration and environment; they are neither runtime feedback nor proofs +//! of an accuracy guarantee. CPU quantities are nanoseconds, never CPU operations. + +use asap_types::post_asap::{ + GroupingStrategy, SketchAlgorithm, SketchParams, SummaryExpr, SummaryFamilyType, SummaryNode, +}; +use asap_types::pre_asap::AggIntent; +use serde::{Deserialize, Serialize}; + +use crate::cost_model::{Cost, CostModel, DefaultCostModel}; +use crate::replacement::{ + accuracy_budget, accuracy_target, default_size_params, ReplacementSubDAG, TargetSubDAG, +}; +use crate::summary_maintenance_lifecycle::SummaryMaintenanceLifecycleCostInputs; + +pub const EVIDENCE_SCHEMA_VERSION: u32 = 1; +pub const EVIDENCE_MODEL_VERSION: &str = "empirical-update-cpu-v1"; + +pub use crate::empirical_resources::ResourceMeasurements; +pub use asap_types::resources::Measurement; + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct DistributionDescriptor { + pub id: String, + pub family: String, + pub sample_count: u64, + pub distinct_count: Option, + /// Generator parameters or trace identity/checksum, including sampling rules. + pub parameters: serde_json::Value, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct EnvironmentDescriptor { + pub id: String, + pub cpu: String, + pub os: String, + pub runtime: String, + pub implementation: String, + pub implementation_version: String, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct MeasurementProvenance { + pub command: String, + pub dataset: String, + pub source_revision: String, + pub repetitions: u32, +} + +/// Observed error on offline ground truth. No confidence or formal guarantee is +/// inferred from these statistics; `metric` defines the meaning of mean/max. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct OfflineError { + pub metric: String, + pub mean: Option, + pub max: Option, + pub trials: u32, + pub ground_truth_method: String, + pub query: serde_json::Value, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct OfflineMeasurement { + pub id: String, + pub algorithm: SketchAlgorithm, + pub params: SketchParams, + pub distribution: DistributionDescriptor, + pub environment: EnvironmentDescriptor, + pub measured_at_unix_seconds: u64, + pub valid_until_unix_seconds: u64, + pub provenance: MeasurementProvenance, + pub metrics: ResourceMeasurements, + pub error: Option, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct EvidenceArtifact { + pub schema_version: u32, + pub benchmark_version: String, + /// Identity of the normalization/cost interpretation, independent of JSON + /// layout and of the sketch implementation's source revision. + pub model_version: String, + pub records: Vec, +} + +/// The caller explicitly chooses the offline applicability context. Matching +/// all descriptors prevents reusing costs solely because a distribution name +/// or hardware label happens to agree. Time is injected for reproducibility. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct EvidenceContext { + pub distribution: DistributionDescriptor, + pub environment: EnvironmentDescriptor, + pub now_unix_seconds: u64, +} + +#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] +pub enum EvidenceError { + #[error("unsupported offline evidence schema version {0}")] + UnsupportedVersion(u32), + #[error("invalid offline evidence: {0}")] + Invalid(String), + #[error("no measurement for algorithm and exact configuration")] + MissingConfiguration, + #[error("offline distribution or environment is incompatible")] + IncompatibleContext, + #[error("offline measurement is expired or dated in the future")] + Stale, + #[error("multiple applicable measurements; select an unambiguous artifact")] + Ambiguous, +} + +pub struct EmpiricalEvidenceProvider { + artifact: EvidenceArtifact, + context: EvidenceContext, +} + +impl EmpiricalEvidenceProvider { + pub fn new( + artifact: EvidenceArtifact, + context: EvidenceContext, + ) -> Result { + artifact.validate()?; + if artifact.model_version != EVIDENCE_MODEL_VERSION { + return invalid("unsupported offline cost model version"); + } + validate_context(&context.distribution, &context.environment)?; + Ok(Self { artifact, context }) + } + + pub fn artifact(&self) -> &EvidenceArtifact { + &self.artifact + } + pub fn context(&self) -> &EvidenceContext { + &self.context + } + + /// Never interpolates between configurations, distributions, or machines. + /// The returned row includes complete provenance for user explanations. + pub fn lookup( + &self, + algorithm: &SketchAlgorithm, + params: &SketchParams, + ) -> Result<&OfflineMeasurement, EvidenceError> { + let configurations: Vec<_> = self + .artifact + .records + .iter() + .filter(|r| &r.algorithm == algorithm && &r.params == params) + .collect(); + if configurations.is_empty() { + return Err(EvidenceError::MissingConfiguration); + } + let compatible: Vec<_> = configurations + .into_iter() + .filter(|r| { + r.distribution == self.context.distribution + && r.environment == self.context.environment + }) + .collect(); + if compatible.is_empty() { + return Err(EvidenceError::IncompatibleContext); + } + let mut valid = compatible.into_iter().filter(|r| { + r.measured_at_unix_seconds <= self.context.now_unix_seconds + && self.context.now_unix_seconds <= r.valid_until_unix_seconds + }); + let first = valid.next().ok_or(EvidenceError::Stale)?; + if valid.next().is_some() { + return Err(EvidenceError::Ambiguous); + } + Ok(first) + } + + /// Reorders only a fully measured comparison, preserving all candidates. + /// For deployment-specific sizing call `lookup` with those exact params. + pub fn rank_candidates( + &self, + intent: &AggIntent, + candidates: &[SketchAlgorithm], + eps: f64, + delta: f64, + ) -> Vec { + let costs: Option> = candidates + .iter() + .map(|algorithm| { + let params = default_size_params(algorithm.clone(), intent, eps, delta); + self.lookup(algorithm, ¶ms) + .ok()? + .metrics + .resources + .cpu + .update_cpu_ns + .as_ref() + .map(|m| (algorithm.clone(), m.value)) + }) + .collect(); + let Some(mut costs) = costs else { + return candidates.to_vec(); + }; + costs.sort_by(|a, b| a.1.total_cmp(&b.1)); + costs.into_iter().map(|(algorithm, _)| algorithm).collect() + } + + /// Costs for one independently instantiated sketch state, in CPU ns. + /// Unknown retention/retirement remain unavailable; CPU time must not be + /// mixed with an existing deployment's unitless or CPU-operation costs. + pub fn lifecycle_cost_inputs( + &self, + summary: &SummaryNode, + ) -> SummaryMaintenanceLifecycleCostInputs { + let SummaryExpr::SummaryAgg { + family: SummaryFamilyType::Sketch(kind, GroupingStrategy::PerSubpopulationInstance), + grouping: GroupingStrategy::PerSubpopulationInstance, + .. + } = &summary.expr + else { + return SummaryMaintenanceLifecycleCostInputs::default(); + }; + let Ok(row) = self.lookup(kind.algorithm(), kind.params()) else { + return SummaryMaintenanceLifecycleCostInputs::default(); + }; + SummaryMaintenanceLifecycleCostInputs { + build_cost: snapshot_build_cpu(row).map(Cost), + maintenance_cost_per_update: row + .metrics + .resources + .cpu + .update_cpu_ns + .as_ref() + .map(|m| Cost(m.value)), + // A point-frequency benchmark read does not price a total-count + // or quantile read. There is no query request in this hook. + summary_read_cost: None, + ..Default::default() + } + } +} + +/// Standalone adapter for the existing planner boundary. Empirical data changes +/// candidate discovery order; final structural scores retain their documented +/// default meaning. Complete plan benefit estimates require downstream raw and +/// summary physical evidence and are deliberately not invented here. +pub struct EmpiricalCostModel { + pub provider: EmpiricalEvidenceProvider, +} + +impl EmpiricalCostModel { + pub fn new(provider: EmpiricalEvidenceProvider) -> Self { + Self { provider } + } +} + +impl CostModel for EmpiricalCostModel { + fn rank_candidates( + &self, + intent: &AggIntent, + candidates: &[SketchAlgorithm], + ) -> Vec { + let Some(accuracy) = accuracy_target(intent) else { + return candidates.to_vec(); + }; + let (eps, delta) = accuracy_budget(accuracy); + self.provider + .rank_candidates(intent, candidates, eps, delta) + } + + // The default size_params formula retains its formal guarantee. Observed + // offline error is insufficient evidence to shrink a sketch safely. + fn estimate_cost(&self, candidate: &ReplacementSubDAG, target: &TargetSubDAG<'_>) -> f64 { + DefaultCostModel.estimate_cost(candidate, target) + } + + fn summary_maintenance_lifecycle_cost_inputs( + &self, + summary: &SummaryNode, + ) -> SummaryMaintenanceLifecycleCostInputs { + self.provider.lifecycle_cost_inputs(summary) + } +} + +impl EvidenceArtifact { + pub fn validate(&self) -> Result<(), EvidenceError> { + if self.schema_version != EVIDENCE_SCHEMA_VERSION { + return Err(EvidenceError::UnsupportedVersion(self.schema_version)); + } + if self.benchmark_version.trim().is_empty() || self.model_version.trim().is_empty() { + return invalid("missing benchmark or model version"); + } + let mut ids = std::collections::HashSet::new(); + for row in &self.records { + if row.id.trim().is_empty() || !ids.insert(&row.id) { + return invalid("empty or duplicate record id"); + } + validate_context(&row.distribution, &row.environment)?; + let p = &row.provenance; + if [&p.command, &p.dataset, &p.source_revision] + .iter() + .any(|s| s.trim().is_empty()) + || p.repetitions == 0 + { + return invalid("missing measurement provenance"); + } + if row.measured_at_unix_seconds > row.valid_until_unix_seconds { + return invalid("reversed validity interval"); + } + if !valid_params(&row.algorithm, &row.params) { + return invalid("invalid or mismatched sketch parameters"); + } + 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(|v| !nonnegative(v)) + { + return invalid("invalid measurement or uncertainty"); + } + } + if let Some(error) = &row.error { + if error.metric.trim().is_empty() + || error.ground_truth_method.trim().is_empty() + || error.trials == 0 + || [error.mean, error.max] + .into_iter() + .flatten() + .any(|v| !nonnegative(v)) + { + return invalid("invalid offline error evidence"); + } + } + } + Ok(()) + } +} + +fn invalid(message: &str) -> Result { + Err(EvidenceError::Invalid(message.into())) +} +fn nonnegative(value: f64) -> bool { + value.is_finite() && value >= 0.0 +} + +fn snapshot_build_cpu(row: &OfflineMeasurement) -> Option { + let cpu = row.metrics.resources.cpu.build_cpu_ns.as_ref()?.value + + row.metrics.resources.cpu.update_cpu_ns.as_ref()?.value + * row.distribution.sample_count as f64 + + snapshot_prepare_cpu(row)?; + nonnegative(cpu).then_some(cpu) +} + +/// 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, +) -> Result<(), EvidenceError> { + if [ + &distribution.id, + &distribution.family, + &environment.id, + &environment.cpu, + &environment.os, + &environment.runtime, + &environment.implementation, + &environment.implementation_version, + ] + .iter() + .any(|s| s.trim().is_empty()) + { + return invalid("missing distribution or environment identity"); + } + if distribution.sample_count == 0 + || distribution + .distinct_count + .is_some_and(|n| n > distribution.sample_count) + || !distribution.parameters.is_object() + { + return invalid("invalid distribution descriptors"); + } + Ok(()) +} + +fn valid_params(algorithm: &SketchAlgorithm, params: &SketchParams) -> bool { + match (algorithm, params) { + (SketchAlgorithm::Kll, SketchParams::Kll { k }) + | (SketchAlgorithm::Kmv, SketchParams::Kmv { k }) + | (SketchAlgorithm::Theta, SketchParams::Theta { k }) => *k > 0, + (SketchAlgorithm::Hll, SketchParams::Hll { precision }) => (4..=18).contains(precision), + (SketchAlgorithm::DDSketch, SketchParams::DDSketch { alpha }) => { + alpha.is_finite() && *alpha > 0.0 && *alpha < 1.0 + } + (SketchAlgorithm::Cms, SketchParams::Cms { width, depth }) + | (SketchAlgorithm::CountSketch, SketchParams::CountSketch { width, depth }) => { + *width > 0 && *depth > 0 + } + ( + SketchAlgorithm::CmsWithHeap, + SketchParams::CmsWithHeap { + width, + depth, + heap_size, + }, + ) + | ( + SketchAlgorithm::CountSketchWithHeap, + SketchParams::CountSketchWithHeap { + width, + depth, + heap_size, + }, + ) => *width > 0 && *depth > 0 && *heap_size > 0, + _ => false, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::replacement::{implementations_for_with, Implementation}; + use asap_types::types::AccuracyTarget; + + /// The documented synthetic wire-format example remains importable and + /// explicitly identifiable as a test fixture. + #[test] + fn checked_in_synthetic_example_is_valid() { + let artifact: EvidenceArtifact = serde_json::from_str(include_str!( + "../tests/data/offline-evidence-synthetic.json" + )) + .unwrap(); + artifact.validate().unwrap(); + assert!(artifact.records[0] + .environment + .implementation + .contains("SYNTHETIC")); + let schema: serde_json::Value = serde_json::from_str(include_str!( + "../../../docs/developer_docs/offline-sketch-evidence.schema.json" + )) + .unwrap(); + assert_eq!( + schema["properties"]["schema_version"]["const"], + artifact.schema_version + ); + assert_eq!( + schema["properties"]["model_version"]["const"], + EVIDENCE_MODEL_VERSION + ); + } + + /// Lifecycle build includes all measured snapshot updates, not just an empty + /// allocation. A missing update measurement cannot become free ingestion. + #[test] + fn lifecycle_build_requires_complete_snapshot_ingestion() { + let (mut artifact, _, _) = fixture(); + let row = &mut artifact.records[0]; + row.metrics.resources.cpu.build_cpu_ns = Some(Measurement { + value: 10.0, + stddev: None, + samples: 1, + method: None, + }); + assert_eq!(snapshot_build_cpu(row), Some(20010.0)); + row.metrics.resources.cpu.prepare_cpu_ns = Some(Measurement { + value: 17.0, + stddev: None, + samples: 1, + method: None, + }); + assert_eq!(snapshot_build_cpu(row), Some(20027.0)); + row.metrics.resources.cpu.update_cpu_ns = None; + assert_eq!(snapshot_build_cpu(row), None); + } + + /// Newly shared optional dimensions receive the same numeric validation. + #[test] + fn optional_prepare_and_scan_measurements_are_validated() { + for prepare in [true, false] { + let (mut artifact, _, _) = fixture(); + let resources = &mut artifact.records[0].metrics.resources; + let field = if prepare { + &mut resources.cpu.prepare_cpu_ns + } else { + &mut resources.scan_bytes + }; + *field = Some(Measurement { + value: -1.0, + stddev: None, + samples: 1, + method: None, + }); + assert!(artifact.validate().is_err()); + } + } + + /// Only the established frequency-sketch contract can omit preparation. + #[test] + fn unmeasured_preparation_for_other_families_keeps_build_unknown() { + let (mut artifact, _, _) = fixture(); + let row = &mut artifact.records[0]; + row.metrics.resources.cpu.build_cpu_ns = Some(Measurement { + value: 10.0, + stddev: None, + samples: 1, + method: None, + }); + assert_eq!(snapshot_build_cpu(row), Some(20010.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_build_cpu(row), None); + row.metrics.resources.cpu.prepare_cpu_ns = Some(Measurement { + value: 17.0, + stddev: None, + samples: 1, + method: None, + }); + assert_eq!(snapshot_build_cpu(row), Some(20027.0)); + } + + fn fixture() -> (EvidenceArtifact, EvidenceContext, AggIntent) { + let distribution = DistributionDescriptor { + id: "unit-test-uniform".into(), + family: "uniform".into(), + sample_count: 1000, + distinct_count: Some(100), + parameters: serde_json::json!({"seed": 7}), + }; + let environment = EnvironmentDescriptor { + id: "unit-test-machine".into(), + cpu: "test CPU".into(), + os: "test OS".into(), + runtime: "test runtime".into(), + implementation: "synthetic test fixture".into(), + implementation_version: "test-v1".into(), + }; + let intent = AggIntent::Count { + accuracy: AccuracyTarget::EpsilonDelta { + epsilon: 0.01, + delta: 0.01, + }, + }; + let records = [ + (SketchAlgorithm::Cms, 20.0), + (SketchAlgorithm::CountSketch, 10.0), + ] + .into_iter() + .map(|(algorithm, cost)| OfflineMeasurement { + id: format!("test-{algorithm:?}"), + params: default_size_params(algorithm.clone(), &intent, 0.01, 0.01), + algorithm, + distribution: distribution.clone(), + environment: environment.clone(), + measured_at_unix_seconds: 100, + valid_until_unix_seconds: 200, + provenance: MeasurementProvenance { + command: "unit test fixture; not a measured benchmark".into(), + dataset: "synthetic fixture".into(), + source_revision: "test".into(), + repetitions: 3, + }, + metrics: ResourceMeasurements { + resources: asap_types::resources::MeasuredResources { + cpu: asap_types::resources::MeasuredCpu { + update_cpu_ns: Some(Measurement { + value: cost, + stddev: Some(1.0), + samples: 3, + method: None, + }), + ..Default::default() + }, + ..Default::default() + }, + }, + error: Some(OfflineError { + metric: "mean absolute relative frequency error".into(), + mean: Some(0.001), + max: None, + trials: 3, + ground_truth_method: "test exact counter fixture".into(), + query: serde_json::json!({"kind":"point_frequency"}), + }), + }) + .collect(); + ( + EvidenceArtifact { + schema_version: EVIDENCE_SCHEMA_VERSION, + benchmark_version: "test-fixture-v1".into(), + model_version: "empirical-update-cpu-v1".into(), + records, + }, + EvidenceContext { + distribution, + environment, + now_unix_seconds: 150, + }, + intent, + ) + } + + /// Real replacement generation follows measured update ranking while keeping + /// every candidate and the same formally sized parameter configurations. + #[test] + fn public_cost_model_changes_replacement_order_without_changing_guarantees() { + let (artifact, context, intent) = fixture(); + let model = + EmpiricalCostModel::new(EmpiricalEvidenceProvider::new(artifact, context).unwrap()); + let default = implementations_for_with(&intent, &DefaultCostModel); + let measured = implementations_for_with(&intent, &model); + assert_eq!(default.len(), measured.len()); + let Implementation::Sketch(first_default) = &default[0] else { + panic!("expected sketch") + }; + let Implementation::Sketch(first_measured) = &measured[0] else { + panic!("expected sketch") + }; + assert_eq!(first_default.algorithm(), &SketchAlgorithm::Cms); + assert_eq!(first_measured.algorithm(), &SketchAlgorithm::CountSketch); + for candidate in &measured { + assert!(default.contains(candidate)); + } + assert_eq!( + model.size_params(SketchAlgorithm::Cms, &intent, 0.001, 0.001), + DefaultCostModel.size_params(SketchAlgorithm::Cms, &intent, 0.001, 0.001) + ); + } + + /// Missing, mismatched and expired evidence preserve the original ranking; + /// a measurement from another configuration is never extrapolated. + #[test] + fn unavailable_evidence_falls_back_with_specific_reasons() { + let (artifact, context, intent) = fixture(); + let candidates = vec![SketchAlgorithm::Cms, SketchAlgorithm::CountSketch]; + for (mut evidence, mut request, expected) in [ + ( + artifact.clone(), + context.clone(), + EvidenceError::MissingConfiguration, + ), + ( + artifact.clone(), + context.clone(), + EvidenceError::IncompatibleContext, + ), + (artifact.clone(), context.clone(), EvidenceError::Stale), + ] { + match expected { + EvidenceError::MissingConfiguration => { + evidence.records.remove(1); + } + EvidenceError::IncompatibleContext => { + request.distribution.parameters = serde_json::json!({"seed":8}); + } + EvidenceError::Stale => { + request.now_unix_seconds = 201; + } + _ => unreachable!(), + } + let provider = EmpiricalEvidenceProvider::new(evidence, request).unwrap(); + assert_eq!( + provider + .lookup(&artifact.records[1].algorithm, &artifact.records[1].params) + .unwrap_err(), + expected + ); + assert_eq!( + provider.rank_candidates(&intent, &candidates, 0.01, 0.01), + candidates + ); + } + let provider = EmpiricalEvidenceProvider::new(artifact, context).unwrap(); + assert_eq!( + provider.rank_candidates(&intent, &candidates, 0.001, 0.01), + candidates + ); + } + + /// Null remains unknown across serialization; zero is accepted only as an + /// explicit valid measurement, and no point-frequency error becomes a bound. + #[test] + fn serialization_preserves_unknown_zero_and_provenance() { + let (mut artifact, context, _) = fixture(); + artifact.records[0].metrics.resources.disk_bytes = Some(Measurement { + value: 0.0, + stddev: None, + samples: 1, + method: None, + }); + let decoded: EvidenceArtifact = + serde_json::from_str(&serde_json::to_string(&artifact).unwrap()).unwrap(); + let provider = EmpiricalEvidenceProvider::new(decoded, context).unwrap(); + let row = provider + .lookup(&artifact.records[0].algorithm, &artifact.records[0].params) + .unwrap(); + assert!(row.metrics.resources.peak_memory_bytes.is_none()); + assert_eq!( + row.metrics.resources.disk_bytes.as_ref().unwrap().value, + 0.0 + ); + assert_eq!(row.provenance, artifact.records[0].provenance); + assert_eq!(row.error.as_ref().unwrap().query["kind"], "point_frequency"); + } + + /// Malformed values, schema versions and ambiguous live records cannot + /// silently become plausible costs. + #[test] + fn invalid_and_ambiguous_artifacts_are_rejected() { + let (artifact, context, _) = fixture(); + let mut bad = artifact.clone(); + bad.schema_version = 2; + assert_eq!(bad.validate(), Err(EvidenceError::UnsupportedVersion(2))); + let mut bad = artifact.clone(); + bad.records[0] + .metrics + .resources + .cpu + .update_cpu_ns + .as_mut() + .unwrap() + .value = f64::NAN; + assert!(bad.validate().is_err()); + let mut bad = artifact.clone(); + bad.records[0].params = SketchParams::Hll { precision: 14 }; + assert!(bad.validate().is_err()); + let mut bad = artifact.clone(); + bad.records[0].provenance.repetitions = 0; + assert!(bad.validate().is_err()); + let mut duplicate = artifact.records[0].clone(); + duplicate.id = "another-live-measurement".into(); + let mut ambiguous = artifact.clone(); + ambiguous.records.push(duplicate); + let provider = EmpiricalEvidenceProvider::new(ambiguous, context).unwrap(); + assert_eq!( + provider.lookup(&artifact.records[0].algorithm, &artifact.records[0].params), + Err(EvidenceError::Ambiguous) + ); + } +} diff --git a/crates/asap-aware-mapping/src/empirical_resources.rs b/crates/asap-aware-mapping/src/empirical_resources.rs new file mode 100644 index 00000000..716392f3 --- /dev/null +++ b/crates/asap-aware-mapping/src/empirical_resources.rs @@ -0,0 +1,202 @@ +//! Measured resource payloads and compatibility with the v1 benchmark wire format. +//! +//! Physical dimensions live in `asap_types::resources`; flat wire structs below +//! exist only to keep archived artifacts readable and preserve their field names. + +use asap_types::resources::PhysicalResources; +use serde::{Deserialize, Serialize}; + +pub use asap_types::resources::{MeasuredCpu, MeasuredResources, Measurement}; + +#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)] +#[serde(from = "SketchWire", into = "SketchWire")] +pub struct ResourceMeasurements { + pub resources: MeasuredResources, +} + +#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)] +#[serde(from = "ExactWire", into = "ExactWire")] +pub struct ExactResourceMeasurements { + pub resources: MeasuredResources, +} + +// Keep the archived flat v1 schema at the serialization boundary only. New +// optional dimensions are omitted when absent, so legacy snapshots round-trip. +#[derive(Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct SketchWire { + build_cpu_ns: Option, + update_cpu_ns: Option, + merge_cpu_ns: Option, + read_cpu_ns: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + prepare_cpu_ns: Option, + retained_bytes: Option, + peak_bytes: Option, + serialized_bytes: Option, + disk_bytes: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + scan_bytes: Option, +} + +#[derive(Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct ExactWire { + empty_build_cpu_ns: Option, + update_cpu_ns: Option, + prepare_cpu_ns: Option, + read_cpu_ns: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + merge_cpu_ns: Option, + retained_bytes: Option, + peak_bytes: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + serialized_bytes: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + disk_bytes: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + scan_bytes: Option, +} + +impl From for ResourceMeasurements { + fn from(w: SketchWire) -> Self { + Self { + resources: PhysicalResources { + cpu: MeasuredCpu { + build_cpu_ns: w.build_cpu_ns, + update_cpu_ns: w.update_cpu_ns, + merge_cpu_ns: w.merge_cpu_ns, + prepare_cpu_ns: w.prepare_cpu_ns, + read_cpu_ns: w.read_cpu_ns, + }, + retained_memory_bytes: w.retained_bytes, + peak_memory_bytes: w.peak_bytes, + serialized_bytes: w.serialized_bytes, + disk_bytes: w.disk_bytes, + scan_bytes: w.scan_bytes, + }, + } + } +} + +impl From for SketchWire { + fn from(value: ResourceMeasurements) -> Self { + let r = value.resources; + Self { + build_cpu_ns: r.cpu.build_cpu_ns, + update_cpu_ns: r.cpu.update_cpu_ns, + merge_cpu_ns: r.cpu.merge_cpu_ns, + prepare_cpu_ns: r.cpu.prepare_cpu_ns, + read_cpu_ns: r.cpu.read_cpu_ns, + retained_bytes: r.retained_memory_bytes, + peak_bytes: r.peak_memory_bytes, + serialized_bytes: r.serialized_bytes, + disk_bytes: r.disk_bytes, + scan_bytes: r.scan_bytes, + } + } +} + +impl From for ExactResourceMeasurements { + fn from(w: ExactWire) -> Self { + Self { + resources: PhysicalResources { + cpu: MeasuredCpu { + build_cpu_ns: w.empty_build_cpu_ns, + update_cpu_ns: w.update_cpu_ns, + merge_cpu_ns: w.merge_cpu_ns, + prepare_cpu_ns: w.prepare_cpu_ns, + read_cpu_ns: w.read_cpu_ns, + }, + retained_memory_bytes: w.retained_bytes, + peak_memory_bytes: w.peak_bytes, + serialized_bytes: w.serialized_bytes, + disk_bytes: w.disk_bytes, + scan_bytes: w.scan_bytes, + }, + } + } +} + +impl From for ExactWire { + fn from(value: ExactResourceMeasurements) -> Self { + let r = value.resources; + Self { + empty_build_cpu_ns: r.cpu.build_cpu_ns, + update_cpu_ns: r.cpu.update_cpu_ns, + merge_cpu_ns: r.cpu.merge_cpu_ns, + prepare_cpu_ns: r.cpu.prepare_cpu_ns, + read_cpu_ns: r.cpu.read_cpu_ns, + retained_bytes: r.retained_memory_bytes, + peak_bytes: r.peak_memory_bytes, + serialized_bytes: r.serialized_bytes, + disk_bytes: r.disk_bytes, + scan_bytes: r.scan_bytes, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn measured(value: f64) -> Option { + Some(Measurement { + value, + stddev: Some(0.5), + samples: 5, + method: Some("test only".into()), + }) + } + + /// The same resource dimensions retain uncertainty through either wire adapter. + #[test] + fn shared_resources_round_trip_without_losing_dimensions() { + let resources = PhysicalResources { + cpu: MeasuredCpu { + build_cpu_ns: measured(1.0), + update_cpu_ns: measured(2.0), + merge_cpu_ns: measured(3.0), + prepare_cpu_ns: measured(4.0), + read_cpu_ns: measured(5.0), + }, + peak_memory_bytes: measured(100.0), + retained_memory_bytes: measured(70.0), + scan_bytes: measured(200.0), + serialized_bytes: measured(40.0), + disk_bytes: measured(4096.0), + }; + let sketch = ResourceMeasurements { + resources: resources.clone(), + }; + let exact = ExactResourceMeasurements { resources }; + assert_eq!( + serde_json::from_value::(serde_json::to_value(&sketch).unwrap()) + .unwrap(), + sketch + ); + assert_eq!( + serde_json::from_value::( + serde_json::to_value(&exact).unwrap() + ) + .unwrap(), + exact + ); + } + + /// Archived names remain accepted, but physical fields use canonical names. + #[test] + fn legacy_fields_map_to_shared_resources() { + let value = serde_json::json!({"empty_build_cpu_ns": measured(3.0), "peak_bytes": measured(100.0), "retained_bytes": measured(70.0)}); + let exact: ExactResourceMeasurements = serde_json::from_value(value).unwrap(); + assert_eq!(exact.resources.cpu.build_cpu_ns, measured(3.0)); + assert_eq!(exact.resources.peak_memory_bytes, measured(100.0)); + assert_eq!(exact.resources.retained_memory_bytes, measured(70.0)); + assert!(exact.resources.scan_bytes.is_none()); + assert!(exact.resources.disk_bytes.is_none()); + assert!( + serde_json::from_value::(serde_json::json!({"cpu_ops": 42})) + .is_err() + ); + } +} diff --git a/crates/asap-aware-mapping/src/lib.rs b/crates/asap-aware-mapping/src/lib.rs index 12654b82..27021625 100644 --- a/crates/asap-aware-mapping/src/lib.rs +++ b/crates/asap-aware-mapping/src/lib.rs @@ -185,6 +185,9 @@ pub mod accuracy; pub mod accuracy_reconciliation; pub mod analytical_cost; pub mod cost_model; +pub mod empirical_comparison; +pub mod empirical_cost; +pub mod empirical_resources; pub mod explanation; pub mod grouping; pub mod physical_operator_statistics; diff --git a/crates/asap-aware-mapping/src/query_physical_lowering.rs b/crates/asap-aware-mapping/src/query_physical_lowering.rs index b52f96a7..79a31789 100644 --- a/crates/asap-aware-mapping/src/query_physical_lowering.rs +++ b/crates/asap-aware-mapping/src/query_physical_lowering.rs @@ -1632,7 +1632,7 @@ mod tests { let estimate = estimate_physical_dag(&dag.nodes, &dag.root, &independent_scope, &dag.evidence) .unwrap(); - assert_eq!(estimate.scan_bytes, 1_600); + assert_eq!(estimate.scan_bytes(), 1_600); let shared_provider = |request: PhysicalNodeRequest<'_>| { let (physical_id, statistics) = match request.operator { @@ -1668,8 +1668,8 @@ mod tests { }, ) .unwrap(); - assert_eq!(comparison.raw.scan_bytes, 1_600); - assert_eq!(comparison.candidate.scan_bytes, 800); + assert_eq!(comparison.raw.scan_bytes(), 1_600); + assert_eq!(comparison.candidate.scan_bytes(), 800); let mut drifted_buffer = shared_dag.clone(); drifted_buffer.nodes[0].output_buffer_bytes += 1; @@ -2167,8 +2167,8 @@ mod tests { ]); let dag = lower_query_physical_dag(&root, &scope, &scripted(&provided)).unwrap(); let estimate = estimate_physical_dag(&dag.nodes, &dag.root, &scope, &dag.evidence).unwrap(); - assert_eq!(estimate.cpu_ops, 200.0); - assert_eq!(estimate.scan_bytes, 800); + assert_eq!(estimate.cpu_ops(), 200.0); + assert_eq!(estimate.scan_bytes(), 800); } #[test] diff --git a/crates/asap-aware-mapping/src/summary_maintenance_cost/estimator.rs b/crates/asap-aware-mapping/src/summary_maintenance_cost/estimator.rs index 53c93052..d7c28ae1 100644 --- a/crates/asap-aware-mapping/src/summary_maintenance_cost/estimator.rs +++ b/crates/asap-aware-mapping/src/summary_maintenance_cost/estimator.rs @@ -545,18 +545,18 @@ pub(super) fn estimate_heterogeneous_summary( if !cpu_ops.is_finite() { return Err(AnalyticalCostError::Overflow); } - Ok(ResourceEstimate { + Ok(ResourceEstimate::new( cpu_ops, - peak_memory_bytes: persistent_bytes + persistent_bytes .checked_add(transient_bytes) .and_then(|bytes| bytes.checked_add(ephemeral_state_bytes)) .ok_or(AnalyticalCostError::Overflow)?, - scan_bytes: scans + scans .values() .try_fold(operator_io_bytes, |sum, (_, bytes)| { sum.checked_add(*bytes).ok_or(AnalyticalCostError::Overflow) })?, - }) + )) } fn add_operator_io( @@ -1070,15 +1070,15 @@ pub(super) fn estimate_incremental_summary_maintenance_with_join( .initial_input_bytes .div_ceil(inputs.initial_input_rows) }; - Ok(ResourceEstimate { + Ok(ResourceEstimate::new( cpu_ops, - peak_memory_bytes: retained_bytes + retained_bytes .checked_add(transient_bytes) .and_then(|bytes| bytes.checked_add(join_bytes)) .ok_or(AnalyticalCostError::Overflow)? .max(bootstrap_row_buffer), - scan_bytes: inputs.initial_source_scan_bytes, - }) + inputs.initial_source_scan_bytes, + )) } pub(super) fn lifecycle_row_counts( diff --git a/crates/asap-aware-mapping/src/summary_maintenance_cost/model.rs b/crates/asap-aware-mapping/src/summary_maintenance_cost/model.rs index 1eb0fac7..e7371650 100644 --- a/crates/asap-aware-mapping/src/summary_maintenance_cost/model.rs +++ b/crates/asap-aware-mapping/src/summary_maintenance_cost/model.rs @@ -637,28 +637,22 @@ impl SummaryMaintenanceCostModel { let evidence = self.canonical_inputs(summary)?; let inputs = evidence.inputs; let insert = validated_operator_cpu("insert_cpu_ops", evidence.insert_cpu_ops).ok()?; - let build = self.calibrated(ResourceEstimate { - cpu_ops: inputs.initial_input_rows as f64 - * inputs.bootstrap_window_count as f64 - * insert, - peak_memory_bytes: 0, - scan_bytes: inputs.initial_source_scan_bytes, - })?; - let maintenance = self.calibrated(ResourceEstimate { - cpu_ops: inputs.active_window_count as f64 * insert, - peak_memory_bytes: 0, - scan_bytes: 0, - })?; + let build = self.calibrated(ResourceEstimate::new( + inputs.initial_input_rows as f64 * inputs.bootstrap_window_count as f64 * insert, + 0, + inputs.initial_source_scan_bytes, + ))?; + let maintenance = self.calibrated(ResourceEstimate::new( + inputs.active_window_count as f64 * insert, + 0, + 0, + ))?; let retained = inputs .active_window_count .checked_add(inputs.retained_window_count)? .checked_mul(inputs.physical_summary_count)? .checked_mul(inputs.state_bytes_per_summary)?; - let retention_total = self.calibrated(ResourceEstimate { - cpu_ops: 0.0, - peak_memory_bytes: retained, - scan_bytes: 0, - })?; + let retention_total = self.calibrated(ResourceEstimate::new(0.0, retained, 0))?; let horizon_seconds = horizon.filter(|value| value.0 > 0.0)?.0; Some(SummaryMaintenanceLifecycleCostInputs { build_cost: Some(build), @@ -1059,8 +1053,8 @@ mod tests { ) .unwrap(); // 10 arrivals * 2 active windows * 2 insert ops + 5 reads * 2 summaries. - assert_eq!(estimate.cpu_ops, 50.0); - assert_eq!(estimate.scan_bytes, 0); + assert_eq!(estimate.cpu_ops(), 50.0); + assert_eq!(estimate.scan_bytes(), 0); } #[test] @@ -1112,7 +1106,7 @@ mod tests { ) .unwrap(); // 10 bootstrap rows * 3 windows * 2 insert ops + 5 reads * 2 summaries. - assert_eq!(estimate.cpu_ops, 70.0); + assert_eq!(estimate.cpu_ops(), 70.0); } #[test] @@ -2604,9 +2598,9 @@ mod tests { ) .unwrap(); // 10 bootstrap + 10 arrivals into two active windows; two states read 5 times. - assert_eq!(estimate.cpu_ops, 90.0); - assert_eq!(estimate.peak_memory_bytes, 1_000); - assert_eq!(estimate.scan_bytes, 640); + assert_eq!(estimate.cpu_ops(), 90.0); + assert_eq!(estimate.peak_memory_bytes(), 1_000); + assert_eq!(estimate.scan_bytes(), 640); } #[test] @@ -2636,9 +2630,9 @@ mod tests { }, ) .unwrap(); - assert_eq!(estimate.cpu_ops, 21.0 + 20.0 + 30.0 + 200.0 + 70.0); + assert_eq!(estimate.cpu_ops(), 21.0 + 20.0 + 30.0 + 200.0 + 70.0); // Three persistent windows plus one transient result, for two instances. - assert_eq!(estimate.peak_memory_bytes, 80); + assert_eq!(estimate.peak_memory_bytes(), 80); } #[test] @@ -2762,7 +2756,7 @@ mod tests { .unwrap(); // Two pre-activation arrivals join the bootstrap; eight more are // maintained through the horizon; five reads are served. - assert_eq!(estimate.cpu_ops, 25.0); + assert_eq!(estimate.cpu_ops(), 25.0); } #[test] @@ -2859,8 +2853,8 @@ mod tests { }), ) .unwrap(); - assert_eq!(estimate.cpu_ops, 77.0); - assert_eq!(estimate.peak_memory_bytes, 64); // 4 persistent states + join memory. + assert_eq!(estimate.cpu_ops(), 77.0); + assert_eq!(estimate.peak_memory_bytes(), 64); // 4 persistent states + join memory. } fn summary_with_operations(merge: bool, subtract: bool, delete: bool) -> Rc { diff --git a/crates/asap-aware-mapping/tests/data/offline-evidence-synthetic.json b/crates/asap-aware-mapping/tests/data/offline-evidence-synthetic.json new file mode 100644 index 00000000..77d06194 --- /dev/null +++ b/crates/asap-aware-mapping/tests/data/offline-evidence-synthetic.json @@ -0,0 +1,67 @@ +{ + "schema_version": 1, + "benchmark_version": "synthetic-fixture-v1", + "model_version": "empirical-update-cpu-v1", + "records": [ + { + "id": "synthetic-test-cms", + "algorithm": "Cms", + "params": { + "Cms": { + "width": 272, + "depth": 5 + } + }, + "distribution": { + "id": "synthetic-test-uniform", + "family": "uniform", + "sample_count": 1000, + "distinct_count": 100, + "parameters": { + "seed": 42 + } + }, + "environment": { + "id": "synthetic-test-machine", + "cpu": "synthetic CPU", + "os": "synthetic OS", + "runtime": "synthetic runtime", + "implementation": "SYNTHETIC TEST FIXTURE, NOT MEASURED", + "implementation_version": "test-v1" + }, + "measured_at_unix_seconds": 100, + "valid_until_unix_seconds": 200, + "provenance": { + "command": "synthetic fixture; no benchmark was run", + "dataset": "synthetic fixture", + "source_revision": "test-fixture-v1", + "repetitions": 3 + }, + "metrics": { + "build_cpu_ns": null, + "update_cpu_ns": { + "value": 20, + "stddev": 1, + "samples": 3, + "method": "fabricated values for schema/integration tests only" + }, + "merge_cpu_ns": null, + "read_cpu_ns": null, + "retained_bytes": null, + "peak_bytes": null, + "serialized_bytes": null, + "disk_bytes": null + }, + "error": { + "metric": "absolute relative frequency error (dimensionless)", + "mean": 0.001, + "max": null, + "trials": 3, + "ground_truth_method": "synthetic fixture only", + "query": { + "kind": "point_frequency" + } + } + } + ] +} diff --git a/crates/devtools/src/bin/dag_export.rs b/crates/devtools/src/bin/dag_export.rs index a0c7e146..4f900556 100644 --- a/crates/devtools/src/bin/dag_export.rs +++ b/crates/devtools/src/bin/dag_export.rs @@ -413,17 +413,17 @@ impl ExportPlannerCostModel<'_> { vec![ CostInput { name: "estimated_cpu_ops".into(), - value: resources.cpu_ops, + value: resources.cpu_ops(), unit: Some("operations".into()), }, CostInput { name: "estimated_peak_memory".into(), - value: resources.peak_memory_bytes as f64, + value: resources.peak_memory_bytes() as f64, unit: Some("bytes".into()), }, CostInput { name: "estimated_scan".into(), - value: resources.scan_bytes as f64, + value: resources.scan_bytes() as f64, unit: Some("bytes".into()), }, ] diff --git a/crates/types/src/lib.rs b/crates/types/src/lib.rs index dbc1665c..8b65cb1c 100644 --- a/crates/types/src/lib.rs +++ b/crates/types/src/lib.rs @@ -23,5 +23,6 @@ pub mod cost; pub mod dag_export; pub mod post_asap; pub mod pre_asap; +pub mod resources; pub mod types; pub mod workload; diff --git a/crates/types/src/resources.rs b/crates/types/src/resources.rs new file mode 100644 index 00000000..894e3465 --- /dev/null +++ b/crates/types/src/resources.rs @@ -0,0 +1,131 @@ +//! Resource dimensions shared by analytical estimates and benchmark evidence. +//! +//! The CPU payload establishes its unit and operation scope; CPU operations +//! must never be interpreted as nanoseconds. Byte values can be exact integers +//! or measurements carrying uncertainty. `None` means unavailable, not zero. + +use serde::{Deserialize, Serialize}; + +/// Modeled CPU work, never measured elapsed or process CPU time. +#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ModeledCpu { + pub cpu_ops: f64, +} + +/// A measured quantity whose unit and operation scope are set by its field. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct Measurement { + pub value: f64, + pub stddev: Option, + pub samples: u32, + /// Measurement procedure and scope, such as allocator heap versus payload. + #[serde(default)] + pub method: Option, +} + +/// Process CPU nanoseconds per named operation, never modeled CPU operations. +/// An absent phase was not measured; it does not imply a free operation. +#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct MeasuredCpu { + /// Empty construction; ingestion is charged separately. + pub build_cpu_ns: Option, + pub update_cpu_ns: Option, + pub merge_cpu_ns: Option, + /// One prepare pass after ingestion and before reads. + pub prepare_cpu_ns: Option, + pub read_cpu_ns: Option, +} + +pub type MeasuredResources = PhysicalResources; + +#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct PhysicalResources { + pub cpu: Cpu, + /// Maximum simultaneously live memory within the reported scope. + pub peak_memory_bytes: Option, + /// Memory still retained at the end of the reported scope. + pub retained_memory_bytes: Option, + /// Bytes read by scan operations, not storage occupancy. + pub scan_bytes: Option, + /// Logical encoded snapshot size, not in-memory state size. + pub serialized_bytes: Option, + /// Allocated filesystem space, not bytes read or written over time. + pub disk_bytes: Option, +} + +impl Default for PhysicalResources { + fn default() -> Self { + Self { + cpu: Cpu::default(), + peak_memory_bytes: None, + retained_memory_bytes: None, + scan_bytes: None, + serialized_bytes: None, + disk_bytes: None, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Distinct CPU payloads reject each other's units at the wire boundary; + /// measured phase metadata survives without filling unknown phases. + #[test] + fn cpu_units_and_measurement_metadata_remain_distinct() { + let modeled = ModeledCpu { cpu_ops: 12.0 }; + assert!( + serde_json::from_value::(serde_json::to_value(modeled).unwrap()).is_err() + ); + let measured = MeasuredCpu { + update_cpu_ns: Some(Measurement { + value: 12.0, + stddev: Some(0.5), + samples: 5, + method: Some("test process CPU clock".into()), + }), + ..Default::default() + }; + let encoded = serde_json::to_value(&measured).unwrap(); + assert!(serde_json::from_value::(encoded.clone()).is_err()); + assert_eq!( + serde_json::from_value::(encoded).unwrap(), + measured + ); + assert!(measured.build_cpu_ns.is_none()); + assert!(measured.read_cpu_ns.is_none()); + } + + /// Default resources preserve unavailable dimensions instead of inventing zeros. + #[test] + fn missing_dimensions_remain_unavailable() { + let resources = PhysicalResources::, u64>::default(); + assert_eq!(resources.cpu, None); + assert_eq!(resources.peak_memory_bytes, None); + assert_eq!(resources.retained_memory_bytes, None); + assert_eq!(resources.scan_bytes, None); + assert_eq!(resources.serialized_bytes, None); + assert_eq!(resources.disk_bytes, None); + } + + /// Storage occupancy and scan traffic are independent even though both use bytes. + #[test] + fn byte_dimensions_round_trip_independently() { + let resources = PhysicalResources { + cpu: Some(12.0), + peak_memory_bytes: Some(100), + retained_memory_bytes: Some(70), + scan_bytes: Some(200), + serialized_bytes: Some(40), + disk_bytes: Some(4096), + }; + let json = serde_json::to_value(resources).unwrap(); + let decoded: PhysicalResources, u64> = serde_json::from_value(json).unwrap(); + assert_eq!(decoded, resources); + } +} diff --git a/docs/developer_docs/offline-comparison-evidence.schema.json b/docs/developer_docs/offline-comparison-evidence.schema.json new file mode 100644 index 00000000..8b579dc2 --- /dev/null +++ b/docs/developer_docs/offline-comparison-evidence.schema.json @@ -0,0 +1,225 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "title": "ASAPPlanner fixed-snapshot offline comparison evidence", + "description": "Disjoint CPU phases while state remains alive; no runtime guarantee or live-stream extrapolation.", + "type": "object", + "additionalProperties": false, + "properties": { + "schema_version": { + "const": 1 + }, + "timing_contract": { + "const": "disjoint_live_state_v1" + }, + "sketch_evidence": { + "$ref": "offline-sketch-evidence.schema.json" + }, + "query_bindings": { + "type": "array", + "items": { + "type": "object", + "additionalProperties": false, + "properties": { + "record_id": { + "type": "string", + "minLength": 1 + }, + "query": { + "$ref": "#/$defs/query" + } + }, + "required": [ + "record_id", + "query" + ] + } + }, + "exact_records": { + "type": "array", + "items": { + "$ref": "#/$defs/exact" + } + } + }, + "required": [ + "schema_version", + "timing_contract", + "sketch_evidence", + "query_bindings", + "exact_records" + ], + "$defs": { + "query": { + "type": "object", + "additionalProperties": false, + "required": [ + "kind", + "value_type", + "probe_set" + ], + "properties": { + "kind": { + "const": "point_frequency" + }, + "value_type": { + "const": "i64" + }, + "probe_set": { + "const": "all_distinct_keys" + } + } + }, + "exact": { + "type": "object", + "additionalProperties": false, + "properties": { + "id": { + "type": "string", + "minLength": 1 + }, + "distribution": { + "$ref": "offline-sketch-evidence.schema.json#/$defs/distribution" + }, + "environment": { + "$ref": "offline-sketch-evidence.schema.json#/$defs/environment" + }, + "query": { + "$ref": "#/$defs/query" + }, + "measured_at_unix_seconds": { + "type": "integer", + "minimum": 0 + }, + "valid_until_unix_seconds": { + "type": "integer", + "minimum": 0 + }, + "provenance": { + "$ref": "offline-sketch-evidence.schema.json#/$defs/provenance" + }, + "metrics": { + "description": "Flat v1 exact-reference wire representation of shared PhysicalResources; empty_build_cpu_ns maps to the shared build_cpu_ns field.", + "type": "object", + "additionalProperties": false, + "properties": { + "empty_build_cpu_ns": { + "anyOf": [ + { + "$ref": "offline-sketch-evidence.schema.json#/$defs/measurement" + }, + { + "type": "null" + } + ] + }, + "update_cpu_ns": { + "anyOf": [ + { + "$ref": "offline-sketch-evidence.schema.json#/$defs/measurement" + }, + { + "type": "null" + } + ] + }, + "merge_cpu_ns": { + "description": "Optional measured merge CPU nanoseconds. Its presence alone does not establish post-merge accuracy or enable merged-state recommendations.", + "anyOf": [ + { + "$ref": "offline-sketch-evidence.schema.json#/$defs/measurement" + }, + { + "type": "null" + } + ] + }, + "prepare_cpu_ns": { + "anyOf": [ + { + "$ref": "offline-sketch-evidence.schema.json#/$defs/measurement" + }, + { + "type": "null" + } + ] + }, + "read_cpu_ns": { + "anyOf": [ + { + "$ref": "offline-sketch-evidence.schema.json#/$defs/measurement" + }, + { + "type": "null" + } + ] + }, + "retained_bytes": { + "anyOf": [ + { + "$ref": "offline-sketch-evidence.schema.json#/$defs/measurement" + }, + { + "type": "null" + } + ] + }, + "serialized_bytes": { + "description": "Optional logical encoded snapshot size in bytes; omitted when unavailable by the compatibility serializer.", + "anyOf": [ + { + "$ref": "offline-sketch-evidence.schema.json#/$defs/measurement" + }, + { + "type": "null" + } + ] + }, + "disk_bytes": { + "description": "Optional allocated filesystem bytes; distinct from scan traffic and serialized length. Omitted when unavailable by the compatibility serializer.", + "anyOf": [ + { + "$ref": "offline-sketch-evidence.schema.json#/$defs/measurement" + }, + { + "type": "null" + } + ] + }, + "scan_bytes": { + "description": "Optional bytes read by scan operations; omitted when unavailable by the compatibility serializer.", + "anyOf": [ + { + "$ref": "offline-sketch-evidence.schema.json#/$defs/measurement" + }, + { + "type": "null" + } + ] + }, + "peak_bytes": { + "anyOf": [ + { + "$ref": "offline-sketch-evidence.schema.json#/$defs/measurement" + }, + { + "type": "null" + } + ] + } + }, + "required": [] + } + }, + "required": [ + "id", + "distribution", + "environment", + "query", + "measured_at_unix_seconds", + "valid_until_unix_seconds", + "provenance", + "metrics" + ] + } + } +} diff --git a/docs/developer_docs/offline-sketch-evidence.md b/docs/developer_docs/offline-sketch-evidence.md new file mode 100644 index 00000000..cd331e57 --- /dev/null +++ b/docs/developer_docs/offline-sketch-evidence.md @@ -0,0 +1,162 @@ +# Consuming offline sketch measurements + +This document is for developers integrating sketch-bench with the planner. The +Rust schema is `asap_aware_mapping::empirical_cost::EvidenceArtifact`; its JSON +schema version is `1`. Required artifact-level `benchmark_version` and +`model_version` identify the producer and cost interpretation independently of +the serialization schema. Producers export offline benchmark measurements using +this contract; benchmark tooling is delivered separately from the core provider. + +The checked-in [JSON Schema](offline-sketch-evidence.schema.json) describes the +wire format. `crates/asap-aware-mapping/tests/data/offline-evidence-synthetic.json` +is an explicitly fabricated format fixture, never benchmark evidence. Runtime +validation additionally checks cross-field constraints and matching context. + +Resource values share the internal `asap_types::resources::PhysicalResources` container. Its CPU payload preserves units: `ModeledCpu { cpu_ops: f64 }` +represents modeled operations, while `MeasuredCpu` contains optional measured +`build_cpu_ns`, `update_cpu_ns`, `merge_cpu_ns`, `prepare_cpu_ns`, and `read_cpu_ns` +values. `Measurement` is defined in the shared types crate; +`MeasuredResources` is `PhysicalResources`. Sharing +the container does not convert modeled CPU operations into measured nanoseconds. + +`ResourceMeasurements` and `ExactResourceMeasurements` retain only a public +`resources` value of that shared measured type and provide v1 wire compatibility. +Rust consumers access measured CPU through `metrics.resources.cpu` and byte +dimensions through `metrics.resources`. The JSON remains flat: there is no new +`resources` or `cpu` object inside `metrics`. These names map as follows: + +| Shared internal field | Sketch v1 JSON field | Exact-reference v1 JSON field | +| --- | --- | --- | +| `cpu.build_cpu_ns` | `build_cpu_ns` | `empty_build_cpu_ns` | +| `cpu.update_cpu_ns` | `update_cpu_ns` | `update_cpu_ns` | +| `cpu.merge_cpu_ns` | `merge_cpu_ns` | `merge_cpu_ns` | +| `cpu.prepare_cpu_ns` | `prepare_cpu_ns` | `prepare_cpu_ns` | +| `cpu.read_cpu_ns` | `read_cpu_ns` | `read_cpu_ns` | +| `retained_memory_bytes` | `retained_bytes` | `retained_bytes` | +| `peak_memory_bytes` | `peak_bytes` | `peak_bytes` | +| `scan_bytes` | `scan_bytes` | `scan_bytes` | +| `serialized_bytes` | `serialized_bytes` | `serialized_bytes` | +| `disk_bytes` | `disk_bytes` | `disk_bytes` | + +Existing v1 field names and null behavior are unchanged. Sketch records can +add optional `prepare_cpu_ns` and `scan_bytes`; exact references can add optional +`merge_cpu_ns`, `serialized_bytes`, `disk_bytes`, and `scan_bytes`. These new +fields are omitted when unavailable by the compatibility serializers and accept +either a `Measurement` or null on input. They use the same numeric, sample-count, +and uncertainty validation as existing measurements. Old artifacts need no +rewrite or schema-version change. Representing a resource dimension does not +by itself add a workload operation or establish its semantic applicability. + +Deserialize an artifact and an `EvidenceContext`, then construct an +`EmpiricalEvidenceProvider::new(artifact, context)`. Construction validates schema +version, configuration, provenance and numeric values. `lookup(algorithm, params)` +requires identical sketch parameters, distribution descriptors and environment +descriptors. The context supplies the evaluation timestamp. An expired, future, +missing, incompatible or ambiguous measurement returns a typed error; none of +these conditions supplies a zero cost. Select an explicit context for each +distribution or machine; the provider does not interpolate between datasets. + +Each measured resource is an optional `Measurement` with `value`, optional `stddev`, +`samples`, and optional `method`. CPU fields are process CPU nanoseconds per +operation; `build_cpu_ns` measures empty construction. Building an ingested +snapshot additionally requires `sample_count × update_cpu_ns`; the lifecycle +helper returns that sum only when both measurements exist. Memory and disk +fields are bytes; `scan_bytes` records bytes read by scans, not storage occupancy. +Producer methods must state what +was measured and how normalization was performed. `retained_bytes` is distinct +from `peak_bytes`, `serialized_bytes`, and `disk_bytes`. Counter payload size +does not establish allocator footprint, process RSS or on-disk storage. Absent +measurements preserve the v1 null behavior, except the newly optional fields +listed above are omitted when unavailable. `stddev: null` means uncertainty was not measured, +not that variability is zero. A zero measurement must have actual evidence. + +The record retains the complete command, dataset identity, implementation +revision, environment and repetition count. Distribution parameters should +include generator configuration and seed, or trace checksum and sampling rules. +Validity intervals are supplied by the producer or deployment policy; they are +an explicit applicability assumption, not a measured property. + +`EmpiricalCostModel` implements the planner's existing `CostModel` boundary. +It derives the planner's default parameter configurations for the requested +accuracy, and orders algorithms by mean measured update CPU only when all +candidates have applicable measurements. Otherwise it preserves discovery order. +It returns every candidate and keeps default sizing. Its final candidate scores +retain `DefaultCostModel`'s dimensionless structural meaning; do not label those +scores as CPU or measured savings. + +Deployment cost models can own the provider and call `lookup` with their own +parameter sizing. This preserves the deployment's other cost and capability +hooks. The provider's lifecycle helper returns available build/update CPU costs +for a single independently instantiated state. It deliberately leaves retention, +retirement and read costs unknown. In particular, a point-frequency benchmark +read does not price a total-count read, even when both use CMS. A deployment must +match readout semantics and supply the missing lifecycle and raw-query evidence +before selecting and pricing a complete physical plan. Never combine these +nanosecond costs with CPU operation counts without explicit calibration. + +`error` contains offline observed statistics and a query descriptor. Its metric +name defines what mean/max refer to; null max does not imply a per-key maximum +was measured. Query semantics and error metrics must agree before the evidence +is shown as applicable to a planner query. Offline frequency error does not +establish total-count or quantile error. The adapter never converts observed +errors into formal accuracy guarantees or runtime feedback, and never shrinks +parameters solely because one dataset had low observed error. + +## Query-matched offline recommendations + +`empirical_comparison::recommend_offline` consumes the companion +[`OfflineComparisonEvidence` JSON format](offline-comparison-evidence.schema.json). +This combines the sketch artifact with explicit query bindings and a separately +identified exact implementation measured on the same machine, OS, runtime and +input distribution. Its `disjoint_live_state_v1` timing contract requires +construction, ingestion, exact preparation and read CPU to be timed separately +with retained state alive. The original upstream consuming-wrapper timings must +not be passed as these disjoint phase measurements. + +`MeasurementQueryBinding` is the producer's explicit assertion identifying the +read/error probe population. The consumer checks that binding and the error +record's readout kind/value type; it cannot recover or certify the original +probe set from an aggregate error number alone. + +The supported workload is an immutable i64 point-frequency snapshot, fully +ingested before a sequence of reads from its measured all-distinct-key probe +population. Each state's input size must equal the measured distribution's sample +count. The model multiplies per-state quantities by the number of identical, +independent state instances. Reads use the measured average over this probe +population; arbitrary keys outside that population are not covered. It does not +estimate sliding windows, interleaved updates, post-merge error, persistence CPU +or retirement. Nonzero merges are explicitly unavailable until matching +post-merge error and an exact merge baseline exist. + +The caller supplies an `EmpiricalAccuracyRequirement`: the exact observed error +metric, maximum accepted mean, and minimum number of offline trials. This is +separate from `AccuracyTarget`. Every candidate must match the readout descriptor, +error metric, trial count and all ordinary distribution/configuration/environment +checks. A zero observed error is neither proof of exactness nor a per-key bound. + +For each compatible sketch, CPU is empty construction plus input count times +update CPU plus read count times read CPU. The exact reference also includes one +complete prepare pass. The comparison ends with retained state. These measured +components produce an estimate of the specified execution sequence, not a newly +measured end-to-end runtime. Unknown required costs reject an alternative. The +objective weights CPU nanoseconds and retained byte-seconds explicitly; unknown +memory cannot be used with a nonzero memory weight. Peak requested allocations +and per-state snapshot disk/serialized sizes remain separate optional reported +dimensions. The disk size does not imply that the modeled workload writes files. + +The result retains every rejected candidate and reason, the exact reference, +selected configuration, and dimensional savings. If no sketch both meets the +offline criterion and beats the exact reference, the exact path wins. If the +exact baseline is missing, stale or incompatible, no benefit recommendation is +available. + +Deployments connecting this result to formal planning supply +`formal_minimums: Some(...)`, computed by their existing sizing formulas. +Selection only admits supported algorithms and configurations that dominate +those minima. `parameters_at_least` is a conservative componentwise check; +deployment-specific layout constraints, including power-of-two widths, remain +the deployment's responsibility. `selected_sketch() == None` means preserve the +exact path. A deployment must additionally match its actual point-frequency +query; a CMS configuration match does not authorize applying these error +observations to PromQL `count_over_time` or bare total counts. diff --git a/docs/developer_docs/offline-sketch-evidence.schema.json b/docs/developer_docs/offline-sketch-evidence.schema.json new file mode 100644 index 00000000..1622155e --- /dev/null +++ b/docs/developer_docs/offline-sketch-evidence.schema.json @@ -0,0 +1,797 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "title": "ASAPPlanner offline sketch evidence v1", + "description": "Offline observations only. Runtime validation additionally checks duplicate IDs, validity intervals, finite numbers, sample cardinalities and contextual applicability.", + "type": "object", + "additionalProperties": false, + "properties": { + "schema_version": { + "const": 1 + }, + "benchmark_version": { + "type": "string", + "minLength": 1 + }, + "model_version": { + "const": "empirical-update-cpu-v1" + }, + "records": { + "type": "array", + "items": { + "$ref": "#/$defs/record" + } + } + }, + "required": [ + "schema_version", + "benchmark_version", + "model_version", + "records" + ], + "$defs": { + "measurement": { + "type": "object", + "additionalProperties": false, + "properties": { + "value": { + "type": "number", + "minimum": 0 + }, + "stddev": { + "anyOf": [ + { + "type": "number", + "minimum": 0 + }, + { + "type": "null" + } + ] + }, + "samples": { + "type": "integer", + "minimum": 1 + }, + "method": { + "anyOf": [ + { + "type": "string", + "minLength": 1 + }, + { + "type": "null" + } + ] + } + }, + "required": [ + "value", + "samples" + ] + }, + "distribution": { + "type": "object", + "additionalProperties": false, + "properties": { + "id": { + "type": "string", + "minLength": 1 + }, + "family": { + "type": "string", + "minLength": 1 + }, + "sample_count": { + "type": "integer", + "minimum": 1 + }, + "distinct_count": { + "anyOf": [ + { + "type": "integer", + "minimum": 0 + }, + { + "type": "null" + } + ] + }, + "parameters": { + "type": "object" + } + }, + "required": [ + "id", + "family", + "sample_count", + "parameters" + ] + }, + "environment": { + "type": "object", + "additionalProperties": false, + "properties": { + "id": { + "type": "string", + "minLength": 1 + }, + "cpu": { + "type": "string", + "minLength": 1 + }, + "os": { + "type": "string", + "minLength": 1 + }, + "runtime": { + "type": "string", + "minLength": 1 + }, + "implementation": { + "type": "string", + "minLength": 1 + }, + "implementation_version": { + "type": "string", + "minLength": 1 + } + }, + "required": [ + "id", + "cpu", + "os", + "runtime", + "implementation", + "implementation_version" + ] + }, + "provenance": { + "type": "object", + "additionalProperties": false, + "properties": { + "command": { + "type": "string", + "minLength": 1 + }, + "dataset": { + "type": "string", + "minLength": 1 + }, + "source_revision": { + "type": "string", + "minLength": 1 + }, + "repetitions": { + "type": "integer", + "minimum": 1 + } + }, + "required": [ + "command", + "dataset", + "source_revision", + "repetitions" + ] + }, + "metrics": { + "description": "Flat v1 wire representation of shared PhysicalResources; internal resource nesting does not change this JSON layout.", + "type": "object", + "additionalProperties": false, + "properties": { + "build_cpu_ns": { + "description": "Empty sketch construction CPU nanoseconds; full snapshot build additionally charges sample_count times update_cpu_ns.", + "anyOf": [ + { + "$ref": "#/$defs/measurement" + }, + { + "type": "null" + } + ] + }, + "update_cpu_ns": { + "anyOf": [ + { + "$ref": "#/$defs/measurement" + }, + { + "type": "null" + } + ] + }, + "merge_cpu_ns": { + "anyOf": [ + { + "$ref": "#/$defs/measurement" + }, + { + "type": "null" + } + ] + }, + "read_cpu_ns": { + "anyOf": [ + { + "$ref": "#/$defs/measurement" + }, + { + "type": "null" + } + ] + }, + "prepare_cpu_ns": { + "description": "Optional measured preparation CPU nanoseconds. Omitted when unavailable by the compatibility serializer.", + "anyOf": [ + { + "$ref": "#/$defs/measurement" + }, + { + "type": "null" + } + ] + }, + "retained_bytes": { + "anyOf": [ + { + "$ref": "#/$defs/measurement" + }, + { + "type": "null" + } + ] + }, + "peak_bytes": { + "anyOf": [ + { + "$ref": "#/$defs/measurement" + }, + { + "type": "null" + } + ] + }, + "serialized_bytes": { + "anyOf": [ + { + "$ref": "#/$defs/measurement" + }, + { + "type": "null" + } + ] + }, + "scan_bytes": { + "description": "Optional bytes read by scan operations; distinct from retained memory and allocated disk space. Omitted when unavailable by the compatibility serializer.", + "anyOf": [ + { + "$ref": "#/$defs/measurement" + }, + { + "type": "null" + } + ] + }, + "disk_bytes": { + "anyOf": [ + { + "$ref": "#/$defs/measurement" + }, + { + "type": "null" + } + ] + } + }, + "required": [] + }, + "error": { + "type": "object", + "additionalProperties": false, + "properties": { + "metric": { + "type": "string", + "minLength": 1 + }, + "mean": { + "anyOf": [ + { + "type": "number", + "minimum": 0 + }, + { + "type": "null" + } + ] + }, + "max": { + "anyOf": [ + { + "type": "number", + "minimum": 0 + }, + { + "type": "null" + } + ] + }, + "trials": { + "type": "integer", + "minimum": 1 + }, + "ground_truth_method": { + "type": "string", + "minLength": 1 + }, + "query": {} + }, + "required": [ + "metric", + "trials", + "ground_truth_method", + "query" + ] + }, + "record": { + "type": "object", + "additionalProperties": false, + "properties": { + "id": { + "type": "string", + "minLength": 1 + }, + "algorithm": { + "enum": [ + "Kll", + "Cms", + "Hll", + "DDSketch", + "CmsWithHeap", + "Kmv", + "Theta", + "CountSketch", + "CountSketchWithHeap" + ] + }, + "params": { + "oneOf": [ + { + "type": "object", + "additionalProperties": false, + "properties": { + "Kll": { + "type": "object", + "additionalProperties": false, + "properties": { + "k": { + "type": "integer", + "minimum": 1 + } + }, + "required": [ + "k" + ] + } + }, + "required": [ + "Kll" + ] + }, + { + "type": "object", + "additionalProperties": false, + "properties": { + "Cms": { + "type": "object", + "additionalProperties": false, + "properties": { + "width": { + "type": "integer", + "minimum": 1 + }, + "depth": { + "type": "integer", + "minimum": 1 + } + }, + "required": [ + "width", + "depth" + ] + } + }, + "required": [ + "Cms" + ] + }, + { + "type": "object", + "additionalProperties": false, + "properties": { + "Hll": { + "type": "object", + "additionalProperties": false, + "properties": { + "precision": { + "type": "integer", + "minimum": 4, + "maximum": 18 + } + }, + "required": [ + "precision" + ] + } + }, + "required": [ + "Hll" + ] + }, + { + "type": "object", + "additionalProperties": false, + "properties": { + "DDSketch": { + "type": "object", + "additionalProperties": false, + "properties": { + "alpha": { + "type": "number", + "exclusiveMinimum": 0, + "exclusiveMaximum": 1 + } + }, + "required": [ + "alpha" + ] + } + }, + "required": [ + "DDSketch" + ] + }, + { + "type": "object", + "additionalProperties": false, + "properties": { + "CmsWithHeap": { + "type": "object", + "additionalProperties": false, + "properties": { + "width": { + "type": "integer", + "minimum": 1 + }, + "depth": { + "type": "integer", + "minimum": 1 + }, + "heap_size": { + "type": "integer", + "minimum": 1 + } + }, + "required": [ + "width", + "depth", + "heap_size" + ] + } + }, + "required": [ + "CmsWithHeap" + ] + }, + { + "type": "object", + "additionalProperties": false, + "properties": { + "Kmv": { + "type": "object", + "additionalProperties": false, + "properties": { + "k": { + "type": "integer", + "minimum": 1 + } + }, + "required": [ + "k" + ] + } + }, + "required": [ + "Kmv" + ] + }, + { + "type": "object", + "additionalProperties": false, + "properties": { + "Theta": { + "type": "object", + "additionalProperties": false, + "properties": { + "k": { + "type": "integer", + "minimum": 1 + } + }, + "required": [ + "k" + ] + } + }, + "required": [ + "Theta" + ] + }, + { + "type": "object", + "additionalProperties": false, + "properties": { + "CountSketch": { + "type": "object", + "additionalProperties": false, + "properties": { + "width": { + "type": "integer", + "minimum": 1 + }, + "depth": { + "type": "integer", + "minimum": 1 + } + }, + "required": [ + "width", + "depth" + ] + } + }, + "required": [ + "CountSketch" + ] + }, + { + "type": "object", + "additionalProperties": false, + "properties": { + "CountSketchWithHeap": { + "type": "object", + "additionalProperties": false, + "properties": { + "width": { + "type": "integer", + "minimum": 1 + }, + "depth": { + "type": "integer", + "minimum": 1 + }, + "heap_size": { + "type": "integer", + "minimum": 1 + } + }, + "required": [ + "width", + "depth", + "heap_size" + ] + } + }, + "required": [ + "CountSketchWithHeap" + ] + } + ] + }, + "distribution": { + "$ref": "#/$defs/distribution" + }, + "environment": { + "$ref": "#/$defs/environment" + }, + "measured_at_unix_seconds": { + "type": "integer", + "minimum": 0 + }, + "valid_until_unix_seconds": { + "type": "integer", + "minimum": 0 + }, + "provenance": { + "$ref": "#/$defs/provenance" + }, + "metrics": { + "$ref": "#/$defs/metrics" + }, + "error": { + "anyOf": [ + { + "$ref": "#/$defs/error" + }, + { + "type": "null" + } + ] + } + }, + "required": [ + "id", + "algorithm", + "params", + "distribution", + "environment", + "measured_at_unix_seconds", + "valid_until_unix_seconds", + "provenance", + "metrics" + ], + "allOf": [ + { + "if": { + "properties": { + "algorithm": { + "const": "Kll" + } + } + }, + "then": { + "properties": { + "params": { + "required": [ + "Kll" + ] + } + } + } + }, + { + "if": { + "properties": { + "algorithm": { + "const": "Cms" + } + } + }, + "then": { + "properties": { + "params": { + "required": [ + "Cms" + ] + } + } + } + }, + { + "if": { + "properties": { + "algorithm": { + "const": "Hll" + } + } + }, + "then": { + "properties": { + "params": { + "required": [ + "Hll" + ] + } + } + } + }, + { + "if": { + "properties": { + "algorithm": { + "const": "DDSketch" + } + } + }, + "then": { + "properties": { + "params": { + "required": [ + "DDSketch" + ] + } + } + } + }, + { + "if": { + "properties": { + "algorithm": { + "const": "CmsWithHeap" + } + } + }, + "then": { + "properties": { + "params": { + "required": [ + "CmsWithHeap" + ] + } + } + } + }, + { + "if": { + "properties": { + "algorithm": { + "const": "Kmv" + } + } + }, + "then": { + "properties": { + "params": { + "required": [ + "Kmv" + ] + } + } + } + }, + { + "if": { + "properties": { + "algorithm": { + "const": "Theta" + } + } + }, + "then": { + "properties": { + "params": { + "required": [ + "Theta" + ] + } + } + } + }, + { + "if": { + "properties": { + "algorithm": { + "const": "CountSketch" + } + } + }, + "then": { + "properties": { + "params": { + "required": [ + "CountSketch" + ] + } + } + } + }, + { + "if": { + "properties": { + "algorithm": { + "const": "CountSketchWithHeap" + } + } + }, + "then": { + "properties": { + "params": { + "required": [ + "CountSketchWithHeap" + ] + } + } + } + } + ] + } + } +}