refactor(planner): plan without deadlines over every tumbling-window width (#1185, part 2)

crates/cost/src/ranking.rs

@@ -1,6 +1,5 @@
 //! How a cost model orders valid plans ([`PlanRanking`]): Stage 3 ranks its
-//! candidates with it, and the facade picks the best of its tumbling-window-width
-//! rounds' selections with it (#1047).
+//! candidates with it.
 use std::cmp::Ordering;
 
 use crate::{ClusterObjective, CostModel, PlanCost, QueryLatency, ResourceUsage};

crates/devtools/src/bin/sketch_coverage.rs

@@ -67,7 +67,6 @@ fn analyze_corpus(
             &NoAccuracyEvidence,
             &DataWorkload::default(),
             None,
-            None,
         ) else {
             rejected += 1;
             continue;

crates/devtools/src/bin/stage_pipeline.rs

@@ -24,11 +24,10 @@
 // Writes an `asap-stage-pipeline/v1` document (tools/dag-viewer) with the
 // four planner stages as the facade runs them (`asap_planner::run_planning`).
 // The default document keeps the selected plan; `--exhaustive` adds the
-// reference enumeration: the selected round's memo choices, each implemented
+// reference enumeration: Pass 2's memo choices, each implemented
 // by Stage 2 with every materialization assignment, all priced by the
 // reference pricer (`asap_plan_selection::reference_pricer`), under
-// `reference_pricing`. `--milp-time-limit-s` bounds planning; a timeout keeps
-// the best plan found, never an optimality claim for unexplored candidates.
+// `reference_pricing`.
 // Unlike `plan_installable`'s default settings, Stage 3 ignores accuracy
 // guarantees unless `--enforce-accuracy-guarantees` is given (below);
 // `accuracy_guarantees` is `"ignored"` or `"enforced"`.
@@ -49,7 +48,7 @@
 //     built it, then under `--exhaustive` each enumerated choice's physical
 //     plans `E<n>-<k>`; an enumerated choice Stage 2 cannot implement is
 //     listed under `pruned` with the reason. `pruned_candidates` lists the
-//     physical candidates Stage 2 pruned in the selected round
+//     physical candidates Stage 2 pruned
 //     (`PhysicalASAPMemoCandidates::pruned_candidates`): each with its memo
 //     `group`, `operator_node`, `operator`, `implementation_root`, `timing`,
 //     `materialization` and `reason` (`PruneReason`:
@@ -61,7 +60,7 @@
 //     rejected ones taking each cause in turn;
 //   - stage3_plan_selection: Stage 3's `Selection`: `selected_dag`
 //     (`DagOfSelectedCandidates`: `candidate_of_node`, the candidate id of
-//     the selected round's Stage 2 output each node runs, one node per run;
+//     Stage 2's output each node runs, one node per run;
 //     `edges`, each `producer` node feeding input `input` of a `consumer`
 //     node; and `node_of_output_reference`),
 //     `nodes_with_proven_non_negative_weights`, the plan's `cost` (its
@@ -97,9 +96,8 @@
 //     and the plan's peak memory when over 10^12 bytes and the cost model
 //     declares no memory capacity (`large_peak_memory_warning`); absent when
 //     there are none;
-//   - planning_report: the facade's `PlanningReport`: Pass 1's time and, per
-//     tumbling-window-width round, Pass 2's and Stage 2's counts and each stage's time,
-//     with Stage 3's search report or error;
+//   - planning_report: the facade's `PlanningReport`: each stage's time,
+//     Pass 2's and Stage 2's counts, and Stage 3's search report;
 //   - reference_pricing (with `--exhaustive`): the reference pricer's
 //     selection among the enumerated plans, named as Stage 3's `Selection`
 //     names its fields: the shown candidates' `cost_per_candidate` and
@@ -114,7 +112,7 @@
 //     `pruned_candidates_not_enumerated` (rejections naming no enumerated
 //     candidate) and `pruned_candidate_count_by_cause`.
 // `memo` (unless `--no-memo`) is the memo the search built, from the
-// planning round whose plan was selected: `nodes`, one node
+// planning run: `nodes`, one node
 // table for the whole section (`MemoNodes`: each distinct node once, matched
 // by its serialized JSON, children as indices into the table and before
 // their parents); `steps`, one per
@@ -315,7 +313,7 @@ const USAGE: &str =
 [--name <name>]... [--table <json>...] [--max-error <f64> --failure-probability <f64> --accuracy-metric <metric>] [--sample-interval-ms <u64>] [--series-count <u64>] [--rows-per-second <f64>] [--rows-held <u64>])) \
 [--cost-model usage-over-time|aws-fargate-us-east-1|provisioned-capacity|cluster-capacity [--cluster-memory-bytes <n> \
 --cluster-disk-bytes <n> [--cluster-cpu-cores <n>] --cluster-objective lowest-latency|least-memory]] \
