fix(plan-selection): prove Count-Min weights over a gauge declared non-negative

crates/logical-asap-optimization/src/accuracy/composition.rs

@@ -51,8 +51,8 @@ pub(crate) fn source_norm_statistic(norm: &str, source: &str) -> String {
 /// Whether every value folded is non-negative, for values that are raw
 /// samples of `metric` or plain sums of them, e.g.
 /// `values_non_negative[m]`: an indicator, 1 when proven. Plan selection
-/// binds it to 1 when the data workload declares `metric` a counter, and
-/// leaves it unbound otherwise.
+/// binds it to 1 when the data workload declares `metric` a counter or
+/// non-negative, and leaves it unbound otherwise.
 pub fn samples_sign_statistic(metric: &str) -> String {
     format!("{VALUES_NON_NEGATIVE}[{metric}]")
 }

crates/plan-selection/src/lib.rs

@@ -741,8 +741,8 @@ pub enum RejectionCause {
     /// it and no deployment evidence meets it.
     AccuracyTargetMissed,
     /// A query's plan reads a Count-Min sketch whose update weights are not
-    /// proven non-negative (e.g. samples of a metric not declared a
-    /// counter). The executor cannot run it, so the deployment's accuracy
+    /// proven non-negative (e.g. samples of a metric declared neither a
+    /// counter nor non-negative). The executor cannot run it, so the deployment's accuracy
     /// model cannot accept it either.
     CountMinWeightsNotProvenNonNegative,
     /// It reduces each series' raw samples with an exact aggregate the
@@ -1997,7 +1997,8 @@ pub fn resize_for_failure_probability(
 /// unknown leaves the target unmet. Count-Min is admitted only over update
 /// weights proven non-negative, which its bound and the executor assume: an open
 /// [`WeightDomain::SamplesOfMetric`] requirement is met when `data` declares
-/// the metric a counter ([`confirmed_weights`]), and unmet otherwise.
+/// the metric a counter or non-negative ([`confirmed_weights`]), and unmet
+/// otherwise.
 ///
 /// A target that sets no metric accepts a guarantee in any metric that meets
 /// its ε, δ and scope ([`AccuracyModel::satisfies`]); one that sets a metric
@@ -2048,9 +2049,10 @@ fn accuracy_check(
                 )
             {
                 let unconfirmed = match &input.weight_domain {
-                    WeightDomain::SamplesOfMetric { metric } => {
-                        format!(" (samples of `{metric}`, which is not declared a counter)")
-                    }
+                    WeightDomain::SamplesOfMetric { metric } => format!(
+                        " (samples of `{metric}`, which is not declared a counter or \
+                         non-negative)"
+                    ),
                     WeightDomain::CounterDerivativeOfMetric { metric } => format!(
                         " (rates or increases of `{metric}`, which is not declared a counter)"
                     ),
@@ -2213,11 +2215,15 @@ fn guarantee_check(
     })
 }
 
-/// `domain` with an open [`WeightDomain::SamplesOfMetric`] requirement
-/// answered from `data`'s declared metric types: proven when the metric is
-/// declared a counter, unknown otherwise. No type is inferred from a name.
+/// `domain` with an open requirement answered from `data`'s declared source
+/// facts ([`WeightDomain::confirmed`]): samples are proven when the metric is
+/// declared a counter or non-negative, a counter derivative only when it is
+/// declared a counter; unknown otherwise. No type is inferred from a name.
 pub fn confirmed_weights(domain: &WeightDomain, data: &DataWorkload) -> WeightDomain {
-    domain.confirmed(|metric| declared_counter(data, metric))
+    domain.confirmed(
+        |metric| declared_counter(data, metric),
+        |metric| declared_non_negative(data, metric),
+    )
 }
 
 /// Whether `data` declares every sample of `metric` non-negative: the
@@ -4710,6 +4716,84 @@ mod tests {
         }
     }
 
+    /// A gauge declared non-negative
+    /// ([`SourceFacts::values_non_negative`]) proves its raw samples, and
+    /// plain sums of them, non-negative: the Count-Min-with-heap top-k of
+    /// `topk by (job) (10, sum_over_time(m[1m]))` (the ASAPQuery workloads'
+    /// top-k-sum shape) passes the weight check, and the selected plan carries
+    /// [`NonNegativeWeightProof::DeclaredNonNegativeSamples`]. A rate or
+    /// increase of that gauge stays unproven: a reset correction over a
+    /// gauge can still give a negative value.
+    #[test]
+    fn count_min_over_samples_of_a_gauge_declared_non_negative_is_admitted() {
+        use asap_types::ir::schema::NonNegativeWeightProof;
+        let target = AccuracyTarget::max_error(0.01)
+            .failure_probability(0.001)
+            .accuracy_metric(AccuracyMetric::ABSOLUTE_STREAM_L1);
+        let mut gauge = with_metric_types(data(), &[("m", MetricType::Gauge)]);
+        gauge.sources.get_mut("m").unwrap().values_non_negative = true;
+        for (query, admitted) in [
+            ("topk by (job) (10, sum_over_time(m[1m]))", true),
+            ("topk(10, sum by (job) (increase(m[5m])))", false),
+        ] {
+            let candidates = all_candidates(query, &target);
+            let count_min: Vec<&PhysicalCandidate> = candidates
+                .iter()
+                .filter(|c| count_min_builds(c).contains(&SketchAlgorithm::CmsWithHeap))
+                .collect();
+            assert!(!count_min.is_empty(), "{query}");
+            let rejected = match select_plan(
+                &candidates,
+                &[entry_with(target.clone(), repeating(10_000))],
+                &gauge,
+                PlanningModels::builtin(),
+            ) {
+                Ok(selection) => selection.rejected,
+                Err(SelectionError::NoValidCandidate(rejected)) => rejected,
+                Err(error) => panic!("{query}: {error}"),
+            };
+            for candidate in &count_min {
+                let weights_rejected = rejected.iter().any(|r| {
+                    r.id == candidate.id
+                        && r.cause == RejectionCause::CountMinWeightsNotProvenNonNegative
+                });
+                assert_eq!(
+                    weights_rejected, !admitted,
+                    "{query}, {}: {rejected:?}",
+                    candidate.id
+                );
+                if !admitted {
+                    continue;
+                }
+                // The proof the executor reads.
+                let proofs: Vec<WeightDomain> = with_confirmed_weights(&candidate.roots, &gauge)
+                    .iter()
+                    .flat_map(root_operators)
+                    .flat_map(OperatorNode::reachable)
+                    .filter_map(|n| match n.asap() {
+                        Some(ASAPOp::SummaryAgg {
+                            input,
+                            family: FieldDataType::Sketch(kind, _),
+                            ..
+                        }) if *kind.algorithm() == SketchAlgorithm::CmsWithHeap => {
+                            Some(input.weight_domain.clone())
+                        }
+                        _ => None,
+                    })
+                    .collect();
+                assert!(!proofs.is_empty(), "{}", candidate.id);
+                assert!(
+                    proofs.iter().all(|domain| *domain
+                        == WeightDomain::NonNegative {
+                            proof: NonNegativeWeightProof::DeclaredNonNegativeSamples
+                        }),
+                    "{}: {proofs:?}",
+                    candidate.id
+                );
+            }
+        }
+    }
+
     /// 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.
@@ -4928,6 +5012,14 @@ mod tests {
                 proof: NonNegativeWeightProof::CounterSamples
             }]
         );
