nordisk/skade

MERGE INTO — the row-level write path

Status: landed 2026-09-03, lane SK. Builds on lane FR (b4613a9, the InsertOp refusal) and lane FR's merge-on-read proof.

The hole this fills

Before this, skade could append and it could delete by equality. It could not change a row in place through SQL. Concretely:

The refusal was not a skade decision. DataFusion 55 plans MERGE INTO end to end — datafusion-sql/src/statement.rs:2422 merge_to_plan → WriteOp::MergeInto → datafusion/src/physical_planner.rs:838 — and then calls TableProvider::merge_into, whose default is not_impl_err!. No provider in this tree implemented it. And it could not: the vendored iceberg crate's src/transaction/ had exactly one data verb, fast_append. There was no overwrite, rewrite-files, RowDelta or delete-file action, and upstream iceberg-rust 0.10.1 (newest, Aug 2026) still has none — issues #1104 "Support RowDeltaAction" and #735 "Delete support" are open with prerequisites in progress.

So this lane is two pieces that only work together:

1. iceberg::transaction::OverwriteFilesAction — a real Iceberg overwrite commit, in the vendored core.
2. IcebergTableProvider::merge_into — the DataFusion hook that drives it.

Piece 1 — OverwriteFilesAction

vendor/iceberg-arrow59/src/transaction/overwrite.rs, beside append.rs, in the same shape.

The generic machinery was already there and already named the operation: SnapshotProducer::summary (snapshot.rs:421) branches on operation() == Operation::Overwrite, and the SnapshotProduceOperation trait doc (snapshot.rs:80) says an overwrite "may exclude manifests for partitions being overwritten". existing_manifest IS the seam where replaced files are dropped. FastAppendOperation uses it to carry everything forward; this uses it to carry most things forward.

let tx = Transaction::new(&table);
tx.overwrite_files()
    .add_data_files(new_files)          // the rewritten rows
    .delete_files(replaced_paths)       // the files they replace
    .apply(tx)?
    .commit(catalog)
    .await?;

One snapshot, operation = overwrite, one catalog pointer swap.

How a data file is removed

A manifest is immutable, so removing a file means writing a new manifest in its place. Per data manifest in the current snapshot's manifest list:

Dropping a wholly-dead manifest is what FastAppendOperation::existing_manifest already does (has_added_files() || has_existing_files()), and it is what stops a table merged N times from accumulating N empty manifests. The vendored crate has no expire_snapshots, so nothing needs the dropped tombstones.

Delete manifests (ManifestContentType::Deletes — the equality-delete files delete_equality writes) are carried forward untouched. This action replaces DATA files; it does not reason about somebody else's merge-on-read tombstones.

The three refusals inside the action

• adds nothing and removes nothing → PreconditionFailed. A no-op that produces a snapshot is worse than an error: it grows the metadata tree and lets a caller believe a merge did something. This refusal is why merge_into can skip the commit entirely instead of "successfully" writing nothing.
• the same path removed and re-added → DataInvalid. The rewritten manifest would carry the path as both Deleted and Added in one snapshot.
• a named removal is not live → DataInvalid, listing the paths. This is what makes "a failed merge leaves the table exactly as it was" true rather than "the merge dropped half its work": a stale view of the table fails the whole commit, it does not partially apply.

Piece 2 — IcebergTableProvider::merge_into

vendor/iceberg-datafusion-arrow59/src/physical_plan/merge.rs + the hook in src/table/mod.rs.

DataFusion hands the provider (datafusion-session-55.0.0/src/table.rs:394):

• source: Arc<dyn ExecutionPlan> — the USING clause,
• merge_schema: DFSchemaRef — target columns then source columns, target qualified by the real table name (the SQL alias is canonicalised away in merge_to_plan step 6),
• on: Expr — the ON condition over that combined schema,
• clauses: Vec<MergeIntoClause> — kind + optional AND predicate + action.

Every expression is compiled to a PhysicalExpr at plan time, against the caller's &dyn Session. That is the only point where a user-defined function inside a clause can still be resolved: the returned plan executes with a TaskContext only.

Execution, per data file

for each live data file F:
    read F's rows            (through ArrowReader → schema evolution + any
                              committed equality-delete file already applied)
    FULL hash join F ⋈ source on the ON equality keys
    per joined row, pick the FIRST clause of the applicable category whose
      predicate holds
    if no row of F was updated or deleted  →  leave F alone
    else                                   →  write F's surviving/updated rows
                                              to a new data file, mark F removed
