feat: range evaluations (step, span) in the workload; pane grid, per-step pricing and pane retention

crates/devtools/src/bin/stage_pipeline.rs

@@ -4,6 +4,11 @@
 // planner-layering-3a and planner-layering-3b: #509 Example 3, Patterns A
 // and B; 3a has 972 candidates, all priced, 64 shown by default)
 // cargo run -p asap-devtools --bin stage_pipeline -- \
+//     --workload dashboard.json --out run.json
+// (a `PlanningWorkload` JSON file, e.g. with a repeating query's
+// `"range_evaluation": {"step_ms": 60000, "span_ms": 3600000}`; PromQL, or
+// SQL over `--table` declarations)
+// cargo run -p asap-devtools --bin stage_pipeline -- \
 //     --promql "topk by (job) (10, rate(x[1m]))" --max-error 0.01 --failure-probability 0.001 --out run.json
 // cargo run -p asap-devtools --bin stage_pipeline -- \
 //     --table '{"name": "metrics", "columns": [{"name": "ts", "type": "timestamp"},
@@ -47,7 +52,10 @@
 //     once if materialized, else once per query (or materialized node)
 //     that reads it, so a shared NotMaterialized node is charged once per
 //     query. When selection fails, `selected` is null, with the rejections,
-//     or with `error` for any other failure.
+//     or with `error` for any other failure. `pane_retention_ms` maps each
+//     pane chain the selected plan keeps beyond one evaluation (by node id)
+//     to how far back from the newest evaluation the runtime must keep its
+//     panes: the window plus the longest range-evaluation span reading it.
 // When Pass 1 fails, the document still has the Stage 0 DAG; Stage 1 has
 // `error` and no candidates, and nothing is selected.
 //
