From 68c7397b24b56a9f524203210b8a3de5e4af4e90 Mon Sep 17 00:00:00 2001
From: jianguotian <18464293+jianguotian@users.noreply.github.com>
Date: Mon, 14 Sep 2026 10:50:38 +0800
Subject: [PATCH 1/6] [core] Support one-shot forced manifest rewrite
---
docs/docs/flink/procedures/compaction.md | 22 +-
docs/docs/spark/procedures/maintenance.md | 23 ++
docs/generated/core_configuration.html | 6 +
.../java/org/apache/paimon/CoreOptions.java | 66 ++++
.../operation/ManifestCompactDryRun.java | 1 +
.../paimon/operation/ManifestFileSorter.java | 82 ++++-
.../paimon/manifest/ManifestFileMetaTest.java | 294 ++++++++++++++++++
.../flink/action/CompactManifestAction.java | 3 +-
.../procedure/CompactManifestProcedure.java | 22 +-
.../CompactManifestProcedureITCase.java | 38 +++
.../procedure/CompactManifestProcedure.java | 19 +-
.../CompactManifestProcedureTest.scala | 3 +-
12 files changed, 567 insertions(+), 12 deletions(-)
diff --git a/docs/docs/flink/procedures/compaction.md b/docs/docs/flink/procedures/compaction.md
index 71c2fd531d7b..ae8727594eec 100644
--- a/docs/docs/flink/procedures/compaction.md
+++ b/docs/docs/flink/procedures/compaction.md
@@ -250,6 +250,12 @@ To compact_manifest the manifests. Arguments:
- manifest_sort_max_rewrite_size (String, optional): maximum manifest size rewritten by one sort pass.
+- manifest_sort_order (String, optional): target layout for a one-shot rewrite. Supported values are `bucket-first` and `partition-first`. Setting it enables manifest sort and forces existing manifests to be rewritten. `bucket-first` requires a bucketed table, and explicit sort orders are not supported for data evolution tables.
+
+When `manifest_sort_order` is omitted, the existing layout selection remains unchanged: bucketed tables use bucket-first, non-bucket tables use partition-first, and data evolution tables use RowID sorting when RowID metadata is available.
+
+Set `manifest-sort.force-rewrite=true` in `options` together with `manifest_sort_enabled=true` to rewrite already compacted manifest runs using the current sort order. Use it only as a one-shot dynamic option. The existing `manifest_sort_max_rewrite_size` rewrite budget semantics still apply; raise it to migrate more manifests in one invocation.
+
**Syntax**
```sql
@@ -263,7 +269,8 @@ CALL [catalog.]sys.compact_manifest(
`table` => 'identifier',
`manifest_sort_enabled` => true,
`manifest_sort_partition_field` => 'dt',
- `manifest_sort_max_rewrite_size` => '1 gb'
+ `manifest_sort_max_rewrite_size` => '1 gb',
+ `manifest_sort_order` => 'partition-first'
);
```
@@ -280,6 +287,19 @@ CALL sys.compact_manifest(
`manifest_sort_partition_field` => 'dt',
`manifest_sort_max_rewrite_size` => '1 gb'
);
+
+CALL sys.compact_manifest(
+ `table` => 'default.T',
+ `manifest_sort_order` => 'partition-first',
+ `manifest_sort_max_rewrite_size` => '1 gb'
+);
+
+-- Switch the same bucketed table back to bucket-first layout.
+CALL sys.compact_manifest(
+ `table` => 'default.T',
+ `manifest_sort_order` => 'bucket-first',
+ `manifest_sort_max_rewrite_size` => '1 gb'
+);
```
## rescale
diff --git a/docs/docs/spark/procedures/maintenance.md b/docs/docs/spark/procedures/maintenance.md
index 1244848028fc..e636a9f03569 100644
--- a/docs/docs/spark/procedures/maintenance.md
+++ b/docs/docs/spark/procedures/maintenance.md
@@ -126,6 +126,16 @@ 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.
+- `manifest_sort_order` (`STRING`, optional): target layout for a one-shot rewrite. Supported values are `bucket-first` and `partition-first`. Setting it enables manifest sort and forces existing manifests to be rewritten. `bucket-first` requires a bucketed table, and explicit sort orders are not supported for data evolution tables.
+
+When `manifest_sort_order` is omitted, the existing layout selection remains unchanged: bucketed
+tables use bucket-first, non-bucket tables use partition-first, and data evolution tables use RowID
+sorting when RowID metadata is available.
+
+Set `manifest-sort.force-rewrite=true` in `options` together with `manifest_sort_enabled=true` to
+rewrite already compacted manifest runs using the current sort order. Use it only as a one-shot
+dynamic option. The existing `manifest_sort_max_rewrite_size` rewrite budget semantics still apply;
+raise it to migrate more manifests in one invocation.
```sql
CALL sys.compact_manifest(`table` => 'default.T');
@@ -138,6 +148,19 @@ CALL sys.compact_manifest(
manifest_sort_partition_field => 'dt',
manifest_sort_max_rewrite_size => '1 gb'
);
+
+CALL sys.compact_manifest(
+ `table` => 'default.T',
+ manifest_sort_order => 'partition-first',
+ manifest_sort_max_rewrite_size => '1 gb'
+);
+
+-- Switch the same bucketed table back to bucket-first layout.
+CALL sys.compact_manifest(
+ `table` => 'default.T',
+ manifest_sort_order => 'bucket-first',
+ manifest_sort_max_rewrite_size => '1 gb'
+);
```
## materialize_deletion_vectors
diff --git a/docs/generated/core_configuration.html b/docs/generated/core_configuration.html
index ac5730017523..2ee3a3321c16 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -1065,6 +1065,12 @@
| Boolean |
Whether to invoke manifest sort rewrite during commit.
Note: enabling this changes the semantics of 'manifest.merge-min-count'. In the sort rewrite path, small manifest files within the rewrite budget are sorted and merged directly, so the minimum-count gate no longer prevents merging a small number of under-budget manifest files when full compaction is not triggered. |
+
+ manifest-sort.force-rewrite |
+ false |
+ Boolean |
+ When 'manifest-sort.enabled' is true, force an explicit manifest compaction to rewrite already compacted manifest runs using the configured manifest sort order. The existing 'manifest-sort.max-rewrite-size' rewrite budget semantics still apply. This should be supplied as a one-shot dynamic option for maintenance, not persisted for routine writes. |
+
manifest-sort.max-rewrite-size |
256 mb |
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index 7b0665c50296..1a237eb4d63a 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -599,6 +599,27 @@ public InlineElement getDescription() {
+ " skipped. Set to a larger value to allow more aggressive"
+ " sort rewriting. The cap only limits the sorted rewrite portion and full/minor cleanup may still happen beyond it.");
+ public static final ConfigOption MANIFEST_SORT_FORCE_REWRITE =
+ key("manifest-sort.force-rewrite")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "When 'manifest-sort.enabled' is true, force an explicit manifest"
+ + " compaction to rewrite already compacted manifest runs using"
+ + " the configured manifest sort order."
+ + " The existing 'manifest-sort.max-rewrite-size' rewrite budget"
+ + " semantics still apply."
+ + " This should be supplied as a one-shot dynamic option for"
+ + " maintenance, not persisted for routine writes.");
+
+ @ExcludeFromDocumentation("Only used by compact_manifest maintenance procedure")
+ public static final ConfigOption MANIFEST_SORT_ORDER =
+ key("manifest-sort.order")
+ .enumType(ManifestSortOrder.class)
+ .noDefaultValue()
+ .withDescription(
+ "Target manifest layout for a one-shot manifest sort rewrite.");
+
public static final ConfigOption MANIFEST_MERGE_OPTIMIZE_ENABLED =
key("manifest.merge-optimize.enabled")
.booleanType()
@@ -3227,6 +3248,15 @@ public long manifestSortMaxRewriteSize() {
return options.get(MANIFEST_SORT_MAX_REWRITE_SIZE).getBytes();
}
+ public boolean manifestSortForceRewrite() {
+ return options.get(MANIFEST_SORT_FORCE_REWRITE);
+ }
+
+ @Nullable
+ public ManifestSortOrder manifestSortOrder() {
+ return options.getOptional(MANIFEST_SORT_ORDER).orElse(null);
+ }
+
public boolean manifestMergeOptimizeEnabled() {
return options.get(MANIFEST_MERGE_OPTIMIZE_ENABLED);
}
@@ -5431,6 +5461,42 @@ public static List> getOptions() {
})
.collect(Collectors.toSet());
+ /** Target layout for an explicit manifest sort rewrite. */
+ public enum ManifestSortOrder implements DescribedEnum {
+ BUCKET_FIRST("bucket-first", "Sort manifest entries by bucket before partition."),
+ PARTITION_FIRST("partition-first", "Sort manifest entries by partition.");
+
+ private final String value;
+ private final String description;
+
+ ManifestSortOrder(String value, String description) {
+ this.value = value;
+ this.description = description;
+ }
+
+ public static ManifestSortOrder fromString(String value) {
+ for (ManifestSortOrder order : values()) {
+ if (order.value.equalsIgnoreCase(value.trim())) {
+ return order;
+ }
+ }
+ throw new IllegalArgumentException(
+ String.format(
+ "Unsupported manifest sort order '%s'. Supported values are 'bucket-first' and 'partition-first'.",
+ value));
+ }
+
+ @Override
+ public String toString() {
+ return value;
+ }
+
+ @Override
+ public InlineElement getDescription() {
+ return text(description);
+ }
+ }
+
/** Specifies the sort engine for table with primary key. */
public enum SortEngine implements DescribedEnum {
MIN_HEAP("min-heap", "Use min-heap for multiway sorting."),
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestCompactDryRun.java b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestCompactDryRun.java
index f80554ce6ea4..aebcc1622d00 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestCompactDryRun.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestCompactDryRun.java
@@ -121,6 +121,7 @@ private static List buildLevelSortedRunsForDryRun(
manifests,
options.manifestSortPartitionField(),
partitionType,
+ options.manifestSortOrder(),
options.bucket() > 0 || options.bucket() == BucketMode.POSTPONE_BUCKET);
ManifestFileSorter.ClassifyResult classifyResult =
ManifestFileSorter.classifyManifests(
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileSorter.java b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileSorter.java
index 87fd4611e72b..0f1250696829 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileSorter.java
+++ b/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 forceRewrite;
final boolean runMergeOptimizeEnabled;
final ManifestSortKey sortKey;
final RowType partitionType;
@@ -91,6 +92,7 @@ static class CompactionContext {
CompactionContext(
boolean fullCompaction,
+ boolean forceRewrite,
boolean runMergeOptimizeEnabled,
ManifestSortKey sortKey,
RowType partitionType,
@@ -100,6 +102,7 @@ static class CompactionContext {
List levelRuns,
List pickedRuns) {
this.fullCompaction = fullCompaction;
+ this.forceRewrite = forceRewrite;
this.runMergeOptimizeEnabled = runMergeOptimizeEnabled;
this.sortKey = sortKey;
this.partitionType = partitionType;
@@ -156,11 +159,13 @@ static List trySortCompaction(
@Nullable IOManager ioManager)
throws Exception {
String sortPartitionField = options.manifestSortPartitionField();
+ CoreOptions.ManifestSortOrder sortOrder = options.manifestSortOrder();
boolean bucketed = options.bucket() > 0 || options.bucket() == BucketMode.POSTPONE_BUCKET;
boolean runMergeOptimizeEnabled = options.manifestMergeOptimizeEnabled();
long suggestedMetaSize = options.manifestTargetSize().getBytes();
int suggestedMinMetaCount = options.manifestMergeMinCount();
long fullCompactionThreshold = options.manifestFullCompactionThresholdSize().getBytes();
+ boolean forceRewrite = options.manifestSortForceRewrite();
long maxRewriteSize = options.manifestSortMaxRewriteSize();
int maxSizeAmplificationPercent = options.maxSizeAmplificationPercent();
int sortedRunSizeRatio = options.sortedRunSizeRatio();
@@ -175,12 +180,14 @@ static List trySortCompaction(
manifestFile,
partitionType,
sortPartitionField,
+ sortOrder,
bucketed,
options.dataEvolutionEnabled(),
runMergeOptimizeEnabled,
suggestedMetaSize,
suggestedMinMetaCount,
fullCompactionThreshold,
+ forceRewrite,
maxRewriteSize,
maxSizeAmplificationPercent,
sortedRunSizeRatio,
@@ -195,6 +202,7 @@ static List trySortCompaction(
manifestFile,
partitionType,
sortPartitionField,
+ sortOrder,
bucketed,
options.dataEvolutionEnabled(),
runMergeOptimizeEnabled,
@@ -219,12 +227,14 @@ private static Optional> tryFullCompaction(
ManifestFile manifestFile,
RowType partitionType,
String sortPartitionField,
+ @Nullable CoreOptions.ManifestSortOrder sortOrder,
boolean bucketed,
boolean dataEvolutionEnabled,
boolean runMergeOptimizeEnabled,
long suggestedMetaSize,
int suggestedMinMetaCount,
long fullCompactionThreshold,
+ boolean forceRewrite,
long maxRewriteSize,
int maxSizeAmplificationPercent,
int sortedRunSizeRatio,
@@ -232,7 +242,9 @@ private static Optional> tryFullCompaction(
@Nullable Integer manifestReadParallelism)
throws Exception {
// Step 1: Check if full compaction threshold is met
- if (!reachesFullCompactionThreshold(input, suggestedMetaSize, fullCompactionThreshold)) {
+ if (!forceRewrite
+ && !reachesFullCompactionThreshold(
+ input, suggestedMetaSize, fullCompactionThreshold)) {
return Optional.empty();
}
// Step 2: Prepare compaction context
@@ -240,9 +252,11 @@ private static Optional> tryFullCompaction(
prepareCompaction(
input,
true,
+ forceRewrite,
manifestFile,
partitionType,
sortPartitionField,
+ sortOrder,
bucketed,
dataEvolutionEnabled,
runMergeOptimizeEnabled,
@@ -254,6 +268,9 @@ private static Optional> tryFullCompaction(
try {
List levelRuns = ctx.levelRuns;
List pickedRuns = ctx.pickedRuns;
+ if (forceRewrite) {
+ pickedRuns = new ArrayList<>(levelRuns);
+ }
if (pickedRuns.isEmpty() && ctx.defaultCompactFiles.isEmpty()) {
LOG.debug(
@@ -283,9 +300,23 @@ private static Optional> tryFullCompaction(
}
pickedFiles.addAll(ctx.defaultCompactFiles.keySet());
- // Step 4: Split into sections and merge small adjacent sections
- List sections = splitIntoSections(pickedFiles, ctx);
- sections = mergeSmallAdjacentSections(sections, suggestedMetaSize);
+ // Step 4: Split into sections and merge small adjacent sections. A forced rewrite
+ // intentionally uses one global section so entries from different already-compacted
+ // manifests can be clustered using the current sort order.
+ List sections;
+ if (forceRewrite) {
+ 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={}.",
@@ -327,6 +358,7 @@ private static List tryMinorCompaction(
ManifestFile manifestFile,
RowType partitionType,
String sortPartitionField,
+ @Nullable CoreOptions.ManifestSortOrder sortOrder,
boolean bucketed,
boolean dataEvolutionEnabled,
boolean runMergeOptimizeEnabled,
@@ -343,9 +375,11 @@ private static List tryMinorCompaction(
prepareCompaction(
input,
false,
+ false,
manifestFile,
partitionType,
sortPartitionField,
+ sortOrder,
bucketed,
dataEvolutionEnabled,
runMergeOptimizeEnabled,
@@ -458,9 +492,11 @@ private static List tryMinorCompaction(
private static CompactionContext prepareCompaction(
List input,
boolean fullCompaction,
+ boolean forceRewrite,
ManifestFile manifestFile,
RowType partitionType,
String sortPartitionField,
+ @Nullable CoreOptions.ManifestSortOrder sortOrder,
boolean bucketed,
boolean dataEvolutionEnabled,
boolean runMergeOptimizeEnabled,
@@ -475,7 +511,12 @@ private static CompactionContext prepareCompaction(
// Step 1: Resolve sort key. Data evolution tables prefer RowID ranges when available.
ManifestSortKey sortKey =
createSortKey(
- dataEvolutionEnabled, input, sortPartitionField, partitionType, bucketed);
+ dataEvolutionEnabled,
+ input,
+ sortPartitionField,
+ partitionType,
+ sortOrder,
+ bucketed);
// Step 2: Classify manifests into LSM files and collect delete entries.
ClassifyResult classification =
@@ -500,6 +541,7 @@ private static CompactionContext prepareCompaction(
return new CompactionContext(
fullCompaction,
+ forceRewrite,
useRunMergeOptimize,
sortKey,
partitionType,
@@ -1080,7 +1122,9 @@ private static void rewriteSection(
@Nullable Integer manifestReadParallelism)
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
+ && !ctx.forceRewrite
+ && !ctx.defaultCompactFiles.getOrDefault(section.get(0), false)) {
output.addUnchanged(section.get(0));
return;
}
@@ -1195,6 +1239,27 @@ static ManifestSortKey createSortKey(
String sortPartitionField,
RowType partitionType,
boolean bucketed) {
+ return createSortKey(
+ dataEvolutionEnabled, input, sortPartitionField, partitionType, null, bucketed);
+ }
+
+ static ManifestSortKey createSortKey(
+ boolean dataEvolutionEnabled,
+ List input,
+ String sortPartitionField,
+ RowType partitionType,
+ @Nullable CoreOptions.ManifestSortOrder sortOrder,
+ boolean bucketed) {
+ if (sortOrder != null && dataEvolutionEnabled) {
+ throw new IllegalArgumentException(
+ "Explicit manifest sort order is not supported for data evolution tables.");
+ }
+
+ if (sortOrder == CoreOptions.ManifestSortOrder.BUCKET_FIRST && !bucketed) {
+ throw new IllegalArgumentException(
+ "Manifest sort order 'bucket-first' requires a bucketed table.");
+ }
+
boolean rowIdSort = dataEvolutionEnabled && ManifestFileMeta.allContainsRowId(input);
if (rowIdSort) {
// RowID sorting uses the configured partition field as the primary key when specified,
@@ -1224,7 +1289,10 @@ static ManifestSortKey createSortKey(
RecordComparator fieldComparator =
CodeGenUtils.newRecordComparator(
partitionType.getFieldTypes(), new int[] {sortFieldIndex});
- if (bucketed) {
+ boolean useBucketSort =
+ sortOrder == CoreOptions.ManifestSortOrder.BUCKET_FIRST
+ || (sortOrder == null && bucketed);
+ if (useBucketSort) {
boolean compareManifestBuckets =
input.stream()
.allMatch(meta -> meta.minBucket() != null && meta.maxBucket() != null);
diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTest.java
index 6d67eddfbdcd..7d0abc7c7d73 100644
--- a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTest.java
@@ -1352,6 +1352,264 @@ public void testManifestSortUsesBucketAsPrimaryKeyForBucketedTable(int bucket) {
.containsExactly(0, 1, 0, 1);
}
+ @Test
+ public void testManifestSortForceRewriteAlreadyCompactedRuns() {
+ List physical =
+ Arrays.asList(
+ makeManifest(makeBucketEntry("a-3", 0, 3), makeBucketEntry("a-1", 0, 1)),
+ makeManifest(makeBucketEntry("b-2", 1, 2), makeBucketEntry("b-0", 1, 0)));
+ long targetSize = CoreOptions.MANIFEST_TARGET_FILE_SIZE.defaultValue().getBytes();
+ List input =
+ physical.stream()
+ .map(meta -> copyAsLegacyWithFileSize(meta, targetSize))
+ .collect(Collectors.toList());
+ assertThat(input)
+ .allMatch(
+ meta ->
+ meta.minBucket() == null
+ && meta.maxBucket() == null
+ && meta.totalBuckets() == null);
+
+ Options testOptions = new Options();
+ testOptions.set(CoreOptions.MANIFEST_SORT_ENABLED, true);
+ testOptions.set(CoreOptions.MANIFEST_SORT_MAX_REWRITE_SIZE.key(), "1G");
+ testOptions.set(CoreOptions.BUCKET, 4);
+
+ List unchanged =
+ ManifestFileMerger.merge(
+ input,
+ manifestFile,
+ getPartitionType(),
+ CoreOptions.fromMap(testOptions.toMap()));
+ assertThat(unchanged).containsExactlyInAnyOrderElementsOf(input);
+
+ testOptions.set(CoreOptions.MANIFEST_SORT_FORCE_REWRITE, true);
+ List rewritten =
+ ManifestFileMerger.merge(
+ input,
+ manifestFile,
+ getPartitionType(),
+ CoreOptions.fromMap(testOptions.toMap()));
+
+ assertEquivalentEntries(input, rewritten);
+ assertThat(rewritten)
+ .extracting(ManifestFileMeta::fileName)
+ .doesNotContainAnyElementsOf(
+ input.stream()
+ .map(ManifestFileMeta::fileName)
+ .collect(Collectors.toList()));
+ assertThat(readEntries(rewritten))
+ .extracting(ManifestEntry::bucket)
+ .containsExactly(0, 1, 2, 3);
+ assertThat(rewritten).hasSize(1);
+ assertThat(rewritten.get(0).minBucket()).isZero();
+ assertThat(rewritten.get(0).maxBucket()).isEqualTo(3);
+ assertThat(rewritten.get(0).totalBuckets()).isEqualTo(240);
+
+ testOptions.set(CoreOptions.MANIFEST_SORT_FORCE_REWRITE, false);
+ List afterMigration =
+ ManifestFileMerger.merge(
+ rewritten,
+ manifestFile,
+ getPartitionType(),
+ CoreOptions.fromMap(testOptions.toMap()));
+ assertThat(afterMigration).containsExactlyElementsOf(rewritten);
+ }
+
+ @Test
+ public void testManifestSortForceRewriteSwitchesLayout() {
+ List input =
+ Arrays.asList(
+ makeManifest(
+ makeBucketEntry("a-p0-b3", 0, 3), makeBucketEntry("a-p1-b1", 1, 1)),
+ makeManifest(
+ makeBucketEntry("b-p0-b2", 0, 2),
+ makeBucketEntry("b-p1-b0", 1, 0)));
+
+ Options testOptions = new Options();
+ testOptions.set(CoreOptions.MANIFEST_SORT_ENABLED, true);
+ testOptions.set(CoreOptions.MANIFEST_SORT_FORCE_REWRITE, true);
+ testOptions.set(CoreOptions.MANIFEST_SORT_MAX_REWRITE_SIZE.key(), "1G");
+ testOptions.set(CoreOptions.BUCKET, 4);
+ testOptions.set(
+ CoreOptions.MANIFEST_SORT_ORDER, CoreOptions.ManifestSortOrder.PARTITION_FIRST);
+
+ List partitionFirst =
+ ManifestFileMerger.merge(
+ input,
+ manifestFile,
+ getPartitionType(),
+ CoreOptions.fromMap(testOptions.toMap()));
+ assertEquivalentEntries(input, partitionFirst);
+ assertThat(readEntries(partitionFirst))
+ .extracting(entry -> entry.partition().getInt(0))
+ .containsExactly(0, 0, 1, 1);
+
+ testOptions.set(
+ CoreOptions.MANIFEST_SORT_ORDER, CoreOptions.ManifestSortOrder.BUCKET_FIRST);
+ List bucketFirst =
+ ManifestFileMerger.merge(
+ partitionFirst,
+ manifestFile,
+ getPartitionType(),
+ CoreOptions.fromMap(testOptions.toMap()));
+ assertEquivalentEntries(partitionFirst, bucketFirst);
+ assertThat(readEntries(bucketFirst))
+ .extracting(ManifestEntry::bucket)
+ .containsExactly(0, 1, 2, 3);
+ }
+
+ @Test
+ public void testManifestSortExplicitOrderValidation() {
+ List input =
+ Collections.singletonList(makeManifest(makeBucketEntry("file", 0, 0)));
+
+ Options testOptions = new Options();
+ testOptions.set(CoreOptions.MANIFEST_SORT_ENABLED, true);
+ testOptions.set(CoreOptions.MANIFEST_SORT_FORCE_REWRITE, true);
+ testOptions.set(
+ CoreOptions.MANIFEST_SORT_ORDER, CoreOptions.ManifestSortOrder.BUCKET_FIRST);
+ assertThat(
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ ManifestFileMerger.merge(
+ input,
+ manifestFile,
+ getPartitionType(),
+ CoreOptions.fromMap(testOptions.toMap()))))
+ .hasMessage("Manifest sort order 'bucket-first' requires a bucketed table.");
+
+ testOptions.set(
+ CoreOptions.MANIFEST_SORT_ORDER, CoreOptions.ManifestSortOrder.PARTITION_FIRST);
+ testOptions.set(CoreOptions.DATA_EVOLUTION_ENABLED, true);
+ assertThat(
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ ManifestFileMerger.merge(
+ input,
+ manifestFile,
+ getPartitionType(),
+ CoreOptions.fromMap(testOptions.toMap()))))
+ .hasMessage(
+ "Explicit manifest sort order is not supported for data evolution tables.");
+ }
+
+ @Test
+ public void testManifestSortForceRewriteAllLevelRuns() {
+ List physical =
+ Arrays.asList(
+ makeManifest(makeBucketEntry("a-3", 0, 3), makeBucketEntry("a-1", 2, 1)),
+ makeManifest(makeBucketEntry("b-2", 1, 2), makeBucketEntry("b-0", 3, 0)));
+ long targetSize = CoreOptions.MANIFEST_TARGET_FILE_SIZE.defaultValue().getBytes();
+ List input =
+ physical.stream()
+ .map(meta -> copyAsLegacyWithFileSize(meta, targetSize))
+ .collect(Collectors.toList());
+
+ Options testOptions = new Options();
+ testOptions.set(CoreOptions.MANIFEST_SORT_ENABLED, true);
+ testOptions.set(CoreOptions.MANIFEST_SORT_MAX_REWRITE_SIZE.key(), "1G");
+ testOptions.set(CoreOptions.BUCKET, 4);
+
+ // Without bucket metadata, manifest sort falls back to the overlapping partition ranges
+ // [0, 2] and [1, 3]. The dry run therefore shows that these files form two level runs.
+ FileStoreTable table = mock(FileStoreTable.class, RETURNS_DEEP_STUBS);
+ Snapshot snapshot = mock(Snapshot.class);
+ when(table.options()).thenReturn(testOptions.toMap());
+ when(table.store().snapshotManager().latestSnapshot()).thenReturn(snapshot);
+ when(table.store().manifestListFactory().create().readDataManifests(snapshot))
+ .thenReturn(input);
+ when(table.store().manifestFileFactory().create()).thenReturn(manifestFile);
+ when(table.schema().logicalPartitionType()).thenReturn(getPartitionType());
+ assertThat(ManifestCompactDryRun.execute(table))
+ .endsWith("Manifest sort level files: L0=0, L1=0, L2=0, L3=1, L4=1.");
+
+ List unchanged =
+ ManifestFileMerger.merge(
+ input,
+ manifestFile,
+ getPartitionType(),
+ CoreOptions.fromMap(testOptions.toMap()));
+ assertThat(unchanged).containsExactlyInAnyOrderElementsOf(input);
+
+ testOptions.set(CoreOptions.MANIFEST_SORT_FORCE_REWRITE, true);
+ List rewritten =
+ ManifestFileMerger.merge(
+ input,
+ manifestFile,
+ getPartitionType(),
+ CoreOptions.fromMap(testOptions.toMap()));
+
+ assertEquivalentEntries(input, rewritten);
+ assertThat(rewritten)
+ .extracting(ManifestFileMeta::fileName)
+ .doesNotContainAnyElementsOf(
+ input.stream()
+ .map(ManifestFileMeta::fileName)
+ .collect(Collectors.toList()));
+ assertThat(readEntries(rewritten))
+ .extracting(ManifestEntry::bucket)
+ .containsExactly(0, 1, 2, 3);
+ }
+
+ @Test
+ public void testManifestSortForceRewriteSingleManifest() {
+ ManifestFileMeta physical =
+ makeManifest(makeBucketEntry("file-3", 0, 3), makeBucketEntry("file-0", 0, 0));
+ ManifestFileMeta input =
+ copyWithFileSize(
+ physical, CoreOptions.MANIFEST_TARGET_FILE_SIZE.defaultValue().getBytes());
+
+ Options testOptions = new Options();
+ testOptions.set(CoreOptions.MANIFEST_SORT_ENABLED, true);
+ testOptions.set(CoreOptions.MANIFEST_SORT_FORCE_REWRITE, true);
+ testOptions.set(CoreOptions.BUCKET, 4);
+ List rewritten =
+ ManifestFileMerger.merge(
+ Collections.singletonList(input),
+ manifestFile,
+ getPartitionType(),
+ CoreOptions.fromMap(testOptions.toMap()));
+
+ assertEquivalentEntries(Collections.singletonList(input), rewritten);
+ assertThat(rewritten)
+ .extracting(ManifestFileMeta::fileName)
+ .doesNotContain(input.fileName());
+ assertThat(readEntries(rewritten)).extracting(ManifestEntry::bucket).containsExactly(0, 3);
+ }
+
+ @Test
+ public void testManifestSortForceRewriteRespectsRewriteBudget() {
+ long targetSize = CoreOptions.MANIFEST_TARGET_FILE_SIZE.defaultValue().getBytes();
+ List input = new ArrayList<>();
+ for (int i = 0; i < 4; i++) {
+ input.add(
+ copyWithFileSize(makeManifest(makeBucketEntry("file-" + i, 0, i)), targetSize));
+ }
+
+ Options testOptions = new Options();
+ testOptions.set(CoreOptions.MANIFEST_SORT_ENABLED, true);
+ testOptions.set(CoreOptions.MANIFEST_SORT_FORCE_REWRITE, true);
+ testOptions.set(CoreOptions.MANIFEST_SORT_MAX_REWRITE_SIZE.key(), "1B");
+ testOptions.set(CoreOptions.BUCKET, 4);
+ List rewritten =
+ ManifestFileMerger.merge(
+ input,
+ manifestFile,
+ getPartitionType(),
+ CoreOptions.fromMap(testOptions.toMap()));
+
+ Set inputNames =
+ input.stream().map(ManifestFileMeta::fileName).collect(Collectors.toSet());
+ assertThat(rewritten)
+ .extracting(ManifestFileMeta::fileName)
+ .filteredOn(inputNames::contains)
+ .hasSize(2);
+ assertEquivalentEntries(input, rewritten);
+ }
+
@ParameterizedTest
@ValueSource(ints = {-1, 4, -2})
public void testManifestSortDryRunUsesBucketRangesForBucketedTable(int bucket) {
@@ -2830,6 +3088,42 @@ private ManifestEntry makeBucketEntry(String fileName, int partition, int bucket
return ManifestEntry.create(entry.kind(), entry.partition(), bucket, 240, entry.file());
}
+ private ManifestFileMeta copyWithFileSize(ManifestFileMeta meta, long fileSize) {
+ return new ManifestFileMeta(
+ meta.fileName(),
+ fileSize,
+ meta.numAddedFiles(),
+ meta.numDeletedFiles(),
+ meta.partitionStats(),
+ meta.schemaId(),
+ meta.minBucket(),
+ meta.maxBucket(),
+ meta.minLevel(),
+ meta.maxLevel(),
+ meta.minRowId(),
+ meta.maxRowId(),
+ meta.totalBuckets(),
+ meta.extraFiles());
+ }
+
+ private ManifestFileMeta copyAsLegacyWithFileSize(ManifestFileMeta meta, long fileSize) {
+ return new ManifestFileMeta(
+ meta.fileName(),
+ fileSize,
+ meta.numAddedFiles(),
+ meta.numDeletedFiles(),
+ meta.partitionStats(),
+ meta.schemaId(),
+ null,
+ null,
+ meta.minLevel(),
+ meta.maxLevel(),
+ meta.minRowId(),
+ meta.maxRowId(),
+ null,
+ meta.extraFiles());
+ }
+
/** Create a ManifestEntry with a 3-field partition row (region, dt, hour). */
private ManifestEntry makeMultiPartEntry(
boolean isAdd, String fileName, int region, int dt, int hour) {
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactManifestAction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactManifestAction.java
index 4b70de86dbf3..7c0c123339eb 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactManifestAction.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactManifestAction.java
@@ -69,7 +69,8 @@ public void executeLocally() throws Exception {
dryRun,
manifestSortEnabled,
manifestSortPartitionField,
- manifestSortMaxRewriteSize);
+ manifestSortMaxRewriteSize,
+ null);
for (String result : results) {
LOG.info(result);
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CompactManifestProcedure.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CompactManifestProcedure.java
index 1bae106b9f6b..a12c9abcdb8b 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CompactManifestProcedure.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CompactManifestProcedure.java
@@ -59,6 +59,10 @@ public String identifier() {
@ArgumentHint(
name = "manifest_sort_max_rewrite_size",
type = @DataTypeHint("STRING"),
+ isOptional = true),
+ @ArgumentHint(
+ name = "manifest_sort_order",
+ type = @DataTypeHint("STRING"),
isOptional = true)
})
public String[] call(
@@ -68,7 +72,8 @@ public String[] call(
@Nullable Boolean dryRun,
@Nullable Boolean manifestSortEnabled,
@Nullable String manifestSortPartitionField,
- @Nullable String manifestSortMaxRewriteSize)
+ @Nullable String manifestSortMaxRewriteSize,
+ @Nullable String manifestSortOrder)
throws Exception {
FileStoreTable table = (FileStoreTable) table(tableId);
@@ -88,6 +93,21 @@ public String[] call(
dynamicOptions.put(
CoreOptions.MANIFEST_SORT_MAX_REWRITE_SIZE.key(), manifestSortMaxRewriteSize);
}
+ if (manifestSortOrder != null) {
+ if (Boolean.FALSE.equals(manifestSortEnabled)
+ || "false"
+ .equalsIgnoreCase(
+ dynamicOptions.get(CoreOptions.MANIFEST_SORT_ENABLED.key()))) {
+ throw new IllegalArgumentException(
+ "'manifest_sort_order' cannot be used with 'manifest_sort_enabled=false'.");
+ }
+ CoreOptions.ManifestSortOrder order =
+ CoreOptions.ManifestSortOrder.fromString(manifestSortOrder);
+ dynamicOptions.put(CoreOptions.MANIFEST_SORT_ORDER.key(), order.toString());
+ dynamicOptions.put(CoreOptions.MANIFEST_SORT_ENABLED.key(), Boolean.TRUE.toString());
+ dynamicOptions.put(
+ CoreOptions.MANIFEST_SORT_FORCE_REWRITE.key(), Boolean.TRUE.toString());
+ }
table = table.copy(dynamicOptions);
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CompactManifestProcedureITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CompactManifestProcedureITCase.java
index a40fa8f905f6..cef9509018f1 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CompactManifestProcedureITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CompactManifestProcedureITCase.java
@@ -18,6 +18,7 @@
package org.apache.paimon.flink.procedure;
+import org.apache.paimon.CoreOptions;
import org.apache.paimon.flink.CatalogITCaseBase;
import org.apache.paimon.flink.action.ActionFactory;
import org.apache.paimon.flink.action.CompactManifestAction;
@@ -123,6 +124,24 @@ public void testManifestSortParameters() throws Exception {
sql(procedure);
Assertions.assertThat(table.snapshotManager().latestSnapshot().id())
.isEqualTo(compactSnapshotId);
+
+ String forceRewriteProcedure =
+ "CALL sys.compact_manifest("
+ + "`table` => 'default.T_SORT', "
+ + "`manifest_sort_partition_field` => 'dt', "
+ + "`manifest_sort_max_rewrite_size` => '1 gb', "
+ + "`manifest_sort_order` => 'partition-first')";
+ sql(forceRewriteProcedure);
+ long forceRewriteSnapshotId = table.snapshotManager().latestSnapshot().id();
+ Assertions.assertThat(forceRewriteSnapshotId).isEqualTo(compactSnapshotId + 1);
+ Assertions.assertThat(paimonTable("T_SORT").options())
+ .doesNotContainKeys(
+ CoreOptions.MANIFEST_SORT_FORCE_REWRITE.key(),
+ CoreOptions.MANIFEST_SORT_ORDER.key());
+
+ sql(procedure);
+ Assertions.assertThat(table.snapshotManager().latestSnapshot().id())
+ .isEqualTo(forceRewriteSnapshotId);
}
@Test
@@ -141,6 +160,25 @@ public void testManifestSortParametersValidation() {
+ "`manifest_sort_partition_field` => 'missing')"))
.hasStackTraceContaining(
"'manifest-sort.partition-field' = 'missing' is not a partition field");
+
+ Assertions.assertThatThrownBy(
+ () ->
+ sql(
+ "CALL sys.compact_manifest("
+ + "`table` => 'default.T_INVALID', "
+ + "`manifest_sort_order` => 'unknown')"))
+ .hasStackTraceContaining(
+ "Unsupported manifest sort order 'unknown'. Supported values are 'bucket-first' and 'partition-first'.");
+
+ Assertions.assertThatThrownBy(
+ () ->
+ sql(
+ "CALL sys.compact_manifest("
+ + "`table` => 'default.T_INVALID', "
+ + "`manifest_sort_enabled` => false, "
+ + "`manifest_sort_order` => 'partition-first')"))
+ .hasStackTraceContaining(
+ "'manifest_sort_order' cannot be used with 'manifest_sort_enabled=false'.");
}
@Test
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactManifestProcedure.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactManifestProcedure.java
index 25b417e5b0a7..89c15761d3b8 100644
--- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactManifestProcedure.java
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactManifestProcedure.java
@@ -59,7 +59,8 @@ public class CompactManifestProcedure extends BaseProcedure {
ProcedureParameter.optional("dry_run", BooleanType),
ProcedureParameter.optional("manifest_sort_enabled", BooleanType),
ProcedureParameter.optional("manifest_sort_partition_field", StringType),
- ProcedureParameter.optional("manifest_sort_max_rewrite_size", StringType)
+ ProcedureParameter.optional("manifest_sort_max_rewrite_size", StringType),
+ ProcedureParameter.optional("manifest_sort_order", StringType)
};
private static final StructType OUTPUT_TYPE =
@@ -91,6 +92,7 @@ public InternalRow[] call(InternalRow args) {
Boolean manifestSortEnabled = args.isNullAt(3) ? null : args.getBoolean(3);
String manifestSortPartitionField = args.isNullAt(4) ? null : args.getString(4);
String manifestSortMaxRewriteSize = args.isNullAt(5) ? null : args.getString(5);
+ String manifestSortOrder = args.isNullAt(6) ? null : args.getString(6);
Table table = loadSparkTable(tableIdent).getTable();
HashMap dynamicOptions = new HashMap<>();
@@ -107,6 +109,21 @@ public InternalRow[] call(InternalRow args) {
dynamicOptions.put(
CoreOptions.MANIFEST_SORT_MAX_REWRITE_SIZE.key(), manifestSortMaxRewriteSize);
}
+ if (manifestSortOrder != null) {
+ if (Boolean.FALSE.equals(manifestSortEnabled)
+ || "false"
+ .equalsIgnoreCase(
+ dynamicOptions.get(CoreOptions.MANIFEST_SORT_ENABLED.key()))) {
+ throw new IllegalArgumentException(
+ "'manifest_sort_order' cannot be used with 'manifest_sort_enabled=false'.");
+ }
+ CoreOptions.ManifestSortOrder order =
+ CoreOptions.ManifestSortOrder.fromString(manifestSortOrder);
+ dynamicOptions.put(CoreOptions.MANIFEST_SORT_ORDER.key(), order.toString());
+ dynamicOptions.put(CoreOptions.MANIFEST_SORT_ENABLED.key(), Boolean.TRUE.toString());
+ dynamicOptions.put(
+ CoreOptions.MANIFEST_SORT_FORCE_REWRITE.key(), Boolean.TRUE.toString());
+ }
table = table.copy(dynamicOptions);
if (dryRun) {
diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactManifestProcedureTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactManifestProcedureTest.scala
index 76368e6f240c..4dab5d46484d 100644
--- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactManifestProcedureTest.scala
+++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactManifestProcedureTest.scala
@@ -69,7 +69,8 @@ class CompactManifestProcedureTest extends PaimonSparkTestBase with StreamTest {
"dry_run => true, " +
"manifest_sort_enabled => true, " +
"manifest_sort_partition_field => 'dt', " +
- "manifest_sort_max_rewrite_size => '1gb')")
+ "manifest_sort_max_rewrite_size => '1gb', " +
+ "manifest_sort_order => 'partition-first')")
.collectAsList()
Assertions.assertThat(dryRunRows.get(0).getBoolean(0)).isTrue
From 1592745594ecb04fff7d10524384d422904a580a Mon Sep 17 00:00:00 2001
From: mingfeng
Date: Mon, 14 Sep 2026 13:23:39 +0800
Subject: [PATCH 2/6] [core] Restrict forced manifest rewrite to maintenance
---
docs/generated/core_configuration.html | 6 ---
.../java/org/apache/paimon/CoreOptions.java | 1 +
.../operation/ManifestCompactDryRun.java | 13 +++---
.../paimon/schema/SchemaValidation.java | 15 ++++++-
.../paimon/manifest/ManifestFileMetaTest.java | 27 +++++++++++++
.../paimon/schema/SchemaValidationTest.java | 40 +++++++++++++++++++
6 files changed, 89 insertions(+), 13 deletions(-)
diff --git a/docs/generated/core_configuration.html b/docs/generated/core_configuration.html
index 2ee3a3321c16..ac5730017523 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -1065,12 +1065,6 @@
Boolean |
Whether to invoke manifest sort rewrite during commit.
Note: enabling this changes the semantics of 'manifest.merge-min-count'. In the sort rewrite path, small manifest files within the rewrite budget are sorted and merged directly, so the minimum-count gate no longer prevents merging a small number of under-budget manifest files when full compaction is not triggered. |
-
- manifest-sort.force-rewrite |
- false |
- Boolean |
- When 'manifest-sort.enabled' is true, force an explicit manifest compaction to rewrite already compacted manifest runs using the configured manifest sort order. The existing 'manifest-sort.max-rewrite-size' rewrite budget semantics still apply. This should be supplied as a one-shot dynamic option for maintenance, not persisted for routine writes. |
-
manifest-sort.max-rewrite-size |
256 mb |
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index 1a237eb4d63a..8e345dd998f3 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -599,6 +599,7 @@ public InlineElement getDescription() {
+ " skipped. Set to a larger value to allow more aggressive"
+ " sort rewriting. The cap only limits the sorted rewrite portion and full/minor cleanup may still happen beyond it.");
+ @ExcludeFromDocumentation("Only used by compact_manifest maintenance procedure")
public static final ConfigOption MANIFEST_SORT_FORCE_REWRITE =
key("manifest-sort.force-rewrite")
.booleanType()
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestCompactDryRun.java b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestCompactDryRun.java
index aebcc1622d00..3c2abfbdb451 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestCompactDryRun.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestCompactDryRun.java
@@ -110,11 +110,13 @@ private static List buildLevelSortedRunsForDryRun(
RowType partitionType,
CoreOptions options) {
long suggestedMetaSize = options.manifestTargetSize().getBytes();
+ boolean forceRewrite = options.manifestSortForceRewrite();
boolean fullCompaction =
- ManifestFileSorter.reachesFullCompactionThreshold(
- manifests,
- suggestedMetaSize,
- options.manifestFullCompactionThresholdSize().getBytes());
+ forceRewrite
+ || ManifestFileSorter.reachesFullCompactionThreshold(
+ manifests,
+ suggestedMetaSize,
+ options.manifestFullCompactionThresholdSize().getBytes());
ManifestFileSorter.ManifestSortKey sortKey =
ManifestFileSorter.createSortKey(
options.dataEvolutionEnabled(),
@@ -135,7 +137,8 @@ private static List buildLevelSortedRunsForDryRun(
// 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
+ if (!forceRewrite
+ && fullCompaction
&& classifyResult.compactWithoutSort.isEmpty()
&& new ManifestPickStrategy(
options.maxSizeAmplificationPercent(), options.sortedRunSizeRatio())
diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
index 996fceaa97bf..25c7e7b3a2a4 100644
--- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
+++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
@@ -417,7 +417,7 @@ public static void validateTableSchema(TableSchema schema, Set dynamicOp
validatePkClusteringOverride(options);
- validateManifestSort(schema, options);
+ validateManifestSort(schema, options, dynamicOptionKeys);
}
/**
@@ -2019,7 +2019,18 @@ public static void validatePkClusteringOverride(CoreOptions options) {
}
}
- private static void validateManifestSort(TableSchema schema, CoreOptions options) {
+ private static void validateManifestSort(
+ TableSchema schema, CoreOptions options, Set dynamicOptionKeys) {
+ for (ConfigOption option :
+ Arrays.asList(
+ CoreOptions.MANIFEST_SORT_FORCE_REWRITE, CoreOptions.MANIFEST_SORT_ORDER)) {
+ checkArgument(
+ !schema.options().containsKey(option.key())
+ || dynamicOptionKeys.contains(option.key()),
+ "'%s' is only supported as a dynamic option for explicit manifest compaction.",
+ option.key());
+ }
+
if (options.manifestSortEnabled()) {
if (!options.dataEvolutionEnabled()) {
checkArgument(
diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTest.java
index 7d0abc7c7d73..452f96d59d0d 100644
--- a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTest.java
@@ -1554,6 +1554,33 @@ public void testManifestSortForceRewriteAllLevelRuns() {
.containsExactly(0, 1, 2, 3);
}
+ @Test
+ public void testManifestSortForceRewriteDryRunUsesFullCompaction() {
+ List input =
+ Arrays.asList(
+ makeManifest(makeEntry(true, "base", 0)),
+ makeManifest(
+ makeEntry(false, "base", 0), makeEntry(true, "replacement", 0)));
+
+ Options testOptions = new Options();
+ testOptions.set(CoreOptions.MANIFEST_SORT_ENABLED, true);
+ testOptions.set(CoreOptions.MANIFEST_SORT_FORCE_REWRITE, true);
+ testOptions.set(CoreOptions.MANIFEST_TARGET_FILE_SIZE.key(), "1B");
+ testOptions.set(CoreOptions.MANIFEST_FULL_COMPACTION_FILE_SIZE.key(), Long.MAX_VALUE + "B");
+
+ FileStoreTable table = mock(FileStoreTable.class, RETURNS_DEEP_STUBS);
+ Snapshot snapshot = mock(Snapshot.class);
+ when(table.options()).thenReturn(testOptions.toMap());
+ when(table.store().snapshotManager().latestSnapshot()).thenReturn(snapshot);
+ when(table.store().manifestListFactory().create().readDataManifests(snapshot))
+ .thenReturn(input);
+ when(table.store().manifestFileFactory().create()).thenReturn(manifestFile);
+ when(table.schema().logicalPartitionType()).thenReturn(getPartitionType());
+
+ assertThat(ManifestCompactDryRun.execute(table))
+ .endsWith("Manifest sort level files: L0=0, L1=0, L2=0, L3=0, L4=0.");
+ }
+
@Test
public void testManifestSortForceRewriteSingleManifest() {
ManifestFileMeta physical =
diff --git a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
index c9588f4552fa..a891d8c1068d 100644
--- a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
@@ -36,6 +36,7 @@
import java.util.Map;
import static java.util.Collections.emptyList;
+import static java.util.Collections.singleton;
import static java.util.Collections.singletonList;
import static org.apache.paimon.CoreOptions.BUCKET;
import static org.apache.paimon.CoreOptions.DATA_EVOLUTION_ENABLED;
@@ -2049,6 +2050,45 @@ void testManifestSortValidation() {
.hasMessageContaining("is not a partition field");
}
+ @Test
+ void testManifestSortMaintenanceOptionsAreDynamicOnly() {
+ List fields =
+ Arrays.asList(
+ new DataField(0, "f0", DataTypes.INT()),
+ new DataField(1, "f1", DataTypes.INT()));
+ Map baseOptions = new HashMap<>();
+ baseOptions.put(CoreOptions.MANIFEST_SORT_ENABLED.key(), "true");
+ baseOptions.put(BUCKET.key(), "-1");
+
+ Map forceOptions = new HashMap<>(baseOptions);
+ forceOptions.put(CoreOptions.MANIFEST_SORT_FORCE_REWRITE.key(), "true");
+ TableSchema forceSchema =
+ new TableSchema(1, fields, 10, singletonList("f0"), emptyList(), forceOptions, "");
+ assertThatThrownBy(() -> validateTableSchema(forceSchema))
+ .hasMessage(
+ "'manifest-sort.force-rewrite' is only supported as a dynamic option for explicit manifest compaction.");
+ assertThatNoException()
+ .isThrownBy(
+ () ->
+ validateTableSchema(
+ forceSchema,
+ singleton(CoreOptions.MANIFEST_SORT_FORCE_REWRITE.key())));
+
+ Map orderOptions = new HashMap<>(baseOptions);
+ orderOptions.put(CoreOptions.MANIFEST_SORT_ORDER.key(), "partition-first");
+ TableSchema orderSchema =
+ new TableSchema(1, fields, 10, singletonList("f0"), emptyList(), orderOptions, "");
+ assertThatThrownBy(() -> validateTableSchema(orderSchema))
+ .hasMessage(
+ "'manifest-sort.order' is only supported as a dynamic option for explicit manifest compaction.");
+ assertThatNoException()
+ .isThrownBy(
+ () ->
+ validateTableSchema(
+ orderSchema,
+ singleton(CoreOptions.MANIFEST_SORT_ORDER.key())));
+ }
+
@Test
public void testMergeOnReadCoexistsWithVisibilityCallback() {
Map options = new HashMap<>();
From dfcb41002e37e9e2d2904c34f7ff8fd0047f8a09 Mon Sep 17 00:00:00 2001
From: mingfeng
Date: Mon, 14 Sep 2026 13:55:33 +0800
Subject: [PATCH 3/6] [core] Validate explicit manifest sort order eagerly
---
.../paimon/schema/SchemaValidation.java | 10 +++++
.../paimon/schema/SchemaValidationTest.java | 42 +++++++++++++++++++
.../CompactManifestProcedureITCase.java | 9 ++++
3 files changed, 61 insertions(+)
diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
index 25c7e7b3a2a4..72cb80f526b1 100644
--- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
+++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
@@ -2031,6 +2031,16 @@ private static void validateManifestSort(
option.key());
}
+ CoreOptions.ManifestSortOrder sortOrder = options.manifestSortOrder();
+ checkArgument(
+ sortOrder == null || !options.dataEvolutionEnabled(),
+ "Explicit manifest sort order is not supported for data evolution tables.");
+ checkArgument(
+ sortOrder != CoreOptions.ManifestSortOrder.BUCKET_FIRST
+ || options.bucket() > 0
+ || options.bucket() == BucketMode.POSTPONE_BUCKET,
+ "Manifest sort order 'bucket-first' requires a bucketed table.");
+
if (options.manifestSortEnabled()) {
if (!options.dataEvolutionEnabled()) {
checkArgument(
diff --git a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
index a891d8c1068d..a1c31c24bbfb 100644
--- a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
@@ -2089,6 +2089,48 @@ void testManifestSortMaintenanceOptionsAreDynamicOnly() {
singleton(CoreOptions.MANIFEST_SORT_ORDER.key())));
}
+ @Test
+ void testManifestSortMaintenanceOrderValidation() {
+ List fields =
+ Arrays.asList(
+ new DataField(0, "f0", DataTypes.INT()),
+ new DataField(1, "f1", DataTypes.INT()));
+ Map options = new HashMap<>();
+ options.put(CoreOptions.MANIFEST_SORT_ENABLED.key(), "true");
+ options.put(CoreOptions.MANIFEST_SORT_ORDER.key(), "bucket-first");
+ options.put(BUCKET.key(), "-1");
+
+ TableSchema schema =
+ new TableSchema(1, fields, 10, singletonList("f0"), emptyList(), options, "");
+ assertThatThrownBy(
+ () ->
+ validateTableSchema(
+ schema, singleton(CoreOptions.MANIFEST_SORT_ORDER.key())))
+ .hasMessage("Manifest sort order 'bucket-first' requires a bucketed table.");
+
+ options.put(BUCKET.key(), "4");
+ options.put(CoreOptions.BUCKET_KEY.key(), "f1");
+ assertThatNoException()
+ .isThrownBy(
+ () ->
+ validateTableSchema(
+ schema.copy(options),
+ singleton(CoreOptions.MANIFEST_SORT_ORDER.key())));
+
+ options.put(CoreOptions.MANIFEST_SORT_ORDER.key(), "partition-first");
+ options.put(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true");
+ options.put(CoreOptions.ROW_TRACKING_ENABLED.key(), "true");
+ options.put(BUCKET.key(), "-1");
+ options.remove(CoreOptions.BUCKET_KEY.key());
+ assertThatThrownBy(
+ () ->
+ validateTableSchema(
+ schema.copy(options),
+ singleton(CoreOptions.MANIFEST_SORT_ORDER.key())))
+ .hasMessage(
+ "Explicit manifest sort order is not supported for data evolution tables.");
+ }
+
@Test
public void testMergeOnReadCoexistsWithVisibilityCallback() {
Map options = new HashMap<>();
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CompactManifestProcedureITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CompactManifestProcedureITCase.java
index cef9509018f1..4e892aab6ae9 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CompactManifestProcedureITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CompactManifestProcedureITCase.java
@@ -179,6 +179,15 @@ public void testManifestSortParametersValidation() {
+ "`manifest_sort_order` => 'partition-first')"))
.hasStackTraceContaining(
"'manifest_sort_order' cannot be used with 'manifest_sort_enabled=false'.");
+
+ Assertions.assertThatThrownBy(
+ () ->
+ sql(
+ "CALL sys.compact_manifest("
+ + "`table` => 'default.T_INVALID', "
+ + "`manifest_sort_order` => 'bucket-first')"))
+ .hasStackTraceContaining(
+ "Manifest sort order 'bucket-first' requires a bucketed table.");
}
@Test
From 0c23e13f85b86ba9a6ffe3fb6b685a99c0c6b4eb Mon Sep 17 00:00:00 2001
From: mingfeng
Date: Mon, 14 Sep 2026 15:21:00 +0800
Subject: [PATCH 4/6] [core] Respect table options for forced manifest rewrite
---
docs/docs/flink/procedures/compaction.md | 19 +--
docs/docs/spark/procedures/maintenance.md | 22 +--
.../java/org/apache/paimon/CoreOptions.java | 51 +------
.../operation/ManifestCompactDryRun.java | 1 -
.../paimon/operation/ManifestFileSorter.java | 72 ++-------
.../paimon/schema/SchemaValidation.java | 23 +--
.../paimon/manifest/ManifestFileMetaTest.java | 143 ++++++++----------
.../paimon/schema/SchemaValidationTest.java | 58 +------
.../flink/action/CompactManifestAction.java | 3 +-
.../procedure/CompactManifestProcedure.java | 22 +--
.../CompactManifestProcedureITCase.java | 37 +----
.../procedure/CompactManifestProcedure.java | 19 +--
.../CompactManifestProcedureTest.scala | 3 +-
13 files changed, 101 insertions(+), 372 deletions(-)
diff --git a/docs/docs/flink/procedures/compaction.md b/docs/docs/flink/procedures/compaction.md
index ae8727594eec..22090db69850 100644
--- a/docs/docs/flink/procedures/compaction.md
+++ b/docs/docs/flink/procedures/compaction.md
@@ -250,11 +250,7 @@ To compact_manifest the manifests. Arguments:
- manifest_sort_max_rewrite_size (String, optional): maximum manifest size rewritten by one sort pass.
-- manifest_sort_order (String, optional): target layout for a one-shot rewrite. Supported values are `bucket-first` and `partition-first`. Setting it enables manifest sort and forces existing manifests to be rewritten. `bucket-first` requires a bucketed table, and explicit sort orders are not supported for data evolution tables.
-
-When `manifest_sort_order` is omitted, the existing layout selection remains unchanged: bucketed tables use bucket-first, non-bucket tables use partition-first, and data evolution tables use RowID sorting when RowID metadata is available.
-
-Set `manifest-sort.force-rewrite=true` in `options` together with `manifest_sort_enabled=true` to rewrite already compacted manifest runs using the current sort order. Use it only as a one-shot dynamic option. The existing `manifest_sort_max_rewrite_size` rewrite budget semantics still apply; raise it to migrate more manifests in one invocation.
+Set `manifest-sort.force-rewrite=true` in `options` together with `manifest_sort_enabled=true` to rewrite already compacted manifest runs using the layout selected from the table options. Use it only as a one-shot dynamic option. The existing `manifest_sort_max_rewrite_size` rewrite budget semantics still apply; raise it to migrate more manifests in one invocation.
**Syntax**
@@ -269,8 +265,7 @@ CALL [catalog.]sys.compact_manifest(
`table` => 'identifier',
`manifest_sort_enabled` => true,
`manifest_sort_partition_field` => 'dt',
- `manifest_sort_max_rewrite_size` => '1 gb',
- `manifest_sort_order` => 'partition-first'
+ `manifest_sort_max_rewrite_size` => '1 gb'
);
```
@@ -290,14 +285,8 @@ CALL sys.compact_manifest(
CALL sys.compact_manifest(
`table` => 'default.T',
- `manifest_sort_order` => 'partition-first',
- `manifest_sort_max_rewrite_size` => '1 gb'
-);
-
--- Switch the same bucketed table back to bucket-first layout.
-CALL sys.compact_manifest(
- `table` => 'default.T',
- `manifest_sort_order` => 'bucket-first',
+ `options` => 'manifest-sort.force-rewrite=true',
+ `manifest_sort_enabled` => true,
`manifest_sort_max_rewrite_size` => '1 gb'
);
```
diff --git a/docs/docs/spark/procedures/maintenance.md b/docs/docs/spark/procedures/maintenance.md
index e636a9f03569..9cabe8571323 100644
--- a/docs/docs/spark/procedures/maintenance.md
+++ b/docs/docs/spark/procedures/maintenance.md
@@ -126,16 +126,10 @@ 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.
-- `manifest_sort_order` (`STRING`, optional): target layout for a one-shot rewrite. Supported values are `bucket-first` and `partition-first`. Setting it enables manifest sort and forces existing manifests to be rewritten. `bucket-first` requires a bucketed table, and explicit sort orders are not supported for data evolution tables.
-
-When `manifest_sort_order` is omitted, the existing layout selection remains unchanged: bucketed
-tables use bucket-first, non-bucket tables use partition-first, and data evolution tables use RowID
-sorting when RowID metadata is available.
-
Set `manifest-sort.force-rewrite=true` in `options` together with `manifest_sort_enabled=true` to
-rewrite already compacted manifest runs using the current sort order. Use it only as a one-shot
-dynamic option. The existing `manifest_sort_max_rewrite_size` rewrite budget semantics still apply;
-raise it to migrate more manifests in one invocation.
+rewrite already compacted manifest runs using the layout selected from the table options. Use it
+only as a one-shot dynamic option. The existing `manifest_sort_max_rewrite_size` rewrite budget
+semantics still apply; raise it to migrate more manifests in one invocation.
```sql
CALL sys.compact_manifest(`table` => 'default.T');
@@ -151,14 +145,8 @@ CALL sys.compact_manifest(
CALL sys.compact_manifest(
`table` => 'default.T',
- manifest_sort_order => 'partition-first',
- manifest_sort_max_rewrite_size => '1 gb'
-);
-
--- Switch the same bucketed table back to bucket-first layout.
-CALL sys.compact_manifest(
- `table` => 'default.T',
- manifest_sort_order => 'bucket-first',
+ options => 'manifest-sort.force-rewrite=true',
+ manifest_sort_enabled => true,
manifest_sort_max_rewrite_size => '1 gb'
);
```
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index 8e345dd998f3..f6f9892670f4 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -607,20 +607,12 @@ public InlineElement getDescription() {
.withDescription(
"When 'manifest-sort.enabled' is true, force an explicit manifest"
+ " compaction to rewrite already compacted manifest runs using"
- + " the configured manifest sort order."
+ + " the layout selected from the table options."
+ " The existing 'manifest-sort.max-rewrite-size' rewrite budget"
+ " semantics still apply."
+ " This should be supplied as a one-shot dynamic option for"
+ " maintenance, not persisted for routine writes.");
- @ExcludeFromDocumentation("Only used by compact_manifest maintenance procedure")
- public static final ConfigOption MANIFEST_SORT_ORDER =
- key("manifest-sort.order")
- .enumType(ManifestSortOrder.class)
- .noDefaultValue()
- .withDescription(
- "Target manifest layout for a one-shot manifest sort rewrite.");
-
public static final ConfigOption MANIFEST_MERGE_OPTIMIZE_ENABLED =
key("manifest.merge-optimize.enabled")
.booleanType()
@@ -3253,11 +3245,6 @@ public boolean manifestSortForceRewrite() {
return options.get(MANIFEST_SORT_FORCE_REWRITE);
}
- @Nullable
- public ManifestSortOrder manifestSortOrder() {
- return options.getOptional(MANIFEST_SORT_ORDER).orElse(null);
- }
-
public boolean manifestMergeOptimizeEnabled() {
return options.get(MANIFEST_MERGE_OPTIMIZE_ENABLED);
}
@@ -5462,42 +5449,6 @@ public static List> getOptions() {
})
.collect(Collectors.toSet());
- /** Target layout for an explicit manifest sort rewrite. */
- public enum ManifestSortOrder implements DescribedEnum {
- BUCKET_FIRST("bucket-first", "Sort manifest entries by bucket before partition."),
- PARTITION_FIRST("partition-first", "Sort manifest entries by partition.");
-
- private final String value;
- private final String description;
-
- ManifestSortOrder(String value, String description) {
- this.value = value;
- this.description = description;
- }
-
- public static ManifestSortOrder fromString(String value) {
- for (ManifestSortOrder order : values()) {
- if (order.value.equalsIgnoreCase(value.trim())) {
- return order;
- }
- }
- throw new IllegalArgumentException(
- String.format(
- "Unsupported manifest sort order '%s'. Supported values are 'bucket-first' and 'partition-first'.",
- value));
- }
-
- @Override
- public String toString() {
- return value;
- }
-
- @Override
- public InlineElement getDescription() {
- return text(description);
- }
- }
-
/** Specifies the sort engine for table with primary key. */
public enum SortEngine implements DescribedEnum {
MIN_HEAP("min-heap", "Use min-heap for multiway sorting."),
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestCompactDryRun.java b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestCompactDryRun.java
index 3c2abfbdb451..480871a4a6da 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestCompactDryRun.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestCompactDryRun.java
@@ -123,7 +123,6 @@ private static List buildLevelSortedRunsForDryRun(
manifests,
options.manifestSortPartitionField(),
partitionType,
- options.manifestSortOrder(),
options.bucket() > 0 || options.bucket() == BucketMode.POSTPONE_BUCKET);
ManifestFileSorter.ClassifyResult classifyResult =
ManifestFileSorter.classifyManifests(
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileSorter.java b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileSorter.java
index 0f1250696829..f6981de56577 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileSorter.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileSorter.java
@@ -159,7 +159,6 @@ static List trySortCompaction(
@Nullable IOManager ioManager)
throws Exception {
String sortPartitionField = options.manifestSortPartitionField();
- CoreOptions.ManifestSortOrder sortOrder = options.manifestSortOrder();
boolean bucketed = options.bucket() > 0 || options.bucket() == BucketMode.POSTPONE_BUCKET;
boolean runMergeOptimizeEnabled = options.manifestMergeOptimizeEnabled();
long suggestedMetaSize = options.manifestTargetSize().getBytes();
@@ -180,7 +179,6 @@ static List trySortCompaction(
manifestFile,
partitionType,
sortPartitionField,
- sortOrder,
bucketed,
options.dataEvolutionEnabled(),
runMergeOptimizeEnabled,
@@ -202,7 +200,6 @@ static List trySortCompaction(
manifestFile,
partitionType,
sortPartitionField,
- sortOrder,
bucketed,
options.dataEvolutionEnabled(),
runMergeOptimizeEnabled,
@@ -227,7 +224,6 @@ private static Optional> tryFullCompaction(
ManifestFile manifestFile,
RowType partitionType,
String sortPartitionField,
- @Nullable CoreOptions.ManifestSortOrder sortOrder,
boolean bucketed,
boolean dataEvolutionEnabled,
boolean runMergeOptimizeEnabled,
@@ -256,7 +252,6 @@ private static Optional> tryFullCompaction(
manifestFile,
partitionType,
sortPartitionField,
- sortOrder,
bucketed,
dataEvolutionEnabled,
runMergeOptimizeEnabled,
@@ -302,7 +297,7 @@ private static Optional> tryFullCompaction(
// Step 4: Split into sections and merge small adjacent sections. A forced rewrite
// intentionally uses one global section so entries from different already-compacted
- // manifests can be clustered using the current sort order.
+ // manifests can be clustered using the layout selected from the table options.
List sections;
if (forceRewrite) {
long totalSize = 0L;
@@ -358,7 +353,6 @@ private static List tryMinorCompaction(
ManifestFile manifestFile,
RowType partitionType,
String sortPartitionField,
- @Nullable CoreOptions.ManifestSortOrder sortOrder,
boolean bucketed,
boolean dataEvolutionEnabled,
boolean runMergeOptimizeEnabled,
@@ -379,7 +373,6 @@ private static List tryMinorCompaction(
manifestFile,
partitionType,
sortPartitionField,
- sortOrder,
bucketed,
dataEvolutionEnabled,
runMergeOptimizeEnabled,
@@ -496,7 +489,6 @@ private static CompactionContext prepareCompaction(
ManifestFile manifestFile,
RowType partitionType,
String sortPartitionField,
- @Nullable CoreOptions.ManifestSortOrder sortOrder,
boolean bucketed,
boolean dataEvolutionEnabled,
boolean runMergeOptimizeEnabled,
@@ -511,12 +503,7 @@ private static CompactionContext prepareCompaction(
// Step 1: Resolve sort key. Data evolution tables prefer RowID ranges when available.
ManifestSortKey sortKey =
createSortKey(
- dataEvolutionEnabled,
- input,
- sortPartitionField,
- partitionType,
- sortOrder,
- bucketed);
+ dataEvolutionEnabled, input, sortPartitionField, partitionType, bucketed);
// Step 2: Classify manifests into LSM files and collect delete entries.
ClassifyResult classification =
@@ -904,18 +891,6 @@ 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) {
- rewriteSection(
- section.files,
- output,
- sortNewFiles,
- ctx,
- manifestFile,
- manifestReadParallelism);
- continue;
- }
-
// Phase 1: budget not yet exhausted -- perform aggressive sort rewrite.
if (!budgetExhausted) {
// Phase 1a: section fits within the remaining budget -- sort and rewrite it
@@ -928,7 +903,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
@@ -1008,7 +984,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;
@@ -1086,7 +1063,8 @@ private static void unsortedCompactSection(
sortNewFiles,
ctx,
manifestFile,
- manifestReadParallelism);
+ manifestReadParallelism,
+ false);
candidates.clear();
candidatesSize = 0;
}
@@ -1100,7 +1078,8 @@ private static void unsortedCompactSection(
sortNewFiles,
ctx,
manifestFile,
- manifestReadParallelism);
+ manifestReadParallelism,
+ false);
} else {
output.addAllUnchanged(candidates);
}
@@ -1119,11 +1098,12 @@ private static void rewriteSection(
List sortNewFiles,
CompactionContext ctx,
ManifestFile manifestFile,
- @Nullable Integer manifestReadParallelism)
+ @Nullable Integer manifestReadParallelism,
+ boolean allowForceRewrite)
throws Exception {
// Skip rewrite for single file not in delete-range.
if (section.size() == 1
- && !ctx.forceRewrite
+ && !(allowForceRewrite && ctx.forceRewrite)
&& !ctx.defaultCompactFiles.getOrDefault(section.get(0), false)) {
output.addUnchanged(section.get(0));
return;
@@ -1239,27 +1219,6 @@ static ManifestSortKey createSortKey(
String sortPartitionField,
RowType partitionType,
boolean bucketed) {
- return createSortKey(
- dataEvolutionEnabled, input, sortPartitionField, partitionType, null, bucketed);
- }
-
- static ManifestSortKey createSortKey(
- boolean dataEvolutionEnabled,
- List input,
- String sortPartitionField,
- RowType partitionType,
- @Nullable CoreOptions.ManifestSortOrder sortOrder,
- boolean bucketed) {
- if (sortOrder != null && dataEvolutionEnabled) {
- throw new IllegalArgumentException(
- "Explicit manifest sort order is not supported for data evolution tables.");
- }
-
- if (sortOrder == CoreOptions.ManifestSortOrder.BUCKET_FIRST && !bucketed) {
- throw new IllegalArgumentException(
- "Manifest sort order 'bucket-first' requires a bucketed table.");
- }
-
boolean rowIdSort = dataEvolutionEnabled && ManifestFileMeta.allContainsRowId(input);
if (rowIdSort) {
// RowID sorting uses the configured partition field as the primary key when specified,
@@ -1289,10 +1248,7 @@ static ManifestSortKey createSortKey(
RecordComparator fieldComparator =
CodeGenUtils.newRecordComparator(
partitionType.getFieldTypes(), new int[] {sortFieldIndex});
- boolean useBucketSort =
- sortOrder == CoreOptions.ManifestSortOrder.BUCKET_FIRST
- || (sortOrder == null && bucketed);
- if (useBucketSort) {
+ if (bucketed) {
boolean compareManifestBuckets =
input.stream()
.allMatch(meta -> meta.minBucket() != null && meta.maxBucket() != null);
diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
index 72cb80f526b1..8144112db6c5 100644
--- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
+++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
@@ -2021,25 +2021,12 @@ public static void validatePkClusteringOverride(CoreOptions options) {
private static void validateManifestSort(
TableSchema schema, CoreOptions options, Set dynamicOptionKeys) {
- for (ConfigOption option :
- Arrays.asList(
- CoreOptions.MANIFEST_SORT_FORCE_REWRITE, CoreOptions.MANIFEST_SORT_ORDER)) {
- checkArgument(
- !schema.options().containsKey(option.key())
- || dynamicOptionKeys.contains(option.key()),
- "'%s' is only supported as a dynamic option for explicit manifest compaction.",
- option.key());
- }
-
- CoreOptions.ManifestSortOrder sortOrder = options.manifestSortOrder();
- checkArgument(
- sortOrder == null || !options.dataEvolutionEnabled(),
- "Explicit manifest sort order is not supported for data evolution tables.");
checkArgument(
- sortOrder != CoreOptions.ManifestSortOrder.BUCKET_FIRST
- || options.bucket() > 0
- || options.bucket() == BucketMode.POSTPONE_BUCKET,
- "Manifest sort order 'bucket-first' requires a bucketed table.");
+ !schema.options().containsKey(CoreOptions.MANIFEST_SORT_FORCE_REWRITE.key())
+ || dynamicOptionKeys.contains(
+ CoreOptions.MANIFEST_SORT_FORCE_REWRITE.key()),
+ "'%s' is only supported as a dynamic option for explicit manifest compaction.",
+ CoreOptions.MANIFEST_SORT_FORCE_REWRITE.key());
if (options.manifestSortEnabled()) {
if (!options.dataEvolutionEnabled()) {
diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTest.java
index 452f96d59d0d..0b6a34ba57a4 100644
--- a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTest.java
@@ -1416,86 +1416,6 @@ public void testManifestSortForceRewriteAlreadyCompactedRuns() {
assertThat(afterMigration).containsExactlyElementsOf(rewritten);
}
- @Test
- public void testManifestSortForceRewriteSwitchesLayout() {
- List input =
- Arrays.asList(
- makeManifest(
- makeBucketEntry("a-p0-b3", 0, 3), makeBucketEntry("a-p1-b1", 1, 1)),
- makeManifest(
- makeBucketEntry("b-p0-b2", 0, 2),
- makeBucketEntry("b-p1-b0", 1, 0)));
-
- Options testOptions = new Options();
- testOptions.set(CoreOptions.MANIFEST_SORT_ENABLED, true);
- testOptions.set(CoreOptions.MANIFEST_SORT_FORCE_REWRITE, true);
- testOptions.set(CoreOptions.MANIFEST_SORT_MAX_REWRITE_SIZE.key(), "1G");
- testOptions.set(CoreOptions.BUCKET, 4);
- testOptions.set(
- CoreOptions.MANIFEST_SORT_ORDER, CoreOptions.ManifestSortOrder.PARTITION_FIRST);
-
- List partitionFirst =
- ManifestFileMerger.merge(
- input,
- manifestFile,
- getPartitionType(),
- CoreOptions.fromMap(testOptions.toMap()));
- assertEquivalentEntries(input, partitionFirst);
- assertThat(readEntries(partitionFirst))
- .extracting(entry -> entry.partition().getInt(0))
- .containsExactly(0, 0, 1, 1);
-
- testOptions.set(
- CoreOptions.MANIFEST_SORT_ORDER, CoreOptions.ManifestSortOrder.BUCKET_FIRST);
- List bucketFirst =
- ManifestFileMerger.merge(
- partitionFirst,
- manifestFile,
- getPartitionType(),
- CoreOptions.fromMap(testOptions.toMap()));
- assertEquivalentEntries(partitionFirst, bucketFirst);
- assertThat(readEntries(bucketFirst))
- .extracting(ManifestEntry::bucket)
- .containsExactly(0, 1, 2, 3);
- }
-
- @Test
- public void testManifestSortExplicitOrderValidation() {
- List input =
- Collections.singletonList(makeManifest(makeBucketEntry("file", 0, 0)));
-
- Options testOptions = new Options();
- testOptions.set(CoreOptions.MANIFEST_SORT_ENABLED, true);
- testOptions.set(CoreOptions.MANIFEST_SORT_FORCE_REWRITE, true);
- testOptions.set(
- CoreOptions.MANIFEST_SORT_ORDER, CoreOptions.ManifestSortOrder.BUCKET_FIRST);
- assertThat(
- assertThrows(
- IllegalArgumentException.class,
- () ->
- ManifestFileMerger.merge(
- input,
- manifestFile,
- getPartitionType(),
- CoreOptions.fromMap(testOptions.toMap()))))
- .hasMessage("Manifest sort order 'bucket-first' requires a bucketed table.");
-
- testOptions.set(
- CoreOptions.MANIFEST_SORT_ORDER, CoreOptions.ManifestSortOrder.PARTITION_FIRST);
- testOptions.set(CoreOptions.DATA_EVOLUTION_ENABLED, true);
- assertThat(
- assertThrows(
- IllegalArgumentException.class,
- () ->
- ManifestFileMerger.merge(
- input,
- manifestFile,
- getPartitionType(),
- CoreOptions.fromMap(testOptions.toMap()))))
- .hasMessage(
- "Explicit manifest sort order is not supported for data evolution tables.");
- }
-
@Test
public void testManifestSortForceRewriteAllLevelRuns() {
List physical =
@@ -1637,6 +1557,69 @@ public void testManifestSortForceRewriteRespectsRewriteBudget() {
assertEquivalentEntries(input, rewritten);
}
+ @Test
+ public void testManifestSortForceRewriteDoesNotExceedBudgetForSingletonTail() {
+ long targetSize = CoreOptions.MANIFEST_TARGET_FILE_SIZE.defaultValue().getBytes();
+ List input = new ArrayList<>();
+ for (int i = 0; i < 3; i++) {
+ input.add(
+ copyWithFileSize(makeManifest(makeBucketEntry("file-" + i, 0, i)), targetSize));
+ }
+
+ Options testOptions = new Options();
+ testOptions.set(CoreOptions.MANIFEST_SORT_ENABLED, true);
+ testOptions.set(CoreOptions.MANIFEST_SORT_FORCE_REWRITE, true);
+ testOptions.set(CoreOptions.MANIFEST_SORT_MAX_REWRITE_SIZE.key(), "1B");
+ testOptions.set(CoreOptions.BUCKET, 4);
+ List rewritten =
+ ManifestFileMerger.merge(
+ input,
+ manifestFile,
+ getPartitionType(),
+ CoreOptions.fromMap(testOptions.toMap()));
+
+ Set inputNames =
+ input.stream().map(ManifestFileMeta::fileName).collect(Collectors.toSet());
+ assertThat(rewritten)
+ .extracting(ManifestFileMeta::fileName)
+ .filteredOn(inputNames::contains)
+ .hasSize(1);
+ assertEquivalentEntries(input, rewritten);
+ }
+
+ @Test
+ public void testManifestSortForceRewriteDoesNotRewriteTailBeyondBudget() {
+ long targetSize = CoreOptions.MANIFEST_TARGET_FILE_SIZE.defaultValue().getBytes();
+ List input = new ArrayList<>();
+ for (int i = 0; i < 4; i++) {
+ input.add(
+ copyWithFileSize(makeManifest(makeBucketEntry("file-" + i, 0, i)), targetSize));
+ }
+ input.add(
+ copyWithFileSize(
+ makeManifest(makeBucketEntry("small-file", 0, 4)), targetSize - 1));
+
+ Options testOptions = new Options();
+ testOptions.set(CoreOptions.MANIFEST_SORT_ENABLED, true);
+ testOptions.set(CoreOptions.MANIFEST_SORT_FORCE_REWRITE, true);
+ testOptions.set(CoreOptions.MANIFEST_SORT_MAX_REWRITE_SIZE.key(), "1B");
+ testOptions.set(CoreOptions.BUCKET, 8);
+ List rewritten =
+ ManifestFileMerger.merge(
+ input,
+ manifestFile,
+ getPartitionType(),
+ CoreOptions.fromMap(testOptions.toMap()));
+
+ Set inputNames =
+ input.stream().map(ManifestFileMeta::fileName).collect(Collectors.toSet());
+ assertThat(rewritten)
+ .extracting(ManifestFileMeta::fileName)
+ .filteredOn(inputNames::contains)
+ .hasSize(3);
+ assertEquivalentEntries(input, rewritten);
+ }
+
@ParameterizedTest
@ValueSource(ints = {-1, 4, -2})
public void testManifestSortDryRunUsesBucketRangesForBucketedTable(int bucket) {
diff --git a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
index a1c31c24bbfb..f66d623a0e23 100644
--- a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
@@ -2051,7 +2051,7 @@ void testManifestSortValidation() {
}
@Test
- void testManifestSortMaintenanceOptionsAreDynamicOnly() {
+ void testManifestSortForceRewriteIsDynamicOnly() {
List fields =
Arrays.asList(
new DataField(0, "f0", DataTypes.INT()),
@@ -2073,62 +2073,6 @@ void testManifestSortMaintenanceOptionsAreDynamicOnly() {
validateTableSchema(
forceSchema,
singleton(CoreOptions.MANIFEST_SORT_FORCE_REWRITE.key())));
-
- Map orderOptions = new HashMap<>(baseOptions);
- orderOptions.put(CoreOptions.MANIFEST_SORT_ORDER.key(), "partition-first");
- TableSchema orderSchema =
- new TableSchema(1, fields, 10, singletonList("f0"), emptyList(), orderOptions, "");
- assertThatThrownBy(() -> validateTableSchema(orderSchema))
- .hasMessage(
- "'manifest-sort.order' is only supported as a dynamic option for explicit manifest compaction.");
- assertThatNoException()
- .isThrownBy(
- () ->
- validateTableSchema(
- orderSchema,
- singleton(CoreOptions.MANIFEST_SORT_ORDER.key())));
- }
-
- @Test
- void testManifestSortMaintenanceOrderValidation() {
- List fields =
- Arrays.asList(
- new DataField(0, "f0", DataTypes.INT()),
- new DataField(1, "f1", DataTypes.INT()));
- Map options = new HashMap<>();
- options.put(CoreOptions.MANIFEST_SORT_ENABLED.key(), "true");
- options.put(CoreOptions.MANIFEST_SORT_ORDER.key(), "bucket-first");
- options.put(BUCKET.key(), "-1");
-
- TableSchema schema =
- new TableSchema(1, fields, 10, singletonList("f0"), emptyList(), options, "");
- assertThatThrownBy(
- () ->
- validateTableSchema(
- schema, singleton(CoreOptions.MANIFEST_SORT_ORDER.key())))
- .hasMessage("Manifest sort order 'bucket-first' requires a bucketed table.");
-
- options.put(BUCKET.key(), "4");
- options.put(CoreOptions.BUCKET_KEY.key(), "f1");
- assertThatNoException()
- .isThrownBy(
- () ->
- validateTableSchema(
- schema.copy(options),
- singleton(CoreOptions.MANIFEST_SORT_ORDER.key())));
-
- options.put(CoreOptions.MANIFEST_SORT_ORDER.key(), "partition-first");
- options.put(CoreOptions.DATA_EVOLUTION_ENABLED.key(), "true");
- options.put(CoreOptions.ROW_TRACKING_ENABLED.key(), "true");
- options.put(BUCKET.key(), "-1");
- options.remove(CoreOptions.BUCKET_KEY.key());
- assertThatThrownBy(
- () ->
- validateTableSchema(
- schema.copy(options),
- singleton(CoreOptions.MANIFEST_SORT_ORDER.key())))
- .hasMessage(
- "Explicit manifest sort order is not supported for data evolution tables.");
}
@Test
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactManifestAction.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactManifestAction.java
index 7c0c123339eb..4b70de86dbf3 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactManifestAction.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactManifestAction.java
@@ -69,8 +69,7 @@ public void executeLocally() throws Exception {
dryRun,
manifestSortEnabled,
manifestSortPartitionField,
- manifestSortMaxRewriteSize,
- null);
+ manifestSortMaxRewriteSize);
for (String result : results) {
LOG.info(result);
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CompactManifestProcedure.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CompactManifestProcedure.java
index a12c9abcdb8b..1bae106b9f6b 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CompactManifestProcedure.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CompactManifestProcedure.java
@@ -59,10 +59,6 @@ public String identifier() {
@ArgumentHint(
name = "manifest_sort_max_rewrite_size",
type = @DataTypeHint("STRING"),
- isOptional = true),
- @ArgumentHint(
- name = "manifest_sort_order",
- type = @DataTypeHint("STRING"),
isOptional = true)
})
public String[] call(
@@ -72,8 +68,7 @@ public String[] call(
@Nullable Boolean dryRun,
@Nullable Boolean manifestSortEnabled,
@Nullable String manifestSortPartitionField,
- @Nullable String manifestSortMaxRewriteSize,
- @Nullable String manifestSortOrder)
+ @Nullable String manifestSortMaxRewriteSize)
throws Exception {
FileStoreTable table = (FileStoreTable) table(tableId);
@@ -93,21 +88,6 @@ public String[] call(
dynamicOptions.put(
CoreOptions.MANIFEST_SORT_MAX_REWRITE_SIZE.key(), manifestSortMaxRewriteSize);
}
- if (manifestSortOrder != null) {
- if (Boolean.FALSE.equals(manifestSortEnabled)
- || "false"
- .equalsIgnoreCase(
- dynamicOptions.get(CoreOptions.MANIFEST_SORT_ENABLED.key()))) {
- throw new IllegalArgumentException(
- "'manifest_sort_order' cannot be used with 'manifest_sort_enabled=false'.");
- }
- CoreOptions.ManifestSortOrder order =
- CoreOptions.ManifestSortOrder.fromString(manifestSortOrder);
- dynamicOptions.put(CoreOptions.MANIFEST_SORT_ORDER.key(), order.toString());
- dynamicOptions.put(CoreOptions.MANIFEST_SORT_ENABLED.key(), Boolean.TRUE.toString());
- dynamicOptions.put(
- CoreOptions.MANIFEST_SORT_FORCE_REWRITE.key(), Boolean.TRUE.toString());
- }
table = table.copy(dynamicOptions);
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CompactManifestProcedureITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CompactManifestProcedureITCase.java
index 4e892aab6ae9..213e6b73afb4 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CompactManifestProcedureITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CompactManifestProcedureITCase.java
@@ -128,16 +128,15 @@ public void testManifestSortParameters() throws Exception {
String forceRewriteProcedure =
"CALL sys.compact_manifest("
+ "`table` => 'default.T_SORT', "
+ + "`options` => 'manifest-sort.force-rewrite=true', "
+ + "`manifest_sort_enabled` => true, "
+ "`manifest_sort_partition_field` => 'dt', "
- + "`manifest_sort_max_rewrite_size` => '1 gb', "
- + "`manifest_sort_order` => 'partition-first')";
+ + "`manifest_sort_max_rewrite_size` => '1 gb')";
sql(forceRewriteProcedure);
long forceRewriteSnapshotId = table.snapshotManager().latestSnapshot().id();
Assertions.assertThat(forceRewriteSnapshotId).isEqualTo(compactSnapshotId + 1);
Assertions.assertThat(paimonTable("T_SORT").options())
- .doesNotContainKeys(
- CoreOptions.MANIFEST_SORT_FORCE_REWRITE.key(),
- CoreOptions.MANIFEST_SORT_ORDER.key());
+ .doesNotContainKey(CoreOptions.MANIFEST_SORT_FORCE_REWRITE.key());
sql(procedure);
Assertions.assertThat(table.snapshotManager().latestSnapshot().id())
@@ -160,34 +159,6 @@ public void testManifestSortParametersValidation() {
+ "`manifest_sort_partition_field` => 'missing')"))
.hasStackTraceContaining(
"'manifest-sort.partition-field' = 'missing' is not a partition field");
-
- Assertions.assertThatThrownBy(
- () ->
- sql(
- "CALL sys.compact_manifest("
- + "`table` => 'default.T_INVALID', "
- + "`manifest_sort_order` => 'unknown')"))
- .hasStackTraceContaining(
- "Unsupported manifest sort order 'unknown'. Supported values are 'bucket-first' and 'partition-first'.");
-
- Assertions.assertThatThrownBy(
- () ->
- sql(
- "CALL sys.compact_manifest("
- + "`table` => 'default.T_INVALID', "
- + "`manifest_sort_enabled` => false, "
- + "`manifest_sort_order` => 'partition-first')"))
- .hasStackTraceContaining(
- "'manifest_sort_order' cannot be used with 'manifest_sort_enabled=false'.");
-
- Assertions.assertThatThrownBy(
- () ->
- sql(
- "CALL sys.compact_manifest("
- + "`table` => 'default.T_INVALID', "
- + "`manifest_sort_order` => 'bucket-first')"))
- .hasStackTraceContaining(
- "Manifest sort order 'bucket-first' requires a bucketed table.");
}
@Test
diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactManifestProcedure.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactManifestProcedure.java
index 89c15761d3b8..25b417e5b0a7 100644
--- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactManifestProcedure.java
+++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactManifestProcedure.java
@@ -59,8 +59,7 @@ public class CompactManifestProcedure extends BaseProcedure {
ProcedureParameter.optional("dry_run", BooleanType),
ProcedureParameter.optional("manifest_sort_enabled", BooleanType),
ProcedureParameter.optional("manifest_sort_partition_field", StringType),
- ProcedureParameter.optional("manifest_sort_max_rewrite_size", StringType),
- ProcedureParameter.optional("manifest_sort_order", StringType)
+ ProcedureParameter.optional("manifest_sort_max_rewrite_size", StringType)
};
private static final StructType OUTPUT_TYPE =
@@ -92,7 +91,6 @@ public InternalRow[] call(InternalRow args) {
Boolean manifestSortEnabled = args.isNullAt(3) ? null : args.getBoolean(3);
String manifestSortPartitionField = args.isNullAt(4) ? null : args.getString(4);
String manifestSortMaxRewriteSize = args.isNullAt(5) ? null : args.getString(5);
- String manifestSortOrder = args.isNullAt(6) ? null : args.getString(6);
Table table = loadSparkTable(tableIdent).getTable();
HashMap dynamicOptions = new HashMap<>();
@@ -109,21 +107,6 @@ public InternalRow[] call(InternalRow args) {
dynamicOptions.put(
CoreOptions.MANIFEST_SORT_MAX_REWRITE_SIZE.key(), manifestSortMaxRewriteSize);
}
- if (manifestSortOrder != null) {
- if (Boolean.FALSE.equals(manifestSortEnabled)
- || "false"
- .equalsIgnoreCase(
- dynamicOptions.get(CoreOptions.MANIFEST_SORT_ENABLED.key()))) {
- throw new IllegalArgumentException(
- "'manifest_sort_order' cannot be used with 'manifest_sort_enabled=false'.");
- }
- CoreOptions.ManifestSortOrder order =
- CoreOptions.ManifestSortOrder.fromString(manifestSortOrder);
- dynamicOptions.put(CoreOptions.MANIFEST_SORT_ORDER.key(), order.toString());
- dynamicOptions.put(CoreOptions.MANIFEST_SORT_ENABLED.key(), Boolean.TRUE.toString());
- dynamicOptions.put(
- CoreOptions.MANIFEST_SORT_FORCE_REWRITE.key(), Boolean.TRUE.toString());
- }
table = table.copy(dynamicOptions);
if (dryRun) {
diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactManifestProcedureTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactManifestProcedureTest.scala
index 4dab5d46484d..76368e6f240c 100644
--- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactManifestProcedureTest.scala
+++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactManifestProcedureTest.scala
@@ -69,8 +69,7 @@ class CompactManifestProcedureTest extends PaimonSparkTestBase with StreamTest {
"dry_run => true, " +
"manifest_sort_enabled => true, " +
"manifest_sort_partition_field => 'dt', " +
- "manifest_sort_max_rewrite_size => '1gb', " +
- "manifest_sort_order => 'partition-first')")
+ "manifest_sort_max_rewrite_size => '1gb')")
.collectAsList()
Assertions.assertThat(dryRunRows.get(0).getBoolean(0)).isTrue
From 2b4a1a18f162802cfd4747af82de772f7eb676bf Mon Sep 17 00:00:00 2001
From: mingfeng
Date: Mon, 14 Sep 2026 16:22:53 +0800
Subject: [PATCH 5/6] [core] Make compact_manifest perform full manifest sort
---
docs/docs/flink/procedures/compaction.md | 11 +---
docs/docs/spark/procedures/maintenance.md | 14 +----
.../java/org/apache/paimon/CoreOptions.java | 18 ------
.../paimon/operation/FileStoreCommitImpl.java | 20 +++---
.../operation/ManifestCompactDryRun.java | 13 ++--
.../paimon/operation/ManifestFileMerger.java | 18 +++++-
.../paimon/operation/ManifestFileSorter.java | 39 ++++++------
.../paimon/schema/SchemaValidation.java | 12 +---
.../paimon/manifest/ManifestFileMetaTest.java | 62 +++++--------------
.../paimon/operation/FileStoreCommitTest.java | 12 ++--
.../operation/ManifestFileMergerTest.java | 4 +-
.../ManifestFileMergerTestUtils.java | 40 ++++++++++++
.../paimon/schema/SchemaValidationTest.java | 26 --------
.../CompactManifestProcedureITCase.java | 20 +-----
14 files changed, 118 insertions(+), 191 deletions(-)
create mode 100644 paimon-core/src/test/java/org/apache/paimon/operation/ManifestFileMergerTestUtils.java
diff --git a/docs/docs/flink/procedures/compaction.md b/docs/docs/flink/procedures/compaction.md
index 22090db69850..5b26f1dfdab4 100644
--- a/docs/docs/flink/procedures/compaction.md
+++ b/docs/docs/flink/procedures/compaction.md
@@ -250,7 +250,9 @@ To compact_manifest the manifests. Arguments:
- manifest_sort_max_rewrite_size (String, optional): maximum manifest size rewritten by one sort pass.
-Set `manifest-sort.force-rewrite=true` in `options` together with `manifest_sort_enabled=true` to rewrite already compacted manifest runs using the layout selected from the table options. Use it only as a one-shot dynamic option. The existing `manifest_sort_max_rewrite_size` rewrite budget semantics still apply; raise it to migrate more manifests in one invocation.
+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**
@@ -282,13 +284,6 @@ CALL sys.compact_manifest(
`manifest_sort_partition_field` => 'dt',
`manifest_sort_max_rewrite_size` => '1 gb'
);
-
-CALL sys.compact_manifest(
- `table` => 'default.T',
- `options` => 'manifest-sort.force-rewrite=true',
- `manifest_sort_enabled` => true,
- `manifest_sort_max_rewrite_size` => '1 gb'
-);
```
## rescale
diff --git a/docs/docs/spark/procedures/maintenance.md b/docs/docs/spark/procedures/maintenance.md
index 9cabe8571323..9952a70cc552 100644
--- a/docs/docs/spark/procedures/maintenance.md
+++ b/docs/docs/spark/procedures/maintenance.md
@@ -126,10 +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.
-Set `manifest-sort.force-rewrite=true` in `options` together with `manifest_sort_enabled=true` to
-rewrite already compacted manifest runs using the layout selected from the table options. Use it
-only as a one-shot dynamic option. The existing `manifest_sort_max_rewrite_size` rewrite budget
-semantics still apply; raise it to migrate more manifests in one invocation.
+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');
@@ -142,13 +141,6 @@ CALL sys.compact_manifest(
manifest_sort_partition_field => 'dt',
manifest_sort_max_rewrite_size => '1 gb'
);
-
-CALL sys.compact_manifest(
- `table` => 'default.T',
- options => 'manifest-sort.force-rewrite=true',
- manifest_sort_enabled => true,
- manifest_sort_max_rewrite_size => '1 gb'
-);
```
## materialize_deletion_vectors
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index f6f9892670f4..7b0665c50296 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -599,20 +599,6 @@ public InlineElement getDescription() {
+ " skipped. Set to a larger value to allow more aggressive"
+ " sort rewriting. The cap only limits the sorted rewrite portion and full/minor cleanup may still happen beyond it.");
- @ExcludeFromDocumentation("Only used by compact_manifest maintenance procedure")
- public static final ConfigOption MANIFEST_SORT_FORCE_REWRITE =
- key("manifest-sort.force-rewrite")
- .booleanType()
- .defaultValue(false)
- .withDescription(
- "When 'manifest-sort.enabled' is true, force an explicit manifest"
- + " compaction to rewrite already compacted manifest runs using"
- + " the layout selected from the table options."
- + " The existing 'manifest-sort.max-rewrite-size' rewrite budget"
- + " semantics still apply."
- + " This should be supplied as a one-shot dynamic option for"
- + " maintenance, not persisted for routine writes.");
-
public static final ConfigOption MANIFEST_MERGE_OPTIMIZE_ENABLED =
key("manifest.merge-optimize.enabled")
.booleanType()
@@ -3241,10 +3227,6 @@ public long manifestSortMaxRewriteSize() {
return options.get(MANIFEST_SORT_MAX_REWRITE_SIZE).getBytes();
}
- public boolean manifestSortForceRewrite() {
- return options.get(MANIFEST_SORT_FORCE_REWRITE);
- }
-
public boolean manifestMergeOptimizeEnabled() {
return options.get(MANIFEST_MERGE_OPTIMIZE_ENABLED);
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
index 2325d85ff7ed..0f60d6a3cb7c 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
+++ b/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 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);
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestCompactDryRun.java b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestCompactDryRun.java
index 480871a4a6da..f80554ce6ea4 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestCompactDryRun.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestCompactDryRun.java
@@ -110,13 +110,11 @@ private static List buildLevelSortedRunsForDryRun(
RowType partitionType,
CoreOptions options) {
long suggestedMetaSize = options.manifestTargetSize().getBytes();
- boolean forceRewrite = options.manifestSortForceRewrite();
boolean fullCompaction =
- forceRewrite
- || ManifestFileSorter.reachesFullCompactionThreshold(
- manifests,
- suggestedMetaSize,
- options.manifestFullCompactionThresholdSize().getBytes());
+ ManifestFileSorter.reachesFullCompactionThreshold(
+ manifests,
+ suggestedMetaSize,
+ options.manifestFullCompactionThresholdSize().getBytes());
ManifestFileSorter.ManifestSortKey sortKey =
ManifestFileSorter.createSortKey(
options.dataEvolutionEnabled(),
@@ -136,8 +134,7 @@ private static List buildLevelSortedRunsForDryRun(
// 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 (!forceRewrite
- && fullCompaction
+ if (fullCompaction
&& classifyResult.compactWithoutSort.isEmpty()
&& new ManifestPickStrategy(
options.maxSizeAmplificationPercent(), options.sortedRunSizeRatio())
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileMerger.java b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileMerger.java
index e3f8c7af7671..1dd630d077b3 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileMerger.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileMerger.java
@@ -57,6 +57,16 @@ public static List merge(
RowType partitionType,
CoreOptions options,
@Nullable IOManager ioManager) {
+ return merge(input, manifestFile, partitionType, options, ioManager, false);
+ }
+
+ static List merge(
+ List 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 newFilesForAbort = new ArrayList<>();
@@ -66,7 +76,13 @@ public static List 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()) {
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileSorter.java b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileSorter.java
index f6981de56577..e581f2f301c2 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileSorter.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileSorter.java
@@ -71,7 +71,7 @@ public class ManifestFileSorter {
/** Context object that carries shared state across compaction methods. */
static class CompactionContext {
final boolean fullCompaction;
- final boolean forceRewrite;
+ final boolean fullSort;
final boolean runMergeOptimizeEnabled;
final ManifestSortKey sortKey;
final RowType partitionType;
@@ -92,7 +92,7 @@ static class CompactionContext {
CompactionContext(
boolean fullCompaction,
- boolean forceRewrite,
+ boolean fullSort,
boolean runMergeOptimizeEnabled,
ManifestSortKey sortKey,
RowType partitionType,
@@ -102,7 +102,7 @@ static class CompactionContext {
List levelRuns,
List pickedRuns) {
this.fullCompaction = fullCompaction;
- this.forceRewrite = forceRewrite;
+ this.fullSort = fullSort;
this.runMergeOptimizeEnabled = runMergeOptimizeEnabled;
this.sortKey = sortKey;
this.partitionType = partitionType;
@@ -156,7 +156,8 @@ static List 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;
@@ -164,7 +165,6 @@ static List trySortCompaction(
long suggestedMetaSize = options.manifestTargetSize().getBytes();
int suggestedMinMetaCount = options.manifestMergeMinCount();
long fullCompactionThreshold = options.manifestFullCompactionThresholdSize().getBytes();
- boolean forceRewrite = options.manifestSortForceRewrite();
long maxRewriteSize = options.manifestSortMaxRewriteSize();
int maxSizeAmplificationPercent = options.maxSizeAmplificationPercent();
int sortedRunSizeRatio = options.sortedRunSizeRatio();
@@ -185,7 +185,7 @@ static List trySortCompaction(
suggestedMetaSize,
suggestedMinMetaCount,
fullCompactionThreshold,
- forceRewrite,
+ fullSort,
maxRewriteSize,
maxSizeAmplificationPercent,
sortedRunSizeRatio,
@@ -230,7 +230,7 @@ private static Optional> tryFullCompaction(
long suggestedMetaSize,
int suggestedMinMetaCount,
long fullCompactionThreshold,
- boolean forceRewrite,
+ boolean fullSort,
long maxRewriteSize,
int maxSizeAmplificationPercent,
int sortedRunSizeRatio,
@@ -238,7 +238,7 @@ private static Optional> tryFullCompaction(
@Nullable Integer manifestReadParallelism)
throws Exception {
// Step 1: Check if full compaction threshold is met
- if (!forceRewrite
+ if (!fullSort
&& !reachesFullCompactionThreshold(
input, suggestedMetaSize, fullCompactionThreshold)) {
return Optional.empty();
@@ -248,7 +248,7 @@ private static Optional> tryFullCompaction(
prepareCompaction(
input,
true,
- forceRewrite,
+ fullSort,
manifestFile,
partitionType,
sortPartitionField,
@@ -262,10 +262,8 @@ private static Optional> tryFullCompaction(
manifestReadParallelism);
try {
List levelRuns = ctx.levelRuns;
- List pickedRuns = ctx.pickedRuns;
- if (forceRewrite) {
- pickedRuns = new ArrayList<>(levelRuns);
- }
+ List pickedRuns =
+ fullSort ? new ArrayList<>(levelRuns) : ctx.pickedRuns;
if (pickedRuns.isEmpty() && ctx.defaultCompactFiles.isEmpty()) {
LOG.debug(
@@ -295,11 +293,10 @@ private static Optional> tryFullCompaction(
}
pickedFiles.addAll(ctx.defaultCompactFiles.keySet());
- // Step 4: Split into sections and merge small adjacent sections. A forced rewrite
- // intentionally uses one global section so entries from different already-compacted
- // manifests can be clustered using the layout selected from the table options.
+ // 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 sections;
- if (forceRewrite) {
+ if (fullSort) {
long totalSize = 0L;
boolean hasDefaultCompactFile = false;
for (ManifestFileMeta file : pickedFiles) {
@@ -485,7 +482,7 @@ private static List tryMinorCompaction(
private static CompactionContext prepareCompaction(
List input,
boolean fullCompaction,
- boolean forceRewrite,
+ boolean fullSort,
ManifestFile manifestFile,
RowType partitionType,
String sortPartitionField,
@@ -528,7 +525,7 @@ private static CompactionContext prepareCompaction(
return new CompactionContext(
fullCompaction,
- forceRewrite,
+ fullSort,
useRunMergeOptimize,
sortKey,
partitionType,
@@ -1099,11 +1096,11 @@ private static void rewriteSection(
CompactionContext ctx,
ManifestFile manifestFile,
@Nullable Integer manifestReadParallelism,
- boolean allowForceRewrite)
+ boolean allowFullRewrite)
throws Exception {
// Skip rewrite for single file not in delete-range.
if (section.size() == 1
- && !(allowForceRewrite && ctx.forceRewrite)
+ && !(allowFullRewrite && ctx.fullSort)
&& !ctx.defaultCompactFiles.getOrDefault(section.get(0), false)) {
output.addUnchanged(section.get(0));
return;
diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
index 8144112db6c5..996fceaa97bf 100644
--- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
+++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
@@ -417,7 +417,7 @@ public static void validateTableSchema(TableSchema schema, Set dynamicOp
validatePkClusteringOverride(options);
- validateManifestSort(schema, options, dynamicOptionKeys);
+ validateManifestSort(schema, options);
}
/**
@@ -2019,15 +2019,7 @@ public static void validatePkClusteringOverride(CoreOptions options) {
}
}
- private static void validateManifestSort(
- TableSchema schema, CoreOptions options, Set dynamicOptionKeys) {
- checkArgument(
- !schema.options().containsKey(CoreOptions.MANIFEST_SORT_FORCE_REWRITE.key())
- || dynamicOptionKeys.contains(
- CoreOptions.MANIFEST_SORT_FORCE_REWRITE.key()),
- "'%s' is only supported as a dynamic option for explicit manifest compaction.",
- CoreOptions.MANIFEST_SORT_FORCE_REWRITE.key());
-
+ private static void validateManifestSort(TableSchema schema, CoreOptions options) {
if (options.manifestSortEnabled()) {
if (!options.dataEvolutionEnabled()) {
checkArgument(
diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTest.java
index 0b6a34ba57a4..557f2131345d 100644
--- a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTest.java
@@ -33,6 +33,7 @@
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.operation.ManifestCompactDryRun;
import org.apache.paimon.operation.ManifestFileMerger;
+import org.apache.paimon.operation.ManifestFileMergerTestUtils;
import org.apache.paimon.options.Options;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.schema.FileSystemSchemaManager;
@@ -1353,7 +1354,7 @@ public void testManifestSortUsesBucketAsPrimaryKeyForBucketedTable(int bucket) {
}
@Test
- public void testManifestSortForceRewriteAlreadyCompactedRuns() {
+ public void testManifestSortFullCompactionAlreadyCompactedRuns() {
List physical =
Arrays.asList(
makeManifest(makeBucketEntry("a-3", 0, 3), makeBucketEntry("a-1", 0, 1)),
@@ -1373,6 +1374,7 @@ public void testManifestSortForceRewriteAlreadyCompactedRuns() {
Options testOptions = new Options();
testOptions.set(CoreOptions.MANIFEST_SORT_ENABLED, true);
testOptions.set(CoreOptions.MANIFEST_SORT_MAX_REWRITE_SIZE.key(), "1G");
+ testOptions.set(CoreOptions.MANIFEST_FULL_COMPACTION_FILE_SIZE.key(), Long.MAX_VALUE + "B");
testOptions.set(CoreOptions.BUCKET, 4);
List unchanged =
@@ -1383,9 +1385,8 @@ public void testManifestSortForceRewriteAlreadyCompactedRuns() {
CoreOptions.fromMap(testOptions.toMap()));
assertThat(unchanged).containsExactlyInAnyOrderElementsOf(input);
- testOptions.set(CoreOptions.MANIFEST_SORT_FORCE_REWRITE, true);
List rewritten =
- ManifestFileMerger.merge(
+ ManifestFileMergerTestUtils.fullMerge(
input,
manifestFile,
getPartitionType(),
@@ -1406,7 +1407,7 @@ public void testManifestSortForceRewriteAlreadyCompactedRuns() {
assertThat(rewritten.get(0).maxBucket()).isEqualTo(3);
assertThat(rewritten.get(0).totalBuckets()).isEqualTo(240);
- testOptions.set(CoreOptions.MANIFEST_SORT_FORCE_REWRITE, false);
+ testOptions.set(CoreOptions.MANIFEST_FULL_COMPACTION_FILE_SIZE.key(), Long.MAX_VALUE + "B");
List afterMigration =
ManifestFileMerger.merge(
rewritten,
@@ -1417,7 +1418,7 @@ public void testManifestSortForceRewriteAlreadyCompactedRuns() {
}
@Test
- public void testManifestSortForceRewriteAllLevelRuns() {
+ public void testManifestSortFullCompactionAllLevelRuns() {
List physical =
Arrays.asList(
makeManifest(makeBucketEntry("a-3", 0, 3), makeBucketEntry("a-1", 2, 1)),
@@ -1431,6 +1432,7 @@ public void testManifestSortForceRewriteAllLevelRuns() {
Options testOptions = new Options();
testOptions.set(CoreOptions.MANIFEST_SORT_ENABLED, true);
testOptions.set(CoreOptions.MANIFEST_SORT_MAX_REWRITE_SIZE.key(), "1G");
+ testOptions.set(CoreOptions.MANIFEST_FULL_COMPACTION_FILE_SIZE.key(), Long.MAX_VALUE + "B");
testOptions.set(CoreOptions.BUCKET, 4);
// Without bucket metadata, manifest sort falls back to the overlapping partition ranges
@@ -1454,9 +1456,8 @@ public void testManifestSortForceRewriteAllLevelRuns() {
CoreOptions.fromMap(testOptions.toMap()));
assertThat(unchanged).containsExactlyInAnyOrderElementsOf(input);
- testOptions.set(CoreOptions.MANIFEST_SORT_FORCE_REWRITE, true);
List rewritten =
- ManifestFileMerger.merge(
+ ManifestFileMergerTestUtils.fullMerge(
input,
manifestFile,
getPartitionType(),
@@ -1475,34 +1476,7 @@ public void testManifestSortForceRewriteAllLevelRuns() {
}
@Test
- public void testManifestSortForceRewriteDryRunUsesFullCompaction() {
- List input =
- Arrays.asList(
- makeManifest(makeEntry(true, "base", 0)),
- makeManifest(
- makeEntry(false, "base", 0), makeEntry(true, "replacement", 0)));
-
- Options testOptions = new Options();
- testOptions.set(CoreOptions.MANIFEST_SORT_ENABLED, true);
- testOptions.set(CoreOptions.MANIFEST_SORT_FORCE_REWRITE, true);
- testOptions.set(CoreOptions.MANIFEST_TARGET_FILE_SIZE.key(), "1B");
- testOptions.set(CoreOptions.MANIFEST_FULL_COMPACTION_FILE_SIZE.key(), Long.MAX_VALUE + "B");
-
- FileStoreTable table = mock(FileStoreTable.class, RETURNS_DEEP_STUBS);
- Snapshot snapshot = mock(Snapshot.class);
- when(table.options()).thenReturn(testOptions.toMap());
- when(table.store().snapshotManager().latestSnapshot()).thenReturn(snapshot);
- when(table.store().manifestListFactory().create().readDataManifests(snapshot))
- .thenReturn(input);
- when(table.store().manifestFileFactory().create()).thenReturn(manifestFile);
- when(table.schema().logicalPartitionType()).thenReturn(getPartitionType());
-
- assertThat(ManifestCompactDryRun.execute(table))
- .endsWith("Manifest sort level files: L0=0, L1=0, L2=0, L3=0, L4=0.");
- }
-
- @Test
- public void testManifestSortForceRewriteSingleManifest() {
+ public void testManifestSortFullCompactionSingleManifest() {
ManifestFileMeta physical =
makeManifest(makeBucketEntry("file-3", 0, 3), makeBucketEntry("file-0", 0, 0));
ManifestFileMeta input =
@@ -1511,10 +1485,9 @@ public void testManifestSortForceRewriteSingleManifest() {
Options testOptions = new Options();
testOptions.set(CoreOptions.MANIFEST_SORT_ENABLED, true);
- testOptions.set(CoreOptions.MANIFEST_SORT_FORCE_REWRITE, true);
testOptions.set(CoreOptions.BUCKET, 4);
List rewritten =
- ManifestFileMerger.merge(
+ ManifestFileMergerTestUtils.fullMerge(
Collections.singletonList(input),
manifestFile,
getPartitionType(),
@@ -1528,7 +1501,7 @@ public void testManifestSortForceRewriteSingleManifest() {
}
@Test
- public void testManifestSortForceRewriteRespectsRewriteBudget() {
+ public void testManifestSortFullCompactionRespectsRewriteLimit() {
long targetSize = CoreOptions.MANIFEST_TARGET_FILE_SIZE.defaultValue().getBytes();
List input = new ArrayList<>();
for (int i = 0; i < 4; i++) {
@@ -1538,11 +1511,10 @@ public void testManifestSortForceRewriteRespectsRewriteBudget() {
Options testOptions = new Options();
testOptions.set(CoreOptions.MANIFEST_SORT_ENABLED, true);
- testOptions.set(CoreOptions.MANIFEST_SORT_FORCE_REWRITE, true);
testOptions.set(CoreOptions.MANIFEST_SORT_MAX_REWRITE_SIZE.key(), "1B");
testOptions.set(CoreOptions.BUCKET, 4);
List rewritten =
- ManifestFileMerger.merge(
+ ManifestFileMergerTestUtils.fullMerge(
input,
manifestFile,
getPartitionType(),
@@ -1558,7 +1530,7 @@ public void testManifestSortForceRewriteRespectsRewriteBudget() {
}
@Test
- public void testManifestSortForceRewriteDoesNotExceedBudgetForSingletonTail() {
+ public void testManifestSortFullCompactionDoesNotExceedLimitForSingletonTail() {
long targetSize = CoreOptions.MANIFEST_TARGET_FILE_SIZE.defaultValue().getBytes();
List input = new ArrayList<>();
for (int i = 0; i < 3; i++) {
@@ -1568,11 +1540,10 @@ public void testManifestSortForceRewriteDoesNotExceedBudgetForSingletonTail() {
Options testOptions = new Options();
testOptions.set(CoreOptions.MANIFEST_SORT_ENABLED, true);
- testOptions.set(CoreOptions.MANIFEST_SORT_FORCE_REWRITE, true);
testOptions.set(CoreOptions.MANIFEST_SORT_MAX_REWRITE_SIZE.key(), "1B");
testOptions.set(CoreOptions.BUCKET, 4);
List rewritten =
- ManifestFileMerger.merge(
+ ManifestFileMergerTestUtils.fullMerge(
input,
manifestFile,
getPartitionType(),
@@ -1588,7 +1559,7 @@ public void testManifestSortForceRewriteDoesNotExceedBudgetForSingletonTail() {
}
@Test
- public void testManifestSortForceRewriteDoesNotRewriteTailBeyondBudget() {
+ public void testManifestSortFullCompactionDoesNotRewriteTailBeyondLimit() {
long targetSize = CoreOptions.MANIFEST_TARGET_FILE_SIZE.defaultValue().getBytes();
List input = new ArrayList<>();
for (int i = 0; i < 4; i++) {
@@ -1601,11 +1572,10 @@ public void testManifestSortForceRewriteDoesNotRewriteTailBeyondBudget() {
Options testOptions = new Options();
testOptions.set(CoreOptions.MANIFEST_SORT_ENABLED, true);
- testOptions.set(CoreOptions.MANIFEST_SORT_FORCE_REWRITE, true);
testOptions.set(CoreOptions.MANIFEST_SORT_MAX_REWRITE_SIZE.key(), "1B");
testOptions.set(CoreOptions.BUCKET, 8);
List rewritten =
- ManifestFileMerger.merge(
+ ManifestFileMergerTestUtils.fullMerge(
input,
manifestFile,
getPartitionType(),
diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
index c96b7b711195..437944fae5d9 100644
--- a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
@@ -1430,21 +1430,17 @@ public void testManifestCompact(boolean skipOnWriteOnly, boolean writeOnly) thro
}
@Test
- public void testManifestSortCompactManifestRespectsCompactionThresholds() {
+ public void testManifestSortCompactManifestUsesFullCompactionThresholds() {
Options options = new Options();
options.set(CoreOptions.MANIFEST_SORT_ENABLED, true);
options.set(CoreOptions.MANIFEST_MERGE_MIN_COUNT, 100);
options.set(CoreOptions.MANIFEST_FULL_COMPACTION_FILE_SIZE.key(), Long.MAX_VALUE + "B");
CoreOptions compactOptions =
- FileStoreCommitImpl.manifestCompactionOptions(
- new CoreOptions(options),
- Collections.emptyList(),
- TestKeyValueGenerator.DEFAULT_PART_TYPE);
+ FileStoreCommitImpl.manifestCompactionOptions(new CoreOptions(options));
- assertThat(compactOptions.manifestMergeMinCount()).isEqualTo(100);
- assertThat(compactOptions.manifestFullCompactionThresholdSize().getBytes())
- .isEqualTo(Long.MAX_VALUE);
+ assertThat(compactOptions.manifestMergeMinCount()).isEqualTo(1);
+ assertThat(compactOptions.manifestFullCompactionThresholdSize().getBytes()).isEqualTo(1);
}
@Test
diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/ManifestFileMergerTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/ManifestFileMergerTest.java
index cfd1cb8fc1bd..57b4dadec48f 100644
--- a/paimon-core/src/test/java/org/apache/paimon/operation/ManifestFileMergerTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/operation/ManifestFileMergerTest.java
@@ -81,9 +81,7 @@ public void testManifestSortFallsBackToForcedLegacyMergeWithoutRowId() {
assertThat(ManifestFileMerger.canUseManifestSort(input, NO_PARTITION_TYPE, tableOptions))
.isFalse();
- CoreOptions compactOptions =
- FileStoreCommitImpl.manifestCompactionOptions(
- tableOptions, input, NO_PARTITION_TYPE);
+ CoreOptions compactOptions = FileStoreCommitImpl.manifestCompactionOptions(tableOptions);
assertThat(compactOptions.manifestMergeMinCount()).isEqualTo(1);
assertThat(compactOptions.manifestFullCompactionThresholdSize().getBytes()).isEqualTo(1);
diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/ManifestFileMergerTestUtils.java b/paimon-core/src/test/java/org/apache/paimon/operation/ManifestFileMergerTestUtils.java
new file mode 100644
index 000000000000..87d93d128530
--- /dev/null
+++ b/paimon-core/src/test/java/org/apache/paimon/operation/ManifestFileMergerTestUtils.java
@@ -0,0 +1,40 @@
+/*
+ * 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.operation;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.manifest.ManifestFile;
+import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.types.RowType;
+
+import java.util.List;
+
+/** Test access to explicit full manifest compaction. */
+public class ManifestFileMergerTestUtils {
+
+ private ManifestFileMergerTestUtils() {}
+
+ public static List fullMerge(
+ List input,
+ ManifestFile manifestFile,
+ RowType partitionType,
+ CoreOptions options) {
+ return ManifestFileMerger.merge(input, manifestFile, partitionType, options, null, true);
+ }
+}
diff --git a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
index f66d623a0e23..c9588f4552fa 100644
--- a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
@@ -36,7 +36,6 @@
import java.util.Map;
import static java.util.Collections.emptyList;
-import static java.util.Collections.singleton;
import static java.util.Collections.singletonList;
import static org.apache.paimon.CoreOptions.BUCKET;
import static org.apache.paimon.CoreOptions.DATA_EVOLUTION_ENABLED;
@@ -2050,31 +2049,6 @@ void testManifestSortValidation() {
.hasMessageContaining("is not a partition field");
}
- @Test
- void testManifestSortForceRewriteIsDynamicOnly() {
- List fields =
- Arrays.asList(
- new DataField(0, "f0", DataTypes.INT()),
- new DataField(1, "f1", DataTypes.INT()));
- Map baseOptions = new HashMap<>();
- baseOptions.put(CoreOptions.MANIFEST_SORT_ENABLED.key(), "true");
- baseOptions.put(BUCKET.key(), "-1");
-
- Map forceOptions = new HashMap<>(baseOptions);
- forceOptions.put(CoreOptions.MANIFEST_SORT_FORCE_REWRITE.key(), "true");
- TableSchema forceSchema =
- new TableSchema(1, fields, 10, singletonList("f0"), emptyList(), forceOptions, "");
- assertThatThrownBy(() -> validateTableSchema(forceSchema))
- .hasMessage(
- "'manifest-sort.force-rewrite' is only supported as a dynamic option for explicit manifest compaction.");
- assertThatNoException()
- .isThrownBy(
- () ->
- validateTableSchema(
- forceSchema,
- singleton(CoreOptions.MANIFEST_SORT_FORCE_REWRITE.key())));
- }
-
@Test
public void testMergeOnReadCoexistsWithVisibilityCallback() {
Map options = new HashMap<>();
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CompactManifestProcedureITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CompactManifestProcedureITCase.java
index 213e6b73afb4..a2e0f85fcee3 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CompactManifestProcedureITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/CompactManifestProcedureITCase.java
@@ -18,7 +18,6 @@
package org.apache.paimon.flink.procedure;
-import org.apache.paimon.CoreOptions;
import org.apache.paimon.flink.CatalogITCaseBase;
import org.apache.paimon.flink.action.ActionFactory;
import org.apache.paimon.flink.action.CompactManifestAction;
@@ -123,24 +122,7 @@ public void testManifestSortParameters() throws Exception {
long compactSnapshotId = table.snapshotManager().latestSnapshot().id();
sql(procedure);
Assertions.assertThat(table.snapshotManager().latestSnapshot().id())
- .isEqualTo(compactSnapshotId);
-
- String forceRewriteProcedure =
- "CALL sys.compact_manifest("
- + "`table` => 'default.T_SORT', "
- + "`options` => 'manifest-sort.force-rewrite=true', "
- + "`manifest_sort_enabled` => true, "
- + "`manifest_sort_partition_field` => 'dt', "
- + "`manifest_sort_max_rewrite_size` => '1 gb')";
- sql(forceRewriteProcedure);
- long forceRewriteSnapshotId = table.snapshotManager().latestSnapshot().id();
- Assertions.assertThat(forceRewriteSnapshotId).isEqualTo(compactSnapshotId + 1);
- Assertions.assertThat(paimonTable("T_SORT").options())
- .doesNotContainKey(CoreOptions.MANIFEST_SORT_FORCE_REWRITE.key());
-
- sql(procedure);
- Assertions.assertThat(table.snapshotManager().latestSnapshot().id())
- .isEqualTo(forceRewriteSnapshotId);
+ .isEqualTo(compactSnapshotId + 1);
}
@Test
From 13725eb0dd64d4492a286548fc7dd6ed372b2f0c Mon Sep 17 00:00:00 2001
From: mingfeng
Date: Mon, 14 Sep 2026 17:27:26 +0800
Subject: [PATCH 6/6] [core] Preserve singleton handling in manifest sort
---
.../paimon/operation/ManifestFileSorter.java | 14 ++++++++
.../paimon/manifest/ManifestFileMetaTest.java | 35 +++++++++++++++++++
2 files changed, 49 insertions(+)
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileSorter.java b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileSorter.java
index e581f2f301c2..a44b94a310bd 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileSorter.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileSorter.java
@@ -888,6 +888,20 @@ private static void rewriteSections(
for (int i = 0; i < sections.size(); i++) {
Section section = sections.get(i);
+ // 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,
+ false);
+ continue;
+ }
+
// Phase 1: budget not yet exhausted -- perform aggressive sort rewrite.
if (!budgetExhausted) {
// Phase 1a: section fits within the remaining budget -- sort and rewrite it
diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTest.java
index 557f2131345d..ad3641b121b4 100644
--- a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTest.java
@@ -1590,6 +1590,41 @@ public void testManifestSortFullCompactionDoesNotRewriteTailBeyondLimit() {
assertEquivalentEntries(input, rewritten);
}
+ @Test
+ public void testManifestSortUnchangedSingletonDoesNotConsumeRewriteLimit() {
+ long targetSize = CoreOptions.MANIFEST_TARGET_FILE_SIZE.defaultValue().getBytes();
+ List input = new ArrayList<>();
+ input.add(copyWithFileSize(makeManifest(makeEntry(true, "singleton", 0)), targetSize));
+ for (int i = 0; i < 5; i++) {
+ input.add(
+ copyWithFileSize(
+ makeManifest(
+ makeEntry(true, "range-" + i + "-1", 1),
+ makeEntry(true, "range-" + i + "-2", 2)),
+ targetSize));
+ }
+
+ Options testOptions = new Options();
+ testOptions.set(CoreOptions.MANIFEST_SORT_ENABLED, true);
+ testOptions.set(CoreOptions.MANIFEST_SORT_MAX_REWRITE_SIZE.key(), "4M");
+ testOptions.set(CoreOptions.BUCKET, -1);
+ List merged =
+ ManifestFileMerger.merge(
+ input,
+ manifestFile,
+ getPartitionType(),
+ CoreOptions.fromMap(testOptions.toMap()));
+
+ Set inputNames =
+ input.stream().map(ManifestFileMeta::fileName).collect(Collectors.toSet());
+ assertThat(merged).extracting(ManifestFileMeta::fileName).contains(input.get(0).fileName());
+ assertThat(merged)
+ .extracting(ManifestFileMeta::fileName)
+ .filteredOn(inputNames::contains)
+ .hasSize(4);
+ assertEquivalentEntries(input, merged);
+ }
+
@ParameterizedTest
@ValueSource(ints = {-1, 4, -2})
public void testManifestSortDryRunUsesBucketRangesForBucketedTable(int bucket) {