[core] Simplify force rewrite files in Compact Task

Author: JingsongLi

docs/layouts/shortcodes/generated/core_configuration.html

@@ -183,7 +183,7 @@
             <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-compact-all-files</h5></td>
+            <td><h5>compaction.force-rewrite-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>

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

@@ -151,8 +151,8 @@ 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_COMPACT_ALL_FILES =
-            key("compaction.force-compact-all-files")
+    public static final ConfigOption<Boolean> COMPACTION_FORCE_REWRITE_ALL_FILES =
+            key("compaction.force-rewrite-all-files")
                     .booleanType()
                     .defaultValue(false)
                     .withDescription(
@@ -2473,8 +2473,8 @@ public String externalSpecificFS() {
         return options.get(DATA_FILE_EXTERNAL_PATHS_SPECIFIC_FS);
     }
 
-    public Boolean forceCompactAllFiles() {
-        return options.get(COMPACTION_FORCE_COMPACT_ALL_FILES);
+    public Boolean forceRewriteAllFiles() {
+        return options.get(COMPACTION_FORCE_REWRITE_ALL_FILES);
     }
 
     public String partitionTimestampFormatter() {

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

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

paimon-core/src/main/java/org/apache/paimon/compact/CompactUnit.java

@@ -25,31 +25,40 @@
 import java.util.List;
 
 /** A files unit for compaction. */
-public interface CompactUnit {
+public class CompactUnit {
 
-    int outputLevel();
+    private final int outputLevel;
+    private final List<DataFileMeta> files;
+    private final boolean fileRewrite;
 
-    List<DataFileMeta> files();
+    public CompactUnit(int outputLevel, List<DataFileMeta> files, boolean fileRewrite) {
+        this.outputLevel = outputLevel;
+        this.files = files;
+        this.fileRewrite = fileRewrite;
+    }
+
+    public int outputLevel() {
+        return outputLevel;
+    }
+
+    public List<DataFileMeta> files() {
+        return files;
+    }
+
+    public boolean fileRewrite() {
+        return fileRewrite;
+    }
 
-    static CompactUnit fromLevelRuns(int outputLevel, List<LevelSortedRun> runs) {
+    public static CompactUnit fromLevelRuns(int outputLevel, List<LevelSortedRun> runs) {
         List<DataFileMeta> files = new ArrayList<>();
         for (LevelSortedRun run : runs) {
             files.addAll(run.run().files());
         }
-        return fromFiles(outputLevel, files);
+        return fromFiles(outputLevel, files, false);
     }
 
-    static CompactUnit fromFiles(int outputLevel, List<DataFileMeta> files) {
-        return new CompactUnit() {
-            @Override
-            public int outputLevel() {
-                return outputLevel;
-            }
-
-            @Override
-            public List<DataFileMeta> files() {
-                return files;
-            }
-        };
+    public static CompactUnit fromFiles(
+            int outputLevel, List<DataFileMeta> files, boolean fileRewrite) {
+        return new CompactUnit(outputLevel, files, fileRewrite);
     }
 }

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

@@ -32,7 +32,6 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Optional;
-import java.util.stream.Collectors;
 
 /** Compact strategy to decide which files to select for compaction. */
 public interface CompactStrategy {
@@ -56,16 +55,19 @@ static Optional<CompactUnit> pickFullCompaction(
             List<LevelSortedRun> runs,
             @Nullable RecordLevelExpire recordLevelExpire,
             @Nullable DeletionVectorsMaintainer dvMaintainer,
-            boolean forceCompactAllFiles) {
+            boolean forceRewriteAllFiles) {
         int maxLevel = numLevels - 1;
         if (runs.isEmpty()) {
             // no sorted run, no need to compact
             return Optional.empty();
-        } else if ((runs.size() == 1 && runs.get(0).level() == maxLevel)) {
+        }
+
+        // only max level files
+        if ((runs.size() == 1 && runs.get(0).level() == maxLevel)) {
             List<DataFileMeta> filesToBeCompacted = new ArrayList<>();
 
             for (DataFileMeta file : runs.get(0).run().files()) {
-                if (forceCompactAllFiles) {
+                if (forceRewriteAllFiles) {
                     // add all files when force compacted
                     filesToBeCompacted.add(file);
                 } else if (recordLevelExpire != null && recordLevelExpire.isExpireFile(file)) {
@@ -78,27 +80,14 @@ static Optional<CompactUnit> pickFullCompaction(
                 }
             }
 
-            if (LOG.isDebugEnabled()) {
-                LOG.debug(
-                        "Pick these files which have expired records or dv index for full compaction: {}",
-                        filesToBeCompacted.stream()
-                                .map(
-                                        file ->
-                                                String.format(
-                                                        "(%s, %d, %d)",
-                                                        file.fileName(),
-                                                        file.level(),
-                                                        file.fileSize()))
-                                .collect(Collectors.joining(", ")));
-            }
-
-            if (!filesToBeCompacted.isEmpty()) {
-                return Optional.of(CompactUnit.fromFiles(maxLevel, filesToBeCompacted));
-            } else {
+            if (filesToBeCompacted.isEmpty()) {
                 return Optional.empty();
             }
-        } else {
-            return Optional.of(CompactUnit.fromLevelRuns(maxLevel, runs));
+
+            return Optional.of(CompactUnit.fromFiles(maxLevel, filesToBeCompacted, true));
         }
+
+        // full compaction
+        return Optional.of(CompactUnit.fromLevelRuns(maxLevel, runs));
     }
 }

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

@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.mergetree.compact;
+
+import org.apache.paimon.compact.CompactResult;
+import org.apache.paimon.compact.CompactTask;
+import org.apache.paimon.compact.CompactUnit;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.mergetree.SortedRun;
+import org.apache.paimon.operation.metrics.CompactionMetrics;
+
+import javax.annotation.Nullable;
+
+import java.util.List;
+
+import static java.util.Collections.singletonList;
+
+/** Compact task for file rewrite compaction. */
+public class FileRewriteCompactTask extends CompactTask {
+
+    private final CompactRewriter rewriter;
+    private final int outputLevel;
+    private final List<DataFileMeta> files;
+    private final boolean dropDelete;
+
+    public FileRewriteCompactTask(
+            CompactRewriter rewriter,
+            CompactUnit unit,
+            boolean dropDelete,
+            @Nullable CompactionMetrics.Reporter metricsReporter) {
+        super(metricsReporter);
+        this.rewriter = rewriter;
+        this.outputLevel = unit.outputLevel();
+        this.files = unit.files();
+        this.dropDelete = dropDelete;
+    }
+
+    @Override
+    protected CompactResult doCompact() throws Exception {
+        CompactResult result = new CompactResult();
+        for (DataFileMeta file : files) {
+            rewriteFile(file, result);
+        }
+        return result;
+    }
+
+    private void rewriteFile(DataFileMeta file, CompactResult toUpdate) throws Exception {
+        List<List<SortedRun>> candidate = singletonList(singletonList(SortedRun.fromSingle(file)));
+        toUpdate.merge(rewriter.rewrite(outputLevel, dropDelete, candidate));
+    }
+}