Share one scan of a source across readers with different predicates, labels and time ranges (subsumption derivations) #951

Problem, by example

Each pair below reads the same data source twice, and in each pair one scan could serve both queries. The reasons in the last column come from reading with_union_scan_labels (asap_aware_cse.rs) and memo/rules/scan_labels.rs, not from a planner run.

Expected: each pair gets a candidate in which both queries read one scan, and each applies its own compensation on top. Plan selection compares it with the separate scans.

The cost model charges every physical Scan node for its input rows and its source_read_bytes (plan-selection/src/cost/analytical.rs). A shared scan is paid once; each reader then pays its own Filter/Project per shared row. Whether sharing pays off is therefore a cost decision, not a rewrite.

Correction (2026-10-27, measured in draft PR #962)

The table's reasons came from reading code; running the planner showed two of them wrong, and the runtime constrains the design:

• Examples 3 and 5 are already shared on main. The 5-minute and 1-minute percentile_cont(0.9) read one scan and one pane chain; rate(m[5m]) and rate(m[5m] offset 2m) read one scan, and so do @ readers. Only examples 1, 2 and 4 remain.

• The runtime binds raw samples directly to the TimeRange above a scan (RawDataSampleWindow), so no Filter can sit between a scan and its window. Today only reads with the same window and shift can share one scan; "each reader keeps its own window" (Proposal §2–3, Decision 3) needs the raw-data binding to move to the Scan itself ([Refactor] Time-series TimeRange as a native operator; bind raw data only at Scan; remove the raw-sample selectors #969).

• No cost benefit measured yet. feat(optimizer): share one scan across reads with different predicates, windows and shifts #962 shares one read across same-window reads with different predicates or columns:

  Workload	main	feat(optimizer): share one scan across reads with different predicates, windows and shifts #962
  example 2, per-city SUM/MAX	0.0108458/s	same
  example 2, per-city p90	0.2084510/s (two scans)	0.2162829/s (one shared scan, +3.8%)
  example 3	0.1560126/s	same
  example 5	0.0248025/s	same

  At ingestion time a scan read by two consumers is charged once per consumer, so sharing saves nothing. At query time the joint search missed the better plan: the per-city p90 pair planned separately sums to 0.1048378/s, half of main's joint plan. That search defect is being fixed separately (seed with the separate plans, Search: certify lower bounds for query-time scans and builds #963 lower bounds, Search: exclude a rejected candidate itself, not only its query closure #964 candidate-level cuts); feat(optimizer): share one scan across reads with different predicates, windows and shifts #962 is re-measured after those land, and closed with the reason recorded here if sharing still saves nothing.

• windows_read prices an offset read by its width only, ignoring the offset (Retained raw data as concrete state groups per lookback: offsets, @ and a retention limit #955).

Why it happens

Scan sharing today is "widen, then let identical-expression CSE merge". with_union_scan_labels rewrites every predicate-free scan of one PromQL metric to project the union of the labels, so the scans become identical, and ir::cse then shares them. scan_labels.rs feeds this rewrite the eligible memo candidates (source chains, and per-series builds directly over them) and records the widened source chains as new groups (CandidateOrigin::WidenedScans).

That design only works when the scans differ in projected labels, because CSE needs identical nodes afterwards. Predicates, SQL tables, time ranges and partial sharing do not fit it.

Prior work

• Efficient and Extensible Algorithms for Multi Query Optimization (SIGMOD 2000), §2.1, subsumption derivations:

  • for σ(A<5)(E) and σ(A<10)(E), "the result of e1 can be obtained from the result of e2 by an additional selection";
  • for predicates that do not subsume each other (A=5, A=10), "we create a single new node representing the disjunction of all the selection conditions" and derive each selection from it; "the new node represents the sharing of accesses between the two selection[s]";
  • the same is done for aggregates with different group-bys (a node grouped by both).

  §3.2 then lets the optimizer keep a subsumption derivation only if the shared node is worth materializing. This is the proposal below.

• Apache Calcite shares only identical subtrees across queries: CombineSimpleEquivalenceRule spools the structurally equal sub-plans (by digest) of a multi-root Combine. RelCommonExpressionSuggester notes that shared expressions "could be revealed after applying various transformations and algebraic equivalences", but its only implementation finds identical trees. Containment with compensating predicates (MaterializedViewFilterScanRule, MaterializedViewRule) works only against a declared materialized view; Calcite never creates the wider node itself.

• DataFusion has CommonSubexprEliminate, which works on expressions within one plan node, and MATERIALIZED CTEs, which share one whole subquery. The latter explicitly does not push one reference's filters or projections into the shared body. There is no subsumption sharing.

Proposal: shared-scan rule

This is the same containment-plus-compensation structure as #950, applied to source rows instead of aggregates. Without grouping, a compensating filter may read any column.

1. Normalize source chains

/// A source chain in normalized form: a scan's source, its time range
/// and shift, its conjuncts (scan predicates, label matchers, filters
/// directly above) and the columns it projects.
struct ScanShape {
    source: Source,
    time: TimeSpine,              // TimeRange / TimeShift nodes above the scan
    conjuncts: Vec<ScalarExpr>,   // split at AND
    columns: Vec<ColumnName>,
}

2. Build one shared scan per source

Scans with the same source form one set, whatever their time ranges and shifts. Their shared scan is:

• columns: the union;
• predicate: the conjuncts common to all, AND the OR of each scan's remaining conjunctions (the paper's disjunction node). When the OR would not be a predicate the scan accepts, it is dropped and only the common conjuncts remain;
• time: the samples of every reader's window, each window placed by the reader's range and shift: (t - o - w, t - o] for m[w] offset o, (T - w, T] for m[w] @ T. Overlapping windows merge into one covering interval; disjoint ones (an @ far from now, a large offset) stay an OR of intervals, so the shared scan never reads the gap between them.

3. Compensate per reader

shared scan ▸ Filter(own remaining conjuncts, and ts > now() - w when its range is shorter) ▸ Project(own columns)

The reader must keep its original output contract: schema, unique keys and, for PromQL, series identity. This is the check same_output_contract already makes.

4. Offer candidates; plan selection decides

• The shared scan becomes one new memo group.
• Each reader gets one more candidate that reads it; the original candidates stay.
• Each reader chooses on its own, so partial sharing is possible; the shared scan's cost is paid once if any reader chooses it.
• Downstream rules see identical inputs, so a build over the shared scan is shared by the identical-expression and summary-capability rules, as widened scans are today.

5. Replace label widening

Today's label widening is the special case in which every reader's remaining conjuncts are empty and the compensation is only Project. with_union_scan_labels and the rewrite-then-CSE step go away; scan_labels.rs becomes the memo hook of this rule.

Relation to #950

• When a grouped build is read from a finer grouped build, Reuse a finer aggregate through a compensating filter and a rollup of values or states (view matching) #950 filters and merges states, so the build itself is shared, not only the scan.
• This rule covers what has no reusable summary:
  • exact computations;
  • per-series builds over scans with different matchers;
  • raw rows read at query time.

Scope and acceptance

In scope: examples 1–5; SQL tables and PromQL metrics; different time ranges for both SQL and PromQL range vectors; different time shifts (offset, @).

Out of scope:

• sources joined with other sources;
• ORs that the scan cannot push down (the shared scan then keeps only the common conjuncts).

Acceptance:

• Failing-first tests in which examples 1–5 each get a shared-scan candidate, and every reader of it executes to the same rows as its original scan on representative row sets, including empty and NULL rows.
• A PromQL test in which m[1m] read from the shared m[5m] samples gives the same samples, boundaries included, as the original selector.
• PromQL tests for offset and @: each shifted reader keeps exactly the samples of its original selector, including when the windows are disjoint.
• A cost test in which sharing is selected when the readers' predicates overlap enough, and not selected when the wider scan reads much more than the separate scans.
• The existing scan-label widening tests keep passing.

Decisions (2026-10-27)

1. Predicate of the shared scan: the OR of the remaining conjunctions (the paper's disjunction node) when every remaining conjunction is a conjunction of column-literal comparisons; otherwise only the common conjuncts.
2. One shared scan per source, with per-reader choice. There are no shared scans per subset of readers: subsets are exponential, and per-reader choice already gives partial sharing, though the shared scan stays as wide as all readers.
3. Different time ranges and time shifts over raw rows are in the first version, for both SQL and PromQL.
   • SQL: the shorter reader filters ts > now() - INTERVAL 'w' over the longer reader's rows (example 3).
   • PromQL: a shorter range vector m[1m] is read from the samples of m[5m] by keeping those in its own window. Its boundary must equal PromQL's range-selector window (left-open, (t - w, t]), which a test checks against the original selector.
   • Shifts (offset o, @ T): each reader keeps the samples of its shifted window (t - o - w, t - o] or (T - w, T]. The shared scan covers the union of the readers' windows, as an OR of intervals when they are disjoint. Samples keep their own timestamps, as in PromQL, so what is above the reader (rate, *_over_time) is unchanged.

Related

#950 (finer-aggregate reuse); the rule tables in the design doc of #509.