[core] Support dedicated full compact to external paths

Author: mxdzs0612

docs/layouts/shortcodes/generated/core_configuration.html

@@ -182,6 +182,12 @@
             <td>Double</td>
             <td>Ratio of the deleted rows in a data file to be forced compacted for append-only table.</td>
         </tr>
+        <tr>
+            <td><h5>compaction.force-pick-all-files</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>Whether to force pick all files for a full compaction. Usually seen in a compaction task to external paths.</td>
+        </tr>
         <tr>
             <td><h5>compaction.force-up-level-0</h5></td>
             <td style="word-wrap: break-word;">false</td>

paimon-api/src/main/java/org/apache/paimon/CoreOptions.java

@@ -151,6 +151,13 @@ public class CoreOptions implements Serializable {
                                     + ExternalPathStrategy.SPECIFIC_FS
                                     + ", should be the prefix scheme of the external path, now supported are s3 and oss.");
 
+    public static final ConfigOption<Boolean> COMPACTION_FORCE_PICK_ALL_FILES =
+            key("compaction.force-pick-all-files")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "Whether to force pick all files for a full compaction. Usually seen in a compaction task to external paths.");
+
     @ExcludeFromDocumentation("Internal use only")
     public static final ConfigOption<String> PATH =
             key("path")
@@ -2466,6 +2473,10 @@ public String externalSpecificFS() {
         return options.get(DATA_FILE_EXTERNAL_PATHS_SPECIFIC_FS);
     }
 
+    public Boolean externalCompact() {
+        return options.get(COMPACTION_FORCE_PICK_ALL_FILES);
+    }
+
     public String partitionTimestampFormatter() {
         return options.get(PARTITION_TIMESTAMP_FORMATTER);
     }

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

