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

akar_processor/physical/write_ops/
merge_rel.rs
