[core] Support refresh toCompact files in BucketedAppendCompactManager

Author: xiangyuf

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

@@ -45,6 +45,7 @@
 import java.util.PriorityQueue;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
+import java.util.function.Supplier;
 
 import static java.util.Collections.emptyList;
 
@@ -65,6 +66,8 @@ public class BucketedAppendCompactManager extends CompactFutureManager {
 
     private List<DataFileMeta> compacting;
 
+    private final Supplier<List<DataFileMeta>> dataFilesSupplier;
+
     @Nullable private final CompactionMetrics.Reporter metricsReporter;
 
     public BucketedAppendCompactManager(
@@ -76,6 +79,28 @@ public BucketedAppendCompactManager(
             long targetFileSize,
             CompactRewriter rewriter,
             @Nullable CompactionMetrics.Reporter metricsReporter) {
+        this(
+                executor,
+                restored,
+                dvMaintainer,
+                minFileNum,
+                maxFileNum,
+                targetFileSize,
+                rewriter,
+                metricsReporter,
+                null);
+    }
+
+    public BucketedAppendCompactManager(
+            ExecutorService executor,
+            List<DataFileMeta> restored,
+            @Nullable DeletionVectorsMaintainer dvMaintainer,
+            int minFileNum,
+            int maxFileNum,
+            long targetFileSize,
+            CompactRewriter rewriter,
+            @Nullable CompactionMetrics.Reporter metricsReporter,
+            @Nullable Supplier<List<DataFileMeta>> dataFilesSupplier) {
         this.executor = executor;
         this.dvMaintainer = dvMaintainer;
         this.toCompact = new PriorityQueue<>(fileComparator(false));
@@ -85,10 +110,15 @@ public BucketedAppendCompactManager(
         this.targetFileSize = targetFileSize;
         this.rewriter = rewriter;
         this.metricsReporter = metricsReporter;
+        this.dataFilesSupplier = dataFilesSupplier;
     }
 
     @Override
     public void triggerCompaction(boolean fullCompaction) {
+        if (dataFilesSupplier != null) {
+            refreshToCompact();
+        }
+
         if (fullCompaction) {
             triggerFullCompaction();
         } else {
@@ -120,6 +150,23 @@ private void triggerFullCompaction() {
         toCompact.clear();
     }
 
+    @VisibleForTesting
+    protected void refreshToCompact() {
+        List<DataFileMeta> latestDataFiles = new ArrayList<>(dataFilesSupplier.get());
+        List<DataFileMeta> currentDataFiles = new ArrayList<>(toCompact);
+
+        if (LOG.isDebugEnabled()) {
+            LOG.debug(
+                    "Current bucket latest snapshot files count {}, toCompact files count {}.",
+                    latestDataFiles.size(),
+                    currentDataFiles.size());
+        }
+
+        // deduplicate files
+        latestDataFiles.removeIf(currentDataFiles::remove);
+        toCompact.addAll(latestDataFiles);
+    }
+
     private void recordCompactionsQueuedRequest() {
         if (metricsReporter != null) {
             metricsReporter.increaseCompactionsQueuedCount();

paimon-core/src/main/java/org/apache/paimon/operation/AppendOnlyFixedBucketFileStoreWrite.java

@@ -19,6 +19,7 @@
 package org.apache.paimon.operation;
 
 import org.apache.paimon.CoreOptions;
+import org.apache.paimon.Snapshot;
 import org.apache.paimon.append.BucketedAppendCompactManager;
 import org.apache.paimon.compact.CompactManager;
 import org.apache.paimon.compact.NoopCompactManager;
@@ -28,15 +29,18 @@
 import org.apache.paimon.deletionvectors.DeletionVectorsMaintainer;
 import org.apache.paimon.fs.FileIO;
 import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.manifest.ManifestEntry;
 import org.apache.paimon.types.RowType;
 import org.apache.paimon.utils.FileStorePathFactory;
 import org.apache.paimon.utils.SnapshotManager;
 
 import javax.annotation.Nullable;
 
+import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.ExecutorService;
 import java.util.function.Function;
+import java.util.function.Supplier;
 
 /** {@link AppendOnlyFileStoreWrite} for {@link org.apache.paimon.table.BucketMode#HASH_FIXED}. */
 public class AppendOnlyFixedBucketFileStoreWrite extends AppendOnlyFileStoreWrite {
@@ -85,22 +89,49 @@ protected CompactManager getCompactManager(
                     dvMaintainer != null
                             ? f -> dvMaintainer.deletionVectorOf(f).orElse(null)
                             : null;
-            return new BucketedAppendCompactManager(
-                    compactExecutor,
-                    restoredFiles,
-                    dvMaintainer,
-                    options.compactionMinFileNum(),
-                    options.compactionMaxFileNum().orElse(5),
-                    options.targetFileSize(false),
-                    files -> compactRewrite(partition, bucket, dvFactory, files),
-                    compactionMetrics == null
-                            ? null
-                            : compactionMetrics.createReporter(partition, bucket));
+            if (options.compactionForceRefreshFiles()) {
+                Supplier<List<DataFileMeta>> dataFilesSupplier =
+                        () -> scanBucketFiles(snapshotManager.latestSnapshot(), partition, bucket);
+                return new BucketedAppendCompactManager(
+                        compactExecutor,
+                        restoredFiles,
+                        dvMaintainer,
+                        options.compactionMinFileNum(),
+                        options.compactionMaxFileNum().orElse(5),
+                        options.targetFileSize(false),
+                        files -> compactRewrite(partition, bucket, dvFactory, files),
+                        compactionMetrics == null
+                                ? null
+                                : compactionMetrics.createReporter(partition, bucket),
+                        dataFilesSupplier);
+            } else {
+                return new BucketedAppendCompactManager(
+                        compactExecutor,
+                        restoredFiles,
+                        dvMaintainer,
+                        options.compactionMinFileNum(),
+                        options.compactionMaxFileNum().orElse(5),
+                        options.targetFileSize(false),
+                        files -> compactRewrite(partition, bucket, dvFactory, files),
+                        compactionMetrics == null
+                                ? null
+                                : compactionMetrics.createReporter(partition, bucket));
+            }
         }
     }
 
     @Override
     protected Function<WriterContainer<InternalRow>, Boolean> createWriterCleanChecker() {
         return createConflictAwareWriterCleanChecker(commitUser, snapshotManager);
     }
+
+    private List<DataFileMeta> scanBucketFiles(Snapshot snapshot, BinaryRow partition, int bucket) {
+        List<DataFileMeta> dataFileMetas = new ArrayList<>();
+        List<ManifestEntry> files =
+                scan.withSnapshot(snapshot).withPartitionBucket(partition, bucket).plan().files();
+        for (ManifestEntry entry : files) {
+            dataFileMetas.add(entry.file());
+        }
+        return dataFileMetas;
+    }
 }

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

@@ -191,6 +191,39 @@ public void testPick() {
                 Collections.singletonList(newFile(2621L, 2630L)));
     }
 
+    @Test
+    public void testToCompactRefresh() {
+        int minFileNum = 4;
+        int maxFileNum = 12;
+        long targetFileSize = 1024;
+
+        List<DataFileMeta> totalFiles =
+                Arrays.asList(
+                        newFile(1L, 1024L),
+                        newFile(1025L, 2049L),
+                        newFile(2050L, 2500L),
+                        newFile(2501L, 4096L),
+                        newFile(4097L, 6000L));
+
+        List<DataFileMeta> toCompact = totalFiles.subList(0, 4);
+        List<DataFileMeta> toRefresh = totalFiles.subList(1, 5);
+
+        BucketedAppendCompactManager manager =
+                new BucketedAppendCompactManager(
+                        null, // not used
+                        toCompact,
+                        null,
+                        minFileNum,
+                        maxFileNum,
+                        targetFileSize,
+                        null, // not used
+                        null,
+                        () -> toRefresh);
+
+        manager.refreshToCompact();
+        assertThat(manager.getToCompact()).containsExactlyElementsOf(totalFiles);
+    }
+
     private void innerTest(
             List<DataFileMeta> toCompactBeforePick,
             boolean expectedPresent,