@@ -223,6 +231,7 @@ fn document_of(
 
 fn run(args: Vec<String>) -> Result<(), Failure> {
     let mut example = None;
+    let mut workload_path: Option<String> = None;
     // (language, text, name) in workload order.
     let mut queries: Vec<(QueryLanguage, String, Option<String>)> = Vec::new();
     let mut tables = Vec::new();
@@ -242,6 +251,7 @@ fn run(args: Vec<String>) -> Result<(), Failure> {
         let number = |v: String| v.parse::<f64>().map_err(|e| format!("{v}: {e}"));
         match flag.as_str() {
             "--example" => example = Some(value()?),
+            "--workload" => workload_path = Some(value()?),
             "--promql" => queries.push((QueryLanguage::PromQL, value()?, None)),
             "--sql" => queries.push((
                 QueryLanguage::SQL(SqlDialect::DataFusionSQL),
@@ -295,7 +305,7 @@ fn run(args: Vec<String>) -> Result<(), Failure> {
     let has_sql = queries
         .iter()
         .any(|(language, _, _)| *language != QueryLanguage::PromQL);
-    if !tables.is_empty() && !has_sql {
+    if !tables.is_empty() && !has_sql && workload_path.is_none() {
         return Err("--table needs --sql".into());
     }
     let out = out.ok_or("--out is required")?;
@@ -355,7 +365,43 @@ fn run(args: Vec<String>) -> Result<(), Failure> {
         measured_operation_costs,
         ..default_settings
     };
+    let from_file = match &workload_path {
+        Some(path) if example.is_none() && queries.is_empty() => {
+            let text =
+                std::fs::read_to_string(path).map_err(|e| format!("--workload {path}: {e}"))?;
+            let workload: PlanningWorkload =
+                serde_json::from_str(&text).map_err(|e| format!("--workload {path}: {e}"))?;
+            workload
+                .validate()
+                .map_err(|e| format!("--workload {path}: {e}"))?;
+            Some((workload, sql_catalog(&tables)?))
+        }
+        Some(_) => {
+            return Err(
+                "give exactly one of --example, --workload or --promql/--sql queries".into(),
+            )
+        }
+        None => None,
+    };
     let mut document = match (example.as_deref(), queries.is_empty()) {
+        (None, true) if from_file.is_some() => {
+            let (workload, catalog) = from_file.expect("checked");
+            let roots = match workload.query_workload.language {
+                QueryLanguage::PromQL => lower_promql_roots(&workload)?,
+                _ => lower_sql_roots(&workload, &catalog)?,
+            };
+            let entries: Vec<_> = workload.query_workload.entries().collect();
+            let data = workload.data_workload.clone().unwrap_or_default();
+            let result = run_planning_stages_over(
+                roots.clone(),
+                &entries,
+                &data,
+                Some(max_combinations),
+                &plan_selection,
+                &deployment_capabilities,
+            );
+            document_of(workload_queries(&workload), &roots, result, max_displayed)?
+        }
         (Some(example), true) => {
             let (workload, catalog) = match example {
                 "planner-layering-1" => (example1(), SqlCatalog::new()),
@@ -458,7 +504,11 @@ fn run(args: Vec<String>) -> Result<(), Failure> {
             );
             document_of(described, &roots, result, max_displayed)?
         }
-        _ => return Err("give exactly one of --example or --promql/--sql queries".into()),
+        _ => {
+            return Err(
+                "give exactly one of --example, --workload or --promql/--sql queries".into(),
+            )
+        }
     };
     // Read-only: the deployment inputs the run planned with.
     document["deployment_capabilities"] =
@@ -647,6 +697,11 @@ fn plan_selection_json(
         "selected": selection.selected,
         "rejected": rejections_json(&selection.rejected),
         "front": selection.front,
+        "pane_retention_ms": selection
+            .pane_retention_ms
+            .iter()
+            .map(|(node, ms)| (node.to_string(), json!(ms)))
+            .collect::<serde_json::Map<_, _>>(),
     })
 }
 
@@ -699,6 +754,9 @@ fn workload_queries(workload: &PlanningWorkload) -> Vec<Value> {
             {
                 requirements["repeat_interval_ms"] = json!(interval.0);
             }
+            if let Some(range) = entry.range_evaluation {
+                requirements["range_evaluation"] = json!(range);
+            }
             json!({
                 "id": format!("q{}", index + 1),
                 "language": language,
@@ -810,6 +868,7 @@ fn query_batch(
                         predictability: Default::default(),
                         invocations: 1,
                         execute_at: None,
+                        range_evaluation: None,
                     })
                     .collect(),
             ),

crates/devtools/src/candidate_labels.rs

@@ -485,6 +485,7 @@ mod tests {
                 max_latency_ms: None,
             },
             predictability: Predictability::Predictable { known_at: None },
+            range_evaluation: None,
         };
         let workload = PlanningWorkload {
             query_workload: QueryWorkload {
@@ -551,6 +552,7 @@ mod tests {
                             max_latency_ms: None,
                         },
                         predictability: Predictability::Predictable { known_at: None },
+                        range_evaluation: None,
                     }]),
                 },
                 data_workload: Some(shared_data_workload(

crates/devtools/src/lib.rs

@@ -55,6 +55,7 @@ pub fn lower_promql_with_sample_interval(
                 predictability: Predictability::Unknown,
                 invocations: 1,
                 execute_at: None,
+                range_evaluation: None,
             }]),
             repeating_queries: None,
         },

crates/devtools/src/planner_layering_examples.rs

@@ -53,6 +53,7 @@ pub fn example1() -> PlanningWorkload {
             max_latency_ms,
         },
         predictability: Predictability::Predictable { known_at: None },
+        range_evaluation: None,
     };
     PlanningWorkload {
         query_workload: QueryWorkload {
@@ -130,6 +131,7 @@ pub fn example2() -> (PlanningWorkload, asap_frontend_sql::SqlCatalog) {
                             max_latency_ms: None,
                         },
                         predictability: Predictability::Predictable { known_at: None },
+                        range_evaluation: None,
                     })
                     .collect(),
             ),
@@ -239,6 +241,7 @@ pub fn pattern_a(recurrence: PatternARecurrence, arrival: DataArrival) -> Planni
                         predictability: Predictability::AdHoc,
                         invocations: 1,
                         execute_at: Some(TimestampMs(T_MS)),
+                        range_evaluation: None,
                     })
                     .collect(),
             ),
@@ -258,6 +261,7 @@ pub fn pattern_a(recurrence: PatternARecurrence, arrival: DataArrival) -> Planni
                         predictability: Predictability::Predictable {
                             known_at: Some(TimestampMs(T_MS)),
                         },
+                        range_evaluation: None,
                     })
                     .collect(),
             ),
