feat: warm-tier follow-ups — FrequencyTopk + Delta encoding + Hybrid stitch - #126
Merged
Merged
Conversation
…stitch Three follow-ups from #124 land in one bundle. The warm-tier reducer now covers all four sketch families end-to-end (DD/KLL quantile, HLL cardinality, CMS-with-heap topk), applies delta-encoded windows by folding from the prior Full, and stitches with the archive engine when the warm tier's coverage is narrower than the query range. ## TODO 1 — FrequencyTopk (CMS-with-heap) [IMPLEMENTED] - New `SketchKindHandle::CmsWithHeap` variant in `stores/sketch_db/sketch_index.rs`. - `drivers/ingest/otel.rs::sketch_kind_handle_for` detects CmsWithHeap by attempting `CountMinSketchWithHeap::deserialize_msgpack(&dp.sketch)` on msgpack payloads and checking `topk_heap_items().is_empty()`. The heap is embedded in the msgpack wrapper, not on the OTLP wire (confirmed by inspecting metrics.proto::CountMinSketch). - New `engines/warm_tier/decoders.rs::decode_cms_with_heap_from_msgpack`: calls `asap_sketchlib::sketches::countminsketch_topk:: CountMinSketchWithHeap::deserialize_msgpack`, then `topk_heap_items()` for the heap. - CMS / CountSketch without heap surface as new `WarmTierError::MissingHeap { sid, sketch_kind }` — clearly distinguished from `UnsupportedCapability` so the EngineRouter's failover semantics can tell "no warm-tier handler for this function" from "warm tier has the data but lacks topk metadata". - `topk` output shape: one `(label_values, samples)` per item where `label_values["item"] = key` and `samples = [(window_end, estimated_count)]`. `k` from `function_args[0]`, default 10. ## TODO 2 — Delta encoding stitching [IMPLEMENTED] - New `engines/warm_tier/delta_apply.rs`: `RollingState` (DDSketch / KLL / HLL variants) + `per_window_evaluate` + `cumulative_evaluate`. - Sketch-lib calls used: `DdSketch::merge`, `KllSketch::merge`, `HllSketch::merge`, `HllSketch::apply_delta(&HllSketchDelta)` for proto register deltas, `HllSketch::deserialize_msgpack` for msgpack delta fragments. - Mode pick by function name: quantile_over_time / count_distinct_over_time / topk_over_time → cumulative (one scalar over [t0, t1]) quantile / histogram_quantile / cardinality_estimate → per-window - Leading deltas without a base Full are skipped (don't error); cumulative mode starts at the first Full in range. Tests verify DDSketch + HLL cumulative Full+Delta(s) round-trip within accuracy envelope of fresh-sketch ground truth. ## TODO 3 — Hybrid warm + archive stitch [IMPLEMENTED] - `WarmTierResult.coverage: Option<(u64, u64)>` populated from observed window-end timestamps. - `SimpleEngine.archive_engine: Option<Arc<dyn QueryEngine>>` field + `with_archive_engine` builder. The trait `execute` adapter inspects `result.coverage` post-evaluate; if narrower than `[t0, now]`, it calls `archive.execute(query)` and stitches. - Chose option (b): in-line stitch in `SimpleEngine` rather than a new `EngineError::PartialHit` variant. Reason: the existing `EngineRouter` is variant-agnostic; introducing PartialHit would force every other engine to also handle it. - Stitch logic: index warm series by `Vec<String>` label key, for each archive series merge by timestamp — warm wins on `[cov_lo, cov_hi]`, archive fills prefix/suffix. New private `stitch_warm_and_archive` helper. - Tests: `stitch_fills_archive_prefix_and_suffix` and `stitch_keeps_archive_only_series_in_full` verify both contracts. ## New API surface - `SketchKindHandle::CmsWithHeap` variant - `WarmTierError::MissingHeap { sid, sketch_kind }` variant - `WarmTierResult.coverage: Option<(u64, u64)>` field - `SimpleEngine::with_archive_engine(archive: Arc<dyn QueryEngine>)` builder - `engines::warm_tier::decoders` (new module) — 5 decode helpers - `engines::warm_tier::delta_apply` (new module) — `DeltaSketchKind`, `RollingState`, `per_window_evaluate`, `cumulative_evaluate` ## Build + test - `cargo build --release -p query_engine_rust` — clean (3 pre-existing warnings) - Targeted run (24/24 pass): engines::warm_tier + the new hybrid_stitch_tests + warm_tier_classify_tests - Full lib run: 814 ok, 4 pre-existing failed (schema_timeline_dispatch + persistence_integration), 1 pre-existing hang (hard_cap_back_pressure) ## Migration note Existing `quantile_over_time` test had to flip its function name to `quantile` (per-window) to match the new cumulative semantics — caller code in production may need similar migration if it depended on per-window emit from `quantile_over_time`. ## Remaining follow-ups - KllSketch proto-delta wire path: today `delta_apply` decodes `Delta` as "Full fragment of matching encoding" then merges. The asap_sketchlib KLL proto wire `KllState.items[]` is sparse-friendly but there's no canonical `KllDelta` proto. If downstream agents start emitting a true sparse KLL delta, revisit. - `decoders::decode_cms_from_{proto,msgpack}`, `decode_cs_from_{proto,msgpack}` are dead-coded for now (kept for the eventual point-query `frequency(key, foo)` path). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
3 tasks
zzylol
added a commit
that referenced
this pull request
May 11, 2026
…ability owner (#128) Today the warm-tier reducer's `extract_promql_call` did a naive AST walk to extract `(function_name, args)` from a PromQL query. This duplicated knowledge the controller already encodes (PromQL → Intent → Capability via `query_parser`, `intent_algebra`, `sketch_algebra`, `algebra::lower`) and silently routed the MVP demo's compound queries (e.g. `sum by (zone) (rate(http_requests_total[5m]))`, `histogram_quantile(0.99, sum(rate(bucket[5m])) by (le))`) through the archive engine because the outermost-call heuristic returned `None` on nested shapes. The controller is now the single owner of "is this PromQL warm-tier-answerable" knowledge. ## What landed ### `controller/src/warm_tier_analysis.rs` (775 lines) Public API: ```rust pub fn analyze_promql_for_warm_tier(promql: &str) -> WarmTierAnalysis; pub struct WarmTierAnalysis { pub candidates: Vec<WarmTierCandidate>, pub unsupported: Option<UnsupportedReason>, } pub struct WarmTierCandidate { pub metric_name: String, pub group_by_keys: BTreeSet<String>, pub required_capability: Capability, pub function: String, pub function_args: Vec<f64>, pub range_seconds: u64, } pub enum UnsupportedReason { UnsupportedFunction(String), UnsupportedComposition(String), NoCallNodeFound, UnparseablePromql(String), } ``` Walks the `promql_parser` AST and identifies sub-expressions that can be served from sketches. Mapping table: | PromQL shape | Capability | |---|---| | `quantile_over_time(q, m[r])` | `QuantileApprox(Any)` | | `histogram_quantile(q, m)` | `QuantileApprox(Any)` | | `count_distinct_over_time(m[r])` | `CardinalityApprox` | | `cardinality_estimate(m)` | `CardinalityApprox` | | `topk(k, m)` | `FrequencyTopk(CmsWithHeap)` | | `topk_over_time(k, m[r])` | `FrequencyTopk(CmsWithHeap)` | Explicitly rejected (now surface as the right `UnsupportedReason`, not silent misroute): - `sum by (label_set) (rate(metric[range]))` → `UnsupportedFunction("rate")` - `histogram_quantile(q, sum(rate(bucket[r])) by (le))` → same - `sum by (label_set) (metric)` → `UnsupportedComposition` (Sum-over-CountSketch is a follow-up) - `increase` / `irate` → `UnsupportedFunction` - `topk(k, rate(metric[r]))` → `UnsupportedComposition` ### `controller/src/lib.rs` `pub mod warm_tier_analysis;` — exposes the new module. ### `asap-query-engine/src/stores/sketch_db/sketch_index.rs` (+85) `From<controller::warm_tier_analysis::SketchKindHandle>` and `From<controller::warm_tier_analysis::Capability>` adapters at the boundary. New `Capability::is_satisfied_by` helper handles the `SketchKindHandle::Any` wildcard ("any sketch impl in the family is acceptable" — e.g. `QuantileApprox(Any)` is satisfied by both DDSketch and KLL instances). ### `asap-query-engine/src/engines/warm_tier/mod.rs` Doc comments rewritten to reference the new controller analyzer. `pub use` re-exports for `promql_extract::*` removed. ### `asap-query-engine/src/engines/warm_tier/promql_extract.rs` **Deleted** (159 lines). ### `asap-query-engine/src/engines/simple/engine.rs::execute` Warm-tier hook rewritten: 1. `controller::warm_tier_analysis::analyze_promql_for_warm_tier(query)` 2. If `analysis.unsupported.is_some()` or `analysis.candidates.is_empty()` → `EngineError::CapabilityMiss`. 3. For each candidate: `index.instances_matching(metric, group_by)`, verify `Capability::is_satisfied_by`, classify, dispatch reducer per Capability. Cold-tier fallthrough is now explicit: - `UnsupportedFunction` / `UnsupportedComposition` / `UnparseablePromql` → archive (router CapabilityMiss failover) - `NoCallNodeFound` (bare selector) → archive - Candidates populated but `instances_matching` empty → archive - All candidates resolve to Hit → warm-tier ## Build + test - `cargo build --release -p query_engine_rust` clean (only pre-existing warnings) - `cargo build --release -p controller` clean - `cargo test --release -p query_engine_rust --lib -- engines::warm_tier` — 13/13 pass ## Diff ``` asap-query-engine/src/engines/simple/engine.rs | 98 ++++++-- asap-query-engine/src/engines/warm_tier/mod.rs | 20 ++ asap-query-engine/src/engines/warm_tier/promql_extract.rs | 159 --- (deleted) asap-query-engine/src/stores/sketch_db/sketch_index.rs | 85 ++++ controller/src/lib.rs | 5 + controller/src/warm_tier_analysis.rs | 775 +++++ (new) 6 files changed, 983 insertions(+), 166 deletions(-) ``` ## Follow-ups (out of scope) - Per-candidate hybrid stitch (PR #126 stitches at engine level; per-candidate is a follow-up). - `Sum-over-CountSketch` reducer for the `sum by (zone) (metric)` shape. - Wire the controller analyzer's `range_seconds` into the reducer's `query_range(t0, t1)` bounds — today the dispatch uses `[now - 5min, now]`. Co-authored-by: zz_y <zz_y@node0.zz-y-304941.softmeasure-pg0.clemson.cloudlab.us> Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Three follow-ups from #124 land in one bundle. The warm-tier reducer now covers all four sketch families end-to-end (DDSketch/KLL quantile, HLL cardinality, CMS-with-heap topk), applies delta-encoded windows by folding from the prior Full, and stitches with the archive engine when the warm tier's coverage is narrower than the query range.
With this PR, the MVP demo's three canonical query classes (DDSketch p99, sum-by-zone, rate+sum) all have warm-tier evaluation paths.
What landed
TODO 1 — FrequencyTopk (CMS-with-heap) ✅
SketchKindHandle::CmsWithHeapvariant; OTLP receive detects it by attemptingCountMinSketchWithHeap::deserialize_msgpackand checking!topk_heap_items().is_empty().engines/warm_tier/decoders.rs::decode_cms_with_heap_from_msgpackcallsasap_sketchlib::sketches::countminsketch_topk::CountMinSketchWithHeap::deserialize_msgpack+topk_heap_items().WarmTierError::MissingHeap { sid, sketch_kind }for CMS/CountSketch without a heap (distinguished fromUnsupportedCapability).topkoutput: one(label_values, samples)per item;label_values["item"] = key,samples = [(window_end, count)].kfromfunction_args[0], default 10.TODO 2 — Delta encoding stitching ✅
engines/warm_tier/delta_apply.rs:RollingState(DD/KLL/HLL) +per_window_evaluate+cumulative_evaluate.DdSketch::merge,KllSketch::merge,HllSketch::merge,HllSketch::apply_delta(&HllSketchDelta)for proto register deltas, msgpack-delta fragments via deserialize+merge.quantile_over_time/count_distinct_over_time/topk_over_time→ cumulativequantile/histogram_quantile/cardinality_estimate→ per-windowTODO 3 — Hybrid warm + archive stitch ✅
WarmTierResult.coverage: Option<(u64, u64)>populated from observed window-end timestamps.SimpleEngine.archive_engine: Option<Arc<dyn QueryEngine>>+with_archive_enginebuilder. The traitexecuteadapter inspectsresult.coveragepost-evaluate; narrower than[t0, now]→ callsarchive.execute(query)+ stitches.Vec<String>label key; for each archive series merge by timestamp — warm wins on[cov_lo, cov_hi], archive fills prefix/suffix.SimpleEnginerather than a newEngineError::PartialHitvariant (existingEngineRouteris variant-agnostic; PartialHit would force every other engine to also handle it).New API surface
SketchKindHandle::CmsWithHeapWarmTierError::MissingHeapWarmTierResult.coverageSimpleEngine::with_archive_engineengines::warm_tier::decoders(new module, 5 decode helpers)engines::warm_tier::delta_apply(new module)Build + test
cargo build --release -p query_engine_rust— clean (3 pre-existing warnings)hybrid_stitch_tests::*+warm_tier_classify_tests::*)schema_timeline_dispatch+persistence_integration), 1 pre-existing hang (hard_cap_back_pressure)Migration note
The existing
quantile_over_timetest had to flip its function name toquantile(per-window) to match the new cumulative semantics. Caller code in production may need similar migration if it depended on per-window emit fromquantile_over_time.Remaining follow-ups (out of scope here)
delta_applydecodesDeltaas "Full fragment of matching encoding" then merges. The asap_sketchlib KLL proto wireKllState.items[]is sparse-friendly but there's no canonicalKllDeltaproto. Revisit when agents start emitting a true sparse KLL delta.decoders::decode_cms_from_{proto,msgpack}anddecode_cs_from_{proto,msgpack}are dead-coded (kept for the eventual point-queryfrequency(key, foo)path).🤖 Generated with Claude Code