Shared scans as candidates at ingestion and query time; charge each scan the plan runs #1179

Problem, by example

Ingestion time. Two summaries are maintained at ingestion time over one source m, with 6 series and 6 rows/s:

A: SummaryTumblingWindowChain of a KLL, 10000 ms tumbling windows   refresh window (10 s, 0)
B: SummaryTumblingWindowChain of a DDSketch, 60000 ms tumbling windows   refresh window (60 s, 0)
both read: Scan(m), NotMaterialized, IngestionTime

The Scan can run two ways:

• Separately: one ingestion run per refresh window, and each run scans its own window. m's rows are read twice.
• Shared: one scan of every arriving row, at the finer cadence, feeding both chains. B's 60 s window is filled across six 10 s runs. Rows are read once.

Today this choice is not a candidate. #1151 (#1152–#1154) hard-codes it in the cost function: asap_cost::execution_demand::computations charges a NotMaterialized ingestion-time node once per distinct refresh window of its MatAtIngestion consumers. The design doc says so (docs/design_docs/physical-planning-and-deployment.md, "Refresh groups"): "a stateless ingestion-time node is charged once per refresh window of the views above it". The #1095 group MILP (#1144, #1157) mirrors that rule with one ingestion context I_r per refresh window.

So:

• A and B always pay for two scans, and the plan cannot choose the shared scan.
• A runtime that refreshes per refresh window pays what was priced. A runtime that "updates states sample by sample passes each sample once whatever the windows, so it runs cheaper than priced" (same doc). Its real cost and the priced cost differ, and nothing in the plan says which runtime behaviour was assumed.
• The grouping rule lives in asap-cost. There is no node in the plan for it, so the number of scans the plan pays for is not visible in the selected DAG.

Query time. rate(m[1m]) and rate(m[5m]), both every 10 s, computed at query time over raw data:

• Separately: one scan of the last 1 min and one of the last 5 min, i.e. 6 + 30 = 36 s-equivalents of rows per 10 s.
• Shared: one scan of the last 5 min, the largest scope. The 1 min reader keeps its last minute with a TimeRange. That is 30 s-equivalents.

(Both figures are hand-counted for illustration, not measured.)

#951 proposes this as a Pass 2 rule for different predicates, labels and time ranges, at query time. It does not cover ingestion time, and it does not settle how a shared scan is charged.

What should happen instead:

