From 0a74a9c7d7716c02b5371faf8d333d69ea5febaa Mon Sep 17 00:00:00 2001 From: Denys Tsomenko Date: Sat, 7 Nov 2026 10:13:29 +0200 Subject: [PATCH] feat: add opt-in VALUES column metadata resolution --- AGENTS.md | 6 + datafusion/expr/README.md | 6 + datafusion/expr/src/logical_plan/builder.rs | 179 +++++++++++++++++++- datafusion/expr/src/planner.rs | 18 ++ datafusion/sql/README.md | 6 + datafusion/sql/src/values.rs | 12 +- datafusion/sql/tests/sql_integration.rs | 76 +++++++++ 7 files changed, 294 insertions(+), 9 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 8fdf314ed4b6c..a76abd6ea88ae 100644 --- a/AGENTS.md +++ b/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 diff --git a/datafusion/expr/README.md b/datafusion/expr/README.md index b3ab9a383dbbd..c20d8abe7ea30 100644 --- a/datafusion/expr/README.md +++ b/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. diff --git a/datafusion/expr/src/logical_plan/builder.rs b/datafusion/expr/src/logical_plan/builder.rs index fbd94ec69bacd..1f3acdb8291b1 100644 --- a/datafusion/expr/src/logical_plan/builder.rs +++ b/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], usize) -> Result> + '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>) -> Result { + 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>, + resolver: &ValuesMetadataResolver<'_>, + ) -> Result { + Self::values_impl(values, Some(resolver)) + } + + fn values_impl( + values: Vec>, + resolver: Option<&ValuesMetadataResolver<'_>>, + ) -> Result { 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,7 +335,10 @@ impl LogicalPlanBuilder { Self::infer_inner(values, fields, schema) } - fn infer_data(values: Vec>) -> Result { + fn infer_data( + values: Vec>, + resolver: Option<&ValuesMetadataResolver<'_>>, + ) -> Result { let n_cols = values[0].len(); let schema = DFSchema::empty(); let mut fields = ValuesFields::new(); @@ -314,6 +346,7 @@ impl LogicalPlanBuilder { for j in 0..n_cols { let mut common_type: Option = None; let mut common_metadata: Option = 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], 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); diff --git a/datafusion/expr/src/planner.rs b/datafusion/expr/src/planner.rs index 4658d22325d2f..b0298093d215b 100644 --- a/datafusion/expr/src/planner.rs +++ b/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], + _column: usize, + ) -> Result> { + Ok(None) + } + /// Plan SQL [`sqlparser::ast::DataType`] to DataFusion [`DataType`] /// /// Returns None if not possible diff --git a/datafusion/sql/README.md b/datafusion/sql/README.md index d0e5e498e514c..8f783bdf41476 100644 --- a/datafusion/sql/README.md +++ b/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. diff --git a/datafusion/sql/src/values.rs b/datafusion/sql/src/values.rs index 512ad6b82c0c2..26a5b463e163d 100644 --- a/datafusion/sql/src/values.rs +++ b/datafusion/sql/src/values.rs @@ -422,7 +422,17 @@ impl SqlToRel<'_, S> { .collect::>(); 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() }; diff --git a/datafusion/sql/tests/sql_integration.rs b/datafusion/sql/tests/sql_integration.rs index eb8dd55e5cce6..8670a3f5e8570 100644 --- a/datafusion/sql/tests/sql_integration.rs +++ b/datafusion/sql/tests/sql_integration.rs @@ -7090,6 +7090,82 @@ fn test_no_substring_registered_alt_syntax() { ); } +#[test] +fn test_values_metadata_uses_opt_in_type_planner() -> Result<()> { + use datafusion_common::metadata::FieldMetadata; + use datafusion_expr::planner::TypePlanner; + use std::collections::HashMap; + use std::sync::atomic::{AtomicUsize, Ordering}; + + #[derive(Debug)] + struct WidthPlanner { + calls: AtomicUsize, + accept: bool, + } + + impl TypePlanner for WidthPlanner { + fn plan_type_field( + &self, + sql_type: &sqlparser::ast::DataType, + ) -> Result> { + if matches!(sql_type, sqlparser::ast::DataType::Varchar(Some(_))) { + return Ok(Some(Arc::new( + Field::new("", DataType::Utf8, true).with_metadata(HashMap::from([ + ("example.width".to_owned(), "10".to_owned()), + ])), + ))); + } + Ok(None) + } + + fn plan_values_column_metadata( + &self, + rows: &[Vec], + column: usize, + ) -> Result> { + self.calls.fetch_add(1, Ordering::Relaxed); + assert_eq!(column, 0); + assert_eq!(rows.len(), 3); + assert_eq!(rows[1][0], datafusion_expr::lit("longer")); + Ok(self.accept.then(|| { + FieldMetadata::from(HashMap::from([( + "example.width".to_owned(), + "10".to_owned(), + )])) + })) + } + } + + for accept in [false, true] { + let type_planner = Arc::new(WidthPlanner { + calls: AtomicUsize::new(0), + accept, + }); + let context = MockContextProvider { + state: MockSessionState::default().with_type_planner(type_planner.clone()), + }; + let planner = SqlToRel::new(&context); + let statement = + DFParser::parse_sql("VALUES ('a'), ('longer'), (CAST(NULL AS VARCHAR(10)))")? + .pop_front() + .unwrap(); + let result = planner.statement_to_plan(statement); + assert_eq!(type_planner.calls.load(Ordering::Relaxed), 1); + if accept { + let plan = result?; + assert_eq!(plan.schema().field(0).data_type(), &DataType::Utf8); + assert!(plan.schema().field(0).is_nullable()); + assert_eq!( + plan.schema().field(0).metadata().get("example.width"), + Some(&"10".to_owned()) + ); + } else { + assert_contains!(result.unwrap_err().to_string(), "Inconsistent metadata"); + } + } + Ok(()) +} + #[test] fn test_custom_type_plan() -> Result<()> { let sql = "SELECT DATETIME '2001-01-01 18:00:00'";