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 properties: Vec<(String, akar_parser::ast::Expression)>,
152    pub table_catalog: Arc<TableCatalog>,
153    /// Active transaction id for MVCC + undo recording (P52.18).
154    pub txn_id: Option<u64>,
155    /// Undo sink for rollback records (P52.18).
156    pub undo_sink: Option<Arc<Mutex<Vec<UndoRecord>>>>,
157    /// Typed WAL sink so the edge survives restarts via WAL replay (P60.2).
158    pub wal_sink: Option<WalSink>,
159}
160
161impl PhysicalOperatorExec for PhysicalInsertRel {
162    fn operator_type(&self) -> &str {
163        "insert_rel"
164    }
165
166    fn execute(&self, input: Vec<DataChunk>) -> OperatorResult {
167        let mut inserted_count = 0u64;
168        let mut table = self
169            .table_catalog
170            .get_rel_table_by_name_mut(&self.table_name)
171            .ok_or_else(|| format!("Rel table '{}' not found for INSERT", self.table_name))?;
172
173        // A filtered-out pipeline (all input chunks empty) must stay empty:
174        // nothing was created, so the output row count is zero (P53.25).
175        if !input.is_empty() && input.iter().all(|c| c.size == 0) {
176            return Ok(vec![DataChunk::new(vec![], vec![])]);
177        }
178
179        let num_cols = table.columns.len();
180        let mut rels_to_insert = Vec::new();
181
182        for chunk in &input {
183            let src_name_id = format!("{}.{}", self.src_node_name, "_id");
184            let src_name_pk = format!("{}.{}", self.src_node_name, "id");
185            let src_name_pk_upper = format!("{}.{}", self.src_node_name, "ID");
186            let src_node_col_idx = chunk
187                .field_names
188                .iter()
189                .position(|name| name == &src_name_id)
190                .or_else(|| chunk.field_names.iter().position(|name| name == &self.src_node_name))
191                .or_else(|| chunk.field_names.iter().position(|name| name == &src_name_pk))
192                .or_else(|| chunk.field_names.iter().position(|name| name == &src_name_pk_upper))
193                .or_else(|| {
194                    chunk
195                        .field_names
196                        .iter()
197                        .position(|name| name.eq_ignore_ascii_case(&src_name_pk))
198                })
199                .ok_or_else(|| {
200                    format!(
201                        "Source node variable {} not found (fields: {:?})",
202                        self.src_node_name, chunk.field_names
203                    )
204                })?;
205
206            let dst_name_id = format!("{}.{}", self.dst_node_name, "_id");
207            let dst_name_pk = format!("{}.{}", self.dst_node_name, "id");
208            let dst_name_pk_upper = format!("{}.{}", self.dst_node_name, "ID");
209            let dst_node_col_idx = chunk
210                .field_names
211                .iter()
212                .position(|name| name == &dst_name_id)
213                .or_else(|| chunk.field_names.iter().position(|name| name == &self.dst_node_name))
214                .or_else(|| chunk.field_names.iter().position(|name| name == &dst_name_pk))
215                .or_else(|| chunk.field_names.iter().position(|name| name == &dst_name_pk_upper))
216                .or_else(|| {
217                    chunk
218                        .field_names
219                        .iter()
220                        .position(|name| name.eq_ignore_ascii_case(&dst_name_pk))
221                })
222                .ok_or_else(|| {
223                    format!(
224                        "Destination node variable {} not found (fields: {:?})",
225                        self.dst_node_name, chunk.field_names
226                    )
227                })?;
228
229            if src_node_col_idx >= chunk.fields.len() || dst_node_col_idx >= chunk.fields.len() {
230                return Err("Src/Dst node column index out of bounds in INSERT REL".into());
231            }
232
233            for row in 0..chunk.size {
234                let src_id = if let Some(Value::Int64(val)) = chunk.get_value(src_node_col_idx, row) {
235                    val as u64
236                } else {
237                    0
238                };
239                let dst_id = if let Some(Value::Int64(val)) = chunk.get_value(dst_node_col_idx, row) {
240                    val as u64
241                } else {
242                    0
243                };
244
245                let mut props = vec![Value::Null; num_cols];
246                for (prop_name, expr) in &self.properties {
247                    if let Some(col_idx) = table.columns.iter().position(|c| c.name == *prop_name) {
248                        let val = evaluate_expression_for_row(expr, chunk, row);
249                        props[col_idx] = val;
250                    }
251                }
252
253                rels_to_insert.push((src_id, dst_id, props));
254            }
255        }
256
257        // Batch insert the collected relationships
258        if !rels_to_insert.is_empty() {
259            inserted_count = table
260                .insert_rels_batch(&rels_to_insert)
261                .map_err(|e| format!("BatchInsert rel error: {e}"))?;
262            for (src_id, dst_id, props) in &rels_to_insert {
263                log_rel_insert_record(&self.wal_sink, self.table_id, *src_id, *dst_id, props);
264            }
265            if let Some(sink) = self.undo_sink.as_ref()
266                && let Ok(mut u) = sink.lock()
267            {
268                let num_edges = table.edges.len();
269                for idx in (num_edges - inserted_count as usize)..num_edges {
270                    u.push(UndoRecord::insert(self.table_id, idx as u64));
271                }
272            }
273        }
274
275        tracing::info!("INSERT REL: added {inserted_count} rels to '{}'", self.table_name);
276
277        let mut v = ValueVector::new(PhysicalTypeID::Int64, 1);
278        v.resize(1);
279        v.set_i64(0, inserted_count as i64);
280        let arr = akar_common::arrow_vector::ArrowVector::from_legacy(&v).array;
281        Ok(vec![DataChunk::new(vec![arr], vec![PhysicalTypeID::Int64])])
282    }
283}
akar_processor/physical/write_ops/insert.rs

akar_processor/physical/write_ops/
insert.rs