• Candidates. Sharing a scan is a candidate the planner generates, at ingestion time and at query time: one scan with the largest scope (union of columns and time ranges, the OR of the predicates where Share one scan of a source across readers with different predicates, labels and time ranges (subsumption derivations) #951 allows it), with each reader compensating on top. Next to it stay the separate scans. Stage 3 chooses.
• Cost. A scan is charged for each execution the selected plan actually performs, with that execution's scope. No grouping rule is hidden in the cost function.

Why it happens

• Scan sharing is identity-only. Hash-consing shares a scan only when two readers need the identical scan (same source, columns, predicates and time range). scan_labels.rs widens labels for some per-entity builds, but there is no general shared-scan rule; that is Share one scan of a source across readers with different predicates, labels and time ranges (subsumption derivations) #951, open.
• At ingestion time the sharing is implicit. Ingestion-time NotMaterialized nodes have no context of their own. feat: charge a shared ingestion node once per refresh window (#1151) #1154 decides, in computations, that the views of one refresh window share one run and those of different windows do not. feat(plan-selection): one ingestion pipeline per refresh window in the selection MILP (#1095 §2.5 O3(b)) #1157 then builds I_r contexts so the MILP objective matches that rule. The choice "one scan for several refresh windows" therefore cannot be expressed, and is never priced.
• What already works:
  • Within one refresh window, one run computes a shared node once for all views (ingestion_refresh_groups). This is the "separately per window" option.
  • Within one query-time execution context, the [Feature] Stage 3: no execution-scope copies, exact candidate-level MILP, one cost formula #1095 MILP gives a group one candidate (O4), so a scan read twice in one context is charged once.

Proposal

1. A shared-scan rule generates candidates at both timings.

• Shared candidate. For readers of one source, the rule adds a scan with the covering scope:
  • columns and labels: the union;
  • time: the union of the readers' windows (for ingestion-time readers, the finest refresh cadence among them; for query-time readers, the longest window and the union of offsets);
  • predicates: per Share one scan of a source across readers with different predicates, labels and time ranges (subsumption derivations) #951.
• Compensation. Each reader gets a candidate reading the shared scan through its own compensation: a TimeRange or shift down to its window, Filter, Project.
• Separate candidates. The readers' own scans stay as candidates.
• Share one scan of a source across readers with different predicates, labels and time ranges (subsumption derivations) #951 becomes the query-time half of this rule. The ingestion-time half is new.

2. Ingestion time needs a capability and a runtime contract. A shared ingestion scan at a 10 s cadence feeding a 60 s chain needs the runtime to keep the 60 s chain's newest tumbling window open across six runs, and to add each run's rows to it. That is incremental update, not one refresh per window.

• The deployment declares whether it supports this, next to raw_data_retained and query_time_results_retained.
• Without that capability, only the separate (per refresh window) candidates exist, which is today's behaviour.

3. Cost: charge what runs.

• Each scan candidate the plan selects is charged per execution, for the rows of its scope.
  • A shared ingestion scan runs once per arriving row.
  • Separate per-window scans run once per refresh window that selects them.
  • A query-time scan runs once per execution of its context.
• Remove the refresh-window grouping from asap_cost::execution_demand::computations. Runs within one refresh window stay one execution (the existing ingestion_refresh_groups). A scan shared across refresh windows is an explicit candidate, not a discount.
• In the [Feature] Stage 3: no execution-scope copies, exact candidate-level MILP, one cost formula #1095 MILP:
  • A shared scan is a shareable candidate with its own context, charged once on x[m].
  • A per-window scan stays in its I_r.
  • The objective still equals plan_cost of the decoded plan. That is feat(plan-selection): group selection MILP over execution contexts as Stage 3's default search (#1095) #1144's invariant, now with no hidden rule on either side.

4. The plan shows the scans it pays for. The selected DAG has one Scan node per scan execution kind. The runtime follows the plan: one shared scan, or one scan per refresh window.

Scope and acceptance

In scope:

• the shared-scan rule at ingestion and query time (Share one scan of a source across readers with different predicates, labels and time ranges (subsumption derivations) #951's query-time design, plus the ingestion half);
• the deployment capability for incremental ingestion updates;
• the cost semantics in asap-cost and in the [Feature] Stage 3: no execution-scope copies, exact candidate-level MILP, one cost formula #1095 MILP;
• the design doc's "Refresh groups" section.

Out of scope:

• sharing across different sources;
• choosing tumbling-window widths (Pass 2: offer only gcd tumbling-window widths; finer divisors are dominated #1171).

Acceptance:

• Ingestion, with the capability. For A and B above:
  • the memo has both a shared-scan candidate and per-window scan candidates;
  • the selected plan uses the shared scan when it is cheaper;
  • plan_cost charges one scan of m's rows.
• Ingestion, without the capability. The plan and its cost are as today: one scan per refresh window.
• Query time. For rate(m[1m]) and rate(m[5m]) every 10 s, the memo has a shared 5 min scan read by the 1 min reader through TimeRange(1m). plan_cost charges the scans the selected plan runs.
• Correctness. runtime_install.rs runs both plans with IngestionBuilds::Tumbling (per window) and a new incremental mode. Every answer equals the query-time reference.
• Objective. In the [Feature] Stage 3: no execution-scope copies, exact candidate-level MILP, one cost formula #1095 MILP, objective == plan_cost on these workloads and the memo examples.
• No regressions. corpus_snapshots and sketch_plan_snapshots keep every selected plan's cost or lower it; the PR lists each changed plan.

Status and related work

• To start after feat(plan-selection): group selection MILP over execution contexts as Stage 3's default search (#1095) #1144 is merged to main. The cost semantics and the contexts are [Feature] Stage 3: no execution-scope copies, exact candidate-level MILP, one cost formula #1095's, so this builds on the group MILP, not on today's engine.
• Share one scan of a source across readers with different predicates, labels and time ranges (subsumption derivations) #951: the query-time shared-scan rule (subsumption derivations). This issue generalizes it to ingestion time and fixes the cost semantics.
• Refresh ingestion-time views that share a refresh window in one run, and charge a shared stateless ingestion node once per run #1151 (docs: refresh the ingestion-time views of one refresh window in one run (#1151) #1152–feat: charge a shared ingestion node once per refresh window (#1151) #1154): refresh windows and the per-window charge that this issue replaces with explicit candidates.
• feat(plan-selection): one ingestion pipeline per refresh window in the selection MILP (#1095 §2.5 O3(b)) #1157: I_r contexts in the [Feature] Stage 3: no execution-scope copies, exact candidate-level MILP, one cost formula #1095 MILP.
• Pass 2: offer only gcd tumbling-window widths; finer divisors are dominated #1171: tumbling-window widths. A shared ingestion scan feeds chains of different widths.