feat(planner): plan_installable, the one call a deployment runtime makes (#749)

.github/workflows/rust.yml

@@ -62,6 +62,14 @@ jobs:
       env:
         RUSTC_WRAPPER: sccache
 
+    # A deployment runtime uses only the facade's and the executor's public
+    # APIs. asap-planner's tests check that, but only when the executor links
+    # without its `test-support` feature, which a workspace build turns on.
+    - name: Check the runtime tests against the public APIs alone
+      run: cargo check -p asap-planner --tests --locked
+      env:
+        RUSTC_WRAPPER: sccache
+
   test:
     if: github.event_name != 'pull_request' || github.event.pull_request.draft == false
     runs-on: ubuntu-latest

crates/devtools/src/bin/stage_pipeline.rs

@@ -111,6 +111,13 @@
 // writes from sketch-bench's `rqe_atomic_costs.json`); a sketch node with an
 // entry is priced from it, and each node's `price_source` says `measured`
 // or `analytical`.
+// `--plan-selection-settings <file>` reads all of Stage 3's settings from
+// one JSON file (`PlanSelectionSettings`: `cost_model`, e.g.
+// `{"usage_over_time": {"cost_per_cpu_op": 1e-6, ...}}` with the
+// deployment's own unit costs, `resource_speed`,
+// `non_repeating_query_horizon_seconds` and `measured_operation_costs`;
+// an omitted field takes its default). It replaces `--cost-model`, the
+// `--cluster-*` flags and `--measured-operation-costs`.
 // `--sample-interval-ms` is the sample interval declared for every metric the
 // PromQL queries read (default 15000); the data declares no other facts.
 // `--table '<json>'` (repeatable) declares a table the SQL queries read: `{"name": ...,
@@ -154,7 +161,7 @@ const USAGE: &str =
 [--name <name>]... [--table <json>...] [--max-error <f64> --failure-probability <f64> --accuracy-metric <metric>] [--sample-interval-ms <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|pareto-front]] \
-[--deployment-capabilities <json>] [--max-combinations <n>] [--max-displayed <n>] --out <file>";
+[--plan-selection-settings <file>] [--deployment-capabilities <json>] [--max-combinations <n>] [--max-displayed <n>] --out <file>";
 
 /// The default `--max-combinations`: the most Stage 1 candidates (Pass 2's
 /// included) a run enumerates.
@@ -245,6 +252,7 @@ fn run(args: Vec<String>) -> Result<(), Failure> {
     let mut deployment_capabilities = DeploymentCapabilities::default();
     let mut out = None;
     let mut measured_operation_costs_path: Option<String> = None;
+    let mut plan_selection_settings_path: Option<String> = None;
     let mut args = args.into_iter();
     while let Some(flag) = args.next() {
         let mut value = || args.next().ok_or(format!("{flag} needs a value"));
@@ -299,6 +307,7 @@ fn run(args: Vec<String>) -> Result<(), Failure> {
             }
             "--out" => out = Some(value()?),
             "--measured-operation-costs" => measured_operation_costs_path = Some(value()?),
+            "--plan-selection-settings" => plan_selection_settings_path = Some(value()?),
             other => return Err(format!("unknown argument {other}").into()),
         }
     }
@@ -309,6 +318,20 @@ fn run(args: Vec<String>) -> Result<(), Failure> {
         return Err("--table needs --sql".into());
     }
     let out = out.ok_or("--out is required")?;
+    if plan_selection_settings_path.is_some()
+        && (cost_model.is_some()
+            || cluster_memory.is_some()
+            || cluster_disk.is_some()
+            || cluster_cores.is_some()
+            || cluster_objective.is_some()
+            || measured_operation_costs_path.is_some())
+    {
+        return Err(
+            "--plan-selection-settings replaces --cost-model, the --cluster-* flags \
+                    and --measured-operation-costs"
+                .into(),
+        );
+    }
     let cluster = (cluster_memory, cluster_disk, cluster_objective.as_deref());
     // The settings every flag below leaves at its default; the Fargate
     // preset converts its prices at their `resource_speed`.
@@ -360,10 +383,18 @@ fn run(args: Vec<String>) -> Result<(), Failure> {
         }
         None => Vec::new(),
     };
