Hi Andrei, Thanks for the reply.
On whether the sink should do CDC at all: append plus MERGE works, and it should remain available. But the history suggests the sink was meant to get CDC back once the concerns about equality deletes were addressed: - After the initial sink submission, Bryan opened #10842 [3]. Delta writer support had been left out because of performance concerns over equality deletes, and the issue says: "We should address those concerns and add back the delta writer support, for features like change data capture and upsert mode." It is still open. - In [1], Ryan objected to the sink writing equality deletes, and Bryan hoped the sink would get a solution like the one Flink was working on, for streaming deletes without equality deletes. - Users keep asking for it: #13986, #15046, #15342, #15351, #17542. This proposal is meant to be that solution: the table never gets an equality delete. CDC would be opt-in, and append-only stays the default. Append plus MERGE remains the better fit for two kinds of tables: - Tables whose readers follow rows across updates by row ID. The sink writes an update without reading the old row, so the new row cannot carry the old row ID and gets a new one. The spec describes the same for equality delete writers such as Flink's upsert, which don't read existing rows either [4]. Spark's MERGE reads the old row and keeps its row ID. - Tables where the conversion can't keep up with the commit interval. It runs on the coordinator, and the part 2 benchmark will show where that point is. For other tables, the sink keeps the target table current at the commit interval, without a separate MERGE job per table. On repartitioning, which Ryan named as the main challenge for writing DVs from Kafka Connect [2]: for CDC topics, Kafka already does it. Debezium uses the primary key as the message key, Kafka's default partitioner sends equal keys to the same partition, and Kafka Connect assigns each partition to one task. So within a commit, all changes to a row reach the same task, and the task resolves repeated changes itself. Rows from earlier commits are found by the coordinator's scan, so no keyed state is needed. CDC mode would require topics keyed this way, which is Debezium's default. Bryan, Ryan, since you made the original call, I'd like to hear whether this changes it for you. [1] https://lists.apache.org/thread/96dhf3sj5pc4ql0l8yk8sxgtr78bchrd [2] https://lists.apache.org/thread/wfd3s65dpyzto8g8poffcg1tcqpgk5k7 [3] https://github.com/apache/iceberg/issues/10842 [4] https://iceberg.apache.org/spec/#row-lineage Thanks, 2026년 10월 27일 (화) 오전 5:44, Andrei Tserakhau via dev <[email protected]>님이 작성: > 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 >> >
