akar_processor/physical/write_ops/
insert.rs

1//! Physical operators for INSERT (CreateNode, CreateRel).
2
3use crate::physical::types::{OperatorResult, PhysicalOperatorExec};
4use crate::physical::write_ops::set::evaluate_expression_for_row;
5use akar_common::types::{PhysicalTypeID, Value, physical_type_from_logical};
6use akar_common::vector::{DataChunk, ValueVector};
7use akar_storage::table::TableCatalog;
8use akar_storage::wal::{WalSink, log_insert_record, log_rel_insert_record};
9use akar_transaction::UndoRecord;
10use std::sync::{Arc, Mutex};
11
12/// Physical operator for CREATE NODE.
13pub struct PhysicalInsertNode {
14    pub table_name: String,
15    pub table_id: u64,
16    pub out_var_name: String,
17    pub properties: Vec<(String, akar_parser::ast::Expression)>,
18    pub table_catalog: Arc<TableCatalog>,
19    /// Active transaction id — inserts are recorded in `VersionInfo` for MVCC (P52.18).
20    pub txn_id: Option<u64>,
21    /// Undo sink for rollback records (P52.18).
22    pub undo_sink: Option<Arc<Mutex<Vec<UndoRecord>>>>,
23    /// Typed WAL sink so the row survives restarts via WAL replay (P60.2).
24    pub wal_sink: Option<WalSink>,
25}
26
27impl PhysicalOperatorExec for PhysicalInsertNode {
28    fn operator_type(&self) -> &str {
29        "insert_node"
30    }
31
32    fn execute(&self, input: Vec<DataChunk>) -> OperatorResult {
33        // A filtered-out pipeline (all input chunks empty) must stay empty:
34        // nothing was created, so the output row count is zero (P53.25).
35        if !input.is_empty() && input.iter().all(|c| c.size == 0) {
36            return Ok(vec![DataChunk::new(vec![], vec![])]);
37        }
38
39        let mut assigned_row_ids: Vec<i64> = Vec::new();
40        let mut output_rows: Vec<Vec<Value>> = Vec::new();
41        let mut table = self
42            .table_catalog
43            .get_node_table_by_name_mut(&self.table_name)
44            .ok_or_else(|| format!("Node table '{}' not found for INSERT", self.table_name))?;
45
46        // If input is empty (no previous pipeline), we insert exactly one node.
47        // Otherwise, we insert a node for each row in the input.
48        let chunks = if input.is_empty() {
49            let mut v = ValueVector::new(PhysicalTypeID::Int64, 1);
50            v.resize(1);
51            v.set_i64(0, 0);
52            let arr = akar_common::arrow_vector::ArrowVector::from_legacy(&v).array;
53            vec![DataChunk::new(vec![arr], vec![PhysicalTypeID::Int64])]
54        } else {
55            input
56        };
57
58        let num_cols = table.columns.len();
59        for chunk in &chunks {
60            for row in 0..chunk.size {
61                let mut row_values = vec![Value::Null; num_cols];
62
63                for (prop_name, expr) in &self.properties {
64                    if let Some(col_idx) = table.columns.iter().position(|c| c.name == *prop_name) {
65                        let val = evaluate_expression_for_row(expr, chunk, row);
66                        row_values[col_idx] = val;
67                    }
68                }
69
70                // Add the row to the node table; capture assigned row_id for OCC.
71                // Errors (e.g. NULL primary key) must surface, not silently skip
72                // the row — otherwise UNWIND+CREATE drops input rows (P53.27).
73                let logged_row = self.wal_sink.is_some().then(|| row_values.clone());
74                let row_id = table
75                    .insert_row_with_txn(row_values.clone(), self.txn_id)
76                    .map_err(|e| format!("INSERT NODE row {row} failed in '{}': {e}", self.table_name))?;
77                assigned_row_ids.push(row_id as i64);
78                output_rows.push(row_values);
79                log_insert_record(&self.wal_sink, self.table_id, logged_row.as_deref().unwrap_or(&[]));
80                if let Some(sink) = self.undo_sink.as_ref()
81                    && let Ok(mut u) = sink.lock()
82                {
83                    u.push(UndoRecord::insert(self.table_id, row_id));
84                }
85            }
86        }
87
88        let inserted_count = assigned_row_ids.len();
89        tracing::info!("INSERT NODE: added {inserted_count} rows to '{}'", self.table_name);
90
91        // Nothing was created (the earlier guard also covers all-empty input):
92        // return an empty result with zero rows (P53.25).
93        if inserted_count == 0 {
94            return Ok(vec![DataChunk::new(vec![], vec![])]);
95        }
96
97        let n = inserted_count;
98
99        // Column 0: `_id` — assigned internal row offsets, exposed for OCC
100        // write-set tracking (record_insert_writes reads the `_id` field name,
101        // matching the convention used by MERGE/set.rs).
102        let mut id_v = ValueVector::new(PhysicalTypeID::Int64, n);
103        id_v.resize(n);
104        for (i, rid) in assigned_row_ids.iter().enumerate() {
105            id_v.set_i64(i, *rid);
106        }
107        let mut fields = vec![akar_common::arrow_vector::ArrowVector::from_legacy(&id_v).array];
108        let mut types = vec![PhysicalTypeID::Int64];
109        let mut names = vec!["_id".to_string()];
110
111        // Columns 1..: the created node's property columns bound to `out_var_name`
112        // (e.g. `n.id`, `n.name`) so `RETURN n.id, n.name` resolves to the real
113        // values and the no-RETURN result reports num_rows = n (P73.1). Every row
114        // maps 1:1 to one inserted node. Complex-typed columns (List/Struct) that
115        // the plain ValueVector cannot materialise are skipped to avoid regressing
116        // write-only CREATE of vector/struct node columns.
117        for (col_idx, col) in table.columns.iter().enumerate() {
118            let ptype = physical_type_from_logical(col.logical_type);
119            let mut cv = ValueVector::new(ptype, n);
120            cv.resize(n);
121            let mut buildable = true;
122            for (row_i, row_values) in output_rows.iter().enumerate() {
123                if cv.set_value(row_i, &row_values[col_idx]).is_err() {
124                    buildable = false;
125                    break;
126                }
127            }
128            if !buildable {
129                tracing::warn!(
130                    "INSERT NODE: skipping complex output column '{}' in '{}'",
131                    col.name,
132                    self.table_name
133                );
134                continue;
135            }
136            fields.push(akar_common::arrow_vector::ArrowVector::from_legacy(&cv).array);
137            types.push(ptype);
138            names.push(format!("{}.{}", self.out_var_name, col.name));
139        }
140
141        Ok(vec![DataChunk::new(fields, types).with_names(names)])
142    }
143}
144
145/// Physical operator for CREATE REL.
146pub struct PhysicalInsertRel {
147    pub table_name: String,
148    pub table_id: u64,
149    pub src_node_name: String,
150    pub dst_node_name: String,
151    pub out_var_name: String,
152    pub properties: Vec<(String, akar_parser::ast::Expression)>,
153    pub table_catalog: Arc<TableCatalog>,
154    /// Active transaction id for MVCC + undo recording (P52.18).
155    pub txn_id: Option<u64>,
156    /// Undo sink for rollback records (P52.18).
157    pub undo_sink: Option<Arc<Mutex<Vec<UndoRecord>>>>,
158    /// Typed WAL sink so the edge survives restarts via WAL replay (P60.2).
159    pub wal_sink: Option<WalSink>,
160}
161
162impl PhysicalOperatorExec for PhysicalInsertRel {
163    fn operator_type(&self) -> &str {
164        "insert_rel"
165    }
166
167    fn execute(&self, input: Vec<DataChunk>) -> OperatorResult {
168        let mut inserted_count = 0u64;
169        let mut table = self
170            .table_catalog
171            .get_rel_table_by_name_mut(&self.table_name)
172            .ok_or_else(|| format!("Rel table '{}' not found for INSERT", self.table_name))?;
173
174        // A filtered-out pipeline (all input chunks empty) must stay empty:
175        // nothing was created, so the output row count is zero (P53.25).
176        if !input.is_empty() && input.iter().all(|c| c.size == 0) {
177            return Ok(vec![DataChunk::new(vec![], vec![])]);
178        }
179
180        let num_cols = table.columns.len();
181        let mut rels_to_insert = Vec::new();
182        let mut rel_props: Vec<Vec<Value>> = Vec::new();
183
184        // Flatten the input in row-major form so the write operator's output can
185        // carry the source/destination node columns (`a.id`, `b.id`, `a._id`, ...)
186        // forward for RETURN projection, mirroring the node path (P73.2). The
187        // column schema (names + physical types) is taken from the first chunk;
188        // every input row maps 1:1 to one inserted relationship.
189        let base_names: Vec<String> = input.first().map(|c| c.field_names.clone()).unwrap_or_default();
190        let base_types: Vec<PhysicalTypeID> = input.first().map(|c| c.field_types.clone()).unwrap_or_default();
191        let mut context_rows: Vec<Vec<Value>> = Vec::new();
192
193        for chunk in &input {
194            let src_name_id = format!("{}.{}", self.src_node_name, "_id");
195            let src_name_pk = format!("{}.{}", self.src_node_name, "id");
196            let src_name_pk_upper = format!("{}.{}", self.src_node_name, "ID");
197            let src_node_col_idx = chunk
198                .field_names
199                .iter()
200                .position(|name| name == &src_name_id)
201                .or_else(|| chunk.field_names.iter().position(|name| name == &self.src_node_name))
202                .or_else(|| chunk.field_names.iter().position(|name| name == &src_name_pk))
203                .or_else(|| chunk.field_names.iter().position(|name| name == &src_name_pk_upper))
204                .or_else(|| {
205                    chunk
206                        .field_names
207                        .iter()
208                        .position(|name| name.eq_ignore_ascii_case(&src_name_pk))
209                })
210                .ok_or_else(|| {
211                    format!(
212                        "Source node variable {} not found (fields: {:?})",
213                        self.src_node_name, chunk.field_names
214                    )
215                })?;
216
217            let dst_name_id = format!("{}.{}", self.dst_node_name, "_id");
218            let dst_name_pk = format!("{}.{}", self.dst_node_name, "id");
219            let dst_name_pk_upper = format!("{}.{}", self.dst_node_name, "ID");
220            let dst_node_col_idx = chunk
221                .field_names
222                .iter()
223                .position(|name| name == &dst_name_id)
224                .or_else(|| chunk.field_names.iter().position(|name| name == &self.dst_node_name))
225                .or_else(|| chunk.field_names.iter().position(|name| name == &dst_name_pk))
226                .or_else(|| chunk.field_names.iter().position(|name| name == &dst_name_pk_upper))
227                .or_else(|| {
228                    chunk
229                        .field_names
230                        .iter()
231                        .position(|name| name.eq_ignore_ascii_case(&dst_name_pk))
232                })
233                .ok_or_else(|| {
234                    format!(
235                        "Destination node variable {} not found (fields: {:?})",
236                        self.dst_node_name, chunk.field_names
237                    )
238                })?;
239
240            if src_node_col_idx >= chunk.fields.len() || dst_node_col_idx >= chunk.fields.len() {
241                return Err("Src/Dst node column index out of bounds in INSERT REL".into());
242            }
243
244            for row in 0..chunk.size {
245                let src_id = if let Some(Value::Int64(val)) = chunk.get_value(src_node_col_idx, row) {
246                    val as u64
247                } else {
248                    0
249                };
250                let dst_id = if let Some(Value::Int64(val)) = chunk.get_value(dst_node_col_idx, row) {
251                    val as u64
252                } else {
253                    0
254                };
255
256                let mut props = vec![Value::Null; num_cols];
257                for (prop_name, expr) in &self.properties {
258                    if let Some(col_idx) = table.columns.iter().position(|c| c.name == *prop_name) {
259                        let val = evaluate_expression_for_row(expr, chunk, row);
260                        props[col_idx] = val;
261                    }
262                }
263
264                rels_to_insert.push((src_id, dst_id, props.clone()));
265                rel_props.push(props);
266
267                let mut ctx_row = Vec::with_capacity(chunk.fields.len());
268                for col in 0..chunk.fields.len() {
269                    ctx_row.push(chunk.get_value(col, row).unwrap_or(Value::Null));
270                }
271                context_rows.push(ctx_row);
272            }
273        }
274
275        // The edge indices assigned to this batch begin at the current edge count;
276        // the batch inserts `rels_to_insert` in order, so edge i gets id
277        // `start_edge_idx + i` (used for the `_id` output column and OCC).
278        let start_edge_idx = table.edges.len() as i64;
279
280        // Batch insert the collected relationships
281        if !rels_to_insert.is_empty() {
282            inserted_count = table
283                .insert_rels_batch(&rels_to_insert)
284                .map_err(|e| format!("BatchInsert rel error: {e}"))?;
285            for (src_id, dst_id, props) in &rels_to_insert {
286                log_rel_insert_record(&self.wal_sink, self.table_id, *src_id, *dst_id, props);
287            }
288            if let Some(sink) = self.undo_sink.as_ref()
289                && let Ok(mut u) = sink.lock()
290            {
291                let num_edges = table.edges.len();
292                for idx in (num_edges - inserted_count as usize)..num_edges {
293                    u.push(UndoRecord::insert(self.table_id, idx as u64));
294                }
295            }
296        }
297
298        tracing::info!("INSERT REL: added {inserted_count} rels to '{}'", self.table_name);
299
300        let n = inserted_count as usize;
301        if n == 0 {
302            return Ok(vec![DataChunk::new(vec![], vec![])]);
303        }
304
305        // Column 0: `_id` — assigned internal edge offsets, placed FIRST so OCC
306        // `record_insert_writes` resolves this exact `_id` before any source/dest
307        // `.*_id` node columns carried through below (P73.2).
308        let mut id_v = ValueVector::new(PhysicalTypeID::Int64, n);
309        id_v.resize(n);
310        for (i, _) in context_rows.iter().enumerate().take(n) {
311            id_v.set_i64(i, start_edge_idx + i as i64);
312        }
313        let mut fields = vec![akar_common::arrow_vector::ArrowVector::from_legacy(&id_v).array];
314        let mut types = vec![PhysicalTypeID::Int64];
315        let mut names = vec!["_id".to_string()];
316
317        // Pass through the input (source/destination node) columns so RETURN
318        // projection like `a.id, b.id` resolves to the real values (P73.2).
319        // Complex-typed columns (List/Struct) the plain ValueVector cannot
320        // materialise are skipped to avoid regressing vector/struct passthrough.
321        for (col, name) in base_names.iter().enumerate() {
322            let ptype = base_types.get(col).copied().unwrap_or(PhysicalTypeID::Any);
323            let mut cv = ValueVector::new(ptype, n);
324            cv.resize(n);
325            let mut buildable = true;
326            for (row_i, ctx_row) in context_rows.iter().enumerate().take(n) {
327                if cv.set_value(row_i, &ctx_row[col]).is_err() {
328                    buildable = false;
329                    break;
330                }
331            }
332            if !buildable {
333                tracing::warn!(
334                    "INSERT REL: skipping complex output column '{}' in '{}'",
335                    name,
336                    self.table_name
337                );
338                continue;
339            }
340            fields.push(akar_common::arrow_vector::ArrowVector::from_legacy(&cv).array);
341            types.push(ptype);
342            names.push(name.clone());
343        }
344
345        // Rel property columns bound to `out_var_name` (e.g. `r.weight`, `r.type`)
346        // so RETURN projection resolves the created relationship's real values (P73.2).
347        for (col_idx, col) in table.columns.iter().enumerate() {
348            let ptype = physical_type_from_logical(col.logical_type);
349            let mut cv = ValueVector::new(ptype, n);
350            cv.resize(n);
351            let mut buildable = true;
352            for (row_i, props) in rel_props.iter().enumerate().take(n) {
353                if cv.set_value(row_i, &props[col_idx]).is_err() {
354                    buildable = false;
355                    break;
356                }
357            }
358            if !buildable {
359                tracing::warn!(
360                    "INSERT REL: skipping complex output column '{}' in '{}'",
361                    col.name,
362                    self.table_name
363                );
364                continue;
365            }
366            fields.push(akar_common::arrow_vector::ArrowVector::from_legacy(&cv).array);
367            types.push(ptype);
368            names.push(format!("{}.{}", self.out_var_name, col.name));
369        }
370
371        Ok(vec![DataChunk::new(fields, types).with_names(names)])
372    }
373}
akar_processor/physical/write_ops/insert.rs

akar_processor/physical/write_ops/
insert.rs
