Core: Read DELETE manifests in ReachableFileCleanup

[dkranchii]

Summary

ReachableFileCleanup#findFilesToDelete uses ManifestFiles.readPaths to enumerate live file paths for every manifest scheduled for deletion. readPaths is only defined for DATA manifests — it calls ManifestFiles.read which asserts manifest.content() == ManifestContent.DATA and throws IllegalArgumentException otherwise. Because the surrounding Tasks.foreach(...).retry(3).suppressFailureWhenFinished() catches only IOException explicitly and suppresses everything else, the exception is logged per retry and swallowed, so DELETE manifests reachable only through expired snapshots are silently skipped. The delete files and DV Puffin blobs they reference are left as orphans on object storage.
ReachableFileCleanup is selected by RemoveSnapshots.cleanExpirenapshots whenever specific snapshot IDs are given, when a non-main snapshot was removed, or when there are non-main snapshots — i.e., any workflow that touches branches or tags.
IncrementalFileCleanup.findFilesToDelete already handles this correctly by using ManifestFiles.open, which returns a ManifestReader for either DATA or DELETE manifest content. This PR aligns ReachableFileCleanup with that pattern and adds a regression test.

Change

• Replace both ManifestFiles.readPaths(...) calls in ReachableFileCleanup#findFilesToDelete with ManifestFiles.open(...).select("file_path").liveEntries(), projecting only the file-path column so the read cost is unchanged.
• New test TestRemoveSnapshots#testReachableCleanupWithDeleteManifestFromRemovedBranch stages a position-delete file on a branch, removes the branch, expires older snapshots, and asserts the delete-file location is enumerated for deletion. Gated with assumeThat(formatVersion == 2) and assumeThat(!incrementalCleanup) to target t reachable cleanup path.

Test plan

• ./gradlew :iceberg-core:test --tests "org.apache.iceberg.TestRemoveSnapshots" passes.
• The new regression test fails without the code change and passes with it.
• ./gradlew spotlessCheck passes.

---

AI Disclosure

• Platform/Tool: Cursor

[dramaticlly]

thanks @dkranchii for the contribution, but I think the new test testReachableCleanupWithDeleteManifestFromRemovedBranch still pass without the change in ManifestFiles. Upon further digging, I think the root cause is actually elsewhere and I opened #17763 for the fix.

To be clear, the delete manifests seem to be GCed correctly during snapshot expiration as verified in the existing test

iceberg/core/src/test/java/org/apache/iceberg/TestRemoveSnapshots.java

Lines 1040 to 1094 in 7f879b1

	public void testExpireWithDeleteFiles() {
	assumeThat(formatVersion).as("Delete files only supported in V2 spec").isEqualTo(2);
	// Data Manifest => File_A
	table.newAppend().appendFile(FILE_A).commit();
	Snapshot firstSnapshot = table.currentSnapshot();
	// Data Manifest => FILE_A
	// Delete Manifest => FILE_A_DELETES
	table.newRowDelta().addDeletes(FILE_A_DELETES).commit();
	Snapshot secondSnapshot = table.currentSnapshot();
	assertThat(secondSnapshot.dataManifests(table.io())).hasSize(1);
	assertThat(secondSnapshot.deleteManifests(table.io())).hasSize(1);
	// FILE_A and FILE_A_DELETES move into "DELETED" state
	table
	.newRewrite()
	.rewriteFiles(
	ImmutableSet.of(FILE_A), ImmutableSet.of(FILE_A_DELETES), // deleted
	ImmutableSet.of(FILE_B), ImmutableSet.of(FILE_B_DELETES)) // added
	.validateFromSnapshot(secondSnapshot.snapshotId())
	.commit();
	Snapshot thirdSnapshot = table.currentSnapshot();
	Set<ManifestFile> manifestOfDeletedFiles =
	thirdSnapshot.allManifests(table.io()).stream()
	.filter(ManifestFile::hasDeletedFiles)
	.collect(Collectors.toSet());
	assertThat(manifestOfDeletedFiles).hasSize(2);
	// Need one more commit before manifests of files of DELETED state get cleared from current
	// snapshot.
	table.newAppend().appendFile(FILE_C).commit();
	Snapshot fourthSnapshot = table.currentSnapshot();
	long fourthSnapshotTs = waitUntilAfter(fourthSnapshot.timestampMillis());
	Set<String> deletedFiles = Sets.newHashSet();
	removeSnapshots(table).expireOlderThan(fourthSnapshotTs).deleteWith(deletedFiles::add).commit();
	assertThat(deletedFiles)
	.as("Should remove old delete files and delete file manifests")
	.isEqualTo(
	ImmutableSet.builder()
	.add(FILE_A.location())
	.add(FILE_A_DELETES.location())
	.add(firstSnapshot.manifestListLocation())
	.add(secondSnapshot.manifestListLocation())
	.add(thirdSnapshot.manifestListLocation())
	.addAll(manifestPaths(secondSnapshot, table.io()))
	.addAll(
	manifestOfDeletedFiles.stream()
	.map(ManifestFile::path)
	.collect(Collectors.toList()))
	.build());
	}

. But the implementation might need to update.

Also, I believe it would be great to expand the TestRemoveSnapshots coverage for v3 table as well, if you are willing to help.