akar_processor/physical/
misc.rs

1//! Miscellaneous physical operators (EmptyResult, MultiplicityReducer, Skip, UnionAllScan).
2
3use crate::physical::types::{OperatorResult, PhysicalOperatorExec};
4use akar_common::vector::{DataChunk, ValueVector};
5
6/// Physical operator that always returns an empty result.
7pub struct PhysicalEmptyResult;
8
9impl PhysicalOperatorExec for PhysicalEmptyResult {
10    fn operator_type(&self) -> &str {
11        "empty_result"
12    }
13
14    fn execute(&self, _input: Vec<DataChunk>) -> OperatorResult {
15        Ok(vec![])
16    }
17}
18
19/// Physical operator that reduces the multiplicity of paths (e.g., DISTINCT).
20pub struct PhysicalMultiplicityReducer {
21    pub key_columns: Vec<usize>,
22}
23
24impl PhysicalOperatorExec for PhysicalMultiplicityReducer {
25    fn operator_type(&self) -> &str {
26        "multiplicity_reducer"
27    }
28
29    fn execute(&self, input: Vec<DataChunk>) -> OperatorResult {
30        if input.is_empty() {
31            return Ok(input);
32        }
33
34        let mut result = Vec::new();
35        let mut seen = std::collections::HashSet::new();
36
37        for chunk in input {
38            let mut filter_mask = vec![false; chunk.size];
39            for i in 0..chunk.size {
40                let mut row_keys = Vec::new();
41                for &col_idx in &self.key_columns {
42                    let val = chunk.get_value(col_idx, i).unwrap_or(akar_common::types::Value::Null);
43                    row_keys.push(val);
44                }
45
46                // Using debug format as a fallback hashable representation for arbitrary Value types
47                if seen.insert(format!("{:?}", row_keys)) {
48                    filter_mask[i] = true;
49                }
50            }
51
52            let filtered_size = filter_mask.iter().filter(|&&b| b).count();
53            if filtered_size > 0 {
54                let mut new_fields = Vec::new();
55                let mut new_field_types = Vec::new();
56                for (col_idx, _field) in chunk.fields.iter().enumerate() {
57                    let phys_type = chunk.field_types[col_idx];
58                    let mut new_field = ValueVector::new(phys_type, filtered_size);
59                    let mut current = 0;
60                    for (i, &keep) in filter_mask.iter().enumerate() {
61                        if keep {
62                            if let Some(val) = chunk.get_value(col_idx, i) {
63                                let _ = new_field.set_value(current, &val);
64                            }
65                            current += 1;
66                        }
67                    }
68                    new_fields.push(akar_common::arrow_vector::ArrowVector::from_legacy(&new_field).array);
69                    new_field_types.push(phys_type);
70                }
71                result.push(DataChunk {
72                    fields: new_fields,
73                    field_types: new_field_types,
74                    size: filtered_size,
75                    field_names: chunk.field_names.clone(),
76                    sel_vector: None,
77                });
78            }
79        }
80        Ok(result)
81    }
82}
83
84/// Physical operator for SKIP (OFFSET) in queries.
85pub struct PhysicalSkip {
86    pub skip_count: usize,
87}
88
89impl PhysicalOperatorExec for PhysicalSkip {
90    fn operator_type(&self) -> &str {
91        "skip"
92    }
93
94    fn execute(&self, input: Vec<DataChunk>) -> OperatorResult {
95        let mut remaining_skip = self.skip_count;
96        let mut output = Vec::new();
97
98        for chunk in input {
99            if remaining_skip == 0 {
100                output.push(chunk);
101                continue;
102            }
103
104            if chunk.size <= remaining_skip {
105                remaining_skip -= chunk.size;
106                continue;
107            }
108
109            // Partially skip this chunk
110            let keep_size = chunk.size - remaining_skip;
111            let mut sliced_fields = Vec::new();
112            let mut sliced_types = Vec::new();
113
114            for (col_idx, _field) in chunk.fields.iter().enumerate() {
115                let phys_type = chunk.field_types[col_idx];
116                let mut new_field = ValueVector::new(phys_type, keep_size);
117                for i in 0..keep_size {
118                    let val = chunk
119                        .get_value(col_idx, remaining_skip + i)
120                        .unwrap_or(akar_common::types::Value::Null);
121                    // set_value returns Err only for strings > 255 bytes (legacy
122                    // inline storage limit); drop to NULL instead of panicking.
123                    if new_field.set_value(i, &val).is_err() {
124                        new_field.set_null(i, true);
125                    }
126                }
127                sliced_fields.push(akar_common::arrow_vector::ArrowVector::from_legacy(&new_field).array);
128                sliced_types.push(phys_type);
129            }
130
131            output.push(DataChunk {
132                fields: sliced_fields,
133                field_types: sliced_types,
134                size: keep_size,
135                field_names: chunk.field_names.clone(),
136                sel_vector: None,
137            });
138            remaining_skip = 0;
139        }
140
141        Ok(output)
142    }
143}
144
145/// Physical operator for scanning from a UNION ALL.
146pub struct PhysicalUnionAllScan;
147
148impl PhysicalOperatorExec for PhysicalUnionAllScan {
149    fn operator_type(&self) -> &str {
150        "union_all_scan"
151    }
152
153    fn execute(&self, input: Vec<DataChunk>) -> OperatorResult {
154        Ok(input)
155    }
156}
157
158/// Insert operator — row-level insertion (unlike BatchInsert).
159pub struct PhysicalInsert {
160    pub table_name: String,
161    pub table_id: u64,
162    pub columns: Vec<String>,
163    pub values: Vec<Vec<akar_common::types::Value>>,
164    pub table_catalog: std::sync::Arc<akar_storage::table::TableCatalog>,
165    /// Active transaction id (P52.18).
166    pub txn_id: Option<u64>,
167    /// Undo sink for rollback records (P52.18).
168    pub undo_sink: Option<std::sync::Arc<std::sync::Mutex<Vec<akar_transaction::UndoRecord>>>>,
169    /// Typed WAL sink so inserted rows/edges survive restarts via replay (P60.2).
170    pub wal_sink: Option<akar_storage::wal::WalSink>,
171}
172
173impl PhysicalOperatorExec for PhysicalInsert {
174    fn operator_type(&self) -> &str {
175        "insert"
176    }
177
178    fn execute(&self, _input: Vec<DataChunk>) -> OperatorResult {
179        let mut inserted = 0;
180
181        // Cek jika tabel adalah rel table atau node table
182        if let Some(mut rel_tbl) = self.table_catalog.get_rel_table_by_name_mut(&self.table_name) {
183            // Rel Table Insert
184            let mut rels_to_insert = Vec::new();
185            for row_values in &self.values {
186                if row_values.len() >= 2 {
187                    // Extract src and dst
188                    let src = if let akar_common::types::Value::Int64(v) = row_values[0] {
189                        v as u64
190                    } else {
191                        0
192                    };
193                    let dst = if let akar_common::types::Value::Int64(v) = row_values[1] {
194                        v as u64
195                    } else {
196                        0
197                    };
198                    let props = if row_values.len() > 2 {
199                        row_values[2..].to_vec()
200                    } else {
201                        vec![]
202                    };
203                    rels_to_insert.push((src, dst, props));
204                }
205            }
206            if !rels_to_insert.is_empty() {
207                let start = rel_tbl.edges.len();
208                if let Ok(count) = rel_tbl.insert_rels_batch(&rels_to_insert) {
209                    inserted += count;
210                    for (src, dst, props) in &rels_to_insert {
211                        akar_storage::wal::log_rel_insert_record(&self.wal_sink, self.table_id, *src, *dst, props);
212                    }
213                    if let Some(sink) = self.undo_sink.as_ref()
214                        && let Ok(mut u) = sink.lock()
215                    {
216                        for idx in start..start + count as usize {
217                            u.push(akar_transaction::UndoRecord::insert(self.table_id, idx as u64));
218                        }
219                    }
220                }
221            }
222        } else if let Some(mut node_tbl) = self.table_catalog.get_node_table_by_name_mut(&self.table_name) {
223            // Node Table Insert
224            for row_values in &self.values {
225                if let Ok(row_id) = node_tbl.insert_row_with_txn(row_values.clone(), self.txn_id) {
226                    inserted += 1;
227                    akar_storage::wal::log_insert_record(&self.wal_sink, self.table_id, row_values);
228                    if let Some(sink) = self.undo_sink.as_ref()
229                        && let Ok(mut u) = sink.lock()
230                    {
231                        u.push(akar_transaction::UndoRecord::insert(self.table_id, row_id));
232                    }
233                }
234            }
235        } else {
236            return Err(format!("Table '{}' not found for INSERT", self.table_name).into());
237        }
238
239        let mut v = ValueVector::new(akar_common::types::PhysicalTypeID::Int64, 1);
240        v.resize(1);
241        v.set_i64(0, inserted as i64);
242        let arr = akar_common::arrow_vector::ArrowVector::from_legacy(&v).array;
243        Ok(vec![DataChunk::new(
244            vec![arr],
245            vec![akar_common::types::PhysicalTypeID::Int64],
246        )])
247    }
248}
249
250/// ExtensionClause operator — handles EXTENSION commands (INSTALL, LOAD).
251pub struct PhysicalExtensionClause {
252    pub action: akar_parser::ast::ExtensionAction,
253    pub extension_name: String,
254}
255
256impl PhysicalOperatorExec for PhysicalExtensionClause {
257    fn operator_type(&self) -> &str {
258        "extension_clause"
259    }
260
261    fn execute(&self, _input: Vec<DataChunk>) -> OperatorResult {
262        let msg = match self.action {
263            akar_parser::ast::ExtensionAction::Install => {
264                format!("Extension '{}' installed successfully.", self.extension_name)
265            }
266            akar_parser::ast::ExtensionAction::Load => {
267                // Pseudo-registry for static extensions
268                if self.extension_name.to_lowercase() == "httpfs" {
269                    format!(
270                        "Extension '{}' loaded (HTTP/S3 virtual file system).",
271                        self.extension_name
272                    )
273                } else if self.extension_name.to_lowercase() == "fts" {
274                    format!("Extension '{}' loaded (Full Text Search).", self.extension_name)
275                } else {
276                    format!("Extension '{}' loaded.", self.extension_name)
277                }
278            }
279            akar_parser::ast::ExtensionAction::Uninstall => {
280                format!("Extension '{}' uninstalled.", self.extension_name)
281            }
282        };
283
284        tracing::info!("{}", msg);
285
286        let mut field = ValueVector::new(akar_common::types::PhysicalTypeID::String, 1);
287        let _ = field.set_value(0, &akar_common::types::Value::String(msg));
288        let arr = akar_common::arrow_vector::ArrowVector::from_legacy(&field).array;
289
290        Ok(vec![DataChunk {
291            fields: vec![arr],
292            field_types: vec![akar_common::types::PhysicalTypeID::String],
293            size: 1,
294            field_names: vec!["message".into()],
295            sel_vector: None,
296        }])
297    }
298}
akar_processor/physical/misc.rs

akar_processor/physical/
misc.rs
