feat: declare non-negative gauges so per-series quantile sketches compose through avg (#772)

crates/frontend-promql/src/promql.rs

@@ -805,7 +805,10 @@ fn build_over_sub_dag(outer: Outer, keys: Vec<ColumnRef>, child: Unresolved) ->
             );
             let direct_counter_ranking = matches!(&child, Unresolved::Aggregate {
                 measures, reduction: Reduction::PerEntity, ..
-            } if matches!(measures.as_slice(), [AggIntent::Rate | AggIntent::Increase]));
+            } if matches!(
+                measures.as_slice(),
+                [AggIntent::Rate | AggIntent::Increase | AggIntent::IRate]
+            ));
             if descending && (weighted_counter_ranking || direct_counter_ranking) {
                 return Ok(outer_aggregate(
                     keys,
@@ -1708,7 +1711,12 @@ fn build(inner: Inner, keys: Vec<ColumnRef>, outer: Outer) -> Result<Unresolved>
             // Preserve the counter-value ranking intent. Physical candidates
             // may rebuild a heap over finalized rates or use exact Sort/Limit;
             // neither is allowed to sum raw counter samples as ranking weights.
-            if descending && matches!(inner.func, Some(InnerFunc::Rate | InnerFunc::Increase)) {
+            if descending
+                && matches!(
+                    inner.func,
+                    Some(InnerFunc::Rate | InnerFunc::Increase | InnerFunc::IRate)
+                )
+            {
                 let intent = inner_intent(inner.func.as_ref().expect("counter function"));
                 let ranked = windowed_aggregate(inner, vec![], intent)?;
                 return Ok(Unresolved::Aggregate {

crates/frontend-promql/tests/promql_conformance.rs

@@ -832,6 +832,23 @@ fn topk_over_nested_sum_preserves_weighted_topk_accuracy() {
     assert!(has(&qe, |i| matches!(i, AggIntent::TopK { .. })));
 }
 
+#[test]
+fn topk_over_a_per_series_counter_derivative_is_a_heavy_hitter_ranking() {
+    // `topk(k, rate|increase|irate(m[w]))` ranks each series' derivative: a
+    // heavy-hitter TopK intent directly over the per-series derivative, for
+    // `irate` as for `rate` and `increase`.
+    for function in ["rate", "increase", "irate"] {
+        let qe = ok(&format!("topk(3, {function}(node_cpu_seconds_total[5m]))"));
+        let NonASAPOp::Aggregate { measures, .. } = qe.expect_non_asap() else {
+            panic!("{function}: expected a TopK aggregate, got {qe:?}");
+        };
+        assert!(
+            matches!(measures.as_slice(), [AggIntent::TopK { k: 3, .. }]),
+            "{function}: {measures:?}"
+        );
+    }
+}
+
 #[test]
 fn outer_aggregate_over_nested_aggregate_nests() {
     // `max(sum by (job) (rate(m[5m])))` — an outer cross-series reduction over a

crates/logical-asap-optimization/src/per_computation_candidates.rs

@@ -174,6 +174,10 @@ pub enum PruneReason {
     /// answer the measure's readout from it
     /// ([`DeploymentCapabilities::supports`]).
     UnsupportedByDeployment,
+    /// The quantile's φ is outside [0, 1], where PromQL answers -Inf or
+    /// +Inf: no sketch estimates that, and the executor's sketches refuse
+    /// such a φ, so only the exact answer remains.
+    QuantileOutsideUnitInterval,
 }
 
 /// Unranked candidates of one measure and the candidates pruned for it.
@@ -481,7 +485,14 @@ pub fn computation_candidates_for_intent(
             }
         }
         let summaries = summary_candidates(intent).iter().cloned();
-        if target.is_exact() {
+        if matches!(intent, AggIntent::Quantile { q, .. } if !(0.0..=1.0).contains(q)) {
+            choices.pruned.extend(summaries.map(|algorithm| {
+                (
+                    ComputationCandidate::Summary(algorithm),
+                    PruneReason::QuantileOutsideUnitInterval,
+                )
+            }));
+        } else if target.is_exact() {
             choices.pruned.extend(summaries.map(|algorithm| {
                 (
                     ComputationCandidate::Summary(algorithm),
@@ -940,9 +951,10 @@ pub(crate) fn partitions_over(
 }
 
 /// The metric whose raw samples, or plain sums of them, are every value
-/// `node` outputs, or `None`. Only a time-series scan, a time range over it
-/// and a plain `sum` keep that; any other operator (rate and arithmetic
-/// included) ends it. The metric's type is not read here: whether its
+/// `node` outputs, or `None`. Only a time-series scan, a time range over it,
+/// an `offset`-only time shift (it moves the window, not the samples) and a
+/// plain `sum` keep that; any other operator (rate and arithmetic included)
+/// ends it. The metric's type is not read here: whether its
 /// samples are non-negative is plan selection's question
 /// ([`WeightDomain::SamplesOfMetric`]).
 pub(crate) fn samples_of_metric(node: &OperatorNode) -> Option<String> {
@@ -952,6 +964,7 @@ pub(crate) fn samples_of_metric(node: &OperatorNode) -> Option<String> {
             ..
         } => Some(metric.clone()),
         NonASAPOp::TimeRange { child, .. } => samples_of_metric(child),
+        NonASAPOp::TimeShift { shift, child } if shift.at.is_none() => samples_of_metric(child),
         NonASAPOp::Aggregate {
             measures, child, ..
         } if matches!(measures.as_slice(), [AggIntent::Sum { col: None }]) => {
@@ -1660,8 +1673,10 @@ fn fold_stats(reduction: &Reduction, input: &Rc<OperatorNode>) -> Option<Propaga
 /// values are or estimate, or `None`: [`samples_of_metric`]'s strict rule
 /// over a composed candidate. An exact sum accumulator, a renaming and a
 /// heap estimate whose update weights carry the
-/// [`WeightDomain::SamplesOfMetric`] requirement keep it; anything else
-/// (rate, arithmetic, other functions) ends it.
+/// [`WeightDomain::SamplesOfMetric`] requirement keep it, and so does a
+/// quantile estimate of those values: it is one of them, or between two of
+/// them, so it shares their sign. Anything else (rate, arithmetic, other
+/// functions) ends it. Only the sign of the values is read from it.
 fn estimates_samples_of(node: &OperatorNode) -> Option<String> {
     match &node.operator {
         Operator::ASAP(ASAPOp::FinalizeExactAccumulator { child }) => match child.asap()? {
@@ -1673,6 +1688,25 @@ fn estimates_samples_of(node: &OperatorNode) -> Option<String> {
             } => estimates_samples_of(child),
             _ => None,
         },
+        // Only a φ in [0, 1] picks one of the values; outside it the answer
+        // is -Inf or +Inf, whatever their sign.
+        Operator::ASAP(ASAPOp::SummaryEstimate {
+            summary_input,
+            query: SketchStatistic::Quantile { q },
+        }) if (0.0..=1.0).contains(q) => match summary_input.asap()? {
+            ASAPOp::SummaryAgg {
+                input:
+                    SummaryUpdate {
+                        item: None,
+                        weight: SummaryInputExpr::Column(ColumnRef::SampleValue),
+                        ..
+                    },
+                filter: None,
+                child,
+                ..
+            } => estimates_samples_of(child),
+            _ => None,
+        },
         Operator::ASAP(ASAPOp::SummaryEstimate { summary_input, .. }) => {
             match summary_input.asap()? {
                 ASAPOp::SummaryAgg {
@@ -4353,6 +4387,44 @@ mod tests {
         assert_eq!(inventory.combinations(), enumerated);
     }
 
+    /// A quantile estimate of a metric's samples carries their sign only for
+    /// φ in [0, 1]: outside it PromQL answers -Inf or +Inf. Pass 1 offers
+    /// no sketch for such a φ.
+    #[test]
+    fn a_quantile_estimate_shares_its_samples_sign_only_inside_the_unit_interval() {
+        let estimate_of = |query: &str| {
+            let inventory = inventory(query);
+            (0..inventory.targets.len())
+                .flat_map(|t| composed_with(&inventory, t))
+                .find_map(|(_, root)| {
+                    OperatorNode::reachable(&root)
+                        .into_iter()
+                        .find(|n| matches!(n.asap(), Some(ASAPOp::SummaryEstimate { .. })))
+                })
+        };
+        let inside = estimate_of("quantile_over_time(0.5, m[1m])").expect("a sketch estimate");
+        assert_eq!(estimates_samples_of(&inside).as_deref(), Some("m"));
+        let Some(ASAPOp::SummaryEstimate { summary_input, .. }) = inside.asap() else {
+            unreachable!()
+        };
+        for q in [-1.0, 2.0] {
+            let outside = OperatorNode::with_schema(
+                Operator::ASAP(ASAPOp::SummaryEstimate {
+                    summary_input: Rc::clone(summary_input),
+                    query: SketchStatistic::Quantile { q },
+                }),
+                inside.schema.clone(),
+            );
+            assert_eq!(estimates_samples_of(&outside), None, "q = {q}");
+        }
+        for query in [
+            "quantile_over_time(-1, m[1m])",
+            "quantile_over_time(2, m[1m])",
+        ] {
+            assert!(estimate_of(query).is_none(), "{query}: a sketch is offered");
+        }
+    }
+
     /// The metric whose raw samples, or plain sums of them, a composed node's
     /// values estimate: a whole-expression heap's estimates and an exact
     /// sum over the samples are; a rate or any other function is not. A

crates/plan-selection/src/lib.rs

@@ -1723,6 +1723,17 @@ pub fn confirmed_weights(domain: &WeightDomain, data: &DataWorkload) -> WeightDo
     domain.confirmed(|metric| declared_counter(data, metric))
 }
 
+/// Whether `data` declares every sample of `metric` non-negative: the
+/// metric is declared a counter, or a metric (a gauge, say) with
+/// [`SourceFacts::values_non_negative`](asap_types::workload::SourceFacts::values_non_negative).
+fn declared_non_negative(data: &DataWorkload, metric: &str) -> bool {
+    declared_counter(data, metric)
+        || data
+            .sources
+            .get(metric)
+            .is_some_and(|facts| facts.values_non_negative)
+}
+
 /// Whether `data` declares `metric` a counter. A declared type does not
 /// expire, so its evidence is not checked.
 fn declared_counter(data: &DataWorkload, metric: &str) -> bool {
@@ -1844,9 +1855,10 @@ fn bind_statistics(
 /// metric's samples, from its declared type.
 fn workload_statistic(statistic: &str, series: Option<u64>, data: &DataWorkload) -> Option<f64> {
     // The sign of one metric's samples: proven (1) when the metric is
-    // declared a counter; no type is inferred from its name.
+    // declared a counter or declared non-negative; nothing is inferred from
+    // its name.
     if let Some(metric) = parse_samples_sign_statistic(statistic) {
-        return declared_counter(data, metric).then_some(1.0);
+        return declared_non_negative(data, metric).then_some(1.0);
     }
     if statistic == INPUT_ROW_COUNT {
         return series.map(|rows| rows as f64);
@@ -3863,9 +3875,14 @@ mod tests {
         let target = AccuracyTarget::max_error(0.01)
             .failure_probability(0.001)
             .accuracy_metric(AccuracyMetric::ABSOLUTE_STREAM_L1);
-        for function in ["increase", "rate", "irate"] {
-            let query = format!("topk(10, sum by (job) ({function}(data_total[5m])))");
-            let candidates = all_candidates(&query, &target);
+        // An `offset` moves the window, not the samples' sign.
+        for query in [
+            "topk(10, sum by (job) (increase(data_total[5m])))",
+            "topk(10, sum by (job) (rate(data_total[5m])))",
+            "topk(10, sum by (job) (irate(data_total[5m])))",
+            "topk(10, sum by (job) (rate(data_total[5m] offset 1h)))",
+        ] {
+            let candidates = all_candidates(query, &target);
             let count_min: Vec<&str> = candidates
                 .iter()
                 .filter(|c| count_min_builds(c).contains(&SketchAlgorithm::CmsWithHeap))
@@ -5218,6 +5235,133 @@ mod tests {
         }
     }
 
+    /// The candidates of `query` that build a DDSketch.
+    fn ddsketch_candidates(candidates: &[PhysicalCandidate]) -> Vec<String> {
+        candidates
+            .iter()
+            .filter(|c| {
+                c.dag.nodes.iter().any(|node| {
+                    matches!(&node.payload,
+                        Payload::ASAP(ASAPOp::SummaryAgg { family: FieldDataType::Sketch(kind, _), .. })
+                            if kind.algorithm() == &SketchAlgorithm::DDSketch)
+                })
+            })
+            .map(|c| c.id.clone())
+            .collect()
+    }
+
+    /// A per-series DDSketch quantile is within a relative error, which
+    /// carries through `avg` and `sum` only over non-negative values, and
+    /// through `max` whatever their sign. With `x` declared a gauge whose
+    /// samples are non-negative, the DDSketch candidate of
+    /// `avg by (region) (quantile_over_time(0.99, x[100s]))` (and of its `sum`)
+    /// is valid and accepted with a relative-error guarantee; without the
+    /// declaration it is rejected. `max` is valid either way (issue #772).
+    #[test]
+    fn a_non_negative_gauge_carries_a_ddsketch_relative_bound_through_avg() {
+        let target = AccuracyTarget::max_error(0.01)
+            .failure_probability(0.01)
+            .accuracy_metric(AccuracyMetric::Relative);
+        // An `offset` moves the window, not the samples' sign.
+        for (outer, window, needs_sign) in [
+            ("avg", "x[100s]", true),
+            ("sum", "x[100s]", true),
+            ("max", "x[100s]", false),
+            ("avg", "x[100s] offset 1h", true),
+        ] {
+            let query = format!("{outer} by (region) (quantile_over_time(0.99, {window}))");
+            let candidates = all_candidates(&query, &target);
+            let ddsketch = ddsketch_candidates(&candidates);
+            assert!(!ddsketch.is_empty(), "{query}");
+            for values_non_negative in [true, false] {
+                let mut data = with_metric_types(data(), &[("x", MetricType::Gauge)]);
+                data.sources.get_mut("x").unwrap().values_non_negative = values_non_negative;
+                let selection = select_plan(
+                    &candidates,
+                    &[entry_with(target.clone(), repeating(10_000))],
+                    &data,
+                    PlanningModels::builtin(),
+                )
+                .unwrap();
+                for id in &ddsketch {
+                    let accepted = selection.accepted.get(id);
+                    if values_non_negative || !needs_sign {
+                        let [accepted] = accepted.map(Vec::as_slice).unwrap_or_default() else {
+                            panic!("{query}, {id} not valid: {:?}", selection.rejected);
+                        };
+                        assert_eq!(accepted.accepted_metric, Some(AccuracyMetric::Relative));
+                        assert!(!accepted.accepted_regardless_of_guarantee);
+                    } else {
+                        assert!(accepted.is_none(), "{query}, {id}: {accepted:?}");
+                        let rejection = selection.rejected.iter().find(|r| &r.id == id).unwrap();
+                        assert!(rejection.cause.failed_a_check(), "{rejection:?}");
+                    }
+                }
+            }
+        }
+    }
+
+    /// PromQL answers a quantile with φ outside [0, 1] as -Inf or +Inf,
+    /// which no sketch estimates and the executor's sketches refuse: such a
+    /// quantile gets no sketch candidate, so the exact plan is selected, and
+    /// a sketch never inherits its samples' sign for it, even over a gauge
+    /// declared non-negative.
+    #[test]
+    fn a_quantile_outside_the_unit_interval_is_answered_exactly() {
+        let target = AccuracyTarget::max_error(0.01)
+            .failure_probability(0.01)
+            .accuracy_metric(AccuracyMetric::Relative);
+        for query in [
+            "avg by (region) (quantile_over_time(-1, x[100s]))",
+            "avg by (region) (quantile_over_time(2, x[100s]))",
+            "quantile_over_time(2, x[100s])",
+            "max by (region) (quantile_over_time(-1, x[100s]))",
+        ] {
+            let candidates = all_candidates(query, &target);
+            assert!(
+                candidates.iter().all(|c| !has_summary_estimate(c)),
+                "{query}: a sketch candidate is offered"
+            );
+            let mut data = with_metric_types(data(), &[("x", MetricType::Gauge)]);
+            data.sources.get_mut("x").unwrap().values_non_negative = true;
+            let selection = select_plan(
+                &candidates,
+                &[entry_with(target.clone(), repeating(10_000))],
+                &data,
+                PlanningModels::builtin(),
+            )
+            .unwrap();
+            let selected = candidates
+                .iter()
+                .find(|c| c.id == selection.selected)
+                .unwrap();
+            assert!(
+                !has_summary_estimate(selected),
+                "{query}: {}",
+                selection.selected
+            );
+        }
+    }
+
+    /// `topk(k, irate(m[w]))`, now a heavy-hitter ranking as `rate`'s is,
+    /// still plans: Stages 1-3 compose its candidates and select one.
+    #[test]
+    fn a_direct_irate_ranking_plans() {
+        let target = AccuracyTarget::max_error(0.01).failure_probability(0.001);
+        for function in ["rate", "irate"] {
+            let query = format!("topk(10, {function}(data_total[5m]))");
+            let candidates = all_candidates(&query, &target);
+            let data = with_metric_types(data(), &[("data_total", MetricType::Counter)]);
+            let selection = select_plan(
+                &candidates,
+                &[entry_with(target.clone(), repeating(10_000))],
+                &data,
+                PlanningModels::builtin(),
+            );
+            assert!(selection.is_ok(), "{query}: {selection:?}");
+        }
+    }
+
     /// A candidate that ties the selected one is rejected for losing on
     /// enumeration order, not as costlier.
     #[test]

crates/types/src/workload/mod.rs

@@ -510,6 +510,14 @@ pub struct SourceFacts {
     /// undeclared metric's samples may take any value.
     #[serde(default)]
     pub metric_type: Option<MetricType>,
+    /// Whether every sample of the metric is non-negative (default `false`:
+    /// unknown). A counter's samples always are; this declares it for a
+    /// gauge, such as a latency or a queue length. The deployment must
+    /// ensure it. Plan selection reads it as the sign of the metric's
+    /// samples, which a relative-error bound needs to carry through a
+    /// `sum` or `avg`. Omitted from JSON when `false`.
+    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
+    pub values_non_negative: bool,
     /// PromQL and MetricsQL only: how often each series of the metric
     /// receives a sample (default: unknown). A bare instant selector over
     /// the metric reads the latest sample within it; lowering such a
@@ -1168,6 +1176,7 @@ mod tests {
                     SourceFacts {
                         series_count: Some(10_000),
                         metric_type: Some(MetricType::Counter),
+                        values_non_negative: true,
                         sample_interval: Some(DurationMs(15_000)),
                         stream_norms: Some(StreamNorms {
                             l1: Some(1e9),
@@ -1187,6 +1196,7 @@ mod tests {
                 "sources": {"requests_total": {
                     "series_count": 10000,
                     "metric_type": "counter",
+                    "values_non_negative": true,
                     "sample_interval": 15000,
                     "stream_norms": {"l1": 1e9, "l2": null},
                     "max_distinct_per_evaluation": 100