[Refactor] TableProvider (RawSource renamed) with a ScanRequest, and one BindingDataSources for table slots and materialized-view slots #966

Example

SELECT count(*) FROM requests WHERE status = 500 AND ts >= now() - INTERVAL '5 minutes', where Stage 3 keeps 1-minute panes as a materialized view.

The DAG is Scan(Source::Table { table_ref: "requests" }, predicate status = 500) under TimeRange(Range, 5m), and the pane node under it is materialized.

Today a deployment binds two kinds of data through two interfaces:

After this issue, both are a TableProvider:

• requests gets scan(ScanRequest { time_range_ms: Some((T - 5m, T]), projection: Some(&[status, ts]), predicates: &[status = 500] }).
• The pane view gets scan(ScanRequest { time_range_ms: Some((T - 5m, T]), .. }) and returns the five pane states.

Concrete maintenance problem

• RawSource gives a provider no way to skip rows or columns. That is the reason the raw-sample selectors exist (removed in the next PR of [Tracking] Physical DAG → deployment boundary: InstallablePlan, input slots, TableProvider, BindingDataSources #968).
• Requiring the provider schema to equal the Scan's rules out a real table with more columns than the query reads.
• Materialized views have no interface of their own: the deployment hands the executor an operator. So the deployment's summary store cannot share code or contract with raw-data providers, and the executor cannot tell it which panes a run reads.

Behavior that must remain unchanged

• Every query returns the same answers.
• Compilation opens no readers.
• A connector error is never an empty successful scan.
• A provider that ignores every ScanRequest field stays correct.

Before and after structure

pub trait TableProvider {             // RawSource, renamed
    fn schema(&self) -> SchemaRef;     // contains the slot's columns; may have more
    fn boundedness(&self) -> Boundedness;
    fn scan(&self, request: &ScanRequest, context: RunContext)
        -> Result<OutputStream<'_, Batch>, Error>;
}
pub struct ScanRequest<'a> {
    pub time_range_ms: Option<(i64, i64)>, // (start, end]
    pub projection: Option<&'a [usize]>,  // indices into the provider's schema
    pub predicates: &'a [ScalarExpr],     // over the provider's schema; a hint only
}
pub struct BindingDataSources {
    pub tables: DataSources,                             // Source → provider
    pub views: BTreeMap<NodeId, Arc<dyn TableProvider>>, // view slot → provider
}
impl CompiledPhysicalDAG {
    pub fn instantiate<'a>(&self, data_sources: &'a BindingDataSources) -> Result<PhysicalDAG<'a, Batch, SchemaRef>, Error>;
}

• Table slot. Every Scan leaf compiles to a table slot holding its Source, schema, compiled predicates and window, which is the offsets and widths of the TimeRange/TimeShifts above it. This is computed once, as table_time_operator (physical_planner/mod.rs:725) does today for SQL tables. instantiate turns the slot and its provider into a Scan. At run time the Scan takes T from RunContext's Scope, builds the ScanRequest, projects the provider's columns to the slot's schema, and re-applies the time range and the predicates. A Scan read by several windows requests their union.
• View slot. instantiate wraps the view's provider in a schema-checked source. The request carries the view's time range for this run: the panes the query reads.
• Pushdown is a hint. There is no supports_filters_pushdown and no Exact mode, so the executor always filters (decision Q4 of [Tracking] Physical DAG → deployment boundary: InstallablePlan, input slots, TableProvider, BindingDataSources #968).
• bind and bind_with_data_sources are replaced by instantiate(&BindingDataSources).

Why the proposed structure is minimally complex

The trait is renamed and gains one argument. The two kinds of bound data share one trait. No pushdown negotiation is added, and the correctness of a scan never depends on the provider honoring the request.

Existing unit and end-to-end protection

• crates/executor/tests/{raw_scan,current_timestamp,deployment_computation,precompute_candidates}.rs.
• crates/integration-tests/tests/{sql_to_physical,filtered_aggregates,sketch_rollup,operator_design_examples}.rs.
• crates/planner/tests/{sql_table_scans,sql_per_group_topk}.rs.
• Add tests:
  • a provider whose schema has an extra column;
  • a provider that ignores the request and returns rows outside the range and the predicates (same answer);
  • a provider that applies the request (same answer, fewer rows read);
  • a view provider asked for exactly the panes of (T - 5m, T].

Compatibility or migration impact

This is a public API change in asap_executor:

• RawSource is renamed TableProvider, and its scan takes a ScanRequest;
• bind and bind_with_data_sources are replaced by instantiate(&BindingDataSources).

ASAPQuery-backend main pins an older Planner revision (f46cbf6c) and does not depend on asap_executor yet, so no backend code breaks. The backend's executor integration implements TableProvider from the start. A deployment that also links DataFusion names ours by path (asap_executor::sources::TableProvider).

Parent: #968 (PR 4 of 6; after #965). The next PR moves PromQL time-series windows onto this interface and deletes the raw-sample selectors.