Merge branch 'main' into rm-aggregates-integers

Author: theirix

datafusion/common/src/nested_struct.rs

@@ -4138,20 +4138,16 @@ impl ScalarValue {
 
         let scalar_array = self.to_array()?;
 
-        // For struct types, use name-based casting logic that matches fields by name
-        // and recursively casts nested structs. The field name wrapper is arbitrary
-        // since cast_column only uses the DataType::Struct field definitions inside.
-        let cast_arr = match target_type {
-            DataType::Struct(_) => {
-                // Field name is unused; only the struct's inner field names matter
-                let target_field = Field::new("_", target_type.clone(), true);
-                crate::nested_struct::cast_column(
-                    &scalar_array,
-                    &target_field,
-                    cast_options,
-                )?
-            }
-            _ => cast_with_options(&scalar_array, target_type, cast_options)?,
+        // For types that contain structs (including nested inside Lists, Dictionaries,
+        // etc.), use name-based casting logic that matches struct fields by name and
+        // recursively casts nested structs.
+        let cast_arr = if crate::nested_struct::requires_nested_struct_cast(
+            scalar_array.data_type(),
+            target_type,
+        ) {
+            crate::nested_struct::cast_column(&scalar_array, target_type, cast_options)?
+        } else {
+            cast_with_options(&scalar_array, target_type, cast_options)?
         };
 
         ScalarValue::try_from_array(&cast_arr, 0)

datafusion/common/src/scalar/mod.rs

@@ -20,7 +20,7 @@
 use arrow::{
     array::{Array, ArrayRef, Date32Array, Date64Array, NullArray},
     compute::{CastOptions, kernels, max, min},
-    datatypes::{DataType, Field},
+    datatypes::DataType,
     util::pretty::pretty_format_columns,
 };
 use datafusion_common::internal_datafusion_err;
@@ -313,24 +313,18 @@ fn cast_array_by_name(
         return Ok(Arc::clone(array));
     }
 
-    match cast_type {
-        DataType::Struct(_) => {
-            // Field name is unused; only the struct's inner field names matter
-            let target_field = Field::new("_", cast_type.clone(), true);
-            datafusion_common::nested_struct::cast_column(
-                array,
-                &target_field,
-                cast_options,
-            )
-        }
-        _ => {
-            ensure_date_array_timestamp_bounds(array, cast_type)?;
-            Ok(kernels::cast::cast_with_options(
-                array,
-                cast_type,
-                cast_options,
-            )?)
-        }
+    if datafusion_common::nested_struct::requires_nested_struct_cast(
+        array.data_type(),
+        cast_type,
+    ) {
+        datafusion_common::nested_struct::cast_column(array, cast_type, cast_options)
+    } else {
+        ensure_date_array_timestamp_bounds(array, cast_type)?;
+        Ok(kernels::cast::cast_with_options(
+            array,
+            cast_type,
+            cast_options,
+        )?)
     }
 }
 

datafusion/expr-common/src/columnar_value.rs

@@ -40,13 +40,13 @@ make_udf_expr_and_func!(
     ArrayRemove,
     array_remove,
     array element,
-    "removes the first element from the array equal to the given value.",
+    "removes the first element from the array equal to the given value. NULL elements already in the array are preserved when removing a non-NULL value. If `element` evaluates to NULL, the result is NULL rather than removing NULL entries.",
     array_remove_udf
 );
 
 #[user_doc(
     doc_section(label = "Array Functions"),
-    description = "Removes the first element from the array equal to the given value.",
+    description = "Removes the first element from the array equal to the given value. NULL elements already in the array are preserved when removing a non-NULL value. If `element` evaluates to NULL, the result is NULL rather than removing NULL entries.",
     syntax_example = "array_remove(array, element)",
     sql_example = r#"```sql
 > select array_remove([1, 2, 2, 3, 2, 1, 4], 2);
@@ -55,6 +55,13 @@ make_udf_expr_and_func!(
 +----------------------------------------------+
 | [1, 2, 3, 2, 1, 4]                           |
 +----------------------------------------------+
+
+> select array_remove([1, 2, NULL, 2, 4], 2);
++---------------------------------------------------+
+| array_remove(List([1,2,NULL,2,4]),Int64(2)) |
++---------------------------------------------------+
+| [1, NULL, 2, 4]                              |
++---------------------------------------------------+
 ```"#,
     argument(
         name = "array",
@@ -127,21 +134,28 @@ make_udf_expr_and_func!(
     ArrayRemoveN,
     array_remove_n,
     array element max,
-    "removes the first `max` elements from the array equal to the given value.",
+    "removes the first `max` elements from the array equal to the given value. NULL elements already in the array are preserved when removing a non-NULL value. If `element` evaluates to NULL, the result is NULL rather than removing NULL entries.",
     array_remove_n_udf
 );
 
 #[user_doc(
     doc_section(label = "Array Functions"),
-    description = "Removes the first `max` elements from the array equal to the given value.",
-    syntax_example = "array_remove_n(array, element, max))",
+    description = "Removes the first `max` elements from the array equal to the given value. NULL elements already in the array are preserved when removing a non-NULL value. If `element` evaluates to NULL, the result is NULL rather than removing NULL entries.",
+    syntax_example = "array_remove_n(array, element, max)",
     sql_example = r#"```sql
 > select array_remove_n([1, 2, 2, 3, 2, 1, 4], 2, 2);
 +---------------------------------------------------------+
 | array_remove_n(List([1,2,2,3,2,1,4]),Int64(2),Int64(2)) |
 +---------------------------------------------------------+
 | [1, 3, 2, 1, 4]                                         |
 +---------------------------------------------------------+
+
+> select array_remove_n([1, 2, NULL, 2, 4], 2, 2);
++----------------------------------------------------------+
+| array_remove_n(List([1,2,NULL,2,4]),Int64(2),Int64(2)) |
++----------------------------------------------------------+
+| [1, NULL, 4]                                            |
++----------------------------------------------------------+
 ```"#,
     argument(
         name = "array",
@@ -219,13 +233,13 @@ make_udf_expr_and_func!(
     ArrayRemoveAll,
     array_remove_all,
     array element,
-    "removes all elements from the array equal to the given value.",
+    "removes all elements from the array equal to the given value. NULL elements already in the array are preserved when removing a non-NULL value. If `element` evaluates to NULL, the result is NULL rather than removing NULL entries.",
     array_remove_all_udf
 );
 
 #[user_doc(
     doc_section(label = "Array Functions"),
-    description = "Removes all elements from the array equal to the given value.",
+    description = "Removes all elements from the array equal to the given value. NULL elements already in the array are preserved when removing a non-NULL value. If `element` evaluates to NULL, the result is NULL rather than removing NULL entries.",
     syntax_example = "array_remove_all(array, element)",
     sql_example = r#"```sql
 > select array_remove_all([1, 2, 2, 3, 2, 1, 4], 2);
@@ -234,6 +248,13 @@ make_udf_expr_and_func!(
 +--------------------------------------------------+
 | [1, 3, 1, 4]                                     |
 +--------------------------------------------------+
+
+> select array_remove_all([1, 2, NULL, 2, 4], 2);
++-----------------------------------------------------+
+| array_remove_all(List([1,2,NULL,2,4]),Int64(2)) |
++-----------------------------------------------------+
+| [1, NULL, 4]                                     |
++-----------------------------------------------------+
 ```"#,
     argument(
         name = "array",

datafusion/functions-nested/src/remove.rs

@@ -25,12 +25,11 @@ use std::hash::Hash;
 use std::sync::Arc;
 
 use arrow::array::RecordBatch;
-use arrow::compute::can_cast_types;
 use arrow::datatypes::{DataType, Field, FieldRef, SchemaRef};
 use datafusion_common::{
-    Result, ScalarValue, exec_err,
+    DataFusionError, Result, ScalarValue, exec_err,
     metadata::FieldMetadata,
-    nested_struct::validate_struct_compatibility,
+    nested_struct::validate_data_type_compatibility,
     tree_node::{Transformed, TransformedResult, TreeNode},
 };
 use datafusion_functions::core::getfield::GetFieldFunc;
@@ -487,31 +486,18 @@ impl DefaultPhysicalExprAdapterRewriter {
         physical_field: FieldRef,
         logical_field: &Field,
     ) -> Result<Transformed<Arc<dyn PhysicalExpr>>> {
-        // For struct types, use validate_struct_compatibility which handles:
-        // - Missing fields in source (filled with nulls)
-        // - Extra fields in source (ignored)
-        // - Recursive validation of nested structs
-        // For non-struct types, use Arrow's can_cast_types
-        match (physical_field.data_type(), logical_field.data_type()) {
-            (DataType::Struct(physical_fields), DataType::Struct(logical_fields)) => {
-                validate_struct_compatibility(
-                    physical_fields.as_ref(),
-                    logical_fields.as_ref(),
-                )?;
-            }
-            _ => {
-                let is_compatible =
-                    can_cast_types(physical_field.data_type(), logical_field.data_type());
-                if !is_compatible {
-                    return exec_err!(
-                        "Cannot cast column '{}' from '{}' (physical data type) to '{}' (logical data type)",
+        validate_data_type_compatibility(
+            column.name(),
+            physical_field.data_type(),
+            logical_field.data_type(),
+        )
+        .map_err(|e|
+                     DataFusionError::Execution(format!(
+                        "Cannot cast column '{}' from '{}' (physical data type) to '{}' (logical data type): {e}",
                         column.name(),
                         physical_field.data_type(),
                         logical_field.data_type()
-                    );
-                }
-            }
-        }
+                    )))?;
 
         let cast_expr = Arc::new(CastColumnExpr::new(
             Arc::new(column),
@@ -663,8 +649,8 @@ impl BatchAdapter {
 mod tests {
     use super::*;
     use arrow::array::{
-        BooleanArray, Int32Array, Int64Array, RecordBatch, RecordBatchOptions,
-        StringArray, StringViewArray, StructArray,
+        Array, BooleanArray, GenericListArray, Int32Array, Int64Array, RecordBatch,
+        RecordBatchOptions, StringArray, StringViewArray, StructArray,
     };
     use arrow::datatypes::{Fields, Schema};
     use datafusion_common::{assert_contains, record_batch};
@@ -1282,6 +1268,142 @@ mod tests {
         assert_eq!(extra_values.iter().collect_vec(), vec![None, None, None]);
     }
 
+    /// Test that List<Struct> columns are properly adapted with struct evolution.
+    #[test]
+    fn test_adapt_list_struct_batches() {
+        // Physical: List<{id: Int32, name: Utf8}>
+        let physical_struct_fields: Fields = vec![
+            Field::new("id", DataType::Int32, false),
+            Field::new("name", DataType::Utf8, true),
+        ]
+        .into();
+
+        let struct_array = StructArray::new(
+            physical_struct_fields.clone(),
+            vec![
+                Arc::new(Int32Array::from(vec![1, 2, 3])) as _,
+                Arc::new(StringArray::from(vec![
+                    Some("alice"),
+                    None,
+                    Some("charlie"),
+                ])) as _,
+            ],
+            None,
+        );
+
+        // One list element per row
+        let item_field = Arc::new(Field::new(
+            "item",
+            DataType::Struct(physical_struct_fields.clone()),
+            true,
+        ));
+        let offsets =
+            arrow::buffer::OffsetBuffer::from_lengths(vec![1usize; struct_array.len()]);
+        let list_array = GenericListArray::<i32>::new(
+            item_field,
+            offsets,
+            Arc::new(struct_array),
+            None,
+        );
+
+        let physical_schema = Arc::new(Schema::new(vec![Field::new(
+            "data",
+            DataType::List(Arc::new(Field::new(
+                "item",
+                DataType::Struct(physical_struct_fields),
+                true,
+            ))),
+            false,
+        )]));
+
+        let physical_batch = RecordBatch::try_new(
+            Arc::clone(&physical_schema),
+            vec![Arc::new(list_array)],
+        )
+        .unwrap();
+
+        // Logical: List<{id: Int64, name: Utf8View, extra: Boolean}>
+        let logical_struct_fields: Fields = vec![
+            Field::new("id", DataType::Int64, false),
+            Field::new("name", DataType::Utf8View, true),
+            Field::new("extra", DataType::Boolean, true),
+        ]
+        .into();
+
+        let logical_schema = Arc::new(Schema::new(vec![Field::new(
+            "data",
+            DataType::List(Arc::new(Field::new(
+                "item",
+                DataType::Struct(logical_struct_fields.clone()),
+                true,
+            ))),
+            false,
+        )]));
+
+        let projection = vec![col("data", &logical_schema).unwrap()];
+
+        let factory = DefaultPhysicalExprAdapterFactory;
+        let adapter = factory
+            .create(Arc::clone(&logical_schema), Arc::clone(&physical_schema))
+            .unwrap();
+
+        let adapted_projection = projection
+            .into_iter()
+            .map(|expr| adapter.rewrite(expr).unwrap())
+            .collect_vec();
+
+        let adapted_schema = Arc::new(Schema::new(
+            adapted_projection
+                .iter()
+                .map(|expr| expr.return_field(&physical_schema).unwrap())
+                .collect_vec(),
+        ));
+
+        let res = batch_project(
+            adapted_projection,
+            &physical_batch,
+            Arc::clone(&adapted_schema),
+        )
+        .unwrap();
+
+        assert_eq!(res.num_columns(), 1);
+
+        let result_list = res
+            .column(0)
+            .as_any()
+            .downcast_ref::<GenericListArray<i32>>()
+            .unwrap();
+
+        // Check each list element contains the evolved struct
+        assert_eq!(result_list.len(), 3);
+        let flat_structs = result_list
+            .values()
+            .as_any()
+            .downcast_ref::<StructArray>()
+            .unwrap();
+
+        let id_col = flat_structs.column_by_name("id").unwrap();
+        assert_eq!(id_col.data_type(), &DataType::Int64);
+        let id_values = id_col.as_any().downcast_ref::<Int64Array>().unwrap();
+        assert_eq!(
+            id_values.iter().collect_vec(),
+            vec![Some(1), Some(2), Some(3)]
+        );
+
+        let name_col = flat_structs.column_by_name("name").unwrap();
+        assert_eq!(name_col.data_type(), &DataType::Utf8View);
+        let name_values = name_col.as_any().downcast_ref::<StringViewArray>().unwrap();
+        assert_eq!(
+            name_values.iter().collect_vec(),
+            vec![Some("alice"), None, Some("charlie")]
+        );
+
+        let extra_col = flat_structs.column_by_name("extra").unwrap();
+        assert_eq!(extra_col.data_type(), &DataType::Boolean);
+        let extra_values = extra_col.as_any().downcast_ref::<BooleanArray>().unwrap();
+        assert_eq!(extra_values.iter().collect_vec(), vec![None, None, None]);
+    }
+
     #[test]
     fn test_try_rewrite_struct_field_access() {
         // Test the core logic of try_rewrite_struct_field_access