akar_processor/physical/write_ops/
merge_rel.rs

1//! Physical operator for edge MERGE (P53.20): `MERGE (a)-[r:R {..}]->(b)`.
2
3use crate::physical::types::{OperatorResult, PhysicalOperatorExec};
4use crate::physical::write_ops::set::{PhysicalSet, evaluate_expression_for_row};
5use akar_common::error::ProcessorError;
6use akar_common::types::{PhysicalTypeID, Value};
7use akar_common::vector::{DataChunk, ValueVector};
8use akar_parser::ast::{EdgeDirection, Expression};
9use akar_storage::table::TableCatalog;
10use akar_transaction::UndoRecord;
11use std::sync::{Arc, Mutex};
12
13/// Physical operator for MERGE on an edge pattern.
14///
15/// For each input row (a bound `src`/`dst` node pair, resolved via the
16/// `<src>._id` / `<dst>._id` columns), it matches an existing edge on the rel
17/// table whose endpoints and pattern properties match. If none exists, a new
18/// edge is inserted. Emits one output column `<edge_var>._id` carrying the
19/// matched/inserted edge index per row, so a following `SET <edge_var>.x`
20/// clause can target the right rows.
21pub struct PhysicalMergeRel {
22    pub rel_table_name: String,
23    pub rel_table_id: u64,
24    pub edge_var: String,
25    pub src_node_var: String,
26    pub dst_node_var: String,
27    pub direction: EdgeDirection,
28    pub properties: Vec<(String, Expression)>,
29    pub on_match: Vec<PhysicalSet>,
30    pub on_create: Vec<PhysicalSet>,
31    pub table_catalog: Arc<TableCatalog>,
32    /// Active transaction id (P52.18).
33    pub txn_id: Option<u64>,
34    /// Undo sink for rollback records (P52.18).
35    pub undo_sink: Option<Arc<Mutex<Vec<UndoRecord>>>>,
36}
37
38impl PhysicalOperatorExec for PhysicalMergeRel {
39    fn operator_type(&self) -> &str {
40        "merge_rel"
41    }
42
43    fn execute(&self, input: Vec<DataChunk>) -> OperatorResult {
44        if input.is_empty() || input.iter().all(|c| c.size == 0) {
45            return Ok(input);
46        }
47
48        // Owned snapshot of the rel table adjacency + columns (P53.14 style).
49        let (fwd_adj, rev_adj, cols) = {
50            let rel_table = self
51                .table_catalog
52                .get_rel_table_by_name(&self.rel_table_name)
53                .ok_or_else(|| format!("Rel table '{}' not found", self.rel_table_name))?;
54            (
55                rel_table.fwd_adj.clone(),
56                rel_table.rev_adj.clone(),
57                rel_table.columns.clone(),
58            )
59        };
60
61        let mut output = Vec::with_capacity(input.len());
62
63        for chunk in input {
64            let src_col = chunk
65                .field_names
66                .iter()
67                .position(|n| n == &format!("{}.{}", self.src_node_var, "_id"))
68                .ok_or_else(|| {
69                    format!(
70                        "Bound node '{}' not found in MERGE input. Available fields: {:?}",
71                        self.src_node_var, chunk.field_names
72                    )
73                })?;
74            let dst_col = chunk
75                .field_names
76                .iter()
77                .position(|n| n == &format!("{}.{}", self.dst_node_var, "_id"))
78                .ok_or_else(|| {
79                    format!(
80                        "Bound node '{}' not found in MERGE input. Available fields: {:?}",
81                        self.dst_node_var, chunk.field_names
82                    )
83                })?;
84
85            let mut matched_idx: Vec<Option<u64>> = Vec::with_capacity(chunk.size);
86            let mut matched: Vec<u64> = Vec::new();
87            let mut created: Vec<u64> = Vec::new();
88
89            for row in 0..chunk.size {
90                let src_id = match chunk.get_value(src_col, row) {
91                    Some(Value::Int64(v)) => v as u64,
92                    _ => continue,
93                };
94                let dst_id = match chunk.get_value(dst_col, row) {
95                    Some(Value::Int64(v)) => v as u64,
96                    _ => continue,
97                };
98
99                let candidates: Vec<(u64, usize)> = match self.direction {
100                    EdgeDirection::LeftToRight => fwd_adj.get(&src_id).cloned().unwrap_or_default(),
101                    EdgeDirection::RightToLeft => rev_adj.get(&src_id).cloned().unwrap_or_default(),
102                    EdgeDirection::Both => {
103                        let mut all = fwd_adj.get(&src_id).cloned().unwrap_or_default();
104                        if let Some(rev) = rev_adj.get(&src_id) {
105                            all.extend(rev.iter().cloned());
106                        }
107                        all
108                    }
109                };
110
111                let mut found: Option<usize> = None;
112                for &(dst_offset, edge_idx) in &candidates {
113                    if dst_offset != dst_id {
114                        continue;
115                    }
116                    if self.props_match(edge_idx, &cols, &chunk, row)? {
117                        found = Some(edge_idx);
118                        break;
119                    }
120                }
121
122                match found {
123                    Some(edge_idx) => {
124                        matched.push(edge_idx as u64);
125                        matched_idx.push(Some(edge_idx as u64));
126                    }
127                    None => {
128                        let mut values: Vec<Value> = vec![Value::Null; cols.len()];
129                        for (prop_name, prop_expr) in &self.properties {
130                            if let Some(col_idx) = cols.iter().position(|c| c.name == *prop_name) {
131                                values[col_idx] = evaluate_expression_for_row(prop_expr, &chunk, row);
132                            }
133                        }
134                        let edge_idx = self.insert_rel(src_id, dst_id, values)?;
135                        created.push(edge_idx);
136                        matched_idx.push(Some(edge_idx));
137                    }
138                }
139            }
140
141            // Emit `<edge_var>._id` so a following SET targets these edges.
142            let edge_count = matched_idx.len();
143            let mut v = ValueVector::new(PhysicalTypeID::Int64, edge_count);
144            v.resize(edge_count);
145            for (i, e) in matched_idx.iter().enumerate() {
146                if let Some(idx) = e {
147                    v.set_i64(i, *idx as i64);
148                } else {
149                    v.set_null(i, true);
150                }
151            }
152            let arr = akar_common::arrow_vector::ArrowVector::from_legacy(&v).array;
153            let out = DataChunk::new(vec![arr], vec![PhysicalTypeID::Int64])
154                .with_names(vec![format!("{}.{}", self.edge_var, "_id")]);
155            output.push(out);
156
157            if !matched.is_empty() {
158                self.apply_on_clause(&self.on_match, &matched)?;
159            }
160            if !created.is_empty() {
161                self.apply_on_clause(&self.on_create, &created)?;
162            }
163        }
164
165        Ok(output)
166    }
167}
168
169impl PhysicalMergeRel {
170    /// Evaluate the pattern's inline properties against the row and compare
171    /// them with the candidate edge's stored property values.
172    fn props_match(
173        &self,
174        edge_idx: usize,
175        cols: &[akar_storage::table::ColumnDefinition],
176        chunk: &DataChunk,
177        row: usize,
178    ) -> Result<bool, ProcessorError> {
179        let rel_table = self
180            .table_catalog
181            .get_rel_table_by_name(&self.rel_table_name)
182            .ok_or_else(|| format!("Rel table '{}' not found", self.rel_table_name))?;
183        let props = rel_table.get_edge_properties(edge_idx);
184        for (prop_name, prop_expr) in &self.properties {
185            let col_idx = cols
186                .iter()
187                .position(|c| c.name == *prop_name)
188                .ok_or_else(|| format!("Rel column '{prop_name}' not found"))?;
189            let expected = evaluate_expression_for_row(prop_expr, chunk, row);
190            let actual = props.get(col_idx).cloned().unwrap_or(Value::Null);
191            if expected != actual {
192                return Ok(false);
193            }
194        }
195        Ok(true)
196    }
197
198    /// Insert a new edge, recording an undo record.
199    fn insert_rel(&self, src_id: u64, dst_id: u64, values: Vec<Value>) -> Result<u64, ProcessorError> {
200        let (edge_idx, table_id) = {
201            let mut rel_table = self
202                .table_catalog
203                .get_rel_table_by_name_mut(&self.rel_table_name)
204                .ok_or_else(|| format!("Rel table '{}' not found", self.rel_table_name))?;
205            let edge_idx = rel_table.edges.len() as u64;
206            rel_table
207                .insert_rel(src_id, dst_id, values)
208                .map_err(|e| format!("MERGE CREATE edge failed: {e}"))?;
209            (edge_idx, rel_table.table_id)
210        };
211        if let Some(sink) = self.undo_sink.as_ref()
212            && let Ok(mut u) = sink.lock()
213        {
214            u.push(UndoRecord::insert(table_id, edge_idx));
215        }
216        Ok(edge_idx)
217    }
218
219    /// Apply `ON MATCH SET` / `ON CREATE SET` operations against the given
220    /// edge indices. Each SET op reads the `<edge_var>._id` column, so only
221    /// the edge rows emitted by this merge are touched.
222    fn apply_on_clause(&self, set_ops: &[PhysicalSet], edge_ids: &[u64]) -> Result<(), ProcessorError> {
223        if set_ops.is_empty() || edge_ids.is_empty() {
224            return Ok(());
225        }
226        let n = edge_ids.len();
227        let mut v = ValueVector::new(PhysicalTypeID::Int64, n);
228        v.resize(n);
229        for (i, e) in edge_ids.iter().enumerate() {
230            v.set_i64(i, *e as i64);
231        }
232        let arr = akar_common::arrow_vector::ArrowVector::from_legacy(&v).array;
233        let chunk = DataChunk::new(vec![arr], vec![PhysicalTypeID::Int64])
234            .with_names(vec![format!("{}.{}", self.edge_var, "_id")]);
235        for set_op in set_ops {
236            let _ = set_op.execute(vec![chunk.clone()])?;
237        }
238        Ok(())
239    }
240}
akar_processor/physical/write_ops/merge_rel.rs

akar_processor/physical/write_ops/
merge_rel.rs
