feat: add opt-in VALUES column metadata resolution

AGENTS.md

@@ -55,6 +55,12 @@ For modified code identify local benchmarks(if any) and run them against `main`.
 
 ## Agent Skills
 
+VALUES metadata resolution is opt-in through `TypePlanner` and
+`LogicalPlanBuilder::values_with_metadata_resolver`. Keep the ordinary builder
+strict and the equal-metadata path free of policy calls. A successful cold
+resolver validates a borrowed complete column once; it owns metadata policy,
+never VALUES type coercion/nullability, expression evaluation or CASE semantics.
+
 Repository-specific agent skills live under `.ai/skills/`. Each subdirectory is
 a single skill with a `SKILL.md` (YAML frontmatter + body). Check that
 directory for applicable skills before working on a task; new skills go in

datafusion/expr/README.md

@@ -23,6 +23,12 @@
 
 This crate is a submodule of DataFusion that provides data types and utilities for logical plans and expressions.
 
+The `values_with_metadata_resolver` builder provides an opt-in cold policy
+for conflicting VALUES metadata. Default inference remains strict; equal metadata
+never invokes the policy. A successful policy validates a borrowed full column
+once, while the builder retains type union, nullability and coercion. SQL's existing
+`TypePlanner` can supply the policy without changing target-schema VALUES or CASE.
+
 Most projects should use the [`datafusion`] crate directly, which re-exports
 this module. If you are already using the [`datafusion`] crate, there is no
 reason to use this crate directly in your project as well.

datafusion/expr/src/logical_plan/builder.rs

@@ -68,6 +68,16 @@ use indexmap::IndexSet;
 /// Default table name for unnamed table
 pub const UNNAMED_TABLE: &str = "?table?";
 
