Preserve VALUES names across INSERT planning and parameter binding

datafusion/expr/src/logical_plan/plan.rs

@@ -2118,8 +2118,37 @@ impl LogicalPlan {
             // Build `LogicalPlan::Values` from the values for type inference.
             // We can't use `recompute_schema` because it skips recomputing for
             // `LogicalPlan::Values`.
-            LogicalPlan::Values(Values { values, schema: _ }) => {
-                LogicalPlanBuilder::values(values)?.build()
+            LogicalPlan::Values(Values { values, schema }) => {
+                let mut plan = LogicalPlanBuilder::values(values)?.build()?;
+                let LogicalPlan::Values(rebuilt) = &mut plan else {
+                    return internal_err!("Expected VALUES plan after rebuilding VALUES");
+                };
+                if rebuilt
+                    .schema
+                    .fields()
+                    .iter()
+                    .zip(schema.fields())
+                    .any(|(new, original)| new.name() != original.name())
+                {
+                    let fields = rebuilt
+                        .schema
+                        .fields()
+                        .iter()
+                        .zip(schema.fields())
+                        .map(|(new, original)| {
+                            Arc::new(
+                                new.as_ref()
+                                    .clone()
+                                    .with_name(original.name().to_string()),
+                            )
+                        })
+                        .collect::<Vec<_>>();
+                    rebuilt.schema = Arc::new(DFSchema::from_unqualified_fields(
+                        fields.into(),
+                        rebuilt.schema.metadata().clone(),
+                    )?);
+                }
+                Ok(plan)
             }
             // other plans can just use `recompute_schema` directly.
             plan => plan.recompute_schema(),

datafusion/sql/src/values.rs

@@ -364,6 +364,7 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
         planner_context.set_table_schema(table_schema.clone());
         let mut values = values?;
 
+        let has_target_schema = table_schema.is_some();
         let schema = table_schema.unwrap_or(empty_schema);
         let object_columns = schema
             .fields()
@@ -425,7 +426,7 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
             } else {
                 LogicalPlanBuilder::values_with_schema(values, &schema)?.build()
             };
-            return self.normalize_values_column_names(plan?);
+            return self.normalize_values_column_names(plan?, has_target_schema);
         }
         for row in &values {
             for expr in row {
@@ -566,14 +567,16 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
         } else {
             LogicalPlanBuilder::values_with_schema(values, &values_schema)?.build()
         };
-        self.normalize_values_column_names(plan?)
+        self.normalize_values_column_names(plan?, has_target_schema)
     }
 
     fn normalize_values_column_names(
         &self,
         mut plan: LogicalPlan,
+        has_target_schema: bool,
     ) -> Result<LogicalPlan> {
-        if !self.options.enable_ident_normalization
+        if has_target_schema
+            || !self.options.enable_ident_normalization
             || !self
                 .context_provider
                 .options()

datafusion/sql/tests/sql_integration.rs

@@ -3724,6 +3724,35 @@ fn uppercase_unquoted_identifiers_apply_to_values_columns() {
     )
     .unwrap();
     assert_eq!(unnormalized.schema().field(0).name(), "column1");
+
+    let insert = logical_plan_with_uppercase_unquoted_identifiers(
+        "INSERT INTO \"test_decimal\" VALUES (1, 2)",
+    )
+    .unwrap();
+    let LogicalPlan::Dml(dml) = insert else {
+        panic!("expected INSERT plan");
+    };
+    let LogicalPlan::Projection(projection) = dml.input.as_ref() else {
+        panic!("expected INSERT projection");
+    };
+    assert_eq!(projection.input.schema().field(0).name(), "column1");
+
+    let bound =
+        logical_plan_with_uppercase_unquoted_identifiers("VALUES (CAST($1 AS INT))")
+            .unwrap()
+            .with_param_values(vec![ScalarValue::Int32(Some(7))])
+            .unwrap();
+    assert_eq!(bound.schema().field(0).name(), "COLUMN1");
+    logical_plan_with_uppercase_unquoted_identifiers(
+        "SELECT column1 FROM (VALUES (CAST($1 AS INT)))",
+    )
+    .unwrap()
+    .with_param_values(vec![ScalarValue::Int32(Some(7))])
+    .unwrap();
+    logical_plan_with_uppercase_unquoted_identifiers("SELECT column1 FROM (VALUES (7))")
+        .unwrap()
+        .with_param_values(Vec::<ScalarValue>::new())
+        .unwrap();
 }
 
 #[test]