From 69c3629d8d73834f169b057473604860d52dd3eb Mon Sep 17 00:00:00 2001 From: osipovartem Date: Tue, 20 Oct 2026 22:41:30 +0300 Subject: [PATCH 1/3] Coerce VARIANT values into untyped OBJECT columns on insert --- datafusion/sql/src/statement.rs | 21 +++++++++++++++++- datafusion/sql/tests/common/mod.rs | 9 ++++++++ datafusion/sql/tests/sql_integration.rs | 29 +++++++++++++++++++++++++ 3 files changed, 58 insertions(+), 1 deletion(-) diff --git a/datafusion/sql/src/statement.rs b/datafusion/sql/src/statement.rs index 404299ac3f83b..17fb080e0afaa 100644 --- a/datafusion/sql/src/statement.rs +++ b/datafusion/sql/src/statement.rs @@ -2968,6 +2968,10 @@ impl SqlToRel<'_, S> { .metadata() .get("embucket.snowflake_logical_type") .is_some_and(|value| value == "ARRAY"); + let target_is_object = target_field + .metadata() + .get("embucket.snowflake_logical_type") + .is_some_and(|value| value == "OBJECT"); let source_is_text = matches!( source_field.data_type(), arrow::datatypes::DataType::Utf8 @@ -2980,7 +2984,22 @@ impl SqlToRel<'_, S> { | arrow::datatypes::DataType::LargeList(_) | arrow::datatypes::DataType::FixedSizeList(_, _) ); - let expr = if target_field.extension_type_name() + let expr = if target_is_object + && target_field.data_type() == &arrow::datatypes::DataType::Utf8 + && source_field.extension_type_name() + == Some("arrow.parquet.variant") + { + let to_object = self + .context_provider + .get_function_meta("to_object") + .ok_or_else(|| { + DataFusionError::Plan( + "to_object is required to insert VARIANT into OBJECT storage" + .to_string(), + ) + })?; + Expr::ScalarFunction(ScalarFunction::new_udf(to_object, vec![expr])) + } else if target_field.extension_type_name() == Some("arrow.parquet.variant") && source_field.data_type() != target_field.data_type() && (source_is_array diff --git a/datafusion/sql/tests/common/mod.rs b/datafusion/sql/tests/common/mod.rs index 481d3d14070cc..64fbfbd8ae784 100644 --- a/datafusion/sql/tests/common/mod.rs +++ b/datafusion/sql/tests/common/mod.rs @@ -255,6 +255,15 @@ impl ContextProvider for MockContextProvider { .into(), ), ])), + "untyped_object_target" => Ok(Schema::new(vec![ + Field::new("items", DataType::Utf8, true).with_metadata( + [( + "embucket.snowflake_logical_type".to_string(), + "OBJECT".to_string(), + )] + .into(), + ), + ])), "orders" => Ok(Schema::new(vec![ Field::new("order_id", DataType::UInt32, false), Field::new("o_orderkey", DataType::UInt32, false), diff --git a/datafusion/sql/tests/sql_integration.rs b/datafusion/sql/tests/sql_integration.rs index 7f67c800bbf3d..1ef7d6d2a5759 100644 --- a/datafusion/sql/tests/sql_integration.rs +++ b/datafusion/sql/tests/sql_integration.rs @@ -869,6 +869,35 @@ fn plan_insert_converts_tagged_array_to_variant_storage() -> Result<()> { Ok(()) } +#[test] +fn plan_insert_converts_variant_to_untyped_object_storage() -> Result<()> { + let state = mock_session_state().with_scalar_function(Arc::new(make_udf( + "to_object", + vec![variant_storage_type()], + DataType::Utf8, + ))); + let plan = logical_plan_from_state( + "INSERT INTO untyped_object_target SELECT items FROM plain_variant_target", + &SnowflakeDialect {}, + ParserOptions::default(), + state, + )?; + assert_contains!( + plan.display_indent().to_string(), + "to_object(plain_variant_target.items)" + ); + Ok(()) +} + +#[test] +fn plan_insert_keeps_untyped_object_text_fast_path() -> Result<()> { + let plan = logical_plan( + "INSERT INTO untyped_object_target SELECT items FROM plain_string_source", + )?; + assert!(!plan.display_indent().to_string().contains("to_object(")); + Ok(()) +} + #[test] fn plan_insert_uses_explicit_array_conversion_for_untagged_text() -> Result<()> { let state = mock_session_state().with_scalar_function(Arc::new(make_udf( From 0aa34b4f6916f661b4bece400c9af79e22e49527 Mon Sep 17 00:00:00 2001 From: osipovartem Date: Tue, 20 Oct 2026 23:07:43 +0300 Subject: [PATCH 2/3] Validate untyped OBJECT insert values before schema coercion --- datafusion/sql/src/statement.rs | 13 ++++++++++ datafusion/sql/src/values.rs | 33 ++++++++++++++++++++++++ datafusion/sql/tests/sql_integration.rs | 34 +++++++++++++++++++++++-- 3 files changed, 78 insertions(+), 2 deletions(-) diff --git a/datafusion/sql/src/statement.rs b/datafusion/sql/src/statement.rs index 17fb080e0afaa..ebb220fb358d3 100644 --- a/datafusion/sql/src/statement.rs +++ b/datafusion/sql/src/statement.rs @@ -2984,6 +2984,19 @@ impl SqlToRel<'_, S> { | arrow::datatypes::DataType::LargeList(_) | arrow::datatypes::DataType::FixedSizeList(_, _) ); + if target_is_object + && target_field.data_type() == &arrow::datatypes::DataType::Utf8 + && !source_is_values + && source_is_text + && !source_field + .metadata() + .get("embucket.snowflake_logical_type") + .is_some_and(|value| value == "OBJECT") + { + plan_err!( + "Cannot implicitly convert text to OBJECT during INSERT; use TO_OBJECT" + )?; + } let expr = if target_is_object && target_field.data_type() == &arrow::datatypes::DataType::Utf8 && source_field.extension_type_name() diff --git a/datafusion/sql/src/values.rs b/datafusion/sql/src/values.rs index cc466be7f2e1a..dd9ac739399fe 100644 --- a/datafusion/sql/src/values.rs +++ b/datafusion/sql/src/values.rs @@ -58,6 +58,39 @@ impl SqlToRel<'_, S> { let mut values = values?; let schema = table_schema.unwrap_or(empty_schema); + let object_columns = schema + .fields() + .iter() + .enumerate() + .filter_map(|(index, field)| { + (field.data_type() == &DataType::Utf8 + && field + .metadata() + .get("embucket.snowflake_logical_type") + .is_some_and(|value| value == "OBJECT")) + .then_some(index) + }) + .collect::>(); + for row in &values { + for &index in &object_columns { + let Some(expr) = row.get(index) else { + continue; + }; + let (_, source_field) = expr.to_field(&schema)?; + if matches!( + source_field.data_type(), + DataType::Utf8 | DataType::Utf8View | DataType::LargeUtf8 + ) && !source_field + .metadata() + .get("embucket.snowflake_logical_type") + .is_some_and(|value| value == "OBJECT") + { + plan_err!( + "Cannot implicitly convert text to OBJECT during INSERT; use TO_OBJECT" + )?; + } + } + } let array_columns = schema .fields() .iter() diff --git a/datafusion/sql/tests/sql_integration.rs b/datafusion/sql/tests/sql_integration.rs index 1ef7d6d2a5759..50d735ca501da 100644 --- a/datafusion/sql/tests/sql_integration.rs +++ b/datafusion/sql/tests/sql_integration.rs @@ -890,14 +890,44 @@ fn plan_insert_converts_variant_to_untyped_object_storage() -> Result<()> { } #[test] -fn plan_insert_keeps_untyped_object_text_fast_path() -> Result<()> { +fn plan_insert_keeps_untyped_object_fast_path() -> Result<()> { let plan = logical_plan( - "INSERT INTO untyped_object_target SELECT items FROM plain_string_source", + "INSERT INTO untyped_object_target SELECT items FROM untyped_object_target", )?; assert!(!plan.display_indent().to_string().contains("to_object(")); Ok(()) } +#[test] +fn plan_insert_rejects_plain_text_into_untyped_object() { + let err = logical_plan( + "INSERT INTO untyped_object_target SELECT items FROM plain_string_source", + ) + .expect_err("plain text needs an explicit OBJECT conversion"); + assert_contains!(err.to_string(), "Cannot implicitly convert text to OBJECT"); +} + +#[test] +fn plan_insert_values_rejects_plain_text_into_untyped_object() { + let err = logical_plan("INSERT INTO untyped_object_target VALUES ('{\"a\":1}')") + .expect_err("plain text VALUES needs an explicit OBJECT conversion"); + assert_contains!(err.to_string(), "Cannot implicitly convert text to OBJECT"); +} + +#[test] +fn plan_insert_values_rejects_mixed_null_and_plain_text_into_untyped_object() { + let err = + logical_plan("INSERT INTO untyped_object_target VALUES (NULL), ('{\"a\":1}')") + .expect_err("a mixed VALUES list still needs explicit OBJECT conversion"); + assert_contains!(err.to_string(), "Cannot implicitly convert text to OBJECT"); +} + +#[test] +fn plan_insert_values_accepts_null_for_untyped_object() -> Result<()> { + logical_plan("INSERT INTO untyped_object_target VALUES (NULL)")?; + Ok(()) +} + #[test] fn plan_insert_uses_explicit_array_conversion_for_untagged_text() -> Result<()> { let state = mock_session_state().with_scalar_function(Arc::new(make_udf( From 0a7e53f704aa5f9024d6d340bfc2a2f1b69f2b7b Mon Sep 17 00:00:00 2001 From: osipovartem Date: Wed, 21 Oct 2026 00:53:54 +0300 Subject: [PATCH 3/3] Validate OBJECT set branches and bound inserts Reject incompatible scalar branches, preserve compatible OBJECT/VARIANT insert conversion, and validate bound UNION branches after parameter replacement. --- datafusion/expr/src/logical_plan/plan.rs | 398 ++++++++++++++++++++++- datafusion/sql/src/set_expr.rs | 69 ++++ datafusion/sql/src/statement.rs | 27 +- datafusion/sql/src/values.rs | 333 ++++++++++++++++++- datafusion/sql/tests/sql_integration.rs | 252 +++++++++++++- 5 files changed, 1061 insertions(+), 18 deletions(-) diff --git a/datafusion/expr/src/logical_plan/plan.rs b/datafusion/expr/src/logical_plan/plan.rs index ab00cb0bca87d..68a3869d6531e 100644 --- a/datafusion/expr/src/logical_plan/plan.rs +++ b/datafusion/expr/src/logical_plan/plan.rs @@ -1662,6 +1662,370 @@ impl LogicalPlan { } } +fn bound_source_is_null_literal(source: &LogicalPlan, index: usize) -> Result { + enum Node<'a> { + Plan(&'a LogicalPlan, usize), + Expr(&'a Expr, Option<&'a LogicalPlan>), + } + let mut pending = vec![Node::Plan(source, index)]; + while let Some(node) = pending.pop() { + match node { + Node::Plan(plan, index) => match plan { + LogicalPlan::Projection(node) => { + let Some(expr) = node.expr.get(index) else { + return Ok(false); + }; + pending.push(Node::Expr(expr, Some(node.input.as_ref()))); + } + LogicalPlan::Values(node) => { + for row in &node.values { + let Some(expr) = row.get(index) else { + return Ok(false); + }; + pending.push(Node::Expr(expr, None)); + } + } + LogicalPlan::Filter(node) => pending.push(Node::Plan(&node.input, index)), + LogicalPlan::Limit(node) => pending.push(Node::Plan(&node.input, index)), + LogicalPlan::Sort(node) => pending.push(Node::Plan(&node.input, index)), + LogicalPlan::SubqueryAlias(node) => { + pending.push(Node::Plan(&node.input, index)); + } + LogicalPlan::Subquery(node) => { + pending.push(Node::Plan(&node.subquery, index)); + } + LogicalPlan::Repartition(node) => { + pending.push(Node::Plan(&node.input, index)); + } + LogicalPlan::Distinct(node) => { + pending.push(Node::Plan(node.input(), index)); + } + LogicalPlan::Join(node) => { + let column = Column::from(plan.schema().qualified_field(index)); + if let Some(index) = node.left.schema().maybe_index_of_column(&column) + { + pending.push(Node::Plan(&node.left, index)); + } else if let Some(index) = + node.right.schema().maybe_index_of_column(&column) + { + pending.push(Node::Plan(&node.right, index)); + } else { + return Ok(false); + } + } + LogicalPlan::Aggregate(node) => { + let expr = node.group_expr.get(index).or_else(|| { + index + .checked_sub(node.group_expr.len()) + .and_then(|index| node.aggr_expr.get(index)) + }); + let Some(expr) = expr else { + return Ok(false); + }; + pending.push(Node::Expr(expr, Some(node.input.as_ref()))); + } + LogicalPlan::Window(node) => { + let input_len = node.input.schema().fields().len(); + if index < input_len { + pending.push(Node::Plan(node.input.as_ref(), index)); + } else if let Some(expr) = node.window_expr.get(index - input_len) { + pending.push(Node::Expr(expr, Some(node.input.as_ref()))); + } else { + return Ok(false); + } + } + LogicalPlan::Union(node) => { + pending + .extend(node.inputs.iter().map(|input| Node::Plan(input, index))); + } + _ => return Ok(false), + }, + Node::Expr(expr, input) => match expr { + Expr::Alias(alias) => pending.push(Node::Expr(&alias.expr, input)), + Expr::Column(column) => { + let Some(input) = input else { + return Ok(false); + }; + let Some(index) = input.schema().maybe_index_of_column(column) else { + return Ok(false); + }; + pending.push(Node::Plan(input, index)); + } + Expr::Literal(value, _) if value.is_null() => {} + _ => return Ok(false), + }, + } + } + Ok(true) +} + +fn bound_source_incompatible_branch_type( + source: &LogicalPlan, + index: usize, +) -> Result> { + enum Node<'a> { + Plan(&'a LogicalPlan, usize), + Expr(&'a Expr, Option<&'a LogicalPlan>), + } + let mut pending = vec![(Node::Plan(source, index), 0)]; + while let Some((node, depth)) = pending.pop() { + if depth > 256 { + return plan_err!("OBJECT INSERT source is too deeply nested"); + } + match node { + Node::Plan(plan, index) => { + let field = plan.schema().field(index); + let passes_through_source = match plan { + LogicalPlan::Union(_) + | LogicalPlan::Filter(_) + | LogicalPlan::Limit(_) + | LogicalPlan::Sort(_) + | LogicalPlan::SubqueryAlias(_) + | LogicalPlan::Subquery(_) + | LogicalPlan::Repartition(_) + | LogicalPlan::Distinct(_) + | LogicalPlan::Join(_) => true, + LogicalPlan::Projection(node) => { + node.expr.get(index).is_some_and(|expr| { + let mut expr = expr; + while let Expr::Alias(alias) = expr { + expr = alias.expr.as_ref(); + } + matches!(expr, Expr::Column(_)) + }) + } + LogicalPlan::Aggregate(node) => node + .group_expr + .get(index) + .is_some_and(|expr| matches!(expr, Expr::Column(_))), + LogicalPlan::Window(node) => { + index < node.input.schema().fields().len() + } + _ => false, + }; + if field.data_type() != &DataType::Null && !passes_through_source { + if field.extension_type_name() != Some("arrow.parquet.variant") + && !field + .metadata() + .get("embucket.snowflake_logical_type") + .is_some_and(|logical_type| logical_type == "OBJECT") + && !bound_source_is_null_literal(plan, index)? + { + return Ok(Some(field.data_type().clone())); + } + continue; + } + match plan { + LogicalPlan::Projection(node) => { + if let Some(expr) = node.expr.get(index) { + pending.push(( + Node::Expr(expr, Some(node.input.as_ref())), + depth + 1, + )); + } + } + LogicalPlan::Filter(node) => { + pending.push((Node::Plan(node.input.as_ref(), index), depth + 1)); + } + LogicalPlan::Limit(node) => { + pending.push((Node::Plan(node.input.as_ref(), index), depth + 1)); + } + LogicalPlan::Sort(node) => { + pending.push((Node::Plan(node.input.as_ref(), index), depth + 1)); + } + LogicalPlan::SubqueryAlias(node) => { + pending.push((Node::Plan(node.input.as_ref(), index), depth + 1)); + } + LogicalPlan::Subquery(node) => { + pending + .push((Node::Plan(node.subquery.as_ref(), index), depth + 1)); + } + LogicalPlan::Repartition(node) => { + pending.push((Node::Plan(node.input.as_ref(), index), depth + 1)); + } + LogicalPlan::Distinct(node) => { + pending + .push((Node::Plan(node.input().as_ref(), index), depth + 1)); + } + LogicalPlan::Union(node) => { + pending.extend( + node.inputs.iter().map(|input| { + (Node::Plan(input.as_ref(), index), depth + 1) + }), + ); + } + LogicalPlan::Join(node) => { + let column = Column::from(plan.schema().qualified_field(index)); + if let Some(index) = + node.left.schema().maybe_index_of_column(&column) + { + pending + .push((Node::Plan(node.left.as_ref(), index), depth + 1)); + } else if let Some(index) = + node.right.schema().maybe_index_of_column(&column) + { + pending.push(( + Node::Plan(node.right.as_ref(), index), + depth + 1, + )); + } + } + LogicalPlan::Aggregate(node) => { + let expr = node.group_expr.get(index).or_else(|| { + index + .checked_sub(node.group_expr.len()) + .and_then(|index| node.aggr_expr.get(index)) + }); + if let Some(expr) = expr { + pending.push(( + Node::Expr(expr, Some(node.input.as_ref())), + depth + 1, + )); + } + } + LogicalPlan::Window(node) => { + let input_len = node.input.schema().fields().len(); + if index < input_len { + pending.push(( + Node::Plan(node.input.as_ref(), index), + depth + 1, + )); + } else if let Some(expr) = node.window_expr.get(index - input_len) + { + pending.push(( + Node::Expr(expr, Some(node.input.as_ref())), + depth + 1, + )); + } + } + _ => {} + } + } + Node::Expr(expr, input) => match expr { + Expr::Column(column) => { + if let Some(input) = input + && let Some(index) = input.schema().maybe_index_of_column(column) + { + pending.push((Node::Plan(input, index), depth + 1)); + } + } + Expr::ScalarSubquery(node) => { + pending.push((Node::Plan(node.subquery.as_ref(), 0), depth + 1)); + } + Expr::Case(case) => { + pending.extend(case.when_then_expr.iter().map(|(_, value)| { + (Node::Expr(value.as_ref(), input), depth + 1) + })); + pending.extend( + case.else_expr + .as_deref() + .map(|expr| (Node::Expr(expr, input), depth + 1)), + ); + } + _ => { + expr.apply_children(|child| { + pending.push((Node::Expr(child, input), depth + 1)); + Ok(TreeNodeRecursion::Continue) + })?; + } + }, + } + } + Ok(None) +} + +fn validate_bound_object_insert_source(dml: &DmlStatement) -> Result<()> { + if !matches!(dml.op, WriteOp::Insert(_)) { + return Ok(()); + } + let LogicalPlan::Projection(projection) = dml.input.as_ref() else { + return Ok(()); + }; + if matches!(projection.input.as_ref(), LogicalPlan::Values(_)) { + return Ok(()); + } + for (index, target_field) in dml.target.schema().fields().iter().enumerate() { + if target_field.data_type() != &DataType::Utf8 + || !target_field + .metadata() + .get("embucket.snowflake_logical_type") + .is_some_and(|logical_type| logical_type == "OBJECT") + { + continue; + } + let Some(mut expr) = projection.expr.get(index) else { + continue; + }; + loop { + expr = match expr { + Expr::Alias(alias) => alias.expr.as_ref(), + Expr::Cast(cast) => cast.expr.as_ref(), + _ => break, + }; + } + let Expr::Column(column) = expr else { + continue; + }; + let source_schema = projection.input.schema(); + let source_index = source_schema.index_of_column(column)?; + let source_field = source_schema.field(source_index); + let source_is_object = source_field.extension_type_name() + == Some("arrow.parquet.variant") + || source_field + .metadata() + .get("embucket.snowflake_logical_type") + .is_some_and(|logical_type| logical_type == "OBJECT"); + let incompatible_branch_type = bound_source_incompatible_branch_type( + projection.input.as_ref(), + source_index, + )?; + if (!source_is_object + && source_field.data_type() != &DataType::Null + && !bound_source_is_null_literal(projection.input.as_ref(), source_index)?) + || incompatible_branch_type.is_some() + { + return plan_err!( + "Cannot implicitly convert {} to OBJECT; use TO_OBJECT", + incompatible_branch_type + .as_ref() + .unwrap_or_else(|| source_field.data_type()) + ); + } + } + Ok(()) +} + +fn validate_bound_object_union(union: &Union) -> Result<()> { + for index in 0..union.schema.fields().len() { + let has_object = union.inputs.iter().any(|input| { + input + .schema() + .field(index) + .metadata() + .get("embucket.snowflake_logical_type") + .is_some_and(|logical_type| logical_type == "OBJECT") + }); + if !has_object { + continue; + } + for input in &union.inputs { + let field = input.schema().field(index); + let logical_type = field.metadata().get("embucket.snowflake_logical_type"); + if field.data_type() != &DataType::Null + && logical_type.is_none_or(|logical_type| logical_type != "OBJECT") + && !(field.extension_type_name() == Some("arrow.parquet.variant") + && logical_type.is_none_or(|logical_type| logical_type != "ARRAY")) + { + return plan_err!( + "inconsistent data type for result columns for set operator input branches: OBJECT and {}", + field.data_type() + ); + } + } + } + Ok(()) +} + impl LogicalPlan { /// Return a `LogicalPlan` with all placeholders (e.g $1 $2, /// ...) replaced with corresponding values provided in @@ -1686,10 +2050,32 @@ impl LogicalPlan { } else { let original_name = name_preserver.save(&e); let transformed_expr = e.transform_up(|e| { - if let Expr::Placeholder(Placeholder { id, .. }) = e { + if let Expr::Placeholder(Placeholder { id, field }) = e { let (value, metadata) = param_values .get_placeholders_with_values(&id)? .into_inner(); + if field.as_ref().is_some_and(|field| { + field + .metadata() + .get("embucket.snowflake_logical_type") + .is_some_and(|logical_type| logical_type == "OBJECT") + }) && !value.is_null() + && !metadata.as_ref().is_some_and(|metadata| { + metadata + .inner() + .get("embucket.snowflake_logical_type") + .is_some_and(|logical_type| logical_type == "OBJECT") + || metadata + .inner() + .get("ARROW:extension:name") + .is_some_and(|extension| extension == "arrow.parquet.variant") + }) + { + return plan_err!( + "Cannot implicitly convert {} to OBJECT; use TO_OBJECT", + value.data_type() + ); + } Ok(Transformed::yes(Expr::Literal(value, metadata))) } else { Ok(Transformed::no(e)) @@ -1699,7 +2085,15 @@ impl LogicalPlan { Ok(transformed_expr.update_data(|expr| original_name.restore(expr))) } })? - .map_data(|plan| plan.update_schema_data_type()) + .map_data(|plan| { + let plan = plan.update_schema_data_type()?; + if let LogicalPlan::Dml(dml) = &plan { + validate_bound_object_insert_source(dml)?; + } else if let LogicalPlan::Union(union) = &plan { + validate_bound_object_union(union)?; + } + Ok(plan) + }) }) .map(|res| res.data) } diff --git a/datafusion/sql/src/set_expr.rs b/datafusion/sql/src/set_expr.rs index bf82d10ae50d5..7e292e12c8bfb 100644 --- a/datafusion/sql/src/set_expr.rs +++ b/datafusion/sql/src/set_expr.rs @@ -15,15 +15,83 @@ // specific language governing permissions and limitations // under the License. +use std::collections::HashMap; use std::sync::Arc; use crate::planner::{ContextProvider, PlannerContext, SqlToRel}; +use arrow::datatypes::{DataType, Field}; use datafusion_common::{ DataFusionError, Diagnostic, Result, Span, not_impl_err, plan_err, }; use datafusion_expr::{LogicalPlan, LogicalPlanBuilder}; use sqlparser::ast::{SetExpr, SetOperator, SetQuantifier, Spanned}; +fn is_object_field(field: &Field) -> bool { + field + .metadata() + .get("embucket.snowflake_logical_type") + .is_some_and(|logical_type| logical_type == "OBJECT") +} + +fn is_object_set_compatible(field: &Field) -> bool { + field.data_type() == &DataType::Null + || is_object_field(field) + || (field.extension_type_name() == Some("arrow.parquet.variant") + && !field + .metadata() + .get("embucket.snowflake_logical_type") + .is_some_and(|logical_type| logical_type == "ARRAY")) +} + +fn validate_object_set_column(left: &Field, right: &Field) -> Result<()> { + if (is_object_field(left) && !is_object_set_compatible(right)) + || (is_object_field(right) && !is_object_set_compatible(left)) + { + return plan_err!( + "inconsistent data type for result columns for set operator input branches: OBJECT and {}", + if is_object_field(left) { + right.data_type() + } else { + left.data_type() + } + ); + } + Ok(()) +} + +fn validate_object_set_columns( + left: &LogicalPlan, + right: &LogicalPlan, + set_quantifier: SetQuantifier, +) -> Result<()> { + let left_fields = left.schema().fields(); + let right_fields = right.schema().fields(); + if matches!( + set_quantifier, + SetQuantifier::ByName | SetQuantifier::AllByName | SetQuantifier::DistinctByName + ) { + if !left_fields.iter().any(|field| is_object_field(field)) + && !right_fields.iter().any(|field| is_object_field(field)) + { + return Ok(()); + } + let right_by_name: HashMap<_, _> = right_fields + .iter() + .map(|field| (field.name(), field.as_ref())) + .collect(); + for left in left_fields { + if let Some(right) = right_by_name.get(left.name()) { + validate_object_set_column(left, right)?; + } + } + } else { + for (left, right) in left_fields.iter().zip(right_fields) { + validate_object_set_column(left, right)?; + } + } + Ok(()) +} + impl SqlToRel<'_, S> { pub(super) fn set_expr_to_plan( &self, @@ -139,6 +207,7 @@ impl SqlToRel<'_, S> { right_plan: LogicalPlan, set_quantifier: SetQuantifier, ) -> Result { + validate_object_set_columns(&left_plan, &right_plan, set_quantifier)?; match (op, set_quantifier) { (SetOperator::Union, SetQuantifier::All) => { LogicalPlanBuilder::from(left_plan) diff --git a/datafusion/sql/src/statement.rs b/datafusion/sql/src/statement.rs index ebb220fb358d3..881623f26d562 100644 --- a/datafusion/sql/src/statement.rs +++ b/datafusion/sql/src/statement.rs @@ -2972,6 +2972,21 @@ impl SqlToRel<'_, S> { .metadata() .get("embucket.snowflake_logical_type") .is_some_and(|value| value == "OBJECT"); + let source_is_object_variant_union = target_is_object + && target_field.data_type() == &arrow::datatypes::DataType::Utf8 + && !source_is_values + && crate::values::object_output_is_object_variant_union(&source, v)?; + if target_is_object + && target_field.data_type() == &arrow::datatypes::DataType::Utf8 + && !source_is_values + && source_field.data_type() + == &arrow::datatypes::DataType::Null + && crate::values::object_output_has_placeholder(&source, v)? + { + plan_err!( + "Cannot insert an untyped bound parameter into OBJECT from SELECT" + )?; + } let source_is_text = matches!( source_field.data_type(), arrow::datatypes::DataType::Utf8 @@ -2987,20 +3002,26 @@ impl SqlToRel<'_, S> { if target_is_object && target_field.data_type() == &arrow::datatypes::DataType::Utf8 && !source_is_values - && source_is_text + && source_field.data_type() + != &arrow::datatypes::DataType::Null + && source_field.extension_type_name() + != Some("arrow.parquet.variant") + && !source_is_object_variant_union && !source_field .metadata() .get("embucket.snowflake_logical_type") .is_some_and(|value| value == "OBJECT") { plan_err!( - "Cannot implicitly convert text to OBJECT during INSERT; use TO_OBJECT" + "Cannot implicitly convert {} to OBJECT during INSERT; use TO_OBJECT", + source_field.data_type() )?; } let expr = if target_is_object && target_field.data_type() == &arrow::datatypes::DataType::Utf8 - && source_field.extension_type_name() + && (source_field.extension_type_name() == Some("arrow.parquet.variant") + || source_is_object_variant_union) { let to_object = self .context_provider diff --git a/datafusion/sql/src/values.rs b/datafusion/sql/src/values.rs index dd9ac739399fe..1716dd2cc7fe1 100644 --- a/datafusion/sql/src/values.rs +++ b/datafusion/sql/src/values.rs @@ -21,11 +21,318 @@ use std::sync::Arc; use crate::planner::{ContextProvider, PlannerContext, SqlToRel}; use arrow::datatypes::DataType; use datafusion_common::tree_node::{TreeNode, TreeNodeRecursion}; -use datafusion_common::{DFSchema, DataFusionError, Result, not_impl_err, plan_err}; +use datafusion_common::{ + Column, DFSchema, DataFusionError, Result, not_impl_err, plan_err, +}; use datafusion_expr::expr::ScalarFunction; use datafusion_expr::{Expr, ExprSchemable, LogicalPlan, LogicalPlanBuilder, ScalarUDF}; use sqlparser::ast::Values as SQLValues; +enum ObjectSource<'a> { + Plan(&'a LogicalPlan, usize), + Expr(&'a Expr, Option<&'a LogicalPlan>), +} + +pub(super) fn object_output_has_placeholder( + source: &LogicalPlan, + output_index: usize, +) -> Result { + object_source_has_placeholder(ObjectSource::Plan(source, output_index)) +} + +pub(super) fn object_output_is_object_variant_union( + source: &LogicalPlan, + output_index: usize, +) -> Result { + Ok(object_union_source_kind(source, output_index, 0)? == ObjectUnionKind::Mixed) +} + +#[derive(Clone, Copy, PartialEq, Eq)] +enum ObjectUnionKind { + Object, + Variant, + Null, + Mixed, + Other, +} + +fn object_union_field_kind(field: &arrow::datatypes::Field) -> ObjectUnionKind { + let logical_type = field.metadata().get("embucket.snowflake_logical_type"); + if logical_type.is_some_and(|kind| kind == "OBJECT") { + ObjectUnionKind::Object + } else if field.extension_type_name() == Some("arrow.parquet.variant") + && logical_type.is_none_or(|kind| kind != "ARRAY") + { + ObjectUnionKind::Variant + } else if field.data_type() == &DataType::Null { + ObjectUnionKind::Null + } else { + ObjectUnionKind::Other + } +} + +fn object_union_source_kind( + plan: &LogicalPlan, + index: usize, + depth: usize, +) -> Result { + if depth > 256 { + return plan_err!("OBJECT INSERT source is too deeply nested"); + } + let passthrough = |input: &LogicalPlan, column: &Column| { + input + .schema() + .maybe_index_of_column(column) + .map(|index| object_union_source_kind(input, index, depth + 1)) + }; + match plan { + LogicalPlan::Union(node) => { + let mut has_object = false; + let mut has_variant = false; + for input in &node.inputs { + match object_union_source_kind(input, index, depth + 1)? { + ObjectUnionKind::Object => has_object = true, + ObjectUnionKind::Variant => has_variant = true, + ObjectUnionKind::Mixed => { + has_object = true; + has_variant = true; + } + ObjectUnionKind::Null => {} + ObjectUnionKind::Other => return Ok(ObjectUnionKind::Other), + } + } + Ok(match (has_object, has_variant) { + (true, true) => ObjectUnionKind::Mixed, + (true, false) => ObjectUnionKind::Object, + (false, true) => ObjectUnionKind::Variant, + (false, false) => ObjectUnionKind::Null, + }) + } + LogicalPlan::Projection(node) => { + let Some(mut expr) = node.expr.get(index) else { + return Ok(ObjectUnionKind::Other); + }; + while let Expr::Alias(alias) = expr { + expr = alias.expr.as_ref(); + } + if let Expr::Column(column) = expr + && let Some(kind) = passthrough(node.input.as_ref(), column) + { + return kind; + } + Ok(object_union_field_kind(plan.schema().field(index))) + } + LogicalPlan::Filter(node) => { + object_union_source_kind(&node.input, index, depth + 1) + } + LogicalPlan::Limit(node) => { + object_union_source_kind(&node.input, index, depth + 1) + } + LogicalPlan::Sort(node) => { + object_union_source_kind(&node.input, index, depth + 1) + } + LogicalPlan::SubqueryAlias(node) => { + object_union_source_kind(&node.input, index, depth + 1) + } + LogicalPlan::Subquery(node) => { + object_union_source_kind(&node.subquery, index, depth + 1) + } + LogicalPlan::Repartition(node) => { + object_union_source_kind(&node.input, index, depth + 1) + } + LogicalPlan::Distinct(node) => { + object_union_source_kind(node.input().as_ref(), index, depth + 1) + } + LogicalPlan::Join(node) => { + let column = Column::from(plan.schema().qualified_field(index)); + if let Some(kind) = passthrough(node.left.as_ref(), &column) { + kind + } else if let Some(kind) = passthrough(node.right.as_ref(), &column) { + kind + } else { + Ok(ObjectUnionKind::Other) + } + } + LogicalPlan::Aggregate(node) => { + if let Some(Expr::Column(column)) = node.group_expr.get(index) + && let Some(kind) = passthrough(node.input.as_ref(), column) + { + return kind; + } + Ok(object_union_field_kind(plan.schema().field(index))) + } + LogicalPlan::Window(node) if index < node.input.schema().fields().len() => { + object_union_source_kind(&node.input, index, depth + 1) + } + _ => Ok(object_union_field_kind(plan.schema().field(index))), + } +} + +fn object_value_has_placeholder(expr: &Expr) -> Result { + object_source_has_placeholder(ObjectSource::Expr(expr, None)) +} + +fn object_source_has_placeholder(source: ObjectSource<'_>) -> Result { + let mut pending = vec![(source, 0)]; + while let Some((source, depth)) = pending.pop() { + if depth > 256 { + return plan_err!("OBJECT INSERT source is too deeply nested"); + } + match source { + ObjectSource::Plan(plan, index) => match plan { + LogicalPlan::Projection(node) => { + let expr = node.expr.get(index).ok_or_else(|| { + DataFusionError::Plan( + "OBJECT INSERT source column is missing".into(), + ) + })?; + pending.push(( + ObjectSource::Expr(expr, Some(node.input.as_ref())), + depth + 1, + )); + } + LogicalPlan::Values(node) => { + for row in &node.values { + if let Some(expr) = row.get(index) + && !matches!(expr, Expr::Placeholder(_)) + { + pending.push((ObjectSource::Expr(expr, None), depth + 1)); + } + } + } + LogicalPlan::Limit(node) => { + pending.push(( + ObjectSource::Plan(node.input.as_ref(), index), + depth + 1, + )); + } + LogicalPlan::Sort(node) => { + pending.push(( + ObjectSource::Plan(node.input.as_ref(), index), + depth + 1, + )); + } + LogicalPlan::Filter(node) => { + pending.push(( + ObjectSource::Plan(node.input.as_ref(), index), + depth + 1, + )); + } + LogicalPlan::SubqueryAlias(node) => { + pending.push(( + ObjectSource::Plan(node.input.as_ref(), index), + depth + 1, + )); + } + LogicalPlan::Subquery(node) => { + pending.push(( + ObjectSource::Plan(node.subquery.as_ref(), index), + depth + 1, + )); + } + LogicalPlan::Repartition(node) => { + pending.push(( + ObjectSource::Plan(node.input.as_ref(), index), + depth + 1, + )); + } + LogicalPlan::Distinct(node) => { + pending.push(( + ObjectSource::Plan(node.input().as_ref(), index), + depth + 1, + )); + } + LogicalPlan::Join(node) => { + let column = Column::from(plan.schema().qualified_field(index)); + if let Some(index) = node.left.schema().maybe_index_of_column(&column) + { + pending.push(( + ObjectSource::Plan(node.left.as_ref(), index), + depth + 1, + )); + } else if let Some(index) = + node.right.schema().maybe_index_of_column(&column) + { + pending.push(( + ObjectSource::Plan(node.right.as_ref(), index), + depth + 1, + )); + } else { + return Ok(true); + } + } + LogicalPlan::Aggregate(node) => { + let expr = node.group_expr.get(index).or_else(|| { + index + .checked_sub(node.group_expr.len()) + .and_then(|index| node.aggr_expr.get(index)) + }); + let Some(expr) = expr else { + return Ok(true); + }; + pending.push(( + ObjectSource::Expr(expr, Some(node.input.as_ref())), + depth + 1, + )); + } + LogicalPlan::Window(node) => { + let input_len = node.input.schema().fields().len(); + if index < input_len { + pending.push(( + ObjectSource::Plan(node.input.as_ref(), index), + depth + 1, + )); + } else if let Some(expr) = node.window_expr.get(index - input_len) { + pending.push(( + ObjectSource::Expr(expr, Some(node.input.as_ref())), + depth + 1, + )); + } else { + return Ok(true); + } + } + LogicalPlan::Union(node) => { + pending.extend(node.inputs.iter().map(|input| { + (ObjectSource::Plan(input.as_ref(), index), depth + 1) + })); + } + _ => {} + }, + ObjectSource::Expr(expr, input) => match expr { + Expr::Placeholder(_) => return Ok(true), + Expr::Case(case) => { + pending.extend(case.when_then_expr.iter().map(|(_, value)| { + (ObjectSource::Expr(value.as_ref(), input), depth + 1) + })); + pending.extend( + case.else_expr + .as_deref() + .map(|expr| (ObjectSource::Expr(expr, input), depth + 1)), + ); + } + Expr::Column(column) => { + if let Some(input) = input + && let Some(index) = input.schema().maybe_index_of_column(column) + { + pending.push((ObjectSource::Plan(input, index), depth + 1)); + } + } + Expr::ScalarSubquery(node) => { + pending + .push((ObjectSource::Plan(node.subquery.as_ref(), 0), depth + 1)); + } + _ => { + expr.apply_children(|child| { + pending.push((ObjectSource::Expr(child, input), depth + 1)); + Ok(TreeNodeRecursion::Continue) + })?; + } + }, + } + } + Ok(false) +} + impl SqlToRel<'_, S> { pub(super) fn sql_values_to_plan( &self, @@ -77,16 +384,24 @@ impl SqlToRel<'_, S> { continue; }; let (_, source_field) = expr.to_field(&schema)?; - if matches!( - source_field.data_type(), - DataType::Utf8 | DataType::Utf8View | DataType::LargeUtf8 - ) && !source_field - .metadata() - .get("embucket.snowflake_logical_type") - .is_some_and(|value| value == "OBJECT") + if source_field.data_type() == &DataType::Null + && !matches!(expr, Expr::Placeholder(_)) + && object_value_has_placeholder(expr)? + { + return plan_err!( + "Cannot insert an untyped bound expression into OBJECT" + ); + } + if source_field.data_type() != &DataType::Null + && source_field.extension_type_name() != Some("arrow.parquet.variant") + && !source_field + .metadata() + .get("embucket.snowflake_logical_type") + .is_some_and(|value| value == "OBJECT") { plan_err!( - "Cannot implicitly convert text to OBJECT during INSERT; use TO_OBJECT" + "Cannot implicitly convert {} to OBJECT during INSERT; use TO_OBJECT", + source_field.data_type() )?; } } diff --git a/datafusion/sql/tests/sql_integration.rs b/datafusion/sql/tests/sql_integration.rs index 50d735ca501da..f7796a14a5d42 100644 --- a/datafusion/sql/tests/sql_integration.rs +++ b/datafusion/sql/tests/sql_integration.rs @@ -40,7 +40,7 @@ use datafusion_expr::{ logical_plan::LogicalPlan, test::function_stub::sum_udaf, }; -use datafusion_functions::{string, unicode}; +use datafusion_functions::{core as df_core, string, unicode}; use datafusion_sql::{ parser::DFParser, planner::{NullOrdering, ParserOptions, PlannerContext, SqlToRel}, @@ -889,6 +889,41 @@ fn plan_insert_converts_variant_to_untyped_object_storage() -> Result<()> { Ok(()) } +#[test] +fn plan_insert_converts_object_variant_union_to_untyped_object() -> Result<()> { + for sql in [ + "INSERT INTO untyped_object_target SELECT items FROM untyped_object_target UNION ALL SELECT items FROM plain_variant_target", + "INSERT INTO untyped_object_target SELECT item FROM (SELECT items AS item FROM untyped_object_target UNION ALL SELECT items AS item FROM plain_variant_target)", + "INSERT INTO untyped_object_target SELECT item FROM (SELECT items AS item FROM untyped_object_target UNION ALL SELECT items AS item FROM plain_variant_target) UNION ALL SELECT NULL", + ] { + let state = mock_session_state().with_scalar_function(Arc::new(make_udf( + "to_object", + vec![DataType::Utf8], + DataType::Utf8, + ))); + let plan = logical_plan_from_state( + sql, + &SnowflakeDialect {}, + ParserOptions::default(), + state, + )?; + assert_contains!(plan.display_indent().to_string(), "to_object("); + } + Ok(()) +} + +#[test] +fn plan_insert_rejects_text_outside_object_variant_union() { + let error = logical_plan( + "INSERT INTO untyped_object_target SELECT item FROM (SELECT items AS item FROM untyped_object_target UNION ALL SELECT items AS item FROM plain_variant_target) UNION ALL SELECT '{\"z\":3}'", + ) + .expect_err("an outer text branch must not inherit the inner OBJECT/VARIANT conversion"); + assert_contains!( + error.to_string(), + "Cannot implicitly convert Utf8 to OBJECT" + ); +} + #[test] fn plan_insert_keeps_untyped_object_fast_path() -> Result<()> { let plan = logical_plan( @@ -904,14 +939,14 @@ fn plan_insert_rejects_plain_text_into_untyped_object() { "INSERT INTO untyped_object_target SELECT items FROM plain_string_source", ) .expect_err("plain text needs an explicit OBJECT conversion"); - assert_contains!(err.to_string(), "Cannot implicitly convert text to OBJECT"); + assert_contains!(err.to_string(), "Cannot implicitly convert Utf8 to OBJECT"); } #[test] fn plan_insert_values_rejects_plain_text_into_untyped_object() { let err = logical_plan("INSERT INTO untyped_object_target VALUES ('{\"a\":1}')") .expect_err("plain text VALUES needs an explicit OBJECT conversion"); - assert_contains!(err.to_string(), "Cannot implicitly convert text to OBJECT"); + assert_contains!(err.to_string(), "Cannot implicitly convert Utf8 to OBJECT"); } #[test] @@ -919,7 +954,7 @@ fn plan_insert_values_rejects_mixed_null_and_plain_text_into_untyped_object() { let err = logical_plan("INSERT INTO untyped_object_target VALUES (NULL), ('{\"a\":1}')") .expect_err("a mixed VALUES list still needs explicit OBJECT conversion"); - assert_contains!(err.to_string(), "Cannot implicitly convert text to OBJECT"); + assert_contains!(err.to_string(), "Cannot implicitly convert Utf8 to OBJECT"); } #[test] @@ -928,6 +963,215 @@ fn plan_insert_values_accepts_null_for_untyped_object() -> Result<()> { Ok(()) } +#[test] +fn plan_insert_select_accepts_null_for_untyped_object() -> Result<()> { + logical_plan("INSERT INTO untyped_object_target SELECT NULL")?; + Ok(()) +} + +#[test] +fn plan_insert_rejects_numeric_and_boolean_for_untyped_object() { + for sql in [ + "INSERT INTO untyped_object_target SELECT 42", + "INSERT INTO untyped_object_target SELECT TRUE", + "INSERT INTO untyped_object_target VALUES (42)", + "INSERT INTO untyped_object_target VALUES (TRUE)", + ] { + let err = + logical_plan(sql).expect_err("scalar needs an explicit OBJECT conversion"); + assert_contains!(err.to_string(), "to OBJECT during INSERT"); + } +} + +#[test] +fn plan_insert_values_validates_bound_object_source() -> Result<()> { + let plan = logical_plan_from_state( + "INSERT INTO untyped_object_target VALUES ($1)", + &SnowflakeDialect {}, + ParserOptions::default(), + mock_session_state(), + )?; + for value in [ + ScalarValue::Utf8(Some("{\"a\":1}".to_string())), + ScalarValue::Int64(Some(42)), + ScalarValue::Boolean(Some(true)), + ] { + let err = plan + .clone() + .with_param_values(vec![value]) + .expect_err("bound scalar needs an explicit OBJECT conversion"); + assert_contains!(err.to_string(), "to OBJECT; use TO_OBJECT"); + } + plan.clone().with_param_values(vec![ScalarValue::Null])?; + plan.with_param_values(vec![ScalarValue::Utf8(None)])?; + Ok(()) +} + +#[test] +fn plan_insert_select_rejects_direct_object_placeholder() { + for sql in [ + "INSERT INTO untyped_object_target SELECT $1", + "INSERT INTO untyped_object_target SELECT $1 LIMIT 1", + "INSERT INTO untyped_object_target SELECT $1 AS item ORDER BY item LIMIT 1", + "INSERT INTO untyped_object_target SELECT item FROM (SELECT $1 AS item)", + "INSERT INTO untyped_object_target SELECT v.item FROM (SELECT $1 AS item) v CROSS JOIN (SELECT 1) x", + "INSERT INTO untyped_object_target SELECT v.item FROM (SELECT $1 AS item) v GROUP BY v.item", + "INSERT INTO untyped_object_target SELECT (SELECT $1)", + "INSERT INTO untyped_object_target SELECT COALESCE($1, NULL)", + ] { + let error = logical_plan_from_state( + sql, + &SnowflakeDialect {}, + ParserOptions::default(), + mock_session_state() + .with_scalar_function(Arc::new(df_core::coalesce().as_ref().clone())), + ) + .expect_err("bound SELECT cannot infer OBJECT"); + assert_contains!(error.to_string(), "untyped bound parameter into OBJECT"); + } +} + +#[test] +fn plan_insert_select_allows_placeholder_outside_object_output() -> Result<()> { + for sql in [ + "INSERT INTO untyped_object_target SELECT NULL WHERE $1", + "INSERT INTO untyped_object_target SELECT CASE WHEN $1 THEN NULL ELSE NULL END", + "INSERT INTO untyped_object_target VALUES (CASE WHEN $1 THEN NULL ELSE NULL END)", + ] { + logical_plan_from_state( + sql, + &SnowflakeDialect {}, + ParserOptions::default(), + mock_session_state(), + )?; + } + Ok(()) +} + +#[test] +fn plan_insert_values_rejects_nested_object_placeholder() { + let error = logical_plan_from_state( + "INSERT INTO untyped_object_target VALUES (COALESCE($1, NULL))", + &SnowflakeDialect {}, + ParserOptions::default(), + mock_session_state() + .with_scalar_function(Arc::new(df_core::coalesce().as_ref().clone())), + ) + .expect_err("nested bound VALUES expression cannot infer OBJECT"); + assert_contains!(error.to_string(), "untyped bound expression into OBJECT"); +} + +#[test] +fn plan_insert_select_from_values_checks_bound_object_value() -> Result<()> { + for sql in [ + "INSERT INTO untyped_object_target SELECT column1 FROM (VALUES ($1)) AS v(column1)", + "INSERT INTO untyped_object_target SELECT DISTINCT column1 FROM (VALUES ($1)) AS v(column1)", + "INSERT INTO untyped_object_target SELECT column1 FROM (VALUES ($1)) AS v(column1) UNION ALL SELECT NULL", + ] { + let plan = logical_plan_from_state( + sql, + &SnowflakeDialect {}, + ParserOptions::default(), + mock_session_state(), + )?; + let error = plan + .clone() + .with_param_values(vec![ScalarValue::Utf8(Some("{\"a\":1}".into()))]) + .expect_err("bound text cannot be inserted into OBJECT"); + assert_contains!( + error.to_string(), + "Cannot implicitly convert Utf8 to OBJECT" + ); + plan.clone().with_param_values(vec![ScalarValue::Null])?; + plan.with_param_values(vec![ScalarValue::Utf8(None)])?; + } + Ok(()) +} + +#[test] +fn plan_insert_object_union_checks_bound_scalar_branch() -> Result<()> { + for sql in [ + "INSERT INTO untyped_object_target SELECT items FROM untyped_object_target UNION ALL SELECT $1", + "INSERT INTO untyped_object_target SELECT item FROM (SELECT items AS item FROM untyped_object_target UNION ALL SELECT $1 AS item)", + ] { + let plan = logical_plan_from_state( + sql, + &SnowflakeDialect {}, + ParserOptions::default(), + mock_session_state(), + )?; + let error = plan + .clone() + .with_param_values(vec![ScalarValue::Utf8(Some("{\"a\":1}".into()))]) + .expect_err("a TEXT bind cannot enter an OBJECT union branch"); + assert_contains!(error.to_string(), "inconsistent data type"); + plan.with_param_values(vec![ScalarValue::Null])?; + } + Ok(()) +} + +#[test] +fn plan_select_object_union_checks_bound_scalar_branch() -> Result<()> { + let plan = logical_plan_from_state( + "SELECT items FROM untyped_object_target UNION ALL SELECT $1", + &SnowflakeDialect {}, + ParserOptions::default(), + mock_session_state(), + )?; + for value in [ + ScalarValue::Utf8(Some("{\"a\":1}".into())), + ScalarValue::Utf8(None), + ] { + let error = plan + .clone() + .with_param_values(vec![value]) + .expect_err("a VARCHAR bind cannot enter an OBJECT union branch"); + assert_contains!(error.to_string(), "inconsistent data type"); + } + plan.with_param_values(vec![ScalarValue::Null])?; + Ok(()) +} + +#[test] +fn plan_insert_object_accepts_wide_null_union() -> Result<()> { + let source = std::iter::repeat_n("SELECT NULL", 65) + .collect::>() + .join(" UNION ALL "); + logical_plan(&format!("INSERT INTO untyped_object_target {source}"))?; + Ok(()) +} + +#[test] +fn plan_set_operation_rejects_object_and_plain_scalar() { + for sql in [ + "SELECT items FROM untyped_object_target UNION ALL SELECT items FROM plain_string_source", + "SELECT items FROM plain_string_source UNION ALL SELECT items FROM untyped_object_target", + "SELECT items FROM untyped_object_target UNION ALL SELECT 42", + "SELECT items FROM untyped_object_target UNION ALL SELECT CAST(NULL AS VARCHAR)", + "SELECT items FROM untyped_object_target UNION ALL BY NAME SELECT items FROM plain_string_source", + "INSERT INTO untyped_object_target SELECT items FROM untyped_object_target UNION ALL SELECT items FROM plain_string_source", + ] { + let err = logical_plan(sql) + .expect_err("OBJECT and a scalar must not share a set column"); + assert_contains!(err.to_string(), "inconsistent data type"); + } +} + +#[test] +fn plan_set_operation_accepts_object_and_null() -> Result<()> { + logical_plan("SELECT items FROM untyped_object_target UNION ALL SELECT NULL")?; + logical_plan("SELECT NULL UNION ALL SELECT items FROM untyped_object_target")?; + Ok(()) +} + +#[test] +fn plan_set_operation_accepts_object_and_variant() -> Result<()> { + logical_plan( + "SELECT items FROM untyped_object_target UNION ALL SELECT items FROM plain_variant_target", + )?; + Ok(()) +} + #[test] fn plan_insert_uses_explicit_array_conversion_for_untagged_text() -> Result<()> { let state = mock_session_state().with_scalar_function(Arc::new(make_udf(