feat(query): compose rate-over-Sum into ExactAgg(Sum) candidate (multinode topk dispatch)

[zzylol]

Summary

Closes the rate-over-Sum gap PR #291 documented as the remaining blocker for the multinode demo's topk(5, sum by (zone) (rate(http_requests_total[5m]))) query. The three target shapes now land on the ExactAgg(Sum) sids the control plane mints for counter metrics (no more falling over to the archive's NoData stub):

• rate(http_requests_total[5m]) → 4 zones × per-zone rate via data_source: asap_query.
• sum by (zone) (rate(http_requests_total[5m])) → same 4 zones via data_source: asap_query.
• topk(5, sum by (zone) (rate(http_requests_total[5m]))) → all 4 zones (k≥n) descending via data_source: asap_query.

What changed

Three data-plane changes (control plane untouched, per scope constraint):

1. SketchReducer::evaluate_exact_agg_rate (sister of feat(query): dispatch sum by (...) PromQL to ExactAgg(Sum) sids (closes Option-B downstream) #291's evaluate_exact_agg) — folds EVERY in-window sub-window accumulator per group into ONE merged accumulator, reads Statistic::Sum, then divides by range_seconds to produce events-per-second. Emits one sample per group, timestamped at t1_ms (instant-rate semantics).

2. ASAPQueryEngine compose rule — when a candidate is ExactAgg(Sum-family) with range_seconds > 0 AND the raw PromQL contains a rate(...) / irate(...) call (walked via promql_parser), dispatch to evaluate_exact_agg_rate instead of the plain per-window evaluate_exact_agg. The PromQL walker is the disambiguator the analyzer can't provide on its own (sum by (zone) (rate(metric[5m])) and sum by (zone) (sum_over_time(metric[5m])) arrive at the engine with identical (function, range_seconds, capability) and need different math).

3. ASAPQueryEngine::try_topk_over_rate_fallback — the control-plane analyzer rewrites topk(K, sum by (gbk) (rate(metric[r]))) into FrequencyTopk + FrequencyEstimate candidates (PromQL topk lowers to AggIntent::TopK; the optimizer's CMS-topk binder pins it to a frequency family). Those don't match ExactAgg(Sum) sids — the analyzer-driven path would CapabilityMiss. The fallback lifts (K, metric, group_by_keys, range_seconds, is_topk) from the raw PromQL, finds ExactAgg(Sum) sids via instances_matching, runs evaluate_exact_agg_rate, then post-applies the descending top-K (or ascending bottom-K) slice in-engine.

Plus a routing-table override: resolve_metric_storage flips Gorilla → SketchStore when the table sends a RatePostHoc / Topk query to the archive AND the sketch index has ExactAgg(Sum) sids for the metric. The control plane's build_routing_entry unconditionally puts rate_post_hoc on the archive's claim list (predates ExactAgg sids); flipping it back at request time avoids the control-plane churn.

irate falls out for free (the PromQL walker matches both rate and irate).

Test coverage

• 7 new sketch_reducer tests pin evaluate_exact_agg_rate: divide-by-range, per-group across zones, rack→zone subgroup collapse, no-gbk per-sid preservation, zero-range and MinMax unsupported-capability, empty-window NoData.
• 4 new asap_query_engine end-to-end tests pin the engine dispatch: rate(...) direct, sum by (zone) (rate(...)), topk(5, ...) fallback (K≥n), topk(2, ...) fallback truncation.

cargo test -p data_plane --lib: 743 passed (732 baseline + 7 reducer + 4 engine), 0 failed, 2 ignored. Control plane untouched (738 passed, unchanged).

Smoke-test verification

$ curl --data-urlencode 'query=rate(http_requests_total[5m])' http://localhost:19091/api/v1/query
{"data":{"result":[
  {"metric":{"zone":"z0"},"value":[...,"226.74"]},
  {"metric":{"zone":"z1"},"value":[...,"226.74"]},
  {"metric":{"zone":"z2"},"value":[...,"209.28333..."]},
  {"metric":{"zone":"z3"},"value":[...,"209.26333..."]}
]},"resultType":"vector","infos":["data_source: asap_query"],"status":"success"}

$ curl --data-urlencode 'query=sum by (zone) (rate(http_requests_total[5m]))' http://localhost:19091/api/v1/query
{"data":{"result":[
  {"metric":{"zone":"z0"},"value":[...,"226.74"]},
  ...same 4 zones...
]},"infos":["data_source: asap_query"],...}

$ curl --data-urlencode 'query=topk(5, sum by (zone) (rate(http_requests_total[5m])))' http://localhost:19091/api/v1/query
{"data":{"result":[
  ...all 4 zones, descending by rate...
]},"infos":["data_source: asap_query"],...}

# REGRESSION baselines — both still green:
$ curl --data-urlencode 'query=sum by (zone) (http_requests_total)' http://localhost:19091/api/v1/query
{"data":{"result":[
  {"metric":{"zone":"z0"},"value":[...,"56416"]},
  ...4 zones...
]},"infos":["data_source: asap_query"],...}

$ curl --data-urlencode 'query=quantile_over_time(0.99, http_requests_total_latency_ms[5m])' http://localhost:19091/api/v1/query
{"data":{"result":[
  4 per-zone DDSketch quantile values
]},"infos":["data_source: asap_query"],...}

Pre-fix all three new queries returned {"result":[],"resultType":"vector","infos":["data_source: thanos_query"]} (NoData stub on the archive slot).

Test plan

• cargo build -p data_plane: clean
• cargo test -p data_plane --lib: 743 passed, 0 failed, 2 ignored
• Smoke test: rate(...), sum by (zone) (rate(...)), topk(5, sum by (zone) (rate(...))), topk(2, ...) all return correct values via data_source: asap_query
• Smoke test regression: sum by (zone) (http_requests_total) (PR feat(query): dispatch sum by (...) PromQL to ExactAgg(Sum) sids (closes Option-B downstream) #291) and quantile_over_time(0.99, http_requests_total_latency_ms[5m]) (PR refactor(control_plane): unify Replanner through typed cumulative emit; retire generate_streaming_config_yaml (Option B) #290) still work

🤖 Generated with Claude Code