From 665baa87204fdf5a5d3a1334c3d8e7264828c946 Mon Sep 17 00:00:00 2001
From: jianguotian <18464293+jianguotian@users.noreply.github.com>
Date: Sun, 13 Sep 2026 23:33:44 +0800
Subject: [PATCH 1/3] [core] Support bucket-first sorting for manifest files
---
docs/generated/core_configuration.html | 6 ++
.../java/org/apache/paimon/CoreOptions.java | 14 +++
.../operation/ManifestCompactDryRun.java | 3 +-
.../paimon/operation/ManifestFileSorter.java | 94 +++++++++++++++++--
.../paimon/schema/SchemaValidation.java | 5 +
.../paimon/manifest/ManifestFileMetaTest.java | 60 ++++++++++++
.../operation/ManifestEntryRunMergeTest.java | 74 +++++++++++++++
.../paimon/schema/SchemaValidationTest.java | 17 ++++
8 files changed, 262 insertions(+), 11 deletions(-)
diff --git a/docs/generated/core_configuration.html b/docs/generated/core_configuration.html
index ac5730017523..fb313aef8dd9 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -1059,6 +1059,12 @@
| Integer |
Level threshold of lookup to generate remote lookup files. Level files below this threshold will not generate remote lookup files. |
+
+ manifest-sort.bucket-first |
+ false |
+ Boolean |
+ Sort manifest entries by bucket before the configured partition field. This improves manifest pruning for bucket-key point lookups spanning many partitions, at the cost of wider partition ranges in each manifest. |
+
manifest-sort.enabled |
false |
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index 7b0665c50296..83e330f81e0e 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -589,6 +589,16 @@ public InlineElement getDescription() {
"Partition field name to sort manifest entries by. Validated by"
+ " schema validation, if not configured, defaults to the first partition field.");
+ public static final ConfigOption MANIFEST_SORT_BUCKET_FIRST =
+ key("manifest-sort.bucket-first")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Sort manifest entries by bucket before the configured partition"
+ + " field. This improves manifest pruning for bucket-key point"
+ + " lookups spanning many partitions, at the cost of wider"
+ + " partition ranges in each manifest.");
+
public static final ConfigOption MANIFEST_SORT_MAX_REWRITE_SIZE =
key("manifest-sort.max-rewrite-size")
.memoryType()
@@ -3223,6 +3233,10 @@ public String manifestSortPartitionField() {
return options.get(MANIFEST_SORT_PARTITION_FIELD);
}
+ public boolean manifestSortBucketFirst() {
+ return options.get(MANIFEST_SORT_BUCKET_FIRST);
+ }
+
public long manifestSortMaxRewriteSize() {
return options.get(MANIFEST_SORT_MAX_REWRITE_SIZE).getBytes();
}
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 4ab06f52bfdc..56a31a02ac5e 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
@@ -119,7 +119,8 @@ private static List buildLevelSortedRunsForDryRun(
options.dataEvolutionEnabled(),
manifests,
options.manifestSortPartitionField(),
- partitionType);
+ partitionType,
+ options.manifestSortBucketFirst());
ManifestFileSorter.ClassifyResult classifyResult =
ManifestFileSorter.classifyManifests(
manifests,
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 76c5b0ef5ca0..ba3856849e5d 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
@@ -155,6 +155,7 @@ static List trySortCompaction(
@Nullable IOManager ioManager)
throws Exception {
String sortPartitionField = options.manifestSortPartitionField();
+ boolean sortBucketFirst = options.manifestSortBucketFirst();
boolean runMergeOptimizeEnabled = options.manifestMergeOptimizeEnabled();
long suggestedMetaSize = options.manifestTargetSize().getBytes();
int suggestedMinMetaCount = options.manifestMergeMinCount();
@@ -173,6 +174,7 @@ static List trySortCompaction(
manifestFile,
partitionType,
sortPartitionField,
+ sortBucketFirst,
options.dataEvolutionEnabled(),
runMergeOptimizeEnabled,
suggestedMetaSize,
@@ -192,6 +194,7 @@ static List trySortCompaction(
manifestFile,
partitionType,
sortPartitionField,
+ sortBucketFirst,
options.dataEvolutionEnabled(),
runMergeOptimizeEnabled,
suggestedMetaSize,
@@ -215,6 +218,7 @@ private static Optional> tryFullCompaction(
ManifestFile manifestFile,
RowType partitionType,
String sortPartitionField,
+ boolean sortBucketFirst,
boolean dataEvolutionEnabled,
boolean runMergeOptimizeEnabled,
long suggestedMetaSize,
@@ -238,6 +242,7 @@ private static Optional> tryFullCompaction(
manifestFile,
partitionType,
sortPartitionField,
+ sortBucketFirst,
dataEvolutionEnabled,
runMergeOptimizeEnabled,
suggestedMetaSize,
@@ -321,6 +326,7 @@ private static List tryMinorCompaction(
ManifestFile manifestFile,
RowType partitionType,
String sortPartitionField,
+ boolean sortBucketFirst,
boolean dataEvolutionEnabled,
boolean runMergeOptimizeEnabled,
long suggestedMetaSize,
@@ -339,6 +345,7 @@ private static List tryMinorCompaction(
manifestFile,
partitionType,
sortPartitionField,
+ sortBucketFirst,
dataEvolutionEnabled,
runMergeOptimizeEnabled,
suggestedMetaSize,
@@ -453,6 +460,7 @@ private static CompactionContext prepareCompaction(
ManifestFile manifestFile,
RowType partitionType,
String sortPartitionField,
+ boolean sortBucketFirst,
boolean dataEvolutionEnabled,
boolean runMergeOptimizeEnabled,
long suggestedMetaSize,
@@ -464,7 +472,13 @@ private static CompactionContext prepareCompaction(
boolean useRunMergeOptimize = rowIdSort && runMergeOptimizeEnabled;
// Step 1: Resolve sort key. Data evolution tables prefer RowID ranges when available.
- ManifestSortKey sortKey = createSortKey(rowIdSort, sortPartitionField, partitionType);
+ ManifestSortKey sortKey =
+ createSortKey(
+ dataEvolutionEnabled,
+ input,
+ sortPartitionField,
+ partitionType,
+ sortBucketFirst);
// Step 2: Classify manifests into LSM files and collect delete entries.
ClassifyResult classification =
@@ -1183,14 +1197,42 @@ static ManifestSortKey createSortKey(
List input,
String sortPartitionField,
RowType partitionType) {
+ return createSortKey(dataEvolutionEnabled, input, sortPartitionField, partitionType, false);
+ }
+
+ static ManifestSortKey createSortKey(
+ boolean dataEvolutionEnabled,
+ List input,
+ String sortPartitionField,
+ RowType partitionType,
+ boolean sortBucketFirst) {
+ if (dataEvolutionEnabled && sortBucketFirst) {
+ throw new IllegalArgumentException(
+ String.format(
+ "'%s' is not supported when '%s' is enabled.",
+ CoreOptions.MANIFEST_SORT_BUCKET_FIRST.key(),
+ CoreOptions.DATA_EVOLUTION_ENABLED.key()));
+ }
+
return createSortKey(
dataEvolutionEnabled && ManifestFileMeta.allContainsRowId(input),
sortPartitionField,
- partitionType);
+ partitionType,
+ sortBucketFirst,
+ sortBucketFirst
+ && input.stream()
+ .allMatch(
+ meta ->
+ meta.minBucket() != null
+ && meta.maxBucket() != null));
}
private static ManifestSortKey createSortKey(
- boolean rowIdSort, String sortPartitionField, RowType partitionType) {
+ boolean rowIdSort,
+ String sortPartitionField,
+ RowType partitionType,
+ boolean sortBucketFirst,
+ boolean compareManifestBuckets) {
if (rowIdSort) {
// 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
@@ -1219,7 +1261,12 @@ private static ManifestSortKey createSortKey(
RecordComparator fieldComparator =
CodeGenUtils.newRecordComparator(
partitionType.getFieldTypes(), new int[] {sortFieldIndex});
- return new PartitionSortKey(fieldComparator, partitionType, sortFieldIndex);
+ return new PartitionSortKey(
+ fieldComparator,
+ partitionType,
+ sortFieldIndex,
+ sortBucketFirst,
+ compareManifestBuckets);
}
private static int[] createPartitionSortFields(
@@ -1282,36 +1329,62 @@ private static class PartitionSortKey implements ManifestSortKey {
private final RowType externalSortRowType;
private final int[] externalSortKeyFields;
private final int sortFieldNum;
+ private final boolean compareManifestBuckets;
private PartitionSortKey(
- RecordComparator fieldComparator, RowType partitionType, int sortFieldIndex) {
+ RecordComparator fieldComparator,
+ RowType partitionType,
+ int sortFieldIndex,
+ boolean sortBucketFirst,
+ boolean compareManifestBuckets) {
this.fieldComparator = fieldComparator;
+ this.compareManifestBuckets = compareManifestBuckets;
DataType sortFieldType = partitionType.getTypeAt(sortFieldIndex);
this.sortFieldGetter = InternalRow.createFieldGetter(sortFieldType, sortFieldIndex);
- this.sortFieldNum = 3;
+ this.sortFieldNum = 4;
this.externalSortRowType =
DataTypes.ROW(
sortFieldType,
+ DataTypes.INT(),
DataTypes.TINYINT(),
DataTypes.STRING(),
ManifestEntry.MANIFEST_ROW_TYPE);
- this.externalSortKeyFields = createSequentialFields(sortFieldNum);
+ this.externalSortKeyFields =
+ sortBucketFirst ? new int[] {1, 0, 2, 3} : new int[] {0, 2, 3};
}
@Override
public int compareMin(ManifestFileMeta a, ManifestFileMeta b) {
+ if (compareManifestBuckets) {
+ int bucketComparison = Integer.compare(a.minBucket(), b.minBucket());
+ if (bucketComparison != 0) {
+ return bucketComparison;
+ }
+ }
return fieldComparator.compare(
a.partitionStats().minValues(), b.partitionStats().minValues());
}
@Override
public int compareMax(ManifestFileMeta a, ManifestFileMeta b) {
+ if (compareManifestBuckets) {
+ int bucketComparison = Integer.compare(a.maxBucket(), b.maxBucket());
+ if (bucketComparison != 0) {
+ return bucketComparison;
+ }
+ }
return fieldComparator.compare(
a.partitionStats().maxValues(), b.partitionStats().maxValues());
}
@Override
public boolean isAfterMax(ManifestFileMeta file, ManifestFileMeta maxFile) {
+ if (compareManifestBuckets) {
+ int bucketComparison = Integer.compare(file.minBucket(), maxFile.maxBucket());
+ if (bucketComparison != 0) {
+ return bucketComparison > 0;
+ }
+ }
return fieldComparator.compare(
file.partitionStats().minValues(), maxFile.partitionStats().maxValues())
>= 0;
@@ -1331,13 +1404,14 @@ public int[] externalSortKeyFields() {
public void replaceExternalSortRow(
GenericRow row, ManifestEntry entry, InternalRow binaryManifestRow) {
row.setField(0, sortFieldGetter.getFieldOrNull(entry.partition()));
- row.setField(1, entry.kind().toByteValue());
+ row.setField(1, entry.bucket());
+ row.setField(2, entry.kind().toByteValue());
row.setField(
- 2,
+ 3,
entry instanceof ProjectedManifestEntry
? ((ProjectedManifestEntry) entry).file().fileNameBinary()
: BinaryString.fromString(entry.file().fileName()));
- row.setField(3, binaryManifestRow);
+ row.setField(4, binaryManifestRow);
}
@Override
diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
index 996fceaa97bf..71a0f5842495 100644
--- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
+++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
@@ -2021,6 +2021,11 @@ public static void validatePkClusteringOverride(CoreOptions options) {
private static void validateManifestSort(TableSchema schema, CoreOptions options) {
if (options.manifestSortEnabled()) {
+ checkArgument(
+ !options.dataEvolutionEnabled() || !options.manifestSortBucketFirst(),
+ "'%s' is not supported when '%s' is enabled.",
+ CoreOptions.MANIFEST_SORT_BUCKET_FIRST.key(),
+ CoreOptions.DATA_EVOLUTION_ENABLED.key());
if (!options.dataEvolutionEnabled()) {
checkArgument(
!schema.partitionKeys().isEmpty(),
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 c27d8c012fb3..94d64ec3c375 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
@@ -1286,6 +1286,60 @@ public void testManifestSortWithOverlappingPartitions() {
}
}
+ @Test
+ public void testManifestSortPreservesExistingOrderWhenBucketFirstDisabled() {
+ List input =
+ Arrays.asList(
+ makeManifest(makeBucketEntry("a-3", 0, 3), makeBucketEntry("a-1", 0, 1)),
+ makeManifest(makeBucketEntry("b-2", 0, 2), makeBucketEntry("b-0", 0, 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_FULL_COMPACTION_FILE_SIZE.key(), "1B");
+ List merged =
+ ManifestFileMerger.merge(
+ input,
+ manifestFile,
+ getPartitionType(),
+ CoreOptions.fromMap(testOptions.toMap()));
+
+ assertEquivalentEntries(input, merged);
+ assertThat(readEntries(merged))
+ .extracting(ManifestEntry::bucket)
+ .containsExactly(1, 3, 0, 2);
+ }
+
+ @Test
+ public void testManifestSortCanUseBucketAsPrimaryKey() {
+ List input =
+ Arrays.asList(
+ makeManifest(
+ makeBucketEntry("a-b1-p1", 1, 1), makeBucketEntry("a-b0-p0", 0, 0)),
+ makeManifest(
+ makeBucketEntry("b-b1-p0", 0, 1),
+ makeBucketEntry("b-b0-p1", 1, 0)));
+
+ Options testOptions = new Options();
+ testOptions.set(CoreOptions.MANIFEST_SORT_ENABLED, true);
+ testOptions.set(CoreOptions.MANIFEST_SORT_BUCKET_FIRST, true);
+ testOptions.set(CoreOptions.MANIFEST_TARGET_FILE_SIZE.key(), "1G");
+ testOptions.set(CoreOptions.MANIFEST_FULL_COMPACTION_FILE_SIZE.key(), "1B");
+ List merged =
+ ManifestFileMerger.merge(
+ input,
+ manifestFile,
+ getPartitionType(),
+ CoreOptions.fromMap(testOptions.toMap()));
+
+ assertEquivalentEntries(input, merged);
+ List entries = readEntries(merged);
+ assertThat(entries).extracting(ManifestEntry::bucket).containsExactly(0, 0, 1, 1);
+ assertThat(entries)
+ .extracting(entry -> entry.partition().getInt(0))
+ .containsExactly(0, 1, 0, 1);
+ }
+
@Test
public void testManifestSortMinorCompactionRespectsMergeMinCount() {
List input = new ArrayList<>();
@@ -2708,6 +2762,12 @@ public void testBoundaryEqualityHandling() {
}
}
+ /** Create a ManifestEntry with an explicit bucket. */
+ private ManifestEntry makeBucketEntry(String fileName, int partition, int bucket) {
+ ManifestEntry entry = makeEntry(true, fileName, partition);
+ return ManifestEntry.create(entry.kind(), entry.partition(), bucket, 240, entry.file());
+ }
+
/** Create a ManifestEntry with a 3-field partition row (region, dt, hour). */
private ManifestEntry makeMultiPartEntry(
boolean isAdd, String fileName, int region, int dt, int hour) {
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 2d08a158d5ca..c311eeec6eb4 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
@@ -38,6 +38,7 @@
import org.junit.jupiter.api.io.TempDir;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;
@@ -111,6 +112,26 @@ void testLargeFragmentedManifestsUseRunMerge() throws Exception {
.collect(Collectors.toList()));
}
+ @Test
+ void testBucketFirstManifestComparisonFallsBackForLegacyMetadata() {
+ ManifestFileMeta first = makeManifest(bucketEntry("first", 2, 0));
+ ManifestFileMeta legacy = copyWithoutBucketStats(makeManifest(bucketEntry("legacy", 1, 2)));
+ ManifestFileMeta last = makeManifest(bucketEntry("last", 0, 1));
+
+ ManifestFileSorter.ManifestSortKey sortKey =
+ ManifestFileSorter.createSortKey(
+ false, Arrays.asList(first, legacy, last), null, partitionType, true);
+
+ assertThat(sortKey.compareMin(first, legacy)).isPositive();
+ assertThat(sortKey.compareMin(legacy, last)).isPositive();
+ assertThat(sortKey.compareMin(first, last)).isPositive();
+
+ ManifestFileSorter.ManifestSortKey bucketSortKey =
+ ManifestFileSorter.createSortKey(
+ false, Arrays.asList(first, last), null, partitionType, true);
+ assertThat(bucketSortKey.compareMin(first, last)).isNegative();
+ }
+
private ManifestEntry rowIdEntry(String fileName, long firstRowId) {
return ManifestEntry.create(
FileKind.ADD,
@@ -141,6 +162,59 @@ private ManifestEntry rowIdEntry(String fileName, long firstRowId) {
null));
}
+ private ManifestEntry bucketEntry(String fileName, int partitionValue, int bucket) {
+ BinaryRow entryPartition = new BinaryRow(1);
+ BinaryRowWriter writer = new BinaryRowWriter(entryPartition);
+ writer.writeInt(0, partitionValue);
+ writer.complete();
+
+ return ManifestEntry.create(
+ FileKind.ADD,
+ entryPartition,
+ bucket,
+ 240,
+ DataFileMeta.create(
+ fileName,
+ 0,
+ 1,
+ entryPartition,
+ entryPartition,
+ StatsTestUtils.newEmptySimpleStats(),
+ StatsTestUtils.newEmptySimpleStats(),
+ 0,
+ 0,
+ 0,
+ 0,
+ Collections.emptyList(),
+ Timestamp.fromEpochMillis(200000),
+ 0L,
+ null,
+ FileSource.APPEND,
+ null,
+ null,
+ null,
+ Collections.singletonList("f0"),
+ null));
+ }
+
+ private ManifestFileMeta copyWithoutBucketStats(ManifestFileMeta meta) {
+ return new ManifestFileMeta(
+ meta.fileName(),
+ meta.fileSize(),
+ meta.numAddedFiles(),
+ meta.numDeletedFiles(),
+ meta.partitionStats(),
+ meta.schemaId(),
+ null,
+ null,
+ meta.minLevel(),
+ meta.maxLevel(),
+ meta.minRowId(),
+ meta.maxRowId(),
+ meta.totalBuckets(),
+ meta.extraFiles());
+ }
+
@Override
protected ManifestFile getManifestFile() {
return manifestFile;
diff --git a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
index c9588f4552fa..b4fa41be10b0 100644
--- a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
@@ -2047,6 +2047,23 @@ void testManifestSortValidation() {
options6,
"")))
.hasMessageContaining("is not a partition field");
+
+ // Test 7: bucket-first sorting is incompatible with RowID sorting
+ Map options7 = new HashMap<>(options4);
+ options7.put(CoreOptions.MANIFEST_SORT_BUCKET_FIRST.key(), "true");
+ assertThatThrownBy(
+ () ->
+ validateTableSchema(
+ new TableSchema(
+ 1,
+ fields,
+ 10,
+ emptyList(),
+ emptyList(),
+ options7,
+ "")))
+ .hasMessageContaining(CoreOptions.MANIFEST_SORT_BUCKET_FIRST.key())
+ .hasMessageContaining(DATA_EVOLUTION_ENABLED.key());
}
@Test
From 1cdca5e4798c42fecaac475ca24b8939d7da926d Mon Sep 17 00:00:00 2001
From: jianguotian <18464293+jianguotian@users.noreply.github.com>
Date: Mon, 14 Sep 2026 08:47:18 +0800
Subject: [PATCH 2/3] [core] Add spill coverage for bucket-first manifest
sorting
---
.../paimon/manifest/ManifestFileMetaTest.java | 39 ++++++++++++-------
1 file changed, 26 insertions(+), 13 deletions(-)
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 94d64ec3c375..a0ca91e20fe7 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
@@ -1446,26 +1446,29 @@ public void testManifestSortMaxRewriteSizeSmallerThanTargetFileSizeStillRewrites
.isTrue();
}
- @Test
- public void testManifestSortWithSpillableExternalSortBuffer() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testManifestSortWithSpillableExternalSortBuffer(boolean bucketFirst) {
List input = new ArrayList<>();
for (int manifest = 0; manifest < 4; manifest++) {
List 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);
entries.add(
- makeEntry(
- true,
+ makeBucketEntry(
String.format(
"spill-manifest-%02d-entry-%03d-payload-padding-%040d",
manifest, i, i),
- partition));
+ partition,
+ bucket));
}
input.add(makeManifest(entries.toArray(new ManifestEntry[0])));
}
Options testOptions = new Options();
testOptions.set("manifest-sort.enabled", "true");
+ testOptions.set(CoreOptions.MANIFEST_SORT_BUCKET_FIRST, bucketFirst);
testOptions.set("manifest.full-compaction-threshold-size", "1B");
testOptions.set("page-size", "1kb");
testOptions.set("sort-spill-buffer-size", "4kb");
@@ -1479,15 +1482,25 @@ public void testManifestSortWithSpillableExternalSortBuffer() {
CoreOptions.fromMap(testOptions.toMap()));
assertEquivalentEntries(input, merged);
- for (ManifestFileMeta meta : merged) {
- List 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);
+ List entries = readEntries(merged);
+ for (int i = 1; i < entries.size(); i++) {
+ ManifestEntry previous = entries.get(i - 1);
+ ManifestEntry current = entries.get(i);
+ int comparison = 0;
+ if (bucketFirst) {
+ comparison = Integer.compare(previous.bucket(), current.bucket());
+ }
+ if (comparison == 0) {
+ comparison =
+ Integer.compare(
+ previous.partition().getInt(0), current.partition().getInt(0));
+ }
+ if (comparison == 0) {
+ comparison = previous.file().fileName().compareTo(current.file().fileName());
}
+ assertThat(comparison)
+ .as("Entries should use the configured sort order after spill")
+ .isLessThanOrEqualTo(0);
}
}
From fb445e6ad1b3dd37db544fe327919ca78cce727c Mon Sep 17 00:00:00 2001
From: jianguotian <18464293+jianguotian@users.noreply.github.com>
Date: Mon, 14 Sep 2026 09:19:14 +0800
Subject: [PATCH 3/3] [core] Use bucket-first manifest sorting by default
---
docs/generated/core_configuration.html | 6 -
.../java/org/apache/paimon/CoreOptions.java | 14 --
.../operation/ManifestCompactDryRun.java | 2 +-
.../paimon/operation/ManifestFileSorter.java | 162 +++++++++++-------
.../paimon/schema/SchemaValidation.java | 5 -
.../paimon/manifest/ManifestFileMetaTest.java | 16 +-
.../operation/ManifestEntryRunMergeTest.java | 2 +-
.../paimon/schema/SchemaValidationTest.java | 17 --
8 files changed, 111 insertions(+), 113 deletions(-)
diff --git a/docs/generated/core_configuration.html b/docs/generated/core_configuration.html
index fb313aef8dd9..ac5730017523 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -1059,12 +1059,6 @@
Integer |
Level threshold of lookup to generate remote lookup files. Level files below this threshold will not generate remote lookup files. |
-
- manifest-sort.bucket-first |
- false |
- Boolean |
- Sort manifest entries by bucket before the configured partition field. This improves manifest pruning for bucket-key point lookups spanning many partitions, at the cost of wider partition ranges in each manifest. |
-
manifest-sort.enabled |
false |
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index 83e330f81e0e..7b0665c50296 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -589,16 +589,6 @@ public InlineElement getDescription() {
"Partition field name to sort manifest entries by. Validated by"
+ " schema validation, if not configured, defaults to the first partition field.");
- public static final ConfigOption MANIFEST_SORT_BUCKET_FIRST =
- key("manifest-sort.bucket-first")
- .booleanType()
- .defaultValue(false)
- .withDescription(
- "Sort manifest entries by bucket before the configured partition"
- + " field. This improves manifest pruning for bucket-key point"
- + " lookups spanning many partitions, at the cost of wider"
- + " partition ranges in each manifest.");
-
public static final ConfigOption MANIFEST_SORT_MAX_REWRITE_SIZE =
key("manifest-sort.max-rewrite-size")
.memoryType()
@@ -3233,10 +3223,6 @@ public String manifestSortPartitionField() {
return options.get(MANIFEST_SORT_PARTITION_FIELD);
}
- public boolean manifestSortBucketFirst() {
- return options.get(MANIFEST_SORT_BUCKET_FIRST);
- }
-
public long manifestSortMaxRewriteSize() {
return options.get(MANIFEST_SORT_MAX_REWRITE_SIZE).getBytes();
}
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 56a31a02ac5e..6f1e5f8e4076 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
@@ -120,7 +120,7 @@ private static List buildLevelSortedRunsForDryRun(
manifests,
options.manifestSortPartitionField(),
partitionType,
- options.manifestSortBucketFirst());
+ options.bucket() > 0);
ManifestFileSorter.ClassifyResult classifyResult =
ManifestFileSorter.classifyManifests(
manifests,
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 ba3856849e5d..51f51dabfe30 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
@@ -155,7 +155,7 @@ static List trySortCompaction(
@Nullable IOManager ioManager)
throws Exception {
String sortPartitionField = options.manifestSortPartitionField();
- boolean sortBucketFirst = options.manifestSortBucketFirst();
+ boolean bucketed = options.bucket() > 0;
boolean runMergeOptimizeEnabled = options.manifestMergeOptimizeEnabled();
long suggestedMetaSize = options.manifestTargetSize().getBytes();
int suggestedMinMetaCount = options.manifestMergeMinCount();
@@ -174,7 +174,7 @@ static List trySortCompaction(
manifestFile,
partitionType,
sortPartitionField,
- sortBucketFirst,
+ bucketed,
options.dataEvolutionEnabled(),
runMergeOptimizeEnabled,
suggestedMetaSize,
@@ -194,7 +194,7 @@ static List trySortCompaction(
manifestFile,
partitionType,
sortPartitionField,
- sortBucketFirst,
+ bucketed,
options.dataEvolutionEnabled(),
runMergeOptimizeEnabled,
suggestedMetaSize,
@@ -218,7 +218,7 @@ private static Optional> tryFullCompaction(
ManifestFile manifestFile,
RowType partitionType,
String sortPartitionField,
- boolean sortBucketFirst,
+ boolean bucketed,
boolean dataEvolutionEnabled,
boolean runMergeOptimizeEnabled,
long suggestedMetaSize,
@@ -242,7 +242,7 @@ private static Optional> tryFullCompaction(
manifestFile,
partitionType,
sortPartitionField,
- sortBucketFirst,
+ bucketed,
dataEvolutionEnabled,
runMergeOptimizeEnabled,
suggestedMetaSize,
@@ -326,7 +326,7 @@ private static List tryMinorCompaction(
ManifestFile manifestFile,
RowType partitionType,
String sortPartitionField,
- boolean sortBucketFirst,
+ boolean bucketed,
boolean dataEvolutionEnabled,
boolean runMergeOptimizeEnabled,
long suggestedMetaSize,
@@ -345,7 +345,7 @@ private static List tryMinorCompaction(
manifestFile,
partitionType,
sortPartitionField,
- sortBucketFirst,
+ bucketed,
dataEvolutionEnabled,
runMergeOptimizeEnabled,
suggestedMetaSize,
@@ -460,7 +460,7 @@ private static CompactionContext prepareCompaction(
ManifestFile manifestFile,
RowType partitionType,
String sortPartitionField,
- boolean sortBucketFirst,
+ boolean bucketed,
boolean dataEvolutionEnabled,
boolean runMergeOptimizeEnabled,
long suggestedMetaSize,
@@ -474,11 +474,7 @@ private static CompactionContext prepareCompaction(
// Step 1: Resolve sort key. Data evolution tables prefer RowID ranges when available.
ManifestSortKey sortKey =
createSortKey(
- dataEvolutionEnabled,
- input,
- sortPartitionField,
- partitionType,
- sortBucketFirst);
+ dataEvolutionEnabled, input, sortPartitionField, partitionType, bucketed);
// Step 2: Classify manifests into LSM files and collect delete entries.
ClassifyResult classification =
@@ -1205,34 +1201,8 @@ static ManifestSortKey createSortKey(
List input,
String sortPartitionField,
RowType partitionType,
- boolean sortBucketFirst) {
- if (dataEvolutionEnabled && sortBucketFirst) {
- throw new IllegalArgumentException(
- String.format(
- "'%s' is not supported when '%s' is enabled.",
- CoreOptions.MANIFEST_SORT_BUCKET_FIRST.key(),
- CoreOptions.DATA_EVOLUTION_ENABLED.key()));
- }
-
- return createSortKey(
- dataEvolutionEnabled && ManifestFileMeta.allContainsRowId(input),
- sortPartitionField,
- partitionType,
- sortBucketFirst,
- sortBucketFirst
- && input.stream()
- .allMatch(
- meta ->
- meta.minBucket() != null
- && meta.maxBucket() != null));
- }
-
- private static ManifestSortKey createSortKey(
- boolean rowIdSort,
- String sortPartitionField,
- RowType partitionType,
- boolean sortBucketFirst,
- boolean compareManifestBuckets) {
+ boolean bucketed) {
+ boolean rowIdSort = dataEvolutionEnabled && ManifestFileMeta.allContainsRowId(input);
if (rowIdSort) {
// 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
@@ -1261,12 +1231,14 @@ private static ManifestSortKey createSortKey(
RecordComparator fieldComparator =
CodeGenUtils.newRecordComparator(
partitionType.getFieldTypes(), new int[] {sortFieldIndex});
- return new PartitionSortKey(
- fieldComparator,
- partitionType,
- sortFieldIndex,
- sortBucketFirst,
- compareManifestBuckets);
+ if (bucketed) {
+ boolean compareManifestBuckets =
+ input.stream()
+ .allMatch(meta -> meta.minBucket() != null && meta.maxBucket() != null);
+ return new BucketSortKey(
+ fieldComparator, partitionType, sortFieldIndex, compareManifestBuckets);
+ }
+ return new PartitionSortKey(fieldComparator, partitionType, sortFieldIndex);
}
private static int[] createPartitionSortFields(
@@ -1329,28 +1301,98 @@ private static class PartitionSortKey implements ManifestSortKey {
private final RowType externalSortRowType;
private final int[] externalSortKeyFields;
private final int sortFieldNum;
- private final boolean compareManifestBuckets;
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(),
+ ManifestEntry.MANIFEST_ROW_TYPE);
+ this.externalSortKeyFields = createSequentialFields(sortFieldNum);
+ }
+
+ @Override
+ public int compareMin(ManifestFileMeta a, ManifestFileMeta b) {
+ return fieldComparator.compare(
+ a.partitionStats().minValues(), b.partitionStats().minValues());
+ }
+
+ @Override
+ public int compareMax(ManifestFileMeta a, ManifestFileMeta b) {
+ return fieldComparator.compare(
+ a.partitionStats().maxValues(), b.partitionStats().maxValues());
+ }
+
+ @Override
+ public boolean isAfterMax(ManifestFileMeta file, ManifestFileMeta maxFile) {
+ return fieldComparator.compare(
+ file.partitionStats().minValues(), maxFile.partitionStats().maxValues())
+ >= 0;
+ }
+
+ @Override
+ public RowType externalSortRowType() {
+ return externalSortRowType;
+ }
+
+ @Override
+ public int[] externalSortKeyFields() {
+ return externalSortKeyFields;
+ }
+
+ @Override
+ public void replaceExternalSortRow(
+ GenericRow row, ManifestEntry entry, InternalRow binaryManifestRow) {
+ row.setField(0, sortFieldGetter.getFieldOrNull(entry.partition()));
+ row.setField(1, entry.kind().toByteValue());
+ row.setField(
+ 2,
+ entry instanceof ProjectedManifestEntry
+ ? ((ProjectedManifestEntry) entry).file().fileNameBinary()
+ : BinaryString.fromString(entry.file().fileName()));
+ row.setField(3, binaryManifestRow);
+ }
+
+ @Override
+ public InternalRow binaryManifestRow(BinaryRow row) {
+ return row.getRow(sortFieldNum, ManifestEntry.MANIFEST_ROW_TYPE.getFieldCount());
+ }
+ }
+
+ private static class BucketSortKey implements ManifestSortKey {
+
+ private final PartitionSortKey partitionSortKey;
+ private final InternalRow.FieldGetter sortFieldGetter;
+ private final RowType externalSortRowType;
+ private final int[] externalSortKeyFields;
+ private final int sortFieldNum;
+ private final boolean compareManifestBuckets;
+
+ private BucketSortKey(
RecordComparator fieldComparator,
RowType partitionType,
int sortFieldIndex,
- boolean sortBucketFirst,
boolean compareManifestBuckets) {
- this.fieldComparator = fieldComparator;
+ this.partitionSortKey =
+ new PartitionSortKey(fieldComparator, partitionType, sortFieldIndex);
this.compareManifestBuckets = compareManifestBuckets;
DataType sortFieldType = partitionType.getTypeAt(sortFieldIndex);
this.sortFieldGetter = InternalRow.createFieldGetter(sortFieldType, sortFieldIndex);
this.sortFieldNum = 4;
this.externalSortRowType =
DataTypes.ROW(
- sortFieldType,
DataTypes.INT(),
+ sortFieldType,
DataTypes.TINYINT(),
DataTypes.STRING(),
ManifestEntry.MANIFEST_ROW_TYPE);
- this.externalSortKeyFields =
- sortBucketFirst ? new int[] {1, 0, 2, 3} : new int[] {0, 2, 3};
+ this.externalSortKeyFields = createSequentialFields(sortFieldNum);
}
@Override
@@ -1361,8 +1403,7 @@ public int compareMin(ManifestFileMeta a, ManifestFileMeta b) {
return bucketComparison;
}
}
- return fieldComparator.compare(
- a.partitionStats().minValues(), b.partitionStats().minValues());
+ return partitionSortKey.compareMin(a, b);
}
@Override
@@ -1373,8 +1414,7 @@ public int compareMax(ManifestFileMeta a, ManifestFileMeta b) {
return bucketComparison;
}
}
- return fieldComparator.compare(
- a.partitionStats().maxValues(), b.partitionStats().maxValues());
+ return partitionSortKey.compareMax(a, b);
}
@Override
@@ -1385,9 +1425,7 @@ public boolean isAfterMax(ManifestFileMeta file, ManifestFileMeta maxFile) {
return bucketComparison > 0;
}
}
- return fieldComparator.compare(
- file.partitionStats().minValues(), maxFile.partitionStats().maxValues())
- >= 0;
+ return partitionSortKey.isAfterMax(file, maxFile);
}
@Override
@@ -1403,8 +1441,8 @@ public int[] externalSortKeyFields() {
@Override
public void replaceExternalSortRow(
GenericRow row, ManifestEntry entry, InternalRow binaryManifestRow) {
- row.setField(0, sortFieldGetter.getFieldOrNull(entry.partition()));
- row.setField(1, entry.bucket());
+ row.setField(0, entry.bucket());
+ row.setField(1, sortFieldGetter.getFieldOrNull(entry.partition()));
row.setField(2, entry.kind().toByteValue());
row.setField(
3,
diff --git a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
index 71a0f5842495..996fceaa97bf 100644
--- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
+++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java
@@ -2021,11 +2021,6 @@ public static void validatePkClusteringOverride(CoreOptions options) {
private static void validateManifestSort(TableSchema schema, CoreOptions options) {
if (options.manifestSortEnabled()) {
- checkArgument(
- !options.dataEvolutionEnabled() || !options.manifestSortBucketFirst(),
- "'%s' is not supported when '%s' is enabled.",
- CoreOptions.MANIFEST_SORT_BUCKET_FIRST.key(),
- CoreOptions.DATA_EVOLUTION_ENABLED.key());
if (!options.dataEvolutionEnabled()) {
checkArgument(
!schema.partitionKeys().isEmpty(),
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 a0ca91e20fe7..898f238231ed 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
@@ -1287,7 +1287,7 @@ public void testManifestSortWithOverlappingPartitions() {
}
@Test
- public void testManifestSortPreservesExistingOrderWhenBucketFirstDisabled() {
+ public void testManifestSortPreservesExistingOrderForUnawareBucketTable() {
List input =
Arrays.asList(
makeManifest(makeBucketEntry("a-3", 0, 3), makeBucketEntry("a-1", 0, 1)),
@@ -1311,7 +1311,7 @@ public void testManifestSortPreservesExistingOrderWhenBucketFirstDisabled() {
}
@Test
- public void testManifestSortCanUseBucketAsPrimaryKey() {
+ public void testManifestSortUsesBucketAsPrimaryKeyForBucketedTable() {
List input =
Arrays.asList(
makeManifest(
@@ -1322,7 +1322,7 @@ public void testManifestSortCanUseBucketAsPrimaryKey() {
Options testOptions = new Options();
testOptions.set(CoreOptions.MANIFEST_SORT_ENABLED, true);
- testOptions.set(CoreOptions.MANIFEST_SORT_BUCKET_FIRST, true);
+ testOptions.set(CoreOptions.BUCKET, 4);
testOptions.set(CoreOptions.MANIFEST_TARGET_FILE_SIZE.key(), "1G");
testOptions.set(CoreOptions.MANIFEST_FULL_COMPACTION_FILE_SIZE.key(), "1B");
List merged =
@@ -1448,7 +1448,7 @@ public void testManifestSortMaxRewriteSizeSmallerThanTargetFileSizeStillRewrites
@ParameterizedTest
@ValueSource(booleans = {false, true})
- public void testManifestSortWithSpillableExternalSortBuffer(boolean bucketFirst) {
+ public void testManifestSortWithSpillableExternalSortBuffer(boolean bucketed) {
List input = new ArrayList<>();
for (int manifest = 0; manifest < 4; manifest++) {
List entries = new ArrayList<>();
@@ -1468,7 +1468,9 @@ public void testManifestSortWithSpillableExternalSortBuffer(boolean bucketFirst)
Options testOptions = new Options();
testOptions.set("manifest-sort.enabled", "true");
- testOptions.set(CoreOptions.MANIFEST_SORT_BUCKET_FIRST, bucketFirst);
+ if (bucketed) {
+ testOptions.set(CoreOptions.BUCKET, 4);
+ }
testOptions.set("manifest.full-compaction-threshold-size", "1B");
testOptions.set("page-size", "1kb");
testOptions.set("sort-spill-buffer-size", "4kb");
@@ -1487,7 +1489,7 @@ public void testManifestSortWithSpillableExternalSortBuffer(boolean bucketFirst)
ManifestEntry previous = entries.get(i - 1);
ManifestEntry current = entries.get(i);
int comparison = 0;
- if (bucketFirst) {
+ if (bucketed) {
comparison = Integer.compare(previous.bucket(), current.bucket());
}
if (comparison == 0) {
@@ -1499,7 +1501,7 @@ public void testManifestSortWithSpillableExternalSortBuffer(boolean bucketFirst)
comparison = previous.file().fileName().compareTo(current.file().fileName());
}
assertThat(comparison)
- .as("Entries should use the configured sort order after spill")
+ .as("Entries should use the table's sort order after spill")
.isLessThanOrEqualTo(0);
}
}
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 c311eeec6eb4..1142f2fa1664 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
@@ -113,7 +113,7 @@ void testLargeFragmentedManifestsUseRunMerge() throws Exception {
}
@Test
- void testBucketFirstManifestComparisonFallsBackForLegacyMetadata() {
+ void testBucketedManifestComparisonFallsBackForLegacyMetadata() {
ManifestFileMeta first = makeManifest(bucketEntry("first", 2, 0));
ManifestFileMeta legacy = copyWithoutBucketStats(makeManifest(bucketEntry("legacy", 1, 2)));
ManifestFileMeta last = makeManifest(bucketEntry("last", 0, 1));
diff --git a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
index b4fa41be10b0..c9588f4552fa 100644
--- a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaValidationTest.java
@@ -2047,23 +2047,6 @@ void testManifestSortValidation() {
options6,
"")))
.hasMessageContaining("is not a partition field");
-
- // Test 7: bucket-first sorting is incompatible with RowID sorting
- Map options7 = new HashMap<>(options4);
- options7.put(CoreOptions.MANIFEST_SORT_BUCKET_FIRST.key(), "true");
- assertThatThrownBy(
- () ->
- validateTableSchema(
- new TableSchema(
- 1,
- fields,
- 10,
- emptyList(),
- emptyList(),
- options7,
- "")))
- .hasMessageContaining(CoreOptions.MANIFEST_SORT_BUCKET_FIRST.key())
- .hasMessageContaining(DATA_EVOLUTION_ENABLED.key());
}
@Test