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..2bd7537ce 100644 --- a/control_plane/src/physical/compiler/placement.rs +++ b/control_plane/src/physical/compiler/placement.rs @@ -19,16 +19,21 @@ 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>, + mixed: Vec>, pub(super) trace: Vec, } @@ -40,7 +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.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.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.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, + ) -> impl Iterator, &CompiledPhysicalDag)> { + self.mixed(query) + .into_iter() + .flat_map(|mixed| mixed.branches.iter().zip(&mixed.precompute)) } } @@ -128,15 +166,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 +186,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 +264,7 @@ pub(super) fn place( let mut placement = Placement { ephemeral: vec![Vec::new(); queries.len()], raw: (0..queries.len()).map(|_| None).collect(), + mixed: (0..queries.len()).map(|_| None).collect(), trace: Vec::new(), }; let (Some(workload), Some(data), Some(first)) = ( @@ -341,6 +382,7 @@ pub(super) fn place( &selected_states[query], environment, request.query_retention_margin_ms, + false, ) }); let layout = installed @@ -436,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 { @@ -462,7 +503,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 +511,192 @@ 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]) + // Planner's own native realization of the root owns its layout. + || queries[query].physical_candidate.is_some() + { + 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, + )?; + let is_branch = Rc::ptr_eq(state, branch); + // The executor reads the batch as one window of the query lookback. + 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(costs, Some(retained), lifecycle.evaluation_interval_ms), + ) + .map(|cost| cost.0) + }) + .sum::>() + .map(Cost) + }; + // 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)? + .iter() + .map(|source| { + members + .iter() + .copied() + .find(|&member| Rc::ptr_eq(&states[member].0, source)) + }) + .collect::>>()?; + Some((branch, sources)) + }) + .collect(); + 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 others: Vec = members + .iter() + .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), + "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, 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<_> = 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, &kept) { + Ok(placed) => { + event["selected"] = "mixed".into(); + for &member in &members { + ephemeral[member] = + rebuilt.iter().any(|s| Rc::ptr_eq(s, &states[member].0)); + } + 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 +720,14 @@ 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.ephemeral[query] = mixed.rebuilt.clone(); + placement.mixed[query] = Some(mixed); + } } placement } @@ -524,7 +756,372 @@ fn raw_query_time_program(root: &QueryExpr) -> Result, + rebuilt: Vec>, + branches: Vec>, + precompute: Vec, + retained: Vec>, + program: RawQueryTimeProgram, +} + +/// 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 branches. +const MAX_MIXED_BRANCHES: usize = 8; + +/// 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_BRANCHES} stored branches of a query, not {}; the group decision is kept", + branches.len() + )); + } + 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 first..every { + let cost = branches + .iter() + .enumerate() + .map(|(index, option)| { + if rebuilt >> index & 1 == 1 { + option.alone + } else { + option.beside_raw + } + }) + .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)); + } + } + Ok(best.map(|(rebuilt, cost)| { + ( + (0..branches.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 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, + reduction: planner_types::pre_asap::Reduction::Reduce(keys), + .. + } 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; + } + 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 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, + ) { + 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() + .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 + } + // 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()), + }; + 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()); + } + // 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| { + 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(program) + }) + .collect::, String>>()?; + let program = mixed_program(&compiled, &rebuilt, &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: typed_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 +1397,58 @@ 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 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(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]); + // 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_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, None).unwrap_err(); + assert!(error.contains("at most 8 stored branches"), "{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..e8c9b965f 100644 --- a/control_plane/tests/lifecycle_placement.rs +++ b/control_plane/tests/lifecycle_placement.rs @@ -282,3 +282,108 @@ 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 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() { + for query in [ + "sum(sum_over_time(a[10m])) + sum(sum_over_time(b[1m]))", + "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 a65763dc8..9414f6f9d 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,89 @@ 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; + // 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; + (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 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 + 500, counter(i))) + .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"); + 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_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 - expected).abs() < 1e-9, + "expected {expected}: {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(); +}