Repository navigation
[LAKE-6732] Spark 3.5: scan-based dangling delete removal (backport of apache/iceberg#15727) - #4
Conversation
Backports the scan-based `RemoveDanglingDeletesSparkAction` from apache#15727 onto 1.8.1, plus one local fix not covered by that PR. Problem: the shipped action decides removability from sequence numbers only (`delete.seq <= min(data seq in partition)`), so it cannot see that a delete is provably irrelevant on value bounds. On CDC tables with a monotonically increasing equality key, the partition floor is pinned by full-load data files that compaction never selects (correctly sized, zero attached deletes after bounds pruning, and equality deletes cannot trip tooHighDeleteRatio), so every delete sits above the floor forever. Measured on prod__lake__us.prod_applications_1.decideroutput (2026-08-11): 18,747 equality delete files / 71.4M delete records, of which 18,741 (99.97%) apply to no live data file. 509 data files frozen at seq 51515 pin the floor. 18 consecutive compactions each removed ~1,140 data files while total-delete-files climbed 14,087 -> 18,409 and removed-delete-files was NULL every run. Changes: * RemoveDanglingDeletesSparkAction - replaced with the apache#15727 head file (SHA 000380b), which derives the live delete set from an unfiltered scan (`FileScanTask.deletes()`, honouring sequence number AND value bounds) and removes every delete file in the manifests absent from that set. Three local adaptations, each marked `// AFFIRM:`: - Stream.toList() -> new ArrayList<>(): upstream targets Java 17, 1.8.1 compiles with options.release = 11 (build.gradle:191). - null-guard scan.snapshot(), which NPEs on a table with no current snapshot when the action is invoked directly. - resolve the snapshot once and reuse it for the manifest read, pinning the scan and the manifest list to the same snapshot. * RewriteDataFilesSparkAction - run dangling-delete removal on the "nothing found to rewrite" early return as well. delete-file-threshold normally guarantees a non-empty group set, except in the exact stuck state this feature exists to escape: once the only remaining delete files are already dangling, no data file qualifies for rewrite, totalGroupCount() == 0, and the removal step was skipped forever. Not covered by apache#15727, which does not touch this file; the hole is still present upstream on main. * TestRemoveDanglingDeleteAction - ported from apache#15727. Note testUnpartitionedTable changes from no-op to removing 2 files, because the upstream short-circuit for unpartitioned tables is removed. Deliberately omitted from apache#15727: the DeleteFileIndex sequence-number validation relaxation (lines 205-211). It is separable (different module, no symbol coupling) and unreachable on format-v2 tables, since DeleteFileIndex.build() only populates dvByPath for PUFFIN DVs. 1.8.1 carries the identical block, so the relaxation remains available if we move to v3. Known behaviour change: dangling-delete detection now runs on the driver (scan planning + a read of every delete manifest, thread-pooled but local) rather than over distributed Spark metadata tables. Cost scales with attached-delete density - measured ~712x path redundancy on a random-key table vs 1x on the monotonic-key table that needs the fix. Accepted; size the driver and validate on the largest table first. Not compiled locally - per the build runbook, Java compilation is expected to fail on Macbooks. Needs a Thor build before this leaves draft. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Two follow-ups from self-review of the scan-based dangling-delete backport: - RemoveDanglingDeletesSparkAction: pin the scan itself to the resolved snapshot ID via useSnapshot() instead of relying on scan.snapshot() being called before scan.planFiles(). TableScan#snapshot() re-resolves table().currentSnapshot() on every call, so the safety of the pin was previously an artifact of statement order rather than structural. - TestRewriteDataFilesAction: add a test for the totalGroupCount() == 0 early-return path in RewriteDataFilesSparkAction#execute() -- the exact scenario (no data file ever qualifies for rewrite) that produced NULL removed-delete-files on every prod rewrite_data_files run, and the one piece of this backport not covered by apache#15727 at all.
Three defects found by compiling and running the suites on x86_64/JDK 11. None are in the production fix itself, but one means the backport as previously scoped was incomplete rather than merely minimal. - TestRemoveDanglingDeleteAction: TestHelpers.V2_AND_ABOVE does not exist in 1.8.1 -- it is a post-1.8.1 addition, and a hard compile error here. Use Arrays.asList(2, 3), which is what the sibling TestRewriteDataFilesAction already does in this tree and is semantically identical. - DeleteFileIndex: port apache#15727's DV validation relaxation after all. It was previously omitted as "unreachable on format-v2", which is true of prod but is the wrong conclusion. Under the old code a DV whose data sequence number is below its data file's throws ValidationException out of findDV(), which aborts the caller's scan. The new detection *is* an unfiltered planFiles() scan, so on a v3 table holding exactly the stale DVs this cleanup exists to remove, dangling-delete removal would crash instead of cleaning. Returning null -- the DV simply does not apply -- lets the scan complete and the DV fall out as dangling. Both v3 failures in TestRemoveDanglingDeleteAction were this exception. - TestRewriteDataFilesAction: testRemoveDanglingDeletesWhenNoRewriteGroupSelected asserted against TestHelpers.deleteFiles(), which unions FileScanTask#deletes() over a scan -- the LIVE set, the one set a dangling delete is absent from by definition. The precondition could never pass, and the post-condition would have passed vacuously once it did. Count live delete-file entries from #entries instead, mirroring how TestRemoveDanglingDeleteAction already separates allEntries() from liveEntries(). Also applies spotlessApply to RewriteDataFilesSparkAction and TestRewriteDataFilesAction, which carried format violations from the two preceding commits. Verified on Thor (x86_64, JDK 11.0.31): :iceberg-spark-runtime-3.5_2.12:build -x test -x integrationTest -> iceberg-spark-runtime-3.5_2.12-1.8.1-PATCH.2.jar (44M) :iceberg-core:test --tests "*DeleteFileIndex*" -> BUILD SUCCESSFUL :iceberg-spark-3.5_2.12:test (dangling + rewrite) -> BUILD SUCCESSFUL, 112 tests, 0 failures (was 4 failures before this commit) :iceberg-spark-3.5_2.12:spotlessCheck -> BUILD SUCCESSFUL Note: :iceberg-core:spotlessCheck still fails on three pre-existing format violations in the REST retry files inherited from 1.8.1-affirm-patch (ExponentialHttpRequestRetryStrategy, TestExponentialHttpRequestRetryStrategy, TestRESTCatalog). Left untouched as out of scope for this PR.
| } | ||
|
|
||
| @Override | ||
| public Result execute() { |
There was a problem hiding this comment.
EXPLANATORY COMMENT
What changed: the old code returned early — doing nothing at all — if the table was unpartitioned. That shortcut is gone, so the scan-based check now runs on every table.
Why the shortcut existed. The comment being removed claimed Iceberg's ManifestFilterManager "already performs this table-wide delete on each commit" — i.e. unpartitioned tables clean themselves up, so there is nothing to do here.
Why that is not true. ManifestFilterManager only removes what a commit explicitly hands it. When an operation commits, it passes along a list — "I am replacing these data files", or "drop everything matching this expression" — and the manager drops the matching manifest entries. It never goes looking on its own for delete files that no longer apply to anything.
So the shortcut is correct for exactly one case: a delete file that becomes useless because a later commit replaced the data file it pointed at. That commit names the data file, so cleanup follows.
It does nothing for a delete file that was useless the moment it was written, because no commit ever names it.
Concrete example — this is testUnpartitionedTable, and it is why that test flips from asserting "removes nothing" to "removes 2 files":
commit 1: add delete files D1, D2 <- no data file exists for them to delete yet
commit 2: add data file F
Nothing ever says "remove F", so nothing ever triggers cleanup of D1 and D2. They stay forever.
That is the production bug in miniature. Debezium's upsert writer calls deleteKey(row) before write(row) for every row, including brand-new primary keys that have never existed. Those deletes are dead on arrival — exactly like D1 and D2 above.
The scan-based check does not care how a delete file became useless. It asks the planner "would any reader actually use this?", which covers both cases, so the special case is no longer needed.
| * <li>Collect all delete file entries skipping files from the previous step. | ||
| * </ol> | ||
| */ | ||
| private List<DeleteFile> findDanglingDeletes() { |
There was a problem hiding this comment.
EXPLANATORY COMMENT
Read this one first — it's the shape of the whole method.
Two sets get built, and the answer is the difference between them.
LIVE = every delete file the planner would actually hand a reader
ALL = every delete file that exists in the delete manifests
DANGLING = ALL − LIVE
LIVE comes from planning a real scan and collecting what the planner attached. That matters because the planner already applies both rules that decide whether a delete applies: the sequence-number check and the value-bounds check.
ALL comes from reading the delete manifests directly — it is just "what exists", with no judgement applied.
Anything in ALL but not in LIVE is a delete file that exists but that no reader will ever be given. That is the definition of dangling, and removing it provably cannot change query results.
Why this replaces the old approach. The old code effectively computed ALL and then filtered it by sequence number alone. That misses any delete that is irrelevant because its values do not overlap any data file — which is exactly our case, since a monotonically increasing key (decideroutput's id) puts every new delete's value range above every old data file's range.
| Snapshot snapshot = scan.snapshot(); | ||
| if (snapshot == null) { | ||
| return Collections.emptyList(); | ||
| } |
There was a problem hiding this comment.
EXPLANATORY COMMENT
Two things happen on this line: a null-guard, and pinning the scan.
The null-guard. scan.snapshot() returns null on a table with no current snapshot. Upstream does not check, so calling this action directly (rather than through rewrite_data_files, which happens to null-check earlier) would NPE.
The pin is the important part. TableScan#snapshot() re-resolves table().currentSnapshot() every time it is called — it is not a cached value. So merely storing it in a local and calling planFiles() afterwards is safe only because of the order the statements happen to appear in. Add a second scan.snapshot() call, or move the manifest read above the scan, and the safety silently disappears with no test failing.
scan = scan.useSnapshot(snapshot.snapshotId()) binds the scan to one specific snapshot ID, so both reads see the same table state by construction rather than by luck.
The race this closes. Our Flink writer commits roughly every 60 seconds, so a commit landing mid-run is close to guaranteed:
t0 plan scan -> sees snapshot S1
t1 Flink commits S2, adding a brand-new, perfectly valid delete file D
t2 read delete manifests -> sees snapshot S2, which includes D
D is in ALL (read at S2) but not in LIVE (planned at S1). Since DANGLING = ALL − LIVE, D gets classified as dangling and deleted for real — even though it is live and currently suppressing rows.
That failure is worse than the bug this PR fixes. Dead delete files only cost planning time; removing a live one means suppressed rows silently come back.
| } | ||
| } catch (IOException e) { | ||
| throw new RuntimeIOException(e, "Failed to scan: %s", scan); | ||
| } |
There was a problem hiding this comment.
EXPLANATORY COMMENT
This loop is the "LIVE set" — for every data file task in a full table scan, harvest the delete files that Iceberg's planner actually attached to it. This is the exact same planning code a SELECT * uses, so it's already checking value bounds for equality deletes, not just sequence number.
| } catch (IOException e) { | ||
| throw new RuntimeIOException(e, "Failed to read manifest: %s", manifest); | ||
| } | ||
| } |
There was a problem hiding this comment.
EXPLANATORY COMMENT
And this is "ALL minus LIVE" — walk every delete manifest in the snapshot, and anything not in the LIVE set computed above gets added to danglingDeletes. ManifestReader already filters to live (non-deleted) entries, so we don't need the explicit status < 2 filter the old Spark-dataset version used.
| DeleteFile fileADeletes = fileADeletes(); | ||
| DeleteFile fileA2Deletes = fileA2Deletes(); | ||
| DeleteFile fileBDeletes = fileBDeletes(); | ||
| DeleteFile fileB2Deletes = fileB2Deletes(); |
There was a problem hiding this comment.
EXPLANATORY COMMENT
This test's outcome is unchanged by the diff — it is a regression guard, not new coverage.
What it exercises: deletes committed in the same row-delta as their target data files, so delete and data share a sequence number.
Why that case needs guarding. The equality-delete rule is delete.seq <= min(data seq) — inclusive. That inclusiveness is what makes same-commit deletes work correctly. A rewrite that accidentally made it exclusive would break them silently, and this test is what would catch it.
The only changes here are mechanical: the allEntries()/liveEntries() helper extraction, and dropping an unused fileBDeletes variable that never affected an assertion.
| .map(DeleteFile::location) | ||
| .collect(Collectors.toSet()); | ||
| assertThat(removed) | ||
| .as("Expected 2 delete files removed") |
There was a problem hiding this comment.
EXPLANATORY COMMENT
This test flips its conclusion outright — same setup, opposite expectation.
before: assertThat(result.removedDeleteFiles()).isEmpty() // "No-op for unpartitioned tables"
after: assertThat(removed).hasSize(2)
The setup did not change. What changed is that the old code short-circuited on unpartitioned tables and never looked, on the assumption that they clean themselves up.
Why 2 files should be removed here. The test commits its delete files before appending FILE_UNPARTITIONED, the data file they target. So they never applied to anything from the moment they were written, and no later commit ever names them for removal — nothing triggers a cleanup.
That is the smallest possible reproduction of the production bug: a delete file that is dead on arrival rather than one that dies later. See the comment on the removed fast-path above for why ManifestFilterManager does not catch this.
| .addRows(FILE_A) | ||
| .addRows(FILE_C) | ||
| .addDeletes(fileADeletes) | ||
| // since FILE B doesn't exist, these delete files will be dangling |
There was a problem hiding this comment.
EXPLANATORY COMMENT
Why deletion vectors get their own dedicated test.
A DV is bound to exactly one data file, by construction — the manifest entry carries a required referenced_data_file. A v2 position-delete file is looser: it can be partition-scoped and match anything in its partition.
That makes "is this DV dangling?" the simplest question in the whole suite: does the file it names exist at all? No partition arithmetic, no value bounds.
The setup. fileBDeletes and fileB2Deletes are DVs for FILE_B and FILE_B2 — neither of which is ever appended anywhere in this test. fileADeletes points at FILE_A, which does exist. So two should be removed and one kept.
Why it is separate rather than folded into another test. DVs did not exist when the original algorithm was written — its javadoc only ever described position and equality delete files. This proves the scan-based approach handles the newer type with no DV-specific code: a DV is just another DeleteFile flowing through the same LIVE-vs-ALL difference.
assumeThat(formatVersion).isGreaterThanOrEqualTo(3) skips it on v2, where DVs cannot exist.
| // simply does not apply, so return null instead of throwing. Required by the scan-based | ||
| // dangling-delete detection: it plans a full scan, and throwing here would abort that scan on | ||
| // exactly the stale DVs the cleanup exists to remove. | ||
| if (dv != null && dv.dataSequenceNumber() < seq) { |
There was a problem hiding this comment.
EXPLANATORY COMMENT
What this method does. During scan planning Iceberg asks, for each data file, which delete files apply to it? findDV is the deletion-vector part of that answer. seq is the data file's sequence number; dv.dataSequenceNumber() is the DV's.
What changed: when a DV's sequence number is lower than its data file's, the old code threw. Now it returns null, meaning "this DV does not apply."
Why null is the correct answer. Deletes only apply backwards in time — a delete written at sequence N removes rows from data written before N. So a DV whose sequence sits below the data file it names cannot possibly apply to that file. "Does not apply" is simply true. This is upstream apache#15727's change, not an Affirm one.
Why the exception cannot stay. The new detection works by asking the planner to plan a scan of the whole table, collecting every delete file the planner attached to something, and treating whatever is left over as dangling. So detection is a planFiles() call — and findDV() runs during planning.
That means a single DV anywhere in the table with a stale sequence number throws, the exception escapes planning, planFiles() fails, and the entire dangling-delete detection dies.
Now consider what gives a DV a stale sequence number: compaction rewrites its data file, the replacement gets a higher sequence number, and the DV is left behind pointing at the old one. The exact artifact this cleanup exists to remove is the one that would crash it.
No impact on prod today. dvByPath is only populated for Puffin deletion vectors, which are format-v3. On v2 the dvByPath == null guard a few lines above returns first, so neither the old nor the new code is reached.
| assertThatThrownBy(() -> index.forDataFile(2, FILE_A)) | ||
| .isInstanceOf(ValidationException.class) | ||
| .hasMessageContaining("must be greater than or equal to data file sequence number"); | ||
| assertThat(index.forDataFile(2, FILE_A)).isEmpty(); |
There was a problem hiding this comment.
EXPLANATORY COMMENT
Companion to the DeleteFileIndex change — the same semantic flip, expressed as an assertion.
The setup builds the invalid pairing on purpose: a DV pinned at data sequence 1, then forDataFile(2, FILE_A) asks what applies to FILE_A at sequence 2. So dv.seq (1) < seq (2) — a DV that predates the data file it names. This is the only test that constructs that state deliberately, so it is where the behavioural contract is pinned down: previously "throws ValidationException", now "returns no delete files."
.isEmpty() rather than .isNull() because forDataFile returns DeleteFile[] — the production return null from findDV is folded into an empty array by the caller. Asserting on the array is the right level, since that is what a scan consumes.
The assumeThat(formatVersion).isGreaterThanOrEqualTo(3) skip means this does not run on v2, where dvByPath is null and findDV returns before either code path.
In 1.8.1 this test lives in DeleteFileIndexTestBase rather than TestDeleteFileIndex as upstream has it.
| protected static List<Object> parameters() { | ||
| return Arrays.asList(2, 3); | ||
| protected static List<Integer> parameters() { | ||
| return TestHelpers.V2_AND_ABOVE; |
There was a problem hiding this comment.
EXPLANATORY COMMENT
Why deletion vectors appear in this file at all.
@Parameters makes the whole suite run twice — once against a v2 table and once against v3. Deletion vectors are the v3 replacement for position-delete files: a Puffin-encoded bitmap, one per data file, instead of a separate delete file per operation.
So every test here is implicitly also a v3 test, whether or not it mentions DVs.
Why that is worth keeping. RemoveDanglingDeletesSparkAction is a shared Iceberg action — any table on this Spark version can call it, not only our v2 Debezium tables. Breaking DV handling would be invisible today and a landmine for the first table that moves to v3.
abrarsher23
changed the title
| // for rewrite, so totalGroupCount() == 0 and returning here would leave them in place | ||
| // forever. Not covered by apache/iceberg#15727, which does not touch this file; the hole is | ||
| // still present upstream on main. | ||
| if (removeDanglingDeletes) { |
There was a problem hiding this comment.
I think this fix also deserves its own upstream PR. apache#15727
doesn't touch this file, and the bug is there with the current detection logic
too, so it doesn't need to wait. Happy to review it.
There was a problem hiding this comment.
Looks like kinolaev kinolaev added it in with apache@0d8075c
| ManifestFiles.readDeleteManifest(manifest, table.io(), table.specs())) { | ||
| for (DeleteFile deleteFile : reader) { | ||
| if (!deletes.contains(deleteFile)) { | ||
| danglingDeletes.add(deleteFile); |
There was a problem hiding this comment.
Can we drop the stats here? Every delete file we read keeps all of its column
statistics in driver memory. The set only needs the file
location and its offset and size, and copyWithoutStats() keeps all three.
| danglingDeletes.add(deleteFile); | |
| danglingDeletes.add(deleteFile.copyWithoutStats()); |
Review feedback from @huaxingao on #4. findDanglingDeletes() retains one DeleteFile per dangling delete for the lifetime of the action, and ManifestReader#iterator() hands back a full defensive copy including every column-level metrics map. On a table with a large delete backlog that is a lot of driver heap held for data nothing on this path reads. DeleteFileSet keys on (location, contentOffset, contentSizeInBytes) -- DeleteFileSet.DeleteFileWrapper#equals -- and copyWithoutStats() preserves all three, along with manifestLocation, specId, partition, recordCount, fileSizeInBytes, equalityIds and referencedDataFile. It nulls only columnSizes, valueCounts, nullValueCounts, nanValueCounts, lowerBounds and upperBounds. Nothing on the removal path touches those: ManifestFilterManager uses manifestLocation for its no-deletes fast path, then set membership plus specId/partition, and the snapshot summary needs only recordCount and fileSizeInBytes. Classification is unaffected -- deletes.contains(deleteFile) is evaluated on the unstripped object, before the copy. The reader-projection alternative (select() columns so ManifestReader#iterator drops stats itself, avoiding the copy-then-copy) was rejected: it means hand-enumerating every manifest column the rewrite needs, and record_count lives inside ManifestReader.STATS_COLUMNS, so it is fragile against any schema change. New test testRemovedDeleteFilesDropColumnStats pins all three properties -- classification unchanged, no metrics map survives, and the identity and summary fields that do survive -- with FILE_B_EQ_DELETES as the negative control since it genuinely applies and must be kept. Verified on Thor (x86_64, JDK 11.0.31): :iceberg-spark-3.5_2.12:test --tests "*TestRemoveDanglingDeleteAction" -> BUILD SUCCESSFUL, 10 tests, 0 failures, 1 skipped (DV test on v2) :iceberg-spark-3.5_2.12:spotlessJavaApply -> no reformatting negative control, copyWithoutStats() reverted and the test kept -> only testRemovedDeleteFilesDropColumnStats fails, on both v2 and v3 (valueCounts {2=1L} where null expected); other 7 still pass Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…tes; extract shared DuplicateFileRegistrationGuard Ports the additional scope #8 built while retargeting this guard to the 1.11 base, back onto this branch: - Extracts the detection/repair logic this branch already had inline in RewriteDataFilesSparkAction into a shared DuplicateFileRegistrationGuard, used by three actions instead of one. remove_dangling_deletes reproduces the delete-side "2-of-2 removal" half of the original incident on its own (DeleteFileSet's identity key is (location, contentOffset, contentSizeInBytes), identical across two live registrations of one physical file), and rewrite_position_delete_files is run standalone from oncall/backfill notebooks with no dependency on rewrite_data_files ever having run -- both needed the identical guard, not a second drifting copy of this logic. - New core test suite, TestDuplicateRegistrationRepair (66 tests): argument validation, the keep-lowest-sequence-number invariant, 3+ registrations collapsing to one, delete-file repair, that repairing a data file doesn't strip its own delete coverage, and retry safety against a concurrent commit changing the intended survivor between scan and commit. - New ManifestFilterManager checks: validateNoContradictoryDuplicateRegistrationIntent (rejects a commit that both keeps a survivor and separately deletes that same path outright -- the two instructions silently resolved in the destructive direction before this), and hoists isDuplicateRegistrationToDrop's two call sites out of their `||` chains into local variables, since Java's short-circuiting could skip the survivor-recording side effect when an earlier term in the same chain already matched the entry. - TestRemoveDanglingDeleteAction: one existing test's fixture (testPartitionedDeletesWithDanglingDvs) incidentally re-registers two data files at a second sequence number, which the guard now repairs before the dangling-delete pass runs -- updated its post-repair assertions accordingly, and added a new test driving the "delete side" defect deliberately rather than incidentally. - TestRewriteDataFilesAction: narrowed testDuplicateDataFileRegistrationCheckCanBeDisabled's assertion to just "the opt-out doesn't throw this guard's exception" -- two Thor runs disagreed on what happens downstream of the opt-out (planner bin-packing determines whether the duplicate lands in a selected file group), so asserting no exception at all was asserting a coin flip. - Fixed a pre-existing checkstyle violation in RemoveDanglingDeletesSparkAction (new ArrayList<>() instead of Lists.newArrayList()) while already touching that file -- present since #4 merged, apparently never caught by CI until now. TestRewritePositionDeleteFilesAction's new testIcebergItselfRefusesToCreateADuplicateDvRegistration had to be adapted rather than ported verbatim: it relied on rewrite_position_delete_files converting V2 position deletes to V3 DVs, a capability this branch's RewritePositionDeleteFilesSparkAction doesn't have (it hard-refuses to run against a V3 table at all here). Generates the DV directly via FileGenerationUtil instead, matching TestRemoveDanglingDeleteAction's existing pattern -- same assertion, without depending on a feature this base doesn't have. Also found and fixed, independent of the port: DuplicateFileRegistrationGuard's validateNoContradictoryDuplicateRegistrationIntent on the 1.11 branch has a comment stating it must read callerDeletedFilePaths (not deleteFiles, which the filtering loop also writes to and never clears between commit attempts) but the code still reads deleteFiles -- this port implements what the comment says, not what the code currently does. Worth a look on #8 too. Tested: TestDuplicateRegistrationRepair (66 tests, 0 failures), TestDuplicateFileRegistrationGuard (5 tests, 0 failures), TestRemoveDanglingDeleteAction (12 tests, 2 skipped, 0 failures), TestRewriteDataFilesAction (full suite, 122 tests, 2 skipped, 0 failures), TestRewritePositionDeleteFilesAction (full suite, 19 tests, 0 failures). spotlessCheck and checkstyle clean on every file this commit touches -- iceberg-core:checkstyleTest still fails on this branch, but only on core/src/test/java/org/apache/iceberg/rest/* (unused imports, annotation style), pre-existing since PR #1 merged and unrelated to this change.
4 participants
SUMMARY
Our Flink replication jobs write Debezium CDC data into lake tables. A separate, daily maintenance step —
rewrite_data_files— is supposed to compact those tables, but on ones with a monotonically increasing key it runs, reports success, and removes zero delete files, so they accumulate without bound. On an affected table over 99% of them apply to nothing, yet every one is still checked on every query plan.This PR replaces the dangling-delete cleanup with one that asks Iceberg's own query planner which delete files a reader would actually use, and removes the rest. It is a backport of apache/iceberg#15727, plus one Affirm-only fix upstream lacks:
rewrite_data_filesreturns early when no data file qualifies for rewrite, before reaching the cleanup — exactly the stuck state here, since once the only delete files left are dangling, nothing qualifies. Cleanup now runs on that exit too.This is a two-PR change with one shared CMT. This description is that CMT — it covers both the fix here and the pyspark pin bump that actually activates it, in Affirm/all-the-things#215004. Approvals for both happen against this PR; see RELEASE PLAN below for why the two are split and how they sequence.
Jira: LAKE-6732
ROOT CAUSE & APPROACH
1. How
rewrite_data_filesworks todayThe procedure does two things, in this order: rewrite data files (pick the ones worth merging, merge them, commit the replacements), then remove dangling deletes (drop delete files that no longer apply). The two are connected by sequence numbers, and that is where it breaks down.
A data file's sequence number is set when it is written and only changes if the file is rewritten — step 1 gives every replacement a new, higher number, and a file never picked for rewriting keeps its original forever.
Step 2 only looks at sequence numbers:
Call that lowest number the floor. The rule is safe — a delete older than every surviving data file genuinely cannot apply to any of them — but it is the only thing step 2 checks. It never looks at values, and values are how our deletes die.
Step 1 picks a file on exactly three criteria: wrongly sized, too many delete files attached, or too high a delete ratio. If none fire, the file keeps its sequence number.
2. Why it leaves dangling deletes behind
Take
prod_applications_1.decideroutput, measured read-only againstprod__lake__us:removed-delete-filesThe deletes are dead on values. The equality key is
id, which only increases. A delete written today covers ids around 900,000,000; every older data file holds lower ids. Deletes apply only backwards in time, so the files this delete could apply to are exactly the ones whose id ranges it cannot overlap. Zero matches, by arithmetic.The floor never moves. It sits at 51515, held down by 509 data files (216 GB) from the original full load. It can only rise if those files get rewritten in step 1 — and not one of the three criteria can fire:
The deadlock:
And it compounds. Compaction sizes its output to the target file size, so the output always lands inside the "correctly sized" band — every run manufactures another file that can never be picked again.
The control case.
decisionmodel— same cluster, schedule and writer — sits at 99 delete files. Its key is a randomari, so its deletes do overlap old files, they attach, the files get picked, the floor rises, and cleanup works. Same code, opposite outcome, which is what tells us the diagnosis is right.A second, separate bug.
RewriteDataFilesSparkActionreturns early when no data file qualifies for rewrite, before step 2 runs at all — precisely the state above, since dangling deletes attach to nothing and so make nothing qualify. Not covered by apache#15727, still open upstream.3. How we're fixing it
Stop inferring which deletes are dead, and ask the query planner directly:
LIVEcomes from Iceberg's own scan planning, which applies both the sequence-number rule and the value bounds — the check step 2 was missing. Anything the planner refuses to hand a reader is, by construction, a file no query can use. The early-return hole is closed by running cleanup on both exits ofRewriteDataFilesSparkAction.Why it is safe: we remove only files Iceberg's own planner refuses to hand any reader, so query results provably cannot change — a set-complement argument rather than a bounds check we wrote and must keep in sync. S3 is untouched; old snapshots keep referencing the files until
expire_snapshots, so time travel and tags are unaffected. Cost isO(#files), notO(table size).CHANGES
3 production files, 3 test files, +411 / −181.
RemoveDanglingDeletesSparkAction.javaALL − LIVEset complement. Four adaptations marked// AFFIRM:—new ArrayList<>()instead of Java-16Stream.toList(); a null-guard onscan.snapshot(); the scan pinned viauseSnapshot()so a concurrent commit can't get a live delete file classified as dangling; and retained dangling deletes stripped viacopyWithoutStats(), sinceDeleteFileSetkeys only on(location, contentOffset, contentSizeInBytes)and nothing on the removal path reads a metrics map — so holding every dangling delete's column bounds is pure driver heap.RewriteDataFilesSparkAction.javaremoveDanglingDeletesAndBuild(Builder)shared by both exits. This is the Affirm-only fix.DeleteFileIndex.javaTestRemoveDanglingDeleteAction.java,TestRewriteDataFilesAction.java,DeleteFileIndexTestBase.javaTESTING
Unit tests
TestRemoveDanglingDeleteAction(v2 + v3)TestRewriteDataFilesAction, incl. new zero-rewrite-groups case:iceberg-spark-3.5_2.12:test(both suites):iceberg-core:test --tests "*DeleteFileIndex*":iceberg-spark-3.5_2.12:spotlessCheckAlpine notebook testing against stage & prod tables
Ran the patched jar on an alpine pod directly against real stage and prod tables. The only variable between runs is the jar.
before / after are delete files removed — before is the jar running in prod today, after is this PR. Same table, same command, same config; the jar is the only thing that changes. Only the two prod A/B tables were run both ways.
prod_payments_gateway_1.paymentprod_applications_1.decideroutputprod_users_2.chargeeventstage_applications_1.decideroutputstage_users_1.disclosureData was preserved on every run. Row count and a full-column checksum (
bit_xor(xxhash64(to_json(struct(*)))), order-independent) came back identical at each snapshot and at its own parent — the only interval containing just our operation, since these tables take a CDC commit every ~60s and so a naive before/after count proves nothing. The one exception ischargeevent, verified on row count alone: a full-column checksum over 2.46 B rows / 570 GB exceeded the executor sizing used.Stage Airflow testing
The jar shipped as a real
+affirm.dev1pyspark wheel, pinned inrequirements-3.in, deployed to Stage US via Buildkite, and exercised by the reallake-debezium-compaction-taskDAG — the actual production code path, not a notebook. A/B across the two shards of the same table, so schema, writer and spec are identical and only the jar differs:stage_applications_1— afterstage_applications_2— beforeremoved-delete-filestotal-delete-filesafterFailure-mode testing on stage
Ran against the real
lake-debezium-compaction-taskDAG, deliberately killing things mid-run to check recovery — not just the happy path.kubectl delete --grace-period=0 --force, no graceful shutdown) partway through a rewrite. Airflow's own operator detected the vanished pod and moved the task toup_for_retryon its own; all 6 executors self-terminated cleanly rather than orphaning; the automatic retry ran a fresh rewrite and succeeded (675 files, 0 failures).RISK ASSESSMENT
Checkout impact:
Customer impact if this fails: None. This runs inside the daily lake compaction DAG. The worst realistic failure is that a compaction run errors and delete files keep accumulating — i.e. today's behaviour, which is what the change fixes. The cleanup is metadata-only and never deletes physical files, so a bad run cannot destroy data.
Specific risks and mitigations:
useSnapshot()fixes both the scan and the delete-manifest read to one snapshot, so the two halves cannot disagree. At commit timeBaseRewriteFilescallsfailMissingDeletePaths(), so if a concurrent job already removed one of these files the commit fails loudly instead of mis-applying, andSnapshotProducerretries against the refreshed base. A CDC append cannot resurrect a dangling delete either — new data files get higher sequence numbers, and a delete only applies to data files below its own. Exercised against tables taking a commit every ~60s.table_compactionalready requests, on the fleet's heaviest workload (13,662 data files, 30.5 M pairs). No sizing change needed.RELEASE PLAN
Merging this PR changed nothing in any environment. The rollout is a separate, later step: a jar built from the merged commit, published as a real
2815!3.5.4+affirm.6wheel via the manual process in Producing the wheel below — built fresh againstd4406cd29, not reusing any pre-merge stage artifact. Only once that wheel exists does the rollout happen, in Affirm/all-the-things#215004, which bumpsdeployable/data-replication/requirements-3.into+affirm.6. That PR currently still points at the pre-merge stage-validation branch pinned to+affirm.dev1and is marked DO NOT MERGE; it must be repointed to an+affirm.6pin with a fresh lockfile compile before it can go anywhere.Review changed the code, so the merged commit was re-validated on stage before the release cut. The
+affirm.dev1artifact used in Alpine notebook testing and Stage Airflow testing above was built pre-merge from19f5580c5and bundles jar1.8.1-PATCH.2. Review added one commit —c9227a174,copyWithoutStats()on retained dangling deletes — so the merged treed4406cd29is not byte-identical to what those sections validated. The delta is one line plus a test, covered by 114 unit tests including a mutation check (reverting only the one-line fix makes the new test the sole failure, on both format versions), but unit tests are not real-data evidence.The merged commit was therefore rebuilt as jar
1.8.1-PATCH.3(sha256a10da1e8c028e7615a86fbd8e6477c88f20b9099f3186dcdc64acbffa5f43872, embeddinggit.commit.id=d4406cd29194312becab75b05ac04871a3a9f6af) and shipped to Stage US/CA/UK as+affirm.dev2before any production wheel was cut. The jar version was deliberately bumpedPATCH.2→PATCH.3so snapshot summaries distinguish the two builds — with both stampedPATCH.2, this validation would have been unfalsifiable.Stage re-validation of the merged commit —
stage_applications_1.decideroutput, 2026-08-26 18:18 UTC. Reallake-debezium-compaction-taskDAG, scheduler-generated config copied verbatim from that table's own 17:02 run. Two snapshots, both stamped1.8.1-PATCH.3 (commit d4406cd29…):removed-delete-filesremoved-equality-deletestotal-delete-filestotal-recordstotal-data-filesRow counts are compared against each snapshot's own parent — the only interval containing just our operation, since this table takes a CDC
overwriteroughly every 60 seconds. A natural A/B on the same table the same day: at 17:02 underPATCH.1the run removed nothing and the backlog kept climbing (11,027 → 11,032 over five minutes); at 18:18 underPATCH.3it removed 11,040 and reached zero.What is new post-merge is the version string,
+affirm.dev2→+affirm.6, and publication to the release channel (pypi-local) instead of staying stage-only. Dev version names exist for exactly this — Artifactory versions are immutable, so dry runs before a real release cut get adevN.This description is the CMT for both changes. #215004 shares this CMT rather than carrying a duplicate one — its own description just points back here. Approvals for the rollout happen against this PR. The intended sequence:
data-replicationpin and deploy well before 00:00 UTC. This is a single deploy —data-replicationisn't staged per region, so UK, CA and US all pick up the new wheel at once.rewrite-all=true, which lifts the partition floor and lets the existing cleanup collect delete files, so this jar is a smaller behaviour change there. US is where nothing is removed today, so it's where the fix's effect — and any problem — will show up largest and first.2815!3.5.4+affirm.5and redeploy. The previous wheel is immutable in Artifactory and still present. Nothing to undo in the data — every commit this makes is an Iceberg snapshot, andrewriteFilesnever deletes physical files, so rollback is metadata-only.Feature flag: N/A.
_DEFAULT_PROCEDURE_OPTIONSintable_compaction.py:143already setsremove-dangling-deletes: "true"for every Debezium table, so no config change accompanies the bump — the jar alone is what makes the cleanup actually remove anything.Scope of the pin: per-DT, so the other six DTs on
3.5.4+affirm.5are untouched.Producing the wheel — the manual jar → wheel → DT path
No CI exists for this path — the jar and wheel are built by hand, following our Notion runbooks How to Build Iceberg and Bundle in Spark Locally and Runbook: how to upgrade PySpark at Affirm. Run twice against the merged commit: once as
+affirm.dev2for the stage re-validation above, then as+affirm.6for the release.make requirementsfor Phase 3 dies withrosetta error: Attachment of code signature supplement failedon the x86-translated toolchain.Phase 1 — Iceberg jar (Thor, x86_64, JDK 11.0.31, 16 core / 123 GB):
echo '1.8.1-PATCH.3' > version.txt— the only version mechanism; untracked, do not commit./gradlew :iceberg-spark:iceberg-spark-runtime-3.5_2.12:build -x test -x integrationTest./build/mvn -N install:install-file -DgroupId=org.apache.iceberg -DartifactId=iceberg-spark-runtime-3.5_2.12 -Dversion=1.8.1-PATCH.3 -Dpackaging=jar -Dfile=<jar>git worktree—gradle-git-propertiesuses jgit and throwsRepositoryNotFoundException: .git/worktrees/<name>.Phase 2 — pyspark wheel, SAME host (
~/.m2is the handoff):Affirm/spark, branchbranch-3.5pom.xml:445${iceberg.version}-PATCH.1→-PATCH.3(line 451 leavesiceberg-aws-bundleat stock 1.8.1, matching the two jars in the DT image)affirm-3.5.4offbranch-3.5— the PySpark runbook's prod-release step; dev cuts stay onbranch-3.5python/pyspark/version.py→2815!3.5.4+affirm.6(prod convention is+affirm.X;+affirm.devXfor dry runs)./build/mvn -DskipTests -Pkubernetes clean package -T 4— measured 23:10 min, 22:18 on the rebuildpyenv global 3.9.18 && cd python && python -m build→ 651 MB wheel. SetPYENV_ROOT=/pyenvexplicitly — a non-login shell on the Thor defaults it to$HOME/.pyenv, and$HOMEthere is/workspace, sopyenv globalreportsversion not installedand silently leaves Python 3.7.9 selectedtwine upload -r artifactorytopypi-local— needs SailPoint "Artifactory: PYPI Local Writers"Phase 3 — DT pickup: bump
deployable/data-replication/requirements-3.in:72, re-locksrc/requirements-3.txt(make src/requirements-3.txt), merge, deploy. The lockfile carriesREQUIREMENTS_INPUT_HASH_v1andREQUIREMENTS_COMPILED_HASH_v1trailers, so the pin cannot be hand-edited — the compile has to run. The DT Dockerfile adds no jars; it only setsSPARK_HOME=$DT_PYTHON_SITE_DIR/pyspark.