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};
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 table = self
41            .table_catalog
42            .get_node_table_by_name_mut(&self.table_name)
43            .ok_or_else(|| format!("Node table '{}' not found for INSERT", self.table_name))?;
44
45        // If input is empty (no previous pipeline), we insert exactly one node.
46        // Otherwise, we insert a node for each row in the input.
47        let chunks = if input.is_empty() {
48            let mut v = ValueVector::new(PhysicalTypeID::Int64, 1);
49            v.resize(1);
50            v.set_i64(0, 0);
51            let arr = akar_common::arrow_vector::ArrowVector::from_legacy(&v).array;
52            vec![DataChunk::new(vec![arr], vec![PhysicalTypeID::Int64])]
53        } else {
54            input
55        };
56
57        let num_cols = table.columns.len();
58        for chunk in &chunks {
59            for row in 0..chunk.size {
60                let mut row_values = vec![Value::Null; num_cols];
61
62                for (prop_name, expr) in &self.properties {
63                    if let Some(col_idx) = table.columns.iter().position(|c| c.name == *prop_name) {
64                        let val = evaluate_expression_for_row(expr, chunk, row);
65                        row_values[col_idx] = val;
66                    }
67                }
68
69                // Add the row to the node table; capture assigned row_id for OCC.
70                // Errors (e.g. NULL primary key) must surface, not silently skip
71                // the row — otherwise UNWIND+CREATE drops input rows (P53.27).
72                let logged_row = self.wal_sink.is_some().then(|| row_values.clone());
73                let row_id = table
74                    .insert_row_with_txn(row_values, self.txn_id)
75                    .map_err(|e| format!("INSERT NODE row {row} failed in '{}': {e}", self.table_name))?;
76                assigned_row_ids.push(row_id as i64);
77                log_insert_record(&self.wal_sink, self.table_id, logged_row.as_deref().unwrap_or(&[]));
78                if let Some(sink) = self.undo_sink.as_ref()
79                    && let Ok(mut u) = sink.lock()
80                {
81                    u.push(UndoRecord::insert(self.table_id, row_id));
82                }
83            }
84        }
85
86        let inserted_count = assigned_row_ids.len();
87        tracing::info!("INSERT NODE: added {inserted_count} rows to '{}'", self.table_name);
88
89        let mut count_v = ValueVector::new(PhysicalTypeID::Int64, 1);
90        count_v.resize(1);
91        count_v.set_i64(0, inserted_count as i64);
92        let arr_count = akar_common::arrow_vector::ArrowVector::from_legacy(&count_v).array;
93
94        // Column 1: assigned row IDs (used by record_insert_writes for OCC row-level tracking)
95        let mut ids_v = ValueVector::new(PhysicalTypeID::Int64, assigned_row_ids.len());
96        ids_v.resize(assigned_row_ids.len());
97        for (i, rid) in assigned_row_ids.iter().enumerate() {
98            ids_v.set_i64(i, *rid);
99        }
100        let arr_ids = akar_common::arrow_vector::ArrowVector::from_legacy(&ids_v).array;
101
102        Ok(vec![DataChunk::new(
103            vec![arr_count, arr_ids],
104            vec![PhysicalTypeID::Int64, PhysicalTypeID::Int64],
105        )])
106    }
107}
108
109/// Physical operator for CREATE REL.
110pub struct PhysicalInsertRel {
111    pub table_name: String,
112    pub table_id: u64,
113    pub src_node_name: String,
114    pub dst_node_name: String,
115    pub properties: Vec<(String, akar_parser::ast::Expression)>,
116    pub table_catalog: Arc<TableCatalog>,
117    /// Active transaction id for MVCC + undo recording (P52.18).
118    pub txn_id: Option<u64>,
119    /// Undo sink for rollback records (P52.18).
120    pub undo_sink: Option<Arc<Mutex<Vec<UndoRecord>>>>,
121    /// Typed WAL sink so the edge survives restarts via WAL replay (P60.2).
122    pub wal_sink: Option<WalSink>,
123}
124
125impl PhysicalOperatorExec for PhysicalInsertRel {
126    fn operator_type(&self) -> &str {
127        "insert_rel"
128    }
129
130    fn execute(&self, input: Vec<DataChunk>) -> OperatorResult {
131        let mut inserted_count = 0u64;
132        let mut table = self
133            .table_catalog
134            .get_rel_table_by_name_mut(&self.table_name)
135            .ok_or_else(|| format!("Rel table '{}' not found for INSERT", self.table_name))?;
136
137        // A filtered-out pipeline (all input chunks empty) must stay empty:
138        // nothing was created, so the output row count is zero (P53.25).
139        if !input.is_empty() && input.iter().all(|c| c.size == 0) {
140            return Ok(vec![DataChunk::new(vec![], vec![])]);
141        }
142
143        let num_cols = table.columns.len();
144        let mut rels_to_insert = Vec::new();
145
146        for chunk in &input {
147            let src_name_id = format!("{}.{}", self.src_node_name, "_id");
148            let src_name_pk = format!("{}.{}", self.src_node_name, "id");
149            let src_name_pk_upper = format!("{}.{}", self.src_node_name, "ID");
150            let src_node_col_idx = chunk
151                .field_names
152                .iter()
153                .position(|name| name == &src_name_id)
154                .or_else(|| chunk.field_names.iter().position(|name| name == &self.src_node_name))
155                .or_else(|| chunk.field_names.iter().position(|name| name == &src_name_pk))
156                .or_else(|| chunk.field_names.iter().position(|name| name == &src_name_pk_upper))
157                .or_else(|| {
158                    chunk
159                        .field_names
160                        .iter()
161                        .position(|name| name.eq_ignore_ascii_case(&src_name_pk))
162                })
163                .ok_or_else(|| {
164                    format!(
165                        "Source node variable {} not found (fields: {:?})",
166                        self.src_node_name, chunk.field_names
167                    )
168                })?;
169
170            let dst_name_id = format!("{}.{}", self.dst_node_name, "_id");
171            let dst_name_pk = format!("{}.{}", self.dst_node_name, "id");
172            let dst_name_pk_upper = format!("{}.{}", self.dst_node_name, "ID");
173            let dst_node_col_idx = chunk
174                .field_names
175                .iter()
176                .position(|name| name == &dst_name_id)
177                .or_else(|| chunk.field_names.iter().position(|name| name == &self.dst_node_name))
178                .or_else(|| chunk.field_names.iter().position(|name| name == &dst_name_pk))
179                .or_else(|| chunk.field_names.iter().position(|name| name == &dst_name_pk_upper))
180                .or_else(|| {
181                    chunk
182                        .field_names
183                        .iter()
184                        .position(|name| name.eq_ignore_ascii_case(&dst_name_pk))
185                })
186                .ok_or_else(|| {
187                    format!(
188                        "Destination node variable {} not found (fields: {:?})",
189                        self.dst_node_name, chunk.field_names
190                    )
191                })?;
192
193            if src_node_col_idx >= chunk.fields.len() || dst_node_col_idx >= chunk.fields.len() {
194                return Err("Src/Dst node column index out of bounds in INSERT REL".into());
195            }
196
197            for row in 0..chunk.size {
198                let src_id = if let Some(Value::Int64(val)) = chunk.get_value(src_node_col_idx, row) {
199                    val as u64
200                } else {
201                    0
202                };
203                let dst_id = if let Some(Value::Int64(val)) = chunk.get_value(dst_node_col_idx, row) {
204                    val as u64
205                } else {
206                    0
207                };
208
209                let mut props = vec![Value::Null; num_cols];
210                for (prop_name, expr) in &self.properties {
211                    if let Some(col_idx) = table.columns.iter().position(|c| c.name == *prop_name) {
212                        let val = evaluate_expression_for_row(expr, chunk, row);
213                        props[col_idx] = val;
214                    }
215                }
216
217                rels_to_insert.push((src_id, dst_id, props));
218            }
219        }
220
221        // Batch insert the collected relationships
222        if !rels_to_insert.is_empty() {
223            inserted_count = table
224                .insert_rels_batch(&rels_to_insert)
225                .map_err(|e| format!("BatchInsert rel error: {e}"))?;
226            for (src_id, dst_id, props) in &rels_to_insert {
227                log_rel_insert_record(&self.wal_sink, self.table_id, *src_id, *dst_id, props);
228            }
229            if let Some(sink) = self.undo_sink.as_ref()
230                && let Ok(mut u) = sink.lock()
231            {
232                let num_edges = table.edges.len();
233                for idx in (num_edges - inserted_count as usize)..num_edges {
234                    u.push(UndoRecord::insert(self.table_id, idx as u64));
235                }
236            }
237        }
238
239        tracing::info!("INSERT REL: added {inserted_count} rels to '{}'", self.table_name);
240
241        let mut v = ValueVector::new(PhysicalTypeID::Int64, 1);
242        v.resize(1);
243        v.set_i64(0, inserted_count as i64);
244        let arr = akar_common::arrow_vector::ArrowVector::from_legacy(&v).array;
245        Ok(vec![DataChunk::new(vec![arr], vec![PhysicalTypeID::Int64])])
246    }
247}
akar_processor/physical/write_ops/insert.rs

akar_processor/physical/write_ops/
insert.rs
