Hi all, Bumping this with a short summary, since I'd like to hear from a few people before we decide.
Where it stands: - The proposal: the sink's coordinator converts each commit's equality deletes into DVs before committing, so the sink never commits an equality delete in any configuration. CDC is opt-in and append-only stays the default. v3 now, v4 as the target [1]. - koodin9 posted a benchmark on #18003 [2]: with increasing IDs a conversion stays under 0.4 s, while random UUID keys on a 10,000-file table read every file and take 2 to 42 s depending on storage latency. koodin9 agrees a full read is too expensive on large tables, so part 2 would document the mode for tables that prune well, with an index (#16961 [3]) as the path for the rest. - At the Oct 28 community sync [4], nobody argued for dropping upserts from the sink. The position was to add nothing that produces new equality deletes. I went back to the code after the sync, and two things I said there were off. The mainline sink is append-only today; the upsert mode with equality deletes was in Tabular's version. And #18003 [5] doesn't convert in a background thread: the coordinator converts inside the commit. Questions: - Delta writers were left out of the sink because of equality deletes [6][7]. Does a sink that never commits one address that? If not, what would? - Flink converts after the commit with keyed state [8][9]; this converts before the commit with a scan. Should the two converge for v4, or is it fine for them to differ? Short answers are fine. Even a +1 or -1 on the direction helps. [1] Spec change forbidding new equality deletes in v4: https://github.com/apache/iceberg/pull/17783 [2] Benchmark results for the conversion: https://github.com/apache/iceberg/pull/18003#issuecomment-6573941243 [3] Secondary index spec: https://github.com/apache/iceberg/pull/16961 [4] Oct 28 community sync recording, from 40:39: https://youtu.be/fPTtFTNf_Tc?t=2439 [5] Part 1, the equality delete to DV converter: https://github.com/apache/iceberg/pull/18003 [6] Earlier Kafka Connect delta writer PR, not merged: https://github.com/apache/iceberg/pull/14797 [7] Issue to add delta writer support back to the sink: https://github.com/apache/iceberg/issues/10842 [8] Flink ConvertEqualityDeletes maintenance task: https://github.com/apache/iceberg/pull/15996 [9] Flink ConvertEqualityDeletes integration with IcebergSink: https://github.com/apache/iceberg/pull/17142 best, Andrei On Tue, Oct 27, 2026 at 7:17 AM Péter Váry <[email protected]> wrote: > I'm OOO now, so I just put my first thoughts here. I hope I can come back > later. BTW: Max knows quite a bit of the topic, you can try to reach out to > him. > > - Definitely do EQ delete resolution performance testing before commit > anything to the code - I suspect without an index it will be slow (without > caching, maybe even with an index as well) > - Consider compactions and possible other writes to the table. This will > became bigger problem as the resolution becomes faster. Concurrent commits > might get starved. > - For base snapshot we always use the current snapshot of the table, as > CDC commits stored with equality deletes could not conflict with any > concurrent writes. > > Thanks, Peter > > On Tue, Oct 27, 2026, 05:27 koodin 9 <[email protected]> wrote: > >> 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 >>>> >>>
