From af195f1c6937b9bb67c789f268e561d2600b58ea Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Mon, 19 Oct 2026 02:07:32 +0000 Subject: [PATCH] refactor(planner)!: summary identity and dedup move to the runtime's SDS The Planner says what the plan needs: which nodes are stored, when each is built, its pane, and how long that node's output must be kept. Remove StoredOutput.computation_id (the SHA-256 that mirrored the runtime's definition id) and the step that gave every output of one computation the largest retention; each StoredOutput.retention_ms is now its own node's requirement. Drop the sha2 dependency. The stand-in runtime test dedupes panes with its own cache key and keeps a shared pane for the largest retention among the outputs it serves. --- Cargo.lock | 1 - .../tests/runtime_handoff.rs | 51 ++++++--- crates/planner/Cargo.toml | 1 - crates/planner/src/installable.rs | 86 +++------------ crates/planner/tests/runtime_install.rs | 100 +++++++++++++----- 5 files changed, 121 insertions(+), 118 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index f98e21cd..5953ce73 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -463,7 +463,6 @@ dependencies = [ "futures", "serde", "serde_json", - "sha2", "thiserror 2.0.18", "tokio", ] diff --git a/crates/integration-tests/tests/runtime_handoff.rs b/crates/integration-tests/tests/runtime_handoff.rs index fcba018a..ade8c9ab 100644 --- a/crates/integration-tests/tests/runtime_handoff.rs +++ b/crates/integration-tests/tests/runtime_handoff.rs @@ -2,7 +2,7 @@ //! checked against the Planner's internals; the runtime's own end-to-end //! run, with public APIs only, is `crates/planner/tests/runtime_install.rs`. -use std::collections::BTreeMap; +use std::collections::{BTreeMap, BTreeSet}; use std::sync::Arc; use asap_executor::physical_planner::{compile, frontier_from_timing, InputContract}; @@ -11,7 +11,9 @@ use asap_planner::{ e2e_plan, plan_installable, FrontendInput, InstallablePlan, InstallablePlanError, PlanError, PlanOutput, PlanningModels, UserInput, }; -use asap_types::ir::physical_export::{compile_physical_asap_workload, PhysicalASAPDAG}; +use asap_types::ir::physical_export::{ + compile_physical_asap_workload, PhysicalASAPDAG, PhysicalASAPNodeId, +}; use asap_types::ir::properties::ExecutionTiming; use asap_types::ir::{ASAPOp, NonASAPOp, Operator}; use asap_types::physical::Materialization; @@ -365,8 +367,9 @@ async fn stored_panes_carry_stage3_pane_retention() { /// A chain over an offset input keeps its panes for its window plus the /// offset: `[60m] offset 10m` reaches 70 minutes back (4,200,000 ms), the -/// unshifted query's chain 60 minutes. Their panes are one computation, so -/// each pane is kept for the longer of the two on every output. +/// unshifted query's chain 60 minutes (3,600,000 ms). Each stored output +/// carries its own node's requirement; a runtime that stores the twins' +/// identical panes once keeps them for the larger one. #[tokio::test] async fn an_offset_chain_keeps_its_panes_for_its_window_plus_the_offset() { let plan = installable(&repeating(&[ @@ -374,26 +377,40 @@ async fn an_offset_chain_keeps_its_panes_for_its_window_plus_the_offset() { "quantile_over_time(0.99, latency_ms[60m] offset 10m)", ])) .await; - // The twins' panes share one computation, so every output of it - // carries the longest retention: the offset chain's 70 minutes. - let computations: std::collections::BTreeSet<_> = plan - .stored_outputs - .iter() - .map(|o| &o.computation_id) - .collect(); - assert_eq!(computations.len(), 1); - assert!( - plan.stored_outputs - .iter() - .all(|o| o.retention_ms == 4_200_000), - "{:?}", + assert_eq!(plan.stored_outputs.len(), 12); + // The retention of the stored outputs below each query's root. + let retention_ms_under = |root| { + let below = nodes_below(&plan.dag.dag, root); plan.stored_outputs .iter() + .filter(|o| below.contains(&o.node_id)) .map(|o| o.retention_ms) .collect::>() + }; + assert_eq!( + retention_ms_under(plan.queries[0].root_node_id), + [3_600_000; 6] + ); + assert_eq!( + retention_ms_under(plan.queries[1].root_node_id), + [4_200_000; 6] ); } +/// Every node `root` reads, directly or through other nodes. +fn nodes_below(dag: &PhysicalASAPDAG, root: PhysicalASAPNodeId) -> BTreeSet { + let mut below = BTreeSet::new(); + let mut pending = vec![root]; + while let Some(node) = pending.pop() { + for edge in dag.edges.iter().filter(|edge| edge.consumer == node) { + if below.insert(edge.producer) { + pending.push(edge.producer); + } + } + } + below +} + /// A standalone scalar query (`scalar(sum(latency_ms))`) has no DAG node /// to answer it through the public API, so it is refused with its entry. #[tokio::test] diff --git a/crates/planner/Cargo.toml b/crates/planner/Cargo.toml index 0406ba89..27c95edc 100644 --- a/crates/planner/Cargo.toml +++ b/crates/planner/Cargo.toml @@ -22,7 +22,6 @@ asap-logical-asap-optimization = { path = "../logical-asap-optimization" } asap-executor = { path = "../executor" } serde = { version = "1", features = ["derive"] } serde_json = "1" -sha2 = "0.11" thiserror = "2" [dev-dependencies] diff --git a/crates/planner/src/installable.rs b/crates/planner/src/installable.rs index 152f9099..80217671 100644 --- a/crates/planner/src/installable.rs +++ b/crates/planner/src/installable.rs @@ -31,7 +31,6 @@ use asap_types::ir::{ }; use asap_types::physical::Materialization; use serde::{Deserialize, Serialize}; -use sha2::{Digest, Sha256}; use crate::{e2e_plan, PlanError, UserInput}; @@ -82,22 +81,18 @@ pub struct StoredOutput { /// Which run builds it: an ingestion run over its pane's window, or /// the query run of the evaluation that first reads it. pub built_at: StoredOutputBuild, - /// Hex SHA-256 of what the output computes, independent of node ids - /// and of which pane it reads: stored outputs that differ only in - /// `pane.end_offset_ms` share it, so a runtime builds each pane - /// once; equal across plans of one DAG wire version. - pub computation_id: String, /// The pane of raw samples it is built from. pub pane: RawSampleWindow, - /// How far back from the newest evaluation time the runtime keeps this - /// pane once built (a pane whose window starts earlier is dropped): - /// Stage 3's retention of its chain (`Selection::pane_retention_ms`: - /// its input's offset plus its window plus the longest - /// range-evaluation span of its readers); or, for an output Stage 3 - /// records none for, how far back its pane reaches - /// (`pane.end_offset_ms + pane.width_ms`). Every output with the same - /// `computation_id` carries the largest of these among them, since a - /// runtime keeps their panes as one. + /// How far back from the newest evaluation time this node's output + /// must stay available once built (an output whose pane starts + /// earlier is no longer read): Stage 3's retention of its chain + /// (`Selection::pane_retention_ms`: its input's offset plus its window + /// plus the longest range-evaluation span of its readers); or, for an + /// output Stage 3 records none for, how far back its pane reaches + /// (`pane.end_offset_ms + pane.width_ms`). This is this node's own + /// requirement: a runtime that stores identical outputs of several + /// nodes once must keep them for the largest requirement among those + /// nodes, which is the runtime's choice. pub retention_ms: u64, } @@ -110,10 +105,10 @@ pub enum StoredOutputBuild { /// samples have arrived, and stores the result. IngestionRun, /// Built at query time and kept across evaluations - /// (`MatAtQueryKeptAcrossEvaluations`): when an evaluation needs the - /// pane and the store lacks it, the runtime runs the node in a query - /// run over its pane's samples, stores the result, and binds it like - /// a stored pane. Its `TimeRange` is fed only through this build, so + /// (`MatAtQueryKeptAcrossEvaluations`): when an evaluation needs this + /// output for a pane it has not been built for yet, the runtime runs + /// the node in a query run over its pane's samples, keeps the result, + /// and binds it like an output built at ingestion. Its `TimeRange` is fed only through this build, so /// it is not listed in `query_time_raw_windows`. QueryRun, } @@ -492,7 +487,7 @@ impl<'a> IndexedDag<'a> { } } - /// The pane stored output `node` is built from: follow single inputs + /// The pane of raw samples stored output `node` is built from: follow single inputs /// down to a `TimeRange`, then [`Self::window_at`] it. fn pane_of(&self, node: PhysicalASAPNodeId) -> Result { let without_pane = || InstallablePlanError::StoredOutputWithoutPane { node_id: node }; @@ -518,7 +513,6 @@ fn runtime_inputs_of( retention_ms_by_node: &HashMap, ) -> Result<(Vec, Vec), InstallablePlanError> { let indexed = IndexedDag::new(dag); - let mut computation_hashes = HashMap::new(); let mut stored_outputs = Vec::new(); // Every raw-sample window with its timing, and the ranges of stored // outputs' panes, which are fed through those outputs' builds. @@ -542,10 +536,6 @@ fn runtime_inputs_of( .get(&node.id) .copied() .unwrap_or_else(|| pane.reach_ms()), - computation_id: computation_hash(node.id, &indexed, &mut computation_hashes) - .iter() - .map(|b| format!("{b:02x}")) - .collect(), pane, }); } @@ -557,18 +547,6 @@ fn runtime_inputs_of( raw_windows.push((node.output_state.timing, indexed.window_at(node.id)?)); } } - // Outputs of one computation are one pane store: each keeps its panes - // for the longest retention any of them needs. - let mut longest_retention_ms: HashMap = HashMap::new(); - for output in &stored_outputs { - let longest = longest_retention_ms - .entry(output.computation_id.clone()) - .or_default(); - *longest = (*longest).max(output.retention_ms); - } - for output in &mut stored_outputs { - output.retention_ms = longest_retention_ms[&output.computation_id]; - } let query_time_raw_windows = raw_windows .into_iter() .filter(|(timing, window)| { @@ -580,40 +558,6 @@ fn runtime_inputs_of( Ok((stored_outputs, query_time_raw_windows)) } -/// SHA-256 over a node's payload (input ids zeroed) and output schema, then -/// its inputs' hashes in edge order. An offset `TimeShift` with one input -/// takes its input's hash: the offset says which pane an output reads, not -/// what it computes. -fn computation_hash( - node: PhysicalASAPNodeId, - dag: &IndexedDag, - done: &mut HashMap, -) -> [u8; 32] { - if let Some(hash) = done.get(&node) { - return *hash; - } - let dag_node = dag.nodes[&node]; - let node_inputs = dag.inputs.get(&node).map(Vec::as_slice).unwrap_or_default(); - let hash = match (&dag_node.payload, node_inputs) { - (Operator::NonASAP(NonASAPOp::TimeShift { shift, .. }), [input]) if shift.at.is_none() => { - computation_hash(*input, dag, done) - } - _ => { - let payload = dag_node.payload.map_children(|_| 0); - let mut hasher = Sha256::new(); - hasher.update( - serde_json::to_vec(&(&payload, &dag_node.output_schema)).expect("serializes"), - ); - for &input in node_inputs { - hasher.update(computation_hash(input, dag, done)); - } - hasher.finalize().into() - } - }; - done.insert(node, hash); - hash -} - /// The installable DAG of the timed selected plan, with each expanded /// pane's retention by node id. struct ExpandedWithRetention { diff --git a/crates/planner/tests/runtime_install.rs b/crates/planner/tests/runtime_install.rs index fe248dfd..302dc4d8 100644 --- a/crates/planner/tests/runtime_install.rs +++ b/crates/planner/tests/runtime_install.rs @@ -13,8 +13,8 @@ use asap_executor::values::{Batch, Value}; use asap_executor::{series_labels_of_output_row, RawSampleSelector, Sample}; use asap_planner::{ plan_installable, DeploymentCapabilities, FrontendInput, InstallablePlan, InstallablePlanError, - PhysicalASAPDAG, PlanSelectionSettings, PlanningModels, PlanningWorkload, StoredOutputBuild, - UserInput, + PhysicalASAPDAG, PlanSelectionSettings, PlanningModels, PlanningWorkload, StoredOutput, + StoredOutputBuild, UserInput, }; use futures::{executor::block_on, StreamExt}; @@ -215,24 +215,69 @@ struct DataPlaneRun { /// Per query, the query-time reference's answers. reference_answers_per_query: Vec>, /// How many panes this evaluation built (ingestion or query runs); - /// the others it read came from the store. + /// the others it read came from the cache. panes_built: usize, } -/// One pane the data plane stores: its built state, and how long to keep it. +/// One pane this test caches: its built state, and how long to keep it. struct StoredPane { - /// The built state, bound at each stored output of its computation. + /// The built state, bound at every stored output with its cache key. state: Batch, /// When the pane's window starts, in Unix milliseconds. start_ms: i64, - /// How far back from the newest evaluation the pane is kept. + /// How far back from the newest evaluation the pane is kept: the + /// largest `retention_ms` among the stored outputs it served. retention_ms: u64, } -/// The data plane's pane store, by computation and pane end: stored -/// outputs with the same key are the same state, across queries and -/// evaluations. -type PaneStore = BTreeMap<(String, i64), StoredPane>; +/// This stand-in runtime's pane cache, by the test's own key +/// ([`pane_cache_key`]). The plan carries no summary identity: a real +/// runtime decides which stored outputs are the same state. +type PaneStore = BTreeMap; + +/// The test's own cache key for `output`'s pane ending at `end_ms`: the +/// stored node's and its pane's `Scan`'s operators and output schemas +/// without node ids, the pane's width, and the pane's end. Two pane nodes of one chain (or of +/// an offset twin) that cover the same window get the same key; the pane's +/// offset only says which window it covers at an evaluation. Holds for the +/// plans here, whose stored nodes read their pane's `TimeRange` directly +/// (asserted). +fn pane_cache_key(dag: &PhysicalASAPDAG, output: &StoredOutput, end_ms: i64) -> String { + // The stored node's only input is its pane's TimeRange. + let inputs: Vec<_> = dag + .edges + .iter() + .filter(|edge| edge.consumer == output.node_id) + .map(|edge| edge.producer) + .collect(); + assert_eq!(inputs, [output.pane.time_range_node_id]); + // A node by what it computes rather than by its ids. + let node_of = |id| { + let node = dag.nodes.iter().find(|n| n.id == id).unwrap(); + let mut payload = serde_json::to_value(&node.payload).unwrap(); + drop_child_ids(&mut payload); + serde_json::json!([payload, node.output_schema]) + }; + serde_json::json!({ + "stored_node": node_of(output.node_id), + "pane_width_ms": output.pane.width_ms, + "scan": node_of(output.pane.scan_node_id), + "pane_end_ms": end_ms, + }) + .to_string() +} + +/// `value` without its `child` fields, which hold node ids. +fn drop_child_ids(value: &mut serde_json::Value) { + match value { + serde_json::Value::Object(fields) => { + fields.remove("child"); + fields.values_mut().for_each(drop_child_ids); + } + serde_json::Value::Array(items) => items.iter_mut().for_each(drop_child_ids), + _ => {} + } +} /// The node that answers each query of `plan`, in workload order. fn query_roots(plan: &InstallablePlan) -> Vec { @@ -244,7 +289,7 @@ fn query_roots(plan: &InstallablePlan) -> Vec { /// The data plane at evaluation time `evaluation_ms`: from `plan_json`, /// `samples` and the panes earlier evaluations left in `store`, build each -/// distinct stored pane the store lacks once, answer every query, drop the +/// pane the cache lacks once, answer every query, drop the /// panes past their retention, and answer the queries again with the /// query-time reference. fn run_data_plane( @@ -265,8 +310,9 @@ fn run_data_plane( for output in &installed.stored_outputs { // The pane's window at this evaluation, `(start_ms, end_ms]`. let (start_ms, end_ms) = output.pane.bounds_ms(evaluation_ms); - // The pane's identity: its computation and where it ends. - let pane_key = (output.computation_id.clone(), end_ms); + // The test's own cache key: what the output computes and the + // window its pane covers at this evaluation. + let pane_key = pane_cache_key(dag, output, end_ms); if !store.contains_key(&pane_key) { // Prepared once per window: routes by metric, applies the Scan. let selector = RawSampleSelector::new(dag, &output.pane).unwrap(); @@ -296,7 +342,7 @@ fn run_data_plane( let pane_input = BTreeMap::from([(output.pane.time_range_node_id as u64, rows)]); // The batches that run built. let built = run(dag, pane_input, &[output.node_id as u64], scope).remove(0); - // The pane's state, as the store keeps it, in one batch. + // The pane's state, as the cache keeps it, in one batch. let schema = built[0].schema().clone(); let state_rows = built.iter().flat_map(|b| b.rows().to_vec()).collect(); let state = Batch::try_new(schema, state_rows).unwrap(); @@ -310,7 +356,11 @@ fn run_data_plane( ); panes_built += 1; } - inputs.insert(output.node_id as u64, store[&pane_key].state.clone()); + // One cached pane serves every output with its key, so it is kept + // for the largest retention any of them requires. + let cached = store.get_mut(&pane_key).unwrap(); + cached.retention_ms = cached.retention_ms.max(output.retention_ms); + inputs.insert(output.node_id as u64, cached.state.clone()); } // Raw samples read at query time, other than stored outputs' panes. for window in &installed.query_time_raw_windows { @@ -332,7 +382,7 @@ fn run_data_plane( .iter() .map(|batches| answers(batches)) .collect(); - // The store drops each pane whose window starts more than its + // The cache drops each pane whose window starts more than its // retention before this evaluation. store.retain(|_, pane| pane.start_ms >= evaluation_ms - pane.retention_ms as i64); @@ -407,11 +457,11 @@ async fn label_matchers_select_the_series_a_pane_reads() { } /// The p99 over the last hour and over the hour before the last 10 -/// minutes store 12 panes of one computation; five panes are the same -/// (offsets 10..50 min), so the data plane builds 7, and both answers equal -/// the reference. +/// minutes store six panes each; five of them cover the same windows +/// (offsets 10..50 min), which this runtime's own cache key finds, so it +/// builds 7. Both answers equal the reference. #[tokio::test] -async fn panes_of_one_computation_build_once_across_queries() { +async fn an_hour_and_its_offset_twin_both_equal_the_reference() { let plan_json = plan_json(&[ "quantile_over_time(0.99, latency_ms[60m])", "quantile_over_time(0.99, latency_ms[60m] offset 10m)", @@ -419,12 +469,6 @@ async fn panes_of_one_computation_build_once_across_queries() { .await; let installed: InstallablePlan = serde_json::from_str(&plan_json).unwrap(); assert_eq!(installed.stored_outputs.len(), 12); - let computations: BTreeSet<&str> = installed - .stored_outputs - .iter() - .map(|o| o.computation_id.as_str()) - .collect(); - assert_eq!(computations.len(), 1); let run = run_data_plane( &plan_json, &remote_write_samples(), @@ -542,7 +586,7 @@ async fn query_time_results_kept_across_evaluations_are_built_in_query_runs() { .query_time_raw_windows .iter() .all(|w| !pane_ranges.contains(&w.time_range_node_id))); - // The store the data plane keeps across evaluations. + // The pane cache the data plane keeps across evaluations. let mut store = PaneStore::new(); let first = run_data_plane( &plan_json, @@ -559,6 +603,6 @@ async fn query_time_results_kept_across_evaluations_are_built_in_query_runs() { EVALUATION_MS + 60_000, &mut store, ); - assert_eq!(next.panes_built, 1, "four panes reused from the store"); + assert_eq!(next.panes_built, 1, "four panes reused from the cache"); assert_eq!(next.answers_per_query, next.reference_answers_per_query); }