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": {} + } +}