[koodin9]

This is part 1 of 4 in a series that adds CDC (upsert/delete) writes to the Kafka Connect sink
without committing equality deletes to the table.

Why

#14797 was blocked because the equality deletes committed by a streaming sink slow down every
reader until compaction runs. The v4 spec now forbids writing new equality deletes (#17783).

The sink's coordinator sees every delete file of a commit before the commit happens. So it can
resolve the deleted keys to row positions and commit deletion vectors (DVs) instead. This is the
same idea as Flink's ConvertEqualityDeletes (#15996, #17142), but it needs no staging branch,
no index state and no separate maintenance job.

flowchart LR
    W["Worker<br/>delta writer (part 3)"] -->|"data files +<br/>equality delete files"| C["Coordinator (part 2)"]
    C -->|"equality delete files"| X["EqualityDeleteConverter<br/>(this PR)"]
    X -->|"DVs"| C
    C -->|"RowDelta: data files + DVs"| T[("Iceberg v3 table<br/>no equality deletes")]

Loading

Series

• Part 1 (this PR): EqualityDeleteConverter and its benchmark. No callers, no behavior change.
• Part 2: call the converter from the coordinator and commit the result
  (iceberg.tables.convert-equality-deletes-enabled, validation, retries, docs).
• Part 3: delta writer (iceberg.tables.cdc-field, iceberg.tables.upsert-mode-enabled).
• Part 4: integration tests.

Each part is opened after the previous one is merged, so no merged state can write equality
deletes without the conversion.

What this PR adds

Class	Role
EqualityDeleteConverter	Loads the deleted keys, finds their rows in the base snapshot, writes DVs and returns a Result for the commit.
KeyPositionResolver	Interface that finds the rows holding the deleted keys. An index-backed implementation can replace the scan once the secondary index spec (#16961) lands.
ScanKeyPositionResolver	Default implementation. Plans the snapshot with pruning filters, then reads only the key columns and _pos of the candidate files. Keeps no state.
KeyFilters	Builds the pruning filters: the partition condition, and IN lists or the bounds of the leading key column.

Result maps directly onto the RowDelta that part 2 builds: baseSnapshotId for
validateFromSnapshot, dvFiles for addDeletes (and their referenced data files for
validateDataFilesExist, so a concurrent compaction fails the commit instead of resurrecting rows),
rewrittenDvFiles for removeDeletes, and conflictFilter for conflict detection.

The conversion gives the same rows as committing the equality deletes. A partitioned delete only
applies to data files with the same spec ID and partition values, also after partition evolution.
The pruning filters below only narrow the candidates.

Format version 3 only. A data file can have only one DV, so the new DV also carries the data file's
existing DV or v2 position deletes.

How the candidate files are pruned

1. Manifests are pruned by partition summaries against the delete file's partition.
2. Data files are pruned by partition values and column bounds against the key filter.
3. Row groups are pruned by their statistics (only the key filter is pushed to the reader).
4. Each row is matched exactly against the deleted keys. Rows already deleted by the existing DV or v2 position deletes are skipped.

The key filter uses IN lists (200 values per list, the evaluator limit) for up to 1,000 keys.
Above that it uses the bounds of the leading key column, which can keep most files of the
partition when the keys are scattered; the benchmark below covers this.

Benchmark

EqualityDeleteConverterBenchmark (JMH) times one conversion on an unpartitioned table, where the
key filter alone decides which data files are read. It varies the table size, how keys lie in the
data files (increasing IDs, unsorted or sorted UUIDs), which rows change (recent, scattered, new
keys from upsert), and keys per commit across the IN-list limit. It was planned in
this thread, and the results
are in this comment.

The tables are on local disk by default. BENCHMARK_CATALOG_PROPERTIES names a catalog properties
file instead, and kafka-connect/kafka-connect-runtime/docker/benchmark-catalog.properties uses
the REST catalog and RustFS of the integration tests.

Adding object store latency with toxiproxy

docker compose -f kafka-connect/kafka-connect-runtime/docker/docker-compose.yml up -d iceberg
docker run -d --name toxiproxy --network docker_default -p 8474:8474 -p 19000:19000 \
  ghcr.io/shopify/toxiproxy:2.12.0
curl -X POST localhost:8474/proxies \
  -d '{"name":"s3","listen":"0.0.0.0:19000","upstream":"object-store:9000"}'
# delay every S3 response by 20 ms
curl -X POST localhost:8474/proxies/s3/toxics \
  -d '{"name":"latency","type":"latency","stream":"downstream","attributes":{"latency":20}}'
sed 's#localhost:9000#localhost:19000#' \
  kafka-connect/kafka-connect-runtime/docker/benchmark-catalog.properties > /tmp/toxiproxy.properties
BENCHMARK_CATALOG_PROPERTIES=/tmp/toxiproxy.properties \
  ./gradlew :iceberg-kafka-connect:iceberg-kafka-connect:jmh \
  -PjmhIncludeRegex=EqualityDeleteConverterBenchmark \
  -PjmhOutputPath=benchmark/eq-delete-converter-benchmark.txt

Only S3 requests go through the proxy. The REST catalog is reached directly.

Not covered yet: partitioned layouts, bloom filters, larger data files, and read latency compared
with the same commits written as equality deletes. The envelope is where the conversion finishes
within the commit interval. Past it, the conversion still runs and costs commit time, not
correctness.

Testing

• TestEqualityDeleteConverter (17 tests) on a real v3 table: conversion of earlier commits,
  merging with an existing DV or v2 position deletes, partition pruning (identity and bucket),
  file pruning for clustered and scattered keys and past the IN-list limit, a custom resolver,
  branches including one that does not exist yet, a concurrent append of a deleted key that the
  commit's validation rejects, and rejection of non-equality deletes and format version 2.
• TestKeyFilters (2 tests): the files kept by IN lists and by the bounds.
• ./gradlew -DkafkaVersions=3 :iceberg-kafka-connect:iceberg-kafka-connect:check passes.

---

AI Disclosure

• Model: Claude Opus 5.5
• Platform/Tool: Claude Code
• Human Oversight: fully reviewed
• Prompt Summary: Implement and refine a converter that resolves equality deletes of a Kafka Connect commit to deletion vectors, add a benchmark for it, and simplify the PR description

[koodin9]

@bryanck I'd appreciate your review, since you raised the read-performance concern that stopped #14797.

This PR is part 1 of 4. It adds EqualityDeleteConverter, which resolves the keys of a commit's equality delete files to row positions in the base snapshot and writes deletion vectors, following the approach of Flink's ConvertEqualityDeletes but run inside the sink's coordinator before the commit. The later parts wire it into the coordinator, add the delta writer, and add integration tests. Ordering the series this way means no merged state ever produces equality deletes without the conversion in place.

Nothing calls the converter yet, so there is no behavior change here. If the direction looks wrong to you, I'd rather hear it on this PR than after the writer lands.

[github-actions]

This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions.

[koodin9]

still alive

[koodin9]

@laskoviymishka Could you take a look at this PR, or point me to the right reviewer?

Some context: #14797 (CDC/upsert for the sink) was blocked because it committed equality deletes, and it was closed as stale. Since then, the v4 spec forbids writing new equality deletes (#17783), and Flink converts them to deletion vectors with ConvertEqualityDeletes (#17142).

This series follows the same direction for Kafka Connect. The coordinator resolves the equality deletes of a commit to row positions and commits deletion vectors instead, so the table never contains an equality delete. This PR (part 1 of 4) only adds the converter class. Nothing calls it yet, so there is no behavior change.

Before I continue with the coordinator wiring, is this direction acceptable for the sink, or should I start a [DISCUSS] thread on dev@ first?

Thanks!

[laskoviymishka]

Thanks for the patience on this one, and for ordering the series so no merged state writes equality deletes. On your direction question: converting in the coordinator follows where the project already went with #17783 and Flink's ConvertEqualityDeletes (#17142), so it looks right to me for the sink. I'd still like @bryanck to confirm, since his read-cost concern is what stopped #14797.

For this part, the one thing I'd want before merge is the empty-branch path: it returns an alwaysFalse conflict filter, so a concurrent first commit can keep rows the equality delete would have removed. The version check and the test gaps are inline.

[koodin9]

@laskoviymishka
Thanks for the review. I rebased on main and pushed one commit per comment; each inline reply names its commit.

The last commit, 286bc46, drops the comparison budget and the range splitting, so it is mostly deletions in KeyFilters and its tests. It partly undoes two earlier commits: the range test added in b0e5bc8 now runs through the bounds path, and the splitIntoRanges tests that 8ec2676 moved are removed together with the method. I kept both commits so each comment has its own change, but none of the removed code remains in the final diff.

While fixing the empty-branch path, I also found that a branch that does not exist yet was treated as empty, although a commit to it builds on the current snapshot of main. 92ee01a handles both cases.

[laskoviymishka]

The round-1 threads all landed, thanks. This is a second pass; I'll dive deeper once this round is fixed. Code-wise this is close, but the direction should settle before merge.

That direction is bigger than this stack: v4 drops equality deletes (#17783), and this should eventually be how Kafka Connect does CDC. Could you open the dev@ [DISCUSS] thread you offered? Flink's ConvertEqualityDeletes (#15996, #17142) converts after commit as a stateful maintenance task; this converts before commit with a stateless scan, worth aligning there. @bryanck, @mxm, could you weigh in?

One proposal for that thread: make the CDC path v4-only. On v3, equality deletes are still allowed, and converting them costs a scan on every commit that gets expensive for scattered keys (see the KeyFilters thread). The conversion is worth that cost where equality deletes aren't an option, which is v4. A v4 gate would also let part 3 take keys from the primary-key constraint (#17822) instead of identifier-field-ids, and the format check here would move from >= 3 to >= 4. v3 tables would stay append-only, and CDC would wait until v4 is usable end to end. wdyt?

@pvary, could you check the resolver seam in the existingDeletes() thread?

[laskoviymishka]

Could the tests commit a Result through a small helper, covering a concurrent compaction of a DV's data file and a concurrent in-key append on a non-null base? Conflict validation only covers append and overwrite, so only validateDataFilesExist catches the compaction, and the commit() helper never calls it.

[koodin9]

Added in b02bb42.

The tests now commit a Result through a new helper, commit(Table, Result, DataFile...). It applies the validations the Result Javadoc asks the commit for:

• conflictDetectionFilter(result.conflictFilter()) with validateNoConflictingDataFiles() and validateNoConflictingDeleteFiles()
• validateDataFilesExist(...) for the data files the new deletion vectors point to
• validateFromSnapshot(result.baseSnapshotId()) when the result has a base snapshot

testConflictFilterCoversKeysAppendedAfterBase covers the concurrent in-key append on a non-null base: another writer appends a row with a deleted key after the conversion, and the commit fails with a ValidationException. testConflictFilterCoversKeysWithoutSnapshot is the same case without a base snapshot, and now uses the same helper.

The other tests keep the old helper. Nothing commits concurrently there, so these validations cannot fail.

As you suggested, the compaction case waits for part 2, where it can be tested against the coordinator's real RowDelta.

[laskoviymishka]

@pvary, does this obligation fit the index in #16961? Every resolver has to return the file's existing DV and v2 position deletes, and once the new DV exists, a null for v2 files silently brings their deleted rows back. The draft spec doesn't define whether an entry points back to its row, so an index-backed resolver may have no positions to give.

[laskoviymishka]

Above 1,000 keys this falls back to one min/max range on the first key column. For scattered keys (UUIDs, hashes, updates across history) that means every commit reads the key columns of every file in the partition, or the whole table if unpartitioned, with no cap on planned files. The 1,200-key test already plans all 30 of 30 files. The same range is the conflict filter, so a retry repeats the scan. What envelope (keys per commit, files per partition, key distribution) should the part-2 benchmark cover, and what happens when plannedDataFiles() goes past it?

[koodin9]

What envelope (keys per commit, files per partition, key distribution) should the part-2 benchmark cover?

• Keys per commit: from a few keys to well beyond the IN-list limit, so that both the IN lists and the min/max range are covered.
• Data files per partition: from small partitions to ones where reading the key columns of every file takes longer than the commit interval.
• Key distribution, in two parts:
  • Which rows a commit changes: recent rows, rows scattered across history, and new keys, since upsert mode also deletes the key of every insert.
  • How keys lie in data files: increasing IDs, UUIDs after a sorting compaction, and unsorted UUIDs.
• Layout: unpartitioned, identity-partitioned, and bucketed by the key, each with and without bloom filters on the key columns.

For each case, the benchmark measures conversion time against the commit interval, planned data files, and bytes and manifests read, and compares read latency with the same commits written as equality deletes. The envelope is where the conversion still finishes within the commit interval.

What happens when plannedDataFiles() goes past it?

The conversion still runs. Going past the envelope costs commit time, not correctness. In part 2, the coordinator will:

• Report the planned data files and the conversion time of each commit.
• Warn when a conversion takes longer than the commit interval.
• On a conflict, re-run the conversion (and its scan) against the new snapshot, and fail the commit if it keeps conflicting.

The benchmark results also say how to stay inside the envelope or widen it:

• Layouts that stay inside it: sorting or bucketing by the key, bloom filters, compaction. These go into the docs.
• The first pruning change to make: several ranges instead of one min/max range, one manifest pass for all partition groups, or skipping row groups with bloom filters.
• Beyond that, the resolver seam is where an index (Spec: Secondary index specification #16961) can take over.

As you asked, I'll add a short version of this plan to the description.

[laskoviymishka]

I dug into the v3 part a bit more after the review, and I think I was too strict there. There is a good reason to support this on v3 too, not only v4.

Ryan’s point in the thread behind #17783 is that DV-based writes on v3 give writers a clean upgrade path before the table itself moves to v4 (thread). Flink’s converter also targets v3. v4 can come later once the converter writes colocated DVs.

The cost is on the write side. An equality-delete writer just writes keys. Here we have to read key columns from candidate files before commit. But after that, readers only need the DV instead of doing the equality-delete join on every read, and clients without equality-delete support can still read the table. For Kafka Connect I think that is probably the right trade.

The cases I’d want to understand better in the dev@ thread are mostly the bad key distributions:

• random keys don’t prune well by bounds, so a commit can end up reading key columns from most files in the touched partitions
• with bucketed tables, touching many buckets can also mean planning the same manifests several times because planning currently runs once per partition group (here)
• all of this runs on the coordinator, and a conflict means doing it again

For part 2, I’d benchmark conversion time vs commit interval, plus bytes/manifests read. Recent/scattered/random keys are probably enough to show where this starts getting expensive. I’d also compare that with read latency after accumulating equality deletes for a while. Bloom filters on the key columns are worth trying too.

For this part, I think I only need the concurrent in-key append test on a non-null base, plus a short benchmark plan in the description. The compaction case can wait for part 2 when there is a real RowDelta.

The >1,000-key fallback is what I asked for in round 1, so I’m fine with it here. existingDeletes() also doesn’t block this part.

[koodin9]

@laskoviymishka Thanks for the second pass and the follow-up. Agreed on v3: CDC writes target v3 now, and v4 follows once the converter writes colocated DVs.

I started the dev@ discussion on the direction: https://lists.apache.org/thread/gmkj55oc0k1k72gobsylgfc9ngk9lyxn

[laskoviymishka]

@koodin9 thread went to spam for me, but i did reply onto this, i think for me now main question is from strategic perspective - do we want to disrupt current status quo or no.

[koodin9]

@laskoviymishka Thanks, I replied on the thread.

@bryanck @rdblue, the dev@ thread is now on whether the sink should support CDC. You were part of the original decision to leave it out because of equality deletes, so your view would help.

In short: the coordinator converts each commit's equality deletes into deletion vectors before committing, so the table never gets an equality delete. CDC would be opt-in, and append-only stays the default.

[koodin9]

I added the benchmark to this PR (2c593a4), since Péter asked on the dev@ thread to measure the conversion before anything is merged. Here are the first results.

Setup: a 14-core Mac, an unpartitioned v3 table with 1,000 rows per data file, and 1,000 deleted keys per commit. The table is in RustFS behind toxiproxy, which delays every S3 response by a fixed latency. The worker pool has 14 threads unless noted. Each time is one conversion. How to run it, including the toxiproxy setup, is in the PR description.

Table	Data files read	0 ms	10 ms	20 ms	50 ms	50 ms, 64 threads
10,000 files, random UUID keys	10,000	2.2 s	9.5 s	18.5 s	42.4 s	11.4 s
10,000 files, sorted UUID keys	~1,000	0.38 s	1.2 s	2.1 s	4.6 s	1.5 s
1,000 files, random UUID keys	1,000	0.30 s	1.0 s	2.0 s	4.3 s	1.4 s
10,000 files, increasing IDs, recent rows	1	0.11 s	0.19 s	0.24 s	0.38 s	0.39 s
10,000 files, increasing IDs, new keys	0	0.09 s	0.11 s	0.14 s	0.20 s	0.20 s

• The time is close to the data files read times the request latency, divided by the threads. The number of deleted keys barely changes it.
• When file bounds can prune the keys, as with increasing IDs, a conversion stays under half a second even at 50 ms.
• Random UUID keys read every data file, so the time grows with the table, as Péter expected. I expect bloom filters to help little there, because each file still needs a request. That case needs an index, or partitioning that narrows the files.
• More threads help almost linearly, but the worker pool is shared by the whole JVM.
• These are lower bounds. The latency is fixed with no tail, the files are small, and there is no TLS.

Next, I'd like to add partitioned layouts and larger data files, and run compaction alongside repeated conversion commits.

[pvary]

What are the expectation of the time for the commits? Is a worst case full read acceptable?

[koodin9]

@pvary

What are the expectation of the time for the commits?

Well below the commit interval. Our target is about 3 minutes from a change to its visibility. With a commit interval of 60 to 90 s, the conversion should take at most about 80 s, or about 40 s so that one retry still fits.

Is a worst case full read acceptable?

Not on large tables, I think. When the changes are spread over the key space, every commit reads the key column of the whole table. Our two largest CDC tables are unpartitioned, with about 150 to 160 data files and 1.7 and 2.4 billion rows. Amoro already reads all of these files after each commit to turn the equality deletes into position deletes. It runs one task per data file in parallel, and still takes 14 s and 43 s. I haven't measured the conversion on our storage yet, but doing that in one coordinator on every commit is too expensive.

So I'd document in part 2 that the conversion suits tables where the key filter or the partition spec prunes most data files, such as increasing keys with changes on recent rows. Tables with changes spread over many rows need an index behind KeyPositionResolver, as a follow-up.

[pvary]

Thanks for the answers @koodin9!

These were my main questions. I leave the decision to people with more Kafka knowledge to decide how to proceed.