From b78ade9ae574363ed87e9787602f6f971031e202 Mon Sep 17 00:00:00 2001
From: zzylol <50204836+zzylol@users.noreply.github.com>
Date: Thu, 22 Oct 2026 16:13:00 +0000
Subject: [PATCH] Joint planning search: interchangeable queries take one
 choice; components split by the nodes queries can share

---
 .../tests/joint_search_equivalence.rs         | 267 ++++++++++-
 crates/plan-selection/src/lib.rs              |  17 +-
 crates/plan-selection/src/pass/components.rs  | 327 +++++++++++++-
 .../src/pass/interchangeable.rs               | 415 ++++++++++++++++++
 crates/plan-selection/src/pass/mod.rs         |   1 +
 .../src/pass/query_candidates.rs              |   4 +-
 .../src/pass/stage1_to_stage3.rs              |  30 +-
 ...andom-mixed-seed-862-long-workload-01.json |  50 +--
 .../random-mixed-seed-862-workload-01.json    |  50 +--
 .../random-mixed-seed-862-x5-workload-01.json |  50 +--
 10 files changed, 1072 insertions(+), 139 deletions(-)
 create mode 100644 crates/plan-selection/src/pass/interchangeable.rs

diff --git a/crates/integration-tests/tests/joint_search_equivalence.rs b/crates/integration-tests/tests/joint_search_equivalence.rs
index af632784f..74be1785a 100644
--- a/crates/integration-tests/tests/joint_search_equivalence.rs
+++ b/crates/integration-tests/tests/joint_search_equivalence.rs
@@ -616,7 +616,7 @@ fn random_workloads_over_disjoint_sources_combine_their_components() {
         let entries: Vec<_> = workload.query_workload.entries().collect();
         let data = workload.data_workload.clone().unwrap();
         let models = PlanningModels::builtin().with_plan_selection(&settings[0]);
-        let run = assert_equivalent(&format!("seed {seed}"), &roots, &entries, &data, models);
+        let run = assert_same_cost(&format!("seed {seed}"), &roots, &entries, &data, models);
         if let Ok(selection) = &run.stage3_plan_selection {
             if !selection.search.components.is_empty() {
                 combined += 1;
@@ -635,7 +635,8 @@ fn random_workloads_over_disjoint_sources_combine_their_components() {
 }
 
 /// Queries reading disjoint sources are separate components, and queries
-/// linked through a common source are one, in workload order.
+/// linked through a summary build they can share are one, in workload
+/// order.
 #[test]
 fn components_follow_shared_sources() {
     let mut workload = random_workload_over(1, &["m", "n"]);
@@ -653,14 +654,15 @@ fn components_follow_shared_sources() {
         entry("sum by (job) (sum_over_time(m[5m]))"),
         entry("sum by (job) (rate(c_total[5m]))"),
         entry("max by (job) (max_over_time(n[5m]))"),
-        entry("sum(sum_over_time(m[1m])) / sum(sum_over_time(n[1m]))"),
+        entry("sum by (job) (sum_over_time(m[5m])) / max by (job) (max_over_time(n[5m]))"),
     ]);
     let roots = lower_promql_roots(&workload).unwrap();
     let entries: Vec<_> = workload.query_workload.entries().collect();
     let data = workload.data_workload.clone().unwrap();
     let run = assert_equivalent("linked", &roots, &entries, &data, PlanningModels::builtin());
     let search = &run.stage3_plan_selection.as_ref().unwrap().search;
-    // q4 reads m and n, linking q1 and q3; q2 alone reads c_total.
+    // q4 can share q1's build over m and q3's over n, linking them; q2
+    // alone reads c_total.
     let components: Vec<_> = search
         .components
         .iter()
@@ -898,3 +900,260 @@ fn a_shared_summary_is_combined_with_a_rollup() {
         cost(&selected.id)
     );
 }
+
+/// `roots` planned with the exhaustive search and with the default search
+/// under `models`: both select a plan of the same price (within Stage 3's
+/// tie tolerance), or both find none. Rules that keep one of several plans
+/// of equal price (interchangeable queries) may select another plan than
+/// the exhaustive search's first. Returns the default search's run.
+fn assert_same_cost(
+    label: &str,
+    roots: &[QueryRoot],
+    entries: &[QueryWorkloadEntry],
+    data: &DataWorkload,
+    models: PlanningModels<'_>,
+) -> Stage1ToStage3Run<usize> {
+    let indexed = || roots.iter().cloned().enumerate().collect::<Vec<_>>();
+    let exhaustive = run_stage1_to_stage3_with(
+        indexed(),
+        entries,
+        data,
+        models,
+        Stage1ToStage3Search::exhaustive(),
+    )
+    .unwrap_or_else(|e| panic!("{label}: {e}"));
+    let pruned = run_stage1_to_stage3(indexed(), entries, data, models)
+        .unwrap_or_else(|e| panic!("{label}: {e}"));
+    match (
+        &exhaustive.stage3_plan_selection,
+        &pruned.stage3_plan_selection,
+    ) {
+        (Ok(all), Ok(searched)) => {
+            let price =
+                |s: &asap_plan_selection::Selection| s.costs[&s.selected].cost_per_second.unwrap();
+            let (all, searched) = (price(all), price(searched));
+            assert!(
+                (all - searched).abs() <= 1e-9 * all.abs(),
+                "{label}: exhaustive {all}, pruned {searched}"
+            );
+        }
+        (Err(SelectionError::NoValidCandidate(_)), Err(SelectionError::NoValidCandidate(_))) => {}
+        (all, searched) => panic!("{label}: exhaustive {all:?}, pruned {searched:?}"),
+    }
+    pruned
+}
+
+/// A workload of repeating `queries` (each with its interval in ms) over
+/// the gauge `m`, every query with accuracy target `accuracy`.
+fn workload_of(queries: &[(&str, u32)], accuracy: AccuracyTarget) -> PlanningWorkload {
+    let mut workload = random_workload(1);
+    workload.query_workload.repeating_queries = Some(
+        queries
+            .iter()
+            .map(|&(query, interval_ms)| RepeatingEntry {
+                query: Query(query.into()),
+                demand: RepeatedDemand::FixedInterval(RepetitionInterval(interval_ms)),
+                requirements: QueryRequirements {
+                    accuracy: accuracy.clone(),
+                    max_latency_ms: None,
+                },
+                predictability: Predictability::Predictable { known_at: None },
+                range_evaluation: None,
+            })
+            .collect(),
+    );
+    workload
+}
+
+/// Queries over one gauge that differ at most in a quantile rank, with
+/// equal workload entries, are one class of interchangeable queries: the
+/// default search gives them one shared choice and still selects a plan of
+/// the exhaustive search's price, with raw data kept and without, beside a
+/// query of another shape. A query of another interval at one member's
+/// rank, or a ratio of two members' ranks, can share one member's exact
+/// quantile and not the other's: then there is no class, and the price is
+/// still the exhaustive search's.
+#[test]
+fn interchangeable_queries_select_the_exhaustive_price() {
+    let q = |phi: &str| format!("quantile_over_time({phi}, m[5m])");
+    let (q5, q9, q99) = (q("0.5"), q("0.9"), q("0.99"));
+    let avg = |phi: &str| format!("avg(quantile_over_time({phi}, m[5m]))");
+    let (avg5, avg9) = (avg("0.5"), avg("0.9"));
+    let ratio = format!("{q9} / {q5}");
+    let sum = "sum_over_time(m[5m])";
+    // (label, queries with their intervals, the largest class's size)
+    type Case<'a> = (&'a str, Vec<(&'a str, u32)>, usize);
+    let cases: Vec<Case> = vec![
+        (
+            "three ranks",
+            vec![(&q5, 60_000), (&q9, 60_000), (&q99, 60_000)],
+            3,
+        ),
+        // The 0.9 duplicates can share their exact quantile, which the 0.5
+        // query cannot join: the duplicates are a class of their own.
+        (
+            "two ranks and a duplicate",
+            vec![(&q5, 60_000), (&q9, 60_000), (&q9, 60_000)],
+            2,
+        ),
+        (
+            "averages of two ranks and a duplicate",
+            vec![(&avg9, 60_000), (&avg9, 60_000), (&avg5, 60_000)],
+            2,
+        ),
+        // The 300 s query can share the 0.9 member's exact quantile, not the
+        // 0.5 member's, so the two are no longer alike.
+        (
+            "two intervals",
+            vec![(&q5, 60_000), (&q9, 60_000), (&q9, 300_000)],
+            0,
+        ),
+        // So can the ratio, with each member's exact quantile.
+        (
+            "a ratio of ranks",
+            vec![(&q5, 60_000), (&q9, 60_000), (&ratio, 60_000)],
+            0,
+        ),
+        (
+            "identical sums",
+            vec![(sum, 60_000), (sum, 60_000), (&q5, 60_000)],
+            2,
+        ),
+    ];
+    let deployments = [
+        asap_logical_asap_optimization::DeploymentCapabilities::default(),
+        no_raw_data(),
+    ];
+    for (label, queries, class_size) in cases {
+        for accuracy in [
+            AccuracyTarget::max_error(0.05),
+            AccuracyTarget::max_error(0.05).accuracy_metric(AccuracyMetric::Rank),
+        ] {
+            let workload = workload_of(&queries, accuracy);
+            let roots = lower_promql_roots(&workload).unwrap();
+            let entries: Vec<_> = workload.query_workload.entries().collect();
+            let data = workload.data_workload.clone().unwrap();
+            for deployment in &deployments {
+                let models = PlanningModels::builtin().with_deployment_capabilities(deployment);
+                let run = assert_same_cost(label, &roots, &entries, &data, models);
+                if let Ok(selection) = &run.stage3_plan_selection {
+                    let largest = selection
+                        .search
+                        .interchangeable_queries
+                        .iter()
+                        .map(Vec::len)
+                        .max()
+                        .unwrap_or(0);
+                    assert_eq!(largest, class_size, "{label}: class size");
+                }
+            }
+        }
+    }
+}
+
+/// Two identical SQL queries and a third over another window, over one
+/// table: the identical two are one class of interchangeable queries, and
+/// the default search selects a plan of the exhaustive search's price.
+#[test]
+fn identical_sql_queries_are_one_class() {
+    let (mut workload, catalog) = example2();
+    let entries = workload.query_workload.repeating_queries.as_mut().unwrap();
+    let first = entries[0].clone();
+    let distinct = |minutes: u32| {
+        RepeatingEntry {
+        query: Query(format!(
+            "SELECT COUNT(DISTINCT src_ip) FROM flows WHERE ts >= now() - INTERVAL '{minutes} minute'"
+        )),
+        ..first.clone()
+    }
+    };
+    *entries = vec![distinct(1), distinct(5), distinct(1)];
+    let roots = lower_sql_roots(&workload, &catalog).unwrap();
+    let entries: Vec<_> = workload.query_workload.entries().collect();
+    let data = workload.data_workload.clone().unwrap();
+    let run = assert_same_cost(
+        "identical SQL queries",
+        &roots,
+        &entries,
+        &data,
+        PlanningModels::builtin(),
+    );
+    let search = &run.stage3_plan_selection.as_ref().unwrap().search;
+    assert_eq!(search.interchangeable_queries, vec![vec![0, 2]]);
+}
+
+/// Seeded random workloads over one gauge whose queries repeat each other
+/// or differ in a quantile rank, at one of two intervals, with raw data
+/// kept or not: the default search selects a plan of the exhaustive
+/// search's price.
+#[test]
+fn random_workloads_with_interchangeable_queries_select_the_exhaustive_price() {
+    let pool = [
+        "quantile_over_time(0.5, m[5m])",
+        "quantile_over_time(0.9, m[5m])",
+        "quantile_over_time(0.99, m[5m])",
+        "sum_over_time(m[5m])",
+        "sum by (job) (sum_over_time(m[5m]))",
+        "max_over_time(m[5m])",
+    ];
+    let mut searched = 0;
+    for seed in 1..=30u64 {
+        let mut rng = Rng(seed.wrapping_mul(0x9E37_79B9_7F4A_7C15) | 1);
+        let count = 2 + rng.below(2);
+        let queries: Vec<(&str, u32)> = (0..count)
+            .map(|_| (*rng.pick(&pool), *rng.pick(&[60_000u32, 300_000])))
+            .collect();
+        let workload = workload_of(&queries, AccuracyTarget::max_error(0.05));
+        let roots = lower_promql_roots(&workload).unwrap();
+        let inventory =
+            asap_logical_asap_optimization::per_computation_candidates::generate_per_computation_candidates(
+                roots.iter().cloned().enumerate().collect(),
+            )
+            .unwrap();
+        if inventory.combinations() > MAX_EXHAUSTIVE_COMBINATIONS {
+            continue;
+        }
+        searched += 1;
+        let entries: Vec<_> = workload.query_workload.entries().collect();
+        let data = workload.data_workload.clone().unwrap();
+        let deployment = match seed % 2 {
+            0 => no_raw_data(),
+            _ => asap_logical_asap_optimization::DeploymentCapabilities::default(),
+        };
+        let models = PlanningModels::builtin().with_deployment_capabilities(&deployment);
+        assert_same_cost(&format!("seed {seed}"), &roots, &entries, &data, models);
+    }
+    assert!(
+        searched >= 15,
+        "only {searched} workloads were small enough"
+    );
+}
+
+/// Queries over different windows, or with a latency bound (a check, not a
+/// charge, so the class argument does not cover it), are in no class.
+#[test]
+fn queries_over_other_windows_or_with_a_latency_bound_are_in_no_class() {
+    let mut workload = workload_of(
+        &[
+            ("quantile_over_time(0.5, m[5m])", 60_000),
+            ("quantile_over_time(0.9, m[10m])", 60_000),
+            ("quantile_over_time(0.99, m[5m])", 60_000),
+        ],
+        AccuracyTarget::max_error(0.05),
+    );
+    workload.query_workload.repeating_queries.as_mut().unwrap()[2]
+        .requirements
+        .max_latency_ms = Some(60_000.0);
+    let roots = lower_promql_roots(&workload).unwrap();
+    let entries: Vec<_> = workload.query_workload.entries().collect();
+    let data = workload.data_workload.clone().unwrap();
+    let run = assert_same_cost(
+        "no class",
+        &roots,
+        &entries,
+        &data,
+        PlanningModels::builtin(),
+    );
+    let search = &run.stage3_plan_selection.as_ref().unwrap().search;
+    assert!(search.interchangeable_queries.is_empty());
+}
diff --git a/crates/plan-selection/src/lib.rs b/crates/plan-selection/src/lib.rs
index 11569df70..2b591ce70 100644
--- a/crates/plan-selection/src/lib.rs
+++ b/crates/plan-selection/src/lib.rs
@@ -1015,6 +1015,12 @@ pub struct SearchReport {
     /// deployment that keeps none (nor their identical-expression Pass 2
     /// candidates, which inherit that).
     pub raw_scan_candidates_skipped: u64,
+    /// The classes of interchangeable queries, by query index: queries
+    /// that differ at most in a quantile rank, which the search gave one
+    /// shared choice and the same summary builds (design method 4). Empty
+    /// for the exhaustive search and outside cost model 1.
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub interchangeable_queries: Vec<Vec<usize>>,
     /// Under cost models 1 and 2, the cost per second of the cheapest
     /// valid candidate priced more than rounding (1e-12 of it) above the
     /// cheapest: how close the selection came to a tie. Candidates within
@@ -1024,8 +1030,9 @@ pub struct SearchReport {
     pub runner_up_cost_per_second: Option<f64>,
 }
 
-/// One independent component of the workload: queries that read no source
-/// another component reads, so no Pass 2 rule, raw-data retention or
+/// One independent component of the workload: queries that can read no
+/// node another component's queries read, nor any source when raw data on
+/// disk is priced, so no Pass 2 rule, raw-data retention or
 /// node charge links them to another component.
 #[derive(Debug, Clone, Default, PartialEq, serde::Serialize, serde::Deserialize)]
 pub struct ComponentReport {
@@ -1041,7 +1048,8 @@ pub struct ComponentReport {
 #[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
 #[serde(rename_all = "snake_case")]
 pub enum SearchedAsOne {
-    /// Its queries all read sources that link them.
+    /// Its queries are all linked: through common sources, and over one
+    /// source through a node two of them can read.
     OneComponent,
     /// The cost model's objective or limits are not a sum over components
     /// (cost models 2 and 3: peak cores and cluster limits).
@@ -1061,8 +1069,7 @@ pub enum SearchedAsOne {
     NotAdditive,
     /// The combination of the components' selections is not a candidate
     /// the joint search enumerates: the Pass 2 rules combine only some
-    /// rewrites (top-k reuse is not combined with panes), and cap their
-    /// candidates over the whole workload.
+    /// rewrites (exact top-k reuse is not combined with panes).
     NotInJointSpace,
 }
 
diff --git a/crates/plan-selection/src/pass/components.rs b/crates/plan-selection/src/pass/components.rs
index 0785d108e..bb9dfcfc2 100644
--- a/crates/plan-selection/src/pass/components.rs
+++ b/crates/plan-selection/src/pass/components.rs
@@ -1,16 +1,17 @@
 //! Independent components (joint-planning-search design, method 5): the
-//! workload's queries split into groups that read disjoint sources, each
+//! workload's queries split into groups that no candidate can link, each
 //! searched on its own, and their selections combined.
 //!
 //! Two queries that read no common source cannot interact under any
 //! alternative: every Pass 2 rule (identical expressions, summary
 //! capability, rollup, top-k reuse) shares only work over the same input
 //! rows, raw-data retention is kept per source, and every node's charge
-//! reads only the queries and sources above and below it. Under cost model
-//! 1 the workload's price is the sum of its nodes', so the cheapest plan is
-//! the union of each component's cheapest. Cost models 2 and 3 price or
-//! limit peak cores, a ceiling over the whole workload's work, so their
-//! workloads are searched as one.
+//! reads only the queries and sources above and below it. Two queries over
+//! one source interact only through a node both can read
+//! ([`sharing_components`]). Under cost model 1 the workload's price is the sum
+//! of its nodes', so the cheapest plan is the union of each component's
+//! cheapest. Cost models 2 and 3 price or limit peak cores, a ceiling over
+//! the whole workload's work, so their workloads are searched as one.
 //!
 //! The combined plan is rebuilt as one whole-workload candidate and checked
 //! and priced by Stage 3 again. It is the exhaustive search's selection
@@ -18,22 +19,31 @@
 //! (where the joint enumeration order would decide): each component's
 //! runner-up must cost more than its selection by a margin above that
 //! tolerance. Otherwise, or when the combined price is not the sum, the
-//! workload is searched as one.
+//! workload is searched as one. The combination is not looked up among the
+//! joint search's candidates: each component's Pass 2 rewrites touch only
+//! its own nodes, so together they are a joint candidate whenever the joint
+//! search combines those rules ([`combined_pass2`]), whatever the rules'
+//! enumeration caps over the whole workload.
 
 use std::collections::HashMap;
 use std::rc::Rc;
 
+use asap_logical_asap_optimization::asap_aware_cse::{
+    capability_sharing_candidates, pane_candidates, rollup_candidates, sharing_candidates,
+    topk_reuse_candidates, window_composition_pass2_candidates, with_union_scan_labels,
+    SharedPaneChain,
+};
 use asap_logical_asap_optimization::per_computation_candidates::{
-    generate_per_computation_candidates, PerComputationCandidates,
+    compose_logical_candidate, generate_per_computation_candidates, PerComputationCandidates,
 };
-use asap_logical_asap_optimization::workload_candidates::stage1_candidates_of_choice;
+use asap_logical_asap_optimization::workload_candidates::Pass2Candidate;
 use asap_physical_asap_optimization::optimize_physical_asap_assigned;
-use asap_types::ir::flat::flatten;
 use asap_types::ir::operator::Source;
 use asap_types::ir::{NonASAPOp, Operator, OperatorNode, QueryRoot};
 use asap_types::physical::Materialization;
 use asap_types::workload::{DataWorkload, QueryWorkloadEntry};
 
+use super::query_candidates::{next_digits, query_digits};
 use super::stage1_to_stage3::{
     run_stage1_to_stage3_from_pass1, selected_logical, LogicalCandidate, Stage1ToStage3Run,
     Stage1ToStage3Search,
@@ -101,6 +111,196 @@ pub(crate) fn source_components(roots: &[QueryRoot]) -> Vec<Vec<usize>> {
     components
 }
 
+/// The workload's queries grouped into components that no candidate can
+/// link: [`source_components`], each split further into the queries that
+/// can read a common node, in workload order, ordered by their first query.
+/// `per_computation` is the Pass 1 inventory of the workload's roots;
+/// `entries[q]` is query `q`'s workload entry.
+///
+/// Two queries over one source are linked when they read a common Pass 1
+/// target, or when, for some choice of their own candidates, a Pass 2 rule
+/// can make them read one node: the identical-expression rule (any
+/// identical node, whole or in panes, with each metric's scans widened to
+/// one label set), the summary-capability rule, the window-composition
+/// rule's shared chains, or a reuse rule (rollup or exact top-k reuse).
+/// Stage 2 can build any node two queries read once per evaluation, so
+/// every such node is a link. Queries whose candidates have no node in
+/// common under any rule share no charge: e.g. SQL queries over one table,
+/// whose table scans have no unique key and are never made one node, with
+/// different groupings.
+///
+/// Raw data kept for query-time scans is charged per source, for the
+/// longest window any query scans: with a disk price, that charge links
+/// every query over one source, and the components stay
+/// [`source_components`].
+pub(crate) fn sharing_components<Id: Clone>(
+    per_computation: &PerComputationCandidates<Id>,
+    entries: &[QueryWorkloadEntry],
+    models: &PlanningModels<'_>,
+) -> Vec<Vec<usize>> {
+    let roots: Vec<QueryRoot> = per_computation
+        .roots
+        .iter()
+        .map(|(_, root)| root.clone())
+        .collect();
+    let by_source = source_components(&roots);
+    let disk_priced = match &models.plan_selection.cost_model {
+        CostModel::UsageOverTime(prices) => prices.cost_per_disk_byte_second != 0.0,
+        _ => true,
+    };
+    if disk_priced {
+        return by_source;
+    }
+    let grids: Vec<Option<u64>> = entries
+        .iter()
+        .map(QueryWorkloadEntry::evaluation_grid_ms)
+        .collect();
+    let digits: Vec<Vec<usize>> = roots
+        .iter()
+        .map(|root| query_digits(per_computation, root))
+        .collect();
+    let mut components = Vec::new();
+    for queries in by_source {
+        // Union-find over the component's positions.
+        let mut parent: Vec<usize> = (0..queries.len()).collect();
+        fn find(parent: &mut [usize], i: usize) -> usize {
+            let mut root = i;
+            while parent[root] != root {
+                root = parent[root];
+            }
+            parent[i] = root;
+            root
+        }
+        for i in 0..queries.len() {
+            for j in i + 1..queries.len() {
+                if find(&mut parent, i) == find(&mut parent, j) {
+                    continue;
+                }
+                let (a, b) = (queries[i], queries[j]);
+                if can_share_a_node(per_computation, &digits, &grids, a, b) {
+                    let (ri, rj) = (find(&mut parent, i), find(&mut parent, j));
+                    parent[ri.max(rj)] = ri.min(rj);
+                }
+            }
+        }
+        let mut split: Vec<(usize, Vec<usize>)> = Vec::new();
+        for (i, &query) in queries.iter().enumerate() {
+            let root = find(&mut parent, i);
+            match split.iter_mut().find(|(r, _)| *r == root) {
+                Some((_, members)) => members.push(query),
+                None => split.push((root, vec![query])),
+            }
+        }
+        components.extend(split.into_iter().map(|(_, members)| members));
+    }
+    components.sort_by_key(|members| members[0]);
+    components
+}
+
+/// Whether queries `a` and `b` of `per_computation` can read one node
+/// ([`sharing_components`]): they read a common Pass 1 target, or for some
+/// choice of their own candidates, composed together, a Pass 2 rule links
+/// a node of each. `digits[q]` are query `q`'s digits of a whole-workload choice
+/// ([`query_digits`]); `grids[q]` its evaluation grid.
+fn can_share_a_node<Id: Clone>(
+    per_computation: &PerComputationCandidates<Id>,
+    digits: &[Vec<usize>],
+    grids: &[Option<u64>],
+    a: usize,
+    b: usize,
+) -> bool {
+    if digits[a].iter().any(|d| digits[b].contains(d)) {
+        return true;
+    }
+    let pair = PerComputationCandidates {
+        roots: vec![
+            per_computation.roots[a].clone(),
+            per_computation.roots[b].clone(),
+        ],
+        targets: per_computation.targets.clone(),
+    };
+    let pair_digits: Vec<usize> = digits[a].iter().chain(&digits[b]).copied().collect();
+    let pair_grids = [grids[a], grids[b]];
+    let mut choice = vec![0; per_computation.targets.len()];
+    loop {
+        if !per_computation.absorbed_choice_varies(&choice) {
+            if let Ok(composed) = compose_logical_candidate(&pair, &choice) {
+                let roots: Vec<QueryRoot> = composed.into_iter().map(|(_, root)| root).collect();
+                if rules_link_nodes(&roots, &pair_grids) {
+                    return true;
+                }
+            }
+        }
+        if !next_digits(per_computation, &pair_digits, &mut choice) {
+            return false;
+        }
+    }
+}
+
+/// Whether a Pass 2 rule, on the two-query candidate `roots`, makes both
+/// queries read one node: whole, with the metric's scans widened to one
+/// label set, or in any window-composition candidate (identical pane
+/// chains, or one chain serving nested windows or several intervals).
+fn rules_link_nodes(roots: &[QueryRoot], grids: &[Option<u64>]) -> bool {
+    let reach = |root: &QueryRoot| -> std::collections::HashSet<*const OperatorNode> {
+        root_nodes(root).iter().map(Rc::as_ptr).collect()
+    };
+    let links = |roots: &[QueryRoot]| {
+        let (first, second) = (reach(&roots[0]), reach(&roots[1]));
+        let spans = |nodes: &mut dyn Iterator<Item = &Rc<OperatorNode>>| {
+            let (mut in_first, mut in_second) = (false, false);
+            for node in nodes {
+                in_first |= first.contains(&Rc::as_ptr(node));
+                in_second |= second.contains(&Rc::as_ptr(node));
+            }
+            in_first && in_second
+        };
+        // A rule already made both read one node (a shared pane chain).
+        first.iter().any(|node| second.contains(node))
+            || sharing_candidates(roots)
+                .iter()
+                .any(|c| spans(&mut c.nodes.iter()))
+            || capability_sharing_candidates(roots)
+                .iter()
+                .any(|c| spans(&mut c.members.iter()))
+            || rollup_candidates(roots)
+                .iter()
+                .any(|c| spans(&mut [&c.coarse, &c.finer].into_iter()))
+            || topk_reuse_candidates(roots)
+                .iter()
+                .any(|c| spans(&mut c.members.iter()))
+    };
+    if links(roots) {
+        return true;
+    }
+    // Scans of one metric that project different labels, widened to one.
+    if let Ok(Some(widened)) = with_union_scan_labels(roots) {
+        if links(&widened) {
+            return true;
+        }
+    }
+    let panes = pane_candidates(roots, grids);
+    if panes.is_empty() {
+        return false;
+    }
+    window_composition_pass2_candidates(roots, &panes)
+        .unwrap_or_default()
+        .iter()
+        .any(|in_panes| matches!(&in_panes.roots, Ok(in_panes) if links(in_panes)))
+}
+
+/// Every node `root` reaches.
+fn root_nodes(root: &QueryRoot) -> Vec<Rc<OperatorNode>> {
+    let starts: Vec<&Rc<OperatorNode>> = match root {
+        QueryRoot::Operator(node) => vec![node],
+        QueryRoot::Scalar(expr) => expr.operator_refs(),
+    };
+    starts
+        .into_iter()
+        .flat_map(OperatorNode::reachable)
+        .collect()
+}
+
 /// Why `per_computation`'s workload under `models` and `search` is
 /// searched as one, or `None` when its `components` may be searched
 /// separately.
@@ -256,17 +456,20 @@ pub(crate) fn search_components<Id: Clone + std::fmt::Debug>(
         })
         .collect();
     let joint_choice = joint_choice.ok_or(SearchedAsOne::NotInJointSpace)?;
-    let flat = flatten(&combined).0;
-    let skip_raw = !models.deployment_capabilities.raw_data_retained;
-    let viable = |roots: &[QueryRoot]| {
-        super::stage1_to_stage3::viable(roots, entries, data, models, skip_raw)
-    };
-    let (joint, pass2) =
-        stage1_candidates_of_choice(&joint_inventory, entries, &joint_choice, &viable)
+    let pass2 = combined_pass2(&runs).ok_or(SearchedAsOne::NotInJointSpace)?;
+    // Stage 1 widens every metric's scans to one label set at once, so a
+    // selection with widened scans is a joint candidate only alone.
+    if components.len() > 1 {
+        let composed: Vec<QueryRoot> = compose_logical_candidate(&joint_inventory, &joint_choice)
+            .map_err(|_| SearchedAsOne::NotInJointSpace)?
             .into_iter()
-            .filter_map(|candidate| Some((candidate.roots.ok()?, candidate.pass2)))
-            .find(|(roots, _)| flatten(roots).0 == flat)
-            .ok_or(SearchedAsOne::NotInJointSpace)?;
+            .map(|(_, root)| root)
+            .collect();
+        if widens_a_scan(&combined, &composed) {
+            return Err(SearchedAsOne::NotInJointSpace);
+        }
+    }
+    let joint = combined;
     let mut assignments: Vec<std::vec::IntoIter<Materialization>> = runs
         .iter()
         .map(|run| {
@@ -340,6 +543,17 @@ pub(crate) fn search_components<Id: Clone + std::fmt::Debug>(
         })
         .collect();
     selection.search.runner_up_cost_per_second = None;
+    selection.search.interchangeable_queries = components
+        .iter()
+        .zip(&selections)
+        .flat_map(|(component, s)| {
+            s.search
+                .interchangeable_queries
+                .iter()
+                .map(|class| class.iter().map(|&q| component[q]).collect::<Vec<_>>())
+                .collect::<Vec<_>>()
+        })
+        .collect();
     selection.search.raw_scan_assignments_skipped = selections
         .iter()
         .map(|s| s.search.raw_scan_assignments_skipped)
@@ -368,6 +582,79 @@ pub(crate) fn search_components<Id: Clone + std::fmt::Debug>(
     })
 }
 
+/// Whether some query of `selected` reads a scan its root in `composed`
+/// (its Pass 1 candidate's roots) does not read: one Stage 1 widened to
+/// more labels ([`with_union_scan_labels`]).
+fn widens_a_scan(selected: &[QueryRoot], composed: &[QueryRoot]) -> bool {
+    let scans = |root: &QueryRoot| -> Vec<OperatorNode> {
+        root_nodes(root)
+            .into_iter()
+            .filter(|node| matches!(node.non_asap(), Some(NonASAPOp::Scan { .. })))
+            .map(|node| (*node).clone())
+            .collect()
+    };
+    selected.iter().zip(composed).any(|(selected, composed)| {
+        let original = scans(composed);
+        scans(selected).iter().any(|scan| !original.contains(scan))
+    })
+}
+
+/// What the Pass 2 rules changed in the components' selections, together:
+/// `Some(None)` when no rule changed anything. `None` when the joint
+/// search never combines those rewrites: it builds exact top-k reuse and
+/// tumbling panes into no common candidate (Stage 1's reuse rules follow
+/// only a rollup with panes). Each component's rewrites touch only its own
+/// nodes, so the rest is a candidate of the joint search, up to its
+/// enumeration caps, which bound each rule's list for one workload and are
+/// not a property of the plan.
+fn combined_pass2<Id>(runs: &[Stage1ToStage3Run<Id>]) -> Option<Option<Pass2Candidate>> {
+    let mut combined: Option<Pass2Candidate> = None;
+    for pass2 in runs
+        .iter()
+        .filter_map(|run| run.stage1_selected.as_ref()?.pass2.as_ref())
+    {
+        let into = combined.get_or_insert_with(|| Pass2Candidate {
+            from_pass1_position: 0,
+            shared_builds: Vec::new(),
+            pane_candidates: Vec::new(),
+            shared_pane_chains: Vec::new(),
+            rollup_candidates: Vec::new(),
+            topk_reuse_candidates: Vec::new(),
+            sharing_candidates: Vec::new(),
+        });
+        into.shared_builds
+            .extend(pass2.shared_builds.iter().cloned());
+        // Each component's chain groups are positions in its own panes.
+        let offset = into.pane_candidates.len();
+        into.shared_pane_chains
+            .extend(
+                pass2
+                    .shared_pane_chains
+                    .iter()
+                    .map(|chain| SharedPaneChain {
+                        members: chain.members.iter().map(|&p| p + offset).collect(),
+                        ..chain.clone()
+                    }),
+            );
+        into.pane_candidates
+            .extend(pass2.pane_candidates.iter().cloned());
+        into.rollup_candidates
+            .extend(pass2.rollup_candidates.iter().cloned());
+        into.topk_reuse_candidates
+            .extend(pass2.topk_reuse_candidates.iter().cloned());
+        into.sharing_candidates
+            .extend(pass2.sharing_candidates.iter().cloned());
+    }
+    match &combined {
+        Some(pass2)
+            if !pass2.topk_reuse_candidates.is_empty() && !pass2.pane_candidates.is_empty() =>
+        {
+            None
+        }
+        _ => Some(combined),
+    }
+}
+
 /// `rejection` of a component's search with its query numbers (`q<n>` at
 /// the start of its id or reason, counted within the component) renamed to
 /// the workload's.
diff --git a/crates/plan-selection/src/pass/interchangeable.rs b/crates/plan-selection/src/pass/interchangeable.rs
new file mode 100644
index 000000000..930be4ece
--- /dev/null
+++ b/crates/plan-selection/src/pass/interchangeable.rs
@@ -0,0 +1,415 @@
+//! Interchangeable queries (joint-planning-search design, method 4 for
+//! one class of alternatives): queries that differ at most in a quantile
+//! rank take one shared choice.
+//!
+//! Two queries are interchangeable when their workload entries are equal
+//! but for the query text, neither has a latency bound, each reads only
+//! its own Pass 1 targets, and for every choice of their own candidates
+//! the two queries, each planned alone, compose to the same plan up to the
+//! quantile rank `q` of their quantile computations and estimates; and, when
+//! their ranks differ, no two members have the same rank and no other query
+//! can share a node that carries one of their ranks (a ratio `q0.9 / q0.5`
+//! shares the 0.5 query's exact quantile, not the 0.9 query's, so it tells
+//! them apart; two 0.9 duplicates share theirs, which a 0.5 member cannot
+//! join). Duplicates are a class of their own. Neither
+//! a cost, a summary's size nor a check reads `q`: a quantile estimate is
+//! priced per group, and its sketch is sized for the rank or value error,
+//! the same at every rank.
+//!
+//! Swapping two members then maps every plan to one of the same price, and
+//! every member contributes the same charges to whatever summary build it
+//! reads. Under cost model 1 a
+//! build's price is the cheapest of its materializations, and each of
+//! those is affine in its number of readers from the class (rebuilt per
+//! reader when not materialized; built once at ingestion, or once per
+//! evaluation of the class's common recurrence, however many read it). A
+//! minimum of affine functions is concave, so splitting the class over
+//! several builds never costs less than putting all of it on the best one
+//! of them: some cheapest plan has every member read the same summary
+//! builds, with the same Pass 1 candidate. The default search keeps only
+//! those plans:
+//!
+//! - Pass 1: a whole-workload choice must give every member of a class the
+//!   same candidate for each of its targets ([`Classes::invalid_prefix`]);
+//! - Pass 2: a Stage 1 candidate must have every member of a class read the
+//!   same summary builds, by node ([`Classes::members_read_same_builds`]): the
+//!   same builds shared, in tumbling panes or whole alike (a build over a
+//!   member's quantile reads its rank and stays its own).
+//!
+//! The members' identical nodes are not made one node before the Pass 2
+//! rules run: the identical-expression rule runs after the others, and
+//! sharing first would reach plans in another order (two duplicates' builds
+//! shared, then capability-shared with a third query's, is not a plan of
+//! the exhaustive search and can cost more than the best one).
+//!
+//! The argument needs the sum of node charges: classes are formed only
+//! under cost model 1 without a measured cost table (whose calibration is
+//! fitted over every candidate), and a latency bound, which is not a
+//! charge, keeps a query out of every class.
+
+use std::collections::{HashMap, HashSet};
+use std::rc::Rc;
+
+use asap_logical_asap_optimization::asap_aware_cse::{sharing_candidates, with_union_scan_labels};
+use asap_logical_asap_optimization::per_computation_candidates::{
+    compose_logical_candidate, PerComputationCandidates,
+};
+use asap_types::ir::flat::flatten;
+use asap_types::ir::operator::AggIntent;
+use asap_types::ir::schema::SketchStatistic;
+use asap_types::ir::{ASAPOp, NonASAPOp, Operator, OperatorNode, QueryRoot};
+use asap_types::workload::QueryWorkloadEntry;
+
+use super::query_candidates::{next_digits, query_digits};
+use crate::{CostModel, PlanningModels};
+
+/// The classes of interchangeable queries of one workload, each of two or
+/// more queries.
+pub(crate) struct Classes {
+    /// Each class: its members' query indices, in workload order, and each
+    /// member's own digits of a whole-workload choice, in increasing order.
+    classes: Vec<Vec<(usize, Vec<usize>)>>,
+}
+
+impl Classes {
+    /// No class: every query chooses on its own.
+    pub(crate) fn none() -> Self {
+        Self {
+            classes: Vec::new(),
+        }
+    }
+
+    /// The classes of interchangeable queries of `per_computation` (the
+    /// Pass 1 inventory of the workload, already pruned for the deployment's
+    /// capabilities); none unless `models` prices by cost model 1 without a
+    /// measured cost table.
+    pub(crate) fn of<Id: Clone>(
+        per_computation: &PerComputationCandidates<Id>,
+        entries: &[QueryWorkloadEntry],
+        models: &PlanningModels<'_>,
+    ) -> Self {
+        let additive = matches!(
+            models.plan_selection.cost_model,
+            CostModel::UsageOverTime(_)
+        ) && models
+            .plan_selection
+            .measured_operation_costs
+            .operators
+            .is_empty();
+        if !additive {
+            return Self::none();
+        }
+        let digits: Vec<Vec<usize>> = per_computation
+            .roots
+            .iter()
+            .map(|(_, root)| query_digits(per_computation, root))
+            .collect();
+        // A query reading a target another query reads is in no class.
+        let own = |q: usize| {
+            !digits[q].is_empty()
+                && (0..digits.len())
+                    .all(|other| other == q || digits[other].iter().all(|d| !digits[q].contains(d)))
+        };
+        let mut classes: Vec<Vec<(usize, Vec<usize>)>> = Vec::new();
+        // Each class's first member's plan alone, rank-masked, per own choice.
+        let mut signatures: Vec<Vec<Option<serde_json::Value>>> = Vec::new();
+        for (q, entry) in entries.iter().enumerate() {
+            if entry.requirements.max_latency_ms.is_some() || !own(q) {
+                continue;
+            }
+            let signature = alone_signature(per_computation, &digits[q], q, true);
+            let found = classes
+                .iter()
+                .zip(&signatures)
+                .position(|(class, first_signature)| {
+                    let (first, first_digits) = &class[0];
+                    same_entry(&entries[*first], entry)
+                        && first_digits.len() == digits[q].len()
+                        && first_digits.iter().zip(&digits[q]).all(|(&a, &b)| {
+                            per_computation.targets[a].choice_count()
+                                == per_computation.targets[b].choice_count()
+                        })
+                        && *first_signature == signature
+                });
+            match found {
+                Some(c) => classes[c].push((q, digits[q].clone())),
+                None => {
+                    classes.push(vec![(q, digits[q].clone())]);
+                    signatures.push(signature);
+                }
+            }
+        }
+        // Members of different ranks are interchangeable only if no two of
+        // them have the same rank (a pair of duplicates can share its exact
+        // quantile, which a third member of another rank cannot join) and
+        // no other query can share a node that carries a member's rank (a
+        // ratio `q0.9 / q0.5` can share the 0.5 member's exact quantile but
+        // not the 0.9 member's). Duplicates within such a class form a class
+        // of their own.
+        let mut alike = Vec::new();
+        for class in classes {
+            if class.len() < 2 {
+                continue;
+            }
+            let ranked: Vec<Vec<Option<serde_json::Value>>> = class
+                .iter()
+                .map(|(m, member_digits)| {
+                    alone_signature(per_computation, member_digits, *m, false)
+                })
+                .collect();
+            if ranked.iter().all(|signature| *signature == ranked[0]) {
+                alike.push(class);
+                continue;
+            }
+            let mut same_rank: Vec<Vec<(usize, Vec<usize>)>> = Vec::new();
+            let mut rank_of: Vec<&Vec<Option<serde_json::Value>>> = Vec::new();
+            for (member, signature) in class.iter().zip(&ranked) {
+                match rank_of.iter().position(|seen| *seen == signature) {
+                    Some(r) => same_rank[r].push(member.clone()),
+                    None => {
+                        rank_of.push(signature);
+                        same_rank.push(vec![member.clone()]);
+                    }
+                }
+            }
+            if same_rank.iter().any(|members| members.len() > 1) {
+                alike.extend(same_rank);
+                continue;
+            }
+            let members: Vec<usize> = class.iter().map(|(m, _)| *m).collect();
+            alike.push(
+                class
+                    .into_iter()
+                    .filter(|(m, _)| {
+                        (0..entries.len())
+                            .filter(|x| !members.contains(x))
+                            .all(|x| !shares_a_rank_node(per_computation, &digits, x, *m))
+                    })
+                    .collect(),
+            );
+        }
+        let mut classes = alike;
+        classes.retain(|class| class.len() >= 2);
+        Self { classes }
+    }
+
+    /// `Some(n)` when every whole-workload choice that agrees with `choice`
+    /// on its first `n` digits gives two members of a class different
+    /// candidates; `None` when `choice` gives each class one.
+    pub(crate) fn invalid_prefix(&self, choice: &[usize]) -> Option<usize> {
+        self.classes
+            .iter()
+            .filter_map(|class| {
+                let (_, first) = &class[0];
+                class[1..]
+                    .iter()
+                    .filter_map(|(_, member)| {
+                        // The first digit position where this member differs.
+                        let position =
+                            (0..first.len()).find(|&i| choice[first[i]] != choice[member[i]])?;
+                        Some(first[position].max(member[position]) + 1)
+                    })
+                    .min()
+            })
+            .min()
+    }
+
+    /// Whether every member of each class reads the same summary builds
+    /// (`SummaryAgg` and `PaneChain` nodes, by node) in the Stage 1
+    /// candidate with `roots`, other than builds over a member's quantile
+    /// (e.g. the sums of an average of per-series quantiles), which read
+    /// its rank.
+    pub(crate) fn members_read_same_builds(&self, roots: &[QueryRoot]) -> bool {
+        if self.classes.is_empty() {
+            return true;
+        }
+        // Whether a rank lies at or below each node, by node.
+        let mut ranked: HashMap<*const OperatorNode, bool> = HashMap::new();
+        let builds = |q: usize, ranked: &mut HashMap<*const OperatorNode, bool>| {
+            root_nodes(&roots[q])
+                .into_iter()
+                .filter(|node| {
+                    matches!(
+                        node.asap(),
+                        Some(ASAPOp::SummaryAgg { .. } | ASAPOp::PaneChain { .. })
+                    ) && !rank_below(node, ranked)
+                })
+                .map(|node| Rc::as_ptr(&node))
+                .collect::<HashSet<_>>()
+        };
+        self.classes.iter().all(|class| {
+            let first = builds(class[0].0, &mut ranked);
+            class[1..]
+                .iter()
+                .all(|(q, _)| builds(*q, &mut ranked) == first)
+        })
+    }
+
+    /// Each class's members, by query index, for the search report.
+    pub(crate) fn members(&self) -> Vec<Vec<usize>> {
+        self.classes
+            .iter()
+            .map(|class| class.iter().map(|(q, _)| *q).collect())
+            .collect()
+    }
+}
+
+/// Whether `a` and `b` are the same workload entry but for the query text.
+fn same_entry(a: &QueryWorkloadEntry, b: &QueryWorkloadEntry) -> bool {
+    // Every field but the query text: a new field must be compared here.
+    let QueryWorkloadEntry {
+        query: _,
+        requirements,
+        predictability,
+        recurrence,
+        range_evaluation,
+    } = a;
+    *requirements == b.requirements
+        && *predictability == b.predictability
+        && *recurrence == b.recurrence
+        && *range_evaluation == b.range_evaluation
+}
+
+/// Every node `root` reaches.
+fn root_nodes(root: &QueryRoot) -> Vec<Rc<OperatorNode>> {
+    let starts: Vec<&Rc<OperatorNode>> = match root {
+        QueryRoot::Operator(node) => vec![node],
+        QueryRoot::Scalar(expr) => expr.operator_refs(),
+    };
+    starts
+        .into_iter()
+        .flat_map(OperatorNode::reachable)
+        .collect()
+}
+
+/// Whether a node carrying a quantile rank lies at or below `node`,
+/// memoized by node in `ranked`.
+fn rank_below(node: &Rc<OperatorNode>, ranked: &mut HashMap<*const OperatorNode, bool>) -> bool {
+    if let Some(&known) = ranked.get(&Rc::as_ptr(node)) {
+        return known;
+    }
+    let below = carries_a_rank(node)
+        || node
+            .children()
+            .into_iter()
+            .any(|child| rank_below(child, ranked));
+    ranked.insert(Rc::as_ptr(node), below);
+    below
+}
+
+/// Query `q` of `per_computation` planned alone with each choice of its own
+/// candidates (over `digits`, in mixed-radix order): the composed plan, with
+/// every quantile rank masked when `masked`, or `None` when it does not
+/// compose.
+fn alone_signature<Id: Clone>(
+    per_computation: &PerComputationCandidates<Id>,
+    digits: &[usize],
+    q: usize,
+    masked: bool,
+) -> Vec<Option<serde_json::Value>> {
+    let alone = PerComputationCandidates {
+        roots: vec![per_computation.roots[q].clone()],
+        targets: per_computation.targets.clone(),
+    };
+    let mut choice = vec![0; per_computation.targets.len()];
+    let mut signature = Vec::new();
+    loop {
+        signature.push(
+            compose_logical_candidate(&alone, &choice)
+                .ok()
+                .and_then(|composed| {
+                    let roots: Vec<QueryRoot> =
+                        composed.into_iter().map(|(_, root)| root).collect();
+                    let mut flat = serde_json::to_value(flatten(&roots).0).ok()?;
+                    if masked {
+                        mask_quantile_ranks(&mut flat);
+                    }
+                    Some(flat)
+                }),
+        );
+        if !next_digits(per_computation, digits, &mut choice) {
+            return signature;
+        }
+    }
+}
+
+/// Whether queries `x` and `m` of `per_computation`, for some choice of
+/// their own candidates (over `digits[x]` and `digits[m]`) composed
+/// together, have identical nodes that carry a quantile rank (with each
+/// metric's scans widened to one label set): a node the identical-
+/// expression rule could make one.
+fn shares_a_rank_node<Id: Clone>(
+    per_computation: &PerComputationCandidates<Id>,
+    digits: &[Vec<usize>],
+    x: usize,
+    m: usize,
+) -> bool {
+    let pair = PerComputationCandidates {
+        roots: vec![
+            per_computation.roots[x].clone(),
+            per_computation.roots[m].clone(),
+        ],
+        targets: per_computation.targets.clone(),
+    };
+    let pair_digits: Vec<usize> = digits[x].iter().chain(&digits[m]).copied().collect();
+    let mut choice = vec![0; per_computation.targets.len()];
+    loop {
+        if let Ok(composed) = compose_logical_candidate(&pair, &choice) {
+            let roots: Vec<QueryRoot> = composed.into_iter().map(|(_, root)| root).collect();
+            let widened = with_union_scan_labels(&roots).ok().flatten();
+            for roots in std::iter::once(&roots).chain(widened.as_ref()) {
+                if sharing_candidates(roots)
+                    .iter()
+                    .any(|candidate| carries_a_rank(&candidate.nodes[0]))
+                {
+                    return true;
+                }
+            }
+        }
+        if !next_digits(per_computation, &pair_digits, &mut choice) {
+            return false;
+        }
+    }
+}
+
+/// Whether `node` computes or estimates a quantile at some rank.
+fn carries_a_rank(node: &OperatorNode) -> bool {
+    match &node.operator {
+        Operator::ASAP(ASAPOp::SummaryEstimate { query, .. }) => {
+            matches!(query, SketchStatistic::Quantile { .. })
+        }
+        Operator::NonASAP(NonASAPOp::Aggregate { measures, .. }) => measures
+            .iter()
+            .any(|measure| matches!(measure, AggIntent::Quantile { .. })),
+        _ => false,
+    }
+}
+
+/// `value`, a serialized plan, with the rank `q` of every quantile (a
+/// quantile measure's or a sketch estimate's, and the estimate a
+/// guarantee's provenance names) masked.
+fn mask_quantile_ranks(value: &mut serde_json::Value) {
+    match value {
+        serde_json::Value::Object(map) => {
+            // A sketch estimate's `{"Quantile": {"q": ..}}`.
+            if let Some(serde_json::Value::Object(quantile)) = map.get_mut("Quantile") {
+                if let Some(q) = quantile.get_mut("q") {
+                    *q = serde_json::Value::from(0);
+                }
+            }
+            // A quantile measure's `{"kind": "quantile", "q": .., ..}`.
+            if map.get("kind").and_then(serde_json::Value::as_str) == Some("quantile") {
+                if let Some(q) = map.get_mut("q") {
+                    *q = serde_json::Value::from(0);
+                }
+            }
+            map.values_mut().for_each(mask_quantile_ranks);
+        }
+        // A guarantee's provenance names the estimate it evaluates.
+        serde_json::Value::String(text) if text.starts_with("Quantile { q: ") => {
+            *text = "Quantile".to_string();
+        }
+        serde_json::Value::Array(items) => items.iter_mut().for_each(mask_quantile_ranks),
+        _ => {}
+    }
+}
diff --git a/crates/plan-selection/src/pass/mod.rs b/crates/plan-selection/src/pass/mod.rs
index 90da95f0d..d95cbef92 100644
--- a/crates/plan-selection/src/pass/mod.rs
+++ b/crates/plan-selection/src/pass/mod.rs
@@ -13,6 +13,7 @@
 //! downstream consumers rely on.
 
 mod components;
+mod interchangeable;
 mod query_candidates;
 mod stage1_to_stage3;
 
diff --git a/crates/plan-selection/src/pass/query_candidates.rs b/crates/plan-selection/src/pass/query_candidates.rs
index 1647bb116..03d0ab1aa 100644
--- a/crates/plan-selection/src/pass/query_candidates.rs
+++ b/crates/plan-selection/src/pass/query_candidates.rs
@@ -143,7 +143,7 @@ impl QueryCandidates {
 /// The digits of a whole-workload choice that decide `root`'s plan: the
 /// targets whose aggregate it reads, and the targets whose fused
 /// computation absorbs one of those, in increasing order.
-fn query_digits<Id>(
+pub(super) fn query_digits<Id>(
     per_computation: &PerComputationCandidates<Id>,
     root: &QueryRoot,
 ) -> Vec<usize> {
@@ -174,7 +174,7 @@ fn query_digits<Id>(
 
 /// Mixed-radix increment of `choice` over `digits` only (the last varying
 /// fastest); `false` after the last combination.
-fn next_digits<Id>(
+pub(super) fn next_digits<Id>(
     per_computation: &PerComputationCandidates<Id>,
     digits: &[usize],
     choice: &mut [usize],
diff --git a/crates/plan-selection/src/pass/stage1_to_stage3.rs b/crates/plan-selection/src/pass/stage1_to_stage3.rs
index 51b71f2e1..388358aa7 100644
--- a/crates/plan-selection/src/pass/stage1_to_stage3.rs
+++ b/crates/plan-selection/src/pass/stage1_to_stage3.rs
@@ -52,6 +52,7 @@ use asap_types::ir::schema::schema_support::with_promql_series_identity;
 use asap_types::ir::{OperatorNode, QueryRoot};
 use asap_types::workload::{DataWorkload, QueryLanguage, QueryWorkloadEntry};
 
+use super::interchangeable::Classes;
 use super::query_candidates::QueryCandidates;
 use super::{OptimizationInput, OptimizationPass, OptimizeError, PlanOutput, QueryPlan};
 use crate::PlanningModels;
@@ -227,13 +228,7 @@ pub fn run_stage1_to_stage3_from_pass1<Id: Clone + std::fmt::Debug>(
     models: PlanningModels<'_>,
     search: Stage1ToStage3Search,
 ) -> Stage1ToStage3Run<Id> {
-    let components = super::components::source_components(
-        &per_computation
-            .roots
-            .iter()
-            .map(|(_, root)| root.clone())
-            .collect::<Vec<_>>(),
-    );
+    let components = super::components::sharing_components(&per_computation, entries, &models);
     let searched_as_one = match super::components::searched_as_one(&components, &models, search) {
         Some(reason) => reason,
         None => match super::components::search_components(
@@ -288,6 +283,7 @@ fn search_as_one<Id: Clone + std::fmt::Debug>(
         return run.with_inventory(per_computation);
     }
     let queries = QueryCandidates::check(&per_computation, entries, data, models);
+    let classes = Classes::of(&per_computation, entries, &models);
     // Stage 1 candidates left out because every assignment of theirs scans
     // raw data at query time.
     let stage1_raw_skipped = std::cell::Cell::new(0u64);
@@ -296,7 +292,13 @@ fn search_as_one<Id: Clone + std::fmt::Debug>(
     let stage1 = stage1_logical_asap_optimization_skipping(
         &per_computation,
         entries,
-        |choice| queries.invalid_prefix(choice),
+        |choice| match (
+            queries.invalid_prefix(choice),
+            classes.invalid_prefix(choice),
+        ) {
+            (Some(a), Some(b)) => Some(a.min(b)),
+            (a, b) => a.or(b),
+        },
         |roots| {
             let viable = viable(roots, entries, data, models, skip_raw);
             if !viable {
@@ -313,7 +315,14 @@ fn search_as_one<Id: Clone + std::fmt::Debug>(
         data,
         models,
     );
-    let run = search_stage2_and_stage3(
+    // Members of a class of interchangeable queries read the same builds.
+    let stage1 = stage1.filter(|candidate| {
+        candidate
+            .roots
+            .as_ref()
+            .map_or(true, |roots| classes.members_read_same_builds(roots))
+    });
+    let mut run = search_stage2_and_stage3(
         &per_computation,
         stage1,
         entries,
@@ -325,6 +334,9 @@ fn search_as_one<Id: Clone + std::fmt::Debug>(
         skip_raw,
         &stage1_raw_skipped,
     );
+    if let Ok(selection) = &mut run.selection {
+        selection.search.interchangeable_queries = classes.members();
+    }
     run.with_inventory(per_computation)
 }
 
diff --git a/tools/dag-viewer/examples/random-mixed-seed-862-long-workload-01.json b/tools/dag-viewer/examples/random-mixed-seed-862-long-workload-01.json
index 51babd612..3c75a84b6 100644
--- a/tools/dag-viewer/examples/random-mixed-seed-862-long-workload-01.json
+++ b/tools/dag-viewer/examples/random-mixed-seed-862-long-workload-01.json
@@ -9045,7 +9045,7 @@
           ]
         },
         "display_name": "Q1 exact (nested Sum acc) · Q2 exact (nested Sum acc) · Q3 DDSketch (Sum/Count acc) · Q4 DDSketch (Sum/Count acc) · Q5 DDSketch (Sum/Count acc) · Q6 exact (Sum acc) · Q7 exact (Sum acc) · Q8 exact (Increase acc) · Q9 exact (Rate acc) · one DDSketch · Sum acc for Q1, Q2 in 90 × 20s panes, DDSketch for Q3, Q4, Q5 in 300 × 1s panes, Sum acc for Q6 in 180 × 1s panes, Sum acc for Q7 in 120 × 30s panes, Increase acc for Q8 in 90 × 20s panes, Rate acc for Q9 in 120 × 30s panes · Sum acc rolled up · shared PaneChain (Q3, Q4, Q5), SummaryMerge (Q3, Q4, Q5)",
-        "id": "L1"
+        "id": "L315"
       }
     ],
     "capped": false,
@@ -19654,15 +19654,15 @@
           "scalar_roots": []
         },
         "display_name": "Q1 exact (nested Sum acc) · Q2 exact (nested Sum acc) · Q3 DDSketch (Sum/Count acc) · Q4 DDSketch (Sum/Count acc) · Q5 DDSketch (Sum/Count acc) · Q6 exact (Sum acc) · Q7 exact (Sum acc) · Q8 exact (Increase acc) · Q9 exact (Rate acc) · one DDSketch · Sum acc for Q1, Q2 in 90 × 20s panes, DDSketch for Q3, Q4, Q5 in 300 × 1s panes, Sum acc for Q6 in 180 × 1s panes, Sum acc for Q7 in 120 × 30s panes, Increase acc for Q8 in 90 × 20s panes, Rate acc for Q9 in 120 × 30s panes · Sum acc rolled up · shared PaneChain (Q3, Q4, Q5), SummaryMerge (Q3, Q4, Q5) · ingestion time: Sum acc ×90 panes, DDSketch ×300 panes, Sum acc ×180 panes, Sum acc ×120 panes, Increase acc ×90 panes, Rate acc ×120 panes · query time, shared: FinalizeExactAccumulator, SummaryMerge",
-        "from_logical": "L1",
-        "id": "P390-m79+P1-m3"
+        "from_logical": "L315",
+        "id": "P315-m319"
       }
     ],
     "pruned": []
   },
   "stage3_plan_selection": {
     "accepted": {
-      "P390-m79+P1-m3": [
+      "P315-m319": [
         {
           "accepted_metric": null,
           "accepted_regardless_of_guarantee": false,
@@ -19744,7 +19744,7 @@
       ]
     },
     "costs": {
-      "P390-m79+P1-m3": {
+      "P315-m319": {
         "cost_per_second": 4.5581790928001605e-6,
         "cost_unit": "cost_per_second",
         "per_node": {
@@ -20200,29 +20200,13 @@
     "rejected": [],
     "rejected_by_cause": {},
     "search": {
-      "candidates_checked": 1103,
-      "components": [
-        {
-          "candidates_checked": 1100,
-          "queries": [
-            0,
-            1,
-            2,
-            3,
-            4,
-            5,
-            6
-          ],
-          "selected": "P390-m79"
-        },
-        {
-          "candidates_checked": 2,
-          "queries": [
-            7,
-            8
-          ],
-          "selected": "P1-m3"
-        }
+      "candidates_checked": 933,
+      "components": [],
+      "interchangeable_queries": [
+        [
+          2,
+          4
+        ]
       ],
       "queries": [
         {
@@ -20365,13 +20349,13 @@
           "discarded": []
         }
       ],
-      "raw_scan_assignments_skipped": 71257,
-      "raw_scan_candidates_skipped": 0,
-      "runner_up_cost_per_second": null,
-      "searched_as_one": null
+      "raw_scan_assignments_skipped": 185847,
+      "raw_scan_candidates_skipped": 10752,
+      "runner_up_cost_per_second": 4.558196002856508e-6,
+      "searched_as_one": "not_in_joint_space"
     },
     "search_rejected": [],
-    "selected": "P390-m79+P1-m3",
+    "selected": "P315-m319",
     "stored_outputs": [
       {
         "built_at": "ingestion_run",
diff --git a/tools/dag-viewer/examples/random-mixed-seed-862-workload-01.json b/tools/dag-viewer/examples/random-mixed-seed-862-workload-01.json
index a9bba1683..870ec1441 100644
--- a/tools/dag-viewer/examples/random-mixed-seed-862-workload-01.json
+++ b/tools/dag-viewer/examples/random-mixed-seed-862-workload-01.json
@@ -9045,7 +9045,7 @@
           ]
         },
         "display_name": "Q1 exact (nested Sum acc) · Q2 exact (nested Sum acc) · Q3 DDSketch (Sum/Count acc) · Q4 DDSketch (Sum/Count acc) · Q5 DDSketch (Sum/Count acc) · Q6 exact (Sum acc) · Q7 exact (Sum acc) · Q8 exact (Increase acc) · Q9 exact (Rate acc) · one DDSketch · Sum acc for Q1, Q2 in 30 × 20s panes, DDSketch for Q3, Q4, Q5 in 100 × 1s panes, Sum acc for Q6 in 60 × 1s panes, Sum acc for Q7 in 40 × 30s panes, Increase acc for Q8 in 30 × 20s panes, Rate acc for Q9 in 40 × 30s panes · Sum acc rolled up · shared PaneChain (Q3, Q4, Q5), SummaryMerge (Q3, Q4, Q5)",
-        "id": "L1"
+        "id": "L315"
       }
     ],
     "capped": false,
@@ -19654,15 +19654,15 @@
           "scalar_roots": []
         },
         "display_name": "Q1 exact (nested Sum acc) · Q2 exact (nested Sum acc) · Q3 DDSketch (Sum/Count acc) · Q4 DDSketch (Sum/Count acc) · Q5 DDSketch (Sum/Count acc) · Q6 exact (Sum acc) · Q7 exact (Sum acc) · Q8 exact (Increase acc) · Q9 exact (Rate acc) · one DDSketch · Sum acc for Q1, Q2 in 30 × 20s panes, DDSketch for Q3, Q4, Q5 in 100 × 1s panes, Sum acc for Q6 in 60 × 1s panes, Sum acc for Q7 in 40 × 30s panes, Increase acc for Q8 in 30 × 20s panes, Rate acc for Q9 in 40 × 30s panes · Sum acc rolled up · shared PaneChain (Q3, Q4, Q5), SummaryMerge (Q3, Q4, Q5) · ingestion time: Sum acc ×30 panes, DDSketch ×100 panes, Sum acc ×60 panes, Sum acc ×40 panes, Increase acc ×30 panes, Rate acc ×40 panes · query time, shared: FinalizeExactAccumulator, SummaryMerge",
-        "from_logical": "L1",
-        "id": "P390-m79+P1-m3"
+        "from_logical": "L315",
+        "id": "P315-m319"
       }
     ],
     "pruned": []
   },
   "stage3_plan_selection": {
     "accepted": {
-      "P390-m79+P1-m3": [
+      "P315-m319": [
         {
           "accepted_metric": null,
           "accepted_regardless_of_guarantee": false,
@@ -19744,7 +19744,7 @@
       ]
     },
     "costs": {
-      "P390-m79+P1-m3": {
+      "P315-m319": {
         "cost_per_second": 1.5764190368555072e-6,
         "cost_unit": "cost_per_second",
         "per_node": {
@@ -20200,29 +20200,13 @@
     "rejected": [],
     "rejected_by_cause": {},
     "search": {
-      "candidates_checked": 1103,
-      "components": [
-        {
-          "candidates_checked": 1100,
-          "queries": [
-            0,
-            1,
-            2,
-            3,
-            4,
-            5,
-            6
-          ],
-          "selected": "P390-m79"
-        },
-        {
-          "candidates_checked": 2,
-          "queries": [
-            7,
-            8
-          ],
-          "selected": "P1-m3"
-        }
+      "candidates_checked": 933,
+      "components": [],
+      "interchangeable_queries": [
+        [
+          2,
+          4
+        ]
       ],
       "queries": [
         {
@@ -20365,13 +20349,13 @@
           "discarded": []
         }
       ],
-      "raw_scan_assignments_skipped": 71257,
-      "raw_scan_candidates_skipped": 0,
-      "runner_up_cost_per_second": null,
-      "searched_as_one": null
+      "raw_scan_assignments_skipped": 185847,
+      "raw_scan_candidates_skipped": 10752,
+      "runner_up_cost_per_second": 1.5764359193046939e-6,
+      "searched_as_one": "not_in_joint_space"
     },
     "search_rejected": [],
-    "selected": "P390-m79+P1-m3",
+    "selected": "P315-m319",
     "stored_outputs": [
       {
         "built_at": "ingestion_run",
diff --git a/tools/dag-viewer/examples/random-mixed-seed-862-x5-workload-01.json b/tools/dag-viewer/examples/random-mixed-seed-862-x5-workload-01.json
index ec6ff224f..55661c497 100644
--- a/tools/dag-viewer/examples/random-mixed-seed-862-x5-workload-01.json
+++ b/tools/dag-viewer/examples/random-mixed-seed-862-x5-workload-01.json
@@ -9045,7 +9045,7 @@
           ]
         },
         "display_name": "Q1 exact (nested Sum acc) · Q2 exact (nested Sum acc) · Q3 DDSketch (Sum/Count acc) · Q4 DDSketch (Sum/Count acc) · Q5 DDSketch (Sum/Count acc) · Q6 exact (Sum acc) · Q7 exact (Sum acc) · Q8 exact (Increase acc) · Q9 exact (Rate acc) · one DDSketch · Sum acc for Q1, Q2 in 30 × 20s panes, DDSketch for Q3, Q4, Q5 in 100 × 1s panes, Sum acc for Q6 in 60 × 1s panes, Sum acc for Q7 in 40 × 30s panes, Increase acc for Q8 in 30 × 20s panes, Rate acc for Q9 in 40 × 30s panes · Sum acc rolled up · shared PaneChain (Q3, Q4, Q5), SummaryMerge (Q3, Q4, Q5)",
-        "id": "L1"
+        "id": "L315"
       }
     ],
     "capped": false,
@@ -19654,15 +19654,15 @@
           "scalar_roots": []
         },
         "display_name": "Q1 exact (nested Sum acc) · Q2 exact (nested Sum acc) · Q3 DDSketch (Sum/Count acc) · Q4 DDSketch (Sum/Count acc) · Q5 DDSketch (Sum/Count acc) · Q6 exact (Sum acc) · Q7 exact (Sum acc) · Q8 exact (Increase acc) · Q9 exact (Rate acc) · one DDSketch · Sum acc for Q1, Q2 in 30 × 20s panes, DDSketch for Q3, Q4, Q5 in 100 × 1s panes, Sum acc for Q6 in 60 × 1s panes, Sum acc for Q7 in 40 × 30s panes, Increase acc for Q8 in 30 × 20s panes, Rate acc for Q9 in 40 × 30s panes · Sum acc rolled up · shared PaneChain (Q3, Q4, Q5), SummaryMerge (Q3, Q4, Q5) · ingestion time: Sum acc ×30 panes, DDSketch ×100 panes, Sum acc ×60 panes, Sum acc ×40 panes, Increase acc ×30 panes, Rate acc ×40 panes · query time, shared: FinalizeExactAccumulator, SummaryMerge",
-        "from_logical": "L1",
-        "id": "P390-m79+P1-m3"
+        "from_logical": "L315",
+        "id": "P315-m319"
       }
     ],
     "pruned": []
   },
   "stage3_plan_selection": {
     "accepted": {
-      "P390-m79+P1-m3": [
+      "P315-m319": [
         {
           "accepted_metric": null,
           "accepted_regardless_of_guarantee": false,
@@ -19744,7 +19744,7 @@
       ]
     },
     "costs": {
-      "P390-m79+P1-m3": {
+      "P315-m319": {
         "cost_per_second": 7.882484714275631e-6,
         "cost_unit": "cost_per_second",
         "per_node": {
@@ -20200,29 +20200,13 @@
     "rejected": [],
     "rejected_by_cause": {},
     "search": {
-      "candidates_checked": 1103,
-      "components": [
-        {
-          "candidates_checked": 1100,
-          "queries": [
-            0,
-            1,
-            2,
-            3,
-            4,
-            5,
-            6
-          ],
-          "selected": "P390-m79"
-        },
-        {
-          "candidates_checked": 2,
-          "queries": [
-            7,
-            8
-          ],
-          "selected": "P1-m3"
-        }
+      "candidates_checked": 933,
+      "components": [],
+      "interchangeable_queries": [
+        [
+          2,
+          4
+        ]
       ],
       "queries": [
         {
@@ -20365,13 +20349,13 @@
           "discarded": []
         }
       ],
-      "raw_scan_assignments_skipped": 71257,
-      "raw_scan_candidates_skipped": 0,
-      "runner_up_cost_per_second": null,
-      "searched_as_one": null
+      "raw_scan_assignments_skipped": 185847,
+      "raw_scan_candidates_skipped": 10752,
+      "runner_up_cost_per_second": 7.88256945547401e-6,
+      "searched_as_one": "not_in_joint_space"
     },
     "search_rejected": [],
-    "selected": "P390-m79+P1-m3",
+    "selected": "P315-m319",
     "stored_outputs": [
       {
         "built_at": "ingestion_run",