akar_processor/physical/write_ops/
merge.rs

1//! Physical operator for MERGE.
2
3use crate::physical::types::{OperatorResult, PhysicalOperatorExec};
4use crate::physical::write_ops::set::{PhysicalSet, append_pipeline_columns};
5use akar_common::types::{PhysicalTypeID, Value};
6use akar_common::vector::{DataChunk, ValueVector};
7use akar_storage::table::TableCatalog;
8use akar_transaction::UndoRecord;
9use std::sync::{Arc, Mutex};
10
11/// Physical operator for MERGE.
12/// Represents a combination of MATCH and INSERT (Upsert).
13pub struct PhysicalMerge {
14    pub table_name: String,
15    pub table_id: u64,
16    pub properties: Vec<(String, akar_parser::ast::Expression)>,
17    pub on_match: Vec<PhysicalSet>,
18    pub on_create: Vec<PhysicalSet>,
19    pub table_catalog: Arc<TableCatalog>,
20    /// Active transaction id (P52.18).
21    pub txn_id: Option<u64>,
22    /// Undo sink for rollback records (P52.18).
23    pub undo_sink: Option<Arc<Mutex<Vec<UndoRecord>>>>,
24}
25
26/// Build a single-column chunk carrying physical row indices under the `_id`
27/// pseudo-column name, so a `PhysicalSet` passed as ON MATCH / ON CREATE can
28/// re-target exactly those rows.
29fn row_id_chunk(row_ids: &[u64]) -> DataChunk {
30    let mut v = ValueVector::new(PhysicalTypeID::Int64, row_ids.len());
31    v.resize(row_ids.len());
32    for (i, r) in row_ids.iter().enumerate() {
33        v.set_i64(i, *r as i64);
34    }
35    let arr = akar_common::arrow_vector::ArrowVector::from_legacy(&v).array;
36    DataChunk::new(vec![arr], vec![PhysicalTypeID::Int64]).with_names(vec!["_id".to_string()])
37}
38
39impl PhysicalOperatorExec for PhysicalMerge {
40    fn operator_type(&self) -> &str {
41        "merge"
42    }
43
44    fn execute(&self, input: Vec<DataChunk>) -> OperatorResult {
45        // A filtered-out pipeline stays empty (P53.25): nothing was merged.
46        if !input.is_empty() && input.iter().all(|c| c.size == 0) {
47            return Ok(vec![DataChunk::new(vec![], vec![])]);
48        }
49
50        // Evaluate constant helper (fallback if input is empty)
51        let eval_const = |expr: &akar_parser::ast::Expression, chunk: Option<&DataChunk>, row: usize| -> Value {
52            if let Some(c) = chunk {
53                crate::physical::write_ops::set::evaluate_expression_for_row(expr, c, row)
54            } else {
55                match expr {
56                    akar_parser::ast::Expression::Constant(c) => match c {
57                        akar_parser::ast::Constant::Null => Value::Null,
58                        akar_parser::ast::Constant::Bool(b) => Value::Bool(*b),
59                        akar_parser::ast::Constant::Integer(i) => Value::Int64(*i),
60                        akar_parser::ast::Constant::Float(f) => Value::Double(*f),
61                        akar_parser::ast::Constant::String(s) => Value::String(s.clone()),
62                    },
63                    _ => Value::Null,
64                }
65            }
66        };
67
68        // Get table info to build the row
69        let num_cols = {
70            let tbl = self
71                .table_catalog
72                .get_node_table_by_name(&self.table_name)
73                .ok_or_else(|| format!("Table '{}' not found for MERGE", self.table_name))?;
74            tbl.columns.len()
75        };
76
77        // Handle input chunks for pipeline (or just 1 iteration if empty)
78        let chunks = if input.is_empty() {
79            let mut v = ValueVector::new(PhysicalTypeID::Int64, 1);
80            v.resize(1);
81            v.set_i64(0, 0);
82            let arr = akar_common::arrow_vector::ArrowVector::from_legacy(&v).array;
83            vec![DataChunk::new(vec![arr], vec![PhysicalTypeID::Int64])]
84        } else {
85            input
86        };
87
88        // One output row per processed input row, carrying the physical row id
89        // (matched or newly created) plus its source (chunk, row) so pipeline
90        // columns (e.g. UNWIND variables) survive into the output (P53.31).
91        let mut merged_row_ids: Vec<u64> = Vec::new();
92        let mut source_rows: Vec<(usize, usize)> = Vec::new();
93        let mut matched_ids: Vec<u64> = Vec::new();
94        let mut created_ids: Vec<u64> = Vec::new();
95
96        for (ci, chunk) in chunks.iter().enumerate() {
97            for row in 0..chunk.size {
98                let table_info = self
99                    .table_catalog
100                    .get_node_table_by_name(&self.table_name)
101                    .ok_or_else(|| format!("Table '{}' not found", self.table_name))?;
102
103                // Determine PK property
104                let mut pk_val = None;
105                for col_idx in 0..num_cols {
106                    let col = &table_info.columns[col_idx];
107                    if col.is_primary_key {
108                        if let Some((_, expr)) = self.properties.iter().find(|(n, _)| n == &col.name) {
109                            let val = eval_const(expr, Some(chunk), row);
110                            pk_val = Some(val);
111                        }
112                    }
113                }
114
115                // Match against the existing table. Without a PK property, fall
116                // back to a full scan over the first pattern property (Cypher
117                // MERGE on non-PK fields has no O(1) lookup without a secondary
118                // index).
119                let mut matched: Option<u64> = None;
120                if let Some(val) = &pk_val {
121                    let row_ids = table_info.lookup_by_pk_range(Some(val), true, Some(val), true, 1);
122                    if !row_ids.is_empty() {
123                        matched = Some(row_ids[0]);
124                    }
125                } else if let Some((prop_name, expr)) = self.properties.first() {
126                    let first_val = eval_const(expr, Some(chunk), row);
127                    if let Some(prop_col) = table_info.columns.iter().position(|c| &c.name == prop_name) {
128                        for row_idx in 0..table_info.num_rows as usize {
129                            if let Some(val) = table_info.get_value(row_idx, prop_col)
130                                && val == &first_val
131                            {
132                                matched = Some(row_idx as u64);
133                                break;
134                            }
135                        }
136                    }
137                }
138                drop(table_info);
139
140                if let Some(row_id) = matched {
141                    matched_ids.push(row_id);
142                    merged_row_ids.push(row_id);
143                    source_rows.push((ci, row));
144                } else {
145                    let table_info = self
146                        .table_catalog
147                        .get_node_table_by_name(&self.table_name)
148                        .ok_or_else(|| format!("Table '{}' not found", self.table_name))?;
149                    let mut new_values: Vec<Value> = Vec::new();
150                    for col_idx in 0..num_cols {
151                        let col_name = &table_info.columns[col_idx].name;
152                        if let Some((_, expr)) = self.properties.iter().find(|(n, _)| n == col_name) {
153                            new_values.push(eval_const(expr, Some(chunk), row));
154                        } else if table_info.columns[col_idx].is_primary_key {
155                            return Err(format!("MERGE CREATE requires primary key '{}'", col_name).into());
156                        } else {
157                            new_values.push(Value::Null);
158                        }
159                    }
160                    drop(table_info);
161
162                    if let Some(mut tbl) = self.table_catalog.get_node_table_by_name_mut(&self.table_name) {
163                        let row_id = tbl
164                            .insert_row_with_txn(new_values, self.txn_id)
165                            .map_err(|e| format!("MERGE CREATE failed: {e}"))?;
166                        if let Some(sink) = self.undo_sink.as_ref()
167                            && let Ok(mut u) = sink.lock()
168                        {
169                            u.push(UndoRecord::insert(self.table_id, row_id));
170                        }
171                        created_ids.push(row_id);
172                        merged_row_ids.push(row_id);
173                        source_rows.push((ci, row));
174                    }
175                }
176            }
177        }
178
179        // Apply ON MATCH / ON CREATE SET once per group, targeting the affected
180        // row ids via the `_id` pseudo-column (mirrors the edge MERGE path).
181        for set_op in &self.on_match {
182            if !matched_ids.is_empty() {
183                let chunk = row_id_chunk(&matched_ids);
184                set_op.execute(vec![chunk])?;
185            }
186        }
187        for set_op in &self.on_create {
188            if !created_ids.is_empty() {
189                let chunk = row_id_chunk(&created_ids);
190                set_op.execute(vec![chunk])?;
191            }
192        }
193
194        tracing::info!("MERGE: processed merges in '{}'", self.table_name);
195
196        if merged_row_ids.is_empty() {
197            return Ok(vec![DataChunk::new(vec![], vec![])]);
198        }
199
200        // Output one row per processed input row: the post-update table columns
201        // (named, so a following RETURN resolves `<alias>.<prop>`), the input
202        // pipeline columns, and the `_id` pseudo-column. Previously MERGE
203        // emitted a single count chunk, so `MERGE ... SET ... RETURN n.prop`
204        // resolved the projection against the count (P53.31).
205        let mut output = {
206            let table = self
207                .table_catalog
208                .get_node_table_by_name(&self.table_name)
209                .ok_or_else(|| format!("Node table '{}' not found for MERGE", self.table_name))?;
210            crate::physical::write_ops::set::build_old_row_chunk(&table.columns, &merged_row_ids, &|row_id, col| {
211                table.get_value(row_id as usize, col).cloned()
212            })?
213        };
214        append_pipeline_columns(&mut output, &chunks, &source_rows)?;
215
216        let mut v = ValueVector::new(PhysicalTypeID::Int64, merged_row_ids.len());
217        v.resize(merged_row_ids.len());
218        for (i, r) in merged_row_ids.iter().enumerate() {
219            v.set_i64(i, *r as i64);
220        }
221        output
222            .fields
223            .push(akar_common::arrow_vector::ArrowVector::from_legacy(&v).array);
224        output.field_types.push(PhysicalTypeID::Int64);
225        output.field_names.push("_id".to_string());
226
227        Ok(vec![output])
228    }
229}
akar_processor/physical/write_ops/merge.rs

akar_processor/physical/write_ops/
merge.rs
