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
