Skip to content

Commit ee0ee5a

Browse files
zzylolclaude
andcommitted
refactor(runtime): remove the PromQL fallback evaluator
ASAPPlanner generates candidates and computes their cost, or says why a query has none. Executing what the native operators cannot run belongs to the deployment runtime (ASAPQuery backend), not to the Planner. Remove unified_physical_planner::promql_fallback and its tests, and the fallback paths in compile: raw-series input slots, the fallback lowering and the below-fallback binding check. A node without a native lowering now fails compilation with Error::Invalid. Reject Planner node IDs above u32::MAX, which could collide with helper IDs, and error instead of debug-asserting when a node needs more than 2^16 helpers. Document that forces_query_time also forces an unaligned per-series PromQL BinaryOp to query time. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01W7qG9aFyPij5uWsyAJCxDW
1 parent 8b172df commit ee0ee5a

6 files changed

Lines changed: 71 additions & 2637 deletions

File tree

‎crates/asap-physical-operators/README.md‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -54,6 +54,9 @@ deployment source, including asynchronous I/O; computation operators remain in
5454
the library. The public `planner` export identifies the exact Planner types used
5555
by the crate. The physical compiler currently supports a subset of those types and
5656
operations; it does not interpret an unknown node as external fallback.
57+
`unified_physical_planner::compile` does the same for a `PhysicalASAPDAG`: a
58+
node without a native lowering fails compilation, and the deployment runtime
59+
must execute it.
5760

5861
Plain values preserve Planner scalar/collection types and nullability. Numeric
5962
arithmetic uses matching Int64 or Float64 inputs; integer overflow is an error.

‎crates/asap-physical-operators/src/unified_physical_planner/mod.rs‎

