1use std::collections::{HashMap, HashSet, VecDeque};
2use std::sync::Arc;
3
4use super::{
5 ExecutionError, GraphPath, MatchedEdge, MatchedNode, QueryStats, UnifiedRecord, UnifiedResult,
6};
7use crate::storage::engine::graph_store::{GraphStore, Namespace, StoredNode};
8use crate::storage::engine::graph_table_index::GraphTableIndex;
9use reddb_rql::ast::{
10 CompareOp, EdgeDirection, EdgePattern, FieldRef, Filter, GraphPattern, GraphQuery, JoinQuery,
11 JoinType, NodePattern, NodeSelector, PathQuery, Projection, QueryExpr, TableQuery,
12};
13use reddb_rql::sql_lowering::{
14 effective_graph_filter, effective_graph_projections, effective_path_filter,
15};
16use reddb_types::Value;
17
18pub type EdgeProperties = HashMap<(String, String, String), HashMap<String, Value>>;
19
20pub struct UnifiedExecutor {
21 graph: Arc<GraphStore>,
23 index: Arc<GraphTableIndex>,
25 node_properties: Arc<HashMap<String, HashMap<String, Value>>>,
27 edge_properties: Arc<EdgeProperties>,
30 work_check: Option<fn(u64) -> Result<(), ExecutionError>>,
31}
32
33impl UnifiedExecutor {
34 pub fn new(graph: Arc<GraphStore>, index: Arc<GraphTableIndex>) -> Self {
36 Self::new_with_node_properties(graph, index, HashMap::new())
37 }
38
39 pub fn new_with_node_properties(
41 graph: Arc<GraphStore>,
42 index: Arc<GraphTableIndex>,
43 node_properties: HashMap<String, HashMap<String, Value>>,
44 ) -> Self {
45 Self::new_with_graph_properties(graph, index, node_properties, HashMap::new())
46 }
47
48 pub fn new_with_graph_properties(
49 graph: Arc<GraphStore>,
50 index: Arc<GraphTableIndex>,
51 node_properties: HashMap<String, HashMap<String, Value>>,
52 edge_properties: EdgeProperties,
53 ) -> Self {
54 Self {
55 graph,
56 index,
57 node_properties: Arc::new(node_properties),
58 edge_properties: Arc::new(edge_properties),
59 work_check: None,
60 }
61 }
62
63 fn check_work(&self, work: u64) -> Result<(), ExecutionError> {
64 match self.work_check {
65 Some(check) => check(work),
66 None => Ok(()),
67 }
68 }
69
70 fn matched_node(&self, node: &StoredNode) -> MatchedNode {
71 let mut node = MatchedNode::from_stored(node);
72 if let Some(properties) = self.node_properties.get(&node.id) {
73 node.properties = properties.clone();
74 }
75 node
76 }
77
78 fn matched_edge(
79 &self,
80 source: &str,
81 edge_label: &str,
82 target: &str,
83 weight: f32,
84 ) -> MatchedEdge {
85 let mut edge = MatchedEdge::from_tuple(source, edge_label, target, weight);
86 if let Some(properties) = self.edge_properties.get(&(
87 source.to_string(),
88 edge_label.to_string(),
89 target.to_string(),
90 )) {
91 edge.properties = properties.clone();
92 }
93 edge
94 }
95
96 fn node_stored_property_value(node: &StoredNode, property: &str) -> Option<Value> {
97 if let Some(properties) = match property {
98 "id" => Some(Value::text(node.id.clone())),
99 "label" => Some(Value::text(node.label.clone())),
100 "type" | "node_type" => Some(Value::text(node.node_type.as_str().to_string())),
101 _ => None,
102 } {
103 return Some(properties);
104 }
105
106 None
107 }
108
109 fn node_property_value(&self, node: &StoredNode, property: &str) -> Option<Value> {
110 self.node_properties
111 .get(&node.id)
112 .and_then(|properties| properties.get(property).cloned())
113 .or_else(|| Self::node_stored_property_value(node, property))
114 }
115
116 fn node_property_value_by_id(&self, node_id: &str, property: &str) -> Option<Value> {
117 if property == "id" {
118 return Some(Value::text(node_id.to_string()));
119 }
120 if property == "label" {
121 if let Some(node) = self.graph.get_node(node_id).as_ref() {
122 return Some(Value::text(node.label.clone()));
123 }
124 return None;
125 }
126 if property == "type" || property == "node_type" {
127 return self
128 .graph
129 .get_node(node_id)
130 .map(|node| Value::text(node.node_type.as_str().to_string()));
131 }
132 self.node_properties
133 .get(node_id)
134 .and_then(|properties| properties.get(property).cloned())
135 }
136
137 pub fn execute_on(
142 graph: &GraphStore,
143 query: &QueryExpr,
144 ) -> Result<UnifiedResult, ExecutionError> {
145 Self::execute_on_with_node_properties(graph, query, HashMap::new())
146 }
147
148 pub fn execute_on_with_node_properties(
150 graph: &GraphStore,
151 query: &QueryExpr,
152 node_properties: HashMap<String, HashMap<String, Value>>,
153 ) -> Result<UnifiedResult, ExecutionError> {
154 Self::execute_on_with_graph_properties(graph, query, node_properties, HashMap::new())
155 }
156
157 pub fn execute_on_with_graph_properties(
158 graph: &GraphStore,
159 query: &QueryExpr,
160 node_properties: HashMap<String, HashMap<String, Value>>,
161 edge_properties: EdgeProperties,
162 ) -> Result<UnifiedResult, ExecutionError> {
163 Self::execute_on_with_graph_properties_checked(
164 graph,
165 query,
166 node_properties,
167 edge_properties,
168 None,
169 )
170 }
171
172 pub(crate) fn execute_on_with_graph_properties_checked(
173 graph: &GraphStore,
174 query: &QueryExpr,
175 node_properties: HashMap<String, HashMap<String, Value>>,
176 edge_properties: EdgeProperties,
177 work_check: Option<fn(u64) -> Result<(), ExecutionError>>,
178 ) -> Result<UnifiedResult, ExecutionError> {
179 let mut temp = Self::new_with_graph_properties(
180 Arc::new(GraphStore::new()),
181 Arc::new(GraphTableIndex::new()),
182 node_properties,
183 edge_properties,
184 );
185
186 temp.work_check = work_check;
187 temp.check_work(0)?;
188
189 match query {
190 QueryExpr::Graph(q) => temp.exec_graph_on(graph, q),
191 QueryExpr::Path(q) => temp.exec_path_on(graph, q),
192 QueryExpr::Table(_) => Err(ExecutionError::new(
193 "Table queries require proper executor initialization",
194 )),
195 QueryExpr::Join(_) => Err(ExecutionError::new(
196 "Join queries require proper executor initialization",
197 )),
198 QueryExpr::Vector(_) => Err(ExecutionError::new(
199 "Vector queries require VectorStore integration",
200 )),
201 QueryExpr::Hybrid(_) => Err(ExecutionError::new(
202 "Hybrid queries require VectorStore integration",
203 )),
204 QueryExpr::Insert(_)
205 | QueryExpr::Update(_)
206 | QueryExpr::Delete(_)
207 | QueryExpr::Function(_)
208 | QueryExpr::CreateTable(_)
209 | QueryExpr::CreateCollection(_)
210 | QueryExpr::CreateVector(_)
211 | QueryExpr::DropTable(_)
212 | QueryExpr::DropGraph(_)
213 | QueryExpr::DropVector(_)
214 | QueryExpr::DropDocument(_)
215 | QueryExpr::DropKv(_)
216 | QueryExpr::DropCollection(_)
217 | QueryExpr::Truncate(_)
218 | QueryExpr::AlterTable(_)
219 | QueryExpr::CreateVcsRef(_)
220 | QueryExpr::DropVcsRef(_)
221 | QueryExpr::ForkStore(_)
222 | QueryExpr::PromoteFork(_)
223 | QueryExpr::DropFork(_)
224 | QueryExpr::VcsCommand(_)
225 | QueryExpr::GraphCommand(_)
226 | QueryExpr::SearchCommand(_)
227 | QueryExpr::CreateIndex(_)
228 | QueryExpr::DropIndex(_)
229 | QueryExpr::ProbabilisticCommand(_)
230 | QueryExpr::Ask(_)
231 | QueryExpr::SetConfig { .. }
232 | QueryExpr::ShowConfig { .. }
233 | QueryExpr::Scrub { .. }
234 | QueryExpr::SetSecret { .. }
235 | QueryExpr::DeleteSecret { .. }
236 | QueryExpr::SetKv { .. }
237 | QueryExpr::DeleteKv { .. }
238 | QueryExpr::ShowSecrets { .. }
239 | QueryExpr::SetTenant(_)
240 | QueryExpr::ShowTenant
241 | QueryExpr::CreateTimeSeries(_)
242 | QueryExpr::CreateMetric(_)
243 | QueryExpr::AlterMetric(_)
244 | QueryExpr::CreateSlo(_)
245 | QueryExpr::DropTimeSeries(_)
246 | QueryExpr::CreateQueue(_)
247 | QueryExpr::AlterQueue(_)
248 | QueryExpr::DropQueue(_)
249 | QueryExpr::QueueSelect(_)
250 | QueryExpr::QueueCommand(_)
251 | QueryExpr::KvCommand(_)
252 | QueryExpr::ConfigCommand(_)
253 | QueryExpr::CreateTree(_)
254 | QueryExpr::DropTree(_)
255 | QueryExpr::TreeCommand(_)
256 | QueryExpr::ExplainAlter(_)
257 | QueryExpr::TransactionControl(_)
258 | QueryExpr::MaintenanceCommand(_)
259 | QueryExpr::CreateSchema(_)
260 | QueryExpr::DropSchema(_)
261 | QueryExpr::CreateSequence(_)
262 | QueryExpr::DropSequence(_)
263 | QueryExpr::CopyFrom(_)
264 | QueryExpr::CreateView(_)
265 | QueryExpr::DropView(_)
266 | QueryExpr::RefreshMaterializedView(_)
267 | QueryExpr::CreatePolicy(_)
268 | QueryExpr::DropPolicy(_)
269 | QueryExpr::CreateServer(_)
270 | QueryExpr::DropServer(_)
271 | QueryExpr::CreateForeignTable(_)
272 | QueryExpr::DropForeignTable(_)
273 | QueryExpr::Grant(_)
274 | QueryExpr::Revoke(_)
275 | QueryExpr::AlterUser(_)
276 | QueryExpr::CreateUser(_)
277 | QueryExpr::CreateIamPolicy { .. }
278 | QueryExpr::DropIamPolicy { .. }
279 | QueryExpr::AttachPolicy { .. }
280 | QueryExpr::DetachPolicy { .. }
281 | QueryExpr::ShowPolicies { .. }
282 | QueryExpr::ShowEffectivePermissions { .. }
283 | QueryExpr::RankOf(_)
284 | QueryExpr::ApproxRankOf(_)
285 | QueryExpr::RankRange(_)
286 | QueryExpr::SimulatePolicy { .. }
287 | QueryExpr::LintPolicy { .. }
288 | QueryExpr::MigratePolicyMode { .. }
289 | QueryExpr::CreateMigration(_)
290 | QueryExpr::ApplyMigration(_)
291 | QueryExpr::RollbackMigration(_)
292 | QueryExpr::ExplainMigration(_)
293 | QueryExpr::Explain(_)
294 | QueryExpr::EventsBackfill(_)
295 | QueryExpr::EventsBackfillStatus { .. } => Err(ExecutionError::new(
296 "DML/DDL/Command statements are not supported in UnifiedExecutor",
297 )),
298 }
299 }
300
301 fn exec_graph_on(
307 &self,
308 graph: &GraphStore,
309 query: &GraphQuery,
310 ) -> Result<UnifiedResult, ExecutionError> {
311 let mut result = UnifiedResult::empty();
312 let mut stats = QueryStats::default();
313 let effective_filter = effective_graph_filter(query);
314 let effective_projections = effective_graph_projections(query);
315
316 let matches = self.match_pattern_on(graph, &query.pattern, &mut stats)?;
317
318 for matched in matches {
319 self.check_work(1)?;
320 if Self::graph_limit_reached(result.records.len(), query.limit) {
321 break;
322 }
323 if !self.eval_filter_on_match(&effective_filter, &matched) {
324 continue;
325 }
326 let record = self.project_match(&matched, &effective_projections);
327 result.records.push(record);
328 }
329
330 result.stats = stats;
331 Ok(result)
332 }
333
334 fn exec_path_on(
336 &self,
337 graph: &GraphStore,
338 query: &PathQuery,
339 ) -> Result<UnifiedResult, ExecutionError> {
340 let mut result = UnifiedResult::empty();
341
342 let mut queue: VecDeque<(String, GraphPath)> = VecDeque::new();
344 let mut visited: HashSet<String> = HashSet::new();
345
346 let start_ids = self.resolve_selector_on(graph, &query.from);
348
349 for start in start_ids {
350 queue.push_back((start.clone(), GraphPath::start(&start)));
351 visited.insert(start);
352 }
353
354 let target_ids: HashSet<_> = self
355 .resolve_selector_on(graph, &query.to)
356 .into_iter()
357 .collect();
358 let max_len = query.max_length as usize;
359
360 while let Some((current, path)) = queue.pop_front() {
361 if path.len() > max_len {
362 continue;
363 }
364
365 if target_ids.contains(¤t) && !path.is_empty() {
366 let mut record = UnifiedRecord::new();
367 record.paths.push(path.clone());
368 result.records.push(record);
369 continue;
370 }
371
372 for (edge_type, neighbor, weight) in graph.outgoing_edges(¤t) {
374 if !query.via.is_empty() && !query.via.iter().any(|via| via == edge_type.as_str()) {
377 continue;
378 }
379
380 if !visited.contains(&neighbor) {
381 visited.insert(neighbor.clone());
382 let edge = MatchedEdge::from_tuple(¤t, edge_type, &neighbor, weight);
383 let new_path = path.extend(edge, &neighbor);
384 queue.push_back((neighbor, new_path));
385 }
386 }
387 }
388
389 result.stats.edges_scanned = visited.len() as u64;
390 Ok(result)
391 }
392
393 fn resolve_selector_on(&self, graph: &GraphStore, selector: &NodeSelector) -> Vec<String> {
395 match selector {
396 NodeSelector::ById(id) => vec![id.clone()],
397 NodeSelector::ByType {
398 node_label,
399 filter: _,
400 } => graph
401 .nodes_with_category(node_label)
402 .into_iter()
403 .map(|n| n.id)
404 .collect(),
405 NodeSelector::ByRow { table, row_id } => {
406 if let Some((table_id, row_id)) = match (table.as_str().parse::<u16>(), *row_id) {
407 (Ok(table_id), row_id) => Some((table_id, row_id)),
408 _ => None,
409 } {
410 let mut ids = Vec::new();
411
412 if let Some(node_id) = self.index.get_node_for_row(table_id, row_id) {
414 ids.push(node_id);
415 }
416
417 if ids.is_empty() {
420 ids.extend(graph.iter_nodes().filter_map(|node| {
421 let table_ref = node.table_ref?;
422 if table_ref.table_id == table_id && table_ref.row_id == row_id {
423 Some(node.id)
424 } else {
425 None
426 }
427 }));
428 }
429
430 ids
431 } else {
432 Vec::new()
433 }
434 }
435 }
436 }
437
438 pub fn execute(&self, query: &QueryExpr) -> Result<UnifiedResult, ExecutionError> {
440 match query {
441 QueryExpr::Table(q) => self.exec_table(q),
442 QueryExpr::Graph(q) => self.exec_graph(q),
443 QueryExpr::Join(q) => self.exec_join(q),
444 QueryExpr::Path(q) => self.exec_path(q),
445 QueryExpr::Vector(_) => {
446 Err(ExecutionError::new(
449 "Vector queries not yet implemented in UnifiedExecutor",
450 ))
451 }
452 QueryExpr::Hybrid(_) => {
453 Err(ExecutionError::new(
456 "Hybrid queries not yet implemented in UnifiedExecutor",
457 ))
458 }
459 QueryExpr::Insert(_)
460 | QueryExpr::Update(_)
461 | QueryExpr::Delete(_)
462 | QueryExpr::Function(_)
463 | QueryExpr::CreateTable(_)
464 | QueryExpr::CreateCollection(_)
465 | QueryExpr::CreateVector(_)
466 | QueryExpr::DropTable(_)
467 | QueryExpr::DropGraph(_)
468 | QueryExpr::DropVector(_)
469 | QueryExpr::DropDocument(_)
470 | QueryExpr::DropKv(_)
471 | QueryExpr::DropCollection(_)
472 | QueryExpr::Truncate(_)
473 | QueryExpr::AlterTable(_)
474 | QueryExpr::CreateVcsRef(_)
475 | QueryExpr::DropVcsRef(_)
476 | QueryExpr::ForkStore(_)
477 | QueryExpr::PromoteFork(_)
478 | QueryExpr::DropFork(_)
479 | QueryExpr::VcsCommand(_)
480 | QueryExpr::GraphCommand(_)
481 | QueryExpr::SearchCommand(_)
482 | QueryExpr::CreateIndex(_)
483 | QueryExpr::DropIndex(_)
484 | QueryExpr::ProbabilisticCommand(_)
485 | QueryExpr::Ask(_)
486 | QueryExpr::SetConfig { .. }
487 | QueryExpr::ShowConfig { .. }
488 | QueryExpr::Scrub { .. }
489 | QueryExpr::SetSecret { .. }
490 | QueryExpr::DeleteSecret { .. }
491 | QueryExpr::SetKv { .. }
492 | QueryExpr::DeleteKv { .. }
493 | QueryExpr::ShowSecrets { .. }
494 | QueryExpr::SetTenant(_)
495 | QueryExpr::ShowTenant
496 | QueryExpr::CreateTimeSeries(_)
497 | QueryExpr::CreateMetric(_)
498 | QueryExpr::AlterMetric(_)
499 | QueryExpr::CreateSlo(_)
500 | QueryExpr::DropTimeSeries(_)
501 | QueryExpr::CreateQueue(_)
502 | QueryExpr::AlterQueue(_)
503 | QueryExpr::DropQueue(_)
504 | QueryExpr::QueueSelect(_)
505 | QueryExpr::QueueCommand(_)
506 | QueryExpr::KvCommand(_)
507 | QueryExpr::ConfigCommand(_)
508 | QueryExpr::CreateTree(_)
509 | QueryExpr::DropTree(_)
510 | QueryExpr::TreeCommand(_)
511 | QueryExpr::ExplainAlter(_)
512 | QueryExpr::TransactionControl(_)
513 | QueryExpr::MaintenanceCommand(_)
514 | QueryExpr::CreateSchema(_)
515 | QueryExpr::DropSchema(_)
516 | QueryExpr::CreateSequence(_)
517 | QueryExpr::DropSequence(_)
518 | QueryExpr::CopyFrom(_)
519 | QueryExpr::CreateView(_)
520 | QueryExpr::DropView(_)
521 | QueryExpr::RefreshMaterializedView(_)
522 | QueryExpr::CreatePolicy(_)
523 | QueryExpr::DropPolicy(_)
524 | QueryExpr::CreateServer(_)
525 | QueryExpr::DropServer(_)
526 | QueryExpr::CreateForeignTable(_)
527 | QueryExpr::DropForeignTable(_)
528 | QueryExpr::Grant(_)
529 | QueryExpr::Revoke(_)
530 | QueryExpr::AlterUser(_)
531 | QueryExpr::CreateUser(_)
532 | QueryExpr::CreateIamPolicy { .. }
533 | QueryExpr::DropIamPolicy { .. }
534 | QueryExpr::AttachPolicy { .. }
535 | QueryExpr::DetachPolicy { .. }
536 | QueryExpr::ShowPolicies { .. }
537 | QueryExpr::ShowEffectivePermissions { .. }
538 | QueryExpr::RankOf(_)
539 | QueryExpr::ApproxRankOf(_)
540 | QueryExpr::RankRange(_)
541 | QueryExpr::SimulatePolicy { .. }
542 | QueryExpr::LintPolicy { .. }
543 | QueryExpr::MigratePolicyMode { .. }
544 | QueryExpr::CreateMigration(_)
545 | QueryExpr::ApplyMigration(_)
546 | QueryExpr::RollbackMigration(_)
547 | QueryExpr::ExplainMigration(_)
548 | QueryExpr::Explain(_)
549 | QueryExpr::EventsBackfill(_)
550 | QueryExpr::EventsBackfillStatus { .. } => Err(ExecutionError::new(
551 "DML/DDL/Command statements are not supported in UnifiedExecutor",
552 )),
553 }
554 }
555
556 fn exec_table(&self, _query: &TableQuery) -> Result<UnifiedResult, ExecutionError> {
560 Ok(UnifiedResult::empty())
563 }
564
565 fn exec_graph(&self, query: &GraphQuery) -> Result<UnifiedResult, ExecutionError> {
567 let mut result = UnifiedResult::empty();
568 let mut stats = QueryStats::default();
569
570 let matches = self.match_pattern(&query.pattern, &mut stats)?;
572
573 let effective_filter = effective_graph_filter(query);
574 let effective_projections = effective_graph_projections(query);
575
576 for matched in matches {
577 self.check_work(1)?;
578 if Self::graph_limit_reached(result.records.len(), query.limit) {
579 break;
580 }
581 if !self.eval_filter_on_match(&effective_filter, &matched) {
582 continue;
583 }
584 let record = self.project_match(&matched, &effective_projections);
585 result.push(record);
586 }
587
588 result.stats = stats;
589 Ok(result)
590 }
591
592 fn graph_limit_reached(row_count: usize, limit: Option<u64>) -> bool {
593 limit.is_some_and(|limit| row_count as u64 >= limit)
594 }
595
596 fn match_pattern(
598 &self,
599 pattern: &GraphPattern,
600 stats: &mut QueryStats,
601 ) -> Result<Vec<PatternMatch>, ExecutionError> {
602 self.match_pattern_on(self.graph.as_ref(), pattern, stats)
603 }
604
605 fn match_pattern_on(
606 &self,
607 graph: &GraphStore,
608 pattern: &GraphPattern,
609 stats: &mut QueryStats,
610 ) -> Result<Vec<PatternMatch>, ExecutionError> {
611 if pattern.nodes.is_empty() {
612 return Ok(Vec::new());
613 }
614
615 let first = &pattern.nodes[0];
617 let mut matches = self.find_matching_nodes_on(graph, first, stats)?;
618
619 for edge_pattern in &pattern.edges {
621 matches =
622 self.extend_matches_on(graph, matches, edge_pattern, &pattern.nodes, stats)?;
623 }
624
625 Ok(matches)
626 }
627
628 fn find_matching_nodes_on(
630 &self,
631 graph: &GraphStore,
632 pattern: &NodePattern,
633 stats: &mut QueryStats,
634 ) -> Result<Vec<PatternMatch>, ExecutionError> {
635 let mut matches = Vec::new();
636
637 for node in graph.iter_nodes() {
639 self.check_work(1)?;
640 stats.nodes_scanned += 1;
641
642 if let Some(ref expected) = pattern.node_label {
644 let expected_id = graph.registry.lookup(Namespace::Node, expected);
645 match expected_id {
646 Some(id) if id == node.label_id => {}
647 _ => continue,
648 }
649 }
650
651 let mut match_props = true;
653 for prop_filter in &pattern.properties {
654 if !self.eval_node_property_filter(&node, prop_filter) {
655 match_props = false;
656 break;
657 }
658 }
659
660 if match_props {
661 let mut pm = PatternMatch::new();
662 pm.nodes
663 .insert(pattern.alias.clone(), self.matched_node(&node));
664 matches.push(pm);
665 }
666 }
667
668 Ok(matches)
669 }
670
671 fn extend_matches_on(
673 &self,
674 graph: &GraphStore,
675 matches: Vec<PatternMatch>,
676 edge_pattern: &EdgePattern,
677 node_patterns: &[NodePattern],
678 stats: &mut QueryStats,
679 ) -> Result<Vec<PatternMatch>, ExecutionError> {
680 let mut extended = Vec::new();
681
682 let target_pattern = node_patterns
684 .iter()
685 .find(|n| n.alias == edge_pattern.to)
686 .ok_or_else(|| {
687 ExecutionError::new(format!(
688 "Node alias '{}' not found in pattern",
689 edge_pattern.to
690 ))
691 })?;
692
693 for pm in matches {
694 self.check_work(1)?;
695 let source_node = pm.nodes.get(&edge_pattern.from).ok_or_else(|| {
697 ExecutionError::new(format!(
698 "Source node '{}' not found in match",
699 edge_pattern.from
700 ))
701 })?;
702
703 let edges: Vec<_> = match edge_pattern.direction {
707 EdgeDirection::Outgoing => {
708 graph
709 .outgoing_edges(&source_node.id)
710 .into_iter()
711 .map(|(et, target, w)| (et, target, w, true)) .collect()
713 }
714 EdgeDirection::Incoming => {
715 graph
716 .incoming_edges(&source_node.id)
717 .into_iter()
718 .map(|(et, source, w)| (et, source, w, false)) .collect()
720 }
721 EdgeDirection::Both => {
722 let mut all: Vec<_> = graph
723 .outgoing_edges(&source_node.id)
724 .into_iter()
725 .map(|(et, target, w)| (et, target, w, true))
726 .collect();
727 all.extend(
728 graph
729 .incoming_edges(&source_node.id)
730 .into_iter()
731 .map(|(et, source, w)| (et, source, w, false)),
732 );
733 all
734 }
735 };
736
737 for (etype, other_id, weight, is_outgoing) in edges {
738 self.check_work(1)?;
739 stats.edges_scanned += 1;
740
741 if let Some(ref expected) = edge_pattern.edge_label {
744 if etype.as_str() != expected.as_str() {
745 continue;
746 }
747 }
748
749 let target_id = &other_id;
751
752 if let Some(target_node) = graph.get_node(target_id) {
753 if let Some(ref expected) = target_pattern.node_label {
755 let expected_id = graph.registry.lookup(Namespace::Node, expected);
756 match expected_id {
757 Some(id) if id == target_node.label_id => {}
758 _ => continue,
759 }
760 }
761
762 let mut match_props = true;
764 for prop_filter in &target_pattern.properties {
765 if !self.eval_node_property_filter(&target_node, prop_filter) {
766 match_props = false;
767 break;
768 }
769 }
770
771 if match_props {
772 let mut new_pm = pm.clone();
773 new_pm.nodes.insert(
774 target_pattern.alias.clone(),
775 self.matched_node(&target_node),
776 );
777 if let Some(ref alias) = edge_pattern.alias {
778 let edge = if is_outgoing {
780 self.matched_edge(&source_node.id, &etype, target_id, weight)
781 } else {
782 self.matched_edge(target_id, &etype, &source_node.id, weight)
783 };
784 new_pm.edges.insert(alias.clone(), edge);
785 }
786 extended.push(new_pm);
787 }
788 }
789 }
790 }
791
792 Ok(extended)
793 }
794
795 fn eval_node_property_filter(
797 &self,
798 node: &StoredNode,
799 filter: &reddb_rql::ast::PropertyFilter,
800 ) -> bool {
801 let Some(value) = self.node_property_value(node, filter.name.as_str()) else {
802 return false;
803 };
804
805 self.compare_values(&value, &filter.op, &filter.value)
806 }
807
808 fn compare_values(&self, left: &Value, op: &CompareOp, right: &Value) -> bool {
810 match op {
811 CompareOp::Eq => left == right,
812 CompareOp::Ne => left != right,
813 CompareOp::Lt => self.value_lt(left, right),
814 CompareOp::Le => self.value_lt(left, right) || left == right,
815 CompareOp::Gt => self.value_lt(right, left),
816 CompareOp::Ge => self.value_lt(right, left) || left == right,
817 }
818 }
819
820 fn value_lt(&self, left: &Value, right: &Value) -> bool {
822 match (left, right) {
823 (Value::Integer(a), Value::Integer(b)) => a < b,
824 (Value::Float(a), Value::Float(b)) => a < b,
825 (Value::Integer(a), Value::Float(b)) => (*a as f64) < *b,
826 (Value::Float(a), Value::Integer(b)) => *a < (*b as f64),
827 (Value::Text(a), Value::Text(b)) => a < b,
828 (Value::Timestamp(a), Value::Timestamp(b)) => a < b,
829 _ => false,
830 }
831 }
832
833 fn eval_filter_on_match(&self, filter: &Option<Filter>, matched: &PatternMatch) -> bool {
835 match filter {
836 None => true,
837 Some(f) => self.eval_filter(f, matched),
838 }
839 }
840
841 fn eval_filter(&self, filter: &Filter, matched: &PatternMatch) -> bool {
843 match filter {
844 Filter::Compare { field, op, value } => {
845 let actual = self.get_field_value(field, matched);
846 match actual {
847 Some(v) => self.compare_values(&v, op, value),
848 None => false,
849 }
850 }
851 Filter::CompareFields { left, op, right } => {
852 let l = self.get_field_value(left, matched);
853 let r = self.get_field_value(right, matched);
854 match (l, r) {
855 (Some(lv), Some(rv)) => self.compare_values(&lv, op, &rv),
856 _ => false,
857 }
858 }
859 Filter::CompareExpr { .. } => {
860 false
866 }
867 Filter::And(left, right) => {
868 self.eval_filter(left, matched) && self.eval_filter(right, matched)
869 }
870 Filter::Or(left, right) => {
871 self.eval_filter(left, matched) || self.eval_filter(right, matched)
872 }
873 Filter::Not(inner) => !self.eval_filter(inner, matched),
874 Filter::IsNull(field) => self.get_field_value(field, matched).is_none(),
875 Filter::IsNotNull(field) => self.get_field_value(field, matched).is_some(),
876 Filter::In { field, values } => match self.get_field_value(field, matched) {
877 Some(v) => values.contains(&v),
878 None => false,
879 },
880 Filter::Between { field, low, high } => match self.get_field_value(field, matched) {
881 Some(v) => !self.value_lt(&v, low) && !self.value_lt(high, &v),
882 None => false,
883 },
884 Filter::Like { field, pattern } => match self.get_field_value(field, matched) {
885 Some(Value::Text(s)) => self.match_like(&s, pattern),
886 _ => false,
887 },
888 Filter::StartsWith { field, prefix } => match self.get_field_value(field, matched) {
889 Some(Value::Text(s)) => s.starts_with(prefix),
890 _ => false,
891 },
892 Filter::EndsWith { field, suffix } => match self.get_field_value(field, matched) {
893 Some(Value::Text(s)) => s.ends_with(suffix),
894 _ => false,
895 },
896 Filter::Contains { field, substring } => match self.get_field_value(field, matched) {
897 Some(Value::Text(s)) => s.contains(substring),
898 _ => false,
899 },
900 }
901 }
902
903 fn match_like(&self, text: &str, pattern: &str) -> bool {
905 let regex_pattern = pattern.replace('%', ".*").replace('_', ".");
907
908 if pattern.starts_with('%') && pattern.ends_with('%') {
910 let inner = &pattern[1..pattern.len() - 1];
911 text.contains(inner)
912 } else if let Some(suffix) = pattern.strip_prefix('%') {
913 text.ends_with(suffix)
914 } else if let Some(prefix) = pattern.strip_suffix('%') {
915 text.starts_with(prefix)
916 } else {
917 text == pattern || regex_pattern == text
918 }
919 }
920
921 fn get_field_value(&self, field: &FieldRef, matched: &PatternMatch) -> Option<Value> {
923 match field {
924 FieldRef::NodeId { alias } => {
925 matched.nodes.get(alias).map(|n| Value::text(n.id.clone()))
926 }
927 FieldRef::NodeProperty { alias, property } => matched
928 .nodes
929 .get(alias)
930 .and_then(|n| match property.as_str() {
931 "id" => Some(Value::text(n.id.clone())),
932 "label" => Some(Value::text(n.label.clone())),
933 "type" | "node_type" => Some(Value::text(n.node_label.clone())),
934 _ => n.properties.get(property).cloned(),
935 })
936 .or_else(|| {
937 matched
938 .edges
939 .get(alias)
940 .and_then(|e| Self::edge_property_value(e, property))
941 }),
942 FieldRef::EdgeProperty { alias, property } => matched
943 .edges
944 .get(alias)
945 .and_then(|e| Self::edge_property_value(e, property)),
946 FieldRef::TableColumn { table, column } => {
947 if !table.is_empty() {
953 if let Some(n) = matched.nodes.get(table) {
954 return match column.as_str() {
955 "id" => Some(Value::text(n.id.clone())),
956 "label" => Some(Value::text(n.label.clone())),
957 "type" | "node_type" => Some(Value::text(n.node_label.clone())),
958 other => n.properties.get(other).cloned(),
959 };
960 }
961 if let Some(e) = matched.edges.get(table) {
962 return Self::edge_property_value(e, column);
963 }
964 }
965 None
966 }
967 }
968 }
969
970 fn edge_property_value(edge: &MatchedEdge, property: &str) -> Option<Value> {
971 match property {
972 "weight" => Some(Value::Float(edge.weight as f64)),
973 "from" | "source" => Some(Value::text(edge.from.clone())),
974 "to" | "target" => Some(Value::text(edge.to.clone())),
975 "label" | "type" | "edge_type" => Some(Value::text(edge.edge_label.clone())),
976 other => edge.properties.get(other).cloned(),
977 }
978 }
979
980 fn get_join_value(&self, field: &FieldRef, record: &UnifiedRecord) -> Option<Value> {
982 match field {
983 FieldRef::TableColumn { column, .. } => record.get(column.as_str()).cloned(),
984 FieldRef::NodeId { alias } => record
985 .nodes
986 .get(alias)
987 .map(|node| Value::text(node.id.clone())),
988 FieldRef::NodeProperty { alias, property } => {
989 record
990 .nodes
991 .get(alias)
992 .and_then(|n| match property.as_str() {
993 "id" => Some(Value::text(n.id.clone())),
994 "label" => Some(Value::text(n.label.clone())),
995 "type" | "node_type" => Some(Value::text(n.node_label.clone())),
996 _ => n.properties.get(property).cloned(),
997 })
998 }
999 FieldRef::EdgeProperty { alias, property } => {
1000 record
1001 .edges
1002 .get(alias)
1003 .and_then(|e| match property.as_str() {
1004 "weight" => Some(Value::Float(e.weight as f64)),
1005 "from" | "source" => Some(Value::text(e.from.clone())),
1006 "to" | "target" => Some(Value::text(e.to.clone())),
1007 "label" | "type" | "edge_type" => Some(Value::text(e.edge_label.clone())),
1008 other => e.properties.get(other).cloned(),
1009 })
1010 }
1011 }
1012 }
1013
1014 fn project_match(&self, matched: &PatternMatch, projections: &[Projection]) -> UnifiedRecord {
1016 let mut record = UnifiedRecord::new();
1017
1018 record.nodes = matched.nodes.clone();
1020 record.edges = matched.edges.clone();
1021
1022 for proj in projections {
1024 match proj {
1025 Projection::Field(field, alias) => {
1026 if let (FieldRef::NodeId { alias: node_alias }, None) = (field, alias) {
1031 if let Some(node) = matched.nodes.get(node_alias) {
1032 record.set(&format!("{}.id", node_alias), Value::text(node.id.clone()));
1033 record.set(
1034 &format!("{}.label", node_alias),
1035 Value::text(node.label.clone()),
1036 );
1037 record.set(
1038 &format!("{}.node_type", node_alias),
1039 Value::text(node.node_label.clone()),
1040 );
1041 for (k, v) in &node.properties {
1042 record.set(&format!("{}.{}", node_alias, k), v.clone());
1043 }
1044 continue;
1045 }
1046 if let Some(edge) = matched.edges.get(node_alias) {
1047 record.set(
1048 &format!("{}.from", node_alias),
1049 Value::text(edge.from.clone()),
1050 );
1051 record.set(&format!("{}.to", node_alias), Value::text(edge.to.clone()));
1052 record.set(
1053 &format!("{}.label", node_alias),
1054 Value::text(edge.edge_label.clone()),
1055 );
1056 record.set(
1057 &format!("{}.weight", node_alias),
1058 Value::Float(edge.weight as f64),
1059 );
1060 for (k, v) in &edge.properties {
1061 record.set(&format!("{}.{}", node_alias, k), v.clone());
1062 }
1063 continue;
1064 }
1065 }
1066 if let Some(value) = self.get_field_value(field, matched) {
1067 let key = alias.clone().unwrap_or_else(|| self.field_to_string(field));
1068 record.set(&key, value);
1069 }
1070 }
1071 Projection::All => {
1072 for (alias, node) in &matched.nodes {
1074 record.set(&format!("{}.id", alias), Value::text(node.id.clone()));
1075 record.set(&format!("{}.label", alias), Value::text(node.label.clone()));
1076 }
1077 }
1078 Projection::Column(col) => {
1079 for node in matched.nodes.values() {
1081 match col.as_str() {
1082 "id" => record.set(col, Value::text(node.id.clone())),
1083 "label" => record.set(col, Value::text(node.label.clone())),
1084 _ => {}
1085 }
1086 }
1087 }
1088 Projection::Alias(col, alias) => {
1089 for node in matched.nodes.values() {
1090 match col.as_str() {
1091 "id" => record.set(alias, Value::text(node.id.clone())),
1092 "label" => record.set(alias, Value::text(node.label.clone())),
1093 _ => {}
1094 }
1095 }
1096 }
1097 _ => {} }
1099 }
1100
1101 record
1102 }
1103
1104 fn field_to_string(&self, field: &FieldRef) -> String {
1106 match field {
1107 FieldRef::NodeId { alias } => format!("{}.id", alias),
1108 FieldRef::NodeProperty { alias, property } => format!("{}.{}", alias, property),
1109 FieldRef::EdgeProperty { alias, property } => format!("{}.{}", alias, property),
1110 FieldRef::TableColumn { table, column } => {
1111 if table.is_empty() {
1112 column.clone()
1113 } else {
1114 format!("{}.{}", table, column)
1115 }
1116 }
1117 }
1118 }
1119
1120 fn exec_join(&self, query: &JoinQuery) -> Result<UnifiedResult, ExecutionError> {
1122 let left_result = self.execute(&query.left)?;
1124
1125 let right_result = self.execute(&query.right)?;
1127
1128 let mut result = UnifiedResult::empty();
1130
1131 for left in &left_result.records {
1133 let left_value = self.get_join_value(&query.on.left_field, left);
1134
1135 for right in &right_result.records {
1136 let right_value = self.get_join_value(&query.on.right_field, right);
1137
1138 if left_value == right_value {
1139 let mut merged = left.clone();
1141 merged.nodes.extend(right.nodes.clone());
1142 merged.edges.extend(right.edges.clone());
1143 for (k, v) in right.iter_fields() {
1144 merged.set_arc(k.clone(), v.clone());
1145 }
1146 result.push(merged);
1147 }
1148 }
1149
1150 if matches!(query.join_type, JoinType::LeftOuter) {
1152 if !right_result
1154 .records
1155 .iter()
1156 .any(|r| self.get_join_value(&query.on.right_field, r) == left_value)
1157 {
1158 result.push(left.clone());
1159 }
1160 }
1161 }
1162
1163 Ok(result)
1164 }
1165
1166 fn exec_path(&self, query: &PathQuery) -> Result<UnifiedResult, ExecutionError> {
1168 let mut result = UnifiedResult::empty();
1169 let mut stats = QueryStats::default();
1170
1171 let start_nodes = self.resolve_selector(&query.from, &mut stats)?;
1173
1174 let target_nodes: HashSet<String> = self
1176 .resolve_selector(&query.to, &mut stats)?
1177 .into_iter()
1178 .collect();
1179
1180 for start_id in start_nodes {
1182 let paths = self.bfs_paths(
1183 &start_id,
1184 &target_nodes,
1185 &query.via,
1186 query.max_length,
1187 &mut stats,
1188 )?;
1189
1190 for path in paths {
1191 if effective_path_filter(query).is_some() {
1193 }
1196
1197 let mut record = UnifiedRecord::new();
1198 record.paths.push(path);
1199 result.push(record);
1200 }
1201 }
1202
1203 result.stats = stats;
1204 Ok(result)
1205 }
1206
1207 fn resolve_selector(
1209 &self,
1210 selector: &NodeSelector,
1211 stats: &mut QueryStats,
1212 ) -> Result<Vec<String>, ExecutionError> {
1213 match selector {
1214 NodeSelector::ById(id) => Ok(vec![id.clone()]),
1215 NodeSelector::ByType { node_label, filter } => {
1216 let expected_id = self.graph.registry.lookup(Namespace::Node, node_label);
1217 let mut nodes = Vec::new();
1218 for node in self.graph.iter_nodes() {
1219 stats.nodes_scanned += 1;
1220 if expected_id.map(|id| node.label_id == id).unwrap_or(false) {
1221 let matches_filter = filter
1222 .as_ref()
1223 .map(|f| self.eval_node_property_filter(&node, f))
1224 .unwrap_or(true);
1225 if matches_filter {
1226 nodes.push(node.id.clone());
1227 }
1228 }
1229 }
1230 Ok(nodes)
1231 }
1232 NodeSelector::ByRow { row_id, .. } => {
1233 if let Some(node_id) = self.index.get_node_for_row(0, *row_id) {
1236 Ok(vec![node_id])
1237 } else {
1238 Ok(Vec::new())
1239 }
1240 }
1241 }
1242 }
1243
1244 fn bfs_paths(
1246 &self,
1247 start: &str,
1248 targets: &HashSet<String>,
1249 via: &[String],
1250 max_length: u32,
1251 stats: &mut QueryStats,
1252 ) -> Result<Vec<GraphPath>, ExecutionError> {
1253 let mut paths = Vec::new();
1254 let mut queue: VecDeque<GraphPath> = VecDeque::new();
1255 let mut visited: HashSet<String> = HashSet::new();
1256
1257 queue.push_back(GraphPath::start(start));
1258 visited.insert(start.to_string());
1259
1260 while let Some(current_path) = queue.pop_front() {
1261 let Some(current_node) = current_path.nodes.last() else {
1262 continue;
1263 };
1264
1265 if targets.contains(current_node) && !current_path.is_empty() {
1267 paths.push(current_path.clone());
1268 continue;
1269 }
1270
1271 if current_path.len() >= max_length as usize {
1273 continue;
1274 }
1275
1276 for (edge_type, target_id, weight) in self.graph.outgoing_edges(current_node) {
1278 stats.edges_scanned += 1;
1279
1280 if !via.is_empty() && !via.iter().any(|v| v == edge_type.as_str()) {
1282 continue;
1283 }
1284
1285 if visited.contains(&target_id) {
1287 continue;
1288 }
1289
1290 let edge = MatchedEdge::from_tuple(current_node, edge_type, &target_id, weight);
1291 let new_path = current_path.extend(edge, &target_id);
1292 visited.insert(target_id.clone());
1293 queue.push_back(new_path);
1294 }
1295 }
1296
1297 Ok(paths)
1298 }
1299}
1300
1301#[derive(Debug, Clone, Default)]
1303struct PatternMatch {
1304 nodes: HashMap<String, MatchedNode>,
1305 edges: HashMap<String, MatchedEdge>,
1306}
1307
1308impl PatternMatch {
1309 fn new() -> Self {
1310 Self::default()
1311 }
1312}