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 55d3511b69 [flink][spark] Support manifest sort in compact_manifest
(#9083)
55d3511b69 is described below
commit 55d3511b699e047ac0801c4bc5306c962104bbbf
Author: umi <[email protected]>
AuthorDate: Sat Aug 15 14:35:50 2026 +0800
[flink][spark] Support manifest sort in compact_manifest (#9083)
---
docs/docs/flink/procedures.md | 11 +-
docs/docs/spark/procedures.md | 8 +-
.../paimon/operation/FileStoreCommitImpl.java | 20 ++-
.../paimon/operation/ManifestCompactDryRun.java | 127 ++++++++++++++++--
.../paimon/operation/ManifestFileMerger.java | 11 +-
.../paimon/operation/ManifestFileSorter.java | 25 ++--
.../paimon/manifest/ManifestFileMetaTest.java | 58 +++++++-
.../paimon/manifest/ManifestFileMetaTestBase.java | 4 +-
.../paimon/operation/FileStoreCommitTest.java | 19 +++
.../paimon/operation/ManifestFileMergerTest.java | 102 ++++++++++++++
.../paimon/flink/action/CompactManifestAction.java | 77 +++++++++++
.../flink/action/CompactManifestActionFactory.java | 81 ++++++++++++
.../flink/procedure/CompactManifestProcedure.java | 31 ++++-
.../services/org.apache.paimon.factories.Factory | 1 +
.../procedure/CompactManifestProcedureITCase.java | 147 ++++++++++++++++++++-
.../spark/procedure/CompactManifestProcedure.java | 21 ++-
.../procedure/CompactManifestProcedureTest.scala | 8 +-
17 files changed, 707 insertions(+), 44 deletions(-)
diff --git a/docs/docs/flink/procedures.md b/docs/docs/flink/procedures.md
index 79e735f39b..60ca8ded94 100644
--- a/docs/docs/flink/procedures.md
+++ b/docs/docs/flink/procedures.md
@@ -945,17 +945,22 @@ All available procedures are listed below.
<td>
CALL [catalog.]sys.compact_manifest(`table` => 'identifier')<br/>
CALL [catalog.]sys.compact_manifest(`table` => 'identifier',
'options' => 'key1=value1,key2=value2')<br/>
- CALL [catalog.]sys.compact_manifest(`table` => 'identifier',
`dry_run` => true)
+ CALL [catalog.]sys.compact_manifest(`table` => 'identifier',
`dry_run` => true)<br/>
+ CALL [catalog.]sys.compact_manifest(`table` => 'identifier',
`manifest_sort_enabled` => true, `manifest_sort_partition_field` => 'dt',
`manifest_sort_max_rewrite_size` => '1 gb')
</td>
<td>
To compact_manifest the manifests. Arguments:
<li>table: the target table identifier. Cannot be empty.</li>
<li>options: the additional dynamic options of the table. It
prioritizes higher than original `tableProp` and lower than `procedureArg`.</li>
- <li>dry_run (Boolean, optional): when true, returns manifest
metadata statistics without actually compacting.</li>
+ <li>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.</li>
+ <li>manifest_sort_enabled (Boolean, optional): whether to use
manifest sort rewrite for this invocation.</li>
+ <li>manifest_sort_partition_field (String, optional): partition
field used to sort manifest entries. Defaults to the first partition field.</li>
+ <li>manifest_sort_max_rewrite_size (String, optional): maximum
manifest size rewritten by one sort pass.</li>
</td>
<td>
CALL sys.compact_manifest(`table` => 'default.T')<br/>
- CALL sys.compact_manifest(`table` => 'default.T', `dry_run` => true)
+ CALL sys.compact_manifest(`table` => 'default.T', `dry_run` =>
true)<br/>
+ CALL sys.compact_manifest(`table` => 'default.T',
`manifest_sort_enabled` => true, `manifest_sort_partition_field` => 'dt',
`manifest_sort_max_rewrite_size` => '1 gb')
</td>
</tr>
<tr>
diff --git a/docs/docs/spark/procedures.md b/docs/docs/spark/procedures.md
index bcd046fb73..005c30d687 100644
--- a/docs/docs/spark/procedures.md
+++ b/docs/docs/spark/procedures.md
@@ -466,11 +466,15 @@ This section introduce all available spark procedures
about paimon.
To compact_manifest the manifests. Arguments:
<li>table: the target table identifier. Cannot be empty.</li>
<li>options: the additional dynamic options of the table. It
prioritizes higher than original `tableProp` and lower than `procedureArg`.</li>
- <li>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`.</li>
+ <li>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`.</li>
+ <li>manifest_sort_enabled (Boolean, optional): whether to use
manifest sort rewrite for this invocation.</li>
+ <li>manifest_sort_partition_field (String, optional): partition
field used to sort manifest entries. Defaults to the first partition field.</li>
+ <li>manifest_sort_max_rewrite_size (String, optional): maximum
manifest size rewritten by one sort pass.</li>
</td>
<td>
CALL sys.compact_manifest(`table` => 'default.T')<br/>
- CALL sys.compact_manifest(`table` => 'default.T', dry_run => true)
+ CALL sys.compact_manifest(`table` => 'default.T', dry_run =>
true)<br/>
+ CALL sys.compact_manifest(`table` => 'default.T',
manifest_sort_enabled => true, manifest_sort_partition_field => 'dt',
manifest_sort_max_rewrite_size => '1 gb')
</td>
</tr>
<tr>
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 eb6ef13cef..505faba772 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
@@ -1589,16 +1589,12 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
manifestList.readDataManifests(latestSnapshot);
List<ManifestFileMeta> 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))) {
@@ -1637,6 +1633,20 @@ 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.
+ 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 98c2f01eb7..4ab06f52bf 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 @@ package org.apache.paimon.operation;
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<ManifestFileMeta> 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 class ManifestCompactDryRun {
}
}
- 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<ManifestAdjacentSortedRun> 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<ManifestAdjacentSortedRun>
buildLevelSortedRunsForDryRun(
+ List<ManifestFileMeta> 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<ManifestAdjacentSortedRun> 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<ManifestAdjacentSortedRun> 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 7c5019f1e3..a92efa86d8 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 class ManifestFileMerger {
// 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 class ManifestFileMerger {
}
}
+ static boolean canUseManifestSort(
+ List<ManifestFileMeta> input, RowType partitionType, CoreOptions
options) {
+ return options.manifestSortEnabled()
+ && (partitionType.getFieldCount() > 0
+ || (options.dataEvolutionEnabled() &&
allContainsRowId(input)));
+ }
+
private static List<ManifestFileMeta> tryMinorCompaction(
List<ManifestFileMeta> input,
List<ManifestFileMeta> 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 6ca30b6444..75eb9e7e63 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 @@ public class ManifestFileSorter {
}
/** Result of classifying manifest files. */
- private static class ClassifyResult {
+ static class ClassifyResult {
final List<ManifestFileMeta> lsmFiles;
final CompactFileIdentifierSet deleteEntries;
/**
@@ -228,13 +228,7 @@ public class ManifestFileSorter {
@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 @@ public class ManifestFileSorter {
pickedRuns);
}
+ static boolean reachesFullCompactionThreshold(
+ List<ManifestFileMeta> 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 @@ public class ManifestFileSorter {
*
* @return ClassifyResult containing lsmFiles, deleteEntries, and
compactWithoutSort
*/
- private static ClassifyResult classifyManifests(
+ static ClassifyResult classifyManifests(
List<ManifestFileMeta> input,
boolean fullCompaction,
ManifestFile manifestFile,
@@ -1116,7 +1121,7 @@ public class ManifestFileSorter {
return true;
}
- private static ManifestSortKey createSortKey(
+ static ManifestSortKey createSortKey(
boolean dataEvolutionEnabled,
List<ManifestFileMeta> 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 7dff697e8f..34df06767c 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 class ManifestFileMetaTest extends
ManifestFileMetaTestBase {
}
@Override
- ManifestFile getManifestFile() {
+ protected ManifestFile getManifestFile() {
return manifestFile;
}
@@ -961,6 +961,62 @@ public class ManifestFileMetaTest extends
ManifestFileMetaTestBase {
}
}
+ @Test
+ public void testManifestSortMinorCompactionRespectsMergeMinCount() {
+ List<ManifestFileMeta> 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<ManifestFileMeta> merged =
+ ManifestFileMerger.merge(
+ input,
+ manifestFile,
+ getPartitionType(),
+ CoreOptions.fromMap(testOptions.toMap()));
+
+ Set<String> inputManifestNames =
+
input.stream().map(ManifestFileMeta::fileName).collect(Collectors.toSet());
+ Set<String> retainedInputManifestNames =
+ merged.stream()
+ .map(ManifestFileMeta::fileName)
+ .filter(inputManifestNames::contains)
+ .collect(Collectors.toSet());
+ assertThat(retainedInputManifestNames).hasSize(2);
+ assertEquivalentEntries(input, merged);
+ }
+
+ @Test
+ public void testManifestSortRespectsFullCompactionThreshold() {
+ List<ManifestFileMeta> 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<ManifestFileMeta> 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<ManifestFileMeta> 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 b336bd8dd5..649ebb73ec 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 @@ public abstract class ManifestFileMetaTestBase {
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<ManifestFileMeta> input, List<ManifestFileMeta> 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 d427345fa7..3297383dcb 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.CommitChanges;
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 class FileStoreCommitTest {
.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 0000000000..7df1318902
--- /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<ManifestFileMeta> 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<ManifestFileMeta> merged =
+ ManifestFileMerger.merge(input, manifestFile,
NO_PARTITION_TYPE, compactOptions);
+ List<ManifestEntry> 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 0000000000..4b70de86db
--- /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<String, String> 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 0000000000..be8314a122
--- /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<Action> 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 <warehouse_path> \\\n"
+ + "--database <database_name> \\\n"
+ + "--table <table_name> \\\n"
+ + "[--options <key1=value1,key2=value2>] \\\n"
+ + "[--dry_run <true|false>] \\\n"
+ + "[--manifest-sort.enabled <true|false>] \\\n"
+ + "[--manifest-sort.partition-field <partition_field>]
\\\n"
+ + "[--manifest-sort.max-rewrite-size <memory_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 f1c9156e31..1bae106b9f 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 class CompactManifestProcedure extends ProcedureBase
{
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 class CompactManifestProcedure extends ProcedureBase {
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 882adc4e78..188b50550e 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 cd74fc682d..a40fa8f905 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 class CompactManifestProcedureITCase extends
CatalogITCaseBase {
"[+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 class CompactManifestProcedureITCase extends
CatalogITCaseBase {
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 cc1a42f9e5..25b417e5b0 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 class CompactManifestProcedure extends BaseProcedure
{
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<String, String> 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 7ef982b085..76368e6f24 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
