diff --git a/docs/rqe_sketch_deployment_v1.md b/docs/rqe_sketch_deployment_v1.md index 6414b761..32ecedcb 100644 --- a/docs/rqe_sketch_deployment_v1.md +++ b/docs/rqe_sketch_deployment_v1.md @@ -283,7 +283,7 @@ window size, and query frequency. For deployment `D` serving RAQE `i`: - `lambda`: samples/sec arriving for `D`'s metric, `card(metric.labels) / scrape_interval`. -- `card(G)`: cardinality of `G`, so the number of parallel accumulator instances per window. +- `card(G)`: cardinality of `G`, so the number of parallel accumulator instances per window (one for a shared fixed-size sketch; see below). - `x`, `y`: `D`'s window and slide. - `a_D = x / y`: open windows; each sample is inserted into `a_D` instances of its group. - `S_i`, `T_i`: RAQE `i`'s lookback and interval. @@ -346,27 +346,69 @@ that see far fewer or far more values. DDSketch is sized from the metric's `value_range` when given (capped by values per instance), else from the measured size. -The model uses one formula, `card(G) × m`, which is right for both cells in -use: +The model uses `I × m`, where `I` is instances per window: -- PerGroup + Fixed is that formula as is. -- Shared + PerKey reduces to it because the export divides the exact - accumulators' memory and merge cost by `measured_keys` +- PerGroup + Fixed: `I = card(G)`. +- Shared + PerKey reduces to `I = card(G)` because the export divides the + exact accumulators' memory and merge cost by `measured_keys` (`groups_per_instance`), making `m` a per-group cost. Their query cost is already per group. - -The other two cells are deferred until a family needs them: Shared + Fixed -would be overpriced by `card(G)` (sketch-bench#142), and PerGroup + PerKey -needs the key labels `K`, the metric's labels minus `G`. So for every family -in use, nothing scales with keys below the group: a per-service top-k CMS -costs the same however many endpoints it counts. +- Shared + Fixed: `I = 1`. In code this is the family property + `one_fixed_size_sketch_for_all_groups`, true only for HydraKLL. It applies + to memory and merge; query stays `card(G) × c_qry`, one probe per group. + A fixed-size sketch degrades as groups grow, so it is eligible only when + `card(G)` is at most the `subpopulations` its row was measured at. This + models a sketch keyed by the joined `G` value alone; the current rows fan + out to every label subset (sketch-bench#165), and merging several windows + is not yet measured (sketch-bench#166). + +PerGroup + PerKey is deferred until a family needs it: it needs the key labels +`K`, the metric's labels minus `G`. So for every family in use, nothing scales +with keys below the group: a per-service top-k CMS costs the same however +many endpoints it counts. + +Side by side, with `G = card(G)` and `C` closed windows: + +| | KLL (per group) | exact-sum (per-group normalized) | HydraKLL (shared, fixed) | +|---|---|---|---| +| `m`, `c_mrg` in the row | one group's | whole ÷ groups | whole sketch | +| Ingest CPU | `λ·a_D·c_ins` | `λ·a_D·c_ins` | `λ·a_D·c_ins` | +| Ingest memory | `G·m·a_D` | `G·m·a_D` | `m·a_D` | +| Merge CPU | `G·(n_i−1)·c_mrg/T_i` | same | `(n_i−1)·c_mrg/T_i` | +| Merge memory | `G·m` | same | `m` | +| Query CPU | `G·c_qry/T_i` | same | `G·c_qry/T_i` | +| Query memory | `G ×` output bytes | same | `G ×` output bytes | +| Storage memory | `G·m·C` | same | `m·C` | +| Latency | `G·(c_qry + (n_i−1)·c_mrg)` | same | `(n_i−1)·c_mrg + G·c_qry` | + +### Family properties and key tracker + +Each family has properties, hardcoded by variant in `family_properties` +(an unknown variant panics rather than getting a default): + +- `mergeable_across_windows`: a family without it only serves `S_i = x`. + Every family in use merges. +- `one_fixed_size_sketch_for_all_groups`: the Shared + Fixed cell above. +- `needs_delta_set_key_tracker`: the sketch can't list its groups, so a + DeltaSet (`exact-delta-set`) records each window's keys for the query to + probe. True for HydraKLL. A candidate needing one carries the DeltaSet row + and is dropped if the cost table has none; a deployment without its tracker + is never eligible. The cost table may hold at most one DeltaSet row. + +The tracker is priced as an exact accumulator on the same `x`, `y` and `G`, +in every phase: ingest, merge (union of the `n_i` windows' key sets), query +(enumerate keys) and storage. Its `m` is per key, so it adds `card(G) × m_ds` +per window. The AutoSketch baseline skips shared fixed-size families for now +(sketch-bench#159). Per-RAQE latency is the serial CPU time of one query: ```text -latency_i = card(G) × (c_qry + (n_i − 1) × c_mrg) +latency_i = card(G) × c_qry + I × (n_i − 1) × c_mrg ``` +summed over the sketch and its key tracker, if any. + It is not a wall-clock SLA: it assumes no parallel execution across groups and no cheaper k-way merge. diff --git a/rqe-optimizer/examples/small_problem.rs b/rqe-optimizer/examples/small_problem.rs index 97d23d89..5da09dff 100644 --- a/rqe-optimizer/examples/small_problem.rs +++ b/rqe-optimizer/examples/small_problem.rs @@ -401,7 +401,7 @@ fn main() { ); let mut possible_mappings = Some(1_u64); for raqe in &raqes { - let eligible = eligible_deployments_for(raqe, &deployments); + let eligible = eligible_deployments_for(raqe, &deployments, &facts); println!(" {}: {} eligible deployments", raqe.id, eligible.len()); possible_mappings = possible_mappings.and_then(|count| { count.checked_mul(eligible.len().try_into().expect("usize fits in u64")) @@ -422,7 +422,7 @@ fn main() { return; } - let missing = unservable(&raqes, &deployments); + let missing = unservable(&raqes, &deployments, &facts); if !missing.is_empty() { println!("unservable (no eligible deployment): {missing:?}"); return; @@ -431,7 +431,7 @@ fn main() { let sample_mappings = positive_integer_flag("--sample-mappings", 0); if sample_mappings > 0 { let mut printed = 0; - let result = for_each_mapping_while(&raqes, &deployments, |mapping| { + let result = for_each_mapping_while(&raqes, &deployments, &facts, |mapping| { printed += 1; print_mapping(printed, mapping, &raqes, &deployments); printed < sample_mappings @@ -473,7 +473,7 @@ fn main() { let print_first = positive_integer_flag("--print-first", 0); let started = Instant::now(); let mut processed = 0_u64; - let mapping_count = for_each_mapping(&raqes, &deployments, |mapping| { + let mapping_count = for_each_mapping(&raqes, &deployments, &facts, |mapping| { front.consider(mapping, score(&raqes, &deployments, mapping, &facts)); processed += 1; if processed <= print_first { @@ -500,7 +500,7 @@ fn main() { "WARNING: eager mode will retain every feasible mapping. For larger workloads, use --milp, \ --candidates-only, or --sample-mappings N instead." ); - let mappings = brute_force(&raqes, &deployments); + let mappings = brute_force(&raqes, &deployments, &facts); println!("{} feasible full mappings", mappings.len()); let plan_costs: Vec<_> = mappings diff --git a/rqe-optimizer/src/analytical_cost_model.rs b/rqe-optimizer/src/analytical_cost_model.rs index 7bc85cdb..88895fc5 100644 --- a/rqe-optimizer/src/analytical_cost_model.rs +++ b/rqe-optimizer/src/analytical_cost_model.rs @@ -8,14 +8,15 @@ //! //! [`score`] and the MILP share the per-phase functions below. //! -//! Measured costs are per instance, and a deployment holds `card(G)` instances -//! per window. Exact accumulators hold many groups in one instance, but the -//! export divides their memory and merge cost by the group count, so -//! `card(G) ×` prices them correctly too. -// ponytail: a family with one fixed-size instance for all groups (HydraKLL, -// sketch-bench#142) needs a per-family shape so it isn't multiplied by card(G). - -use crate::{secs, Capability, Deployment, Mapping, Millis, Raqe, WorkloadFacts}; +//! Measured costs are per instance. A deployment holds `card(G)` instances +//! per window, or one if its family keeps [one fixed-size sketch for all +//! groups](crate::FamilyProperties::one_fixed_size_sketch_for_all_groups). +//! Exact accumulators hold many groups in one instance, but the export +//! divides their memory and merge cost by the group count, so `card(G) ×` +//! prices them correctly too. A [key tracker](crate::Deployment::key_tracker) +//! is priced like an exact accumulator on the same windows, in every phase. + +use crate::{secs, AtomicCostEntry, Capability, Deployment, Mapping, Millis, Raqe, WorkloadFacts}; use std::collections::{BTreeMap, BTreeSet}; pub(crate) const BYTES_PER_GIB: f64 = (1u64 << 30) as f64; @@ -77,13 +78,40 @@ impl PlanCost { } } -/// `card(G)`: cardinality of `G`, so the number of parallel accumulator -/// instances per window. +/// `card(G)`: cardinality of `G`, the number of groups. fn group_count(deployment: &Deployment, facts: &WorkloadFacts) -> f64 { facts[&deployment.metric].cardinality[&deployment.grouping_labels] as f64 } -/// `m`: memory of one instance holding `span_ms` of one group's samples: a +/// The sketch and its key tracker, if any, each with its instances per +/// window: 1 for a fixed-size sketch shared by all groups, else `card(G)`. +fn priced_parts<'a>( + deployment: &'a Deployment, + facts: &WorkloadFacts, +) -> impl Iterator { + let properties = deployment.properties(); + assert_eq!( + properties.needs_delta_set_key_tracker, + deployment.key_tracker.is_some(), + "{}: key tracker present iff the family needs one, or the plan is mispriced", + deployment.config.sketch + ); + let groups = group_count(deployment, facts); + let sketch_instances = if properties.one_fixed_size_sketch_for_all_groups { + 1.0 + } else { + groups + }; + std::iter::once((&deployment.config, sketch_instances)).chain( + deployment + .key_tracker + .iter() + .map(move |tracker| (tracker, groups)), + ) +} + +/// `m`: memory of one `config` instance (`deployment`'s sketch or its key +/// tracker) holding `span_ms` of one group's samples: a /// window `x`, or the lookback `L` for a query's merge accumulator. /// DDSketch on a metric with a known value range holds one store (values are /// positive) of [`dd_bucket_count`] buckets, capped by the values it holds, @@ -91,11 +119,11 @@ fn group_count(deployment: &Deployment, facts: &WorkloadFacts) -> f64 { /// an upper bound for a group whose own values span less. Every other row, /// and DDSketch without a range, keeps the measured size. pub(crate) fn instance_memory_bytes( + config: &AtomicCostEntry, deployment: &Deployment, facts: &WorkloadFacts, span_ms: Millis, ) -> f64 { - let config = &deployment.config; let metric = &facts[&deployment.metric]; let alpha = config.sketch_config["params"]["alpha"].as_f64(); match (config.sketch.as_str(), metric.value_range, alpha) { @@ -128,74 +156,86 @@ fn merged_window_count(raqe: &Raqe, deployment: &Deployment) -> f64 { .expect("only eligible pairs are costed") as f64 } -/// CPU `λ · (x/y) · c_ins`; memory `card(G) · m · (x/y)`. +/// Per part: CPU `λ · (x/y) · c_ins`; memory `instances · m · (x/y)`. pub(crate) fn ingest(deployment: &Deployment, facts: &WorkloadFacts) -> PhaseCost { let open_windows = open_window_count(deployment); - PhaseCost { - cpu_secs_per_sec: facts[&deployment.metric].arrival_rate_per_sec() - * open_windows - * deployment.config.insert_cpu_secs, - memory_bytes: group_count(deployment, facts) - * instance_memory_bytes(deployment, facts, deployment.window_ms) - * open_windows, + let arrival_rate = facts[&deployment.metric].arrival_rate_per_sec(); + let mut cost = PhaseCost::default(); + for (config, instances) in priced_parts(deployment, facts) { + cost += PhaseCost { + cpu_secs_per_sec: arrival_rate * open_windows * config.insert_cpu_secs, + memory_bytes: instances + * instance_memory_bytes(config, deployment, facts, deployment.window_ms) + * open_windows, + }; } + cost } -/// CPU `card(G) · (L/x − 1) · c_mrg / T`; memory `card(G) · m`, one -/// accumulator per group. A direct query (`L == x`) merges nothing and costs -/// neither. +/// Per part: CPU `instances · (L/x − 1) · c_mrg / T`; memory `instances · m`, +/// every merged accumulator held at once, each holding the whole lookback. A +/// direct query (`L == x`) merges nothing and costs neither. pub(crate) fn merge(raqe: &Raqe, deployment: &Deployment, facts: &WorkloadFacts) -> PhaseCost { - let groups = group_count(deployment, facts); - let merges_per_group = merged_window_count(raqe, deployment) - 1.0; - PhaseCost { - cpu_secs_per_sec: groups * merges_per_group * deployment.config.merge_cpu_secs - / secs(raqe.interval_ms), - memory_bytes: groups * merge_memory_per_group(raqe, deployment, facts), + let merges_per_instance = merged_window_count(raqe, deployment) - 1.0; + if merges_per_instance == 0.0 { + return PhaseCost::default(); } -} - -/// One accumulator if the query merges at all, holding the whole lookback. -pub(crate) fn merge_memory_per_group( - raqe: &Raqe, - deployment: &Deployment, - facts: &WorkloadFacts, -) -> f64 { - if merged_window_count(raqe, deployment) > 1.0 { - instance_memory_bytes(deployment, facts, raqe.lookback_ms) - } else { - 0.0 + let mut cost = PhaseCost::default(); + for (config, instances) in priced_parts(deployment, facts) { + cost += PhaseCost { + cpu_secs_per_sec: instances * merges_per_instance * config.merge_cpu_secs + / secs(raqe.interval_ms), + memory_bytes: instances + * instance_memory_bytes(config, deployment, facts, raqe.lookback_ms), + }; } + cost } -/// CPU `card(G) · c_qry / T`; memory `card(G) ·` output bytes per group. +/// CPU `card(G) · Σ c_qry / T`, one probe per group of each part; memory +/// `card(G) ·` output bytes per group. pub(crate) fn query(raqe: &Raqe, deployment: &Deployment, facts: &WorkloadFacts) -> PhaseCost { let groups = group_count(deployment, facts); let output_bytes_per_group = match raqe.capability { Capability::TopK => TOPK_ENTRIES * OUTPUT_BYTES_PER_TOPK_ENTRY, _ => OUTPUT_BYTES_PER_VALUE, }; + let query_cpu_secs_per_group: f64 = priced_parts(deployment, facts) + .map(|(config, _)| config.query_cpu_secs) + .sum(); PhaseCost { - cpu_secs_per_sec: groups * deployment.config.query_cpu_secs / secs(raqe.interval_ms), + cpu_secs_per_sec: groups * query_cpu_secs_per_group / secs(raqe.interval_ms), memory_bytes: groups * output_bytes_per_group, } } -/// Closed windows kept to answer `raqe`: `card(G) · m · ((L − x)/y + 1)`. +/// Closed windows kept to answer `raqe`, per part: +/// `instances · m · ((L − x)/y + 1)`. pub(crate) fn storage_bytes(raqe: &Raqe, deployment: &Deployment, facts: &WorkloadFacts) -> f64 { let closed_windows = deployment .closed_instance_count(raqe.lookback_ms) .expect("only eligible pairs are costed") as f64; - group_count(deployment, facts) - * instance_memory_bytes(deployment, facts, deployment.window_ms) - * closed_windows + priced_parts(deployment, facts) + .map(|(config, instances)| { + instances + * instance_memory_bytes(config, deployment, facts, deployment.window_ms) + * closed_windows + }) + .sum() } -/// Serial CPU time of one query, in ms: `card(G) · (c_qry + (L/x − 1) · c_mrg)`. +/// Serial CPU time of one query, in ms, summed over parts: +/// `card(G) · c_qry + instances · (L/x − 1) · c_mrg`. pub(crate) fn query_latency_ms(raqe: &Raqe, deployment: &Deployment, facts: &WorkloadFacts) -> f64 { - let merges_per_group = merged_window_count(raqe, deployment) - 1.0; + let groups = group_count(deployment, facts); + let merges_per_instance = merged_window_count(raqe, deployment) - 1.0; 1000.0 - * group_count(deployment, facts) - * (deployment.config.query_cpu_secs + merges_per_group * deployment.config.merge_cpu_secs) + * priced_parts(deployment, facts) + .map(|(config, instances)| { + groups * config.query_cpu_secs + + instances * merges_per_instance * config.merge_cpu_secs + }) + .sum::() } pub fn score( @@ -271,7 +311,10 @@ mod tests { .value_range = Some((1.0, 1000.0)); let window = |d: &Deployment| d.window_ms; - assert_eq!(instance_memory_bytes(&dd, &ranged, window(&dd)), 80.0); // 10 buckets × 8 B + assert_eq!( + instance_memory_bytes(&dd.config, &dd, &ranged, window(&dd)), + 80.0 + ); // 10 buckets × 8 B assert_eq!(ingest(&dd, &ranged).memory_bytes, 80.0); // 1 sample/sec into a 5 s window: 5 values, fewer than the 10 buckets. @@ -294,11 +337,14 @@ mod tests { // No range, or not DDSketch: the measured size. assert_eq!( - instance_memory_bytes(&dd, &facts(1, 1), window(&dd)), + instance_memory_bytes(&dd.config, &dd, &facts(1, 1), window(&dd)), 1_000.0 ); let cms = deployment(1_000.0, 0.0, 0.0, 0.0, 60_000, 60_000); - assert_eq!(instance_memory_bytes(&cms, &ranged, window(&cms)), 1_000.0); + assert_eq!( + instance_memory_bytes(&cms.config, &cms, &ranged, window(&cms)), + 1_000.0 + ); } #[test] @@ -352,4 +398,47 @@ mod tests { // Two RAQEs × 2 groups × 10 bytes. assert_eq!(result.merge.memory_bytes, 2.0 * 2.0 * 10.0); } + #[test] + fn a_shared_sketch_is_priced_once_and_its_tracker_per_group() { + // Same windows as `scores_each_phase`: 2 open, 3 merged, 5 closed; + // 4 groups, 7 samples/sec. Hydra: 100 bytes for all groups. Tracker: + // 10 bytes per group. + let mut deployment = deployment(100.0, 2.0, 3.0, 5.0, 20_000, 10_000); + deployment.capability = Capability::Quantile; + deployment.config.sketch = "hydra-kll".into(); + deployment.key_tracker = Some(AtomicCostEntry { + sketch: crate::KEY_TRACKER_FAMILY.into(), + mem_bytes_per_instance: 10.0, + insert_cpu_secs: 1.0, + merge_cpu_secs: 1.0, + query_cpu_secs: 1.0, + ..deployment.config.clone() + }); + let raqe = Raqe { + capability: Capability::Quantile, + ..raqe(60_000, 30_000) + }; + let result = score(&[raqe], &[deployment], &vec![0], &facts(4, 7)); + + assert_eq!(result.ingest.cpu_secs_per_sec, 7.0 * 2.0 * (2.0 + 1.0)); + assert_eq!(result.ingest.memory_bytes, 100.0 * 2.0 + 4.0 * 10.0 * 2.0); + assert_eq!( + result.merge.cpu_secs_per_sec, + (2.0 * 3.0 + 4.0 * 2.0 * 1.0) / 30.0 + ); + assert_eq!(result.merge.memory_bytes, 100.0 + 4.0 * 10.0); + assert_eq!(result.query.cpu_secs_per_sec, 4.0 * (5.0 + 1.0) / 30.0); + assert_eq!(result.query.memory_bytes, 4.0 * OUTPUT_BYTES_PER_VALUE); + assert_eq!(result.storage.memory_bytes, 100.0 * 5.0 + 4.0 * 10.0 * 5.0); + // 4 × (5 + 1) query + 2 × 3 Hydra merges + 4 × 2 × 1 tracker merges, in s. + assert_eq!(result.query_latency_ms, vec![38_000.0]); + } + + #[test] + #[should_panic(expected = "key tracker present iff the family needs one")] + fn a_family_that_needs_a_tracker_is_never_priced_without_one() { + let mut deployment = deployment(100.0, 1.0, 1.0, 1.0, 60_000, 60_000); + deployment.config.sketch = "hydra-kll".into(); + ingest(&deployment, &facts(1, 1)); + } } diff --git a/rqe-optimizer/src/autosketch.rs b/rqe-optimizer/src/autosketch.rs index 2180bc28..c46bec73 100644 --- a/rqe-optimizer/src/autosketch.rs +++ b/rqe-optimizer/src/autosketch.rs @@ -11,7 +11,8 @@ //! One extension over the paper: AutoSketch's compiler maps an operator to one //! sketch algorithm and tunes its parameters, while this search covers every //! deployable variant serving the RAQE's capability and keeps the cheapest, -//! so it chooses from the same sketches as the MILP. +//! so it chooses from the same sketches as the MILP, except those shared by +//! all groups (sketch-bench#159). use crate::candidates::gcd; use crate::{AtomicCostEntry, Deployment, Mapping, Millis, Raqe}; @@ -89,6 +90,7 @@ pub fn plan( config: costs[s.selected.expect("unservable RAQEs returned above")].clone(), window_ms, slide_ms, + key_tracker: None, } }) .collect(); @@ -113,6 +115,9 @@ pub fn search( let grids: Vec = raqe .capability .candidate_families(allow_undeployable_families) + // ponytail: ranking compares per-instance memory, which a sketch + // shared by all groups isn't comparable on; sketch-bench#159. + .filter(|variant| !crate::family_properties(variant).one_fixed_size_sketch_for_all_groups) .filter_map(|variant| Grid::new(variant, costs)) .collect(); @@ -455,7 +460,7 @@ mod tests { assert_eq!(window_adapter(&raqes[0]), (3_600_000, 60_000)); assert_eq!(window_adapter(&raqes[1]), (3_600_000, 40_000)); for (r, d) in raqes.iter().zip(&plan.deployments) { - assert!(is_eligible(r, d), "{} not eligible", r.id); + assert!(is_eligible(r, d, &facts(1, 1)), "{} not eligible", r.id); } } diff --git a/rqe-optimizer/src/candidates.rs b/rqe-optimizer/src/candidates.rs index 55283aaf..9c778b89 100644 --- a/rqe-optimizer/src/candidates.rs +++ b/rqe-optimizer/src/candidates.rs @@ -1,7 +1,10 @@ //! Candidate generation and eligibility (§3). -use crate::analytical_cost_model::{instance_memory_bytes, merge_memory_per_group}; -use crate::{AtomicCostEntry, Capability, Deployment, LabelSet, Millis, Raqe, WorkloadFacts}; +use crate::analytical_cost_model; +use crate::{ + AtomicCostEntry, Capability, Deployment, LabelSet, MetricFacts, Millis, Raqe, WorkloadFacts, + KEY_TRACKER_FAMILY, +}; use std::collections::{BTreeMap, BTreeSet}; pub fn gcd(a: Millis, b: Millis) -> Millis { @@ -41,18 +44,45 @@ fn subset_gcds(values: impl IntoIterator) -> BTreeSet { gcds } +/// Whether a fixed-size sketch shared by all groups was measured holding at +/// least `groups` of them. Its accuracy degrades as groups grow, and the row +/// holds one measured point (`subpopulations`). Models a sketch keyed by the +/// joined `G` value alone, so it holds exactly `card(G)` subpopulations; the +/// current rows fan out to every label subset (sketch-bench#165). +// ponytail: one point, so larger groupings are never planned. Rows at more +// group counts (sketch-bench#143 S5) loosen this with no code change. +fn measured_at_group_count(config: &AtomicCostEntry, groups: u64) -> bool { + config + .query_accuracy + .get("subpopulations") + .is_some_and(|&measured| groups as f64 <= measured) +} + /// Windows and slides are multiples of the metric's scrape interval: anything /// finer only splits one scrape's samples. fn candidate_deployments( group: &[&Raqe], costs: &[AtomicCostEntry], - scrape_interval_ms: Millis, + metric_facts: &MetricFacts, allow_undeployable_families: bool, ) -> Vec { if group.is_empty() { return Vec::new(); } let capability = group[0].capability; + let scrape_interval_ms = metric_facts.scrape_interval_ms; + // The tracker is deployable whenever the family needing it is: ASAPQuery + // runs it as DeltaSetAggregator. + let tracker_rows: Vec<_> = costs + .iter() + .filter(|c| c.sketch == KEY_TRACKER_FAMILY) + .collect(); + assert!( + tracker_rows.len() <= 1, + "{} {KEY_TRACKER_FAMILY} rows in the cost table; the tracker's price would depend on row order", + tracker_rows.len() + ); + let tracker_row = tracker_rows.first().copied(); let windows = group .iter() .flat_map(|r| divisors(r.lookback_ms / scrape_interval_ms)) @@ -74,6 +104,15 @@ fn candidate_deployments( .candidate_families(allow_undeployable_families) .any(|family| family == c.sketch) }) { + let key_tracker = + if crate::family_properties(&config.sketch).needs_delta_set_key_tracker { + match tracker_row { + Some(tracker) => Some(tracker.clone()), + None => continue, + } + } else { + None + }; for &slide_ms in &slides { if slide_ms.is_multiple_of(scrape_interval_ms) { deployments.push(Deployment { @@ -84,6 +123,7 @@ fn candidate_deployments( config: config.clone(), window_ms, slide_ms, + key_tracker: key_tracker.clone(), }); } } @@ -131,12 +171,7 @@ pub fn build_all_candidates_unpruned( groups .iter() .flat_map(|(&(_, metric, _, _), group)| { - candidate_deployments( - group, - costs, - facts[metric].scrape_interval_ms, - allow_undeployable_families, - ) + candidate_deployments(group, costs, &facts[metric], allow_undeployable_families) }) .collect() } @@ -144,24 +179,20 @@ pub fn build_all_candidates_unpruned( /// Remove a candidate only when another candidate can replace it in every /// mapping without making any modeled objective worse. /// -/// This comparison is deliberately local to a (capability, metric, -/// spatial_filter, grouping) group, where `card(G)`, arrival rate and query output size are common -/// multipliers. So it compares ingest CPU and memory per group, then latency -/// (merge and query CPU), merge memory and stored memory for each RAQE the -/// dominated candidate can serve. +/// This comparison is local to a (capability, metric, spatial_filter, +/// grouping) group, where query output size is common. It compares ingest CPU +/// and memory, then latency (merge and query CPU), merge memory and stored +/// memory for each RAQE the dominated candidate can serve, all as the +/// analytical cost model prices them: a sketch shared by all groups and a +/// sketch per group don't scale alike with `card(G)`. pub fn prune_dominated_candidates( raqes: &[Raqe], facts: &WorkloadFacts, candidates: Vec, ) -> Vec { - let eligibility: Vec> = candidates + let costs: Vec = candidates .iter() - .map(|candidate| { - raqes - .iter() - .map(|raqe| is_eligible(raqe, candidate)) - .collect() - }) + .map(|candidate| CandidateCosts::new(candidate, raqes, facts)) .collect(); candidates @@ -174,11 +205,9 @@ pub fn prune_dominated_candidates( } match dominance( &candidates[other_index], - &eligibility[other_index], + &costs[other_index], candidate, - &eligibility[candidate_index], - raqes, - facts, + &costs[candidate_index], ) { Dominance::StrictlyBetter => true, // The earlier of two identical candidates wins, so a tie @@ -192,6 +221,37 @@ pub fn prune_dominated_candidates( .collect() } +/// One candidate's costs, computed once for every pairwise comparison. +struct CandidateCosts { + /// Ingest CPU and memory. + ingest: [f64; 2], + /// Per RAQE: latency, merge memory and stored memory if the candidate + /// serves it, else `None`. + per_raqe: Vec>, +} + +impl CandidateCosts { + fn new(candidate: &Deployment, raqes: &[Raqe], facts: &WorkloadFacts) -> Self { + let ingest = analytical_cost_model::ingest(candidate, facts); + let per_raqe = raqes + .iter() + .map(|raqe| { + is_eligible(raqe, candidate, facts).then(|| { + [ + analytical_cost_model::query_latency_ms(raqe, candidate, facts), + analytical_cost_model::merge(raqe, candidate, facts).memory_bytes, + analytical_cost_model::storage_bytes(raqe, candidate, facts), + ] + }) + }) + .collect(); + Self { + ingest: [ingest.cpu_secs_per_sec, ingest.memory_bytes], + per_raqe, + } + } +} + enum Dominance { StrictlyBetter, Equal, @@ -202,55 +262,54 @@ enum Dominance { /// `original` serves and is no worse on any cost. fn dominance( replacement: &Deployment, - replacement_coverage: &[bool], + replacement_costs: &CandidateCosts, original: &Deployment, - original_coverage: &[bool], - raqes: &[Raqe], - facts: &WorkloadFacts, + original_costs: &CandidateCosts, ) -> Dominance { let same_group = replacement.capability == original.capability && replacement.metric == original.metric && replacement.spatial_filter == original.spatial_filter && replacement.grouping_labels == original.grouping_labels; - let covers_original = original_coverage + let covers_original = original_costs + .per_raqe .iter() - .zip(replacement_coverage) - .all(|(&original_serves, &replacement_serves)| !original_serves || replacement_serves); + .zip(&replacement_costs.per_raqe) + .all(|(original_serves, replacement_serves)| { + original_serves.is_none() || replacement_serves.is_some() + }); if !same_group || !covers_original { return Dominance::NotDominating; } - // Each pair is (replacement's cost, original's cost). - let mut costs = vec![ - (ingest_cpu(replacement), ingest_cpu(original)), - ( - ingest_memory(replacement, facts), - ingest_memory(original, facts), - ), - ]; - for (raqe, _) in raqes + // Each pair is (replacement's cost, original's cost), over the RAQEs + // `original` serves. + let mut costs: Vec<(f64, f64)> = replacement_costs + .ingest + .into_iter() + .zip(original_costs.ingest) + .collect(); + for (replacement_raqe, original_raqe) in replacement_costs + .per_raqe .iter() - .zip(original_coverage) - .filter(|(_, &serves)| serves) + .zip(&original_costs.per_raqe) { - costs.push(( - query_latency(replacement, raqe), - query_latency(original, raqe), - )); - costs.push(( - merge_memory_per_group(raqe, replacement, facts), - merge_memory_per_group(raqe, original, facts), - )); - costs.push(( - stored_memory(replacement, raqe, facts), - stored_memory(original, raqe, facts), - )); + if let (Some(replacement_raqe), Some(original_raqe)) = (replacement_raqe, original_raqe) { + costs.extend( + replacement_raqe + .iter() + .copied() + .zip(original_raqe.iter().copied()), + ); + } } - let serves_more = original_coverage + let serves_more = original_costs + .per_raqe .iter() - .zip(replacement_coverage) - .any(|(&original_serves, &replacement_serves)| !original_serves && replacement_serves); + .zip(&replacement_costs.per_raqe) + .any(|(original_serves, replacement_serves)| { + original_serves.is_none() && replacement_serves.is_some() + }); if costs .iter() .any(|(replacement_cost, original_cost)| replacement_cost > original_cost) @@ -267,48 +326,23 @@ fn dominance( } } -fn open_window_count(deployment: &Deployment) -> f64 { - deployment.active_instance_count().unwrap_or(u64::MAX) as f64 -} - -/// Insert CPU per sample. Arrival rate is a common multiplier within a group, -/// so it is left out. -fn ingest_cpu(deployment: &Deployment) -> f64 { - open_window_count(deployment) * deployment.config.insert_cpu_secs -} - -/// Open-window bytes per group. `card(G)` is a common multiplier within a -/// group, so it is left out. -fn ingest_memory(deployment: &Deployment, facts: &WorkloadFacts) -> f64 { - open_window_count(deployment) * instance_memory_bytes(deployment, facts, deployment.window_ms) -} - -fn query_latency(deployment: &Deployment, raqe: &Raqe) -> f64 { - let instances = deployment - .query_instance_count(raqe.lookback_ms) - .expect("candidate coverage only contains exactly tiled RAQEs"); - deployment.config.query_cpu_secs - + instances.saturating_sub(1) as f64 * deployment.config.merge_cpu_secs -} - -/// Closed-window bytes per group when serving `raqe`. -fn stored_memory(deployment: &Deployment, raqe: &Raqe, facts: &WorkloadFacts) -> f64 { - deployment - .closed_instance_count(raqe.lookback_ms) - .expect("candidate coverage only contains exactly tiled RAQEs") as f64 - * instance_memory_bytes(deployment, facts, deployment.window_ms) -} - -pub fn eligible_deployments_for(r: &Raqe, deployments: &[Deployment]) -> Vec { +pub fn eligible_deployments_for( + r: &Raqe, + deployments: &[Deployment], + facts: &WorkloadFacts, +) -> Vec { deployments .iter() .enumerate() - .filter(|(_, d)| is_eligible(r, d)) + .filter(|(_, d)| is_eligible(r, d, facts)) .map(|(i, _)| i) .collect() } -pub fn is_eligible(r: &Raqe, d: &Deployment) -> bool { +/// Whether `d` can serve `r`. Every rule lives here, so the MILP, enumeration +/// and candidate pruning agree on what is valid. +pub fn is_eligible(r: &Raqe, d: &Deployment, facts: &WorkloadFacts) -> bool { + let properties = d.properties(); r.capability == d.capability && r.metric == d.metric && r.spatial_filter == d.spatial_filter @@ -318,6 +352,10 @@ pub fn is_eligible(r: &Raqe, d: &Deployment) -> bool { && d.window_ms.is_multiple_of(d.slide_ms) && r.lookback_ms.is_multiple_of(d.window_ms) && r.interval_ms.is_multiple_of(d.slide_ms) + && (properties.mergeable_across_windows || r.lookback_ms == d.window_ms) + && properties.needs_delta_set_key_tracker == d.key_tracker.is_some() + && (!properties.one_fixed_size_sketch_for_all_groups + || measured_at_group_count(&d.config, facts[&d.metric].cardinality[&d.grouping_labels])) && r.accuracy_ok_for(&d.config, r.lookback_ms / d.window_ms) } @@ -369,19 +407,20 @@ mod tests { config: config.clone(), window_ms, slide_ms: window_ms, + key_tracker: None, }; let r = raqe("a", 60_000, 60_000); // One window reads query_accuracy; 2 reads 1 and 4; 4 reads 4 alone. - assert!(is_eligible(&r, &at_window(60_000))); - assert!(is_eligible(&r, &at_window(30_000))); - assert!(is_eligible(&r, &at_window(15_000))); + assert!(is_eligible(&r, &at_window(60_000), &facts(1, 1))); + assert!(is_eligible(&r, &at_window(30_000), &facts(1, 1))); + assert!(is_eligible(&r, &at_window(15_000), &facts(1, 1))); // 12 and 20 merges bracket the bad count 16 from either side; 60 // reads 16 and 64, so it fails too even though 64 alone passes. - assert!(!is_eligible(&r, &at_window(5_000))); - assert!(!is_eligible(&r, &at_window(3_000))); - assert!(!is_eligible(&r, &at_window(1_000))); + assert!(!is_eligible(&r, &at_window(5_000), &facts(1, 1))); + assert!(!is_eligible(&r, &at_window(3_000), &facts(1, 1))); + assert!(!is_eligible(&r, &at_window(1_000), &facts(1, 1))); // Past the largest count reads only 64. - assert!(is_eligible(&r, &at_window(500))); + assert!(is_eligible(&r, &at_window(500), &facts(1, 1))); } #[test] fn shared_slide_comes_from_subset_gcd_not_all_divisors() { @@ -421,7 +460,8 @@ mod tests { ); assert!(candidates .iter() - .all(|d| is_eligible(&unfiltered, d) != is_eligible(&filtered, d))); + .all(|d| is_eligible(&unfiltered, d, &facts(1, 1)) + != is_eligible(&filtered, d, &facts(1, 1)))); } #[test] @@ -474,21 +514,24 @@ mod tests { config: cost(), window_ms: 120_000, slide_ms: 60_000, + key_tracker: None, }; - assert!(is_eligible(&r, &d)); + assert!(is_eligible(&r, &d, &facts(1, 1))); assert!(!is_eligible( &r, &Deployment { window_ms: 128_000, ..d.clone() - } + }, + &facts(1, 1) )); assert!(!is_eligible( &r, &Deployment { slide_ms: 70_000, ..d - } + }, + &facts(1, 1) )); } @@ -503,6 +546,7 @@ mod tests { config: cost(), window_ms: 60_000, slide_ms: 60_000, + key_tracker: None, }; let fine = Deployment { slide_ms: 30_000, @@ -526,6 +570,7 @@ mod tests { config: cost.clone(), window_ms: 60_000, slide_ms: 60_000, + key_tracker: None, }; let small_window = Deployment { window_ms: 30_000, @@ -557,6 +602,7 @@ mod tests { }, window_ms: 60_000, slide_ms: 60_000, + key_tracker: None, }; // Smaller, faster sketches, but (60 − 20) / 20 + 1 = 3 of them: 18 bytes. let panes = Deployment { @@ -595,6 +641,7 @@ mod tests { }, window_ms: 60_000, slide_ms: 60_000, + key_tracker: None, }; // Smaller and cheaper on every other cost, but merges two windows, // holding a 4-byte accumulator per group. @@ -615,4 +662,109 @@ mod tests { assert_eq!(retained, vec![direct, halves]); } + /// A whole-sketch hydra-kll row measured at `subpopulations` groups, and + /// a per-key DeltaSet row. + fn hydra_and_tracker(subpopulations: f64) -> [AtomicCostEntry; 2] { + [ + AtomicCostEntry { + sketch: "hydra-kll".into(), + mem_bytes_per_instance: 1000.0, + query_accuracy: BTreeMap::from([ + ("err".into(), 0.1), + ("subpopulations".into(), subpopulations), + ]), + ..cost() + }, + AtomicCostEntry { + sketch: KEY_TRACKER_FAMILY.into(), + ..cost() + }, + ] + } + + fn quantile_raqe() -> Raqe { + Raqe { + capability: Capability::Quantile, + ..raqe("r", 60_000, 60_000) + } + } + + #[test] + fn hydra_carries_the_delta_set_tracker_and_needs_one() { + let costs = hydra_and_tracker(100.0); + let candidates = build_all_candidates(&[quantile_raqe()], &costs, &facts(10, 10), true); + assert!(!candidates.is_empty()); + assert!(candidates + .iter() + .all(|d| d.key_tracker.as_ref() == Some(&costs[1]))); + let untracked = build_all_candidates(&[quantile_raqe()], &costs[..1], &facts(10, 10), true); + assert!(untracked.is_empty()); + } + + #[test] + fn hydra_is_planned_only_up_to_its_measured_group_count() { + let costs = hydra_and_tracker(100.0); + let servable = |groups| { + let facts = facts(groups, groups); + let candidates = + build_all_candidates_unpruned(&[quantile_raqe()], &costs, &facts, true); + !eligible_deployments_for(&quantile_raqe(), &candidates, &facts).is_empty() + }; + assert!(servable(100)); + assert!(!servable(101)); + } + + #[test] + fn a_hydra_deployment_without_its_tracker_is_never_eligible() { + let costs = hydra_and_tracker(100.0); + let facts = facts(10, 10); + let candidates = build_all_candidates(&[quantile_raqe()], &costs, &facts, true); + let untracked = Deployment { + key_tracker: None, + ..candidates[0].clone() + }; + assert!(is_eligible(&quantile_raqe(), &candidates[0], &facts)); + assert!(!is_eligible(&quantile_raqe(), &untracked, &facts)); + } + + #[test] + #[should_panic(expected = "exact-delta-set rows in the cost table")] + fn two_tracker_rows_are_refused() { + let [hydra, tracker] = hydra_and_tracker(100.0); + build_all_candidates( + &[quantile_raqe()], + &[hydra, tracker.clone(), tracker], + &facts(10, 10), + true, + ); + } + + #[test] + fn a_shared_sketch_is_not_pruned_by_a_smaller_per_group_one() { + // Same CPU per op, and 10 bytes per group against Hydra's 1000 for + // the whole sketch: compared per group, Hydra is dominated. Over 100 + // groups, memory and ingest tie (the tracker is free here), and over + // 2 merged windows Hydra's latency is 100 queries + 1 merge against + // KLL's 100 + 100. Latency alone keeps Hydra. + let r = Raqe { + lookback_ms: 120_000, + ..quantile_raqe() + }; + let [hydra, tracker] = hydra_and_tracker(100.0); + let kll = AtomicCostEntry { + sketch: "kll-percall".into(), + mem_bytes_per_instance: 10.0, + ..cost() + }; + let tracker = AtomicCostEntry { + mem_bytes_per_instance: 0.0, + insert_cpu_secs: 0.0, + query_cpu_secs: 0.0, + merge_cpu_secs: 0.0, + ..tracker + }; + let candidates = build_all_candidates(&[r], &[kll, hydra, tracker], &facts(100, 100), true); + assert!(candidates.iter().any(|d| d.config.sketch == "hydra-kll")); + assert!(candidates.iter().any(|d| d.config.sketch == "kll-percall")); + } } diff --git a/rqe-optimizer/src/enumerate.rs b/rqe-optimizer/src/enumerate.rs index 48a98ac3..ebcb2ff5 100644 --- a/rqe-optimizer/src/enumerate.rs +++ b/rqe-optimizer/src/enumerate.rs @@ -10,24 +10,32 @@ //! An ILP would replace this module alone: `Mapping` already encodes //! `x_{i,D}` and `y_D`. -use crate::{candidates::eligible_deployments_for, Deployment, Mapping, Raqe}; +use crate::{candidates::eligible_deployments_for, Deployment, Mapping, Raqe, WorkloadFacts}; /// RAQEs nothing in `deployments` can serve (§3.4). Surfaced separately from /// `brute_force`: an unservable RAQE is a modeling gap (tolerance too tight, /// no such config), not a normal empty result. -pub fn unservable(raqes: &[Raqe], deployments: &[Deployment]) -> Vec { +pub fn unservable( + raqes: &[Raqe], + deployments: &[Deployment], + facts: &WorkloadFacts, +) -> Vec { raqes .iter() - .filter(|r| eligible_deployments_for(r, deployments).is_empty()) + .filter(|r| eligible_deployments_for(r, deployments, facts).is_empty()) .map(|r| r.id.clone()) .collect() } /// Every feasible full mapping. Empty if any RAQE is unservable -- check /// [`unservable`] to tell the two cases apart. -pub fn brute_force(raqes: &[Raqe], deployments: &[Deployment]) -> Vec { +pub fn brute_force( + raqes: &[Raqe], + deployments: &[Deployment], + facts: &WorkloadFacts, +) -> Vec { let mut mappings = Vec::new(); - for_each_mapping(raqes, deployments, |mapping| { + for_each_mapping(raqes, deployments, facts, |mapping| { mappings.push(mapping.to_vec()) }); mappings @@ -39,9 +47,10 @@ pub fn brute_force(raqes: &[Raqe], deployments: &[Deployment]) -> Vec { pub fn for_each_mapping( raqes: &[Raqe], deployments: &[Deployment], + facts: &WorkloadFacts, mut visit: impl FnMut(&Mapping), ) -> u64 { - for_each_mapping_while(raqes, deployments, |mapping| { + for_each_mapping_while(raqes, deployments, facts, |mapping| { visit(mapping); true }) @@ -62,11 +71,12 @@ pub struct EnumerationResult { pub fn for_each_mapping_while( raqes: &[Raqe], deployments: &[Deployment], + facts: &WorkloadFacts, mut visit: impl FnMut(&Mapping) -> bool, ) -> EnumerationResult { let choices: Vec> = raqes .iter() - .map(|r| eligible_deployments_for(r, deployments)) + .map(|r| eligible_deployments_for(r, deployments, facts)) .collect(); if choices.iter().any(|c| c.is_empty()) { @@ -157,8 +167,8 @@ mod tests { let table = vec![cost("cms-heap-topk-fastpath-vector2d")]; let deployments = build_all_candidates(&raqes, &table); - assert!(unservable(&raqes, &deployments).is_empty()); - let mappings = brute_force(&raqes, &deployments); + assert!(unservable(&raqes, &deployments, &facts(1, 1)).is_empty()); + let mappings = brute_force(&raqes, &deployments, &facts(1, 1)); assert!( mappings.iter().any(|m| m[0] == m[1]), "expected at least one mapping where both identical RAQEs land on the same deployment" @@ -171,17 +181,20 @@ mod tests { let table: Vec = Vec::new(); // no configs at all let deployments = build_all_candidates(&raqes, &table); - assert_eq!(unservable(&raqes, &deployments), vec!["a".to_string()]); - assert!(brute_force(&raqes, &deployments).is_empty()); + assert_eq!( + unservable(&raqes, &deployments, &facts(1, 1)), + vec!["a".to_string()] + ); + assert!(brute_force(&raqes, &deployments, &facts(1, 1)).is_empty()); } #[test] fn streaming_visits_the_same_mappings_as_eager_enumeration() { let raqes = vec![raqe("a", 60_000, 3_600_000), raqe("b", 60_000, 3_600_000)]; let deployments = build_all_candidates(&raqes, &[cost("cms-heap-topk-fastpath-vector2d")]); - let eager = brute_force(&raqes, &deployments); + let eager = brute_force(&raqes, &deployments, &facts(1, 1)); let mut streamed = Vec::new(); - let count = for_each_mapping(&raqes, &deployments, |mapping| { + let count = for_each_mapping(&raqes, &deployments, &facts(1, 1), |mapping| { streamed.push(mapping.clone()) }); assert_eq!(count as usize, eager.len()); @@ -193,7 +206,7 @@ mod tests { let raqes = vec![raqe("a", 60_000, 3_600_000), raqe("b", 60_000, 3_600_000)]; let deployments = build_all_candidates(&raqes, &[cost("cms-heap-topk-fastpath-vector2d")]); let mut sampled = Vec::new(); - let result = for_each_mapping_while(&raqes, &deployments, |mapping| { + let result = for_each_mapping_while(&raqes, &deployments, &facts(1, 1), |mapping| { sampled.push(mapping.clone()); sampled.len() < 2 }); diff --git a/rqe-optimizer/src/lib.rs b/rqe-optimizer/src/lib.rs index ec76fbdd..f172eed8 100644 --- a/rqe-optimizer/src/lib.rs +++ b/rqe-optimizer/src/lib.rs @@ -70,7 +70,7 @@ impl Capability { Capability::Min => &["exact-min"], Capability::Max => &["exact-max"], Capability::RateOrIncrease => &["exact-increase"], - Capability::Quantile => &["kll-percall", "dd"], + Capability::Quantile => &["kll-percall", "dd", "hydra-kll"], // univmon-cardinality is registered under KeyedCardinality in // sketch-bench, but its ground truth (`KeyedCardinalityGT`) is // "count of keys with a nonzero total" -- plain distinct-key @@ -111,6 +111,54 @@ pub const DEPLOYABLE_FAMILIES: &[&str] = &[ "cms-heap-topk-fastpath-vector2d", // CountMinSketchWithHeap ]; +/// The variant every [`FamilyProperties::needs_delta_set_key_tracker`] +/// deployment pays for alongside its sketch. Never a candidate on its own. +pub const KEY_TRACKER_FAMILY: &str = "exact-delta-set"; + +/// What a variant's algorithm can do, as opposed to what it measured. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct FamilyProperties { + /// Windows fold into one. A family without it only serves `L == x`. + pub mergeable_across_windows: bool, + /// One sketch of fixed size holds every group: the design doc's Shared + + /// Fixed cell. Memory, merge and storage don't scale with `card(G)`. + /// Exact accumulators are Shared + PerKey and `false` here, because the + /// export prices them per group. + pub one_fixed_size_sketch_for_all_groups: bool, + /// The sketch can't list its groups, so a DeltaSet records which keys + /// each window saw, for the query to probe. + pub needs_delta_set_key_tracker: bool, +} + +/// Panics on a variant with no entry: guessing would misprice it. +pub fn family_properties(variant: &str) -> FamilyProperties { + let one_sketch_per_group = FamilyProperties { + mergeable_across_windows: true, + one_fixed_size_sketch_for_all_groups: false, + needs_delta_set_key_tracker: false, + }; + match variant { + "exact-sum" + | "exact-min" + | "exact-max" + | "exact-increase" + | KEY_TRACKER_FAMILY + | "kll-percall" + | "dd" + | "hll" + | "univmon-cardinality" + | "cms-heap-topk-fastpath-vector2d" + | "countsketch-heap-topk-fastpath-vector2d" + | "univmon-topk" => one_sketch_per_group, + "hydra-kll" => FamilyProperties { + mergeable_across_windows: true, + one_fixed_size_sketch_for_all_groups: true, + needs_delta_set_key_tracker: true, + }, + _ => panic!("{variant} has no FamilyProperties; add it to family_properties"), + } +} + /// Facts about one metric, given as input. RAQEs and deployments on the same /// metric read the same samples. #[derive(Debug, Clone)] @@ -307,16 +355,24 @@ pub struct Deployment { pub capability: Capability, pub metric: String, pub spatial_filter: String, - /// `G`: one instance per group, per window. + /// `G`: one instance per group, per window, unless the family keeps one + /// for all groups. pub grouping_labels: LabelSet, pub config: AtomicCostEntry, /// `x`: materialized sketch-window size. pub window_ms: Millis, /// `y`: materialized sketch slide. pub slide_ms: Millis, + /// The [`KEY_TRACKER_FAMILY`] row, for a family that + /// [needs one](FamilyProperties::needs_delta_set_key_tracker). + pub key_tracker: Option, } impl Deployment { + pub fn properties(&self) -> FamilyProperties { + family_properties(&self.config.sketch) + } + /// Number of active overlapping instances receiving each item. pub fn active_instance_count(&self) -> Option { if self.slide_ms == 0 || !self.window_ms.is_multiple_of(self.slide_ms) { @@ -415,6 +471,7 @@ pub(crate) mod test_support { }, window_ms, slide_ms, + key_tracker: None, } } } @@ -486,6 +543,16 @@ mod tests { } } + #[test] + fn every_family_has_properties() { + for capability in Capability::ALL { + for family in capability.families() { + family_properties(family); + } + } + family_properties(KEY_TRACKER_FAMILY); + } + #[test] fn closed_instances_lie_wholly_inside_the_lookback() { // 10-minute windows every minute, over an hour: starts in minutes diff --git a/rqe-optimizer/src/milp.rs b/rqe-optimizer/src/milp.rs index 5ebacdfe..33f4545a 100644 --- a/rqe-optimizer/src/milp.rs +++ b/rqe-optimizer/src/milp.rs @@ -63,7 +63,7 @@ pub fn minimize( ) -> Result { let eligible: Vec> = raqes .iter() - .map(|raqe| eligible_deployments_for(raqe, deployments)) + .map(|raqe| eligible_deployments_for(raqe, deployments, facts)) .collect(); assert!( eligible.iter().all(|choices| !choices.is_empty()), @@ -230,7 +230,7 @@ mod tests { objective: Objective, ) { let facts = facts(1, 1); - let best = brute_force(raqes, deployments) + let best = brute_force(raqes, deployments, &facts) .iter() .map(|mapping| objective.value(&score(raqes, deployments, mapping, &facts))) .min_by(f64::total_cmp) diff --git a/scripts/export_rqe_optimizer_costs.sh b/scripts/export_rqe_optimizer_costs.sh index 1d4e04a0..4164aec2 100755 --- a/scripts/export_rqe_optimizer_costs.sh +++ b/scripts/export_rqe_optimizer_costs.sh @@ -149,9 +149,10 @@ point_spec exact-delta-set exact "" key-set configs/datagen/hydra_columns.yaml i # HydraKLL rows are whole-sketch: one instance serves every group, and memory # and merge cost are not divided per group (`subpop-rank-error` reports no -# `groups_per_instance`). The optimizer prices a row as card(G) x per-instance -# cost, so these rows must not become candidates until #142 lands. Cost and -# accuracy hold at this spec's group count only. The spec gives each label its +# `groups_per_instance`). The optimizer prices them as one shared fixed-size +# sketch, not card(G) instances, and plans them only up to the `subpopulations` +# they were measured at. The keys fan out to every label subset, unlike the +# single-key sketch the optimizer models (#165). The spec gives each label its # own alphabet, so rank error is the sketch's, not label collisions' (#74). echo "==> Quantiles: KLL, DDSketch, HydraKLL" >&2 for k in "${KLL_KS[@]}"; do