feat(planner)!: Stage 2 materialization, the latency check and simplified query requirements (Example 4)

[zzylol]

Supersedes #604 and #606. Group G8 of the ASAPPlanner consolidation (#509 Example 4, Stage 2 materialization). Also simplifies the per-query requirements (QueryRequirements) and passes the workload entries straight into planning.

Why

• Materialization. Stage 2 (physical ASAP-aware optimization) ran every summary build at query time. So Stage 3 (plan selection) never compared keeping state from ingestion time with rebuilding it at every evaluation, or building a shared summary once per evaluation with building it once per query. Example 4's B1 (tumbling KLLs maintained at ingestion time) was never generated, and latency bounds were never checked.
• Requirements. QueryRequirements wrapped its two fields in enums: AccuracyRequirement { Explicit(target), ImplicitExact } and LatencyRequirement { ExplicitMaxMs(ms), Unspecified }. Both say "a value or the default", which AccuracyTarget (default exact) and Option<f64> already say. Planning then took the accuracy targets and the recurrences as two separate lists beside the roots.

What

• Stage 2 enumerates materialization. Each summary build (SummaryAgg or PaneChain) is one materialization candidate (MaterializationCandidate), with up to three options:

  • NotMaterialized;
  • MatAtIngestion(Memory), when the build is eligible (S4);
  • MatAtQuery(Memory), when two or more queries read it. The state is reused within one evaluation only; keeping it across evaluations (B3) and the deployment capability gate are G9.