then: source rows that matched nothing anywhere → the INSERT clauses → one more
      new data file
then: ONE overwrite_files commit, or none at all

The join is HashJoinExec in PartitionMode::CollectLeft with the file on the build side, so the cost is O(rows in F + source rows) per file rather than a nested loop. Unmatched-vs-NULL is disambiguated by a synthetic non-null ordinal appended to the END of each side (__iceberg_merge_target_ordinal, __iceberg_merge_source_ordinal), so the real columns keep the indices merge_schema gave them; the two ordinals are projected back out before any compiled expression sees the batch.

The count

The single count: UInt64 row is updated + deleted + inserted, each counted off the work that produced the commit. It is never the input row count. The test the_count_is_the_warehouses_not_the_inputs pins it in both directions — one merge whose count is below the source row count (4 source rows, 2 matched → 2) and one above it (1 source row, 3 target rows deleted → 3). Returning source.num_rows() fails the first; returning the target row count fails both.

What executes and what refuses, by name

NotMatched and NotMatchedByTarget are collapsed through MergeIntoClauseKind::canonical(), as the DataFusion type doc requires.

Copy-on-write versus merge-on-read — the measurement

skade-sql/tests/merge_cow_vs_mor.rs, oden, 2026-09-03, release build, local filesystem. 200 000 rows in 10 data files; 8 rounds of 100 updated keys.

Keys SPREAD across all 10 files (worst case for copy-on-write):

after 8 rounds on disk: cow 18.8 MB · mor 3.9 MB

Keys CONCENTRATED in 1 file (best case for copy-on-write):

after 8 rounds on disk: cow 5.2 MB · mor 3.9 MB

Read it honestly, in both directions:

• Merge-on-read wins the write, by a lot. ~2 ms against ~34 ms — a 17× cheaper commit — and 0.04 MB against 1.91 MB, a 48× smaller write. It does not degrade with rounds.
• Copy-on-write wins the read, by more, and the gap grows. The full scan goes 38 → 173 ms over 8 merge-on-read rounds (4.6×, linear in the number of delete files) while copy-on-write stays flat at 4–5 ms.
• The crossover is at less than one read per merge. Copy-on-write costs +32 ms once, at write time; merge-on-read costs +31 ms on the first read after the same merge, and more on every read after that. A table that is read even once between merges is already better off copied.
• Disk: copy-on-write's total is bounded by the size of the files it rewrites (18.8 MB spread / 5.2 MB concentrated vs 3.9 MB), and old snapshots keep the originals alive until they are expired.

Copy-on-write is the default, and MERGE INTO is copy-on-write. The argument is the read column, and it is measured rather than asserted: merge-on-read's read cost is not a constant tax, it is a slope, and a warehouse table is read far more than it is merged.

Merge-on-read stays available and unchanged as skade::Table::delete_equality + append, and it is the right call for exactly two shapes: a high-frequency CDC sink where the reader is a delta consumer rather than a full scan, and a table compacted often enough that the delete files never accumulate. Neither is reachable from SQL, on purpose — MERGE INTO should not silently pick the mechanism that makes the next reader slower.

What is NOT done

• Partitioned tables. Refused by name. A rewritten row needs its partition values recomputed; the projection that does that (project_with_partition) sits on the INSERT path ahead of a writer this path does not use.
• Non-equi ON. Refused by name rather than degrading to a nested loop over the whole table.
• A merge-on-read MERGE INTO. There is no RowDeltaAction and no position-delete writer in the transaction layer, so MERGE INTO has exactly one mechanism. If merge-on-read ever becomes selectable it should be a table property (write.update.mode, Iceberg's own spelling), not a second statement.
• Concurrent-merge retry. Two merges racing on the same table: the second fails its RefSnapshotIdMatch (or the not-live-file check) and the caller retries. Nothing is corrupted; nothing is retried for you.

Upstreamability

overwrite.rs (≈330 lines) has no skade type, no korp vocabulary and no assumption about skade's catalog in it — it is an Iceberg transaction action that happens to live here, written to be offered against iceberg-rust #1104 / #735. The only edit it needs in an existing upstream file is three pub(crate) accessors on SnapshotProducer (snapshot_id, commit_uuid, key_metadata), because existing_manifest is handed &self and cannot reach the &mut self manifest-name counter, plus two lines in transaction/mod.rs.

merge.rs is upstreamable in the same sense but against a moving target: it is written for DataFusion 55's merge_into hook, which is newer than any iceberg-datafusion release.