From 68f08040e4af7f08957cee0984a4beb092f6fc92 Mon Sep 17 00:00:00 2001 From: zzylol Date: Sun, 18 Oct 2026 15:36:25 +0000 Subject: [PATCH] feat(sds): define stored summaries from the plan's ingestion-time outputs Each MatAtIngestion output the query side reads (the executor's materialization frontier) reads one tumbling pane of raw samples. Outputs that differ only in which pane they read share one SummaryDefinition, identified by a SHA-256 of their sub-DAG with node ids blanked and offset TimeShifts skipped. Each output binds to its definition with an offset: at evaluation time T it reads the pane ending at T - offset. SummaryStore keeps built panes by definition and end. A plan with a node kept across evaluations is rejected: the backend declares it cannot keep one. #509 Example 4b on a deployment without raw data: its six 10-minute KLL panes become one definition with offsets 0..50 min. --- Cargo.lock | 147 ++++++++++++- data_plane/Cargo.toml | 3 + data_plane/src/lib.rs | 1 + data_plane/src/sds.rs | 252 +++++++++++++++++++++++ data_plane/tests/sds.rs | 151 ++++++++++++++ examples/planner-layering-example3b.json | 44 ++++ examples/planner-layering-example4b.json | 44 ++++ 7 files changed, 641 insertions(+), 1 deletion(-) create mode 100644 data_plane/src/sds.rs create mode 100644 data_plane/tests/sds.rs create mode 100644 examples/planner-layering-example3b.json create mode 100644 examples/planner-layering-example4b.json diff --git a/Cargo.lock b/Cargo.lock index 12a44ff1..47258aa7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -82,6 +82,22 @@ version = "1.0.104" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "330a5ed07fa54e4702c9d6c4174f74427fc0ef6e214bbd677ae50a5099946470" +[[package]] +name = "asap-executor" +version = "0.1.0" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=58999df294fc9b0326341c1f20be62ecc8c2975e#58999df294fc9b0326341c1f20be62ecc8c2975e" +dependencies = [ + "asap-types", + "asap_sketchlib", + "chrono", + "futures", + "regex", + "serde", + "serde_json", + "thiserror 2.0.20", + "tracing", +] + [[package]] name = "asap-frontend-common" version = "0.1.0" @@ -265,6 +281,15 @@ 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" @@ -326,7 +351,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06" dependencies = [ "cfg-if", - "cpufeatures", + "cpufeatures 0.3.1", "rand_core 0.10.1", ] @@ -413,6 +438,15 @@ 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" @@ -422,11 +456,23 @@ dependencies = [ "libc", ] +[[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" dependencies = [ "anyhow", + "asap-executor", + "asap-types", "axum", "clap", "control_plane", @@ -435,6 +481,7 @@ dependencies = [ "reqwest", "serde", "serde_json", + "sha2", "snap", "thiserror 1.0.69", "tokio", @@ -449,6 +496,16 @@ 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" @@ -513,6 +570,21 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "futures" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a31d2a3fbaaeb2af2368bbdd904aa8e812d3c04a1ee10d3171f52d556e5d0a3" +dependencies = [ + "futures-channel", + "futures-core", + "futures-executor", + "futures-io", + "futures-sink", + "futures-task", + "futures-util", +] + [[package]] name = "futures-channel" version = "0.3.34" @@ -520,6 +592,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b1f9e3d69d39e4862ffed03ed071a76f9a13ba1d9109d355b0f0aa6b15e393c4" dependencies = [ "futures-core", + "futures-sink", ] [[package]] @@ -528,6 +601,40 @@ version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e" +[[package]] +name = "futures-executor" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "031b47cf1a3c6cc8bc2fc76cd437f521619387907d469316e7c0bc278f1f5432" +dependencies = [ + "futures-core", + "futures-task", + "futures-util", +] + +[[package]] +name = "futures-io" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53c0fa8157de1303bfffdaa1cc2a673bfffb60102f76b0ef4441659124373fed" + +[[package]] +name = "futures-macro" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9fb9654ba8355388abeb8dcb4fc62f511300867002afc858860463bdd9fe0c44" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.5", +] + +[[package]] +name = "futures-sink" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1944426bf7d03f1d14f708785e4b33efd750b36d48a157b836b3efc15ede8e1d" + [[package]] name = "futures-task" version = "0.3.34" @@ -540,12 +647,27 @@ version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0d50a92467f8ba5dd6e3ee5d4bd04d73ab2e4e1c44474a0674821dfce14b79bc" dependencies = [ + "futures-channel", "futures-core", + "futures-io", + "futures-macro", + "futures-sink", "futures-task", + "memchr", "pin-project-lite", "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" @@ -1562,6 +1684,17 @@ 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" @@ -1960,6 +2093,12 @@ 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" @@ -2019,6 +2158,12 @@ dependencies = [ "time", ] +[[package]] +name = "version_check" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" + [[package]] name = "vob" version = "3.0.6" diff --git a/data_plane/Cargo.toml b/data_plane/Cargo.toml index 2aaf8975..cf8e346d 100644 --- a/data_plane/Cargo.toml +++ b/data_plane/Cargo.toml @@ -5,6 +5,9 @@ edition.workspace = true [dependencies] control_plane.workspace = true +planner-types.workspace = true +asap-executor.workspace = true +sha2 = "0.10" serde.workspace = true serde_json.workspace = true thiserror.workspace = true diff --git a/data_plane/src/lib.rs b/data_plane/src/lib.rs index fd483635..7180dacf 100644 --- a/data_plane/src/lib.rs +++ b/data_plane/src/lib.rs @@ -6,6 +6,7 @@ //! query is forwarded to Prometheus. pub mod remote_write; +pub mod sds; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; diff --git a/data_plane/src/sds.rs b/data_plane/src/sds.rs new file mode 100644 index 00000000..05ca7478 --- /dev/null +++ b/data_plane/src/sds.rs @@ -0,0 +1,252 @@ +//! Stored summaries (SDS): what the data plane stores for a plan's +//! ingestion-time outputs, and how a query finds them again. +//! +//! The Planner's selected DAG marks each output computed at ingestion time +//! (`MatAtIngestion`) and read at query time: the executor's materialization +//! frontier. The control plane installs the DAG with each `PaneChain` +//! expanded into its panes, so each such output reads one tumbling pane of +//! raw samples: +//! +//! ```text +//! frontier node <- ... <- TimeRange(width) <- TimeShift(offset)* <- Scan +//! ``` +//! +//! At evaluation time `T` it covers `[T - offset - width, T - offset)`. +//! Outputs that differ only in `offset` compute the same thing over different +//! panes, so they share one [`SummaryDefinition`]: the data plane builds each +//! pane of a definition once, and every output reads the pane it needs. + +use std::collections::{BTreeMap, HashMap}; + +use planner_types::ir::physical_export::{PhysicalASAPDAG, PhysicalASAPDAGNode}; +use planner_types::ir::schema::Schema; +use planner_types::ir::{NonASAPOp, Operator}; +use planner_types::physical::Materialization; +use sha2::{Digest, Sha256}; + +/// Identifies what a stored summary computes, independent of node ids and of +/// which pane an output reads. Equal for equal computations across plans. +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct DefinitionId(pub String); + +/// One computation the data plane maintains pane by pane. +#[derive(Debug, Clone, PartialEq)] +pub struct SummaryDefinition { + pub id: DefinitionId, + /// Pane width: each stored summary covers `[end - pane_ms, end)`. + pub pane_ms: i64, + /// The frontier 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 raw input of the build: the `TimeRange` node fed one pane's samples. + pub raw_input: u64, + pub output_schema: Schema, +} + +/// How one frontier output reads its definition: at evaluation time `T`, the +/// pane ending at `T - offset_ms`. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct PaneBinding { + pub offset_ms: i64, +} + +/// The stored summaries of one plan. +#[derive(Debug, Clone, PartialEq, Default)] +pub struct Sds { + pub definitions: BTreeMap, + /// Every frontier output, with its definition and pane. + pub outputs: BTreeMap, +} + +#[derive(Debug, thiserror::Error, PartialEq)] +pub enum SdsError { + #[error("node {0} is not in the DAG")] + UnknownNode(u64), + #[error("ingestion-time output {node} does not read one pane of one Scan: {reason}")] + UnsupportedShape { node: u64, reason: String }, + /// The data plane keeps no query-time result across evaluations; it + /// declares `query_time_results_retained: false`, so the Planner should + /// not select such a plan. + #[error("node {0} is kept across evaluations (MatAtQueryKeptAcrossEvaluations), which the data plane does not support")] + KeptAcrossEvaluations(u64), +} + +impl Sds { + /// The definitions behind `frontier`, the DAG outputs computed at + /// ingestion time and read at query time. + pub fn from_dag(dag: &PhysicalASAPDAG, frontier: &[u64]) -> Result { + let nodes: HashMap = + dag.nodes.iter().map(|n| (n.id as u64, n)).collect(); + let mut inputs: HashMap> = HashMap::new(); + for edge in &dag.edges { + inputs + .entry(edge.consumer as u64) + .or_default() + .push(edge.producer as u64); + } + if let Some(node) = dag.nodes.iter().find(|n| { + matches!( + n.materialization, + Materialization::MatAtQueryKeptAcrossEvaluations(_) + ) + }) { + return Err(SdsError::KeptAcrossEvaluations(node.id as u64)); + } + let mut sds = Sds::default(); + let mut hashes = HashMap::new(); + for &node in frontier { + let (raw_input, pane_ms, offset_ms) = pane_of(node, &nodes, &inputs)?; + let id = DefinitionId(hex(&identity(node, &nodes, &inputs, &mut hashes)?)); + let output_schema = nodes[&node].output_schema.clone(); + let definition = + sds.definitions + .entry(id.clone()) + .or_insert_with(|| SummaryDefinition { + id: id.clone(), + pane_ms, + builder: node, + raw_input, + output_schema, + }); + if node < definition.builder { + definition.builder = node; + definition.raw_input = raw_input; + } + sds.outputs.insert(node, (id, PaneBinding { offset_ms })); + } + Ok(sds) + } +} + +/// `(raw input, pane width, offset)` of a frontier output: follow its single +/// inputs down to the `TimeRange`, then sum the `TimeShift`s down to the `Scan`. +fn pane_of( + node: u64, + nodes: &HashMap, + inputs: &HashMap>, +) -> Result<(u64, i64, i64), SdsError> { + let unsupported = |reason: &str| SdsError::UnsupportedShape { + node, + reason: reason.into(), + }; + let only_input = |id: u64| match inputs.get(&id).map(Vec::as_slice) { + Some([input]) => Ok(*input), + _ => Err(unsupported( + "every operator above the Scan must have one input", + )), + }; + let mut current = node; + let (raw_input, pane_ms) = loop { + match relational(nodes, current)? { + Some(NonASAPOp::TimeRange { range, .. }) => { + break (current, range.as_millis() as i64); + } + Some(NonASAPOp::TimeShift { .. }) => { + return Err(unsupported("a TimeShift above the TimeRange")); + } + Some(NonASAPOp::Scan { .. }) => return Err(unsupported("no TimeRange")), + _ => current = only_input(current)?, + } + }; + let mut offset_ms = 0; + current = only_input(raw_input)?; + loop { + match relational(nodes, current)? { + Some(NonASAPOp::TimeShift { shift, .. }) if shift.at.is_none() => { + offset_ms += shift.offset_ms; + current = only_input(current)?; + } + Some(NonASAPOp::Scan { .. }) => return Ok((raw_input, pane_ms, offset_ms)), + _ => { + return Err(unsupported( + "only offset TimeShifts between TimeRange and Scan", + )) + } + } + } +} + +fn relational<'a>( + nodes: &HashMap, + id: u64, +) -> Result>, SdsError> { + match &nodes.get(&id).ok_or(SdsError::UnknownNode(id))?.payload { + Operator::NonASAP(operator) => Ok(Some(operator)), + Operator::ASAP(_) => Ok(None), + } +} + +/// 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)?); + } + let hash: [u8; 32] = hasher.finalize().into(); + hashes.insert(node, hash); + Ok(hash) +} + +fn hex(bytes: &[u8]) -> String { + bytes.iter().map(|b| format!("{b:02x}")).collect() +} + +/// Built panes, by definition and pane end. A pane is stored once, when it +/// is complete, and never changes. +#[derive(Debug, Default)] +pub struct SummaryStore { + panes: BTreeMap<(DefinitionId, i64), S>, +} + +impl SummaryStore { + pub fn new() -> Self { + Self { + panes: BTreeMap::new(), + } + } + + pub fn insert(&mut self, definition: &DefinitionId, end_ms: i64, state: S) { + self.panes.insert((definition.clone(), end_ms), state); + } + + pub fn get(&self, definition: &DefinitionId, end_ms: i64) -> Option<&S> { + self.panes.get(&(definition.clone(), end_ms)) + } + + /// Drops the panes of `definition` that end at or before `end_ms`. + pub fn evict_through(&mut self, definition: &DefinitionId, end_ms: i64) { + self.panes + .retain(|(id, end), _| id != definition || *end > end_ms); + } + + pub fn len(&self) -> usize { + self.panes.len() + } + + pub fn is_empty(&self) -> bool { + self.panes.is_empty() + } +} diff --git a/data_plane/tests/sds.rs b/data_plane/tests/sds.rs new file mode 100644 index 00000000..a25a2d07 --- /dev/null +++ b/data_plane/tests/sds.rs @@ -0,0 +1,151 @@ +use std::path::Path; + +use asap_executor::physical_planner::frontier_from_timing; +use control_plane::{plan_workload, DeploymentInputs, InstalledPlan}; +use data_plane::sds::{DefinitionId, Sds, SdsError, SummaryStore}; +use planner_types::ir::physical_export::PhysicalASAPDAG; +use planner_types::ir::{NonASAPOp, Operator}; +use planner_types::physical::{Materialization, Medium}; + +/// The selected plan of a #509 example workload in `examples/`, for a +/// deployment that keeps raw data or not. +fn plan(example: &str, raw_data_retained: bool) -> InstalledPlan { + let path = Path::new(env!("CARGO_MANIFEST_DIR")) + .join(format!("../examples/planner-layering-{example}.json")); + let workload = serde_json::from_str(&std::fs::read_to_string(path).unwrap()).unwrap(); + let mut inputs = DeploymentInputs::default(); + inputs.deployment_capabilities.raw_data_retained = raw_data_retained; + plan_workload(&workload, &inputs).unwrap() +} + +fn sds_of(dag: &PhysicalASAPDAG) -> Result { + Sds::from_dag(dag, &frontier_from_timing(dag).unwrap()) +} + +/// 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. +#[test] +fn example4b_panes_share_one_definition() { + let plan = plan("example4b", false); + let sds = sds_of(&plan.dag.dag).unwrap(); + assert_eq!(sds.definitions.len(), 1); + let definition = sds.definitions.values().next().unwrap(); + assert_eq!(definition.pane_ms, 600_000); + let mut offsets: Vec = sds + .outputs + .values() + .map(|(id, binding)| { + assert_eq!(id, &definition.id); + binding.offset_ms + }) + .collect(); + offsets.sort(); + assert_eq!(offsets, (0..6).map(|i| i * 600_000).collect::>()); + let raw = plan + .dag + .dag + .nodes + .iter() + .find(|n| n.id as u64 == definition.raw_input) + .unwrap(); + assert!(matches!( + raw.payload, + Operator::NonASAP(NonASAPOp::TimeRange { .. }) + )); +} + +/// A plan computed entirely at query time stores nothing. +#[test] +fn query_time_plans_have_no_definitions() { + let plan = plan("example1", true); + assert_eq!(sds_of(&plan.dag.dag).unwrap(), Sds::default()); +} + +/// Identity depends on what is computed, not on how nodes are numbered. +#[test] +fn identity_ignores_node_numbering() { + let dag = plan("example4b", false).dag.dag; + let mut renumbered = dag.clone(); + let renumber = |id: &usize| id + 100; + for node in &mut renumbered.nodes { + node.id = renumber(&node.id); + node.payload = node.payload.map_children(renumber); + } + for edge in &mut renumbered.edges { + edge.producer = renumber(&edge.producer); + edge.consumer = renumber(&edge.consumer); + } + renumbered.roots = renumbered.roots.iter().map(renumber).collect(); + let ids = |sds: Sds| sds.definitions.into_keys().collect::>(); + assert_eq!( + ids(sds_of(&renumbered).unwrap()), + ids(sds_of(&dag).unwrap()) + ); +} + +/// The same computation over another metric is another definition. +#[test] +fn identity_depends_on_the_source() { + let dag = plan("example4b", false).dag.dag; + let other: PhysicalASAPDAG = serde_json::from_str( + &serde_json::to_string(&dag) + .unwrap() + .replace("latency_ms", "other"), + ) + .unwrap(); + assert_ne!( + sds_of(&other).unwrap().definitions.into_keys().next(), + sds_of(&dag).unwrap().definitions.into_keys().next() + ); +} + +/// An ingestion-time output must read a pane of raw samples. +#[test] +fn rejects_outputs_without_a_pane() { + let dag = plan("example4b", false).dag.dag; + let scan = dag + .nodes + .iter() + .find(|n| matches!(n.payload, Operator::NonASAP(NonASAPOp::Scan { .. }))) + .unwrap(); + let scan = scan.id as u64; + assert_eq!( + Sds::from_dag(&dag, &[scan]), + Err(SdsError::UnsupportedShape { + node: scan, + reason: "no TimeRange".into() + }) + ); +} + +/// A plan that keeps a query-time result across evaluations is not +/// installed: the data plane declares it cannot keep one. +#[test] +fn rejects_results_kept_across_evaluations() { + let mut dag = plan("example4b", true).dag.dag; + let node = &mut dag.nodes[0]; + node.materialization = Materialization::MatAtQueryKeptAcrossEvaluations(Medium::Memory); + let id = node.id as u64; + assert_eq!( + Sds::from_dag(&dag, &[]), + Err(SdsError::KeptAcrossEvaluations(id)) + ); +} + +/// Panes are found by definition and end, and evicted once no query needs them. +#[test] +fn store_finds_and_evicts_panes() { + let (a, b) = (DefinitionId("a".into()), DefinitionId("b".into())); + let mut store = SummaryStore::new(); + for end in [60, 120, 180] { + store.insert(&a, end, format!("a{end}")); + } + store.insert(&b, 60, "b60".to_string()); + assert_eq!(store.get(&a, 120).map(String::as_str), Some("a120")); + assert_eq!(store.get(&a, 90), None); + store.evict_through(&a, 120); + assert_eq!(store.get(&a, 120), None); + assert_eq!(store.get(&a, 180).map(String::as_str), Some("a180")); + assert_eq!(store.get(&b, 60).map(String::as_str), Some("b60")); + assert_eq!(store.len(), 2); +} diff --git a/examples/planner-layering-example3b.json b/examples/planner-layering-example3b.json new file mode 100644 index 00000000..c943e415 --- /dev/null +++ b/examples/planner-layering-example3b.json @@ -0,0 +1,44 @@ +{ + "query_workload": { + "language": "promql", + "query_batch": null, + "repeating_queries": [ + { + "query": "quantile_over_time(0.99, latency_ms[5m])", + "demand": { + "fixed_interval": 60000 + }, + "requirements": { + "accuracy": { + "accuracy_metric": "rank", + "max_error": 0.01, + "failure_probability": 0.01, + "accuracy_scope": "whole_result" + }, + "max_latency_ms": 200.0 + }, + "predictability": { + "predictable": { + "known_at": null + } + } + } + ] + }, + "data_workload": { + "arrival": { + "ingesting": { + "rows_per_second": 66666.66666666667 + } + }, + "sources": { + "latency_ms": { + "series_count": 1000000, + "metric_type": null, + "sample_interval": 15000, + "stream_norms": null, + "max_distinct_per_evaluation": null + } + } + } +} diff --git a/examples/planner-layering-example4b.json b/examples/planner-layering-example4b.json new file mode 100644 index 00000000..ab13a77f --- /dev/null +++ b/examples/planner-layering-example4b.json @@ -0,0 +1,44 @@ +{ + "query_workload": { + "language": "promql", + "query_batch": null, + "repeating_queries": [ + { + "query": "quantile_over_time(0.99, latency_ms[60m])", + "demand": { + "fixed_interval": 600000 + }, + "requirements": { + "accuracy": { + "accuracy_metric": "rank", + "max_error": 0.01, + "failure_probability": 0.01, + "accuracy_scope": "whole_result" + }, + "max_latency_ms": 200.0 + }, + "predictability": { + "predictable": { + "known_at": null + } + } + } + ] + }, + "data_workload": { + "arrival": { + "ingesting": { + "rows_per_second": 1000.0 + } + }, + "sources": { + "latency_ms": { + "series_count": 1000, + "metric_type": null, + "sample_interval": 1000, + "stream_norms": null, + "max_distinct_per_evaluation": null + } + } + } +}