feat(asap-aware-mapping): AvgToSumOverCountStrategy semantic rewrite (#253)

[zzylol]

Closes #253. Part of #33. Stacked on #259 (issue #251) — do not merge before that one.

What

A ReplacementStrategy — AvgToSumOverCountStrategy — implementing the
"semantic-equivalent rewriting" degree of freedom from
docs/asap_aware_mapping.md: rewrites Aggregate{ measures: [Avg{col}], .. }
into Aggregate{ measures: [Sum{col}, Count], .. } |> Project{ sum/count re-divided under the original avg column name }.

Why

boundary::implementation_for_with dispatches Avg straight to
Implementation::PassThrough — a bare avg node has no summary
realization at all, so it can never be a SharedSubtreeStrategy target.
Sum and Count are both ordinary mergeable accumulators
SharedSubtreeStrategy (and a future sketch-family search) already know
how to reuse across a workload; this rewrite reshapes avg into those
pieces without changing what the query computes.

Shape

• matches: a single Avg{col} measure, no HAVING (the same
  bindable shape bind::implement_tree_with requires — mirrors
  SketchFamilyStrategy's own bindable_intent check), further
  restricted to an ordinary by(...) reduction. without(...) and
  Reduction::PerEntity are excluded, for real correctness reasons, not
  style: PerEntity is single-measure by construction
  (aggregate_output_schema's own debug_assert, which this rewrite's
  second measure would violate), and without(...) leaves the aggregate's
  own output schema open (closed: false) while the wrapping Project
  this rewrite always produces forces closed: true — a schema drift
  this strategy declines to introduce.
• replacements: exactly one Replacement::Rewrite candidate — no
  ranking, no "is it worth it" heuristic. Per the issue's own framing:
  earlier drafts needed a manual before/after-CSE cost comparison to
  decide whether rewriting helps; with ReplacementStrategy's
  exhaustive-candidate shape in place that's unnecessary — both the
  original Avg and the rewritten Sum/Count form are just two
  candidates a future cost-based search (issue asap-aware-mapping: Cascades/Volcano-style candidate-plan search engine over ReplacementStrategy #252) can compare, not
  something this strategy pre-decides.
• The Project's sum / count division is explicitly Cast to
  Float64 so the rewritten avg column matches
  AggIntent::Avg::output_column's own (Float64, nullable: false)
  shape regardless of the summed column's own type — an uncast
  Int64 / Int64 division would otherwise silently retype the column,
  which a test in this PR pins down.

Before / after planner output

Example query:

SELECT job, avg(cpu) FROM metrics GROUP BY job

Before — avg has no summary realization at all (implementations_for_with dispatches it to PassThrough), so the only candidate keeps the whole aggregate logical — nothing to share or sketch:

replacements(avg_agg) = [
    Summary(KeepPreAsap(Aggregate {
        reduction: by([job]),
        measures: [Avg { col: None }],
        child: Scan(metrics),
    })),
]

After — AvgToSumOverCountStrategy adds a second candidate: avg reshaped into sum/count under the same grouping, re-divided back by a wrapping Project — two ordinary mergeable accumulators a future sketch-family search / CSE can now actually reuse, where a bare avg node never could:

replacements(avg_agg) = [
    Summary(..),  // unchanged, still offered
    Rewrite(
        Project {
            cols: [Column(0), Cast(Column(1) / Column(2), Float64) AS "avg"],
            child: Aggregate {
                reduction: by([job]),
                measures: [Sum { col: None }, Count { accuracy: Exact }],
                child: Scan(metrics),
            },
        },
    ),
    // rationale: "avg has no summary realization at all (replacement::implementations_for_with
    // dispatches it to PassThrough) ... rewriting it into sum/count ... computes the same result
    // from two ordinary mergeable accumulators SharedSubtreeStrategy (and a future sketch-family
    // search) can actually reuse across the workload"
]

Tests

Positive: an ungrouped and a grouped Avg aggregate both match and
rewrite correctly. Negative: multi-measure, HAVING-bearing, non-Avg
intents, without(...) grouping, Reduction::PerEntity, and non-Aggregate
nodes all correctly don't match. Schema round-tripping:
original.output_schema() == rewritten.output_schema() holds exactly for
an ungrouped aggregate (including an explicit-output-name-override
variant); for a grouped aggregate, column-level equality is asserted
directly, with an explicit test documenting the one known gap (Project
always resets unique_keys, a pre-existing property of every
Project-wrapped rewrite in this IR, not specific to this rewrite —
unique_keys also isn't consumed by anything yet per Schema's own doc
comment). A dedicated test with an Int64-typed summed column exercises
the Cast directly.

lib.rs

Added only pub mod rewrite; — kept minimal since sibling PRs #252/#254/#256
touch the same file from the same base branch in parallel.

Checks

cargo build --workspace --all-targets, cargo test --workspace,
cargo fmt --all -- --check, and
cargo clippy --workspace --all-targets --all-features -- -D warnings all
pass clean.

🤖 Generated with Claude Code