Skip to main content

reddb_server/storage/query/unified/
executor.rs

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 storage
22    graph: Arc<GraphStore>,
23    /// Graph-table index for joins
24    index: Arc<GraphTableIndex>,
25    /// Optional node properties loaded from the unified entity store
26    node_properties: Arc<HashMap<String, HashMap<String, Value>>>,
27    /// Optional edge properties loaded from the unified entity store, keyed by
28    /// `(from, canonical_label, to)` because GraphStore adjacency omits edge ids.
29    edge_properties: Arc<EdgeProperties>,
30}
31
32impl UnifiedExecutor {
33    /// Create a new executor
34    pub fn new(graph: Arc<GraphStore>, index: Arc<GraphTableIndex>) -> Self {
35        Self::new_with_node_properties(graph, index, HashMap::new())
36    }
37
38    /// Create a new executor with node properties
39    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    /// Execute a query directly against a graph reference
129    ///
130    /// This is a convenience method for simple graph-only queries.
131    /// For table joins, use `new()` with proper Arc ownership.
132    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    /// Execute a query directly against a graph reference with custom node properties
140    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::VcsCommand(_)
193            | QueryExpr::GraphCommand(_)
194            | QueryExpr::SearchCommand(_)
195            | QueryExpr::CreateIndex(_)
196            | QueryExpr::DropIndex(_)
197            | QueryExpr::ProbabilisticCommand(_)
198            | QueryExpr::Ask(_)
199            | QueryExpr::SetConfig { .. }
200            | QueryExpr::ShowConfig { .. }
201            | QueryExpr::SetSecret { .. }
202            | QueryExpr::DeleteSecret { .. }
203            | QueryExpr::SetKv { .. }
204            | QueryExpr::DeleteKv { .. }
205            | QueryExpr::ShowSecrets { .. }
206            | QueryExpr::SetTenant(_)
207            | QueryExpr::ShowTenant
208            | QueryExpr::CreateTimeSeries(_)
209            | QueryExpr::CreateMetric(_)
210            | QueryExpr::AlterMetric(_)
211            | QueryExpr::CreateSlo(_)
212            | QueryExpr::DropTimeSeries(_)
213            | QueryExpr::CreateQueue(_)
214            | QueryExpr::AlterQueue(_)
215            | QueryExpr::DropQueue(_)
216            | QueryExpr::QueueSelect(_)
217            | QueryExpr::QueueCommand(_)
218            | QueryExpr::KvCommand(_)
219            | QueryExpr::ConfigCommand(_)
220            | QueryExpr::CreateTree(_)
221            | QueryExpr::DropTree(_)
222            | QueryExpr::TreeCommand(_)
223            | QueryExpr::ExplainAlter(_)
224            | QueryExpr::TransactionControl(_)
225            | QueryExpr::MaintenanceCommand(_)
226            | QueryExpr::CreateSchema(_)
227            | QueryExpr::DropSchema(_)
228            | QueryExpr::CreateSequence(_)
229            | QueryExpr::DropSequence(_)
230            | QueryExpr::CopyFrom(_)
231            | QueryExpr::CreateView(_)
232            | QueryExpr::DropView(_)
233            | QueryExpr::RefreshMaterializedView(_)
234            | QueryExpr::CreatePolicy(_)
235            | QueryExpr::DropPolicy(_)
236            | QueryExpr::CreateServer(_)
237            | QueryExpr::DropServer(_)
238            | QueryExpr::CreateForeignTable(_)
239            | QueryExpr::DropForeignTable(_)
240            | QueryExpr::Grant(_)
241            | QueryExpr::Revoke(_)
242            | QueryExpr::AlterUser(_)
243            | QueryExpr::CreateUser(_)
244            | QueryExpr::CreateIamPolicy { .. }
245            | QueryExpr::DropIamPolicy { .. }
246            | QueryExpr::AttachPolicy { .. }
247            | QueryExpr::DetachPolicy { .. }
248            | QueryExpr::ShowPolicies { .. }
249            | QueryExpr::ShowEffectivePermissions { .. }
250            | QueryExpr::RankOf(_)
251            | QueryExpr::ApproxRankOf(_)
252            | QueryExpr::RankRange(_)
253            | QueryExpr::SimulatePolicy { .. }
254            | QueryExpr::LintPolicy { .. }
255            | QueryExpr::MigratePolicyMode { .. }
256            | QueryExpr::CreateMigration(_)
257            | QueryExpr::ApplyMigration(_)
258            | QueryExpr::RollbackMigration(_)
259            | QueryExpr::ExplainMigration(_)
260            | QueryExpr::EventsBackfill(_)
261            | QueryExpr::EventsBackfillStatus { .. } => Err(ExecutionError::new(
262                "DML/DDL/Command statements are not supported in UnifiedExecutor",
263            )),
264        }
265    }
266
267    /// Execute a graph query on a specific graph reference
268    ///
269    /// Runtime `MATCH` execution uses the passed `graph` because the
270    /// runtime materialises a fresh `GraphStore` per query, so
271    /// `self.graph` is empty here.
272    fn exec_graph_on(
273        &self,
274        graph: &GraphStore,
275        query: &GraphQuery,
276    ) -> Result<UnifiedResult, ExecutionError> {
277        let mut result = UnifiedResult::empty();
278        let mut stats = QueryStats::default();
279        let effective_filter = effective_graph_filter(query);
280        let effective_projections = effective_graph_projections(query);
281
282        let matches = self.match_pattern_on(graph, &query.pattern, &mut stats)?;
283
284        for matched in matches {
285            if Self::graph_limit_reached(result.records.len(), query.limit) {
286                break;
287            }
288            if !self.eval_filter_on_match(&effective_filter, &matched) {
289                continue;
290            }
291            let record = self.project_match(&matched, &effective_projections);
292            result.records.push(record);
293        }
294
295        result.stats = stats;
296        Ok(result)
297    }
298
299    /// Execute a path query on a specific graph reference
300    fn exec_path_on(
301        &self,
302        graph: &GraphStore,
303        query: &PathQuery,
304    ) -> Result<UnifiedResult, ExecutionError> {
305        let mut result = UnifiedResult::empty();
306
307        // BFS to find paths
308        let mut queue: VecDeque<(String, GraphPath)> = VecDeque::new();
309        let mut visited: HashSet<String> = HashSet::new();
310
311        // Get start node IDs from selector
312        let start_ids = self.resolve_selector_on(graph, &query.from);
313
314        for start in start_ids {
315            queue.push_back((start.clone(), GraphPath::start(&start)));
316            visited.insert(start);
317        }
318
319        let target_ids: HashSet<_> = self
320            .resolve_selector_on(graph, &query.to)
321            .into_iter()
322            .collect();
323        let max_len = query.max_length as usize;
324
325        while let Some((current, path)) = queue.pop_front() {
326            if path.len() > max_len {
327                continue;
328            }
329
330            if target_ids.contains(&current) && !path.is_empty() {
331                let mut record = UnifiedRecord::new();
332                record.paths.push(path.clone());
333                result.records.push(record);
334                continue;
335            }
336
337            // Expand to neighbors
338            for (edge_type, neighbor, weight) in graph.outgoing_edges(&current) {
339                // Check via filter — strings, compared against the legacy
340                // edge enum's canonical name.
341                if !query.via.is_empty() && !query.via.iter().any(|via| via == edge_type.as_str()) {
342                    continue;
343                }
344
345                if !visited.contains(&neighbor) {
346                    visited.insert(neighbor.clone());
347                    let edge = MatchedEdge::from_tuple(&current, edge_type, &neighbor, weight);
348                    let new_path = path.extend(edge, &neighbor);
349                    queue.push_back((neighbor, new_path));
350                }
351            }
352        }
353
354        result.stats.edges_scanned = visited.len() as u64;
355        Ok(result)
356    }
357
358    /// Resolve a node selector to IDs on a specific graph
359    fn resolve_selector_on(&self, graph: &GraphStore, selector: &NodeSelector) -> Vec<String> {
360        match selector {
361            NodeSelector::ById(id) => vec![id.clone()],
362            NodeSelector::ByType {
363                node_label,
364                filter: _,
365            } => graph
366                .nodes_with_category(node_label)
367                .into_iter()
368                .map(|n| n.id)
369                .collect(),
370            NodeSelector::ByRow { table, row_id } => {
371                if let Some((table_id, row_id)) = match (table.as_str().parse::<u16>(), *row_id) {
372                    (Ok(table_id), row_id) => Some((table_id, row_id)),
373                    _ => None,
374                } {
375                    let mut ids = Vec::new();
376
377                    // Fast path: query the bidirectional graph-table index first
378                    if let Some(node_id) = self.index.get_node_for_row(table_id, row_id) {
379                        ids.push(node_id);
380                    }
381
382                    // Fallback path: for callers that don't register index mappings yet,
383                    // scan graph nodes directly by table_ref row linkage.
384                    if ids.is_empty() {
385                        ids.extend(graph.iter_nodes().filter_map(|node| {
386                            let table_ref = node.table_ref?;
387                            if table_ref.table_id == table_id && table_ref.row_id == row_id {
388                                Some(node.id)
389                            } else {
390                                None
391                            }
392                        }));
393                    }
394
395                    ids
396                } else {
397                    Vec::new()
398                }
399            }
400        }
401    }
402
403    /// Execute a query
404    pub fn execute(&self, query: &QueryExpr) -> Result<UnifiedResult, ExecutionError> {
405        match query {
406            QueryExpr::Table(q) => self.exec_table(q),
407            QueryExpr::Graph(q) => self.exec_graph(q),
408            QueryExpr::Join(q) => self.exec_join(q),
409            QueryExpr::Path(q) => self.exec_path(q),
410            QueryExpr::Vector(_) => {
411                // Vector execution requires VectorStore integration
412                // This will be implemented in the VectorExecutor
413                Err(ExecutionError::new(
414                    "Vector queries not yet implemented in UnifiedExecutor",
415                ))
416            }
417            QueryExpr::Hybrid(_) => {
418                // Hybrid execution requires both structured and vector execution
419                // This will be implemented in the HybridExecutor
420                Err(ExecutionError::new(
421                    "Hybrid queries not yet implemented in UnifiedExecutor",
422                ))
423            }
424            QueryExpr::Insert(_)
425            | QueryExpr::Update(_)
426            | QueryExpr::Delete(_)
427            | QueryExpr::CreateTable(_)
428            | QueryExpr::CreateCollection(_)
429            | QueryExpr::CreateVector(_)
430            | QueryExpr::DropTable(_)
431            | QueryExpr::DropGraph(_)
432            | QueryExpr::DropVector(_)
433            | QueryExpr::DropDocument(_)
434            | QueryExpr::DropKv(_)
435            | QueryExpr::DropCollection(_)
436            | QueryExpr::Truncate(_)
437            | QueryExpr::AlterTable(_)
438            | QueryExpr::CreateVcsRef(_)
439            | QueryExpr::DropVcsRef(_)
440            | QueryExpr::VcsCommand(_)
441            | QueryExpr::GraphCommand(_)
442            | QueryExpr::SearchCommand(_)
443            | QueryExpr::CreateIndex(_)
444            | QueryExpr::DropIndex(_)
445            | QueryExpr::ProbabilisticCommand(_)
446            | QueryExpr::Ask(_)
447            | QueryExpr::SetConfig { .. }
448            | QueryExpr::ShowConfig { .. }
449            | QueryExpr::SetSecret { .. }
450            | QueryExpr::DeleteSecret { .. }
451            | QueryExpr::SetKv { .. }
452            | QueryExpr::DeleteKv { .. }
453            | QueryExpr::ShowSecrets { .. }
454            | QueryExpr::SetTenant(_)
455            | QueryExpr::ShowTenant
456            | QueryExpr::CreateTimeSeries(_)
457            | QueryExpr::CreateMetric(_)
458            | QueryExpr::AlterMetric(_)
459            | QueryExpr::CreateSlo(_)
460            | QueryExpr::DropTimeSeries(_)
461            | QueryExpr::CreateQueue(_)
462            | QueryExpr::AlterQueue(_)
463            | QueryExpr::DropQueue(_)
464            | QueryExpr::QueueSelect(_)
465            | QueryExpr::QueueCommand(_)
466            | QueryExpr::KvCommand(_)
467            | QueryExpr::ConfigCommand(_)
468            | QueryExpr::CreateTree(_)
469            | QueryExpr::DropTree(_)
470            | QueryExpr::TreeCommand(_)
471            | QueryExpr::ExplainAlter(_)
472            | QueryExpr::TransactionControl(_)
473            | QueryExpr::MaintenanceCommand(_)
474            | QueryExpr::CreateSchema(_)
475            | QueryExpr::DropSchema(_)
476            | QueryExpr::CreateSequence(_)
477            | QueryExpr::DropSequence(_)
478            | QueryExpr::CopyFrom(_)
479            | QueryExpr::CreateView(_)
480            | QueryExpr::DropView(_)
481            | QueryExpr::RefreshMaterializedView(_)
482            | QueryExpr::CreatePolicy(_)
483            | QueryExpr::DropPolicy(_)
484            | QueryExpr::CreateServer(_)
485            | QueryExpr::DropServer(_)
486            | QueryExpr::CreateForeignTable(_)
487            | QueryExpr::DropForeignTable(_)
488            | QueryExpr::Grant(_)
489            | QueryExpr::Revoke(_)
490            | QueryExpr::AlterUser(_)
491            | QueryExpr::CreateUser(_)
492            | QueryExpr::CreateIamPolicy { .. }
493            | QueryExpr::DropIamPolicy { .. }
494            | QueryExpr::AttachPolicy { .. }
495            | QueryExpr::DetachPolicy { .. }
496            | QueryExpr::ShowPolicies { .. }
497            | QueryExpr::ShowEffectivePermissions { .. }
498            | QueryExpr::RankOf(_)
499            | QueryExpr::ApproxRankOf(_)
500            | QueryExpr::RankRange(_)
501            | QueryExpr::SimulatePolicy { .. }
502            | QueryExpr::LintPolicy { .. }
503            | QueryExpr::MigratePolicyMode { .. }
504            | QueryExpr::CreateMigration(_)
505            | QueryExpr::ApplyMigration(_)
506            | QueryExpr::RollbackMigration(_)
507            | QueryExpr::ExplainMigration(_)
508            | QueryExpr::EventsBackfill(_)
509            | QueryExpr::EventsBackfillStatus { .. } => Err(ExecutionError::new(
510                "DML/DDL/Command statements are not supported in UnifiedExecutor",
511            )),
512        }
513    }
514
515    /// Execute a table query
516    /// Note: Without actual table storage access, this returns empty result.
517    /// In production, this would integrate with the table storage engine.
518    fn exec_table(&self, _query: &TableQuery) -> Result<UnifiedResult, ExecutionError> {
519        // Table execution requires table storage integration
520        // For now, return empty result
521        Ok(UnifiedResult::empty())
522    }
523
524    /// Execute a graph query
525    fn exec_graph(&self, query: &GraphQuery) -> Result<UnifiedResult, ExecutionError> {
526        let mut result = UnifiedResult::empty();
527        let mut stats = QueryStats::default();
528
529        // Match the pattern
530        let matches = self.match_pattern(&query.pattern, &mut stats)?;
531
532        let effective_filter = effective_graph_filter(query);
533        let effective_projections = effective_graph_projections(query);
534
535        for matched in matches {
536            if Self::graph_limit_reached(result.records.len(), query.limit) {
537                break;
538            }
539            if !self.eval_filter_on_match(&effective_filter, &matched) {
540                continue;
541            }
542            let record = self.project_match(&matched, &effective_projections);
543            result.push(record);
544        }
545
546        result.stats = stats;
547        Ok(result)
548    }
549
550    fn graph_limit_reached(row_count: usize, limit: Option<u64>) -> bool {
551        limit.is_some_and(|limit| row_count as u64 >= limit)
552    }
553
554    /// Match a graph pattern
555    fn match_pattern(
556        &self,
557        pattern: &GraphPattern,
558        stats: &mut QueryStats,
559    ) -> Result<Vec<PatternMatch>, ExecutionError> {
560        self.match_pattern_on(self.graph.as_ref(), pattern, stats)
561    }
562
563    fn match_pattern_on(
564        &self,
565        graph: &GraphStore,
566        pattern: &GraphPattern,
567        stats: &mut QueryStats,
568    ) -> Result<Vec<PatternMatch>, ExecutionError> {
569        if pattern.nodes.is_empty() {
570            return Ok(Vec::new());
571        }
572
573        // Start with first node pattern
574        let first = &pattern.nodes[0];
575        let mut matches = self.find_matching_nodes_on(graph, first, stats)?;
576
577        // Extend matches for each edge pattern
578        for edge_pattern in &pattern.edges {
579            matches =
580                self.extend_matches_on(graph, matches, edge_pattern, &pattern.nodes, stats)?;
581        }
582
583        Ok(matches)
584    }
585
586    /// Find nodes matching a pattern
587    fn find_matching_nodes_on(
588        &self,
589        graph: &GraphStore,
590        pattern: &NodePattern,
591        stats: &mut QueryStats,
592    ) -> Result<Vec<PatternMatch>, ExecutionError> {
593        let mut matches = Vec::new();
594
595        // Iterate through all nodes
596        for node in graph.iter_nodes() {
597            stats.nodes_scanned += 1;
598
599            // Check label filter (resolved against the graph's registry).
600            if let Some(ref expected) = pattern.node_label {
601                let expected_id = graph.registry.lookup(Namespace::Node, expected);
602                match expected_id {
603                    Some(id) if id == node.label_id => {}
604                    _ => continue,
605                }
606            }
607
608            // Check property filters
609            let mut match_props = true;
610            for prop_filter in &pattern.properties {
611                if !self.eval_node_property_filter(&node, prop_filter) {
612                    match_props = false;
613                    break;
614                }
615            }
616
617            if match_props {
618                let mut pm = PatternMatch::new();
619                pm.nodes
620                    .insert(pattern.alias.clone(), self.matched_node(&node));
621                matches.push(pm);
622            }
623        }
624
625        Ok(matches)
626    }
627
628    /// Extend matches by following an edge pattern
629    fn extend_matches_on(
630        &self,
631        graph: &GraphStore,
632        matches: Vec<PatternMatch>,
633        edge_pattern: &EdgePattern,
634        node_patterns: &[NodePattern],
635        stats: &mut QueryStats,
636    ) -> Result<Vec<PatternMatch>, ExecutionError> {
637        let mut extended = Vec::new();
638
639        // Find the target node pattern
640        let target_pattern = node_patterns
641            .iter()
642            .find(|n| n.alias == edge_pattern.to)
643            .ok_or_else(|| {
644                ExecutionError::new(format!(
645                    "Node alias '{}' not found in pattern",
646                    edge_pattern.to
647                ))
648            })?;
649
650        for pm in matches {
651            // Get the source node
652            let source_node = pm.nodes.get(&edge_pattern.from).ok_or_else(|| {
653                ExecutionError::new(format!(
654                    "Source node '{}' not found in match",
655                    edge_pattern.from
656                ))
657            })?;
658
659            // Get adjacent edges. The tuple is (edge_label, peer_id, weight,
660            // is_outgoing) — we union outgoing and incoming sets and tag
661            // direction so downstream filters know which side the edge came from.
662            let edges: Vec<_> = match edge_pattern.direction {
663                EdgeDirection::Outgoing => {
664                    graph
665                        .outgoing_edges(&source_node.id)
666                        .into_iter()
667                        .map(|(et, target, w)| (et, target, w, true)) // is_outgoing = true
668                        .collect()
669                }
670                EdgeDirection::Incoming => {
671                    graph
672                        .incoming_edges(&source_node.id)
673                        .into_iter()
674                        .map(|(et, source, w)| (et, source, w, false)) // is_outgoing = false
675                        .collect()
676                }
677                EdgeDirection::Both => {
678                    let mut all: Vec<_> = graph
679                        .outgoing_edges(&source_node.id)
680                        .into_iter()
681                        .map(|(et, target, w)| (et, target, w, true))
682                        .collect();
683                    all.extend(
684                        graph
685                            .incoming_edges(&source_node.id)
686                            .into_iter()
687                            .map(|(et, source, w)| (et, source, w, false)),
688                    );
689                    all
690                }
691            };
692
693            for (etype, other_id, weight, is_outgoing) in edges {
694                stats.edges_scanned += 1;
695
696                // Check edge label filter — direct string compare against the
697                // canonical label that the storage layer hands back.
698                if let Some(ref expected) = edge_pattern.edge_label {
699                    if etype.as_str() != expected.as_str() {
700                        continue;
701                    }
702                }
703
704                // The target is the other node
705                let target_id = &other_id;
706
707                if let Some(target_node) = graph.get_node(target_id) {
708                    // Check target node label filter (registry resolution).
709                    if let Some(ref expected) = target_pattern.node_label {
710                        let expected_id = graph.registry.lookup(Namespace::Node, expected);
711                        match expected_id {
712                            Some(id) if id == target_node.label_id => {}
713                            _ => continue,
714                        }
715                    }
716
717                    // Check target property filters
718                    let mut match_props = true;
719                    for prop_filter in &target_pattern.properties {
720                        if !self.eval_node_property_filter(&target_node, prop_filter) {
721                            match_props = false;
722                            break;
723                        }
724                    }
725
726                    if match_props {
727                        let mut new_pm = pm.clone();
728                        new_pm.nodes.insert(
729                            target_pattern.alias.clone(),
730                            self.matched_node(&target_node),
731                        );
732                        if let Some(ref alias) = edge_pattern.alias {
733                            // Create edge with proper from/to direction
734                            let edge = if is_outgoing {
735                                self.matched_edge(&source_node.id, &etype, target_id, weight)
736                            } else {
737                                self.matched_edge(target_id, &etype, &source_node.id, weight)
738                            };
739                            new_pm.edges.insert(alias.clone(), edge);
740                        }
741                        extended.push(new_pm);
742                    }
743                }
744            }
745        }
746
747        Ok(extended)
748    }
749
750    /// Evaluate a property filter on a stored node
751    fn eval_node_property_filter(
752        &self,
753        node: &StoredNode,
754        filter: &crate::storage::query::ast::PropertyFilter,
755    ) -> bool {
756        let Some(value) = self.node_property_value(node, filter.name.as_str()) else {
757            return false;
758        };
759
760        self.compare_values(&value, &filter.op, &filter.value)
761    }
762
763    /// Compare two values with an operator
764    fn compare_values(&self, left: &Value, op: &CompareOp, right: &Value) -> bool {
765        match op {
766            CompareOp::Eq => left == right,
767            CompareOp::Ne => left != right,
768            CompareOp::Lt => self.value_lt(left, right),
769            CompareOp::Le => self.value_lt(left, right) || left == right,
770            CompareOp::Gt => self.value_lt(right, left),
771            CompareOp::Ge => self.value_lt(right, left) || left == right,
772        }
773    }
774
775    /// Less-than comparison for values
776    fn value_lt(&self, left: &Value, right: &Value) -> bool {
777        match (left, right) {
778            (Value::Integer(a), Value::Integer(b)) => a < b,
779            (Value::Float(a), Value::Float(b)) => a < b,
780            (Value::Integer(a), Value::Float(b)) => (*a as f64) < *b,
781            (Value::Float(a), Value::Integer(b)) => *a < (*b as f64),
782            (Value::Text(a), Value::Text(b)) => a < b,
783            (Value::Timestamp(a), Value::Timestamp(b)) => a < b,
784            _ => false,
785        }
786    }
787
788    /// Evaluate a filter on a pattern match
789    fn eval_filter_on_match(&self, filter: &Option<Filter>, matched: &PatternMatch) -> bool {
790        match filter {
791            None => true,
792            Some(f) => self.eval_filter(f, matched),
793        }
794    }
795
796    /// Evaluate a filter expression
797    fn eval_filter(&self, filter: &Filter, matched: &PatternMatch) -> bool {
798        match filter {
799            Filter::Compare { field, op, value } => {
800                let actual = self.get_field_value(field, matched);
801                match actual {
802                    Some(v) => self.compare_values(&v, op, value),
803                    None => false,
804                }
805            }
806            Filter::CompareFields { left, op, right } => {
807                let l = self.get_field_value(left, matched);
808                let r = self.get_field_value(right, matched);
809                match (l, r) {
810                    (Some(lv), Some(rv)) => self.compare_values(&lv, op, &rv),
811                    _ => false,
812                }
813            }
814            Filter::CompareExpr { .. } => {
815                // The unified graph-level executor doesn't yet carry
816                // the `UnifiedRecord` context that expr_eval needs.
817                // Return false (conservative — the predicate is
818                // treated as unmatched) until the executor is
819                // upgraded in Week 5.
820                false
821            }
822            Filter::And(left, right) => {
823                self.eval_filter(left, matched) && self.eval_filter(right, matched)
824            }
825            Filter::Or(left, right) => {
826                self.eval_filter(left, matched) || self.eval_filter(right, matched)
827            }
828            Filter::Not(inner) => !self.eval_filter(inner, matched),
829            Filter::IsNull(field) => self.get_field_value(field, matched).is_none(),
830            Filter::IsNotNull(field) => self.get_field_value(field, matched).is_some(),
831            Filter::In { field, values } => match self.get_field_value(field, matched) {
832                Some(v) => values.contains(&v),
833                None => false,
834            },
835            Filter::Between { field, low, high } => match self.get_field_value(field, matched) {
836                Some(v) => !self.value_lt(&v, low) && !self.value_lt(high, &v),
837                None => false,
838            },
839            Filter::Like { field, pattern } => match self.get_field_value(field, matched) {
840                Some(Value::Text(s)) => self.match_like(&s, pattern),
841                _ => false,
842            },
843            Filter::StartsWith { field, prefix } => match self.get_field_value(field, matched) {
844                Some(Value::Text(s)) => s.starts_with(prefix),
845                _ => false,
846            },
847            Filter::EndsWith { field, suffix } => match self.get_field_value(field, matched) {
848                Some(Value::Text(s)) => s.ends_with(suffix),
849                _ => false,
850            },
851            Filter::Contains { field, substring } => match self.get_field_value(field, matched) {
852                Some(Value::Text(s)) => s.contains(substring),
853                _ => false,
854            },
855        }
856    }
857
858    /// Simple LIKE pattern matching (% and _ wildcards)
859    fn match_like(&self, text: &str, pattern: &str) -> bool {
860        // Simple implementation: convert % to .* and _ to .
861        let regex_pattern = pattern.replace('%', ".*").replace('_', ".");
862
863        // Basic match without regex (for simplicity)
864        if pattern.starts_with('%') && pattern.ends_with('%') {
865            let inner = &pattern[1..pattern.len() - 1];
866            text.contains(inner)
867        } else if let Some(suffix) = pattern.strip_prefix('%') {
868            text.ends_with(suffix)
869        } else if let Some(prefix) = pattern.strip_suffix('%') {
870            text.starts_with(prefix)
871        } else {
872            text == pattern || regex_pattern == text
873        }
874    }
875
876    /// Get a field value from a pattern match
877    fn get_field_value(&self, field: &FieldRef, matched: &PatternMatch) -> Option<Value> {
878        match field {
879            FieldRef::NodeId { alias } => {
880                matched.nodes.get(alias).map(|n| Value::text(n.id.clone()))
881            }
882            FieldRef::NodeProperty { alias, property } => matched
883                .nodes
884                .get(alias)
885                .and_then(|n| match property.as_str() {
886                    "id" => Some(Value::text(n.id.clone())),
887                    "label" => Some(Value::text(n.label.clone())),
888                    "type" | "node_type" => Some(Value::text(n.node_label.clone())),
889                    _ => n.properties.get(property).cloned(),
890                })
891                .or_else(|| {
892                    matched
893                        .edges
894                        .get(alias)
895                        .and_then(|e| Self::edge_property_value(e, property))
896                }),
897            FieldRef::EdgeProperty { alias, property } => matched
898                .edges
899                .get(alias)
900                .and_then(|e| Self::edge_property_value(e, property)),
901            FieldRef::TableColumn { table, column } => {
902                // The shared SQL `WHERE` parser emits `n.foo` as
903                // `TableColumn { table: "n", column: "foo" }` — for
904                // graph MATCH we want to treat that as a node-property
905                // lookup against the matched alias when one exists,
906                // so `MATCH (n) WHERE n.label = 'x'` actually filters.
907                if !table.is_empty() {
908                    if let Some(n) = matched.nodes.get(table) {
909                        return match column.as_str() {
910                            "id" => Some(Value::text(n.id.clone())),
911                            "label" => Some(Value::text(n.label.clone())),
912                            "type" | "node_type" => Some(Value::text(n.node_label.clone())),
913                            other => n.properties.get(other).cloned(),
914                        };
915                    }
916                    if let Some(e) = matched.edges.get(table) {
917                        return Self::edge_property_value(e, column);
918                    }
919                }
920                None
921            }
922        }
923    }
924
925    fn edge_property_value(edge: &MatchedEdge, property: &str) -> Option<Value> {
926        match property {
927            "weight" => Some(Value::Float(edge.weight as f64)),
928            "from" | "source" => Some(Value::text(edge.from.clone())),
929            "to" | "target" => Some(Value::text(edge.to.clone())),
930            "label" | "type" | "edge_type" => Some(Value::text(edge.edge_label.clone())),
931            other => edge.properties.get(other).cloned(),
932        }
933    }
934
935    /// Get a value for join condition
936    fn get_join_value(&self, field: &FieldRef, record: &UnifiedRecord) -> Option<Value> {
937        match field {
938            FieldRef::TableColumn { column, .. } => record.get(column.as_str()).cloned(),
939            FieldRef::NodeId { alias } => record
940                .nodes
941                .get(alias)
942                .map(|node| Value::text(node.id.clone())),
943            FieldRef::NodeProperty { alias, property } => {
944                record
945                    .nodes
946                    .get(alias)
947                    .and_then(|n| match property.as_str() {
948                        "id" => Some(Value::text(n.id.clone())),
949                        "label" => Some(Value::text(n.label.clone())),
950                        "type" | "node_type" => Some(Value::text(n.node_label.clone())),
951                        _ => n.properties.get(property).cloned(),
952                    })
953            }
954            FieldRef::EdgeProperty { alias, property } => {
955                record
956                    .edges
957                    .get(alias)
958                    .and_then(|e| match property.as_str() {
959                        "weight" => Some(Value::Float(e.weight as f64)),
960                        "from" | "source" => Some(Value::text(e.from.clone())),
961                        "to" | "target" => Some(Value::text(e.to.clone())),
962                        "label" | "type" | "edge_type" => Some(Value::text(e.edge_label.clone())),
963                        other => e.properties.get(other).cloned(),
964                    })
965            }
966        }
967    }
968
969    /// Get an index-agnostic view of matched records for projections
970    fn project_match(&self, matched: &PatternMatch, projections: &[Projection]) -> UnifiedRecord {
971        let mut record = UnifiedRecord::new();
972
973        // Copy all matched nodes and edges
974        record.nodes = matched.nodes.clone();
975        record.edges = matched.edges.clone();
976
977        // Extract projected values
978        for proj in projections {
979            match proj {
980                Projection::Field(field, alias) => {
981                    // `RETURN n` (whole-entity projection) expands into
982                    // every property the match holds for that alias —
983                    // id, label, node_type, plus user properties — so
984                    // callers don't get an empty `{}` row.
985                    if let (FieldRef::NodeId { alias: node_alias }, None) = (field, alias) {
986                        if let Some(node) = matched.nodes.get(node_alias) {
987                            record.set(&format!("{}.id", node_alias), Value::text(node.id.clone()));
988                            record.set(
989                                &format!("{}.label", node_alias),
990                                Value::text(node.label.clone()),
991                            );
992                            record.set(
993                                &format!("{}.node_type", node_alias),
994                                Value::text(node.node_label.clone()),
995                            );
996                            for (k, v) in &node.properties {
997                                record.set(&format!("{}.{}", node_alias, k), v.clone());
998                            }
999                            continue;
1000                        }
1001                        if let Some(edge) = matched.edges.get(node_alias) {
1002                            record.set(
1003                                &format!("{}.from", node_alias),
1004                                Value::text(edge.from.clone()),
1005                            );
1006                            record.set(&format!("{}.to", node_alias), Value::text(edge.to.clone()));
1007                            record.set(
1008                                &format!("{}.label", node_alias),
1009                                Value::text(edge.edge_label.clone()),
1010                            );
1011                            record.set(
1012                                &format!("{}.weight", node_alias),
1013                                Value::Float(edge.weight as f64),
1014                            );
1015                            for (k, v) in &edge.properties {
1016                                record.set(&format!("{}.{}", node_alias, k), v.clone());
1017                            }
1018                            continue;
1019                        }
1020                    }
1021                    if let Some(value) = self.get_field_value(field, matched) {
1022                        let key = alias.clone().unwrap_or_else(|| self.field_to_string(field));
1023                        record.set(&key, value);
1024                    }
1025                }
1026                Projection::All => {
1027                    // For All projection, include all node basic info
1028                    for (alias, node) in &matched.nodes {
1029                        record.set(&format!("{}.id", alias), Value::text(node.id.clone()));
1030                        record.set(&format!("{}.label", alias), Value::text(node.label.clone()));
1031                    }
1032                }
1033                Projection::Column(col) => {
1034                    // Try to find a matching column in nodes
1035                    for node in matched.nodes.values() {
1036                        match col.as_str() {
1037                            "id" => record.set(col, Value::text(node.id.clone())),
1038                            "label" => record.set(col, Value::text(node.label.clone())),
1039                            _ => {}
1040                        }
1041                    }
1042                }
1043                Projection::Alias(col, alias) => {
1044                    for node in matched.nodes.values() {
1045                        match col.as_str() {
1046                            "id" => record.set(alias, Value::text(node.id.clone())),
1047                            "label" => record.set(alias, Value::text(node.label.clone())),
1048                            _ => {}
1049                        }
1050                    }
1051                }
1052                _ => {} // Function and Expression projections not supported yet
1053            }
1054        }
1055
1056        record
1057    }
1058
1059    /// Convert a field reference to a string key
1060    fn field_to_string(&self, field: &FieldRef) -> String {
1061        match field {
1062            FieldRef::NodeId { alias } => format!("{}.id", alias),
1063            FieldRef::NodeProperty { alias, property } => format!("{}.{}", alias, property),
1064            FieldRef::EdgeProperty { alias, property } => format!("{}.{}", alias, property),
1065            FieldRef::TableColumn { table, column } => {
1066                if table.is_empty() {
1067                    column.clone()
1068                } else {
1069                    format!("{}.{}", table, column)
1070                }
1071            }
1072        }
1073    }
1074
1075    /// Execute a join query
1076    fn exec_join(&self, query: &JoinQuery) -> Result<UnifiedResult, ExecutionError> {
1077        // Execute left side
1078        let left_result = self.execute(&query.left)?;
1079
1080        // Execute right side
1081        let right_result = self.execute(&query.right)?;
1082
1083        // Perform the join
1084        let mut result = UnifiedResult::empty();
1085
1086        // For each left record, find matching right records
1087        for left in &left_result.records {
1088            let left_value = self.get_join_value(&query.on.left_field, left);
1089
1090            for right in &right_result.records {
1091                let right_value = self.get_join_value(&query.on.right_field, right);
1092
1093                if left_value == right_value {
1094                    // Merge records
1095                    let mut merged = left.clone();
1096                    merged.nodes.extend(right.nodes.clone());
1097                    merged.edges.extend(right.edges.clone());
1098                    for (k, v) in right.iter_fields() {
1099                        merged.set_arc(k.clone(), v.clone());
1100                    }
1101                    result.push(merged);
1102                }
1103            }
1104
1105            // Handle outer joins
1106            if matches!(query.join_type, JoinType::LeftOuter) {
1107                // If no matches found for this left record, still include it
1108                if !right_result
1109                    .records
1110                    .iter()
1111                    .any(|r| self.get_join_value(&query.on.right_field, r) == left_value)
1112                {
1113                    result.push(left.clone());
1114                }
1115            }
1116        }
1117
1118        Ok(result)
1119    }
1120
1121    /// Execute a path query
1122    fn exec_path(&self, query: &PathQuery) -> Result<UnifiedResult, ExecutionError> {
1123        let mut result = UnifiedResult::empty();
1124        let mut stats = QueryStats::default();
1125
1126        // Find start nodes
1127        let start_nodes = self.resolve_selector(&query.from, &mut stats)?;
1128
1129        // Find target nodes
1130        let target_nodes: HashSet<String> = self
1131            .resolve_selector(&query.to, &mut stats)?
1132            .into_iter()
1133            .collect();
1134
1135        // BFS to find paths
1136        for start_id in start_nodes {
1137            let paths = self.bfs_paths(
1138                &start_id,
1139                &target_nodes,
1140                &query.via,
1141                query.max_length,
1142                &mut stats,
1143            )?;
1144
1145            for path in paths {
1146                // Apply filter if present
1147                if effective_path_filter(query).is_some() {
1148                    // Path filtering would require converting path to match
1149                    // For now, include all paths
1150                }
1151
1152                let mut record = UnifiedRecord::new();
1153                record.paths.push(path);
1154                result.push(record);
1155            }
1156        }
1157
1158        result.stats = stats;
1159        Ok(result)
1160    }
1161
1162    /// Resolve a node selector to node IDs
1163    fn resolve_selector(
1164        &self,
1165        selector: &NodeSelector,
1166        stats: &mut QueryStats,
1167    ) -> Result<Vec<String>, ExecutionError> {
1168        match selector {
1169            NodeSelector::ById(id) => Ok(vec![id.clone()]),
1170            NodeSelector::ByType { node_label, filter } => {
1171                let expected_id = self.graph.registry.lookup(Namespace::Node, node_label);
1172                let mut nodes = Vec::new();
1173                for node in self.graph.iter_nodes() {
1174                    stats.nodes_scanned += 1;
1175                    if expected_id.map(|id| node.label_id == id).unwrap_or(false) {
1176                        let matches_filter = filter
1177                            .as_ref()
1178                            .map(|f| self.eval_node_property_filter(&node, f))
1179                            .unwrap_or(true);
1180                        if matches_filter {
1181                            nodes.push(node.id.clone());
1182                        }
1183                    }
1184                }
1185                Ok(nodes)
1186            }
1187            NodeSelector::ByRow { row_id, .. } => {
1188                // Use graph-table index to find linked node
1189                // For now, try direct lookup with table_id=0
1190                if let Some(node_id) = self.index.get_node_for_row(0, *row_id) {
1191                    Ok(vec![node_id])
1192                } else {
1193                    Ok(Vec::new())
1194                }
1195            }
1196        }
1197    }
1198
1199    /// BFS to find paths between nodes
1200    fn bfs_paths(
1201        &self,
1202        start: &str,
1203        targets: &HashSet<String>,
1204        via: &[String],
1205        max_length: u32,
1206        stats: &mut QueryStats,
1207    ) -> Result<Vec<GraphPath>, ExecutionError> {
1208        let mut paths = Vec::new();
1209        let mut queue: VecDeque<GraphPath> = VecDeque::new();
1210        let mut visited: HashSet<String> = HashSet::new();
1211
1212        queue.push_back(GraphPath::start(start));
1213        visited.insert(start.to_string());
1214
1215        while let Some(current_path) = queue.pop_front() {
1216            let Some(current_node) = current_path.nodes.last() else {
1217                continue;
1218            };
1219
1220            // Check if we've reached a target
1221            if targets.contains(current_node) && !current_path.is_empty() {
1222                paths.push(current_path.clone());
1223                continue;
1224            }
1225
1226            // Don't extend beyond max length
1227            if current_path.len() >= max_length as usize {
1228                continue;
1229            }
1230
1231            // Get outgoing edges (each entry: edge_label, target_id, weight)
1232            for (edge_type, target_id, weight) in self.graph.outgoing_edges(current_node) {
1233                stats.edges_scanned += 1;
1234
1235                // Check edge label filter (string compare against canonical name).
1236                if !via.is_empty() && !via.iter().any(|v| v == edge_type.as_str()) {
1237                    continue;
1238                }
1239
1240                // Skip if already visited (prevent cycles)
1241                if visited.contains(&target_id) {
1242                    continue;
1243                }
1244
1245                let edge = MatchedEdge::from_tuple(current_node, edge_type, &target_id, weight);
1246                let new_path = current_path.extend(edge, &target_id);
1247                visited.insert(target_id.clone());
1248                queue.push_back(new_path);
1249            }
1250        }
1251
1252        Ok(paths)
1253    }
1254}
1255
1256/// Internal pattern match state
1257#[derive(Debug, Clone, Default)]
1258struct PatternMatch {
1259    nodes: HashMap<String, MatchedNode>,
1260    edges: HashMap<String, MatchedEdge>,
1261}
1262
1263impl PatternMatch {
1264    fn new() -> Self {
1265        Self::default()
1266    }
1267}