Thanks for writing this up, and for the detailed plan on the PR. I think the design is right, but only if we as a community want CDC in the sink at all. That question should come first.
I don’t think the current setup is bad. The sink can stay append-only, and Spark or another engine can do the MERGE. With Spark, that scales out, keeps row IDs on updates, and doesn’t need a new update/delete path in Kafka Connect. `DebeziumTransform` already adds the `_cdc` fields needed for MERGE. The reason we ended up there was equality deletes. The delta writers proposed for the sink committed them, and that made tables hard to read until compaction ran [1][2]. That part is different now. v3 has deletion vectors, v4 doesn’t allow new equality deletes, and the v4 discussion moved delete resolution to the writer [2]. Flink has to do that too on v4. So I think it’s reasonable to reopen the old decision. But I’d still like to hear from Bryan and Ryan, since they were part of the original call. [1] https://lists.apache.org/thread/96dhf3sj5pc4ql0l8yk8sxgtr78bchrd [2] https://lists.apache.org/thread/wfd3s65dpyzto8g8poffcg1tcqpgk5k7 best, Andrei On Mon, Oct 26, 2026 at 8:54 AM koodin 9 <[email protected]> wrote: > Hi all, > > The review of #18003 asked to settle the direction for CDC (upsert and > delete) writes in the Kafka Connect sink on this list. Here is the proposal. > > Background > - #14797 added a delta writer to the sink but was not merged, mainly > because equality deletes committed by a streaming sink slow down every > reader until compaction runs. v4 now forbids writing new equality deletes > (#17783). > - Flink converts equality deletes after they are committed. > ConvertEqualityDeletes (#15996, #17142) is a maintenance task that keeps a > primary key index in Flink state and converts the equality deletes of a > staging branch, or of the write branch in place, into deletion vectors. > > Proposal > The sink's coordinator sees every file of a commit before committing it. > It resolves the batch's equality deletes against the base snapshot, writes > deletion vectors (merged with each data file's existing DV or v2 position > deletes), and commits the data files and the DVs in one RowDelta. The > equality delete files are never committed, not even as a fallback. > > - Lookup: a stateless scan. Candidate files are pruned by the delete's > partition and the deleted keys, and only the key columns and _pos are read. > The lookup is behind an interface, so an index (#16961) can replace the > scan later. > - Concurrency: the RowDelta validates from the base snapshot, with the > deleted keys as the conflict filter, and checks that the data files the DVs > point to still exist, which catches a concurrent compaction. On a conflict, > the coordinator converts again against the new snapshot, and fails the > commit if conflicts keep coming. > - Series: part 1 is the converter (#18003, in review), part 2 calls it > from the coordinator and adds a benchmark, part 3 adds the delta writer, > part 4 adds integration tests. > > Why before the commit, without state > - Kafka Connect has no keyed state or checkpoints to keep an index in, and > a sink writing to many tables would need an index per table. The same > memory concern came up for Flink's dynamic sink in #17142. > - Readers never see equality deletes, and clients that cannot apply them > can still read the table. > > Cost > The write side pays: every commit reads the key columns of the candidate > files, so the cost follows the number of files that can hold the deleted > keys. The expensive cases are keys that file bounds cannot prune (such as > random UUIDs), upsert mode, where every insert also deletes its key, > bucketed tables, where planning currently runs once per partition group, > and conflicts, which repeat the conversion. All of it runs on the > coordinator. Part 2 will benchmark the conversion time against the commit > interval and compare it with reads over accumulated equality deletes (plan > in the PR description). Past that envelope the conversion still runs, so > the cost shows up as commit latency rather than as equality deletes in the > table. > > Format versions > - v3: deletion vectors let writers move off equality deletes before their > tables move to v4. Flink's converter also targets v3. > - v4: new DVs must be colocated with their data files, so v4 support comes > once the converter writes colocated DVs. > - v2: there are no deletion vectors, so CDC writes would not be supported > on v2 tables. > > Questions > 1. Is converting in the sink's commit path, before the commit and without > state, the right direction for Kafka Connect, alongside Flink's post-commit > conversion? > 2. Is it acceptable that CDC writes require v3 or later, and that going > past the envelope costs commit latency instead of falling back to equality > deletes? > 3. Should the deleted keys come from the identifier fields for now, and > from the primary key in the v4 constraints proposal (#17822) once it lands? > > Links > - #18003: https://github.com/apache/iceberg/pull/18003 > - Benchmark envelope discussion: > https://github.com/apache/iceberg/pull/18003#discussion_r4374820473 > - #14797: https://github.com/apache/iceberg/pull/14797 > - #15996 and #17142 (Flink ConvertEqualityDeletes): > https://github.com/apache/iceberg/pull/15996, > https://github.com/apache/iceberg/pull/17142 > - Earlier thread on Flink's conversion: > https://lists.apache.org/thread/gldqrycmo24r0vo77wzcj9s9b04rtvy7 > - #17783: https://github.com/apache/iceberg/pull/17783 > - #17822: https://github.com/apache/iceberg/pull/17822 > - #16961: https://github.com/apache/iceberg/pull/16961 >
