@@ -85,33 +85,39 @@ pub enum ShareDecision {
 }
 
 /// Default [`CostModel::cse_recompute_cost`]: a structural-size proxy — the
-/// length of `subtree`'s canonical JSON serialization (the same style of
-/// serialization `asap_types::pre_asap::cse`'s own structural hashing uses
-/// to compare subtrees). Cheap to compute, needs no tree traversal of its
-/// own, and scales with the subtree's real structural complexity — a genuinely
-/// tiny leaf costs little to recompute, a deep multi-join subtree costs a
-/// lot. A deployment with real per-row/per-update cost knowledge should
+/// number of *unique* nodes in `subtree`'s DAG
+/// ([`asap_types::pre_asap::cse::dag_node_count`], the same module this
+/// candidate's sharing was detected in). Deliberately **not** a raw
+/// `serde_json` serialization length: after CSE, `subtree` is generally a
+/// DAG, not a tree (a `CseCandidate` only exists because something got
+/// shared), and a naive full serialization re-serializes — over-counts —
+/// any descendant `subtree` already shares internally, once per parent
+/// that references it, instead of once for the whole DAG. `dag_node_count`
+/// dedupes by `Rc` pointer identity, so it charges each unique node's
+/// contribution exactly once regardless of how many places within
+/// `subtree` reference it. Cheap to compute (one pass, no serialization),
+/// and still scales with real structural complexity — a genuinely tiny
+/// leaf costs little to recompute, a deep multi-join subtree costs a lot.
+/// A deployment with real per-row/per-update cost knowledge should
 /// override [`CostModel::cse_recompute_cost`] instead of relying on this.
 pub fn default_cse_recompute_cost(subtree: &QueryExpr) -> f64 {
-    serde_json::to_string(subtree)
-        .map(|s| s.len() as f64)
-        .unwrap_or(1.0)
-        .max(1.0)
+    asap_types::pre_asap::cse::dag_node_count(subtree) as f64
 }
 
 /// Default [`CostModel::cse_shared_maintenance_cost`]: a small
 /// per-[`SummaryFamilyType`] weight, scaled to the same order of magnitude
