diff --git a/data_plane/src/engine.rs b/data_plane/src/engine.rs index b932c460..cda84ad3 100644 --- a/data_plane/src/engine.rs +++ b/data_plane/src/engine.rs @@ -1,6 +1,7 @@ -//! The engine thread: owns the installed plan and everything that executes -//! it. Planner IR holds `Rc`s, so the plan cannot be shared across the HTTP -//! server's threads; handlers send the engine commands instead. +//! The engine thread: owns the installed plan, compiled once when it is +//! installed, and everything that executes it; handlers send the engine +//! commands. The compiled plan is `Send`, so it is compiled on the +//! installing thread and moved to the engine thread. use std::collections::BTreeMap; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; @@ -96,12 +97,12 @@ impl Drop for RunningGuard { } impl EngineHandle { - /// Starts the engine for the plan serialized in `plan_json`, parsed once - /// on the engine thread, and returns the plan's queries. Fails if the + /// Starts the engine for the plan serialized in `plan_json`, parsed and + /// compiled once, and returns the plan's queries. Fails if the /// plan is invalid or the data plane cannot execute it. pub fn spawn(plan_json: String, options: EngineOptions) -> anyhow::Result<(Self, Vec)> { let (commands, receiver) = channel(); - let (ready, started) = std::sync::mpsc::channel(); + let (engine, queries) = Engine::new(&plan_json, options.precompute)?; let running = Arc::new(AtomicBool::new(true)); let queued_samples = Arc::new(AtomicU64::new(0)); let stats = Arc::new(Mutex::new(PrecomputeStats::default())); @@ -116,17 +117,8 @@ impl EngineHandle { .name("engine".into()) .spawn(move || { let _guard = guard; - match Engine::new(&plan_json, options.precompute) { - Ok((engine, queries)) => { - let _ = ready.send(Ok(queries)); - engine.serve(receiver, shared); - } - Err(error) => { - let _ = ready.send(Err(error)); - } - } + engine.serve(receiver, shared); })?; - let queries = started.recv()??; let handle = Self { commands, queued_samples, diff --git a/data_plane/src/lib.rs b/data_plane/src/lib.rs index ce71122c..869f3604 100644 --- a/data_plane/src/lib.rs +++ b/data_plane/src/lib.rs @@ -7,11 +7,13 @@ //! query it cannot answer, is forwarded to Prometheus. pub mod engine; +mod per_sample; pub mod precompute; pub mod prometheus; pub mod query; pub mod remote_write; pub mod sds; +mod series; use std::collections::BTreeMap; use std::sync::atomic::{AtomicU64, Ordering}; diff --git a/data_plane/src/per_sample.rs b/data_plane/src/per_sample.rs new file mode 100644 index 00000000..1dbccf3d --- /dev/null +++ b/data_plane/src/per_sample.rs @@ -0,0 +1,383 @@ +//! Per-sample ingestion: a definition's build, `Scan -> TimeShift* -> +//! 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 +//! 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 +//! takes the value and time into the group's state in place, the code the +//! batch `SummaryAgg` runs per row. +//! - **At pane close:** the states are sealed into a state column and the +//! pane is built from columns (`Batch::try_from_columns`): the group +//! columns' texts, the pane's timestamp, and the states. +//! +//! A group is one series (the group columns include the series identity), +//! so a counter's samples reach its state in the time order the batch build +//! requires of one series. +//! +//! A build of another shape keeps the batch build over the pane's samples: +//! a row filter, a keyed or shared-grid family, an operator between the +//! `TimeRange` and the `SummaryAgg`, matchers on the sample's value or +//! 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::sync::{Arc, OnceLock}; + +use asap_executor::arrow::array::{Array, TimestampMillisecondArray, UInt32Array}; +use asap_executor::planner::ir::operator::Reduction; +use asap_executor::planner::ir::physical_export::PhysicalASAPDAG; +use asap_executor::planner::ir::scalar::ColumnRef; +use asap_executor::planner::ir::schema::{ + DataType, FieldDataType, GroupingStrategy, SummaryInputExpr, +}; +use asap_executor::planner::ir::{ASAPOp, NonASAPOp, Operator}; +use asap_executor::runtime::Scope; +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}; + +/// 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: 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, + /// The input group columns, in the `SummaryAgg`'s order. + groups: Vec, + output: SchemaRef, + /// Output column of the states, and of the pane's timestamp. + state_column: usize, + time_column: usize, + /// Per output text column: its position in a group's key. + output_group_positions: Vec<(usize, usize)>, + /// How much the output timestamp is after the pane's end, found by + /// running the plan's build once. + time_offset_ms: OnceLock, +} + +/// 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. + series_groups: HashMap>, (GroupKey, u32)>, + /// By pane end: its states and the groups that took a sample, or why it + /// failed. + panes: BTreeMap), String>>, +} + +impl OpenPanes { + /// Drops the open panes ending before `end_ms`, which are skipped. + pub(crate) fn drop_before(&mut self, end_ms: i64) { + self.panes = self.panes.split_off(&end_ms); + } +} + +impl PerSampleSpec { + /// The per-sample build of `pane`'s definition, or `None` when its shape + /// keeps the batch build. + pub(crate) fn new(dag: &PhysicalASAPDAG, pane: &SummaryBuild) -> Option { + let definition = &pane.definition; + let node = |id: u64| dag.nodes.iter().find(|n| n.id as u64 == id); + let builder = node(definition.builder)?; + let inputs: Vec = dag + .edges + .iter() + .filter(|e| e.consumer as u64 == definition.builder) + .map(|e| e.producer as u64) + .collect(); + if inputs != [definition.raw_input()] { + return None; + } + let Operator::ASAP(ASAPOp::SummaryAgg { + family, + input: update, + reduction, + grouping, + filter: None, + .. + }) = &builder.payload + else { + return None; + }; + if update.item.is_some() || *grouping != GroupingStrategy::PerSubpopulationInstance { + return None; + } + let input = &node(definition.raw_input())?.output_schema; + let SummaryInputExpr::Column(column) = &update.weight else { + return None; + }; + let name = match column { + ColumnRef::SampleValue => "value", + ColumnRef::Named(name) => name.as_str(), + _ => return None, + }; + let value = input.fields.iter().position(|f| f.name == name)?; + let time = input.time_index?; + let groups: Vec = match reduction { + Reduction::PerEntity => (0..input.fields.len()) + .filter(|&c| c != value && c != time) + .collect(), + Reduction::Reduce(keys) if !keys.is_without() && !keys.keys().is_empty() => { + keys.keys().to_vec() + } + _ => return None, + }; + if groups + .iter() + .any(|&c| input.fields[c].dtype != FieldDataType::Plain(DataType::Utf8)) + { + return None; + } + // The cached prefix holds only if the matchers read labels alone. + let scan = node(definition.window.scan_node_id as u64)?; + let Operator::NonASAP(NonASAPOp::Scan { predicates, .. }) = &scan.payload else { + return None; + }; + let read = serde_json::to_value(predicates).ok()?; + if columns_read(&read).any(|c| c == value || c == time) { + return None; + } + StateColumnUpdater::new(family).ok()?; + let output = Arc::new(definition.output_schema.clone()); + let state_column = output + .fields + .iter() + .position(|f| !matches!(f.dtype, FieldDataType::Plain(_)))?; + let time_column = output.time_index?; + // Every other output column repeats a group column of its name. + let output_group_positions = output + .fields + .iter() + .enumerate() + .filter(|(i, _)| *i != state_column && *i != time_column) + .map(|(i, field)| { + let position = groups + .iter() + .position(|&c| input.fields[c].name == field.name)?; + (field.dtype == FieldDataType::Plain(DataType::Utf8)).then_some((i, position)) + }) + .collect::>>()?; + // A group is one series. + if !groups + .iter() + .any(|&c| input.fields[c].name == SERIES_IDENTITY_COLUMN) + { + return None; + } + Some(Self { + family: family.clone(), + groups, + output, + state_column, + time_column, + output_group_positions, + time_offset_ms: OnceLock::new(), + }) + } + + /// Updates the pane ending at `end_ms`, open in `open`, with a sample + /// of the series whose cached cells are `cells`. + pub(crate) fn update( + &self, + open: &mut OpenPanes, + cells: &Arc, + sample: &RawSample, + end_ms: i64, + ) { + // A stale marker ends a series; it is not a sample of its state. + if is_stale(sample.value) { + return; + } + let group = match open.series_groups.get(&sample.labels) { + Some(group) => Some(group.clone()), + None => { + let key: GroupKey = 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.clone(), (key.clone(), number)); + Some((key, number)) + } + }; + let Some((_, number)) = group else { return }; + let entry = open.panes.entry(end_ms).or_insert_with(|| { + StateColumnUpdater::new(&self.family) + .map(|updater| (updater, Vec::new())) + .map_err(|e| e.to_string()) + }); + let Ok((updater, touched)) = entry else { + return; + }; + let index = number as usize; + let result = updater + .ensure_groups(index + 1) + .and_then(|()| updater.update(number, sample.value, sample.timestamp_ms)); + match result { + Ok(()) => { + if touched.len() <= index { + touched.resize(index + 1, false); + } + touched[index] = true; + } + Err(error) => *entry = Err(error.to_string()), + } + } + + /// The pane ending at `end_ms`, which closes: a row per group (series) + /// that took a sample. + pub(crate) fn close( + &self, + pane: &SummaryBuild, + open: &mut OpenPanes, + end_ms: i64, + scope: &Scope, + ) -> Result { + let (updater, touched) = match open.panes.remove(&end_ms) { + Some(open_pane) => open_pane?, + 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) + } + + /// 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. + fn pane_batch( + &self, + pane: &SummaryBuild, + updater: StateColumnUpdater, + touched: &[u32], + keys: &[GroupKey], + end_ms: i64, + scope: &Scope, + ) -> Result { + let mut rows: Vec<(&GroupKey, u32)> = touched + .iter() + .map(|&number| (&keys[number as usize], number)) + .collect(); + rows.sort_unstable_by(|a, b| a.0.cmp(b.0)); + 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)), + }; + let states = Arc::new(updater.seal()); + let columns = (0..self.output.fields.len()) + .map(|column| { + if column == self.state_column { + return Ok(BatchColumn::States { + indices: rows + .iter() + .map(|(_, number)| *number) + .collect::(), + states: states.clone(), + }); + } + if column == self.time_column { + return Ok(BatchColumn::Plain(Arc::new( + TimestampMillisecondArray::from_iter_values( + rows.iter().map(|_| end_ms + offset), + ), + ))); + } + let (_, position) = self + .output_group_positions + .iter() + .find(|(output, _)| *output == column) + .expect("every other column repeats a group column"); + text_column(rows.iter().map(|(key, _)| key[*position].as_ref())) + .map(BatchColumn::Plain) + }) + .collect::, asap_executor::Error>>() + .map_err(|e| e.to_string())?; + Batch::try_from_columns(self.output.clone(), columns).map_err(|e| e.to_string()) + } + + /// How much the build's output timestamp is after the pane's end: the + /// plan's build run once over one row. + fn calibrate(&self, pane: &SummaryBuild, end_ms: i64, scope: &Scope) -> Result { + let texts = pane + .window + .schema + .fields + .iter() + .map(|field| field.dtype == FieldDataType::Plain(DataType::Utf8)) + .map(|text| text.then(|| Arc::from(""))) + .collect(); + let cells = Arc::new(SeriesCells { texts }); + let row = pane + .window + .batch_of(&[(cells, end_ms, 1.0)]) + .map_err(|e| e.to_string())?; + let built = pane + .run_batch(row, scope.clone()) + .map_err(|e| e.to_string())?; + let time = built + .batches + .iter() + .find(|batch| batch.num_rows() > 0) + .and_then(|batch| { + batch + .column(self.time_column) + .as_any() + .downcast_ref::() + .filter(|times| !times.is_null(0)) + .map(|times| times.value(0)) + }) + .ok_or("the build gave no timestamp")?; + Ok(time - end_ms) + } +} + +/// The column indices an expression's JSON form reads (`{"Column": i}`). +fn columns_read(expression: &serde_json::Value) -> Box + '_> { + match expression { + serde_json::Value::Object(fields) => Box::new(fields.iter().flat_map(|(name, value)| { + let column = match (name.as_str(), value) { + ("Column", serde_json::Value::Number(n)) => n.as_u64().map(|c| c as usize), + _ => None, + }; + column.into_iter().chain(columns_read(value)) + })), + serde_json::Value::Array(items) => Box::new(items.iter().flat_map(columns_read)), + _ => Box::new(std::iter::empty()), + } +} diff --git a/data_plane/src/precompute.rs b/data_plane/src/precompute.rs index 294e778a..90631bc8 100644 --- a/data_plane/src/precompute.rs +++ b/data_plane/src/precompute.rs @@ -1,16 +1,18 @@ //! Precompute engine: builds each stored summary pane once, when the //! watermark passes its end. //! -//! For every [`SummaryDefinition`] the engine buffers the samples of its -//! `Scan`'s metric by tumbling pane. A pane `(end - pane, end]` closes when -//! the watermark (the newest sample time minus the allowed lateness) reaches -//! `end`. Closing it selects the pane's rows from the buffered samples with -//! the executor's `RawSampleSelector` (the `Scan`'s label matchers), then runs -//! the definition's build from its `TimeRange` input up to its output, under -//! ingestion scope. The output is stored and never changes; samples older -//! than a closed pane are dropped. The first pane a definition sees may have -//! started before the data plane did, so it is never built: queries that -//! need it are not answered from stored panes. +//! A pane `(end - pane, end]` of a [`SummaryDefinition`] closes when the +//! watermark (the newest sample time minus the allowed lateness) reaches +//! `end`. Where the definition's shape allows ([`crate::per_sample`]), each +//! sample of its `Scan`'s metric updates its series' summary state in the +//! open pane in place, and closing the pane seals the states into one batch +//! built from columns. Otherwise the engine buffers the pane's samples, and +//! closing it selects the pane's rows with the `Scan`'s label matchers and +//! runs the definition's build from its `TimeRange` input up to its output, +//! under ingestion scope. The output is stored and never changes; samples +//! older than a closed pane are dropped. The first pane a definition sees +//! may have started before the data plane did, so it is never built: +//! queries that need it are not answered from stored panes. //! //! Only samples of a metric some definition reads move the watermark, and a //! sample further ahead of the wall clock than `max_future_ms` is dropped, so @@ -25,15 +27,17 @@ use asap_executor::physical_planner::{compile, CompiledPhysicalDAG, InputContrac use asap_executor::planner::ir::physical_export::PhysicalASAPDAG; use asap_executor::runtime::{Limits, RunContext, Scope}; use asap_executor::values::{Batch, SchemaRef}; -use asap_executor::{RawSampleSelector, Sample}; +use asap_executor::RawSampleSelector; use futures::executor::block_on; use futures::StreamExt; use asap_planner::StoredOutputBuild; +use crate::per_sample::{OpenPanes as OpenStates, PerSampleSpec}; use crate::sds::{ DefinitionId, Sds, Selection, SummaryDefinition, SummaryInstanceKey, SummaryStore, }; +use crate::series::SeriesWindow; /// One raw sample: its series' full label set (including `__name__`), time /// and value. @@ -101,15 +105,36 @@ impl Default for PrecomputeOptions { } } -/// How one definition's summaries are built from raw samples. The -/// precompute engine builds stored outputs built in an `IngestionRun` with it, and the query -/// engine `QueryRun` ones. +/// A summary instance as stored in the SDS: its rows, in the batches its +/// 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 batches: Vec, +} + +impl Pane { + pub fn num_rows(&self) -> usize { + self.batches.iter().map(Batch::num_rows).sum() + } + + /// Every row, for tests; the data plane reads rows only to encode an + /// answer. + pub fn to_rows(&self) -> Vec> { + self.batches.iter().flat_map(Batch::to_rows).collect() + } +} + +/// How one definition's summaries are built from raw samples with the +/// plan's own build. The precompute engine builds stored outputs built in +/// an `IngestionRun` with it where it does not build them per sample, and +/// the query engine `QueryRun` ones. pub struct SummaryBuild { pub definition: SummaryDefinition, - /// Turns a coverage's samples into the build's input: the `Scan`'s label - /// matchers, then the coverage's `TimeRange` rows. - selector: RawSampleSelector, - raw_schema: SchemaRef, + /// 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, build: CompiledPhysicalDAG, } @@ -130,29 +155,27 @@ impl SummaryBuild { if definition.pane_ms <= 0 { return Err(error(format!("pane width {} ms", definition.pane_ms))); } - let raw_schema = Arc::new(output_schema(dag, definition.raw_input()).clone()); - // `window_batch` is called with each coverage's end, so the window's + // `batch` is called with each coverage's end, so the window's // offset from evaluation time is not used. - let selector = RawSampleSelector::new(dag, &definition.window)?; + let window = SeriesWindow::new(dag, &definition.window)?; let build = compile( dag, BTreeMap::from([( definition.raw_input(), - InputContract::bounded(raw_schema.clone()), + InputContract::bounded(window.schema.clone()), )]), &[definition.builder], )?; Ok(Self { definition: definition.clone(), - selector, - raw_schema, + window, build, }) } /// The metric whose samples the build reads. pub fn metric(&self) -> &str { - self.selector.metric() + self.window.metric() } /// Selects the coverage ending at `end_ms` from `samples` with the plan's @@ -162,25 +185,30 @@ impl SummaryBuild { end_ms: i64, samples: &[RawSample], scope: Scope, - ) -> Result { - let selected = self - .selector - .window_batch(end_ms, &executor_samples(samples))?; - let input: Source<'_> = - Box::new(Operator::source(self.raw_schema.clone(), vec![selected])?); + ) -> 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 { + let input: Source<'_> = Box::new(Operator::source(input.schema().clone(), vec![input])?); let build = self .build .instantiate(BTreeMap::from([(self.definition.raw_input(), input)]))?; - let rows = run_root(&build, self.definition.builder, scope)?; - let schema = Arc::new(self.definition.output_schema.clone()); - Ok(Batch::try_new(schema, rows)?) + Ok(Pane { + batches: run(&build, self.build.roots(), scope)?, + }) } } struct Builder { summary: SummaryBuild, - /// Samples of open panes, by pane end. + /// The build run per sample, unless its shape keeps the plan's build. + per_sample: Option, + /// For the plan's build: samples of open panes, by pane end. open: BTreeMap>, + /// For the per-sample build: the open panes' summary states. + states: OpenStates, /// End of the first pane seen, which is skipped. `None` until then. first_pane_end: Option, /// End of the newest pane built or skipped; samples in it or before it @@ -211,18 +239,48 @@ 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, } impl Precompute { /// Compiles one build per definition of `sds` built in an ingestion run, - /// in dependency order; `sds` describes `dag`. + /// in dependency order; `sds` describes `dag`. A definition is built per + /// sample where its shape allows ([`crate::per_sample`]), else with the + /// plan's build over each pane's samples. pub fn new( dag: PhysicalASAPDAG, sds: &Sds, options: PrecomputeOptions, ) -> Result { + Self::with(&dag, sds, options, true) + } + + /// [`Self::new`] building every pane with the plan's own build over its + /// samples: the reference the per-sample builds equal. + pub fn batch( + dag: &PhysicalASAPDAG, + sds: &Sds, + options: PrecomputeOptions, + ) -> Result { + Self::with(dag, sds, options, false) + } + + /// How many definitions are built per sample. + pub fn per_sample_definitions(&self) -> usize { + self.builders + .iter() + .filter(|b| b.per_sample.is_some()) + .count() + } + + fn with( + dag: &PhysicalASAPDAG, + sds: &Sds, + options: PrecomputeOptions, + per_sample: bool, + ) -> Result { + let dag = dag.clone(); let mut builders = Vec::new(); // `IngestionRun` definitions in dependency order; `QueryRun` ones are // the query engine's. @@ -230,9 +288,16 @@ impl Precompute { .ingestion_build_order() .map_err(|e| PrecomputeError::Plan(e.to_string()))?; for id in &order { + let summary = SummaryBuild::new(&dag, &sds.definitions[id])?; + let per_sample = per_sample + .then(|| PerSampleSpec::new(&dag, &summary)) + .flatten(); + tracing::info!(definition = ?id, per_sample = per_sample.is_some(), "pane build"); builders.push(Builder { - summary: SummaryBuild::new(&dag, &sds.definitions[id])?, + summary, + per_sample, open: BTreeMap::new(), + states: OpenStates::default(), first_pane_end: None, built_through: None, }); @@ -360,7 +425,17 @@ impl Precompute { } else if end <= built { Outcome::Late } else { - builder.open.entry(end).or_default().push(sample.clone()); + match &builder.per_sample { + // A series the Scan does not select counts + // 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); + } + } + None => builder.open.entry(end).or_default().push(sample.clone()), + } Outcome::Buffered } } @@ -411,18 +486,33 @@ impl Precompute { self.stats.panes_skipped_expired += (first_kept.saturating_sub(end) / pane_ms) as u64; builder.open = builder.open.split_off(&first_kept); + builder.states.drop_before(first_kept); end = first_kept; } // Every pane after the first through the watermark is built, so // a pane without samples is stored as empty, not missing. while end <= watermark { - let samples = builder.open.remove(&end).unwrap_or_default(); let scope = Scope::Ingestion { window_start_ms: end - pane_ms, window_end_ms: end, revision: 1, }; - match builder.summary.run(end, &samples, scope) { + 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 { + batches: vec![batch], + }), + None => { + let samples = builder.open.remove(&end).unwrap_or_default(); + builder + .summary + .run(end, &samples, scope) + .map_err(|e| e.to_string()) + } + }; + match built { Ok(state) => { self.store.insert( SummaryInstanceKey::new(id.clone(), Selection::ending_at(end)), @@ -466,40 +556,22 @@ pub fn pane_end(timestamp_ms: i64, pane_ms: i64) -> Option { .checked_mul(pane_ms) } -/// `samples` as the executor's [`Sample`]s. The selector takes owned label -/// maps, so each sample's labels are copied. -pub fn executor_samples(samples: &[RawSample]) -> Vec { - samples - .iter() - .map(|s| Sample { - labels: (*s.labels).clone(), - timestamp_ms: s.timestamp_ms, - value: s.value, - }) - .collect() -} - -/// Executes `root` of `dag` to completion and returns its rows. -pub fn run_root( +/// Executes `dag`'s single root to completion and returns its batches. +pub fn run( dag: &asap_executor::plan::PhysicalDAG<'_, Batch, SchemaRef>, - root: u64, + roots: &[u64], scope: Scope, -) -> Result>, asap_executor::Error> { +) -> Result, asap_executor::Error> { block_on(async { let context = RunContext::new(scope, Limits::default())?; - let mut output = dag.execute(&[root], context)?.remove(0); - let mut rows = Vec::new(); + let mut output = dag.execute(roots, context)?.remove(0); + let mut batches = Vec::new(); while let Some(batch) = output.next().await { - rows.extend(batch?.to_rows()); + let batch = batch?; + if batch.num_rows() > 0 { + batches.push((*batch).clone()); + } } - Ok(rows) + Ok(batches) }) } - -fn output_schema(dag: &PhysicalASAPDAG, id: u64) -> &asap_executor::planner::ir::schema::Schema { - &dag.nodes - .iter() - .find(|n| n.id as u64 == id) - .expect("SDS names DAG nodes") - .output_schema -} diff --git a/data_plane/src/query.rs b/data_plane/src/query.rs index 3416a5d7..f608743a 100644 --- a/data_plane/src/query.rs +++ b/data_plane/src/query.rs @@ -29,12 +29,13 @@ use asap_executor::planner::ir::scalar::{CompareOpKind, ScalarValue}; use asap_executor::planner::ir::schema::{DataType, FieldDataType, Schema}; use asap_executor::planner::ir::{NonASAPOp, Operator as IrOperator, ScalarExpr}; use asap_executor::runtime::Scope; +use asap_executor::series_labels_of_output_row; use asap_executor::values::{Batch, SchemaRef, Value}; -use asap_executor::{series_labels_of_output_row, RawSampleSelector}; use asap_planner::{InstallablePlan, StoredOutputBuild}; -use crate::precompute::{executor_samples, run_root, RawSample, SummaryBuild}; +use crate::precompute::{run, Pane, RawSample, SummaryBuild}; 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]` @@ -105,6 +106,8 @@ pub enum Unanswered { struct StoredInput { node: u64, + /// The output's schema, which its panes' rows are in. + schema: SchemaRef, definition: DefinitionId, offset_ms: i64, pane_ms: i64, @@ -114,8 +117,7 @@ struct RawInput { node: u64, matchers: Vec, /// The plan's `Scan` label matchers over the range's samples. - selector: RawSampleSelector, - schema: SchemaRef, + series: SeriesWindow, range_ms: i64, offset_ms: i64, } @@ -123,6 +125,7 @@ struct RawInput { /// A stored output built in a `QueryRun` (`MatAtQueryKeptAcrossEvaluations`). struct QueryRunInput { node: u64, + schema: SchemaRef, definition: DefinitionId, offset_ms: i64, /// The output's own `TimeRange`, which its coverage's samples feed. @@ -242,7 +245,7 @@ impl QueryEngine { &self, query: &str, time_ms: i64, - store: &SummaryStore, + store: &SummaryStore, progress: IngestionProgress, ) -> Result { let (root, program) = self.program(query)?; @@ -256,7 +259,7 @@ impl QueryEngine { let (start_ms, end_ms) = window(time_ms, input.offset_ms, input.range_ms)?; Ok(BoundRawSampleWindow { node: input.node, - metric: input.selector.metric().to_string(), + metric: input.series.metric().to_string(), matchers: input.matchers.clone(), start_ms, end_ms, @@ -308,7 +311,7 @@ impl QueryEngine { &mut self, root: usize, time_ms: i64, - store: &mut SummaryStore, + store: &mut SummaryStore, progress: &IngestionProgress, raw: &BTreeMap>, ) -> Result, Unanswered> { @@ -325,7 +328,7 @@ impl QueryEngine { for input in &program.stored { let pane = pane(input, time_ms, store)?; let source = - Operator::source(pane.schema().clone(), vec![pane.clone()]).map_err(failed)?; + Operator::source(input.schema.clone(), pane.batches.clone()).map_err(failed)?; sources.insert(input.node, Box::new(source)); } for input in &program.query_run { @@ -376,8 +379,7 @@ impl QueryEngine { summary } }; - let source = - Operator::source(summary.schema().clone(), vec![summary]).map_err(failed)?; + let source = Operator::source(input.schema.clone(), summary.batches).map_err(failed)?; sources.insert(input.node, Box::new(source)); } if let Some(watermark) = progress.watermark_ms { @@ -394,19 +396,15 @@ impl QueryEngine { } for input in &program.raw { let samples = raw.get(&input.node).map(Vec::as_slice).unwrap_or_default(); - let rows = input - .selector - .window_batch( - window(time_ms, input.offset_ms, input.range_ms)?.1, - &executor_samples(samples), - ) - .map_err(failed)?; - let source = Operator::source(input.schema.clone(), vec![rows]).map_err(failed)?; + let (_, end_ms) = window(time_ms, input.offset_ms, input.range_ms)?; + let rows = input.series.batch(end_ms, samples).map_err(failed)?; + let source = + Operator::source(input.series.schema.clone(), vec![rows]).map_err(failed)?; sources.insert(input.node, Box::new(source)); } let dag = program.compiled.instantiate(sources).map_err(failed)?; - let rows = run_root(&dag, program.compiled.roots()[0], scope).map_err(failed)?; - result_samples(&program.output, rows) + let batches = run(&dag, program.compiled.roots(), scope).map_err(failed)?; + result_samples(&program.output, &batches) } /// The build of a `QueryRun` definition a compiled root reads. @@ -444,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 Batch, Unanswered> { + store: &'a SummaryStore, +) -> Result<&'a Pane, 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!( @@ -490,6 +488,7 @@ fn program( match sds.definitions[definition].built_at { StoredOutputBuild::IngestionRun => stored.push(StoredInput { node: id, + schema: Arc::new(nodes[&id].output_schema.clone()), definition: definition.clone(), offset_ms: binding.offset_ms, pane_ms: sds.definitions[definition].pane_ms, @@ -500,6 +499,7 @@ fn program( .map_err(Clone::clone)?; query_run.push(QueryRunInput { node: id, + schema: Arc::new(nodes[&id].output_schema.clone()), definition: definition.clone(), offset_ms: binding.offset_ms, raw_input: binding.raw_input, @@ -513,13 +513,12 @@ fn program( let window = windows.get(&id).ok_or_else(|| { format!("node {id}: a TimeRange the plan lists no window for") })?; - let selector = - RawSampleSelector::new(dag, window).map_err(|e| format!("node {id}: {e}"))?; + let series = + SeriesWindow::new(dag, window).map_err(|e| format!("node {id}: {e}"))?; raw.push(RawInput { node: id, matchers: label_matchers(dag, window.scan_node_id as u64), - selector, - schema: Arc::new(nodes[&id].output_schema.clone()), + series, range_ms: window.width_ms, offset_ms: window.end_offset_ms, }); @@ -541,7 +540,10 @@ fn program( inputs.insert(node, InputContract::bounded(schema)); } for input in &raw { - inputs.insert(input.node, InputContract::bounded(input.schema.clone())); + inputs.insert( + input.node, + InputContract::bounded(input.series.schema.clone()), + ); } let compiled = compile(dag, inputs, &[root]).map_err(|e| e.to_string())?; let mut query_run_fetch_node: BTreeMap = BTreeMap::new(); @@ -610,9 +612,11 @@ fn label_matchers(dag: &PhysicalASAPDAG, scan: u64) -> Vec { .collect() } -/// PromQL vector samples from result rows: labels from the series identity -/// and the other text columns, the value from the one `Float64` column. -fn result_samples(schema: &Schema, rows: Vec>) -> Result, Unanswered> { +/// PromQL vector samples from a root's result batches: labels from the +/// series identity and the other text columns, the value from the one +/// `Float64` column. The one place rows are read: the answer's JSON +/// encoding. +fn result_samples(schema: &Schema, batches: &[Batch]) -> Result, Unanswered> { let unsupported = |reason: &str| Unanswered::Unsupported(format!("result rows: {reason}")); let values: Vec = schema .fields @@ -624,7 +628,9 @@ fn result_samples(schema: &Schema, rows: Vec>) -> Result TimeShift* -> TimeRange`) +//! reads samples of one metric. Each series is looked at once: the +//! executor's `RawSampleSelector` applies the `Scan`'s label matchers to +//! it, and its text cells (its series identity and label columns) are +//! computed and cached. A window's batch is then built from the samples' +//! times and values and their series' cached cells, as Arrow columns +//! (`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::{ + ArrayRef, DictionaryArray, Float64Array, StringArray, TimestampMillisecondArray, UInt32Array, +}; +use asap_executor::planner::ir::physical_export::{PhysicalASAPDAG, RawSampleWindow}; +use asap_executor::planner::ir::schema::{DataType, FieldDataType}; +use asap_executor::planner::ir::{NonASAPOp, Operator, TimeRangeKind}; +use asap_executor::values::{Batch, BatchColumn, SchemaRef}; +use asap_executor::{series_identity, RawSampleSelector, Sample}; + +use crate::precompute::RawSample; + +/// The `Scan` schema's column that holds a series' full identity. +pub(crate) const SERIES_IDENTITY_COLUMN: &str = "$promql_series_identity"; + +/// The Prometheus stale marker: a NaN that ends a series, not a sample. +pub(crate) const STALE_MARKER: u64 = 0x7ff0_0000_0000_0002; + +/// Whether `value` is the Prometheus stale marker. +pub(crate) fn is_stale(value: f64) -> bool { + value.to_bits() == STALE_MARKER +} + +/// What one column of a window's input holds. +#[derive(Debug, Clone, PartialEq)] +enum ColumnKind { + Time, + Value, + /// The series identity, or a label (empty when the series lacks it). + Text, +} + +/// One selected series' text cells, by input column (`None` for the time +/// and value columns). +#[derive(Debug, PartialEq)] +pub(crate) struct SeriesCells { + pub(crate) texts: Vec>>, +} + +/// Each series' cells, or `None` when the `Scan` does not select it. +type SeriesCache = HashMap>, Option>>; + +/// 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 { + selector: RawSampleSelector, + /// The `TimeRange`'s output schema, which the batch is in. + pub(crate) schema: SchemaRef, + kinds: Vec, + width_ms: i64, + /// An instant selector keeps only each series' latest sample. + latest_sample_per_series: bool, + series: RwLock, +} + +impl SeriesWindow { + /// The window `window` of `dag`. + pub(crate) fn new( + dag: &PhysicalASAPDAG, + window: &RawSampleWindow, + ) -> Result { + let invalid = |message: &str| asap_executor::Error::Invalid(message.into()); + let selector = RawSampleSelector::new(dag, window)?; + let time_range = dag + .nodes + .iter() + .find(|n| n.id == window.time_range_node_id) + .ok_or_else(|| invalid("window names an unknown TimeRange"))?; + let Operator::NonASAP(NonASAPOp::TimeRange { kind, .. }) = &time_range.payload else { + return Err(invalid("window's time range is not a TimeRange")); + }; + let schema = Arc::new(time_range.output_schema.clone()); + let kinds = schema + .fields + .iter() + .enumerate() + .map(|(index, field)| { + if Some(index) == schema.time_index { + Ok(ColumnKind::Time) + } else if field.name == "value" + && field.dtype == FieldDataType::Plain(DataType::Float64) + { + Ok(ColumnKind::Value) + } else if field.dtype == FieldDataType::Plain(DataType::Utf8) { + Ok(ColumnKind::Text) + } else { + Err(invalid("unsupported PromQL source column")) + } + }) + .collect::, _>>()?; + Ok(Self { + selector, + schema, + kinds, + width_ms: window.width_ms, + latest_sample_per_series: *kind == TimeRangeKind::Instant, + series: RwLock::new(HashMap::new()), + }) + } + + /// The metric the window reads. + pub(crate) fn metric(&self) -> &str { + self.selector.metric() + } + + /// 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 + .series + .read() + .unwrap_or_else(|e| e.into_inner()) + .get(&sample.labels) + { + return cells.clone(); + } + let cells = self.series_cells(sample); + self.series + .write() + .unwrap_or_else(|e| e.into_inner()) + .insert(sample.labels.clone(), cells.clone()); + cells + } + + 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. + let one = Sample { + labels: (*sample.labels).clone(), + timestamp_ms: sample.timestamp_ms, + value: 0.0, + }; + let selected = self + .selector + .window_batch(sample.timestamp_ms, &[one]) + .ok()?; + if selected.num_rows() != 1 { + return None; + } + let identity: Arc = series_identity(&sample.labels).ok()?.into(); + let texts = self + .kinds + .iter() + .zip(&self.schema.fields) + .map(|(kind, field)| match kind { + ColumnKind::Text if field.name == SERIES_IDENTITY_COLUMN => Some(identity.clone()), + ColumnKind::Text => Some( + sample + .labels + .get(&field.name) + .map_or_else(|| Arc::from(""), |v| Arc::from(v.as_str())), + ), + _ => None, + }) + .collect(); + Some(Arc::new(SeriesCells { texts })) + } + + /// The window's input batch for the window ending at `end_ms`, as the + /// executor's `RawSampleSelector::window_batch` gives it: the samples + /// in `(end_ms - width, end_ms]` of the series the `Scan` selects + /// (only each series' latest for an instant selector), in the + /// `TimeRange`'s schema. A stale marker is never a row: a range window + /// drops it, and an instant selector whose latest sample is one has no + /// row for that series. + pub(crate) fn batch( + &self, + end_ms: i64, + samples: &[RawSample], + ) -> Result { + let start_ms = end_ms - self.width_ms; + let in_window = samples + .iter() + .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(); + for sample in in_window { + let entry = latest.entry(&sample.labels).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)) { + 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; + 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() + } + _ => { + let cells = self.cells(sample); + current = Some((&sample.labels, cells.clone())); + cells + } + }; + if let Some(cells) = cells { + points.push((cells, sample.timestamp_ms, sample.value)); + } + } + } + self.batch_of(&points) + } + + /// The batch of `points`: each sample's series cells, time and value. + pub(crate) fn batch_of( + &self, + points: &[(Arc, i64, f64)], + ) -> Result { + let columns = self + .kinds + .iter() + .enumerate() + .map(|(index, kind)| { + let array: ArrayRef = match kind { + ColumnKind::Time => Arc::new(TimestampMillisecondArray::from_iter_values( + points.iter().map(|(_, time, _)| *time), + )), + ColumnKind::Value => Arc::new(Float64Array::from_iter_values( + points.iter().map(|(_, _, value)| *value), + )), + ColumnKind::Text => text_column( + points + .iter() + .map(|(cells, _, _)| cells.texts[index].as_deref().unwrap_or("")), + )?, + }; + Ok(BatchColumn::Plain(array)) + }) + .collect::, asap_executor::Error>>()?; + Batch::try_from_columns(self.schema.clone(), columns) + } +} + +/// A text column in the batch layout: a dictionary of the distinct texts, +/// with each row's code. +pub(crate) fn text_column<'a>( + texts: impl Iterator, +) -> Result { + let mut codes: HashMap<&str, u32> = HashMap::new(); + let mut values: Vec<&str> = Vec::new(); + let keys: UInt32Array = texts + .map(|text| { + let next = values.len() as u32; + *codes.entry(text).or_insert_with(|| { + values.push(text); + next + }) + }) + .collect(); + let dictionary = DictionaryArray::try_new(keys, Arc::new(StringArray::from(values))) + .map_err(|e| asap_executor::Error::Invalid(e.to_string()))?; + Ok(Arc::new(dictionary)) +} diff --git a/data_plane/tests/per_sample.rs b/data_plane/tests/per_sample.rs new file mode 100644 index 00000000..af42e185 --- /dev/null +++ b/data_plane/tests/per_sample.rs @@ -0,0 +1,169 @@ +//! Ingestion updates each open pane's summary state per sample, through the +//! executor's `StateColumnUpdater`, and builds the pane from columns; the +//! panes it stores equal the plan's own build over the same samples. + +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; +use asap_executor::values::Value; +use data_plane::precompute::{Pane, Precompute, PrecomputeOptions, RawSample}; +use data_plane::sds::Sds; +use data_plane::sds::Selection; + +const MINUTE: i64 = 60_000; + +fn options() -> PrecomputeOptions { + PrecomputeOptions { + allowed_lateness_ms: 0, + ..PrecomputeOptions::default() + } +} + +/// Samples every 15 s in `[from, to)` of `metric` for each `job`; values +/// grow with time and differ by job. +fn samples(metric: &str, jobs: usize, from: i64, to: i64) -> Vec { + let series: Vec<_> = (0..jobs) + .map(|j| { + Arc::new(BTreeMap::from([ + ("__name__".to_string(), metric.to_string()), + ("job".to_string(), format!("job{j}")), + ("instance".to_string(), format!("{}", j % 3)), + ])) + }) + .collect(); + (from..to) + .step_by(15_000) + .flat_map(|t| { + series.iter().enumerate().map(move |(j, labels)| RawSample { + labels: labels.clone(), + timestamp_ms: t, + value: (j * 1000) as f64 + (t / 15_000 % 97) as f64 + t as f64 / 1e6, + }) + }) + .collect() +} + +/// The state in `value`, as comparable data: an exact accumulator's +/// serialized bytes, or a quantile sketch's quantiles at every percent. +/// KLL seeds its compaction randomly per sketch, so two builds of the same +/// samples agree on content, not on bytes; below k they hold every sample. +fn state_of(value: &Value) -> Vec { + let Value::Summary { family, state } = value else { + panic!("not a state: {value:?}"); + }; + if let Some(exact) = state.as_any().downcast_ref::() { + return serde_json::to_vec(exact).unwrap(); + } + assert!(matches!(family, FieldDataType::Sketch(..)), "{family:?}"); + (1..100) + .flat_map(|p| { + let quantile = SketchStatistic::Quantile { + q: p as f64 / 100.0, + }; + state.estimate(&quantile).unwrap().to_le_bytes() + }) + .collect() +} + +/// 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>> { + let mut rows: Vec>> = pane + .to_rows() + .into_iter() + .map(|row| { + row.iter() + .map(|value| match value { + Value::Summary { .. } => state_of(value), + other => other.key().unwrap(), + }) + .collect() + }) + .collect(); + rows.sort(); + rows +} + +/// For Example 1's counter rates and sums over 10 s panes, Example 3b's +/// 1-minute KLL panes and Example 4b's 10-minute KLL panes, on a deployment +/// without raw data: every pane built per sample equals the executor's batch build of the same samples, row by row. +#[test] +fn per_sample_panes_equal_the_batch_build() { + for (example, metric, to) in [ + ("example1", "http_requests_total", 12 * MINUTE), + ("example3b", "latency_ms", 12 * MINUTE), + ("example4b", "latency_ms", 75 * MINUTE), + ] { + let plan = common::plan(example, false); + let sds = Sds::from_plan(&plan).unwrap(); + let all = samples(metric, 7, MINUTE, to + 1); + let mut batch = Precompute::batch(&plan.dag.dag, &sds, options()).unwrap(); + batch.ingest(&all); + assert_eq!(batch.per_sample_definitions(), 0); + { + let mut per_sample = Precompute::new(plan.dag.dag.clone(), &sds, options()).unwrap(); + assert_eq!( + per_sample.per_sample_definitions(), + sds.definitions.len(), + "{example}: every definition is built per sample" + ); + per_sample.ingest(&all); + assert_eq!(per_sample.stats, batch.stats, "{example}"); + let mut compared = 0; + for definition in sds.definitions.values() { + let mut end = 0; + while end <= to { + let selection = Selection::ending_at(end); + let expected = batch.store.get(&definition.id, selection); + let actual = per_sample.store.get(&definition.id, selection); + assert_eq!(actual.is_some(), expected.is_some(), "{example} pane {end}"); + if let (Some(actual), Some(expected)) = (actual, expected) { + for batch in actual.batches.iter().chain(&expected.batches) { + assert_eq!(**batch.schema(), definition.output_schema); + } + assert_eq!(cells(&actual), cells(&expected), "{example} pane {end}"); + compared += usize::from(expected.num_rows() > 0); + } + end += definition.pane_ms; + } + } + assert!(compared > 0, "{example}: no pane compared"); + } + } +} + +/// The plan's label matchers apply per sample as in the batch build: only +/// the selected series get a state. +#[test] +fn per_sample_panes_apply_label_matchers() { + let mut workload = common::workload("example4b"); + workload.query_workload.repeating_queries.as_mut().unwrap()[0].query = + asap_executor::planner::workload::Query( + "quantile_over_time(0.99, latency_ms{instance=\"1\"}[60m])".into(), + ); + let plan = control_plane::plan_workload(&workload, &common::inputs(false)).unwrap(); + let sds = Sds::from_plan(&plan).unwrap(); + let id = sds.definitions.keys().next().unwrap().clone(); + let all = samples("latency_ms", 7, MINUTE, 25 * MINUTE); + let mut per_sample = Precompute::new(plan.dag.dag.clone(), &sds, options()).unwrap(); + assert_eq!(per_sample.per_sample_definitions(), 1); + per_sample.ingest(&all); + let mut batch = Precompute::batch(&plan.dag.dag, &sds, options()).unwrap(); + batch.ingest(&all); + let (actual, expected) = ( + per_sample + .store + .get(&id, Selection::ending_at(20 * MINUTE)) + .unwrap(), + batch + .store + .get(&id, Selection::ending_at(20 * MINUTE)) + .unwrap(), + ); + assert_eq!(cells(&expected).len(), 2, "jobs 1 and 4 are instance 1"); + assert_eq!(cells(&actual), cells(&expected)); +} diff --git a/data_plane/tests/query_run.rs b/data_plane/tests/query_run.rs index 85ececb0..d8547d03 100644 --- a/data_plane/tests/query_run.rs +++ b/data_plane/tests/query_run.rs @@ -8,9 +8,9 @@ mod common; use std::collections::BTreeMap; use std::sync::Arc; -use asap_executor::values::Batch; use asap_planner::{InstallablePlan, StoredOutputBuild}; use data_plane::engine::{EngineHandle, EngineOptions}; +use data_plane::precompute::Pane; use data_plane::precompute::RawSample; use data_plane::query::{ BoundRawSampleWindow, IngestionProgress, QueryEngine, QueryRunStats, ResultSample, @@ -23,7 +23,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, }