[Epic] Add MERGE INTO support for DataFusion integration

[wirybeaver]

What's the feature are you trying to implement?

Add support for SQL MERGE INTO (UPSERT) operations in the iceberg-datafusion integration. This enables atomic row-level updates and inserts based on join conditions, essential for CDC pipelines, incremental updates, and data synchronization. I already have a PoC branch.

Refer to the insert_into support

The Spark SPJ (Storage Partition Join) style is the key optimization I wanted to introduce. The Datafusion currently doesn't support merge_into sql parsing and logic plan yet. I am contributing the "MERGE INTO" in datafusion as well: apache/datafusion#20746.

SQL Example:

MERGE INTO target_table t
USING source_table s
ON t.id = s.id
WHEN MATCHED THEN
  UPDATE SET t.value = s.value
WHEN NOT MATCHED THEN
  INSERT (id, value) VALUES (s.id, s.value)

---

Query Execution Plan (CoW Mode)

Baseline Plan (Unpartitioned Table)

┌─────────────────────────────┐
│   IcebergMergeCommitExec    │  Commits via RowDelta transaction
│   (add + remove data files) │  Outputs: record count
└──────────────┬──────────────┘
               │
┌──────────────▼──────────────┐
│   CoalescePartitionsExec    │  Merges all partitions into
│                             │  single stream for atomic commit
└──────────────┬──────────────┘
               │
┌──────────────▼──────────────┐
│   IcebergMergeWriteExec     │  Writes merged rows to new Parquet
│                             │  files via TaskWriter; tracks
│                             │  _file values as deleted_files
└──────────────┬──────────────┘
               │
┌──────────────▼──────────────┐
│      IcebergMergeExec       │  FULL OUTER JOIN (HashJoinExec)
│                             │  Classifies rows:
│                             │    MATCHED → apply UPDATE exprs
│                             │    NOT MATCHED → apply INSERT exprs
└──────┬──────────────┬───────┘
       │              │
┌──────▼──────┐ ┌─────▼───────┐
│ IcebergTable│ │  Source Plan │
│    Scan     │ │  (any exec) │
│ (target,    │ │             │
│  with _file)│ │             │
└─────────────┘ └─────────────┘

SPJ-Optimized Plan (Partitioned Table)

When all partition columns appear in the join keys and use hash-compatible
transforms (Identity or Bucket), the optimizer wraps both sides with
repartitioning to eliminate cross-partition shuffles:

┌─────────────────────────────┐
│   IcebergMergeCommitExec    │
└──────────────┬──────────────┘
               │
┌──────────────▼──────────────┐
│   CoalescePartitionsExec    │
└──────────────┬──────────────┘
               │
┌──────────────▼──────────────┐
│   IcebergMergeWriteExec     │
└──────────────┬──────────────┘
               │
┌──────────────▼──────────────┐
│      IcebergMergeExec       │
└──────┬──────────────┬───────┘
       │              │
┌──────▼──────┐ ┌─────▼───────┐
│Repartition  │ │Repartition  │
│Exec (Hash   │ │Exec (Hash   │
│on _partition)│ │on _partition)│
└──────┬──────┘ └──────┬──────┘
       │               │
┌──────▼──────┐ ┌──────▼──────┐
│Projection   │ │Projection   │
│Exec (adds   │ │Exec (adds   │
│_partition)  │ │_partition)  │
└──────┬──────┘ └──────┬──────┘
       │               │
┌──────▼──────┐ ┌──────▼──────┐
│ IcebergTable│ │ Source Plan  │
│   Scan      │ │ (any exec)  │
│(target,     │ │             │
│ with _file) │ │             │
└─────────────┘ └─────────────┘

---

The following tasks are already completed on the PoC branch. Will raise formal PRs one after another as the fork repo doesn't support stacking PRs.

• feat(transaction): add RowDelta transaction action for row-level modi… #2203
• Add IcebergMergeExec with HashJoinExec integration and row classification
• Add IcebergMergeWriteExec and IcebergMergeCommitExec nodes
• Implement full MERGE execution logic with file tracking
• Integrate MERGE INTO into IcebergTableProvider
• Add comprehensive MERGE INTO integration tests
• Add partition-aware merge optimization (spark storage partition join style)

Willingness to contribute

I would be willing to contribute to this feature with guidance from the Iceberg Rust community

[tuantran0910]

Hi @wirybeaver , may I ask that ehat do you think about the performance of this merge (for example 100000 rows) when running on large table (more than 2M rows)?

[raghav-reglobe]

Hi @wirybeaver — following on from the RowDelta thread: we're starting work on the merge_into implementation for the DataFusion provider, building on the RowDelta lineage in #2678 (which we've been running against production V3 MoR tables for a few weeks now). Our workload is an SCD2 upsert expressed as MERGE INTO with a window-function USING source, so the plan is the generic hook body — join on the ON condition, apply the WHEN clauses, and commit the deletion-vector demotes + appends as a single atomic snapshot.

We'll track the DataFusion-side hook (apache/datafusion#22988) as it evolves and adapt to whatever shape lands there. If you've made progress beyond the PoC branch or have design opinions, very happy to sync so we don't duplicate — otherwise we'll share a PR here once we have something reviewable.

[wirybeaver]

Thanks for the interest, @raghav-reglobe. To clarify the current plan and avoid duplication:

I plan to own the full end-to-end implementation of MERGE INTO for CoW mode in iceberg-rust, from the DataFusion table provider integration through query planning to execution. The PoC branch referenced in the issue description already demonstrates this with comprehensive end-to-end tests as proof of the implementation.

Since GitHub doesn't support stacked PRs for forked repos, I'm landing this as a series of incremental PRs to keep each one reviewable — posting thousands of lines in a single PR would not be productive for reviewers. The open-source review process is rigorous, so the landing pace is measured.

The critical path for CoW MERGE INTO is:

1. MERGE INTO support is released in the next DataFusion version; iceberg-rust upgrades to that version.
2. RowDelta action fully supports CoW mode.
3. My incremental PRs for MERGE INTO get fully landed.

If your team has an urgent internal need for MoR, I'd suggest branching off the PoC branch to unblock yourselves in the interim — but for the upstream contribution, I'll continue driving the CoW implementation to completion.

That said, contributions are very welcome — particularly if someone wants to contribute additional query execution nodes (e.g., MergeIntoExec variants or custom physical operators) needed for MoR mode. Happy to coordinate to ensure the designs align.

The issue description is updated with Query Plans just a while ago for your reference.