-/// as [`default_cse_recompute_cost`]'s typical output, reflecting that
-/// families differ in how expensive they are to keep *continuously
-/// updated* for the life of a workload — an exact accumulator is the
-/// cheapest (an O(1) merge), sketches/samples cost more (a whole data
-/// structure to update per new row), wavelets/fitted models cost the most
-/// (coefficient/parameter maintenance). These weights are illustrative, not
-/// measured — a deployment with real memory/update-cost numbers should
-/// override [`CostModel::cse_shared_maintenance_cost`] instead of relying on
-/// this table.
+/// as [`default_cse_recompute_cost`]'s typical output (a small node
+/// count, not a byte length), reflecting that families differ in how
+/// expensive they are to keep *continuously updated* for the life of a
+/// workload — an exact accumulator is the cheapest (an O(1) merge),
+/// sketches/samples cost more (a whole data structure to update per new
+/// row), wavelets/fitted models cost the most (coefficient/parameter
+/// maintenance). These weights are illustrative, not measured — a
+/// deployment with real memory/update-cost numbers should override
+/// [`CostModel::cse_shared_maintenance_cost`] instead of relying on this
+/// table.
 pub fn default_cse_shared_maintenance_cost(family: &SummaryFamilyType) -> f64 {
-    const UNIT: f64 = 60.0;
+    const UNIT: f64 = 1.0;
     let weight = match family {
         SummaryFamilyType::Plain(_) => 1.0,
         SummaryFamilyType::ExactAggregate(..) => 1.0,
@@ -423,6 +429,49 @@ mod tests {
         assert!(default_cse_recompute_cost(&nested) > default_cse_recompute_cost(&leaf));
     }
 
+    /// The DAG-awareness this proxy exists for: a subtree that internally
+    /// re-references one shared descendant (e.g. after single-query CSE,
+    /// `x op x` collapsing both branches onto one `Rc`) must cost the same
+    /// as if that descendant only appeared once — not double, the way a
+    /// naive tree-shaped size measure (a full serialization, or an
+    /// identity-blind recursive walk) would count it.
+    #[test]
+    fn default_recompute_cost_does_not_double_count_an_internally_shared_descendant() {
+        use asap_types::pre_asap::expr_ir::ScalarValue;
+        use asap_types::pre_asap::query_expr::{JoinKind, Predicate};
+
+        let true_pred = || {
+            Predicate(std::rc::Rc::new(QueryExpr::Literal(ScalarValue::Boolean(
+                true,
+            ))))
+        };
+        let shared_leaf = std::rc::Rc::new(scan());
+        let no_sharing = QueryExpr::Join {
+            kind: JoinKind::Inner,
+            pred: true_pred(),
+            left: std::rc::Rc::new(scan()),
+            right: std::rc::Rc::new(scan()),
+        };
+        let with_sharing = QueryExpr::Join {
+            kind: JoinKind::Inner,
+            pred: true_pred(),
+            left: std::rc::Rc::clone(&shared_leaf),
+            right: std::rc::Rc::clone(&shared_leaf),
+        };
+        assert_eq!(
+            default_cse_recompute_cost(&no_sharing),
+            3.0,
+            "no sharing: Join + 2 independent Scans = 3 unique nodes"
+        );
+        assert_eq!(
+            default_cse_recompute_cost(&with_sharing),
+            2.0,
+            "internal sharing: Join + 1 shared Scan (referenced twice) = \
+             2 unique nodes, not 3 — a tree-shaped size measure would \
+             wrongly charge for the shared Scan twice"
+        );
+    }
+
     #[test]
     fn default_shared_maintenance_cost_orders_families_cheapest_to_priciest() {
         let exact = default_cse_shared_maintenance_cost(&SummaryFamilyType::ExactAggregate(
@@ -468,10 +517,10 @@ mod tests {
                     family: "gaussian_mixture".into(),
                 },
             )),
-            // A single, cheap-to-recompute leaf against an
-            // expensive-to-maintain family: maintenance should dominate
-            // (scan()'s recompute cost, 264, is well under StatModel's
-            // maintenance cost, 360).
+            // A single, cheap-to-recompute leaf (scan() alone is 1 DAG
+            // node, recompute_total = 1) against an expensive-to-maintain
+            // family (StatModel, maintenance cost 6.0): maintenance should
+            // dominate.
             consumer_count: 1,
         };
         assert_eq!(
 
@@ -164,6 +164,110 @@ fn structural_hash(node: &QueryExpr) -> u64 {
     hasher.finish()
 }
 
+/// Count of *unique* nodes reachable from `root`, deduplicated by `Rc`
+/// pointer identity (`Rc::as_ptr`) — the real size of the DAG rooted at
+/// `root`, not a tree-walk count.
+///
+/// After [`share_common_subtrees`] runs (or even before it, for a tree a
+/// front end already built with internal `Rc` sharing — e.g. re-running
+/// CSE, or a single-query repeated subexpression), `root` is generally a
+/// **DAG**, not a tree — that is this whole module's premise. Anything that
+/// walks `root` as if every reference were a fresh subtree (a naive
+/// recursive walk with no identity tracking, or a naive full
+/// `serde_json` serialization — `Rc`'s `Serialize` impl serializes the
+/// pointee's *value* at every occurrence, it does not dedupe by identity)
+/// re-visits/re-counts an already-shared descendant once per parent that
+/// references it, over-counting relative to the actual work of holding it
+/// in memory or recomputing it once. This function is the DAG-correct
+/// alternative: each unique node is counted exactly once, regardless of
+/// how many places within `root` reference it.
+///
+/// `pub` so cost-aware callers outside this crate (e.g.
+/// `asap_aware_mapping::CostModel::cse_recompute_cost`'s default) have a
+/// DAG-correct structural-size proxy available, instead of reaching for
+/// something tree-shaped like a raw serialization length.
+///
+/// Same operator-child traversal scope as [`share_common_subtrees`] itself
+/// (see the module doc's "Algorithm" section, and this module's private
+/// `rebuild_children`) — a scalar subexpression embedded in a wrapper
+/// position (`Predicate`, `ProjectItem.expr`, `Aggregate.having`, …) is not
+/// separately visited, matching this module's own stated scope; it's
+/// counted as part of its owning operator node, the same node
+/// `rebuild_children` treats as a single opaque leaf for interning
+/// purposes.
+pub fn dag_node_count(root: &QueryExpr) -> usize {
+    let mut seen: std::collections::HashSet<*const QueryExpr> = std::collections::HashSet::new();
+    count_unique(root, &mut seen)
+}
+
+/// One node's own contribution (`1`) plus each *not-yet-seen* operator
+/// child's contribution — exhaustive over every `QueryExpr` variant,
+/// enumerating the same fields [`rebuild_children`] does (kept as a
+/// separate, read-only traversal rather than threaded through
+/// `rebuild_children` itself, since that function consumes and rebuilds
+/// its input while this one only ever reads it).
+fn count_unique(node: &QueryExpr, seen: &mut std::collections::HashSet<*const QueryExpr>) -> usize {
+    use QueryExpr::*;
+
+    /// Visit one `Rc`-held child: counts (and recurses into) it only the
+    /// first time its pointer is seen, `0` on every later occurrence —
+    /// this is the actual dedup step.
+    fn visit(
+        child: &Rc<QueryExpr>,
+        seen: &mut std::collections::HashSet<*const QueryExpr>,
+    ) -> usize {
+        if seen.insert(Rc::as_ptr(child)) {
+            count_unique(child, seen)
+        } else {
+            0
+        }
+    }
+
+    1 + match node {
+        Scan { .. } | PromqlScalar(_) | QueryTimestamp => 0,
+        PromqlVectorFromScalar(c) | PromqlScalarFromVector(c) => visit(c, seen),
+        PromqlRelabel { child, .. }
+        | PromqlInfoEnrich { child, .. }
+        | PromqlSeriesSample { child, .. }
+        | Filter { child, .. }
+        | Project { child, .. }
+        | Aggregate { child, .. }
+        | Dedup { child, .. }
+        | Sort { child, .. }
+        | Limit { child, .. }
+        | PromqlSubquery { child, .. }
+        | TimeRange { child, .. }
+        | TimeShift { child, .. }
+        | SQLWindowFunc { child, .. } => visit(child, seen),
+        // `Concat`'s branches are stored by value (`Vec<QueryExpr>`, not
+        // `Rc<QueryExpr>` — see `rebuild_children`'s `intern_owned` use for
+        // this variant), so a branch has no `Rc` identity of its own to
+        // dedup on at this position; still recurse into each in case an
+        // `Rc`-shared descendant appears further down.
+        Concat { children } => children.iter().map(|c| count_unique(c, seen)).sum(),
+        Join { left, right, .. } | SetOp { left, right, .. } => {
+            visit(left, seen) + visit(right, seen)
+        }
+        BinaryOp { lhs, rhs, .. } => visit(lhs, seen) + visit(rhs, seen),
+        // Scalar variants (issue #205) — never descended into, matching
+        // `rebuild_children`'s own scope exactly (see its trailing match
+        // arm and this module's "Algorithm" section).
+        Column(_)
+        | Literal(_)
+        | Compare { .. }
+        | BoolAnd(_)
+        | BoolOr(_)
+        | Not(_)
+        | IsNull(_)
+        | IsNotNull(_)
+        | Cast { .. }
+        | InList { .. }
+        | FunctionCall { .. }
+        | Arithmetic { .. }
+        | Case { .. } => 0,
+    }
+}
+
 /// Recurse into `child`, then intern the result. `Rc::try_unwrap` recovers
 /// the owned node without cloning in the overwhelmingly common case — a
 /// tree freshly built by a front end / `resolve_root`, not yet shared by any
@@ -516,6 +620,67 @@ mod tests {
         );
     }
 
+    // ── dag_node_count ───────────────────────────────────────────────────
+
+    #[test]
+    fn dag_node_count_is_the_naive_count_when_nothing_is_shared() {
+        // scan() alone: 1 node.
+        assert_eq!(dag_node_count(&scan()), 1);
+        // quantile_agg's own child is a fresh, unshared scan(): 2 nodes.
+        assert_eq!(dag_node_count(&quantile_agg(vec![1], Some(2), 0.5)), 2);
+    }
+
+    #[test]
+    fn dag_node_count_deduplicates_an_internally_shared_subtree() {
+        // Same shape as `single_query_shares_its_own_repeated_subtree`: a
+        // BinaryOp whose two branches are the *same* Rc after
+        // `share_common_subtrees` (2 nodes: Scan + Aggregate) — the root
+        // itself makes 3 unique nodes total (BinaryOp, Aggregate, Scan),
+        // not 5 (which a tree-walk / naive serialization, counting the
+        // shared branch's 2 nodes twice, would report).
+        let agg = quantile_agg(vec![1], Some(2), 0.5);
+        let root = QueryExpr::BinaryOp {
+            op: BinaryOpKind::Compare(crate::pre_asap::expr_ir::CompareOpKind::Eq),
+            lhs: Rc::new(agg.clone()),
+            rhs: Rc::new(agg),
+            vector_match: None,
+        };
+        let shared = share_common_subtrees(vec![("q", root)]);
+        let [(_, root)] = shared.as_slice() else {
+            panic!("expected 1 root");
+        };
+        assert_eq!(
+            dag_node_count(root),
+            3,
+            "the shared branch's 2 nodes must be counted once, not once per \
+             occurrence — got {} for {root:?}",
+            dag_node_count(root)
+        );
+    }
+
+    #[test]
+    fn dag_node_count_deduplicates_across_two_workload_roots() {
+        // Two workload roots sharing one Aggregate after
+        // `share_common_subtrees` (the `duplicate_workload_queries_...`
+        // shape from `crates/integration-tests/tests/cse.rs`, built
+        // directly here): each root's own `dag_node_count` must report the
+        // shared subtree's real size once, not double-count anything —
+        // there's nothing *to* double-count from a single root's own count
+        // in this case (no root references the shared node twice), so this
+        // pins the simpler, more common case that a per-candidate cost
+        // proxy (`CseCandidate::subtree` in `asap-aware-mapping`) actually
+        // exercises: counting one occurrence's own reachable DAG size.
+        let a = quantile_agg(vec![1], Some(2), 0.5);
+        let b = quantile_agg(vec![1], Some(2), 0.5);
+        let shared = share_common_subtrees(vec![("a", a), ("b", b)]);
+        let [(_, ra), (_, rb)] = shared.as_slice() else {
+            panic!("expected 2 roots");
+        };
+        assert!(Rc::ptr_eq(ra, rb), "fixture sanity: the two roots merged");
+        assert_eq!(dag_node_count(ra), 2);
+        assert_eq!(dag_node_count(rb), 2);
+    }
+
     #[test]
     fn dedup_gates_sharing_the_same_as_aggregate() {
         // `Dedup { cols }` adds `cols` as a unique key — so two identical
 
@@ -80,9 +80,15 @@ independently (`RecomputeIndependently`) per that one decision.
 
 ## Defaults
 
-`cse_recompute_cost`'s default is a structural-size proxy (the length of the
-same canonical-JSON serialization `cse::structural_hash` already computes for
-the subtree — no second traversal). `cse_shared_maintenance_cost`'s default
+`cse_recompute_cost`'s default is a structural-size proxy: `cse::dag_node_count`,
+the number of *unique* nodes in the subtree's DAG (deduplicated by `Rc`
+pointer identity), not a raw serialization length. This distinction matters
+here specifically — a `CseCandidate`'s subtree is, by definition, something
+CSE already found sharing in, so it's generally a DAG, not a tree; a naive
+tree-shaped size measure (a full `serde_json` serialization, or a recursive
+walk with no identity tracking) would re-count any descendant the subtree
+already shares internally once per parent that reaches it, over-stating the
+real cost of holding or recomputing it once. `cse_shared_maintenance_cost`'s default
 is a small per-`SummaryFamilyType` weight table (exact accumulators cheapest,
 sketches/samples/wavelets/stat-models progressively more expensive to keep
 continuously updated) scaled to the same order of magnitude as typical