diff --git a/crates/asap_types/src/aggregation_config.rs b/crates/asap_types/src/aggregation_config.rs index a8d4dfd3..d3d2da5f 100644 --- a/crates/asap_types/src/aggregation_config.rs +++ b/crates/asap_types/src/aggregation_config.rs @@ -124,11 +124,11 @@ impl AggregationConfig { } /// `PolicyFingerprint::as_u64()` — the u64-form handle used by the - /// legacy aggregation-id-keyed call sites. **Always** equal to - /// `self.policy_fingerprint().as_u64()`. Provided so callers that - /// previously read `config.aggregation_id` can keep the same - /// arithmetic without owning the typed handle. - pub fn aggregation_id(&self) -> u64 { + /// policy-fingerprint-keyed call sites (e.g. `StreamingConfig`'s + /// `HashMap` keys). **Always** equal to + /// `self.policy_fingerprint().as_u64()`. The value is content- + /// addressed identity, NOT a controller-allocated counter id. + pub fn policy_fp_u64(&self) -> u64 { self.policy_fingerprint().as_u64() } @@ -437,16 +437,16 @@ mod tests { ); } - /// The `aggregation_id()` accessor is exactly the fingerprint u64. + /// The `policy_fp_u64()` accessor is exactly the fingerprint u64. #[test] - fn aggregation_id_accessor_equals_fingerprint_u64() { + fn policy_fp_u64_accessor_equals_fingerprint_u64() { let cfg = AggregationConfig::from_yaml_data( &sample_yaml(false), None, QueryLanguage::promql, ) .expect("parse"); - assert_eq!(cfg.aggregation_id(), cfg.policy_fingerprint().as_u64()); + assert_eq!(cfg.policy_fp_u64(), cfg.policy_fingerprint().as_u64()); } /// PR 5: `serialize_to_json` no longer emits `aggregationId`. diff --git a/crates/asap_types/src/capability_matching.rs b/crates/asap_types/src/capability_matching.rs index 952fdc0b..ce853a6f 100644 --- a/crates/asap_types/src/capability_matching.rs +++ b/crates/asap_types/src/capability_matching.rs @@ -397,7 +397,7 @@ pub fn spatial_filter_compatible(config_filter: &str, req_filter: &str) -> bool /// avoids the key-aggregation hunt when both shapes serve the /// statistic — which is the common case for `Statistic::Sum` /// matching both `Sum` and `CountMinSketch`. -/// 3. **Tie-break on `aggregation_id()` (the policy fingerprint).** +/// 3. **Tie-break on `policy_fp_u64()` (the policy fingerprint).** /// Deterministic across runs and hosts; fixes the /// HashMap-iteration-order flake on `avg_finds_sum_and_count`. pub fn aggregation_priority(a: &AggregationConfig, b: &AggregationConfig) -> Ordering { @@ -407,7 +407,7 @@ pub fn aggregation_priority(a: &AggregationConfig, b: &AggregationConfig) -> Ord .cmp(&a.window_size) // `false < true` in Rust's bool Ord → single-pop sorts FIRST. .then_with(|| a_multi.cmp(&b_multi)) - .then_with(|| a.aggregation_id().cmp(&b.aggregation_id())) + .then_with(|| a.policy_fp_u64().cmp(&b.policy_fp_u64())) } // --------------------------------------------------------------------------- @@ -463,7 +463,7 @@ pub fn find_compatible_aggregation( ); if !ok { debug!( - agg_id = c.aggregation_id(), + policy_fp = c.policy_fp_u64(), agg_type = %c.aggregation_type, metric = %c.metric, window_size_s = c.window_size, @@ -489,7 +489,7 @@ pub fn find_compatible_aggregation( debug!( statistic = ?stat, num_candidates = candidates.len(), - chosen_agg_id = candidates[0].aggregation_id(), + chosen_policy_fp = candidates[0].policy_fp_u64(), chosen_agg_type = %candidates[0].aggregation_type, chosen_window_size_s = candidates[0].window_size, "capability matching: found candidates, chose best", @@ -538,17 +538,17 @@ pub fn find_compatible_aggregation( debug!( metric = %requirements.metric, - value_agg_id = value_agg.aggregation_id(), + value_policy_fp = value_agg.policy_fp_u64(), value_agg_type = %value_agg.aggregation_type, - key_agg_id = key_agg.aggregation_id(), + key_policy_fp = key_agg.policy_fp_u64(), key_agg_type = %key_agg.aggregation_type, "capability matching: resolved", ); Some(AggregationIdInfo { - aggregation_id_for_value: value_agg.aggregation_id(), + aggregation_id_for_value: value_agg.policy_fp_u64(), aggregation_type_for_value: value_agg.aggregation_type, - aggregation_id_for_key: key_agg.aggregation_id(), + aggregation_id_for_key: key_agg.policy_fp_u64(), aggregation_type_for_key: key_agg.aggregation_type, }) } @@ -620,7 +620,7 @@ mod tests { fn single_config(config: AggregationConfig) -> HashMap { let mut m = HashMap::new(); - m.insert(config.aggregation_id(), config); + m.insert(config.policy_fp_u64(), config); m } @@ -629,7 +629,7 @@ mod tests { #[test] fn basic_sum_match() { let cfg = make_config(1, "cpu", "Sum", "", 300, "tumbling", &[], ""); - let expected = cfg.aggregation_id(); + let expected = cfg.policy_fp_u64(); let configs = single_config(cfg); let result = find_compatible_aggregation( &configs, @@ -642,7 +642,7 @@ mod tests { #[test] fn quantile_any_value_finds_kll() { let cfg = make_config(2, "lat", "DatasketchesKLL", "", 300, "tumbling", &[], ""); - let expected = cfg.aggregation_id(); + let expected = cfg.policy_fp_u64(); let configs = single_config(cfg); // quantile value (0.5 or 0.9) is NOT part of QueryRequirements — both should find the same config let r1 = find_compatible_aggregation( @@ -660,7 +660,7 @@ mod tests { #[test] fn quantile_matches_hydrarkll() { let cfg = make_config(3, "lat", "HydraKLL", "", 300, "tumbling", &[], ""); - let expected = cfg.aggregation_id(); + let expected = cfg.policy_fp_u64(); let configs = single_config(cfg); let result = find_compatible_aggregation( &configs, @@ -757,10 +757,10 @@ mod tests { fn window_priority_largest_wins() { let small = make_config(1, "cpu", "Sum", "", 300, "tumbling", &[], ""); let large = make_config(2, "cpu", "Sum", "", 900, "tumbling", &[], ""); - let expected = large.aggregation_id(); + let expected = large.policy_fp_u64(); let mut configs = HashMap::new(); - configs.insert(small.aggregation_id(), small); - configs.insert(large.aggregation_id(), large); + configs.insert(small.policy_fp_u64(), small); + configs.insert(large.policy_fp_u64(), large); // 900_000 ms is divisible by both 300 s and 900 s — prefer 900 s let result = find_compatible_aggregation( &configs, @@ -1025,8 +1025,8 @@ mod tests { let sum = make_config(1, "cpu", "Sum", "", 300, "tumbling", &["job"], ""); let cnt = make_config(2, "cpu", "CountMinSketch", "", 300, "tumbling", &["job"], ""); let mut configs = HashMap::new(); - configs.insert(sum.aggregation_id(), sum); - configs.insert(cnt.aggregation_id(), cnt); + configs.insert(sum.policy_fp_u64(), sum); + configs.insert(cnt.policy_fp_u64(), cnt); let result = find_compatible_aggregation( &configs, &req( @@ -1046,8 +1046,8 @@ mod tests { // Count config has different window_size — must be rejected let cnt = make_config(2, "cpu", "CountMinSketch", "", 900, "tumbling", &["job"], ""); let mut configs = HashMap::new(); - configs.insert(sum.aggregation_id(), sum); - configs.insert(cnt.aggregation_id(), cnt); + configs.insert(sum.policy_fp_u64(), sum); + configs.insert(cnt.policy_fp_u64(), cnt); let result = find_compatible_aggregation( &configs, &req( @@ -1220,9 +1220,9 @@ mod tests { &[], "", ); - let expected = cfg.aggregation_id(); + let expected = cfg.policy_fp_u64(); let mut configs = HashMap::new(); - configs.insert(cfg.aggregation_id(), cfg); + configs.insert(cfg.policy_fp_u64(), cfg); let result = find_compatible_aggregation( &configs, &req( diff --git a/crates/asap_types/src/policy_fingerprint.rs b/crates/asap_types/src/policy_fingerprint.rs index 7ca86924..a330c69f 100644 --- a/crates/asap_types/src/policy_fingerprint.rs +++ b/crates/asap_types/src/policy_fingerprint.rs @@ -261,10 +261,10 @@ mod tests { /// vacuously true. Kept as a doc-comment anchor; no runtime test /// is needed. #[test] - fn aggregation_id_accessor_equals_fingerprint_u64() { + fn policy_fp_u64_accessor_equals_fingerprint_u64() { let a = cfg("http_lat", AggregationType::Sum, HashMap::new(), vec![], 60, ""); assert_eq!( - a.aggregation_id(), + a.policy_fp_u64(), PolicyFingerprint::from_config(&a).as_u64(), ); } diff --git a/crates/asap_types/src/streaming_config.rs b/crates/asap_types/src/streaming_config.rs index 0ae983ba..35b5c12a 100644 --- a/crates/asap_types/src/streaming_config.rs +++ b/crates/asap_types/src/streaming_config.rs @@ -111,9 +111,9 @@ impl StreamingConfig { QueryLanguage::promql, )?; // PR 5: the map key IS the policy-fingerprint u64. - // `AggregationConfig::aggregation_id()` is the canonical + // `AggregationConfig::policy_fp_u64()` is the canonical // accessor for this value. - aggregation_configs.insert(config.aggregation_id(), config); + aggregation_configs.insert(config.policy_fp_u64(), config); } } @@ -181,7 +181,7 @@ aggregations:\n\ assert_eq!(cfg.aggregation_configs.len(), 1); let (k, v) = cfg.aggregation_configs.iter().next().unwrap(); assert_ne!(*k, 0, "derived id is not the 0 sentinel"); - assert_eq!(*k, v.aggregation_id(), "map key equals fingerprint u64"); + assert_eq!(*k, v.policy_fp_u64(), "map key equals fingerprint u64"); assert_eq!(v.metric, "cpu_seconds"); } diff --git a/data_plane/src/drivers/ingest/otel.rs b/data_plane/src/drivers/ingest/otel.rs index 015418f7..41c6adeb 100644 --- a/data_plane/src/drivers/ingest/otel.rs +++ b/data_plane/src/drivers/ingest/otel.rs @@ -598,7 +598,7 @@ fn resolve_bucket_sid_for_agg_config( let sid = ingest_state .series_resolver .resolve(&config.metric, &fp, &agg_kind_canonical); - let policy_fp = asap_types::PolicyFingerprint(config.aggregation_id()); + let policy_fp = asap_types::PolicyFingerprint(config.policy_fp_u64()); (sid, policy_fp) } @@ -2874,7 +2874,7 @@ mod sid_bucketing_tests { /// - each sid equals what `SeriesIdResolver::lookup` records for /// `(metric, "zone=;", ExactAgg-canonical)` — i.e. the /// bucket identity is folded into sid via the resolver - /// - policy_fp = config.aggregation_id() on every message + /// - policy_fp = config.policy_fp_u64() on every message /// - samples in each bucket are exactly the DPs whose `zone` /// attribute matches that bucket (the GROUP-BY semantic) /// @@ -2898,9 +2898,9 @@ mod sid_bucketing_tests { let metric = "cpu_seconds"; let cfg = sum_agg_config(metric, &["zone"]); - let policy_fp = asap_types::PolicyFingerprint(cfg.aggregation_id()); + let policy_fp = asap_types::PolicyFingerprint(cfg.policy_fp_u64()); let mut configs = HashMap::new(); - configs.insert(cfg.aggregation_id(), cfg.clone()); + configs.insert(cfg.policy_fp_u64(), cfg.clone()); let streaming = StreamingConfig::new(configs); let hot_reload = HotReloadStreamingConfig::new(streaming); @@ -2957,7 +2957,7 @@ mod sid_bucketing_tests { // Both buckets carry the same policy_fp (one source config). for (_, pf, _, _) in &groups { - assert_eq!(*pf, policy_fp, "policy_fp must equal config.aggregation_id()"); + assert_eq!(*pf, policy_fp, "policy_fp must equal config.policy_fp_u64()"); } // sids must be non-zero (zero is reserved on the wire) and distinct. diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs index 233a6229..b61368fe 100644 --- a/data_plane/src/drivers/query/servers/http.rs +++ b/data_plane/src/drivers/query/servers/http.rs @@ -2725,7 +2725,7 @@ aggregations: // PR 5: streaming-config is keyed on the policy // fingerprint. Build a marker→fingerprint map so the test // POSTs the right id on the wire. - let fp = cfg.aggregation_id(); + let fp = cfg.policy_fp_u64(); marker_to_fp.insert(*marker, fp); agg_map.insert(fp, cfg); } diff --git a/data_plane/src/precompute_engine/output_sink.rs b/data_plane/src/precompute_engine/output_sink.rs index 576ee8ef..189b132d 100644 --- a/data_plane/src/precompute_engine/output_sink.rs +++ b/data_plane/src/precompute_engine/output_sink.rs @@ -203,7 +203,7 @@ mod tests { fn sum_agg_config(_id: u64, metric: &str, grouping_keys: &[&str]) -> AggregationConfig { // `_id` is unused after PR 5 — identity is content-addressed // via `PolicyFingerprint::from_config`. Callers obtain the id - // via `config.aggregation_id()`. + // via `config.policy_fp_u64()`. AggregationConfig { aggregation_type: AggregationType::Sum, aggregation_sub_type: String::new(), @@ -229,7 +229,7 @@ mod tests { #[test] fn sketch_index_sink_writes_to_index() { let cfg = sum_agg_config(7, "cpu_seconds", &["zone"]); - let agg_id = cfg.aggregation_id(); + let agg_id = cfg.policy_fp_u64(); let mut configs = HashMap::new(); configs.insert(agg_id, cfg); let streaming = StreamingConfig::new(configs); diff --git a/data_plane/src/precompute_engine/worker.rs b/data_plane/src/precompute_engine/worker.rs index d4680ffb..9be15769 100644 --- a/data_plane/src/precompute_engine/worker.rs +++ b/data_plane/src/precompute_engine/worker.rs @@ -1285,7 +1285,7 @@ mod tests { ) -> AggregationConfig { // `_id` is unused after PR 5 — identity is content-addressed // via `PolicyFingerprint::from_config`. Callers below build the - // streaming-config map by reading `config.aggregation_id()` + // streaming-config map by reading `config.policy_fp_u64()` // from the returned value. let window_type = if slide_secs == 0 || slide_secs == window_secs { WindowType::Tumbling 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 1d3177f9..43a7e2a4 100644 --- a/data_plane/src/query_engines/asap_query_engine/engine.rs +++ b/data_plane/src/query_engines/asap_query_engine/engine.rs @@ -1722,7 +1722,7 @@ mod hot_reload_phase2_tests { fn cfg_with_agg(id: u64, metric: &str) -> (StreamingConfig, u64) { let mut map = std::collections::HashMap::new(); let cfg = dummy_agg(id, metric); - let fp = cfg.aggregation_id(); + let fp = cfg.policy_fp_u64(); map.insert(fp, cfg); (StreamingConfig::new(map), fp) } diff --git a/data_plane/src/storage_engines/sketch_db/backfill/mod.rs b/data_plane/src/storage_engines/sketch_db/backfill/mod.rs index 66b2fa6d..4ab3db95 100644 --- a/data_plane/src/storage_engines/sketch_db/backfill/mod.rs +++ b/data_plane/src/storage_engines/sketch_db/backfill/mod.rs @@ -559,7 +559,7 @@ impl BackfillRegistry { windows_total: u64, data_retention_ms: Option, ) -> Result { - let agg_id = config.aggregation_id(); + let agg_id = config.policy_fp_u64(); // Time-disjoint invariant: live ingest writes `[created_at, ∞)` // so backfill must stay strictly inside `[0, created_at)` or // touch the boundary exactly. diff --git a/data_plane/src/storage_engines/sketch_db/backfill/processor.rs b/data_plane/src/storage_engines/sketch_db/backfill/processor.rs index 55bd7bbb..9484d91c 100644 --- a/data_plane/src/storage_engines/sketch_db/backfill/processor.rs +++ b/data_plane/src/storage_engines/sketch_db/backfill/processor.rs @@ -439,14 +439,14 @@ mod tests { fn streaming_config_with(config: AggregationConfig) -> Arc { let mut map = std::collections::HashMap::new(); - map.insert(config.aggregation_id(), config); + map.insert(config.policy_fp_u64(), config); Arc::new(StreamingConfig::new(map)) } #[tokio::test] async fn happy_path_writes_one_output_per_group() { let cfg = sum_config(1, "latency", vec!["svc"]); - let fp = cfg.aggregation_id(); + let fp = cfg.policy_fp_u64(); let streaming = streaming_config_with(cfg.clone()); let hot = HotReloadStreamingConfig::from_arc(streaming.clone()); let registry = Arc::new(BackfillRegistry::new()); @@ -514,7 +514,7 @@ mod tests { #[tokio::test] async fn empty_samples_complete_without_write() { let cfg = sum_config(1, "m", vec![]); - let fp = cfg.aggregation_id(); + let fp = cfg.policy_fp_u64(); let streaming = streaming_config_with(cfg); let hot = HotReloadStreamingConfig::from_arc(streaming.clone()); let registry = Arc::new(BackfillRegistry::new()); @@ -536,7 +536,7 @@ mod tests { // Exercise the full chain: BackfillWorker drives the // processor over a job that covers 4 windows. let cfg = sum_config(1, "latency", vec!["svc"]); - let fp = cfg.aggregation_id(); + let fp = cfg.policy_fp_u64(); let streaming = streaming_config_with(cfg); let hot = HotReloadStreamingConfig::from_arc(streaming.clone()); let registry = Arc::new(BackfillRegistry::new()); @@ -692,7 +692,7 @@ mod tests { fn create_checked_rejects_end_past_created_at() { use super::super::CreateError; let cfg = sum_config(1, "m", vec![]); - let expected_fp = cfg.aggregation_id(); + let expected_fp = cfg.policy_fp_u64(); let created = now_ms(); let registry = BackfillRegistry::new(); @@ -742,7 +742,7 @@ mod tests { fn create_checked_rejects_start_older_than_data_retention() { use super::super::CreateError; let cfg = sum_config(1, "m", vec![]); - let expected_fp = cfg.aggregation_id(); + let expected_fp = cfg.policy_fp_u64(); let created = now_ms(); let registry = BackfillRegistry::new(); // Created-at is now_ms(), so data retention of 1 hour with @@ -846,7 +846,7 @@ mod tests { use crate::storage_engines::sketch_db::index::{SidLookup, SketchStore}; let cfg = sum_config(1, "latency", vec!["svc"]); - let fp = cfg.aggregation_id(); + let fp = cfg.policy_fp_u64(); let streaming = streaming_config_with(cfg.clone()); let hot = HotReloadStreamingConfig::from_arc(streaming.clone()); let registry = Arc::new(BackfillRegistry::new()); diff --git a/data_plane/src/storage_engines/sketch_db/backfill/service.rs b/data_plane/src/storage_engines/sketch_db/backfill/service.rs index 1e5f164d..5141ba8d 100644 --- a/data_plane/src/storage_engines/sketch_db/backfill/service.rs +++ b/data_plane/src/storage_engines/sketch_db/backfill/service.rs @@ -350,7 +350,7 @@ mod tests { fn streaming_with(cfg: AggregationConfig) -> Arc { let mut m = std::collections::HashMap::new(); - m.insert(cfg.aggregation_id(), cfg); + m.insert(cfg.policy_fp_u64(), cfg); Arc::new(StreamingConfig::new(m)) } @@ -380,7 +380,7 @@ mod tests { #[tokio::test(flavor = "current_thread")] async fn service_drains_queued_job_to_complete() { let cfg = sum_config(1, "latency"); - let agg_fp = cfg.aggregation_id(); + let agg_fp = cfg.policy_fp_u64(); let streaming = streaming_with(cfg); let hot = HotReloadStreamingConfig::from_arc(streaming.clone()); let registry = Arc::new(BackfillRegistry::new()); @@ -427,7 +427,7 @@ mod tests { #[tokio::test(flavor = "current_thread")] async fn service_marks_job_failed_when_reader_factory_fails() { let cfg = sum_config(1, "latency"); - let agg_fp = cfg.aggregation_id(); + let agg_fp = cfg.policy_fp_u64(); let streaming = streaming_with(cfg); let hot = HotReloadStreamingConfig::from_arc(streaming.clone()); let registry = Arc::new(BackfillRegistry::new()); @@ -461,7 +461,7 @@ mod tests { #[tokio::test(flavor = "current_thread")] async fn service_processes_multiple_jobs_in_id_order() { let cfg = sum_config(1, "latency"); - let agg_fp = cfg.aggregation_id(); + let agg_fp = cfg.policy_fp_u64(); let streaming = streaming_with(cfg); let hot = HotReloadStreamingConfig::from_arc(streaming.clone()); let registry = Arc::new(BackfillRegistry::new()); diff --git a/data_plane/src/storage_engines/sketch_db/lifecycle/eviction.rs b/data_plane/src/storage_engines/sketch_db/lifecycle/eviction.rs index 563c8009..4ae5ef26 100644 --- a/data_plane/src/storage_engines/sketch_db/lifecycle/eviction.rs +++ b/data_plane/src/storage_engines/sketch_db/lifecycle/eviction.rs @@ -285,7 +285,7 @@ mod tests { let mut id_to_fp = HashMap::new(); for &id in ids { let cfg = sum_agg_config(id); - let fp = cfg.aggregation_id(); + let fp = cfg.policy_fp_u64(); id_to_fp.insert(id, fp); map.insert(fp, cfg); } diff --git a/data_plane/src/storage_engines/types/hot_reload_config.rs b/data_plane/src/storage_engines/types/hot_reload_config.rs index 6667440f..e7234923 100644 --- a/data_plane/src/storage_engines/types/hot_reload_config.rs +++ b/data_plane/src/storage_engines/types/hot_reload_config.rs @@ -173,7 +173,7 @@ mod tests { let mut id_to_fp = std::collections::HashMap::new(); for &id in ids { let cfg = dummy_agg(id); - let fp = cfg.aggregation_id(); + let fp = cfg.policy_fp_u64(); id_to_fp.insert(id, fp); map.insert(fp, cfg); } diff --git a/data_plane/src/tests/test_utilities/engine_factories.rs b/data_plane/src/tests/test_utilities/engine_factories.rs index d3204958..c8d23760 100644 --- a/data_plane/src/tests/test_utilities/engine_factories.rs +++ b/data_plane/src/tests/test_utilities/engine_factories.rs @@ -105,7 +105,7 @@ pub fn create_engine_single_pop_with_aggregated( num_aggregates_to_retain: None, table_name: None, value_column: None}; - let agg_id = agg_config.aggregation_id(); + let agg_id = agg_config.policy_fp_u64(); aggregation_configs.insert(agg_id, agg_config); let streaming_config = Arc::new(StreamingConfig { @@ -188,7 +188,7 @@ pub fn create_engine_dual_input( num_aggregates_to_retain: None, table_name: None, value_column: None}; - let value_id = value_agg_config.aggregation_id(); + let value_id = value_agg_config.policy_fp_u64(); aggregation_configs.insert(value_id, value_agg_config); // Keys aggregation @@ -209,7 +209,7 @@ pub fn create_engine_dual_input( num_aggregates_to_retain: None, table_name: None, value_column: None}; - let keys_id = keys_agg_config.aggregation_id(); + let keys_id = keys_agg_config.policy_fp_u64(); aggregation_configs.insert(keys_id, keys_agg_config); let streaming_config = Arc::new(StreamingConfig { @@ -282,7 +282,7 @@ pub fn create_engine_two_metrics( num_aggregates_to_retain: None, table_name: None, value_column: None}; - let id_a = agg_config_a.aggregation_id(); + let id_a = agg_config_a.policy_fp_u64(); aggregation_configs.insert(id_a, agg_config_a); let agg_config_b = AggregationConfig { @@ -302,7 +302,7 @@ pub fn create_engine_two_metrics( num_aggregates_to_retain: None, table_name: None, value_column: None}; - let id_b = agg_config_b.aggregation_id(); + let id_b = agg_config_b.policy_fp_u64(); aggregation_configs.insert(id_b, agg_config_b); let streaming_config = Arc::new(StreamingConfig { @@ -379,7 +379,7 @@ pub fn create_engine_three_metrics( num_aggregates_to_retain: None, table_name: None, value_column: None}; - let id = cfg.aggregation_id(); + let id = cfg.policy_fp_u64(); ids.push(id); aggregation_configs.insert(id, cfg); } @@ -439,7 +439,7 @@ pub fn create_engine_multi_timestamp( num_aggregates_to_retain: None, table_name: None, value_column: None}; - let agg_id = agg_config.aggregation_id(); + let agg_id = agg_config.policy_fp_u64(); aggregation_configs.insert(agg_id, agg_config); let streaming_config = Arc::new(StreamingConfig { @@ -493,7 +493,7 @@ pub fn create_engine_multi_timestamp_with_window( num_aggregates_to_retain: None, table_name: None, value_column: None}; - let agg_id = agg_config.aggregation_id(); + let agg_id = agg_config.policy_fp_u64(); aggregation_configs.insert(agg_id, agg_config); let streaming_config = Arc::new(StreamingConfig {