From bce77c9e1f8fe79a5a1643f90f91551a14e3ab98 Mon Sep 17 00:00:00 2001 From: zzylol Date: Wed, 7 Oct 2026 03:07:54 +0000 Subject: [PATCH] feat(sds): define stored summaries from the plan's ingestion-time outputs Each ingestion-time 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 TimeShift offsets zeroed. 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. #509 Example 4b's six 10-minute KLL panes become one definition with offsets 0..50 min. Co-Authored-By: Claude Opus 5.5 --- Cargo.lock | 108 ++++++++--- data_plane/Cargo.toml | 4 +- data_plane/src/lib.rs | 1 + data_plane/src/sds.rs | 232 +++++++++++++++++++++++ data_plane/tests/sds.rs | 145 ++++++++++++++ examples/planner-layering-example3b.json | 71 +++++++ examples/planner-layering-example4b.json | 71 +++++++ 7 files changed, 604 insertions(+), 28 deletions(-) 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 cac83d6d..8ca158eb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -295,6 +295,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" @@ -356,7 +365,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", ] @@ -443,6 +452,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" @@ -452,11 +470,22 @@ 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", @@ -466,6 +495,7 @@ dependencies = [ "reqwest", "serde", "serde_json", + "sha2", "snap", "thiserror 1.0.69", "tokio", @@ -480,6 +510,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" @@ -510,7 +550,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -632,6 +672,16 @@ 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" @@ -791,7 +841,7 @@ dependencies = [ "libc", "percent-encoding", "pin-project-lite", - "socket2 0.5.10", + "socket2", "tokio", "tower-service", "tracing", @@ -1110,7 +1160,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -1276,7 +1326,7 @@ dependencies = [ "quinn-udp", "rustc-hash", "rustls", - "socket2 0.5.10", + "socket2", "thiserror 2.0.20", "tokio", "tracing", @@ -1314,9 +1364,9 @@ dependencies = [ "cfg_aliases", "libc", "once_cell", - "socket2 0.5.10", + "socket2", "tracing", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -1648,6 +1698,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" @@ -1691,16 +1752,6 @@ version = "1.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "199905e6153d6405f9728fe44daace35f8f837bbf830bb6e85fbd5828709a886" -[[package]] -name = "socket2" -version = "0.5.10" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e22376abed350d73dd1cd119b57ffccad95b4e585a7cda43e286245ce23c0678" -dependencies = [ - "libc", - "windows-sys 0.52.0", -] - [[package]] name = "socket2" version = "0.6.5" @@ -1907,7 +1958,7 @@ dependencies = [ "parking_lot", "pin-project-lite", "signal-hook-registry", - "socket2 0.6.5", + "socket2", "tokio-macros", "windows-sys 0.61.2", ] @@ -2056,6 +2107,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" @@ -2115,6 +2172,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" @@ -2311,15 +2374,6 @@ dependencies = [ "windows-targets", ] -[[package]] -name = "windows-sys" -version = "0.59.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1e38bc4d79ed67fd075bcc251a1c39b32a1776bbe92e5bef1f0bf1f8c531853b" -dependencies = [ - "windows-targets", -] - [[package]] name = "windows-sys" version = "0.61.2" diff --git a/data_plane/Cargo.toml b/data_plane/Cargo.toml index d11d1479..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 @@ -19,6 +22,5 @@ prost = "0.13" snap = "1.1" [dev-dependencies] -planner-types.workspace = true tower = { version = "0.5", features = ["util"] } http-body-util = "0.1" 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..4dc662b0 --- /dev/null +++ b/data_plane/src/sds.rs @@ -0,0 +1,232 @@ +//! 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 and +//! read at query time (the executor's materialization frontier). 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::export::{LogicalASAPOperatorPayload as Payload, NonASAPOpKind}; +use planner_types::ir::physical_export::{PhysicalASAPDAG, PhysicalASAPDAGNode}; +use planner_types::ir::schema::Schema; +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 }, +} + +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| (u64::from(n.id.0), n)).collect(); + let mut inputs: HashMap> = HashMap::new(); + for edge in &dag.edges { + inputs + .entry(u64::from(edge.consumer.0)) + .or_default() + .push(u64::from(edge.producer.0)); + } + 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(NonASAPOpKind::TimeRange { range, .. }) => { + break (current, range.as_millis() as i64); + } + Some(NonASAPOpKind::TimeShift { .. }) => { + return Err(unsupported("a TimeShift above the TimeRange")); + } + Some(NonASAPOpKind::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(NonASAPOpKind::TimeShift { shift }) if shift.at.is_none() => { + offset_ms += shift.offset_ms; + current = only_input(current)?; + } + Some(NonASAPOpKind::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 { + Payload::Relational { operator } => Ok(Some(operator)), + _ => Ok(None), + } +} + +/// SHA-256 over a node's payload and output schema, then its inputs' +/// identities in edge order. `TimeShift` offsets are zeroed, so outputs that +/// read different panes of the same computation share an identity. +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))?; + let mut payload = dag_node.payload.clone(); + if let Payload::Relational { + operator: NonASAPOpKind::TimeShift { shift }, + } = &mut payload + { + shift.offset_ms = 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..7dffbba1 --- /dev/null +++ b/data_plane/tests/sds.rs @@ -0,0 +1,145 @@ +use std::path::Path; + +use asap_executor::physical_planner::frontier_from_timing; +use control_plane::{backend_capabilities, plan_workload, InstalledPlan}; +use data_plane::sds::{DefinitionId, Sds, SdsError, SummaryStore}; +use planner_types::ir::export::{LogicalASAPOperatorPayload as Payload, NonASAPOpKind}; +use planner_types::ir::physical_export::PhysicalASAPDAG; + +/// The selected plan of a #509 example workload in `examples/`. +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 capabilities = planner_types::deployment::DeploymentCapabilities { + raw_data_retained, + ..backend_capabilities() + }; + plan_workload(&workload, &capabilities).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| u64::from(n.id.0) == definition.raw_input) + .unwrap(); + assert!(matches!( + raw.payload, + Payload::Relational { + operator: NonASAPOpKind::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 renumbered: PhysicalASAPDAG = serde_json::from_str( + &serde_json::to_string(&dag) + .unwrap() + // Node ids appear as `"id":n`, `"producer":n`, `"consumer":n` and roots. + .replace("\"id\":", "\"id\":1") + .replace("\"producer\":", "\"producer\":1") + .replace("\"consumer\":", "\"consumer\":1") + .replace("\"roots\":[", "\"roots\":[1"), + ) + .unwrap(); + 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, + Payload::Relational { + operator: NonASAPOpKind::Scan { .. } + } + ) + }) + .unwrap(); + let scan = u64::from(scan.id.0); + assert_eq!( + Sds::from_dag(&dag, &[scan]), + Err(SdsError::UnsupportedShape { + node: scan, + reason: "no TimeRange".into() + }) + ); +} + +/// 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..94b55608 --- /dev/null +++ b/examples/planner-layering-example3b.json @@ -0,0 +1,71 @@ +{ + "query_workload": { + "language": "promql", + "query_batch": null, + "repeating_queries": [ + { + "query": "quantile_over_time(0.99, latency_ms[5m])", + "demand": { + "fixed_interval": 60000 + }, + "requirements": { + "accuracy": { + "explicit": { + "EpsilonDelta": { + "epsilon": 0.01, + "delta": 0.01 + } + } + }, + "response_latency": { + "explicit_max_ms": 200.0 + } + }, + "predictability": { + "predictable": { + "known_at": null + } + }, + "time_selection": { + "scope": "real_time", + "lookback": 300000, + "as_of": null + } + } + ] + }, + "data_workload": { + "arrival": "continuously_ingesting", + "data_ingestion_interval": { + "value": 15000, + "source": "declared", + "observed_at_ms": null, + "valid_for_ms": null + }, + "ingestion_volume": { + "value": null, + "source": "unknown", + "observed_at_ms": null, + "valid_for_ms": null + }, + "ingestion_rate": { + "value": 66666.66666666667, + "source": "declared", + "observed_at_ms": null, + "valid_for_ms": null + }, + "input_cardinality": { + "value": 1000000, + "source": "declared", + "observed_at_ms": null, + "valid_for_ms": null + }, + "distribution": { + "value": "zipf", + "source": "declared", + "observed_at_ms": null, + "valid_for_ms": null + }, + "metric_types": {} + } +} diff --git a/examples/planner-layering-example4b.json b/examples/planner-layering-example4b.json new file mode 100644 index 00000000..81aaceb5 --- /dev/null +++ b/examples/planner-layering-example4b.json @@ -0,0 +1,71 @@ +{ + "query_workload": { + "language": "promql", + "query_batch": null, + "repeating_queries": [ + { + "query": "quantile_over_time(0.99, latency_ms[60m])", + "demand": { + "fixed_interval": 600000 + }, + "requirements": { + "accuracy": { + "explicit": { + "EpsilonDelta": { + "epsilon": 0.01, + "delta": 0.01 + } + } + }, + "response_latency": { + "explicit_max_ms": 200.0 + } + }, + "predictability": { + "predictable": { + "known_at": null + } + }, + "time_selection": { + "scope": "real_time", + "lookback": 3600000, + "as_of": null + } + } + ] + }, + "data_workload": { + "arrival": "continuously_ingesting", + "data_ingestion_interval": { + "value": 1000, + "source": "declared", + "observed_at_ms": null, + "valid_for_ms": null + }, + "ingestion_volume": { + "value": null, + "source": "unknown", + "observed_at_ms": null, + "valid_for_ms": null + }, + "ingestion_rate": { + "value": 1000.0, + "source": "declared", + "observed_at_ms": null, + "valid_for_ms": null + }, + "input_cardinality": { + "value": 1000, + "source": "declared", + "observed_at_ms": null, + "valid_for_ms": null + }, + "distribution": { + "value": "zipf", + "source": "declared", + "observed_at_ms": null, + "valid_for_ms": null + }, + "metric_types": {} + } +}