feat: rate, increase and irate of a declared counter are non-negative Count-Min weights (#773)

crates/executor/src/operators/current_series.rs

@@ -95,7 +95,7 @@ pub(super) fn execute<'a>(
             .groups_in_key_order()
             .into_iter()
             .filter_map(|series| latest[series as usize])
-            .filter(|&row| values.value(row as usize).to_bits() != 0x7ff0_0000_0000_0002)
+            .filter(|&row| values.value(row as usize).to_bits() != STALE_MARKER)
             .collect::<Vec<_>>();
         let selected = batch.take(&arrow::array::UInt32Array::from(rows))?;
         let stamped = std::sync::Arc::new(arrow::array::TimestampMillisecondArray::from(vec![

crates/executor/src/operators/row_reference/current_series.rs

@@ -64,7 +64,7 @@ pub(super) fn execute<'a>(
             let Value::Float64(sample) = rows[index][value] else {
                 unreachable!()
             };
-            if sample.to_bits() == 0x7ff0_0000_0000_0002 {
+            if sample.to_bits() == crate::operators::STALE_MARKER {
                 continue;
             }
             workspace.grow(row_bytes(&rows[index]))?;

crates/executor/src/operators/row_reference/mod.rs

@@ -344,7 +344,7 @@ mod tests {
                     .map(|i| {
                         let series = g.next() % 4;
                         let value = match g.next() % 9 {
-                            0 => f64::from_bits(0x7ff0_0000_0000_0002),
+                            0 => f64::from_bits(crate::operators::STALE_MARKER),
                             n => (n as f64) * 1.5,
                         };
                         vec![

crates/frontend-promql/src/promql.rs

@@ -798,7 +798,10 @@ fn build_over_sub_dag(outer: Outer, keys: Vec<ColumnRef>, child: Unresolved) ->
                     ..
                 } if matches!(measures.as_slice(), [AggIntent::Sum { .. }])
                     && matches!(sum_child.as_ref(), Unresolved::Aggregate { measures, .. }
-                        if matches!(measures.as_slice(), [AggIntent::Rate | AggIntent::Increase]))
+                        if matches!(
+                            measures.as_slice(),
+                            [AggIntent::Rate | AggIntent::Increase | AggIntent::IRate]
+                        ))
             );
             let direct_counter_ranking = matches!(&child, Unresolved::Aggregate {
                 measures, reduction: Reduction::PerEntity, ..

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

@@ -945,7 +945,7 @@ pub(crate) fn partitions_over(
 /// included) ends it. The metric's type is not read here: whether its
 /// samples are non-negative is plan selection's question
 /// ([`WeightDomain::SamplesOfMetric`]).
-fn samples_of_metric(node: &OperatorNode) -> Option<String> {
+pub(crate) fn samples_of_metric(node: &OperatorNode) -> Option<String> {
     match node.non_asap()? {
         NonASAPOp::Scan {
             source: Source::TimeSeries { metric },

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

@@ -285,9 +285,28 @@ pub(crate) fn realize_keyed_additive_summary_input(
     if asap_types::ir::operator::non_asap::any_measure_filtered(filters) {
         return None;
     }
-    let counter_input = matches!(measures.as_slice(), [AggIntent::Sum { .. }])
-        && matches!(raw_child.non_asap(), Some(NonASAPOp::Aggregate { measures, .. })
-            if matches!(measures.as_slice(), [AggIntent::Rate | AggIntent::Increase]));
+    // A grouped sum of per-series `rate`/`increase`/`irate` values: the
+    // metric they derive from, whose declared type decides their sign.
+    let derivative_of = match (measures.as_slice(), raw_child.non_asap()) {
+        (
+            [AggIntent::Sum { .. }],
+            Some(NonASAPOp::Aggregate {
+                measures: derivative,
+                child: samples,
+                ..
+            }),
+        ) if matches!(
+            derivative.as_slice(),
+            [AggIntent::Rate | AggIntent::Increase | AggIntent::IRate]
+        ) =>
+        {
+            Some(crate::per_computation_candidates::samples_of_metric(
+                samples,
+            ))
+        }
+        _ => None,
+    };
+    let counter_input = derivative_of.is_some();
     let weight = match measures.as_slice() {
         [AggIntent::Count { .. }] => SummaryInputExpr::Constant(1.0),
         [AggIntent::Sum { .. }] if counter_input => {
@@ -309,13 +328,16 @@ pub(crate) fn realize_keyed_additive_summary_input(
         [AggIntent::Count { .. }] => WeightDomain::NonNegative {
             proof: NonNegativeWeightProof::UnitCount,
         },
-        [AggIntent::Sum { .. }] if counter_input => WeightDomain::NonNegative {
-            proof: NonNegativeWeightProof::ResetAwareCounterDerivative,
+        _ => match derivative_of {
+            Some(Some(metric)) => WeightDomain::CounterDerivativeOfMetric { metric },
+            _ => WeightDomain::UnknownOrSigned,
         },
-        _ => WeightDomain::UnknownOrSigned,
     };
     if matches!(heap_algorithm, SketchAlgorithm::CmsWithHeap)
-        && !matches!(weight_domain, WeightDomain::NonNegative { .. })
+        && !matches!(
+            weight_domain,
+            WeightDomain::NonNegative { .. } | WeightDomain::CounterDerivativeOfMetric { .. }
+        )
     {
         // Unsupported: value-weighted CMS requires non-negative update evidence; use CountSketch for arbitrary values.
         return None;

crates/plan-selection/src/lib.rs

@@ -1608,6 +1608,9 @@ fn accuracy_check(
                     WeightDomain::SamplesOfMetric { metric } => {
                         format!(" (samples of `{metric}`, which is not declared a counter)")
                     }
+                    WeightDomain::CounterDerivativeOfMetric { metric } => format!(
+                        " (rates or increases of `{metric}`, which is not declared a counter)"
+                    ),
                     _ => String::new(),
                 };
                 // Count-Min (plain, CmsWithHeap or Hydra Count-Min) over
@@ -3826,6 +3829,76 @@ mod tests {
         }
     }
 
+    /// The Count-Min sketch algorithms (plain or with a heap) `candidate`
+    /// builds.
+    fn count_min_builds(candidate: &PhysicalCandidate) -> Vec<SketchAlgorithm> {
+        candidate
+            .dag
+            .nodes
+            .iter()
+            .filter_map(|node| match &node.payload {
+                Payload::ASAP(ASAPOp::SummaryAgg {
+                    family: FieldDataType::Sketch(kind, _),
+                    ..
+                }) if matches!(
+                    kind.algorithm(),
+                    SketchAlgorithm::Cms | SketchAlgorithm::CmsWithHeap
+                ) =>
+                {
+                    Some(kind.algorithm().clone())
+                }
+                _ => None,
+            })
+            .collect()
+    }
+
+    /// `increase` (and `rate`, `irate`) over a counter is non-negative, since
+    /// resets are corrected: the Count-Min-with-heap top-k of
+    /// `topk(10, sum by (job) (increase(data_total[5m])))` passes the weight
+    /// check when `data_total` is declared a counter, and goes on to the
+    /// accuracy and cost checks. Over a gauge it is still rejected for its
+    /// weights (issue #773).
+    #[test]
+    fn count_min_over_a_counter_derivative_needs_a_declared_counter() {
+        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);
+            let count_min: Vec<&str> = candidates
+                .iter()
+                .filter(|c| count_min_builds(c).contains(&SketchAlgorithm::CmsWithHeap))
+                .map(|c| c.id.as_str())
+                .collect();
+            assert!(!count_min.is_empty(), "{query}");
+            for (metric_type, admitted) in [(MetricType::Counter, true), (MetricType::Gauge, false)]
+            {
+                let data = with_metric_types(data(), &[("data_total", metric_type)]);
+                let rejected = match select_plan(
+                    &candidates,
+                    &[entry_with(target.clone(), repeating(10_000))],
+                    &data,
+                    PlanningModels::builtin(),
+                ) {
+                    Ok(selection) => selection.rejected,
+                    Err(SelectionError::NoValidCandidate(rejected)) => rejected,
+                    Err(error) => panic!("{query}: {error}"),
+                };
+                for id in &count_min {
+                    let weights_rejected = rejected.iter().any(|r| {
+                        &r.id == id
+                            && r.cause == RejectionCause::CountMinWeightsNotProvenNonNegative
+                    });
+                    assert_eq!(
+                        weights_rejected, !admitted,
+                        "{query} over a {metric_type:?}, {id}: {rejected:?}"
+                    );
+                }
+            }
+        }
+    }
+
     /// The sign gate is an indicator that only 1 (proven) opens: any other
     /// value a binder supplies, 0 included, leaves it unbound rather than
     /// reading as a perfect bound. Other statistics pass through.

crates/types/src/ir/schema/sketch.rs

@@ -694,21 +694,41 @@ pub enum WeightDomain {
     SamplesOfMetric {
         metric: String,
     },
+    /// An open requirement, not a proof: the weights are reset-corrected
+    /// `rate`, `increase` or `irate` values of `metric`'s samples, or plain
+    /// sums of them. Reset correction keeps them non-negative exactly when
+    /// the samples are, that is when `metric` is a counter; over a gauge a
+    /// negative sample can give a negative value. Plan selection confirms
+    /// it from the declared metric types, as
+    /// [`SamplesOfMetric`](Self::SamplesOfMetric). Unconfirmed, it counts as
+    /// not proven.
+    CounterDerivativeOfMetric {
+        metric: String,
+    },
 }
 
 impl WeightDomain {
-    /// The domain once `is_counter` has answered an open
-    /// [`SamplesOfMetric`](Self::SamplesOfMetric) requirement: proven
-    /// ([`NonNegativeWeightProof::CounterSamples`]) when the metric is a
-    /// counter, else unknown. Any other domain is returned unchanged.
+    /// The domain once `is_counter` has answered an open requirement
+    /// ([`SamplesOfMetric`](Self::SamplesOfMetric) or
+    /// [`CounterDerivativeOfMetric`](Self::CounterDerivativeOfMetric)):
+    /// proven when the metric is a counter
+    /// ([`NonNegativeWeightProof::CounterSamples`] or
+    /// [`NonNegativeWeightProof::ResetAwareCounterDerivative`]), else
+    /// unknown. Any other domain is returned unchanged.
     pub fn confirmed(&self, is_counter: impl Fn(&str) -> bool) -> WeightDomain {
         match self {
             WeightDomain::SamplesOfMetric { metric } if is_counter(metric) => {
                 WeightDomain::NonNegative {
                     proof: NonNegativeWeightProof::CounterSamples,
                 }
             }
-            WeightDomain::SamplesOfMetric { .. } => WeightDomain::UnknownOrSigned,
+            WeightDomain::CounterDerivativeOfMetric { metric } if is_counter(metric) => {
+                WeightDomain::NonNegative {
+                    proof: NonNegativeWeightProof::ResetAwareCounterDerivative,
+                }
+            }
+            WeightDomain::SamplesOfMetric { .. }
+            | WeightDomain::CounterDerivativeOfMetric { .. } => WeightDomain::UnknownOrSigned,
             other => other.clone(),
         }
     }
@@ -718,8 +738,11 @@ impl WeightDomain {
 #[serde(rename_all = "snake_case")]
 pub enum NonNegativeWeightProof {
     UnitCount,
-    /// PromQL counter reset correction produces a non-negative increase; rate
-    /// divides that increase by a positive duration.
+    /// `rate`, `increase` or `irate` of a counter's samples, or plain sums
+    /// of them: reset correction over non-negative samples gives a
+    /// non-negative increase, and a rate divides it by a positive duration.
+    /// Plan selection gives it on confirming a
+    /// [`WeightDomain::CounterDerivativeOfMetric`] requirement.
     ResetAwareCounterDerivative,
     /// Raw samples of a metric declared a counter
     /// ([`MetricType::Counter`](crate::workload::MetricType::Counter)), or