From 4798b4e198be86ec01746a090884a197a5a78414 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Tue, 1 Sep 2026 20:30:49 -0400 Subject: [PATCH 1/6] fix(precompute): route keyed set aggregators correctly --- .../src/query_logics/enums.rs | 2 + .../precompute_engine/accumulator_factory.rs | 245 ++++++++++++++++-- 2 files changed, 220 insertions(+), 27 deletions(-) diff --git a/asap-common/dependencies/rs/promql_utilities/src/query_logics/enums.rs b/asap-common/dependencies/rs/promql_utilities/src/query_logics/enums.rs index 0411028..6af8c27 100644 --- a/asap-common/dependencies/rs/promql_utilities/src/query_logics/enums.rs +++ b/asap-common/dependencies/rs/promql_utilities/src/query_logics/enums.rs @@ -326,6 +326,8 @@ impl AggregationType { | AggregationType::CountMinSketch | AggregationType::CountMinSketchWithHeap | AggregationType::HydraKLL + | AggregationType::SetAggregator + | AggregationType::DeltaSetAggregator ) } diff --git a/asap-query-engine/src/precompute_engine/accumulator_factory.rs b/asap-query-engine/src/precompute_engine/accumulator_factory.rs index 7044cd2..be1660d 100644 --- a/asap-query-engine/src/precompute_engine/accumulator_factory.rs +++ b/asap-query-engine/src/precompute_engine/accumulator_factory.rs @@ -1,9 +1,9 @@ use crate::data_model::{AggregateCore, AggregationType, KeyByLabelValues, Measurement}; use crate::precompute_operators::{ CountMinSketchAccumulator, CountMinSketchWithHeapAccumulator, DatasketchesKLLAccumulator, - HllAccumulator, HydraKllSketchAccumulator, IncreaseAccumulator, MinMaxAccumulator, - MultipleIncreaseAccumulator, MultipleMinMaxAccumulator, MultipleSumAccumulator, SumAccumulator, - DEFAULT_HLL_PRECISION, + DeltaSetAggregatorAccumulator, HllAccumulator, HydraKllSketchAccumulator, IncreaseAccumulator, + MinMaxAccumulator, MultipleIncreaseAccumulator, MultipleMinMaxAccumulator, + MultipleSumAccumulator, SetAggregatorAccumulator, SumAccumulator, DEFAULT_HLL_PRECISION, }; use asap_types::aggregation_config::AggregationConfig; @@ -43,7 +43,7 @@ pub trait AccumulatorUpdater: Send { /// Feed a single (value, timestamp_ms) pair — for SingleSubpopulation types. fn update_single(&mut self, value: f64, timestamp_ms: i64); - /// Feed a keyed (key, value, timestamp_ms) triple — for MultipleSubpopulation types. + /// Feed a keyed (key, value, timestamp_ms) triple — for keyed aggregation types. fn update_keyed(&mut self, key: &KeyByLabelValues, value: f64, timestamp_ms: i64); /// Extract the final accumulator as a boxed `AggregateCore`. @@ -66,7 +66,7 @@ pub trait AccumulatorUpdater: Send { /// Reset internal state for reuse (avoids re-allocation). fn reset(&mut self); - /// Whether this updater is keyed (MultipleSubpopulation). + /// Whether this updater is keyed (multi-population or key-tracking). fn is_keyed(&self) -> bool; /// Estimated memory usage in bytes. @@ -347,6 +347,112 @@ impl AccumulatorUpdater for HllAccumulatorUpdater { } } +// --------------------------------------------------------------------------- +// SetAggregatorUpdater +// --------------------------------------------------------------------------- + +/// Updater for `AggregationType::SetAggregator`, which records the distinct +/// aggregation keys present in the current window. +pub struct SetAggregatorUpdater { + acc: SetAggregatorAccumulator, +} + +impl SetAggregatorUpdater { + pub fn new() -> Self { + Self { + acc: SetAggregatorAccumulator::new(), + } + } +} + +impl Default for SetAggregatorUpdater { + fn default() -> Self { + Self::new() + } +} + +impl AccumulatorUpdater for SetAggregatorUpdater { + 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.add_key(key.clone()); + } + + impl_accumulator_methods!(acc); + + fn reset(&mut self) { + self.acc = SetAggregatorAccumulator::new(); + } + + fn is_keyed(&self) -> bool { + true + } + + fn memory_usage_bytes(&self) -> usize { + std::mem::size_of::() + + self.acc.added.len() * std::mem::size_of::() + } +} + +// --------------------------------------------------------------------------- +// DeltaSetAggregatorUpdater +// --------------------------------------------------------------------------- + +/// Updater for `AggregationType::DeltaSetAggregator`, which records keys added +/// during the current delta window. Removals are produced by the accumulator +/// merge path when delta windows are combined. +pub struct DeltaSetAggregatorUpdater { + acc: DeltaSetAggregatorAccumulator, +} + +impl DeltaSetAggregatorUpdater { + pub fn new() -> Self { + Self { + acc: DeltaSetAggregatorAccumulator::new(), + } + } +} + +impl Default for DeltaSetAggregatorUpdater { + fn default() -> Self { + Self::new() + } +} + +impl AccumulatorUpdater for DeltaSetAggregatorUpdater { + 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.add_key(key.clone()); + } + + impl_accumulator_methods!(acc); + + fn reset(&mut self) { + self.acc = DeltaSetAggregatorAccumulator::new(); + } + + fn is_keyed(&self) -> bool { + true + } + + fn memory_usage_bytes(&self) -> usize { + std::mem::size_of::() + + (self.acc.added.len() + self.acc.removed.len()) + * std::mem::size_of::() + } +} + // --------------------------------------------------------------------------- // MultipleSumAccumulatorUpdater // --------------------------------------------------------------------------- @@ -679,7 +785,7 @@ impl AccumulatorUpdater for HydraKllAccumulatorUpdater { // Config helpers // --------------------------------------------------------------------------- -/// Return `true` if `config` produces a keyed (MultipleSubpopulation) updater, +/// Return `true` if `config` produces a keyed (multi-population or key-tracking) updater, /// without allocating an updater object. /// /// **Contract:** this must agree with every concrete `AccumulatorUpdater::is_keyed()` @@ -695,6 +801,8 @@ pub fn config_is_keyed(config: &AggregationConfig) -> bool { | AggregationType::CountMinSketch | AggregationType::CountMinSketchWithHeap | AggregationType::HydraKLL + | AggregationType::SetAggregator + | AggregationType::DeltaSetAggregator ) } @@ -819,13 +927,7 @@ pub fn create_accumulator_updater( "DatasketchesKLL" | "datasketches_kll" | "KLL" | "kll" => { Ok(Box::new(KllAccumulatorUpdater::new(kll_k_param(config)?))) } - other => { - tracing::warn!( - "Unknown SingleSubpopulation sub_type '{}', defaulting to Sum", - other - ); - Ok(Box::new(SumAccumulatorUpdater::new())) - } + other => Err(format!("Unknown SingleSubpopulation sub_type '{other}'")), }, AggregationType::MultipleSubpopulation => match sub_type { "Sum" | "sum" => Ok(Box::new(MultipleSumAccumulatorUpdater::new())), @@ -842,13 +944,7 @@ pub fn create_accumulator_updater( row_num, col_num, k, ))) } - other => { - tracing::warn!( - "Unknown MultipleSubpopulation sub_type '{}', defaulting to Sum", - other - ); - Ok(Box::new(MultipleSumAccumulatorUpdater::new())) - } + other => Err(format!("Unknown MultipleSubpopulation sub_type '{other}'")), }, AggregationType::DatasketchesKLL => { Ok(Box::new(KllAccumulatorUpdater::new(kll_k_param(config)?))) @@ -887,13 +983,8 @@ pub fn create_accumulator_updater( AggregationType::HLL => Ok(Box::new(HllAccumulatorUpdater::new(hll_precision_param( config, )))), - other => { - tracing::warn!( - "Unknown aggregation_type '{:?}', defaulting to SingleSubpopulation Sum", - other - ); - Ok(Box::new(SumAccumulatorUpdater::new())) - } + AggregationType::SetAggregator => Ok(Box::new(SetAggregatorUpdater::new())), + AggregationType::DeltaSetAggregator => Ok(Box::new(DeltaSetAggregatorUpdater::new())), } } @@ -1431,6 +1522,28 @@ mod tests { ) } + fn key_aggregation_config(aggregation_type: AggregationType) -> AggregationConfig { + AggregationConfig::new( + 477, + aggregation_type, + String::new(), + std::collections::HashMap::new(), + promql_utilities::data_model::key_by_label_names::KeyByLabelNames::new(vec![]), + promql_utilities::data_model::key_by_label_names::KeyByLabelNames::new(vec![]), + promql_utilities::data_model::key_by_label_names::KeyByLabelNames::new(vec![]), + String::new(), + 60_000, + 0, + WindowType::Tumbling, + "metric".to_string(), + "metric".to_string(), + None, + None, + None, + None, + ) + } + #[test] fn test_cms_with_heap_factory_routes_to_heap_accumulator_and_is_keyed() { // CountMinSketchWithHeap must build a CmsWithHeapAccumulatorUpdater whose @@ -1523,4 +1636,82 @@ mod tests { assert_eq!(cms.query_key(&key), 0.0, "reset must clear the sketch"); assert!(cms.get_topk_keys().is_empty(), "reset must clear the heap"); } + + #[test] + fn test_issue_477_key_aggregator_factory_routes_and_tracks_keys() { + // Regression for #477: planner-generated key aggregations must not fall + // through the factory's scalar Sum updater, or keyed approximate queries + // lose the subpopulation keys needed for enumeration. + let key_a = KeyByLabelValues::new_with_labels(vec!["a".to_string()]); + let key_b = KeyByLabelValues::new_with_labels(vec!["b".to_string()]); + + for (aggregation_type, expected_type_name) in [ + (AggregationType::SetAggregator, "SetAggregatorAccumulator"), + ( + AggregationType::DeltaSetAggregator, + "DeltaSetAggregatorAccumulator", + ), + ] { + let config = key_aggregation_config(aggregation_type); + assert!(config_is_keyed(&config)); + assert!(aggregation_type.is_keyed()); + + let mut updater = create_accumulator_updater(&config).unwrap(); + assert!(updater.is_keyed()); + updater.update_keyed(&key_a, 10.0, 1_000); + updater.update_keyed(&key_b, 20.0, 2_000); + updater.update_keyed(&key_a, 30.0, 3_000); + + let accumulator = updater.take_accumulator(); + assert_eq!(accumulator.type_name(), expected_type_name); + assert_eq!(accumulator.get_accumulator_type(), aggregation_type); + let keys = accumulator + .get_keys() + .expect("key aggregators must enumerate tracked keys"); + assert_eq!(keys.len(), 2); + assert!(keys.contains(&key_a)); + assert!(keys.contains(&key_b)); + } + } + + #[test] + fn test_key_aggregator_updater_reset_clears_keys() { + let key = KeyByLabelValues::new_with_labels(vec!["reset-me".to_string()]); + + for aggregation_type in [ + AggregationType::SetAggregator, + AggregationType::DeltaSetAggregator, + ] { + let config = key_aggregation_config(aggregation_type); + let mut updater = create_accumulator_updater(&config).unwrap(); + updater.update_keyed(&key, 1.0, 0); + updater.reset(); + + assert!( + updater + .snapshot_accumulator() + .get_keys() + .expect("key aggregators must enumerate tracked keys") + .is_empty(), + "reset must clear {aggregation_type:?} keys" + ); + } + } + + #[test] + fn test_factory_rejects_unknown_subpopulation_sub_type() { + let mut config = key_aggregation_config(AggregationType::SingleSubpopulation); + config.aggregation_sub_type = "not-an-aggregation".to_string(); + let err = create_accumulator_updater(&config) + .err() + .expect("unknown subpopulation subtype must not default to Sum"); + assert!(err.contains("Unknown SingleSubpopulation sub_type")); + + let mut config = key_aggregation_config(AggregationType::MultipleSubpopulation); + config.aggregation_sub_type = "not-an-aggregation".to_string(); + let err = create_accumulator_updater(&config) + .err() + .expect("unknown subpopulation subtype must not default to MultipleSum"); + assert!(err.contains("Unknown MultipleSubpopulation sub_type")); + } } From ef0903612f3c7a2cc192baf17bf31d59f2b76c2e Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Tue, 1 Sep 2026 21:03:21 -0400 Subject: [PATCH 2/6] fix(precompute): preserve DeltaSetAggregator window deltas --- .../src/precompute_engine/worker.rs | 157 +++++++++++++++++- 1 file changed, 154 insertions(+), 3 deletions(-) diff --git a/asap-query-engine/src/precompute_engine/worker.rs b/asap-query-engine/src/precompute_engine/worker.rs index f2e151b..6f05cd5 100644 --- a/asap-query-engine/src/precompute_engine/worker.rs +++ b/asap-query-engine/src/precompute_engine/worker.rs @@ -1,4 +1,4 @@ -use crate::data_model::{AggregateCore, KeyByLabelValues, PrecomputedOutput}; +use crate::data_model::{AggregateCore, AggregationType, KeyByLabelValues, PrecomputedOutput}; use crate::precompute_engine::accumulator_factory::{ create_accumulator_updater, AccumulatorUpdater, }; @@ -6,9 +6,10 @@ use crate::precompute_engine::config::LateDataPolicy; use crate::precompute_engine::output_sink::OutputSink; use crate::precompute_engine::series_router::WorkerMessage; use crate::precompute_engine::window_manager::WindowManager; +use crate::precompute_operators::delta_set_aggregator_accumulator::DeltaSetAggregatorAccumulator; use crate::precompute_operators::sum_accumulator::SumAccumulator; use asap_types::aggregation_config::AggregationConfig; -use std::collections::{BTreeMap, HashMap}; +use std::collections::{BTreeMap, HashMap, HashSet}; use std::sync::atomic::{AtomicI64, AtomicUsize, Ordering}; use std::sync::Arc; use tokio::sync::mpsc; @@ -25,6 +26,11 @@ struct GroupState { window_manager: WindowManager, /// Active panes keyed by pane_start_ms. active_panes: BTreeMap>, + /// Key population emitted by the previous non-empty DeltaSetAggregator + /// window for this (aggregation_id, group_key). DeltaSetAggregator outputs + /// are differences between consecutive window populations, so this state + /// must live at group scope rather than inside a single pane updater. + delta_set_previous_keys: HashSet, /// Per-group watermark: tracks the maximum timestamp seen across all /// series in this group on this worker. previous_watermark_ms: i64, @@ -260,7 +266,7 @@ impl Worker { continue; }; - for (group_key_str, state) in inner { + for (group_key_str, mut state) in inner { if state.previous_watermark_ms == i64::MIN { continue; // No samples received — nothing to emit. } @@ -290,6 +296,21 @@ impl Worker { if let Some(accumulator) = merge_panes_for_window(&mut active_panes, &pane_starts) { + let accumulator = match finalize_closed_accumulator( + accumulator, + &state.config, + &mut state.delta_set_previous_keys, + ) { + Ok(accumulator) => accumulator, + Err(e) => { + warn!( + "Worker {}: failed to finalize DeltaSetAggregator \ + for removed agg_id={}: {}", + self.id, agg_id, e + ); + continue; + } + }; let group_key_lv = build_group_key_label_values(&group_key_str); let output = PrecomputedOutput::new( @@ -366,6 +387,7 @@ impl Worker { window_manager: WindowManager::new(config.window_size_ms, config.slide_interval_ms), config, active_panes: BTreeMap::new(), + delta_set_previous_keys: HashSet::new(), previous_watermark_ms: i64::MIN, pane_wall_clock_last_touch_ms: BTreeMap::new(), }; @@ -499,6 +521,11 @@ impl Worker { if let Some(accumulator) = merge_panes_for_window(&mut state.active_panes, &pane_starts) { + let accumulator = finalize_closed_accumulator( + accumulator, + &state.config, + &mut state.delta_set_previous_keys, + )?; let key = build_group_key_label_values(group_key); let output = PrecomputedOutput::new( *window_start as u64, @@ -644,6 +671,11 @@ impl Worker { if let Some(accumulator) = merge_panes_for_window(&mut state.active_panes, &pane_starts) { + let accumulator = finalize_closed_accumulator( + accumulator, + &state.config, + &mut state.delta_set_previous_keys, + )?; let key = build_group_key_label_values(group_key); let output = PrecomputedOutput::new( *window_start as u64, @@ -728,6 +760,11 @@ impl Worker { if let Some(accumulator) = merge_panes_for_window(&mut state.active_panes, &pane_starts) { + let accumulator = finalize_closed_accumulator( + accumulator, + &state.config, + &mut state.delta_set_previous_keys, + )?; let key = build_group_key_label_values(group_key); let output = PrecomputedOutput::new( *window_start as u64, @@ -1036,6 +1073,52 @@ fn merge_panes_for_window( merged } +/// Convert a closed DeltaSetAggregator population into the stateful delta +/// format used by the Arroyo implementation: keys newly present in this +/// window go in `added`, and keys absent from this window go in `removed`. +/// +/// The updater can only collect the current window's observed keys. The +/// previous population therefore belongs to `GroupState`, which survives the +/// per-pane updater lifecycle and is isolated for each `(agg_id, group_key)`. +fn finalize_closed_accumulator( + accumulator: Box, + config: &AggregationConfig, + previous_delta_set_keys: &mut HashSet, +) -> Result, Box> { + if config.aggregation_type != AggregationType::DeltaSetAggregator { + return Ok(accumulator); + } + + let delta = accumulator + .as_any() + .downcast_ref::() + .ok_or_else(|| { + format!( + "DeltaSetAggregator config produced {} instead of DeltaSetAggregatorAccumulator", + accumulator.type_name() + ) + })?; + let current_keys: HashSet = delta + .get_keys() + .ok_or("DeltaSetAggregator accumulator could not resolve its current keys")? + .into_iter() + .collect(); + + let added = current_keys + .difference(previous_delta_set_keys) + .cloned() + .collect(); + let removed = previous_delta_set_keys + .difference(¤t_keys) + .cloned() + .collect(); + *previous_delta_set_keys = current_keys; + + Ok(Box::new(DeltaSetAggregatorAccumulator::new_with_sets( + added, removed, + ))) +} + #[cfg(test)] mod tests { use super::*; @@ -1240,6 +1323,7 @@ mod tests { use crate::precompute_engine::config::LateDataPolicy; use crate::precompute_engine::output_sink::CapturingOutputSink; use crate::precompute_operators::datasketches_kll_accumulator::DatasketchesKLLAccumulator; + use crate::precompute_operators::delta_set_aggregator_accumulator::DeltaSetAggregatorAccumulator; use crate::precompute_operators::multiple_sum_accumulator::MultipleSumAccumulator; use crate::precompute_operators::sum_accumulator::SumAccumulator; use asap_sketchlib::KllSketch; @@ -1446,6 +1530,73 @@ mod tests { ); } + #[test] + fn test_delta_set_aggregator_emits_changes_relative_to_previous_window() { + // Regression for the stateful DeltaSetAggregator contract: each + // non-empty window emits keys added to or removed from the previous + // window, matching asap-summary-ingest's Arroyo UDAF. + let config = make_agg_config_full( + 2, + "cpu", + AggregationType::DeltaSetAggregator, + "", + 1_000, + 1_000, + vec![], + vec!["host"], + ); + let mut agg_configs = HashMap::new(); + agg_configs.insert(2, config); + + let sink = Arc::new(CapturingOutputSink::new()); + let mut worker = make_worker( + arc_configs(agg_configs), + sink.clone(), + false, + 0, + LateDataPolicy::Drop, + ); + + worker + .process_group_samples(2, "", vec![("cpu{host=\"a\"}".to_string(), 100, 1.0)]) + .unwrap(); + worker + .process_group_samples(2, "", vec![("cpu{host=\"b\"}".to_string(), 1_000, 1.0)]) + .unwrap(); + + let first_window = sink + .drain() + .into_iter() + .find(|(output, _)| output.start_timestamp == 0) + .expect("first DeltaSetAggregator window should be emitted"); + let first_delta = first_window + .1 + .as_any() + .downcast_ref::() + .expect("first output should be DeltaSetAggregatorAccumulator"); + let key_a = KeyByLabelValues::new_with_labels(vec!["a".to_string()]); + assert!(first_delta.added.contains(&key_a)); + assert!(first_delta.removed.is_empty()); + + worker + .process_group_samples(2, "", vec![("cpu{host=\"b\"}".to_string(), 2_000, 1.0)]) + .unwrap(); + + let second_window = sink + .drain() + .into_iter() + .find(|(output, _)| output.start_timestamp == 1_000) + .expect("second DeltaSetAggregator window should be emitted"); + let second_delta = second_window + .1 + .as_any() + .downcast_ref::() + .expect("second output should be DeltaSetAggregatorAccumulator"); + let key_b = KeyByLabelValues::new_with_labels(vec!["b".to_string()]); + assert!(second_delta.added.contains(&key_b)); + assert!(second_delta.removed.contains(&key_a)); + } + // ----------------------------------------------------------------------- // Test: GROUP BY — multiple series merged into same group accumulator // ----------------------------------------------------------------------- From 1e1abe1b1ef42377b604573a261691dea2e81348 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Tue, 1 Sep 2026 21:34:01 -0400 Subject: [PATCH 3/6] fix(precompute): drop late DeltaSetAggregator samples --- .../src/precompute_engine/config.rs | 2 + .../precompute_engine_design_doc.md | 37 +++++++---- .../src/precompute_engine/worker.rs | 65 +++++++++++++++++++ 3 files changed, 91 insertions(+), 13 deletions(-) diff --git a/asap-query-engine/src/precompute_engine/config.rs b/asap-query-engine/src/precompute_engine/config.rs index b5f1268..4fcc300 100644 --- a/asap-query-engine/src/precompute_engine/config.rs +++ b/asap-query-engine/src/precompute_engine/config.rs @@ -6,6 +6,8 @@ pub enum LateDataPolicy { /// Drop late samples that arrive after their window has closed. Drop, /// Forward late samples to the store to be merged with existing window data. + /// Unsupported for `DeltaSetAggregator`, whose stateful key deltas cannot + /// be repaired by appending an independent mini-accumulator. ForwardToStore, } diff --git a/asap-query-engine/src/precompute_engine/precompute_engine_design_doc.md b/asap-query-engine/src/precompute_engine/precompute_engine_design_doc.md index 55070a1..2aa97f7 100644 --- a/asap-query-engine/src/precompute_engine/precompute_engine_design_doc.md +++ b/asap-query-engine/src/precompute_engine/precompute_engine_design_doc.md @@ -13,7 +13,7 @@ and VictoriaMetrics remote write), buffers them, computes windowed aggregations - Watermark-based windowed aggregation (tumbling and sliding windows) - Shared-nothing worker design: series are hash-partitioned across threads with no cross-worker coordination - Pluggable accumulator types (Sum, Min/Max, Increase, KLL, CMS, HydraKLL) -- Configurable late-data handling (Drop or ForwardToStore) +- Configurable late-data handling (Drop or ForwardToStore for supported aggregators) - Optional raw passthrough mode for bypassing aggregation ## 2. Architecture @@ -100,7 +100,7 @@ Late data handling: │ Watermark W=8 ─┘ t=3 < W - allowed_lateness(2) = 6? - 3 < 6 → YES, late → DROP (or ForwardToStore) + 3 < 6 → YES, late → DROP (or ForwardToStore for supported aggregators) ``` ### Cross-group watermark propagation @@ -205,7 +205,7 @@ pub struct PrecomputeEngineConfig { pub enum LateDataPolicy { Drop, // Silently discard late samples for closed windows - ForwardToStore, // Emit a mini-accumulator for query-time merge + ForwardToStore, // Emit a mini-accumulator for query-time merge (not DeltaSetAggregator) } ``` @@ -396,7 +396,7 @@ from `active_panes`. Remaining panes are read non-destructively via a. Compute pane_start = pane_start_for(ts) b. If pane was evicted (late data for closed window): → late_data_policy == Drop: skip - → late_data_policy == ForwardToStore: create mini-accumulator, emit + → late_data_policy == ForwardToStore: create mini-accumulator, emit (except DeltaSetAggregator, which drops) c. Else: get-or-create pane in active_panes, feed value (1 update per sample) 5. Detect newly closed windows via closed_windows(prev_wm, current_wm) 6. For each closed window: @@ -585,7 +585,7 @@ Workers have independent watermarks. For a standard Prometheus scrape (all insta For staggered multi-source producers arriving at different times, the incompleteness window is bounded by the spread of producer arrival times. In both cases the result is **eventually consistent**: once all contributing workers have emitted, the store holds a complete set of accumulators and queries return the correct merged value. -This deferred-merge design is intentional — it preserves the shared-nothing worker architecture with zero ingest-time cross-worker coordination. The store's append-multiple-per-window design and the query-time merge handle the fan-in correctly for both cross-series aggregation and `ForwardToStore` late data. +This deferred-merge design is intentional — it preserves the shared-nothing worker architecture with zero ingest-time cross-worker coordination. The store's append-multiple-per-window design and the query-time merge handle the fan-in correctly for both cross-series aggregation and supported `ForwardToStore` late data. ### Sliding windows with cross-worker GROUP BY @@ -795,12 +795,18 @@ When `LateDataPolicy::ForwardToStore` is active and a late sample falls into an already-closed (and potentially already-merged) window, the first-tier worker emits a `PartialWindowAggregate` for that window as it does today. +This policy is not supported for `DeltaSetAggregator`. Its output is a +stateful difference between consecutive key populations, so an independent +late mini-accumulator cannot be appended safely. The worker drops such late +samples and logs a warning. + The merge tier treats these late partials as **store appends**, not as corrections to a finalized canonical entry. The canonical merged output written at finalization time remains unchanged. The store accumulates the late partial alongside it, and query-time `SummaryMergeMultipleExec` merges them on read. -This is consistent with the existing `ForwardToStore` semantics and avoids the +For supported aggregation types, this is consistent with the existing +`ForwardToStore` semantics and avoids the need to read-modify-write a finalized store entry. The benefit of the merge tier (one canonical output per window) applies only to on-time data; late corrections fall back to the same append + query-time-merge path as the non-merge-tier @@ -921,7 +927,8 @@ With the merge tier enabled for an aggregation: This removes the current ambiguity where the store can return multiple exact matches for one logical output window and the query layer must decide whether to merge or pick one. Late corrections (via `ForwardToStore`) continue to use -query-time merge for their incremental updates. +query-time merge for their incremental updates. `DeltaSetAggregator` is the +exception: late samples are dropped rather than forwarded. #### Why a separate merge worker is preferable to routing by grouping key @@ -1054,8 +1061,9 @@ struct StoreKeyData { ``` Multiple entries per `(start_ts, end_ts)` are allowed — they are appended, not -overwritten. This is what makes `ForwardToStore` late-data policy work: the late -mini-accumulator is stored alongside the original window accumulator. +overwritten. This is what makes the supported `ForwardToStore` late-data policy +work: the late mini-accumulator is stored alongside the original window +accumulator. `DeltaSetAggregator` does not use this path. ### Read path / query-time merge @@ -1091,10 +1099,12 @@ For case 2, the `LateDataPolicy` controls behavior: - **Drop**: log at debug level and skip. No ghost accumulator is created (fixing the original bug where `or_insert_with` would create orphaned entries). -- **ForwardToStore**: create a fresh `AccumulatorUpdater`, feed the single - late sample, wrap as `PrecomputedOutput`, and push into the same `emit_batch` - as normal closed-window outputs. The store appends it alongside the original - window data, and query-time merge combines them. +- **ForwardToStore** (except `DeltaSetAggregator`): create a fresh + `AccumulatorUpdater`, feed the single late sample, wrap as `PrecomputedOutput`, + and push into the same `emit_batch` as normal closed-window outputs. The store + appends it alongside the original window data, and query-time merge combines + them. `DeltaSetAggregator` drops the sample because an append-only correction + cannot preserve stateful deltas. ## 8. Concurrency Model @@ -1176,6 +1186,7 @@ store with the Kafka consumer path. | `test_groupby_separate_emits_per_series` | Two series (`host=A`, `host=B`) on same worker -> 2 independent `MultipleSumAccumulator` emits (no ingest-time cross-series merge) | | `test_late_data_drop` | Sample behind `watermark - allowed_lateness_ms` with `Drop` policy -> 0 emits | | `test_late_data_forward_to_store` | Late sample for evicted pane with `ForwardToStore` -> 1 emit as mini-accumulator with correct window bounds and sum | + | `test_late_data_forward_to_store_drops_delta_set_aggregator` | Late sample for evicted `DeltaSetAggregator` pane with `ForwardToStore` -> 0 emits | - **Unit tests -- other modules**: `window_manager.rs` (tumbling/sliding arithmetic, pane enumeration, closure detection), `series_buffer.rs` (ordering, watermark), `accumulator_factory.rs` (updater creation and reset), `series_router.rs` (consistent hash routing), `config.rs` (defaults). diff --git a/asap-query-engine/src/precompute_engine/worker.rs b/asap-query-engine/src/precompute_engine/worker.rs index 6f05cd5..b3b79db 100644 --- a/asap-query-engine/src/precompute_engine/worker.rs +++ b/asap-query-engine/src/precompute_engine/worker.rs @@ -472,6 +472,16 @@ impl Worker { ); continue; } + LateDataPolicy::ForwardToStore + if state.config.aggregation_type == AggregationType::DeltaSetAggregator => + { + warn!( + "Dropping late DeltaSetAggregator sample for evicted pane [{}, {}): ForwardToStore is unsupported for stateful key deltas", + pane_start, + pane_end + ); + continue; + } LateDataPolicy::ForwardToStore => { let mut updater = create_accumulator_updater(&state.config)?; apply_sample(&mut *updater, series_key, *val, *ts, &state.config); @@ -2464,6 +2474,61 @@ mod tests { ); } + #[test] + fn test_late_data_forward_to_store_drops_delta_set_aggregator() { + let config = make_agg_config_full( + 6, + "cpu", + AggregationType::DeltaSetAggregator, + "", + 10_000, + 0, + vec![], + vec!["host"], + ); + let mut agg_configs = HashMap::new(); + agg_configs.insert(6, config); + + let sink = Arc::new(CapturingOutputSink::new()); + let (_tx, rx) = tokio::sync::mpsc::channel(1); + let wm = Arc::new(AtomicI64::new(i64::MIN)); + let mut worker = Worker::new( + 0, + rx, + sink.clone(), + arc_configs(agg_configs), + WorkerRuntimeConfig { + max_buffer_per_series: 10_000, + allowed_lateness_ms: 15_000, + pass_raw_samples: false, + raw_mode_aggregation_id: 0, + late_data_policy: LateDataPolicy::ForwardToStore, + wall_clock_grace_period_ms: 0, + }, + Arc::new(AtomicUsize::new(0)), + wm.clone(), + vec![wm], + ); + + worker + .process_group_samples(6, "", group_samples("cpu{host=\"a\"}", vec![(500, 1.0)])) + .unwrap(); + worker + .process_group_samples(6, "", group_samples("cpu{host=\"a\"}", vec![(20_000, 0.0)])) + .unwrap(); + let _ = sink.drain(); + + worker + .process_group_samples(6, "", group_samples("cpu{host=\"b\"}", vec![(8_000, 55.0)])) + .unwrap(); + + assert_eq!( + sink.len(), + 0, + "ForwardToStore must drop late DeltaSetAggregator samples" + ); + } + // ----------------------------------------------------------------------- // Test: worker from streaming_config YAML // ----------------------------------------------------------------------- From 2bc39f5fe7129bc1fa5c2bffa6c38630d5119194 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 2 Sep 2026 08:13:56 -0400 Subject: [PATCH 4/6] fix(precompute): handle first-batch window ordering --- .../src/precompute_engine/worker.rs | 74 ++++++++++++++++++- 1 file changed, 72 insertions(+), 2 deletions(-) diff --git a/asap-query-engine/src/precompute_engine/worker.rs b/asap-query-engine/src/precompute_engine/worker.rs index b3b79db..919f53b 100644 --- a/asap-query-engine/src/precompute_engine/worker.rs +++ b/asap-query-engine/src/precompute_engine/worker.rs @@ -435,12 +435,25 @@ impl Worker { .map(|(_, ts, _)| *ts) .max() .unwrap_or(i64::MIN); + let batch_min_ts = samples + .iter() + .map(|(_, ts, _)| *ts) + .min() + .unwrap_or(batch_max_ts); let previous_wm = state.previous_watermark_ms; let current_wm = if batch_max_ts > previous_wm { batch_max_ts } else { previous_wm }; + // On the first batch there is no prior watermark. Use the earliest + // sample as the closure baseline so a batch spanning multiple windows + // can close windows older than that sample after all samples are routed. + let closure_previous_wm = if previous_wm == i64::MIN { + batch_min_ts + } else { + previous_wm + }; let mut emit_batch: Vec<(PrecomputedOutput, Box)> = Vec::new(); @@ -460,7 +473,7 @@ impl Worker { // Check if pane was already evicted (late data for a closed window) if !state.active_panes.contains_key(&pane_start) - && current_wm >= pane_start + state.window_manager.window_size_ms() + && previous_wm >= pane_start + state.window_manager.window_size_ms() { let window_start = pane_start; let window_end = pane_start + state.window_manager.window_size_ms(); @@ -523,7 +536,9 @@ impl Worker { } // Check for closed windows - let closed = state.window_manager.closed_windows(previous_wm, current_wm); + let closed = state + .window_manager + .closed_windows(closure_previous_wm, current_wm); for window_start in &closed { let (_, window_end) = state.window_manager.window_bounds(*window_start); @@ -2529,6 +2544,61 @@ mod tests { ); } + #[test] + fn test_delta_set_aggregator_keeps_on_time_samples_in_first_multi_window_batch() { + let config = make_agg_config_full( + 7, + "cpu", + AggregationType::DeltaSetAggregator, + "", + 10_000, + 0, + vec![], + vec!["host"], + ); + let mut agg_configs = HashMap::new(); + agg_configs.insert(7, config); + + let sink = Arc::new(CapturingOutputSink::new()); + let mut worker = make_worker( + arc_configs(agg_configs), + sink.clone(), + false, + 0, + LateDataPolicy::ForwardToStore, + ); + + worker + .process_group_samples( + 7, + "", + vec![ + ("cpu{host=\"a\"}".to_string(), 500, 1.0), + ("cpu{host=\"b\"}".to_string(), 20_000, 1.0), + ], + ) + .unwrap(); + worker + .process_group_samples(7, "", vec![("cpu{host=\"b\"}".to_string(), 30_000, 1.0)]) + .unwrap(); + + let first_window = sink + .drain() + .into_iter() + .find(|(output, _)| output.start_timestamp == 0) + .expect("first DeltaSetAggregator window should be emitted"); + let first_delta = first_window + .1 + .as_any() + .downcast_ref::() + .expect("first output should be DeltaSetAggregatorAccumulator"); + let key_a = KeyByLabelValues::new_with_labels(vec!["a".to_string()]); + assert!( + first_delta.added.contains(&key_a), + "on-time key in the first batch must not be treated as late" + ); + } + // ----------------------------------------------------------------------- // Test: worker from streaming_config YAML // ----------------------------------------------------------------------- From a362aba79e37a759ec8233249e1e50a76d5db620 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 2 Sep 2026 09:29:08 -0400 Subject: [PATCH 5/6] docs(precompute): clarify delta set updater comment --- .../src/precompute_engine/accumulator_factory.rs | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/asap-query-engine/src/precompute_engine/accumulator_factory.rs b/asap-query-engine/src/precompute_engine/accumulator_factory.rs index be1660d..7387927 100644 --- a/asap-query-engine/src/precompute_engine/accumulator_factory.rs +++ b/asap-query-engine/src/precompute_engine/accumulator_factory.rs @@ -403,9 +403,11 @@ impl AccumulatorUpdater for SetAggregatorUpdater { // DeltaSetAggregatorUpdater // --------------------------------------------------------------------------- -/// Updater for `AggregationType::DeltaSetAggregator`, which records keys added -/// during the current delta window. Removals are produced by the accumulator -/// merge path when delta windows are combined. +/// Updater for `AggregationType::DeltaSetAggregator`, which records keys observed +/// during the current window. The worker's window-finalization step compares that +/// population with the previous window to produce added and removed keys; the +/// accumulator merge path only preserves the correct state when delta buckets are +/// combined. pub struct DeltaSetAggregatorUpdater { acc: DeltaSetAggregatorAccumulator, } From baccd300e5066e1e495546a556059b8403bd3221 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 2 Sep 2026 09:47:30 -0400 Subject: [PATCH 6/6] fix(planner): include benchmark binary in Docker cache layer --- asap-planner-rs/Dockerfile | 1 + 1 file changed, 1 insertion(+) diff --git a/asap-planner-rs/Dockerfile b/asap-planner-rs/Dockerfile index 59dabd0..438955e 100644 --- a/asap-planner-rs/Dockerfile +++ b/asap-planner-rs/Dockerfile @@ -21,6 +21,7 @@ RUN mkdir -p asap-query-engine/src && echo "fn main() {}" > asap-query-engine/sr mkdir -p asap-planner-rs/src/bin && echo "fn main() {}" > asap-planner-rs/src/main.rs && \ echo "fn main() {}" > asap-planner-rs/src/bin/optimizer_cli.rs && \ echo "fn main() {}" > asap-planner-rs/src/bin/candidate_gen_dump.rs && \ + echo "fn main() {}" > asap-planner-rs/src/bin/benchmark_promql_status.rs && \ echo "pub fn placeholder() {}" >> asap-planner-rs/src/lib.rs # Build dependencies (this layer will be cached)