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 6b6e805008 [core] Fix bucket-first manifest sorting for postpone tables
6b6e805008 is described below
commit 6b6e8050088194512bffa72015a9562445a4f756
Author: JingsongLi <[email protected]>
AuthorDate: Mon Sep 14 10:15:19 2026 +0800
[core] Fix bucket-first manifest sorting for postpone tables
---
.../paimon/operation/ManifestCompactDryRun.java | 3 +-
.../paimon/operation/ManifestEntryRunMerge.java | 6 +-
.../paimon/operation/ManifestFileSorter.java | 25 ++------
.../paimon/manifest/ManifestFileMetaTest.java | 73 ++++++++++++++++++----
.../operation/ManifestEntryRunMergeTest.java | 7 ++-
.../operation/ManifestRewriteCleanupTest.java | 2 +-
6 files changed, 76 insertions(+), 40 deletions(-)
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 6f1e5f8e40..f80554ce6e 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
@@ -24,6 +24,7 @@ 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.BucketMode;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.types.RowType;
@@ -120,7 +121,7 @@ public class ManifestCompactDryRun {
manifests,
options.manifestSortPartitionField(),
partitionType,
- options.bucket() > 0);
+ options.bucket() > 0 || options.bucket() ==
BucketMode.POSTPONE_BUCKET);
ManifestFileSorter.ClassifyResult classifyResult =
ManifestFileSorter.classifyManifests(
manifests,
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 740b23f58d..aeb8e6ef62 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
@@ -71,7 +71,7 @@ final class ManifestEntryRunMerge {
@Nullable
static List<ManifestFileMeta> sortAndWriteFullEntries(
List<ManifestFileMeta> section,
- ManifestFileSorter.RowIdEntrySortKey sortKey,
+ ManifestFileSorter.RowIdSortKey sortKey,
RowType partitionType,
ManifestFile manifestFile,
List<ManifestFileMeta> newFilesForAbort,
@@ -102,7 +102,7 @@ final class ManifestEntryRunMerge {
@Nullable
static Pair<List<ManifestFileMeta>, List<ManifestFileMeta>>
sortAndWriteMinorEntries(
List<ManifestFileMeta> section,
- ManifestFileSorter.RowIdEntrySortKey sortKey,
+ ManifestFileSorter.RowIdSortKey sortKey,
RowType partitionType,
ManifestFile manifestFile,
List<ManifestFileMeta> newFilesForAbort,
@@ -133,7 +133,7 @@ final class ManifestEntryRunMerge {
@Nullable
private static ManifestEntryRunMergePlan discoverRuns(
List<ManifestFileMeta> section,
- ManifestFileSorter.RowIdEntrySortKey sortKey,
+ ManifestFileSorter.RowIdSortKey sortKey,
RowType partitionType,
ManifestFile manifestFile,
CollectedDeletes deletes,
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 51f51dabfe..87fd4611e7 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
@@ -32,6 +32,7 @@ import org.apache.paimon.manifest.ManifestFile;
import org.apache.paimon.manifest.ManifestFileMeta;
import org.apache.paimon.manifest.ProjectedManifestEntry;
import org.apache.paimon.partition.PartitionPredicate;
+import org.apache.paimon.table.BucketMode;
import org.apache.paimon.types.DataType;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
@@ -155,7 +156,7 @@ public class ManifestFileSorter {
@Nullable IOManager ioManager)
throws Exception {
String sortPartitionField = options.manifestSortPartitionField();
- boolean bucketed = options.bucket() > 0;
+ boolean bucketed = options.bucket() > 0 || options.bucket() ==
BucketMode.POSTPONE_BUCKET;
boolean runMergeOptimizeEnabled =
options.manifestMergeOptimizeEnabled();
long suggestedMetaSize = options.manifestTargetSize().getBytes();
int suggestedMinMetaCount = options.manifestMergeMinCount();
@@ -1110,7 +1111,7 @@ public class ManifestFileSorter {
sorted =
ManifestEntryRunMerge.sortAndWriteFullEntries(
section,
- (RowIdEntrySortKey) ctx.sortKey,
+ (RowIdSortKey) ctx.sortKey,
ctx.partitionType,
manifestFile,
sortNewFiles,
@@ -1152,7 +1153,7 @@ public class ManifestFileSorter {
sorted =
ManifestEntryRunMerge.sortAndWriteMinorEntries(
section,
- (RowIdEntrySortKey) ctx.sortKey,
+ (RowIdSortKey) ctx.sortKey,
ctx.partitionType,
manifestFile,
sortNewFiles,
@@ -1188,14 +1189,6 @@ public class ManifestFileSorter {
return true;
}
- static ManifestSortKey createSortKey(
- boolean dataEvolutionEnabled,
- List<ManifestFileMeta> input,
- String sortPartitionField,
- RowType partitionType) {
- return createSortKey(dataEvolutionEnabled, input, sortPartitionField,
partitionType, false);
- }
-
static ManifestSortKey createSortKey(
boolean dataEvolutionEnabled,
List<ManifestFileMeta> input,
@@ -1289,11 +1282,6 @@ public class ManifestFileSorter {
InternalRow binaryManifestRow(BinaryRow row);
}
- interface RowIdEntrySortKey extends ManifestSortKey {
-
- int comparePartitions(BinaryRow left, BinaryRow right);
- }
-
private static class PartitionSortKey implements ManifestSortKey {
private final RecordComparator fieldComparator;
@@ -1458,7 +1446,7 @@ public class ManifestFileSorter {
}
}
- private static class RowIdSortKey implements RowIdEntrySortKey {
+ static class RowIdSortKey implements ManifestSortKey {
@Nullable private final RecordComparator partitionComparator;
private final InternalRow.FieldGetter[] partitionFieldGetters;
@@ -1545,8 +1533,7 @@ public class ManifestFileSorter {
return row.getRow(sortFieldNum,
ManifestEntry.MANIFEST_ROW_TYPE.getFieldCount());
}
- @Override
- public int comparePartitions(BinaryRow left, BinaryRow right) {
+ int comparePartitions(BinaryRow left, BinaryRow right) {
return partitionComparator == null ? 0 :
partitionComparator.compare(left, right);
}
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 898f238231..6d67eddfbd 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
@@ -19,6 +19,7 @@
package org.apache.paimon.manifest;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.Snapshot;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.BinaryRowWriter;
import org.apache.paimon.data.Timestamp;
@@ -30,11 +31,13 @@ import org.apache.paimon.fs.SeekableInputStream;
import org.apache.paimon.fs.SeekableInputStreamWrapper;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.operation.ManifestCompactDryRun;
import org.apache.paimon.operation.ManifestFileMerger;
import org.apache.paimon.options.Options;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.schema.FileSystemSchemaManager;
import org.apache.paimon.stats.StatsTestUtils;
+import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.types.IntType;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.FailingFileIO;
@@ -48,6 +51,7 @@ import org.junit.jupiter.api.RepeatedTest;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import org.junit.jupiter.params.provider.NullSource;
import org.junit.jupiter.params.provider.ValueSource;
@@ -80,6 +84,9 @@ import java.util.stream.LongStream;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assumptions.assumeTrue;
+import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
/** Tests for {@link ManifestFileMeta}. */
public class ManifestFileMetaTest extends ManifestFileMetaTestBase {
@@ -1310,19 +1317,22 @@ public class ManifestFileMetaTest extends
ManifestFileMetaTestBase {
.containsExactly(1, 3, 0, 2);
}
- @Test
- public void testManifestSortUsesBucketAsPrimaryKeyForBucketedTable() {
+ @ParameterizedTest
+ @ValueSource(ints = {4, -2})
+ public void testManifestSortUsesBucketAsPrimaryKeyForBucketedTable(int
bucket) {
+ int firstBucket = bucket == -2 ? -2 : 0;
List<ManifestFileMeta> input =
Arrays.asList(
makeManifest(
- makeBucketEntry("a-b1-p1", 1, 1),
makeBucketEntry("a-b0-p0", 0, 0)),
+ makeBucketEntry("a-b1-p1", 1, 1),
+ makeBucketEntry("a-first-p0", 0, firstBucket)),
makeManifest(
makeBucketEntry("b-b1-p0", 0, 1),
- makeBucketEntry("b-b0-p1", 1, 0)));
+ makeBucketEntry("b-first-p1", 1,
firstBucket)));
Options testOptions = new Options();
testOptions.set(CoreOptions.MANIFEST_SORT_ENABLED, true);
- testOptions.set(CoreOptions.BUCKET, 4);
+ testOptions.set(CoreOptions.BUCKET, bucket);
testOptions.set(CoreOptions.MANIFEST_TARGET_FILE_SIZE.key(), "1G");
testOptions.set(CoreOptions.MANIFEST_FULL_COMPACTION_FILE_SIZE.key(),
"1B");
List<ManifestFileMeta> merged =
@@ -1334,12 +1344,48 @@ public class ManifestFileMetaTest extends
ManifestFileMetaTestBase {
assertEquivalentEntries(input, merged);
List<ManifestEntry> entries = readEntries(merged);
-
assertThat(entries).extracting(ManifestEntry::bucket).containsExactly(0, 0, 1,
1);
+ assertThat(entries)
+ .extracting(ManifestEntry::bucket)
+ .containsExactly(firstBucket, firstBucket, 1, 1);
assertThat(entries)
.extracting(entry -> entry.partition().getInt(0))
.containsExactly(0, 1, 0, 1);
}
+ @ParameterizedTest
+ @ValueSource(ints = {-1, 4, -2})
+ public void testManifestSortDryRunUsesBucketRangesForBucketedTable(int
bucket) {
+ int firstBucket = bucket == -2 ? -2 : 0;
+ List<ManifestFileMeta> input =
+ Arrays.asList(
+ makeManifest(
+ makeBucketEntry("a-p0", 0, firstBucket),
+ makeBucketEntry("a-p1", 1, firstBucket)),
+ makeManifest(makeBucketEntry("b-p0", 0, 1),
makeBucketEntry("b-p1", 1, 1)));
+
+ Options testOptions = new Options();
+ testOptions.set(CoreOptions.MANIFEST_SORT_ENABLED, true);
+ testOptions.set(CoreOptions.BUCKET, bucket);
+ testOptions.set(CoreOptions.MANIFEST_TARGET_FILE_SIZE.key(), "1B");
+ testOptions.set(CoreOptions.MANIFEST_FULL_COMPACTION_FILE_SIZE.key(),
Long.MAX_VALUE + "B");
+
+ FileStoreTable table = mock(FileStoreTable.class, RETURNS_DEEP_STUBS);
+ Snapshot snapshot = mock(Snapshot.class);
+ when(table.options()).thenReturn(testOptions.toMap());
+
when(table.store().snapshotManager().latestSnapshot()).thenReturn(snapshot);
+
when(table.store().manifestListFactory().create().readDataManifests(snapshot))
+ .thenReturn(input);
+
when(table.store().manifestFileFactory().create()).thenReturn(manifestFile);
+
when(table.schema().logicalPartitionType()).thenReturn(getPartitionType());
+
+ // Overlapping partition ranges form one run only when buckets take
precedence.
+ assertThat(ManifestCompactDryRun.execute(table))
+ .endsWith(
+ bucket == -1
+ ? "Manifest sort level files: L0=0, L1=0,
L2=0, L3=1, L4=1."
+ : "Manifest sort level files: L0=0, L1=0,
L2=0, L3=0, L4=2.");
+ }
+
@Test
public void testManifestSortMinorCompactionRespectsMergeMinCount() {
List<ManifestFileMeta> input = new ArrayList<>();
@@ -1447,30 +1493,31 @@ public class ManifestFileMetaTest extends
ManifestFileMetaTestBase {
}
@ParameterizedTest
- @ValueSource(booleans = {false, true})
- public void testManifestSortWithSpillableExternalSortBuffer(boolean
bucketed) {
+ @CsvSource({"-1, false", "4, true", "-2, true"})
+ public void testManifestSortWithSpillableExternalSortBuffer(int bucket,
boolean bucketed) {
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;
- int bucket = Math.floorMod(manifest * 31 + i * 17, 4);
+ int entryBucket = Math.floorMod(manifest * 31 + i * 17, 4);
+ if (bucket == -2 && entryBucket == 3) {
+ entryBucket = -2;
+ }
entries.add(
makeBucketEntry(
String.format(
"spill-manifest-%02d-entry-%03d-payload-padding-%040d",
manifest, i, i),
partition,
- bucket));
+ entryBucket));
}
input.add(makeManifest(entries.toArray(new ManifestEntry[0])));
}
Options testOptions = new Options();
testOptions.set("manifest-sort.enabled", "true");
- if (bucketed) {
- testOptions.set(CoreOptions.BUCKET, 4);
- }
+ testOptions.set(CoreOptions.BUCKET, bucket);
testOptions.set("manifest.full-compaction-threshold-size", "1B");
testOptions.set("page-size", "1kb");
testOptions.set("sort-spill-buffer-size", "4kb");
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
index 1142f2fa16..dbd8ef66a1 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/ManifestEntryRunMergeTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/ManifestEntryRunMergeTest.java
@@ -76,9 +76,10 @@ class ManifestEntryRunMergeTest extends
ManifestFileMetaTestBase {
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);
+ ManifestFileSorter.RowIdSortKey sortKey =
+ (ManifestFileSorter.RowIdSortKey)
+ ManifestFileSorter.createSortKey(
+ true, manifests, null, partitionType, false);
CollectedDeletes deletes = new CollectedDeletes(true);
List<ManifestFileMeta> output;
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/ManifestRewriteCleanupTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/ManifestRewriteCleanupTest.java
index 2049011bc8..8247d23262 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/ManifestRewriteCleanupTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/ManifestRewriteCleanupTest.java
@@ -598,7 +598,7 @@ class ManifestRewriteCleanupTest extends
ManifestFileMetaTestBase {
private ManifestFileSorter.ManifestSortKey failingSortKey(
List<ManifestFileMeta> input, int successfulRowsBeforeFailure,
AssertionError failure) {
return new FailingManifestSortKey(
- ManifestFileSorter.createSortKey(true, input, null,
PARTITION_TYPE),
+ ManifestFileSorter.createSortKey(true, input, null,
PARTITION_TYPE, false),
successfulRowsBeforeFailure,
failure);
}