• optimize_physical_asap returns a Vec: one physical candidate per assignment, the all-NotMaterialized one first. An assignment that puts a build at ingestion time over a build that is not is not generated (docs: propose workload-wide planning, summary sharing, and materialization #509 Algorithm 3, line 13). The enumeration is uncapped. Ids are P<n>, then P<n>-m<k>. Each candidate carries its assignment (materialization: each summary build's option, by DAG node id).

• Candidates hold planning data only. PhysicalCandidate, LogicalCandidate, Stage1Candidate and PrunedCandidate no longer have display_name. The viewer labels (e.g. "Q1 exact · Q2 CMS+heap · ingestion time: Kll ×5 panes") are pure functions in asap_devtools::candidate_labels, written into the stage_pipeline document only.

• Ingestion-time eligibility (S4). A build may run at ingestion time only when all of these hold:

  • the data is continuously ingesting or mixed (unknown arrival counts as not ingesting);
  • every query that reads it is repeated and Predictable;
  • it has no filter;
  • if it is a counter state (Rate, Increase, IRate): the data is continuously ingesting and every source below it is a time series (PromQL). Assumption: continuously ingesting PromQL data arrives in time order per series, as Prometheus assumes (it rejects out-of-order samples). Table (SQL) sources, mixed data (whose at-rest part may load in any order) and unknown arrival stay ineligible;
  • it is a pane chain, or a whole-window build whose readers share one FixedIntervalAt (a known phase), over an unshifted window no longer than the interval;
  • everything below it can run at ingestion time.

  Each refusal is an IngestionPruneReason. That type and MaterializationCandidate are crate-private Stage 2 bookkeeping. On the planner side, the timing pass now rejects a filtered build placed at ingestion time (FilteredSummaryAtIngestion), which matches the executor's check.

• Cross-query phase split. split_shared_by_phase now takes every root of the workload. A node that one query reads at ingestion time and another at query time is copied, one copy per phase.

• MatAtQuery memory. A MatAtQuery state is reused only within one evaluation, so it is charged only for the time it is held: its bytes × the latency of the longest query reading it, once per evaluation (close to nothing). It is not charged retained byte-seconds over the interval. Keeping state across evaluations (B3) is G9 and will be charged by its real retention.

• Pane chains at ingestion time. The planner prices a chain as a chain: only the newest pane pays the update, over one second of arriving rows, and the chain keeps N + 1 panes (bytes × seconds). expand_pane_chains now carries the chain's timing and materialization onto its panes. Before this PR it dropped them.

• Latency check (S6). This follows sketch-bench docs/rqe_optimizer_cost_model.md §5. A query's latency is its longest chain: the compaction c_D of the newest window of an ingestion-time state it reads, plus its own query-time work ℓ, each on a full core, with elastic CPU (no queueing). A candidate with any query over its max_latency_ms is invalid, and the reason gives the latency. The milliseconds come from two new ResourceCalibration fields, ms_per_cpu_op and ms_per_scan_byte (1 ns and 0.1 ns in the illustrative calibration). G9 makes them deployment-suppliable. CandidateCost::latency reports every query's latency.

• QueryRequirements { accuracy: AccuracyTarget, max_latency_ms: Option<f64> }. AccuracyRequirement and LatencyRequirement are removed. JSON: an omitted accuracy is AccuracyTarget::exact(), an omitted max_latency_ms imposes no bound, unknown fields are rejected, and the old tagged forms ("implicit_exact", {"explicit": …}, response_latency) are not accepted. There is no compatibility path.

• Planning takes the workload entries. optimize_physical_asap, physical_candidate_count, select_plan, run_stage1_to_stage3 (and _from_pass1), run_stage2_and_stage3 and run_planning_stages_over take &[QueryWorkloadEntry], one per query in root order. They read the accuracy target, latency bound, recurrence and predictability from it. The separate targets / recurrence parameters are removed. Every query now has a target (default exact), so "a query without a target" no longer exists. QueryWorkloadEntry, Query and QueryRecurrence derive Default.

• Devtool. The stage_pipeline 10^4 limit now also counts Stage 2's physical candidates, before it builds them. Its document's per-query latency_ms is now max_latency_ms.

• Docs. The PR changes four Markdown files: docs/user_guide_docs/stage-viewer.md (the Example 1 and 3b rows), docs/design_docs/proposals/planner-layering-example4-acceptance.md (test statuses and the resolved open question), docs/design_docs/architecture/input-output-workflow.md and docs/develop_docs/library-api.md (the new QueryRequirements).

How

• stage2_physical_asap_optimization.rs:
  • materialization_candidates finds the builds and their readers, then checks the S4 rules against the readers' entries. As a final check, it applies timings with only the build and the builds below it at ingestion time.
  • materialization_assignments walks the mixed-radix product of the options and skips assignments that are not down-closed.
  • For each assignment, the roots are split by phase (all roots together) and then timed with one memo. The result is exported.
• stage3_plan_selection.rs:
  • select_plan takes each entry's accuracy target, recurrence and max_latency_ms. The internal per-query targets are no longer Option.
  • price records each node's work in milliseconds. That is one computation at query time, or one second of arriving rows at ingestion time.
  • query_latencies sums the query-time work each query reaches, counting each node once. It adds the largest compaction among the ingestion-time states the query reads: (k − 1) · I · ms_per_cpu_op, where k = ⌈ingest ms per s / 1000⌉, at least 1, and I is the number of states per window.
  • Kept memory: a MatAtIngestion chain keeps state / N · (N + 1) and a whole window 2 × state, every second. A MatAtQuery node records (state, evaluations per second) while priced. Once the latencies are known, price adds cost_per_retained_byte_second × state × (longest reader's latency) × evaluations per second. Each query's reach (its walk, stopping at ingestion-time nodes) is computed once.
  • select_plan rejects a candidate whose latency is over a bound before recording its cost.
• pass/stage1_to_stage3.rs: the driver passes the entries to Stage 2 and Stage 3, and the Stage 1 recurrences come from them.
• devtools/candidate_labels.rs: candidate_labels(&PlanningStagesRun) derives each L<n> label from the Pass 1 choice (choice_label) and the Pass 2 rules applied (pass2_label, including feat(planner): Pass 2 reuse rules (rollup, exact top-k reuse) and nested sum merge in Pass 1 (#680) #716's reuse rules). Each physical label is its logical label plus the non-NotMaterialized entries of its assignment (physical_label). PlanningStagesRun keeps the Pass 1 inventory and the Pass 2 records they read.

Key code interfaces added / modified

crates/types/src/workload.rs

/// Independent accuracy and response-latency constraints attached to one query in the workload.
/// JSON: {"accuracy": {"error": 0.01, "metric": "rank"}, "max_latency_ms": 200.0}; both fields
/// may be omitted; unknown fields and the old tagged forms are rejected. The type stays.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct QueryRequirements {
    /// The accuracy the query's answer must meet; omitted: AccuracyTarget::exact().
    #[serde(default = "AccuracyTarget::exact")]
    pub accuracy: AccuracyTarget,   // was AccuracyRequirement
    /// The longest one evaluation may take, in ms; None (the default): no bound.
    /// Stage 3 rejects a candidate whose latency is above it.
    #[serde(default)]
    pub max_latency_ms: Option<f64>, // new; was response_latency: LatencyRequirement
}

// removed: pub enum AccuracyRequirement { Explicit(AccuracyTarget), ImplicitExact }
//          (an AccuracyTarget, default exact, says the same)
// removed: pub enum LatencyRequirement { ExplicitMaxMs(f64), Unspecified }
//          (an Option<f64> says the same)
//          and its From<&QueryWorkloadEntry> / From<QueryRecurrence> impls
//          (planning reads the QueryWorkloadEntry itself)

/// Planner-facing normalized form of a BatchEntry or RepeatingEntry; planning takes one per
/// query, in root order. Stage 2 reads recurrence and predictability, Stage 3 the accuracy
/// target, the latency bound and the recurrence. The type stays; it now derives Default
/// (empty query, exact, no latency bound, unknown predictability and recurrence).
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)] // Default: new
#[serde(deny_unknown_fields)]
pub struct QueryWorkloadEntry {
    /// The query text in the workload's language.
    pub query: Query,                       // unchanged
    /// Its accuracy target and latency bound (default: exact, no bound).
    pub requirements: QueryRequirements,    // unchanged (the type changed, above)
    /// How far ahead it is known: a build may run at ingestion time only for Predictable queries.
    pub predictability: Predictability,     // unchanged
    /// How often it runs: sets how often each node is priced; only repeated queries may keep
    /// state from ingestion time.
    pub recurrence: QueryRecurrence,        // unchanged
    /// The event time it reads.
    pub time_selection: TimeSelection,      // unchanged
}

/// A raw query string; derives Default (the empty string) for QueryWorkloadEntry's Default.
#[derive(Debug, Clone, Default, PartialEq, Eq, Hash, Serialize, Deserialize)] // Default: new
pub struct Query(pub String);

/// How often a query runs; derives Default (Unknown) for QueryWorkloadEntry's Default.
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)] // Default: new
pub enum QueryRecurrence {
    /// A finite batch: `invocations` runs, at `execute_at` when known.
    OneTime { invocations: u64, execute_at: Option<TimestampMs> }, // unchanged
    /// Repeated demand (fixed interval, with or without a phase, scheduled, estimated rate).
    Repeated(RepeatedDemand),                                       // unchanged
    /// Not known; the default.
    #[default]
    Unknown,                                                        // unchanged
}

