[core] Support dedicated full compact to external paths for pk table

Author: mxdzs0612

docs/layouts/shortcodes/generated/core_configuration.html

@@ -356,6 +356,12 @@
             <td>Boolean</td>
             <td>Optional endInput check partition expire used in case of batch mode or bounded stream.</td>
         </tr>
+        <tr>
+            <td><h5>external.compact</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>Whether to execute full compaction to external paths.</td>
+        </tr>
         <tr>
             <td><h5>fields.default-aggregate-function</h5></td>
             <td style="word-wrap: break-word;">(none)</td>

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

@@ -151,6 +151,12 @@ 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_EXTERNAL =
+            key("compaction.external")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription("Whether to execute full compaction to external paths.");
+
     @ExcludeFromDocumentation("Internal use only")
     public static final ConfigOption<String> PATH =
             key("path")
@@ -2457,6 +2463,10 @@ public String externalSpecificFS() {
         return options.get(DATA_FILE_EXTERNAL_PATHS_SPECIFIC_FS);
     }
 
+    public Boolean externalCompact() {
+        return options.get(COMPACTION_EXTERNAL);
+    }
+
     public String partitionTimestampFormatter() {
         return options.get(PARTITION_TIMESTAMP_FORMATTER);
     }

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/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/mergetree/MergeTreeTestBase.java

@@ -542,7 +542,8 @@ private MergeTreeCompactManager createCompactManager(
                 null,
                 false,
                 options.needLookup(),
-                null);
+                null,
+                false);
     }
 
     static class MockFailResultCompactionManager extends MergeTreeCompactManager {
@@ -566,7 +567,8 @@ public MockFailResultCompactionManager(
                     null,
                     false,
                     false,
-                    null);
+                    null,
+                    false);
         }
 
         protected CompactResult obtainCompactResult()

paimon-core/src/test/java/org/apache/paimon/mergetree/compact/MergeTreeCompactManagerTest.java

@@ -208,7 +208,8 @@ public void testIsCompacting() {
                         null,
                         false,
                         true,
-                        null);
+                        null,
+                        false);
 
         MergeTreeCompactManager defaultManager =
                 new MergeTreeCompactManager(
@@ -223,7 +224,8 @@ public void testIsCompacting() {
                         null,
                         false,
                         false,
-                        null);
+                        null,
+                        false);
 
         assertThat(lookupManager.compactNotCompleted()).isTrue();
         assertThat(defaultManager.compactNotCompleted()).isFalse();
@@ -259,7 +261,8 @@ private void innerTest(
                         null,
                         false,
                         false,
-                        null);
+                        null,
+                        false);
         manager.triggerCompaction(false);
         manager.getCompactionResult(true);
         List<LevelMinMax> outputs =

paimon-e2e-tests/src/test/java/org/apache/paimon/tests/FlinkProceduresE2eTest.java

@@ -121,4 +121,66 @@ public void testCompact() throws Exception {
         // check that second part of test data are compacted
         checkResult("20221205, 1, 101", "20221206, 1, 101");
     }
+
+    @Test
+    public void testExternalCompact() throws Exception {
+        // create hive catalog to test catalog loading
+        String warehouse = HDFS_ROOT + "/" + UUID.randomUUID() + ".warehouse";
+        String externalPath = HDFS_ROOT + "/" + UUID.randomUUID() + ".warehouse/external";
+        String catalogDdl =
+                String.format(
+                        "CREATE CATALOG ts_catalog WITH (\n"
+                                + "    'type' = 'paimon',\n"
+                                + "    'warehouse' = '%s',\n"
+                                + "    'metastore' = 'hive',\n"
+                                + "    'uri' = 'thrift://hive-metastore:9083'\n"
+                                + ");",
+                        warehouse);
+        String useCatalogCmd = "USE CATALOG ts_catalog;";
+
+        String tableDdl =
+                "CREATE TABLE IF NOT EXISTS ts_table (\n"
+                        + "    dt STRING,\n"
+                        + "    k INT,\n"
+                        + "    v INT,\n"
+                        + "    PRIMARY KEY (dt, k) NOT ENFORCED\n"
+                        + ") PARTITIONED BY (dt) WITH (\n"
+                        + "    'data-file.external-paths' = '"
+                        + externalPath
+                        + "',\n"
+                        + "    'bucket' = '1',\n"
+                        + "    'write-only' = 'true'\n"
+                        + ");";
+
+        String insert1 =
+                "INSERT INTO ts_table VALUES ('2023-01-13', 0, 0), ('2023-01-14', 0, 0), ('2023-01-13', 0, 0), "
+                        + "('2023-01-15', 0, 1), ('2023-01-15', 0, 1), ('2023-01-15', 0, 1), "
+                        + "('2023-01-16', 1, 0), ('2023-01-17', 1, 0), ('2023-01-18', 1, 0), "
+                        + "('2023-01-19', 1, 1), ('2023-01-20', 1, 1), ('2023-01-21', 1, 1);";
+
+        String insert2 =
+                "INSERT INTO ts_table VALUES ('2023-01-13', 0, 1), ('2023-01-14', 0, 1), ('2023-01-13', 0, 1), "
+                        + "('2023-01-15', 0, 2), ('2023-01-15', 0, 2), ('2023-01-15', 0, 2), "
+                        + "('2023-01-16', 1, 3), ('2023-01-17', 1, 3), ('2023-01-18', 1, 3), "
+                        + "('2023-01-16', 2, 3), ('2023-01-17', 2, 3), ('2023-01-18', 2, 3), ";
+
+        runBatchSql(
+                "SET 'table.dml-sync' = 'true';\n" + insert1, catalogDdl, useCatalogCmd, tableDdl);
+        runBatchSql(
+                "SET 'table.dml-sync' = 'true';\n" + insert2, catalogDdl, useCatalogCmd, tableDdl);
+
+        Thread.sleep(5000);
+
+        // execute external compact procedure
+        String callStatement;
+        if (System.getProperty("test.flink.main.version").compareTo("1.18") == 0) {
+            callStatement = "CALL sys.compact('default.ts_table');";
+        } else {
+            callStatement =
+                    "CALL sys.compact(table => 'default.ts_table', partition_idle_time =>'1s'," +
+                            " options => 'external.compact=true,data-file.external-paths.strategy=specific-fs,data-file.external-paths.specific-fs=hdfs');;";
+        }
+
+        runStreamingSql(callStatement, catalogDdl, useCatalogCmd);
+    }
 }