[core] Make compact_manifest perform a full manifest sort

docs/docs/flink/procedures/compaction.md

@@ -250,6 +250,10 @@ To compact_manifest the manifests. Arguments:
 
 - manifest_sort_max_rewrite_size (String, optional): maximum manifest size rewritten by one sort pass.
 
+When manifest sort is enabled, `compact_manifest` performs a full sort using the layout selected
+from the table options. The existing `manifest_sort_max_rewrite_size` limit still controls the
+amount of manifest data rewritten in one invocation.
+
 **Syntax**
 
 ```sql

docs/docs/spark/procedures/maintenance.md

@@ -126,6 +126,9 @@ Compact manifest files.
 - `manifest_sort_enabled` (`BOOLEAN`, optional): whether to use manifest sort rewrite for this invocation.
 - `manifest_sort_partition_field` (`STRING`, optional): partition field used to sort manifest entries. Defaults to the first partition field.
 - `manifest_sort_max_rewrite_size` (`STRING`, optional): maximum manifest size rewritten by one sort pass.
+When manifest sort is enabled, `compact_manifest` performs a full sort using the layout selected
+from the table options. The existing `manifest_sort_max_rewrite_size` limit still controls the
+amount of manifest data rewritten in one invocation.
 
 ```sql
 CALL sys.compact_manifest(`table` => 'default.T');

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

@@ -1615,8 +1615,9 @@ private boolean compactManifestOnce() {
                         mergeBeforeManifests,
                         manifestFile,
                         partitionType,
-                        manifestCompactionOptions(options, mergeBeforeManifests, partitionType),
-                        ioManager);
+                        manifestCompactionOptions(options),
+                        ioManager,
+                        true);
 
         if (new HashSet<>(mergeBeforeManifests).equals(new HashSet<>(mergeAfterManifests))) {
             // no need to commit this snapshot, because no compact were happened
@@ -1655,17 +1656,12 @@ 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.
+    static CoreOptions manifestCompactionOptions(CoreOptions options) {
+        // Use copied options so explicit manifest compaction always takes the full-compaction path
+        // without changing the table options used by regular commits.
         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));
-        }
+        compactOptions.set(CoreOptions.MANIFEST_MERGE_MIN_COUNT, 1);
+        compactOptions.set(CoreOptions.MANIFEST_FULL_COMPACTION_FILE_SIZE, MemorySize.ofBytes(1));
         return new CoreOptions(compactOptions);
     }
 

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

