From 72f7618b3a74b29dccc9b9825f9e5d44750ba0bd Mon Sep 17 00:00:00 2001 From: zzylol Date: Wed, 21 Oct 2026 00:12:02 +0000 Subject: [PATCH 1/2] perf(data-plane): intern series at decode; key per-series structures by id Each label set is interned once, when Remote Write or a raw-sample fetch delivers it: an ahash of its label bytes finds or creates one shared Series with a dense u32 id and its metric. Samples carry the Arc; the raw-sample window's per-series cells, the per-sample group numbers and the instant selector's latest-sample map key by the id, and the metric is read from the series instead of the label map. --- Cargo.lock | 1 + data_plane/Cargo.toml | 1 + data_plane/src/lib.rs | 14 +-- data_plane/src/per_sample.rs | 12 +-- data_plane/src/precompute.rs | 9 +- data_plane/src/prometheus.rs | 3 +- data_plane/src/series.rs | 35 ++++---- data_plane/src/series_ids.rs | 154 +++++++++++++++++++++++++++++++++ data_plane/tests/per_sample.rs | 3 +- data_plane/tests/precompute.rs | 3 +- data_plane/tests/query.rs | 3 +- data_plane/tests/query_run.rs | 3 +- data_plane/tests/series_ids.rs | 52 +++++++++++ 13 files changed, 247 insertions(+), 46 deletions(-) create mode 100644 data_plane/src/series_ids.rs create mode 100644 data_plane/tests/series_ids.rs diff --git a/Cargo.lock b/Cargo.lock index 5920ce46..053ddae8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1037,6 +1037,7 @@ dependencies = [ name = "data_plane" version = "0.1.0" dependencies = [ + "ahash", "anyhow", "asap-executor", "asap-planner", diff --git a/data_plane/Cargo.toml b/data_plane/Cargo.toml index 8ce0da62..06ab9f98 100644 --- a/data_plane/Cargo.toml +++ b/data_plane/Cargo.toml @@ -7,6 +7,7 @@ edition.workspace = true asap-planner.workspace = true asap-executor.workspace = true sha2 = "0.10" +ahash = "0.8" 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 869f3604..07633be8 100644 --- a/data_plane/src/lib.rs +++ b/data_plane/src/lib.rs @@ -14,6 +14,7 @@ pub mod query; pub mod remote_write; pub mod sds; mod series; +pub mod series_ids; use std::collections::BTreeMap; use std::sync::atomic::{AtomicU64, Ordering}; @@ -118,13 +119,12 @@ async fn write(State(state): State>, body: Bytes) -> Response { Ok(request) => { let mut samples = Vec::new(); for series in request.timeseries { - let labels: Arc> = Arc::new( - series - .labels - .into_iter() - .map(|label| (label.name, label.value)) - .collect(), - ); + let pairs: Vec<(&str, &str)> = series + .labels + .iter() + .map(|label| (label.name.as_str(), label.value.as_str())) + .collect(); + let labels = series_ids::intern_pairs(&pairs); // Stale markers are kept: the executor's sample selector // drops them from range windows and ends instant selectors. samples.extend(series.samples.into_iter().map(|s| RawSample { diff --git a/data_plane/src/per_sample.rs b/data_plane/src/per_sample.rs index 1dbccf3d..cc8c8505 100644 --- a/data_plane/src/per_sample.rs +++ b/data_plane/src/per_sample.rs @@ -22,7 +22,7 @@ //! time, or groups that are not series. The //! Planner places no such build at ingestion time for the PromQL examples. -use std::collections::{BTreeMap, HashMap}; +use std::collections::BTreeMap; use std::sync::{Arc, OnceLock}; use asap_executor::arrow::array::{Array, TimestampMillisecondArray, UInt32Array}; @@ -46,7 +46,7 @@ type GroupKey = Arc<[Arc]>; /// Groups numbered `0..`, in the order they are first seen. #[derive(Default)] struct GroupNumbers { - numbers: HashMap, + numbers: ahash::HashMap, keys: Vec, } @@ -83,8 +83,8 @@ pub(crate) struct PerSampleSpec { pub(crate) struct OpenPanes { /// The numbers of the groups (series) seen. numbers: GroupNumbers, - /// Each series' group key and number. - series_groups: HashMap>, (GroupKey, u32)>, + /// Each series' group key and number, by series id. + series_groups: ahash::HashMap, /// By pane end: its states and the groups that took a sample, or why it /// failed. panes: BTreeMap), String>>, @@ -213,7 +213,7 @@ impl PerSampleSpec { if is_stale(sample.value) { return; } - let group = match open.series_groups.get(&sample.labels) { + let group = match open.series_groups.get(&sample.labels.id) { Some(group) => Some(group.clone()), None => { let key: GroupKey = self @@ -223,7 +223,7 @@ impl PerSampleSpec { .collect(); let number = open.numbers.number(&key); open.series_groups - .insert(sample.labels.clone(), (key.clone(), number)); + .insert(sample.labels.id, (key.clone(), number)); Some((key, number)) } }; diff --git a/data_plane/src/precompute.rs b/data_plane/src/precompute.rs index 90631bc8..4fdf3239 100644 --- a/data_plane/src/precompute.rs +++ b/data_plane/src/precompute.rs @@ -38,12 +38,13 @@ use crate::sds::{ DefinitionId, Sds, Selection, SummaryDefinition, SummaryInstanceKey, SummaryStore, }; use crate::series::SeriesWindow; +use crate::series_ids::Series; -/// One raw sample: its series' full label set (including `__name__`), time -/// and value. +/// One raw sample: its series (interned: its id and full label set, +/// `__name__` included), time and value. #[derive(Debug, Clone, PartialEq)] pub struct RawSample { - pub labels: Arc>, + pub labels: Arc, pub timestamp_ms: i64, pub value: f64, } @@ -395,7 +396,7 @@ impl Precompute { self.newest_clock_ms = self.newest_clock_ms.max(Some(now_ms)); let latest_allowed_ms = now_ms.saturating_add(self.max_future_ms); for sample in samples { - let metric = sample.labels.get("__name__").map(String::as_str); + let metric = Some(&*sample.labels.metric).filter(|m| !m.is_empty()); let query_run = metric.is_some_and(|m| self.query_run_metrics.contains_key(m)); if !query_run && !self diff --git a/data_plane/src/prometheus.rs b/data_plane/src/prometheus.rs index 2bbd90dd..bdee9247 100644 --- a/data_plane/src/prometheus.rs +++ b/data_plane/src/prometheus.rs @@ -2,7 +2,6 @@ //! parameters and results, and fetching raw samples for query-time ranges. use std::collections::BTreeMap; -use std::sync::Arc; use serde_json::{json, Value}; @@ -159,7 +158,7 @@ pub async fn fetch( let mut samples = Vec::new(); for series in series { let labels: BTreeMap = serde_json::from_value(series["metric"].clone())?; - let labels = Arc::new(labels); + let labels = crate::series_ids::intern(&labels); for point in series["values"].as_array().ok_or_else(malformed)? { let seconds = point[0].as_f64().ok_or_else(malformed)?; let value = point[1].as_str().ok_or_else(malformed)?; diff --git a/data_plane/src/series.rs b/data_plane/src/series.rs index 59215d54..9ffd9375 100644 --- a/data_plane/src/series.rs +++ b/data_plane/src/series.rs @@ -9,7 +9,6 @@ //! (`Batch::try_from_columns`): the columns the selector builds, without //! a `Sample` or a row per sample. -use std::collections::{BTreeMap, HashMap}; use std::sync::{Arc, RwLock}; use asap_executor::arrow::array::{ @@ -50,8 +49,9 @@ pub(crate) struct SeriesCells { pub(crate) texts: Vec>>, } -/// Each series' cells, or `None` when the `Scan` does not select it. -type SeriesCache = HashMap>, Option>>; +/// Each series' cells, or `None` when the `Scan` does not select it, by +/// series id. +type SeriesCache = ahash::HashMap>>; /// One raw-sample window of the plan, prepared to turn samples into its /// input batch. Shared by every thread that reads the window. @@ -107,7 +107,7 @@ impl SeriesWindow { kinds, width_ms: window.width_ms, latest_sample_per_series: *kind == TimeRangeKind::Instant, - series: RwLock::new(HashMap::new()), + series: RwLock::new(ahash::HashMap::default()), }) } @@ -123,7 +123,7 @@ impl SeriesWindow { .series .read() .unwrap_or_else(|e| e.into_inner()) - .get(&sample.labels) + .get(&sample.labels.id) { return cells.clone(); } @@ -131,7 +131,7 @@ impl SeriesWindow { self.series .write() .unwrap_or_else(|e| e.into_inner()) - .insert(sample.labels.clone(), cells.clone()); + .insert(sample.labels.id, cells.clone()); cells } @@ -139,7 +139,7 @@ impl SeriesWindow { // The selector applies the Scan's matchers to the series' labels; // the value is not the sample's, which may be a stale marker. let one = Sample { - labels: (*sample.labels).clone(), + labels: (**sample.labels).clone(), timestamp_ms: sample.timestamp_ms, value: 0.0, }; @@ -187,33 +187,30 @@ impl SeriesWindow { .filter(|s| start_ms < s.timestamp_ms && s.timestamp_ms <= end_ms); let mut points: Vec<(Arc, i64, f64)> = Vec::new(); if self.latest_sample_per_series { - let mut latest: BTreeMap<&BTreeMap, &RawSample> = BTreeMap::new(); + let mut latest: ahash::HashMap = ahash::HashMap::default(); for sample in in_window { - let entry = latest.entry(&sample.labels).or_insert(sample); + let entry = latest.entry(sample.labels.id).or_insert(sample); if sample.timestamp_ms >= entry.timestamp_ms { *entry = sample; } } // In label order, as the selector emits them. - for sample in latest.into_values().filter(|s| !is_stale(s.value)) { + let mut latest: Vec<&RawSample> = latest.into_values().collect(); + latest.sort_unstable_by(|a, b| a.labels.cmp(&b.labels)); + for sample in latest.into_iter().filter(|s| !is_stale(s.value)) { if let Some(cells) = self.cells(sample) { points.push((cells, sample.timestamp_ms, sample.value)); } } } else { // Samples of one series are together: look each up once. - let mut current: Option<(&Arc>, Option>)> = - None; + let mut current: Option<(u32, Option>)> = None; for sample in in_window.filter(|s| !is_stale(s.value)) { let cells = match ¤t { - Some((labels, cells)) - if Arc::ptr_eq(labels, &sample.labels) || **labels == sample.labels => - { - cells.clone() - } + Some((id, cells)) if *id == sample.labels.id => cells.clone(), _ => { let cells = self.cells(sample); - current = Some((&sample.labels, cells.clone())); + current = Some((sample.labels.id, cells.clone())); cells } }; @@ -260,7 +257,7 @@ impl SeriesWindow { pub(crate) fn text_column<'a>( texts: impl Iterator, ) -> Result { - let mut codes: HashMap<&str, u32> = HashMap::new(); + let mut codes: ahash::HashMap<&str, u32> = ahash::HashMap::default(); let mut values: Vec<&str> = Vec::new(); let keys: UInt32Array = texts .map(|text| { diff --git a/data_plane/src/series_ids.rs b/data_plane/src/series_ids.rs new file mode 100644 index 00000000..b5fc84e8 --- /dev/null +++ b/data_plane/src/series_ids.rs @@ -0,0 +1,154 @@ +//! Series interning: each label set is mapped to one shared [`Series`], with +//! a `u32` id, once, when Remote Write (or a raw-sample fetch from +//! Prometheus) delivers it. Everything downstream (the per-series cells of +//! each raw-sample window, the per-sample group numbers, the latest sample +//! per series of an instant selector) keys by the id, and the metric is read +//! from the series, so no stage builds, hashes or compares a label map per +//! sample. +//! +//! The interner is process-wide: ids are the same for a label set whichever +//! request, worker or query thread meets it. It never forgets a series. + +use std::collections::BTreeMap; +use std::hash::{BuildHasher, Hasher}; +use std::ops::Deref; +use std::sync::{Arc, OnceLock, RwLock}; + +use ahash::RandomState; + +/// One label set, interned. +#[derive(Debug)] +pub struct Series { + /// The series' id, dense from 0 in the order series are first seen. + pub id: u32, + /// The `__name__` label (empty without one). + pub metric: Arc, + labels: BTreeMap, +} + +impl Deref for Series { + type Target = BTreeMap; + + fn deref(&self) -> &Self::Target { + &self.labels + } +} + +impl PartialEq for Series { + fn eq(&self, other: &Self) -> bool { + self.id == other.id + } +} + +impl Eq for Series {} + +impl PartialOrd for Series { + fn partial_cmp(&self, other: &Self) -> Option { + Some(self.cmp(other)) + } +} + +impl Ord for Series { + /// By labels, as Prometheus orders series. + fn cmp(&self, other: &Self) -> std::cmp::Ordering { + if self.id == other.id { + return std::cmp::Ordering::Equal; + } + self.labels.cmp(&other.labels) + } +} + +impl std::hash::Hash for Series { + fn hash(&self, state: &mut H) { + state.write_u32(self.id); + } +} + +#[derive(Default)] +struct Interner { + /// Series ids by label-set hash; a hash collision lists several. + by_hash: ahash::HashMap>, + series: Vec>, +} + +fn interner() -> &'static RwLock { + static INTERNER: OnceLock> = OnceLock::new(); + INTERNER.get_or_init(|| RwLock::new(Interner::default())) +} + +/// Fixed seeds, so a label set hashes (and shards) the same in every run. +fn hasher() -> ahash::AHasher { + RandomState::with_seeds(0x5eed_0001, 0x5eed_0002, 0x5eed_0003, 0x5eed_0004).build_hasher() +} + +/// The hash of `pairs`, sorted by name: each name and value with its +/// length, so a byte moving across a boundary changes the hash. +fn hash_sorted<'a>(pairs: impl Iterator) -> u64 { + let mut hasher = hasher(); + for (name, value) in pairs { + hasher.write_usize(name.len()); + hasher.write(name.as_bytes()); + hasher.write_usize(value.len()); + hasher.write(value.as_bytes()); + } + hasher.finish() +} + +/// The series of the label pairs `pairs`, in any order (Remote Write sends +/// them sorted by name); interned on first sight. +pub fn intern_pairs, V: AsRef>(pairs: &[(N, V)]) -> Arc { + let mut sorted: Vec<(&str, &str)> = pairs + .iter() + .map(|(n, v)| (n.as_ref(), v.as_ref())) + .collect(); + if !sorted.windows(2).all(|w| w[0].0 <= w[1].0) { + sorted.sort_unstable_by(|a, b| a.0.cmp(b.0)); + } + let hash = hash_sorted(sorted.iter().copied()); + let matches = |series: &Series| { + series.labels.len() == sorted.len() + && series + .labels + .iter() + .zip(&sorted) + .all(|((n, v), (name, value))| n == name && v == value) + }; + { + let interner = interner().read().expect("interner lock"); + for &id in interner.by_hash.get(&hash).into_iter().flatten() { + let series = &interner.series[id as usize]; + if matches(series) { + return series.clone(); + } + } + } + let mut interner = interner().write().expect("interner lock"); + for &id in interner.by_hash.get(&hash).into_iter().flatten() { + let series = &interner.series[id as usize]; + if matches(series) { + return series.clone(); + } + } + let labels: BTreeMap = sorted + .iter() + .map(|(n, v)| (n.to_string(), v.to_string())) + .collect(); + let id = u32::try_from(interner.series.len()).expect("fewer than 2^32 series"); + let series = Arc::new(Series { + id, + metric: labels.get("__name__").map_or("", String::as_str).into(), + labels, + }); + interner.series.push(series.clone()); + interner.by_hash.entry(hash).or_default().push(id); + series +} + +/// The series of `labels`; interned on first sight. +pub fn intern(labels: &BTreeMap) -> Arc { + let pairs: Vec<(&str, &str)> = labels + .iter() + .map(|(n, v)| (n.as_str(), v.as_str())) + .collect(); + intern_pairs(&pairs) +} diff --git a/data_plane/tests/per_sample.rs b/data_plane/tests/per_sample.rs index af42e185..7af2e74a 100644 --- a/data_plane/tests/per_sample.rs +++ b/data_plane/tests/per_sample.rs @@ -5,7 +5,6 @@ mod common; use std::collections::BTreeMap; -use std::sync::Arc; use asap_executor::planner::ir::schema::{FieldDataType, SketchStatistic}; use asap_executor::summary_kernels::exact::ExactAccumulator; @@ -28,7 +27,7 @@ fn options() -> PrecomputeOptions { fn samples(metric: &str, jobs: usize, from: i64, to: i64) -> Vec { let series: Vec<_> = (0..jobs) .map(|j| { - Arc::new(BTreeMap::from([ + data_plane::series_ids::intern(&BTreeMap::from([ ("__name__".to_string(), metric.to_string()), ("job".to_string(), format!("job{j}")), ("instance".to_string(), format!("{}", j % 3)), diff --git a/data_plane/tests/precompute.rs b/data_plane/tests/precompute.rs index d2f078f6..dd2ebc6d 100644 --- a/data_plane/tests/precompute.rs +++ b/data_plane/tests/precompute.rs @@ -1,7 +1,6 @@ mod common; use std::collections::BTreeMap; -use std::sync::Arc; use asap_executor::planner::workload::{PlanningWorkload, Query}; use asap_planner::InstallablePlan; @@ -32,7 +31,7 @@ fn precompute_with(plan: InstallablePlan, allowed_lateness_ms: u64) -> (Precompu fn sample(metric: &str, job: &str, timestamp_ms: i64, value: f64) -> RawSample { RawSample { - labels: Arc::new(BTreeMap::from([ + labels: data_plane::series_ids::intern(&BTreeMap::from([ ("__name__".to_string(), metric.to_string()), ("job".to_string(), job.to_string()), ])), diff --git a/data_plane/tests/query.rs b/data_plane/tests/query.rs index 2e12f204..2fc3b87f 100644 --- a/data_plane/tests/query.rs +++ b/data_plane/tests/query.rs @@ -1,7 +1,6 @@ mod common; use std::collections::BTreeMap; -use std::sync::Arc; use asap_planner::InstallablePlan; use data_plane::precompute::{Precompute, PrecomputeOptions, RawSample}; @@ -42,7 +41,7 @@ fn samples( .map(|(metric, labels)| { let mut map = BTreeMap::from([("__name__".to_string(), metric.to_string())]); map.extend(labels.iter().map(|(k, v)| (k.to_string(), v.to_string()))); - Arc::new(map) + data_plane::series_ids::intern(&map) }) .collect(); (from..to) diff --git a/data_plane/tests/query_run.rs b/data_plane/tests/query_run.rs index d8547d03..d03d07d1 100644 --- a/data_plane/tests/query_run.rs +++ b/data_plane/tests/query_run.rs @@ -6,7 +6,6 @@ mod common; use std::collections::BTreeMap; -use std::sync::Arc; use asap_planner::{InstallablePlan, StoredOutputBuild}; use data_plane::engine::{EngineHandle, EngineOptions}; @@ -68,7 +67,7 @@ fn samples() -> Vec { let series: Vec<_> = ["a", "b"] .iter() .map(|job| { - Arc::new(BTreeMap::from([ + data_plane::series_ids::intern(&BTreeMap::from([ ("__name__".to_string(), "latency_ms".to_string()), ("job".to_string(), job.to_string()), ])) diff --git a/data_plane/tests/series_ids.rs b/data_plane/tests/series_ids.rs new file mode 100644 index 00000000..144c5834 --- /dev/null +++ b/data_plane/tests/series_ids.rs @@ -0,0 +1,52 @@ +//! Series are interned once, at decode: each label set gets one `u32` id +//! and one shared `Series`, which everything downstream keys by. + +use std::collections::BTreeMap; +use std::sync::Arc; + +use data_plane::series_ids::{intern, intern_pairs}; + +fn labels(pairs: &[(&str, &str)]) -> BTreeMap { + pairs + .iter() + .map(|(k, v)| (k.to_string(), v.to_string())) + .collect() +} + +/// The same label set is one series whether it arrives as a map or as +/// Remote Write label pairs in any order; another set is another series. +#[test] +fn one_label_set_is_one_series() { + let map = labels(&[("__name__", "up"), ("job", "api"), ("instance", "1")]); + let a = intern(&map); + let b = intern_pairs(&[("job", "api"), ("__name__", "up"), ("instance", "1")]); + assert!(Arc::ptr_eq(&a, &b)); + assert_eq!(a.id, b.id); + assert_eq!(&*a.metric, "up"); + assert_eq!(**a, map); + let other = intern_pairs(&[("__name__", "up"), ("job", "api"), ("instance", "2")]); + assert_ne!(other.id, a.id); + // A value that moves across a label boundary is another label set. + let shifted = intern_pairs(&[("__name__", "up"), ("job", "ap"), ("instance", "i1")]); + assert_ne!(shifted.id, a.id); +} + +/// Interning from many threads at once gives each label set one id. +#[test] +fn interning_is_consistent_across_threads() { + let threads: Vec<_> = (0..8) + .map(|_| { + std::thread::spawn(|| { + (0..500) + .map(|i| intern_pairs(&[("__name__", "m"), ("i", &i.to_string())]).id) + .collect::>() + }) + }) + .collect(); + let ids: Vec> = threads.into_iter().map(|t| t.join().unwrap()).collect(); + assert!(ids.windows(2).all(|w| w[0] == w[1])); + let mut distinct = ids[0].clone(); + distinct.sort(); + distinct.dedup(); + assert_eq!(distinct.len(), 500); +} From 244e56dbfc3cee7d2e364486ff862e59815a82e1 Mon Sep 17 00:00:00 2001 From: zzylol Date: Wed, 21 Oct 2026 00:27:00 +0000 Subject: [PATCH 2/2] fix(precompute): bound per-series state; drop resent samples; name the batcher Follow-ups from the per-sample ingestion review: - An open pane numbers its own groups, so it holds a summary state only for the series that took a sample in it, not one per series ever seen. - A series unseen for longer than its definition's retention plus one pane is forgotten, with its cached cells in the raw-sample window; a query-time window forgets series unseen for the plan's longest retention past its width. - A sample is looked up once, by its interned series id, and a run of samples into one pane reuses the series' group number there. - A sample not after its series' previous one in the pane (a resent Remote Write sample) is dropped and counted in samples_dropped_duplicate instead of failing the pane or being counted twice. - The output timestamp offset is an error when the build gives none, instead of defaulting to zero. - Pane is SummaryInstance, and SeriesWindow is RawSampleWindowBatcher. --- data_plane/src/lib.rs | 2 +- data_plane/src/per_sample.rs | 212 +++++++++++------- data_plane/src/precompute.rs | 110 +++++++-- data_plane/src/query.rs | 26 ++- .../src/{series.rs => raw_sample_window.rs} | 64 +++++- data_plane/tests/per_sample.rs | 39 +++- data_plane/tests/query_run.rs | 4 +- 7 files changed, 336 insertions(+), 121 deletions(-) rename data_plane/src/{series.rs => raw_sample_window.rs} (81%) diff --git a/data_plane/src/lib.rs b/data_plane/src/lib.rs index 07633be8..c3e9a29c 100644 --- a/data_plane/src/lib.rs +++ b/data_plane/src/lib.rs @@ -11,9 +11,9 @@ mod per_sample; pub mod precompute; pub mod prometheus; pub mod query; +mod raw_sample_window; pub mod remote_write; pub mod sds; -mod series; pub mod series_ids; use std::collections::BTreeMap; diff --git a/data_plane/src/per_sample.rs b/data_plane/src/per_sample.rs index cc8c8505..4721566a 100644 --- a/data_plane/src/per_sample.rs +++ b/data_plane/src/per_sample.rs @@ -2,7 +2,7 @@ //! TimeRange(pane) -> SummaryAgg`, run one sample at a time, in columns. //! //! - **Per series, once:** the stateless prefix. The window's -//! [`SeriesWindow`] applies the `Scan`'s label matchers to the series and +//! [`RawSampleWindowBatcher`] applies the `Scan`'s label matchers to the series and //! caches its text cells; the cells of the `SummaryAgg`'s group columns //! are its group. //! - **Per sample:** the executor's `StateColumnUpdater` for the open pane @@ -38,30 +38,13 @@ use asap_executor::values::{Batch, BatchColumn, SchemaRef}; use asap_executor::StateColumnUpdater; use crate::precompute::{RawSample, SummaryBuild}; -use crate::series::{is_stale, text_column, SeriesCells, SERIES_IDENTITY_COLUMN}; +use crate::raw_sample_window::{ + is_stale, text_column, RawSampleWindowBatcher, SeriesCells, SERIES_IDENTITY_COLUMN, +}; /// A group: its texts in the `SummaryAgg`'s group columns, in their order. type GroupKey = Arc<[Arc]>; -/// Groups numbered `0..`, in the order they are first seen. -#[derive(Default)] -struct GroupNumbers { - numbers: ahash::HashMap, - keys: Vec, -} - -impl GroupNumbers { - fn number(&mut self, key: &GroupKey) -> u32 { - if let Some(&number) = self.numbers.get(key) { - return number; - } - let number = self.keys.len() as u32; - self.keys.push(key.clone()); - self.numbers.insert(key.clone(), number); - number - } -} - /// How a definition is built per sample; compiled once per definition. pub(crate) struct PerSampleSpec { family: FieldDataType, @@ -78,16 +61,45 @@ pub(crate) struct PerSampleSpec { time_offset_ms: OnceLock, } +/// One series as a per-sample build has seen it. +struct SeriesEntry { + /// Its group, or `None` when the `Scan` does not select it. + key: Option, + /// Its newest sample time: a series unseen for longer than the + /// definition's retention is forgotten. + newest_ms: i64, + /// The open pane it last took a sample in, and its group number there, + /// so a run of samples into one pane looks the series up once. + pane: Option<(i64, u32)>, +} + +/// One open pane: a summary state per series that took a sample in it. +struct OpenPane { + updater: StateColumnUpdater, + /// Each group's key and newest sample time, by its number in the pane. + groups: Vec<(GroupKey, i64)>, + /// Group numbers, by series id. + numbers: ahash::HashMap, +} + +/// What a per-sample build did with one sample. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum Taken { + /// Taken into its series' state, or not a sample of any (a stale + /// marker, or a series the `Scan` does not select). + Taken, + /// Not after its series' previous sample in the pane (a resent sample, + /// for instance): dropped, not failing the pane or counted twice. + Duplicate, +} + /// The open panes of a definition built per sample. #[derive(Default)] pub(crate) struct OpenPanes { - /// The numbers of the groups (series) seen. - numbers: GroupNumbers, - /// Each series' group key and number, by series id. - series_groups: ahash::HashMap, - /// By pane end: its states and the groups that took a sample, or why it - /// failed. - panes: BTreeMap), String>>, + /// Each series seen and not forgotten, by series id. + series: ahash::HashMap, + /// By pane end: its states, or why it failed. + panes: BTreeMap>, } impl OpenPanes { @@ -95,6 +107,17 @@ impl OpenPanes { pub(crate) fn drop_before(&mut self, end_ms: i64) { self.panes = self.panes.split_off(&end_ms); } + + /// The series remembered, and each open pane's number of states. + #[cfg(test)] + pub(crate) fn sizes(&self) -> (usize, Vec) { + let states = self + .panes + .values() + .map(|pane| pane.as_ref().map_or(0, |p| p.updater.group_count())) + .collect(); + (self.series.len(), states) + } } impl PerSampleSpec { @@ -200,104 +223,131 @@ impl PerSampleSpec { }) } - /// Updates the pane ending at `end_ms`, open in `open`, with a sample - /// of the series whose cached cells are `cells`. + /// Takes `sample` into the pane ending at `end_ms`, open in `open`: + /// its series' summary state there is updated in place. `window` gives + /// a series' group the first time it is seen. pub(crate) fn update( &self, + window: &RawSampleWindowBatcher, open: &mut OpenPanes, - cells: &Arc, sample: &RawSample, end_ms: i64, - ) { + ) -> Taken { // A stale marker ends a series; it is not a sample of its state. if is_stale(sample.value) { - return; + return Taken::Taken; } - let group = match open.series_groups.get(&sample.labels.id) { - Some(group) => Some(group.clone()), - None => { - let key: GroupKey = self - .groups + let id = sample.labels.id; + let entry = open.series.entry(id).or_insert_with(|| SeriesEntry { + key: window.cells(sample).map(|cells| { + self.groups .iter() .map(|&c| cells.texts[c].clone().unwrap_or_else(|| Arc::from(""))) - .collect(); - let number = open.numbers.number(&key); - open.series_groups - .insert(sample.labels.id, (key.clone(), number)); - Some((key, number)) - } + .collect() + }), + newest_ms: sample.timestamp_ms, + pane: None, + }); + entry.newest_ms = entry.newest_ms.max(sample.timestamp_ms); + let Some(key) = &entry.key else { + return Taken::Taken; }; - let Some((_, number)) = group else { return }; - let entry = open.panes.entry(end_ms).or_insert_with(|| { + let pane = open.panes.entry(end_ms).or_insert_with(|| { StateColumnUpdater::new(&self.family) - .map(|updater| (updater, Vec::new())) + .map(|updater| OpenPane { + updater, + groups: Vec::new(), + numbers: ahash::HashMap::default(), + }) .map_err(|e| e.to_string()) }); - let Ok((updater, touched)) = entry else { - return; + let Ok(pane) = pane else { + return Taken::Taken; }; - let index = number as usize; - let result = updater - .ensure_groups(index + 1) - .and_then(|()| updater.update(number, sample.value, sample.timestamp_ms)); + let number = match entry.pane { + Some((end, number)) if end == end_ms => Ok(number), + _ => match pane.numbers.get(&id) { + Some(&number) => Ok(number), + None => pane.updater.add_group().inspect(|&number| { + pane.numbers.insert(id, number); + pane.groups.push((key.clone(), i64::MIN)); + }), + }, + }; + let result = number.and_then(|number| { + entry.pane = Some((end_ms, number)); + let newest = &mut pane.groups[number as usize].1; + // As Prometheus does, a series' sample not after its previous one + // (a resent sample, for instance) is dropped: a counter's state + // refuses it, and any other state would count it twice. + if sample.timestamp_ms <= *newest { + return Ok(Taken::Duplicate); + } + *newest = sample.timestamp_ms; + pane.updater + .update(number, sample.value, sample.timestamp_ms) + .map(|()| Taken::Taken) + }); match result { - Ok(()) => { - if touched.len() <= index { - touched.resize(index + 1, false); - } - touched[index] = true; + Ok(taken) => taken, + Err(error) => { + open.panes.insert(end_ms, Err(error.to_string())); + Taken::Taken } - Err(error) => *entry = Err(error.to_string()), } } /// The pane ending at `end_ms`, which closes: a row per group (series) - /// that took a sample. + /// that took a sample. Then the series not seen for longer than + /// `keep_ms` before the pane's end are forgotten. pub(crate) fn close( &self, - pane: &SummaryBuild, + build: &SummaryBuild, open: &mut OpenPanes, end_ms: i64, + keep_ms: i64, scope: &Scope, ) -> Result { - let (updater, touched) = match open.panes.remove(&end_ms) { - Some(open_pane) => open_pane?, + let oldest_ms = end_ms.saturating_sub(keep_ms); + open.series.retain(|_, entry| entry.newest_ms > oldest_ms); + build.window.forget_unseen(oldest_ms); + let (updater, groups) = match open.panes.remove(&end_ms) { + Some(pane) => { + let pane = pane?; + (pane.updater, pane.groups) + } None => ( StateColumnUpdater::new(&self.family).map_err(|e| e.to_string())?, Vec::new(), ), }; - let touched: Vec = touched - .iter() - .enumerate() - .filter(|(_, &took)| took) - .map(|(number, _)| number as u32) - .collect(); - self.pane_batch(pane, updater, &touched, &open.numbers.keys, end_ms, scope) + self.pane_batch(build, updater, &groups, end_ms, scope) } - /// The pane's batch: a row per touched group, in group key order as the - /// batch build emits them, with its texts, the pane's timestamp and its - /// state. + /// The pane's batch: a row per group, in group key order as the batch + /// build emits them, with its texts, the pane's timestamp and its state. fn pane_batch( &self, - pane: &SummaryBuild, + build: &SummaryBuild, updater: StateColumnUpdater, - touched: &[u32], - keys: &[GroupKey], + groups: &[(GroupKey, i64)], end_ms: i64, scope: &Scope, ) -> Result { - let mut rows: Vec<(&GroupKey, u32)> = touched + let mut rows: Vec<(&GroupKey, u32)> = groups .iter() - .map(|&number| (&keys[number as usize], number)) + .enumerate() + .map(|(number, (key, _))| (key, number as u32)) .collect(); rows.sort_unstable_by(|a, b| a.0.cmp(b.0)); + // Found once; a pane fails, and the next one tries again, until the + // build gives a timestamp. let offset = match self.time_offset_ms.get() { Some(offset) => *offset, - None => *self - .time_offset_ms - .get_or_init(|| self.calibrate(pane, end_ms, scope).unwrap_or(0)), + None => { + let offset = self.calibrate(build, end_ms, scope)?; + *self.time_offset_ms.get_or_init(|| offset) + } }; let states = Arc::new(updater.seal()); let columns = (0..self.output.fields.len()) diff --git a/data_plane/src/precompute.rs b/data_plane/src/precompute.rs index 4fdf3239..cf6c4779 100644 --- a/data_plane/src/precompute.rs +++ b/data_plane/src/precompute.rs @@ -33,11 +33,11 @@ use futures::StreamExt; use asap_planner::StoredOutputBuild; -use crate::per_sample::{OpenPanes as OpenStates, PerSampleSpec}; +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::series::SeriesWindow; use crate::series_ids::Series; /// One raw sample: its series (interned: its id and full label set, @@ -69,6 +69,10 @@ pub struct PrecomputeStats { /// Samples of a pane already built or skipped, other than the first /// pane after startup. pub samples_dropped_late: u64, + /// Samples, of a stored output built per sample, whose time is not + /// after their series' previous sample in the pane (a resent sample, + /// for instance): dropped, as Prometheus drops them. + pub samples_dropped_duplicate: u64, /// Samples of the first pane after startup, which is never built. pub samples_skipped_first_pane: u64, /// Samples more than `max_future_ms` ahead of the wall clock. @@ -110,11 +114,11 @@ impl Default for PrecomputeOptions { /// build produced (one, when built per sample). Batches share their /// buffers when cloned, so binding a stored pane to a query copies nothing. #[derive(Debug, Clone, Default)] -pub struct Pane { +pub struct SummaryInstance { pub batches: Vec, } -impl Pane { +impl SummaryInstance { pub fn num_rows(&self) -> usize { self.batches.iter().map(Batch::num_rows).sum() } @@ -135,7 +139,7 @@ pub struct SummaryBuild { /// The definition's `RawSampleWindow`: a coverage's raw samples to the /// build's input batch, column by column, with the `Scan`'s label /// matchers. - pub(crate) window: SeriesWindow, + pub(crate) window: RawSampleWindowBatcher, build: CompiledPhysicalDAG, } @@ -158,7 +162,9 @@ impl SummaryBuild { } // `batch` is called with each coverage's end, so the window's // offset from evaluation time is not used. - let window = SeriesWindow::new(dag, &definition.window)?; + // 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 build = compile( dag, BTreeMap::from([( @@ -186,17 +192,21 @@ impl SummaryBuild { end_ms: i64, samples: &[RawSample], scope: Scope, - ) -> Result { + ) -> Result { self.run_batch(self.window.batch(end_ms, samples)?, scope) } /// The build over `input`, a coverage's selected input batch. - pub(crate) fn run_batch(&self, input: Batch, scope: Scope) -> Result { + pub(crate) fn run_batch( + &self, + input: Batch, + scope: Scope, + ) -> Result { let input: Source<'_> = Box::new(Operator::source(input.schema().clone(), vec![input])?); let build = self .build .instantiate(BTreeMap::from([(self.definition.raw_input(), input)]))?; - Ok(Pane { + Ok(SummaryInstance { batches: run(&build, self.build.roots(), scope)?, }) } @@ -221,6 +231,7 @@ struct Builder { #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] enum Outcome { Late, + Duplicate, FirstPane, Buffered, } @@ -240,7 +251,7 @@ pub struct Precompute { /// The newest event-time barrier a producer declared: every pane ending /// at or before it is closed. barrier_ms: Option, - pub store: SummaryStore, + pub store: SummaryStore, pub stats: PrecomputeStats, } @@ -431,13 +442,17 @@ impl Precompute { // as taken, as the plan's build takes and // then filters it. Some(spec) => { - if let Some(cells) = builder.summary.window.cells(sample) { - spec.update(&mut builder.states, &cells, sample, end); + let window = &builder.summary.window; + match spec.update(window, &mut builder.states, sample, end) { + Taken::Taken => Outcome::Buffered, + Taken::Duplicate => Outcome::Duplicate, } } - None => builder.open.entry(end).or_default().push(sample.clone()), + None => { + builder.open.entry(end).or_default().push(sample.clone()); + Outcome::Buffered + } } - Outcome::Buffered } } }; @@ -448,6 +463,7 @@ impl Precompute { Some(Outcome::Buffered) => self.stats.samples_accepted += 1, Some(Outcome::FirstPane) => self.stats.samples_skipped_first_pane += 1, Some(Outcome::Late) => self.stats.samples_dropped_late += 1, + Some(Outcome::Duplicate) => self.stats.samples_dropped_duplicate += 1, } self.newest_sample_ms = self.newest_sample_ms.max(Some(sample.timestamp_ms)); if let Some(newest) = metric.and_then(|m| self.query_run_metrics.get_mut(m)) { @@ -466,6 +482,11 @@ impl Precompute { continue; }; let pane_ms = builder.summary.definition.pane_ms; + let keep_ms = builder + .summary + .definition + .retention_ms + .saturating_add(pane_ms); let id = &builder.summary.definition.id; // The newest evaluation stored panes can serve is at the end of // the newest closed pane. Panes are kept for one extra pane @@ -501,8 +522,8 @@ impl Precompute { let built = match &builder.per_sample { // The pane's states, sealed into one batch. Some(spec) => spec - .close(&builder.summary, &mut builder.states, end, &scope) - .map(|batch| Pane { + .close(&builder.summary, &mut builder.states, end, keep_ms, &scope) + .map(|batch| SummaryInstance { batches: vec![batch], }), None => { @@ -576,3 +597,60 @@ pub fn run( Ok(batches) }) } + +#[cfg(test)] +mod tests { + use super::*; + use crate::series_ids::intern; + + const MINUTE: i64 = 60_000; + + fn sample(job: &str, timestamp_ms: i64) -> RawSample { + RawSample { + labels: intern(&BTreeMap::from([ + ("__name__".to_string(), "latency_ms".to_string()), + ("job".to_string(), job.to_string()), + ])), + timestamp_ms, + value: 1.0, + } + } + + /// Per-sample state follows the active series: an open pane holds a + /// state only for the series that took a sample in it, and a series + /// unseen for longer than the retention plus one pane is forgotten, + /// with its cached cells. Example 4b: 10-minute KLL panes kept 60 min. + #[test] + fn per_series_state_follows_the_active_series() { + let path = std::path::Path::new(env!("CARGO_MANIFEST_DIR")) + .join("../examples/planner-layering-example4b.json"); + let workload = serde_json::from_str(&std::fs::read_to_string(path).unwrap()).unwrap(); + let mut inputs = control_plane::DeploymentInputs::default(); + inputs.deployment_capabilities.raw_data_retained = false; + let plan = control_plane::plan_workload(&workload, &inputs).unwrap(); + let sds = Sds::from_plan(&plan).unwrap(); + let options = PrecomputeOptions { + allowed_lateness_ms: 0, + ..PrecomputeOptions::default() + }; + let mut engine = Precompute::new(plan.dag.dag, &sds, options).unwrap(); + assert_eq!((sds.definitions.len(), engine.builders.len(), engine.per_sample_definitions()), (1, 1, 1)); + let jobs: Vec = (0..10).map(|i| format!("job{i}")).collect(); + let mut samples = Vec::new(); + for t in (MINUTE..25 * MINUTE).step_by(15_000) { + samples.extend(jobs.iter().map(|job| sample(job, t))); + } + engine.ingest_at(&samples, 300 * MINUTE); + let builder = &engine.builders[0]; + assert_eq!(builder.states.sizes(), (10, vec![10]), "(20, 30 min] open"); + let only_a: Vec = (25 * MINUTE..200 * MINUTE) + .step_by(15_000) + .map(|t| sample("job0", t)) + .collect(); + engine.ingest_at(&only_a, 300 * MINUTE); + let builder = &engine.builders[0]; + assert_eq!(builder.states.sizes(), (1, vec![1])); + assert!(builder.summary.window.cached_series() <= 1); + assert_eq!(engine.stats.panes_failed, 0); + } +} diff --git a/data_plane/src/query.rs b/data_plane/src/query.rs index f608743a..bff9b0e2 100644 --- a/data_plane/src/query.rs +++ b/data_plane/src/query.rs @@ -33,9 +33,9 @@ use asap_executor::series_labels_of_output_row; use asap_executor::values::{Batch, SchemaRef, Value}; use asap_planner::{InstallablePlan, StoredOutputBuild}; -use crate::precompute::{run, Pane, RawSample, SummaryBuild}; +use crate::precompute::{run, RawSample, SummaryBuild, SummaryInstance}; +use crate::raw_sample_window::RawSampleWindowBatcher; use crate::sds::{DefinitionId, Sds, Selection, SummaryInstanceKey, SummaryStore}; -use crate::series::SeriesWindow; /// 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]` @@ -117,7 +117,7 @@ struct RawInput { node: u64, matchers: Vec, /// The plan's `Scan` label matchers over the range's samples. - series: SeriesWindow, + series: RawSampleWindowBatcher, range_ms: i64, offset_ms: i64, } @@ -245,7 +245,7 @@ impl QueryEngine { &self, query: &str, time_ms: i64, - store: &SummaryStore, + store: &SummaryStore, progress: IngestionProgress, ) -> Result { let (root, program) = self.program(query)?; @@ -311,7 +311,7 @@ impl QueryEngine { &mut self, root: usize, time_ms: i64, - store: &mut SummaryStore, + store: &mut SummaryStore, progress: &IngestionProgress, raw: &BTreeMap>, ) -> Result, Unanswered> { @@ -442,8 +442,8 @@ fn window(time_ms: i64, offset_ms: i64, width_ms: i64) -> Result<(i64, i64), Una fn pane<'a>( input: &StoredInput, time_ms: i64, - store: &'a SummaryStore, -) -> Result<&'a Pane, Unanswered> { + store: &'a SummaryStore, +) -> Result<&'a SummaryInstance, Unanswered> { let (_, end) = window(time_ms, input.offset_ms, input.pane_ms)?; if end.rem_euclid(input.pane_ms) != 0 { return Err(Unanswered::MissingPane(format!( @@ -477,6 +477,14 @@ fn program( .or_default() .push(edge.producer as u64); } + // A query-time window caches a series for as long as any summary is + // kept. + let longest_retention_ms = sds + .definitions + .values() + .map(|d| d.retention_ms) + .max() + .unwrap_or(0); let (mut stored, mut query_run, mut raw) = (Vec::new(), Vec::new(), Vec::new()); let mut pending = vec![root]; let mut seen = BTreeSet::new(); @@ -513,8 +521,8 @@ fn program( let window = windows.get(&id).ok_or_else(|| { format!("node {id}: a TimeRange the plan lists no window for") })?; - let series = - SeriesWindow::new(dag, window).map_err(|e| format!("node {id}: {e}"))?; + let series = RawSampleWindowBatcher::new(dag, window, longest_retention_ms) + .map_err(|e| format!("node {id}: {e}"))?; raw.push(RawInput { node: id, matchers: label_matchers(dag, window.scan_node_id as u64), diff --git a/data_plane/src/series.rs b/data_plane/src/raw_sample_window.rs similarity index 81% rename from data_plane/src/series.rs rename to data_plane/src/raw_sample_window.rs index 9ffd9375..31a853b8 100644 --- a/data_plane/src/series.rs +++ b/data_plane/src/raw_sample_window.rs @@ -9,6 +9,7 @@ //! (`Batch::try_from_columns`): the columns the selector builds, without //! a `Sample` or a row per sample. +use std::sync::atomic::{AtomicI64, Ordering}; use std::sync::{Arc, RwLock}; use asap_executor::arrow::array::{ @@ -49,13 +50,15 @@ pub(crate) struct SeriesCells { pub(crate) texts: Vec>>, } -/// Each series' cells, or `None` when the `Scan` does not select it, by -/// series id. -type SeriesCache = ahash::HashMap>>; +/// Each series' cells, or `None` when the `Scan` does not select it, with +/// the newest sample time the series was looked up with, by series id. +type SeriesCache = ahash::HashMap>, AtomicI64)>; -/// One raw-sample window of the plan, prepared to turn samples into its -/// input batch. Shared by every thread that reads the window. -pub(crate) struct SeriesWindow { +/// One `RawSampleWindow` of the plan, prepared to turn raw samples into its +/// input batch. A series is forgotten once it has not been seen for +/// `keep_ms`, so the cache holds the series of recent windows, not every +/// series ever seen. +pub(crate) struct RawSampleWindowBatcher { selector: RawSampleSelector, /// The `TimeRange`'s output schema, which the batch is in. pub(crate) schema: SchemaRef, @@ -64,13 +67,19 @@ pub(crate) struct SeriesWindow { /// An instant selector keeps only each series' latest sample. latest_sample_per_series: bool, series: RwLock, + /// How long a series not seen stays cached, past the window's width. + keep_ms: i64, + /// The window end series were last forgotten at. + forgotten_at_ms: AtomicI64, } -impl SeriesWindow { - /// The window `window` of `dag`. +impl RawSampleWindowBatcher { + /// The window `window` of `dag`, which caches a series until it has + /// not been seen for `keep_ms` past the window's width. pub(crate) fn new( dag: &PhysicalASAPDAG, window: &RawSampleWindow, + keep_ms: i64, ) -> Result { let invalid = |message: &str| asap_executor::Error::Invalid(message.into()); let selector = RawSampleSelector::new(dag, window)?; @@ -108,6 +117,8 @@ impl SeriesWindow { width_ms: window.width_ms, latest_sample_per_series: *kind == TimeRangeKind::Instant, series: RwLock::new(ahash::HashMap::default()), + keep_ms, + forgotten_at_ms: AtomicI64::new(i64::MIN), }) } @@ -119,22 +130,54 @@ impl SeriesWindow { /// The cells of `sample`'s series, or `None` when the `Scan` does not /// select it; computed once per series. pub(crate) fn cells(&self, sample: &RawSample) -> Option> { - if let Some(cells) = self + if let Some((cells, newest_ms)) = self .series .read() .unwrap_or_else(|e| e.into_inner()) .get(&sample.labels.id) { + newest_ms.fetch_max(sample.timestamp_ms, Ordering::Relaxed); return cells.clone(); } let cells = self.series_cells(sample); self.series .write() .unwrap_or_else(|e| e.into_inner()) - .insert(sample.labels.id, cells.clone()); + .insert( + sample.labels.id, + (cells.clone(), AtomicI64::new(sample.timestamp_ms)), + ); cells } + /// How many series are cached. + #[cfg(test)] + pub(crate) fn cached_series(&self) -> usize { + self.series.read().unwrap_or_else(|e| e.into_inner()).len() + } + + /// Forgets the series whose newest sample is at or before `oldest_ms`. + pub(crate) fn forget_unseen(&self, oldest_ms: i64) { + self.forgotten_at_ms.fetch_max(oldest_ms, Ordering::Relaxed); + self.series + .write() + .unwrap_or_else(|e| e.into_inner()) + .retain(|_, (_, newest_ms)| newest_ms.load(Ordering::Relaxed) > oldest_ms); + } + + /// Forgets the series not seen for `keep_ms` past the width of the + /// window ending at `end_ms`, at most once per `keep_ms` of window ends, + /// so a scan of the cache is amortized over many windows. + fn forget_old(&self, end_ms: i64) { + let oldest_ms = end_ms + .saturating_sub(self.width_ms) + .saturating_sub(self.keep_ms); + let last = self.forgotten_at_ms.load(Ordering::Relaxed); + if oldest_ms > last.saturating_add(self.keep_ms.max(1)) { + self.forget_unseen(oldest_ms); + } + } + fn series_cells(&self, sample: &RawSample) -> Option> { // The selector applies the Scan's matchers to the series' labels; // the value is not the sample's, which may be a stale marker. @@ -219,6 +262,7 @@ impl SeriesWindow { } } } + self.forget_old(end_ms); self.batch_of(&points) } diff --git a/data_plane/tests/per_sample.rs b/data_plane/tests/per_sample.rs index 7af2e74a..90d8ecf6 100644 --- a/data_plane/tests/per_sample.rs +++ b/data_plane/tests/per_sample.rs @@ -9,7 +9,7 @@ use std::collections::BTreeMap; use asap_executor::planner::ir::schema::{FieldDataType, SketchStatistic}; use asap_executor::summary_kernels::exact::ExactAccumulator; use asap_executor::values::Value; -use data_plane::precompute::{Pane, Precompute, PrecomputeOptions, RawSample}; +use data_plane::precompute::{Precompute, PrecomputeOptions, RawSample, SummaryInstance}; use data_plane::sds::Sds; use data_plane::sds::Selection; @@ -70,7 +70,7 @@ fn state_of(value: &Value) -> Vec { /// Every cell of `pane` as comparable data, its rows sorted. (Tests read /// rows to compare them; the data plane does not.) -fn cells(pane: &Pane) -> Vec>> { +fn cells(pane: &SummaryInstance) -> Vec>> { let mut rows: Vec>> = pane .to_rows() .into_iter() @@ -166,3 +166,38 @@ fn per_sample_panes_apply_label_matchers() { 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 +/// after its series' previous sample: it is dropped and counted, and the +/// pane is built as without it, instead of failing. +#[test] +fn a_resent_sample_is_dropped_not_failing_or_doubling_the_pane() { + let plan = common::plan("example1", false); + let sds = Sds::from_plan(&plan).unwrap(); + let all = samples("http_requests_total", 3, MINUTE, 3 * MINUTE + 1); + let mut resent = all.clone(); + let again = resent[10].clone(); + resent.insert(11, again); + let mut batch = Precompute::batch(&plan.dag.dag, &sds, options()).unwrap(); + batch.ingest(&all); + let mut per_sample = Precompute::new(plan.dag.dag.clone(), &sds, options()).unwrap(); + assert_eq!(per_sample.per_sample_definitions(), sds.definitions.len()); + per_sample.ingest(&resent); + assert_eq!(per_sample.stats.samples_dropped_duplicate, 1); + assert_eq!(per_sample.stats.panes_failed, 0); + assert_eq!(per_sample.stats.panes_built, batch.stats.panes_built); + 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); + assert_eq!( + actual.map(cells), + expected.map(cells), + "pane ending at {end}" + ); + end += definition.pane_ms; + } + } +} diff --git a/data_plane/tests/query_run.rs b/data_plane/tests/query_run.rs index d03d07d1..e8561d59 100644 --- a/data_plane/tests/query_run.rs +++ b/data_plane/tests/query_run.rs @@ -9,8 +9,8 @@ use std::collections::BTreeMap; use asap_planner::{InstallablePlan, StoredOutputBuild}; use data_plane::engine::{EngineHandle, EngineOptions}; -use data_plane::precompute::Pane; use data_plane::precompute::RawSample; +use data_plane::precompute::SummaryInstance; use data_plane::query::{ BoundRawSampleWindow, IngestionProgress, QueryEngine, QueryRunStats, ResultSample, }; @@ -22,7 +22,7 @@ const P99: &str = "quantile_over_time(0.99, latency_ms[5m])"; /// A query engine and the SDS store its stored outputs built in a `QueryRun` go to. struct Fixture { engine: QueryEngine, - store: SummaryStore, + store: SummaryStore, definition: DefinitionId, }