diff --git a/crates/types/src/ir/materialized_view.rs b/crates/types/src/ir/materialized_view.rs index 7d83bea19..27e99090b 100644 --- a/crates/types/src/ir/materialized_view.rs +++ b/crates/types/src/ir/materialized_view.rs @@ -18,10 +18,13 @@ //! before it. A runtime refreshes a view once all of its inputs are //! complete: each raw-data window's samples have arrived (sealed) and each //! view it reads is refreshed. It compiles the DAG with the executor for -//! that view's node, with each view it reads as a materialized view slot -//! and raw data read by `Source` at each `Scan`. It refreshes the view in an -//! ingestion run ([`MaterializedViewRefresh::AtIngestion`]) or in the query -//! run of the evaluation that first reads it +//! that view's node (for a view refreshed at ingestion, together with the +//! rest of its refresh group, below), with each view it reads that the run +//! does not refresh as a materialized view slot and raw data read by +//! `Source` at each `Scan`. It +//! refreshes the view in an ingestion run +//! ([`MaterializedViewRefresh::AtIngestion`]) or in the query run of the +//! evaluation that first reads it //! ([`MaterializedViewRefresh::OnFirstRead`]), keeps the result, and binds //! it the same way wherever the DAG reads that node. An ingestion run over //! `(window_start_ms, window_end_ms]` refreshes the windows ending at @@ -35,6 +38,21 @@ //! A query run binds the views it reaches from the roots without passing //! another view; a view only other views read is read through them. //! +//! ## Refresh groups +//! +//! Views refreshed in an ingestion run are refreshed by **refresh group**: +//! the views whose windows have one width `W` and end at one offset `p` +//! modulo `W` (their **refresh window**; a view with an unbounded window is +//! a group of its own). A runtime refreshes a group with one ingestion run +//! per window `(e - W, e]`, `e ≡ p (mod W)`. The run's roots are the views of +//! the group whose state for the window ending at `e` is not stored yet: in +//! steady state the newest pane of each chain, since older panes are the +//! same states, stored by earlier runs. A view of the group that a root +//! reads is computed in the run; views of other groups are materialized +//! view slots. A node several roots share runs once per run, so a `Scan` +//! below a KLL pane chain and a DDSketch pane chain of equal pane width +//! reads each pane's raw data once. +//! //! Each view's `coverage` says what its stored state is: a stable id of its //! definition (what it computes) and its selection (which rows went in, //! with time relative to the evaluation time). Two views with the same diff --git a/docs/design_docs/physical-planning-and-deployment.md b/docs/design_docs/physical-planning-and-deployment.md index 00c574e7a..04391939f 100644 --- a/docs/design_docs/physical-planning-and-deployment.md +++ b/docs/design_docs/physical-planning-and-deployment.md @@ -288,8 +288,9 @@ window; that requires supported boundary handling or a different candidate. ## 3. Materialization → Physical DAG Operator names in the diagrams of this section and §4 (`NativeKllBuild`, -`NativeKllMerge`, `NativeP50`, `NativeQuantile`, `KllStateOutput`) are -illustrative. The executor implements them as `SummaryAgg` (a KLL build), +`NativeDDSketchBuild`, `NativeKllMerge`, `NativeP50`, `NativeQuantile`, +`KllStateOutput`) are illustrative. The executor implements them as +`SummaryAgg` (a KLL or DDSketch build), `SummaryMerge` and `SummaryEvaluation` (a quantile readout). The **Physical Plan Compiler** consumes both computation semantics and maintenance @@ -574,21 +575,65 @@ ingestion window, or a query evaluation time `T`. - In an ingestion run over `(window_start, window_end]`, each `TimeRange(w)` keeps `(window_end - w, window_end]`. A windowed raw data slot asks for `(window_start, window_end]`, and a materialized view slot for `(window_end - w, window_end]`. A view refreshed at ingestion whose raw windows, or the windows of the views it reads, end at different offsets cannot be built in one run, so installation refuses it. - A store never implements PromQL or SQL semantics. The compile check compiles these operators, so a plan whose window the executor cannot run is rejected before it is installed. +### Refresh groups + +Two views refreshed at ingestion can be built in one ingestion run, with the +run's interval equal to each one's window, exactly when they have the same +**refresh window**: windows of width `W` that end `p` after a multiple of `W` +(`0 ≤ p < W`), relative to the evaluation time. A view whose `view_window` is +`(T - o - W, T - o]` has the refresh window `(W, o mod W)`. All panes of a +pane chain have their newest pane's refresh window. The source, the summary +family and where the state is kept do not enter. A view with an unbounded +window is a group of its own. + +A **refresh group** is the set of views refreshed at ingestion that have one +refresh window `(W, p)`. A runtime refreshes a refresh group with one +ingestion run per window `(e - W, e]`, `e ≡ p (mod W)`: + +- The run's roots are the views of the group whose state for the window + ending at `e` is not stored yet. Pane `i` of a chain at evaluation `T` is + the same state as the newest pane at `T - i·W`, so in steady state the + roots are the newest pane of each chain of the group; the older panes are + already stored. +- A view of the group that a root reads is computed in the run, not read + from a slot. The views of other groups that the roots read are + materialized view slots. +- The executor runs a node that several roots share once per run, so a `Scan` + below several views of the group reads its raw data once per window. +- Views of different refresh windows are refreshed in different runs. With a + 10-second and a 60-second pane chain over one `Scan`, the 10-second chain + needs five pane ends the 60-second chain does not, so the `Scan` is read + once for each chain. + +The refresh window is derived from the DAG alone (time ranges, shifts and pane +widths), like the views and their retention. Sealing, lateness, threading and +the alignment of the window grid stay the deployment's. A runtime that +splits a group (for example, because its sources seal at different times) +computes the same states with more runs. + ### Running example For the KLL p50/p99 example, with one-minute panes materialized at ingestion -over `Source::TimeSeries { metric: "latency" }`: +over `Source::TimeSeries { metric: "latency" }`, plus a p99.9 with relative +error answered by a DDSketch chain with one-minute panes over the same `Scan`: ```text Install (once): - views = plan.materialized_views()? // [pane nodes, refresh = at ingestion, retention] - refresh = compile(dag, [], [pane_0]) // Scan → TimeRange(1m) → NativeKllBuild(k=200) - query = compile(dag, views[pane_0..pane_4], [p50, p99]) // 5 materialized view slots → NativeKllMerge → NativeP50, NativeP99 - -Each complete pane (12:04, 12:05], in an ingestion run: + views = plan.materialized_views()? // [KLL and DDSketch pane nodes, refresh = at ingestion, retention] + // both chains have refresh window (1m, 0): one refresh group of ten pane views + refresh = compile(dag, [], [kll_pane_0, dd_pane_0]) // the newest pane of each chain + // Scan → TimeRange(1m) → NativeKllBuild(k=200) + // → NativeDDSketchBuild(α=0.01) + query = compile(dag, views[kll_pane_0..kll_pane_4], [p50, p99]) // 5 materialized view slots → NativeKllMerge → NativeP50, NativeP99 + (the p99.9 query DAG reads dd_pane_0..dd_pane_4 the same way) + +Each complete pane (12:04, 12:05], in one ingestion run whose roots are the +group's views with no stored state for the pane ending 12:05 (kll_pane_0 and +dd_pane_0; the older panes were stored by earlier runs): refresh.instantiate(store) - Scan calls read_raw_data(TimeSeries { latency }, time_range_ms = (12:04, 12:05]) - store the KllState in the summary store under the view's coverage and interval + Scan calls read_raw_data(TimeSeries { latency }, time_range_ms = (12:04, 12:05]) once; + both builds read its rows + store the KllState and the DDSketchState in the summary store, each under its view's coverage and interval Query at T = 12:05 over (12:00, 12:05]: the installed DAG has the chain expanded: five pane views, pane i under TimeShift(i·1m) @@ -636,7 +681,7 @@ The complete example makes the ownership boundary explicit: | **Physical Plan Compiler** | Lower to native KLL build, merge, and readout operators | | **Physical DAG** | Define precompute and query DAGs with typed input/output boundaries | | **Deployment interface** | Read the `latency` raw data slot with `read_raw_data`, and each pane's materialized view slot with `read_materialized_view`, from one `DataStore` | -| **Deployment** | Schedule pane refreshes, store and resolve panes, run queries | +| **Deployment** | Schedule one ingestion run per refresh group and window, store and resolve panes, run queries | ```text Logical: @@ -662,7 +707,7 @@ does not introduce additional planning decisions. The physical layer does not own raw ingestion, storage formats, or decoding persisted bytes into typed state. It compiles computation over input slots: windows, summary build, merge (for example KLL merge), sketch estimates and -exact finalization. The deployment schedules each pane's refresh, stores and +exact finalization. The deployment schedules each refresh group's runs, stores and reads stored state, and serves raw data and views to the slots through its `DataStore`. The plan file is the only persisted handoff; a compiled physical DAG is derived from it by the executor and is not persisted.