From 66c21ec5a96d19a58a301b3f3c1f61e6b497f7b8 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Tue, 20 Oct 2026 16:03:08 +0000 Subject: [PATCH 1/2] refactor(executor): use the shared STALE_MARKER in current-series selection current_series and its row reference compared sample bits against the literal 0x7ff0_0000_0000_0002; they now read operators::STALE_MARKER, its one definition. --- crates/executor/src/operators/current_series.rs | 2 +- crates/executor/src/operators/row_reference/current_series.rs | 2 +- crates/executor/src/operators/row_reference/mod.rs | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/crates/executor/src/operators/current_series.rs b/crates/executor/src/operators/current_series.rs index bbd4acea1..d2bb586e0 100644 --- a/crates/executor/src/operators/current_series.rs +++ b/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::>(); let selected = batch.take(&arrow::array::UInt32Array::from(rows))?; let stamped = std::sync::Arc::new(arrow::array::TimestampMillisecondArray::from(vec![ diff --git a/crates/executor/src/operators/row_reference/current_series.rs b/crates/executor/src/operators/row_reference/current_series.rs index b6a6b4f93..b84f0e789 100644 --- a/crates/executor/src/operators/row_reference/current_series.rs +++ b/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]))?; diff --git a/crates/executor/src/operators/row_reference/mod.rs b/crates/executor/src/operators/row_reference/mod.rs index ecfa85307..80c547970 100644 --- a/crates/executor/src/operators/row_reference/mod.rs +++ b/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![ From 4dc7cd0ce2588bebb03b00bb8a8078b45d925925 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Tue, 20 Oct 2026 16:03:08 +0000 Subject: [PATCH 2/2] feat(plan-selection): rate, increase and irate of a declared counter are non-negative Count-Min weights (#773) - WeightDomain::CounterDerivativeOfMetric { metric }: an open requirement that heap weights are reset-corrected rate/increase/irate values of `metric`'s samples (or plain sums of them). Plan selection confirms it as NonNegative { ResetAwareCounterDerivative } when the data workload declares `metric` a counter, and leaves it unproven otherwise. - The keyed top-k realization over `sum by (...) (rate|increase|irate(m))` records that requirement instead of proving it from the operator alone: over a gauge, a negative sample can give a negative rate, so the Count-Min-with-heap candidate is now rejected with CountMinWeightsNotProvenNonNegative. Over a declared counter it passes the weight check and goes on to the accuracy and cost checks. - The PromQL frontend recognizes `topk(k, sum by (...) (irate(m[w])))` as a heavy-hitter ranking, as it does for rate and increase. - Test: topk(10, sum by (job) (increase|rate|irate(data_total[5m]))) with data_total declared a counter and a gauge. --- crates/frontend-promql/src/promql.rs | 5 +- .../src/per_computation_candidates.rs | 2 +- .../src/realization.rs | 36 +++++++-- crates/plan-selection/src/lib.rs | 73 +++++++++++++++++++ crates/types/src/ir/schema/sketch.rs | 37 ++++++++-- 5 files changed, 137 insertions(+), 16 deletions(-) diff --git a/crates/frontend-promql/src/promql.rs b/crates/frontend-promql/src/promql.rs index f05dc4474..89775fc81 100644 --- a/crates/frontend-promql/src/promql.rs +++ b/crates/frontend-promql/src/promql.rs @@ -798,7 +798,10 @@ fn build_over_sub_dag(outer: Outer, keys: Vec, 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, .. diff --git a/crates/logical-asap-optimization/src/per_computation_candidates.rs b/crates/logical-asap-optimization/src/per_computation_candidates.rs index 86ad98f8d..59a2f0953 100644 --- a/crates/logical-asap-optimization/src/per_computation_candidates.rs +++ b/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 { +pub(crate) fn samples_of_metric(node: &OperatorNode) -> Option { match node.non_asap()? { NonASAPOp::Scan { source: Source::TimeSeries { metric }, diff --git a/crates/logical-asap-optimization/src/realization.rs b/crates/logical-asap-optimization/src/realization.rs index 9eec31df5..e845d9c44 100644 --- a/crates/logical-asap-optimization/src/realization.rs +++ b/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; diff --git a/crates/plan-selection/src/lib.rs b/crates/plan-selection/src/lib.rs index d83f9719c..ead610339 100644 --- a/crates/plan-selection/src/lib.rs +++ b/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 { + 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. diff --git a/crates/types/src/ir/schema/sketch.rs b/crates/types/src/ir/schema/sketch.rs index 78c80569f..b280844cb 100644 --- a/crates/types/src/ir/schema/sketch.rs +++ b/crates/types/src/ir/schema/sketch.rs @@ -694,13 +694,27 @@ 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) => { @@ -708,7 +722,13 @@ impl WeightDomain { 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