[zzylol]

Merge unit. This is PR 2b of #1207's merge unit (PRs 0–9), between PR 2 (#1230, refactor/1207-memo) and PR 3 (#1232). It is stacked on PR 2 (8095a51); the PRs merge into main together, after PR 9. Planning still panics at the first todo!() of Pass 1 (PR 3); see "Test status". One commit: 58549f2.

Why

Schema fields carried a table qualifier (Field::table), and every operator rule had to keep it right: a Project replaced it with its own qualifier, an aggregate copied it onto its keys, and two patches put it back where rules lost it (with_aggregate_qualifiers in Pass 1, the requalification in OperatorNode::with_new_children / validate_structure). Summary coverage named a selection's column by (table, name) (ColumnIdentity), so a rename made the same source column look like another column, and Memo::add_candidate's exact schema compare rejected a readout renamed to an aggregate's column because its qualifier differed.

The concepts doc (memo-groups-properties-requirements.md, section 3, "Column lineage") replaces the qualifier by a group property: each output column's lineage, the source column its values are copied from, or Computed. A front end binds a qualified name such as f.src with its own scope while it lowers, and the IR never needs the qualifier again.

PR 2 added column_sign, a second per-column group property, with its own copy of every operator's column layout (aggregates, chains, readouts). Lineage needs the same layout, so this PR derives both from one.

What

