Skip to content

[core] Support bucket-first sorting for manifest files #9792

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
JingsongLi merged 3 commits into from
Sep 14, 2026
Merged

[core] Support bucket-first sorting for manifest files #9792

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
665baa8
[core] Support bucket-first sorting for manifest files
jianguotian Sep 13, 2026
1cdca5e
[core] Add spill coverage for bucket-first manifest sorting
jianguotian Sep 14, 2026
fb445e6
[core] Use bucket-first manifest sorting by default
jianguotian Sep 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion paimon-core/src/main/java/org/apache/paimon/operation/ManifestCompactDryRun.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -119,7 +119,8 @@ private static List<ManifestAdjacentSortedRun> buildLevelSortedRunsForDryRun(
options.dataEvolutionEnabled(),
manifests,
options.manifestSortPartitionField(),
partitionType);
partitionType,
options.bucket() > 0);
ManifestFileSorter.ClassifyResult classifyResult =
ManifestFileSorter.classifyManifests(
manifests,
Expand Down
126 changes: 119 additions & 7 deletions paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileSorter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -155,6 +155,7 @@ static List<ManifestFileMeta> trySortCompaction(
@Nullable IOManager ioManager)
throws Exception {
String sortPartitionField = options.manifestSortPartitionField();
boolean bucketed = options.bucket() > 0;
boolean runMergeOptimizeEnabled = options.manifestMergeOptimizeEnabled();
long suggestedMetaSize = options.manifestTargetSize().getBytes();
int suggestedMinMetaCount = options.manifestMergeMinCount();
Expand All @@ -173,6 +174,7 @@ static List<ManifestFileMeta> trySortCompaction(
manifestFile,
partitionType,
sortPartitionField,
bucketed,
options.dataEvolutionEnabled(),
runMergeOptimizeEnabled,
suggestedMetaSize,
Expand All @@ -192,6 +194,7 @@ static List<ManifestFileMeta> trySortCompaction(
manifestFile,
partitionType,
sortPartitionField,
bucketed,
options.dataEvolutionEnabled(),
runMergeOptimizeEnabled,
suggestedMetaSize,
Expand All @@ -215,6 +218,7 @@ private static Optional<List<ManifestFileMeta>> tryFullCompaction(
ManifestFile manifestFile,
RowType partitionType,
String sortPartitionField,
boolean bucketed,
boolean dataEvolutionEnabled,
boolean runMergeOptimizeEnabled,
long suggestedMetaSize,
Expand All @@ -238,6 +242,7 @@ private static Optional<List<ManifestFileMeta>> tryFullCompaction(
manifestFile,
partitionType,
sortPartitionField,
bucketed,
dataEvolutionEnabled,
runMergeOptimizeEnabled,
suggestedMetaSize,
Expand Down Expand Up @@ -321,6 +326,7 @@ private static List<ManifestFileMeta> tryMinorCompaction(
ManifestFile manifestFile,
RowType partitionType,
String sortPartitionField,
boolean bucketed,
boolean dataEvolutionEnabled,
boolean runMergeOptimizeEnabled,
long suggestedMetaSize,
Expand All @@ -339,6 +345,7 @@ private static List<ManifestFileMeta> tryMinorCompaction(
manifestFile,
partitionType,
sortPartitionField,
bucketed,
dataEvolutionEnabled,
runMergeOptimizeEnabled,
suggestedMetaSize,
Expand Down Expand Up @@ -453,6 +460,7 @@ private static CompactionContext prepareCompaction(
ManifestFile manifestFile,
RowType partitionType,
String sortPartitionField,
boolean bucketed,
boolean dataEvolutionEnabled,
boolean runMergeOptimizeEnabled,
long suggestedMetaSize,
Expand All @@ -464,7 +472,9 @@ 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, bucketed);

// Step 2: Classify manifests into LSM files and collect delete entries.
ClassifyResult classification =
Expand Down Expand Up @@ -1183,14 +1193,16 @@ static ManifestSortKey createSortKey(
List<ManifestFileMeta> input,
String sortPartitionField,
RowType partitionType) {
return createSortKey(
dataEvolutionEnabled && ManifestFileMeta.allContainsRowId(input),
sortPartitionField,
partitionType);
return createSortKey(dataEvolutionEnabled, input, sortPartitionField, partitionType, false);
}

private static ManifestSortKey createSortKey(
boolean rowIdSort, String sortPartitionField, RowType partitionType) {
static ManifestSortKey createSortKey(
boolean dataEvolutionEnabled,
List<ManifestFileMeta> input,
String sortPartitionField,
RowType partitionType,
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
Expand Down Expand Up @@ -1219,6 +1231,13 @@ private static ManifestSortKey createSortKey(
RecordComparator fieldComparator =
CodeGenUtils.newRecordComparator(
partitionType.getFieldTypes(), new int[] {sortFieldIndex});
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);
}

Expand Down Expand Up @@ -1346,6 +1365,99 @@ public InternalRow binaryManifestRow(BinaryRow row) {
}
}

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 compareManifestBuckets) {
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(
DataTypes.INT(),
sortFieldType,
DataTypes.TINYINT(),
DataTypes.STRING(),
ManifestEntry.MANIFEST_ROW_TYPE);
this.externalSortKeyFields = createSequentialFields(sortFieldNum);
}

@Override
public int compareMin(ManifestFileMeta a, ManifestFileMeta b) {
if (compareManifestBuckets) {
int bucketComparison = Integer.compare(a.minBucket(), b.minBucket());
if (bucketComparison != 0) {
return bucketComparison;
}
}
return partitionSortKey.compareMin(a, b);
}

@Override
public int compareMax(ManifestFileMeta a, ManifestFileMeta b) {
if (compareManifestBuckets) {
int bucketComparison = Integer.compare(a.maxBucket(), b.maxBucket());
if (bucketComparison != 0) {
return bucketComparison;
}
}
return partitionSortKey.compareMax(a, b);
}

@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 partitionSortKey.isAfterMax(file, maxFile);
}

@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, entry.bucket());
row.setField(1, sortFieldGetter.getFieldOrNull(entry.partition()));
row.setField(2, entry.kind().toByteValue());
row.setField(
3,
entry instanceof ProjectedManifestEntry
? ((ProjectedManifestEntry) entry).file().fileNameBinary()
: BinaryString.fromString(entry.file().fileName()));
row.setField(4, binaryManifestRow);
}

@Override
public InternalRow binaryManifestRow(BinaryRow row) {
return row.getRow(sortFieldNum, ManifestEntry.MANIFEST_ROW_TYPE.getFieldCount());
}
}

private static class RowIdSortKey implements RowIdEntrySortKey {

@Nullable private final RecordComparator partitionComparator;
Expand Down
101 changes: 88 additions & 13 deletions paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1286,6 +1286,60 @@ public void testManifestSortWithOverlappingPartitions() {
}
}

@Test
public void testManifestSortPreservesExistingOrderForUnawareBucketTable() {
List<ManifestFileMeta> 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<ManifestFileMeta> 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 testManifestSortUsesBucketAsPrimaryKeyForBucketedTable() {
List<ManifestFileMeta> 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.BUCKET, 4);
testOptions.set(CoreOptions.MANIFEST_TARGET_FILE_SIZE.key(), "1G");
testOptions.set(CoreOptions.MANIFEST_FULL_COMPACTION_FILE_SIZE.key(), "1B");
List<ManifestFileMeta> merged =
ManifestFileMerger.merge(
input,
manifestFile,
getPartitionType(),
CoreOptions.fromMap(testOptions.toMap()));

assertEquivalentEntries(input, merged);
List<ManifestEntry> 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<ManifestFileMeta> input = new ArrayList<>();
Expand Down Expand Up @@ -1392,26 +1446,31 @@ public void testManifestSortMaxRewriteSizeSmallerThanTargetFileSizeStillRewrites
.isTrue();
}

@Test
public void testManifestSortWithSpillableExternalSortBuffer() {
@ParameterizedTest
@ValueSource(booleans = {false, true})
public void testManifestSortWithSpillableExternalSortBuffer(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);
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");
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");
Expand All @@ -1425,15 +1484,25 @@ public void testManifestSortWithSpillableExternalSortBuffer() {
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);
List<ManifestEntry> 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 (bucketed) {
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 table's sort order after spill")
.isLessThanOrEqualTo(0);
}
}

Expand Down Expand Up @@ -2708,6 +2777,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) {
Expand Down
Loading
Loading