Skip to content

Commit bbf67d9

Browse files
gstvgrluvatonmartin-gLiaCastaneda
authored
Add lambda support and array_transform udf (#21679)
This a clean version of #18921 to make it easier to review **this is a breaking change due to adding variant to `Expr` enum, new methods on traits `Session`, `FunctionRegistry` and `ContextProvider` and a new arg on `TaskContext::new`** This PR adds support for lambdas and the `array_transform` function used to test the lambda implementation. Example usage: ```sql SELECT array_transform([2, 3], v -> v != 2); [false, true] -- arbitrally nested lambdas are also supported SELECT array_transform([[[2, 3]]], m -> array_transform(m, l -> array_transform(l, v -> v*2))); [[[4, 6]]] ``` Note: column capture has been removed for now and will be added on a follow on PR, see #21172 Some comments on code snippets of this doc show what value each struct, variant or field would hold after planning the first example above. Some literals are simplified pseudo code 3 new `Expr` variants are added, `HigherOrderFunction`, owing a new trait `HigherOrderUDF`, which is like a `ScalarFunction`/`ScalarUDFImpl` with support for lambdas, `Lambda`, for the lambda body and it's parameters names, and `LambdaVariable`, which is like `Column` but for lambdas parameters. Their logical representations: ```rust enum Expr { // array_transform([2, 3], v -> v != 2) HigherOrderFunction(HigherOrderFunction), // v -> v != 2 Lambda(Lambda), // v, of the lambda body: v != 2 LambdaVariable(LambdaVariable), ... } // array_transform([2, 3], v -> v != 2) struct HigherOrderFunction { // global instance of array_transform pub func: Arc<dyn HigherOrderUDF>, // [Expr::ScalarValue([2, 3]), Expr::Lambda(v -> v != 2)] pub args: Vec<Expr>, } // v -> v != 2 struct Lambda { // ["v"] pub params: Vec<String>, // v != 2 pub body: Box<Expr>, } // v, of the lambda body: v != 2 struct LambdaVariable { // "v" pub name: String, // Field::new("", DataType::Int32, false) // Note: a follow on PR will make this field optional // to free expr_api from specifying it beforehand, // and add resolve_lambda_variables method to Expr, // similar to Expr::Placeholder, see #21172 pub field: FieldRef, pub spans: Spans, } ``` The example would be planned into a tree like this: ``` HigherOrderFunctionExpression name: array_transform children: 1. ListExpression [2,3] 2. LambdaExpression parameters: ["v"] body: BinaryExpression (!=) left: LambdaVariableExpression("v", Field::new("", Int32, false)) right: LiteralExpression("2") ``` The physical counterparts definition: ```rust struct HigherOrderFunctionExpr { // global instance of array_transform fun: Arc<dyn HigherOrderUDF>, // "array_transform" name: String, // [LiteralExpr([2, 3], LambdaExpr("v -> v != 2"))] args: Vec<Arc<dyn PhysicalExpr>>, // [1], the positions at args that contains lambdas lambda_positions: Vec<usize>, // Field::new("", DataType::new_list(DataType::Boolean, false), false) return_field: FieldRef, config_options: Arc<ConfigOptions>, } struct LambdaExpr { // ["v"] params: Vec<String>, // v -> v != 2 body: Arc<dyn PhysicalExpr>, } struct LambdaVariable { // Field::new("v", DataType::Int32, false) field: FieldRef, // 0, the first and only parameter, "v" index: usize, } ``` Note: For those who primarly wants to check if this lambda implementation supports their usecase and don't want to spend much time here, it's okay to skip most collapsed blocks, as those serve mostly to help code reviewers, with the exception of `HigherOrderUDF` and the `array_transform` implementation of `HigherOrderUDF` relevant methods, collapsed due to their size The added `HigherOrderUDF` trait is almost a clone of `ScalarUDFImpl`, with the exception of: 1. `return_field_from_args` and `invoke_with_args`, where now `args.args` is a list of enums with two variants: `Value` or `Lambda` instead of a list of values 2. the addition of `lambda_parameters`, which return a `Field` for each parameter supported for every lambda argument based on the `Field` of the non lambda arguments 3. the removal of `return_field` and the deprecated ones `is_nullable` and `display_name`. 4. Not yet includes analogues to the methods preimage, placement, evaluate_bounds, propagate_constraints, output_ordering and preserves_lex_ordering <details><summary>HigherOrderUDF</summary> ```rust trait HigherOrderUDF { /// Return the field of all the parameters supported by all the supported lambdas of this function /// based on the field of the value arguments. If a lambda support multiple parameters, or if multiple /// lambdas are supported and some are optional, all should be returned, /// regardless of whether they are used on a particular invocation /// /// Tip: If you have a [`HigherOrderFunction`] invocation, you can call the helper /// [`HigherOrderFunction::lambda_parameters`] instead of this method directly /// /// [`HigherOrderFunction`]: crate::expr::HigherOrderFunction /// [`HigherOrderFunction::lambda_parameters`]: crate::expr::HigherOrderFunction::lambda_parameters /// /// Example for array_transform: /// /// `array_transform([2.0, 8.0], v -> v > 4.0)` /// /// ```ignore /// let lambda_parameters = array_transform.lambda_parameters(&[ /// Arc::new(Field::new("", DataType::new_list(DataType::Float32, false))), // the Field of the literal `[2, 8]` /// ])?; /// /// assert_eq!( /// lambda_parameters, /// vec![ /// // the lambda supported parameters, regardless of how many are actually used /// vec![ /// // the value being transformed /// Field::new("", DataType::Float32, false), /// ] /// ] /// ) /// ``` /// /// The implementation can assume that some other part of the code has coerced /// the actual argument types to match [`Self::signature`]. fn lambda_parameters(&self, value_fields: &[FieldRef]) -> Result<Vec<Vec<Field>>>; fn return_field_from_args(&self, args: LambdaReturnFieldArgs) -> Result<FieldRef>; fn invoke_with_args(&self, args: HigherOrderFunctionArgs) -> Result<ColumnarValue>; // ... omitted methods that are similar in ScalarUDFImpl } /// An argument to a lambda function #[derive(Clone, Debug, PartialEq, Eq)] pub enum ValueOrLambda<V, L> { /// A value with associated data Value(V), /// A lambda with associated data Lambda(L), } /// Information about arguments passed to the function /// /// This structure contains metadata about how the function was called /// such as the type of the arguments, any scalar arguments and if the /// arguments can (ever) be null /// /// See [`HigherOrderUDF::return_field_from_args`] for more information #[derive(Clone, Debug)] pub struct LambdaReturnFieldArgs<'a> { /// The data types of the arguments to the function /// /// If argument `i` to the function is a lambda, it will be the field of the result of the /// lambda if evaluated with the parameters returned from [`HigherOrderUDF::lambda_parameters`] /// /// For example, with `array_transform([1], v -> v == 5)` /// this field will be `[ /// ValueOrLambda::Value(Field::new("", DataType::List(DataType::Int32), false)), /// ValueOrLambda::Lambda(Field::new("", DataType::Boolean, false)) /// ]` pub arg_fields: &'a [ValueOrLambda<FieldRef, FieldRef>], /// Is argument `i` to the function a scalar (constant)? /// /// If the argument `i` is not a scalar, it will be None /// /// For example, if a function is called like `array_transform([1], v -> v == 5)` /// this field will be `[Some(ScalarValue::List(...), None]` pub scalar_arguments: &'a [Option<&'a ScalarValue>], } /// Arguments passed to [`HigherOrderUDF::invoke_with_args`] when invoking a /// lambda function. #[derive(Debug, Clone)] pub struct HigherOrderFunctionArgs { /// The evaluated arguments and lambdas to the function pub args: Vec<ValueOrLambda<ColumnarValue, LambdaArgument>>, /// Field associated with each arg, if it exists /// For lambdas, it will be the field of the result of /// the lambda if evaluated with the parameters /// returned from [`HigherOrderUDF::lambda_parameters`] pub arg_fields: Vec<ValueOrLambda<FieldRef, FieldRef>>, /// The number of rows in record batch being evaluated pub number_rows: usize, /// The return field of the lambda function returned /// (from `return_field_from_args`) when creating the /// physical expression from the logical expression pub return_field: FieldRef, /// The config options at execution time pub config_options: Arc<ConfigOptions>, } /// A lambda argument to a HigherOrderFunction #[derive(Clone, Debug)] pub struct LambdaArgument { /// The parameters defined in this lambda /// /// For example, for `array_transform([2], v -> -v)`, /// this will be `vec![Field::new("v", DataType::Int32, true)]` params: Vec<FieldRef>, /// The body of the lambda /// /// For example, for `array_transform([2], v -> -v)`, /// this will be the physical expression of `-v` body: Arc<dyn PhysicalExpr>, } impl LambdaArgument { /// Evaluate this lambda /// `args` should evalute to the value of each parameter /// of the correspondent lambda returned in [HigherOrderUDF::lambda_parameters]. pub fn evaluate( &self, args: &[&dyn Fn() -> Result<ArrayRef>], ) -> Result<ColumnarValue> { let columns = args .iter() .take(self.params.len()) .map(|arg| arg()) .collect::<Result<_>>()?; let schema = Arc::new(Schema::new(self.params.clone())); let batch = RecordBatch::try_new(schema, columns)?; self.body.evaluate(&batch) } } ``` </details> <details><summary>array_transform lambda_parameters implementation</summary> ```rust impl HigherOrderUDF for ArrayTransform { fn lambda_parameters(&self, value_fields: &[FieldRef]) -> Result<Vec<Vec<Field>>> { let list = if value_fields.len() == 1 { &value_fields[0] } else { return plan_err!( "{} function requires 1 value arguments, got {}", self.name(), value_fields.len() ); }; let field = match list.data_type() { DataType::List(field) => field, DataType::LargeList(field) => field, DataType::FixedSizeList(field, _) => field, _ => return plan_err!("expected list, got {list}"), }; // we don't need to check whether the lambda contains more than two parameters, // e.g. array_transform([], (v, i, j) -> v+i+j), as datafusion will do that for us let value = Field::new("", field.data_type().clone(), field.is_nullable()) .with_metadata(field.metadata().clone()); Ok(vec![vec![value]]) } } ``` </details> <details><summary>array_transform return_field_from_args implementation</summary> ```rust fn value_lambda_pair<'a, V: Debug, L: Debug>( name: &str, args: &'a [ValueOrLambda<V, L>], ) -> Result<(&'a V, &'a L)> { let [value, lambda] = take_function_args(name, args)?; let (ValueOrLambda::Value(value), ValueOrLambda::Lambda(lambda)) = (value, lambda) else { return plan_err!( "{name} expects a value followed by a lambda, got {value:?} and {lambda:?}" ); }; Ok((value, lambda)) } impl HigherOrderUDF for ArrayTransform { fn return_field_from_args( &self, args: HigherOrderReturnFieldArgs, ) -> Result<Arc<Field>> { let (list, lambda) = value_lambda_pair(self.name(), args.arg_fields)?; // lambda is the resulting field of executing the lambda body // with the parameters returned in lambda_parameters let field = Arc::new(Field::new( Field::LIST_FIELD_DEFAULT_NAME, lambda.data_type().clone(), lambda.is_nullable(), )); let return_type = match list.data_type() { DataType::List(_) => DataType::List(field), DataType::LargeList(_) => DataType::LargeList(field), DataType::FixedSizeList(_, size) => DataType::FixedSizeList(field, *size), other => plan_err!("expected list, got {other}")?, }; Ok(Arc::new(Field::new("", return_type, list.is_nullable()))) } } ``` </details> <details><summary>array_transform invoke_with_args implementation</summary> ```rust impl HigherOrderUDF for ArrayTransform { fn invoke_with_args(&self, args: HigherOrderFunctionArgs) -> Result<ColumnarValue> { let (list, lambda) = value_lambda_pair(self.name(), &args.args)?; let list_array = list.to_array(args.number_rows)?; // Fast path for fully null input array and also the only way to safely work with // a fully null fixed size list array as it can't be handled by remove_list_null_values below if list_array.null_count() == list_array.len() { return Ok(ColumnarValue::Array(new_null_array( args.return_type(), list_array.len(), ))); } // as per list_values docs, if list_array is sliced, list_values will be sliced too, // so before constructing the transformed array below, we must adjust the list offsets with // adjust_offsets_for_slice let list_values = list_values(&list_array)?; // by passing closures, lambda.evaluate can evaluate only those actually needed let values_param = || Ok(Arc::clone(&list_values)); // call the transforming lambda let transformed_values = lambda .evaluate(&[&values_param])? .into_array(list_values.len())?; let field = match args.return_field.data_type() { DataType::List(field) | DataType::LargeList(field) | DataType::FixedSizeList(field, _) => Arc::clone(field), _ => { return exec_err!( "{} expected ScalarFunctionArgs.return_field to be a list, got {}", self.name(), args.return_field ); } }; let transformed_list = match list_array.data_type() { DataType::List(_) => { let list = list_array.as_list(); // since we called list_values above which would return sliced values for // a sliced list, we must adjust the offsets here as otherwise they would be invalid let adjusted_offsets = adjust_offsets_for_slice(list); Arc::new(ListArray::new( field, adjusted_offsets, transformed_values, list.nulls().cloned(), )) as ArrayRef } DataType::LargeList(_) => { let large_list = list_array.as_list(); // since we called list_values above which would return sliced values for // a sliced list, we must adjust the offsets here as otherwise they would be invalid let adjusted_offsets = adjust_offsets_for_slice(large_list); Arc::new(LargeListArray::new( field, adjusted_offsets, transformed_values, large_list.nulls().cloned(), )) } DataType::FixedSizeList(_, value_length) => { Arc::new(FixedSizeListArray::new( field, *value_length, transformed_values, list_array.as_fixed_size_list().nulls().cloned(), )) } other => exec_err!("expected list, got {other}")?, }; Ok(ColumnarValue::Array(transformed_list)) } } ``` </details> <details><summary>How relevant HigherOrderUDF methods would be called and what they would return during planning and evaluation of the example</summary> ```rust // this is called at sql planning let lambda_parameters = lambda_udf.lambda_parameters(&[ Field::new("", DataType::new_list(DataType::Int32, false), false), // the Field of the [2, 3] literal ])?; assert_eq!( lambda_parameters, vec![ // the parameters that *can* be declared on the lambda, and not only // those actually declared: the implementation doesn't need to care // about it vec![ Field::new("", DataType::Int32, false), // the list inner value ]] ); // this is called every time ExprSchemable is called on a HigherOrderFunction let return_field = array_transform.return_field_from_args(&LambdaReturnFieldArgs { arg_fields: &[ ValueOrLambda::Value(Field::new("", DataType::new_list(DataType::Int32, false), false)), ValueOrLambda::Lambda(Field::new("", DataType::Boolean, false)), // the return_field of the expression "v != 2" when "v" is of the type returned in lambda_parameters ], scalar_arguments // irrelevant })?; assert_eq!(return_field, Field::new("", DataType::new_list(DataType::Boolean, false), false)); let value = array_transform.evaluate(&HigherOrderFunctionArgs { args: vec![ ValueOrLambda::Value(List([2, 3])), ValueOrLambda::Lambda(LambdaArgument of `v -> v != 2`), ], arg_fields, // same as above number_rows: 1, return_field, // same as above config_options, // irrelevant })?; assert_eq!(value, BooleanArray::from([false, true])) ``` </details> <br> <br> A pair HigherOrderUDF/HigherOrderUDFImpl like ScalarFunction was not used because those exist only [to maintain backwards compatibility with the older API](https://docs.rs/datafusion/latest/datafusion/logical_expr/struct.ScalarUDF.html#api-note) #8045 </details> <br> Why `LambdaVariable` and not `Column`: Existing tree traversals that operate on columns would break if some column nodes referenced to a lambda parameter and not a real column. In the example query, projection pushdown would try to push the lambda parameter "v", which won't exist in table "t". Example of code of another traversal that would break: ```rust fn minimize_join_filter(expr: Arc<dyn PhysicalExpr>, ...) -> JoinFilter { let mut used_columns = HashSet::new(); expr.apply(|expr| { if let Some(col) = expr.as_any().downcast_ref::<Column>() { // if this is a lambda column, this function will break used_columns.insert(col.index()); } Ok(TreeNodeRecursion::Continue) }); ... } ``` Furthermore, the implemention of `ExprSchemable` and `PhysicalExpr::return_field` for `Column` expects that the schema it receives as a argument contains an entry for its name, which is not the case for lambda parameters. By including a `FieldRef` on `LambdaVariable` that should be resolved during construction time in the sql planner, `ExprSchemable` and `PhysicalExpr::return_field` simply return it's own Field: <details><summary>LambdaVariable ExprSchemable and PhysicalExpr::return_field implementation </summary> ```rust impl ExprSchemable for Expr { fn to_field( &self, schema: &dyn ExprSchema, ) -> Result<(Option<TableReference>, Arc<Field>)> { let (relation, schema_name) = self.qualified_name(); let field = match self { Expr::LambdaVariable(l) => Ok(Arc::clone(&l.field)), ... }?; Ok(( relation, Arc::new(field.as_ref().clone().with_name(schema_name)), )) } ... } impl PhysicalExpr for LambdaVariable { fn return_field(&self, _input_schema: &Schema) -> Result<FieldRef> { Ok(Arc::clone(&self.field)) } ... } ``` </details> <br> <details><summary>Possible alternatives discarded due to complexity, requiring downstream changes and implementation size:</summary> 1. Add a new set of TreeNode methods that provides the set of lambdas parameters names seen during the traversal, so column nodes can be tested if they refer to a regular column or to a lambda parameter. Any downstream user that wants to support lambdas would need use those methods instead of the existing ones. This also would add 1k+ lines to the PR. ```rust impl Expr { pub fn transform_with_lambdas_params< F: FnMut(Self, &HashSet<String>) -> Result<Transformed<Self>>, >( self, mut f: F, ) -> Result<Transformed<Self>> {} } ``` How minimize_join_filter would looks like: ```rust fn minimize_join_filter(expr: Arc<dyn PhysicalExpr>, ...) -> JoinFilter { let mut used_columns = HashSet::new(); expr.apply_with_lambdas_params(|expr, lambdas_params| { if let Some(col) = expr.as_any().downcast_ref::<Column>() { // dont include lambdas parameters if !lambdas_params.contains(col.name()) { used_columns.insert(col.index()); } } Ok(TreeNodeRecursion::Continue) }) ... } ``` 2. Add a flag to the Column node indicating if it refers to a lambda parameter. Still requires checking for it on existing tree traversals that works on Columns (30+) and also downstream. ```rust //logical struct Column { pub relation: Option<TableReference>, pub name: String, pub spans: Spans, pub is_lambda_parameter: bool, } //physical struct Column { name: String, index: usize, is_lambda_parameter: bool, } ``` How minimize_join_filter would look like: ```rust fn minimize_join_filter(expr: Arc<dyn PhysicalExpr>, ...) -> JoinFilter { let mut used_columns = HashSet::new(); expr.apply(|expr| { if let Some(col) = expr.as_any().downcast_ref::<Column>() { // dont include lambdas parameters if !col.is_lambda_parameter { used_columns.insert(col.index()); } } Ok(TreeNodeRecursion::Continue) }) ... } ``` 1. Add a new set of TreeNode methods that provides a schema that includes the lambdas parameters for the scope of the node being visited/transformed: ```rust impl Expr { pub fn transform_with_schema< F: FnMut(Self, &DFSchema) -> Result<Transformed<Self>>, >( self, schema: &DFSchema, f: F, ) -> Result<Transformed<Self>> { ... } ... other methods } ``` For any given HigherOrderFunction found during the traversal, a new schema is created for each lambda argument that contains it's parameter, returned from HigherOrderUDF::lambda_parameters How it would look like: ```rust pub fn infer_placeholder_types(self, schema: &DFSchema) -> Result<(Expr, bool)> { let mut has_placeholder = false; // Provide the schema as the first argument. // Transforming closure receive an adjusted_schema as argument self.transform_with_schema(schema, |mut expr, adjusted_schema| { match &mut expr { // Default to assuming the arguments are the same type Expr::BinaryExpr(BinaryExpr { left, op: _, right }) => { // use adjusted_schema and not schema. Those expressions may contain // columns referring to a lambda parameter, which Field would only be // available in adjusted_schema and not in schema rewrite_placeholder(left.as_mut(), right.as_ref(), adjusted_schema)?; rewrite_placeholder(right.as_mut(), left.as_ref(), adjusted_schema)?; } .... ``` 2. Make available trought LogicalPlan and ExecutionPlan nodes a schema that includes all lambdas parameters from all expressions owned by the node, and use this schema for tree traversals. For nodes which won't own any expression, the regular schema can be returned ```rust impl LogicalPlan { fn lambda_extended_schema(&self) -> &DFSchema; } trait ExecutionPlan { fn lambda_extended_schema(&self) -> &DFSchema; } //usage impl LogicalPlan { pub fn replace_params_with_values( self, param_values: &ParamValues, ) -> Result<LogicalPlan> { self.transform_up_with_subqueries(|plan| { // use plan.lambda_extended_schema() containing lambdas parameters // instead of plan.schema() which wont let lambda_extended_schema = Arc::clone(plan.lambda_extended_schema()); let name_preserver = NamePreserver::new(&plan); plan.map_expressions(|e| { // if this expression is child of lambda and contain columns referring it's parameters // the lambda_extended_schema already contain them let (e, has_placeholder) = e.infer_placeholder_types(&lambda_extended_schema)?; .... ``` </details> <br> --------- Co-authored-by: Raz Luvaton <16746759+rluvaton@users.noreply.github.com> Co-authored-by: Martin Grigorov <martin-g@users.noreply.github.com> Co-authored-by: Lía Adriana <lia.castaneda@datadoghq.com>
1 parent 54a5515 commit bbf67d9

61 files changed

Lines changed: 4784 additions & 99 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

Cargo.lock

Lines changed: 1 addition & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

datafusion-examples/examples/sql_ops/frontend.rs

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -22,8 +22,8 @@ use datafusion::common::{TableReference, plan_err};
2222
use datafusion::config::ConfigOptions;
2323
use datafusion::error::Result;
2424
use datafusion::logical_expr::{
25-
AggregateUDF, Expr, LogicalPlan, ScalarUDF, TableProviderFilterPushDown, TableSource,
26-
WindowUDF,
25+
AggregateUDF, Expr, HigherOrderUDF, LogicalPlan, ScalarUDF,
26+
TableProviderFilterPushDown, TableSource, WindowUDF,
2727
};
2828
use datafusion::optimizer::{
2929
Analyzer, AnalyzerRule, Optimizer, OptimizerConfig, OptimizerContext, OptimizerRule,
@@ -154,6 +154,10 @@ impl ContextProvider for MyContextProvider {
154154
None
155155
}
156156

157+
fn get_higher_order_meta(&self, _name: &str) -> Option<Arc<dyn HigherOrderUDF>> {
158+
None
159+
}
160+
157161
fn get_aggregate_meta(&self, _name: &str) -> Option<Arc<AggregateUDF>> {
158162
None
159163
}
@@ -174,6 +178,10 @@ impl ContextProvider for MyContextProvider {
174178
Vec::new()
175179
}
176180

181+
fn higher_order_function_names(&self) -> Vec<String> {
182+
Vec::new()
183+
}
184+
177185
fn udaf_names(&self) -> Vec<String> {
178186
Vec::new()
179187
}

datafusion/catalog-listing/src/helpers.rs

Lines changed: 16 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -85,12 +85,26 @@ pub fn expr_applicable_for_cols(col_names: &[&str], expr: &Expr) -> bool {
8585
| Expr::ScalarSubquery(_)
8686
| Expr::SetComparison(_)
8787
| Expr::GroupingSet(_)
88-
| Expr::Case(_) => Ok(TreeNodeRecursion::Continue),
88+
| Expr::Case(_)
89+
| Expr::Lambda(_)
90+
| Expr::LambdaVariable(_) => Ok(TreeNodeRecursion::Continue),
8991

9092
Expr::ScalarFunction(scalar_function) => {
9193
match scalar_function.func.signature().volatility {
9294
Volatility::Immutable => Ok(TreeNodeRecursion::Continue),
9395
// TODO: Stable functions could be `applicable`, but that would require access to the context
96+
// https://github.com/apache/datafusion/issues/21690
97+
Volatility::Stable | Volatility::Volatile => {
98+
is_applicable = false;
99+
Ok(TreeNodeRecursion::Stop)
100+
}
101+
}
102+
}
103+
Expr::HigherOrderFunction(hof) => {
104+
match hof.func.signature().volatility {
105+
Volatility::Immutable => Ok(TreeNodeRecursion::Continue),
106+
// TODO: Stable functions could be `applicable`, but that would require access to the context
107+
// https://github.com/apache/datafusion/issues/21690
94108
Volatility::Stable | Volatility::Volatile => {
95109
is_applicable = false;
96110
Ok(TreeNodeRecursion::Stop)
@@ -102,6 +116,7 @@ pub fn expr_applicable_for_cols(col_names: &[&str], expr: &Expr) -> bool {
102116
// - AGGREGATE and WINDOW should not end up in filter conditions, except maybe in some edge cases
103117
// - Can `Wildcard` be considered as a `Literal`?
104118
// - ScalarVariable could be `applicable`, but that would require access to the context
119+
// https://github.com/apache/datafusion/issues/21690
105120
// TODO: remove the next line after `Expr::Wildcard` is removed
106121
#[expect(deprecated)]
107122
Expr::AggregateFunction { .. }

datafusion/common/src/utils/mod.rs

Lines changed: 258 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -24,14 +24,19 @@ pub mod proxy;
2424
pub mod string_utils;
2525

2626
use crate::assert_or_internal_err;
27-
use crate::error::{_exec_datafusion_err, _internal_datafusion_err};
27+
use crate::error::{_exec_datafusion_err, _exec_err, _internal_datafusion_err};
2828
use crate::{Result, ScalarValue};
2929
use arrow::array::{
3030
Array, ArrayRef, FixedSizeListArray, LargeListArray, ListArray, OffsetSizeTrait,
3131
cast::AsArray,
3232
};
33+
use arrow::array::{
34+
Datum, GenericListArray, Int32Array, Int64Array, MutableArrayData, make_array,
35+
};
3336
use arrow::array::{LargeListViewArray, ListViewArray};
3437
use arrow::buffer::{OffsetBuffer, ScalarBuffer};
38+
use arrow::compute::kernels::cmp::neq;
39+
use arrow::compute::kernels::length::length;
3540
use arrow::compute::{SortColumn, SortOptions, partition};
3641
use arrow::datatypes::{DataType, Field, SchemaRef};
3742
#[cfg(feature = "sql")]
@@ -1005,11 +1010,137 @@ pub fn take_function_args<const N: usize, T>(
10051010
})
10061011
}
10071012

1013+
/// Returns the inner values of a list, or an error otherwise
1014+
/// For [`ListArray`] and [`LargeListArray`], if it's sliced, it returns a
1015+
/// sliced array too. Therefore, too reconstruct a list using it,
1016+
/// you must adjust the offsets using [`adjust_offsets_for_slice`]
1017+
pub fn list_values(array: &dyn Array) -> Result<ArrayRef> {
1018+
match array.data_type() {
1019+
DataType::List(_) => Ok(sliced_list_values(array.as_list::<i32>())),
1020+
DataType::LargeList(_) => Ok(sliced_list_values(array.as_list::<i64>())),
1021+
DataType::FixedSizeList(_, _) => {
1022+
Ok(Arc::clone(array.as_fixed_size_list().values()))
1023+
}
1024+
other => _exec_err!("expected list, got {other}"),
1025+
}
1026+
}
1027+
1028+
fn sliced_list_values<O: OffsetSizeTrait>(list: &GenericListArray<O>) -> ArrayRef {
1029+
let values = list.values();
1030+
let offsets = list.offsets();
1031+
1032+
if let (Some(first), Some(last)) = (offsets.first(), offsets.last()) {
1033+
let first = first.as_usize();
1034+
let last = last.as_usize();
1035+
1036+
if first != 0 || last != values.len() {
1037+
return values.slice(first, last - first);
1038+
}
1039+
}
1040+
1041+
Arc::clone(values)
1042+
}
1043+
1044+
/// If `list` is sliced, returns an adjusted offset buffer so that
1045+
/// it points to the sliced portion of the list values, and not the whole list values
1046+
pub fn adjust_offsets_for_slice<O: OffsetSizeTrait>(
1047+
list: &GenericListArray<O>,
1048+
) -> OffsetBuffer<O> {
1049+
let offsets = list.offsets();
1050+
1051+
if let (Some(first), Some(last)) = (offsets.first(), offsets.last())
1052+
&& (!first.is_zero() || last.as_usize() != list.values().len())
1053+
{
1054+
let offsets = offsets.iter().map(|offset| *offset - *first).collect();
1055+
1056+
//todo: use unsafe Offset::new_unchecked?
1057+
return OffsetBuffer::new(offsets);
1058+
}
1059+
1060+
offsets.clone()
1061+
}
1062+
1063+
/// For lists and large lists, truncates the sublist of null values
1064+
/// Otherwise returns an error
1065+
pub fn remove_list_null_values(array: &ArrayRef) -> Result<ArrayRef> {
1066+
// todo: handle list view and map
1067+
match array.data_type() {
1068+
DataType::List(_) => Ok(Arc::new(truncate_list_nulls(array.as_list::<i32>())?)),
1069+
DataType::LargeList(_) => {
1070+
Ok(Arc::new(truncate_list_nulls(array.as_list::<i64>())?))
1071+
}
1072+
dt => _exec_err!("expected List or LargeList, got {dt}"),
1073+
}
1074+
}
1075+
1076+
fn truncate_list_nulls<O: OffsetSizeTrait>(
1077+
list: &GenericListArray<O>,
1078+
) -> Result<GenericListArray<O>> {
1079+
if let Some(nulls) = list.nulls()
1080+
&& nulls.null_count() > 0
1081+
{
1082+
let lengths = length(list)?;
1083+
let zero: &dyn Datum = if lengths.data_type() == &DataType::Int32 {
1084+
&Int32Array::new_scalar(0)
1085+
} else {
1086+
&Int64Array::new_scalar(0)
1087+
};
1088+
1089+
let not_empty = neq(&lengths, zero)?;
1090+
let null_and_non_empty = &!nulls.inner() & not_empty.values();
1091+
1092+
if null_and_non_empty.count_set_bits() > 0 {
1093+
let array_data = list.values().to_data();
1094+
let offsets = list.offsets();
1095+
let capacity = offsets[offsets.len() - 1] - offsets[0];
1096+
let mut mutable_array_data =
1097+
MutableArrayData::new(vec![&array_data], false, capacity.as_usize());
1098+
1099+
let valid_or_empty = nulls.inner() | &!not_empty.values();
1100+
1101+
for (start, end) in valid_or_empty.set_slices() {
1102+
mutable_array_data.extend(
1103+
0,
1104+
offsets[start].as_usize(),
1105+
offsets[end].as_usize(),
1106+
);
1107+
}
1108+
1109+
let lengths = std::iter::zip(offsets.lengths(), nulls)
1110+
.map(|(length, is_valid)| if is_valid { length } else { 0 });
1111+
1112+
let offsets = OffsetBuffer::from_lengths(lengths);
1113+
let values = make_array(mutable_array_data.freeze());
1114+
1115+
let field = match list.data_type() {
1116+
DataType::List(field) => field,
1117+
DataType::LargeList(field) => field,
1118+
_ => unreachable!(),
1119+
};
1120+
1121+
return Ok(GenericListArray::try_new(
1122+
Arc::clone(field),
1123+
offsets,
1124+
values,
1125+
list.nulls().cloned(),
1126+
)?);
1127+
}
1128+
}
1129+
Ok(list.clone())
1130+
}
1131+
10081132
#[cfg(test)]
10091133
mod tests {
1134+
use std::sync::Arc;
1135+
10101136
use super::*;
10111137
use crate::ScalarValue::Null;
1012-
use arrow::array::Float64Array;
1138+
use arrow::{
1139+
array::{Float64Array, Int32Array},
1140+
buffer::NullBuffer,
1141+
datatypes::Int32Type,
1142+
};
1143+
use sqlparser::ast::Ident;
10131144

10141145
#[test]
10151146
fn test_bisect_linear_left_and_right() -> Result<()> {
@@ -1309,4 +1440,129 @@ mod tests {
13091440
assert_eq!(expected, transposed);
13101441
Ok(())
13111442
}
1443+
1444+
#[test]
1445+
fn test_sliced_list_values() {
1446+
let data = vec![
1447+
Some(vec![Some(0), Some(1), Some(2)]),
1448+
None,
1449+
Some(vec![Some(3), None, Some(5)]),
1450+
Some(vec![Some(6), Some(7)]),
1451+
];
1452+
1453+
let list = ListArray::from_iter_primitive::<Int32Type, _, _>(data);
1454+
1455+
assert_eq!(
1456+
sliced_list_values(&list).as_primitive(),
1457+
&Int32Array::from(vec![
1458+
Some(0),
1459+
Some(1),
1460+
Some(2),
1461+
Some(3),
1462+
None,
1463+
Some(5),
1464+
Some(6),
1465+
Some(7)
1466+
])
1467+
);
1468+
1469+
assert_eq!(
1470+
sliced_list_values(&list.slice(0, 1)).as_primitive(),
1471+
&Int32Array::from(vec![Some(0), Some(1), Some(2)])
1472+
);
1473+
1474+
assert_eq!(
1475+
sliced_list_values(&list.slice(2, 1)).as_primitive(),
1476+
&Int32Array::from(vec![Some(3), None, Some(5)])
1477+
);
1478+
1479+
assert_eq!(
1480+
sliced_list_values(&list.slice(3, 1)).as_primitive(),
1481+
&Int32Array::from(vec![Some(6), Some(7)])
1482+
);
1483+
1484+
assert!(sliced_list_values(&list.slice(0, 0)).is_empty());
1485+
assert!(sliced_list_values(&list.slice(1, 0)).is_empty());
1486+
assert!(sliced_list_values(&list.slice(3, 0)).is_empty());
1487+
}
1488+
1489+
#[test]
1490+
fn test_adjust_offsets() {
1491+
let data = vec![
1492+
Some(vec![Some(0), Some(1), Some(2)]),
1493+
None,
1494+
Some(vec![Some(3), None, Some(5)]),
1495+
Some(vec![Some(6), Some(7)]),
1496+
];
1497+
let list = ListArray::from_iter_primitive::<Int32Type, _, _>(data);
1498+
1499+
assert_eq!(
1500+
adjust_offsets_for_slice(&list),
1501+
OffsetBuffer::from_lengths([3, 0, 3, 2])
1502+
);
1503+
1504+
assert_eq!(
1505+
adjust_offsets_for_slice(&list.slice(0, 1)),
1506+
OffsetBuffer::from_lengths([3])
1507+
);
1508+
1509+
assert_eq!(
1510+
adjust_offsets_for_slice(&list.slice(1, 2)),
1511+
OffsetBuffer::from_lengths([0, 3])
1512+
);
1513+
1514+
assert_eq!(
1515+
adjust_offsets_for_slice(&list.slice(1, 3)),
1516+
OffsetBuffer::from_lengths([0, 3, 2])
1517+
);
1518+
1519+
assert_eq!(
1520+
adjust_offsets_for_slice(&list.slice(0, 0)),
1521+
OffsetBuffer::from_lengths([])
1522+
);
1523+
1524+
assert_eq!(
1525+
adjust_offsets_for_slice(&list.slice(1, 0)),
1526+
OffsetBuffer::from_lengths([])
1527+
);
1528+
1529+
assert_eq!(
1530+
adjust_offsets_for_slice(&list.slice(3, 0)),
1531+
OffsetBuffer::from_lengths([])
1532+
);
1533+
}
1534+
1535+
fn create_i32_list(
1536+
values: impl Into<Int32Array>,
1537+
offsets: OffsetBuffer<i32>,
1538+
nulls: Option<NullBuffer>,
1539+
) -> ListArray {
1540+
let list_field = Arc::new(Field::new_list_field(DataType::Int32, true));
1541+
1542+
ListArray::new(list_field, offsets, Arc::new(values.into()), nulls)
1543+
}
1544+
1545+
#[test]
1546+
fn test_remove_list_null_values_list() {
1547+
let list = Arc::new(create_i32_list(
1548+
vec![100, 20, 10, 0, 0, 0, 0, 1, 50],
1549+
OffsetBuffer::<i32>::from_lengths(vec![3, 4, 0, 2, 0]),
1550+
Some(NullBuffer::from(vec![true, false, false, true, false])),
1551+
)) as ArrayRef;
1552+
1553+
let res = remove_list_null_values(&list).unwrap();
1554+
let res = res.as_list::<i32>();
1555+
1556+
let expected = Arc::new(create_i32_list(
1557+
vec![100, 20, 10, 1, 50],
1558+
OffsetBuffer::<i32>::from_lengths(vec![3, 0, 0, 2, 0]),
1559+
Some(NullBuffer::from(vec![true, false, false, true, false])),
1560+
)) as ArrayRef;
1561+
let expected = expected.as_list::<i32>();
1562+
1563+
assert_eq!(res, expected);
1564+
// check above skips inner value of nulls
1565+
assert_eq!(res.values(), expected.values());
1566+
assert_eq!(res.offsets(), expected.offsets());
1567+
}
13121568
}

0 commit comments

Comments
 (0)