Lines changed: 19 additions & 110 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,7 @@
11
//! Compile logical computation to native operators with typed external inputs.
22
//! Compilation needs no readers; deployment resolves inputs after selection.
3+
//! A node without a native lowering fails compilation; the deployment runtime
4+
//! must execute it.
35
use crate::operators::ReadoutQuery;
46
use crate::summary_kernels::exact::ExactReadout;
57
use crate::{
@@ -34,7 +36,6 @@ fn invalid(message: impl Into<String>) -> Error {
3436
pub type Source<'a> = Box<dyn PhysicalOperator<Batch, SchemaRef> + 'a>;
3537

3638
pub mod precompute;
37-
pub mod promql_fallback;
3839
pub mod promql_rows;
3940
pub mod promql_values;
4041

@@ -118,9 +119,11 @@ thread_local! {
118119
/// Planner ID range, so every boundary choice yields a subgraph of the same
119120
/// lowering and candidate cuts need not renumber operators. A node lowering to
120121
/// several helpers takes consecutive indices below its base.
121-
fn helper_id(node: NodeId, index: u64) -> NodeId {
122-
debug_assert!(node <= u64::from(u32::MAX) && index < 1 << 16);
123-
u64::MAX - (node << 16) - index
122+
fn helper_id(node: NodeId, index: u64) -> Result<NodeId, Error> {
123+
if index >= 1 << 16 {
124+
return Err(invalid(format!("node {node} needs more than 2^16 helpers")));
125+
}
126+
Ok(u64::MAX - (node << 16) - index)
124127
}
125128

126129
fn compile_internal(
@@ -129,6 +132,10 @@ fn compile_internal(
129132
roots: &[NodeId],
130133
) -> Result<CompiledPhysicalDAG, Error> {
131134
preflight_depth(dag)?;
135+
// Helper IDs lie above the u32 range; a larger Planner ID could collide.
136+
if let Some(node) = dag.nodes.iter().find(|n| n.id as u64 > u64::from(u32::MAX)) {
137+
return Err(invalid(format!("Planner node ID {} exceeds u32", node.id)));
138+
}
132139
let restored = logical::restore(dag)?;
133140
dag.validate().map_err(|e| invalid(e.to_string()))?;
134141
let nodes = dag
@@ -156,35 +163,7 @@ fn compile_internal(
156163
.or_default()
157164
.push(edge.producer as u64);
158165
}
159-
let mut fallback = BTreeMap::new();
160-
for (&id, root) in &restored {
161-
let raw_summary_input = matches!(root.non_asap(), Some(NonASAPOp::TimeRange { .. }))
162-
&& dag.edges.iter().any(|e| {
163-
e.producer as u64 == id
164-
&& matches!(
165-
nodes[&(e.consumer as u64)].payload,
166-
Payload::ASAP(ASAPOp::SummaryAgg { .. })
167-
)
168-
});
169-
if !root.contains_asap() && !raw_summary_input {
170-
if let Ok(lowered) = promql_fallback::lower(root) {
171-
fallback.insert(id, lowered);
172-
}
173-
}
174-
}
175-
let known = |id: &NodeId| {
176-
nodes.contains_key(id)
177-
|| promql_fallback::raw_series_owner(*id).is_some_and(|owner| {
178-
matches!(
179-
nodes.get(&owner),
180-
Some(PhysicalASAPDAGNode {
181-
payload: Payload::NonASAP(_),
182-
..
183-
})
184-
)
185-
})
186-
};
187-
if !sources.keys().all(known) {
166+
if !sources.keys().all(|id| nodes.contains_key(id)) {
188167
return Err(invalid("source binding names an unknown node"));
189168
}
190169
let mut ordered = Vec::new();
@@ -202,30 +181,16 @@ fn compile_internal(
202181
return Err(invalid(format!("missing root {id}")));
203182
}
204183
pending.push((id, true));
205-
if !sources.contains_key(&id) && !fallback.contains_key(&id) {
184+
if !sources.contains_key(&id) {
206185
for &input in dependencies.get(&id).into_iter().flatten() {
207186
pending.push((input, false));
208187
}
209188
}
210189
}
211-
// The PromQL fallback reads raw series slots, so a binding below a node it
212-
// evaluates is never read. (A binding below another bound input is just
213-
// superseded by it, as in `cut`.)
214-
let mut below_fallback = BTreeSet::new();
215-
let mut stack = ordered
216-
.iter()
217-
.filter(|id| !sources.contains_key(id) && fallback.contains_key(id))
218-
.flat_map(|id| dependencies.get(id).into_iter().flatten().copied())
219-
.collect::<Vec<_>>();
220-
while let Some(id) = stack.pop() {
221-
if below_fallback.insert(id) {
222-
stack.extend(dependencies.get(&id).into_iter().flatten().copied());
223-
}
224-
}
225190
let mut dag = CompiledPhysicalDAG::new(roots.to_vec());
226191
for id in ordered {
227192
let node = nodes[&id];
228-
let mut auxiliary = helper_id(id, 0);
193+
let auxiliary = helper_id(id, 0)?;
229194
let output = Arc::new(node.output_schema.clone());
230195
crate::values::validate_schema(&output)?;
231196
if let Some(source) = sources.remove(&id) {
@@ -255,56 +220,6 @@ fn compile_internal(
255220
inputs = vec![auxiliary];
256221
schemas.truncate(1);
257222
}
258-
if let Some(promql_fallback::Lowering {
259-
selectors,
260-
mut steps,
261-
}) = fallback.remove(&id)
262-
{
263-
let mut slots = Vec::new();
264-
for (i, (_, schema)) in selectors.iter().enumerate() {
265-
let slot = promql_fallback::raw_series_input(id, i);
266-
match sources.remove(&slot) {
267-
Some(contract) if &contract.schema == schema => {
268-
dag.add_input(slot, contract)?
269-
}
270-
Some(_) => {
271-
return Err(invalid(format!(
272-
"node {id}: raw series input {slot} differs from the selector schema"
273-
)))
274-
}
275-
None => {
276-
return Err(invalid(format!(
277-
"node {id}: PromQL fallback requires raw series input {slot}"
278-
)))
279-
}
280-
}
281-
slots.push(slot);
282-
}
283-
let (last, last_inputs) = steps
284-
.pop()
285-
.ok_or_else(|| invalid("empty PromQL lowering"))?;
286-
let mut ids = Vec::new();
287-
let resolve = |inputs: Vec<promql_fallback::Input>, ids: &[NodeId]| {
288-
inputs
289-
.into_iter()
290-
.map(|input| match input {
291-
promql_fallback::Input::Raw(i) => slots[i],
292-
promql_fallback::Input::Step(i) => ids[i],
293-
})
294-
.collect::<Vec<_>>()
295-
};
296-
for (operator, inputs) in steps {
297-
dag.add(auxiliary, resolve(inputs, &ids), operator)?;
298-
ids.push(auxiliary);
299-
auxiliary -= 1;
300-
}
301-
dag.add(
302-
id,
303-
resolve(last_inputs, &ids),
304-
last.with_output_schema(output)?,
305-
)?;
306-
continue;
307-
}
308223
if let Payload::ASAP(ASAPOp::MaintainPopulation { population, .. }) = &node.payload {
309224
use planner_types::post_asap::maintained_population::PopulationInput;
310225
let PopulationInput::CurrentSeries(spec) = &population.input else {
@@ -366,10 +281,10 @@ fn compile_internal(
366281
row_values::population_aggregate(&input, &spec.grouping, evaluation)?;
367282
let last = chain.pop().expect("nonempty chain");
368283
let mut inputs = inputs;
369-
for operator in chain {
370-
dag.add(auxiliary, inputs, operator)?;
371-
inputs = vec![auxiliary];
372-
auxiliary -= 1;
284+
for (index, operator) in chain.into_iter().enumerate() {
285+
let helper = helper_id(id, index as u64)?;
286+
dag.add(helper, inputs, operator)?;
287+
inputs = vec![helper];
373288
}
374289
dag.add(id, inputs, last.with_output_schema(output)?)?;
375290
continue;
@@ -530,7 +445,6 @@ fn compile_internal(
530445
} else {
531446
dag.add(id, vec![auxiliary], project)?;
532447
}
533-
auxiliary -= 1;
534448
continue;
535449
}
536450
}
@@ -570,11 +484,6 @@ fn compile_internal(
570484
}
571485
}
572486
}
573-
if let Some(id) = sources.keys().find(|id| below_fallback.contains(id)) {
574-
return Err(invalid(format!(
575-
"input binding {id} is below a PromQL fallback node, which never reads it"
576-
)));
577-
}
578487
dag.validate()?;
579488
Ok(dag)
580489
}
@@ -942,7 +851,7 @@ fn bind_operation(node: &PhysicalASAPDAGNode, inputs: &[SchemaRef]) -> Result<Op
942851
)
943852
}
944853
_ => Err(invalid(
945-
"physical operation has no native binding; no fallback is installed",
854+
"physical operation has no native binding; the deployment runtime must execute it",
946855
)),
947856
}
948857
}

0 commit comments

Comments
 (0)