feat(pre-asap): structural CSE via bottom-up hash-consing (stages 1+2)

crates/asap-aware-mapping/src/bind.rs

@@ -91,6 +91,62 @@ pub fn implement_tree_with(
     logical(expr)
 }
 
+/// Bind a whole workload's worth of already-CSE'd roots
+/// ([`asap_types::pre_asap::cse::share_common_subtrees`]'s output), reusing
+/// one bound [`SummaryNode`] wherever two roots share the same `Rc` (issue
+/// #212, #222, #223 stage 2).
+///
+/// This is a real caller for `share_common_subtrees`, wired up deliberately:
+/// the pass's own landing plan calls out that its predecessor
+/// (`asap-plan::cse::dedupe_subtrees`) was deleted in #192 for being unwired
+/// dead code, and lands this memoization alongside it so that never becomes
+/// true again.
+///
+/// The memo key is `Rc::as_ptr` — pointer identity, not a second
+/// `PartialEq`/structural-equality pass. `share_common_subtrees` already made
+/// the (non-negotiable, `PartialEq`-checked) sharing decision; this only
+/// needs to recognize when it already bound the exact `Rc` a later root
+/// hands back, and is deliberately *not* a general "does an available
+/// `Implementation` satisfy this one" lookup — that subsumption question is
+/// `asap_aware_mapping::boundary::Matcher`'s documented, deliberately-unfilled
+/// job, not this one's.
+///
+/// Only whole-root sharing is memoized (matching two workload roots that are
+/// themselves the same `Rc<QueryExpr>` after CSE) — [`implement_tree`] is
+/// called at most once per distinct root pointer, but it still walks each
+/// such tree's own internal structure fresh; a subtree shared only *below*
+/// two different roots' top level does not additionally memoize inside that
+/// walk. Widening this to sub-root memoization is future work.
+pub fn implement_workload<Id>(
+    roots: Vec<(Id, Rc<QueryExpr>)>,
+) -> Vec<(Id, Result<Rc<SummaryNode>, ImplementError>)> {
+    implement_workload_with(roots, &DefaultCostModel)
+}
+
+/// Like [`implement_workload`], but ranks candidate summaries via
+/// `cost_model` instead of the built-in static preference order (see
+/// [`crate::cost_model`]).
+pub fn implement_workload_with<Id>(
+    roots: Vec<(Id, Rc<QueryExpr>)>,
+    cost_model: &dyn CostModel,
+) -> Vec<(Id, Result<Rc<SummaryNode>, ImplementError>)> {
+    let mut memo: std::collections::HashMap<*const QueryExpr, Rc<SummaryNode>> =
+        std::collections::HashMap::new();
+    roots
+        .into_iter()
+        .map(|(id, expr)| {
+            let ptr = Rc::as_ptr(&expr);
+            let result = match memo.get(&ptr) {
+                Some(cached) => Ok(Rc::clone(cached)),
+                None => implement_tree_with(&expr, cost_model).inspect(|node| {
+                    memo.insert(ptr, Rc::clone(node));
+                }),
+            };
+            (id, result)
+        })
+        .collect()
+}
+
 /// Translate an [`Implementation`] into the `(family, needs a
 /// SummaryEstimate readout)` pair [`bind_summary_agg`] needs, or `None` for
 /// `PassThrough` (the caller falls back to [`logical`]).

crates/asap-aware-mapping/src/lib.rs

@@ -36,9 +36,13 @@
 //!   `AggIntent → Implementation` (a summary family's own `(Kind, Params)`,
 //!   or an exact accumulator) sized to the `AccuracyTarget` (issue #98).
 //!   [`boundary::implementation_for`] is the per-node decision;
