[flink][spark] Support manifest sort in compact_manifest

docs/docs/flink/procedures.md

@@ -925,17 +925,22 @@ All available procedures are listed below.
       <td>
          CALL [catalog.]sys.compact_manifest(`table` => 'identifier')<br/>
          CALL [catalog.]sys.compact_manifest(`table` => 'identifier', 'options' => 'key1=value1,key2=value2')<br/>
-         CALL [catalog.]sys.compact_manifest(`table` => 'identifier', `dry_run` => true)
+         CALL [catalog.]sys.compact_manifest(`table` => 'identifier', `dry_run` => true)<br/>
+         CALL [catalog.]sys.compact_manifest(`table` => 'identifier', `manifest_sort_enabled` => true, `manifest_sort_partition_field` => 'dt', `manifest_sort_max_rewrite_size` => '1 gb')
       </td>
       <td>
          To compact_manifest the manifests. Arguments:
             <li>table: the target table identifier. Cannot be empty.</li>
             <li>options: the additional dynamic options of the table. It prioritizes higher than original `tableProp` and lower than `procedureArg`.</li>
-            <li>dry_run (Boolean, optional): when true, returns manifest metadata statistics without actually compacting.</li>
+            <li>dry_run (Boolean, optional): when true, returns manifest metadata statistics without actually compacting. When manifest sort is enabled, the result also contains the number of manifest files in each level built by manifest sort.</li>
+            <li>manifest_sort_enabled (Boolean, optional): whether to use manifest sort rewrite for this invocation.</li>
+            <li>manifest_sort_partition_field (String, optional): partition field used to sort manifest entries. Defaults to the first partition field.</li>
+            <li>manifest_sort_max_rewrite_size (String, optional): maximum manifest size rewritten by one sort pass.</li>
       </td>
       <td>
          CALL sys.compact_manifest(`table` => 'default.T')<br/>
-         CALL sys.compact_manifest(`table` => 'default.T', `dry_run` => true)
+         CALL sys.compact_manifest(`table` => 'default.T', `dry_run` => true)<br/>
+         CALL sys.compact_manifest(`table` => 'default.T', `manifest_sort_enabled` => true, `manifest_sort_partition_field` => 'dt', `manifest_sort_max_rewrite_size` => '1 gb')
       </td>
    </tr>
    <tr>

docs/docs/spark/procedures.md

@@ -441,11 +441,15 @@ This section introduce all available spark procedures about paimon.
          To compact_manifest the manifests. Arguments:
             <li>table: the target table identifier. Cannot be empty.</li>
             <li>options: the additional dynamic options of the table. It prioritizes higher than original `tableProp` and lower than `procedureArg`.</li>
-            <li>dry_run (Boolean, optional): when true, logs manifest metadata statistics without actually compacting. The result is printed to the application log; the SQL return value is still `true`.</li>
+            <li>dry_run (Boolean, optional): when true, logs manifest metadata statistics without actually compacting. When manifest sort is enabled, the log also contains the number of manifest files in each level built by manifest sort. The result is printed to the application log; the SQL return value is still `true`.</li>
+            <li>manifest_sort_enabled (Boolean, optional): whether to use manifest sort rewrite for this invocation.</li>
+            <li>manifest_sort_partition_field (String, optional): partition field used to sort manifest entries. Defaults to the first partition field.</li>
+            <li>manifest_sort_max_rewrite_size (String, optional): maximum manifest size rewritten by one sort pass.</li>
       </td>
       <td>
          CALL sys.compact_manifest(`table` => 'default.T')<br/>
-         CALL sys.compact_manifest(`table` => 'default.T', dry_run => true)
+         CALL sys.compact_manifest(`table` => 'default.T', dry_run => true)<br/>
+         CALL sys.compact_manifest(`table` => 'default.T', manifest_sort_enabled => true, manifest_sort_partition_field => 'dt', manifest_sort_max_rewrite_size => '1 gb')
       </td>
    </tr>
    <tr>

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

