From 76c37181b74315327c8ee57066dc373deac6aa0f Mon Sep 17 00:00:00 2001
From: Arnav Balyan <arnavbalyan1@gmail.com>
Date: Thu, 4 Jun 2026 15:41:42 +0530
Subject: [PATCH] update

---
 .../append/BucketedAppendCompactManager.java  | 16 +++++-----
 .../BucketedAppendCompactManagerTest.java     | 29 +++++++++++++++++++
 2 files changed, 38 insertions(+), 7 deletions(-)

diff --git a/paimon-core/src/main/java/org/apache/paimon/append/BucketedAppendCompactManager.java b/paimon-core/src/main/java/org/apache/paimon/append/BucketedAppendCompactManager.java
index 93f09c42f36e..82da4e30bbcf 100644
--- a/paimon-core/src/main/java/org/apache/paimon/append/BucketedAppendCompactManager.java
+++ b/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
diff --git a/paimon-core/src/test/java/org/apache/paimon/append/BucketedAppendCompactManagerTest.java b/paimon-core/src/test/java/org/apache/paimon/append/BucketedAppendCompactManagerTest.java
index 73ee94b4c854..40b1cd13b306 100644
--- a/paimon-core/src/test/java/org/apache/paimon/append/BucketedAppendCompactManagerTest.java
+++ b/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,