[zzylol]

Addresses #840 (questions 1 and 5).

Why

In #840's workload, Q1/Q2 (quantile_over_time over 1 m and 5 m), Q3 (quantile by (host, job) (0.5, data)) and Q4 (quantile by (job) (0.5, data)) all read the rows of data. The PromQL frontend projects only the labels each query uses into its Scan (ts, value, $promql_series_identity for Q1/Q2, plus host, job for Q3 and job for Q4), and structural CSE shares only nodes with equal schemas. So the identical-expression rule never offered to share Q3's or Q4's scan, nor Q3 and Q4's identical instant time ranges. Since #855, Stage 2 can build a shared scan once per evaluation, so sharing these scans can now be priced fairly.

What

• The identical-expression rule also tries each candidate with every predicate-free time-series scan of a metric widened to the union of the labels its scans project. Q1–Q4 then share one scan, and Q3 and Q4 one time range.
• Each query keeps its own columns: nodes that pass the labels on keep the wider rows (time ranges and shifts, and a per-series build or pane chain without a filter, with its merges, estimates and finalizations); any other reader gets a Project that restores its columns by name, as fix: one finest-grain stored summary serves several groupings (quantiles at the estimate level, sums at the state level) #804 did for per-series builds.
• Only the widened candidates that share a scan are kept, and they are listed after every other Pass 2 candidate, so a plan reached by both the summary-capability rule and the widened identical-expression rule keeps the former on a tie.
• planner-layering.md (Pass 2 section) and the asap_aware_cse module doc describe it.

How

with_union_scan_labels(roots) walks the DAG children first and builds, per metric, the union schema of its unfiltered scans (labels inserted before the series identity, in walk order; a label with two types disables widening for that metric). It returns None when every metric's scans already agree. Otherwise it rebuilds the DAG: widened scans; passes_labels_on nodes rebuilt over the wider children; every other node over narrowed(child) (a Project by name). It gives up (None) when a reader would have to narrow a summary state, or when a grouped or filtered pane chain would read a Project (plan selection reads a pane chain's window from the rows directly below it; the executor also needs a per-series build's TimeRange directly below it). Guarantees are recomposed. Pass2List::identical_expression runs the rule on the original roots as before, then on the widened roots, keeping those that share a Scan in Pass2List.widened, appended at the end.

Key code interfaces added / modified

crates/logical-asap-optimization/src/asap_aware_cse.rs

/// `roots` with each metric's predicate-free time-series scans widened to the union of
/// their labels, each other reader narrowed back by a `Project`; `None` when nothing
/// needs widening or it cannot be done. Pass 2's identical-expression rule calls it.
pub fn with_union_scan_labels(
    roots: &[QueryRoot],
) -> Result<Option<Vec<QueryRoot>>, PerComputationCandidateError>;   // new
/// The source and schema of a time-series `Scan` without predicates.
fn unfiltered_series_scan(node: &OperatorNode) -> Option<(&Source, &Schema)>;   // new
/// Whether `node` keeps a wider child's labels in its own output.
fn passes_labels_on(node: &OperatorNode) -> bool;                                // new
/// `wide` with only `schema`'s columns, by name; `None` for a summary state.
fn narrowed(
    wide: &Rc<OperatorNode>,
    schema: &Schema,
) -> Result<Option<Rc<OperatorNode>>, PerComputationCandidateError>;           // new

crates/logical-asap-optimization/src/workload_candidates.rs

/// Pass 2 candidates of one Pass 1 candidate, in listing order. Private; the type stays.
struct Pass2List<'v> {
    /// The flattened DAGs listed so far, to skip duplicates.
    seen: Vec<asap_types::ir::flat::FlatDag>,   // unchanged
    /// The Pass 2 candidates, in listing order.
    out: Vec<Pass2>,                             // unchanged
    /// The identical-expression candidates over widened scans, appended after `out`
    /// so a tie keeps the unwidened candidate.
    widened: Vec<Pass2>,                         // new
    /// Whether a candidate can have a valid implementation.
    viable: &'v dyn Fn(&[QueryRoot]) -> bool,    // unchanged
}

Usage, end to end: #840's 4-query workload through stage_pipeline at this commit.

# 1. What the user writes (#840, with Q4): four repeating PromQL queries every 10 s,
#    ε = 0.01, δ = 0.001, over `data` (gauge, 10,000 series, one sample per 15 s):
#    Q1 quantile_over_time(0.5, data[1m])     Q2 quantile_over_time(0.9, data[5m])
#    Q3 quantile by (host, job) (0.5, data)   Q4 quantile by (job) (0.5, data)
cargo run -p asap-devtools --bin stage_pipeline -- --workload wl.json \
  --exhaustive --enforce-accuracy-guarantees --max-displayed 100000 --out out.json
