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

akar_processor/physical/write_ops/
insert.rs
