Spark 3.5: Fix RewriteDataFiles with partial progress enabled and max-failed-commits larger than total-file-group (#12120)

Author: manuzhang

Diff for: spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteDataFilesSparkAction.java

@@ -377,7 +377,8 @@ private Builder doExecuteWithPartialProgress(
     // stop commit service
     commitService.close();
 
-    int failedCommits = maxCommits - commitService.succeededCommits();
+    int totalCommits = Math.min(ctx.totalGroupCount(), maxCommits);
+    int failedCommits = totalCommits - commitService.succeededCommits();
     if (failedCommits > 0 && failedCommits <= maxFailedCommits) {
       LOG.warn(
           "{} is true but {} rewrite commits failed. Check the logs to determine why the individual "

Diff for: spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteDataFilesAction.java

@@ -1320,6 +1320,34 @@ public void testParallelPartialProgressWithMaxFailedCommits() {
     shouldHaveACleanCache(table);
   }
 
+  @TestTemplate
+  public void testParallelPartialProgressWithMaxFailedCommitsLargerThanTotalFileGroup() {
+    Table table = createTable(20);
+    int fileSize = averageFileSize(table);
+
+    List<Object[]> originalData = currentData();
+
+    RewriteDataFilesSparkAction rewrite =
+        basicRewrite(table)
+            .option(
+                RewriteDataFiles.MAX_FILE_GROUP_SIZE_BYTES, Integer.toString(fileSize * 2 + 1000))
+            .option(RewriteDataFiles.MAX_CONCURRENT_FILE_GROUP_REWRITES, "3")
+            .option(RewriteDataFiles.PARTIAL_PROGRESS_ENABLED, "true")
+            // Since we can have at most one commit per file group and there are only 10 file
+            // groups, actual number of commits is 10
+            .option(RewriteDataFiles.PARTIAL_PROGRESS_MAX_COMMITS, "20")
+            .option(RewriteDataFiles.PARTIAL_PROGRESS_MAX_FAILED_COMMITS, "0");
+    rewrite.execute();
+
+    table.refresh();
+
+    List<Object[]> postRewriteData = currentData();
+    assertEquals("We shouldn't have changed the data", originalData, postRewriteData);
+    shouldHaveSnapshots(table, 11);
+    shouldHaveNoOrphans(table);
+    shouldHaveACleanCache(table);
+  }
+
   @TestTemplate
   public void testInvalidOptions() {
     Table table = createTable(20);