@@ -1629,16 +1629,12 @@ private boolean compactManifestOnce() {
                 manifestList.readDataManifests(latestSnapshot);
         List<ManifestFileMeta> mergeAfterManifests;
 
-        // the fist trial: use a copied options with forced full compaction settings
-        Options compactOptions = Options.fromMap(options.toMap());
-        compactOptions.set(CoreOptions.MANIFEST_MERGE_MIN_COUNT, 1);
-        compactOptions.set(CoreOptions.MANIFEST_FULL_COMPACTION_FILE_SIZE, MemorySize.ofBytes(1));
         mergeAfterManifests =
                 ManifestFileMerger.merge(
                         mergeBeforeManifests,
                         manifestFile,
                         partitionType,
-                        new CoreOptions(compactOptions),
+                        manifestCompactionOptions(options, mergeBeforeManifests, partitionType),
                         ioManager);
 
         if (new HashSet<>(mergeBeforeManifests).equals(new HashSet<>(mergeAfterManifests))) {
@@ -1677,6 +1673,20 @@ private boolean compactManifestOnce() {
         return commitSnapshotImpl(latestSnapshot, newSnapshot, emptyList());
     }
 
+    static CoreOptions manifestCompactionOptions(
+            CoreOptions options, List<ManifestFileMeta> manifests, RowType partitionType) {
+        // Use a copied options with forced full compaction settings for the legacy merge path.
+        // Manifest sort has its own full/minor picking strategy and should respect its configured
+        // thresholds.
+        Options compactOptions = Options.fromMap(options.toMap());
+        if (!ManifestFileMerger.canUseManifestSort(manifests, partitionType, options)) {
+            compactOptions.set(CoreOptions.MANIFEST_MERGE_MIN_COUNT, 1);
+            compactOptions.set(
+                    CoreOptions.MANIFEST_FULL_COMPACTION_FILE_SIZE, MemorySize.ofBytes(1));
+        }
+        return new CoreOptions(compactOptions);
+    }
+
     private boolean commitSnapshotImpl(
             @Nullable Snapshot baseSnapshot,
             Snapshot newSnapshot,

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

@@ -20,30 +20,33 @@
 
 import org.apache.paimon.CoreOptions;
 import org.apache.paimon.Snapshot;
+import org.apache.paimon.manifest.ManifestFile;
 import org.apache.paimon.manifest.ManifestFileMeta;
 import org.apache.paimon.manifest.ManifestList;
 import org.apache.paimon.options.MemorySize;
 import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.types.RowType;
 
+import java.util.ArrayList;
 import java.util.List;
 
 /** Dry run for manifest compaction. Reads only existing metadata, never writes files. */
 public class ManifestCompactDryRun {
 
     public static String execute(FileStoreTable table) {
+        CoreOptions options = new CoreOptions(table.options());
         Snapshot latestSnapshot = table.store().snapshotManager().latestSnapshot();
         if (latestSnapshot == null) {
-            return "Dry run: no snapshot exists.";
+            return appendEmptyManifestSortLevels("Dry run: no snapshot exists.", options);
         }
 
         ManifestList manifestList = table.store().manifestListFactory().create();
         List<ManifestFileMeta> manifests = manifestList.readDataManifests(latestSnapshot);
 
         if (manifests.isEmpty()) {
-            return "Dry run: 0 manifest files.";
+            return appendEmptyManifestSortLevels("Dry run: 0 manifest files.", options);
         }
 
-        CoreOptions options = new CoreOptions(table.options());
         long suggestedMetaSize = options.manifestTargetSize().getBytes();
 
         long totalFiles = manifests.size();
@@ -63,15 +66,113 @@ public static String execute(FileStoreTable table) {
             }
         }
 
-        return String.format(
-                "Dry run: %d manifest files (%s), "
-                        + "%d deleted entries in %d files, "
-                        + "%d undersized files (< %s).",
-                totalFiles,
-                MemorySize.ofBytes(totalSize),
-                totalDeletedEntries,
-                filesWithDeletedEntries,
-                smallFiles,
-                MemorySize.ofBytes(suggestedMetaSize));
+        String summary =
+                String.format(
+                        "Dry run: %d manifest files (%s), "
+                                + "%d deleted entries in %d files, "
+                                + "%d undersized files (< %s).",
+                        totalFiles,
+                        MemorySize.ofBytes(totalSize),
+                        totalDeletedEntries,
+                        filesWithDeletedEntries,
+                        smallFiles,
+                        MemorySize.ofBytes(suggestedMetaSize));
+
+        if (!options.manifestSortEnabled()) {
+            return summary;
+        }
+
+        RowType partitionType = table.schema().logicalPartitionType();
+        if (partitionType.getFieldCount() == 0
+                && !(options.dataEvolutionEnabled()
+                        && ManifestFileMeta.allContainsRowId(manifests))) {
+            return summary + " Manifest sort level files: unavailable (no sortable field).";
+        }
+
+        long[] levelFileCounts = new long[ManifestPickStrategy.MAX_LEVEL + 1];
+        List<ManifestAdjacentSortedRun> levelRuns =
+                buildLevelSortedRunsForDryRun(
+                        manifests,
+                        table.store().manifestFileFactory().create(),
+                        partitionType,
+                        options);
+        for (ManifestAdjacentSortedRun run : levelRuns) {
+            levelFileCounts[run.level()] += run.files().size();
+        }
+
+        return appendManifestSortLevels(summary, levelFileCounts);
+    }
+
+    private static List<ManifestAdjacentSortedRun> buildLevelSortedRunsForDryRun(
+            List<ManifestFileMeta> manifests,
+            ManifestFile manifestFile,
+            RowType partitionType,
+            CoreOptions options) {
+        long suggestedMetaSize = options.manifestTargetSize().getBytes();
+        boolean fullCompaction =
+                ManifestFileSorter.reachesFullCompactionThreshold(
+                        manifests,
+                        suggestedMetaSize,
+                        options.manifestFullCompactionThresholdSize().getBytes());
+        ManifestFileSorter.ManifestSortKey sortKey =
+                ManifestFileSorter.createSortKey(
+                        options.dataEvolutionEnabled(),
+                        manifests,
+                        options.manifestSortPartitionField(),
+                        partitionType);
+        ManifestFileSorter.ClassifyResult classifyResult =
+                ManifestFileSorter.classifyManifests(
+                        manifests,
+                        fullCompaction,
+                        manifestFile,
+                        partitionType,
+                        suggestedMetaSize,
+                        options.scanManifestParallelism());
+        List<ManifestAdjacentSortedRun> levelRuns = buildLevelSortedRuns(classifyResult, sortKey);
+
+        // A full compaction with no work falls through to the minor path. Mirror that fallback so
+        // the reported levels describe the path which a real compaction would use.
+        if (fullCompaction
+                && classifyResult.compactWithoutSort.isEmpty()
+                && new ManifestPickStrategy(
+                                options.maxSizeAmplificationPercent(), options.sortedRunSizeRatio())
+                        .pick(levelRuns)
+                        .isEmpty()) {
+            classifyResult =
+                    ManifestFileSorter.classifyManifests(
+                            manifests,
+                            false,
+                            manifestFile,
+                            partitionType,
+                            suggestedMetaSize,
+                            options.scanManifestParallelism());
+            levelRuns = buildLevelSortedRuns(classifyResult, sortKey);
+        }
+        return levelRuns;
+    }
+
+    private static List<ManifestAdjacentSortedRun> buildLevelSortedRuns(
+            ManifestFileSorter.ClassifyResult classifyResult,
+            ManifestFileSorter.ManifestSortKey sortKey) {
+        return classifyResult.lsmFiles.isEmpty()
+                ? new ArrayList<>()
+                : ManifestFileSorter.buildLevelSortedRuns(classifyResult.lsmFiles, sortKey);
+    }
+
+    private static String appendEmptyManifestSortLevels(String summary, CoreOptions options) {
+        return options.manifestSortEnabled()
+                ? appendManifestSortLevels(summary, new long[ManifestPickStrategy.MAX_LEVEL + 1])
+                : summary;
+    }
+
+    private static String appendManifestSortLevels(String summary, long[] levelFileCounts) {
+        return summary
+                + String.format(
+                        " Manifest sort level files: L0=%d, L1=%d, L2=%d, L3=%d, L4=%d.",
+                        levelFileCounts[0],
+                        levelFileCounts[1],
+                        levelFileCounts[2],
+                        levelFileCounts[3],
+                        levelFileCounts[4]);
     }
 }

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

@@ -89,9 +89,7 @@ public static List<ManifestFileMeta> merge(
             // If manifest-sort.enabled is enabled and there are sortable fields, use
             // trySortRewrite. Data evolution tables sort by RowID when all manifest files contain
             // RowID ranges, so they do not require partition fields.
-            if (options.manifestSortEnabled()
-                    && (partitionType.getFieldCount() > 0
-                            || (options.dataEvolutionEnabled() && allContainsRowId(input)))) {
+            if (canUseManifestSort(input, partitionType, options)) {
                 return ManifestFileSorter.trySortCompaction(
                         input, newFilesForAbort, manifestFile, partitionType, options, ioManager);
             } else {
@@ -124,6 +122,13 @@ public static List<ManifestFileMeta> merge(
         }
     }
 
+    static boolean canUseManifestSort(
+            List<ManifestFileMeta> input, RowType partitionType, CoreOptions options) {
+        return options.manifestSortEnabled()
+                && (partitionType.getFieldCount() > 0
+                        || (options.dataEvolutionEnabled() && allContainsRowId(input)));
+    }
+
     private static List<ManifestFileMeta> tryMinorCompaction(
             List<ManifestFileMeta> input,
             List<ManifestFileMeta> newFilesForAbort,

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

@@ -110,7 +110,7 @@ boolean isMarkedForUnsortedCompaction(ManifestFileMeta file) {
     }
 
     /** Result of classifying manifest files. */
-    private static class ClassifyResult {
+    static class ClassifyResult {
         final List<ManifestFileMeta> lsmFiles;
         final DeletedIdentifierSet deleteEntries;
         /**
@@ -228,13 +228,7 @@ private static Optional<List<ManifestFileMeta>> tryFullCompaction(
             @Nullable Integer manifestReadParallelism)
             throws Exception {
         // Step 1: Check if full compaction threshold is met
-        long totalDeltaFileSize = 0;
-        for (ManifestFileMeta file : input) {
-            if (file.numDeletedFiles() > 0 || file.fileSize() < suggestedMetaSize) {
-                totalDeltaFileSize += file.fileSize();
-            }
-        }
-        if (totalDeltaFileSize < fullCompactionThreshold) {
+        if (!reachesFullCompactionThreshold(input, suggestedMetaSize, fullCompactionThreshold)) {
             return Optional.empty();
         }
         // Step 2: Prepare compaction context
@@ -490,6 +484,17 @@ private static CompactionContext prepareCompaction(
                 pickedRuns);
     }
 
+    static boolean reachesFullCompactionThreshold(
+            List<ManifestFileMeta> input, long suggestedMetaSize, long fullCompactionThreshold) {
+        long totalDeltaFileSize = 0;
+        for (ManifestFileMeta file : input) {
+            if (file.numDeletedFiles() > 0 || file.fileSize() < suggestedMetaSize) {
+                totalDeltaFileSize += file.fileSize();
+            }
+        }
+        return totalDeltaFileSize >= fullCompactionThreshold;
+    }
+
     /**
      * Classify manifest files into default-compaction group and LSM group.
      *
@@ -501,7 +506,7 @@ private static CompactionContext prepareCompaction(
      *
      * @return ClassifyResult containing lsmFiles, deleteEntries, and compactWithoutSort
      */
-    private static ClassifyResult classifyManifests(
+    static ClassifyResult classifyManifests(
             List<ManifestFileMeta> input,
             boolean fullCompaction,
             ManifestFile manifestFile,
@@ -1118,7 +1123,7 @@ private static boolean containsNoDeleteEntries(List<ManifestFileMeta> section) {
         return true;
     }
 
-    private static ManifestSortKey createSortKey(
+    static ManifestSortKey createSortKey(
             boolean dataEvolutionEnabled,
             List<ManifestFileMeta> input,
             String sortPartitionField,