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 std::sync::Arc;
8
9// ==================== Delete ====================
10
11/// Physical operator for DELETE — removes rows from a node or rel table.
12pub struct PhysicalDelete {
13    pub table_name: String,
14    pub table_id: u64,
15    pub primary_key_column: String,
16    pub is_node: bool,
17    pub detach: bool,
18    /// Row indices to delete (found by the scan/filter pipeline).
19    pub row_indices: Vec<u64>,
20    pub table_catalog: Arc<TableCatalog>,
21}
22
23impl PhysicalOperatorExec for PhysicalDelete {
24    fn operator_type(&self) -> &str {
25        "delete"
26    }
27
28    fn execute(&self, input: Vec<DataChunk>) -> OperatorResult {
29        // Collect row indices from input chunks
30        let mut rows_to_delete: Vec<u64> = self.row_indices.clone();
31
32        // If input has data, extract row indices from it
33        for chunk in &input {
34            for row in 0..chunk.size {
35                if !chunk.fields.is_empty() {
36                    if let Some(akar_common::types::Value::Int64(val)) = chunk.get_value(0, row) {
37                        rows_to_delete.push(val as u64);
38                    }
39                }
40            }
41        }
42
43        if rows_to_delete.is_empty() {
44            // No rows to delete - still return success
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            return Ok(vec![DataChunk::new(vec![arr], vec![PhysicalTypeID::Int64])]);
50        }
51
52        // Delete rows from the table
53        let mut deleted = 0u64;
54        if self.is_node {
55            for &row_idx in &rows_to_delete {
56                if !self.detach && self.table_catalog.has_incident_edges(self.table_id, row_idx) {
57                    return Err(format!(
58                        "Cannot delete node {} because it has incident edges (use DETACH DELETE)",
59                        row_idx
60                    )
61                    .into());
62                }
63            }
64            if self.detach {
65                for &row_idx in &rows_to_delete {
66                    self.table_catalog.detach_node(self.table_id, row_idx);
67                }
68            }
69            if let Some(mut table) = self.table_catalog.get_node_table_by_name_mut(&self.table_name) {
70                for &row_idx in &rows_to_delete {
71                    if table.delete_row(row_idx).is_ok() {
72                        deleted += 1;
73                    }
74                }
75            } else {
76                return Err(format!("Node table '{}' not found for DELETE", self.table_name).into());
77            }
78        } else {
79            if let Some(mut table) = self.table_catalog.get_rel_table_by_name_mut(&self.table_name) {
80                for &edge_idx in &rows_to_delete {
81                    if table.delete_edge(edge_idx as usize).is_ok() {
82                        deleted += 1;
83                    }
84                }
85            } else {
86                return Err(format!("Rel table '{}' not found for DELETE", self.table_name).into());
87            }
88        }
89
90        tracing::info!("DELETE: removed {deleted} rows from '{}'", self.table_name);
91
92        let mut v = ValueVector::new(PhysicalTypeID::Int64, 1);
93        v.resize(1);
94        v.set_i64(0, deleted as i64);
95        let arr = akar_common::arrow_vector::ArrowVector::from_legacy(&v).array;
96        Ok(vec![DataChunk::new(vec![arr], vec![PhysicalTypeID::Int64])])
97    }
98}
99
100/// Convert an AST Constant to a Value.
101pub fn ast_constant_to_value(c: &Constant) -> Value {
102    match c {
103        Constant::Null => Value::Null,
104        Constant::Bool(b) => Value::Bool(*b),
105        Constant::Integer(i) => Value::Int64(*i),
106        Constant::Float(f) => Value::Double(*f),
107        Constant::String(s) => Value::String(s.clone()),
108    }
109}
akar_processor/physical/write_ops/delete.rs

akar_processor/physical/write_ops/
delete.rs
