This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 967b242ba9 [core] Make compact_manifest perform a full manifest sort
(#9802)
967b242ba9 is described below
commit 967b242ba960687c53da0d6a85fe41d67f18361d
Author: jianguotian <[email protected]>
AuthorDate: Mon Sep 14 17:42:10 2026 +0800
[core] Make compact_manifest perform a full manifest sort (#9802)
---
docs/docs/flink/procedures/compaction.md | 4 +
docs/docs/spark/procedures/maintenance.md | 3 +
.../paimon/operation/FileStoreCommitImpl.java | 20 +-
.../paimon/operation/ManifestFileMerger.java | 18 +-
.../paimon/operation/ManifestFileSorter.java | 65 ++++-
.../paimon/manifest/ManifestFileMetaTest.java | 309 +++++++++++++++++++++
.../paimon/operation/FileStoreCommitTest.java | 12 +-
.../paimon/operation/ManifestFileMergerTest.java | 4 +-
.../operation/ManifestFileMergerTestUtils.java | 40 +++
.../procedure/CompactManifestProcedureITCase.java | 2 +-
10 files changed, 437 insertions(+), 40 deletions(-)
diff --git a/docs/docs/flink/procedures/compaction.md
b/docs/docs/flink/procedures/compaction.md
index 71c2fd531d..5b26f1dfda 100644
--- a/docs/docs/flink/procedures/compaction.md
+++ b/docs/docs/flink/procedures/compaction.md
@@ -250,6 +250,10 @@ To compact_manifest the manifests. Arguments:
- manifest_sort_max_rewrite_size (String, optional): maximum manifest size
rewritten by one sort pass.
+When manifest sort is enabled, `compact_manifest` performs a full sort using
the layout selected
+from the table options. The existing `manifest_sort_max_rewrite_size` limit
still controls the
+amount of manifest data rewritten in one invocation.
+
**Syntax**
```sql
diff --git a/docs/docs/spark/procedures/maintenance.md
b/docs/docs/spark/procedures/maintenance.md
index 1244848028..9952a70cc5 100644
--- a/docs/docs/spark/procedures/maintenance.md
+++ b/docs/docs/spark/procedures/maintenance.md
@@ -126,6 +126,9 @@ Compact manifest files.
- `manifest_sort_enabled` (`BOOLEAN`, optional): whether to use manifest sort
rewrite for this invocation.
- `manifest_sort_partition_field` (`STRING`, optional): partition field used
to sort manifest entries. Defaults to the first partition field.
- `manifest_sort_max_rewrite_size` (`STRING`, optional): maximum manifest size
rewritten by one sort pass.
+When manifest sort is enabled, `compact_manifest` performs a full sort using
the layout selected
+from the table options. The existing `manifest_sort_max_rewrite_size` limit
still controls the
+amount of manifest data rewritten in one invocation.
```sql
CALL sys.compact_manifest(`table` => 'default.T');
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 2325d85ff7..0f60d6a3cb 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 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
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 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
return commitSnapshotImpl(latestSnapshot, newSnapshot, emptyList());
}
- static CoreOptions manifestCompactionOptions(
- CoreOptions options, List<ManifestFileMeta> manifests, RowType
partitionType) {
- // Use a copied options with forced full compaction settings for the
legacy merge path.
- // Manifest sort has its own full/minor picking strategy and should
respect its configured
- // thresholds.
+ static CoreOptions manifestCompactionOptions(CoreOptions options) {
+ // Use copied options so explicit manifest compaction always takes the
full-compaction path
+ // without changing the table options used by regular commits.
Options compactOptions = Options.fromMap(options.toMap());
- if (!ManifestFileMerger.canUseManifestSort(manifests, partitionType,
options)) {
- compactOptions.set(CoreOptions.MANIFEST_MERGE_MIN_COUNT, 1);
- compactOptions.set(
- CoreOptions.MANIFEST_FULL_COMPACTION_FILE_SIZE,
MemorySize.ofBytes(1));
- }
+ compactOptions.set(CoreOptions.MANIFEST_MERGE_MIN_COUNT, 1);
+ compactOptions.set(CoreOptions.MANIFEST_FULL_COMPACTION_FILE_SIZE,
MemorySize.ofBytes(1));
return new CoreOptions(compactOptions);
}
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 688d59f492..79d21e025e 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
@@ -58,6 +58,16 @@ public class ManifestFileMerger {
RowType partitionType,
CoreOptions options,
@Nullable IOManager ioManager) {
+ return merge(input, manifestFile, partitionType, options, ioManager,
false);
+ }
+
+ static List<ManifestFileMeta> merge(
+ List<ManifestFileMeta> input,
+ ManifestFile manifestFile,
+ RowType partitionType,
+ CoreOptions options,
+ @Nullable IOManager ioManager,
+ boolean fullCompaction) {
// these are the newly created manifest files, clean them up if
exception occurs
List<ManifestFileMeta> newFilesForAbort = new ArrayList<>();
@@ -66,7 +76,13 @@ public class ManifestFileMerger {
// partition fields for manifest sort rewrite.
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 94255953fc..30351f9e79 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 fullSort;
final boolean runMergeOptimizeEnabled;
final ManifestSortKey sortKey;
final RowType partitionType;
@@ -91,6 +92,7 @@ public class ManifestFileSorter {
CompactionContext(
boolean fullCompaction,
+ boolean fullSort,
boolean runMergeOptimizeEnabled,
ManifestSortKey sortKey,
RowType partitionType,
@@ -100,6 +102,7 @@ public class ManifestFileSorter {
List<ManifestAdjacentSortedRun> levelRuns,
List<ManifestAdjacentSortedRun> pickedRuns) {
this.fullCompaction = fullCompaction;
+ this.fullSort = fullSort;
this.runMergeOptimizeEnabled = runMergeOptimizeEnabled;
this.sortKey = sortKey;
this.partitionType = partitionType;
@@ -153,7 +156,8 @@ public class ManifestFileSorter {
ManifestFile manifestFile,
RowType partitionType,
CoreOptions options,
- @Nullable IOManager ioManager)
+ @Nullable IOManager ioManager,
+ boolean fullSort)
throws Exception {
String sortPartitionField = options.manifestSortPartitionField();
boolean bucketed = options.bucket() > 0 || options.bucket() ==
BucketMode.POSTPONE_BUCKET;
@@ -181,6 +185,7 @@ public class ManifestFileSorter {
suggestedMetaSize,
suggestedMinMetaCount,
fullCompactionThreshold,
+ fullSort,
maxRewriteSize,
maxSizeAmplificationPercent,
sortedRunSizeRatio,
@@ -225,6 +230,7 @@ public class ManifestFileSorter {
long suggestedMetaSize,
int suggestedMinMetaCount,
long fullCompactionThreshold,
+ boolean fullSort,
long maxRewriteSize,
int maxSizeAmplificationPercent,
int sortedRunSizeRatio,
@@ -232,7 +238,9 @@ public class ManifestFileSorter {
@Nullable Integer manifestReadParallelism)
throws Exception {
// Step 1: Check if full compaction threshold is met
- if (!reachesFullCompactionThreshold(input, suggestedMetaSize,
fullCompactionThreshold)) {
+ if (!fullSort
+ && !reachesFullCompactionThreshold(
+ input, suggestedMetaSize, fullCompactionThreshold)) {
return Optional.empty();
}
// Step 2: Prepare compaction context
@@ -240,6 +248,7 @@ public class ManifestFileSorter {
prepareCompaction(
input,
true,
+ fullSort,
manifestFile,
partitionType,
sortPartitionField,
@@ -253,7 +262,8 @@ public class ManifestFileSorter {
manifestReadParallelism);
try {
List<ManifestAdjacentSortedRun> levelRuns = ctx.levelRuns;
- List<ManifestAdjacentSortedRun> pickedRuns = ctx.pickedRuns;
+ List<ManifestAdjacentSortedRun> pickedRuns =
+ fullSort ? new ArrayList<>(levelRuns) : ctx.pickedRuns;
if (pickedRuns.isEmpty() && ctx.defaultCompactFiles.isEmpty()) {
LOG.debug(
@@ -283,9 +293,22 @@ public class ManifestFileSorter {
}
pickedFiles.addAll(ctx.defaultCompactFiles.keySet());
- // Step 4: Split into sections and merge small adjacent sections
- List<Section> sections = splitIntoSections(pickedFiles, ctx);
- sections = mergeSmallAdjacentSections(sections, suggestedMetaSize);
+ // Step 4: A full sort uses one global section so entries from all
existing runs can be
+ // clustered using the layout selected from the table options.
+ List<Section> sections;
+ if (fullSort) {
+ long totalSize = 0L;
+ boolean hasDefaultCompactFile = false;
+ for (ManifestFileMeta file : pickedFiles) {
+ totalSize += file.fileSize();
+ hasDefaultCompactFile |=
ctx.isMarkedForDefaultCompaction(file);
+ }
+ sections = new ArrayList<>();
+ sections.add(new Section(pickedFiles, totalSize,
hasDefaultCompactFile));
+ } else {
+ sections = splitIntoSections(pickedFiles, ctx);
+ sections = mergeSmallAdjacentSections(sections,
suggestedMetaSize);
+ }
LOG.info(
"Manifest sort full compact: pickedFiles={}, sections={}.",
@@ -343,6 +366,7 @@ public class ManifestFileSorter {
prepareCompaction(
input,
false,
+ false,
manifestFile,
partitionType,
sortPartitionField,
@@ -458,6 +482,7 @@ public class ManifestFileSorter {
private static CompactionContext prepareCompaction(
List<ManifestFileMeta> input,
boolean fullCompaction,
+ boolean fullSort,
ManifestFile manifestFile,
RowType partitionType,
String sortPartitionField,
@@ -500,6 +525,7 @@ public class ManifestFileSorter {
return new CompactionContext(
fullCompaction,
+ fullSort,
useRunMergeOptimize,
sortKey,
partitionType,
@@ -862,15 +888,17 @@ public class ManifestFileSorter {
for (int i = 0; i < sections.size(); i++) {
Section section = sections.get(i);
- // A single-file section is always handled directly, regardless of
the budget.
- if (section.files.size() == 1) {
+ // Preserve the ordinary-compaction shortcut: an unchanged
singleton must not consume
+ // the sort rewrite limit. Explicit full sort intentionally
rewrites the singleton.
+ if (!ctx.fullSort && section.files.size() == 1) {
rewriteSection(
section.files,
output,
sortNewFiles,
ctx,
manifestFile,
- manifestReadParallelism);
+ manifestReadParallelism,
+ false);
continue;
}
@@ -886,7 +914,8 @@ public class ManifestFileSorter {
sortNewFiles,
ctx,
manifestFile,
- manifestReadParallelism);
+ manifestReadParallelism,
+ true);
} else {
// Phase 1b: first overflow -- split the section at the
budget boundary,
// rewrite the affordable head, and append the remaining
tail back for later
@@ -966,7 +995,8 @@ public class ManifestFileSorter {
}
}
- rewriteSection(headFiles, output, sortNewFiles, ctx, manifestFile,
manifestReadParallelism);
+ rewriteSection(
+ headFiles, output, sortNewFiles, ctx, manifestFile,
manifestReadParallelism, true);
if (tailFiles.isEmpty()) {
return null;
@@ -1044,7 +1074,8 @@ public class ManifestFileSorter {
sortNewFiles,
ctx,
manifestFile,
- manifestReadParallelism);
+ manifestReadParallelism,
+ false);
candidates.clear();
candidatesSize = 0;
}
@@ -1058,7 +1089,8 @@ public class ManifestFileSorter {
sortNewFiles,
ctx,
manifestFile,
- manifestReadParallelism);
+ manifestReadParallelism,
+ false);
} else {
output.addAllUnchanged(candidates);
}
@@ -1077,10 +1109,13 @@ public class ManifestFileSorter {
List<ManifestFileMeta> sortNewFiles,
CompactionContext ctx,
ManifestFile manifestFile,
- @Nullable Integer manifestReadParallelism)
+ @Nullable Integer manifestReadParallelism,
+ boolean allowFullRewrite)
throws Exception {
// Skip rewrite for single file not in delete-range.
- if (section.size() == 1 &&
!ctx.defaultCompactFiles.getOrDefault(section.get(0), false)) {
+ if (section.size() == 1
+ && !(allowFullRewrite && ctx.fullSort)
+ && !ctx.defaultCompactFiles.getOrDefault(section.get(0),
false)) {
output.addUnchanged(section.get(0));
return;
}
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 6d67eddfbd..ad3641b121 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.fs.local.LocalFileIO;
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;
@@ -1352,6 +1353,278 @@ public class ManifestFileMetaTest extends
ManifestFileMetaTestBase {
.containsExactly(0, 1, 0, 1);
}
+ @Test
+ public void testManifestSortFullCompactionAlreadyCompactedRuns() {
+ List<ManifestFileMeta> 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<ManifestFileMeta> 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.MANIFEST_FULL_COMPACTION_FILE_SIZE.key(),
Long.MAX_VALUE + "B");
+ testOptions.set(CoreOptions.BUCKET, 4);
+
+ List<ManifestFileMeta> unchanged =
+ ManifestFileMerger.merge(
+ input,
+ manifestFile,
+ getPartitionType(),
+ CoreOptions.fromMap(testOptions.toMap()));
+ assertThat(unchanged).containsExactlyInAnyOrderElementsOf(input);
+
+ List<ManifestFileMeta> rewritten =
+ ManifestFileMergerTestUtils.fullMerge(
+ 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_FULL_COMPACTION_FILE_SIZE.key(),
Long.MAX_VALUE + "B");
+ List<ManifestFileMeta> afterMigration =
+ ManifestFileMerger.merge(
+ rewritten,
+ manifestFile,
+ getPartitionType(),
+ CoreOptions.fromMap(testOptions.toMap()));
+ assertThat(afterMigration).containsExactlyElementsOf(rewritten);
+ }
+
+ @Test
+ public void testManifestSortFullCompactionAllLevelRuns() {
+ List<ManifestFileMeta> 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<ManifestFileMeta> 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.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
+ // [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<ManifestFileMeta> unchanged =
+ ManifestFileMerger.merge(
+ input,
+ manifestFile,
+ getPartitionType(),
+ CoreOptions.fromMap(testOptions.toMap()));
+ assertThat(unchanged).containsExactlyInAnyOrderElementsOf(input);
+
+ List<ManifestFileMeta> rewritten =
+ ManifestFileMergerTestUtils.fullMerge(
+ 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 testManifestSortFullCompactionSingleManifest() {
+ 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.BUCKET, 4);
+ List<ManifestFileMeta> rewritten =
+ ManifestFileMergerTestUtils.fullMerge(
+ 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 testManifestSortFullCompactionRespectsRewriteLimit() {
+ long targetSize =
CoreOptions.MANIFEST_TARGET_FILE_SIZE.defaultValue().getBytes();
+ List<ManifestFileMeta> 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_MAX_REWRITE_SIZE.key(),
"1B");
+ testOptions.set(CoreOptions.BUCKET, 4);
+ List<ManifestFileMeta> rewritten =
+ ManifestFileMergerTestUtils.fullMerge(
+ input,
+ manifestFile,
+ getPartitionType(),
+ CoreOptions.fromMap(testOptions.toMap()));
+
+ Set<String> inputNames =
+
input.stream().map(ManifestFileMeta::fileName).collect(Collectors.toSet());
+ assertThat(rewritten)
+ .extracting(ManifestFileMeta::fileName)
+ .filteredOn(inputNames::contains)
+ .hasSize(2);
+ assertEquivalentEntries(input, rewritten);
+ }
+
+ @Test
+ public void
testManifestSortFullCompactionDoesNotExceedLimitForSingletonTail() {
+ long targetSize =
CoreOptions.MANIFEST_TARGET_FILE_SIZE.defaultValue().getBytes();
+ List<ManifestFileMeta> 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_MAX_REWRITE_SIZE.key(),
"1B");
+ testOptions.set(CoreOptions.BUCKET, 4);
+ List<ManifestFileMeta> rewritten =
+ ManifestFileMergerTestUtils.fullMerge(
+ input,
+ manifestFile,
+ getPartitionType(),
+ CoreOptions.fromMap(testOptions.toMap()));
+
+ Set<String> 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 testManifestSortFullCompactionDoesNotRewriteTailBeyondLimit() {
+ long targetSize =
CoreOptions.MANIFEST_TARGET_FILE_SIZE.defaultValue().getBytes();
+ List<ManifestFileMeta> 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_MAX_REWRITE_SIZE.key(),
"1B");
+ testOptions.set(CoreOptions.BUCKET, 8);
+ List<ManifestFileMeta> rewritten =
+ ManifestFileMergerTestUtils.fullMerge(
+ input,
+ manifestFile,
+ getPartitionType(),
+ CoreOptions.fromMap(testOptions.toMap()));
+
+ Set<String> inputNames =
+
input.stream().map(ManifestFileMeta::fileName).collect(Collectors.toSet());
+ assertThat(rewritten)
+ .extracting(ManifestFileMeta::fileName)
+ .filteredOn(inputNames::contains)
+ .hasSize(3);
+ assertEquivalentEntries(input, rewritten);
+ }
+
+ @Test
+ public void testManifestSortUnchangedSingletonDoesNotConsumeRewriteLimit()
{
+ long targetSize =
CoreOptions.MANIFEST_TARGET_FILE_SIZE.defaultValue().getBytes();
+ List<ManifestFileMeta> 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<ManifestFileMeta> merged =
+ ManifestFileMerger.merge(
+ input,
+ manifestFile,
+ getPartitionType(),
+ CoreOptions.fromMap(testOptions.toMap()));
+
+ Set<String> 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) {
@@ -2830,6 +3103,42 @@ public class ManifestFileMetaTest extends
ManifestFileMetaTestBase {
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-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
index c96b7b7111..437944fae5 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 class FileStoreCommitTest {
}
@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 cfd1cb8fc1..57b4dadec4 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 class ManifestFileMergerTest extends
ManifestFileMetaTestBase {
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 0000000000..87d93d1285
--- /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<ManifestFileMeta> fullMerge(
+ List<ManifestFileMeta> input,
+ ManifestFile manifestFile,
+ RowType partitionType,
+ CoreOptions options) {
+ return ManifestFileMerger.merge(input, manifestFile, partitionType,
options, null, true);
+ }
+}
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 a40fa8f905..a2e0f85fce 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
@@ -122,7 +122,7 @@ public class CompactManifestProcedureITCase extends
CatalogITCaseBase {
long compactSnapshotId = table.snapshotManager().latestSnapshot().id();
sql(procedure);
Assertions.assertThat(table.snapshotManager().latestSnapshot().id())
- .isEqualTo(compactSnapshotId);
+ .isEqualTo(compactSnapshotId + 1);
}
@Test
