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

akar_processor/physical/write_ops/
delete.rs
