Select complete workload candidates with exhaustive sharing and lifecycles

[zzylol]

Problem: selection picks per root, but #509 asks for the cheapest complete plan for the whole workload

#509 §Stages and their decisions says a candidate is a DAG "for the whole workload, not for one query", and that stage 3 is "the only step that chooses one candidate DAG". #509 §3 Plan selection says:

Cost is evaluated for the whole workload rather than per query, which is what lets one shared summary beat several cheaper independent ones: the cost of a shared summary is estimated once, with the demand of all its consumers.

#509 §Pass 2 adds: "Applying a rule adds a shared candidate and keeps the independent candidates, so selection can compare both."

The shipped MajorPass does not do this. It ranks choices locally, assembles each root, interns identical producers with CSE, and then reconciles shared states. replacement.rs says so itself: "This is not an exhaustive search over combinations of choices for a provably-global optimum". Two cases it cannot get right:

Example 1: locally expensive, globally cheaper

The fixture in crates/integration-tests/tests/complete_workload_selection.rs has two queries over one table:

SELECT SUM(value) FROM events
SELECT MAX(value) FROM events

Its cost model (InteractingCosts) quotes:

Choice	Per-root quote	Complete workload quote
raw recompute	1.0	2.0 (any assignment with a raw root)
summary	build 10.0	0.25 (only when both roots use summaries)

Compared root by root, raw (1.0) beats a summary build (10.0) for each query. The cheapest complete workload is "both summaries" at 0.25. A per-root comparison never sees that number.

Example 2: share some consumers, not all

Three identical SUM(value) queries read one producer. The possible sharing layouts are the set partitions of the three readers:

{0,1,2}      all share one state
{0,1} {2}    0 and 1 share, 2 is independent
{0,2} {1}
{0} {1,2}
{0} {1} {2}  all independent

CSE interns all three into one state ({0,1,2}), and MajorPass falls back to unshared roots only when the union cannot be costed. The three partial layouts are never generated, so they cannot win even when they are cheapest. planner-layering-status.md listed both gaps before this PR: "It does not enumerate all partial sharing partitions", and the cheapest complete candidate guarantee "need[s] a complete workload search/selection path".

Scope

This PR covers the #509 §3 requirement "choose the cheapest valid plan for the whole workload" over the candidates the planner can already generate: registered Pass 1 alternatives, exact pane compositions from #566 (window-composition rule), partial sharing of identical producers (Pass 2 identical-expression rule), and the lifecycle (materialization) alternatives of each state. It leaves out: resizing a shared summary to its strictest consumer (#509 §Pass 2), parallelism, partitioning and resources (TODO in #509 §2), new accuracy certificates, and asserting the illustrative 54/156 inventory counts from #509's examples.

Proposed method

A new pass, CompletePass (name complete), runs as stage 3 plan selection. It enumerates complete workload assignments, prices each one with a single workload-level cost hook, and returns the cheapest. It is opt-in. major stays the default because it accepts rank-only cost models, and a rank cannot certify a complete cost.

CompletePass::enumerate does five nested steps. Every step is bounded by max_candidates.

