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

akar_processor/processor/mapper/
mod.rs
