feat: Add SQL planner, physical planner, and TableProvider hook for MERGE INTO by wirybeaver · Pull Request #22988 · apache/datafusion

wirybeaver · 2026-06-17T01:22:39Z

Which issue does this PR close?

Follow-up to #20763 (merged) which added MergeIntoOp, MergeIntoClause, and proto types.

Rationale for this change

MERGE INTO (SQL:2003) is a widely-used DML statement for upsert/conditional update workloads. This PR wires the types introduced in #20763 through the SQL planner, physical planner, and TableProvider trait so that table implementations can actually execute merge operations.

What changes are included in this PR?

datafusion/catalog — TableProvider trait extension

Add merge_into(source, on, clauses) async method with a default not_impl_err impl so existing providers are unaffected.

datafusion/sql — SQL → LogicalPlan

statement.rs: parse Statement::Merge into LogicalPlan::Dml with WriteOp::MergeInto.
Resolve the target table and plan the USING source into a LogicalPlan.
Build a combined target+source schema to resolve ON and WHEN expressions.
Convert ON condition and WHEN MATCHED / NOT MATCHED clauses to DataFusion Expr.

datafusion/expr — expression plumbing

MergeIntoOp::exprs(): stable iteration over all expressions (ON, then per-clause predicate + action values).
MergeIntoOp::with_new_exprs(): rebuild op from a transformed expr vector.
Branch LogicalPlan::apply_expressions, map_expressions, and with_new_exprs on WriteOp::MergeInto so optimizers can rewrite merge expressions. Other WriteOp variants are unchanged.

datafusion/core — physical planner dispatch

Dispatch WriteOp::MergeInto in the physical planner.
Recover the TableProvider via source_as_provider(), extract the source ExecutionPlan, and call TableProvider::merge_into.

Are these changes tested?

The SQL planner path is exercised by datafusion/proto/tests/cases/roundtrip_logical_plan.rs (proto round-trip for MergeInto).
Unit tests for MergeIntoOp::exprs / with_new_exprs are included in dml.rs.
End-to-end integration tests require a TableProvider that implements merge_into; that is left to follow-up once a concrete provider (e.g. Delta Lake) adopts the hook.

Are there any user-facing changes?

TableProvider gains a new merge_into method. The default implementation returns not_impl_err, so existing implementations compile without changes.
MERGE INTO <table> USING <source> ON <cond> WHEN ... SQL syntax is now accepted by the DataFusion SQL parser and planner.

Add merge_into async method to TableProvider trait for MERGE INTO DML support. The method accepts: - source: ExecutionPlan representing the USING clause - on: Expr representing the ON join condition - clauses: Vec<MergeIntoClause> for WHEN MATCHED/NOT MATCHED actions Default implementation returns not_impl_err for tables that don't support MERGE INTO operations.

kosiew

@wirybeaver
Thanks for the work on MERGE planning. I found two issues that look like they should be fixed before this lands.

kosiew · 2026-06-24T10:53:18Z

+        let target_table_source = self
+            .context_provider
+            .get_table_source(target_table_ref.clone())?;
+        let target_schema = Arc::new(DFSchema::try_from_qualified_schema(


I think target aliases need a bit more handling here. Right now MERGE INTO target AS t USING source AS s ON t.id = s.id ... accepts the alias, but target_schema is still qualified with the table name rather than t. That means normal alias-qualified references in the ON clause or WHEN AND expressions will not resolve.

Could we either reject target aliases explicitly for now, or qualify the target side with the alias so this matches the rest of SQL planning?

Fixed — when an alias is present we now use it as the schema qualifier via TableReference::bare(alias.name.value.clone()), so t.col resolves correctly in ON and WHEN expressions.

kosiew · 2026-06-24T10:53:18Z

+                MergeIntoAction::Update(assignments)
+            }
+            ast::MergeAction::Insert(insert_expr) => {
+                let columns: Vec<String> = insert_expr


I think the MERGE INSERT actions need the same kind of validation that regular INSERT planning does. At the moment, unknown columns, duplicate columns, and mismatched columns / values lengths can all plan successfully. Empty columns also does not appear to require values for every target column.

That can leave TableProviders receiving malformed actions instead of a validated logical operation. Could we normalize and validate target column names, reject duplicates, and verify that the value count matches either the explicit columns or the full target schema?

Fixed — we now validate before building the action: duplicate column names are rejected, each named column is checked against the target schema via field_with_unqualified_name, and the value count must equal either the explicit column count (when a column list is given) or the full target schema width (when none is given).

kosiew · 2026-06-24T10:53:18Z

+            })
+            .collect::<Result<Vec<_>>>()?;
+
+        // 6. Build the DmlStatement


