Design: pre-ASAP structural CSE via bottom-up hash-consing over Rc<QueryExpr>

[zzylol]

Summary

Design for the pre-ASAP structural CSE pass whose placement was settled in
#222 (primary pass in asap-types::pre_asap, before implement; a
secondary, narrower pass can run post-ASAP over already-implemented
SummaryNodes). This issue is the algorithm and the landing plan; #212 is
the parent tracking issue, #222 is "where it runs."

Representation prerequisite already landed: QueryExpr<C>'s child
fields are now Rc<QueryExpr<C>> instead of Box<QueryExpr<C>> (branch
refactor/pre-asap-rc-sharing, mechanical, zero behavior change — 403
tests pass before and after). That change only adds the capability to
share a subtree (the same Rc referenced from more than one parent); it
does not make any sharing happen. This issue is the pass that does.

Algorithm: canonicalize first, then bottom-up hash-consing

CSE only runs on an already-resolve_root'd (parsed, bound, canonicalized)
tree — structural matching is meaningless before canonicalization has
converged semantically-equivalent queries onto one shape (docs/pre-asap-ir.md
design principle 3; median(latency) vs approx_percentile_cont(latency, 0.5)
already lower to an identical AggIntent::Quantile today, per
sql_lowering.rs's median_is_the_same_intent_as_an_explicit_half_percentile).

Classic hash-consing / value-numbering, adapted to QueryExpr:

// crates/types/src/pre_asap/cse.rs (new)

use std::collections::HashMap;
use std::rc::Rc;

use super::names::QueryId; // or equivalent
use super::query_expr::QueryExpr;

struct InternTable {
    // hash is a coarse filter only — see "Correctness" below.
    buckets: HashMap<u64, Vec<Rc<QueryExpr>>>,
}

impl InternTable {
    /// Bottom-up: intern every child first, then this node. Children are
    /// already `Rc`s by construction, so a parent's hash/equality
    /// naturally incorporates whether its children were themselves shared.
    fn intern(&mut self, node: QueryExpr) -> Rc<QueryExpr> {
        let h = structural_hash(&node); // lifted from dag_export.rs, not reimplemented
        let bucket = self.buckets.entry(h).or_default();
        if let Some(existing) = bucket.iter().find(|e| e.as_ref() == &node) {
            return Rc::clone(existing); // hash matched AND PartialEq matched
        }
        let rc = Rc::new(node);
        bucket.push(Rc::clone(&rc));
        rc
    }
}

/// Share structurally-identical subtrees across a workload's query roots.
/// Every root's *value* is unchanged (`PartialEq`-equal to its input) —
/// only its internal `Rc` structure may now alias another root's.
pub fn share_common_subtrees(
    roots: Vec<(QueryId, QueryExpr)>,
) -> Vec<(QueryId, Rc<QueryExpr>)> {
    let mut table = InternTable { buckets: HashMap::new() };
    roots
        .into_iter()
        .map(|(id, expr)| (id, intern_bottom_up(&mut table, expr)))
        .collect()
}

intern_bottom_up recurses into each QueryExpr variant's children,
interns each child (getting back a possibly-shared Rc), rebuilds the node
with those Rc children, then interns the rebuilt node itself.

Correctness: hash is a filter, PartialEq is the decision

This is the one non-negotiable rule. A false positive here — two
subtrees wrongly judged shareable — is a wrong query answer, not a missed
optimization: two different queries would read each other's data. hash
(DefaultHasher/SipHash, no collision-freedom guarantee) may only narrow
the candidate set; InternTable::intern's .find(|e| e.as_ref() == &node)
PartialEq check is what actually decides sharing, every time, no
exceptions for "the hash probably didn't collide."

This also means the pass is safe by construction against the case #212
flagged as a real historical bug (issue #115): AggIntent::Quantile now
carries its input column and its AccuracyTarget, both PartialEq fields,
so Quantile(x, 0.99, ε=0.01) and Quantile(x, 0.99, ε=0.001) — or
Quantile(x, ..) vs Quantile(y, ..) — are never merged. This is
intentionally conservative: it only recognizes exact structural matches,
not "a stricter-accuracy summary could also answer a looser request." That
subsumption question already has a documented, deliberately-unfilled home
(asap-aware-mapping::boundary::Matcher) — CSE here does not attempt it.

Legality: gated by Schema::unique_keys

Structural equality alone is necessary but not sufficient — a subtree with
no provable row identity (Schema::unique_keys empty /
Schema::is_reusable() false) is not hoisted even if structurally
identical to another, matching the rule the (now-deleted) prior CSE attempt
already encoded and the doc comment unique_keys already carries ("feeds
CSE's producer-sharing legality check").

Where this plugs into the pipeline

resolve_root (parse → bind → canonicalize)   — per query, unchanged
        │
        ▼
share_common_subtrees(roots)                  — NEW, workload-level, this issue
        │
        ▼
implement_workload(shared_roots)              — NEW, asap-aware-mapping, memoized

asap-aware-mapping::bind::implement_tree runs per query today. Its
workload counterpart should memoize on Rc identity (Rc::as_ptr), not
re-derive sharing from scratch:

pub fn implement_workload(
    roots: Vec<(QueryId, Rc<QueryExpr>)>, // already through share_common_subtrees
) -> Vec<(QueryId, Rc<SummaryNode>)> {
    let mut memo: HashMap<*const QueryExpr, Rc<SummaryNode>> = HashMap::new();
    roots.into_iter().map(|(id, expr)| (id, implement_memoized(&expr, &mut memo))).collect()
}

Two QueryExpr roots sharing the same Rc<QueryExpr::Aggregate{..}> after
share_common_subtrees collapse to one Rc<SummaryNode::SummaryAgg{..}> —
a summary genuinely built once — with no second structural-equality pass
needed at the post-ASAP layer for this kind of sharing.

What this pass does not cover (see #222)

Two structurally different pre-ASAP nodes — Quantile(x, 0.99) and
Quantile(x, 0.95) — can still legitimately share one built sketch,
read out twice. That's invisible to pre-ASAP structural equality by
construction (different q) and is exactly the secondary, post-ASAP pass
#222 describes: keyed on (SummaryKind, params, source-subtree) once both
are already bound, not on QueryExpr equality.

Landing plan (avoid repeating why the old pass got deleted)

The original asap-plan::cse::dedupe_subtrees (QueryExpr::Ref /
CseWorkloadPlan) was removed in #192 for being unwired dead code — "no
producer exists" — not for a wrong algorithm. Land in this order so that
never becomes true again:

1. share_common_subtrees + InternTable, in asap-types::pre_asap::cse.
   Unit tests re-covering the old pass's pinned cases (distinct-column
   Quantiles don't merge; no unique_keys → no merge; median /
   approx_percentile_cont(.,0.5) do merge).
2. implement_workload in asap-aware-mapping, wired to call (1) — a real
   caller before anything else, even if only exercised by an integration
   test initially.
3. dag_export's structural_hash becomes literally the same function (1)
   uses, instead of a parallel reimplementation; tools/dag-viewer's
   "shared subtree" highlighting starts reflecting (1)'s real output
   instead of the current hash-proxy (see tools/dag-viewer/README.md's
   own "this is a proxy, not real CSE" caveat).
4. Only then: wire workload-level CSE credit into CostModel
   (asap-aware-mapping::cost_model's module doc already names this as
   planned).

Scope / open questions carried from #212

• Single-query CSE (a repeated sub-expression within one query, e.g.
  the same grouped Aggregate referenced twice on two BinaryOp branches)
  falls out of the same share_common_subtrees call for free — a workload
  of size 1 still interns bottom-up within that one tree. No separate
  mechanism needed.
• Not covered here: cross-batch / persistent sharing (reusing a summary
  built for a previous workload run) — that's deployment/materialized-
  view territory (l3-summary-bound-ir.md's deployment-supplied
  find_candidates), out of scope for this planning-time pass.

Related

• New feature: add common sub expr elimination #212 — parent "add CSE" issue
• CSE detection should run in pre-ASAP IR; post-ASAP can run a secondary pass #222 — placement decision (pre-ASAP primary, post-ASAP secondary)
• refactor/pre-asap-rc-sharing branch — the Box→Rc prerequisite,
  mechanical, tests green (403/403, no clippy warnings)
• crates/types/src/dag_export.rs (structural_hash, to be shared with this pass)
• crates/types/src/pre_asap/schema.rs (unique_keys / is_reusable)
• crates/asap-aware-mapping/src/cost_model.rs (module doc names
  "workload-level CSE credit" as planned)