fix(query-engine): counter-reset correction + Prometheus extrapolation for rate()/increase() (#476)

[zaoxing]

Summary

Fixes #476. ASAPQuery's accelerated rate() / increase() returned wrong values because the increase precompute did no counter-reset correction. The accumulator stored only the first and last sample per series and computed last − start, so any counter reset inside the window was silently lost — e.g. a stream 10 → 100 → 5 → 30 returned 20 instead of the correct 120.

This PR brings the precompute path to full Prometheus rate/increase semantics across both ingestion engines (native precompute and Arroyo):

• Counter-reset correction — accumulated per-sample at ingest (Prometheus counterCorrection), since only the window endpoints are retained and a reset between them is unrecoverable at query time.
• Boundary extrapolation — the result is extrapolated to the range-vector boundaries, and rate divides by the range duration (not the sampled interval) — matching extrapolatedRate in Prometheus promql/functions.go.

The extrapolated() implementation was verified line-for-line against the current Prometheus source (float counter path), including the exact clamp order (threshold clamp → counter zero-point clamp) and the factor = 1.0 fallback when the sampled interval is zero. The histogram, anchored/smoothed, and created-timestamp branches are intentionally not reproduced (not applicable to this codebase).

Changes

• IncreaseAccumulator (increase_accumulator.rs): added counter_reset_correction and sample_count fields; reset detection in update(); Prometheus extrapolation in query() with a reset-corrected fallback for non-PromQL callers (SQL/Elastic); boundary-aware merge_pair() used by both merge paths; serde (JSON + binary) carries the new fields with tolerant legacy decode.
• MultipleIncreaseAccumulator (multiple_increase_accumulator.rs): extended the MeasurementData MessagePack wire struct with #[serde(default)] fields (back-compatible), fixed the manual byte-length math, routed per-key merges through merge_pair, and forwarded query_kwargs.
• Arroyo UDF templates (templates/udfs/multipleincrease_.rs, templates/hashed_key_udfs/multipleincrease_.rs): same reset/count logic, sorting per-key samples by timestamp first.
• Query plumbing: range-vector boundaries threaded into query_kwargs for the instant path (promql.rs) and per-step in the range path (mod.rs); shared key constants in enums.rs.

Testing

• New unit tests: single/multiple resets, merge boundary correction (order-independent), extrapolation parity (incl. two cases pinning the exact Prometheus clamp order), fallback, serde round-trip + legacy decode, Arroyo round-trip + legacy blob.
• cargo test -p query_engine_rust --lib — 532 passed, 0 failed; clippy clean; full workspace builds.
• Arroyo UDF bodies compiled and behavior-checked offline (the live validate_udfs.py needs a running Arroyo server).

Follow-up

No automated numeric parity test against a real Prometheus exists yet. Recommend a manual check on a resetting counter via the asap-quickstart stack (Grafana shows ASAPQuery vs Prometheus side-by-side) before relying on it in production.

It is worth additional checking if this implementation is capatiable with VictoriaMetrics extrapolation.

[milindsrivastava1997]

This updates only the range query execution path, but not the instant query execution path. Note that Prometheus "instant" queries also operate on a range of data. Range queries are simply a way to express multiple instant queries executing at different timesteps.

When you send a query request to Prometheus, it can be an instant query, evaluated at one point in time, or a range query at equally-spaced steps between a start and an end time. PromQL works exactly the same in each case; the range query is just like an instant query run multiple times at different timestamps.

https://prometheus.io/docs/prometheus/latest/querying/basics/

[milindsrivastava1997]

There are actually three execution paths for rate()/increase(), not two:

1. Single-metric instant (handle_query_promql → execute_context → execute_query_pipeline) — gets RANGE_START_MS_KWARG/RANGE_END_MS_KWARG via build_query_kwargs_promql's new Statistic::Rate | Statistic::Increase arm (promql.rs:144-101).
2. Range query, single or binary (handle_range_query_promql / handle_binary_expr_range_promql → execute_range_query_pipeline) — gets the new step_kwargs with per-step RANGE_START_MS_KWARG/RANGE_END_MS_KWARG (mod.rs:1511-1550). ✅ Fixed.
3. Binary-arithmetic instant (handle_binary_expr_promql → build_arm_logical_plan → DataFusion physical plan → SummaryInferExec) — this is a completely separate code path, and SummaryInferExec::infer_op_to_kwargs (engines/physical/summary_infer_exec.rs:167-187) only
   builds kwargs for Quantile/Median/TopK; everything else, including InferOperation::ExtractIncrease/ExtractRate, falls through to _ => None. That None propagates to IncreaseAccumulator::query(statistic, None), which hits the fallback branch — reset-corrected but not
   extrapolated.

[zaoxing]

Using my phone, will double check. But I don’t think counter-reset problem happens on a single-metric instant query as all points are in one timestamp? Oh, with some lookback windows.

[milindsrivastava1997]

@zaoxing This is confusing about Prometheus but instant query != instant vector.
Instant vector - all data points at 1 timestamp
Range vector - data points across multiple timestamp
Instant query - a query evaluated at 1 timestamp, but may span data points across multiple timestamps. For e.g. avg_over_time(data[5m]) @ t=10:00:00 is an instant query evaluated at 10am that consider data from 9:55am to 10am
Range query - an instant query that is evaluated multiple times, at multiple timestamps.

[zaoxing]

Yep, will revision.

[milindsrivastava1997]

Why 1 and not 0?

[milindsrivastava1997]

Claude says:

1 is correct for new()'s actual contract — it's only ever called (in production code) with starting == last_seen, i.e. exactly one real sample has been observed, so 0 would be wrong. I checked every production call site (accumulator_factory.rs) and they all respect
this.

But the question exposes that the contract is unenforced: new()'s signature happily accepts two distinct starting/last_seen values and will still silently record sample_count: 1. Several existing tests already construct accumulators this way (e.g.
plan_execution_tests.rs:311-312, store_correctness_tests.rs:845, accumulator_serde_tests.rs:115/119) — treating new() as a generic "two-endpoint" constructor. Luckily I checked: the ones that would actually break (test_old_vs_new_increase) are already #[ignore]d, and
the others never call .query()/.extrapolated(), so nothing currently fails — but it's a latent trap. If any future caller uses new() the same way with two distinct values and then calls rate()/increase() with range kwargs, extrapolated() will wrongly return
sample_count < 2's "insufficient samples" error path is actually fine there (1 < 2 → None, correctly conservative) — the worse case is average_duration_between_samples = sampled_interval / 0 if someone passes sample_count explicitly wrong via new_full, or just
generally undercounts real sample density for extrapolation math on synthetic fixtures.

