From 572ba4a24bbd2e62f76f95841fb2f7b5b363125d Mon Sep 17 00:00:00 2001
From: zzylol <50204836+zzylol@users.noreply.github.com>
Date: Wed, 28 Oct 2026 20:39:41 +0000
Subject: [PATCH 1/6] test(memo): equal keyless deterministic results should
 share one group

The per-series quantile in #1011's example has no provable unique key, so
CSE keeps G2 and G3 as two groups even though both readers of G2 / G3 read
the same rows. The new memo test fails on main (groups 2 and 3).

A guard test pins that an unordered LIMIT, whose copies may return
different rows, stays unshared after the legality rule changes.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 crates/logical-asap-optimization/src/memo.rs | 32 +++++++++++++++++++-
 crates/types/src/ir/cse.rs                   | 16 ++++++++++
 2 files changed, 47 insertions(+), 1 deletion(-)

diff --git a/crates/logical-asap-optimization/src/memo.rs b/crates/logical-asap-optimization/src/memo.rs
index fe57bb5a2..21d39da43 100644
--- a/crates/logical-asap-optimization/src/memo.rs
+++ b/crates/logical-asap-optimization/src/memo.rs
@@ -235,8 +235,10 @@ fn pair(
 mod tests {
     use super::*;
     use crate::test_support::lower_promql;
+    use asap_types::ir::operator::operator_properties::BinaryOpKind;
     use asap_types::ir::properties::AccuracyTarget;
-    use asap_types::ir::ScalarExpr;
+    use asap_types::ir::scalar::expr_ir::ArithmeticOpKind;
+    use asap_types::ir::{NonASAPOp, ScalarExpr};
 
     fn query(text: &str) -> QueryRoot {
         QueryRoot::Operator(lower_promql(text, AccuracyTarget::max_error(0.01)))
@@ -273,6 +275,34 @@ mod tests {
         }
     }
 
+    /// A per-series quantile has no provable unique key, but it is
+    /// deterministic: both readers of `G2 / G3` read the same rows, so the
+    /// two equal sub-expressions are one group read twice.
+    #[test]
+    fn equal_keyless_deterministic_results_share_one_group() {
+        let roots = vec![(
+            0,
+            QueryRoot::Operator(lower_promql(
+                "avg by (job) (quantile_over_time(0.5, data[5m]) / quantile_over_time(0.5, data[5m]) + quantile_over_time(0.9, data[5m]))",
+                AccuracyTarget::max_error(0.01).failure_probability(0.001),
+            )),
+        )];
+        let memo = LogicalMemo::from_roots(&roots);
+        let division = memo
+            .groups
+            .iter()
+            .find(|group| {
+                matches!(group.node.non_asap(), Some(NonASAPOp::BinaryOp { operator, .. })
+                    if operator.kind == BinaryOpKind::Arithmetic(ArithmeticOpKind::Div))
+            })
+            .expect("a `G2 / G3` group");
+        assert_eq!(
+            division.children[0], division.children[1],
+            "both operands of the division are one group"
+        );
+        assert_eq!(memo.groups.len(), 7);
+    }
+
     /// Scalar references are roots of their query; literals need no invented node.
     #[test]
     fn scalar_and_literal_roots_keep_workload_positions() {
diff --git a/crates/types/src/ir/cse.rs b/crates/types/src/ir/cse.rs
index a6a91b768..940889c9f 100644
--- a/crates/types/src/ir/cse.rs
+++ b/crates/types/src/ir/cse.rs
@@ -746,6 +746,22 @@ mod tests {
         assert!(Rc::ptr_eq(&ra, &rb));
     }
 
+    /// `LIMIT 10` without an order may return any 10 rows, so two copies may
+    /// differ; one shared copy would make both readers see the same rows.
+    #[test]
+    fn unordered_limits_are_not_shared() {
+        let limit = || {
+            node(NonASAPOp::Limit {
+                n: Some(10),
+                offset: 0,
+                partition_by: GroupKeys::default(),
+                child: scan(),
+            })
+        };
+        let (ra, rb) = two_roots(limit(), limit());
+        assert!(!Rc::ptr_eq(&ra, &rb));
+    }
+
     #[test]
     fn group_keys_gate_still_prevented_when_partition_by_without_used() {
         let without_agg = || {

From 7609056438650f7a69c34b70987d2335b5035c81 Mon Sep 17 00:00:00 2001
From: zzylol <50204836+zzylol@users.noreply.github.com>
Date: Wed, 28 Oct 2026 21:15:20 +0000
Subject: [PATCH 2/6] fix(memo): share equal nodes that return the same rows
 for the same input

Group identity no longer needs a provable unique key. A non-ASAP node is
legal to share when every execution over the same input rows returns the
same output rows (NonASAPOp::returns_same_rows_for_same_input_rows); the
hard-coded list of shareable shapes goes. Tie order of a stable Sort
counts as fixed, so exact top-k and a Limit over a Sort stay shared; an
unordered Limit, a Dedup on a column subset and a position-reading
window function without ORDER BY are never shared.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../src/memo/groups.rs                        |  18 +--
 .../src/memo/shared_producer_snapshot.rs      |   7 +-
 crates/types/src/ir/cse.rs                    | 112 +++++++++---------
 crates/types/src/ir/operator/non_asap.rs      |  61 +++++++++-
 .../design_docs/proposals/planner-layering.md |  12 +-
 5 files changed, 141 insertions(+), 69 deletions(-)

diff --git a/crates/logical-asap-optimization/src/memo/groups.rs b/crates/logical-asap-optimization/src/memo/groups.rs
index 5fac56fc6..4234426c0 100644
--- a/crates/logical-asap-optimization/src/memo/groups.rs
+++ b/crates/logical-asap-optimization/src/memo/groups.rs
@@ -848,9 +848,9 @@ mod tests {
 
     /// `q0.9 / q0.5` beside a plain `q0.9`: each of the ratio's quantile
     /// groups offers every sketch for the whole budget and for half of it,
-    /// since its query has two inputs that may be approximate; the plain
-    /// query's quantile group (its own: its unkeyed window is not shared)
-    /// offers only the whole budget. A ratio of quantiles carries no
+    /// since its query has two inputs that may be approximate. The plain
+    /// query's quantile is the numerator's group: equal per-series quantiles
+    /// return the same rows, so they are one group with both readers. A ratio of quantiles carries no
     /// guarantee, so composition does not split its budget: the
     /// whole-budget candidate is the exhaustive search's sizing, and the
     /// half-budget one, larger, is never the cheaper valid one.
@@ -875,12 +875,12 @@ mod tests {
                 .collect();
             assert_eq!(shares, vec![1, 2, 1, 2]);
         }
-        let plain = group_of_kind(&memo, 1, "Aggregate");
-        let plain_shares: Vec<usize> = sketch_shares(&memo, plain)
-            .into_iter()
-            .map(|(_, share)| share)
-            .collect();
-        assert_eq!(plain_shares, vec![1, 1]);
+        let plain = memo
+            .groups
+            .iter()
+            .position(|group| group.roots_of == vec![1])
+            .unwrap();
+        assert_eq!(plain, operand);
         let whole_kll = &sketch_shares(&memo, operand)[0];
         assert_eq!(whole_kll.1, 1);
         assert_eq!(
diff --git a/crates/logical-asap-optimization/src/memo/shared_producer_snapshot.rs b/crates/logical-asap-optimization/src/memo/shared_producer_snapshot.rs
index 1a2755de6..179945fe3 100644
--- a/crates/logical-asap-optimization/src/memo/shared_producer_snapshot.rs
+++ b/crates/logical-asap-optimization/src/memo/shared_producer_snapshot.rs
@@ -1,6 +1,9 @@
 //! Pins the candidates the shared-producer cases generate, recorded before
 //! they moved onto one generation path: each workload's expanded memo
 //! (groups, candidates, their origins and children) hashes to a fixed value.
+//! The top-k pins were re-recorded when equal keyless results became one
+//! group: the readers' shared `sum_over_time` is now one group, not one per
+//! reader.
 use super::candidates::{try_populate_pass1, CandidateMemo, CandidateOrigin};
 use super::state_configurations::PaneWidthDomain;
 use super::state_groups::expand_concrete_summary_state_groups_in_domain;
@@ -112,7 +115,7 @@ fn top_k_reuse_candidates_are_unchanged() {
         .iter()
         .flatten()
         .any(|candidate| candidate.origin == CandidateOrigin::TopKReuse));
-    assert_eq!(fingerprint(&memo), (313, 16940643818053118362));
+    assert_eq!(fingerprint(&memo), (69, 14825718632741491576));
 }
 
 /// Quantile readers of one input at different sizes.
@@ -161,5 +164,5 @@ fn several_top_k_sets_are_unchanged() {
         ],
         60_000,
     );
-    assert_eq!(fingerprint(&memo), (994, 9213568444657927922));
+    assert_eq!(fingerprint(&memo), (504, 3265274154981812875));
 }
diff --git a/crates/types/src/ir/cse.rs b/crates/types/src/ir/cse.rs
index 940889c9f..bd819c93c 100644
--- a/crates/types/src/ir/cse.rs
+++ b/crates/types/src/ir/cse.rs
@@ -35,18 +35,15 @@
 //! ## Legality
 //!
 //! Structural equality is necessary but not sufficient. A non-ASAP node is
-//! only ever *returned* as a match for another when its output has a provable
-//! unique key (`Schema::has_unique_key()`): a producer's output can only be
-//! shared across consumers when its row identity is stable across reads, so
-//! a `without(..)` grouping, a `Concat`/`SetOp` that
-//! drops its keys, … is always inserted fresh even when it is structurally
-//! identical to something already interned. A `Scan` is the exception: equal
-//! scans read the same source rows, so every reader of one shared scan reads
-//! the rows it would have read alone, a SQL table's keyless rows included.
-//! Global scalar reductions are also shareable: their single result has stable
-//! identity even though the schema does not declare the empty unique key.
-//! Equal windows directly over scans and exact TopK retaining explicit series
-//! identity also preserve the identity of their selected rows.
+//! only ever *returned* as a match for another when every execution over the
+//! same input rows returns the same output rows
+//! ([`NonASAPOp::returns_same_rows_for_same_input_rows`]). Equal nodes are
+//! interned children first, so two equal nodes read one input, and one copy
+//! returns each reader the rows it would have computed alone. A unique key
+//! is not needed: equal per-series quantiles without a provable key are one
+//! node. An operator that picks rows by input order alone (an unordered
+//! `LIMIT`) is always inserted fresh, because two copies may return
+//! different rows.
 //! An ASAP node (summary state and its evaluations) has no such gate: equal
 //! operator, schema and guarantee make it shareable, exactly as post-ASAP
 //! sharing decided before this IR.
@@ -216,33 +213,9 @@ impl InternTable {
         // for something else; it still occupies a fresh slot in the bucket
         // (harmless: later scans require legality of the new node too).
         let reusable = node.is_asap()
-            || node.schema.has_unique_key()
-            || matches!(node.non_asap(), Some(NonASAPOp::Scan { .. }))
-            // Equal windows over one scan preserve its row identities, even
-            // for keyless SQL tables. Consumers can share that window's rows.
-            || matches!(node.non_asap(), Some(NonASAPOp::TimeRange { child, .. })
-                if matches!(child.non_asap(), Some(NonASAPOp::Scan { .. })))
-            // Exact PromQL TopK retains each selected series' identity. The
-            // ranked projection can drop the timestamp part of the input key,
-            // but identical selections still return the same series rows.
-            || (node.result_kind == crate::ir::OperatorResultKind::InstantVector
-                && node.schema.fields.iter().any(|field| field.name == crate::ir::schema::PROMQL_SERIES_IDENTITY)
-                && matches!(node.non_asap(), Some(NonASAPOp::Aggregate { measures, .. })
-                    if matches!(measures.as_slice(), [crate::ir::operator::AggIntent::TopK { .. }])))
-            || matches!(node.non_asap(), Some(NonASAPOp::Aggregate {
-                reduction: crate::ir::operator::Reduction::Reduce(keys), measures, ..
-            }) if !keys.is_without() && keys.is_empty() && measures.iter().all(|measure| matches!(measure,
-                crate::ir::operator::AggIntent::Count { .. }
-                | crate::ir::operator::AggIntent::Sum { .. }
-                | crate::ir::operator::AggIntent::Min { .. }
-                | crate::ir::operator::AggIntent::Max { .. }
-                | crate::ir::operator::AggIntent::Avg { .. }
-                | crate::ir::operator::AggIntent::StdDev { .. }
-                | crate::ir::operator::AggIntent::Variance { .. }
-                | crate::ir::operator::AggIntent::PearsonCorr { .. }
-                | crate::ir::operator::AggIntent::Quantile { .. }
-                | crate::ir::operator::AggIntent::Cardinality { .. }
-            )));
+            || node
+                .non_asap()
+                .is_some_and(NonASAPOp::returns_same_rows_for_same_input_rows);
         let bucket = self.buckets.entry(hash).or_default();
         if reusable {
             if let Some(existing) = bucket
@@ -592,8 +565,8 @@ mod tests {
             "the vector read by scalar(v) is a child and must be interned"
         );
         assert!(
-            !Rc::ptr_eq(&ra, &rb),
-            "the scalar bridge itself has no unique key and stays separate"
+            Rc::ptr_eq(&ra, &rb),
+            "the scalar bridge has no unique key but returns the same rows"
         );
     }
 
@@ -732,18 +705,20 @@ mod tests {
         );
     }
 
+    /// `DISTINCT` (every column) returns the same rows each time; a `Dedup`
+    /// on some columns keeps an unspecified row per key.
     #[test]
-    fn dedup_gates_sharing_the_same_as_aggregate() {
-        // `Dedup { cols }` adds `cols` as a unique key, so two identical
-        // `Dedup`s merge even though their keyless `Scan`s could not.
-        let dedup = || {
+    fn dedup_shares_only_on_every_column() {
+        let dedup = |cols: Vec<usize>| {
             node(NonASAPOp::Dedup {
-                cols: vec![1],
+                cols,
                 child: scan(),
             })
         };
-        let (ra, rb) = two_roots(dedup(), dedup());
+        let (ra, rb) = two_roots(dedup(vec![]), dedup(vec![]));
         assert!(Rc::ptr_eq(&ra, &rb));
+        let (ra, rb) = two_roots(dedup(vec![1]), dedup(vec![1]));
+        assert!(!Rc::ptr_eq(&ra, &rb));
     }
 
     /// `LIMIT 10` without an order may return any 10 rows, so two copies may
@@ -762,8 +737,33 @@ mod tests {
         assert!(!Rc::ptr_eq(&ra, &rb));
     }
 
+    /// A `LIMIT` over a `Sort` keeps the first rows of a fixed order.
     #[test]
-    fn group_keys_gate_still_prevented_when_partition_by_without_used() {
+    fn ordered_limits_are_shared() {
+        let limit = || {
+            node(NonASAPOp::Limit {
+                n: Some(10),
+                offset: 0,
+                partition_by: GroupKeys::default(),
+                child: node(NonASAPOp::Sort {
+                    keys: vec![crate::ir::scalar::SortKey {
+                        expr: ScalarExpr::column(2),
+                        ascending: false,
+                        nulls_first: false,
+                    }],
+                    partition_by: GroupKeys::default(),
+                    child: scan(),
+                }),
+            })
+        };
+        let (ra, rb) = two_roots(limit(), limit());
+        assert!(Rc::ptr_eq(&ra, &rb));
+    }
+
+    /// `count without (service)` has no provable unique key, but it computes
+    /// the same rows from the same input, so equal copies are one node.
+    #[test]
+    fn without_grouping_is_shared() {
         let without_agg = || {
             node(NonASAPOp::Aggregate {
                 reduction: Reduction::Reduce(GroupKeys::without(vec![0])),
@@ -779,7 +779,7 @@ mod tests {
         let a = without_agg();
         assert!(!a.schema.has_unique_key());
         let (ra, rb) = two_roots(a, without_agg());
-        assert!(!Rc::ptr_eq(&ra, &rb));
+        assert!(Rc::ptr_eq(&ra, &rb));
     }
 
     #[test]
@@ -804,14 +804,20 @@ mod tests {
         let (done, completion) = std::sync::mpsc::channel();
         let worker = std::thread::spawn(move || {
             fn keyed_diamond() -> Rc<OperatorNode> {
-                // BinaryOp over a keyless scan has no unique key at any level,
-                // so none of the 24 levels is shareable; the `Dedup` on top is.
+                // An unordered `Limit` at each of the 24 levels is never
+                // shared, so comparing the `Dedup` on top walks the diamond.
                 let mut current = scan();
                 for _ in 0..24 {
-                    current = compare(Rc::clone(&current), current);
+                    let limited = node(NonASAPOp::Limit {
+                        n: Some(10),
+                        offset: 0,
+                        partition_by: GroupKeys::default(),
+                        child: current,
+                    });
+                    current = compare(Rc::clone(&limited), limited);
                 }
                 node(NonASAPOp::Dedup {
-                    cols: vec![1],
+                    cols: vec![],
                     child: current,
                 })
             }
diff --git a/crates/types/src/ir/operator/non_asap.rs b/crates/types/src/ir/operator/non_asap.rs
index d945ea841..2f534cd6c 100644
--- a/crates/types/src/ir/operator/non_asap.rs
+++ b/crates/types/src/ir/operator/non_asap.rs
@@ -10,7 +10,8 @@ use super::node::{OperatorNode, OperatorResultKind};
 use crate::ir::operator::agg_intent::AggIntent;
 use crate::ir::operator::operator_properties::{
     BinaryOpKind, ConcatDiscriminatorKey, GroupKeys, InfoMatcher, JoinKind, Reduction,
-    RelationalSetOpKind, SampleKind, Source, TimeShift, VectorMatch, WindowFrame, WindowFuncKind,
+    RelationalSetOpKind, SampleKind, Source, TimeShift, VectorMatch, WindowFrame, WindowFrameUnits,
+    WindowFuncKind,
 };
 use crate::ir::scalar::{Predicate, ProjectItem, ScalarExpr, SortKey};
 use crate::ir::schema::aggregate_schema::aggregate_output_schema;
@@ -500,6 +501,64 @@ impl<C> NonASAPOp<C> {
 }
 
 impl NonASAPOp {
+    /// Whether every execution over the same input rows returns the same
+    /// output rows, so two equal nodes over one input can be computed (or
+    /// stored) once and read by both readers.
+    ///
+    /// Ties are the one place the input order decides: a `Sort` keeps tied
+    /// rows in input order (a stable sort), and an exact top-k ranks with that
+    /// sort. That tie order counts as fixed. An operator that picks rows by
+    /// input order alone does not return the same rows:
+    /// - a `Limit` or `OFFSET` whose child is not a `Sort` (`LIMIT 10` may
+    ///   return any 10 rows);
+    /// - a `Dedup` on some but not all columns (which row of each key
+    ///   survives is unspecified);
+    /// - a window function that reads row positions (`ROW_NUMBER`, `LAG`,
+    ///   `FIRST_VALUE`, a `ROWS` frame, …) without an `ORDER BY`.
+    ///
+    /// Every other operator computes its output from the input rows' values
+    /// only. `limitk` and `limit_ratio` pick the same series on every
+    /// evaluation by definition ([`SampleKind`]).
+    pub fn returns_same_rows_for_same_input_rows(&self) -> bool {
+        use NonASAPOp::*;
+        match self {
+            Limit {
+                n, offset, child, ..
+            } => (n.is_none() && *offset == 0) || matches!(child.non_asap(), Some(Sort { .. })),
+            Dedup { cols, child } => {
+                cols.is_empty()
+                    || (0..child.schema.fields.len()).all(|column| cols.contains(&column))
+            }
+            SQLWindowFunc {
+                func,
+                order_by,
+                frame,
+                ..
+            } => {
+                let reads_row_positions = match func {
+                    WindowFuncKind::RowNumber
+                    | WindowFuncKind::Lag
+                    | WindowFuncKind::Lead
+                    | WindowFuncKind::LagInFrame
+                    | WindowFuncKind::LeadInFrame
+                    | WindowFuncKind::FirstValue
+                    | WindowFuncKind::LastValue
+                    | WindowFuncKind::NthValue(_) => true,
+                    WindowFuncKind::Rank | WindowFuncKind::DenseRank => false,
+                    WindowFuncKind::Sum
+                    | WindowFuncKind::Avg
+                    | WindowFuncKind::Count
+                    | WindowFuncKind::Min
+                    | WindowFuncKind::Max => frame
+                        .as_ref()
+                        .is_some_and(|frame| frame.units == WindowFrameUnits::Rows),
+                };
+                !reads_row_positions || !order_by.is_empty()
+            }
+            _ => true,
+        }
+    }
+
     /// Output schema derived from this operator's parameters and its
     /// children's (already derived) schemas.
     pub fn output_schema(&self) -> Result<Schema, SchemaDerivationError> {
diff --git a/docs/design_docs/proposals/planner-layering.md b/docs/design_docs/proposals/planner-layering.md
index 37134791d..84d1026e2 100644
--- a/docs/design_docs/proposals/planner-layering.md
+++ b/docs/design_docs/proposals/planner-layering.md
@@ -286,14 +286,18 @@ lists `LogicalMemo::from_roots` with Pass 1. `LogicalMemo::from_roots` runs
 `share_common_sub_dags` (`crates/types/src/ir/cse.rs`) over the frontends'
 roots and numbers the shared nodes as groups. Two structurally equal nodes
 become one group only when sharing them is legal, that is, when every reader
-of the shared node reads the rows it would read alone:
+of the shared node reads the rows it would read alone. Equal nodes are
+interned children first, so they always read one input, and a node is legal
+to share when every execution over the same input rows returns the same
+output rows (`NonASAPOp::returns_same_rows_for_same_input_rows`). A unique
+key is not needed.
 
 | Equal nodes | Share one group |
 |---|---|
 | ASAP nodes (summary states and their evaluations) | Always (equal operator, schema and guarantee) |
-| `Scan`s | Always: they read the same source rows |
-| Windows directly over a scan, global scalar reductions, exact top-k keeping the series identity | Always: their rows keep a stable identity |
-| Any other non-ASAP node | Only when its output schema has a provable unique key (`Schema::has_unique_key`) |
+| Non-ASAP nodes that compute their output from the input rows' values: scans, windows, filters, projections, aggregates (exact top-k included), joins, binary operators, `limitk` | Always |
+| A `Sort` keeps tied rows in input order (a stable sort); a `Limit` over a `Sort` and an exact top-k rank with that order | Always: the tie order counts as fixed |
+| A `Limit` or offset not over a `Sort`, a `Dedup` on some but not all columns, a window function that reads row positions (`ROW_NUMBER`, `LAG`, `FIRST_VALUE`, a `ROWS` frame) without an `ORDER BY` | Never: they pick rows by input order alone, so two copies may return different rows |
 
 Equal nodes that may not share stay separate groups, also within one query.
 They may still read one shared child group.

From f58d14dd1852150910f0234d06cba6a70e84b803 Mon Sep 17 00:00:00 2001
From: zzylol <50204836+zzylol@users.noreply.github.com>
Date: Wed, 28 Oct 2026 21:23:37 +0000
Subject: [PATCH 3/6] docs: CSE legality by same rows, not unique key, in the
 IR docs and memo example

pre-asap-ir.md and the planner-layering memo example explained sharing by
the unique key; both now state the same-rows rule. Schema::has_unique_key
no longer claims to be the CSE gate.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 crates/types/src/ir/schema/mod.rs              | 4 +---
 docs/design_docs/proposals/planner-layering.md | 9 ++++-----
 docs/develop_docs/pre-asap-ir.md               | 9 +++++----
 3 files changed, 10 insertions(+), 12 deletions(-)

diff --git a/crates/types/src/ir/schema/mod.rs b/crates/types/src/ir/schema/mod.rs
index ce15c3f92..1a34d8ba5 100644
--- a/crates/types/src/ir/schema/mod.rs
+++ b/crates/types/src/ir/schema/mod.rs
@@ -347,9 +347,7 @@ impl Schema {
             .position(|c| c.name == name && c.table.as_deref() == Some(table))
     }
 
-    /// Whether this schema has *any* provable unique key — the signal a
-    /// reuse-aware planning pass needs to decide whether a producer's output
-    /// can be safely shared across consumers.
+    /// Whether this schema has *any* provable unique key.
     pub fn has_unique_key(&self) -> bool {
         !self.unique_keys.is_empty()
     }
diff --git a/docs/design_docs/proposals/planner-layering.md b/docs/design_docs/proposals/planner-layering.md
index 84d1026e2..8a75aaecb 100644
--- a/docs/design_docs/proposals/planner-layering.md
+++ b/docs/design_docs/proposals/planner-layering.md
@@ -360,11 +360,10 @@ avg by (job) (quantile_over_time(0.5, data[5m]) / quantile_over_time(0.5, data[5
 
 G1 is one group for its two readers, because a window directly over a scan
 may be shared. The two `quantile_over_time(0.5, data[5m])` are one group, G2:
-the PromQL frontend adds a series-identity column (`$promql_series_identity`)
-to every series, so the per-series quantile's output has a unique key
-(series identity and timestamp), and equal non-ASAP nodes with a unique key
-share one group. A non-ASAP node without a unique key, such as a
-`without(..)` grouping, would stay one group per reader.
+both compute the same rows from the same input, so one copy serves both
+operands. A unique key is not needed (the PromQL planning path happens to
+give this output one, series identity and timestamp); an unordered `LIMIT`
+would stay one group per reader.
 
 **Step 2, Stage 1 Pass 1.** `try_populate_pass1` gives G2 and G4 seven
 candidates each. G0, G1, G3 and G5 get one each, their own node. G6 gets
diff --git a/docs/develop_docs/pre-asap-ir.md b/docs/develop_docs/pre-asap-ir.md
index 82ed4438b..bfe197a2c 100644
--- a/docs/develop_docs/pre-asap-ir.md
+++ b/docs/develop_docs/pre-asap-ir.md
@@ -87,9 +87,10 @@ state and only appear below an ASAP operator; a scalar expression that reads suc
 typing error (`ScalarExpr::scalar_type`), because state must be read out before a value can use
 it. Column references are positional `ColumnId`s (indexes into the input schema), never names.
 
-`Schema::has_unique_key()` is the legality gate CSE uses: a non-ASAP producer is only shared
-across consumers when its row identity is provable. A `Scan` is shared without one: equal scans
-read the same source rows, a SQL table's keyless rows included.
+CSE shares a non-ASAP producer across consumers when every execution over the same input rows
+returns the same output rows (`NonASAPOp::returns_same_rows_for_same_input_rows`); a unique key
+is not needed. An unordered `Limit`, a `Dedup` on some but not all columns and a window function
+that reads row positions without an `ORDER BY` are never shared.
 
 ## Scalar expressions
 
@@ -154,7 +155,7 @@ The result is `Rc<OperatorNode>`. `lower_promql_workload`, `lower_sql` / `lower_
 Workload search then runs structural CSE (`asap_types::ir::cse::share_common_sub_dags`) once
 across every root: bottom-up hash-consing where the structural hash is only a filter and the
 typed `PartialEq` decides sharing, following scalar references like any other input, and
-gated by `Schema::has_unique_key()` for non-ASAP producers other than scans.
+gated by `NonASAPOp::returns_same_rows_for_same_input_rows` for non-ASAP producers.
 
 ## Fields and column references
 

From 7f292b351252535424c690c52e6bea355dfc70bf Mon Sep 17 00:00:00 2001
From: zzylol <50204836+zzylol@users.noreply.github.com>
Date: Thu, 29 Oct 2026 00:25:31 +0000
Subject: [PATCH 4/6] review: Limit over Sort needs matching partitions; stale
 unique-key comments; pins assert the shared group

- A Limit over a Sort partitioned differently still picks rows by input
  order, so it is not shared.
- The sharing rule's doc states its assumptions: scalar functions the IR
  accepts are deterministic, and Extension intents return the same rows.
- Comments in memo/groups.rs, asap_aware_cse.rs, summary_inputs.rs and
  summary_sharing.rs no longer describe the unique-key gate.
- The top-k snapshot pins also assert that the readers' sum_over_time is
  one group.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../src/asap_aware_cse.rs                     |  3 ++-
 crates/logical-asap-optimization/src/memo.rs  |  2 +-
 .../src/memo/groups.rs                        |  7 ++++---
 .../src/memo/rules/summary_inputs.rs          |  6 +++---
 .../src/memo/shared_producer_snapshot.rs      | 20 +++++++++++++++++++
 crates/planner/tests/summary_sharing.rs       |  4 ++--
 crates/types/src/ir/cse.rs                    | 14 ++++++++-----
 crates/types/src/ir/operator/non_asap.rs      | 20 ++++++++++++++-----
 8 files changed, 56 insertions(+), 20 deletions(-)

diff --git a/crates/logical-asap-optimization/src/asap_aware_cse.rs b/crates/logical-asap-optimization/src/asap_aware_cse.rs
index 022445c0c..063516d65 100644
--- a/crates/logical-asap-optimization/src/asap_aware_cse.rs
+++ b/crates/logical-asap-optimization/src/asap_aware_cse.rs
@@ -12,7 +12,8 @@
 //! A *sharing candidate* ([`SharingCandidate`]) of a whole-workload
 //! candidate is a set of two or more of its nodes with identical input and
 //! computation semantics that are legal to share: the `ir::cse` interning
-//! merges only ASAP nodes, scans, and nodes with a unique key.
+//! merges ASAP nodes and non-ASAP nodes that return the same rows for the
+//! same input rows.
 //! [`identical_expression_pass2_candidates`] returns one Pass 2 candidate
 //! per non-empty subset of the sharing candidates (Algorithm 2, line 31):
 //! each sharing candidate in the subset becomes one node serving every
diff --git a/crates/logical-asap-optimization/src/memo.rs b/crates/logical-asap-optimization/src/memo.rs
index 21d39da43..f87506099 100644
--- a/crates/logical-asap-optimization/src/memo.rs
+++ b/crates/logical-asap-optimization/src/memo.rs
@@ -276,7 +276,7 @@ mod tests {
     }
 
     /// A per-series quantile has no provable unique key, but it is
-    /// deterministic: both readers of `G2 / G3` read the same rows, so the
+    /// deterministic: both operands of `G2 / G2` read the same rows, so the
     /// two equal sub-expressions are one group read twice.
     #[test]
     fn equal_keyless_deterministic_results_share_one_group() {
diff --git a/crates/logical-asap-optimization/src/memo/groups.rs b/crates/logical-asap-optimization/src/memo/groups.rs
index 4234426c0..9641184a7 100644
--- a/crates/logical-asap-optimization/src/memo/groups.rs
+++ b/crates/logical-asap-optimization/src/memo/groups.rs
@@ -4,9 +4,10 @@
 //! Groups are the workload's logical DAG interned by
 //! [`asap_types::ir::cse::share_common_sub_dags`], the identical-expression rule's legality
 //! included: structurally equal nodes are one group when sharing them is
-//! legal (an ASAP node, a scan, or one whose schema has a unique key).
-//! Structurally equal keyless aggregates remain separate; eligible scans
-//! and keyed aggregates can share a group.
+//! legal: an ASAP node, or a non-ASAP node that returns the same rows for
+//! the same input rows (`NonASAPOp::returns_same_rows_for_same_input_rows`).
+//! A unique key is not needed; an unordered `LIMIT` stays one group per
+//! reader.
 //!
 //! A Pass 1 target's aggregate group gets one candidate per combination of
 //! its measures' Pass 1 candidates: the aggregate alone composed by
diff --git a/crates/logical-asap-optimization/src/memo/rules/summary_inputs.rs b/crates/logical-asap-optimization/src/memo/rules/summary_inputs.rs
index 5b275d228..46c20c0b1 100644
--- a/crates/logical-asap-optimization/src/memo/rules/summary_inputs.rs
+++ b/crates/logical-asap-optimization/src/memo/rules/summary_inputs.rs
@@ -80,9 +80,9 @@ fn canonical_preprocessing(
         return Ok(true);
     }
     if boundaries.contains(&Rc::as_ptr(node)) {
-        // Logical TimeRange groups can remain distinct because their rows lack
-        // a unique key. The existing state-sharing rule nevertheless permits
-        // identical range descriptors over the same canonical source.
+        // Logical TimeRange groups can remain distinct when their inputs
+        // differ structurally. The existing state-sharing rule nevertheless
+        // permits identical range descriptors over the same canonical source.
         if let Some(group) = memo
             .groups
             .iter()
diff --git a/crates/logical-asap-optimization/src/memo/shared_producer_snapshot.rs b/crates/logical-asap-optimization/src/memo/shared_producer_snapshot.rs
index 179945fe3..c633ae133 100644
--- a/crates/logical-asap-optimization/src/memo/shared_producer_snapshot.rs
+++ b/crates/logical-asap-optimization/src/memo/shared_producer_snapshot.rs
@@ -58,6 +58,23 @@ fn expanded(queries: &[(&str, f64)], step_ms: u32) -> CandidateMemo {
         .memo
 }
 
+/// Whether one `sum_over_time` group (an aggregate directly over a window
+/// of `metric`) is reached by every query in `queries`.
+fn one_sum_over_time_group_for(memo: &CandidateMemo, metric: &str, queries: &[usize]) -> bool {
+    use asap_types::ir::operator::Source;
+    use asap_types::ir::NonASAPOp;
+    memo.groups.iter().any(|group| {
+        let Some(NonASAPOp::Aggregate { child, .. }) = group.node.non_asap() else {
+            return false;
+        };
+        let Some(NonASAPOp::TimeRange { child: scan, .. }) = child.non_asap() else {
+            return false;
+        };
+        matches!(scan.non_asap(), Some(NonASAPOp::Scan { source: Source::TimeSeries { metric: m }, .. }) if m == metric)
+            && queries.iter().all(|query| group.queries.contains(query))
+    })
+}
+
 fn signature(node: &Rc<asap_types::ir::OperatorNode>) -> String {
     serde_json::to_string(&flatten(&[QueryRoot::Operator(Rc::clone(node))]).0).unwrap()
 }
@@ -115,6 +132,7 @@ fn top_k_reuse_candidates_are_unchanged() {
         .iter()
         .flatten()
         .any(|candidate| candidate.origin == CandidateOrigin::TopKReuse));
+    assert!(one_sum_over_time_group_for(&memo, "m_gauge", &[0, 1]));
     assert_eq!(fingerprint(&memo), (69, 14825718632741491576));
 }
 
@@ -164,5 +182,7 @@ fn several_top_k_sets_are_unchanged() {
         ],
         60_000,
     );
+    assert!(one_sum_over_time_group_for(&memo, "m_gauge", &[0, 1, 2]));
+    assert!(one_sum_over_time_group_for(&memo, "other", &[3, 4]));
     assert_eq!(fingerprint(&memo), (504, 3265274154981812875));
 }
diff --git a/crates/planner/tests/summary_sharing.rs b/crates/planner/tests/summary_sharing.rs
index c08c57273..98a015f4d 100644
--- a/crates/planner/tests/summary_sharing.rs
+++ b/crates/planner/tests/summary_sharing.rs
@@ -497,8 +497,8 @@ async fn cross_series_p50_and_p99_share_one_producer() {
     assert_eq!(unique_deployments(&output), 1);
 }
 
-/// An ungrouped aggregate has no unique key, so pre-ASAP CSE keeps the two
-/// copies apart; their identical producers (rate, then sum) are shared here.
+/// Two identical ungrouped queries: their identical producers (rate, then
+/// sum) are shared.
 #[tokio::test]
 async fn identical_ungrouped_queries_share_their_producers() {
     let query = ("sum(rate(x[5m]))", 0.01);
diff --git a/crates/types/src/ir/cse.rs b/crates/types/src/ir/cse.rs
index bd819c93c..1873b8452 100644
--- a/crates/types/src/ir/cse.rs
+++ b/crates/types/src/ir/cse.rs
@@ -737,10 +737,12 @@ mod tests {
         assert!(!Rc::ptr_eq(&ra, &rb));
     }
 
-    /// A `LIMIT` over a `Sort` keeps the first rows of a fixed order.
+    /// A `LIMIT` over a `Sort` with the same partitions keeps the first rows
+    /// of a fixed order; over a `Sort` partitioned differently, which rows
+    /// come first still depends on input order.
     #[test]
-    fn ordered_limits_are_shared() {
-        let limit = || {
+    fn limits_over_a_sort_with_their_partitions_are_shared() {
+        let limit = |sort_partitions: GroupKeys| {
             node(NonASAPOp::Limit {
                 n: Some(10),
                 offset: 0,
@@ -751,13 +753,15 @@ mod tests {
                         ascending: false,
                         nulls_first: false,
                     }],
-                    partition_by: GroupKeys::default(),
+                    partition_by: sort_partitions,
                     child: scan(),
                 }),
             })
         };
-        let (ra, rb) = two_roots(limit(), limit());
+        let (ra, rb) = two_roots(limit(GroupKeys::default()), limit(GroupKeys::default()));
         assert!(Rc::ptr_eq(&ra, &rb));
+        let (ra, rb) = two_roots(limit(GroupKeys::by(vec![1])), limit(GroupKeys::by(vec![1])));
+        assert!(!Rc::ptr_eq(&ra, &rb));
     }
 
     /// `count without (service)` has no provable unique key, but it computes
diff --git a/crates/types/src/ir/operator/non_asap.rs b/crates/types/src/ir/operator/non_asap.rs
index 2f534cd6c..957bb438b 100644
--- a/crates/types/src/ir/operator/non_asap.rs
+++ b/crates/types/src/ir/operator/non_asap.rs
@@ -509,8 +509,8 @@ impl NonASAPOp {
     /// rows in input order (a stable sort), and an exact top-k ranks with that
     /// sort. That tie order counts as fixed. An operator that picks rows by
     /// input order alone does not return the same rows:
-    /// - a `Limit` or `OFFSET` whose child is not a `Sort` (`LIMIT 10` may
-    ///   return any 10 rows);
+    /// - a `Limit` or `OFFSET` whose child is not a `Sort` with the same
+    ///   partitions (`LIMIT 10` may return any 10 rows);
     /// - a `Dedup` on some but not all columns (which row of each key
     ///   survives is unspecified);
     /// - a window function that reads row positions (`ROW_NUMBER`, `LAG`,
@@ -518,13 +518,23 @@ impl NonASAPOp {
     ///
     /// Every other operator computes its output from the input rows' values
     /// only. `limitk` and `limit_ratio` pick the same series on every
-    /// evaluation by definition ([`SampleKind`]).
+    /// evaluation by definition ([`SampleKind`]). This assumes every scalar
+    /// function the IR accepts is deterministic (no `random()`; unregistered
+    /// SQL functions are rejected) and that a deployment's
+    /// `AggIntent::Extension` returns the same rows for the same input.
     pub fn returns_same_rows_for_same_input_rows(&self) -> bool {
         use NonASAPOp::*;
         match self {
             Limit {
-                n, offset, child, ..
-            } => (n.is_none() && *offset == 0) || matches!(child.non_asap(), Some(Sort { .. })),
+                n,
+                offset,
+                partition_by,
+                child,
+            } => {
+                (n.is_none() && *offset == 0)
+                    || matches!(child.non_asap(), Some(Sort { partition_by: sorted_by, .. })
+                        if sorted_by == partition_by)
+            }
             Dedup { cols, child } => {
                 cols.is_empty()
                     || (0..child.schema.fields.len()).all(|column| cols.contains(&column))

From 18218ad06bbdfa495f91f704f0023850923c5859 Mon Sep 17 00:00:00 2001
From: zzylol <50204836+zzylol@users.noreply.github.com>
Date: Thu, 29 Oct 2026 00:30:38 +0000
Subject: [PATCH 5/6] refactor(cse): name the sharing rule
 is_shareable_when_equal and list every operator

- NonASAPOp::returns_same_rows_for_same_input_rows -> is_shareable_when_equal,
  which says what CSE asks; the same-rows definition stays in its doc.
- The match lists every NonASAPOp variant, so a new operator must decide
  whether it may be shared.
- Dedup on every column is a length check over its distinct columns.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .../src/memo/groups.rs                        |  2 +-
 crates/types/src/ir/cse.rs                    |  4 +--
 crates/types/src/ir/operator/non_asap.rs      | 29 +++++++++++++++----
 .../design_docs/proposals/planner-layering.md |  2 +-
 docs/develop_docs/pre-asap-ir.md              |  4 +--
 5 files changed, 30 insertions(+), 11 deletions(-)

diff --git a/crates/logical-asap-optimization/src/memo/groups.rs b/crates/logical-asap-optimization/src/memo/groups.rs
index 9641184a7..a0c19947f 100644
--- a/crates/logical-asap-optimization/src/memo/groups.rs
+++ b/crates/logical-asap-optimization/src/memo/groups.rs
@@ -5,7 +5,7 @@
 //! [`asap_types::ir::cse::share_common_sub_dags`], the identical-expression rule's legality
 //! included: structurally equal nodes are one group when sharing them is
 //! legal: an ASAP node, or a non-ASAP node that returns the same rows for
-//! the same input rows (`NonASAPOp::returns_same_rows_for_same_input_rows`).
+//! the same input rows (`NonASAPOp::is_shareable_when_equal`).
 //! A unique key is not needed; an unordered `LIMIT` stays one group per
 //! reader.
 //!
diff --git a/crates/types/src/ir/cse.rs b/crates/types/src/ir/cse.rs
index 1873b8452..66e716ef4 100644
--- a/crates/types/src/ir/cse.rs
+++ b/crates/types/src/ir/cse.rs
@@ -37,7 +37,7 @@
 //! Structural equality is necessary but not sufficient. A non-ASAP node is
 //! only ever *returned* as a match for another when every execution over the
 //! same input rows returns the same output rows
-//! ([`NonASAPOp::returns_same_rows_for_same_input_rows`]). Equal nodes are
+//! ([`NonASAPOp::is_shareable_when_equal`]). Equal nodes are
 //! interned children first, so two equal nodes read one input, and one copy
 //! returns each reader the rows it would have computed alone. A unique key
 //! is not needed: equal per-series quantiles without a provable key are one
@@ -215,7 +215,7 @@ impl InternTable {
         let reusable = node.is_asap()
             || node
                 .non_asap()
-                .is_some_and(NonASAPOp::returns_same_rows_for_same_input_rows);
+                .is_some_and(NonASAPOp::is_shareable_when_equal);
         let bucket = self.buckets.entry(hash).or_default();
         if reusable {
             if let Some(existing) = bucket
diff --git a/crates/types/src/ir/operator/non_asap.rs b/crates/types/src/ir/operator/non_asap.rs
index 957bb438b..8d3b0cc6b 100644
--- a/crates/types/src/ir/operator/non_asap.rs
+++ b/crates/types/src/ir/operator/non_asap.rs
@@ -501,8 +501,9 @@ impl<C> NonASAPOp<C> {
 }
 
 impl NonASAPOp {
-    /// Whether every execution over the same input rows returns the same
-    /// output rows, so two equal nodes over one input can be computed (or
+    /// Whether two structurally equal copies of this node may be one shared
+    /// node: every execution over the same input rows returns the same
+    /// output rows, so one copy over their one input can be computed (or
     /// stored) once and read by both readers.
     ///
     /// Ties are the one place the input order decides: a `Sort` keeps tied
@@ -522,7 +523,7 @@ impl NonASAPOp {
     /// function the IR accepts is deterministic (no `random()`; unregistered
     /// SQL functions are rejected) and that a deployment's
     /// `AggIntent::Extension` returns the same rows for the same input.
-    pub fn returns_same_rows_for_same_input_rows(&self) -> bool {
+    pub fn is_shareable_when_equal(&self) -> bool {
         use NonASAPOp::*;
         match self {
             Limit {
@@ -535,9 +536,11 @@ impl NonASAPOp {
                     || matches!(child.non_asap(), Some(Sort { partition_by: sorted_by, .. })
                         if sorted_by == partition_by)
             }
+            // Empty `cols` means every column.
             Dedup { cols, child } => {
                 cols.is_empty()
-                    || (0..child.schema.fields.len()).all(|column| cols.contains(&column))
+                    || cols.iter().collect::<std::collections::BTreeSet<_>>().len()
+                        == child.schema.fields.len()
             }
             SQLWindowFunc {
                 func,
@@ -565,7 +568,23 @@ impl NonASAPOp {
                 };
                 !reads_row_positions || !order_by.is_empty()
             }
-            _ => true,
+            Scan { .. }
+            | Values { .. }
+            | Filter { .. }
+            | Project { .. }
+            | Aggregate { .. }
+            | Join { .. }
+            | SetOp { .. }
+            | Concat { .. }
+            | Sort { .. }
+            | BinaryOp { .. }
+            | TimeRange { .. }
+            | TimeShift { .. }
+            | PromqlVectorFromScalar(_)
+            | PromqlRelabel { .. }
+            | PromqlInfoEnrich { .. }
+            | PromqlSeriesSample { .. }
+            | PromqlSubquery { .. } => true,
         }
     }
 
diff --git a/docs/design_docs/proposals/planner-layering.md b/docs/design_docs/proposals/planner-layering.md
index 8a75aaecb..fd5acbe63 100644
--- a/docs/design_docs/proposals/planner-layering.md
+++ b/docs/design_docs/proposals/planner-layering.md
@@ -289,7 +289,7 @@ become one group only when sharing them is legal, that is, when every reader
 of the shared node reads the rows it would read alone. Equal nodes are
 interned children first, so they always read one input, and a node is legal
 to share when every execution over the same input rows returns the same
-output rows (`NonASAPOp::returns_same_rows_for_same_input_rows`). A unique
+output rows (`NonASAPOp::is_shareable_when_equal`). A unique
 key is not needed.
 
 | Equal nodes | Share one group |
diff --git a/docs/develop_docs/pre-asap-ir.md b/docs/develop_docs/pre-asap-ir.md
index bfe197a2c..d0360cf43 100644
--- a/docs/develop_docs/pre-asap-ir.md
+++ b/docs/develop_docs/pre-asap-ir.md
@@ -88,7 +88,7 @@ typing error (`ScalarExpr::scalar_type`), because state must be read out before
 it. Column references are positional `ColumnId`s (indexes into the input schema), never names.
 
 CSE shares a non-ASAP producer across consumers when every execution over the same input rows
-returns the same output rows (`NonASAPOp::returns_same_rows_for_same_input_rows`); a unique key
+returns the same output rows (`NonASAPOp::is_shareable_when_equal`); a unique key
 is not needed. An unordered `Limit`, a `Dedup` on some but not all columns and a window function
 that reads row positions without an `ORDER BY` are never shared.
 
@@ -155,7 +155,7 @@ The result is `Rc<OperatorNode>`. `lower_promql_workload`, `lower_sql` / `lower_
 Workload search then runs structural CSE (`asap_types::ir::cse::share_common_sub_dags`) once
 across every root: bottom-up hash-consing where the structural hash is only a filter and the
 typed `PartialEq` decides sharing, following scalar references like any other input, and
-gated by `NonASAPOp::returns_same_rows_for_same_input_rows` for non-ASAP producers.
+gated by `NonASAPOp::is_shareable_when_equal` for non-ASAP producers.
 
 ## Fields and column references
 

From d9821d2720beac988673725ed25bd2dc1f3e9157 Mon Sep 17 00:00:00 2001
From: zzylol <50204836+zzylol@users.noreply.github.com>
Date: Thu, 29 Oct 2026 00:53:58 +0000
Subject: [PATCH 6/6] test(devtools): keyless ratio-of-quantiles memo has 25
 choices; only the exact DAG is valid

Without the series identity the two quantile 0.5 are now one group, so the
memo has 5 x 5 = 25 choices (9 whole-budget), not 343. Its exact DAG prices
at 0.0032169/s against 0.0037502/s for the same exact DAG with the identity
column, so its price is no longer compared with the identity-lowered
selection (#1035); the production-lowering test keeps that check.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 crates/devtools/tests/memo_pricing.rs | 43 +++++++--------------------
 1 file changed, 11 insertions(+), 32 deletions(-)

diff --git a/crates/devtools/tests/memo_pricing.rs b/crates/devtools/tests/memo_pricing.rs
index 902cf3496..784d45eec 100644
--- a/crates/devtools/tests/memo_pricing.rs
+++ b/crates/devtools/tests/memo_pricing.rs
@@ -67,7 +67,7 @@ struct Planned {
 /// `stage_pipeline` lowers each series' identity as a column
 /// (`lower_promql_roots`); without it (`series_identity` false, as the
 /// memo-enumeration tests lower), the ratio's two `quantile 0.5` have no
-/// provable unique key and stay separate groups.
+/// provable unique key, but they are still one group.
 fn planned(workload: &PlanningWorkload, series_identity: bool) -> Planned {
     let roots = if series_identity {
         lower_promql_roots(workload).unwrap()
@@ -147,21 +147,6 @@ fn price_every_choice(
     )
 }
 
-fn assert_none_cheaper(priced: &[PricedChoice], baseline: SelectedBaseline) {
-    let cheaper: Vec<_> = priced
-        .iter()
-        .enumerate()
-        .filter(|(_, p)| p.cheaper_than_selected)
-        .map(|(i, p)| (format!("E{}", i + 1), p.cost_per_second))
-        .collect();
-    assert!(
-        cheaper.is_empty(),
-        "selected {:?}; cheaper: {cheaper:?}",
-        baseline.cost_per_second
-    );
-    assert!(priced.iter().any(|p| p.valid));
-}
-
 /// `stage_pipeline --price-enumerated 200` on `args`' workload: the listed
 /// choices of the enumeration's step, each priced.
 fn priced_document(args: &[&str]) -> Value {
@@ -242,15 +227,16 @@ fn ratio_of_quantiles_no_listed_choice_is_cheaper_than_the_selected_plan() {
     }
 }
 
-/// Lowered without series identity, the two `quantile 0.5` stay separate
-/// groups: 7 × 7 × 7 = 343 DAGs, 27 of them sized for the whole budget. The
-/// executor compiles none of their per-series summaries without the series
-/// identity, so only the exact DAG is valid, and it is not cheaper than the
-/// plan selected for the same query as `stage_pipeline` lowers it. (The
-/// default search finds no plan for this lowering in its budget, so the
-/// selection comes from the series-identity lowering.)
+/// Lowered without series identity, the two `quantile 0.5` have no unique
+/// key but return the same rows, so they are still one group: 5 × 5 = 25
+/// DAGs (two distinct approximate inputs), 9 of them sized for the whole
+/// budget. The executor compiles none of their per-series summaries without
+/// the series identity, so only the exact DAG is valid. Its price is not
+/// compared with the plan selected for the series-identity lowering: its
+/// rows carry no identity column, so it prices lower than the same exact
+/// DAG with the identity (#1035).
 #[test]
-fn ratio_of_quantiles_343_pass1_choices_none_cheaper_than_the_selected_plan() {
+fn ratio_of_quantiles_25_pass1_choices_without_identity_only_exact_is_valid() {
     let without_identity = planned(&ratio_of_quantiles(), false);
     let enumeration = enumerate(&without_identity.memo, 100_000);
     let whole_budget = enumeration
@@ -262,21 +248,14 @@ fn ratio_of_quantiles_343_pass1_choices_none_cheaper_than_the_selected_plan() {
             })
         })
         .count();
-    assert_eq!((enumeration.choices.len(), whole_budget), (343, 27));
+    assert_eq!((enumeration.choices.len(), whole_budget), (25, 9));
     let with_identity = planned(&ratio_of_quantiles(), true);
     let deployment = deployment();
     for enforce in [false, true] {
         let settings = settings(enforce);
         let models = models(&settings, &deployment);
         let baseline = selected(&with_identity, models);
-        let started = std::time::Instant::now();
         let priced = price_every_choice(&without_identity, models, baseline);
-        eprintln!(
-            "priced {} choices in {:.2?}",
-            priced.len(),
-            started.elapsed()
-        );
-        assert_none_cheaper(&priced, baseline);
         assert_eq!(priced.iter().filter(|p| p.valid).count(), 1);
     }
 }