Keep annotated untyped NULL neutral during CASE coercion

datafusion/expr/src/expr_schema.rs

@@ -234,8 +234,20 @@ fn case_field_metadata(case: &Case, schema: &dyn ExprSchema) -> Result<FieldRef>
                 let Some(source) = fields.pop() else {
                     return internal_err!("Missing CASE cast input field");
                 };
+                let untyped_null =
+                    source.certainly_null && source.field.data_type().is_null();
+                let mut field = cast_output_field(&source.field, target, force_nullable);
+                // A type-only coercion must not make an untyped NULL constrain CASE metadata.
+                if untyped_null
+                    && target.metadata().is_empty()
+                    && !field.metadata().is_empty()
+                {
+                    field = Arc::new(field.as_ref().clone().with_metadata(
+                        std::collections::HashMap::<String, String>::new(),
+                    ));
+                }
                 fields.push(BranchField {
-                    field: cast_output_field(&source.field, target, force_nullable),
+                    field,
                     certainly_null: source.certainly_null,
                 });
             }

datafusion/optimizer/src/analyzer/type_coercion.rs

@@ -2761,6 +2761,55 @@ mod test {
         Ok(())
     }
 
+    #[test]
+    fn test_case_coercion_preserves_metadata_with_annotated_null() -> Result<()> {
+        let metadata = std::collections::HashMap::from([
+            (
+                "ARROW:extension:name".to_string(),
+                "example.case".to_string(),
+            ),
+            ("structured_type".to_string(), "ARRAY(NUMBER)".to_string()),
+        ]);
+        let schema = DFSchema::from_unqualified_fields(
+            vec![
+                Field::new("structured", DataType::Int32, false)
+                    .with_metadata(metadata.clone()),
+            ]
+            .into(),
+            std::collections::HashMap::new(),
+        )?;
+        let conflicting = std::collections::HashMap::from([
+            (
+                "ARROW:extension:name".to_string(),
+                "example.other".to_string(),
+            ),
+            ("structured_type".to_string(), "OBJECT".to_string()),
+        ]);
+        for null_metadata in [metadata.clone(), conflicting] {
+            let annotated_null = Expr::Literal(
+                ScalarValue::Null,
+                Some(expr::FieldMetadata::from(null_metadata)),
+            );
+            let case = Case {
+                expr: None,
+                when_then_expr: vec![(Box::new(lit(true)), Box::new(col("structured")))],
+                else_expr: Some(Box::new(annotated_null)),
+            };
+            assert_eq!(
+                Expr::Case(case.clone()).to_field(&schema)?.1.metadata(),
+                &metadata
+            );
+
+            let coerced = coerce_case_expression(case, &schema)?;
+            assert!(matches!(coerced.else_expr.as_deref(), Some(Expr::Cast(_))));
+            assert_eq!(
+                Expr::Case(coerced).to_field(&schema)?.1.metadata(),
+                &metadata
+            );
+        }
+        Ok(())
+    }
+
     #[test]
     fn test_case_expression_coercion() -> Result<()> {
         let schema = Arc::new(DFSchema::from_unqualified_fields(