diff --git a/asap-query-engine/src/drivers/query/adapters/prometheus_http.rs b/asap-query-engine/src/drivers/query/adapters/prometheus_http.rs index 69e9d811..aa80a103 100644 --- a/asap-query-engine/src/drivers/query/adapters/prometheus_http.rs +++ b/asap-query-engine/src/drivers/query/adapters/prometheus_http.rs @@ -25,10 +25,10 @@ pub struct PrometheusResponse { #[serde(skip_serializing_if = "Option::is_none")] pub error: Option, /// Non-error advisories — maps to Prometheus's top-level - /// `warnings: []` field. Phase 3b-2-b uses this to surface - /// partial results from the §7 schema-timeline dispatcher - /// (query spans a reconfigure boundary with a non-combinable - /// statistic or a Purged segment). + /// `warnings: []` field. The schema-timeline dispatcher uses + /// this to surface partial results when a query spans a + /// reconfigure boundary with a non-combinable statistic or a + /// Purged segment. #[serde(default, skip_serializing_if = "Vec::is_empty")] pub warnings: Vec, } @@ -221,8 +221,8 @@ impl QueryResponseAdapter for PrometheusHttpAdapter { StatusCode::INTERNAL_SERVER_ERROR })?; - // Thread through any Phase 3 timeline-dispatch warnings so - // they land on the top-level `warnings` field, matching + // Thread through any schema-timeline dispatcher warnings + // so they land on the top-level `warnings` field, matching // Prometheus's native API. let warnings = result.query_result.warnings().to_vec(); let response = if warnings.is_empty() { @@ -560,9 +560,9 @@ mod tests { #[test] fn success_response_with_warnings_serialises_the_top_level_field() { - // This is the Phase 3b-2-b contract: a Partial result coming - // out of the §7 timeline dispatcher lands on Prometheus's - // native `warnings: []` field at the top of the response, + // Contract: a Partial result coming out of the §7 + // schema-timeline dispatcher lands on Prometheus's native + // `warnings: []` field at the top of the response, // matching upstream behaviour for warning-carrying queries. let r = PrometheusResponse::success_with_warnings( json!({"resultType": "vector", "result": []}), diff --git a/asap-query-engine/src/engines/query_result.rs b/asap-query-engine/src/engines/query_result.rs index 5a7437c1..6c8fc72d 100644 --- a/asap-query-engine/src/engines/query_result.rs +++ b/asap-query-engine/src/engines/query_result.rs @@ -26,10 +26,10 @@ impl QueryResult { }) } - /// Phase 3b-2-b: construct an instant vector with a non-empty - /// warnings list. Used by the timeline dispatcher when the query - /// spans a reconfigure boundary and one or more segments could - /// not contribute to the answer (non-combinable statistic, purged + /// Construct an instant vector with a non-empty warnings list. + /// Used by the schema-timeline dispatcher when the query spans + /// a reconfigure boundary and one or more segments could not + /// contribute to the answer (non-combinable statistic, purged /// data, or agg_id removed from config mid-flight). Prometheus's /// native JSON surface carries these back to the caller via the /// top-level `warnings` field, matching the upstream contract. @@ -70,7 +70,7 @@ pub struct InstantVector { pub timestamp: u64, /// Non-error advisories attached to this result, surfaced on /// Prometheus's top-level `warnings` field. Empty for - /// single-schema queries; populated by the Phase 3b-2-b timeline + /// single-schema queries; populated by the schema-timeline /// dispatcher when one or more segments produced a /// [`crate::engines::timeline_dispatch::CombinedResult::Partial`] /// (non-combinable statistic, purged coverage, or agg_id diff --git a/asap-query-engine/src/engines/simple_engine.rs b/asap-query-engine/src/engines/simple_engine.rs index 927600d3..e6f7b454 100644 --- a/asap-query-engine/src/engines/simple_engine.rs +++ b/asap-query-engine/src/engines/simple_engine.rs @@ -159,13 +159,15 @@ pub struct SimpleEngine { /// 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 + /// dispatch (`docs/design-sketch-db.md`). The combiner lives in /// [`crate::engines::timeline_dispatch`] and the lookup primitive - /// is already exposed on [`crate::stores::sketch_db::SchemaRegistry`]. + /// is exposed on [`crate::stores::sketch_db::SchemaRegistry`]; + /// the engine consults the registry on every query to resolve + /// which agg_id owns each sub-range of the query's time window. /// - /// Defaults to an empty registry so pre-Phase-3 call-sites keep - /// compiling. Production wire-up (`main.rs`) uses + /// Defaults to an empty registry so call-sites that don't + /// participate in schema-timeline dispatch 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, @@ -400,9 +402,9 @@ impl SimpleEngine { /// 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). + /// store module to build a timeline (and so dispatch-wiring + /// tests can mock by swapping the registry rather than + /// monkey-patching the engine). pub fn timeline_for_query( &self, metric: &str, @@ -2725,12 +2727,12 @@ impl SimpleEngine { } } - // Phase 3b-2-b: try the §7 schema-timeline dispatch first. - // Returns Some only when the query's [t1, t2] range crosses a - // reconfigure boundary AND the statistic is combinable. In - // every other case (single-schema range, non-combinable - // statistic, unparseable query) it returns None and we fall - // through to the default single-agg path below. + // Try the §7 schema-timeline dispatch first. Returns Some + // only when the query's [t1, t2] range crosses a + // reconfigure boundary (i.e. two or more agg_ids own pieces + // of the range). In every other case (single-schema range, + // unparseable query) it returns None and we fall through + // to the default single-agg path below. if let Some(result) = self.try_handle_query_promql_via_timeline(&query, time) { let total_query_duration = query_start_time.elapsed(); debug!( @@ -2863,8 +2865,8 @@ impl SimpleEngine { /// pipeline runs; only the "which aggregation covers this /// query" step is replaced. /// - /// Phase 3b-2-a (refactor) — the caller for this entry point is - /// Phase 3b-2-b's per-segment dispatch: for each + /// Caller: the per-segment dispatch in + /// [`Self::try_handle_query_promql_via_timeline`]. For each /// `TimelineSegment` returned by /// [`crate::stores::sketch_db::SchemaRegistry::timeline_for_metric`], /// the dispatch builds a context targeting that segment's @@ -2954,7 +2956,7 @@ impl SimpleEngine { /// `QueryConfig` exact-string match first, then fall back to /// capability-based matching (with controller miss-notification /// if wired). Extracted from `build_query_execution_context_promql` - /// so the Phase 3b-2-b per-segment dispatch can choose NOT to + /// so the per-segment timeline dispatch can choose NOT to /// auto-resolve (it has a forced agg_id from the timeline). fn resolve_agg_info_promql( &self, @@ -2981,7 +2983,7 @@ impl SimpleEngine { } /// Build an `AggregationIdInfo` from a single forced `agg_id`, - /// for the Phase 3b-2-b per-segment dispatch. Uses the same + /// for the per-segment timeline dispatch. Uses the same /// "one agg covers both key and value" shape as the single- /// aggregation branch in `get_aggregation_id_info` (line /// ~1881), so downstream dispatch treats this agg identically @@ -3006,8 +3008,7 @@ impl SimpleEngine { }) } - /// Phase 3b-2-b: per-segment dispatch across the §7 schema - /// timeline. + /// Per-segment dispatch across the §7 schema timeline. /// /// Returns `Some(result)` when `SchemaRegistry::timeline_for_metric` /// yields two or more segments for the query's metric within its @@ -3033,7 +3034,7 @@ impl SimpleEngine { /// plus one or more `warnings` strings explaining the schema /// boundary, the dropped groups, and the unresolved segments. /// - /// Delivers the user-visible Phase 3 outcome documented in + /// Delivers the user-visible outcome documented in /// `docs/design-sketch-db.md` §7: queries spanning a reconfigure /// boundary no longer see a silent data cliff — additive stats /// get the combined answer, and non-combinable stats get an @@ -3084,7 +3085,7 @@ impl SimpleEngine { t1, t2, statistic = ?stat, - "Phase 3 timeline dispatch: evaluating per-segment" + "schema-timeline dispatch: evaluating per-segment" ); // Phase 4: per-segment evaluation. Each segment's agg_id @@ -3140,6 +3141,11 @@ impl SimpleEngine { } }; + debug!( + agg_id = segment.agg_id, + count = per_segment_results.len(), + "schema-timeline dispatch: segment produced results" + ); for el in per_segment_results { per_group .entry(Some(el.labels)) @@ -3150,6 +3156,11 @@ impl SimpleEngine { }); } } + debug!( + groups = per_group.len(), + unresolved = unresolved.len(), + "schema-timeline dispatch: about to combine" + ); // Phase 5: per-group combine. Group-by label-tuple so the // combiner folds per-segment scalars into one final scalar @@ -5671,7 +5682,7 @@ mod aux_pushdown_tests { } } -// ── Phase 3b-2-a: build_query_execution_context_promql_for_agg_id tests ── +// ── build_query_execution_context_promql_for_agg_id (forced-agg) tests ── #[cfg(test)] mod forced_agg_id_tests { @@ -5704,7 +5715,7 @@ mod forced_agg_id_tests { /// The new forced-agg-id entry point produces a context when /// the forced `agg_id` matches the one the auto-resolver /// would have picked. Basic smoke test; §7 timeline dispatch - /// tests in Phase 3b-2-b will exercise it against multiple + /// per-segment timeline dispatch tests exercise it against multiple /// agg_ids. #[test] fn forced_agg_id_produces_context_for_known_id() { @@ -5732,7 +5743,7 @@ mod forced_agg_id_tests { } /// Unknown `agg_id` returns `None` without panicking or - /// polluting the auto-resolver state. Phase 3b-2-b relies on + /// polluting the auto-resolver state. Per-segment timeline dispatch relies on /// this to gracefully skip timeline segments whose agg_id /// disappeared from the StreamingConfig mid-query. #[test] @@ -5760,7 +5771,7 @@ mod forced_agg_id_tests { /// Forced and auto-resolved contexts should be observably /// equivalent for the common one-agg case (where the /// auto-resolver would have picked the same id). The - /// invariant that matters for Phase 3b-2-b: dispatching + /// invariant that matters for per-segment timeline dispatch: dispatching /// through the forced path against the single covering /// segment yields the same answer as the existing path. #[test] diff --git a/asap-query-engine/src/engines/timeline_dispatch.rs b/asap-query-engine/src/engines/timeline_dispatch.rs index 082a69b2..a8e6baa0 100644 --- a/asap-query-engine/src/engines/timeline_dispatch.rs +++ b/asap-query-engine/src/engines/timeline_dispatch.rs @@ -17,10 +17,10 @@ //! * 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. +//! The query engine wires this primitive into +//! `SimpleEngine::try_handle_query_promql_via_timeline`, which +//! runs the per-segment evaluation loop and feeds the scalars back +//! through `combine_statistic`. //! //! ## Statistic combinability //! diff --git a/asap-query-engine/src/stores/sketch_db/backfill.rs b/asap-query-engine/src/stores/sketch_db/backfill.rs index a41675aa..6a6957da 100644 --- a/asap-query-engine/src/stores/sketch_db/backfill.rs +++ b/asap-query-engine/src/stores/sketch_db/backfill.rs @@ -10,8 +10,9 @@ //! agg_id — say the operator widens a CMS from 256 to 2048, or swaps //! in KLL200 on top of a metric that previously had only CMS — the new //! agg has zero history. Queries spanning the reconfigure boundary -//! either see a data cliff (which Phase 3's schema timeline at least -//! surfaces honestly) or have to fall back to the exact DB. +//! either see a data cliff (which the §7 schema timeline surfaces +//! honestly via `Partial` results + `warnings`) or have to fall +//! back to the exact DB. //! //! Backfill closes that gap: a `BackfillJob` reads raw samples from //! the exact DB for a `(agg_id, time_range)` window, rebuilds the diff --git a/asap-query-engine/src/stores/sketch_db/schema.rs b/asap-query-engine/src/stores/sketch_db/schema.rs index da9627ad..7386cf9f 100644 --- a/asap-query-engine/src/stores/sketch_db/schema.rs +++ b/asap-query-engine/src/stores/sketch_db/schema.rs @@ -32,9 +32,9 @@ //! * `SchemaRegistry` — an in-memory map keyed by `agg_id` that the //! ingest path consults. Built from the current `StreamingConfig` //! snapshot at construction; reconciled event-driven by the -//! `POST /api/v1/streaming-config` swap handler (Phase 2b). +//! `POST /api/v1/streaming-config` swap handler. //! -//! Phase 3a added the §7 **schema timeline** read API: +//! **§7 schema timeline read API:** //! //! * `TimelineSegment` + `TimelineCoverage` types. //! * `SchemaRegistry::timeline_for_metric(metric, t1_ms, t2_ms)` @@ -42,12 +42,11 @@ //! `[t1, t2]` for a given metric. Derived on-demand from registry //! state — no separate index to keep consistent. //! -//! Phase 3b lands the combiner used by the query engine to stitch -//! per-segment scalars into a single result (see -//! `crate::engines::timeline_dispatch`); Phase 3b-2 will wire that -//! combiner into the PromQL dispatch. +//! The query engine stitches per-segment scalars into a single +//! result via the combiner in `crate::engines::timeline_dispatch`, +//! wired through `SimpleEngine::try_handle_query_promql_via_timeline`. //! -//! Phase 2c (this commit) adds **on-disk schema persistence**: +//! **On-disk schema persistence:** //! //! * `SchemaRegistry::load_or_new_from_config(path, &StreamingConfig)` //! reads a JSON snapshot if present (preserving `created_at_ms` / @@ -65,7 +64,7 @@ //! will be filled in once the sketch types' theoretical bounds are //! vendored. //! * `combine_statistic()` and `PartialResult` for cross-segment -//! result stitching — Phase 3b. +//! result stitching — see `crate::engines::timeline_dispatch`. //! * Compaction policy that reads `AggStatus` to throttle as expiry //! approaches — §9.2 of the design. @@ -571,15 +570,15 @@ impl SchemaRegistry { /// caller sees a coverage hole and can fall back to the exact /// DB per §7.3. /// - /// ## Current limitations (Phase 3a scope) + /// ## Current limitations /// /// * `created_at_ms` is currently the wall-clock at which the /// backend first observed the schema, not necessarily when the - /// first datapoint was written. After a restart without on-disk - /// schema persistence (Phase 2c) the timeline reflects only the - /// *post-restart* history. This is the right-edge-of-time - /// behaviour the precompute engine already had pre-Phase-3; - /// Phase 2c closes this gap. + /// first datapoint was written. Without on-disk schema + /// persistence, the timeline reflects only the *post-restart* + /// history — matching the right-edge-of-time behaviour the + /// precompute engine had before the timeline read API existed. + /// On-disk schema persistence closes that gap. /// * All segments are returned, including those whose schema is /// `Expired`. The caller inspects `TimelineSegment::status` to /// decide whether data is still readable. @@ -589,9 +588,9 @@ impl SchemaRegistry { /// Linear in the number of schemas for the given metric (one /// pass to collect + sort). For the metric counts typical of /// sketch DB deployments (dozens of metrics × a handful of - /// schemas each) this is well under a microsecond. Phase 3b's - /// query path will call this once per `query_metric()` so the - /// cost is amortised across the query. + /// schemas each) this is well under a microsecond. The query + /// path calls this once per query, so the cost is amortised + /// across the query. pub fn timeline_for_metric( &self, metric: &str, @@ -741,10 +740,11 @@ pub struct TimelineSegment { /// Coarse classification of whether a [`TimelineSegment`]'s data is /// expected to be readable from the sketch store. §7.3 of the design -/// doc lays out the full state machine; Phase 3a exposes the two -/// states we can determine purely from schema metadata. Phase 3b / -/// Phase 5 (backfill) will extend this with `BackfillInProgress` and -/// with finer-grained per-window coverage. +/// doc lays out the full state machine; this enum exposes the two +/// states we can determine purely from schema metadata. Follow-up +/// work (cross-segment stitching, backfill coverage) will extend +/// this with `BackfillInProgress` and finer-grained per-window +/// coverage. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum TimelineCoverage { /// Data is (or was) written by the live ingest path and the @@ -917,7 +917,7 @@ mod tests { assert_eq!(r.get(1).unwrap().status(), AggStatus::Expired); } - // --- §7 timeline tests (Phase 3a) --- + // --- §7 timeline_for_metric tests --- /// Build a schema with explicit timestamps, bypassing the /// wall-clock path. `metric_override` defaults to `metric_{id}` diff --git a/asap-query-engine/src/tests/mod.rs b/asap-query-engine/src/tests/mod.rs index 010b50df..fca55a2a 100644 --- a/asap-query-engine/src/tests/mod.rs +++ b/asap-query-engine/src/tests/mod.rs @@ -7,6 +7,7 @@ pub mod persistence_integration_tests; pub mod persistence_perf_tests; pub mod prometheus_forwarding_tests; pub mod query_equivalence_tests; +pub mod schema_timeline_dispatch_tests; pub mod sql_pattern_matching_tests; pub mod store_correctness_tests; pub mod trait_design_tests; diff --git a/asap-query-engine/src/tests/schema_timeline_dispatch_tests.rs b/asap-query-engine/src/tests/schema_timeline_dispatch_tests.rs new file mode 100644 index 00000000..cefea2e3 --- /dev/null +++ b/asap-query-engine/src/tests/schema_timeline_dispatch_tests.rs @@ -0,0 +1,306 @@ +//! End-to-end tests for the schema-timeline query dispatcher. +//! +//! Exercises the full path from a PromQL query → schema registry +//! lookup → per-segment store query → `combine_statistic` → +//! Prometheus `warnings`, on a real `SimpleEngine` + +//! `SimpleMapStore` + `SchemaRegistry` with two agg_ids for the +//! same metric and a reconfigure boundary inside the query range. +//! +//! Contract validated: queries that span a reconfigure boundary +//! do not see a silent data cliff. Combinable statistics (Count / +//! Sum / Min / Max) get the stitched answer; non-combinable or +//! Purged segments surface as explicit `warnings` on the response +//! so the caller knows the answer is partial. +//! +//! Lives inside the crate (not `tests/`) so we can reach the +//! `#[cfg(test)] insert_raw_for_testing` helper on `SchemaRegistry` +//! without leaking a test-only API into the public crate surface. + +use std::collections::HashMap; +use std::sync::Arc; + +use asap_types::aggregation_config::AggregationConfig; +use asap_types::aggregation_reference::AggregationReference; +use asap_types::enums::{AggregationType, WindowType}; +use asap_types::promql_schema::PromQLSchema; +use asap_types::query_config::QueryConfig; +use promql_utilities::data_model::key_by_label_names::KeyByLabelNames; + +use crate::data_model::{ + CleanupPolicy, HotReloadStreamingConfig, InferenceConfig, KeyByLabelValues, PrecomputedOutput, + QueryLanguage, SchemaConfig, StreamingConfig, +}; +use crate::engines::{QueryResult, SimpleEngine}; +use crate::precompute_operators::sum_accumulator::SumAccumulator; +use crate::stores::sketch_db::simple_map_store::SimpleMapStore; +use crate::stores::sketch_db::{AggSchema, SchemaRegistry}; +use crate::stores::Store; + +const METRIC: &str = "sensor_reading"; + +// Timeline layout used by the tests. Picked so that an instant +// query at `QUERY_TIME_SEC` produces a range that straddles the +// reconfigure boundary between `agg_1` and `agg_2`. +// +// * `BOUNDARY_MS` — where `agg_1.retired_at_ms` == `agg_2.created_at_ms`. +// * `AGG1_SAMPLE_MS` — at-boundary-ish stamp used for agg_1's seeded +// window so the clipped `[QUERY_START_MS, BOUNDARY_MS]` sub-query +// finds it. +// * `AGG2_SAMPLE_MS` — post-boundary stamp used for agg_2's seeded +// window so the clipped `[BOUNDARY_MS, QUERY_TIME_MS]` sub-query +// finds it. +const QUERY_TIME_SEC: f64 = 501.0; +const QUERY_TIME_MS: u64 = 501_000; +const BOUNDARY_MS: u64 = 500_500; +const AGG1_SAMPLE_MS: u64 = 500_000; +const AGG2_SAMPLE_MS: u64 = 501_000; + +fn make_agg_config(id: u64) -> AggregationConfig { + AggregationConfig::new( + id, + AggregationType::Sum, + String::new(), + HashMap::new(), + KeyByLabelNames::new(vec!["host".to_string()]), + KeyByLabelNames::empty(), + KeyByLabelNames::empty(), + String::new(), + 1, + 1, + WindowType::Tumbling, + String::new(), + METRIC.to_string(), + None, + None, + None, + None, + ) +} + +/// Construct a schema with explicit lifecycle timestamps, bypassing +/// the wall-clock `new_active` path so we can pin a schema into +/// `Retired` or `Expired` status for coverage testing. +fn fixed_schema( + agg_id: u64, + created_at_ms: u64, + retired_at_ms: Option, + expires_at_ms: Option, +) -> AggSchema { + let mut base = AggSchema::new_active(make_agg_config(agg_id)); + base.created_at_ms = created_at_ms; + base.retired_at_ms = retired_at_ms; + base.expires_at_ms = expires_at_ms; + base +} + +/// Instant PromQL query used by the tests. Runs through the +/// OnlySpatial aggregation pattern (op=sum) with a `by (host)` +/// modifier — the engine's `format_final_results` path only +/// emits keyed output elements, so the query must be grouped for +/// the result vector to be non-empty. +const TEST_QUERY: &str = "sum by (host) (sensor_reading)"; + +fn build_engine( + streaming_config: Arc, + schemas: Arc, + store: Arc, + query_for_agg_id: u64, +) -> SimpleEngine { + let mut inference_config = + InferenceConfig::new(QueryLanguage::promql, CleanupPolicy::NoCleanup); + let promql_schema = PromQLSchema::new().add_metric( + METRIC.to_string(), + KeyByLabelNames::new(vec!["host".to_string()]), + ); + inference_config.schema = SchemaConfig::PromQL(promql_schema); + // Pin the test query to the active agg so the probe resolution + // succeeds; the dispatcher still visits every timeline segment + // regardless of which one the probe picked. + inference_config.query_configs = vec![QueryConfig::new(TEST_QUERY.to_string()) + .add_aggregation(AggregationReference::new(query_for_agg_id, None))]; + + let hot_reload = HotReloadStreamingConfig::from_arc(streaming_config); + SimpleEngine::new_with_hot_reload( + store, + inference_config, + hot_reload, + 1, + QueryLanguage::promql, + ) + .with_schema_registry(schemas) +} + +/// Insert a single `SumAccumulator` window at `ts` into `agg_id`. +/// Uses `(ts, ts)` for the window start/end pair to match the +/// existing test-utility pattern in `engine_factories` — the engine +/// treats those as single-point buckets aligned to the tumbling +/// window, so a query whose range contains `ts` picks up the data. +fn seed_sum_at(store: &SimpleMapStore, agg_id: u64, ts: u64, host: &str, sum: f64) { + let key = Some(KeyByLabelValues { + labels: vec![host.to_string()], + }); + let output = PrecomputedOutput::new(ts, ts, key, agg_id); + let acc = SumAccumulator::with_sum(sum); + store + .insert_precomputed_output(output, Box::new(acc)) + .expect("seed insert must succeed"); +} + +/// Two schemas for the same metric, both answerable: agg_1 is +/// Retired-but-not-Expired (coverage=Sketch) with data in its own +/// lifetime, agg_2 is Active with data post-boundary. Sum is +/// combinable, so the dispatcher folds 10.0 + 20.0 into +/// `Full(30.0)` — no warnings, no data cliff. +#[test] +fn sum_query_across_reconfigure_boundary_returns_combined_full_result() { + let mut agg_map = HashMap::new(); + agg_map.insert(1u64, make_agg_config(1)); + agg_map.insert(2u64, make_agg_config(2)); + let streaming_config = Arc::new(StreamingConfig::new(agg_map)); + + let schemas = Arc::new(SchemaRegistry::empty()); + // agg_1: Retired at the boundary but not yet Expired, so + // coverage stays `Sketch` and the dispatcher evaluates it. + schemas.insert_raw_for_testing(fixed_schema(1, 0, Some(BOUNDARY_MS), Some(u64::MAX / 4))); + // agg_2: Active from the boundary onwards. + schemas.insert_raw_for_testing(fixed_schema(2, BOUNDARY_MS, None, None)); + + let store = Arc::new(SimpleMapStore::new( + streaming_config.clone(), + CleanupPolicy::NoCleanup, + )); + // Data placed so the instant query at `QUERY_TIME_SEC` sweeps + // `[QUERY_START_MS, QUERY_TIME_MS]`. After the dispatcher clips + // per segment: + // agg_1's sub-range is `[QUERY_START_MS, BOUNDARY_MS]` + // agg_2's sub-range is `[BOUNDARY_MS, QUERY_TIME_MS]` + seed_sum_at(&store, 1, AGG1_SAMPLE_MS, "A", 10.0); + seed_sum_at(&store, 2, AGG2_SAMPLE_MS, "A", 20.0); + + let engine = build_engine(streaming_config, schemas, store, 2); + + let (_labels, qr) = engine + .handle_query_promql(TEST_QUERY.to_string(), QUERY_TIME_SEC) + .expect("query must produce a result"); + + assert!( + qr.warnings().is_empty(), + "Sum is cleanly combinable — no warnings expected, got {:?}", + qr.warnings() + ); + match qr { + QueryResult::Vector(iv) => { + assert_eq!(iv.values.len(), 1, "one combined scalar across segments"); + assert!( + (iv.values[0].value - 30.0).abs() < 1e-9, + "expected 10 + 20 = 30.0 across the reconfigure boundary, got {}", + iv.values[0].value + ); + } + other => panic!("expected instant vector, got {other:?}"), + } +} + +/// agg_1 Expired (coverage=Purged, unresolved); agg_2 Active with +/// data. `combine_statistic(Sum)` on a combinable stat with a +/// non-empty `unresolved` list returns `Partial { covered: Some, +/// missing: [...] }`. The engine surfaces the partial through +/// `QueryResult::warnings()`. +#[test] +fn sum_query_with_purged_segment_returns_partial_with_warnings() { + let mut agg_map = HashMap::new(); + agg_map.insert(1u64, make_agg_config(1)); + agg_map.insert(2u64, make_agg_config(2)); + let streaming_config = Arc::new(StreamingConfig::new(agg_map)); + + let schemas = Arc::new(SchemaRegistry::empty()); + // agg_1: retired at the boundary so its lifetime [0, BOUNDARY_MS) + // overlaps the query range, but `expires_at_ms` is in the past + // so `status()` returns `Expired` → `coverage_for` returns + // `Purged`. The dispatcher treats this segment as unresolved + // and forces a Partial combine. + schemas.insert_raw_for_testing(fixed_schema(1, 0, Some(BOUNDARY_MS), Some(1_000))); + schemas.insert_raw_for_testing(fixed_schema(2, BOUNDARY_MS, None, None)); + + let store = Arc::new(SimpleMapStore::new( + streaming_config.clone(), + CleanupPolicy::NoCleanup, + )); + // Only agg_2 has data; agg_1's data is assumed gone with the + // Purged classification. + seed_sum_at(&store, 2, AGG2_SAMPLE_MS, "A", 20.0); + + let engine = build_engine(streaming_config, schemas, store, 2); + + let (_labels, qr) = engine + .handle_query_promql(TEST_QUERY.to_string(), QUERY_TIME_SEC) + .expect("query must produce a result even with a Partial combine"); + + assert!( + !qr.warnings().is_empty(), + "Purged segment must populate warnings — got empty list" + ); + let joined = qr.warnings().join(" | "); + assert!( + joined.contains("partial result") && joined.contains(METRIC), + "warnings should explain the partial + reference the metric: {joined}" + ); + assert!( + joined.contains("agg_id=1"), + "warnings should enumerate the unresolved agg_id=1: {joined}" + ); + + match qr { + QueryResult::Vector(iv) => { + assert_eq!(iv.values.len(), 1, "best-effort covered sum"); + assert!( + (iv.values[0].value - 20.0).abs() < 1e-9, + "covered sum is agg_2's 20.0; got {}", + iv.values[0].value + ); + } + other => panic!("expected instant vector, got {other:?}"), + } +} + +/// Single-schema regression guard: when the timeline has only one +/// segment, the dispatcher returns `None`, the default single-agg +/// path handles the query, no warnings attach. +#[test] +fn single_schema_query_falls_through_to_default_path() { + let mut agg_map = HashMap::new(); + agg_map.insert(7u64, make_agg_config(7)); + let streaming_config = Arc::new(StreamingConfig::new(agg_map)); + + let schemas = Arc::new(SchemaRegistry::empty()); + schemas.insert_raw_for_testing(fixed_schema(7, 0, None, None)); + + let store = Arc::new(SimpleMapStore::new( + streaming_config.clone(), + CleanupPolicy::NoCleanup, + )); + seed_sum_at(&store, 7, QUERY_TIME_MS, "A", 42.0); + + let engine = build_engine(streaming_config, schemas, store, 7); + + let (_labels, qr) = engine + .handle_query_promql(TEST_QUERY.to_string(), QUERY_TIME_SEC) + .expect("query must produce a result"); + + assert!( + qr.warnings().is_empty(), + "single-schema path must not attach warnings: {:?}", + qr.warnings() + ); + match qr { + QueryResult::Vector(iv) => { + assert_eq!(iv.values.len(), 1); + assert!( + (iv.values[0].value - 42.0).abs() < 1e-9, + "single-agg answer should be 42.0, got {}", + iv.values[0].value + ); + } + other => panic!("expected instant vector, got {other:?}"), + } +}