crates/asap-aware-mapping/src/stage2_physical_asap_optimization.rs

/// One CandidatePhysicalASAPDAG (Stage 2 output), derived from exactly one
/// CandidateLogicalASAPDAG. Planning data only; display labels are derived by the devtools.
/// The type stays.
#[derive(Debug, Clone)]
pub struct PhysicalCandidate {
    /// `P<n>` (nothing materialized) or `P<n>-m<k>`, set by the caller.
    pub id: String,                       // unchanged
    /// The id of the logical candidate it implements.
    pub from_logical: String,             // unchanged
    /// The timed query roots, one per query in workload order, that `dag` exports.
    pub roots: Vec<QueryRoot>,            // unchanged
    /// The exported physical DAG Stage 3 prices.
    pub dag: PhysicalASAPDAG,             // unchanged
    /// Its materialization assignment: the option chosen for each summary build
    /// (SummaryAgg or PaneChain) of `dag`, by node id, NotMaterialized ones included.
    pub materialization: BTreeMap<PhysicalASAPNodeId, Materialization>, // new
    // removed: display_name (now asap_devtools::candidate_labels)
}
impl PhysicalCandidate {
    /// Whether any summary build is materialized (false for `P<n>`).
    pub fn materializes_any(&self) -> bool; // new
}

/// Stage 2 for one logical candidate: implement every operator, then one physical candidate
/// per materialization assignment, all-NotMaterialized first. `entries[i]` is query i's
/// workload entry (a missing one is unknown). Err only when an operator has no
/// implementation (exact top-k): the caller prunes that candidate.
pub fn optimize_physical_asap(
    from_logical: &str,
    roots: &[QueryRoot],
    entries: &[QueryWorkloadEntry], // new
    data: &DataWorkload,            // new
) -> Result<Vec<PhysicalCandidate>, PhysicalASAPOptimizationError>; // was Result<PhysicalCandidate, _>

/// How many candidates `optimize_physical_asap` would return, counted without building
/// them; the devtool's candidate limit reads it.
pub fn physical_candidate_count(roots: &[QueryRoot], entries: &[QueryWorkloadEntry], data: &DataWorkload)
    -> Result<usize, PhysicalASAPOptimizationError>; // new

/// Crate-private Stage 2 bookkeeping: one summary build Stage 2 may store, with its
/// materialization options. Not a whole-workload plan like PhysicalCandidate: the physical
/// candidates are the combinations of these options (the doc comment has an example).
#[derive(Debug, Clone)]
pub(crate) struct MaterializationCandidate { // new
    /// The build in the implemented DAG.
    pub(crate) node: Rc<OperatorNode>,
    /// The queries (indices into the roots) that read it.
    pub(crate) readers: Vec<usize>,
    /// `Ok` when `MatAtIngestion` is an option.
    pub(crate) at_ingestion: Result<(), IngestionPruneReason>,
}
impl MaterializationCandidate {
    /// Its options, NotMaterialized first: MatAtIngestion when eligible, MatAtQuery when two
    /// or more queries read it.
    pub(crate) fn options(&self) -> Vec<Materialization>;
}

/// Crate-private: why a summary build cannot be materialized at ingestion time (S4).
/// These are workload facts, not errors: the build keeps its query-time options.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum IngestionPruneReason { // new
    /// The data workload is at rest, or its arrival is unknown.
    NotIngesting,
    /// A query reading it is not repeated.
    NotRepeated,
    /// A query reading it is not predictable.
    NotPredictable,
    /// It has a filter, which only a query-time build can apply.
    Filtered,
    /// It is a counter state, whose update needs each series' samples in time order, over
    /// data not taken to arrive in that order: only continuously ingesting time-series
    /// (PromQL) sources are.
    CounterNeedsTimeOrder,
    /// A whole-window build whose readers do not all run at one fixed interval with a known
    /// phase, or whose window is longer than that interval, shifted, or not a range.
    WindowNotFixed,
    /// Something below it can only run at query time (carries the timing error).
    UpstreamAtQueryTime(String),
}

crates/asap-aware-mapping/src/stage3_plan_selection.rs, analytical_cost.rs

/// Stage 3: reject candidates that miss a query's accuracy target or latency bound, price the
/// rest per second, select the cheapest. `entries[i]` is the workload entry of root i.
/// A MatAtQuery state is charged only while one evaluation holds it.
pub fn select_plan(
    cands: &[PhysicalCandidate],
    entries: &[QueryWorkloadEntry], // was targets: &[Option<AccuracyTarget>], recurrence: &[QueryRecurrence]
    data: &DataWorkload,
    models: PlanningModels<'_>,
) -> Result<Selection, SelectionError>;

pub enum SelectionError {
    /// Bad input: a different number of workload entries than the candidate has query roots.
    EntryCount { candidate: String, roots: usize, entries: usize }, // was TargetCount { candidate, roots, targets, recurrences }
    /// Internal: a node could not be priced.
    Cost { candidate: String, node: PhysicalASAPNodeId, error: AnalyticalCostError }, // unchanged
    /// Unsatisfiable request: every candidate was rejected (the rejections say why).
    NoValidCandidate(Vec<Rejection>),                                                   // unchanged
    /// Internal: a candidate's summaries could not be resized for its failure probabilities.
    Sizing { candidate: String, error: SizingError },                                   // unchanged
}

