Cherry-pick apache/datafusion#25015 onto branch-55

datafusion/substrait/src/logical_plan/consumer/expr/literal.rs

@@ -16,7 +16,9 @@
 // under the License.
 
 use crate::logical_plan::consumer::SubstraitConsumer;
-use crate::logical_plan::consumer::types::from_substrait_type;
+use crate::logical_plan::consumer::types::{
+    from_substrait_type, substrait_type_to_field,
+};
 use crate::logical_plan::consumer::utils::{DEFAULT_TIMEZONE, next_struct_field_name};
 use crate::variation_const::FLOAT_16_TYPE_NAME;
 #[expect(deprecated)]
@@ -338,19 +340,17 @@ pub(crate) fn from_substrait_literal(
                 Some(v) => Ok(v),
                 _ => plan_err!("Missing value type for empty map"),
             }?;
-            let key_type = from_substrait_type(consumer, key, dfs_names, name_idx)?;
-            let value_type = from_substrait_type(consumer, value, dfs_names, name_idx)?;
+            let key_field = substrait_type_to_field(consumer, key, dfs_names, name_idx)?
+                .with_name("key")
+                .with_nullable(false);
+            let value_field =
+                substrait_type_to_field(consumer, value, dfs_names, name_idx)?
+                    .with_name("value");
 
             // new_empty_array on a MapType creates a too empty array
             // We want it to contain an empty struct array to align with an empty MapBuilder one
-            let entries = Field::new_struct(
-                "entries",
-                vec![
-                    Field::new("key", key_type, false),
-                    Field::new("value", value_type, true),
-                ],
-                false,
-            );
+            let entries =
+                Field::new_struct("entries", vec![key_field, value_field], false);
             let struct_array =
                 new_empty_array(entries.data_type()).as_struct().to_owned();
             ScalarValue::Map(Arc::new(MapArray::new(

datafusion/substrait/src/logical_plan/consumer/substrait_consumer.rs

@@ -36,7 +36,7 @@ use datafusion::common::{
 use datafusion::execution::{FunctionRegistry, SessionState};
 use datafusion::logical_expr::expr::LambdaVariable;
 use datafusion::logical_expr::{Expr, Extension, LogicalPlan};
-use std::collections::VecDeque;
+use std::collections::{HashMap, VecDeque};
 use std::sync::{Arc, RwLock};
 use substrait::proto::expression as substrait_expression;
 use substrait::proto::expression::{
@@ -500,6 +500,28 @@ pub trait SubstraitConsumer: Send + Sync + Sized {
         )
     }
 
+    /// Optional Arrow field metadata for a Substrait type.
+    ///
+    /// Arrow cannot represent some Substrait types natively. A user-defined
+    /// "json" type, for example, becomes a plain `DataType::Utf8`, which loses
+    /// the fact that the values are JSON. A consumer can return metadata here
+    /// to keep the original type recoverable from the Arrow schema.
+    ///
+    /// The consumer sees the whole [`Type`], so it can key the metadata off a
+    /// user-defined type, a type variation, or anything else it recognizes.
+    /// The metadata is attached to the Arrow [`Field`] built for this type,
+    /// including a type nested inside a list, map, or struct.
+    ///
+    /// The default attaches nothing.
+    ///
+    /// [`Field`]: datafusion::arrow::datatypes::Field
+    fn consume_type_metadata(
+        &self,
+        _typ: &Type,
+    ) -> datafusion::common::Result<Option<HashMap<String, String>>> {
+        Ok(None)
+    }
+
     fn consume_user_defined_literal(
         &self,
         user_defined_literal: &proto::expression::literal::UserDefined,