-    let plan_selection = PlanSelectionSettings {
-        cost_model,
-        measured_operation_costs,
-        ..default_settings
+    let plan_selection = match &plan_selection_settings_path {
+        Some(path) => {
+            let text = std::fs::read_to_string(path)
+                .map_err(|e| format!("--plan-selection-settings {path}: {e}"))?;
+            serde_json::from_str(&text)
+                .map_err(|e| format!("--plan-selection-settings {path}: {e}"))?
+        }
+        None => PlanSelectionSettings {
+            cost_model,
+            measured_operation_costs,
+            ..default_settings
+        },
     };
     let from_file = match &workload_path {
         Some(path) if example.is_none() && queries.is_empty() => {

crates/devtools/tests/stage_pipeline.rs

@@ -949,3 +949,42 @@ fn workload_file_with_a_range_evaluation_is_planned() {
     );
     std::fs::remove_file(path).unwrap();
 }
+
+/// `--plan-selection-settings <file>` loads Stage 3's settings from JSON:
+/// cost model 2 with the illustrative prices selects what
+/// `--cost-model provisioned-capacity` selects for Example 4 (P1). The file
+/// replaces the cost-model flags, so combining them is refused.
+#[test]
+fn plan_selection_settings_flag_loads_the_settings_from_json() {
+    let path = std::env::temp_dir().join(format!(
+        "stage_pipeline_{}_plan_selection_settings.json",
+        std::process::id()
+    ));
+    let settings = asap_plan_selection::PlanSelectionSettings {
+        cost_model: asap_plan_selection::CostModel::ProvisionedCapacity(
+            asap_plan_selection::ResourceUnitCost::illustrative(),
+        ),
+        ..Default::default()
+    };
+    std::fs::write(&path, serde_json::to_string(&settings).unwrap()).unwrap();
+    let path_arg = path.to_str().unwrap();
+    let document = generate_example(
+        "planner-layering-3b",
+        &["--plan-selection-settings", path_arg],
+    );
+    assert_eq!(document["stage3_plan_selection"]["selected"], "P1");
+    let (code, stderr) = failure(&[
+        "--example",
+        "planner-layering-3b",
+        "--plan-selection-settings",
+        path_arg,
+        "--cost-model",
+        "usage-over-time",
+    ]);
+    assert_eq!(code, Some(2));
+    assert!(
+        stderr.contains("--plan-selection-settings replaces --cost-model"),
+        "{stderr}"
+    );
+    std::fs::remove_file(path).unwrap();
+}

crates/executor/Cargo.toml

@@ -15,7 +15,13 @@ thiserror = "2"
 regex = "1"
 
 
+[features]
+# Executor internals the Planner's own tests use (the PromQL row encoding,
+# the timing frontier); a deployment runtime does not need them.
+test-support = []
+
 [dev-dependencies]
+asap-executor = { path = ".", features = ["test-support"] }
 rmp-serde = "1"
 asap-logical-asap-optimization = { path = "../logical-asap-optimization" }
 asap-plan-selection = { path = "../plan-selection" }

crates/executor/src/lib.rs

@@ -13,7 +13,11 @@ pub use traits::*;
 
 pub use expressions::arithmetic;
 pub mod capability;
+mod samples;
+pub use samples::{series_labels_of_output_row, RawSampleSelector, Sample};
+mod supported_summaries;
 pub use summary_kernels::factory;
+pub use supported_summaries::supported_summary_families;
 
 /// The exact Planner contract used by these kernels.
 pub use planner_types as planner;

crates/executor/src/physical_planner/candidates.rs

@@ -96,6 +96,7 @@ pub fn cut_candidate(
 /// A query-time node feeding an ingestion-time node has no valid placement.
 /// Only ingestion timing is read: a `MatAtQuery` output is not stored by
 /// this frontier yet (choosing and storing it is Stage 2's job).
+#[cfg(any(test, feature = "test-support"))]
 pub fn frontier_from_timing(dag: &PhysicalASAPDAG) -> Result<Vec<NodeId>, Error> {
     use planner_types::ir::properties::ExecutionTiming::IngestionTime;
     let timing = dag

crates/executor/src/physical_planner/mod.rs

@@ -19,6 +19,9 @@ use planner_types::ir::scalar::{ColumnRef, CompareOpKind};
 use planner_types::ir::schema::{
     DataType, FamilyCapability, FieldDataType, SketchStatistic, SummaryInputExpr,
 };
+// `promql_rows` reads `LogicalOperator` and `OperatorNode` through `super::*`
+// only in its test-support helpers.
+#[cfg_attr(not(any(test, feature = "test-support")), allow(unused_imports))]
 use planner_types::ir::{ASAPOp, NonASAPOp, Operator as LogicalOperator, OperatorNode, ScalarExpr};
 mod logical;
 use std::{
@@ -35,14 +38,22 @@ fn invalid(message: impl Into<String>) -> Error {
 pub type Source<'a> = Box<dyn PhysicalOperator<Batch, SchemaRef> + 'a>;
 
 pub mod precompute;
+// The executor's PromQL row encoding and the timing frontier are internals:
+// a deployment runtime feeds samples with `crate::RawSampleSelector`
+// and reads the stored outputs from the installable plan. The Planner's own
+// tests reach them with the `test-support` feature.
+#[cfg(feature = "test-support")]
 pub mod promql_rows;
+#[cfg(not(feature = "test-support"))]
+pub(crate) mod promql_rows;
 pub mod promql_values;
 
 mod candidates;
+#[cfg(feature = "test-support")]
+pub use candidates::frontier_from_timing;
 pub use candidates::{
-    compile_candidate, compile_candidates, cut_candidate, enumerate_frontiers,
-    frontier_from_timing, select_candidate, CandidateCost, CandidateSelection,
-    CompiledPhysicalPlan,
+    compile_candidate, compile_candidates, cut_candidate, enumerate_frontiers, select_candidate,
+    CandidateCost, CandidateSelection, CompiledPhysicalPlan,
 };
 
 mod compiled;

crates/executor/src/physical_planner/promql_rows.rs

@@ -1,5 +1,10 @@
 //! A bounded PromQL source row carries the entire label set, not just labels
 //! mentioned by the query. The source adapter owns this lossless encoding.
+// Some helpers here serve only the Planner's own tests (`test-support`).
+#![cfg_attr(
+    not(any(test, feature = "test-support")),
+    allow(dead_code, unused_imports)
+)]
 use super::*;
 use planner_types::ir::physical_export::{
     compile_physical_asap_dag, compile_physical_asap_dag_with_node_ids,