[Tracking] Physical DAG → deployment boundary: InstallablePlan, input slots, TableProvider, BindingDataSources #968

Example

rate(http_requests_total{job="api"}[5m]), where Stage 3 stores 1-minute panes at ingestion time.

Today a deployment runtime that installs this plan has to:

1. read InstallablePlan.stored_outputs and query_time_raw_windows. These are fields the Planner derives from the DAG, and InstallablePlan::validate recomputes them and compares them;
2. call RawDataSampleSelector::new(dag, window) and window_batch(end, &[RawDataSample]). These apply the window, the label matchers, instant lookback and stale markers outside the DAG, and the runtime binds the result at the TimeRange node;
3. build an InputContract for every boundary from the node's output_schema, and bind stored state as a Source at the pane node.

A SQL table with the same window goes through a different path (RawSource + Scan, or TableRowSelector). The executor's public API also carries legacy types (Statistic, KeyByLabelValues, Measurement).

After this tracking issue, the runtime:

1. reads InstallablePlan { dag, query_texts }, and calls materialized_views() to learn which views to maintain;
2. compiles each view's refresh and the query run once (compile(dag, views: &[MaterializedView], roots)), with no readers;
3. implements one DataStore over its raw data and its summary store: read_raw_data(source, RawDataRequest) for the http_requests_total time series or a SQL table, and read_materialized_view(view, MaterializedViewRequest) for the panes;
4. calls instantiate(Arc<dyn DataStore>) with one store handle per run. The DAG's own Scan and TimeRange apply the window and predicates for PromQL and SQL alike.

Layers and ownership

Concepts

InstallablePlan holds only what the DAG cannot derive:

pub struct InstallablePlan {
    pub wire_version: u32,        // versions the whole plan file
    pub dag: PhysicalASAPDAG,     // a materialized node carries its retention_ms
    pub query_texts: Vec<String>,     // query i is answered by dag.roots[i]
}
impl InstallablePlan {
    pub fn materialized_views(&self) -> Vec<MaterializedView>; // node, refresh, coverage, inputs, retention
}

Input slot. A position in a compiled DAG where data is read at run time, and that position's schema, which is the DAG node's output_schema. There are two kinds, and no other: the executor cannot cut a DAG at an arbitrary node.

• raw data slot: a Scan leaf. It keeps the Scan's Source, schema, predicates and window (the union of its readers' windows, scan_window), and is read by Source.
• materialized view slot: a materialized view (a node whose output is kept across evaluations) that a DAG reads. It keeps the MaterializedView and the window its state summarizes (PhysicalASAPDAG::view_window: the union of the windows of every TimeRange below it), and is read by the MaterializedView. The same node is the root of the DAG that refreshes it.

pub fn compile(dag: &PhysicalASAPDAG, views: &[MaterializedView], roots: &[NodeId])
    -> Result<CompiledPhysicalDAG, Error>; // opens no readers

CompiledPhysicalDAG is derived inside the executor. It is not persisted and has no serde.

DataStore is the only data interface between the executor and a deployment. Its method names say what is read, because the two contracts differ:

pub trait DataStore {
    fn read_raw_data(&self, source: &Source, request: &RawDataRequest, context: RunContext)
        -> Result<OutputStream<'_, Batch>, Error>;
    fn read_materialized_view(&self, view: &MaterializedView, request: &MaterializedViewRequest, context: RunContext)
        -> Result<OutputStream<'_, Batch>, Error>;
}
pub struct RawDataRequest<'a> {
    pub time_range_ms: Option<(i64, i64)>, // (start, end], from the slot's window and the run's time
    pub schema: &'a SchemaRef,             // the slot's columns; the store returns exactly these
    pub predicates: &'a [ScalarExpr<()>],  // a hint: the executor re-applies every predicate
}
pub struct MaterializedViewRequest<'a> {
    pub time_range_ms: (i64, i64),         // (start, end], the interval the view's state summarizes in this run
    pub schema: &'a SchemaRef,             // group columns and one state column
}

A time series is a table (ts, value, *labels), as Source::TimeSeries declares. A view read returns exactly the states whose intervals tile the requested range: never an observation twice, never another definition's state, never missing state as empty (an error instead). Only SDS definitions cross the interface (the view's coverage and schema); instances stay inside the store.

Instantiation and runs.

impl CompiledPhysicalDAG {
    pub fn instantiate(&self, store: Arc<dyn DataStore>) -> Result<PhysicalDAG<'static, Batch, SchemaRef>, Error>;
}

A deployment passes one store handle per run, which presents one consistent state to every read of that run; how it keeps the snapshot is the deployment's choice. RunContext/Scope carry only what the executor uses (scope, limits, cancellation, query range).

Operators own semantics; the store only reads.

