Hi Andrei, To your 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? Yes, the goal is to get rid of equality deletes, which we can only do by not writing them anymore. >- 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? >From the user perspective, it shouldn't matter if the implementations are engine specific, as long as the writes are correct. However, I think we should try to share as much as possible for the V4 milestone. The code in https://github.com/apache/iceberg/pull/18003 looks modular enough to plug in a different implementation. Your proposal mentions that there won't be an index for now, but ad-hoc scanning to resolve the equality deletes is the only option. I have some doubts how scalable this approach is going to be. It surely wastes a lot of IO, even if a slight delay (evident in https://github.com/apache/iceberg/pull/18003#issuecomment-6573941243) is acceptable. All in all, a reasonable first approach to getting rid of equality deletes. Thanks, Max On Fri, Nov 13, 2026 at 10:51 AM Andrei Tserakhau via dev <[email protected]> wrote: > > 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
