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

akar_processor/processor/mapper/
mod.rs
