raphaelpfl opened a new issue, #17542: URL: https://github.com/apache/iceberg/issues/17542
## Motivation The pre-donation project ([tabular-io/iceberg-kafka-connect](https://github.com/tabular-io/iceberg-kafka-connect), now archived) advertised "Row mutations (update/delete rows), upsert mode" as a feature, configured via `iceberg.tables.cdc-field` (op codes `I`/`U`/`D`) and `iceberg.tables.upsert-mode-enabled`. Neither made it into the code donated to `apache/iceberg` - the current `kafka-connect` module is append-only. Users are still hitting this gap with the *exact* old config keys, which are now silently ignored (unknown connector properties): #15046. Both the OP and a commenter report duplicate rows in Iceberg even with `upsert-mode-enabled`/`cdc-field` set, using a real Debezium source. Digging into *why* it wouldn't have worked even if those keys were still wired up: the old CDC docs only ever show a value that carries the op code inline (e.g. `_cdc_op: "I"/"U"/"D"`) - never a delete event with a null value. In practice, with a Debezium source and the `ExtractNewRecordState` SMT, delete events arrive at the sink as a standard Kafka tombstone (null value, key only) regardless of `delete.handling.mode` - not as a rewritten row carrying an op field. So the original design would not have handled a real Debezium delete correctly even if it had survived the donation. ## What we prototyped and validated We (Junto Seguros) built a fix on top of `apache-iceberg-1.11.0` and tested it end-to-end against a real pipeline: Debezium (SQL Server) -> Kafka -> this sink -> Iceberg (REST catalog / AWS S3 Tables). We've since rebased it cleanly onto current `main` (no conflicts touching any of the files below). Summary: * **`cdc-field`** (`iceberg.table.<table>.cdc-field`, with a connector-wide default `iceberg.tables.default-cdc-field`): the named record-value field is compared against Debezium op codes `c`/`r` (insert), `u` (update: delete then insert), `d` (delete). Backed by Iceberg equality-deletes (`BaseEqualityDeltaWriter`), not a table rewrite. * **Delete via tombstone**: a null-value record, with `cdc-field` enabled and a non-null key, is treated as delete-by-key. This is the gap described above - it works regardless of `delete.handling.mode` and isn't Debezium-specific (any CDC source emitting standard tombstones on delete works). * **Partitioned tables**: a key-only delete can't compute a partition value (no full row available), so it's written as an equality-delete under an *unpartitioned* spec from the table's spec history; Iceberg applies it against every partition on read. Tables auto-created by the connector always get such a spec for free (created unpartitioned, then evolved). A pre-existing partitioned table without that history can opt into a one-time spec-evolution migration (`cdc-auto-migrate-partition-spec`) that adds one without changing how future writes are partitioned. * **Topic-based routing** (`route-by-topic-enabled` + `dynamic-topic-namespace` + `topic-table-regex`): routes by Kafka topic name instead of a record-value field, so it works correctly for null-value tombstones (unlike `dynamic-enabled`, which inspects the value and would silently drop every delete) and lets one connector instance fan out to hundreds/thousands of source tables (one topic per table, Debezium-style naming) without a connector-per-table topology. All of this is covered by a JUnit suite (161 tests, including the module's pre-existing suite) using the module's existing `InMemoryCatalog`/`WriterTestBase` conventions, plus manual end-to-end validation against a live pipeline (including a schema-evolution-during-CDC scenario). Working branch, rebased onto current `main`, split into small reviewable commits (config -> writer -> wiring -> topic routing -> partitioned-table support -> docs): https://github.com/raphaelpfl/iceberg/tree/cdc-upsert-delete-support - posted for early visibility, not as a ready-to-merge PR; opening the PR itself once there's some directional alignment on the questions below. ## Open questions for maintainers, before a PR A few things affect public config surface and would benefit from maintainer input before we invest in a polished PR: 1. **Backward-compat naming**: reuse the exact pre-donation key (`iceberg.tables.cdc-field`, connector-wide) so existing tutorials/blog posts/#15046 itself start working again, or keep our per-table shape (`iceberg.table.<table>.cdc-field` + `iceberg.tables.default-cdc-field`)? We can support both if that's preferred. 2. **Op code casing**: the old design used uppercase `I`/`U`/`D`; our prototype matches Debezium's lowercase `c`/`r`/`u`/`d`, since that's what we tested against. Should this be configurable (an op-code mapping), fixed to one convention, or support a couple of well-known conventions out of the box? 3. **`upsert-mode-enabled`**: the old design had this as an *independent* flag (equality-delete before every insert, no `cdc-field` needed - useful for idempotent/exactly-once-ish reprocessing without real deletes). We haven't implemented this; open to adding it as a follow-up if there's interest. 4. **Topic-based routing vs. extending `dynamic-enabled`**: is a new routing mode the right shape, or would maintainers prefer `dynamic-enabled` itself became null-safe (e.g. falling back to the topic name when the record value is null)? 5. **`cdc-auto-migrate-partition-spec`**: any preference between an opt-in flag (our approach, since it mutates table metadata - two partition-spec evolutions), always attempting it automatically, or leaving that kind of migration to a separate offline procedure entirely? ## Contribution We can turn this into a PR with small, reviewable commits (config -> writer -> tombstone handling -> topic routing -> partitioned-table support -> tests -> docs) once there's rough alignment on the questions above. Part of this implementation was drafted with AI assistance (Claude); all code has been reviewed line-by-line, tested end-to-end against a real pipeline, and is covered by the JUnit suite mentioned above, per the project's [AI-assisted contribution guidelines](https://iceberg.apache.org/contribute/#guidelines-for-ai-assisted-contributions). cc @laskoviymishka @AnatolyPopov, since you both reviewed the most recent Kafka Connect changes - would appreciate your take on the open questions above whenever you have a moment. --- **Query engine**: Kafka Connect **Willingness to contribute**: I would be willing to contribute this improvement/feature with guidance from the Iceberg community -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