[zaoxing]

fix(query-engine): Prometheus counter-reset correction + extrapolation for rate()/increase() across all execution paths

Closes #476. Completes the work started in #492.

Problem

ASAP's accelerated rate() / increase() did not reproduce Prometheus counter
semantics. Two gaps:

1. No counter-reset correction. A counter restart within a window (a sample
   lower than its predecessor) was not corrected, so the increase was
   under-counted whenever the counter reset.
2. No extrapolation. Prometheus extrapolates the observed increase to the
   range-vector boundaries (extrapolatedRate in promql/functions.go); ASAP
   returned the raw endpoint difference, so results diverged from Prometheus —
   especially for short/edge-aligned windows and counters near zero.

There are three execution paths for rate()/increase(), and the fix has to
land in all of them:

#	Path	Entry point	Mechanism
1	Instant, single expression	handle_query_promql("increase(m[w])")	in-memory pipeline
2	Range query	handle_range_query_promql(...)	pipeline, per-step boundaries
3	Instant, binary arithmetic	handle_query_promql("increase(m[w]) * k")	DataFusion SummaryInferExec

Path #3 is a separate DataFusion code path: the range boundaries were dropped at
the logical→physical boundary because the InferOperation enum couldn't carry
them, so the accumulator fell back to the non-extrapolated branch.

What changed

Accumulator (counter-reset correction + extrapolation)

