feat(ir): read summary coverage source and population from the subtree

[zzylol]

Closes #570.

Why

#567 records each summary state's coverage (source, time range, population), and #560 allows a SummaryMerge only when its inputs' coverage is provably disjoint. But a SummaryAgg's coverage is a declaration that nothing checks against the state's own subtree, so a wrong one merges silently.

Before this PR:

A = SummaryAgg(filter: region = 'us')   declared {region: eu} × [0,1)   ← wrong, A holds US data
B = SummaryAgg(filter: region = 'us')   declared {region: us} × [0,1)

SummaryMerge(A, B) → accepted ("eu" ≠ "us" looks disjoint)
merged state       → every US observation in [0,1) counted twice

After this PR: A.with_coverage(...) fails with PopulationMismatch, because A's filters say {region: us}. Producers no longer write coverage by hand: SummaryCoverage::for_summary(&agg, time_ms) builds the declaration that passes.

What is read from the subtree

Part	Where it comes from	Checked?
source	the one Scan below the SummaryAgg (OperatorNode::scanned_source)	yes, must be equal
population	field = 'text' conjuncts of SummaryAgg.filter, Filter nodes and Scan.predicates, on a path that passes only through Filter and TimeRange (OperatorNode::derived_population)	yes, every region must equal it exactly
time_ms	the producer: a pane's bounds are a materialization fact, not in the logical plan (#601 sets them)	no, still declared

• Exact equality, not a subset: declaring fewer conditions than the filters apply overstates the data, and declaring more lets a merge with another value pass.
• Columns are named from the Scan operator's own schema. Filter and TimeRange keep their input columns, so every column id on the path is a Scan column. A node on the path that renames a field (with_schema) makes the population unknown, because a name could then stand for a different source column.
• Unknown population (another operator on the path, a renamed field, or a predicate that is not one text equality per field: >, regex, IN, OR, region = 'us' AND region = 'eu'): the declaration must be unrestricted, and SummaryMerge accepts such states only when every input has the same input and filter below its SummaryAgg (time panes of one computation). Otherwise it fails with UnprovenPopulation. This also stops a merge of states that read different columns under one name.
• Rebuilds keep coverage: OperatorNode::map_children keeps a SummaryAgg's declared time bounds and reads its source and population again from the new input, so rebuilding a merge (e.g. in canonicalize) keeps working.
• Several sources: a SummaryAgg whose input reads two sources (a join, or an IN subquery over another table) has no single source. It is valid without coverage and cannot be merged (UnknownInput).

Key code interfaces

crates/types/src/ir/summary_coverage.rs

impl SummaryCoverage {
    /// Source and population from `summary`'s subtree, over `time_ms`.
    /// None unless `summary` is a SummaryAgg reading exactly one source.
    pub fn for_summary(summary: &OperatorNode, time_ms: Option<Range<i64>>) -> Option<Self>;
}

impl OperatorNode {
    pub fn scanned_source(&self) -> Option<Source>;
    /// None when the path holds another operator, renames a field, or a
    /// predicate is not `field = 'text'`. Empty map = no restriction.
    pub fn derived_population(&self) -> Option<BTreeMap<String, String>>;
}

pub enum CoverageError {
    // ... existing variants
    PopulationMismatch,   // declared population differs from the filters
    UnprovenPopulation,   // merge of unknown-population states with different inputs
}

Tests

summary_coverage_examples.rs now builds every #560 example with a real region = '…' filter matching its declared population. summary_coverage_derivation.rs adds:

Test	Behaviour
population_combines_every_filter_on_the_path	SummaryAgg.filter, a Filter node and Scan.predicates, through a TimeRange, give {job, region, tier}
unsupported_predicates_leave_the_population_unknown	>, OR, and one field equal to two values
renamed_field_does_not_fake_a_disjoint_population	regression: a renamed tier column cannot pass as region
renamed_update_column_does_not_merge	regression: Named("latency") reading size on a renamed scan is rejected
nested_merges_compose	merges of merges; unknown-population inputs stay rejected through nesting
rebuilding_keeps_coverage	regression: map_children keeps time bounds and re-reads population; a merge rebuilds
summary_over_two_sources_has_no_coverage	regression: a join of two sources validates without coverage and cannot merge

The four regression tests fail on the code before the fix commit.

Out of scope

• Checking time bounds: TimeRange holds a relative duration, so absolute pane bounds cannot be checked in the logical IR.
• Reading populations through Project or Join; widen the path when a real plan needs it.

Stack and validation

Order: #645 (merged) → #567 (merged) → #560 → this PR → #539 → #540 → #541 → #542 → #543.

#542 builds planned coverage with for_summary. This PR passes the workspace tests (1,645), cargo fmt --check, and workspace/all-target/all-feature Clippy with warnings denied. The fixes come from an independent review of #560 and #646.

🤖 Generated with Claude Code

[Selvomega]

The naming of time_ms seems not good. I assume you want conceptually a "range" here?

[zzylol]

yes, I will rename to "time_range"

[Selvomega]

Bug here: If derived_population() returns None, it will be converted to the default legal value of population.
The problem is in derived_population() semantic, None means I cannot handle this population so it should not be used. While this treacherous unwrap_or_default() silently convert this into a legal value!

Spotting this bug actually makes me feel better: At least it proves reading code is still somehow useful.

[Selvomega]

Same. Check if there is bug here

[Selvomega]

Seemingly a bug. Check other comments related

[Selvomega]

This function is also treacherous and shady. It is called for_summary but practically only works for SummaryAgg nodes. Some problems in PR #560 is related to this.