Avoid target schema propagation into INSERT SELECT

datafusion/core/src/datasource/view_test.rs

@@ -289,10 +289,10 @@ mod tests {
 
         insta::assert_snapshot!(batches_to_string(&results),@r"
         +---------+---------+---------+
-        | column2 | column1 | column3 |
+        | column1 | column2 | column3 |
         +---------+---------+---------+
-        | 2       | 1       | 3       |
-        | 5       | 4       | 6       |
+        | 1       | 2       | 3       |
+        | 4       | 5       | 6       |
         +---------+---------+---------+
         ");
 

datafusion/core/tests/sql/unparser.rs

@@ -378,7 +378,7 @@ async fn optimized_duckdb_unparse_qualifies_nested_passthrough_column() -> Resul
     // `o` (which is only the base-table alias one level deeper). The bug emitted
     // `"o"."order_id"` inside that derived table; the fix emits a bare column.
     let expected = concat!(
-        r#"SELECT "o"."order_id", "o"."discount_pct_2" "#,
+        r#"SELECT "oi"."order_id", "o"."discount_pct_2" "#,
         r#"FROM "warehouse"."main"."order_items" AS "oi" "#,
         r#"INNER JOIN (SELECT "order_id", "#,
         r#"CASE WHEN "__common_expr_1" IS NOT NULL "#,

datafusion/sql/src/statement.rs

@@ -2828,6 +2828,7 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
         let table_name = self.object_name_to_table_reference(table_name)?;
         let table_source = self.context_provider.get_table_source(table_name.clone())?;
         let table_schema = DFSchema::try_from(table_source.schema())?;
+        let source_is_values = matches!(source.body.as_ref(), SetExpr::Values(_));
 
         let columns: Vec<Ident> = columns
             .into_iter()
@@ -2927,9 +2928,11 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
         // Projection
         let mut planner_context =
             PlannerContext::new().with_prepare_param_data_types(prepare_param_data_types);
-        planner_context.set_table_schema(Some(DFSchemaRef::new(
-            DFSchema::from_unqualified_fields(fields.clone(), Default::default())?,
-        )));
+        if source_is_values {
+            planner_context.set_table_schema(Some(DFSchemaRef::new(
+                DFSchema::from_unqualified_fields(fields.clone(), Default::default())?,
+            )));
+        }
         let source = self.query_to_plan(*source, &mut planner_context)?;
         if fields.len() != source.schema().fields().len() {
             plan_err!("Column count doesn't match insert query!")?;

datafusion/sql/tests/sql_integration.rs

@@ -737,6 +737,15 @@ fn plan_insert_no_target_columns() {
     );
 }
 
+#[test]
+fn plan_insert_select_expression_from_values() {
+    let sql = "INSERT INTO array (\"left\") \
+               SELECT make_array(column1) FROM (VALUES (1), (2))";
+    let plan = logical_plan_with_dialect(sql, &GenericDialect {}).unwrap();
+
+    assert_contains!(plan.display_indent().to_string(), "make_array(column1)");
+}
+
 #[rstest]
 #[case::duplicate_columns(
     "INSERT INTO test_decimal (id, price, price) VALUES (1, 2, 3), (4, 5, 6)",