[zzylol]

Why

The Count-Min top-k of topk(10, sum by (job) (increase(data_total[5m]))) adds each series' increase as a weight, and Count-Min's bound and the executor both need those weights to be non-negative. Reset correction keeps rate, increase and irate non-negative only when the samples themselves are, that is, when the metric is a counter. So whether these candidates are valid depends on the metric's declared type.

What main did differed from what the issue describes. I checked it with the acceptance test below:

• increase and rate: Pass 1 proved the weights non-negative (ResetAwareCounterDerivative) from the operator alone, without reading the metric's type. The Count-Min-with-heap candidate therefore passed the weight check over a counter, which is what the issue asks for, but also over a gauge, where the proof does not hold.
• irate: the frontend did not lower topk(k, sum by (...) (irate(...))) to a heavy-hitter top-k, so there was no Count-Min candidate at all.

I could not reproduce the rejection the issue reports over a counter on this shape. This PR implements the expected behaviour the issue states: admitted over a declared counter, rejected over a gauge.

What

• Over a metric declared a counter, the weights of a rate, increase or irate top-k are proven non-negative. The Count-Min-with-heap candidate passes the weight check and still goes through the accuracy and cost checks.
• Over a gauge, or a metric with no declared type, that candidate is rejected with CountMinWeightsNotProvenNonNegative. The reason names the metric: … (rates or increases of \data_total`, which is not declared a counter)`.
• topk(k, sum by (...) (irate(m[w]))) is lowered to a heavy-hitter top-k, as it already was for rate and increase.
• The review nit from fix: run exact per-series range reductions and reject per-entity aggregates the executor cannot run (#770) #775 is folded in: current_series.rs and its row reference compare against the shared STALE_MARKER instead of the literal.

How

This follows the existing WeightDomain::SamplesOfMetric pattern. Logical optimization records an open requirement without reading the data workload, and Stage 3 confirms it from the declared metric types.

• Realization. The keyed top-k realization of a grouped sum of per-series rate, increase or irate records CounterDerivativeOfMetric { metric } instead of an unconditional proof. The metric comes from the derivative's input, through the same samples_of_metric rule.
• Stage 3. WeightDomain::confirmed turns the requirement into NonNegative { ResetAwareCounterDerivative } for a declared counter, and into UnknownOrSigned otherwise. Stage 3's existing Count-Min weight check and with_confirmed_weights (the plan the executor runs) both call it, so the executor still receives the proof it already keys on.
• When the metric is unknown. If the derivative's input is not raw samples of one metric (for example a subquery), the weights are UnknownOrSigned, so no Count-Min candidate is built. Count-Sketch, which needs no sign, is unaffected.

Key code interfaces added / modified

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

/// The sign of a summary update's weights: a proof, unknown, or an open
/// requirement that plan selection answers from the declared metric types.
/// The type stays; one variant is added.
pub enum WeightDomain {
    /// No proof; the default.
    UnknownOrSigned, // unchanged
    /// Proven non-negative, with the proof.
    NonNegative { proof: NonNegativeWeightProof }, // unchanged
    /// Open: raw samples of `metric` or plain sums of them; non-negative exactly
    /// when `metric` is a counter.
    SamplesOfMetric { metric: String }, // unchanged
    /// Open: reset-corrected rate/increase/irate values of `metric`'s samples, or
    /// plain sums of them; non-negative exactly when `metric` is a counter.
    /// Unconfirmed, it counts as not proven (the request is unsatisfiable
    /// over a gauge, not bad input). JSON kind "counter_derivative_of_metric".
    CounterDerivativeOfMetric { metric: String }, // new
}

impl WeightDomain {
    /// The domain once `is_counter` has answered an open requirement:
    /// CounterSamples or ResetAwareCounterDerivative for a counter, else unknown.
    pub fn confirmed(&self, is_counter: impl Fn(&str) -> bool) -> WeightDomain; // unchanged signature; now also answers CounterDerivativeOfMetric
}

Usage (the test count_min_over_a_counter_derivative_needs_a_declared_counter):

// The user's query, at an absolute (stream-L1) error of 0.01 with δ = 0.001.
let target = AccuracyTarget::max_error(0.01).failure_probability(0.001)
    .accuracy_metric(AccuracyMetric::ABSOLUTE_STREAM_L1);
// Stages 1-2: the exact plan (P1), the Count-Min-with-heap top-k (P2) and the
// Count-Sketch-with-heap top-k (P3). P2's update carries
// CounterDerivativeOfMetric { metric: "data_total" }.
let candidates = all_candidates("topk(10, sum by (job) (increase(data_total[5m])))", &target);
// The data workload declares data_total a counter (or a gauge).
let data = with_metric_types(data(), &[("data_total", MetricType::Counter)]);
// Stage 3: accuracy_check confirms P2's weights from `data`.
let rejected = select_plan(&candidates, &[entry_with(target, repeating(10_000))], &data,
                           PlanningModels::builtin())?.rejected;
// Counter: P2 is not rejected for its weights. It then fails NoAccuracyGuarantee
// under the built-in model, since a top-k estimate carries no guarantee; a
// deployment accuracy model may accept it.
// Gauge: P2 is rejected with CountMinWeightsNotProvenNonNegative, "… (rates or
// increases of `data_total`, which is not declared a counter)".

Before this PR

query over data_total	declared counter	declared gauge
topk(10, sum by (job) (increase(...))), Count-Min-with-heap	passes the weight check	passes the weight check (unsound)
same with rate	passes	passes (unsound)
same with irate	no Count-Min candidate	no Count-Min candidate

After this PR

query over data_total	declared counter	declared gauge
increase, rate or irate, Count-Min-with-heap	passes the weight check, then the accuracy and cost checks	rejected with CountMinWeightsNotProvenNonNegative

Verification

Regression test. plan-selection count_min_over_a_counter_derivative_needs_a_declared_counter covers increase, rate and irate, each with data_total declared a counter and a gauge. On main it failed: over a gauge, increase's Count-Min candidate was rejected only with NoAccuracyGuarantee, not for its weights. It passes now.

Commands run at the head commit. Each was run with CARGO_PROFILE_DEV_DEBUG=0 CARGO_PROFILE_TEST_DEBUG=0 CARGO_INCREMENTAL=0, which only reduces disk use. All passed.

• cargo test --workspace --locked: exit 0, 150 test binaries, 0 failures.
• cargo fmt --all -- --check
• cargo clippy -p asap-types -p asap-executor -p asap-plan-selection -p asap-logical-asap-optimization -p asap-frontend-promql --all-targets --all-features --locked -- -D warnings
• cargo check -p asap-planner --tests --locked

Not applicable: performance measurements (no hot-path change).

Limitations.

• Executing a weighted heap top-k end to end is still blocked by the top-k output shape (fix(ir): top-k sketch readouts return selected rows; Example 1 expectations follow planner output #579; weighted_topk_binding is ignored for that reason). This PR changes which candidates pass Stage 3's weight check, not execution.
• The review was not independent: the author reviewed the diff.

Closes #773