akar_processor/processor/
mod.rs

1//! Query processor — maps logical operators to physical operators and executes them.
2//!
3//! Pipeline execution model:
4//! 1. Scan operators produce raw DataChunks
5//! 2. Filter removes non-matching rows
6//! 3. Projection selects/transforms columns
7//! 4. Limit/OrderBy/Aggregate are applied last
8
9pub mod chunk_helpers;
10pub mod join_helpers;
11pub mod mapper;
12pub mod plan_serializer;
13pub mod projection_helper;
14#[cfg(test)]
15mod tests;
16#[cfg(test)]
17mod tests_only;
18pub mod union_helpers;
19
20pub use chunk_helpers::*;
21pub use join_helpers::*;
22pub use mapper::*;
23pub use plan_serializer::*;
24pub use projection_helper::*;
25pub use union_helpers::*;
26
27use crate::physical_operator::*;
28use akar_common::error::ProcessorError;
29use akar_common::types::{PhysicalTypeID, Value};
30use akar_common::vector::{DataChunk, ValueVector};
31use akar_function::registry::{FunctionRegistry, TableFunction};
32use akar_planner::logical_operator::LogicalOperator;
33use akar_storage::table::TableCatalog;
34use std::sync::{Arc, Mutex};
35
36pub type SequenceFn = Arc<dyn Fn(&str, bool) -> Result<Value, ProcessorError> + Send + Sync>;
37pub type SubqueryFn = Arc<dyn Fn(&akar_parser::ast::Query) -> Result<Vec<DataChunk>, ProcessorError> + Send + Sync>;
38
39/// DDL operations that require the schema-level Catalog (from Akar-catalog).
40/// These are dispatched via callback because the processor layer doesn't
41/// directly own the schema catalog.
42#[derive(Debug, Clone)]
43pub enum SchemaDdlOp {
44    CreateSequence {
45        name: String,
46        if_not_exists: bool,
47        start_value: i64,
48        increment: i64,
49        min_value: i64,
50        max_value: i64,
51        cycle: bool,
52    },
53    DropSequence {
54        name: String,
55        if_exists: bool,
56    },
57    ExportDatabase {
58        file_path: String,
59        file_type: String,
60        schema_only: bool,
61    },
62    ImportDatabase {
63        file_path: String,
64        query: String,
65        index_query: String,
66    },
67}
68pub type SchemaDdlFn = Arc<dyn Fn(SchemaDdlOp) -> Result<String, ProcessorError> + Send + Sync>;
69
70pub trait StandaloneCallHandler: Send + Sync {
71    fn execute_call(
72        &self,
73        name: &str,
74        args: &[akar_parser::ast::Expression],
75    ) -> Result<Vec<akar_common::vector::DataChunk>, ProcessorError>;
76}
77
78pub trait StandaloneCallFn: Send + Sync {
79    fn execute(
80        &self,
81        args: &[akar_parser::ast::Expression],
82    ) -> Result<Vec<Vec<akar_common::types::Value>>, ProcessorError>;
83    fn aliases(&self) -> Vec<&'static str>;
84}
85
86#[derive(Default)]
87pub struct StandaloneCallRegistry {
88    handlers: std::collections::HashMap<String, std::sync::Arc<dyn StandaloneCallFn>>,
89}
90
91impl StandaloneCallRegistry {
92    pub fn new() -> Self {
93        Self::default()
94    }
95
96    pub fn register(&mut self, handler: std::sync::Arc<dyn StandaloneCallFn>) {
97        for alias in handler.aliases() {
98            self.handlers.insert(alias.to_lowercase(), handler.clone());
99        }
100    }
101
102    pub fn get(&self, name: &str) -> Option<std::sync::Arc<dyn StandaloneCallFn>> {
103        self.handlers.get(&name.to_lowercase()).cloned()
104    }
105}
106
107/// The query processor executes a physical plan and produces result chunks.
108pub struct QueryProcessor {
109    function_registry: Option<Arc<Mutex<FunctionRegistry>>>,
110    table_catalog: Option<Arc<TableCatalog>>,
111    vfs: Option<Arc<akar_common::file_system::VirtualFileSystemRegistry>>,
112    standalone_call_handler: Option<Arc<dyn StandaloneCallHandler>>,
113    /// Callback for sequence operations (nextval/currval).
114    /// Takes (sequence_name, is_nextval) and returns the resulting value.
115    sequence_fn: Option<SequenceFn>,
116    /// Callback for executing subqueries.
117    subquery_fn: Option<SubqueryFn>,
118    /// Callback for schema-level DDL operations (CREATE/DROP SEQUENCE, EXPORT/IMPORT DATABASE).
119    schema_ddl_fn: Option<SchemaDdlFn>,
120    /// MVCC snapshot timestamp for read isolation.
121    snapshot_ts: Option<u64>,
122    /// Commit history for MVCC visibility checks.
123    commit_history: Vec<(u64, u64)>,
124    /// Row-level write set accumulated during execution.
125    /// Populated by the mapper after each write operation (SET, DELETE, INSERT).
126    /// Read by the connection layer after execution for OCC conflict detection.
127    written_rows: Mutex<Vec<(u64, u64)>>,
128}
129
130impl QueryProcessor {
131    pub fn new() -> Self {
132        Self {
133            function_registry: None,
134            table_catalog: None,
135            vfs: None,
136            standalone_call_handler: None,
137            sequence_fn: None,
138            subquery_fn: None,
139            schema_ddl_fn: None,
140            snapshot_ts: None,
141            commit_history: Vec::new(),
142            written_rows: Mutex::new(Vec::new()),
143        }
144    }
145
146    /// Create a processor with access to the function registry.
147    pub fn with_registry(registry: Arc<Mutex<FunctionRegistry>>) -> Self {
148        Self {
149            function_registry: Some(registry),
150            table_catalog: None,
151            vfs: None,
152            standalone_call_handler: None,
153            sequence_fn: None,
154            subquery_fn: None,
155            schema_ddl_fn: None,
156            snapshot_ts: None,
157            commit_history: Vec::new(),
158            written_rows: Mutex::new(Vec::new()),
159        }
160    }
161
162    /// Create a processor with function registry, table catalog access, and VFS.
163    pub fn with_catalog(
164        registry: Arc<Mutex<FunctionRegistry>>,
165        table_catalog: Arc<TableCatalog>,
166        vfs: Arc<akar_common::file_system::VirtualFileSystemRegistry>,
167    ) -> Self {
168        Self {
169            function_registry: Some(registry),
170            table_catalog: Some(table_catalog),
171            vfs: Some(vfs),
172            standalone_call_handler: None,
173            sequence_fn: None,
174            subquery_fn: None,
175            schema_ddl_fn: None,
176            snapshot_ts: None,
177            commit_history: Vec::new(),
178            written_rows: Mutex::new(Vec::new()),
179        }
180    }
181
182    /// Set the sequence operation callback (for nextval/currval).
183
184    pub fn with_standalone_call_handler(mut self, handler: Arc<dyn StandaloneCallHandler>) -> Self {
185        self.standalone_call_handler = Some(handler);
186        self
187    }
188
189    pub fn with_sequence_fn(mut self, f: SequenceFn) -> Self {
190        self.sequence_fn = Some(f);
191        self
192    }
193
194    /// Set the subquery operation callback.
195    pub fn with_subquery_fn(mut self, f: SubqueryFn) -> Self {
196        self.subquery_fn = Some(f);
197        self
198    }
199
200    /// Set the schema DDL callback (for CREATE/DROP SEQUENCE, EXPORT/IMPORT DATABASE).
201    pub fn with_schema_ddl_fn(mut self, f: SchemaDdlFn) -> Self {
202        self.schema_ddl_fn = Some(f);
203        self
204    }
205
206    /// Set MVCC snapshot parameters for read isolation.
207    pub fn with_snapshot(mut self, snapshot_ts: Option<u64>, commit_history: Vec<(u64, u64)>) -> Self {
208        self.snapshot_ts = snapshot_ts;
209        self.commit_history = commit_history;
210        self
211    }
212
213    /// Execute a sequence of logical operators by mapping them to physical operators.
214    pub fn execute(&self, operators: &[LogicalOperator]) -> Result<Vec<DataChunk>, ProcessorError> {
215        self.execute_internal(operators)
216    }
217
218    pub fn execute_internal(&self, operators: &[LogicalOperator]) -> Result<Vec<DataChunk>, ProcessorError> {
219        if operators.is_empty() {
220            return Ok(vec![DataChunk {
221                fields: vec![],
222                field_types: vec![],
223                size: 0,
224                field_names: vec![],
225                sel_vector: None,
226            }]);
227        }
228
229        let mut intermediate_result: Option<Vec<DataChunk>> = None;
230
231        for (i, op) in operators.iter().enumerate() {
232            let current = intermediate_result.take().unwrap_or_else(|| {
233                let mut dummy = DataChunk::new(vec![], vec![]);
234                dummy.size = 1;
235                vec![dummy]
236            });
237            let next_op = operators.get(i + 1);
238
239            let mut ctx = mapper::ExecutionContext {
240                processor: self,
241                function_registry: self.function_registry.clone(),
242                table_catalog: self.table_catalog.clone(),
243                vfs: self.vfs.clone(),
244                standalone_call_handler: self.standalone_call_handler.clone(),
245                sequence_fn: self.sequence_fn.clone(),
246                subquery_fn: self.subquery_fn.clone(),
247                schema_ddl_fn: self.schema_ddl_fn.clone(),
248                snapshot_ts: self.snapshot_ts,
249                commit_history: self.commit_history.clone(),
250                written_rows: Vec::new(),
251            };
252
253            let result = mapper::PlanMapper::map_and_execute(op, next_op, current, &mut ctx)?;
254
255            // Accumulate written rows from this operator into the processor's write set
256            if !ctx.written_rows.is_empty() {
257                if let Ok(mut writes) = self.written_rows.lock() {
258                    writes.append(&mut ctx.written_rows);
259                }
260            }
261
262            if let LogicalOperator::ScanRel(_) = op {
263                // Accumulate: extend rather than replace for ScanRel
264                match &mut intermediate_result {
265                    Some(existing) => existing.extend(result),
266                    None => intermediate_result = Some(result),
267                }
268            } else {
269                intermediate_result = Some(result);
270            }
271        }
272
273        Ok(intermediate_result.unwrap_or_default())
274    }
275
276    /// Take the accumulated write set from the processor.
277    /// Returns all (table_id, row_id) pairs written during the last execution.
278    pub fn take_written_rows(&self) -> Vec<(u64, u64)> {
279        self.written_rows
280            .lock()
281            .map(|mut w| std::mem::take(&mut *w))
282            .unwrap_or_default()
283    }
284
285    /// Execute a table function call by looking up the function in the registry
286    /// and dispatching to the appropriate handler.
287    fn execute_table_function(
288        &self,
289        tf: &akar_planner::logical_operator::LogicalTableFunctionCall,
290    ) -> Result<Vec<DataChunk>, ProcessorError> {
291        let func_name = &tf.function_name;
292        let args: Vec<Value> = Vec::new(); // args would be evaluated from expressions
293
294        // Look up the function in the registry
295        if let Some(ref registry) = self.function_registry {
296            let reg = registry.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
297            if let Some(tbl_fn) = reg.get_table(func_name) {
298                match tbl_fn {
299                    TableFunction::CustomTable { execute, .. } => {
300                        let mut chunk = DataChunk::new(Vec::new(), Vec::new());
301                        (execute)(&args, &mut chunk)?;
302                        Ok(vec![chunk])
303                    }
304                    TableFunction::ScanCsv { .. }
305                    | TableFunction::ScanParquet { .. }
306                    | TableFunction::ScanJson { .. }
307                    | TableFunction::ListTables
308                    | TableFunction::ShowColumns { .. }
309                    | TableFunction::CurrentSetting { .. } => Err(format!(
310                        "Table function '{}' cannot be executed dynamically (no callback)",
311                        func_name
312                    )
313                    .into()),
314                    TableFunction::Custom { name } if name == "vector_similarity_scan" => {
315                        // Evaluate args: [table_name, column_name, query_vector, top_k]
316                        // For CALL statement, args are parsed as expressions. We need to evaluate them.
317                        // For now, parse from the function args
318                        drop(reg);
319                        self.execute_vector_similarity_scan(tf)
320                    }
321                    TableFunction::Custom { name } => {
322                        Err(format!("Custom table function '{}' has no registered handler", name).into())
323                    }
324                }
325            } else {
326                Err(format!("Table function '{}' not found", func_name).into())
327            }
328        } else {
329            Err(format!(
330                "Cannot execute table function '{}': no function registry available",
331                func_name
332            )
333            .into())
334        }
335    }
336
337    /// Execute a `vector_similarity_scan` table function call.
338    ///
339    /// Expects CALL vector_similarity_scan(table_name, column_name, query_vector, top_k)
340    /// and dispatches to PhysicalVectorSimilarityScan with the processor's TableCatalog.
341    fn execute_vector_similarity_scan(
342        &self,
343        tf: &akar_planner::logical_operator::LogicalTableFunctionCall,
344    ) -> Result<Vec<DataChunk>, ProcessorError> {
345        // Evaluate arguments from expressions (they should be constants or simple vars)
346        if tf.args.len() < 4 {
347            return Err(
348                "vector_similarity_scan requires 4 arguments: table_name, column_name, query_vector, top_k".into(),
349            );
350        }
351
352        // For CALL statements, args arrive as Expression AST nodes.
353        // Evaluate them to Values. The simplest approach: evaluate constants inline.
354        fn eval_expr_to_value(expr: &akar_parser::ast::Expression) -> Option<Value> {
355            match expr {
356                akar_parser::ast::Expression::Constant(c) => match c {
357                    akar_parser::ast::Constant::String(s) => Some(Value::String(s.clone())),
358                    akar_parser::ast::Constant::Integer(i) => Some(Value::Int64(*i)),
359                    akar_parser::ast::Constant::Float(f) => Some(Value::Double(*f)),
360                    akar_parser::ast::Constant::Bool(b) => Some(Value::Bool(*b)),
361                    akar_parser::ast::Constant::Null => Some(Value::Null),
362                },
363                akar_parser::ast::Expression::List(items) => {
364                    let vals: Vec<Value> = items.iter().filter_map(eval_expr_to_value).collect();
365                    Some(Value::List(vals))
366                }
367                _ => None, // Non-constant expression — skip
368            }
369        }
370
371        let table_name = match eval_expr_to_value(&tf.args[0]) {
372            Some(Value::String(s)) => s,
373            _ => return Err("First argument to vector_similarity_scan must be a table name string".into()),
374        };
375
376        let _column_name = match eval_expr_to_value(&tf.args[1]) {
377            Some(Value::String(s)) => s,
378            _ => return Err("Second argument to vector_similarity_scan must be a column name string".into()),
379        };
380
381        let query_vector = match eval_expr_to_value(&tf.args[2]) {
382            Some(Value::List(items)) => {
383                let mut vec = Vec::with_capacity(items.len());
384                for item in &items {
385                    match item {
386                        Value::Double(d) => vec.push(*d),
387                        Value::Int64(i) => vec.push(*i as f64),
388                        Value::Int32(i) => vec.push(*i as f64),
389                        Value::Float(f) => vec.push(*f as f64),
390                        _ => return Err("query_vector must be a list of numbers".into()),
391                    }
392                }
393                vec
394            }
395            _ => return Err("Third argument to vector_similarity_scan must be a list of numbers".into()),
396        };
397
398        let top_k = match eval_expr_to_value(&tf.args[3]) {
399            Some(Value::Int64(k)) if k > 0 => k as u64,
400            _ => return Err("Fourth argument to vector_similarity_scan must be a positive integer".into()),
401        };
402
403        // Find the vector index on this table
404        let tc = self
405            .table_catalog
406            .clone()
407            .ok_or_else(|| "No table catalog available for vector_similarity_scan".to_string())?;
408
409        // Look for a vector index matching this table name
410        let index_name = {
411            let mut found = None;
412            for entry in tc.all_vector_indexes() {
413                if entry.table_name == table_name {
414                    found = Some(entry.name.clone());
415                    break;
416                }
417            }
418            found.ok_or_else(|| format!("No vector index found on table '{}'", table_name))?
419        };
420
421        // Dispatch to PhysicalVectorSimilarityScan
422        let scan = PhysicalVectorSimilarityScan {
423            index_name,
424            index_id: 0,
425            query_vector,
426            top_k,
427            table_name,
428            table_catalog: Some(tc),
429        };
430        scan.execute(vec![])
431    }
432
433    /// Execute a single expression against a DataChunk and return a ValueVector of results.
434    pub fn evaluate_expression(
435        _expr: &akar_parser::ast::Expression,
436        _chunk: &DataChunk,
437    ) -> Result<ValueVector, ProcessorError> {
438        // Placeholder: return a dummy Int64 vector
439        let size = _chunk.size;
440        let mut v = ValueVector::new(PhysicalTypeID::Int64, size);
441        for i in 0..size {
442            v.set_i64(i, 0);
443        }
444        v.resize(size);
445        Ok(v)
446    }
447}
448
449impl Default for QueryProcessor {
450    fn default() -> Self {
451        Self::new()
452    }
453}
akar_processor/processor/mod.rs

akar_processor/processor/
mod.rs
