Upserts into Iceberg are easy… in batch workloads. But in streaming architectures, and especially high-frequency CDC, this is still very much a data engineering challenge. We explained the nature of that challenge in our previous article. Today we're covering the way we solve it at Etleap, focusing on the architecture we use to convert equality deletes into deletion vectors (DVs) - which allows us to deliver low-latency Iceberg tables (<2 minutes) that work in any query engine.
If you've already read Why Streaming Updates Break Iceberg Compaction, feel free to skip to the next section.
Iceberg files are immutable, so a CDC update is a delete of the old row plus an insert of the new one. Deletes are the tricky part - there are two ways to do them in an Iceberg lakehouse, and both come with tradeoffs.
Equality deletes are fast to write but slow to read (and unsupported by major readers). An equality delete is key-based: “delete the row where id = 42”. On write, the key comes straight from the change event - so a writer such as a Flink upsert sink can emit one delete per changed key without reading the table. But at read time, Iceberg has no key-row index, so every query has to check each row it scans against the whole set of deleted keys. The performance hit is so bad that Snowflake and Databricks won't read a table that contains equality deletes.
Deletion vectors (DVs) are fast to read but slow to write. A DV is position-based: it says "delete row 7 of file X". That makes reading cheap: the engine skips the marked rows in the file it's already reading. Writing, however, is expensive: the writer has to figure out which files contain the rows that need deleting, which means a lookup against the table for every changed key.
To keep the lookup out of the streaming job, pipelines commonly append changes to a staging table and run a scheduled Spark MERGE INTO in merge-on-read mode. Less-frequent merges impact data freshness. More frequent ones are more likely to collide with already-running compaction jobs - which will throw a Cannot commit, missing data files error. Equality deletes do not have this problem because the delete references column values, not file paths. Workarounds that make collisions rarer - retries, pausing ingestion, compacting less often - cost you either latency or query performance.
Type of operation | Write | Read | Compaction conflict |
|---|---|---|---|
Equality delete | Fast: needs only the key from the change event, no table read | Slow: every query checks each row against the deleted keys. Snowflake and Databricks won't read the table | None, if compaction runs with |
Deletion vector | Slow: needs a lookup to find each key's file and row | Fast: the engine skips the marked rows | Yes, whichever commits second fails: |
Branching is a very powerful, and often underutilised, feature of Apache Iceberg. Similarly to Git commits, Iceberg table snapshots follow a lineage, and that lineage can have separate branches.
At Etleap, we use branching to convert key-based equality deletes into deletion vectors, while maintaining near real-time latency, and with minimal compaction conflicts.
The core of our approach is that we write into two separate branches of the same table:
realtime is the write branch: a Flink streaming job commits new data files and equality deletes to this branch as changes arrive. Since we are only writing equality deletes at this stage, we can maintain very low-latency ingestion.
main is the read branch: it references the same data files but with DVs in place of equality deletes. Since here we have only DVs, we get a version of the table that's read-optimised and queryable in any engine (including Snowflake and Databricks).
A single process writes to main, so no other job can add an equality delete to it or collide with its commits.
Compaction runs on realtime. Since this is the equality deletes branch, there are no collisions between jobs (see above).
Why one table instead of two? Because both branches belong to one table, they can share data files. Nothing needs to be copied, and snapshot expiry rules prevent a file referenced by either branch from being deleted. Working with one table also leads to lower operational costs since there is one catalog entry to manage rather than two.
The main challenge in implementing this solution is to keep both branches in sync, so that query engines read data from main with near real-time data freshness - typically under two minutes behind the source.
To achieve this, we implemented the conversion as a continuous Flink streaming job that runs independently of ingestion and table maintenance. Each cycle, it picks up every realtime snapshot committed since the previous cycle and combines them into one set of changes. To minimise added latency, it avoids unneeded reads and works with metadata wherever it can.
The sync happens in three stages, each running as an operator within the same Flink job:
Differ: works out what changed, from metadata alone. It reads the manifests of the new realtime snapshots and lists the files added and removed: new data and delete files, plus any files a compaction replaced. Because it never opens a data file, it's fast; and because it is only concerned with changes at the file level, it can replicate any change from one branch to the other. The differ emits metadata instructions that can be used to apply the change on the other branch, as well as to pinpoint the files that need to be scanned in order to convert the equality deletes into DVs.
Converter: finds the rows impacted by an equality delete. This is the only stage that reads data files, and thus the most expensive one. It receives just the parts of the diff that involve new equality deletes or compacted files, scans the older data files those deletes could apply to, and writes one DV per data file. The converter runs as a parallel fleet that can be scaled horizontally: each data file goes to exactly one converter instance, so it's read only once per cycle. If a data file already has a DV, its entries are merged into the new one, because the Iceberg spec allows only one DV per data file. Like the differ, the converter outputs only metadata.
Committer: turns it all into one snapshot on main. It collects the output of the differ and every converter instance, applies the combined list of metadata changes, and commits them to main as a single snapshot. It keeps each file's original sequence number, which is key to avoiding compaction conflicts (more on that below).
Let's see what this looks like in practice.
Say you run an online store and replicate its orders table into Iceberg. So far there are six purchase orders, numbered 1 to 6. Then, two new updates are streamed in as customers change and cancel their orders. Let's say the first update changes order 2, cancels order 5, and adds order 7. In the second update, a customer changes order 4 and cancels order 1. For the sake of our example, if an order is cancelled, it is deleted from the data model.
In our Iceberg table, what we would see is this:
Orders 1-6 sit in two data files that are on both branches: A (sequence number 3) contains orders 1-3, and B (sequence number 4) contains orders 4-6.
The ingestion job commits each update to realtime as soon as it lands.
Update 1 writes data file C (the updated order 2, and order 7) plus equality delete E1 (ids 2 and 5, as well as 7), with sequence number 5.
Update 2 writes data file D (the updated order 4) plus equality delete E2 (ids 1 and 4), sequence number 6.
On its next cycle, the conversion job picks up both commits and passes them through the three stages:
Differ (1) identifies the changed files: it reads only the manifests of the two new snapshots, without opening data files, and finds two new data files and two equality deletes. The new delete files are paired with existing data files. Both E1 and E2 affect files A and B, so the differ generates pairings of (A, E1, E2) and (B, E1, E2). Sequence numbers and column statistics are used where possible to prune the files that need to be checked.
Converter (2) finds the impacted rows: the pairings emitted by the differ are allocated to the converters. One converter instance will look up the positions of the rows in E1 and E2 and write a DV for file A. Another instance will do the same for file B. Any existing DVs for those data files are merged into the output DV. E2 may also apply to file C, so the converters will do the same lookup, but because no deletes in E2 actually match rows in file C, no DV is emitted.
Committer (3) builds the matching snapshot on main: it adds the new data files by reference, plus the DVs, in a single commit. The equality deletes never reach main.
main now shows orders 2 (updated), 3, 4 (updated), 6 and 7 (new), with no equality deletes for any engine to trip over. Two ingest commits are handled in a single cycle of the conversion job and one commit on main, and the only files the job wrote were the DVs.
One of the main challenges with merge-on-read based implementations are compaction conflicts. A DV points at a specific data file; if a row is deleted in a file that a compaction is currently replacing, the compaction fails with Cannot commit, found new delete for replaced data file. The compaction then needs to be retried, which leaves more time for new deletes to land, so on a busy table compaction can fail again and again, creating a "doom loop" of slower queries and worse latency.
Our solution eliminates the most painful parts of this problem. Compaction only runs on the realtime branch, where every delete is an equality delete - so it doesn't matter that compaction moves rows into a new file. As long as use-starting-sequence-number is enabled, the new file keeps the old sequence number, and deletes that arrive mid-compaction still apply to it. When it comes to syncing the branches, the conversion job mirrors the compaction on main like any other change, swapping the old files for the compacted one.
This approach shifts the conflict from being between ingest and compaction, to being between two consecutive delete conversions, which is a much more manageable problem.
Let’s look at an example:
We have two files on the realtime branch that are being compacted. At the same time an update is also made to the realtime branch. These two operations don't conflict because the update writes equality deletes.
The conversion job may replicate that update to the main branch before the compaction has completed. The converted DV will then reference a data file that the compaction is about to replace.
After the compaction completes, the compacted file will still contain the rows that were deleted after the compaction began. When the conversion job mirrors the compaction on the main branch, it must recognise that the new data file has an earlier sequence number than the equality delete and perform the conversion a second time, rewriting the DV to point at the compacted file.
In the worst-case, we must convert that same equality delete file twice. If the compaction had completed before the first conversion ran, then the delete file would only need to be converted once. Any further compactions would remove the delete file, so this never happens a third time.
Either way, it's a fixed amount of repeated work which you can budget for. There's no longer the possibility of "doom loops" that require interventions such as pausing real-time ingestion to apply fixes.
Consistently fresh data. Changes from the source typically reach main in under two minutes. Ingestion doesn't pause for compaction. Compaction can't get stuck in a retry loop. If the conversion job falls behind, it catches up in a single run instead of handling each realtime commit separately.
Readable in your data warehouse. main holds only DVs, so Snowflake and Databricks can query it like any other Iceberg table.
Data is not copied or rewritten. Data files cross from realtime to main as pointers; the conversion job only writes DVs. You don't spend compute or storage on copying data.
Each conversion benefits from previous cycles. When a new equality delete applies to a data file that's already on main, the converter reads that file together with its current DV. Rows removed by earlier deletes are already marked, so the converter never has to read old equality deletes again.
Fast and efficient queries. Since main is using position-based deletes, query engines can read the data they need without slow and costly lookups. And because compaction conflicts cost less and happen less frequently, we get more consistent compaction that further improves query performance and reduces costs.
main can trail realtime, briefly. Readers on main see a change only after the next conversion run. In practice the gap is small, as we schedule the conversion job to run more often than ingest commits.
A coarser history on main. Each run combines every new realtime commit into one commit on main. The result is that time travel on main moves in steps of one conversion, not one ingest commit. We could keep a one-to-one history, but it may lead to wasted work as we’ll need to convert deletes that later compactions have already removed.
No row lineage. The v3 spec's row lineage doesn't cover rows updated through equality deletes. The writer doesn't read the old row, so it can't carry the row's ID over, and updates count as a delete plus a brand-new row. However, this is true for any pipeline that writes equality deletes.
The conversion still does a lookup. Turning an equality delete into a DV means scanning the data files it could apply to. Bigger tables mean more work per run.
That last point leads us to our plans for the future of streaming pipelines:
Iceberg tables currently do not have an index beyond min-max values attached to each file. With a true, relational database-style index from key to file and row, the converter could go straight to the rows it needs - which would make each run faster, especially on large tables. It would also reduce S3 costs, specifically GET requests.
This is still a work-in-progress on our end, so watch this space. In any case, we expect branching to stay central to resolving compaction conflicts.
So far we've been looking under the hood. Let's briefly look above it.
In Etleap, the process we described runs behind an ordinary pipeline. You create it in a few clicks: pick a source such as Postgres and the table you want, add any transformations (PII masking, say), and choose where to load it. The pipeline takes an initial snapshot of the table, then streams changes through Flink into realtime. The conversion job keeps main in sync, and compaction and snapshot expiry run automatically, as the table needs them.
Here you can see how an Etleap pipeline running on a table of more than 400 million records picks up a batch of cancelled trades in Postgres, and a couple of minutes later they show up in a Snowflake query:
_VIDEO_:https://www.youtube.com/watch?v=9FlOQgePGAU
If you're building this yourself, the ingredients needed are:
Two branches on one table.
A streaming Flink job that mirrors file changes from one to the other and converts equality deletes on the way. It needs to be fast enough not to become its own bottleneck - in our solution, that means reading only metadata wherever possible, and limiting file reads to once per cycle.
Merging new deletes into each file's existing DV, and keeping every file's original sequence number on main.
Compaction with use-starting-sequence-number on the write branch only.
A rule to reconvert any deletes that landed while a compaction was running.
If you'd rather not build it, you know where to find us.
