diff --git a/docs/docs/flink/procedures.md b/docs/docs/flink/procedures.md index e3039f69b380..ebd22a6fb970 100644 --- a/docs/docs/flink/procedures.md +++ b/docs/docs/flink/procedures.md @@ -925,17 +925,22 @@ All available procedures are listed below. CALL [catalog.]sys.compact_manifest(`table` => 'identifier')
CALL [catalog.]sys.compact_manifest(`table` => 'identifier', 'options' => 'key1=value1,key2=value2')
- CALL [catalog.]sys.compact_manifest(`table` => 'identifier', `dry_run` => true) + CALL [catalog.]sys.compact_manifest(`table` => 'identifier', `dry_run` => true)
+ CALL [catalog.]sys.compact_manifest(`table` => 'identifier', `manifest_sort_enabled` => true, `manifest_sort_partition_field` => 'dt', `manifest_sort_max_rewrite_size` => '1 gb') To compact_manifest the manifests. Arguments:
  • table: the target table identifier. Cannot be empty.
  • options: the additional dynamic options of the table. It prioritizes higher than original `tableProp` and lower than `procedureArg`.
    • -
  • dry_run (Boolean, optional): when true, returns manifest metadata statistics without actually compacting.
    • +
  • dry_run (Boolean, optional): when true, returns manifest metadata statistics without actually compacting. When manifest sort is enabled, the result also contains the number of manifest files in each level built by manifest sort.
    • +
  • 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.
    • CALL sys.compact_manifest(`table` => 'default.T')
    - CALL sys.compact_manifest(`table` => 'default.T', `dry_run` => true) + CALL sys.compact_manifest(`table` => 'default.T', `dry_run` => true)
    + CALL sys.compact_manifest(`table` => 'default.T', `manifest_sort_enabled` => true, `manifest_sort_partition_field` => 'dt', `manifest_sort_max_rewrite_size` => '1 gb') diff --git a/docs/docs/spark/procedures.md b/docs/docs/spark/procedures.md index 33cdb9dc244f..a0db38a01aee 100644 --- a/docs/docs/spark/procedures.md +++ b/docs/docs/spark/procedures.md @@ -441,11 +441,15 @@ This section introduce all available spark procedures about paimon. To compact_manifest the manifests. Arguments:
  • table: the target table identifier. Cannot be empty.
  • options: the additional dynamic options of the table. It prioritizes higher than original `tableProp` and lower than `procedureArg`.
    • -
  • dry_run (Boolean, optional): when true, logs manifest metadata statistics without actually compacting. The result is printed to the application log; the SQL return value is still `true`.
    • +
  • dry_run (Boolean, optional): when true, logs manifest metadata statistics without actually compacting. When manifest sort is enabled, the log also contains the number of manifest files in each level built by manifest sort. The result is printed to the application log; the SQL return value is still `true`.
    • +
  • 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.
    • CALL sys.compact_manifest(`table` => 'default.T')
    - CALL sys.compact_manifest(`table` => 'default.T', dry_run => true) + CALL sys.compact_manifest(`table` => 'default.T', dry_run => true)
    + CALL sys.compact_manifest(`table` => 'default.T', manifest_sort_enabled => true, manifest_sort_partition_field => 'dt', manifest_sort_max_rewrite_size => '1 gb') 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 88d6083dec0d..ae4bd518b7dc 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 @@ -1629,16 +1629,12 @@ private boolean compactManifestOnce() { manifestList.readDataManifests(latestSnapshot); List mergeAfterManifests; - // the fist trial: use a copied options with forced full compaction settings - Options compactOptions = Options.fromMap(options.toMap()); - compactOptions.set(CoreOptions.MANIFEST_MERGE_MIN_COUNT, 1); - compactOptions.set(CoreOptions.MANIFEST_FULL_COMPACTION_FILE_SIZE, MemorySize.ofBytes(1)); mergeAfterManifests = ManifestFileMerger.merge( mergeBeforeManifests, manifestFile, partitionType, - new CoreOptions(compactOptions), + manifestCompactionOptions(options, mergeBeforeManifests, partitionType), ioManager); if (new HashSet<>(mergeBeforeManifests).equals(new HashSet<>(mergeAfterManifests))) { @@ -1677,6 +1673,20 @@ private boolean compactManifestOnce() { return commitSnapshotImpl(latestSnapshot, newSnapshot, emptyList()); } + static CoreOptions manifestCompactionOptions( + CoreOptions options, List manifests, RowType partitionType) { + // Use a copied options with forced full compaction settings for the legacy merge path. + // Manifest sort has its own full/minor picking strategy and should respect its configured + // thresholds. + Options compactOptions = Options.fromMap(options.toMap()); + if (!ManifestFileMerger.canUseManifestSort(manifests, partitionType, options)) { + compactOptions.set(CoreOptions.MANIFEST_MERGE_MIN_COUNT, 1); + compactOptions.set( + CoreOptions.MANIFEST_FULL_COMPACTION_FILE_SIZE, MemorySize.ofBytes(1)); + } + return new CoreOptions(compactOptions); + } + private boolean commitSnapshotImpl( @Nullable Snapshot baseSnapshot, Snapshot newSnapshot, 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 98c2f01eb727..4ab06f52bfdc 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 @@ -20,30 +20,33 @@ import org.apache.paimon.CoreOptions; import org.apache.paimon.Snapshot; +import org.apache.paimon.manifest.ManifestFile; import org.apache.paimon.manifest.ManifestFileMeta; import org.apache.paimon.manifest.ManifestList; import org.apache.paimon.options.MemorySize; import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.types.RowType; +import java.util.ArrayList; import java.util.List; /** Dry run for manifest compaction. Reads only existing metadata, never writes files. */ public class ManifestCompactDryRun { public static String execute(FileStoreTable table) { + CoreOptions options = new CoreOptions(table.options()); Snapshot latestSnapshot = table.store().snapshotManager().latestSnapshot(); if (latestSnapshot == null) { - return "Dry run: no snapshot exists."; + return appendEmptyManifestSortLevels("Dry run: no snapshot exists.", options); } ManifestList manifestList = table.store().manifestListFactory().create(); List manifests = manifestList.readDataManifests(latestSnapshot); if (manifests.isEmpty()) { - return "Dry run: 0 manifest files."; + return appendEmptyManifestSortLevels("Dry run: 0 manifest files.", options); } - CoreOptions options = new CoreOptions(table.options()); long suggestedMetaSize = options.manifestTargetSize().getBytes(); long totalFiles = manifests.size(); @@ -63,15 +66,113 @@ public static String execute(FileStoreTable table) { } } - return String.format( - "Dry run: %d manifest files (%s), " - + "%d deleted entries in %d files, " - + "%d undersized files (< %s).", - totalFiles, - MemorySize.ofBytes(totalSize), - totalDeletedEntries, - filesWithDeletedEntries, - smallFiles, - MemorySize.ofBytes(suggestedMetaSize)); + String summary = + String.format( + "Dry run: %d manifest files (%s), " + + "%d deleted entries in %d files, " + + "%d undersized files (< %s).", + totalFiles, + MemorySize.ofBytes(totalSize), + totalDeletedEntries, + filesWithDeletedEntries, + smallFiles, + MemorySize.ofBytes(suggestedMetaSize)); + + if (!options.manifestSortEnabled()) { + return summary; + } + + RowType partitionType = table.schema().logicalPartitionType(); + if (partitionType.getFieldCount() == 0 + && !(options.dataEvolutionEnabled() + && ManifestFileMeta.allContainsRowId(manifests))) { + return summary + " Manifest sort level files: unavailable (no sortable field)."; + } + + long[] levelFileCounts = new long[ManifestPickStrategy.MAX_LEVEL + 1]; + List levelRuns = + buildLevelSortedRunsForDryRun( + manifests, + table.store().manifestFileFactory().create(), + partitionType, + options); + for (ManifestAdjacentSortedRun run : levelRuns) { + levelFileCounts[run.level()] += run.files().size(); + } + + return appendManifestSortLevels(summary, levelFileCounts); + } + + private static List buildLevelSortedRunsForDryRun( + List manifests, + ManifestFile manifestFile, + RowType partitionType, + CoreOptions options) { + long suggestedMetaSize = options.manifestTargetSize().getBytes(); + boolean fullCompaction = + ManifestFileSorter.reachesFullCompactionThreshold( + manifests, + suggestedMetaSize, + options.manifestFullCompactionThresholdSize().getBytes()); + ManifestFileSorter.ManifestSortKey sortKey = + ManifestFileSorter.createSortKey( + options.dataEvolutionEnabled(), + manifests, + options.manifestSortPartitionField(), + partitionType); + ManifestFileSorter.ClassifyResult classifyResult = + ManifestFileSorter.classifyManifests( + manifests, + fullCompaction, + manifestFile, + partitionType, + suggestedMetaSize, + options.scanManifestParallelism()); + List levelRuns = buildLevelSortedRuns(classifyResult, sortKey); + + // A full compaction with no work falls through to the minor path. Mirror that fallback so + // the reported levels describe the path which a real compaction would use. + if (fullCompaction + && classifyResult.compactWithoutSort.isEmpty() + && new ManifestPickStrategy( + options.maxSizeAmplificationPercent(), options.sortedRunSizeRatio()) + .pick(levelRuns) + .isEmpty()) { + classifyResult = + ManifestFileSorter.classifyManifests( + manifests, + false, + manifestFile, + partitionType, + suggestedMetaSize, + options.scanManifestParallelism()); + levelRuns = buildLevelSortedRuns(classifyResult, sortKey); + } + return levelRuns; + } + + private static List buildLevelSortedRuns( + ManifestFileSorter.ClassifyResult classifyResult, + ManifestFileSorter.ManifestSortKey sortKey) { + return classifyResult.lsmFiles.isEmpty() + ? new ArrayList<>() + : ManifestFileSorter.buildLevelSortedRuns(classifyResult.lsmFiles, sortKey); + } + + private static String appendEmptyManifestSortLevels(String summary, CoreOptions options) { + return options.manifestSortEnabled() + ? appendManifestSortLevels(summary, new long[ManifestPickStrategy.MAX_LEVEL + 1]) + : summary; + } + + private static String appendManifestSortLevels(String summary, long[] levelFileCounts) { + return summary + + String.format( + " Manifest sort level files: L0=%d, L1=%d, L2=%d, L3=%d, L4=%d.", + levelFileCounts[0], + levelFileCounts[1], + levelFileCounts[2], + levelFileCounts[3], + levelFileCounts[4]); } } 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 0313b2b12b8d..65afa3c499b9 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 @@ -89,9 +89,7 @@ public static List merge( // If manifest-sort.enabled is enabled and there are sortable fields, use // trySortRewrite. Data evolution tables sort by RowID when all manifest files contain // RowID ranges, so they do not require partition fields. - if (options.manifestSortEnabled() - && (partitionType.getFieldCount() > 0 - || (options.dataEvolutionEnabled() && allContainsRowId(input)))) { + if (canUseManifestSort(input, partitionType, options)) { return ManifestFileSorter.trySortCompaction( input, newFilesForAbort, manifestFile, partitionType, options, ioManager); } else { @@ -124,6 +122,13 @@ public static List merge( } } + static boolean canUseManifestSort( + List input, RowType partitionType, CoreOptions options) { + return options.manifestSortEnabled() + && (partitionType.getFieldCount() > 0 + || (options.dataEvolutionEnabled() && allContainsRowId(input))); + } + private static List tryMinorCompaction( List input, List newFilesForAbort, 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 40a76a5914f0..5baefdc17011 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 @@ -110,7 +110,7 @@ boolean isMarkedForUnsortedCompaction(ManifestFileMeta file) { } /** Result of classifying manifest files. */ - private static class ClassifyResult { + static class ClassifyResult { final List lsmFiles; final DeletedIdentifierSet deleteEntries; /** @@ -228,13 +228,7 @@ private static Optional> tryFullCompaction( @Nullable Integer manifestReadParallelism) throws Exception { // Step 1: Check if full compaction threshold is met - long totalDeltaFileSize = 0; - for (ManifestFileMeta file : input) { - if (file.numDeletedFiles() > 0 || file.fileSize() < suggestedMetaSize) { - totalDeltaFileSize += file.fileSize(); - } - } - if (totalDeltaFileSize < fullCompactionThreshold) { + if (!reachesFullCompactionThreshold(input, suggestedMetaSize, fullCompactionThreshold)) { return Optional.empty(); } // Step 2: Prepare compaction context @@ -490,6 +484,17 @@ private static CompactionContext prepareCompaction( pickedRuns); } + static boolean reachesFullCompactionThreshold( + List input, long suggestedMetaSize, long fullCompactionThreshold) { + long totalDeltaFileSize = 0; + for (ManifestFileMeta file : input) { + if (file.numDeletedFiles() > 0 || file.fileSize() < suggestedMetaSize) { + totalDeltaFileSize += file.fileSize(); + } + } + return totalDeltaFileSize >= fullCompactionThreshold; + } + /** * Classify manifest files into default-compaction group and LSM group. * @@ -501,7 +506,7 @@ private static CompactionContext prepareCompaction( * * @return ClassifyResult containing lsmFiles, deleteEntries, and compactWithoutSort */ - private static ClassifyResult classifyManifests( + static ClassifyResult classifyManifests( List input, boolean fullCompaction, ManifestFile manifestFile, @@ -1118,7 +1123,7 @@ private static boolean containsNoDeleteEntries(List section) { return true; } - private static ManifestSortKey createSortKey( + static ManifestSortKey createSortKey( boolean dataEvolutionEnabled, List input, String sortPartitionField, 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 7dff697e8fc6..34df06767cc5 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 @@ -791,7 +791,7 @@ public RowType getPartitionType() { } @Override - ManifestFile getManifestFile() { + protected ManifestFile getManifestFile() { return manifestFile; } @@ -961,6 +961,62 @@ public void testManifestSortWithOverlappingPartitions() { } } + @Test + public void testManifestSortMinorCompactionRespectsMergeMinCount() { + List input = new ArrayList<>(); + for (int i = 0; i < 4; i++) { + input.add(makeManifest(makeEntry(true, "file-" + i, 0))); + } + + Options testOptions = new Options(); + testOptions.set(CoreOptions.MANIFEST_SORT_ENABLED, true); + testOptions.set(CoreOptions.MANIFEST_TARGET_FILE_SIZE.key(), "1G"); + testOptions.set(CoreOptions.MANIFEST_MERGE_MIN_COUNT, 100); + testOptions.set(CoreOptions.MANIFEST_FULL_COMPACTION_FILE_SIZE.key(), Long.MAX_VALUE + "B"); + testOptions.set(CoreOptions.MANIFEST_SORT_MAX_REWRITE_SIZE.key(), "1B"); + + List merged = + ManifestFileMerger.merge( + input, + manifestFile, + getPartitionType(), + CoreOptions.fromMap(testOptions.toMap())); + + Set inputManifestNames = + input.stream().map(ManifestFileMeta::fileName).collect(Collectors.toSet()); + Set retainedInputManifestNames = + merged.stream() + .map(ManifestFileMeta::fileName) + .filter(inputManifestNames::contains) + .collect(Collectors.toSet()); + assertThat(retainedInputManifestNames).hasSize(2); + assertEquivalentEntries(input, merged); + } + + @Test + public void testManifestSortRespectsFullCompactionThreshold() { + 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_TARGET_FILE_SIZE.key(), "1B"); + testOptions.set(CoreOptions.MANIFEST_FULL_COMPACTION_FILE_SIZE.key(), Long.MAX_VALUE + "B"); + + List merged = + ManifestFileMerger.merge( + input, + manifestFile, + getPartitionType(), + CoreOptions.fromMap(testOptions.toMap())); + + assertThat(merged).containsExactlyElementsOf(input); + assertThat(readEntries(merged)).anyMatch(entry -> entry.kind() == FileKind.DELETE); + } + @Test public void testManifestSortMaxRewriteSizeSmallerThanTargetFileSizeStillRewrites() { List input = new ArrayList<>(); diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTestBase.java b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTestBase.java index b336bd8dd505..649ebb73ec99 100644 --- a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTestBase.java +++ b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTestBase.java @@ -105,9 +105,9 @@ protected ManifestFileMeta makeManifest(ManifestEntry... entries) { return getManifestFile().write(Arrays.asList(entries)).get(0); } - abstract ManifestFile getManifestFile(); + protected abstract ManifestFile getManifestFile(); - abstract RowType getPartitionType(); + protected abstract RowType getPartitionType(); protected void assertEquivalentEntries( List input, List merged) { 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 f326be18dcab..826f3d06d491 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 @@ -50,6 +50,7 @@ import org.apache.paimon.operation.commit.ConflictDetection; import org.apache.paimon.operation.commit.ManifestEntryChanges; import org.apache.paimon.operation.commit.RetryCommitResult; +import org.apache.paimon.options.Options; import org.apache.paimon.predicate.PredicateBuilder; import org.apache.paimon.schema.Schema; import org.apache.paimon.schema.SchemaManager; @@ -1282,6 +1283,24 @@ public void testManifestCompact() throws Exception { .isEqualTo(0); } + @Test + public void testManifestSortCompactManifestRespectsCompactionThresholds() { + 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); + + assertThat(compactOptions.manifestMergeMinCount()).isEqualTo(100); + assertThat(compactOptions.manifestFullCompactionThresholdSize().getBytes()) + .isEqualTo(Long.MAX_VALUE); + } + @Test public void testRtasAppendAfterTruncateResetsInheritedIndexAndStats() throws Exception { TestFileStore store = createStore(false, 1, CoreOptions.ChangelogProducer.NONE); 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 new file mode 100644 index 000000000000..7df1318902ca --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/operation/ManifestFileMergerTest.java @@ -0,0 +1,102 @@ +/* + * 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.FileKind; +import org.apache.paimon.manifest.ManifestEntry; +import org.apache.paimon.manifest.ManifestFile; +import org.apache.paimon.manifest.ManifestFileMeta; +import org.apache.paimon.manifest.ManifestFileMetaTestBase; +import org.apache.paimon.options.Options; +import org.apache.paimon.types.RowType; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.util.Arrays; +import java.util.List; +import java.util.stream.Collectors; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for {@link ManifestFileMerger}. */ +public class ManifestFileMergerTest extends ManifestFileMetaTestBase { + + private static final RowType NO_PARTITION_TYPE = RowType.of(); + + @TempDir java.nio.file.Path tempDir; + private ManifestFile manifestFile; + + @BeforeEach + public void beforeEach() { + manifestFile = createManifestFile(tempDir.toString()); + } + + @Test + public void testManifestSortFallsBackToForcedLegacyMergeWithoutRowId() { + List input = + Arrays.asList( + makeManifest(makeEntry(true, "base", null)), + makeManifest( + makeEntry(false, "base", null), + makeEntry(true, "replacement", null))); + + Options options = new Options(); + options.set(CoreOptions.MANIFEST_SORT_ENABLED, true); + options.set(CoreOptions.DATA_EVOLUTION_ENABLED, true); + options.set(CoreOptions.MANIFEST_MERGE_MIN_COUNT, 100); + options.set(CoreOptions.MANIFEST_FULL_COMPACTION_FILE_SIZE.key(), Long.MAX_VALUE + "B"); + CoreOptions tableOptions = new CoreOptions(options); + + assertThat(ManifestFileMerger.canUseManifestSort(input, NO_PARTITION_TYPE, tableOptions)) + .isFalse(); + + CoreOptions compactOptions = + FileStoreCommitImpl.manifestCompactionOptions( + tableOptions, input, NO_PARTITION_TYPE); + assertThat(compactOptions.manifestMergeMinCount()).isEqualTo(1); + assertThat(compactOptions.manifestFullCompactionThresholdSize().getBytes()).isEqualTo(1); + + List merged = + ManifestFileMerger.merge(input, manifestFile, NO_PARTITION_TYPE, compactOptions); + List mergedEntries = + merged.stream() + .flatMap( + meta -> + manifestFile.read(meta.fileName(), meta.fileSize()) + .stream()) + .collect(Collectors.toList()); + assertThat(mergedEntries).noneMatch(entry -> entry.kind() == FileKind.DELETE); + assertThat(mergedEntries) + .extracting(entry -> entry.file().fileName()) + .containsExactly("replacement"); + } + + @Override + public ManifestFile getManifestFile() { + return manifestFile; + } + + @Override + public RowType getPartitionType() { + return NO_PARTITION_TYPE; + } +} 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 new file mode 100644 index 000000000000..4b70de86dbf3 --- /dev/null +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactManifestAction.java @@ -0,0 +1,77 @@ +/* + * 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.flink.action; + +import org.apache.paimon.flink.procedure.CompactManifestProcedure; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Map; + +/** Compact manifest action for Flink. */ +public class CompactManifestAction extends ActionBase implements LocalAction { + + private static final Logger LOG = LoggerFactory.getLogger(CompactManifestAction.class); + + private final String database; + private final String table; + private final String options; + private final Boolean dryRun; + private final Boolean manifestSortEnabled; + private final String manifestSortPartitionField; + private final String manifestSortMaxRewriteSize; + + public CompactManifestAction( + String database, + String table, + Map catalogConfig, + String options, + Boolean dryRun, + Boolean manifestSortEnabled, + String manifestSortPartitionField, + String manifestSortMaxRewriteSize) { + super(catalogConfig); + this.database = database; + this.table = table; + this.options = options; + this.dryRun = dryRun; + this.manifestSortEnabled = manifestSortEnabled; + this.manifestSortPartitionField = manifestSortPartitionField; + this.manifestSortMaxRewriteSize = manifestSortMaxRewriteSize; + } + + @Override + public void executeLocally() throws Exception { + CompactManifestProcedure procedure = new CompactManifestProcedure(); + procedure.withCatalog(catalog); + String[] results = + procedure.call( + null, + database + "." + table, + options, + dryRun, + manifestSortEnabled, + manifestSortPartitionField, + manifestSortMaxRewriteSize); + for (String result : results) { + LOG.info(result); + } + } +} diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactManifestActionFactory.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactManifestActionFactory.java new file mode 100644 index 000000000000..be8314a12239 --- /dev/null +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactManifestActionFactory.java @@ -0,0 +1,81 @@ +/* + * 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.flink.action; + +import java.util.Optional; + +/** Factory to create {@link CompactManifestAction}. */ +public class CompactManifestActionFactory implements ActionFactory { + + public static final String IDENTIFIER = "compact_manifest"; + + private static final String OPTIONS = "options"; + private static final String DRY_RUN = "dry_run"; + private static final String MANIFEST_SORT_ENABLED = "manifest_sort.enabled"; + private static final String MANIFEST_SORT_PARTITION_FIELD = "manifest_sort.partition_field"; + private static final String MANIFEST_SORT_MAX_REWRITE_SIZE = "manifest_sort.max_rewrite_size"; + + @Override + public String identifier() { + return IDENTIFIER; + } + + @Override + public Optional create(MultipleParameterToolAdapter params) { + CompactManifestAction action = + new CompactManifestAction( + params.getRequired(DATABASE), + params.getRequired(TABLE), + catalogConfigMap(params), + params.get(OPTIONS), + params.getBoolean(DRY_RUN, false), + params.getBoolean(MANIFEST_SORT_ENABLED, null), + params.get(MANIFEST_SORT_PARTITION_FIELD), + params.get(MANIFEST_SORT_MAX_REWRITE_SIZE)); + return Optional.of(action); + } + + @Override + public void printHelp() { + System.out.println( + "Action \"compact_manifest\" compacts manifest files of the specified table."); + System.out.println(); + + System.out.println("Syntax:"); + System.out.println( + " compact_manifest \\\n" + + "--warehouse \\\n" + + "--database \\\n" + + "--table \\\n" + + "[--options ] \\\n" + + "[--dry_run ] \\\n" + + "[--manifest-sort.enabled ] \\\n" + + "[--manifest-sort.partition-field ] \\\n" + + "[--manifest-sort.max-rewrite-size ]"); + System.out.println(); + + System.out.println("Example:"); + System.out.println( + " compact_manifest --warehouse s3://path/to/warehouse \\\n" + + "--database default --table T \\\n" + + "--manifest-sort.enabled true \\\n" + + "--manifest-sort.partition-field dt \\\n" + + "--manifest-sort.max-rewrite-size 1gb"); + } +} 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 f1c9156e319f..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 @@ -47,13 +47,28 @@ public String identifier() { argument = { @ArgumentHint(name = "table", type = @DataTypeHint("STRING")), @ArgumentHint(name = "options", type = @DataTypeHint("STRING"), isOptional = true), - @ArgumentHint(name = "dry_run", type = @DataTypeHint("BOOLEAN"), isOptional = true) + @ArgumentHint(name = "dry_run", type = @DataTypeHint("BOOLEAN"), isOptional = true), + @ArgumentHint( + name = "manifest_sort_enabled", + type = @DataTypeHint("BOOLEAN"), + isOptional = true), + @ArgumentHint( + name = "manifest_sort_partition_field", + type = @DataTypeHint("STRING"), + isOptional = true), + @ArgumentHint( + name = "manifest_sort_max_rewrite_size", + type = @DataTypeHint("STRING"), + isOptional = true) }) public String[] call( ProcedureContext procedureContext, String tableId, @Nullable String options, - @Nullable Boolean dryRun) + @Nullable Boolean dryRun, + @Nullable Boolean manifestSortEnabled, + @Nullable String manifestSortPartitionField, + @Nullable String manifestSortMaxRewriteSize) throws Exception { FileStoreTable table = (FileStoreTable) table(tableId); @@ -61,6 +76,18 @@ public String[] call( ProcedureUtils.putIfNotEmpty( dynamicOptions, CoreOptions.COMMIT_USER_PREFIX.key(), COMMIT_USER); ProcedureUtils.putAllOptions(dynamicOptions, options); + if (manifestSortEnabled != null) { + dynamicOptions.put( + CoreOptions.MANIFEST_SORT_ENABLED.key(), Boolean.toString(manifestSortEnabled)); + } + if (manifestSortPartitionField != null) { + dynamicOptions.put( + CoreOptions.MANIFEST_SORT_PARTITION_FIELD.key(), manifestSortPartitionField); + } + if (manifestSortMaxRewriteSize != null) { + dynamicOptions.put( + CoreOptions.MANIFEST_SORT_MAX_REWRITE_SIZE.key(), manifestSortMaxRewriteSize); + } table = table.copy(dynamicOptions); diff --git a/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory b/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory index e0eee1005a76..7ad4149df360 100644 --- a/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory +++ b/paimon-flink/paimon-flink-common/src/main/resources/META-INF/services/org.apache.paimon.factories.Factory @@ -16,6 +16,7 @@ ### action factories org.apache.paimon.flink.action.CopyFilesActionFactory org.apache.paimon.flink.action.CompactActionFactory +org.apache.paimon.flink.action.CompactManifestActionFactory org.apache.paimon.flink.action.CompactDatabaseActionFactory org.apache.paimon.flink.action.DropPartitionActionFactory org.apache.paimon.flink.action.DeleteActionFactory 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 cd74fc682d6d..a40fa8f905f6 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 @@ -19,11 +19,16 @@ package org.apache.paimon.flink.procedure; import org.apache.paimon.flink.CatalogITCaseBase; +import org.apache.paimon.flink.action.ActionFactory; +import org.apache.paimon.flink.action.CompactManifestAction; +import org.apache.paimon.table.FileStoreTable; import org.assertj.core.api.Assertions; import org.junit.jupiter.api.Test; import java.util.Objects; +import java.util.regex.Matcher; +import java.util.regex.Pattern; /** IT Case for {@link CompactManifestProcedure}. */ public class CompactManifestProcedureITCase extends CatalogITCaseBase { @@ -76,6 +81,130 @@ public void testManifestCompactProcedure() { "[+I[1, 101, 15, 20221208], +I[4, 1001, 16, 20221208], +I[5, 10001, 15, 20221209]]"); } + @Test + public void testManifestSortParameters() throws Exception { + sql( + "CREATE TABLE T_SORT (" + + " k INT," + + " v STRING," + + " dt STRING" + + ") PARTITIONED BY (dt) WITH (" + + " 'write-only' = 'true'," + + " 'manifest.full-compaction-threshold-size' = '10000 T'," + + " 'bucket' = '-1'" + + ")"); + + sql("INSERT INTO T_SORT VALUES (1, '10', '20221208'), (2, '20', '20221209')"); + sql("INSERT OVERWRITE T_SORT VALUES (1, '11', '20221208'), (2, '21', '20221209')"); + + Assertions.assertThat( + sql("SELECT sum(num_deleted_files) FROM T_SORT$manifests") + .get(0) + .getField(0)) + .isEqualTo(2L); + + String procedure = + "CALL sys.compact_manifest(" + + "`table` => 'default.T_SORT', " + + "`options` => 'manifest-sort.partition-field=missing', " + + "`manifest_sort_enabled` => true, " + + "`manifest_sort_partition_field` => 'dt', " + + "`manifest_sort_max_rewrite_size` => '1 gb')"; + sql(procedure); + + Assertions.assertThat( + sql("SELECT sum(num_deleted_files) FROM T_SORT$manifests") + .get(0) + .getField(0)) + .isEqualTo(0L); + + FileStoreTable table = paimonTable("T_SORT"); + long compactSnapshotId = table.snapshotManager().latestSnapshot().id(); + sql(procedure); + Assertions.assertThat(table.snapshotManager().latestSnapshot().id()) + .isEqualTo(compactSnapshotId); + } + + @Test + public void testManifestSortParametersValidation() { + sql( + "CREATE TABLE T_INVALID (k INT, dt STRING) PARTITIONED BY (dt) WITH (" + + " 'bucket' = '-1'" + + ")"); + + Assertions.assertThatThrownBy( + () -> + sql( + "CALL sys.compact_manifest(" + + "`table` => 'default.T_INVALID', " + + "`manifest_sort_enabled` => true, " + + "`manifest_sort_partition_field` => 'missing')")) + .hasStackTraceContaining( + "'manifest-sort.partition-field' = 'missing' is not a partition field"); + } + + @Test + public void testManifestCompactWithoutSnapshotDoesNotCommit() throws Exception { + sql( + "CREATE TABLE T_EMPTY (k INT, dt STRING) PARTITIONED BY (dt) WITH (" + + " 'bucket' = '-1'" + + ")"); + FileStoreTable table = paimonTable("T_EMPTY"); + Assertions.assertThat(table.snapshotManager().latestSnapshot()).isNull(); + + sql( + "CALL sys.compact_manifest(" + + "`table` => 'default.T_EMPTY', " + + "`manifest_sort_enabled` => true, " + + "`manifest_sort_partition_field` => 'dt')"); + + Assertions.assertThat(table.snapshotManager().latestSnapshot()).isNull(); + } + + @Test + public void testManifestCompactActionWithManifestSort() throws Exception { + sql( + "CREATE TABLE T_ACTION (" + + " k INT," + + " v STRING," + + " dt STRING" + + ") PARTITIONED BY (dt) WITH (" + + " 'write-only' = 'true'," + + " 'manifest.full-compaction-threshold-size' = '10000 T'," + + " 'bucket' = '-1'" + + ")"); + sql("INSERT INTO T_ACTION VALUES (1, '10', '20221208'), (2, '20', '20221209')"); + sql("INSERT OVERWRITE T_ACTION VALUES (1, '11', '20221208'), (2, '21', '20221209')"); + + CompactManifestAction action = + ActionFactory.createAction( + new String[] { + "compact_manifest", + "--warehouse", + path, + "--database", + "default", + "--table", + "T_ACTION", + "--manifest-sort.enabled", + "true", + "--manifest-sort.partition-field", + "dt", + "--manifest-sort.max-rewrite-size", + "1gb" + }) + .filter(CompactManifestAction.class::isInstance) + .map(CompactManifestAction.class::cast) + .orElseThrow(() -> new RuntimeException("Failed to create action")); + action.run(); + + Assertions.assertThat( + sql("SELECT sum(num_deleted_files) FROM T_ACTION$manifests") + .get(0) + .getField(0)) + .isEqualTo(0L); + } + @Test public void testManifestCompactProcedureWithBranch() { sql( @@ -160,13 +289,29 @@ public void testManifestCompactDryRun() { String dryRunResult = Objects.requireNonNull( - sql("CALL sys.compact_manifest(`table` => 'default.T', `dry_run` => true)") + sql("CALL sys.compact_manifest(" + + "`table` => 'default.T', " + + "`options` => 'manifest.target-file-size=1B', " + + "`dry_run` => true, " + + "`manifest_sort_enabled` => true, " + + "`manifest_sort_partition_field` => 'dt')") .get(0) .getField(0)) .toString(); Assertions.assertThat(dryRunResult).startsWith("Dry run:"); Assertions.assertThat(dryRunResult).contains("deleted entries in"); + Matcher levelCounts = + Pattern.compile( + "Manifest sort level files: L0=(\\d+), L1=(\\d+), L2=(\\d+), L3=(\\d+), L4=(\\d+)\\.") + .matcher(dryRunResult); + Assertions.assertThat(levelCounts.find()).isTrue(); + long leveledManifestFiles = 0; + for (int i = 1; i <= 5; i++) { + leveledManifestFiles += Long.parseLong(levelCounts.group(i)); + } + Assertions.assertThat(leveledManifestFiles) + .isEqualTo(sql("SELECT count(*) FROM T$manifests").get(0).getField(0)); // verify dry run did not actually compact Assertions.assertThat( 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 cc1a42f9e52c..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 @@ -18,6 +18,7 @@ package org.apache.paimon.spark.procedure; +import org.apache.paimon.CoreOptions; import org.apache.paimon.operation.ManifestCompactDryRun; import org.apache.paimon.table.FileStoreTable; import org.apache.paimon.table.Table; @@ -55,7 +56,10 @@ public class CompactManifestProcedure extends BaseProcedure { new ProcedureParameter[] { ProcedureParameter.required("table", StringType), ProcedureParameter.optional("options", StringType), - ProcedureParameter.optional("dry_run", BooleanType) + 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) }; private static final StructType OUTPUT_TYPE = @@ -84,10 +88,25 @@ public InternalRow[] call(InternalRow args) { Identifier tableIdent = toIdentifier(args.getString(0), PARAMETERS[0].name()); String options = args.isNullAt(1) ? null : args.getString(1); boolean dryRun = !args.isNullAt(2) && args.getBoolean(2); + 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); Table table = loadSparkTable(tableIdent).getTable(); HashMap dynamicOptions = new HashMap<>(); ProcedureUtils.putAllOptions(dynamicOptions, options); + if (manifestSortEnabled != null) { + dynamicOptions.put( + CoreOptions.MANIFEST_SORT_ENABLED.key(), Boolean.toString(manifestSortEnabled)); + } + if (manifestSortPartitionField != null) { + dynamicOptions.put( + CoreOptions.MANIFEST_SORT_PARTITION_FIELD.key(), manifestSortPartitionField); + } + if (manifestSortMaxRewriteSize != null) { + dynamicOptions.put( + CoreOptions.MANIFEST_SORT_MAX_REWRITE_SIZE.key(), manifestSortMaxRewriteSize); + } 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 7ef982b085ee..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 @@ -63,7 +63,13 @@ class CompactManifestProcedureTest extends PaimonSparkTestBase with StreamTest { Assertions.assertThat(deletedBefore).isGreaterThan(0L) val dryRunRows = spark - .sql("CALL sys.compact_manifest(table => 'T2', dry_run => true)") + .sql( + "CALL sys.compact_manifest(" + + "table => 'T2', " + + "dry_run => true, " + + "manifest_sort_enabled => true, " + + "manifest_sort_partition_field => 'dt', " + + "manifest_sort_max_rewrite_size => '1gb')") .collectAsList() Assertions.assertThat(dryRunRows.get(0).getBoolean(0)).isTrue