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