fix(cse): share equal nodes that return the same rows for the same input, not only keyed ones

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);
     }
 }

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

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 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() {
+        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() {

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::is_shareable_when_equal`).
+//! 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
@@ -848,9 +849,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 +876,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!(

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()

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;
@@ -55,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()
 }
@@ -112,7 +132,8 @@ fn top_k_reuse_candidates_are_unchanged() {
         .iter()
         .flatten()
         .any(|candidate| candidate.origin == CandidateOrigin::TopKReuse));
-    assert_eq!(fingerprint(&memo), (313, 16940643818053118362));
+    assert!(one_sum_over_time_group_for(&memo, "m_gauge", &[0, 1]));
+    assert_eq!(fingerprint(&memo), (69, 14825718632741491576));
 }
 
 /// Quantile readers of one input at different sizes.
@@ -161,5 +182,7 @@ fn several_top_k_sets_are_unchanged() {
         ],
         60_000,
     );
-    assert_eq!(fingerprint(&memo), (994, 9213568444657927922));
+    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));
 }

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);