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..1b794b3a6672a 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;
 
@@ -215,69 +218,111 @@ async fn merge_into_context() -> SessionContext {
     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}"
+    );
 }
 
-#[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.
-    let ctx = merge_into_context().await;
+async fn merge_operation(ctx: &SessionContext, sql: &str) -> Box<MergeIntoOp> {
+    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
+}
 
-    for target_ref in ["target", "public.target", "datafusion.public.target"] {
-        assert_merge_sql_error(
-            &ctx,
-            &format!(
-                "MERGE INTO {target_ref} AS t USING source AS target \
-                 ON t.id = target.id WHEN MATCHED THEN DELETE"
-            ),
-            &format!(
-                "MERGE source may not use the target table name '{target_ref}' \
-                 as a qualifier"
-            ),
-        )
-        .await;
-    }
+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_subqueries_correlated_to_target_alias() {
+async fn merge_into_preserves_target_alias_in_correlated_subquery() {
     let ctx = merge_into_context().await;
-    assert_merge_sql_error(
-        &ctx,
-        "MERGE INTO target AS t USING source AS s \
+    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 [
-        "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",
+        direct_exists,
+        direct_in,
+        direct_any,
+        direct_all,
+        direct_scalar,
     ] {
-        assert_merge_physical_error(&ctx, sql, "MERGE INTO not supported for Base table")
-            .await;
+        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")
+    ));
 }
 
 #[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<DFSchema> {
+        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 <expr>`.
     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<TableReference>,
+        on: Expr,
+        clauses: Vec<MergeIntoClause>,
+    ) -> 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<DFSchema> {
+        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<DFSchema> {
+        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<Transformed<LogicalPlan>> {
         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<Transformed<LogicalPlan>> {
-        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<Option<DFSchema>> {
+    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<D>(deserializer: D) -> std::result::Result<GeneratedField, D::Error>
@@ -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<LogicalExprNode>>,
     #[prost(message, repeated, tag = "2")]
     pub clauses: ::prost::alloc::vec::Vec<MergeIntoClauseNode>,
+    /// 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<TableReference>,
 }
 /// 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<MergeIntoOp, Error> {
@@ -88,7 +101,7 @@ fn parse_merge_into_op(
         .iter()
         .map(|c| parse_merge_into_clause(c, ctx, codec))
         .collect::<Result<Vec<_>, 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::<Result<Vec<_>, 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<S: ContextProvider> 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<S: ContextProvider> SqlToRel<'_, S> {
             })
             .collect::<Result<Vec<_>>>()?;
 
-        // 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::<Result<Vec<_>>>()?;
-            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<S: ContextProvider> 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> {
-        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<bool> {
-        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..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
@@ -75,7 +84,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
@@ -203,20 +212,208 @@ 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
 ##########
 
-# 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
+# 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;