+/// Optional VALUES column-metadata policy. Called at most once per column, only
+/// after unequal metadata is found. The complete input rows are borrowed so a
+/// policy can validate every contributor without cloning expressions.
+///
+/// `None` declines resolution and retains the original strict metadata error.
+/// `Some`, including empty metadata, replaces that column's inferred metadata.
+/// Errors are propagated. Type coercion and nullability remain builder-owned.
+pub type ValuesMetadataResolver<'a> =
+    dyn Fn(&[Vec<Expr>], usize) -> Result<Option<FieldMetadata>> + 'a;
+
 /// Options for [`LogicalPlanBuilder`]
 #[derive(Default, Debug, Clone)]
 pub struct LogicalPlanBuilderOptions {
@@ -211,6 +221,25 @@ impl LogicalPlanBuilder {
     ///
     /// If the values include params/binders such as $1, $2, $3, etc, then the `param_data_types` should be provided.
     pub fn values(values: Vec<Vec<Expr>>) -> Result<Self> {
+        Self::values_impl(values, None)
+    }
+
+    /// Build VALUES using an opt-in policy for otherwise conflicting metadata.
+    ///
+    /// Equal-metadata columns retain the ordinary path and never call `resolver`.
+    /// A successful resolver must validate the complete requested column: it is
+    /// not called again for later rows. Other columns remain independent.
+    pub fn values_with_metadata_resolver(
+        values: Vec<Vec<Expr>>,
+        resolver: &ValuesMetadataResolver<'_>,
+    ) -> Result<Self> {
+        Self::values_impl(values, Some(resolver))
+    }
+
+    fn values_impl(
+        values: Vec<Vec<Expr>>,
+        resolver: Option<&ValuesMetadataResolver<'_>>,
+    ) -> Result<Self> {
         if values.is_empty() {
             return plan_err!("Values list cannot be empty");
         }
@@ -230,7 +259,7 @@ impl LogicalPlanBuilder {
         }
 
         // Infer from data itself
-        Self::infer_data(values)
+        Self::infer_data(values, resolver)
     }
 
     /// Create a values list based relation, and the schema is inferred from data itself or table schema if provided, consuming
@@ -306,14 +335,18 @@ impl LogicalPlanBuilder {
         Self::infer_inner(values, fields, schema)
     }
 
-    fn infer_data(values: Vec<Vec<Expr>>) -> Result<Self> {
+    fn infer_data(
+        values: Vec<Vec<Expr>>,
+        resolver: Option<&ValuesMetadataResolver<'_>>,
+    ) -> Result<Self> {
         let n_cols = values[0].len();
         let schema = DFSchema::empty();
         let mut fields = ValuesFields::new();
 
         for j in 0..n_cols {
             let mut common_type: Option<DataType> = None;
             let mut common_metadata: Option<FieldMetadata> = None;
+            let mut metadata_resolved = false;
             let mut nullable = false;
             for (i, row) in values.iter().enumerate() {
                 let value = &row[j];
@@ -327,15 +360,22 @@ impl LogicalPlanBuilder {
 
                 let is_null_literal =
                     matches!(value, Expr::Literal(value, _) if value.is_null());
-                if !is_null_literal {
+                if !is_null_literal && !metadata_resolved {
                     let metadata = value.metadata(&schema)?;
                     if let Some(ref cm) = common_metadata {
                         if &metadata != cm {
-                            return plan_err!(
-                                "Inconsistent metadata across values list at row {i} column {j}. Was {:?} but found {:?}",
-                                cm,
-                                metadata
-                            );
+                            if let Some(resolver) = resolver
+                                && let Some(resolved) = resolver(&values, j)?
+                            {
+                                common_metadata = Some(resolved);
+                                metadata_resolved = true;
+                            } else {
+                                return plan_err!(
+                                    "Inconsistent metadata across values list at row {i} column {j}. Was {:?} but found {:?}",
+                                    cm,
+                                    metadata
+                                );
+                            }
                         }
                     } else {
                         common_metadata = Some(metadata.clone());
@@ -3290,6 +3330,129 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn test_values_metadata_resolver_is_cold_and_borrows_full_column() -> Result<()> {
+        use std::cell::Cell;
+
+        let calls = Cell::new(0);
+        let metadata = FieldMetadata::from(HashMap::from([(
+            "example.width".to_owned(),
+            "10".to_owned(),
+        )]));
+        let values = vec![
+            vec![lit("a")],
+            vec![lit("longer")],
+            vec![Expr::Cast(expr::Cast::new_from_field(
+                Box::new(lit(ScalarValue::Null)),
+                Arc::new(
+                    Field::new("", DataType::Utf8, true)
+                        .with_metadata(metadata.to_hashmap()),
+                ),
+            ))],
+        ];
+        let first = &raw const values[0][0];
+        let resolver = |rows: &[Vec<Expr>], column| {
+            calls.set(calls.get() + 1);
+            assert_eq!(column, 0);
+            assert_eq!(rows.len(), 3);
+            assert_eq!(&raw const rows[0][0], first);
+            assert_eq!(rows[1][0], lit("longer"));
+            Ok(Some(metadata.clone()))
+        };
+        let plan = LogicalPlanBuilder::values_with_metadata_resolver(values, &resolver)?
+            .build()?;
+        assert_eq!(calls.get(), 1);
+        assert_eq!(plan.schema().field(0).data_type(), &DataType::Utf8);
+        assert!(plan.schema().field(0).is_nullable());
+        assert_eq!(plan.schema().field(0).metadata(), &metadata.to_hashmap());
+
+        calls.set(0);
+        LogicalPlanBuilder::values_with_metadata_resolver(
+            vec![vec![lit("a")], vec![lit("b")]],
+            &resolver,
+        )?;
+        assert_eq!(calls.get(), 0, "equal metadata must not call the policy");
+        Ok(())
+    }
+
+    #[test]
+    fn test_values_metadata_resolver_declines_or_errors_without_fallback() -> Result<()> {
+        let metadata = FieldMetadata::from(HashMap::from([(
+            "example.width".to_owned(),
+            "10".to_owned(),
+        )]));
+        let values = || {
+            vec![
+                vec![lit_with_metadata("a", Some(metadata.clone()))],
+                vec![lit("b")],
+            ]
+        };
+        let original = LogicalPlanBuilder::values(values()).unwrap_err();
+        let declined =
+            LogicalPlanBuilder::values_with_metadata_resolver(values(), &|_, _| Ok(None))
+                .unwrap_err();
+        assert_eq!(original.strip_backtrace(), declined.strip_backtrace());
+        let rejected =
+            LogicalPlanBuilder::values_with_metadata_resolver(values(), &|_, _| {
+                plan_err!("metadata policy refused")
+            })
+            .unwrap_err();
+        assert_eq!(
+            rejected.strip_backtrace(),
+            "Error during planning: metadata policy refused"
+        );
+
+        let empty =
+            LogicalPlanBuilder::values_with_metadata_resolver(values(), &|_, _| {
+                Ok(Some(FieldMetadata::default()))
+            })?
+            .build()?;
+        assert!(empty.schema().field(0).metadata().is_empty());
+        Ok(())
+    }
+
+    #[test]
+    fn test_values_metadata_resolver_runs_once_per_conflicting_column() -> Result<()> {
+        use std::cell::Cell;
+
+        let calls = Cell::new(0);
+        let values = (0..8)
+            .map(|index| {
+                (0..2)
+                    .map(|column| {
+                        lit_with_metadata(
+                            index,
+                            Some(FieldMetadata::from(HashMap::from([(
+                                "example".to_owned(),
+                                format!("{column}:{index}"),
+                            )]))),
+                        )
+                    })
+                    .collect()
+            })
+            .collect();
+        let plan = LogicalPlanBuilder::values_with_metadata_resolver(
+            values,
+            &|rows, column| {
+                calls.set(calls.get() + 1);
+                assert_eq!(rows.len(), 8);
+                Ok(Some(FieldMetadata::from(HashMap::from([(
+                    "example".to_owned(),
+                    column.to_string(),
+                )]))))
+            },
+        )?
+        .build()?;
+        assert_eq!(calls.get(), 2);
+        for column in 0..2 {
+            assert_eq!(
+                plan.schema().field(column).metadata().get("example"),
+                Some(&column.to_string())
+            );
+        }
+        Ok(())
+    }
+
     #[test]
     fn test_unique_field_aliases() {
         let t1_field_1 = Field::new("a", DataType::Int32, false);

datafusion/expr/src/planner.rs

@@ -29,6 +29,8 @@ use crate::{
 };
 use arrow::datatypes::{DataType, Field, FieldRef, SchemaRef};
 use datafusion_common::datatype::DataTypeExt;
+#[cfg(feature = "sql")]
+use datafusion_common::metadata::FieldMetadata;
 use datafusion_common::{
     DFSchema, Result, TableReference, config::ConfigOptions,
     file_options::file_type::FileType, not_impl_err,
@@ -472,6 +474,22 @@ pub trait RelationPlannerContext {
 ///
 /// [Extending SQL in DataFusion: from ->> to TABLESAMPLE blog]: https://datafusion.apache.org/blog/2026/01/12/extending-sql
 pub trait TypePlanner: Debug + Send + Sync {
+    /// Resolve conflicting metadata for one inferred VALUES column.
+    ///
+    /// Called only on a metadata mismatch, at most once for each column. Input
+    /// rows are borrowed and already planned expressions; inspect the complete
+    /// selected column before returning a replacement. `None` preserves the
+    /// builder's strict error. `Some`, even empty metadata, accepts a complete
+    /// replacement; an error refuses planning. This does not alter expressions,
+    /// type union/coercion, nullability, CASE, or target-schema VALUES semantics.
+    fn plan_values_column_metadata(
+        &self,
+        _rows: &[Vec<Expr>],
+        _column: usize,
+    ) -> Result<Option<FieldMetadata>> {
+        Ok(None)
+    }
+
     /// Plan SQL [`sqlparser::ast::DataType`] to DataFusion [`DataType`]
     ///
     /// Returns None if not possible

datafusion/sql/README.md

@@ -25,6 +25,12 @@ project that requires a SQL query planner and does not make any assumptions abou
 will be translated to a physical plan. For example, there is no concept of row-based versus columnar execution in the
 logical plan.
 
+The optional `TypePlanner::plan_values_column_metadata` hook can resolve an
+inferred VALUES column's conflicting metadata. Its default implementation declines
+resolution, retaining the strict error. The SQL planner delegates to the logical
+builder's cold resolver once per conflicting column. It does not embed a
+product-specific width policy or change VALUES with a target schema.
+
 Note that the [`datafusion`] crate re-exports this module. If you are already
 using the [`datafusion`] crate in your project, there is no reason to use this
 crate directly in your project as well.

datafusion/sql/src/values.rs

@@ -422,7 +422,17 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
             .collect::<Vec<_>>();
         if array_columns.is_empty() {
             let plan = if schema.fields().is_empty() {
-                LogicalPlanBuilder::values(values)?.build()
+                if let Some(type_planner) = self.context_provider.get_type_planner() {
+                    LogicalPlanBuilder::values_with_metadata_resolver(
+                        values,
+                        &|rows, column| {
+                            type_planner.plan_values_column_metadata(rows, column)
+                        },
+                    )?
+                    .build()
+                } else {
+                    LogicalPlanBuilder::values(values)?.build()
+                }
             } else {
                 LogicalPlanBuilder::values_with_schema(values, &schema)?.build()
             };