akar_processor/processor/mapper/
map_update.rs

1use super::ExecutionContext;
2use crate::physical_operator::*;
3use akar_common::error::ProcessorError;
4use akar_common::vector::DataChunk;
5use akar_planner::logical_operator::LogicalOperator;
6
7pub fn map_and_execute_update(
8    op: &LogicalOperator,
9    current_input: Vec<DataChunk>,
10    ctx: &mut ExecutionContext,
11) -> Result<Vec<DataChunk>, ProcessorError> {
12    match op {
13        LogicalOperator::Set(sl) => {
14            let table_catalog = ctx
15                .table_catalog
16                .clone()
17                .ok_or_else(|| "No table catalog available for SET".to_string())?;
18
19            let set_op = PhysicalSet {
20                table_name: sl.table_name.clone(),
21                table_id: sl.table_id,
22                is_node: sl.is_node,
23                items: sl.items.clone(),
24                table_catalog,
25                txn_id: ctx.txn_id,
26                undo_sink: Some(ctx.processor.undo_sink()),
27                function_registry: ctx.function_registry.clone(),
28                emit_count: sl.emit_count,
29            };
30            let result = set_op.execute(current_input)?;
31            // Record written rows for OCC conflict detection
32            record_set_writes(sl.table_id, &result, ctx);
33            Ok(result)
34        }
35        LogicalOperator::Delete(dl) => {
36            let table_catalog = ctx
37                .table_catalog
38                .clone()
39                .ok_or_else(|| "No table catalog available for DELETE".to_string())?;
40
41            let delete_op = PhysicalDelete {
42                table_name: dl.table_name.clone(),
43                table_id: dl.table_id,
44                primary_key_column: dl.primary_key_column.clone(),
45                is_node: dl.is_node,
46                detach: dl.detach,
47                row_indices: Vec::new(),
48                table_catalog,
49                txn_id: ctx.txn_id,
50                undo_sink: Some(ctx.processor.undo_sink()),
51            };
52            let result = delete_op.execute(current_input)?;
53            // Record written rows for OCC conflict detection
54            record_delete_writes(dl.table_id, &result, ctx);
55            Ok(result)
56        }
57        LogicalOperator::CreateNode(cn) => {
58            let table_catalog = ctx
59                .table_catalog
60                .clone()
61                .ok_or_else(|| "No table catalog available for CREATE".to_string())?;
62
63            let create_node_op = PhysicalInsertNode {
64                table_name: cn.table_name.clone(),
65                table_id: cn.table_id,
66                out_var_name: cn.out_var_name.clone(),
67                properties: cn.properties.clone(),
68                table_catalog,
69                txn_id: ctx.txn_id,
70                undo_sink: Some(ctx.processor.undo_sink()),
71            };
72            let result = create_node_op.execute(current_input)?;
73            // Record written rows for OCC conflict detection
74            record_insert_writes(cn.table_id, &result, ctx);
75            Ok(result)
76        }
77        LogicalOperator::CreateRel(cr) => {
78            let table_catalog = ctx
79                .table_catalog
80                .clone()
81                .ok_or_else(|| "No table catalog available for CREATE".to_string())?;
82
83            let create_rel_op = PhysicalInsertRel {
84                table_name: cr.table_name.clone(),
85                table_id: cr.table_id,
86                src_node_name: cr.src_node_name.clone(),
87                dst_node_name: cr.dst_node_name.clone(),
88                properties: cr.properties.clone(),
89                table_catalog,
90                txn_id: ctx.txn_id,
91                undo_sink: Some(ctx.processor.undo_sink()),
92            };
93            let result = create_rel_op.execute(current_input)?;
94            // Record written rows for OCC conflict detection
95            record_insert_writes(cr.table_id, &result, ctx);
96            Ok(result)
97        }
98        LogicalOperator::Extend(ex) => {
99            let table_catalog = ctx
100                .table_catalog
101                .clone()
102                .ok_or_else(|| "No table catalog available for Extend".to_string())?;
103
104            let extend_op = PhysicalExtend {
105                rel_table_name: ex.rel_table_name.clone(),
106                rel_table_id: ex.rel_table_id,
107                rel_var: ex.rel_var.clone(),
108                bound_node_var: ex.bound_node_var.clone(),
109                direction: ex.direction.clone(),
110                dst_node_var: ex.dst_node_var.clone(),
111                dst_table_name: ex.dst_table_name.clone(),
112                dst_table_id: ex.dst_table_id,
113                table_catalog,
114            };
115            let result = extend_op.execute(current_input)?;
116            // Record written rows for OCC conflict detection
117            record_insert_writes(ex.rel_table_id, &result, ctx);
118            Ok(result)
119        }
120        LogicalOperator::OptionalExtend(oe) => {
121            let table_catalog = ctx
122                .table_catalog
123                .clone()
124                .ok_or_else(|| "No table catalog available for OptionalExtend".to_string())?;
125
126            let input = if oe.children.is_empty() {
127                current_input
128            } else {
129                ctx.execute_children(&oe.children)?
130            };
131
132            let optional_extend_op = PhysicalOptionalExtend {
133                rel_table_name: oe.rel_table_name.clone(),
134                rel_table_id: oe.rel_table_id,
135                rel_var: oe.rel_var.clone(),
136                src_node_var: oe.src_node_var.clone(),
137                dst_node_var: oe.dst_node_var.clone(),
138                direction: oe.direction.clone(),
139                table_catalog,
140            };
141            Ok(optional_extend_op.execute(input)?)
142        }
143        LogicalOperator::Merge(m) => {
144            let table_catalog = ctx
145                .table_catalog
146                .clone()
147                .ok_or_else(|| "No table catalog available for MERGE".to_string())?;
148
149            let mut on_match_ops = Vec::new();
150            for set_item in &m.on_match {
151                on_match_ops.push(PhysicalSet {
152                    table_name: set_item.table_name.clone(),
153                    table_id: set_item.table_id,
154                    is_node: set_item.is_node,
155                    items: set_item.items.clone(),
156                    table_catalog: table_catalog.clone(),
157                    txn_id: ctx.txn_id,
158                    undo_sink: Some(ctx.processor.undo_sink()),
159                    function_registry: ctx.function_registry.clone(),
160                    emit_count: false,
161                });
162            }
163
164            let mut on_create_ops = Vec::new();
165            for set_item in &m.on_create {
166                on_create_ops.push(PhysicalSet {
167                    table_name: set_item.table_name.clone(),
168                    table_id: set_item.table_id,
169                    is_node: set_item.is_node,
170                    items: set_item.items.clone(),
171                    table_catalog: table_catalog.clone(),
172                    txn_id: ctx.txn_id,
173                    undo_sink: Some(ctx.processor.undo_sink()),
174                    function_registry: ctx.function_registry.clone(),
175                    emit_count: false,
176                });
177            }
178
179            let merge_op = PhysicalMerge {
180                table_name: m.table_name.clone(),
181                table_id: m.table_id,
182                properties: m.properties.clone(),
183                on_match: on_match_ops,
184                on_create: on_create_ops,
185                table_catalog,
186                txn_id: ctx.txn_id,
187                undo_sink: Some(ctx.processor.undo_sink()),
188            };
189            let result = merge_op.execute(current_input)?;
190            // Record written rows for OCC conflict detection
191            record_insert_writes(m.table_id, &result, ctx);
192            Ok(result)
193        }
194        LogicalOperator::MergeRel(mr) => {
195            let table_catalog = ctx
196                .table_catalog
197                .clone()
198                .ok_or_else(|| "No table catalog available for MERGE".to_string())?;
199
200            let mut on_match_ops = Vec::new();
201            for set_item in &mr.on_match {
202                on_match_ops.push(PhysicalSet {
203                    table_name: set_item.table_name.clone(),
204                    table_id: set_item.table_id,
205                    is_node: set_item.is_node,
206                    items: set_item.items.clone(),
207                    table_catalog: table_catalog.clone(),
208                    txn_id: ctx.txn_id,
209                    undo_sink: Some(ctx.processor.undo_sink()),
210                    function_registry: ctx.function_registry.clone(),
211                    emit_count: false,
212                });
213            }
214
215            let mut on_create_ops = Vec::new();
216            for set_item in &mr.on_create {
217                on_create_ops.push(PhysicalSet {
218                    table_name: set_item.table_name.clone(),
219                    table_id: set_item.table_id,
220                    is_node: set_item.is_node,
221                    items: set_item.items.clone(),
222                    table_catalog: table_catalog.clone(),
223                    txn_id: ctx.txn_id,
224                    undo_sink: Some(ctx.processor.undo_sink()),
225                    function_registry: ctx.function_registry.clone(),
226                    emit_count: false,
227                });
228            }
229
230            let merge_rel_op = PhysicalMergeRel {
231                rel_table_name: mr.rel_table_name.clone(),
232                rel_table_id: mr.rel_table_id,
233                edge_var: mr.edge_var.clone(),
234                src_node_var: mr.src_node_var.clone(),
235                dst_node_var: mr.dst_node_var.clone(),
236                direction: akar_parser::ast::EdgeDirection::LeftToRight,
237                properties: mr.properties.clone(),
238                on_match: on_match_ops,
239                on_create: on_create_ops,
240                table_catalog,
241                txn_id: ctx.txn_id,
242                undo_sink: Some(ctx.processor.undo_sink()),
243            };
244            let result = merge_rel_op.execute(current_input)?;
245            // Record written rows for OCC conflict detection
246            record_insert_writes(mr.rel_table_id, &result, ctx);
247            Ok(result)
248        }
249        LogicalOperator::CopyFrom(cf) => {
250            let table_catalog = ctx
251                .table_catalog
252                .clone()
253                .ok_or_else(|| "No table catalog available for COPY FROM".to_string())?;
254
255            // Get column definitions from the table catalog
256            let columns = if let Some(node_table) = table_catalog.get_node_table_by_name(&cf.table_name) {
257                node_table.columns.clone()
258            } else if let Some(rel_table) = table_catalog.get_rel_table_by_name(&cf.table_name) {
259                rel_table.columns.clone()
260            } else {
261                return Err(format!("Table '{}' not found in storage catalog", cf.table_name).into());
262            };
263
264            let copy_op = PhysicalCopyFrom {
265                table_name: cf.table_name.clone(),
266                table_id: cf.table_id,
267                file_path: cf.file_path.clone(),
268                columns,
269                options: cf.options.clone(),
270                table_catalog,
271                vfs: ctx
272                    .vfs
273                    .clone()
274                    .ok_or_else(|| "VFS not initialized in processor".to_string())?,
275                txn_id: ctx.txn_id,
276                undo_sink: Some(ctx.processor.undo_sink()),
277            };
278            let result = copy_op.execute(current_input)?;
279            // Record written rows for OCC conflict detection
280            record_insert_writes(cf.table_id, &result, ctx);
281            Ok(result)
282        }
283        LogicalOperator::BatchInsert(bi) => {
284            let table_catalog = ctx
285                .table_catalog
286                .clone()
287                .ok_or_else(|| "No table catalog available for BATCH INSERT".to_string())?;
288
289            let batch_op = PhysicalBatchInsert {
290                table_name: bi.table_name.clone(),
291                table_id: bi.table_id,
292                rows: bi.rows.clone(),
293                table_catalog,
294                txn_id: ctx.txn_id,
295                undo_sink: Some(ctx.processor.undo_sink()),
296            };
297            let result = batch_op.execute(current_input)?;
298            // Record written rows for OCC conflict detection
299            record_insert_writes(bi.table_id, &result, ctx);
300            Ok(result)
301        }
302        LogicalOperator::Insert(i) => {
303            let exec = crate::physical::misc::PhysicalInsert {
304                table_name: i.table_name.clone(),
305                table_id: i.table_id,
306                columns: i.columns.clone(),
307                values: i.values.clone(),
308                table_catalog: ctx.table_catalog.clone().unwrap(),
309                txn_id: ctx.txn_id,
310                undo_sink: Some(ctx.processor.undo_sink()),
311            };
312            let result = exec.execute(current_input)?;
313            // Record written rows for OCC conflict detection
314            record_insert_writes(i.table_id, &result, ctx);
315            Ok(result)
316        }
317        _ => Err(format!("Not an update operator: {:?}", op).into()),
318    }
319}
320
321/// Record rows written by a SET operation for OCC conflict detection.
322/// The result DataChunk carries the updated row indices under the `_id`
323/// pseudo-column (P53.30); older outputs put a single updated-count in column 0.
324fn record_set_writes(table_id: u64, result: &[DataChunk], ctx: &mut ExecutionContext) {
325    if let Some(chunk) = result.first() {
326        let id_col = chunk
327            .field_names
328            .iter()
329            .position(|n| n == "_id" || n.ends_with("._id"))
330            .unwrap_or(0);
331        for row in 0..chunk.size {
332            if !chunk.fields.is_empty() {
333                if let Some(akar_common::types::Value::Int64(row_idx)) = chunk.get_value(id_col, row) {
334                    ctx.written_rows.push((table_id, row_idx as u64));
335                }
336            }
337        }
338    }
339}
340
341/// Record rows written by a DELETE operation for OCC conflict detection.
342/// The result DataChunk contains the row indices that were deleted (first column).
343fn record_delete_writes(table_id: u64, result: &[DataChunk], ctx: &mut ExecutionContext) {
344    if let Some(chunk) = result.first() {
345        for row in 0..chunk.size {
346            if !chunk.fields.is_empty() {
347                if let Some(akar_common::types::Value::Int64(row_idx)) = chunk.get_value(0, row) {
348                    ctx.written_rows.push((table_id, row_idx as u64));
349                }
350            }
351        }
352    }
353}
354
355/// Record rows written by an INSERT operation for OCC conflict detection.
356/// When the result chunk contains an `_id` pseudo-column (Merge output, P53.31)
357/// or a second column with assigned row IDs (Create/BatchInsert), tracks at row
358/// level. Otherwise, row-level inserts are not tracked (PK uniqueness is
359/// enforced by the storage layer's hash index).
360fn record_insert_writes(table_id: u64, result: &[DataChunk], ctx: &mut ExecutionContext) {
361    if let Some(chunk) = result.first() {
362        // Column 0 = inserted_count, Column 1 = assigned row IDs (legacy); a
363        // Merge output names the row ids `_id` at its last column instead.
364        if let Some(id_col) = chunk.field_names.iter().position(|n| n == "_id" || n.ends_with("._id")) {
365            for row in 0..chunk.size {
366                if let Some(akar_common::types::Value::Int64(row_id)) = chunk.get_value(id_col, row) {
367                    ctx.written_rows.push((table_id, row_id as u64));
368                }
369            }
370        } else if chunk.fields.len() > 1 {
371            for row in 0..chunk.fields[1].len() {
372                if let Some(akar_common::types::Value::Int64(row_id)) = chunk.get_value(1, row) {
373                    ctx.written_rows.push((table_id, row_id as u64));
374                }
375            }
376        }
377    }
378}
akar_processor/processor/mapper/map_update.rs

akar_processor/processor/mapper/
map_update.rs
