From 19c25704ac4f0d0927a2737dff394c1f71b03fd4 Mon Sep 17 00:00:00 2001 From: zzylol Date: Mon, 19 Oct 2026 21:14:17 +0000 Subject: [PATCH] feat(data-plane): columnar engine on the Planner's Arrow operators; compile once; per-sample panes from columns Re-pin ASAPPlanner to main 92bc1296: the executor runs every operator on Arrow batches, compiled plans and RawSampleSelector are Send + Sync, and StateColumnUpdater builds a state column sample by sample. - Compile once: the plan's ingestion side (Ingestion) and its query roots (QueryEngine) are compiled when the plan is installed and shared by every ingestion worker and query thread through Arc. - Raw samples to columns: a raw-sample window's input batch is built from the samples' times and values and each series' cached cells (series identity and label columns, dictionary-encoded), with the Scan's matchers applied once per series through RawSampleSelector. Query-time raw windows, kept panes and the batch-build fallback all use it. - Ingestion per sample, no rows: a definition whose group is a series runs StateColumnUpdater::update in place per sample; at pane close each worker seals its states and builds its part with Batch::try_from_columns. Workers own disjoint series, so a pane is the workers' batches, shared by Arc in the SDS. Another shape keeps the plan's build, now run once over every worker's samples of the pane. - Stored panes bind through Operator::source, which treats them as held outside the run. - Stored panes are kept one pane width longer than the plan's retention: an evaluation at T runs once samples just after T have arrived, so the newest stored pane can be past T. - Rows are read only to encode answers as Prometheus JSON. --- Cargo.lock | 38 +-- Cargo.toml | 4 +- README.md | 11 +- data_plane/src/engine.rs | 53 +--- data_plane/src/lib.rs | 1 + data_plane/src/per_sample.rs | 411 +++++++++++++++++----------- data_plane/src/precompute.rs | 475 +++++++++++++++++---------------- data_plane/src/query.rs | 127 ++++----- data_plane/src/series.rs | 255 ++++++++++++++++++ data_plane/tests/answers.rs | 2 +- data_plane/tests/per_sample.rs | 25 +- data_plane/tests/precompute.rs | 93 ++++++- data_plane/tests/range.rs | 13 +- 13 files changed, 949 insertions(+), 559 deletions(-) create mode 100644 data_plane/src/series.rs diff --git a/Cargo.lock b/Cargo.lock index c55c0c8b..e462fd74 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -360,8 +360,10 @@ dependencies = [ [[package]] name = "asap-executor" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=f0dc1825805f4cbb8866b3baa15bed4319c4e1ab#f0dc1825805f4cbb8866b3baa15bed4319c4e1ab" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=92bc1296fd33802e8ce3fcab6b75f8e3205c3960#92bc1296fd33802e8ce3fcab6b75f8e3205c3960" dependencies = [ + "ahash", + "arrow", "asap-types", "asap_sketchlib", "chrono", @@ -376,7 +378,7 @@ dependencies = [ [[package]] name = "asap-frontend-common" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=f0dc1825805f4cbb8866b3baa15bed4319c4e1ab#f0dc1825805f4cbb8866b3baa15bed4319c4e1ab" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=92bc1296fd33802e8ce3fcab6b75f8e3205c3960#92bc1296fd33802e8ce3fcab6b75f8e3205c3960" dependencies = [ "asap-types", "serde", @@ -386,7 +388,7 @@ dependencies = [ [[package]] name = "asap-frontend-metricsql" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=f0dc1825805f4cbb8866b3baa15bed4319c4e1ab#f0dc1825805f4cbb8866b3baa15bed4319c4e1ab" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=92bc1296fd33802e8ce3fcab6b75f8e3205c3960#92bc1296fd33802e8ce3fcab6b75f8e3205c3960" dependencies = [ "asap-frontend-common", "asap-types", @@ -397,7 +399,7 @@ dependencies = [ [[package]] name = "asap-frontend-promql" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=f0dc1825805f4cbb8866b3baa15bed4319c4e1ab#f0dc1825805f4cbb8866b3baa15bed4319c4e1ab" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=92bc1296fd33802e8ce3fcab6b75f8e3205c3960#92bc1296fd33802e8ce3fcab6b75f8e3205c3960" dependencies = [ "asap-frontend-common", "asap-types", @@ -407,7 +409,7 @@ dependencies = [ [[package]] name = "asap-frontend-sql" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=f0dc1825805f4cbb8866b3baa15bed4319c4e1ab#f0dc1825805f4cbb8866b3baa15bed4319c4e1ab" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=92bc1296fd33802e8ce3fcab6b75f8e3205c3960#92bc1296fd33802e8ce3fcab6b75f8e3205c3960" dependencies = [ "asap-frontend-common", "asap-sql-function-catalog", @@ -419,7 +421,7 @@ dependencies = [ [[package]] name = "asap-logical-asap-optimization" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=f0dc1825805f4cbb8866b3baa15bed4319c4e1ab#f0dc1825805f4cbb8866b3baa15bed4319c4e1ab" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=92bc1296fd33802e8ce3fcab6b75f8e3205c3960#92bc1296fd33802e8ce3fcab6b75f8e3205c3960" dependencies = [ "asap-types", "asap_sketchlib", @@ -431,7 +433,7 @@ dependencies = [ [[package]] name = "asap-physical-asap-optimization" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=f0dc1825805f4cbb8866b3baa15bed4319c4e1ab#f0dc1825805f4cbb8866b3baa15bed4319c4e1ab" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=92bc1296fd33802e8ce3fcab6b75f8e3205c3960#92bc1296fd33802e8ce3fcab6b75f8e3205c3960" dependencies = [ "asap-logical-asap-optimization", "asap-types", @@ -441,7 +443,7 @@ dependencies = [ [[package]] name = "asap-plan-selection" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=f0dc1825805f4cbb8866b3baa15bed4319c4e1ab#f0dc1825805f4cbb8866b3baa15bed4319c4e1ab" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=92bc1296fd33802e8ce3fcab6b75f8e3205c3960#92bc1296fd33802e8ce3fcab6b75f8e3205c3960" dependencies = [ "asap-logical-asap-optimization", "asap-physical-asap-optimization", @@ -455,7 +457,7 @@ dependencies = [ [[package]] name = "asap-planner" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=f0dc1825805f4cbb8866b3baa15bed4319c4e1ab#f0dc1825805f4cbb8866b3baa15bed4319c4e1ab" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=92bc1296fd33802e8ce3fcab6b75f8e3205c3960#92bc1296fd33802e8ce3fcab6b75f8e3205c3960" dependencies = [ "asap-executor", "asap-frontend-metricsql", @@ -472,12 +474,12 @@ dependencies = [ [[package]] name = "asap-sql-function-catalog" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=f0dc1825805f4cbb8866b3baa15bed4319c4e1ab#f0dc1825805f4cbb8866b3baa15bed4319c4e1ab" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=92bc1296fd33802e8ce3fcab6b75f8e3205c3960#92bc1296fd33802e8ce3fcab6b75f8e3205c3960" [[package]] name = "asap-types" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=f0dc1825805f4cbb8866b3baa15bed4319c4e1ab#f0dc1825805f4cbb8866b3baa15bed4319c4e1ab" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=92bc1296fd33802e8ce3fcab6b75f8e3205c3960#92bc1296fd33802e8ce3fcab6b75f8e3205c3960" dependencies = [ "serde", "serde_json", @@ -1783,7 +1785,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -2640,7 +2642,7 @@ checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" [[package]] name = "metricsql_common" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=f0dc1825805f4cbb8866b3baa15bed4319c4e1ab#f0dc1825805f4cbb8866b3baa15bed4319c4e1ab" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=92bc1296fd33802e8ce3fcab6b75f8e3205c3960#92bc1296fd33802e8ce3fcab6b75f8e3205c3960" dependencies = [ "chrono", ] @@ -2648,7 +2650,7 @@ dependencies = [ [[package]] name = "metricsql_parser" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=f0dc1825805f4cbb8866b3baa15bed4319c4e1ab#f0dc1825805f4cbb8866b3baa15bed4319c4e1ab" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=92bc1296fd33802e8ce3fcab6b75f8e3205c3960#92bc1296fd33802e8ce3fcab6b75f8e3205c3960" dependencies = [ "ahash", "chrono", @@ -3130,7 +3132,7 @@ dependencies = [ "once_cell", "socket2", "tracing", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -3378,7 +3380,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -3778,7 +3780,7 @@ dependencies = [ "getrandom 0.4.3", "once_cell", "rustix", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -4345,7 +4347,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index d2633f34..a6448940 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -10,8 +10,8 @@ version = "0.1.0" # The backend depends on two Planner crates, at one immutable revision: # `asap-planner` to plan and `asap-executor` to execute. Planner IR types # are reached through `asap_executor::planner`. -asap-planner = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "f0dc1825805f4cbb8866b3baa15bed4319c4e1ab" } -asap-executor = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "f0dc1825805f4cbb8866b3baa15bed4319c4e1ab" } +asap-planner = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "92bc1296fd33802e8ce3fcab6b75f8e3205c3960" } +asap-executor = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "92bc1296fd33802e8ce3fcab6b75f8e3205c3960" } serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" diff --git a/README.md b/README.md index 9ed8b0ed..19207735 100644 --- a/README.md +++ b/README.md @@ -46,11 +46,12 @@ across the ingestion workers by a hash of their label set; a pane closes passes its end, and later samples for it are dropped. The first pane after startup is skipped because it may be partial. Each query, including every step of a range query, runs on one query thread; queries read the stored -panes concurrently, by shared reference. Ingestion updates each open pane's summary -state per sample, with the updater the Planner executor's batch build uses; -a definition whose build has another shape (a row filter, a keyed family, -operators between its range and its summary) is built from the pane's -buffered samples instead. The data plane uses jemalloc as its allocator. +panes concurrently, by shared reference. The plan is compiled once and shared by every thread. Ingestion updates +each open pane's summary state per sample with the executor's +`StateColumnUpdater`, and builds the pane from columns; a definition whose +build has another shape (a row filter, a keyed family, operators between +its range and its summary, groups that are not series) is built once over +the pane's samples instead. Batches are Arrow columns throughout. The data plane uses jemalloc as its allocator. An instant query whose text is one of the plan's queries is answered by the plan (`x-asap-execution: plan`). Its stored inputs read the panes ending at diff --git a/data_plane/src/engine.rs b/data_plane/src/engine.rs index 88cf86d4..bff04e52 100644 --- a/data_plane/src/engine.rs +++ b/data_plane/src/engine.rs @@ -1,10 +1,10 @@ //! The data plane's threads: ingestion workers that build stored panes, and //! a pool of query threads that answer planned queries. //! -//! Planner IR and the executor's compiled plans hold `Rc`s, so they cannot -//! be shared across threads: every worker and query thread reads the plan -//! and compiles its own. What threads share is plain data: the stored panes, -//! the kept query-time results, and the watermarks. +//! The plan is compiled once, when it is installed: its ingestion side +//! ([`Ingestion`]) and its query roots ([`QueryEngine`]). Every thread +//! shares them, with the stored panes, the kept query-time results and the +//! watermarks. //! //! - **Ingestion.** Series are sharded by a hash of their label set; each //! worker owns the open panes of its shard. A worker's watermark is its @@ -22,7 +22,7 @@ use std::sync::{Arc, Mutex}; use asap_planner::InstallablePlan; use tokio::sync::{mpsc, oneshot}; -use crate::precompute::{PrecomputeStats, RawSample, ShardWorker, Shared}; +use crate::precompute::{Ingestion, PrecomputeStats, RawSample, ShardWorker, Shared}; use crate::query::{KeptResults, QueryEngine, RawRange, ResultSample, Unanswered}; use crate::raw_buffer::RawBuffer; use crate::sds::Sds; @@ -152,6 +152,9 @@ impl EngineHandle { horizons.values_mut().for_each(|ms| *ms = override_ms); } tracing::info!(?horizons, "local raw buffer"); + let ingestion = Arc::new(Ingestion::new(&plan, &sds, true)?); + let kept = Arc::new(KeptResults::default()); + let queries_engine = Arc::new(QueryEngine::with_kept(&plan, &sds, kept)); drop(plan); let raw_buffer = (!horizons.is_empty()).then(|| { Arc::new(RawBuffer::with_horizons( @@ -159,7 +162,6 @@ impl EngineHandle { threads.ingestion_workers, )) }); - let plan_json = Arc::new(plan_json); let precompute = Arc::new(Shared::new(threads.ingestion_workers, allowed_lateness_ms)); let (senders, receivers): (Vec<_>, Vec<_>) = (0..precompute.workers()) .map(|_| mpsc::channel(QUEUE)) @@ -168,27 +170,11 @@ impl EngineHandle { // senders let the channels close when the handle is dropped. let peers: Vec<_> = senders.iter().map(mpsc::Sender::downgrade).collect(); for (index, mut commands) in receivers.into_iter().enumerate() { - let (plan_json, shared, peers, raw_buffer) = ( - plan_json.clone(), - precompute.clone(), - peers.clone(), - raw_buffer.clone(), - ); + let mut worker = ShardWorker::new(index, ingestion.clone(), precompute.clone()) + .with_raw_buffer(raw_buffer.clone()); + let peers = peers.clone(); spawn(format!("ingest-{index}"), move |ready| { - let started = install(&plan_json).and_then(|(plan, sds)| { - Ok(ShardWorker::new(&plan, &sds, index, shared, true)? - .with_raw_buffer(raw_buffer)) - }); - let mut worker = match started { - Ok(worker) => { - let _ = ready.send(Ok(())); - worker - } - Err(error) => { - let _ = ready.send(Err(error)); - return; - } - }; + let _ = ready.send(Ok(())); while let Some(command) = commands.blocking_recv() { match command { WorkerCommand::Ingest(samples) => { @@ -214,22 +200,9 @@ impl EngineHandle { } let (queries, jobs) = sync_channel(QUEUE); let jobs = Arc::new(Mutex::new(jobs)); - let kept = Arc::new(KeptResults::default()); for index in 0..threads.query_threads.max(1) { - let (plan_json, shared, kept, jobs) = ( - plan_json.clone(), - precompute.clone(), - kept.clone(), - jobs.clone(), - ); + let (engine, shared, jobs) = (queries_engine.clone(), precompute.clone(), jobs.clone()); spawn(format!("query-{index}"), move |ready| { - let engine = match install(&plan_json) { - Ok((plan, sds)) => QueryEngine::with_kept(&plan, &sds, kept), - Err(error) => { - let _ = ready.send(Err(error)); - return; - } - }; let _ = ready.send(Ok(())); serve_queries(&engine, &shared, &jobs); })?; diff --git a/data_plane/src/lib.rs b/data_plane/src/lib.rs index 3a008b9b..4852113d 100644 --- a/data_plane/src/lib.rs +++ b/data_plane/src/lib.rs @@ -17,6 +17,7 @@ pub mod raw_buffer; pub mod remote_read; 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 index 655bb676..6307bed7 100644 --- a/data_plane/src/per_sample.rs +++ b/data_plane/src/per_sample.rs @@ -1,84 +1,102 @@ //! Per-sample ingestion: a definition's build, `Scan -> TimeShift* -> -//! TimeRange(pane) -> SummaryAgg`, run one sample at a time instead of -//! over a pane's buffered rows. +//! TimeRange(pane) -> SummaryAgg`, run one sample at a time, in columns. //! -//! - **Per series, once:** the stateless prefix. The executor's -//! `RawSampleSelector` turns the series' first sample into its input row, -//! which applies the `Scan`'s label matchers; the row's group columns give -//! its group. The result is cached by label set. -//! - **Per sample:** the value updates its open pane's state for its group -//! in place, through the updater the executor's batch build uses for the -//! family (`asap_executor::factory::create_planner_accumulator`), with the -//! same value and timestamp, in the same order. -//! - **Per group, once:** the output row's layout. The definition's compiled -//! build runs over the group's first row, and its output row is kept as -//! the group's template; at pane close each group's row is its template -//! with the pane's timestamp and the group's state. +//! - **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 build of another shape (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) keeps the batch build. +//! A group is one series (the group columns include the series identity), +//! and workers own disjoint series, so each worker's part of a pane is +//! complete for its groups: it is sealed and stored as its own batch, and +//! the pane is the workers' batches. There is nothing to merge. +//! +//! 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 (which could span workers). The +//! Planner places no such build at ingestion time for the PromQL examples. use std::collections::{BTreeMap, HashMap}; -use std::sync::Arc; +use std::sync::{Arc, OnceLock}; -use asap_executor::factory::{create_planner_accumulator, AccumulatorUpdater}; +use asap_executor::arrow::array::{Array, TimestampMillisecondArray, UInt32Array}; use asap_executor::planner::ir::operator::Reduction; use asap_executor::planner::ir::scalar::ColumnRef; use asap_executor::planner::ir::schema::{ - ExactKind, FieldDataType, GroupingStrategy, SummaryInputExpr, SummaryUpdate, + DataType, FieldDataType, GroupingStrategy, SummaryInputExpr, }; use asap_executor::planner::ir::{ASAPOp, NonASAPOp, Operator}; use asap_executor::runtime::Scope; -use asap_executor::values::{Batch, Value}; +use asap_executor::values::{Batch, BatchColumn, SchemaRef}; +use asap_executor::StateColumnUpdater; use asap_planner::PhysicalASAPDAG; use crate::precompute::{PaneBuild, RawSample}; +use crate::series::{text_column, SeriesCells}; -type Labels = Arc>; +/// The `Scan` schema's column that holds a series' full identity. +const SERIES_IDENTITY_COLUMN: &str = "$promql_series_identity"; -/// One group of a definition: its key and what its output rows repeat. -struct Group { - /// The group's first input row, for its template. - row: Batch, - /// Its output row, with the timestamp and state to be replaced; built - /// when the group's first pane closes. - template: Option>, -} +/// A group: its texts in the `SummaryAgg`'s group columns, in their order. +type GroupKey = Arc<[Arc]>; -/// One group's state in one open pane. -struct OpenState { - updater: Box, - /// The previous sample's time, for counters, which need increasing times. - previous_ms: Option, +/// Groups numbered `0..`, in the order they are first seen. +#[derive(Default)] +struct GroupNumbers { + numbers: HashMap, + keys: Vec, } -/// One open pane: its groups' states, by group key, or why it failed. -type OpenPane = Result>, OpenState>, String>; +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 + } +} -/// A definition's build, run per sample. -pub(crate) struct PerSampleBuild { +/// How a definition is built per sample; one per definition, shared by +/// every ingestion worker. +pub(crate) struct PerSampleSpec { family: FieldDataType, - update: SummaryUpdate, - grouping: GroupingStrategy, - /// Input columns: the value, the time, the group key. - value: usize, - time: usize, + /// The input group columns, in the `SummaryAgg`'s order. groups: Vec, - /// Counters (rate, increase) need strictly increasing times in a group. - ordered_time: bool, - /// Output columns of the state and the timestamp. + output: SchemaRef, + /// Output column of the states, and of the pane's timestamp. state_column: usize, time_column: usize, - /// Each series' group key, or `None` if the `Scan` does not select it. - series: HashMap>>>, - group_rows: HashMap>, Group>, - /// How much the output timestamp is after the pane's end. - time_offset_ms: Option, - open: BTreeMap, + /// 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, +} + +/// One worker's open panes of a definition built per sample. +#[derive(Default)] +pub(crate) struct OpenPanes { + /// This worker's numbers for its groups (its series). + numbers: GroupNumbers, + /// Each series' group key, or `None` when the `Scan` drops it, by the + /// address of its cached cells. + series_groups: HashMap>, + /// By pane end: its states and the groups that took a sample, or why it + /// failed. + panes: BTreeMap), String>>, } -impl PerSampleBuild { +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: &PaneBuild) -> Option { @@ -128,6 +146,12 @@ impl PerSampleBuild { } _ => 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 { @@ -137,143 +161,206 @@ impl PerSampleBuild { if columns_read(&read).any(|c| c == value || c == time) { return None; } - create_planner_accumulator(family, update, grouping).ok()?; - let output = &definition.output_schema; + 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?; - let ordered_time = matches!( - family, - FieldDataType::ExactAggregate(ExactKind::Rate | ExactKind::Increase, _) - ); + // 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, so it never spans workers. + if !groups + .iter() + .any(|&c| input.fields[c].name == SERIES_IDENTITY_COLUMN) + { + return None; + } Some(Self { family: family.clone(), - update: update.clone(), - grouping: grouping.clone(), - value, - time, groups, - ordered_time, + output, state_column, time_column, - series: HashMap::new(), - group_rows: HashMap::new(), - time_offset_ms: None, - open: BTreeMap::new(), + output_group_positions, + time_offset_ms: OnceLock::new(), }) } - /// Updates the pane ending at `end_ms` with `sample`. - pub(crate) fn update(&mut self, pane: &PaneBuild, sample: &RawSample, end_ms: i64) { - let key = match self.series.get(&sample.labels) { - Some(key) => key.clone(), + /// 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, + ) { + let address = Arc::as_ptr(cells) as usize; + let group = match open.series_groups.get(&address) { + Some(group) => group.clone(), None => { - let key = self.series_group(pane, sample); - self.series.insert(sample.labels.clone(), key.clone()); - key + let key: GroupKey = self + .groups + .iter() + .map(|&c| cells.texts[c].clone().unwrap_or_else(|| Arc::from(""))) + .collect(); + let number = open.numbers.number(&key); + // The cells stay cached by the window, so the address is + // never reused for another series. + let group = Some((key, number)); + open.series_groups.insert(address, group.clone()); + group } }; - let Some(key) = key else { return }; - let Ok(states) = self - .open - .entry(end_ms) - .or_insert_with(|| Ok(BTreeMap::new())) - else { + 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 state = match states.get_mut(&key) { - Some(state) => state, - None => { - let updater = - create_planner_accumulator(&self.family, &self.update, &self.grouping) - .expect("checked at install"); - states.entry(key).or_insert(OpenState { - updater, - previous_ms: None, - }) + 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; } - }; - let time = sample.timestamp_ms; - let result = if self.ordered_time && state.previous_ms.is_some_and(|p| time <= p) { - Err("counter samples must have strictly increasing timestamps within each group".into()) - } else { - state.previous_ms = Some(time); - state - .updater - .update_value(&Value::Float64(sample.value), time) - }; - if let Err(error) = result { - self.open.insert(end_ms, Err(error)); + Err(error) => *entry = Err(error.to_string()), } } - /// The group of `sample`'s series: its input row's group columns, if - /// the `Scan` selects it. - fn series_group(&mut self, pane: &PaneBuild, sample: &RawSample) -> Option>> { - let row = pane - .selector - .rows(sample.timestamp_ms, &[sample.to_sample()]) - .ok()?; - let cells = row.rows().first()?; - debug_assert!(matches!(cells[self.value], Value::Float64(_))); - debug_assert!(matches!(cells[self.time], Value::Timestamp(_))); - let key = self - .groups + /// This worker's part of the pane ending at `end_ms`, which closes: a + /// row per group (series) that took a sample. + pub(crate) fn close( + &self, + pane: &PaneBuild, + 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() - .map(|&c| cells[c].key()) - .collect::, _>>() - .ok()?; - self.group_rows.entry(key.clone()).or_insert(Group { - row, - template: None, - }); - Some(key) + .enumerate() + .filter(|(_, &took)| took) + .map(|(number, _)| number as u32) + .collect(); + self.pane_batch(pane, updater, &touched, &open.numbers.keys, end_ms, scope) } - /// The pane ending at `end_ms`, which closes: each group's row, in group - /// key order as the batch build emits them. - pub(crate) fn close( - &mut self, + /// 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: &PaneBuild, + updater: StateColumnUpdater, + touched: &[u32], + keys: &[GroupKey], end_ms: i64, - scope: Scope, + scope: &Scope, ) -> Result { - let states = self - .open - .remove(&end_ms) - .unwrap_or_else(|| Ok(BTreeMap::new()))?; - let mut rows = Vec::with_capacity(states.len()); - for (key, state) in states { - let group = self.group_rows.get_mut(&key).expect("grouped series"); - if group.template.is_none() { - let built = pane - .run_rows(group.row.clone(), scope.clone()) - .map_err(|e| e.to_string())?; - let [row] = built.rows() else { - return Err("a group's build did not give one row".into()); - }; - let Value::Timestamp(time) = row[self.time_column] else { - return Err("output time column is not a timestamp".into()); - }; - let offset = time - end_ms; - if *self.time_offset_ms.get_or_insert(offset) != offset { - return Err("output timestamps differ between groups".into()); + 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(), + }); } - group.template = Some(row.clone()); - } - let mut row = group.template.clone().expect("built above"); - row[self.time_column] = - Value::Timestamp(end_ms + self.time_offset_ms.expect("set with the template")); - row[self.state_column] = Value::Summary { - family: self.family.clone(), - state: Arc::from(state.updater.into_accumulator()), - }; - rows.push(row); - } - let schema = Arc::new(pane.definition.output_schema.clone()); - Batch::try_new(schema, rows).map_err(|e| e.to_string()) + 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: &PaneBuild, 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 + .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) } } diff --git a/data_plane/src/precompute.rs b/data_plane/src/precompute.rs index 82d28123..5cb9f5c2 100644 --- a/data_plane/src/precompute.rs +++ b/data_plane/src/precompute.rs @@ -1,17 +1,19 @@ //! Precompute engine: builds each stored summary pane once, when the //! watermark passes its end. //! -//! For every [`SummaryDefinition`] built at ingestion time the engine -//! buffers the samples of its pane'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 with -//! the executor's `RawSampleSelector`, which applies the plan's `Scan` -//! (its label matchers), then runs the definition's build from its -//! `TimeRange` 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. +//! The plan's ingestion side is compiled once ([`Ingestion`]) and shared by +//! the ingestion workers, each of which owns a shard of the series. A pane +//! `(end - pane, end]` closes when the global watermark (the minimum over +//! the workers of their newest sample, minus the allowed lateness) reaches +//! `end`. A definition is built per sample where its shape allows +//! ([`crate::per_sample`]): each sample updates its group's open state in +//! place, and each worker's part of the pane is built from columns at +//! close. Otherwise each worker keeps the pane's samples, and the worker +//! that completes the pane runs the plan's build once over all of them +//! ([`PaneBuild`]). The pane is stored and never changes; samples +//! older than a closed pane are dropped. The first pane may have started +//! before the data plane did, so it is never built: queries that need it +//! are not answered from stored panes. use std::collections::{BTreeMap, HashMap}; use std::hash::{Hash, Hasher}; @@ -21,17 +23,16 @@ use std::sync::{Arc, Mutex, OnceLock}; use asap_executor::operators::Operator; use asap_executor::physical_planner::{compile, CompiledPhysicalDAG, InputContract, Source}; use asap_executor::plan::PhysicalDAG; -use asap_executor::planner::ir::schema::FieldDataType; use asap_executor::runtime::{Limits, RunContext, Scope}; -use asap_executor::values::{Batch, SchemaRef, Value}; -use asap_executor::{RawSampleSelector, Sample}; +use asap_executor::values::{Batch, SchemaRef}; use asap_planner::{InstallablePlan, PhysicalASAPDAG}; use futures::executor::block_on; use futures::StreamExt; -use crate::per_sample::PerSampleBuild; +use crate::per_sample::{OpenPanes, PerSampleSpec}; use crate::raw_buffer::RawBuffer; use crate::sds::{DefinitionId, Sds, SummaryDefinition, SummaryStore}; +use crate::series::SeriesWindow; /// One raw sample: its series' full label set (including `__name__`), time /// and value. Samples of one series share their labels. @@ -42,17 +43,6 @@ pub struct RawSample { pub value: f64, } -impl RawSample { - /// The executor's sample, for a `RawSampleSelector`. - pub fn to_sample(&self) -> Sample { - Sample { - labels: (*self.labels).clone(), - timestamp_ms: self.timestamp_ms, - value: self.value, - } - } -} - #[derive(Debug, thiserror::Error)] pub enum PrecomputeError { #[error("stored summary {0:?}: {1}")] @@ -74,13 +64,19 @@ pub struct PrecomputeStats { pub panes_failed: u64, } -/// How the panes of one definition are built from raw samples. The -/// precompute engine builds stored panes with it, and the query engine kept -/// ones. +/// A pane: its batches, one row per group across them. Built per sample at +/// ingestion time, the batches are the workers' parts, whose groups (their +/// series) are disjoint. +pub type Pane = Vec; + +/// How the panes of one definition are built from raw samples with the +/// plan's own build. Compiled once and shared by every thread: the +/// precompute engine builds stored panes with it (where it does not build +/// them per sample), and the query engine kept ones. pub(crate) struct PaneBuild { pub(crate) definition: SummaryDefinition, - /// Selects one pane's rows from raw samples. - pub(crate) selector: RawSampleSelector, + /// The pane's raw-sample window: its input batch from raw samples. + pub(crate) window: SeriesWindow, build: CompiledPhysicalDAG, } @@ -90,47 +86,102 @@ impl PaneBuild { dag: &PhysicalASAPDAG, definition: &SummaryDefinition, ) -> Result { - let selector = RawSampleSelector::new(dag, &definition.window) + let window = SeriesWindow::new(dag, &definition.window) .map_err(|e| PrecomputeError::Definition(definition.id.clone(), e.to_string()))?; - let raw_schema = Arc::new(output_schema(dag, definition.raw_input()).clone()); let build = compile( dag, - BTreeMap::from([(definition.raw_input(), InputContract::bounded(raw_schema))]), + BTreeMap::from([( + definition.raw_input(), + InputContract::bounded(window.schema.clone()), + )]), &[definition.builder], )?; Ok(Self { definition: definition.clone(), - selector, + window, build, }) } - /// The pane ending at `end_ms`, built from `samples` under `scope`: - /// the selector keeps the pane's samples that the plan's `Scan` selects, - /// then the build runs over them. + /// The pane ending at `end_ms`, built from `samples` under `scope`: the + /// samples of the pane the plan's `Scan` selects, then the build. pub(crate) fn run( &self, - samples: &[Sample], + samples: &[RawSample], end_ms: i64, scope: Scope, - ) -> Result { - self.run_rows(self.selector.rows(end_ms, samples)?, scope) + ) -> Result { + self.run_batch(self.window.batch(end_ms, samples)?, scope) } - /// The build over `rows`, input rows already selected, under `scope`. - pub(crate) fn run_rows(&self, rows: Batch, scope: Scope) -> Result { - let input: Source<'_> = Box::new(Operator::source(rows.schema().clone(), vec![rows])?); + /// The build over `input`, the pane's selected input batch, under `scope`. + pub(crate) fn run_batch(&self, input: Batch, scope: Scope) -> Result { + let source: 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(&build, self.build.roots(), scope)?; - let schema = Arc::new(self.definition.output_schema.clone()); - Ok(Batch::try_new(schema, rows)?) + .instantiate(BTreeMap::from([(self.definition.raw_input(), source)]))?; + Ok(run(&build, self.build.roots(), scope)?) } } -/// One worker's part of a pane, or why its build failed. -type PanePart = Result; +/// One definition built at ingestion time. +pub(crate) struct IngestionBuild { + pane: PaneBuild, + /// The build run per sample, unless its shape keeps the batch build. + per_sample: Option, +} + +/// The plan's ingestion side, compiled once and shared by every ingestion +/// worker: one build per definition built at ingestion time. +pub struct Ingestion { + builds: Vec, +} + +impl Ingestion { + /// The builds of `sds`'s ingestion-time definitions, the stored + /// summaries of `plan`: per sample where its shape allows + /// ([`crate::per_sample`]) and `per_sample` is set, else over each + /// pane's buffered samples. + pub fn new( + plan: &InstallablePlan, + sds: &Sds, + per_sample: bool, + ) -> Result { + let builds = sds + .definitions + .values() + .map(|definition| { + let pane = PaneBuild::new(&plan.dag.dag, definition)?; + let per_sample = per_sample + .then(|| PerSampleSpec::new(&plan.dag.dag, &pane)) + .flatten(); + tracing::info!(definition = ?definition.id, per_sample = per_sample.is_some(), "pane build"); + Ok(IngestionBuild { pane, per_sample }) + }) + .collect::>()?; + Ok(Self { builds }) + } + + /// How many definitions are built per sample. + pub fn per_sample_definitions(&self) -> usize { + self.builds + .iter() + .filter(|b| b.per_sample.is_some()) + .count() + } +} + +/// One worker's part of a pane. +enum PanePart { + /// Built per sample: the rows of the worker's groups. + Batch(Batch), + /// For the batch build: the worker's samples of the pane, built with + /// the other workers' at once, since a group may span workers. + Samples(Vec), +} + +/// One worker's part of a pane, or why it failed. +type PartOrError = Result; /// No value yet in an `AtomicI64` slot. const UNSET: i64 = i64::MIN; @@ -140,7 +191,7 @@ const UNSET: i64 = i64::MIN; /// not yet complete. pub struct Shared { /// Stored panes, by definition and pane end. - pub store: SummaryStore, + pub store: SummaryStore, allowed_lateness_ms: i64, /// Each worker's newest sample time, `UNSET` before its first. newest_sample_ms: Vec, @@ -151,7 +202,7 @@ pub struct Shared { announced_ms: AtomicI64, /// Partial panes by definition and end, until every worker has built /// its part. - pending: Mutex>>, + pending: Mutex>>, accepted: AtomicU64, dropped: AtomicU64, panes_built: AtomicU64, @@ -200,9 +251,16 @@ impl Shared { } } - /// Adds one worker's part of the pane of `definition` ending at `end_ms`. - /// The worker that adds the last part merges them and stores the pane. - fn add_part(&self, definition: &SummaryDefinition, end_ms: i64, part: Result) { + /// Adds one worker's part of the pane of `build` ending at `end_ms`. + /// The worker that adds the last part completes and stores the pane. + fn add_part( + &self, + build: &IngestionBuild, + end_ms: i64, + scope: &Scope, + part: Result, + ) { + let definition = &build.pane.definition; let parts = { let mut pending = self.pending.lock().expect("pending panes lock"); let key = (definition.id.clone(), end_ms); @@ -213,19 +271,33 @@ impl Shared { } pending.remove(&key).expect("present") }; - match parts - .into_iter() - .collect::, _>>() - .and_then(|parts| merge_parts(definition, parts)) - { + let complete = || -> Result { + let mut pane = Pane::new(); + let mut samples = Vec::new(); + for part in parts { + match part? { + PanePart::Batch(batch) => pane.push(batch), + PanePart::Samples(part) => samples.extend(part), + } + } + if build.per_sample.is_none() { + pane = build + .pane + .run(&samples, end_ms, scope.clone()) + .map_err(|e| e.to_string())?; + } + Ok(pane) + }; + match complete() { Ok(pane) => { self.store.insert(&definition.id, end_ms, pane); self.panes_built.fetch_add(1, Ordering::Relaxed); - // The newest evaluation that reads stored panes is at most - // the newest stored pane's end, so retention is measured - // from it, not from the watermark, which may be past it. - self.store - .evict_before(&definition.id, definition.oldest_kept_end_ms(end_ms)); + // The newest evaluation that reads stored panes is about the + // newest stored pane's end, but may lag it: an evaluation at + // T runs once samples up to just after T have arrived. One + // pane width of margin keeps the panes it reads. + let oldest = definition.oldest_kept_end_ms(end_ms) - definition.pane_ms; + self.store.evict_before(&definition.id, oldest); } Err(error) => { tracing::error!(definition = ?definition.id, end_ms, %error, "pane build failed"); @@ -235,55 +307,6 @@ impl Shared { } } -/// One pane from the parts the workers built over their own series: their -/// rows, with the states of a group several workers saw merged by the -/// family's merge. Each worker's part has one row per group, so a group -/// present in one part only is kept as built. -fn merge_parts(definition: &SummaryDefinition, parts: Vec) -> Result { - let schema = Arc::new(definition.output_schema.clone()); - let state = definition - .output_schema - .fields - .iter() - .position(|f| !matches!(f.dtype, FieldDataType::Plain(_))); - let mut rows: BTreeMap>, Vec> = BTreeMap::new(); - for part in &parts { - for row in part.rows() { - let key = row - .iter() - .enumerate() - .filter(|(i, _)| Some(*i) != state) - .map(|(_, value)| value.key()) - .collect::, _>>() - .map_err(|e| e.to_string())?; - match (rows.get_mut(&key), state) { - (Some(existing), Some(state)) => { - let ( - Value::Summary { state: left, .. }, - Value::Summary { - family, - state: right, - }, - ) = (&existing[state], &row[state]) - else { - return Err("summary state column without a state".into()); - }; - let merged = left.merge_with(right.as_ref()).map_err(|e| e.to_string())?; - existing[state] = Value::Summary { - family: family.clone(), - state: Arc::from(merged), - }; - } - (Some(_), None) => return Err("duplicate group without a state".into()), - (None, _) => { - rows.insert(key, row.clone()); - } - } - } - } - Batch::try_new(schema, rows.into_values().collect()).map_err(|e| e.to_string()) -} - /// The ingestion worker of `workers` that owns the series with `labels`. pub fn shard_of(labels: &BTreeMap, workers: usize) -> usize { let mut hasher = std::collections::hash_map::DefaultHasher::new(); @@ -291,25 +314,27 @@ pub fn shard_of(labels: &BTreeMap, workers: usize) -> usize { (hasher.finish() % workers.max(1) as u64) as usize } -struct Builder { - pane: PaneBuild, - /// The build run per sample, unless its shape keeps the batch build. - per_sample: Option, - /// For the batch build, this worker's samples of open panes, by pane end. - open: BTreeMap>, +/// One worker's open panes of one definition. +#[derive(Default)] +struct OpenBuild { + /// For the batch build: this worker's samples of open panes, by end. + samples: BTreeMap>, + /// For the per-sample build: the open panes' states. + states: OpenPanes, /// End of the newest pane this worker built its part of, or skipped; /// samples in it or before it are dropped. `None` before the first /// sample anywhere. built_through: Option, } -/// One ingestion worker: buffers the samples of its shard of the series and -/// builds its part of each pane once the global watermark passes it. It -/// holds the plan's compiled builds, which are not `Send`, so each worker -/// thread compiles its own. +/// One ingestion worker: takes the samples of its shard of the series into +/// its open panes and builds its part of each pane once the global +/// watermark passes it. The compiled builds are shared; the open panes are +/// the worker's own. pub struct ShardWorker { index: usize, - builders: Vec, + ingestion: Arc, + open: Vec, shared: Arc, /// The global watermark this worker has built its parts through. closed_through_ms: Option, @@ -319,43 +344,21 @@ pub struct ShardWorker { } impl ShardWorker { - /// Worker `index` of `shared`'s workers, with one build per - /// ingestion-time definition of `sds`, the stored summaries of `plan`: - /// per sample where its shape allows ([`crate::per_sample`]) and - /// `per_sample` is set, else over each pane's buffered samples. - pub fn new( - plan: &InstallablePlan, - sds: &Sds, - index: usize, - shared: Arc, - per_sample: bool, - ) -> Result { - let builders = sds - .definitions - .values() - .map(|definition| { - let pane = PaneBuild::new(&plan.dag.dag, definition)?; - let per_sample = per_sample - .then(|| PerSampleBuild::new(&plan.dag.dag, &pane)) - .flatten(); - if index == 0 { - tracing::info!(definition = ?definition.id, per_sample = per_sample.is_some(), "pane build"); - } - Ok(Builder { - pane, - per_sample, - open: BTreeMap::new(), - built_through: None, - }) - }) - .collect::>()?; - Ok(Self { + /// Worker `index` of `shared`'s workers, running `ingestion`'s builds. + pub fn new(index: usize, ingestion: Arc, shared: Arc) -> Self { + let open = ingestion + .builds + .iter() + .map(|_| OpenBuild::default()) + .collect(); + Self { index, - builders, + ingestion, + open, shared, closed_through_ms: None, raw_buffer: None, - }) + } } /// Also keeps this worker's samples in `raw_buffer`, its shard @@ -365,10 +368,10 @@ impl ShardWorker { self } - /// Buffers `samples`, all of this worker's shard, into their panes, then - /// builds this worker's part of every pane the global watermark has - /// passed. Returns whether this moved the global watermark past what - /// was announced, so the caller tells the other workers to + /// Takes `samples`, all of this worker's shard, into their open panes, + /// then builds this worker's part of every pane the global watermark + /// has passed. Returns whether this moved the global watermark past + /// what was announced, so the caller tells the other workers to /// [`Self::advance`]. pub fn ingest(&mut self, samples: &[RawSample]) -> bool { let Some(first) = samples.iter().map(|s| s.timestamp_ms).min() else { @@ -382,36 +385,59 @@ impl ShardWorker { .start_ms .compare_exchange(UNSET, first, Ordering::AcqRel, Ordering::Acquire); let start_ms = shared.start_ms.load(Ordering::Acquire); - let (mut accepted, mut dropped) = (0, 0); - for sample in samples { - let metric = sample.labels.get("__name__").map(String::as_str); - let mut buffered = false; - for builder in &mut self.builders { - if metric != Some(builder.pane.selector.metric()) { + let mut buffered = vec![false; samples.len()]; + let mut dropped = 0; + let ingestion = self.ingestion.clone(); + // Samples of one series arrive together: each build looks the + // series up once per run. + let mut offset = 0; + for run in samples.chunk_by(|a, b| Arc::ptr_eq(&a.labels, &b.labels)) { + let metric = run[0].labels.get("__name__").map(String::as_str); + for (build, open) in ingestion.builds.iter().zip(&mut self.open) { + if metric != Some(build.pane.window.metric()) { continue; } - let pane_ms = builder.pane.definition.pane_ms; - let end = pane_end(sample.timestamp_ms, pane_ms); + let pane_ms = build.pane.definition.pane_ms; // The pane of the first sample may be partial: skip it. - let built = *builder + let built = *open .built_through .get_or_insert(pane_end(start_ms, pane_ms)); - if end <= built { - dropped += 1; - continue; - } - match &mut builder.per_sample { - Some(per_sample) => per_sample.update(&builder.pane, sample, end), - None => builder - .open - .entry(end) - .or_default() - .push(sample.to_sample()), + let cells = match &build.per_sample { + Some(_) => match build.pane.window.cells(&run[0]) { + Some(cells) => Some(cells), + // Not selected: counted as taken, as the batch + // build takes and then filters it. + None => { + for (i, sample) in run.iter().enumerate() { + if pane_end(sample.timestamp_ms, pane_ms) <= built { + dropped += 1; + } else { + buffered[offset + i] = true; + } + } + continue; + } + }, + None => None, + }; + for (i, sample) in run.iter().enumerate() { + let end = pane_end(sample.timestamp_ms, pane_ms); + if end <= built { + dropped += 1; + continue; + } + match (&build.per_sample, &cells) { + (Some(spec), Some(cells)) => { + spec.update(&mut open.states, cells, sample, end) + } + _ => open.samples.entry(end).or_default().push(sample.clone()), + } + buffered[offset + i] = true; } - buffered = true; } - accepted += u64::from(buffered); + offset += run.len(); } + let accepted = buffered.iter().filter(|&&b| b).count() as u64; shared.accepted.fetch_add(accepted, Ordering::Relaxed); shared.dropped.fetch_add(dropped, Ordering::Relaxed); let newest = samples @@ -430,9 +456,9 @@ impl ShardWorker { } /// Builds this worker's part of every pane the global watermark has - /// passed: from its buffered samples, or empty. Every worker builds every - /// pane after the first, so a pane without samples is stored as empty, - /// not missing. + /// passed: from its open states or samples, or empty. Every worker + /// builds every pane after the first, so a pane without samples is + /// stored as empty, not missing. pub fn advance(&mut self) { if let Some(watermark) = self.shared.watermark_ms() { self.advance_to(watermark); @@ -448,9 +474,10 @@ impl ShardWorker { } self.closed_through_ms = Some(watermark); let start_ms = self.shared.start_ms.load(Ordering::Acquire); - for builder in &mut self.builders { - let pane_ms = builder.pane.definition.pane_ms; - let built = *builder + let ingestion = self.ingestion.clone(); + for (build, open) in ingestion.builds.iter().zip(&mut self.open) { + let pane_ms = build.pane.definition.pane_ms; + let built = *open .built_through .get_or_insert(pane_end(start_ms, pane_ms)); let mut end = built + pane_ms; @@ -460,18 +487,16 @@ impl ShardWorker { window_end_ms: end, revision: 1, }; - let part = match &mut builder.per_sample { - Some(per_sample) => per_sample.close(&builder.pane, end, scope), - None => { - let samples = builder.open.remove(&end).unwrap_or_default(); - builder - .pane - .run(&samples, end, scope) - .map_err(|e| e.to_string()) - } + let part = match &build.per_sample { + Some(spec) => spec + .close(&build.pane, &mut open.states, end, &scope) + .map(PanePart::Batch), + None => Ok(PanePart::Samples( + open.samples.remove(&end).unwrap_or_default(), + )), }; - self.shared.add_part(&builder.pane.definition, end, part); - builder.built_through = Some(end); + self.shared.add_part(build, end, &scope, part); + open.built_through = Some(end); end += pane_ms; } } @@ -482,6 +507,7 @@ impl ShardWorker { /// shard of the series: the threaded engine's precompute without threads. pub struct Precompute { shared: Arc, + ingestion: Arc, workers: Vec, } @@ -502,40 +528,36 @@ impl Precompute { allowed_lateness_ms: i64, shards: usize, ) -> Result { - Self::workers(plan, sds, allowed_lateness_ms, shards, true) + let ingestion = Arc::new(Ingestion::new(plan, sds, true)?); + Ok(Self::workers(ingestion, allowed_lateness_ms, shards)) } /// One worker that builds every pane over its buffered samples with - /// the executor's batch build: the reference per-sample builds equal. + /// the plan's own build: the reference per-sample builds equal. pub fn batch( plan: &InstallablePlan, sds: &Sds, allowed_lateness_ms: i64, ) -> Result { - Self::workers(plan, sds, allowed_lateness_ms, 1, false) + let ingestion = Arc::new(Ingestion::new(plan, sds, false)?); + Ok(Self::workers(ingestion, allowed_lateness_ms, 1)) } - fn workers( - plan: &InstallablePlan, - sds: &Sds, - allowed_lateness_ms: i64, - shards: usize, - per_sample: bool, - ) -> Result { + fn workers(ingestion: Arc, allowed_lateness_ms: i64, shards: usize) -> Self { let shared = Arc::new(Shared::new(shards, allowed_lateness_ms)); let workers = (0..shared.workers()) - .map(|index| ShardWorker::new(plan, sds, index, shared.clone(), per_sample)) - .collect::>()?; - Ok(Self { shared, workers }) + .map(|index| ShardWorker::new(index, ingestion.clone(), shared.clone())) + .collect(); + Self { + shared, + ingestion, + workers, + } } /// How many definitions are built per sample. pub fn per_sample_definitions(&self) -> usize { - self.workers[0] - .builders - .iter() - .filter(|b| b.per_sample.is_some()) - .count() + self.ingestion.per_sample_definitions() } /// Routes `samples` to their workers, then lets every worker build the @@ -557,7 +579,7 @@ impl Precompute { self.shared.watermark_ms() } - pub fn store(&self) -> &SummaryStore { + pub fn store(&self) -> &SummaryStore { &self.shared.store } @@ -591,20 +613,23 @@ fn limits() -> Limits { limits } -/// Executes `dag`'s single root to completion and returns its rows. +/// Executes `dag`'s single root to completion and returns its batches. pub fn run( dag: &PhysicalDAG<'_, Batch, SchemaRef>, roots: &[u64], scope: Scope, -) -> Result>, asap_executor::Error> { +) -> Result, asap_executor::Error> { block_on(async { let context = RunContext::new(scope, limits())?; let mut output = dag.execute(roots, context)?.remove(0); - let mut rows = Vec::new(); + let mut batches = Vec::new(); while let Some(batch) = output.next().await { - rows.extend(batch?.rows().iter().cloned()); + let batch = batch?; + if batch.num_rows() > 0 { + batches.push((*batch).clone()); + } } - Ok(rows) + Ok(batches) }) } diff --git a/data_plane/src/query.rs b/data_plane/src/query.rs index 43247a52..6b3a12d8 100644 --- a/data_plane/src/query.rs +++ b/data_plane/src/query.rs @@ -30,18 +30,17 @@ use std::sync::{Arc, Mutex}; use asap_executor::operators::Operator; use asap_executor::physical_planner::{compile, CompiledPhysicalDAG, InputContract, Source}; -use asap_executor::plan::{Boundedness, Emission, PhysicalOperator, PlanProperties}; use asap_executor::planner::ir::physical_export::RawSampleWindow; use asap_executor::planner::ir::schema::{DataType, FieldDataType, Schema}; use asap_executor::planner::ir::{NonASAPOp, Operator as IrOperator}; -use asap_executor::runtime::{Input, OutputStream, RunContext, Scope}; +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, Sample}; use asap_planner::{InstallablePlan, PhysicalASAPDAG}; -use futures::StreamExt; -use crate::precompute::{output_schema, run, PaneBuild, RawSample}; +use crate::precompute::{output_schema, run, Pane, PaneBuild, RawSample}; use crate::sds::{DefinitionId, Sds, SummaryStore}; +use crate::series::SeriesWindow; /// Raw samples a query needs: `metric`'s samples in `(start_ms, end_ms]`. #[derive(Debug, Clone, PartialEq, Eq)] @@ -74,6 +73,8 @@ pub enum Unanswered { struct StoredInput { node: u64, + /// The output's rows, which its panes hold. + schema: SchemaRef, definition: DefinitionId, offset_ms: i64, pane_ms: i64, @@ -81,6 +82,7 @@ struct StoredInput { struct KeptInput { node: u64, + schema: SchemaRef, definition: DefinitionId, offset_ms: i64, /// The output's own `TimeRange`, which its raw samples feed. @@ -88,10 +90,11 @@ struct KeptInput { } struct RawInput { - /// The window's `TimeRange`, which its rows feed. + /// The window's `TimeRange`, which its batch feeds. node: u64, window: RawSampleWindow, - selector: RawSampleSelector, + /// Its input batch from raw samples, column by column. + series: SeriesWindow, } struct Program { @@ -119,7 +122,7 @@ pub struct KeptResultStats { pub struct KeptResults { /// Kept panes (`MatAtQueryKeptAcrossEvaluations`), by definition and /// pane end, in memory. - pub panes: SummaryStore, + pub panes: SummaryStore, built: AtomicU64, reused: AtomicU64, /// Newest evaluation time, from which kept panes are evicted; `None` @@ -145,9 +148,8 @@ impl KeptResults { } } -/// One thread's compiled roots of the plan. The executor's compiled plans -/// are not `Send`, so each query thread builds its own; they share the -/// stored panes and the [`KeptResults`]. +/// The plan's compiled roots, compiled once and shared by every query +/// thread, with the [`KeptResults`] they share. pub struct QueryEngine { /// Results kept across evaluations, shared with the other query threads. pub kept: Arc, @@ -203,7 +205,7 @@ impl QueryEngine { &self, query: &str, time_ms: i64, - store: &SummaryStore, + store: &SummaryStore, ) -> Result<(usize, Vec), Unanswered> { let (root, program) = self.program(query)?; for input in &program.stored { @@ -216,7 +218,7 @@ impl QueryEngine { let (start_ms, end_ms) = input.window.bounds_ms(time_ms); RawRange { node: input.node, - metric: input.selector.metric().to_string(), + metric: input.series.metric().to_string(), start_ms, end_ms, } @@ -228,7 +230,7 @@ impl QueryEngine { let build = self.kept_build(&input.definition); ranges.push(RawRange { node: input.raw_input, - metric: build.selector.metric().to_string(), + metric: build.window.metric().to_string(), start_ms: end_ms - build.definition.pane_ms, end_ms, }); @@ -247,7 +249,7 @@ impl QueryEngine { &self, root: usize, time_ms: i64, - store: &SummaryStore, + store: &SummaryStore, watermark_ms: Option, raw: &BTreeMap>, ) -> Result, Unanswered> { @@ -264,7 +266,9 @@ impl QueryEngine { for input in &program.stored { sources.insert( input.node, - Box::new(PaneSource(pane(input, time_ms, store)?)), + Box::new( + pane_source(&*pane(input, time_ms, store)?, &input.schema).map_err(failed)?, + ), ); } for input in &program.kept { @@ -291,7 +295,7 @@ impl QueryEngine { }; let pane = self .kept_build(&input.definition) - .run(&samples_of(samples), end_ms, pane_scope) + .run(samples, end_ms, pane_scope) .map_err(|e| Unanswered::Execution(e.to_string()))?; self.kept.built.fetch_add(1, Ordering::Relaxed); // Kept only once closed, as the precompute engine closes @@ -303,15 +307,13 @@ impl QueryEngine { } } }; - sources.insert(input.node, Box::new(PaneSource(pane))); + let source = pane_source(&pane, &input.schema).map_err(failed)?; + sources.insert(input.node, Box::new(source)); } for input in &program.raw { let samples = raw.get(&input.node).map(Vec::as_slice).unwrap_or_default(); let (_, end_ms) = input.window.bounds_ms(time_ms); - let rows = input - .selector - .rows(end_ms, &samples_of(samples)) - .map_err(failed)?; + let rows = input.series.batch(end_ms, samples).map_err(failed)?; let source = Operator::source(rows.schema().clone(), vec![rows]).map_err(failed)?; sources.insert(input.node, Box::new(source)); } @@ -325,8 +327,8 @@ impl QueryEngine { } } let dag = program.compiled.instantiate(sources).map_err(failed)?; - let rows = run(&dag, program.compiled.roots(), 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 root answering `query` at each of `steps` (ascending) and the @@ -339,7 +341,7 @@ impl QueryEngine { &self, query: &str, steps: &[i64], - store: &SummaryStore, + store: &SummaryStore, ) -> Result<(usize, Vec), Unanswered> { let (root, program) = self.program(query)?; let (Some(&first), Some(&last)) = (steps.first(), steps.last()) else { @@ -355,7 +357,7 @@ impl QueryEngine { .iter() .map(|input| RawRange { node: input.node, - metric: input.selector.metric().to_string(), + metric: input.series.metric().to_string(), start_ms: input.window.bounds_ms(first).0, end_ms: input.window.bounds_ms(last).1, }) @@ -375,7 +377,7 @@ impl QueryEngine { let build = self.kept_build(definition); ranges.push(RawRange { node: build.definition.raw_input(), - metric: build.selector.metric().to_string(), + metric: build.window.metric().to_string(), start_ms: first_end_ms - build.definition.pane_ms, end_ms: last_end_ms, }); @@ -394,7 +396,7 @@ impl QueryEngine { &self, root: usize, steps: &[i64], - store: &SummaryStore, + store: &SummaryStore, watermark_ms: Option, raw: &BTreeMap>, ) -> Result>, Unanswered> { @@ -500,56 +502,19 @@ fn slice(runs: &[&[RawSample]], start_ms: i64, end_ms: i64) -> Vec { samples } -/// `samples` as the executor's samples, for a `RawSampleSelector`. -fn samples_of(samples: &[RawSample]) -> Vec { - samples.iter().map(RawSample::to_sample).collect() -} - -/// A stored or kept pane bound as a source of a compiled root. The pane is -/// shared, not copied into the source; the executor's sources emit owned -/// batches, so each run takes one copy of its rows. -struct PaneSource(Arc); - -impl PhysicalOperator for PaneSource { - fn name(&self) -> &str { - "StoredPane" - } - - fn properties(&self, _inputs: &[PlanProperties]) -> PlanProperties { - PlanProperties { - boundedness: Boundedness::Bounded, - emission: Emission::Incremental, - } - } - - fn input_schemas(&self) -> Vec { - Vec::new() - } - - fn output_schema(&self) -> SchemaRef { - self.0.schema().clone() - } - - fn start<'a>( - &'a self, - _inputs: Vec>, - _context: RunContext, - ) -> Result, asap_executor::Error> { - let batch = (*self.0).clone(); - Ok(futures::stream::once(async move { Ok(batch) }).boxed_local()) - } - - fn output_bytes(&self, value: &Batch) -> usize { - value.bytes() - } +/// A stored or kept pane bound as a source of a compiled root. The batches +/// share the pane's buffers, and the executor treats them as held outside +/// the run, so the run is not charged for their states. +fn pane_source(pane: &Pane, schema: &SchemaRef) -> Result { + Operator::source(schema.clone(), pane.clone()) } /// The stored pane `input` reads at `time_ms`. fn pane( input: &StoredInput, time_ms: i64, - store: &SummaryStore, -) -> Result, Unanswered> { + store: &SummaryStore, +) -> Result, Unanswered> { let end = time_ms - input.offset_ms; if end.rem_euclid(input.pane_ms) != 0 { return Err(Unanswered::MissingPane(format!( @@ -592,6 +557,7 @@ fn program( if let Some((definition, binding)) = sds.outputs.get(&id) { 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, @@ -602,6 +568,7 @@ fn program( kept_builds[definition].as_ref().map_err(Clone::clone)?; kept.push(KeptInput { node: id, + schema: Arc::new(nodes[&id].output_schema.clone()), definition: definition.clone(), offset_ms: binding.offset_ms, raw_input: binding.raw_input, @@ -609,11 +576,11 @@ fn program( continue; } if let Some(window) = windows.get(&id) { - let selector = RawSampleSelector::new(dag, window).map_err(|e| e.to_string())?; + let series = SeriesWindow::new(dag, window).map_err(|e| e.to_string())?; raw.push(RawInput { node: id, window: (*window).clone(), - selector, + series, }); continue; } @@ -646,9 +613,11 @@ fn program( }) } -/// 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 @@ -660,7 +629,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 its first +//! sample, and its text cells (its series identity and label columns) are +//! computed and cached by label set. 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 same columns the selector +//! builds, without making 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::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 asap_planner::PhysicalASAPDAG; + +use crate::precompute::RawSample; + +/// The `Scan` schema's column that holds a series' full identity. +const SERIES_IDENTITY_COLUMN: &str = "$promql_series_identity"; + +pub(crate) type Labels = Arc>; + +/// 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>>, +} + +/// 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, + /// Each series' cells, or `None` when the `Scan` does not select it. + 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().expect("series lock").get(&sample.labels) { + return cells.clone(); + } + let cells = self.series_cells(sample); + self.series + .write() + .expect("series lock") + .insert(sample.labels.clone(), cells.clone()); + cells + } + + fn series_cells(&self, sample: &RawSample) -> Option> { + let one = Sample { + labels: (*sample.labels).clone(), + timestamp_ms: sample.timestamp_ms, + value: sample.value, + }; + // The selector applies the Scan's matchers to the series' labels. + 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`: 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. + 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<&Labels, &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; + } + } + for sample in latest.into_values() { + 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<(&Labels, Option>)> = None; + for sample in in_window { + let cells = match ¤t { + Some((labels, cells)) if Arc::ptr_eq(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/answers.rs b/data_plane/tests/answers.rs index 4b21f6d8..cdbbaf16 100644 --- a/data_plane/tests/answers.rs +++ b/data_plane/tests/answers.rs @@ -364,7 +364,7 @@ async fn answers_planned_range_queries() { #[tokio::test] async fn forwards_whole_range_queries_a_step_cannot_answer() { let (_, app) = example4b(101 * MINUTE).await; - for (start, step, reason) in [("5400", "5m", "pane grid"), ("4800", "10m", "not stored")] { + for (start, step, reason) in [("5400", "5m", "pane grid"), ("4200", "10m", "not stored")] { let params = [ ("query", P99), ("start", start), diff --git a/data_plane/tests/per_sample.rs b/data_plane/tests/per_sample.rs index d15a3098..d581ba27 100644 --- a/data_plane/tests/per_sample.rs +++ b/data_plane/tests/per_sample.rs @@ -1,6 +1,6 @@ //! Ingestion updates each open pane's summary state per sample, through the -//! executor's own updater; the panes it stores equal the executor's batch -//! build of the same samples. +//! 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; @@ -61,10 +61,13 @@ fn state_of(value: &Value) -> Vec { .collect() } -/// Every cell of `pane` as comparable data, rows in order. -fn cells(pane: &Batch) -> Vec>> { - pane.rows() +/// Every cell of `pane` as comparable data, its rows sorted: a pane built +/// on several workers holds one batch per worker. (Tests read rows to +/// compare them; the data plane does not.) +fn cells(pane: &[Batch]) -> Vec>> { + let mut rows: Vec>> = pane .iter() + .flat_map(Batch::to_rows) .map(|row| { row.iter() .map(|value| match value { @@ -73,7 +76,9 @@ fn cells(pane: &Batch) -> Vec>> { }) .collect() }) - .collect() + .collect(); + rows.sort(); + rows } /// For Example 1's counter rates and sums over 10 s panes, Example 3b's @@ -110,9 +115,11 @@ fn per_sample_panes_equal_the_batch_build() { let actual = per_sample.store().get(&definition.id, end); assert_eq!(actual.is_some(), expected.is_some(), "{example} pane {end}"); if let (Some(actual), Some(expected)) = (actual, expected) { - assert_eq!(actual.schema(), expected.schema()); + for batch in actual.iter().chain(expected.iter()) { + assert_eq!(**batch.schema(), definition.output_schema); + } assert_eq!(cells(&actual), cells(&expected), "{example} pane {end}"); - compared += usize::from(!expected.rows().is_empty()); + compared += usize::from(expected.iter().any(|b| b.num_rows() > 0)); } end += definition.pane_ms; } @@ -144,6 +151,6 @@ fn per_sample_panes_apply_label_matchers() { per_sample.store().get(&id, 20 * MINUTE).unwrap(), batch.store().get(&id, 20 * MINUTE).unwrap(), ); - assert_eq!(expected.rows().len(), 2, "jobs 1 and 4 are instance 1"); + 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/precompute.rs b/data_plane/tests/precompute.rs index 035b10ea..3baa549d 100644 --- a/data_plane/tests/precompute.rs +++ b/data_plane/tests/precompute.rs @@ -52,7 +52,11 @@ fn builds_panes_after_the_first_as_the_watermark_passes() { ); for end in [2 * PANE, 3 * PANE] { let pane = engine.store().get(&id, end).expect("built pane"); - assert_eq!(pane.rows().len(), 2, "one state per series"); + assert_eq!( + pane.iter().map(|b| b.num_rows()).sum::(), + 2, + "one state per series" + ); } assert_eq!( engine.store().get(&id, 4 * PANE).map(|_| ()), @@ -87,9 +91,36 @@ fn stores_empty_panes() { let id = sds.definitions.keys().next().unwrap().clone(); engine.ingest(&samples("latency_ms", &["a"], 0, 20 * MINUTE)); engine.ingest(&[sample("latency_ms", "a", 45 * MINUTE, 1.0)]); - assert_eq!(engine.store().get(&id, 2 * PANE).unwrap().rows().len(), 1); - assert_eq!(engine.store().get(&id, 3 * PANE).unwrap().rows().len(), 0); - assert_eq!(engine.store().get(&id, 4 * PANE).unwrap().rows().len(), 0); + assert_eq!( + engine + .store() + .get(&id, 2 * PANE) + .unwrap() + .iter() + .map(|b| b.num_rows()) + .sum::(), + 1 + ); + assert_eq!( + engine + .store() + .get(&id, 3 * PANE) + .unwrap() + .iter() + .map(|b| b.num_rows()) + .sum::(), + 0 + ); + assert_eq!( + engine + .store() + .get(&id, 4 * PANE) + .unwrap() + .iter() + .map(|b| b.num_rows()) + .sum::(), + 0 + ); } /// The plan's label matchers select which series a pane summarizes. @@ -104,7 +135,16 @@ fn applies_label_matchers() { let (mut engine, sds) = precompute(plan_workload(&workload, &common::inputs(false)).unwrap()); let id = sds.definitions.keys().next().unwrap().clone(); engine.ingest(&samples("latency_ms", &["a", "b"], 0, 25 * MINUTE)); - assert_eq!(engine.store().get(&id, 2 * PANE).unwrap().rows().len(), 1); + assert_eq!( + engine + .store() + .get(&id, 2 * PANE) + .unwrap() + .iter() + .map(|b| b.num_rows()) + .sum::(), + 1 + ); } /// Panes are `(end - width, end]`, like PromQL ranges: a sample at a @@ -119,10 +159,11 @@ fn panes_are_left_open() { assert_eq!(pane_end(-PANE, PANE), -PANE); } -/// A pane stays stored while an output reads it: Example 4b's oldest output -/// reads the pane ending 50 min before the evaluation, so with the watermark -/// at 90 min the pane ending at 40 min is kept and the one at 30 min, which -/// starts before the plan's 60-minute retention, is evicted. +/// A pane stays stored while an output reads it, plus one pane width for an +/// evaluation behind the newest pane: Example 4b's oldest output reads the +/// pane ending 50 min before the evaluation, so with the newest pane ending +/// at 90 min the panes ending at 40 and 30 min are kept and the one at 20 +/// min, before the plan's 60-minute retention and the margin, is evicted. #[test] fn keeps_panes_for_the_plans_retention() { let (mut engine, sds) = precompute(common::plan("example4b", false)); @@ -134,9 +175,10 @@ fn keeps_panes_for_the_plans_retention() { engine.store().get(&id, 4 * PANE).is_some(), "read at offset 50 min" ); + assert!(engine.store().get(&id, 3 * PANE).is_some(), "the margin"); assert!( - engine.store().get(&id, 3 * PANE).is_none(), - "older than the retention" + engine.store().get(&id, 2 * PANE).is_none(), + "older than the retention and the margin" ); } @@ -179,7 +221,7 @@ fn a_pane_closes_once_every_shard_has_passed_it() { .get(&id, 2 * PANE) .expect("both shards passed 20 min"); assert_eq!( - pane.rows().len(), + pane.iter().map(|b| b.num_rows()).sum::(), 2, "one state per series, from both shards" ); @@ -231,3 +273,30 @@ fn sharded_panes_answer_like_one_shard() { } } } + +/// An evaluation at T runs once samples just after T have arrived, so the +/// newest stored pane can be past T. Example 1 without raw data stores +/// 10-second panes for 60 s: with samples every 15 s through 10:15, an +/// evaluation at 10:00 still finds the pane ending at 9:10, which it reads +/// at offset 50 s, though the newest pane ends at 10:10. +#[test] +fn keeps_the_panes_of_an_evaluation_behind_the_newest_pane() { + let (mut engine, sds) = precompute(common::plan("example1", false)); + let all = samples( + "http_requests_total", + &["a", "b"], + MINUTE, + 10 * MINUTE + 15_001, + ); + engine.ingest(&all); + for definition in sds.definitions.values() { + assert_eq!(definition.pane_ms, 10_000); + assert!( + engine + .store() + .get(&definition.id, 9 * MINUTE + 10_000) + .is_some(), + "read by the evaluation at 10:00" + ); + } +} diff --git a/data_plane/tests/range.rs b/data_plane/tests/range.rs index ca1829f2..a0443edb 100644 --- a/data_plane/tests/range.rs +++ b/data_plane/tests/range.rs @@ -7,9 +7,8 @@ mod common; use std::collections::BTreeMap; use std::sync::Arc; -use asap_executor::values::Batch; use asap_planner::InstallablePlan; -use data_plane::precompute::{Precompute, RawSample}; +use data_plane::precompute::{Pane, Precompute, RawSample}; use data_plane::query::{KeptResultStats, QueryEngine, RawRange, ResultSample, Unanswered}; use data_plane::sds::{Sds, SummaryStore}; @@ -80,7 +79,7 @@ fn instant( engine: &mut QueryEngine, query: &str, time_ms: i64, - store: &SummaryStore, + store: &SummaryStore, all: &[RawSample], ) -> Vec { let (root, ranges) = engine.raw_ranges(query, time_ms, store).unwrap(); @@ -97,7 +96,7 @@ fn range( engine: &mut QueryEngine, query: &str, steps: &[i64], - store: &SummaryStore, + store: &SummaryStore, all: &[RawSample], ) -> (Vec>, Vec) { let watermark_ms = all.iter().map(|s| s.timestamp_ms).max(); @@ -110,7 +109,7 @@ fn range_at_watermark( engine: &mut QueryEngine, query: &str, steps: &[i64], - store: &SummaryStore, + store: &SummaryStore, watermark_ms: Option, all: &[RawSample], ) -> (Vec>, Vec) { @@ -218,10 +217,10 @@ fn every_step_reads_stored_panes() { }; // A step 5 minutes apart lands off the 10-minute grid. assert!(missing(&[90 * MINUTE, 95 * MINUTE]).contains("pane grid")); - // 80 min reads the pane ending at 30 min, already evicted: the whole + // 70 min reads the pane ending at 20 min, already evicted: the whole // range is forwarded even though its later steps could be answered. assert!(missing( - &minutes(8..=10) + &minutes(7..=10) .into_iter() .map(|t| t * 10) .collect::>()