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 8f47383d8f [core] Support manifest-sort.partition-field for Data
Evolution Table (#8340)
8f47383d8f is described below
commit 8f47383d8fdf582c3d4267aa6d8a5642a9675735
Author: umi <[email protected]>
AuthorDate: Wed Jun 24 14:38:17 2026 +0800
[core] Support manifest-sort.partition-field for Data Evolution Table
(#8340)
---
.../paimon/operation/ManifestFileSorter.java | 32 ++++-
.../paimon/manifest/ManifestFileMetaTest.java | 139 +++++++++++++++++++++
2 files changed, 165 insertions(+), 6 deletions(-)
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 33720c4aeb..71edb1924e 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
@@ -1099,13 +1099,11 @@ public class ManifestFileSorter {
String sortPartitionField,
RowType partitionType) {
if (dataEvolutionEnabled && ManifestFileMeta.allContainsRowId(input)) {
- // RowID sorting uses the full partition row as the primary key to
preserve partition
- // locality, then orders files by RowID. The optional
manifest-sort.partition-field is
- // only used by the partition-sort fallback when RowID stats are
incomplete.
+ // 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
+ // orders files by RowID.
RecordComparator partitionComparator =
- partitionType.getFieldCount() == 0
- ? null
- :
CodeGenUtils.newRecordComparator(partitionType.getFieldTypes());
+ createPartitionComparator(sortPartitionField,
partitionType);
return new RowIdSortKey(partitionComparator);
}
@@ -1129,6 +1127,28 @@ public class ManifestFileSorter {
return new PartitionSortKey(fieldComparator);
}
+ @Nullable
+ private static RecordComparator createPartitionComparator(
+ String sortPartitionField, RowType partitionType) {
+ if (sortPartitionField != null && !sortPartitionField.isEmpty()) {
+ int sortFieldIndex =
partitionType.getFieldNames().indexOf(sortPartitionField);
+ if (sortFieldIndex < 0) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Cannot resolve sort field '%s' for manifest
sort rewrite.",
+ sortPartitionField));
+ }
+ return CodeGenUtils.newRecordComparator(
+ partitionType.getFieldTypes(), new int[] {sortFieldIndex});
+ }
+
+ if (partitionType.getFieldCount() == 0) {
+ return null;
+ }
+
+ return CodeGenUtils.newRecordComparator(partitionType.getFieldTypes());
+ }
+
interface ManifestSortKey {
int compareMin(ManifestFileMeta a, ManifestFileMeta b);
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 0357a26e52..42638aa24f 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
@@ -1118,6 +1118,67 @@ public class ManifestFileMetaTest extends
ManifestFileMetaTestBase {
}
}
+ @Test
+ public void
testDataEvolutionManifestSortUsesConfiguredPartitionFieldBeforeRowId() {
+ RowType multiPartitionType = RowType.of(new IntType(), new IntType(),
new IntType());
+ ManifestFile multiPartManifestFile =
createManifestFileForPartitionType(multiPartitionType);
+
+ List<ManifestFileMeta> input = new ArrayList<>();
+ input.add(
+ multiPartManifestFile
+ .write(
+ Arrays.asList(
+ makeMultiPartRowIdEntry(
+ "region10-dt2-row30", 10, 2,
0, 30, 5),
+ makeMultiPartRowIdEntry(
+ "region20-dt1-row5", 20, 1, 0,
5, 5)))
+ .get(0));
+ input.add(
+ multiPartManifestFile
+ .write(
+ Arrays.asList(
+ makeMultiPartRowIdEntry(
+ "region5-dt2-row10", 5, 2, 0,
10, 5),
+ makeMultiPartRowIdEntry(
+ "region0-dt1-row20", 0, 1, 0,
20, 5)))
+ .get(0));
+
+ Options defaultOptions = new Options();
+ defaultOptions.set("manifest-sort.enabled", "true");
+ defaultOptions.set("data-evolution.enabled", "true");
+ defaultOptions.set("manifest.full-compaction-threshold-size", "1B");
+ List<ManifestFileMeta> sortedByFullPartition =
+ ManifestFileMerger.merge(
+ input,
+ multiPartManifestFile,
+ multiPartitionType,
+ CoreOptions.fromMap(defaultOptions.toMap()));
+ assertThat(readFileNames(multiPartManifestFile, sortedByFullPartition))
+ .containsExactly(
+ "region0-dt1-row20",
+ "region5-dt2-row10",
+ "region10-dt2-row30",
+ "region20-dt1-row5");
+
+ Options configuredFieldOptions = new Options();
+ configuredFieldOptions.set("manifest-sort.enabled", "true");
+ configuredFieldOptions.set("manifest-sort.partition-field", "f1");
+ configuredFieldOptions.set("data-evolution.enabled", "true");
+ configuredFieldOptions.set("manifest.full-compaction-threshold-size",
"1B");
+ List<ManifestFileMeta> sortedByDt =
+ ManifestFileMerger.merge(
+ input,
+ multiPartManifestFile,
+ multiPartitionType,
+ CoreOptions.fromMap(configuredFieldOptions.toMap()));
+ assertThat(readFileNames(multiPartManifestFile, sortedByDt))
+ .containsExactly(
+ "region20-dt1-row5",
+ "region0-dt1-row20",
+ "region5-dt2-row10",
+ "region10-dt2-row30");
+ }
+
@Test
public void
testDataEvolutionManifestSortFallsBackToPartitionWhenRowIdStatsMissing() {
List<ManifestFileMeta> input = new ArrayList<>();
@@ -1621,6 +1682,84 @@ public class ManifestFileMetaTest extends
ManifestFileMetaTestBase {
null));
}
+ private ManifestFile createManifestFileForPartitionType(RowType
partitionType) {
+ Path path = new Path(tempDir.toString());
+ FileIO fileIO = FileIOFinder.find(path);
+ return new ManifestFile.Factory(
+ fileIO,
+ new SchemaManager(fileIO, path),
+ partitionType,
+ avro,
+ "zstd",
+ new FileStorePathFactory(
+ path,
+ partitionType,
+ "default",
+ CoreOptions.FILE_FORMAT.defaultValue(),
+ CoreOptions.DATA_FILE_PREFIX.defaultValue(),
+
CoreOptions.CHANGELOG_FILE_PREFIX.defaultValue(),
+
CoreOptions.PARTITION_GENERATE_LEGACY_NAME.defaultValue(),
+
CoreOptions.FILE_SUFFIX_INCLUDE_COMPRESSION.defaultValue(),
+ CoreOptions.FILE_COMPRESSION.defaultValue(),
+ null,
+ null,
+ CoreOptions.ExternalPathStrategy.NONE,
+ null,
+ false,
+ null),
+ Long.MAX_VALUE,
+ null)
+ .create();
+ }
+
+ private ManifestEntry makeMultiPartRowIdEntry(
+ String fileName, int region, int dt, int hour, long firstRowId,
long rowCount) {
+ BinaryRow binaryRow = new BinaryRow(3);
+ BinaryRowWriter writer = new BinaryRowWriter(binaryRow);
+ writer.writeInt(0, region);
+ writer.writeInt(1, dt);
+ writer.writeInt(2, hour);
+ writer.complete();
+
+ return ManifestEntry.create(
+ FileKind.ADD,
+ binaryRow,
+ 0,
+ 0,
+ DataFileMeta.create(
+ fileName,
+ 0,
+ rowCount,
+ binaryRow,
+ binaryRow,
+ StatsTestUtils.newEmptySimpleStats(),
+ StatsTestUtils.newEmptySimpleStats(),
+ 0,
+ 0,
+ 0,
+ 0,
+ Collections.emptyList(),
+ Timestamp.fromEpochMillis(200000),
+ 0L,
+ null,
+ FileSource.APPEND,
+ null,
+ null,
+ firstRowId,
+ Collections.singletonList("f0")));
+ }
+
+ private List<String> readFileNames(
+ ManifestFile manifestFile, List<ManifestFileMeta> manifestMetas) {
+ List<String> fileNames = new ArrayList<>();
+ for (ManifestFileMeta meta : manifestMetas) {
+ for (ManifestEntry entry : manifestFile.read(meta.fileName(),
meta.fileSize())) {
+ fileNames.add(entry.file().fileName());
+ }
+ }
+ return fileNames;
+ }
+
/** Create a ManifestEntry with row ID metadata for data evolution
manifest sort tests. */
private ManifestEntry makeRowIdEntry(
boolean isAdd, String fileName, int partition, long firstRowId,
long rowCount) {
