Skip to content

Commit b1ac24b

Browse files
fix(planner): bump rqe-optimizer to S8 and split Sum/Count/TopK capabilities (#801)
* fix(planner): bump rqe-optimizer to S8 and split Sum/Count/TopK capabilities sketch-bench split SumOrCount into Sum and Count and TopK into TopKByValue and TopKByCount, so the two halves of an avg rewrite, and value- vs count-ranked top-k, no longer share one deployment. A topk with unknown weighting is now an error. solve_milp returns S8's MilpSolution (active deployments with retained count, per-Raqe deployment and merged count); the CLI prints both. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix(planner): reuse rqe-optimizer's unservable check once per item - The unservable pre-check calls rqe_optimizer::enumerate::unservable instead of re-implementing it, and checks one Raqe per item, since an item's occurrences are identical. - TopkWeightingUnknown no longer implies a bare topk is invalid. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
1 parent acebd27 commit b1ac24b

3 files changed

Lines changed: 130 additions & 53 deletions

File tree

‎Cargo.lock‎

Lines changed: 3 additions & 3 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎asap-planner-rs/src/bin/optimizer_cli.rs‎

Lines changed: 14 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -188,33 +188,35 @@ fn run_milp(args: &Args, config: &ControllerConfig) -> anyhow::Result<()> {
188188
tracing::debug!(?objective, cost_rows = costs.len(), "milp: inputs loaded");
189189

190190
let workload = build_milp_workload(config, &facts, args.data_ingestion_interval_ms)?;
191-
let (deployments, solution) = solve_milp(&workload, &facts, &costs, objective)?;
192-
193-
let mut active: Vec<usize> = solution.mapping.clone();
194-
active.sort();
195-
active.dedup();
196-
println!("=== Deployments: {} ===", active.len());
197-
for &d in &active {
198-
let dep = &deployments[d];
191+
let solution = solve_milp(&workload, &facts, &costs, objective)?;
192+
193+
println!("=== Deployments: {} ===", solution.deployments.len());
194+
for (d, planned) in solution.deployments.iter().enumerate() {
195+
let dep = &planned.deployment;
199196
println!(
200-
" [{d}] {:?} {} config={} metric={} grouping={:?} window={}ms slide={}ms",
197+
" [{d}] {:?} {} config={} metric={} grouping={:?} window={}ms slide={}ms retained={} key_tracker={}",
201198
dep.capability,
202199
dep.config.sketch,
203200
dep.config.sketch_config,
204201
dep.metric,
205202
dep.grouping_labels,
206203
dep.window_ms,
207204
dep.slide_ms,
205+
planned.retained_instance_count,
206+
dep.key_tracker.is_some(),
208207
);
209208
}
210209
println!("\n=== Raqes: {} ===", workload.raqes.len());
211-
for ((raqe, &d), latency_ms) in workload
210+
for ((raqe, planned), latency_ms) in workload
212211
.raqes
213212
.iter()
214-
.zip(&solution.mapping)
213+
.zip(&solution.raqes)
215214
.zip(&solution.plan_cost.query_latency_ms)
216215
{
217-
println!(" {} -> [{d}] latency={latency_ms:.3e}ms", raqe.id);
216+
println!(
217+
" {} -> [{}] n={} latency={latency_ms:.3e}ms",
218+
raqe.id, planned.deployment, planned.merged_instance_count
219+
);
218220
}
219221
let cost = &solution.plan_cost;
220222
println!(

‎asap-planner-rs/src/optimizer/milp.rs‎

Lines changed: 113 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -2,11 +2,11 @@
22
//! workload config into `Raqe`s and solves for the cheapest deployments.
33
44
use promql_utilities::query_logics::enums::Statistic;
5-
use rqe_optimizer::candidates::{build_all_candidates, eligible_deployments_for};
5+
use rqe_optimizer::candidates::build_all_candidates;
6+
use rqe_optimizer::enumerate::unservable;
67
use rqe_optimizer::milp::{minimize, MilpSolution, Objective};
78
use rqe_optimizer::{
8-
validate_facts, AccuracyDirection, AtomicCostEntry, Capability, Deployment, LabelSet, Raqe,
9-
WorkloadFacts,
9+
validate_facts, AccuracyDirection, AtomicCostEntry, Capability, LabelSet, Raqe, WorkloadFacts,
1010
};
1111
use thiserror::Error;
1212

@@ -37,34 +37,38 @@ pub enum MilpError {
3737
query: String,
3838
statistics: Vec<Statistic>,
3939
},
40+
#[error("query {query:?}: topk ranking (by value or by sample count) is unknown")]
41+
TopkWeightingUnknown { query: String },
4042
#[error("no eligible deployment for raqes: {0:?}")]
4143
Unservable(Vec<String>),
4244
#[error("MILP solve failed: {0}")]
4345
Solver(String),
4446
}
4547

46-
/// Candidate deployments and the MILP's choice among them. Only families in
47-
/// sketch-bench's `DEPLOYABLE_FAMILIES` are candidates.
48+
/// The cheapest plan. Only families in sketch-bench's `DEPLOYABLE_FAMILIES`
49+
/// are candidates.
4850
pub fn solve_milp(
4951
workload: &MilpWorkload,
5052
facts: &WorkloadFacts,
5153
costs: &[AtomicCostEntry],
5254
objective: Objective,
53-
) -> Result<(Vec<Deployment>, MilpSolution), MilpError> {
55+
) -> Result<MilpSolution, MilpError> {
5456
let deployments = build_all_candidates(&workload.raqes, costs, facts, false);
5557
tracing::debug!(
5658
candidates = deployments.len(),
5759
cost_rows = costs.len(),
5860
"milp: built candidates"
5961
);
60-
let mut missing = Vec::new();
61-
for raqe in &workload.raqes {
62-
let eligible = eligible_deployments_for(raqe, &deployments).len();
63-
tracing::debug!(raqe = %raqe.id, eligible, "milp: eligible deployments");
64-
if eligible == 0 {
65-
missing.push(raqe.id.clone());
66-
}
67-
}
62+
// Occurrences of one item are identical Raqes (and contiguous), so
63+
// checking the first of each is enough.
64+
let one_per_item: Vec<Raqe> = workload
65+
.raqes
66+
.iter()
67+
.enumerate()
68+
.filter(|&(i, _)| i == 0 || workload.raqe_items[i] != workload.raqe_items[i - 1])
69+
.map(|(_, raqe)| raqe.clone())
70+
.collect();
71+
let missing = unservable(&one_per_item, &deployments, facts);
6872
if !missing.is_empty() {
6973
return Err(MilpError::Unservable(missing));
7074
}
@@ -74,10 +78,10 @@ pub fn solve_milp(
7478
objective = objective.value(&solution.plan_cost),
7579
cpu_secs_per_sec = solution.plan_cost.cpu_secs_per_sec(),
7680
memory_bytes = solution.plan_cost.memory_bytes(),
77-
mapping = ?solution.mapping,
81+
deployments = solution.deployments.len(),
7882
"milp: solved"
7983
);
80-
Ok((deployments, solution))
84+
Ok(solution)
8185
}
8286

8387
/// The MILP's view of a workload.
@@ -166,7 +170,9 @@ fn item_to_raqe(item: &OptimizerItem) -> Result<Raqe, MilpError> {
166170
statistics: req.statistics.clone(),
167171
});
168172
};
169-
let capability = capability(*statistic);
173+
let Some(capability) = capability(*statistic, req.topk_count_events) else {
174+
return Err(MilpError::TopkWeightingUnknown { query });
175+
};
170176
if !(0.0..=1.0).contains(&item.accuracy_sla) {
171177
return Err(MilpError::AccuracySlaOutOfRange {
172178
query,
@@ -179,7 +185,7 @@ fn item_to_raqe(item: &OptimizerItem) -> Result<Raqe, MilpError> {
179185
// TopK keeps one heap per `topk by` bucket; its `grouping_labels` is the
180186
// output label set (every label), which would cost one heap per series.
181187
let grouping_labels: LabelSet = match capability {
182-
Capability::TopK => req
188+
Capability::TopKByValue | Capability::TopKByCount => req
183189
.topk_by_labels
184190
.as_ref()
185191
.map(|labels| labels.labels.iter().cloned().collect())
@@ -202,16 +208,23 @@ fn item_to_raqe(item: &OptimizerItem) -> Result<Raqe, MilpError> {
202208
})
203209
}
204210

205-
fn capability(statistic: Statistic) -> Capability {
206-
match statistic {
207-
Statistic::Sum | Statistic::Count => Capability::SumOrCount,
211+
/// `None` for a topk whose weighting is unknown: value-ranked
212+
/// (`sum_over_time`) and count-ranked (`count_over_time`) top-k need separate
213+
/// heaps.
214+
fn capability(statistic: Statistic, topk_count_events: Option<bool>) -> Option<Capability> {
215+
Some(match statistic {
216+
Statistic::Sum => Capability::Sum,
217+
Statistic::Count => Capability::Count,
208218
Statistic::Rate | Statistic::Increase => Capability::RateOrIncrease,
209219
Statistic::Min => Capability::Min,
210220
Statistic::Max => Capability::Max,
211221
Statistic::Quantile => Capability::Quantile,
212222
Statistic::Cardinality => Capability::Cardinality,
213-
Statistic::Topk => Capability::TopK,
214-
}
223+
Statistic::Topk => match topk_count_events? {
224+
true => Capability::TopKByCount,
225+
false => Capability::TopKByValue,
226+
},
227+
})
215228
}
216229

217230
/// `accuracy_sla` is required accuracy: error metrics must stay within
@@ -222,13 +235,14 @@ fn accuracy_target(
222235
) -> (&'static str, f64, AccuracyDirection) {
223236
let max_error = 1.0 - accuracy_sla + SLA_EPSILON;
224237
match capability {
225-
Capability::SumOrCount
238+
Capability::Sum
239+
| Capability::Count
226240
| Capability::Min
227241
| Capability::Max
228242
| Capability::RateOrIncrease
229243
| Capability::Cardinality => (RELATIVE_ERROR, max_error, AccuracyDirection::LowerIsBetter),
230244
Capability::Quantile => (MAX_RANK_ERROR, max_error, AccuracyDirection::LowerIsBetter),
231-
Capability::TopK => (
245+
Capability::TopKByValue | Capability::TopKByCount => (
232246
PRECISION_AT_K,
233247
accuracy_sla - SLA_EPSILON,
234248
AccuracyDirection::HigherIsBetter,
@@ -352,10 +366,10 @@ metrics:
352366
}
353367

354368
#[test]
355-
fn sum_maps_to_sum_or_count_with_relative_error_ceiling() {
369+
fn sum_maps_to_sum_with_relative_error_ceiling() {
356370
let w = workload(&group("sum by (job) (http_requests_total)", 0.99));
357371
let r = &w.raqes[0];
358-
assert_eq!(r.capability, Capability::SumOrCount);
372+
assert_eq!(r.capability, Capability::Sum);
359373
assert_eq!(r.grouping_labels, labels(&["job"]));
360374
assert_eq!(r.accuracy_metric, RELATIVE_ERROR);
361375
assert_eq!(r.accuracy_direction, AccuracyDirection::LowerIsBetter);
@@ -364,6 +378,34 @@ metrics:
364378
assert_eq!(r.latency_sla_ms, None);
365379
}
366380

381+
#[test]
382+
fn count_and_topk_weighting_map_to_their_own_capabilities() {
383+
let cap = |query| workload(&group(query, 0.99)).raqes[0].capability;
384+
assert_eq!(
385+
cap("sum by (job) (count_over_time(http_requests_total[1m]))"),
386+
Capability::Count
387+
);
388+
assert_eq!(
389+
cap("topk(5, sum_over_time(http_requests_total[1m]))"),
390+
Capability::TopKByValue
391+
);
392+
assert_eq!(
393+
cap("topk(5, count_over_time(http_requests_total[1m]))"),
394+
Capability::TopKByCount
395+
);
396+
}
397+
398+
#[test]
399+
fn topk_with_unknown_weighting_is_an_error() {
400+
let w = workload(&group("topk(5, http_requests_total)", 0.99));
401+
let mut item = w.items[0].clone();
402+
item.requirements.topk_count_events = None;
403+
assert!(matches!(
404+
item_to_raqe(&item),
405+
Err(MilpError::TopkWeightingUnknown { .. })
406+
));
407+
}
408+
367409
#[test]
368410
fn quantile_uses_max_rank_error() {
369411
let w = workload(&group(
@@ -380,7 +422,7 @@ metrics:
380422
fn topk_requires_precision_and_groups_by_its_buckets() {
381423
let w = workload(&group("topk by (job) (5, http_requests_total)", 0.9));
382424
let r = &w.raqes[0];
383-
assert_eq!(r.capability, Capability::TopK);
425+
assert_eq!(r.capability, Capability::TopKByValue);
384426
assert_eq!(r.accuracy_metric, PRECISION_AT_K);
385427
assert_eq!(r.accuracy_direction, AccuracyDirection::HigherIsBetter);
386428
assert!((r.accuracy_sla - 0.9).abs() < 1e-6);
@@ -427,10 +469,13 @@ metrics:
427469
let config = config(&(group(query, 0.99) + &group(query, 0.99)));
428470
let facts = facts(&config);
429471
let w = build_milp_workload(&config, &facts, SCRAPE_MS).unwrap();
430-
let (deployments, solution) =
431-
solve_milp(&w, &facts, &costs(), Objective::default()).unwrap();
432-
assert_eq!(solution.mapping[0], solution.mapping[1]);
433-
assert_eq!(deployments[solution.mapping[0]].config.sketch, "exact-sum");
472+
let solution = solve_milp(&w, &facts, &costs(), Objective::default()).unwrap();
473+
assert_eq!(solution.deployments.len(), 1);
474+
assert_eq!(solution.raqes[0].deployment, solution.raqes[1].deployment);
475+
assert_eq!(
476+
solution.deployments[0].deployment.config.sketch,
477+
"exact-sum"
478+
);
434479
}
435480

436481
#[test]
@@ -446,6 +491,17 @@ metrics:
446491
assert!(matches!(err, MilpError::Unservable(ids) if ids.len() == 1));
447492
}
448493

494+
#[test]
495+
fn unservable_query_is_reported_once_per_item() {
496+
let quantile = group("quantile_over_time(0.99, http_requests_total[5m])", 0.999);
497+
let config = config(&(quantile.clone() + &quantile));
498+
let facts = facts(&config);
499+
let w = build_milp_workload(&config, &facts, SCRAPE_MS).unwrap();
500+
assert_eq!(w.raqes.len(), 2);
501+
let err = solve_milp(&w, &facts, &costs(), Objective::default()).unwrap_err();
502+
assert!(matches!(err, MilpError::Unservable(ids) if ids == [w.raqes[0].id.clone()]));
503+
}
504+
449505
#[test]
450506
fn solve_picks_a_deployable_family_per_capability() {
451507
let groups = group("sum by (job) (http_requests_total)", 0.99)
@@ -454,16 +510,35 @@ metrics:
454510
let config = config(&groups);
455511
let facts = facts(&config);
456512
let w = build_milp_workload(&config, &facts, SCRAPE_MS).unwrap();
457-
let (deployments, solution) =
458-
solve_milp(&w, &facts, &costs(), Objective::default()).unwrap();
513+
let solution = solve_milp(&w, &facts, &costs(), Objective::default()).unwrap();
459514
let chosen: BTreeMap<Capability, &str> = w
460515
.raqes
461516
.iter()
462-
.zip(&solution.mapping)
463-
.map(|(r, &d)| (r.capability, deployments[d].config.sketch.as_str()))
517+
.zip(&solution.raqes)
518+
.map(|(r, p)| {
519+
let sketch = &solution.deployments[p.deployment].deployment.config.sketch;
520+
(r.capability, sketch.as_str())
521+
})
464522
.collect();
465-
assert_eq!(chosen[&Capability::SumOrCount], "exact-sum");
523+
assert_eq!(chosen[&Capability::Sum], "exact-sum");
466524
assert_eq!(chosen[&Capability::Quantile], "kll-percall");
467-
assert_eq!(chosen[&Capability::TopK], "cms-heap-topk-fastpath-vector2d");
525+
assert_eq!(
526+
chosen[&Capability::TopKByValue],
527+
"cms-heap-topk-fastpath-vector2d"
528+
);
529+
}
530+
531+
#[test]
532+
fn avg_sum_and_count_get_separate_deployments() {
533+
// One exact-sum accumulator can't answer both halves of the avg
534+
// rewrite; sharing would undercount ingest and memory.
535+
let config = config(&group("avg by (job) (http_requests_total)", 0.99));
536+
let facts = facts(&config);
537+
let w = build_milp_workload(&config, &facts, SCRAPE_MS).unwrap();
538+
let caps: Vec<_> = w.raqes.iter().map(|r| r.capability).collect();
539+
assert_eq!(caps.len(), 2);
540+
assert!(caps.contains(&Capability::Sum) && caps.contains(&Capability::Count));
541+
let solution = solve_milp(&w, &facts, &costs(), Objective::default()).unwrap();
542+
assert_eq!(solution.deployments.len(), 2);
468543
}
469544
}

0 commit comments

Comments
 (0)