feat(asap-aware-mapping): CSE stage 4 — Volcano/Cascades cost-based sharing (#237, #223, #212)

[zzylol]

Summary

Implements #223's stage 4 for real: CostModel::cse_share_decision decides
whether a CSE-detected shared subtree is actually worth sharing, via a real
Volcano/Cascades-style cost comparison rather than a fixed rule.

The design discussion for #237 (rule-based vs. cost-based framework) now
lives in docs/cse-cost-model-decision.md, not in code comments. See that
doc for the full reasoning: why cost-based, why not a full plan-search
engine, and the layering constraint that forces CSE detection
(asap_types::pre_asap::cse) to stay cost-agnostic while the cost-aware
decision lives downstream in asap-aware-mapping.

Core algorithm — NOT dynamic programming

Worth being explicit about this: nothing here is a DP recurrence over a
state space. It's two separate, simpler things — a hash-consing pass
(pre-existing, #235) and a per-candidate greedy cost comparison (this PR).

Detection (share_common_subtrees, already merged in #235, unchanged
by this PR) — bottom-up hash-consing / value-numbering:

function intern_bottom_up(node):
    node.children = [intern_bottom_up(c) for c in node.children]   # children first
    return table.intern(node)

function InternTable.intern(node):
    h = structural_hash(node)                    # coarse filter only
    if node.has_unique_key():
        for candidate in buckets[h]:
            if candidate == node:                 # PartialEq — the real decision
                return candidate                   # reuse: this IS the sharing
    buckets[h].push(Rc::new(node))
    return buckets[h].last()

Sharing decision (this PR, bind::implement_workload_with +
CostModel::cse_share_decision) — a single left-to-right pass with a
pre-count, then one independent cost comparison per shared candidate,
cached so every later occurrence reuses the same answer:

function implement_workload_with(roots, cost_model):
    # Pass 1 (O(R)): true total consumer count per shared Rc pointer,
    # known up front rather than growing as roots are processed.
    consumer_count = count_occurrences_by_pointer(roots)

    memo = {}   # ptr -> (bound_summary, decision)
    for (id, expr) in roots:
        ptr = pointer_of(expr)
        if ptr in memo:
            (bound, decision) = memo[ptr]
            yield (id, bound if decision == Share else implement_tree(expr, cost_model))
        else:
            bound = implement_tree(expr, cost_model)
            n = consumer_count[ptr]
            decision = Share if n <= 1 else cost_model.cse_share_decision(
                CseCandidate(subtree=expr, bound_summary=bound, consumer_count=n)
            )
            memo[ptr] = (bound, decision)
            yield (id, bound)

function cse_share_decision(candidate):
    recompute_total = cse_recompute_cost(candidate) * candidate.consumer_count
    shared = cse_shared_maintenance_cost(candidate)
    return Share if shared <= recompute_total else RecomputeIndependently

Why not DP: each CseCandidate's decision is evaluated independently —
no candidate's Share/RecomputeIndependently choice depends on, or
constrains, any other candidate's choice, and there's no recurrence
combining subproblem solutions toward one globally-optimal objective (the
shape OPT[i] = f(OPT[i-1], …) takes). The memo above is answer-caching
for efficiency (decide once per unique shared subtree, not once per
occurrence) — necessary because a naive per-occurrence decision could
disagree with itself across occurrences of the same subtree, but caching a
computed answer isn't sufficient by itself to make an algorithm DP. This is
also why a full Volcano/Cascades-style plan-enumeration search was
explicitly rejected in docs/cse-cost-model-decision.md: this repo has no
DP-based plan memo/search space anywhere, and the binary "share or don't"
question this PR answers doesn't need one — a direct cost comparison is
the whole algorithm.

What changed in code (not just docs)

• CostModel gains:
  • CseCandidate — a detected, legality-gated shared subtree with its
    bound SummaryNode and true consumer_count.
  • ShareDecision — Share or RecomputeIndependently.
  • cse_recompute_cost / cse_shared_maintenance_cost — overridable cost
    hooks, with concrete default heuristics.
  • cse_share_decision — a real default body: share iff the estimated
    shared-maintenance cost is no greater than the estimated total
    recompute cost (cse_recompute_cost * consumer_count). A genuine cost
    comparison, not a hardcoded rule.
• bind::implement_workload_with now computes each shared subtree's true
  consumer_count via a whole-workload pre-pass, decides ShareDecision
  once per shared subtree, and applies it consistently to every occurrence.
• cse_recompute_cost's default (default_cse_recompute_cost) is
  DAG-aware, not tree-shaped: it calls the new
  asap_types::pre_asap::cse::dag_node_count, which counts unique nodes
  reachable from a subtree (deduplicated by Rc pointer identity), rather
  than a naive serde_json serialization length — a CseCandidate's
  subtree is, by definition, something CSE already found sharing in, so
  it's generally a DAG, not a tree, and a naive full serialization would
  re-count any descendant it already shares internally once per parent
  that references it.

Tests

• cost_model.rs: unit tests pinning the default cost comparison's
  behavior in both directions, a test confirming a custom CostModel's
  overridden cost hooks are actually consulted, and a test proving the
  DAG-vs-tree fix directly (a Join with two independent Scan children
  costs 3 unique nodes; the same Join with both children pointing at one
  shared Scan costs 2, not 3).
• cse.rs: unit tests for dag_node_count — the naive count when nothing
  is shared, correct dedup of an internally-shared subtree, and correct
  dedup across two workload roots sharing one subtree.
• bind.rs: a test confirming a CostModel that declines sharing produces
  genuinely independent (non-Rc::ptr_eq) SummaryNodes.
• All pre-existing CSE tests, including the integration tests'
  DefaultCostModel-sharing assertions, pass unchanged — default cost
  weights were calibrated against a real lowered query so today's sharing
  behavior for realistic subtrees is preserved.

Test plan

• cargo build --workspace --all-targets
• cargo test --workspace (444 passed, 0 failed)
• cargo fmt --all -- --check
• cargo clippy --workspace --all-targets --all-features -- -D warnings
• cargo doc --workspace --no-deps (no new broken-intra-doc-link warnings)

Closes #237. Closes #223. Closes #212.

🤖 Generated with Claude Code

[zzylol]

Fixed per feedback: after CSE, a CseCandidate's subtree is generally a DAG, not a tree (it only exists because something got shared) — but default_cse_recompute_cost measured it via serde_json::to_string(subtree).len(), and Rc's Serialize impl re-serializes the pointee's value at every occurrence, not once by identity. That silently over-counted any descendant the subtree already shares internally (e.g. a single-query x op x case, both branches collapsed onto one Rc), once per parent that reaches it.

Added asap_types::pre_asap::cse::dag_node_count(root) -> usize: counts unique nodes reachable from root, deduplicated by Rc pointer identity, using the same operator-child traversal scope share_common_subtrees itself uses. default_cse_recompute_cost now calls this instead of serde_json.

Recalibrated default_cse_shared_maintenance_cost's UNIT constant (60.0 → 1.0) for the new magnitude — node counts are small integers, not byte lengths. Verified this preserves existing default-sharing behavior: all pre-existing CSE tests, including the integration tests' DefaultCostModel-sharing assertions, pass unchanged.

New test demonstrating the fix directly: a Join with two independent Scan children costs 3 (unique nodes), the same Join with both children pointing at one shared Scan costs 2, not 3.

cargo build/test/fmt/clippy/doc all pass clean (444 tests, 0 failed).

[zzylol]

@milindsrivastava1997 FYI for your candidate selection model of asap mapping