From 2606512a9e9972e92eca11f9c420224bdd6b746a Mon Sep 17 00:00:00 2001 From: xuanyili Date: Mon, 17 Aug 2026 07:09:12 +0000 Subject: [PATCH 1/4] fix: preserve MERGE target qualifier bindings Keep SQL-visible target qualifiers distinct from provider identity so correlated subqueries and source-name collisions retain their intended meaning. --- CONTEXT.md | 20 ++ datafusion/core/src/physical_planner.rs | 16 +- datafusion/core/tests/sql/sql_api.rs | 293 ++++++++++++++++-- datafusion/expr/src/logical_plan/dml.rs | 125 +++++++- .../expr/src/logical_plan/invariants.rs | 9 +- .../src/analyzer/function_rewrite.rs | 22 +- .../optimizer/src/analyzer/type_coercion.rs | 25 +- .../optimizer/src/rewrite_set_comparison.rs | 16 +- .../simplify_expressions/simplify_exprs.rs | 19 +- datafusion/optimizer/src/utils.rs | 13 +- .../proto-models/proto/datafusion.proto | 5 +- .../proto-models/src/generated/pbjson.rs | 18 ++ .../proto-models/src/generated/prost.rs | 6 +- .../proto/src/logical_plan/from_proto.rs | 17 +- datafusion/proto/src/logical_plan/to_proto.rs | 3 + .../tests/cases/roundtrip_logical_plan.rs | 44 ++- datafusion/session/src/table.rs | 6 +- datafusion/sql/src/statement.rs | 131 +------- datafusion/sql/tests/sql_integration.rs | 8 +- .../sqllogictest/test_files/merge_into.slt | 10 +- .../0001-preserve-merge-target-qualifier.md | 21 ++ .../library-user-guide/upgrading/55.0.0.md | 17 + 22 files changed, 593 insertions(+), 251 deletions(-) create mode 100644 CONTEXT.md create mode 100644 docs/adr/0001-preserve-merge-target-qualifier.md diff --git a/CONTEXT.md b/CONTEXT.md new file mode 100644 index 0000000000000..a003520224314 --- /dev/null +++ b/CONTEXT.md @@ -0,0 +1,20 @@ +# DataFusion SQL Name Resolution + +This context distinguishes catalog identity, SQL-visible names, and lexical bindings +while planning DataFusion statements. + +## Language + +**Provider identity**: +The catalog-resolved table reference used to locate a target `TableProvider`. +_Avoid_: Target qualifier, target alias + +**Visible qualifier**: +The normalized SQL relation name exposed in one lexical scope, such as `t` after +`target AS t`. The same text may name different relations in nested scopes. +_Avoid_: Relation identity, binding identity + +**Relation binding**: +The particular relation occurrence selected by lexical name resolution. A relation +binding is distinct from both its provider identity and its visible qualifier. +_Avoid_: Qualifier string diff --git a/datafusion/core/src/physical_planner.rs b/datafusion/core/src/physical_planner.rs index 2792c9c7a6faa..641aee4634b8b 100644 --- a/datafusion/core/src/physical_planner.rs +++ b/datafusion/core/src/physical_planner.rs @@ -869,11 +869,9 @@ impl DefaultPhysicalPlanner { e.context(format!("MERGE INTO operation on table '{table_name}'")) })?; let input_exec = children.one()?; - let target_schema = DFSchema::try_from_qualified_schema( - table_name.clone(), - &target.schema(), - )?; - let merge_schema = Arc::new(target_schema.join(input.schema())?); + let merge_schema = Arc::new( + merge_op.expression_schema(&target.schema(), input.schema())?, + ); provider .merge_into( session_state, @@ -3571,8 +3569,8 @@ mod tests { ctx.register_table("source", source)?; ctx.sql( - "MERGE INTO target AS t USING source AS s ON t.id = s.id \ - WHEN MATCHED AND t.id > s.id THEN DELETE", + "MERGE INTO target AS t USING source AS target ON t.id = target.id \ + WHEN MATCHED AND t.id > target.id THEN DELETE", ) .await? .create_physical_plan() @@ -3583,11 +3581,11 @@ mod tests { captured.as_ref().expect("merge_into should be called"); assert_eq!(*clause_count, 1); assert_eq!( - merge_schema.index_of_column(&Column::new(Some("target"), "id"))?, + merge_schema.index_of_column(&Column::new(Some("t"), "id"))?, 0 ); assert_eq!( - merge_schema.index_of_column(&Column::new(Some("s"), "id"))?, + merge_schema.index_of_column(&Column::new(Some("target"), "id"))?, 1 ); assert_contains!(physical_on, "index: 0"); diff --git a/datafusion/core/tests/sql/sql_api.rs b/datafusion/core/tests/sql/sql_api.rs index ca18406a8e40d..43bed215bef6e 100644 --- a/datafusion/core/tests/sql/sql_api.rs +++ b/datafusion/core/tests/sql/sql_api.rs @@ -16,7 +16,10 @@ // under the License. use datafusion::prelude::*; -use datafusion_common::assert_contains; +use datafusion_common::tree_node::{TreeNode, TreeNodeRecursion}; +use datafusion_common::{TableReference, assert_contains}; +use datafusion_expr::dml::MergeIntoOp; +use datafusion_expr::{Expr, LogicalPlan, WriteOp}; use tempfile::TempDir; @@ -212,72 +215,306 @@ async fn merge_into_context() -> SessionContext { let ctx = SessionContext::new(); ctx.sql("CREATE TABLE target (id INT)").await.unwrap(); ctx.sql("CREATE TABLE source (id INT)").await.unwrap(); + ctx.sql("CREATE TABLE \"Target\" (id INT)").await.unwrap(); + ctx.sql("CREATE SCHEMA \"CaseSchema\"").await.unwrap(); + ctx.sql("CREATE TABLE \"CaseSchema\".\"Target\" (id INT)") + .await + .unwrap(); ctx } -async fn assert_merge_sql_error(ctx: &SessionContext, sql: &str, expected: &str) { - let err = ctx.sql(sql).await.unwrap_err(); - assert_contains!(err.strip_backtrace(), expected); -} - async fn assert_merge_physical_error(ctx: &SessionContext, sql: &str, expected: &str) { - let err = ctx + let result = ctx .sql(sql) .await - .unwrap() + .unwrap_or_else(|error| panic!("failed to plan MERGE SQL:\n{sql}\n{error}")) .create_physical_plan() - .await - .unwrap_err(); - assert_contains!(err.strip_backtrace(), expected); + .await; + let err = match result { + Ok(_) => panic!("expected physical planning to fail:\n{sql}"), + Err(error) => error, + }; + let actual = err.strip_backtrace(); + assert!( + actual.contains(expected), + "MERGE SQL:\n{sql}\n\nExpected:\n{expected}\n\nActual:\n{actual}" + ); +} + +async fn merge_operation(ctx: &SessionContext, sql: &str) -> Box { + let plan = ctx.state().create_logical_plan(sql).await.unwrap(); + let LogicalPlan::Dml(dml) = plan else { + panic!("expected MERGE DML") + }; + let WriteOp::MergeInto(merge_op) = dml.op else { + panic!("expected MERGE operation") + }; + merge_op +} + +fn has_outer_reference_to(expr: &Expr, qualifier: &TableReference) -> bool { + let mut found = false; + expr.apply(|expr| { + let outer_refs = match expr { + Expr::Exists(exists) => Some(&exists.subquery.outer_ref_columns), + Expr::InSubquery(in_subquery) => { + Some(&in_subquery.subquery.outer_ref_columns) + } + Expr::SetComparison(set_comparison) => { + Some(&set_comparison.subquery.outer_ref_columns) + } + Expr::ScalarSubquery(subquery) => Some(&subquery.outer_ref_columns), + _ => None, + }; + found = outer_refs.is_some_and(|outer_refs| { + outer_refs.iter().any(|expr| { + matches!( + expr, + Expr::OuterReferenceColumn(_, column) + if column.relation.as_ref() == Some(qualifier) + ) + }) + }); + Ok(if found { + TreeNodeRecursion::Stop + } else { + TreeNodeRecursion::Continue + }) + }) + .unwrap(); + found } #[tokio::test] -async fn merge_into_rejects_source_alias_colliding_with_target_name() { - // Canonicalizing `t.id` to `target.id` must not collapse it onto a source - // that also uses `target` as its qualifier. +async fn merge_into_distinguishes_target_alias_from_source_qualifier() { let ctx = merge_into_context().await; - for target_ref in ["target", "public.target", "datafusion.public.target"] { - assert_merge_sql_error( + for target_ref in [ + "target", + "public.target", + "datafusion.public.target", + "\"Target\"", + "\"CaseSchema\".\"Target\"", + ] { + assert_merge_physical_error( &ctx, &format!( "MERGE INTO {target_ref} AS t USING source AS target \ ON t.id = target.id WHEN MATCHED THEN DELETE" ), + "MERGE INTO not supported for Base table", + ) + .await; + } + + for (target_alias, target_column) in [ + ("T", "t.id"), + ("\"T\"", "\"T\".id"), + ("\"public.target\"", "\"public.target\".id"), + ] { + assert_merge_physical_error( + &ctx, &format!( - "MERGE source may not use the target table name '{target_ref}' \ - as a qualifier" + "MERGE INTO target AS {target_alias} USING source AS target \ + ON {target_column} = target.id WHEN MATCHED THEN DELETE" ), + "MERGE INTO not supported for Base table", + ) + .await; + } + + for source in [ + "(SELECT id FROM source) AS target", + "target AS source_target", + ] { + assert_merge_physical_error( + &ctx, + &format!( + "MERGE INTO target AS t USING {source} \ + ON true WHEN MATCHED THEN DELETE" + ), + "MERGE INTO not supported for Base table", + ) + .await; + } + + for (target_ref, target_column, expected_qualifier) in [ + ( + "\"Target\"", + "\"Target\".id", + TableReference::bare("Target"), + ), + ( + "\"CaseSchema\".\"Target\"", + "\"CaseSchema\".\"Target\".id", + TableReference::partial("CaseSchema", "Target"), + ), + ] { + let sql = format!( + "MERGE INTO {target_ref} USING source AS s \ + ON {target_column} = s.id WHEN MATCHED THEN DELETE" + ); + let merge_op = merge_operation(&ctx, &sql).await; + assert_eq!(merge_op.target_qualifier(), &expected_qualifier); + assert_merge_physical_error( + &ctx, + &sql, + "MERGE INTO not supported for Base table", ) .await; } } #[tokio::test] -async fn merge_into_rejects_subqueries_correlated_to_target_alias() { +async fn merge_into_rejects_outer_source_qualifier_matching_target_alias() { let ctx = merge_into_context().await; - assert_merge_sql_error( - &ctx, - "MERGE INTO target AS t USING source AS s \ + for (target_alias, source_alias, expected) in [ + ("t", "t", "target qualifier 't'"), + ("t", "T", "target qualifier 't'"), + ("\"T\"", "\"T\"", "target qualifier 'T'"), + ] { + let err = ctx + .sql(&format!( + "MERGE INTO target AS {target_alias} USING source AS {source_alias} \ + ON true WHEN MATCHED THEN DELETE" + )) + .await + .unwrap_err(); + assert_contains!(err.strip_backtrace(), expected); + assert_contains!(err.strip_backtrace(), "conflicts with a source qualifier"); + } + + let err = ctx + .sql( + "MERGE INTO target AS t \ + USING (SELECT id AS source_id FROM source) AS t \ + ON true WHEN MATCHED THEN DELETE", + ) + .await + .unwrap_err(); + assert_contains!( + err.strip_backtrace(), + "MERGE target qualifier 't' conflicts with a source qualifier" + ); + + for target_ref in ["public.target", "datafusion.public.target"] { + let err = ctx + .sql(&format!( + "MERGE INTO {target_ref} USING source AS target \ + ON true WHEN MATCHED THEN DELETE" + )) + .await + .unwrap_err(); + assert_contains!(err.strip_backtrace(), "conflicts with a source qualifier"); + } +} + +#[tokio::test] +async fn merge_into_preserves_target_alias_in_correlated_subquery() { + let ctx = merge_into_context().await; + let direct_exists = "MERGE INTO target AS t USING source AS s \ ON EXISTS (SELECT 1 FROM source AS x WHERE x.id = t.id) \ - WHEN MATCHED THEN DELETE", - "MERGE subqueries correlated to target alias 't' are not supported", - ) - .await; + WHEN MATCHED THEN DELETE"; + let direct_in = "MERGE INTO target AS t USING source AS s \ + ON t.id IN (SELECT x.id FROM source AS x WHERE x.id = t.id) \ + WHEN MATCHED THEN DELETE"; + let direct_any = "MERGE INTO target AS t USING source AS s \ + ON t.id = ANY (SELECT x.id FROM source AS x WHERE x.id = t.id) \ + WHEN MATCHED THEN DELETE"; + let direct_all = "MERGE INTO target AS t USING source AS s \ + ON t.id = ALL (SELECT x.id FROM source AS x WHERE x.id = t.id) \ + WHEN MATCHED THEN DELETE"; + let direct_scalar = "MERGE INTO target AS t USING source AS s \ + ON t.id = (SELECT max(x.id) FROM source AS x WHERE x.id = t.id) \ + WHEN MATCHED THEN DELETE"; - // Source-correlated and uncorrelated subqueries remain supported through - // logical optimization. for sql in [ + direct_exists, + direct_in, + direct_any, + direct_all, + direct_scalar, + ] { + let merge_op = merge_operation(&ctx, sql).await; + assert_eq!(merge_op.target_qualifier(), &TableReference::bare("t")); + assert!(has_outer_reference_to( + &merge_op.on, + &TableReference::bare("t") + )); + } + + let shadowed_correlation = "MERGE INTO target AS t USING source AS s \ + ON EXISTS (SELECT 1 FROM source AS t \ + WHERE EXISTS (SELECT 1 FROM source AS x WHERE x.id = t.id)) \ + WHEN MATCHED THEN DELETE"; + let merge_op = merge_operation(&ctx, shadowed_correlation).await; + assert!(!has_outer_reference_to( + &merge_op.on, + &TableReference::bare("t") + )); + + for sql in [ + // Direct and deeply nested target correlation. + direct_exists, + direct_in, + direct_any, + direct_all, + direct_scalar, + "MERGE INTO target AS t USING source AS s \ + ON EXISTS (SELECT 1 FROM source AS q \ + WHERE EXISTS (SELECT 1 FROM source AS x WHERE x.id = t.id)) \ + WHEN MATCHED THEN DELETE", + "MERGE INTO target AS t USING source AS s \ + ON EXISTS (SELECT 1 FROM source AS q \ + CROSS JOIN LATERAL (SELECT t.id) AS l) \ + WHEN MATCHED THEN DELETE", + "MERGE INTO target AS t USING source AS s \ + ON EXISTS (SELECT 1 FROM source AS q CROSS JOIN (\ + source AS x CROSS JOIN LATERAL (SELECT t.id) AS l\ + )) WHEN MATCHED THEN DELETE", + // Source correlation, no correlation, and nested shadowing. "MERGE INTO target AS t USING source AS s \ ON EXISTS (SELECT 1 FROM source AS x WHERE x.id = s.id) \ WHEN MATCHED THEN DELETE", "MERGE INTO target AS t USING source AS s \ ON t.id = ANY (SELECT id FROM source) \ WHEN MATCHED THEN DELETE", + shadowed_correlation, + "MERGE INTO target AS t USING source AS s \ + ON EXISTS (SELECT EXISTS (\ + SELECT 1 FROM source AS x WHERE x.id = t.id\ + ) FROM source AS t) \ + WHEN MATCHED THEN DELETE", + "MERGE INTO target AS t USING source AS s \ + ON EXISTS (SELECT 1 FROM source AS t \ + WHERE EXISTS (SELECT 1 FROM source AS q \ + WHERE EXISTS (SELECT 1 FROM source AS x WHERE x.id = t.id))) \ + WHEN MATCHED THEN DELETE", + "MERGE INTO target AS t USING source AS s \ + ON EXISTS (SELECT 1 FROM source AS t \ + CROSS JOIN LATERAL (SELECT t.id) AS l) \ + WHEN MATCHED THEN DELETE", + "MERGE INTO target AS t USING source AS s \ + ON EXISTS (SELECT 1 FROM source AS t CROSS JOIN (\ + source AS x CROSS JOIN LATERAL (SELECT t.id) AS l\ + )) WHEN MATCHED THEN DELETE", ] { assert_merge_physical_error(&ctx, sql, "MERGE INTO not supported for Base table") .await; } + + // LIMIT expressions intentionally use an empty local scope. The nested + // `t.id` therefore resolves to the MERGE target rather than `source AS t`, + // and reaches the existing generic scalar-subquery placement restriction. + assert_merge_physical_error( + &ctx, + "MERGE INTO target AS t USING source AS s \ + ON EXISTS (SELECT t.id FROM source AS t \ + LIMIT (SELECT t.id FROM source AS x LIMIT 1)) \ + WHEN MATCHED THEN DELETE", + "Correlated scalar subquery can only be used", + ) + .await; } #[tokio::test] diff --git a/datafusion/expr/src/logical_plan/dml.rs b/datafusion/expr/src/logical_plan/dml.rs index 7717dfaff7a33..261faf3226f6d 100644 --- a/datafusion/expr/src/logical_plan/dml.rs +++ b/datafusion/expr/src/logical_plan/dml.rs @@ -23,7 +23,9 @@ use std::sync::Arc; use arrow::datatypes::{DataType, Field, Schema}; use datafusion_common::file_options::file_type::FileType; -use datafusion_common::{DFSchemaRef, Result, TableReference, internal_err}; +use datafusion_common::{ + DFSchema, DFSchemaRef, Result, TableReference, internal_err, plan_err, +}; use crate::{Expr, LogicalPlan, TableSource}; @@ -206,6 +208,16 @@ impl DmlStatement { pub fn name(&self) -> &str { self.op.name() } + + /// Build the target-plus-source schema used by MERGE expressions. + pub fn merge_schema(&self) -> Result { + let WriteOp::MergeInto(merge_op) = &self.op else { + return internal_err!( + "DmlStatement::merge_schema requires a MERGE operation" + ); + }; + merge_op.expression_schema(&self.target.schema(), self.input.schema()) + } } // Manual implementation needed because of `table_schema` and `output_schema` fields. @@ -299,8 +311,15 @@ impl Display for InsertOp { } /// Describes a MERGE INTO operation's parameters. +/// +/// [`Self::target_qualifier`] is the SQL-visible relation name used by +/// expressions. The target's catalog/provider identity remains in +/// [`DmlStatement::table_name`]. #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Hash)] +#[non_exhaustive] pub struct MergeIntoOp { + /// The target relation name visible to expressions in the MERGE scope. + target_qualifier: TableReference, /// The join condition from `ON `. pub on: Expr, /// The WHEN clauses, in the order they appeared in the SQL. @@ -308,6 +327,55 @@ pub struct MergeIntoOp { } impl MergeIntoOp { + /// Create a MERGE operation. + pub fn new( + target_qualifier: impl Into, + on: Expr, + clauses: Vec, + ) -> Self { + Self { + target_qualifier: target_qualifier.into(), + on, + clauses, + } + } + + /// Return the target relation name visible to MERGE expressions. + pub fn target_qualifier(&self) -> &TableReference { + &self.target_qualifier + } + + /// Build the schema used to resolve expressions owned by this operation. + /// + /// Target fields precede source fields. The visible target qualifier must + /// not also identify a source relation in the outer MERGE scope. + pub fn expression_schema( + &self, + target_schema: &Schema, + source_schema: &DFSchema, + ) -> Result { + Self::expression_schema_for(&self.target_qualifier, target_schema, source_schema) + } + + /// Build a MERGE expression schema before constructing the operation. + pub fn expression_schema_for( + target_qualifier: &TableReference, + target_schema: &Schema, + source_schema: &DFSchema, + ) -> Result { + if source_schema.iter().any(|(qualifier, _)| { + qualifier.is_some_and(|qualifier| qualifier.resolved_eq(target_qualifier)) + }) { + return plan_err!( + "MERGE target qualifier '{}' conflicts with a source qualifier", + target_qualifier + ); + } + + DFSchema::try_from_qualified_schema(target_qualifier.clone(), target_schema)? + .join(source_schema) + } + /// Count of top-level [`Expr`]s owned by this operation (no allocation). /// /// Matches the length of [`Self::exprs`] and the `exprs` vec consumed by @@ -403,7 +471,11 @@ impl MergeIntoOp { } }) .collect(); - Ok(Self { on, clauses }) + Ok(Self { + target_qualifier: self.target_qualifier.clone(), + on, + clauses, + }) } } @@ -503,9 +575,10 @@ mod tests { #[test] fn write_op_merge_into_name_and_display() { - let op = WriteOp::MergeInto(Box::new(MergeIntoOp { - on: col("id").eq(col("source_id")), - clauses: vec![MergeIntoClause { + let op = WriteOp::MergeInto(Box::new(MergeIntoOp::new( + "target", + col("id").eq(col("source_id")), + vec![MergeIntoClause { kind: MergeIntoClauseKind::Matched, predicate: Some(col("qty").gt(lit(0_i64))), action: MergeIntoAction::Update(vec![( @@ -513,7 +586,7 @@ mod tests { col("source_qty"), )]), }], - })); + ))); assert_eq!(op.name(), "MergeInto"); assert_eq!(format!("{op}"), "MergeInto"); } @@ -548,9 +621,10 @@ mod tests { #[test] fn merge_into_op_exprs_round_trip() { - let op = MergeIntoOp { - on: col("id").eq(col("source_id")), - clauses: vec![ + let op = MergeIntoOp::new( + "target", + col("id").eq(col("source_id")), + vec![ MergeIntoClause { kind: MergeIntoClauseKind::Matched, predicate: Some(col("qty").gt(lit(0_i64))), @@ -573,7 +647,7 @@ mod tests { action: MergeIntoAction::Delete, }, ], - }; + ); let exprs = op.exprs(); assert_eq!(exprs.len(), 7); @@ -584,14 +658,37 @@ mod tests { #[test] fn merge_into_op_with_new_exprs_length_mismatch() { - let op = MergeIntoOp { - on: col("id").eq(col("source_id")), - clauses: vec![], - }; + let op = MergeIntoOp::new("target", col("id").eq(col("source_id")), vec![]); let err = op.with_new_exprs(vec![]).unwrap_err(); assert!( err.to_string().contains("expected 1 expressions, got 0"), "unexpected error: {err}" ); } + + #[test] + fn merge_into_schema_uses_visible_qualifier_as_outer_scope_binding() -> Result<()> { + let target_schema = Schema::new(vec![Field::new("id", DataType::Int32, false)]); + let source_schema = + Schema::new(vec![Field::new("source_id", DataType::Int32, false)]); + let op = MergeIntoOp::new("t", col("t.id").eq(col("target.source_id")), vec![]); + + let source = DFSchema::try_from_qualified_schema("target", &source_schema)?; + let schema = op.expression_schema(&target_schema, &source)?; + assert!(schema.has_column(&datafusion_common::Column::new(Some("t"), "id"))); + assert!( + schema + .has_column(&datafusion_common::Column::new(Some("target"), "source_id")) + ); + + let colliding_source = DFSchema::try_from_qualified_schema("t", &source_schema)?; + let err = op + .expression_schema(&target_schema, &colliding_source) + .unwrap_err(); + assert!( + err.to_string() + .contains("target qualifier 't' conflicts with a source qualifier") + ); + Ok(()) + } } diff --git a/datafusion/expr/src/logical_plan/invariants.rs b/datafusion/expr/src/logical_plan/invariants.rs index f36653694c21d..1041d2df8f444 100644 --- a/datafusion/expr/src/logical_plan/invariants.rs +++ b/datafusion/expr/src/logical_plan/invariants.rs @@ -195,7 +195,12 @@ pub fn check_subquery_expr( } }?; match outer_plan { - LogicalPlan::Projection(_) | LogicalPlan::Filter(_) => Ok(()), + LogicalPlan::Projection(_) + | LogicalPlan::Filter(_) + | LogicalPlan::Dml(DmlStatement { + op: WriteOp::MergeInto(_), + .. + }) => Ok(()), LogicalPlan::Aggregate(Aggregate { group_expr, aggr_expr, @@ -213,7 +218,7 @@ pub fn check_subquery_expr( } _ => plan_err!( "Correlated scalar subquery can only be used in Projection, \ - Filter, Aggregate plan nodes" + Filter, Aggregate and MERGE DML plan nodes" ), }?; } diff --git a/datafusion/optimizer/src/analyzer/function_rewrite.rs b/datafusion/optimizer/src/analyzer/function_rewrite.rs index a66e3ccc0cf8a..9e40bb066e003 100644 --- a/datafusion/optimizer/src/analyzer/function_rewrite.rs +++ b/datafusion/optimizer/src/analyzer/function_rewrite.rs @@ -22,10 +22,10 @@ use datafusion_common::config::ConfigOptions; use datafusion_common::tree_node::{Transformed, TreeNode}; use datafusion_common::{DFSchema, Result}; -use crate::utils::NamePreserver; +use crate::utils::{NamePreserver, merge_into_schema}; +use datafusion_expr::LogicalPlan; use datafusion_expr::expr_rewriter::FunctionRewrite; use datafusion_expr::utils::merge_schema; -use datafusion_expr::{DmlStatement, LogicalPlan, WriteOp}; use std::sync::Arc; /// Analyzer rule that invokes [`FunctionRewrite`]s on expressions @@ -59,20 +59,10 @@ impl ApplyFunctionRewrites { } // MERGE expressions reference the target table, which is not one of - // `plan.inputs()`. Rebuild the target schema from the DML's - // `table_name` and `target` so those columns resolve. - if let LogicalPlan::Dml(DmlStatement { - op: WriteOp::MergeInto(_), - table_name, - target, - .. - }) = &plan - { - let target_schema = DFSchema::try_from_qualified_schema( - table_name.clone(), - &target.schema(), - )?; - schema.merge(&target_schema); + // `plan.inputs()`. Use the operation's visible qualifier when adding + // its target schema. + if let Some(merge_schema) = merge_into_schema(&plan)? { + schema = merge_schema; } let name_preserver = NamePreserver::new(&plan); diff --git a/datafusion/optimizer/src/analyzer/type_coercion.rs b/datafusion/optimizer/src/analyzer/type_coercion.rs index feadc0370bfd5..c3c787b2945e3 100644 --- a/datafusion/optimizer/src/analyzer/type_coercion.rs +++ b/datafusion/optimizer/src/analyzer/type_coercion.rs @@ -23,7 +23,7 @@ use itertools::{Itertools as _, izip}; use std::sync::{Arc, LazyLock}; use crate::analyzer::AnalyzerRule; -use crate::utils::NamePreserver; +use crate::utils::{NamePreserver, merge_into_schema}; use arrow::datatypes::{DataType, Field, IntervalUnit, Schema, TimeUnit}; use arrow::temporal_conversions::SECONDS_IN_DAY; @@ -130,18 +130,10 @@ fn analyze_internal( } // MERGE expressions (ON / WHEN clauses) reference the target table, which - // is not one of `plan.inputs()`. Rebuild the target schema from the DML's - // `table_name` and `target` so those columns resolve during coercion. - if let LogicalPlan::Dml(DmlStatement { - op: WriteOp::MergeInto(_), - table_name, - target, - .. - }) = &plan - { - let target_schema = - DFSchema::try_from_qualified_schema(table_name.clone(), &target.schema())?; - schema.merge(&target_schema); + // is not one of `plan.inputs()`. Use the operation's visible qualifier + // when adding its target schema. + if let Some(merge_schema) = merge_into_schema(&plan)? { + schema = merge_schema; } // merge the outer schema for correlated subqueries @@ -1648,9 +1640,10 @@ mod test { // target schema to be visible to the analyzer, which only sees // `plan.inputs()` (the source plan) by default. let on = col("target.id").eq(col("source.id")); - let merge_op = MergeIntoOp { + let merge_op = MergeIntoOp::new( + "target", on, - clauses: vec![ + vec![ MergeIntoClause { kind: MergeIntoClauseKind::Matched, predicate: None, @@ -1668,7 +1661,7 @@ mod test { }, }, ], - }; + ); let plan = LogicalPlan::Dml(DmlStatement::new( target_table_name, target_source, diff --git a/datafusion/optimizer/src/rewrite_set_comparison.rs b/datafusion/optimizer/src/rewrite_set_comparison.rs index 18712c5335205..a5af4b1978caf 100644 --- a/datafusion/optimizer/src/rewrite_set_comparison.rs +++ b/datafusion/optimizer/src/rewrite_set_comparison.rs @@ -19,13 +19,14 @@ //! `> ALL`) into boolean expressions built from `EXISTS` subqueries //! that capture SQL three-valued logic. +use crate::utils::merge_into_schema; use crate::{OptimizerConfig, OptimizerRule}; use datafusion_common::tree_node::{Transformed, TreeNode}; use datafusion_common::{Column, DFSchema, ExprSchema, Result, ScalarValue, plan_err}; use datafusion_expr::expr::{self, Exists, SetComparison, SetQuantifier}; use datafusion_expr::logical_plan::Subquery; use datafusion_expr::logical_plan::builder::LogicalPlanBuilder; -use datafusion_expr::{DmlStatement, Expr, LogicalPlan, WriteOp, lit}; +use datafusion_expr::{Expr, LogicalPlan, lit}; use std::sync::Arc; use datafusion_expr::utils::merge_schema; @@ -45,17 +46,8 @@ impl RewriteSetComparison { fn rewrite_plan(&self, plan: LogicalPlan) -> Result> { let mut schema = merge_schema(&plan.inputs()); - if let LogicalPlan::Dml(DmlStatement { - op: WriteOp::MergeInto(_), - table_name, - target, - .. - }) = &plan - { - schema.merge(&DFSchema::try_from_qualified_schema( - table_name.clone(), - &target.schema(), - )?); + if let Some(merge_schema) = merge_into_schema(&plan)? { + schema = merge_schema; } plan.map_expressions(|expr| { expr.transform_up(|expr| rewrite_set_comparison(expr, &schema)) diff --git a/datafusion/optimizer/src/simplify_expressions/simplify_exprs.rs b/datafusion/optimizer/src/simplify_expressions/simplify_exprs.rs index 1c5a4a1869ddb..41dd24b6d857f 100644 --- a/datafusion/optimizer/src/simplify_expressions/simplify_exprs.rs +++ b/datafusion/optimizer/src/simplify_expressions/simplify_exprs.rs @@ -21,17 +21,17 @@ use std::sync::Arc; use datafusion_common::tree_node::{Transformed, TreeNode}; use datafusion_common::{Column, DFSchema, DFSchemaRef, DataFusionError, Result}; +use datafusion_expr::Expr; use datafusion_expr::logical_plan::{Aggregate, LogicalPlan, Projection}; use datafusion_expr::simplify::SimplifyContext; use datafusion_expr::utils::{ columnize_expr, find_aggregate_exprs, grouping_set_to_exprlist, merge_schema, }; -use datafusion_expr::{DmlStatement, Expr, WriteOp}; use super::ExprSimplifier; use crate::optimizer::ApplyOrder; use crate::simplify_expressions::linear_aggregates::rewrite_multiple_linear_aggregates; -use crate::utils::NamePreserver; +use crate::utils::{NamePreserver, merge_into_schema}; use crate::{OptimizerConfig, OptimizerRule}; /// Optimizer Pass that simplifies [`LogicalPlan`]s by rewriting @@ -77,19 +77,8 @@ impl SimplifyExpressions { plan: LogicalPlan, config: &dyn OptimizerConfig, ) -> Result> { - let schema = if let LogicalPlan::Dml(DmlStatement { - op: WriteOp::MergeInto(_), - table_name, - target, - .. - }) = &plan - { - let mut schema = merge_schema(&plan.inputs()); - schema.merge(&DFSchema::try_from_qualified_schema( - table_name.clone(), - &target.schema(), - )?); - DFSchemaRef::new(schema) + let schema = if let Some(merge_schema) = merge_into_schema(&plan)? { + DFSchemaRef::new(merge_schema) } else if !plan.inputs().is_empty() { DFSchemaRef::new(merge_schema(&plan.inputs())) } else if let LogicalPlan::TableScan(scan) = &plan { diff --git a/datafusion/optimizer/src/utils.rs b/datafusion/optimizer/src/utils.rs index d4ac31e8a517c..d6a5480861c02 100644 --- a/datafusion/optimizer/src/utils.rs +++ b/datafusion/optimizer/src/utils.rs @@ -30,7 +30,7 @@ use datafusion_expr::execution_props::ExecutionProps; use datafusion_expr::expr::{Exists, InSubquery, SetComparison}; use datafusion_expr::expr_rewriter::replace_col; use datafusion_expr::physical_planning_context::PhysicalPlanningContext; -use datafusion_expr::{ColumnarValue, Expr, logical_plan::LogicalPlan}; +use datafusion_expr::{ColumnarValue, Expr, WriteOp, logical_plan::LogicalPlan}; use datafusion_physical_expr::create_physical_expr; use log::{debug, trace}; use std::sync::Arc; @@ -39,6 +39,17 @@ use std::sync::Arc; /// as it was initially placed here and then moved elsewhere. pub use datafusion_expr::expr_rewriter::NamePreserver; +/// Return the expression schema for a MERGE DML node. +pub(crate) fn merge_into_schema(plan: &LogicalPlan) -> Result> { + let LogicalPlan::Dml(dml) = plan else { + return Ok(None); + }; + let WriteOp::MergeInto(_) = &dml.op else { + return Ok(None); + }; + dml.merge_schema().map(Some) +} + /// Invokes `f` with the index, within `schema`, of every column referenced by /// `expr` — including columns reached through a correlated subquery's outer /// references. Columns absent from `schema` are skipped. diff --git a/datafusion/proto-models/proto/datafusion.proto b/datafusion/proto-models/proto/datafusion.proto index d98b67a66e0a9..71d4472958a76 100644 --- a/datafusion/proto-models/proto/datafusion.proto +++ b/datafusion/proto-models/proto/datafusion.proto @@ -332,10 +332,13 @@ message DmlNode{ MergeIntoOpNode merge_into = 6; } -// Carries the ON condition and WHEN clauses of a MERGE INTO operation. +// Carries the target qualifier, ON condition, and WHEN clauses of a MERGE INTO operation. message MergeIntoOpNode { LogicalExprNode on = 1; repeated MergeIntoClauseNode clauses = 2; + // SQL-visible target qualifier. Absent in payloads written before this field + // was introduced; readers then fall back to DmlNode.table_name. + TableReference target_qualifier = 3; } // A single WHEN clause within a MERGE INTO statement. diff --git a/datafusion/proto-models/src/generated/pbjson.rs b/datafusion/proto-models/src/generated/pbjson.rs index 21309bb2d0941..b64438cdd0c67 100644 --- a/datafusion/proto-models/src/generated/pbjson.rs +++ b/datafusion/proto-models/src/generated/pbjson.rs @@ -15231,6 +15231,9 @@ impl serde::Serialize for MergeIntoOpNode { if !self.clauses.is_empty() { len += 1; } + if self.target_qualifier.is_some() { + len += 1; + } let mut struct_ser = serializer.serialize_struct("datafusion.MergeIntoOpNode", len)?; if let Some(v) = self.on.as_ref() { struct_ser.serialize_field("on", v)?; @@ -15238,6 +15241,9 @@ impl serde::Serialize for MergeIntoOpNode { if !self.clauses.is_empty() { struct_ser.serialize_field("clauses", &self.clauses)?; } + if let Some(v) = self.target_qualifier.as_ref() { + struct_ser.serialize_field("targetQualifier", v)?; + } struct_ser.end() } } @@ -15250,12 +15256,15 @@ impl<'de> serde::Deserialize<'de> for MergeIntoOpNode { const FIELDS: &[&str] = &[ "on", "clauses", + "target_qualifier", + "targetQualifier", ]; #[allow(clippy::enum_variant_names)] enum GeneratedField { On, Clauses, + TargetQualifier, } impl<'de> serde::Deserialize<'de> for GeneratedField { fn deserialize(deserializer: D) -> std::result::Result @@ -15279,6 +15288,7 @@ impl<'de> serde::Deserialize<'de> for MergeIntoOpNode { match value { "on" => Ok(GeneratedField::On), "clauses" => Ok(GeneratedField::Clauses), + "targetQualifier" | "target_qualifier" => Ok(GeneratedField::TargetQualifier), _ => Err(serde::de::Error::unknown_field(value, FIELDS)), } } @@ -15300,6 +15310,7 @@ impl<'de> serde::Deserialize<'de> for MergeIntoOpNode { { let mut on__ = None; let mut clauses__ = None; + let mut target_qualifier__ = None; while let Some(k) = map_.next_key()? { match k { GeneratedField::On => { @@ -15314,11 +15325,18 @@ impl<'de> serde::Deserialize<'de> for MergeIntoOpNode { } clauses__ = Some(map_.next_value()?); } + GeneratedField::TargetQualifier => { + if target_qualifier__.is_some() { + return Err(serde::de::Error::duplicate_field("targetQualifier")); + } + target_qualifier__ = map_.next_value()?; + } } } Ok(MergeIntoOpNode { on: on__, clauses: clauses__.unwrap_or_default(), + target_qualifier: target_qualifier__, }) } } diff --git a/datafusion/proto-models/src/generated/prost.rs b/datafusion/proto-models/src/generated/prost.rs index d830624322e14..1862a57c30137 100644 --- a/datafusion/proto-models/src/generated/prost.rs +++ b/datafusion/proto-models/src/generated/prost.rs @@ -527,13 +527,17 @@ pub mod dml_node { } } } -/// Carries the ON condition and WHEN clauses of a MERGE INTO operation. +/// Carries the target qualifier, ON condition, and WHEN clauses of a MERGE INTO operation. #[derive(Clone, PartialEq, ::prost::Message)] pub struct MergeIntoOpNode { #[prost(message, optional, boxed, tag = "1")] pub on: ::core::option::Option<::prost::alloc::boxed::Box>, #[prost(message, repeated, tag = "2")] pub clauses: ::prost::alloc::vec::Vec, + /// SQL-visible target qualifier. Absent in payloads written before this field + /// was introduced; readers then fall back to DmlNode.table_name. + #[prost(message, optional, tag = "3")] + pub target_qualifier: ::core::option::Option, } /// A single WHEN clause within a MERGE INTO statement. #[derive(Clone, PartialEq, ::prost::Message)] diff --git a/datafusion/proto/src/logical_plan/from_proto.rs b/datafusion/proto/src/logical_plan/from_proto.rs index d4d0ea7292ffe..b97a85742c590 100644 --- a/datafusion/proto/src/logical_plan/from_proto.rs +++ b/datafusion/proto/src/logical_plan/from_proto.rs @@ -69,13 +69,26 @@ pub fn parse_write_op( .to_string(), ) })?; - WriteOp::MergeInto(Box::new(parse_merge_into_op(merge_into, ctx, codec)?)) + let target_qualifier = super::from_table_reference( + merge_into + .target_qualifier + .as_ref() + .or(node.table_name.as_ref()), + "MERGE INTO ", + )?; + WriteOp::MergeInto(Box::new(parse_merge_into_op( + merge_into, + target_qualifier, + ctx, + codec, + )?)) } }) } fn parse_merge_into_op( op: &protobuf::MergeIntoOpNode, + target_qualifier: TableReference, ctx: &TaskContext, codec: &dyn LogicalExtensionCodec, ) -> Result { @@ -88,7 +101,7 @@ fn parse_merge_into_op( .iter() .map(|c| parse_merge_into_clause(c, ctx, codec)) .collect::, Error>>()?; - Ok(MergeIntoOp { on, clauses }) + Ok(MergeIntoOp::new(target_qualifier, on, clauses)) } fn parse_merge_into_clause( diff --git a/datafusion/proto/src/logical_plan/to_proto.rs b/datafusion/proto/src/logical_plan/to_proto.rs index 16c3468465541..deeaea8a03373 100644 --- a/datafusion/proto/src/logical_plan/to_proto.rs +++ b/datafusion/proto/src/logical_plan/to_proto.rs @@ -589,6 +589,9 @@ pub fn serialize_merge_into_op( .iter() .map(|c| serialize_merge_into_clause(c, codec)) .collect::, Error>>()?, + target_qualifier: Some(protobuf::TableReference::from( + op.target_qualifier().clone(), + )), }) } diff --git a/datafusion/proto/tests/cases/roundtrip_logical_plan.rs b/datafusion/proto/tests/cases/roundtrip_logical_plan.rs index a450f7a7e888f..a7792abcd881a 100644 --- a/datafusion/proto/tests/cases/roundtrip_logical_plan.rs +++ b/datafusion/proto/tests/cases/roundtrip_logical_plan.rs @@ -675,9 +675,10 @@ async fn roundtrip_logical_plan_dml_merge_into() -> Result<()> { other => panic!("expected TableScan, got {other:?}"), }; - let merge = WriteOp::MergeInto(Box::new(MergeIntoOp { - on: col("a").eq(lit(1_i64)), - clauses: vec![ + let merge = WriteOp::MergeInto(Box::new(MergeIntoOp::new( + "target_alias", + col("target_alias.a").eq(lit(1_i64)), + vec![ MergeIntoClause { kind: MergeIntoClauseKind::Matched, predicate: Some(col("b").gt(lit(ScalarValue::Decimal128( @@ -709,7 +710,7 @@ async fn roundtrip_logical_plan_dml_merge_into() -> Result<()> { action: MergeIntoAction::Delete, }, ], - })); + ))); let plan = LogicalPlan::Dml(DmlStatement::new( "t1".into(), @@ -721,6 +722,16 @@ async fn roundtrip_logical_plan_dml_merge_into() -> Result<()> { let bytes = logical_plan_to_bytes(&plan)?; let round_trip = logical_plan_from_bytes(&bytes, &ctx.task_ctx())?; assert_eq!(format!("{plan}"), format!("{round_trip}")); + let LogicalPlan::Dml(round_trip) = round_trip else { + panic!("expected DML plan") + }; + let WriteOp::MergeInto(round_trip) = round_trip.op else { + panic!("expected MERGE operation") + }; + assert_eq!( + round_trip.target_qualifier(), + &TableReference::bare("target_alias") + ); Ok(()) } @@ -745,6 +756,9 @@ fn parse_write_op_merge_into_without_payload_errors() { fn dml_node_with_merge_payload(payload: protobuf::MergeIntoOpNode) -> protobuf::DmlNode { protobuf::DmlNode { dml_type: protobuf::dml_node::Type::MergeInto.into(), + table_name: Some(protobuf::TableReference::from(TableReference::bare( + "target", + ))), merge_into: Some(Box::new(payload)), ..Default::default() } @@ -757,12 +771,31 @@ fn parse_merge_into_op_missing_on_errors() { let node = dml_node_with_merge_payload(protobuf::MergeIntoOpNode { on: None, clauses: vec![], + target_qualifier: None, }); let err = from_proto::parse_write_op(&node, ctx.task_ctx().as_ref(), &codec) .expect_err("missing `on` must fail"); assert!(err.to_string().contains("`on`"), "unexpected error: {err}"); } +#[test] +fn parse_merge_into_op_without_target_qualifier_uses_table_name() { + let ctx = SessionContext::new(); + let codec = DefaultLogicalExtensionCodec {}; + let on = serialize_expr(&lit(true), &codec).unwrap(); + let node = dml_node_with_merge_payload(protobuf::MergeIntoOpNode { + on: Some(Box::new(on)), + clauses: vec![], + target_qualifier: None, + }); + + let op = from_proto::parse_write_op(&node, ctx.task_ctx().as_ref(), &codec).unwrap(); + let WriteOp::MergeInto(op) = op else { + panic!("expected MERGE operation") + }; + assert_eq!(op.target_qualifier(), &TableReference::bare("target")); +} + #[test] fn parse_merge_into_clause_unknown_kind_errors() { let ctx = SessionContext::new(); @@ -779,6 +812,7 @@ fn parse_merge_into_clause_unknown_kind_errors() { )), }), }], + target_qualifier: None, }); let err = from_proto::parse_write_op(&node, ctx.task_ctx().as_ref(), &codec) .expect_err("unknown clause kind tag must fail"); @@ -800,6 +834,7 @@ fn parse_merge_into_clause_missing_action_errors() { predicate: None, action: None, }], + target_qualifier: None, }); let err = from_proto::parse_write_op(&node, ctx.task_ctx().as_ref(), &codec) .expect_err("missing clause `action` must fail"); @@ -821,6 +856,7 @@ fn parse_merge_into_action_missing_oneof_errors() { predicate: None, action: Some(protobuf::MergeIntoActionNode { action: None }), }], + target_qualifier: None, }); let err = from_proto::parse_write_op(&node, ctx.task_ctx().as_ref(), &codec) .expect_err("missing action oneof must fail"); diff --git a/datafusion/session/src/table.rs b/datafusion/session/src/table.rs index a6cddd10ce80a..cf2aee03c382a 100644 --- a/datafusion/session/src/table.rs +++ b/datafusion/session/src/table.rs @@ -416,9 +416,13 @@ pub trait TableProvider: Any + Debug + Sync + Send { /// The `merge_schema` contains the target columns followed by the source /// columns, preserving their logical qualifiers. Providers can use this /// schema to resolve the logical expressions against the combined rows - /// they construct while executing the merge. + /// they construct while executing the merge. Providers should identify + /// target fields by this leading field range rather than comparing their + /// qualifiers with the provider's catalog name. /// The `on` condition is the join predicate from the ON clause. /// The `clauses` describe the WHEN MATCHED / WHEN NOT MATCHED actions. + /// These logical expressions may contain residual subqueries. Providers + /// must either support those subqueries or return an explicit error. /// /// Returns an [`ExecutionPlan`] producing a single row with `count` (UInt64). // Hand-written `#[async_trait]` expansion to reduce compile time. See diff --git a/datafusion/sql/src/statement.rs b/datafusion/sql/src/statement.rs index 33791227b3f81..bef35c3acf8d7 100644 --- a/datafusion/sql/src/statement.rs +++ b/datafusion/sql/src/statement.rs @@ -33,7 +33,6 @@ use arrow::datatypes::{Field, FieldRef, Fields}; use datafusion_common::error::_plan_err; use datafusion_common::format::ExplainStatementOptions; use datafusion_common::parsers::CompressionTypeVariant; -use datafusion_common::tree_node::{Transformed, TreeNode, TreeNodeRecursion}; use datafusion_common::{ Column, Constraint, Constraints, DFSchema, DFSchemaRef, DataFusionError, Result, ScalarValue, SchemaError, SchemaReference, TableReference, ToDFSchema, exec_err, @@ -2507,8 +2506,11 @@ impl SqlToRel<'_, S> { self.plan_from_tables(vec![source_table_with_joins], &mut planner_context)?; // 3. Build a combined schema for resolving expressions in ON and WHEN clauses - let combined_schema = - Arc::new(target_schema.as_ref().join(source_plan.schema())?); + let combined_schema = Arc::new(MergeIntoOp::expression_schema_for( + &target_qualifier, + &target_table_source.schema(), + source_plan.schema(), + )?); // 4. Convert the ON condition from sqlparser Expr to datafusion Expr let on_expr = self.sql_to_expr(*on, &combined_schema, &mut planner_context)?; @@ -2527,61 +2529,10 @@ impl SqlToRel<'_, S> { }) .collect::>>()?; - // 6. Build the MERGE operation. Column references to the target may be - // qualified with the SQL alias (`MERGE INTO target AS t ... t.col`). - // Canonicalize those to the real target table qualifier so the stored - // plan is independent of the alias: this lets the analyzer passes and - // proto deserialization rebuild the target schema from `table_name` - // alone, without carrying the alias as extra state. - let mut merge_op = MergeIntoOp { - on: on_expr, - clauses: df_clauses, - }; - if target_qualifier != target_table_ref { - // Target references in correlated subqueries are represented as - // `OuterReferenceColumn`s inside the embedded logical plan. The - // alias canonicalization below only rewrites top-level expression - // columns, so accepting such a subquery would leave the target - // alias in the public MERGE representation. Reject this case until - // the alias can be rewritten scope-safely inside subquery plans. - for expr in merge_op.exprs() { - if Self::has_outer_reference_to_qualifier(expr, &target_qualifier)? { - return not_impl_err!( - "MERGE subqueries correlated to target alias \ - '{target_qualifier}' are not supported" - ); - } - } - - // Canonicalizing target columns to `target_table_ref` is only safe - // when the source does not already use that qualifier. If it does - // (e.g. `MERGE INTO target AS t USING source AS target`), the two - // namespaces would collapse and later resolution could silently - // pick the source column for a target reference. Reject that - // collision rather than change the meaning of the condition. - if source_plan.schema().iter().any(|(qualifier, _)| { - qualifier.is_some_and(|q| q.resolved_eq(&target_table_ref)) - }) { - return plan_err!( - "MERGE source may not use the target table name '{target_table_ref}' \ - as a qualifier while the target is aliased as '{target_qualifier}'; \ - use a different source alias" - ); - } - let canonical = merge_op - .exprs() - .into_iter() - .cloned() - .map(|expr| { - Self::canonicalize_target_qualifier( - expr, - &target_qualifier, - &target_table_ref, - ) - }) - .collect::>>()?; - merge_op = merge_op.with_new_exprs(canonical)?; - } + // 6. Preserve the target's visible qualifier in the public MERGE + // representation. It is a scope-local SQL name, distinct from the + // provider identity stored in `DmlStatement::table_name`. + let merge_op = MergeIntoOp::new(target_qualifier, on_expr, df_clauses); Ok(LogicalPlan::Dml(DmlStatement::new( target_table_ref, @@ -2591,70 +2542,6 @@ impl SqlToRel<'_, S> { ))) } - /// Rewrite every [`Expr::Column`] qualified with `from` to instead use - /// `to`, leaving all other columns untouched. Used to canonicalize MERGE - /// target-alias references to the real target table qualifier. - fn canonicalize_target_qualifier( - expr: Expr, - from: &TableReference, - to: &TableReference, - ) -> Result { - expr.transform(|expr| match expr { - Expr::Column(col) if col.relation.as_ref() == Some(from) => Ok( - Transformed::yes(Expr::Column(Column::new(Some(to.clone()), col.name))), - ), - other => Ok(Transformed::no(other)), - }) - .map(|transformed| transformed.data) - } - - /// Return true if an expression contains a subquery whose embedded plan - /// has an outer reference qualified by `qualifier`. - fn has_outer_reference_to_qualifier( - expr: &Expr, - qualifier: &TableReference, - ) -> Result { - let mut found = false; - expr.apply(|expr| { - let subquery = match expr { - Expr::Exists(exists) => Some(&exists.subquery), - Expr::InSubquery(in_subquery) => Some(&in_subquery.subquery), - Expr::SetComparison(set_comparison) => Some(&set_comparison.subquery), - Expr::ScalarSubquery(subquery) => Some(subquery), - _ => None, - }; - - if let Some(subquery) = subquery { - subquery.subquery.apply_with_subqueries(|plan| { - plan.apply_expressions(|expr| { - expr.apply(|expr| { - if let Expr::OuterReferenceColumn(_, column) = expr - && column.relation.as_ref() == Some(qualifier) - { - found = true; - Ok(TreeNodeRecursion::Stop) - } else { - Ok(TreeNodeRecursion::Continue) - } - }) - })?; - Ok(if found { - TreeNodeRecursion::Stop - } else { - TreeNodeRecursion::Continue - }) - })?; - } - - Ok(if found { - TreeNodeRecursion::Stop - } else { - TreeNodeRecursion::Continue - }) - })?; - Ok(found) - } - fn merge_target_column_name( &self, name: &ObjectName, diff --git a/datafusion/sql/tests/sql_integration.rs b/datafusion/sql/tests/sql_integration.rs index 00103bfd9f56a..6d05368fcc50e 100644 --- a/datafusion/sql/tests/sql_integration.rs +++ b/datafusion/sql/tests/sql_integration.rs @@ -3708,7 +3708,7 @@ fn select_groupby_orderby_aggregate_on_non_selected_column_original_issue() { } #[test] -fn plan_merge_into_canonicalizes_qualifiers_and_preserves_quoted_columns() { +fn plan_merge_into_preserves_target_qualifier_and_quoted_columns() { let plan = logical_plan( "MERGE INTO person_quoted_cols AS t USING j2 AS s ON t.id = s.j2_id \ WHEN MATCHED THEN UPDATE SET \"First Name\" = s.j2_string \ @@ -3722,7 +3722,11 @@ fn plan_merge_into_canonicalizes_qualifiers_and_preserves_quoted_columns() { panic!("expected MergeInto, got {:?}", dml.op); }; - assert_eq!(merge_op.on.to_string(), "person_quoted_cols.id = s.j2_id"); + assert_eq!( + merge_op.target_qualifier(), + &datafusion_common::TableReference::bare("t") + ); + assert_eq!(merge_op.on.to_string(), "t.id = s.j2_id"); let datafusion_expr::dml::MergeIntoAction::Update(assignments) = &merge_op.clauses[0].action diff --git a/datafusion/sqllogictest/test_files/merge_into.slt b/datafusion/sqllogictest/test_files/merge_into.slt index f868bcbdc4862..c921a26f8d007 100644 --- a/datafusion/sqllogictest/test_files/merge_into.slt +++ b/datafusion/sqllogictest/test_files/merge_into.slt @@ -75,7 +75,7 @@ physical_plan_error 02)caused by 03)This feature is not implemented: MERGE INTO not supported for Base table -# Aliased target and source: alias is canonicalized to the table name +# Aliased target and source preserve their visible qualifiers query TT explain merge into target as t using source as s on t.id = s.id when matched and s.is_active then update set val = s.val @@ -206,13 +206,13 @@ when matched then update set s.val = 'x'; # Planning errors: qualifier and alias handling ########## -# Source alias may not collide with the target table name when the target is aliased -statement error DataFusion error: Error during planning: MERGE source may not use the target table name 'target' as a qualifier while the target is aliased as 't'; use a different source alias +# A source alias may equal the target table name when the target has another alias +statement error DataFusion error: MERGE INTO operation on table 'target'\ncaused by\nThis feature is not implemented: MERGE INTO not supported for Base table merge into target as t using source as target on t.id = target.id when matched then delete; -# Subqueries correlated to the target alias are not supported yet -statement error DataFusion error: This feature is not implemented: MERGE subqueries correlated to target alias 't' are not supported +# Target-correlated subqueries preserve the visible target qualifier +statement error DataFusion error: MERGE INTO operation on table 'target'\ncaused by\nThis feature is not implemented: MERGE INTO not supported for Base table merge into target as t using source as s on exists (select 1 from source x where x.id = t.id) when matched then delete; diff --git a/docs/adr/0001-preserve-merge-target-qualifier.md b/docs/adr/0001-preserve-merge-target-qualifier.md new file mode 100644 index 0000000000000..07a9cfbcff968 --- /dev/null +++ b/docs/adr/0001-preserve-merge-target-qualifier.md @@ -0,0 +1,21 @@ +# Preserve the MERGE target's visible qualifier + +MERGE will persist its target's visible SQL qualifier separately from the target +provider identity and will preserve that qualifier through logical analysis, +optimization, serialization, and provider dispatch. This removes the need to rewrite +qualifier strings or reconstruct lexical scope, allowing nested aliases to shadow the +MERGE alias and allowing a source qualifier to equal the target's real table name. +DataFusion-wide scoped relation-binding IDs would model identity more rigorously, but +their public-API and optimizer migration is deliberately deferred to a separate +architecture project. + +The MERGE representation and provider contract must therefore treat the qualifier as +a scope-visible name, not provider identity. New MERGE protobuf payloads are not +promised to be readable by older DataFusion versions; new readers decode pre-field +payloads by using the target table name as the qualifier. + +A shared target-first expression-schema builder will enforce visible target/source +qualifier uniqueness for SQL-planned and programmatically constructed MERGE plans. +Providers may receive residual logical subqueries in MERGE expressions and must +either handle them or return an explicit unsupported error; generic subquery +execution is outside this decision. diff --git a/docs/source/library-user-guide/upgrading/55.0.0.md b/docs/source/library-user-guide/upgrading/55.0.0.md index 4483fe76059b2..ce44213c1f044 100644 --- a/docs/source/library-user-guide/upgrading/55.0.0.md +++ b/docs/source/library-user-guide/upgrading/55.0.0.md @@ -44,6 +44,23 @@ key-field removal, or narrowing Map-key casts must be updated to retain case-sensitive field names and key identity. Ordinary Maps without semantic Struct children continue to use Arrow's normal Map cast behavior. +### `MergeIntoOp` must be constructed with its target qualifier + +`MergeIntoOp` now preserves the SQL-visible target qualifier separately from the +target table's catalog identity. The struct is also non-exhaustive, so replace +struct literals with `MergeIntoOp::new`: + +```rust,ignore +// Before +let op = MergeIntoOp { on, clauses }; + +// After +let op = MergeIntoOp::new(target_qualifier, on, clauses); +``` + +Use `MergeIntoOp::target_qualifier()` to read the qualifier. MERGE planning and +provider dispatch use it to distinguish an aliased target from source relations. + ### `TableProvider::scan` takes `Option<&[usize]>` for the projection `TableProvider::scan` and the related APIs listed below previously took the From 59bd564f6eba4da9c016b374681422e8ec973e89 Mon Sep 17 00:00:00 2001 From: xuanyili Date: Tue, 18 Aug 2026 06:02:53 +0000 Subject: [PATCH 2/4] chore: keep design notes local Remove worktree-only context and ADR files from the published change while retaining them through local excludes. --- CONTEXT.md | 20 ------------------ .../0001-preserve-merge-target-qualifier.md | 21 ------------------- 2 files changed, 41 deletions(-) delete mode 100644 CONTEXT.md delete mode 100644 docs/adr/0001-preserve-merge-target-qualifier.md diff --git a/CONTEXT.md b/CONTEXT.md deleted file mode 100644 index a003520224314..0000000000000 --- a/CONTEXT.md +++ /dev/null @@ -1,20 +0,0 @@ -# DataFusion SQL Name Resolution - -This context distinguishes catalog identity, SQL-visible names, and lexical bindings -while planning DataFusion statements. - -## Language - -**Provider identity**: -The catalog-resolved table reference used to locate a target `TableProvider`. -_Avoid_: Target qualifier, target alias - -**Visible qualifier**: -The normalized SQL relation name exposed in one lexical scope, such as `t` after -`target AS t`. The same text may name different relations in nested scopes. -_Avoid_: Relation identity, binding identity - -**Relation binding**: -The particular relation occurrence selected by lexical name resolution. A relation -binding is distinct from both its provider identity and its visible qualifier. -_Avoid_: Qualifier string diff --git a/docs/adr/0001-preserve-merge-target-qualifier.md b/docs/adr/0001-preserve-merge-target-qualifier.md deleted file mode 100644 index 07a9cfbcff968..0000000000000 --- a/docs/adr/0001-preserve-merge-target-qualifier.md +++ /dev/null @@ -1,21 +0,0 @@ -# Preserve the MERGE target's visible qualifier - -MERGE will persist its target's visible SQL qualifier separately from the target -provider identity and will preserve that qualifier through logical analysis, -optimization, serialization, and provider dispatch. This removes the need to rewrite -qualifier strings or reconstruct lexical scope, allowing nested aliases to shadow the -MERGE alias and allowing a source qualifier to equal the target's real table name. -DataFusion-wide scoped relation-binding IDs would model identity more rigorously, but -their public-API and optimizer migration is deliberately deferred to a separate -architecture project. - -The MERGE representation and provider contract must therefore treat the qualifier as -a scope-visible name, not provider identity. New MERGE protobuf payloads are not -promised to be readable by older DataFusion versions; new readers decode pre-field -payloads by using the target table name as the qualifier. - -A shared target-first expression-schema builder will enforce visible target/source -qualifier uniqueness for SQL-planned and programmatically constructed MERGE plans. -Providers may receive residual logical subqueries in MERGE expressions and must -either handle them or return an explicit unsupported error; generic subquery -execution is outside this decision. From 9ecfa52140030097071a40f7c4126afc7fa459ff Mon Sep 17 00:00:00 2001 From: xuanyili Date: Tue, 18 Aug 2026 06:09:16 +0000 Subject: [PATCH 3/4] docs: remove unreleased MERGE migration note MergeIntoOp has not appeared in a release, so users do not need before-and-after upgrade guidance. --- .../library-user-guide/upgrading/55.0.0.md | 17 ----------------- 1 file changed, 17 deletions(-) diff --git a/docs/source/library-user-guide/upgrading/55.0.0.md b/docs/source/library-user-guide/upgrading/55.0.0.md index ce44213c1f044..4483fe76059b2 100644 --- a/docs/source/library-user-guide/upgrading/55.0.0.md +++ b/docs/source/library-user-guide/upgrading/55.0.0.md @@ -44,23 +44,6 @@ key-field removal, or narrowing Map-key casts must be updated to retain case-sensitive field names and key identity. Ordinary Maps without semantic Struct children continue to use Arrow's normal Map cast behavior. -### `MergeIntoOp` must be constructed with its target qualifier - -`MergeIntoOp` now preserves the SQL-visible target qualifier separately from the -target table's catalog identity. The struct is also non-exhaustive, so replace -struct literals with `MergeIntoOp::new`: - -```rust,ignore -// Before -let op = MergeIntoOp { on, clauses }; - -// After -let op = MergeIntoOp::new(target_qualifier, on, clauses); -``` - -Use `MergeIntoOp::target_qualifier()` to read the qualifier. MERGE planning and -provider dispatch use it to distinguish an aliased target from source relations. - ### `TableProvider::scan` takes `Option<&[usize]>` for the projection `TableProvider::scan` and the related APIs listed below previously took the From 3d19096ee1136ace48541f7f52a660ac873020ff Mon Sep 17 00:00:00 2001 From: xuanyili Date: Fri, 21 Aug 2026 00:38:24 +0000 Subject: [PATCH 4/4] test: migrate MERGE SQL coverage to SLT Keep only binding-specific Rust assertions while moving SQL behavior cases to the faster, more maintainable sqllogictest suite. --- datafusion/core/tests/sql/sql_api.rs | 192 ---------------- .../sqllogictest/test_files/merge_into.slt | 210 +++++++++++++++++- 2 files changed, 208 insertions(+), 194 deletions(-) diff --git a/datafusion/core/tests/sql/sql_api.rs b/datafusion/core/tests/sql/sql_api.rs index 43bed215bef6e..1b794b3a6672a 100644 --- a/datafusion/core/tests/sql/sql_api.rs +++ b/datafusion/core/tests/sql/sql_api.rs @@ -215,11 +215,6 @@ async fn merge_into_context() -> SessionContext { let ctx = SessionContext::new(); ctx.sql("CREATE TABLE target (id INT)").await.unwrap(); ctx.sql("CREATE TABLE source (id INT)").await.unwrap(); - ctx.sql("CREATE TABLE \"Target\" (id INT)").await.unwrap(); - ctx.sql("CREATE SCHEMA \"CaseSchema\"").await.unwrap(); - ctx.sql("CREATE TABLE \"CaseSchema\".\"Target\" (id INT)") - .await - .unwrap(); ctx } @@ -285,130 +280,6 @@ fn has_outer_reference_to(expr: &Expr, qualifier: &TableReference) -> bool { found } -#[tokio::test] -async fn merge_into_distinguishes_target_alias_from_source_qualifier() { - let ctx = merge_into_context().await; - - for target_ref in [ - "target", - "public.target", - "datafusion.public.target", - "\"Target\"", - "\"CaseSchema\".\"Target\"", - ] { - assert_merge_physical_error( - &ctx, - &format!( - "MERGE INTO {target_ref} AS t USING source AS target \ - ON t.id = target.id WHEN MATCHED THEN DELETE" - ), - "MERGE INTO not supported for Base table", - ) - .await; - } - - for (target_alias, target_column) in [ - ("T", "t.id"), - ("\"T\"", "\"T\".id"), - ("\"public.target\"", "\"public.target\".id"), - ] { - assert_merge_physical_error( - &ctx, - &format!( - "MERGE INTO target AS {target_alias} USING source AS target \ - ON {target_column} = target.id WHEN MATCHED THEN DELETE" - ), - "MERGE INTO not supported for Base table", - ) - .await; - } - - for source in [ - "(SELECT id FROM source) AS target", - "target AS source_target", - ] { - assert_merge_physical_error( - &ctx, - &format!( - "MERGE INTO target AS t USING {source} \ - ON true WHEN MATCHED THEN DELETE" - ), - "MERGE INTO not supported for Base table", - ) - .await; - } - - for (target_ref, target_column, expected_qualifier) in [ - ( - "\"Target\"", - "\"Target\".id", - TableReference::bare("Target"), - ), - ( - "\"CaseSchema\".\"Target\"", - "\"CaseSchema\".\"Target\".id", - TableReference::partial("CaseSchema", "Target"), - ), - ] { - let sql = format!( - "MERGE INTO {target_ref} USING source AS s \ - ON {target_column} = s.id WHEN MATCHED THEN DELETE" - ); - let merge_op = merge_operation(&ctx, &sql).await; - assert_eq!(merge_op.target_qualifier(), &expected_qualifier); - assert_merge_physical_error( - &ctx, - &sql, - "MERGE INTO not supported for Base table", - ) - .await; - } -} - -#[tokio::test] -async fn merge_into_rejects_outer_source_qualifier_matching_target_alias() { - let ctx = merge_into_context().await; - for (target_alias, source_alias, expected) in [ - ("t", "t", "target qualifier 't'"), - ("t", "T", "target qualifier 't'"), - ("\"T\"", "\"T\"", "target qualifier 'T'"), - ] { - let err = ctx - .sql(&format!( - "MERGE INTO target AS {target_alias} USING source AS {source_alias} \ - ON true WHEN MATCHED THEN DELETE" - )) - .await - .unwrap_err(); - assert_contains!(err.strip_backtrace(), expected); - assert_contains!(err.strip_backtrace(), "conflicts with a source qualifier"); - } - - let err = ctx - .sql( - "MERGE INTO target AS t \ - USING (SELECT id AS source_id FROM source) AS t \ - ON true WHEN MATCHED THEN DELETE", - ) - .await - .unwrap_err(); - assert_contains!( - err.strip_backtrace(), - "MERGE target qualifier 't' conflicts with a source qualifier" - ); - - for target_ref in ["public.target", "datafusion.public.target"] { - let err = ctx - .sql(&format!( - "MERGE INTO {target_ref} USING source AS target \ - ON true WHEN MATCHED THEN DELETE" - )) - .await - .unwrap_err(); - assert_contains!(err.strip_backtrace(), "conflicts with a source qualifier"); - } -} - #[tokio::test] async fn merge_into_preserves_target_alias_in_correlated_subquery() { let ctx = merge_into_context().await; @@ -452,69 +323,6 @@ async fn merge_into_preserves_target_alias_in_correlated_subquery() { &merge_op.on, &TableReference::bare("t") )); - - for sql in [ - // Direct and deeply nested target correlation. - direct_exists, - direct_in, - direct_any, - direct_all, - direct_scalar, - "MERGE INTO target AS t USING source AS s \ - ON EXISTS (SELECT 1 FROM source AS q \ - WHERE EXISTS (SELECT 1 FROM source AS x WHERE x.id = t.id)) \ - WHEN MATCHED THEN DELETE", - "MERGE INTO target AS t USING source AS s \ - ON EXISTS (SELECT 1 FROM source AS q \ - CROSS JOIN LATERAL (SELECT t.id) AS l) \ - WHEN MATCHED THEN DELETE", - "MERGE INTO target AS t USING source AS s \ - ON EXISTS (SELECT 1 FROM source AS q CROSS JOIN (\ - source AS x CROSS JOIN LATERAL (SELECT t.id) AS l\ - )) WHEN MATCHED THEN DELETE", - // Source correlation, no correlation, and nested shadowing. - "MERGE INTO target AS t USING source AS s \ - ON EXISTS (SELECT 1 FROM source AS x WHERE x.id = s.id) \ - WHEN MATCHED THEN DELETE", - "MERGE INTO target AS t USING source AS s \ - ON t.id = ANY (SELECT id FROM source) \ - WHEN MATCHED THEN DELETE", - shadowed_correlation, - "MERGE INTO target AS t USING source AS s \ - ON EXISTS (SELECT EXISTS (\ - SELECT 1 FROM source AS x WHERE x.id = t.id\ - ) FROM source AS t) \ - WHEN MATCHED THEN DELETE", - "MERGE INTO target AS t USING source AS s \ - ON EXISTS (SELECT 1 FROM source AS t \ - WHERE EXISTS (SELECT 1 FROM source AS q \ - WHERE EXISTS (SELECT 1 FROM source AS x WHERE x.id = t.id))) \ - WHEN MATCHED THEN DELETE", - "MERGE INTO target AS t USING source AS s \ - ON EXISTS (SELECT 1 FROM source AS t \ - CROSS JOIN LATERAL (SELECT t.id) AS l) \ - WHEN MATCHED THEN DELETE", - "MERGE INTO target AS t USING source AS s \ - ON EXISTS (SELECT 1 FROM source AS t CROSS JOIN (\ - source AS x CROSS JOIN LATERAL (SELECT t.id) AS l\ - )) WHEN MATCHED THEN DELETE", - ] { - assert_merge_physical_error(&ctx, sql, "MERGE INTO not supported for Base table") - .await; - } - - // LIMIT expressions intentionally use an empty local scope. The nested - // `t.id` therefore resolves to the MERGE target rather than `source AS t`, - // and reaches the existing generic scalar-subquery placement restriction. - assert_merge_physical_error( - &ctx, - "MERGE INTO target AS t USING source AS s \ - ON EXISTS (SELECT t.id FROM source AS t \ - LIMIT (SELECT t.id FROM source AS x LIMIT 1)) \ - WHEN MATCHED THEN DELETE", - "Correlated scalar subquery can only be used", - ) - .await; } #[tokio::test] diff --git a/datafusion/sqllogictest/test_files/merge_into.slt b/datafusion/sqllogictest/test_files/merge_into.slt index c921a26f8d007..fc64a35723520 100644 --- a/datafusion/sqllogictest/test_files/merge_into.slt +++ b/datafusion/sqllogictest/test_files/merge_into.slt @@ -44,6 +44,15 @@ insert into source values (2, 'xxxx', true); statement ok insert into source values (4, 'yyyy', false); +statement ok +create table "Target"(id int); + +statement ok +create schema "CaseSchema"; + +statement ok +create table "CaseSchema"."Target"(id int); + ########## # Logical planning @@ -203,7 +212,7 @@ merge into target as t using source as s on t.id = s.id when matched then update set s.val = 'x'; ########## -# Planning errors: qualifier and alias handling +# Qualifier and alias handling ########## # A source alias may equal the target table name when the target has another alias @@ -211,12 +220,200 @@ statement error DataFusion error: MERGE INTO operation on table 'target'\ncaused merge into target as t using source as target on t.id = target.id when matched then delete; -# Target-correlated subqueries preserve the visible target qualifier +# Qualified and quoted target names preserve their aliases +statement error DataFusion error: MERGE INTO operation on table 'public.target'\ncaused by\nThis feature is not implemented: MERGE INTO not supported for Base table +merge into public.target as t using source as target on t.id = target.id +when matched then delete; + +statement error DataFusion error: MERGE INTO operation on table 'datafusion.public.target'\ncaused by\nThis feature is not implemented: MERGE INTO not supported for Base table +merge into datafusion.public.target as t using source as target on t.id = target.id +when matched then delete; + +statement error DataFusion error: MERGE INTO operation on table 'Target'\ncaused by\nThis feature is not implemented: MERGE INTO not supported for Base table +merge into "Target" as t using source as target on t.id = target.id +when matched then delete; + +statement error DataFusion error: MERGE INTO operation on table 'CaseSchema.Target'\ncaused by\nThis feature is not implemented: MERGE INTO not supported for Base table +merge into "CaseSchema"."Target" as t using source as target on t.id = target.id +when matched then delete; + +# Alias normalization and case sensitivity +statement error DataFusion error: MERGE INTO operation on table 'target'\ncaused by\nThis feature is not implemented: MERGE INTO not supported for Base table +merge into target as T using source as target on t.id = target.id +when matched then delete; + +statement error DataFusion error: MERGE INTO operation on table 'target'\ncaused by\nThis feature is not implemented: MERGE INTO not supported for Base table +merge into target as "T" using source as target on "T".id = target.id +when matched then delete; + +statement error DataFusion error: MERGE INTO operation on table 'target'\ncaused by\nThis feature is not implemented: MERGE INTO not supported for Base table +merge into target as "public.target" using source as target on "public.target".id = target.id +when matched then delete; + +# Derived sources and self-MERGE keep distinct qualifiers +statement error DataFusion error: MERGE INTO operation on table 'target'\ncaused by\nThis feature is not implemented: MERGE INTO not supported for Base table +merge into target as t using (select id from source) as target on true +when matched then delete; + +statement error DataFusion error: MERGE INTO operation on table 'target'\ncaused by\nThis feature is not implemented: MERGE INTO not supported for Base table +merge into target as t using target as source_target on true +when matched then delete; + +# An unaliased quoted target remains its visible qualifier +statement error DataFusion error: MERGE INTO operation on table 'Target'\ncaused by\nThis feature is not implemented: MERGE INTO not supported for Base table +merge into "Target" using source as s on "Target".id = s.id +when matched then delete; + +statement error DataFusion error: MERGE INTO operation on table 'CaseSchema.Target'\ncaused by\nThis feature is not implemented: MERGE INTO not supported for Base table +merge into "CaseSchema"."Target" using source as s on "CaseSchema"."Target".id = s.id +when matched then delete; + +# The source may not collide with the visible target qualifier +statement error DataFusion error: Error during planning: MERGE target qualifier 't' conflicts with a source qualifier +merge into target as t using source as t on true +when matched then delete; + +statement error DataFusion error: Error during planning: MERGE target qualifier 't' conflicts with a source qualifier +merge into target as t using source as T on true +when matched then delete; + +statement error DataFusion error: Error during planning: MERGE target qualifier 'T' conflicts with a source qualifier +merge into target as "T" using source as "T" on true +when matched then delete; + +statement error DataFusion error: Error during planning: MERGE target qualifier 't' conflicts with a source qualifier +merge into target as t using (select id as source_id from source) as t on true +when matched then delete; + +statement error DataFusion error: Error during planning: MERGE target qualifier 'public.target' conflicts with a source qualifier +merge into public.target using source as target on true +when matched then delete; + +statement error DataFusion error: Error during planning: MERGE target qualifier 'datafusion.public.target' conflicts with a source qualifier +merge into datafusion.public.target using source as target on true +when matched then delete; + +########## +# Correlated subqueries and alias scopes +########## + +# Direct target correlation statement error DataFusion error: MERGE INTO operation on table 'target'\ncaused by\nThis feature is not implemented: MERGE INTO not supported for Base table merge into target as t using source as s on exists (select 1 from source x where x.id = t.id) when matched then delete; +statement error DataFusion error: MERGE INTO operation on table 'target'\ncaused by\nThis feature is not implemented: MERGE INTO not supported for Base table +merge into target as t using source as s +on t.id in (select x.id from source x where x.id = t.id) +when matched then delete; + +statement error DataFusion error: MERGE INTO operation on table 'target'\ncaused by\nThis feature is not implemented: MERGE INTO not supported for Base table +merge into target as t using source as s +on t.id = any (select x.id from source x where x.id = t.id) +when matched then delete; + +statement error DataFusion error: MERGE INTO operation on table 'target'\ncaused by\nThis feature is not implemented: MERGE INTO not supported for Base table +merge into target as t using source as s +on t.id = all (select x.id from source x where x.id = t.id) +when matched then delete; + +statement error DataFusion error: MERGE INTO operation on table 'target'\ncaused by\nThis feature is not implemented: MERGE INTO not supported for Base table +merge into target as t using source as s +on t.id = (select max(x.id) from source x where x.id = t.id) +when matched then delete; + +# Deep target correlation and lateral scopes +statement error DataFusion error: MERGE INTO operation on table 'target'\ncaused by\nThis feature is not implemented: MERGE INTO not supported for Base table +merge into target as t using source as s +on exists ( + select 1 from source q + where exists (select 1 from source x where x.id = t.id) +) +when matched then delete; + +statement error DataFusion error: MERGE INTO operation on table 'target'\ncaused by\nThis feature is not implemented: MERGE INTO not supported for Base table +merge into target as t using source as s +on exists ( + select 1 from source q + cross join lateral (select t.id) l +) +when matched then delete; + +statement error DataFusion error: MERGE INTO operation on table 'target'\ncaused by\nThis feature is not implemented: MERGE INTO not supported for Base table +merge into target as t using source as s +on exists ( + select 1 from source q cross join ( + source x cross join lateral (select t.id) l + ) +) +when matched then delete; + +# Source correlation and uncorrelated subqueries remain supported +statement error DataFusion error: MERGE INTO operation on table 'target'\ncaused by\nThis feature is not implemented: MERGE INTO not supported for Base table +merge into target as t using source as s +on exists (select 1 from source x where x.id = s.id) +when matched then delete; + +statement error DataFusion error: MERGE INTO operation on table 'target'\ncaused by\nThis feature is not implemented: MERGE INTO not supported for Base table +merge into target as t using source as s +on t.id = any (select id from source) +when matched then delete; + +# Inner aliases may shadow the target alias +statement error DataFusion error: MERGE INTO operation on table 'target'\ncaused by\nThis feature is not implemented: MERGE INTO not supported for Base table +merge into target as t using source as s +on exists ( + select 1 from source t + where exists (select 1 from source x where x.id = t.id) +) +when matched then delete; + +statement error DataFusion error: MERGE INTO operation on table 'target'\ncaused by\nThis feature is not implemented: MERGE INTO not supported for Base table +merge into target as t using source as s +on exists ( + select exists (select 1 from source x where x.id = t.id) + from source t +) +when matched then delete; + +statement error DataFusion error: MERGE INTO operation on table 'target'\ncaused by\nThis feature is not implemented: MERGE INTO not supported for Base table +merge into target as t using source as s +on exists ( + select 1 from source t + where exists ( + select 1 from source q + where exists (select 1 from source x where x.id = t.id) + ) +) +when matched then delete; + +statement error DataFusion error: MERGE INTO operation on table 'target'\ncaused by\nThis feature is not implemented: MERGE INTO not supported for Base table +merge into target as t using source as s +on exists ( + select 1 from source t + cross join lateral (select t.id) l +) +when matched then delete; + +statement error DataFusion error: MERGE INTO operation on table 'target'\ncaused by\nThis feature is not implemented: MERGE INTO not supported for Base table +merge into target as t using source as s +on exists ( + select 1 from source t cross join ( + source x cross join lateral (select t.id) l + ) +) +when matched then delete; + +# LIMIT expressions use an empty scope; this correlated scalar placement remains unsupported +statement error DataFusion error: Invalid \(non-executable\) plan after Analyzer\ncaused by\nError during planning: Correlated scalar subquery can only be used in Projection, Filter, Aggregate and MERGE DML plan nodes +merge into target as t using source as s +on exists ( + select t.id from source t + limit (select t.id from source x limit 1) +) +when matched then delete; + ########## # Planning errors: unsupported syntax ########## @@ -241,6 +438,15 @@ statement error DataFusion error: This feature is not implemented: MERGE INSERT merge into target using source on target.id = source.id when not matched then insert row; +statement ok +drop table "Target"; + +statement ok +drop table "CaseSchema"."Target"; + +statement ok +drop schema "CaseSchema"; + statement ok drop table target;