-[--cost-inputs <file>] [--enforce-accuracy-guarantees] [--deployment-capabilities <json>] [--exhaustive [--max-combinations <n>]] [--milp-time-limit-s <f64>] [--max-displayed <n>] \
+[--cost-inputs <file>] [--enforce-accuracy-guarantees] [--deployment-capabilities <json>] [--exhaustive [--max-combinations <n>]] [--max-displayed <n>] \
 [--no-memo] [--max-memo-candidates <n>] [--memo-enumeration-limit <n>] [--max-enumerated <n>] [--price-enumerated <n>] --out <file>";
 
 /// Default evaluator-assignment budget for an exhaustive archive run.
@@ -404,9 +402,6 @@ fn run(args: Vec<String>) -> Result<(), Failure> {
     let mut cost_inputs_path: Option<String> = None;
     let mut enforce_accuracy_guarantees = false;
     let mut exhaustive = false;
-    // `--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<u64> = None;
     let mut max_combinations_given = false;
     let mut write_memo = true;
     let mut memo_caps = MemoCaps {
@@ -489,12 +484,6 @@ fn run(args: Vec<String>) -> Result<(), Failure> {
             "--price-enumerated" => {
                 memo_caps.choices_to_price = value()?.parse().map_err(|e| format!("{e}"))?
             }
-            "--milp-time-limit-s" => {
-                let seconds: f64 = value()?
-                    .parse()
-                    .map_err(|e| format!("--milp-time-limit-s: {e}"))?;
-                milp_time_limit_ms = Some((seconds * 1e3) as u64);
-            }
             other => return Err(format!("unknown argument {other}").into()),
         }
     }
@@ -746,10 +735,7 @@ fn run(args: Vec<String>) -> Result<(), Failure> {
         &data,
         models,
         // The document shows every stage's memo.
-        PlanSearchSettings {
-            time_limit_ms: milp_time_limit_ms,
-            inspect: true,
-        },
+        PlanSearchSettings { inspect: true },
         exhaustive.then_some(max_combinations),
     );
     // A rate, increase or irate over a declared gauge, or an aggregation of
@@ -1407,7 +1393,7 @@ fn accepted_json(queries: &[AcceptedAccuracy]) -> Value {
 }
 
 /// Stage 3's selection ([`Selection`]): the selected DAG over the selected
-/// round's Stage 2 candidates (one node per run of a candidate), the nodes
+/// Stage 2 candidates (one node per run of a candidate), the nodes
 /// whose weights it proved non-negative, the plan's cost, the accuracy accepted per query, the
 /// candidates it pruned and its search report; `error` when the facade
 /// selected no plan.

crates/devtools/src/memo_pricing.rs

@@ -254,7 +254,7 @@ pub struct EnumeratedChoice {
 }
 
 /// `stage_pipeline --exhaustive`: the first `max_choices` choices of the
-/// selected round's Pass 2 memo (`run.logical_asap_candidates`), each composed,
+/// Pass 2 memo (`run.logical_asap_candidates`), each composed,
 /// implemented by Stage 2 with every materialization assignment, and all
 /// their physical plans priced together by the reference pricer.
 #[derive(Debug)]

crates/devtools/src/planning_stages.rs

@@ -21,9 +21,9 @@ pub const DEFAULT_REFERENCE_CHOICES: usize = 10_000;
 pub struct PlanningStagesRun {
     /// The query-language frontends' roots, one per workload entry.
     pub stage0_query_language_frontends: Vec<QueryRoot>,
-    /// The facade's run: Pass 1, the tumbling-window-width rounds and the selection.
+    /// The facade's run: every stage and the selection.
     pub run: PlanningRun,
-    /// `--exhaustive`: the selected round's memo choices, each implemented
+    /// `--exhaustive`: Pass 2's memo choices, each implemented
     /// by Stage 2 and all priced by the reference pricer.
     pub reference: Option<ReferenceEnumeration>,
 }
@@ -188,11 +188,8 @@ pub fn run_planning_stages_over(
         entries,
         data,
         models,
-        // The reference enumeration reads the selected round's memo.
-        PlanSearchSettings {
-            inspect: true,
-            ..PlanSearchSettings::default()
-        },
+        // The reference enumeration reads Pass 2's memo.
+        PlanSearchSettings { inspect: true },
         Some(max_combinations.unwrap_or(DEFAULT_REFERENCE_CHOICES)),
     )
 }