@@ -295,6 +299,7 @@ pub fn pattern_b() -> PlanningWorkload {
                     max_latency_ms: Some(200.0),
                 },
                 predictability: Predictability::Predictable { known_at: None },
+                range_evaluation: None,
             }]),
         },
         data_workload: Some(shared_data_workload(

crates/devtools/src/planning_stages.rs

@@ -173,9 +173,8 @@ pub fn run_planning_stages_over(
             .without_unsupported_summaries(deployment);
     // Count Stage 1 candidates as they are enumerated and stop as soon as
     // the count passes the limit; never truncate.
-    let recurrence: Vec<_> = entries.iter().map(|e| e.recurrence.clone()).collect();
     let mut stage1 = Vec::new();
-    for candidate in stage1_logical_asap_optimization(&per_computation, &recurrence) {
+    for candidate in stage1_logical_asap_optimization(&per_computation, entries) {
         if max_combinations.is_some_and(|limit| stage1.len() == limit) {
             return Err(PlanningStagesError::TooManyCandidates {
                 limit: stage1.len(),

crates/devtools/tests/stage_pipeline.rs

@@ -885,3 +885,67 @@ fn logical_candidate_labels_are_distinct() {
     assert_eq!(labels.len(), 9, "{labels:#?}");
     assert_eq!(distinct.len(), 9, "{labels:#?}");
 }
+
+/// `--workload` plans a `PlanningWorkload` JSON file. A dashboard panel
+/// (`max_over_time(m[5m])` every minute over 1M series) answered as the
+/// last hour at 1-minute steps (`range_evaluation`) is priced as 61
+/// evaluations per request: the kept panes (`P2-m2`, 70.51 cost/s) beat
+/// rebuilding the window at every step (`P1`, 127.08), and the document
+/// reports their retention, 1 h + 5 min. The workload section shows the
+/// range. An invalid range is refused.
+#[test]
+fn workload_file_with_a_range_evaluation_is_planned() {
+    let workload = serde_json::json!({
+        "query_workload": {
+            "language": "promql",
+            "query_batch": null,
+            "repeating_queries": [{
+                "query": "max_over_time(m[5m])",
+                "demand": {"fixed_interval": 60000},
+                "requirements": {},
+                "predictability": {"predictable": {"known_at": null}},
+                "range_evaluation": {"step_ms": 60000, "span_ms": 3600000},
+            }],
+        },
+        "data_workload": {
+            "arrival": {"ingesting": {"rows_per_second": 66666.67}},
+            "sources": {"m": {"series_count": 1000000, "sample_interval": 15000}},
+        },
+    });
+    let path = std::env::temp_dir().join(format!(
+        "stage_pipeline_{}_range_workload.json",
+        std::process::id()
+    ));
+    std::fs::write(&path, workload.to_string()).unwrap();
+    let document = document_of(&[
+        "--workload",
+        path.to_str().unwrap(),
+        "--deployment-capabilities",
+        r#"{"query_time_results_retained": true}"#,
+    ]);
+    assert_eq!(
+        document["workload"]["queries"][0]["requirements"]["range_evaluation"],
+        serde_json::json!({"step_ms": 60000, "span_ms": 3600000})
+    );
+    let selection = &document["stage3_plan_selection"];
+    assert_eq!(selection["selected"], "P2-m2");
+    let cost = |id: &str| selection["costs"][id]["cost_per_second"].as_f64().unwrap();
+    assert!((cost("P2-m2") - 70.5133).abs() < 1e-3, "{}", cost("P2-m2"));
+    assert!((cost("P1") - 127.0833).abs() < 1e-3, "{}", cost("P1"));
+    assert_eq!(
+        selection["pane_retention_ms"],
+        serde_json::json!({"1": 3_900_000})
+    );
+
+    let mut invalid = workload.clone();
+    invalid["query_workload"]["repeating_queries"][0]["range_evaluation"]["step_ms"] =
+        serde_json::json!(0);
+    std::fs::write(&path, invalid.to_string()).unwrap();
+    let (status, stderr) = failure(&["--workload", path.to_str().unwrap()]);
+    assert_ne!(status, Some(0));
+    assert!(
+        stderr.contains("step_ms must be greater than zero"),
+        "{stderr}"
+    );
+    std::fs::remove_file(path).unwrap();
+}

crates/executor/src/physical_planner/candidates.rs

@@ -356,6 +356,7 @@ mod tests {
                     predictability: Predictability::Unknown,
                     invocations: 1,
                     execute_at: None,
+                    range_evaluation: None,
                 }]),
                 repeating_queries: None,
             },

crates/executor/tests/common/mod.rs

@@ -75,6 +75,7 @@ fn run_stage1_to_stage3(
             },
             predictability: Default::default(),
             recurrence: Default::default(),
+            range_evaluation: None,
         }],
         &data,
         asap_plan_selection::PlanningModels::builtin(),

crates/executor/tests/current_series_heap.rs

@@ -318,6 +318,7 @@ fn planner_current_series_candidate_compiles_with_dynamic_identity() {
                 predictability: Predictability::Unknown,
                 invocations: 1,
                 execute_at: None,
+                range_evaluation: None,
             }]),
             repeating_queries: None,
         },

