diff --git a/Cargo.lock b/Cargo.lock index 7fdd3f01..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=7565c47e30609d9571d8a9a372bf5e195754b2bb#7565c47e30609d9571d8a9a372bf5e195754b2bb" +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=7565c47e30609d9571d8a9a372bf5e195754b2bb#7565c47e30609d9571d8a9a372bf5e195754b2bb" +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=7565c47e30609d9571d8a9a372bf5e195754b2bb#7565c47e30609d9571d8a9a372bf5e195754b2bb" +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=7565c47e30609d9571d8a9a372bf5e195754b2bb#7565c47e30609d9571d8a9a372bf5e195754b2bb" +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=7565c47e30609d9571d8a9a372bf5e195754b2bb#7565c47e30609d9571d8a9a372bf5e195754b2bb" +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=7565c47e30609d9571d8a9a372bf5e195754b2bb#7565c47e30609d9571d8a9a372bf5e195754b2bb" +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=7565c47e30609d9571d8a9a372bf5e195754b2bb#7565c47e30609d9571d8a9a372bf5e195754b2bb" +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=7565c47e30609d9571d8a9a372bf5e195754b2bb#7565c47e30609d9571d8a9a372bf5e195754b2bb" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=05fb36e4e9ff0b8e2ff6d43685785117ec0b9225#05fb36e4e9ff0b8e2ff6d43685785117ec0b9225" 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]] @@ -474,15 +469,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 +530,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 +634,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 +649,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 +667,6 @@ dependencies = [ "reqwest", "serde", "serde_json", - "sha2", "snap", "thiserror 1.0.69", "tokio", @@ -715,16 +681,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 +833,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" @@ -1932,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" @@ -2007,17 +1944,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 +2351,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" @@ -2777,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 1c511734..83288fc0 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 = "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 = "05fb36e4e9ff0b8e2ff6d43685785117ec0b9225", default-features = false, features = ["promql"] } +asap-executor = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "05fb36e4e9ff0b8e2ff6d43685785117ec0b9225" } 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..8a36243f 100644 --- a/README.md +++ b/README.md @@ -99,9 +99,24 @@ 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. + +`--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/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 588a67ee..a9bbbd2c 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 @@ -123,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/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/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/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..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 @@ -35,9 +37,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, @@ -99,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 { @@ -106,6 +111,7 @@ impl Default for PrecomputeOptions { Self { allowed_lateness_ms: 15_000, max_future_ms: 5 * 60_000, + stream_start_ms: None, } } } @@ -140,6 +146,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 +172,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 +215,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)?, }) @@ -243,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 @@ -321,7 +332,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); } @@ -338,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, @@ -430,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 { @@ -537,7 +560,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 +579,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..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::{DefinitionId, 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]` @@ -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, + &input.definition.selection_ending_at(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 = input.definition.selection_ending_at(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") @@ -460,7 +467,10 @@ fn pane<'a>( ))); } store - .get(&input.definition, Selection::ending_at(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", @@ -473,7 +483,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 +572,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..364c3ef2 100644 --- a/data_plane/src/sds.rs +++ b/data_plane/src/sds.rs @@ -1,65 +1,96 @@ -//! 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, 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 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 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`'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, + } + } +} + +/// 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 +98,23 @@ 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(), + 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 { + self.key().selection_ending_at(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 +139,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 +153,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 +174,116 @@ 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), + pane_ms: end - start, + }; 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: key.pane_ms, + 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; - } - if let Some(id) = stored.get(&node) { - if !found.contains(id) { - found.push(id.clone()); - } - 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()); } - 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 +315,57 @@ 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); - } - } - 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)?); +/// `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 hash: [u8; 32] = hasher.finalize().into(); - hashes.insert(node, hash); - Ok(hash) + let columns = SelectionBox { + columns: selection.columns.clone(), + relative_time: None, + }; + Arc::from(serde_json::to_string(&columns).expect("selections serialize")) } -fn hex(bytes: &[u8]) -> String { - bytes.iter().map(|b| format!("{b:02x}")).collect() -} - -/// 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 - 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 coverage ending at `end_ms`. - pub fn ending_at(end_ms: i64) -> Self { - Self { end_ms } + /// 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, + } } } @@ -349,16 +413,32 @@ 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) } - /// 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); + /// 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, &definition.selection_ending_at(end_ms)) + } + + /// Drops the instances of `definition` with `oldest`'s label constraints + /// 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 { + end_ms: i64::MIN, + ..oldest.clone() + }; + 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..45330585 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,15 +115,15 @@ 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) { 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; @@ -154,17 +153,11 @@ 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)); + 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 @@ -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/planner_executor_answers.rs b/data_plane/tests/planner_executor_answers.rs new file mode 100644 index 00000000..c9938490 --- /dev/null +++ b/data_plane/tests/planner_executor_answers.rs @@ -0,0 +1,786 @@ +//! 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, one request per second, 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()); + // 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_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); + } + 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}" + ); + } +} + +/// 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/precompute.rs b/data_plane/tests/precompute.rs index dd2ebc6d..19a8c7c9 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; @@ -599,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()); +} diff --git a/data_plane/tests/query_run.rs b/data_plane/tests/query_run.rs index e8561d59..72ec4e5a 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()), 0)); 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..375e183c 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,43 +138,52 @@ 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))); } -/// 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) = (DefinitionId("a".into()), 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.clone(), Selection::ending_at(end)), + SummaryInstanceKey::new(a.id.clone(), a.selection_ending_at(end)), format!("a{end}"), ); } store.insert( - SummaryInstanceKey::new(b.clone(), Selection::ending_at(60)), - "b60".to_string(), + SummaryInstanceKey::new(a30.id.clone(), a30.selection_ending_at(120)), + "a30-120".to_string(), ); - assert_eq!( - store.get(&a, Selection::ending_at(120)).map(String::as_str), - Some("a120") + store.insert( + SummaryInstanceKey::new(b.id.clone(), Selection::ending_at(60, 60)), + "b60".to_string(), ); - 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_pane(&a, 120).map(String::as_str), Some("a120")); assert_eq!( - store.get(&a, Selection::ending_at(180)).map(String::as_str), - Some("a180") + 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, &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(&b, Selection::ending_at(60)).map(String::as_str), - Some("b60") + store.get_pane(&a30, 120).map(String::as_str), + Some("a30-120") ); - assert_eq!(store.len(), 2); + assert_eq!(store.get_pane(&b, 60).map(String::as_str), Some("b60")); + assert_eq!(store.len(), 3); } /// `definition` under another identity, reading `stored_inputs`. @@ -203,7 +196,9 @@ 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()), definition.pane_ms) + }) .collect(); definition } @@ -220,13 +215,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 +236,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()), template.pane_ms), + input: DefinitionKey::unconstrained(DefinitionId("q".into()), template.pane_ms), }) ); } @@ -269,7 +264,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();