akar_processor/processor/mapper/
mod.rs

1pub mod map_aggregate;
2pub mod map_ddl;
3pub mod map_join;
4pub mod map_projection;
5pub mod map_scan;
6pub mod map_update;
7
8use crate::physical::types::PhysicalOperatorExec;
9use crate::processor::QueryProcessor;
10use akar_common::error::ProcessorError;
11use akar_common::types::physical_type_from_logical;
12use akar_common::vector::DataChunk;
13use akar_function::registry::FunctionRegistry;
14use akar_planner::logical_operator::LogicalOperator;
15use akar_storage::table::TableCatalog;
16use arrow::array::ArrayRef;
17use std::sync::{Arc, Mutex};
18
19use super::{SchemaDdlFn, SequenceFn, StandaloneCallHandler, SubqueryFn};
20
21/// Shared state threaded through the mapper functions
22pub struct ExecutionContext<'p> {
23    pub processor: &'p QueryProcessor,
24    pub function_registry: Option<Arc<Mutex<FunctionRegistry>>>,
25    pub table_catalog: Option<Arc<TableCatalog>>,
26    pub vfs: Option<Arc<akar_common::file_system::VirtualFileSystemRegistry>>,
27    pub standalone_call_handler: Option<Arc<dyn StandaloneCallHandler>>,
28    pub sequence_fn: Option<SequenceFn>,
29    pub subquery_fn: Option<SubqueryFn>,
30    pub schema_ddl_fn: Option<SchemaDdlFn>,
31    /// MVCC snapshot timestamp. When `Some(ts)`, reads are isolated to data
32    /// committed at or before `ts`. `None` means read最新 (no isolation).
33    pub snapshot_ts: Option<u64>,
34    /// Commit history for MVCC visibility checks: `(txn_id, commit_ts)` pairs.
35    pub commit_history: Vec<(u64, u64)>,
36    /// Row-level write set for OCC conflict detection.
37    /// Populated by the mapper after each write operation (SET, DELETE, INSERT).
38    /// The connection layer reads this after execution and calls `record_write()`.
39    pub written_rows: Vec<(u64, u64)>,
40}
41
42impl<'p> ExecutionContext<'p> {
43    pub fn execute_children(&mut self, operators: &[LogicalOperator]) -> Result<Vec<DataChunk>, ProcessorError> {
44        self.processor.execute_internal(operators)
45    }
46
47    /// Resolve table data and column definitions for a scan node.
48    /// When `snapshot_ts` is set on the context, uses MVCC-aware scan.
49    pub fn resolve_scan_data<'b>(
50        &self,
51        table_name: &str,
52        predicate: Option<(usize, &'b str, &'b akar_common::types::Value)>,
53    ) -> (
54        Option<Vec<Vec<akar_common::types::Value>>>,
55        Vec<akar_storage::table::ColumnDefinition>,
56        u64,
57    ) {
58        if let Some(ref tc) = self.table_catalog {
59            // Try node table first
60            if let Some(node_table) = tc.get_node_table_by_name(table_name) {
61                let num_rows = node_table.num_rows;
62                if num_rows > 0 {
63                    let data = if self.snapshot_ts.is_some() {
64                        // MVCC-aware scan: filter by snapshot visibility
65                        node_table.to_column_major_data_with_snapshot_and_predicate(
66                            predicate,
67                            self.snapshot_ts,
68                            &self.commit_history,
69                        )
70                    } else {
71                        node_table.to_column_major_data_with_predicate(predicate)
72                    };
73                    return (Some(data), node_table.columns.clone(), num_rows);
74                }
75            }
76            // Try rel table
77            if let Some(rel_table) = tc.get_rel_table_by_name(table_name) {
78                let num_rows = rel_table.num_rows;
79                if num_rows > 0 {
80                    return (
81                        Some(rel_table.to_column_major_data()),
82                        rel_table.columns.clone(),
83                        num_rows,
84                    );
85                }
86            }
87        }
88        (None, Vec::new(), 0)
89    }
90
91    /// Resolve scan data directly into Arrow arrays, bypassing the
92    /// `Vec<Vec<Value>>` intermediate materialization.
93    ///
94    /// Reads from NodeTable's NodeGroup column chunks, converts each
95    /// ColumnChunk to an Arrow array, then concatenates per-group arrays
96    /// into one array per column.
97    ///
98    /// When `snapshot_ts` is set, falls back to the Vec<Vec<Value>> path
99    /// since Arrow arrays don't support MVCC version chain traversal.
100    pub fn resolve_scan_arrow_data(
101        &self,
102        table_name: &str,
103    ) -> (Option<Vec<ArrayRef>>, Vec<akar_storage::table::ColumnDefinition>, u64) {
104        // When MVCC snapshot is active, skip the Arrow fast path — Arrow
105        // arrays don't support version chain traversal. The caller will
106        // fall back to the Vec<Vec<Value>> path which uses MVCC-aware reads.
107        if self.snapshot_ts.is_some() {
108            return (None, Vec::new(), 0);
109        }
110
111        if let Some(ref tc) = self.table_catalog {
112            if let Some(node_table) = tc.get_node_table_by_name(table_name) {
113                let num_rows = node_table.num_rows;
114                if num_rows > 0 {
115                    let mut column_arrays: Vec<Vec<ArrayRef>> = vec![Vec::new(); node_table.columns.len()];
116                    for ng in &node_table.node_groups {
117                        for (col_idx, col_chunk) in ng.columns.iter().enumerate() {
118                            let phys_type = physical_type_from_logical(node_table.columns[col_idx].logical_type);
119                            let arr = col_chunk.to_arrow_array(phys_type);
120                            column_arrays[col_idx].push(arr);
121                        }
122                    }
123                    // Concatenate per-group arrays into one array per column
124                    let concat_arrays: Vec<ArrayRef> = column_arrays
125                        .into_iter()
126                        .map(|group_arrays| {
127                            if group_arrays.len() == 1 {
128                                group_arrays.into_iter().next().unwrap()
129                            } else {
130                                let refs: Vec<&dyn arrow::array::Array> =
131                                    group_arrays.iter().map(|a| a.as_ref()).collect();
132                                arrow::compute::concat(&refs)
133                                    .unwrap_or_else(|_| group_arrays.into_iter().next().unwrap())
134                            }
135                        })
136                        .collect();
137                    return (Some(concat_arrays), node_table.columns.clone(), num_rows);
138                }
139            }
140        }
141        (None, Vec::new(), 0)
142    }
143}
144
145pub struct PlanMapper;
146
147impl PlanMapper {
148    pub fn map_and_execute(
149        op: &LogicalOperator,
150        next_op: Option<&LogicalOperator>,
151        current_input: Vec<DataChunk>,
152        ctx: &mut ExecutionContext,
153    ) -> Result<Vec<DataChunk>, ProcessorError> {
154        match op {
155            // Scans
156            LogicalOperator::ScanNode(s) => map_scan::map_and_execute_scan_node(s, next_op, current_input, ctx),
157            LogicalOperator::ScanRel(_)
158            | LogicalOperator::VectorSimilarityScan(_)
159            | LogicalOperator::ArtIndexRangeScan(_)
160            | LogicalOperator::IndexLookup(_)
161            | LogicalOperator::ExpressionsScan(_)
162            | LogicalOperator::PathPropertyProbe(_) => map_scan::map_and_execute_scan(op, current_input, ctx),
163
164            // Joins
165            LogicalOperator::HashJoin(_)
166            | LogicalOperator::SemiJoin(_)
167            | LogicalOperator::AntiJoin(_)
168            | LogicalOperator::Intersect(_)
169            | LogicalOperator::CrossProduct(_)
170            | LogicalOperator::OptionalMatch(_)
171            | LogicalOperator::RecursiveExtend(_) => map_join::map_and_execute_join(op, current_input, ctx),
172
173            // Aggregates
174            LogicalOperator::Aggregate(_) | LogicalOperator::CountRelTable(_) => {
175                map_aggregate::map_and_execute_aggregate(op, current_input, ctx)
176            }
177
178            // Updates
179            LogicalOperator::Set(_)
180            | LogicalOperator::Delete(_)
181            | LogicalOperator::CreateNode(_)
182            | LogicalOperator::CreateRel(_)
183            | LogicalOperator::Merge(_)
184            | LogicalOperator::Extend(_)
185            | LogicalOperator::BatchInsert(_)
186            | LogicalOperator::Insert(_)
187            | LogicalOperator::CopyFrom(_) => map_update::map_and_execute_update(op, current_input, ctx),
188
189            // Union
190            LogicalOperator::Union(u) => {
191                use crate::processor::union_helpers::{flatten_union_child, merge_union_chunks};
192                let left_ops = flatten_union_child(&u.left);
193                let right_ops = flatten_union_child(&u.right);
194                let left = ctx.execute_children(&left_ops)?;
195                let right = ctx.execute_children(&right_ops)?;
196                merge_union_chunks(left, right, u.all)
197            }
198
199            // Projections & Filters
200            LogicalOperator::Projection(_)
201            | LogicalOperator::Filter(_)
202            | LogicalOperator::TopK(_)
203            | LogicalOperator::OrderBy(_)
204            | LogicalOperator::Limit(_)
205            | LogicalOperator::Flatten(_)
206            | LogicalOperator::Unwind(_)
207            | LogicalOperator::Partitioner(_) => map_projection::map_and_execute_projection(op, current_input, ctx),
208
209            // DDL & Others
210            LogicalOperator::CreateNodeTable(_)
211            | LogicalOperator::CreateRelTable(_)
212            | LogicalOperator::DropTable(_)
213            | LogicalOperator::AlterTable(_)
214            | LogicalOperator::CreateIndex(_)
215            | LogicalOperator::DropIndex(_)
216            | LogicalOperator::CreateVectorIndex(_)
217            | LogicalOperator::CreateSequence(_)
218            | LogicalOperator::DropSequence(_)
219            | LogicalOperator::CreateDml(_)
220            | LogicalOperator::ExportDatabase(_)
221            | LogicalOperator::ImportDatabase(_)
222            | LogicalOperator::CreateFtsIndex(_)
223            | LogicalOperator::FtsScan(_)
224            | LogicalOperator::EmptyResult(_)
225            | LogicalOperator::MultiplicityReducer(_)
226            | LogicalOperator::Skip(_)
227            | LogicalOperator::ExtensionClause(_)
228            | LogicalOperator::StandaloneCall(_)
229            | LogicalOperator::TableFunctionCall(_)
230            | LogicalOperator::Foreach(_)
231            | LogicalOperator::Explain(_) => map_ddl::map_and_execute_ddl(op, current_input, ctx),
232
233            LogicalOperator::Accumulate(_) => {
234                let result = crate::physical_operator::PhysicalAccumulate.execute(current_input)?;
235                Ok(result)
236            }
237        }
238    }
239}
akar_processor/processor/mapper/mod.rs

akar_processor/processor/mapper/
mod.rs
