From 6832db87d7a6f17bbb39bb24132e95180c9c014b Mon Sep 17 00:00:00 2001 From: zz_y Date: Thu, 10 Sep 2026 10:35:40 -0600 Subject: [PATCH] Persist executable DAG bindings in installed plans --- control_plane/src/physical/compiler.rs | 172 +++++++++++++- .../src/physical/executable_binding.rs | 220 +++++++++++++++++- control_plane/src/query_plan.rs | 151 +++++++++++- .../drivers/ingest/prometheus_remote_write.rs | 1 + data_plane/src/drivers/query/servers/http.rs | 1 + data_plane/src/main.rs | 1 + .../src/precompute_engine/subdag_scheduler.rs | 2 + .../asap_query_engine/post_asap_readout.rs | 3 +- .../types/hot_reload_config.rs | 3 +- 9 files changed, 530 insertions(+), 24 deletions(-) diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs index 72ef55df..421f8201 100644 --- a/control_plane/src/physical/compiler.rs +++ b/control_plane/src/physical/compiler.rs @@ -317,6 +317,11 @@ pub struct PrecomputePlan { pub schemas: Vec, pub producers: Vec, pub materializations: Vec, + /// Planner semantic DAGs and backend-owned placement for this generation. + /// Empty only for legacy/config-only construction paths. + #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] + pub executable_dags: + BTreeMap, } #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] @@ -547,6 +552,7 @@ impl PrecomputePlan { schemas, producers, materializations, + executable_dags: BTreeMap::new(), }; plan.validate()?; Ok(plan) @@ -604,6 +610,16 @@ impl PrecomputePlan { if !valid_ingest { return Err(PrecomputePlanError::UnsupportedIngestEndpoint); } + for (query_id, installed) in &self.executable_dags { + if query_id != &installed.document.query_id { + return Err(PrecomputePlanError::CatalogContract( + "post-ASAP DAG map key differs from document query ID".into(), + )); + } + installed + .validate() + .map_err(PrecomputePlanError::CatalogContract)?; + } let mut materializations = BTreeSet::new(); for materialization in &self.materializations { // HLL is supported as an ingested sketch envelope, not as a raw @@ -2094,6 +2110,16 @@ impl PhysicalCompiler { reason: "backend-local target cannot declare Collector producers".into(), }); } + let query_ids = request + .queries + .iter() + .map(|query| query.query_id.as_str()) + .collect::>(); + if query_ids.len() != request.queries.len() { + return Err(CompileError::Snapshot( + "planning query IDs must be unique within a plan generation".into(), + )); + } if let Some(workload) = &request.query_workload { let entries = workload.entries().collect::>(); @@ -2136,6 +2162,8 @@ impl PhysicalCompiler { let mut executable_dags = vec![None::; request.queries.len()]; let mut node_bindings = BTreeMap::<(usize, PostAsapNodeId), asap_types::PolicyFingerprint>::new(); + let mut query_node_bindings = + BTreeMap::<(usize, PostAsapNodeId), crate::query_plan::QueryNodeId>::new(); let consumers = materialization_consumers( &request.queries, environment.target, @@ -2621,22 +2649,38 @@ impl PhysicalCompiler { cumulative_readout: true, }; let mut entry = if request.hybrid_execution { - QueryPlanEntry::compile_bound_composable( + QueryPlanEntry::compile_bound_composable_mapped( query.query_id.clone(), canonical.clone(), &query.post_asap, instant, FallbackPolicy::ExactBackend, binding, + |node, query_node| { + if let Some(post_asap_node) = executable_dags[query_index] + .as_ref() + .and_then(|compiled| compiled.node_ids.node_id(node)) + { + query_node_bindings.insert((query_index, post_asap_node), query_node); + } + }, ) } else { - QueryPlanEntry::compile_bound( + QueryPlanEntry::compile_bound_mapped( query.query_id.clone(), canonical.clone(), &query.post_asap, instant, FallbackPolicy::ExactBackend, binding, + |node, query_node| { + if let Some(post_asap_node) = executable_dags[query_index] + .as_ref() + .and_then(|compiled| compiled.node_ids.node_id(node)) + { + query_node_bindings.insert((query_index, post_asap_node), query_node); + } + }, ) }?; if request.hybrid_execution { @@ -2662,6 +2706,60 @@ impl PhysicalCompiler { clickhouse_context: None, entries: query_entries, }; + for (query_index, compiled) in executable_dags.iter().enumerate() { + let Some(compiled) = compiled else { continue }; + let query_id = request.queries[query_index].query_id.clone(); + let mut placements = BTreeMap::new(); + let mut precompute_sinks = Vec::new(); + for node in &compiled.dag.nodes { + let placement = + if let Some(definition) = node_bindings.get(&(query_index, node.id)).copied() { + precompute_sinks.push(node.id); + crate::physical::executable_binding::BackendNodeBinding::Materialization { + summary_definition: definition.into(), + } + } else if node.output_state.timing + == planner_types::post_asap::ExecutionTiming::MaintenanceTime + { + super::executable_binding::BackendNodeBinding::MaintenanceInput + } else { + match query_node_bindings.get(&(query_index, node.id)).copied() { + Some(query_node) => { + super::executable_binding::BackendNodeBinding::Query { query_node } + } + None => super::executable_binding::BackendNodeBinding::QueryInput, + } + }; + placements.insert(node.id, placement); + } + precompute_sinks.sort(); + let installed = crate::physical::executable_binding::InstalledPostAsapDag { + document: super::executable_binding::PostAsapDagDocument::from_executable( + query_id.clone(), + &compiled.dag, + ) + .map_err(|reason| CompileError::Query { + query_id: query_id.clone(), + reason, + })?, + binding: super::executable_binding::BackendExecutableBinding { + nodes: placements, + query_sink: compiled.dag.root, + query_plan_sink: query_plan + .entries + .values() + .find(|entry| entry.query_id == query_id) + .expect("compiled query entry exists") + .root, + precompute_sinks, + }, + }; + installed.validate().map_err(|reason| CompileError::Query { + query_id: query_id.clone(), + reason, + })?; + precompute_plan.executable_dags.insert(query_id, installed); + } for materialization in &mut precompute_plan.materializations { let fingerprint = materialization.policy_fingerprint(); let max_lookback_ms = query_plan @@ -2696,6 +2794,22 @@ impl PhysicalCompiler { .unwrap_or(DEFAULT_RETAINED_SUMMARY_MEMORY_BUDGET_BYTES), )?; query_plan.validate(&materialization_fingerprints)?; + for (query_id, installed) in &precompute_plan.executable_dags { + let entry = query_plan + .entries + .values() + .find(|entry| &entry.query_id == query_id) + .ok_or_else(|| CompileError::Query { + query_id: query_id.clone(), + reason: "installed post-ASAP DAG has no query-plan entry".into(), + })?; + installed + .validate_query_plan(entry) + .map_err(|reason| CompileError::Query { + query_id: query_id.clone(), + reason, + })?; + } let summary_catalog = super::summary_catalog::SummaryCatalog::from_materializations( envelope.plan_id, envelope.plan_version, @@ -4195,7 +4309,9 @@ mod tests { #[test] fn hybrid_weighted_topk_installs_only_candidates_and_delegates_filtered_exact_values() { - use crate::query_plan::{logical::LogicalOperator, QueryPlanNode}; + use crate::query_plan::{ + logical::LogicalOperator, ExternalExactInput, ExternalExactOutput, QueryPlanNode, + }; let query = "topk(2, sum by (job) (rate(m[1m])))"; let evidence = TopKMembershipEvidence { selected_lower_bound: 101.0, @@ -4223,11 +4339,15 @@ mod tests { }; assert!(matches!( &entry.nodes[&inputs[1]], - QueryPlanNode::Logical { - operator: LogicalOperator::CandidateExactSubquery { query, item_label }, + QueryPlanNode::ExternalExact { + request, inputs: exact_inputs, - } if query == "sum by (job) (rate(m[1m]))" - && item_label == "job" + } if request.language == crate::query_plan::QueryLanguage::PromQl + && request.expression == "sum by (job) (rate(m[1m]))" + && request.output == ExternalExactOutput::InstantVector + && request.input_contracts == vec![ExternalExactInput::CandidateMembership { + item_label: "job".into(), + }] && exact_inputs == &vec![inputs[0]] )); assert!(entry.nodes.values().all(|node| !matches!( @@ -4238,6 +4358,27 @@ mod tests { .. } ))); + let installed = plan + .precompute_plan + .executable_dags + .get(&entry.query_id) + .expect("compiled query retains its Planner DAG and backend placement"); + installed.validate().expect("typed DAG document"); + installed + .validate_query_plan(entry) + .expect("query node bindings"); + assert_eq!(installed.binding.query_plan_sink, entry.root); + assert!(installed.binding.nodes.values().any(|placement| matches!( + placement, + crate::physical::executable_binding::BackendNodeBinding::Materialization { .. } + ))); + let encoded = serde_json::to_value(installed).unwrap(); + let decoded: crate::physical::executable_binding::InstalledPostAsapDag = + serde_json::from_value(encoded).unwrap(); + assert_eq!(&decoded, installed); + decoded + .validate() + .expect("round-tripped typed DAG document"); } #[test] @@ -4413,6 +4554,23 @@ mod tests { request_with_evidence(query_id, promql, None).expect("post-ASAP selection") } + #[test] + fn duplicate_query_ids_cannot_overwrite_installed_dag_documents() { + let mut workload = request("shared-id", "max_over_time(a[1m])"); + let mut second = workload.queries[0].clone(); + second.query_string = "max_over_time(b[1m])".into(); + workload.queries.push(second); + + let error = PhysicalCompiler + .compile(workload, environment(10_000)) + .unwrap_err(); + assert!(matches!( + error, + CompileError::Snapshot(reason) + if reason == "planning query IDs must be unique within a plan generation" + )); + } + #[test] fn metricsql_compilation_publishes_a_language_tagged_query_entry() { let query = "default_rollup(m[1m])"; diff --git a/control_plane/src/physical/executable_binding.rs b/control_plane/src/physical/executable_binding.rs index c49aba35..8959790c 100644 --- a/control_plane/src/physical/executable_binding.rs +++ b/control_plane/src/physical/executable_binding.rs @@ -1,10 +1,215 @@ use std::collections::{BTreeMap, BTreeSet}; use asap_types::sds::SummaryDefinitionId; -use planner_types::post_asap::{ExecutableDag, ExecutionTiming, PostAsapNodeId}; +use planner_types::post_asap::{ + EdgeRole, ExecutableDag, ExecutableDagEdge, ExecutableDagNode, ExecutableOperator, + ExecutionDataState, ExecutionTiming, GroupingEdgeCompatibility, PostAsapNodeId, + WindowEdgeCompatibility, +}; use serde::{Deserialize, Serialize}; -use crate::query_plan::QueryNodeId; +use crate::query_plan::{QueryNodeId, QueryPlanEntry}; + +pub const POST_ASAP_DAG_DOCUMENT_SCHEMA_VERSION: u32 = 1; + +/// Versioned, language-neutral Planner DAG persisted with an installed plan. +/// Plan lifecycle belongs to the enclosing `PrecomputePlan`; this document +/// carries semantic identity only and does not duplicate its envelope. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct PostAsapDagDocument { + pub schema_version: u32, + pub query_id: String, + pub nodes: Vec, + pub edges: Vec, + pub root: PostAsapNodeId, +} + +/// Send/Sync wire form of one Planner node. Planner's in-memory payload may +/// contain `Rc`; the canonical JSON payload preserves its tagged type without +/// leaking that process-local ownership choice into installed runtime state. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct PostAsapDagNodeDocument { + pub id: PostAsapNodeId, + pub operator: ExecutableOperator, + pub payload: serde_json::Value, + pub output_state: ExecutionDataState, + pub output_schema: serde_json::Value, + pub guarantee: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct PostAsapDagEdgeDocument { + pub producer: PostAsapNodeId, + pub consumer: PostAsapNodeId, + pub role: EdgeRole, + pub intermediate_schema: serde_json::Value, + pub data_state: ExecutionDataState, + pub grouping: GroupingEdgeCompatibility, + pub window: WindowEdgeCompatibility, +} + +impl PostAsapDagDocument { + pub fn from_executable(query_id: String, dag: &ExecutableDag) -> Result { + let nodes = dag + .nodes + .iter() + .map(|node| { + Ok(PostAsapDagNodeDocument { + id: node.id, + operator: node.operator, + payload: serde_json::to_value(&node.payload).map_err(|e| e.to_string())?, + output_state: node.output_state, + output_schema: serde_json::to_value(&node.output_schema) + .map_err(|e| e.to_string())?, + guarantee: node + .guarantee + .as_ref() + .map(serde_json::to_value) + .transpose() + .map_err(|e| e.to_string())?, + }) + }) + .collect::, String>>()?; + let edges = dag + .edges + .iter() + .map(|edge| { + Ok(PostAsapDagEdgeDocument { + producer: edge.producer, + consumer: edge.consumer, + role: edge.role, + intermediate_schema: serde_json::to_value(&edge.intermediate_schema) + .map_err(|e| e.to_string())?, + data_state: edge.data_state, + grouping: edge.grouping, + window: edge.window, + }) + }) + .collect::, String>>()?; + Ok(Self { + schema_version: POST_ASAP_DAG_DOCUMENT_SCHEMA_VERSION, + query_id, + nodes, + edges, + root: dag.root, + }) + } + + pub fn decode(&self) -> Result { + let node_ids = self + .nodes + .iter() + .map(|node| node.id) + .collect::>(); + if node_ids.len() != self.nodes.len() { + return Err("duplicate post-ASAP DAG node ID".into()); + } + let edge_ids = self + .edges + .iter() + .map(|edge| { + ( + edge.producer, + edge.consumer, + serde_json::to_string(&edge.role).expect("EdgeRole is serializable"), + ) + }) + .collect::>(); + if edge_ids.len() != self.edges.len() { + return Err("duplicate post-ASAP DAG edge".into()); + } + let nodes = self + .nodes + .iter() + .map(|node| { + let payload: planner_types::post_asap::ExecutableOperatorPayload = + serde_json::from_value(node.payload.clone()).map_err(|e| e.to_string())?; + if payload.operator() != node.operator { + return Err(format!( + "post-ASAP node {} operator disagrees with payload", + node.id.0 + )); + } + Ok(ExecutableDagNode { + id: node.id, + operator: node.operator, + payload, + output_state: node.output_state, + output_schema: serde_json::from_value(node.output_schema.clone()) + .map_err(|e| e.to_string())?, + guarantee: node + .guarantee + .clone() + .map(serde_json::from_value) + .transpose() + .map_err(|e| e.to_string())?, + }) + }) + .collect::, String>>()?; + let edges = self + .edges + .iter() + .map(|edge| { + Ok(ExecutableDagEdge { + producer: edge.producer, + consumer: edge.consumer, + role: edge.role, + intermediate_schema: serde_json::from_value(edge.intermediate_schema.clone()) + .map_err(|e| e.to_string())?, + data_state: edge.data_state, + grouping: edge.grouping, + window: edge.window, + }) + }) + .collect::, String>>()?; + Ok(ExecutableDag { + nodes, + edges, + root: self.root, + }) + } +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct InstalledPostAsapDag { + pub document: PostAsapDagDocument, + pub binding: BackendExecutableBinding, +} + +impl InstalledPostAsapDag { + pub fn validate(&self) -> Result<(), String> { + if self.document.schema_version != POST_ASAP_DAG_DOCUMENT_SCHEMA_VERSION + || self.document.query_id.trim().is_empty() + { + return Err("invalid post-ASAP DAG document identity/version".into()); + } + self.binding.validate(&self.document.decode()?) + } + + pub fn validate_query_plan(&self, query: &QueryPlanEntry) -> Result<(), String> { + if query.query_id != self.document.query_id { + return Err("post-ASAP document and query plan identity disagree".into()); + } + if query.root != self.binding.query_plan_sink { + return Err("backend query-plan sink disagrees with installed query root".into()); + } + for placement in self.binding.nodes.values() { + if let BackendNodeBinding::Query { query_node } = placement { + if !query.nodes.contains_key(query_node) { + return Err(format!( + "backend binding references missing query node {}", + query_node.0 + )); + } + } + } + Ok(()) + } +} /// Backend-owned physical identities and sink placement for one Planner /// semantic DAG. Planner node IDs remain stable semantic references; the @@ -14,6 +219,7 @@ use crate::query_plan::QueryNodeId; pub struct BackendExecutableBinding { pub nodes: BTreeMap, pub query_sink: PostAsapNodeId, + pub query_plan_sink: QueryNodeId, pub precompute_sinks: Vec, } @@ -23,6 +229,9 @@ pub enum BackendNodeBinding { Query { query_node: QueryNodeId, }, + /// Semantic read node absorbed into a larger installed query node, such + /// as an external exact subtree boundary. + QueryInput, MaintenanceInput, Materialization { summary_definition: SummaryDefinitionId, @@ -49,6 +258,7 @@ impl BackendExecutableBinding { for node in &dag.nodes { match (node.output_state.timing, self.node(node.id).unwrap()) { (ExecutionTiming::ReadTime, BackendNodeBinding::Query { .. }) + | (ExecutionTiming::ReadTime, BackendNodeBinding::QueryInput) | (ExecutionTiming::MaintenanceTime, BackendNodeBinding::MaintenanceInput) | (ExecutionTiming::MaintenanceTime, BackendNodeBinding::Materialization { .. }) => { } @@ -60,12 +270,6 @@ impl BackendExecutableBinding { } } } - if !matches!( - self.node(self.query_sink), - Some(BackendNodeBinding::Query { .. }) - ) { - return Err("backend query sink is not query-placed".into()); - } for sink in &self.precompute_sinks { if !matches!( self.node(*sink), diff --git a/control_plane/src/query_plan.rs b/control_plane/src/query_plan.rs index ac546bd3..8d3a8192 100644 --- a/control_plane/src/query_plan.rs +++ b/control_plane/src/query_plan.rs @@ -303,18 +303,45 @@ impl QueryPlanEntry { } pub fn compile_bound( + query_id: String, + canonical_query: String, + root: &Rc, + instant: InstantExecution, + fallback: FallbackPolicy, + bind: F, + ) -> Result + where + F: FnMut( + &Rc, + &SummaryFamilyType, + ) -> Result, + { + Self::compile_bound_mapped( + query_id, + canonical_query, + root, + instant, + fallback, + bind, + |_, _| {}, + ) + } + + pub fn compile_bound_mapped( query_id: String, canonical_query: String, root: &Rc, instant: InstantExecution, fallback: FallbackPolicy, mut bind: F, + mut lowered: G, ) -> Result where F: FnMut( &Rc, &SummaryFamilyType, ) -> Result, + G: FnMut(&Rc, QueryNodeId), { let mut compiler = DagCompiler { next_id: 0, @@ -323,6 +350,7 @@ impl QueryPlanEntry { bind: &mut bind, logical_source: None, preserve_relational: false, + lowered: Some(&mut lowered), }; let root = compiler.lower(root)?; Ok(Self { @@ -359,6 +387,7 @@ impl QueryPlanEntry { bind: &mut bind, logical_source: None, preserve_relational: true, + lowered: None, }; let root = compiler.lower(root)?; Ok(Self { @@ -376,18 +405,49 @@ impl QueryPlanEntry { /// Compile selected summary nodes and verified native residuals into one DAG. /// This is a distinct physical alternative; native execution remains available. pub fn compile_bound_composable( + query_id: String, + canonical_query: String, + root: &Rc, + instant: InstantExecution, + fallback: FallbackPolicy, + bind: F, + ) -> Result + where + F: FnMut( + &Rc, + &SummaryFamilyType, + ) -> Result, + { + Self::compile_bound_composable_mapped( + query_id, + canonical_query, + root, + instant, + fallback, + bind, + |_, _| {}, + ) + } + + /// Compile a composable query while exposing the stable mapping from + /// Planner semantic nodes to installed query nodes. The control-plane + /// physical compiler uses this to persist backend placement without + /// relying on pointer values or reconstructing query shape later. + pub fn compile_bound_composable_mapped( query_id: String, canonical_query: String, root: &Rc, instant: InstantExecution, fallback: FallbackPolicy, mut bind: F, + mut lowered: G, ) -> Result where F: FnMut( &Rc, &SummaryFamilyType, ) -> Result, + G: FnMut(&Rc, QueryNodeId), { let mut compiler = DagCompiler { next_id: 0, @@ -396,6 +456,7 @@ impl QueryPlanEntry { bind: &mut bind, logical_source: Some(canonical_query.clone()), preserve_relational: false, + lowered: Some(&mut lowered), }; let root = compiler.lower(root)?; let mut entry = Self { @@ -427,6 +488,25 @@ impl QueryPlanEntry { if matches!(node, QueryPlanNode::Scalar { value } if !value.is_finite()) { return Err(QueryPlanError::Invalid("non-finite scalar constant".into())); } + if let QueryPlanNode::ExternalExact { request, inputs } = node { + if request.expression.trim().is_empty() { + return Err(QueryPlanError::Invalid( + "external exact expression must not be empty".into(), + )); + } + if request.input_contracts.len() != inputs.len() { + return Err(QueryPlanError::Invalid( + "external exact input contracts must match DAG inputs".into(), + )); + } + if request.input_contracts.iter().any(|contract| { + matches!(contract, ExternalExactInput::CandidateMembership { item_label } if item_label.is_empty()) + }) { + return Err(QueryPlanError::Invalid( + "external exact candidate item label must not be empty".into(), + )); + } + } if let QueryPlanNode::CandidateTopK { k, completeness, .. } = node @@ -521,6 +601,38 @@ pub enum PhysicalGrouping { Reduce(Vec), } +/// Result shape promised by an external exact engine. The backend uses this +/// contract to type-check downstream DAG nodes without depending on an +/// engine-specific response envelope. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] +pub enum ExternalExactOutput { + Scalar, + InstantVector, + RangeVector, + Relation { schema: serde_json::Value }, +} + +/// How an ordinary DAG input constrains an external exact evaluation. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)] +pub enum ExternalExactInput { + CandidateMembership { item_label: String }, +} + +/// Language-neutral request contract for an exact subtree. Evaluation time is +/// inherited from the containing QueryPlanEntry, avoiding a second time-range +/// envelope that could drift from the installed query plan. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct ExternalExactRequest { + pub language: QueryLanguage, + pub expression: String, + pub output: ExternalExactOutput, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub input_contracts: Vec, +} + #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] #[serde(tag = "op", rename_all = "snake_case", deny_unknown_fields)] pub enum QueryPlanNode { @@ -570,6 +682,12 @@ pub enum QueryPlanNode { grouping: logical::Grouping, completeness: CandidateCompleteness, }, + /// An exact subtree evaluated outside ASAP. Its results enter the query DAG + /// like any other node output and may depend on summary-produced inputs. + ExternalExact { + request: ExternalExactRequest, + inputs: Vec, + }, ExactFallback { reason: String, }, @@ -586,7 +704,9 @@ impl QueryPlanNode { | Self::Relational { input, .. } | Self::SummaryEstimate { input, .. } | Self::ExactReadout { input, .. } => std::slice::from_ref(input), - Self::SummaryMerge { inputs } | Self::Logical { inputs, .. } => inputs, + Self::SummaryMerge { inputs } + | Self::Logical { inputs, .. } + | Self::ExternalExact { inputs, .. } => inputs, Self::CandidateTopK { inputs, .. } => inputs, } } @@ -649,6 +769,7 @@ struct DagCompiler<'a, F> { bind: &'a mut F, logical_source: Option, preserve_relational: bool, + lowered: Option<&'a mut dyn FnMut(&Rc, QueryNodeId)>, } impl DagCompiler<'_, F> @@ -677,7 +798,9 @@ where } for (local, mut physical) in nodes { match &mut physical { - QueryPlanNode::Logical { inputs, .. } | QueryPlanNode::SummaryMerge { inputs } => { + QueryPlanNode::Logical { inputs, .. } + | QueryPlanNode::SummaryMerge { inputs } + | QueryPlanNode::ExternalExact { inputs, .. } => { for input in inputs { *input = remap[input]; } @@ -717,6 +840,9 @@ where // second runtime readout node. let child_id = self.lower(child)?; self.seen.insert(identity, child_id); + if let Some(lowered) = &mut self.lowered { + lowered(node, child_id); + } return Ok(child_id); } let id = QueryNodeId(self.next_id); @@ -738,7 +864,11 @@ where _ => None, }; if let Some((root, nodes)) = residual { - return self.graft(id, root, nodes); + let id = self.graft(id, root, nodes)?; + if let Some(lowered) = &mut self.lowered { + lowered(node, id); + } + return Ok(id); } let physical = match &node.expr { @@ -964,10 +1094,14 @@ where self.next_id += 1; self.nodes.insert( value_id, - QueryPlanNode::Logical { - operator: logical::LogicalOperator::CandidateExactSubquery { - query: aggregate.expr.to_string(), - item_label, + QueryPlanNode::ExternalExact { + request: ExternalExactRequest { + language: QueryLanguage::PromQl, + expression: aggregate.expr.to_string(), + output: ExternalExactOutput::InstantVector, + input_contracts: vec![ExternalExactInput::CandidateMembership { + item_label, + }], }, inputs: vec![candidate_input], }, @@ -1182,6 +1316,9 @@ where }, }; self.nodes.insert(id, physical); + if let Some(lowered) = &mut self.lowered { + lowered(node, id); + } Ok(id) } } diff --git a/data_plane/src/drivers/ingest/prometheus_remote_write.rs b/data_plane/src/drivers/ingest/prometheus_remote_write.rs index e58a94d7..bf8473e6 100644 --- a/data_plane/src/drivers/ingest/prometheus_remote_write.rs +++ b/data_plane/src/drivers/ingest/prometheus_remote_write.rs @@ -789,6 +789,7 @@ mod tests { }, schemas: Vec::new(), producers: Vec::new(), + executable_dags: Default::default(), materializations: streaming.aggregation_configs.values().cloned().collect(), }, transmission_plan: TransmissionPlan { diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs index aaccfd43..dd6625cc 100644 --- a/data_plane/src/drivers/query/servers/http.rs +++ b/data_plane/src/drivers/query/servers/http.rs @@ -2756,6 +2756,7 @@ mod tests { }, schemas: Vec::new(), producers: Vec::new(), + executable_dags: Default::default(), materializations: Vec::new(), }, transmission_plan: TransmissionPlan { diff --git a/data_plane/src/main.rs b/data_plane/src/main.rs index e15a047f..cf421a5c 100644 --- a/data_plane/src/main.rs +++ b/data_plane/src/main.rs @@ -759,6 +759,7 @@ async fn main() -> Result<()> { .values() .cloned() .collect(), + executable_dags: Default::default(), }; let initial_transmission_plan = control_plane::physical::compiler::TransmissionPlan { summary_catalog: None, diff --git a/data_plane/src/precompute_engine/subdag_scheduler.rs b/data_plane/src/precompute_engine/subdag_scheduler.rs index 94895b9a..dc83ec0b 100644 --- a/data_plane/src/precompute_engine/subdag_scheduler.rs +++ b/data_plane/src/precompute_engine/subdag_scheduler.rs @@ -172,6 +172,7 @@ mod tests { )]) .collect(), query_sink: PostAsapNodeId(4), + query_plan_sink: control_plane::query_plan::QueryNodeId(9), precompute_sinks: vec![PostAsapNodeId(3)], } } @@ -323,6 +324,7 @@ mod tests { .into_iter() .collect(), query_sink: PostAsapNodeId(0), + query_plan_sink: control_plane::query_plan::QueryNodeId(1), precompute_sinks: vec![PostAsapNodeId(1)], }; assert!(matches!( diff --git a/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs b/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs index b2c1d8f5..9914b702 100644 --- a/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs +++ b/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs @@ -287,7 +287,8 @@ impl QueryNodeRuntime for PhysicalQueryRuntime<'_> { } QueryPlanNode::Logical { .. } | QueryPlanNode::CandidateTopK { .. } - | QueryPlanNode::Relational { .. } => Err(PhysicalNodeError::Fallback( + | QueryPlanNode::Relational { .. } + | QueryPlanNode::ExternalExact { .. } => Err(PhysicalNodeError::Fallback( "logical node requires installed logical runtime".into(), )), QueryPlanNode::ExactFallback { reason } => { diff --git a/data_plane/src/storage_engines/types/hot_reload_config.rs b/data_plane/src/storage_engines/types/hot_reload_config.rs index 598c980b..ceec6ba5 100644 --- a/data_plane/src/storage_engines/types/hot_reload_config.rs +++ b/data_plane/src/storage_engines/types/hot_reload_config.rs @@ -683,7 +683,8 @@ mod tests { }, schemas: Vec::new(), producers: Vec::new(), - materializations: Vec::new(), + executable_dags: Default::default(), + materializations: Vec::new(), }, transmission_plan: control_plane::physical::compiler::TransmissionPlan { summary_catalog: None,