-//!   [`bind::implement_tree`] drives it over a whole tree — see the
-//!   terminology section below for why these two are named around
-//!   "implementation" rather than "bind".
+//!   [`bind::implement_tree`] drives it over a whole tree, and
+//!   [`bind::implement_workload`] drives it over a whole workload's roots —
+//!   memoized on `Rc` identity so two roots that
+//!   `asap_types::pre_asap::cse::share_common_subtrees` already collapsed
+//!   onto one shared subtree bind to one shared `SummaryNode` too (issue
+//!   #212, #222, #223) — see the terminology section below for why these are
+//!   named around "implementation" rather than "bind".
 //! - [`cost_model`] — the [`CostModel`](cost_model::CostModel) trait every
 //!   deployment's cost-based sketch selection plugs into (issues #6, #33).
 //!   `asap-plan` itself only ships [`DefaultCostModel`](cost_model::DefaultCostModel),
@@ -87,7 +91,10 @@ pub mod bind;
 pub mod boundary;
 pub mod cost_model;
 
-pub use bind::{implement_tree, implement_tree_with, ImplementError};
+pub use bind::{
+    implement_tree, implement_tree_with, implement_workload, implement_workload_with,
+    ImplementError,
+};
 pub use boundary::{
     implementation_for, implementation_for_with, summary_candidates, Implementation, Matcher,
 };

crates/integration-tests/tests/cse.rs