1. Logical alternatives (Pass 1). Build the search space with search_workload_with_targets and the default strategies, then list every workload DAG with CandidateLogicalASAPDAGs::enumerate_candidate_dags. Its rejected_assemblies seed the inventory's rejection list.
2. Pane compositions (Pass 2, window composition). For each root, enumerate_window_compositions (Generate exact window composition and materialization candidates #566) returns the original state plus each exact pane composition. The pass takes the Cartesian product across roots.
3. Sharing partitions (Pass 2, CSE). sharing_partitions first interns the roots with share_common_sub_dags. Every node reached by more than one reader forms a class. For each class it enumerates every set partition of its readers. It then rebuilds each reader's DAG: readers in the same group get the same Rc, others get their own copy.
4. Union demand per state. For each assembly, every reachable node is mapped to exactly the entry indices that reach it. enumerate_assembled_plans re-enumerates each deployment's lifecycle alternatives against that union, so a shared state is costed with all of its consumers and an unshared one with only its own.
5. Lifecycle assignments (physical materialization). SummaryMaintenanceLifecycleCandidates::enumerate_plans returns every legal, fully costed lifecycle combination for a root, instead of only the cheapest. The pass takes the product across roots.

Each full assignment is then checked and priced:

• compatible rejects it if one shared Rc has different lifecycle guarantees or window frameworks in different plans. Every plan must also pass execution_timed_dag().
• CostModel::complete_workload_candidate_cost quotes the whole assignment. None or a non-finite/negative value is recorded as a rejection reason, never treated as zero.
• Priced assignments become PlanOutputs with workload_total_cost set.

optimize returns the candidate with the smallest workload_total_cost, or an error listing all rejection reasons. Exceeding any budget (logical, pane, partition, assembly, lifecycle, or assignment) is an error even if a priced candidate was already found. There is no heuristic fallback.

The default cost hook is additive. It sums each root's selected total cost and subtracts a deployment's cost each time the same Rc appears again, so a shared producer is charged once. Deployments whose roots interact (as in Example 1) override the hook. Workloads with scalar roots need an override too: the default returns None because the per-root hooks do not price scalar execution.

Key code interfaces

crates/asap-aware-mapping/src/pass/complete.rs (new):

#[derive(Debug, Clone, Copy)]
pub struct CompletePass {
    pub max_candidates: usize,
}
impl Default for CompletePass { /* max_candidates: 65_536 */ }

#[derive(Debug)]
pub struct CompleteWorkloadInventory {
    pub candidates: Vec<PlanOutput>,
    pub rejected: Vec<String>,
}

impl CompletePass {
    pub fn enumerate(
        &self,
        input: OptimizationInput<'_>,
    ) -> Result<CompleteWorkloadInventory, OptimizeError>;
}

impl OptimizationPass for CompletePass {
    fn name(&self) -> &'static str { "complete" }
    fn optimize(&self, input: OptimizationInput<'_>) -> Result<PlanOutput, OptimizeError>;
}

crates/asap-aware-mapping/src/cost_model.rs, new CostModel method with an additive default:

fn complete_workload_candidate_cost(
    &self,
    plans: &[SummaryMaintenanceLifecyclePlan],
    scalar_roots: &[(usize, asap_types::ir::ScalarExpr)],
) -> Option<Cost>;

crates/asap-aware-mapping/src/pass/mod.rs:

pub struct PlanOutput {
    pub plans: Vec<QueryLifecyclePlan>,
    pub scalar_roots: Vec<(usize, asap_types::ir::ScalarExpr)>,
    pub workload_total_cost: Option<crate::cost_model::Cost>, // new
}

pub enum OptimizeError {
    #[error("complete workload selection: {0}")]
    CompleteSelection(String), // new
    // ...
}

// PassRegistry::with_builtin() now registers MajorPass ("major") and CompletePass::default() ("complete").

crates/asap-aware-mapping/src/summary_maintenance_lifecycle.rs:

#[derive(Clone)] // new
pub struct SummaryMaintenanceLifecycleCandidates<'a> { /* unchanged */ }

impl SummaryMaintenanceLifecycleCandidates<'_> {
    pub fn enumerate_plans(
        &self,
        limit: usize,
    ) -> Result<Vec<SummaryMaintenanceLifecyclePlan>, SummaryMaintenanceLifecycleChoiceError>;
}

pub enum SummaryMaintenanceLifecycleChoiceError {
    #[error("lifecycle enumeration exceeds candidate budget; no partial inventory returned")]
    BudgetExceeded, // new
    // ...
}

Usage:

// Through the planner facade (crates/planner re-exports CompletePass and CompleteWorkloadInventory):
let input = UserInput::new(&workload, frontend, models, lifecycle)
    .with_pass(&CompletePass::default());

// Or directly on a ParsedWorkload:
let output = optimize(&CompletePass::default(), input)?;
let cost = output.workload_total_cost; // Some(..) for this pass

Crate-private helpers: enumerate_assembled_plans (new), enumerate_with_profile (now pub(crate)), sharing_partitions, compatible, product.

Fields

CompletePass

Field	Type	Meaning
max_candidates	usize	Budget for each exhaustive stage: logical DAGs, pane compositions per root, sharing partitions, assemblies, lifecycle plans per root, and total priced assignments. Exceeding it returns OptimizeError::CompleteSelection with no partial inventory. 0 is rejected ("requires a positive budget"). Default 65_536. Set by the caller.

CompletePass::enumerate

Item	Meaning
input: OptimizationInput<'_>	The same input every pass takes: parsed workload, PlanningModels (cost, accuracy, evidence, capabilities) and LifecycleInput (now_ms, horizon). Validated first.
returns CompleteWorkloadInventory	Every priced complete assignment, plus reasons for the rest.

CompleteWorkloadInventory

Field	Type	Meaning
candidates	Vec<PlanOutput>	One per priced, valid assignment. Each has one QueryLifecyclePlan per workload entry, scalar_roots copied from the workload, and workload_total_cost = Some(..).
rejected	Vec<String>	Sorted, deduplicated reasons: logical rejected_assemblies, lifecycle enumeration errors, "shared state has inconsistent lifecycle/window assignment", "complete workload cost is unknown", "complete workload cost is invalid".

OptimizationPass for CompletePass

Item	Meaning
name()	"complete"; the key in PassRegistry.
optimize(input)	Runs enumerate and returns the candidate with the minimum workload_total_cost (f64::total_cmp). No candidate → CompleteSelection("no feasible priced complete workload: <reasons>").

CostModel::complete_workload_candidate_cost

Parameter / result	Meaning
plans	One SummaryMaintenanceLifecyclePlan per workload root, in the assembly's root order. A shared state is the same Rc in each plan that reads it.
scalar_roots	The workload's exact scalar expressions, keyed by entry index.
returns Option<Cost>	The complete quote. None means unknown; the pass records it as a rejection.

The default reads these existing plan fields: selected_raw_recompute picks raw_recompute_total_cost or summary_total_cost as the root's cost; deployments[].summary (by Rc::as_ptr) detects repeats; for a repeat, summary_maintenance_lifecycle_guarantee names the chosen lifecycle and the matching entry in alternatives supplies the total_cost to subtract. It returns None if scalar_roots is non-empty, if any needed cost or guarantee is missing, or if a cost is non-finite or negative.

PlanOutput::workload_total_cost: Option<Cost>. Set by a complete-cost pass to the certified quote of the selected workload. MajorPass leaves it None. check_contract rejects a non-finite or negative value with ContractViolation("invalid complete workload cost").

OptimizeError::CompleteSelection(String): every CompletePass failure: budget exceeded, zero budget, no feasible priced workload, logical or pane enumeration errors.

SummaryMaintenanceLifecycleCandidates::enumerate_plans

Item	Meaning
limit: usize	Maximum number of lifecycle combinations. 0, overflow, or a product above limit → BudgetExceeded.
returns Vec<SummaryMaintenanceLifecyclePlan>	For each deployment, every eligible alternative is tried in every combination through the existing select. Combinations that fail with NoCompleteEstimate or IncompatibleEvaluationSchedules are skipped. Any other error is returned.

SummaryMaintenanceLifecycleChoiceError::BudgetExceeded: new variant for the case above. CompletePass turns any lifecycle error containing "budget" into a hard error; other lifecycle errors become rejection reasons.

#[derive(Clone)] on SummaryMaintenanceLifecycleCandidates: needed because select consumes self and enumerate_plans calls it once per combination.

Examples

End to end: Example 1 (globally_cheapest_assignment_executes_both_queries)

Input: the two-query fixture (SUM, MAX over events), InteractingCosts, LifecycleInput::new(0), CompletePass { max_candidates: 4096 }.

1. Logical inventory: each root has a raw alternative and summary alternatives.
2. No recurrence, so no pane compositions.
3. Only nodes both roots reach (such as the events scan) form sharing classes, and their partitions are enumerated.
4. Lifecycle assignments are enumerated per root.
5. Each assignment is quoted: 0.25 if no root selected raw recompute, else 2.0.

Output: the inventory contains an all-raw candidate, but optimize returns the all-summary one with workload_total_cost == Some(Cost(0.25)). The test then executes both selected roots on rows [1.0, 2.0] and gets 3.0 (SUM) and 2.0 (MAX).

Partial sharing (three_consumers_can_share_only_a_subset)

Three identical SUM(value) entries. PartialSharingCosts quotes 0.1 only when reader 0 and reader 1 share a state and reader 2 does not; otherwise 10.0, and raw is 1000.0. The result has workload_total_cost == Some(Cost(0.1)), states[0] and states[1] are the same Rc, and states[2] is a different one. This is the {0,1} {2} layout from Example 2.

Shared producer charged once (shared_producer_cost_is_not_multiplied_by_consumer_count)

Two identical SUM(value) entries, data at rest, Horizon(100.0), AdditiveCosts (default hook, build cost 10.0, other lifecycle costs zero). The selected plans share one Rc with the same lifecycle guarantee, the lifecycle is SummaryMaintenanceLifecycle::Shared { .. }, and workload_total_cost == Some(Cost(10.0)): one build, not two.

Accepted and rejected cases

Case	Result
Valid assignment, hook returns finite non-negative cost	candidate
Shared Rc with different lifecycle guarantee or window framework across plans	rejected: "shared state has inconsistent lifecycle/window assignment"
A plan's execution_timed_dag() fails	rejected: same reason
Hook returns None (e.g. default hook with missing costs or scalar roots)	rejected: "complete workload cost is unknown"
Hook returns NaN, infinite or negative	rejected: "complete workload cost is invalid"
Lifecycle enumeration fails for a root, not budget-related	assembly skipped, reason recorded
Every assignment rejected	Err: "no feasible priced complete workload: …" (test uses PlanningModels::builtin())
Any stage exceeds max_candidates	Err containing "budget", no inventory (test uses max_candidates: 1)
max_candidates == 0	Err: "complete search requires a positive budget"

Out of scope

• Resizing a shared state's parameters to its strictest consumer. Only identical admitted producers are partitioned.
• New sketch accuracy certificates, historical approximate EH implementations, parallelism/partitioning/resource search.
• Physical placement choices beyond what a deployment's complete_workload_candidate_cost override prices. Native compile_materialization_candidates supplies the executable frontier for such models.
• The guarantee holds only over registered alternatives with complete cost evidence. The illustrative 54/156 counts from docs: propose workload-wide planning, summary sharing, and materialization #509 are not asserted.

Stack and validation

Stack: #566 → #568 → #569. Base: stack/509-21-window-composition (#566). Next: #569, which adds the temporal merge timestamp regression and fix. Addresses the complete workload selection path in #509 (§3 Plan selection). docs/develop_docs/planner-layering-status.md updates the sharing and whole-workload rows and adds a "Complete workload selection acceptance" section.

Validation: 1,637 workspace tests/doctests pass (2 existing ignores). The four tests in crates/integration-tests/tests/complete_workload_selection.rs execute a globally winning combination that local ranking misses, select a partial sharing partition, verify one retained producer is charged once, and reject unknown costs and budget exhaustion. All-feature workspace Clippy, formatting and diff checks pass.

🤖 Generated with Claude Code

[zzylol]

Parked as draft: PR priorities changed (see #528). Order is now (A) finish #511 operator sharing, (B) the #572 crate/module reorganization, (C) #509 end-to-end stages. This PR sits on the old stack/528-legacy-physical-base chain, and Phase B moves the files it touches. Its content will be re-scoped onto the new layout in Phase C.

🤖 Generated with Claude Code