From 6376a8f673dc4a844a709db06572bc5cde490724 Mon Sep 17 00:00:00 2001 From: Zeying Zhu Date: Fri, 17 Apr 2026 16:46:59 -0400 Subject: [PATCH] =?UTF-8?q?feat(sketch-db):=20Phase=203b=20=E2=80=94=20cro?= =?UTF-8?q?ss-schema=20result=20combiner=20+=20engine=20registry=20wiring?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Lands the §7 combiner primitive used by the query path to stitch per-segment scalars across a schema change, plus the minimal SimpleEngine plumbing needed to reach a live SchemaRegistry. Does **not** yet re-route the existing dispatch through the combiner — that is Phase 3b-2, a pure-plumbing follow-up PR. ## New: `src/engines/timeline_dispatch.rs` * `CombinedResult { Full(f64) | Partial { covered, missing } }` — the result type that surfaces the schema-change failure mode explicitly instead of silently returning "the wrong answer" when a statistic can't span a schema boundary. * `SegmentValue { segment, value }` — per-segment scalar input. * `combine_statistic(stat, &segments, &unresolved) -> CombinedResult` implementing §7.3 combinability: - Count / Sum: additive. - Min / Max: pointwise. - Cardinality / Increase / Rate / Quantile / Topk: non-combinable at the scalar level — caller must merge underlying sketches or fall back to the exact DB. * 12 unit tests covering each statistic, empty-input, single-segment, and unresolved-with-covered scenarios. ## SimpleEngine wiring * Adds `schema_registry: Arc` field, defaulted to `SchemaRegistry::empty()` so the ~35 existing `SimpleEngine::new*` call-sites keep compiling unchanged. * `with_schema_registry(Arc) -> Self` builder alongside `with_controller_client`. * `timeline_for_query(metric, t1_ms, t2_ms) -> Vec` thin delegate so the engine's own query path (and Phase 3b-2 tests) don't need to reach into the store module to build a timeline. ## main.rs plumbing Defers the `Arc::new(engine)` wrap until after the precompute engine is constructed, so the engine can share `precompute_ingest_state.schemas` — the same `Arc` the ingest path is already reconciling and the HTTP swap handler is driving event-driven (Phase 2b). When precompute isn't enabled the engine keeps its default empty registry and `timeline_for_query` returns no segments, matching pre-Phase-3 behaviour. 559 lib tests pass (up from 547: +12 combiner tests); clippy/fmt clean. No changes to existing engine tests or call-sites. Co-Authored-By: Claude Opus 4.7 (1M context) --- asap-query-engine/src/engines/mod.rs | 2 + .../src/engines/simple_engine.rs | 42 +++ .../src/engines/timeline_dispatch.rs | 350 ++++++++++++++++++ asap-query-engine/src/main.rs | 17 +- 4 files changed, 409 insertions(+), 2 deletions(-) create mode 100644 asap-query-engine/src/engines/timeline_dispatch.rs diff --git a/asap-query-engine/src/engines/mod.rs b/asap-query-engine/src/engines/mod.rs index 2bc64fb2..06ed0db1 100644 --- a/asap-query-engine/src/engines/mod.rs +++ b/asap-query-engine/src/engines/mod.rs @@ -2,8 +2,10 @@ pub mod logical; pub mod physical; pub mod query_result; pub mod simple_engine; +pub mod timeline_dispatch; pub mod window_merger; pub use query_result::{InstantVector, QueryResult, RangeVector, RangeVectorElement, Sample}; pub use simple_engine::SimpleEngine; +pub use timeline_dispatch::{combine_statistic, CombinedResult}; pub use window_merger::{create_window_merger, NaiveMerger, WindowMerger}; diff --git a/asap-query-engine/src/engines/simple_engine.rs b/asap-query-engine/src/engines/simple_engine.rs index 50405111..e931b57f 100644 --- a/asap-query-engine/src/engines/simple_engine.rs +++ b/asap-query-engine/src/engines/simple_engine.rs @@ -158,6 +158,17 @@ pub struct SimpleEngine { /// misses fall through to the §5.2 fallback silently, matching /// pre-PR-G behavior. Set via `with_controller_client`. controller_client: Option>, + /// Per-`agg_id` schema registry used for §7 schema-timeline + /// dispatch (`docs/design-sketch-db.md`). In Phase 3b this is + /// stored for future query-path wiring; the combiner lives in + /// [`crate::engines::timeline_dispatch`] and the lookup primitive + /// is already exposed on [`crate::stores::sketch_db::SchemaRegistry`]. + /// + /// Defaults to an empty registry so pre-Phase-3 call-sites keep + /// compiling. Production wire-up (`main.rs`) uses + /// [`Self::with_schema_registry`] to share the same registry the + /// ingest path is reconciling. + schema_registry: Arc, } impl SimpleEngine { @@ -338,6 +349,7 @@ impl SimpleEngine { controller_patterns, query_language, controller_client: None, + schema_registry: Arc::new(crate::stores::sketch_db::SchemaRegistry::empty()), } } @@ -371,6 +383,36 @@ impl SimpleEngine { self } + /// Attach the shared `SchemaRegistry` the ingest path is + /// reconciling so queries can resolve the §7 schema timeline for + /// a metric. Typically called from `main.rs` with the same + /// `Arc` held by `IngestState::schemas` and the + /// HTTP streaming-config swap handler so all three observe the + /// same lifecycle transitions. + pub fn with_schema_registry( + mut self, + registry: Arc, + ) -> Self { + self.schema_registry = registry; + self + } + + /// Resolve the §7 schema timeline for a metric over a query range. + /// Thin delegate to `SchemaRegistry::timeline_for_metric` so the + /// engine's own query-path code does not need to reach into the + /// store module to build a timeline (and so tests for the + /// dispatch wiring — Phase 3b-2 — can mock by swapping the + /// registry rather than monkey-patching the engine). + pub fn timeline_for_query( + &self, + metric: &str, + t1_ms: u64, + t2_ms: u64, + ) -> Vec { + self.schema_registry + .timeline_for_metric(metric, t1_ms, t2_ms) + } + /// Look up a compatible aggregation for the given requirements, /// and if none exists, fire a capability-miss notification to /// the controller (fire-and-forget, does not block the query). diff --git a/asap-query-engine/src/engines/timeline_dispatch.rs b/asap-query-engine/src/engines/timeline_dispatch.rs new file mode 100644 index 00000000..082a69b2 --- /dev/null +++ b/asap-query-engine/src/engines/timeline_dispatch.rs @@ -0,0 +1,350 @@ +//! Cross-schema result combination for the §7 schema-timeline query +//! dispatch ([`design-sketch-db.md`](../../../../docs/design-sketch-db.md)). +//! +//! When a metric-range query spans a reconfigure boundary, the +//! `SchemaRegistry::timeline_for_metric` call returns multiple +//! [`TimelineSegment`]s each owned by a distinct `agg_id`. The query +//! engine evaluates the statistic **per segment** against its owning +//! aggregation's precompute, then hands the per-segment scalars to +//! [`combine_statistic`] here to stitch them into a single answer. +//! +//! ## What this module is — and is not +//! +//! This is a pure combiner over already-evaluated scalar results. It +//! does NOT: +//! * touch the store or accumulators, +//! * know how per-segment evaluation works (the engine does that), +//! * decide fallback policy for `Purged` segments (the caller does +//! that before calling here). +//! +//! Phase 3b-2 will wire this into +//! `SimpleEngine::build_query_execution_context_promql`; this PR +//! (Phase 3b) lands the primitive and its correctness tests so that +//! wiring is a pure plumbing change. +//! +//! ## Statistic combinability +//! +//! See §7.3 of the design doc. Summarised: +//! +//! | Statistic | Combinable across sketch types? | +//! |---|---| +//! | `Count`, `Sum` | Yes — sum. | +//! | `Min`, `Max` | Yes — pointwise min / max. | +//! | `Cardinality` (HLL) | Yes in principle (HLL OR merge on the sketch itself), but at the **scalar** level — which is what this module receives — distinct counts from different HLL parameterisations cannot be OR-merged. Treated as **non-combinable** here. | +//! | `Increase`, `Rate` | Non-combinable at the scalar level — they depend on endpoint samples; stitching needs the raw counters, not their deltas. | +//! | `Quantile`, `Topk` | Non-combinable at the scalar level — stitching requires merging the underlying KLL / CMS sketches, which may differ in parameters across schemas. | +//! +//! Non-combinable statistics return [`CombinedResult::Partial`] +//! carrying whatever combinable prefix we could compute plus the +//! list of segments that couldn't contribute. The caller decides +//! how to render that (warning, fall-through to exact DB, or +//! error). + +use promql_utilities::query_logics::enums::Statistic; + +use crate::stores::sketch_db::TimelineSegment; + +/// Result of combining per-segment scalars for a single statistic. +/// +/// `Full(value)` means every segment contributed and the combination +/// is semantically equivalent to a single-schema evaluation over the +/// entire range. `Partial` surfaces the failure mode explicitly so +/// the user knows they are looking at a schema-change artifact. +#[derive(Debug, Clone, PartialEq)] +pub enum CombinedResult { + /// All segments combined cleanly. Value is the stitched result. + Full(f64), + /// The statistic isn't combinable across schema boundaries, or + /// the caller flagged one or more segments as uncomputable + /// (e.g. a `TimelineCoverage::Purged` segment whose data no + /// longer lives in the sketch store). + /// + /// `covered` is the combined result over the segments we could + /// evaluate; `missing` lists the segments we couldn't. Note that + /// `covered` may be meaningless for the user (e.g. a Quantile + /// over only part of the range is not "the p99 of the range"), + /// so it's up to the caller to decide whether to display it. + Partial { + covered: Option, + missing: Vec, + }, +} + +/// Input to [`combine_statistic`]: one per-segment scalar plus the +/// segment it came from (for provenance in the `Partial` case). +#[derive(Debug, Clone, PartialEq)] +pub struct SegmentValue { + pub segment: TimelineSegment, + pub value: f64, +} + +/// Combine per-segment scalars for a single statistic across a +/// multi-schema timeline. +/// +/// Contract: +/// +/// * `segments` is the output of the engine's per-segment evaluator, +/// one `SegmentValue` per [`TimelineSegment`] returned by +/// `timeline_for_metric`. An empty input means no schema covers the +/// query range; returns `Full(0.0)` for `Count` / `Sum` (zero is +/// the identity), and `Partial { covered: None, missing: [] }` for +/// everything else (no sensible default). +/// * `unresolved` is the list of segments the engine could not +/// evaluate (e.g. `Purged` coverage, or a capability miss). They +/// are folded directly into the `Partial` output's `missing` list. +pub fn combine_statistic( + statistic: Statistic, + segments: &[SegmentValue], + unresolved: &[TimelineSegment], +) -> CombinedResult { + // Fast path: nothing to combine and nothing missing. + if segments.is_empty() && unresolved.is_empty() { + return match statistic { + Statistic::Count | Statistic::Sum => CombinedResult::Full(0.0), + _ => CombinedResult::Partial { + covered: None, + missing: Vec::new(), + }, + }; + } + + // If the caller handed us any unresolved segments, the result is + // Partial regardless of the statistic. We still compute the + // best-effort `covered` so callers that want to show it can. + let has_unresolved = !unresolved.is_empty(); + + let covered = match statistic { + Statistic::Count | Statistic::Sum => { + // Additive: zero identity, so even with zero segments + // the value is 0.0. + Some(segments.iter().map(|s| s.value).sum::()) + } + Statistic::Min => segments + .iter() + .map(|s| s.value) + .fold(None, |acc, v| Some(acc.map_or(v, |a: f64| a.min(v)))), + Statistic::Max => segments + .iter() + .map(|s| s.value) + .fold(None, |acc, v| Some(acc.map_or(v, |a: f64| a.max(v)))), + // Non-combinable at the scalar level — see module doc. + Statistic::Cardinality + | Statistic::Increase + | Statistic::Rate + | Statistic::Quantile + | Statistic::Topk => None, + }; + + let combinable = matches!( + statistic, + Statistic::Count | Statistic::Sum | Statistic::Min | Statistic::Max + ); + + if combinable && !has_unresolved { + // `covered` is always `Some` for combinable statistics with + // at least one input (Count/Sum guarantee it even at zero). + match covered { + Some(v) => CombinedResult::Full(v), + None => CombinedResult::Partial { + covered: None, + missing: Vec::new(), + }, + } + } else { + CombinedResult::Partial { + covered, + missing: unresolved.to_vec(), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::stores::sketch_db::{AggStatus, TimelineCoverage}; + + fn seg(agg_id: u64, start: u64, end: u64, status: AggStatus) -> TimelineSegment { + TimelineSegment { + agg_id, + start_ms: start, + end_ms: end, + status, + coverage: match status { + AggStatus::Expired => TimelineCoverage::Purged, + _ => TimelineCoverage::Sketch, + }, + } + } + + fn sv(agg_id: u64, start: u64, end: u64, status: AggStatus, v: f64) -> SegmentValue { + SegmentValue { + segment: seg(agg_id, start, end, status), + value: v, + } + } + + #[test] + fn sum_is_additive_across_segments() { + let vs = vec![ + sv(1, 0, 100, AggStatus::Retired, 10.0), + sv(2, 100, 200, AggStatus::Active, 15.0), + ]; + assert_eq!( + combine_statistic(Statistic::Sum, &vs, &[]), + CombinedResult::Full(25.0), + ); + } + + #[test] + fn count_is_additive_across_segments() { + let vs = vec![ + sv(1, 0, 100, AggStatus::Retired, 7.0), + sv(2, 100, 200, AggStatus::Active, 3.0), + ]; + assert_eq!( + combine_statistic(Statistic::Count, &vs, &[]), + CombinedResult::Full(10.0), + ); + } + + #[test] + fn min_takes_pointwise_min() { + let vs = vec![ + sv(1, 0, 100, AggStatus::Retired, 5.0), + sv(2, 100, 200, AggStatus::Active, 2.5), + ]; + assert_eq!( + combine_statistic(Statistic::Min, &vs, &[]), + CombinedResult::Full(2.5), + ); + } + + #[test] + fn max_takes_pointwise_max() { + let vs = vec![ + sv(1, 0, 100, AggStatus::Retired, 5.0), + sv(2, 100, 200, AggStatus::Active, 9.25), + ]; + assert_eq!( + combine_statistic(Statistic::Max, &vs, &[]), + CombinedResult::Full(9.25), + ); + } + + #[test] + fn quantile_across_schemas_is_partial() { + let vs = vec![ + sv(1, 0, 100, AggStatus::Retired, 0.95), + sv(2, 100, 200, AggStatus::Active, 0.97), + ]; + match combine_statistic(Statistic::Quantile, &vs, &[]) { + CombinedResult::Partial { covered, missing } => { + assert!(covered.is_none()); + assert!(missing.is_empty()); + } + other => panic!("expected Partial, got {other:?}"), + } + } + + #[test] + fn topk_across_schemas_is_partial() { + let vs = vec![sv(1, 0, 100, AggStatus::Retired, 42.0)]; + assert!(matches!( + combine_statistic(Statistic::Topk, &vs, &[]), + CombinedResult::Partial { .. } + )); + } + + #[test] + fn cardinality_across_schemas_is_partial() { + let vs = vec![ + sv(1, 0, 100, AggStatus::Retired, 1_000.0), + sv(2, 100, 200, AggStatus::Active, 1_500.0), + ]; + assert!(matches!( + combine_statistic(Statistic::Cardinality, &vs, &[]), + CombinedResult::Partial { .. } + )); + } + + #[test] + fn additive_with_unresolved_segment_is_partial_with_covered_set() { + let vs = vec![sv(1, 0, 100, AggStatus::Retired, 10.0)]; + let missing = vec![seg(2, 100, 200, AggStatus::Expired)]; + match combine_statistic(Statistic::Sum, &vs, &missing) { + CombinedResult::Partial { + covered, + missing: m, + } => { + assert_eq!(covered, Some(10.0)); + assert_eq!(m.len(), 1); + assert_eq!(m[0].agg_id, 2); + } + other => panic!("expected Partial with covered=Some(10.0), got {other:?}"), + } + } + + #[test] + fn empty_input_yields_additive_zero_for_count_and_sum() { + assert_eq!( + combine_statistic(Statistic::Count, &[], &[]), + CombinedResult::Full(0.0), + ); + assert_eq!( + combine_statistic(Statistic::Sum, &[], &[]), + CombinedResult::Full(0.0), + ); + } + + #[test] + fn empty_input_yields_partial_none_for_non_additive() { + // Min / Max / Quantile / Topk / Cardinality / Increase / Rate + // all lack an identity element, so an empty range produces + // Partial { covered: None } rather than a misleading 0.0. + for stat in [ + Statistic::Min, + Statistic::Max, + Statistic::Quantile, + Statistic::Topk, + Statistic::Cardinality, + Statistic::Increase, + Statistic::Rate, + ] { + match combine_statistic(stat, &[], &[]) { + CombinedResult::Partial { covered, missing } => { + assert!( + covered.is_none() && missing.is_empty(), + "{stat:?} should yield empty Partial", + ); + } + other => panic!("{stat:?} should be Partial on empty input, got {other:?}"), + } + } + } + + #[test] + fn single_segment_additive_returns_that_value() { + let vs = vec![sv(1, 0, 100, AggStatus::Active, 42.0)]; + assert_eq!( + combine_statistic(Statistic::Sum, &vs, &[]), + CombinedResult::Full(42.0), + ); + } + + #[test] + fn unresolved_only_no_segments_returns_partial_with_no_covered() { + let missing = vec![seg(1, 0, 100, AggStatus::Expired)]; + match combine_statistic(Statistic::Sum, &[], &missing) { + CombinedResult::Partial { + covered, + missing: m, + } => { + // Sum over zero segments is 0.0 (identity), so covered is + // Some(0.0); the missing list still surfaces the gap. + assert_eq!(covered, Some(0.0)); + assert_eq!(m.len(), 1); + } + other => panic!("expected Partial, got {other:?}"), + } + } +} diff --git a/asap-query-engine/src/main.rs b/asap-query-engine/src/main.rs index d3cbba0e..a8dca2a1 100644 --- a/asap-query-engine/src/main.rs +++ b/asap-query-engine/src/main.rs @@ -372,7 +372,7 @@ async fn main() -> Result<()> { // (PR E phase 2). Without sharing the handle, SimpleEngine // would take a one-time snapshot at construction and ignore // subsequent swaps. - let engine = { + let mut engine = { let mut engine = SimpleEngine::new_with_hot_reload( store.clone(), inference_config, @@ -406,7 +406,10 @@ async fn main() -> Result<()> { (pass --controller-endpoint= to enable)" ); } - Arc::new(engine) + // `Arc::new(engine)` is deferred until after the precompute + // engine is constructed so we can hand the same `SchemaRegistry` + // (§7 timeline source) to both via `with_schema_registry`. + engine }; // Setup Kafka consumer (only when not using precompute engine as the streaming backend) @@ -511,6 +514,16 @@ async fn main() -> Result<()> { (None, None) }; + // Hand the precompute engine's `SchemaRegistry` to the query + // engine so both observe the same §7 timeline (design-sketch-db.md + // §6 / §7). When precompute isn't enabled the engine keeps its + // default empty registry — queries that need the timeline will + // simply see no segments and fall through to the legacy path. + if let Some(ingest_state) = precompute_ingest_state.as_ref() { + engine = engine.with_schema_registry(ingest_state.schemas.clone()); + } + let engine = Arc::new(engine); + // Setup OTLP receiver (after precompute engine so it can share the ingest state) let otel_handle = if args.enable_otel_ingest { let otel_config = OtlpReceiverConfig {