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 653175d8e2 [core][flink] Support external sort for manifest sort
(#8357)
653175d8e2 is described below
commit 653175d8e23cec69035e8636dd342a0c08ac970e
Author: umi <[email protected]>
AuthorDate: Sun Jun 28 00:12:58 2026 +0800
[core][flink] Support external sort for manifest sort (#8357)
Improve manifest file sort compaction to avoid heap-heavy in-memory
sorting for large manifest entries by introducing spillable external
sorting. The change also propagates a Paimon IOManager through commit
paths so manifest sort can use configured spill directories in Flink,
while non-Flink/default contexts can keep IOManager creation local to
the sorter.
---
.../apache/paimon/operation/FileStoreCommit.java | 3 +
.../paimon/operation/FileStoreCommitImpl.java | 17 +-
.../operation/ManifestEntryExternalSort.java | 304 ++++++++++++++++++
.../paimon/operation/ManifestFileMerger.java | 12 +-
.../paimon/operation/ManifestFileSorter.java | 342 ++++++++++++---------
.../org/apache/paimon/table/sink/TableCommit.java | 6 +
.../apache/paimon/table/sink/TableCommitImpl.java | 7 +
.../paimon/manifest/ManifestFileMetaTest.java | 95 ++++++
.../org/apache/paimon/flink/sink/Committer.java | 31 ++
.../paimon/flink/sink/CommitterOperator.java | 6 +-
.../apache/paimon/flink/sink/StoreCommitter.java | 14 +-
11 files changed, 681 insertions(+), 156 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java
b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java
index dbd316d8fa..2dab92854c 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java
@@ -19,6 +19,7 @@
package org.apache.paimon.operation;
import org.apache.paimon.Snapshot;
+import org.apache.paimon.disk.IOManager;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.manifest.ManifestCommittable;
import org.apache.paimon.operation.metrics.CommitMetrics;
@@ -34,6 +35,8 @@ import java.util.Map;
/** Commit operation which provides commit and overwrite. */
public interface FileStoreCommit extends AutoCloseable {
+ FileStoreCommit withIOManager(IOManager ioManager);
+
FileStoreCommit ignoreEmptyCommit(boolean ignoreEmptyCommit);
FileStoreCommit withPartitionExpire(PartitionExpire partitionExpire);
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 b97b12f74e..6495d10c27 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
@@ -25,6 +25,7 @@ import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.catalog.SnapshotCommit;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.disk.IOManager;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.io.DataFilePathFactory;
@@ -164,6 +165,7 @@ public class FileStoreCommitImpl implements FileStoreCommit
{
private boolean appendCommitCheckConflict = false;
private long lastCommittedSnapshotId = -1L;
@Nullable private Snapshot.Operation operation;
+ @Nullable private IOManager ioManager;
public FileStoreCommitImpl(
SnapshotCommit snapshotCommit,
@@ -228,6 +230,12 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
this.commitCleaner = new CommitCleaner(manifestList, manifestFile,
indexManifestFile);
}
+ @Override
+ public FileStoreCommit withIOManager(IOManager ioManager) {
+ this.ioManager = ioManager;
+ return this;
+ }
+
@Override
public FileStoreCommit ignoreEmptyCommit(boolean ignoreEmptyCommit) {
this.ignoreEmptyCommit = ignoreEmptyCommit;
@@ -1019,7 +1027,11 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
} else {
mergeAfterManifests =
ManifestFileMerger.merge(
- mergeBeforeManifests, manifestFile,
partitionType, options);
+ mergeBeforeManifests,
+ manifestFile,
+ partitionType,
+ options,
+ ioManager);
}
baseManifestList = manifestList.write(mergeAfterManifests);
@@ -1317,7 +1329,8 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
mergeBeforeManifests,
manifestFile,
partitionType,
- new CoreOptions(compactOptions));
+ new CoreOptions(compactOptions),
+ ioManager);
if (new HashSet<>(mergeBeforeManifests).equals(new
HashSet<>(mergeAfterManifests))) {
// no need to commit this snapshot, because no compact were
happened
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryExternalSort.java
b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryExternalSort.java
new file mode 100644
index 0000000000..5f2c342cc4
--- /dev/null
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryExternalSort.java
@@ -0,0 +1,304 @@
+/*
+ * 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.codegen.CodeGenUtils;
+import org.apache.paimon.codegen.RecordComparator;
+import org.apache.paimon.compression.CompressOptions;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.io.RollingFileWriter;
+import org.apache.paimon.manifest.FileEntry;
+import org.apache.paimon.manifest.FileKind;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.manifest.ManifestEntrySerializer;
+import org.apache.paimon.manifest.ManifestFile;
+import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.options.MemorySize;
+import org.apache.paimon.sort.BinaryExternalSortBuffer;
+import org.apache.paimon.utils.Filter;
+import org.apache.paimon.utils.MutableObjectIterator;
+import org.apache.paimon.utils.Pair;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Function;
+
+import static
org.apache.paimon.utils.ManifestReadThreadPool.sequentialBatchedExecute;
+
+/** Spillable external sort utilities for manifest entries. */
+public class ManifestEntryExternalSort {
+
+ static Pair<List<ManifestFileMeta>, List<ManifestFileMeta>>
sortAndWriteMinorEntries(
+ List<ManifestFileMeta> section,
+ ManifestFileSorter.ManifestSortKey sortKey,
+ ExternalSortConfig config,
+ ManifestFile manifestFile,
+ List<ManifestFileMeta> newFilesForAbort,
+ @Nullable Integer manifestReadParallelism)
+ throws Exception {
+ try (EntrySorter sorter = new EntrySorter(sortKey, config)) {
+ Map<FileEntry.Identifier, ManifestEntry> deleteEntries = new
HashMap<>();
+ Function<ManifestFileMeta, List<ManifestEntry>> reader =
+ meta -> manifestFile.read(meta.fileName(),
meta.fileSize());
+ for (ManifestEntry entry :
+ sequentialBatchedExecute(reader, section,
manifestReadParallelism)) {
+ if (entry.kind() == FileKind.DELETE) {
+ deleteEntries.put(entry.identifier(), entry);
+ } else {
+ sorter.write(entry);
+ }
+ }
+
+ List<ManifestFileMeta> addFiles =
+ sorter.writeSurvivingAddsToManifest(manifestFile,
deleteEntries);
+ // Register ADD files for abort cleanup right after they are
written, before the
+ // DELETE files below. Otherwise, if sortAndWriteDeleteEntries
throws, the already
+ // written ADD manifest files would not be in newFilesForAbort and
would leak as
+ // orphan files on commit abort.
+ newFilesForAbort.addAll(addFiles);
+ List<ManifestFileMeta> deleteFiles =
+ sortAndWriteDeleteEntries(deleteEntries.values(), sortKey,
manifestFile);
+ newFilesForAbort.addAll(deleteFiles);
+ return Pair.of(addFiles, deleteFiles);
+ }
+ }
+
+ static List<ManifestFileMeta> sortAndWriteFullEntries(
+ List<ManifestFileMeta> section,
+ ManifestFileSorter.ManifestSortKey sortKey,
+ ExternalSortConfig config,
+ ManifestFile manifestFile,
+ List<ManifestFileMeta> newFilesForAbort,
+ Set<FileEntry.Identifier> deleteEntries,
+ @Nullable Integer manifestReadParallelism)
+ throws Exception {
+ try (EntrySorter sorter = new EntrySorter(sortKey, config)) {
+ Function<ManifestFileMeta, List<ManifestEntry>> reader =
+ meta -> {
+ List<ManifestEntry> batch = new ArrayList<>();
+ for (ManifestEntry entry :
+ manifestFile.read(
+ meta.fileName(),
+ meta.fileSize(),
+ FileEntry.addFilter(),
+ Filter.alwaysTrue())) {
+ if (!deleteEntries.contains(entry.identifier())) {
+ batch.add(entry);
+ }
+ }
+ return batch;
+ };
+ for (ManifestEntry entry :
+ sequentialBatchedExecute(reader, section,
manifestReadParallelism)) {
+ sorter.write(entry);
+ }
+ List<ManifestFileMeta> files =
sorter.writeToManifest(manifestFile);
+ newFilesForAbort.addAll(files);
+ return files;
+ }
+ }
+
+ /** Config used by manifest entry external sort. */
+ static class ExternalSortConfig {
+ final long bufferSize;
+ final int pageSize;
+ final int maxNumFileHandles;
+ final CompressOptions compression;
+ final MemorySize maxDiskSize;
+ @Nullable final IOManager ioManager;
+
+ ExternalSortConfig(
+ long bufferSize,
+ int pageSize,
+ int maxNumFileHandles,
+ CompressOptions compression,
+ MemorySize maxDiskSize,
+ @Nullable IOManager ioManager) {
+ this.bufferSize = bufferSize;
+ this.pageSize = pageSize;
+ this.maxNumFileHandles = maxNumFileHandles;
+ this.compression = compression;
+ this.maxDiskSize = maxDiskSize;
+ this.ioManager = ioManager;
+ }
+
+ static ExternalSortConfig from(CoreOptions options, @Nullable
IOManager ioManager) {
+ return new ExternalSortConfig(
+ options.sortSpillBufferSize(),
+ options.pageSize(),
+ options.localSortMaxNumFileHandles(),
+ options.spillCompressOptions(),
+ options.writeBufferSpillDiskSize(),
+ ioManager);
+ }
+ }
+
+ /** Spillable sorter that stores sort keys plus serialized manifest
entries in BinaryRow. */
+ private static class EntrySorter implements AutoCloseable {
+ private final ManifestFileSorter.ManifestSortKey sortKey;
+ private final ManifestEntrySerializer entrySerializer;
+ private final IOManager ioManager;
+ private final boolean ownedIOManager;
+ private final BinaryExternalSortBuffer sortBuffer;
+
+ private EntrySorter(ManifestFileSorter.ManifestSortKey sortKey,
ExternalSortConfig config) {
+ this.sortKey = sortKey;
+ this.entrySerializer = new ManifestEntrySerializer();
+ this.ioManager =
+ config.ioManager == null
+ ?
IOManager.create(System.getProperty("java.io.tmpdir"))
+ : config.ioManager;
+ this.ownedIOManager = config.ioManager == null;
+ this.sortBuffer =
+ BinaryExternalSortBuffer.create(
+ ioManager,
+ sortKey.externalSortRowType(),
+ sortKey.externalSortKeyFields(),
+ config.bufferSize,
+ config.pageSize,
+ config.maxNumFileHandles,
+ config.compression,
+ config.maxDiskSize);
+ }
+
+ private void write(ManifestEntry entry) throws Exception {
+ sortBuffer.write(
+ sortKey.toExternalSortRow(entry,
entrySerializer.serializeToBytes(entry)));
+ }
+
+ private boolean isEmpty() {
+ return sortBuffer.isEmpty();
+ }
+
+ private List<ManifestFileMeta> writeToManifest(ManifestFile
manifestFile) throws Exception {
+ if (isEmpty()) {
+ return Collections.emptyList();
+ }
+
+ RollingFileWriter<ManifestEntry, ManifestFileMeta> writer =
+ manifestFile.createRollingWriter();
+ Exception exception = null;
+ try {
+ MutableObjectIterator<BinaryRow> iterator =
sortBuffer.sortedIterator();
+ BinaryRow reuse = new
BinaryRow(sortKey.externalSortRowType().getFieldCount());
+ BinaryRow row;
+ while ((row = iterator.next(reuse)) != null) {
+
writer.write(entrySerializer.deserializeFromBytes(sortKey.entryBytes(row)));
+ }
+ } catch (Exception e) {
+ exception = e;
+ } finally {
+ if (exception != null) {
+ writer.abort();
+ throw exception;
+ }
+ writer.close();
+ }
+ return writer.result();
+ }
+
+ private List<ManifestFileMeta> writeSurvivingAddsToManifest(
+ ManifestFile manifestFile, Map<FileEntry.Identifier,
ManifestEntry> deleteEntries)
+ throws Exception {
+ if (isEmpty()) {
+ return Collections.emptyList();
+ }
+
+ RollingFileWriter<ManifestEntry, ManifestFileMeta> writer =
+ manifestFile.createRollingWriter();
+ Exception exception = null;
+ try {
+ MutableObjectIterator<BinaryRow> iterator =
sortBuffer.sortedIterator();
+ BinaryRow reuse = new
BinaryRow(sortKey.externalSortRowType().getFieldCount());
+ BinaryRow row;
+ while ((row = iterator.next(reuse)) != null) {
+ ManifestEntry entry =
+
entrySerializer.deserializeFromBytes(sortKey.entryBytes(row));
+ if (deleteEntries.remove(entry.identifier()) != null) {
+ continue;
+ }
+ writer.write(entry);
+ }
+ } catch (Exception e) {
+ exception = e;
+ } finally {
+ if (exception != null) {
+ writer.abort();
+ throw exception;
+ }
+ writer.close();
+ }
+ return writer.result();
+ }
+
+ @Override
+ public void close() throws Exception {
+ sortBuffer.clear();
+ if (ownedIOManager) {
+ ioManager.close();
+ }
+ }
+ }
+
+ private static List<ManifestFileMeta> sortAndWriteDeleteEntries(
+ Collection<ManifestEntry> entries,
+ ManifestFileSorter.ManifestSortKey sortKey,
+ ManifestFile manifestFile)
+ throws Exception {
+ if (entries.isEmpty()) {
+ return Collections.emptyList();
+ }
+
+ List<ManifestEntry> sorted = new ArrayList<>(entries);
+ RecordComparator comparator =
+ CodeGenUtils.newRecordComparator(
+ sortKey.externalSortRowType().getFieldTypes(),
+ sortKey.externalSortKeyFields());
+ sorted.sort(
+ (a, b) ->
+ comparator.compare(
+ sortKey.toExternalSortRow(a, new byte[0]),
+ sortKey.toExternalSortRow(b, new byte[0])));
+
+ RollingFileWriter<ManifestEntry, ManifestFileMeta> writer =
+ manifestFile.createRollingWriter();
+ Exception exception = null;
+ try {
+ writer.write(sorted);
+ } catch (Exception e) {
+ exception = e;
+ } finally {
+ if (exception != null) {
+ writer.abort();
+ throw exception;
+ }
+ writer.close();
+ }
+ return writer.result();
+ }
+}
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 f1caa03bcf..0313b2b12b 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
@@ -20,6 +20,7 @@ package org.apache.paimon.operation;
import org.apache.paimon.CoreOptions;
import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.disk.IOManager;
import org.apache.paimon.io.RollingFileWriter;
import org.apache.paimon.manifest.FileEntry;
import org.apache.paimon.manifest.ManifestEntry;
@@ -66,6 +67,15 @@ public class ManifestFileMerger {
ManifestFile manifestFile,
RowType partitionType,
CoreOptions options) {
+ return merge(input, manifestFile, partitionType, options, null);
+ }
+
+ public static List<ManifestFileMeta> merge(
+ List<ManifestFileMeta> input,
+ ManifestFile manifestFile,
+ RowType partitionType,
+ CoreOptions options,
+ @Nullable IOManager ioManager) {
// Extract configuration from options
long suggestedMetaSize = options.manifestTargetSize().getBytes();
int suggestedMinMetaCount = options.manifestMergeMinCount();
@@ -83,7 +93,7 @@ public class ManifestFileMerger {
&& (partitionType.getFieldCount() > 0
|| (options.dataEvolutionEnabled() &&
allContainsRowId(input)))) {
return ManifestFileSorter.trySortCompaction(
- input, newFilesForAbort, manifestFile, partitionType,
options);
+ input, newFilesForAbort, manifestFile, partitionType,
options, ioManager);
} else {
// Otherwise try full compaction first, then minor compaction
if needed
Optional<List<ManifestFileMeta>> fullCompacted =
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 71edb1924e..6201029968 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
@@ -22,15 +22,18 @@ import org.apache.paimon.CoreOptions;
import org.apache.paimon.codegen.CodeGenUtils;
import org.apache.paimon.codegen.RecordComparator;
import org.apache.paimon.data.BinaryRow;
-import org.apache.paimon.io.RollingFileWriter;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.disk.IOManager;
import org.apache.paimon.manifest.FileEntry;
-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.partition.PartitionPredicate;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
-import org.apache.paimon.utils.Filter;
import org.apache.paimon.utils.Pair;
import org.slf4j.Logger;
@@ -51,10 +54,6 @@ import java.util.Map;
import java.util.Optional;
import java.util.PriorityQueue;
import java.util.Set;
-import java.util.function.Function;
-
-import static java.util.Collections.singletonList;
-import static
org.apache.paimon.utils.ManifestReadThreadPool.sequentialBatchedExecute;
/**
* Manifest file sorter that sorts and rewrites manifest files by a configured
partition field, or
@@ -68,6 +67,7 @@ public class ManifestFileSorter {
static class CompactionContext {
final boolean fullCompaction;
final ManifestSortKey sortKey;
+ final ManifestEntryExternalSort.ExternalSortConfig externalSortConfig;
final Set<FileEntry.Identifier> deleteEntries;
/**
* Manifest files that need unsorted compaction.
@@ -85,12 +85,14 @@ public class ManifestFileSorter {
CompactionContext(
boolean fullCompaction,
ManifestSortKey sortKey,
+ ManifestEntryExternalSort.ExternalSortConfig
externalSortConfig,
Set<FileEntry.Identifier> deleteEntries,
Map<ManifestFileMeta, Boolean> compactWithoutSort,
List<ManifestAdjacentSortedRun> levelRuns,
List<ManifestAdjacentSortedRun> pickedRuns) {
this.fullCompaction = fullCompaction;
this.sortKey = sortKey;
+ this.externalSortConfig = externalSortConfig;
this.deleteEntries = deleteEntries;
this.compactWithoutSort = compactWithoutSort;
this.levelRuns = levelRuns;
@@ -139,7 +141,8 @@ public class ManifestFileSorter {
List<ManifestFileMeta> newFilesForAbort,
ManifestFile manifestFile,
RowType partitionType,
- CoreOptions options)
+ CoreOptions options,
+ @Nullable IOManager ioManager)
throws Exception {
String sortPartitionField = options.manifestSortPartitionField();
long suggestedMetaSize = options.manifestTargetSize().getBytes();
@@ -149,6 +152,8 @@ public class ManifestFileSorter {
int maxSizeAmplificationPercent =
options.maxSizeAmplificationPercent();
int sortedRunSizeRatio = options.sortedRunSizeRatio();
Integer manifestReadParallelism = options.scanManifestParallelism();
+ ManifestEntryExternalSort.ExternalSortConfig externalSortConfig =
+ ManifestEntryExternalSort.ExternalSortConfig.from(options,
ioManager);
Optional<List<ManifestFileMeta>> fullCompacted =
tryFullCompaction(
@@ -164,6 +169,7 @@ public class ManifestFileSorter {
maxRewriteSize,
maxSizeAmplificationPercent,
sortedRunSizeRatio,
+ externalSortConfig,
manifestReadParallelism);
if (fullCompacted.isPresent()) {
return fullCompacted.get();
@@ -180,6 +186,7 @@ public class ManifestFileSorter {
maxRewriteSize,
maxSizeAmplificationPercent,
sortedRunSizeRatio,
+ externalSortConfig,
manifestReadParallelism);
}
@@ -202,6 +209,7 @@ public class ManifestFileSorter {
long maxRewriteSize,
int maxSizeAmplificationPercent,
int sortedRunSizeRatio,
+ ManifestEntryExternalSort.ExternalSortConfig externalSortConfig,
@Nullable Integer manifestReadParallelism)
throws Exception {
// Step 1: Check if full compaction threshold is met
@@ -226,6 +234,7 @@ public class ManifestFileSorter {
suggestedMetaSize,
maxSizeAmplificationPercent,
sortedRunSizeRatio,
+ externalSortConfig,
manifestReadParallelism);
List<ManifestAdjacentSortedRun> levelRuns = ctx.levelRuns;
List<ManifestAdjacentSortedRun> pickedRuns = ctx.pickedRuns;
@@ -305,6 +314,7 @@ public class ManifestFileSorter {
long maxRewriteSize,
int maxSizeAmplificationPercent,
int sortedRunSizeRatio,
+ ManifestEntryExternalSort.ExternalSortConfig externalSortConfig,
@Nullable Integer manifestReadParallelism)
throws Exception {
// Step 1: Prepare compaction context (early-return if nothing to
compact)
@@ -319,6 +329,7 @@ public class ManifestFileSorter {
suggestedMetaSize,
maxSizeAmplificationPercent,
sortedRunSizeRatio,
+ externalSortConfig,
manifestReadParallelism);
List<ManifestAdjacentSortedRun> levelRuns = ctx.levelRuns;
List<ManifestAdjacentSortedRun> pickedRuns = ctx.pickedRuns;
@@ -427,6 +438,7 @@ public class ManifestFileSorter {
long suggestedMetaSize,
int maxSizeAmplificationPercent,
int sortedRunSizeRatio,
+ ManifestEntryExternalSort.ExternalSortConfig externalSortConfig,
@Nullable Integer manifestReadParallelism) {
// Step 1: Resolve sort key. Data evolution tables prefer RowID ranges
when available.
@@ -456,6 +468,7 @@ public class ManifestFileSorter {
return new CompactionContext(
fullCompaction,
sortKey,
+ externalSortConfig,
classifyResult.deleteEntries,
classifyResult.compactWithoutSort,
levelRuns,
@@ -489,7 +502,7 @@ public class ManifestFileSorter {
classifiedDeleteEntries =
FileEntry.readDeletedEntries(manifestFile, input,
manifestReadParallelism);
- // Build partition predicate from delete entries for overlap
detection
+ // Build partition predicate from delete entries for overlap
detection.
if (classifiedDeleteEntries.isEmpty()) {
predicate = PartitionPredicate.ALWAYS_FALSE;
} else {
@@ -942,7 +955,9 @@ public class ManifestFileSorter {
return;
}
- if (ctx.fullCompaction) {
+ // Add-only minor sections can use the full rewrite path to avoid
keeping DELETE entries in
+ // memory.
+ if (ctx.fullCompaction || containsNoDeleteEntries(section)) {
rewriteFull(section, output, sortNewFiles, ctx, manifestFile,
manifestReadParallelism);
} else {
rewriteMinor(section, output, sortNewFiles, ctx, manifestFile,
manifestReadParallelism);
@@ -961,43 +976,24 @@ public class ManifestFileSorter {
ManifestFile manifestFile,
@Nullable Integer manifestReadParallelism)
throws Exception {
- // Read surviving ADD entries: filter out entries cancelled by
deleteEntries.
- Function<ManifestFileMeta, List<ManifestEntry>> reader =
- meta -> {
- List<ManifestEntry> batch = new ArrayList<>();
- for (ManifestEntry entry :
- manifestFile.read(
- meta.fileName(),
- meta.fileSize(),
- FileEntry.addFilter(),
- Filter.alwaysTrue())) {
- if (!ctx.deleteEntries.contains(entry.identifier())) {
- batch.add(entry);
- }
- }
- return batch;
- };
-
- List<ManifestEntry> entries = new ArrayList<>();
- for (ManifestEntry entry :
- sequentialBatchedExecute(reader, section,
manifestReadParallelism)) {
- entries.add(entry);
- }
-
- if (!entries.isEmpty()) {
- List<ManifestFileMeta> sorted = sortAndWriteEntries(entries,
ctx.sortKey, manifestFile);
+ List<ManifestFileMeta> sorted =
+ ManifestEntryExternalSort.sortAndWriteFullEntries(
+ section,
+ ctx.sortKey,
+ ctx.externalSortConfig,
+ manifestFile,
+ sortNewFiles,
+ ctx.deleteEntries,
+ manifestReadParallelism);
+ if (!sorted.isEmpty()) {
output.addSortedFiles(sorted);
- sortNewFiles.addAll(sorted);
}
}
/**
- * Minor compaction path: read entries with ADD/DELETE classified in a
single pass per file,
- * then sort each group independently and write them to output.
- *
- * <p>Each file is read in parallel (via sequentialBatchedExecute). The
reader classifies
- * entries into ADD and DELETE within each file, returning a Pair. Results
are merged in the
- * main thread.
+ * Minor compaction path: collect DELETE entries in memory while
external-sorting all entries,
+ * then write surviving ADD entries from the sorted stream and remaining
DELETE entries from
+ * memory.
*/
private static void rewriteMinor(
List<ManifestFileMeta> section,
@@ -1007,90 +1003,31 @@ public class ManifestFileSorter {
ManifestFile manifestFile,
@Nullable Integer manifestReadParallelism)
throws Exception {
- // Read and classify ADD/DELETE in one pass per file.
- Function<ManifestFileMeta, List<Pair<List<ManifestEntry>,
List<ManifestEntry>>>> reader =
- meta -> {
- List<ManifestEntry> addBatch = new ArrayList<>();
- List<ManifestEntry> deleteBatch = new ArrayList<>();
- for (ManifestEntry entry :
- manifestFile.read(meta.fileName(),
meta.fileSize())) {
- if (entry.kind() == FileKind.ADD) {
- addBatch.add(entry);
- } else {
- deleteBatch.add(entry);
- }
- }
- return singletonList(Pair.of(addBatch, deleteBatch));
- };
-
- Map<FileEntry.Identifier, ManifestEntry> addMap = new HashMap<>();
- List<ManifestEntry> minorDeleteEntries = new ArrayList<>();
- for (Pair<List<ManifestEntry>, List<ManifestEntry>> pair :
- sequentialBatchedExecute(reader, section,
manifestReadParallelism)) {
- for (ManifestEntry entry : pair.getLeft()) {
- addMap.put(entry.identifier(), entry);
- }
- minorDeleteEntries.addAll(pair.getRight());
- }
-
- // Cancel out ADD+DELETE pairs with the same identifier within the
section.
- minorDeleteEntries.removeIf(
- manifestEntry -> addMap.remove(manifestEntry.identifier()) !=
null);
- List<ManifestEntry> addEntries = new ArrayList<>(addMap.values());
+ Pair<List<ManifestFileMeta>, List<ManifestFileMeta>> sorted =
+ ManifestEntryExternalSort.sortAndWriteMinorEntries(
+ section,
+ ctx.sortKey,
+ ctx.externalSortConfig,
+ manifestFile,
+ sortNewFiles,
+ manifestReadParallelism);
- if (!addEntries.isEmpty()) {
- List<ManifestFileMeta> sorted =
- sortAndWriteEntries(addEntries, ctx.sortKey, manifestFile);
- output.addSortedFiles(sorted);
- sortNewFiles.addAll(sorted);
+ if (!sorted.getLeft().isEmpty()) {
+ output.addSortedFiles(sorted.getLeft());
}
- if (!minorDeleteEntries.isEmpty()) {
- List<ManifestFileMeta> sorted =
- sortAndWriteEntries(minorDeleteEntries, ctx.sortKey,
manifestFile);
- output.addDeleteFiles(sorted);
- sortNewFiles.addAll(sorted);
+ if (!sorted.getRight().isEmpty()) {
+ output.addDeleteFiles(sorted.getRight());
}
}
- /** Sort entries and write them to a new manifest file with proper error
handling. */
- private static List<ManifestFileMeta> sortAndWriteEntries(
- List<ManifestEntry> entries, ManifestSortKey sortKey, ManifestFile
manifestFile)
- throws Exception {
- entries.sort((a, b) -> compareSortKey(a, b, sortKey));
- RollingFileWriter<ManifestEntry, ManifestFileMeta> writer =
- manifestFile.createRollingWriter();
- Exception exception = null;
- try {
- writer.write(entries);
- } catch (Exception e) {
- exception = e;
- } finally {
- if (exception != null) {
- writer.abort();
- throw exception;
+ private static boolean containsNoDeleteEntries(List<ManifestFileMeta>
section) {
+ for (ManifestFileMeta meta : section) {
+ if (meta.numDeletedFiles() > 0) {
+ return false;
}
- writer.close();
}
- return writer.result();
- }
-
- /**
- * Compare two {@link ManifestEntry}s by the composite key {@code
(sort-key, kind, fileName)}.
- * {@code fileName} is used as the tie-breaker so that all entries sharing
the same sort-field
- * value AND the same data file are emitted contiguously.
- */
- private static int compareSortKey(ManifestEntry a, ManifestEntry b,
ManifestSortKey sortKey) {
- int c = sortKey.compareEntry(a, b);
- if (c != 0) {
- return c;
- }
- // ADD before DELETE
- int kindCmp = a.kind().compareTo(b.kind());
- if (kindCmp != 0) {
- return kindCmp;
- }
- return a.file().fileName().compareTo(b.file().fileName());
+ return true;
}
private static ManifestSortKey createSortKey(
@@ -1102,9 +1039,11 @@ public class ManifestFileSorter {
// RowID sorting uses the configured partition field as the
primary key when specified,
// otherwise it uses the full partition row to preserve partition
locality. It then
// orders files by RowID.
+ int[] partitionSortFields =
+ createPartitionSortFields(sortPartitionField,
partitionType);
RecordComparator partitionComparator =
- createPartitionComparator(sortPartitionField,
partitionType);
- return new RowIdSortKey(partitionComparator);
+ createPartitionComparator(partitionType,
partitionSortFields);
+ return new RowIdSortKey(partitionComparator, partitionType,
partitionSortFields);
}
if (partitionType.getFieldCount() == 0) {
@@ -1124,11 +1063,10 @@ public class ManifestFileSorter {
RecordComparator fieldComparator =
CodeGenUtils.newRecordComparator(
partitionType.getFieldTypes(), new int[]
{sortFieldIndex});
- return new PartitionSortKey(fieldComparator);
+ return new PartitionSortKey(fieldComparator, partitionType,
sortFieldIndex);
}
- @Nullable
- private static RecordComparator createPartitionComparator(
+ private static int[] createPartitionSortFields(
String sortPartitionField, RowType partitionType) {
if (sortPartitionField != null && !sortPartitionField.isEmpty()) {
int sortFieldIndex =
partitionType.getFieldNames().indexOf(sortPartitionField);
@@ -1138,15 +1076,24 @@ public class ManifestFileSorter {
"Cannot resolve sort field '%s' for manifest
sort rewrite.",
sortPartitionField));
}
- return CodeGenUtils.newRecordComparator(
- partitionType.getFieldTypes(), new int[] {sortFieldIndex});
+ return new int[] {sortFieldIndex};
}
- if (partitionType.getFieldCount() == 0) {
- return null;
+ int fieldCount = partitionType.getFieldCount();
+ int[] sortFields = new int[fieldCount];
+ for (int i = 0; i < fieldCount; i++) {
+ sortFields[i] = i;
}
+ return sortFields;
+ }
- return CodeGenUtils.newRecordComparator(partitionType.getFieldTypes());
+ @Nullable
+ private static RecordComparator createPartitionComparator(
+ RowType partitionType, int[] sortFields) {
+ if (sortFields.length == 0) {
+ return null;
+ }
+ return CodeGenUtils.newRecordComparator(partitionType.getFieldTypes(),
sortFields);
}
interface ManifestSortKey {
@@ -1157,15 +1104,36 @@ public class ManifestFileSorter {
boolean isAfterMax(ManifestFileMeta file, ManifestFileMeta maxFile);
- int compareEntry(ManifestEntry a, ManifestEntry b);
+ RowType externalSortRowType();
+
+ int[] externalSortKeyFields();
+
+ InternalRow toExternalSortRow(ManifestEntry entry, byte[] entryBytes);
+
+ byte[] entryBytes(BinaryRow row);
}
private static class PartitionSortKey implements ManifestSortKey {
private final RecordComparator fieldComparator;
+ private final InternalRow.FieldGetter sortFieldGetter;
+ private final RowType externalSortRowType;
+ private final int[] externalSortKeyFields;
+ private final int sortFieldNum;
- private PartitionSortKey(RecordComparator fieldComparator) {
+ private PartitionSortKey(
+ RecordComparator fieldComparator, RowType partitionType, int
sortFieldIndex) {
this.fieldComparator = fieldComparator;
+ DataType sortFieldType = partitionType.getTypeAt(sortFieldIndex);
+ this.sortFieldGetter =
InternalRow.createFieldGetter(sortFieldType, sortFieldIndex);
+ this.sortFieldNum = 3;
+ this.externalSortRowType =
+ DataTypes.ROW(
+ sortFieldType,
+ DataTypes.TINYINT(),
+ DataTypes.STRING(),
+ DataTypes.BYTES());
+ this.externalSortKeyFields = createSequentialFields(sortFieldNum);
}
@Override
@@ -1188,17 +1156,60 @@ public class ManifestFileSorter {
}
@Override
- public int compareEntry(ManifestEntry a, ManifestEntry b) {
- return fieldComparator.compare(a.partition(), b.partition());
+ public RowType externalSortRowType() {
+ return externalSortRowType;
+ }
+
+ @Override
+ public int[] externalSortKeyFields() {
+ return externalSortKeyFields;
+ }
+
+ @Override
+ public InternalRow toExternalSortRow(ManifestEntry entry, byte[]
entryBytes) {
+ GenericRow row = new
GenericRow(externalSortRowType.getFieldCount());
+ row.setField(0, sortFieldGetter.getFieldOrNull(entry.partition()));
+ row.setField(1, entry.kind().toByteValue());
+ row.setField(2, BinaryString.fromString(entry.file().fileName()));
+ row.setField(3, entryBytes);
+ return row;
+ }
+
+ @Override
+ public byte[] entryBytes(BinaryRow row) {
+ return row.getBinary(sortFieldNum);
}
}
private static class RowIdSortKey implements ManifestSortKey {
@Nullable private final RecordComparator partitionComparator;
-
- private RowIdSortKey(@Nullable RecordComparator partitionComparator) {
+ private final InternalRow.FieldGetter[] partitionFieldGetters;
+ private final RowType externalSortRowType;
+ private final int[] externalSortKeyFields;
+ private final int sortFieldNum;
+
+ private RowIdSortKey(
+ @Nullable RecordComparator partitionComparator,
+ RowType partitionType,
+ int[] partitionSortFields) {
this.partitionComparator = partitionComparator;
+ this.partitionFieldGetters =
+ createPartitionFieldGetters(partitionType,
partitionSortFields);
+
+ List<DataType> fieldTypes = new ArrayList<>();
+ for (int partitionSortField : partitionSortFields) {
+ fieldTypes.add(partitionType.getTypeAt(partitionSortField));
+ }
+ fieldTypes.add(DataTypes.BIGINT());
+ fieldTypes.add(DataTypes.BIGINT());
+ fieldTypes.add(DataTypes.BIGINT());
+ fieldTypes.add(DataTypes.TINYINT());
+ fieldTypes.add(DataTypes.STRING());
+ fieldTypes.add(DataTypes.BYTES());
+ this.externalSortRowType = DataTypes.ROW(fieldTypes.toArray(new
DataType[0]));
+ this.sortFieldNum = externalSortRowType.getFieldCount() - 1;
+ this.externalSortKeyFields = createSequentialFields(sortFieldNum);
}
@Override
@@ -1234,23 +1245,34 @@ public class ManifestFileSorter {
}
@Override
- public int compareEntry(ManifestEntry a, ManifestEntry b) {
- int c = 0;
- if (partitionComparator != null) {
- c = partitionComparator.compare(a.partition(), b.partition());
- if (c != 0) {
- return c;
- }
- }
- c = Long.compare(a.file().nonNullFirstRowId(),
b.file().nonNullFirstRowId());
- if (c != 0) {
- return c;
- }
- c = Long.compare(rowIdRangeEnd(a), rowIdRangeEnd(b));
- if (c != 0) {
- return c;
+ public RowType externalSortRowType() {
+ return externalSortRowType;
+ }
+
+ @Override
+ public int[] externalSortKeyFields() {
+ return externalSortKeyFields;
+ }
+
+ @Override
+ public InternalRow toExternalSortRow(ManifestEntry entry, byte[]
entryBytes) {
+ GenericRow row = new
GenericRow(externalSortRowType.getFieldCount());
+ int pos = 0;
+ for (InternalRow.FieldGetter partitionFieldGetter :
partitionFieldGetters) {
+ row.setField(pos++,
partitionFieldGetter.getFieldOrNull(entry.partition()));
}
- return Long.compare(b.file().maxSequenceNumber(),
a.file().maxSequenceNumber());
+ row.setField(pos++, entry.file().nonNullFirstRowId());
+ row.setField(pos++, rowIdRangeEnd(entry));
+ row.setField(pos++, Long.MAX_VALUE -
entry.file().maxSequenceNumber());
+ row.setField(pos++, entry.kind().toByteValue());
+ row.setField(pos++,
BinaryString.fromString(entry.file().fileName()));
+ row.setField(pos, entryBytes);
+ return row;
+ }
+
+ @Override
+ public byte[] entryBytes(BinaryRow row) {
+ return row.getBinary(sortFieldNum);
}
private int comparePartitionMin(ManifestFileMeta a, ManifestFileMeta
b) {
@@ -1292,6 +1314,26 @@ public class ManifestFileSorter {
}
}
+ private static int[] createSequentialFields(int fieldCount) {
+ int[] fields = new int[fieldCount];
+ for (int i = 0; i < fieldCount; i++) {
+ fields[i] = i;
+ }
+ return fields;
+ }
+
+ private static InternalRow.FieldGetter[] createPartitionFieldGetters(
+ RowType partitionType, int[] partitionSortFields) {
+ InternalRow.FieldGetter[] fieldGetters =
+ new InternalRow.FieldGetter[partitionSortFields.length];
+ for (int i = 0; i < partitionSortFields.length; i++) {
+ int fieldIndex = partitionSortFields[i];
+ fieldGetters[i] =
+
InternalRow.createFieldGetter(partitionType.getTypeAt(fieldIndex), fieldIndex);
+ }
+ return fieldGetters;
+ }
+
/**
* Resolve the partition field to sort manifests by.
*
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommit.java
b/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommit.java
index 26f4ffd163..f616891406 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommit.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommit.java
@@ -19,6 +19,7 @@
package org.apache.paimon.table.sink;
import org.apache.paimon.annotation.Public;
+import org.apache.paimon.disk.IOManager;
import org.apache.paimon.metrics.MetricRegistry;
import org.apache.paimon.table.Table;
@@ -36,6 +37,11 @@ import java.util.List;
@Public
public interface TableCommit extends AutoCloseable {
+ /** With {@link IOManager}, this is needed if commit-time external sort is
used. */
+ default TableCommit withIOManager(IOManager ioManager) {
+ return this;
+ }
+
/** Set {@link MetricRegistry} to table commit. */
TableCommit withMetricRegistry(MetricRegistry registry);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java
b/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java
index 6a9e3046cc..9ec742b354 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java
@@ -21,6 +21,7 @@ package org.apache.paimon.table.sink;
import org.apache.paimon.Snapshot;
import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.consumer.ConsumerManager;
+import org.apache.paimon.disk.IOManager;
import org.apache.paimon.fs.Path;
import org.apache.paimon.index.IndexPathFactory;
import org.apache.paimon.io.DataFileMeta;
@@ -177,6 +178,12 @@ public class TableCommitImpl implements InnerTableCommit {
return this;
}
+ @Override
+ public TableCommitImpl withIOManager(IOManager ioManager) {
+ commit.withIOManager(ioManager);
+ return this;
+ }
+
@Override
public InnerTableCommit withMetricRegistry(MetricRegistry registry) {
commit.withMetrics(new CommitMetrics(registry, tableName));
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 42638aa24f..eced1bf4ef 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
@@ -22,6 +22,7 @@ import org.apache.paimon.CoreOptions;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.BinaryRowWriter;
import org.apache.paimon.data.Timestamp;
+import org.apache.paimon.disk.IOManager;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.FileIOFinder;
import org.apache.paimon.fs.Path;
@@ -960,6 +961,96 @@ public class ManifestFileMetaTest extends
ManifestFileMetaTestBase {
}
}
+ @Test
+ public void testManifestSortWithSpillableExternalSortBuffer() {
+ List<ManifestFileMeta> input = new ArrayList<>();
+ for (int manifest = 0; manifest < 4; manifest++) {
+ List<ManifestEntry> entries = new ArrayList<>();
+ for (int i = 0; i < 80; i++) {
+ int partition = manifest % 2 == 0 ? 79 - i : i;
+ entries.add(
+ makeEntry(
+ true,
+ String.format(
+
"spill-manifest-%02d-entry-%03d-payload-padding-%040d",
+ manifest, i, i),
+ partition));
+ }
+ input.add(makeManifest(entries.toArray(new ManifestEntry[0])));
+ }
+
+ Options testOptions = new Options();
+ testOptions.set("manifest-sort.enabled", "true");
+ testOptions.set("manifest.full-compaction-threshold-size", "1B");
+ testOptions.set("page-size", "1kb");
+ testOptions.set("sort-spill-buffer-size", "4kb");
+ testOptions.set("local-sort.max-num-file-handles", "2");
+
+ List<ManifestFileMeta> merged =
+ ManifestFileMerger.merge(
+ input,
+ manifestFile,
+ getPartitionType(),
+ CoreOptions.fromMap(testOptions.toMap()));
+
+ assertEquivalentEntries(input, merged);
+ for (ManifestFileMeta meta : merged) {
+ List<ManifestEntry> entries = manifestFile.read(meta.fileName(),
meta.fileSize());
+ for (int i = 1; i < entries.size(); i++) {
+ int prevPartition = entries.get(i - 1).partition().getInt(0);
+ int currPartition = entries.get(i).partition().getInt(0);
+ assertThat(currPartition)
+ .as("Entries within a manifest should be sorted after
spill")
+ .isGreaterThanOrEqualTo(prevPartition);
+ }
+ }
+ }
+
+ @Test
+ public void testManifestSortUsesExternalIOManagerWithoutClosingIt() throws
Exception {
+ List<ManifestFileMeta> input = new ArrayList<>();
+ for (int manifest = 0; manifest < 2; manifest++) {
+ List<ManifestEntry> entries = new ArrayList<>();
+ for (int i = 0; i < 40; i++) {
+ int partition = manifest == 0 ? 39 - i : i;
+ entries.add(
+ makeEntry(
+ true,
+ String.format(
+
"external-io-manager-%02d-entry-%03d-payload-%040d",
+ manifest, i, i),
+ partition));
+ }
+ input.add(makeManifest(entries.toArray(new ManifestEntry[0])));
+ }
+
+ Options testOptions = new Options();
+ testOptions.set("manifest-sort.enabled", "true");
+ testOptions.set("manifest.full-compaction-threshold-size", "1B");
+ testOptions.set("page-size", "1kb");
+ testOptions.set("sort-spill-buffer-size", "4kb");
+
+ java.nio.file.Path spillBase = tempDir.resolve("manifest-spill");
+ java.nio.file.Files.createDirectories(spillBase);
+ IOManager ioManager = IOManager.create(spillBase.toString());
+ try {
+ List<ManifestFileMeta> merged =
+ ManifestFileMerger.merge(
+ input,
+ manifestFile,
+ getPartitionType(),
+ CoreOptions.fromMap(testOptions.toMap()),
+ ioManager);
+
+ assertEquivalentEntries(input, merged);
+ assertThat(spillBase.toFile().list((dir, name) ->
name.startsWith("paimon-")))
+ .isNotEmpty();
+ } finally {
+ ioManager.close();
+ }
+ assertThat(spillBase.toFile().list()).isEmpty();
+ }
+
/**
* Test that sort rewrite correctly eliminates DELETE entries and their
corresponding ADD
* entries. The key condition is that totalDeltaFileSize must reach
manifestFullCompactionSize
@@ -1021,6 +1112,10 @@ public class ManifestFileMetaTest extends
ManifestFileMetaTestBase {
Options testOptions = new Options();
testOptions.set("manifest-sort.enabled", "true");
testOptions.set("manifest.full-compaction-threshold-size", "10B");
+ testOptions.set("manifest-sort.max-rewrite-size", "1B");
+ testOptions.set("page-size", "1kb");
+ testOptions.set("sort-spill-buffer-size", "4kb");
+ testOptions.set("local-sort.max-num-file-handles", "2");
List<ManifestFileMeta> merged =
ManifestFileMerger.merge(
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/Committer.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/Committer.java
index f1684be96e..a556f6cc3f 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/Committer.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/Committer.java
@@ -94,6 +94,11 @@ public interface Committer<CommitT, GlobalCommitT> extends
AutoCloseable {
int getParallelism();
int getSubtaskIndex();
+
+ @Nullable
+ default String[] tempDirs() {
+ return null;
+ }
}
static Context createContext(
@@ -104,6 +109,26 @@ public interface Committer<CommitT, GlobalCommitT> extends
AutoCloseable {
StateStore stateStore,
int parallelism,
int subtaskIndex) {
+ return createContext(
+ commitUser,
+ metricGroup,
+ streamingCheckpointEnabled,
+ isRestored,
+ stateStore,
+ parallelism,
+ subtaskIndex,
+ null);
+ }
+
+ static Context createContext(
+ String commitUser,
+ @Nullable MetricGroup metricGroup,
+ boolean streamingCheckpointEnabled,
+ boolean isRestored,
+ StateStore stateStore,
+ int parallelism,
+ int subtaskIndex,
+ @Nullable String[] tempDirs) {
return new Committer.Context() {
@Override
public String commitUser() {
@@ -139,6 +164,12 @@ public interface Committer<CommitT, GlobalCommitT> extends
AutoCloseable {
public int getSubtaskIndex() {
return subtaskIndex;
}
+
+ @Override
+ @Nullable
+ public String[] tempDirs() {
+ return tempDirs;
+ }
};
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CommitterOperator.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CommitterOperator.java
index 034ded9c64..9056440cf6 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CommitterOperator.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CommitterOperator.java
@@ -142,7 +142,11 @@ public class CommitterOperator<CommitT, GlobalCommitT>
extends AbstractStreamOpe
context.isRestored(),
new
OperatorBackendStateStore(context.getOperatorStateStore()),
parallelism,
- index);
+ index,
+ getContainingTask()
+ .getEnvironment()
+ .getIOManager()
+ .getSpillingDirectoriesPaths());
committer = committerFactory.create(committerContext);
committableStateManager.initializeState(committerContext, committer);
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreCommitter.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreCommitter.java
index 48daabc8d8..4c353517c3 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreCommitter.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreCommitter.java
@@ -19,6 +19,7 @@
package org.apache.paimon.flink.sink;
import org.apache.paimon.annotation.VisibleForTesting;
+import org.apache.paimon.disk.IOManager;
import org.apache.paimon.flink.metrics.FlinkMetricRegistry;
import org.apache.paimon.flink.sink.listener.CommitListeners;
import org.apache.paimon.io.DataFileMeta;
@@ -29,6 +30,7 @@ import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.table.sink.CommitMessageImpl;
import org.apache.paimon.table.sink.TableCommit;
import org.apache.paimon.table.sink.TableCommitImpl;
+import org.apache.paimon.utils.IOUtils;
import javax.annotation.Nullable;
@@ -45,6 +47,7 @@ public class StoreCommitter implements Committer<Committable,
ManifestCommittabl
private final TableCommitImpl commit;
@Nullable private final CommitterMetrics committerMetrics;
private final CommitListeners commitListeners;
+ @Nullable private final IOManager commitIOManager;
private final boolean allowLogOffsetDuplicate;
public StoreCommitter(FileStoreTable table, TableCommit commit, Context
context) {
@@ -62,6 +65,14 @@ public class StoreCommitter implements
Committer<Committable, ManifestCommittabl
} catch (Exception e) {
throw new RuntimeException(e);
}
+
+ String[] tempDirs = context.tempDirs();
+ if (tempDirs == null) {
+ this.commitIOManager = null;
+ } else {
+ this.commitIOManager = IOManager.create(tempDirs);
+ this.commit.withIOManager(commitIOManager);
+ }
allowLogOffsetDuplicate = table.bucketMode() ==
BucketMode.BUCKET_UNAWARE;
}
@@ -132,8 +143,7 @@ public class StoreCommitter implements
Committer<Committable, ManifestCommittabl
@Override
public void close() throws Exception {
- commit.close();
- commitListeners.close();
+ IOUtils.closeAll(commit, commitListeners, commitIOManager);
}
public boolean allowLogOffsetDuplicate() {
