Spark 4.1: Optimize ExpireSnapshotsSparkAction with manifest-level filtering #15154

joyhaldar · 2026-01-27T14:32:19Z

This PR optimizes ExpireSnapshotsSparkAction by filtering at the manifest level first, then reading content files only from orphaned manifests. Approach is similar to ReachableFileCleanup but uses distributed Spark operations.

Changes:

Added early exits when no snapshots expired or no orphaned manifests
Path level except to find orphaned manifests before reading content files
Compute manifest lists and stats separately to enable early exit without reading content files
Join to get orphaned manifest details, then read only those content files
Added contentFilesFromManifestDF() to read content files from a filtered manifest DataFrame (existing contentFileDS() only accepts snapshot IDs, not a filtered DataFrame)
Added emptyFileInfoDS() helper
Changed ReadManifest to protected in BaseSparkAction

Before

All Expired Files ------+
                        +--> Except --> Orphaned Files
All Live Files ---------+
     (reads all manifests)

After

                    +--> No expired snapshots? --> Return empty (Exit Early)
                    |
Expired Snapshots --+
                    |
                    +--> Find orphaned manifest paths via except
                              |
                              +--> No orphaned manifests? --> Return manifest lists + stats (Exit Early)
                              |
                              +--> Join to get orphaned manifest details
                                        |
                                        +--> Read content files only from orphaned manifests
                                                    |
                                                    +--> Except with live content files --> Orphaned Files

Tests:

testEarlyExitWhenNoOrphanedManifests
testManifestReusedAcrossSnapshots

References:

Early exit when no expired snapshots or no orphaned manifests, similar to ReachableFileCleanup.cleanFiles()

iceberg/core/src/main/java/org/apache/iceberg/ReachableFileCleanup.java

