diff --git a/docs/docs/flink/procedures/compaction.md b/docs/docs/flink/procedures/compaction.md index 71c2fd531d7b..5b26f1dfdab4 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 1244848028fc..9952a70cc552 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 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/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 87fd4611e72b..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 @@ -71,6 +71,7 @@ public class ManifestFileSorter { /** Context object that carries shared state across compaction methods. */ static class CompactionContext { final boolean fullCompaction; + final boolean fullSort; final boolean runMergeOptimizeEnabled; final ManifestSortKey sortKey; final RowType partitionType; @@ -91,6 +92,7 @@ static class CompactionContext { CompactionContext( boolean fullCompaction, + boolean fullSort, boolean runMergeOptimizeEnabled, ManifestSortKey sortKey, RowType partitionType, @@ -100,6 +102,7 @@ static class CompactionContext { List levelRuns, List pickedRuns) { this.fullCompaction = fullCompaction; + this.fullSort = fullSort; this.runMergeOptimizeEnabled = runMergeOptimizeEnabled; this.sortKey = sortKey; this.partitionType = partitionType; @@ -153,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; @@ -181,6 +185,7 @@ static List trySortCompaction( suggestedMetaSize, suggestedMinMetaCount, fullCompactionThreshold, + fullSort, maxRewriteSize, maxSizeAmplificationPercent, sortedRunSizeRatio, @@ -225,6 +230,7 @@ private static Optional> tryFullCompaction( long suggestedMetaSize, int suggestedMinMetaCount, long fullCompactionThreshold, + boolean fullSort, long maxRewriteSize, int maxSizeAmplificationPercent, int sortedRunSizeRatio, @@ -232,7 +238,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 (!fullSort + && !reachesFullCompactionThreshold( + input, suggestedMetaSize, fullCompactionThreshold)) { return Optional.empty(); } // Step 2: Prepare compaction context @@ -240,6 +248,7 @@ private static Optional> tryFullCompaction( prepareCompaction( input, true, + fullSort, manifestFile, partitionType, sortPartitionField, @@ -253,7 +262,8 @@ private static Optional> tryFullCompaction( manifestReadParallelism); try { List levelRuns = ctx.levelRuns; - List pickedRuns = ctx.pickedRuns; + List pickedRuns = + fullSort ? new ArrayList<>(levelRuns) : ctx.pickedRuns; if (pickedRuns.isEmpty() && ctx.defaultCompactFiles.isEmpty()) { LOG.debug( @@ -283,9 +293,22 @@ 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: 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 (fullSort) { + long totalSize = 0L; + boolean hasDefaultCompactFile = false; + for (ManifestFileMeta file : pickedFiles) { + totalSize += file.fileSize(); + hasDefaultCompactFile |= ctx.isMarkedForDefaultCompaction(file); + } + sections = new ArrayList<>(); + sections.add(new Section(pickedFiles, totalSize, hasDefaultCompactFile)); + } else { + sections = splitIntoSections(pickedFiles, ctx); + sections = mergeSmallAdjacentSections(sections, suggestedMetaSize); + } LOG.info( "Manifest sort full compact: pickedFiles={}, sections={}.", @@ -343,6 +366,7 @@ private static List tryMinorCompaction( prepareCompaction( input, false, + false, manifestFile, partitionType, sortPartitionField, @@ -458,6 +482,7 @@ private static List tryMinorCompaction( private static CompactionContext prepareCompaction( List input, boolean fullCompaction, + boolean fullSort, ManifestFile manifestFile, RowType partitionType, String sortPartitionField, @@ -500,6 +525,7 @@ private static CompactionContext prepareCompaction( return new CompactionContext( fullCompaction, + fullSort, useRunMergeOptimize, sortKey, partitionType, @@ -862,15 +888,17 @@ private static void rewriteSections( for (int i = 0; i < sections.size(); i++) { Section section = sections.get(i); - // A single-file section is always handled directly, regardless of the budget. - if (section.files.size() == 1) { + // Preserve the ordinary-compaction shortcut: an unchanged singleton must not consume + // the sort rewrite limit. Explicit full sort intentionally rewrites the singleton. + if (!ctx.fullSort && section.files.size() == 1) { rewriteSection( section.files, output, sortNewFiles, ctx, manifestFile, - manifestReadParallelism); + manifestReadParallelism, + false); continue; } @@ -886,7 +914,8 @@ private static void rewriteSections( sortNewFiles, ctx, manifestFile, - manifestReadParallelism); + manifestReadParallelism, + true); } else { // Phase 1b: first overflow -- split the section at the budget boundary, // rewrite the affordable head, and append the remaining tail back for later @@ -966,7 +995,8 @@ private static Section splitSectionAndRewriteHead( } } - rewriteSection(headFiles, output, sortNewFiles, ctx, manifestFile, manifestReadParallelism); + rewriteSection( + headFiles, output, sortNewFiles, ctx, manifestFile, manifestReadParallelism, true); if (tailFiles.isEmpty()) { return null; @@ -1044,7 +1074,8 @@ private static void unsortedCompactSection( sortNewFiles, ctx, manifestFile, - manifestReadParallelism); + manifestReadParallelism, + false); candidates.clear(); candidatesSize = 0; } @@ -1058,7 +1089,8 @@ private static void unsortedCompactSection( sortNewFiles, ctx, manifestFile, - manifestReadParallelism); + manifestReadParallelism, + false); } else { output.addAllUnchanged(candidates); } @@ -1077,10 +1109,13 @@ private static void rewriteSection( List 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 6d67eddfbdcd..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 @@ -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; @@ -1352,6 +1353,278 @@ public void testManifestSortUsesBucketAsPrimaryKeyForBucketedTable(int bucket) { .containsExactly(0, 1, 0, 1); } + @Test + public void testManifestSortFullCompactionAlreadyCompactedRuns() { + 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.MANIFEST_FULL_COMPACTION_FILE_SIZE.key(), Long.MAX_VALUE + "B"); + testOptions.set(CoreOptions.BUCKET, 4); + + List unchanged = + ManifestFileMerger.merge( + input, + manifestFile, + getPartitionType(), + CoreOptions.fromMap(testOptions.toMap())); + assertThat(unchanged).containsExactlyInAnyOrderElementsOf(input); + + List 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 afterMigration = + ManifestFileMerger.merge( + rewritten, + manifestFile, + getPartitionType(), + CoreOptions.fromMap(testOptions.toMap())); + assertThat(afterMigration).containsExactlyElementsOf(rewritten); + } + + @Test + public void testManifestSortFullCompactionAllLevelRuns() { + 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.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 unchanged = + ManifestFileMerger.merge( + input, + manifestFile, + getPartitionType(), + CoreOptions.fromMap(testOptions.toMap())); + assertThat(unchanged).containsExactlyInAnyOrderElementsOf(input); + + List 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 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 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 rewritten = + ManifestFileMergerTestUtils.fullMerge( + 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); + } + + @Test + public void testManifestSortFullCompactionDoesNotExceedLimitForSingletonTail() { + 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_MAX_REWRITE_SIZE.key(), "1B"); + testOptions.set(CoreOptions.BUCKET, 4); + List rewritten = + ManifestFileMergerTestUtils.fullMerge( + 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 testManifestSortFullCompactionDoesNotRewriteTailBeyondLimit() { + 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_MAX_REWRITE_SIZE.key(), "1B"); + testOptions.set(CoreOptions.BUCKET, 8); + List rewritten = + ManifestFileMergerTestUtils.fullMerge( + 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); + } + + @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) { @@ -2830,6 +3103,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-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-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..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 @@ -122,7 +122,7 @@ public void testManifestSortParameters() throws Exception { long compactSnapshotId = table.snapshotManager().latestSnapshot().id(); sql(procedure); Assertions.assertThat(table.snapshotManager().latestSnapshot().id()) - .isEqualTo(compactSnapshotId); + .isEqualTo(compactSnapshotId + 1); } @Test