From 9786766995a2d7059ce4b9bc19936d9fb5a5a484 Mon Sep 17 00:00:00 2001 From: zzylol Date: Wed, 30 Sep 2026 13:40:14 +0000 Subject: [PATCH 1/2] feat: choose placement per state with bounded-lag mixing A query that alone consumes its states may retain some and rebuild the others from raw series at query time, as #803 executes: raw inputs at t_q, each stored native batch at its newest complete window within the lag bound. Placement enumerates per query (at most 8 units) the assignments that retain at least one native branch (an ungrouped Sum or sketch over retained readouts, with its sources) and rebuild every other state. Each unit is priced with #804's model as it would run: retained as complete windows sliding at the evaluation interval, or rebuilt folding only its own selector's samples. The cheapest mix replaces the group decision only when priced and strictly cheaper; otherwise the group decision stands. The compiled plan types the mixed root with the series identity, installs each stored branch with a precompute program building its batch, and emits one physical program over raw Scans and ReadMaterialization inputs. Co-Authored-By: Claude Opus 5.5 --- control_plane/src/physical/compiler.rs | 150 ++++- .../src/physical/compiler/placement.rs | 606 +++++++++++++++++- control_plane/tests/lifecycle_placement.rs | 100 +++ .../support/lifecycle_placement_process.rs | 105 ++- 4 files changed, 925 insertions(+), 36 deletions(-) diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index fc5dc4365..b04e57d7f 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -1627,6 +1627,11 @@ impl DeploymentPlanCompiler { request.queries[id].selected_plan_root = root; } let placement = placement::place(&request, &environment, frontend); + for (index, query) in request.queries.iter_mut().enumerate() { + if let Some(root) = placement.root(index) { + query.selected_plan_root = Rc::clone(root); + } + } let population_operators = super::maintained_population::operators(&request)?; let mut compiled_materializations = Vec::with_capacity(request.queries.len()); let mut collector_materializations = Vec::with_capacity(request.queries.len()); @@ -1646,6 +1651,7 @@ impl DeploymentPlanCompiler { &request.queries, environment.target, request.allow_mixed_summary_and_exact_execution, + |index| placement.native_branches(index), )?; let mut lifecycle_estimates = BTreeMap::::new(); @@ -1668,9 +1674,10 @@ impl DeploymentPlanCompiler { let selected = if super::maintained_population::supported_node(&node) { Vec::new() } else { - collect_selected_materializations( + collect_selected_materializations_with( &node, request.allow_mixed_summary_and_exact_execution, + placement.native_branches(query_index), ) .map_err(|reason| CompileError::Query { query_id: query.query_id.clone(), @@ -1780,7 +1787,13 @@ impl DeploymentPlanCompiler { .retained_physical()? .is_some_and(|candidate| candidate.precompute.is_some()); for (ordinal, selected) in selected.into_iter().enumerate() { - let branch_query = state_query(query, &selected, &cohort_nodes, native_cohort); + let branch_query = state_query( + query, + &selected, + &cohort_nodes, + native_cohort, + placement.beside_raw(query_index, &selected.node), + ); let query = &branch_query; let lifecycle_costs = SummaryMaintenanceLifecycleCostInputs { build_cost: Some(Cost(query.summary_lifecycle_inputs.costs.build)), @@ -1993,6 +2006,7 @@ impl DeploymentPlanCompiler { reason: error.to_string(), })?; if !native_cohort + && !placement.beside_raw(query_index, &selected.node) && runtime_materialization.window_size != runtime_materialization.slide_interval { @@ -2352,11 +2366,24 @@ impl DeploymentPlanCompiler { None }; let mut entry = if let Some(raw) = placement.raw_program(query_index) { + let stored = raw + .stored + .iter() + .map(|(slot, node)| { + let SummaryExpr::SummaryAgg { family, .. } = &node.expr else { + return Err(CompileError::Snapshot( + "stored input beside raw series is not a summary state".into(), + )); + }; + Ok((*slot, binding(node, family)?)) + }) + .collect::, CompileError>>()?; Ok(raw_query_time_entry( query, canonical.clone(), &original_root(query, query_index, &request.canonical_roots)?, raw, + stored, )?) } else if let Some((source, program)) = native_rate { let native_state_binding = if let SummaryExpr::SummaryAgg { @@ -2592,6 +2619,20 @@ impl DeploymentPlanCompiler { query_id: query_id.clone(), reason, })?; + for (branch, program) in placement.branch_programs(query_index) { + let sink = compiled.node_ids.node_id(branch).ok_or_else(|| { + CompileError::Snapshot("stored branch is absent from its DAG".into()) + })?; + installed.native_programs.insert( + sink, + serde_json::from_slice( + &program + .encode() + .map_err(|e| CompileError::Snapshot(e.to_string()))?, + ) + .map_err(|e| CompileError::Snapshot(e.to_string()))?, + ); + } if let Some(program) = request.queries[query_index] .retained_physical()? .and_then(|physical| physical.precompute) @@ -2857,23 +2898,36 @@ fn raw_query_time_entry( canonical: String, root_expr: &QueryExpr, raw: &placement::RawQueryTimeProgram, + stored: Vec<(u64, MaterializationBinding)>, ) -> Result { let mut nodes = BTreeMap::new(); let mut inputs = Vec::new(); let mut source_nodes = Vec::new(); - for (ordinal, (slot, scan)) in raw.scans.iter().enumerate() { + let leaves = raw + .scans + .iter() + .map(|(slot, scan)| { + ( + *slot, + crate::query_plan::QueryPlanNode::Logical { + operator: scan.clone(), + inputs: vec![], + }, + ) + }) + .chain(stored.into_iter().map(|(slot, binding)| { + ( + slot, + crate::query_plan::QueryPlanNode::ReadMaterialization { binding }, + ) + })); + for (ordinal, (slot, node)) in leaves.enumerate() { let id = crate::query_plan::QueryNodeId(ordinal as u64); - nodes.insert( - id, - crate::query_plan::QueryPlanNode::Logical { - operator: scan.clone(), - inputs: vec![], - }, - ); + nodes.insert(id, node); inputs.push(id); - source_nodes.push(*slot); + source_nodes.push(slot); } - let root = crate::query_plan::QueryNodeId(raw.scans.len() as u64); + let root = crate::query_plan::QueryNodeId(inputs.len() as u64); nodes.insert( root, crate::query_plan::QueryPlanNode::Physical { @@ -3858,11 +3912,15 @@ struct PlannerPhysicalSelection { /// `query` as the consumer of `state` alone: its window, grouping, and the /// window implementations that can install it. +/// A state read beside query-time raw inputs, like a native cohort state, is +/// installed as one complete window sliding at the evaluation interval: the +/// executor reads it as a stored native batch within one slide of t_q. fn state_query( query: &QueryCompilationInput, state: &SelectedMaterialization, cohort_nodes: &BTreeSet, native_cohort: bool, + beside_raw: bool, ) -> QueryCompilationInput { let mut branch = query.clone(); branch.query_lookback_ms = state @@ -3877,14 +3935,12 @@ fn state_query( let cohort = cohort_nodes.contains(&(Rc::as_ptr(&state.node) as usize)); branch.window_realization_candidates.retain(|candidate| { candidate.window_secs.saturating_mul(1_000) == lookback_ms - && if cohort { - if native_cohort { - windows::is_complete_window(candidate) - && candidate.slide_secs.saturating_mul(1000) - == u64::from(query.summary_lifecycle_inputs.evaluation_interval_ms) - } else { - windows::is_full_cohort(candidate) - } + && if beside_raw || (cohort && native_cohort) { + windows::is_complete_window(candidate) + && candidate.slide_secs.saturating_mul(1000) + == u64::from(query.summary_lifecycle_inputs.evaluation_interval_ms) + } else if cohort { + windows::is_full_cohort(candidate) } else { !candidate.cohort_only } @@ -4444,10 +4500,11 @@ pub(crate) fn aggregation_config_for_materialization( .context("build materialization from physical aggregation") } -fn materialization_consumers( +fn materialization_consumers<'a>( queries: &[QueryCompilationInput], target: PhysicalDeploymentTarget, composable: bool, + native_branches: impl Fn(usize) -> &'a [Rc], ) -> Result>, CompileError> { let mut consumers = BTreeMap::<_, BTreeSet<_>>::new(); // Until logical lifecycle costing carries a derived-program identity, @@ -4455,11 +4512,15 @@ fn materialization_consumers( let mut cohort_programs = BTreeMap::>::new(); for (index, query) in queries.iter().enumerate() { - let states = collect_selected_materializations(&query.selected_plan_root, composable) - .map_err(|reason| CompileError::Query { - query_id: query.query_id.clone(), - reason, - })?; + let states = collect_selected_materializations_with( + &query.selected_plan_root, + composable, + native_branches(index), + ) + .map_err(|reason| CompileError::Query { + query_id: query.query_id.clone(), + reason, + })?; for state in states { if composable && state.window_secs.is_some_and(|window| { @@ -4555,14 +4616,29 @@ fn collect_selected_materializations( node: &Rc, composable: bool, ) -> Result, String> { + collect_selected_materializations_with(node, composable, &[]) +} + +/// [`collect_selected_materializations`] where each of `native_branches` is +/// maintained by its own complete native graph, as a whole root with a +/// fixed-window realization is: its Sum over readouts is a stored state. +fn collect_selected_materializations_with( + node: &Rc, + composable: bool, + native_branches: &[Rc], +) -> Result, String> { + #[allow(clippy::too_many_arguments)] fn walk( node: &Rc, readout: Option<&SketchQuery>, composable: bool, native_maintenance: bool, + branches: &[Rc], inherited_grouping: Option>, selected: &mut Vec, ) -> Result<(), String> { + let native_maintenance = + native_maintenance || branches.iter().any(|branch| Rc::ptr_eq(branch, node)); let grouping = if composable { if let SummaryExpr::SummaryAgg { reduction, child, .. @@ -4612,7 +4688,15 @@ fn collect_selected_materializations( }; if let Some(source) = &immutable_sources { for source in source { - walk(source, None, composable, native_maintenance, None, selected)?; + walk( + source, + None, + composable, + native_maintenance, + branches, + None, + selected, + )?; } } match &node.expr { @@ -4628,6 +4712,7 @@ fn collect_selected_materializations( readout, composable, native_maintenance, + branches, grouping.clone(), selected, )?; @@ -4636,6 +4721,7 @@ fn collect_selected_materializations( readout, composable, native_maintenance, + branches, grouping.clone(), selected, )?; @@ -4646,6 +4732,7 @@ fn collect_selected_materializations( readout, composable, native_maintenance, + branches, grouping.clone(), selected, )?; @@ -4656,6 +4743,7 @@ fn collect_selected_materializations( readout, composable, native_maintenance, + branches, grouping.clone(), selected, )?; @@ -4664,6 +4752,7 @@ fn collect_selected_materializations( readout, composable, native_maintenance, + branches, grouping.clone(), selected, )?; @@ -4676,6 +4765,7 @@ fn collect_selected_materializations( readout, composable, native_maintenance, + branches, grouping.clone(), selected, )?; @@ -4684,6 +4774,7 @@ fn collect_selected_materializations( readout, composable, native_maintenance, + branches, grouping.clone(), selected, )?; @@ -4704,6 +4795,7 @@ fn collect_selected_materializations( readout, composable, native_maintenance, + branches, grouping.clone(), selected, )?; @@ -4724,6 +4816,7 @@ fn collect_selected_materializations( Some(query), composable, native_maintenance, + branches, grouping.clone(), selected, )?, @@ -4734,6 +4827,7 @@ fn collect_selected_materializations( readout, composable, native_maintenance, + branches, grouping.clone(), selected, )?; @@ -4842,6 +4936,7 @@ fn collect_selected_materializations( None, composable, placement::root_fixed_window_candidate(node).is_ok(), + native_branches, None, &mut selected, )?; @@ -6060,6 +6155,7 @@ pub(crate) mod tests { &queries, PhysicalDeploymentTarget::BackendLocalRemoteWrite, true, + |_| &[], ) .unwrap_err(); assert!( diff --git a/control_plane/src/physical/compiler/placement.rs b/control_plane/src/physical/compiler/placement.rs index 156786e3a..526aba900 100644 --- a/control_plane/src/physical/compiler/placement.rs +++ b/control_plane/src/physical/compiler/placement.rs @@ -19,16 +19,28 @@ use crate::query_plan::query_time::{LabelMatch, LabelMatcher, QueryTimeOperator} /// A query whose every state is `Ephemeral` keeps nothing between queries. /// Planner's PromQL Fallback lowering compiles it once over raw-series inputs, -/// each read at query time by the matching range-selector `Scan`. +/// each read at query time by the matching range-selector `Scan`. A mixed +/// query also reads its retained states, as stored native batches. pub(super) struct RawQueryTimeProgram { pub(super) program: CompiledPhysicalDag, pub(super) scans: Vec<(u64, QueryTimeOperator)>, + /// Retained states read beside the raw inputs. The executor binds each at + /// its newest complete window within its lag bound (#803). + pub(super) stored: Vec<(u64, Rc)>, } #[derive(Default)] pub(super) struct Placement { ephemeral: Vec>>, raw: Vec>, + /// Native branches a mixed query maintains and reads as stored batches. + native: Vec>>, + /// The precompute program building each native branch's batch. + branch_programs: Vec, CompiledPhysicalDag)>>, + /// Retained states of a mixed query, each installed as a complete window. + complete_window: Vec>>, + /// A mixed query's root, its raw selectors typed with the series identity. + roots: Vec>>, pub(super) trace: Vec, } @@ -42,6 +54,30 @@ impl Placement { pub(super) fn raw_program(&self, query: usize) -> Option<&RawQueryTimeProgram> { self.raw.get(query).and_then(Option::as_ref) } + + /// Whether retained `state` serves stored input beside `query`'s raw inputs. + pub(super) fn beside_raw(&self, query: usize, state: &Rc) -> bool { + self.complete_window + .get(query) + .is_some_and(|states| states.iter().any(|s| Rc::ptr_eq(s, state))) + } + + /// The root a mixed query compiles instead of Planner's selected root. + pub(super) fn root(&self, query: usize) -> Option<&Rc> { + self.roots.get(query).and_then(Option::as_ref) + } + + pub(super) fn native_branches(&self, query: usize) -> &[Rc] { + self.native.get(query).map_or(&[], Vec::as_slice) + } + + /// Each native branch of `query` with the precompute program building it. + pub(super) fn branch_programs( + &self, + query: usize, + ) -> &[(Rc, CompiledPhysicalDag)] { + self.branch_programs.get(query).map_or(&[], Vec::as_slice) + } } /// Backend lifecycle prices for any summary state. Retention charges the @@ -128,15 +164,16 @@ impl CostModel for LifecycleCosts { } /// The window implementation compilation installs for `state` when `query` -/// retains it, with the number of states that layout keeps in the store for -/// the state's own window. A derived state reading a longer window over it -/// retains more, so this is a lower bound there. +/// retains it, possibly `beside_raw` inputs, with the number of states that +/// layout keeps in the store for the state's own window. A derived state +/// reading a longer window over it retains more, so this is a lower bound there. fn installed_window( query: &QueryCompilationInput, state: &SelectedMaterialization, query_states: &[SelectedMaterialization], environment: &PhysicalDeploymentContext, retention_margin_ms: u64, + beside_raw: bool, ) -> Option<(WindowRealizationCandidate, u64)> { let native_cohort = query .retained_physical() @@ -147,6 +184,7 @@ fn installed_window( state, &super::windows::cohort_nodes(query_states), native_cohort, + beside_raw, ); let model = ControlPlaneCostModel::new(branch.accuracy_target.clone()) .with_window_implementation_costs( @@ -224,6 +262,10 @@ pub(super) fn place( let mut placement = Placement { ephemeral: vec![Vec::new(); queries.len()], raw: (0..queries.len()).map(|_| None).collect(), + native: vec![Vec::new(); queries.len()], + branch_programs: (0..queries.len()).map(|_| Vec::new()).collect(), + complete_window: vec![Vec::new(); queries.len()], + roots: vec![None; queries.len()], trace: Vec::new(), }; let (Some(workload), Some(data), Some(first)) = ( @@ -341,6 +383,7 @@ pub(super) fn place( &selected_states[query], environment, request.query_retention_margin_ms, + false, ) }); let layout = installed @@ -462,7 +505,7 @@ pub(super) fn place( .sum::>() .map(Cost) }; - let ephemeral: Vec = (0..states.len()) + let mut ephemeral: Vec = (0..states.len()) .map(|state| { rebuild_is_cheaper( total(linked[state], |decision| decision.2), @@ -470,6 +513,172 @@ pub(super) fn place( ) }) .collect(); + // A query that alone consumes its states may instead retain some and + // rebuild the others: raw inputs are read at t_q and each stored native + // batch at its newest complete window within the lag bound (#803). A + // retained unit is a native branch, read as its batch, with its sources; + // every other state is rebuilt. Each option is priced as it would run. + let mut mixed: Vec> = (0..queries.len()).map(|_| None).collect(); + for (query, owned) in query_states.iter().enumerate() { + let members: Vec = owned.iter().filter_map(index_of).collect(); + if !raw_bindable + || members.len() < 2 + || members.iter().any(|&member| states[member].1 != [query]) + { + continue; + } + let root = &queries[query].selected_plan_root; + let lifecycle = &queries[query].summary_lifecycle_inputs; + // Rebuilt beside retained state, a program folds only the samples of + // the rebuilt states' own selectors. + let alone = |member: usize| { + let ( + _, + QueryTimeOperator::Scan { + range_ms: Some(range_ms), + .. + }, + ) = leaf_selector(&states[member].0)? + else { + return None; + }; + let costs = LifecycleUnitCosts { + build: lifecycle.costs.build + + update_rate? * range_ms as f64 / 1_000.0 + * lifecycle.costs.maintenance_per_update, + ..lifecycle.costs.clone() + }; + price( + &states[member].0, + &[query], + SummaryMaintenanceLifecycle::Ephemeral, + &model(costs, None, lifecycle.evaluation_interval_ms), + ) + }; + let stored = |branch: &Rc, sources: &[usize]| { + let installed = + collect_selected_materializations_with(root, true, std::slice::from_ref(branch)) + .ok()?; + let unit: Vec<_> = std::iter::once(Rc::clone(branch)) + .chain(sources.iter().map(|&member| Rc::clone(&states[member].0))) + .collect(); + unit.iter() + .map(|state| { + let selected = installed + .iter() + .find(|selected| Rc::ptr_eq(&selected.node, state))?; + let (window, retained) = installed_window( + &queries[query], + selected, + &installed, + environment, + request.query_retention_margin_ms, + true, + )?; + // The executor reads the batch as one window of the query lookback. + if Rc::ptr_eq(state, branch) + && window.window_secs.saturating_mul(1_000) + != queries[query].query_lookback_ms + { + return None; + } + price( + state, + &[query], + SummaryMaintenanceLifecycle::ContinuouslyMaintained, + &model( + lifecycle.costs.clone(), + Some(retained), + lifecycle.evaluation_interval_ms, + ), + ) + .map(|cost| cost.0) + }) + .sum::>() + .map(Cost) + }; + // Each unit: an optional native branch and the member states it + // retains or rebuilds; a state outside every branch is rebuilt. + let mut units: Vec<(Option>, Vec)> = batch_branches(root) + .into_iter() + .filter_map(|branch| { + let sources = immutable_materialization_sources(&branch)? + .iter() + .map(|source| { + members + .iter() + .copied() + .find(|&member| Rc::ptr_eq(&states[member].0, source)) + }) + .collect::>>()?; + Some((Some(branch), sources)) + }) + .collect(); + for &member in &members { + if !units.iter().any(|(_, unit)| unit.contains(&member)) { + units.push((None, vec![member])); + } + } + let options: Vec = units + .iter() + .map(|(branch, unit)| MixedOption { + beside_raw: branch.as_ref().and_then(|branch| stored(branch, unit)), + alone: unit + .iter() + .map(|&member| alone(member).map(|cost| cost.0)) + .sum::>() + .map(Cost), + }) + .collect(); + let group = if ephemeral[members[0]] { + total(linked[members[0]], |decision| decision.3) + } else { + total(linked[members[0]], |decision| decision.2) + }; + let target = &queries[query].accuracy_target; + let mut event = json!({ + "stage": "deployment.mixed_placement", + "query_ids": [&queries[query].query_id], + "group_cost": group.map(|cost| cost.0), + "units": units.iter().zip(&options).map(|((branch, unit), option)| json!({ + "native_branch_id": branch.as_ref().map(|branch| crate::planner_selection::explained_root_id(branch, target)), + "logical_root_ids": unit.iter().map(|&member| crate::planner_selection::explained_root_id(&states[member].0, target)).collect::>(), + "retained_beside_raw_cost": option.beside_raw.map(|cost| cost.0), + "rebuilt_cost": option.alone.map(|cost| cost.0), + })).collect::>(), + "selected": "group", + }); + match cheapest_mix(&options) { + Err(reason) => event["reason"] = reason.into(), + Ok(Some((rebuild, cost))) if rebuild_is_cheaper(group, Some(cost)) => { + event["mixed_cost"] = cost.0.into(); + let mut rebuilt = Vec::new(); + let mut branches = Vec::new(); + for ((branch, unit), &rebuild) in units.iter().zip(&rebuild) { + match branch { + Some(branch) if !rebuild => branches.push(Rc::clone(branch)), + _ => { + rebuilt.extend(unit.iter().map(|&member| Rc::clone(&states[member].0))) + } + } + } + match mixed_placement(root, &rebuilt, &branches) { + Ok(placed) => { + event["selected"] = "mixed".into(); + for ((_, unit), &rebuild) in units.iter().zip(&rebuild) { + for &member in unit { + ephemeral[member] = rebuild; + } + } + mixed[query] = Some(placed); + } + Err(reason) => event["reason"] = reason.into(), + } + } + Ok(_) => {} + } + placement.trace.push(event); + } for (state_index, bindable, retained, rebuilt, retained_states) in decisions { let (state, consumers) = &states[state_index]; placement.trace.push(json!({ @@ -493,9 +702,18 @@ pub(super) fn place( placement.raw[query] = raw_programs[query].as_ref().map(|raw| RawQueryTimeProgram { program: raw.program.clone(), scans: raw.scans.clone(), + stored: Vec::new(), }); } placement.ephemeral[query] = chosen; + if let Some(mixed) = mixed[query].take() { + placement.roots[query] = Some(mixed.root); + placement.ephemeral[query] = mixed.rebuilt; + placement.native[query] = mixed.branches.iter().map(|(b, _)| Rc::clone(b)).collect(); + placement.branch_programs[query] = mixed.branches; + placement.complete_window[query] = mixed.retained; + placement.raw[query] = Some(mixed.program); + } } placement } @@ -524,7 +742,337 @@ fn raw_query_time_program(root: &QueryExpr) -> Result, + rebuilt: Vec>, + branches: Vec<(Rc, CompiledPhysicalDag)>, + retained: Vec>, + program: RawQueryTimeProgram, +} + +/// A unit's prices in a mixed plan: retained, with its batch read beside raw +/// inputs, and rebuilt from its own raw selectors. `None` is inadmissible or +/// unpriced. +struct MixedOption { + beside_raw: Option, + alone: Option, +} + +/// Per-state placement enumerates all 2^n assignments of a query's n units. +const MAX_MIXED_UNITS: usize = 8; + +/// The cheapest assignment that retains at least one unit and rebuilds at +/// least one, as the units to rebuild, or `None` when no such assignment is +/// admissible and priced. +fn cheapest_mix(options: &[MixedOption]) -> Result, Cost)>, String> { + if options.len() > MAX_MIXED_UNITS { + return Err(format!( + "per-state placement enumerates at most {MAX_MIXED_UNITS} units of a query, not {}; the group decision is kept", + options.len() + )); + } + let every = (1u32 << options.len()) - 1; + let mut best: Option<(u32, f64)> = None; + for rebuilt in 1..every { + let cost = options + .iter() + .enumerate() + .map(|(index, option)| { + if rebuilt >> index & 1 == 1 { + option.alone + } else { + option.beside_raw + } + .map(|cost| cost.0) + }) + .sum::>(); + if let Some(cost) = cost.filter(|cost| best.is_none_or(|(_, known)| *cost < known)) { + best = Some((rebuilt, cost)); + } + } + Ok(best.map(|(rebuilt, cost)| { + ( + (0..options.len()) + .map(|index| rebuilt >> index & 1 == 1) + .collect(), + Cost(cost), + ) + })) +} + +/// Branches of `root` the executor can read as stored native batches beside +/// raw inputs (#803): an ungrouped Sum or sketch the precompute program builds +/// from retained readouts. A state over raw samples is stored per population. +fn batch_branches(root: &Rc) -> Vec> { + fn walk(node: &Rc, branches: &mut Vec>) { + let ungrouped = matches!(&node.expr, SummaryExpr::SummaryAgg { + family: SummaryFamilyType::Sketch(..) + | SummaryFamilyType::ExactAggregate(planner_types::post_asap::ExactKind::Sum, _), + reduction: planner_types::pre_asap::Reduction::Reduce(keys), + .. + } if keys.keys().is_empty() && !keys.is_without()); + if ungrouped && over_readouts(node) && immutable_materialization_sources(node).is_some() { + branches.push(Rc::clone(node)); + return; + } + match &node.expr { + SummaryExpr::ValueOperation { child, .. } | SummaryExpr::SummaryAgg { child, .. } => { + walk(child, branches) + } + SummaryExpr::SummaryEstimate { summary_input, .. } => walk(summary_input, branches), + SummaryExpr::BinaryOp { lhs, rhs, .. } => { + walk(lhs, branches); + walk(rhs, branches); + } + _ => {} + } + } + let mut branches = Vec::new(); + walk(root, &mut branches); + branches +} + +/// The raw range selector a leaf state summarizes, and its query-time `Scan`. +fn leaf_selector(state: &SummaryNode) -> Option<(Rc, QueryTimeOperator)> { + let SummaryExpr::SummaryAgg { child, .. } = &state.expr else { + return None; + }; + let SummaryExpr::KeepPreAsap(selector) = &child.expr else { + return None; + }; + let scan = range_selector_scan(selector).ok()?; + Some((Rc::clone(selector), scan)) +} + +/// A leaf state of an identity-typed root: its original and retyped nodes, +/// its raw leaf, and the `Scan` that reads that leaf at query time. +struct TypedState { + original: Rc, + state: Rc, + leaf: Rc, + scan: QueryTimeOperator, +} + +/// `root` with each of `states`' raw selectors typed with the complete series +/// identity: raw rows read at query time carry it, and Planner's native +/// realization of a branch needs it. A per-series node keeps its input's +/// columns, so it gains the identity too; every other node is unchanged. +fn identity_typed( + root: &Rc, + states: &[Rc], +) -> Result<(Rc, Vec), String> { + use asap_physical_operators::physical_planner::promql_rows::SERIES_IDENTITY_COLUMN; + fn names(schema: &planner_types::post_asap::SummarySchema) -> Vec<&str> { + schema + .fields + .iter() + .map(|field| field.name.as_str()) + .collect() + } + // A node that passed its old input's columns through passes the identity on. + fn follow( + next: &mut SummaryNode, + old: &planner_types::post_asap::SummarySchema, + new: &planner_types::post_asap::SummarySchema, + ) { + if names(&next.schema) == names(old) { + if let Some(identity) = new + .fields + .iter() + .find(|field| field.name == SERIES_IDENTITY_COLUMN) + { + next.schema.fields.push(identity.clone()); + } + } + } + fn retype( + node: &Rc, + states: &[Rc], + typed: &mut Vec, + ) -> Result, String> { + let mut next = node.as_ref().clone(); + if states.iter().any(|state| Rc::ptr_eq(state, node)) { + let (selector, scan) = leaf_selector(node).ok_or("state has no raw selector")?; + let selector = + asap_physical_operators::physical_planner::promql_rows::with_series_identity( + &selector, + ) + .map_err(|e| e.to_string())?; + let SummaryExpr::SummaryAgg { child, .. } = &mut next.expr else { + unreachable!("leaf_selector matched a SummaryAgg") + }; + let mut leaf = crate::planner_selection::keep_pre_asap(&selector) + .map_err(|e| e.to_string())? + .as_ref() + .clone(); + leaf.guarantee = child.guarantee.clone(); + let (old, leaf) = (Rc::clone(child), Rc::new(leaf)); + *child = Rc::clone(&leaf); + follow(&mut next, &old.schema, &leaf.schema); + let state = Rc::new(next); + typed.push(TypedState { + original: Rc::clone(node), + state: Rc::clone(&state), + leaf, + scan, + }); + return Ok(state); + } + let unchanged = match &mut next.expr { + SummaryExpr::KeepPreAsap(_) => true, + SummaryExpr::ValueOperation { child, .. } | SummaryExpr::SummaryAgg { child, .. } => { + let old = Rc::clone(child); + *child = retype(child, states, typed)?; + let unchanged = Rc::ptr_eq(&old, child); + let new = Rc::clone(child); + follow(&mut next, &old.schema, &new.schema); + unchanged + } + SummaryExpr::SummaryEstimate { summary_input, .. } => { + let old = Rc::clone(summary_input); + *summary_input = retype(summary_input, states, typed)?; + Rc::ptr_eq(&old, summary_input) + } + SummaryExpr::BinaryOp { lhs, rhs, .. } => { + let (left, right) = (Rc::clone(lhs), Rc::clone(rhs)); + *lhs = retype(lhs, states, typed)?; + *rhs = retype(rhs, states, typed)?; + Rc::ptr_eq(&left, lhs) && Rc::ptr_eq(&right, rhs) + } + _ => return Err("mixed placement supports no such operator".into()), + }; + Ok(if unchanged { + Rc::clone(node) + } else { + Rc::new(next) + }) + } + let mut typed = Vec::new(); + let root = retype(root, states, &mut typed)?; + if typed.len() != states.len() { + return Err("a state is not reachable as a leaf of the root".into()); + } + Ok((root, typed)) +} + +/// Type `root` for a mixed plan that rebuilds `rebuilt` from raw rows and +/// reads each of `branches` as its stored native batch. Raw rows and Planner's +/// native precompute of a branch both carry the complete series identity. +fn mixed_placement( + root: &Rc, + rebuilt: &[Rc], + branches: &[Rc], +) -> Result { + let mut leaves = rebuilt.to_vec(); + for branch in branches { + leaves.extend(immutable_materialization_sources(branch).ok_or("branch has no sources")?); + } + let (root, typed) = identity_typed(root, &leaves)?; + let (rebuilt, sources): (Vec<_>, Vec<_>) = typed + .into_iter() + .partition(|state| rebuilt.iter().any(|s| Rc::ptr_eq(s, &state.original))); + // A branch is the Sum or sketch over readouts of its typed sources. + let typed_branches: Vec<_> = batch_branches(&root) + .into_iter() + .filter(|branch| { + immutable_materialization_sources(branch).is_some_and(|inputs| { + !inputs.is_empty() + && inputs + .iter() + .all(|input| sources.iter().any(|s| Rc::ptr_eq(&s.state, input))) + }) + }) + .collect(); + if typed_branches.len() != branches.len() { + return Err("a stored branch lost its sources while typing".into()); + } + let compiled = planner_types::post_asap::compile_post_asap_dag_with_node_ids(&root) + .map_err(|e| e.to_string())?; + let id = |node: &Rc| { + compiled + .node_ids + .node_id(node) + .map(|id| u64::from(id.0)) + .ok_or("mixed input is absent from the compiled DAG") + }; + // As in Planner's fixed-window realization, precompute builds the batch + // from the complete states of its sources. + let precompute = typed_branches + .iter() + .map(|branch| { + let inputs = immutable_materialization_sources(branch) + .expect("filtered above") + .iter() + .map(|source| { + let schema = std::sync::Arc::new(source.schema.clone()); + Ok((id(source)?, InputContract::bounded(schema))) + }) + .collect::, String>>()?; + let program = compile(&compiled.dag, inputs, &[id(branch)?]) + .map_err(|e| format!("stored branch precompute: {e}"))?; + Ok((Rc::clone(branch), program)) + }) + .collect::, String>>()?; + let program = mixed_program(&compiled, &rebuilt, &typed_branches)?; + let mut retained = typed_branches; + retained.extend(sources.into_iter().map(|state| state.state)); + Ok(MixedPlacement { + root, + rebuilt: rebuilt.into_iter().map(|state| state.state).collect(), + branches: precompute, + retained, + program, + }) +} + +/// Compile the typed DAG with each `stored` batch as a stored input and each +/// `rebuilt` state built from its leaf's raw rows at query time. +fn mixed_program( + compiled: &planner_types::post_asap::PostAsapDagCompilation, + rebuilt: &[TypedState], + stored: &[Rc], +) -> Result { + let id = |node: &Rc| { + compiled + .node_ids + .node_id(node) + .map(|id| u64::from(id.0)) + .ok_or("mixed input is absent from the compiled DAG") + }; + let mut inputs = BTreeMap::new(); + let mut scans = Vec::new(); + for state in rebuilt { + let slot = id(&state.leaf)?; + let schema = std::sync::Arc::new(state.leaf.schema.clone()); + inputs.insert(slot, InputContract::bounded(schema)); + scans.push((slot, state.scan.clone())); + } + let stored = stored + .iter() + .map(|batch| { + let slot = id(batch)?; + let schema = std::sync::Arc::new(batch.schema.clone()); + inputs.insert(slot, InputContract::bounded(schema)); + Ok((slot, Rc::clone(batch))) + }) + .collect::, String>>()?; + let program = compile(&compiled.dag, inputs, &[u64::from(compiled.dag.root.0)]) + .map_err(|e| e.to_string())?; + Ok(RawQueryTimeProgram { + program, + scans, + stored, + }) } /// `TimeRange { range, [TimeShift { offset }], Scan }` as a Prometheus range @@ -800,6 +1348,52 @@ fn find(root: &Rc, summary: &SummaryNode) -> Option mod tests { use super::*; + fn option(beside_raw: Option, alone: Option) -> MixedOption { + MixedOption { + beside_raw: beside_raw.map(Cost), + alone: alone.map(Cost), + } + } + + // The cheapest mix retains and rebuilds at least one unit each, and a unit + // with no stored batch is always rebuilt. + #[test] + fn cheapest_mix_is_the_cheapest_admissible_assignment() { + let (rebuilt, cost) = + cheapest_mix(&[option(None, Some(5.0)), option(Some(1.0), Some(9.0))]) + .unwrap() + .unwrap(); + assert_eq!((rebuilt, cost.0), (vec![true, false], 6.0)); + let (rebuilt, _) = cheapest_mix(&[ + option(Some(4.0), Some(1.0)), + option(Some(1.0), Some(4.0)), + option(Some(1.0), Some(4.0)), + ]) + .unwrap() + .unwrap(); + assert_eq!(rebuilt, [true, false, false]); + assert!( + cheapest_mix(&[option(None, Some(1.0)), option(None, Some(1.0))]) + .unwrap() + .is_none() + ); + assert!( + cheapest_mix(&[option(Some(1.0), None), option(Some(1.0), Some(1.0))]) + .unwrap() + .is_some_and(|(rebuilt, _)| rebuilt == [false, true]) + ); + } + + // Beyond the enumeration cap the search reports why it kept the group decision. + #[test] + fn cheapest_mix_rejects_more_units_than_its_cap() { + let options: Vec<_> = (0..=MAX_MIXED_UNITS) + .map(|_| option(Some(1.0), Some(1.0))) + .collect(); + let error = cheapest_mix(&options).unwrap_err(); + assert!(error.contains("at most 8 units"), "{error}"); + } + // Only a priced, strictly cheaper rebuild moves a state to query time. #[test] fn unpriced_alternatives_never_displace_retention() { diff --git a/control_plane/tests/lifecycle_placement.rs b/control_plane/tests/lifecycle_placement.rs index fa8dd3991..628005286 100644 --- a/control_plane/tests/lifecycle_placement.rs +++ b/control_plane/tests/lifecycle_placement.rs @@ -282,3 +282,103 @@ fn out_of_phase_consumers_are_priced_as_separate_installs() { + cost(&alone_second, "continuously_maintained_cost"); assert!((cost(&shared, "continuously_maintained_cost") - expected).abs() < 1e-9); } + +/// Two per-series Rate states under ungrouped Sums; the query lookback is +/// the 10m window. +const MIXED_QUERY: &str = "sum(rate(a[1m])) + sum(rate(b[10m]))"; + +/// The fixture evaluating `query` under `store` per retained byte-second. +fn two_state_plan(query: &str, store: f64) -> CompiledPhysicalPlan { + let mut wire = serde_json::to_value(fixture(store, false)).unwrap(); + wire["query_workload"]["repeating_queries"][0]["query"] = query.into(); + selected_plan(serde_json::from_value(wire).unwrap()) +} + +fn mixed_event(plan: &CompiledPhysicalPlan) -> Value { + let [event] = plan + .planner_selection_trace + .iter() + .filter(|entry| entry["stage"] == "deployment.mixed_placement") + .cloned() + .collect::>() + .try_into() + .unwrap(); + event +} + +// Where rebuilding the 1m Rate beats retaining it but the 10m branch is cheap +// as one complete window, the query rebuilds `a` from raw series beside the +// stored native batch of `sum(rate(b[10m]))`, the cheapest admissible choice. +#[test] +fn cheapest_assignment_retains_one_state_beside_a_rebuilt_one() { + let plan = two_state_plan(MIXED_QUERY, 1e-3); + assert_eq!(placements(&plan), ["ephemeral", "continuously_maintained"]); + let event = mixed_event(&plan); + assert_eq!(event["selected"], "mixed"); + assert!(cost(&event, "mixed_cost") < cost(&event, "group_cost")); + let installed = &plan.precompute_plan.materializations; + assert_eq!(installed.len(), 2, "b's Rate and its Sum batch"); + for state in installed { + assert_eq!(state.metric, "b"); + assert_eq!( + ( + state.window_size, + state.slide_interval, + &state.window_layout + ), + ( + 600, + 10, + &asap_types::WindowMaterializationLayout::FullWindow + ), + "a complete window sliding at the evaluation interval" + ); + } + assert!(installed.iter().any(|state| state.derived_input.is_some())); + assert_eq!( + raw_scans(&plan), + [&QueryTimeOperator::Scan { + metric: Some("a".into()), + matchers: vec![], + range_ms: Some(60_000), + offset_ms: 0, + }] + ); + let entry = plan.query_plan.entries.values().next().unwrap(); + entry.recover_vector_physical_dag().unwrap(); + assert!(entry.mixes_raw_and_stored_inputs()); +} + +// Without a store price, retaining both states stays cheapest; the plan is +// the group decision and has no raw input. +#[test] +fn free_store_keeps_the_group_decision_for_two_states() { + let plan = two_state_plan(MIXED_QUERY, 0.0); + assert_eq!( + placements(&plan), + ["continuously_maintained", "continuously_maintained"] + ); + assert_eq!(mixed_event(&plan)["selected"], "group"); + assert!(raw_scans(&plan).is_empty()); +} + +// Leaf Sum states are stored per population, not as native batches, so no +// mix is admissible and both states move together. +#[test] +fn inadmissible_mix_keeps_the_group_decision() { + let plan = two_state_plan( + "sum(sum_over_time(a[10m])) + sum(sum_over_time(b[1m]))", + 1e-3, + ); + assert_eq!(placements(&plan), ["ephemeral", "ephemeral"]); + let event = mixed_event(&plan); + assert_eq!(event["selected"], "group"); + assert!(event["units"] + .as_array() + .unwrap() + .iter() + .all(|unit| unit["retained_beside_raw_cost"].is_null())); + assert!(plan.precompute_plan.materializations.is_empty()); + let entry = plan.query_plan.entries.values().next().unwrap(); + assert!(!entry.mixes_raw_and_stored_inputs()); +} diff --git a/data_plane/tests/support/lifecycle_placement_process.rs b/data_plane/tests/support/lifecycle_placement_process.rs index a65763dc8..433d803a4 100644 --- a/data_plane/tests/support/lifecycle_placement_process.rs +++ b/data_plane/tests/support/lifecycle_placement_process.rs @@ -17,10 +17,25 @@ struct Deployment { /// Start the production binary from the fixture priced with `store` per /// retained byte-second. Its Prometheus serves `samples` of `m` for raw reads. async fn deploy(store: f64, samples: Vec<(i64, f64)>) -> Deployment { + deploy_query(QUERY, "m", store, samples, false).await +} + +/// [`deploy`] for `query`, whose Prometheus serves `samples` of `metric` for +/// its one-minute raw selector. Stored native batches are read only from +/// pinned Remote Write `revisions`. +async fn deploy_query( + query: &str, + metric: &'static str, + store: f64, + samples: Vec<(i64, f64)>, + revisions: bool, +) -> Deployment { + let raw_selector = format!(r#"{{__name__="{metric}"}}[60000ms]"#); let mut fixture: Value = serde_json::from_str(include_str!( "../../../docs/examples/asapquery-planning-snapshot.json" )) .unwrap(); + fixture["query_workload"]["repeating_queries"][0]["query"] = query.into(); fixture["implementation"]["scrape_interval_ms"] = 1000.into(); fixture["data_workload"]["data_ingestion_interval"]["value"] = 1000.into(); fixture["implementation"]["lifecycle_costs"]["store_per_byte_second"] = store.into(); @@ -30,7 +45,7 @@ async fn deploy(store: f64, samples: Vec<(i64, f64)>) -> Deployment { let requests = Arc::new(Mutex::new(Vec::>::new())); let recorded = requests.clone(); let matrix = serde_json::json!({"status": "success", "data": {"resultType": "matrix", - "result": [{"metric": {"__name__": "m", "instance": "a"}, + "result": [{"metric": {"__name__": metric, "instance": "a"}, "values": samples.iter().map(|(ms, value)| serde_json::json!([*ms as f64 / 1000.0, value.to_string()])).collect::>()}]}}); let empty = serde_json::json!({"status": "success", "data": {"resultType": "vector", "result": []}}); @@ -45,7 +60,7 @@ async fn deploy(store: f64, samples: Vec<(i64, f64)>) -> Deployment { "/api/v1/query", get(move |Query(params): Query>| { let recorded = recorded.clone(); - let raw = params.get("query").map(String::as_str) == Some(RAW_SELECTOR); + let raw = params.get("query") == Some(&raw_selector); let response = if raw { matrix.clone() } else { empty.clone() }; async move { recorded.lock().await.push(params); @@ -61,8 +76,15 @@ async fn deploy(store: f64, samples: Vec<(i64, f64)>) -> Deployment { let path = output.path().join("planning.json"); std::fs::write(&path, serde_json::to_vec(&priced).unwrap()).unwrap(); let port = unused_port(); + let mut command = Command::new(env!("CARGO_BIN_EXE_data_plane")); + if revisions { + command + .arg("--remote-write-revision-dir") + .arg(output.path().join("revisions")) + .args(["--remote-write-correction-horizon-ms", "3600000"]); + } let mut child = ChildGuard( - Command::new(env!("CARGO_BIN_EXE_data_plane")) + command .args(["--profile", "asapquery", "--planning-snapshot"]) .arg(path) .args([ @@ -180,3 +202,80 @@ async fn expensive_summary_store_rebuilds_state_from_raw_series() { ); deployment.prometheus.abort(); } + +// With a mid-priced store, the query rebuilds `sum(rate(a[1m]))` from raw `a` +// read at t_q and reads `sum(rate(b[10m]))` as the stored batch of its newest +// complete window; the answer is exact for that window, whose lag is within +// the default bound of one 10 s slide. +#[tokio::test] +async fn mixed_placement_combines_raw_series_with_stored_state_within_the_lag_bound() { + const MIXED: &str = "sum(rate(a[1m])) + sum(rate(b[10m]))"; + let origin = origin_ms(); + let at_ms = origin + 650_000; + // Counters rising 1/s (raw `a`) and 2/s (stored `b`): each rate is exact + // for any window their samples span. + let raw: Vec<_> = (0..6) + .map(|i| { + let ms = at_ms - 50_000 + i * 10_000; + (ms, 10_000.0 + (ms - origin) as f64 / 1000.0) + }) + .collect(); + let deployment = deploy_query(MIXED, "a", 1e-3, raw, true).await; + let client = reqwest::Client::new(); + // Samples past t_q close the stored window that ends at t_q. + let stored: Vec<_> = (1..=700) + .map(|i| (origin + i * 1000, 10_000.0 + 2.0 * i as f64)) + .collect(); + let wire = WriteRequest { + timeseries: vec![series_with_labels("b", &[("instance", "a")], &stored)], + }; + assert_eq!(remote_write(&client, &deployment.backend, &wire).await, 204); + let mut last = Value::Null; + let mut lag = None; + for _ in 0..80 { + let response = client + .get(format!("{}/api/v1/query", deployment.backend)) + .query(&[ + ("query", MIXED.to_string()), + ("time", (at_ms as f64 / 1000.0).to_string()), + ]) + .send() + .await + .unwrap(); + lag = response + .headers() + .get("x-asap-stored-input-lag-ms") + .map(|value| value.to_str().unwrap().parse::().unwrap()); + last = response.json().await.unwrap(); + if is_warm(&last) && lag.is_some() { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + } + let lag = lag.unwrap_or_else(|| { + let log = std::fs::read_to_string(deployment._output.path().join("query_engine.log")) + .unwrap_or_default(); + panic!("no mixed answer: {last}\nbackend log:\n{log}") + }); + assert!(lag <= 10_000, "lag {lag} beyond one slide"); + // Exact reference: rate(a) over (t_q - 1m, t_q] plus rate(b) over the + // stored window (t_q - lag - 10m, t_q - lag]. + let value = first_value(&last, "value").unwrap(); + assert!((value - 3.0).abs() < 1e-9, "{last}"); + let requests = deployment.requests.lock().await; + assert!( + requests.iter().any(|request| { + request.get("query").map(String::as_str) == Some(r#"{__name__="a"}[60000ms]"#) + && request.get("time").map(String::as_str) + == Some(format!("{:.3}", at_ms as f64 / 1000.0).as_str()) + }), + "raw `a` is read at t_q: {requests:?}" + ); + assert!( + !requests.iter().any(|request| request + .get("query") + .is_some_and(|query| query.contains(r#""b""#))), + "stored `b` is not read from Prometheus: {requests:?}" + ); + deployment.prometheus.abort(); +} From 0e32911e8cf45f416c04f4a5c9db8ff8163eb8e8 Mon Sep 17 00:00:00 2001 From: zzylol Date: Wed, 30 Sep 2026 14:18:48 +0000 Subject: [PATCH 2/2] fix: constrain and price admissible mixed placement branches Preserve group placement for unsupported or shared branch layouts, scale complete-window maintenance by overlap, and verify the selected stored window in the process oracle. Co-Authored-By: Claude Opus 5.5 --- .../src/physical/compiler/placement.rs | 301 +++++++++++------- control_plane/tests/lifecycle_placement.rs | 37 ++- .../support/lifecycle_placement_process.rs | 19 +- 3 files changed, 213 insertions(+), 144 deletions(-) diff --git a/control_plane/src/physical/compiler/placement.rs b/control_plane/src/physical/compiler/placement.rs index 526aba900..2bd7537ce 100644 --- a/control_plane/src/physical/compiler/placement.rs +++ b/control_plane/src/physical/compiler/placement.rs @@ -33,14 +33,7 @@ pub(super) struct RawQueryTimeProgram { pub(super) struct Placement { ephemeral: Vec>>, raw: Vec>, - /// Native branches a mixed query maintains and reads as stored batches. - native: Vec>>, - /// The precompute program building each native branch's batch. - branch_programs: Vec, CompiledPhysicalDag)>>, - /// Retained states of a mixed query, each installed as a complete window. - complete_window: Vec>>, - /// A mixed query's root, its raw selectors typed with the series identity. - roots: Vec>>, + mixed: Vec>, pub(super) trace: Vec, } @@ -52,31 +45,40 @@ impl Placement { } pub(super) fn raw_program(&self, query: usize) -> Option<&RawQueryTimeProgram> { - self.raw.get(query).and_then(Option::as_ref) + self.raw + .get(query) + .and_then(Option::as_ref) + .or_else(|| self.mixed(query).map(|mixed| &mixed.program)) + } + + fn mixed(&self, query: usize) -> Option<&MixedPlacement> { + self.mixed.get(query).and_then(Option::as_ref) } /// Whether retained `state` serves stored input beside `query`'s raw inputs. pub(super) fn beside_raw(&self, query: usize, state: &Rc) -> bool { - self.complete_window - .get(query) - .is_some_and(|states| states.iter().any(|s| Rc::ptr_eq(s, state))) + self.mixed(query) + .is_some_and(|mixed| mixed.retained.iter().any(|s| Rc::ptr_eq(s, state))) } /// The root a mixed query compiles instead of Planner's selected root. pub(super) fn root(&self, query: usize) -> Option<&Rc> { - self.roots.get(query).and_then(Option::as_ref) + self.mixed(query).map(|mixed| &mixed.root) } + /// Native branches a mixed query maintains and reads as stored batches. pub(super) fn native_branches(&self, query: usize) -> &[Rc] { - self.native.get(query).map_or(&[], Vec::as_slice) + self.mixed(query).map_or(&[], |mixed| &mixed.branches) } /// Each native branch of `query` with the precompute program building it. pub(super) fn branch_programs( &self, query: usize, - ) -> &[(Rc, CompiledPhysicalDag)] { - self.branch_programs.get(query).map_or(&[], Vec::as_slice) + ) -> impl Iterator, &CompiledPhysicalDag)> { + self.mixed(query) + .into_iter() + .flat_map(|mixed| mixed.branches.iter().zip(&mixed.precompute)) } } @@ -262,10 +264,7 @@ pub(super) fn place( let mut placement = Placement { ephemeral: vec![Vec::new(); queries.len()], raw: (0..queries.len()).map(|_| None).collect(), - native: vec![Vec::new(); queries.len()], - branch_programs: (0..queries.len()).map(|_| Vec::new()).collect(), - complete_window: vec![Vec::new(); queries.len()], - roots: vec![None; queries.len()], + mixed: (0..queries.len()).map(|_| None).collect(), trace: Vec::new(), }; let (Some(workload), Some(data), Some(first)) = ( @@ -479,10 +478,9 @@ pub(super) fn place( .sum::>(); decisions.push((state_index, bindable, retained, rebuilt, retained_states)); } - // A query rebuilds either all of its states or none: raw query-time inputs - // and exact subtrees share no snapshot with installed state. States linked - // through a query therefore move together, and only when the raw programs - // of all their queries cost less than retaining all of them. + // First choose a group baseline: states linked through a query move + // together when rebuilding all consumers costs less than retaining them. + // The bounded-lag search below may replace it with an admissible mix. let index_of = |state: &Rc| states.iter().position(|(s, _)| Rc::ptr_eq(s, state)); let mut linked: Vec = (0..states.len()).collect(); for owned in &query_states { @@ -524,6 +522,8 @@ pub(super) fn place( if !raw_bindable || members.len() < 2 || members.iter().any(|&member| states[member].1 != [query]) + // Planner's own native realization of the root owns its layout. + || queries[query].physical_candidate.is_some() { continue; } @@ -575,31 +575,39 @@ pub(super) fn place( request.query_retention_margin_ms, true, )?; + let is_branch = Rc::ptr_eq(state, branch); // The executor reads the batch as one window of the query lookback. - if Rc::ptr_eq(state, branch) + if is_branch && window.window_secs.saturating_mul(1_000) != queries[query].query_lookback_ms { return None; } + // Each raw sample updates every open complete window. The + // batch is built once per window from finished readouts. + let overlap = if is_branch { + 1.0 + } else { + window.window_secs.div_ceil(window.slide_secs.max(1)) as f64 + }; + let costs = LifecycleUnitCosts { + maintenance_per_update: lifecycle.costs.maintenance_per_update * overlap, + ..lifecycle.costs.clone() + }; price( state, &[query], SummaryMaintenanceLifecycle::ContinuouslyMaintained, - &model( - lifecycle.costs.clone(), - Some(retained), - lifecycle.evaluation_interval_ms, - ), + &model(costs, Some(retained), lifecycle.evaluation_interval_ms), ) .map(|cost| cost.0) }) .sum::>() .map(Cost) }; - // Each unit: an optional native branch and the member states it - // retains or rebuilds; a state outside every branch is rebuilt. - let mut units: Vec<(Option>, Vec)> = batch_branches(root) + // A branch retains its sources with it; a state outside every branch + // is always rebuilt. Branches sharing a source are not enumerated. + let branches: Vec<(Rc, Vec)> = batch_branches(root) .into_iter() .filter_map(|branch| { let sources = immutable_materialization_sources(&branch)? @@ -611,64 +619,74 @@ pub(super) fn place( .find(|&member| Rc::ptr_eq(&states[member].0, source)) }) .collect::>>()?; - Some((Some(branch), sources)) + Some((branch, sources)) }) .collect(); - for &member in &members { - if !units.iter().any(|(_, unit)| unit.contains(&member)) { - units.push((None, vec![member])); - } + let mut sourced: Vec = branches.iter().flat_map(|(_, s)| s.clone()).collect(); + let count = sourced.len(); + sourced.sort_unstable(); + sourced.dedup(); + if branches.is_empty() || sourced.len() != count { + continue; } - let options: Vec = units + let others: Vec = members .iter() - .map(|(branch, unit)| MixedOption { - beside_raw: branch.as_ref().and_then(|branch| stored(branch, unit)), - alone: unit - .iter() - .map(|&member| alone(member).map(|cost| cost.0)) - .sum::>() - .map(Cost), + .copied() + .filter(|member| !sourced.contains(member)) + .collect(); + let rebuild_all = |unit: &[usize]| { + unit.iter() + .map(|&member| alone(member).map(|cost| cost.0)) + .sum::>() + .map(Cost) + }; + let options: Vec = branches + .iter() + .map(|(branch, sources)| MixedOption { + beside_raw: stored(branch, sources), + alone: rebuild_all(sources), }) .collect(); + let others_cost = (!others.is_empty()).then(|| rebuild_all(&others)); let group = if ephemeral[members[0]] { total(linked[members[0]], |decision| decision.3) } else { total(linked[members[0]], |decision| decision.2) }; let target = &queries[query].accuracy_target; + let id = |node: &Rc| crate::planner_selection::explained_root_id(node, target); let mut event = json!({ "stage": "deployment.mixed_placement", "query_ids": [&queries[query].query_id], "group_cost": group.map(|cost| cost.0), - "units": units.iter().zip(&options).map(|((branch, unit), option)| json!({ - "native_branch_id": branch.as_ref().map(|branch| crate::planner_selection::explained_root_id(branch, target)), - "logical_root_ids": unit.iter().map(|&member| crate::planner_selection::explained_root_id(&states[member].0, target)).collect::>(), + "branches": branches.iter().zip(&options).map(|((branch, sources), option)| json!({ + "native_branch_id": id(branch), + "logical_root_ids": sources.iter().map(|&member| id(&states[member].0)).collect::>(), "retained_beside_raw_cost": option.beside_raw.map(|cost| cost.0), "rebuilt_cost": option.alone.map(|cost| cost.0), })).collect::>(), + "others_rebuilt_cost": others_cost.flatten().map(|cost| cost.0), "selected": "group", }); - match cheapest_mix(&options) { + match cheapest_mix(&options, others_cost) { Err(reason) => event["reason"] = reason.into(), Ok(Some((rebuild, cost))) if rebuild_is_cheaper(group, Some(cost)) => { event["mixed_cost"] = cost.0.into(); - let mut rebuilt = Vec::new(); - let mut branches = Vec::new(); - for ((branch, unit), &rebuild) in units.iter().zip(&rebuild) { - match branch { - Some(branch) if !rebuild => branches.push(Rc::clone(branch)), - _ => { - rebuilt.extend(unit.iter().map(|&member| Rc::clone(&states[member].0))) - } + let mut rebuilt: Vec<_> = others.iter().map(|&m| Rc::clone(&states[m].0)).collect(); + let mut kept = Vec::new(); + for ((branch, sources), &rebuild) in branches.iter().zip(&rebuild) { + if rebuild { + rebuilt.extend(sources.iter().map(|&m| Rc::clone(&states[m].0))); + } else { + kept.push(Rc::clone(branch)); } } - match mixed_placement(root, &rebuilt, &branches) { + match mixed_placement(root, &rebuilt, &kept) { Ok(placed) => { event["selected"] = "mixed".into(); - for ((_, unit), &rebuild) in units.iter().zip(&rebuild) { - for &member in unit { - ephemeral[member] = rebuild; - } + for &member in &members { + ephemeral[member] = + rebuilt.iter().any(|s| Rc::ptr_eq(s, &states[member].0)); } mixed[query] = Some(placed); } @@ -707,12 +725,8 @@ pub(super) fn place( } placement.ephemeral[query] = chosen; if let Some(mixed) = mixed[query].take() { - placement.roots[query] = Some(mixed.root); - placement.ephemeral[query] = mixed.rebuilt; - placement.native[query] = mixed.branches.iter().map(|(b, _)| Rc::clone(b)).collect(); - placement.branch_programs[query] = mixed.branches; - placement.complete_window[query] = mixed.retained; - placement.raw[query] = Some(mixed.program); + placement.ephemeral[query] = mixed.rebuilt.clone(); + placement.mixed[query] = Some(mixed); } } placement @@ -755,36 +769,43 @@ fn raw_query_time_program(root: &QueryExpr) -> Result, rebuilt: Vec>, - branches: Vec<(Rc, CompiledPhysicalDag)>, + branches: Vec>, + precompute: Vec, retained: Vec>, program: RawQueryTimeProgram, } -/// A unit's prices in a mixed plan: retained, with its batch read beside raw -/// inputs, and rebuilt from its own raw selectors. `None` is inadmissible or -/// unpriced. +/// A native branch's prices in a mixed plan: retained with its sources and +/// read as its batch, or its sources rebuilt from their raw selectors. `None` +/// is inadmissible or unpriced. struct MixedOption { beside_raw: Option, alone: Option, } -/// Per-state placement enumerates all 2^n assignments of a query's n units. -const MAX_MIXED_UNITS: usize = 8; +/// Per-state placement enumerates all 2^n assignments of a query's n branches. +const MAX_MIXED_BRANCHES: usize = 8; -/// The cheapest assignment that retains at least one unit and rebuilds at -/// least one, as the units to rebuild, or `None` when no such assignment is -/// admissible and priced. -fn cheapest_mix(options: &[MixedOption]) -> Result, Cost)>, String> { - if options.len() > MAX_MIXED_UNITS { +/// The cheapest assignment that retains at least one branch and rebuilds at +/// least one state, as the branches to rebuild. `others` is the cost of the +/// states outside every branch, which are always rebuilt, if there are any. +/// `None` when no such assignment is admissible and priced. +fn cheapest_mix( + branches: &[MixedOption], + others: Option>, +) -> Result, Cost)>, String> { + if branches.len() > MAX_MIXED_BRANCHES { return Err(format!( - "per-state placement enumerates at most {MAX_MIXED_UNITS} units of a query, not {}; the group decision is kept", - options.len() + "per-state placement enumerates at most {MAX_MIXED_BRANCHES} stored branches of a query, not {}; the group decision is kept", + branches.len() )); } - let every = (1u32 << options.len()) - 1; + let every = (1u32 << branches.len()) - 1; + // Rebuilding nothing is a mix only when other states are rebuilt. + let first = if others.is_some() { 0 } else { 1 }; let mut best: Option<(u32, f64)> = None; - for rebuilt in 1..every { - let cost = options + for rebuilt in first..every { + let cost = branches .iter() .enumerate() .map(|(index, option)| { @@ -793,8 +814,9 @@ fn cheapest_mix(options: &[MixedOption]) -> Result, Cost)>, St } else { option.beside_raw } - .map(|cost| cost.0) }) + .chain(others) + .map(|cost| cost.map(|cost| cost.0)) .sum::>(); if let Some(cost) = cost.filter(|cost| best.is_none_or(|(_, known)| *cost < known)) { best = Some((rebuilt, cost)); @@ -802,7 +824,7 @@ fn cheapest_mix(options: &[MixedOption]) -> Result, Cost)>, St } Ok(best.map(|(rebuilt, cost)| { ( - (0..options.len()) + (0..branches.len()) .map(|index| rebuilt >> index & 1 == 1) .collect(), Cost(cost), @@ -811,16 +833,24 @@ fn cheapest_mix(options: &[MixedOption]) -> Result, Cost)>, St } /// Branches of `root` the executor can read as stored native batches beside -/// raw inputs (#803): an ungrouped Sum or sketch the precompute program builds -/// from retained readouts. A state over raw samples is stored per population. +/// raw inputs (#803): an ungrouped Sum or heap sketch the precompute program +/// builds from retained readouts. A state over raw samples is stored per population. fn batch_branches(root: &Rc) -> Vec> { fn walk(node: &Rc, branches: &mut Vec>) { + use planner_types::post_asap::{ExactKind, SketchAlgorithm}; + // The families #803 reads out independently of the evaluation time. let ungrouped = matches!(&node.expr, SummaryExpr::SummaryAgg { - family: SummaryFamilyType::Sketch(..) - | SummaryFamilyType::ExactAggregate(planner_types::post_asap::ExactKind::Sum, _), + family, reduction: planner_types::pre_asap::Reduction::Reduce(keys), .. - } if keys.keys().is_empty() && !keys.is_without()); + } if keys.keys().is_empty() && !keys.is_without() && match family { + SummaryFamilyType::Sketch(kind, _) => matches!( + kind.algorithm(), + SketchAlgorithm::CmsWithHeap | SketchAlgorithm::CountSketchWithHeap + ), + SummaryFamilyType::ExactAggregate(ExactKind::Sum, _) => true, + _ => false, + }); if ungrouped && over_readouts(node) && immutable_materialization_sources(node).is_some() { branches.push(Rc::clone(node)); return; @@ -879,13 +909,21 @@ fn identity_typed( .map(|field| field.name.as_str()) .collect() } - // A node that passed its old input's columns through passes the identity on. + // A readout or per-series state that passed its old input's columns + // through passes the identity on; any other node keeps its schema. fn follow( next: &mut SummaryNode, old: &planner_types::post_asap::SummarySchema, new: &planner_types::post_asap::SummarySchema, ) { - if names(&next.schema) == names(old) { + let per_series = match &next.expr { + SummaryExpr::ValueOperation { .. } => true, + SummaryExpr::SummaryAgg { reduction, .. } => { + matches!(reduction, planner_types::pre_asap::Reduction::PerEntity) + } + _ => false, + }; + if per_series && names(&next.schema) == names(old) { if let Some(identity) = new .fields .iter() @@ -938,15 +976,23 @@ fn identity_typed( follow(&mut next, &old.schema, &new.schema); unchanged } + // These nodes' schemas do not follow their inputs, so an input + // that gained the identity has no typed form here. SummaryExpr::SummaryEstimate { summary_input, .. } => { let old = Rc::clone(summary_input); *summary_input = retype(summary_input, states, typed)?; + if old.schema != summary_input.schema { + return Err("an estimate over a per-series input cannot be retyped".into()); + } Rc::ptr_eq(&old, summary_input) } SummaryExpr::BinaryOp { lhs, rhs, .. } => { let (left, right) = (Rc::clone(lhs), Rc::clone(rhs)); *lhs = retype(lhs, states, typed)?; *rhs = retype(rhs, states, typed)?; + if left.schema != lhs.schema || right.schema != rhs.schema { + return Err("a per-series binary input cannot be retyped".into()); + } Rc::ptr_eq(&left, lhs) && Rc::ptr_eq(&right, rhs) } _ => return Err("mixed placement supports no such operator".into()), @@ -996,6 +1042,8 @@ fn mixed_placement( if typed_branches.len() != branches.len() { return Err("a stored branch lost its sources while typing".into()); } + // Compilation numbers the same root identically when it installs the + // plan, so these slots and precompute roots name its installed nodes. let compiled = planner_types::post_asap::compile_post_asap_dag_with_node_ids(&root) .map_err(|e| e.to_string())?; let id = |node: &Rc| { @@ -1020,16 +1068,17 @@ fn mixed_placement( .collect::, String>>()?; let program = compile(&compiled.dag, inputs, &[id(branch)?]) .map_err(|e| format!("stored branch precompute: {e}"))?; - Ok((Rc::clone(branch), program)) + Ok(program) }) .collect::, String>>()?; let program = mixed_program(&compiled, &rebuilt, &typed_branches)?; - let mut retained = typed_branches; + let mut retained = typed_branches.clone(); retained.extend(sources.into_iter().map(|state| state.state)); Ok(MixedPlacement { root, rebuilt: rebuilt.into_iter().map(|state| state.state).collect(), - branches: precompute, + branches: typed_branches, + precompute, retained, program, }) @@ -1355,43 +1404,49 @@ mod tests { } } - // The cheapest mix retains and rebuilds at least one unit each, and a unit - // with no stored batch is always rebuilt. + // The cheapest mix retains at least one branch and rebuilds at least one + // state; states outside every branch are always rebuilt. #[test] fn cheapest_mix_is_the_cheapest_admissible_assignment() { - let (rebuilt, cost) = - cheapest_mix(&[option(None, Some(5.0)), option(Some(1.0), Some(9.0))]) - .unwrap() - .unwrap(); - assert_eq!((rebuilt, cost.0), (vec![true, false], 6.0)); - let (rebuilt, _) = cheapest_mix(&[ - option(Some(4.0), Some(1.0)), - option(Some(1.0), Some(4.0)), - option(Some(1.0), Some(4.0)), - ]) + let (rebuilt, cost) = cheapest_mix(&[option(Some(1.0), Some(9.0))], Some(Some(Cost(5.0)))) + .unwrap() + .unwrap(); + assert_eq!((rebuilt, cost.0), (vec![false], 6.0)); + let (rebuilt, _) = cheapest_mix( + &[ + option(Some(4.0), Some(1.0)), + option(Some(1.0), Some(4.0)), + option(Some(1.0), Some(4.0)), + ], + None, + ) .unwrap() .unwrap(); assert_eq!(rebuilt, [true, false, false]); - assert!( - cheapest_mix(&[option(None, Some(1.0)), option(None, Some(1.0))]) - .unwrap() - .is_none() - ); - assert!( - cheapest_mix(&[option(Some(1.0), None), option(Some(1.0), Some(1.0))]) - .unwrap() - .is_some_and(|(rebuilt, _)| rebuilt == [false, true]) - ); + // One branch alone and nothing else to rebuild is no mix. + assert!(cheapest_mix(&[option(Some(1.0), Some(1.0))], None) + .unwrap() + .is_none()); + // An unpriced state outside every branch prices every mix out. + assert!(cheapest_mix(&[option(Some(1.0), Some(1.0))], Some(None)) + .unwrap() + .is_none()); + assert!(cheapest_mix( + &[option(None, Some(1.0)), option(Some(1.0), Some(1.0))], + None + ) + .unwrap() + .is_some_and(|(rebuilt, _)| rebuilt == [true, false])); } // Beyond the enumeration cap the search reports why it kept the group decision. #[test] - fn cheapest_mix_rejects_more_units_than_its_cap() { - let options: Vec<_> = (0..=MAX_MIXED_UNITS) + fn cheapest_mix_rejects_more_branches_than_its_cap() { + let options: Vec<_> = (0..=MAX_MIXED_BRANCHES) .map(|_| option(Some(1.0), Some(1.0))) .collect(); - let error = cheapest_mix(&options).unwrap_err(); - assert!(error.contains("at most 8 units"), "{error}"); + let error = cheapest_mix(&options, None).unwrap_err(); + assert!(error.contains("at most 8 stored branches"), "{error}"); } // Only a priced, strictly cheaper rebuild moves a state to query time. diff --git a/control_plane/tests/lifecycle_placement.rs b/control_plane/tests/lifecycle_placement.rs index 628005286..e8c9b965f 100644 --- a/control_plane/tests/lifecycle_placement.rs +++ b/control_plane/tests/lifecycle_placement.rs @@ -362,23 +362,28 @@ fn free_store_keeps_the_group_decision_for_two_states() { assert!(raw_scans(&plan).is_empty()); } -// Leaf Sum states are stored per population, not as native batches, so no -// mix is admissible and both states move together. +// Leaf Sum states are stored per population and a grouped Sum is no single +// batch, so neither query has an admissible mix: its states move together. #[test] fn inadmissible_mix_keeps_the_group_decision() { - let plan = two_state_plan( + for query in [ "sum(sum_over_time(a[10m])) + sum(sum_over_time(b[1m]))", - 1e-3, - ); - assert_eq!(placements(&plan), ["ephemeral", "ephemeral"]); - let event = mixed_event(&plan); - assert_eq!(event["selected"], "group"); - assert!(event["units"] - .as_array() - .unwrap() - .iter() - .all(|unit| unit["retained_beside_raw_cost"].is_null())); - assert!(plan.precompute_plan.materializations.is_empty()); - let entry = plan.query_plan.entries.values().next().unwrap(); - assert!(!entry.mixes_raw_and_stored_inputs()); + "sum by (job) (rate(a[1m])) + sum by (job) (rate(b[10m]))", + ] { + let plan = two_state_plan(query, 1e-3); + let placed = placements(&plan); + assert!(placed.len() >= 2, "{query}: {placed:?}"); + assert!( + placed.iter().all(|selected| selected == &placed[0]), + "{query}: {placed:?}" + ); + assert!(plan + .planner_selection_trace + .iter() + .filter(|entry| entry["stage"] == "deployment.mixed_placement") + .all(|event| event["selected"] == "group")); + for entry in plan.query_plan.entries.values() { + assert!(!entry.mixes_raw_and_stored_inputs(), "{query}"); + } + } } diff --git a/data_plane/tests/support/lifecycle_placement_process.rs b/data_plane/tests/support/lifecycle_placement_process.rs index 433d803a4..9414f6f9d 100644 --- a/data_plane/tests/support/lifecycle_placement_process.rs +++ b/data_plane/tests/support/lifecycle_placement_process.rs @@ -212,8 +212,9 @@ async fn mixed_placement_combines_raw_series_with_stored_state_within_the_lag_bo const MIXED: &str = "sum(rate(a[1m])) + sum(rate(b[10m]))"; let origin = origin_ms(); let at_ms = origin + 650_000; - // Counters rising 1/s (raw `a`) and 2/s (stored `b`): each rate is exact - // for any window their samples span. + // Raw `a` rises 1/s. Stored `b` rises 2/s, then 4/s over the last 5 s + // before t_q, so its rate tells which stored window was read. Its samples + // sit mid-second, off every window boundary. let raw: Vec<_> = (0..6) .map(|i| { let ms = at_ms - 50_000 + i * 10_000; @@ -223,8 +224,9 @@ async fn mixed_placement_combines_raw_series_with_stored_state_within_the_lag_bo let deployment = deploy_query(MIXED, "a", 1e-3, raw, true).await; let client = reqwest::Client::new(); // Samples past t_q close the stored window that ends at t_q. + let counter = |i: i64| 10_000.0 + 2.0 * i as f64 + 2.0 * (i - 645).max(0) as f64; let stored: Vec<_> = (1..=700) - .map(|i| (origin + i * 1000, 10_000.0 + 2.0 * i as f64)) + .map(|i| (origin + i * 1000 + 500, counter(i))) .collect(); let wire = WriteRequest { timeseries: vec![series_with_labels("b", &[("instance", "a")], &stored)], @@ -258,10 +260,17 @@ async fn mixed_placement_combines_raw_series_with_stored_state_within_the_lag_bo panic!("no mixed answer: {last}\nbackend log:\n{log}") }); assert!(lag <= 10_000, "lag {lag} beyond one slide"); + assert_eq!(lag % 10_000, 0, "stored windows end on the 10 s grid"); // Exact reference: rate(a) over (t_q - 1m, t_q] plus rate(b) over the - // stored window (t_q - lag - 10m, t_q - lag]. + // stored window (t_s - 10m, t_s], t_s = t_q - lag. Its samples span 599 s + // and extrapolate half a second to each window edge. + let end = 650 - lag as i64 / 1000; + let expected = 1.0 + (counter(end - 1) - counter(end - 600)) / 599.0; let value = first_value(&last, "value").unwrap(); - assert!((value - 3.0).abs() < 1e-9, "{last}"); + assert!( + (value - expected).abs() < 1e-9, + "expected {expected}: {last}" + ); let requests = deployment.requests.lock().await; assert!( requests.iter().any(|request| {