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

akar_processor/physical/write_ops/
insert.rs