@@ -57,6 +57,16 @@ public static List<ManifestFileMeta> merge(
             RowType partitionType,
             CoreOptions options,
             @Nullable IOManager ioManager) {
+        return merge(input, manifestFile, partitionType, options, ioManager, false);
+    }
+
+    static List<ManifestFileMeta> merge(
+            List<ManifestFileMeta> input,
+            ManifestFile manifestFile,
+            RowType partitionType,
+            CoreOptions options,
+            @Nullable IOManager ioManager,
+            boolean fullCompaction) {
         // these are the newly created manifest files, clean them up if exception occurs
         List<ManifestFileMeta> newFilesForAbort = new ArrayList<>();
 
@@ -66,7 +76,13 @@ public static List<ManifestFileMeta> merge(
             // RowID ranges, so they do not require partition fields.
             if (canUseManifestSort(input, partitionType, options)) {
                 return ManifestFileSorter.trySortCompaction(
-                        input, newFilesForAbort, manifestFile, partitionType, options, ioManager);
+                        input,
+                        newFilesForAbort,
+                        manifestFile,
+                        partitionType,
+                        options,
+                        ioManager,
+                        fullCompaction);
             }
 
             if (options.manifestMergeOptimizeEnabled()) {

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

@@ -71,6 +71,7 @@ public class ManifestFileSorter {
     /** Context object that carries shared state across compaction methods. */
     static class CompactionContext {
         final boolean fullCompaction;
+        final boolean fullSort;
         final boolean runMergeOptimizeEnabled;
         final ManifestSortKey sortKey;
         final RowType partitionType;
@@ -91,6 +92,7 @@ static class CompactionContext {
 
         CompactionContext(
                 boolean fullCompaction,
+                boolean fullSort,
                 boolean runMergeOptimizeEnabled,
                 ManifestSortKey sortKey,
                 RowType partitionType,
@@ -100,6 +102,7 @@ static class CompactionContext {
                 List<ManifestAdjacentSortedRun> levelRuns,
                 List<ManifestAdjacentSortedRun> pickedRuns) {
             this.fullCompaction = fullCompaction;
+            this.fullSort = fullSort;
             this.runMergeOptimizeEnabled = runMergeOptimizeEnabled;
             this.sortKey = sortKey;
             this.partitionType = partitionType;
@@ -153,7 +156,8 @@ static List<ManifestFileMeta> trySortCompaction(
             ManifestFile manifestFile,
             RowType partitionType,
             CoreOptions options,
-            @Nullable IOManager ioManager)
+            @Nullable IOManager ioManager,
+            boolean fullSort)
             throws Exception {
         String sortPartitionField = options.manifestSortPartitionField();
         boolean bucketed = options.bucket() > 0 || options.bucket() == BucketMode.POSTPONE_BUCKET;
@@ -181,6 +185,7 @@ static List<ManifestFileMeta> trySortCompaction(
                         suggestedMetaSize,
                         suggestedMinMetaCount,
                         fullCompactionThreshold,
+                        fullSort,
                         maxRewriteSize,
                         maxSizeAmplificationPercent,
                         sortedRunSizeRatio,
@@ -225,21 +230,25 @@ private static Optional<List<ManifestFileMeta>> tryFullCompaction(
             long suggestedMetaSize,
             int suggestedMinMetaCount,
             long fullCompactionThreshold,
+            boolean fullSort,
             long maxRewriteSize,
             int maxSizeAmplificationPercent,
             int sortedRunSizeRatio,
             ManifestEntryExternalSort.ExternalSortConfig externalSortConfig,
             @Nullable Integer manifestReadParallelism)
             throws Exception {
         // Step 1: Check if full compaction threshold is met
-        if (!reachesFullCompactionThreshold(input, suggestedMetaSize, fullCompactionThreshold)) {
+        if (!fullSort
+                && !reachesFullCompactionThreshold(
+                        input, suggestedMetaSize, fullCompactionThreshold)) {
             return Optional.empty();
         }
         // Step 2: Prepare compaction context
         CompactionContext ctx =
                 prepareCompaction(
                         input,
                         true,
+                        fullSort,
                         manifestFile,
                         partitionType,
                         sortPartitionField,
@@ -253,7 +262,8 @@ private static Optional<List<ManifestFileMeta>> tryFullCompaction(
                         manifestReadParallelism);
         try {
             List<ManifestAdjacentSortedRun> levelRuns = ctx.levelRuns;
-            List<ManifestAdjacentSortedRun> pickedRuns = ctx.pickedRuns;
+            List<ManifestAdjacentSortedRun> pickedRuns =
+                    fullSort ? new ArrayList<>(levelRuns) : ctx.pickedRuns;
 
             if (pickedRuns.isEmpty() && ctx.defaultCompactFiles.isEmpty()) {
                 LOG.debug(
@@ -283,9 +293,22 @@ private static Optional<List<ManifestFileMeta>> tryFullCompaction(
             }
             pickedFiles.addAll(ctx.defaultCompactFiles.keySet());
 
-            // Step 4: Split into sections and merge small adjacent sections
-            List<Section> sections = splitIntoSections(pickedFiles, ctx);
-            sections = mergeSmallAdjacentSections(sections, suggestedMetaSize);
+            // Step 4: A full sort uses one global section so entries from all existing runs can be
+            // clustered using the layout selected from the table options.
+            List<Section> sections;
+            if (fullSort) {
+                long totalSize = 0L;
+                boolean hasDefaultCompactFile = false;
+                for (ManifestFileMeta file : pickedFiles) {
+                    totalSize += file.fileSize();
+                    hasDefaultCompactFile |= ctx.isMarkedForDefaultCompaction(file);
+                }
+                sections = new ArrayList<>();
+                sections.add(new Section(pickedFiles, totalSize, hasDefaultCompactFile));
+            } else {
+                sections = splitIntoSections(pickedFiles, ctx);
+                sections = mergeSmallAdjacentSections(sections, suggestedMetaSize);
+            }
 
             LOG.info(
                     "Manifest sort full compact: pickedFiles={}, sections={}.",
@@ -343,6 +366,7 @@ private static List<ManifestFileMeta> tryMinorCompaction(
                 prepareCompaction(
                         input,
                         false,
+                        false,
                         manifestFile,
                         partitionType,
                         sortPartitionField,
@@ -458,6 +482,7 @@ private static List<ManifestFileMeta> tryMinorCompaction(
     private static CompactionContext prepareCompaction(
             List<ManifestFileMeta> input,
             boolean fullCompaction,
+            boolean fullSort,
             ManifestFile manifestFile,
             RowType partitionType,
             String sortPartitionField,
@@ -500,6 +525,7 @@ private static CompactionContext prepareCompaction(
 
         return new CompactionContext(
                 fullCompaction,
+                fullSort,
                 useRunMergeOptimize,
                 sortKey,
                 partitionType,
@@ -862,15 +888,17 @@ private static void rewriteSections(
         for (int i = 0; i < sections.size(); i++) {
             Section section = sections.get(i);
 
-            // A single-file section is always handled directly, regardless of the budget.
-            if (section.files.size() == 1) {
+            // Preserve the ordinary-compaction shortcut: an unchanged singleton must not consume
+            // the sort rewrite limit. Explicit full sort intentionally rewrites the singleton.
+            if (!ctx.fullSort && section.files.size() == 1) {
                 rewriteSection(
                         section.files,
                         output,
                         sortNewFiles,
                         ctx,
                         manifestFile,
-                        manifestReadParallelism);
+                        manifestReadParallelism,
+                        false);
                 continue;
             }
 
@@ -886,7 +914,8 @@ private static void rewriteSections(
                             sortNewFiles,
                             ctx,
                             manifestFile,
-                            manifestReadParallelism);
+                            manifestReadParallelism,
+                            true);
                 } else {
                     // Phase 1b: first overflow -- split the section at the budget boundary,
                     // rewrite the affordable head, and append the remaining tail back for later
@@ -966,7 +995,8 @@ private static Section splitSectionAndRewriteHead(
             }
         }
 
-        rewriteSection(headFiles, output, sortNewFiles, ctx, manifestFile, manifestReadParallelism);
+        rewriteSection(
+                headFiles, output, sortNewFiles, ctx, manifestFile, manifestReadParallelism, true);
 
         if (tailFiles.isEmpty()) {
             return null;
@@ -1044,7 +1074,8 @@ private static void unsortedCompactSection(
                         sortNewFiles,
                         ctx,
                         manifestFile,
-                        manifestReadParallelism);
+                        manifestReadParallelism,
+                        false);
                 candidates.clear();
                 candidatesSize = 0;
             }
@@ -1058,7 +1089,8 @@ private static void unsortedCompactSection(
                         sortNewFiles,
                         ctx,
                         manifestFile,
-                        manifestReadParallelism);
+                        manifestReadParallelism,
+                        false);
             } else {
                 output.addAllUnchanged(candidates);
             }
@@ -1077,10 +1109,13 @@ private static void rewriteSection(
             List<ManifestFileMeta> sortNewFiles,
             CompactionContext ctx,
             ManifestFile manifestFile,
-            @Nullable Integer manifestReadParallelism)
+            @Nullable Integer manifestReadParallelism,
+            boolean allowFullRewrite)
             throws Exception {
         // Skip rewrite for single file not in delete-range.
-        if (section.size() == 1 && !ctx.defaultCompactFiles.getOrDefault(section.get(0), false)) {
+        if (section.size() == 1
+                && !(allowFullRewrite && ctx.fullSort)
+                && !ctx.defaultCompactFiles.getOrDefault(section.get(0), false)) {
             output.addUnchanged(section.get(0));
             return;
         }