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 6b80445762 [core] Support bucket-first sorting for manifest files
(#9792)
6b80445762 is described below
commit 6b804457620f991ee9fec9cf8d151b77344eab56
Author: jianguotian <[email protected]>
AuthorDate: Mon Sep 14 09:51:21 2026 +0800
[core] Support bucket-first sorting for manifest files (#9792)
---
.../paimon/operation/ManifestCompactDryRun.java | 3 +-
.../paimon/operation/ManifestFileSorter.java | 126 +++++++++++++++++++--
.../paimon/manifest/ManifestFileMetaTest.java | 101 ++++++++++++++---
.../operation/ManifestEntryRunMergeTest.java | 74 ++++++++++++
4 files changed, 283 insertions(+), 21 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 4ab06f52bf..6f1e5f8e40 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 @@ public class ManifestCompactDryRun {
options.dataEvolutionEnabled(),
manifests,
options.manifestSortPartitionField(),
- partitionType);
+ partitionType,
+ 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 76c5b0ef5c..51f51dabfe 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 @@ public class ManifestFileSorter {
@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();
@@ -173,6 +174,7 @@ public class ManifestFileSorter {
manifestFile,
partitionType,
sortPartitionField,
+ bucketed,
options.dataEvolutionEnabled(),
runMergeOptimizeEnabled,
suggestedMetaSize,
@@ -192,6 +194,7 @@ public class ManifestFileSorter {
manifestFile,
partitionType,
sortPartitionField,
+ bucketed,
options.dataEvolutionEnabled(),
runMergeOptimizeEnabled,
suggestedMetaSize,
@@ -215,6 +218,7 @@ public class ManifestFileSorter {
ManifestFile manifestFile,
RowType partitionType,
String sortPartitionField,
+ boolean bucketed,
boolean dataEvolutionEnabled,
boolean runMergeOptimizeEnabled,
long suggestedMetaSize,
@@ -238,6 +242,7 @@ public class ManifestFileSorter {
manifestFile,
partitionType,
sortPartitionField,
+ bucketed,
dataEvolutionEnabled,
runMergeOptimizeEnabled,
suggestedMetaSize,
@@ -321,6 +326,7 @@ public class ManifestFileSorter {
ManifestFile manifestFile,
RowType partitionType,
String sortPartitionField,
+ boolean bucketed,
boolean dataEvolutionEnabled,
boolean runMergeOptimizeEnabled,
long suggestedMetaSize,
@@ -339,6 +345,7 @@ public class ManifestFileSorter {
manifestFile,
partitionType,
sortPartitionField,
+ bucketed,
dataEvolutionEnabled,
runMergeOptimizeEnabled,
suggestedMetaSize,
@@ -453,6 +460,7 @@ public class ManifestFileSorter {
ManifestFile manifestFile,
RowType partitionType,
String sortPartitionField,
+ boolean bucketed,
boolean dataEvolutionEnabled,
boolean runMergeOptimizeEnabled,
long suggestedMetaSize,
@@ -464,7 +472,9 @@ public class ManifestFileSorter {
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 =
@@ -1183,14 +1193,16 @@ public class ManifestFileSorter {
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
@@ -1219,6 +1231,13 @@ public class ManifestFileSorter {
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);
}
@@ -1346,6 +1365,99 @@ public class ManifestFileSorter {
}
}
+ 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;
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 c27d8c012f..898f238231 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 class ManifestFileMetaTest extends
ManifestFileMetaTestBase {
}
}
+ @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<>();
@@ -1392,26 +1446,31 @@ public class ManifestFileMetaTest extends
ManifestFileMetaTestBase {
.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");
@@ -1425,15 +1484,25 @@ public class ManifestFileMetaTest extends
ManifestFileMetaTestBase {
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);
}
}
@@ -2708,6 +2777,12 @@ public class ManifestFileMetaTest extends
ManifestFileMetaTestBase {
}
}
+ /** 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 2d08a158d5..1142f2fa16 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.Test;
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 @@ class ManifestEntryRunMergeTest extends
ManifestFileMetaTestBase {
.collect(Collectors.toList()));
}
+ @Test
+ void testBucketedManifestComparisonFallsBackForLegacyMetadata() {
+ 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 @@ class ManifestEntryRunMergeTest extends
ManifestFileMetaTestBase {
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;
