diff --git a/Cargo.lock b/Cargo.lock index 7be2448a..e173aa68 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -364,7 +364,7 @@ dependencies = [ [[package]] name = "asap-aware-mapping" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=b8b5d705215200361b6a2c3b389d513d554077ca#b8b5d705215200361b6a2c3b389d513d554077ca" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=029ff2fe041172c94c2d32c90b185bc83c5e8a57#029ff2fe041172c94c2d32c90b185bc83c5e8a57" dependencies = [ "asap-types", "serde", @@ -375,7 +375,7 @@ dependencies = [ [[package]] name = "asap-frontend-promql" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=b8b5d705215200361b6a2c3b389d513d554077ca#b8b5d705215200361b6a2c3b389d513d554077ca" +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)", @@ -384,7 +384,7 @@ dependencies = [ [[package]] name = "asap-frontend-sql" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=b8b5d705215200361b6a2c3b389d513d554077ca#b8b5d705215200361b6a2c3b389d513d554077ca" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=029ff2fe041172c94c2d32c90b185bc83c5e8a57#029ff2fe041172c94c2d32c90b185bc83c5e8a57" dependencies = [ "asap-sql-function-catalog", "asap-types", @@ -407,12 +407,12 @@ dependencies = [ [[package]] name = "asap-sql-function-catalog" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=b8b5d705215200361b6a2c3b389d513d554077ca#b8b5d705215200361b6a2c3b389d513d554077ca" +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=b8b5d705215200361b6a2c3b389d513d554077ca#b8b5d705215200361b6a2c3b389d513d554077ca" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=029ff2fe041172c94c2d32c90b185bc83c5e8a57#029ff2fe041172c94c2d32c90b185bc83c5e8a57" dependencies = [ "serde", "serde_json", diff --git a/Cargo.toml b/Cargo.toml index 268aedba..ae7e90f8 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 (current-series Planner PR). # Alias upstream asap-types because this workspace also defines asap_types. -planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "b8b5d705215200361b6a2c3b389d513d554077ca" } -asap-aware-mapping = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "b8b5d705215200361b6a2c3b389d513d554077ca" } -asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "b8b5d705215200361b6a2c3b389d513d554077ca" } -asap-frontend-sql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "b8b5d705215200361b6a2c3b389d513d554077ca" } +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"] } @@ -46,3 +46,4 @@ reqwest = { version = "0.12", default-features = false, features = ["json", "rus 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 dc34920b..ceaa441f 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/clickhouse.rs b/control_plane/src/clickhouse.rs index 96c3b22f..f1e1b1df 100644 --- a/control_plane/src/clickhouse.rs +++ b/control_plane/src/clickhouse.rs @@ -1224,7 +1224,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/backend_wire.rs b/control_plane/src/emit/backend_wire.rs index 89d23d44..6eaad495 100644 --- a/control_plane/src/emit/backend_wire.rs +++ b/control_plane/src/emit/backend_wire.rs @@ -230,9 +230,8 @@ pub(crate) fn build_backend_aggregation_json(agg: &BackendAggregation) -> JsonVa match kind { ExactKind::Sum => "Sum", ExactKind::Count => "Count", - // Typed `Min` shares the MinMax accumulator on the wire; the - // readout direction is carried by the aggregation sub-type. - ExactKind::MinMax | ExactKind::Min => "MinMax", + ExactKind::Min => "Min", + ExactKind::Max => "Max", ExactKind::Increase => "Increase", ExactKind::Rate => "Rate", ExactKind::IRate => "IRate", @@ -286,20 +285,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, - // `MinMax` and the typed `Min` share one accumulator on the wire, so the - // readout direction has to travel in the sub-type or the backend cannot - // tell which end of the window the query wants. - "aggregationSubType": match &agg.family { - planner_types::post_asap::SummaryFamilyType::ExactAggregate( - planner_types::post_asap::ExactKind::MinMax, - _, - ) => "max", - planner_types::post_asap::SummaryFamilyType::ExactAggregate( - planner_types::post_asap::ExactKind::Min, - _, - ) => "min", - _ => "", - }, + "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 353302ca..bc6968d5 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -727,7 +727,7 @@ fn has_unsafe_raw_entity_leaf( let preserves_series_state = matches!( family, SummaryFamilyType::ExactAggregate( - ExactKind::Increase | ExactKind::Rate | ExactKind::MinMax | ExactKind::Min, + ExactKind::Increase | ExactKind::Rate | ExactKind::Min | ExactKind::Max, _ ) ); @@ -1063,7 +1063,7 @@ impl PhysicalPlanCompiler { && (!matches!( state.family, SummaryFamilyType::ExactAggregate( - planner_types::post_asap::ExactKind::MinMax, + planner_types::post_asap::ExactKind::Max, _ ) ) || crate::query_plan::residual::selected_range_max_materialization( @@ -2694,10 +2694,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, } @@ -2715,7 +2717,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() { @@ -3874,8 +3876,8 @@ pub(crate) mod tests { let plan = PhysicalPlanCompiler.compile_promql(input, env).unwrap(); assert_eq!(plan.precompute_plan.materializations.len(), 1); assert_eq!( - plan.precompute_plan.materializations[0].aggregation_sub_type, - "min" + plan.precompute_plan.materializations[0].aggregation_type, + asap_types::AggregationType::Min ); assert!(plan .query_plan diff --git a/control_plane/src/physical/post_asap/matcher.rs b/control_plane/src/physical/post_asap/matcher.rs index 3cb1a80d..43fdf6bb 100644 --- a/control_plane/src/physical/post_asap/matcher.rs +++ b/control_plane/src/physical/post_asap/matcher.rs @@ -189,8 +189,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, @@ -285,10 +285,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 ecbb1153..0e513c60 100644 --- a/control_plane/src/physical/runtime_capability.rs +++ b/control_plane/src/physical/runtime_capability.rs @@ -63,7 +63,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 @@ -351,8 +351,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 { @@ -360,7 +360,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) ) } @@ -394,9 +395,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)) } @@ -537,24 +541,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 @@ -762,10 +778,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)); } @@ -805,11 +821,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 6f54c173..15ce225a 100644 --- a/control_plane/src/query_plan.rs +++ b/control_plane/src/query_plan.rs @@ -823,8 +823,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, } } @@ -913,7 +913,7 @@ pub(crate) fn exact_value_executable(node: &SummaryNode) -> bool { | ExactKind::Increase | ExactKind::Rate | ExactKind::Min - | ExactKind::MinMax + | ExactKind::Max ) } else { // Raw producer grouping may move through additive reductions, @@ -1169,6 +1169,90 @@ mod catalog_binding_tests { mod tests { use super::*; + #[test] + fn guarded_division_retains_checks_in_both_query_compilers() { + // Both compilers retain the finite/relative guard supplied by Planner. + let query = "avg_over_time(m[5m])"; + let canonical = crate::query_parser::parse_query_expr_canonical( + query, + planner_types::types::AccuracyTarget::Exact, + ) + .unwrap(); + let root = crate::planner_selection::select_summary_default(&canonical).unwrap(); + 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 { + full_window_slide_ms: None, + 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_mapped( + "guarded".into(), + query.into(), + &guarded, + instant, + FallbackPolicy::ExactBackend, + bind, + |_, _| {}, + ) + } else { + compile_bound_mapped( + "guarded".into(), + query.into(), + &guarded, + instant, + FallbackPolicy::ExactBackend, + bind, + |_, _| {}, + ) + } + .unwrap(); + let QueryPlanNode::Logical { + operator: residual::ResidualQueryOperator::Binary { operation, .. }, + .. + } = &entry.nodes[&entry.root] + else { + panic!( + "expected guarded division (composable={composable}): {:?}", + entry.nodes + ); + }; + assert_eq!( + *operation, + if relative { + residual::BinaryOperation::CheckedDiv + } else { + residual::BinaryOperation::FiniteDiv + } + ); + assert!(!entry.materialization_bindings().is_empty()); + } + } + } + #[test] fn canonical_identity_ignores_formatting() { assert_eq!( diff --git a/control_plane/src/query_plan/residual.rs b/control_plane/src/query_plan/residual.rs index bc21e6a1..05845755 100644 --- a/control_plane/src/query_plan/residual.rs +++ b/control_plane/src/query_plan/residual.rs @@ -766,7 +766,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/crates/asap_types/src/accumulator_spec.rs b/crates/asap_types/src/accumulator_spec.rs index ab3fc669..482ef6d6 100644 --- a/crates/asap_types/src/accumulator_spec.rs +++ b/crates/asap_types/src/accumulator_spec.rs @@ -41,9 +41,13 @@ //! //! Backend-specific execution details remain deliberately separate: //! -//! - **Min/max direction.** Planner's `ExactKind::Min` and legacy maximum -//! `ExactKind::MinMax` map to the shared wire accumulator with an explicit -//! aggregation subtype. The physical compiler preserves this typed direction. +//! - **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 @@ -231,12 +235,12 @@ impl AggregationConfig { SummaryFamilyType::ExactAggregate(ExactKind::Increase, ExactParams::Increase), false, ), - MinMax => ( - if sub_type.eq_ignore_ascii_case("min") { - SummaryFamilyType::ExactAggregate(ExactKind::Min, ExactParams::Min) - } else { - SummaryFamilyType::ExactAggregate(ExactKind::MinMax, ExactParams::MinMax) - }, + Min => ( + SummaryFamilyType::ExactAggregate(ExactKind::Min, ExactParams::Min), + false, + ), + Max => ( + SummaryFamilyType::ExactAggregate(ExactKind::Max, ExactParams::Max), false, ), DatasketchesKLL => ( @@ -256,12 +260,12 @@ impl AggregationConfig { SummaryFamilyType::ExactAggregate(ExactKind::Increase, ExactParams::Increase), true, ), - MultipleMinMax => ( - if sub_type.eq_ignore_ascii_case("min") { - SummaryFamilyType::ExactAggregate(ExactKind::Min, ExactParams::Min) - } else { - SummaryFamilyType::ExactAggregate(ExactKind::MinMax, ExactParams::MinMax) - }, + MultipleMin => ( + SummaryFamilyType::ExactAggregate(ExactKind::Min, ExactParams::Min), + true, + ), + MultipleMax => ( + SummaryFamilyType::ExactAggregate(ExactKind::Max, ExactParams::Max), true, ), HydraKLL => { @@ -396,12 +400,12 @@ impl AggregationConfig { SummaryFamilyType::ExactAggregate(ExactKind::Sum, ExactParams::Sum), false, ), - "Min" | "min" | "Max" | "max" => ( - if sub_type.eq_ignore_ascii_case("min") { - SummaryFamilyType::ExactAggregate(ExactKind::Min, ExactParams::Min) - } else { - 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" => ( @@ -428,12 +432,12 @@ impl AggregationConfig { SummaryFamilyType::ExactAggregate(ExactKind::Sum, ExactParams::Sum), true, ), - "Min" | "min" | "Max" | "max" => ( - if sub_type.eq_ignore_ascii_case("min") { - SummaryFamilyType::ExactAggregate(ExactKind::Min, ExactParams::Min) - } else { - 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 a19734e7..ccdcbec0 100644 --- a/crates/asap_types/src/aggregation_type.rs +++ b/crates/asap_types/src/aggregation_type.rs @@ -15,12 +15,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, @@ -40,11 +42,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", @@ -65,7 +69,8 @@ impl AggregationType { AggregationType::MultipleSubpopulation | AggregationType::MultipleSum | AggregationType::MultipleIncrease - | AggregationType::MultipleMinMax + | AggregationType::MultipleMin + | AggregationType::MultipleMax | AggregationType::CountMinSketch | AggregationType::CountMinSketchWithHeap | AggregationType::CountSketch @@ -89,11 +94,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), @@ -109,7 +116,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) } @@ -117,7 +125,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) @@ -127,6 +136,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 554fe269..a219ff03 100644 --- a/crates/asap_types/src/precompute_plan.rs +++ b/crates/asap_types/src/precompute_plan.rs @@ -177,8 +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, @@ -194,7 +199,7 @@ 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, diff --git a/crates/asap_types/src/query_plan.rs b/crates/asap_types/src/query_plan.rs index 3b56492d..5bdd2605 100644 --- a/crates/asap_types/src/query_plan.rs +++ b/crates/asap_types/src/query_plan.rs @@ -633,8 +633,8 @@ pub enum ExactReadout { Count, Increase, Rate, - Max, Min, + Max, } #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] diff --git a/crates/asap_types/src/sds.rs b/crates/asap_types/src/sds.rs index f24f2779..c0621614 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 b22ed208..b6d3bfd7 100644 --- a/data_plane/src/drivers/ingest/prometheus_remote_write.rs +++ b/data_plane/src/drivers/ingest/prometheus_remote_write.rs @@ -718,8 +718,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 913e5c14..2721bc68 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 462c824c..43a779fc 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 // --------------------------------------------------------------------------- @@ -912,7 +905,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 @@ -1053,23 +1047,22 @@ pub fn create_accumulator_updater(config: &AggregationConfig) -> Box { - Box::new(MinMaxAccumulatorUpdater::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::MinMax | ExactKind::Min, _), true) => { - Box::new(MultipleMinMaxAccumulatorUpdater::new( - config.aggregation_sub_type.eq_ignore_ascii_case("max"), - )) + (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) => { @@ -1337,13 +1330,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] @@ -1479,7 +1472,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/operators/max_accumulator.rs b/data_plane/src/precompute_engine/operators/max_accumulator.rs new file mode 100644 index 00000000..733c2028 --- /dev/null +++ b/data_plane/src/precompute_engine/operators/max_accumulator.rs @@ -0,0 +1,248 @@ +use crate::storage_engines::types::{ + AggregateCore, AggregationType, AuxStats, MergeableAccumulator, SerializableToSink, + SingleSubpopulationAggregate, +}; +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()) + } +} + +#[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 00000000..e69cda83 --- /dev/null +++ b/data_plane/src/precompute_engine/operators/min_accumulator.rs @@ -0,0 +1,253 @@ +use crate::storage_engines::types::{ + AggregateCore, AggregationType, AuxStats, MergeableAccumulator, SerializableToSink, + SingleSubpopulationAggregate, +}; +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()) + } +} + +#[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 dbf7ed92..00000000 --- a/data_plane/src/precompute_engine/operators/min_max_accumulator.rs +++ /dev/null @@ -1,395 +0,0 @@ -use crate::storage_engines::types::{ - AggregateCore, AggregationType, AuxStats, MergeableAccumulator, SerializableToSink, - SingleSubpopulationAggregate, -}; -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()) - } -} - -#[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 7d7c8466..af284459 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 00000000..1865d268 --- /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 00000000..00c25040 --- /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 e65766c4..00000000 --- a/data_plane/src/precompute_engine/operators/multiple_min_max_accumulator.rs +++ /dev/null @@ -1,486 +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 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"), - } - } - - #[cfg(test)] - 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/query_engines/asap_query_engine/catalog_resolver.rs b/data_plane/src/query_engines/asap_query_engine/catalog_resolver.rs index bcd989c4..ba07c0b9 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,14 +82,11 @@ impl ResolvedMaterialization<'_> { ExactReadout::Increase | ExactReadout::Rate => { matches!(aggregation_type, Increase | MultipleIncrease) } - ExactReadout::Min => { - matches!(aggregation_type, MinMax | MultipleMinMax) - && aggregation_sub_type.eq_ignore_ascii_case("min") - } - 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 ee0417a3..9a40f98d 100644 --- a/data_plane/src/query_engines/asap_query_engine/engine.rs +++ b/data_plane/src/query_engines/asap_query_engine/engine.rs @@ -1390,7 +1390,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; @@ -1549,8 +1550,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()) @@ -1799,8 +1800,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. @@ -1819,7 +1820,7 @@ mod asap_tier_classify_tests { }, ); let engine = ASAPQueryEngine::new(15000).with_sketch_index(idx); - let query = "count(unique_users_per_min)"; + let query = "count(distinct_over_time(unique_users_per_min[1m]))"; let instant = engine.execute_at(query, 2_000).await; let range = engine .execute_range_promql_modern(query, 1_000, 2_000, 1_000) @@ -1849,7 +1850,7 @@ mod asap_tier_classify_tests { &[], 1000, ); - let query = "count(unique_users_per_min)"; + let query = "count(distinct_over_time(unique_users_per_min[1m]))"; let entry = test_plan::entry( query, &config, @@ -1896,7 +1897,7 @@ mod asap_tier_classify_tests { &[], 1000, ); - let query = "count(unique_users_per_min)"; + let query = "count(distinct_over_time(unique_users_per_min[1m]))"; let entry = test_plan::entry( query, &config, @@ -1915,7 +1916,7 @@ mod asap_tier_classify_tests { ); 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)" ); @@ -2012,7 +2013,7 @@ mod asap_tier_classify_tests { &[], 1000, ); - let query = "count(unique_users_global)"; + let query = "count(distinct_over_time(unique_users_global[1m]))"; let entry = test_plan::entry( query, &config, @@ -2027,7 +2028,7 @@ mod asap_tier_classify_tests { let result = engine .execute_at(query, w_end) .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 { @@ -2035,7 +2036,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/post_asap_readout.rs b/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs index 4962ec64..41b41ab7 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 @@ -931,7 +931,7 @@ mod tests { 1000, ); let entry = crate::query_engines::asap_query_engine::test_plan::entry( - "count(unique_users)", + "count(distinct_over_time(unique_users[1m]))", &config, asap_types::query_plan::PhysicalGrouping::PerEntity, 1000, @@ -995,7 +995,7 @@ mod tests { 1000, ); let entry = test_plan::entry( - "count(unique_users)", + "count(distinct_over_time(unique_users[1m]))", &config, PhysicalGrouping::Reduce(vec![]), 1000, @@ -1009,8 +1009,8 @@ mod tests { 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_executor.rs b/data_plane/src/query_engines/asap_query_engine/summary_executor.rs index 07dd86c3..d4bb3841 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,14 +253,14 @@ impl GroupState { asap_types::query_plan::ExactReadout::Rate, AggregationType::Increase | AggregationType::MultipleIncrease, ) => asap_types::Statistic::Rate, - ( - asap_types::query_plan::ExactReadout::Max, - AggregationType::MinMax | AggregationType::MultipleMinMax, - ) => asap_types::Statistic::Max, ( asap_types::query_plan::ExactReadout::Min, - AggregationType::MinMax | AggregationType::MultipleMinMax, + AggregationType::Min | AggregationType::MultipleMin, ) => asap_types::Statistic::Min, + ( + asap_types::query_plan::ExactReadout::Max, + AggregationType::Max | AggregationType::MultipleMax, + ) => asap_types::Statistic::Max, _ => return None, }; @@ -287,26 +288,37 @@ impl GroupState { } if matches!( agg_type, - AggregationType::MinMax | AggregationType::MultipleMinMax - ) && matches!( - readout, - asap_types::query_plan::ExactReadout::Max | asap_types::query_plan::ExactReadout::Min - ) { + 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::() + .downcast_ref::() .map(|a| a.value) }) .collect::>>()? .into_iter() - .reduce(if readout == asap_types::query_plan::ExactReadout::Min { - f64::min - } else { - f64::max - }); + .reduce(f64::min); + } + if matches!( + agg_type, + AggregationType::Max | AggregationType::MultipleMax + ) && readout == asap_types::query_plan::ExactReadout::Max + { + return entries + .iter() + .flat_map(|windows| windows.values()) + .map(|acc| { + acc.as_any() + .downcast_ref::() + .map(|a| a.value) + }) + .collect::>>()? + .into_iter() + .reduce(f64::max); } let mut merged: Option> = None; for windows in entries { @@ -596,18 +608,26 @@ impl QueryExecutionContext<'_> { )); } } - if binding.full_window_slide_ms.is_none() - && 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) = binding + .full_window_slide_ms + .is_none() + .then(|| { + self.index + .query_rollup_range(reduction, sid, self.t0_ms, self.t1_ms) + }) + .flatten() + { for (labels, value) in series { let key = match &binding.output_grouping { PhysicalGrouping::PerEntity => labels, @@ -615,12 +635,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, }); @@ -1236,10 +1259,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), @@ -3243,26 +3270,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); @@ -3270,8 +3297,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 5c0acd00..f3780b04 100644 --- a/data_plane/src/storage_engines/sketch_db/accuracy.rs +++ b/data_plane/src/storage_engines/sketch_db/accuracy.rs @@ -92,10 +92,12 @@ fn derive_sketch_only(config: &AggregationConfig) -> AccuracyProfile { // here too before its retirement.) AggregationType::Sum | AggregationType::Increase - | AggregationType::MinMax + | AggregationType::Min + | AggregationType::Max | AggregationType::MultipleSum | AggregationType::MultipleIncrease - | AggregationType::MultipleMinMax => AccuracyProfile::exact(), + | AggregationType::MultipleMin + | AggregationType::MultipleMax => AccuracyProfile::exact(), AggregationType::CountMinSketch => { let (rows, cols) = cms_params(config); @@ -532,7 +534,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 = 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 0266ddc1..ef6c54ce 100644 --- a/data_plane/src/storage_engines/sketch_db/index/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs @@ -92,7 +92,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; @@ -103,7 +103,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) @@ -112,9 +115,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. @@ -208,7 +210,7 @@ pub struct SummarySeriesMetadata { /// 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, @@ -432,12 +434,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), } } @@ -1450,7 +1454,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. /// @@ -1498,11 +1502,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) @@ -1517,9 +1531,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 0848da10..83778690 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);