[zzylol]

Merge unit. This is PR 2 of #1207's merge unit (PRs 0–9). It is stacked on PR 1 (#1227, refactor/1207-properties), which is stacked on PR 0b (#1229) and PR 0 (#1226). The PRs merge into main together, after PR 9. This PR fills in the Memo methods, Stage 0 and the group properties PR 0 left as todo!(): hash-consing, candidate storage, each query with its accuracy target and $n_q$, column_sign, depends_on_evaluation_time and aggregate_coverage; and it lowers PromQL @ end() / @ start() above the subtree they anchor.

Why

The concepts doc (section 3) says rules add groups only through Memo::find_or_add_group and candidates only through Memo::add_candidate, and that every group property is derived once, when the group is created, from its input groups. Pass 1 (PR 3) and Pass 2 (PR 4) are built on these methods; Stage 2 prunes Count-Min builds by column_sign (PR 5), PR 1's delivers_on reads depends_on_evaluation_time, and Pass 2 splits a query's failure probability over its $n_q$ values. PR 0 left all of them as todo!(), so planning stopped at Stage 0.

@ end() / @ start() pin a subtree to one end of the request. Lowered in place, the pinned selector was a group of its own and shared nothing with the same selector without @ (#1251). The approved design (2026-11-08) lifts the @ above the subtree, so the subtree is the ordinary group.

What

• Memo (types/src/stages/candidates.rs):
  • find_or_add_group(expression) hash-conses on the expression (operator, arguments, input group ids) and derives GroupProperties once with derive_group_properties; an expression that breaks the node rules returns the error and changes nothing. An expression whose copies may return different rows (an unordered LIMIT) always gets a new group.
  • add_candidate(group, candidate) derives the candidate's properties the same way, requires its schema and result kind to equal the group's, drops a duplicate (same expression, same θ), and keeps parents and queries up to date.
  • add_query(MemoQuery) lists the query on every group its root reaches.
  • Memo::new(declared_column_signs) (new): the memo holds the signs the data workload declares, which a Scan's column_sign reads (P6, the recommendation; private field declared_column_signs).
• Stage 0 (stage0-memo): memo_of_queries(roots, accuracy_targets, data) builds the groups bottom-up, collects the declared signs of the sources the queries scan (SourceFacts::column_signs; a declared counter's value is non-negative without a declaration), and adds each query as a MemoQuery with its accuracy target and $n_q$ = values_per_result(root, series_read(root, data), data) (None when data bounds nothing).
• GroupProperties::column_sign (types/src/stages/column_sign.rs, new module): per output column, per the doc's "Value sign" table. Scan: declared sign, else Unknown; Filter/TimeRange/TimeShift/Sort/Limit/Dedup/column Project: the input's; abs NonNegative, negation flips, numeric literals and casts; sum/avg/min/max/quantile/top-k/first/last keep the value's sign; counts and spreads NonNegative; rate/irate/increase NonNegative over a NonNegative input; a state records the sign of its update values; a merge keeps the sign its inputs agree on; a readout reads the state's (a distinct count, L2 norm or entropy is NonNegative); anything else Unknown.
• GroupProperties::depends_on_evaluation_time: true for a TimeRange and for a scalar expression reading time() / now(); false for TimeShift { at: Some(Timestamp(T)) } (pinned); otherwise true when an input group's is.
• aggregate_coverage for summary states, chains, TimeRange/TimeShift over a chain, merges and exact Aggregates. Time is an interval on the timestamp column with IntervalConstraintLimit ends, left-open and right-closed: m[5m] offset 1h → (t − 65m, t − 60m], @ T → fixed timestamps. SQL folds a timestamp cast of a literal and resolution drops a cast to the operand's own type, so ts >= '2026-11-01' lifts. SummaryMerge(TimeRange(TimeShift(Chain[w](X)))) covers what SummaryAgg(TimeRange(TimeShift(X))) does, and add_candidate accepts it in that group. Counter states merge only over intervals disjoint in time.
• @ end() / @ start() (frontend-promql/src/promql.rs): the selector or subquery X is lowered without the @ (its offset and an @ T stay in X), and TimeShift { offset_ms: 0, at: Some(End | Start) } is put above it. Instant and range-evaluated queries lower alike; PromqlLowerer::range_evaluated (the previous revision's bool) is gone. The coverage walk stops at that TimeShift: it is the definition's input_group.
• Docs. Concepts doc (docs/memo-group-model, 10e5852): Memo::new and declared_column_signs; the "Value sign" rows for counts, rates, comparisons and readouts; depends_on_evaluation_time covers time()/now(). asap-primitive-schema.md §4.2.2 (coverage over memo groups, the time limits table, the @ rows).

Decisions

Item	Decision	Date
A group is its expression; both methods return Result; one set of node rules for groups and candidates	as implemented	approved 2026-11-06
Exact Aggregate coverage, time as an interval with IntervalConstraintLimit, @ T fixed, SQL timestamp filters, DisjointInTime, chain coverage and the merge over a chain's windows	as implemented	approved 2026-11-07
Intervals (t − r, t]	as implemented	approved 2026-11-08
@ end()/@ start() lifted above the anchored subtree as TimeShift { at }(X), X without @; range_evaluated deleted; coverage leaves the TimeShift(@) unlifted	as implemented	approved 2026-11-08
ValueSign / column_sign derived per operator, Scan from declared source facts	as implemented	approved 2026-11-08
depends_on_evaluation_time derived in Stage 0	as implemented	approved 2026-11-09
P6: the memo stores the declared signs (Memo::new, private declared_column_signs)	as implemented	approved 2026-11-09
MemoQuery::accuracy_target: Option<_> (P17), memo_of_queries(roots, accuracy_targets, data) (P5)	as implemented	approved 2026-11-09

Pending (new here, not in the doc before this PR)

#	Item	Why	Proposed
Q1	SourceColumn { source, column } in ir/operator/operator_properties.rs, beside Source	The key of declared_column_signs. The doc defines it under "Column lineage"; PR 2b (lineage) will use the same type for ColumnLineage::SourceColumn and the selection keys, and delete ColumnIdentity.	approve the placement (PR 2b imports it)
Q2	Value-sign rows beyond the approved table: numeric literal/cast; first/last_over_time; counts and spreads (count_values, changes, resets, group, absent, present_over_time, stddev, stdvar, L2 norm, entropy) NonNegative; rate/irate/increase NonNegative only over a NonNegative input; vector comparison/and/unless keep the left value's sign, or the agreed sign, bool comparisons NonNegative; a distinct count, L2 norm or entropy readout and an exact count's finalization NonNegative	The approved rows said "rate, increase: NonNegative" and "readouts: the state's recorded sign". A rate over values of either sign can be negative (a decrease reads as a reset: −5 then −10 gives −10), and a distinct count over negative values is still non-negative, so the literal rows would derive wrong signs. Doc updated (10e5852).	approve
Q3	depends_on_evaluation_time is true for time() / now()	The approved text named only TimeRange and @ T; a filter ts > now() reads the evaluation time too.	approve
Q4	@ lifted above the selector only, not above the functions over it	rate(m[5m] @ end()) lowers to rate(TimeShift(@ end())(m[5m])): the doc's "the subtree m[5m]". Lifting further (TimeShift(@ end())(rate(m[5m])), equal while nothing above reads the evaluation time) would also share the rate with instant queries; not done.	keep; lifting further is a later rule if wanted

How

• Lookup. find_or_add_group derives the properties, hashes the operator's arguments together with the input group ids (hash_consing::expression_hash), and compares the bucket with same_expression. Only shareable expressions are indexed.
• Candidates. add_candidate checks before changing anything, then appends, adds parents, and walks down from the inputs inserting the group's queries in sorted position (it stops below a group that already had them).
• Stage 0. group_of maps each node by address to its group (a diamond is visited once), children first. declared_column_signs walks the roots once for the scanned sources and reads DataWorkload::series_facts, so a histogram's _sum/_count series get the family's declarations and the counter rule.
• Signs. output_column_sign(operator, output, declared) follows each operator's output layout (aggregate_output_schema's for aggregates and states, chain_schema's for a chain: window_end Unknown, the build's own time column dropped; a merge over a chain's windows reads the chain's build). A layout it does not know (a readout whose output does not keep the state's columns) is Unknown everywhere. debug_assert checks one sign per output column.
• Time. time_constraint sums the offsets of the TimeShifts on the path, takes the one range width and the one @ T, and returns the interval; the walk in of_aggregate does not pass a TimeShift { at: End | Start }.

Key code interfaces added / modified

types/src/stages/candidates.rs, types/src/stages/memo.rs

/// The memo, Stage 0's and Stage 1's one type (stays).
#[derive(Debug, Clone, Default)]
pub struct Memo {
    /// The groups, by id; added only through `find_or_add_group`. // unchanged (private, PR 0)
    groups: Vec<MemoGroup>,
    /// The queries, by index in the workload. // unchanged (PR 0)
    pub queries: Vec<MemoQuery>,
    /// The shareable groups by the hash of their expression. // new (private)
    groups_by_expression_hash: HashMap<u64, Vec<MemoGroupId>>,
    /// The declared sign of each source column the workload declares one for. // new (private, P6)
    declared_column_signs: BTreeMap<SourceColumn, ValueSign>,
}

impl Memo {
    /// A memo without groups whose `Scan`s read `declared_column_signs`; Stage 0 creates it. // new
    pub fn new(declared_column_signs: BTreeMap<SourceColumn, ValueSign>) -> Self;
    /// The group of `expression`, added with its derived properties if new. // body filled in (PR 0 signature)
    pub fn find_or_add_group(&mut self, expression: Operator<MemoGroupId>) -> Result<MemoGroupId, SchemaDerivationError>;
    /// Adds `query`; its root's groups and every group below them list it. // body filled in
    pub fn add_query(&mut self, query: MemoQuery) -> usize;
    /// Adds `candidate` to `group`, or returns the duplicate's id. // body filled in
    pub fn add_candidate(&mut self, group: MemoGroupId, candidate: CandidateForMemoGroup) -> Result<CandidateForMemoGroupId, CandidateError>;
}

/// What every candidate of a group produces (stays; PR 0's fields, now all derived).
pub struct GroupProperties {
    /// Output schema. // unchanged
    pub schema: Schema,
    /// Output kind. // unchanged
    pub result_kind: OperatorResultKind,
    /// Per output column: the sign of its values. // derived here (was todo!())
    pub column_sign: Vec<ValueSign>,
    /// Whether the result changes with the evaluation time. // derived here (was todo!())
    pub depends_on_evaluation_time: bool,
    /// What an aggregate group covers; `None` for any other group. // chains, window selections, exact Aggregates added
    pub aggregate_coverage: Option<SummaryCoverage>,
    /// Whether two equal expressions return the same rows. // unchanged
    pub shareable_when_equal: bool,
}

types/src/ir/operator/operator_properties.rs

/// A column of a source (a `Scan`'s metric or table), by name: what the data workload declares facts about. // new (Q1)
pub struct SourceColumn {
    /// The metric or table name. // new
    pub source: String,
    /// The column name. // new
    pub column: String,
}

stage0-memo/src/lib.rs

/// Stage 0: the memo groups of the queries' roots, each query with its target and n_q; Scans read `data`'s signs. // body filled in (PR 0 signature)
pub fn memo_of_queries(roots: &[QueryRoot], accuracy_targets: &[Option<AccuracyTarget>], data: &DataWorkload) -> Result<Memo, SchemaDerivationError>;

PromQL lowering has no interface change; its output shape for @ end() / @ start() changes (TimeShift { at } above the selector).

Usage

The workload sum by (host) (rate(requests_total[1m])) and topk(5, sum by (host) (gauge)) over data declaring requests_total a counter and nothing for gauge (Stage 0 tests a_scan_reads_the_declared_column_signs, each_query_enters_with_the_values_one_evaluation_returns; types test a_top_k_ranks_values_of_its_inputs_sign):

// The front end lowers each query to a node DAG; data declares `requests_total` a counter, `gauge` nothing,
// and `series_count: 1000` for `requests_total`.
let roots: Vec<QueryRoot> = lower_promql_query_workload(&workload)?;          // two operator roots
let targets = vec![None, None];                                                // accuracy not enforced
// Stage 0: the groups, the declared signs {requests_total.value: NonNegative}, and each query's n_q.
let memo: Memo = memo_of_queries(&roots, &targets, &data)?;
// rate(requests_total[1m]) per series: value NonNegative (a counter's rate); the sum by host keeps it.
// sum by (host)(gauge): value Unknown, so Stage 2 (PR 5) prunes the Count-Min heap builds of the top-k
// (`CountMinWeightsNotProvenNonNegative`) and the query uses an exact top-k.
// The `[1m]` groups and everything above them have depends_on_evaluation_time == true.
// memo.queries[0].values_per_evaluation: the 1000 series it reads bound it (values_per_result).

And @ end() (lib test at_start_and_at_end_are_lifted_above_the_selector_they_anchor, types test at_end_and_at_start_share_the_subtree_they_anchor_and_stay_unlifted):

range query  quantile_over_time(0.9, latency[5m] @ end())
instant query quantile_over_time(0.9, latency[5m])

G1 Scan latency;  G2 TimeRange[5m](G1)                       (shared by both)
G3 SummaryAgg[KLL](G2)          coverage ts ∈ (t − 5m, t], input_group G1
G4 TimeShift(@ end())(G2);  G5 SummaryAgg[KLL](G4)            coverage: input_group G4, no time constraint

Major algorithm changes

Stage 0 hash-consing.

• Problem. Stage 0's node equality (cse::same_node, deleted in PR 0) and the group lookup the rules need were two equalities that could disagree (on a guarantee, for one).
• Before. Nodes interned bottom-up into an Rc table, then numbered and paired with groups.
• After. 1. group_of calls Memo::find_or_add_group(operator over child groups) bottom-up. 2. add_query per root, with its target and $n_q$.
• Correctness. Two expressions are one group only if their arguments are equal and their input groups are the same, so equal groups compute equal values by induction; a candidate must reproduce its group's schema and result kind. Tests: Stage 0's 15, memo_tests 10.
• Complexity. O(arguments) per lookup (children are ids); O(N) for N nodes; one extra walk over the roots for the scanned sources.
• Limits. A shareable parent over unshareable children is no longer shared.

Aggregate coverage with time intervals.

• Problem. PR 0 left time lifting, chain coverage and the counter merge check as todo!(); @ T and SQL timestamp filters had no representation; exact aggregates had no coverage.
• Before. One range TimeRange gave a separate relative time field; a chain expanded to a build over count × width.
• After. time_constraint gives (anchor − o − W, anchor − o] on the timestamp column; a chain is a build over its input with time unconstrained, and a window selection over it adds the interval; DisjointInTime compares the timestamp column; an exact Aggregate is a top unlifted operator; the walk stops at TimeShift { at: End | Start }.
• Correctness. A row is in a TimeRange's window iff its timestamp is in the interval; equal coverage means equal rows. Tests: memo_summary_coverage.rs (13).
• Complexity. One pass over the liftable operators.
• Limits. Time is not lifted from paths with several range TimeRanges or @ Ts.

Value sign and evaluation-time dependence: new derivations, one pass per group over its operator and its input groups' stored properties (O(output columns)). Tests: memo_group_properties.rs (5), Stage 0's a_scan_reads_the_declared_column_signs.

Before this PR

On PR 1, memo_of_queries, find_or_add_group, add_candidate and add_query panic at todo!(), so every planning test stops at Stage 0. m[5m] @ end() lowered as TimeRange(TimeShift(@ end())(Scan)) in a range query and as m[5m] in an instant one, through a range_evaluated flag.

After this PR

Input	After
sum by (host) (rate(requests_total[1m])), a counter	sum column NonNegative, depends_on_evaluation_time true
sum by (host) (gauge), nothing declared	sum column Unknown
latency declared {"value": "non_negative"}, quantile_over_time(0.99, latency[1m])	state and quantile NonNegative
m[5m] @ 1700000000	depends_on_evaluation_time false; coverage ts ∈ (1699999700000, 1700000000000]
latency[5m] @ end()	TimeShift(@ end()) over the group of latency[5m], shared with instant queries
a query over m with 1000 declared series	MemoQuery::values_per_evaluation == Some(1000)
KLL k = 200 and k = 345 over one input	one group, two candidates; a duplicate returns the first id

Planning passes Stage 0 and stops at Pass 1's target discovery (PR 3).

Deleted, and what replaces each

Deleted	Replaced by
PromqlLowerer::range_evaluated and the range_evaluated parameter of lower_query, vs_parts, time_shift, extract_matrix (previous revision)	anchored / selector in promql.rs: @ end() / @ start() above the subtree, the same for every query
test at_end_and_at_start_share_coverage_only_in_an_instant_query (types)	at_end_and_at_start_share_the_subtree_they_anchor_and_stay_unlifted
test at_start_and_at_end_stay_only_in_a_range_query (promql lib)	at_start_and_at_end_are_lifted_above_the_selector_they_anchor
Stage 0 test differing_accuracy_targets_are_two_groups_over_one_scan_group (measures carry no target since PR 0)	queries_that_differ_only_in_their_accuracy_target_are_one_group
the todo!()s for column_sign, depends_on_evaluation_time, Stage 0, add_query, and #[allow(unreachable_code)] on derive_group_properties	the derivations

Self-review: over-abstraction, duplication, dead code

• Over-abstraction. New: Memo::new and the private declared_column_signs (P6, approved), SourceColumn (Q1, the doc's type), private column_sign.rs functions, MemoGroupAsInput::group_properties() (pub(crate), one accessor the two derivations share). No trait or wrapper added; the sign derivation is not generic over OperatorInput because only memo groups have signs.
• Duplication. column_sign.rs repeats the output-column layouts of aggregate_output_schema and chain_schema (keys, then measures; a per-entity reduction in place). PR 2b's column_lineage derives the same layouts; when PR 2b is rebased it should map both properties through one layout function instead of a second copy (noted in the PR 2b handoff). Memo keeps Default (an empty memo without declared signs, used by tests and by Memo::new) beside Memo::new.
• Dead code. None: clippy -D warnings; the range_evaluated plumbing is gone; the stale #1207 PR 2 comments are removed.
• Fixed in this round. The coverage walk passed TimeShift { at: End | Start } and returned no interval for the whole path; it now stops there, so the time path below it stays the anchored group's own.

Verification

cargo test -p <crate> --all-features --no-fail-fast at the head (8095a51). Every failure panics at a later PR's todo!() (or is the stale fixture, PR 9); planning now stops at Pass 1's target discovery (PR 3).

Crate	Passed	Failed	Ignored	Fails at
asap-types	194	0	0	—
asap-accuracy	60	0	0	—
asap-cost	31	1	0	check θ, which the exported node carries (PR 6): 1
asap-frontend-common	49	0	0	—
asap-sql-function-catalog	10	0	0	—
metricsql_common	3	0	0	—
asap-frontend-promql	218	0	0	—
asap-frontend-metricsql	15	0	0	—
asap-frontend-sql	174	0	0	—
asap-executor	210	63	1	check θ, which the exported node carries (PR 6): 32; the Hydra grid's sizes are θ, which the exported node carries (PR 6): 13; each materialized node's retention_ms is its h_v, which Stage 3 sets (PR 6): 8; the weighted-frequency θ, which the exported node carries (PR 6): 7; check the weighted-frequency θ, which the exported node carries (PR 6): 2; discover targets over the memo groups, not node trees (PR 3): 1
asap-planner	8	18	1	discover targets over the memo groups, not node trees (PR 3): 18
asap-devtools	37	56	3	discover targets over the memo groups, not node trees (PR 3): 48; a group's node from its expression (PR 9): 5; an asap_planner subprocess that stops at a todo!(): 2; label a group by its expression (PR 9): 1
asap-integration-tests	83	51	4	discover targets over the memo groups, not node trees (PR 3): 47; each materialized node's retention_ms is its h_v, which Stage 3 sets (PR 6): 3; the stale fixture (old wire format; PR 9): 1
asap-stage3-plan-selection	3	0	0	—
asap-stage0-memo	15	0	0	—
asap-stage2-physical-optimization	0	0	0	—
asap-stage1-pass2-reuse-candidates	0	0	0	—
asap-stage1-pass1-per-computation-candidates	11	0	0	—
Total	1121	189	9

Compared by test name with PR 1 at 8030d3fe6: no test that passed fails. 45 tests are new and pass (asap-types 30, asap-stage0-memo 15, one each in asap-frontend-promql and asap-frontend-sql); PR 1's delivers_on_a_chain_when_its_width_divides_the_context_times and delivers_on_a_stored_result_on_its_group_grid failed at this PR's todo!() and pass. Tests that failed at PR 2's todo!() on PR 1 now fail at PR 3's (target discovery) or PR 6's.

• cargo clippy --workspace --all-targets --all-features -- -D warnings: clean. cargo fmt --all --check: clean. crate_graph: 6 passed.
• New or rewritten tests, each with a doc line stating the behavior: memo_group_properties.rs (5: sign table rows for row operators, aggregates, states/merges/readouts, the doc's top-k example; depends_on_evaluation_time true/false), at_end_and_at_start_share_the_subtree_they_anchor_and_stay_unlifted, Stage 0's each_query_enters_with_the_values_one_evaluation_returns, a_scan_reads_the_declared_column_signs, queries_that_differ_only_in_their_accuracy_target_are_one_group, promql lib at_start_and_at_end_are_lifted_above_the_selector_they_anchor, conformance at_modifier_lowers_to_a_time_shift (updated).
• Rebased 2026-11-09 onto PR 1 8030d3fe6 (previous head kept as backup/1207-memo-fe0ff3357); the four previous commits are squashed into one.