diff --git a/crates/asap-aware-mapping/src/analytical_cost.rs b/crates/asap-aware-mapping/src/analytical_cost.rs index 63b8edf3..ed2a316d 100644 --- a/crates/asap-aware-mapping/src/analytical_cost.rs +++ b/crates/asap-aware-mapping/src/analytical_cost.rs @@ -8,7 +8,7 @@ use std::collections::{HashMap, HashSet}; use asap_types::post_asap::{SketchAlgorithm, SketchParams}; -pub use asap_types::resources::ModeledCpu; +pub use asap_types::resources::{CacheCapacityEvidence, CacheEvidence, CacheProfile, ModeledCpu}; use asap_types::workload::DataArrival; use serde::{Deserialize, Serialize}; @@ -21,7 +21,132 @@ use crate::physical_operator_statistics::{ /// /// This identifies the estimation method, not a physical executor or runtime /// implementation version. -pub const ANALYTICAL_COST_MODEL_VERSION: &str = "analytical-cost-v1"; +pub const ANALYTICAL_COST_MODEL_VERSION: &str = "analytical-cost-v2-cache-aware"; + +#[derive(Debug, Clone, Copy)] +struct ResolvedCacheProfile { + integer_executions: Option, + cpu_execution_factor: f64, + scan_execution_factor: f64, + result_hit_ratio: f64, + buffer_hit_ratio: f64, + buffer_miss_bytes: u64, + buffer_working_set_bytes: u64, +} + +fn resolve_cache_profile( + profile: &CacheProfile, + evaluation_count: u64, + data_arrival: DataArrival, +) -> Result { + if profile.version().trim().is_empty() { + return Err(AnalyticalCostError::InvalidCacheEvidence( + "blank cache profile version", + )); + } + let CacheProfile::Evidence(evidence) = profile else { + return Ok(ResolvedCacheProfile { + integer_executions: Some(evaluation_count), + cpu_execution_factor: evaluation_count as f64, + scan_execution_factor: evaluation_count as f64, + result_hit_ratio: 0.0, + buffer_hit_ratio: 0.0, + buffer_miss_bytes: 1, + buffer_working_set_bytes: 1, + }); + }; + let declared = evidence + .distinct_evaluations + .checked_add(evidence.repeated_identical_evaluations) + .ok_or(AnalyticalCostError::Overflow)?; + if declared != evaluation_count || evidence.distinct_evaluations == 0 { + return Err(AnalyticalCostError::InvalidCacheEvidence( + "cache demand requires an initial distinct evaluation and must sum to evaluation count", + )); + } + fn miss_fraction(value: CacheCapacityEvidence) -> Result { + if value.working_set_bytes == 0 { + return Err(AnalyticalCostError::InvalidCacheEvidence( + "cache working set must be non-zero", + )); + } + Ok( + value.working_set_bytes.saturating_sub(value.capacity_bytes) as f64 + / value.working_set_bytes as f64, + ) + } + if evidence + .result_invalidation_ratio + .is_some_and(|ratio| !ratio.is_finite() || !(0.0..=1.0).contains(&ratio)) + { + return Err(AnalyticalCostError::InvalidCacheEvidence( + "result-cache invalidation ratio must be finite and in [0, 1]", + )); + } + let invalidation = + match data_arrival { + DataArrival::AtRest => { + if evidence + .result_invalidation_ratio + .is_some_and(|ratio| ratio != 0.0) + { + return Err(AnalyticalCostError::InvalidCacheEvidence( + "at-rest data cannot invalidate cached results", + )); + } + 0.0 + } + DataArrival::ContinuouslyIngesting => evidence.result_invalidation_ratio.ok_or( + AnalyticalCostError::InvalidCacheEvidence( + "continuous ingestion requires result-cache invalidation evidence", + ), + )?, + DataArrival::Mixed | DataArrival::Unknown => { + return Err(AnalyticalCostError::InvalidCacheEvidence( + "cache invalidation cannot be derived for mixed or unknown data arrival", + )); + } + }; + // Subtract integer byte counts before conversion: 1 - residency can + // round a nonempty uncovered working set to zero near full capacity. + let result_miss_ratio = + invalidation + miss_fraction(evidence.result_cache)? * (1.0 - invalidation); + let buffer_miss_ratio = miss_fraction(evidence.buffer_cache)?; + let result_hit_ratio = 1.0 - result_miss_ratio; + let buffer_hit_ratio = 1.0 - buffer_miss_ratio; + let cpu_execution_factor = evidence.distinct_evaluations as f64 + + evidence.repeated_identical_evaluations as f64 * result_miss_ratio; + Ok(ResolvedCacheProfile { + integer_executions: if evidence.result_cache.capacity_bytes == 0 || invalidation == 1.0 { + Some(evaluation_count) + } else if evidence.result_cache.capacity_bytes >= evidence.result_cache.working_set_bytes + && invalidation == 0.0 + { + Some(evidence.distinct_evaluations) + } else { + None + }, + cpu_execution_factor, + scan_execution_factor: cpu_execution_factor * buffer_miss_ratio, + result_hit_ratio, + buffer_hit_ratio, + buffer_miss_bytes: evidence + .buffer_cache + .working_set_bytes + .saturating_sub(evidence.buffer_cache.capacity_bytes), + buffer_working_set_bytes: evidence.buffer_cache.working_set_bytes, + }) +} + +/// Derive cache hit ratios from the shared assumptions for this workload. +pub fn cache_hit_ratios( + profile: &CacheProfile, + evaluation_count: u64, + data_arrival: DataArrival, +) -> Result<(f64, f64), AnalyticalCostError> { + let resolved = resolve_cache_profile(profile, evaluation_count, data_arrival)?; + Ok((resolved.result_hit_ratio, resolved.buffer_hit_ratio)) +} /// Conversion from physical dimensions to one deployment-specific objective. /// Memory's coefficient means cost units per retained byte over this model's @@ -342,6 +467,7 @@ pub struct PhysicalDagEstimateRequest<'a> { pub root: &'a str, pub scope: &'a ComparisonScope, pub statistics: &'a dyn OperatorStatisticsProvider, + pub cache_profile: &'a CacheProfile, } #[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)] @@ -357,13 +483,25 @@ pub fn estimate_physical_dag_comparison( candidate: PhysicalDagEstimateRequest<'_>, ) -> Result { validate_comparison_scopes(raw.scope, candidate.scope)?; + if raw.cache_profile != candidate.cache_profile { + return Err(AnalyticalCostError::ComparisonScopeMismatch( + "cache profile", + )); + } Ok(PhysicalDagComparisonEstimate { - raw: estimate_physical_dag(raw.nodes, raw.root, raw.scope, raw.statistics)?, - candidate: estimate_physical_dag( + raw: estimate_physical_dag_with_cache( + raw.nodes, + raw.root, + raw.scope, + raw.statistics, + raw.cache_profile, + )?, + candidate: estimate_physical_dag_with_cache( candidate.nodes, candidate.root, candidate.scope, candidate.statistics, + candidate.cache_profile, )?, }) } @@ -376,8 +514,19 @@ pub fn estimate_physical_dag( root: &str, scope: &ComparisonScope, statistics: &(impl OperatorStatisticsProvider + ?Sized), +) -> Result { + estimate_physical_dag_with_cache(nodes, root, scope, statistics, &CacheProfile::no_cache()) +} + +pub fn estimate_physical_dag_with_cache( + nodes: &[PhysicalDagNode], + root: &str, + scope: &ComparisonScope, + statistics: &(impl OperatorStatisticsProvider + ?Sized), + cache_profile: &CacheProfile, ) -> Result { let evaluation_count = scope.validate()?; + let cache = resolve_cache_profile(cache_profile, evaluation_count, scope.data_arrival)?; let by_id: HashMap<&str, &PhysicalDagNode> = nodes.iter().map(|n| (n.id.as_str(), n)).collect(); if by_id.len() != nodes.len() { return Err(AnalyticalCostError::InvalidPhysicalDag("duplicate node id")); @@ -493,18 +642,40 @@ pub fn estimate_physical_dag( for id in order { let node = by_id[id]; let local = estimate_operator(node.operator, resolved_statistics[id].clone())?; - let executions = match node.execution { - ExecutionMultiplicity::Once => 1, - ExecutionMultiplicity::PerEvaluation => evaluation_count, + let (cpu_executions, scan_executions) = match node.execution { + ExecutionMultiplicity::Once => ( + 1.0, + cache.buffer_miss_bytes as f64 / cache.buffer_working_set_bytes as f64, + ), + ExecutionMultiplicity::PerEvaluation => { + (cache.cpu_execution_factor, cache.scan_execution_factor) + } + }; + cpu_ops += local.cpu_ops() * cpu_executions; + let integer_executions = match node.execution { + ExecutionMultiplicity::Once => Some(1), + ExecutionMultiplicity::PerEvaluation => cache.integer_executions, + }; + let local_scan = if let Some(executions) = integer_executions { + let total = u128::from(local.scan_bytes()) * u128::from(executions); + let denominator = u128::from(cache.buffer_working_set_bytes); + let numerator = u128::from(cache.buffer_miss_bytes); + // Split the product before scaling. Both products fit u128 because + // numerator <= denominator; retain exact bytes even above 2^53. + let adjusted = (total / denominator) * numerator + + ((total % denominator) * numerator).div_ceil(denominator); + u64::try_from(adjusted).map_err(|_| AnalyticalCostError::Overflow)? + } else { + let adjusted = local.scan_bytes() as f64 * scan_executions; + // u64::MAX rounds up to 2^64 in f64. Equality is already outside + // the integer domain; casting it would silently saturate. + if !adjusted.is_finite() || adjusted >= u64::MAX as f64 { + return Err(AnalyticalCostError::Overflow); + } + adjusted.ceil() as u64 }; - cpu_ops += local.cpu_ops() * executions as f64; scan_bytes = scan_bytes - .checked_add( - local - .scan_bytes() - .checked_mul(executions) - .ok_or(AnalyticalCostError::Overflow)?, - ) + .checked_add(local_scan) .ok_or(AnalyticalCostError::Overflow)?; peak_memory_bytes = peak_memory_bytes.max( live_bytes @@ -2010,6 +2181,8 @@ pub enum AnalyticalCostError { InvalidCalibration(&'static str, f64), #[error("at least one calibration coefficient must be positive")] ZeroCalibration, + #[error("invalid or incomplete cache evidence: {0}")] + InvalidCacheEvidence(&'static str), #[error("algorithm {0:?} does not match parameters {1:?}")] ParameterMismatch(SketchAlgorithm, SketchParams), #[error("{0} needs a value-range/bin-count model before it can be estimated")] @@ -3256,6 +3429,7 @@ mod tests { )]); let raw_scope = comparison_scope(); let mut candidate_scope = raw_scope.clone(); + let no_cache = CacheProfile::no_cache(); candidate_scope.sources[0].source_snapshot_id = "catalog-version-43".into(); assert_eq!( @@ -3265,12 +3439,14 @@ mod tests { root: "scan", scope: &raw_scope, statistics: &provided, + cache_profile: &no_cache, }, PhysicalDagEstimateRequest { nodes: &nodes, root: "scan", scope: &candidate_scope, statistics: &provided, + cache_profile: &no_cache, }, ), Err(AnalyticalCostError::ComparisonScopeMismatch("sources")) @@ -3455,4 +3631,367 @@ mod tests { .is_err() ); } + + fn cache_test_scan() -> (Vec, HashMap) { + let edge = EdgeStatistics { + rows: 100, + bytes: 1_000, + }; + ( + vec![PhysicalDagNode { + id: "scan".into(), + operator: PhysicalOperator::Scan, + children: vec![], + source_coverage: Some(comparison_scope().sources[0].clone()), + output_buffer_bytes: 0, + retained_bytes: 0, + execution: ExecutionMultiplicity::PerEvaluation, + }], + HashMap::from([( + "scan".into(), + OperatorStatistics::Scan { + edges: unary_edges(edge, edge), + source_read_bytes: 1_000, + }, + )]), + ) + } + + fn cache_profile(result_capacity: u64, buffer_capacity: u64) -> CacheProfile { + CacheProfile::Evidence(CacheEvidence { + version: "cache-evidence-7".into(), + distinct_evaluations: 2, + repeated_identical_evaluations: 4, + result_cache: CacheCapacityEvidence { + working_set_bytes: 100, + capacity_bytes: result_capacity, + }, + buffer_cache: CacheCapacityEvidence { + working_set_bytes: 100, + capacity_bytes: buffer_capacity, + }, + result_invalidation_ratio: None, + }) + } + + // Legacy mapping imports are aliases of the shared schema, not a second type. + #[test] + fn shared_cache_types_work_through_legacy_mapping_imports() { + let central: asap_types::resources::CacheProfile = cache_profile(100, 50); + let legacy: CacheProfile = + serde_json::from_value(serde_json::to_value(¢ral).unwrap()).unwrap(); + fn accepts_shared(_: &asap_types::resources::CacheProfile) {} + accepts_shared(&legacy); + assert_eq!(central, legacy); + assert_eq!( + cache_hit_ratios(¢ral, 6, DataArrival::AtRest).unwrap(), + (1.0, 0.5) + ); + } + + #[test] + fn no_cache_reproduces_legacy_estimate_and_cache_effects_are_distinct() { + let (nodes, evidence) = cache_test_scan(); + let scope = comparison_scope(); + let legacy = estimate_physical_dag(&nodes, "scan", &scope, &evidence).unwrap(); + let explicit_no_cache = estimate_physical_dag_with_cache( + &nodes, + "scan", + &scope, + &evidence, + &CacheProfile::no_cache(), + ) + .unwrap(); + assert_eq!(legacy, explicit_no_cache); + assert_eq!(legacy.cpu_ops(), 600.0); + assert_eq!(legacy.scan_bytes(), 6_000); + + let result_only = estimate_physical_dag_with_cache( + &nodes, + "scan", + &scope, + &evidence, + &cache_profile(100, 0), + ) + .unwrap(); + assert_eq!(result_only.cpu_ops(), 200.0); + assert_eq!(result_only.scan_bytes(), 2_000); + + let result_and_buffer = estimate_physical_dag_with_cache( + &nodes, + "scan", + &scope, + &evidence, + &cache_profile(100, 50), + ) + .unwrap(); + assert_eq!(result_and_buffer.cpu_ops(), result_only.cpu_ops()); + assert_eq!(result_and_buffer.scan_bytes(), 1_000); + } + + #[test] + fn cache_hits_are_monotone_and_reduce_summary_benefit() { + let (nodes, evidence) = cache_test_scan(); + let scope = comparison_scope(); + let low = estimate_physical_dag_with_cache( + &nodes, + "scan", + &scope, + &evidence, + &cache_profile(0, 0), + ) + .unwrap(); + let high = estimate_physical_dag_with_cache( + &nodes, + "scan", + &scope, + &evidence, + &cache_profile(100, 0), + ) + .unwrap(); + assert!(high.cpu_ops() <= low.cpu_ops()); + assert!(high.scan_bytes() <= low.scan_bytes()); + + // A build-once summary has fixed work; as exact result-cache hits + // rise, the raw-minus-summary advantage cannot increase. + let mut summary_nodes = nodes.clone(); + summary_nodes[0].execution = ExecutionMultiplicity::Once; + let summary = estimate_physical_dag_with_cache( + &summary_nodes, + "scan", + &scope, + &evidence, + &cache_profile(0, 0), + ) + .unwrap(); + assert!(high.cpu_ops() - summary.cpu_ops() <= low.cpu_ops() - summary.cpu_ops()); + } + + #[test] + fn cache_evidence_fails_closed_for_bad_demand_and_streaming_invalidation() { + let mut profile = cache_profile(100, 100); + let CacheProfile::Evidence(evidence) = &mut profile else { + unreachable!() + }; + evidence.distinct_evaluations = 1; + assert!(matches!( + cache_hit_ratios(&profile, 6, DataArrival::AtRest), + Err(AnalyticalCostError::InvalidCacheEvidence(_)) + )); + + let profile = cache_profile(100, 100); + assert!(matches!( + cache_hit_ratios(&profile, 6, DataArrival::ContinuouslyIngesting), + Err(AnalyticalCostError::InvalidCacheEvidence(_)) + )); + } + + // Cache-enabled I/O must keep exact integers and reject unrepresentable totals. + #[test] + fn cache_io_preserves_large_integer_bytes_and_detects_overflow() { + let (mut nodes, mut evidence) = cache_test_scan(); + let mut scope = comparison_scope(); + scope.horizon.0 = 20_000; + let mut cache = cache_profile(0, 0); + let CacheProfile::Evidence(inputs) = &mut cache else { + unreachable!() + }; + inputs.distinct_evaluations = 1; + inputs.repeated_identical_evaluations = 1; + let OperatorStatistics::Scan { + source_read_bytes, .. + } = evidence.get_mut("scan").unwrap() + else { + unreachable!() + }; + *source_read_bytes = 1_u64 << 63; + assert_eq!( + estimate_physical_dag_with_cache(&nodes, "scan", &scope, &evidence, &cache), + Err(AnalyticalCostError::Overflow) + ); + let OperatorStatistics::Scan { + source_read_bytes, .. + } = evidence.get_mut("scan").unwrap() + else { + unreachable!() + }; + *source_read_bytes = (1_u64 << 53) + 1; + nodes[0].execution = ExecutionMultiplicity::Once; + assert_eq!( + estimate_physical_dag_with_cache(&nodes, "scan", &scope, &evidence, &cache) + .unwrap() + .scan_bytes(), + (1_u64 << 53) + 1 + ); + } + + // Repeated demand needs an initial evaluation; contradictory AtRest evidence is invalid. + #[test] + fn cache_evidence_rejects_impossible_demand_and_invalid_at_rest_invalidation() { + let mut cache = cache_profile(100, 0); + let CacheProfile::Evidence(inputs) = &mut cache else { + unreachable!() + }; + inputs.distinct_evaluations = 0; + inputs.repeated_identical_evaluations = 6; + assert!(cache_hit_ratios(&cache, 6, DataArrival::AtRest).is_err()); + for ratio in [f64::NAN, f64::INFINITY, -1.0, 0.5] { + let mut cache = cache_profile(100, 0); + let CacheProfile::Evidence(inputs) = &mut cache else { + unreachable!() + }; + inputs.result_invalidation_ratio = Some(ratio); + assert!(cache_hit_ratios(&cache, 6, DataArrival::AtRest).is_err()); + } + } + + // Buffer residency affects source reads equally for raw scans and summary builds. + #[test] + fn buffer_cache_applies_to_build_once_scans_without_reducing_cpu() { + let (mut nodes, evidence) = cache_test_scan(); + nodes[0].execution = ExecutionMultiplicity::Once; + let scope = comparison_scope(); + let estimate = estimate_physical_dag_with_cache( + &nodes, + "scan", + &scope, + &evidence, + &cache_profile(100, 50), + ) + .unwrap(); + assert_eq!(estimate.cpu_ops(), 100.0); + assert_eq!(estimate.scan_bytes(), 500); + } + + // A sub-ULP uncovered fraction must not become a fabricated full cache hit. + #[test] + fn partial_cache_preserves_tiny_buffer_misses() { + let (mut nodes, mut evidence) = cache_test_scan(); + nodes[0].execution = ExecutionMultiplicity::Once; + let working_set = 1_u64 << 63; + let OperatorStatistics::Scan { + source_read_bytes, .. + } = evidence.get_mut("scan").unwrap() + else { + unreachable!() + }; + *source_read_bytes = working_set; + let mut profile = cache_profile(0, 0); + let CacheProfile::Evidence(inputs) = &mut profile else { + unreachable!() + }; + inputs.buffer_cache = CacheCapacityEvidence { + working_set_bytes: working_set, + capacity_bytes: working_set - 1, + }; + let estimate = estimate_physical_dag_with_cache( + &nodes, + "scan", + &comparison_scope(), + &evidence, + &profile, + ) + .unwrap(); + assert_eq!(estimate.scan_bytes(), 1); + } + + // A result-cache miss remains chargeable even when the displayed hit rounds to one. + #[test] + fn partial_cache_preserves_tiny_result_misses() { + let working_set = 1_u64 << 63; + let mut profile = cache_profile(0, 0); + let CacheProfile::Evidence(inputs) = &mut profile else { + unreachable!() + }; + inputs.distinct_evaluations = 1; + inputs.repeated_identical_evaluations = working_set; + inputs.result_cache = CacheCapacityEvidence { + working_set_bytes: working_set, + capacity_bytes: working_set - 1, + }; + assert_eq!( + resolve_cache_profile(&profile, working_set + 1, DataArrival::AtRest) + .unwrap() + .cpu_execution_factor, + 2.0 + ); + } + + // Byte accounting remains exact above f64 integer precision after cache scaling. + #[test] + fn partial_buffer_cache_rounds_integer_bytes_without_precision_loss() { + let (mut nodes, mut evidence) = cache_test_scan(); + nodes[0].execution = ExecutionMultiplicity::Once; + let OperatorStatistics::Scan { + source_read_bytes, .. + } = evidence.get_mut("scan").unwrap() + else { + unreachable!() + }; + *source_read_bytes = (1_u64 << 54) + 2; + let estimate = estimate_physical_dag_with_cache( + &nodes, + "scan", + &comparison_scope(), + &evidence, + &cache_profile(0, 50), + ) + .unwrap(); + assert_eq!(estimate.scan_bytes(), (1_u64 << 53) + 1); + } + + // A one-third miss fraction must not round five ordinary bytes up to six. + #[test] + fn partial_buffer_cache_does_not_round_an_exact_small_total_up() { + let (mut nodes, mut evidence) = cache_test_scan(); + nodes[0].execution = ExecutionMultiplicity::Once; + let OperatorStatistics::Scan { + source_read_bytes, .. + } = evidence.get_mut("scan").unwrap() + else { + unreachable!() + }; + *source_read_bytes = 15; + let mut profile = cache_profile(0, 0); + let CacheProfile::Evidence(inputs) = &mut profile else { + unreachable!() + }; + inputs.buffer_cache = CacheCapacityEvidence { + working_set_bytes: 3, + capacity_bytes: 2, + }; + assert_eq!( + estimate_physical_dag_with_cache( + &nodes, + "scan", + &comparison_scope(), + &evidence, + &profile + ) + .unwrap() + .scan_bytes(), + 5 + ); + } + + // Declared invalidations bound otherwise-resident results under ingestion. + #[test] + fn streaming_invalidation_bounds_result_hits_and_changes_physical_work() { + let (nodes, evidence) = cache_test_scan(); + let mut scope = comparison_scope(); + scope.data_arrival = DataArrival::ContinuouslyIngesting; + let mut profile = cache_profile(100, 50); + let CacheProfile::Evidence(inputs) = &mut profile else { + unreachable!() + }; + inputs.result_invalidation_ratio = Some(0.5); + let estimate = + estimate_physical_dag_with_cache(&nodes, "scan", &scope, &evidence, &profile).unwrap(); + assert_eq!( + cache_hit_ratios(&profile, 6, scope.data_arrival).unwrap(), + (0.5, 0.5) + ); + assert_eq!(estimate.cpu_ops(), 400.0); + assert_eq!(estimate.scan_bytes(), 2_000); + } } diff --git a/crates/asap-aware-mapping/src/physical_plan_cost_model.rs b/crates/asap-aware-mapping/src/physical_plan_cost_model.rs index 0761e326..6d2d8245 100644 --- a/crates/asap-aware-mapping/src/physical_plan_cost_model.rs +++ b/crates/asap-aware-mapping/src/physical_plan_cost_model.rs @@ -4,6 +4,7 @@ use std::{cell::RefCell, rc::Rc}; use asap_types::post_asap::{SketchAlgorithm, SummaryExpr, SummaryNode}; use asap_types::pre_asap::{AggIntent, QueryExpr}; +use asap_types::resources::CacheProfile; use crate::analytical_cost::{ estimate_physical_dag_comparison, AnalyticalCostError, @@ -27,6 +28,7 @@ use crate::replacement::{Replacement, ReplacementSubDAG, TargetSubDAG}; pub struct PhysicalEvidenceSnapshot { pub version: String, pub scope: ComparisonScope, + pub cache_profile: CacheProfile, } /// Deployment evidence needed to price one planner alternative. @@ -90,6 +92,11 @@ impl<'a> PhysicalPlanCostModel<'a> { provider: &'a dyn PlannerPhysicalPlanProvider, calibration: ResourceCalibration, ) -> Result { + if calibration.version.trim().is_empty() { + return Err(AnalyticalCostError::MissingOrStale( + "resource_calibration.version", + )); + } calibration.validate()?; Ok(Self { provider, @@ -159,12 +166,14 @@ impl<'a> PhysicalPlanCostModel<'a> { root: &raw.root, scope, statistics: &raw, + cache_profile: &snapshot.cache_profile, }, PhysicalDagEstimateRequest { nodes: &replacement.nodes, root: &replacement.root, scope, statistics: &replacement, + cache_profile: &snapshot.cache_profile, }, )?; let raw_cost = Cost(resources.raw.calibrated_cost(&self.calibration)?); @@ -413,6 +422,7 @@ mod tests { Ok(PhysicalEvidenceSnapshot { version: "test-snapshot-1".into(), scope: scope(), + cache_profile: CacheProfile::no_cache(), }) } @@ -482,6 +492,99 @@ mod tests { ); } + // Ranking must retain the uncovered byte of a nearly resident buffer cache. + #[test] + fn tiny_buffer_misses_still_affect_global_selection() { + use crate::analytical_cost::{CacheCapacityEvidence, CacheEvidence}; + const WORKING_SET: u64 = 1_u64 << 63; + struct AlmostResident(TestProvider); + impl PlannerPhysicalPlanProvider for AlmostResident { + fn capture_evidence_snapshot( + &self, + target: &TargetSubDAG<'_>, + ) -> Result { + let mut snapshot = self.0.capture_evidence_snapshot(target)?; + snapshot.cache_profile = CacheProfile::Evidence(CacheEvidence { + version: "almost-resident-v1".into(), + distinct_evaluations: 10, + repeated_identical_evaluations: 0, + result_invalidation_ratio: None, + result_cache: CacheCapacityEvidence { + working_set_bytes: 1, + capacity_bytes: 0, + }, + buffer_cache: CacheCapacityEvidence { + working_set_bytes: WORKING_SET, + capacity_bytes: WORKING_SET - 1, + }, + }); + Ok(snapshot) + } + fn query_node_evidence( + &self, + snapshot: &PhysicalEvidenceSnapshot, + request: PhysicalNodeRequest<'_>, + ) -> Result { + let mut evidence = self.0.query_node_evidence(snapshot, request)?; + if let OperatorStatistics::Scan { + source_read_bytes, .. + } = &mut evidence.statistics + { + *source_read_bytes = WORKING_SET; + } + Ok(evidence) + } + fn summary_physical_dag( + &self, + snapshot: &PhysicalEvidenceSnapshot, + summary: &Rc, + target: &TargetSubDAG<'_>, + ) -> Result { + self.0.summary_physical_dag(snapshot, summary, target) + } + } + let space = crate::replacement::search_workload_with( + vec![("q", query())], + &crate::replacement::default_strategies(), + ); + let provider = AlmostResident(TestProvider::new(true, WORKING_SET)); + let model = PhysicalPlanCostModel::new( + &provider, + ResourceCalibration { + cost_per_cpu_op: 0.0, + cost_per_scan_byte: 1.0, + cost_per_retained_byte: 0.0, + version: "disk-only-v1".into(), + }, + ) + .unwrap(); + let selected = space.global_selection(&model); + assert!( + selected + .for_target(&space.roots[0].1) + .unwrap() + .chosen + .is_some(), + "one remaining build read must rank ahead of ten remaining raw reads" + ); + } + + // Versioned physical ranking cannot accept an anonymous calibration generation. + #[test] + fn blank_calibration_version_is_rejected_before_ranking() { + let provider = TestProvider::new(true, 800); + for version in ["", " \t\n"] { + let mut coefficients = calibration(); + coefficients.version = version.into(); + assert!(matches!( + PhysicalPlanCostModel::new(&provider, coefficients), + Err(AnalyticalCostError::MissingOrStale( + "resource_calibration.version" + )) + )); + } + } + #[test] fn missing_summary_evidence_keeps_the_raw_target() { let root = query(); @@ -557,6 +660,7 @@ mod tests { Ok(PhysicalEvidenceSnapshot { version: " \t".into(), scope: scope(), + cache_profile: CacheProfile::no_cache(), }) } diff --git a/crates/asap-aware-mapping/src/query_physical_lowering.rs b/crates/asap-aware-mapping/src/query_physical_lowering.rs index 79a31789..39d04ced 100644 --- a/crates/asap-aware-mapping/src/query_physical_lowering.rs +++ b/crates/asap-aware-mapping/src/query_physical_lowering.rs @@ -1647,6 +1647,7 @@ mod tests { }) }; let shared_scope = scope(vec![source_coverage]); + let no_cache = crate::analytical_cost::CacheProfile::no_cache(); let shared_dag = lower_query_physical_dag(&root, &shared_scope, &shared_provider).unwrap(); assert_eq!(shared_dag.nodes.len(), 2); assert_eq!( @@ -1659,12 +1660,14 @@ mod tests { root: &dag.root, scope: &independent_scope, statistics: &dag, + cache_profile: &no_cache, }, PhysicalDagEstimateRequest { nodes: &shared_dag.nodes, root: &shared_dag.root, scope: &shared_scope, statistics: &shared_dag, + cache_profile: &no_cache, }, ) .unwrap(); diff --git a/crates/devtools/src/bin/dag_export.rs b/crates/devtools/src/bin/dag_export.rs index 4f900556..75cc3818 100644 --- a/crates/devtools/src/bin/dag_export.rs +++ b/crates/devtools/src/bin/dag_export.rs @@ -55,8 +55,8 @@ use std::rc::Rc; use std::time::Instant; use asap_aware_mapping::analytical_cost::{ - AnalyticalCostError, EvidenceBackedPhysicalDag as PhysicalDag, PhysicalNodeEvidence, - ResourceCalibration, ANALYTICAL_COST_MODEL_VERSION, + cache_hit_ratios, AnalyticalCostError, EvidenceBackedPhysicalDag as PhysicalDag, + PhysicalNodeEvidence, ResourceCalibration, ANALYTICAL_COST_MODEL_VERSION, }; #[cfg(test)] use asap_aware_mapping::cost_model::DefaultCostModel; @@ -82,6 +82,7 @@ use asap_types::post_asap::{CompositionOperator, SketchQuery, SummaryFamilyType} use asap_types::pre_asap::cse::{structural_hash, HashCache}; use asap_types::pre_asap::query_expr::QueryExpr; use asap_types::pre_asap::schema::{Column, DataType, Schema}; +use asap_types::resources::CacheProfile; use asap_types::types::AccuracyTarget; #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] @@ -126,6 +127,8 @@ struct ComparisonScopeEvidence { lookback_ms: Option, as_of_ms: Option, sources: Vec, + #[serde(default = "CacheProfile::no_cache")] + cache_profile: CacheProfile, } impl ComparisonScopeEvidence { @@ -293,6 +296,7 @@ impl PlannerPhysicalPlanProvider for ExportPhysicalProvider<'_> { Ok(PhysicalEvidenceSnapshot { version: self.evidence_version.into(), scope: self.target.scope.resolve()?, + cache_profile: self.target.scope.cache_profile.clone(), }) } @@ -409,8 +413,67 @@ impl ExportPlannerCostModel<'_> { return winner_cost_annotations(); } let version = format!("{}+{}", ANALYTICAL_COST_MODEL_VERSION, calibration.version); + let scope = &provider.target.scope; + let Ok((result_hits, buffer_hits)) = cache_hit_ratios( + &scope.cache_profile, + scope.evaluation_count, + scope.data_arrival, + ) else { + return winner_cost_annotations(); + }; + let input = |name: &str, value: f64, unit: &str| CostInput { + name: name.into(), + value, + unit: Some(unit.into()), + }; + let mut cache_inputs = vec![ + input( + "evaluation_count", + scope.evaluation_count as f64, + "evaluations", + ), + input("result_cache_hit_ratio", result_hits, "ratio"), + input("buffer_cache_hit_ratio", buffer_hits, "ratio"), + ]; + if let CacheProfile::Evidence(evidence) = &scope.cache_profile { + cache_inputs.extend([ + input( + "distinct_evaluations", + evidence.distinct_evaluations as f64, + "evaluations", + ), + input( + "repeated_identical_evaluations", + evidence.repeated_identical_evaluations as f64, + "evaluations", + ), + input( + "result_cache_working_set", + evidence.result_cache.working_set_bytes as f64, + "bytes", + ), + input( + "result_cache_capacity", + evidence.result_cache.capacity_bytes as f64, + "bytes", + ), + input( + "buffer_cache_working_set", + evidence.buffer_cache.working_set_bytes as f64, + "bytes", + ), + input( + "buffer_cache_capacity", + evidence.buffer_cache.capacity_bytes as f64, + "bytes", + ), + ]); + if let Some(ratio) = evidence.result_invalidation_ratio { + cache_inputs.push(input("result_cache_invalidation_ratio", ratio, "ratio")); + } + } let inputs = |resources: asap_aware_mapping::analytical_cost::ResourceEstimate| { - vec![ + let mut inputs = vec![ CostInput { name: "estimated_cpu_ops".into(), value: resources.cpu_ops(), @@ -426,7 +489,9 @@ impl ExportPlannerCostModel<'_> { value: resources.scan_bytes() as f64, unit: Some("bytes".into()), }, - ] + ]; + inputs.extend(cache_inputs.iter().cloned()); + inputs }; let baseline = CostAnnotation::modeled( estimate.raw_cost.0, @@ -434,7 +499,8 @@ impl ExportPlannerCostModel<'_> { &version, inputs(estimate.resources.raw), ) - .with_evidence_version(&self.document.evidence_version); + .with_evidence_version(&self.document.evidence_version) + .with_cache_profile(snapshot_cache_version(&provider)); let selected = CostAnnotation::modeled( estimate.candidate_cost.0, CostUnit::CostUnits, @@ -442,7 +508,8 @@ impl ExportPlannerCostModel<'_> { inputs(estimate.resources.candidate), ) .with_baseline(BaselineRef::PreAsapRecomputation, estimate.raw_cost.0) - .with_evidence_version(&self.document.evidence_version); + .with_evidence_version(&self.document.evidence_version) + .with_cache_profile(snapshot_cache_version(&provider)); let benefit = CostAnnotation { value: selected.delta, unit: CostUnit::CostUnits, @@ -452,6 +519,7 @@ impl ExportPlannerCostModel<'_> { benefit_ratio: selected.benefit_ratio, model_version: Some(version), evidence_version: Some(self.document.evidence_version.clone()), + cache_profile: Some(snapshot_cache_version(&provider).into()), benchmark_id: None, inputs: Vec::new(), }; @@ -459,6 +527,10 @@ impl ExportPlannerCostModel<'_> { } } +fn snapshot_cache_version<'a>(provider: &'a ExportPhysicalProvider<'_>) -> &'a str { + provider.target.scope.cache_profile.version() +} + impl CostModel for ExportPlannerCostModel<'_> { fn candidate_cost_covers_complete_plan(&self) -> bool { true @@ -1479,6 +1551,7 @@ mod tests { predicates: vec![], info_matchers: vec![], }], + cache_profile: CacheProfile::no_cache(), } } @@ -1663,6 +1736,7 @@ mod tests { annotation.evidence_version.as_deref(), Some("test-evidence-v1") ); + assert_eq!(annotation.cache_profile.as_deref(), Some("no-cache-v1")); assert!(annotation .model_version .as_deref() @@ -1675,6 +1749,121 @@ mod tests { assert!(!selected.inputs.iter().any(|input| input.name == "topk_k")); } + #[test] + fn legacy_cache_json_defaults_to_named_no_cache_but_malformed_profiles_fail() { + // Existing evidence files keep their costs and acquire explicit provenance. + let (query, candidate, document) = cost_fixture(); + let mut json = serde_json::to_value(&document).unwrap(); + json["targets"][0]["scope"] + .as_object_mut() + .unwrap() + .remove("cache_profile"); + let parsed = parse_planner_cost_document(&json.to_string()).unwrap(); + let target = Rc::new(query); + let legacy = ExportPlannerCostModel { document: &parsed }.annotations(&candidate, &target); + let explicit = ExportPlannerCostModel { + document: &document, + } + .annotations(&candidate, &target); + assert_eq!(legacy, explicit); + assert_eq!(legacy.0.cache_profile.as_deref(), Some("no-cache-v1")); + let exported = serde_json::to_value(legacy.0).unwrap(); + assert_eq!(exported["cache_profile"], "no-cache-v1"); + + json["targets"][0]["scope"]["cache_profile"] = serde_json::json!({"profile": "evidence"}); + assert!(parse_planner_cost_document(&json.to_string()).is_err()); + json["targets"][0]["scope"]["cache_profile"] = serde_json::Value::Null; + assert!(parse_planner_cost_document(&json.to_string()).is_err()); + } + + #[test] + fn cache_json_affects_ranking_and_exports_declared_evidence() { + // Identical repeats hit the result cache; distinct evaluations still execute. + let (query, candidate, document) = cost_fixture(); + let target_rc = Rc::new(query); + let target = asap_aware_mapping::replacement::TargetSubDAG::new(&target_rc); + let no_cache = ExportPlannerCostModel { + document: &document, + } + .annotations(&candidate, &target_rc); + let mut json = serde_json::to_value(&document).unwrap(); + json["targets"][0]["scope"]["cache_profile"] = serde_json::json!({ + "profile": "evidence", + "version": "warm-cache-v1", + "distinct_evaluations": 2, + "repeated_identical_evaluations": 8, + "result_cache": {"working_set_bytes": 100, "capacity_bytes": 100}, + "buffer_cache": {"working_set_bytes": 1000, "capacity_bytes": 500}, + "result_invalidation_ratio": null + }); + let parsed = parse_planner_cost_document(&json.to_string()).unwrap(); + let model = ExportPlannerCostModel { document: &parsed }; + let (baseline, selected, _) = model.annotations(&candidate, &target_rc); + let value = |annotation: &CostAnnotation, name: &str| { + annotation + .inputs + .iter() + .find(|input| input.name == name) + .unwrap() + .value + }; + assert_eq!( + value(&baseline, "estimated_cpu_ops"), + value(&no_cache.0, "estimated_cpu_ops") * 0.2 + ); + assert_eq!( + value(&baseline, "estimated_scan"), + value(&no_cache.0, "estimated_scan") * 0.1 + ); + assert_eq!(value(&baseline, "result_cache_hit_ratio"), 1.0); + assert_eq!(value(&baseline, "buffer_cache_hit_ratio"), 0.5); + assert_eq!(value(&baseline, "distinct_evaluations"), 2.0); + assert_eq!(value(&baseline, "repeated_identical_evaluations"), 8.0); + assert_eq!(value(&baseline, "result_cache_working_set"), 100.0); + assert_eq!(value(&baseline, "buffer_cache_capacity"), 500.0); + assert_eq!(baseline.cache_profile.as_deref(), Some("warm-cache-v1")); + assert!(selected.value.unwrap() < no_cache.1.value.unwrap()); + assert_eq!( + model.candidate_cost(&candidate, &target).unwrap().0, + selected.value.unwrap() + ); + let exported = serde_json::to_value(&selected).unwrap(); + assert_eq!(exported["cache_profile"], "warm-cache-v1"); + assert!(exported["inputs"] + .as_array() + .unwrap() + .iter() + .any(|input| { input["name"] == "buffer_cache_capacity" && input["unit"] == "bytes" })); + + json["targets"][0]["scope"]["cache_profile"]["distinct_evaluations"] = 10.into(); + json["targets"][0]["scope"]["cache_profile"]["repeated_identical_evaluations"] = 0.into(); + json["targets"][0]["scope"]["cache_profile"]["buffer_cache"]["capacity_bytes"] = 0.into(); + let distinct = parse_planner_cost_document(&json.to_string()).unwrap(); + let distinct = ExportPlannerCostModel { + document: &distinct, + } + .annotations(&candidate, &target_rc); + assert_eq!(distinct.0.value, no_cache.0.value); + assert_eq!( + value(&distinct.0, "estimated_cpu_ops"), + value(&no_cache.0, "estimated_cpu_ops") + ); + assert_eq!( + value(&distinct.0, "estimated_scan"), + value(&no_cache.0, "estimated_scan") + ); + + json["targets"][0]["scope"]["cache_profile"]["distinct_evaluations"] = 9.into(); + let invalid = parse_planner_cost_document(&json.to_string()).unwrap(); + let invalid = ExportPlannerCostModel { document: &invalid }; + assert!(invalid.candidate_cost(&candidate, &target).is_none()); + assert!(invalid + .annotations(&candidate, &target_rc) + .0 + .value + .is_none()); + } + #[test] fn duplicate_target_candidate_and_query_evidence_each_fail_closed() { let (query, candidate, document) = cost_fixture(); diff --git a/crates/types/src/cost.rs b/crates/types/src/cost.rs index 0c1f8166..4bfe537c 100644 --- a/crates/types/src/cost.rs +++ b/crates/types/src/cost.rs @@ -122,6 +122,9 @@ pub struct CostAnnotation { /// visible even when the analytical formulas are unchanged. #[serde(skip_serializing_if = "Option::is_none")] pub evidence_version: Option, + /// Named, versioned cache assumption used by the physical cost model. + #[serde(skip_serializing_if = "Option::is_none")] + pub cache_profile: Option, #[serde(skip_serializing_if = "Option::is_none")] pub benchmark_id: Option, #[serde(default, skip_serializing_if = "Vec::is_empty")] @@ -143,6 +146,7 @@ impl CostAnnotation { benefit_ratio: None, model_version: None, evidence_version: None, + cache_profile: None, benchmark_id: None, inputs: Vec::new(), } @@ -174,6 +178,7 @@ impl CostAnnotation { benefit_ratio: None, model_version: Some(model_version), evidence_version: None, + cache_profile: None, benchmark_id: None, inputs, } @@ -211,6 +216,18 @@ impl CostAnnotation { self.evidence_version = Some(evidence_version); self } + + pub fn with_cache_profile(mut self, cache_profile: impl Into) -> Self { + if self.source != CostSource::Modeled || self.value.is_none() { + return self; + } + let cache_profile = cache_profile.into(); + if cache_profile.trim().is_empty() { + return Self::unavailable(self.unit); + } + self.cache_profile = Some(cache_profile); + self + } } /// `delta / baseline_value`, or `None` when `baseline_value <= 0` — the @@ -309,6 +326,7 @@ where let mut missing_any = false; let mut model_versions: Vec = Vec::new(); let mut evidence_versions: Vec = Vec::new(); + let mut cache_profiles: Vec> = Vec::new(); for (workload_node_id, annotation) in entries { if let Some(id) = workload_node_id { @@ -346,12 +364,15 @@ where evidence_versions.push(version.clone()); } } + if !cache_profiles.contains(&annotation.cache_profile) { + cache_profiles.push(annotation.cache_profile.clone()); + } } // One workload total must not silently combine different immutable // catalog/runtime generations. Such a subtotal is not a comparable // snapshot even though each component is individually numeric. - if evidence_versions.len() > 1 { + if evidence_versions.len() > 1 || cache_profiles.len() > 1 { missing_any = true; } @@ -373,6 +394,11 @@ where } else { None }, + cache_profile: if cache_profiles.len() == 1 { + cache_profiles.remove(0) + } else { + None + }, benchmark_id: None, inputs: Vec::new(), } @@ -418,6 +444,7 @@ where (Some(baseline_value), Some(selected_value)) if baseline_cost.unit == selected_cost.unit && baseline_cost.evidence_version == selected_cost.evidence_version + && baseline_cost.cache_profile == selected_cost.cache_profile && !model_version.trim().is_empty() => { let delta = baseline_value - selected_value; @@ -430,6 +457,7 @@ where benefit_ratio: benefit_ratio(baseline_value, delta), model_version: Some(model_version), evidence_version: selected_cost.evidence_version.clone(), + cache_profile: selected_cost.cache_profile.clone(), benchmark_id: None, inputs: Vec::new(), } @@ -500,6 +528,46 @@ mod tests { assert_eq!(unavailable.value, None); } + #[test] + fn modeled_annotations_export_cache_profile_provenance() { + let annotation = ann(3.0, CostUnit::CostUnits).with_cache_profile("no-cache-v1"); + assert_eq!(annotation.cache_profile.as_deref(), Some("no-cache-v1")); + assert_eq!( + serde_json::to_value(annotation).unwrap()["cache_profile"], + "no-cache-v1" + ); + + let unavailable = ann(3.0, CostUnit::CostUnits).with_cache_profile(" \t"); + assert_eq!(unavailable.source, CostSource::Unavailable); + } + + // Totals and benefits must not combine known and incompatible/unknown cache assumptions. + #[test] + fn workload_cache_provenance_must_be_complete_and_equal() { + let raw = ann(10.0, CostUnit::CostUnits).with_cache_profile("no-cache-v1"); + let warm = ann(1.0, CostUnit::CostUnits).with_cache_profile("warm-v1"); + let unknown = ann(2.0, CostUnit::CostUnits); + assert_eq!( + sum_workload_costs([(None, &raw), (None, &unknown)]) + .unwrap() + .value, + None + ); + assert_eq!( + workload_cost_summary([(None, &raw, &warm)], "v1") + .unwrap() + .benefit + .value, + None + ); + assert_eq!( + sum_workload_costs([(None, &raw), (None, &raw)]) + .unwrap() + .cache_profile, + raw.cache_profile + ); + } + #[test] fn benefit_ratio_unavailable_when_baseline_not_positive() { assert_eq!(benefit_ratio(0.0, 5.0), None); diff --git a/crates/types/src/resources.rs b/crates/types/src/resources.rs index 894e3465..d1bde516 100644 --- a/crates/types/src/resources.rs +++ b/crates/types/src/resources.rs @@ -3,72 +3,18 @@ //! 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. +//! Cache assumptions share this schema namespace but are not additive resource +//! consumption; their numerical interpretation belongs to an estimator. -use serde::{Deserialize, Serialize}; +pub mod cache; +pub mod cpu; +pub mod measurement; +pub mod physical; +pub use cache::{CacheCapacityEvidence, CacheEvidence, CacheProfile}; -/// 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, - } - } -} +pub use cpu::{MeasuredCpu, ModeledCpu}; +pub use measurement::Measurement; +pub use physical::{MeasuredResources, PhysicalResources}; #[cfg(test)] mod tests { diff --git a/crates/types/src/resources/cache.rs b/crates/types/src/resources/cache.rs new file mode 100644 index 00000000..f3e47630 --- /dev/null +++ b/crates/types/src/resources/cache.rs @@ -0,0 +1,115 @@ +//! Shared cache assumptions. Numerical estimation belongs to the cost model. + +use serde::{Deserialize, Serialize}; + +/// Versioned deployment evidence describing query-result and buffer caching. +/// These are deployment assumptions, not additive CPU or byte consumption. +/// Estimators validate and interpret them against a comparison workload. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(tag = "profile", rename_all = "snake_case", deny_unknown_fields)] +pub enum CacheProfile { + NoCache { version: String }, + Evidence(CacheEvidence), +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct CacheEvidence { + pub version: String, + /// Evaluations whose parameterization is not identical to a preceding + /// evaluation and therefore cannot use the result cache. + pub distinct_evaluations: u64, + /// Evaluations identical to a preceding evaluation and eligible for a + /// result-cache hit. + pub repeated_identical_evaluations: u64, + pub result_cache: CacheCapacityEvidence, + pub buffer_cache: CacheCapacityEvidence, + /// Fraction of otherwise-resident result entries invalidated by arriving + /// data. Required for continuously ingesting data; `AtRest` permits only + /// zero or omitted invalidation evidence. + pub result_invalidation_ratio: Option, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct CacheCapacityEvidence { + pub working_set_bytes: u64, + pub capacity_bytes: u64, +} + +impl CacheProfile { + pub fn no_cache() -> Self { + Self::NoCache { + version: "no-cache-v1".into(), + } + } + + pub fn version(&self) -> &str { + match self { + Self::NoCache { version } => version, + Self::Evidence(evidence) => &evidence.version, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + // Moving the definitions preserves the established tagged JSON contract. + #[test] + fn existing_cache_json_round_trips_through_shared_types() { + let no_cache = json!({"profile": "no_cache", "version": "no-cache-v1"}); + assert_eq!( + serde_json::to_value(CacheProfile::no_cache()).unwrap(), + no_cache + ); + assert_eq!( + serde_json::from_value::(no_cache) + .unwrap() + .version(), + "no-cache-v1" + ); + let evidence = json!({ + "profile": "evidence", "version": "cache-evidence-7", + "distinct_evaluations": 2, "repeated_identical_evaluations": 4, + "result_cache": {"working_set_bytes": 100, "capacity_bytes": 100}, + "buffer_cache": {"working_set_bytes": 9007199254740993_u64, "capacity_bytes": 50}, + "result_invalidation_ratio": null + }); + let decoded: CacheProfile = serde_json::from_value(evidence.clone()).unwrap(); + assert_eq!(decoded.version(), "cache-evidence-7"); + assert_eq!(serde_json::to_value(decoded).unwrap(), evidence); + } + + // Unknown invalidation stays unknown; required capacity cannot default to zero. + #[test] + fn shared_cache_json_preserves_unknowns_and_rejects_missing_or_extra_dimensions() { + let mut evidence = json!({ + "profile": "evidence", "version": "v1", + "distinct_evaluations": 1, "repeated_identical_evaluations": 0, + "result_cache": {"working_set_bytes": 100, "capacity_bytes": 0}, + "buffer_cache": {"working_set_bytes": 100, "capacity_bytes": 0} + }); + let CacheProfile::Evidence(decoded) = + serde_json::from_value::(evidence.clone()).unwrap() + else { + panic!("wrong variant") + }; + assert_eq!(decoded.result_invalidation_ratio, None); + evidence["result_cache"] + .as_object_mut() + .unwrap() + .remove("capacity_bytes"); + assert!(serde_json::from_value::(evidence).is_err()); + assert!(serde_json::from_value::(json!({ + "working_set_bytes": 100, "capacity_bytes": 10, "cpu_ops": 1 + })) + .is_err()); + assert!(serde_json::from_value::(json!({ + "profile": "no_cache", "version": "v1", "scan_bytes": 0 + })) + .is_err()); + } +} diff --git a/crates/types/src/resources/cpu.rs b/crates/types/src/resources/cpu.rs new file mode 100644 index 00000000..960f5cd2 --- /dev/null +++ b/crates/types/src/resources/cpu.rs @@ -0,0 +1,26 @@ +//! CPU quantities with explicit modeled-work or measured-time units. + +use serde::{Deserialize, Serialize}; + +use super::measurement::Measurement; + +/// 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, +} + +/// 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, +} diff --git a/crates/types/src/resources/measurement.rs b/crates/types/src/resources/measurement.rs new file mode 100644 index 00000000..96b51b0f --- /dev/null +++ b/crates/types/src/resources/measurement.rs @@ -0,0 +1,15 @@ +//! Measurement values and uncertainty metadata shared across resource dimensions. + +use serde::{Deserialize, Serialize}; + +/// 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, +} diff --git a/crates/types/src/resources/physical.rs b/crates/types/src/resources/physical.rs new file mode 100644 index 00000000..a7f9cf85 --- /dev/null +++ b/crates/types/src/resources/physical.rs @@ -0,0 +1,36 @@ +//! Resource aggregation with independent byte dimensions and explicit CPU units. + +use serde::{Deserialize, Serialize}; + +use super::{cpu::MeasuredCpu, measurement::Measurement}; + +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, + } + } +} diff --git a/docs/design_docs/asap-aware-mapping/analytical-resource-cost.md b/docs/design_docs/asap-aware-mapping/analytical-resource-cost.md index cb576cc4..e7954cbc 100644 --- a/docs/design_docs/asap-aware-mapping/analytical-resource-cost.md +++ b/docs/design_docs/asap-aware-mapping/analytical-resource-cost.md @@ -109,7 +109,7 @@ Execution multiplicity has exactly two values: | Value | Executions in one comparison horizon | Typical use | |---|---:|---| | `Once` | `1` | Bootstrap/build work whose result is retained for later reads. | -| `PerEvaluation` | `evaluation_count` | Raw recomputation or query-side work repeated for every invocation. | +| `PerEvaluation` | cache-adjusted demand | Raw recomputation or query-side work repeated for cache misses. | `Once` means once **within the comparison horizon**, not once for the lifetime of a process or deployment. `PerEvaluation` means once per query invocation, @@ -117,14 +117,63 @@ not once per input row, grouping key, DAG edge, or consumer. Per-row work is already represented by the operator's local formula; active-window fan-out and the number of physical summary instances are separate inputs. +Every evidence snapshot carries a named, versioned `CacheProfile`. The +`no-cache-v1` profile preserves the original formulas exactly. A modeled +profile splits demand into distinct parameterizations and repeated-identical +evaluations, and supplies working-set and capacity evidence independently for +the result cache and buffer/page cache. Working sets must be nonzero; zero +capacity explicitly disables that cache. Missing, non-finite, or inconsistent +evidence makes the estimate unavailable. At least one distinct evaluation is +required: the result cache is filled by evaluations in this horizon. Buffer +residency is a steady-state capacity/working-set model; it does not model +cold-start warming or access order. + +`asap_types::resources` is the single definition site for `CacheProfile`, +`CacheEvidence`, and `CacheCapacityEvidence` (implemented in `resources/cache.rs`). +These schemas describe cache assumptions, not additive CPU or byte consumption. +The mapping module re-exports the same types for existing import paths, while +`no_cache()` and `version()` remain shared metadata methods. Workload validation +and numeric estimation stay in mapping: callers use +`analytical_cost::cache_hit_ratios(&profile, count, arrival)` instead of the former +inherent `profile.hit_ratios(...)` method. The exporter consumes the central type +directly. Tagged JSON, required capacities, and optional invalidation are unchanged. + +Uncovered buffer bytes are subtracted before converting to floating point, so +near-full residency cannot erase a nonzero miss. With an integral number of +executions, buffer-adjusted scan bytes use exact integer arithmetic and round +up only after scaling; large byte counts do not lose precision above `2^53`. +The physical-ranking entry point also rejects blank calibration versions, +keeping every admitted comparison bound to a named coefficient generation. + +Define `R = min(1, result_capacity / result_working_set)` and +`B = min(1, buffer_capacity / buffer_working_set)`. For at-rest data the +result invalidation ratio is zero. Continuously ingesting data must declare an +invalidation ratio `I` in `[0, 1]`; mixed or unknown arrival fails closed. + +```text +result_hit_ratio = R * (1 - I) +cpu_executions = distinct + repeated_identical * (1 - result_hit_ratio) +scan_executions = cpu_executions * (1 - B) +``` + +Thus a result-cache hit elides both CPU and scans, while a buffer-cache hit +elides only storage bytes, including scans used to build a summary. `Once` +nodes remain one CPU execution: retained summary +state is already in memory and is not charged a per-evaluation storage read. +Raw and candidate plans must use the same cache profile. Its version is +exported in `CostAnnotation.cache_profile` alongside formula and evidence +versions. + For physical node `n`, define: ```text -executions(n) = 1 if n.execution = Once - = evaluation_count if n.execution = PerEvaluation +cpu_executions(n) = 1 if n.execution = Once + = cache-adjusted CPU demand otherwise +scan_executions(n) = 1 - B if n.execution = Once + = cache-adjusted scan demand otherwise -total_cpu_ops = sum(local_cpu_ops(n) * executions(n)) -total_scan = sum(local_scan_bytes(n) * executions(n)) +total_cpu_ops = sum(local_cpu_ops(n) * cpu_executions(n)) +total_scan = sum(local_scan_bytes(n) * scan_executions(n)) ``` Memory is not multiplied by `executions(n)`. `peak_memory_bytes` is the maximum @@ -1128,7 +1177,6 @@ complete deployment model may additionally require: - source and spill writes; - network transfer; -- cache residency; - parallelism and contention; - allocator fragmentation; - wall-clock critical-path latency; @@ -1137,3 +1185,12 @@ complete deployment model may additionally require: Those dimensions should extend the resource vector and calibration. They must not be silently represented as zero. Consumers render unavailable costs as `Not estimated`. + +## Benchmark validation + +Experiments used to calibrate or validate these formulas must either disable +result and buffer caches, or vary query parameters so the workload represents +distinct queries that a retained summary can serve but a result cache cannot. +Repeatedly timing one identical query measures cache behavior rather than the +relative physical-plan cost. Benchmark reports must record which cache profile +was used. diff --git a/tools/dag-viewer/test_render.py b/tools/dag-viewer/test_render.py index bc78fcfb..d5b28f9f 100644 --- a/tools/dag-viewer/test_render.py +++ b/tools/dag-viewer/test_render.py @@ -1,9 +1,5 @@ -"""Unit tests for render.py's data-merging and HTML-assembly logic -- -deliberately exercised without py_mini_racer or a browser (see render.py's -own ad hoc py_mini_racer validation, not committed here, and PR #249's -description for how the shared viewer.js/node-style.js logic itself was -checked). These tests only cover what render.py adds on top of index.html: -merging input files and correctly inlining/embedding into one HTML page. +"""Tests for render.py's data merging and HTML assembly, plus viewer cache +provenance checks executed with optional py_mini_racer (no browser required). Run with (from the repo root): python3 -m unittest discover -s tools/dag-viewer -p 'test_render.py' (or `cd tools/dag-viewer && python3 -m unittest test_render`, or @@ -32,6 +28,11 @@ HERE = Path(__file__).resolve().parent +try: + from py_mini_racer import py_mini_racer +except ImportError: + py_mini_racer = None + def named_graph(name: str, source: str = "SELECT 1") -> dict: """A minimal well-formed NamedGraph: one leaf Scan node, its own root.""" @@ -315,5 +316,69 @@ def test_malformed_json_gives_a_clean_error_not_a_traceback(self): self.assertNotIn("Traceback", result.stderr) +@unittest.skipIf(py_mini_racer is None, "viewer tests require py_mini_racer") +class ViewerCacheTests(unittest.TestCase): + def setUp(self): + self.js = py_mini_racer.MiniRacer() + source = (HERE / "viewer.js").read_text() + for name in ["escapeHtml", "formatCostUnit", "formatBaselineRef", + "formatCostNumber", "renderCostAnnotation", + "computeSelectionWorkloadCost"]: + function = re.search(r"^function " + name + r"\(.*?^}", source, re.M | re.S) + self.assertIsNotNone(function, name) + self.js.eval(function.group(0)) + + @staticmethod + def query(profile, selected_profile=None, batch=0): + def annotation(value, cache): + result = {"value": value, "unit": "CostUnits", "source": "Modeled"} + if cache is not None: + result["cache_profile"] = cache + return result + + return {"sourceBatch": batch, "post_graph": {"nodes": [{"decision": { + "id": 1, + "baseline_cost": annotation(10, profile), + "selected_cost": annotation(4, selected_profile or profile), + }}]}} + + def test_cache_profile_and_inputs_are_rendered(self): + """The sidebar exposes the assumptions behind a cache-adjusted cost.""" + annotation = self.query("warm-cache-v1")["post_graph"]["nodes"][0]["decision"]["baseline_cost"] + annotation["inputs"] = [{"name": "result_cache_hit_ratio", "value": 0.5, "unit": "ratio"}] + html = self.js.call("renderCostAnnotation", "Baseline", annotation) + self.assertIn("cache warm-cache-v1", html) + self.assertIn("result_cache_hit_ratio", html) + self.assertIn("0.5 ratio", html) + + def test_same_cache_profile_aggregates_and_preserves_provenance(self): + """Comparable decisions total normally and retain their common profile.""" + result = self.js.call("computeSelectionWorkloadCost", [self.query("warm-v1"), self.query("warm-v1", batch=1)]) + self.assertEqual(result["baseline_cost"]["value"], 20) + self.assertEqual(result["benefit"]["value"], 12) + self.assertEqual(result["benefit"]["cache_profile"], "warm-v1") + + def test_legacy_profiles_preserve_existing_totals(self): + """Models without cache provenance retain their existing aggregation.""" + result = self.js.call("computeSelectionWorkloadCost", [self.query(None), self.query(None, batch=1)]) + self.assertEqual(result["benefit"]["value"], 12) + self.assertIsNone(result["benefit"]["cache_profile"]) + + def test_different_or_mixed_cache_profiles_make_totals_unavailable(self): + """Selection totals cannot claim benefits across incompatible assumptions.""" + cases = [ + [self.query("cold-v1"), self.query("warm-v1", batch=1)], + [self.query("cold-v1", selected_profile="warm-v1")], + [self.query("warm-v1"), self.query(None, batch=1)], + [self.query(None), self.query("warm-v1", batch=1)], + ] + for queries in cases: + with self.subTest(queries=queries): + result = self.js.call("computeSelectionWorkloadCost", queries) + for annotation in result.values(): + self.assertIsNone(annotation["value"]) + self.assertEqual(annotation["source"], "Unavailable") + + if __name__ == "__main__": unittest.main() diff --git a/tools/dag-viewer/viewer.js b/tools/dag-viewer/viewer.js index e371c34f..dfe2e329 100644 --- a/tools/dag-viewer/viewer.js +++ b/tools/dag-viewer/viewer.js @@ -723,6 +723,7 @@ function renderCostAnnotation(title, annotation) { const provenanceParts = []; if (annotation.model_version) provenanceParts.push(`model ${annotation.model_version}`); if (annotation.evidence_version) provenanceParts.push(`evidence ${annotation.evidence_version}`); + if (annotation.cache_profile) provenanceParts.push(`cache ${annotation.cache_profile}`); if (annotation.benchmark_id) provenanceParts.push(`benchmark ${annotation.benchmark_id}`); const inputsHtml = (annotation.inputs || []).length ? `
    ${annotation.inputs.map((input) => `
  • ${escapeHtml(input.name)}${escapeHtml(String(input.value))}${input.unit ? ' ' + escapeHtml(input.unit) : ''}
    • `).join('')}
` @@ -790,6 +791,8 @@ function computeSelectionWorkloadCost(selected) { let selectedSum = 0; let any = false; let unavailable = false; + let cacheProfile = null; + let hasCacheProfile = false; for (const query of selected) { const nodes = (query.post_graph && query.post_graph.nodes) || []; for (const node of nodes) { @@ -804,6 +807,17 @@ function computeSelectionWorkloadCost(selected) { if (unit === null) unit = baseline.unit; if (baseline.unit !== unit || selectedCost.unit !== unit) return null; // unit-incompatible aggregation is rejected any = true; + // Preserve legacy models without cache provenance, but never mix them + // with cache-aware estimates (matching Rust workload aggregation). + const profile = baseline.cache_profile ?? null; + if ((profile !== null && (typeof profile !== 'string' || !profile.trim())) + || (selectedCost.cache_profile ?? null) !== profile + || (hasCacheProfile && cacheProfile !== profile)) { + unavailable = true; + } else { + cacheProfile = profile; + hasCacheProfile = true; + } if (baseline.value === null || baseline.value === undefined || selectedCost.value === null || selectedCost.value === undefined) { unavailable = true; continue; @@ -819,12 +833,13 @@ function computeSelectionWorkloadCost(selected) { } const delta = baselineSum - selectedSum; return { - baseline_cost: { value: baselineSum, unit, source: 'Modeled' }, - selected_cost: { value: selectedSum, unit, source: 'Modeled' }, + baseline_cost: { value: baselineSum, unit, source: 'Modeled', cache_profile: cacheProfile }, + selected_cost: { value: selectedSum, unit, source: 'Modeled', cache_profile: cacheProfile }, benefit: { value: delta, unit, source: 'Modeled', + cache_profile: cacheProfile, baseline: { kind: 'PreAsapRecomputation' }, benefit_ratio: baselineSum > 0 ? delta / baselineSum : null, },