akar_processor/physical/
misc.rs

1//! Miscellaneous physical operators (EmptyResult, MultiplicityReducer, Skip, UnionAllScan).
2
3use crate::physical::common::hash_row;
4use crate::physical::types::{OperatorResult, PhysicalOperatorExec};
5use akar_common::types::Value;
6use akar_common::vector::{DataChunk, ValueVector};
7
8/// Physical operator that always returns an empty result.
9pub struct PhysicalEmptyResult;
10
11impl PhysicalOperatorExec for PhysicalEmptyResult {
12    fn operator_type(&self) -> &str {
13        "empty_result"
14    }
15
16    fn execute(&self, _input: Vec<DataChunk>) -> OperatorResult {
17        Ok(vec![])
18    }
19}
20
21/// Physical operator that reduces the multiplicity of paths (e.g., DISTINCT).
22pub struct PhysicalMultiplicityReducer {
23    pub key_columns: Vec<usize>,
24}
25
26impl PhysicalOperatorExec for PhysicalMultiplicityReducer {
27    fn operator_type(&self) -> &str {
28        "multiplicity_reducer"
29    }
30
31    fn execute(&self, input: Vec<DataChunk>) -> OperatorResult {
32        if input.is_empty() {
33            return Ok(input);
34        }
35
36        let mut result = Vec::new();
37        // Hash-bucket membership index over kept keys: hash -> indices into
38        // `kept_keys`. Exact row equality is checked only on hash collision,
39        // so two distinct rows sharing a hash are never wrongly merged. This
40        // replaces the old per-row `format!("{:?}", row_keys)` allocation +
41        // `HashSet<String>` dedup (O(1) per row, deterministic equality).
42        let mut buckets: std::collections::HashMap<u64, Vec<usize>> = std::collections::HashMap::new();
43        let mut kept_keys: Vec<Vec<Value>> = Vec::new();
44
45        for chunk in input {
46            let mut filter_mask = vec![false; chunk.size];
47            for i in 0..chunk.size {
48                let mut row_keys = Vec::with_capacity(self.key_columns.len());
49                for &col_idx in &self.key_columns {
50                    let val = chunk.get_value(col_idx, i).unwrap_or(Value::Null);
51                    row_keys.push(val);
52                }
53
54                let hash = hash_row(&row_keys);
55                let is_dup = buckets
56                    .get(&hash)
57                    .is_some_and(|bucket| bucket.iter().any(|&k| kept_keys[k] == row_keys));
58                if !is_dup {
59                    match buckets.get_mut(&hash) {
60                        Some(bucket) => bucket.push(kept_keys.len()),
61                        None => {
62                            buckets.insert(hash, vec![kept_keys.len()]);
63                        }
64                    }
65                    kept_keys.push(row_keys);
66                    filter_mask[i] = true;
67                }
68            }
69
70            let filtered_size = filter_mask.iter().filter(|&&b| b).count();
71            if filtered_size > 0 {
72                let mut new_fields = Vec::new();
73                let mut new_field_types = Vec::new();
74                for (col_idx, _field) in chunk.fields.iter().enumerate() {
75                    let phys_type = chunk.field_types[col_idx];
76                    let mut new_field = ValueVector::new(phys_type, filtered_size);
77                    let mut current = 0;
78                    for (i, &keep) in filter_mask.iter().enumerate() {
79                        if keep {
80                            if let Some(val) = chunk.get_value(col_idx, i) {
81                                let _ = new_field.set_value(current, &val);
82                            }
83                            current += 1;
84                        }
85                    }
86                    new_fields.push(akar_common::arrow_vector::ArrowVector::from_legacy(&new_field).array);
87                    new_field_types.push(phys_type);
88                }
89                result.push(DataChunk {
90                    fields: new_fields,
91                    field_types: new_field_types,
92                    size: filtered_size,
93                    field_names: chunk.field_names.clone(),
94                    sel_vector: None,
95                });
96            }
97        }
98        Ok(result)
99    }
100}
101
102/// Physical operator for SKIP (OFFSET) in queries.
103pub struct PhysicalSkip {
104    pub skip_count: usize,
105}
106
107impl PhysicalOperatorExec for PhysicalSkip {
108    fn operator_type(&self) -> &str {
109        "skip"
110    }
111
112    fn execute(&self, input: Vec<DataChunk>) -> OperatorResult {
113        let mut remaining_skip = self.skip_count;
114        let mut output = Vec::new();
115
116        for chunk in input {
117            if remaining_skip == 0 {
118                output.push(chunk);
119                continue;
120            }
121
122            if chunk.size <= remaining_skip {
123                remaining_skip -= chunk.size;
124                continue;
125            }
126
127            // Partially skip this chunk
128            let keep_size = chunk.size - remaining_skip;
129            let mut sliced_fields = Vec::new();
130            let mut sliced_types = Vec::new();
131
132            for (col_idx, _field) in chunk.fields.iter().enumerate() {
133                let phys_type = chunk.field_types[col_idx];
134                let mut new_field = ValueVector::new(phys_type, keep_size);
135                for i in 0..keep_size {
136                    let val = chunk
137                        .get_value(col_idx, remaining_skip + i)
138                        .unwrap_or(akar_common::types::Value::Null);
139                    // set_value returns Err only for strings > 255 bytes (legacy
140                    // inline storage limit); drop to NULL instead of panicking.
141                    if new_field.set_value(i, &val).is_err() {
142                        new_field.set_null(i, true);
143                    }
144                }
145                sliced_fields.push(akar_common::arrow_vector::ArrowVector::from_legacy(&new_field).array);
146                sliced_types.push(phys_type);
147            }
148
149            output.push(DataChunk {
150                fields: sliced_fields,
151                field_types: sliced_types,
152                size: keep_size,
153                field_names: chunk.field_names.clone(),
154                sel_vector: None,
155            });
156            remaining_skip = 0;
157        }
158
159        Ok(output)
160    }
161}
162
163/// Physical operator for scanning from a UNION ALL.
164pub struct PhysicalUnionAllScan;
165
166impl PhysicalOperatorExec for PhysicalUnionAllScan {
167    fn operator_type(&self) -> &str {
168        "union_all_scan"
169    }
170
171    fn execute(&self, input: Vec<DataChunk>) -> OperatorResult {
172        Ok(input)
173    }
174}
175
176/// Insert operator — row-level insertion (unlike BatchInsert).
177pub struct PhysicalInsert {
178    pub table_name: String,
179    pub table_id: u64,
180    pub columns: Vec<String>,
181    pub values: Vec<Vec<akar_common::types::Value>>,
182    pub table_catalog: std::sync::Arc<akar_storage::table::TableCatalog>,
183    /// Active transaction id (P52.18).
184    pub txn_id: Option<u64>,
185    /// Undo sink for rollback records (P52.18).
186    pub undo_sink: Option<std::sync::Arc<std::sync::Mutex<Vec<akar_transaction::UndoRecord>>>>,
187    /// Typed WAL sink so inserted rows/edges survive restarts via replay (P60.2).
188    pub wal_sink: Option<akar_storage::wal::WalSink>,
189}
190
191impl PhysicalOperatorExec for PhysicalInsert {
192    fn operator_type(&self) -> &str {
193        "insert"
194    }
195
196    fn execute(&self, _input: Vec<DataChunk>) -> OperatorResult {
197        let mut inserted = 0;
198
199        // Cek jika tabel adalah rel table atau node table
200        if let Some(mut rel_tbl) = self.table_catalog.get_rel_table_by_name_mut(&self.table_name) {
201            // Rel Table Insert
202            let mut rels_to_insert = Vec::new();
203            for row_values in &self.values {
204                if row_values.len() >= 2 {
205                    // Extract src and dst
206                    let src = if let akar_common::types::Value::Int64(v) = row_values[0] {
207                        v as u64
208                    } else {
209                        0
210                    };
211                    let dst = if let akar_common::types::Value::Int64(v) = row_values[1] {
212                        v as u64
213                    } else {
214                        0
215                    };
216                    let props = if row_values.len() > 2 {
217                        row_values[2..].to_vec()
218                    } else {
219                        vec![]
220                    };
221                    rels_to_insert.push((src, dst, props));
222                }
223            }
224            if !rels_to_insert.is_empty() {
225                let start = rel_tbl.edges.len();
226                if let Ok(count) = rel_tbl.insert_rels_batch(&rels_to_insert) {
227                    inserted += count;
228                    for (src, dst, props) in &rels_to_insert {
229                        akar_storage::wal::log_rel_insert_record(&self.wal_sink, self.table_id, *src, *dst, props);
230                    }
231                    if let Some(sink) = self.undo_sink.as_ref()
232                        && let Ok(mut u) = sink.lock()
233                    {
234                        for idx in start..start + count as usize {
235                            u.push(akar_transaction::UndoRecord::insert(self.table_id, idx as u64));
236                        }
237                    }
238                }
239            }
240        } else if let Some(mut node_tbl) = self.table_catalog.get_node_table_by_name_mut(&self.table_name) {
241            // Node Table Insert
242            for row_values in &self.values {
243                if let Ok(row_id) = node_tbl.insert_row_with_txn(row_values.clone(), self.txn_id) {
244                    inserted += 1;
245                    akar_storage::wal::log_insert_record(&self.wal_sink, self.table_id, row_values);
246                    if let Some(sink) = self.undo_sink.as_ref()
247                        && let Ok(mut u) = sink.lock()
248                    {
249                        u.push(akar_transaction::UndoRecord::insert(self.table_id, row_id));
250                    }
251                }
252            }
253        } else {
254            return Err(format!("Table '{}' not found for INSERT", self.table_name).into());
255        }
256
257        let mut v = ValueVector::new(akar_common::types::PhysicalTypeID::Int64, 1);
258        v.resize(1);
259        v.set_i64(0, inserted as i64);
260        let arr = akar_common::arrow_vector::ArrowVector::from_legacy(&v).array;
261        Ok(vec![DataChunk::new(
262            vec![arr],
263            vec![akar_common::types::PhysicalTypeID::Int64],
264        )])
265    }
266}
267
268/// ExtensionClause operator — handles EXTENSION commands (INSTALL, LOAD).
269pub struct PhysicalExtensionClause {
270    pub action: akar_parser::ast::ExtensionAction,
271    pub extension_name: String,
272}
273
274impl PhysicalOperatorExec for PhysicalExtensionClause {
275    fn operator_type(&self) -> &str {
276        "extension_clause"
277    }
278
279    fn execute(&self, _input: Vec<DataChunk>) -> OperatorResult {
280        let msg = match self.action {
281            akar_parser::ast::ExtensionAction::Install => {
282                format!("Extension '{}' installed successfully.", self.extension_name)
283            }
284            akar_parser::ast::ExtensionAction::Load => {
285                // Pseudo-registry for static extensions
286                if self.extension_name.to_lowercase() == "httpfs" {
287                    format!(
288                        "Extension '{}' loaded (HTTP/S3 virtual file system).",
289                        self.extension_name
290                    )
291                } else if self.extension_name.to_lowercase() == "fts" {
292                    format!("Extension '{}' loaded (Full Text Search).", self.extension_name)
293                } else {
294                    format!("Extension '{}' loaded.", self.extension_name)
295                }
296            }
297            akar_parser::ast::ExtensionAction::Uninstall => {
298                format!("Extension '{}' uninstalled.", self.extension_name)
299            }
300        };
301
302        tracing::info!("{}", msg);
303
304        let mut field = ValueVector::new(akar_common::types::PhysicalTypeID::String, 1);
305        let _ = field.set_value(0, &akar_common::types::Value::String(msg));
306        let arr = akar_common::arrow_vector::ArrowVector::from_legacy(&field).array;
307
308        Ok(vec![DataChunk {
309            fields: vec![arr],
310            field_types: vec![akar_common::types::PhysicalTypeID::String],
311            size: 1,
312            field_names: vec!["message".into()],
313            sel_vector: None,
314        }])
315    }
316}
317
318#[cfg(test)]
319mod tests {
320    use super::*;
321    use akar_common::types::PhysicalTypeID;
322
323    fn make_chunk(cols: &[Vec<Value>], names: &[&str]) -> DataChunk {
324        let mut fields = Vec::with_capacity(cols.len());
325        let mut types = Vec::with_capacity(cols.len());
326        for col in cols {
327            let first = col
328                .iter()
329                .find(|v| !matches!(v, Value::Null))
330                .unwrap_or(&Value::Int64(0));
331            let ptype = match first {
332                Value::Int64(_) => PhysicalTypeID::Int64,
333                Value::Double(_) => PhysicalTypeID::Double,
334                Value::Float(_) => PhysicalTypeID::Float,
335                Value::Bool(_) => PhysicalTypeID::Bool,
336                Value::String(_) => PhysicalTypeID::String,
337                _ => PhysicalTypeID::Int64,
338            };
339            let mut v = ValueVector::new(ptype, col.len().max(1));
340            for (i, val) in col.iter().enumerate() {
341                let _ = v.set_value(i, val);
342            }
343            v.resize(col.len());
344            fields.push(akar_common::arrow_vector::ArrowVector::from_legacy(&v).array);
345            types.push(ptype);
346        }
347        let mut chunk = DataChunk::new(fields, types);
348        chunk.field_names = names.iter().map(|s| s.to_string()).collect();
349        chunk
350    }
351
352    fn reducer(key_columns: Vec<usize>) -> PhysicalMultiplicityReducer {
353        PhysicalMultiplicityReducer { key_columns }
354    }
355
356    #[test]
357    fn test_multiplicity_reducer_empty_input_returns_empty() {
358        let out = reducer(vec![0]).execute(Vec::new()).unwrap();
359        assert!(out.is_empty());
360    }
361
362    #[test]
363    fn test_multiplicity_reducer_dedup_across_chunks() {
364        // The same key appearing in a later chunk must be dropped (the seen set
365        // persists across all input chunks); first-seen order is preserved.
366        let c1 = make_chunk(&[vec![Value::Int64(1), Value::Int64(2)]], &["x"]);
367        let c2 = make_chunk(&[vec![Value::Int64(2), Value::Int64(3)]], &["x"]);
368        let out = reducer(vec![0]).execute(vec![c1, c2]).unwrap();
369        assert_eq!(out.len(), 2, "one output chunk per non-empty filtered chunk");
370        assert_eq!(out[0].size, 2);
371        assert_eq!(out[1].size, 1, "row 2 is a duplicate of c1's row 2");
372        assert_eq!(out[0].get_value(0, 0), Some(Value::Int64(1)));
373        assert_eq!(out[0].get_value(0, 1), Some(Value::Int64(2)));
374        assert_eq!(out[1].get_value(0, 0), Some(Value::Int64(3)));
375    }
376
377    #[test]
378    fn test_multiplicity_reducer_key_columns_subset() {
379        // Only key_columns participate in the dedup key: second row shares its
380        // key with the first so it is dropped even though col 1 differs.
381        let chunk = make_chunk(
382            &[
383                vec![Value::Int64(1), Value::Int64(1)],
384                vec![Value::String("a".into()), Value::String("b".into())],
385            ],
386            &["id", "tag"],
387        );
388        let out = reducer(vec![0]).execute(vec![chunk]).unwrap();
389        assert_eq!(out[0].size, 1);
390        assert_eq!(out[0].get_value(0, 0), Some(Value::Int64(1)));
391        assert_eq!(out[0].get_value(1, 0), Some(Value::String("a".into())));
392    }
393
394    #[test]
395    fn test_multiplicity_reducer_multicolumn_key() {
396        // The key is the whole tuple of key_columns, not the first one alone:
397        // rows sharing col 0 but differing in col 1 are distinct.
398        let chunk = make_chunk(
399            &[
400                vec![Value::Int64(1), Value::Int64(1), Value::Int64(1)],
401                vec![
402                    Value::String("a".into()),
403                    Value::String("b".into()),
404                    Value::String("a".into()),
405                ],
406            ],
407            &["id", "tag"],
408        );
409        let out = reducer(vec![0, 1]).execute(vec![chunk]).unwrap();
410        assert_eq!(out[0].size, 2, "(1,a) and (1,b) kept; (1,a) repeated dropped");
411        assert_eq!(out[0].get_value(0, 0), Some(Value::Int64(1)));
412        assert_eq!(out[0].get_value(1, 0), Some(Value::String("a".into())));
413        assert_eq!(out[0].get_value(1, 1), Some(Value::String("b".into())));
414    }
415
416    #[test]
417    fn test_multiplicity_reducer_nan_not_deduped() {
418        // IEEE equality: NaN != NaN, so two NaN keys are both kept. The old
419        // string-format key would have merged them (debug "NaN" == "NaN").
420        let chunk = make_chunk(&[vec![Value::Double(f64::NAN), Value::Double(f64::NAN)]], &["x"]);
421        let out = reducer(vec![0]).execute(vec![chunk]).unwrap();
422        assert_eq!(out[0].size, 2);
423    }
424}
akar_processor/physical/misc.rs

akar_processor/physical/
misc.rs