crates/devtools/tests/corpus_snapshots.rs

@@ -15,12 +15,8 @@
 //! and accepted accuracy per query, the materialized views, tumbling-window retention,
 //! and whether the search proved the plan optimal; for a query that does not
 //! plan, the error. Candidate ids, round and evaluation counters, timings, the
-//! Stage 1 and Stage 2 candidate lists and the memo vary from run to run
-//! (round budgets are wall-clock), so they are left out. Floats are rounded
-//! to 12 significant digits. A run that the time limit stopped fails instead
-//! of being recorded, unless the query has a `time_limit_reason` in the
-//! manifest (its plan is then recorded and compared as usual); a query with a
-//! `skip_reason` is skipped.
+//! Stage 1 and Stage 2 candidate lists and the memo are left out. Floats are
+//! rounded to 12 significant digits. A query with a `skip_reason` is skipped.
 //!
 //! The same binary holds the sketch plan snapshots (`sketch_plan_snapshots.rs`),
 //! which reuse these helpers on multi-query workloads with accuracy targets.
@@ -33,10 +29,6 @@ use serde_json::{json, Map, Value};
 #[path = "sketch_plan_snapshots.rs"]
 mod sketch_plan_snapshots;
 
-/// Per query. The slowest kept queries take about 60 s on one core, so this
-/// leaves room for a slower CI runner while still failing a runaway search.
-const TIME_LIMIT_S: &str = "240";
-
 /// Held for the whole of each planner snapshot test, so the corpus and
 /// sketch plan tests do not run at once (`--include-ignored` would otherwise
 /// run them in parallel) and slow each other's searches.
@@ -69,14 +61,13 @@ fn plan(workload: &Value, args: &[String], scratch: &Path) -> Value {
     stage_pipeline(&input_args, scratch)
 }
 
-/// Runs `stage_pipeline` (without the memo section, under the
-/// [`TIME_LIMIT_S`] budget) with `args`, which name its input; panics (a
-/// planner crash) on failure.
+/// Runs `stage_pipeline` (without the memo section) with `args`, which name
+/// its input; panics (a planner crash) on failure.
 fn stage_pipeline(args: &[String], scratch: &Path) -> Value {
     let output = scratch.join("document.json");
     let mut command = Command::new(env!("CARGO_BIN_EXE_stage_pipeline"));
     command
-        .args(["--milp-time-limit-s", TIME_LIMIT_S, "--no-memo"])
+        .arg("--no-memo")
         .args(args)
         .arg("--out")
         .arg(&output);
@@ -89,40 +80,14 @@ fn stage_pipeline(args: &[String], scratch: &Path) -> Value {
     serde_json::from_str(&std::fs::read_to_string(output).unwrap()).unwrap()
 }
 
-/// Whether the wall-clock budget ended the search: planning (Pass 1 and
-/// every tumbling-window-width round) used the whole budget. A round cut short by its
-/// own budget is not visible in the report.
-fn hit_time_limit(document: &Value) -> bool {
-    let report = &document["planning_report"];
-    let Some(limit) = report["time_limit_ms"].as_f64() else {
-        return false;
-    };
-    let rounds: f64 = report["rounds"]
-        .as_array()
-        .into_iter()
-        .flatten()
-        .map(|round| {
-            ["pass2_ms", "stage2_ms", "stage3_ms"]
-                .iter()
-                .filter_map(|key| round[*key].as_f64())
-                .sum::<f64>()
-        })
-        .sum();
-    report["pass1_ms"].as_f64().unwrap_or(0.0) + rounds >= limit
-}
-
-/// 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.
+/// The selected plan's search status (`optimal`), or null without a plan.
 fn search_status(document: &Value) -> &Value {
     &document["stage3_plan_selection"]["search_report"]["status"]
 }
 
-/// Whether the plan is proven optimal: its search says so and every legal
-/// tumbling-window width was searched.
+/// Whether the plan is proven optimal: its search says so.
 fn proven_optimal(document: &Value) -> bool {
     search_status(document) == "optimal"
-        && document["planning_report"]["tumbling_window_widths_pending"] != json!(true)
 }
 
 /// `value` with every non-integer number rounded to 12 significant digits.
@@ -301,15 +266,6 @@ fn corpus_plans_match_snapshots() {
                 &args,
                 &scratch,
             );
-            // A `time_limit_reason` accepts a run the time limit ended; the
-            // plan and cost are still compared with the snapshot.
-            if hit_time_limit(&document) && query.get("time_limit_reason").is_none() {
-                failures.push(format!(
-                    "{name}/{id}: the {TIME_LIMIT_S} s time limit ended the search, \
-                     so its plan depends on machine speed"
-                ));
-                continue;
-            }
             let actual = serde_json::to_string_pretty(&snapshot_of(&document)).unwrap() + "\n";
             let path = dir.join(format!("{id}.json"));
             if update {
@@ -361,7 +317,7 @@ fn snapshot_keeps_only_the_plan() {
             "accepted_accuracy_per_query": [],
             "materialized_views": [],
             "tumbling_window_retention_ms": {"1": 60000},
-            "search_report": {"status": "deadline_reached", "solver_ms": 1},
+            "search_report": {"solver_ms": 1},
         },
     });
     assert_eq!(
@@ -404,36 +360,13 @@ fn snapshot_of_a_failed_query() {
     );
 }
 