/// Whole-workload cost of one valid candidate. The type stays.
pub struct CandidateCost {
    /// The sum of `per_node`, per second.
    pub total: f64,                                       // unchanged
    /// The cost unit, "cost_per_second".
    pub unit: &'static str,                               // unchanged
    /// Which model priced it.
    pub source: &'static str,                             // unchanged
    /// Each node's cost, multiplier, rate and detail.
    pub per_node: BTreeMap<PhysicalASAPNodeId, NodeCost>, // unchanged
    /// One per query, in workload order: its latency per evaluation.
    pub latency: Vec<QueryLatency>,                       // new
}

/// One query's latency per evaluation (sketch-bench cost model §5): its longest chain of
/// work, each step on a full core.
pub struct QueryLatency { // new
    /// Ms to merge the ingest workers' partial copies of the newest window of the largest
    /// ingestion-time state the query reads; 0 when ingestion fits on one core or it reads none.
    pub compaction_ms: f64,
    /// Ms of the query-time work the query reaches, each node once.
    pub query_ms: f64,
}
impl QueryLatency {
    /// compaction_ms + query_ms: what is checked against the query's max_latency_ms.
    pub fn total_ms(&self) -> f64;
}

/// Conversion from physical dimensions to one deployment-specific objective. The type stays.
pub struct ResourceCalibration {
    /// Cost of one CPU operation.
    pub cost_per_cpu_op: f64,               // unchanged
    /// Cost of scanning one byte.
    pub cost_per_scan_byte: f64,            // unchanged
    /// Cost per retained byte (cost model 2; plan selection ignores it).
    pub cost_per_retained_byte: f64,        // unchanged
    /// Cost of one byte kept for one second.
    pub cost_per_retained_byte_second: f64, // unchanged
    /// Seconds over which one-off and unknown queries are spread (default 3600).
    pub horizon_s: f64,                     // unchanged
    /// Ms one CPU operation takes on one core, for the latency check; serde default 0
    /// (work takes no time). Illustrative: 1e-6.
    pub ms_per_cpu_op: f64,                 // new
    /// Ms scanning one byte takes on one core; serde default 0. Illustrative: 1e-7.
    pub ms_per_scan_byte: f64,              // new
    /// Calibration version string.
    pub version: String,                    // unchanged
}

crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs

