akar_processor/physical/write_ops/
delete.rs

1//! Auto-extracted from physical_operator.rs
2use crate::physical::types::{OperatorResult, PhysicalOperatorExec};
3use akar_common::types::{PhysicalTypeID, Value};
4use akar_common::vector::{DataChunk, ValueVector};
5use akar_parser::ast::Constant;
6use akar_storage::table::TableCatalog;
7use akar_transaction::UndoRecord;
8use std::sync::{Arc, Mutex};
9
10// ==================== Delete ====================
11
12/// Physical operator for DELETE — removes rows from a node or rel table.
13pub struct PhysicalDelete {
14    pub table_name: String,
15    pub table_id: u64,
16    pub primary_key_column: String,
17    pub is_node: bool,
18    pub detach: bool,
19    /// Row indices to delete (found by the scan/filter pipeline).
20    pub row_indices: Vec<u64>,
21    pub table_catalog: Arc<TableCatalog>,
22    /// Active transaction id — deletes are recorded in `VersionInfo` for MVCC (P52.18).
23    pub txn_id: Option<u64>,
24    /// Undo sink for rollback records (P52.18).
25    pub undo_sink: Option<Arc<Mutex<Vec<UndoRecord>>>>,
26}
27
28impl PhysicalOperatorExec for PhysicalDelete {
29    fn operator_type(&self) -> &str {
30        "delete"
31    }
32
33    fn execute(&self, input: Vec<DataChunk>) -> OperatorResult {
34        // Collect row indices from input chunks
35        let mut rows_to_delete: Vec<u64> = self.row_indices.clone();
36
37        // If input has data, extract row indices from it.
38        // The scan emits the physical row index as the `<alias>._id` column
39        // (last column); reading column 0 would treat the first *property*
40        // value as a row index (wrong row or an out-of-range no-op).
41        for chunk in &input {
42            let row_id_col = row_id_column_index(chunk);
43            for row in 0..chunk.size {
44                if !chunk.fields.is_empty() {
45                    if let Some(akar_common::types::Value::Int64(val)) = chunk.get_value(row_id_col.unwrap_or(0), row) {
46                        rows_to_delete.push(val as u64);
47                    }
48                }
49            }
50        }
51
52        if rows_to_delete.is_empty() {
53            // No rows to delete — emit zero-row chunks so a following
54            // `RETURN count(*)` reports 0 (the old 1-row count chunk made it
55            // report 1 regardless; CREATE/SET already follow the per-row
56            // convention, P53.25/P53.30).
57            let mut out = Vec::with_capacity(input.len());
58            for mut chunk in input {
59                chunk.resize(0);
60                chunk.sel_vector = None;
61                out.push(chunk);
62            }
63            return Ok(out);
64        }
65
66        // Delete rows from the table
67        let mut deleted = 0u64;
68        if self.is_node {
69            for &row_idx in &rows_to_delete {
70                if !self.detach && self.table_catalog.has_incident_edges(self.table_id, row_idx) {
71                    return Err(format!(
72                        "Cannot delete node {} because it has incident edges (use DETACH DELETE)",
73                        row_idx
74                    )
75                    .into());
76                }
77            }
78            if self.detach {
79                for &row_idx in &rows_to_delete {
80                    self.table_catalog.detach_node(self.table_id, row_idx);
81                }
82            }
83            if let Some(mut table) = self.table_catalog.get_node_table_by_name_mut(&self.table_name) {
84                for &row_idx in &rows_to_delete {
85                    // Capture pre-delete row data for rollback (P52.18) and
86                    // record the delete in VersionInfo for MVCC isolation.
87                    if let Some(sink) = self.undo_sink.as_ref()
88                        && let Ok(mut u) = sink.lock()
89                    {
90                        let old_data = table.row_undo_bytes(row_idx);
91                        u.push(UndoRecord::delete(self.table_id, row_idx, old_data));
92                    }
93                    if table.delete_row_with_txn(row_idx, self.txn_id).is_ok() {
94                        deleted += 1;
95                    }
96                }
97            } else {
98                return Err(format!("Node table '{}' not found for DELETE", self.table_name).into());
99            }
100        } else {
101            if let Some(mut table) = self.table_catalog.get_rel_table_by_name_mut(&self.table_name) {
102                for &edge_idx in &rows_to_delete {
103                    if let Some(sink) = self.undo_sink.as_ref()
104                        && let Ok(mut u) = sink.lock()
105                    {
106                        let old_data = table.edge_undo_bytes(edge_idx as usize);
107                        u.push(UndoRecord::delete(self.table_id, edge_idx, old_data));
108                    }
109                    if table.delete_edge(edge_idx as usize).is_ok() {
110                        deleted += 1;
111                    }
112                }
113            } else {
114                return Err(format!("Rel table '{}' not found for DELETE", self.table_name).into());
115            }
116        }
117
118        tracing::info!("DELETE: removed {deleted} rows from '{}'", self.table_name);
119
120        // Emit one row per deleted row so a following `RETURN count(*)` reports
121        // the actual deleted count (P53.37c). Column 0 carries the deleted count
122        // in every row, preserving the count-chunk contract for consumers that
123        // read `get_i64(0, 0)`. An all-failed delete emits zero rows.
124        let n = deleted as usize;
125        let mut v = ValueVector::new(PhysicalTypeID::Int64, n);
126        v.resize(n);
127        for i in 0..n {
128            v.set_i64(i, deleted as i64);
129        }
130        let arr = akar_common::arrow_vector::ArrowVector::from_legacy(&v).array;
131        Ok(vec![DataChunk::new(vec![arr], vec![PhysicalTypeID::Int64])])
132    }
133}
134
135/// Convert an AST Constant to a Value.
136pub fn ast_constant_to_value(c: &Constant) -> Value {
137    match c {
138        Constant::Null => Value::Null,
139        Constant::Bool(b) => Value::Bool(*b),
140        Constant::Integer(i) => Value::Int64(*i),
141        Constant::Float(f) => Value::Double(*f),
142        Constant::String(s) => Value::String(s.clone()),
143    }
144}
145
146/// Locate the physical row index column in a scan-produced chunk.
147///
148/// Node scans append an internal node id column (`<alias>._id` = row offset)
149/// as the last column of each chunk (see `resolve_scan_arrow_data` and
150/// `resolve_scan_data`). Write operators (DELETE/SET) must read row indices
151/// from that column; reading column 0 would use the first *property* value
152/// as a row index. Falls back to `None` when the chunk carries no `_id`
153/// column (e.g. rel table scans), letting callers keep the legacy behaviour.
154pub fn row_id_column_index(chunk: &DataChunk) -> Option<usize> {
155    chunk.field_names.iter().position(|n| n == "_id" || n.ends_with("._id"))
156}
akar_processor/physical/write_ops/delete.rs

akar_processor/physical/write_ops/
delete.rs
