diff --git a/crates/asap-aware-mapping/src/boundary_cost.rs b/crates/asap-aware-mapping/src/boundary_cost.rs new file mode 100644 index 00000000..2474e72e --- /dev/null +++ b/crates/asap-aware-mapping/src/boundary_cost.rs @@ -0,0 +1,232 @@ +//! Byte estimates at deployment-declared physical boundaries. + +use crate::analytical_cost::{ + estimate_physical_dag, AnalyticalCostError, EvidenceBackedPhysicalDag, ExecutionMultiplicity, + PhysicalDagNode, +}; +use crate::physical_operator_statistics::{ComparisonScope, OperatorStatistics}; +pub use asap_types::resources::{BoundaryKind, BoundaryResources, MaterializationMedium}; +use serde::{Deserialize, Serialize}; +use std::collections::{HashMap, HashSet}; + +pub const BOUNDARY_MODEL_VERSION: &str = "physical-boundary-bytes-v1"; + +/// A physical action on a producer output, distinct from a logical DAG edge. +/// With `consumer = None`, one action serves all consumers. With a consumer, +/// it is a separate action per execution of that downstream physical node. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct PhysicalBoundary { + pub id: String, + pub consumer: Option, + pub kind: BoundaryKind, + pub logical_bytes: u64, + /// Encoded payload per execution; compression is explicit evidence. + pub encoded_bytes: u64, + /// Copies actually transferred or written (e.g. broadcast replicas). + pub copies: u64, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct BoundaryNodeEvidence { + pub node: PhysicalDagNode, + pub statistics: OperatorStatistics, + pub boundaries: Vec, +} + +/// Complete supplementary physical binding captured atomically with the +/// planner snapshot. An empty per-node list explicitly asserts no boundary. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct BoundaryProfile { + pub evidence_version: String, + pub observed_at_ms: u64, + pub valid_until_ms: u64, + pub calibration: BoundaryCalibration, + pub plans: Vec, +} + +/// Boundaries belong to a complete alternative: a producer's consumers may +/// differ between plans even when its physical identity is unchanged. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct BoundaryPlanEvidence { + pub root: String, + pub nodes: HashMap, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct BoundaryCalibration { + pub version: String, + pub cost_per_network_byte: f64, + pub cost_per_materialization_byte: f64, +} + +impl BoundaryCalibration { + pub fn cost(&self, value: BoundaryResources) -> Result { + if self.version.trim().is_empty() + || [ + self.cost_per_network_byte, + self.cost_per_materialization_byte, + ] + .iter() + .any(|value| !value.is_finite() || *value < 0.0) + { + return Err(invalid("invalid boundary calibration")); + } + let cost = value.network_bytes as f64 * self.cost_per_network_byte + + value.materialization_bytes as f64 * self.cost_per_materialization_byte; + if !cost.is_finite() { + return Err(AnalyticalCostError::Overflow); + } + Ok(cost) + } +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct BoundaryEstimate { + pub total: BoundaryResources, + pub per_node: HashMap, + pub per_boundary: 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) +} + +pub fn estimate_boundaries( + dag: &EvidenceBackedPhysicalDag, + scope: &ComparisonScope, + profile: &BoundaryProfile, + evidence_version: &str, +) -> Result { + 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("boundary profile")); + } + let mut matching_plans = profile.plans.iter().filter(|plan| { + plan.root == dag.root + && plan.nodes.len() == dag.nodes.len() + && dag.nodes.iter().all(|node| { + plan.nodes.get(&node.id).is_some_and(|evidence| { + evidence.node == *node + && dag.evidence.get(&node.id).is_some_and(|physical| { + physical.physical_id == node.id + && physical.output_buffer_bytes == node.output_buffer_bytes + && evidence.statistics == physical.statistics + }) + }) + }) + }); + let plan = matching_plans + .next() + .ok_or(AnalyticalCostError::MissingOrStale( + "boundary physical plan", + ))?; + if matching_plans.next().is_some() { + return Err(invalid("ambiguous boundary physical plan")); + } + let by_id: HashMap<_, _> = dag + .nodes + .iter() + .map(|node| (node.id.as_str(), node)) + .collect(); + let mut reachable = HashSet::new(); + let mut pending = vec![dag.root.as_str()]; + while let Some(id) = pending.pop() { + if reachable.insert(id) { + pending.extend(by_id[id].children.iter().map(String::as_str)); + } + } + let mut total = BoundaryResources::default(); + let mut per_node = HashMap::new(); + let mut per_boundary = HashMap::new(); + for id in &reachable { + let node = by_id[id]; + let evidence = &plan.nodes[*id]; + let mut local = BoundaryResources::default(); + for boundary in &evidence.boundaries { + if boundary.id.trim().is_empty() || per_boundary.contains_key(&boundary.id) { + return Err(invalid("duplicate or blank physical boundary identity")); + } + if boundary.logical_bytes != evidence.statistics.output().bytes + || (boundary.logical_bytes == 0) != (boundary.encoded_bytes == 0) + || boundary.copies == 0 + { + return Err(invalid( + "boundary payload is incompatible with producer output", + )); + } + let execution = if let Some(consumer) = &boundary.consumer { + if !reachable.contains(consumer.as_str()) + || !by_id[consumer.as_str()] + .children + .iter() + .any(|child| child == *id) + { + return Err(invalid( + "boundary consumer is not a reachable physical edge", + )); + } + by_id[consumer.as_str()].execution + } else { + node.execution + }; + let executions = match execution { + ExecutionMultiplicity::Once => 1, + ExecutionMultiplicity::PerEvaluation => evaluations, + }; + let bytes = boundary + .encoded_bytes + .checked_mul(boundary.copies) + .and_then(|bytes| bytes.checked_mul(executions)) + .ok_or(AnalyticalCostError::Overflow)?; + let mut term = BoundaryResources::default(); + match &boundary.kind { + BoundaryKind::Network { + source_location, + destination_location, + } => { + if source_location.trim().is_empty() + || destination_location.trim().is_empty() + || source_location.trim() == destination_location.trim() + { + return Err(invalid( + "network boundary needs distinct non-empty locations", + )); + } + term.network_bytes = bytes; + } + BoundaryKind::Materialization { .. } => term.materialization_bytes = bytes, + } + local = local + .checked_add(term) + .ok_or(AnalyticalCostError::Overflow)?; + per_boundary.insert(boundary.id.clone(), term); + } + total = total + .checked_add(local) + .ok_or(AnalyticalCostError::Overflow)?; + per_node.insert((*id).into(), local); + } + Ok(BoundaryEstimate { + total, + per_node, + per_boundary, + model_version: BOUNDARY_MODEL_VERSION.into(), + evidence_version: profile.evidence_version.clone(), + calibration_version: profile.calibration.version.clone(), + cost: profile.calibration.cost(total)?, + }) +} diff --git a/crates/asap-aware-mapping/src/lib.rs b/crates/asap-aware-mapping/src/lib.rs index e682fe66..89a44acc 100644 --- a/crates/asap-aware-mapping/src/lib.rs +++ b/crates/asap-aware-mapping/src/lib.rs @@ -184,6 +184,7 @@ pub mod accuracy; pub mod accuracy_reconciliation; pub mod analytical_cost; +pub mod boundary_cost; pub mod cost_model; pub mod empirical_comparison; pub mod empirical_cost; 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 86025910..ad70cdc3 100644 --- a/crates/asap-aware-mapping/src/physical_plan_cost_model.rs +++ b/crates/asap-aware-mapping/src/physical_plan_cost_model.rs @@ -30,6 +30,7 @@ pub struct PhysicalEvidenceSnapshot { pub scope: ComparisonScope, pub cache_profile: CacheProfile, pub storage_io: Option, + pub boundaries: Option, } /// Deployment evidence needed to price one planner alternative. @@ -71,6 +72,10 @@ pub struct PhysicalPlanComparison { crate::storage_io::StorageEstimate, crate::storage_io::StorageEstimate, )>, + pub boundaries: Option<( + crate::boundary_cost::BoundaryEstimate, + crate::boundary_cost::BoundaryEstimate, + )>, } /// Planner cost model that admits only complete, cheaper physical plans. @@ -104,8 +109,7 @@ impl<'a> PhysicalPlanCostModel<'a> { "resource_calibration.version", )); } - // A zero base objective may be supplemented by storage coefficients; - // that check needs the target's immutable evidence snapshot. + // Supplemental storage or boundary pricing may supply a zero-base objective. match calibration.validate() { Ok(()) | Err(AnalyticalCostError::ZeroCalibration) => {} Err(error) => return Err(error), @@ -167,6 +171,15 @@ impl<'a> PhysicalPlanCostModel<'a> { )); } let scope = &snapshot.scope; + if snapshot.boundaries.is_some() + && matches!(snapshot.cache_profile, CacheProfile::Evidence(_)) + { + // Scope-based boundary multiplicity does not describe which actions + // cache hits skip; do not mix pre-cache byte work with discounted CPU. + return Err(AnalyticalCostError::MissingOrStale( + "cache-aware boundary execution evidence", + )); + } let evidence = QueryEvidence { provider: self.provider, snapshot: &snapshot, @@ -218,6 +231,26 @@ impl<'a> PhysicalPlanCostModel<'a> { )) }) .transpose()?; + let boundaries = snapshot + .boundaries + .as_ref() + .map(|profile| { + Ok(( + crate::boundary_cost::estimate_boundaries( + &raw, + scope, + profile, + &snapshot.version, + )?, + crate::boundary_cost::estimate_boundaries( + &replacement, + scope, + profile, + &snapshot.version, + )?, + )) + }) + .transpose()?; let (mut raw_cost, mut candidate_cost) = match self.calibration.validate() { Ok(()) => ( Cost(resources.raw.calibrated_cost(&self.calibration)?), @@ -237,7 +270,13 @@ impl<'a> PhysicalPlanCostModel<'a> { .into_iter() .any(|coefficient| coefficient > 0.0) }); - if !has_storage_objective { + // Boundary estimation above validates evidence and pricing. + // At least one dimension must have a positive coefficient. + let has_boundary_objective = snapshot.boundaries.as_ref().is_some_and(|profile| { + profile.calibration.cost_per_network_byte > 0.0 + || profile.calibration.cost_per_materialization_byte > 0.0 + }); + if !has_storage_objective && !has_boundary_objective { return Err(AnalyticalCostError::ZeroCalibration); } (Cost(0.0), Cost(0.0)) @@ -248,6 +287,10 @@ impl<'a> PhysicalPlanCostModel<'a> { raw_cost.0 += raw.cost; candidate_cost.0 += candidate.cost; } + if let Some((raw, candidate)) = &boundaries { + 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); } @@ -256,6 +299,7 @@ impl<'a> PhysicalPlanCostModel<'a> { raw_cost, candidate_cost, storage_io, + boundaries, }) } } @@ -405,6 +449,7 @@ mod tests { storage_io: Option, summary_available: bool, candidate_scan_bytes: u64, + boundaries: Option, snapshot_calls: Cell, raw_evidence_calls: Cell, } @@ -415,6 +460,7 @@ mod tests { storage_io: None, summary_available, candidate_scan_bytes, + boundaries: None, snapshot_calls: Cell::new(0), raw_evidence_calls: Cell::new(0), } @@ -500,6 +546,7 @@ mod tests { scope: scope(), cache_profile: CacheProfile::no_cache(), storage_io: self.storage_io.clone(), + boundaries: self.boundaries.clone(), }) } @@ -674,6 +721,131 @@ mod tests { assert!(estimate.storage_io.is_none()); } + // Combined objectives must identify the base calibration even when its + // coefficients are zero and boundaries supply the entire objective. + #[test] + fn blank_base_calibration_version_is_rejected() { + let provider = TestProvider::new(true, 800); + for version in ["", " \t\n"] { + for boundary_only in [false, true] { + let mut calibration = calibration(); + calibration.version = version.into(); + if boundary_only { + calibration.cost_per_cpu_op = 0.0; + calibration.cost_per_scan_byte = 0.0; + calibration.cost_per_retained_byte = 0.0; + } + assert!( + PhysicalPlanCostModel::new(&provider, calibration).is_err(), + "blank base provenance accepted (boundary_only={boundary_only})" + ); + } + } + } + + // Explicit byte pricing can rank complete plans without pricing CPU or scans. + #[test] + fn boundary_only_objective_ranks_complete_plans() { + use crate::boundary_cost::*; + let root = query(); + let target = TargetSubDAG::new(&root); + let mut provider = TestProvider::new(true, 800); + let raw = PhysicalPlanCostModel::new(&provider, calibration()) + .unwrap() + .target_evidence(&target) + .unwrap() + .1; + let candidate = provider.summary_dag(&scope()); + provider.boundaries = Some(BoundaryProfile { + evidence_version: "test-snapshot-1".into(), + observed_at_ms: 900, + valid_until_ms: 2000, + calibration: BoundaryCalibration { + version: "network-only-v1".into(), + cost_per_network_byte: 1.0, + cost_per_materialization_byte: 0.0, + }, + plans: [&raw, &candidate] + .into_iter() + .map(|dag| BoundaryPlanEvidence { + root: dag.root.clone(), + nodes: dag + .nodes + .iter() + .map(|node| { + let statistics = dag.evidence[&node.id].statistics.clone(); + let bytes = statistics.output().bytes; + let boundaries = if matches!(node.operator, PhysicalOperator::Scan) { + vec![PhysicalBoundary { + id: "scan-transfer".into(), + consumer: None, + kind: BoundaryKind::Network { + source_location: "edge".into(), + destination_location: "backend".into(), + }, + logical_bytes: bytes, + encoded_bytes: bytes, + copies: 1, + }] + } else { + vec![] + }; + ( + node.id.clone(), + BoundaryNodeEvidence { + node: node.clone(), + statistics, + boundaries, + }, + ) + }) + .collect(), + }) + .collect(), + }); + let zero_base = ResourceCalibration { + cost_per_cpu_op: 0.0, + cost_per_scan_byte: 0.0, + cost_per_retained_byte: 0.0, + version: "boundary-only-v1".into(), + }; + let space = crate::replacement::search_workload_with( + vec![("q", Rc::clone(&root))], + &crate::replacement::default_strategies(), + ); + let model = PhysicalPlanCostModel::new(&provider, zero_base.clone()).unwrap(); + let selected = space.global_selection(&model); + assert!(selected + .for_target(&space.roots[0].1) + .unwrap() + .chosen + .is_some()); + drop(model); + for coefficient in [0.0, -1.0, f64::NAN] { + provider + .boundaries + .as_mut() + .unwrap() + .calibration + .cost_per_network_byte = coefficient; + let model = PhysicalPlanCostModel::new(&provider, zero_base.clone()).unwrap(); + assert!(space + .global_selection(&model) + .for_target(&space.roots[0].1) + .unwrap() + .chosen + .is_none()); + } + provider.boundaries = None; + let model = PhysicalPlanCostModel::new(&provider, zero_base).unwrap(); + assert!(space + .global_selection(&model) + .for_target(&space.roots[0].1) + .unwrap() + .chosen + .is_none()); + } + #[test] fn global_selection_uses_complete_physical_comparison() { let root = query(); @@ -862,6 +1034,7 @@ mod tests { scope: scope(), cache_profile: CacheProfile::no_cache(), storage_io: None, + boundaries: None, }) } diff --git a/crates/asap-aware-mapping/tests/boundary_cost.rs b/crates/asap-aware-mapping/tests/boundary_cost.rs new file mode 100644 index 00000000..3da56535 --- /dev/null +++ b/crates/asap-aware-mapping/tests/boundary_cost.rs @@ -0,0 +1,442 @@ +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::boundary_cost::*; + +// Legacy mapping imports and the shared resource API are the very same Rust types. +#[test] +fn mapping_resource_reexports_are_wire_compatible_shared_types() { + let shared = asap_types::resources::BoundaryResources { + network_bytes: 480, + materialization_bytes: 40, + }; + let legacy: BoundaryResources = shared; + assert_eq!( + serde_json::to_value(legacy).unwrap(), + serde_json::json!({"network_bytes": 480, "materialization_bytes": 40}) + ); + let shared_kind = asap_types::resources::BoundaryKind::Materialization { + medium: asap_types::resources::MaterializationMedium::Disk, + }; + let mut boundary = transfer("persist", None); + boundary.kind = shared_kind; + let json = serde_json::to_value(&boundary).unwrap(); + assert_eq!( + json["kind"], + serde_json::json!({"kind": "materialization", "medium": "disk"}) + ); + assert_eq!( + serde_json::from_value::(json).unwrap(), + boundary + ); +} + +fn profile(dag: &EvidenceBackedPhysicalDag) -> BoundaryProfile { + BoundaryProfile { + evidence_version: "evidence-v1".into(), + observed_at_ms: 90, + valid_until_ms: 200, + calibration: BoundaryCalibration { + version: "bytes-v1".into(), + cost_per_network_byte: 2.0, + cost_per_materialization_byte: 3.0, + }, + plans: vec![BoundaryPlanEvidence { + root: dag.root.clone(), + nodes: dag + .nodes + .iter() + .map(|node| { + ( + node.id.clone(), + BoundaryNodeEvidence { + node: node.clone(), + statistics: dag.evidence[&node.id].statistics.clone(), + boundaries: vec![], + }, + ) + }) + .collect(), + }], + } +} + +fn transfer(id: &str, consumer: Option<&str>) -> PhysicalBoundary { + PhysicalBoundary { + id: id.into(), + consumer: consumer.map(str::to_owned), + kind: BoundaryKind::Network { + source_location: "edge".into(), + destination_location: "backend".into(), + }, + logical_bytes: 80, + encoded_bytes: 40, + copies: 2, + } +} + +fn profiles_for_alternatives(dags: &[&EvidenceBackedPhysicalDag]) -> BoundaryProfile { + let mut combined = profile(dags[0]); + combined.plans.clear(); + for dag in dags { + let mut alternative = profile(dag); + alternative.plans[0] + .nodes + .get_mut("scan") + .unwrap() + .boundaries = vec![transfer(&format!("wire-{}", dag.root), Some(&dag.root))]; + combined.plans.extend(alternative.plans); + } + combined +} + +// Alternative consumers may share a producer identity without sharing transfers. +#[test] +fn shared_producer_boundaries_are_scoped_to_each_alternative() { + let (dag, scope) = fixture(); + let alternative = |root: &str| { + let mut value = dag.clone(); + value.root = root.into(); + value + .nodes + .retain(|node| node.id == "scan" || node.id == root); + value.evidence.retain(|id, _| id == "scan" || id == root); + value + }; + let raw = alternative("left"); + let candidate = alternative("right"); + let profile = profiles_for_alternatives(&[&raw, &candidate]); + for plan in [&raw, &candidate] { + let estimate = estimate_boundaries(plan, &scope, &profile, "evidence-v1").unwrap(); + assert_eq!(estimate.total.network_bytes, 240); + assert_eq!(estimate.per_boundary.len(), 1); + assert!(estimate + .per_boundary + .contains_key(&format!("wire-{}", plan.root))); + } +} + +// A boundary binding must identify exactly one complete physical alternative. +#[test] +fn missing_ambiguous_and_mismatched_plan_bindings_are_rejected() { + let (dag, scope) = fixture(); + for case in 0..6 { + let mut profile = profile(&dag); + match case { + 0 => profile.plans.clear(), + 1 => profile.plans.push(profile.plans[0].clone()), + 2 => profile.plans[0].root = "left".into(), + 3 => { + profile.plans[0].nodes.remove("left"); + } + 4 => { + profile.plans[0] + .nodes + .get_mut("scan") + .unwrap() + .node + .execution = ExecutionMultiplicity::Once + } + 5 => { + let OperatorStatistics::Scan { + source_read_bytes, .. + } = &mut profile.plans[0].nodes.get_mut("scan").unwrap().statistics + else { + unreachable!() + }; + *source_read_bytes += 1; + } + _ => unreachable!(), + } + assert!( + estimate_boundaries(&dag, &scope, &profile, "evidence-v1").is_err(), + "case {case}" + ); + } +} + +// Ordinary in-memory edges contribute no traffic; shared transfers count once. +#[test] +fn memory_edges_are_free_and_shared_transfer_is_counted_once() { + let (dag, scope) = fixture(); + let mut profile = profile(&dag); + assert_eq!( + estimate_boundaries(&dag, &scope, &profile, "evidence-v1") + .unwrap() + .total, + BoundaryResources::default() + ); + profile.plans[0] + .nodes + .get_mut("scan") + .unwrap() + .boundaries + .push(transfer("shared", None)); + let estimate = estimate_boundaries(&dag, &scope, &profile, "evidence-v1").unwrap(); + assert_eq!(estimate.total.network_bytes, 240); // 40 encoded bytes × 2 replicas × 3 evaluations + assert_eq!(estimate.total.materialization_bytes, 0); + assert_eq!(estimate.cost, 480.0); + assert_eq!(estimate.per_node["scan"].network_bytes, 240); + assert_eq!(estimate.per_boundary["shared"].network_bytes, 240); +} + +// A retained producer materializes once and transfers separately to each reader. +#[test] +fn materialization_once_and_transfers_per_consumer_have_distinct_multiplicity() { + let (mut dag, scope) = fixture(); + dag.nodes[0].execution = ExecutionMultiplicity::Once; + dag.nodes[0].retained_bytes = 80; + let mut profile = profile(&dag); + let mut materialize = transfer("persist", None); + materialize.kind = BoundaryKind::Materialization { + medium: MaterializationMedium::Disk, + }; + materialize.copies = 1; + profile.plans[0].nodes.get_mut("scan").unwrap().boundaries = vec![ + materialize, + transfer("left-wire", Some("left")), + transfer("right-wire", Some("right")), + ]; + let estimate = estimate_boundaries(&dag, &scope, &profile, "evidence-v1").unwrap(); + assert_eq!(estimate.total.materialization_bytes, 40); + assert_eq!(estimate.total.network_bytes, 480); + assert_eq!(estimate.cost, 1080.0); +} + +// Invalid endpoint, byte evidence, duplicate identity, and overflow fail closed. +#[test] +fn incompatible_boundary_evidence_is_rejected() { + let (dag, scope) = fixture(); + for case in 0..7 { + let mut profile = profile(&dag); + let mut boundary = transfer("wire", Some("left")); + match case { + 0 => boundary.consumer = Some("root".into()), + 1 => boundary.logical_bytes = 79, + 2 => boundary.copies = 0, + 3 => boundary.encoded_bytes = 0, + 4 => { + boundary.kind = BoundaryKind::Network { + source_location: "same".into(), + destination_location: "same".into(), + } + } + 5 => boundary.copies = u64::MAX, + 6 => profile.plans[0] + .nodes + .get_mut("scan") + .unwrap() + .boundaries + .push(boundary.clone()), + _ => unreachable!(), + } + profile.plans[0] + .nodes + .get_mut("scan") + .unwrap() + .boundaries + .push(boundary); + assert!( + estimate_boundaries(&dag, &scope, &profile, "evidence-v1").is_err(), + "case {case}" + ); + } +} + +// Missing evidence and non-finite coefficients cannot be treated as zero. +#[test] +fn missing_stale_and_non_finite_evidence_is_rejected() { + let (dag, scope) = fixture(); + let mut evidence = profile(&dag); + evidence.plans[0].nodes.remove("left"); + assert!(estimate_boundaries(&dag, &scope, &evidence, "evidence-v1").is_err()); + evidence = profile(&dag); + evidence.valid_until_ms = 100; + assert!(estimate_boundaries(&dag, &scope, &evidence, "evidence-v1").is_err()); + evidence = profile(&dag); + evidence.calibration.cost_per_network_byte = f64::INFINITY; + assert!(estimate_boundaries(&dag, &scope, &evidence, "evidence-v1").is_err()); + assert!(estimate_boundaries(&dag, &scope, &profile(&dag), "different").is_err()); +} + +// Individual actions may fit while accumulation across actions or nodes overflows. +#[test] +fn boundary_accumulation_and_calibrated_cost_overflow_are_rejected() { + use asap_aware_mapping::analytical_cost::AnalyticalCostError; + let (mut dag, mut scope) = fixture(); + scope.recurrence = QueryRecurrence::OneTime { + invocations: 1, + execute_at: None, + }; + for node in &mut dag.nodes { + node.execution = ExecutionMultiplicity::Once; + } + for different_nodes in [false, true] { + let mut evidence = profile(&dag); + for (id, node) in [ + ("first", "scan"), + ("second", if different_nodes { "left" } else { "scan" }), + ] { + let mut boundary = transfer(id, None); + boundary.copies = 1; + boundary.encoded_bytes = u64::MAX / 2 + 1; + evidence.plans[0] + .nodes + .get_mut(node) + .unwrap() + .boundaries + .push(boundary); + } + assert!(matches!( + estimate_boundaries(&dag, &scope, &evidence, "evidence-v1"), + Err(AnalyticalCostError::Overflow) + )); + } + let mut evidence = profile(&dag); + evidence.plans[0] + .nodes + .get_mut("scan") + .unwrap() + .boundaries + .push(transfer("wire", None)); + evidence.calibration.cost_per_network_byte = f64::MAX; + assert!(matches!( + estimate_boundaries(&dag, &scope, &evidence, "evidence-v1"), + Err(AnalyticalCostError::Overflow) + )); +} diff --git a/crates/devtools/src/bin/dag_export.rs b/crates/devtools/src/bin/dag_export.rs index 3eb51a9a..8c95fab6 100644 --- a/crates/devtools/src/bin/dag_export.rs +++ b/crates/devtools/src/bin/dag_export.rs @@ -90,6 +90,8 @@ use asap_types::types::AccuracyTarget; struct PlannerCostDocument { #[serde(default, skip_serializing_if = "Option::is_none")] storage_io: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + boundaries: Option, /// Immutable catalog/runtime evidence generation shared by this file. evidence_version: String, calibration: ResourceCalibration, @@ -279,6 +281,7 @@ fn plan_values_match_inner( struct ExportPhysicalProvider<'a> { storage_io: Option<&'a asap_aware_mapping::storage_io::StorageIoProfile>, + boundaries: Option<&'a asap_aware_mapping::boundary_cost::BoundaryProfile>, evidence_version: &'a str, target: &'a TargetPhysicalEvidence, candidate: &'a CandidatePhysicalEvidence, @@ -301,6 +304,7 @@ impl PlannerPhysicalPlanProvider for ExportPhysicalProvider<'_> { scope: self.target.scope.resolve()?, cache_profile: self.target.scope.cache_profile.clone(), storage_io: self.storage_io.cloned(), + boundaries: self.boundaries.cloned(), }) } @@ -390,6 +394,7 @@ impl ExportPlannerCostModel<'_> { Some(( ExportPhysicalProvider { storage_io: self.document.storage_io.as_ref(), + boundaries: self.document.boundaries.as_ref(), evidence_version: &self.document.evidence_version, target: target_evidence, candidate: candidate_evidence, @@ -424,6 +429,12 @@ impl ExportPlannerCostModel<'_> { storage.model_version, storage.calibration_version )); } + if let Some((boundary, _)) = &estimate.boundaries { + version.push_str(&format!( + "+{}+{}", + boundary.model_version, boundary.calibration_version + )); + } let scope = &provider.target.scope; let Ok((result_hits, buffer_hits)) = cache_hit_ratios( &scope.cache_profile, @@ -537,6 +548,40 @@ impl ExportPlannerCostModel<'_> { raw_inputs.extend(storage_inputs(raw)); candidate_inputs.extend(storage_inputs(candidate)); } + let boundary_inputs = + |estimate: &asap_aware_mapping::boundary_cost::BoundaryEstimate| { + let mut terms: Vec<_> = estimate + .total + .terms() + .into_iter() + .map(|(name, value)| CostInput { + name: name.into(), + value: value as f64, + unit: Some("bytes".into()), + }) + .collect(); + for (prefix, entries) in [ + ("physical_node", &estimate.per_node), + ("boundary", &estimate.per_boundary), + ] { + let mut ids: Vec<_> = entries.keys().collect(); + ids.sort(); + for id in ids { + terms.extend(entries[id].terms().into_iter().map(|(name, value)| { + CostInput { + name: format!("{prefix}:{id}:{name}"), + value: value as f64, + unit: Some("bytes".into()), + } + })); + } + } + terms + }; + if let Some((raw, candidate)) = &estimate.boundaries { + raw_inputs.extend(boundary_inputs(raw)); + candidate_inputs.extend(boundary_inputs(candidate)); + } let baseline = CostAnnotation::modeled( estimate.raw_cost.0, CostUnit::CostUnits, @@ -1756,6 +1801,265 @@ mod tests { .is_none()); } + // Declared boundaries survive JSON and affect the selected physical plan. + #[test] + fn boundary_bytes_export_and_change_plan_selection() { + use asap_aware_mapping::boundary_cost::*; + 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 = BoundaryProfile { + evidence_version: document.evidence_version.clone(), + observed_at_ms: 900, + valid_until_ms: 2000, + calibration: BoundaryCalibration { + version: "bytes-v1".into(), + cost_per_network_byte: 1.0, + cost_per_materialization_byte: 1.0, + }, + plans: vec![], + }; + for dag in [&raw, &candidate_dag] { + profile.plans.push(BoundaryPlanEvidence { + root: dag.root.clone(), + nodes: dag + .nodes + .iter() + .map(|node| { + ( + node.id.clone(), + BoundaryNodeEvidence { + node: node.clone(), + statistics: dag.evidence[&node.id].statistics.clone(), + boundaries: vec![], + }, + ) + }) + .collect(), + }); + } + profile.plans[1] + .nodes + .get_mut("summary-read") + .unwrap() + .boundaries + .push(PhysicalBoundary { + id: "summary-transfer".into(), + consumer: None, + kind: BoundaryKind::Network { + source_location: "edge".into(), + destination_location: "backend".into(), + }, + logical_bytes: 2400, + encoded_bytes: 1200, + copies: 1, + }); + document.boundaries = 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 == "network_bytes") + .unwrap() + .value, + 0.0 + ); + assert_eq!( + selected + .inputs + .iter() + .find(|term| term.name == "network_bytes") + .unwrap() + .value, + 12_000.0 + ); + assert!(selected + .inputs + .iter() + .any(|term| term.name == "physical_node:summary-read:network_bytes")); + assert!(selected + .inputs + .iter() + .any(|term| term.name == "boundary:summary-transfer:network_bytes")); + assert!(selected + .model_version + .as_ref() + .unwrap() + .contains(BOUNDARY_MODEL_VERSION)); + assert_eq!( + selected.evidence_version.as_deref(), + Some("test-evidence-v1") + ); + let mut boundary_only = parsed.clone(); + boundary_only.calibration.cost_per_cpu_op = 0.0; + boundary_only.calibration.cost_per_scan_byte = 0.0; + boundary_only.calibration.cost_per_retained_byte = 0.0; + let (baseline, selected, _) = ExportPlannerCostModel { + document: &boundary_only, + } + .annotations(&candidate, &root); + assert_eq!(baseline.value, Some(0.0)); + assert_eq!(selected.value, Some(12_000.0)); + + // Independent supplemental objectives add once, and either one can + // price a zero-base plan while retaining both sets of export evidence. + { + use asap_aware_mapping::storage_io::*; + let mut joint = boundary_only.clone(); + let mut storage = StorageIoProfile { + evidence_version: joint.evidence_version.clone(), + observed_at_ms: 900, + valid_until_ms: 2000, + calibration: StorageCalibration { + version: "joint-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![], + }; + storage.nodes.insert( + node.id.clone(), + StorageNodeEvidence { + node: node.clone(), + statistics, + accesses, + }, + ); + } + } + joint.storage_io = Some(storage); + let joint = + parse_planner_cost_document(&serde_json::to_string(&joint).unwrap()).unwrap(); + let (raw_cost, candidate_cost, _) = + ExportPlannerCostModel { document: &joint }.annotations(&candidate, &root); + assert_eq!(raw_cost.value, Some(160.0)); + assert_eq!(candidate_cost.value, Some(12_010.0)); + for annotation in [&raw_cost, &candidate_cost] { + let version = annotation.model_version.as_ref().unwrap(); + assert!(version.contains(STORAGE_IO_MODEL_VERSION)); + assert!(version.contains(BOUNDARY_MODEL_VERSION)); + assert!(version.contains("joint-requests-v1")); + assert_eq!(annotation.cache_profile.as_deref(), Some("no-cache-v1")); + assert!(annotation + .inputs + .iter() + .any(|term| term.name == "object_get_operations")); + assert!(annotation + .inputs + .iter() + .any(|term| term.name == "network_bytes")); + } + let mut storage_only = joint.clone(); + storage_only.boundaries = None; + assert_eq!( + ExportPlannerCostModel { + document: &storage_only + } + .candidate_cost(&candidate, &target), + Some(Cost(10.0)) + ); + let mut neither = storage_only; + neither.storage_io = None; + assert!(ExportPlannerCostModel { document: &neither } + .annotations(&candidate, &root) + .0 + .value + .is_none()); + } + + // Explicit boundaries have no post-cache execution-count evidence. + // A nontrivial cache profile must fail closed instead of combining + // discounted CPU with pre-cache transfer multiplicity. + let mut combined = serde_json::to_value(&parsed).unwrap(); + combined["targets"][0]["scope"]["cache_profile"] = serde_json::json!({ + "profile": "evidence", "version": "combined-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 combined = parse_planner_cost_document(&combined.to_string()).unwrap(); + let combined_model = ExportPlannerCostModel { + document: &combined, + }; + assert!(combined_model.candidate_cost(&candidate, &target).is_none()); + assert!(combined_model + .annotations(&candidate, &root) + .0 + .value + .is_none()); + let (uncached_raw, uncached_selected, _) = + ExportPlannerCostModel { document: &parsed }.annotations(&candidate, &root); + for annotation in [&uncached_raw, &uncached_selected] { + assert_eq!(annotation.cache_profile.as_deref(), Some("no-cache-v1")); + assert_eq!( + annotation.evidence_version.as_deref(), + Some("test-evidence-v1") + ); + assert!(annotation + .model_version + .as_ref() + .unwrap() + .contains(BOUNDARY_MODEL_VERSION)); + } + + let mut ambiguous = parsed.clone(); + let plans = &mut ambiguous.boundaries.as_mut().unwrap().plans; + plans.push(plans[0].clone()); + let model = ExportPlannerCostModel { + document: &ambiguous, + }; + assert!(model.candidate_cost(&candidate, &target).is_none()); + assert!(model.annotations(&candidate, &root).0.value.is_none()); + document + .boundaries + .as_mut() + .unwrap() + .calibration + .cost_per_network_byte = 1000.0; + assert!(ExportPlannerCostModel { + document: &document + } + .candidate_cost(&candidate, &target) + .is_none()); + document.boundaries.as_mut().unwrap().plans[0] + .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, @@ -1908,6 +2212,7 @@ mod tests { }; let document = PlannerCostDocument { storage_io: None, + boundaries: None, evidence_version: "test-evidence-v1".into(), calibration: ResourceCalibration { cost_per_cpu_op: 1.0, @@ -2221,6 +2526,7 @@ mod tests { let second_dag = cheap_candidate_dag(); let document = PlannerCostDocument { storage_io: None, + boundaries: None, evidence_version: "test-evidence-v1".into(), calibration: ResourceCalibration { cost_per_cpu_op: 1.0, diff --git a/crates/types/src/resources.rs b/crates/types/src/resources.rs index d2107a7f..67734816 100644 --- a/crates/types/src/resources.rs +++ b/crates/types/src/resources.rs @@ -6,18 +6,19 @@ //! Cache assumptions share this schema namespace but are not additive resource //! consumption; their numerical interpretation belongs to an estimator. +pub mod boundary; pub mod cache; pub mod cpu; pub mod measurement; pub mod physical; pub mod storage; +pub use boundary::{BoundaryKind, BoundaryResources, MaterializationMedium}; pub use cache::{CacheCapacityEvidence, CacheEvidence, CacheProfile}; pub use storage::StorageResources; pub use cpu::{MeasuredCpu, ModeledCpu}; pub use measurement::Measurement; pub use physical::{MeasuredResources, PhysicalResources}; - #[cfg(test)] mod tests { use super::*; diff --git a/crates/types/src/resources/boundary.rs b/crates/types/src/resources/boundary.rs new file mode 100644 index 00000000..e1972af5 --- /dev/null +++ b/crates/types/src/resources/boundary.rs @@ -0,0 +1,137 @@ +//! Physical transfer and materialization dimensions, independent of planner policy. +//! +//! A boundary is either a network transfer between execution locations or a +//! materialization of an intermediate result in memory, on disk, or in an +//! object store. This module defines their kinds, materialization media, and +//! separate byte-work counters; it is not limited to network resources. +//! +//! Counters describe bytes transferred or materialized, not retained memory, +//! allocated disk space, or storage request counts. Estimation, evidence +//! validation, and calibration belong to the cost model, not this schema. + +use serde::{Deserialize, Serialize}; + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] +pub enum BoundaryKind { + /// Transfer encoded data between execution locations. + Network { + source_location: String, + destination_location: String, + }, + /// Write an intermediate result to the specified medium for later use. + Materialization { medium: MaterializationMedium }, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum MaterializationMedium { + Memory, + Disk, + ObjectStore, +} + +/// Byte work at explicitly declared physical actions, not storage occupancy. +/// Network traffic and materialization writes remain separate dimensions. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct BoundaryResources { + /// Bytes transferred across explicitly declared network boundaries. + pub network_bytes: u64, + /// Bytes written across materialization boundaries, regardless of medium. + pub materialization_bytes: u64, +} + +impl BoundaryResources { + pub fn terms(self) -> [(&'static str, u64); 2] { + [ + ("network_bytes", self.network_bytes), + ("materialization_bytes", self.materialization_bytes), + ] + } + + /// Add independent dimensions without wrapping or partially mutating either input. + pub fn checked_add(self, other: Self) -> Option { + Some(Self { + network_bytes: self.network_bytes.checked_add(other.network_bytes)?, + materialization_bytes: self + .materialization_bytes + .checked_add(other.materialization_bytes)?, + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// The shared vocabulary retains the original flat counters and tagged kind JSON. + #[test] + fn boundary_json_format_is_unchanged() { + let bytes = BoundaryResources { + network_bytes: 480, + materialization_bytes: 40, + }; + let json = serde_json::json!({"network_bytes": 480, "materialization_bytes": 40}); + assert_eq!(serde_json::to_value(bytes).unwrap(), json); + assert_eq!( + serde_json::from_value::(json).unwrap(), + bytes + ); + let network = BoundaryKind::Network { + source_location: "edge".into(), + destination_location: "backend".into(), + }; + assert_eq!( + serde_json::to_value(network).unwrap(), + serde_json::json!({ + "kind": "network", "source_location": "edge", "destination_location": "backend" + }) + ); + for (medium, name) in [ + (MaterializationMedium::Memory, "memory"), + (MaterializationMedium::Disk, "disk"), + (MaterializationMedium::ObjectStore, "object_store"), + ] { + let kind = BoundaryKind::Materialization { medium }; + let json = serde_json::json!({"kind": "materialization", "medium": name}); + assert_eq!(serde_json::to_value(&kind).unwrap(), json); + assert_eq!(serde_json::from_value::(json).unwrap(), kind); + } + } + + /// Either dimension overflowing returns None without changing the original counters. + #[test] + fn checked_add_preserves_dimensions_and_rejects_overflow() { + let first = BoundaryResources { + network_bytes: 2, + materialization_bytes: 3, + }; + let second = BoundaryResources { + network_bytes: 5, + materialization_bytes: 7, + }; + assert_eq!( + first.checked_add(second), + Some(BoundaryResources { + network_bytes: 7, + materialization_bytes: 10, + }) + ); + for overflowing in [ + BoundaryResources { + network_bytes: u64::MAX, + materialization_bytes: 0, + }, + BoundaryResources { + network_bytes: 0, + materialization_bytes: u64::MAX, + }, + ] { + assert!(first.checked_add(overflowing).is_none()); + } + assert_eq!( + first.terms(), + [("network_bytes", 2), ("materialization_bytes", 3)] + ); + } +} 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 cc444a7a..a618a957 100644 --- a/docs/design_docs/asap-aware-mapping/analytical-resource-cost.md +++ b/docs/design_docs/asap-aware-mapping/analytical-resource-cost.md @@ -1174,6 +1174,11 @@ request counts using the optional storage profile. See [storage operation estimates](../../developer_docs/storage-operation-costs.md) for units, formulas, evidence requirements, and calibration. +The physical-plan adapter supports explicit network and materialization +boundaries through optional deployment evidence. See +[physical boundary estimates](../../developer_docs/physical-boundary-costs.md) +for kinds, execution multiplicity, evidence, and calibration. + An estimate is unavailable when required evidence, a physical formula, or a finite horizon is missing. Structural node counts are never substituted. @@ -1181,7 +1186,7 @@ The current scalar includes CPU, retained memory, and source/disk reads. A complete deployment model may additionally require: - source and spill writes; -- network transfer; +- latency and contention at network boundaries; - parallelism and contention; - allocator fragmentation; - wall-clock critical-path latency; diff --git a/docs/developer_docs/physical-boundary-costs.md b/docs/developer_docs/physical-boundary-costs.md new file mode 100644 index 00000000..4b0d2965 --- /dev/null +++ b/docs/developer_docs/physical-boundary-costs.md @@ -0,0 +1,99 @@ +# Physical boundary byte estimates + +`asap_types::resources` owns the canonical `BoundaryResources`, `BoundaryKind`, +and `MaterializationMedium` definitions in `resources/boundary.rs`. The mapping +crate re-exports those same types from `boundary_cost` for import compatibility; +all estimator and export consumers therefore use shared definitions, not copies. +Their existing JSON format is unchanged. The shared byte counters support checked +addition without depending on planner errors. Snapshot binding, validation, +calibration, and ranking remain in the mapping crate. Boundary traffic/write work +is distinct from CPU work, scanned bytes, and stored byte occupancy; it is not +collapsed into the generic CPU/byte resource container. + +The physical-plan adapter accepts an optional `BoundaryProfile` in the +immutable `PhysicalEvidenceSnapshot`. `dag_export --planner-cost-json` accepts +the same profile in a top-level `boundaries` field. With no profile, these +dimensions remain unestimated and the existing resource objective is preserved. + +The profile's `plans` list binds boundaries to complete physical alternatives. +Each plan supplies its `root` and a `nodes` map containing every physical node, +its authoritative statistics, and an explicit list of boundary actions on its +output. An empty boundary list declares ordinary in-memory dataflow with no +boundary traffic. Exactly one plan must match the root, complete node set, and +node/statistics snapshot; missing or ambiguous matches fail closed. This lets +alternatives reuse a producer identity while declaring different transfers to +their respective consumers. All alternatives share the profile's immutable +evidence generation and calibration. Evidence +must match the immutable snapshot version and be current at planning time: +`observed_at_ms <= planning_time < valid_until_ms`. + +Supported boundaries are: + +| Boundary | Required evidence | Dimension | +|---|---|---| +| Network/exchange/deployment transfer | Distinct nonempty source and destination locations | Network bytes | +| Materialization/persistence | Memory, disk, or object-store medium | Materialization bytes | + +Every boundary also declares its unique physical ID, output logical bytes, +encoded bytes per execution, and positive copy count. Logical bytes must equal +the producer's output statistic. Encoded bytes capture an explicit compression +or serialization estimate; empty and nonempty payloads must agree with the +logical edge. A remote persistence operation may declare both a network action +and a materialization action with distinct IDs; these contribute to different +dimensions and require separate calibration coefficients. + +```text +boundary_bytes = encoded_bytes * copies * executions +``` + +For a shared boundary (`consumer: null`), execution multiplicity comes from +the producer. For a per-consumer boundary, `consumer` must identify an actual, +reachable parent of the producer; multiplicity comes from that consumer. +`Once` means one execution; `PerEvaluation` uses the comparison scope's demand. +Shared producers are traversed once regardless of fan-out. Duplicate boundary +IDs fail closed instead of being ambiguously counted or silently dropped. +Boundary profiles currently require `CacheProfile::NoCache`. Cache evidence does +not identify which physical transfers or materializations are skipped on a hit, +and boundary execution counts derive from the comparison scope rather than a +post-cache schedule. Combining a boundary profile with `CacheProfile::Evidence` +therefore fails closed until cache-aware boundary execution evidence exists. +Supported exports retain the no-cache profile provenance alongside boundary +model/calibration versions. + +Example: a retained producer materializes 40 encoded bytes once. Two consumers +each receive two copies of those 40 bytes over three evaluations. The totals +are 40 materialization bytes and 480 network bytes. Logical in-memory edges +without a boundary action add no traffic. A retained producer can therefore +have a once-only persistence action and repeated transfers to its readers. + +`BoundaryEstimate` keeps network and materialization totals separate and returns +per-node and per-boundary terms with model, evidence, and calibration provenance. +A `BoundaryCalibration` supplies finite, nonnegative cost coefficients in the +same cost units as the base resource model, and a nonempty version. Both +alternatives use that profile when ranking. This models byte work, not transfer +latency, bandwidth contention, memory lifetime, or storage requests. +Base CPU, scan, and retained-memory coefficients may all be zero when at least +one boundary coefficient is positive. An absent boundary profile or an entirely +zero objective remains unavailable for ranking. +Both the base and boundary calibration versions must be nonempty, including +when the base coefficients are all zero; combined annotations identify both. + +Annotations expose totals in `bytes`, plus terms named +`physical_node::` and `boundary::`. The existing +viewer cost sidebar displays all terms and the combined formula/calibration +version (`physical-boundary-bytes-v1`) with the evidence version. No network +traffic is inferred from logical edges, operator buffers, or scan bytes. + +Unknown endpoints, mismatched payloads, absent node evidence, duplicate IDs, +stale evidence, invalid coefficients, and integer overflow return typed errors; +ranking/export report the comparison as unavailable. This extends the physical +plan adapter; lifecycle-specific summary-maintenance costing and caching are +separate follow-up integration points. + +Verification: + +```sh +cargo test -p asap-aware-mapping --test boundary_cost +cargo test -p asap-devtools --bin dag_export boundary_bytes_export_and_change_plan_selection +python3 -m unittest discover -s tools/dag-viewer -p 'test_render.py' +``` diff --git a/tools/dag-viewer/test_render.py b/tools/dag-viewer/test_render.py index d5b28f9f..c3463a7a 100644 --- a/tools/dag-viewer/test_render.py +++ b/tools/dag-viewer/test_render.py @@ -56,6 +56,29 @@ def named_graph(name: str, source: str = "SELECT 1") -> dict: class LoadWorkloadTests(unittest.TestCase): + def test_boundary_terms_and_provenance_survive_standalone_export(self): + """The standalone viewer retains byte totals, physical terms, and provenance.""" + graph = named_graph("boundary-example") + annotation = { + "value": 1080.0, + "unit": "CostUnits", + "source": "Modeled", + "model_version": "physical-boundary-bytes-v1+bytes-v1", + "evidence_version": "evidence-v1", + "inputs": [ + {"name": "network_bytes", "value": 480, "unit": "bytes"}, + {"name": "materialization_bytes", "value": 40, "unit": "bytes"}, + {"name": "physical_node:scan:network_bytes", "value": 480, "unit": "bytes"}, + {"name": "boundary:persist:materialization_bytes", "value": 40, "unit": "bytes"}, + ], + } + graph["graph"]["nodes"][0]["selected_cost"] = annotation + html = render({"queries": [graph]}) + for term in annotation["inputs"]: + self.assertIn(term["name"], html) + self.assertIn(annotation["model_version"], html) + self.assertIn(annotation["evidence_version"], html) + def test_loads_summary_maintenance_export_as_a_lifecycle_plan(self): graph = named_graph("unused")["graph"] summary = {