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

akar_processor/processor/mapper/
map_update.rs