# 2. Stage 1, Pass 2 (Pass2List::identical_expression): with_union_scan_labels widens
#    the four scans to [ts, value, host, job, $promql_series_identity]. Q1/Q2's per-series
#    KLLs keep the wider rows (passes_labels_on); their estimates and Q3/Q4's grouped
#    builds read Projects back to their own columns. Interning now finds Scan (Q1–Q4)
#    and TimeRange (Q3, Q4) identical; both candidates share a scan, so both are kept.
# 3. Stage 2 (#855) offers the shared Scan and TimeRange built once per evaluation.
# 4. Stage 3 prices the shared scan once per evaluation over the 5 m window:
#      scan 200000 samples; x 0.1000/s   (multiplier 1)
# 5. What the user gets back (stage3_plan_selection):
#      P330-m1  Q1 Kll · Q2 Kll · Q3 Kll · Q4 Kll · shared Scan (Q1, Q2) · query time, shared: Scan   0.1536/s  selected
#      P329     Q1 Kll · Q2 Kll · Q3 Kll · Q4 Kll                                                     0.1702/s  costlier
#      P339-m3  Q1 Kll · Q2 Kll · Q3 Kll · Q4 Kll · shared Scan (Q1, Q2, Q3, Q4), TimeRange (Q3, Q4)
#                 · query time, shared: Scan, TimeRange                                             0.2068/s  costlier

Before this PR

• Instant quantile by (host, job) next to windowed quantiles: no scan sharing, no panes, no ingest-time variant, 100-group default #840 Q1–Q4 (on main with feat(stage2): offer a shared non-summary node built once per evaluation #855): 468 logical and 1,161 physical candidates; the only sharing offered is shared Scan (Q1, Q2). Q1–Q3: 156 and 387.

After this PR

• Q1–Q4: 954 logical and 3,357 physical candidates; Q1–Q4 can share one scan, and Q3 and Q4 one time range. Q1–Q3: 237 and 639.
• The shared Q1–Q4 scan is priced as one 5 m scan per evaluation (200,000 samples, multiplier 1), but it does not pay here: 0.2068/s against 0.1702/s for separate scans, because the widened scan carries Q3's labels (12.8 MB instead of 6.4 MB per evaluation) and Q3 and Q4 filter 5 m of rows for their latest samples instead of scanning 15 s. The selection stays shared Scan (Q1, Q2) at 0.1536/s (Q1–Q3: 0.1458/s), as Instant quantile by (host, job) next to windowed quantiles: no scan sharing, no panes, no ingest-time variant, 100-group default #840's analysis predicted.
• The committed seed-862 viewer fixtures select the same plan at the same cost; only candidate ids shift.
• Limitations: a grouped build over a Project is sized without the source's declared non-negativity (plan selection looks for the scan or time range directly below it). Widening is all-or-nothing per metric, so Q3 and Q4 cannot share a 15 s scan while Q1 and Q2 keep theirs. A whole-window grouped build that reads a Project is not offered at ingestion time (Stage 2 looks for its TimeRange directly below it); its unwidened counterpart keeps that option. The widened candidates roughly double the logical candidates of a workload whose scans of one metric project different labels.

Verification

• Failing first: scans_projecting_different_labels_are_shared (crates/devtools/tests/stage_pipeline.rs, Instant quantile by (host, job) next to windowed quantiles: no scan sharing, no panes, no ingest-time variant, 100-group default #840's 4-query workload) has no "shared Scan (Q1, Q2, Q3, Q4)" candidate on main; it passes now: one scan with ts, value, host, job, $promql_series_identity, multiplier 1, "scan 200000 samples; x 0.1000/s", 0.2068/s against 0.1702/s, and the Q1/Q2 shared scan selected.
• scans_of_one_metric_with_other_labels_share_once_widened (unit): sum by (job) (m) and max by (host, job) (m) share nothing, then a Scan and a TimeRange once widened; each root keeps its schema; agreeing scans return None.
• joint_search_equivalence::a_shared_summary_is_combined_with_a_rollup counts a build shared by either rule.
• with_union_scan_labels keeps the scans' unique key on the series identity and ts after widening (unit test), and does not widen scans that disagree on whether they hold every label.
• Independent review (asap-code-review): fixed the stale unique-key positions, the closed-flag conflict, a failed rebuild now skips widening, doc wording; the non-negativity sizing and all-or-nothing widening are stated as limitations; the interface options are left as they are.
• On this head, rebased onto main: cargo fmt --all -- --check; cargo clippy -p asap-logical-asap-optimization -p asap-devtools -p asap-integration-tests --all-targets --all-features --locked -- -D warnings clean; asap-logical-asap-optimization lib (164), asap-integration-tests joint_search_equivalence (12), asap-devtools stage_pipeline (36, including the committed fixtures) pass.