@@ -0,0 +1,134 @@
+//! End-to-end pre-ASAP CSE → implementation pin (issue #212, #222, #223).
+//!
+//! Drives the full staged pipeline this issue lands: two independently
+//! lowered `QueryExpr` trees → `share_common_subtrees` (stage 1,
+//! `asap-types::pre_asap::cse`) → `implement_workload` (stage 2,
+//! `asap-aware-mapping`) — and asserts the sharing that stage 1 decides
+//! survives into stage 2's bound output as one genuinely shared
+//! `Rc<SummaryNode>`, not just one shared `Rc<QueryExpr>`. This is the "real
+//! caller" the issue's landing plan requires before `share_common_subtrees`
+//! is allowed to exist at all (its predecessor,
+//! `asap-plan::cse::dedupe_subtrees`, was deleted in #192 for being unwired
+//! dead code).
+
+use std::rc::Rc;
+
+use asap_aware_mapping::implement_workload;
+use asap_frontend_promql::lower_promql;
+use asap_types::pre_asap::cse::share_common_subtrees;
+use asap_types::pre_asap::query_expr::QueryExpr;
+use asap_types::types::AccuracyTarget;
+
+/// Two workload entries that happen to submit the exact same query (a
+/// realistic case — two dashboards, or a query fired both standalone and as
+/// part of a larger batch) collapse onto one shared `Rc<QueryExpr>` after
+/// `share_common_subtrees`, and then onto one genuinely-shared, single build
+/// `Rc<SummaryNode>` after `implement_workload` — no second structural-
+/// equality pass at the post-ASAP layer needed for this kind of sharing.
+#[test]
+fn duplicate_workload_queries_share_one_bound_summary() {
+    // Grouped (`by (job)`), so the shared `Aggregate`'s output schema carries
+    // a provable unique key — the legality gate `share_common_subtrees`
+    // enforces (see `asap-types::pre_asap::cse`'s module doc) — and its
+    // `ExactAggregate(Sum)` realization is deterministic regardless of the
+    // accuracy target, so this pins the sharing mechanism itself rather than
+    // any one particular summary-family choice.
+    let query = "sum by (job) (http_requests_total)";
+    let a = lower_promql(query, AccuracyTarget::Exact).expect("query a failed to lower");
+    let b = lower_promql(query, AccuracyTarget::Exact).expect("query b failed to lower");
+
+    // Independently lowered: not yet sharing any `Rc`, even though they are
+    // structurally identical (`resolve_root` gives each call its own fresh
+    // tree).
+    assert_eq!(
+        a, b,
+        "fixture sanity: identical query text lowers identically"
+    );
+
+    let shared = share_common_subtrees(vec![("a", a), ("b", b)]);
+    let [(_, qa), (_, qb)] = shared.as_slice() else {
+        panic!("expected 2 roots");
+    };
+    assert!(
+        Rc::ptr_eq(qa, qb),
+        "share_common_subtrees must collapse the two identical roots onto one Rc<QueryExpr>"
+    );
+
+    let bound = implement_workload(shared);
+    let [(_, ra), (_, rb)] = bound.as_slice() else {
+        panic!("expected 2 bound results");
+    };
+    let ra = ra.as_ref().expect("query a failed to bind");
+    let rb = rb.as_ref().expect("query b failed to bind");
+    assert!(
+        Rc::ptr_eq(ra, rb),
+        "implement_workload must reuse one bound SummaryNode for the shared Rc<QueryExpr>, \
+         got two independently-built nodes: {ra:?} vs {rb:?}"
+    );
+}
+
+/// The negative control: two workload entries whose queries are NOT
+/// structurally identical must not be conflated by `implement_workload`'s
+/// `Rc::as_ptr` memoization — different `Rc<QueryExpr>` roots always bind
+/// independently.
+#[test]
+fn distinct_workload_queries_do_not_share_a_bound_summary() {
+    let a = lower_promql("sum by (job) (http_requests_total)", AccuracyTarget::Exact)
+        .expect("query a failed to lower");
+    let b = lower_promql("sum by (job) (http_response_total)", AccuracyTarget::Exact)
+        .expect("query b failed to lower");
+    assert_ne!(a, b, "fixture sanity: the two queries differ");
+
+    let shared = share_common_subtrees(vec![("a", a), ("b", b)]);
+    let [(_, qa), (_, qb)] = shared.as_slice() else {
+        panic!("expected 2 roots");
+    };
+    assert!(!Rc::ptr_eq(qa, qb));
+
+    let bound = implement_workload(shared);
+    let [(_, ra), (_, rb)] = bound.as_slice() else {
+        panic!("expected 2 bound results");
+    };
+    let ra = ra.as_ref().expect("query a failed to bind");
+    let rb = rb.as_ref().expect("query b failed to bind");
+    assert!(
+        !Rc::ptr_eq(ra, rb),
+        "distinct queries must bind independently"
+    );
+}
+
+/// Single-query CSE (a repeated sub-expression within one query) also
+/// survives through `implement_workload`: the two grouped-`Aggregate`
+/// branches of a `BinaryOp` collapse to one shared `Rc<QueryExpr>` in
+/// `share_common_subtrees`, and to one shared bound `Rc<SummaryNode>` here.
+#[test]
+fn single_query_repeated_subexpression_shares_one_bound_summary() {
+    let query = "sum by (job) (http_requests_total) / sum by (job) (http_requests_total)";
+    let expr = lower_promql(query, AccuracyTarget::Exact).expect("query failed to lower");
+
+    let shared = share_common_subtrees(vec![("q", expr)]);
+    let [(_, root)] = shared.as_slice() else {
+        panic!("expected 1 root");
+    };
+    let QueryExpr::BinaryOp { lhs, rhs, .. } = root.as_ref() else {
+        panic!("expected a BinaryOp root, got {root:?}");
+    };
+    assert!(
+        Rc::ptr_eq(lhs, rhs),
+        "the two identical sum-by-job branches must collapse onto one Rc<QueryExpr>"
+    );
+
+    // Bind both branches (as if they were two roots of a tiny sub-workload)
+    // through the same memoized `implement_workload`, and confirm they bind
+    // to the identical `Rc<SummaryNode>`.
+    let bound = implement_workload(vec![("lhs", Rc::clone(lhs)), ("rhs", Rc::clone(rhs))]);
+    let [(_, ra), (_, rb)] = bound.as_slice() else {
+        panic!("expected 2 bound results");
+    };
+    let ra = ra.as_ref().expect("lhs failed to bind");
+    let rb = rb.as_ref().expect("rhs failed to bind");
+    assert!(
+        Rc::ptr_eq(ra, rb),
+        "the shared branch must bind to one shared SummaryNode"
+    );
+}