iceberg: add a copy-on-write merge strategy to the output

CHANGELOG.md

@@ -3,6 +3,18 @@ Changelog
 
 All notable changes to this project will be documented in this file.
 
+## 4.104.0 - 2026-08-04
+
+### Added
+
+- iceberg: Add an opt-in `merge_strategy: copy-on-write` so `upsert` and `delete` land as plain data files readable by engine-backed catalogs such as Snowflake and the Databricks Unity Catalog. ([@Jeffail](https://github.com/Jeffail), [#4666](https://github.com/redpanda-data/connect/pull/4666))
+
+### Fixed
+
+- iceberg: No-timezone `timestamp` columns are now written to parquet with the spec-correct `isAdjustedToUTC=false` annotation; the encoding is pinned per table via the `redpanda-connect.timestamp-encoding` property so existing tables never change or mix encodings. ([@Jeffail](https://github.com/Jeffail), [#4666](https://github.com/redpanda-data/connect/pull/4666))
+- iceberg: Row-mutation commits are now idempotent across ambiguous catalog responses, preventing duplicate rows on redelivery after a commit whose response was lost; mutation snapshots carry a `redpanda-connect.commit-id` summary property. ([@Jeffail](https://github.com/Jeffail), [#4666](https://github.com/redpanda-data/connect/pull/4666))
+- iceberg: Table properties a catalog refuses to accept (for example Unity Catalog's managed keys) are now learned from the rejection and stripped from subsequent commits instead of failing every mutation. ([@Jeffail](https://github.com/Jeffail), [#4666](https://github.com/redpanda-data/connect/pull/4666))
+
 ## 4.103.1 - 2026-07-31
 
 ### Added

cmd/tools/integration/packages.json

@@ -25,7 +25,7 @@
   {"path":"./internal/impl/gcp/enterprise/changestreams"},
   {"path":"./internal/impl/gcp/enterprise/changestreams/metadata"},
   {"path":"./internal/impl/hdfs"},
-  {"path":"./internal/impl/iceberg/integration"},
+  {"path":"./internal/impl/iceberg/integration","timeout":"15m"},
   {"path":"./internal/impl/influxdb"},
   {"path":"./internal/impl/jira"},
   {"path":"./internal/impl/kafka","timeout":"10m"},

docs/modules/components/pages/outputs/iceberg.adoc

@@ -122,6 +122,7 @@ output:
     case_sensitive_columns: true
     row_operation: insert
     identifier_fields: []
+    merge_strategy: merge-on-read
     storage:
       aws_s3:
         bucket: my-iceberg-data # No default (required)
@@ -231,11 +232,11 @@ By default this output is append-only — every message becomes a new row (`row_
 
 `row_operation` supports interpolation, so the operation can be driven by the data itself — for example by mapping a change-data-capture stream's operation field — but no CDC-specific format is assumed (see the change-data-capture example below). It is named `row_operation` to distinguish it from Iceberg's snapshot-level operation.
 
-`upsert` and `delete` require `identifier_fields` and use Iceberg merge-on-read equality deletes, which require table format version 2. A version-1 table is automatically upgraded to version 2 on the first `upsert`/`delete`; *this upgrade is irreversible*.
+`upsert` and `delete` require `identifier_fields`. How those mutations are materialised on disk — and, critically, which query engines can then read the table — is controlled by `merge_strategy`. See <<merge-strategies,Merge strategies: merge-on-read vs copy-on-write>> below for the decision guide, support matrix, and maintenance guidance.
 
-*Identifier fields.* `identifier_fields` must reference existing table columns of a primitive, non-floating-point type. A static `upsert`/`delete` is validated at startup; an interpolated `row_operation` is validated per message at write time, so an empty `identifier_fields` is not caught until the first `upsert`/`delete` message arrives. Identifier columns of a temporal type (`timestamp`, `timestamptz`, `date`, `time`) must arrive as time values, not bare numbers — a numeric epoch is ambiguous as a delete key and is rejected at write time; convert it to a timestamp upstream. If the table is partitioned, every partition source column must be one of the `identifier_fields`, since equality deletes are partition-scoped.
+*Identifier fields.* `identifier_fields` must reference existing table columns of a primitive, non-floating-point type. A static `upsert`/`delete` is validated at startup; an interpolated `row_operation` is validated per message at write time, so an empty `identifier_fields` is not caught until the first `upsert`/`delete` message arrives. Identifier columns of a temporal type (`timestamp`, `timestamptz`, `date`, `time`) must arrive as time values, not bare numbers — a numeric epoch is ambiguous as a delete key and is rejected at write time; convert it to a timestamp upstream. If the table is partitioned, `merge-on-read` additionally requires every partition source column to be one of the `identifier_fields`, since equality deletes are partition-scoped; `copy-on-write` carries no such restriction (it rewrites whole files by filter and routes new rows by value, and can even move a key across partitions). See the support matrix for the merge-key types each strategy accepts.
 
-When this output auto-creates a table (via `schema_evolution`), the `identifier_fields` columns are created as *required* and registered as the table's Iceberg identifier-field-ids, so downstream engines and other writers see the primary key. A consequence is that a null or missing value in an identifier column is rejected on write, even for `insert`. Identifier columns must therefore be present at creation — in the first message or declared via `schema_metadata`. Pre-existing tables are never modified.
+Under `merge-on-read`, when this output auto-creates a table (via `schema_evolution`), the `identifier_fields` columns are created as *required* and registered as the table's Iceberg identifier-field-ids, so downstream engines and other writers see the primary key. A consequence is that a null or missing value in an identifier column is rejected on write, even for `insert`; identifier columns must therefore be present at creation — in the first message or declared via `schema_metadata`. Under `copy-on-write` the identifier fields are used only as the connector-side merge key and are *not* registered as identifier-field-ids, so auto-created columns are not forced required (this is also what lets engine-backed catalogs such as the Databricks Unity Catalog, which rejects identifier-field-ids at table creation, accept the `CREATE TABLE`). Pre-existing tables are never modified.
 
 *Batching and ordering.* Within a single batch the last `upsert`/`delete` per `identifier_fields` key wins. Each batch containing an `upsert`/`delete` is committed as its own snapshot (these commits are never coalesced, which is required for correctness), so a high-throughput mutation workload produces one snapshot per batch. Size batches accordingly and run regular table maintenance (snapshot expiry and compaction) to keep metadata manageable. Pure `insert`-only batches keep the original append fast path, which does coalesce commits.
 
@@ -246,6 +247,52 @@ Ordering only holds *within* a batch. With more than one batch in flight, concur
 `insert` is an unconditional append and is *not* keyed or de-duplicated. For keyed data (including change-data-capture), map create/read events to `upsert`, never `insert` — mixing `insert` with `upsert`/`delete` on the same key in one batch produces duplicate rows.
 ====
 
+[[merge-strategies]]
+=== Merge strategies: merge-on-read vs copy-on-write
+
+`merge_strategy` controls how `upsert`/`delete` mutations are written, which sets both the write cost and — most importantly — which query engines can read the table.
+
+* `merge-on-read` (the default) writes Iceberg v2 equality-delete files and applies them at read time. Writes stay cheap and streaming-friendly, but only catalog-native / Flink-world engines (Apache Polaris, Flink, Trino, Spark) can read equality deletes. Engine-backed catalogs — Snowflake and the Databricks Unity Catalog — cannot. It requires table format version 2, so a version-1 table is automatically upgraded to version 2 on the first `upsert`/`delete`; *this upgrade is irreversible*.
+* `copy-on-write` instead rewrites whole data files so the table only ever holds plain data files — no equality- or positional-delete files. Every engine that reads Iceberg data files can read the result: Snowflake and the Databricks Unity Catalog as well as Polaris, Flink and Trino. Because it writes only plain data files it works on version-1 or version-2 tables and never forces the irreversible v1->v2 upgrade.
+
+*Which to choose.*
+
+* Choose `merge-on-read` for streaming or high-throughput mutation into a lake read by Polaris, Flink, Trino or Spark, where write cost must stay low.
+* Choose `copy-on-write` when the table must be correct on Snowflake or the Databricks Unity Catalog (or any engine that cannot read equality deletes), and the workload is batch or moderate-throughput so the write amplification is acceptable.
+
+*Copy-on-write support matrix.*
+
+* *Column types:* all flat primitives (`boolean`, `int`, `long`, `float`, `double`, `string`, `date`, `time`, `timestamp`, `timestamptz`, `decimal`, `uuid`, `binary`, `fixed`), and nested `struct`/`list`/`map` columns whose leaves are all supported primitives.
+* *Merge-key (`identifier_fields`) types:* `int`, `long`, `string`, `date`, `time`, `timestamp`, `timestamptz` and `uuid`. `decimal` and `boolean` merge keys are *not* supported and error with a message pointing you at `merge-on-read` (an upstream limitation in the Iceberg library's overwrite filter — it cannot apply a `decimal` or `boolean` predicate when rewriting files); both are fine as non-key columns.
+* *Partitioned tables:* supported, with no requirement that the partition columns be a subset of `identifier_fields`. A `copy-on-write` `upsert` can even move a key from one partition to another.
+* *Table format:* version 1 or version 2, with no forced upgrade.
+
+*Write amplification and throughput.* `copy-on-write` rewrites every data file that contains a touched key: a batch of K keys scattered over M files rewrites roughly K/M of the table, and touching even a single key in a file rewrites that whole file — so a one-row change to a 512 MB file rewrites all 512 MB. To keep amplification low, sort the table by the identifier key so a batch's keys cluster into as few files as possible, and use large batches. This is a batch / moderate-throughput mode, not a streaming one.
+
+*Memory.* Under `copy-on-write` the whole new-row batch is materialised in memory as a single Arrow record while the batch commits, so a keyed batch's memory scales with its total row bytes. Size keyed batches to stay within the process memory budget rather than making them arbitrarily large.
+
+*Maintenance.* Because every mutating batch rewrites files and adds a snapshot, a high-churn `copy-on-write` workload accumulates data files and snapshots quickly. Run regular table maintenance: compaction (rewrite / bin-pack data files), snapshot expiry, and orphan-file removal. Orphan-file removal matters specifically because a `copy-on-write` commit that fails *ambiguously* (the catalog may or may not have recorded it) can leave newly-written data files unreferenced; the connector cleans these up best-effort, but periodic orphan-file removal is the definitive backstop.
+
+*Copy-on-write limitations.*
+
+* A `decimal` merge key is not supported — use `merge-on-read` for a decimal key (a `decimal` non-key column is fine).
+* Schema evolution covers new *top-level* columns only; new fields appearing inside an existing nested `struct`/`list`/`map` column are not auto-surfaced for evolution.
+* It is a batch / moderate-throughput mode: expect heavy write amplification under scattered, high-frequency keyed mutations.
+* Tables pinned to the legacy timestamp encoding whose schema contains a no-timezone `timestamp` column reject `upsert`/`delete` — see <<timestamp-encoding,Timestamp encoding on existing tables>> for why and for the migration path.
+
+[[timestamp-encoding]]
+=== Timestamp encoding on existing tables
+
+Older versions of this output annotated no-timezone `timestamp` columns in the parquet files they wrote with `isAdjustedToUTC=true` — the annotation the Iceberg spec reserves for `timestamptz`. The stored microsecond instants are correct, and appends and most readers are unaffected, but the annotation makes some readers treat the column as UTC-adjusted, and it prevents `copy-on-write` from rewriting those files (the file's annotation reads back as `timestamptz`, which cannot be written into a `timestamp` column). Current versions write the spec-correct `isAdjustedToUTC=false`.
+
+To guarantee an existing table never ends up with a mix of the two annotations, the encoding is pinned *per table* via the table property `redpanda-connect.timestamp-encoding` (`spec` or `legacy`):
+
+* Tables created by this output carry `redpanda-connect.timestamp-encoding: spec` from creation.
+* For an existing table without the property, the output resolves the encoding automatically on first contact and stamps the result onto the table: if the schema has no no-timezone `timestamp` column, or the table has no data files, it resolves `spec`; otherwise the output inspects one data file's parquet footer and adopts whatever that file already contains (`legacy` for `isAdjustedToUTC=true`). A table that cannot be probed (unreadable file) fails the write rather than risk mixing annotations.
+* Once stamped, the property is authoritative and the probe never runs again. An unrecognised property value is a hard error.
+
+A table pinned `legacy` keeps receiving the legacy annotation on every new file — byte-identical to what previous releases wrote — so appends and `merge-on-read` continue working unchanged forever. The one restriction is mutating `copy-on-write` (`upsert`/`delete`): it must rewrite existing files, which the legacy annotation prevents, so such writes fail upfront with an actionable error (pure `insert` batches still work). To migrate a legacy table to the spec encoding: rewrite/compact the table's data files with an engine that writes the spec annotation (e.g. Spark's `rewrite_data_files`), then set the table property `redpanda-connect.timestamp-encoding` to `spec`, keeping any running instances of this output that write to the table stopped (or restarting them) around the migration — a live writer only re-reads the property when its writer is recreated. Alternatively, keep the table on `merge-on-read`.
+
 
 == Performance
 
@@ -297,6 +344,53 @@ output:
         region: us-east-1
 ```
 
+--
+CDC into a Snowflake- or Databricks-readable table (copy-on-write)::
++
+--
+
+Materialize a change-data-capture stream into an Iceberg table that must be read by an engine-backed catalog such as Snowflake or the Databricks Unity Catalog, which cannot read merge-on-read equality deletes. `merge_strategy: copy-on-write` rewrites whole data files so the table only ever holds plain data files that every engine can read. As with any keyed workload it requires `max_in_flight: 1`; because copy-on-write rewrites every file containing a touched key, prefer large batches and a table sorted by the identifier key, and run regular compaction and snapshot expiry.
+
+```yaml
+input:
+  redpanda:
+    seed_brokers: [ localhost:9092 ]
+    topics: [ dbserver.inventory.customers ]
+    consumer_group: iceberg_sink_cow
+
+pipeline:
+  processors:
+    - mapping: |
+        meta op = match this.op {
+          "d" => "delete",
+          _   => "upsert",
+        }
+        root = this.after | this.before
+
+output:
+  iceberg:
+    catalog:
+      url: http://localhost:8181/api/catalog
+    namespace: inventory
+    table: customers
+    row_operation: ${! metadata("op") }
+    identifier_fields: [ id ]
+    # copy-on-write produces only plain data files, so Snowflake and the
+    # Databricks Unity Catalog can read the result (unlike equality deletes).
+    merge_strategy: copy-on-write
+    # Keyed writes must stay ordered; a single batch in flight prevents a stale
+    # update from overwriting a newer one for the same key.
+    max_in_flight: 1
+    # Amortise the file rewrites over larger commits.
+    batching:
+      count: 5000
+      period: 30s
+    storage:
+      aws_s3:
+        bucket: my-iceberg-data
+        region: us-east-1
+```
+
 --
 ======
 
@@ -683,6 +777,25 @@ identifier_fields:
   - user_id
 ```
 
+=== `merge_strategy`
+
+How `upsert` and `delete` are materialised on disk.
+
+* `merge-on-read` (the default) writes Iceberg v2 equality-delete files. Deletes are applied at read time, so writes stay cheap and streaming-friendly, but only catalog-native / Flink-world engines can read the result — engine-backed catalogs such as Snowflake and the Databricks Unity Catalog cannot read equality deletes.
+* `copy-on-write` rewrites whole data files so the table only ever contains plain data files (no delete files), which every engine can read — including Snowflake and Databricks Unity Catalog. It works on version-1 or version-2 tables and never forces the irreversible v1->v2 upgrade. The trade-off is heavy write amplification: each mutating batch rewrites every data file that contains a touched key, so it is a batch / moderate-throughput mode. Sort the table by the identifier key and use large batches so each rewrite touches as few files as possible.
+
+See the <<merge-strategies,Merge strategies>> section above for the full decision guide, copy-on-write support matrix (column and merge-key types, partitioning, table format), and maintenance guidance.
+
+
+*Type*: `string`
+
+*Default*: `"merge-on-read"`
+
+Options:
+`merge-on-read`
+, `copy-on-write`
+.
+
 === `storage`
 
 Storage backend configuration for data files. Exactly one of `aws_s3`, `gcp_cloud_storage`, or `azure_blob_storage` must be specified.

internal/impl/iceberg/catalogx/catalog.go

@@ -1,4 +1,4 @@
-// Copyright 2025 Redpanda Data, Inc.
+// Copyright 2026 Redpanda Data, Inc.
 //
 // Licensed as a Redpanda Enterprise file under the Redpanda Community
 // License (the "License"); you may not use this file except in compliance with
@@ -319,6 +319,25 @@ func (c *Client) loadCatalog() catalog.Catalog {
 	return c.catalog.Load()
 }
 
+// TableIO returns a table.CatalogIO view of the client — the interface a
+// *table.Table binds its commits and refreshes to. It resolves the client's
+// current underlying REST catalog on every call, so the returned value stays
+// valid across the auth-driven catalog refreshes the client performs
+// internally.
+func (c *Client) TableIO() table.CatalogIO {
+	return clientTableIO{c}
+}
+
+type clientTableIO struct{ c *Client }
+
+func (t clientTableIO) LoadTable(ctx context.Context, ident table.Identifier) (*table.Table, error) {
+	return t.c.loadCatalog().LoadTable(ctx, ident)
+}
+
+func (t clientTableIO) CommitTable(ctx context.Context, ident table.Identifier, reqs []table.Requirement, updates []table.Update) (table.Metadata, string, error) {
+	return t.c.loadCatalog().CommitTable(ctx, ident, reqs, updates)
+}
+
 // isNamespaceAlreadyExists checks if the error indicates the namespace already exists.
 func isNamespaceAlreadyExists(err error) bool {
 	return errors.Is(err, catalog.ErrNamespaceAlreadyExists)