Skip to content

Commit 257c07d

Browse files
zzylolzzylol
committed
Retain selected native candidates and execute revisioned query ensembles
1 parent cd0bbe3 commit 257c07d

13 files changed

Lines changed: 596 additions & 67 deletions

File tree

‎control_plane/src/main.rs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -590,6 +590,7 @@ fn compile_physical_plan_request(
590590
query_id: query.query_id,
591591
query_string: query.query_string,
592592
selected_plan_root: post_asap,
593+
physical_candidate: None,
593594
legacy_query_source: planner_types::pre_asap::Source::TimeSeries {
594595
metric: query.metric,
595596
},

‎control_plane/src/physical/compiler.rs‎

Lines changed: 112 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -67,6 +67,9 @@ pub struct QueryCompilationInput {
6767
/// Planner-selected post-ASAP DAG. The physical compiler must not
6868
/// re-select a summary family from pre-ASAP input.
6969
pub selected_plan_root: Rc<SummaryNode>,
70+
/// Encoded Planner PhysicalCandidate retained before deployment pricing.
71+
/// Installation decodes this candidate; it must not compile another graph.
72+
pub physical_candidate: Option<Vec<u8>>,
7073
/// Legacy catalog source retained for request compatibility. Physical
7174
/// materialization sources are derived from each post-ASAP SummaryAgg;
7275
/// it also remains part of the cost manifest workload identity.
@@ -88,6 +91,46 @@ pub struct QueryCompilationInput {
8891
pub materialization_runtime_policy: RuntimeRulePolicy,
8992
}
9093

94+
impl QueryCompilationInput {
95+
/// Physical planning precedes deployment feasibility and cost selection.
96+
pub(crate) fn retain_physical_candidate(&mut self) -> Result<(), CompileError> {
97+
use asap_physical_operators::physical_planner::{promql_rows, PhysicalCandidate};
98+
let candidate =
99+
promql_rows::compile_fixed_window_rate_aggregation(&self.selected_plan_root)
100+
.or_else(|_| {
101+
promql_rows::compile_current_series_readout(&self.selected_plan_root)
102+
.or_else(|_| {
103+
promql_rows::compile_rate_ranking(&self.selected_plan_root)
104+
.map(|(_, dag)| dag)
105+
})
106+
.map(|query| PhysicalCandidate {
107+
precompute: None,
108+
query,
109+
materialized_outputs: BTreeMap::new(),
110+
})
111+
})
112+
.ok();
113+
self.physical_candidate = candidate
114+
.map(|candidate| candidate.encode())
115+
.transpose()
116+
.map_err(|e| CompileError::Snapshot(e.to_string()))?;
117+
Ok(())
118+
}
119+
120+
fn retained_physical(
121+
&self,
122+
) -> Result<Option<asap_physical_operators::physical_planner::PhysicalCandidate>, CompileError>
123+
{
124+
self.physical_candidate
125+
.as_ref()
126+
.map(|bytes| {
127+
asap_physical_operators::physical_planner::PhysicalCandidate::decode(bytes)
128+
.map_err(|e| CompileError::Snapshot(e.to_string()))
129+
})
130+
.transpose()
131+
}
132+
}
133+
91134
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
92135
#[serde(deny_unknown_fields)]
93136
pub struct WindowRealizationCostQuote {
@@ -917,6 +960,7 @@ impl BackendLocalPlanningInput {
917960
query_id,
918961
query_string: query_string.clone(),
919962
selected_plan_root: post_asap,
963+
physical_candidate: None,
920964
legacy_query_source: Source::TimeSeries {
921965
metric: source_hint,
922966
},
@@ -1084,6 +1128,12 @@ impl BackendLocalPlanningInput {
10841128
}
10851129
planner_candidate_forests.push(forest);
10861130
}
1131+
for query in queries
1132+
.iter_mut()
1133+
.chain(planner_candidate_forests.iter_mut().flatten())
1134+
{
1135+
query.retain_physical_candidate()?;
1136+
}
10871137
// Composable lowering residualizes unsafe leaves individually; retain Planner siblings.
10881138
Ok((
10891139
PhysicalCompilationRequest {
@@ -1571,7 +1621,10 @@ impl DeploymentPlanCompiler {
15711621
for state in &selected {
15721622
if matches!(&state.node.expr, SummaryExpr::SummaryAgg {
15731623
reduction: planner_types::pre_asap::Reduction::Reduce(keys), ..
1574-
} if keys.is_empty()) || asap_physical_operators::physical_planner::promql_rows::compile_fixed_window_rate_aggregation(&query.selected_plan_root).is_ok()
1624+
} if keys.is_empty())
1625+
|| query
1626+
.retained_physical()?
1627+
.is_some_and(|candidate| candidate.precompute.is_some())
15751628
{
15761629
if let Some(sources) = immutable_materialization_sources(&state.node) {
15771630
canonical_nodes.insert(Rc::as_ptr(&state.node) as usize);
@@ -1582,7 +1635,9 @@ impl DeploymentPlanCompiler {
15821635
}
15831636
}
15841637
let cohort_nodes = windows::cohort_nodes(&selected);
1585-
let native_cohort = asap_physical_operators::physical_planner::promql_rows::compile_fixed_window_rate_aggregation(&query.selected_plan_root).is_ok();
1638+
let native_cohort = query
1639+
.retained_physical()?
1640+
.is_some_and(|candidate| candidate.precompute.is_some());
15861641
for (ordinal, selected) in selected.into_iter().enumerate() {
15871642
let mut branch_query = query.clone();
15881643
branch_query.query_lookback_ms = selected
@@ -2162,23 +2217,39 @@ impl DeploymentPlanCompiler {
21622217
} else {
21632218
false
21642219
};
2165-
let native_rate =
2166-
asap_physical_operators::physical_planner::promql_rows::compile_rate_ranking(
2167-
&query.selected_plan_root,
2168-
)
2169-
.ok();
2170-
let native_rate = native_rate.or_else(|| {
2171-
let physical = asap_physical_operators::physical_planner::promql_rows::compile_fixed_window_rate_aggregation(&query.selected_plan_root).ok()?;
2172-
fn heap(node: &Rc<SummaryNode>) -> Option<Rc<SummaryNode>> {
2173-
match &node.expr {
2174-
SummaryExpr::SummaryAgg { family: SummaryFamilyType::Sketch(..) | SummaryFamilyType::ExactAggregate(planner_types::post_asap::ExactKind::Sum, _), .. } => Some(node.clone()),
2175-
SummaryExpr::ValueOperation { child, .. } => heap(child),
2176-
SummaryExpr::SummaryEstimate { summary_input, .. } => heap(summary_input),
2177-
_ => None,
2178-
}
2179-
}
2180-
Some((heap(&query.selected_plan_root)?, physical.query))
2181-
});
2220+
let native_rate = if population_operators[query_index].is_none() {
2221+
query
2222+
.retained_physical()?
2223+
.map(|physical| {
2224+
let program = physical.query;
2225+
let sources = program
2226+
.input_contracts()
2227+
.map(|(id, _)| id)
2228+
.collect::<Vec<_>>();
2229+
let [source] = sources.as_slice() else {
2230+
return Err(CompileError::Snapshot(
2231+
"native candidate requires one bound input".into(),
2232+
));
2233+
};
2234+
let source = executable_dags[query_index]
2235+
.as_ref()
2236+
.and_then(|compiled| {
2237+
u32::try_from(*source).ok().and_then(|id| {
2238+
compiled.node_ids.summary_node(PostAsapNodeId(id))
2239+
})
2240+
})
2241+
.cloned()
2242+
.ok_or_else(|| {
2243+
CompileError::Snapshot(
2244+
"selected physical input has no semantic binding".into(),
2245+
)
2246+
})?;
2247+
Ok((source, program))
2248+
})
2249+
.transpose()?
2250+
} else {
2251+
None
2252+
};
21822253
let mut entry = if let Some((source, program)) = native_rate {
21832254
let native_state_binding = if let SummaryExpr::SummaryAgg {
21842255
family:
@@ -2371,7 +2442,10 @@ impl DeploymentPlanCompiler {
23712442
}
23722443
super::maintained_population::install_native_topk(
23732444
&mut entry,
2374-
&query.selected_plan_root,
2445+
query
2446+
.retained_physical()?
2447+
.as_ref()
2448+
.map(|candidate| &candidate.query),
23752449
)?;
23762450
let catalog_key = QueryPlan::catalog_key(entry.language, &canonical);
23772451
if query_entries.insert(catalog_key, entry).is_some() {
@@ -2414,10 +2488,23 @@ impl DeploymentPlanCompiler {
24142488
query_id: query_id.clone(),
24152489
reason,
24162490
})?;
2417-
if let Ok(physical) = asap_physical_operators::physical_planner::promql_rows::compile_fixed_window_rate_aggregation(&request.queries[query_index].selected_plan_root) {
2418-
let program = physical.precompute.ok_or_else(|| CompileError::Snapshot("fixed-window candidate has no maintenance program".into()))?;
2419-
let sink = PostAsapNodeId(u32::try_from(program.roots()[0]).map_err(|_| CompileError::Snapshot("maintenance sink overflow".into()))?);
2420-
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()))?);
2491+
if let Some(program) = request.queries[query_index]
2492+
.retained_physical()?
2493+
.and_then(|physical| physical.precompute)
2494+
{
2495+
let sink = PostAsapNodeId(
2496+
u32::try_from(program.roots()[0])
2497+
.map_err(|_| CompileError::Snapshot("precompute sink overflow".into()))?,
2498+
);
2499+
installed.native_programs.insert(
2500+
sink,
2501+
serde_json::from_slice(
2502+
&program
2503+
.encode()
2504+
.map_err(|e| CompileError::Snapshot(e.to_string()))?,
2505+
)
2506+
.map_err(|e| CompileError::Snapshot(e.to_string()))?,
2507+
);
24212508
}
24222509
query_plan
24232510
.selected_dags
@@ -5546,6 +5633,7 @@ pub(crate) mod tests {
55465633
query_id: query_id.into(),
55475634
query_string: promql.into(),
55485635
selected_plan_root: post_asap,
5636+
physical_candidate: None,
55495637
legacy_query_source: Source::TimeSeries { metric: "m".into() },
55505638
query_lookback_ms: 60_000,
55515639
group_by_labels: vec![],

‎control_plane/src/physical/maintained_population.rs‎

Lines changed: 5 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -161,7 +161,7 @@ pub(super) fn operator(
161161
/// Planner before this candidate is priced or installed.
162162
pub(super) fn install_native_topk(
163163
entry: &mut asap_types::query_plan::QueryPlanEntry,
164-
selected: &std::rc::Rc<SummaryNode>,
164+
compiled: Option<&asap_physical_operators::physical_planner::CompiledPhysicalDag>,
165165
) -> Result<(), CompileError> {
166166
use asap_types::query_plan::QueryPlanNode;
167167
let Some(QueryPlanNode::Logical {
@@ -178,14 +178,10 @@ pub(super) fn install_native_topk(
178178
if population.grouping.without {
179179
return Ok(());
180180
}
181-
let compiled =
182-
asap_physical_operators::physical_planner::promql_rows::compile_current_series_readout(
183-
selected,
184-
)
185-
.map_err(|error| CompileError::Query {
186-
query_id: entry.query_id.clone(),
187-
reason: error.to_string(),
188-
})?;
181+
let compiled = compiled.ok_or_else(|| CompileError::Query {
182+
query_id: entry.query_id.clone(),
183+
reason: "selected TopK candidate has no retained physical DAG".into(),
184+
})?;
189185
let encoded = compiled.encode().map_err(|error| CompileError::Query {
190186
query_id: entry.query_id.clone(),
191187
reason: error.to_string(),

‎control_plane/src/physical/workload_cost.rs‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -882,6 +882,11 @@ fn enumerate_frontier_candidates(
882882
}
883883
}
884884
}
885+
for candidate in &mut candidates {
886+
for query in &mut candidate.queries {
887+
query.retain_physical_candidate()?;
888+
}
889+
}
885890
Ok(candidates)
886891
}
887892

‎control_plane/tests/native_rate_topk.rs‎

Lines changed: 48 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -276,3 +276,51 @@ fn grouped_rate_has_native_query_and_maintenance_candidates() {
276276
"{errors:#?}"
277277
);
278278
}
279+
280+
// Deployment must install the exact retained Planner graphs, including roots,
281+
// typed inputs and materialization frontiers, rather than compiling replacements.
282+
#[test]
283+
fn selected_native_graphs_survive_deployment_unchanged() {
284+
use asap_physical_operators::physical_planner::PhysicalCandidate;
285+
let (request, environment) = fixture(true).into_physical_compilation_request().unwrap();
286+
let mut checked = 0;
287+
let mut checked_precompute = 0;
288+
for candidate in enumerate_exact_and_materialized_candidates(request).unwrap() {
289+
let expected = candidate
290+
.queries
291+
.iter()
292+
.filter_map(|query| {
293+
query.physical_candidate.as_ref().map(|bytes| {
294+
(
295+
query.query_id.clone(),
296+
PhysicalCandidate::decode(bytes).unwrap(),
297+
)
298+
})
299+
})
300+
.collect::<std::collections::BTreeMap<_, _>>();
301+
let Ok(plan) = DeploymentPlanCompiler.compile_promql(candidate, environment.clone()) else {
302+
continue;
303+
};
304+
for entry in plan.query_plan.entries.values() {
305+
let Some(expected) = expected.get(&entry.query_id) else {
306+
continue;
307+
};
308+
let actual = entry
309+
.physical_dag
310+
.as_ref()
311+
.expect("retained candidate must be installed");
312+
let encoded: Value = serde_json::from_slice(&expected.query.encode().unwrap()).unwrap();
313+
assert_eq!(*actual, encoded);
314+
if let Some(precompute) = &expected.precompute {
315+
let encoded: Value = serde_json::from_slice(&precompute.encode().unwrap()).unwrap();
316+
assert!(plan.precompute_plan.executable_dags[&entry.query_id]
317+
.native_programs
318+
.values()
319+
.any(|actual| actual == &encoded));
320+
checked_precompute += 1;
321+
}
322+
checked += 1;
323+
}
324+
}
325+
assert!(checked > 0 && checked_precompute > 0);
326+
}

‎data_plane/src/precompute_engine/maintenance_runtime.rs‎

Lines changed: 24 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1550,7 +1550,9 @@ fn execute_finite_complete_populations(
15501550
cohort.inputs(),
15511551
window,
15521552
max_bytes,
1553-
)?;
1553+
0,
1554+
)
1555+
.map_err(|e| e.to_string())?;
15541556
let mut output = crate::storage_engines::types::PrecomputedOutput::new(
15551557
window.0,
15561558
window.1,
@@ -4556,6 +4558,27 @@ pub(crate) fn execute_revision_outputs(
45564558
{
45574559
continue;
45584560
}
4561+
if let Some(program) = installed.native_program(*sink)? {
4562+
let batch = super::native_precompute::execute(
4563+
installed, &program, &selected, window, limit, revision,
4564+
)?;
4565+
let payload =
4566+
asap_physical_operators::stored_state::native::encode_batch(&batch)?;
4567+
if payload.len() > limit {
4568+
return Err(asap_physical_operators::Error::MemoryLimit.into());
4569+
}
4570+
result.get_mut(target).unwrap().push(RevisionRecord {
4571+
reference: plan
4572+
.installed_precompute_plan
4573+
.stored_output_reference(*target)
4574+
.ok_or("native revision output has no installed binding")?,
4575+
group: BTreeMap::new(),
4576+
start_ms: window.0,
4577+
end_ms: window.1,
4578+
payload,
4579+
});
4580+
continue;
4581+
}
45594582
let (_, mut key) = prepare_frozen_maintenance_sink(
45604583
installed,
45614584
&plan.precompute_plan.materializations,

0 commit comments

Comments
 (0)