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

akar_processor/processor/mapper/
map_update.rs