@@ -60,6 +60,7 @@ public class BucketedAppendCompactManager extends CompactFutureManager {
     private final PriorityQueue<DataFileMeta> toCompact;
     private final int minFileNum;
     private final long targetFileSize;
+    private final boolean externalCompact;
     private final CompactRewriter rewriter;
 
     private List<DataFileMeta> compacting;
@@ -72,6 +73,7 @@ public BucketedAppendCompactManager(
             @Nullable DeletionVectorsMaintainer dvMaintainer,
             int minFileNum,
             long targetFileSize,
+            boolean externalCompact,
             CompactRewriter rewriter,
             @Nullable CompactionMetrics.Reporter metricsReporter) {
         this.executor = executor;
@@ -80,6 +82,7 @@ public BucketedAppendCompactManager(
         this.toCompact.addAll(restored);
         this.minFileNum = minFileNum;
         this.targetFileSize = targetFileSize;
+        this.externalCompact = externalCompact;
         this.rewriter = rewriter;
         this.metricsReporter = metricsReporter;
     }
@@ -99,8 +102,9 @@ private void triggerFullCompaction() {
                 "A compaction task is still running while the user "
                         + "forces a new compaction. This is unexpected.");
         // if deletion vector enables, always trigger compaction.
-        if (toCompact.isEmpty()
-                || (dvMaintainer == null && toCompact.size() < FULL_COMPACT_MIN_FILE)) {
+        if (!externalCompact
+                && (toCompact.isEmpty()
+                        || (dvMaintainer == null && toCompact.size() < FULL_COMPACT_MIN_FILE))) {
             return;
         }
 
@@ -114,6 +118,7 @@ private void triggerFullCompaction() {
                                 dvMaintainer,
                                 toCompact,
                                 targetFileSize,
+                                externalCompact,
                                 rewriter,
                                 metricsReporter));
         recordCompactionsQueuedRequest();
@@ -238,25 +243,28 @@ public static class FullCompactTask extends CompactTask {
         private final DeletionVectorsMaintainer dvMaintainer;
         private final LinkedList<DataFileMeta> toCompact;
         private final long targetFileSize;
+        private final boolean externalCompact;
         private final CompactRewriter rewriter;
 
         public FullCompactTask(
                 DeletionVectorsMaintainer dvMaintainer,
                 Collection<DataFileMeta> inputs,
                 long targetFileSize,
+                boolean externalCompact,
                 CompactRewriter rewriter,
                 @Nullable CompactionMetrics.Reporter metricsReporter) {
             super(metricsReporter);
             this.dvMaintainer = dvMaintainer;
             this.toCompact = new LinkedList<>(inputs);
             this.targetFileSize = targetFileSize;
+            this.externalCompact = externalCompact;
             this.rewriter = rewriter;
         }
 
         @Override
         protected CompactResult doCompact() throws Exception {
             // remove large files
-            while (!toCompact.isEmpty()) {
+            while (!externalCompact && !toCompact.isEmpty()) {
                 DataFileMeta file = toCompact.peekFirst();
                 // the data file with deletion file always need to be compacted.
                 if (file.fileSize() >= targetFileSize && !hasDeletionFile(file)) {
@@ -281,7 +289,7 @@ protected CompactResult doCompact() throws Exception {
                         small++;
                     }
                 }
-                if (small > big && toCompact.size() >= FULL_COMPACT_MIN_FILE) {
+                if (externalCompact || (small > big && toCompact.size() >= FULL_COMPACT_MIN_FILE)) {
                     return compact(null, toCompact, rewriter);
                 } else {
                     return result(emptyList(), emptyList());

paimon-core/src/main/java/org/apache/paimon/mergetree/compact/CompactStrategy.java

@@ -55,7 +55,8 @@ static Optional<CompactUnit> pickFullCompaction(
             int numLevels,
             List<LevelSortedRun> runs,
             @Nullable RecordLevelExpire recordLevelExpire,
-            @Nullable DeletionVectorsMaintainer dvMaintainer) {
+            @Nullable DeletionVectorsMaintainer dvMaintainer,
+            boolean externalCompact) {
         int maxLevel = numLevels - 1;
         if (runs.isEmpty()) {
             // no sorted run, no need to compact
@@ -64,7 +65,10 @@ static Optional<CompactUnit> pickFullCompaction(
             List<DataFileMeta> filesToBeCompacted = new ArrayList<>();
 
             for (DataFileMeta file : runs.get(0).run().files()) {
-                if (recordLevelExpire != null && recordLevelExpire.isExpireFile(file)) {
+                if (externalCompact) {
+                    // add all files when it is an external compaction
+                    filesToBeCompacted.add(file);
+                } else if (recordLevelExpire != null && recordLevelExpire.isExpireFile(file)) {
                     // check record level expire for large files
                     filesToBeCompacted.add(file);
                 } else if (dvMaintainer != null

paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManager.java

@@ -65,6 +65,7 @@ public class MergeTreeCompactManager extends CompactFutureManager {
     @Nullable private final DeletionVectorsMaintainer dvMaintainer;
     private final boolean lazyGenDeletionFile;
     private final boolean needLookup;
+    private final boolean externalCompact;
 
     @Nullable private final RecordLevelExpire recordLevelExpire;
 
@@ -80,7 +81,8 @@ public MergeTreeCompactManager(
             @Nullable DeletionVectorsMaintainer dvMaintainer,
             boolean lazyGenDeletionFile,
             boolean needLookup,
-            @Nullable RecordLevelExpire recordLevelExpire) {
+            @Nullable RecordLevelExpire recordLevelExpire,
+            boolean externalCompact) {
         this.executor = executor;
         this.levels = levels;
         this.strategy = strategy;
@@ -93,6 +95,7 @@ public MergeTreeCompactManager(
         this.lazyGenDeletionFile = lazyGenDeletionFile;
         this.recordLevelExpire = recordLevelExpire;
         this.needLookup = needLookup;
+        this.externalCompact = externalCompact;
 
         MetricUtils.safeCall(this::reportMetrics, LOG);
     }
@@ -135,7 +138,11 @@ public void triggerCompaction(boolean fullCompaction) {
             }
             optionalUnit =
                     CompactStrategy.pickFullCompaction(
-                            levels.numberOfLevels(), runs, recordLevelExpire, dvMaintainer);
+                            levels.numberOfLevels(),
+                            runs,
+                            recordLevelExpire,
+                            dvMaintainer,
+                            externalCompact);
         } else {
             if (taskFuture != null) {
                 return;
@@ -210,7 +217,8 @@ private void submitCompaction(CompactUnit unit, boolean dropDelete) {
                         metricsReporter,
                         compactDfSupplier,
                         dvMaintainer,
-                        recordLevelExpire);
+                        recordLevelExpire,
+                        externalCompact);
         if (LOG.isDebugEnabled()) {
             LOG.debug(
                     "Pick these files (name, level, size) for compaction: {}",

paimon-core/src/main/java/org/apache/paimon/mergetree/compact/MergeTreeCompactTask.java

@@ -55,6 +55,7 @@ public class MergeTreeCompactTask extends CompactTask {
     private int upgradeFilesNum;
 
     @Nullable private final RecordLevelExpire recordLevelExpire;
+    private final boolean externalCompact;
     @Nullable private final DeletionVectorsMaintainer dvMaintainer;
 
     public MergeTreeCompactTask(
@@ -67,7 +68,8 @@ public MergeTreeCompactTask(
             @Nullable CompactionMetrics.Reporter metricsReporter,
             Supplier<CompactDeletionFile> compactDfSupplier,
             @Nullable DeletionVectorsMaintainer dvMaintainer,
-            @Nullable RecordLevelExpire recordLevelExpire) {
+            @Nullable RecordLevelExpire recordLevelExpire,
+            boolean externalCompact) {
         super(metricsReporter);
         this.minFileSize = minFileSize;
         this.rewriter = rewriter;
@@ -78,6 +80,7 @@ public MergeTreeCompactTask(
         this.dropDelete = dropDelete;
         this.maxLevel = maxLevel;
         this.recordLevelExpire = recordLevelExpire;
+        this.externalCompact = externalCompact;
 
         this.upgradeFilesNum = 0;
     }
@@ -127,21 +130,23 @@ protected String logMetric(
     private void upgrade(DataFileMeta file, CompactResult toUpdate) throws Exception {
         if (file.level() == outputLevel) {
             if (isContainExpiredRecords(file)
+                    || externalCompact
                     || (dvMaintainer != null
                             && dvMaintainer.deletionVectorOf(file.fileName()).isPresent())) {
                 /*
                  * 1. if the large file in maxLevel has expired records, we need to rewrite it.
-                 * 2. if the large file in maxLevel has corresponding deletion vector, we need to rewrite it.
+                 * 2. if it is an external compaction, we need to rewrite all files.
+                 * 3. if the large file in maxLevel has corresponding deletion vector, we need to rewrite it.
                  */
                 rewriteFile(file, toUpdate);
             }
             return;
         }
 
         if (outputLevel != maxLevel || file.deleteRowCount().map(d -> d == 0).orElse(false)) {
-            if (isContainExpiredRecords(file)) {
-                // if the file which could be directly upgraded has expired records, we need to
-                // rewrite it
+            if (isContainExpiredRecords(file) || externalCompact) {
+                // if the file which could be directly upgraded has expired records, or it is an
+                // external compaction, we need to rewrite it
                 rewriteFile(file, toUpdate);
             } else {
                 CompactResult upgradeResult = rewriter.upgrade(outputLevel, file);

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

@@ -91,6 +91,7 @@ protected CompactManager getCompactManager(
                     dvMaintainer,
                     options.compactionMinFileNum(),
                     options.targetFileSize(false),
+                    options.externalCompact(),
                     files -> compactRewrite(partition, bucket, dvFactory, files),
                     compactionMetrics == null
                             ? null

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

@@ -290,7 +290,8 @@ private CompactManager createCompactManager(
                     dvMaintainer,
                     options.prepareCommitWaitCompaction(),
                     options.needLookup(),
-                    recordLevelExpire);
+                    recordLevelExpire,
+                    options.externalCompact());
         }
     }
 

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

@@ -652,6 +652,7 @@ private Pair<AppendOnlyWriter, List<DataFileMeta>> createWriter(
                         null,
                         MIN_FILE_NUM,
                         targetFileSize,
+                        false,
                         compactBefore -> {
                             latch.await();
                             return compactBefore.isEmpty()

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

@@ -207,6 +207,7 @@ private void innerTest(
                         null,
                         minFileNum,
                         targetFileSize,
+                        false,
                         null, // not used
                         null);
         Optional<List<DataFileMeta>> actual = manager.pickCompactBefore();