• A raw data slot becomes a Scan over the store. The Scan computes time_range_ms from its window and the run's Scope, passes the request, checks each batch against the slot's schema, and re-applies the time range and the predicates.
• A materialized view slot asks for the view's window: (T - o - w, T - o] in a query run, (window_end - w, window_end] in an ingestion run.
• A time-series TimeRange is a native operator. A range window filters ts. An instant window keeps each series' latest row in the lookback, and a series whose latest sample is a stale marker has no row.
• A Scan shared by several windows requests the union of their ranges, and each TimeRange filters its own.

Decisions (2026-10-27)

Also decided:

• RawSource is renamed TableProvider; TableProvider, ScanRequest, DataSources, BindingDataSources and MemorySource are then replaced by DataStore and InMemoryDataStore ([Refactor] Read raw data and materialized views through one DataStore (read_raw_data, read_materialized_view), not per-run Source/NodeId → TableProvider maps #995).
• A node whose output is kept across evaluations (MatAtIngestion or MatAtQueryKeptAcrossEvaluations) is a materialized view. StoredOutput becomes MaterializedView, StoredOutputBuild becomes its refresh mode, and a stored slot is a materialized view slot. A MatAtQuery node is not a view, because its output is discarded after one evaluation. The names follow DataFusion's materialized-views extension (datafusion-contrib/datafusion-materialized-views). There, a view is a TableProvider that also implements Materialized { query(), rewrite_readiness() }, queries read it as a table, and mv_dependencies gives its build graph. Ours map as follows: query() is the sub-DAG under the node, and coverage.definition_id identifies it; readiness is the deployment's (the summary store's status); the build graph is MaterializedView.inputs.
• InstallablePlan keeps its name. DataFusion's ExecutionPlan corresponds to our PhysicalOperator, not to InstallablePlan.
• The summary store (SDS) stays in ASAPQuery-backend. The Planner and the executor never see storage identity.

PRs, in order

Each sub-issue is one PR.

1. [Docs] Physical DAG → deployment interface in the physical planning design doc #971: design doc: docs/design_docs/physical-planning-and-deployment.md §4–§5 describe these concepts.
2. [Refactor] Input slots take their schema from the physical DAG node; remove InputContract #965: an input slot is a position and the node's schema; remove InputContract; CompiledPhysicalDAG drops serde.
3. [Refactor] Remove legacy public types from asap_executor (Statistic, KeyByLabelValues, Measurement) and the uncalled query_time module #967: remove legacy public types (Statistic, KeyByLabelValues, Measurement) and the uncalled query_time module. Independent of the others.
4. [Refactor] TableProvider (RawSource renamed) with a ScanRequest, and one BindingDataSources for table slots and materialized-view slots #966: TableProvider (renamed RawSource) with ScanRequest; table and materialized view slots; BindingDataSources; materialized views through TableProvider. After [Refactor] Input slots take their schema from the physical DAG node; remove InputContract #965.
5. [Refactor] Time-series TimeRange as a native operator; bind raw data only at Scan; remove the raw-sample selectors #969: time-series TimeRange as a native operator; raw data bound only at Scan; remove the raw-sample selectors. After [Refactor] TableProvider (RawSource renamed) with a ScanRequest, and one BindingDataSources for table slots and materialized-view slots #966.
6. [Refactor] InstallablePlan holds only a versioned DAG and query texts; remove PhysicalASAPDAGDocument; StoredOutput becomes MaterializedView #970: InstallablePlan holds only a versioned DAG and the query texts (PhysicalASAPDAGDocument removed); StoredOutput is renamed MaterializedView; retention on the node. After [Refactor] Time-series TimeRange as a native operator; bind raw data only at Scan; remove the raw-sample selectors #969.
7. [Refactor] Read raw data and materialized views through one DataStore (read_raw_data, read_materialized_view), not per-run Source/NodeId → TableProvider maps #995: one DataStore (read_raw_data, read_materialized_view) replaces TableProvider/BindingDataSources; compile takes MaterializedViews; Scope drops revision; node-id cut paths removed. After [Refactor] InstallablePlan holds only a versioned DAG and query texts; remove PhysicalASAPDAGDocument; StoredOutput becomes MaterializedView #970.

Status (2026-10-28)

All sub-issues are done; this issue is closed. The system architecture doc (#1000) is reviewed separately. Follow-ups, tracked outside it:

• per-root query-time Scan retention (today's value can over-report);
• Stage 3 does not cost range spans, and costs a kept pane chain at its newest pane only (Retained raw data as concrete state groups per lookback: offsets, @ and a retention limit #955);
• memo costing (direct_profile, proposals, lower_bounds) does not see TimeShift offsets;
• open point: group-column predicates as a hint on a view read, once decode cost is measured.

Out of scope

• How ASAPQuery-backend implements its providers and summary store.
• Filter pushdown beyond hints (Q4), and limit pushdown.