/// Stage 1 → 2 → 3 over the frontends' roots; `entries[i]` is root i's workload entry.
pub fn run_stage1_to_stage3<Id: Clone + Debug>(
    roots: Vec<(Id, QueryRoot)>,
    entries: &[QueryWorkloadEntry], // was targets: &[Option<AccuracyTarget>], recurrence: &[QueryRecurrence]
    data: &DataWorkload,
    models: PlanningModels<'_>,
) -> Result<Stage1ToStage3Run<Id>, PerComputationCandidateError>;
/// The same after Pass 1, so a caller can read the inventory first.
pub fn run_stage1_to_stage3_from_pass1<Id>(per_computation: PerComputationCandidates<Id>,
    entries: &[QueryWorkloadEntry], // was targets, recurrence
    data: &DataWorkload, models: PlanningModels<'_>) -> Stage1ToStage3Run<Id>;
/// Stage 2 and Stage 3 over Stage 1 candidates a caller enumerated.
pub fn run_stage2_and_stage3<Id>(per_computation: PerComputationCandidates<Id>, stage1: Vec<Stage1Candidate>,
    entries: &[QueryWorkloadEntry], // was targets, recurrence
    data: &DataWorkload, models: PlanningModels<'_>) -> Stage1ToStage3Run<Id>;

/// One whole-workload CandidateLogicalASAPDAG. The type stays.
pub struct LogicalCandidate {
    /// `L<n>`, numbered in enumeration order.
    pub id: String,                                                  // unchanged
    /// The Pass 1 choice: targets[t].choices.candidates[choice[t]].
    pub choice: Vec<usize>,                                          // unchanged
    /// The composed roots, or why the choice did not compose.
    pub roots: Result<Vec<QueryRoot>, PerComputationCandidateError>, // unchanged
    // removed: display_name
}
/// One Stage 1 candidate before Stage 2 numbers it. The type stays.
pub struct Stage1Candidate {
    /// The Pass 1 choice; a Pass 2 candidate has its Pass 1 candidate's.
    pub choice: Vec<usize>,                                          // unchanged
    /// The composed roots, or why they did not compose.
    pub roots: Result<Vec<QueryRoot>, PerComputationCandidateError>, // unchanged
    /// For a Pass 2 candidate: what it shares of which Pass 1 candidate.
    pub pass2_of: Option<Pass2Of>,                                   // unchanged
    // removed: display_name
}
/// A logical candidate Stage 2 could not implement. The type stays.
pub struct PrunedCandidate {
    /// The logical candidate's id.
    pub from_logical: String, // unchanged
    /// Why (e.g. exact top-k has no implementation).
    pub reason: String,       // unchanged
    // removed: display_name
}

crates/types/src/ir/timing.rs, post_asap/execution_data_state.rs, ir/pane_chain.rs

/// Copy, one copy per timing, every sub-DAG that two roots (or two paths of one root) reach
/// at two timings; returns the rewritten roots in order and the assignment remapped onto them.
pub fn split_shared_by_phase(
    roots: &[Rc<OperatorNode>],              // was root: &Rc<OperatorNode>
    assignment: &MaterializationAssignment,
) -> (Vec<Rc<OperatorNode>>, MaterializationAssignment); // was (Rc<OperatorNode>, _)

/// Whether a node is a SummaryAgg or PaneChain with a filter, which never runs at ingestion time.
pub fn is_filtered_build(node: &OperatorNode) -> bool; // new

/// Why the materialization timing pass rejects a plan. The type stays; one variant is new,
/// every other is unchanged.
pub enum ExecutionDataStateError {
    /// Internal: a malformed maintained population.
    InvalidMaintainedPopulation,                                         // unchanged
    /// Invalid plan: a query-time value placed beneath a maintained summary.
    EvaluationUnderMaintenance { edge: &'static str, child: ExecutionDataState }, // unchanged
    /// Invalid plan: an edge whose child data state the parent does not accept.
    IllegalChildDataState { edge: &'static str, child: ExecutionDataState },      // unchanged
    /// Invalid plan: a SummaryAgg over non-accumulator summary state.
    UnsupportedStateComposition { family: String },                      // unchanged
    /// Invalid plan: one shared node given two timings by its consumers.
    ConflictingTiming { first: ExecutionDataState, second: ExecutionDataState }, // unchanged
    /// Invalid plan: an ingestion-time root that is not stored.
    UnstoredIngestionRoot { operator: &'static str },                    // unchanged
    /// Invalid plan: an explicit materialization against a builder-fixed timing.
    MaterializationConflictsWithFixedTiming { operator: &'static str, fixed: ExecutionTiming,
        materialization: Materialization },                              // unchanged
    /// Invalid plan: an explicit MatAtIngestion (or a fixed ingestion timing) on a filtered
    /// summary build. The all-at-ingestion default leaves such a build at query time.
    FilteredSummaryAtIngestion { operator: &'static str },               // new
    /// Internal: a malformed maintenance binary operator.
    InvalidMaintenanceBinary,                                            // unchanged
    /// Internal: a malformed checked division.
    InvalidCheckedDivision,                                              // unchanged
    /// Invalid plan: an exact operator with a non-Plain input column.
    NonPlainOperand { column: String, dtype: String },                   // unchanged
    /// Unsupported: a reserved ASAP operator with no timing contract.
    UnimplementedOperator { operator: &'static str },                    // unchanged
    /// Invalid plan: MatAtIngestion over a sub-DAG that can only run at query time.
    UpstreamNotAtIngestion { operator: &'static str },                   // unchanged
    /// Internal: a node reached by export without a timing.
    UntimedNode { operator: &'static str },                              // unchanged
    /// Internal: the exported DAG failed its own validation.
    InvalidPhysicalDAG(PhysicalASAPDAGValidationError),                  // unchanged
}

/// Expand every PaneChain into its panes; each pane now takes its chain's timing and
/// materialization, and its range and shift take the chain's timing (previously dropped).
pub fn expand_pane_chains(root: &Rc<OperatorNode>) -> Result<Rc<OperatorNode>, SchemaDerivationError>;

crates/devtools/src/candidate_labels.rs (new), planning_stages.rs

/// The label of every logical (`L<n>`) and physical (`P<n>`, `P<n>-m<k>`) candidate, by id.
/// For the stage_pipeline viewer document only; planning never reads labels.
pub fn candidate_labels(run: &PlanningStagesRun) -> BTreeMap<String, String>; // new
/// A Pass 1 choice, e.g. "Q1 exact · Q2 CMS+heap" (exact accumulators in parentheses).
pub fn choice_label<Id>(inventory: &PerComputationCandidates<Id>, choice: &[usize]) -> String; // new
/// What a Pass 2 candidate adds, e.g. " · one UnivMon for Q1, Q3 · Kll for Q1 in 5 × 1m panes".
pub fn pass2_label(pass2: &Pass2Candidate) -> String; // new
/// `logical` plus what the candidate materializes, e.g. "… · ingestion time: Kll ×5 panes".
pub fn physical_label(logical: &str, candidate: &PhysicalCandidate) -> String; // new

/// Every stage's output for one devtool run. The type stays.
pub struct PlanningStagesRun {
    /// The query-language frontends' roots, one per workload entry.
    pub stage0_query_language_frontends: Vec<QueryRoot>,                // unchanged
    /// Stage 1, Pass 1: the inventory the logical candidates' `choice` indexes.
    pub stage1_pass1_per_computation: PerComputationCandidates<usize>, // new
    /// Stage 1, Pass 2: the rules each Pass 2 candidate applied.
    pub stage1_pass2: Vec<Pass2Candidate>,                              // new
    /// Stage 1's logical candidates, `L<n>`.
    pub stage1_logical_asap_optimization: Vec<LogicalCandidate>,       // unchanged
    /// Stage 2's physical candidates; `P<n>` and `P<n>-m<k>` implement `L<n>`.
    pub stage2_physical_asap_optimization: Vec<PhysicalCandidate>,     // unchanged
    /// Logical candidates Stage 2 could not implement.
    pub stage2_pruned: Vec<PrunedCandidate>,                            // unchanged
    /// Stage 3's selection, or why it failed.
    pub stage3_plan_selection: Result<Selection, SelectionError>,      // unchanged
}

/// The devtool pipeline over roots from any mix of query languages; `entries[i]` is root i's
/// workload entry.
pub fn run_planning_stages_over(
    roots: Vec<QueryRoot>,
    entries: &[QueryWorkloadEntry], // was targets: &[AccuracyTarget], recurrence: &[QueryRecurrence]
    data: &DataWorkload,
    max_combinations: Option<usize>,
) -> Result<PlanningStagesRun, PlanningStagesError>;

/// Why the devtool pipeline stopped. The type stays.
pub enum PlanningStagesError {
    /// Bad input: Pass 1 failed; there are only the frontends' roots.
    Pass1(String),                                   // unchanged
    /// Request too large: Stage 1 enumerated more candidates than the limit (exit status 3).
    TooManyCandidates { limit: usize },              // unchanged
    /// Request too large: Stage 2 would build more physical candidates than the limit,
    /// counted, not built (exit status 3); planning it needs early pruning (#673).
    TooManyPhysicalCandidates { limit: usize },      // new
    /// Internal: the frontends' roots or a composed candidate failed validation.
    Validation(String),                              // unchanged
}

The asap-stage-pipeline/v1 document keeps its shape. Each candidate's display_name comes from candidate_labels, there are more physical candidates (P<n>-m<k>), and each query's requirements.latency_ms is now requirements.max_latency_ms.

Usage

End to end, as a test runs it on the head (#509 Example 4, Pattern B). The query's workload entry, with max_latency_ms, goes through Stage 2 materialization and Stage 3, and ends at the selected plan, its materialization and cost:

// 1. What the user writes: one live p99 panel over the last 5 min, evaluated
//    every minute, within rank error 0.01 (δ 0.01) and at most 200 ms per
//    evaluation, over 1M series arriving every 15 s (#509 Example 4, Pattern B).
let workload = PlanningWorkload {
    query_workload: QueryWorkload {
        language: QueryLanguage::PromQL,
        query_batch: None,
        repeating_queries: Some(vec![RepeatingEntry {
            query: Query("quantile_over_time(0.99, latency[5m])".into()),
            demand: RepeatedDemand::FixedInterval(RepetitionInterval(60_000)),
            requirements: QueryRequirements {
                accuracy: AccuracyTarget::epsilon_delta(0.01, 0.01).with_metric(ErrorMetric::Rank),
                max_latency_ms: Some(200.0), // new: was LatencyRequirement::ExplicitMaxMs(200.0)
            },
            predictability: Predictability::Predictable { known_at: None },
            time_selection: TimeSelection {
                scope: QueryTimeScope::RealTime,
                lookback: Some(DurationMs(300_000)),
                as_of: None,
            },
        }]),
    },
    data_workload: Some(asap_devtools::planner_layering_examples::shared_data_workload(
        DataArrival::ContinuouslyIngesting, // 1M series, one sample per 15 s each
    )),
};
// 2. `entries`: one QueryWorkloadEntry per query, in root order. Stage 2 reads
//    each one's recurrence and predictability (may state be kept from
//    ingestion time?); Stage 3 its accuracy target, max_latency_ms and recurrence.
let entries: Vec<QueryWorkloadEntry> = workload.query_workload.entries().collect();
// `data`: the arrival and rates Stage 2 checks for ingestion time and Stage 3 prices with.
let data = workload.data_workload.clone().unwrap();
// `roots`: the PromQL frontend's root per query (with each series' identity
// as a column), numbered for Pass 1.
let roots: Vec<(usize, QueryRoot)> = asap_frontend_promql::lower_promql_workload(&workload, 0)
    .unwrap()
    .iter()
    .map(|root| asap_types::ir::schema_support::with_promql_series_identity(root).unwrap())
    .map(QueryRoot::Operator)
    .enumerate()
    .collect();
// 3. Stage 1 → 2 → 3. `run` holds every stage's candidates and the selection.
let run = run_stage1_to_stage3(roots, &entries, &data, PlanningModels::builtin()).unwrap();
// Stage 2 output: one physical candidate per materialization assignment.
// `ids`: (id, materializes anything?) for each.
let ids: Vec<(&str, bool)> = run
    .stage2_physical_asap_optimization
    .iter()
    .map(|p| (p.id.as_str(), p.materializes_any()))
    .collect();
assert_eq!(
    ids,
    [("P1", false), ("P2", false), ("P3", false), ("P3-m1", true),
     ("P4", false), ("P5", false), ("P5-m1", true)]
);
// `b1`: P3-m1, the KLL in five 1-min panes maintained at ingestion time (B1).
let b1 = run.stage2_physical_asap_optimization.iter().find(|p| p.id == "P3-m1").unwrap();
assert!(b1.materialization.values().any(|m| matches!(m, Materialization::MatAtIngestion(_))));
// 4. Stage 3: `selection` has every valid candidate's cost and latency.
let selection = run.stage3_plan_selection.as_ref().unwrap();
// `b1_cost`: B1 keeps 6 panes of 1M per-series KLLs (~107 GB), so it costs
// far more per second, although its query takes only 7 ms.
let b1_cost = &selection.costs["P3-m1"];
assert_eq!((b1_cost.total.round(), b1_cost.latency[0].total_ms()), (13327.0, 7.0));
// 5. What the user gets back: `selected`, the cheapest valid plan, P2: one
//    whole-window KLL rebuilt each evaluation, nothing materialized.
let selected = run.selected().unwrap();
// `cost`: its price per second and its latency per evaluation, within 200 ms.
let cost = &selection.costs[&selected.id];
assert_eq!(selected.id, "P2");
assert!(!selected.materializes_any());
assert_eq!((format!("{:.3}", cost.total), cost.latency[0].total_ms()), ("2.100".into(), 126.0));
// `selection.accepted`: what Stage 3 accepted for the query (rank, KLL).
assert_eq!(selection.accepted["P2"][0].as_ref().unwrap().to_string(), "metric rank; accepted rank (Kll)");

Before this PR

• Every logical candidate had exactly one physical candidate, with everything at query time.
• expand_pane_chains dropped a chain's materialization.
• split_shared_by_phase worked on one root only.
• Latency bounds were ignored.
• A query's requirements were AccuracyRequirement / LatencyRequirement enums, and planning took separate accuracy-target and recurrence lists.
• Example 4: every materialization test passed vacuously, or was ignored.

After this PR (built-in models, illustrative constants)

Workload	Logical → physical	Selected	Notes
Example 1	27 → 57	P1, all exact, 7.94 /s (unchanged)	30 plans maintain Q1's rate panes, Q2's 10-s panes or both (counter panes are eligible over ingesting PromQL data). Maintained exact panes cost 12.94 (e.g. P4-m1, Q1's rate panes), against 8.54 rebuilt. No candidate breaks Q2's 100 ms bound
Example 2	82 → 184	P1, all exact, 6.08 (as on main)	102 plans materialize: MatAtQuery for builds two queries share, and panes at ingestion time. 164 invalid (accuracy)
Example 3a / 4A as given (once, ad hoc)	972 → 972	P485, 39,828.8	No eligible build, and no build is read by two queries. 844 invalid
4A at rest	972 → 972	P485, 597.43	Nothing at ingestion time
4A monthly	12,500 → 67,228	P4125, 55.32 (nothing materialized)	54,728 plans maintain 365- or 73-pane DDSketch/KLL chains. Above the devtool's 10^4 limit
4B (= 3b)	5 → 7	P2, whole-window KLL, 2.100	B2 (P3, query-time panes) 2.183, 131 ms. B1 (P3-m1) 13,326.5 /s: it keeps 6 panes × 1M per-series KLLs ≈ 107 GB, and its query takes 7 ms. Whole-window KLL takes 126 ms. All are within 200 ms

Example 4 tests:

Test	Before	After
ingestion-time upstream is ingestion time	passes (vacuously)	passes, over B1 and the monthly chains
at rest runs nothing at ingestion time	passes (vacuously)	passes
materialized output covers its consumers	passes (vacuously)	passes. retention_ms now derives the kept span (B1: 6 min ≥ 5 min)
B1 builds at ingestion, merges at query time	ignored	passes
tumbling KLL has B1 and B2 (new, implementer test)	—	passes
B1/B2/B3 present, B3 keeps windows, B2 costs most	ignored	ignored: needs B3 (G9, #625)
B1 cheapest	ignored	ignored: the built-in model contradicts it (numbers above). Needs B3 and the model-relative rewrite (Q49, G9)
sliding KLL	ignored	ignored: sliding windows
Pattern A (A1–A3)	ignored	ignored: the shared EH (#628, #634)

Verification

• New unit tests:
  • types: a filtered build never runs at ingestion time; a cross-root split copies only the scan two queries read at two phases; expansion carries the chain's timing.
  • Stage 2: the pane chain options; the S4 refusals (at rest, unknown arrival, not predictable, one-off, whole window without phase or too long, filter); rate and increase panes over ingesting PromQL data run at ingestion time, while a counter over a table source or mixed data stays at query time; MatAtQuery for a build read by two queries; the scan split across queries. These two counter tests and the MatAtQuery memory test were written failing first.
  • Stage 3: kept state at ingestion time costs 2 × state per second, and a MatAtQuery state costs state × evaluation latency × evaluations per second; an ingestion chain pays one second of rows × (depth + 1) and keeps 6 × series × state bytes; the latency bound rejects B2 with its latency and selects B1; compaction appears once ingestion needs more than one core.
• End to end: tumbling_pane_merge.rs takes the planner's pane chain assigned MatAtIngestion, expands it (the panes carry MatAtIngestion), builds the panes in Scope::Ingestion runs and merges them at query time. The result is identical to the single build for sum, p99, rate and increase.
• New: query_requirements_json_is_a_target_and_an_optional_bound (written failing first): {} parses to exact with no bound, max_latency_ms round-trips, and the old tagged forms and unknown fields are rejected.
• Regression tests that changed:
  • Tests now build QueryWorkloadEntry values (test_support::entry / entry_with in asap-aware-mapping, ..Default::default() elsewhere) or read workload.query_workload.entries().
  • count_min_is_admitted_without_a_target is deleted: every query now has a target, and under any target a Count-Min over a metric not declared a counter is rejected.
  • summary_sharing: four tests that now pass are un-ignored (quantiles_with_equal_params_share_one_producer, cross_series_p50_and_p99_share_one_producer, identical_ungrouped_queries_share_their_producers, identical_sql_percentiles_share_one_producer). The other three stay ignored because plan selection still picks an exact plan for one of their queries.
  • reuse_rules (feat(planner): Pass 2 reuse rules (rollup, exact top-k reuse) and nested sum merge in Pass 1 (#680) #716): the rollup test now reads Stage 2's own MatAtQuery instead of the hand-written stand-in, and plan selection picks the rollup unaided. The exact top-k test keeps the stand-in, because Stage 2 builds a shared summary once per evaluation, but not a shared exact ranking.
  • operator_design_examples::batch_planning_replaces_summary_operators now selects one MatAtQuery SUM state for both queries.
  • Example 1/2 Stage 2 tests now check one unmaterialized candidate per logical candidate, plus the materialized ones.
  • The devtool limit test (57 physical) and the Example 1 viewer fixture were regenerated.
  • Label tests moved to asap_devtools::candidate_labels (Example 1: choice, Pass 2 and ingestion-time labels; Example 2: "query time, shared: UnivMon"). Stage 2 tests assert the candidate's materialization instead.
  • exact_quantile_state_costs_more_than_a_sketch_over_large_groups keeps the state at ingestion time, since MatAtQuery state is no longer kept across evaluations.
• Local checks, after rebasing onto cde2ca3d (feat(planner): Pass 2 reuse rules (rollup, exact top-k reuse) and nested sum merge in Pass 1 (#680) #716):
  • cargo nextest run --workspace passes: 1856 tests, plus the two slow Example 4 tests run separately.
  • cargo fmt --all -- --check and cargo clippy --workspace --all-targets --all-features --locked -- -D warnings pass.
  • cargo test --doc -p asap-types passes, and so does tools/verify_metricsql_external_consumer.sh.
  • The Usage above was run as a test on the head.
• tools/dag-viewer: python3 -m unittest test_server test_viewer passes (50 tests).

Limitations and follow-ups

• Slow Example 4 tests. Example 4's monthly variant now plans 67,228 physical candidates in about 150 s. The two invariant tests that run it take about 2.5 minutes each. The enumeration is uncapped by design; early pruning is Planning roadmap: move ASAPQuery planning into ASAPPlanner, MILP plan selection over shared DAGs, benchmark-based accuracy #671/Prune the candidate search space early without discarding a possibly-best valid plan #673.
• G9: B3 (state kept across evaluations), the capability gate, deployment-supplied ms_per_cpu_op / ms_per_scan_byte, and Example 4's model-relative B1/B2 tests (Q49).
• Counter assumption: counter builds are maintained only over continuously ingesting time-series (PromQL) data, which is taken to arrive in time order per series. Nothing checks this at runtime. MaintainPopulation stays as before.
• Stage Viewer: the Example 2 row in stage-viewer.md does not yet mention the MatAtQuery plans. Docs are tracked in Update the docs after the G1–G11 consolidation groups land #677.
• Prune reasons: why a build cannot run at ingestion time (IngestionPruneReason) is now internal. If it should reach the user, it would go through the candidate or report, not the MaterializationCandidate type.

[zzylol]

Can the display label etc DAG viewer related fields being separate from the physical candidate structures themselves?

[zzylol]

MaterializationUnit I also didn't understand what is this for

[zzylol]

Can the display label etc DAG viewer related fields being separate from the physical candidate structures themselves?

Yes, and it should be. Planned change in this PR:

• Remove display_name and materialization_label from PhysicalCandidate (and display_name from LogicalCandidate), so the Stage 1/2/3 structs hold only planning data: id, origin, roots, DAG, and the materialization assignment.
• The labels become pure functions over that data in the devtools crate (stage_pipeline / planning_stages). Only the viewer document gets them, e.g. label_physical(&candidate) -> String derived from the candidate's assignment.
• Tests that assert labels move with them.

[zzylol]

MaterializationUnit I also didn't understand what is this for

It is one summary build that Stage 2 can decide to store, plus its materialization options. It is now named MaterializationCandidate. It is not a whole-workload plan like PhysicalCandidate: the physical candidates are the combinations of these options. Example, two queries over one metric:

q1: quantile_over_time(0.99, latency[5m])   every 1m
q2: max_over_time(latency[5m])              every 1m

Say Stage 1 chose KLL panes for q1 and exact max panes for q2. Stage 2 then finds two materialization candidates:

materialization candidate	readers	options
Kll ×5 panes (q1)	q1	NotMaterialized, MatAtIngestion
Max acc ×5 panes (q2)	q2	NotMaterialized, MatAtIngestion

(A whole-window exact max would not get MatAtIngestion here: its 5m window is longer than the 1m interval.) One read by two or more queries also gets MatAtQuery (built once per evaluation, reused). Stage 2's physical candidates are the product of the options (here 2 × 2 = 4), less assignments that put a build at ingestion time over one that is not, and Stage 3 prices each one. at_ingestion records why ingestion is ruled out when it is (e.g. Filtered, NotRepeated).

It is only Stage 2's enumeration bookkeeping, so in this PR it is crate-private, with this example in its doc comment instead of being public API. The public result stays PhysicalCandidate, which carries its assignment in materialization.

[zzylol]

Renaming MaterializationUnit → MaterializationCandidate (crate-private), per the review discussion. Its doc comment clarifies it is one summary build Stage 2 may store, with its options, not a whole-workload plan like PhysicalCandidate.

[zzylol]

/// How one query is demanded, as planning reads it. Stage 2 reads recurrence and
/// predictability; Stage 3 reads recurrence and the latency bound.
pub struct QueryDemand { // new
pub recurrence: QueryRecurrence,
pub predictability: Predictability,
pub response_latency: LatencyRequirement,
}
/// From a workload entry: its recurrence, predictability and latency requirement.
impl From<&QueryWorkloadEntry> for QueryDemand;
/// A query known only by its recurrence: Predictability::Unknown, LatencyRequirement::Unspecified.
impl From for QueryDemand;

what's the difference between QueryDemand and Queryworkload/requirement?

[zzylol]

what's the difference between QueryDemand and Queryworkload/requirement?

There's no real difference. QueryDemand copies three fields out of QueryWorkloadEntry (recurrence, predictability, and the latency part of requirements) and adds nothing. It only exists because the Stage 2/3 functions take roots plus separate per-query lists (targets, recurrence) instead of the workload entries.

Change in this PR:

• Remove QueryDemand. optimize_physical_asap, select_plan and run_stage1_to_stage3 take &[QueryWorkloadEntry] and read accuracy, latency, recurrence and predictability from it. The separate targets and recurrence parameters go too.
• In the same PR, simplify QueryRequirements to { accuracy: AccuracyTarget, max_latency_ms: Option<f64> }, removing the AccuracyRequirement and LatencyRequirement wrappers. Omitted fields mean exact and no latency bound.
• Tests build entries with one small helper.