Skip to main content

akar_processor/
expression_evaluator.rs

1//! Expression evaluator — recursively evaluates an expression tree against a DataChunk.
2//!
3//! This replaces the ad-hoc `PhysicalFilter::evaluate_expression` with a proper
4//! expression evaluator that dispatches function calls through the scalar function
5//! registry (Akar-function::evaluate_scalar), supporting:
6//! - Variable:        reads values from a DataChunk field by name
7//! - Constant:        returns a literal value
8//! - BinaryOp:        dispatches to arithmetic/comparison/boolean scalar functions
9//! - UnaryOp:         dispatches to NOT/negate scalar functions
10//! - FunctionCall:    resolves the function name in the registry and evaluates
11//! - PropertyAccess:  reads a property from a struct/object expression
12//! - List/Map:        evaluated via list_creation/map_creation scalar functions
13
14use akar_common::arrow_vector::{ArrowVector, VectorAccess};
15use akar_common::error::ProcessorError;
16use akar_common::types::{PhysicalTypeID, Value};
17use akar_common::vector::{DataChunk, ValueVector};
18use akar_function::registry::{FunctionRegistry, ScalarFunction};
19use akar_function::scalar::evaluate_scalar;
20use akar_parser::ast::{BinaryOp, Constant, Expression, Query, UnaryOp};
21use arrow::array::{Array, ArrayRef};
22use std::sync::{Arc, Mutex};
23
24pub type SubqueryFn = Arc<dyn Fn(&Query) -> Result<Vec<DataChunk>, ProcessorError> + Send + Sync>;
25pub type SequenceFn = Arc<dyn Fn(&str, bool) -> Result<Value, ProcessorError> + Send + Sync>;
26
27/// Extract a named property from a runtime map/struct value. Returns
28/// `Value::Null` when `val` is not a map/struct or the key is absent.
29///
30/// Used to resolve property access on a variable bound to a map/struct column,
31/// e.g. `UNWIND $rows AS row CREATE (:Memory {id: row.id})` (P53.26).
32pub fn map_property_value(val: &Value, prop: &str) -> Value {
33    match val {
34        Value::Map(entries) => entries
35            .iter()
36            .find(|(k, _)| matches!(k, Value::String(s) if s == prop))
37            .map(|(_, v)| v.clone())
38            .unwrap_or(Value::Null),
39        Value::Struct(fields) => fields
40            .iter()
41            .find(|(name, _)| name == prop)
42            .map(|(_, v)| v.clone())
43            .unwrap_or(Value::Null),
44        _ => Value::Null,
45    }
46}
47
48/// Evaluates expressions against DataChunks using the function registry.
49impl std::fmt::Debug for ExpressionEvaluator {
50    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
51        f.debug_struct("ExpressionEvaluator").finish()
52    }
53}
54
55pub struct ExpressionEvaluator {
56    registry: Arc<Mutex<FunctionRegistry>>,
57    /// Optional callback to execute subqueries at evaluation time.
58    /// Takes a parsed Query and returns DataChunks.
59    pub subquery_fn: Option<SubqueryFn>,
60    /// Optional callback for sequence operations (nextval/currval).
61    /// Takes (sequence_name, is_nextval) and returns the resulting value.
62    pub sequence_fn: Option<SequenceFn>,
63}
64
65impl ExpressionEvaluator {
66    pub fn new(registry: Arc<Mutex<FunctionRegistry>>) -> Self {
67        Self {
68            registry,
69            subquery_fn: None,
70            sequence_fn: None,
71        }
72    }
73
74    /// Set the subquery execution callback.
75    pub fn with_subquery_fn(mut self, f: SubqueryFn) -> Self {
76        self.subquery_fn = Some(f);
77        self
78    }
79
80    /// Set the sequence operation callback (for nextval/currval).
81    pub fn with_sequence_fn(mut self, f: SequenceFn) -> Self {
82        self.sequence_fn = Some(f);
83        self
84    }
85
86    /// Evaluate a subquery by calling the stored callback.
87    fn evaluate_subquery(&self, query: &Query) -> Result<Vec<DataChunk>, ProcessorError> {
88        if let Some(ref f) = self.subquery_fn {
89            f(query)
90        } else {
91            Err("No subquery executor configured".into())
92        }
93    }
94
95    /// Evaluate an expression for every row in the chunk, returning an ArrowVector
96    /// directly. Delegates to evaluate_arrow which uses Arrow compute kernels for
97    /// the hot path (comparisons, arithmetic, boolean ops).
98    pub fn evaluate_to_arrow(&self, expr: &Expression, chunk: &DataChunk) -> Result<ArrowVector, ProcessorError> {
99        self.evaluate_arrow(expr, chunk)
100    }
101
102    /// Evaluate an expression for every row in the chunk, returning a ValueVector.
103    pub fn evaluate(&self, expr: &Expression, chunk: &DataChunk) -> Result<ValueVector, ProcessorError> {
104        match expr {
105            Expression::Constant(c) => self.evaluate_constant(c, chunk.size),
106            Expression::Variable(name) => self.evaluate_variable(name, chunk),
107            Expression::PropertyAccess(obj, prop) => self.evaluate_property_access(obj, prop, chunk),
108            Expression::FunctionCall(name, args) => {
109                let refs: Vec<&Expression> = args.iter().collect();
110                self.evaluate_function_call(name, &refs, chunk)
111            }
112            Expression::BinaryOp(op, left, right) => self.evaluate_binary_op(op, left, right, chunk),
113            Expression::UnaryOp(op, inner) => self.evaluate_unary_op(op, inner, chunk),
114            Expression::List(items) => self.evaluate_list_literal(items, chunk),
115            Expression::Map(items) => self.evaluate_map_literal(items, chunk),
116            Expression::Parameter(_) => {
117                // Parameters should be substituted by the binder/prepared statement layer.
118                // If they reach the evaluator, return a null Int64 vector matching chunk size.
119                let mut v = ValueVector::new(akar_common::types::PhysicalTypeID::Int64, chunk.size);
120                v.resize(chunk.size);
121                for i in 0..chunk.size {
122                    v.set_null(i, true);
123                }
124                Ok(v)
125            }
126            Expression::ExistsSubquery(query) => {
127                // Evaluate EXISTS subquery: execute the inner query against
128                // the database. If it returns at least one row → true, else false.
129                // For uncorrelated subqueries, execute once and fill all rows.
130                let result = self.evaluate_subquery(query)?;
131                let exists = !result.is_empty() && result.iter().any(|c| c.size > 0);
132                let mut v = ValueVector::new(akar_common::types::PhysicalTypeID::Bool, chunk.size);
133                v.resize(chunk.size);
134                for i in 0..chunk.size {
135                    store_value_in_vector_simple(&mut v, i, &Value::Bool(exists))?;
136                }
137                Ok(v)
138            }
139            Expression::Case(case_expr) => self.evaluate_case(case_expr, chunk),
140            Expression::Star => {
141                Err("STAR expression should be expanded by the binder before reaching the evaluator".into())
142            }
143            Expression::ListPredicate {
144                quantifier,
145                list,
146                var_name,
147                predicate,
148            } => self.evaluate_list_predicate(quantifier, list, var_name, predicate, chunk),
149            Expression::Lambda { .. } => {
150                Err("Lambda expression should only appear as argument to list_transform/filter/reduce".into())
151            }
152        }
153    }
154
155    // ==================== Arrow-native evaluation ====================
156
157    /// Evaluate an expression returning an ArrowVector directly.
158    /// For operations supported by Arrow compute kernels (comparisons,
159    /// arithmetic, boolean), this is vectorized and avoids Value enum boxing.
160    pub fn evaluate_arrow(&self, expr: &Expression, chunk: &DataChunk) -> Result<ArrowVector, ProcessorError> {
161        match expr {
162            Expression::Constant(c) => self.evaluate_arrow_constant(c, chunk.size),
163            Expression::Variable(name) => self.evaluate_arrow_variable(name, chunk),
164            Expression::PropertyAccess(obj, prop) => self.evaluate_arrow_property_access(obj, prop, chunk),
165            Expression::FunctionCall(name, args) => {
166                let refs: Vec<&Expression> = args.iter().collect();
167                self.evaluate_arrow_function_call(name, &refs, chunk)
168            }
169            Expression::BinaryOp(op, left, right) => self.evaluate_arrow_binary_op(op, left, right, chunk),
170            Expression::UnaryOp(op, inner) => self.evaluate_arrow_unary_op(op, inner, chunk),
171            Expression::List(items) => self.evaluate_arrow_list_literal(items, chunk),
172            Expression::Map(items) => self.evaluate_arrow_map_literal(items, chunk),
173            // Fallback for complex types: use evaluate + from_legacy
174            _ => {
175                let legacy = self.evaluate(expr, chunk)?;
176                Ok(ArrowVector::from_legacy(&legacy))
177            }
178        }
179    }
180
181    /// Evaluate a list literal directly to an Arrow `ListArray`, avoiding the
182    /// `ValueVector` (which has no side-storage for variable-length list
183    /// payloads). Each row's element expressions are evaluated per-row.
184    fn evaluate_arrow_list_literal(
185        &self,
186        items: &[Expression],
187        chunk: &DataChunk,
188    ) -> Result<ArrowVector, ProcessorError> {
189        let num_rows = chunk.size;
190        let mut row_results: Vec<Value> = Vec::with_capacity(num_rows);
191        for row in 0..num_rows {
192            let mut list_values = Vec::with_capacity(items.len());
193            for item in items {
194                let item_vec = self.evaluate(item, chunk)?;
195                let val = if row < item_vec.size() {
196                    item_vec.get_value(row).unwrap_or(Value::Null)
197                } else {
198                    Value::Null
199                };
200                list_values.push(val);
201            }
202            row_results.push(Value::List(list_values));
203        }
204        build_arrow_from_values(&row_results, PhysicalTypeID::List, num_rows)
205    }
206
207    /// Evaluate a map literal directly to an Arrow `StructArray` (a `{..}`
208    /// literal), again bypassing the `ValueVector`.
209    fn evaluate_arrow_map_literal(
210        &self,
211        items: &[(String, Expression)],
212        chunk: &DataChunk,
213    ) -> Result<ArrowVector, ProcessorError> {
214        let num_rows = chunk.size;
215        let mut row_results: Vec<Value> = Vec::with_capacity(num_rows);
216        for row in 0..num_rows {
217            let mut entries = Vec::with_capacity(items.len());
218            for (key, item) in items {
219                let item_vec = self.evaluate(item, chunk)?;
220                let val = if row < item_vec.size() {
221                    item_vec.get_value(row).unwrap_or(Value::Null)
222                } else {
223                    Value::Null
224                };
225                entries.push((key.clone(), val));
226            }
227            row_results.push(Value::Struct(entries));
228        }
229        build_arrow_from_values(&row_results, PhysicalTypeID::Struct, num_rows)
230    }
231
232    /// Evaluate a constant directly as ArrowVector using typed Arrow builders.
233    fn evaluate_arrow_constant(&self, c: &Constant, size: usize) -> Result<ArrowVector, ProcessorError> {
234        match c {
235            Constant::Null => {
236                let mut builder = arrow::array::Int64Builder::with_capacity(size);
237                builder.append_nulls(size);
238                Ok(ArrowVector::new(Arc::new(builder.finish()), PhysicalTypeID::Int64))
239            }
240            Constant::Bool(b) => {
241                let mut builder = arrow::array::BooleanBuilder::with_capacity(size);
242                for _ in 0..size {
243                    builder.append_value(*b);
244                }
245                Ok(ArrowVector::new(Arc::new(builder.finish()), PhysicalTypeID::Bool))
246            }
247            Constant::Integer(i) => {
248                let mut builder = arrow::array::Int64Builder::with_capacity(size);
249                let v = *i;
250                for _ in 0..size {
251                    builder.append_value(v);
252                }
253                Ok(ArrowVector::new(Arc::new(builder.finish()), PhysicalTypeID::Int64))
254            }
255            Constant::Float(f) => {
256                let mut builder = arrow::array::Float64Builder::with_capacity(size);
257                let v = *f;
258                for _ in 0..size {
259                    builder.append_value(v);
260                }
261                Ok(ArrowVector::new(Arc::new(builder.finish()), PhysicalTypeID::Double))
262            }
263            Constant::String(s) => {
264                let mut builder = arrow::array::StringBuilder::with_capacity(size, size * s.len().max(1));
265                for _ in 0..size {
266                    builder.append_value(s);
267                }
268                Ok(ArrowVector::new(Arc::new(builder.finish()), PhysicalTypeID::String))
269            }
270        }
271    }
272
273    /// Evaluate a variable expression, converting the ValueVector to ArrowVector.
274    fn evaluate_arrow_variable(&self, name: &str, chunk: &DataChunk) -> Result<ArrowVector, ProcessorError> {
275        let idx = if let Ok(idx) = name.parse::<usize>() {
276            idx
277        } else if !chunk.field_names.is_empty() {
278            if let Some(idx) = chunk.field_names.iter().position(|n| n == name) {
279                idx
280            } else {
281                return Err(format!("Variable '{}' not found in field_names", name).into());
282            }
283        } else {
284            return Err(format!("Variable '{}' not found (chunk has no field_names)", name).into());
285        };
286
287        let field = chunk
288            .fields
289            .get(idx)
290            .ok_or_else(|| format!("Variable '{}' (index {}) not found in chunk fields", name, idx))?;
291        Ok(ArrowVector::new(field.clone(), chunk.field_types[idx]))
292    }
293
294    /// Evaluate a property access expression, returning ArrowVector.
295    fn evaluate_arrow_property_access(
296        &self,
297        obj: &Expression,
298        prop: &str,
299        chunk: &DataChunk,
300    ) -> Result<ArrowVector, ProcessorError> {
301        let qualified_prop = if let Expression::Variable(var_name) = obj {
302            format!("{}.{}", var_name, prop)
303        } else {
304            prop.to_string()
305        };
306
307        if !chunk.field_names.is_empty()
308            && let Some(idx) = chunk.field_names.iter().position(|n| n == &qualified_prop)
309        {
310            let field = chunk
311                .fields
312                .get(idx)
313                .ok_or_else(|| format!("Property '{}' not found in chunk", prop))?;
314            return Ok(ArrowVector::new(field.clone(), chunk.field_types[idx]));
315        }
316        // P53.26: `row.id` where `row` is a map/struct column — extract the key
317        // from each row's value instead of returning the whole object. This must
318        // run BEFORE the bare-name match below, otherwise `row.content` in a
319        // snapshot that also carries a plain `content` table column resolves to
320        // the table column instead of the UNWIND variable.
321        if let Expression::Variable(var_name) = obj
322            && let Some(idx) = chunk.field_names.iter().position(|n| n == var_name)
323        {
324            let extracted: Vec<Value> = (0..chunk.size)
325                .map(|i| map_property_value(&chunk.get_value(idx, i).unwrap_or(Value::Null), prop))
326                .collect();
327            if extracted.iter().any(|v| !matches!(v, Value::Null)) {
328                let result_type = extracted
329                    .iter()
330                    .find(|v| !matches!(v, Value::Null))
331                    .map(|v| v.physical_type())
332                    .unwrap_or(PhysicalTypeID::Int64);
333                return build_arrow_from_values(&extracted, result_type, chunk.size);
334            }
335        }
336        if !chunk.field_names.is_empty()
337            && let Some(idx) = chunk.field_names.iter().position(|n| n == prop)
338        {
339            let field = chunk
340                .fields
341                .get(idx)
342                .ok_or_else(|| format!("Property '{}' not found in chunk", prop))?;
343            return Ok(ArrowVector::new(field.clone(), chunk.field_types[idx]));
344        }
345        let legacy = self.evaluate(obj, chunk)?;
346        Ok(ArrowVector::from_legacy(&legacy))
347    }
348
349    /// Evaluate a binary operation using Arrow compute kernels when possible.
350    fn evaluate_arrow_binary_op(
351        &self,
352        op: &BinaryOp,
353        left: &Expression,
354        right: &Expression,
355        chunk: &DataChunk,
356    ) -> Result<ArrowVector, ProcessorError> {
357        match op {
358            BinaryOp::In | BinaryOp::NotIn => {
359                let legacy = self.evaluate_in_op(op, left, right, chunk)?;
360                return Ok(ArrowVector::from_legacy(&legacy));
361            }
362            BinaryOp::Concat | BinaryOp::StartsWith | BinaryOp::EndsWith | BinaryOp::Contains | BinaryOp::Like => {
363                let func_name = match op {
364                    BinaryOp::Concat => "concat",
365                    BinaryOp::StartsWith => "starts_with",
366                    BinaryOp::EndsWith => "ends_with",
367                    BinaryOp::Contains => "contains",
368                    BinaryOp::Like => "like",
369                    _ => unreachable!(),
370                };
371                return self.evaluate_arrow_function_call(func_name, &[left, right], chunk);
372            }
373            _ => {}
374        }
375
376        let kernel_name = match op {
377            BinaryOp::Add => "add",
378            BinaryOp::Subtract => "sub",
379            BinaryOp::Multiply => "mul",
380            BinaryOp::Divide => "div",
381            BinaryOp::Modulo => "mod",
382            BinaryOp::Equal => "eq",
383            BinaryOp::NotEqual => "neq",
384            BinaryOp::LessThan => "lt",
385            BinaryOp::LessThanOrEqual => "lt_eq",
386            BinaryOp::GreaterThan => "gt",
387            BinaryOp::GreaterThanOrEqual => "gt_eq",
388            BinaryOp::And => "and",
389            BinaryOp::Or => "or",
390            BinaryOp::Xor => "xor",
391            _ => return Err(format!("Unsupported binary op: {:?}", op).into()),
392        };
393
394        let left_arrow = self.evaluate_arrow(left, chunk)?;
395        let right_arrow = self.evaluate_arrow(right, chunk)?;
396
397        match self.apply_arrow_kernel(kernel_name, &left_arrow, &right_arrow) {
398            Ok(result) => Ok(result),
399            Err(_) => {
400                let legacy = self.evaluate_binary_op(op, left, right, chunk)?;
401                Ok(ArrowVector::from_legacy(&legacy))
402            }
403        }
404    }
405
406    /// Evaluate a unary operation using Arrow compute kernels when possible.
407    fn evaluate_arrow_unary_op(
408        &self,
409        op: &UnaryOp,
410        inner: &Expression,
411        chunk: &DataChunk,
412    ) -> Result<ArrowVector, ProcessorError> {
413        match op {
414            UnaryOp::Not => {
415                let inner_arrow = self.evaluate_arrow(inner, chunk)?;
416                self.apply_arrow_unary_kernel("not", &inner_arrow).or_else(|_| {
417                    let legacy = self.evaluate_unary_op(&UnaryOp::Not, inner, chunk)?;
418                    Ok(ArrowVector::from_legacy(&legacy))
419                })
420            }
421            UnaryOp::Negate => {
422                let inner_arrow = self.evaluate_arrow(inner, chunk)?;
423                self.apply_arrow_unary_kernel("negate", &inner_arrow).or_else(|_| {
424                    let legacy = self.evaluate_unary_op(&UnaryOp::Negate, inner, chunk)?;
425                    Ok(ArrowVector::from_legacy(&legacy))
426                })
427            }
428            UnaryOp::IsNull => {
429                // P53.25: `{rel_var} IS NULL` where the variable expands to
430                // multiple prefixed columns (e.g. the edge columns emitted by
431                // OptionalExtend) is true iff ALL of its columns are NULL.
432                // A bare variable has no exact column here, so evaluate the
433                // null mask directly over the prefixed columns.
434                if let Expression::Variable(name) = inner
435                    && !chunk.field_names.is_empty()
436                    && !chunk.field_names.iter().any(|n| n == name)
437                {
438                    let prefix = format!("{}.", name);
439                    let cols: Vec<usize> = chunk
440                        .field_names
441                        .iter()
442                        .enumerate()
443                        .filter(|(_, n)| n.starts_with(prefix.as_str()))
444                        .map(|(i, _)| i)
445                        .collect();
446                    if !cols.is_empty() {
447                        let size = chunk.size;
448                        let mut acc: Option<arrow::array::BooleanArray> = None;
449                        for &col in &cols {
450                            let field = chunk
451                                .fields
452                                .get(col)
453                                .ok_or_else(|| format!("Variable '{}' column {} not found", name, col))?;
454                            let nulls: Vec<bool> = (0..size).map(|i| field.is_null(i)).collect();
455                            let m = arrow::array::BooleanArray::from(nulls);
456                            acc = Some(match acc {
457                                Some(a) => arrow::compute::kernels::boolean::and(&a, &m)
458                                    .map_err(|e| format!("Arrow is_null over variable failed: {e}"))?,
459                                None => m,
460                            });
461                        }
462                        return Ok(ArrowVector::new(Arc::new(acc.unwrap()), PhysicalTypeID::Bool));
463                    }
464                }
465                let inner_arrow = self.evaluate_arrow(inner, chunk)?;
466                self.apply_arrow_unary_kernel("is_null", &inner_arrow)
467            }
468            UnaryOp::IsNotNull => {
469                if let Expression::Variable(name) = inner
470                    && !chunk.field_names.is_empty()
471                    && !chunk.field_names.iter().any(|n| n == name)
472                {
473                    let prefix = format!("{}.", name);
474                    let cols: Vec<usize> = chunk
475                        .field_names
476                        .iter()
477                        .enumerate()
478                        .filter(|(_, n)| n.starts_with(prefix.as_str()))
479                        .map(|(i, _)| i)
480                        .collect();
481                    if !cols.is_empty() {
482                        let size = chunk.size;
483                        let mut acc: Option<arrow::array::BooleanArray> = None;
484                        for &col in &cols {
485                            let field = chunk
486                                .fields
487                                .get(col)
488                                .ok_or_else(|| format!("Variable '{}' column {} not found", name, col))?;
489                            let nulls: Vec<bool> = (0..size).map(|i| field.is_null(i)).collect();
490                            let m = arrow::array::BooleanArray::from(nulls);
491                            acc = Some(match acc {
492                                Some(a) => arrow::compute::kernels::boolean::and(&a, &m)
493                                    .map_err(|e| format!("Arrow is_not_null over variable failed: {e}"))?,
494                                None => m,
495                            });
496                        }
497                        let all_null = acc.unwrap();
498                        let not_all = arrow::compute::kernels::boolean::not(&all_null)
499                            .map_err(|e| format!("Arrow is_not_null over variable failed: {e}"))?;
500                        return Ok(ArrowVector::new(Arc::new(not_all), PhysicalTypeID::Bool));
501                    }
502                }
503                let inner_arrow = self.evaluate_arrow(inner, chunk)?;
504                self.apply_arrow_unary_kernel("is_not_null", &inner_arrow)
505            }
506        }
507    }
508
509    /// Apply an Arrow binary compute kernel.
510    fn apply_arrow_kernel(
511        &self,
512        name: &str,
513        left: &ArrowVector,
514        right: &ArrowVector,
515    ) -> Result<ArrowVector, ProcessorError> {
516        use arrow::compute::kernels::boolean::{and_kleene, or_kleene};
517        use arrow::compute::kernels::cmp::{eq, gt, gt_eq, lt, lt_eq, neq};
518        use arrow::compute::kernels::numeric::{add, div, mul, rem, sub};
519
520        let result: ArrayRef = match name {
521            "add" => Arc::new(add(&left.array, &right.array).map_err(|e| format!("Arrow {name} failed: {e}"))?),
522            "sub" => Arc::new(sub(&left.array, &right.array).map_err(|e| format!("Arrow {name} failed: {e}"))?),
523            "mul" => Arc::new(mul(&left.array, &right.array).map_err(|e| format!("Arrow {name} failed: {e}"))?),
524            "div" => Arc::new(div(&left.array, &right.array).map_err(|e| format!("Arrow {name} failed: {e}"))?),
525            "mod" => Arc::new(rem(&left.array, &right.array).map_err(|e| format!("Arrow {name} failed: {e}"))?),
526            "eq" => Arc::new(eq(&left.array, &right.array).map_err(|e| format!("Arrow {name} failed: {e}"))?),
527            "neq" => Arc::new(neq(&left.array, &right.array).map_err(|e| format!("Arrow {name} failed: {e}"))?),
528            "lt" => Arc::new(lt(&left.array, &right.array).map_err(|e| format!("Arrow {name} failed: {e}"))?),
529            "lt_eq" => Arc::new(lt_eq(&left.array, &right.array).map_err(|e| format!("Arrow {name} failed: {e}"))?),
530            "gt" => Arc::new(gt(&left.array, &right.array).map_err(|e| format!("Arrow {name} failed: {e}"))?),
531            "gt_eq" => Arc::new(gt_eq(&left.array, &right.array).map_err(|e| format!("Arrow {name} failed: {e}"))?),
532            "and" => {
533                let l = left
534                    .array
535                    .as_any()
536                    .downcast_ref::<arrow::array::BooleanArray>()
537                    .ok_or_else(|| format!("Arrow {name}: expected BooleanArray, got {:?}", left.array.data_type()))?;
538                let r = right
539                    .array
540                    .as_any()
541                    .downcast_ref::<arrow::array::BooleanArray>()
542                    .ok_or_else(|| format!("Arrow {name}: expected BooleanArray, got {:?}", right.array.data_type()))?;
543                Arc::new(and_kleene(l, r).map_err(|e| format!("Arrow {name} failed: {e}"))?)
544            }
545            "or" => {
546                let l = left
547                    .array
548                    .as_any()
549                    .downcast_ref::<arrow::array::BooleanArray>()
550                    .ok_or_else(|| format!("Arrow {name}: expected BooleanArray, got {:?}", left.array.data_type()))?;
551                let r = right
552                    .array
553                    .as_any()
554                    .downcast_ref::<arrow::array::BooleanArray>()
555                    .ok_or_else(|| format!("Arrow {name}: expected BooleanArray, got {:?}", right.array.data_type()))?;
556                Arc::new(or_kleene(l, r).map_err(|e| format!("Arrow {name} failed: {e}"))?)
557            }
558            "xor" => {
559                // XOR = (l AND NOT r) OR (NOT l AND r)
560                let l = left
561                    .array
562                    .as_any()
563                    .downcast_ref::<arrow::array::BooleanArray>()
564                    .ok_or_else(|| format!("Arrow {name}: expected BooleanArray, got {:?}", left.array.data_type()))?;
565                let r = right
566                    .array
567                    .as_any()
568                    .downcast_ref::<arrow::array::BooleanArray>()
569                    .ok_or_else(|| format!("Arrow {name}: expected BooleanArray, got {:?}", right.array.data_type()))?;
570                let not_r =
571                    arrow::compute::kernels::boolean::not(r).map_err(|e| format!("Arrow xor/not failed: {e}"))?;
572                let not_l =
573                    arrow::compute::kernels::boolean::not(l).map_err(|e| format!("Arrow xor/not failed: {e}"))?;
574                let l_and_not_r = and_kleene(l, &not_r).map_err(|e| format!("Arrow xor/and failed: {e}"))?;
575                let not_l_and_r = and_kleene(&not_l, r).map_err(|e| format!("Arrow xor/and failed: {e}"))?;
576                Arc::new(or_kleene(&l_and_not_r, &not_l_and_r).map_err(|e| format!("Arrow xor/or failed: {e}"))?)
577            }
578            _ => return Err(format!("Unknown binary kernel: {name}").into()),
579        };
580
581        let phys_type = match name {
582            "eq" | "neq" | "lt" | "lt_eq" | "gt" | "gt_eq" | "and" | "or" | "xor" => PhysicalTypeID::Bool,
583            _ => left.physical_type,
584        };
585
586        Ok(ArrowVector::new(result, phys_type))
587    }
588
589    /// Apply an Arrow unary compute kernel.
590    fn apply_arrow_unary_kernel(&self, name: &str, arr: &ArrowVector) -> Result<ArrowVector, ProcessorError> {
591        use arrow::compute::kernels::boolean::{is_not_null, is_null, not};
592        use arrow::compute::kernels::numeric::neg;
593
594        let result: ArrayRef = match name {
595            "not" => {
596                let arr_ref = arr
597                    .array
598                    .as_any()
599                    .downcast_ref::<arrow::array::BooleanArray>()
600                    .ok_or_else(|| format!("Arrow {name}: expected BooleanArray, got {:?}", arr.array.data_type()))?;
601                Arc::new(not(arr_ref).map_err(|e| format!("Arrow {name} failed: {e}"))?)
602            }
603            "negate" => Arc::new(neg(&*arr.array).map_err(|e| format!("Arrow {name} failed: {e}"))?),
604            "is_null" => Arc::new(is_null(&*arr.array).map_err(|e| format!("Arrow {name} failed: {e}"))?),
605            "is_not_null" => Arc::new(is_not_null(&*arr.array).map_err(|e| format!("Arrow {name} failed: {e}"))?),
606            _ => return Err(format!("Unknown unary kernel: {name}").into()),
607        };
608
609        let phys_type = if matches!(name, "is_null" | "is_not_null" | "not") {
610            PhysicalTypeID::Bool
611        } else {
612            arr.physical_type
613        };
614
615        Ok(ArrowVector::new(result, phys_type))
616    }
617
618    /// Evaluate a function call, producing ArrowVector directly.
619    /// For fixed-width return types, uses typed Vec<T> collection to
620    /// avoid the intermediate ValueVector allocation.
621    fn evaluate_arrow_function_call(
622        &self,
623        name: &str,
624        args: &[&Expression],
625        chunk: &DataChunk,
626    ) -> Result<ArrowVector, ProcessorError> {
627        // Lambda-based functions use complex control flow — stick with evaluate
628        if let Some(_lambda) = self.extract_lambda_arg(args) {
629            match name {
630                "list_transform" | "list_filter" | "list_reduce" => {
631                    let legacy = self.evaluate_function_call(name, args, chunk)?;
632                    return Ok(ArrowVector::from_legacy(&legacy));
633                }
634                _ => {}
635            }
636        }
637
638        // Evaluate arguments as ArrowVectors
639        let arg_arrows: Vec<ArrowVector> = args
640            .iter()
641            .map(|arg| self.evaluate_arrow(arg, chunk))
642            .collect::<Result<Vec<_>, _>>()?;
643
644        if arg_arrows.is_empty() {
645            return Err(format!("Function '{}' requires at least one argument", name).into());
646        }
647
648        let num_rows = arg_arrows[0].size();
649
650        // Look up the function
651        let func = {
652            let reg = self.registry.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
653            reg.get_scalar(name).cloned()
654        };
655        let func = match func {
656            Some(f) => f,
657            None => return Err(format!("Unknown function: '{}'", name).into()),
658        };
659
660        // SequenceOp needs the callback — delegate to evaluate
661        if matches!(func, ScalarFunction::SequenceOp { .. }) {
662            let legacy = self.evaluate_function_call(name, args, chunk)?;
663            return Ok(ArrowVector::from_legacy(&legacy));
664        }
665
666        let mut first_error: Option<String> = None;
667        let mut row_results: Vec<Value> = Vec::with_capacity(num_rows);
668
669        for row in 0..num_rows {
670            let arg_values: Vec<Value> = arg_arrows
671                .iter()
672                .map(|arr| {
673                    if row < arr.size() && !arr.is_null(row) {
674                        arr.get_value(row).unwrap_or(Value::Null)
675                    } else {
676                        Value::Null
677                    }
678                })
679                .collect();
680
681            if !name.eq_ignore_ascii_case("coalesce")
682                && !name.eq_ignore_ascii_case("ifnull")
683                && arg_values.iter().any(|v| matches!(v, Value::Null))
684            {
685                row_results.push(Value::Null);
686                continue;
687            }
688
689            match evaluate_scalar(&func, &arg_values) {
690                Ok(val) => row_results.push(val),
691                Err(e) => {
692                    row_results.push(Value::Null);
693                    if first_error.is_none() {
694                        first_error = Some(e);
695                    }
696                }
697            }
698        }
699
700        let result_type = row_results
701            .iter()
702            .find(|v| !matches!(v, Value::Null))
703            .map(|v| v.physical_type())
704            .unwrap_or(PhysicalTypeID::Int64);
705
706        // Build Arrow array directly from typed Vec<T> — avoids ValueVector allocation
707        let arrow_result = build_arrow_from_values(&row_results, result_type, num_rows)?;
708
709        if row_results.iter().all(|v| matches!(v, Value::Null))
710            && let Some(e) = first_error
711        {
712            return Err(e.into());
713        }
714
715        Ok(arrow_result)
716    }
717
718    /// Evaluate a constant expression — returns a vector filled with the constant value.
719    fn evaluate_constant(&self, c: &Constant, size: usize) -> Result<ValueVector, ProcessorError> {
720        let val: Value = match c {
721            Constant::Null => Value::Null,
722            Constant::Bool(b) => Value::Bool(*b),
723            Constant::Integer(i) => Value::Int64(*i),
724            Constant::Float(f) => Value::Double(*f),
725            Constant::String(s) => Value::String(s.clone()),
726        };
727
728        let physical_type = val.physical_type();
729        let mut v = ValueVector::new(physical_type, size);
730        v.resize(size);
731        for i in 0..size {
732            store_value_in_vector(&mut v, i, &val)?;
733        }
734        Ok(v)
735    }
736
737    /// Evaluate a variable expression — reads a field from the DataChunk by variable name.
738    /// The variable name is matched against:
739    /// 1. Numeric index (binder resolves names to positions, e.g., "0", "1")
740    /// 2. Chunk field names (e.g., "title", "d.title")
741    /// 3. Falls back to the first field (legacy compatibility) if no match found.
742    fn evaluate_variable(&self, name: &str, chunk: &DataChunk) -> Result<ValueVector, ProcessorError> {
743        if let Ok(idx) = name.parse::<usize>() {
744            if idx >= chunk.fields.len() {
745                return Err(format!(
746                    "Variable '{}' (index {}) not found in chunk with {} fields",
747                    name,
748                    idx,
749                    chunk.fields.len()
750                )
751                .into());
752            }
753            let phys_type = chunk.field_types[idx];
754            let mut v = ValueVector::new(phys_type, chunk.size);
755            v.resize(chunk.size);
756            for i in 0..chunk.size {
757                if let Some(val) = chunk.get_value(idx, i) {
758                    store_value_in_vector(&mut v, i, &val)?;
759                } else {
760                    v.set_null(i, true);
761                }
762            }
763            return Ok(v);
764        }
765
766        if !chunk.field_names.is_empty() {
767            if let Some(idx) = chunk.field_names.iter().position(|n| n == name) {
768                let phys_type = chunk.field_types[idx];
769                let mut v = ValueVector::new(phys_type, chunk.size);
770                v.resize(chunk.size);
771                for i in 0..chunk.size {
772                    if let Some(val) = chunk.get_value(idx, i) {
773                        store_value_in_vector(&mut v, i, &val)?;
774                    } else {
775                        v.set_null(i, true);
776                    }
777                }
778                return Ok(v);
779            }
780            // P53.23: previously this silently fell back to column 0, so a
781            // missing column (e.g. an ORDER BY key referencing a non-projected
782            // column) produced a silently-wrong result instead of an error.
783            return Err(format!(
784                "Variable '{}' not found in chunk field_names {:?}",
785                name, chunk.field_names
786            )
787            .into());
788        }
789
790        if !chunk.fields.is_empty() {
791            let phys_type = chunk.field_types[0];
792            let mut v = ValueVector::new(phys_type, chunk.size);
793            v.resize(chunk.size);
794            for i in 0..chunk.size {
795                if let Some(val) = chunk.get_value(0, i) {
796                    store_value_in_vector(&mut v, i, &val)?;
797                } else {
798                    v.set_null(i, true);
799                }
800            }
801            Ok(v)
802        } else {
803            Ok(ValueVector::new(akar_common::types::PhysicalTypeID::Int64, 0))
804        }
805    }
806
807    /// Evaluate a property access expression — resolves the property name to a
808    /// column index using `chunk.field_names`, then returns that column's data.
809    ///
810    /// Falls back to evaluating the object expression (legacy behaviour) if no
811    /// `field_names` are available on the chunk.
812    fn evaluate_property_access(
813        &self,
814        obj: &Expression,
815        prop: &str,
816        chunk: &DataChunk,
817    ) -> Result<ValueVector, ProcessorError> {
818        // Build the qualified property name (e.g., "t.name")
819        let qualified_prop = if let Expression::Variable(var_name) = obj {
820            format!("{}.{}", var_name, prop)
821        } else {
822            prop.to_string()
823        };
824
825        // Fast path: look up the property by name in the chunk's field names
826        // (exact qualified match only).
827        if !chunk.field_names.is_empty()
828            && let Some(idx) = chunk.field_names.iter().position(|n| n == &qualified_prop)
829        {
830            if chunk.fields.get(idx).is_none() {
831                return Err(format!("Column '{}' (index {}) not found in chunk", prop, idx).into());
832            }
833            let phys_type = chunk.field_types[idx];
834            let mut v = ValueVector::new(phys_type, chunk.size);
835            v.resize(chunk.size);
836            for i in 0..chunk.size {
837                if let Some(val) = chunk.get_value(idx, i) {
838                    store_value_in_vector(&mut v, i, &val)?;
839                } else {
840                    v.set_null(i, true);
841                }
842            }
843            return Ok(v);
844        }
845        // P53.26: `row.id` where `row` is a map/struct column — extract the key
846        // from each row's value instead of returning the whole object. Runs
847        // BEFORE the bare-name match so a plain `content` table column does not
848        // shadow an UNWIND map variable's `row.content`.
849        if let Expression::Variable(var_name) = obj
850            && let Some(idx) = chunk.field_names.iter().position(|n| n == var_name)
851        {
852            let extracted: Vec<Value> = (0..chunk.size)
853                .map(|i| map_property_value(&chunk.get_value(idx, i).unwrap_or(Value::Null), prop))
854                .collect();
855            if extracted.iter().any(|v| !matches!(v, Value::Null)) {
856                let result_type = extracted
857                    .iter()
858                    .find(|v| !matches!(v, Value::Null))
859                    .map(|v| v.physical_type())
860                    .unwrap_or(PhysicalTypeID::Int64);
861                let mut v = ValueVector::new(result_type, chunk.size);
862                v.resize(chunk.size);
863                for (i, val) in extracted.iter().enumerate() {
864                    if matches!(val, Value::Null) {
865                        v.set_null(i, true);
866                    } else {
867                        store_value_in_vector(&mut v, i, val)?;
868                    }
869                }
870                return Ok(v);
871            }
872        }
873        // Bare-name match (plain table columns, e.g. `SET m.content = content`).
874        if !chunk.field_names.is_empty()
875            && let Some(idx) = chunk.field_names.iter().position(|n| n == prop)
876        {
877            if chunk.fields.get(idx).is_none() {
878                return Err(format!("Column '{}' (index {}) not found in chunk", prop, idx).into());
879            }
880            let phys_type = chunk.field_types[idx];
881            let mut v = ValueVector::new(phys_type, chunk.size);
882            v.resize(chunk.size);
883            for i in 0..chunk.size {
884                if let Some(val) = chunk.get_value(idx, i) {
885                    store_value_in_vector(&mut v, i, &val)?;
886                } else {
887                    v.set_null(i, true);
888                }
889            }
890            return Ok(v);
891        }
892        // Fallback: evaluate the object expression (returns first column — legacy behaviour).
893        self.evaluate(obj, chunk)
894    }
895
896    /// Evaluate a function call expression.
897    fn evaluate_function_call(
898        &self,
899        name: &str,
900        args: &[&Expression],
901        chunk: &DataChunk,
902    ) -> Result<ValueVector, ProcessorError> {
903        // Handle lambda-based list functions at expression level.
904        // These cannot go through the normal scalar function pipeline because
905        // lambda expressions are not Values and must be evaluated per-element.
906        if let Some(lambda) = self.extract_lambda_arg(args) {
907            match name {
908                "list_transform" => return self.evaluate_list_transform(args, lambda, chunk),
909                "list_filter" => return self.evaluate_list_filter(args, lambda, chunk),
910                "list_reduce" => return self.evaluate_list_reduce(args, lambda, chunk),
911                _ => {}
912            }
913        }
914
915        // Evaluate all argument expressions first
916        let arg_vectors: Vec<ValueVector> = args
917            .iter()
918            .map(|arg| self.evaluate(arg, chunk))
919            .collect::<Result<Vec<_>, _>>()?;
920
921        if arg_vectors.is_empty() {
922            return Err(format!("Function '{}' requires at least one argument", name).into());
923        }
924
925        let num_rows = arg_vectors[0].size();
926
927        // Look up the function in the registry
928        let func = {
929            let reg = self.registry.lock().map_err(|e| format!("Lock poisoned: {e}"))?;
930            reg.get_scalar(name).cloned()
931        };
932
933        let func = match func {
934            Some(f) => f,
935            None => return Err(format!("Unknown function: '{}'", name).into()),
936        };
937
938        // Handle SequenceOp (nextval/currval) via callback with catalog access
939        if matches!(func, ScalarFunction::SequenceOp { .. }) {
940            return self.evaluate_sequence_op(name, &func, &arg_vectors, num_rows);
941        }
942
943        // Evaluate each row exactly once, then infer result type from cached values.
944        // This avoids double-invoking side-effecting functions (e.g., nextval custom scalar).
945        let mut row_results: Vec<Value> = Vec::with_capacity(num_rows);
946        let mut first_error: Option<String> = None;
947
948        for row in 0..num_rows {
949            let arg_values: Vec<Value> = arg_vectors
950                .iter()
951                .map(|vec| {
952                    if row < vec.size() && !vec.is_null(row) {
953                        vec.get_value(row).unwrap_or(Value::Null)
954                    } else {
955                        Value::Null
956                    }
957                })
958                .collect();
959
960            // Three-valued logic for AND/OR (short-circuit)
961            if name == "AND" || name == "OR" {
962                let l = &arg_values[0];
963                let r = &arg_values[1];
964                if name == "AND" {
965                    if matches!(l, Value::Bool(false)) || matches!(r, Value::Bool(false)) {
966                        row_results.push(Value::Bool(false));
967                        continue;
968                    }
969                    if matches!(l, Value::Null) || matches!(r, Value::Null) {
970                        row_results.push(Value::Null);
971                        continue;
972                    }
973                } else {
974                    if matches!(l, Value::Bool(true)) || matches!(r, Value::Bool(true)) {
975                        row_results.push(Value::Bool(true));
976                        continue;
977                    }
978                    if matches!(l, Value::Null) || matches!(r, Value::Null) {
979                        row_results.push(Value::Null);
980                        continue;
981                    }
982                }
983            }
984
985            // If any argument is null, the result is null (SQL NULL semantics)
986            // Coalesce/ifnull are exempt — they intentionally inspect null args
987            if !name.eq_ignore_ascii_case("coalesce")
988                && !name.eq_ignore_ascii_case("ifnull")
989                && arg_values.iter().any(|v| matches!(v, Value::Null))
990            {
991                row_results.push(Value::Null);
992                continue;
993            }
994
995            match evaluate_scalar(&func, &arg_values) {
996                Ok(val) => row_results.push(val),
997                Err(e) => {
998                    row_results.push(Value::Null);
999                    if first_error.is_none() {
1000                        first_error = Some(e);
1001                    }
1002                }
1003            }
1004        }
1005
1006        let result_type = row_results
1007            .iter()
1008            .find(|v| !matches!(v, Value::Null))
1009            .map(|v| v.physical_type())
1010            .unwrap_or(akar_common::types::PhysicalTypeID::Int64);
1011
1012        let mut result_vec = ValueVector::new(result_type, num_rows);
1013        result_vec.resize(num_rows);
1014        for (row, val) in row_results.iter().enumerate() {
1015            store_value_in_vector(&mut result_vec, row, val)?;
1016        }
1017
1018        if row_results.iter().all(|v| matches!(v, Value::Null))
1019            && let Some(e) = first_error
1020        {
1021            return Err(e.into());
1022        }
1023
1024        Ok(result_vec)
1025    }
1026
1027    /// Evaluate a binary operation.
1028    fn evaluate_binary_op(
1029        &self,
1030        op: &BinaryOp,
1031        left: &Expression,
1032        right: &Expression,
1033        chunk: &DataChunk,
1034    ) -> Result<ValueVector, ProcessorError> {
1035        // Map AST BinaryOp to a scalar function name
1036        let func_name = match op {
1037            BinaryOp::Add => "+",
1038            BinaryOp::Subtract => "-",
1039            BinaryOp::Multiply => "*",
1040            BinaryOp::Divide => "/",
1041            BinaryOp::Modulo => "%",
1042            BinaryOp::Equal => "=",
1043            BinaryOp::NotEqual => "<>",
1044            BinaryOp::LessThan => "<",
1045            BinaryOp::LessThanOrEqual => "<=",
1046            BinaryOp::GreaterThan => ">",
1047            BinaryOp::GreaterThanOrEqual => ">=",
1048            BinaryOp::And => "AND",
1049            BinaryOp::Or => "OR",
1050            BinaryOp::Xor => "XOR",
1051            BinaryOp::Concat => "concat",
1052            // Handled inline — not mapped to scalar function
1053            BinaryOp::In | BinaryOp::NotIn => {
1054                return self.evaluate_in_op(op, left, right, chunk);
1055            }
1056            BinaryOp::StartsWith => "starts_with",
1057            BinaryOp::EndsWith => "ends_with",
1058            BinaryOp::Contains => "contains",
1059            BinaryOp::Like => {
1060                return self.evaluate_function_call("like", &[left, right], chunk);
1061            }
1062        };
1063
1064        // Treat as a function call with two arguments
1065        self.evaluate_function_call(func_name, &[left, right], chunk)
1066    }
1067
1068    /// Evaluate a unary operation.
1069    fn evaluate_unary_op(
1070        &self,
1071        op: &UnaryOp,
1072        inner: &Expression,
1073        chunk: &DataChunk,
1074    ) -> Result<ValueVector, ProcessorError> {
1075        match op {
1076            UnaryOp::Not => self.evaluate_function_call("NOT", std::slice::from_ref(&inner), chunk),
1077            UnaryOp::Negate => self.evaluate_function_call("-", std::slice::from_ref(&inner), chunk),
1078            UnaryOp::IsNull => {
1079                let vec = self.evaluate(inner, chunk)?;
1080                let num_rows = vec.size();
1081                let mut result = ValueVector::new(akar_common::types::PhysicalTypeID::Bool, num_rows);
1082                result.resize(num_rows);
1083                for i in 0..num_rows {
1084                    let is_null = vec.is_null(i) || matches!(vec.get_value(i), Some(Value::Null) | None);
1085                    store_value_in_vector_simple(&mut result, i, &Value::Bool(is_null))?;
1086                }
1087                Ok(result)
1088            }
1089            UnaryOp::IsNotNull => {
1090                let vec = self.evaluate(inner, chunk)?;
1091                let num_rows = vec.size();
1092                let mut result = ValueVector::new(akar_common::types::PhysicalTypeID::Bool, num_rows);
1093                result.resize(num_rows);
1094                for i in 0..num_rows {
1095                    let is_null = vec.is_null(i) || matches!(vec.get_value(i), Some(Value::Null) | None);
1096                    store_value_in_vector_simple(&mut result, i, &Value::Bool(!is_null))?;
1097                }
1098                Ok(result)
1099            }
1100        }
1101    }
1102
1103    /// Evaluate `x IN list` and `x NOT IN list` operators.
1104    fn evaluate_in_op(
1105        &self,
1106        op: &BinaryOp,
1107        left: &Expression,
1108        right: &Expression,
1109        chunk: &DataChunk,
1110    ) -> Result<ValueVector, ProcessorError> {
1111        let left_arr = self.evaluate_arrow(left, chunk)?;
1112        let num_rows = chunk.size;
1113        let mut result = ValueVector::new(akar_common::types::PhysicalTypeID::Bool, num_rows);
1114        result.resize(num_rows);
1115        for row in 0..num_rows {
1116            let lv = left_arr.get_value(row).unwrap_or(Value::Null);
1117            if matches!(lv, Value::Null) {
1118                result.set_null(row, true);
1119                continue;
1120            }
1121            let (in_list, has_null) = match right {
1122                Expression::List(items) => {
1123                    let mut matched = false;
1124                    let mut has_null_item = false;
1125                    for item in items {
1126                        let item_vec = self.evaluate(item, chunk)?;
1127                        let iv = item_vec.get_value(row).unwrap_or(Value::Null);
1128                        if matches!(iv, Value::Null) {
1129                            has_null_item = true;
1130                        } else if iv == lv {
1131                            matched = true;
1132                            break;
1133                        }
1134                    }
1135                    (matched, has_null_item)
1136                }
1137                _ => {
1138                    let right_arr = self.evaluate_arrow(right, chunk)?;
1139                    let rv = right_arr.get_value(row).unwrap_or(Value::Null);
1140                    match &rv {
1141                        Value::List(ritems) => {
1142                            let matched = ritems.contains(&lv);
1143                            let has_null_item = if matched {
1144                                false
1145                            } else {
1146                                ritems.iter().any(|item| matches!(item, Value::Null))
1147                            };
1148                            (matched, has_null_item)
1149                        }
1150                        _ => (lv == rv, false),
1151                    }
1152                }
1153            };
1154            let result_val = if *op == BinaryOp::NotIn { !in_list } else { in_list };
1155            if !in_list && has_null {
1156                result.set_null(row, true);
1157            } else {
1158                store_value_in_vector_simple(&mut result, row, &Value::Bool(result_val))?;
1159            }
1160        }
1161        Ok(result)
1162    }
1163
1164    /// Evaluate a `CASE [subject] WHEN ... THEN ... [ELSE ...] END` expression.
1165    fn evaluate_case(
1166        &self,
1167        case_expr: &akar_parser::ast::CaseExpr,
1168        chunk: &DataChunk,
1169    ) -> Result<ValueVector, ProcessorError> {
1170        let num_rows = chunk.size;
1171        // Evaluate subject (if any)
1172        let subject_vec = if let Some(subj) = &case_expr.subject {
1173            Some(self.evaluate(subj, chunk)?)
1174        } else {
1175            None
1176        };
1177
1178        // We need to find the result type first by speculatively checking THEN exprs
1179        // Strategy: evaluate all WHEN/THEN in order per row
1180        // Determine result type from first THEN expr evaluation
1181        let result_type = {
1182            let first_then = self.evaluate(&case_expr.alternatives[0].then, chunk)?;
1183            first_then.physical_type()
1184        };
1185
1186        let mut result = ValueVector::new(result_type, num_rows);
1187        result.resize(num_rows);
1188
1189        // For each row, find the matching branch
1190        for row in 0..num_rows {
1191            let subject_val = subject_vec.as_ref().and_then(|sv| sv.get_value(row));
1192
1193            let mut matched = false;
1194            for alt in &case_expr.alternatives {
1195                let when_vec = self.evaluate(&alt.when, chunk)?;
1196                let when_val = when_vec.get_value(row).unwrap_or(Value::Null);
1197
1198                // Simple CASE: compare subject == when_val
1199                // Searched CASE: when_val is a boolean condition
1200                let branch_taken = if let Some(ref sv) = subject_val {
1201                    when_val != Value::Null && when_val == *sv
1202                } else {
1203                    matches!(when_val, Value::Bool(true))
1204                };
1205
1206                if branch_taken {
1207                    let then_vec = self.evaluate(&alt.then, chunk)?;
1208                    let then_val = then_vec.get_value(row).unwrap_or(Value::Null);
1209                    store_value_in_vector(&mut result, row, &then_val)?;
1210                    matched = true;
1211                    break;
1212                }
1213            }
1214
1215            if !matched {
1216                if let Some(else_e) = &case_expr.else_expr {
1217                    let else_vec = self.evaluate(else_e, chunk)?;
1218                    let else_val = else_vec.get_value(row).unwrap_or(Value::Null);
1219                    store_value_in_vector(&mut result, row, &else_val)?;
1220                } else {
1221                    result.set_null(row, true);
1222                }
1223            }
1224        }
1225
1226        Ok(result)
1227    }
1228
1229    /// Evaluate a list literal expression.
1230    fn evaluate_list_literal(&self, items: &[Expression], chunk: &DataChunk) -> Result<ValueVector, ProcessorError> {
1231        if items.is_empty() {
1232            let mut v = ValueVector::new(akar_common::types::PhysicalTypeID::List, chunk.size);
1233            v.resize(chunk.size);
1234            for i in 0..chunk.size {
1235                store_value_in_vector(&mut v, i, &Value::List(vec![]))?;
1236            }
1237            return Ok(v);
1238        }
1239
1240        let num_rows = chunk.size;
1241        let mut result_vec = ValueVector::new(akar_common::types::PhysicalTypeID::List, num_rows);
1242        result_vec.resize(num_rows);
1243
1244        for row in 0..num_rows {
1245            // For each row, evaluate each item expression against the same chunk
1246            // (list literals reference the whole chunk, so we get the per-row values)
1247            let mut list_values = Vec::with_capacity(items.len());
1248            for item in items {
1249                let item_vec = self.evaluate(item, chunk)?;
1250                let val = if row < item_vec.size() {
1251                    item_vec.get_value(row).unwrap_or(Value::Null)
1252                } else {
1253                    Value::Null
1254                };
1255                list_values.push(val);
1256            }
1257            store_value_in_vector(&mut result_vec, row, &Value::List(list_values))?;
1258        }
1259
1260        Ok(result_vec)
1261    }
1262
1263    /// Evaluate a map literal expression.
1264    fn evaluate_map_literal(
1265        &self,
1266        items: &[(String, Expression)],
1267        chunk: &DataChunk,
1268    ) -> Result<ValueVector, ProcessorError> {
1269        let num_rows = chunk.size;
1270        let mut result_vec = ValueVector::new(akar_common::types::PhysicalTypeID::Struct, num_rows);
1271        result_vec.resize(num_rows);
1272
1273        for row in 0..num_rows {
1274            let mut map_values = Vec::with_capacity(items.len());
1275            for (key, item) in items {
1276                let item_vec = self.evaluate(item, chunk)?;
1277                let val = if row < item_vec.size() {
1278                    item_vec.get_value(row).unwrap_or(Value::Null)
1279                } else {
1280                    Value::Null
1281                };
1282                map_values.push((Value::String(key.clone()), val));
1283            }
1284            store_value_in_vector(&mut result_vec, row, &Value::Map(map_values))?;
1285        }
1286
1287        Ok(result_vec)
1288    }
1289
1290    /// Evaluate an ANY/ALL/NONE/SINGLE list predicate.
1291    /// Evaluates the list expression, then for each element evaluates the
1292    /// predicate and applies the quantifier logic.
1293    fn evaluate_list_predicate(
1294        &self,
1295        quantifier: &akar_parser::ast::Quantifier,
1296        list: &Expression,
1297        _var_name: &str,
1298        predicate: &Expression,
1299        chunk: &DataChunk,
1300    ) -> Result<ValueVector, ProcessorError> {
1301        // Evaluate the list expression to get a ValueVector
1302        let list_vec = self.evaluate(list, chunk)?;
1303        let num_rows = chunk.size;
1304        let mut result = ValueVector::new(akar_common::types::PhysicalTypeID::Bool, num_rows);
1305        result.resize(num_rows);
1306
1307        for row in 0..num_rows {
1308            let list_val = list_vec.get_value(row).unwrap_or(Value::Null);
1309            let items = match &list_val {
1310                Value::List(items) => items.as_slice(),
1311                _ => {
1312                    // Not a list → false for all quantifiers
1313                    store_value_in_vector(&mut result, row, &Value::Bool(false))?;
1314                    continue;
1315                }
1316            };
1317
1318            // For each element, create a mini-chunk with the variable bound
1319            // and evaluate the predicate
1320            let mut true_count = 0u64;
1321            for item in items {
1322                // Create a single-row chunk with the variable as first field
1323                let mut elem_vec = ValueVector::new(item.physical_type(), 1);
1324                elem_vec.resize(1);
1325                store_value_in_vector(&mut elem_vec, 0, item)?;
1326                let mini_chunk = {
1327                    let arrow_fields = vec![&elem_vec]
1328                        .into_iter()
1329                        .map(|v| akar_common::arrow_vector::ArrowVector::from_legacy(v).array)
1330                        .collect::<Vec<_>>();
1331                    let arrow_field_types = vec![&elem_vec]
1332                        .into_iter()
1333                        .map(|v| v.physical_type())
1334                        .collect::<Vec<_>>();
1335                    DataChunk::new(arrow_fields, arrow_field_types)
1336                };
1337
1338                let pred_vec = self.evaluate(predicate, &mini_chunk)?;
1339                let pred_val = pred_vec.get_value(0).unwrap_or(Value::Null);
1340
1341                if matches!(pred_val, Value::Bool(true)) {
1342                    true_count += 1;
1343                }
1344            }
1345
1346            // Apply quantifier logic
1347            let elem_count = items.len() as u64;
1348            let bool_result = match quantifier {
1349                akar_parser::ast::Quantifier::Any => true_count > 0,
1350                akar_parser::ast::Quantifier::All => !items.is_empty() && true_count == elem_count,
1351                akar_parser::ast::Quantifier::None => true_count == 0,
1352                akar_parser::ast::Quantifier::Single => true_count == 1,
1353            };
1354            store_value_in_vector(&mut result, row, &Value::Bool(bool_result))?;
1355        }
1356
1357        Ok(result)
1358    }
1359
1360    /// Extract the Lambda expression from function call arguments, if present.
1361    fn extract_lambda_arg<'a>(&self, args: &'a [&Expression]) -> Option<&'a Expression> {
1362        args.iter().find(|a| matches!(a, Expression::Lambda { .. })).copied()
1363    }
1364
1365    /// Evaluate `list_transform(list, x -> body)` — apply lambda to each element.
1366    fn evaluate_list_transform(
1367        &self,
1368        args: &[&Expression],
1369        lambda: &Expression,
1370        chunk: &DataChunk,
1371    ) -> Result<ValueVector, ProcessorError> {
1372        let list_expr = args
1373            .iter()
1374            .find(|a| !matches!(a, Expression::Lambda { .. }))
1375            .ok_or("list_transform requires a list argument")?;
1376
1377        let (var_name, body) = match lambda {
1378            Expression::Lambda { var_name, body } => (var_name, body),
1379            _ => return Err("Expected lambda expression".into()),
1380        };
1381
1382        let list_vec = self.evaluate(list_expr, chunk)?;
1383        let num_rows = chunk.size;
1384        let mut result = ValueVector::new(akar_common::types::PhysicalTypeID::List, num_rows);
1385        result.resize(num_rows);
1386
1387        for row in 0..num_rows {
1388            let list_val = list_vec.get_value(row).unwrap_or(Value::Null);
1389            let items = match list_val {
1390                Value::List(items) => items,
1391                _ => {
1392                    store_value_in_vector(&mut result, row, &Value::List(vec![]))?;
1393                    continue;
1394                }
1395            };
1396
1397            let mut transformed: Vec<Value> = Vec::with_capacity(items.len());
1398            for item in items {
1399                let mut elem_vec = ValueVector::new(item.physical_type(), 1);
1400                elem_vec.resize(1);
1401                store_value_in_vector(&mut elem_vec, 0, &item)?;
1402                let mut mini_chunk = {
1403                    let arrow_fields = vec![&elem_vec]
1404                        .into_iter()
1405                        .map(|v| akar_common::arrow_vector::ArrowVector::from_legacy(v).array)
1406                        .collect::<Vec<_>>();
1407                    let arrow_field_types = vec![&elem_vec]
1408                        .into_iter()
1409                        .map(|v| v.physical_type())
1410                        .collect::<Vec<_>>();
1411                    DataChunk::new(arrow_fields, arrow_field_types)
1412                };
1413                mini_chunk.field_names.push(var_name.clone());
1414
1415                let body_vec = self.evaluate(body, &mini_chunk)?;
1416                let body_val = body_vec.get_value(0).unwrap_or(Value::Null);
1417                transformed.push(body_val);
1418            }
1419            store_value_in_vector(&mut result, row, &Value::List(transformed))?;
1420        }
1421
1422        Ok(result)
1423    }
1424
1425    /// Evaluate `list_filter(list, x -> predicate)` — keep elements where predicate is true.
1426    fn evaluate_list_filter(
1427        &self,
1428        args: &[&Expression],
1429        lambda: &Expression,
1430        chunk: &DataChunk,
1431    ) -> Result<ValueVector, ProcessorError> {
1432        let list_expr = args
1433            .iter()
1434            .find(|a| !matches!(a, Expression::Lambda { .. }))
1435            .ok_or("list_filter requires a list argument")?;
1436
1437        let (var_name, body) = match lambda {
1438            Expression::Lambda { var_name, body } => (var_name, body),
1439            _ => return Err("Expected lambda expression".into()),
1440        };
1441
1442        let list_vec = self.evaluate(list_expr, chunk)?;
1443        let num_rows = chunk.size;
1444        let mut result = ValueVector::new(akar_common::types::PhysicalTypeID::List, num_rows);
1445        result.resize(num_rows);
1446
1447        for row in 0..num_rows {
1448            let list_val = list_vec.get_value(row).unwrap_or(Value::Null);
1449            let items = match list_val {
1450                Value::List(items) => items,
1451                _ => {
1452                    store_value_in_vector(&mut result, row, &Value::List(vec![]))?;
1453                    continue;
1454                }
1455            };
1456
1457            let mut filtered: Vec<Value> = Vec::with_capacity(items.len());
1458            for item in items {
1459                let mut elem_vec = ValueVector::new(item.physical_type(), 1);
1460                elem_vec.resize(1);
1461                store_value_in_vector(&mut elem_vec, 0, &item)?;
1462                let mut mini_chunk = {
1463                    let arrow_fields = vec![&elem_vec]
1464                        .into_iter()
1465                        .map(|v| akar_common::arrow_vector::ArrowVector::from_legacy(v).array)
1466                        .collect::<Vec<_>>();
1467                    let arrow_field_types = vec![&elem_vec]
1468                        .into_iter()
1469                        .map(|v| v.physical_type())
1470                        .collect::<Vec<_>>();
1471                    DataChunk::new(arrow_fields, arrow_field_types)
1472                };
1473                mini_chunk.field_names.push(var_name.clone());
1474
1475                let pred_vec = self.evaluate(body, &mini_chunk)?;
1476                let pred_val = pred_vec.get_value(0).unwrap_or(Value::Null);
1477
1478                if matches!(pred_val, Value::Bool(true)) {
1479                    filtered.push(item);
1480                } else if let Value::Int64(x) = pred_val {
1481                    if x != 0 {
1482                        filtered.push(item);
1483                    }
1484                }
1485            }
1486            store_value_in_vector(&mut result, row, &Value::List(filtered))?;
1487        }
1488
1489        Ok(result)
1490    }
1491
1492    /// Evaluate `list_reduce(list, (acc, x) -> body, initial)` — fold over list.
1493    fn evaluate_list_reduce(
1494        &self,
1495        args: &[&Expression],
1496        lambda: &Expression,
1497        chunk: &DataChunk,
1498    ) -> Result<ValueVector, ProcessorError> {
1499        let list_expr = args
1500            .iter()
1501            .find(|a| !matches!(a, Expression::Lambda { .. }))
1502            .ok_or("list_reduce requires a list argument")?;
1503
1504        // Find initial value — the argument that is not the list and not the lambda
1505        let initial_expr = args
1506            .iter()
1507            .filter(|a| !matches!(a, Expression::Lambda { .. }))
1508            .nth(1) // Second non-lambda arg (first is list)
1509            .ok_or("list_reduce requires an initial value argument")?;
1510
1511        let (var_name, body) = match lambda {
1512            Expression::Lambda { var_name, body } => (var_name, body),
1513            _ => return Err("Expected lambda expression".into()),
1514        };
1515
1516        // list_reduce uses (acc, x) -> expr where acc is first var, x is second
1517        let acc_name = var_name.clone();
1518        let elem_name = match body.as_ref() {
1519            Expression::BinaryOp(_op, left, right) => {
1520                // Try to infer the element variable from the body pattern
1521                // Most common: acc + x where x is a Variable
1522                let left_var = if let Expression::Variable(v) = left.as_ref() {
1523                    Some(v.clone())
1524                } else {
1525                    None
1526                };
1527                let right_var = if let Expression::Variable(v) = right.as_ref() {
1528                    Some(v.clone())
1529                } else {
1530                    None
1531                };
1532
1533                if left_var.as_deref() == Some(&acc_name) {
1534                    right_var.unwrap_or_default()
1535                } else if right_var.as_deref() == Some(&acc_name) {
1536                    left_var.unwrap_or_default()
1537                } else {
1538                    String::new()
1539                }
1540            }
1541            _ => String::new(),
1542        };
1543
1544        let list_vec = self.evaluate(list_expr, chunk)?;
1545        let initial_vec = self.evaluate(initial_expr, chunk)?;
1546        let num_rows = chunk.size;
1547        let mut result = ValueVector::new(akar_common::types::PhysicalTypeID::Int64, num_rows);
1548        result.resize(num_rows);
1549
1550        // Pre-build field_names template (avoids N String clones per list element)
1551        let field_names_template = {
1552            let mut names = vec![acc_name.clone()];
1553            if !elem_name.is_empty() {
1554                names.push(elem_name.clone());
1555            }
1556            names
1557        };
1558
1559        for row in 0..num_rows {
1560            let list_val = list_vec.get_value(row).unwrap_or(Value::Null);
1561            let items = match list_val {
1562                Value::List(items) => items,
1563                _ => {
1564                    store_value_in_vector(&mut result, row, &Value::Null)?;
1565                    continue;
1566                }
1567            };
1568
1569            let mut acc = initial_vec.get_value(row).unwrap_or(Value::Null);
1570            for item in items {
1571                // Create mini-chunk with acc as field 0 and item as field 1
1572                let mut acc_vec = ValueVector::new(acc.physical_type(), 1);
1573                acc_vec.resize(1);
1574                store_value_in_vector(&mut acc_vec, 0, &acc)?;
1575                let mut elem_vec = ValueVector::new(item.physical_type(), 1);
1576                elem_vec.resize(1);
1577                store_value_in_vector(&mut elem_vec, 0, &item)?;
1578                let mut mini_chunk = {
1579                    let arrow_fields = vec![&acc_vec, &elem_vec]
1580                        .into_iter()
1581                        .map(|v| akar_common::arrow_vector::ArrowVector::from_legacy(v).array)
1582                        .collect::<Vec<_>>();
1583                    let arrow_field_types = vec![&acc_vec, &elem_vec]
1584                        .into_iter()
1585                        .map(|v| v.physical_type())
1586                        .collect::<Vec<_>>();
1587                    DataChunk::new(arrow_fields, arrow_field_types)
1588                };
1589                mini_chunk.field_names = field_names_template.clone();
1590
1591                let body_vec = self.evaluate(body, &mini_chunk)?;
1592                acc = body_vec.get_value(0).unwrap_or(Value::Null);
1593            }
1594            store_value_in_vector(&mut result, row, &acc)?;
1595        }
1596
1597        Ok(result)
1598    }
1599}
1600
1601/// Simplified store_value_in_vector that accepts any Value type.
1602/// This is a copy adapted from physical_operator.rs.
1603fn store_value_in_vector_simple(v: &mut ValueVector, row: usize, val: &Value) -> Result<(), String> {
1604    match val {
1605        Value::Null => {
1606            v.set_null(row, true);
1607        }
1608        Value::Bool(x) => {
1609            if v.physical_type() == akar_common::types::PhysicalTypeID::Bool {
1610                v.data_mut()[row] = if *x { 1 } else { 0 };
1611                v.set_null(row, false);
1612            }
1613        }
1614        Value::Int64(x) => {
1615            let offset = row * 8;
1616            if offset + 8 <= v.data().len() {
1617                v.data_mut()[offset..offset + 8].copy_from_slice(&x.to_le_bytes());
1618                v.set_null(row, false);
1619            }
1620        }
1621        Value::UInt64(x) => {
1622            let offset = row * 8;
1623            if offset + 8 <= v.data().len() {
1624                v.data_mut()[offset..offset + 8].copy_from_slice(&x.to_le_bytes());
1625                v.set_null(row, false);
1626            }
1627        }
1628        Value::Double(x) => {
1629            let offset = row * 8;
1630            if offset + 8 <= v.data().len() {
1631                v.data_mut()[offset..offset + 8].copy_from_slice(&x.to_le_bytes());
1632                v.set_null(row, false);
1633            }
1634        }
1635        Value::String(s) => {
1636            let bytes = s.as_bytes();
1637            if bytes.len() > 255 {
1638                return Err(format!(
1639                    "Cannot store string of {} bytes: inline string storage limit is 255 bytes",
1640                    bytes.len()
1641                ));
1642            }
1643            let offset = row * 256;
1644            if offset < v.data().len() {
1645                v.data_mut()[offset] = bytes.len() as u8;
1646                if offset + 1 + bytes.len() <= v.data().len() {
1647                    v.data_mut()[offset + 1..offset + 1 + bytes.len()].copy_from_slice(bytes);
1648                }
1649                v.set_null(row, false);
1650            }
1651        }
1652        _ => {
1653            v.set_null(row, true);
1654        }
1655    }
1656    Ok(())
1657}
1658
1659/// Store a Value into a ValueVector at the given row index.
1660/// This is a copy of the helper from physical_operator.rs, kept here for independence.
1661fn store_value_in_vector(v: &mut ValueVector, row: usize, val: &Value) -> Result<(), String> {
1662    match val {
1663        Value::Null => {
1664            v.set_null(row, true);
1665        }
1666        Value::Bool(x) => {
1667            if v.physical_type() == akar_common::types::PhysicalTypeID::Bool {
1668                v.data_mut()[row] = if *x { 1 } else { 0 };
1669                v.set_null(row, false);
1670            }
1671        }
1672        Value::Int64(x) => {
1673            let offset = row * 8;
1674            if offset + 8 <= v.data().len() {
1675                v.data_mut()[offset..offset + 8].copy_from_slice(&x.to_le_bytes());
1676                v.set_null(row, false);
1677            }
1678        }
1679        Value::UInt64(x) => {
1680            let offset = row * 8;
1681            if offset + 8 <= v.data().len() {
1682                v.data_mut()[offset..offset + 8].copy_from_slice(&x.to_le_bytes());
1683                v.set_null(row, false);
1684            }
1685        }
1686        Value::Date(x) => {
1687            let offset = row * 8;
1688            if offset + 8 <= v.data().len() {
1689                v.data_mut()[offset..offset + 8].copy_from_slice(&(x.0 as i64).to_le_bytes());
1690                v.set_null(row, false);
1691            }
1692        }
1693        Value::Timestamp(x) | Value::TimestampNs(x) | Value::TimestampMs(x) | Value::TimestampSec(x) => {
1694            let offset = row * 8;
1695            if offset + 8 <= v.data().len() {
1696                v.data_mut()[offset..offset + 8].copy_from_slice(&x.0.to_le_bytes());
1697                v.set_null(row, false);
1698            }
1699        }
1700        Value::TimestampTz(x) => {
1701            let offset = row * 8;
1702            if offset + 8 <= v.data().len() {
1703                v.data_mut()[offset..offset + 8].copy_from_slice(&x.0.to_le_bytes());
1704                v.set_null(row, false);
1705            }
1706        }
1707        Value::DTime(x) => {
1708            let offset = row * 8;
1709            if offset + 8 <= v.data().len() {
1710                v.data_mut()[offset..offset + 8].copy_from_slice(&x.to_le_bytes());
1711                v.set_null(row, false);
1712            }
1713        }
1714        Value::Int32(x) => {
1715            let offset = row * 4;
1716            if offset + 4 <= v.data().len() {
1717                v.data_mut()[offset..offset + 4].copy_from_slice(&x.to_le_bytes());
1718                v.set_null(row, false);
1719            }
1720        }
1721        Value::Double(x) => {
1722            let offset = row * 8;
1723            if offset + 8 <= v.data().len() {
1724                v.data_mut()[offset..offset + 8].copy_from_slice(&x.to_le_bytes());
1725                v.set_null(row, false);
1726            }
1727        }
1728        Value::Float(x) => {
1729            let offset = row * 4;
1730            if offset + 4 <= v.data().len() {
1731                v.data_mut()[offset..offset + 4].copy_from_slice(&x.to_le_bytes());
1732                v.set_null(row, false);
1733            }
1734        }
1735        Value::String(s) => {
1736            let bytes = s.as_bytes();
1737            if bytes.len() > 255 {
1738                return Err(format!(
1739                    "Cannot store string of {} bytes: inline string storage limit is 255 bytes",
1740                    bytes.len()
1741                ));
1742            }
1743            let offset = row * 256;
1744            if offset < v.data().len() {
1745                v.data_mut()[offset] = bytes.len() as u8;
1746                if offset + 1 + bytes.len() <= v.data().len() {
1747                    v.data_mut()[offset + 1..offset + 1 + bytes.len()].copy_from_slice(bytes);
1748                }
1749                v.set_null(row, false);
1750            }
1751        }
1752        _ => {
1753            // For complex types (List, Struct, etc.), store as null
1754            v.set_null(row, true);
1755        }
1756    }
1757    Ok(())
1758}
1759
1760/// Build an ArrowVector from a Vec<Value>, using typed builders to
1761/// avoid the intermediate ValueVector allocation.
1762pub(crate) fn build_arrow_from_values(
1763    values: &[Value],
1764    phys_type: PhysicalTypeID,
1765    num_rows: usize,
1766) -> Result<ArrowVector, ProcessorError> {
1767    match phys_type {
1768        PhysicalTypeID::Bool => {
1769            let mut builder = arrow::array::BooleanBuilder::with_capacity(num_rows);
1770            for v in values {
1771                match v {
1772                    Value::Null => builder.append_null(),
1773                    Value::Bool(b) => builder.append_value(*b),
1774                    _ => builder.append_null(),
1775                }
1776            }
1777            Ok(ArrowVector::new(Arc::new(builder.finish()), phys_type))
1778        }
1779        PhysicalTypeID::Int64 => {
1780            let mut builder = arrow::array::Int64Builder::with_capacity(num_rows);
1781            for v in values {
1782                match v {
1783                    Value::Null => builder.append_null(),
1784                    Value::Int64(n) => builder.append_value(*n),
1785                    Value::Int32(n) => builder.append_value(*n as i64),
1786                    Value::Date(n) => builder.append_value(n.0 as i64),
1787                    Value::Timestamp(n) | Value::TimestampNs(n) | Value::TimestampMs(n) | Value::TimestampSec(n) => {
1788                        builder.append_value(n.0)
1789                    }
1790                    Value::TimestampTz(n) => builder.append_value(n.0),
1791                    Value::DTime(n) => builder.append_value(*n),
1792                    _ => builder.append_null(),
1793                }
1794            }
1795            Ok(ArrowVector::new(Arc::new(builder.finish()), phys_type))
1796        }
1797        PhysicalTypeID::Int32 => {
1798            let mut builder = arrow::array::Int32Builder::with_capacity(num_rows);
1799            for v in values {
1800                match v {
1801                    Value::Null => builder.append_null(),
1802                    Value::Int32(n) => builder.append_value(*n),
1803                    _ => builder.append_null(),
1804                }
1805            }
1806            Ok(ArrowVector::new(Arc::new(builder.finish()), phys_type))
1807        }
1808        PhysicalTypeID::Double => {
1809            let mut builder = arrow::array::Float64Builder::with_capacity(num_rows);
1810            for v in values {
1811                match v {
1812                    Value::Null => builder.append_null(),
1813                    Value::Double(n) => builder.append_value(*n),
1814                    _ => builder.append_null(),
1815                }
1816            }
1817            Ok(ArrowVector::new(Arc::new(builder.finish()), phys_type))
1818        }
1819        PhysicalTypeID::Float => {
1820            let mut builder = arrow::array::Float32Builder::with_capacity(num_rows);
1821            for v in values {
1822                match v {
1823                    Value::Null => builder.append_null(),
1824                    Value::Float(n) => builder.append_value(*n),
1825                    Value::Double(n) => builder.append_value(*n as f32),
1826                    _ => builder.append_null(),
1827                }
1828            }
1829            Ok(ArrowVector::new(Arc::new(builder.finish()), phys_type))
1830        }
1831        PhysicalTypeID::String => {
1832            let mut builder = arrow::array::StringBuilder::with_capacity(num_rows, num_rows * 16);
1833            for v in values {
1834                match v {
1835                    Value::Null => builder.append_null(),
1836                    Value::String(s) => builder.append_value(s),
1837                    _ => builder.append_null(),
1838                }
1839            }
1840            Ok(ArrowVector::new(Arc::new(builder.finish()), phys_type))
1841        }
1842        PhysicalTypeID::List | PhysicalTypeID::Array | PhysicalTypeID::Struct => {
1843            let array = akar_common::arrow_vector::arrow_array_from_values(values);
1844            Ok(ArrowVector::new(array, phys_type))
1845        }
1846        _ => {
1847            // Unsupported type — fall back to creating an Arrow array with all nulls
1848            let mut builder = arrow::array::Int64Builder::with_capacity(num_rows);
1849            builder.append_nulls(num_rows);
1850            Ok(ArrowVector::new(Arc::new(builder.finish()), PhysicalTypeID::Int64))
1851        }
1852    }
1853}
1854
1855impl ExpressionEvaluator {
1856    /// Evaluate a sequence operation (nextval/currval) using the sequence callback.
1857    /// Extracts the first string argument as the sequence name and delegates to the callback.
1858    fn evaluate_sequence_op(
1859        &self,
1860        name: &str,
1861        func: &ScalarFunction,
1862        arg_vectors: &[ValueVector],
1863        num_rows: usize,
1864    ) -> Result<ValueVector, ProcessorError> {
1865        let is_nextval = match func {
1866            ScalarFunction::SequenceOp { is_nextval } => *is_nextval,
1867            _ => return Err(format!("Internal error: expected SequenceOp for '{}'", name).into()),
1868        };
1869
1870        let seq_fn = self
1871            .sequence_fn
1872            .as_ref()
1873            .ok_or_else(|| format!("No sequence callback configured for '{}'", name))?;
1874
1875        let mut result_vec = ValueVector::new(akar_common::types::PhysicalTypeID::Int64, num_rows);
1876        result_vec.resize(num_rows);
1877
1878        for row in 0..num_rows {
1879            let seq_name = if row < arg_vectors[0].size() && !arg_vectors[0].is_null(row) {
1880                match arg_vectors[0].get_value(row) {
1881                    Some(Value::String(s)) => s,
1882                    _ => return Err("nextval/currval requires a string argument (sequence name)".into()),
1883                }
1884            } else {
1885                result_vec.set_null(row, true);
1886                continue;
1887            };
1888
1889            match seq_fn(&seq_name, is_nextval) {
1890                Ok(val) => {
1891                    store_value_in_vector(&mut result_vec, row, &val)?;
1892                }
1893                Err(e) => {
1894                    result_vec.set_null(row, true);
1895                    if row == 0 {
1896                        return Err(e);
1897                    }
1898                }
1899            }
1900        }
1901
1902        Ok(result_vec)
1903    }
1904}
1905
1906#[cfg(test)]
1907mod tests {
1908    use super::*;
1909    use akar_common::types::PhysicalTypeID;
1910    use akar_common::vector::ValueVector;
1911    use akar_function::registry::FunctionRegistry;
1912    use hashbrown::HashMap;
1913
1914    fn make_registry() -> Arc<Mutex<FunctionRegistry>> {
1915        Arc::new(Mutex::new(FunctionRegistry::new()))
1916    }
1917
1918    fn make_chunk(values: &[i64]) -> DataChunk {
1919        let mut v = ValueVector::new(PhysicalTypeID::Int64, values.len());
1920        v.resize(values.len());
1921        for (i, val) in values.iter().enumerate() {
1922            v.set_i64(i, *val);
1923        }
1924        {
1925            let arrow_fields = vec![akar_common::arrow_vector::ArrowVector::from_legacy(&v).array];
1926            let arrow_field_types = vec![v.physical_type()];
1927            DataChunk::new(arrow_fields, arrow_field_types)
1928        }
1929    }
1930
1931    #[test]
1932    fn test_evaluate_constant_int() {
1933        let eval = ExpressionEvaluator::new(make_registry());
1934        let expr = Expression::Constant(Constant::Integer(42));
1935        let chunk = make_chunk(&[]);
1936        let result = eval.evaluate(&expr, &chunk).unwrap();
1937        assert_eq!(result.size(), 0);
1938    }
1939
1940    #[test]
1941    fn test_evaluate_constant_bool() {
1942        let eval = ExpressionEvaluator::new(make_registry());
1943        let expr = Expression::Constant(Constant::Bool(true));
1944        let chunk = make_chunk(&[1, 2, 3]);
1945        let result = eval.evaluate(&expr, &chunk).unwrap();
1946        assert_eq!(result.size(), 3);
1947        // All rows should be true
1948        for i in 0..3 {
1949            assert!(!result.is_null(i));
1950        }
1951    }
1952
1953    #[test]
1954    fn test_evaluate_variable() {
1955        let eval = ExpressionEvaluator::new(make_registry());
1956        let chunk = make_chunk(&[10, 20, 30]);
1957        let expr = Expression::Variable("0".into());
1958        let result = eval.evaluate(&expr, &chunk).unwrap();
1959        assert_eq!(result.size(), 3);
1960        assert_eq!(result.get_i64(0), Some(10));
1961        assert_eq!(result.get_i64(1), Some(20));
1962        assert_eq!(result.get_i64(2), Some(30));
1963    }
1964
1965    #[test]
1966    fn test_evaluate_binary_equal() {
1967        let eval = ExpressionEvaluator::new(make_registry());
1968        // 0 = 0 → true, 1 = 0 → false, 2 = 0 → false
1969        let left = Box::new(Expression::Variable("0".into()));
1970        let right = Box::new(Expression::Constant(Constant::Integer(0)));
1971        let expr = Expression::BinaryOp(BinaryOp::Equal, left, right);
1972        let chunk = make_chunk(&[0, 1, 2]);
1973        let result = eval.evaluate(&expr, &chunk).unwrap();
1974        assert_eq!(result.size(), 3);
1975        assert_eq!(result.get_value(0), Some(Value::Bool(true)));
1976        assert_eq!(result.get_value(1), Some(Value::Bool(false)));
1977        assert_eq!(result.get_value(2), Some(Value::Bool(false)));
1978    }
1979
1980    #[test]
1981    fn test_evaluate_binary_greater_than() {
1982        let eval = ExpressionEvaluator::new(make_registry());
1983        // 3 > 2 → true, 1 > 2 → false, 5 > 2 → true
1984        let left = Box::new(Expression::Variable("0".into()));
1985        let right = Box::new(Expression::Constant(Constant::Integer(2)));
1986        let expr = Expression::BinaryOp(BinaryOp::GreaterThan, left, right);
1987        let chunk = make_chunk(&[3, 1, 5]);
1988        let result = eval.evaluate(&expr, &chunk).unwrap();
1989        assert_eq!(result.size(), 3);
1990        assert_eq!(result.get_value(0), Some(Value::Bool(true)));
1991        assert_eq!(result.get_value(1), Some(Value::Bool(false)));
1992        assert_eq!(result.get_value(2), Some(Value::Bool(true)));
1993    }
1994
1995    #[test]
1996    fn test_evaluate_binary_and() {
1997        let eval = ExpressionEvaluator::new(make_registry());
1998        // We need two columns: set up chunk with column 0 as bools and column 1 as bools
1999        let mut v0 = ValueVector::new(PhysicalTypeID::Bool, 2);
2000        v0.resize(2);
2001        store_value_in_vector(&mut v0, 0, &Value::Bool(true)).unwrap();
2002        store_value_in_vector(&mut v0, 1, &Value::Bool(false)).unwrap();
2003        let mut v1 = ValueVector::new(PhysicalTypeID::Bool, 2);
2004        v1.resize(2);
2005        store_value_in_vector(&mut v1, 0, &Value::Bool(true)).unwrap();
2006        store_value_in_vector(&mut v1, 1, &Value::Bool(true)).unwrap();
2007        let chunk = {
2008            let arrow_fields = vec![
2009                akar_common::arrow_vector::ArrowVector::from_legacy(&v0).array,
2010                akar_common::arrow_vector::ArrowVector::from_legacy(&v1).array,
2011            ];
2012            let arrow_field_types = vec![v0.physical_type(), v1.physical_type()];
2013            DataChunk::new(arrow_fields, arrow_field_types)
2014        };
2015
2016        // true AND true → true, false AND true → false
2017        let left = Box::new(Expression::Variable("0".into()));
2018        let right = Box::new(Expression::Variable("1".into()));
2019        let expr = Expression::BinaryOp(BinaryOp::And, left, right);
2020        let result = eval.evaluate(&expr, &chunk).unwrap();
2021        assert_eq!(result.size(), 2);
2022        assert_eq!(result.get_value(0), Some(Value::Bool(true)));
2023        assert_eq!(result.get_value(1), Some(Value::Bool(false)));
2024    }
2025
2026    #[test]
2027    fn test_evaluate_function_call_string_length() {
2028        let eval = ExpressionEvaluator::new(make_registry());
2029        let chunk = akar_common::vector::DataChunk::new(vec![], vec![]);
2030        let expr = Expression::FunctionCall(
2031            "length".into(),
2032            vec![Expression::Constant(Constant::String("hello".into()))],
2033        );
2034        let result = eval.evaluate(&expr, &chunk).unwrap();
2035        assert_eq!(result.size(), 0);
2036    }
2037
2038    #[test]
2039    fn test_evaluate_not() {
2040        let eval = ExpressionEvaluator::new(make_registry());
2041        let mut v = ValueVector::new(PhysicalTypeID::Bool, 3);
2042        v.resize(3);
2043        store_value_in_vector(&mut v, 0, &Value::Bool(true)).unwrap();
2044        store_value_in_vector(&mut v, 1, &Value::Bool(false)).unwrap();
2045        store_value_in_vector(&mut v, 2, &Value::Bool(true)).unwrap();
2046        let chunk = {
2047            let arrow_fields = vec![akar_common::arrow_vector::ArrowVector::from_legacy(&v).array];
2048            let arrow_field_types = vec![v.physical_type()];
2049            DataChunk::new(arrow_fields, arrow_field_types)
2050        };
2051        // NOT of column 0 — true→false, false→true, true→false
2052        let expr = Expression::UnaryOp(UnaryOp::Not, Box::new(Expression::Variable("0".into())));
2053        let result = eval.evaluate(&expr, &chunk).unwrap();
2054        assert_eq!(result.size(), 3);
2055        assert_eq!(result.get_value(0), Some(Value::Bool(false)));
2056        assert_eq!(result.get_value(1), Some(Value::Bool(true)));
2057        assert_eq!(result.get_value(2), Some(Value::Bool(false)));
2058    }
2059
2060    #[test]
2061    fn test_sequence_nextval_currval_with_callback() {
2062        let state = Arc::new(Mutex::new(HashMap::new()));
2063        state.lock().unwrap().insert("my_seq".to_string(), 10_i64);
2064
2065        let state_for_fn = state.clone();
2066        let seq_fn: Arc<dyn Fn(&str, bool) -> Result<Value, ProcessorError> + Send + Sync> =
2067            Arc::new(move |seq_name: &str, is_nextval: bool| {
2068                let mut map = state_for_fn.lock().map_err(|e| format!("Lock error: {e}"))?;
2069                let current = map
2070                    .get_mut(seq_name)
2071                    .ok_or_else(|| format!("Sequence '{}' not found", seq_name))?;
2072                if is_nextval {
2073                    let out = *current;
2074                    *current += 2;
2075                    Ok(Value::Int64(out))
2076                } else {
2077                    Ok(Value::Int64(*current))
2078                }
2079            });
2080
2081        let eval = ExpressionEvaluator::new(make_registry()).with_sequence_fn(seq_fn);
2082        let chunk = make_chunk(&[1, 2, 3]);
2083
2084        let nextval_expr = Expression::FunctionCall(
2085            "nextval".into(),
2086            vec![Expression::Constant(Constant::String("my_seq".into()))],
2087        );
2088        let nextvals = eval.evaluate(&nextval_expr, &chunk).unwrap();
2089        assert_eq!(nextvals.get_value(0), Some(Value::Int64(10)));
2090        assert_eq!(nextvals.get_value(1), Some(Value::Int64(12)));
2091        assert_eq!(nextvals.get_value(2), Some(Value::Int64(14)));
2092
2093        let currval_expr = Expression::FunctionCall(
2094            "currval".into(),
2095            vec![Expression::Constant(Constant::String("my_seq".into()))],
2096        );
2097        let curr = eval.evaluate(&currval_expr, &make_chunk(&[1])).unwrap();
2098        assert_eq!(curr.get_value(0), Some(Value::Int64(16)));
2099    }
2100
2101    #[test]
2102    fn test_sequence_requires_callback() {
2103        let eval = ExpressionEvaluator::new(make_registry());
2104        let expr = Expression::FunctionCall(
2105            "nextval".into(),
2106            vec![Expression::Constant(Constant::String("my_seq".into()))],
2107        );
2108        let err = eval.evaluate(&expr, &make_chunk(&[1])).unwrap_err();
2109        assert!(
2110            err.to_string().contains("No sequence callback configured"),
2111            "Unexpected error: {err}"
2112        );
2113    }
2114
2115    #[test]
2116    fn test_sequence_requires_string_arg() {
2117        let seq_fn: Arc<dyn Fn(&str, bool) -> Result<Value, ProcessorError> + Send + Sync> =
2118            Arc::new(|_seq_name: &str, _is_nextval: bool| Ok(Value::Int64(1)));
2119        let eval = ExpressionEvaluator::new(make_registry()).with_sequence_fn(seq_fn);
2120
2121        let expr = Expression::FunctionCall("nextval".into(), vec![Expression::Constant(Constant::Integer(42))]);
2122        let err = eval.evaluate(&expr, &make_chunk(&[1])).unwrap_err();
2123        assert!(
2124            err.to_string().contains("requires a string argument"),
2125            "Unexpected error: {err}"
2126        );
2127    }
2128}