uqa_graph/
operator_impls.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! `Operator` trait wrappers for graph operators that previously only
8//! existed as inherent methods (`execute(&G)`). They support generic
9//! [`uqa_operators::Operator`] composition. The engine's exhaustive
10//! `OperatorTree` driver invokes the corresponding graph primitives against
11//! its named graph store; SQL functions represented by those IR nodes use the
12//! same optimized physical path.
13
14use std::sync::Arc;
15
16use uqa_core::{IndexStats, PostingList};
17use uqa_operators::base::{ExecutionContext, Operator, OperatorResult};
18use uqa_operators::PathWeightPredicate;
19use uqa_storage::StorageBackendError;
20
21use crate::cypher::{CypherQuery, CypherWriter};
22use crate::memory_store::MemoryGraphStore;
23use crate::operators::{WeightedPathQuery, DEFAULT_GRAPH_SCORE};
24use crate::rpq::RegularPathExpr;
25
26/// `WeightedPathQueryOperator` evaluates bounded DFA walks, sums a numeric
27/// edge property, and keeps endpoints for which the accumulated weight passes
28/// `predicate`. The selectivity value is retained solely for planning; it
29/// never substitutes for physical predicate evaluation.
30pub struct WeightedPathQueryOperator {
31    pub path_expr: RegularPathExpr,
32    pub graph_store: Arc<MemoryGraphStore>,
33    pub graph_name: String,
34    pub start_vertex: Option<u64>,
35    pub weight_property: String,
36    pub default_edge_weight: f64,
37    pub max_hops: usize,
38    pub predicate: PathWeightPredicate,
39    pub predicate_selectivity: f64,
40    pub score: f64,
41}
42
43impl WeightedPathQueryOperator {
44    #[must_use]
45    pub fn new(
46        path_expr: RegularPathExpr,
47        graph_store: Arc<MemoryGraphStore>,
48        graph_name: impl Into<String>,
49    ) -> Self {
50        Self {
51            path_expr,
52            graph_store,
53            graph_name: graph_name.into(),
54            start_vertex: None,
55            weight_property: "weight".to_string(),
56            default_edge_weight: 1.0,
57            max_hops: 16,
58            predicate: Arc::new(|_| true),
59            predicate_selectivity: 1.0,
60            score: DEFAULT_GRAPH_SCORE,
61        }
62    }
63
64    #[must_use]
65    pub fn from_vertex(mut self, start: u64) -> Self {
66        self.start_vertex = Some(start);
67        self
68    }
69
70    #[must_use]
71    pub fn with_predicate_selectivity(mut self, sel: f64) -> Self {
72        self.predicate_selectivity = sel;
73        self
74    }
75
76    #[must_use]
77    pub fn with_predicate(
78        mut self,
79        predicate: impl Fn(f64) -> bool + Send + Sync + 'static,
80        selectivity: f64,
81    ) -> Self {
82        self.predicate = Arc::new(predicate);
83        self.predicate_selectivity = selectivity;
84        self
85    }
86
87    #[must_use]
88    pub fn with_weight_property(mut self, property: impl Into<String>) -> Self {
89        self.weight_property = property.into();
90        self
91    }
92
93    #[must_use]
94    pub fn with_default_edge_weight(mut self, weight: f64) -> Self {
95        self.default_edge_weight = weight;
96        self
97    }
98
99    #[must_use]
100    pub fn with_max_hops(mut self, max_hops: usize) -> Self {
101        self.max_hops = max_hops;
102        self
103    }
104
105    #[must_use]
106    pub fn with_score(mut self, score: f64) -> Self {
107        self.score = score;
108        self
109    }
110}
111
112impl Operator for WeightedPathQueryOperator {
113    fn execute(&self, _ctx: &ExecutionContext) -> OperatorResult {
114        let mut query = WeightedPathQuery::new(
115            self.path_expr.clone(),
116            &self.graph_name,
117            &self.weight_property,
118            Arc::clone(&self.predicate),
119        );
120        query.default_edge_weight = self.default_edge_weight;
121        query.max_hops = self.max_hops;
122        query.score = self.score;
123        if let Some(v) = self.start_vertex {
124            query = query.from_vertex(v);
125        }
126        Ok(query
127            .execute(self.graph_store.as_ref())
128            .map_err(|error| StorageBackendError::Other(error.to_string()))?
129            .to_posting_list())
130    }
131
132    fn cost_estimate(&self, stats: &IndexStats) -> f64 {
133        // O(V^2 * |R|) approximation; the exact cardinality model
134        // lives in `uqa_planner::cardinality::estimate_rpq`. This
135        // stays a coarse fallback so the trait impl is self-contained.
136        let n = stats.total_docs as f64;
137        n * n
138    }
139}
140
141/// `CypherQueryOperator` — execute a parsed openCypher query against
142/// a named graph and project the resulting `(start, end)` vertex
143/// pairs into a posting list for downstream operator composition.
144///
145/// Routes through [`CypherWriter`] because it covers the full clause set,
146/// including
147/// CREATE/MERGE/SET/DELETE/UNWIND. Read-only queries still flow
148/// through the same path; the writer leaves the store untouched
149/// when the query is read-only.
150pub struct CypherQueryOperator {
151    pub graph_store: Arc<parking_lot::RwLock<MemoryGraphStore>>,
152    pub query: CypherQuery,
153    pub graph_name: String,
154    pub params: std::collections::BTreeMap<String, uqa_core::Value>,
155}
156
157impl CypherQueryOperator {
158    #[must_use]
159    pub fn new(
160        graph_store: Arc<parking_lot::RwLock<MemoryGraphStore>>,
161        query: CypherQuery,
162        graph_name: impl Into<String>,
163    ) -> Self {
164        Self {
165            graph_store,
166            query,
167            graph_name: graph_name.into(),
168            params: std::collections::BTreeMap::new(),
169        }
170    }
171
172    #[must_use]
173    pub fn with_params(
174        mut self,
175        params: std::collections::BTreeMap<String, uqa_core::Value>,
176    ) -> Self {
177        self.params = params;
178        self
179    }
180}
181
182impl Operator for CypherQueryOperator {
183    fn execute(&self, _ctx: &ExecutionContext) -> OperatorResult {
184        // The Cypher executor needs a unique borrow of the store. The
185        // engine passes `Arc<RwLock<...>>` so concurrent readers can
186        // share the store while writers serialize through the lock.
187        let mut guard = self.graph_store.write();
188        let mut writer = CypherWriter::new(&mut *guard, self.graph_name.clone())
189            .with_params(self.params.clone());
190        let (_cols, rows) = writer
191            .execute(&self.query)
192            .map_err(|error| StorageBackendError::Other(error.to_string()))?;
193        // Project bound vertex/edge ids out of the result rows. The
194        // posting list carries one entry per distinct vertex id seen,
195        // so downstream operators can intersect / union the result
196        // against any other graph-result set.
197        let mut ids: std::collections::BTreeSet<u64> = std::collections::BTreeSet::new();
198        for row in rows {
199            for value in row.values() {
200                if let uqa_core::Value::Int(n) = value {
201                    if *n >= 0 {
202                        ids.insert(*n as u64);
203                    }
204                }
205            }
206        }
207        let entries: Vec<uqa_core::PostingEntry> = ids
208            .into_iter()
209            .map(|id| uqa_core::PostingEntry::new(id, uqa_core::Payload::default()))
210            .collect();
211        Ok(PostingList::from_sorted_unchecked(entries))
212    }
213
214    fn cost_estimate(&self, stats: &IndexStats) -> f64 {
215        stats.total_docs as f64
216    }
217}
uqa_graph/operator_impls.rs

uqa_graph/
operator_impls.rs
