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;
5use akar_common::types::{PhysicalTypeID, Value};
6use akar_common::vector::{DataChunk, ValueVector};
7use akar_storage::table::TableCatalog;
8use std::sync::Arc;
9
10/// Physical operator for MERGE.
11/// Represents a combination of MATCH and INSERT (Upsert).
12pub struct PhysicalMerge {
13    pub table_name: String,
14    pub table_id: u64,
15    pub properties: Vec<(String, akar_parser::ast::Expression)>,
16    pub on_match: Vec<PhysicalSet>,
17    pub on_create: Vec<PhysicalSet>,
18    pub table_catalog: Arc<TableCatalog>,
19}
20
21impl PhysicalOperatorExec for PhysicalMerge {
22    fn operator_type(&self) -> &str {
23        "merge"
24    }
25
26    fn execute(&self, _input: Vec<DataChunk>) -> OperatorResult {
27        let mut _merged_count: u64 = 0;
28
29        // Evaluate constant helper (fallback if input is empty)
30        let eval_const = |expr: &akar_parser::ast::Expression, chunk: Option<&DataChunk>, row: usize| -> Value {
31            if let Some(c) = chunk {
32                crate::physical::write_ops::set::evaluate_expression_for_row(expr, c, row)
33            } else {
34                match expr {
35                    akar_parser::ast::Expression::Constant(c) => match c {
36                        akar_parser::ast::Constant::Null => Value::Null,
37                        akar_parser::ast::Constant::Bool(b) => Value::Bool(*b),
38                        akar_parser::ast::Constant::Integer(i) => Value::Int64(*i),
39                        akar_parser::ast::Constant::Float(f) => Value::Double(*f),
40                        akar_parser::ast::Constant::String(s) => Value::String(s.clone()),
41                    },
42                    _ => Value::Null,
43                }
44            }
45        };
46
47        // Get table info to build the row
48        let num_cols = {
49            let tbl = self
50                .table_catalog
51                .get_node_table_by_name(&self.table_name)
52                .ok_or_else(|| format!("Table '{}' not found for MERGE", self.table_name))?;
53            tbl.columns.len()
54        };
55
56        // Handle input chunks for pipeline (or just 1 iteration if empty)
57        let chunks = if _input.is_empty() {
58            let mut v = ValueVector::new(PhysicalTypeID::Int64, 1);
59            v.resize(1);
60            v.set_i64(0, 0);
61            let arr = akar_common::arrow_vector::ArrowVector::from_legacy(&v).array;
62            vec![DataChunk::new(vec![arr], vec![PhysicalTypeID::Int64])]
63        } else {
64            _input
65        };
66
67        for chunk in &chunks {
68            for row in 0..chunk.size {
69                let table_info = self
70                    .table_catalog
71                    .get_node_table_by_name(&self.table_name)
72                    .ok_or_else(|| format!("Table '{}' not found", self.table_name))?;
73
74                // Determine PK property
75                let mut pk_val = None;
76                for col_idx in 0..num_cols {
77                    let col = &table_info.columns[col_idx];
78                    if col.is_primary_key {
79                        if let Some((_, expr)) = self.properties.iter().find(|(n, _)| n == &col.name) {
80                            let val = eval_const(expr, Some(chunk), row);
81                            pk_val = Some(val);
82                        }
83                    }
84                }
85
86                let is_node = true; // For now assuming node merge
87                let mut matched = false;
88
89                if is_node {
90                    if let Some(val) = &pk_val {
91                        let row_ids = table_info.lookup_by_pk_range(Some(val), true, Some(val), true, 1);
92                        if !row_ids.is_empty() {
93                            matched = true;
94                        }
95                    } else {
96                        // Without PK, we cannot effectively match a node in O(1).
97                        // Cypher MERGE on non-PK fields requires full table scan or secondary index.
98                        // For this implementation, we fail if PK is not provided.
99                        if let Some((prop_name, expr)) = self.properties.first() {
100                            let first_val = eval_const(expr, Some(chunk), row);
101                            if let Some(prop_col) = table_info.columns.iter().position(|c| &c.name == prop_name) {
102                                // Fallback scan
103                                for row_idx in 0..table_info.num_rows as usize {
104                                    if let Some(val) = table_info.get_value(row_idx, prop_col) {
105                                        if val == &first_val {
106                                            matched = true;
107                                            break;
108                                        }
109                                    }
110                                }
111                            }
112                        }
113                    }
114                } else {
115                    // Rel merge fallback
116                }
117                drop(table_info);
118
119                if matched {
120                    for set_op in &self.on_match {
121                        let single_chunk = DataChunk::new(chunk.fields.clone(), chunk.field_types.clone()); // pass the row
122                        let _ = set_op.execute(vec![single_chunk])?;
123                    }
124                } else {
125                    let table_info = self.table_catalog.get_node_table_by_name(&self.table_name).unwrap();
126                    let mut new_values: Vec<Value> = Vec::new();
127                    for col_idx in 0..num_cols {
128                        let col_name = &table_info.columns[col_idx].name;
129                        if let Some((_, expr)) = self.properties.iter().find(|(n, _)| n == col_name) {
130                            new_values.push(eval_const(expr, Some(chunk), row));
131                        } else if table_info.columns[col_idx].is_primary_key {
132                            return Err(format!("MERGE CREATE requires primary key '{}'", col_name).into());
133                        } else {
134                            new_values.push(Value::Null);
135                        }
136                    }
137                    drop(table_info);
138
139                    if let Some(mut tbl) = self.table_catalog.get_node_table_by_name_mut(&self.table_name) {
140                        tbl.insert_row(new_values)
141                            .map_err(|e| format!("MERGE CREATE failed: {e}"))?;
142                        _merged_count += 1;
143                    }
144
145                    for set_op in &self.on_create {
146                        let single_chunk = DataChunk::new(chunk.fields.clone(), chunk.field_types.clone());
147                        let _ = set_op.execute(vec![single_chunk])?;
148                    }
149                }
150            }
151        }
152
153        tracing::info!("MERGE: processed merges in '{}'", self.table_name);
154
155        let mut v = ValueVector::new(PhysicalTypeID::Int64, 1);
156        v.resize(1);
157        v.set_i64(0, _merged_count as i64);
158        let arr = akar_common::arrow_vector::ArrowVector::from_legacy(&v).array;
159        Ok(vec![DataChunk::new(vec![arr], vec![PhysicalTypeID::Int64])])
160    }
161}
akar_processor/physical/write_ops/merge.rs

akar_processor/physical/write_ops/
merge.rs