+        let mut gauge = with_metric_types(data(), &[("m", MetricType::Gauge)]);
+        gauge.sources.get_mut("m").unwrap().values_non_negative = true;
+        assert_eq!(
+            domains(&with_confirmed_weights(&p2.roots, &gauge)),
+            [WeightDomain::NonNegative {
+                proof: NonNegativeWeightProof::DeclaredNonNegativeSamples
+            }]
+        );
         assert_eq!(
             domains(&with_confirmed_weights(&p2.roots, &data())),
             [WeightDomain::UnknownOrSigned]

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

@@ -693,10 +693,12 @@ pub enum WeightDomain {
     },
     /// An open requirement, not a proof: the weights are raw samples of
     /// `metric`, or plain sums of them (no rate, no other arithmetic), so they
-    /// are non-negative exactly when `metric` is a counter. Logical
-    /// optimization records it without reading the data workload; plan
-    /// selection confirms it from the declared metric types
-    /// ([`SourceFacts::metric_type`](crate::workload::SourceFacts::metric_type)).
+    /// are non-negative when every sample of `metric` is: `metric` is
+    /// declared a counter, or declared non-negative. Logical optimization
+    /// records it without reading the data workload; plan selection confirms
+    /// it from the declared source facts
+    /// ([`SourceFacts::metric_type`](crate::workload::SourceFacts::metric_type),
+    /// [`SourceFacts::values_non_negative`](crate::workload::SourceFacts::values_non_negative)).
     /// Unconfirmed, it counts as not proven.
     SamplesOfMetric {
         metric: String,
@@ -715,20 +717,31 @@ pub enum WeightDomain {
 }
 
 impl WeightDomain {
-    /// 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
+    /// The domain once the declared facts have answered an open requirement.
+    /// [`SamplesOfMetric`](Self::SamplesOfMetric) is proven when the metric
+    /// is a counter ([`NonNegativeWeightProof::CounterSamples`]) or else
+    /// declared non-negative
+    /// ([`NonNegativeWeightProof::DeclaredNonNegativeSamples`]);
+    /// [`CounterDerivativeOfMetric`](Self::CounterDerivativeOfMetric) only
+    /// when it is a counter
+    /// ([`NonNegativeWeightProof::ResetAwareCounterDerivative`]). Otherwise
     /// unknown. Any other domain is returned unchanged.
-    pub fn confirmed(&self, is_counter: impl Fn(&str) -> bool) -> WeightDomain {
+    pub fn confirmed(
+        &self,
+        is_counter: impl Fn(&str) -> bool,
+        is_declared_non_negative: impl Fn(&str) -> bool,
+    ) -> WeightDomain {
         match self {
             WeightDomain::SamplesOfMetric { metric } if is_counter(metric) => {
                 WeightDomain::NonNegative {
                     proof: NonNegativeWeightProof::CounterSamples,
                 }
             }
+            WeightDomain::SamplesOfMetric { metric } if is_declared_non_negative(metric) => {
+                WeightDomain::NonNegative {
+                    proof: NonNegativeWeightProof::DeclaredNonNegativeSamples,
+                }
+            }
             WeightDomain::CounterDerivativeOfMetric { metric } if is_counter(metric) => {
                 WeightDomain::NonNegative {
                     proof: NonNegativeWeightProof::ResetAwareCounterDerivative,
@@ -757,6 +770,12 @@ pub enum NonNegativeWeightProof {
     /// gives it on confirming a
     /// [`WeightDomain::SamplesOfMetric`] requirement.
     CounterSamples,
+    /// Raw samples of a metric declared non-negative
+    /// ([`SourceFacts::values_non_negative`](crate::workload::SourceFacts::values_non_negative)),
+    /// such as a gauge, or plain sums of them. Plan selection gives it on
+    /// confirming a [`WeightDomain::SamplesOfMetric`] requirement for a
+    /// metric not declared a counter.
+    DeclaredNonNegativeSamples,
     /// A constant weight greater than 0, such as `Constant(2.0)`: every row
     /// adds the same positive amount, so no weight is negative.
     PositiveConstant,

crates/types/src/workload/mod.rs

@@ -527,7 +527,8 @@ pub struct SourceFacts {
     /// 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`.
+    /// `sum` or `avg`, and as proof that Count-Min weights over the raw
+    /// samples are non-negative. 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

tools/dag-viewer/examples/random-mixed-seed-862-long-workload-01.json

@@ -20552,7 +20552,7 @@
           "candidates_checked": 3,
           "discarded": [
             {
-              "cause": "count_min_weights_not_proven_non_negative",
+              "cause": "no_accuracy_guarantee",
               "choice": [
                 [
                   10,
@@ -20563,7 +20563,7 @@
                   0
                 ]
               ],
-              "reason": "CmsWithHeap needs non-negative update weights, and these are not proven non-negative (samples of `data`, which is not declared a counter)"
+              "reason": "the output has no guarantee, and the deployment has no evidence for it, so it cannot meet target AccuracyTarget { accuracy_metric: None, max_error: 0.05000000000000005, failure_probability: Some(0.3), accuracy_scope: WholeResult }"
             },
             {
               "cause": "no_accuracy_guarantee",
@@ -20585,7 +20585,7 @@
           "candidates_checked": 3,
           "discarded": [
             {
-              "cause": "count_min_weights_not_proven_non_negative",
+              "cause": "no_accuracy_guarantee",
               "choice": [
                 [
                   12,
@@ -20596,7 +20596,7 @@
                   0
                 ]
               ],
-              "reason": "CmsWithHeap needs non-negative update weights, and these are not proven non-negative (samples of `data`, which is not declared a counter)"
+              "reason": "the output has no guarantee, and the deployment has no evidence for it, so it cannot meet target AccuracyTarget { accuracy_metric: None, max_error: 0.05000000000000005, failure_probability: Some(0.3), accuracy_scope: WholeResult }"
             },
             {
               "cause": "no_accuracy_guarantee",

tools/dag-viewer/examples/random-mixed-seed-862-workload-01.json

@@ -20552,7 +20552,7 @@
           "candidates_checked": 3,
           "discarded": [
             {
-              "cause": "count_min_weights_not_proven_non_negative",
+              "cause": "no_accuracy_guarantee",
               "choice": [
                 [
                   10,
@@ -20563,7 +20563,7 @@
                   0
                 ]
               ],
-              "reason": "CmsWithHeap needs non-negative update weights, and these are not proven non-negative (samples of `data`, which is not declared a counter)"
+              "reason": "the output has no guarantee, and the deployment has no evidence for it, so it cannot meet target AccuracyTarget { accuracy_metric: None, max_error: 0.05000000000000005, failure_probability: Some(0.3), accuracy_scope: WholeResult }"
             },
             {
               "cause": "no_accuracy_guarantee",
@@ -20585,7 +20585,7 @@
           "candidates_checked": 3,
           "discarded": [
             {
-              "cause": "count_min_weights_not_proven_non_negative",
+              "cause": "no_accuracy_guarantee",
               "choice": [
                 [
                   12,
@@ -20596,7 +20596,7 @@
                   0
                 ]
               ],
-              "reason": "CmsWithHeap needs non-negative update weights, and these are not proven non-negative (samples of `data`, which is not declared a counter)"
+              "reason": "the output has no guarantee, and the deployment has no evidence for it, so it cannot meet target AccuracyTarget { accuracy_metric: None, max_error: 0.05000000000000005, failure_probability: Some(0.3), accuracy_scope: WholeResult }"
             },
             {
               "cause": "no_accuracy_guarantee",

tools/dag-viewer/examples/random-mixed-seed-862-x5-workload-01.json

@@ -20552,7 +20552,7 @@
           "candidates_checked": 3,
           "discarded": [
             {
-              "cause": "count_min_weights_not_proven_non_negative",
+              "cause": "no_accuracy_guarantee",
               "choice": [
                 [
                   10,
@@ -20563,7 +20563,7 @@
                   0
                 ]
               ],
-              "reason": "CmsWithHeap needs non-negative update weights, and these are not proven non-negative (samples of `data`, which is not declared a counter)"
+              "reason": "the output has no guarantee, and the deployment has no evidence for it, so it cannot meet target AccuracyTarget { accuracy_metric: None, max_error: 0.05000000000000005, failure_probability: Some(0.3), accuracy_scope: WholeResult }"
             },
             {
               "cause": "no_accuracy_guarantee",
@@ -20585,7 +20585,7 @@
           "candidates_checked": 3,
           "discarded": [
             {
-              "cause": "count_min_weights_not_proven_non_negative",
+              "cause": "no_accuracy_guarantee",
               "choice": [
                 [
                   12,
@@ -20596,7 +20596,7 @@
                   0
                 ]
               ],
-              "reason": "CmsWithHeap needs non-negative update weights, and these are not proven non-negative (samples of `data`, which is not declared a counter)"
+              "reason": "the output has no guarantee, and the deployment has no evidence for it, so it cannot meet target AccuracyTarget { accuracy_metric: None, max_error: 0.05000000000000005, failure_probability: Some(0.3), accuracy_scope: WholeResult }"
             },
             {
               "cause": "no_accuracy_guarantee",