From e27236d15399298c6a73aee40399a0eeac23db66 Mon Sep 17 00:00:00 2001 From: zz_y Date: Sun, 13 Sep 2026 11:21:34 -0600 Subject: [PATCH 1/2] refactor: split the MinMax aggregation family into Min and Max A summary's family said `MinMax` and its direction rode alongside it as a string. `build_backend_aggregation_json` emitted `aggregationType: MinMax` and recovered "max" from the family in an `if matches!` wedged inside the `json!` literal; `accumulator_factory` re-derived the direction with `aggregation_sub_type.eq_ignore_ascii_case("max")`; `MinMaxAccumulator` carried `sub_type: String` and panicked on anything that was not "min" or "max". Every one of those is a place where the direction can disagree with the state it labels, and the failure is silent: a minimum answered as a maximum. Worse, `capability_for` mapped both `AggIntent::Min` and `AggIntent::Max` onto `ExactAgg(MinMax)`, so a deployed minimum summary was a legal candidate for a `max_over_time` read. Direction is the family now, end to end: * `AggregationType::{Min, Max, MultipleMin, MultipleMax}` replace `MinMax`/`MultipleMinMax`. `FromStr` rejects the retired names rather than guessing a direction -- there is no safe guess. * `ExactStateKind::{Min, Max}` and `ExactReadout::Min` give the state contract and the readout contract the same two-sided shape. * The wire is `aggregationType: Min` / `Max` with an empty `aggregationSubType`; nothing reads that field for a direction. * `MinAccumulator`/`MaxAccumulator` and `MultipleMinAccumulator`/ `MultipleMaxAccumulator` replace the `sub_type`-carrying pair. Each answers exactly one `Statistic`, refuses to merge with the other direction, and serializes without a direction byte. * `RollupReduction::Min` joins `Max`, so `min_over_time` gets the same derived-rollup fast path `max_over_time` already had; which rollup a pane feeds follows from the accumulator's type. This needs ASAPPlanner's own split (`ExactKind::MinMax` -> `Max` alongside the existing `Min`), so the planner pin moves to 029ff2fe. Two other changes ride in with that pin: * `BinaryOperator` gained `checked_relative_division` / `checked_finite_division`, and `avg_over_time` now realizes as a guarded Sum/Count division instead of staying archive-only -- `avg_over_time_is_not_yet_realizable_matching_capability_for_today` asserted the old outcome and is rewritten to assert the new one. * PromQL `count(v)` is a row count upstream now, and the distinct-count idiom is `count(distinct_over_time(v[w]))`. The four HLL tests that spelled it `count(v)` are retargeted. Co-Authored-By: Claude Opus 5 (1M context) --- Cargo.lock | 10 +- Cargo.toml | 9 +- .../examples/audit_clickhouse_corpus.rs | 2 +- control_plane/src/asap_tier_implement.rs | 33 +- control_plane/src/clickhouse.rs | 2 +- control_plane/src/emit/stage_config.rs | 16 +- control_plane/src/physical/compiler.rs | 14 +- .../src/physical/post_asap/matcher.rs | 8 +- .../src/physical/runtime_capability.rs | 48 +- control_plane/src/query_plan.rs | 3 +- control_plane/src/query_plan/logical.rs | 2 +- control_plane/src/query_planning.rs | 3 +- control_plane/src/replan.rs | 10 +- control_plane/tests/offline_evidence.rs | 2 + crates/asap_types/src/accumulator_spec.rs | 44 +- crates/asap_types/src/aggregation_type.rs | 45 +- crates/asap_types/src/precompute_plan.rs | 11 +- crates/asap_types/src/query_plan.rs | 1 + crates/asap_types/src/sds.rs | 2 +- .../drivers/ingest/prometheus_remote_write.rs | 6 +- data_plane/src/lib.rs | 2 +- .../precompute_engine/accumulator_factory.rs | 193 ++++--- .../precompute_engine/maintenance_runtime.rs | 2 + .../operators/max_accumulator.rs | 269 ++++++++++ .../operators/min_accumulator.rs | 274 ++++++++++ .../operators/min_max_accumulator.rs | 463 ---------------- .../src/precompute_engine/operators/mod.rs | 12 +- .../operators/multiple_max_accumulator.rs | 338 ++++++++++++ .../operators/multiple_min_accumulator.rs | 338 ++++++++++++ .../operators/multiple_min_max_accumulator.rs | 493 ------------------ .../src/precompute_engine/subdag_scheduler.rs | 2 + .../asap_query_engine/catalog_resolver.rs | 15 +- .../query_engines/asap_query_engine/engine.rs | 32 +- .../asap_query_engine/live_serve.rs | 20 +- .../asap_query_engine/post_asap_readout.rs | 22 +- .../asap_query_engine/summary_exec.rs | 2 + .../asap_query_engine/summary_executor.rs | 98 ++-- .../src/storage_engines/sketch_db/accuracy.rs | 12 +- .../storage_engines/sketch_db/index/mod.rs | 42 +- .../src/storage_engines/sketch_db/sds.rs | 10 +- 40 files changed, 1671 insertions(+), 1239 deletions(-) create mode 100644 data_plane/src/precompute_engine/operators/max_accumulator.rs create mode 100644 data_plane/src/precompute_engine/operators/min_accumulator.rs delete mode 100644 data_plane/src/precompute_engine/operators/min_max_accumulator.rs create mode 100644 data_plane/src/precompute_engine/operators/multiple_max_accumulator.rs create mode 100644 data_plane/src/precompute_engine/operators/multiple_min_accumulator.rs delete mode 100644 data_plane/src/precompute_engine/operators/multiple_min_max_accumulator.rs diff --git a/Cargo.lock b/Cargo.lock index 4ec2c5938..e00797745 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -373,7 +373,7 @@ dependencies = [ [[package]] name = "asap-aware-mapping" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=ca7546de792d74aee8231e9a1100ca893d9e86d3#ca7546de792d74aee8231e9a1100ca893d9e86d3" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=029ff2fe041172c94c2d32c90b185bc83c5e8a57#029ff2fe041172c94c2d32c90b185bc83c5e8a57" dependencies = [ "asap-types", "serde", @@ -384,7 +384,7 @@ dependencies = [ [[package]] name = "asap-frontend-promql" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=ca7546de792d74aee8231e9a1100ca893d9e86d3#ca7546de792d74aee8231e9a1100ca893d9e86d3" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=029ff2fe041172c94c2d32c90b185bc83c5e8a57#029ff2fe041172c94c2d32c90b185bc83c5e8a57" dependencies = [ "asap-types", "promql-parser 0.10.0 (git+https://github.com/ProjectASAP/promql-parser?rev=9fede7eecca923c9882fe256484d00d37f8706cb)", @@ -393,7 +393,7 @@ dependencies = [ [[package]] name = "asap-frontend-sql" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=ca7546de792d74aee8231e9a1100ca893d9e86d3#ca7546de792d74aee8231e9a1100ca893d9e86d3" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=029ff2fe041172c94c2d32c90b185bc83c5e8a57#029ff2fe041172c94c2d32c90b185bc83c5e8a57" dependencies = [ "asap-sql-function-catalog", "asap-types", @@ -416,12 +416,12 @@ dependencies = [ [[package]] name = "asap-sql-function-catalog" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=ca7546de792d74aee8231e9a1100ca893d9e86d3#ca7546de792d74aee8231e9a1100ca893d9e86d3" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=029ff2fe041172c94c2d32c90b185bc83c5e8a57#029ff2fe041172c94c2d32c90b185bc83c5e8a57" [[package]] name = "asap-types" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=ca7546de792d74aee8231e9a1100ca893d9e86d3#ca7546de792d74aee8231e9a1100ca893d9e86d3" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=029ff2fe041172c94c2d32c90b185bc83c5e8a57#029ff2fe041172c94c2d32c90b185bc83c5e8a57" dependencies = [ "serde", "serde_json", diff --git a/Cargo.toml b/Cargo.toml index 3e61c62d0..836774c65 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -20,10 +20,10 @@ asap_sketchlib = { git = "https://github.com/ProjectASAP/asap_sketchlib", branch [workspace.dependencies] # Keep Planner frontends, selection, and IR on the same immutable revision. # Alias upstream asap-types because this workspace also defines asap_types. -planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "ca7546de792d74aee8231e9a1100ca893d9e86d3" } -asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "ca7546de792d74aee8231e9a1100ca893d9e86d3" } -asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "ca7546de792d74aee8231e9a1100ca893d9e86d3" } -asap-frontend-sql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "ca7546de792d74aee8231e9a1100ca893d9e86d3" } +planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "029ff2fe041172c94c2d32c90b185bc83c5e8a57" } +asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "029ff2fe041172c94c2d32c90b185bc83c5e8a57" } +asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "029ff2fe041172c94c2d32c90b185bc83c5e8a57" } +asap-frontend-sql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "029ff2fe041172c94c2d32c90b185bc83c5e8a57" } # Shared external deps (used by 2+ crates) serde = { version = "1.0", features = ["derive"] } @@ -45,3 +45,4 @@ arc-swap = "1.7" asap_types = { path = "crates/asap_types" } asap_otel_proto = { path = "crates/asap_otel_proto" } indexmap = { version = "2.0", features = ["serde"] } + diff --git a/control_plane/examples/audit_clickhouse_corpus.rs b/control_plane/examples/audit_clickhouse_corpus.rs index e7497b8d1..f1d03d454 100644 --- a/control_plane/examples/audit_clickhouse_corpus.rs +++ b/control_plane/examples/audit_clickhouse_corpus.rs @@ -26,7 +26,7 @@ struct Row { fn publication_inputs(schema: &Schema, sql: String) -> ClickHouseSqlWorkload { let mut materialization = PrecomputeMaterialization::new( - AggregationType::MinMax, + AggregationType::Max, String::new(), std::collections::HashMap::from([("variant".into(), json!(2))]), KeyByLabelNames::new(vec!["labels".into()]), diff --git a/control_plane/src/asap_tier_implement.rs b/control_plane/src/asap_tier_implement.rs index 02809866a..3061e5aa7 100644 --- a/control_plane/src/asap_tier_implement.rs +++ b/control_plane/src/asap_tier_implement.rs @@ -258,20 +258,33 @@ mod tests { } #[test] - fn avg_over_time_is_not_yet_realizable_matching_capability_for_today() { - // Avg = Sum / Count needs a cross-policy join implement_tree_in_with - // doesn't build (matches capability_for(&AggIntent::Avg) => None - // on the flat path -- see asap_tier_analysis.rs and lower.rs's - // AggFunc::Avg comment). Use avg_over_time (a range-vector - // function), not bare instant avg(...) -- only the former is - // guaranteed to lower through AggFunc::Avg in this frontend. + fn avg_over_time_realizes_as_a_checked_sum_over_count_division() { + // Avg = Sum / Count. Planner rewrites the temporal average into a + // read-time division of two exact accumulators, guarded by + // `checked_finite_division` so an overflowing quotient falls back to + // the original query instead of serving an infinity. Use + // avg_over_time (a range-vector function), not bare instant avg(...) + // -- only the former is guaranteed to lower through AggFunc::Avg in + // this frontend. let roots = implement_promql_for_asap_tier("avg_over_time(http_requests_total[5m])") .expect("parses and implements"); assert_eq!(roots.len(), 1); + let SummaryExpr::BinaryOp { operator, .. } = &roots[0].expr else { + panic!("avg_over_time realizes as a division: {:?}", roots[0].expr); + }; assert!( - matches!(roots[0].expr, SummaryExpr::KeepPreAsap(_)), - "Avg has no ASAP-tier realization yet on either path: {:?}", - roots[0].expr, + matches!( + operator.kind, + planner_types::pre_asap::BinaryOpKind::Arithmetic( + planner_types::pre_asap::ArithmeticOpKind::Div + ) + ), + "expected a division operator: {operator:?}", + ); + assert!( + operator.checked_finite_division, + "the temporal-average rewrite must stay guarded against an \ + overflowing quotient: {operator:?}", ); } diff --git a/control_plane/src/clickhouse.rs b/control_plane/src/clickhouse.rs index 4abd29e6c..c646ae555 100644 --- a/control_plane/src/clickhouse.rs +++ b/control_plane/src/clickhouse.rs @@ -1212,7 +1212,7 @@ mod tests { ("variant", serde_json::json!(1)), ); let count_60 = materialization( - AggregationType::MinMax, + AggregationType::Max, "requests", 60, 10, diff --git a/control_plane/src/emit/stage_config.rs b/control_plane/src/emit/stage_config.rs index 6d94e1643..d2110e479 100644 --- a/control_plane/src/emit/stage_config.rs +++ b/control_plane/src/emit/stage_config.rs @@ -2938,7 +2938,10 @@ fn build_gateway_merge_block(mp: &GatewayMergeProcessor) -> Value { /// /// * `aggregationType` — sketch family. /// * `aggregationSubType` — always empty; reserved for future -/// sub-family distinctions. +/// sub-family distinctions. Min/max direction used to ride here +/// because Planner had one `MinMax` accumulator for both; it is +/// `aggregationType: Min` / `Max` now, so nothing reads this field +/// to pick a direction. /// * `metric` — source metric the aggregation runs over. /// * `labels.{grouping,rollup,aggregated}` — three label lists the /// backend's `KeyByLabelNames` parser keys on. Today the typed L5 @@ -2969,7 +2972,8 @@ pub(crate) fn build_backend_aggregation_json(agg: &BackendAggregation) -> JsonVa match kind { ExactKind::Sum => "Sum", ExactKind::Count => "Count", - ExactKind::MinMax => "MinMax", + ExactKind::Min => "Min", + ExactKind::Max => "Max", ExactKind::Increase => "Increase", ExactKind::Rate => "Rate", ExactKind::IRate => "IRate", @@ -3023,13 +3027,7 @@ pub(crate) fn build_backend_aggregation_json(agg: &BackendAggregation) -> JsonVa clamp_window_secs(Some(agg.window_secs)).expect("clamp_window_secs preserves Some"); json!({ "aggregationType": aggregation_type, - "aggregationSubType": if matches!( - &agg.family, - planner_types::post_asap::SummaryFamilyType::ExactAggregate( - planner_types::post_asap::ExactKind::MinMax, - _ - ) - ) { "max" } else { "" }, + "aggregationSubType": "", "metric": agg.metric_name, "labels": { "grouping": agg.grouping, diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index 2ea40ab23..526493fea 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -780,7 +780,7 @@ fn has_unsafe_raw_entity_leaf( let preserves_series_state = matches!( family, SummaryFamilyType::ExactAggregate( - ExactKind::Increase | ExactKind::Rate | ExactKind::MinMax, + ExactKind::Increase | ExactKind::Rate | ExactKind::Max, _ ) ); @@ -1091,7 +1091,7 @@ impl PhysicalCompiler { && (!matches!( state.family, SummaryFamilyType::ExactAggregate( - planner_types::post_asap::ExactKind::MinMax, + planner_types::post_asap::ExactKind::Max, _ ) ) || crate::query_plan::logical::selected_range_max_materialization( @@ -2657,10 +2657,12 @@ fn retained_state_bytes(materialization: &asap_types::PrecomputeMaterialization) A::DDSketch => 64 * 1024, A::Sum | A::Increase - | A::MinMax + | A::Min + | A::Max | A::MultipleSum | A::MultipleIncrease - | A::MultipleMinMax + | A::MultipleMin + | A::MultipleMax | A::SingleSubpopulation | A::MultipleSubpopulation => 256, } @@ -2678,7 +2680,7 @@ fn retained_partition_count( if materialization.partitioning == Some(asap_types::sds::PopulationPartitioning::PerEntity) || matches!( materialization.aggregation_type, - A::Increase | A::MultipleIncrease | A::MinMax | A::MultipleMinMax + A::Increase | A::MultipleIncrease | A::Min | A::Max | A::MultipleMin | A::MultipleMax ) || !materialization.grouping_labels.names().is_empty() { @@ -5189,6 +5191,8 @@ pub(crate) mod tests { lhs: selected.clone(), rhs: selected.clone(), operator: planner_types::post_asap::BinaryOperator { + checked_relative_division: false, + checked_finite_division: false, kind: planner_types::pre_asap::BinaryOpKind::Arithmetic( planner_types::pre_asap::ArithmeticOpKind::Add, ), diff --git a/control_plane/src/physical/post_asap/matcher.rs b/control_plane/src/physical/post_asap/matcher.rs index f5a671e8d..356fa7c36 100644 --- a/control_plane/src/physical/post_asap/matcher.rs +++ b/control_plane/src/physical/post_asap/matcher.rs @@ -192,7 +192,8 @@ mod tests { match kind { ExactKind::Sum => ExactParams::Sum, ExactKind::Count => ExactParams::Count, - ExactKind::MinMax => ExactParams::MinMax, + ExactKind::Min => ExactParams::Min, + ExactKind::Max => ExactParams::Max, ExactKind::Increase => ExactParams::Increase, ExactKind::Rate => ExactParams::Rate, ExactKind::IRate => ExactParams::IRate, @@ -287,10 +288,7 @@ mod tests { fn exact_accumulator_requires_the_exact_same_kind() { let m = SummaryFamilyMatcher; assert!(m.is_satisfied_by(&accumulator(ExactKind::Sum), &accumulator(ExactKind::Sum))); - assert!(!m.is_satisfied_by( - &accumulator(ExactKind::Sum), - &accumulator(ExactKind::MinMax) - )); + assert!(!m.is_satisfied_by(&accumulator(ExactKind::Sum), &accumulator(ExactKind::Max))); assert!(!m.is_satisfied_by( &accumulator(ExactKind::Increase), &accumulator(ExactKind::Rate) diff --git a/control_plane/src/physical/runtime_capability.rs b/control_plane/src/physical/runtime_capability.rs index 67f6caf67..1e2050d16 100644 --- a/control_plane/src/physical/runtime_capability.rs +++ b/control_plane/src/physical/runtime_capability.rs @@ -62,7 +62,7 @@ pub enum Capability { /// wire format can answer this. `Any` required matches either /// `CmsWithHeap` or `CountSketchWithHeap`. FrequencyTopk(Option), - /// Exact-aggregation ASAP-tier state — Sum / Count / MinMax / Avg / + /// Exact-aggregation ASAP-tier state — Sum / Count / Min / Max / Avg / /// Rate / Increase / SetAggregator etc. Backed by a per-accumulator /// payload (`AggPayload::ExactAgg` in the data plane). One variant /// per [`AggregationType`] — the inner enum names the concrete @@ -386,8 +386,8 @@ fn sketch_algorithms_compatible( /// True when `available` is the multi-population equivalent of /// `required`'s single-population variant — i.e. a `MultipleSum` /// policy can serve a `Sum` query (via re-aggregation across keys), -/// `MultipleIncrease` can serve `Increase`, `MultipleMinMax` can -/// serve `MinMax`. Asymmetric: this returns `false` for the reverse +/// `MultipleIncrease` can serve `Increase`, `MultipleMax` can +/// serve `Max`. Asymmetric: this returns `false` for the reverse /// direction (single-pop can't recover keys that have been collapsed /// away). fn multi_pop_satisfies_single(required: AggregationType, available: AggregationType) -> bool { @@ -395,7 +395,8 @@ fn multi_pop_satisfies_single(required: AggregationType, available: AggregationT (required, available), (AggregationType::Sum, AggregationType::MultipleSum) | (AggregationType::Increase, AggregationType::MultipleIncrease) - | (AggregationType::MinMax, AggregationType::MultipleMinMax) + | (AggregationType::Min, AggregationType::MultipleMin) + | (AggregationType::Max, AggregationType::MultipleMax) ) } @@ -428,9 +429,12 @@ pub fn capability_for(intent: &AggIntent) -> Option { } match intent { AggIntent::Sum { .. } => Some(Capability::ExactAgg(AggregationType::Sum)), - AggIntent::Min { .. } | AggIntent::Max { .. } => { - Some(Capability::ExactAgg(AggregationType::MinMax)) - } + // Direction is part of the capability: a stored minimum cannot + // answer `max_over_time` and vice versa, so these must not + // collapse onto one `ExactAgg` the way they did while Planner + // had a single `MinMax` accumulator. + AggIntent::Min { .. } => Some(Capability::ExactAgg(AggregationType::Min)), + AggIntent::Max { .. } => Some(Capability::ExactAgg(AggregationType::Max)), AggIntent::Increase | AggIntent::Rate => { Some(Capability::ExactAgg(AggregationType::Increase)) } @@ -571,24 +575,36 @@ mod tests { } #[test] - fn capability_for_min_returns_exact_agg_minmax() { + fn capability_for_min_returns_exact_agg_min() { // Min/Max are exact, mergeable accumulators -- no approximation // needed at all -- matching ASAPController's own // `crates/plan/src/boundary.rs` treatment. assert_eq!( capability_for(&AggIntent::Min { col: None }), - Some(Capability::ExactAgg(AggregationType::MinMax)) + Some(Capability::ExactAgg(AggregationType::Min)) ); } #[test] - fn capability_for_max_returns_exact_agg_minmax() { + fn capability_for_max_returns_exact_agg_max() { assert_eq!( capability_for(&AggIntent::Max { col: None }), - Some(Capability::ExactAgg(AggregationType::MinMax)) + Some(Capability::ExactAgg(AggregationType::Max)) ); } + #[test] + fn exact_agg_min_and_max_do_not_satisfy_each_other() { + // The whole point of splitting the family: a deployed minimum + // sid must never be routed a `max_over_time` read. + assert!(!Capability::ExactAgg(AggregationType::Min) + .is_satisfied_by(&Capability::ExactAgg(AggregationType::Max))); + assert!(!Capability::ExactAgg(AggregationType::Max) + .is_satisfied_by(&Capability::ExactAgg(AggregationType::Min))); + assert!(!Capability::ExactAgg(AggregationType::Min) + .is_satisfied_by(&Capability::ExactAgg(AggregationType::MultipleMax))); + } + #[test] fn capability_for_rate_increase_route_to_exact_agg_increase() { // PR-6 follow-up: Rate and Increase route to ASAP-tier @@ -805,10 +821,10 @@ mod tests { #[test] fn is_satisfied_by_exact_agg_different_types_do_not_match() { - // Sum required, MinMax indexed → no match. No wildcard for + // Sum required, Max indexed → no match. No wildcard for // ExactAgg — every agg_type stands on its own. let required = Capability::ExactAgg(AggregationType::Sum); - let indexed = Capability::ExactAgg(AggregationType::MinMax); + let indexed = Capability::ExactAgg(AggregationType::Max); assert!(!required.is_satisfied_by(&indexed)); } @@ -848,11 +864,13 @@ mod tests { let cases = [ AggregationType::Sum, AggregationType::Increase, - AggregationType::MinMax, + AggregationType::Min, + AggregationType::Max, AggregationType::DatasketchesKLL, AggregationType::MultipleSum, AggregationType::MultipleIncrease, - AggregationType::MultipleMinMax, + AggregationType::MultipleMin, + AggregationType::MultipleMax, AggregationType::HydraKLL, AggregationType::CountMinSketch, AggregationType::CountMinSketchWithHeap, diff --git a/control_plane/src/query_plan.rs b/control_plane/src/query_plan.rs index bcce290d2..ce1f059a3 100644 --- a/control_plane/src/query_plan.rs +++ b/control_plane/src/query_plan.rs @@ -886,7 +886,8 @@ fn exact_readout(family: &SummaryFamilyType) -> Option { SummaryFamilyType::ExactAggregate(ExactKind::Count, _) => Some(ExactReadout::Count), SummaryFamilyType::ExactAggregate(ExactKind::Increase, _) => Some(ExactReadout::Increase), SummaryFamilyType::ExactAggregate(ExactKind::Rate, _) => Some(ExactReadout::Rate), - SummaryFamilyType::ExactAggregate(ExactKind::MinMax, _) => Some(ExactReadout::Max), + SummaryFamilyType::ExactAggregate(ExactKind::Min, _) => Some(ExactReadout::Min), + SummaryFamilyType::ExactAggregate(ExactKind::Max, _) => Some(ExactReadout::Max), _ => None, } } diff --git a/control_plane/src/query_plan/logical.rs b/control_plane/src/query_plan/logical.rs index e802ea4ba..a29658c41 100644 --- a/control_plane/src/query_plan/logical.rs +++ b/control_plane/src/query_plan/logical.rs @@ -903,7 +903,7 @@ pub(crate) fn selected_range_max_materialization( if !matches!( &node.expr, SummaryExpr::SummaryAgg { - family: SummaryFamilyType::ExactAggregate(ExactKind::MinMax, _), + family: SummaryFamilyType::ExactAggregate(ExactKind::Max, _), reduction: planner_types::pre_asap::Reduction::PerEntity, .. } diff --git a/control_plane/src/query_planning.rs b/control_plane/src/query_planning.rs index 87bc1704c..128809375 100644 --- a/control_plane/src/query_planning.rs +++ b/control_plane/src/query_planning.rs @@ -215,7 +215,8 @@ fn planned_capability( | planner_types::post_asap::ExactKind::IRate => { asap_types::AggregationType::Increase } - planner_types::post_asap::ExactKind::MinMax => asap_types::AggregationType::MinMax, + planner_types::post_asap::ExactKind::Min => asap_types::AggregationType::Min, + planner_types::post_asap::ExactKind::Max => asap_types::AggregationType::Max, }; Capability::ExactAgg(agg) } diff --git a/control_plane/src/replan.rs b/control_plane/src/replan.rs index 4ee23c58a..ff1abb8bb 100644 --- a/control_plane/src/replan.rs +++ b/control_plane/src/replan.rs @@ -588,7 +588,7 @@ impl Replanner { /// `count(metric)`) AND the role classifies as /// Sum/Count/Other/Topk, synthesize a single ExactAgg-shaped /// `BackendStageConfig` carrying an `agg_type_override` of - /// `"Sum"` / `"Increase"` / `"MinMax"` so the cumulative + /// `"Sum"` / `"Increase"` / `"Min"` / `"Max"` so the cumulative /// streaming-config still surfaces the metric to the backend. /// Without this fallback the typed cumulative POST would omit /// every Sum-shaped metric and `sum by (zone) (…)` queries would @@ -653,7 +653,8 @@ impl Replanner { let exact_kind = match agg_type_override? { "Sum" => planner_types::post_asap::ExactKind::Sum, "Count" => planner_types::post_asap::ExactKind::Count, - "MinMax" => planner_types::post_asap::ExactKind::MinMax, + "Min" => planner_types::post_asap::ExactKind::Min, + "Max" => planner_types::post_asap::ExactKind::Max, "Increase" | "Rate" => planner_types::post_asap::ExactKind::Increase, _ => return None, }; @@ -662,9 +663,8 @@ impl Replanner { planner_types::post_asap::ExactKind::Count => { planner_types::post_asap::ExactParams::Count } - planner_types::post_asap::ExactKind::MinMax => { - planner_types::post_asap::ExactParams::MinMax - } + planner_types::post_asap::ExactKind::Min => planner_types::post_asap::ExactParams::Min, + planner_types::post_asap::ExactKind::Max => planner_types::post_asap::ExactParams::Max, planner_types::post_asap::ExactKind::Increase => { planner_types::post_asap::ExactParams::Increase } diff --git a/control_plane/tests/offline_evidence.rs b/control_plane/tests/offline_evidence.rs index 0c4a936b1..abcd47ef4 100644 --- a/control_plane/tests/offline_evidence.rs +++ b/control_plane/tests/offline_evidence.rs @@ -355,6 +355,8 @@ fn binary_summary_has_explicit_warm_tier_fallback() { lhs: child.clone(), rhs: child.clone(), operator: BinaryOperator { + checked_relative_division: false, + checked_finite_division: false, kind: BinaryOpKind::Arithmetic(planner_types::pre_asap::ArithmeticOpKind::Div), vector_match: None, }, diff --git a/crates/asap_types/src/accumulator_spec.rs b/crates/asap_types/src/accumulator_spec.rs index 5e6202b6b..482ef6d67 100644 --- a/crates/asap_types/src/accumulator_spec.rs +++ b/crates/asap_types/src/accumulator_spec.rs @@ -41,11 +41,13 @@ //! //! Backend-specific execution details remain deliberately separate: //! -//! - **Min/max direction.** Planner's `ExactParams::MinMax` carries no fields — -//! upstream doesn't model a direction axis. `accumulator_factory.rs` -//! keeps reading `AggregationConfig::aggregation_sub_type` directly -//! for this one bit (`eq_ignore_ascii_case("max")`), exactly as it did -//! before this refactor. +//! - **Min/max direction.** Direction is part of the family now, not a +//! string riding alongside it: `AggregationType::{Min, Max}` (and the +//! keyed `{MultipleMin, MultipleMax}`) map to `ExactKind::Min` and +//! `ExactKind::Max` respectively — upstream still spells its +//! maximum accumulator `MinMax`, but it is a maximum. Nothing reads +//! `AggregationConfig::aggregation_sub_type` for the direction any +//! more, so a min state can no longer content-address onto a max one. //! - **HydraKLL's `(row, col)` tiling.** `SketchParams::Kll` carries //! only `k` — upstream has no concept of the CMS-like grid-of-KLL-cells //! layout `HydraKllSketchAccumulator` uses to parallelize a keyed KLL @@ -233,8 +235,12 @@ impl AggregationConfig { SummaryFamilyType::ExactAggregate(ExactKind::Increase, ExactParams::Increase), false, ), - MinMax => ( - SummaryFamilyType::ExactAggregate(ExactKind::MinMax, ExactParams::MinMax), + Min => ( + SummaryFamilyType::ExactAggregate(ExactKind::Min, ExactParams::Min), + false, + ), + Max => ( + SummaryFamilyType::ExactAggregate(ExactKind::Max, ExactParams::Max), false, ), DatasketchesKLL => ( @@ -254,8 +260,12 @@ impl AggregationConfig { SummaryFamilyType::ExactAggregate(ExactKind::Increase, ExactParams::Increase), true, ), - MultipleMinMax => ( - SummaryFamilyType::ExactAggregate(ExactKind::MinMax, ExactParams::MinMax), + MultipleMin => ( + SummaryFamilyType::ExactAggregate(ExactKind::Min, ExactParams::Min), + true, + ), + MultipleMax => ( + SummaryFamilyType::ExactAggregate(ExactKind::Max, ExactParams::Max), true, ), HydraKLL => { @@ -390,8 +400,12 @@ impl AggregationConfig { SummaryFamilyType::ExactAggregate(ExactKind::Sum, ExactParams::Sum), false, ), - "Min" | "min" | "Max" | "max" => ( - SummaryFamilyType::ExactAggregate(ExactKind::MinMax, ExactParams::MinMax), + "Min" | "min" => ( + SummaryFamilyType::ExactAggregate(ExactKind::Min, ExactParams::Min), + false, + ), + "Max" | "max" => ( + SummaryFamilyType::ExactAggregate(ExactKind::Max, ExactParams::Max), false, ), "Increase" | "increase" => ( @@ -418,8 +432,12 @@ impl AggregationConfig { SummaryFamilyType::ExactAggregate(ExactKind::Sum, ExactParams::Sum), true, ), - "Min" | "min" | "Max" | "max" => ( - SummaryFamilyType::ExactAggregate(ExactKind::MinMax, ExactParams::MinMax), + "Min" | "min" => ( + SummaryFamilyType::ExactAggregate(ExactKind::Min, ExactParams::Min), + true, + ), + "Max" | "max" => ( + SummaryFamilyType::ExactAggregate(ExactKind::Max, ExactParams::Max), true, ), "Increase" | "increase" => ( diff --git a/crates/asap_types/src/aggregation_type.rs b/crates/asap_types/src/aggregation_type.rs index f1e4f0f97..791f09033 100644 --- a/crates/asap_types/src/aggregation_type.rs +++ b/crates/asap_types/src/aggregation_type.rs @@ -31,12 +31,14 @@ pub enum AggregationType { // ---------- single-population (non-keyed) ---------- Sum, Increase, - MinMax, + Min, + Max, DatasketchesKLL, // ---------- multi-population (keyed) ---------- MultipleSum, MultipleIncrease, - MultipleMinMax, + MultipleMin, + MultipleMax, HydraKLL, CountMinSketch, CountMinSketchWithHeap, @@ -56,11 +58,13 @@ impl AggregationType { match self { AggregationType::Sum => "Sum", AggregationType::Increase => "Increase", - AggregationType::MinMax => "MinMax", + AggregationType::Min => "Min", + AggregationType::Max => "Max", AggregationType::DatasketchesKLL => "DatasketchesKLL", AggregationType::MultipleSum => "MultipleSum", AggregationType::MultipleIncrease => "MultipleIncrease", - AggregationType::MultipleMinMax => "MultipleMinMax", + AggregationType::MultipleMin => "MultipleMin", + AggregationType::MultipleMax => "MultipleMax", AggregationType::HydraKLL => "HydraKLL", AggregationType::CountMinSketch => "CountMinSketch", AggregationType::CountMinSketchWithHeap => "CountMinSketchWithHeap", @@ -81,7 +85,8 @@ impl AggregationType { AggregationType::MultipleSubpopulation | AggregationType::MultipleSum | AggregationType::MultipleIncrease - | AggregationType::MultipleMinMax + | AggregationType::MultipleMin + | AggregationType::MultipleMax | AggregationType::CountMinSketch | AggregationType::CountMinSketchWithHeap | AggregationType::CountSketch @@ -95,7 +100,8 @@ impl AggregationType { matches!( self, AggregationType::MultipleSum - | AggregationType::MultipleMinMax + | AggregationType::MultipleMin + | AggregationType::MultipleMax | AggregationType::MultipleIncrease | AggregationType::CountMinSketch | AggregationType::CountMinSketchWithHeap @@ -128,11 +134,13 @@ impl FromStr for AggregationType { // Canonical names "Sum" => Ok(AggregationType::Sum), "Increase" => Ok(AggregationType::Increase), - "MinMax" => Ok(AggregationType::MinMax), + "Min" => Ok(AggregationType::Min), + "Max" => Ok(AggregationType::Max), "DatasketchesKLL" => Ok(AggregationType::DatasketchesKLL), "MultipleSum" => Ok(AggregationType::MultipleSum), "MultipleIncrease" => Ok(AggregationType::MultipleIncrease), - "MultipleMinMax" => Ok(AggregationType::MultipleMinMax), + "MultipleMin" => Ok(AggregationType::MultipleMin), + "MultipleMax" => Ok(AggregationType::MultipleMax), "HydraKLL" => Ok(AggregationType::HydraKLL), "CountMinSketch" => Ok(AggregationType::CountMinSketch), "CountMinSketchWithHeap" => Ok(AggregationType::CountMinSketchWithHeap), @@ -148,7 +156,8 @@ impl FromStr for AggregationType { "IncreaseAccumulator" | "IncreaseAggregator" | "increase" => { Ok(AggregationType::Increase) } - "MinMaxAccumulator" | "MinMaxAggregator" | "min_max" => Ok(AggregationType::MinMax), + "MinAccumulator" | "MinAggregator" | "min" => Ok(AggregationType::Min), + "MaxAccumulator" | "MaxAggregator" | "max" => Ok(AggregationType::Max), "DatasketchesKLLAccumulator" | "KLL" | "kll" | "datasketches_kll" => { Ok(AggregationType::DatasketchesKLL) } @@ -156,7 +165,8 @@ impl FromStr for AggregationType { "MultipleIncreaseAccumulator" | "multiple_increase" => { Ok(AggregationType::MultipleIncrease) } - "MultipleMinMaxAccumulator" | "multiple_min_max" => Ok(AggregationType::MultipleMinMax), + "MultipleMinAccumulator" | "multiple_min" => Ok(AggregationType::MultipleMin), + "MultipleMaxAccumulator" | "multiple_max" => Ok(AggregationType::MultipleMax), "HydraKllSketchAccumulator" | "hydra_kll" => Ok(AggregationType::HydraKLL), "CountMinSketchAccumulator" | "CMS" | "cms" | "count_min_sketch" => { Ok(AggregationType::CountMinSketch) @@ -166,6 +176,21 @@ impl FromStr for AggregationType { Ok(AggregationType::CountSketch) } "CountSketchWithHeapAccumulator" => Ok(AggregationType::CountSketchWithHeap), + // Retired names. `MinMax` used to be one accumulator whose + // direction rode alongside in `aggregationSubType`; the two + // directions are separate types now, so there is no safe + // direction to guess here -- resolving a min workload as a + // max one is silently wrong, not merely imprecise. + "MinMax" + | "MinMaxAccumulator" + | "MinMaxAggregator" + | "min_max" + | "MultipleMinMax" + | "MultipleMinMaxAccumulator" + | "multiple_min_max" => Err(format!( + "Retired aggregation type: '{s}' -- min and max are separate types now, \ + use 'Min'/'Max' (or 'MultipleMin'/'MultipleMax')" + )), _ => Err(format!("Unknown aggregation type: '{s}'")), } } diff --git a/crates/asap_types/src/precompute_plan.rs b/crates/asap_types/src/precompute_plan.rs index 4a6ce9d71..a219ff03b 100644 --- a/crates/asap_types/src/precompute_plan.rs +++ b/crates/asap_types/src/precompute_plan.rs @@ -177,7 +177,13 @@ pub enum StateFamilyContract { pub enum ExactStateKind { Sum, Count, - MinMax, + /// Exact minimum. Distinct state from [`ExactStateKind::Max`] -- a + /// stored minimum cannot answer a maximum query, so the two must + /// never share a content address. + Min, + /// Exact maximum. Planner still spells this `ExactKind::Max` for + /// historical reasons; it is a maximum accumulator. + Max, Increase, Rate, IRate, @@ -193,7 +199,8 @@ impl TryFrom<&SummaryFamilyType> for StateFamilyContract { kind: match kind { ExactKind::Sum => ExactStateKind::Sum, ExactKind::Count => ExactStateKind::Count, - ExactKind::MinMax => ExactStateKind::MinMax, + ExactKind::Max => ExactStateKind::Max, + ExactKind::Min => ExactStateKind::Min, ExactKind::Increase => ExactStateKind::Increase, ExactKind::Rate => ExactStateKind::Rate, ExactKind::IRate => ExactStateKind::IRate, diff --git a/crates/asap_types/src/query_plan.rs b/crates/asap_types/src/query_plan.rs index 8b0c48e8c..679f17ad3 100644 --- a/crates/asap_types/src/query_plan.rs +++ b/crates/asap_types/src/query_plan.rs @@ -627,6 +627,7 @@ pub enum ExactReadout { Count, Increase, Rate, + Min, Max, } diff --git a/crates/asap_types/src/sds.rs b/crates/asap_types/src/sds.rs index a44fff11d..3caa44224 100644 --- a/crates/asap_types/src/sds.rs +++ b/crates/asap_types/src/sds.rs @@ -668,7 +668,7 @@ impl FidelityGuarantee { (aggregation_type, self), (A::UnivMon, UnivMonFrequency { .. }) | ( - A::Sum | A::MultipleSum | A::MinMax | A::MultipleMinMax, + A::Sum | A::MultipleSum | A::Min | A::Max | A::MultipleMin | A::MultipleMax, Exact ) | (A::Increase | A::MultipleIncrease, ExactCounter { .. }) diff --git a/data_plane/src/drivers/ingest/prometheus_remote_write.rs b/data_plane/src/drivers/ingest/prometheus_remote_write.rs index b37d4c798..0ddf1a8b6 100644 --- a/data_plane/src/drivers/ingest/prometheus_remote_write.rs +++ b/data_plane/src/drivers/ingest/prometheus_remote_write.rs @@ -710,8 +710,10 @@ fn route_messages( config.aggregation_type, asap_types::AggregationType::Increase | asap_types::AggregationType::MultipleIncrease - | asap_types::AggregationType::MinMax - | asap_types::AggregationType::MultipleMinMax + | asap_types::AggregationType::Min + | asap_types::AggregationType::Max + | asap_types::AggregationType::MultipleMin + | asap_types::AggregationType::MultipleMax )); let grouping_pairs: Vec<(&str, &str)> = if series_scoped { Vec::new() diff --git a/data_plane/src/lib.rs b/data_plane/src/lib.rs index 1273bbc0a..1920450f2 100644 --- a/data_plane/src/lib.rs +++ b/data_plane/src/lib.rs @@ -43,7 +43,7 @@ pub use storage_engines::types::{ }; pub use precompute_engine::operators::{ - IncreaseAccumulator, MinMaxAccumulator, MultipleSumAccumulator, SumAccumulator, + IncreaseAccumulator, MaxAccumulator, MinAccumulator, MultipleSumAccumulator, SumAccumulator, }; pub use storage_engines::StoreResult; diff --git a/data_plane/src/precompute_engine/accumulator_factory.rs b/data_plane/src/precompute_engine/accumulator_factory.rs index 81ae8fb92..2947d5334 100644 --- a/data_plane/src/precompute_engine/accumulator_factory.rs +++ b/data_plane/src/precompute_engine/accumulator_factory.rs @@ -1,8 +1,9 @@ use crate::precompute_engine::operators::{ CountMinSketchAccumulator, CountMinSketchWithHeapAccumulator, CountSketchAccumulator, CountSketchWithHeapAccumulator, DDSketchAccumulator, DatasketchesKLLAccumulator, - HydraKllSketchAccumulator, IncreaseAccumulator, MinMaxAccumulator, MultipleIncreaseAccumulator, - MultipleMinMaxAccumulator, MultipleSumAccumulator, SumAccumulator, + HydraKllSketchAccumulator, IncreaseAccumulator, MaxAccumulator, MinAccumulator, + MultipleIncreaseAccumulator, MultipleMaxAccumulator, MultipleMinAccumulator, + MultipleSumAccumulator, SumAccumulator, }; use crate::storage_engines::types::{ AggregateCore, AggregationType, KeyByLabelValues, Measurement, @@ -148,55 +149,51 @@ impl AccumulatorUpdater for SumAccumulatorUpdater { } // --------------------------------------------------------------------------- -// MinMaxAccumulatorUpdater +// MinAccumulatorUpdater / MaxAccumulatorUpdater // --------------------------------------------------------------------------- -pub struct MinMaxAccumulatorUpdater { - acc: MinMaxAccumulator, - is_max: bool, -} +macro_rules! extremum_updater { + ($updater:ident, $acc:ty) => { + #[derive(Default)] + pub struct $updater { + acc: $acc, + } -impl MinMaxAccumulatorUpdater { - pub fn new(is_max: bool) -> Self { - Self { - acc: if is_max { - MinMaxAccumulator::new_max() - } else { - MinMaxAccumulator::new_min() - }, - is_max, + impl $updater { + pub fn new() -> Self { + Self::default() + } } - } -} -impl AccumulatorUpdater for MinMaxAccumulatorUpdater { - fn update_single(&mut self, value: f64, _timestamp_ms: i64) { - self.acc.update(value); - } + impl AccumulatorUpdater for $updater { + fn update_single(&mut self, value: f64, _timestamp_ms: i64) { + self.acc.update(value); + } - fn update_keyed(&mut self, _key: &KeyByLabelValues, value: f64, timestamp_ms: i64) { - self.update_single(value, timestamp_ms); - } + fn update_keyed(&mut self, _key: &KeyByLabelValues, value: f64, timestamp_ms: i64) { + self.update_single(value, timestamp_ms); + } - impl_clone_accumulator_methods!(acc); + impl_clone_accumulator_methods!(acc); - fn reset(&mut self) { - self.acc = if self.is_max { - MinMaxAccumulator::new_max() - } else { - MinMaxAccumulator::new_min() - }; - } + fn reset(&mut self) { + self.acc = <$acc>::new(); + } - fn is_keyed(&self) -> bool { - false - } + fn is_keyed(&self) -> bool { + false + } - fn memory_usage_bytes(&self) -> usize { - std::mem::size_of::() - } + fn memory_usage_bytes(&self) -> usize { + std::mem::size_of::<$acc>() + } + } + }; } +extremum_updater!(MinAccumulatorUpdater, MinAccumulator); +extremum_updater!(MaxAccumulatorUpdater, MaxAccumulator); + // --------------------------------------------------------------------------- // IncreaseAccumulatorUpdater // --------------------------------------------------------------------------- @@ -432,59 +429,55 @@ impl AccumulatorUpdater for MultipleSumAccumulatorUpdater { } // --------------------------------------------------------------------------- -// MultipleMinMaxAccumulatorUpdater +// MultipleMinAccumulatorUpdater / MultipleMaxAccumulatorUpdater // --------------------------------------------------------------------------- -pub struct MultipleMinMaxAccumulatorUpdater { - acc: MultipleMinMaxAccumulator, - is_max: bool, -} +macro_rules! multiple_extremum_updater { + ($updater:ident, $acc:ty) => { + #[derive(Default)] + pub struct $updater { + acc: $acc, + } -impl MultipleMinMaxAccumulatorUpdater { - pub fn new(is_max: bool) -> Self { - Self { - acc: if is_max { - MultipleMinMaxAccumulator::new_max() - } else { - MultipleMinMaxAccumulator::new_min() - }, - is_max, + impl $updater { + pub fn new() -> Self { + Self::default() + } } - } -} -impl AccumulatorUpdater for MultipleMinMaxAccumulatorUpdater { - fn update_single(&mut self, _value: f64, _timestamp_ms: i64) { - debug_assert!( - false, - "update_single called on keyed updater; use update_keyed" - ); - } + impl AccumulatorUpdater for $updater { + fn update_single(&mut self, _value: f64, _timestamp_ms: i64) { + debug_assert!( + false, + "update_single called on keyed updater; use update_keyed" + ); + } - fn update_keyed(&mut self, key: &KeyByLabelValues, value: f64, _timestamp_ms: i64) { - self.acc.update(key.clone(), value); - } + fn update_keyed(&mut self, key: &KeyByLabelValues, value: f64, _timestamp_ms: i64) { + self.acc.update(key.clone(), value); + } - impl_clone_accumulator_methods!(acc); + impl_clone_accumulator_methods!(acc); - fn reset(&mut self) { - self.acc = if self.is_max { - MultipleMinMaxAccumulator::new_max() - } else { - MultipleMinMaxAccumulator::new_min() - }; - } + fn reset(&mut self) { + self.acc = <$acc>::new(); + } - fn is_keyed(&self) -> bool { - true - } + fn is_keyed(&self) -> bool { + true + } - fn memory_usage_bytes(&self) -> usize { - std::mem::size_of::() - + self.acc.values.len() * (std::mem::size_of::() + 8) - } + fn memory_usage_bytes(&self) -> usize { + std::mem::size_of::<$acc>() + + self.acc.values.len() * (std::mem::size_of::() + 8) + } + } + }; } +multiple_extremum_updater!(MultipleMinAccumulatorUpdater, MultipleMinAccumulator); +multiple_extremum_updater!(MultipleMaxAccumulatorUpdater, MultipleMaxAccumulator); + // --------------------------------------------------------------------------- // MultipleIncreaseAccumulatorUpdater // --------------------------------------------------------------------------- @@ -911,7 +904,8 @@ pub fn config_is_keyed(config: &AggregationConfig) -> bool { AggregationType::MultipleSubpopulation | AggregationType::MultipleSum | AggregationType::MultipleIncrease - | AggregationType::MultipleMinMax + | AggregationType::MultipleMin + | AggregationType::MultipleMax | AggregationType::CountMinSketch | AggregationType::CountMinSketchWithHeap | AggregationType::CountSketch @@ -1059,21 +1053,22 @@ pub fn create_accumulator_updater(config: &AggregationConfig) -> Box Box::new( - MinMaxAccumulatorUpdater::new(config.aggregation_sub_type.eq_ignore_ascii_case("max")), - ), - (SummaryFamilyType::ExactAggregate(ExactKind::MinMax, _), true) => { - Box::new(MultipleMinMaxAccumulatorUpdater::new( - config.aggregation_sub_type.eq_ignore_ascii_case("max"), - )) + // Direction comes off the family itself now. It used to be read + // back out of `aggregation_sub_type` because Planner had one + // `MinMax` accumulator for both directions, which meant a config + // whose sub_type was lost or misspelled silently built the wrong + // extremum. + (SummaryFamilyType::ExactAggregate(ExactKind::Min, _), false) => { + Box::new(MinAccumulatorUpdater::new()) + } + (SummaryFamilyType::ExactAggregate(ExactKind::Min, _), true) => { + Box::new(MultipleMinAccumulatorUpdater::new()) + } + (SummaryFamilyType::ExactAggregate(ExactKind::Max, _), false) => { + Box::new(MaxAccumulatorUpdater::new()) + } + (SummaryFamilyType::ExactAggregate(ExactKind::Max, _), true) => { + Box::new(MultipleMaxAccumulatorUpdater::new()) } (SummaryFamilyType::ExactAggregate(ExactKind::Increase, _), false) => { @@ -1346,13 +1341,13 @@ mod tests { #[test] fn test_minmax_updater() { - let mut updater = MinMaxAccumulatorUpdater::new(true); + let mut updater = MaxAccumulatorUpdater::new(); updater.update_single(5.0, 1000); updater.update_single(3.0, 2000); updater.update_single(7.0, 3000); let acc = updater.take_accumulator(); - assert_eq!(acc.type_name(), "MinMaxAccumulator"); + assert_eq!(acc.type_name(), "MaxAccumulator"); } #[test] @@ -1488,7 +1483,7 @@ mod tests { "" ))); assert!(config_is_keyed(&make_config( - AggregationType::MultipleMinMax, + AggregationType::MultipleMax, "" ))); assert!(config_is_keyed(&make_config( diff --git a/data_plane/src/precompute_engine/maintenance_runtime.rs b/data_plane/src/precompute_engine/maintenance_runtime.rs index ab1f35d42..7afdc3961 100644 --- a/data_plane/src/precompute_engine/maintenance_runtime.rs +++ b/data_plane/src/precompute_engine/maintenance_runtime.rs @@ -3255,6 +3255,8 @@ mod tests { ]; operation.output_schema.time_index = Some(0); let mut operator = BinaryOperator { + checked_relative_division: false, + checked_finite_division: false, kind: BinaryOpKind::Arithmetic(ArithmeticOpKind::Sub), vector_match: None, }; diff --git a/data_plane/src/precompute_engine/operators/max_accumulator.rs b/data_plane/src/precompute_engine/operators/max_accumulator.rs new file mode 100644 index 000000000..9dc62fb2a --- /dev/null +++ b/data_plane/src/precompute_engine/operators/max_accumulator.rs @@ -0,0 +1,269 @@ +use crate::storage_engines::types::{ + AggregateCore, AggregationType, AuxStats, MergeableAccumulator, SerializableToSink, + SingleSubpopulationAggregate, SingleSubpopulationAggregateFactory, +}; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use std::collections::HashMap; + +use asap_types::Statistic; + +/// Exact maximum over one population, mergeable by comparison. +/// +/// See [`MinAccumulator`](super::min_accumulator::MinAccumulator) for why the +/// two directions are separate types rather than one accumulator carrying a +/// `sub_type` string. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct MaxAccumulator { + pub value: f64, +} + +impl Default for MaxAccumulator { + fn default() -> Self { + Self::new() + } +} + +impl MaxAccumulator { + pub fn new() -> Self { + Self { + value: f64::NEG_INFINITY, + } + } + + pub fn with_value(value: f64) -> Self { + Self { value } + } + + pub fn update(&mut self, value: f64) { + if value > self.value { + self.value = value; + } + } + + pub fn deserialize_from_json(data: &Value) -> Result> { + let value = data["value"] + .as_f64() + .ok_or("Missing or invalid 'value' field")?; + Ok(Self::with_value(value)) + } + + pub fn deserialize_from_bytes(buffer: &[u8]) -> Result> { + if buffer.len() < 8 { + return Err("Buffer too short".into()); + } + let value = f64::from_le_bytes([ + buffer[0], buffer[1], buffer[2], buffer[3], buffer[4], buffer[5], buffer[6], buffer[7], + ]); + Ok(Self::with_value(value)) + } +} + +impl SerializableToSink for MaxAccumulator { + fn serialize_to_json(&self) -> Value { + serde_json::json!({ "value": self.value }) + } + + fn serialize_to_bytes(&self) -> Vec { + self.value.to_le_bytes().to_vec() + } +} + +impl MergeableAccumulator for MaxAccumulator { + fn merge_accumulators( + accumulators: Vec, + ) -> Result> { + if accumulators.is_empty() { + return Err("No accumulators to merge".into()); + } + let mut result = MaxAccumulator::new(); + for acc in accumulators { + result.update(acc.value); + } + Ok(result) + } +} + +impl AggregateCore for MaxAccumulator { + fn clone_boxed_core(&self) -> Box { + Box::new(self.clone()) + } + + fn type_name(&self) -> &'static str { + "MaxAccumulator" + } + + fn as_any(&self) -> &dyn std::any::Any { + self + } + + fn as_any_mut(&mut self) -> &mut dyn std::any::Any { + self + } + + fn merge_with( + &self, + other: &dyn AggregateCore, + ) -> Result, Box> { + if other.get_accumulator_type() != self.get_accumulator_type() { + return Err(format!( + "Cannot merge MaxAccumulator with {}", + other.get_accumulator_type() + ) + .into()); + } + let other_max = other + .as_any() + .downcast_ref::() + .ok_or("Failed to downcast to MaxAccumulator")?; + let mut merged = self.clone(); + merged.update(other_max.value); + Ok(Box::new(merged)) + } + + fn get_accumulator_type(&self) -> AggregationType { + AggregationType::Max + } + + fn approx_memory_bytes(&self) -> usize { + std::mem::size_of::() + } + + fn aux_stats(&self) -> AuxStats { + // The sentinel `f64::NEG_INFINITY` from `new()` is surfaced as-is; the + // query engine already treats it as "no data yet", the same way it + // does for `query_statistic`. + AuxStats { + max: Some(self.value), + ..AuxStats::empty() + } + } + + fn get_keys(&self) -> Option> { + None + } + + fn query_statistic( + &self, + statistic: asap_types::Statistic, + _key: &Option, + _query_kwargs: &std::collections::HashMap, + ) -> Result> { + use crate::storage_engines::types::SingleSubpopulationAggregate; + self.query(statistic, None) + } +} + +impl SingleSubpopulationAggregate for MaxAccumulator { + fn query( + &self, + statistic: Statistic, + query_kwargs: Option<&HashMap>, + ) -> Result> { + if query_kwargs.is_some() { + return Err("MaxAccumulator does not support query parameters".into()); + } + match statistic { + Statistic::Max => Ok(self.value), + other => Err(format!("Unsupported statistic in MaxAccumulator: {other:?}").into()), + } + } + + fn clone_boxed(&self) -> Box { + Box::new(self.clone()) + } +} + +#[derive(Debug, Default, Clone, Copy)] +pub struct MaxAccumulatorFactory; + +impl SingleSubpopulationAggregateFactory for MaxAccumulatorFactory { + fn merge_accumulators( + &self, + accumulators: Vec>, + ) -> Result, Box> + { + let mut result = f64::NEG_INFINITY; + for acc in accumulators { + result = result.max(acc.query(Statistic::Max, None)?); + } + Ok(Box::new(MaxAccumulator::with_value(result))) + } + + fn create_default(&self) -> Box { + Box::new(MaxAccumulator::new()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn keeps_the_largest_update() { + let mut acc = MaxAccumulator::new(); + acc.update(10.0); + acc.update(5.0); + acc.update(15.0); + + assert_eq!(acc.value, 15.0); + assert_eq!( + crate::SingleSubpopulationAggregate::query(&acc, Statistic::Max, None).unwrap(), + 15.0 + ); + } + + #[test] + fn refuses_to_answer_a_minimum_query() { + let acc = MaxAccumulator::with_value(15.0); + assert!(crate::SingleSubpopulationAggregate::query(&acc, Statistic::Min, None).is_err()); + } + + #[test] + fn merges_by_taking_the_largest() { + let merged = + >::merge_accumulators(vec![ + MaxAccumulator::with_value(10.0), + MaxAccumulator::with_value(5.0), + MaxAccumulator::with_value(15.0), + ]) + .unwrap(); + assert_eq!(merged.value, 15.0); + } + + #[test] + fn refuses_to_merge_with_a_minimum() { + use super::super::min_accumulator::MinAccumulator; + let max = MaxAccumulator::with_value(15.0); + let min = MinAccumulator::with_value(5.0); + assert!(max.merge_with(&min).is_err()); + } + + #[test] + fn round_trips_through_both_serializations() { + let acc = MaxAccumulator::with_value(42.5); + + let json = acc.serialize_to_json(); + assert_eq!( + MaxAccumulator::deserialize_from_json(&json).unwrap().value, + 42.5 + ); + + let bytes = acc.serialize_to_bytes(); + assert_eq!( + MaxAccumulator::deserialize_from_bytes(&bytes) + .unwrap() + .value, + 42.5 + ); + } + + #[test] + fn aux_stats_expose_max_only() { + let aux = MaxAccumulator::with_value(99.0).aux_stats(); + assert_eq!(aux.max, Some(99.0)); + assert_eq!(aux.min, None); + assert_eq!(aux.try_answer(Statistic::Max), Some(99.0)); + assert_eq!(aux.try_answer(Statistic::Min), None); + } +} diff --git a/data_plane/src/precompute_engine/operators/min_accumulator.rs b/data_plane/src/precompute_engine/operators/min_accumulator.rs new file mode 100644 index 000000000..e68a2cf81 --- /dev/null +++ b/data_plane/src/precompute_engine/operators/min_accumulator.rs @@ -0,0 +1,274 @@ +use crate::storage_engines::types::{ + AggregateCore, AggregationType, AuxStats, MergeableAccumulator, SerializableToSink, + SingleSubpopulationAggregate, SingleSubpopulationAggregateFactory, +}; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use std::collections::HashMap; + +use asap_types::Statistic; + +/// Exact minimum over one population, mergeable by comparison. +/// +/// The sibling [`MaxAccumulator`](super::max_accumulator::MaxAccumulator) is a +/// separate type on purpose: these two used to be one `MinMaxAccumulator` +/// whose direction lived in a `sub_type: String`, which meant every layer +/// above -- the wire `aggregationSubType`, the accumulator factory, the +/// summary catalog -- had to carry the direction alongside the family and +/// could silently answer a `min_over_time` read from maximum state. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct MinAccumulator { + pub value: f64, +} + +impl Default for MinAccumulator { + fn default() -> Self { + Self::new() + } +} + +impl MinAccumulator { + pub fn new() -> Self { + Self { + value: f64::INFINITY, + } + } + + pub fn with_value(value: f64) -> Self { + Self { value } + } + + pub fn update(&mut self, value: f64) { + if value < self.value { + self.value = value; + } + } + + pub fn deserialize_from_json(data: &Value) -> Result> { + let value = data["value"] + .as_f64() + .ok_or("Missing or invalid 'value' field")?; + Ok(Self::with_value(value)) + } + + pub fn deserialize_from_bytes(buffer: &[u8]) -> Result> { + if buffer.len() < 8 { + return Err("Buffer too short".into()); + } + let value = f64::from_le_bytes([ + buffer[0], buffer[1], buffer[2], buffer[3], buffer[4], buffer[5], buffer[6], buffer[7], + ]); + Ok(Self::with_value(value)) + } +} + +impl SerializableToSink for MinAccumulator { + fn serialize_to_json(&self) -> Value { + serde_json::json!({ "value": self.value }) + } + + fn serialize_to_bytes(&self) -> Vec { + self.value.to_le_bytes().to_vec() + } +} + +impl MergeableAccumulator for MinAccumulator { + fn merge_accumulators( + accumulators: Vec, + ) -> Result> { + if accumulators.is_empty() { + return Err("No accumulators to merge".into()); + } + let mut result = MinAccumulator::new(); + for acc in accumulators { + result.update(acc.value); + } + Ok(result) + } +} + +impl AggregateCore for MinAccumulator { + fn clone_boxed_core(&self) -> Box { + Box::new(self.clone()) + } + + fn type_name(&self) -> &'static str { + "MinAccumulator" + } + + fn as_any(&self) -> &dyn std::any::Any { + self + } + + fn as_any_mut(&mut self) -> &mut dyn std::any::Any { + self + } + + fn merge_with( + &self, + other: &dyn AggregateCore, + ) -> Result, Box> { + if other.get_accumulator_type() != self.get_accumulator_type() { + return Err(format!( + "Cannot merge MinAccumulator with {}", + other.get_accumulator_type() + ) + .into()); + } + let other_min = other + .as_any() + .downcast_ref::() + .ok_or("Failed to downcast to MinAccumulator")?; + let mut merged = self.clone(); + merged.update(other_min.value); + Ok(Box::new(merged)) + } + + fn get_accumulator_type(&self) -> AggregationType { + AggregationType::Min + } + + fn approx_memory_bytes(&self) -> usize { + std::mem::size_of::() + } + + fn aux_stats(&self) -> AuxStats { + // The sentinel `f64::INFINITY` from `new()` is surfaced as-is; the + // query engine already treats it as "no data yet", the same way it + // does for `query_statistic`. + AuxStats { + min: Some(self.value), + ..AuxStats::empty() + } + } + + fn get_keys(&self) -> Option> { + None + } + + fn query_statistic( + &self, + statistic: asap_types::Statistic, + _key: &Option, + _query_kwargs: &std::collections::HashMap, + ) -> Result> { + use crate::storage_engines::types::SingleSubpopulationAggregate; + self.query(statistic, None) + } +} + +impl SingleSubpopulationAggregate for MinAccumulator { + fn query( + &self, + statistic: Statistic, + query_kwargs: Option<&HashMap>, + ) -> Result> { + if query_kwargs.is_some() { + return Err("MinAccumulator does not support query parameters".into()); + } + match statistic { + Statistic::Min => Ok(self.value), + other => Err(format!("Unsupported statistic in MinAccumulator: {other:?}").into()), + } + } + + fn clone_boxed(&self) -> Box { + Box::new(self.clone()) + } +} + +#[derive(Debug, Default, Clone, Copy)] +pub struct MinAccumulatorFactory; + +impl SingleSubpopulationAggregateFactory for MinAccumulatorFactory { + fn merge_accumulators( + &self, + accumulators: Vec>, + ) -> Result, Box> + { + let mut result = f64::INFINITY; + for acc in accumulators { + result = result.min(acc.query(Statistic::Min, None)?); + } + Ok(Box::new(MinAccumulator::with_value(result))) + } + + fn create_default(&self) -> Box { + Box::new(MinAccumulator::new()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn keeps_the_smallest_update() { + let mut acc = MinAccumulator::new(); + acc.update(10.0); + acc.update(5.0); + acc.update(15.0); + + assert_eq!(acc.value, 5.0); + assert_eq!( + crate::SingleSubpopulationAggregate::query(&acc, Statistic::Min, None).unwrap(), + 5.0 + ); + } + + #[test] + fn refuses_to_answer_a_maximum_query() { + let acc = MinAccumulator::with_value(5.0); + assert!(crate::SingleSubpopulationAggregate::query(&acc, Statistic::Max, None).is_err()); + } + + #[test] + fn merges_by_taking_the_smallest() { + let merged = + >::merge_accumulators(vec![ + MinAccumulator::with_value(10.0), + MinAccumulator::with_value(5.0), + MinAccumulator::with_value(15.0), + ]) + .unwrap(); + assert_eq!(merged.value, 5.0); + } + + #[test] + fn refuses_to_merge_with_a_maximum() { + use super::super::max_accumulator::MaxAccumulator; + let min = MinAccumulator::with_value(5.0); + let max = MaxAccumulator::with_value(15.0); + assert!(min.merge_with(&max).is_err()); + } + + #[test] + fn round_trips_through_both_serializations() { + let acc = MinAccumulator::with_value(42.5); + + let json = acc.serialize_to_json(); + assert_eq!( + MinAccumulator::deserialize_from_json(&json).unwrap().value, + 42.5 + ); + + let bytes = acc.serialize_to_bytes(); + assert_eq!( + MinAccumulator::deserialize_from_bytes(&bytes) + .unwrap() + .value, + 42.5 + ); + } + + #[test] + fn aux_stats_expose_min_only() { + let aux = MinAccumulator::with_value(3.5).aux_stats(); + assert_eq!(aux.min, Some(3.5)); + assert_eq!(aux.max, None); + assert_eq!(aux.count, None); + assert_eq!(aux.sum, None); + assert_eq!(aux.try_answer(Statistic::Min), Some(3.5)); + assert_eq!(aux.try_answer(Statistic::Max), None); + } +} diff --git a/data_plane/src/precompute_engine/operators/min_max_accumulator.rs b/data_plane/src/precompute_engine/operators/min_max_accumulator.rs deleted file mode 100644 index 4111f4ce9..000000000 --- a/data_plane/src/precompute_engine/operators/min_max_accumulator.rs +++ /dev/null @@ -1,463 +0,0 @@ -use crate::storage_engines::types::{ - AggregateCore, AggregationType, AuxStats, MergeableAccumulator, SerializableToSink, - SingleSubpopulationAggregate, SingleSubpopulationAggregateFactory, -}; -use serde::{Deserialize, Serialize}; -use serde_json::Value; -use std::collections::HashMap; - -use asap_types::Statistic; - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct MinMaxAccumulator { - pub value: f64, - pub sub_type: String, // "min" or "max" -} - -impl MinMaxAccumulator { - pub fn new_min() -> Self { - Self { - value: f64::INFINITY, - sub_type: "min".to_string(), - } - } - - pub fn new_max() -> Self { - Self { - value: f64::NEG_INFINITY, - sub_type: "max".to_string(), - } - } - - pub fn new(sub_type: String) -> Self { - match sub_type.as_str() { - "min" => Self::new_min(), - "max" => Self::new_max(), - _ => panic!("sub_type must be 'min' or 'max'"), - } - } - - pub fn with_value(value: f64, sub_type: String) -> Self { - if sub_type != "min" && sub_type != "max" { - panic!("sub_type must be 'min' or 'max'"); - } - Self { value, sub_type } - } - - pub fn update(&mut self, value: f64) { - match self.sub_type.as_str() { - "min" => { - if value < self.value { - self.value = value; - } - } - "max" => { - if value > self.value { - self.value = value; - } - } - _ => panic!("Invalid sub_type"), - } - } - - pub fn deserialize_from_json(data: &Value) -> Result> { - let value = data["value"] - .as_f64() - .ok_or("Missing or invalid 'value' field")?; - let sub_type = data["sub_type"] - .as_str() - .ok_or("Missing or invalid 'sub_type' field")? - .to_string(); - - if sub_type != "min" && sub_type != "max" { - return Err("sub_type must be 'min' or 'max'".into()); - } - - Ok(Self::with_value(value, sub_type)) - } - - pub fn deserialize_from_bytes(buffer: &[u8]) -> Result> { - if buffer.len() < 9 { - return Err("Buffer too short".into()); - } - - let value = f64::from_le_bytes([ - buffer[0], buffer[1], buffer[2], buffer[3], buffer[4], buffer[5], buffer[6], buffer[7], - ]); - - let sub_type = match buffer[8] { - 0 => "min".to_string(), - 1 => "max".to_string(), - _ => return Err("Invalid sub_type byte".into()), - }; - - Ok(Self::with_value(value, sub_type)) - } -} - -impl SerializableToSink for MinMaxAccumulator { - fn serialize_to_json(&self) -> Value { - serde_json::json!({ - "value": self.value, - "sub_type": self.sub_type - }) - } - - fn serialize_to_bytes(&self) -> Vec { - let mut bytes = self.value.to_le_bytes().to_vec(); - let sub_type_byte = match self.sub_type.as_str() { - "min" => 0u8, - "max" => 1u8, - _ => panic!("Invalid sub_type"), - }; - bytes.push(sub_type_byte); - bytes - } -} - -impl MergeableAccumulator for MinMaxAccumulator { - fn merge_accumulators( - accumulators: Vec, - ) -> Result> { - if accumulators.is_empty() { - return Err("No accumulators to merge".into()); - } - - let sub_type = &accumulators[0].sub_type; - - // Verify all accumulators have the same sub_type - for acc in &accumulators { - if acc.sub_type != *sub_type { - return Err("Cannot merge accumulators with different sub_types".into()); - } - } - - let mut result = MinMaxAccumulator::new(sub_type.clone()); - - for acc in accumulators { - result.update(acc.value); - } - - Ok(result) - } -} - -impl AggregateCore for MinMaxAccumulator { - fn clone_boxed_core(&self) -> Box { - Box::new(self.clone()) - } - - fn type_name(&self) -> &'static str { - "MinMaxAccumulator" - } - - fn as_any(&self) -> &dyn std::any::Any { - self - } - - fn as_any_mut(&mut self) -> &mut dyn std::any::Any { - self - } - - fn merge_with( - &self, - other: &dyn AggregateCore, - ) -> Result, Box> { - // Check if other is also a MinMaxAccumulator - if other.get_accumulator_type() != self.get_accumulator_type() { - return Err(format!( - "Cannot merge MinMaxAccumulator with {}", - other.get_accumulator_type() - ) - .into()); - } - - // Downcast to MinMaxAccumulator - let other_minmax = other - .as_any() - .downcast_ref::() - .ok_or("Failed to downcast to MinMaxAccumulator")?; - - if self.sub_type != other_minmax.sub_type { - return Err("Cannot merge MinMaxAccumulators with different sub_types".into()); - } - let mut merged = self.clone(); - merged.update(other_minmax.value); - - Ok(Box::new(merged)) - } - - fn get_accumulator_type(&self) -> AggregationType { - AggregationType::MinMax - } - - fn approx_memory_bytes(&self) -> usize { - // f64 + small sub_type String. - std::mem::size_of::() + self.sub_type.capacity() - } - - fn aux_stats(&self) -> AuxStats { - // A single MinMaxAccumulator instance holds either a min - // or a max depending on sub_type — never both. Surface it - // in the matching aux field so `min_over_time` / `max_over_time` - // queries can read it without deserialising the accumulator. - // - // Sentinel values (±∞ from `new_min()` / `new_max()`) are - // surfaced as-is; the query engine already handles those as - // "no data yet" the same way it does today via `query_statistic`. - match self.sub_type.as_str() { - "min" => AuxStats { - min: Some(self.value), - ..AuxStats::empty() - }, - "max" => AuxStats { - max: Some(self.value), - ..AuxStats::empty() - }, - _ => AuxStats::empty(), - } - } - - fn get_keys(&self) -> Option> { - None - } - - fn query_statistic( - &self, - statistic: asap_types::Statistic, - _key: &Option, - _query_kwargs: &std::collections::HashMap, - ) -> Result> { - use crate::storage_engines::types::SingleSubpopulationAggregate; - self.query(statistic, None) - } -} - -impl SingleSubpopulationAggregate for MinMaxAccumulator { - fn query( - &self, - statistic: Statistic, - query_kwargs: Option<&HashMap>, - ) -> Result> { - // MinMaxAccumulator doesn't use query_kwargs, assert it's None - if query_kwargs.is_some() { - return Err("MinMaxAccumulator does not support query parameters".into()); - } - - match (statistic, self.sub_type.as_str()) { - (Statistic::Min, "min") => Ok(self.value), - (Statistic::Max, "max") => Ok(self.value), - _ => Err(format!( - "Unsupported statistic in MinMaxAccumulator: {:?} for sub_type: {}", - statistic, self.sub_type - ) - .into()), - } - } - - fn clone_boxed(&self) -> Box { - Box::new(self.clone()) - } -} - -// Factory implementation for merging -pub struct MinMaxAccumulatorFactory { - pub sub_type: String, -} - -impl MinMaxAccumulatorFactory { - pub fn new_min() -> Self { - Self { - sub_type: "min".to_string(), - } - } - - pub fn new_max() -> Self { - Self { - sub_type: "max".to_string(), - } - } -} - -impl SingleSubpopulationAggregateFactory for MinMaxAccumulatorFactory { - fn merge_accumulators( - &self, - accumulators: Vec>, - ) -> Result, Box> - { - if accumulators.is_empty() { - return match self.sub_type.as_str() { - "min" => Ok(Box::new(MinMaxAccumulator::new_min())), - "max" => Ok(Box::new(MinMaxAccumulator::new_max())), - _ => Err(format!("Unsupported sub_type: {}", self.sub_type).into()), - }; - } - - let mut result_value = match self.sub_type.as_str() { - "min" => f64::INFINITY, - "max" => f64::NEG_INFINITY, - _ => return Err(format!("Unsupported sub_type: {}", self.sub_type).into()), - }; - - for acc in accumulators { - let value = match self.sub_type.as_str() { - "min" => acc.query(Statistic::Min, None)?, - "max" => acc.query(Statistic::Max, None)?, - _ => return Err(format!("Unsupported sub_type: {}", self.sub_type).into()), - }; - - result_value = match self.sub_type.as_str() { - "min" => result_value.min(value), - "max" => result_value.max(value), - _ => return Err(format!("Unsupported sub_type: {}", self.sub_type).into()), - }; - } - - Ok(Box::new(MinMaxAccumulator::with_value( - result_value, - self.sub_type.clone(), - ))) - } - - fn create_default(&self) -> Box { - match self.sub_type.as_str() { - "min" => Box::new(MinMaxAccumulator::new_min()), - "max" => Box::new(MinMaxAccumulator::new_max()), - _ => Box::new(MinMaxAccumulator::new_min()), // Default fallback - } - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_min_accumulator() { - let mut acc = MinMaxAccumulator::new_min(); - acc.update(10.0); - acc.update(5.0); - acc.update(15.0); - - assert_eq!(acc.value, 5.0); - assert_eq!( - crate::SingleSubpopulationAggregate::query(&acc, Statistic::Min, None).unwrap(), - 5.0 - ); - assert!(crate::SingleSubpopulationAggregate::query(&acc, Statistic::Max, None).is_err()); - } - - #[test] - fn test_max_accumulator() { - let mut acc = MinMaxAccumulator::new_max(); - acc.update(10.0); - acc.update(5.0); - acc.update(15.0); - - assert_eq!(acc.value, 15.0); - assert_eq!( - crate::SingleSubpopulationAggregate::query(&acc, Statistic::Max, None).unwrap(), - 15.0 - ); - assert!(crate::SingleSubpopulationAggregate::query(&acc, Statistic::Min, None).is_err()); - } - - #[test] - fn test_merge_min_accumulators() { - let acc1 = MinMaxAccumulator::with_value(10.0, "min".to_string()); - let acc2 = MinMaxAccumulator::with_value(5.0, "min".to_string()); - let acc3 = MinMaxAccumulator::with_value(15.0, "min".to_string()); - - let merged = - >::merge_accumulators( - vec![acc1, acc2, acc3], - ) - .unwrap(); - assert_eq!(merged.value, 5.0); - assert_eq!(merged.sub_type, "min"); - } - - #[test] - fn test_merge_max_accumulators() { - let acc1 = MinMaxAccumulator::with_value(10.0, "max".to_string()); - let acc2 = MinMaxAccumulator::with_value(5.0, "max".to_string()); - let acc3 = MinMaxAccumulator::with_value(15.0, "max".to_string()); - - let merged = - >::merge_accumulators( - vec![acc1, acc2, acc3], - ) - .unwrap(); - assert_eq!(merged.value, 15.0); - assert_eq!(merged.sub_type, "max"); - } - - #[test] - fn test_merge_different_types_error() { - let acc1 = MinMaxAccumulator::with_value(10.0, "min".to_string()); - let acc2 = MinMaxAccumulator::with_value(5.0, "max".to_string()); - - assert!( - >::merge_accumulators( - vec![acc1, acc2] - ) - .is_err() - ); - } - - #[test] - fn test_serialization() { - let acc = MinMaxAccumulator::with_value(42.5, "min".to_string()); - - // Test JSON serialization - let json = acc.serialize_to_json(); - let deserialized = MinMaxAccumulator::deserialize_from_json(&json).unwrap(); - assert_eq!(acc.value, deserialized.value); - assert_eq!(acc.sub_type, deserialized.sub_type); - - // Test byte serialization - let bytes = acc.serialize_to_bytes(); - let deserialized_bytes = MinMaxAccumulator::deserialize_from_bytes(&bytes).unwrap(); - assert_eq!(acc.value, deserialized_bytes.value); - assert_eq!(acc.sub_type, deserialized_bytes.sub_type); - } - - #[test] - fn test_single_subpopulation_aggregate_trait() { - let acc: Box = - Box::new(MinMaxAccumulator::with_value(42.0, "max".to_string())); - - assert_eq!(acc.query(Statistic::Max, None).unwrap(), 42.0); - assert!(acc.query(Statistic::Min, None).is_err()); - assert_eq!(acc.type_name(), "MinMaxAccumulator"); - } - - #[test] - fn aux_stats_min_variant_exposes_min_only() { - let acc = MinMaxAccumulator::with_value(3.5, "min".to_string()); - let aux = acc.aux_stats(); - assert_eq!(aux.min, Some(3.5)); - assert_eq!(aux.max, None); - assert_eq!(aux.count, None); - assert_eq!(aux.sum, None); - } - - #[test] - fn aux_stats_max_variant_exposes_max_only() { - let acc = MinMaxAccumulator::with_value(99.0, "max".to_string()); - let aux = acc.aux_stats(); - assert_eq!(aux.max, Some(99.0)); - assert_eq!(aux.min, None); - } - - #[test] - fn aux_stats_try_answer_on_min_max() { - let m = MinMaxAccumulator::with_value(7.0, "min".to_string()); - assert_eq!(m.aux_stats().try_answer(Statistic::Min), Some(7.0)); - assert_eq!(m.aux_stats().try_answer(Statistic::Max), None); // not tracked - - let x = MinMaxAccumulator::with_value(7.0, "max".to_string()); - assert_eq!(x.aux_stats().try_answer(Statistic::Max), Some(7.0)); - assert_eq!(x.aux_stats().try_answer(Statistic::Min), None); - } -} diff --git a/data_plane/src/precompute_engine/operators/mod.rs b/data_plane/src/precompute_engine/operators/mod.rs index 7d7c84666..af284459e 100644 --- a/data_plane/src/precompute_engine/operators/mod.rs +++ b/data_plane/src/precompute_engine/operators/mod.rs @@ -8,9 +8,11 @@ pub mod edge_runtime_adapter; pub mod hll_sketch_accumulator; pub mod hydra_kll_accumulator; pub mod increase_accumulator; -pub mod min_max_accumulator; +pub mod max_accumulator; +pub mod min_accumulator; pub mod multiple_increase_accumulator; -pub mod multiple_min_max_accumulator; +pub mod multiple_max_accumulator; +pub mod multiple_min_accumulator; pub mod multiple_sum_accumulator; pub mod sketch_envelope_accumulator; pub mod sum_accumulator; @@ -25,9 +27,11 @@ pub use dd_sketch_accumulator::*; pub use hll_sketch_accumulator::*; pub use hydra_kll_accumulator::*; pub use increase_accumulator::*; -pub use min_max_accumulator::*; +pub use max_accumulator::*; +pub use min_accumulator::*; pub use multiple_increase_accumulator::*; -pub use multiple_min_max_accumulator::*; +pub use multiple_max_accumulator::*; +pub use multiple_min_accumulator::*; pub use multiple_sum_accumulator::*; pub use sketch_envelope_accumulator::*; pub use sum_accumulator::*; diff --git a/data_plane/src/precompute_engine/operators/multiple_max_accumulator.rs b/data_plane/src/precompute_engine/operators/multiple_max_accumulator.rs new file mode 100644 index 000000000..1865d2686 --- /dev/null +++ b/data_plane/src/precompute_engine/operators/multiple_max_accumulator.rs @@ -0,0 +1,338 @@ +use crate::storage_engines::types::{ + AggregateCore, AggregationType, KeyByLabelValues, MergeableAccumulator, + MultipleSubpopulationAggregate, SerializableToSink, +}; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use std::collections::HashMap; + +use asap_types::Statistic; + +/// Exact per-key maximum over many populations, mergeable by comparison. +/// +/// The minimum direction is +/// [`MultipleMinAccumulator`](super::multiple_min_accumulator::MultipleMinAccumulator), +/// a separate type: these used to be one `MultipleMinMaxAccumulator` whose +/// direction lived in a `sub_type` string that every layer above had to carry +/// alongside the family. +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct MultipleMaxAccumulator { + pub values: HashMap, +} + +impl MultipleMaxAccumulator { + pub fn new() -> Self { + Self::default() + } + + pub fn new_with_values(values: HashMap) -> Self { + Self { values } + } + + pub fn update(&mut self, key: KeyByLabelValues, value: f64) { + let current = self.values.entry(key).or_insert(f64::NEG_INFINITY); + if value > *current { + *current = value; + } + } + + pub fn add_value(&mut self, key: KeyByLabelValues, value: f64) { + self.values.insert(key, value); + } + + pub fn deserialize_from_json(data: &Value) -> Result> { + let values_data = data["values"] + .as_object() + .ok_or("Missing or invalid 'values' field")?; + + let mut values = HashMap::new(); + for (key_str, value) in values_data { + let key_json: Value = serde_json::from_str(key_str)?; + let key = KeyByLabelValues::deserialize_from_json(&key_json)?; + let val = value.as_f64().ok_or("Invalid value")?; + values.insert(key, val); + } + + Ok(Self { values }) + } + + pub fn deserialize_from_bytes(buffer: &[u8]) -> Result> { + let mut offset = 0; + + // Read number of entries + if buffer.len() < 4 { + return Err("Buffer too short for entry count".into()); + } + let num_entries = u32::from_le_bytes([ + buffer[offset], + buffer[offset + 1], + buffer[offset + 2], + buffer[offset + 3], + ]) as usize; + offset += 4; + + let mut values = HashMap::new(); + + for _ in 0..num_entries { + // Read key length and data + if buffer.len() < offset + 4 { + return Err("Buffer too short for key length".into()); + } + let key_length = u32::from_le_bytes([ + buffer[offset], + buffer[offset + 1], + buffer[offset + 2], + buffer[offset + 3], + ]) as usize; + offset += 4; + + if buffer.len() < offset + key_length { + return Err("Buffer too short for key data".into()); + } + let key = + KeyByLabelValues::deserialize_from_bytes(&buffer[offset..offset + key_length])?; + offset += key_length; + + // Read value + if buffer.len() < offset + 8 { + return Err("Buffer too short for value".into()); + } + let value = f64::from_le_bytes([ + buffer[offset], + buffer[offset + 1], + buffer[offset + 2], + buffer[offset + 3], + buffer[offset + 4], + buffer[offset + 5], + buffer[offset + 6], + buffer[offset + 7], + ]); + offset += 8; + + values.insert(key, value); + } + + Ok(Self { values }) + } +} + +impl SerializableToSink for MultipleMaxAccumulator { + fn serialize_to_json(&self) -> Value { + let mut values_obj = serde_json::Map::new(); + for (key, value) in &self.values { + let key_json = key.serialize_to_json(); + let key_str = serde_json::to_string(&key_json).unwrap(); + values_obj.insert( + key_str, + Value::Number(serde_json::Number::from_f64(*value).unwrap()), + ); + } + + serde_json::json!({ "values": values_obj }) + } + + fn serialize_to_bytes(&self) -> Vec { + let mut buffer = Vec::new(); + + // Write number of entries + buffer.extend_from_slice(&(self.values.len() as u32).to_le_bytes()); + + // Write each key-value pair + for (key, value) in &self.values { + let key_bytes = key.serialize_to_bytes(); + + // Write key length and data + buffer.extend_from_slice(&(key_bytes.len() as u32).to_le_bytes()); + buffer.extend_from_slice(&key_bytes); + + // Write value + buffer.extend_from_slice(&value.to_le_bytes()); + } + + buffer + } +} + +impl AggregateCore for MultipleMaxAccumulator { + fn clone_boxed_core(&self) -> Box { + Box::new(self.clone()) + } + + fn type_name(&self) -> &'static str { + "MultipleMaxAccumulator" + } + + fn as_any(&self) -> &dyn std::any::Any { + self + } + + fn as_any_mut(&mut self) -> &mut dyn std::any::Any { + self + } + + fn merge_with( + &self, + other: &dyn AggregateCore, + ) -> Result, Box> { + if other.get_accumulator_type() != self.get_accumulator_type() { + return Err(format!( + "Cannot merge MultipleMaxAccumulator with {}", + other.get_accumulator_type() + ) + .into()); + } + + let other_multiple = other + .as_any() + .downcast_ref::() + .ok_or("Failed to downcast to MultipleMaxAccumulator")?; + + let merged = Self::merge_accumulators(vec![self.clone(), other_multiple.clone()])?; + + Ok(Box::new(merged)) + } + + fn get_accumulator_type(&self) -> AggregationType { + AggregationType::MultipleMax + } + + fn approx_memory_bytes(&self) -> usize { + const BYTES_PER_ENTRY: usize = 96; + std::mem::size_of::() + self.values.len() * BYTES_PER_ENTRY + } + + fn get_keys(&self) -> Option> { + Some(self.values.keys().cloned().collect()) + } + + fn query_statistic( + &self, + statistic: asap_types::Statistic, + key: &Option, + query_kwargs: &std::collections::HashMap, + ) -> Result> { + use crate::storage_engines::types::MultipleSubpopulationAggregate; + let key_val = key + .as_ref() + .ok_or("Key required for MultipleMaxAccumulator")?; + self.query(statistic, key_val, Some(query_kwargs)) + } +} + +impl MultipleSubpopulationAggregate for MultipleMaxAccumulator { + fn query( + &self, + statistic: Statistic, + key: &KeyByLabelValues, + _query_kwargs: Option<&HashMap>, + ) -> Result> { + match statistic { + Statistic::Max => self + .values + .get(key) + .copied() + .ok_or_else(|| format!("Key {key} not found in MultipleMaxAccumulator").into()), + other => { + Err(format!("Unsupported statistic in MultipleMaxAccumulator: {other:?}").into()) + } + } + } + + fn clone_boxed(&self) -> Box { + Box::new(self.clone()) + } +} + +impl MergeableAccumulator for MultipleMaxAccumulator { + fn merge_accumulators( + accumulators: Vec, + ) -> Result> { + if accumulators.is_empty() { + return Err("No accumulators to merge".into()); + } + + let mut result = MultipleMaxAccumulator::new(); + + for acc in accumulators { + for (key, value) in acc.values { + result.update(key, value); + } + } + + Ok(result) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn key(value: &str) -> KeyByLabelValues { + KeyByLabelValues::new_with_labels(vec![value.to_string()]) + } + + #[test] + fn keeps_the_largest_per_key() { + let mut acc = MultipleMaxAccumulator::new(); + acc.update(key("a"), 10.0); + acc.update(key("a"), 5.0); + acc.update(key("a"), 15.0); + acc.update(key("b"), 7.0); + + assert_eq!(acc.query(Statistic::Max, &key("a"), None).unwrap(), 15.0); + assert_eq!(acc.query(Statistic::Max, &key("b"), None).unwrap(), 7.0); + } + + #[test] + fn refuses_the_opposite_statistic_and_unknown_keys() { + let mut acc = MultipleMaxAccumulator::new(); + acc.update(key("a"), 1.0); + assert!(acc.query(Statistic::Min, &key("a"), None).is_err()); + assert!(acc.query(Statistic::Max, &key("missing"), None).is_err()); + } + + #[test] + fn merges_per_key() { + let mut left = MultipleMaxAccumulator::new(); + left.update(key("a"), 10.0); + let mut right = MultipleMaxAccumulator::new(); + right.update(key("a"), 5.0); + right.update(key("b"), 3.0); + + let merged = >::merge_accumulators(vec![left, right]) + .unwrap(); + + assert_eq!(merged.query(Statistic::Max, &key("a"), None).unwrap(), 10.0); + assert_eq!(merged.query(Statistic::Max, &key("b"), None).unwrap(), 3.0); + } + + #[test] + fn refuses_to_merge_with_the_opposite_direction() { + use super::super::multiple_min_accumulator::MultipleMinAccumulator; + let mine = MultipleMaxAccumulator::new(); + let theirs = MultipleMinAccumulator::new(); + assert!(mine.merge_with(&theirs).is_err()); + } + + #[test] + fn round_trips_through_both_serializations() { + let mut acc = MultipleMaxAccumulator::new(); + acc.update(key("a"), 4.0); + + let json = acc.serialize_to_json(); + let from_json = MultipleMaxAccumulator::deserialize_from_json(&json).unwrap(); + assert_eq!( + from_json.query(Statistic::Max, &key("a"), None).unwrap(), + 4.0 + ); + + let bytes = acc.serialize_to_bytes(); + let from_bytes = MultipleMaxAccumulator::deserialize_from_bytes(&bytes).unwrap(); + assert_eq!( + from_bytes.query(Statistic::Max, &key("a"), None).unwrap(), + 4.0 + ); + } +} diff --git a/data_plane/src/precompute_engine/operators/multiple_min_accumulator.rs b/data_plane/src/precompute_engine/operators/multiple_min_accumulator.rs new file mode 100644 index 000000000..00c250402 --- /dev/null +++ b/data_plane/src/precompute_engine/operators/multiple_min_accumulator.rs @@ -0,0 +1,338 @@ +use crate::storage_engines::types::{ + AggregateCore, AggregationType, KeyByLabelValues, MergeableAccumulator, + MultipleSubpopulationAggregate, SerializableToSink, +}; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use std::collections::HashMap; + +use asap_types::Statistic; + +/// Exact per-key minimum over many populations, mergeable by comparison. +/// +/// The maximum direction is +/// [`MultipleMaxAccumulator`](super::multiple_max_accumulator::MultipleMaxAccumulator), +/// a separate type: these used to be one `MultipleMinMaxAccumulator` whose +/// direction lived in a `sub_type` string that every layer above had to carry +/// alongside the family. +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct MultipleMinAccumulator { + pub values: HashMap, +} + +impl MultipleMinAccumulator { + pub fn new() -> Self { + Self::default() + } + + pub fn new_with_values(values: HashMap) -> Self { + Self { values } + } + + pub fn update(&mut self, key: KeyByLabelValues, value: f64) { + let current = self.values.entry(key).or_insert(f64::INFINITY); + if value < *current { + *current = value; + } + } + + pub fn add_value(&mut self, key: KeyByLabelValues, value: f64) { + self.values.insert(key, value); + } + + pub fn deserialize_from_json(data: &Value) -> Result> { + let values_data = data["values"] + .as_object() + .ok_or("Missing or invalid 'values' field")?; + + let mut values = HashMap::new(); + for (key_str, value) in values_data { + let key_json: Value = serde_json::from_str(key_str)?; + let key = KeyByLabelValues::deserialize_from_json(&key_json)?; + let val = value.as_f64().ok_or("Invalid value")?; + values.insert(key, val); + } + + Ok(Self { values }) + } + + pub fn deserialize_from_bytes(buffer: &[u8]) -> Result> { + let mut offset = 0; + + // Read number of entries + if buffer.len() < 4 { + return Err("Buffer too short for entry count".into()); + } + let num_entries = u32::from_le_bytes([ + buffer[offset], + buffer[offset + 1], + buffer[offset + 2], + buffer[offset + 3], + ]) as usize; + offset += 4; + + let mut values = HashMap::new(); + + for _ in 0..num_entries { + // Read key length and data + if buffer.len() < offset + 4 { + return Err("Buffer too short for key length".into()); + } + let key_length = u32::from_le_bytes([ + buffer[offset], + buffer[offset + 1], + buffer[offset + 2], + buffer[offset + 3], + ]) as usize; + offset += 4; + + if buffer.len() < offset + key_length { + return Err("Buffer too short for key data".into()); + } + let key = + KeyByLabelValues::deserialize_from_bytes(&buffer[offset..offset + key_length])?; + offset += key_length; + + // Read value + if buffer.len() < offset + 8 { + return Err("Buffer too short for value".into()); + } + let value = f64::from_le_bytes([ + buffer[offset], + buffer[offset + 1], + buffer[offset + 2], + buffer[offset + 3], + buffer[offset + 4], + buffer[offset + 5], + buffer[offset + 6], + buffer[offset + 7], + ]); + offset += 8; + + values.insert(key, value); + } + + Ok(Self { values }) + } +} + +impl SerializableToSink for MultipleMinAccumulator { + fn serialize_to_json(&self) -> Value { + let mut values_obj = serde_json::Map::new(); + for (key, value) in &self.values { + let key_json = key.serialize_to_json(); + let key_str = serde_json::to_string(&key_json).unwrap(); + values_obj.insert( + key_str, + Value::Number(serde_json::Number::from_f64(*value).unwrap()), + ); + } + + serde_json::json!({ "values": values_obj }) + } + + fn serialize_to_bytes(&self) -> Vec { + let mut buffer = Vec::new(); + + // Write number of entries + buffer.extend_from_slice(&(self.values.len() as u32).to_le_bytes()); + + // Write each key-value pair + for (key, value) in &self.values { + let key_bytes = key.serialize_to_bytes(); + + // Write key length and data + buffer.extend_from_slice(&(key_bytes.len() as u32).to_le_bytes()); + buffer.extend_from_slice(&key_bytes); + + // Write value + buffer.extend_from_slice(&value.to_le_bytes()); + } + + buffer + } +} + +impl AggregateCore for MultipleMinAccumulator { + fn clone_boxed_core(&self) -> Box { + Box::new(self.clone()) + } + + fn type_name(&self) -> &'static str { + "MultipleMinAccumulator" + } + + fn as_any(&self) -> &dyn std::any::Any { + self + } + + fn as_any_mut(&mut self) -> &mut dyn std::any::Any { + self + } + + fn merge_with( + &self, + other: &dyn AggregateCore, + ) -> Result, Box> { + if other.get_accumulator_type() != self.get_accumulator_type() { + return Err(format!( + "Cannot merge MultipleMinAccumulator with {}", + other.get_accumulator_type() + ) + .into()); + } + + let other_multiple = other + .as_any() + .downcast_ref::() + .ok_or("Failed to downcast to MultipleMinAccumulator")?; + + let merged = Self::merge_accumulators(vec![self.clone(), other_multiple.clone()])?; + + Ok(Box::new(merged)) + } + + fn get_accumulator_type(&self) -> AggregationType { + AggregationType::MultipleMin + } + + fn approx_memory_bytes(&self) -> usize { + const BYTES_PER_ENTRY: usize = 96; + std::mem::size_of::() + self.values.len() * BYTES_PER_ENTRY + } + + fn get_keys(&self) -> Option> { + Some(self.values.keys().cloned().collect()) + } + + fn query_statistic( + &self, + statistic: asap_types::Statistic, + key: &Option, + query_kwargs: &std::collections::HashMap, + ) -> Result> { + use crate::storage_engines::types::MultipleSubpopulationAggregate; + let key_val = key + .as_ref() + .ok_or("Key required for MultipleMinAccumulator")?; + self.query(statistic, key_val, Some(query_kwargs)) + } +} + +impl MultipleSubpopulationAggregate for MultipleMinAccumulator { + fn query( + &self, + statistic: Statistic, + key: &KeyByLabelValues, + _query_kwargs: Option<&HashMap>, + ) -> Result> { + match statistic { + Statistic::Min => self + .values + .get(key) + .copied() + .ok_or_else(|| format!("Key {key} not found in MultipleMinAccumulator").into()), + other => { + Err(format!("Unsupported statistic in MultipleMinAccumulator: {other:?}").into()) + } + } + } + + fn clone_boxed(&self) -> Box { + Box::new(self.clone()) + } +} + +impl MergeableAccumulator for MultipleMinAccumulator { + fn merge_accumulators( + accumulators: Vec, + ) -> Result> { + if accumulators.is_empty() { + return Err("No accumulators to merge".into()); + } + + let mut result = MultipleMinAccumulator::new(); + + for acc in accumulators { + for (key, value) in acc.values { + result.update(key, value); + } + } + + Ok(result) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn key(value: &str) -> KeyByLabelValues { + KeyByLabelValues::new_with_labels(vec![value.to_string()]) + } + + #[test] + fn keeps_the_smallest_per_key() { + let mut acc = MultipleMinAccumulator::new(); + acc.update(key("a"), 10.0); + acc.update(key("a"), 5.0); + acc.update(key("a"), 15.0); + acc.update(key("b"), 7.0); + + assert_eq!(acc.query(Statistic::Min, &key("a"), None).unwrap(), 5.0); + assert_eq!(acc.query(Statistic::Min, &key("b"), None).unwrap(), 7.0); + } + + #[test] + fn refuses_the_opposite_statistic_and_unknown_keys() { + let mut acc = MultipleMinAccumulator::new(); + acc.update(key("a"), 1.0); + assert!(acc.query(Statistic::Max, &key("a"), None).is_err()); + assert!(acc.query(Statistic::Min, &key("missing"), None).is_err()); + } + + #[test] + fn merges_per_key() { + let mut left = MultipleMinAccumulator::new(); + left.update(key("a"), 10.0); + let mut right = MultipleMinAccumulator::new(); + right.update(key("a"), 5.0); + right.update(key("b"), 3.0); + + let merged = >::merge_accumulators(vec![left, right]) + .unwrap(); + + assert_eq!(merged.query(Statistic::Min, &key("a"), None).unwrap(), 5.0); + assert_eq!(merged.query(Statistic::Min, &key("b"), None).unwrap(), 3.0); + } + + #[test] + fn refuses_to_merge_with_the_opposite_direction() { + use super::super::multiple_max_accumulator::MultipleMaxAccumulator; + let mine = MultipleMinAccumulator::new(); + let theirs = MultipleMaxAccumulator::new(); + assert!(mine.merge_with(&theirs).is_err()); + } + + #[test] + fn round_trips_through_both_serializations() { + let mut acc = MultipleMinAccumulator::new(); + acc.update(key("a"), 4.0); + + let json = acc.serialize_to_json(); + let from_json = MultipleMinAccumulator::deserialize_from_json(&json).unwrap(); + assert_eq!( + from_json.query(Statistic::Min, &key("a"), None).unwrap(), + 4.0 + ); + + let bytes = acc.serialize_to_bytes(); + let from_bytes = MultipleMinAccumulator::deserialize_from_bytes(&bytes).unwrap(); + assert_eq!( + from_bytes.query(Statistic::Min, &key("a"), None).unwrap(), + 4.0 + ); + } +} diff --git a/data_plane/src/precompute_engine/operators/multiple_min_max_accumulator.rs b/data_plane/src/precompute_engine/operators/multiple_min_max_accumulator.rs deleted file mode 100644 index 46c9277d6..000000000 --- a/data_plane/src/precompute_engine/operators/multiple_min_max_accumulator.rs +++ /dev/null @@ -1,493 +0,0 @@ -use crate::storage_engines::types::{ - AggregateCore, AggregationType, KeyByLabelValues, MergeableAccumulator, - MultipleSubpopulationAggregate, SerializableToSink, -}; -use serde::{Deserialize, Serialize}; -use serde_json::Value; -use std::collections::HashMap; - -use asap_types::Statistic; - -/// Accumulator that maintains separate min/max values for multiple keys -/// Allows querying min/max for specific label combinations -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct MultipleMinMaxAccumulator { - pub values: HashMap, - pub sub_type: String, // "min" or "max" -} - -impl MultipleMinMaxAccumulator { - pub fn new(sub_type: String) -> Self { - if sub_type != "min" && sub_type != "max" { - panic!("sub_type must be 'min' or 'max'"); - } - - Self { - values: HashMap::new(), - sub_type, - } - } - - pub fn new_min() -> Self { - Self::new("min".to_string()) - } - - pub fn new_max() -> Self { - Self::new("max".to_string()) - } - - pub fn new_with_values(values: HashMap, sub_type: String) -> Self { - if sub_type != "min" && sub_type != "max" { - panic!("sub_type must be 'min' or 'max'"); - } - - Self { values, sub_type } - } - - pub fn update(&mut self, key: KeyByLabelValues, value: f64) { - match self.sub_type.as_str() { - "min" => { - let current = self.values.entry(key).or_insert(f64::INFINITY); - if value < *current { - *current = value; - } - } - "max" => { - let current = self.values.entry(key).or_insert(f64::NEG_INFINITY); - if value > *current { - *current = value; - } - } - _ => panic!("Invalid sub_type"), - } - } - - pub fn add_value(&mut self, key: KeyByLabelValues, value: f64) { - self.values.insert(key, value); - } - - pub fn deserialize_from_json(data: &Value) -> Result> { - let sub_type = data["sub_type"] - .as_str() - .ok_or("Missing or invalid 'sub_type' field")? - .to_string(); - - if sub_type != "min" && sub_type != "max" { - return Err("sub_type must be 'min' or 'max'".into()); - } - - let values_data = data["values"] - .as_object() - .ok_or("Missing or invalid 'values' field")?; - - let mut values = HashMap::new(); - for (key_str, value) in values_data { - let key_json: Value = serde_json::from_str(key_str)?; - let key = KeyByLabelValues::deserialize_from_json(&key_json)?; - let val = value.as_f64().ok_or("Invalid value")?; - values.insert(key, val); - } - - Ok(Self { values, sub_type }) - } - - pub fn deserialize_from_bytes( - buffer: &[u8], - sub_type: String, - ) -> Result> { - if sub_type != "min" && sub_type != "max" { - return Err("sub_type must be 'min' or 'max'".into()); - } - - let mut offset = 0; - - // Read number of entries - if buffer.len() < 4 { - return Err("Buffer too short for entry count".into()); - } - let num_entries = u32::from_le_bytes([ - buffer[offset], - buffer[offset + 1], - buffer[offset + 2], - buffer[offset + 3], - ]) as usize; - offset += 4; - - let mut values = HashMap::new(); - - for _ in 0..num_entries { - // Read key length and data - if buffer.len() < offset + 4 { - return Err("Buffer too short for key length".into()); - } - let key_length = u32::from_le_bytes([ - buffer[offset], - buffer[offset + 1], - buffer[offset + 2], - buffer[offset + 3], - ]) as usize; - offset += 4; - - if buffer.len() < offset + key_length { - return Err("Buffer too short for key data".into()); - } - let key = - KeyByLabelValues::deserialize_from_bytes(&buffer[offset..offset + key_length])?; - offset += key_length; - - // Read value - if buffer.len() < offset + 8 { - return Err("Buffer too short for value".into()); - } - let value = f64::from_le_bytes([ - buffer[offset], - buffer[offset + 1], - buffer[offset + 2], - buffer[offset + 3], - buffer[offset + 4], - buffer[offset + 5], - buffer[offset + 6], - buffer[offset + 7], - ]); - offset += 8; - - values.insert(key, value); - } - - Ok(Self { values, sub_type }) - } -} - -impl SerializableToSink for MultipleMinMaxAccumulator { - fn serialize_to_json(&self) -> Value { - let mut values_obj = serde_json::Map::new(); - for (key, value) in &self.values { - let key_json = key.serialize_to_json(); - let key_str = serde_json::to_string(&key_json).unwrap(); - values_obj.insert( - key_str, - Value::Number(serde_json::Number::from_f64(*value).unwrap()), - ); - } - - serde_json::json!({ - "values": values_obj, - "sub_type": self.sub_type - }) - } - - fn serialize_to_bytes(&self) -> Vec { - let mut buffer = Vec::new(); - - // Write number of entries - buffer.extend_from_slice(&(self.values.len() as u32).to_le_bytes()); - - // Write each key-value pair - for (key, value) in &self.values { - let key_bytes = key.serialize_to_bytes(); - - // Write key length and data - buffer.extend_from_slice(&(key_bytes.len() as u32).to_le_bytes()); - buffer.extend_from_slice(&key_bytes); - - // Write value - buffer.extend_from_slice(&value.to_le_bytes()); - } - - buffer - } -} - -impl AggregateCore for MultipleMinMaxAccumulator { - fn clone_boxed_core(&self) -> Box { - Box::new(self.clone()) - } - - fn type_name(&self) -> &'static str { - "MultipleMinMaxAccumulator" - } - - fn as_any(&self) -> &dyn std::any::Any { - self - } - - fn as_any_mut(&mut self) -> &mut dyn std::any::Any { - self - } - - fn merge_with( - &self, - other: &dyn AggregateCore, - ) -> Result, Box> { - // Check if other is also a MultipleMinMaxAccumulator - if other.get_accumulator_type() != self.get_accumulator_type() { - return Err(format!( - "Cannot merge MultipleMinMaxAccumulator with {}", - other.get_accumulator_type() - ) - .into()); - } - - // Downcast to MultipleMinMaxAccumulator - let other_multiple_minmax = other - .as_any() - .downcast_ref::() - .ok_or("Failed to downcast to MultipleMinMaxAccumulator")?; - - // Use the existing merge_accumulators method - let merged = Self::merge_accumulators(vec![self.clone(), other_multiple_minmax.clone()])?; - - Ok(Box::new(merged)) - } - - fn get_accumulator_type(&self) -> AggregationType { - AggregationType::MultipleMinMax - } - - fn approx_memory_bytes(&self) -> usize { - const BYTES_PER_ENTRY: usize = 96; - std::mem::size_of::() + self.values.len() * BYTES_PER_ENTRY + self.sub_type.capacity() - } - - fn get_keys(&self) -> Option> { - Some(self.values.keys().cloned().collect()) - } - - fn query_statistic( - &self, - statistic: asap_types::Statistic, - key: &Option, - query_kwargs: &std::collections::HashMap, - ) -> Result> { - use crate::storage_engines::types::MultipleSubpopulationAggregate; - let key_val = key - .as_ref() - .ok_or("Key required for MultipleMinMaxAccumulator")?; - self.query(statistic, key_val, Some(query_kwargs)) - } -} - -impl MultipleSubpopulationAggregate for MultipleMinMaxAccumulator { - fn query( - &self, - statistic: Statistic, - key: &KeyByLabelValues, - _query_kwargs: Option<&HashMap>, - ) -> Result> { - // Query specific key - match statistic { - Statistic::Min => { - if self.sub_type == "min" { - self.values.get(key).copied().ok_or_else(|| { - format!("Key {key} not found in MultipleMinMaxAccumulator").into() - }) - } else { - Err("Cannot query Min statistic from Max accumulator".into()) - } - } - Statistic::Max => { - if self.sub_type == "max" { - self.values.get(key).copied().ok_or_else(|| { - format!("Key {key} not found in MultipleMinMaxAccumulator").into() - }) - } else { - Err("Cannot query Max statistic from Min accumulator".into()) - } - } - _ => Err( - format!("Unsupported statistic in MultipleMinMaxAccumulator: {statistic:?}").into(), - ), - } - } - - fn clone_boxed(&self) -> Box { - Box::new(self.clone()) - } -} - -impl MergeableAccumulator for MultipleMinMaxAccumulator { - fn merge_accumulators( - accumulators: Vec, - ) -> Result> { - if accumulators.is_empty() { - return Err("No accumulators to merge".into()); - } - - let sub_type = accumulators[0].sub_type.clone(); - - // Verify all accumulators have the same sub_type - for acc in &accumulators { - if acc.sub_type != sub_type { - return Err("Cannot merge accumulators with different sub_types".into()); - } - } - - let mut result = MultipleMinMaxAccumulator::new(sub_type.clone()); - - for acc in accumulators { - for (key, value) in acc.values { - match result.values.get(&key) { - Some(existing_value) => match sub_type.as_str() { - "min" => { - if value < *existing_value { - result.values.insert(key, value); - } - } - "max" => { - if value > *existing_value { - result.values.insert(key, value); - } - } - _ => unreachable!(), - }, - None => { - result.values.insert(key, value); - } - } - } - } - - Ok(result) - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_multiple_min_max_accumulator_creation() { - let min_acc = MultipleMinMaxAccumulator::new_min(); - assert_eq!(min_acc.sub_type, "min"); - assert!(min_acc.values.is_empty()); - - let max_acc = MultipleMinMaxAccumulator::new_max(); - assert_eq!(max_acc.sub_type, "max"); - assert!(max_acc.values.is_empty()); - } - - #[test] - fn test_multiple_min_accumulator_update() { - let mut acc = MultipleMinMaxAccumulator::new_min(); - - let key1 = KeyByLabelValues::new_with_labels(vec!["web".to_string()]); - let key2 = KeyByLabelValues::new_with_labels(vec!["api".to_string()]); - - acc.update(key1.clone(), 10.0); - acc.update(key1.clone(), 5.0); // Should update to smaller value - acc.update(key1.clone(), 15.0); // Should not update (larger) - acc.update(key2.clone(), 20.0); - - assert_eq!(acc.values.get(&key1), Some(&5.0)); - assert_eq!(acc.values.get(&key2), Some(&20.0)); - } - - #[test] - fn test_multiple_max_accumulator_update() { - let mut acc = MultipleMinMaxAccumulator::new_max(); - - let key = KeyByLabelValues::new_with_labels(vec!["web".to_string()]); - - acc.update(key.clone(), 10.0); - acc.update(key.clone(), 5.0); // Should not update (smaller) - acc.update(key.clone(), 15.0); // Should update to larger value - - assert_eq!(acc.values.get(&key), Some(&15.0)); - } - - #[test] - fn test_multiple_min_max_accumulator_query() { - let mut min_acc = MultipleMinMaxAccumulator::new_min(); - let mut max_acc = MultipleMinMaxAccumulator::new_max(); - - let key = KeyByLabelValues::new_with_labels(vec!["web".to_string()]); - - min_acc.add_value(key.clone(), 5.0); - max_acc.add_value(key.clone(), 15.0); - - // Test queries with the specific key - assert_eq!( - crate::MultipleSubpopulationAggregate::query(&min_acc, Statistic::Min, &key, None) - .unwrap(), - 5.0 - ); - assert_eq!( - crate::MultipleSubpopulationAggregate::query(&max_acc, Statistic::Max, &key, None) - .unwrap(), - 15.0 - ); - - // Test error cases - assert!( - crate::MultipleSubpopulationAggregate::query(&min_acc, Statistic::Max, &key, None) - .is_err() - ); - assert!( - crate::MultipleSubpopulationAggregate::query(&max_acc, Statistic::Min, &key, None) - .is_err() - ); - assert!( - crate::MultipleSubpopulationAggregate::query(&min_acc, Statistic::Sum, &key, None) - .is_err() - ); - } - - #[test] - fn test_multiple_min_max_accumulator_merge() { - let mut acc1 = MultipleMinMaxAccumulator::new_min(); - let mut acc2 = MultipleMinMaxAccumulator::new_min(); - - let key1 = KeyByLabelValues::new_with_labels(vec!["web".to_string()]); - - let key2 = KeyByLabelValues::new_with_labels(vec!["api".to_string()]); - - acc1.add_value(key1.clone(), 10.0); - acc1.add_value(key2.clone(), 20.0); - - acc2.add_value(key1.clone(), 5.0); // Smaller value, should be used - - let merged = >::merge_accumulators(vec![acc1, acc2]) - .unwrap(); - - assert_eq!(merged.values.get(&key1), Some(&5.0)); // Should use smaller value - assert_eq!(merged.values.get(&key2), Some(&20.0)); // Should be preserved - } - - #[test] - fn test_multiple_min_max_accumulator_serialization() { - let mut acc = MultipleMinMaxAccumulator::new_min(); - - let key = KeyByLabelValues::new_with_labels(vec!["web".to_string()]); - - acc.add_value(key.clone(), 42.5); - - // Test JSON serialization - let json = acc.serialize_to_json(); - let deserialized = MultipleMinMaxAccumulator::deserialize_from_json(&json).unwrap(); - assert_eq!(deserialized.values.get(&key), Some(&42.5)); - assert_eq!(deserialized.sub_type, "min"); - - // Test byte serialization - let bytes = acc.serialize_to_bytes(); - let deserialized_bytes = - MultipleMinMaxAccumulator::deserialize_from_bytes(&bytes, "min".to_string()).unwrap(); - assert_eq!(deserialized_bytes.values.get(&key), Some(&42.5)); - assert_eq!(deserialized_bytes.sub_type, "min"); - } - - #[test] - fn test_trait_object() { - let mut acc = MultipleMinMaxAccumulator::new_min(); - - let key = KeyByLabelValues::new_with_labels(vec!["web".to_string()]); - - acc.add_value(key.clone(), 42.0); - - let trait_obj: Box = Box::new(acc); - - // Test type name through trait object - assert_eq!(trait_obj.type_name(), "MultipleMinMaxAccumulator"); - } -} diff --git a/data_plane/src/precompute_engine/subdag_scheduler.rs b/data_plane/src/precompute_engine/subdag_scheduler.rs index 69d67c401..918f84779 100644 --- a/data_plane/src/precompute_engine/subdag_scheduler.rs +++ b/data_plane/src/precompute_engine/subdag_scheduler.rs @@ -328,6 +328,8 @@ mod tests { binary.payload = ExecutableOperatorPayload::Binary { timing: planner_types::post_asap::ExecutionTiming::MaintenanceTime, operator: BinaryOperator { + checked_relative_division: false, + checked_finite_division: false, kind: BinaryOpKind::Arithmetic(ArithmeticOpKind::Sub), vector_match: None, }, diff --git a/data_plane/src/query_engines/asap_query_engine/catalog_resolver.rs b/data_plane/src/query_engines/asap_query_engine/catalog_resolver.rs index 3009c68cc..8b06d6bd7 100644 --- a/data_plane/src/query_engines/asap_query_engine/catalog_resolver.rs +++ b/data_plane/src/query_engines/asap_query_engine/catalog_resolver.rs @@ -55,8 +55,10 @@ impl ResolvedMaterialization<'_> { | AggregationType::MultipleSum | AggregationType::Increase | AggregationType::MultipleIncrease - | AggregationType::MinMax - | AggregationType::MultipleMinMax, + | AggregationType::Min + | AggregationType::Max + | AggregationType::MultipleMin + | AggregationType::MultipleMax, .. } ) @@ -80,10 +82,11 @@ impl ResolvedMaterialization<'_> { ExactReadout::Increase | ExactReadout::Rate => { matches!(aggregation_type, Increase | MultipleIncrease) } - ExactReadout::Max => { - matches!(aggregation_type, MinMax | MultipleMinMax) - && aggregation_sub_type.eq_ignore_ascii_case("max") - } + // Direction is the family now -- no `aggregation_sub_type` + // cross-check, and a minimum summary can no longer be + // offered up for a maximum readout. + ExactReadout::Min => matches!(aggregation_type, Min | MultipleMin), + ExactReadout::Max => matches!(aggregation_type, Max | MultipleMax), }, QueryPlanNode::SummaryEstimate { query, .. } => match query { QueryReadout::Quantile { q } => { diff --git a/data_plane/src/query_engines/asap_query_engine/engine.rs b/data_plane/src/query_engines/asap_query_engine/engine.rs index 516b518d7..c22bb1fc5 100644 --- a/data_plane/src/query_engines/asap_query_engine/engine.rs +++ b/data_plane/src/query_engines/asap_query_engine/engine.rs @@ -1451,7 +1451,8 @@ mod sketch_query_tests { mod aux_pushdown_tests { use super::*; use crate::precompute_engine::operators::{ - min_max_accumulator::MinMaxAccumulator, sum_accumulator::SumAccumulator, + max_accumulator::MaxAccumulator, min_accumulator::MinAccumulator, + sum_accumulator::SumAccumulator, }; use crate::storage_engines::types::AggregationType; use asap_types::Statistic; @@ -1610,8 +1611,8 @@ mod aux_pushdown_tests { #[test] fn real_min_max_accumulator_uses_aux_fast_path() { let engine = make_engine(); - let min_acc = MinMaxAccumulator::with_value(3.0, "min".to_string()); - let max_acc = MinMaxAccumulator::with_value(99.0, "max".to_string()); + let min_acc = MinAccumulator::with_value(3.0); + let max_acc = MaxAccumulator::with_value(99.0); assert_eq!( engine .query_precompute_for_statistic(&min_acc, &Statistic::Min, &None, &HashMap::new()) @@ -2039,8 +2040,8 @@ mod asap_tier_classify_tests { /// REGRESSION of the HLL `count(metric)` "No result" e2e failure /// (`controller_plan_to_query_full_roundtrip_hll`) isolated to the - /// engine layer. `count(unique_users_per_min)` is the distinct-count - /// idiom. The Planner DAG represents this as a cardinality readout, + /// engine layer. `count(distinct_over_time(unique_users_per_min[w]))` + /// is the distinct-count idiom (bare `count(v)` is a row count). The Planner DAG represents this as a cardinality readout, /// so the executor returns the HLL distinct-count directly. A single /// FULL HLL frame (~500 users) is used so /// the instant projection reads the real estimate. @@ -2065,14 +2066,17 @@ mod asap_tier_classify_tests { ); let engine = build_engine_with_index(idx); - let result = engine.execute("count(unique_users_per_min)").await.expect( - "count(hll_metric) must dispatch to the Cardinality family \ - via the candidate capability (empty trace function) and \ - return the HLL distinct-count, NOT capability-miss", - ); + let result = engine + .execute("count(distinct_over_time(unique_users_per_min[1m]))") + .await + .expect( + "the distinct-count idiom must dispatch to the Cardinality \ + family via the candidate capability (empty trace function) \ + and return the HLL distinct-count, NOT capability-miss", + ); assert!( result_nonempty(&result), - "count(unique_users_per_min) over an HLL sid must return a \ + "the distinct-count idiom over an HLL sid must return a \ non-empty cardinality estimate (regression: empty `asap_query` \ No-result)" ); @@ -2165,9 +2169,9 @@ mod asap_tier_classify_tests { let engine = build_engine_with_index(idx); let result = engine - .execute("count(unique_users_global)") + .execute("count(distinct_over_time(unique_users_global[1m]))") .await - .expect("global count(hll_metric) must answer, not capability-miss"); + .expect("global distinct count over HLL sids must answer, not capability-miss"); // GLOBAL distinct is a single scalar — exactly one element. let est = match &result { @@ -2175,7 +2179,7 @@ mod asap_tier_classify_tests { assert_eq!( v.values.len(), 1, - "global count() must collapse to ONE merged estimate, got {} \ + "a global distinct count must collapse to ONE merged estimate, got {} \ (per-series leak): {v:?}", v.values.len() ); diff --git a/data_plane/src/query_engines/asap_query_engine/live_serve.rs b/data_plane/src/query_engines/asap_query_engine/live_serve.rs index ccfb7333a..00033b002 100644 --- a/data_plane/src/query_engines/asap_query_engine/live_serve.rs +++ b/data_plane/src/query_engines/asap_query_engine/live_serve.rs @@ -498,20 +498,28 @@ mod tests { // `result.is_none()`: the grouping-ambiguity gate declined this // shape because an empty `by` couldn't be told apart from "reduce // everything" (ASAPController#163). With `Reduction` (#165) the - // executor resolves it -- `count(...)` lowers to `Reduce([])`, both - // sids share one group key, and the new path serves the correctly - // merged answer instead of falling back. + // executor resolves it -- the outer aggregator lowers to + // `Reduce([])`, both sids share one group key, and the new path + // serves the correctly merged answer instead of falling back. + // + // The distinct-count idiom is `count(distinct_over_time(v[w]))`; + // bare `count(v)` is a row count upstream. let idx = SketchStore::new(); register_hll(&idx, 1, "svc-a", &["a", "b", "c"]); register_hll(&idx, 2, "svc-b", &["d", "e", "f"]); - let result = - try_serve_from_summary_executor(&idx, "count(unique_users)", 1_000, 2_000, true); + let result = try_serve_from_summary_executor( + &idx, + "count(distinct_over_time(unique_users[1m]))", + 1_000, + 2_000, + true, + ); let result = result .expect("global-merge shape is no longer ambiguous -- it must be served, not declined"); assert_eq!( result.series.len(), 1, - "a by-less count() must merge both sids into ONE series, got {:?}", + "a by-less distinct count must merge both sids into ONE series, got {:?}", result.series ); // Disjoint item sets {a,b,c} + {d,e,f} -> merged cardinality ~6. diff --git a/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs b/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs index b91738c34..1c7d80f3d 100644 --- a/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs +++ b/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs @@ -1097,21 +1097,31 @@ mod tests { // shape at all. // // With `Reduction` (ASAPController#165) that ambiguity is gone: - // `count(...)` is a genuine aggregation operator, so it lowers to + // the outer aggregator is a genuine reduction, so it lowers to // `Reduce([])` and `resolve_group_key` gives every candidate the // SAME group key -- the two sids MERGE into one answer, which is // what the query actually asked for. No gate, no fallback. + // + // The distinct-count idiom is `count(distinct_over_time(v[w]))`; + // bare `count(v)` is a row count upstream and an HLL sid rightly + // cannot serve it. let idx = SketchStore::new(); register_hll(&idx, 1, "svc-a", &["a", "b", "c"]); register_hll(&idx, 2, "svc-b", &["d", "e", "f"]); - let outcome = - execute_post_asap_readout(&idx, "count(unique_users)", 1_000, 2_000, true, accuracy()) - .expect("should execute"); + let outcome = execute_post_asap_readout( + &idx, + "count(distinct_over_time(unique_users[1m]))", + 1_000, + 2_000, + true, + accuracy(), + ) + .expect("should execute"); assert_eq!( outcome.series.len(), 1, - "a by-less count() is a full reduction -- both HLL sids must merge into ONE \ - series, not stay split (and not be declined), got {:?}", + "a by-less distinct count is a full reduction -- both HLL sids must merge into \ + ONE series, not stay split (and not be declined), got {:?}", outcome.series ); // Disjoint item sets {a,b,c} + {d,e,f} -> merged cardinality ~6. diff --git a/data_plane/src/query_engines/asap_query_engine/summary_exec.rs b/data_plane/src/query_engines/asap_query_engine/summary_exec.rs index 672363d12..ff0ef957c 100644 --- a/data_plane/src/query_engines/asap_query_engine/summary_exec.rs +++ b/data_plane/src/query_engines/asap_query_engine/summary_exec.rs @@ -548,6 +548,8 @@ mod tests { lhs: child.clone(), rhs: child.clone(), operator: planner_types::post_asap::BinaryOperator { + checked_relative_division: false, + checked_finite_division: false, kind: planner_types::pre_asap::BinaryOpKind::Arithmetic( planner_types::pre_asap::ArithmeticOpKind::Div, ), diff --git a/data_plane/src/query_engines/asap_query_engine/summary_executor.rs b/data_plane/src/query_engines/asap_query_engine/summary_executor.rs index 0b3583e42..33ee946ff 100644 --- a/data_plane/src/query_engines/asap_query_engine/summary_executor.rs +++ b/data_plane/src/query_engines/asap_query_engine/summary_executor.rs @@ -43,7 +43,7 @@ //! //! `find_candidates`/`fetch_state`/`merge_states` ALSO recognize //! `AggKind::ExactAgg` sids for `ExactKind::{Sum, Increase}` (see -//! `exact_agg_kind_match`'s doc for why `MinMax`/`Count`/`Rate` aren't +//! `exact_agg_kind_match`'s doc for why `Min`/`Max`/`Count`/`Rate` aren't //! matched) — one sid is one aggregation, read out directly, with no //! special-casing of exact-vs-approximate at the `find_candidates`/merge //! level. But `readout`/`SketchQuery` NEVER see these: `asap_aware_mapping::bind` @@ -71,7 +71,8 @@ use planner_types::post_asap::{ use planner_types::pre_asap::{ColumnId, ColumnRef, QueryExpr, Reduction, Source}; use crate::precompute_engine::operators::increase_accumulator::IncreaseAccumulator; -use crate::precompute_engine::operators::min_max_accumulator::MinMaxAccumulator; +use crate::precompute_engine::operators::max_accumulator::MaxAccumulator; +use crate::precompute_engine::operators::min_accumulator::MinAccumulator; use crate::storage_engines::sketch_db::data::{AggKind, SketchConfig, SketchTimeSeries}; use crate::storage_engines::sketch_db::index::{SketchSampleState, SketchStore}; use crate::storage_engines::sketch_db::query::delta_apply::{ @@ -194,7 +195,7 @@ impl GroupState { /// module's doc for why `readout()`/`SketchQuery` never see these). /// /// `None` for a `Sketch` state, a group with no windows in range, or - /// a merge/query failure. `AggregationType::MinMax` (and any other + /// a merge/query failure. `AggregationType::Min`/`Max` (and any other /// type `exact_agg_kind_match` doesn't match) can't reach a /// `GroupState::ExactAgg` via `find_candidates` in the first place — /// the fallback arm here is defensive, not a real path. @@ -252,9 +253,13 @@ impl GroupState { asap_types::query_plan::ExactReadout::Rate, AggregationType::Increase | AggregationType::MultipleIncrease, ) => asap_types::Statistic::Rate, + ( + asap_types::query_plan::ExactReadout::Min, + AggregationType::Min | AggregationType::MultipleMin, + ) => asap_types::Statistic::Min, ( asap_types::query_plan::ExactReadout::Max, - AggregationType::MinMax | AggregationType::MultipleMinMax, + AggregationType::Max | AggregationType::MultipleMax, ) => asap_types::Statistic::Max, _ => return None, }; @@ -283,7 +288,24 @@ impl GroupState { } if matches!( agg_type, - AggregationType::MinMax | AggregationType::MultipleMinMax + AggregationType::Min | AggregationType::MultipleMin + ) && readout == asap_types::query_plan::ExactReadout::Min + { + return entries + .iter() + .flat_map(|windows| windows.values()) + .map(|acc| { + acc.as_any() + .downcast_ref::() + .map(|a| a.value) + }) + .collect::>>()? + .into_iter() + .reduce(f64::min); + } + if matches!( + agg_type, + AggregationType::Max | AggregationType::MultipleMax ) && readout == asap_types::query_plan::ExactReadout::Max { return entries @@ -291,7 +313,7 @@ impl GroupState { .flat_map(|windows| windows.values()) .map(|acc| { acc.as_any() - .downcast_ref::() + .downcast_ref::() .map(|a| a.value) }) .collect::>>()? @@ -569,16 +591,21 @@ impl QueryExecutionContext<'_> { )); } } - if matches!( - agg_type, - AggregationType::MinMax | AggregationType::MultipleMinMax - ) { - if let Some(series) = self.index.query_rollup_range( + if let Some((reduction, is_min)) = match agg_type { + AggregationType::Min | AggregationType::MultipleMin => Some(( + crate::storage_engines::sketch_db::index::RollupReduction::Min, + true, + )), + AggregationType::Max | AggregationType::MultipleMax => Some(( crate::storage_engines::sketch_db::index::RollupReduction::Max, - sid, - self.t0_ms, - self.t1_ms, - ) { + false, + )), + _ => None, + } { + if let Some(series) = self + .index + .query_rollup_range(reduction, sid, self.t0_ms, self.t1_ms) + { for (labels, value) in series { let key = match &binding.output_grouping { PhysicalGrouping::PerEntity => labels, @@ -586,12 +613,15 @@ impl QueryExecutionContext<'_> { project_group_key(keys, &labels) } }; - let accumulator = - MinMaxAccumulator::with_value(value, "max".to_string()); + let accumulator: Arc = if is_min { + Arc::new(MinAccumulator::with_value(value)) + } else { + Arc::new(MaxAccumulator::with_value(value)) + }; by_group.entry(key).or_default().push(GroupState::ExactAgg { entries: vec![Rc::new(BTreeMap::from([( self.t1_ms as i64, - Arc::new(accumulator) as Arc, + accumulator, )]))], agg_type, }); @@ -1207,10 +1237,14 @@ fn summary_family_matches_sketch( /// -> ExactKind::Increase` — confirmed against that module's own /// dispatch table rather than invented here). /// -/// `ExactKind::Count`/`Rate`/`MinMax` are not matched by this legacy -/// family-discovery path because their final operation is ambiguous from the -/// stored accumulator alone. Installed QueryPlans carry an explicit -/// `ExactReadout`, and `read_bound_materialization` serves those forms safely. +/// `ExactKind::Count`/`Rate`/`Min`/`Max` are not matched by this legacy +/// family-discovery path. For `Count`/`Rate` the final operation is ambiguous +/// from the stored accumulator alone. `Min`/`Max` were excluded for a reason +/// that no longer holds -- direction used to be unrecoverable once a summary +/// reached `AggKind::ExactAgg`, and is now the family itself -- but admitting +/// them here widens candidate discovery beyond the family split and is left +/// as follow-up. Installed QueryPlans carry an explicit `ExactReadout`, and +/// `read_bound_materialization` serves those forms safely. fn summary_family_matches_exact(family: &SummaryFamilyType, agg_type: AggregationType) -> bool { matches!( (family, agg_type), @@ -3188,26 +3222,26 @@ mod tests { } #[test] - fn minmax_exactagg_sid_is_not_matched() { - // `ExactKind::MinMax` is deliberately NOT matched against - // ExactAgg sids (see `exact_agg_kind_match`'s doc: no direction - // info survives to `AggKind::ExactAgg`) -- must fail over as - // NoCandidates, not silently guess a direction. + fn max_exactagg_sid_is_not_matched() { + // `ExactKind::Max` is deliberately NOT matched against ExactAgg + // sids by the legacy family-discovery path (see + // `summary_family_matches_exact`'s doc) -- must fail over as + // NoCandidates rather than widen discovery here. let idx = SketchStore::new(); let sid = 1u64; let mut meta = sum_exact_agg_meta(sid, "latency_max_ms", &[]); meta.agg_kind = crate::storage_engines::sketch_db::index::AggKind::ExactAgg { - agg_type: asap_types::AggregationType::MinMax, + agg_type: asap_types::AggregationType::Max, parameters_canonical: String::new(), spatial_filter_canonical: String::new(), }; - meta.capability = Some(Capability::ExactAgg(asap_types::AggregationType::MinMax)); + meta.capability = Some(Capability::ExactAgg(asap_types::AggregationType::Max)); idx.register(meta); idx.append_precompute( sid, BTreeMap::new(), (T0, T0 + 1000), - Box::new(crate::precompute_engine::operators::MinMaxAccumulator::new_min()), + Box::new(crate::precompute_engine::operators::MaxAccumulator::new()), ); let child = scan_node("latency_max_ms", None); @@ -3215,8 +3249,8 @@ mod tests { expr: SummaryExpr::SummaryAgg { child, family: SummaryFamilyType::ExactAggregate( - planner_types::post_asap::ExactKind::MinMax, - planner_types::post_asap::ExactParams::MinMax, + planner_types::post_asap::ExactKind::Max, + planner_types::post_asap::ExactParams::Max, ), input: planner_types::post_asap::SummaryUpdate { item: None, diff --git a/data_plane/src/storage_engines/sketch_db/accuracy.rs b/data_plane/src/storage_engines/sketch_db/accuracy.rs index 6134e7695..e89a6ce5c 100644 --- a/data_plane/src/storage_engines/sketch_db/accuracy.rs +++ b/data_plane/src/storage_engines/sketch_db/accuracy.rs @@ -95,10 +95,12 @@ impl BackendAccuracyProfile for AccuracyProfile { // here too before its retirement.) AggregationType::Sum | AggregationType::Increase - | AggregationType::MinMax + | AggregationType::Min + | AggregationType::Max | AggregationType::MultipleSum | AggregationType::MultipleIncrease - | AggregationType::MultipleMinMax => Self::exact(), + | AggregationType::MultipleMin + | AggregationType::MultipleMax => Self::exact(), // CountMinSketch: classic Cormode-Muthukrishnan bound. // ε = e/w, δ = 1/2^d with w = width, d = depth. We @@ -449,7 +451,11 @@ mod tests { #[test] fn min_max_increase_are_exact() { - for t in [AggregationType::MinMax, AggregationType::Increase] { + for t in [ + AggregationType::Min, + AggregationType::Max, + AggregationType::Increase, + ] { let p = AccuracyProfile::derive(&base_config(t, HashMap::new())); assert_eq!(p.kind, AccuracyKind::Exact); } diff --git a/data_plane/src/storage_engines/sketch_db/index/mod.rs b/data_plane/src/storage_engines/sketch_db/index/mod.rs index 80114880d..450372abf 100644 --- a/data_plane/src/storage_engines/sketch_db/index/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs @@ -96,7 +96,7 @@ fn reconstruct_exact_agg( bytes: &[u8], ) -> Option> { use crate::precompute_engine::operators::{ - IncreaseAccumulator, MinMaxAccumulator, MultipleIncreaseAccumulator, + IncreaseAccumulator, MaxAccumulator, MinAccumulator, MultipleIncreaseAccumulator, MultipleSumAccumulator, SumAccumulator, }; use crate::storage_engines::types::AggregateCore; @@ -107,7 +107,10 @@ fn reconstruct_exact_agg( "IncreaseAccumulator" => IncreaseAccumulator::deserialize_from_bytes(bytes) .ok() .map(|a| Box::new(a) as Box), - "MinMaxAccumulator" => MinMaxAccumulator::deserialize_from_bytes(bytes) + "MinAccumulator" => MinAccumulator::deserialize_from_bytes(bytes) + .ok() + .map(|a| Box::new(a) as Box), + "MaxAccumulator" => MaxAccumulator::deserialize_from_bytes(bytes) .ok() .map(|a| Box::new(a) as Box), "MultipleSumAccumulator" => MultipleSumAccumulator::deserialize_from_bytes(bytes) @@ -116,9 +119,8 @@ fn reconstruct_exact_agg( "MultipleIncreaseAccumulator" => MultipleIncreaseAccumulator::deserialize_from_bytes(bytes) .ok() .map(|a| Box::new(a) as Box), - // `MultipleMinMaxAccumulator` needs an external `sub_type` - // (min/max) not recorded in the part, and the sketch-backed - // accumulator forms have no generic byte factory — both are left + // The keyed `MultipleMin`/`MultipleMax` forms and the + // sketch-backed accumulators have no generic byte factory — left // to the deferred exact-agg/sketch precompute read-back work (see // PR follow-up note). They are still served from memory; only the // evicted-to-disk portion is skipped for these types. @@ -221,7 +223,7 @@ pub struct SketchInstanceMetadata { /// M2.3 — the canonical "what kind of aggregation lives at this /// sid" descriptor. Replaces the M2-era `sketch_kind` + /// `sketch_config` field pair so a single registry can host both - /// sketches and partial-accumulator (Sum/Count/Avg/Rate/MinMax) + /// sketches and partial-accumulator (Sum/Count/Avg/Rate/Min/Max) /// state. pub agg_kind: AggKind, /// Approximate accuracy bound — `Some` for sketch-backed sids, @@ -460,12 +462,14 @@ impl ReductionRollupSeries { /// belong here rather than as additional top-level `SketchStore` fields. #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] pub enum RollupReduction { + Min, Max, } impl RollupReduction { fn combine(self, left: f64, right: f64) -> f64 { match self { + Self::Min => left.min(right), Self::Max => left.max(right), } } @@ -1468,7 +1472,7 @@ impl SketchStore { data } - /// Append a window's exact-aggregation (Sum/Count/Avg/Rate/MinMax) + /// Append a window's exact-aggregation (Sum/Count/Avg/Rate/Min/Max) /// state under `sid`. Mirror of [`Self::append_sample`] for the /// exact-agg branch — Phase 5 M2.3.3. /// @@ -1516,11 +1520,21 @@ impl SketchStore { return false; } let _mutation = self.begin_state_mutation(); - let max_value = payload + // Extremum state feeds the derived rollup series that serves + // `min_over_time` / `max_over_time` without walking every pane. + // Both directions are their own accumulator type, so the reduction + // follows from the payload's type rather than from a `sub_type` + // string that had to agree with it. + let rollup_value = payload .as_any() - .downcast_ref::() - .filter(|acc| acc.sub_type == "max") - .map(|acc| acc.value); + .downcast_ref::() + .map(|acc| (RollupReduction::Min, acc.value)) + .or_else(|| { + payload + .as_any() + .downcast_ref::() + .map(|acc| (RollupReduction::Max, acc.value)) + }); let store = self .series .entry(sid) @@ -1535,9 +1549,11 @@ impl SketchStore { guard.last_write_unix_ms = now_ms(); let retention_horizon_ms = guard.retention_horizon_ms; drop(guard); - if let Some(value) = max_value.filter(|_| self.persistence_read.read().unwrap().is_none()) { + if let Some((reduction, value)) = + rollup_value.filter(|_| self.persistence_read.read().unwrap().is_none()) + { self.rollups.append( - RollupReduction::Max, + reduction, sid, series_label_values, window, diff --git a/data_plane/src/storage_engines/sketch_db/sds.rs b/data_plane/src/storage_engines/sketch_db/sds.rs index 38a3d98de..aece4a0c8 100644 --- a/data_plane/src/storage_engines/sketch_db/sds.rs +++ b/data_plane/src/storage_engines/sketch_db/sds.rs @@ -357,7 +357,7 @@ mod tests { .bind(metadata(1, "cpu", "", AggregationType::Sum, 7)) .unwrap(); let b = registry - .bind(metadata(2, "cpu", "", AggregationType::MinMax, 7)) + .bind(metadata(2, "cpu", "", AggregationType::Max, 7)) .unwrap(); assert!(!Arc::ptr_eq(&a.summary_descriptor, &b.summary_descriptor)); @@ -407,13 +407,7 @@ mod tests { registry.install_catalog(Arc::new(catalog)).unwrap(); let binding = registry - .bind(metadata( - 1, - "wrong-local-copy", - "", - AggregationType::MinMax, - 7, - )) + .bind(metadata(1, "wrong-local-copy", "", AggregationType::Max, 7)) .unwrap(); assert_eq!(binding.summary_descriptor.as_ref(), &summary); assert_eq!(binding.data_descriptor.as_ref(), &data); From 24f49df7b82675ba07b76066e4ede086eb7632f5 Mon Sep 17 00:00:00 2001 From: zz_y Date: Sun, 13 Sep 2026 20:19:01 -0600 Subject: [PATCH 2/2] fix(planning): preserve guarded division fallback semantics Before: Planner division guards were discarded when lowering average rewrites and relative divisions into installed binary nodes. After: use a verified original exact subtree where available, or request exact fallback. Remove query placements pruned by exact subtree execution. Regression reproduced before the fix. Validation: 787 control-plane, 1200 data-plane, and 107 shared-type unit tests passed; formatting and clippy for both planes with all targets and warnings denied passed. --- control_plane/src/physical/compiler.rs | 4 ++ control_plane/src/query_plan.rs | 94 ++++++++++++++++++++++++++ 2 files changed, 98 insertions(+) diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index 526493fea..e0b5f23ce 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -1690,6 +1690,10 @@ impl PhysicalCompiler { // backend-local range index leaf. crate::query_plan::logical::finalize_residuals(&mut entry)?; } + // An exact subtree can absorb guarded arithmetic and prune its + // children. Those semantic nodes no longer have local query placements. + query_node_bindings + .retain(|(index, _), node| *index != query_index || entry.nodes.contains_key(node)); if metricsql { entry.language = crate::query_plan::QueryLanguage::MetricsQl; } diff --git a/control_plane/src/query_plan.rs b/control_plane/src/query_plan.rs index ce1f059a3..cb7ac7519 100644 --- a/control_plane/src/query_plan.rs +++ b/control_plane/src/query_plan.rs @@ -307,6 +307,13 @@ where self.next_id += 1; self.seen.insert(identity, id); let residual = match (&self.logical_source, &node.expr) { + (Some(original), SummaryExpr::BinaryOp { operator, .. }) + if operator.checked_relative_division || operator.checked_finite_division => + { + // Use the original exact subtree when its semantic identity can + // be proved; otherwise the explicit fallback below retries the query. + logical::selected_residual_nodes(original, node).ok() + } (Some(original), SummaryExpr::KeepPreAsap(expr)) => { Some(logical::residual_nodes(original, expr)?) } @@ -330,6 +337,15 @@ where } let physical = match &node.expr { + // Installed binary nodes cannot represent these guards. In particular, + // an overflowing sum/count rewrite must retry the original average. + SummaryExpr::BinaryOp { operator, .. } + if operator.checked_relative_division || operator.checked_finite_division => + { + QueryPlanNode::ExactFallback { + reason: "guarded summary division requires exact execution".into(), + } + } SummaryExpr::RelationalJoin { left, right, @@ -1054,6 +1070,84 @@ fn physical_grouping( mod tests { use super::*; + #[test] + fn guarded_division_falls_back_in_both_query_compilers() { + // Neither installed binary representation can retain Planner's division guards. + let query = "avg_over_time(m[5m])"; + let roots = crate::asap_tier_implement::implement_promql_for_asap_tier(query).unwrap(); + let root = &roots[0]; + let SummaryExpr::BinaryOp { operator, .. } = &root.expr else { + panic!("expected the Planner's average rewrite"); + }; + assert!(operator.checked_finite_division); + for relative in [false, true] { + let mut guarded = root.as_ref().clone(); + let SummaryExpr::BinaryOp { operator, .. } = &mut guarded.expr else { + unreachable!(); + }; + operator.checked_finite_division = !relative; + operator.checked_relative_division = relative; + let guarded = Rc::new(guarded); + for composable in [false, true] { + let instant = InstantExecution { + lookback_ms: 300_000, + full_history: false, + cumulative_readout: false, + }; + let bind = |_: &Rc, _: &SummaryFamilyType| { + Ok(MaterializationBinding { + materialization: PolicyFingerprint(7).into(), + output_grouping: PhysicalGrouping::PerEntity, + window_ms: 300_000, + pane_origin_ms: Some(0), + readout_lookback_ms: Some(300_000), + item_labels: Vec::new(), + }) + }; + let entry = if composable { + compile_bound_composable( + "guarded".into(), + query.into(), + &guarded, + instant, + FallbackPolicy::ExactBackend, + bind, + ) + } else { + compile_bound( + "guarded".into(), + query.into(), + &guarded, + instant, + FallbackPolicy::ExactBackend, + bind, + ) + } + .unwrap(); + if composable && !relative { + let QueryPlanNode::Logical { + operator: logical::LogicalOperator::ExactSubquery { query: exact_query }, + .. + } = &entry.nodes[&entry.root] + else { + panic!("expected the original exact average: {:?}", entry.nodes); + }; + assert_eq!(exact_query, query); + } else { + assert!( + matches!( + entry.nodes[&entry.root], + QueryPlanNode::ExactFallback { .. } + ), + "guard discarded (relative={relative}, composable={composable}): {:?}", + entry.nodes + ); + } + assert!(entry.materialization_bindings().is_empty()); + } + } + } + #[test] fn canonical_identity_ignores_formatting() { assert_eq!(