[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; the PRs merge into main together, after PR 9. Planning still panics at the first todo!() of Pass 1 (PR 3); see "Test status".

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), 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.

What

• GroupProperties::column_lineage: Vec<ColumnLineage>, one entry per output column in schema order, derived once in derive_group_properties from the input groups' lineage by the doc's per-operator table.
• Summary coverage names a selection's column by lineage: ColumnIdentity { source, column } is a source column. 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 rollup's group-key check compares 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 the new 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 doc: asap-primitive-schema.md §4.2.2 (worked example G1–G10 and "Which column a condition is on") uses lineage: latency.job ∈ {api}, t.region ∈ {us}. §2's table and the listings of Field, Schema and ColumnIdentity follow.
• Tracking: implement the optimization formulation (memo candidates, properties, edges y_{u,c,i}, MILP) as named code per stage #1207's Order table row 2b now uses the doc's names (SourceColumn / Computed).
• 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
qualifier copy in named_as_chain	types/src/ir/operator/tumbling_window_chain.rs	nothing
with_aggregate_qualifiers	stage1-pass1-per-computation-candidates/src/per_computation_candidates.rs	nothing: no qualifier to restore
ColumnIdentity::{table, name}	types/src/ir/properties/summary_coverage.rs	ColumnIdentity::{source, column}
private build_input_schema	same	private build_input (the input group, for its schema and lineage)
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 and sql_full_query_retains_project_and_binds_inner_aggregate	types, integration-tests	—

Pending approval (not in the doc or #1207)

Item	Where	Why
UnresolvedOp::Scan::qualifier: Option<String> (pub field)	frontend-common/src/unresolved.rs	The scope a SQL Scan qualifies its columns by, now that the schema has no qualifier. None for PromQL / MetricsQL.
ColumnIdentity field names source, column; test-support ColumnIdentity::new(source, column) (was new(table, name))	types/src/ir/properties/summary_coverage.rs	"ColumnIdentity is a SourceColumn": its fields are SourceColumn's.
Module asap_types::ir::properties::column_lineage	types/src/ir/properties/column_lineage.rs	Where ColumnLineage and its derivation live, next to summary_coverage.
pub(crate) Operator::<I>::output_column_lineage(&self, output, lineage_of), pub(crate) MemoGroupAsInput::column_lineage(); private Scope, Bound (front end), column_lineage (coverage)	types, frontend-common	Derivation; reading a group's lineage; the front end's binding scope.
ResolveError::Ambiguous message "matches several columns" (was "several unqualified columns")	types/src/ir/scalar/column_resolution.rs	No column is qualified any more.

How

• Derivation. output_column_lineage matches the operator: a Scan gives SourceColumn { source: table or metric, column: field name }; Filter, Sort, Limit, Dedup, TimeRange, TimeShift and the other row-selecting PromQL operators pass their input's lineage; a Project column item keeps its input column's lineage, an expression item is Computed; an Aggregate and a SummaryAgg follow aggregate_output_schema's layout (a by key keeps its lineage, a without reduction the kept labels, a per-series reduction every column but the value; measures and the state are Computed); a readout (SummaryEstimate, FinalizeExactAccumulator) keeps its state's plain columns in place; a Join concatenates; SetOp, Concat and a SummaryMerge keep a column's lineage only where their inputs agree; a BinaryOp keeps the left side's labels (and the right side's for or / group_right only where both agree). Anything else, or a rule whose result has not one entry per output column, gives Computed for every column.
• Coverage. of_summary_agg reads the lineage of the SummaryAgg's input (a group's stored lineage, or a temporary build's, derived from its input group's). residual lifts a conjunct only when its column's lineage is a SourceColumn no other column shares, and keys the selection by it. regrouped maps each rollup 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 SummaryAgg'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 is derived once per group: O(columns) per group (a lineage vector is copied from each input group). residual 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/ir/properties/column_lineage.rs

/// Where an output column's values come from (new type).
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ColumnLineage {
    /// Copied unchanged from a column of a source (a Scan's table or metric). // new
    SourceColumn { source: String, column: String },
    /// Computed by an operator: a measure, an expression, a summary state or readout. // new
    Computed,
}

crates/types/src/stages/memo.rs

/// 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,
    /// One entry per output column, in schema order. // new
    pub column_lineage: Vec<ColumnLineage>,
    /// What a summary-state group covers; `None` for any other group. // unchanged
    pub summary_coverage: Option<SummaryCoverage>,
    /// Whether two equal expressions may be one group. // unchanged
    pub shareable_when_equal: bool,
}

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