Small cleanup suggestion: this could return Ok(LogicalPlan::Dml(DmlStatement::new(...))) directly instead of binding let plan and then returning Ok(plan).

Done, returns Ok(LogicalPlan::Dml(...)) directly now.

kosiew · 2026-06-24T10:53:18Z

+                    .assignments
+                    .into_iter()
+                    .map(|assign| {
+                        let col_name = match &assign.target {


Small follow-up suggestion: the update-column extraction and insert-column extraction both walk an ObjectName and take the last identifier. It might be worth adding a small private helper that returns Result<String> and produces a planner error for non-ident parts, rather than using unwrap() at each call site.

Done — extracted ident_from_object_name_last as a private associated function that returns Result<String> and errors on non-ident parts. Both UPDATE and INSERT column extraction now go through it.

kosiew

@wirybeaver
Thanks for the updates here. I think a couple of identifier normalization details still need to be tightened up before this is ready.

kosiew · 2026-06-29T05:30:02Z

+        // `MERGE INTO target AS t`. Fall back to the table reference itself.
+        let target_qualifier = target_alias
+            .as_ref()
+            .map(|a| TableReference::bare(a.name.value.clone()))


Nice fix to use the target alias as the schema qualifier here. One remaining issue is that this uses a.name.value.clone() directly, so unquoted aliases are not normalized the same way as the rest of planning.

For example, MERGE INTO target AS T ... ON T.id = s.id can store the target qualifier as T, while expression planning normalizes T.id to t.id. That can still make alias-qualified target references fail.

Could we build the alias qualifier with self.ident_normalizer.normalize(a.name.clone()) before passing it to TableReference::bare(...)?

Good catch — now using self.ident_normalizer.normalize(a.name.clone()) before passing to TableReference::bare, so unquoted aliases like T are lowercased to match expression planning.

kosiew · 2026-06-29T05:30:02Z

+                let columns: Vec<String> = insert_expr
+                    .columns
+                    .iter()
+                    .map(|c| Self::ident_from_object_name_last(c))


I think the MERGE INSERT column handling still needs one more normalization step. The extracted column names currently keep the raw Ident.value from ident_from_object_name_last, so validation happens before SQL identifier normalization.

That means something like MERGE ... INSERT (ID) VALUES (...) can be rejected against a target column named id, and duplicate detection may not match regular INSERT behavior.

Could we normalize the extracted column identifiers before duplicate checks and before the field_with_unqualified_name lookup, matching regular INSERT planning?

Fixed — both UPDATE and INSERT column names are now normalized through self.ident_normalizer.normalize(Ident::new(...)) before duplicate checks and schema lookups, matching regular INSERT planning behavior.

Implement merge_to_plan and merge_clause_to_plan in SQL planner: - Parse Statement::Merge into LogicalPlan::Dml with WriteOp::MergeInto - Resolve target table and plan source (USING clause) as LogicalPlan - Build combined schema for target + source to resolve ON and WHEN expressions - Convert ON condition and WHEN clauses to DataFusion Expr - Handle UPDATE, INSERT, and DELETE actions in WHEN clauses Add physical planner dispatch for WriteOp::MergeInto: - Use source_as_provider() to recover the TableProvider from the TableSource - Extract source ExecutionPlan from children - Call TableProvider::merge_into with source plan, ON condition, and clauses - Wrap errors with MERGE INTO operation context Wire MergeInto's expressions through LogicalPlan tree-traversal so optimizers can rewrite them: add MergeIntoOp::exprs() (stable iteration order: on, then per-clause predicate + action value Exprs) and MergeIntoOp::with_new_exprs() to rebuild the op from a transformed expr vector. Branch LogicalPlan::apply_expressions, map_expressions, and with_new_exprs on WriteOp::MergeInto to use these helpers; other WriteOp variants continue to expose no expressions as before.

raghav-reglobe · 2026-07-03T18:04:47Z

Took this branch for a spin against a provider-side consumer (we're building an Iceberg SCD2 upsert on this hook), and I think there's one blocking issue the current tests don't reach: any target-column reference in the ON or WHEN expressions fails analysis, so a realistic MERGE can't execute through SessionContext::sql().

Minimal repro (custom provider registered as t, MemTable as batch):

MERGE INTO t USING batch s ON t.id = s.id
WHEN MATCHED THEN UPDATE SET val = s.val

Error: Context("type_coercion", SchemaError(FieldNotFound { field: Column { relation: Some(Bare { table: "t" }), name: "id" }, valid_fields: [s.id, s.val, …] }))

What's happening: the Dml node's only input is the planned USING source, but MergeIntoOp's expressions reference the combined target×source namespace. Since this PR wires those expressions into the generic tree-node traversal (tree_node.rs), the analyzer's TypeCoercion pass rewrites them against the node's input schema — which is source-only — and errors on the first target reference. As a probe, ON s.id = s.id (source-only) sails through analysis and reaches merge_into() fine, which pins the root cause. Everything else held up nicely in testing, for what it's worth — a window-function subquery as the USING source (ROW_NUMBER/LEAD), multi-condition ON, clause predicates, and the INSERT validation all plan exactly as expected.

Two directions I can see:

Minimal: don't expose MergeIntoOp exprs through the generic traversal (keep exprs()/with_new_exprs as inherent methods only). The analyzer/optimizer then leave the node alone, and providers receive un-coerced expressions — acceptable for providers that do their own resolution, and it ships a working v1.
Fuller: record the target's resolution qualifier on MergeIntoOp (the alias when present, else the table reference — today the alias is dropped after planning), and teach TypeCoercion to resolve MergeInto exprs against target-schema ⋈ input-schema. Coercion on these expressions genuinely matters (e.g. an Int32 target key against an Int64 window output in ON), and the recorded qualifier also gives providers an explicit way to tell target references from source references instead of inferring by absence.

I'd lean toward 2, but 1 seems fine to unblock if you'd rather keep this PR small. Happy to share the test module I used — a CaptureMergeProvider in the style of dml_planning.rs with basic + SCD2-shaped (window-subquery USING, triple-condition ON, clause predicate) cases — if it's useful as a starting point for end-to-end coverage here. Also happy to put up a patch for either direction if that helps.

github-actions Bot added sql SQL Planner logical-expr Logical plan and expressions core Core DataFusion crate catalog Related to the catalog crate labels Jun 17, 2026

kosiew requested changes Jun 24, 2026

View reviewed changes

wirybeaver force-pushed the feature/mergeinto branch from cb89e53 to 6f0dfe1 Compare June 25, 2026 20:05

wirybeaver mentioned this pull request Jun 28, 2026

Support MERGE INTO #20746

Open

2 tasks

kosiew requested changes Jun 29, 2026

View reviewed changes

wirybeaver force-pushed the feature/mergeinto branch from 6f0dfe1 to 799ed48 Compare July 2, 2026 14:27

raghav-reglobe mentioned this pull request Jul 3, 2026

[Epic] Add MERGE INTO support for DataFusion integration apache/iceberg-rust#2201

Open

7 tasks

Uh oh!

Conversation

wirybeaver commented Jun 17, 2026

Which issue does this PR close?

Rationale for this change

What changes are included in this PR?

Are these changes tested?

Are there any user-facing changes?

Uh oh!

kosiew left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kosiew left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

raghav-reglobe commented Jul 3, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

kosiew left a comment •

edited

Loading

kosiew left a comment •

edited

Loading