From 4b21b99da535248dbf69fbca80821c65b15f1352 Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 21 Jul 2026 08:06:40 -0600 Subject: [PATCH 1/3] feat(asap_types, data_plane): split AggregationType into AccumulatorSpec Step 5 of the sketch-identity unification (scratchpad/artifacts/enum- unification-plan.md, section 6/8). data_plane's AggregationConfig has represented "which accumulator to run, on what, with what parameters" as three loosely-typed things: a 16-variant AggregationType enum that conflates sketch identity with the keyed/unkeyed axis, a raw aggregation_sub_type string consulted only for two "wrapper" variants, and an untyped parameters: HashMap bag. The real dispatch (accumulator_factory.rs::create_accumulator_updater) was a 14-arm match with a second string-typed dispatch layer underneath it. This adds asap_types::AccumulatorSpec { kind: SummaryKind, params: SummaryParams, grouping: Option }, converging data_plane onto the same asap_sketch::SummaryKind/SummaryParams representation control_plane already uses (Stage 3, merged), extended with the keyed/unkeyed grouping axis AggregationType wrongly folded into identity. accumulator_factory.rs now dispatches on AccumulatorSpec.kind instead of the AggregationType + string combo; the SingleSubpopulation/MultipleSubpopulation string-matching arms collapse away now that grouping is a sibling field. Every one of the old 14 match arms' behavior is preserved exactly, including the CMS- heap vs bare-CMS distinction, the bare-CountSketch-shares-CmsAccumulator quirk, and all three fallback/unknown-warning paths (same warning text, same default updater per path). Two decisions worth flagging for review: - Additive, not a replacement. AggregationConfig keeps its aggregation_type/aggregation_sub_type/parameters fields untouched. PolicyFingerprint::from_config hashes those three fields directly and its own module doc calls the byte layout it produces a stability contract ("any such change invalidates every deployed fingerprint and forces a cold-start rebuild") -- so policy_fingerprint.rs is not touched by this change at all. Separately, AggregationType turned out to be read by ~40 files across data_plane/asap_types (query-time capability matching, persistence, reconciliation, index maintenance) well beyond accumulator_factory.rs, so full removal was judged too large to land and review safely in one PR. AccumulatorSpec is computed on demand from AggregationConfig's existing fields via AggregationConfig::accumulator_spec(); full removal of the old fields is follow-up work, not done here. - Three details don't fit asap_sketch's upstream types and still read AggregationConfig/its parameters map directly, documented in accumulator_spec.rs's module doc: min/max direction (SummaryParams:: MinMax carries no fields), HydraKLL's (row, col) tiling grid (SummaryParams::Kll carries only k), and top-k weight_mode (no upstream concept at all). The wire format (aggregationType/aggregationSubType/parameters JSON and YAML keys) is unaffected -- AggregationConfig::from_yaml/from_json still parse those key names generically, unchanged. cargo test -p asap_types: 78 passed (18 new, covering every AggregationType variant's resolution, both wrapper sub_type alias lists, all three error paths, and a fingerprint-stability regression guard). cargo test -p data_plane --lib: 881 passed, 2 ignored -- no change from the pre-existing baseline. cargo build --workspace: clean. Co-Authored-By: Claude Sonnet 5 --- crates/asap_types/Cargo.toml | 9 + crates/asap_types/src/accumulator_spec.rs | 677 ++++++++++++++++++ crates/asap_types/src/lib.rs | 2 + .../precompute_engine/accumulator_factory.rs | 325 +++++---- 4 files changed, 863 insertions(+), 150 deletions(-) create mode 100644 crates/asap_types/src/accumulator_spec.rs diff --git a/crates/asap_types/Cargo.toml b/crates/asap_types/Cargo.toml index c3b6b683..b1ca38c9 100644 --- a/crates/asap_types/Cargo.toml +++ b/crates/asap_types/Cargo.toml @@ -17,3 +17,12 @@ xxhash-rust = { version = "0.8", features = ["xxh64"] } # unification (scratchpad/artifacts/enum-unification-plan.md). Pin matches # control_plane's -- see control_plane/Cargo.toml's comment for the rationale. asap-ir = { git = "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/ProjectASAP/ASAPController", rev = "01745cceac857be21fd1d80584c045e6f932ebfc" } +# Step 5 of the sketch-identity unification (see +# scratchpad/artifacts/enum-unification-plan.md): `AccumulatorSpec` +# (accumulator_spec.rs) converges data_plane's identity representation +# onto ASAPController's `SummaryKind`/`SummaryParams`, same as +# control_plane already does (Stage 3, merged). Pin MUST match +# control_plane's pin exactly (`control_plane/Cargo.toml`) -- two +# different revs of the same git dependency in one workspace resolve +# to two distinct Rust types that won't unify. +asap-sketch = { git = "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/ProjectASAP/ASAPController", rev = "01745cceac857be21fd1d80584c045e6f932ebfc" } diff --git a/crates/asap_types/src/accumulator_spec.rs b/crates/asap_types/src/accumulator_spec.rs new file mode 100644 index 00000000..a2064c37 --- /dev/null +++ b/crates/asap_types/src/accumulator_spec.rs @@ -0,0 +1,677 @@ +//! `AccumulatorSpec` — data_plane's replacement for the +//! `AggregationType` + `aggregation_sub_type: String` + untyped +//! `parameters: HashMap` triple. +//! +//! **Step 5 of the sketch-identity unification** (see +//! `scratchpad/artifacts/enum-unification-plan.md`, §7-8). Converges +//! accumulator *identity* onto ASAPController's `asap_sketch::SummaryKind` +//! / `SummaryParams` — the same representation `control_plane` already +//! uses as of Stage 3 (merged) — extended with the one axis that +//! representation doesn't have: keyed-vs-unkeyed grouping, which +//! `AggregationType` wrongly folded into identity (`Sum` vs +//! `MultipleSum`, etc.) instead of modeling as a sibling field. +//! +//! ## This is an additive representation, not a replacement (yet) +//! +//! `AggregationConfig` keeps its `aggregation_type` / `aggregation_sub_type` +//! / `parameters` fields untouched. Two hard constraints ruled out full +//! removal in this pass: +//! +//! 1. **`PolicyFingerprint` hash stability.** [`crate::policy_fingerprint`] +//! hashes `aggregation_type` / `aggregation_sub_type` / `parameters` +//! directly, and its own module doc is explicit that the byte layout +//! it produces is a stability *contract* ("Don't reorder fields... +//! any such change invalidates every deployed fingerprint and forces +//! a cold-start rebuild"). Changing what feeds that hash — even by +//! routing it through an equivalent typed shape — risks producing a +//! different byte sequence for the same logical policy, which strands +//! on-disk sids after a deploy. `policy_fingerprint.rs` is +//! deliberately **not touched** by this module; it keeps reading the +//! original three fields, unchanged. +//! 2. **Consumer fan-out.** `AggregationType` is read by ~40 files across +//! `data_plane` and `asap_types` — persistence (`sid_metadata.json` +//! round-trip), query-time capability matching +//! (`capability_matching.rs`, unrelated to accumulator dispatch), +//! the query engine, reconciliation, index maintenance — not just +//! `accumulator_factory.rs` (the single highest-risk consumer, and +//! the one this module targets). Migrating all of them in one PR was +//! judged too large to land and review safely; that's tracked as +//! follow-up, not done here. +//! +//! So: `AccumulatorSpec` is *computed from* `AggregationConfig`'s +//! existing fields via [`AggregationConfig::accumulator_spec`], and +//! consumed by `data_plane::precompute_engine::accumulator_factory` +//! instead of the raw fields. The wire format (`aggregationType` / +//! `aggregationSubType` / `parameters` JSON/YAML keys) is completely +//! unaffected — nothing here changes how `AggregationConfig::from_yaml` +//! / `from_json` parse or how `serialize_to_json` emits. +//! +//! ## What doesn't fit `SummaryKind`/`SummaryParams` +//! +//! `asap_sketch`'s types are ASAPController's, not ours to extend from +//! this repo, and two data_plane-specific details don't fit them: +//! +//! - **Min/max direction.** `SummaryParams::MinMax` carries no fields — +//! upstream doesn't model a direction axis. `accumulator_factory.rs` +//! keeps reading `AggregationConfig::aggregation_sub_type` directly +//! for this one bit (`eq_ignore_ascii_case("max")`), exactly as it did +//! before this refactor. +//! - **HydraKLL's `(row, col)` tiling.** `SummaryParams::Kll` carries +//! only `k` — upstream has no concept of the CMS-like grid-of-KLL-cells +//! layout `HydraKllSketchAccumulator` uses to parallelize a keyed KLL +//! across many populations. `accumulator_factory.rs` calls +//! [`cms_params`] directly for the `(SummaryKind::Kll, keyed=true)` +//! arm, same extraction the plain CMS arms use, because `w`/`d` are +//! genuinely the same wire keys for both. +//! - **Top-k ranking mode (`weight_mode`).** Not a sketch structural +//! parameter — a data_plane-only "what to accumulate" axis +//! (`accumulator_factory::TopkWeight`) with no upstream equivalent. +//! Stays a raw-`parameters`-reading helper in `accumulator_factory.rs`. + +use serde_json::Value; + +use crate::aggregation_config::AggregationConfig; +use crate::key_by_label_names::KeyByLabelNames; +use promql_utilities::query_logics::enums::AggregationType; + +pub use asap_sketch::{SummaryKind, SummaryParams}; + +/// Data_plane's typed replacement for +/// `(aggregation_type, aggregation_sub_type, parameters)`: which +/// accumulator to run (`kind`), with what tuning (`params`), and +/// whether it's keyed by a group-by label set (`grouping`). +/// +/// Computed on demand from an [`AggregationConfig`] via +/// [`AggregationConfig::accumulator_spec`] — not stored on the config +/// itself, so there is exactly one source of truth for the fields that +/// feed [`crate::policy_fingerprint::PolicyFingerprint`]. +#[derive(Debug, Clone, PartialEq)] +pub struct AccumulatorSpec { + /// Which accumulator family — identity only (no keyed/unkeyed axis, + /// no heap-vs-bare ambiguity: heap-bearing sketches are their own + /// `SummaryKind` variant, e.g. `CmsWithHeap` vs `Cms`). + pub kind: SummaryKind, + /// Typed tuning parameters matching `kind` (no `HashMap` lookups — + /// see the module doc for the handful of details that still need + /// one, kept in `accumulator_factory.rs` since `SummaryParams` has + /// no field for them). + pub params: SummaryParams, + /// `Some(labels)` for a keyed (multi-population) accumulator, + /// `None` for a single-population one. This is the axis + /// `AggregationType` wrongly folded into identity (`Sum` vs + /// `MultipleSum`) — here it's a sibling field instead. + pub grouping: Option, +} + +/// Why [`AggregationConfig::accumulator_spec`] couldn't resolve a config +/// into an [`AccumulatorSpec`]. Each variant matches one of the three +/// distinct fallback paths `accumulator_factory::create_accumulator_updater` +/// took pre-Step-5 — preserved verbatim (including which default +/// updater and which warning text each one produced) so this refactor +/// changes *how* the dispatch is expressed, not what it does for any +/// input. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum AccumulatorSpecError { + /// `aggregation_type` was `SingleSubpopulation` with an + /// `aggregation_sub_type` string not in the recognized alias list. + /// Pre-Step-5 this defaulted to `SumAccumulatorUpdater`. + UnknownSingleSubpopulationSubType(String), + /// `aggregation_type` was `MultipleSubpopulation` with an + /// unrecognized `aggregation_sub_type`. Pre-Step-5 this defaulted + /// to `MultipleSumAccumulatorUpdater` (note: a *different* default + /// than the `SingleSubpopulation` case). + UnknownMultipleSubpopulationSubType(String), + /// `aggregation_type` itself has no accumulator-dispatch mapping. + /// Today this is only ever `AggregationType::HLL` — it's a real + /// `SummaryKind::Hll` identity and `control_plane` can emit + /// `aggregationType: HLL` on the wire, but + /// `accumulator_factory::create_accumulator_updater` never grew a + /// real HLL arm (HLL accumulators are built via the SketchEnvelope + /// ingest path instead, bypassing raw-value dispatch). Pre-existing + /// gap, not introduced by this refactor — preserved as-is. + UnmappedAggregationType(AggregationType), +} + +impl std::fmt::Display for AccumulatorSpecError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::UnknownSingleSubpopulationSubType(s) => { + write!( + f, + "Unknown SingleSubpopulation sub_type '{s}', defaulting to Sum" + ) + } + Self::UnknownMultipleSubpopulationSubType(s) => { + write!( + f, + "Unknown MultipleSubpopulation sub_type '{s}', defaulting to Sum" + ) + } + Self::UnmappedAggregationType(t) => write!( + f, + "Unknown aggregation_type '{t:?}', defaulting to SingleSubpopulation Sum" + ), + } + } +} + +impl std::error::Error for AccumulatorSpecError {} + +impl AggregationConfig { + /// Resolve this config's `(aggregation_type, aggregation_sub_type, + /// parameters)` triple into a typed [`AccumulatorSpec`]. + /// + /// Mirrors `accumulator_factory::create_accumulator_updater`'s + /// pre-Step-5 dispatch exactly — same sub_type alias lists, same + /// numeric defaults, same three fallback paths (see + /// [`AccumulatorSpecError`]) — just re-expressed as data instead of + /// as a 14-arm match baked into the accumulator constructor. + pub fn accumulator_spec(&self) -> Result { + use AggregationType::*; + + let sub_type = self.aggregation_sub_type.as_str(); + + let (kind, params, keyed): (SummaryKind, SummaryParams, bool) = match self.aggregation_type + { + Sum => (SummaryKind::Sum, SummaryParams::Sum, false), + Increase => (SummaryKind::Increase, SummaryParams::Increase, false), + MinMax => (SummaryKind::MinMax, SummaryParams::MinMax, false), + DatasketchesKLL => ( + SummaryKind::Kll, + SummaryParams::Kll { + k: kll_k_param(self) as u32, + }, + false, + ), + MultipleSum => (SummaryKind::Sum, SummaryParams::Sum, true), + MultipleIncrease => (SummaryKind::Increase, SummaryParams::Increase, true), + MultipleMinMax => (SummaryKind::MinMax, SummaryParams::MinMax, true), + HydraKLL => ( + SummaryKind::Kll, + SummaryParams::Kll { + k: kll_k_param(self) as u32, + }, + true, + ), + CountMinSketch => { + let (row_num, col_num) = cms_params(self); + ( + SummaryKind::Cms, + SummaryParams::Cms { + width: col_num as u32, + depth: row_num as u32, + }, + true, + ) + } + CountMinSketchWithHeap => { + let (row_num, col_num) = cms_params(self); + let heap_size = heap_size_param(self); + ( + SummaryKind::CmsWithHeap, + SummaryParams::CmsWithHeap { + width: col_num as u32, + depth: row_num as u32, + heap_size: heap_size as u32, + }, + true, + ) + } + CountSketch => { + let (row_num, col_num) = cms_params(self); + ( + SummaryKind::CountSketch, + SummaryParams::CountSketch { + width: col_num as u32, + depth: row_num as u32, + }, + true, + ) + } + CountSketchWithHeap => { + let (row_num, col_num) = cms_params(self); + let heap_size = heap_size_param(self); + ( + SummaryKind::CountSketchWithHeap, + SummaryParams::CountSketchWithHeap { + width: col_num as u32, + depth: row_num as u32, + heap_size: heap_size as u32, + }, + true, + ) + } + DDSketch => ( + SummaryKind::DDSketch, + SummaryParams::DDSketch { + alpha: ddsketch_alpha_param(self), + }, + false, + ), + HLL => return Err(AccumulatorSpecError::UnmappedAggregationType(HLL)), + SingleSubpopulation => match sub_type { + "Sum" | "sum" => (SummaryKind::Sum, SummaryParams::Sum, false), + "Min" | "min" => (SummaryKind::MinMax, SummaryParams::MinMax, false), + "Max" | "max" => (SummaryKind::MinMax, SummaryParams::MinMax, false), + "Increase" | "increase" => (SummaryKind::Increase, SummaryParams::Increase, false), + "DatasketchesKLL" | "datasketches_kll" | "KLL" | "kll" => ( + SummaryKind::Kll, + SummaryParams::Kll { + k: kll_k_param(self) as u32, + }, + false, + ), + other => { + return Err(AccumulatorSpecError::UnknownSingleSubpopulationSubType( + other.to_string(), + )) + } + }, + MultipleSubpopulation => match sub_type { + "Sum" | "sum" => (SummaryKind::Sum, SummaryParams::Sum, true), + "Min" | "min" => (SummaryKind::MinMax, SummaryParams::MinMax, true), + "Max" | "max" => (SummaryKind::MinMax, SummaryParams::MinMax, true), + "Increase" | "increase" => (SummaryKind::Increase, SummaryParams::Increase, true), + "CountMinSketch" | "count_min_sketch" | "CMS" | "cms" => { + let (row_num, col_num) = cms_params(self); + ( + SummaryKind::Cms, + SummaryParams::Cms { + width: col_num as u32, + depth: row_num as u32, + }, + true, + ) + } + "HydraKLL" | "hydra_kll" => ( + SummaryKind::Kll, + SummaryParams::Kll { + k: kll_k_param(self) as u32, + }, + true, + ), + other => { + return Err(AccumulatorSpecError::UnknownMultipleSubpopulationSubType( + other.to_string(), + )) + } + }, + }; + + let grouping = if keyed { + Some(self.grouping_labels.clone()) + } else { + None + }; + + Ok(AccumulatorSpec { + kind, + params, + grouping, + }) + } +} + +// --------------------------------------------------------------------------- +// Raw-parameter extraction helpers. +// +// Relocated verbatim from `data_plane::precompute_engine::accumulator_factory` +// (same names, same behavior, same defaults) — this is now their one +// definition; `accumulator_factory.rs` re-exports them via `use` so its +// existing unit tests (`cms_params_reads_canonical_w_d_keys`, +// `test_kll_k_param_capital_k`, ...) keep passing unchanged. +// --------------------------------------------------------------------------- + +/// Extract the KLL `k` parameter. Capital `"K"` takes precedence over +/// lowercase `"k"` to match the convention used by the top-level +/// aggregation type arms. Defaults to 200. +pub fn kll_k_param(config: &AggregationConfig) -> u16 { + config + .parameters + .get("K") + .or_else(|| config.parameters.get("k")) + .and_then(|v| v.as_u64()) + .and_then(|v| u16::try_from(v).ok()) + .unwrap_or(200) +} + +/// Extract `(row_num, col_num)` for CMS / HydraKLL configs. +/// +/// Reads canonical `d` (depth = rows) / `w` (width = cols) keys — +/// matches what the control plane's `sketch_params_to_json` emits and +/// what `sketch_config_to_params` uses for OTLP policy_fp content +/// matching. Defaults to `(4, 1000)`. +pub fn cms_params(config: &AggregationConfig) -> (usize, usize) { + let row_num = config + .parameters + .get("d") + .and_then(|v| v.as_u64()) + .unwrap_or(4) as usize; + let col_num = config + .parameters + .get("w") + .and_then(|v| v.as_u64()) + .unwrap_or(1000) as usize; + (row_num, col_num) +} + +/// Top-k heap size for the `*WithHeap` configs. Reads `heap_size` / `k` +/// from `parameters`; defaults to 20 (the heap holds the top-k +/// candidates — it must be >= the largest `k` a query asks for). +pub fn heap_size_param(config: &AggregationConfig) -> usize { + config + .parameters + .get("heap_size") + .or_else(|| config.parameters.get("k")) + .or_else(|| config.parameters.get("K")) + .and_then(|v| v.as_u64()) + .map(|v| v as usize) + .filter(|&v| v > 0) + .unwrap_or(20) +} + +/// Pull `relativeAccuracy` (or canonical aliases) out of a +/// streaming-config aggregation entry. Defaults to 0.01 (1% rel-err, +/// the same default the agent's `ddsketchprocessor` uses). +pub fn ddsketch_alpha_param(config: &AggregationConfig) -> f64 { + let parsed = param_f64(config, "relativeAccuracy") + .or_else(|| param_f64(config, "relative_accuracy")) + .or_else(|| param_f64(config, "alpha")) + .unwrap_or(0.01); + if parsed > 0.0 && parsed < 1.0 { + parsed + } else { + tracing::warn!( + "DDSketch relativeAccuracy {} out of (0,1); using default 0.01", + parsed + ); + 0.01 + } +} + +fn param_f64(config: &AggregationConfig, key: &str) -> Option { + config.parameters.get(key).and_then(Value::as_f64) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::enums::WindowType; + use crate::key_by_label_names::KeyByLabelNames; + use std::collections::HashMap; + + #[allow(clippy::too_many_arguments)] + fn make_config( + agg_type: AggregationType, + sub_type: &str, + params: HashMap, + grouping_labels: Vec<&str>, + ) -> AggregationConfig { + AggregationConfig::new( + agg_type, + sub_type.to_string(), + params, + KeyByLabelNames::new(grouping_labels.into_iter().map(|s| s.to_string()).collect()), + KeyByLabelNames::empty(), + KeyByLabelNames::empty(), + String::new(), + 60, + 60, + WindowType::Tumbling, + String::new(), + "m".to_string(), + None, + None, + None, + ) + } + + // ---- direct (non-wrapper) variants ------------------------------ + + #[test] + fn sum_is_unkeyed_sum() { + let cfg = make_config(AggregationType::Sum, "", HashMap::new(), vec![]); + let spec = cfg.accumulator_spec().expect("resolves"); + assert_eq!(spec.kind, SummaryKind::Sum); + assert_eq!(spec.params, SummaryParams::Sum); + assert!(spec.grouping.is_none()); + } + + #[test] + fn multiple_sum_is_keyed_sum() { + let cfg = make_config( + AggregationType::MultipleSum, + "", + HashMap::new(), + vec!["zone"], + ); + let spec = cfg.accumulator_spec().expect("resolves"); + assert_eq!(spec.kind, SummaryKind::Sum); + assert_eq!( + spec.grouping, + Some(KeyByLabelNames::new(vec!["zone".to_string()])) + ); + } + + #[test] + fn datasketches_kll_reads_k_param() { + let mut params = HashMap::new(); + params.insert("k".to_string(), serde_json::json!(128)); + let cfg = make_config(AggregationType::DatasketchesKLL, "", params, vec![]); + let spec = cfg.accumulator_spec().expect("resolves"); + assert_eq!(spec.kind, SummaryKind::Kll); + assert_eq!(spec.params, SummaryParams::Kll { k: 128 }); + assert!(spec.grouping.is_none()); + } + + #[test] + fn hydra_kll_is_keyed_kll() { + let mut params = HashMap::new(); + params.insert("k".to_string(), serde_json::json!(64)); + let cfg = make_config(AggregationType::HydraKLL, "", params, vec!["host"]); + let spec = cfg.accumulator_spec().expect("resolves"); + assert_eq!(spec.kind, SummaryKind::Kll); + assert_eq!(spec.params, SummaryParams::Kll { k: 64 }); + assert!(spec.grouping.is_some()); + } + + #[test] + fn count_min_sketch_maps_to_cms_params() { + let mut params = HashMap::new(); + params.insert("d".to_string(), serde_json::json!(7)); + params.insert("w".to_string(), serde_json::json!(2048)); + let cfg = make_config(AggregationType::CountMinSketch, "", params, vec!["host"]); + let spec = cfg.accumulator_spec().expect("resolves"); + assert_eq!(spec.kind, SummaryKind::Cms); + assert_eq!( + spec.params, + SummaryParams::Cms { + width: 2048, + depth: 7 + } + ); + assert!(spec.grouping.is_some()); + } + + #[test] + fn count_min_sketch_with_heap_maps_to_cms_with_heap_params() { + let mut params = HashMap::new(); + params.insert("d".to_string(), serde_json::json!(4)); + params.insert("w".to_string(), serde_json::json!(256)); + params.insert("heap_size".to_string(), serde_json::json!(8)); + let cfg = make_config( + AggregationType::CountMinSketchWithHeap, + "", + params, + vec!["host"], + ); + let spec = cfg.accumulator_spec().expect("resolves"); + assert_eq!(spec.kind, SummaryKind::CmsWithHeap); + assert_eq!( + spec.params, + SummaryParams::CmsWithHeap { + width: 256, + depth: 4, + heap_size: 8 + } + ); + } + + /// Documented existing quirk (see `accumulator_factory.rs`): bare + /// `CountSketch` gets its own `SummaryKind` identity here, but + /// `accumulator_factory` routes it through the same + /// `CmsAccumulatorUpdater` as bare CMS — no dedicated heap-less + /// Count-Sketch accumulator exists. This test locks in the *identity* + /// resolution only; the shared-accumulator behavior is exercised in + /// `accumulator_factory.rs`'s own tests. + #[test] + fn count_sketch_gets_its_own_kind_identity() { + let cfg = make_config(AggregationType::CountSketch, "", HashMap::new(), vec!["h"]); + let spec = cfg.accumulator_spec().expect("resolves"); + assert_eq!(spec.kind, SummaryKind::CountSketch); + } + + #[test] + fn ddsketch_reads_relative_accuracy_alpha() { + let mut params = HashMap::new(); + params.insert("relativeAccuracy".to_string(), serde_json::json!(0.02)); + let cfg = make_config(AggregationType::DDSketch, "", params, vec![]); + let spec = cfg.accumulator_spec().expect("resolves"); + assert_eq!(spec.kind, SummaryKind::DDSketch); + assert_eq!(spec.params, SummaryParams::DDSketch { alpha: 0.02 }); + assert!(spec.grouping.is_none()); + } + + #[test] + fn hll_is_unmapped_preserving_pre_step5_gap() { + let cfg = make_config(AggregationType::HLL, "", HashMap::new(), vec![]); + let err = cfg.accumulator_spec().expect_err("HLL has no dispatch arm"); + assert_eq!( + err, + AccumulatorSpecError::UnmappedAggregationType(AggregationType::HLL) + ); + } + + // ---- wrapper (Single/MultipleSubpopulation) variants ------------ + + #[test] + fn single_subpopulation_sum_alias() { + for alias in ["Sum", "sum"] { + let cfg = make_config( + AggregationType::SingleSubpopulation, + alias, + HashMap::new(), + vec![], + ); + let spec = cfg.accumulator_spec().expect("resolves"); + assert_eq!(spec.kind, SummaryKind::Sum); + assert!(spec.grouping.is_none()); + } + } + + #[test] + fn multiple_subpopulation_cms_alias() { + for alias in ["CountMinSketch", "count_min_sketch", "CMS", "cms"] { + let cfg = make_config( + AggregationType::MultipleSubpopulation, + alias, + HashMap::new(), + vec!["host"], + ); + let spec = cfg.accumulator_spec().expect("resolves"); + assert_eq!(spec.kind, SummaryKind::Cms); + assert!(spec.grouping.is_some()); + } + } + + #[test] + fn multiple_subpopulation_hydra_kll_alias() { + for alias in ["HydraKLL", "hydra_kll"] { + let cfg = make_config( + AggregationType::MultipleSubpopulation, + alias, + HashMap::new(), + vec!["host"], + ); + let spec = cfg.accumulator_spec().expect("resolves"); + assert_eq!(spec.kind, SummaryKind::Kll); + assert!(spec.grouping.is_some()); + } + } + + #[test] + fn unknown_single_subpopulation_sub_type_errors() { + let cfg = make_config( + AggregationType::SingleSubpopulation, + "Bogus", + HashMap::new(), + vec![], + ); + let err = cfg.accumulator_spec().expect_err("unknown sub_type"); + assert_eq!( + err, + AccumulatorSpecError::UnknownSingleSubpopulationSubType("Bogus".to_string()) + ); + } + + #[test] + fn unknown_multiple_subpopulation_sub_type_errors() { + let cfg = make_config( + AggregationType::MultipleSubpopulation, + "Bogus", + HashMap::new(), + vec!["host"], + ); + let err = cfg.accumulator_spec().expect_err("unknown sub_type"); + assert_eq!( + err, + AccumulatorSpecError::UnknownMultipleSubpopulationSubType("Bogus".to_string()) + ); + } + + #[test] + fn error_display_matches_pre_step5_warning_text() { + assert_eq!( + AccumulatorSpecError::UnknownSingleSubpopulationSubType("Bogus".to_string()) + .to_string(), + "Unknown SingleSubpopulation sub_type 'Bogus', defaulting to Sum" + ); + assert_eq!( + AccumulatorSpecError::UnknownMultipleSubpopulationSubType("Bogus".to_string()) + .to_string(), + "Unknown MultipleSubpopulation sub_type 'Bogus', defaulting to Sum" + ); + assert_eq!( + AccumulatorSpecError::UnmappedAggregationType(AggregationType::HLL).to_string(), + "Unknown aggregation_type 'HLL', defaulting to SingleSubpopulation Sum" + ); + } + + // ---- PolicyFingerprint stability guard --------------------------- + + /// `accumulator_spec()` must be a pure, additional *read* of + /// `AggregationConfig` — it must not change what + /// `PolicyFingerprint::from_config` hashes. This locks in a fixed + /// fingerprint for a fixed config as a tripwire: if this test ever + /// needs its expected constant updated, `policy_fingerprint.rs` + /// changed in a way that breaks on-disk sid compatibility, which is + /// exactly what Step 5 was required not to do. + #[test] + fn accumulator_spec_does_not_perturb_policy_fingerprint() { + let cfg = make_config(AggregationType::DDSketch, "", HashMap::new(), vec![]); + let fp_before = cfg.policy_fp_u64(); + let _ = cfg.accumulator_spec(); // takes &self — must not mutate `cfg` + let fp_after = cfg.policy_fp_u64(); + assert_eq!( + fp_before, fp_after, + "calling accumulator_spec() must not change the fingerprint \ + PolicyFingerprint::from_config computes from this config" + ); + + // An independently-built config with identical content must + // still agree — proves accumulator_spec() reads, never writes, + // the fields PolicyFingerprint::from_config hashes. + let cfg2 = make_config(AggregationType::DDSketch, "", HashMap::new(), vec![]); + assert_eq!(fp_after, cfg2.policy_fp_u64()); + } +} diff --git a/crates/asap_types/src/lib.rs b/crates/asap_types/src/lib.rs index 701b5c0a..1a867c19 100644 --- a/crates/asap_types/src/lib.rs +++ b/crates/asap_types/src/lib.rs @@ -1,3 +1,4 @@ +pub mod accumulator_spec; pub mod aggregation_config; pub mod aggregation_type; pub mod enums; @@ -9,6 +10,7 @@ pub mod query_requirements; pub mod traits; pub mod utils; +pub use accumulator_spec::{AccumulatorSpec, AccumulatorSpecError, SummaryKind, SummaryParams}; pub use aggregation_config::*; pub use aggregation_type::AggregationType; pub use enums::*; diff --git a/data_plane/src/precompute_engine/accumulator_factory.rs b/data_plane/src/precompute_engine/accumulator_factory.rs index aacd8e76..43115aa4 100644 --- a/data_plane/src/precompute_engine/accumulator_factory.rs +++ b/data_plane/src/precompute_engine/accumulator_factory.rs @@ -7,6 +7,21 @@ use crate::storage_engines::types::{ AggregateCore, AggregationType, KeyByLabelValues, Measurement, }; use asap_types::aggregation_config::AggregationConfig; +// Step 5 (sketch-identity unification, see +// scratchpad/artifacts/enum-unification-plan.md): dispatch below is +// driven by `AccumulatorSpec` (SummaryKind + typed SummaryParams + +// keyed-axis grouping) instead of raw `AggregationType` + +// `aggregation_sub_type` string matching. Numeric params come straight +// off `spec.params` (typed, no HashMap lookups) except `cms_params`, +// kept as a raw-`parameters` read for the one case `SummaryParams` has +// no field for: HydraKLL's `(row, col)` tiling grid (see +// `asap_types::accumulator_spec`'s module doc for why). `cms_params` +// now lives there — the only place that still needs the other three +// former local helpers (`kll_k_param`, `heap_size_param`, +// `ddsketch_alpha_param`) is that module's own `AccumulatorSpec` +// construction, so they aren't re-imported here. +use asap_types::accumulator_spec::{cms_params, AccumulatorSpecError}; +use asap_types::{SummaryKind, SummaryParams}; /// Generate the two boilerplate clone-based `AccumulatorUpdater` methods /// for updaters whose inner `acc` field implements `Clone + AggregateCore`. @@ -343,28 +358,6 @@ impl AccumulatorUpdater for DDSketchAccumulatorUpdater { } } -/// Pull `relativeAccuracy` (or canonical aliases) out of a -/// streaming-config aggregation entry. Defaults to 0.01 (1% rel- -/// err, the same default the agent's `ddsketchprocessor` uses). -fn ddsketch_alpha_param(config: &AggregationConfig) -> f64 { - let parsed = config - .parameters - .get("relativeAccuracy") - .or_else(|| config.parameters.get("relative_accuracy")) - .or_else(|| config.parameters.get("alpha")) - .and_then(|v| v.as_f64()) - .unwrap_or(0.01); - if parsed > 0.0 && parsed < 1.0 { - parsed - } else { - tracing::warn!( - "DDSketch relativeAccuracy {} out of (0,1); using default 0.01", - parsed - ); - 0.01 - } -} - // --------------------------------------------------------------------------- // MultipleSumAccumulatorUpdater // --------------------------------------------------------------------------- @@ -756,55 +749,6 @@ pub fn config_is_keyed(config: &AggregationConfig) -> bool { ) } -/// Extract the KLL `k` parameter. Capital `"K"` takes precedence over lowercase -/// `"k"` to match the convention used by the top-level aggregation type arms. -fn kll_k_param(config: &AggregationConfig) -> u16 { - config - .parameters - .get("K") - .or_else(|| config.parameters.get("k")) - .and_then(|v| v.as_u64()) - .and_then(|v| u16::try_from(v).ok()) - .unwrap_or(200) -} - -/// Extract `(row_num, col_num)` for CMS / HydraKLL configs. -/// -/// Reads canonical `d` (depth = rows) / `w` (width = cols) keys — -/// matches what the control plane's `sketch_params_to_json` emits -/// and what `sketch_config_to_params` uses for OTLP policy_fp -/// content matching. The legacy `row_num` / `col_num` form (the -/// only pre-PR-268 reader) was retired in lock-step with the -/// asapcollector migration to canonical keys. -fn cms_params(config: &AggregationConfig) -> (usize, usize) { - let row_num = config - .parameters - .get("d") - .and_then(|v| v.as_u64()) - .unwrap_or(4) as usize; - let col_num = config - .parameters - .get("w") - .and_then(|v| v.as_u64()) - .unwrap_or(1000) as usize; - (row_num, col_num) -} - -/// Top-k heap size for the `*WithHeap` configs. Reads `heap_size` / -/// `k` from `parameters`; defaults to 20 (the heap holds the top-k -/// candidates — it must be ≥ the largest `k` a query asks for). -fn heap_size_param(config: &AggregationConfig) -> usize { - config - .parameters - .get("heap_size") - .or_else(|| config.parameters.get("k")) - .or_else(|| config.parameters.get("K")) - .and_then(|v| v.as_u64()) - .map(|v| v as usize) - .filter(|&v| v > 0) - .unwrap_or(20) -} - /// Top-k ranking quantity for the `*WithHeap` configs. /// /// Selected by `parameters["weight_mode"]` (or alias `topk_weight`): @@ -831,75 +775,157 @@ fn topk_weight_param(config: &AggregationConfig) -> TopkWeight { } } -/// Extract `(row_num, col_num, k)` for HydraKLL configs. -fn hydra_kll_params(config: &AggregationConfig) -> (usize, usize, u16) { - let (row_num, col_num) = cms_params(config); - (row_num, col_num, kll_k_param(config)) -} - // --------------------------------------------------------------------------- // Factory function // --------------------------------------------------------------------------- +/// Read the KLL `k` out of `SummaryParams::Kll`. `accumulator_spec()` +/// always pairs `SummaryKind::Kll` with `SummaryParams::Kll`, so the +/// other arm is unreachable from a `spec` this module builds itself. +fn kll_k(params: &SummaryParams) -> u16 { + match params { + // Lossless: `accumulator_spec()` only ever stores a value that + // already fit in `u16` (via `kll_k_param`'s own `u16::try_from` + // fallback) widened to `u32`. + SummaryParams::Kll { k } => *k as u16, + other => unreachable!( + "accumulator_spec() paired SummaryKind::Kll with non-Kll params: {other:?}" + ), + } +} + +/// Read `(width, depth)` out of `SummaryParams::Cms` or `::CountSketch` +/// — same shape, different variant per bare-sketch identity. +fn cms_dims(params: &SummaryParams) -> (usize, usize) { + match params { + SummaryParams::Cms { width, depth } | SummaryParams::CountSketch { width, depth } => { + (*width as usize, *depth as usize) + } + other => unreachable!( + "accumulator_spec() paired SummaryKind::Cms/CountSketch with unexpected params: {other:?}" + ), + } +} + +/// Read `(width, depth, heap_size)` out of `SummaryParams::CmsWithHeap` +/// or `::CountSketchWithHeap`. +fn cms_heap_dims(params: &SummaryParams) -> (usize, usize, usize) { + match params { + SummaryParams::CmsWithHeap { + width, + depth, + heap_size, + } + | SummaryParams::CountSketchWithHeap { + width, + depth, + heap_size, + } => (*width as usize, *depth as usize, *heap_size as usize), + other => unreachable!( + "accumulator_spec() paired a WithHeap SummaryKind with unexpected params: {other:?}" + ), + } +} + +/// Read the DDSketch relative-accuracy `alpha` out of `SummaryParams::DDSketch`. +fn ddsketch_alpha(params: &SummaryParams) -> f64 { + match params { + SummaryParams::DDSketch { alpha } => *alpha, + other => unreachable!( + "accumulator_spec() paired SummaryKind::DDSketch with non-DDSketch params: {other:?}" + ), + } +} + /// Create an appropriate `AccumulatorUpdater` from an `AggregationConfig`. +/// +/// Dispatches on [`asap_types::AccumulatorSpec`] — `SummaryKind` identity +/// plus the keyed/unkeyed `grouping` axis — instead of the pre-Step-5 +/// `AggregationType` + `aggregation_sub_type` string combo. See +/// `asap_types::accumulator_spec`'s module doc for why min/max direction, +/// HydraKLL's `(row, col)` tiling, and top-k `weight_mode` still read +/// `config` directly rather than going through `SummaryParams` — none of +/// those three have a field in ASAPController's upstream type to live in. pub fn create_accumulator_updater(config: &AggregationConfig) -> Box { - let sub_type = config.aggregation_sub_type.as_str(); - - match config.aggregation_type { - AggregationType::SingleSubpopulation => match sub_type { - "Sum" | "sum" => Box::new(SumAccumulatorUpdater::new()), - "Min" | "min" => Box::new(MinMaxAccumulatorUpdater::new(false)), - "Max" | "max" => Box::new(MinMaxAccumulatorUpdater::new(true)), - "Increase" | "increase" => Box::new(IncreaseAccumulatorUpdater::new()), - "DatasketchesKLL" | "datasketches_kll" | "KLL" | "kll" => { - Box::new(KllAccumulatorUpdater::new(kll_k_param(config))) - } - other => { - tracing::warn!( - "Unknown SingleSubpopulation sub_type '{}', defaulting to Sum", - other - ); - Box::new(SumAccumulatorUpdater::new()) - } - }, - AggregationType::MultipleSubpopulation => match sub_type { - "Sum" | "sum" => Box::new(MultipleSumAccumulatorUpdater::new()), - "Min" | "min" => Box::new(MultipleMinMaxAccumulatorUpdater::new(false)), - "Max" | "max" => Box::new(MultipleMinMaxAccumulatorUpdater::new(true)), - "Increase" | "increase" => Box::new(MultipleIncreaseAccumulatorUpdater::new()), - "CountMinSketch" | "count_min_sketch" | "CMS" | "cms" => { - let (row_num, col_num) = cms_params(config); - Box::new(CmsAccumulatorUpdater::new(row_num, col_num)) - } - "HydraKLL" | "hydra_kll" => { - let (row_num, col_num, k) = hydra_kll_params(config); - Box::new(HydraKllAccumulatorUpdater::new(row_num, col_num, k)) - } - other => { - tracing::warn!( - "Unknown MultipleSubpopulation sub_type '{}', defaulting to Sum", - other - ); - Box::new(MultipleSumAccumulatorUpdater::new()) - } - }, - AggregationType::DatasketchesKLL => { - Box::new(KllAccumulatorUpdater::new(kll_k_param(config))) + let spec = match config.accumulator_spec() { + Ok(spec) => spec, + // Three fallback paths, preserved verbatim from the pre-Step-5 + // dispatch: same warning text, same default updater per case + // (Single- and MultipleSubpopulation default to *different* + // updaters — see `AccumulatorSpecError`'s doc). + Err(AccumulatorSpecError::UnknownSingleSubpopulationSubType(sub_type)) => { + tracing::warn!( + "Unknown SingleSubpopulation sub_type '{}', defaulting to Sum", + sub_type + ); + return Box::new(SumAccumulatorUpdater::new()); } - AggregationType::MultipleSum => Box::new(MultipleSumAccumulatorUpdater::new()), - AggregationType::MultipleIncrease => Box::new(MultipleIncreaseAccumulatorUpdater::new()), - AggregationType::MultipleMinMax => Box::new(MultipleMinMaxAccumulatorUpdater::new( - sub_type.eq_ignore_ascii_case("max"), + Err(AccumulatorSpecError::UnknownMultipleSubpopulationSubType(sub_type)) => { + tracing::warn!( + "Unknown MultipleSubpopulation sub_type '{}', defaulting to Sum", + sub_type + ); + return Box::new(MultipleSumAccumulatorUpdater::new()); + } + Err(AccumulatorSpecError::UnmappedAggregationType(other)) => { + tracing::warn!( + "Unknown aggregation_type '{:?}', defaulting to SingleSubpopulation Sum", + other + ); + return Box::new(SumAccumulatorUpdater::new()); + } + }; + + let keyed = spec.grouping.is_some(); + + match (&spec.kind, keyed) { + (SummaryKind::Sum, false) => Box::new(SumAccumulatorUpdater::new()), + (SummaryKind::Sum, true) => Box::new(MultipleSumAccumulatorUpdater::new()), + + // Min/max direction isn't part of `SummaryParams::MinMax` + // (upstream models no direction axis) — read straight off + // `aggregation_sub_type`, exactly as the pre-Step-5 dispatch did + // for the direct `AggregationType::MinMax`/`MultipleMinMax` + // arms. `accumulator_spec()` only resolves a wrapper's sub_type + // to `SummaryKind::MinMax` for an exact "Min"/"min"/"Max"/"max" + // match, so re-deriving via `eq_ignore_ascii_case("max")` here + // reproduces the same true/false split for that path too. + (SummaryKind::MinMax, false) => Box::new(MinMaxAccumulatorUpdater::new( + config.aggregation_sub_type.eq_ignore_ascii_case("max"), )), - AggregationType::Sum => Box::new(SumAccumulatorUpdater::new()), - AggregationType::MinMax => Box::new(MinMaxAccumulatorUpdater::new( - sub_type.eq_ignore_ascii_case("max"), + (SummaryKind::MinMax, true) => Box::new(MultipleMinMaxAccumulatorUpdater::new( + config.aggregation_sub_type.eq_ignore_ascii_case("max"), )), - AggregationType::Increase => Box::new(IncreaseAccumulatorUpdater::new()), - AggregationType::CountMinSketch => { + + (SummaryKind::Increase, false) => Box::new(IncreaseAccumulatorUpdater::new()), + (SummaryKind::Increase, true) => Box::new(MultipleIncreaseAccumulatorUpdater::new()), + + (SummaryKind::Kll, false) => Box::new(KllAccumulatorUpdater::new(kll_k(&spec.params))), + // HydraKLL: `k` comes off the typed params like the unkeyed case, + // but the `(row, col)` tiling grid has no `SummaryParams::Kll` + // field to live in (see `asap_types::accumulator_spec`'s module + // doc) — read it the same way bare CMS does, via `cms_params`. + (SummaryKind::Kll, true) => { let (row_num, col_num) = cms_params(config); + Box::new(HydraKllAccumulatorUpdater::new( + row_num, + col_num, + kll_k(&spec.params), + )) + } + + // Bare CMS / bare CountSketch: pre-Step-5 quirk preserved + // exactly — CountSketch has no dedicated heap-less accumulator, + // so it shares `CmsAccumulatorUpdater` with plain CMS + // (point-frequency only; top-k needs the heap-bearing variant + // below). `keyed=false` can't actually arise here today (no + // `AggregationType` resolves to bare Cms/CountSketch unkeyed — + // see accumulator_spec.rs), matched anyway as a safe default. + (SummaryKind::Cms, _) | (SummaryKind::CountSketch, _) => { + let (row_num, col_num) = cms_dims(&spec.params); Box::new(CmsAccumulatorUpdater::new(row_num, col_num)) } + // Heap-bearing top-k variants (raw-input ingest path): route to // the real `CmsHeapAccumulatorUpdater` so the per-policy top-k // heap is BUILT (heap-less CMS could not answer `topk(...)` — @@ -910,35 +936,34 @@ pub fn create_accumulator_updater(config: &AggregationConfig) -> Box { - let (row_num, col_num) = cms_params(config); + (SummaryKind::CmsWithHeap, _) | (SummaryKind::CountSketchWithHeap, _) => { + let (row_num, col_num, heap_size) = cms_heap_dims(&spec.params); Box::new(CmsHeapAccumulatorUpdater::new( row_num, col_num, - heap_size_param(config), + heap_size, topk_weight_param(config), )) } - // Heap-LESS CountSketch (raw-input ingest path): route to - // `CmsAccumulatorUpdater` — it handles the same `(rows, cols)` - // matrix shape and answers point-frequency only. CountSketch - // proper has no top-k heap, so `topk(...)` against it routes - // through the heap-bearing variant above. - AggregationType::CountSketch => { - let (row_num, col_num) = cms_params(config); - Box::new(CmsAccumulatorUpdater::new(row_num, col_num)) - } - AggregationType::HydraKLL => { - let (row_num, col_num, k) = hydra_kll_params(config); - Box::new(HydraKllAccumulatorUpdater::new(row_num, col_num, k)) - } - AggregationType::DDSketch => Box::new(DDSketchAccumulatorUpdater::new( - ddsketch_alpha_param(config), - )), - other => { + + (SummaryKind::DDSketch, _) => Box::new(DDSketchAccumulatorUpdater::new(ddsketch_alpha( + &spec.params, + ))), + + // `SummaryKind::Hll` / `Count` / `Rate` / `Kmv` / `Theta`: no + // `AggregationType` resolves to one of these via + // `accumulator_spec()`'s `Ok` path today — HLL is caught by + // `AccumulatorSpecError::UnmappedAggregationType` above (see its + // doc for why: a pre-existing gap, not introduced here), and the + // other four have no `AggregationType` counterpart at all. Kept + // as an explicit warning fallback rather than `unreachable!()` + // so a future `SummaryKind` this dispatch doesn't yet know how + // to build fails safe instead of panicking. + (other_kind, keyed) => { tracing::warn!( - "Unknown aggregation_type '{:?}', defaulting to SingleSubpopulation Sum", - other + "SummaryKind {:?} (keyed={}) has no accumulator_factory mapping, defaulting to Sum", + other_kind, + keyed ); Box::new(SumAccumulatorUpdater::new()) } From 4a41f311b5127867381e6f66911ab7f43e609787 Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 21 Jul 2026 08:46:54 -0600 Subject: [PATCH 2/3] fix(asap_types): repoint accumulator_spec.rs import after promql_utilities retirement promql_utilities was deleted in #403 (merged to main after this branch was cut); AggregationType now lives in-crate at asap_types::aggregation_type. Update the one remaining import site picked up by the rebase. Co-Authored-By: Claude Sonnet 5 --- crates/asap_types/src/accumulator_spec.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/asap_types/src/accumulator_spec.rs b/crates/asap_types/src/accumulator_spec.rs index a2064c37..a97c8c81 100644 --- a/crates/asap_types/src/accumulator_spec.rs +++ b/crates/asap_types/src/accumulator_spec.rs @@ -72,7 +72,7 @@ use serde_json::Value; use crate::aggregation_config::AggregationConfig; use crate::key_by_label_names::KeyByLabelNames; -use promql_utilities::query_logics::enums::AggregationType; +use crate::AggregationType; pub use asap_sketch::{SummaryKind, SummaryParams}; From c58239a62f47e8194c492070b72b18d3993cbcff Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 21 Jul 2026 10:02:22 -0600 Subject: [PATCH 3/3] fix(asap_types): rename WindowType to WindowKind in accumulator_spec.rs tests Picked up by rebasing onto main post-#405 (WindowType retired in favor of asap_ir::WindowKind). Test-only fixture reference, no behavior change. Co-Authored-By: Claude Sonnet 5 --- Cargo.lock | 1 + crates/asap_types/src/accumulator_spec.rs | 4 ++-- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index fb7de5e5..bb127f0b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -419,6 +419,7 @@ version = "0.1.0" dependencies = [ "anyhow", "asap-ir", + "asap-sketch", "clap 4.6.1", "serde", "serde_json", diff --git a/crates/asap_types/src/accumulator_spec.rs b/crates/asap_types/src/accumulator_spec.rs index a97c8c81..4ea86881 100644 --- a/crates/asap_types/src/accumulator_spec.rs +++ b/crates/asap_types/src/accumulator_spec.rs @@ -396,7 +396,7 @@ fn param_f64(config: &AggregationConfig, key: &str) -> Option { #[cfg(test)] mod tests { use super::*; - use crate::enums::WindowType; + use crate::enums::WindowKind; use crate::key_by_label_names::KeyByLabelNames; use std::collections::HashMap; @@ -417,7 +417,7 @@ mod tests { String::new(), 60, 60, - WindowType::Tumbling, + WindowKind::Tumbling, String::new(), "m".to_string(), None,