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

akar_processor/processor/mapper/
map_update.rs
