Infer data type from schema for Values and add struct coercion to coalesce

datafusion/expr-common/src/type_coercion/binary.rs

@@ -534,6 +534,7 @@ fn type_union_resolution_coercion(
     }
 }
 
+/// Handle type union resolution including struct type and others.
 pub fn try_type_union_resolution(data_types: &[DataType]) -> Result<Vec<DataType>> {
     let err = match try_type_union_resolution_with_struct(data_types) {
         Ok(struct_types) => return Ok(struct_types),

datafusion/expr/src/logical_plan/builder.rs

@@ -170,6 +170,29 @@ impl LogicalPlanBuilder {
         })))
     }
 
+    pub fn values(values: Vec<Vec<Expr>>) -> Result<Self> {
+        if values.is_empty() {
+            return plan_err!("Values list cannot be empty");
+        }
+        let n_cols = values[0].len();
+        if n_cols == 0 {
+            return plan_err!("Values list cannot be zero length");
+        }
+        for (i, row) in values.iter().enumerate() {
+            if row.len() != n_cols {
+                return plan_err!(
+                    "Inconsistent data length across values list: got {} values in row {} but expected {}",
+                    row.len(),
+                    i,
+                    n_cols
+                );
+            }
+        }
+
+        // Infer from data itself
+        Self::infer_data(values)
+    }
+
     /// Create a values list based relation, and the schema is inferred from data itself or table schema if provided, consuming
     /// `value`. See the [Postgres VALUES](https://www.postgresql.org/docs/current/queries-values.html)
     /// documentation for more details.
@@ -179,7 +202,10 @@ impl LogicalPlanBuilder {
     /// so it's usually better to override the default names with a table alias list.
     ///
     /// 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>>, schema: Option<&DFSchemaRef>) -> Result<Self> {
+    pub fn values_with_schema(
+        values: Vec<Vec<Expr>>,
+        schema: Option<&DFSchemaRef>,
+    ) -> Result<Self> {
         if values.is_empty() {
             return plan_err!("Values list cannot be empty");
         }
@@ -200,14 +226,17 @@ impl LogicalPlanBuilder {
 
         // Check the type of value against the schema
         if let Some(schema) = schema {
-            Self::infer_from_schema(values, schema)
+            Self::infer_values_from_schema(values, schema)
         } else {
             // Infer from data itself
             Self::infer_data(values)
         }
     }
 
-    fn infer_from_schema(values: Vec<Vec<Expr>>, schema: &DFSchema) -> Result<Self> {
+    fn infer_values_from_schema(
+        values: Vec<Vec<Expr>>,
+        schema: &DFSchema,
+    ) -> Result<Self> {
         let n_cols = values[0].len();
         let mut field_types: Vec<DataType> = Vec::with_capacity(n_cols);
         for j in 0..n_cols {
@@ -2364,10 +2393,10 @@ mod tests {
     fn test_union_after_join() -> Result<()> {
         let values = vec![vec![lit(1)]];
 
-        let left = LogicalPlanBuilder::values(values.clone(), None)?
+        let left = LogicalPlanBuilder::values_with_schema(values.clone(), None)?
             .alias("left")?
             .build()?;
-        let right = LogicalPlanBuilder::values(values, None)?
+        let right = LogicalPlanBuilder::values_with_schema(values, None)?
             .alias("right")?
             .build()?;
 

datafusion/optimizer/src/simplify_expressions/simplify_exprs.rs

@@ -417,7 +417,7 @@ mod tests {
             Box::new(lit(1)),
         ));
         let values = vec![vec![expr1, expr2]];
-        let plan = LogicalPlanBuilder::values(values, None)?.build()?;
+        let plan = LogicalPlanBuilder::values_with_schema(values, None)?.build()?;
 
         let expected = "\
         Values: (Int32(3) AS Int32(1) + Int32(2), Int32(1) AS Int32(2) - Int32(1))";

datafusion/proto/src/logical_plan/mod.rs

@@ -282,7 +282,7 @@ impl AsLogicalPlan for LogicalPlanNode {
                         .map_err(|e| e.into())
                 }?;
 
-                LogicalPlanBuilder::values(values, None)?.build()
+                LogicalPlanBuilder::values_with_schema(values, None)?.build()
             }
             LogicalPlanType::Projection(projection) => {
                 let input: LogicalPlan =

datafusion/sql/src/planner.rs

@@ -139,7 +139,7 @@ pub struct PlannerContext {
     /// FROM clauses, this should become a suffix of the `outer_query_schema`.
     outer_from_schema: Option<DFSchemaRef>,
     /// The query schema defined by the table
-    table_schema: Option<DFSchemaRef>,
+    create_table_schema: Option<DFSchemaRef>,
 }
 
 impl Default for PlannerContext {
@@ -156,7 +156,7 @@ impl PlannerContext {
             ctes: HashMap::new(),
             outer_query_schema: None,
             outer_from_schema: None,
-            table_schema: None,
+            create_table_schema: None,
         }
     }
 
@@ -188,12 +188,12 @@ impl PlannerContext {
         &mut self,
         mut schema: Option<DFSchemaRef>,
     ) -> Option<DFSchemaRef> {
-        std::mem::swap(&mut self.table_schema, &mut schema);
+        std::mem::swap(&mut self.create_table_schema, &mut schema);
         schema
     }
 
     pub fn table_schema(&self) -> Option<DFSchemaRef> {
-        self.table_schema.clone()
+        self.create_table_schema.clone()
     }
 
     // Return a clone of the outer FROM schema

datafusion/sql/src/values.rs

@@ -46,9 +46,9 @@ impl<'a, S: ContextProvider> SqlToRel<'a, S> {
             .collect::<Result<Vec<_>>>()?;
 
         if schema.fields().is_empty() {
-            LogicalPlanBuilder::values(values, None)?.build()
+            LogicalPlanBuilder::values_with_schema(values, None)?.build()
         } else {
-            LogicalPlanBuilder::values(values, Some(&schema))?.build()
+            LogicalPlanBuilder::values_with_schema(values, Some(&schema))?.build()
         }
     }
 }