• IncreaseAccumulator now tracks counter_reset_correction (Prometheus
  counterCorrection, accumulated per-sample in update) and sample_count.
• extrapolated(is_rate, range_start_ms, range_end_ms) reproduces Prometheus
  extrapolatedRate step-for-step (threshold clamp → counter zero-point clamp,
  in that order; factor = 1.0 fallback for a zero sampled interval).
• Boundary-aware merge_pair adds a seam correction when a reset straddles two
  tumbling windows, so windowed ingest + merge equals a single window over the
  full series.
• Same treatment for the multi-population MultipleIncreaseAccumulator.

Path #1 & #2 (pipeline)

• build_query_kwargs_promql supplies RANGE_START_MS_KWARG / RANGE_END_MS_KWARG
  for the Rate | Increase case (instant); the range pipeline refreshes them
  per output step.

Path #3 (binary-arithmetic instant / DataFusion) — the remaining gap

• InferOperation::ExtractIncrease / ExtractRate now carry an optional
  (range_start_ms, range_end_ms) payload — exactly how Quantile/TopK carry
  their parameters — so the boundaries survive the logical→physical boundary.
• plan_builder populates the bounds from the query kwargs;
  SummaryInferExec::infer_op_to_kwargs re-materializes them so both single- and
  multi-population accumulators extrapolate. None payload → reset-corrected
  fallback (non-range callers, e.g. SQL/Elastic), so no behavior change there.

Single-population Increase on the DataFusion path (arroyo serde)

• Previously SketchType::Increase had no arroyo/MessagePack deserializer, so a
  single-pop increase/rate arm inside a binary expression silently fell back to
  Prometheus (correct but not accelerated). Added
  IncreaseAccumulator::{serialize,deserialize}_from_bytes_arroyo and wired
  Increase into serialize_accumulator_arroyo, deserialize_accumulator, and
  deserialize_single_subpopulation. Single-pop increase/rate in binary
  expressions is now accelerated and extrapolated.

Backward compatibility

• Native and arroyo/MessagePack decoders tolerate legacy blobs written before the
  reset/sample-count fields existed (trailing fields default; an absent
  sample_count decodes as 0 → treated as 2 so extrapolation stays defined).
• The single-pop arroyo wire form is byte-compatible with one
  MultipleIncrease entry and forward-compatible with a future single-increase
  ingest UDF.

Correctness verification

Beyond unit tests, this PR adds cross-path and synthetic differential
coverage that checks ASAP against an independent reimplementation of
Prometheus extrapolatedRate operating on the raw (timestamp, value) sample
stream (not on ASAP's reduced accumulator state). The reference is anchored to
hand-computed Prometheus values, then used as the oracle across generated series.

• All three paths verified end-to-end through the real dispatch, with
  Prometheus-parity values (e.g. instant/range/binary all agree; rate*2
  applies arithmetic on the extrapolated rate).
• Synthetic suite: 5,000+ generated cases (reset patterns, even & uneven
  spacing, 2–24 samples, 4 range alignments, rate & increase); merge-path cases
  (random tumbling-window splits → merge → equals full-series reference);
  multi-population per-key cases; named edge cases (reset on first/last,
  consecutive resets, zero-start zero-clamp, flat counter, 2-sample minimum).
• Sensitivity-checked: temporarily disabling reset correction breaks 4 of 6
  synthetic tests with precise diagnostics, confirming the differential tests
  aren't vacuous (reverted).

Full suite: 550 lib tests + integration binaries pass, 0 failures;
cargo fmt --check clean; workspace builds.

Out of scope / follow-ups

• SketchType::MinMax still returns NotImplemented on the DataFusion serde
  path (separate accumulator, unrelated to counters).
• ASAP retains only window endpoints + correction + count; since Prometheus's own
  extrapolatedRate derives its interval estimate from just (first_ts, last_ts, N), this matches Prometheus exactly for arbitrary spacing. The per-sample
  created-timestamp / smoothing branches Prometheus uses for very recent counters
  are not reproduced (ASAP does not track created timestamps).