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::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    /// Execute a graph query on a specific graph reference
273    ///
274    /// Runtime `MATCH` execution uses the passed `graph` because the
275    /// runtime materialises a fresh `GraphStore` per query, so
276    /// `self.graph` is empty here.
277    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    /// Execute a path query on a specific graph reference
305    fn exec_path_on(
306        &self,
307        graph: &GraphStore,
308        query: &PathQuery,
309    ) -> Result<UnifiedResult, ExecutionError> {
310        let mut result = UnifiedResult::empty();
311
312        // BFS to find paths
313        let mut queue: VecDeque<(String, GraphPath)> = VecDeque::new();
314        let mut visited: HashSet<String> = HashSet::new();
315
316        // Get start node IDs from selector
317        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(&current) && !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            // Expand to neighbors
343            for (edge_type, neighbor, weight) in graph.outgoing_edges(&current) {
344                // Check via filter — strings, compared against the legacy
345                // edge enum's canonical name.
346                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(&current, 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    /// Resolve a node selector to IDs on a specific graph
364    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                    // Fast path: query the bidirectional graph-table index first
383                    if let Some(node_id) = self.index.get_node_for_row(table_id, row_id) {
384                        ids.push(node_id);
385                    }
386
387                    // Fallback path: for callers that don't register index mappings yet,
388                    // scan graph nodes directly by table_ref row linkage.
389                    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    /// Execute a query
409    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                // Vector execution requires VectorStore integration
417                // This will be implemented in the VectorExecutor
418                Err(ExecutionError::new(
419                    "Vector queries not yet implemented in UnifiedExecutor",
420                ))
421            }
422            QueryExpr::Hybrid(_) => {
423                // Hybrid execution requires both structured and vector execution
424                // This will be implemented in the HybridExecutor
425                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    /// Execute a table query
526    /// Note: Without actual table storage access, this returns empty result.
527    /// In production, this would integrate with the table storage engine.
528    fn exec_table(&self, _query: &TableQuery) -> Result<UnifiedResult, ExecutionError> {
529        // Table execution requires table storage integration
530        // For now, return empty result
531        Ok(UnifiedResult::empty())
532    }
533
534    /// Execute a graph query
535    fn exec_graph(&self, query: &GraphQuery) -> Result<UnifiedResult, ExecutionError> {
536        let mut result = UnifiedResult::empty();
537        let mut stats = QueryStats::default();
538
539        // Match the pattern
540        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    /// Match a graph pattern
565    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        // Start with first node pattern
584        let first = &pattern.nodes[0];
585        let mut matches = self.find_matching_nodes_on(graph, first, stats)?;
586
587        // Extend matches for each edge pattern
588        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    /// Find nodes matching a pattern
597    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        // Iterate through all nodes
606        for node in graph.iter_nodes() {
607            stats.nodes_scanned += 1;
608
609            // Check label filter (resolved against the graph's registry).
610            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            // Check property filters
619            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    /// Extend matches by following an edge pattern
639    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        // Find the target node pattern
650        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            // Get the source node
662            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            // Get adjacent edges. The tuple is (edge_label, peer_id, weight,
670            // is_outgoing) — we union outgoing and incoming sets and tag
671            // direction so downstream filters know which side the edge came from.
672            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)) // is_outgoing = true
678                        .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)) // is_outgoing = false
685                        .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                // Check edge label filter — direct string compare against the
707                // canonical label that the storage layer hands back.
708                if let Some(ref expected) = edge_pattern.edge_label {
709                    if etype.as_str() != expected.as_str() {
710                        continue;
711                    }
712                }
713
714                // The target is the other node
715                let target_id = &other_id;
716
717                if let Some(target_node) = graph.get_node(target_id) {
718                    // Check target node label filter (registry resolution).
719                    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                    // Check target property filters
728                    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                            // Create edge with proper from/to direction
744                            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    /// Evaluate a property filter on a stored node
761    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    /// Compare two values with an operator
774    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    /// Less-than comparison for values
786    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    /// Evaluate a filter on a pattern match
799    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    /// Evaluate a filter expression
807    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                // The unified graph-level executor doesn't yet carry
826                // the `UnifiedRecord` context that expr_eval needs.
827                // Return false (conservative — the predicate is
828                // treated as unmatched) until the executor is
829                // upgraded in Week 5.
830                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    /// Simple LIKE pattern matching (% and _ wildcards)
869    fn match_like(&self, text: &str, pattern: &str) -> bool {
870        // Simple implementation: convert % to .* and _ to .
871        let regex_pattern = pattern.replace('%', ".*").replace('_', ".");
872
873        // Basic match without regex (for simplicity)
874        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    /// Get a field value from a pattern match
887    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                // The shared SQL `WHERE` parser emits `n.foo` as
913                // `TableColumn { table: "n", column: "foo" }` — for
914                // graph MATCH we want to treat that as a node-property
915                // lookup against the matched alias when one exists,
916                // so `MATCH (n) WHERE n.label = 'x'` actually filters.
917                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    /// Get a value for join condition
946    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    /// Get an index-agnostic view of matched records for projections
980    fn project_match(&self, matched: &PatternMatch, projections: &[Projection]) -> UnifiedRecord {
981        let mut record = UnifiedRecord::new();
982
983        // Copy all matched nodes and edges
984        record.nodes = matched.nodes.clone();
985        record.edges = matched.edges.clone();
986
987        // Extract projected values
988        for proj in projections {
989            match proj {
990                Projection::Field(field, alias) => {
991                    // `RETURN n` (whole-entity projection) expands into
992                    // every property the match holds for that alias —
993                    // id, label, node_type, plus user properties — so
994                    // callers don't get an empty `{}` row.
995                    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 All projection, include all node basic info
1038                    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                    // Try to find a matching column in nodes
1045                    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                _ => {} // Function and Expression projections not supported yet
1063            }
1064        }
1065
1066        record
1067    }
1068
1069    /// Convert a field reference to a string key
1070    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    /// Execute a join query
1086    fn exec_join(&self, query: &JoinQuery) -> Result<UnifiedResult, ExecutionError> {
1087        // Execute left side
1088        let left_result = self.execute(&query.left)?;
1089
1090        // Execute right side
1091        let right_result = self.execute(&query.right)?;
1092
1093        // Perform the join
1094        let mut result = UnifiedResult::empty();
1095
1096        // For each left record, find matching right records
1097        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                    // Merge records
1105                    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            // Handle outer joins
1116            if matches!(query.join_type, JoinType::LeftOuter) {
1117                // If no matches found for this left record, still include it
1118                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    /// Execute a path query
1132    fn exec_path(&self, query: &PathQuery) -> Result<UnifiedResult, ExecutionError> {
1133        let mut result = UnifiedResult::empty();
1134        let mut stats = QueryStats::default();
1135
1136        // Find start nodes
1137        let start_nodes = self.resolve_selector(&query.from, &mut stats)?;
1138
1139        // Find target nodes
1140        let target_nodes: HashSet<String> = self
1141            .resolve_selector(&query.to, &mut stats)?
1142            .into_iter()
1143            .collect();
1144
1145        // BFS to find paths
1146        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                // Apply filter if present
1157                if effective_path_filter(query).is_some() {
1158                    // Path filtering would require converting path to match
1159                    // For now, include all paths
1160                }
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    /// Resolve a node selector to node IDs
1173    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                // Use graph-table index to find linked node
1199                // For now, try direct lookup with table_id=0
1200                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    /// BFS to find paths between nodes
1210    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            // Check if we've reached a target
1231            if targets.contains(current_node) && !current_path.is_empty() {
1232                paths.push(current_path.clone());
1233                continue;
1234            }
1235
1236            // Don't extend beyond max length
1237            if current_path.len() >= max_length as usize {
1238                continue;
1239            }
1240
1241            // Get outgoing edges (each entry: edge_label, target_id, weight)
1242            for (edge_type, target_id, weight) in self.graph.outgoing_edges(current_node) {
1243                stats.edges_scanned += 1;
1244
1245                // Check edge label filter (string compare against canonical name).
1246                if !via.is_empty() && !via.iter().any(|v| v == edge_type.as_str()) {
1247                    continue;
1248                }
1249
1250                // Skip if already visited (prevent cycles)
1251                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/// Internal pattern match state
1267#[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}