1use 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
27pub 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
48impl 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 pub subquery_fn: Option<SubqueryFn>,
60 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 pub fn with_subquery_fn(mut self, f: SubqueryFn) -> Self {
76 self.subquery_fn = Some(f);
77 self
78 }
79
80 pub fn with_sequence_fn(mut self, f: SequenceFn) -> Self {
82 self.sequence_fn = Some(f);
83 self
84 }
85
86 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 pub fn evaluate_to_arrow(&self, expr: &Expression, chunk: &DataChunk) -> Result<ArrowVector, ProcessorError> {
99 self.evaluate_arrow(expr, chunk)
100 }
101
102 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 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 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 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 _ => {
175 let legacy = self.evaluate(expr, chunk)?;
176 Ok(ArrowVector::from_legacy(&legacy))
177 }
178 }
179 }
180
181 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 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 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 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 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 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 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 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 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 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 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, ¬_r).map_err(|e| format!("Arrow xor/and failed: {e}"))?;
575 let not_l_and_r = and_kleene(¬_l, r).map_err(|e| format!("Arrow xor/and failed: {e}"))?;
576 Arc::new(or_kleene(&l_and_not_r, ¬_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 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 fn evaluate_arrow_function_call(
622 &self,
623 name: &str,
624 args: &[&Expression],
625 chunk: &DataChunk,
626 ) -> Result<ArrowVector, ProcessorError> {
627 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 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 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 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 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 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 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 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 fn evaluate_property_access(
813 &self,
814 obj: &Expression,
815 prop: &str,
816 chunk: &DataChunk,
817 ) -> Result<ValueVector, ProcessorError> {
818 let qualified_prop = if let Expression::Variable(var_name) = obj {
820 format!("{}.{}", var_name, prop)
821 } else {
822 prop.to_string()
823 };
824
825 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 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 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 self.evaluate(obj, chunk)
894 }
895
896 fn evaluate_function_call(
898 &self,
899 name: &str,
900 args: &[&Expression],
901 chunk: &DataChunk,
902 ) -> Result<ValueVector, ProcessorError> {
903 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 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 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 if matches!(func, ScalarFunction::SequenceOp { .. }) {
940 return self.evaluate_sequence_op(name, &func, &arg_vectors, num_rows);
941 }
942
943 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 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 !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 fn evaluate_binary_op(
1029 &self,
1030 op: &BinaryOp,
1031 left: &Expression,
1032 right: &Expression,
1033 chunk: &DataChunk,
1034 ) -> Result<ValueVector, ProcessorError> {
1035 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 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 self.evaluate_function_call(func_name, &[left, right], chunk)
1066 }
1067
1068 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 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 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 let subject_vec = if let Some(subj) = &case_expr.subject {
1173 Some(self.evaluate(subj, chunk)?)
1174 } else {
1175 None
1176 };
1177
1178 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 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 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 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 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 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 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 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 store_value_in_vector(&mut result, row, &Value::Bool(false))?;
1314 continue;
1315 }
1316 };
1317
1318 let mut true_count = 0u64;
1321 for item in items {
1322 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 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 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 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 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 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 let initial_expr = args
1506 .iter()
1507 .filter(|a| !matches!(a, Expression::Lambda { .. }))
1508 .nth(1) .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 let acc_name = var_name.clone();
1518 let elem_name = match body.as_ref() {
1519 Expression::BinaryOp(_op, left, right) => {
1520 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 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 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
1601fn 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
1659fn 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 v.set_null(row, true);
1755 }
1756 }
1757 Ok(())
1758}
1759
1760pub(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 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 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 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 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 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 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 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 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}