From 78030350072869d9053d8659f5c1948997b0316e Mon Sep 17 00:00:00 2001 From: zzylol Date: Wed, 21 Oct 2026 01:17:01 +0000 Subject: [PATCH 1/5] feat(sds): adopt ASAPPlanner #787: the plan's definition id and full selection Pin ASAPPlanner 4111e620 (#787: StoredOutput gives its SummaryCoverage, definition id and selection, and its inputs instead of a pane). - A definition's identity is the plan's coverage.definition_id; the SDS's own sub-DAG hash is gone (and with it the sha2 dependency). - The definition id leaves the selection out, so outputs of one id with other label constraints are other SummaryDefinitions (DefinitionKey: id and label constraints), and a summary instance is keyed by the id and its concrete Selection: the label constraints and the pane end the plan's relative time binds to. - A definition's pane width and each output's offset come from its selection's relative time; its RawSampleWindow is the stored output's raw-sample input; its stored inputs are the plan's inputs.stored_outputs. A selection that is not one left-open time box, or several raw-sample windows, is refused. --- Cargo.lock | 84 +----- Cargo.toml | 4 +- control_plane/tests/plan_workload.rs | 13 +- data_plane/Cargo.toml | 1 - data_plane/src/per_sample.rs | 8 +- data_plane/src/precompute.rs | 39 ++- data_plane/src/query.rs | 53 ++-- data_plane/src/sds.rs | 367 ++++++++++++++++----------- data_plane/tests/per_sample.rs | 23 +- data_plane/tests/precompute.rs | 167 ++---------- data_plane/tests/query_run.rs | 14 +- data_plane/tests/sds.rs | 146 +++++------ 12 files changed, 408 insertions(+), 511 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 7fdd3f01..468457e8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -252,7 +252,7 @@ dependencies = [ [[package]] name = "asap-executor" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=7565c47e30609d9571d8a9a372bf5e195754b2bb#7565c47e30609d9571d8a9a372bf5e195754b2bb" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=4111e62014aa1944fa546075a4b161fe390a2439#4111e62014aa1944fa546075a4b161fe390a2439" dependencies = [ "ahash", "arrow", @@ -270,7 +270,7 @@ dependencies = [ [[package]] name = "asap-frontend-common" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=7565c47e30609d9571d8a9a372bf5e195754b2bb#7565c47e30609d9571d8a9a372bf5e195754b2bb" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=4111e62014aa1944fa546075a4b161fe390a2439#4111e62014aa1944fa546075a4b161fe390a2439" dependencies = [ "asap-types", "serde", @@ -280,7 +280,7 @@ dependencies = [ [[package]] name = "asap-frontend-promql" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=7565c47e30609d9571d8a9a372bf5e195754b2bb#7565c47e30609d9571d8a9a372bf5e195754b2bb" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=4111e62014aa1944fa546075a4b161fe390a2439#4111e62014aa1944fa546075a4b161fe390a2439" dependencies = [ "asap-frontend-common", "asap-types", @@ -290,7 +290,7 @@ dependencies = [ [[package]] name = "asap-logical-asap-optimization" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=7565c47e30609d9571d8a9a372bf5e195754b2bb#7565c47e30609d9571d8a9a372bf5e195754b2bb" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=4111e62014aa1944fa546075a4b161fe390a2439#4111e62014aa1944fa546075a4b161fe390a2439" dependencies = [ "asap-types", "asap_sketchlib", @@ -302,7 +302,7 @@ dependencies = [ [[package]] name = "asap-physical-asap-optimization" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=7565c47e30609d9571d8a9a372bf5e195754b2bb#7565c47e30609d9571d8a9a372bf5e195754b2bb" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=4111e62014aa1944fa546075a4b161fe390a2439#4111e62014aa1944fa546075a4b161fe390a2439" dependencies = [ "asap-logical-asap-optimization", "asap-types", @@ -312,7 +312,7 @@ dependencies = [ [[package]] name = "asap-plan-selection" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=7565c47e30609d9571d8a9a372bf5e195754b2bb#7565c47e30609d9571d8a9a372bf5e195754b2bb" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=4111e62014aa1944fa546075a4b161fe390a2439#4111e62014aa1944fa546075a4b161fe390a2439" dependencies = [ "asap-logical-asap-optimization", "asap-physical-asap-optimization", @@ -326,7 +326,7 @@ dependencies = [ [[package]] name = "asap-planner" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=7565c47e30609d9571d8a9a372bf5e195754b2bb#7565c47e30609d9571d8a9a372bf5e195754b2bb" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=4111e62014aa1944fa546075a4b161fe390a2439#4111e62014aa1944fa546075a4b161fe390a2439" dependencies = [ "asap-executor", "asap-frontend-promql", @@ -341,7 +341,7 @@ dependencies = [ [[package]] name = "asap-types" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=7565c47e30609d9571d8a9a372bf5e195754b2bb#7565c47e30609d9571d8a9a372bf5e195754b2bb" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=4111e62014aa1944fa546075a4b161fe390a2439#4111e62014aa1944fa546075a4b161fe390a2439" dependencies = [ "serde", "serde_json", @@ -474,15 +474,6 @@ version = "2.13.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3ded4057c258ba199e2d26386d3af3780957ecaee6c4ef4041c6b4b8b97c0b06" -[[package]] -name = "block-buffer" -version = "0.10.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3078c7629b62d3f0439517fa394996acacc5cbc91c5a20d8c658e77abd503a71" -dependencies = [ - "generic-array", -] - [[package]] name = "bumpalo" version = "3.20.3" @@ -544,7 +535,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06" dependencies = [ "cfg-if", - "cpufeatures 0.3.1", + "cpufeatures", "rand_core 0.10.1", ] @@ -648,15 +639,6 @@ version = "0.8.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" -[[package]] -name = "cpufeatures" -version = "0.2.17" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "59ed5838eebb26a2bb2e58f6d5b5316989ae9d08bab10e0e6d103e656d1b0280" -dependencies = [ - "libc", -] - [[package]] name = "cpufeatures" version = "0.3.1" @@ -672,16 +654,6 @@ version = "0.2.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "460fbee9c2c2f33933d720630a6a0bac33ba7053db5344fac858d4b8952d77d5" -[[package]] -name = "crypto-common" -version = "0.1.7" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "78c8292055d1c1df0cce5d180393dc8cce0abec0a7102adb6c7b1eef6016d60a" -dependencies = [ - "generic-array", - "typenum", -] - [[package]] name = "data_plane" version = "0.1.0" @@ -700,7 +672,6 @@ dependencies = [ "reqwest", "serde", "serde_json", - "sha2", "snap", "thiserror 1.0.69", "tokio", @@ -715,16 +686,6 @@ version = "0.5.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7cd812cc2bc1d69d4764bd80df88b4317eaef9e773c75226407d9bc0876b211c" -[[package]] -name = "digest" -version = "0.10.7" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" -dependencies = [ - "block-buffer", - "crypto-common", -] - [[package]] name = "displaydoc" version = "0.2.7" @@ -877,16 +838,6 @@ dependencies = [ "slab", ] -[[package]] -name = "generic-array" -version = "0.14.7" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "85649ca51fd72272d7821adaf274ad91c288277713d9c18820d8499a7ff69e9a" -dependencies = [ - "typenum", - "version_check", -] - [[package]] name = "getopts" version = "0.2.24" @@ -2007,17 +1958,6 @@ dependencies = [ "serde", ] -[[package]] -name = "sha2" -version = "0.10.9" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283" -dependencies = [ - "cfg-if", - "cpufeatures 0.2.17", - "digest", -] - [[package]] name = "sharded-slab" version = "0.1.7" @@ -2425,12 +2365,6 @@ dependencies = [ "rand 0.10.2", ] -[[package]] -name = "typenum" -version = "1.20.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" - [[package]] name = "unicode-ident" version = "1.0.24" diff --git a/Cargo.toml b/Cargo.toml index 1c511734..c1950b1e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -10,8 +10,8 @@ version = "0.1.0" # The backend depends on two Planner crates, at one immutable revision: # `asap-planner` to plan and `asap-executor` to execute. Planner IR types # are reached through `asap_executor::planner`. -asap-planner = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "7565c47e30609d9571d8a9a372bf5e195754b2bb", default-features = false, features = ["promql"] } -asap-executor = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "7565c47e30609d9571d8a9a372bf5e195754b2bb" } +asap-planner = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "4111e62014aa1944fa546075a4b161fe390a2439", default-features = false, features = ["promql"] } +asap-executor = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "4111e62014aa1944fa546075a4b161fe390a2439" } serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" diff --git a/control_plane/tests/plan_workload.rs b/control_plane/tests/plan_workload.rs index 588a67ee..ac8336af 100644 --- a/control_plane/tests/plan_workload.rs +++ b/control_plane/tests/plan_workload.rs @@ -50,8 +50,11 @@ fn example4b_stores_panes_with_the_planners_retention() { .stored_outputs .iter() .map(|output| { - assert_eq!(output.pane.width_ms, 600_000); - output.pane.end_offset_ms + let [window] = output.inputs.raw_sample_windows.as_slice() else { + panic!("one raw-sample window: {output:?}"); + }; + assert_eq!(window.width_ms, 600_000); + window.end_offset_ms }) .collect(); offsets.sort(); @@ -104,7 +107,11 @@ fn selects_stored_outputs_built_in_a_query_run() { .filter(|o| o.built_at == asap_planner::StoredOutputBuild::QueryRun) .collect(); assert_eq!(query_run.len(), 5, "five one-minute summaries"); - assert!(query_run.iter().all(|o| o.pane.width_ms == 60_000)); + assert!(query_run.iter().all(|o| o + .inputs + .raw_sample_windows + .iter() + .all(|w| w.width_ms == 60_000))); } /// A workload is planned whole or refused: when one query has no plan the diff --git a/data_plane/Cargo.toml b/data_plane/Cargo.toml index 06ab9f98..c09d2582 100644 --- a/data_plane/Cargo.toml +++ b/data_plane/Cargo.toml @@ -6,7 +6,6 @@ edition.workspace = true [dependencies] asap-planner.workspace = true asap-executor.workspace = true -sha2 = "0.10" ahash = "0.8" serde.workspace = true serde_json.workspace = true diff --git a/data_plane/src/per_sample.rs b/data_plane/src/per_sample.rs index 4721566a..4a6fe764 100644 --- a/data_plane/src/per_sample.rs +++ b/data_plane/src/per_sample.rs @@ -125,6 +125,8 @@ impl PerSampleSpec { /// keeps the batch build. pub(crate) fn new(dag: &PhysicalASAPDAG, pane: &SummaryBuild) -> Option { let definition = &pane.definition; + let window = definition.window.as_ref()?; + let raw_input = window.time_range_node_id as u64; let node = |id: u64| dag.nodes.iter().find(|n| n.id as u64 == id); let builder = node(definition.builder)?; let inputs: Vec = dag @@ -133,7 +135,7 @@ impl PerSampleSpec { .filter(|e| e.consumer as u64 == definition.builder) .map(|e| e.producer as u64) .collect(); - if inputs != [definition.raw_input()] { + if inputs != [raw_input] { return None; } let Operator::ASAP(ASAPOp::SummaryAgg { @@ -150,7 +152,7 @@ impl PerSampleSpec { if update.item.is_some() || *grouping != GroupingStrategy::PerSubpopulationInstance { return None; } - let input = &node(definition.raw_input())?.output_schema; + let input = &node(raw_input)?.output_schema; let SummaryInputExpr::Column(column) = &update.weight else { return None; }; @@ -177,7 +179,7 @@ impl PerSampleSpec { return None; } // The cached prefix holds only if the matchers read labels alone. - let scan = node(definition.window.scan_node_id as u64)?; + let scan = node(window.scan_node_id as u64)?; let Operator::NonASAP(NonASAPOp::Scan { predicates, .. }) = &scan.payload else { return None; }; diff --git a/data_plane/src/precompute.rs b/data_plane/src/precompute.rs index 8c9d6ce9..b2ff4d73 100644 --- a/data_plane/src/precompute.rs +++ b/data_plane/src/precompute.rs @@ -35,9 +35,7 @@ use asap_planner::StoredOutputBuild; use crate::per_sample::{OpenPanes as OpenStates, PerSampleSpec, Taken}; use crate::raw_sample_window::RawSampleWindowBatcher; -use crate::sds::{ - DefinitionId, Sds, Selection, SummaryDefinition, SummaryInstanceKey, SummaryStore, -}; +use crate::sds::{DefinitionId, Sds, SummaryDefinition, SummaryInstanceKey, SummaryStore}; use crate::series_ids::Series; /// One raw sample: its series (interned: its id and full label set, @@ -140,6 +138,8 @@ pub struct SummaryBuild { /// build's input batch, column by column, with the `Scan`'s label /// matchers. pub(crate) window: RawSampleWindowBatcher, + /// The `TimeRange` node the window's batch feeds. + raw_input: u64, build: CompiledPhysicalDAG, } @@ -164,17 +164,19 @@ impl SummaryBuild { // offset from evaluation time is not used. // A series is cached for as long as the definition's summaries are kept. let keep_ms = definition.retention_ms.saturating_add(definition.pane_ms); - let window = RawSampleWindowBatcher::new(dag, &definition.window, keep_ms)?; + let Some(raw) = &definition.window else { + return Err(error("reads no raw-sample window".into())); + }; + let raw_input = raw.time_range_node_id as u64; + let window = RawSampleWindowBatcher::new(dag, raw, keep_ms)?; let build = compile( dag, - BTreeMap::from([( - definition.raw_input(), - InputContract::bounded(window.schema.clone()), - )]), + BTreeMap::from([(raw_input, InputContract::bounded(window.schema.clone()))]), &[definition.builder], )?; Ok(Self { definition: definition.clone(), + raw_input, window, build, }) @@ -205,7 +207,7 @@ impl SummaryBuild { let input: Source<'_> = Box::new(Operator::source(input.schema().clone(), vec![input])?); let build = self .build - .instantiate(BTreeMap::from([(self.definition.raw_input(), input)]))?; + .instantiate(BTreeMap::from([(self.raw_input, input)]))?; Ok(SummaryInstance { batches: run(&build, self.build.roots(), scope)?, }) @@ -321,7 +323,10 @@ impl Precompute { let mut query_run_metrics = BTreeMap::new(); for definition in sds.definitions.values() { if definition.built_at == StoredOutputBuild::QueryRun { - match RawSampleSelector::new(&dag, &definition.window) { + let Some(window) = &definition.window else { + continue; + }; + match RawSampleSelector::new(&dag, window) { Ok(selector) => { query_run_metrics.insert(selector.metric().to_string(), None); } @@ -537,7 +542,10 @@ impl Precompute { match built { Ok(state) => { self.store.insert( - SummaryInstanceKey::new(id.clone(), Selection::ending_at(end)), + SummaryInstanceKey::new( + id.clone(), + builder.summary.definition.selection_ending_at(end), + ), state, ); self.stats.panes_built += 1; @@ -553,8 +561,13 @@ impl Precompute { None => break, } } - self.store - .evict_before(id, Selection::ending_at(oldest_kept_end)); + self.store.evict_before( + id, + &builder + .summary + .definition + .selection_ending_at(oldest_kept_end), + ); } } } diff --git a/data_plane/src/query.rs b/data_plane/src/query.rs index 6f16c3cf..14951c87 100644 --- a/data_plane/src/query.rs +++ b/data_plane/src/query.rs @@ -35,7 +35,7 @@ use asap_planner::{InstallablePlan, StoredOutputBuild}; use crate::precompute::{run, RawSample, SummaryBuild, SummaryInstance}; use crate::raw_sample_window::RawSampleWindowBatcher; -use crate::sds::{DefinitionId, Sds, Selection, SummaryInstanceKey, SummaryStore}; +use crate::sds::{DefinitionKey, Sds, Selection, SummaryInstanceKey, SummaryStore}; /// A bound raw-sample window: a `RawSampleWindow` of the plan bound to one /// evaluation time, i.e. `metric`'s raw samples in `(start_ms, end_ms]` @@ -108,7 +108,7 @@ struct StoredInput { node: u64, /// The output's schema, which its panes' rows are in. schema: SchemaRef, - definition: DefinitionId, + definition: DefinitionKey, offset_ms: i64, pane_ms: i64, } @@ -126,7 +126,7 @@ struct RawInput { struct QueryRunInput { node: u64, schema: SchemaRef, - definition: DefinitionId, + definition: DefinitionKey, offset_ms: i64, /// The output's own `TimeRange`, which its coverage's samples feed. raw_input: u64, @@ -138,7 +138,7 @@ struct Program { query_run: Vec, /// For each `QueryRun` definition the root reads, the node its one /// fetch is keyed by: the lowest `TimeRange` among its outputs. - query_run_fetch_node: BTreeMap, + query_run_fetch_node: BTreeMap, raw: Vec, output: Schema, } @@ -187,10 +187,10 @@ pub struct QueryEngine { pub query_run_stats: QueryRunStats, /// How each `QueryRun` definition's summaries are built, or why they /// cannot be. - query_run_builds: BTreeMap>, + query_run_builds: BTreeMap>, /// The label matchers of each `QueryRun` definition's `Scan`, pushed into /// its fetch. - query_run_matchers: BTreeMap>, + query_run_matchers: BTreeMap>, /// Each query, with its compiled root or why it cannot run. roots: Vec<(String, Result)>, } @@ -209,7 +209,7 @@ impl QueryEngine { .filter(|d| d.built_at == StoredOutputBuild::QueryRun) .map(|d| { let build = SummaryBuild::new(dag, d).map_err(|e| e.to_string()); - (d.id.clone(), build) + (d.key(), build) }) .collect(); let roots = plan @@ -223,8 +223,12 @@ impl QueryEngine { let query_run_matchers = query_run_builds .keys() .map(|id| { - let scan = sds.definitions[id].window.scan_node_id as u64; - (id.clone(), label_matchers(dag, scan)) + let matchers = sds.definitions[id] + .window + .as_ref() + .map(|w| label_matchers(dag, w.scan_node_id as u64)) + .unwrap_or_default(); + (id.clone(), matchers) }) .collect(); Self { @@ -268,12 +272,15 @@ impl QueryEngine { .collect::>()?; // One fetch per `QueryRun` definition, covering all its missing // summaries. - let mut missing: BTreeMap<&DefinitionId, (i64, i64)> = BTreeMap::new(); + let mut missing: BTreeMap<&DefinitionKey, (i64, i64)> = BTreeMap::new(); for input in &program.query_run { let build = self.query_run_build(&input.definition); let (start_ms, end_ms) = window(time_ms, input.offset_ms, build.definition.pane_ms)?; if store - .get(&input.definition, Selection::ending_at(end_ms)) + .get( + &input.definition.id, + &selection_of(&input.definition, end_ms), + ) .is_none() { let span = missing @@ -336,8 +343,8 @@ impl QueryEngine { .as_ref() .expect("compiled roots read only `QueryRun` definitions that build"); let (_, end_ms) = window(time_ms, input.offset_ms, build.definition.pane_ms)?; - let selection = Selection::ending_at(end_ms); - let summary = match store.get(&input.definition, selection) { + let selection = selection_of(&input.definition, end_ms); + let summary = match store.get(&input.definition.id, &selection) { Some(summary) => { self.query_run_stats.reused += 1; summary.clone() @@ -372,7 +379,7 @@ impl QueryEngine { self.query_run_stats.stored_while_metric_behind += 1; } store.insert( - SummaryInstanceKey::new(input.definition.clone(), selection), + SummaryInstanceKey::new(input.definition.id.clone(), selection), summary.clone(), ); } @@ -391,7 +398,7 @@ impl QueryEngine { let newest_closed_end = watermark.div_euclid(pane_ms).saturating_mul(pane_ms); let oldest = definition.oldest_kept_end_ms(newest_closed_end.saturating_sub(pane_ms)); - store.evict_before(&definition.id, Selection::ending_at(oldest)); + store.evict_before(&definition.id, &definition.selection_ending_at(oldest)); } } for input in &program.raw { @@ -416,7 +423,7 @@ impl QueryEngine { } /// The build of a `QueryRun` definition a compiled root reads. - fn query_run_build(&self, definition: &DefinitionId) -> &SummaryBuild { + fn query_run_build(&self, definition: &DefinitionKey) -> &SummaryBuild { self.query_run_builds[definition] .as_ref() .expect("compiled roots read only `QueryRun` definitions that build") @@ -436,6 +443,14 @@ impl QueryEngine { } } +/// The concrete selection of `definition`'s pane ending at `end_ms`. +fn selection_of(definition: &DefinitionKey, end_ms: i64) -> Selection { + Selection { + labels: definition.labels.clone(), + end_ms, + } +} + /// `(start, end]` of a window `width_ms` wide ending `offset_ms` before /// `time_ms`, or why it does not fit in `i64` milliseconds. fn window(time_ms: i64, offset_ms: i64, width_ms: i64) -> Result<(i64, i64), Unanswered> { @@ -460,7 +475,7 @@ fn pane<'a>( ))); } store - .get(&input.definition, Selection::ending_at(end)) + .get(&input.definition.id, &selection_of(&input.definition, end)) .ok_or_else(|| { Unanswered::MissingPane(format!( "pane ending at {end} of {:?} is not stored", @@ -473,7 +488,7 @@ fn pane<'a>( fn program( dag: &PhysicalASAPDAG, sds: &Sds, - query_run_builds: &BTreeMap>, + query_run_builds: &BTreeMap>, windows: &HashMap, root: u64, ) -> Result { @@ -562,7 +577,7 @@ fn program( ); } let compiled = compile(dag, inputs, &[root]).map_err(|e| e.to_string())?; - let mut query_run_fetch_node: BTreeMap = BTreeMap::new(); + let mut query_run_fetch_node: BTreeMap = BTreeMap::new(); for input in &query_run { let node = query_run_fetch_node .entry(input.definition.clone()) diff --git a/data_plane/src/sds.rs b/data_plane/src/sds.rs index 299c7528..4dd0a3ae 100644 --- a/data_plane/src/sds.rs +++ b/data_plane/src/sds.rs @@ -1,65 +1,92 @@ -//! Stored summaries (SDS): the immutable summaries the data plane stores -//! for a plan, and how a query finds them again. +//! Stored summaries (SDS): the summary instances the data plane stores for a +//! plan, and how a query finds them again. //! //! The installed plan lists every output it stores across evaluations -//! (`InstallablePlan::stored_outputs`): which run builds it (an ingestion -//! run, or the query run of the first evaluation that reads it), what it -//! covers, and how long it must stay available. Today the Planner's contract -//! states each stored output's coverage as one pane of raw samples: at -//! evaluation time `T` it covers `(T - end_offset - width, T - end_offset]`, -//! the PromQL range convention. +//! (`InstallablePlan::stored_outputs`): which run builds it (an +//! `IngestionRun`, or the `QueryRun` of the first evaluation that reads it), +//! its `SummaryCoverage` (the plan's definition id and selection), the +//! `RawSampleWindow`s and stored outputs its build reads, and its +//! `retention_ms`. //! -//! Outputs that differ only in which coverage they read compute the same -//! thing, so they share one [`SummaryDefinition`]: the data plane builds each -//! summary of a definition once, keyed by its coverage end, and every output -//! reads the one it needs. Identity is the SDS's own: the Planner states what -//! each node computes, not which nodes are equal. +//! A definition's identity is the plan's definition id. Outputs with the same +//! definition id and the same label constraints in their selection differ +//! only in time, so they share one [`SummaryDefinition`]: the data plane +//! builds each pane once and every output reads the one it needs. Outputs +//! with the same definition id but other label constraints are other +//! [`SummaryDefinition`]s ([`DefinitionKey`]). A summary instance is keyed by +//! its definition id and its concrete [`Selection`]: those label constraints +//! and the time interval `(end - width, end]` the plan's relative time binds +//! to at evaluation time. //! -//! A stored output may in future be built from other stored outputs -//! (rollups over time or across series). [`Sds::ingestion_build_order`] -//! orders definitions so that each one's stored inputs come first; the -//! precompute engine refuses such a definition until it can build it once -//! all its stored inputs are sealed. +//! A stored output may be built from other stored outputs (rollups over time +//! or across series). [`Sds::ingestion_build_order`] orders definitions so +//! that each one's stored inputs come first; the precompute engine refuses +//! such a definition until it can build it once all its stored inputs are +//! sealed. use std::collections::{BTreeMap, HashMap}; -use asap_executor::planner::ir::physical_export::{ - PhysicalASAPDAG, PhysicalASAPDAGNode, RawSampleWindow, -}; +use std::ops::Bound; +use std::sync::Arc; + +use asap_executor::planner::ir::physical_export::{PhysicalASAPDAGNode, RawSampleWindow}; +use asap_executor::planner::ir::properties::summary_coverage::SelectionBox; use asap_executor::planner::ir::schema::Schema; -use asap_executor::planner::ir::{NonASAPOp, Operator}; use asap_planner::{InstallablePlan, StoredOutputBuild}; -use sha2::{Digest, Sha256}; -/// Identifies what a stored summary computes, independent of node ids and of -/// which data an output reads: the hash of its coverage definition. Equal for -/// equal computations across plans. When the Planner states each stored -/// output's coverage (ASAPPlanner #781), this becomes the plan's definition -/// id instead of the SDS's own hash. +/// What a stored summary computes: the plan's +/// `StoredCoverage::definition_id`, equal on every node, plan and process +/// that computes the same thing, whatever its selection. #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] pub struct DefinitionId(pub String); -/// One computation the data plane stores summary by summary. +/// The label constraints of a selection (its `SelectionBox` columns), in the +/// plan's JSON form; empty when it has none. +pub type LabelConstraints = Arc; + +/// One [`SummaryDefinition`] of the SDS: a definition id and the label +/// constraints its outputs share. +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct DefinitionKey { + pub id: DefinitionId, + pub labels: LabelConstraints, +} + +impl DefinitionKey { + /// `id` with no label constraints. + pub fn unconstrained(id: DefinitionId) -> Self { + Self { + id, + labels: Arc::from(""), + } + } +} + +/// One computation the data plane stores pane by pane, for one set of label +/// constraints. #[derive(Debug, Clone, PartialEq)] pub struct SummaryDefinition { pub id: DefinitionId, + /// The label constraints of its outputs' selection. + pub labels: LabelConstraints, /// Which run builds its stored outputs (`StoredOutputBuild`): an /// `IngestionRun`, or the `QueryRun` of the first evaluation that needs /// one (stored once its coverage is closed). `IngestionRun` when any /// output of the definition says so. pub built_at: StoredOutputBuild, - /// Coverage width: each stored summary covers `(end - pane_ms, end]`. + /// Pane width: each summary instance covers `(end - pane_ms, end]`. pub pane_ms: i64, /// The output that builds a pane; any output of the definition computes /// the same, and this is the first in node order. pub builder: u64, - /// The builder's pane: its `TimeRange`, fed one pane's samples, and the - /// `Scan` that selects them. - pub window: RawSampleWindow, + /// The `RawSampleWindow` the builder reads: its `TimeRange`, fed one + /// pane's raw samples, and the `Scan` that selects them. `None` for a + /// definition built from other stored outputs only. + pub window: Option, pub output_schema: Schema, /// The definitions of the stored outputs the builder reads, which must /// be built first. Empty while every stored output reads raw samples. - pub stored_inputs: Vec, + pub stored_inputs: Vec, /// How far back from the newest evaluation time a summary stays available: /// the largest `retention_ms` the plan states for the definition's /// outputs. A pane that starts earlier is no longer read. @@ -67,9 +94,25 @@ pub struct SummaryDefinition { } impl SummaryDefinition { - /// The builder's `TimeRange`, which one pane's samples feed. - pub fn raw_input(&self) -> u64 { - self.window.time_range_node_id as u64 + /// Its key in [`Sds::definitions`]. + pub fn key(&self) -> DefinitionKey { + DefinitionKey { + id: self.id.clone(), + labels: self.labels.clone(), + } + } + + /// The concrete selection of its pane ending at `end_ms`. + pub fn selection_ending_at(&self, end_ms: i64) -> Selection { + Selection { + labels: self.labels.clone(), + end_ms, + } + } + + /// The builder's `TimeRange`, which one pane's raw samples feed. + pub fn raw_input(&self) -> Option { + self.window.as_ref().map(|w| w.time_range_node_id as u64) } /// The end of the oldest pane still read when the newest evaluation is @@ -94,10 +137,10 @@ pub struct PaneBinding { /// The stored summaries of one plan. #[derive(Debug, Clone, PartialEq, Default)] pub struct Sds { - /// Every definition, built at ingestion or at query time, by identity. - pub definitions: BTreeMap, - /// Every stored output, with its definition and coverage. - pub outputs: BTreeMap, + /// Every definition, built in an `IngestionRun` or a `QueryRun`. + pub definitions: BTreeMap, + /// Every stored output, with its definition and pane. + pub outputs: BTreeMap, } #[derive(Debug, thiserror::Error, PartialEq)] @@ -108,14 +151,18 @@ pub enum SdsError { RetentionOverflow(u64), /// Stored outputs read each other in a cycle. #[error("stored outputs read each other in a cycle through {0:?}")] - Cycle(DefinitionId), + Cycle(DefinitionKey), /// An `IngestionRun` definition reads a `QueryRun` one, which is built /// only when an evaluation needs it, so never in time for ingestion. #[error("stored output built in an IngestionRun ({definition:?}) reads one built in a QueryRun ({input:?})")] IngestionReadsQueryRun { - definition: DefinitionId, - input: DefinitionId, + definition: DefinitionKey, + input: DefinitionKey, }, + /// A stored output's coverage or inputs are not what the data plane can + /// build. + #[error("stored output {node}: {reason}")] + Unsupported { node: u64, reason: String }, } impl Sds { @@ -125,89 +172,115 @@ impl Sds { let dag = &plan.dag.dag; let nodes: HashMap = dag.nodes.iter().map(|n| (n.id as u64, n)).collect(); - let inputs = inputs_of(dag); let mut sds = Sds::default(); - let mut hashes = HashMap::new(); + let mut stored_inputs: HashMap> = HashMap::new(); for output in &plan.stored_outputs { let node = output.node_id as u64; - let (definitions, outputs) = (&mut sds.definitions, &mut sds.outputs); - let id = DefinitionId(hex(&identity(node, &nodes, &inputs, &mut hashes)?)); + let unsupported = |reason: &str| SdsError::Unsupported { + node, + reason: reason.into(), + }; + let output_schema = nodes + .get(&node) + .ok_or(SdsError::UnknownNode(node))? + .output_schema + .clone(); + let [selection] = output.coverage.selection.as_slice() else { + return Err(unsupported("its selection is not one box")); + }; + let (start, end) = relative_time(selection) + .ok_or_else(|| unsupported("its selection has no left-open time interval"))?; + let window = match output.inputs.raw_sample_windows.as_slice() { + [] => None, + [window] => Some(window.clone()), + _ => return Err(unsupported("it reads several raw-sample windows")), + }; + if window.is_none() && output.inputs.stored_outputs.is_empty() { + return Err(unsupported("it reads no input")); + } + let key = DefinitionKey { + id: DefinitionId(output.coverage.definition_id.clone()), + labels: label_constraints(selection), + }; let retention_ms = i64::try_from(output.retention_ms) .map_err(|_| SdsError::RetentionOverflow(node))?; - let output_schema = nodes[&node].output_schema.clone(); - let definition = definitions - .entry(id.clone()) - .or_insert_with(|| SummaryDefinition { - id: id.clone(), - built_at: output.built_at, - pane_ms: output.pane.width_ms, - builder: node, - window: output.pane.clone(), - output_schema, - stored_inputs: Vec::new(), - retention_ms, - }); + let definition = + sds.definitions + .entry(key.clone()) + .or_insert_with(|| SummaryDefinition { + id: key.id.clone(), + labels: key.labels.clone(), + built_at: output.built_at, + pane_ms: end - start, + builder: node, + window: window.clone(), + output_schema, + stored_inputs: Vec::new(), + retention_ms, + }); if output.built_at == StoredOutputBuild::IngestionRun { definition.built_at = StoredOutputBuild::IngestionRun; } if node < definition.builder { definition.builder = node; - definition.window = output.pane.clone(); + definition.window = window.clone(); } definition.retention_ms = definition.retention_ms.max(retention_ms); let binding = PaneBinding { - offset_ms: output.pane.end_offset_ms, - raw_input: output.pane.time_range_node_id as u64, + offset_ms: -end, + raw_input: window + .as_ref() + .map_or(node, |w| w.time_range_node_id as u64), }; - outputs.insert(node, (id, binding)); + sds.outputs.insert(node, (key, binding)); + stored_inputs.insert( + node, + output + .inputs + .stored_outputs + .iter() + .map(|n| *n as u64) + .collect(), + ); } - sds.link_stored_inputs(&inputs); + sds.link_stored_inputs(&stored_inputs)?; sds.ingestion_build_order()?; Ok(sds) } /// Records, for each definition, the definitions of the stored outputs - /// its builder reads: those found below the builder before any raw - /// input. - fn link_stored_inputs(&mut self, inputs: &HashMap>) { - let stored: HashMap = self + /// its builder reads, as the plan lists them (`StoredOutputInputs`). + fn link_stored_inputs(&mut self, inputs: &HashMap>) -> Result<(), SdsError> { + let stored: HashMap = self .outputs .iter() - .map(|(node, (id, _))| (*node, id.clone())) + .map(|(node, (key, _))| (*node, key.clone())) .collect(); for definition in self.definitions.values_mut() { let mut found = Vec::new(); - let mut pending: Vec = - inputs.get(&definition.builder).cloned().unwrap_or_default(); - let mut seen = std::collections::HashSet::new(); - while let Some(node) = pending.pop() { - if !seen.insert(node) { - continue; + for input in inputs.get(&definition.builder).into_iter().flatten() { + let key = stored.get(input).ok_or(SdsError::UnknownNode(*input))?; + if !found.contains(key) { + found.push(key.clone()); } - if let Some(id) = stored.get(&node) { - if !found.contains(id) { - found.push(id.clone()); - } - continue; - } - pending.extend(inputs.get(&node).into_iter().flatten().copied()); } found.sort(); definition.stored_inputs = found; } + Ok(()) } /// The definitions built in ingestion runs, each after every definition /// it reads (`stored_inputs`). Fails on a cycle, and when an /// `IngestionRun` definition reads a `QueryRun` one. - pub fn ingestion_build_order(&self) -> Result, SdsError> { + pub fn ingestion_build_order(&self) -> Result, SdsError> { let mut order = Vec::new(); - let mut state: HashMap<&DefinitionId, bool> = HashMap::new(); // false: visiting + let mut state: HashMap<&DefinitionKey, bool> = HashMap::new(); // false: visiting fn visit<'a>( sds: &'a Sds, - id: &'a DefinitionId, - state: &mut HashMap<&'a DefinitionId, bool>, - order: &mut Vec, + id: &'a DefinitionKey, + state: &mut HashMap<&'a DefinitionKey, bool>, + order: &mut Vec, ) -> Result<(), SdsError> { match state.get(id) { Some(true) => return Ok(()), @@ -239,69 +312,45 @@ impl Sds { } } -/// Each node's inputs, in edge order. -fn inputs_of(dag: &PhysicalASAPDAG) -> HashMap> { - let mut inputs: HashMap> = HashMap::new(); - for edge in &dag.edges { - inputs - .entry(edge.consumer as u64) - .or_default() - .push(edge.producer as u64); +/// `(start, end]` of `selection`'s relative time, the offsets from the +/// evaluation time a pane covers, when it is a left-open interval. +fn relative_time(selection: &SelectionBox) -> Option<(i64, i64)> { + match selection.relative_time? { + (Bound::Excluded(start), Bound::Included(end)) if start < end => Some((start, end)), + _ => None, } - inputs } -/// SHA-256 over a node's payload and output schema, then its inputs' -/// identities in edge order. The payload's input ids are zeroed, so the -/// identity does not depend on node numbering. An offset `TimeShift` has its -/// input's identity: the offset says which pane an output reads -/// ([`PaneBinding`]), not what it computes, so the panes of one computation -/// share an identity whether or not their range is shifted. -fn identity( - node: u64, - nodes: &HashMap, - inputs: &HashMap>, - hashes: &mut HashMap, -) -> Result<[u8; 32], SdsError> { - if let Some(hash) = hashes.get(&node) { - return Ok(*hash); - } - let dag_node = nodes.get(&node).ok_or(SdsError::UnknownNode(node))?; - if let Operator::NonASAP(NonASAPOp::TimeShift { shift, .. }) = &dag_node.payload { - if let (None, Some([input])) = (shift.at, inputs.get(&node).map(Vec::as_slice)) { - let hash = identity(*input, nodes, inputs, hashes)?; - hashes.insert(node, hash); - return Ok(hash); - } +/// `selection`'s label constraints in the plan's JSON form; empty when it has +/// none. +fn label_constraints(selection: &SelectionBox) -> LabelConstraints { + if selection.columns.is_empty() { + return Arc::from(""); } - 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 inputs.get(&node).into_iter().flatten() { - hasher.update(identity(input, nodes, inputs, hashes)?); - } - let hash: [u8; 32] = hasher.finalize().into(); - hashes.insert(node, hash); - Ok(hash) -} - -fn hex(bytes: &[u8]) -> String { - bytes.iter().map(|b| format!("{b:02x}")).collect() + let columns = SelectionBox { + columns: selection.columns.clone(), + relative_time: None, + }; + Arc::from(serde_json::to_string(&columns).expect("selections serialize")) } -/// Which data one summary instance covers. For now only time: the interval -/// `(end_ms - width, end_ms]`, the width being its definition's coverage -/// width. When the Planner states each stored output's coverage (definition -/// and selection, ASAPPlanner #781), this becomes the plan's selection. -#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)] +/// The concrete selection of one summary instance: its definition's label +/// constraints and the time interval `(end_ms - width, end_ms]` the plan's +/// relative time binds to at evaluation time, the width being its +/// definition's pane width. +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] pub struct Selection { + pub labels: LabelConstraints, pub end_ms: i64, } impl Selection { - /// The coverage ending at `end_ms`. + /// The pane ending at `end_ms`, with no label constraints. pub fn ending_at(end_ms: i64) -> Self { - Self { end_ms } + Self { + labels: Arc::from(""), + end_ms, + } } } @@ -349,16 +398,38 @@ impl SummaryStore { } /// The instance of `definition` covering `selection`, if stored. Borrows - /// the definition id, so a lookup allocates nothing. - pub fn get(&self, definition: &DefinitionId, selection: Selection) -> Option<&S> { - self.instances.get(definition)?.get(&selection) + /// both, so a lookup allocates nothing. + pub fn get(&self, definition: &DefinitionId, selection: &Selection) -> Option<&S> { + self.instances.get(definition)?.get(selection) + } + + /// The instance of `definition`'s pane ending at `end_ms`, if stored. + pub fn get_pane(&self, definition: &DefinitionKey, end_ms: i64) -> Option<&S> { + self.get( + &definition.id, + &Selection { + labels: definition.labels.clone(), + end_ms, + }, + ) } - /// Drops the instances of `definition` whose coverage ends before - /// `oldest`'s. - pub fn evict_before(&mut self, definition: &DefinitionId, oldest: Selection) { - if let Some(instances) = self.instances.get_mut(definition) { - *instances = instances.split_off(&oldest); + /// Drops the instances of `definition` with `oldest`'s label constraints + /// whose pane ends before `oldest`'s. + pub fn evict_before(&mut self, definition: &DefinitionId, oldest: &Selection) { + let Some(instances) = self.instances.get_mut(definition) else { + return; + }; + let from = Selection { + labels: oldest.labels.clone(), + end_ms: i64::MIN, + }; + let expired: Vec = instances + .range(from..oldest.clone()) + .map(|(selection, _)| selection.clone()) + .collect(); + for selection in expired { + instances.remove(&selection); } } diff --git a/data_plane/tests/per_sample.rs b/data_plane/tests/per_sample.rs index 90d8ecf6..40558d10 100644 --- a/data_plane/tests/per_sample.rs +++ b/data_plane/tests/per_sample.rs @@ -11,7 +11,6 @@ use asap_executor::summary_kernels::exact::ExactAccumulator; use asap_executor::values::Value; use data_plane::precompute::{Precompute, PrecomputeOptions, RawSample, SummaryInstance}; use data_plane::sds::Sds; -use data_plane::sds::Selection; const MINUTE: i64 = 60_000; @@ -116,9 +115,9 @@ fn per_sample_panes_equal_the_batch_build() { for definition in sds.definitions.values() { let mut end = 0; while end <= to { - let selection = Selection::ending_at(end); - let expected = batch.store.get(&definition.id, selection); - let actual = per_sample.store.get(&definition.id, selection); + let selection = definition.selection_ending_at(end); + let expected = batch.store.get(&definition.id, &selection); + let actual = per_sample.store.get(&definition.id, &selection); assert_eq!(actual.is_some(), expected.is_some(), "{example} pane {end}"); if let (Some(actual), Some(expected)) = (actual, expected) { for batch in actual.batches.iter().chain(&expected.batches) { @@ -154,14 +153,8 @@ fn per_sample_panes_apply_label_matchers() { let mut batch = Precompute::batch(&plan.dag.dag, &sds, options()).unwrap(); batch.ingest(&all); let (actual, expected) = ( - per_sample - .store - .get(&id, Selection::ending_at(20 * MINUTE)) - .unwrap(), - batch - .store - .get(&id, Selection::ending_at(20 * MINUTE)) - .unwrap(), + per_sample.store.get_pane(&id, 20 * MINUTE).unwrap(), + batch.store.get_pane(&id, 20 * MINUTE).unwrap(), ); assert_eq!(cells(&expected).len(), 2, "jobs 1 and 4 are instance 1"); assert_eq!(cells(&actual), cells(&expected)); @@ -189,9 +182,9 @@ fn a_resent_sample_is_dropped_not_failing_or_doubling_the_pane() { for definition in sds.definitions.values() { let mut end = 0; while end <= 3 * MINUTE { - let selection = Selection::ending_at(end); - let expected = batch.store.get(&definition.id, selection); - let actual = per_sample.store.get(&definition.id, selection); + let selection = definition.selection_ending_at(end); + let expected = batch.store.get(&definition.id, &selection); + let actual = per_sample.store.get(&definition.id, &selection); assert_eq!( actual.map(cells), expected.map(cells), diff --git a/data_plane/tests/precompute.rs b/data_plane/tests/precompute.rs index dd2ebc6d..21887d8a 100644 --- a/data_plane/tests/precompute.rs +++ b/data_plane/tests/precompute.rs @@ -6,7 +6,7 @@ use asap_executor::planner::workload::{PlanningWorkload, Query}; use asap_planner::InstallablePlan; use control_plane::plan_workload; use data_plane::precompute::{pane_end, Precompute, PrecomputeOptions, RawSample}; -use data_plane::sds::{Sds, Selection}; +use data_plane::sds::Sds; const MINUTE: i64 = 60_000; const PANE: i64 = 10 * MINUTE; @@ -57,25 +57,16 @@ fn builds_panes_after_the_first_as_the_watermark_passes() { // The first sample, at 1 min, opens pane (0, 10 min], which is skipped. engine.ingest(&samples("latency_ms", &["a", "b"], MINUTE, 35 * MINUTE)); assert_eq!( - engine - .store - .get(&id, Selection::ending_at(PANE)) - .map(|_| ()), + engine.store.get_pane(&id, PANE).map(|_| ()), None, "first pane" ); for end in [2 * PANE, 3 * PANE] { - let pane = engine - .store - .get(&id, Selection::ending_at(end)) - .expect("built pane"); + let pane = engine.store.get_pane(&id, end).expect("built pane"); assert_eq!(pane.num_rows(), 2, "one state per series"); } assert_eq!( - engine - .store - .get(&id, Selection::ending_at(4 * PANE)) - .map(|_| ()), + engine.store.get_pane(&id, 4 * PANE).map(|_| ()), None, "open pane" ); @@ -116,30 +107,9 @@ fn stores_empty_panes() { let id = sds.definitions.keys().next().unwrap().clone(); engine.ingest(&samples("latency_ms", &["a"], 0, 20 * MINUTE)); engine.ingest(&[sample("latency_ms", "a", 45 * MINUTE, 1.0)]); - assert_eq!( - engine - .store - .get(&id, Selection::ending_at(2 * PANE)) - .unwrap() - .num_rows(), - 1 - ); - assert_eq!( - engine - .store - .get(&id, Selection::ending_at(3 * PANE)) - .unwrap() - .num_rows(), - 0 - ); - assert_eq!( - engine - .store - .get(&id, Selection::ending_at(4 * PANE)) - .unwrap() - .num_rows(), - 0 - ); + assert_eq!(engine.store.get_pane(&id, 2 * PANE).unwrap().num_rows(), 1); + assert_eq!(engine.store.get_pane(&id, 3 * PANE).unwrap().num_rows(), 0); + assert_eq!(engine.store.get_pane(&id, 4 * PANE).unwrap().num_rows(), 0); } /// The plan's label matchers select which series a pane summarizes. @@ -154,14 +124,7 @@ fn applies_label_matchers() { let (mut engine, sds) = precompute(plan_workload(&workload, &common::inputs(false)).unwrap()); let id = sds.definitions.keys().next().unwrap().clone(); engine.ingest(&samples("latency_ms", &["a", "b"], 0, 25 * MINUTE)); - assert_eq!( - engine - .store - .get(&id, Selection::ending_at(2 * PANE)) - .unwrap() - .num_rows(), - 1 - ); + assert_eq!(engine.store.get_pane(&id, 2 * PANE).unwrap().num_rows(), 1); } /// Panes are `(end - width, end]`, like PromQL ranges: a sample at a @@ -192,23 +155,9 @@ fn engine_assigns_boundary_samples_to_the_pane_they_end() { sample("latency_ms", "a", 45 * MINUTE, 1.0), ]); // (10 min, 20 min]: series a and b, the one at its end included. - assert_eq!( - engine - .store - .get(&id, Selection::ending_at(2 * PANE)) - .unwrap() - .num_rows(), - 2 - ); + assert_eq!(engine.store.get_pane(&id, 2 * PANE).unwrap().num_rows(), 2); // (20 min, 30 min]: only c, one millisecond after the boundary. - assert_eq!( - engine - .store - .get(&id, Selection::ending_at(3 * PANE)) - .unwrap() - .num_rows(), - 1 - ); + assert_eq!(engine.store.get_pane(&id, 3 * PANE).unwrap().num_rows(), 1); } /// Samples of a metric no definition reads do not move the watermark, even @@ -259,18 +208,9 @@ fn watermark_jump_builds_only_retained_panes() { engine.stats.panes_skipped_expired, (30 * DAY - 90 * MINUTE) as u64 / PANE as u64 ); - assert!(engine - .store - .get(&id, Selection::ending_at(30 * DAY)) - .is_some()); - assert!(engine - .store - .get(&id, Selection::ending_at(30 * DAY - 7 * PANE)) - .is_none()); - assert!(engine - .store - .get(&id, Selection::ending_at(2 * PANE)) - .is_none()); + assert!(engine.store.get_pane(&id, 30 * DAY).is_some()); + assert!(engine.store.get_pane(&id, 30 * DAY - 7 * PANE).is_none()); + assert!(engine.store.get_pane(&id, 2 * PANE).is_none()); assert_eq!(engine.store.len(), 7); } @@ -312,30 +252,15 @@ fn evicts_panes_past_retention() { let (mut engine, sds) = precompute(common::plan("example4b", false)); let id = sds.definitions.keys().next().unwrap().clone(); engine.ingest(&samples("latency_ms", &["a"], 0, 25 * MINUTE)); - assert!(engine - .store - .get(&id, Selection::ending_at(2 * PANE)) - .is_some()); + assert!(engine.store.get_pane(&id, 2 * PANE).is_some()); // Watermark 100 min, retention 60 min plus one pane: the oldest pane // read by the evaluation at 90 min is (30, 40 min], so panes ending // before 40 min are evicted. engine.ingest(&[sample("latency_ms", "a", 100 * MINUTE, 1.0)]); - assert!(engine - .store - .get(&id, Selection::ending_at(2 * PANE)) - .is_none()); - assert!(engine - .store - .get(&id, Selection::ending_at(3 * PANE)) - .is_none()); - assert!(engine - .store - .get(&id, Selection::ending_at(4 * PANE)) - .is_some()); - assert!(engine - .store - .get(&id, Selection::ending_at(10 * PANE)) - .is_some()); + assert!(engine.store.get_pane(&id, 2 * PANE).is_none()); + assert!(engine.store.get_pane(&id, 3 * PANE).is_none()); + assert!(engine.store.get_pane(&id, 4 * PANE).is_some()); + assert!(engine.store.get_pane(&id, 10 * PANE).is_some()); assert_eq!(engine.store.len(), 7); } @@ -350,14 +275,7 @@ fn accepts_out_of_order_samples_within_lateness() { engine.ingest(&[sample("latency_ms", "b", 15 * MINUTE, 1.0)]); engine.ingest(&[sample("latency_ms", "a", 26 * MINUTE, 1.0)]); // Watermark 21 min: (10, 20 min] holds b's late sample. - assert_eq!( - engine - .store - .get(&id, Selection::ending_at(2 * PANE)) - .unwrap() - .num_rows(), - 1 - ); + assert_eq!(engine.store.get_pane(&id, 2 * PANE).unwrap().num_rows(), 1); assert_eq!(engine.stats.samples_dropped_late, 0); engine.ingest(&[sample("latency_ms", "c", 18 * MINUTE, 1.0)]); assert_eq!(engine.stats.samples_dropped_late, 1); @@ -377,14 +295,7 @@ fn stale_markers_are_not_samples() { sample("latency_ms", "b", 14 * MINUTE, stale), sample("latency_ms", "a", 25 * MINUTE, 1.0), ]); - assert_eq!( - engine - .store - .get(&id, Selection::ending_at(2 * PANE)) - .unwrap() - .num_rows(), - 1 - ); + assert_eq!(engine.store.get_pane(&id, 2 * PANE).unwrap().num_rows(), 1); } /// One series stamped ahead of the wall clock, though within @@ -400,10 +311,7 @@ fn series_ahead_of_the_clock_does_not_close_panes_early() { let now = base + 9 * MINUTE; engine.ingest_at(&[sample("latency_ms", "b", base + 13 * MINUTE, 1.0)], now); assert!( - engine - .store - .get(&id, Selection::ending_at(base + PANE)) - .is_none(), + engine.store.get_pane(&id, base + PANE).is_none(), "(10, 20 min] still open" ); engine.ingest_at(&[sample("latency_ms", "a", base + 9 * MINUTE, 1.0)], now); @@ -445,7 +353,7 @@ fn keeps_every_pane_the_newest_served_evaluation_reads() { engine.ingest(&samples("latency_ms", &["a"], MINUTE, 75 * MINUTE)); for end in (2..=7).map(|i| i * PANE) { assert!( - engine.store.get(&id, Selection::ending_at(end)).is_some(), + engine.store.get_pane(&id, end).is_some(), "pane ending at {end}" ); } @@ -463,7 +371,7 @@ fn keeps_one_pane_of_margin_beyond_the_retention() { engine.ingest(&samples("latency_ms", &["a"], MINUTE, 85 * MINUTE)); for end in (2..=8).map(|i| i * PANE) { assert!( - engine.store.get(&id, Selection::ending_at(end)).is_some(), + engine.store.get_pane(&id, end).is_some(), "pane ending at {end}" ); } @@ -511,26 +419,13 @@ fn barrier_closes_the_last_pane_of_a_stopped_series() { &samples("latency_ms", &["a"], MINUTE, 25 * MINUTE), 31 * MINUTE, ); - assert!(engine - .store - .get(&id, Selection::ending_at(2 * PANE)) - .is_some()); + assert!(engine.store.get_pane(&id, 2 * PANE).is_some()); assert!( - engine - .store - .get(&id, Selection::ending_at(3 * PANE)) - .is_none(), + engine.store.get_pane(&id, 3 * PANE).is_none(), "(20, 30 min] open" ); assert_eq!(engine.advance_watermark_at(3 * PANE, 31 * MINUTE), 3 * PANE); - assert_eq!( - engine - .store - .get(&id, Selection::ending_at(3 * PANE)) - .unwrap() - .num_rows(), - 1 - ); + assert_eq!(engine.store.get_pane(&id, 3 * PANE).unwrap().num_rows(), 1); } /// A repeated or lower barrier does nothing, and a barrier ahead of the @@ -547,15 +442,9 @@ fn barrier_is_monotonic_and_capped_at_the_wall_clock() { engine.advance_watermark_at(4 * PANE, 31 * MINUTE), 31 * MINUTE ); - assert!(engine - .store - .get(&id, Selection::ending_at(3 * PANE)) - .is_some()); + assert!(engine.store.get_pane(&id, 3 * PANE).is_some()); assert!( - engine - .store - .get(&id, Selection::ending_at(4 * PANE)) - .is_none(), + engine.store.get_pane(&id, 4 * PANE).is_none(), "capped at 31 min" ); let stats = engine.stats; diff --git a/data_plane/tests/query_run.rs b/data_plane/tests/query_run.rs index e8561d59..5f2c3bb0 100644 --- a/data_plane/tests/query_run.rs +++ b/data_plane/tests/query_run.rs @@ -14,7 +14,7 @@ use data_plane::precompute::SummaryInstance; use data_plane::query::{ BoundRawSampleWindow, IngestionProgress, QueryEngine, QueryRunStats, ResultSample, }; -use data_plane::sds::{DefinitionId, Sds, Selection, SummaryStore}; +use data_plane::sds::{DefinitionId, DefinitionKey, Sds, SummaryStore}; const MINUTE: i64 = 60_000; const P99: &str = "quantile_over_time(0.99, latency_ms[5m])"; @@ -23,7 +23,7 @@ const P99: &str = "quantile_over_time(0.99, latency_ms[5m])"; struct Fixture { engine: QueryEngine, store: SummaryStore, - definition: DefinitionId, + definition: DefinitionKey, } fn fixture(plan: &InstallablePlan) -> Fixture { @@ -36,8 +36,8 @@ fn fixture(plan: &InstallablePlan) -> Fixture { .definitions .values() .find(|d| d.built_at == StoredOutputBuild::QueryRun) - .map(|d| d.id.clone()) - .unwrap_or_else(|| DefinitionId(String::new())); + .map(|d| d.key()) + .unwrap_or_else(|| DefinitionKey::unconstrained(DefinitionId(String::new()))); Fixture { engine: QueryEngine::new(plan, &sds), store: SummaryStore::new(), @@ -50,11 +50,7 @@ impl Fixture { fn ends(&self) -> Vec { (0..=30) .map(|m| m * MINUTE) - .filter(|end| { - self.store - .get(&self.definition, Selection::ending_at(*end)) - .is_some() - }) + .filter(|end| self.store.get_pane(&self.definition, *end).is_some()) .map(|end| end / MINUTE) .collect() } diff --git a/data_plane/tests/sds.rs b/data_plane/tests/sds.rs index f4663089..dd6b54ca 100644 --- a/data_plane/tests/sds.rs +++ b/data_plane/tests/sds.rs @@ -1,9 +1,11 @@ mod common; use asap_executor::planner::ir::{NonASAPOp, Operator}; -use asap_planner::{InstallablePlan, StoredOutputBuild}; +use asap_planner::StoredOutputBuild; use common::plan; -use data_plane::sds::{DefinitionId, Sds, SdsError, Selection, SummaryInstanceKey, SummaryStore}; +use data_plane::sds::{ + DefinitionId, DefinitionKey, Sds, SdsError, Selection, SummaryInstanceKey, SummaryStore, +}; /// Example 4b's six ingestion-time KLL panes are one definition with a /// 10-minute pane; each output reads the pane `i * 10min` before evaluation, @@ -19,7 +21,7 @@ fn example4b_panes_share_one_definition() { .outputs .values() .map(|(id, binding)| { - assert_eq!(id, &definition.id); + assert_eq!(id, &definition.key()); binding.offset_ms }) .collect(); @@ -32,7 +34,7 @@ fn example4b_panes_share_one_definition() { .dag .nodes .iter() - .find(|n| n.id as u64 == definition.raw_input()) + .find(|n| Some(n.id as u64) == definition.raw_input()) .unwrap(); assert!(matches!( raw.payload, @@ -59,61 +61,46 @@ fn query_time_plans_have_no_definitions() { assert_eq!(Sds::from_plan(&plan).unwrap(), Sds::default()); } -/// `plan` with every node id shifted by 100. -fn renumbered(plan: &InstallablePlan) -> InstallablePlan { - let mut plan = plan.clone(); - let renumber = |id: &usize| id + 100; - let dag = &mut plan.dag.dag; - for node in &mut dag.nodes { - node.id = renumber(&node.id); - node.payload = node.payload.map_children(renumber); - } - for edge in &mut dag.edges { - edge.producer = renumber(&edge.producer); - edge.consumer = renumber(&edge.consumer); - } - dag.roots = dag.roots.iter().map(renumber).collect(); - for output in &mut plan.stored_outputs { - output.node_id = renumber(&output.node_id); - output.pane.time_range_node_id = renumber(&output.pane.time_range_node_id); - output.pane.scan_node_id = renumber(&output.pane.scan_node_id); - } - plan -} - -/// Identity depends on what is computed, not on how nodes are numbered. +/// A definition's identity is the plan's definition id, the same for every +/// output whose selection differs only in time. #[test] -fn identity_ignores_node_numbering() { +fn identity_is_the_plans_definition_id() { let plan = plan("example4b", false); - let ids = |sds: Sds| sds.definitions.into_keys().collect::>(); - assert_eq!( - ids(Sds::from_plan(&renumbered(&plan)).unwrap()), - ids(Sds::from_plan(&plan).unwrap()) - ); + let sds = Sds::from_plan(&plan).unwrap(); + let definition = sds.definitions.values().next().unwrap(); + for output in &plan.stored_outputs { + assert_eq!(definition.id.0, output.coverage.definition_id); + } + assert_eq!(&*definition.labels, "", "no label constraints"); } -/// The same computation over another metric is another definition. +/// Outputs with the same definition id but other label constraints in +/// their selection are other definitions, and their summary instances other +/// keys: the plan's definition id leaves the selection out. #[test] -fn identity_depends_on_the_source() { - let plan = plan("example4b", false); - let other: InstallablePlan = serde_json::from_str( - &serde_json::to_string(&plan) - .unwrap() - .replace("latency_ms", "other"), - ) - .unwrap(); - assert_ne!( - Sds::from_plan(&other) - .unwrap() - .definitions - .into_keys() - .next(), - Sds::from_plan(&plan) - .unwrap() - .definitions - .into_keys() - .next() - ); +fn label_constraints_separate_definitions_of_one_id() { + let mut json = serde_json::to_value(common::workload("example4b")).unwrap(); + let queries = json["query_workload"]["repeating_queries"] + .as_array_mut() + .unwrap(); + let mut other = queries[0].clone(); + queries[0]["query"] = "quantile_over_time(0.99, latency_ms{job=\"a\"}[60m])".into(); + other["query"] = "quantile_over_time(0.99, latency_ms{job=\"b\"}[60m])".into(); + queries.push(other); + let workload = serde_json::from_value(json).unwrap(); + let plan = control_plane::plan_workload(&workload, &common::inputs(false)).unwrap(); + let sds = Sds::from_plan(&plan).unwrap(); + assert_eq!(sds.definitions.len(), 2, "{:?}", sds.definitions.keys()); + let keys: Vec<&DefinitionKey> = sds.definitions.keys().collect(); + assert_eq!(keys[0].id, keys[1].id, "one definition id"); + assert_ne!(keys[0].labels, keys[1].labels); + let mut store = SummaryStore::new(); + for (i, key) in keys.iter().enumerate() { + let selection = sds.definitions[*key].selection_ending_at(600_000); + store.insert(SummaryInstanceKey::new(key.id.clone(), selection), i); + } + assert_eq!(store.get_pane(keys[0], 600_000), Some(&0)); + assert_eq!(store.get_pane(keys[1], 600_000), Some(&1)); } /// Stored outputs built in a `QueryRun` are stored summaries too: their @@ -141,10 +128,7 @@ fn ingestion_wins_when_outputs_share_a_definition() { let sds = Sds::from_plan(&plan).unwrap(); let definition = sds.definitions.values().next().unwrap(); assert_eq!(definition.built_at, StoredOutputBuild::IngestionRun); - assert_eq!( - sds.ingestion_build_order().unwrap(), - vec![definition.id.clone()] - ); + assert_eq!(sds.ingestion_build_order().unwrap(), vec![definition.key()]); } /// Every stored output reads raw samples today, so no definition has stored @@ -154,7 +138,7 @@ fn build_order_follows_stored_inputs() { let mut sds = Sds::from_plan(&plan("example4b", false)).unwrap(); let definition = sds.definitions.values_mut().next().unwrap(); assert!(definition.stored_inputs.is_empty()); - let id = definition.id.clone(); + let id = definition.key(); definition.stored_inputs = vec![id.clone()]; assert_eq!(sds.ingestion_build_order(), Err(SdsError::Cycle(id))); } @@ -163,33 +147,27 @@ fn build_order_follows_stored_inputs() { /// them. #[test] fn store_finds_and_evicts_panes() { - let (a, b) = (DefinitionId("a".into()), DefinitionId("b".into())); + let (a, b) = ( + DefinitionKey::unconstrained(DefinitionId("a".into())), + DefinitionKey::unconstrained(DefinitionId("b".into())), + ); let mut store = SummaryStore::new(); for end in [60, 120, 180] { store.insert( - SummaryInstanceKey::new(a.clone(), Selection::ending_at(end)), + SummaryInstanceKey::new(a.id.clone(), Selection::ending_at(end)), format!("a{end}"), ); } store.insert( - SummaryInstanceKey::new(b.clone(), Selection::ending_at(60)), + SummaryInstanceKey::new(b.id.clone(), Selection::ending_at(60)), "b60".to_string(), ); - assert_eq!( - store.get(&a, Selection::ending_at(120)).map(String::as_str), - Some("a120") - ); - assert_eq!(store.get(&a, Selection::ending_at(90)), None); - store.evict_before(&a, Selection::ending_at(180)); - assert_eq!(store.get(&a, Selection::ending_at(120)), None); - assert_eq!( - store.get(&a, Selection::ending_at(180)).map(String::as_str), - Some("a180") - ); - assert_eq!( - store.get(&b, Selection::ending_at(60)).map(String::as_str), - Some("b60") - ); + assert_eq!(store.get_pane(&a, 120).map(String::as_str), Some("a120")); + assert_eq!(store.get_pane(&a, 90), None); + store.evict_before(&a.id, &Selection::ending_at(180)); + assert_eq!(store.get_pane(&a, 120), None); + assert_eq!(store.get_pane(&a, 180).map(String::as_str), Some("a180")); + assert_eq!(store.get_pane(&b, 60).map(String::as_str), Some("b60")); assert_eq!(store.len(), 2); } @@ -203,7 +181,7 @@ fn renamed( definition.id = DefinitionId(name.into()); definition.stored_inputs = stored_inputs .iter() - .map(|input| DefinitionId(input.to_string())) + .map(|input| DefinitionKey::unconstrained(DefinitionId(input.to_string()))) .collect(); definition } @@ -220,13 +198,13 @@ fn build_order_puts_inputs_first() { renamed(template, "b", &["c"]), renamed(template, "c", &[]), ] { - sds.definitions.insert(definition.id.clone(), definition); + sds.definitions.insert(definition.key(), definition); } let order: Vec = sds .ingestion_build_order() .unwrap() .into_iter() - .map(|id| id.0) + .map(|key| key.id.0) .collect(); assert_eq!(order, ["c", "b", "a"]); } @@ -241,13 +219,13 @@ fn build_order_refuses_ingestion_reading_query_run() { query_run.built_at = StoredOutputBuild::QueryRun; let mut sds = Sds::default(); for definition in [renamed(template, "i", &["q"]), query_run] { - sds.definitions.insert(definition.id.clone(), definition); + sds.definitions.insert(definition.key(), definition); } assert_eq!( sds.ingestion_build_order(), Err(SdsError::IngestionReadsQueryRun { - definition: DefinitionId("i".into()), - input: DefinitionId("q".into()), + definition: DefinitionKey::unconstrained(DefinitionId("i".into())), + input: DefinitionKey::unconstrained(DefinitionId("q".into())), }) ); } @@ -269,7 +247,7 @@ fn query_run_outputs_share_one_definition() { let mut offsets: Vec = sds .outputs .values() - .filter(|(id, _)| id == &definition.id) + .filter(|(id, _)| id == &definition.key()) .map(|(_, binding)| binding.offset_ms) .collect(); offsets.sort(); From 1de950a64283c33be01df536066db2e730c95f07 Mon Sep 17 00:00:00 2001 From: zzylol Date: Wed, 21 Oct 2026 09:57:58 +0000 Subject: [PATCH 2/5] feat(deps): pin ASAPPlanner main ef52e175; plan with ignore_accuracy_guarantees - Both Planner crates move to ef52e175 (sketchlib planner-pin/0.4.0-plus: interpolated and signed DDSketch quantiles, top-k with series labels, Stage 3's executor compile check, joint search, shared stored summaries). - control_plane: --plan-selection-settings (the Planner's PlanSelectionSettings JSON) and --ignore-accuracy-guarantees; prints the queries accepted without a guarantee. --measured-operation-costs reads MeasuredOperationCosts. - serde_json float_roundtrip: a plan file reloads to the plan it was written from (its predicted usage is floats). - data_plane/tests/planner_executor_answers.rs: the data plane answers as the Planner's executor for DDSketch panes over zero and signed samples, interpolated quantiles, top-k labels, counter top-k without guarantees, and jointly planned shared states. --- Cargo.lock | 40 +- Cargo.toml | 8 +- README.md | 15 +- control_plane/src/main.rs | 28 +- control_plane/tests/plan_workload.rs | 34 + data_plane/tests/per_sample.rs | 6 +- data_plane/tests/planner_executor_answers.rs | 692 +++++++++++++++++++ 7 files changed, 782 insertions(+), 41 deletions(-) create mode 100644 data_plane/tests/planner_executor_answers.rs diff --git a/Cargo.lock b/Cargo.lock index 468457e8..47124d8a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -252,7 +252,7 @@ dependencies = [ [[package]] name = "asap-executor" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=4111e62014aa1944fa546075a4b161fe390a2439#4111e62014aa1944fa546075a4b161fe390a2439" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=ef52e175d75f074b2634e6e332b4ec47f8d02fc0#ef52e175d75f074b2634e6e332b4ec47f8d02fc0" dependencies = [ "ahash", "arrow", @@ -270,7 +270,7 @@ dependencies = [ [[package]] name = "asap-frontend-common" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=4111e62014aa1944fa546075a4b161fe390a2439#4111e62014aa1944fa546075a4b161fe390a2439" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=ef52e175d75f074b2634e6e332b4ec47f8d02fc0#ef52e175d75f074b2634e6e332b4ec47f8d02fc0" dependencies = [ "asap-types", "serde", @@ -280,7 +280,7 @@ dependencies = [ [[package]] name = "asap-frontend-promql" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=4111e62014aa1944fa546075a4b161fe390a2439#4111e62014aa1944fa546075a4b161fe390a2439" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=ef52e175d75f074b2634e6e332b4ec47f8d02fc0#ef52e175d75f074b2634e6e332b4ec47f8d02fc0" dependencies = [ "asap-frontend-common", "asap-types", @@ -290,7 +290,7 @@ dependencies = [ [[package]] name = "asap-logical-asap-optimization" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=4111e62014aa1944fa546075a4b161fe390a2439#4111e62014aa1944fa546075a4b161fe390a2439" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=ef52e175d75f074b2634e6e332b4ec47f8d02fc0#ef52e175d75f074b2634e6e332b4ec47f8d02fc0" dependencies = [ "asap-types", "asap_sketchlib", @@ -302,7 +302,7 @@ dependencies = [ [[package]] name = "asap-physical-asap-optimization" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=4111e62014aa1944fa546075a4b161fe390a2439#4111e62014aa1944fa546075a4b161fe390a2439" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=ef52e175d75f074b2634e6e332b4ec47f8d02fc0#ef52e175d75f074b2634e6e332b4ec47f8d02fc0" dependencies = [ "asap-logical-asap-optimization", "asap-types", @@ -312,7 +312,7 @@ dependencies = [ [[package]] name = "asap-plan-selection" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=4111e62014aa1944fa546075a4b161fe390a2439#4111e62014aa1944fa546075a4b161fe390a2439" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=ef52e175d75f074b2634e6e332b4ec47f8d02fc0#ef52e175d75f074b2634e6e332b4ec47f8d02fc0" dependencies = [ "asap-logical-asap-optimization", "asap-physical-asap-optimization", @@ -326,7 +326,7 @@ dependencies = [ [[package]] name = "asap-planner" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=4111e62014aa1944fa546075a4b161fe390a2439#4111e62014aa1944fa546075a4b161fe390a2439" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=ef52e175d75f074b2634e6e332b4ec47f8d02fc0#ef52e175d75f074b2634e6e332b4ec47f8d02fc0" dependencies = [ "asap-executor", "asap-frontend-promql", @@ -341,7 +341,7 @@ dependencies = [ [[package]] name = "asap-types" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=4111e62014aa1944fa546075a4b161fe390a2439#4111e62014aa1944fa546075a4b161fe390a2439" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=ef52e175d75f074b2634e6e332b4ec47f8d02fc0#ef52e175d75f074b2634e6e332b4ec47f8d02fc0" dependencies = [ "serde", "serde_json", @@ -350,20 +350,15 @@ dependencies = [ [[package]] name = "asap_sketchlib" -version = "0.3.0" -source = "git+https://github.com/ProjectASAP/asap_sketchlib?rev=3f44cf1c2d823b844c17796f24408e03863149ee#3f44cf1c2d823b844c17796f24408e03863149ee" +version = "0.4.0" +source = "git+https://github.com/ProjectASAP/asap_sketchlib?rev=63e565506426020c505ef0b84e0c91f450d09e7c#63e565506426020c505ef0b84e0c91f450d09e7c" dependencies = [ - "bincode", - "bytes", - "prost", "rand 0.9.5", "rmp-serde", "serde", - "serde-big-array", "serde_bytes", "smallvec", "twox-hash", - "xxhash-rust", ] [[package]] @@ -1883,15 +1878,6 @@ dependencies = [ "serde_derive", ] -[[package]] -name = "serde-big-array" -version = "0.5.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "11fc7cc2c76d73e0f27ee52abbd64eec84d46f370c88371120433196934e4b7f" -dependencies = [ - "serde", -] - [[package]] name = "serde_bytes" version = "0.11.19" @@ -2711,12 +2697,6 @@ version = "0.6.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3ad82d2a33cdc9674dc7465672f271e096168fcdbe0f799d9e6db8c5892679dc" -[[package]] -name = "xxhash-rust" -version = "0.8.18" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "aee1b19627c7c60102ab80d3a9cbe18de90bfe03bfa6c3715447681f0e8c8af6" - [[package]] name = "yoke" version = "0.8.3" diff --git a/Cargo.toml b/Cargo.toml index c1950b1e..5d668476 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -10,11 +10,13 @@ version = "0.1.0" # The backend depends on two Planner crates, at one immutable revision: # `asap-planner` to plan and `asap-executor` to execute. Planner IR types # are reached through `asap_executor::planner`. -asap-planner = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "4111e62014aa1944fa546075a4b161fe390a2439", default-features = false, features = ["promql"] } -asap-executor = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "4111e62014aa1944fa546075a4b161fe390a2439" } +asap-planner = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "ef52e175d75f074b2634e6e332b4ec47f8d02fc0", default-features = false, features = ["promql"] } +asap-executor = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "ef52e175d75f074b2634e6e332b4ec47f8d02fc0" } serde = { version = "1.0", features = ["derive"] } -serde_json = "1.0" +# `float_roundtrip`: a plan file reloads to the plan it was written from +# (its predicted usage and prices are floats). +serde_json = { version = "1.0", features = ["float_roundtrip"] } thiserror = "1.0" anyhow = "1.0" tracing = "0.1" diff --git a/README.md b/README.md index c64ee10b..ad474f8b 100644 --- a/README.md +++ b/README.md @@ -99,9 +99,18 @@ cargo run -p data_plane -- --plan plan.json --prometheus-url http://localhost:90 `DeploymentCapabilities`: every summary the Planner can build, raw data kept in Prometheus, and stored outputs built in a `QueryRun` (`MatAtQueryKeptAcrossEvaluations`). For example `{"raw_data_retained": false}` makes the Planner build every input at -ingestion time. `--measured-operation-costs costs.json` gives Stage 3 the -deployment's measured sketch operation costs; otherwise it prices every node -with cost model 1 and illustrative unit costs. +ingestion time. + +`--plan-selection-settings settings.json` replaces the Planner's default +`PlanSelectionSettings` (Stage 3's cost model, resource speed, measured +operation costs and `ignore_accuracy_guarantees`); an omitted field keeps its +default, cost model 1 with illustrative unit costs. +`--measured-operation-costs costs.json` gives Stage 3 the deployment's +measured operation costs (`MeasuredOperationCosts`). +`--ignore-accuracy-guarantees` accepts every query output whatever its +accuracy guarantee, so candidates are compared on cost alone; the plan's +`selection.queries_accepted_without_guarantee` lists the queries accepted +only because of it. `--forward-timeout-seconds` bounds each forwarded query (default 120, matching Prometheus's default query timeout); a query that runs longer gets 502. diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs index 6eefba92..76b77a91 100644 --- a/control_plane/src/main.rs +++ b/control_plane/src/main.rs @@ -15,11 +15,23 @@ struct Args { /// backend's data plane executes and keeps. #[arg(long)] deployment_capabilities: Option, - /// Planner `MeasuredOperationCost` list (JSON) that Stage 3 prices - /// sketch operations from; by default every node is priced from the + /// Planner `PlanSelectionSettings` (JSON): Stage 3's cost model, + /// resource speed, measured operation costs and whether accuracy + /// guarantees are ignored; an omitted field takes the Planner's + /// default. + #[arg(long)] + plan_selection_settings: Option, + /// Planner `MeasuredOperationCosts` (JSON, or the earlier list of + /// `MeasuredOperationCost`) that Stage 3 prices operators from, in place + /// of the settings' table; by default every node is priced from the /// Planner's formulas. #[arg(long)] measured_operation_costs: Option, + /// Accept every query output regardless of its accuracy guarantee + /// (`PlanSelectionSettings::ignore_accuracy_guarantees`): candidates are + /// compared on cost alone; executability checks still apply. + #[arg(long)] + ignore_accuracy_guarantees: bool, /// Where to write the plan the data plane installs. #[arg(long)] out: PathBuf, @@ -41,11 +53,23 @@ fn main() -> anyhow::Result<()> { if let Some(path) = &args.deployment_capabilities { inputs.deployment_capabilities = read_json(path)?; } + if let Some(path) = &args.plan_selection_settings { + inputs.plan_selection = read_json(path)?; + } if let Some(path) = &args.measured_operation_costs { inputs.plan_selection.measured_operation_costs = read_json(path)?; } + inputs.plan_selection.ignore_accuracy_guarantees |= args.ignore_accuracy_guarantees; let plan = plan_workload(&workload, &inputs)?; save_plan(&plan, &args.out)?; + if let Some(selection) = &plan.selection { + if !selection.queries_accepted_without_guarantee.is_empty() { + println!( + "queries accepted without an accuracy guarantee (workload indices): {:?}", + selection.queries_accepted_without_guarantee + ); + } + } println!( "wrote {} ({} queries, {} nodes)", args.out.display(), diff --git a/control_plane/tests/plan_workload.rs b/control_plane/tests/plan_workload.rs index ac8336af..a9bbbd2c 100644 --- a/control_plane/tests/plan_workload.rs +++ b/control_plane/tests/plan_workload.rs @@ -130,3 +130,37 @@ fn refuses_a_workload_with_a_query_it_cannot_plan() { let error = plan_workload(&workload, &inputs).unwrap_err(); assert!(matches!(error, PlanError::Planning(_)), "{error}"); } + +/// The MVP benchmark's p99 query over a gauge not declared non-negative: +/// without raw data, every candidate fails a check (the exact ones scan raw +/// samples at query time, the merged DDSketch panes carry no bound through +/// `avg`), so planning fails. With `ignore_accuracy_guarantees`, loaded from +/// the Planner's settings JSON as `--plan-selection-settings` reads it, the +/// cheapest sketch plan is selected and the report marks the query as +/// accepted without a guarantee. +#[test] +fn ignoring_accuracy_guarantees_plans_the_undeclared_p99_query() { + let workload: PlanningWorkload = serde_json::from_str( + r#"{"query_workload": {"language": "promql", "repeating_queries": [{ + "query": "avg by (region, service, job) (quantile_over_time(0.99, data[100s]))", + "demand": {"fixed_interval": 1000}, + "requirements": {"accuracy": {"max_error": 0.05, "failure_probability": 0.3, + "accuracy_scope": "whole_result"}}, + "predictability": {"predictable": {"known_at": null}}}]}, + "data_workload": {"arrival": {"ingesting": {"rows_per_second": 10000.0}}, + "sources": {"data": {"series_count": 1000, "metric_type": "gauge", + "sample_interval": 100}}}}"#, + ) + .unwrap(); + let mut inputs = DeploymentInputs::default(); + inputs.deployment_capabilities.raw_data_retained = false; + let error = plan_workload(&workload, &inputs).unwrap_err(); + assert!(matches!(error, PlanError::Planning(_)), "{error}"); + + inputs.plan_selection = + serde_json::from_str(r#"{"ignore_accuracy_guarantees": true}"#).unwrap(); + let plan = plan_workload(&workload, &inputs).unwrap(); + assert!(!plan.stored_outputs.is_empty(), "panes built at ingestion"); + let selection = plan.selection.as_ref().unwrap(); + assert_eq!(selection.queries_accepted_without_guarantee, [0]); +} diff --git a/data_plane/tests/per_sample.rs b/data_plane/tests/per_sample.rs index 40558d10..45330585 100644 --- a/data_plane/tests/per_sample.rs +++ b/data_plane/tests/per_sample.rs @@ -123,7 +123,7 @@ fn per_sample_panes_equal_the_batch_build() { for batch in actual.batches.iter().chain(&expected.batches) { assert_eq!(**batch.schema(), definition.output_schema); } - assert_eq!(cells(&actual), cells(&expected), "{example} pane {end}"); + assert_eq!(cells(actual), cells(expected), "{example} pane {end}"); compared += usize::from(expected.num_rows() > 0); } end += definition.pane_ms; @@ -156,8 +156,8 @@ fn per_sample_panes_apply_label_matchers() { per_sample.store.get_pane(&id, 20 * MINUTE).unwrap(), batch.store.get_pane(&id, 20 * MINUTE).unwrap(), ); - assert_eq!(cells(&expected).len(), 2, "jobs 1 and 4 are instance 1"); - assert_eq!(cells(&actual), cells(&expected)); + assert_eq!(cells(expected).len(), 2, "jobs 1 and 4 are instance 1"); + assert_eq!(cells(actual), cells(expected)); } /// A resent sample (a Remote Write retry, for instance) is not diff --git a/data_plane/tests/planner_executor_answers.rs b/data_plane/tests/planner_executor_answers.rs new file mode 100644 index 00000000..9d0b0bd5 --- /dev/null +++ b/data_plane/tests/planner_executor_answers.rs @@ -0,0 +1,692 @@ +//! The data plane answers what the Planner's executor answers, for the +//! plans the Planner selects without raw samples: DDSketch panes over zero +//! and signed samples (interpolated quantiles), top-k answers that keep the +//! chosen series' labels, counter top-k planned with accuracy guarantees +//! ignored, and queries planned jointly over one shared stored state. +//! +//! Each test plans with the backend's control plane, installs the plan with +//! `--no-forward` (every answer computed from the plan), writes the samples +//! by Remote Write, closes the panes with the event-time barrier, and +//! checks the answers against the exact ones, as the Planner's own runtime +//! tests do. + +mod common; + +use std::collections::BTreeMap; +use std::sync::Arc; + +use asap_planner::{InstallablePlan, PlanningWorkload}; +use axum::body::Body; +use axum::http::{Request, StatusCode}; +use axum::Router; +use control_plane::plan_workload; +use data_plane::engine::EngineOptions; +use data_plane::remote_write::{encode, Label, Sample, TimeSeries, WriteRequest}; +use data_plane::{router, DataPlane, EXECUTION_HEADER}; +use http_body_util::BodyExt; +use serde_json::Value; +use tower::ServiceExt; + +/// The evaluation time of every test. +const EVALUATION_MS: i64 = 3_600_000; + +/// Labels including `__name__`. +type Labels = BTreeMap; + +/// One raw sample: its series' labels, its time and value. +struct Point { + labels: Labels, + timestamp_ms: i64, + value: f64, +} + +/// The selected plan of `workload` for a deployment that keeps no raw +/// samples, with accuracy guarantees ignored or not. +fn plan_without_raw_samples( + workload: &PlanningWorkload, + ignore_accuracy_guarantees: bool, +) -> InstallablePlan { + let mut inputs = common::inputs(false); + inputs.plan_selection.ignore_accuracy_guarantees = ignore_accuracy_guarantees; + plan_workload(workload, &inputs).unwrap_or_else(|e| panic!("{e}")) +} + +/// A data plane running `plan` with `--no-forward`, its panes built from +/// `points` written by Remote Write and closed by the barrier at +/// `EVALUATION_MS`. +async fn run(plan: &InstallablePlan, points: &[Point]) -> (Router, Arc) { + let options = EngineOptions { + no_forward: true, + ..EngineOptions::default() + }; + let state = Arc::new( + DataPlane::with_options( + serde_json::to_string(plan).unwrap(), + // Never asked: nothing is forwarded and no raw sample is read. + "http://127.0.0.1:1".into(), + options, + ) + .unwrap(), + ); + let app = router(state.clone()); + let mut by_series: BTreeMap<&Labels, Vec> = BTreeMap::new(); + for point in points { + by_series.entry(&point.labels).or_default().push(Sample { + timestamp: point.timestamp_ms, + value: point.value, + }); + } + let timeseries = by_series + .into_iter() + .map(|(labels, mut samples)| { + samples.sort_by_key(|s| s.timestamp); + TimeSeries { + labels: labels + .iter() + .map(|(name, value)| Label { + name: name.clone(), + value: value.clone(), + }) + .collect(), + samples, + } + }) + .collect(); + let written = app + .clone() + .oneshot( + Request::post("/api/v1/write") + .body(Body::from(encode(&WriteRequest { timeseries }))) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(written.status(), StatusCode::NO_CONTENT); + for _ in 0..1000 { + if state.engine.queued_samples() == 0 { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + let barrier = app + .clone() + .oneshot( + Request::post(format!( + "/api/v1/precompute/watermark?time_ms={EVALUATION_MS}" + )) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(barrier.status(), StatusCode::OK); + (app, state) +} + +/// The answer of `query` at `EVALUATION_MS`: each result series' labels and +/// value, sorted by labels. Panics unless the plan answered it. +async fn answer(app: &Router, query: &str) -> Vec<(Labels, f64)> { + let uri = format!( + "/api/v1/query?{}", + form_urlencoded::Serializer::new(String::new()) + .append_pair("query", query) + .append_pair("time", &(EVALUATION_MS as f64 / 1000.0).to_string()) + .finish() + ); + let response = app + .clone() + .oneshot(Request::get(uri).body(Body::empty()).unwrap()) + .await + .unwrap(); + let status = response.status(); + let execution = response + .headers() + .get(EXECUTION_HEADER) + .map(|v| v.to_str().unwrap().to_string()); + let body = response.into_body().collect().await.unwrap().to_bytes(); + let body: Value = serde_json::from_slice(&body).unwrap(); + assert_eq!(status, StatusCode::OK, "{query}: {body}"); + assert_eq!(execution.as_deref(), Some("plan"), "{query}: {body}"); + let mut rows: Vec<(Labels, f64)> = body["data"]["result"] + .as_array() + .unwrap() + .iter() + .map(|series| { + let labels = series["metric"] + .as_object() + .unwrap() + .iter() + .map(|(k, v)| (k.clone(), v.as_str().unwrap().to_string())) + .collect(); + let value = series["value"][1].as_str().unwrap().parse().unwrap(); + (labels, value) + }) + .collect(); + rows.sort_by(|a, b| a.0.cmp(&b.0)); + rows +} + +fn labels(pairs: &[(&str, &str)]) -> Labels { + pairs + .iter() + .map(|(k, v)| (k.to_string(), v.to_string())) + .collect() +} + +/// The MVP benchmark's p99 query with `outer` over the per-series p99s of a +/// gauge over 100 s, per (region, service, job), every second, within 0.05 +/// at failure probability 0.3, over 1,000 `data` series sampled every +/// 100 ms, declared non-negative or not. +fn p99_workload(outer: &str, values_non_negative: bool) -> PlanningWorkload { + quantile_workload(outer, 0.99, values_non_negative) +} + +/// [`p99_workload`] with the quantile `q`. +fn quantile_workload(outer: &str, q: f64, values_non_negative: bool) -> PlanningWorkload { + serde_json::from_str(&format!( + r#"{{"query_workload": {{"language": "promql", "repeating_queries": [{{ + "query": "{outer} by (region, service, job) (quantile_over_time({q}, data[100s]))", + "demand": {{"fixed_interval": 1000}}, + "requirements": {{"accuracy": {{"max_error": 0.05, "failure_probability": 0.3, + "accuracy_scope": "whole_result"}}}}, + "predictability": {{"predictable": {{"known_at": null}}}}}}]}}, + "data_workload": {{"arrival": {{"ingesting": {{"rows_per_second": 10000.0}}}}, + "sources": {{"data": {{"series_count": 1000, "metric_type": "gauge", + "sample_interval": 100, "values_non_negative": {values_non_negative}}}}}}}}}"# + )) + .unwrap() +} + +/// Each (region, service, job) group's per-series windows +/// `(EVALUATION_MS - 100 s, EVALUATION_MS]`. +type Windows = BTreeMap<(String, String, String), Vec>>; + +/// `data` samples of series `0..series`, `value(series, step)` every +/// `interval_ms` over the `steps` steps ending just before `EVALUATION_MS`, +/// two series per (region, service, job) group; and each group's windows. +fn data_samples( + series: usize, + steps: i64, + interval_ms: i64, + value: impl Fn(usize, i64) -> f64, +) -> (Vec, Windows) { + let mut points = Vec::new(); + let mut windows: Windows = BTreeMap::new(); + for s in 0..series { + let (region, service, job) = (format!("r{}", s % 2), format!("s{}", s / 4), "j".into()); + let series_labels = labels(&[ + ("__name__", "data"), + ("region", ®ion), + ("service", &service), + ("job", "j"), + ("instance", &format!("i{s}")), + ]); + let mut window = Vec::new(); + for step in 0..steps { + let timestamp_ms = EVALUATION_MS - steps * interval_ms + 50 + step * interval_ms; + let v = value(s, step); + points.push(Point { + labels: series_labels.clone(), + timestamp_ms, + value: v, + }); + if timestamp_ms > EVALUATION_MS - 100_000 { + window.push(v); + } + } + windows + .entry((region, service, job)) + .or_default() + .push(window); + } + (points, windows) +} + +/// PromQL's `quantile_over_time(q, ...)` of `window`: linear interpolation +/// at rank `q × (n − 1)`. +fn interpolated_quantile(q: f64, window: &[f64]) -> f64 { + let mut sorted = window.to_vec(); + sorted.sort_by(f64::total_cmp); + let rank = q * (sorted.len() - 1) as f64; + let (low, high) = (rank.floor() as usize, rank.ceil() as usize); + sorted[low] + (sorted[high] - sorted[low]) * (rank - low as f64) +} + +/// Every group's answer is within the relative error 0.05 of `combine` of +/// its series' exact p99s. +fn assert_p99s_within_alpha( + answers: &[(Labels, f64)], + windows: &Windows, + combine: impl Fn(&[f64]) -> f64, +) { + assert_eq!(answers.len(), windows.len(), "{answers:?}"); + for (labels, value) in answers { + let key = ( + labels["region"].clone(), + labels["service"].clone(), + labels["job"].clone(), + ); + let p99s: Vec = windows[&key] + .iter() + .map(|window| interpolated_quantile(0.99, window)) + .collect(); + let exact = combine(&p99s); + assert!( + (value - exact).abs() <= 0.05 * exact.abs(), + "{labels:?}: {value} vs exact {exact}" + ); + } +} + +fn average(values: &[f64]) -> f64 { + values.iter().sum::() / values.len() as f64 +} + +/// A non-negative gauge that reports exactly 0 is summarized by the +/// selected DDSketch panes (zero has its own bucket): each series' window +/// holds a third zeros, and every group's average of p99s is within 0.05 of +/// the exact average of PromQL's interpolated p99s. +#[tokio::test] +async fn ddsketch_panes_over_a_gauge_reporting_zero_answer_within_alpha() { + let query = "avg by (region, service, job) (quantile_over_time(0.99, data[100s]))"; + let plan = plan_without_raw_samples(&p99_workload("avg", true), false); + assert!(!plan.stored_outputs.is_empty(), "panes built at ingestion"); + let (points, windows) = data_samples(8, 1_100, 100, |s, step| { + (s + 1) as f64 * [0.0, 3.0, 5.0][(step % 3) as usize] + }); + let (app, _) = run(&plan, &points).await; + assert_p99s_within_alpha(&answer(&app, query).await, &windows, average); +} + +/// Signed samples (negative, zero and positive) are summarized by the +/// selected DDSketch panes through their negative store and zero bucket: +/// the `max` of the per-series p99s is within 0.05 of the exact value, +/// including groups whose samples are all negative. +#[tokio::test] +async fn ddsketch_panes_over_signed_samples_answer_within_alpha() { + let query = "max by (region, service, job) (quantile_over_time(0.99, data[100s]))"; + let plan = plan_without_raw_samples(&p99_workload("max", false), false); + assert!(!plan.stored_outputs.is_empty(), "panes built at ingestion"); + // Series 0..4 cross zero; series 4..8 are all negative. + let (points, windows) = data_samples(8, 1_100, 100, |s, step| { + let scale = (s + 1) as f64; + if s < 4 { + scale * [-7.0, -3.0, 0.0, 2.0, 4.0][(step % 5) as usize] + } else { + scale * [-9.0, -8.0, -5.0, -1.0][(step % 4) as usize] + } + }); + let (app, _) = run(&plan, &points).await; + let max = |p99s: &[f64]| p99s.iter().copied().fold(f64::NEG_INFINITY, f64::max); + assert_p99s_within_alpha(&answer(&app, query).await, &windows, max); +} + +/// A quantile of few samples interpolates between the two nearest, as +/// PromQL does: each series' window holds 1..=10 scaled by the series, so +/// its median is 5.5 × scale (rank 4.5), not a sample (5 or 6 × scale). +/// Every group's average of medians is within 0.05 of the exact one. +#[tokio::test] +async fn ddsketch_quantiles_interpolate_between_samples() { + let query = "avg by (region, service, job) (quantile_over_time(0.5, data[100s]))"; + let plan = plan_without_raw_samples(&quantile_workload("avg", 0.5, true), false); + assert!(!plan.stored_outputs.is_empty(), "panes built at ingestion"); + // Eleven steps: the first, before the window, opens the panes (the + // first pane after startup is never built). + let (points, windows) = data_samples(8, 11, 10_000, |s, step| (s + 1) as f64 * step as f64); + assert!(windows.values().flatten().all(|window| window.len() == 10)); + let (app, _) = run(&plan, &points).await; + let answers = answer(&app, query).await; + assert_eq!(answers.len(), windows.len(), "{answers:?}"); + for (labels, value) in &answers { + let key = ( + labels["region"].clone(), + labels["service"].clone(), + labels["job"].clone(), + ); + let medians: Vec = windows[&key] + .iter() + .map(|window| interpolated_quantile(0.5, window)) + .collect(); + let exact = average(&medians); + assert!( + (value - exact).abs() <= 0.05 * exact, + "{labels:?}: {value} vs exact {exact}" + ); + } +} + +/// `data` samples for the top-k tests: three per series inside the last +/// minute, each a third of its sum, after a sample (1,000) two minutes +/// earlier that opens the panes (the first pane after startup is never +/// built) and no answer reads. Series +/// `e1`..`e3` are in `east` with sums 6, 60 and 30; `w1`, `w2` in `west` +/// with sums 9 and 3. +fn topk_samples() -> Vec { + let mut points = Vec::new(); + for (instance, region, service, sum) in [ + ("e1", "east", "api", 6.0), + ("e2", "east", "web", 60.0), + ("e3", "east", "db", 30.0), + ("w1", "west", "api", 9.0), + ("w2", "west", "web", 3.0), + ] { + for step in [8, 3, 2, 1] { + points.push(Point { + labels: labels(&[ + ("__name__", "data"), + ("instance", instance), + ("region", region), + ("service", service), + ]), + timestamp_ms: EVALUATION_MS - step * 15_000, + value: if step == 8 { 1000.0 } else { sum / 3.0 }, + }); + } + } + points +} + +/// `query` every second within 0.05 at failure probability 0.3 over 1,000 +/// `data` gauge series sampled every 100 ms: the benchmark's top-k shape. +/// Planned with accuracy guarantees ignored, as the benchmark plans it. +fn topk_workload(query: &str) -> PlanningWorkload { + serde_json::from_str(&format!( + r#"{{"query_workload": {{"language": "promql", "repeating_queries": [{{ + "query": {query:?}, + "demand": {{"fixed_interval": 1000}}, + "requirements": {{"accuracy": {{"max_error": 0.05, "failure_probability": 0.3, + "accuracy_scope": "whole_result"}}}}, + "predictability": {{"predictable": {{"known_at": null}}}}}}]}}, + "data_workload": {{"arrival": {{"ingesting": {{"rows_per_second": 10000.0}}}}, + "sources": {{"data": {{"series_count": 1000, "sample_interval": 100, + "metric_type": "gauge"}}}}}}}}"# + )) + .unwrap() +} + +/// PromQL `topk` returns whole input series: `topk by (region) (2, +/// sum_over_time(data[1m]))` answers, per region, the two series with the +/// largest sums, each with its full label set (region, service and +/// instance; `sum_over_time` drops the metric name), not only `region`. +#[tokio::test] +async fn topk_answers_carry_the_chosen_series_labels() { + let query = "topk by (region) (2, sum_over_time(data[1m]))"; + let plan = plan_without_raw_samples(&topk_workload(query), true); + let (app, _) = run(&plan, &topk_samples()).await; + let series = |instance: &str, region: &str, service: &str| { + labels(&[ + ("instance", instance), + ("region", region), + ("service", service), + ]) + }; + let expected = [ + (series("e2", "east", "web"), 60.0), + (series("e3", "east", "db"), 30.0), + (series("w1", "west", "api"), 9.0), + (series("w2", "west", "web"), 3.0), + ]; + let answered = answer(&app, query).await; + assert_eq!(answered.len(), expected.len(), "{answered:?}"); + for ((labels, value), (want_labels, want)) in answered.iter().zip(&expected) { + assert_eq!(labels, want_labels); + assert!((value - want).abs() < 1e-9, "{labels:?}: {value}"); + } +} + +/// Every top-k and bottom-k form, grouped or not, over sums, counts and +/// rates, answers whole input series. +#[tokio::test] +async fn every_topk_form_answers_whole_series() { + for (query, expected) in [ + ("topk(2, sum_over_time(data[1m]))", 2), + ("topk by (region) (1, count_over_time(data[1m]))", 2), + ("topk by (region) (1, rate(data[1m]))", 2), + ("bottomk by (region) (1, sum_over_time(data[1m]))", 2), + ("bottomk(3, sum_over_time(data[1m]))", 3), + ] { + let plan = plan_without_raw_samples(&topk_workload(query), true); + let (app, _) = run(&plan, &topk_samples()).await; + let answered = answer(&app, query).await; + assert_eq!(answered.len(), expected, "{query}: {answered:?}"); + for (labels, _) in &answered { + assert!( + labels.contains_key("instance") && labels.contains_key("service"), + "{query}: {labels:?}" + ); + } + } +} + +/// The per-second rate of series `instance` (1 or 2) of job `job` (1..=12): +/// distinct for every series and every job's sum. +fn counter_rate(job: i64, instance: i64) -> f64 { + job as f64 + 0.1 * instance as f64 +} + +/// `data_total` samples of 12 jobs with 2 series each, every 10 s from +/// 595 s before `EVALUATION_MS`, each growing linearly at its rate, so its +/// 5-minute `increase` is 300 × its rate. +fn counter_samples() -> Vec { + let mut points = Vec::new(); + for job in 1..=12 { + for instance in 1..=2 { + for step in 0..60 { + points.push(Point { + labels: labels(&[ + ("__name__", "data_total"), + ("instance", &format!("{job}-{instance}")), + ("job", &format!("j{job:02}")), + ]), + timestamp_ms: EVALUATION_MS - 595_000 + step * 10_000, + value: counter_rate(job, instance) * (5 + step * 10) as f64, + }); + } + } + } + points +} + +/// The ASAPQuery workloads' counter top-k, `topk(10, sum by (job) +/// ((data_total[5m])))` for `increase` and `rate`, planned with accuracy +/// guarantees ignored and no raw samples: the plan the Planner's executor +/// compiles runs here too and answers the exact top 10 jobs (3..=12). +#[tokio::test] +async fn counter_topk_planned_without_guarantees_answers_exactly() { + for (function, seconds) in [("increase", 300.0), ("rate", 1.0)] { + let query = format!("topk(10, sum by (job) ({function}(data_total[5m])))"); + let workload: PlanningWorkload = serde_json::from_str(&format!( + r#"{{"query_workload": {{"language": "promql", "repeating_queries": [{{ + "query": {query:?}, + "demand": {{"fixed_interval": 60000}}, + "requirements": {{"accuracy": {{"max_error": 0.05, "failure_probability": 0.3}}}}, + "predictability": {{"predictable": {{"known_at": null}}}}}}]}}, + "data_workload": {{"arrival": {{"ingesting": {{"rows_per_second": 100000.0}}}}, + "sources": {{"data_total": {{"series_count": 10000, "metric_type": "counter", + "sample_interval": 100}}}}}}}}"# + )) + .unwrap(); + let plan = plan_without_raw_samples(&workload, true); + let (app, _) = run(&plan, &counter_samples()).await; + let answered = answer(&app, &query).await; + let expected: Vec<(Labels, f64)> = (3..=12) + .map(|job| { + ( + labels(&[("job", &format!("j{job:02}"))]), + (counter_rate(job, 1) + counter_rate(job, 2)) * seconds, + ) + }) + .collect(); + assert_eq!(answered.len(), expected.len(), "{function}: {answered:?}"); + for ((labels, value), (want_labels, want)) in answered.iter().zip(&expected) { + assert_eq!(labels, want_labels, "{function}"); + assert!( + (value - want).abs() <= 1e-9 * want, + "{function} {labels:?}: {value} vs {want}" + ); + } + } +} + +/// `queries` every 10 s at `accuracy` (JSON) over six ingesting series of +/// `data`, a non-negative gauge: one workload, planned jointly. +fn joint_workload(queries: &[&str], accuracy: &str) -> PlanningWorkload { + let entries: Vec = queries + .iter() + .map(|query| { + format!( + r#"{{"query": {query:?}, "demand": {{"fixed_interval": 10000}}, + "requirements": {accuracy}, + "predictability": {{"predictable": {{"known_at": null}}}}}}"# + ) + }) + .collect(); + serde_json::from_str(&format!( + r#"{{"query_workload": {{"language": "promql", "repeating_queries": [{}]}}, + "data_workload": {{"arrival": {{"ingesting": {{"rows_per_second": 0.6}}}}, + "sources": {{"data": {{"series_count": 6, "sample_interval": 10000, + "metric_type": "gauge", "values_non_negative": true}}}}}}}}"#, + entries.join(",") + )) + .unwrap() +} + +/// Six series of `data`: `region` east/west, `service` api/web/db, `job` +/// batch/online; positive values, one every 10 s over the last two +/// minutes; and each series' window `(EVALUATION_MS - 100 s, +/// EVALUATION_MS]`. +fn grouped_series_samples() -> (Vec, Vec<(Labels, Vec)>) { + let mut points = Vec::new(); + let mut windows = Vec::new(); + for (s, (region, service, job)) in [ + ("east", "api", "batch"), + ("east", "web", "online"), + ("east", "db", "batch"), + ("west", "api", "online"), + ("west", "web", "batch"), + ("west", "db", "online"), + ] + .into_iter() + .enumerate() + { + let series_labels = labels(&[ + ("__name__", "data"), + ("region", region), + ("service", service), + ("job", job), + ]); + let mut window = Vec::new(); + for step in 0..12 { + let timestamp_ms = EVALUATION_MS - 120_000 + 5_000 + step * 10_000; + let value = (s as f64 + 1.0) * 10.0 + step as f64; + points.push(Point { + labels: series_labels.clone(), + timestamp_ms, + value, + }); + if timestamp_ms > EVALUATION_MS - 100_000 { + window.push(value); + } + } + let mut answer_labels = series_labels; + answer_labels.remove("__name__"); + windows.push((answer_labels, window)); + } + (points, windows) +} + +/// Quantile queries over one metric and window with different `by` labels +/// (ASAPQuery's seed-862 quantile queries), planned jointly, share one +/// per-series DDSketch, and each answers within 0.05 of its exact value. +#[tokio::test] +async fn jointly_planned_quantiles_share_one_sketch_and_answer_within_alpha() { + let queries = [ + "quantile_over_time(0.9, data[100s])", + "avg by (region) (quantile_over_time(0.99, data[100s]))", + ]; + let accuracy = r#"{"accuracy": {"accuracy_metric": "relative", "max_error": 0.05, + "failure_probability": 0.01, "accuracy_scope": "per_value"}}"#; + let plan = plan_without_raw_samples(&joint_workload(&queries, accuracy), false); + let sketch_definitions: std::collections::BTreeSet<&str> = plan + .stored_outputs + .iter() + .filter(|output| { + let node = plan + .dag + .dag + .nodes + .iter() + .find(|n| n.id == output.node_id) + .unwrap(); + serde_json::to_string(&node.payload) + .unwrap() + .contains("DDSketch") + }) + .map(|output| output.coverage.definition_id.as_str()) + .collect(); + assert_eq!(sketch_definitions.len(), 1, "{sketch_definitions:?}"); + let (points, windows) = grouped_series_samples(); + let (app, _) = run(&plan, &points).await; + + let per_series = answer(&app, queries[0]).await; + assert_eq!(per_series.len(), 6, "{per_series:?}"); + for (labels, value) in &per_series { + let (_, window) = windows.iter().find(|(l, _)| l == labels).unwrap(); + let exact = interpolated_quantile(0.9, window); + assert!( + (value - exact).abs() <= 0.05 * exact, + "{labels:?}: {value} vs {exact}" + ); + } + let per_region = answer(&app, queries[1]).await; + assert_eq!(per_region.len(), 2, "{per_region:?}"); + for (labels, value) in &per_region { + let p99s: Vec = windows + .iter() + .filter(|(l, _)| l["region"] == labels["region"]) + .map(|(_, window)| interpolated_quantile(0.99, window)) + .collect(); + let exact = average(&p99s); + assert!( + (value - exact).abs() <= 0.05 * exact, + "{labels:?}: {value} vs {exact}" + ); + } +} + +/// A fine and a coarse sum over one metric and window (ASAPQuery's +/// sum-fine and sum-coarse), planned jointly, are both answered exactly +/// from the stored per-series sums. +#[tokio::test] +async fn jointly_planned_fine_and_coarse_sums_answer_exactly() { + let queries = [ + "sum by (region, service, job) (sum_over_time(data[100s]))", + "sum by (region) (sum_over_time(data[100s]))", + ]; + let plan = plan_without_raw_samples(&joint_workload(&queries, "{}"), false); + let (points, windows) = grouped_series_samples(); + let (app, _) = run(&plan, &points).await; + let fine = answer(&app, queries[0]).await; + assert_eq!(fine.len(), 6, "{fine:?}"); + for (labels, value) in &fine { + let (_, window) = windows.iter().find(|(l, _)| l == labels).unwrap(); + let exact: f64 = window.iter().sum(); + assert!( + (value - exact).abs() < 1e-9, + "{labels:?}: {value} vs {exact}" + ); + } + let coarse = answer(&app, queries[1]).await; + assert_eq!(coarse.len(), 2, "{coarse:?}"); + for (labels, value) in &coarse { + let exact: f64 = windows + .iter() + .filter(|(l, _)| l["region"] == labels["region"]) + .flat_map(|(_, window)| window) + .sum(); + assert!( + (value - exact).abs() < 1e-9, + "{labels:?}: {value} vs {exact}" + ); + } +} From 070ffe5d907a8b3e5b699e7cfe52623c53a0febf Mon Sep 17 00:00:00 2001 From: zzylol Date: Wed, 21 Oct 2026 10:25:07 +0000 Subject: [PATCH 3/5] feat(deps): pin ASAPPlanner main 05fb36e4 (#808: faster installable-plan compile check) --- Cargo.lock | 16 ++++++++-------- Cargo.toml | 4 ++-- 2 files changed, 10 insertions(+), 10 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 47124d8a..32bcc279 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -252,7 +252,7 @@ dependencies = [ [[package]] name = "asap-executor" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=ef52e175d75f074b2634e6e332b4ec47f8d02fc0#ef52e175d75f074b2634e6e332b4ec47f8d02fc0" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=05fb36e4e9ff0b8e2ff6d43685785117ec0b9225#05fb36e4e9ff0b8e2ff6d43685785117ec0b9225" dependencies = [ "ahash", "arrow", @@ -270,7 +270,7 @@ dependencies = [ [[package]] name = "asap-frontend-common" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=ef52e175d75f074b2634e6e332b4ec47f8d02fc0#ef52e175d75f074b2634e6e332b4ec47f8d02fc0" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=05fb36e4e9ff0b8e2ff6d43685785117ec0b9225#05fb36e4e9ff0b8e2ff6d43685785117ec0b9225" dependencies = [ "asap-types", "serde", @@ -280,7 +280,7 @@ dependencies = [ [[package]] name = "asap-frontend-promql" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=ef52e175d75f074b2634e6e332b4ec47f8d02fc0#ef52e175d75f074b2634e6e332b4ec47f8d02fc0" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=05fb36e4e9ff0b8e2ff6d43685785117ec0b9225#05fb36e4e9ff0b8e2ff6d43685785117ec0b9225" dependencies = [ "asap-frontend-common", "asap-types", @@ -290,7 +290,7 @@ dependencies = [ [[package]] name = "asap-logical-asap-optimization" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=ef52e175d75f074b2634e6e332b4ec47f8d02fc0#ef52e175d75f074b2634e6e332b4ec47f8d02fc0" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=05fb36e4e9ff0b8e2ff6d43685785117ec0b9225#05fb36e4e9ff0b8e2ff6d43685785117ec0b9225" dependencies = [ "asap-types", "asap_sketchlib", @@ -302,7 +302,7 @@ dependencies = [ [[package]] name = "asap-physical-asap-optimization" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=ef52e175d75f074b2634e6e332b4ec47f8d02fc0#ef52e175d75f074b2634e6e332b4ec47f8d02fc0" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=05fb36e4e9ff0b8e2ff6d43685785117ec0b9225#05fb36e4e9ff0b8e2ff6d43685785117ec0b9225" dependencies = [ "asap-logical-asap-optimization", "asap-types", @@ -312,7 +312,7 @@ dependencies = [ [[package]] name = "asap-plan-selection" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=ef52e175d75f074b2634e6e332b4ec47f8d02fc0#ef52e175d75f074b2634e6e332b4ec47f8d02fc0" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=05fb36e4e9ff0b8e2ff6d43685785117ec0b9225#05fb36e4e9ff0b8e2ff6d43685785117ec0b9225" dependencies = [ "asap-logical-asap-optimization", "asap-physical-asap-optimization", @@ -326,7 +326,7 @@ dependencies = [ [[package]] name = "asap-planner" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=ef52e175d75f074b2634e6e332b4ec47f8d02fc0#ef52e175d75f074b2634e6e332b4ec47f8d02fc0" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=05fb36e4e9ff0b8e2ff6d43685785117ec0b9225#05fb36e4e9ff0b8e2ff6d43685785117ec0b9225" dependencies = [ "asap-executor", "asap-frontend-promql", @@ -341,7 +341,7 @@ dependencies = [ [[package]] name = "asap-types" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=ef52e175d75f074b2634e6e332b4ec47f8d02fc0#ef52e175d75f074b2634e6e332b4ec47f8d02fc0" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=05fb36e4e9ff0b8e2ff6d43685785117ec0b9225#05fb36e4e9ff0b8e2ff6d43685785117ec0b9225" dependencies = [ "serde", "serde_json", diff --git a/Cargo.toml b/Cargo.toml index 5d668476..83288fc0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -10,8 +10,8 @@ version = "0.1.0" # The backend depends on two Planner crates, at one immutable revision: # `asap-planner` to plan and `asap-executor` to execute. Planner IR types # are reached through `asap_executor::planner`. -asap-planner = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "ef52e175d75f074b2634e6e332b4ec47f8d02fc0", default-features = false, features = ["promql"] } -asap-executor = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "ef52e175d75f074b2634e6e332b4ec47f8d02fc0" } +asap-planner = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "05fb36e4e9ff0b8e2ff6d43685785117ec0b9225", default-features = false, features = ["promql"] } +asap-executor = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "05fb36e4e9ff0b8e2ff6d43685785117ec0b9225" } serde = { version = "1.0", features = ["derive"] } # `float_roundtrip`: a plan file reloads to the plan it was written from From f56ac9bf0ea0c86f26ac31c485230210b844156c Mon Sep 17 00:00:00 2001 From: zzylol Date: Wed, 21 Oct 2026 11:42:06 +0000 Subject: [PATCH 4/5] fix(sds): panes of one definition id and two widths are separate definitions A jointly planned workload can store one computation (one definition id, the same label constraints) in panes of two widths, e.g. per-series sums in 1-second panes for `sum_over_time(data[1m])` every second and in 30-second panes for `sum_over_time(data[20m])` every 30 s. The SDS keyed definitions by (id, label constraints), so both shared one definition with the first width: the 20-minute query merged forty 1-second panes and answered 1/30 of its sum. DefinitionKey and Selection now carry the pane width: a definition is (id, label constraints, pane width), and a summary instance's concrete selection is (label constraints, width, end), the interval (end - width, end]. --- data_plane/src/query.rs | 19 +-- data_plane/src/sds.rs | 71 +++++---- data_plane/tests/planner_executor_answers.rs | 158 +++++++++++++++---- data_plane/tests/query_run.rs | 2 +- data_plane/tests/sds.rs | 41 +++-- 5 files changed, 203 insertions(+), 88 deletions(-) diff --git a/data_plane/src/query.rs b/data_plane/src/query.rs index 14951c87..515e9893 100644 --- a/data_plane/src/query.rs +++ b/data_plane/src/query.rs @@ -35,7 +35,7 @@ use asap_planner::{InstallablePlan, StoredOutputBuild}; use crate::precompute::{run, RawSample, SummaryBuild, SummaryInstance}; use crate::raw_sample_window::RawSampleWindowBatcher; -use crate::sds::{DefinitionKey, Sds, Selection, SummaryInstanceKey, SummaryStore}; +use crate::sds::{DefinitionKey, Sds, SummaryInstanceKey, SummaryStore}; /// A bound raw-sample window: a `RawSampleWindow` of the plan bound to one /// evaluation time, i.e. `metric`'s raw samples in `(start_ms, end_ms]` @@ -279,7 +279,7 @@ impl QueryEngine { if store .get( &input.definition.id, - &selection_of(&input.definition, end_ms), + &input.definition.selection_ending_at(end_ms), ) .is_none() { @@ -343,7 +343,7 @@ impl QueryEngine { .as_ref() .expect("compiled roots read only `QueryRun` definitions that build"); let (_, end_ms) = window(time_ms, input.offset_ms, build.definition.pane_ms)?; - let selection = selection_of(&input.definition, end_ms); + let selection = input.definition.selection_ending_at(end_ms); let summary = match store.get(&input.definition.id, &selection) { Some(summary) => { self.query_run_stats.reused += 1; @@ -443,14 +443,6 @@ impl QueryEngine { } } -/// The concrete selection of `definition`'s pane ending at `end_ms`. -fn selection_of(definition: &DefinitionKey, end_ms: i64) -> Selection { - Selection { - labels: definition.labels.clone(), - end_ms, - } -} - /// `(start, end]` of a window `width_ms` wide ending `offset_ms` before /// `time_ms`, or why it does not fit in `i64` milliseconds. fn window(time_ms: i64, offset_ms: i64, width_ms: i64) -> Result<(i64, i64), Unanswered> { @@ -475,7 +467,10 @@ fn pane<'a>( ))); } store - .get(&input.definition.id, &selection_of(&input.definition, end)) + .get( + &input.definition.id, + &input.definition.selection_ending_at(end), + ) .ok_or_else(|| { Unanswered::MissingPane(format!( "pane ending at {end} of {:?} is not stored", diff --git a/data_plane/src/sds.rs b/data_plane/src/sds.rs index 4dd0a3ae..364c3ef2 100644 --- a/data_plane/src/sds.rs +++ b/data_plane/src/sds.rs @@ -9,14 +9,16 @@ //! `retention_ms`. //! //! A definition's identity is the plan's definition id. Outputs with the same -//! definition id and the same label constraints in their selection differ -//! only in time, so they share one [`SummaryDefinition`]: the data plane -//! builds each pane once and every output reads the one it needs. Outputs -//! with the same definition id but other label constraints are other -//! [`SummaryDefinition`]s ([`DefinitionKey`]). A summary instance is keyed by -//! its definition id and its concrete [`Selection`]: those label constraints -//! and the time interval `(end - width, end]` the plan's relative time binds -//! to at evaluation time. +//! definition id, the same label constraints in their selection and the +//! same pane width differ only in where their time interval ends, so they +//! share one [`SummaryDefinition`]: the data plane builds each pane once and +//! every output reads the one it needs. Outputs with the same definition id +//! but other label constraints or another pane width (e.g. one query's +//! 1-second panes and another's 30-second panes of the same per-series sum) +//! are other [`SummaryDefinition`]s ([`DefinitionKey`]). A summary instance +//! is keyed by its definition id and its concrete [`Selection`]: those label +//! constraints and the time interval `(end - width, end]` the plan's +//! relative time binds to at evaluation time. //! //! A stored output may be built from other stored outputs (rollups over time //! or across series). [`Sds::ingestion_build_order`] orders definitions so @@ -44,20 +46,22 @@ pub struct DefinitionId(pub String); /// plan's JSON form; empty when it has none. pub type LabelConstraints = Arc; -/// One [`SummaryDefinition`] of the SDS: a definition id and the label -/// constraints its outputs share. +/// One [`SummaryDefinition`] of the SDS: a definition id, and the label +/// constraints and pane width its outputs share. #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] pub struct DefinitionKey { pub id: DefinitionId, pub labels: LabelConstraints, + pub pane_ms: i64, } impl DefinitionKey { - /// `id` with no label constraints. - pub fn unconstrained(id: DefinitionId) -> Self { + /// `id`'s panes `pane_ms` wide with no label constraints. + pub fn unconstrained(id: DefinitionId, pane_ms: i64) -> Self { Self { id, labels: Arc::from(""), + pane_ms, } } } @@ -99,15 +103,13 @@ impl SummaryDefinition { DefinitionKey { id: self.id.clone(), labels: self.labels.clone(), + pane_ms: self.pane_ms, } } /// The concrete selection of its pane ending at `end_ms`. pub fn selection_ending_at(&self, end_ms: i64) -> Selection { - Selection { - labels: self.labels.clone(), - end_ms, - } + self.key().selection_ending_at(end_ms) } /// The builder's `TimeRange`, which one pane's raw samples feed. @@ -201,6 +203,7 @@ impl Sds { let key = DefinitionKey { id: DefinitionId(output.coverage.definition_id.clone()), labels: label_constraints(selection), + pane_ms: end - start, }; let retention_ms = i64::try_from(output.retention_ms) .map_err(|_| SdsError::RetentionOverflow(node))?; @@ -211,7 +214,7 @@ impl Sds { id: key.id.clone(), labels: key.labels.clone(), built_at: output.built_at, - pane_ms: end - start, + pane_ms: key.pane_ms, builder: node, window: window.clone(), output_schema, @@ -335,20 +338,32 @@ fn label_constraints(selection: &SelectionBox) -> LabelConstraints { } /// The concrete selection of one summary instance: its definition's label -/// constraints and the time interval `(end_ms - width, end_ms]` the plan's -/// relative time binds to at evaluation time, the width being its -/// definition's pane width. +/// constraints and the time interval `(end_ms - pane_ms, end_ms]` the +/// plan's relative time binds to at evaluation time. #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] pub struct Selection { pub labels: LabelConstraints, + pub pane_ms: i64, pub end_ms: i64, } impl Selection { - /// The pane ending at `end_ms`, with no label constraints. - pub fn ending_at(end_ms: i64) -> Self { + /// The pane `(end_ms - pane_ms, end_ms]`, with no label constraints. + pub fn ending_at(end_ms: i64, pane_ms: i64) -> Self { Self { labels: Arc::from(""), + pane_ms, + end_ms, + } + } +} + +impl DefinitionKey { + /// The concrete selection of this definition's pane ending at `end_ms`. + pub fn selection_ending_at(&self, end_ms: i64) -> Selection { + Selection { + labels: self.labels.clone(), + pane_ms: self.pane_ms, end_ms, } } @@ -405,24 +420,18 @@ impl SummaryStore { /// The instance of `definition`'s pane ending at `end_ms`, if stored. pub fn get_pane(&self, definition: &DefinitionKey, end_ms: i64) -> Option<&S> { - self.get( - &definition.id, - &Selection { - labels: definition.labels.clone(), - end_ms, - }, - ) + self.get(&definition.id, &definition.selection_ending_at(end_ms)) } /// Drops the instances of `definition` with `oldest`'s label constraints - /// whose pane ends before `oldest`'s. + /// and pane width whose pane ends before `oldest`'s. pub fn evict_before(&mut self, definition: &DefinitionId, oldest: &Selection) { let Some(instances) = self.instances.get_mut(definition) else { return; }; let from = Selection { - labels: oldest.labels.clone(), end_ms: i64::MIN, + ..oldest.clone() }; let expired: Vec = instances .range(from..oldest.clone()) diff --git a/data_plane/tests/planner_executor_answers.rs b/data_plane/tests/planner_executor_answers.rs index 9d0b0bd5..c9938490 100644 --- a/data_plane/tests/planner_executor_answers.rs +++ b/data_plane/tests/planner_executor_answers.rs @@ -52,7 +52,8 @@ fn plan_without_raw_samples( } /// A data plane running `plan` with `--no-forward`, its panes built from -/// `points` written by Remote Write and closed by the barrier at +/// `points` written by Remote Write, one request per second, and closed by +/// the barrier at /// `EVALUATION_MS`. async fn run(plan: &InstallablePlan, points: &[Point]) -> (Router, Arc) { let options = EngineOptions { @@ -69,39 +70,49 @@ async fn run(plan: &InstallablePlan, points: &[Point]) -> (Router, Arc> = BTreeMap::new(); + // One Remote Write request per second of event time, in time order, as + // a paced feeder sends them: a pane longer than a second takes its + // samples over several requests. + let mut by_second: BTreeMap>> = BTreeMap::new(); for point in points { - by_series.entry(&point.labels).or_default().push(Sample { - timestamp: point.timestamp_ms, - value: point.value, - }); + by_second + .entry(point.timestamp_ms.div_euclid(1000)) + .or_default() + .entry(&point.labels) + .or_default() + .push(Sample { + timestamp: point.timestamp_ms, + value: point.value, + }); + } + for by_series in by_second.into_values() { + let timeseries = by_series + .into_iter() + .map(|(labels, mut samples)| { + samples.sort_by_key(|s| s.timestamp); + TimeSeries { + labels: labels + .iter() + .map(|(name, value)| Label { + name: name.clone(), + value: value.clone(), + }) + .collect(), + samples, + } + }) + .collect(); + let written = app + .clone() + .oneshot( + Request::post("/api/v1/write") + .body(Body::from(encode(&WriteRequest { timeseries }))) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(written.status(), StatusCode::NO_CONTENT); } - let timeseries = by_series - .into_iter() - .map(|(labels, mut samples)| { - samples.sort_by_key(|s| s.timestamp); - TimeSeries { - labels: labels - .iter() - .map(|(name, value)| Label { - name: name.clone(), - value: value.clone(), - }) - .collect(), - samples, - } - }) - .collect(); - let written = app - .clone() - .oneshot( - Request::post("/api/v1/write") - .body(Body::from(encode(&WriteRequest { timeseries }))) - .unwrap(), - ) - .await - .unwrap(); - assert_eq!(written.status(), StatusCode::NO_CONTENT); for _ in 0..1000 { if state.engine.queued_samples() == 0 { break; @@ -690,3 +701,86 @@ async fn jointly_planned_fine_and_coarse_sums_answer_exactly() { ); } } + +/// ASAPQuery's seed-862 top-k sums over one metric and two windows, +/// planned jointly: `[1m]` every second and `[20m]` every 30 s. The plan +/// stores per-series sums of one definition id in 1-second panes for the +/// first and 30-second panes for the second, with the same (no) label +/// constraints. Each query merges its own panes: every series' answer is +/// its sum over its query's whole window. +#[tokio::test] +async fn panes_of_one_definition_and_two_widths_are_stored_apart() { + let queries = [ + ( + "topk by (region, service) (4, sum_over_time(data[1m]))", + 1000, + 60.0, + ), + ( + "topk by (region, service) (4, sum_over_time(data[20m]))", + 30000, + 1200.0, + ), + ]; + let entries: Vec = queries + .iter() + .map(|(query, interval, _)| { + format!( + r#"{{"query": {query:?}, "demand": {{"fixed_interval": {interval}}}, + "requirements": {{"accuracy": {{"max_error": 0.05, "failure_probability": 0.3, + "accuracy_scope": "whole_result"}}}}, + "predictability": {{"predictable": {{"known_at": null}}}}}}"# + ) + }) + .collect(); + let workload: PlanningWorkload = serde_json::from_str(&format!( + r#"{{"query_workload": {{"language": "promql", "repeating_queries": [{}]}}, + "data_workload": {{"arrival": {{"ingesting": {{"rows_per_second": 10000.0}}}}, + "sources": {{"data": {{"series_count": 1000, "sample_interval": 100, + "metric_type": "gauge", "values_non_negative": true}}}}}}}}"#, + entries.join(",") + )) + .unwrap(); + let plan = plan_without_raw_samples(&workload, true); + // One definition id, two pane widths. + let mut widths: BTreeMap<&str, std::collections::BTreeSet> = BTreeMap::new(); + for output in &plan.stored_outputs { + for window in &output.inputs.raw_sample_windows { + widths + .entry(output.coverage.definition_id.as_str()) + .or_default() + .insert(window.width_ms); + } + } + assert!( + widths.values().any(|w| w.len() == 2), + "the case this tests: {widths:?}" + ); + // Two series in one (region, service), ten samples per second each + // over the last 21 minutes (the first minute, before the window, opens + // the panes): 0.1 and 0.2, so 1 and 2 per second. + let mut points = Vec::new(); + for (instance, value) in [("a", 0.1), ("b", 0.2)] { + for step in 0..12_600 { + points.push(Point { + labels: labels(&[ + ("__name__", "data"), + ("instance", instance), + ("region", "r"), + ("service", "s"), + ]), + timestamp_ms: EVALUATION_MS - 1_260_000 + 50 + step * 100, + value, + }); + } + } + let (app, _) = run(&plan, &points).await; + for (query, _, seconds) in queries { + let answered = answer(&app, query).await; + let values: Vec = answered.iter().map(|(_, v)| *v).collect(); + assert_eq!(values.len(), 2, "{query}: {answered:?}"); + for (value, want) in values.iter().zip([seconds, 2.0 * seconds]) { + assert!((value - want).abs() < 1e-6, "{query}: {answered:?}"); + } + } +} diff --git a/data_plane/tests/query_run.rs b/data_plane/tests/query_run.rs index 5f2c3bb0..72ec4e5a 100644 --- a/data_plane/tests/query_run.rs +++ b/data_plane/tests/query_run.rs @@ -37,7 +37,7 @@ fn fixture(plan: &InstallablePlan) -> Fixture { .values() .find(|d| d.built_at == StoredOutputBuild::QueryRun) .map(|d| d.key()) - .unwrap_or_else(|| DefinitionKey::unconstrained(DefinitionId(String::new()))); + .unwrap_or_else(|| DefinitionKey::unconstrained(DefinitionId(String::new()), 0)); Fixture { engine: QueryEngine::new(plan, &sds), store: SummaryStore::new(), diff --git a/data_plane/tests/sds.rs b/data_plane/tests/sds.rs index dd6b54ca..375e183c 100644 --- a/data_plane/tests/sds.rs +++ b/data_plane/tests/sds.rs @@ -143,32 +143,47 @@ fn build_order_follows_stored_inputs() { assert_eq!(sds.ingestion_build_order(), Err(SdsError::Cycle(id))); } -/// Panes are found by definition and end, and evicted once no query needs -/// them. +/// Panes are found by definition, width and end, and evicted once no query +/// needs them. Panes of one definition id with two widths (`a`'s 60 and 30) +/// are kept apart: neither is found or evicted as the other. #[test] fn store_finds_and_evicts_panes() { - let (a, b) = ( - DefinitionKey::unconstrained(DefinitionId("a".into())), - DefinitionKey::unconstrained(DefinitionId("b".into())), + let (a, a30, b) = ( + DefinitionKey::unconstrained(DefinitionId("a".into()), 60), + DefinitionKey::unconstrained(DefinitionId("a".into()), 30), + DefinitionKey::unconstrained(DefinitionId("b".into()), 60), ); let mut store = SummaryStore::new(); for end in [60, 120, 180] { store.insert( - SummaryInstanceKey::new(a.id.clone(), Selection::ending_at(end)), + SummaryInstanceKey::new(a.id.clone(), a.selection_ending_at(end)), format!("a{end}"), ); } store.insert( - SummaryInstanceKey::new(b.id.clone(), Selection::ending_at(60)), + SummaryInstanceKey::new(a30.id.clone(), a30.selection_ending_at(120)), + "a30-120".to_string(), + ); + store.insert( + SummaryInstanceKey::new(b.id.clone(), Selection::ending_at(60, 60)), "b60".to_string(), ); assert_eq!(store.get_pane(&a, 120).map(String::as_str), Some("a120")); + assert_eq!( + store.get_pane(&a30, 120).map(String::as_str), + Some("a30-120") + ); + assert_eq!(store.get_pane(&a30, 60), None); assert_eq!(store.get_pane(&a, 90), None); - store.evict_before(&a.id, &Selection::ending_at(180)); + store.evict_before(&a.id, &a.selection_ending_at(180)); assert_eq!(store.get_pane(&a, 120), None); assert_eq!(store.get_pane(&a, 180).map(String::as_str), Some("a180")); + assert_eq!( + store.get_pane(&a30, 120).map(String::as_str), + Some("a30-120") + ); assert_eq!(store.get_pane(&b, 60).map(String::as_str), Some("b60")); - assert_eq!(store.len(), 2); + assert_eq!(store.len(), 3); } /// `definition` under another identity, reading `stored_inputs`. @@ -181,7 +196,9 @@ fn renamed( definition.id = DefinitionId(name.into()); definition.stored_inputs = stored_inputs .iter() - .map(|input| DefinitionKey::unconstrained(DefinitionId(input.to_string()))) + .map(|input| { + DefinitionKey::unconstrained(DefinitionId(input.to_string()), definition.pane_ms) + }) .collect(); definition } @@ -224,8 +241,8 @@ fn build_order_refuses_ingestion_reading_query_run() { assert_eq!( sds.ingestion_build_order(), Err(SdsError::IngestionReadsQueryRun { - definition: DefinitionKey::unconstrained(DefinitionId("i".into())), - input: DefinitionKey::unconstrained(DefinitionId("q".into())), + definition: DefinitionKey::unconstrained(DefinitionId("i".into()), template.pane_ms), + input: DefinitionKey::unconstrained(DefinitionId("q".into()), template.pane_ms), }) ); } From 4a984b32144b8fa346eeed23a0803e71e61d7a44 Mon Sep 17 00:00:00 2001 From: zzylol Date: Wed, 21 Oct 2026 12:48:33 +0000 Subject: [PATCH 5/5] feat(precompute): --stream-start-ms builds the first pane that starts at the stream's start The first pane each stored output sees is skipped, since it may have started before the data plane did. When the producers start their stream at the data plane's start (the benchmark feeds from an aligned start S), that pane is complete, and a query whose window reaches back to S was refused: e.g. `rate(data_total[20m])` and `sum_over_time(data[20m])` at S + 20 min. A producer can now declare where its stream starts (PrecomputeOptions:: stream_start_ms, `--stream-start-ms`): a first pane starting at or after it is built; one starting before it is still skipped. --- README.md | 6 +++++ data_plane/src/main.rs | 7 ++++++ data_plane/src/precompute.rs | 24 +++++++++++++++--- data_plane/tests/precompute.rs | 45 ++++++++++++++++++++++++++++++++++ 4 files changed, 79 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index ad474f8b..8a36243f 100644 --- a/README.md +++ b/README.md @@ -112,6 +112,12 @@ accuracy guarantee, so candidates are compared on cost alone; the plan's `selection.queries_accepted_without_guarantee` lists the queries accepted only because of it. +`--stream-start-ms T` declares that the producers write every raw sample at +or after `T`: the first pane of each stored output that starts at or after +`T` is complete and is built. Without it, the first pane each stored output +sees may have started before the data plane did, so it is skipped and a +query whose window needs it is not answered from the plan. + `--forward-timeout-seconds` bounds each forwarded query (default 120, matching Prometheus's default query timeout); a query that runs longer gets 502. diff --git a/data_plane/src/main.rs b/data_plane/src/main.rs index 8aa337b3..8053ffc7 100644 --- a/data_plane/src/main.rs +++ b/data_plane/src/main.rs @@ -27,6 +27,12 @@ struct Args { /// Samples further ahead of the wall clock than this are dropped. #[arg(long, default_value_t = PrecomputeOptions::default().max_future_ms)] max_future_ms: u64, + /// Where the producers' stream starts (ms since the epoch): every raw + /// sample at or after it is written to the data plane, so a first pane + /// that starts at or after it is complete and is built. By default the + /// first pane each definition sees is skipped. + #[arg(long)] + stream_start_ms: Option, /// Samples queued for the engine beyond which Remote Write gets 503. #[arg(long, default_value_t = EngineOptions::default().max_queued_samples)] max_queued_samples: u64, @@ -53,6 +59,7 @@ async fn main() -> anyhow::Result<()> { precompute: PrecomputeOptions { allowed_lateness_ms: args.allowed_lateness_ms, max_future_ms: args.max_future_ms, + stream_start_ms: args.stream_start_ms, }, max_queued_samples: args.max_queued_samples, no_forward: args.no_forward, diff --git a/data_plane/src/precompute.rs b/data_plane/src/precompute.rs index b2ff4d73..69cac61e 100644 --- a/data_plane/src/precompute.rs +++ b/data_plane/src/precompute.rs @@ -12,7 +12,9 @@ //! under ingestion scope. The output is stored and never changes; samples //! older than a closed pane are dropped. The first pane a definition sees //! may have started before the data plane did, so it is never built: -//! queries that need it are not answered from stored panes. +//! queries that need it are not answered from stored panes. A producer that +//! declares where its stream starts (`stream_start_ms`) makes a first pane +//! starting at or after that point complete, and it is built. //! //! Only samples of a metric some definition reads move the watermark, and a //! sample further ahead of the wall clock than `max_future_ms` is dropped, so @@ -97,6 +99,11 @@ pub struct PrecomputeOptions { pub allowed_lateness_ms: u64, /// Samples further ahead of the wall clock than this are dropped. pub max_future_ms: u64, + /// Where the producer's stream starts: every raw sample at or after it + /// reaches the data plane. A definition's first pane is then built when + /// it starts at or after it. `None`: the first pane may have started + /// before the data plane did, so it is never built. + pub stream_start_ms: Option, } impl Default for PrecomputeOptions { @@ -104,6 +111,7 @@ impl Default for PrecomputeOptions { Self { allowed_lateness_ms: 15_000, max_future_ms: 5 * 60_000, + stream_start_ms: None, } } } @@ -245,6 +253,7 @@ pub struct Precompute { query_run_metrics: BTreeMap>, allowed_lateness_ms: i64, max_future_ms: i64, + stream_start_ms: Option, newest_sample_ms: Option, /// The newest wall-clock time an ingest ran at: the watermark never /// passes it, so a series stamped ahead of the others cannot close @@ -343,6 +352,7 @@ impl Precompute { query_run_metrics, allowed_lateness_ms: millis(options.allowed_lateness_ms), max_future_ms: millis(options.max_future_ms), + stream_start_ms: options.stream_start_ms, newest_sample_ms: None, newest_clock_ms: None, barrier_ms: None, @@ -435,8 +445,16 @@ impl Precompute { let this = match pane_end(sample.timestamp_ms, builder.summary.definition.pane_ms) { None => Outcome::Late, Some(end) => { - let first = *builder.first_pane_end.get_or_insert(end); - let built = *builder.built_through.get_or_insert(end); + let pane_ms = builder.summary.definition.pane_ms; + // A first pane that starts at or after the declared + // stream start is complete: nothing is skipped, and + // the panes from it on are built. + let complete = self + .stream_start_ms + .is_some_and(|start| end - pane_ms >= start); + let skipped = if complete { end - pane_ms } else { end }; + let first = *builder.first_pane_end.get_or_insert(skipped); + let built = *builder.built_through.get_or_insert(skipped); if end == first { Outcome::FirstPane } else if end <= built { diff --git a/data_plane/tests/precompute.rs b/data_plane/tests/precompute.rs index 21887d8a..19a8c7c9 100644 --- a/data_plane/tests/precompute.rs +++ b/data_plane/tests/precompute.rs @@ -488,3 +488,48 @@ fn input_stays_open_after_a_barrier() { engine.ingest_at(&[sample("latency_ms", "a", 29 * MINUTE, 1.0)], 36 * MINUTE); assert_eq!(engine.stats.samples_dropped_late, 1); } + +/// A producer that declares where its stream starts (`stream_start_ms`): +/// every sample at or after it reaches the data plane. The first pane seen +/// is then complete when it starts at or after the declared start, and is +/// built; one that starts before it is still skipped. +#[test] +fn the_first_pane_is_built_when_it_starts_at_the_declared_stream_start() { + let plan = common::plan("example4b", false); + let sds = Sds::from_plan(&plan).unwrap(); + let id = sds.definitions.keys().next().unwrap().clone(); + let with_start = |stream_start_ms| { + let options = PrecomputeOptions { + allowed_lateness_ms: 0, + stream_start_ms: Some(stream_start_ms), + ..PrecomputeOptions::default() + }; + Precompute::new(plan.dag.dag.clone(), &sds, options).unwrap() + }; + // The stream starts at the pane boundary 10 min: pane (10, 20 min] is + // complete. + let mut engine = with_start(PANE); + engine.ingest(&samples( + "latency_ms", + &["a", "b"], + PANE + 15_000, + 35 * MINUTE, + )); + let pane = engine + .store + .get_pane(&id, 2 * PANE) + .expect("first pane built"); + assert_eq!(pane.num_rows(), 2); + assert_eq!(engine.stats.samples_skipped_first_pane, 0); + // The stream starts at 15 min, inside (10, 20 min]: that pane may have + // missed samples before 15 min, so it is skipped. + let mut engine = with_start(PANE + 5 * MINUTE); + engine.ingest(&samples( + "latency_ms", + &["a", "b"], + PANE + 5 * MINUTE, + 35 * MINUTE, + )); + assert!(engine.store.get_pane(&id, 2 * PANE).is_none()); + assert!(engine.store.get_pane(&id, 3 * PANE).is_some()); +}