[Feature] Stage 3: no execution-scope copies, exact candidate-level MILP, one cost formula #1095

Example

One metric data (1,000 series, raw samples retained) and four queries:

q1: quantile_over_time(0.9, data[15m])                      # every 10 s, max_error 0.05
q2: quantile_over_time(0.9, data[1h])                       # every 60 s, max_error 0.05
q3: sum by (instance, region) (avg_over_time(data[1m]))     # every 10 s, exact
q4: sum by (instance)         (avg_over_time(data[1m]))     # every 10 s, exact

A good plan:

• maintains one KLL pane chain over 10 s panes at ingestion time (10 s = gcd of q1's and q2's grids, §2.6), and q1 and q2 both merge their windows from it;
• reads the incoming samples once per pane, even when a second sketch with the same pane width (a DDSketch chain, or a KLL chain of another size) is also maintained: both are refreshed in one ingestion run (Refresh ingestion-time views that share a refresh window in one run, and charge a shared stateless ingestion node once per run #1151);
• materializes q3's 150-group sum at query time (MatAtQuery) and rolls it up to 50 groups for q4;
• below that sum, avg_over_time builds an exact Sum and an exact Count over the same TimeRange(Scan), and the scan runs once per evaluation, not once per build.

Each item is a choice among candidates that give the same result at different cost, and the choices are coupled: the KLL chain is cheap only if both quantile queries read it, the scan only if both builds read the same run. Stage 3 must make them together, at the plan's exact cost, on workloads such as the 9-query random-mixed-seed-862/workload-01 (10,000 series) within 8 GB.

The current Stage 3 does not. Stage 2 pre-allocates up to max(contexts, uses) identical copies of each candidate, and the MILP gets one binary per copy and per (consumer copy, producer copy). On workload-01 the model grows from 0.7M to 2.3M to 6.6M variables over three refinement rounds, and HiGHS aborts with std::bad_alloc at 7.7 GB. Cost is computed in five places, three of them approximations. So the objective is not the plan's cost, and the propose → evaluate → cut loop rarely closes the gap.

This issue specifies the replacement, an exact MILP over memo groups and execution contexts. The four queries above are the running example throughout.

1. Problem statement

Input. Stage 2's PhysicalASAPMemoCandidates. Each candidate is one (operator node, timing, materialization) of a memo group, and lists per input the candidates of the input group it may read (producer_candidates_per_input). Stage 3 also gets the query workload (rate, accuracy target, latency bound per query), the data workload with its bounds (rows, G, source rates), the accuracy model and evidence, and the cost inputs (ResourceUnitCost, ResourceSpeed, measured operation costs, frozen calibration factor, non-repeating horizon).

Output. A DAG with one node per run of a candidate. Every input of every node has exactly one allowed producer, and every output reference is served.

Objective. Minimize cost per second under cost model 1 (usage over time), exactly as asap_cost::plan_cost computes it.

Constraints. Every query meets its latency bound and accuracy target. Every reader constraint holds: MinimumQueryReaders, HighestSharedQueryMaterialization, AllowedOutputs and IncompatibleOutputs.

Interface. Stage 3 depends only on asap-types, asap-accuracy and asap-cost, and it calls no other stage.

pub fn select_plan(
    physical: &PhysicalASAPMemoCandidates,
    entries: &[QueryWorkloadEntry],              // accuracy target, latency bound, recurrence per query
    data: &DataWorkload,                         // with its bounds (rows, G, source rates)
    accuracy_checks: &AccuracyCheckInputs<'_>,   // { model: &dyn AccuracyModel, evidence: &dyn AccuracyEvidenceProvider }
    cost_inputs: &CostInputs,                    // asap-cost; cost model 1 only
    realizer: &dyn AssignmentRealizer,           // asap_types::stages; implemented by Stage 2, built by the facade
    deadline: Option<Instant>,                   // the only budget: one MILP solve, one realization
) -> Result<Selection, ErrorSelectingPlan>;      // infeasible | no plan by the deadline | RealizerRejected | UnsupportedCostModel

pub struct Selection {
    pub selected_dag: DagOfSelectedCandidates,                        // candidate_of_node, edges, node_of_output_reference
    pub nodes_with_proven_non_negative_weights: Vec<SelectedNodeId>,  // count-min proofs; the facade applies them
    pub cost: PlanCost,                                               // asap_cost::plan_cost
    pub accepted_accuracy_per_query: Vec<AcceptedAccuracy>,
    pub pruned_candidates: Vec<PrunedCandidate<PhysicalCandidateId>>,
    pub search_report: SearchReport,  // status (Optimal | DeadlineReached), cost_lower_bound_per_second (HiGHS dual
                                      // bound; relative_gap() is a method), milp_variables, milp_constraints,
                                      // model_construction_ms, solver_ms, validation_ms
}

DagOfSelectedCandidates::validate(&physical) checks one allowed producer per input, acyclicity, outputs from query_output_candidates, and reader constraints. AssignmentRealizer::realize(&physical, &selected) -> Result<RealizedPlan, PhysicalCandidateError> builds RealizedPlan { plan, pane_retention_ms, operator_node_of_selected_node }; Stage 3 calls it once to validate, and the facade again to build the final plan. A rejection is a Stage 2 catalog bug (an illegal combination allowed), returned as RealizerRejected, never cut and re-solved. Only select_plan, the report types and large_peak_memory_warning are pub. raw_data_retained is a Stage 2 input only, and ignore_accuracy_guarantees arrives as an accept-all AccuracyModel.

2. Problem structure

Stage 3 makes two kinds of choices among candidates: exact or summary (§2.1) and reuse or not (§2.2). The second splits into four subproblems: which readers share one result (§2.3), which shared result each reader takes (§2.4), where to cut (§2.5), and which window a shared result covers (§2.6). §2.7 explains why the four must be solved together. Each step names the property (O1–O9) that the formulation in §3 relies on.

2.1 Choice 1: exact or summary

A summary is a local replacement for an exact computation. q1 can compute its quantile exactly over the raw samples in its 15 m window, or merge KLL panes. Which one is cheaper depends on scale (Figure 2). Exact cost grows with the rows in the window. A summary starts higher (build, kept state) but stays nearly flat, as long as its size meets the accuracy target. The crossover comes from measured costs (asap-cost with the sketch benchmarks), not from a rule. So both are candidates of q1's root group, and the choice is one of Stage 3's one-hot choices.

Figure 2. Choice 1 for q1's quantile.

 cost per second
   ▲                              exact: grows with rows in the window
   │                            ╱
   │                          ╱
   │                        ╱
   │  ────────────────────╳─────────────────  summary: build + kept state, nearly flat
   │                    ╱ ▲
   │                  ╱   crossover, from measured costs
   │                ╱
   │              ╱
   └───────────────────────────────────────►  scale (series × samples in the window)

O2. Choose per memo group. A memo group holds the candidates that give the same output result or the same summary state. The KLL chains with k = 200 and k = 400 are two candidates of one group, since a summary state's size configurations are candidates of its group; a DDSketch chain is another group. Because every candidate of a group gives the same result, a reader's input statistics do not depend on which candidate of the input group is picked. "Which candidate gives group g" is therefore a one-hot choice, and Stage 2's timing, materialization and pane rules only gate which producers a reader may use (compat).

Without sharing, choice 1 would be local. Each query picks its cheapest candidate per group with a DP over its own plan tree. Sharing breaks this.

2.2 Choice 2: reuse or not

One producer may serve several consumers: common subexpressions within one query, or shared results across queries (multi-query optimization). Three ideas decide it, from naive to complete:

• Enumerate. Compare "reuse" and "not reuse" per consumer, each with its own cost.
• Amortize. A shared result's cost is spread over its readers: $\big(P_m + \sum_{i=1}^{n} Q_i\big)/n$ per consumer.
• Choose the readers. Which consumers share the producer is itself a choice. The chosen readers decide the shared result's size and accuracy, so $P_m$ is not fixed: for one kind of result, adding readers can only raise its cost.

Existing view selection assumes the opposite. View selectors score a view $j$ with a fixed benefit per reader ($u_{ij}$ in BigSubs, a cost-model estimate in the tuners and learned selectors). Consumers are independent, so views are chosen first and each query's optimizer uses them afterwards. That works when a view is computed from base data at a fixed cost and its readers' costs do not depend on one another. Neither holds in Stage 3. The running example shows three ways the benefit of keeping a result depends on choices made elsewhere:

• The kept result's own plan. Materializing q3's sum pays off only if, inside its context, the Sum and Count builds read one TimeRange(Scan) run. With two scan runs the same view costs more than q3 and q4 computing their answers separately. That is exactly the plan Stage 3 keeps a MatAtQuery build whose NotMaterialized sibling avoids a second scan (#923 workload) #1142 records, where the view was kept but planned badly. The view's own plan may in turn read other kept results, such as an ingestion-time pre-aggregate. So the cost is a chain, not a fixed number (§2.5).
• The readers' plans and accuracy. The shared KLL chain is cheaper than a DDSketch only if both q1 and q2 read it, and only at a size whose error fits both queries' accuracy budgets. Pick the size first and the readers afterwards, and you either over-size the chain or force one reader onto a second sketch. Stage 3 misses cheaper plans in Stage 2's output: rangeheavy t5-q09, or-query and three-quantile workloads #1149's or-query is this case (§2.3).
• Costs that are maxima over readers. A shared result is held until its slowest reader finishes and runs at the rate of its schedule group. Its source keeps raw data as far back as the widest window reading it. Adding one reader can raise all three, so a view's benefit is not a sum of per-reader terms (§2.3).

2.3 Subproblem 1: fix the cut, choose the readers

Fix one shared result $m$, say the KLL chain: a cut in the plan. Its possible readers are $C_1, \dots, C_n$. A set $S$ of them reads $m$; every other reader $j$ computes its own input with producer cost $P_j$ (Figure 3). Then
$$C_{\text{total}}(S) = P_m(S) + \sum_{i \in S} Q_i + \sum_{j \notin S} \big(P_j + Q_j\big).$$

Figure 3. Subproblem 1: one shared result m at a fixed cut.

        Q_1      Q_2      Q_3                Q_4
         ▲        ▲        ▲                  ▲
  ───────┼────────┼────────┼──── cut ────     │
         └────────┼────────┘                [P_4]   C_4 ∉ S: its own producer
               ┌──┴──┐
               │  m  │  P_m(S): size  = tightest accuracy budget in S
               └─────┘          held  = slowest reader in S
                                rate  = fastest schedule group in S
                                raw   = widest window in S

$P_m(S)$ is a maximum over readers. Each part of $m$'s cost is set by the most demanding reader in $S$: size by the tightest accuracy budget, held time by the slowest reader, rate by the fastest schedule group, and raw retention by the widest window. So $P_m$ never decreases as $S$ grows. $C_{\text{total}}$ can still go either way: adding reader $j$ saves $P_j$ and costs $P_m(S \cup {j}) - P_m(S)$.

Consequence: readers join by threshold. Sort the readers by one requirement, e.g. accuracy budget (Figure 4). Once $S$ contains the tightest reader, adding any looser reader costs nothing on that dimension. So an optimal $S$ is "every reader up to a threshold", giving $n + 1$ options instead of $2^n$. With several requirements, there is one threshold per dimension.

Figure 4. Readers of one KLL chain, sorted by accuracy budget (loosest first).

   reader         C_a     q1      q2      C_d
   max_error      0.10    0.05    0.05    0.01
   threshold 0.05 ├───────────────────┤            one size for ε = 0.05: C_a, q1, q2 join at no extra size
   threshold 0.01 ├───────────────────────────┤    one larger size serves all four

$Q_i$ is not quite fixed: merging a larger sketch costs more. But it depends only on the level $m$ is built at, so it moves with the threshold.

The formulation uses levels, not a free maximum. The size configurations of a summary state are candidates of its group (O2). They are the levels in Figure 4, and each reader picks one that meets its budget (row 8), so "size = tightest reader" is a choice among levels. The schedule-group rate, raw retention and scan windows are threshold binaries too. Held time and pane retention stay epigraph variables over continuous latency and span (row 10).

O9. Exact cost from one implementation. Every reader-set-dependent term becomes a level choice, a threshold binary or an epigraph with non-negative cost, so minimization drives it to its true value. Every coefficient comes from asap-cost; Stage 3 computes no cost of its own.

Which producers can merge into one cut. The analysis assumes that each $P_j$ could be served by $m$. Those producers must be found first. They are exactly the candidates of one memo group whose timing, materialization and panes Stage 2 allows the reader to use (O2, compat), so this grouping is given by the memo and Stage 2.

2.4 Subproblem 2: several shared results, each reader takes one

With the cuts fixed there are several shared results, $m_1, \dots, m_k$, and many reader inputs. Each shared result can serve several readers, and each reader input takes one of them or computes its own (Figure 5). This is BigSubs' bipartite graph of subexpressions and queries. It is not a matching, since one result serves many readers and costs something to open. It is uncapacitated facility location: shared results are facilities, reader inputs are customers, and the opening cost is $P_m(S)$ from §2.3. Facility location is NP-hard in general, and here the opening cost also depends on which customers connect.

Figure 5. Subproblem 2: shared results (facilities, open if x[m] = 1) and reader inputs (customers).

   shared results                        reader inputs
   ┌──────────────────────┐
   │ KLL chain, k = 200   │─────────────●  q1: merge 15 m of panes
   └──────────────────────┘╲
                            ╲───────────●  q2: merge 1 h of panes
   ┌──────────────────────┐
   │ KLL chain, k = 400   │   not opened
   └──────────────────────┘
   ┌──────────────────────┐
   │ DDSketch chain       │   not opened
   └──────────────────────┘
   ┌──────────────────────┐─────────────●  q3: root Σ
   │ Σ as MatAtQuery      │
   └──────────────────────┘─────────────●  q4: roll-up reads Σ
                                         ●  a reader with its own producer: P_j + Q_j

Two things make it harder than textbook facility location.

• A reader may have several inputs. It takes one result per input: q3's avg reads a Sum and a Count, and Stage 3 misses cheaper plans in Stage 2's output: rangeheavy t5-q09, or-query and three-quantile workloads #1149's or-query reads two sketches.
• The inputs depend on the reader's own plan. Exact or summary (§2.1) decides whether q1 reads a KLL chain at all. So the assignment and every reader's plan are chosen together.

O6. Reader constraints name output references. AllowedOutputs and IncompatibleOutputs refer to output references, so every reader constraint is written per output reference, not per query.

§5 (d) uses this structure directly. With the open results fixed, each query's plan is a separate problem; column generation or Lagrangian relaxation prices the open results.

2.5 Subproblem 3: where to cut

The cuts are not given. Any memo group with a shareable candidate can be cut, and the candidates come from enumerating the memo. Cuts nest along a chain (Figure 6, left). q3's sum can be cut at the sum, at the builds, or at an ingestion-time pre-aggregate below them. The cost of an upper cut depends on whether a lower one is open. That is #1142.

Figure 6. Subproblem 3. Left: nested cuts along q3's plan. Right: the two kinds of shared node.

          q3      q4                                  across contexts: cut        within one context: O4
            ╲    ╱                                      q1          q2              Σ's context
            [ Σ ]        ─── cut 3: Σ as MatAtQuery       ╲        ╱              Sum      Count
              │                                       ═════╪══════╪═════ cut          ╲    ╱
            [avg]                                         [KLL chain]                 [ T ]
            ╱   ╲                                      charged once on x[m]      one candidate,
        [Sum]   [Count]  ─── cut 2: builds kept                                  run once per execution
            ╲   ╱
            [ T ]        ─── cut 1: ingestion-time pre-aggregate
              │
            [ S ]

Optimal substructure, and where it breaks. If a plan is a tree, its cost is the sum of its subtrees' costs, $C_{\text{total}} = C_l + C_r$. A DP over memo groups can then keep, per group, the Pareto frontier over (cost, latency, approximate shares used). A node with two readers breaks this. In q3's sum, the Sum and Count builds both read TimeRange(Scan). The DP charges the scan once under each build, and it may pick a different scan candidate for each, but the plan runs one scan (Roy et al.'s $\mathrm{numuses}$). Shared nodes come in two kinds, handled differently (Figure 6, right):

So a context is a region in which every node runs once per execution, and the DP's "one best candidate per group" holds inside it.

O3. Sharing depends on where a candidate runs. A candidate runs once for all its readers if it runs at ingestion time or is materialized; otherwise it runs once per place that needs it. We call such a place an execution context: a region between cuts. Each has its own executions per second:

• (a) a query evaluation q, at rate(q);
• (b) an ingestion pipeline I_r per refresh window r (pane width and phase, Refresh ingestion-time views that share a refresh window in one run, and charge a shared stateless ingestion node once per run #1151): the deployment refreshes all ingestion-time views of one refresh window in one run, so a NotMaterialized (stateless) ingestion-time candidate runs once per second of arriving data in each I_r that needs it;
• (c) each selected shareable candidate m (MatAtIngestion state or sketch configuration, pane chain, MatAtQuery, MatAtQueryKeptAcrossEvaluations, pane chain kept across evaluations): the context of m's own inputs, at m's rate (1 at ingestion time, its schedule-group rate at query time).

The KLL chain is a context of kind (c) at rate 1, and its input, the ingestion scan, sits in I_{10 s}. This is exact, since I_r runs a stateless candidate once per second whoever in it reads it; chains of different pane widths run in different I_r, and each scans. It is also small: one ingestion scan serves the KLL and DDSketch chains alike when they have the same pane width, and an ingestion-time m's context holds only one boundary read per input. q3's materialized sum is a context at the rate of schedule group {q3, q4}. Its avg, Sum build, Count build and TimeRange(Scan) run inside it, once per execution. A context reads a shareable candidate or a candidate of an I_r across a boundary, without recomputing what is below it. Latency and accuracy reach through m.

O1. Copies give the solver nothing to choose. q3 and q4 may each run a NotMaterialized TimeRange(Scan) in their own evaluations. These are two runs of one candidate, not two candidates. So Stage 2 emits one candidate per (operator node, timing, materialization), and Stage 3 decides how often each candidate runs: a NotMaterialized query-time candidate once per context that selects it, a shareable candidate once. The selection is a DAG of runs, so a candidate that runs twice is two nodes.

O4. One candidate per group per context. Inside a region the DP's rule applies: one best candidate per group. In q3's sum context, the Sum and Count builds both read the TimeRange(Scan) group. If each could pick its own run, the plan would scan twice. So within a context a group is given by at most one candidate and computed once, however many readers it has there. Across contexts there is no such limit. Two configurations of one summary state can both be selected, since each is its own context and each reader context picks one. A query that reads a group through an ingestion-time sketch and also directly at query time may use different candidates of the group, one picked in an I_r and one in q.

The rule gives up three kinds of plans, each visible as a cost change in the exactness comparison:

• two readers in one context that need, or are cheaper with, different candidates of one group, e.g. disjoint compat sets, or two size configurations of one summary state read in one query evaluation;
• one shared query-time candidate run twice over producers that differ in latency or accuracy for different readers (O5);
• two output references of one query with the same root group, served by different candidates.

Multi-query optimizers make a stronger assumption: one best plan per equivalence node for the whole plan (Roy et al.). BigSubs forbids two overlapping subexpressions in one query. Both lose these plans too, without stating it (§7).

O5. One context per shareable candidate, no instance index. A second run of q3's sum over other producers gives the same result at higher cost, so under cost model 1 it never pays off. The argument fails only when producers differ in latency or accuracy, which are constraints, not costs: q1 with a tight latency bound needs m over an ingestion-time pre-aggregate, while q2 with an exact target needs m over the raw scan. That is the second lost class above. In return, each shareable candidate needs one binary x[m] and no instance variables or symmetry breaking. This is the standard assumption in multi-query optimization and view selection. A materialized node runs once with one plan (Roy et al.), and there is one $z_j$ per subexpression (BigSubs). When latency bounds call for a second copy, Kodiak adds a coarser cube, which in our model is a different group (§7).

O7. Acyclic by construction. The memo group graph is acyclic and p follows group edges, so a selection needs no ordering rows. Contexts follow the same graph, so "q depends on m" is well founded.

O8. Choose cuts with binaries, not with DP keys. A DP could carry the cuts in its state: per group, a Pareto frontier keyed by the shared results the part reads. But the keys multiply. Binary operators multiply both sides, almost every key differs, and dominance prunes nothing. One binary per (context, group, candidate) makes the same choices with a model that grows with the sum of choices. The DP stays where no cut lies: private subtrees enter the MILP as Pareto options (§5, 3b).

2.6 Subproblem 4: choose the window

A shared result over time has a window shape as well as a size: the pane width of a pane chain, and the window of a shared query-time scan. Both are set by the readers, in the same way the size is (§2.3).

Pane width. A pane chain cuts the stream into panes of width $w$. Each reader merges the panes that cover its window (Li et al., "No pane, no gain"; Arasu & Widom for several queries). Reader $i$ can use width $w$ only if $w$ divides its grid $\gamma_i = \gcd(\text{window}_i, \text{interval}i)$. So a set $S$ of readers can share one chain only at a width that divides $\gamma(S) = \gcd{i \in S} \gamma_i$ (Figure 7).

Figure 7. Pane widths for one KLL chain. Reader i can use width w only if w divides γ_i = gcd(window_i, interval_i).

   q1: γ = gcd(15 m, 10 s) = 10 s          q2: γ = gcd(1 h, 60 s) = 60 s

   width w   readers that can use it   per second: panes closed   kept: panes   per evaluation: panes merged
   60 s      q2                        1/60                       60            q2: 60
   10 s      q1, q2   ← γ({q1, q2})    1/10                       360           q1: 90, q2: 360
    5 s      q1, q2                    1/5                        720           q1: 180, q2: 720
    1 s      q1, q2                    1                          3,600         q1: 900, q2: 3,600

The structure is subproblem 1's, on the divisor lattice instead of a sorted list:

• Adding readers can only make the width finer. $\gamma(S \cup {j})$ divides $\gamma(S)$. A finer width closes more panes per second and keeps more panes. Unlike accuracy, it also raises every reader's own cost $Q_i$, since each evaluation merges $\text{window}_i / w$ panes. So adding a reader with a fine grid can make the chain worse for every reader already on it. Then a second chain at a coarser width, for the coarse readers, is the better plan.
• For fixed readers, the coarsest legal width is enough if cost does not increase with $w$ (closing, keeping and merging panes all fall as $w$ grows). Then the widths worth generating are the grids $\gamma_i$ and their pairwise gcds, closed under gcd. These are the "levels" of the window, as sizes are for accuracy. Every other divisor gives the same reader set at higher cost.

Today Pass 2 generates widths over the full divisor domain (336 PaneChain groups for one query in #1048). Because that is too many to generate at once, the facade runs Stage 3 once per pane-width round (G9). If the precondition holds, the gcd closure is a much smaller domain that still contains the optimum. Then a single solve over it could replace the rounds (§8).

Scan window. A shared query-time scan reads the union of its readers' (offset, width) intervals. Its far end is the largest reach of any reader, and it sets the scan's cost and the source's raw retention. Its near end is the smallest offset. Both are maxima (minima) over readers, so they take the threshold encoding of §2.3: $\lambda_{\text{far}}$, $\lambda_{\text{near}}$ and $v$ in row 10. The precondition is the same: usage does not decrease as the window grows at either end (§4).

In the formulation, each pane width is a different group (a PaneChain shared build per width). The choice of width is therefore subproblem 2's choice of facility, and the readers that may use a width are fixed by compat.

2.7 Why one problem

The four subproblems share their decisions. The readers chosen in §2.3 set the opening cost that §2.4 pays, and its window (§2.6). Both depend on where §2.5 cuts. The cuts in turn depend on each reader's choice between exact and summary (§2.1). Solving them in sequence (cuts, then readers, then plans) repeats the two-step mistake of §2.2. So §3 encodes them in one MILP:

3. Formulation

We encode Stage 3 as one MILP, built in four steps. §3.2 chooses a candidate per group in each context. §3.3 shares candidates across contexts and tracks which queries depend on them. §3.4 adds latency and accuracy. §3.5 encodes the cost. Notation and the preprocessing that runs before the model is built come first (§3.1). We illustrate each step on the running example, whose groups and contexts Figure 1 shows. Table 1 lists all variables and Table 2 all constraints.

Figure 1. Running example: memo groups and the execution contexts [..] that choose a candidate for them.

[q1]  root Q1: merge 15 m of panes ─┐            [q3]  root Σ: read m_Σ across the boundary ──┐
[q2]  root Q2: merge 1 h of panes ──┤            [q4]  root Q4: roll-up of Σ ─────────────────┤
                                    ▼                                                          ▼
[m = KLL chain, rate 1]  K = {KLL k=200, KLL k=400}  [m_Σ = Σ as MatAtQuery, rate(G), G = {q3, q4}]
[m = DDSketch chain]     D = {DDSketch}                 Σ ← avg ← {Sum build, Count build} ← T: TimeRange(Scan)
                                    │
                                    ▼
[I_10s, rate 1]          S: ingestion scan, read by every ingestion-time chain with 10 s panes

3.1 Notation and preprocessing

Candidate dominance (exact under cost model 1). Delete $c$ when some $c'$ has all of the following:

• the same memo group, timing, materialization, shareability and held state;
• the same producers per input ($\mathrm{compat}(c',n) = \mathrm{compat}(c,n)$);
• the same readers (every (reader candidate, input) that may read $c$ may read $c'$, and nothing else), output references, reader constraints and keep flags;
• the same $(\mathrm{approx}, \mathrm{parts})$ and a guarantee at least as strong;
• the same output schema, sources, state family and operator kind (pane chains: the same panes), so readers are costed with the same formulas;
• every number at most $c$'s: CPU operations and scanned bytes per execution, latency, output rows and bytes, kept memory (average and peak), disk bytes, held bytes and raw data kept.

Both usages must be plan-independent. A candidate whose usage depends on a scan window or pane retention is compared only with one whose numbers are at most its own at every level. Assumption: a reader's usage is non-decreasing in its input's rows and bytes. Among mutually dominating candidates the smallest id stays, and the deleted candidate's readers read the dominating one. No optimal plan uses a dominated candidate. This is the only pruning; shared-candidate costs stay in the MILP.

Accuracy weights (composition). For each query $q$ and each approximate $c$ that $q$ can reach in any context, we rebuild $q$'s outputs with $c$ approximate and every other candidate exact. asap-accuracy composes the guarantees and runs the selected plan's check (guarantee_check, deployment model):

• every output is exact ($c$ does not reach the output): no weight;
• every output meets the target: $w(q,c) = 1$, one share of $q$'s budget (asap_accuracy::budget);
• otherwise $c$ is unusable for $q$.

Some pairs of candidates are built with both approximate: those usable alone, both sized for at least two shares, whose values meet in one plan of $q$. A failing pair is excluded. The candidates of one operator node are checked once, together.

These rules are conservative, so A1 may reject a plan that satisfies accepts:

• a candidate that fails through one reader of $q$ is unusable through all;
• a candidate or pair that exceeds MAX_COMPOSITIONS builds is unusable (NotChecked);
• a pair with a member sized for one share is excluded by the shares row.

Three or more approximate values are never built together. There, A1 relies on the equal-split proof of asap_accuracy::budget, and the exactness gate must include such queries. Pass 1 already prunes structurally unboundable candidates (guarantee None, e.g. a relative bound through avg), so unusable weights are rare.

Table 1. Variables.

Table 2. Constraints.

3.2 Choosing a candidate per context

We begin with the choices inside a context. Let the binary variable $p[\kappa,g,c]$ indicate that in context $\kappa$, group $g$ is given by candidate $c$. A valid choice must serve every output (1), give each group at most one candidate per context (2), fill every input with an allowed producer (3), and select nothing that no one reads (4).

Constraint (1). For each output reference $r$ we add $\sum_{c \in C(\mathrm{root}(r))} p[q(r), \mathrm{root}(r), c] = 1$, so every output is served by exactly one candidate.

Constraint (2). For each $\kappa$ and $g \in G(\kappa)$ we add $\sum_{c \in C(g)} p[\kappa,g,c] \le 1$. Within one context, a group is computed at most once, however many readers it has there (O4).

Example 3.1. In the context of q3's materialized sum $m_\Sigma$, the Sum build and the Count build both read group $T$, TimeRange(Scan). Constraint (2) gives $T$ one candidate in $m_\Sigma$, so a single run of the scan serves both builds and is charged once per execution of $m_\Sigma$. A plan in which the Sum build is materialized on its own (MatAtQuery, hence its own context) is still expressible. There, $T$ runs in two contexts and is charged twice, so the solver picks the shared run whenever it is cheaper.

Constraint (3). For each $c$ that runs in $\kappa$ and each input $n$ of $c$, and for $m$'s own inputs, we add
$$\textstyle\sum_{c' \in \mathrm{compat}(c,n)} p[\kappa, h(c,n), c'] \ge p[\kappa,g,c], \qquad \sum_{c' \in \mathrm{compat}(m,n)} p[m, h(m,n), c'] \ge x[m].$$
A boundary read also needs what it reads: $p[\kappa,g,c] \le x[c]$ for shareable $c$, and $p[m,g,c] \le p[I_{r(m)},g,c]$ for $c$ of $I_{r(m)}$ read by an ingestion-time $m$. compat carries Stage 2's timing, materialization and pane rules, the way physical properties gate implementation choices in MILP join ordering.

Example 3.2. q1's candidate "merge 15 m of KLL panes" reads group $K$, and its compat lists only KLL pane chains. Selecting it forces $p[q_1, K, c'] = 1$ for some KLL chain $c'$, say k = 200. That chain is read across the boundary, so $p[q_1,K,c'] \le x[c']$ selects it.

Constraint (4). For each $\kappa$, $g \in G(\kappa)$ and $c'$, we add
$$p[\kappa,g,c'] \le \sum_{\substack{(g'',c'') \text{ in } \kappa \text{ reading } g \ c' \in \mathrm{compat}(c'',n)}} p[\kappa,g'',c''] + \mathbb{1}[g = \mathrm{root}(r),, \kappa = q(r)] + \mathbb{1}[\kappa = m,, g \text{ input of } m], x[m] + \mathbb{1}[\kappa = I_r] \sum_{m:, r(m) = r} p[m,g,c'].$$
A run is selected only if a reader in its context, an output reference, $m$ itself, or (for $I_r$) a shareable ingestion-time context of refresh window $r$ reads it. This makes $p$ exact, which (6) and (9) need.

Example 3.3. Suppose Stage 2 puts MinimumQueryReaders(2) on the KLL chain, and q2's merge reads the DDSketch chain instead. Without (4), the solver could still set $p[q_2,K,\text{KLL}] = 1$, a phantom read that makes q2 count as a reader. Constraint (4) forbids this, because nothing in q2's context reads $K$.

3.3 Sharing and reach

Next we connect contexts. Let $x[c]$ indicate that shareable candidate $c$ is selected. Let $a[q,c]$ indicate that query $q$ depends on $c$, directly or through other shareable candidates. Let $z[m,\mathcal{G}]$ indicate that schedule group $\mathcal{G}$ runs shareable query-time candidate $m$.

Constraint (5). For shareable $m \in C(g)$ we add $x[m] \ge p[\kappa,g,m]$ for every reading $\kappa$, and $x[m] \le \sum_\kappa p[\kappa,g,m]$. Each run is then charged where it runs:

• a shareable candidate once on $x[c]$ (kept state, and the ingestion work of a shareable ingestion-time candidate), plus, at query time, on $z[m,\mathcal{G}]$ at the schedule group's rate;
• a candidate of $I_r$ once on $p[I_r,g,c]$ at rate 1, however many contexts of refresh window $r$ read it; once per $I_r$ when chains of several refresh windows read it;
• a NotMaterialized query-time candidate on $p[q,g,c]$ at $\mathrm{rate}(q)$, or on $\mathrm{AND}(z[m,\mathcal{G}], p[m,g,c])$ at $\mathrm{rate}(\mathcal{G})$ in context $m$.

Example 3.4. q1 and q2 both read the KLL chain: $p[q_1,K,c'] = p[q_2,K,c'] = 1$ force $x[c'] = 1$, and the chain's standing cost is charged once. If a DDSketch chain with 10 s panes is also selected, both chains read the ingestion scan $S$ in $I_{10 s}$. The cost sits on $p[I_{10 s},S,s]$ once, and the two boundary reads $p[m,S,s] \le p[I_{10 s},S,s]$ cost nothing. A DDSketch chain with 60 s panes would read $S$ in $I_{60 s}$, charged a second time.

Constraint (6). For query $q$ and shareable $c$ in group $g$, we add
$$a[q,c] \ge p[q,g,c], \quad a[q,c] \ge a[q,m] + p[m,g,c] - 1, \quad a[q,c] \le p[q,g,c] + \textstyle\sum_m \mathrm{AND}(a[q,m], p[m,g,c]),$$
with the ANDs two-sided. Shareable ingestion-time candidates are contexts too, so chains of ingestion state are covered, such as a pane chain over a sketch. An $I_r$ holds no approximate candidate and no compaction, so no reach into it is needed. The lower rows suffice for cost, latency and accuracy, where a larger $a$ only hurts. The upper rows stop a free $a$ from satisfying MinimumQueryReaders. Reach is well founded because contexts follow the acyclic group graph (O7).

Constraint (9). Reader constraints are written per output reference on $p$, $x$ and $a$ (O6). For example, MinimumQueryReaders(k) on shareable $c$ is $\sum_q a[q,c] \ge k, x[c]$, and AllowedOutputs on $r$ fixes $p[q(r), \mathrm{root}(r), c] = 0$ for each disallowed $c$.

3.4 Latency and accuracy

Let $L[\kappa]$ be the latency of context $\kappa$ in ms. Let $T[q,m]$ be $m$'s latency as counted in $q$'s. Let $K[q]$ be the largest compaction among the ingestion-time candidates $q$ depends on.

Constraint (7). We count latency per run, with plan_cost's definition:

• $L[\kappa] \ge \sum_{g \in G(\kappa)} \sum_{c \text{ runs in } \kappa} d(c), p[\kappa,g,c] + \sum_{\text{query-time } m \text{ read in } \kappa} T[\kappa,m]$ for a query or a shareable query-time context, and $L[m]$ adds $d(m)$. Ingestion-time contexts add only compaction.
• $T[q,m] \ge L[m] - \bar L[m](1 - a[q,m])$, one row per $(q,m)$ with $m$ at query time.
• $L[q] \ge \dots + K[q]$ with $K[q] \ge k(c), a[q,c]$ for each shareable ingestion-time $c$, and $L[q] \le \mathrm{bound}(q)$.
• A window-dependent scan adds its window levels' $\Delta d$ (constraint 10).

A (context, group) pair is one run and counts once. $d(c)$ is work_ms_per_execution, the number plan_cost sums. Latency variables appear only in $\le$ bound rows and in held-time epigraphs with positive cost, so they take their true values.

Example 3.5. q4's latency is its roll-up plus $T[q_4, m_\Sigma]$. Here $L[m_\Sigma]$ adds $d$ of $\Sigma$, avg, the two builds and the single run of $T$, each once, however many paths in q4's plan reach $m_\Sigma$.

Constraint (8), accuracy (A1). Let $\mathrm{dep}(q,c)$ be:

• $p[q,g,c]$ for $c$ computed in $q$;
• $a[q,c]$ for shareable $c$;
• $y[q,m,g,c] \ge a[q,m] + p[m,g,c] - 1$ for $c$ computed in context $m$.

With $\bar N_q$ the most approximate values $q$ can depend on, we add
$$N_q = \textstyle\sum_{c:,w(q,c)=1} \mathrm{dep}(q,c), \qquad N_q \le \mathrm{parts}(c) + (\bar N_q - \mathrm{parts}(c))(1 - \mathrm{dep}(q,c)) \ \ \forall c: w(q,c)=1,$$
together with $\mathrm{dep}(q,c) = 0$ for unusable $c$ and $\mathrm{dep}(q,c_1) + \mathrm{dep}(q,c_2) \le 1$ for excluded pairs. Each approximate value $q$ depends on counts as one share. Each $\mathrm{dep}$ appears only where a larger value tightens a row, so the one-sided $y$ is exact. Pass 1 sizes each approximate value for an equal share of $\mathrm{parts}(c)$, so reading at most $\mathrm{parts}(c)$ shares meets the target for the checked compositions. Decode checks satisfies and returns an error on a miss.

Example 3.6. If the KLL chain was sized for one share ($\mathrm{parts} = 1$), then $\mathrm{dep}(q_1, \text{KLL}) = 1$ forces $N_{q_1} \le 1$. q1 may then depend on no other approximate value.

3.5 Cost

Finally we encode the cost. Cost per second is a sum over nodes of (work per execution × executions per second + bytes held × seconds held per second) × unit cost, plus raw retention. It has three layers, all computed by asap-cost.

• Layer 1, node usage, is plan-independent and computed once per candidate: asap_cost::node_usage(node, input statistics, data, speed, measured costs) -> NodeUsage { cpu_ops_per_execution, scanned_bytes_per_execution, work_ms_per_execution (via ResourceSpeed), state_bytes (kept result or state), working_memory_bytes }. An ingestion-time execution is one second of arriving data. Input statistics are those of the input groups (O2). Only a summary state's configuration and a scan's window change them.
• Layer 2, execution demand, depends on the readers and is decided by the MILP.
  • Executions per second: 1 for each $I_r$ and for shareable ingestion-time candidates; $\mathrm{rate}(q)$ in $q$. For a shared query-time $m$ and its context, the rate comes from the queries reaching $m$, under the current asap_cost::execution_demand policy: equal (interval, phase, step) take the maximum span, different ones add, and non-repeating queries take the batch maximum over the horizon. execution_demand::schedule_groups(readers) -> Vec<ScheduleGroup { readers, evaluations_per_second }> gives the grouping, and evaluation_rate is written on top of it.
  • Seconds held: ingestion-time and kept state, always (a kept pane chain grows with its readers' longest span). $m$'s result and its context's working memory, from completion until the slowest reader finishes. Working memory in a query context, for that evaluation's latency.
  • Raw retention: per source, the longest window among the selected query-time scans.
• Layer 3, cost model. Cost model 1 is Σ unit cost × usage, which is linear. asap_cost::plan_cost(dag, usage, demand, cost_inputs) -> PlanCost costs a whole plan, both for the final cost and for the decode check. Cost models 2 and 3 return UnsupportedCostModel. A follow-up issue adds them under the same layers: model 2 with peak memory and cores as linear sums, model 3 with limits as linear rows plus an epigraph for the worst-query latency.

milp/cost_terms.rs only turns these numbers into coefficients and auxiliary rows: NodeUsage × unit cost, one $z$ per ScheduleGroup, held-time epigraphs, and threshold levels. It has no usage formula, schedule rule or unit cost of its own.

Several terms are maxima over the readers (O9). Let $H_{\text{query}}[q,c]$, $H[m]$ and $W[m,c,\mathcal{G}]$ be held times. Let $S[m]$ or $\pi[m,\ell]$ be a kept pane chain's retention, $\lambda_{\text{far}}$ and $\lambda_{\text{near}}$ a scan's window levels, and $v[\mathit{src},\ell]$ a source's raw retention levels (Table 1).

Constraint (10). We add:

• Executions: $z[m,\mathcal{G}] \ge a[q,m]$ for $q \in \mathcal{G}$. The group's rate is that of its fastest reader, by levels.
• Working memory in a query context: $H_{\text{query}}[q,c] \ge L[q] - \bar L[q](1 - p[q,g,c])$.
• $m$'s result and the working memory in $m$'s context, held until the slowest reader finishes: $H[m] \ge L[q] - \bar L[q](1 - a[q,m])$ per reader $q$, and $W[m,c,\mathcal{G}] \ge H[m] - \bar H[m](2 - z[m,\mathcal{G}] - p[m,g,c])$, with one indicator for $c = m$.
• Kept pane chain: $S[m] \ge \mathrm{span}(q), a[q,m]$ when the cost is linear, or $\pi[m,\ell] \ge a[q,m]$ for each reader with $\mathrm{span}(q) \ge \ell$.
• Window-dependent scan: the window is the union of the readers' (offset, width) intervals (PhysicalASAPDAG::scan_window), with far end $\max(\text{offset}+\text{width})$ and near end $\min(\text{offset})$. We add $\lambda_{\text{far}}[\kappa,h,\ell] \ge p[\kappa,g,c]$ for each reader whose far end is at level $\ell$ or beyond, and $\lambda_{\text{near}}[\kappa,h,\ell] \ge p[\kappa,g,c]$ for each reader whose near end is at $\ell$ or closer. For a shared scan, these rows use $a$ or $x$.
• Raw retention: $v[\mathit{src},\ell] \ge \lambda_{\text{far}}$ of each scan that reads $\mathit{src}$ at level $\ell$ or beyond.

Example 3.7. q3 and q4 both read $m_\Sigma$, so $H[m_\Sigma] \ge L[q_3]$ and $H[m_\Sigma] \ge L[q_4]$. The sum's result, and the working memory of the builds in its context, are held until the slower of the two finishes. The rows cost held × unit memory cost × $\mathrm{rate}(\mathcal{G})$ through $W$.

Objective. We minimize:

Four linearizations appear:

• an epigraph for a maximum with linear cost: held time, compaction, latency through shared candidates, and pane retention when memory is linear in the span;
• threshold binaries for a maximum with step cost: raw retention, window-dependent scans in two dimensions, and pane retention otherwise (the threshold-OR max-delay encoding);
• an AND binary for a product of indicators;
• a big-M row, bounded by the continuous variable's own upper bound, for held time × selection.

With constraints (1)–(10) and this objective, every selection is a valid DAG of runs, and its objective is its cost (§4).

4. Decoding and exactness

Exactness. Every coefficient is non-negative, and every auxiliary variable (one-sided ANDs, epigraphs, thresholds) appears only where a larger value costs more or tightens a row. At an optimum each takes its true value, so the objective equals plan_cost. Two preconditions are assumptions, tested by the cost equality gate: usage, latency and raw retention are non-decreasing as a scan's window grows at either end; a kept pane chain's memory is non-decreasing in its readers' longest span. In the example, the ingestion scan read by both the KLL and DDSketch chains is charged once, on $p[I_{10 s},h,c]$ (both chains have 10 s panes), which row 4 makes 1 exactly when some context reads it; each sketch's read $p[m,h,c] \le p[I_{10 s},h,c]$ costs nothing itself. At a DeadlineReached incumbent, auxiliaries may sit above their true values, so the objective is an upper bound on plan_cost.

One solve, one realization. Every check that can reject a plan runs before or inside the MILP: cost is the objective; accuracy (A1) and latency are rows; count-min weight proofs are filtered per candidate. Raw-data retention is not a Stage 3 check, since Stage 2 generates no query-time raw scan without retained raw data. Executability of per-entity aggregates is a Stage 2 catalog rule (not generated, or stated as a reader constraint). Then:

1. Build DagOfSelectedCandidates: one node per (context, group) with $p[\kappa,g,c]=1$ and $c$ a NotMaterialized query-time candidate running in $\kappa$; one node per $x[c]=1$; in each context an edge from the node each input selects (row 3) to its reader, and a boundary read is an edge from the shared node. A TimeRange(Scan) selected in q3's sum context and in a query context decodes to two nodes; the KLL chain read by q1 and q2 to one.
2. Call the realizer once. It checks only Stage 2's combination rules (timing, dependencies, DAG validator).
3. Cost with plan_cost: equal to the objective within tolerance at Optimal, at most the objective at DeadlineReached (and reported). Otherwise return an error, never an unchecked plan.
4. Check satisfies. A miss is a bug in the weights or sizing: return an error, never cut and re-solve.
5. Test gate on the corpus and memo examples: plans A1 admits ⊆ plans satisfies admits; each plan only satisfies admits comes from a §3.1 conservative exclusion; the corpus includes queries with three or more approximate inputs in one composition.

The status is Optimal with gap 0, or DeadlineReached with HiGHS's dual bound as cost_lower_bound_per_second.

5. Solving strategy

An exact model can still be too hard to solve to optimality by the deadline on a large workload (§6.1). Following hybrid MILP join ordering, we use the MILP where sharing makes it necessary, and cheaper exact methods elsewhere. Four questions remain: (a) how to obtain a reference plan; (b) what couples queries; (c) which exact decompositions shrink the model; and (d) how to obtain a strong bound when they are not enough. Algorithm 1 combines the answers. Each step after the first is added only if measurements show it is needed. The acceptance gates are the same for every step.

(a) Reference plan and warm start. We solve one MILP, warm-started from a cheap reference plan: each query's cheapest plan, then greedy sharing. We then measure on workload-01 and rangeheavy-t8-t5: model size (the $m$ contexts separately), solve time and gap.

(b) What couples queries. Without sharing, the problem separates per query. Four things couple queries:

• (i) shareable candidates $x[c]$, such as the KLL chain;
• (ii) raw retention per source;
• (iii) schedule groups $z$;
• (iv) a shared query-time $m$ whose context enters its readers' latency and accuracy, such as $m_\Sigma$ read by q4.

(i)–(iii) appear only in the objective. (iv) appears in constraints.

(c) Exact decompositions.

• 3a. Connected components. Query sets that share no shareable candidate and no source are independent MILPs, solved in parallel. This gives nothing when all queries read one metric, as in the example, and helps on mixed-source workloads.
• 3b. Private subtrees as Pareto options. Consider a subtree below a group that only one query reaches and that reads no shareable candidate. It interacts with nothing else. A DP keeps its Pareto set over (cost, latency, approximate shares used), and the MILP sees those few options instead of the subtree's groups. The DP state has only the two dimensions latency and shares, so it does not multiply as in O8. The MILP then covers only the shared core. This subsumes fixing private groups whose best candidate dominates.

(d) Per-query decomposition (3c), only if 3a–3b are not enough.

• Structure. The problem has the structure of uncapacitated facility location. Shareable candidates are facilities with an opening cost, and each query picks one plan.
• Column generation. A column is one query's full plan. The master problem picks one column per query and the open shareable candidates. Pricing is a DP over the query's memo, with shareable candidates at their dual prices.
• Lagrangian relaxation. Relaxing $x[c] \ge p[\kappa,g,c]$ instead splits the problem into one subproblem per query and one per shareable candidate.
• Bound. Either bound is much stronger than the full MILP's LP relaxation. That relaxation gives about 0 on rangeheavy-t8-t5 and workload-01, so branching never starts.
• Coupling (iv). A shared query-time $m$ becomes a "customer" with its own columns, plus linking rows for its readers' latency and accuracy.
• First version. HiGHS has no branch-and-price, so the first version generates columns at the root only and solves the MILP over them. The result is not proven optimal, but the root bound gives a true gap, reported as DeadlineReached with cost_lower_bound_per_second.

Algorithm 1: select_plan
 1  Function SelectPlan(physical, entries, data, accuracy_checks, cost_inputs, realizer, deadline):
 2      // Preprocess (§3.1): dominance, accuracy weights
 3      cands ← prune_dominated(physical);  w ← accuracy_weights(cands, entries, accuracy_checks)
 4      // (a) Reference plan: each query's cheapest plan, then greedy sharing
 5      ref ← reference_plan(cands, cost_inputs)
 6      // (c) 3a, if needed: independent components (no shared candidate, no shared source)
 7      sol ← ∅
 8      for comp ∈ components(cands):
 9          // (c) 3b, if needed: private subtrees enter as Pareto options over (cost, latency, shares)
10          model ← build_milp(comp, w, cost_inputs)                 // constraints (1)–(10), §3.5 objective
11          sol ← sol ∪ solve(model, warm_start = ref, deadline)     // (d) 3c, if needed: root column generation
12      dag ← decode(sol)                                            // §4
13      plan ← realizer.realize(physical, dag)                       // once; rejection → RealizerRejected
14      check plan_cost(plan) = objective (≤ at DeadlineReached); check satisfies(plan)
15      return Selection

6. Model size

About $\sum_\kappa \sum_{g \in G(\kappa)} |C(g)|$ binaries $p$, plus one $x$ per shareable candidate, one $a$ per (query, reachable shareable candidate), one $z$ per (shareable query-time candidate, schedule group), and the threshold levels. Query contexts add up over queries. The $m$ contexts add $\sum_m |G(m)|$ over all possible shareable candidates, which the query count does not bound. An ingestion-time $m$ adds little (one boundary read per input); a query-time $m$ such as q3's sum carries a copy of the NotMaterialized choices below it.

6.1 Measured on the prototype (#1144)

Budget 300 s, accuracy enforced, aws-fargate-us-east-1, raw data retained, workload-01 under ulimit -v 8388608. The facade runs Stage 3 once per pane-width round; rows show the selected round, and the largest round where it differs.

Every selected plan's plan_cost equals the objective, and the MILP matches the reference enumeration on example1, ratio-of-quantiles and rangeheavy-t5-q05-q09. Memory is not the limit (under 1 GB everywhere). The weak LP bound is: on workload-01's largest round the incumbent never improves on the warm start (3.54e-6, dual 1.03e-6). This points to 3c for the bound, and to shared input contexts across materializations (§8) for size, ahead of 3b.

6.2 What the prototype implements

main still runs the old engine (search/refinement/: copies, propose → evaluate → cut). The design exists only in the draft prototype #1144 (feat/1095-group-milp, switched on by ASAP_STAGE3_SEARCH=milp). Its coverage of §2 and §5:

7. Related work

Multi-query optimization over an AND-OR DAG. Roy, Seshadri, Sudarshan & Bhobe (SIGMOD 2000) merge the queries into one AND-OR DAG. Equivalence (OR) nodes are results; operation (AND) nodes are ways to compute them; physical equivalence nodes, one per required property, refine the OR nodes. Given a materialized set $M$, Volcano keeps one best plan per equivalence node, and a node in $M$ is computed once with that plan. The number of times a node runs depends on its ancestors: $\mathrm{numuses}(e) = \sum_{p} U(p)$, where $U(p) = 1$ if $p \in M$ and $\mathrm{numuses}(p)$ otherwise. This coupling makes exhaustive search exponential, so Volcano-SH, Volcano-RU and a greedy heuristic are used instead. Kathuria & Sudarshan (PODS 2017) recast the greedy as maximizing a submodular benefit (which may be negative) and prove its approximation factor is the best possible unless P = NP. Mistry, Roy, Sudarshan & Ramamritham (SIGMOD 2001) use the same DAG to choose subexpressions kept permanently or transiently for view maintenance. Zhou, Larson, Freytag & Lehner (SIGMOD 2007) add covering subexpressions for similar consumers, and Silva, Larson & Zhou (ICDE 2012) share subexpressions in SCOPE scripts under each consumer's physical requirements.

Memo groups are the OR nodes, candidates the AND nodes, compat the physical-property refinement, and $x[c]$ the materialized set $M$. Execution contexts make $\mathrm{numuses}$ exact inside the MILP: a context is the "$U(p) = 1$" boundary, and a candidate runs at its context's rate (O3). O5 is Roy et al.'s assumption that a materialized node runs once with one plan. O4 is weaker than theirs: they fix one plan per equivalence node for the whole consolidated plan, while we fix one per (group, context). Two contexts may therefore pick different candidates of one group, e.g. an ingestion-time sketch and a query-time read.

Materialized-view selection at scale. BigSubs (Jindal, Karanasos, Rao & Patel, PVLDB 2018) selects SCOPE subexpressions with an ILP. In it, $z_j$ means subexpression $j$ is materialized and $y_{ij}$ means query $i$ reads it, with $y_{ij} \le z_j$ and a storage budget. An interaction row forbids one query from using two subexpressions that cover each other, a coarser form of O4. The global ILP does not scale, so BigSubs maps it to bipartite labeling: subexpression labels are flipped probabilistically, and every query solves a small local ILP given them. This is the heuristic counterpart of step 3c, which fixes the shared $x$ and solves per query, but updates the shared variables with Lagrangian prices or column generation. CloudViews (Jindal et al., SIGMOD 2018; production results EDBT 2021) finds repeated computation in recurring jobs by plan signatures, which correspond to our Stage 0 hash-consing. It selects views BigSubs-style and materializes each view online in the first job that computes it, like MatAtQuery in a schedule group. Anderson et al. (arXiv 2606.03772, 2026) add incremental view-maintenance cost to a BigSubs-style ILP and search it locally. In all three, a view's utility is additive and fixed per (query, view), and a view is computed from base data. Chains of materializations, maxima over readers, latency bounds and accuracy budgets are not modelled.

Kodiak (Liu, Song, Gangam, Lo & Elmeleegy, PVLDB 2016) selects cubes for an ad-analytics service. It minimizes weighted latency under latency bounds for SLA queries and a storage budget, maintains the cubes with an ETL pipeline, and routes each query to the smallest view that can answer it. It adds coarser cubes for SLA queries even when the finest cube could answer them, because rolling up from the finest cube at query time breaks the SLA. In our model a coarser cube is a different memo group, so that plan is kept. O5 loses only the narrower case of the same candidate built twice over different producers.

Workload-driven tuners select views together with indexes: AutoAdmin in SQL Server (Agrawal, Chaudhuri & Narasayya, VLDB 2000), and Oracle's automatic materialized views built on extended covering subexpressions (Ahmed, Bello, Witkowski & Kumar, PVLDB 2020). Learned selectors frame view selection as an ILP or MDP solved with deep reinforcement learning and a learned benefit estimator: RLView (Yuan et al., ICDE 2020) and AutoView (Han et al., ICDE 2021, TKDE 2022). Batch and dataflow systems share or cache intermediate results across jobs: MRShare (Nykiel et al., PVLDB 2010) groups MapReduce jobs by a cost model, Nectar (Gunda et al., OSDI 2010) caches derived datasets keyed by the program that computed them, and ReStore (Elghandour & Aboulnaga, PVLDB 2012) keeps whole-job and operator outputs of Pig workflows. These systems decide which results to keep or reuse; Stage 3 must also decide the plan of each kept result and of every reader, at exact shared cost.

In all of these, a view has a fixed benefit per reader and is chosen before the readers' plans; §2.2 shows three ways this fails in Stage 3, and §2.3–§2.5 give the structure that replaces it.

Shared aggregation over streams. Arasu & Widom (VLDB 2004) cut the stream at the window edges of all queries into slices, and keep partial aggregates per slice to share them across sliding-window queries; our pane chains and pane-width rounds play this role. Krishnamurthy, Wu & Franklin (SIGMOD 2006) share fragments across queries with different windows and predicates (our scan-window union and predicate-OR shared scans). Cutty (Carbone et al., CIKM 2016) shares aggregates for user-defined windows such as sessions, and Scotty (Traub et al., ICDE 2018) generalizes stream slicing to out-of-order data. Zhang, Koudas, Ooi & Srivastava (SIGMOD 2005) add phantoms in Gigascope's low-level operators: finer aggregations no query asked for, which feed several coarser group-bys. They choose the phantoms and size their hash tables under a memory budget with heuristics. The ingestion pipelines $I_r$ together with finer-aggregate reuse is the same idea, chosen exactly. Dobra, Garofalakis, Gehrke & Rastogi (EDBT 2004; Information Systems 2009) share sketches across aggregate queries, state when sharing preserves the estimates, and prove that finding the optimal sketch-sharing configuration is NP-hard. Row 8 and the shared KLL chain of q1 and q2 address the same problem; we size each sketch configuration as a candidate and choose the sharing in the MILP.

Choosing synopses for approximate queries. BlinkDB (Agarwal et al., EuroSys 2013) chooses stratified samples with a MILP that maximizes weighted template coverage under a storage budget, and picks a sample size per query from its error or latency bound. Taster (Olma, Papapetrou, Appuswamy & Ailamaki, ICDE 2019) injects samples and sketches into plans, materializes them, reuses them across queries, and adapts them online to the workload and a storage budget. Both choose synopses, but neither chooses the plans built on them jointly with sharing, latency and accuracy.

MILP for query optimization. Trummer & Koch (SIGMOD 2017) select join orders by MILP with a one-hot implementation choice per join position, gated by the producer's physical properties; rows 2–3 apply this pattern to memo groups with compat as the gate. Trummer & Koch (PVLDB 2016) encode MQO, one plan per query plus pairwise savings for shared work, as a QUBO; groups and contexts generalize the pairwise model. Schönberger, Trummer & Mauerer (2025) compute a heuristic reference plan and run the MILP only on the part that matters; §5 follows them with a warm start and the MILP on the shared core (3b). Step 3c uses standard facility-location and Dantzig–Wolfe decomposition.

What is new here. Prior work fixes or simplifies at least one of the following; this design handles all of them in one exact MILP:

• exact cost under shared execution, with rates, held time as a maximum over readers, and window unions, all from one cost implementation;
• chains of materializations, where a view's own plan is chosen jointly with its readers through contexts;
• per-query accuracy budgets composed across shared sketches;
• latency bounds that reach through shared results.

The prior work makes other simplifications: additive per-view utilities, views computed only from base data, heuristic sharing, or synopses chosen offline. BigSubs' labeling scheme and Roy et al.'s greedy are natural baselines and primal heuristics for §5.

8. Open points

In the design.

• Size of the $m$ contexts — measured, closed (perf(plan-selection): accuracy pair checks reuse built outputs; shared input contexts measured (#1095 §8) #1156). On the §6.1 workloads no operator node has two shareable candidates of one timing (every pair is an ingestion-time pane chain plus a query-time one), so sharing an input context removes no variable. The size comes from nested MatAtQuery candidates of different nodes, each copying the NotMaterialized stack below it; that is for 3b/3c. The 44 s build of rangeheavy-t8-t5's largest round was 96 % accuracy checks, now 7.6 s by keeping accuracy builds across checks.
• Lost plans (O4). The residual case, one shared candidate over producers differing in latency or accuracy, is the one to watch in the exactness comparison.
• Pane widths from the gcd closure (§2.6). If cost does not increase with pane width, generating only the readers' grids closed under gcd keeps the optimum and would let one solve replace the pane-width rounds (G9, [Refactor] Pane-width rounds without cross-stage calls: rerun Pass 2 per round, or pass an incremental Pass 2 state #1048). To check: the precondition in asap-cost (including per-pane sizing), then selected plans and costs of the gcd closure vs the complete domain on the corpus, Planner aborts with bad_alloc under 8 GB on random-mixed-seed-862 workload-01 when raw data is retained #1036 and the paper workloads.
• Levels for held time — measured, closed (feat(plan-selection): selection MILP solved per independent part, LP gap diagnosis, tighter big-M bounds (#1095 §5 3a, §6.1) #1155). The big-M families (held time, $V$, accuracy, latency) carry under 1 % of the LP gap, and dropping any of them keeps 97–100 % of the LP bound. The pure LP (2–7 % of the best cost) is weak because a query spreads over many alternative paths and pays a shared lower node only fractionally. HiGHS's root cuts recover 98–100 % of the bound, except on workload-01's largest round (72 %). The structural fix is 3c, where a column carries its query's full demand.
• Monotonicity preconditions of the scan-window and pane-retention encodings are assumptions. A counterexample in asap-cost must surface as a cost-equality failure, not a silent mis-ranking.

Found by the prototype, not yet part of the design. Each is an open question; the workaround is one possible answer.

9. Tracking list: from → to

[x] = already on refactor/stage-boundaries (#1047 P7, #1054); the rest starts after #1047 P7.

Interface

• route.rs / candidate_connection_search → select.rs / select_plan (§1 signature)
• Stage 3's raw_data_retained / ignore_accuracy_guarantees → AccuracyCheckInputs { model, evidence }; raw_data_retained Stage 2 only; accept-all model from the facade
• Selection.{selected_plan_id, plan, cost_by_plan/costs, accepted_accuracy_by_plan/accepted, front/pareto_front, rejected/rejected_plans} → deleted (the facade builds the plan from selected_dag); Selection.pane_retention_ms → RealizedPlan; CandidateCost → PlanCost
• SearchReport.{evaluations, refinements, milp_solves, max_evaluations, runner_up_cost_per_second}, EvaluationLimitReached, rejected_by_cause/search_rejected → pruned_candidates, pane_widths_pending → facade round state; new fields as in §1
• SearchReport.candidates_checked (still in report.rs) → deleted
• build_selected_physical_plan → AssignmentRealizer::realize; SelectedPhysicalCandidate → RealizedPlan; realizer rejection → ErrorSelectingPlan::RealizerRejected
• cost models 2/3 under the new search → UnsupportedCostModel
• crate-wide pub → only select_plan, report types, large_peak_memory_warning
• Stage 3 fields read by the DAG viewer (stages.js), tools/benchmark_pane_search.py, tools/validate_joint_workloads.py → kept or renamed in those readers in the same PR; Pass 2 / Stage 2 counts and timings → the facade's run report

Copies and the selection type

• Stage 2 scope copies (catalog/scopes.rs), PhysicalCandidateKey::scope_copy, PruneReason::PreviousScopeCopyPruned → one candidate per (operator node, timing, materialization); the old engine expands its copies in Stage 3 (search/candidates/scopes.rs) until the switch
• ReaderConstraint::ActivationRequires (old engine only) → deleted with the switch
• class layer (classes.rs, cap(k), "copies suffice") → deleted
• SelectedCandidates, CandidateInstance, ProducerConsumerConnection → DagOfSelectedCandidates, SelectedNodeId, SelectedEdge; nodes_by_candidate → operator_node_of_selected_node; candidates_with_proven_non_negative_weights → nodes_with_proven_non_negative_weights
• free fn validate_selected_connections (still in search/candidates/dag.rs) → DagOfSelectedCandidates::validate
• memo accuracy parts → OperatorNodeOfCandidateForMemoGroup::accuracy_budget_parts

Search

• refinement/planner.rs per-round joint-cost seed and the direct / baseline / neighborhood / seed phases → one MILP over execution contexts, warm-started from a reference plan
• evaluator loop (max_evaluations, EvaluationLimitReached, no-good and exact-cost cuts) → one solve, one realization
• plain lower-bound MILP (model.rs) → group MILP (milp/model.rs, rows 1–10)
• warm start over copies → warm start over groups
• price-option projection → deleted
• pane-width rounds → out of Stage 3 ([Refactor] One named type per stage boundary: PerComputationLogicalMemoCandidates → LogicalASAPMemoCandidates → PhysicalASAPMemoCandidates #1047 P7); keep the facade's round interface ([Refactor] Pane-width rounds without cross-stage calls: rerun Pass 2 per round, or pass an incremental Pass 2 state #1048) until the new search solves the full width domain at once on Planner aborts with bad_alloc under 8 GB on random-mixed-seed-862 workload-01 when raw data is retained #1036 and the paper workloads, then delete the rounds
• checks.rs per_entity_aggregate_violation → Stage 2 catalog rule (per_entity_aggregate_executable)
• Stage 3 raw-retention check → none (Stage 2 input); whole-plan count-min proof → per-candidate filter
• no Stage 3 pruning → candidate dominance, recorded in pruned_candidates (PruneReason)

Cost

• pricing.rs, price_node, PricedNode, price_at_scale → asap_cost::node_usage / NodeUsage, asap_cost::plan_cost
• general_schedule_rates.rs, QueryScheduleRates (re-implements evaluation_rate, :368-413) → asap_cost::execution_demand::schedule_groups, ScheduleGroup; evaluation_rate on top of it
• lower_bounds.rs, direct_profile.rs, supplier_price_options.rs (now producer_price_options.rs), direct_cost.rs (DirectCostProfile, NodeCostProfile), general_schedule.rs, general_schedule_rates.rs, direct_price_options.rs → milp/cost_terms.rs
• threshold-OR max-delay encoding (direct_cost.rs) → kept, on the group model (raw retention, scan windows)
• ranked enumeration for cost models 2/3 (refinement/objectives.rs) → deleted with the old search; follow-up issue filed before the deletion
• ClusterObjective::ParetoFront, Selection.front → deleted
• request fees in the objective → added after selection

Dead in production → deleted

• ResourceBudget, DagOfPhysicalCandidatesWithCosts.resource_budgets (candidates/dag.rs) with validation (:97-125), MILP rows (model.rs), projection copy (direct_price_options.rs), gate (components.rs)
• #[cfg(test)] modules candidates/{configurations,factoring,inventory,physical,reconstruction,scopes}.rs (~3.1k lines), refinement/pruning.rs → asap-physical-asap-optimization
• 25 CandidateConnectionReport fields (seed_*, direct_cost_*, joint_price_solve_calls, certified_query_cuts, early_query_pruning_*, validation_rejection_*, catalog_scope, max_refinements, time_limit_ms, …)
• LowerBoundMetrics fields except rounded_down_candidates (lower_bounds.rs); QueryScheduleRates.queries (general_schedule_rates.rs); structural_baseline, baseline_weights (search.rs); remaining_seconds: Option<f64> (model.rs, always Some)
• merged-round-N- renaming (route.rs); resize_for_failure_probability (sizing.rs, [Refactor] Crate organization by stage: per-crate functionality, cross-stage code into asap-accuracy / asap-cost / asap-types, enforced crate boundaries, public vs private #1054 X2)
• pub PlanSelector and methods, PassRegistry, PassNameConflict, pub LARGE_PEAK_MEMORY_BYTES, with_confirmed_weights re-export → private or deleted

Duplicated → one copy

• selected-plan cost (MILP objective, DirectCostProfile::selected_cost direct_cost.rs, evaluator) → plan_cost = objective
• modeled_cost re-summed from local_cost (model.rs, direct_price_options.rs, warm_start.rs, search.rs, objectives.rs, components.rs) → the objective
• reader-set rate (general_schedule_rates.rs) and per-query rates (direct_profile.rs, general_schedule_rates.rs, proposals.rs, lower_bounds.rs) → asap_cost::execution_demand
• fixed vs joint cost profile (direct_profile.rs / supplier_price_options.rs); direct_cost::encode / general_schedule::encode → cost_terms.rs
• AND linearization (model.rs, direct_cost.rs, general_schedule.rs, general_schedule_rates.rs) → one helper
• reach on a selected graph (types/.../graph.rs, warm_start.rs, direct_cost.rs); possible reach (model.rs / direct_cost.rs) → one each
• observation cuts (model.rs, direct_price_options.rs); "remove candidates" surgery (search.rs, planner.rs, materialization.rs); "one copy per ActivationRequires" (search.rs, planner.rs) → deleted
• relative gap formula (search.rs, planner.rs, components.rs, route.rs) → SearchReport::relative_gap; tolerances 1e-9/1e-8/1e-12/1e-7 → one
• deadline checks (planner.rs, objectives.rs, components.rs, materialization.rs, route.rs) → one, in select.rs; throw-away PlanSelector::new (planner.rs, components.rs, route.rs) → one validation
• reader-constraint validation (dag.rs / Stage 2 catalog/mod.rs) → Stage 2's
• DagOfPhysicalCandidatesWithCosts::unpriced graph copy (adds one f64) → read &PhysicalASAPMemoCandidates
• Kahn order (direct_cost.rs / supplier_price_options.rs), four template-node view builders, lookback (direct_profile.rs / proposals.rs), HLL certificate (direct_profile.rs / supplier_price_options.rs) → one each

Over-abstracted → simpler

• #[path] modules (model.rs, direct_cost.rs, general_schedule.rs, direct_profile.rs) → plain mod
• solve_with_objective(…, direct_profile: Option, projection: Option) → one model builder; PriceOptionProjection (direct_price_options.rs) → deleted
• forwarding chain planner.rs:291 → search.rs:182 → :205 → :224, single-use propose closure → one function; forwarders solve_*_with_metrics, extract/extract_latest_panes, materialize_configured_candidate, validated_seed_after_stop, five search_candidates* → inlined or deleted
• RouteOutcome [x], PhysicalCandidatesWithCosts { physical, dag } → &PhysicalASAPMemoCandidates; InputSupplierCandidates [x], QueryOutputCandidates → plain vectors
• string calibration_scope, report status → enums; [x] pub use search as pass → one path
• bug: objectives.rs reports calibration_scope = "provided_frozen" without a frozen factor → fixed or deleted with the file

Renames ("cost" not "price", "producer" not "supplier")

• supplier_candidates_per_input, SupplierCandidates → producer_candidates_per_input, ProducerCandidates
• refinement/ → removed, validation in evaluate.rs; local_cost → NodeUsage; refinements, max_refinements, candidates_checked → deleted; lower_bound, actual_cost → cost_lower_bound, validated_cost_per_second
• (direct_cost.rs, general_schedule_rates.rs, pricing.rs, route.rs: above)

Module layout of asap-plan-selection

• search/{candidates,costing,milp,refinement}/ →

  lib.rs          select_plan, Selection, SearchReport (pub); nothing else pub
  select.rs       driver, warm start, deadline
  groups.rs       execution contexts and their memo groups, candidate dominance, accuracy weights
  milp/model.rs   variables and structural rows (coverage, validity, sharing, reach, reader constraints, latency, accuracy)
  milp/cost_terms.rs  cost model 1 objective from asap-cost numbers
  milp/decode.rs  solution → DagOfSelectedCandidates
  evaluate.rs     one realizer call + PlanSelector (pub(crate)) + cost equality check
  checks/         accuracy, accuracy binding, latency, retention
  report.rs
  

• pricing.rs → asap-cost; route.rs round loop, pipeline.rs, query_pruning.rs, search/mod.rs pass API → facade; sizing.rs → out of Stage 3 ([Refactor] Crate organization by stage: per-crate functionality, cross-stage code into asap-accuracy / asap-cost / asap-types, enforced crate boundaries, public vs private #1054 X2, now asap-accuracy)
• whole-workload select_plan (reference_pricer) → devtools reference pricer

Rollout

• new search behind an internal Stage 3 setting → default once the gates pass; replaced code deleted in the same series; no runtime fallback

Scope and acceptance

Stage 3 only; the Stage 2 → 3 input and the Stage 3 output types (§1) are fixed. Users read only Stages 0–2 for now, so Stage 3's output and behavior may change freely: tests asserting dropped outputs (whole-plan rejection causes, the Pareto front) are rewritten against pruned_candidates or deleted, and the DAG viewer's Stage 3 panel shows only what Selection has.

• Memory (Planner aborts with bad_alloc under 8 GB on random-mixed-seed-862 workload-01 when raw data is retained #1036): seed-862/workload-01 with raw_data_retained: true plans under ulimit -v 8388608.
• Missed plans (Stage 3 keeps a MatAtQuery build whose NotMaterialized sibling avoids a second scan (#923 workload) #1142, Stage 3 misses cheaper plans in Stage 2's output: rangeheavy t5-q09, or-query and three-quantile workloads #1149): on sharing_issue_fixtures/923.json (q3/q4 of the example), the Sum and Count builds share one TimeRange(Scan) run and sharing_issue_regressions::exact_coarse_sum_reuses_fine_sum passes; on the Stage 3 misses cheaper plans in Stage 2's output: rangeheavy t5-q09, or-query and three-quantile workloads #1149 workloads (the or of two quantiles, rangeheavy-t5-q09-15m-1h, the three-quantile workload), Stage 3 selects the cheaper plan in Stage 2's output (e.g. the KLL pane chain for q0.5).
• Time: the memo examples and the paper workloads (W-RangeHeavy, W-RollupHeavy, W-Mixed, 8-user) return a validated plan within 300 s; report MILP size and solves.
• The realizer never rejects: before switching the default, count realizer rejections of decoded plans on the corpus snapshots, memo examples and paper workloads; each reason becomes a Stage 2 catalog rule; the gate is zero.
• Exactness, three ways: compare with costing every candidate DAG the memo stands for (Memo: convert memo-group candidates into the vector of all candidate DAGs #1002, stage_pipeline: price every enumerated candidate DAG and flag any cheaper than the selected plan #1004) and with --exhaustive; run the equivalence seeds, Two windowed quantiles over one metric: shared scan costs more, and nested-window KLL panes are never shared #839, Instant quantile by (host, job) next to windowed quantiles: no scan sharing, no panes, no ingest-time variant, 100-group default #840 and q24h; list per workload the cost changes from the design's semantics (one candidate per group per context with its lost plans, working memory held for the slowest reader, request fees outside the objective).
• Unit tests:
  • a NotMaterialized query-time candidate selected in two contexts decodes to two nodes; a shareable candidate read by two queries to one; a group read by two readers in one context to one node, costed once;
  • a query reading a group through an ingestion-time sketch and directly at query time may use different candidates in the two contexts;
  • two configurations of one summary state are both selected when two queries are cheaper reading different ones, and one ingestion scan read by both is costed once when they have the same refresh window, and once per refresh window otherwise;
  • a locally more expensive choice that enables sharing is selected when cheaper overall;
  • a NotMaterialized input of a MatAtQuery candidate read by two queries in one schedule group runs once, in the candidate's context, costed as plan_cost costs it;
  • each cost term of §3.5 equals plan_cost on hand-built plans, including held memory with readers of different latency, raw retention with two windows, a scan read through two offsets, and a kept pane chain with readers of different spans;
  • a plan over the latency bound is infeasible; A1 (row 8) ⊆ satisfies on the corpus and memo examples, each difference from a listed conservative exclusion;
  • cost models 2 and 3 return UnsupportedCostModel.
• Rollout: behind an internal Stage 3 setting until the gates pass, then default; replaced code deleted in the same series; no runtime fallback.

Non-goals

• Changing Stage 1 or Stage 2 beyond emitting no scope copies and adding accuracy_budget_parts.
• Changing accuracy semantics, or cost semantics beyond §3.5: working memory held for the slowest reader, request fees outside the objective; the asap_cost::execution_demand schedule policy ("different intervals add") is unchanged.
• Tuning HiGHS or the model builder.

Related

#1036 (the failure); #888 (MILP over the memo); #879, #927 (whole workloads in minutes); #1142, #1149 (missed plans); #1015 (Stage 3 review); #1047, #1054 (stage boundaries, crates); #1048 (pane-width rounds); #1144 (prototype); #1002, #1004 (enumeration); #951 (shared scans); #929 (sizes once); #1046 (dead code).

🤖 Generated with Claude Code