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, Value};
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                    // MVCC-aware scan: filter by snapshot visibility
64                    let (mut data, ids) = if self.snapshot_ts.is_some() {
65                        node_table.to_column_major_data_with_snapshot_and_predicate_and_ids(
66                            predicate,
67                            self.snapshot_ts,
68                            &self.commit_history,
69                        )
70                    } else {
71                        node_table.to_column_major_data_with_predicate_and_ids(predicate)
72                    };
73                    // Append the internal node id column (`<var>._id` = row offset).
74                    // Extend/insert/join operators resolve nodes through this column;
75                    // it is the same id space used by rel COPY (PK -> offset).
76                    data.push(ids.into_iter().map(|id| Value::Int64(id as i64)).collect());
77                    let mut columns = node_table.columns.clone();
78                    columns.push(akar_storage::table::ColumnDefinition {
79                        name: "_id".to_string(),
80                        logical_type: akar_common::types::LogicalTypeID::Int64,
81                        is_primary_key: false,
82                        compression: akar_common::enums::CompressionType::Uncompressed,
83                    });
84                    return (Some(data), columns, num_rows);
85                }
86            }
87            // Try rel table
88            if let Some(rel_table) = tc.get_rel_table_by_name(table_name) {
89                let num_rows = rel_table.num_rows;
90                if num_rows > 0 {
91                    return (
92                        Some(rel_table.to_column_major_data()),
93                        rel_table.columns.clone(),
94                        num_rows,
95                    );
96                }
97            }
98        }
99        (None, Vec::new(), 0)
100    }
101
102    /// Resolve scan data directly into Arrow arrays, bypassing the
103    /// `Vec<Vec<Value>>` intermediate materialization.
104    ///
105    /// Reads from NodeTable's NodeGroup column chunks, converts each
106    /// ColumnChunk to an Arrow array, then concatenates per-group arrays
107    /// into one array per column.
108    ///
109    /// When `snapshot_ts` is set, falls back to the Vec<Vec<Value>> path
110    /// since Arrow arrays don't support MVCC version chain traversal.
111    pub fn resolve_scan_arrow_data(
112        &self,
113        table_name: &str,
114    ) -> (Option<Vec<ArrayRef>>, Vec<akar_storage::table::ColumnDefinition>, u64) {
115        // When MVCC snapshot is active, skip the Arrow fast path — Arrow
116        // arrays don't support version chain traversal. The caller will
117        // fall back to the Vec<Vec<Value>> path which uses MVCC-aware reads.
118        if self.snapshot_ts.is_some() {
119            return (None, Vec::new(), 0);
120        }
121
122        if let Some(ref tc) = self.table_catalog {
123            if let Some(node_table) = tc.get_node_table_by_name(table_name) {
124                let num_rows = node_table.num_rows;
125                if num_rows > 0 {
126                    let mut column_arrays: Vec<Vec<ArrayRef>> = vec![Vec::new(); node_table.columns.len()];
127                    for ng in &node_table.node_groups {
128                        for (col_idx, col_chunk) in ng.columns.iter().enumerate() {
129                            let phys_type = physical_type_from_logical(node_table.columns[col_idx].logical_type);
130                            let arr = col_chunk.to_arrow_array(phys_type);
131                            column_arrays[col_idx].push(arr);
132                        }
133                    }
134                    // Concatenate per-group arrays into one array per column
135                    let mut concat_arrays: Vec<ArrayRef> = column_arrays
136                        .into_iter()
137                        .map(|group_arrays| {
138                            if group_arrays.len() == 1 {
139                                group_arrays.into_iter().next().unwrap()
140                            } else {
141                                let refs: Vec<&dyn arrow::array::Array> =
142                                    group_arrays.iter().map(|a| a.as_ref()).collect();
143                                arrow::compute::concat(&refs)
144                                    .unwrap_or_else(|_| group_arrays.into_iter().next().unwrap())
145                            }
146                        })
147                        .collect();
148                    // Append the internal node id column (`<var>._id` = row offset).
149                    // Since Arrow arrays are concatenated in node-group order and no
150                    // zone-map/predicate filtering happens here, offsets are 0..num_rows.
151                    let id_array: ArrayRef = std::sync::Arc::new(
152                        arrow::array::Int64Array::from((0..num_rows as i64).collect::<Vec<i64>>()),
153                    );
154                    concat_arrays.push(id_array);
155                    let mut columns = node_table.columns.clone();
156                    columns.push(akar_storage::table::ColumnDefinition {
157                        name: "_id".to_string(),
158                        logical_type: akar_common::types::LogicalTypeID::Int64,
159                        is_primary_key: false,
160                        compression: akar_common::enums::CompressionType::Uncompressed,
161                    });
162                    return (Some(concat_arrays), columns, num_rows);
163                }
164            }
165        }
166        (None, Vec::new(), 0)
167    }
168}
169
170pub struct PlanMapper;
171
172impl PlanMapper {
173    pub fn map_and_execute(
174        op: &LogicalOperator,
175        next_op: Option<&LogicalOperator>,
176        current_input: Vec<DataChunk>,
177        ctx: &mut ExecutionContext,
178    ) -> Result<Vec<DataChunk>, ProcessorError> {
179        match op {
180            // Scans
181            LogicalOperator::ScanNode(s) => map_scan::map_and_execute_scan_node(s, next_op, current_input, ctx),
182            LogicalOperator::ScanRel(_)
183            | LogicalOperator::VectorSimilarityScan(_)
184            | LogicalOperator::ArtIndexRangeScan(_)
185            | LogicalOperator::IndexLookup(_)
186            | LogicalOperator::ExpressionsScan(_)
187            | LogicalOperator::PathPropertyProbe(_) => map_scan::map_and_execute_scan(op, current_input, ctx),
188
189            // Joins
190            LogicalOperator::HashJoin(_)
191            | LogicalOperator::SemiJoin(_)
192            | LogicalOperator::AntiJoin(_)
193            | LogicalOperator::Intersect(_)
194            | LogicalOperator::CrossProduct(_)
195            | LogicalOperator::OptionalMatch(_)
196            | LogicalOperator::RecursiveExtend(_) => map_join::map_and_execute_join(op, current_input, ctx),
197
198            // Aggregates
199            LogicalOperator::Aggregate(_) | LogicalOperator::CountRelTable(_) => {
200                map_aggregate::map_and_execute_aggregate(op, current_input, ctx)
201            }
202
203            // Updates
204            LogicalOperator::Set(_)
205            | LogicalOperator::Delete(_)
206            | LogicalOperator::CreateNode(_)
207            | LogicalOperator::CreateRel(_)
208            | LogicalOperator::Merge(_)
209            | LogicalOperator::Extend(_)
210            | LogicalOperator::BatchInsert(_)
211            | LogicalOperator::Insert(_)
212            | LogicalOperator::CopyFrom(_) => map_update::map_and_execute_update(op, current_input, ctx),
213
214            // Union
215            LogicalOperator::Union(u) => {
216                use crate::processor::union_helpers::{flatten_union_child, merge_union_chunks};
217                let left_ops = flatten_union_child(&u.left);
218                let right_ops = flatten_union_child(&u.right);
219                let left = ctx.execute_children(&left_ops)?;
220                let right = ctx.execute_children(&right_ops)?;
221                merge_union_chunks(left, right, u.all)
222            }
223
224            // Projections & Filters
225            LogicalOperator::Projection(_)
226            | LogicalOperator::Filter(_)
227            | LogicalOperator::TopK(_)
228            | LogicalOperator::OrderBy(_)
229            | LogicalOperator::Limit(_)
230            | LogicalOperator::Flatten(_)
231            | LogicalOperator::Unwind(_)
232            | LogicalOperator::Partitioner(_) => map_projection::map_and_execute_projection(op, current_input, ctx),
233
234            // DDL & Others
235            LogicalOperator::CreateNodeTable(_)
236            | LogicalOperator::CreateRelTable(_)
237            | LogicalOperator::DropTable(_)
238            | LogicalOperator::AlterTable(_)
239            | LogicalOperator::CreateIndex(_)
240            | LogicalOperator::DropIndex(_)
241            | LogicalOperator::CreateVectorIndex(_)
242            | LogicalOperator::CreateSequence(_)
243            | LogicalOperator::DropSequence(_)
244            | LogicalOperator::CreateDml(_)
245            | LogicalOperator::ExportDatabase(_)
246            | LogicalOperator::ImportDatabase(_)
247            | LogicalOperator::CreateFtsIndex(_)
248            | LogicalOperator::FtsScan(_)
249            | LogicalOperator::EmptyResult(_)
250            | LogicalOperator::MultiplicityReducer(_)
251            | LogicalOperator::Skip(_)
252            | LogicalOperator::ExtensionClause(_)
253            | LogicalOperator::StandaloneCall(_)
254            | LogicalOperator::TableFunctionCall(_)
255            | LogicalOperator::Foreach(_)
256            | LogicalOperator::Explain(_) => map_ddl::map_and_execute_ddl(op, current_input, ctx),
257
258            LogicalOperator::Accumulate(_) => {
259                let result = crate::physical_operator::PhysicalAccumulate.execute(current_input)?;
260                Ok(result)
261            }
262        }
263    }
264}
akar_processor/processor/mapper/mod.rs

akar_processor/processor/mapper/
mod.rs
