[Refactor] Read raw data and materialized views through one DataStore (read_raw_data, read_materialized_view), not per-run Source/NodeId → TableProvider maps #995

Design: docs/design_docs/proposals/data-store.md (PR #996). It covers the interface and the store contract; how a store implements it is the deployment's design.

Example

KLL p50/p99 of latency over five 1-minute panes kept as materialized views (the running example of docs/design_docs/physical-planning-and-deployment.md). The installed DAG has the chain expanded into five pane nodes, pane_0…pane_4, each one a materialized view with the same coverage.definition_id. Each pane's refresh DAG reads raw data through Scan(Source::TimeSeries { metric: "latency" }).

Today, before every run, a deployment builds two routing maps (crates/executor/src/sources/mod.rs):

let mut tables = DataSources::default();
tables.register(Source::TimeSeries { metric: "latency".into() }, Arc::new(SampleBuffer::new(..)))?;
let mut materialized_views = BTreeMap::new();
for view in plan.materialized_views()? {                  // node_id, coverage, refresh, retention
    let states = sds.states_for(&view.coverage);          // the deployment's summary store
    materialized_views.insert(view.node_id, Arc::new(StoreProvider::new(states)) as Arc<dyn TableProvider>);
}
query.instantiate(&BindingDataSources { tables, materialized_views })?;

The deployment's real questions are "read raw data of table S" and "read the stored states of definition D for time range R".

What it should be: one store, with one method per kind of input, named after what it reads:

pub trait DataStore {
    /// Rows of the raw table or time series `source` names; a hint, the executor re-filters.
    fn read_raw_data(&self, source: &Source, request: &RawDataRequest, context: RunContext)
        -> Result<OutputStream<'_, Batch>, Error>;
    /// The stored states of `view` whose pane ends in `request.time_range_ms` (`(start, end]`); exact.
    fn read_materialized_view(&self, view: &MaterializedView, request: &MaterializedViewRequest, context: RunContext)
        -> Result<OutputStream<'_, Batch>, Error>;
}
// RawDataRequest { time_range_ms: Option<(i64, i64)>, schema, predicates }   (hint)
// MaterializedViewRequest { time_range_ms: (i64, i64), schema }              (exact)
// CompiledDAG::instantiate(&self, store: Arc<dyn DataStore>)

The deployment passes one handle per run, and it presents one consistent snapshot to every read of that run. How it keeps the snapshot is the deployment's choice.

What crosses the interface: SDS definitions, never instances

Concrete maintenance problem

• Three names for raw data. Source names the table, DataSources maps Source → TableProvider, and TableProvider reads it. A deployment registers one provider per table, although its storage already dispatches by name.

• The deployment must know node ids. It does not otherwise need them: the summary store is keyed by definition_id and coverage.

• The trait doesn't fit a view. A materialized view is bound through TableProvider, but:

  • its rows are not re-filtered;
  • projection and predicates are always empty for it;
  • its states are not charged to the run's memory.

  Implementing TableProvider invites a provider to ignore the time range, which the doc then has to forbid in prose.

• Both maps are rebuilt every run, although they are fixed per installed plan.

• Scope carries a revision the executor never reads. It is a store snapshot concept; the run's handle keeps the snapshot instead.

Behavior that must remain unchanged

• Every query returns the same answers.
• Raw data is still re-filtered by the executor (time range, predicates); the request stays a hint.
• A pane view requests the same time range as today (materialized_view_window, physical_planner/mod.rs). A view that requests None today now requests (T - lookback, T], where lookback is the longest lookback over its raw-data windows and the views it reads.
• A view's states are still held outside the run's memory limit.

Before and after structure

• Interface: DataStore with read_raw_data and read_materialized_view, in crates/executor/src/sources/data_store.rs. RawDataRequest { time_range_ms, schema, predicates } replaces ScanRequest, and MaterializedViewRequest { time_range_ms, schema } is new; the store returns the slot's columns, and the executor checks each batch.
• Removed: TableProvider, ScanRequest, DataSources, BindingDataSources, and Scope's revision.
• Compile time: compile records, for each materialized view slot, the MaterializedView it stands for. instantiate(store) wires every slot to the store.
• MaterializedView stays in asap-types (crates/types/src/ir/materialized_view.rs), so the executor can name it without depending on the planner.
• test-support: InMemoryDataStore (batches by Source and by view) replaces the hand-written providers in tests.
• Open point: boundaries that are not views. precompute::compile composes fragments whose Scan frontiers are bound by node id, and tests cut at a TimeRange (deployment_computation, tumbling_pane_merge). Either keep a crate-internal / test-support node-id binding for those, or give each such boundary a MaterializedView. The PR decides and documents which.
• The design doc (§4, "Binding data sources") and [Tracking] Physical DAG → deployment boundary: InstallablePlan, input slots, TableProvider, BindingDataSources #968's decision Q1 are updated.

Why the proposed structure is minimally complex

One trait with two named methods replaces two per-run maps, a catalog type, a provider trait and a binding struct. No new identity is invented: Source and MaterializedView already exist and are what the plan names.

Existing unit and end-to-end protection

• crates/executor/tests/{stored_pane_memory_limit,deployment_computation,precompute_candidates,raw_scan}.rs.
• crates/planner/tests/runtime_install.rs.
• crates/integration-tests/tests/{materialized_views,materialization_e2e,tumbling_pane_merge,kll_pane_execution}.rs.
• Add a test: a store keyed only by Source, and by definition_id and time range, answers the five-pane KLL query, and the result equals today's binding.

Compatibility or migration impact

This is a public API change in asap_executor: instantiate takes Arc<dyn DataStore>, and the binding types above are removed. ASAPQuery-backend main does not depend on asap_executor yet. Its integration implements DataStore over its raw data storage and SDS directly.

Parent: #968. Raised in discussion: the output of a precompute DAG and the input of a query DAG are a write and a read to the store for a definition id and its metadata, so a deployment should not need a notion of input slot.