Aggregate output-schema derivation is duplicated by hand between query_expr.rs and column_resolution.rs

[milindsrivastava1997]

Problem

There are two independent implementations of "what is the output schema of an Aggregate { by, aggs } node":

1. QueryExpr::output_schema_in's Aggregate arm — the canonical, general-purpose schema derivation used everywhere.
2. column_resolution::output_schema_for_aggregate — a separate function used only to resolve a HAVING predicate's column references against the aggregate's output (since HAVING refers to aggregate results, not the aggregate's input columns).

The second is explicitly documented as a hand-mirror of the first ("Mirrors QueryExpr::output_schema_in's Aggregate arm") rather than a shared call — and it already lacks a branch the original has: the per-series-reduction special case (is_range_child / AggIntent::is_per_series(), which detects rate/increase/*_over_time and preserves label columns instead of collapsing to by ++ aggs). Both implementations also independently re-derive the same value-column probing logic (find "value" or the sole non-by column) and the same unique_keys/closed construction.

Today this is latent, not user-visible: HAVING is SQL-only and the per-series-reduction shape is PromQL-only, so the code path where the two would actually disagree is never exercised by either lowerer. But nothing prevents that from changing, and no test asserts the two functions agree on a shared input.

Reproduction / demonstration

Side-by-side, the two functions:

crates/core/src/intent_algebra/query_expr.rs (output_schema_in, Aggregate arm) — includes:

let is_range_child = matches!(child.as_ref(), QueryExpr::TimeRange { .. });
if by.is_empty() && aggs.len() == 1 && (aggs[0].is_per_series() || is_range_child) {
    return Ok(per_series_reduction_schema(&in_schema, &aggs[0]));
}

crates/core/src/intent_algebra/column_resolution.rs (output_schema_for_aggregate) — has no equivalent branch; it goes straight to the by ++ aggs construction regardless of whether the aggregate is a per-series reduction:

pub fn output_schema_for_aggregate(
    input: &Schema,
    by: &[ColumnId],
    aggs: &[AggIntent],
    output_names: &[String],
) -> Schema {
    let mut out_cols: Vec<Column> = Vec::with_capacity(by.len() + aggs.len());
    for &id in by { ... }
    // no is_per_series / TimeRange check anywhere in this function
    ...

A unit test constructing a per-series Aggregate (e.g. by: [], aggs: [AggIntent::Rate], child: TimeRange { .. }) with a having predicate and comparing output_schema_for_aggregate's result against output_schema_in's result for the same node would fail today, demonstrating the divergence — no such test currently exists.