From f377ae2479c865012a09b4f5cb504363da752c1a Mon Sep 17 00:00:00 2001 From: zz_y Date: Mon, 7 Sep 2026 10:11:25 -0600 Subject: [PATCH 1/5] feat(cost): estimate physical storage operation counts --- crates/asap-aware-mapping/src/lib.rs | 1 + .../src/physical_plan_cost_model.rs | 41 ++- crates/asap-aware-mapping/src/storage_io.rs | 292 ++++++++++++++++++ crates/asap-aware-mapping/tests/storage_io.rs | 245 +++++++++++++++ crates/devtools/src/bin/dag_export.rs | 186 ++++++++++- .../analytical-resource-cost.md | 5 + .../developer_docs/storage-operation-costs.md | 66 ++++ 7 files changed, 830 insertions(+), 6 deletions(-) create mode 100644 crates/asap-aware-mapping/src/storage_io.rs create mode 100644 crates/asap-aware-mapping/tests/storage_io.rs create mode 100644 docs/developer_docs/storage-operation-costs.md diff --git a/crates/asap-aware-mapping/src/lib.rs b/crates/asap-aware-mapping/src/lib.rs index 27021625..e682fe66 100644 --- a/crates/asap-aware-mapping/src/lib.rs +++ b/crates/asap-aware-mapping/src/lib.rs @@ -197,6 +197,7 @@ pub mod recurrence; pub mod replacement; pub mod rewrite; pub mod rollup; +pub mod storage_io; pub mod summary_maintenance_cost; pub mod summary_maintenance_dag_export; pub mod summary_maintenance_lifecycle; 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 6d2d8245..051a4ba7 100644 --- a/crates/asap-aware-mapping/src/physical_plan_cost_model.rs +++ b/crates/asap-aware-mapping/src/physical_plan_cost_model.rs @@ -29,6 +29,7 @@ pub struct PhysicalEvidenceSnapshot { pub version: String, pub scope: ComparisonScope, pub cache_profile: CacheProfile, + pub storage_io: Option, } /// Deployment evidence needed to price one planner alternative. @@ -61,11 +62,15 @@ pub trait PlannerPhysicalPlanProvider { } /// Dimensional comparison retained for explanations and verification. -#[derive(Debug, Clone, Copy, PartialEq)] +#[derive(Debug, Clone, PartialEq)] pub struct PhysicalPlanComparison { pub resources: PhysicalDagComparisonEstimate, pub raw_cost: Cost, pub candidate_cost: Cost, + pub storage_io: Option<( + crate::storage_io::StorageEstimate, + crate::storage_io::StorageEstimate, + )>, } /// Planner cost model that admits only complete, cheaper physical plans. @@ -176,12 +181,40 @@ impl<'a> PhysicalPlanCostModel<'a> { cache_profile: &snapshot.cache_profile, }, )?; - let raw_cost = Cost(resources.raw.calibrated_cost(&self.calibration)?); - let candidate_cost = Cost(resources.candidate.calibrated_cost(&self.calibration)?); + let storage_io = snapshot + .storage_io + .as_ref() + .map(|profile| { + Ok(( + crate::storage_io::estimate_storage_io( + &raw, + scope, + profile, + &snapshot.version, + )?, + crate::storage_io::estimate_storage_io( + &replacement, + scope, + profile, + &snapshot.version, + )?, + )) + }) + .transpose()?; + let mut raw_cost = Cost(resources.raw.calibrated_cost(&self.calibration)?); + let mut candidate_cost = Cost(resources.candidate.calibrated_cost(&self.calibration)?); + if let Some((raw, candidate)) = &storage_io { + raw_cost.0 += raw.cost; + candidate_cost.0 += candidate.cost; + } + if !raw_cost.0.is_finite() || !candidate_cost.0.is_finite() { + return Err(AnalyticalCostError::Overflow); + } Ok(PhysicalPlanComparison { resources, raw_cost, candidate_cost, + storage_io, }) } } @@ -423,6 +456,7 @@ mod tests { version: "test-snapshot-1".into(), scope: scope(), cache_profile: CacheProfile::no_cache(), + storage_io: None, }) } @@ -661,6 +695,7 @@ mod tests { version: " \t".into(), scope: scope(), cache_profile: CacheProfile::no_cache(), + storage_io: None, }) } diff --git a/crates/asap-aware-mapping/src/storage_io.rs b/crates/asap-aware-mapping/src/storage_io.rs new file mode 100644 index 00000000..1bfeaad8 --- /dev/null +++ b/crates/asap-aware-mapping/src/storage_io.rs @@ -0,0 +1,292 @@ +//! Request counts for explicitly bound physical storage accesses. + +use std::collections::{HashMap, HashSet}; + +use serde::{Deserialize, Serialize}; + +use crate::analytical_cost::{ + estimate_physical_dag, AnalyticalCostError, EvidenceBackedPhysicalDag, ExecutionMultiplicity, + PhysicalDagNode, PhysicalOperator, +}; +use crate::physical_operator_statistics::{ComparisonScope, OperatorStatistics}; + +pub const STORAGE_IO_MODEL_VERSION: &str = "storage-requests-v1"; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum StorageOperation { + DiskRead, + DiskWrite, + ObjectGet, + ObjectPut, +} + +/// Each extent is one independent contiguous disk range, object, or multipart +/// payload. Requests cannot coalesce across extent boundaries. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct StorageAccess { + pub operation: StorageOperation, + pub extent_bytes: Vec, + pub bytes_per_request: u64, +} + +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct StorageResources { + pub disk_reads: u64, + pub disk_writes: u64, + pub object_gets: u64, + pub object_puts: u64, +} + +impl StorageResources { + pub fn terms(self) -> [(&'static str, u64); 4] { + [ + ("disk_read_operations", self.disk_reads), + ("disk_write_operations", self.disk_writes), + ("object_get_operations", self.object_gets), + ("object_put_operations", self.object_puts), + ] + } + fn add(&mut self, other: Self) -> Result<(), AnalyticalCostError> { + self.disk_reads = self + .disk_reads + .checked_add(other.disk_reads) + .ok_or(AnalyticalCostError::Overflow)?; + self.disk_writes = self + .disk_writes + .checked_add(other.disk_writes) + .ok_or(AnalyticalCostError::Overflow)?; + self.object_gets = self + .object_gets + .checked_add(other.object_gets) + .ok_or(AnalyticalCostError::Overflow)?; + self.object_puts = self + .object_puts + .checked_add(other.object_puts) + .ok_or(AnalyticalCostError::Overflow)?; + Ok(()) + } +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct StorageCalibration { + pub version: String, + pub cost_per_disk_read: f64, + pub cost_per_disk_write: f64, + pub cost_per_object_get: f64, + pub cost_per_object_put: f64, +} + +impl StorageCalibration { + pub fn cost(&self, value: StorageResources) -> Result { + if self.version.trim().is_empty() { + return Err(invalid("blank storage calibration version")); + } + let mut cost = 0.0; + for (coefficient, count) in [ + (self.cost_per_disk_read, value.disk_reads), + (self.cost_per_disk_write, value.disk_writes), + (self.cost_per_object_get, value.object_gets), + (self.cost_per_object_put, value.object_puts), + ] { + if !coefficient.is_finite() || coefficient < 0.0 { + return Err(invalid("invalid storage calibration coefficient")); + } + cost += coefficient * count as f64; + } + if !cost.is_finite() { + return Err(AnalyticalCostError::Overflow); + } + Ok(cost) + } +} + +/// Atomic deployment snapshot. Every reachable physical node needs an entry, +/// including an explicit empty access list for memory-only operators. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct StorageIoProfile { + pub evidence_version: String, + pub observed_at_ms: u64, + pub valid_until_ms: u64, + pub calibration: StorageCalibration, + pub nodes: HashMap, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct StorageNodeEvidence { + pub node: PhysicalDagNode, + pub statistics: OperatorStatistics, + pub accesses: Vec, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct StorageEstimate { + pub total: StorageResources, + pub per_node: HashMap, + pub model_version: String, + pub evidence_version: String, + pub calibration_version: String, + pub cost: f64, +} + +fn invalid(reason: &'static str) -> AnalyticalCostError { + AnalyticalCostError::InvalidPhysicalDag(reason) +} + +/// ceil(bytes/request_size), rounded independently per extent and execution. +/// No metadata, retry, or speculative prefetch operations are inferred. +pub fn request_count(access: &StorageAccess) -> Result { + if access.bytes_per_request == 0 { + return Err(invalid("storage request size must be positive")); + } + access.extent_bytes.iter().try_fold(0_u64, |sum, bytes| { + let count = + bytes / access.bytes_per_request + u64::from(bytes % access.bytes_per_request != 0); + sum.checked_add(count).ok_or(AnalyticalCostError::Overflow) + }) +} + +pub fn estimate_storage_io( + dag: &EvidenceBackedPhysicalDag, + scope: &ComparisonScope, + profile: &StorageIoProfile, + evidence_version: &str, +) -> Result { + // Also prove source coverage, edge consistency, execution legality and DAG + // identity before using supplementary deployment evidence. + estimate_physical_dag(&dag.nodes, &dag.root, scope, dag)?; + let evaluations = scope.validate()?; + if profile.evidence_version.trim().is_empty() + || profile.evidence_version != evidence_version + || profile.observed_at_ms > scope.planning_time.0 + || profile.valid_until_ms <= scope.planning_time.0 + { + return Err(AnalyticalCostError::MissingOrStale("storage I/O profile")); + } + let by_id: HashMap<_, _> = dag + .nodes + .iter() + .map(|node| (node.id.as_str(), node)) + .collect(); + let mut seen = HashSet::new(); + let mut pending = vec![dag.root.as_str()]; + let mut total = StorageResources::default(); + let mut per_node = HashMap::new(); + while let Some(id) = pending.pop() { + if !seen.insert(id) { + continue; + } + let node = by_id[id]; + pending.extend(node.children.iter().map(String::as_str)); + let evidence = profile.nodes.get(id).ok_or_else(|| { + AnalyticalCostError::MissingOperatorStatistics(format!("storage:{id}")) + })?; + if evidence.node != *node || evidence.statistics != dag.evidence[id].statistics { + return Err(invalid( + "storage evidence differs from physical node snapshot", + )); + } + let executions = match node.execution { + ExecutionMultiplicity::Once => 1, + ExecutionMultiplicity::PerEvaluation => evaluations, + }; + let mut local = StorageResources::default(); + let mut read_bytes = 0_u64; + for access in &evidence.accesses { + let count = request_count(access)? + .checked_mul(executions) + .ok_or(AnalyticalCostError::Overflow)?; + let mut term = StorageResources::default(); + match access.operation { + StorageOperation::DiskRead => term.disk_reads = count, + StorageOperation::DiskWrite => term.disk_writes = count, + StorageOperation::ObjectGet => term.object_gets = count, + StorageOperation::ObjectPut => term.object_puts = count, + } + if matches!( + access.operation, + StorageOperation::DiskRead | StorageOperation::ObjectGet + ) { + for bytes in &access.extent_bytes { + read_bytes = read_bytes + .checked_add(*bytes) + .ok_or(AnalyticalCostError::Overflow)?; + } + } + local.add(term)?; + } + if node.operator == PhysicalOperator::Scan { + let OperatorStatistics::Scan { + source_read_bytes, .. + } = &evidence.statistics + else { + unreachable!() + }; + if read_bytes != *source_read_bytes { + return Err(invalid( + "storage scan access bytes differ from source read bytes", + )); + } + } + total.add(local)?; + per_node.insert(id.into(), local); + } + Ok(StorageEstimate { + total, + per_node, + model_version: STORAGE_IO_MODEL_VERSION.into(), + evidence_version: profile.evidence_version.clone(), + calibration_version: profile.calibration.version.clone(), + cost: profile.calibration.cost(total)?, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + // Separate objects need separate requests even when their total fits. + #[test] + fn counts_extents_independently_and_is_monotone() { + let mut access = StorageAccess { + operation: StorageOperation::ObjectGet, + extent_bytes: vec![5, 5], + bytes_per_request: 8, + }; + assert_eq!(request_count(&access).unwrap(), 2); + access.bytes_per_request = 4; + assert_eq!(request_count(&access).unwrap(), 4); + access.extent_bytes.push(9); + assert_eq!(request_count(&access).unwrap(), 7); + access.bytes_per_request = 0; + assert!(request_count(&access).is_err()); + } + // Full-width integer evidence must neither wrap nor lose precision. + #[test] + fn checked_rounding_handles_maximum_bytes_and_overflow() { + let mut access = StorageAccess { + operation: StorageOperation::DiskRead, + extent_bytes: vec![u64::MAX], + bytes_per_request: 2, + }; + assert_eq!(request_count(&access).unwrap(), 1_u64 << 63); + access.bytes_per_request = 1; + access.extent_bytes.push(1); + assert_eq!(request_count(&access), Err(AnalyticalCostError::Overflow)); + } + // Invalid deployment coefficients cannot produce a usable cost. + #[test] + fn rejects_non_finite_calibration() { + let calibration = StorageCalibration { + version: "v1".into(), + cost_per_disk_read: f64::NAN, + cost_per_disk_write: 0.0, + cost_per_object_get: 0.0, + cost_per_object_put: 0.0, + }; + assert!(calibration.cost(StorageResources::default()).is_err()); + } +} diff --git a/crates/asap-aware-mapping/tests/storage_io.rs b/crates/asap-aware-mapping/tests/storage_io.rs new file mode 100644 index 00000000..0d6e4a77 --- /dev/null +++ b/crates/asap-aware-mapping/tests/storage_io.rs @@ -0,0 +1,245 @@ +use asap_aware_mapping::analytical_cost::{ + EvidenceBackedPhysicalDag, ExecutionMultiplicity, PhysicalDagNode, PhysicalNodeEvidence, + PhysicalOperator, +}; +use asap_aware_mapping::physical_operator_statistics::{ + ComparisonScope, EdgeStatistics, OperatorStatistics, SourceCoverage, UnaryEdgeStatistics, +}; +use asap_types::pre_asap::query_expr::Source; +use asap_types::workload::{ + DataArrival, DurationMs, QueryRecurrence, QueryTimeScope, TimeSelection, TimestampMs, +}; +use std::collections::HashMap; + +fn fixture() -> (EvidenceBackedPhysicalDag, ComparisonScope) { + let coverage = SourceCoverage { + source: Source::Table { + table_ref: "events".into(), + }, + source_snapshot_id: "snapshot".into(), + predicates: vec![], + info_matchers: vec![], + }; + let scope = ComparisonScope { + data_arrival: DataArrival::AtRest, + planning_time: TimestampMs(100), + horizon: DurationMs(1000), + recurrence: QueryRecurrence::OneTime { + invocations: 3, + execute_at: None, + }, + time_selection: TimeSelection { + scope: QueryTimeScope::Longitudinal, + lookback: Some(DurationMs(1000)), + as_of: Some(TimestampMs(100)), + }, + sources: vec![coverage.clone()], + }; + let edge = EdgeStatistics { + rows: 10, + bytes: 80, + }; + let unary = UnaryEdgeStatistics { + input: edge, + output: edge, + promql: None, + }; + let nodes = vec![ + PhysicalDagNode { + id: "scan".into(), + operator: PhysicalOperator::Scan, + children: vec![], + source_coverage: Some(coverage), + output_buffer_bytes: 0, + retained_bytes: 0, + execution: ExecutionMultiplicity::PerEvaluation, + }, + PhysicalDagNode { + id: "left".into(), + operator: PhysicalOperator::PassThrough, + children: vec!["scan".into()], + source_coverage: None, + output_buffer_bytes: 0, + retained_bytes: 0, + execution: ExecutionMultiplicity::PerEvaluation, + }, + PhysicalDagNode { + id: "right".into(), + operator: PhysicalOperator::PassThrough, + children: vec!["scan".into()], + source_coverage: None, + output_buffer_bytes: 0, + retained_bytes: 0, + execution: ExecutionMultiplicity::PerEvaluation, + }, + PhysicalDagNode { + id: "root".into(), + operator: PhysicalOperator::Concat, + children: vec!["left".into(), "right".into()], + source_coverage: None, + output_buffer_bytes: 0, + retained_bytes: 0, + execution: ExecutionMultiplicity::PerEvaluation, + }, + ]; + let evidence = HashMap::from([ + ( + "scan".into(), + PhysicalNodeEvidence { + physical_id: "scan".into(), + statistics: OperatorStatistics::Scan { + edges: unary, + source_read_bytes: 80, + }, + output_buffer_bytes: 0, + }, + ), + ( + "left".into(), + PhysicalNodeEvidence { + physical_id: "left".into(), + statistics: OperatorStatistics::PassThrough { edges: unary }, + output_buffer_bytes: 0, + }, + ), + ( + "right".into(), + PhysicalNodeEvidence { + physical_id: "right".into(), + statistics: OperatorStatistics::PassThrough { edges: unary }, + output_buffer_bytes: 0, + }, + ), + ( + "root".into(), + PhysicalNodeEvidence { + physical_id: "root".into(), + statistics: OperatorStatistics::Concat { + inputs: vec![edge, edge], + output: EdgeStatistics { + rows: 20, + bytes: 160, + }, + promql: None, + }, + output_buffer_bytes: 0, + }, + ), + ]); + ( + EvidenceBackedPhysicalDag { + nodes, + root: "root".into(), + evidence, + }, + scope, + ) +} + +use asap_aware_mapping::storage_io::*; + +fn profile(dag: &EvidenceBackedPhysicalDag) -> StorageIoProfile { + StorageIoProfile { + evidence_version: "evidence-v1".into(), + observed_at_ms: 90, + valid_until_ms: 200, + calibration: StorageCalibration { + version: "requests-v1".into(), + cost_per_disk_read: 1.0, + cost_per_disk_write: 2.0, + cost_per_object_get: 3.0, + cost_per_object_put: 4.0, + }, + nodes: dag + .nodes + .iter() + .map(|node| { + ( + node.id.clone(), + StorageNodeEvidence { + node: node.clone(), + statistics: dag.evidence[&node.id].statistics.clone(), + accesses: if node.id == "scan" { + vec![StorageAccess { + operation: StorageOperation::ObjectGet, + extent_bytes: vec![40, 40], + bytes_per_request: 32, + }] + } else { + vec![] + }, + }, + ) + }) + .collect(), + } +} + +// The shared scan reads two objects once per evaluation, despite two parents. +#[test] +fn shared_io_counts_rounding_then_evaluations_and_explicit_writes() { + let (dag, scope) = fixture(); + let mut profile = profile(&dag); + profile + .nodes + .get_mut("root") + .unwrap() + .accesses + .push(StorageAccess { + operation: StorageOperation::DiskWrite, + extent_bytes: vec![160], + bytes_per_request: 64, + }); + let estimate = estimate_storage_io(&dag, &scope, &profile, "evidence-v1").unwrap(); + assert_eq!(estimate.total.object_gets, 12); + assert_eq!(estimate.total.disk_writes, 9); + assert_eq!(estimate.cost, 54.0); + assert_eq!(estimate.per_node["scan"].object_gets, 12); + assert_eq!(estimate.per_node["left"], StorageResources::default()); + assert_eq!(estimate.model_version, STORAGE_IO_MODEL_VERSION); +} + +// A retained scan is read once; downstream in-memory consumers add no reads. +#[test] +fn retained_state_does_not_reread_storage() { + let (mut dag, scope) = fixture(); + dag.nodes[0].execution = ExecutionMultiplicity::Once; + dag.nodes[0].retained_bytes = 80; + let estimate = estimate_storage_io(&dag, &scope, &profile(&dag), "evidence-v1").unwrap(); + assert_eq!(estimate.total.object_gets, 4); +} + +// Incomplete, expired, rebound, or inconsistent scan evidence cannot be ranked. +#[test] +fn incomplete_stale_and_incompatible_evidence_is_rejected() { + let (dag, scope) = fixture(); + let original = profile(&dag); + let mut missing = original.clone(); + missing.nodes.remove("left"); + assert!(estimate_storage_io(&dag, &scope, &missing, "evidence-v1").is_err()); + let mut stale = original.clone(); + stale.valid_until_ms = 100; + assert!(estimate_storage_io(&dag, &scope, &stale, "evidence-v1").is_err()); + stale.valid_until_ms = 200; + stale.observed_at_ms = 101; + assert!(estimate_storage_io(&dag, &scope, &stale, "evidence-v1").is_err()); + let mut mismatch = original.clone(); + mismatch.nodes.get_mut("scan").unwrap().accesses[0].extent_bytes = vec![79]; + assert!(estimate_storage_io(&dag, &scope, &mismatch, "evidence-v1").is_err()); + mismatch = original.clone(); + mismatch.nodes.get_mut("scan").unwrap().node.execution = ExecutionMultiplicity::Once; + assert!(estimate_storage_io(&dag, &scope, &mismatch, "evidence-v1").is_err()); + assert!(estimate_storage_io(&dag, &scope, &original, "another-version").is_err()); +} + +// Zero and non-finite values in wire evidence are rejected without defaults. +#[test] +fn wire_evidence_is_strict() { + let (dag, _) = fixture(); + let mut json = serde_json::to_value(profile(&dag)).unwrap(); + json["nodes"]["scan"]["accesses"][0]["bytes_per_request"] = serde_json::json!(1.5); + assert!(serde_json::from_value::(json).is_err()); + let mut json = serde_json::to_value(profile(&dag)).unwrap(); + json["calibration"]["cost_per_disk_read"] = serde_json::Value::Null; + assert!(serde_json::from_value::(json).is_err()); +} diff --git a/crates/devtools/src/bin/dag_export.rs b/crates/devtools/src/bin/dag_export.rs index 75cc3818..1362caac 100644 --- a/crates/devtools/src/bin/dag_export.rs +++ b/crates/devtools/src/bin/dag_export.rs @@ -88,6 +88,8 @@ use asap_types::types::AccuracyTarget; #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] #[serde(deny_unknown_fields)] struct PlannerCostDocument { + #[serde(default, skip_serializing_if = "Option::is_none")] + storage_io: Option, /// Immutable catalog/runtime evidence generation shared by this file. evidence_version: String, calibration: ResourceCalibration, @@ -276,6 +278,7 @@ fn plan_values_match_inner( } struct ExportPhysicalProvider<'a> { + storage_io: Option<&'a asap_aware_mapping::storage_io::StorageIoProfile>, evidence_version: &'a str, target: &'a TargetPhysicalEvidence, candidate: &'a CandidatePhysicalEvidence, @@ -297,6 +300,7 @@ impl PlannerPhysicalPlanProvider for ExportPhysicalProvider<'_> { version: self.evidence_version.into(), scope: self.target.scope.resolve()?, cache_profile: self.target.scope.cache_profile.clone(), + storage_io: self.storage_io.cloned(), }) } @@ -385,6 +389,7 @@ impl ExportPlannerCostModel<'_> { } Some(( ExportPhysicalProvider { + storage_io: self.document.storage_io.as_ref(), evidence_version: &self.document.evidence_version, target: target_evidence, candidate: candidate_evidence, @@ -412,7 +417,13 @@ impl ExportPlannerCostModel<'_> { if !provider.all_query_evidence_used() { return winner_cost_annotations(); } - let version = format!("{}+{}", ANALYTICAL_COST_MODEL_VERSION, calibration.version); + let mut version = format!("{}+{}", ANALYTICAL_COST_MODEL_VERSION, calibration.version); + if let Some((storage, _)) = &estimate.storage_io { + version.push_str(&format!( + "+{}+{}", + storage.model_version, storage.calibration_version + )); + } let scope = &provider.target.scope; let Ok((result_hits, buffer_hits)) = cache_hit_ratios( &scope.cache_profile, @@ -493,11 +504,44 @@ impl ExportPlannerCostModel<'_> { inputs.extend(cache_inputs.iter().cloned()); inputs }; + let storage_inputs = |storage: &asap_aware_mapping::storage_io::StorageEstimate| { + let mut terms: Vec<_> = storage + .total + .terms() + .into_iter() + .map(|(name, value)| CostInput { + name: name.into(), + value: value as f64, + unit: Some("operations".into()), + }) + .collect(); + let mut ids: Vec<_> = storage.per_node.keys().collect(); + ids.sort(); + for id in ids { + terms.extend( + storage.per_node[id] + .terms() + .into_iter() + .map(|(name, value)| CostInput { + name: format!("physical_node:{id}:{name}"), + value: value as f64, + unit: Some("operations".into()), + }), + ); + } + terms + }; + let mut raw_inputs = inputs(estimate.resources.raw); + let mut candidate_inputs = inputs(estimate.resources.candidate); + if let Some((raw, candidate)) = &estimate.storage_io { + raw_inputs.extend(storage_inputs(raw)); + candidate_inputs.extend(storage_inputs(candidate)); + } let baseline = CostAnnotation::modeled( estimate.raw_cost.0, CostUnit::CostUnits, &version, - inputs(estimate.resources.raw), + raw_inputs, ) .with_evidence_version(&self.document.evidence_version) .with_cache_profile(snapshot_cache_version(&provider)); @@ -505,7 +549,7 @@ impl ExportPlannerCostModel<'_> { estimate.candidate_cost.0, CostUnit::CostUnits, &version, - inputs(estimate.resources.candidate), + candidate_inputs, ) .with_baseline(BaselineRef::PreAsapRecomputation, estimate.raw_cost.0) .with_evidence_version(&self.document.evidence_version) @@ -1534,6 +1578,140 @@ mod tests { } } + fn fixture_raw_dag( + query: &QueryExpr, + candidate: &ReplacementSubDAG, + document: &PlannerCostDocument, + ) -> PhysicalDag { + let model = ExportPlannerCostModel { document }; + let root = Rc::new(query.clone()); + let target = asap_aware_mapping::replacement::TargetSubDAG::new(&root); + let (provider, _) = model.bound(candidate, &target).unwrap(); + let snapshot = provider.capture_evidence_snapshot(&target).unwrap(); + let evidence = + |request: PhysicalNodeRequest<'_>| provider.query_node_evidence(&snapshot, request); + lower_query_physical_dag(&root, &snapshot.scope, &evidence).unwrap() + } + + // JSON evidence reaches calibrated ranking and structured annotation inputs. + #[test] + fn storage_requests_export_and_change_plan_selection() { + use asap_aware_mapping::storage_io::*; + let (query, candidate, mut document) = cost_fixture(); + let raw = fixture_raw_dag(&query, &candidate, &document); + let candidate_dag = cheap_candidate_dag(); + let root = Rc::new(query.clone()); + let target = asap_aware_mapping::replacement::TargetSubDAG::new(&root); + assert!(ExportPlannerCostModel { + document: &document + } + .candidate_cost(&candidate, &target) + .is_some()); + let mut profile = StorageIoProfile { + evidence_version: document.evidence_version.clone(), + observed_at_ms: 900, + valid_until_ms: 2000, + calibration: StorageCalibration { + version: "requests-v1".into(), + cost_per_disk_read: 1.0, + cost_per_disk_write: 1.0, + cost_per_object_get: 1.0, + cost_per_object_put: 1.0, + }, + nodes: std::collections::HashMap::new(), + }; + for dag in [&raw, &candidate_dag] { + for node in &dag.nodes { + let statistics = dag.evidence[&node.id].statistics.clone(); + let accesses = match &statistics { + OperatorStatistics::Scan { + source_read_bytes, .. + } => vec![StorageAccess { + operation: StorageOperation::ObjectGet, + extent_bytes: vec![*source_read_bytes], + bytes_per_request: 4096, + }], + _ => vec![], + }; + profile.nodes.insert( + node.id.clone(), + StorageNodeEvidence { + node: node.clone(), + statistics, + accesses, + }, + ); + } + } + document.storage_io = Some(profile); + let parsed = + parse_planner_cost_document(&serde_json::to_string(&document).unwrap()).unwrap(); + let model = ExportPlannerCostModel { document: &parsed }; + let (baseline, selected, _) = model.annotations(&candidate, &root); + assert_eq!( + baseline + .inputs + .iter() + .find(|term| term.name == "object_get_operations") + .unwrap() + .value, + 160.0 + ); + assert_eq!( + selected + .inputs + .iter() + .find(|term| term.name == "object_get_operations") + .unwrap() + .value, + 10.0 + ); + assert!(selected + .inputs + .iter() + .any(|term| term.name == "physical_node:summary-read:object_get_operations")); + assert!(selected + .model_version + .as_ref() + .unwrap() + .contains(STORAGE_IO_MODEL_VERSION)); + assert_eq!( + selected.evidence_version.as_deref(), + Some("test-evidence-v1") + ); + document + .storage_io + .as_mut() + .unwrap() + .nodes + .get_mut("summary-read") + .unwrap() + .accesses + .push(StorageAccess { + operation: StorageOperation::DiskWrite, + extent_bytes: vec![1_000_000], + bytes_per_request: 1, + }); + assert!(ExportPlannerCostModel { + document: &document + } + .candidate_cost(&candidate, &target) + .is_none()); + document + .storage_io + .as_mut() + .unwrap() + .nodes + .remove(&raw.root); + assert!(ExportPlannerCostModel { + document: &document + } + .annotations(&candidate, &root) + .0 + .value + .is_none()); + } + fn test_scope() -> ComparisonScopeEvidence { ComparisonScopeEvidence { data_arrival: asap_types::workload::DataArrival::AtRest, @@ -1685,6 +1863,7 @@ mod tests { } }; let document = PlannerCostDocument { + storage_io: None, evidence_version: "test-evidence-v1".into(), calibration: ResourceCalibration { cost_per_cpu_op: 1.0, @@ -1997,6 +2176,7 @@ mod tests { *source_read_bytes = 1_024; let second_dag = cheap_candidate_dag(); let document = PlannerCostDocument { + storage_io: None, evidence_version: "test-evidence-v1".into(), calibration: ResourceCalibration { cost_per_cpu_op: 1.0, 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 e7954cbc..cc444a7a 100644 --- a/docs/design_docs/asap-aware-mapping/analytical-resource-cost.md +++ b/docs/design_docs/asap-aware-mapping/analytical-resource-cost.md @@ -1169,6 +1169,11 @@ available, that candidate is unavailable rather than costed as in-memory. ## Unsupported and future dimensions +The physical-plan adapter can also estimate explicitly evidenced disk/object +request counts using the optional storage profile. See +[storage operation estimates](../../developer_docs/storage-operation-costs.md) +for units, formulas, evidence requirements, and calibration. + An estimate is unavailable when required evidence, a physical formula, or a finite horizon is missing. Structural node counts are never substituted. diff --git a/docs/developer_docs/storage-operation-costs.md b/docs/developer_docs/storage-operation-costs.md new file mode 100644 index 00000000..81bbc2ed --- /dev/null +++ b/docs/developer_docs/storage-operation-costs.md @@ -0,0 +1,66 @@ +# Storage operation estimates + +The physical-plan ranking adapter accepts an optional `StorageIoProfile` in +its immutable `PhysicalEvidenceSnapshot`. `dag_export --planner-cost-json` +accepts the same profile in the document's top-level `storage_io` field. +Omitting the profile preserves the existing CPU/memory/scan-byte objective; +operation counts are unestimated, not inferred to be zero. + +A supplied profile must cover every reachable physical node, with an explicit +empty `accesses` list for nodes doing no storage I/O. Entries bind the complete +`PhysicalDagNode` and `OperatorStatistics`, so a reused ID cannot silently +borrow evidence from a different plan. Profiles may contain additional nodes +for other alternatives. Their evidence generation must equal the planner +snapshot version, and `observed_at_ms <= planning_time < valid_until_ms`. + +Each access supplies a storage operation, independent extent sizes in bytes, +and a positive effective payload size per request. Disk extents represent +contiguous ranges; object extents represent separately requested objects or +multipart payloads. Requests cannot coalesce across extents. + +```text +operations_per_execution = sum(ceil(extent_bytes / bytes_per_request)) +operations = operations_per_execution * executions +executions = 1 for Once, scope evaluation count for PerEvaluation +``` + +Zero-length extents cost zero data requests. Empty-object creation, metadata +requests, retries, seek latency, prefetch, and multipart control requests are +outside this payload model. The caller must supply the actual physical +request payload limit, not assume an object-store or disk block size. + +For example, two 40-byte objects with 32-byte requests require four GETs per +execution. Three evaluations require 12 GETs, even if two parents share the +scan. A retained `Once` scan requires four GETs over the same horizon. A +160-byte write with a 64-byte request size requires three writes per execution. +Round before multiplying by demand, using checked integer arithmetic. + +The output keeps four operation-count dimensions: disk reads, disk writes, +object GETs, and object PUTs. Counts do not replace byte estimates. A versioned +`StorageCalibration` assigns a finite, nonnegative cost per operation in the +same cost units as the base calibration. Both alternatives use the same +profile and coefficients before ranking. Scan read extents must add up to the +scan's authoritative `source_read_bytes`; explicit additional storage actions +can be bound to other physical nodes. + +`StorageEstimate` returns totals and per-node terms with model, evidence, and +calibration versions. DAG annotations expose counts as `CostInput`s with +`operations` units. Per-node terms use `physical_node::`; +annotation provenance identifies `storage-requests-v1` and its calibration. +The viewer displays these terms in the existing cost sidebar. The base byte +estimate and storage request estimate remain independently inspectable. + +Missing entries, expired/future evidence, incompatible node snapshots, zero +request sizes, invalid calibration, and overflow return typed analytical +errors. When used by plan ranking/export they make that comparison unavailable. +This profile extends the physical-plan adapter; the separate summary-maintenance +lifecycle estimator retains its existing dimensions. Cache behavior is a +separate model input and is not inferred here. + +Verification: + +```sh +cargo test -p asap-aware-mapping --test storage_io +cargo test -p asap-aware-mapping --lib storage_io +cargo test -p asap-devtools --bin dag_export storage_requests_export_and_change_plan_selection +``` From 2408179b27c90a0d6fcb0bce81b898104419ff62 Mon Sep 17 00:00:00 2001 From: zz_y Date: Mon, 7 Sep 2026 10:45:13 -0600 Subject: [PATCH 2/5] fix(cost): allow validated storage-only objectives --- .../src/physical_plan_cost_model.rs | 125 +++++++++++++++++- crates/devtools/src/bin/dag_export.rs | 14 ++ .../developer_docs/storage-operation-costs.md | 7 + 3 files changed, 142 insertions(+), 4 deletions(-) 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 051a4ba7..ead5c3d5 100644 --- a/crates/asap-aware-mapping/src/physical_plan_cost_model.rs +++ b/crates/asap-aware-mapping/src/physical_plan_cost_model.rs @@ -102,7 +102,12 @@ impl<'a> PhysicalPlanCostModel<'a> { "resource_calibration.version", )); } - calibration.validate()?; + // A zero base objective may be supplemented by storage coefficients; + // that check needs the target's immutable evidence snapshot. + match calibration.validate() { + Ok(()) | Err(AnalyticalCostError::ZeroCalibration) => {} + Err(error) => return Err(error), + } Ok(Self { provider, calibration, @@ -201,8 +206,32 @@ impl<'a> PhysicalPlanCostModel<'a> { )) }) .transpose()?; - let mut raw_cost = Cost(resources.raw.calibrated_cost(&self.calibration)?); - let mut candidate_cost = Cost(resources.candidate.calibrated_cost(&self.calibration)?); + let (mut raw_cost, mut candidate_cost) = match self.calibration.validate() { + Ok(()) => ( + Cost(resources.raw.calibrated_cost(&self.calibration)?), + Cost(resources.candidate.calibrated_cost(&self.calibration)?), + ), + Err(AnalyticalCostError::ZeroCalibration) => { + // Storage estimation above validates the coefficients and + // evidence. At least one priced dimension must remain. + let has_storage_objective = snapshot.storage_io.as_ref().is_some_and(|profile| { + let calibration = &profile.calibration; + [ + calibration.cost_per_disk_read, + calibration.cost_per_disk_write, + calibration.cost_per_object_get, + calibration.cost_per_object_put, + ] + .into_iter() + .any(|coefficient| coefficient > 0.0) + }); + if !has_storage_objective { + return Err(AnalyticalCostError::ZeroCalibration); + } + (Cost(0.0), Cost(0.0)) + } + Err(error) => return Err(error), + }; if let Some((raw, candidate)) = &storage_io { raw_cost.0 += raw.cost; candidate_cost.0 += candidate.cost; @@ -361,6 +390,7 @@ mod tests { } struct TestProvider { + storage_io: Option, summary_available: bool, candidate_scan_bytes: u64, snapshot_calls: Cell, @@ -370,6 +400,7 @@ mod tests { impl TestProvider { fn new(summary_available: bool, candidate_scan_bytes: u64) -> Self { Self { + storage_io: None, summary_available, candidate_scan_bytes, snapshot_calls: Cell::new(0), @@ -456,7 +487,7 @@ mod tests { version: "test-snapshot-1".into(), scope: scope(), cache_profile: CacheProfile::no_cache(), - storage_io: None, + storage_io: self.storage_io.clone(), }) } @@ -508,6 +539,92 @@ mod tests { } } + // A request-only objective ranks complete evidence and rejects absent, + // zero, or invalid supplemental calibration. + #[test] + fn storage_only_objective_requires_positive_valid_storage_calibration() { + use crate::storage_io::*; + let root = query(); + let target = TargetSubDAG::new(&root); + let mut provider = TestProvider::new(true, 800); + let snapshot = provider.capture_evidence_snapshot(&target).unwrap(); + let raw = lower_query_physical_dag( + &root, + &snapshot.scope, + &QueryEvidence { + provider: &provider, + snapshot: &snapshot, + }, + ) + .unwrap(); + let mut profile = StorageIoProfile { + evidence_version: snapshot.version, + observed_at_ms: 900, + valid_until_ms: 2000, + calibration: StorageCalibration { + version: "requests-v1".into(), + cost_per_disk_read: 0.0, + cost_per_disk_write: 0.0, + cost_per_object_get: 1.0, + cost_per_object_put: 0.0, + }, + nodes: HashMap::new(), + }; + for dag in [raw, provider.summary_dag(&snapshot.scope)] { + for node in dag.nodes { + let statistics = dag.evidence[&node.id].statistics.clone(); + let accesses = match statistics { + OperatorStatistics::Scan { + source_read_bytes, .. + } => vec![StorageAccess { + operation: StorageOperation::ObjectGet, + extent_bytes: vec![source_read_bytes], + bytes_per_request: 400, + }], + _ => vec![], + }; + profile.nodes.insert( + node.id.clone(), + StorageNodeEvidence { + node, + statistics, + accesses, + }, + ); + } + } + let base = ResourceCalibration { + cost_per_cpu_op: 0.0, + cost_per_scan_byte: 0.0, + cost_per_retained_byte: 0.0, + version: "unused-base-v1".into(), + }; + let candidates = + crate::replacement::SketchAlgorithmStrategy::default_cost_model().replacements(&target); + provider.storage_io = Some(profile.clone()); + let model = PhysicalPlanCostModel::new(&provider, base.clone()).unwrap(); + let estimate = model.estimate_candidate(&candidates[0], &target).unwrap(); + assert_eq!(estimate.raw_cost.0, 20.0); + assert_eq!(estimate.candidate_cost.0, 2.0); + assert_eq!( + model.candidate_cost(&candidates[0], &target), + Some(Cost(2.0)) + ); + drop(model); + for coefficient in [0.0, -1.0, f64::NAN] { + profile.calibration.cost_per_object_get = coefficient; + provider.storage_io = Some(profile.clone()); + let model = PhysicalPlanCostModel::new(&provider, base.clone()).unwrap(); + assert!(model.estimate_candidate(&candidates[0], &target).is_err()); + } + provider.storage_io = None; + let model = PhysicalPlanCostModel::new(&provider, base).unwrap(); + assert_eq!( + model.estimate_candidate(&candidates[0], &target), + Err(AnalyticalCostError::ZeroCalibration) + ); + } + #[test] fn global_selection_uses_complete_physical_comparison() { let root = query(); diff --git a/crates/devtools/src/bin/dag_export.rs b/crates/devtools/src/bin/dag_export.rs index 1362caac..833503e9 100644 --- a/crates/devtools/src/bin/dag_export.rs +++ b/crates/devtools/src/bin/dag_export.rs @@ -1679,6 +1679,20 @@ mod tests { selected.evidence_version.as_deref(), Some("test-evidence-v1") ); + // JSON evidence also supports an objective priced only by requests. + let mut requests_only = parsed.clone(); + requests_only.calibration.cost_per_cpu_op = 0.0; + requests_only.calibration.cost_per_scan_byte = 0.0; + requests_only.calibration.cost_per_retained_byte = 0.0; + let requests_only = + parse_planner_cost_document(&serde_json::to_string(&requests_only).unwrap()).unwrap(); + let model = ExportPlannerCostModel { + document: &requests_only, + }; + assert_eq!(model.candidate_cost(&candidate, &target), Some(Cost(10.0))); + let (baseline, selected, _) = model.annotations(&candidate, &root); + assert_eq!(baseline.value, Some(160.0)); + assert_eq!(selected.value, Some(10.0)); document .storage_io .as_mut() diff --git a/docs/developer_docs/storage-operation-costs.md b/docs/developer_docs/storage-operation-costs.md index 81bbc2ed..34f19b8b 100644 --- a/docs/developer_docs/storage-operation-costs.md +++ b/docs/developer_docs/storage-operation-costs.md @@ -43,6 +43,13 @@ profile and coefficients before ranking. Scan read extents must add up to the scan's authoritative `source_read_bytes`; explicit additional storage actions can be bound to other physical nodes. +A request-only objective may set all base CPU, scan-byte, and memory +coefficients to zero if the snapshot supplies valid storage evidence and at +least one positive storage coefficient. With a zero base objective, missing +storage evidence or an all-zero storage calibration makes the comparison +unavailable. This check occurs when the target snapshot is available; +standalone base-resource calibration still rejects an all-zero objective. + `StorageEstimate` returns totals and per-node terms with model, evidence, and calibration versions. DAG annotations expose counts as `CostInput`s with `operations` units. Per-node terms use `physical_node::`; From 529ba2469346113f92a4c28bac618abbe5076a09 Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 8 Sep 2026 12:57:47 -0600 Subject: [PATCH 3/5] fix(cost): require base calibration provenance for storage ranking --- .../src/physical_plan_cost_model.rs | 26 +++++ crates/asap-aware-mapping/tests/storage_io.rs | 109 ++++++++++++++++++ .../developer_docs/storage-operation-costs.md | 3 + 3 files changed, 138 insertions(+) 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 ead5c3d5..a5777bec 100644 --- a/crates/asap-aware-mapping/src/physical_plan_cost_model.rs +++ b/crates/asap-aware-mapping/src/physical_plan_cost_model.rs @@ -97,6 +97,8 @@ impl<'a> PhysicalPlanCostModel<'a> { provider: &'a dyn PlannerPhysicalPlanProvider, calibration: ResourceCalibration, ) -> Result { + // Exported scores retain base calibration provenance even when the + // objective is priced entirely by supplementary storage operations. if calibration.version.trim().is_empty() { return Err(AnalyticalCostError::MissingOrStale( "resource_calibration.version", @@ -539,6 +541,30 @@ mod tests { } } + /// Both base-priced and storage-only objectives need identifiable calibration. + #[test] + fn blank_base_calibration_version_is_rejected_before_snapshot_lookup() { + let provider = TestProvider::new(true, 800); + for version in ["", " \t\n"] { + for zero_base in [false, true] { + let mut base = calibration(); + base.version = version.into(); + if zero_base { + base.cost_per_cpu_op = 0.0; + base.cost_per_scan_byte = 0.0; + base.cost_per_retained_byte = 0.0; + } + assert!(matches!( + PhysicalPlanCostModel::new(&provider, base), + Err(AnalyticalCostError::MissingOrStale( + "resource_calibration.version" + )) + )); + } + } + assert_eq!(provider.snapshot_calls.get(), 0); + } + // A request-only objective ranks complete evidence and rejects absent, // zero, or invalid supplemental calibration. #[test] diff --git a/crates/asap-aware-mapping/tests/storage_io.rs b/crates/asap-aware-mapping/tests/storage_io.rs index 0d6e4a77..88f88097 100644 --- a/crates/asap-aware-mapping/tests/storage_io.rs +++ b/crates/asap-aware-mapping/tests/storage_io.rs @@ -243,3 +243,112 @@ fn wire_evidence_is_strict() { json["calibration"]["cost_per_disk_read"] = serde_json::Value::Null; assert!(serde_json::from_value::(json).is_err()); } + +// Zero-length payloads add no requests, without turning absent reads into zero. +#[test] +fn zero_extents_and_all_request_kinds_preserve_dimensions() { + let (dag, scope) = fixture(); + let mut profile = profile(&dag); + profile.nodes.get_mut("scan").unwrap().accesses[0] + .extent_bytes + .push(0); + profile.nodes.get_mut("root").unwrap().accesses = [ + StorageOperation::DiskRead, + StorageOperation::DiskWrite, + StorageOperation::ObjectGet, + StorageOperation::ObjectPut, + ] + .into_iter() + .map(|operation| StorageAccess { + operation, + extent_bytes: vec![0, 1, 9], + bytes_per_request: 8, + }) + .collect(); + let estimate = estimate_storage_io(&dag, &scope, &profile, "evidence-v1").unwrap(); + assert_eq!( + estimate.total, + StorageResources { + disk_reads: 9, + disk_writes: 9, + object_gets: 21, + object_puts: 9, + } + ); + assert_eq!(estimate.cost, 126.0); + profile.nodes.get_mut("scan").unwrap().accesses.clear(); + assert!(estimate_storage_io(&dag, &scope, &profile, "evidence-v1").is_err()); +} + +// Valid local integer counts must not wrap when scaled or composed across nodes. +#[test] +fn multiplicity_and_cross_node_overflow_are_unavailable() { + let (dag, mut scope) = fixture(); + let mut profile = profile(&dag); + let large = StorageAccess { + operation: StorageOperation::DiskWrite, + extent_bytes: vec![u64::MAX], + bytes_per_request: 1, + }; + profile + .nodes + .get_mut("root") + .unwrap() + .accesses + .push(large.clone()); + assert_eq!( + estimate_storage_io(&dag, &scope, &profile, "evidence-v1"), + Err(asap_aware_mapping::analytical_cost::AnalyticalCostError::Overflow) + ); + scope.recurrence = QueryRecurrence::OneTime { + invocations: 1, + execute_at: None, + }; + profile + .nodes + .get_mut("left") + .unwrap() + .accesses + .push(StorageAccess { + extent_bytes: vec![1], + ..large + }); + assert_eq!( + estimate_storage_io(&dag, &scope, &profile, "evidence-v1"), + Err(asap_aware_mapping::analytical_cost::AnalyticalCostError::Overflow) + ); +} + +// Reusing map keys must not bind another node, source snapshot, or statistics. +#[test] +fn storage_node_identity_statistics_and_calibration_provenance_are_bound() { + let (dag, scope) = fixture(); + let original = profile(&dag); + let mut bad = original.clone(); + bad.nodes.get_mut("scan").unwrap().node.id = "another-scan".into(); + assert!(estimate_storage_io(&dag, &scope, &bad, "evidence-v1").is_err()); + bad = original.clone(); + bad.nodes + .get_mut("scan") + .unwrap() + .node + .source_coverage + .as_mut() + .unwrap() + .source_snapshot_id = "another-source".into(); + assert!(estimate_storage_io(&dag, &scope, &bad, "evidence-v1").is_err()); + bad = original.clone(); + if let OperatorStatistics::Scan { + source_read_bytes, .. + } = &mut bad.nodes.get_mut("scan").unwrap().statistics + { + *source_read_bytes = 79; + } + assert!(estimate_storage_io(&dag, &scope, &bad, "evidence-v1").is_err()); + bad = original.clone(); + bad.calibration.version = " \t".into(); + assert!(estimate_storage_io(&dag, &scope, &bad, "evidence-v1").is_err()); + let mut missing = dag.clone(); + missing.evidence.remove("scan"); + assert!(estimate_storage_io(&missing, &scope, &original, "evidence-v1").is_err()); +} diff --git a/docs/developer_docs/storage-operation-costs.md b/docs/developer_docs/storage-operation-costs.md index 34f19b8b..dd93057b 100644 --- a/docs/developer_docs/storage-operation-costs.md +++ b/docs/developer_docs/storage-operation-costs.md @@ -49,6 +49,9 @@ least one positive storage coefficient. With a zero base objective, missing storage evidence or an all-zero storage calibration makes the comparison unavailable. This check occurs when the target snapshot is available; standalone base-resource calibration still rejects an all-zero objective. +The physical-plan adapter requires a nonblank base calibration version even +for a request-only objective, so exported combined model provenance remains +identifiable; a storage calibration version cannot substitute for it. `StorageEstimate` returns totals and per-node terms with model, evidence, and calibration versions. DAG annotations expose counts as `CostInput`s with From aa486bf7b6e9a3e9f0f7aa1ec327910efee096b5 Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 8 Sep 2026 13:11:30 -0600 Subject: [PATCH 4/5] refactor(resources): share storage operation count dimensions --- .../src/physical_plan_cost_model.rs | 13 +++ crates/asap-aware-mapping/src/storage_io.rs | 49 ++-------- crates/asap-aware-mapping/tests/storage_io.rs | 21 +++++ crates/types/src/resources.rs | 2 + crates/types/src/resources/storage.rs | 94 +++++++++++++++++++ .../developer_docs/storage-operation-costs.md | 8 ++ 6 files changed, 147 insertions(+), 40 deletions(-) create mode 100644 crates/types/src/resources/storage.rs 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 a5777bec..f1437647 100644 --- a/crates/asap-aware-mapping/src/physical_plan_cost_model.rs +++ b/crates/asap-aware-mapping/src/physical_plan_cost_model.rs @@ -651,6 +651,19 @@ mod tests { ); } + /// Shared count defaults must not turn an omitted profile into measured zero I/O. + #[test] + fn missing_storage_profile_remains_unestimated() { + let root = query(); + let target = TargetSubDAG::new(&root); + let candidates = + crate::replacement::SketchAlgorithmStrategy::default_cost_model().replacements(&target); + let provider = TestProvider::new(true, 800); + let model = PhysicalPlanCostModel::new(&provider, calibration()).unwrap(); + let estimate = model.estimate_candidate(&candidates[0], &target).unwrap(); + assert!(estimate.storage_io.is_none()); + } + #[test] fn global_selection_uses_complete_physical_comparison() { let root = query(); diff --git a/crates/asap-aware-mapping/src/storage_io.rs b/crates/asap-aware-mapping/src/storage_io.rs index 1bfeaad8..5ff603ca 100644 --- a/crates/asap-aware-mapping/src/storage_io.rs +++ b/crates/asap-aware-mapping/src/storage_io.rs @@ -4,6 +4,9 @@ use std::collections::{HashMap, HashSet}; use serde::{Deserialize, Serialize}; +// Keep the original public import path while sharing the sole type definition. +pub use asap_types::resources::StorageResources; + use crate::analytical_cost::{ estimate_physical_dag, AnalyticalCostError, EvidenceBackedPhysicalDag, ExecutionMultiplicity, PhysicalDagNode, PhysicalOperator, @@ -31,44 +34,6 @@ pub struct StorageAccess { pub bytes_per_request: u64, } -#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] -pub struct StorageResources { - pub disk_reads: u64, - pub disk_writes: u64, - pub object_gets: u64, - pub object_puts: u64, -} - -impl StorageResources { - pub fn terms(self) -> [(&'static str, u64); 4] { - [ - ("disk_read_operations", self.disk_reads), - ("disk_write_operations", self.disk_writes), - ("object_get_operations", self.object_gets), - ("object_put_operations", self.object_puts), - ] - } - fn add(&mut self, other: Self) -> Result<(), AnalyticalCostError> { - self.disk_reads = self - .disk_reads - .checked_add(other.disk_reads) - .ok_or(AnalyticalCostError::Overflow)?; - self.disk_writes = self - .disk_writes - .checked_add(other.disk_writes) - .ok_or(AnalyticalCostError::Overflow)?; - self.object_gets = self - .object_gets - .checked_add(other.object_gets) - .ok_or(AnalyticalCostError::Overflow)?; - self.object_puts = self - .object_puts - .checked_add(other.object_puts) - .ok_or(AnalyticalCostError::Overflow)?; - Ok(()) - } -} - #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub struct StorageCalibration { @@ -217,7 +182,9 @@ pub fn estimate_storage_io( .ok_or(AnalyticalCostError::Overflow)?; } } - local.add(term)?; + local = local + .checked_add(term) + .ok_or(AnalyticalCostError::Overflow)?; } if node.operator == PhysicalOperator::Scan { let OperatorStatistics::Scan { @@ -232,7 +199,9 @@ pub fn estimate_storage_io( )); } } - total.add(local)?; + total = total + .checked_add(local) + .ok_or(AnalyticalCostError::Overflow)?; per_node.insert(id.into(), local); } Ok(StorageEstimate { diff --git a/crates/asap-aware-mapping/tests/storage_io.rs b/crates/asap-aware-mapping/tests/storage_io.rs index 88f88097..d8fd8614 100644 --- a/crates/asap-aware-mapping/tests/storage_io.rs +++ b/crates/asap-aware-mapping/tests/storage_io.rs @@ -138,6 +138,27 @@ fn fixture() -> (EvidenceBackedPhysicalDag, ComparisonScope) { use asap_aware_mapping::storage_io::*; +// The compatibility import and shared resource namespace expose one Rust type. +#[test] +fn storage_estimates_use_the_shared_resource_type_without_wire_changes() { + let (dag, scope) = fixture(); + let estimate = estimate_storage_io(&dag, &scope, &profile(&dag), "evidence-v1").unwrap(); + let shared: asap_types::resources::StorageResources = estimate.total; + let legacy: asap_aware_mapping::storage_io::StorageResources = shared; + assert_eq!(shared, legacy); + let wire = serde_json::to_value(&estimate).unwrap(); + assert_eq!( + wire["total"], + serde_json::json!({ + "disk_reads": 0, "disk_writes": 0, "object_gets": 12, "object_puts": 0, + }) + ); + assert_eq!( + serde_json::from_value::(wire).unwrap(), + estimate + ); +} + fn profile(dag: &EvidenceBackedPhysicalDag) -> StorageIoProfile { StorageIoProfile { evidence_version: "evidence-v1".into(), diff --git a/crates/types/src/resources.rs b/crates/types/src/resources.rs index d1bde516..e7241ebc 100644 --- a/crates/types/src/resources.rs +++ b/crates/types/src/resources.rs @@ -10,6 +10,8 @@ pub mod cache; pub mod cpu; pub mod measurement; pub mod physical; +pub mod storage; +pub use storage::StorageResources; pub use cache::{CacheCapacityEvidence, CacheEvidence, CacheProfile}; pub use cpu::{MeasuredCpu, ModeledCpu}; diff --git a/crates/types/src/resources/storage.rs b/crates/types/src/resources/storage.rs new file mode 100644 index 00000000..e33c47c6 --- /dev/null +++ b/crates/types/src/resources/storage.rs @@ -0,0 +1,94 @@ +//! Storage request-count dimensions, independent of estimation and calibration. + +use serde::{Deserialize, Serialize}; + +/// Explicit counts of physical data requests, not transferred bytes or CPU work. +/// Unavailable storage evidence is represented by the enclosing optional profile, +/// not by constructing an all-zero count vector. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct StorageResources { + pub disk_reads: u64, + pub disk_writes: u64, + pub object_gets: u64, + pub object_puts: u64, +} + +impl StorageResources { + pub fn terms(self) -> [(&'static str, u64); 4] { + [ + ("disk_read_operations", self.disk_reads), + ("disk_write_operations", self.disk_writes), + ("object_get_operations", self.object_gets), + ("object_put_operations", self.object_puts), + ] + } + + /// Compose exact counts without wrapping or partially updating an input. + pub fn checked_add(self, other: Self) -> Option { + Some(Self { + disk_reads: self.disk_reads.checked_add(other.disk_reads)?, + disk_writes: self.disk_writes.checked_add(other.disk_writes)?, + object_gets: self.object_gets.checked_add(other.object_gets)?, + object_puts: self.object_puts.checked_add(other.object_puts)?, + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Moving the type preserves its four required integer wire fields exactly. + #[test] + fn storage_counts_keep_the_existing_wire_shape() { + let wire = serde_json::json!({ + "disk_reads": 1, "disk_writes": 2, "object_gets": 3, "object_puts": 4, + }); + let counts: StorageResources = serde_json::from_value(wire.clone()).unwrap(); + assert_eq!(serde_json::to_value(counts).unwrap(), wire); + assert_eq!(counts.terms()[2], ("object_get_operations", 3)); + for invalid in [ + serde_json::json!(null), + serde_json::json!({}), + serde_json::json!({ + "disk_reads": 1.5, "disk_writes": 2, "object_gets": 3, "object_puts": 4, + }), + ] { + assert!(serde_json::from_value::(invalid).is_err()); + } + } + + /// Every independent dimension fails closed on integer addition overflow. + #[test] + fn storage_count_addition_is_checked_in_each_dimension() { + let one = StorageResources { + disk_reads: 1, + disk_writes: 1, + object_gets: 1, + object_puts: 1, + }; + let zero = StorageResources::default(); + assert_eq!(zero.checked_add(one), Some(one)); + for counts in [ + StorageResources { + disk_reads: u64::MAX, + ..zero + }, + StorageResources { + disk_writes: u64::MAX, + ..zero + }, + StorageResources { + object_gets: u64::MAX, + ..zero + }, + StorageResources { + object_puts: u64::MAX, + ..zero + }, + ] { + assert_eq!(counts.checked_add(one), None); + assert_eq!(counts.checked_add(zero), Some(counts)); + } + } +} diff --git a/docs/developer_docs/storage-operation-costs.md b/docs/developer_docs/storage-operation-costs.md index dd93057b..03b0d06b 100644 --- a/docs/developer_docs/storage-operation-costs.md +++ b/docs/developer_docs/storage-operation-costs.md @@ -43,6 +43,14 @@ profile and coefficients before ranking. Scan read extents must add up to the scan's authoritative `source_read_bytes`; explicit additional storage actions can be bound to other physical nodes. +Their sole data type is `asap_types::resources::StorageResources`, defined in +the shared resources module alongside CPU and byte dimensions. The mapping +crate re-exports it at `asap_aware_mapping::storage_io::StorageResources` for +source compatibility; the four integer JSON fields are unchanged. Pure term +enumeration and checked addition live with the shared type. Access profiles, +request-count estimation, calibration, and ranking remain in the mapping +crate. An omitted profile still means unavailable operation counts, not zero. + A request-only objective may set all base CPU, scan-byte, and memory coefficients to zero if the snapshot supplies valid storage evidence and at least one positive storage coefficient. With a zero base objective, missing From 160d23a949c2a4e99a97acf5784b7f42726ce8e4 Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 8 Sep 2026 13:29:39 -0600 Subject: [PATCH 5/5] fix(cost): reject unsupported cache and storage evidence combinations --- .../src/physical_plan_cost_model.rs | 10 +++++++ crates/devtools/src/bin/dag_export.rs | 30 +++++++++++++++++++ crates/types/src/resources.rs | 2 +- .../developer_docs/storage-operation-costs.md | 9 ++++-- 4 files changed, 48 insertions(+), 3 deletions(-) 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 f1437647..86025910 100644 --- a/crates/asap-aware-mapping/src/physical_plan_cost_model.rs +++ b/crates/asap-aware-mapping/src/physical_plan_cost_model.rs @@ -156,6 +156,16 @@ impl<'a> PhysicalPlanCostModel<'a> { target: &TargetSubDAG<'_>, ) -> Result { let (snapshot, raw) = self.target_evidence(target)?; + // Aggregate cache hit ratios do not identify which independently rounded + // extents issue requests. Do not mix cache-adjusted bytes/CPU with + // uncached request counts until cache-aware extent evidence is available. + if snapshot.storage_io.is_some() + && matches!(&snapshot.cache_profile, CacheProfile::Evidence(_)) + { + return Err(AnalyticalCostError::InvalidCacheEvidence( + "storage operation costs require an explicit no-cache profile; cache-aware extent evidence is unavailable", + )); + } let scope = &snapshot.scope; let evidence = QueryEvidence { provider: self.provider, diff --git a/crates/devtools/src/bin/dag_export.rs b/crates/devtools/src/bin/dag_export.rs index 833503e9..3eb51a9a 100644 --- a/crates/devtools/src/bin/dag_export.rs +++ b/crates/devtools/src/bin/dag_export.rs @@ -1675,6 +1675,18 @@ mod tests { .as_ref() .unwrap() .contains(STORAGE_IO_MODEL_VERSION)); + assert_eq!(selected.cache_profile.as_deref(), Some("no-cache-v1")); + for name in ["result_cache_hit_ratio", "buffer_cache_hit_ratio"] { + assert_eq!( + selected + .inputs + .iter() + .find(|term| term.name == name) + .unwrap() + .value, + 0.0 + ); + } assert_eq!( selected.evidence_version.as_deref(), Some("test-evidence-v1") @@ -1693,6 +1705,24 @@ mod tests { let (baseline, selected, _) = model.annotations(&candidate, &root); assert_eq!(baseline.value, Some(160.0)); assert_eq!(selected.value, Some(10.0)); + // Aggregate cache hit assumptions cannot locate cached extents or + // reconstruct independently rounded physical storage requests. + let mut cached = serde_json::to_value(&requests_only).unwrap(); + cached["targets"][0]["scope"]["cache_profile"] = serde_json::json!({ + "profile": "evidence", "version": "cached-with-storage-v1", + "distinct_evaluations": 1, "repeated_identical_evaluations": 9, + "result_cache": {"working_set_bytes": 100, "capacity_bytes": 100}, + "buffer_cache": {"working_set_bytes": 100, "capacity_bytes": 100}, + "result_invalidation_ratio": null, + }); + let cached = parse_planner_cost_document(&cached.to_string()).unwrap(); + let cached_model = ExportPlannerCostModel { document: &cached }; + assert!(cached_model.candidate_cost(&candidate, &target).is_none()); + assert!(cached_model + .annotations(&candidate, &root) + .0 + .value + .is_none()); document .storage_io .as_mut() diff --git a/crates/types/src/resources.rs b/crates/types/src/resources.rs index e7241ebc..d2107a7f 100644 --- a/crates/types/src/resources.rs +++ b/crates/types/src/resources.rs @@ -11,8 +11,8 @@ pub mod cpu; pub mod measurement; pub mod physical; pub mod storage; -pub use storage::StorageResources; pub use cache::{CacheCapacityEvidence, CacheEvidence, CacheProfile}; +pub use storage::StorageResources; pub use cpu::{MeasuredCpu, ModeledCpu}; pub use measurement::Measurement; diff --git a/docs/developer_docs/storage-operation-costs.md b/docs/developer_docs/storage-operation-costs.md index 03b0d06b..564a6534 100644 --- a/docs/developer_docs/storage-operation-costs.md +++ b/docs/developer_docs/storage-operation-costs.md @@ -72,8 +72,13 @@ Missing entries, expired/future evidence, incompatible node snapshots, zero request sizes, invalid calibration, and overflow return typed analytical errors. When used by plan ranking/export they make that comparison unavailable. This profile extends the physical-plan adapter; the separate summary-maintenance -lifecycle estimator retains its existing dimensions. Cache behavior is a -separate model input and is not inferred here. +lifecycle estimator retains its existing dimensions. Combined physical-plan +ranking currently supports storage profiles only with an explicit `NoCache` +profile. `CacheProfile::Evidence` together with storage evidence makes the +comparison unavailable: aggregate cache hit ratios cannot identify which +independently rounded extents issue requests. Supporting that combination +requires cache-aware extent/request evidence; the adapter does not mix +cache-adjusted bytes and CPU with uncached operation counts. Verification: