Sixty Thousand Deletes, Nothing Deleted

The bookshop’s orders table, sixty thousand rows in PostgreSQL, snapshotted through Debezium and Kafka into an empty Iceberg table by a Flink upsert job.

  initial snapshot: CONVERGED in 7s — rows 60,025, sum(order_id) 1,801,830,364
  2 data files, deletes {content: (files, rows)} = {2: (2, 60,025)}

Seven seconds to converge, and every check passes. And the table that holds sixty thousand freshly inserted rows also holds sixty thousand equality-delete rows, one per insert, in two delete files. Nothing was deleted. An upsert sink does not know whether a key exists. It expresses every write as “delete this key from anything older, then insert”, and it does that for the first row of an empty table exactly as it does for the thousandth update of a busy one. Every checkpoint of a change-data table is half deletes — by construction — before the source has deleted anything.

Getting the source changes into Iceberg is only half the job. The landing table has to converge after a burst, repeated changes to one key, and a full replay. Downstream consumers then need to read what changed — a request the landing table turns out to handle poorly. The second table introduced below solves that separate problem.

The pipeline, and the three things that broke it once

The source is PostgreSQL 18 with wal_level=logical, feeding Debezium 3.6.2 on Kafka Connect using pgoutput, and Kafka 4.3.1. Flink 2.1.3 reads the topic in its debezium-json format into an Iceberg table with a primary key and write.upsert.enabled, through Polaris. Chapter 1 stood this pipeline up. Three failures in the Gate 1 lab explain the setup choices used in every run below.

The source has to send the whole old row. PostgreSQL’s default replica identity sends only the key on an update, and Flink’s changelog format needs the previous row to retract it. The first update after a clean snapshot failed the job with “the before field of UPDATE message is null”. ALTER TABLE orders REPLICA IDENTITY FULL is the fix, at the source, before the connector exists.

A deleted connector is not a forgotten one. Connect keeps offsets by connector name, so a connector re-registered under its old name resumes at a log position whose replication slot is gone. Every run here names its connector, slot and publication with a timestamp.

A consumer group remembers too. Flink’s Kafka source and, chapter 11 noted, the Connect sink both keep committed offsets under their group name, so earliest-offset only applies to a group that has never existed. Per-run group names, again.

The convergence test

The pipeline needs a check at the destination: has Iceberg caught up with PostgreSQL? This function compares row counts, counts per status and sums of two key columns on both sides, then polls until they agree.

def pg_state():   # PostgreSQL
    n, s1, s2 = psql("SELECT count(*), sum(order_id), sum(customer_id) FROM orders")
    by = dict(psql("SELECT status, count(*) FROM orders GROUP BY 1"))
    return n, s1, s2, by

def ice_state():  # Iceberg, through Spark
    n, s1, s2 = sql("SELECT count(*), sum(order_id), sum(customer_id) FROM ice.stage15.orders_cdc")
    by = dict(sql("SELECT status, count(*) FROM ice.stage15.orders_cdc GROUP BY 1"))
    return n, s1, s2, by

def converge(timeout=300):
    want = pg_state(); t0 = time.time()
    while time.time() - t0 < timeout:
        if ice_state() == want: return time.time() - t0
        time.sleep(3)
    raise AssertionError(f"not converged: {want} vs {ice_state()}")

A count alone passes when an update was lost; a status distribution alone passes when two rows swapped states; the key sums catch a row replaced by a different row. Together they are cheap and they were wrong zero times in this chapter, which is the property a convergence test needs before it is scheduled against production.

  initial snapshot:      CONVERGED in 7s
  after the burst:       CONVERGED in 4s     (1,000 updates, 100 inserts, 50 deletes, one transaction)
  after the churn:       CONVERGED in 4s     (one key: five updates, a delete, a re-insert)
  replayed table:        CONVERGED in 3s     (the whole topic again, into a second table)

The columns the test compares are the ones that moved. Across the burst, the count went from 60,025 to 60,075, placed fell by 900 and shipped rose by a thousand, cancelled fell by fifty, and the sum of order ids rose by the hundred new identities. A test that checked the count alone would have passed a run that lost every update. Four seconds is one checkpoint interval, which is the floor: a change is in the table at the next commit after it arrives, and never sooner. The bookshop’s freshness number for its change tables is the checkpoint interval plus Debezium’s lag, and chapter 13 puts both on the dashboard.

Ordering: one key, seven changes

One order is enough to test whether successive changes survive the pipeline. Here it is updated five times in quick succession, deleted, then re-inserted with the same id. The final row must reflect the re-insert.

  order 8031: postgres reordered|7  /  iceberg [status='reordered', customer_id=7]
  4 data files, deletes = {1: (1, 5), 2: (4, 61,177)}

The final state matches, and the file listing shows how. The five updates that landed inside one checkpoint became five position deletes against that checkpoint’s own data file — each update retracted the previous one in the same file. The delete-and-re-insert became an equality delete for the key and a new row with a higher sequence number. The re-inserted row wins over everything older by the rule that equality deletes only apply to older files. Ordering within a key is Kafka’s guarantee, partitioned by key, and the upsert sink turns that order into sequence numbers. What was not tested is ordering across partitions, which Kafka does not guarantee and this table does not need, because every row’s history is one key.

Idempotency: a replay converges

Chapter 11 replayed a savepoint into an append table and got seventy-three thousand duplicates. Here the replay starts at the topic’s earliest offset and writes into a second change-data table. Comparing the two tables shows whether replay reconstructs the same final state.

  a second job, from the earliest offset, into a second table
  replayed table: CONVERGED in 3s
  orders_cdc EXCEPT orders_replay: 0 rows;  the other way: 0 rows -> identical
  replay: 1 data file, deletes = {1: (1, 1,056), 2: (1, 60,126)}

Identical, row for row, because an upsert is idempotent by key — replaying the topic re-applies every change in order and lands on the same final state. The replayed table’s layout is also a fair picture of what a replay costs. The whole history was consumed in one checkpoint: one data file, and 1,056 position deletes, one for every change to a key that the same checkpoint had already written. Those 1,056 are the burst’s thousand updates and fifty deletes plus the churned key’s six, resolved inside a single file, which is what the same changes cost the original run across four checkpoints and four files. A keyed change table can be rebuilt from its topic at any time — and an append table cannot be rebuilt from anything. That is the operational difference between chapter 11’s events and this chapter’s orders, and it decides which of the two gets the retention policy that keeps the topic.

What the deletes cost, and the rewrite

After the snapshot, the burst, the churn and the second round of updates, the landing table’s shape and its read cost.

  before: 5 data files, deletes = {1: (1, 5), 2: (5, 61,378)};  count(*) = 60,075, GETs 29
  rewrite_data_files (delete-file-threshold 1): 5 -> 1
  after:  1 data file, no delete files;                          count(*) = 60,075, GETs 2
  Trino count(*) = 60,075

Sixty-one thousand equality-delete rows for a table of sixty thousand — and a count that opens twenty-nine objects to answer. The delete files arrived one per checkpoint that carried a change: two after the snapshot, three after the burst, four after the churn, five after the last round, with the delete rows climbing from 60,025 to 61,378. A checkpoint with no changes adds nothing, so on a quiet source the table stops growing and on a busy one it grows by one delete file per interval. The rewrite from chapter 8, with a delete threshold of one, folds it to one file and no deletes, and both engines that apply equality deletes agree before and after. Chapter 11’s schedule for upsert streams holds here unchanged: the data-file rewrite, by checkpoint count.

Out, attempt one: the landing table cannot serve its own changes

A consumer wants the delta between two snapshots of orders_cdc: the snapshot before the burst and the one after. Iceberg has two mechanisms for that, and both were tried.

  create_changelog_view(identifier_columns => order_id, compute_updates):  OK -> burst_changes
  SELECT _change_type, count(*) FROM burst_changes:   FAILED: Delete files are currently not supported in changelog scans
  incremental append read (start-snapshot-id, end-snapshot-id):  OK -> 0 rows

The changelog view is created without complaint and fails the first time it is read. A changelog scan in this release cannot apply delete files, and every snapshot of an upsert table carries them. The incremental read is worse — it succeeds and returns nothing. An incremental append scan yields only append snapshots. An upsert commit is an overwrite, so the scan skips every one of them and reports zero rows with a green exit. Without a reconciliation check, a consumer could keep reporting “no changes” while the source changes underneath it.

The table that lands change data cannot serve change data. Not in this release, not with equality deletes in it — and the mechanism that looks like it works is the one to be most afraid of.

Out, attempt two: the curated table and its row lineage

The answer is the pattern Book 1’s twentieth chapter — on deletion vectors and row lineage — made possible. A second table, format version 3, maintained from the landing table by a copy-on-write MERGE INTO. Its rows carry an identity and the sequence number of the commit that last changed them.

MERGE INTO ice.stage15.orders_curated t
USING (SELECT order_id, customer_id, DATE_ADD(DATE '1970-01-01', order_date) AS order_date, status
       FROM ice.stage15.orders_cdc) s
ON t.order_id = s.order_id
WHEN MATCHED AND (t.status <> s.status OR t.customer_id <> s.customer_id) THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *
WHEN NOT MATCHED BY SOURCE THEN DELETE

  first merge:   60,075 rows, 60,075 distinct _row_id, watermark max(_last_updated_sequence_number) = 1
  source changed: 200 shipped -> delivered, order 8031 -> closed;  landing table CONVERGED in 4s
  second merge:  watermark 1 -> 2
  consumer delta (_last_updated_sequence_number > 1):  201 rows  {delivered: 200, closed: 1}
  rows at or below the first watermark:  59,874
  _row_id: 60,075 distinct, range 0..60074, unchanged after a merge that rewrote the files