/// A column of the definition child's output, by its lineage's source column (stays).
pub struct ColumnIdentity {
    /// The table or metric a `Scan` reads. // new, was `table: Option<String>`
    pub(crate) source: String,
    /// The column's name there. // new, was `name: String`
    pub(crate) column: String,
}
impl ColumnIdentity {
    /// Test support. // was `new(table: Option<String>, name)`
    pub fn new(source: impl Into<String>, column: impl Into<String>) -> Self;
}

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
}

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 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:
let mut memo = Memo::default(); // the memo of the workload
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 lineage
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], accuracy: AccuracyTarget::max_error(0.01) }],
        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];
// its coverage selection is { flows.proto ∈ {tcp} } (ColumnIdentity::new("flows", "proto"))
let g5 = memo // SummaryEstimate[cardinality]: columns `src`, `cardinality`; [flows.src, Computed]
    .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 θ, no window
    logical_properties: LogicalProperties::new(&expression, None, None).unwrap(),
    expression,
    input_requirements: vec![LogicalRequirements::new(None)],
    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 (PR 3 had a stopgap that compared columns ignoring the qualifier, removed again for this PR).
• The resolver bound qualified names through the qualifier on each field.

After this PR

• Memo groups carry column_lineage; coverage, its merge check and its rollup 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.
• 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.

Wire format

The exported Schema JSON no longer has a "table" key on fields, and ColumnIdentity 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.

Verification

• New tests (types/tests/column_lineage.rs), one per doc example:
  • 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): the renaming Project keeps flows.src; a selection on the renamed area is flows.region; the rollup key check passes.
  • join_columns_of_one_name_have_distinct_lineage (example 3): sales.region vs users.region in lineage and in selections.
  • scan_predicates_lift_and_their_states_merge (example 4, in memo_summary_coverage.rs, now also asserting latency.job): latency.job ∈ {api} and {web} merge.
  • a_predicate_on_a_computed_column_stays_unlifted.
  • a_qualified_group_key_binds_its_join_side (frontend-sql): GROUP BY h.service over metrics m JOIN hosts h binds the right side's service.
• cargo fmt --all --check, cargo clippy --workspace --all-features --tests --examples -- -D warnings: clean.
• Changed crates' tests: asap-types, asap-frontend-common, asap-frontend-sql, asap-frontend-promql, asap-frontend-metricsql, asap-stage0-memo pass.

Test status

cargo test -p <crate> --all-features --no-fail-fast per crate at b5daf86 (the code commit; the head adds one assertion to a new test and the doc). The +4 passed are the new tests (types +3 net: four new, qualified_field_serde_roundtrip deleted; frontend-sql +1). Every failure is a todo!() of a later PR, with the same causes and counts as PR 2. PR 2 (#1230, e710906) for comparison: 1136 passed / 229 failed / 9 ignored.

Crate	Passed	Failed	Ignored	Fails at
asap-types	195	0	0	—
asap-accuracy	75	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	175	0	0	—
asap-stage0-memo	13	0	0	—
asap-stage1-pass1-per-computation-candidates	12	1	0	the guarantee of θ from the candidate's LogicalProperties::parameters (PR 4): 1
asap-stage1-pass2-reuse-candidates	0	0	0	—
asap-stage2-physical-optimization	0	0	0	—
asap-stage3-plan-selection	3	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; the view's coverage from its memo group's GroupProperties::summary_coverage (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	9	19	1	discover targets over the memo groups, not node trees (PR 3): 19
asap-devtools	38	58	3	discover targets over the memo groups, not node trees (PR 3): 50; 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	84	87	4	discover targets over the memo groups, not node trees (PR 3): 83; the view's coverage from its memo group's GroupProperties::summary_coverage (PR 6): 3; the stale fixture (old wire format; PR 9): 1
Total	1140	229	9