Hello everyone, I would like to request reviews on three PRs I opened earlier this year to improve equality delete performance and compaction in Spark:
1. **Build equality delete map once per task** - https://github.com/apache/iceberg/pull/16956 2. **Scan-based action to remove dangling deletes** - https://github.com/apache/iceberg/pull/15727 3. **Revert disabling deletes cache for rewrites** - https://github.com/apache/iceberg/pull/15714 ### Context & Motivation I am aware of the community's general sentiment regarding equality deletes and the recent vote on deprecating them in Iceberg V4. However, for many CDC and streaming ingestion workloads, equality deletes remain the only viable option today: - Projects like [debezium-server-iceberg]( https://github.com/memiiso/debezium-server-iceberg) offer an easy and robust way to stream data from Debezium-supported databases directly into Iceberg without complex intermediate infrastructure. - I built [iceberg-flink-kafka]( https://github.com/kinolaev/iceberg-flink-kafka) based on Flink's Dynamic Iceberg Sink to upsert data from Kafka topics, since upsert mode is unlikely to return to the official Kafka Connect sink ([#14797]( https://github.com/apache/iceberg/pull/14797)). Both solutions provide automatic table creation and schema evolution with minimal maintenance overhead, and both rely heavily on equality deletes. If there is an alternative open-source approach that handles multi-table ingestion with a single configuration file without equality deletes, please let me know. In practice, equality deletes work quite well on tables with billions of rows when compacted regularly. The main pain point is Spark's read and compaction efficiency - which these PRs aim to address. --- ### 1. Build equality delete map once per task **PR:** https://github.com/apache/iceberg/pull/16956 This PR lazily builds a shared equality delete map once per task rather than recomputing it for every single data file. Additionally, it fixes double-counting of equality delete files during task weight calculation. (Trino already adopts a similar approach, which is likely a key reason it currently outperforms Spark on tables with equality deletes.) **Example:** Suppose a task processes 2 data files with ID ranges `[1, 3]` and `[2, 4]`, and 3 delete files with IDs `(1) [1, 3]`, `(2) [2]`, `(3) [4]`. Both data files reference delete files 1 and 2. Currently, Spark accounts for them twice during task building, which prevents grouping data files into a single task. Then it constructs the delete set `{1, 2, 3}` for the first data file, drops it, and rebuilds `{1, 2, 3, 4}` for the second. With this PR, Spark builds a delete map with keys `{1, 2, 3}` and their max sequence numbers as values, then simply appends key `4` when moving to the second data file without rebuilding the entire map from scratch. --- ### 2. Scan-based action to remove dangling deletes **PR:** https://github.com/apache/iceberg/pull/15727 This PR updates `RemoveDanglingDeletesSparkAction` to prune dangling delete files even when older data files (with lower sequence numbers) exist. The implementation is cleaner and more robust: it performs a table scan to find referenced deletes and removes all unreferenced delete files. **Example:** Consider the following commit history: ``` seq 4: INSERT [1000010..1000020] seq 3: DELETE [1000001..1000002] seq 2: INSERT [1000001..1000010] seq 1: INSERT [ 1..1000000] ``` Compaction will only combine small files with seq 2 and 4, leaving the data file at seq 1 as is: ``` seq 5: INSERT [1000003..1000020] (compacted data) seq 3: DELETE [1000001..1000002] (now dangling) seq 1: INSERT [ 1..1000000] ``` The delete file at seq 3 is dangling (no longer referenced by any active data file), but the current implementation fails to remove it because it does not evaluate file-level bounds/metrics. Consequently, manifests accumulate stale delete files over time. --- ### 3. Revert disabling deletes cache for rewrites **PR:** https://github.com/apache/iceberg/pull/15714 This fix is critical for rewrite/compaction performance. In Iceberg 1.10.0, delete file caching was hardcoded to `false` during compaction. This was originally introduced as a workaround because when the cache cannot fit all delete files, reading position (!) deletes in Spark degrades severely (described in detail in [#11648]( https://github.com/apache/iceberg/issues/11648#issuecomment-5387856932)). However, disabling caching across the board severely penalizes equality delete compaction: Spark must re-read the same delete file from storage for every data file referencing it, causing S3/object store request counts to skyrocket. --- Together, these changes significantly improve Spark's read throughput and compaction efficiency when dealing with equality deletes. I would greatly appreciate any reviews and feedback, and I am happy to address comments to help get these merged. Best regards, Sergei Nikolaev
