From c8226f54c324faf0ca97d1905bb4e5a39b749a23 Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 07:48:09 -0600 Subject: [PATCH] refactor: use semantic DAG identity for lifecycle plans --- .../src/summary_maintenance_cost/model.rs | 24 +++++++++++--- .../src/summary_maintenance_dag_export.rs | 15 +++++---- .../src/summary_maintenance_lifecycle.rs | 32 +++++++++++-------- 3 files changed, 47 insertions(+), 24 deletions(-) diff --git a/crates/asap-aware-mapping/src/summary_maintenance_cost/model.rs b/crates/asap-aware-mapping/src/summary_maintenance_cost/model.rs index bbed7158..72f7546d 100644 --- a/crates/asap-aware-mapping/src/summary_maintenance_cost/model.rs +++ b/crates/asap-aware-mapping/src/summary_maintenance_cost/model.rs @@ -1282,12 +1282,12 @@ mod tests { .unwrap(); assert_eq!( - plan.selected_physical_plan_id.as_deref(), + plan.selected_window_implementation_id.as_deref(), Some("low-retention-layout") ); assert_eq!( crate::summary_maintenance_dag_export::export_summary_maintenance_plan(&plan) - .selected_physical_plan_id + .selected_window_implementation_id .as_deref(), Some("low-retention-layout") ); @@ -2137,7 +2137,10 @@ mod tests { plan.deployments[0].selected_window_framework, Some(SummaryWindowFramework::ExponentialHistogram) ); - assert_eq!(plan.selected_physical_plan_id.as_deref(), Some("eh-v1")); + assert_eq!( + plan.selected_window_implementation_id.as_deref(), + Some("eh-v1") + ); let guarantee = plan.window_accuracy_guarantee.as_ref().unwrap(); assert_eq!(guarantee.metric, ErrorMetric::RelativeValue); assert!((guarantee.bound.evaluate().unwrap() - 0.05).abs() < f64::EPSILON); @@ -2147,7 +2150,10 @@ mod tests { exported.deployments[0].selected_window_framework, Some(SummaryWindowFramework::ExponentialHistogram) ); - assert_eq!(exported.selected_physical_plan_id.as_deref(), Some("eh-v1")); + assert_eq!( + exported.selected_window_implementation_id.as_deref(), + Some("eh-v1") + ); assert_eq!( exported.window_accuracy_guarantee.unwrap().metric, ErrorMetric::RelativeValue @@ -2519,7 +2525,15 @@ mod tests { .unwrap(); assert!(state_plan.summary_total_cost.is_some()); - let child = summary_with_operations(true, false, false); + let child_readout = summary_with_operations(true, false, false); + let SummaryExpr::SummaryEstimate { + summary_input: child, + .. + } = &child_readout.expr + else { + unreachable!(); + }; + let child = Rc::clone(child); let nested = Rc::new(SummaryNode { expr: SummaryExpr::SummaryAgg { child, diff --git a/crates/asap-aware-mapping/src/summary_maintenance_dag_export.rs b/crates/asap-aware-mapping/src/summary_maintenance_dag_export.rs index 266fb02f..55f1ddf8 100644 --- a/crates/asap-aware-mapping/src/summary_maintenance_dag_export.rs +++ b/crates/asap-aware-mapping/src/summary_maintenance_dag_export.rs @@ -12,7 +12,7 @@ use serde::Serialize; use asap_types::dag_export::{self, SummaryDagGraph}; use asap_types::post_asap::{ - ResultGuarantee, SummaryExpr, SummaryMaintenanceLifecycle, + PostAsapNodeId, ResultGuarantee, SummaryExpr, SummaryMaintenanceLifecycle, SummaryMaintenanceLifecycleGuarantee, SummaryNode, SummaryWindowFramework, }; @@ -30,7 +30,10 @@ pub struct SummaryMaintenanceDagExport { pub expected_reads: Option, pub selected_raw_recompute: bool, #[serde(skip_serializing_if = "Option::is_none")] - pub selected_physical_plan_id: Option, + /// Provider implementation key. The legacy JSON field name is retained + /// until the surrounding export receives its own schema-version bump. + #[serde(rename = "selected_physical_plan_id")] + pub selected_window_implementation_id: Option, pub summary_total_cost: Option, #[serde(skip_serializing_if = "Option::is_none")] pub window_accuracy_guarantee: Option, @@ -39,7 +42,7 @@ pub struct SummaryMaintenanceDagExport { #[derive(Debug, Clone, Serialize)] pub struct SummaryMaintenanceDeploymentExport { - pub summary_index: usize, + pub post_asap_node_id: PostAsapNodeId, #[serde(skip_serializing_if = "Option::is_none")] pub selected_window_framework: Option, #[serde(skip_serializing_if = "Option::is_none")] @@ -65,7 +68,7 @@ pub fn export_summary_maintenance_plan( .deployments .iter() .map(|deployment| SummaryMaintenanceDeploymentExport { - summary_index: deployment.summary_index, + post_asap_node_id: deployment.post_asap_node_id, selected_window_framework: deployment.selected_window_framework.clone(), selected: deployment .summary_maintenance_lifecycle_guarantee @@ -106,7 +109,7 @@ pub fn export_summary_maintenance_plan( update_rate_per_second: plan.update_rate.map(|rate| rate.0), expected_reads: plan.expected_reads, selected_raw_recompute: plan.selected_raw_recompute, - selected_physical_plan_id: plan.selected_physical_plan_id.clone(), + selected_window_implementation_id: plan.selected_window_implementation_id.clone(), summary_total_cost: plan.summary_total_cost.map(|cost| cost.0), window_accuracy_guarantee: plan.window_accuracy_guarantee.clone(), raw_recompute_total_cost: plan.raw_recompute_total_cost.map(|cost| cost.0), @@ -116,7 +119,7 @@ pub fn export_summary_maintenance_plan( /// Walk in the same post-order as `dag_export::export_summary` and attach a /// deployment directly to every flattened occurrence of its `SummaryAgg`. /// This makes the decision visible to graph consumers without asking them to -/// reconstruct pointer identity from `summary_index` or graph position. +/// reconstruct pointer identity from graph position. fn annotate_lifecycle_deployments( node: &SummaryNode, graph: &mut SummaryDagGraph, diff --git a/crates/asap-aware-mapping/src/summary_maintenance_lifecycle.rs b/crates/asap-aware-mapping/src/summary_maintenance_lifecycle.rs index 91f21eda..464ed13d 100644 --- a/crates/asap-aware-mapping/src/summary_maintenance_lifecycle.rs +++ b/crates/asap-aware-mapping/src/summary_maintenance_lifecycle.rs @@ -20,7 +20,8 @@ use std::collections::{HashMap, HashSet}; use std::rc::Rc; use asap_types::post_asap::{ - EvaluationSchedule, OutputRepresentation, ResultGuarantee, SummaryExpr, + compile_executable_dag_with_node_ids, EvaluationSchedule, ExecutionDataStateError, + OutputRepresentation, PostAsapNodeId, ResultGuarantee, SummaryExpr, SummaryMaintenanceLifecycle, SummaryMaintenanceLifecycleGuarantee, SummaryMaintenanceMode, SummaryNode, SummaryWindowFramework, }; @@ -160,9 +161,10 @@ impl SummaryMaintenanceLifecycleAlternative { /// One unique summary-state deployment. Shared `Rc` nodes are emitted once. #[derive(Debug, Clone)] pub struct SummaryMaintenanceDeployment { - /// Traversal-local ordinal used to associate this deployment with exports; - /// it is not a persistent identity across independently planned DAGs. - pub summary_index: usize, + /// Identity of this summary in the exported post-ASAP semantic DAG. + /// It is scoped to one plan version and is not a summary definition or + /// summary instance identity. + pub post_asap_node_id: PostAsapNodeId, /// The unique materialized `SummaryAgg` represented by this deployment. pub summary: Rc, /// Lifecycle, evaluation, and representation commitment selected for this @@ -198,7 +200,7 @@ pub struct SummaryMaintenanceLifecyclePlan { pub selected_raw_recompute: bool, /// Provider-owned identity of the selected complete physical deployment /// (for example a tumbling, sliding, or exponential-histogram plan). - pub selected_physical_plan_id: Option, + pub selected_window_implementation_id: Option, /// Cost of the selected set of summary deployments, when fully known. pub summary_total_cost: Option, /// Composed accuracy guarantee supplied by the selected physical window @@ -244,6 +246,8 @@ pub enum SummaryMaintenanceLifecyclePlanError { EmptyWorkloadDemand, #[error("workload entry index {index} appears more than once in one demand binding")] DuplicateWorkloadEntry { index: usize }, + #[error(transparent)] + InvalidPostAsapDag(#[from] ExecutionDataStateError), } #[derive(Debug, thiserror::Error)] @@ -358,11 +362,11 @@ fn plan_summary_maintenance_lifecycles_with_profile( } let mut summaries = Vec::new(); collect_summary_aggs(&root, &mut HashSet::new(), &mut summaries); + let node_ids = compile_executable_dag_with_node_ids(&root)?.node_ids; let components = summary_state_components(&summaries); let mut deployments: Vec = summaries .into_iter() - .enumerate() - .map(|(summary_index, summary)| { + .map(|summary| { let alternatives = alternatives_for( &facts, horizon, @@ -371,7 +375,9 @@ fn plan_summary_maintenance_lifecycles_with_profile( cost_model.summary_maintenance_lifecycle_cost_inputs_for_horizon(&summary, horizon), ); SummaryMaintenanceDeployment { - summary_index, + post_asap_node_id: node_ids + .node_id(&summary) + .expect("collected summary belongs to the compiled DAG"), summary, summary_maintenance_lifecycle_guarantee: None, selected_window_framework: None, @@ -391,7 +397,7 @@ fn plan_summary_maintenance_lifecycles_with_profile( &facts.required_accuracy, ); let summary_total_cost = complete_estimate.as_ref().map(|estimate| estimate.cost); - let selected_physical_plan_id = complete_estimate + let selected_window_implementation_id = complete_estimate .as_ref() .and_then(|estimate| estimate.physical_plan_id.clone()); let window_accuracy_guarantee = complete_estimate @@ -406,7 +412,7 @@ fn plan_summary_maintenance_lifecycles_with_profile( update_rate: facts.update_rate, expected_reads: facts.reads, selected_raw_recompute, - selected_physical_plan_id, + selected_window_implementation_id, summary_total_cost, window_accuracy_guarantee, raw_recompute_total_cost: None, @@ -508,7 +514,7 @@ pub fn materialize_with_summary_maintenance_lifecycles( plan.root = crate::replacement::keep_pre_asap(target)?; plan.deployments.clear(); plan.selected_raw_recompute = true; - plan.selected_physical_plan_id = None; + plan.selected_window_implementation_id = None; plan.summary_total_cost = None; plan.window_accuracy_guarantee = None; } @@ -2073,7 +2079,7 @@ mod tests { fn whole_candidate_cost_is_evaluated_before_selecting_a_lifecycle() { let root = summary(); let mut deployments = vec![SummaryMaintenanceDeployment { - summary_index: 0, + post_asap_node_id: PostAsapNodeId(0), summary: Rc::clone(&root), summary_maintenance_lifecycle_guarantee: None, selected_window_framework: None, @@ -2136,7 +2142,7 @@ mod tests { ]; let mut deployments: Vec<_> = (0..13) .map(|summary_index| SummaryMaintenanceDeployment { - summary_index, + post_asap_node_id: PostAsapNodeId(summary_index as u32), summary: Rc::clone(&root), summary_maintenance_lifecycle_guarantee: None, selected_window_framework: None,