fix: avoid extraneous casts for equivalent nested types

datafusion/expr/src/type_coercion/functions.rs

@@ -130,7 +130,7 @@ pub fn fields_with_udf<F: UDFCoercionExt>(
     let valid_types = get_valid_types_with_udf(type_signature, &current_types, func)?;
     if valid_types
         .iter()
-        .any(|data_type| data_type == &current_types)
+        .any(|data_type| data_types_match(data_type, &current_types))
     {
         return Ok(current_fields.to_vec());
     }
@@ -236,7 +236,7 @@ pub fn data_types(
         get_valid_types(function_name.as_ref(), type_signature, current_types)?;
     if valid_types
         .iter()
-        .any(|data_type| data_type == current_types)
+        .any(|data_type| data_types_match(data_type, current_types))
     {
         return Ok(current_types.to_vec());
     }
@@ -307,6 +307,34 @@ fn try_coerce_types(
     )
 }
 
+fn data_types_match(valid_types: &[DataType], current_types: &[DataType]) -> bool {
+    fn field_matches(valid: &FieldRef, current: &FieldRef) -> bool {
+        valid.is_nullable() == current.is_nullable()
+            && data_type_matches(valid.data_type(), current.data_type())
+    }
+
+    fn data_type_matches(valid: &DataType, current: &DataType) -> bool {
+        match (valid, current) {
+            (valid, current) if valid == current => true,
+            (DataType::List(valid), DataType::List(current))
+            | (DataType::LargeList(valid), DataType::LargeList(current)) => {
+                field_matches(valid, current)
+            }
+            (
+                DataType::FixedSizeList(valid, valid_size),
+                DataType::FixedSizeList(current, current_size),
+            ) => valid_size == current_size && field_matches(valid, current),
+            _ => false,
+        }
+    }
+
+    valid_types.len() == current_types.len()
+        && valid_types
+            .iter()
+            .zip(current_types)
+            .all(|(valid_type, current_type)| data_type_matches(valid_type, current_type))
+}
+
 fn get_valid_types_with_udf<F: UDFCoercionExt>(
     signature: &TypeSignature,
     current_types: &[DataType],
@@ -757,6 +785,10 @@ fn maybe_data_types(
     for (i, valid_type) in valid_types.iter().enumerate() {
         let current_type = &current_types[i];
 
+        // Keep exact equality here. Some kernels such as `make_array`
+        // require nested field names/order to match exactly at runtime.
+        // Structural-equivalence short-circuiting is handled earlier by
+        // `data_types_match`.
         if current_type == valid_type {
             new_type.push(current_type.clone())
         } else {
@@ -789,6 +821,10 @@ fn maybe_data_types_without_coercion(
     for (i, valid_type) in valid_types.iter().enumerate() {
         let current_type = &current_types[i];
 
+        // Keep exact equality here. Some kernels such as `make_array`
+        // require nested field names/order to match exactly at runtime.
+        // Structural-equivalence short-circuiting is handled earlier by
+        // `data_types_match`.
         if current_type == valid_type {
             new_type.push(current_type.clone())
         } else if can_cast_types(current_type, valid_type) {
@@ -1044,6 +1080,99 @@ mod tests {
         }
     }
 
+    #[test]
+    fn test_maybe_data_types_uses_exact_nested_types() {
+        let struct_fields = vec![
+            Field::new("id", DataType::Utf8, true),
+            Field::new("prim", DataType::Boolean, true),
+        ];
+        let current_type = DataType::List(Arc::new(Field::new(
+            "item",
+            DataType::Struct(struct_fields.clone().into()),
+            true,
+        )));
+        let valid_type = DataType::List(Arc::new(Field::new(
+            "element",
+            DataType::Struct(struct_fields.into()),
+            true,
+        )));
+
+        assert!(current_type.equals_datatype(&valid_type));
+        assert_ne!(current_type, valid_type);
+        assert_eq!(
+            maybe_data_types(std::slice::from_ref(&valid_type), &[current_type]),
+            Some(vec![valid_type])
+        );
+    }
+
+    #[test]
+    fn test_maybe_data_types_without_coercion_uses_exact_nested_types() {
+        let valid_type = DataType::Struct(
+            vec![
+                Field::new("a", DataType::Int64, true),
+                Field::new("b", DataType::Int64, true),
+            ]
+            .into(),
+        );
+        let current_type = DataType::Struct(
+            vec![
+                Field::new("b", DataType::Int64, true),
+                Field::new("a", DataType::Int64, true),
+            ]
+            .into(),
+        );
+
+        assert!(current_type.equals_datatype(&valid_type));
+        assert_ne!(current_type, valid_type);
+        assert_eq!(
+            maybe_data_types_without_coercion(
+                std::slice::from_ref(&valid_type),
+                &[current_type],
+            ),
+            Some(vec![valid_type])
+        );
+    }
+
+    #[test]
+    fn test_data_types_match_ignores_list_field_name() {
+        let struct_fields = vec![
+            Field::new("id", DataType::Utf8, true),
+            Field::new("prim", DataType::Boolean, true),
+        ];
+        let current_type = DataType::List(Arc::new(Field::new(
+            "item",
+            DataType::Struct(struct_fields.clone().into()),
+            true,
+        )));
+        let valid_type = DataType::List(Arc::new(Field::new(
+            "element",
+            DataType::Struct(struct_fields.into()),
+            true,
+        )));
+
+        assert!(data_types_match(&[valid_type], &[current_type]));
+    }
+
+    #[test]
+    fn test_data_types_match_respects_struct_field_order() {
+        let valid_type = DataType::Struct(
+            vec![
+                Field::new("a", DataType::Int64, true),
+                Field::new("b", DataType::Int64, true),
+            ]
+            .into(),
+        );
+        let current_type = DataType::Struct(
+            vec![
+                Field::new("b", DataType::Int64, true),
+                Field::new("a", DataType::Int64, true),
+            ]
+            .into(),
+        );
+
+        assert!(!data_types_match(&[valid_type], &[current_type]));
+    }
+
     #[test]
     fn test_get_valid_types_numeric() -> Result<()> {
         let get_valid_types_flatten =
@@ -1223,6 +1352,36 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn test_fields_with_udf_preserves_equivalent_nested_types() -> Result<()> {
+        let struct_fields = vec![
+            Field::new("id", DataType::Utf8, true),
+            Field::new("prim", DataType::Boolean, true),
+        ];
+        let current_type = DataType::List(Arc::new(Field::new(
+            "item",
+            DataType::Struct(struct_fields.clone().into()),
+            true,
+        )));
+        let signature_type = DataType::List(Arc::new(Field::new(
+            "element",
+            DataType::Struct(struct_fields.into()),
+            true,
+        )));
+
+        assert!(current_type.equals_datatype(&signature_type));
+
+        let current_fields = vec![Arc::new(Field::new("field", current_type, true))];
+        let coerced_fields = fields_with_udf(
+            &current_fields,
+            &MockUdf(Signature::exact(vec![signature_type], Volatility::Stable)),
+        )?;
+
+        assert_eq!(coerced_fields, current_fields);
+
+        Ok(())
+    }
+
     #[test]
     fn test_nested_wildcard_fixed_size_lists() -> Result<()> {
         let type_into = DataType::FixedSizeList(

datafusion/optimizer/src/analyzer/type_coercion.rs

@@ -1770,6 +1770,42 @@ mod test {
         )
     }
 
+    #[test]
+    fn scalar_function_preserves_equivalent_nested_types() -> Result<()> {
+        let struct_fields = vec![
+            Field::new("id", Utf8, true),
+            Field::new("prim", DataType::Boolean, true),
+        ];
+        let current_type = DataType::List(Arc::new(Field::new(
+            "item",
+            DataType::Struct(struct_fields.clone().into()),
+            true,
+        )));
+        let signature_type = DataType::List(Arc::new(Field::new(
+            "element",
+            DataType::Struct(struct_fields.into()),
+            true,
+        )));
+        let empty = empty_with_type(current_type);
+        let fun = ScalarUDF::new_from_impl(TestScalarUDF {
+            signature: Signature::exact(vec![signature_type], Volatility::Stable),
+        });
+        let scalar_function_expr =
+            Expr::ScalarFunction(ScalarFunction::new_udf(Arc::new(fun), vec![col("a")]));
+        let plan = LogicalPlan::Projection(Projection::try_new(
+            vec![scalar_function_expr],
+            empty,
+        )?);
+
+        assert_analyzed_plan_eq!(
+            plan,
+            @r"
+        Projection: TestScalarUDF(a)
+          EmptyRelation: rows=0
+        "
+        )
+    }
+
     #[test]
     fn agg_udaf() -> Result<()> {
         let empty = empty();