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 c6418dd5ad [core] Avoid external sort for fragmented manifests (#9257)
c6418dd5ad is described below
commit c6418dd5ad6c066278ecbbccee8ef54679c7e127
Author: YeJunHao <[email protected]>
AuthorDate: Mon Aug 17 13:42:29 2026 +0800
[core] Avoid external sort for fragmented manifests (#9257)
---
.../paimon/operation/ManifestEntryRunMerge.java | 19 +--
.../operation/ManifestEntryRunMergePlan.java | 21 +--
.../paimon/manifest/ManifestFileMetaTest.java | 4 +-
.../operation/ManifestEntryRunMergeTest.java | 152 +++++++++++++++++++++
4 files changed, 162 insertions(+), 34 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryRunMerge.java
b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryRunMerge.java
index 18cd08dfb3..7eb4f66e32 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryRunMerge.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryRunMerge.java
@@ -59,7 +59,6 @@ import static
org.apache.paimon.utils.Preconditions.checkState;
final class ManifestEntryRunMerge {
private static final int FRAGMENTED_RUN_THRESHOLD = 64;
- private static final long MAX_IN_MEMORY_FRAGMENTED_ENTRIES = 25_000L;
private static final int MAX_STREAM_CURSORS = 128;
private static final int MAX_STREAM_READ_AMPLIFICATION = 8;
@@ -93,7 +92,7 @@ final class ManifestEntryRunMerge {
if (plan == null) {
return null;
}
- return plan.mergeToManifest(sortKey, manifestFile, newFilesForAbort);
+ return plan.mergeToManifest(manifestFile, newFilesForAbort);
}
/**
@@ -125,7 +124,7 @@ final class ManifestEntryRunMerge {
if (plan == null) {
return null;
}
- return plan.mergeMinorToManifest(sortKey, manifestFile,
newFilesForAbort);
+ return plan.mergeMinorToManifest(manifestFile, newFilesForAbort);
} finally {
deletes.release();
}
@@ -146,7 +145,6 @@ final class ManifestEntryRunMerge {
new SortPartitionDictionary(sortKey::comparePartitions);
List<ManifestEntryRunMergePlan.Source.Spec> sources = new
ArrayList<>();
int streamCursorCount = 0;
- long inMemoryEntries = 0;
List<Discovery.DiscoveredManifest> discovered = new
ArrayList<>(section.size());
if (section.size() <= 1
|| (manifestReadParallelism != null && manifestReadParallelism
<= 1)) {
@@ -234,11 +232,6 @@ final class ManifestEntryRunMerge {
ManifestFileMeta meta = section.get(manifestIndex);
Discovery.DiscoveredManifest manifest =
discovered.get(manifestIndex);
if (manifest.fragmented) {
- long entryCount = meta.numAddedFiles() +
meta.numDeletedFiles();
- inMemoryEntries += entryCount;
- if (inMemoryEntries > MAX_IN_MEMORY_FRAGMENTED_ENTRIES) {
- return null;
- }
sources.add(new
ManifestEntryRunMergePlan.Source.FragmentedManifestSpec(meta));
streamCursorCount++;
} else {
@@ -297,7 +290,6 @@ final class ManifestEntryRunMerge {
boolean hasPrevious = false;
long runStart = 0;
long position = 0;
- long entryCount = meta.numAddedFiles() + meta.numDeletedFiles();
boolean fragmented = false;
ProjectedManifestEntry entry =
ProjectedManifestEntry.ENTRY_LAYOUT_PROJECTION.createEntry();
while (reader.hasNext()) {
@@ -336,9 +328,6 @@ final class ManifestEntryRunMerge {
meta, runStart, position, blocks));
runStart = position;
if (runs.size() >= FRAGMENTED_RUN_THRESHOLD) {
- if (entryCount > MAX_IN_MEMORY_FRAGMENTED_ENTRIES) {
- return
Discovery.DiscoveredManifest.requiresExternalSort();
- }
fragmented = true;
runs.clear();
blocks.clear();
@@ -366,9 +355,7 @@ final class ManifestEntryRunMerge {
meta, runStart, position, blocks));
}
if (exceedsStreamingReadAmplification(runs, blocks.size())) {
- return entryCount > MAX_IN_MEMORY_FRAGMENTED_ENTRIES
- ? Discovery.DiscoveredManifest.requiresExternalSort()
- : Discovery.DiscoveredManifest.fragmented();
+ return Discovery.DiscoveredManifest.fragmented();
}
return Discovery.DiscoveredManifest.runs(runs, blocks);
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryRunMergePlan.java
b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryRunMergePlan.java
index 997bfb362b..1b1d9562a0 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryRunMergePlan.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryRunMergePlan.java
@@ -69,15 +69,12 @@ final class ManifestEntryRunMergePlan {
}
List<ManifestFileMeta> mergeToManifest(
- ManifestFileSorter.RowIdEntrySortKey sortKey,
- ManifestFile manifestFile,
- List<ManifestFileMeta> newFilesForAbort)
- throws Exception {
+ ManifestFile manifestFile, List<ManifestFileMeta>
newFilesForAbort) throws Exception {
List<Cursor> cursors = new ArrayList<>(sources.size());
Exception failure = null;
try {
for (Source.Spec source : sources) {
- Cursor cursor = source.open(manifestFile, sortKey, deletes,
minor, partitions);
+ Cursor cursor = source.open(manifestFile, deletes, minor,
partitions);
cursors.add(cursor);
cursor.advance();
}
@@ -104,15 +101,12 @@ final class ManifestEntryRunMergePlan {
}
Pair<List<ManifestFileMeta>, List<ManifestFileMeta>> mergeMinorToManifest(
- ManifestFileSorter.RowIdEntrySortKey sortKey,
- ManifestFile manifestFile,
- List<ManifestFileMeta> newFilesForAbort)
- throws Exception {
+ ManifestFile manifestFile, List<ManifestFileMeta>
newFilesForAbort) throws Exception {
List<Cursor> cursors = new ArrayList<>(sources.size());
Exception failure = null;
try {
for (Source.Spec source : sources) {
- Cursor cursor = source.open(manifestFile, sortKey, deletes,
minor, partitions);
+ Cursor cursor = source.open(manifestFile, deletes, minor,
partitions);
cursors.add(cursor);
cursor.advance();
}
@@ -264,7 +258,6 @@ final class ManifestEntryRunMergePlan {
Cursor open(
ManifestFile manifestFile,
- ManifestFileSorter.RowIdEntrySortKey sortKey,
CollectedDeletes deletes,
boolean minor,
ManifestEntryRunMerge.SortPartitionDictionary partitions)
@@ -306,7 +299,6 @@ final class ManifestEntryRunMergePlan {
@Override
public Cursor open(
ManifestFile manifestFile,
- ManifestFileSorter.RowIdEntrySortKey sortKey,
CollectedDeletes deletes,
boolean minor,
ManifestEntryRunMerge.SortPartitionDictionary partitions)
@@ -327,13 +319,11 @@ final class ManifestEntryRunMergePlan {
@Override
public Cursor open(
ManifestFile manifestFile,
- ManifestFileSorter.RowIdEntrySortKey sortKey,
CollectedDeletes deletes,
boolean minor,
ManifestEntryRunMerge.SortPartitionDictionary partitions)
throws Exception {
- return new InMemoryManifestCursor(
- manifestFile, meta, sortKey, deletes, minor,
partitions);
+ return new InMemoryManifestCursor(manifestFile, meta, deletes,
minor, partitions);
}
}
}
@@ -673,7 +663,6 @@ final class ManifestEntryRunMergePlan {
InMemoryManifestCursor(
ManifestFile manifestFile,
ManifestFileMeta meta,
- ManifestFileSorter.RowIdEntrySortKey sortKey,
CollectedDeletes deletes,
boolean minor,
ManifestEntryRunMerge.SortPartitionDictionary partitions)
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 fe345048c3..6da1b8d2f7 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
@@ -1922,7 +1922,7 @@ public class ManifestFileMetaTest extends
ManifestFileMetaTestBase {
}
@Test
- public void
testDataEvolutionManifestRunMergeFallsBackForLargeFragmentedManifest() {
+ public void
testDataEvolutionManifestRunMergeSortsLargeFragmentedManifest() {
List<ManifestEntry> firstManifest = new ArrayList<>();
List<ManifestEntry> secondManifest = new ArrayList<>();
for (long firstRowId = 25_000; firstRowId >= 12_500; firstRowId--) {
@@ -1956,7 +1956,7 @@ public class ManifestFileMetaTest extends
ManifestFileMetaTestBase {
}
@Test
- public void
testDataEvolutionMinorRunMergeFallsBackForLargeFragmentedManifest() {
+ public void testDataEvolutionMinorRunMergeSortsLargeFragmentedManifest() {
List<ManifestEntry> fragmentedEntries = new ArrayList<>();
for (long firstRowId = 25_000; firstRowId >= 0; firstRowId--) {
fragmentedEntries.add(
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/ManifestEntryRunMergeTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/ManifestEntryRunMergeTest.java
new file mode 100644
index 0000000000..f93d1e7134
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/ManifestEntryRunMergeTest.java
@@ -0,0 +1,152 @@
+/*
+ * 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.data.BinaryRow;
+import org.apache.paimon.data.BinaryRowWriter;
+import org.apache.paimon.data.Timestamp;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.manifest.CollectedDeletes;
+import org.apache.paimon.manifest.FileKind;
+import org.apache.paimon.manifest.FileSource;
+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.stats.StatsTestUtils;
+import org.apache.paimon.types.DataTypes;
+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.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link ManifestEntryRunMerge}. */
+class ManifestEntryRunMergeTest extends ManifestFileMetaTestBase {
+
+ private static final int ENTRY_COUNT = 25_001;
+
+ @TempDir java.nio.file.Path tempDir;
+
+ private final RowType partitionType = RowType.of(DataTypes.INT());
+ private final BinaryRow partition = new BinaryRow(1);
+ private ManifestFile manifestFile;
+
+ @BeforeEach
+ void beforeEach() {
+ BinaryRowWriter writer = new BinaryRowWriter(partition);
+ writer.writeInt(0, 0);
+ writer.complete();
+ manifestFile = createManifestFile(tempDir.toString());
+ }
+
+ @Test
+ void testLargeFragmentedManifestsUseRunMerge() throws Exception {
+ List<ManifestEntry> firstManifest = new ArrayList<>();
+ List<ManifestEntry> secondManifest = new ArrayList<>();
+ for (long firstRowId = ENTRY_COUNT - 1L; firstRowId >= 0;
firstRowId--) {
+ ManifestEntry entry = rowIdEntry("row-" + firstRowId, firstRowId);
+ (firstRowId >= ENTRY_COUNT / 2 ? firstManifest :
secondManifest).add(entry);
+ }
+
+ List<ManifestFileMeta> manifests = new ArrayList<>();
+ manifests.add(makeManifest(firstManifest.toArray(new
ManifestEntry[0])));
+ manifests.add(makeManifest(secondManifest.toArray(new
ManifestEntry[0])));
+ ManifestFileSorter.RowIdEntrySortKey sortKey =
+ (ManifestFileSorter.RowIdEntrySortKey)
+ ManifestFileSorter.createSortKey(true, manifests,
null, partitionType);
+
+ CollectedDeletes deletes = new CollectedDeletes(true);
+ List<ManifestFileMeta> output;
+ try {
+ output =
+ ManifestEntryRunMerge.sortAndWriteFullEntries(
+ manifests,
+ sortKey,
+ partitionType,
+ manifestFile,
+ new ArrayList<>(),
+ deletes,
+ 128,
+ 1);
+ } finally {
+ deletes.release();
+ }
+
+ assertThat(output).isNotNull();
+ assertThat(
+ output.stream()
+ .flatMap(
+ meta ->
+
manifestFile.read(meta.fileName(), meta.fileSize())
+ .stream())
+ .map(entry -> entry.file().nonNullFirstRowId())
+ .collect(Collectors.toList()))
+ .containsExactlyElementsOf(
+ java.util.stream.LongStream.range(0, ENTRY_COUNT)
+ .boxed()
+ .collect(Collectors.toList()));
+ }
+
+ private ManifestEntry rowIdEntry(String fileName, long firstRowId) {
+ return ManifestEntry.create(
+ FileKind.ADD,
+ partition,
+ 0,
+ 0,
+ DataFileMeta.create(
+ fileName,
+ 0,
+ 1,
+ partition,
+ partition,
+ StatsTestUtils.newEmptySimpleStats(),
+ StatsTestUtils.newEmptySimpleStats(),
+ 0,
+ 0,
+ 0,
+ 0,
+ Collections.emptyList(),
+ Timestamp.fromEpochMillis(200000),
+ 0L,
+ null,
+ FileSource.APPEND,
+ null,
+ null,
+ firstRowId,
+ Collections.singletonList("f0")));
+ }
+
+ @Override
+ protected ManifestFile getManifestFile() {
+ return manifestFile;
+ }
+
+ @Override
+ protected RowType getPartitionType() {
+ return partitionType;
+ }
+}