• GroupProperties::column_lineage: Vec<ColumnLineage>, one entry per output column in schema order, derived once in derive_group_properties.
• One per-operator column layout for lineage and sign. stages/column_sign.rs (PR 2) becomes stages/column_properties.rs: output_column_properties returns each output column's lineage and sign together. A column that copies an input column keeps both; a computed one is Computed, with the sign the operator's rule gives (PR 2's rules, unchanged). GroupProperties keeps the two fields the doc lists, column_lineage and column_sign.
• One SourceColumn. PR 2's SourceColumn { source, column } beside Source (ir/operator/operator_properties.rs) is the only one: it is the declared-sign key (PR 2), ColumnLineage::SourceColumn(SourceColumn), and the key of SelectionBox::columns. ColumnIdentity is deleted. It stays beside Source because the data workload's sign declarations and Stage 0 already use it there, and lineage and coverage import it from there; the earlier copy in ir/properties/column_lineage.rs is deleted with that module.
• Chains and window selections (Tracking: implement the optimization formulation (memo candidates, properties, edges y_{u,c,i}, MILP) as named code per stage #1207 restack plan, PR 2b line): a SummaryTumblingWindowChain has window_end (column 0) Computed, its keys their input's lineage, its state Computed; a TimeRange or TimeShift over a chain passes the chain's lineage through; the SummaryMerge of a chain's selected windows has the chain build's layout (keys, then state). TimeShift { at: End | Start } above a selector (PR 2's lowering of @ end() / @ start()) passes the selector's lineage through.
• Summary coverage names a selection's column by lineage. A conjunct on a Computed column is not lifted and stays in unlifted_operators; neither is one on a column whose lineage another column of the SummaryAgg's input shares. The regrouping merge's key check compares lineage, and the timestamp column a DisjointInTime merge and a window selection constrain is named by its lineage.
• Schema fields lose table; the IR Project loses qualifier. Field::with_table and Schema::column_id_qualified are deleted; resolve_column_ref resolves a ColumnRef::Qualified by its unique name.
• Front ends (frontend-common's resolver, used by SQL, PromQL and MetricsQL) bind qualified names with their own scope: each resolved operator carries the qualifier of each output column next to its node. The SQL front end sets UnresolvedOp::Scan::qualifier to the table or alias.
• Deleted patches: with_aggregate_qualifiers (Pass 1) and the requalification in node.rs and tumbling_window_chain.rs. Memo::add_candidate stays strict (exact schema compare).
• Design docs: asap-primitive-schema.md §4.2.2 (worked example G1–G10 and "Which column a condition is on") uses lineage: latency.job ∈ {api}, latency.ts ∈ (t − 60 s, t], t.region ∈ {us}; §2's table and the listings of Field, Schema, SelectionBox and SourceColumn follow. The concepts doc's lineage table gains the chain-selection, @ end() / @ start() and SummaryMerge rows and says lineage and sign share one layout (docs/memo-group-model 1662257).
• Not in this PR: export and wire formats (PR 6). See "Wire format".

Deleted

Item	Where	Replaced by
Field::table, Field::with_table	types/src/ir/schema/mod.rs	GroupProperties::column_lineage; front-end scope
Schema::column_id_qualified	same	Scope::resolve (front end, private)
NonASAPOp::Project::qualifier	types/src/ir/operator/non_asap.rs	the front end's UnresolvedOp::Project::qualifier only
requalification in OperatorNode::with_new_children and validate_structure	types/src/ir/operator/node.rs	nothing: only names may be overridden
with_aggregate_qualifiers	stage1-pass1-per-computation-candidates	nothing: no qualifier to restore
ColumnIdentity { table, name } and its test-support new	types/src/ir/properties/summary_coverage.rs	SourceColumn (beside Source)
private column_identity	same	private source_column (by lineage)
private build_input_schema	same	private build_input (the input group, for its schema and lineage)
stages/column_sign.rs: output_column_sign and its per-operator layout (non_asap_signs, asap_signs, aggregate_signs, state_signs, readout_signs, project_signs, agreed)	types/src/stages/	stages/column_properties.rs: one layout for lineage and sign
three copies of the topk_{k} output-name rule	non_asap.rs, the sign/lineage rule, agg_intent.rs	topk_output_name (ir/schema/topk.rs)
test qualified_field_serde_roundtrip, the qualified case of partitions_over_matches_keys_by_column, the qualifier asserts of rebuilding_preserves_only_explicit_naming_overrides	types	—

How

• One layout. output_column_properties(operator, output, declared, columns_of) matches the operator once and returns a ColumnProperties { lineage, sign } per output column. A Scan gives SourceColumn { source: table or metric, column: field name } with its declared sign; Filter, Sort, Limit, Dedup, TimeRange, TimeShift and the other row-selecting PromQL operators pass their input's columns; a Project column item copies its input column, an expression item is Computed with expression_sign; an Aggregate and a SummaryAgg follow aggregate_output_schema's layout (a by key copied, a without reduction the kept labels, a per-series reduction every column but the value; measures and the state Computed with measure_sign / the update's sign); a chain is its build's layout without the build's time column, after a computed window_end; a readout copies its state's plain columns; a Join concatenates; SetOp, Concat and a SummaryMerge keep what their inputs agree on; a BinaryOp keeps the left side's columns (and, for or / group_right, what both sides agree on), computes the value, and copies the labels the match adds from the right side. derive_group_properties reads a group's columns with columns_of_group and splits the result into column_lineage and column_sign. Each rule gives one entry per output column (debug_assert_eq!).
• Coverage. of_aggregate reads the lineage of the aggregate's input (a group's stored lineage, or a temporary build's, derived by the same output_column_properties from its inputs'). residual lifts a conjunct only when source_column names its column: a SourceColumn lineage no other column shares; the selection is keyed by it. regrouped maps each key to the build input's column of equal lineage.
• Front end. resolve returns a private Bound { node, qualifiers }. Qualified references resolve in a Scope { schema, qualifiers }: a column with that qualifier and name first, else the unique name. Qualifiers follow the old field rules: a Scan its own, a pass-through operator its input's, a Project its alias or none, an aggregate's by keys their input's, a Join left then right, a SetOp the left side's; PromQL operators have none.

Major algorithm changes

Summary coverage: which conjuncts are lifted.

• Problem. Selections named columns by (table, name). SELECT source … FROM (SELECT f.src AS source FROM flows f) named the column (none, source), and one grouped by f.src named it (f, src): the same column looked like two. A conjunct on a computed column (a rate value) was lifted although no source column names it.
• Before. 1. Collect the conjuncts the walk can lift (residual). 2. Lift one when it is a value set or interval on one column whose (table, name) is unique in the aggregate's input. 3. Key the selection by (table, name).
• After. 1. Same. 2. Lift one when it is a value set or interval on one column whose lineage is a SourceColumn unique among the input's columns. 3. Key the selection by that source column.
• Correctness. A source column's lineage is the same in every group whose column copies it, so two selections on one lineage constrain the same values, renamed or not; distinct lineages are distinct source columns (a join's two regions). A Computed column has no source to compare by, so it stays in the definition, which merges only equal definitions. Tests: types/tests/column_lineage.rs and memo_summary_coverage.rs.
• Complexity. Lineage and sign are derived once per group, O(columns) per group (the input groups' columns are copied once per rule). source_column counts namesakes as before, O(columns) per conjunct.
• Limits. rate(m[5m]) > 0 no longer lifts value ∈ (0, ∞): value is computed by rate. A top-k readout's and a population evaluation's columns are Computed (conservative).

Key code interfaces

crates/types/src/stages/memo.rs

/// Where an output column's values come from, a group property derived from
/// the input groups (new type; `asap_types::stages::ColumnLineage`, beside `ValueSign`).
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ColumnLineage {
    /// Copied unchanged from a column of a source. // new
    SourceColumn(SourceColumn),
    /// Computed by an operator: a measure, an expression, a summary state or readout. // new
    Computed,
}

/// What every candidate of a group produces (stays).
pub struct GroupProperties {
    /// Fields: name, type, nullability; no table qualifier. // unchanged type, fields lose `table`
    pub schema: Schema,
    /// The output category. // unchanged
    pub result_kind: OperatorResultKind,
    /// Per output column, in schema order: its source column or `Computed`. // new
    pub column_lineage: Vec<ColumnLineage>,
    /// Per output column, in schema order: the sign of its values. // unchanged (now derived with lineage)
    pub column_sign: Vec<ValueSign>,
    /// Whether the result changes with the evaluation time. // unchanged
    pub depends_on_evaluation_time: bool,
    /// What an aggregate group covers. // unchanged (selections keyed by `SourceColumn`)
    pub aggregate_coverage: Option<SummaryCoverage>,
    /// Whether two equal expressions may be one group. // unchanged
    pub shareable_when_equal: bool,
}

crates/types/src/ir/operator/operator_properties.rs

/// A column of a source (a `Scan`'s metric or table), by name: the declared-sign key,
/// the lineage of a column copied from it, and a selection's key (stays; PR 2's, now the only one).
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
pub struct SourceColumn {
    /// The table or metric a `Scan` reads. // unchanged
    pub source: String,
    /// The column's name in that source. // unchanged
    pub column: String,
}

crates/types/src/ir/properties/summary_coverage.rs

/// One box of a coverage selection (stays).
pub struct SelectionBox {
    /// Per restricted column, its constraint; the timestamp column is one of them. // key was `ColumnIdentity`
    pub(crate) columns: BTreeMap<SourceColumn, Constraint>,
}
impl SelectionBox {
    /// Test support. // key was `ColumnIdentity`
    pub fn new(columns: BTreeMap<SourceColumn, Constraint>) -> Self;
    /// The per-column constraints. // key was `ColumnIdentity`
    pub fn columns(&self) -> &BTreeMap<SourceColumn, Constraint>;
}
// removed: pub struct ColumnIdentity { table: Option<String>, name: String } and ColumnIdentity::new

crates/types/src/stages/column_properties.rs (was column_sign.rs; crate-private)

/// One output column's lineage and sign (new, pub(crate)).
pub(crate) struct ColumnProperties {
    /// Its source column or `Computed`. // new
    pub(crate) lineage: ColumnLineage,
    /// The sign of its values. // new
    pub(crate) sign: ValueSign,
}
/// Each output column's lineage and sign, from the inputs' columns and, for a `Scan`, the declared signs.
pub(crate) fn output_column_properties<I: OperatorInput>(
    operator: &Operator<I>,
    output: &Schema,
    declared: &BTreeMap<SourceColumn, ValueSign>,
    columns_of: &dyn Fn(&I) -> Vec<ColumnProperties>,
) -> Vec<ColumnProperties>; // was output_column_sign(operator, output, declared) -> Vec<ValueSign>
/// A memo group's columns, read from its `group_properties`. // new
pub(crate) fn columns_of_group(group: &MemoGroupAsInput<'_>) -> Vec<ColumnProperties>;

crates/types/src/ir/schema/mod.rs

/// One field of a `Schema` (stays).
pub struct Field<T = FieldDataType> {
    /// Field name. // unchanged
    pub name: String,
    /// Plain value or summary state. // unchanged
    pub dtype: T,
    /// Whether NULL is allowed. // unchanged
    pub nullable: bool,
    // removed: pub table: Option<String>
}
impl<T> Field<T> {
    /// A field. // unchanged
    pub fn new(name: impl Into<String>, dtype: T, nullable: bool) -> Self;
    // removed: pub fn with_table(self, table: impl Into<String>) -> Self;
    /// A reference to this field by its name. // was qualified when the field had a table
    pub fn column_ref(&self) -> ColumnRef;
}
impl Schema {
    // removed: pub(crate) fn column_id_qualified(&self, table: &str, name: &str) -> Option<ColumnId>;
}

crates/types/src/ir/operator/non_asap.rs

pub enum NonASAPOp<C = Rc<OperatorNode>> {
    /// π — projection (stays).
    Project {
        /// Output items. // unchanged
        cols: Vec<ProjectItem<C>>,
        // removed: qualifier: Option<String>
        /// Input. // unchanged
        child: C,
    },
    // every other variant unchanged
}

crates/frontend-common/src/unresolved.rs

pub enum UnresolvedOp {
    /// A leaf (stays).
    Scan {
        /// Table or metric. // unchanged
        source: Source,
        /// Leaf predicates. // unchanged
        predicates: Vec<UnresolvedPredicate>,
        /// Catalog schema (SQL) or `None` (usage-derived). // unchanged
        schema: Option<Schema>,
        /// The SQL table or alias qualifying the leaf's columns; `None` for PromQL. // new
        qualifier: Option<String>,
    },
    /// A projection (stays).
    Project {
        /// Output items. // unchanged
        cols: Vec<UnresolvedProjectItem>,
        /// A derived table's alias qualifying every output column. // unchanged (doc only)
        qualifier: Option<String>,
        /// Input. // unchanged
        child: Rc<UnresolvedOp>,
    },
    // every other variant unchanged
}

crates/planner/src/lib.rs

pub use asap_types::ir::operator::SourceColumn; // new (was `ColumnIdentity` from summary_coverage)
pub use asap_types::ir::properties::summary_coverage::{Constraint, IntervalConstraintLimit, SelectionBox}; // `ColumnIdentity` removed

Usage

A SQL query, lowered and put in the memo; then Pass 1 (PR 3) adds an HLL readout as a candidate of the aggregate's group. Pass 1 is todo!() until PR 3, so an_hll_readout_renamed_to_the_aggregate_column_is_its_candidate (types/tests/column_lineage.rs, passing at this head) builds the groups Stage 0 and Pass 1 would add (helpers scan, filter, equals, hll, column_of in that file):

// The query the user writes: distinct destinations per source, TCP only.
//   SELECT f.src, approx_distinct(f.dst) FROM flows f WHERE f.proto = 'tcp' GROUP BY f.src
// The SQL front end lowers `FROM flows f` to UnresolvedOp::Scan { qualifier: Some("f"), .. };
// the resolver binds `f.src`, `f.dst`, `f.proto` in that scope to columns 0, 1, 2.
// Stage 0 adds one group per operator, deriving lineage and sign together:
let mut memo = Memo::default(); // the workload's memo; no declared signs, so every sign is Unknown
let g1 = memo // Scan flows; lineage [flows.src, flows.dst, flows.proto, flows.ts]
    .find_or_add_group(scan("flows", &["src", "dst", "proto", "ts"]))
    .unwrap();
let g2 = memo.find_or_add_group(filter(equals(2, "tcp"), g1)).unwrap(); // Filter: g1's columns
let g3 = memo // the query's aggregate: columns `src`, `approx_distinct(dst)`; lineage [flows.src, Computed]
    .find_or_add_group(Operator::NonASAP(NonASAPOp::Aggregate {
        reduction: Reduction::by(vec![0]),
        measures: vec![AggIntent::Cardinality { cols: vec![1] }],
        output_names: vec!["approx_distinct(dst)".into()],
        filters: vec![],
        having: None,
        child: g2,
    }))
    .unwrap();
// Pass 1 adds the HLL state and its readout as groups:
let g4 = memo.find_or_add_group(hll(g2, vec![0], "dst")).unwrap(); // SummaryAgg[HLL] by src; [flows.src, Computed]
// g4's aggregate_coverage selection is { flows.proto ∈ {tcp} }, keyed by column_of("flows", "proto")
let g5 = memo // SummaryEstimate[cardinality]: columns `src`, `cardinality`; [flows.src, Computed], sign NonNegative
    .find_or_add_group(Operator::ASAP(ASAPOp::SummaryEstimate {
        summary_input: g4,
        query: SketchStatistic::Cardinality,
    }))
    .unwrap();
// ... and the readout renamed to g3's column names as a candidate of g3:
let operator = Operator::NonASAP(NonASAPOp::Project { // columns `src`, `approx_distinct(dst)`
    cols: vec![
        ProjectItem { alias: Some("src".into()), expr: ScalarExpr::Column(0) },
        ProjectItem { alias: Some("approx_distinct(dst)".into()), expr: ScalarExpr::Column(1) },
    ],
    child: g5,
});
let expression = CandidateExpression { operator }; // the candidate's expression over g5
let candidate = CandidateForMemoGroup { // an exact candidate: no θ
    logical_properties: LogicalProperties::new(&expression, None).unwrap(),
    expression,
    input_requirements: vec![LogicalRequirements::default()],
    fused_groups: vec![],
    origin: OriginOfCandidateForMemoGroup::Pass1,
};
// Its schema equals g3's (no qualifier to differ), so the strict add_candidate accepts it:
let id = memo.add_candidate(g3, candidate).unwrap(); // CandidateForMemoGroupId { group: g3, index_in_group: 0 }

Before this PR

• Every operator rule and two patches kept Field::table right; coverage named columns by (table, name), so a rename broke column identity, and a readout renamed to an aggregate's column failed add_candidate's exact schema compare.
• The resolver bound qualified names through the qualifier on each field.
• PR 2's column_sign had its own copy of every operator's column layout.

After this PR

• Memo groups carry column_lineage, derived with column_sign from one layout; coverage, its merge checks and its regrouping key check read it. Renamed columns keep their identity; the two regions of a join stay apart; a computed column is never named in a selection; a chain's keys, its window selections and @ end() / @ start() keep their lineage.
• Schema fields have only name, type and nullability. Qualified names exist only in the front end, bound during lowering.
• PR 3 (Pass 1) can add readouts renamed to the aggregate's columns with the strict add_candidate, and names columns by SourceColumn.

Wire format

The exported Schema JSON no longer has a "table" key on fields, and a selection key (SourceColumn) serializes as {source, column} (only in a materialized view's coverage, which is todo!() until PR 6). Older plan files still deserialize (a field's unknown keys are ignored). No consumer reads a qualifier (the executor matches columns by name), so nothing is derived at the export boundary; PR 6 owns the wire version.

Pending approval

New or changed public items and behaviour not in the doc or #1207 (conservative choices):

1. ColumnLineage lives in asap_types::stages (stages/memo.rs, beside ValueSign), not in a module of its own (ir::properties::column_lineage is deleted). Definition above. Example: use asap_types::stages::{ColumnLineage, ValueSign}; then assert_eq!(memo.groups()[g.0 as usize].group_properties.column_lineage[0], ColumnLineage::SourceColumn(SourceColumn { source: "flows".into(), column: "src".into() })).
2. SourceColumn stays beside Source (feat(stage0, types): memo hash-consing, group properties (column_sign, depends_on_evaluation_time, aggregate_coverage), @ end()/@ start() above their subtree (#1207 PR 2) #1230 Q1), with pub fields and no constructor (the 11-07 copy had pub(crate) fields and a test-support new). Example: SourceColumn { source: "latency".into(), column: "job".into() } is the lineage of every job column copied from the series latency and the key of a selection latency.job ∈ {api}.
3. The planner facade re-exports SourceColumn (asap_planner::SourceColumn) in place of ColumnIdentity, since it keys SelectionBox::columns.
4. Sign of the labels a BinaryOp match adds: copied from the right side's column together with its lineage, where PR 2 gave Unknown; for or / group_right, a left label keeps the sign both sides agree on (PR 2 kept the left side's). One layout copies a column's lineage and sign together, so the two cannot disagree. Example: in a * on(job) group_left(team) b, the output's last column team has lineage b.team and b's team sign (PR 2: always Unknown). Labels are strings, whose sign is Unknown unless declared, so no current contract reads a different value. The value column's sign rules are PR 2's, unchanged.
5. Kept from the 11-07 review (approved there): UnresolvedOp::Scan::qualifier: Option<String>; ResolveError::Ambiguous message "matches several columns"; resolve_column_ref ignores a qualifier.

Verification

• New tests (types/tests/column_lineage.rs), one per doc example plus the chain rules:
  • an_hll_readout_renamed_to_the_aggregate_column_is_its_candidate (example 1): lineage of G1–G5 and add_candidate accepting the renamed readout.
  • a_rename_keeps_lineage_and_the_rollup_key_check_compares_it (example 2).
  • join_columns_of_one_name_have_distinct_lineage (example 3).
  • scan_predicates_lift_and_their_states_merge (example 4, in memo_summary_coverage.rs, asserting latency.job).
  • a_predicate_on_a_computed_column_stays_unlifted.
  • a_chain_computes_window_end_and_state_and_its_window_selections_pass_lineage_through: chain [Computed, latency.job, Computed]; TimeShift and TimeRange over it the same; the merge of the selected windows [latency.job, Computed] (under the 11-07 rule this merge failed the length debug_assert).
  • a_time_shift_at_end_or_start_above_a_selector_passes_its_lineage_through.
  • a_qualified_group_key_binds_its_join_side (frontend-sql).
• PR 2's sign tests (memo_group_properties.rs, stage0's a_scan_reads_the_declared_column_signs) pass unchanged over the shared layout.
• cargo fmt --all --check; cargo clippy --workspace --all-features --tests -- -D warnings: clean. crate_graph (integration-tests) passes.

Test status

cargo test -p <crate> --all-features --no-fail-fast per crate at 58549f2. Compared by test name with PR 2 (#1230, 8095a51: 1121 passed / 189 failed / 9 ignored): no test that passes on PR 2 fails here. +7 passed are the new tests (types: the six column_lineage tests; frontend-sql: a_qualified_group_key_binds_its_join_side), −1 the deleted qualified_field_serde_roundtrip. Every failure is a todo!() of a later PR, with the same causes and counts as PR 2.

Crate	Passed	Failed	Ignored	Fails at
asap-types	199	0	0	—
asap-accuracy	60	0	0	—
asap-cost	31	1	0	#1207 PR 6: check θ, which the exported node carries: 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	175	0	0	—
asap-executor	210	63	1	#1207 PR 6: check θ, which the exported node carries: 32; #1207 PR 6: the Hydra grid's sizes are θ, which the exported node carries: 13; #1207 PR 6: each materialized node's retention_ms is its h_v, which Stage 3 sets: 8; #1207 PR 6: the weighted-frequency θ, which the exported node carries: 7; #1207 PR 6: check the weighted-frequency θ, which the exported node carries: 2; #1207 PR 3: discover targets over the memo groups, not node trees: 1
asap-planner	8	18	1	#1207 PR 3: discover targets over the memo groups, not node trees: 18
asap-devtools	37	56	3	#1207 PR 3: discover targets over the memo groups, not node trees: 48; #1207 PR 9: a group's node from its expression: 5; OTHER: assertion failed: output.status.success(): 2; #1207 PR 9: label a group by its expression: 1
asap-integration-tests	83	51	4	#1207 PR 3: discover targets over the memo groups, not node trees: 47; #1207 PR 6: each materialized node's retention_ms is its h_v, which Stage 3 sets: 3; OTHER: called Result::unwrap() on an Err value: Error("invalid type: sequence, expected string or map", line: 0, column: 0): 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	1127	189	9