diff --git a/crates/devtools/src/bin/stage_pipeline.rs b/crates/devtools/src/bin/stage_pipeline.rs index 37c2de8b4..90d637f51 100644 --- a/crates/devtools/src/bin/stage_pipeline.rs +++ b/crates/devtools/src/bin/stage_pipeline.rs @@ -404,7 +404,8 @@ fn run(args: Vec) -> Result<(), Failure> { let mut cost_inputs_path: Option = None; let mut enforce_accuracy_guarantees = false; let mut exhaustive = false; - // `--milp-time-limit-s` limits the MILP search. + // `--milp-time-limit-s` bounds the facade's planning budget (Pass 2 and + // its rounds); Stage 3's MILP always runs to optimality (#1185). let mut milp_time_limit_ms: Option = None; let mut max_combinations_given = false; let mut write_memo = true; diff --git a/crates/devtools/tests/corpus_snapshots.rs b/crates/devtools/tests/corpus_snapshots.rs index 10ef56a76..119ba4dfb 100644 --- a/crates/devtools/tests/corpus_snapshots.rs +++ b/crates/devtools/tests/corpus_snapshots.rs @@ -111,8 +111,9 @@ fn hit_time_limit(document: &Value) -> bool { report["pass1_ms"].as_f64().unwrap_or(0.0) + rounds >= limit } -/// The selected plan's search status (`optimal` or `deadline_reached`), or -/// null without a plan. +/// The selected plan's search status (`optimal`; `deadline_reached` in +/// documents from before the MILP ran without a time limit), or null +/// without a plan. fn search_status(document: &Value) -> &Value { &document["stage3_plan_selection"]["search_report"]["status"] } diff --git a/crates/devtools/tests/corpus_snapshots/o11y_bench/q05.json b/crates/devtools/tests/corpus_snapshots/o11y_bench/q05.json index 1e72f4c5c..b986b993f 100644 --- a/crates/devtools/tests/corpus_snapshots/o11y_bench/q05.json +++ b/crates/devtools/tests/corpus_snapshots/o11y_bench/q05.json @@ -2,7 +2,7 @@ "accuracy_guarantees": "enforced", "materialized_views": [], "materialized_views_error": "raw-data window of node 1 is unsupported: a subquery or a TimeShift above the range", - "proven_optimal": false, + "proven_optimal": true, "query": "max_over_time((sum(rate(http_requests_total{job=\"order-service\",status=~\"5..\"}[5m])) / sum(rate(http_requests_total{job=\"order-service\"}[5m])))[6h:1m])", "selected": { "accepted": [ diff --git a/crates/devtools/tests/corpus_snapshots/o11y_bench/q19.json b/crates/devtools/tests/corpus_snapshots/o11y_bench/q19.json index 5c0c496f3..2d5150c66 100644 --- a/crates/devtools/tests/corpus_snapshots/o11y_bench/q19.json +++ b/crates/devtools/tests/corpus_snapshots/o11y_bench/q19.json @@ -6242,7 +6242,7 @@ "stage2_node_id": 7 } ], - "proven_optimal": false, + "proven_optimal": true, "query": "sum(increase(http_requests_total{job=\"payment-service\",status=~\"5..\"}[1h])) / sum(increase(http_requests_total{job=\"payment-service\"}[1h]))", "selected": { "accepted": [ diff --git a/crates/devtools/tests/corpus_snapshots/o11y_bench/q25.json b/crates/devtools/tests/corpus_snapshots/o11y_bench/q25.json index 61dd6c49f..9f4988b12 100644 --- a/crates/devtools/tests/corpus_snapshots/o11y_bench/q25.json +++ b/crates/devtools/tests/corpus_snapshots/o11y_bench/q25.json @@ -4562,7 +4562,7 @@ "stage2_node_id": 7 } ], - "proven_optimal": false, + "proven_optimal": true, "query": "sum(rate(http_requests_total{status=~\"5..\",job=~\"user-service|order-service|payment-service\"}[1h])) / sum(rate(http_requests_total{job=~\"user-service|order-service|payment-service\"}[1h]))", "selected": { "accepted": [ diff --git a/crates/devtools/tests/sketch_plan_snapshots.rs b/crates/devtools/tests/sketch_plan_snapshots.rs index c23f7dcd5..c9047256c 100644 --- a/crates/devtools/tests/sketch_plan_snapshots.rs +++ b/crates/devtools/tests/sketch_plan_snapshots.rs @@ -23,11 +23,7 @@ //! select a different plan on a slower machine. Only a search that proves its //! plan optimal is independent of speed, so every workload must end with //! status `optimal` and no tumbling-window width left unsearched; any other ending -//! fails instead of being recorded. A workload with a -//! `deadline_reached_reason` in the manifest may instead end with status -//! `deadline_reached` (no tumbling-window width left unsearched); its snapshot records -//! that status, so the plan, cost and status must still match. Each -//! kept workload gave the same plan in three runs and on one core. +//! fails instead of being recorded. Each kept workload gave the same plan in three runs and on one core. //! Workloads whose search stops at a budget (the raw-retained memo examples, //! full random-mixed workloads, three or more quantile queries over 10,000 //! series, the `or` and `COUNT(DISTINCT)` workloads, single quantiles under @@ -146,12 +142,7 @@ fn sketch_plans_match_snapshots() { )); continue; } - // A `deadline_reached_reason` accepts a search that ended at its - // deadline after every tumbling-window width; plan and cost are still compared. - let deadline_accepted = entry.get("deadline_reached_reason").is_some() - && search_status(&document) == "deadline_reached" - && document["planning_report"]["tumbling_window_widths_pending"] != json!(true); - if !proven_optimal(&document) && !deadline_accepted { + if !proven_optimal(&document) { failures.push(format!( "{id}: the plan is not proven optimal (status {}, tumbling-window widths \ pending {}), so a round's budget may have chosen it; drop the \ diff --git a/crates/devtools/tests/sketch_plan_snapshots/manifest.json b/crates/devtools/tests/sketch_plan_snapshots/manifest.json index af6caf6ac..c7e77493d 100644 --- a/crates/devtools/tests/sketch_plan_snapshots/manifest.json +++ b/crates/devtools/tests/sketch_plan_snapshots/manifest.json @@ -18,8 +18,7 @@ "id": "shared-quantiles-1h-mvp", "covers": "two queries share one DDSketch pane chain (q0.5 and q0.9 over 1 h)", "workload": "tools/dag-viewer/memo-examples/promql/rangeheavy-t5-q05-q09-1h.json", - "deployment": "mvp", - "deadline_reached_reason": "the MILP stalls numerically (lower bound about 8e-7 relative below the cost) and reports deadline_reached; the plan and cost are the same as when it was proven optimal" + "deployment": "mvp" }, { "id": "nested-windows-15m-1h-mvp", @@ -37,8 +36,7 @@ "id": "quantile-single-mvp", "covers": "single quantile with an accuracy target selects a KLL pane chain", "workload": "crates/devtools/tests/sketch_plan_snapshots/workloads/quantile-single.json", - "deployment": "mvp", - "deadline_reached_reason": "the MILP stalls numerically (lower bound about 8e-7 relative below the cost) and reports deadline_reached; the plan and cost are the same as when it was proven optimal" + "deployment": "mvp" }, { "id": "quantile-grouped-mvp", diff --git a/crates/devtools/tests/sketch_plan_snapshots/quantile-single-mvp.json b/crates/devtools/tests/sketch_plan_snapshots/quantile-single-mvp.json index b1a914db3..4f357790a 100644 --- a/crates/devtools/tests/sketch_plan_snapshots/quantile-single-mvp.json +++ b/crates/devtools/tests/sketch_plan_snapshots/quantile-single-mvp.json @@ -1142,11 +1142,11 @@ "stage2_node_id": 1 } ], - "proven_optimal": false, + "proven_optimal": true, "queries": [ "quantile_over_time(0.99, data[5m])" ], - "search_status": "deadline_reached", + "search_status": "optimal", "selected": { "accepted": [ { diff --git a/crates/devtools/tests/sketch_plan_snapshots/shared-quantiles-1h-mvp.json b/crates/devtools/tests/sketch_plan_snapshots/shared-quantiles-1h-mvp.json index 0c2de0f45..227c9d39c 100644 --- a/crates/devtools/tests/sketch_plan_snapshots/shared-quantiles-1h-mvp.json +++ b/crates/devtools/tests/sketch_plan_snapshots/shared-quantiles-1h-mvp.json @@ -2282,12 +2282,12 @@ "stage2_node_id": 1 } ], - "proven_optimal": false, + "proven_optimal": true, "queries": [ "quantile_over_time(0.5, data[1h])", "quantile_over_time(0.9, data[1h])" ], - "search_status": "deadline_reached", + "search_status": "optimal", "selected": { "accepted": [ { diff --git a/crates/devtools/tests/stage_pipeline.rs b/crates/devtools/tests/stage_pipeline.rs index a9bd69ad3..fdc39e876 100644 --- a/crates/devtools/tests/stage_pipeline.rs +++ b/crates/devtools/tests/stage_pipeline.rs @@ -2605,13 +2605,10 @@ fn milp_search_objective_equals_plan_cost_on_example1_and_rangeheavy() { let selection = &document["stage3_plan_selection"]; assert!(selects_a_plan(selection), "{args:?}: {selection}"); let report = &selection["search_report"]; - // Optimal or not by the deadline, the selected plan's `plan_cost` - // equals the MILP's objective: the search errors on a difference, - // and no term is encoded inexactly here. - assert!( - report["status"] == "optimal" || report["status"] == "deadline_reached", - "{args:?}: {report}" - ); + // The MILP runs to proven optimality, and the selected plan's + // `plan_cost` equals its objective: the search errors on a + // difference, and no term is encoded inexactly here. + assert_eq!(report["status"], "optimal", "{args:?}: {report}"); assert!( report.get("inexact_objective").is_none(), "{args:?}: {report}" diff --git a/crates/plan-selection/src/groups.rs b/crates/plan-selection/src/groups.rs index 659ce3b14..1421f5d2a 100644 --- a/crates/plan-selection/src/groups.rs +++ b/crates/plan-selection/src/groups.rs @@ -33,8 +33,8 @@ //! grow with the range-evaluation span of the queries reaching it //! ([`asap_cost::tumbling_window_retention_ms`], [`asap_cost::retained_tumbling_windows`]). //! -//! Every combination of these options is costed (at most -//! [`MAX_CONTEXTS`]); a combination `asap-cost` cannot cost is skipped +//! Every combination of these options is costed; a combination +//! `asap-cost` cannot cost is skipped //! (no plan can use it). When the numbers agree, the usage is //! plan-independent; otherwise [`CandidateData::usage_depends_on`] says which //! options changed it, and [`CandidateData::usage`] holds the first costed @@ -125,11 +125,6 @@ use crate::SelectionModels; use asap_types::stages::{PhysicalCandidateId, ReaderConstraint}; -/// The most contexts (combinations of producer outputs, scan windows and -/// pane retentions) one candidate is costed under; beyond it the candidate's usage -/// is marked [`UsageContext::TooManyContexts`]. -pub(crate) const MAX_CONTEXTS: usize = 256; - /// What the MILP is built from (module documentation). #[derive(Debug, Clone)] pub(crate) struct MilpInputs { @@ -233,11 +228,6 @@ impl Default for CandidateData { } } -/// The most producer choices [`CandidateData::usage_per_producers`] lists; beyond -/// it the candidate's usage is marked [`UsageContext::TooManyContexts`] and -/// its first costed usage stands for every choice. -pub(crate) const MAX_PRODUCER_CHOICES: usize = 256; - /// One candidate's usage, from [`asap_cost::node_usage`] (module /// documentation). #[derive(Debug, Clone, Copy, PartialEq)] @@ -300,8 +290,6 @@ pub(crate) enum UsageContext { /// window from a source without `rows_held`); the usage is the first /// costed one's. UncostableContext, - /// More than [`MAX_CONTEXTS`] combinations: not all were costed. - TooManyContexts, } /// A candidate dominance deleted. @@ -330,9 +318,8 @@ pub(crate) enum UnusableReason { /// Some output's guarantee misses the target /// ([`RejectionCause::AccuracyTargetMissed`](crate::report::RejectionCause::AccuracyTargetMissed)). AccuracyTargetMissed, - /// Not checked, so not usable: more than - /// [`accuracy_positions::MAX_COMPOSITIONS`] outputs to build, or a node - /// that does not rebuild over its producers. + /// Not checked, so not usable: a node that does not rebuild over its + /// producers. NotChecked, } @@ -1129,7 +1116,7 @@ fn window_read_through(path: &[&PhysicalASAPDAGNode]) -> ScanWindow { /// windows (one per path through time shifts) and, per execution context, /// the unions of the window kinds its readers there read (one scan run read /// by several readers reads the union, `PhysicalASAPDAG::scan_window`): -/// every union of up to [`MAX_WINDOW_KINDS`] kinds, else the union of all. +/// every union of the kinds. /// A read as an output or as a shareable candidate's own input is /// `Unbounded`. Distinct; `Unbounded` alone when nothing reads it. fn scan_window_options( @@ -1239,13 +1226,7 @@ fn window_contains(outer: ScanWindow, inner: ScanWindow) -> bool { } } -/// The most window kinds of one scan run whose every union is costed (and -/// encoded exactly, [`crate::milp::cost_terms`]); beyond it only the union -/// of all. -pub(crate) const MAX_WINDOW_KINDS: usize = 4; - -/// Adds to `windows` every union of a nonempty subset of `kinds` (only the -/// union of all beyond [`MAX_WINDOW_KINDS`]). +/// Adds to `windows` every union of a nonempty subset of `kinds`. fn add_unions(kinds: &[ScanWindow], windows: &mut Vec) { use crate::milp::cost_terms::scan_window_read_at as window_union; let mut push = |window: ScanWindow| { @@ -1253,10 +1234,13 @@ fn add_unions(kinds: &[ScanWindow], windows: &mut Vec) { windows.push(window); } }; - if kinds.len() > MAX_WINDOW_KINDS { - push(window_union(kinds.iter().copied())); - return; - } + // Every union of the kinds is enumerated (#1185: no cap); a shift past + // the word size would silently enumerate nothing. + assert!( + kinds.len() < usize::BITS as usize, + "a scan run with {} window kinds has too many unions to enumerate", + kinds.len() + ); for set in 1usize..(1 << kinds.len()) { push(window_union( (0..kinds.len()) @@ -1624,12 +1608,12 @@ fn candidate_data( .map(Vec::len) .chain([windows.len(), retained.len()]) .try_fold(1usize, |total, options| total.checked_mul(options)) - .unwrap_or(usize::MAX); + .expect("the number of context combinations fits in usize"); // Each costed combination: its usage, its output, and its picks // (the option per input, the window and the retained panes). let mut costed: Vec<(CandidateUsage, ProducerOutput<'_>, Picks)> = Vec::new(); let mut error = None; - for index in 0..combinations.min(MAX_CONTEXTS) { + for index in 0..combinations { let mut rest = index; let mut pick = |options: usize| { let picked = rest % options; @@ -1787,48 +1771,45 @@ fn candidate_data( .try_fold(1usize, |total, producers| { total.checked_mul(producers.len()) }) - .unwrap_or(usize::MAX); - if choices > MAX_PRODUCER_CHOICES { - depends_on.push(UsageContext::TooManyContexts); - } else { - for index in 0..choices { - let mut rest = index; - let chosen: Vec<&(ProducerChoice, Vec)> = options_of_producers - .iter() - .map(|producers| { - let picked = &producers[rest % producers.len()]; - rest /= producers.len(); - picked - }) - .collect(); - let matching: Vec<&(CandidateUsage, usize)> = costed - .iter() - .zip(&given) - .filter(|((_, _, picks), _)| { - inputs_changing_usage.iter().zip(&chosen).all( - |(&input, (_, options))| options.contains(&picks.options[input]), - ) - }) - .map(|(_, gives)| gives) - .collect(); - let costliest = matching.iter().copied().max_by(|a, b| { - let total = |usage: &CandidateUsage| { - usage.cost_per_execution.unwrap_or(0.0) - + usage.standing_cost_per_second.unwrap_or(0.0) - + usage.work_ms_per_execution - }; - total(&a.0).total_cmp(&total(&b.0)) - }); - if matching.iter().any(|gives| Some(*gives) != costliest) { - result[class.0].producer_usage_ambiguous = true; - } - result[class.0].usage_per_producers.push(( - chosen.iter().map(|(choice, _)| *choice).collect(), - costliest.copied(), - )); + .expect("the number of producer choices fits in usize"); + for index in 0..choices { + let mut rest = index; + let chosen: Vec<&(ProducerChoice, Vec)> = options_of_producers + .iter() + .map(|producers| { + let picked = &producers[rest % producers.len()]; + rest /= producers.len(); + picked + }) + .collect(); + let matching: Vec<&(CandidateUsage, usize)> = costed + .iter() + .zip(&given) + .filter(|((_, _, picks), _)| { + inputs_changing_usage + .iter() + .zip(&chosen) + .all(|(&input, (_, options))| options.contains(&picks.options[input])) + }) + .map(|(_, gives)| gives) + .collect(); + let costliest = matching.iter().copied().max_by(|a, b| { + let total = |usage: &CandidateUsage| { + usage.cost_per_execution.unwrap_or(0.0) + + usage.standing_cost_per_second.unwrap_or(0.0) + + usage.work_ms_per_execution + }; + total(&a.0).total_cmp(&total(&b.0)) + }); + if matching.iter().any(|gives| Some(*gives) != costliest) { + result[class.0].producer_usage_ambiguous = true; } - result[class.0].inputs_changing_usage = inputs_changing_usage; + result[class.0].usage_per_producers.push(( + chosen.iter().map(|(choice, _)| *choice).collect(), + costliest.copied(), + )); } + result[class.0].inputs_changing_usage = inputs_changing_usage; } if retained.len() > 1 { let retained_changes_usage = changes(&|a: &Picks, b: &Picks| { @@ -1886,9 +1867,6 @@ fn candidate_data( if uncostable_context { depends_on.push(UsageContext::UncostableContext); } - if combinations > MAX_CONTEXTS { - depends_on.push(UsageContext::TooManyContexts); - } result[class.0].output_count = produced.len(); outputs[class.0] = Some(Ok(produced)); result[class.0].usage = Ok(first); diff --git a/crates/plan-selection/src/groups/accuracy_positions.rs b/crates/plan-selection/src/groups/accuracy_positions.rs index 710ea029e..fa81af61f 100644 --- a/crates/plan-selection/src/groups/accuracy_positions.rs +++ b/crates/plan-selection/src/groups/accuracy_positions.rs @@ -35,11 +35,6 @@ use crate::checks::guarantee_check; use crate::report::RejectionCause; use crate::SelectionModels; -/// The most outputs one candidate (or query) is built as for one set of -/// approximate candidates; beyond it the set is unusable -/// ([`UnusableReason::NotChecked`]). -pub(crate) const MAX_COMPOSITIONS: usize = 64; - /// What the accuracy rows (#1095 §2.0 row 9) are built from. #[derive(Debug, Clone, Default)] pub(crate) struct AccuracyPositions { @@ -375,9 +370,7 @@ impl Checker<'_> { } per_reference.push(distinct_for_readers(outputs, &mut before.keys)); } - let Some(roots) = self.roots(&per_reference) else { - return Outcome::Fails(UnusableReason::NotChecked); - }; + let roots = self.roots(&per_reference); let mut outcome = Outcome::Exact; for root in roots { if root_guarantee(&root).is_some_and(|g| g.is_exact()) { @@ -402,24 +395,22 @@ impl Checker<'_> { outcome } - /// The query roots over each combination of `per_reference`'s outputs; - /// `None` beyond [`MAX_COMPOSITIONS`]. - fn roots(&self, per_reference: &[Vec>]) -> Option> { - let combinations = combinations(per_reference)?; - Some( - combinations - .into_iter() - .map(|chosen| match self.root { - QueryRoot::Operator(_) => QueryRoot::Operator(chosen[0].clone()), - QueryRoot::Scalar(expr) => { - let mut next = chosen.into_iter(); - QueryRoot::Scalar(expr.map_operator_refs(&mut |_| { + /// The query roots over each combination of `per_reference`'s outputs. + fn roots(&self, per_reference: &[Vec>]) -> Vec { + combinations(per_reference) + .into_iter() + .map(|chosen| match self.root { + QueryRoot::Operator(_) => QueryRoot::Operator(chosen[0].clone()), + QueryRoot::Scalar(expr) => { + let mut next = chosen.into_iter(); + QueryRoot::Scalar( + expr.map_operator_refs(&mut |_| { next.next().expect("one output per reference") - })) - } - }) - .collect(), - ) + }), + ) + } + }) + .collect() } /// `class`'s node with an exact guarantee, built once. @@ -540,7 +531,7 @@ impl Checker<'_> { /// timing and materialization and its guarantee composed again; outputs the /// same for their readers kept once. fn rebuilt(node: &Rc, per_input: &[Vec>], keys: &mut Keys) -> Built { - let combinations = combinations(per_input).ok_or(UnusableReason::NotChecked)?; + let combinations = combinations(per_input); let mut outputs = Vec::with_capacity(combinations.len()); for chosen in combinations { let mut next = chosen.into_iter(); @@ -558,14 +549,10 @@ fn rebuilt(node: &Rc, per_input: &[Vec>], keys: & Ok(distinct_for_readers(outputs, keys)) } -/// Every choice of one entry per list, in order; `None` beyond -/// [`MAX_COMPOSITIONS`]. One empty list gives none. -fn combinations(lists: &[Vec]) -> Option>> { +/// Every choice of one entry per list, in order. One empty list gives none. +fn combinations(lists: &[Vec]) -> Vec> { let mut combinations: Vec> = vec![Vec::new()]; for list in lists { - if combinations.len().saturating_mul(list.len()) > MAX_COMPOSITIONS { - return None; - } combinations = combinations .into_iter() .flat_map(|prefix| { @@ -577,7 +564,7 @@ fn combinations(lists: &[Vec]) -> Option>> { }) .collect(); } - Some(combinations) + combinations } /// `outputs` with outputs the same for their readers kept once (the diff --git a/crates/plan-selection/src/milp/cost_terms.rs b/crates/plan-selection/src/milp/cost_terms.rs index 6f5c3afc5..7814f7b3c 100644 --- a/crates/plan-selection/src/milp/cost_terms.rs +++ b/crates/plan-selection/src/milp/cost_terms.rs @@ -69,8 +69,6 @@ use crate::groups::{ use super::model::{Reader, Scope, SelectionMilp, SelectionMilpError}; -use crate::groups::MAX_WINDOW_KINDS; - /// The numbers of one run `asap-cost` computed, as the objective and the /// latency rows read them. #[derive(Debug, Clone, Copy, Default, PartialEq)] @@ -158,7 +156,6 @@ pub(crate) fn run_terms( for usage_context in data.usage_depends_on.clone() { match usage_context { UsageContext::ProducerOnACycle => milp.note_inexact("usage: a producer on a cycle"), - UsageContext::TooManyContexts => milp.note_inexact("usage: too many contexts"), UsageContext::ProducerOutput | UsageContext::ScanWindowOfReader | UsageContext::PaneRetentionOfReachingQueries @@ -575,7 +572,7 @@ fn window_terms( Some(Vec::new()) } }, - count if count <= MAX_WINDOW_KINDS => { + count => { // `σ_W`: the exact OR, over the reading contexts, of "the // context reads the scan and a reader at kind `W` is used" // (`w`, exact from both sides at integral binaries). @@ -626,6 +623,10 @@ fn window_terms( sigma.push(or); } let mut terms = Vec::new(); + assert!( + count < usize::BITS as usize, + "a scan run with {count} window kinds has too many unions to encode" + ); for set in 1usize..(1 << count) { let window = scan_window_read_at( (0..count) @@ -662,18 +663,6 @@ fn window_terms( } Some(terms) } - _ => { - milp.note_inexact("a scan read at more window kinds than encoded (widest charged)"); - let widest = scan_window_read_at(kinds.iter().copied()); - usage_at(milp, widest).map(|(numbers, output)| { - vec![RunTerm { - indicator: run.into(), - numbers, - plain: true, - output, - }] - }) - } }; milp.window_cache.insert((context, scan), terms.clone()); terms diff --git a/crates/plan-selection/src/milp/lp_diagnosis.rs b/crates/plan-selection/src/milp/lp_diagnosis.rs index d1418de1c..d8bdc113f 100644 --- a/crates/plan-selection/src/milp/lp_diagnosis.rs +++ b/crates/plan-selection/src/milp/lp_diagnosis.rs @@ -14,16 +14,16 @@ //! - `runs`: the charged `p` and `x` variables with the largest difference //! between the incumbent and the LP, with their values in both. //! -//! The deadline is pushed back by the diagnosis's time, so the MILP keeps -//! its budget; a diagnosed run's wall time (and the facade's later rounds) -//! is still not comparable with an undiagnosed one. +//! A diagnosed run's wall time is not comparable with an undiagnosed one. +//! The LPs and the root node run without a time limit: an LP solve and a +//! one-node MIP solve end on their own. use std::collections::BTreeMap; use std::time::Instant; use good_lp::solvers::highs::highs; use good_lp::{constraint, variable, Constraint, Expression, ProblemVariables, SolverModel}; -use super::model::{objective_scale, SelectionMilp, HIGHS_THREADS}; +use super::model::{objective_scale, SelectionMilp}; /// The environment variable naming the file the diagnosis appends to. const DIAGNOSIS_ENV: &str = "ASAP_STAGE3_LP_DIAGNOSIS"; @@ -31,9 +31,6 @@ const DIAGNOSIS_ENV: &str = "ASAP_STAGE3_LP_DIAGNOSIS"; /// The runs listed per solve. const RUNS_SHOWN: usize = 40; -/// The time limit of each LP and of the root node. -const LP_TIME_LIMIT_S: f64 = 300.0; - /// One LP (or root node) solved. #[derive(Debug, Clone, serde::Serialize)] struct Bound { @@ -114,8 +111,7 @@ impl Diagnosis { .with_all(rows); let mut model = problem.try_into_inner().expect("a HiGHS model"); model.set_option("output_flag", false); - model.set_option("threads", HIGHS_THREADS); - model.set_option("time_limit", LP_TIME_LIMIT_S); + model.set_option("threads", milp.highs_threads); if root { // The root node, then stop: its bound is the root bound. model.set_option("mip_max_nodes", 1); @@ -227,7 +223,6 @@ impl Diagnosis { incumbent: Option<&[f64]>, objective_per_second: f64, dual_bound_per_second: f64, - status: &str, ) { let by_family = |columns: Option<&[f64]>| -> Option> { let columns = columns.filter(|columns| !columns.is_empty())?; @@ -264,7 +259,6 @@ impl Diagnosis { let line = serde_json::json!({ "runs": runs, "variables": self.column_families.len(), - "status": status, "objective_per_second": objective_per_second, "dual_bound_per_second": dual_bound_per_second, "lp": self.lp, diff --git a/crates/plan-selection/src/milp/model.rs b/crates/plan-selection/src/milp/model.rs index c71952886..ffd268768 100644 --- a/crates/plan-selection/src/milp/model.rs +++ b/crates/plan-selection/src/milp/model.rs @@ -65,10 +65,6 @@ pub(crate) enum SelectionMilpError { EntryCount { roots: usize, entries: usize }, /// No plan meets every row. Infeasible, - /// HiGHS stopped (at the deadline) without an integral incumbent. - NoSolution { status: String }, - /// The deadline passed before the solve. - DeadlineReached, /// A rate or another number `asap-cost` could not compute. Cost(String), /// HiGHS failed. @@ -84,6 +80,19 @@ pub(crate) struct SelectionMilpOptions { /// Solve each independent part alone ([`independent_parts`]); off only /// to compare with one solve in tests. pub(crate) split_into_parts: bool, + /// HiGHS's threads (its `threads` option). One thread makes ties among + /// plans of equal cost break the same way on every run. + pub(crate) highs_threads: i32, +} + +impl Default for SelectionMilpOptions { + fn default() -> Self { + Self { + enforce_accuracy: true, + split_into_parts: true, + highs_threads: 1, + } + } } /// Which part of the model a variable or row belongs to, for the size @@ -277,6 +286,8 @@ pub(crate) struct SelectionMilp<'a> { BTreeMap<(usize, PhysicalCandidateId, usize), Vec<(ResultSlot, Variable)>>, /// [`SelectionMilpOptions::split_into_parts`]. pub(crate) split_into_parts: bool, + /// [`SelectionMilpOptions::highs_threads`]. + pub(crate) highs_threads: i32, } /// Variables and rows of one model. @@ -291,26 +302,15 @@ pub(crate) struct SelectionMilpSize { pub(crate) inexact: BTreeMap<&'static str, usize>, } -/// How the solve ended with a plan. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub(crate) enum SelectionMilpStatus { - Optimal, - /// HiGHS's time limit (the deadline), with an integral incumbent. - TimeLimit, -} - -/// The raw solution, ready to decode. +/// The raw solution of a solve run to proven optimality, ready to decode. #[derive(Debug, Clone, PartialEq)] pub(crate) struct SelectionMilpSolution { - pub(crate) status: SelectionMilpStatus, /// The objective: cost per second without the request fees. pub(crate) cost_per_second: f64, /// The request fees per second, the same for every plan. pub(crate) request_cost_per_second: f64, /// HiGHS's dual bound on `cost_per_second`. pub(crate) cost_lower_bound_per_second: f64, - /// HiGHS's relative gap. - pub(crate) relative_gap: f64, /// Per context, the options with `p[ctx, c] = 1`. pub(crate) chosen: Vec>, /// The shareable candidates with `x[m] = 1`. @@ -508,6 +508,7 @@ pub(crate) fn build_selection_milp<'a>( window_cache: HashMap::new(), slot_links: BTreeMap::new(), split_into_parts: options.split_into_parts, + highs_threads: options.highs_threads, }; choice_variables(&mut milp); structural_rows(&mut milp); @@ -1820,7 +1821,6 @@ pub(crate) struct PartSolve { pub(crate) variables: usize, pub(crate) binaries: usize, pub(crate) rows: usize, - pub(crate) status: String, pub(crate) solve_ms: u64, /// HiGHS's dual bound on the part's objective (cost per second). pub(crate) dual_bound_per_second: f64, @@ -1837,10 +1837,9 @@ struct PartModel { queries: usize, } -/// A part solved: its columns' values, the HiGHS objective and dual bound -/// (cost per second), and the status. +/// A part solved to proven optimality: its columns' values, and the HiGHS +/// objective and dual bound (cost per second). struct SolvedPart { - status: SelectionMilpStatus, columns: Vec, objective_per_second: f64, dual_bound_per_second: f64, @@ -1849,27 +1848,18 @@ struct SolvedPart { /// Solves `milp` with HiGHS, one solve per independent part /// ([`independent_parts`]) when `split_into_parts`, each part from the warm -/// start with its share of the time left before `deadline` (by variables). +/// start and run to proven optimality (no time limit). pub(crate) fn solve_selection_milp( milp: SelectionMilp<'_>, - deadline: Option, build_ms: u64, ) -> Result { let size = milp.size(); + let highs_threads = milp.highs_threads; let warm_started = Instant::now(); - let search_budget = Instant::now() - + deadline - .map(|deadline| deadline.saturating_duration_since(Instant::now()) / 5) - .unwrap_or(WARM_START_MOST_TIME) - .min(WARM_START_MOST_TIME); let completion = super::completion::Completion::new(&milp); - let start = warm_start::warm_start(&milp, &completion, Some(search_budget)); + let start = warm_start::warm_start(&milp, &completion); let warm_start_ms = warm_started.elapsed().as_millis() as u64; let warm_start_cost_per_second = start.as_ref().map(|start| start.cost_per_second); - // The diagnosis's LP solves do not count against the deadline: it is - // pushed back by their time. - #[cfg(feature = "lp-diagnosis")] - let diagnosis_started = Instant::now(); #[cfg(feature = "lp-diagnosis")] let diagnosis = { let column_of: HashMap = milp @@ -1889,8 +1879,6 @@ pub(crate) fn solve_selection_milp( super::lp_diagnosis::Diagnosis::run(&milp, start_columns.as_deref()) .map(|diagnosis| (diagnosis, start_columns)) }; - #[cfg(feature = "lp-diagnosis")] - let deadline = deadline.map(|deadline| deadline + diagnosis_started.elapsed()); let parts = if milp.split_into_parts { independent_parts(&milp) } else { @@ -1977,24 +1965,10 @@ pub(crate) fn solve_selection_milp( } models }; - // Smallest first: time a small part leaves goes to the larger ones. - let mut order: Vec = (0..models.len()).collect(); - order.sort_by_key(|&index| models[index].columns.len()); - let mut remaining_columns: usize = models.iter().map(|model| model.columns.len()).sum(); - let mut models: Vec> = models.into_iter().map(Some).collect(); let mut columns = vec![0.0; column_count]; - let mut solved_parts: Vec> = (0..models.len()).map(|_| None).collect(); + let mut solved_parts: Vec = Vec::with_capacity(models.len()); let started = Instant::now(); - for index in order { - let model = models[index].take().expect("solved once"); - let share = model.columns.len().max(1) as f64 / remaining_columns.max(1) as f64; - remaining_columns -= model.columns.len(); - let part_deadline = deadline.map(|deadline| { - Instant::now() - + deadline - .saturating_duration_since(Instant::now()) - .mul_f64(share) - }); + for model in models { let start = start_columns.as_ref().map(|start| { model .columns @@ -2003,25 +1977,13 @@ pub(crate) fn solve_selection_milp( .collect::>() }); let part_columns = model.columns.clone(); - let solved = solve_part(model, start, part_deadline)?; + let solved = solve_part(model, start, highs_threads)?; for (&column, &value) in part_columns.iter().zip(&solved.columns) { columns[column] = value; } - solved_parts[index] = Some(solved); + solved_parts.push(solved); } let solve_ms = started.elapsed().as_millis() as u64; - let solved_parts: Vec = solved_parts - .into_iter() - .map(|part| part.expect("every part solved")) - .collect(); - let status = if solved_parts - .iter() - .all(|part| part.status == SelectionMilpStatus::Optimal) - { - SelectionMilpStatus::Optimal - } else { - SelectionMilpStatus::TimeLimit - }; let highs_objective: f64 = solved_parts .iter() .map(|part| part.objective_per_second) @@ -2030,17 +1992,10 @@ pub(crate) fn solve_selection_milp( .iter() .map(|part| part.dual_bound_per_second) .sum(); - let relative_gap = match status { - SelectionMilpStatus::Optimal => 0.0, - SelectionMilpStatus::TimeLimit => { - (highs_objective - dual_bound) / highs_objective.abs().max(f64::MIN_POSITIVE) - } - }; // The objective of the integral plan: the incumbent's binaries, // rounded, with every continuous variable at its cheapest value (the // solver's own value carries its tolerances, relative to the scaled - // objective). At the deadline, HiGHS's incumbent objective (an upper - // bound) when the completion does not apply. + // objective). HiGHS's objective when the completion does not apply. let objective_value = match completion.complete_columns(&columns) { Ok(completed) => completed.cost_per_second, Err(reason) => { @@ -2061,16 +2016,13 @@ pub(crate) fn solve_selection_milp( .map(|completed| completed.columns.as_slice()), objective_value, dual_bound, - &format!("{status:?}"), ); } let on = |variable: Variable| columns[column_of[&variable]] > 0.5; Ok(SelectionMilpSolution { - status, cost_per_second: objective_value, request_cost_per_second, cost_lower_bound_per_second: dual_bound, - relative_gap, chosen: choice .iter() .map(|options| { @@ -2104,12 +2056,12 @@ pub(crate) fn solve_selection_milp( }) } -/// Solves one part once with HiGHS, from `start` (its columns' values) when -/// given, with the time left before `deadline` as its time limit. +/// Solves one part once with HiGHS to proven optimality, from `start` (its +/// columns' values) when given, on `highs_threads` threads. fn solve_part( model: PartModel, start: Option>, - deadline: Option, + highs_threads: i32, ) -> Result { let PartModel { variables, @@ -2144,49 +2096,31 @@ fn solve_part( model.set_option("output_flag", false); // HiGHS's MIP search is mostly serial; a thread per core costs // gigabytes of address space on a large host (#1036 under 8 GB). - model.set_option("threads", HIGHS_THREADS); - let started_with_plan = start.is_some(); + model.set_option("threads", highs_threads); if let Some(start) = &start { model .try_set_solution(Some(start), None, None, None) .map_err(|status| SelectionMilpError::Solver(format!("warm start: {status:?}")))?; } - let mut only_read_back = false; - if let Some(deadline) = deadline { - let seconds = deadline - .saturating_duration_since(Instant::now()) - .as_secs_f64(); - if seconds <= 0.0 { - if !started_with_plan { - return Err(SelectionMilpError::DeadlineReached); - } - only_read_back = true; - } - model.set_option("time_limit", seconds.max(MOMENT_TO_READ_THE_START_S)); - } let started = Instant::now(); let solved = model .try_solve() .map_err(|status| SelectionMilpError::Solver(format!("{status:?}")))?; let solve_ms = started.elapsed().as_millis() as u64; use highs::{HighsModelStatus, HighsSolutionStatus}; - let status = match solved.status() { - HighsModelStatus::Optimal if only_read_back => SelectionMilpStatus::TimeLimit, - HighsModelStatus::Optimal => SelectionMilpStatus::Optimal, + // No limit is set, so HiGHS stops only at proven optimality or + // infeasibility; any other status is a solver failure. + match solved.status() { + HighsModelStatus::Optimal => {} HighsModelStatus::Infeasible | HighsModelStatus::UnboundedOrInfeasible => { return Err(SelectionMilpError::Infeasible) } - HighsModelStatus::ReachedTimeLimit - | HighsModelStatus::ReachedInterrupt - | HighsModelStatus::ReachedIterationLimit - | HighsModelStatus::ReachedMemoryLimit - | HighsModelStatus::ReachedSolutionLimit => SelectionMilpStatus::TimeLimit, other => return Err(SelectionMilpError::Solver(format!("{other:?}"))), - }; + } if solved.primal_solution_status() != HighsSolutionStatus::Feasible { - return Err(SelectionMilpError::NoSolution { - status: format!("{:?}", solved.status()), - }); + return Err(SelectionMilpError::Solver( + "HiGHS reported an optimal status without a feasible solution".into(), + )); } // HiGHS leaves `mip_dual_bound` at 0 for a part without binaries (an // LP): its optimal objective is its bound. @@ -2194,13 +2128,10 @@ fn solve_part( solved .double_info_value(c"mip_dual_bound") .unwrap_or(f64::NAN) - } else if status == SelectionMilpStatus::Optimal { - solved.objective_value() } else { - f64::NAN + solved.objective_value() } / scale; Ok(SolvedPart { - status, columns: solved.get_solution().columns().to_vec(), objective_per_second: solved.objective_value() / scale, dual_bound_per_second: dual_bound, @@ -2209,22 +2140,12 @@ fn solve_part( variables: part_columns.len(), binaries, rows: row_count, - status: format!("{status:?}"), solve_ms, dual_bound_per_second: dual_bound, }, }) } -/// HiGHS's threads (its `threads` option). -pub(crate) const HIGHS_THREADS: i32 = 4; - -/// HiGHS's time limit when the warm start used up the time to the deadline. -const MOMENT_TO_READ_THE_START_S: f64 = 1e-3; - -/// The most time the warm start takes. -const WARM_START_MOST_TIME: std::time::Duration = std::time::Duration::from_secs(60); - /// The factor the objective is multiplied by for HiGHS: the warm start's /// cost becomes about 1,000, its largest coefficient at most 10⁷; without /// a warm start cost, the largest coefficient becomes 1,000. diff --git a/crates/plan-selection/src/milp/model/tests.rs b/crates/plan-selection/src/milp/model/tests.rs index de376d437..e6809472d 100644 --- a/crates/plan-selection/src/milp/model/tests.rs +++ b/crates/plan-selection/src/milp/model/tests.rs @@ -1,5 +1,4 @@ use std::collections::{BTreeMap, BTreeSet}; -use std::time::Duration; use super::*; use crate::groups::{prepare_milp_inputs, AccuracyWeight, IngestionPipeline}; @@ -177,23 +176,22 @@ pub(crate) struct Checked { /// `fixture`'s MILP under `cost_inputs`, with `restrict` adding rows first: /// solved, decoded, realized by Stage 2 and costed by -/// `asap_cost::plan_cost`. At `Optimal` the objective plus the request fees -/// must equal `plan_cost`'s cost; at the deadline be at least it. +/// `asap_cost::plan_cost`. The objective plus the request fees must equal +/// `plan_cost`'s cost. pub(crate) fn solve_and_check( fixture: &Fixture, cost_inputs: &CostInputs, restrict: impl FnOnce(&mut SelectionMilp<'_>), ) -> Result { - solve_and_check_by(fixture, cost_inputs, restrict, None, true) + solve_and_check_by(fixture, cost_inputs, restrict, true) } -/// [`solve_and_check`] with HiGHS's time limit `limit`, and the -/// independent parts solved apart when `split_into_parts`. +/// [`solve_and_check`] with the independent parts solved apart when +/// `split_into_parts`. fn solve_and_check_by( fixture: &Fixture, cost_inputs: &CostInputs, restrict: impl FnOnce(&mut SelectionMilp<'_>), - limit: Option, split_into_parts: bool, ) -> Result { let inputs = prepare_milp_inputs( @@ -208,13 +206,13 @@ fn solve_and_check_by( &fixture.entries, cost_inputs, SelectionMilpOptions { - enforce_accuracy: true, split_into_parts, + ..SelectionMilpOptions::default() }, )?; assert!(milp.inexact.is_empty(), "inexact terms: {:?}", milp.inexact); restrict(&mut milp); - let solution = solve_selection_milp(milp, limit.map(|limit| Instant::now() + limit), 0)?; + let solution = solve_selection_milp(milp, 0)?; let dag = decode( &fixture.physical, &inputs, @@ -260,21 +258,14 @@ fn solve_and_check_by( .expect("plan_cost costs the plan"); let expected = plan.cost_per_second.expect("cost model 1") - solution.request_cost_per_second; let tolerance = 1e-6 * expected.abs().max(1e-12); - match solution.status { - SelectionMilpStatus::Optimal => assert!( - (solution.cost_per_second - expected).abs() <= tolerance, - "MILP objective {:e} != plan_cost {:e} (without {:e} of request fees); per node: {:#?}", - solution.cost_per_second, - expected, - solution.request_cost_per_second, - plan.per_node - ), - SelectionMilpStatus::TimeLimit => assert!( - expected <= solution.cost_per_second + tolerance, - "plan_cost {expected:e} above the incumbent's objective {:e}", - solution.cost_per_second - ), - } + assert!( + (solution.cost_per_second - expected).abs() <= tolerance, + "MILP objective {:e} != plan_cost {:e} (without {:e} of request fees); per node: {:#?}", + solution.cost_per_second, + expected, + solution.request_cost_per_second, + plan.per_node + ); Ok(Checked { solution, plan, @@ -387,8 +378,7 @@ pub(crate) fn two_windows() -> Fixture { #[test] fn the_optimum_costs_what_plan_cost_says() { for costs in [cost_inputs(), aws_fargate_cost_inputs()] { - let checked = solve_and_check(&shared_sum(), &costs, |_| {}).unwrap(); - assert_eq!(checked.solution.status, SelectionMilpStatus::Optimal); + solve_and_check(&shared_sum(), &costs, |_| {}).unwrap(); } } @@ -833,10 +823,7 @@ fn a_phantom_reader_does_not_satisfy_minimum_query_readers() { let fixture = shared_sum(); let costs = cost_inputs(); let inputs = prepare_milp_inputs(&fixture.physical, &fixture.entries, &fixture.data, &costs); - let options = SelectionMilpOptions { - enforce_accuracy: true, - split_into_parts: true, - }; + let options = SelectionMilpOptions::default(); let mut milp = build_selection_milp( &fixture.physical, &inputs, @@ -878,7 +865,7 @@ fn a_phantom_reader_does_not_satisfy_minimum_query_readers() { milp.builder.row("test", constraint!(choice <= 0.0)); } assert_eq!( - solve_selection_milp(milp, None, 0).err(), + solve_selection_milp(milp, 0).err(), Some(SelectionMilpError::Infeasible) ); } @@ -967,10 +954,7 @@ fn cost_models_2_and_3_are_unsupported() { }; let inputs = prepare_milp_inputs(&fixture.physical, &fixture.entries, &fixture.data, &costs); - let options = SelectionMilpOptions { - enforce_accuracy: true, - split_into_parts: true, - }; + let options = SelectionMilpOptions::default(); assert!(matches!( build_selection_milp( &fixture.physical, @@ -1018,14 +1002,11 @@ fn accuracy_rows_follow_the_composition_on_a_ratio() { inputs, &fixture.entries, &costs, - SelectionMilpOptions { - enforce_accuracy: true, - split_into_parts: true, - }, + SelectionMilpOptions::default(), ) .unwrap(); restrict(&mut milp); - solve_selection_milp(milp, None, 0).map(|_| ()) + solve_selection_milp(milp, 0).map(|_| ()) }; // Each candidate's runs: `x[m]`, or `p` in a context computing it. let runs_of = |milp: &SelectionMilp<'_>, candidates: &[PhysicalCandidateId]| -> Expression { @@ -1079,7 +1060,7 @@ fn the_warm_start_is_a_feasible_plan_costed_exactly() { let checked = solve_and_check(&fixture, &cost_inputs(), |milp| { let completion = crate::milp::completion::Completion::new(milp); let start = - crate::milp::warm_start::warm_start(milp, &completion, None).expect("a warm start"); + crate::milp::warm_start::warm_start(milp, &completion).expect("a warm start"); start_cost = Some(start.cost_per_second); let values: HashMap = start.values.into_iter().collect(); let mut fixed = Vec::new(); @@ -1103,26 +1084,6 @@ fn the_warm_start_is_a_feasible_plan_costed_exactly() { } } -// A deadline yields a plan: HiGHS starts from the warm start, so even no -// time ends with a validated incumbent, reported as the deadline. -#[test] -fn a_deadline_yields_the_warm_starts_plan() { - let checked = solve_and_check_by( - &two_windows(), - &cost_inputs(), - |_| {}, - Some(Duration::ZERO), - true, - ) - .unwrap(); - assert_eq!(checked.solution.status, SelectionMilpStatus::TimeLimit); - let start = checked - .solution - .warm_start_cost_per_second - .expect("the warm start was costed"); - assert!(checked.solution.cost_per_second <= start * (1.0 + 1e-9)); -} - // RangeHeavy templates 7 and 4: the widened scan puts a `Project` in the // memo group of the scan it reads. The group's roots are split by // self-read depth, so the plan is acyclic, valid, accepted by the realizer @@ -1154,8 +1115,7 @@ fn a_widened_scan_reading_its_own_memo_group_plans_exactly() { "the fixture has a root reading its own group" ); for costs in [cost_inputs(), aws_fargate_cost_inputs()] { - let checked = solve_and_check(&fixture, &costs, |_| {}).unwrap(); - assert_eq!(checked.solution.status, SelectionMilpStatus::Optimal); + solve_and_check(&fixture, &costs, |_| {}).unwrap(); } } @@ -1172,10 +1132,7 @@ fn independent_query_sets_solve_apart_to_the_same_plan() { let fixture = through_stages_over(&queries, &["m", "n"], true, false, false); let costs = cost_inputs(); let inputs = prepare_milp_inputs(&fixture.physical, &fixture.entries, &fixture.data, &costs); - let options = SelectionMilpOptions { - enforce_accuracy: true, - split_into_parts: true, - }; + let options = SelectionMilpOptions::default(); let milp = build_selection_milp( &fixture.physical, &inputs, @@ -1192,12 +1149,10 @@ fn independent_query_sets_solve_apart_to_the_same_plan() { assert_eq!(columns, milp.builder.variables.len()); assert_eq!(rows, milp.builder.rows.len()); - let apart = solve_and_check_by(&fixture, &costs, |_| {}, None, true).unwrap(); - let whole = solve_and_check_by(&fixture, &costs, |_| {}, None, false).unwrap(); + let apart = solve_and_check_by(&fixture, &costs, |_| {}, true).unwrap(); + let whole = solve_and_check_by(&fixture, &costs, |_| {}, false).unwrap(); assert_eq!(apart.solution.parts.len(), 2); assert_eq!(whole.solution.parts.len(), 1); - assert_eq!(apart.solution.status, SelectionMilpStatus::Optimal); - assert_eq!(whole.solution.status, SelectionMilpStatus::Optimal); let tolerance = 1e-9 * whole.solution.cost_per_second.abs(); assert!( (apart.solution.cost_per_second - whole.solution.cost_per_second).abs() <= tolerance, @@ -1225,10 +1180,7 @@ fn big_m_per_row(fixture: &Fixture, family: &str, loose: bool) -> Vec { &inputs, &fixture.entries, &costs, - SelectionMilpOptions { - enforce_accuracy: true, - split_into_parts: true, - }, + SelectionMilpOptions::default(), ) .unwrap(); LOOSE_BIG_M.with(|cell| cell.set(false)); @@ -1276,8 +1228,6 @@ fn a_tighter_context_execution_bound_keeps_the_plan() { let looser = solve_and_check(&fixture, &aws_fargate_cost_inputs(), |_| {}); LOOSE_BIG_M.with(|cell| cell.set(false)); let looser = looser.unwrap(); - assert_eq!(tightened.solution.status, SelectionMilpStatus::Optimal); - assert_eq!(looser.solution.status, SelectionMilpStatus::Optimal); let tolerance = 1e-9 * looser.solution.cost_per_second.abs(); assert!( (tightened.solution.cost_per_second - looser.solution.cost_per_second).abs() <= tolerance diff --git a/crates/plan-selection/src/milp/warm_start.rs b/crates/plan-selection/src/milp/warm_start.rs index 9f520a3b2..b7de6ae37 100644 --- a/crates/plan-selection/src/milp/warm_start.rs +++ b/crates/plan-selection/src/milp/warm_start.rs @@ -1,6 +1,5 @@ //! A feasible starting plan for the group selection MILP (#1095 §2.0, -//! "Solving strategy" step 1), so HiGHS starts with an incumbent and a -//! deadline yields a plan. +//! "Solving strategy" step 1), so HiGHS starts with an incumbent. //! //! **Each query's cheapest plan.** Queries are planned one after another //! into one plan. A query's plan is derived from its output references @@ -25,7 +24,6 @@ //! reversed, are costed exactly by the MILP's own rows and objective //! ([`Completion`]); the cheapest feasible one is the start. use std::collections::{BTreeMap, BTreeSet, HashMap}; -use std::time::Instant; use asap_cost::execution_demand::evaluation_rate; use asap_types::stages::{output_references, PhysicalCandidateId, ReaderConstraint}; @@ -83,19 +81,19 @@ enum Undo { SlotRead(usize, PhysicalCandidateId, usize), } -/// A feasible plan (module documentation), or `None` when none was found -/// before `budget`. +/// A feasible plan (module documentation), or `None` when none was found. +/// It tries a fixed number of query orders ([`MORE_ORDERS`]), so every run +/// starts from the same plan. pub(crate) fn warm_start( milp: &SelectionMilp<'_>, completion: &Completion, - budget: Option, ) -> Option { let estimates = Estimates::new(milp); let mut best: Option = None; let count = milp.query_count(); let mut orders: Vec> = vec![(0..count).collect(), (0..count).rev().collect()]; - // More orders, while the budget lasts: a fixed pseudo-random sequence - // of permutations, so runs are reproducible. + // More orders: a fixed pseudo-random sequence of permutations, so runs + // are reproducible. let mut state: u64 = 0x9e37_79b9_7f4a_7c15; for _ in 0..MORE_ORDERS { let mut order: Vec = (0..count).collect(); @@ -112,11 +110,8 @@ pub(crate) fn warm_start( // queries that may reach it. let strategies = [(false, false), (true, false), (true, true)]; let mut tried = 0; - 'orders: for order in &orders { + for order in &orders { for &(share, amortized) in &strategies { - if budget.is_some_and(|budget| Instant::now() >= budget) && best.is_some() { - break 'orders; - } tried += 1; let Some(plan) = sequential_plan(milp, &estimates, order, share, amortized, &BTreeSet::new()) @@ -162,7 +157,7 @@ pub(crate) fn warm_start( sharable.sort_by(|a, b| b.0.cmp(&a.0).then(a.1.cmp(&b.1))); let mut preferred: BTreeSet = BTreeSet::new(); for (_, m) in sharable { - if best.is_none() || budget.is_some_and(|budget| Instant::now() >= budget) { + if best.is_none() { break; } let mut trial = preferred.clone(); @@ -193,8 +188,7 @@ pub(crate) fn warm_start( best } -/// The pseudo-random orders tried after workload order and its reverse, -/// within the budget. +/// The pseudo-random orders tried after workload order and its reverse. const MORE_ORDERS: usize = 30; /// The plan's binaries: `p[ctx, c]` and `x[m]` set to 1. diff --git a/crates/plan-selection/src/report.rs b/crates/plan-selection/src/report.rs index 30b43c3aa..77fb843ae 100644 --- a/crates/plan-selection/src/report.rs +++ b/crates/plan-selection/src/report.rs @@ -283,11 +283,9 @@ impl SearchReport { #[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)] #[serde(rename_all = "snake_case")] pub enum SearchStatus { - /// The search proved the plan optimal over the catalog it was given. + /// The search proved the plan optimal over the catalog it was given + /// (the only outcome: the solve has no time limit). Optimal, - /// The search stopped at the deadline (or another solver limit) with a - /// validated plan; the gap to the lower bound may be above zero. - DeadlineReached, } /// Why Stage 3 selected no plan. @@ -316,9 +314,6 @@ pub enum ErrorSelectingPlan { }, #[error("no valid candidate: {0:?}")] NoValidCandidate(Vec), - /// The search stopped at the deadline before it found a validated plan. - #[error("no plan by the deadline: {0}")] - NoPlanByDeadline(String), /// Stage 2 rejected a plan Stage 3 decoded from its catalog: a Stage 2 /// catalog bug. #[error("the realizer rejected the selected plan: {0}")] diff --git a/crates/plan-selection/src/select.rs b/crates/plan-selection/src/select.rs index 5c9628762..ed5546432 100644 --- a/crates/plan-selection/src/select.rs +++ b/crates/plan-selection/src/select.rs @@ -5,16 +5,15 @@ //! //! One MILP solve and one realization, no loop: //! 1. the MILP's inputs (candidate dominance, execution contexts, accuracy -//! weights) and the model, solved once with HiGHS from a warm start, -//! the deadline as its time limit; +//! weights) and the model, solved once with HiGHS from a warm start to +//! proven optimality (no time limit); //! 2. the solution decoded into a [`DagOfSelectedCandidates`] //! ([`crate::milp::decode`]) and validated against the catalog; //! 3. the plan realized once through the injected realizer: a rejection is //! a Stage 2 catalog bug ([`ErrorSelectingPlan::RealizerRejected`]); //! 4. exact checks: `asap_cost::plan_cost` of the realized plan minus the -//! request fees equals the objective within a relative 10⁻⁶ at -//! `Optimal`, and is at most the objective at the deadline (otherwise an -//! error, or, when the model encoded some term inexactly, +//! request fees equals the objective within a relative 10⁻⁶ (otherwise +//! an error, or, when the model encoded some term inexactly, //! [`SearchReport::inexact_objective`]); every query's accuracy //! (`satisfies`) and latency hold, or it is an error. //! @@ -41,7 +40,6 @@ use crate::groups::{prepare_milp_inputs_with, MilpInputs}; use crate::milp::decode::{decode, ChosenCandidates}; use crate::milp::model::{ build_selection_milp, solve_selection_milp, SelectionMilpError, SelectionMilpOptions, - SelectionMilpStatus, }; use crate::models::{AccuracyCheckInputs, SelectionModels}; use crate::report::{ @@ -59,10 +57,10 @@ const COST_EQUALITY_TOLERANCE: f64 = 1e-6; /// target (`accuracy_checks`) and latency bound, and priced under /// `cost_inputs`. `entries[q]` is query `q`'s workload entry (accuracy /// target, latency bound, recurrence); `data` the data workload with its -/// bounds. `deadline` is the only budget. +/// bounds. The solve has no time limit: it runs to proven optimality. /// -/// Err: invalid or unsupported cost inputs, an infeasible catalog, no plan -/// by the deadline, or the realizer rejected the selected plan +/// Err: invalid or unsupported cost inputs, an infeasible catalog, a +/// solver failure, or the realizer rejected the selected plan /// ([`ErrorSelectingPlan::RealizerRejected`]: a Stage 2 catalog bug). pub fn select_plan( physical: &PhysicalASAPMemoCandidates, @@ -71,7 +69,6 @@ pub fn select_plan( accuracy_checks: &AccuracyCheckInputs<'_>, cost_inputs: &CostInputs, realizer: &dyn AssignmentRealizer, - deadline: Option, ) -> Result { cost_inputs .validate() @@ -103,12 +100,14 @@ pub fn select_plan( cost_inputs, SelectionMilpOptions { enforce_accuracy, - split_into_parts: true, + ..SelectionMilpOptions::default() }, ) .map_err(selection_error)?; let build_ms = construction_started.elapsed().as_millis() as u64; - let solution = solve_selection_milp(milp, deadline, build_ms).map_err(selection_error)?; + let solution = solve_selection_milp(milp, build_ms).map_err(|error| { + with_uncostable_reasons(selection_error(error), &inputs, &pruned_candidates) + })?; let validation_started = Instant::now(); let chosen = ChosenCandidates { @@ -147,12 +146,7 @@ pub fn select_plan( let objective_per_second = solution.cost_per_second; let difference = (objective_per_second - plan_cost_per_second).abs(); let tolerance = COST_EQUALITY_TOLERANCE * plan_cost_per_second.abs().max(f64::MIN_POSITIVE); - // At the deadline the incumbent's auxiliary variables may sit above - // their true values: the objective is an upper bound on `plan_cost`. - let within = match solution.status { - SelectionMilpStatus::Optimal => difference <= tolerance, - SelectionMilpStatus::TimeLimit => plan_cost_per_second <= objective_per_second + tolerance, - }; + let within = difference <= tolerance; report_run(&solution, objective_per_second, plan_cost_per_second); let inexact_objective = if within { None @@ -192,10 +186,7 @@ pub fn select_plan( accepted_accuracy_per_query, pruned_candidates, search_report: SearchReport { - status: match solution.status { - SelectionMilpStatus::Optimal => SearchStatus::Optimal, - SelectionMilpStatus::TimeLimit => SearchStatus::DeadlineReached, - }, + status: SearchStatus::Optimal, // The dual bound is on the objective, which leaves out the // request fees every plan pays; the selection's cost has them. cost_lower_bound_per_second: solution @@ -233,18 +224,47 @@ fn selection_error(error: SelectionMilpError) -> ErrorSelectingPlan { constraints together (the selection MILP is infeasible)" .into(), ), - SelectionMilpError::NoSolution { status } => ErrorSelectingPlan::NoPlanByDeadline(format!( - "HiGHS stopped ({status}) without an integral plan" - )), - SelectionMilpError::DeadlineReached => { - ErrorSelectingPlan::NoPlanByDeadline("the deadline passed before the solve".into()) - } SelectionMilpError::Cost(reason) | SelectionMilpError::Solver(reason) => { ErrorSelectingPlan::Solver(reason) } } } +/// An infeasible selection with the reasons `asap-cost` could not cost +/// the candidates the model left out (e.g. a source missing the fact its +/// pricing needs), each once in candidate order: those candidates may be +/// why no plan is left. +fn with_uncostable_reasons( + error: ErrorSelectingPlan, + inputs: &MilpInputs, + pruned_candidates: &[PrunedCandidate], +) -> ErrorSelectingPlan { + let ErrorSelectingPlan::NoValidCandidate(mut rejections) = error else { + return error; + }; + let mut reasons: Vec<&str> = Vec::new(); + for pruned in pruned_candidates { + if pruned.reason != PruneReason::UsageNotCostable { + continue; + } + if let Err(reason) = &inputs.candidate_data[pruned.candidate.0].usage { + if !reasons.contains(&reason.as_str()) { + reasons.push(reason); + } + } + } + if let Some(rejection) = rejections.first_mut() { + if !reasons.is_empty() { + rejection.reason = format!( + "{}; candidates left out because they could not be costed: {}", + rejection.reason, + reasons.join("; ") + ); + } + } + ErrorSelectingPlan::NoValidCandidate(rejections) +} + /// What Stage 3 removes before the MILP: dominated candidates, and those /// `asap-cost` could not cost (the model leaves them out), in candidate /// order. @@ -326,7 +346,6 @@ fn report_run( return; }; let line = serde_json::json!({ - "status": format!("{:?}", solution.status), "variables": solution.size.variables, "binaries": solution.size.binaries, "rows": solution.size.rows, @@ -341,7 +360,6 @@ fn report_run( "objective_per_second": objective_per_second, "plan_cost_per_second": plan_cost_per_second, "dual_bound_per_second": solution.cost_lower_bound_per_second, - "relative_gap": solution.relative_gap, "parts": solution.parts, }); use std::io::Write; diff --git a/crates/plan-selection/src/select/tests.rs b/crates/plan-selection/src/select/tests.rs index 4bc561ec1..68ce48681 100644 --- a/crates/plan-selection/src/select/tests.rs +++ b/crates/plan-selection/src/select/tests.rs @@ -28,7 +28,6 @@ fn select( }, &cost_inputs(), realizer, - None, ) } @@ -166,7 +165,6 @@ fn cost_model_2_is_unsupported() { }, &costs, &Rejecting, - None, ); assert!(matches!( result, diff --git a/crates/plan-selection/src/selector.rs b/crates/plan-selection/src/selector.rs index b3573dfbc..21a417487 100644 --- a/crates/plan-selection/src/selector.rs +++ b/crates/plan-selection/src/selector.rs @@ -16,7 +16,8 @@ use crate::checks::{accuracy_check, accuracy_targets, latency_violation}; use crate::report::{ AcceptedAccuracy, ErrorSelectingPlan, Rejection, RejectionCause, WholePlanSelection, }; -use crate::{AnalyticalCostError, SelectionModels}; +use crate::SelectionModels; +use asap_cost::analytical::AnalyticalCostError; use asap_cost::{ kept_tumbling_window_chains, node_map, plan_cost, query_reach, tumbling_window_retention_ms, }; diff --git a/crates/planner/src/run.rs b/crates/planner/src/run.rs index b16c10888..e2568b7e0 100644 --- a/crates/planner/src/run.rs +++ b/crates/planner/src/run.rs @@ -506,7 +506,6 @@ fn plan_round( &models.accuracy_check_inputs(), models.cost_inputs, realizer, - deadline, ); report.stage3_ms = elapsed_ms(started); let fatal = match &selection { @@ -517,8 +516,7 @@ fn plan_round( Err(error) => { report.error = Some(error.to_string()); match error { - ErrorSelectingPlan::NoPlanByDeadline(_) - | ErrorSelectingPlan::EvaluationDeadlineExceeded => None, + ErrorSelectingPlan::EvaluationDeadlineExceeded => None, // An infeasible restricted catalog does not make the complete // one infeasible. ErrorSelectingPlan::NoValidCandidate(_) diff --git a/docs/design_docs/architecture/input-output-workflow.md b/docs/design_docs/architecture/input-output-workflow.md index 446b5af2c..0e62400dc 100644 --- a/docs/design_docs/architecture/input-output-workflow.md +++ b/docs/design_docs/architecture/input-output-workflow.md @@ -291,8 +291,9 @@ candidate whose output has no guarantee for an approximate target. * `selection.accepted_accuracy_per_query`: per query, the accuracy metric Stage 3 accepted; * `selection.pruned_candidates`: the candidates Stage 3 ruled out; -* `selection.search_report`: how the search ended (`Optimal` or - `DeadlineReached`), its lower bound, the MILP's size and its times. +* `selection.search_report`: how the search ended (`Optimal`: the MILP + runs to proven optimality, with no time limit), its lower bound, the + MILP's size and its times. Planner selects because it now has the inputs a backend used to own: the deployment declares its cost model, measurements and capabilities, and every diff --git a/docs/design_docs/proposals/planner-layering.md b/docs/design_docs/proposals/planner-layering.md index 91b86bfb3..200b27a90 100644 --- a/docs/design_docs/proposals/planner-layering.md +++ b/docs/design_docs/proposals/planner-layering.md @@ -605,7 +605,7 @@ the memo path's entry points. | 1, Pass 2 | One Pass 1 choice per computation → that choice rewritten by each sharing rule, every subset of sharing choices (no longer in the code) | `Memo`; the entries' recurrence and range steps → the same memo with each rule's candidates added to the groups they implement | the memo rules (`pass2/memo/rules/`), `pass2::memo::window_candidates` | | 1, concrete summary-state groups | (part of Pass 2) | `Memo` → the memo with concrete summary-state groups: one group per shared size or tumbling-window layout that readers can read | `add_shared_summary_state_groups`, Pass 2's last step; the Pass 2 output is `LogicalASAPMemoCandidates` (`generate_logical_asap_memo_candidates`) | | 2 | One whole-workload logical candidate; entries (recurrence, predictability), `DataWorkload`, query-time retention → its physical candidates, one per materialization assignment | `LogicalASAPMemoCandidates`; entries, `DataWorkload`, query-time and raw-data retention → `PhysicalASAPMemoCandidates`: every operator node of every candidate with each allowed timing and materialization (one physical candidate each, no copies), the producers of each input, reader constraints (each applied to every selected run of its candidate), and the pruned candidates (`PrunedCandidate`) | `generate_physical_asap_memo_candidates` (`asap_physical_asap_optimization`) | -| 3 | Every physical candidate; accuracy targets, accuracy model and evidence, cost model → one plan | `PhysicalASAPMemoCandidates`; entries, `DataWorkload`, `AccuracyCheckInputs`, `CostInputs`, Stage 2's realizer, a deadline → `Selection`: the selected plan (`selected_dag`, a `DagOfSelectedCandidates`: one node per run of a candidate, so a candidate recomputed by two readers or shared as two instances over different inputs is two nodes, with an edge per input), its cost, the accepted accuracy per query and the search report | `select_plan` (`asap_plan_selection`); the facade (`run_planning`, output `PlanningRun`) runs it per tumbling-window-width round and builds the selected `PhysicalWorkloadPlan` with `AssignmentRealizer::realize`, one plan node per selected node | +| 3 | Every physical candidate; accuracy targets, accuracy model and evidence, cost model → one plan | `PhysicalASAPMemoCandidates`; entries, `DataWorkload`, `AccuracyCheckInputs`, `CostInputs`, Stage 2's realizer → `Selection`: the selected plan (`selected_dag`, a `DagOfSelectedCandidates`: one node per run of a candidate, so a candidate recomputed by two readers or shared as two instances over different inputs is two nodes, with an edge per input), its cost, the accepted accuracy per query and the search report | `select_plan` (`asap_plan_selection`); the facade (`run_planning`, output `PlanningRun`) runs it per tumbling-window-width round and builds the selected `PhysicalWorkloadPlan` with `AssignmentRealizer::realize`, one plan node per selected node | | 4. Execution | The selected `PhysicalASAPDAG` → query results | Same: an `InstallablePlan` (`{dag, query_texts}`) → query results | The deployment runtime installs the `InstallablePlan` | What changed with the memo: diff --git a/docs/develop_docs/library-api.md b/docs/develop_docs/library-api.md index ec9166102..3b2267408 100644 --- a/docs/develop_docs/library-api.md +++ b/docs/develop_docs/library-api.md @@ -192,7 +192,6 @@ asap_plan_selection::select_plan( accuracy_checks: &AccuracyCheckInputs<'_>, // accuracy model and evidence cost_inputs: &CostInputs, realizer: &dyn AssignmentRealizer, // Stage 2's, e.g. Stage2AssignmentRealizer - deadline: Option, ) -> Result ``` diff --git a/tools/dag-viewer/README.md b/tools/dag-viewer/README.md index 15d0a2c8a..d86637799 100644 --- a/tools/dag-viewer/README.md +++ b/tools/dag-viewer/README.md @@ -356,7 +356,8 @@ are ignored, `stage_pipeline`'s default. materialization assignment and all priced by the reference pricer. `--max-combinations` bounds the choices enumerated. - `stage3_plan_selection.search_report` reports how the search ended - (`optimal` or `deadline_reached`), its cost lower bound when it has one, + (`optimal`; documents from before the MILP ran without a time limit may + say `deadline_reached`), its cost lower bound when it has one, the MILP's size and the model, solver and validation times. Historical committed documents with the earlier Stage 3 section (`costs`, `selected`, `rejected`, `search`) remain readable.