diff --git a/datafusion/expr/src/higher_order_function.rs b/datafusion/expr/src/higher_order_function.rs index 413714f498164..c300be8f6fcfe 100644 --- a/datafusion/expr/src/higher_order_function.rs +++ b/datafusion/expr/src/higher_order_function.rs @@ -24,7 +24,7 @@ use crate::expr::{ use crate::type_coercion::functions::value_fields_with_higher_order_udf; use crate::udf_eq::UdfEq; use crate::{ColumnarValue, Documentation, Expr, ExprSchemable}; -use arrow::array::{ArrayRef, RecordBatch}; +use arrow::array::{ArrayRef, RecordBatch, RecordBatchOptions}; use arrow::datatypes::{DataType, FieldRef, Schema}; use arrow_schema::SchemaRef; use datafusion_common::config::ConfigOptions; @@ -239,6 +239,26 @@ pub struct LambdaArgument { /// For example, for `array_transform([2], v -> -v)`, /// this will be `vec![Field::new("v", DataType::Int32, true)]` params: Vec, + /// Indices into [`Self::params`] of the parameters that are actually + /// referenced by [`Self::body`] (taking nested-lambda shadowing into + /// account), in the original declaration order of `params`. + /// + /// [`Self::evaluate`] only evaluates and pushes the closures whose + /// corresponding parameter index appears here, so unused declared + /// parameters leave no slot in the merged batch and the body's compressed + /// column indices line up directly with what the evaluator built. + /// + /// Callers who already have a `LambdaExpr` should pass + /// `LambdaExpr::used_param_indices()` directly to [`Self::new`] — both + /// are indices into the same positionally-aligned `params` list. + /// + /// Every index here must be `< params.len()`; see the precondition on + /// [`Self::new`]. + /// + /// Relies on captures sorting before this lambda's own params in the + /// planner's (un-projected) index space, which is what makes + /// `captures ++ used_params` below line up with the projected body. + used_param_indices: Vec, /// The body of the lambda /// /// For example, for `array_transform([2], v -> -v)`, @@ -257,26 +277,45 @@ pub struct LambdaArgument { } impl LambdaArgument { + /// # Preconditions + /// + /// Every index in `used_param_indices` must be `< params.len()`; + /// violating this panics on out-of-bounds indexing below. Callers should + /// pass `LambdaExpr::used_param_indices()`, which always indexes into the + /// same `params` list, rather than constructing indices by hand. pub fn new( params: Vec, body: Arc, captures: Option, + used_param_indices: &[usize], ) -> Self { - let fields = match &captures { + debug_assert!( + used_param_indices.iter().all(|i| *i < params.len()), + "used_param_indices contains an index out of bounds for params \ + (len {}): {:?}", + params.len(), + used_param_indices + ); + + let used_param_indices = used_param_indices.to_vec(); + let effective_params = used_param_indices.iter().map(|i| Arc::clone(¶ms[*i])); + + let fields: Vec = match &captures { Some(batch) => batch .schema_ref() .fields() .iter() .cloned() - .chain(params.clone()) + .chain(effective_params) .collect(), - None => params.clone(), + None => effective_params.collect(), }; let schema = Arc::new(Schema::new(fields)); Self { params, + used_param_indices, body, schema, captures, @@ -287,6 +326,11 @@ impl LambdaArgument { /// `args` should evaluate to the value of each parameter /// of the correspondent lambda returned in [HigherOrderUDFImpl::lambda_parameters]. /// + /// Only the closures in `args` for parameters the lambda body actually + /// references are called; closures for declared-but-unused parameters + /// are skipped entirely. Callers should not rely on every closure in + /// `args` being invoked. + /// /// `spread_captures` is responsible for transforming the captured column arrays /// so they align with the evaluation batch. Captures are snapshotted from the /// outer batch at construction time, giving one value per outer row, but the @@ -344,6 +388,7 @@ impl LambdaArgument { spread_captures.as_ref(), Arc::clone(&self.schema), &self.params, + &self.used_param_indices, args, )?; @@ -355,6 +400,7 @@ fn merge_captures_with_variables( captures: Option<&RecordBatch>, schema: SchemaRef, params: &[FieldRef], + used_param_indices: &[usize], variables: &[&dyn Fn() -> Result], ) -> Result { if variables.len() < params.len() { @@ -365,23 +411,42 @@ fn merge_captures_with_variables( ); } + let push_param_arrays = |columns: &mut Vec| -> Result<()> { + for &i in used_param_indices { + columns.push(variables[i]()?); + } + Ok(()) + }; + let columns = match captures { Some(captures) => { let mut columns = captures.columns().to_vec(); - - for arg in &variables[..params.len()] { - columns.push(arg()?); - } - + push_param_arrays(&mut columns)?; + columns + } + None => { + let mut columns = Vec::with_capacity(used_param_indices.len()); + push_param_arrays(&mut columns)?; columns } - None => variables - .iter() - .take(params.len()) - .map(|arg| arg()) - .collect::>()?, }; + if columns.is_empty() { + // No columns to derive a row count from, so borrow one variable's + // array length instead (all variables have the same length). + let row_count = variables.first().ok_or_else(|| { + internal_datafusion_err!( + "merge_captures_with_variables: no variables to derive a row count from" + ) + })?()? + .len(); + return Ok(RecordBatch::try_new_with_options( + schema, + vec![], + &RecordBatchOptions::new().with_row_count(Some(row_count)), + )?); + } + Ok(RecordBatch::try_new(schema, columns)?) } @@ -1681,4 +1746,109 @@ mod tests { Some(Arc::new(Field::new(name, dt, nullable))), )) } + + /// A physical expression that reads the column at a fixed index of the + /// batch it is evaluated against, for exercising [`LambdaArgument`] + /// directly without depending on `datafusion-physical-expr`. + #[derive(Debug, Eq, PartialEq, Hash)] + struct ColumnAt(usize); + + impl std::fmt::Display for ColumnAt { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "column_at({})", self.0) + } + } + + impl PhysicalExpr for ColumnAt { + fn evaluate(&self, batch: &RecordBatch) -> Result { + Ok(ColumnarValue::Array(Arc::clone(batch.column(self.0)))) + } + + fn children(&self) -> Vec<&Arc> { + vec![] + } + + fn with_new_children( + self: Arc, + _children: Vec>, + ) -> Result> { + Ok(self) + } + + fn fmt_sql(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{self}") + } + } + + /// `(k, v) -> v` with only `v` used must push `v`'s array, not `k`'s. + #[test] + fn test_lambda_argument_evaluate_pushes_only_used_param() { + use arrow::array::Int32Array; + + let k_field = Arc::new(Field::new("k", DataType::Int32, true)); + let v_field = Arc::new(Field::new("v", DataType::Int32, true)); + + let body = Arc::new(ColumnAt(0)) as Arc; + let lambda_arg = LambdaArgument::new(vec![k_field, v_field], body, None, &[1]); + + let k_values: ArrayRef = Arc::new(Int32Array::from(vec![100, 200, 300])); + let v_values: ArrayRef = Arc::new(Int32Array::from(vec![1, 2, 3])); + let k_closure = || -> Result { Ok(Arc::clone(&k_values)) }; + let v_closure = || -> Result { Ok(Arc::clone(&v_values)) }; + let args: Vec<&dyn Fn() -> Result> = vec![&k_closure, &v_closure]; + + let result = lambda_arg + .evaluate(&args, |arrays| Ok(arrays.to_vec())) + .unwrap(); + let ColumnarValue::Array(result) = result else { + unreachable!() + }; + + assert_eq!( + result.as_any().downcast_ref::().unwrap(), + &Int32Array::from(vec![1, 2, 3]), + "body should read v's values, not k's" + ); + } + + /// Same as above, but with a capture occupying the leading slot. + #[test] + fn test_lambda_argument_evaluate_pushes_only_used_param_with_captures() { + use arrow::array::Int32Array; + + let cap_field = Arc::new(Field::new("cap", DataType::Int32, true)); + let k_field = Arc::new(Field::new("k", DataType::Int32, true)); + let v_field = Arc::new(Field::new("v", DataType::Int32, true)); + + let body = Arc::new(ColumnAt(1)) as Arc; + + let cap_values: ArrayRef = Arc::new(Int32Array::from(vec![9, 9, 9])); + let captures = RecordBatch::try_new( + Arc::new(Schema::new(vec![cap_field])), + vec![cap_values], + ) + .unwrap(); + + let lambda_arg = + LambdaArgument::new(vec![k_field, v_field], body, Some(captures), &[1]); + + let k_values: ArrayRef = Arc::new(Int32Array::from(vec![100, 200, 300])); + let v_values: ArrayRef = Arc::new(Int32Array::from(vec![1, 2, 3])); + let k_closure = || -> Result { Ok(Arc::clone(&k_values)) }; + let v_closure = || -> Result { Ok(Arc::clone(&v_values)) }; + let args: Vec<&dyn Fn() -> Result> = vec![&k_closure, &v_closure]; + + let result = lambda_arg + .evaluate(&args, |arrays| Ok(arrays.to_vec())) + .unwrap(); + let ColumnarValue::Array(result) = result else { + unreachable!() + }; + + assert_eq!( + result.as_any().downcast_ref::().unwrap(), + &Int32Array::from(vec![1, 2, 3]), + "body should read v's values, not k's or the capture's" + ); + } } diff --git a/datafusion/physical-expr/src/expressions/lambda.rs b/datafusion/physical-expr/src/expressions/lambda.rs index cab2eea64fcf4..95bb5db0b328e 100644 --- a/datafusion/physical-expr/src/expressions/lambda.rs +++ b/datafusion/physical-expr/src/expressions/lambda.rs @@ -31,7 +31,7 @@ use arrow::{ }; use datafusion_common::{ HashMap, plan_err, - tree_node::{Transformed, TreeNode, TreeNodeRecursion}, + tree_node::{Transformed, TreeNode, TreeNodeRecursion, TreeNodeVisitor}, }; use datafusion_common::{HashSet, Result, internal_err}; use datafusion_expr::ColumnarValue; @@ -43,6 +43,7 @@ pub struct LambdaExpr { body: Arc, projected_body: Arc, projection: Vec, + used_param_indices: Vec, } // Manually derive PartialEq and Hash to work around https://github.com/rust-lang/rust/issues/78808 [https://github.com/apache/datafusion/issues/13196] @@ -60,7 +61,7 @@ impl Hash for LambdaExpr { } impl LambdaExpr { - /// Create a new lambda expression with the given parameters and body + /// Create a new lambda expression with the given parameters and body. pub fn try_new(params: Vec, body: Arc) -> Result { if !all_unique(¶ms) { return plan_err!( @@ -75,27 +76,30 @@ impl LambdaExpr { } fn new(params: Vec, body: Arc) -> Self { - let mut used_column_indices = HashSet::new(); + let own_params: HashSet = params.iter().cloned().collect(); - body.apply(|node| { - if let Some(col) = node.downcast_ref::() { - used_column_indices.insert(col.index()); - } else if let Some(var) = node.downcast_ref::() { - used_column_indices.insert(var.index()); - } - - Ok(TreeNodeRecursion::Continue) - }) - .expect("closure should be infallible"); + let mut visitor = CollectUsedVisitor { + own_params: &own_params, + used_indices: HashSet::new(), + used_param_names: HashSet::new(), + shadow_stack: Vec::new(), + }; + body.visit(&mut visitor).expect("visitor is infallible"); + let CollectUsedVisitor { + used_indices, + used_param_names, + .. + } = visitor; - let mut projection = used_column_indices.into_iter().collect::>(); + let mut projection = used_indices.into_iter().collect::>(); projection.sort(); let column_index_map = projection .iter() + .copied() .enumerate() - .map(|(projected, original)| (*original, projected)) + .map(|(new_idx, original)| (original, new_idx)) .collect::>(); let projected_body = Arc::clone(&body) @@ -124,11 +128,19 @@ impl LambdaExpr { .expect("closure should be infallible") .data; + let used_param_indices = params + .iter() + .enumerate() + .filter(|(_, name)| used_param_names.contains(*name)) + .map(|(i, _)| i) + .collect(); + Self { params, body, projected_body, projection, + used_param_indices, } } @@ -170,6 +182,75 @@ impl LambdaExpr { pub(crate) fn projected_body(&self) -> &Arc { &self.projected_body } + + /// Indices into [`params`](Self::params) of the parameters the body + /// actually references, in declaration order. See `CollectUsedVisitor` + /// in this module. + /// + /// Relies on the planner appending each lambda's own params after + /// captures, matching the `captures ++ used_params` layout + /// `LambdaArgument::new` builds. + pub fn used_param_indices(&self) -> &[usize] { + &self.used_param_indices + } +} + +/// Walks the body of a [`LambdaExpr`] and collects, on a single pass: +/// +/// * `used_indices` — every `Column` / `LambdaVariable` index referenced +/// anywhere in the tree (including inside nested lambdas). This drives +/// the `projection` used to slice the outer batch. +/// * `used_param_names` — the subset of *this* lambda's `own_params` that +/// the body actually references. +/// +/// A nested lambda can declare its own parameter with the same name as +/// one of `own_params` — a distinct variable that happens to reuse the +/// name (variable shadowing). E.g. in +/// `(k, v) -> func(col, (k, v2) -> k + v2 + v)`, the inner `k` is not +/// `own_params`' `k`; only `v` should flow up as used, not `k`. +/// +/// `shadow_stack` holds one frame per nested `LambdaExpr` currently being +/// visited, each frame being that lambda's own parameter names. A +/// `LambdaVariable` only counts toward `used_param_names` if its name +/// isn't in any active frame (i.e. not shadowed). +/// +/// The stack is maintained via `TreeNodeVisitor`'s `f_down` / `f_up`: +/// push a frame when entering a nested [`LambdaExpr`], pop it when leaving. +struct CollectUsedVisitor<'a> { + own_params: &'a HashSet, + used_indices: HashSet, + used_param_names: HashSet, + shadow_stack: Vec>, +} + +impl TreeNodeVisitor<'_> for CollectUsedVisitor<'_> { + type Node = Arc; + + fn f_down(&mut self, node: &Self::Node) -> Result { + if let Some(col) = node.downcast_ref::() { + self.used_indices.insert(col.index()); + } else if let Some(var) = node.downcast_ref::() { + self.used_indices.insert(var.index()); + + let name = var.name(); + let shadowed = self.shadow_stack.iter().any(|frame| frame.contains(name)); + if !shadowed && self.own_params.contains(name) { + self.used_param_names.insert(name.to_string()); + } + } else if let Some(nested) = node.downcast_ref::() { + self.shadow_stack + .push(nested.params.iter().cloned().collect()); + } + + Ok(TreeNodeRecursion::Continue) + } + + fn f_up(&mut self, node: &Self::Node) -> Result { + if node.downcast_ref::().is_some() { + self.shadow_stack.pop(); + } + Ok(TreeNodeRecursion::Continue) + } } impl std::fmt::Display for LambdaExpr { @@ -234,7 +315,7 @@ impl PhysicalExpr for LambdaExpr { } } -/// Create a lambda expression +/// Create a lambda expression. pub fn lambda( params: impl IntoIterator>, body: Arc, @@ -273,10 +354,15 @@ fn check_async_udf(body: &Arc) -> Result<()> { #[cfg(test)] mod tests { - use crate::expressions::{NoOp, lambda::lambda}; - use arrow::{array::RecordBatch, datatypes::Schema}; + use crate::expressions::{Column, LambdaVariable, NoOp, lambda::lambda}; + use arrow::{ + array::RecordBatch, + datatypes::{DataType, Field, Schema}, + }; use std::sync::Arc; + use super::LambdaExpr; + #[test] fn test_lambda_evaluate() { let lambda = lambda(["a"], Arc::new(NoOp::new())).unwrap(); @@ -288,4 +374,125 @@ mod tests { fn test_lambda_duplicate_name() { assert!(lambda(["a", "a"], Arc::new(NoOp::new())).is_err()); } + + /// A two-parameter lambda whose body only references the second + /// parameter (`v`) must report only `v` as used. The higher-order + /// function uses this set to push only `v` into the merged batch, so + /// the body's compressed `LambdaVariable` index for `v` lines up with + /// the batch layout. + #[test] + fn test_used_params_collects_only_referenced_param() { + let v_field = Arc::new(Field::new("v", DataType::Int32, true)); + let body = Arc::new(LambdaVariable::new(1, Arc::clone(&v_field))); + + let lambda = + LambdaExpr::try_new(vec!["k".to_string(), "v".to_string()], body).unwrap(); + + assert_eq!(lambda.projection(), &[1]); + assert_eq!(lambda.used_param_indices(), &[1]); + } + + /// A body that references neither declared parameter reports no used params. + #[test] + fn test_used_params_all_unused() { + let body = Arc::new(NoOp::new()); + + let lambda = + LambdaExpr::try_new(vec!["k".to_string(), "v".to_string()], body).unwrap(); + + assert!(lambda.projection().is_empty()); + assert!(lambda.used_param_indices().is_empty()); + } + + /// A three-parameter lambda that skips the middle parameter reports only the ends as used. + #[test] + fn test_used_params_three_params_middle_unused() { + let a_field = Arc::new(Field::new("a", DataType::Int32, true)); + let c_field = Arc::new(Field::new("c", DataType::Int32, true)); + let body = Arc::new(crate::expressions::BinaryExpr::new( + Arc::new(LambdaVariable::new(0, Arc::clone(&a_field))), + datafusion_expr::Operator::Plus, + Arc::new(LambdaVariable::new(2, Arc::clone(&c_field))), + )); + + let lambda = LambdaExpr::try_new( + vec!["a".to_string(), "b".to_string(), "c".to_string()], + body, + ) + .unwrap(); + + assert_eq!(lambda.used_param_indices(), &[0, 2]); + } + + /// Referencing params out of declaration order still reports both as used. + #[test] + fn test_used_params_both_used_in_reverse_reference_order() { + let k_field = Arc::new(Field::new("k", DataType::Int32, true)); + let v_field = Arc::new(Field::new("v", DataType::Int32, true)); + let body = Arc::new(crate::expressions::BinaryExpr::new( + Arc::new(LambdaVariable::new(1, Arc::clone(&v_field))), + datafusion_expr::Operator::Plus, + Arc::new(LambdaVariable::new(0, Arc::clone(&k_field))), + )); + + let lambda = + LambdaExpr::try_new(vec!["k".to_string(), "v".to_string()], body).unwrap(); + + assert_eq!(lambda.projection(), &[0, 1]); + assert_eq!(lambda.used_param_indices(), &[0, 1]); + } + + /// Inside a nested lambda that re-declares one of the outer parameter + /// names, only the non-shadowed outer references should be reported as + /// used by the outer lambda. In + /// `(k, v) -> func(col, (k, v2) -> k + v2 + v)` the inner `k` shadows + /// the outer `k`, so the outer lambda must only see `v` as used. + #[test] + fn test_used_params_handles_shadowing_inside_nested_lambda() { + let outer_k_field = Arc::new(Field::new("k", DataType::Int32, true)); + let outer_v_field = Arc::new(Field::new("v", DataType::Int32, true)); + let inner_v2_field = Arc::new(Field::new("v2", DataType::Int32, true)); + + // Inner lambda body references "k" (inner's), "v2" (inner's), and + // "v" (outer's). Build it directly with the dense compressed + // indices the inner LambdaExpr::new would produce: sorted referenced + // indices, so the names alone matter here — what matters for + // shadow tracking is the names, not the indices. + let inner_body: Arc = + Arc::new(crate::expressions::BinaryExpr::new( + Arc::new(crate::expressions::BinaryExpr::new( + Arc::new(LambdaVariable::new(1, Arc::clone(&outer_k_field))), + datafusion_expr::Operator::Plus, + Arc::new(LambdaVariable::new(2, Arc::clone(&inner_v2_field))), + )), + datafusion_expr::Operator::Plus, + Arc::new(LambdaVariable::new(0, Arc::clone(&outer_v_field))), + )); + let inner_lambda = Arc::new( + LambdaExpr::try_new(vec!["k".to_string(), "v2".to_string()], inner_body) + .unwrap(), + ); + + // Outer body wraps the inner lambda in a binary op next to a + // regular column reference so the walk has something non-trivial + // to descend through. The outer body references the inner lambda + // via `inner_lambda`. + let outer_body: Arc = + Arc::new(crate::expressions::BinaryExpr::new( + Arc::new(Column::new("col", 0)), + datafusion_expr::Operator::Plus, + inner_lambda, + )); + + let outer_lambda = + LambdaExpr::try_new(vec!["k".to_string(), "v".to_string()], outer_body) + .unwrap(); + + assert_eq!( + outer_lambda.used_param_indices(), + &[1], + "only outer's `v` (index 1) should be reported as used; `k` (index 0) is \ + shadowed inside the nested lambda" + ); + } } diff --git a/datafusion/physical-expr/src/higher_order_function.rs b/datafusion/physical-expr/src/higher_order_function.rs index 7390eb33a0922..e28b38bd7c8c1 100644 --- a/datafusion/physical-expr/src/higher_order_function.rs +++ b/datafusion/physical-expr/src/higher_order_function.rs @@ -353,6 +353,7 @@ impl PhysicalExpr for HigherOrderFunctionExpr { } else { Some(batch.project(&projection)?) }, + lambda.used_param_indices(), ))) } ArgSlot::Value => { @@ -509,15 +510,18 @@ mod tests { use super::*; use crate::HigherOrderFunctionExpr; + use crate::create_physical_expr; use crate::expressions::Column; use crate::expressions::NoOp; use crate::expressions::lambda; use crate::expressions::not; - use arrow::array::NullArray; use arrow::array::RecordBatchOptions; + use arrow::array::{ArrayRef, Int32Array}; use arrow::datatypes::{DataType, Field, Schema}; use datafusion_common::Result; use datafusion_common::assert_contains; + use datafusion_expr::execution_props::ExecutionProps; + use datafusion_expr::physical_planning_context::PhysicalPlanningContext; use datafusion_expr::{ HigherOrderFunctionArgs, HigherOrderSignature, HigherOrderUDF, HigherOrderUDFImpl, }; @@ -545,9 +549,11 @@ mod tests { _step: usize, _fields: &[ValueOrLambda>], ) -> Result { - Ok(LambdaParametersProgress::Complete(vec![vec![Arc::new( - Field::new("", DataType::Null, true), - )]])) + // Offer two params; single-param lambdas just ignore the second. + Ok(LambdaParametersProgress::Complete(vec![vec![ + Arc::new(Field::new("", DataType::Int32, true)), + Arc::new(Field::new("", DataType::Int32, true)), + ]])) } fn return_field_from_args( @@ -567,7 +573,18 @@ mod tests { ) -> Result { match &args.args[0] { ValueOrLambda::Lambda(lambda) => lambda.evaluate( - &[&|| Ok(Arc::new(NullArray::new(args.number_rows)))], + &[ + // Sentinel for the first param, distinct from the second's value. + &|| { + Ok(Arc::new(Int32Array::from(vec![-1000; args.number_rows])) + as ArrayRef) + }, + &|| { + Ok(Arc::new(Int32Array::from_iter_values( + (0..args.number_rows as i32).map(|i| 10 * (i + 1)), + )) as ArrayRef) + }, + ], |arrays| Ok(arrays.to_vec()), ), ValueOrLambda::Value(value) => Ok(value.clone()), @@ -715,4 +732,54 @@ mod tests { "mock_function received a lambda via with_new_children at position 0 that wasn't a lambda before" ); } + + /// Exercises the real planner end to end (not hand-picked indices) to + /// check the "captures before own-params" layout invariant. + #[test] + fn test_higher_order_function_two_lambda_params_capture_and_unused_param() { + use datafusion_common::DFSchema; + use datafusion_expr::expr::{HigherOrderFunction, LambdaVariable}; + use datafusion_expr::{Expr, col, lambda as logical_lambda}; + + let fun = Arc::new(HigherOrderUDF::new_from_impl(MockHigherOrderUDF { + signature: HigherOrderSignature::variadic_any(Volatility::Stable), + })); + + // Body uses capture "a" and param "v"; param "k" is left unused. + let v = Expr::LambdaVariable(LambdaVariable::new( + "v".to_string(), + Some(Arc::new(Field::new("v", DataType::Int32, true))), + )); + let body = col("a") + v; + let lambda_expr = logical_lambda(["k", "v"], body); + + let schema = DFSchema::from_unqualified_fields( + vec![Field::new("a", DataType::Int32, false)].into(), + std::collections::HashMap::new(), + ) + .unwrap(); + + let physical_expr = create_physical_expr( + &Expr::HigherOrderFunction(HigherOrderFunction::new(fun, vec![lambda_expr])), + &schema, + &ExecutionProps::new(), + &PhysicalPlanningContext::default(), + ) + .unwrap(); + + let batch = RecordBatch::try_new( + Arc::clone(schema.inner()), + vec![Arc::new(Int32Array::from(vec![1, 2, 3])) as ArrayRef], + ) + .unwrap(); + + let result = physical_expr.evaluate(&batch).unwrap(); + let ColumnarValue::Array(result) = result else { + unreachable!() + }; + + // a + v; k's sentinel (-1000) must not leak into the result. + let expected = Int32Array::from(vec![11, 22, 33]); + assert_eq!(result.as_ref(), &expected); + } }