Skip to content

refactor(stage3): execution contexts x^mat, x_{κ,v}, y_{κ,u,c,i}, h_v and constraint (R) (#1207 PR 7) - #1237

Draft
zzylol wants to merge 6 commits into
refactor/1207-stage3-baselinefrom
refactor/1207-contexts
Draft

zzylol wants to merge 6 commits into
refactor/1207-stage3-baselinefrom
refactor/1207-contexts

Conversation

@zzylol

@zzylol zzylol commented Nov 7, 2026

Copy link
Copy Markdown
Contributor

Stacked on #1236 (PR 6, Stage 3 baseline). Part of #1207 (PR 7 of the order), #1204. Draft; not to be merged on its own (one merge unit, PRs 0–9).

Why

PR 6's baseline charges every candidate a fixed $w_v$: a query-time candidate was costed once per schedule of every query reaching its group, whether or not they select it (PR 6 spec question 2), and the range under a tumbling-window chain was charged though the realizer builds the chain over the range's input (question 3). The concepts doc §5 and #1204 items 1 and 4 replace this with execution contexts and the retention amount $h_v$ with constraint (R).

What

  • Execution contexts $\kappa$: the queries' evaluations (one per distinct (QueryRecurrence, RangeEvaluation), deduplicated by equality), the ingestion pipeline $I_r$ of each refresh window (as asap-cost groups ingestion runs), and $\kappa_m$, an AcrossEvaluations candidate's own computation.
  • Variables: $x^{\mathrm{mat}}v$ per AcrossEvaluations candidate, $x{\kappa,v}$ per None/OneEvaluation candidate and context, $y_{\kappa,u,c,i}$ per edge and context of its consumer (only when an AcrossEvaluations producer delivers_on $\kappa$), $h_v$ continuous, plus $z$, $t$. Constraints (A)–(F) hold per execution; (R) $h_v \ge \operatorname{need},y$ (and $\ge \operatorname{need},z$ for a query reading $v$ itself).
  • Objective $\sum W^{\mathrm{mat}}v x^{\mathrm{mat}}v + \sum \rho\kappa c{\kappa,v} x_{\kappa,v} + \sum \pi_v h_v$, every coefficient from asap_cost::node_usage, plus the runtime's request fee (a constant).
  • Decode: one DagOfSelectedCandidates node per selected execution, so a candidate run in two contexts is two nodes (the realizer keeps them apart).
  • delivers_on for IngestionPipeline/Materialized contexts (the PR 0 todo!): false for a windowed AcrossEvaluations producer (times not known from the memo).
  • Every todo!("#1207 PR 7 …") is implemented (constraint (R); delivers_on).
  • Objective vs plan_cost: exact on small workloads (test below); otherwise SearchReport::inexact_objective names the plan_cost terms the formulation leaves out, and a difference with none of them is ErrorSelectingPlan::ObjectiveDiffersFromPlanCost.

Not here: (K) accuracy (PR 8), tools/snapshots (PR 9); latency (L) and budgets (B) are out of scope (#1204 items 5, 6).

Representation, rules, enumeration (#1207 (a)/(b)/(c))

  • (a) Representation: no memo change. MILP variables mirror executions: $x^{\mathrm{mat}}v$, $x{\kappa,v}$, $y_{\kappa,u,c,i}$, $h_v$.
  • (b) Rules: none.
  • (c) Enumeration: the MILP over executions; the reference enumerator (Appendix A) is now over executions, with contexts, delivers_on and retention, and agrees with the MILP on 30 random cost sets × 4 schedules (two queries, two chains) plus PR 6's 100 cases.

Notation → code

Notation Code
$\kappa$, $\mathcal K$ ExecutionContext (types), ExecutionContexts::contexts (deduplicated by equality, indexed by ExecutionContextId); $I_r$ keyed by RefreshWindow (RefreshWindowId indexes them)
$x^{\mathrm{mat}}_v$ ExecutionContextVariables::x_materialized
$x_{\kappa,v}$ ExecutionContextVariables::x_in_execution_context
$y_{\kappa,u,c,i}$ ExecutionContextVariables::y_in_execution_context (key: consumer's context, PhysicalEdgeId)
$h_v$ ExecutionContextVariables::retention_amount
$z_{q,v}$, $t$ SelectionVariables::{z, t} ($t$ per execution on cyclic components)
(A)–(F) input_satisfaction_constraints, producer_selection_constraints, query_coverage_constraints, acyclicity_constraints, utilization_constraints (per execution)
(R), $\operatorname{need}_{\kappa,v,c,i}$ retention_constraints; TumblingWindowChainGeometry::retention_need_windows
$W^{\mathrm{mat}}v$, $\rho\kappa c_{\kappa,v}$ ExecutionCosts::cost_per_second (execution_cost.rs)
$\pi_v$ ExecutionCosts::retention_cost_per_window_per_second
$\rho_\kappa$ execution_cost::executions_per_second
delivers_on per context ExecutionContexts::new (y only when PhysicalCandidate::delivers_on(memo, κ))

How

  1. ExecutionContexts::new: walk from each query output in its evaluation context; an execution in $\kappa$ reads a None producer in $\kappa$, a OneEvaluation producer in $\kappa$ only when $\kappa$ is an evaluation, an AcrossEvaluations producer once if it delivers on $\kappa$; an AcrossEvaluations execution's inputs run in its computation context ($I_r$ at ingestion time with a bounded refresh window, else $\kappa_m$). Query outputs never come from an (IngestionTime, None) result.
  2. execution_costs: Kahn's order over executions, one representative producer per input (as PR 6), each execution costed by node_usage at $\rho_\kappa$ (evaluation rate; 1/s at ingestion; one build per window for a query-time chain). A query-time scan reads the windows of the ranges that run in its context (a chain's newest window in $\kappa_m$). The range a chain is built over costs 0 in the chain's context. A windowed build is costed with the realizer's whole-chain merge, which its consumers read. $\pi_v$ = cost retaining $n{+}1$ windows − cost retaining $n$; $W^{\mathrm{mat}}_v$ = cost at $n$ − $\pi_v n$.
  3. solve_selection: build, solve (HiGHS, gap 0), decode one node per execution.
  4. select_plan: realize, plan_cost, compare, report.

Major algorithm changes

Stage 3 selection over execution contexts

  • Problem. Baseline $w_v$ was fixed per candidate. E.g. p99 and p50 over latency[10m] (PR 6 plan below): plan_cost 7.25e-3/s; here 6.25e-3/s (one shared DDSketch OneEvaluation in the queries' context).
  • Before / after. Before: one $x_v$ per candidate, $w_v$ per schedule of every query reaching its group. After: How 1–4.
  • Correctness. (A)+(C)+(F)+(E) per execution make every node reachable and acyclic; validate checks it. Tests: per constraint family; the_milp_agrees_with_the_reference_enumeration_over_contexts_and_retention; the_objective_equals_plan_cost_on_small_workloads (end to end).
  • Complexity. Executions ≤ $|\mathcal K| \cdot |\mathcal C|$ (only those reachable from a query output); $y$ ≤ $|\mathcal K| \cdot |\mathcal E^{adm}|$; (R) one per edge out of a materialized producer. Measured below (78–452 variables, 18–178 ms).
  • Assumptions and limits. Query latency ignored (decided 2026-11-06): a result counts as read when its evaluation starts. Cost model 1 only. Inexact against plan_cost where the formulation and asap-cost differ (spec questions).

Deleted (replaced in this PR)

Deleted Replaced by
candidate_cost.rs (candidate_costs, computations_per_second, scan_window_of) execution_cost.rs (execution_costs, executions_per_second, scan_window_in_context)
SelectionVariables::{x, y} ExecutionContextVariables
SelectionInputs::{cost_per_second, entries, delivers_to_consumer} ExecutionContexts + ExecutionCosts
FIXED_COST_PER_CANDIDATE inexact term named terms (HELD_MEMORY, …)
todo! in delivers_on, retention_constraints implemented

ObjectiveDiffersFromPlanCost is kept and now used.

New abstractions not in #1207 or the doc (pending approval)

All Stage 3 crate-private:

  • Execution { Materialized(PhysicalCandidateId), InContext(ExecutionContextId, PhysicalCandidateId) }: one node of the selected DAG.
  • ExecutionContexts (contexts, refresh windows, executions, edges y may select; output_execution, execution_in, computation_context, input_context).
  • TumblingWindowChainGeometry (width, window count, offset, @ anchor, range input; refresh_window, retention_need_windows).
  • ExecutionCosts (incl. representative_producers, uncostable), executions_per_second, scan_window_in_context, objective_of, SolvedSelection::execution_of_node, request_cost_per_second.
  • SelectionVariables keeps only z and t, with t: HashMap<Execution, Variable> (doc: t: Vec), only on cyclic components.
  • Rules Stage 3 states: OneEvaluation runs only in evaluation contexts; no query output from (IngestionTime, None); a query reading a materialized result directly also bounds $h_v$; inexact-term names in InexactObjective::inexact_terms.

Spec questions

  1. plan_cost vs the formulation (makes the objective inexact; recommend aligning asap-cost, not done here): (a) plan_cost runs a not-materialized node once per query reaching it, the formulation once per context; (b) a query-time chain: plan_cost builds per request of its queries, the formulation (and the runtime's windows refreshed on first read) once per window — they differ whenever $w &lt;$ interval.
  2. Held memory: plan_cost holds query-time results for query latency; latency is out of scope, so the objective has no such term (every query-time plan with memory cost is inexact).
  3. Representative producer: $c_{\kappa,v}$ is costed over one producer per input; a consumer's cost depends on the selected producer's output (e.g. a sketch vs exact state). Constant $c_{\kappa,v}$ cannot express it.
  4. Not encodable as a constant, but linear with a new variable (not added, would be new families): raw data retained per source (Stage 3 MILP: implement the formulation variables x, y, z, t and rows (A)–(I) as named code over explicit edges #1204 item 4's $h_{\text{src}}$, max over scans); the merge over a chain materialized across evaluations (one per context reading it); the range under a None/OneEvaluation chain in an evaluation context (free only if no non-chain consumer reads it); scan windows of the selected consumers only.
  5. delivers_on in IngestionPipeline/Materialized: returns false (a RefreshWindowId and a PhysicalCandidateId cannot be resolved from the memo). Stage 2 offers AcrossEvaluations only to builds, whose consumers read them at query time, so no edge is lost today. The doc's table could say so.
  6. Doc §5: "an upper bound on $h_v$ fixed per candidate when Stage 2 builds it" — not implemented (no need for one; $h_v$ is minimized).

Lower-PR items (not changed)

  • ErrorSelectingPlan::SelectedPlanFailedCheck and RejectionCause::{LatencyBoundExceeded, LatencyOverRepeatInterval, OverClusterCapacity} are never constructed; the first's doc mentions budget shares (PR 6/PR 8).
  • RejectionCause::Infeasible's doc still mentions "the catalog's consumer constraints" ((J) deleted) (PR 6).
  • PR 6's a_free_candidate_nothing_reads_is_not_selected test is named for (F) but relies on delivers_on (fine, unchanged).

Key code interfaces added / modified

types memo_candidates.rs, execution_contexts.rs, model.rs, execution_cost.rs, select.rs, report.rs

impl PhysicalCandidate {
    /// Whether a consumer running in `context` can read this producer. // body filled: false for a
    /// windowed AcrossEvaluations producer in IngestionPipeline / Materialized (was todo!)
    pub fn delivers_on(&self, memo: &Memo, context: &ExecutionContext) -> bool;
}

/// Stage 3's search report. // stays
pub struct SearchReport {
    pub status: SearchStatus,                        // unchanged
    pub cost_lower_bound_per_second: Option<f64>,    // unchanged (now includes the request fee)
    pub milp_variables: usize,                       // unchanged
    pub milp_constraints: usize,                     // unchanged
    pub model_construction_ms: u64,                  // unchanged
    pub solver_ms: u64,                              // unchanged
    pub validation_ms: u64,                          // unchanged
    /// Set when plan_cost differs from the objective, with the plan_cost terms the
    /// formulation leaves out that the selection meets. // doc changed
    pub inexact_objective: Option<InexactObjective>,
}

/// Objective and plan_cost per second (both with request fees), and the named terms. // stays
pub struct InexactObjective {
    pub objective_per_second: f64,                   // unchanged
    pub plan_cost_per_second: f64,                   // unchanged
    /// Term name → selected executions it may concern. // was one "fixed cost per candidate" term
    pub inexact_terms: BTreeMap<String, usize>,
}

pub enum ErrorSelectingPlan {
    // other variants unchanged
    /// plan_cost differs from the objective and no named term explains it: a cost-coefficient bug. // now constructed
    ObjectiveDiffersFromPlanCost { objective_per_second: f64, plan_cost_per_second: f64 },
}

// Stage 3 private (pending approval, above):
pub(crate) enum Execution { Materialized(PhysicalCandidateId), InContext(ExecutionContextId, PhysicalCandidateId) }
pub(crate) struct ExecutionContextVariables {
    x_materialized: HashMap<PhysicalCandidateId, Variable>,                                // x^mat_v
    x_in_execution_context: HashMap<(ExecutionContextId, PhysicalCandidateId), Variable>, // x_{κ,v}
    y_in_execution_context: HashMap<(ExecutionContextId, PhysicalEdgeId), Variable>,      // y_{κ,u,c,i}
    retention_amount: HashMap<PhysicalCandidateId, Variable>,                             // h_v
}
fn retention_constraints(inputs: &SelectionInputs<'_>, selection: &SelectionVariables,
    variables: &ExecutionContextVariables) -> Vec<Constraint>; // (R), was todo!

Usage (end to end; crates/planner/tests/execution_contexts.rs at 8a466a4):

// The user's workload: exact sum(rate(latency[1m])) every 10 s and max(rate(latency[5m])) every 15 s,
// both at phase 0, over 1,000 stored series sampled every 15 s; cost model 1 with memory free.
let output = plan(&[("sum(rate(latency[1m]))", 10_000, 0), ("max(rate(latency[5m]))", 15_000, 0)],
                  &memory_free()).await;
//   e2e_plan → run_planning → select_plan:
//     ExecutionContexts::new: two evaluation contexts (10 s, 15 s); the scan of latency runs in each
//     execution_costs: ρ = 0.1/s and 0.0667/s; each scan reads its own context's range (1m, 5m)
//     solve_selection: x_{κ,v}, y_{κ,u,c,i}, z; decode: one node per selected execution
//     realize + plan_cost: equal to the objective
let selection = output.selection.expect("a plan is selected");
assert_eq!(selection.search_report.inexact_objective, None);        // objective == plan_cost
assert!(selection.search_report.cost_lower_bound_per_second.is_some()); // the solver's bound is the plan's

Before this PR

Per-candidate fixed $w_v$; a None result shared across differently scheduled queries was one node costed for all schedules; chain retention fixed at its window; every plan reported inexact with one generic term.

After this PR

Representative plans (cost model 1 defaults; fixed_interval_at 10 min, phase 0; 1,000 series at 1 s); MILP size and solve time:

=== p99 over 60m (raw data retained: false)
cost/s 3.352e0; MILP 78 variables, 110 constraints, solve 25 ms; inexact objective: Some((3.3522033333333328, 3.3522133398133325, [("a consumer costed over the output of ano", 1), ("memory a query-time result holds while i", 2), ("the merge the realizer reads a tumbling-", 1)]))
  6 × ASAPSummaryAgg IngestionTime AcrossEvaluations(Memory) θ=Some(DDSketch { alpha: 0.1 })
  1 × ASAPSummaryEstimate QueryTime OneEvaluation(Memory) θ=None
  1 × ASAPSummaryMerge QueryTime None θ=None
  1 × NonASAPProject QueryTime OneEvaluation(Memory) θ=None
  1 × NonASAPScan IngestionTime None θ=None
  6 × NonASAPTimeRange IngestionTime None θ=None
  5 × NonASAPTimeShift IngestionTime None θ=None
=== p99 over 60m (raw data retained: true)
cost/s 3.844e-2; MILP 314 variables, 394 constraints, solve 168 ms; inexact objective: Some((0.03720333333333333, 0.03843937688, [("a consumer costed over the output of ano", 1), ("a query-time scan: plan_cost reads the w", 1), ("memory a query-time result holds while i", 5)]))
  1 × ASAPSummaryAgg QueryTime None θ=Some(DDSketch { alpha: 0.005 })
  1 × ASAPSummaryEstimate QueryTime OneEvaluation(Memory) θ=None
  1 × NonASAPProject QueryTime OneEvaluation(Memory) θ=None
  1 × NonASAPScan QueryTime OneEvaluation(Memory) θ=None
  1 × NonASAPTimeRange QueryTime OneEvaluation(Memory) θ=None
=== p99 and p50 over 10m (raw data retained: true)
cost/s 6.249e-3; MILP 452 variables, 565 constraints, solve 178 ms; inexact objective: Some((0.006206666666666666, 0.006249023076293333, [("a consumer costed over the output of ano", 2), ("a not-materialized execution in a contex", 2), ("a query-time scan: plan_cost reads the w", 1), ("memory a query-time result holds while i", 7)]))
  1 × ASAPSummaryAgg QueryTime OneEvaluation(Memory) θ=Some(DDSketch { alpha: 0.005 })
  2 × ASAPSummaryEstimate QueryTime OneEvaluation(Memory) θ=None
  1 × NonASAPProject QueryTime None θ=None
  1 × NonASAPProject QueryTime OneEvaluation(Memory) θ=None
  1 × NonASAPScan QueryTime None θ=None
  1 × NonASAPTimeRange QueryTime OneEvaluation(Memory) θ=None
=== sum by job of rate 5m, exact (raw data retained: true)
cost/s 3.905e-3; MILP 60 variables, 78 constraints, solve 18 ms; inexact objective: Some((0.0039050000000000005, 0.003905046906860001, [("a consumer costed over the output of ano", 1), ("a query-time scan: plan_cost reads the w", 1), ("memory a query-time result holds while i", 5)]))
  1 × ASAPFinalizeExactAccumulator QueryTime None θ=None
  1 × ASAPSummaryAgg QueryTime None θ=None
  1 × NonASAPAggregate QueryTime None θ=None
  1 × NonASAPScan QueryTime None θ=None
  1 × NonASAPTimeRange QueryTime None θ=None

Compared with PR 6 (#1236 body): p99+p50 10m 7.25e-3 → 6.25e-3/s (one DDSketch OneEvaluation shared in the context); p99 60m raw retained 3.829e-2 → 3.844e-2/s (DDSketch alpha 0.1 → 0.005: the two tie in the objective, which leaves out held memory — spec question 2); the other two unchanged. MILP sizes 88/88, 374/372, 522/521, 78/76 → 78/110, 314/394, 452/565, 60/78 (variables/constraints).

Verification

  • cargo fmt --all; cargo clippy --workspace --all-features --all-targets -- -D warnings clean; crate_graph passes (in integration-tests).
  • New tests (stage3): contexts_are_deduplicated_by_equality, a_not_materialized_candidate_runs_once_per_context_that_selects_it ($x_{\kappa,v}$), a_one_evaluation_result_is_shared_within_its_context, two_queries_at_phases_0_and_5_s_read_one_5_s_chain (Stage 3 MILP: implement the formulation variables x, y, z, t and rows (A)–(I) as named code over explicit edges #1204 regression; $x^{\mathrm{mat}}$, $y$ per context, delivers_on per context), a_chain_retains_what_its_most_demanding_consumer_reads ((R), 6 vs 11 windows), a_chain_reads_its_inputs_in_its_own_computation_context ($\kappa_m$), chains_at_ingestion_share_the_pipeline_of_their_refresh_window ($I_r$ dedup), execution_costs_follow_the_context_of_each_execution ($\rho_\kappa$, range under a chain 0, $\pi_v$), the_milp_agrees_with_the_reference_enumeration_over_contexts_and_retention; PR 6's (A)–(F) tests adapted. types: delivers_on cases for pipeline/materialized contexts. planner: the_objective_equals_plan_cost_on_small_workloads, held_memory_is_reported_as_an_inexact_term.
  • Independent review (fresh agent) of the diff; fixed: (R) for a query reading a materialized result directly, unexplained objective differences are an error again, held-memory term only when memory has a cost, slices instead of clones, execution-cost tests, stale test doc, wording. Not taken (spec questions 1, 4–6): extra variables for the chain merge / range-under-chain / raw data per source; geometry from the range group's expression (range groups hold only TimeRange candidates today); code dedup of per-crate index maps (minor).

Test table (cargo test -p <crate> --all-features --no-fail-fast, at 8a466a4; PR 6: 1330/94/9):

crate passed failed ignored
asap-executor 281 0 1
asap-types 197 0 0
asap-frontend-promql 218 0 0
asap-frontend-common 49 0 0
asap-planner 26 5 1
asap-accuracy 54 0 0
asap-cost 37 0 0
asap-frontend-metricsql 15 0 0
asap-frontend-sql 175 0 0
asap-sql-function-catalog 10 0 0
asap-stage0-memo 13 0 0
asap-stage1-pass1-per-computation-candidates 33 0 0
asap-stage1-pass2-reuse-candidates 20 0 0
asap-stage2-physical-optimization 15 0 0
asap-stage3-plan-selection 20 0 0
metricsql_common 3 0 0
asap-devtools 42 52 3
asap-integration-tests 133 37 4
total 1341 94 9

Failures: same set as PR 6, except sharing_issue_regressions::grouped_filtered_quantile_plans now passes and planner_fuzz::quantile_rank_witness_prices_from_the_same_sketch is now inconclusive (its witness contains ASAP operators, which Pass 1 refuses — the PR 6 class of 12 fuzz failures, PR 9's tool rework). Each stops where PR 6's did: devtools 37 at PR 9 todo!()s, 13 fuzz inconclusive, 2 CLI JSON (PR 9); planner 5 (Pass 2 metric rule / raw data, PR 4 spec question; plans until (K), PR 8; Stage 2 executor-support gap, PR 5); integration 37 (Stage 2 infeasible outputs, PR 5/Pass 1; pinned plan shapes, PR 8/PR 9; θ-less or pre-14 snapshots, PR 9; joint/accuracy-policy, PR 8). No failure is ObjectiveDiffersFromPlanCost or stops at a PR 7 todo!().

…ed computation (#1207 PR 7)

A windowed AcrossEvaluations producer read inside an ingestion pipeline
or another AcrossEvaluations candidate's computation never delivers: when
those contexts run is not known from the memo. Replaces the PR 7 todo.
… ρ_κ c_{κ,v}, h_v and constraint (R) (#1207 PR 7)

Stage 3 selects per execution: an AcrossEvaluations candidate once
(x^mat_v), a None or OneEvaluation candidate once per context κ that
selects it (x_{κ,v}), every edge per context of its consumer
(y_{κ,u,c,i}, only when an AcrossEvaluations producer delivers on κ).
Contexts are the queries' evaluations (deduplicated by equality), the
ingestion pipelines of refresh windows, and materialized computations.
Constraints (A)-(F) hold per execution; (R) h_v >= need·y prices
retained tumbling windows at π_v. The objective is
Σ W^mat x^mat + Σ ρ_κ c_{κ,v} x_{κ,v} + Σ π_v h_v, costed per execution
by asap-cost; the range under a chain costs nothing (the realizer
bypasses it). Replaces the baseline's per-candidate w_v
(candidate_cost.rs) and its delivers_to_consumer check.
… inexact objective names the plan_cost terms the formulation leaves out (#1207 PR 7)

A tumbling-window build is costed with the merge the realizer reads it
through (charged in its context when it is not materialized), so its
consumers see the merged states. When plan_cost of the realized plan
differs from the objective, SearchReport::inexact_objective names the
terms the formulation does not encode that the selection meets (memory
held for query latency, runs per query, per-request chain builds, the
merge over a materialized chain, representative producers, ranges under
chains, query-time scans). The objective includes the runtime's request
fee, a constant. ObjectiveDiffersFromPlanCost, never constructed, is
deleted.
…d result; an unexplained objective difference is an error again; held memory only when memory has a cost; slices; execution_costs test (#1207 PR 7)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

optimizer Workload optimization and plan selection refactor

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant