Skip to content

refactor(backend): drop row_num/col_num legacy fallback from cms_params #269

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
zzylol merged 1 commit into from
May 16, 2026
Merged

refactor(backend): drop row_num/col_num legacy fallback from cms_params #269

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 11 additions & 35 deletions data_plane/src/precompute_engine/accumulator_factory.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -650,23 +650,21 @@ 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.
/// 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")
.or_else(|| config.parameters.get("row_num"))
.and_then(|v| v.as_u64())
.unwrap_or(4) as usize;
let col_num = config
.parameters
.get("w")
.or_else(|| config.parameters.get("col_num"))
.and_then(|v| v.as_u64())
.unwrap_or(1000) as usize;
(row_num, col_num)
Expand Down Expand Up @@ -979,10 +977,13 @@ mod tests {
}

#[test]
fn cms_params_accepts_w_d_canonical_and_row_col_num_legacy() {
fn cms_params_reads_canonical_w_d_keys() {
use std::collections::HashMap;
// Canonical `w`/`d` form — what the control plane's
// `sketch_params_to_json` emits today.
// `sketch_params_to_json` emits and what asapcollector
// streaming-config YAMLs ship (asapcollector PR
// `sync-config-canonical-w-d` migrated them in lock-step
// with the legacy-fallback removal).
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));
Expand All @@ -1005,31 +1006,6 @@ mod tests {
);
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,
Expand Down
49 changes: 22 additions & 27 deletions data_plane/src/storage_engines/sketch_db/accuracy.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -138,11 +138,10 @@ impl AccuracyProfile {

// CountMinSketch: classic Cormode-Muthukrishnan bound.
// ε = e/w, δ = 1/2^d with w = width, d = depth. We
// pull w from `parameters["col_num"]` and d from
// `parameters["row_num"]` because that's how the
// existing `accumulator_factory::cms_params` names
// them; fall back to the factory's (rows=4, cols=1000)
// defaults if absent.
// pull w from `parameters["w"]` and d from
// `parameters["d"]` — the canonical keys the controller
// emits and `accumulator_factory::cms_params` reads.
// Defaults to (rows=4, cols=1000) when absent.
AggregationType::CountMinSketch => {
let (rows, cols) = cms_params(config);
// Using natural e ≈ 2.71828 for tighter bound.
Expand Down Expand Up @@ -301,23 +300,19 @@ 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.
/// Reads canonical `d` (depth = rows) / `w` (width = cols) keys.
/// The legacy `row_num` / `col_num` form was retired in lock-step
/// with the asapcollector migration to canonical keys — see
/// `accumulator_factory::cms_params` for the matching change.
fn cms_params(config: &AggregationConfig) -> (u64, u64) {
let rows = config
.parameters
.get("d")
.or_else(|| config.parameters.get("row_num"))
.and_then(|v| v.as_u64())
.unwrap_or(4);
let cols = config
.parameters
.get("w")
.or_else(|| config.parameters.get("col_num"))
.and_then(|v| v.as_u64())
.unwrap_or(1000);
(rows, cols)
Expand Down Expand Up @@ -499,8 +494,8 @@ mod tests {
#[test]
fn cms_epsilon_is_e_over_w() {
let mut params = HashMap::new();
params.insert("row_num".to_string(), json!(5));
params.insert("col_num".to_string(), json!(2718));
params.insert("d".to_string(), json!(5));
params.insert("w".to_string(), json!(2718));
let p = AccuracyProfile::derive(&base_config(AggregationType::CountMinSketch, params));
// e / 2718 ≈ 0.0010001 — very close to 0.001.
assert_eq!(p.kind, AccuracyKind::AdditiveFrequency);
Expand All @@ -525,8 +520,8 @@ mod tests {
// Large heap: 1/heap_size (= 1e-4) dominates the e/w CMS
// bound (e/1e6 ≈ 2.72e-6). Expect ε = 1/heap.
let mut params = HashMap::new();
params.insert("row_num".to_string(), json!(5));
params.insert("col_num".to_string(), json!(1_000_000));
params.insert("d".to_string(), json!(5));
params.insert("w".to_string(), json!(1_000_000));
params.insert("heap_size".to_string(), json!(10_000));
let p = AccuracyProfile::derive(&base_config(
AggregationType::CountMinSketchWithHeap,
Expand All @@ -542,8 +537,8 @@ mod tests {
// Generously-sized heap (1e6) + narrow CMS (w=100) →
// 1/heap (1e-6) ≪ e/w (2.7e-2), so the CMS bound dominates.
let mut params = HashMap::new();
params.insert("row_num".to_string(), json!(4));
params.insert("col_num".to_string(), json!(100));
params.insert("d".to_string(), json!(4));
params.insert("w".to_string(), json!(100));
params.insert("heap_size".to_string(), json!(1_000_000));
let p = AccuracyProfile::derive(&base_config(
AggregationType::CountMinSketchWithHeap,
Expand All @@ -556,8 +551,8 @@ mod tests {
#[test]
fn cms_with_heap_uses_default_heap_size_100() {
let mut params = HashMap::new();
params.insert("row_num".to_string(), json!(4));
params.insert("col_num".to_string(), json!(1000));
params.insert("d".to_string(), json!(4));
params.insert("w".to_string(), json!(1000));
// heap_size absent → default 100 → 1/100 = 0.01 dominates
// e/1000 ≈ 0.00272.
let p = AccuracyProfile::derive(&base_config(
Expand All @@ -574,8 +569,8 @@ mod tests {
// "heap_size" — all three should work.
for alias in ["heap_size", "topk", "k"] {
let mut params = HashMap::new();
params.insert("row_num".to_string(), json!(4));
params.insert("col_num".to_string(), json!(1_000_000));
params.insert("d".to_string(), json!(4));
params.insert("w".to_string(), json!(1_000_000));
params.insert(alias.to_string(), json!(500));
let p = AccuracyProfile::derive(&base_config(
AggregationType::CountMinSketchWithHeap,
Expand All @@ -592,8 +587,8 @@ mod tests {
#[test]
fn countsketch_epsilon_is_one_over_sqrt_w() {
let mut params = HashMap::new();
params.insert("row_num".to_string(), json!(4));
params.insert("col_num".to_string(), json!(100));
params.insert("d".to_string(), json!(4));
params.insert("w".to_string(), json!(100));
let p = AccuracyProfile::derive(&base_config(AggregationType::CountSketch, params));
assert_eq!(p.kind, AccuracyKind::AdditiveFrequency);
assert!((p.epsilon - 0.1).abs() < 1e-9); // 1/√100 = 0.1
Expand Down Expand Up @@ -691,8 +686,8 @@ mod tests {
// i.e. CountSketch is a factor ~1.65 tighter than CMS on
// epsilon alone. Confirms the bounds are not copy-pasted.
let mut params = HashMap::new();
params.insert("row_num".to_string(), json!(4));
params.insert("col_num".to_string(), json!(10000));
params.insert("d".to_string(), json!(4));
params.insert("w".to_string(), json!(10000));
let cms = AccuracyProfile::derive(&base_config(
AggregationType::CountMinSketch,
params.clone(),
Expand Down
32 changes: 16 additions & 16 deletions data_plane/src/tests/accuracy_empirical_validation_tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,8 +73,8 @@ fn cms_3x1000_matches_cormode_muthukrishnan_e_over_w() {
// Cormode-Muthukrishnan 2005: ε = e/w, δ = 1/2^d.
// For (w=1000, d=3): ε = e/1000 ≈ 2.71828e-3, δ = 0.125.
let mut params = HashMap::new();
params.insert("row_num".to_string(), json!(3u64));
params.insert("col_num".to_string(), json!(1000u64));
params.insert("d".to_string(), json!(3u64));
params.insert("w".to_string(), json!(1000u64));
let p = AccuracyProfile::derive(&cfg(AggregationType::CountMinSketch, params));
assert_eq!(p.kind, AccuracyKind::AdditiveFrequency);
assert!((p.epsilon - std::f64::consts::E / 1000.0).abs() < 1e-12);
Expand All @@ -86,8 +86,8 @@ fn count_sketch_4x10000_matches_charikar_one_over_sqrt_w() {
// Charikar-Chen-Farach-Colton 2002: ε = 1/√w for signed
// counter sketch.
let mut params = HashMap::new();
params.insert("row_num".to_string(), json!(4u64));
params.insert("col_num".to_string(), json!(10_000u64));
params.insert("d".to_string(), json!(4u64));
params.insert("w".to_string(), json!(10_000u64));
let p = AccuracyProfile::derive(&cfg(AggregationType::CountSketch, params));
assert_eq!(p.kind, AccuracyKind::AdditiveFrequency);
assert!((p.epsilon - 0.01).abs() < 1e-12);
Expand Down Expand Up @@ -123,8 +123,8 @@ fn cms_with_heap_top_k_combines_cms_and_retention_bounds() {
// (w=1000, d=3, heap=50): CMS bound e/w ≈ 2.72e-3, heap
// bound 1/50 = 0.02. Heap dominates → ε = 0.02.
let mut params = HashMap::new();
params.insert("row_num".to_string(), json!(3u64));
params.insert("col_num".to_string(), json!(1000u64));
params.insert("d".to_string(), json!(3u64));
params.insert("w".to_string(), json!(1000u64));
params.insert("heap_size".to_string(), json!(50u64));
let p = AccuracyProfile::derive(&cfg(AggregationType::CountMinSketchWithHeap, params));
assert_eq!(p.kind, AccuracyKind::TopK);
Expand Down Expand Up @@ -154,8 +154,8 @@ fn cms_epsilon_shrinks_monotonically_with_width() {
let mut last = f64::INFINITY;
for w in [100u64, 500, 2000, 10_000, 100_000] {
let mut params = HashMap::new();
params.insert("row_num".to_string(), json!(4u64));
params.insert("col_num".to_string(), json!(w));
params.insert("d".to_string(), json!(4u64));
params.insert("w".to_string(), json!(w));
let eps = AccuracyProfile::derive(&cfg(AggregationType::CountMinSketch, params)).epsilon;
assert!(eps < last, "CMS ε at w={w}: {eps} should be < {last}");
last = eps;
Expand All @@ -167,8 +167,8 @@ fn cms_delta_shrinks_monotonically_with_depth() {
let mut last = f64::INFINITY;
for d in [2u64, 3, 4, 5, 6, 8] {
let mut params = HashMap::new();
params.insert("row_num".to_string(), json!(d));
params.insert("col_num".to_string(), json!(1000u64));
params.insert("d".to_string(), json!(d));
params.insert("w".to_string(), json!(1000u64));
let delta = AccuracyProfile::derive(&cfg(AggregationType::CountMinSketch, params)).delta;
assert!(delta < last, "CMS δ at d={d}: {delta} should be < {last}");
last = delta;
Expand All @@ -180,8 +180,8 @@ fn countsketch_epsilon_shrinks_monotonically_with_width() {
let mut last = f64::INFINITY;
for w in [100u64, 500, 2000, 10_000, 100_000] {
let mut params = HashMap::new();
params.insert("row_num".to_string(), json!(4u64));
params.insert("col_num".to_string(), json!(w));
params.insert("d".to_string(), json!(4u64));
params.insert("w".to_string(), json!(w));
let eps = AccuracyProfile::derive(&cfg(AggregationType::CountSketch, params)).epsilon;
assert!(
eps < last,
Expand Down Expand Up @@ -229,8 +229,8 @@ fn cms_with_heap_epsilon_shrinks_monotonically_with_heap_size_when_heap_dominate
let mut last = f64::INFINITY;
for heap in [10u64, 100, 1000, 10_000, 100_000] {
let mut params = HashMap::new();
params.insert("row_num".to_string(), json!(5u64));
params.insert("col_num".to_string(), json!(1_000_000u64));
params.insert("d".to_string(), json!(5u64));
params.insert("w".to_string(), json!(1_000_000u64));
params.insert("heap_size".to_string(), json!(heap));
let eps =
AccuracyProfile::derive(&cfg(AggregationType::CountMinSketchWithHeap, params)).epsilon;
Expand All @@ -252,8 +252,8 @@ fn relative_ordering_of_bounds_matches_published_intuition() {
// `countsketch_oxide_matches_cms_oxide` sanity in
// sketch-bench.
let mut params = HashMap::new();
params.insert("row_num".to_string(), json!(4u64));
params.insert("col_num".to_string(), json!(10_000u64));
params.insert("d".to_string(), json!(4u64));
params.insert("w".to_string(), json!(10_000u64));
let cms = AccuracyProfile::derive(&cfg(AggregationType::CountMinSketch, params.clone()));
let cs = AccuracyProfile::derive(&cfg(AggregationType::CountSketch, params));
// CMS at w=10_000: e/10_000 ≈ 0.000272
Expand Down