Lines 76 to 82 in 83653ba

    
           if (!deletionCandidates.isEmpty()) { 
        
             Set<ManifestFile> currentManifests = ConcurrentHashMap.newKeySet(); 
        
             Set<ManifestFile> manifestsToDelete = 
        
                 pruneReferencedManifests( 
        
                     snapshotsAfterExpiration, deletionCandidates, currentManifests::add); 
        
             if (!manifestsToDelete.isEmpty()) {

Finding orphaned manifests by removing current references, similar to ReachableFileCleanup.pruneReferencedManifests()

iceberg/core/src/main/java/org/apache/iceberg/ReachableFileCleanup.java

Lines 107 to 140 in 83653ba

    
           private Set<ManifestFile> pruneReferencedManifests( 
        
               Set<Snapshot> snapshots, 
        
               Set<ManifestFile> deletionCandidates, 
        
               Consumer<ManifestFile> currentManifestCallback) { 
        
             Set<ManifestFile> candidateSet = ConcurrentHashMap.newKeySet(); 
        
             candidateSet.addAll(deletionCandidates); 
        
             Tasks.foreach(snapshots) 
        
                 .retry(3) 
        
                 .stopOnFailure() 
        
                 .throwFailureWhenFinished() 
        
                 .executeWith(planExecutorService) 
        
                 .onFailure( 
        
                     (snapshot, exc) -> 
        
                         LOG.warn( 
        
                             "Failed to determine manifests for snapshot {}", snapshot.snapshotId(), exc)) 
        
                 .run( 
        
                     snapshot -> { 
        
                       try (CloseableIterable<ManifestFile> manifestFiles = readManifests(snapshot)) { 
        
                         for (ManifestFile manifestFile : manifestFiles) { 
        
                           candidateSet.remove(manifestFile); 
        
                           if (candidateSet.isEmpty()) { 
        
                             return; 
        
                           } 
        
                           currentManifestCallback.accept(manifestFile.copy()); 
        
                         } 
        
                       } catch (IOException e) { 
        
                         throw new RuntimeIOException( 
        
                             e, "Failed to close manifest list: %s", snapshot.manifestListLocation()); 
        
                       } 
        
                     }); 
        
             return candidateSet; 
        
           }

Reading content files only from orphaned manifests, similar to ReachableFileCleanup.findFilesToDelete()

iceberg/core/src/main/java/org/apache/iceberg/ReachableFileCleanup.java

Lines 169 to 188 in 83653ba

    
           private Set<String> findFilesToDelete( 
        
               Set<ManifestFile> manifestFilesToDelete, Set<ManifestFile> currentManifestFiles) { 
        
             Set<String> filesToDelete = ConcurrentHashMap.newKeySet(); 
        
             Tasks.foreach(manifestFilesToDelete) 
        
                 .retry(3) 
        
                 .suppressFailureWhenFinished() 
        
                 .executeWith(planExecutorService) 
        
                 .onFailure( 
        
                     (item, exc) -> 
        
                         LOG.warn( 
        
                             "Failed to determine live files in manifest {}. Retrying", item.path(), exc)) 
        
                 .run( 
        
                     manifest -> { 
        
                       try (CloseableIterable<String> paths = ManifestFiles.readPaths(manifest, fileIO)) { 
        
                         paths.forEach(filesToDelete::add); 
        
                       } catch (IOException e) { 
        
                         throw new RuntimeIOException(e, "Failed to read manifest file: %s", manifest); 
        
                       } 
        
                     });

Co-authored-by: Joy Haldar <joy.haldar@target.com>

…oin-based filtering

Copilot

Pull request overview

This PR optimizes ExpireSnapshotsSparkAction by replacing driver-side collection with distributed Spark operations for manifest filtering. Instead of reading content files from all manifests in expired snapshots, the implementation now filters at the manifest level first using join-based operations, then reads content files only from orphaned manifests.

Changes:

Added early exit paths when no snapshots are expired or no orphaned manifests exist
Implemented distributed join-based filtering to identify orphaned manifests before reading their content files
Refactored helper methods in BaseSparkAction to support the new distributed approach

Reviewed changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated 2 comments.

File	Description
ExpireSnapshotsSparkAction.java	Replaced driver-side collection logic with distributed Spark operations for manifest-level filtering and added `contentFilesFromManifestDF()` method
BaseSparkAction.java	Added `emptyFileInfoDS()` helper method and changed `ReadManifest` visibility to `protected`
TestExpireSnapshotsAction.java	Updated expected job count in `testUseLocalIterator()` test from 4 to 12

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Copilot · 2026-01-27T16:28:56Z

spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestExpireSnapshotsAction.java

              .as(
                  "Expected total number of jobs with stream-results should match the expected number")
-              .isEqualTo(4L);
+              .isEqualTo(12L);


The expected job count increased from 4 to 12 due to the new distributed operations. Consider adding a comment explaining why this specific count is expected, or add a test case that validates the optimization logic (e.g., verifying early exits when no orphaned manifests exist).

Added a comment explaining the job count.

Copilot · 2026-01-27T16:28:56Z

spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/ExpireSnapshotsSparkAction.java

+      Dataset<FileInfo> liveStats = statisticsFileDS(updatedTable, null);
+      Dataset<FileInfo> orphanedStats = expiredStats.except(liveStats);
+
+      if (orphanedManifestPaths.isEmpty()) {


Using isEmpty() on a Dataset triggers a Spark action that collects data to the driver. Consider using first() wrapped in a try-catch or take(1).length == 0 to avoid potentially expensive operations when checking if a dataset is empty.

Suggested change

if (orphanedManifestPaths.isEmpty()) {

boolean hasOrphanedManifestPaths = orphanedManifestPaths.limit(1).toLocalIterator().hasNext();

if (!hasOrphanedManifestPaths) {

Thanks for the review. Dataset.isEmpty() uses limit(1) and executeTake(1), it only fetches a single row to check emptiness, not the full dataset.

Source: https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/classic/Dataset.scala#L557-L560

rambleraptor · 2026-01-28T18:05:45Z

spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestExpireSnapshotsAction.java

@@ -1200,10 +1200,12 @@ public void testUseLocalIterator() {

          checkExpirationResults(1L, 0L, 0L, 1L, 2L, results);


Is it possible for you to write another test for this functionality?

Thank you for your review Alex.

Sorry about that, I have added two tests for the optimization:

testEarlyExitWhenNoOrphanedManifests

testManifestReusedAcrossSnapshots

Let me know if I have misunderstood your comment and if you were looking for something different.

Thanks a lot! I'll take a look at this first thing tomorrow. The tests help make it easier for myself and others to digest what exactly the code should be doing.

Thank you Alex. I have also added a References section to the PR description linking to the patterns in ReachableFileCleanup that this is based on. Please let me know if it helps with the review.

rambleraptor

I'm not an expert on this area of the codebase, but the rough idea seems reasonable:

Find list of orphaned manifest lists / stats
Get list of files from there

rambleraptor · 2026-01-29T20:56:15Z

spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/BaseSparkAction.java

  }

-  private static class ReadManifest implements FlatMapFunction<ManifestFileBean, FileInfo> {
+  protected static class ReadManifest implements FlatMapFunction<ManifestFileBean, FileInfo> {


Just making this protected seems fine, but I'd love to get another opinion here.

rambleraptor · 2026-01-29T21:06:53Z

spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/ExpireSnapshotsSparkAction.java

-      Dataset<FileInfo> validFileDS = fileDS(updatedMetadata);

-      // fetch files referenced by expired snapshots
+      // find IDs of expired snapshots


Can you add some comments to break up these code sections? I think it helps to understand the flow of the code

amogh-jahagirdar

Thanks @joyhaldar it's still a bit unclear to me why the new changes are significantly improving the execution? If we look at how fileDS works, and how spark would execute the antijoin I think we'd be implicitly covered? Do we have any numbers before/after this change or any particular cases which are egregiously inefficient at the moment?

amogh-jahagirdar · 2026-01-29T21:58:33Z

spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/ExpireSnapshotsSparkAction.java

-      // fetch files referenced by expired snapshots
+      // find IDs of expired snapshots
      Set<Long> deletedSnapshotIds = findExpiredSnapshotIds(originalMetadata, updatedMetadata);
-      Dataset<FileInfo> deleteCandidateFileDS = fileDS(originalMetadata, deletedSnapshotIds);


Hm, a lot of the cases called out in the PR description should be implicitly handled when you look at how fileDS works? e.g. we're creating the set of files from the set of manifests, and if there are no manifests it's already an empty set that we're doing the anti-join against.

Do we have any particular cases that we see improve after this change (if there are numbers that would be helpful)?

joyhaldar and others added 3 commits January 26, 2026 19:07

Spark: Optimize ExpireSnapshotsSparkAction with manifest-level pruning

987b454

Co-authored-by: Joy Haldar <joy.haldar@target.com>

Optimize ExpireSnapshotsSparkAction by replacing collectAsList with j…

6fcf4c3

…oin-based filtering

Optimize ExpireSnapshotsSparkAction by replacing collectAsList with j…

323b7ac

…oin-based filtering

github-actions bot added the spark label Jan 27, 2026

manuzhang changed the title ~~Spark: Optimize ExpireSnapshotsSparkAction with manifest-level filtering~~ Spark 4.1: Optimize ExpireSnapshotsSparkAction with manifest-level filtering Jan 27, 2026

manuzhang requested a review from Copilot January 27, 2026 16:27

Copilot AI reviewed Jan 27, 2026

View reviewed changes

Add comment explaining Spark job count in stream-results test

d92cd03

joyhaldar marked this pull request as ready for review January 28, 2026 03:55

rambleraptor reviewed Jan 28, 2026

View reviewed changes

Add tests for manifest-level filtering optimization

7afb798

rambleraptor reviewed Jan 29, 2026

View reviewed changes

amogh-jahagirdar requested changes Jan 29, 2026

View reviewed changes

	if (!deletionCandidates.isEmpty()) {
	Set<ManifestFile> currentManifests = ConcurrentHashMap.newKeySet();
	Set<ManifestFile> manifestsToDelete =
	pruneReferencedManifests(
	snapshotsAfterExpiration, deletionCandidates, currentManifests::add);

	if (!manifestsToDelete.isEmpty()) {

	private Set<ManifestFile> pruneReferencedManifests(
	Set<Snapshot> snapshots,
	Set<ManifestFile> deletionCandidates,
	Consumer<ManifestFile> currentManifestCallback) {
	Set<ManifestFile> candidateSet = ConcurrentHashMap.newKeySet();
	candidateSet.addAll(deletionCandidates);
	Tasks.foreach(snapshots)
	.retry(3)
	.stopOnFailure()
	.throwFailureWhenFinished()
	.executeWith(planExecutorService)
	.onFailure(
	(snapshot, exc) ->
	LOG.warn(
	"Failed to determine manifests for snapshot {}", snapshot.snapshotId(), exc))
	.run(
	snapshot -> {
	try (CloseableIterable<ManifestFile> manifestFiles = readManifests(snapshot)) {
	for (ManifestFile manifestFile : manifestFiles) {
	candidateSet.remove(manifestFile);
	if (candidateSet.isEmpty()) {
	return;
	}

	currentManifestCallback.accept(manifestFile.copy());
	}
	} catch (IOException e) {
	throw new RuntimeIOException(
	e, "Failed to close manifest list: %s", snapshot.manifestListLocation());
	}
	});

	return candidateSet;
	}

	private Set<String> findFilesToDelete(
	Set<ManifestFile> manifestFilesToDelete, Set<ManifestFile> currentManifestFiles) {
	Set<String> filesToDelete = ConcurrentHashMap.newKeySet();

	Tasks.foreach(manifestFilesToDelete)
	.retry(3)
	.suppressFailureWhenFinished()
	.executeWith(planExecutorService)
	.onFailure(
	(item, exc) ->
	LOG.warn(
	"Failed to determine live files in manifest {}. Retrying", item.path(), exc))
	.run(
	manifest -> {
	try (CloseableIterable<String> paths = ManifestFiles.readPaths(manifest, fileIO)) {
	paths.forEach(filesToDelete::add);
	} catch (IOException e) {
	throw new RuntimeIOException(e, "Failed to read manifest file: %s", manifest);
	}
	});

	if (orphanedManifestPaths.isEmpty()) {
	boolean hasOrphanedManifestPaths = orphanedManifestPaths.limit(1).toLocalIterator().hasNext();
	if (!hasOrphanedManifestPaths) {

		@@ -1200,10 +1200,12 @@ public void testUseLocalIterator() {

		checkExpirationResults(1L, 0L, 0L, 1L, 2L, results);

Spark 4.1: Optimize ExpireSnapshotsSparkAction with manifest-level filtering #15154

Are you sure you want to change the base?

Spark 4.1: Optimize ExpireSnapshotsSparkAction with manifest-level filtering #15154

Conversation

joyhaldar commented Jan 27, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Copilot AI Jan 27, 2026

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Copilot AI Jan 27, 2026

Choose a reason for hiding this comment

Uh oh!

joyhaldar Jan 28, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rambleraptor left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

amogh-jahagirdar left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

joyhaldar commented Jan 27, 2026 •

edited

Loading

joyhaldar Jan 28, 2026 •

edited

Loading