The consumer’s query is one line: rows whose last-updated sequence number is above the watermark it recorded last time. It returned exactly the 201 rows the source changed — and not the 59,874 that the copy-on-write merge physically rewrote without changing. The sequence number is carried through a rewrite and only advanced by a change. What counts as a change is the merge’s WHEN MATCHED AND clause, which compared status and customer and nothing else. A source update to a column that clause ignores is correctly not a change to any consumer, and that clause is therefore the contract every consumer downstream is reading against. The row ids did not move either; the range after a merge that rewrote every file is the range the first merge assigned. That is the property that makes a v3 table a change source: identity and change-sequence are per row and survive the file being rewritten around them.

Two things bit on the way, and both are in the record. A filter on _last_updated_sequence_number works. An aggregate on it, max(), is pushed down to Iceberg and fails with “Cannot find field _last_updated_sequence_number in struct”. The lineage columns are not in the schema the pushdown binds against. The first merge’s query only worked because a count(DISTINCT …) beside it happened to disable pushdown. The fix is spark.sql.iceberg.aggregate-push-down.enabled=false for the session that computes watermarks, and the finding is that a metadata column is a column for filtering and not yet one for aggregating.

The two-table shape

The two tables now have distinct jobs. The landing table absorbs source changes and supports replay; the curated table gives consumers row identity and a sequence-number watermark. Their maintenance and retention policies follow from those jobs.

	Landing table	Curated table
written by	Flink upsert sink, every checkpoint	MERGE INTO, scheduled
format	v2, equality and position deletes	v3, copy-on-write, row lineage
serves	the convergence test; rebuilds from the topic	every consumer, by sequence-number watermark
maintenance	data-file rewrite by checkpoint count	compaction by size, like any batch table
retention	hot class	audit class, tags at the merges that matter

The landing table exists to be correct and rebuildable. The curated table exists to be read. Trying to make one table do both is what attempt one measured.

What to decide

Ship the convergence test with the pipeline, scheduled, with the count and sums and the per-status distribution, and alert on the time it takes to pass. Four seconds is a checkpoint — forty is a stalled job.

Name every connector, slot, publication and consumer group per deployment. The three defects above are all one defect — something remembered a position after the thing that held it was gone.

Keep the topic long enough to rebuild the landing table, because a keyed table can be rebuilt from it and the replay converges. The topic’s retention is the landing table’s backup.

Do not read changes from the landing table. The changelog scan refuses it and the incremental scan lies. Consumers read the curated table with a sequence-number watermark, and the watermark query runs with aggregate pushdown off.

Rewrite the landing table by checkpoint count, as chapter 11 schedules, and merge into the curated table on the consumers’ freshness requirement, not the writer’s.

What was not run

Ordering across Kafka partitions, which this single-key-per-row table does not depend on. A Debezium restart from an earlier log position, which is the source-side replay; only the topic-side replay was run. A schema change at the source, which chapter 11 ran on the sink side only. Debezium’s tombstone events, which were turned off. And a v3 landing table: the Flink runtime here writes v2, and whether it can write deletion vectors instead of equality deletes is Appendix A’s question.

Exercises

1. Break the convergence test on purpose. Stop the Flink job, make a hundred updates at the source, and run the test. Then resume from a savepoint and run it again, timing the second pass.

Show answer

The first pass fails on the status distribution and on the key sums if any of the updated rows changed customer_id; the count still passes, which is why the count alone is not a test. The second pass converges one checkpoint after the resumed job catches up, and the time it reports is the outage plus one interval, which is the number the freshness alert should carry.

2. Watermark two consumers. Record max(_last_updated_sequence_number) from the curated table twice, an hour apart, and count the rows between the two watermarks. Compare with the number of source rows changed in that hour from the convergence test’s status deltas.

Show answer

They match exactly when every changed source row changed a column the merge compares. They differ by the rows whose change was to a column the WHEN MATCHED AND clause ignores, which the merge correctly did not rewrite and the consumer correctly does not see. The clause is the definition of “changed” for every consumer downstream, and it belongs in a code review.

Final thoughts

Change data arrived in seven seconds and converged every time it was asked to: through a burst, through seven changes to one key, and through a full replay. The test that proved it is nine lines long. The table it landed in was half deletes from the first commit and could be rebuilt from its topic at will. It could not tell a consumer what had changed — once loudly and once in silence. The table beside it could, to the row, because a v3 table carries identity and a change-sequence through every rewrite.

The convergence time is also a useful signal for operations: it tells the platform how long source changes take to become readable. Part III starts by putting that signal beside table-health metrics, then works through incidents, authorisation and recovery. The pipeline is built; keeping it dependable requires knowing when it has stopped doing what these checks established.

Next: The Rule Was Right, the List Was Wrong