-/// Planning that used its whole wall-clock budget is detected; planning
-/// within it, or with no report, is not.
-#[test]
-fn time_limit_detection() {
-    let report = |pass1: f64, stage3: f64| {
-        json!({"planning_report": {
-            "pass1_ms": pass1, "time_limit_ms": 1000,
-            "rounds": [{"pass2_ms": 1.0, "stage2_ms": 1.0, "stage3_ms": stage3}],
-        }})
-    };
-    assert!(hit_time_limit(&report(10.0, 990.0)));
-    assert!(!hit_time_limit(&report(10.0, 12.0)));
-    assert!(!hit_time_limit(
-        &json!({"stage3_plan_selection": {"error": "x"}})
-    ));
-}
-
-/// An optimal search proves the plan optimal only when no legal tumbling-window width
-/// was left unsearched.
+/// Only an `optimal` search status proves the plan optimal.
 #[test]
-fn proven_optimal_needs_every_tumbling_window_width() {
-    let document = |status: &str, pending: bool| {
-        json!({
-            "stage3_plan_selection": {"search_report": {"status": status}},
-            "planning_report": {"tumbling_window_widths_pending": pending},
-        })
-    };
-    assert!(proven_optimal(&document("optimal", false)));
-    assert!(!proven_optimal(&document("optimal", true)));
-    assert!(!proven_optimal(&document("deadline_reached", false)));
+fn proven_optimal_needs_an_optimal_status() {
+    let document =
+        |status: Value| json!({"stage3_plan_selection": {"search_report": {"status": status}}});
+    assert!(proven_optimal(&document(json!("optimal"))));
+    assert!(!proven_optimal(&document(Value::Null)));
 }
 
 /// The failure message names the first differing JSON path.

crates/devtools/tests/corpus_snapshots/o11y_bench/corpus.json

@@ -78,8 +78,6 @@
     {
       "id": "q05",
       "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])"
-,
-      "time_limit_reason": "searching every pane-width round to Complete runs past the 240 s limit; the plan and cost found before it are the same as when planning stopped after 16 s"
     },
     {
       "id": "q06",

crates/devtools/tests/memo_pricing.rs

@@ -215,7 +215,6 @@ fn assert_no_listed_choice_is_cheaper(document: &Value) -> Vec<Value> {
 /// Q2's summary reads miss its 100 ms latency bound, and no valid DAG is
 /// cheaper than the selection.
 #[test]
-#[ignore = "nondeterministic: plans under a wall-clock budget; revisit after Stage 3 design (#1175)"]
 fn example1_no_listed_choice_is_cheaper_than_the_selected_plan() {
     let document = priced_document(&["--example", "planner-layering-1"]);
     let choices = assert_no_listed_choice_is_cheaper(&document);
@@ -233,7 +232,6 @@ fn example1_no_listed_choice_is_cheaper_than_the_selected_plan() {
 /// has a guarantee through `/` and `+` over a KLL rank error or a DDSketch
 /// relative one, so every listed DAG is exact and valid.
 #[test]
-#[ignore = "nondeterministic: plans under a wall-clock budget; revisit after Stage 3 design (#1175)"]
 fn ratio_of_quantiles_no_listed_choice_is_cheaper_than_the_selected_plan() {
     let ignored = assert_no_listed_choice_is_cheaper(&priced_document(&[
         "--workload",

crates/devtools/tests/planner_fuzz.rs

@@ -140,7 +140,6 @@ fn exact_duplicate_of_a_shared_query_holds() {
 /// A bounded worker campaign writes the summary with the design-doc coverage
 /// table and one draft per finding, including cases stopped by its time limit.
 #[test]
-#[ignore = "nondeterministic: plans under a wall-clock budget; revisit after Stage 3 design (#1175)"]
 fn campaign_writes_summary_coverage_and_drafts() {
     use asap_devtools::planner_fuzz::report::{run_campaign, Campaign};
     let out = Path::new(env!("CARGO_TARGET_TMPDIR"))