crates/executor/tests/deployment_computation.rs

@@ -33,6 +33,7 @@ fn lower_with(query: &str, accuracy: AccuracyTarget) -> Rc<planner_types::ir::Op
                 predictability: Predictability::Unknown,
                 invocations: 1,
                 execute_at: None,
+                range_evaluation: None,
             }]),
             repeating_queries: None,
         },

crates/executor/tests/planspace_series_identity_heap.rs

@@ -26,6 +26,7 @@ fn lower(query: &str, accuracy: &AccuracyTarget) -> Rc<OperatorNode> {
                 predictability: Predictability::Unknown,
                 invocations: 1,
                 execute_at: None,
+                range_evaluation: None,
             }]),
             repeating_queries: None,
         },

crates/executor/tests/precompute_candidates.rs

@@ -39,6 +39,7 @@ fn exact_promql_root(query: &str) -> std::rc::Rc<planner_types::ir::OperatorNode
                 predictability: Predictability::Unknown,
                 invocations: 1,
                 execute_at: None,
+                range_evaluation: None,
             }]),
             repeating_queries: None,
         },
@@ -632,6 +633,7 @@ fn population_topk_cuts_equal_per_frontier_compilation() {
                 predictability: Predictability::Unknown,
                 invocations: 1,
                 execute_at: None,
+                range_evaluation: None,
             }]),
             repeating_queries: None,
         },

crates/executor/tests/weighted_topk_binding.rs

@@ -196,6 +196,7 @@ pub fn lower_promql(
                 predictability: Predictability::Unknown,
                 invocations: 1,
                 execute_at: None,
+                range_evaluation: None,
             }]),
             repeating_queries: None,
         },

crates/frontend-metricsql/tests/lowering_semantics.rs

@@ -29,6 +29,7 @@ fn workload(language: QueryLanguage, query: &str) -> PlanningWorkload {
                 predictability: Default::default(),
                 invocations: 1,
                 execute_at: None,
+                range_evaluation: None,
             }]),
             repeating_queries: None,
         },

crates/frontend-promql/src/lib.rs

@@ -139,6 +139,7 @@ mod tests {
                     predictability: Default::default(),
                     invocations: 1,
                     execute_at: None,
+                    range_evaluation: None,
                 }]),
                 repeating_queries: None,
             },

crates/frontend-promql/tests/promql_lowering.rs

@@ -1173,6 +1173,7 @@ fn batch_lowers_each_entry_and_reads_per_query_accuracy() {
                     predictability: Predictability::Unknown,
                     invocations: 1,
                     execute_at: None,
+                    range_evaluation: None,
                 },
                 BatchEntry {
                     query: Query("quantile_over_time(0.9, b[5m])".into()),
@@ -1183,6 +1184,7 @@ fn batch_lowers_each_entry_and_reads_per_query_accuracy() {
                     predictability: Predictability::Unknown,
                     invocations: 1,
                     execute_at: None,
+                    range_evaluation: None,
                 },
             ]),
             repeating_queries: None,
@@ -1206,6 +1208,7 @@ fn batch_rejects_non_promql_language() {
                 predictability: Predictability::Unknown,
                 invocations: 1,
                 execute_at: None,
+                range_evaluation: None,
             }]),
             repeating_queries: None,
         },