From c2fe69a640d68f88b8ec861d174d44fee5fe2cd2 Mon Sep 17 00:00:00 2001 From: zz_y Date: Sat, 16 May 2026 08:14:56 -0600 Subject: [PATCH] fix(backend): cms_params accepts canonical w/d + legacy row_num/col_num MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two `cms_params` helpers (accumulator_factory + accuracy) read the CMS / CountSketch dimensions from `AggregationConfig.parameters`. Pre-PR they only looked at the legacy `row_num` / `col_num` keys — the keys the control plane's `sketch_params_to_json` emits (and that PR #258's `sketch_config_to_params` standardized on for OTLP policy_fp content matching) are `w` / `d`. Result was a silent mismatch: the OTLP modified-sketch ingest path content-matched on `w` / `d`, but raw-input ingest (precompute worker + sketch_db backfill) read from `row_num` / `col_num` — meaning static streaming-config YAMLs that ship in asapcollector with `row_num` / `col_num` worked for raw-input ingest but failed policy_fp lookup on the modified-OTLP side, while controller- emitted configs with `w` / `d` worked for OTLP but defaulted to `(4, 1000)` on the raw-input side. Both helpers now read canonical `w` / `d` first and fall back to `row_num` / `col_num`. Both naming conventions work everywhere; old asapcollector configs keep working, controller-emitted configs keep working, and a follow-up asapcollector PR will migrate the static YAMLs to the canonical form. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../precompute_engine/accumulator_factory.rs | 86 ++++++++++++++++++- .../src/storage_engines/sketch_db/accuracy.rs | 12 ++- 2 files changed, 94 insertions(+), 4 deletions(-) diff --git a/data_plane/src/precompute_engine/accumulator_factory.rs b/data_plane/src/precompute_engine/accumulator_factory.rs index edeb2bb1..0b38f727 100644 --- a/data_plane/src/precompute_engine/accumulator_factory.rs +++ b/data_plane/src/precompute_engine/accumulator_factory.rs @@ -649,15 +649,24 @@ fn kll_k_param(config: &AggregationConfig) -> u16 { } /// Extract `(row_num, col_num)` for CMS / HydraKLL configs. +/// +/// Reads from the canonical `d` (depth = rows) / `w` (width = cols) +/// keys first — these match what the control plane's +/// `sketch_params_to_json` emits and what `sketch_config_to_params` +/// uses for OTLP policy_fp content matching. Falls back to the +/// legacy `row_num` / `col_num` keys so older asapcollector +/// configs continue to work. fn cms_params(config: &AggregationConfig) -> (usize, usize) { let row_num = config .parameters - .get("row_num") + .get("d") + .or_else(|| config.parameters.get("row_num")) .and_then(|v| v.as_u64()) .unwrap_or(4) as usize; let col_num = config .parameters - .get("col_num") + .get("w") + .or_else(|| config.parameters.get("col_num")) .and_then(|v| v.as_u64()) .unwrap_or(1000) as usize; (row_num, col_num) @@ -968,4 +977,77 @@ mod tests { .expect("should be KLL"); assert_eq!(kll.inner.k, 50, "k should be 50 from capital-K param"); } + + #[test] + fn cms_params_accepts_w_d_canonical_and_row_col_num_legacy() { + use std::collections::HashMap; + // Canonical `w`/`d` form — what the control plane's + // `sketch_params_to_json` emits today. + let mut params = HashMap::new(); + params.insert("d".to_string(), serde_json::Value::from(7_u64)); + params.insert("w".to_string(), serde_json::Value::from(2048_u64)); + let config = AggregationConfig::new( + AggregationType::CountMinSketch, + String::new(), + params, + 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, + 0, + WindowType::Tumbling, + "m".to_string(), + "m".to_string(), + None, + None, + None, + ); + assert_eq!(super::cms_params(&config), (7, 2048)); + + // Legacy `row_num`/`col_num` form — older asapcollector + // configs still ship these; the helper must keep accepting + // them so existing Docker e2e setups don't silently degrade. + let mut legacy = HashMap::new(); + legacy.insert("row_num".to_string(), serde_json::Value::from(7_u64)); + legacy.insert("col_num".to_string(), serde_json::Value::from(2048_u64)); + let legacy_config = AggregationConfig::new( + AggregationType::CountMinSketch, + String::new(), + legacy, + 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, + 0, + WindowType::Tumbling, + "m".to_string(), + "m".to_string(), + None, + None, + None, + ); + assert_eq!(super::cms_params(&legacy_config), (7, 2048)); + + // Empty params — defaults `(4, 1000)`. + let empty_config = AggregationConfig::new( + AggregationType::CountMinSketch, + String::new(), + 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, + 0, + WindowType::Tumbling, + "m".to_string(), + "m".to_string(), + None, + None, + None, + ); + assert_eq!(super::cms_params(&empty_config), (4, 1000)); + } } diff --git a/data_plane/src/storage_engines/sketch_db/accuracy.rs b/data_plane/src/storage_engines/sketch_db/accuracy.rs index a577b8c7..6b79c115 100644 --- a/data_plane/src/storage_engines/sketch_db/accuracy.rs +++ b/data_plane/src/storage_engines/sketch_db/accuracy.rs @@ -301,15 +301,23 @@ impl AccuracyProfile { // says it's OK for them to drift — this module is the single // authority on *accuracy*, not on *construction*. +/// Reads from the canonical `d` (depth = rows) / `w` (width = cols) +/// keys first — these match what the control plane's +/// `sketch_params_to_json` emits and what +/// `accumulator_factory::cms_params` reads. Falls back to the legacy +/// `row_num` / `col_num` keys so older asapcollector configs +/// continue to work. fn cms_params(config: &AggregationConfig) -> (u64, u64) { let rows = config .parameters - .get("row_num") + .get("d") + .or_else(|| config.parameters.get("row_num")) .and_then(|v| v.as_u64()) .unwrap_or(4); let cols = config .parameters - .get("col_num") + .get("w") + .or_else(|| config.parameters.get("col_num")) .and_then(|v| v.as_u64()) .unwrap_or(1000); (rows, cols)