docs: refresh the ingestion-time views of one refresh window in one run (#1151)

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

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.