uqa_planner/
executor.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Planner-to-physical bridge + plan executor.
8//!
9//! [`PlanExecutor`] is the planner-side entry point for executing an
10//! [`OperatorTree`] through a runtime driver. It records root timing
11//! statistics and produces an `EXPLAIN`-style tree string.
12
13use std::time::Instant;
14
15use uqa_core::{GeneralizedPostingList, PostingList};
16use uqa_graph::GraphPostingList;
17use uqa_operators::{DeepFusionLayer, OperatorTree};
18
19/// Statistics collected during plan execution.
20#[derive(Debug, Clone, Default)]
21pub struct ExecutionStats {
22    pub operator_name: String,
23    pub elapsed_ms: f64,
24    pub result_count: usize,
25    pub children: Vec<ExecutionStats>,
26}
27
28/// Materialised value produced by a physical [`OperatorTree`] node.
29///
30/// Most operators preserve one document id per row and therefore emit a
31/// [`PostingList`]. Graph operators retain their subgraph side table in a
32/// [`GraphPostingList`], while join operators preserve an ordered tuple of
33/// document ids in a [`GeneralizedPostingList`]. Keeping the carriers distinct
34/// prevents set operations from silently applying ordinary payload precedence
35/// to graph metadata or comparing synthetic join enumeration positions.
36#[derive(Debug, Clone, PartialEq)]
37pub enum OperatorOutput {
38    Posting(PostingList),
39    Graph(GraphPostingList),
40    Generalized(GeneralizedPostingList),
41}
42
43impl OperatorOutput {
44    #[must_use]
45    pub fn len(&self) -> usize {
46        match self {
47            Self::Posting(result) => result.len(),
48            Self::Graph(result) => result.len(),
49            Self::Generalized(result) => result.len(),
50        }
51    }
52
53    #[must_use]
54    pub fn is_empty(&self) -> bool {
55        match self {
56            Self::Posting(result) => result.is_empty(),
57            Self::Graph(result) => result.is_empty(),
58            Self::Generalized(result) => result.is_empty(),
59        }
60    }
61
62    #[must_use]
63    pub fn as_posting(&self) -> Option<&PostingList> {
64        match self {
65            Self::Posting(result) => Some(result),
66            Self::Graph(_) | Self::Generalized(_) => None,
67        }
68    }
69
70    #[must_use]
71    pub fn as_graph(&self) -> Option<&GraphPostingList> {
72        match self {
73            Self::Graph(result) => Some(result),
74            Self::Posting(_) | Self::Generalized(_) => None,
75        }
76    }
77
78    #[must_use]
79    pub fn as_generalized(&self) -> Option<&GeneralizedPostingList> {
80        match self {
81            Self::Posting(_) | Self::Graph(_) => None,
82            Self::Generalized(result) => Some(result),
83        }
84    }
85}
86
87impl From<PostingList> for OperatorOutput {
88    fn from(value: PostingList) -> Self {
89        Self::Posting(value)
90    }
91}
92
93impl From<GraphPostingList> for OperatorOutput {
94    fn from(value: GraphPostingList) -> Self {
95        Self::Graph(value)
96    }
97}
98
99impl From<GeneralizedPostingList> for OperatorOutput {
100    fn from(value: GeneralizedPostingList) -> Self {
101        Self::Generalized(value)
102    }
103}
104
105impl ExecutionStats {
106    pub fn new(name: impl Into<String>) -> Self {
107        Self {
108            operator_name: name.into(),
109            elapsed_ms: 0.0,
110            result_count: 0,
111            children: Vec::new(),
112        }
113    }
114}
115
116/// Driver that knows how to execute an [`OperatorTree`] node. The
117/// engine implements this trait (it owns the per-operator dispatch,
118/// child execution, and the runtime context); the planner-side
119/// `PlanExecutor` only wraps root timing and stats collection.
120pub trait OperatorTreeDriver {
121    /// Failure produced by the physical runtime while materialising a
122    /// node or one of its descendants.
123    type Error;
124
125    fn execute_node(&self, op: &OperatorTree) -> Result<OperatorOutput, Self::Error>;
126}
127
128/// Executor wrapper for [`OperatorTree`].
129pub struct PlanExecutor<'d, D: OperatorTreeDriver> {
130    pub driver: &'d D,
131    last_stats: Option<ExecutionStats>,
132}
133
134impl<'d, D: OperatorTreeDriver> PlanExecutor<'d, D> {
135    pub fn new(driver: &'d D) -> Self {
136        Self {
137            driver,
138            last_stats: None,
139        }
140    }
141
142    /// Execute `op` and capture stats. Driver failures are returned to
143    /// the caller instead of being indistinguishable from an empty
144    /// physical result.
145    pub fn execute(&mut self, op: &OperatorTree) -> Result<OperatorOutput, D::Error> {
146        self.last_stats = None;
147        let start = Instant::now();
148        let result = self.driver.execute_node(op)?;
149        let mut stats = ExecutionStats::new(operator_name(op));
150        stats.elapsed_ms = start.elapsed().as_secs_f64() * 1000.0;
151        stats.result_count = result.len();
152        self.last_stats = Some(stats);
153        Ok(result)
154    }
155
156    pub fn last_stats(&self) -> Option<&ExecutionStats> {
157        self.last_stats.as_ref()
158    }
159
160    pub fn explain(&self, op: &OperatorTree) -> String {
161        let mut lines: Vec<String> = Vec::new();
162        explain_recursive(op, &mut lines, 0);
163        lines.join("\n")
164    }
165}
166
167/// Stable human-readable name for an [`OperatorTree`] variant.
168pub fn operator_name(op: &OperatorTree) -> String {
169    match op {
170        OperatorTree::Empty => "EmptyOp",
171        OperatorTree::Term {
172            top_k:
173                Some(uqa_operators::TextTopKPlan {
174                    strategy: uqa_operators::TextTopKStrategy::Wand,
175                    ..
176                }),
177            ..
178        } => "WANDTopKOp",
179        OperatorTree::Term {
180            top_k:
181                Some(uqa_operators::TextTopKPlan {
182                    strategy: uqa_operators::TextTopKStrategy::BlockMaxWand,
183                    ..
184                }),
185            ..
186        } => "BlockMaxWANDTopKOp",
187        OperatorTree::Term { .. } => "TermOp",
188        OperatorTree::Filter { .. } => "FilterOp",
189        OperatorTree::Facet { .. } => "FacetOp",
190        OperatorTree::Score { .. } => "ScoreOp",
191        OperatorTree::BayesianScore { .. } => "BayesianScoreQuery",
192        OperatorTree::BayesianMatchWithPrior { .. } => "BayesianMatchWithPriorOp",
193        OperatorTree::Intersect(_) => "IntersectOp",
194        OperatorTree::Union(_) => "UnionOp",
195        OperatorTree::Complement(_) => "ComplementOp",
196        OperatorTree::Composed(_) => "ComposedOp",
197        OperatorTree::EncodeGraphPosting { .. } => "EncodeGraphPostingOp",
198        OperatorTree::VectorSimilarity { .. } => "VectorSimOp",
199        OperatorTree::KNN { .. } => "KNNOp",
200        OperatorTree::CalibratedVectorMatch { .. } => "CalibratedVectorMatchOp",
201        OperatorTree::CosineProbability(_) => "CosineProbabilityOp",
202        OperatorTree::BayesianEvidenceFusion { .. } => "BayesianEvidenceFusion",
203        OperatorTree::RobustPositiveEvidencePool { .. } => "RobustPositiveEvidencePool",
204        OperatorTree::ProbBoolFusion { .. } => "ProbBoolFusion",
205        OperatorTree::ProbNot { .. } => "ProbNot",
206        OperatorTree::AttentionFusion { .. } => "AttentionFusion",
207        OperatorTree::LearnedFusion { .. } => "LearnedFusion",
208        OperatorTree::SparseThreshold { .. } => "SparseThreshold",
209        OperatorTree::Traverse { .. } => "TraverseOp",
210        OperatorTree::GraphNeighbors { .. } => "GraphNeighborsOp",
211        OperatorTree::GraphEdges { .. } => "GraphEdgesOp",
212        OperatorTree::PatternMatch { .. } => "PatternMatchOp",
213        OperatorTree::RegularPathQuery { .. } => "RPQOp",
214        OperatorTree::GraphJoin { .. } => "GraphJoinOp",
215        OperatorTree::IndexScan { .. } => "IndexScanOp",
216        OperatorTree::Aggregate { .. } => "AggregateOp",
217        OperatorTree::GroupBy { .. } => "GroupByOp",
218        OperatorTree::MultiStage { .. } => "MultiStage",
219        OperatorTree::MultiFieldSearch { .. } => "MultiFieldSearchOp",
220        OperatorTree::HybridTextVector { .. } => "HybridTextVectorOp",
221        OperatorTree::SemanticFilter { .. } => "SemanticFilterOp",
222        OperatorTree::VectorExclusion { .. } => "VectorExclusionOp",
223        OperatorTree::FacetVector { .. } => "FacetVectorOp",
224        OperatorTree::VertexAggregation { .. } => "VertexAggregationOp",
225        OperatorTree::WeightedPathQuery { .. } => "WeightedPathQueryOp",
226        OperatorTree::MessagePassing { .. } => "MessagePassingOp",
227        OperatorTree::GraphEmbedding { .. } => "GraphEmbeddingOp",
228        OperatorTree::PageRank { .. } => "PageRankOp",
229        OperatorTree::HITS { .. } => "HITSOp",
230        OperatorTree::BetweennessCentrality { .. } => "BetweennessCentralityOp",
231        OperatorTree::TextSimilarityJoin { .. } => "TextSimilarityJoinOp",
232        OperatorTree::VectorSimilarityJoin { .. } => "VectorSimilarityJoinOp",
233        OperatorTree::HybridJoin { .. } => "HybridJoinOp",
234        OperatorTree::CrossParadigmJoin { .. } => "CrossParadigmJoinOp",
235        OperatorTree::TemporalTraverse { .. } => "TemporalTraverseOp",
236        OperatorTree::TemporalPatternMatch { .. } => "TemporalPatternMatchOp",
237        OperatorTree::ProgressiveFusion { .. } => "ProgressiveFusionOp",
238        OperatorTree::DeepFusion { .. } => "DeepFusion",
239        OperatorTree::DeepPredict { .. } => "DeepPredictOp",
240        OperatorTree::Opaque { kind, .. } => return kind.clone(),
241    }
242    .to_string()
243}
244
245#[expect(
246    clippy::too_many_lines,
247    reason = "executor exhaustively maps every operator variant and output carrier"
248)]
249fn explain_recursive(op: &OperatorTree, lines: &mut Vec<String>, indent: usize) {
250    let prefix = "  ".repeat(indent);
251    match op {
252        OperatorTree::Term {
253            query,
254            field,
255            scoring,
256            top_k,
257        } => {
258            lines.push(format!(
259                "{prefix}TermOp(term={query:?}, field={field:?}, scoring={scoring:?}, top_k={top_k:?})"
260            ));
261        }
262        OperatorTree::VectorSimilarity {
263            threshold, field, ..
264        } => {
265            lines.push(format!(
266                "{prefix}VectorSimOp(threshold={threshold}, field={field:?})"
267            ));
268        }
269        OperatorTree::KNN { k, field, .. } => {
270            lines.push(format!("{prefix}KNNOp(k={k}, field={field:?})"));
271        }
272        OperatorTree::IndexScan {
273            field, index_name, ..
274        } => {
275            lines.push(format!(
276                "{prefix}IndexScanOp(field={field:?}, index={index_name:?})"
277            ));
278        }
279        OperatorTree::Score {
280            query_terms,
281            field,
282            source,
283            ..
284        } => {
285            lines.push(format!(
286                "{prefix}ScoreOp(scorer=Scorer, terms={query_terms:?}, field={field:?})"
287            ));
288            explain_recursive(source, lines, indent + 1);
289        }
290        OperatorTree::BayesianScore { source, field } => {
291            lines.push(format!("{prefix}BayesianScoreQuery(field={field:?})"));
292            explain_recursive(source, lines, indent + 1);
293        }
294        OperatorTree::Filter { field, source, .. } => {
295            lines.push(format!("{prefix}FilterOp(field={field:?})"));
296            if let Some(src) = source {
297                explain_recursive(src, lines, indent + 1);
298            }
299        }
300        OperatorTree::BayesianEvidenceFusion { signals, base_rate } => {
301            lines.push(format!(
302                "{prefix}BayesianEvidenceFusion(base_rate={base_rate:?}, signals={})",
303                signals.len()
304            ));
305            for signal in signals {
306                explain_recursive(signal, lines, indent + 1);
307            }
308        }
309        OperatorTree::RobustPositiveEvidencePool { signals, alpha, .. } => {
310            lines.push(format!(
311                "{prefix}RobustPositiveEvidencePool(alpha={alpha}, signals={})",
312                signals.len()
313            ));
314            for sig in signals {
315                explain_recursive(sig, lines, indent + 1);
316            }
317        }
318        OperatorTree::ProbBoolFusion { signals, mode } => {
319            lines.push(format!(
320                "{prefix}ProbBoolFusion(mode={mode:?}, signals={})",
321                signals.len()
322            ));
323            for sig in signals {
324                explain_recursive(sig, lines, indent + 1);
325            }
326        }
327        OperatorTree::ProbNot { signal, .. } => {
328            lines.push(format!("{prefix}ProbNot"));
329            explain_recursive(signal, lines, indent + 1);
330        }
331        OperatorTree::AttentionFusion { signals, .. } => {
332            lines.push(format!(
333                "{prefix}AttentionFusion(signals={})",
334                signals.len()
335            ));
336            for sig in signals {
337                explain_recursive(sig, lines, indent + 1);
338            }
339        }
340        OperatorTree::LearnedFusion { signals, .. } => {
341            lines.push(format!("{prefix}LearnedFusion(signals={})", signals.len()));
342            for sig in signals {
343                explain_recursive(sig, lines, indent + 1);
344            }
345        }
346        OperatorTree::Traverse {
347            start_vertex,
348            label,
349            max_hops,
350            ..
351        } => {
352            lines.push(format!(
353                "{prefix}TraverseOp(start={start_vertex}, label={label:?}, hops={max_hops})"
354            ));
355        }
356        OperatorTree::PatternMatch { pattern, .. } => {
357            lines.push(format!(
358                "{prefix}PatternMatchOp(vertices={}, edges={})",
359                pattern.vertex_patterns.len(),
360                pattern.edge_patterns.len()
361            ));
362        }
363        OperatorTree::RegularPathQuery { start_vertex, .. } => {
364            lines.push(format!("{prefix}RPQOp(start={start_vertex})"));
365        }
366        OperatorTree::Intersect(ops) => {
367            lines.push(format!("{prefix}Intersect"));
368            for child in ops {
369                explain_recursive(child, lines, indent + 1);
370            }
371        }
372        OperatorTree::Union(ops) => {
373            lines.push(format!("{prefix}Union"));
374            for child in ops {
375                explain_recursive(child, lines, indent + 1);
376            }
377        }
378        OperatorTree::Complement(inner) => {
379            lines.push(format!("{prefix}Complement"));
380            explain_recursive(inner, lines, indent + 1);
381        }
382        OperatorTree::Composed(ops) => {
383            lines.push(format!("{prefix}Composed"));
384            for child in ops {
385                explain_recursive(child, lines, indent + 1);
386            }
387        }
388        OperatorTree::EncodeGraphPosting { source } => {
389            lines.push(format!("{prefix}EncodeGraphPosting"));
390            explain_recursive(source, lines, indent + 1);
391        }
392        OperatorTree::SparseThreshold { source, threshold } => {
393            lines.push(format!("{prefix}SparseThreshold(threshold={threshold})"));
394            explain_recursive(source, lines, indent + 1);
395        }
396        OperatorTree::MessagePassing { source } => {
397            lines.push(format!("{prefix}MessagePassingOp"));
398            explain_recursive(source, lines, indent + 1);
399        }
400        OperatorTree::GraphEmbedding { source } => {
401            lines.push(format!("{prefix}GraphEmbeddingOp"));
402            explain_recursive(source, lines, indent + 1);
403        }
404        OperatorTree::MultiStage { stages } => {
405            lines.push(format!("{prefix}MultiStage(stages={})", stages.len()));
406            for (i, entry) in stages.iter().enumerate() {
407                lines.push(format!("{prefix}  Stage {i} (cutoff={:?}):", entry.cutoff));
408                explain_recursive(&entry.child, lines, indent + 2);
409            }
410        }
411        OperatorTree::DeepFusion {
412            layers,
413            alpha,
414            gating,
415        } => {
416            lines.push(format!(
417                "{prefix}DeepFusion(layers={}, alpha={alpha}, gating={gating:?})",
418                layers.len()
419            ));
420            for (i, layer) in layers.iter().enumerate() {
421                match layer {
422                    DeepFusionLayer::Signal { signals } => {
423                        lines.push(format!("{prefix}  Layer {i} (signals={}):", signals.len()));
424                        for sig in signals {
425                            explain_recursive(sig, lines, indent + 2);
426                        }
427                    }
428                    DeepFusionLayer::Propagate {
429                        edge_label,
430                        aggregation,
431                        direction,
432                    } => {
433                        lines.push(format!(
434                            "{prefix}  Layer {i} (propagate={edge_label:?}, aggregation={aggregation:?}, direction={direction:?}):"
435                        ));
436                    }
437                    DeepFusionLayer::Conv {
438                        edge_label,
439                        hop_weights,
440                        direction,
441                    } => {
442                        lines.push(format!(
443                            "{prefix}  Layer {i} (convolve={edge_label:?}, hop_weights={hop_weights:?}, direction={direction:?}):"
444                        ));
445                    }
446                    DeepFusionLayer::Pool {
447                        edge_label,
448                        pool_size,
449                        method,
450                        direction,
451                    } => {
452                        lines.push(format!(
453                            "{prefix}  Layer {i} (pool={edge_label:?}, size={pool_size}, method={method:?}, direction={direction:?}):"
454                        ));
455                    }
456                    DeepFusionLayer::Flatten => {
457                        lines.push(format!("{prefix}  Layer {i} (flatten):"));
458                    }
459                    DeepFusionLayer::Softmax => {
460                        lines.push(format!("{prefix}  Layer {i} (softmax):"));
461                    }
462                    DeepFusionLayer::Dense {
463                        output_channels,
464                        input_channels,
465                        ..
466                    } => {
467                        lines.push(format!(
468                            "{prefix}  Layer {i} (dense, input={input_channels}, output={output_channels}):"
469                        ));
470                    }
471                    DeepFusionLayer::BatchNorm { epsilon } => {
472                        lines.push(format!(
473                            "{prefix}  Layer {i} (batch_norm, epsilon={epsilon}):"
474                        ));
475                    }
476                    DeepFusionLayer::Dropout { probability } => {
477                        lines.push(format!(
478                            "{prefix}  Layer {i} (dropout, probability={probability}):"
479                        ));
480                    }
481                }
482            }
483        }
484        OperatorTree::Opaque { kind, children, .. } => {
485            lines.push(format!("{prefix}{kind}"));
486            for child in children {
487                explain_recursive(child, lines, indent + 1);
488            }
489        }
490        other => {
491            lines.push(format!("{prefix}{}", operator_name(other)));
492        }
493    }
494}
495
496#[cfg(test)]
497mod tests {
498    use super::*;
499    use std::sync::atomic::{AtomicUsize, Ordering};
500
501    struct EmptyDriver;
502    impl OperatorTreeDriver for EmptyDriver {
503        type Error = std::convert::Infallible;
504
505        fn execute_node(&self, _op: &OperatorTree) -> Result<OperatorOutput, Self::Error> {
506            Ok(PostingList::new().into())
507        }
508    }
509
510    #[test]
511    fn explain_renders_intersect_tree() {
512        let op = OperatorTree::Intersect(vec![
513            OperatorTree::Term {
514                query: "rust".into(),
515                field: Some("body".into()),
516                scoring: None,
517                top_k: None,
518            },
519            OperatorTree::Filter {
520                field: "year".into(),
521                predicate: uqa_core::Predicate::Equals(uqa_core::Value::Int(2026)),
522                source: None,
523            },
524        ]);
525        let driver = EmptyDriver;
526        let executor = PlanExecutor::new(&driver);
527        let text = executor.explain(&op);
528        assert!(text.contains("Intersect"));
529        assert!(text.contains("TermOp"));
530        assert!(text.contains("FilterOp"));
531    }
532
533    #[test]
534    fn execute_collects_timing_stats() {
535        let op = OperatorTree::Term {
536            query: "x".into(),
537            field: Some("body".into()),
538            scoring: None,
539            top_k: None,
540        };
541        let driver = EmptyDriver;
542        let mut executor = PlanExecutor::new(&driver);
543        let _ = executor.execute(&op).expect("infallible driver");
544        let stats = executor.last_stats().expect("stats");
545        assert_eq!(stats.operator_name, "TermOp");
546        assert_eq!(stats.result_count, 0);
547        assert!(stats.elapsed_ms >= 0.0);
548    }
549
550    #[test]
551    fn execute_delegates_to_driver_once() {
552        struct CountingDriver {
553            calls: AtomicUsize,
554        }
555
556        impl OperatorTreeDriver for CountingDriver {
557            type Error = std::convert::Infallible;
558
559            fn execute_node(&self, _op: &OperatorTree) -> Result<OperatorOutput, Self::Error> {
560                self.calls.fetch_add(1, Ordering::SeqCst);
561                Ok(PostingList::new().into())
562            }
563        }
564
565        let op = OperatorTree::Intersect(vec![
566            OperatorTree::Term {
567                query: "rust".into(),
568                field: Some("body".into()),
569                scoring: None,
570                top_k: None,
571            },
572            OperatorTree::Term {
573                query: "search".into(),
574                field: Some("body".into()),
575                scoring: None,
576                top_k: None,
577            },
578        ]);
579        let driver = CountingDriver {
580            calls: AtomicUsize::new(0),
581        };
582        let mut executor = PlanExecutor::new(&driver);
583
584        let _ = executor.execute(&op).expect("infallible driver");
585
586        assert_eq!(driver.calls.load(Ordering::SeqCst), 1);
587    }
588
589    #[test]
590    fn execute_propagates_driver_errors() {
591        struct FailingDriver;
592
593        impl OperatorTreeDriver for FailingDriver {
594            type Error = &'static str;
595
596            fn execute_node(&self, _op: &OperatorTree) -> Result<OperatorOutput, Self::Error> {
597                Err("storage failure")
598            }
599        }
600
601        let driver = FailingDriver;
602        let mut executor = PlanExecutor::new(&driver);
603        let error = executor
604            .execute(&OperatorTree::Empty)
605            .expect_err("driver failure must be returned");
606
607        assert_eq!(error, "storage failure");
608        assert!(executor.last_stats().is_none());
609    }
610}
uqa_planner/executor.rs

uqa_planner/
executor.rs
