[core] Fix incomplete large file filtering in append only compaction

paimon-core/src/main/java/org/apache/paimon/append/BucketedAppendCompactManager.java

@@ -39,6 +39,7 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Comparator;
+import java.util.Iterator;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Optional;
@@ -268,14 +269,15 @@ public FullCompactTask(
         @Override
         protected CompactResult doCompact() throws Exception {
             // remove large files
-            while (!forceRewriteAllFiles && !toCompact.isEmpty()) {
-                DataFileMeta file = toCompact.peekFirst();
-                // the data file with deletion file always need to be compacted.
-                if (file.fileSize() >= compactionFileSize && !hasDeletionFile(file)) {
-                    toCompact.poll();
-                    continue;
+            if (!forceRewriteAllFiles) {
+                Iterator<DataFileMeta> it = toCompact.iterator();
+                while (it.hasNext()) {
+                    DataFileMeta file = it.next();
+                    // the data file with deletion file always need to be compacted.
+                    if (file.fileSize() >= compactionFileSize && !hasDeletionFile(file)) {
+                        it.remove();
+                    }
                 }
-                break;
             }
 
             // do compaction

paimon-core/src/test/java/org/apache/paimon/append/BucketedAppendCompactManagerTest.java

@@ -18,11 +18,13 @@
 
 package org.apache.paimon.append;
 
+import org.apache.paimon.compact.CompactResult;
 import org.apache.paimon.io.DataFileMeta;
 
 import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
@@ -189,6 +191,33 @@ public void testPick() {
                 Collections.singletonList(newFile(2621L, 2630L)));
     }
 
+    @Test
+    public void testFullCompactTaskFiltersInterleavedLargeFiles() throws Exception {
+        long targetFileSize = 1024L;
+        DataFileMeta large1 = newFile(1L, 2048L);
+        DataFileMeta small1 = newFile(2049L, 2100L);
+        DataFileMeta large2 = newFile(2101L, 4148L);
+        DataFileMeta small2 = newFile(4149L, 4200L);
+        DataFileMeta large3 = newFile(4201L, 6248L);
+        DataFileMeta small3 = newFile(6249L, 6300L);
+        List<DataFileMeta> inputs = Arrays.asList(large1, small1, large2, small2, large3, small3);
+
+        List<DataFileMeta> rewriterSaw = new ArrayList<>();
+        BucketedAppendCompactManager.CompactRewriter rewriter =
+                files -> {
+                    rewriterSaw.addAll(files);
+                    return Collections.emptyList();
+                };
+
+        BucketedAppendCompactManager.FullCompactTask task =
+                new BucketedAppendCompactManager.FullCompactTask(
+                        null, inputs, targetFileSize, false, rewriter, null);
+        CompactResult result = task.call();
+
+        assertThat(rewriterSaw).containsExactly(small1, small2, small3);
+        assertThat(result.before()).containsExactly(small1, small2, small3);
+    }
+
     private void innerTest(
             List<DataFileMeta> toCompactBeforePick,
             boolean expectedPresent,