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 50f28639d8 [core] Sort data evolution manifests by RowID (#8329)
50f28639d8 is described below
commit 50f28639d8e6036473b2d595f873c66b554f1804
Author: Jingsong Lee <[email protected]>
AuthorDate: Wed Jun 24 09:25:10 2026 +0800
[core] Sort data evolution manifests by RowID (#8329)
Support manifest sort rewrite for data evolution tables by using
RowID-aware sort keys. Partitioned data evolution tables sort entries by
partition first, then RowID range, while non-partitioned tables sort
directly by RowID.
---
.../operation/ManifestAdjacentSortedRun.java | 14 +-
.../paimon/operation/ManifestFileMerger.java | 10 +-
.../paimon/operation/ManifestFileSorter.java | 297 ++++++++++++++++-----
.../org/apache/paimon/schema/SchemaValidation.java | 10 +-
.../paimon/manifest/ManifestFileMetaTest.java | 176 ++++++++++++
.../manifest/NoPartitionManifestFileMetaTest.java | 67 +++++
.../apache/paimon/schema/SchemaValidationTest.java | 59 ++++
7 files changed, 545 insertions(+), 88 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestAdjacentSortedRun.java
b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestAdjacentSortedRun.java
index ca0797c213..b076387ab2 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestAdjacentSortedRun.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestAdjacentSortedRun.java
@@ -26,15 +26,13 @@ import java.util.Objects;
import java.util.stream.Collectors;
/**
- * A {@code ManifestAdjacentSortedRun} is a list of {@link ManifestFileMeta}s
sorted by a single
- * partition field (the configured manifest sort field). The intervals {@code
- * [partitionStats.minValues[k], partitionStats.maxValues[k]]} of these
manifests do not overlap on
- * field {@code k}, where {@code k} is the configured sort field index.
+ * A {@code ManifestAdjacentSortedRun} is a list of {@link ManifestFileMeta}s
sorted by manifest
+ * sort key. The sort-key intervals of these manifests do not overlap.
*
- * <p><b>Boundary Equality:</b> Files with boundary-touching intervals (min ==
previous.max) are
- * considered non-overlapping and can be placed in the same SortedRun. This
reduces the number of
- * runs and improves compaction efficiency. However, such files may be
separated into different
- * Sections during splitIntoSections to avoid merge-sort overhead.
+ * <p><b>Boundary Equality:</b> Partition-field sorting treats
boundary-touching intervals (min ==
+ * previous.max) as non-overlapping, so they can be placed in the same
SortedRun. This reduces the
+ * number of runs and improves compaction efficiency. RowID sorting treats
row-id ranges as
+ * inclusive, so boundary-touching row-id ranges are considered overlapping.
*/
public class ManifestAdjacentSortedRun {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileMerger.java
b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileMerger.java
index f899aa7178..f1caa03bcf 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileMerger.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileMerger.java
@@ -46,6 +46,7 @@ import java.util.Set;
import java.util.function.Function;
import static java.util.Collections.singletonList;
+import static org.apache.paimon.manifest.ManifestFileMeta.allContainsRowId;
import static
org.apache.paimon.utils.ManifestReadThreadPool.sequentialBatchedExecute;
import static org.apache.paimon.utils.Preconditions.checkArgument;
@@ -75,9 +76,12 @@ public class ManifestFileMerger {
List<ManifestFileMeta> newFilesForAbort = new ArrayList<>();
try {
- // If manifest-sort.enabled is enabled and there are partition
fields, use
- // trySortRewrite
- if (options.manifestSortEnabled() && partitionType.getFieldCount()
> 0) {
+ // If manifest-sort.enabled is enabled and there are sortable
fields, use
+ // trySortRewrite. Data evolution tables sort by RowID when all
manifest files contain
+ // RowID ranges, so they do not require partition fields.
+ if (options.manifestSortEnabled()
+ && (partitionType.getFieldCount() > 0
+ || (options.dataEvolutionEnabled() &&
allContainsRowId(input)))) {
return ManifestFileSorter.trySortCompaction(
input, newFilesForAbort, manifestFile, partitionType,
options);
} else {
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 bbdf5e14bc..33720c4aeb 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
@@ -56,7 +56,10 @@ import java.util.function.Function;
import static java.util.Collections.singletonList;
import static
org.apache.paimon.utils.ManifestReadThreadPool.sequentialBatchedExecute;
-/** Manifest file sorter that sorts and rewrites manifest files by a
configured partition field. */
+/**
+ * Manifest file sorter that sorts and rewrites manifest files by a configured
partition field, or
+ * by RowID for data evolution tables.
+ */
public class ManifestFileSorter {
private static final Logger LOG =
LoggerFactory.getLogger(ManifestFileSorter.class);
@@ -64,7 +67,7 @@ public class ManifestFileSorter {
/** Context object that carries shared state across compaction methods. */
static class CompactionContext {
final boolean fullCompaction;
- final RecordComparator fieldComparator;
+ final ManifestSortKey sortKey;
final Set<FileEntry.Identifier> deleteEntries;
/**
* Manifest files that need unsorted compaction.
@@ -81,13 +84,13 @@ public class ManifestFileSorter {
CompactionContext(
boolean fullCompaction,
- RecordComparator fieldComparator,
+ ManifestSortKey sortKey,
Set<FileEntry.Identifier> deleteEntries,
Map<ManifestFileMeta, Boolean> compactWithoutSort,
List<ManifestAdjacentSortedRun> levelRuns,
List<ManifestAdjacentSortedRun> pickedRuns) {
this.fullCompaction = fullCompaction;
- this.fieldComparator = fieldComparator;
+ this.sortKey = sortKey;
this.deleteEntries = deleteEntries;
this.compactWithoutSort = compactWithoutSort;
this.levelRuns = levelRuns;
@@ -154,6 +157,7 @@ public class ManifestFileSorter {
manifestFile,
partitionType,
sortPartitionField,
+ options.dataEvolutionEnabled(),
suggestedMetaSize,
suggestedMinMetaCount,
fullCompactionThreshold,
@@ -170,6 +174,7 @@ public class ManifestFileSorter {
manifestFile,
partitionType,
sortPartitionField,
+ options.dataEvolutionEnabled(),
suggestedMetaSize,
suggestedMinMetaCount,
maxRewriteSize,
@@ -190,6 +195,7 @@ public class ManifestFileSorter {
ManifestFile manifestFile,
RowType partitionType,
String sortPartitionField,
+ boolean dataEvolutionEnabled,
long suggestedMetaSize,
int suggestedMinMetaCount,
long fullCompactionThreshold,
@@ -216,6 +222,7 @@ public class ManifestFileSorter {
manifestFile,
partitionType,
sortPartitionField,
+ dataEvolutionEnabled,
suggestedMetaSize,
maxSizeAmplificationPercent,
sortedRunSizeRatio,
@@ -292,6 +299,7 @@ public class ManifestFileSorter {
ManifestFile manifestFile,
RowType partitionType,
String sortPartitionField,
+ boolean dataEvolutionEnabled,
long suggestedMetaSize,
int suggestedMinMetaCount,
long maxRewriteSize,
@@ -307,6 +315,7 @@ public class ManifestFileSorter {
manifestFile,
partitionType,
sortPartitionField,
+ dataEvolutionEnabled,
suggestedMetaSize,
maxSizeAmplificationPercent,
sortedRunSizeRatio,
@@ -414,21 +423,15 @@ public class ManifestFileSorter {
ManifestFile manifestFile,
RowType partitionType,
String sortPartitionField,
+ boolean dataEvolutionEnabled,
long suggestedMetaSize,
int maxSizeAmplificationPercent,
int sortedRunSizeRatio,
@Nullable Integer manifestReadParallelism) {
- // Step 1: Resolve sort field and build comparator for partition
ordering.
- String sortField = resolveSortField(sortPartitionField, partitionType);
- if (sortField == null) {
- throw new IllegalArgumentException(
- "Cannot resolve sort field for manifest sort rewrite.");
- }
- int sortFieldIndex = partitionType.getFieldNames().indexOf(sortField);
- RecordComparator fieldComparator =
- CodeGenUtils.newRecordComparator(
- partitionType.getFieldTypes(), new int[]
{sortFieldIndex});
+ // Step 1: Resolve sort key. Data evolution tables prefer RowID ranges
when available.
+ ManifestSortKey sortKey =
+ createSortKey(dataEvolutionEnabled, input, sortPartitionField,
partitionType);
// Step 2: Classify manifests into LSM files and collect delete
entries.
ClassifyResult classifyResult =
@@ -443,9 +446,7 @@ public class ManifestFileSorter {
// Step 3: Build level-sorted runs from LSM files based on partition
order.
List<ManifestAdjacentSortedRun> levelRuns =
- lsmFiles.isEmpty()
- ? new ArrayList<>()
- : buildLevelSortedRuns(lsmFiles, fieldComparator);
+ lsmFiles.isEmpty() ? new ArrayList<>() :
buildLevelSortedRuns(lsmFiles, sortKey);
// Step 4: Pick runs for compaction using size amplification and ratio
strategy.
ManifestPickStrategy pickStrategy =
@@ -454,7 +455,7 @@ public class ManifestFileSorter {
return new CompactionContext(
fullCompaction,
- fieldComparator,
+ sortKey,
classifyResult.deleteEntries,
classifyResult.compactWithoutSort,
levelRuns,
@@ -529,18 +530,15 @@ public class ManifestFileSorter {
* largest to level 1~4, rest to level 0).
*/
static List<ManifestAdjacentSortedRun> buildLevelSortedRuns(
- List<ManifestFileMeta> input, RecordComparator fieldComparator) {
+ List<ManifestFileMeta> input, ManifestSortKey sortKey) {
// Step 1: Sort by min value (if equal, then by max value)
input.sort(
(a, b) -> {
- int cmp =
- fieldComparator.compare(
- a.partitionStats().minValues(),
b.partitionStats().minValues());
+ int cmp = sortKey.compareMin(a, b);
if (cmp != 0) {
return cmp;
}
- return fieldComparator.compare(
- a.partitionStats().maxValues(),
b.partitionStats().maxValues());
+ return sortKey.compareMax(a, b);
});
// Step 2: Interval graph coloring algorithm - assign files to runs
@@ -550,9 +548,7 @@ public class ManifestFileSorter {
(r1, r2) -> {
ManifestFileMeta last1 = r1.get(r1.size() - 1);
ManifestFileMeta last2 = r2.get(r2.size() - 1);
- return fieldComparator.compare(
- last1.partitionStats().maxValues(),
- last2.partitionStats().maxValues());
+ return sortKey.compareMax(last1, last2);
});
for (ManifestFileMeta file : input) {
@@ -562,16 +558,11 @@ public class ManifestFileSorter {
List<ManifestFileMeta> newRun = new ArrayList<>();
newRun.add(file);
runs.offer(newRun);
- } else if (fieldComparator.compare(
- file.partitionStats().minValues(),
- earliestRun.get(earliestRun.size() -
1).partitionStats().maxValues())
- >= 0) {
- // Current file's min >= run's max, append to this run
+ } else if (sortKey.isAfterMax(file,
earliestRun.get(earliestRun.size() - 1))) {
+ // Current file's min is after the run's max, append to this
run
// Note: When min == max (boundary equality), files are
considered
- // non-overlapping and can be placed in the same SortedRun.
This allows
- // building fewer SortedRuns, improving compaction efficiency
while
- // maintaining correct sort order. However, these files may
later be separated
- // into different Sections during splitIntoSections to avoid
merge-sort overhead.
+ // non-overlapping for partition sort and can be placed in the
same SortedRun.
+ // RowID sort uses inclusive ranges, so boundary equality is
treated as overlap.
//
// See ManifestAdjacentSortedRun class comment for the full
boundary equality
// semantics.
@@ -612,17 +603,14 @@ public class ManifestFileSorter {
*/
static List<Section> splitIntoSections(
List<ManifestFileMeta> pickedFiles, CompactionContext ctx) {
- RecordComparator fieldComparator = ctx.fieldComparator;
+ ManifestSortKey sortKey = ctx.sortKey;
pickedFiles.sort(
(a, b) -> {
- int cmp =
- fieldComparator.compare(
- a.partitionStats().minValues(),
b.partitionStats().minValues());
+ int cmp = sortKey.compareMin(a, b);
if (cmp != 0) {
return cmp;
}
- return fieldComparator.compare(
- a.partitionStats().maxValues(),
b.partitionStats().maxValues());
+ return sortKey.compareMax(a, b);
});
List<Section> sections = new ArrayList<>();
@@ -633,25 +621,13 @@ public class ManifestFileSorter {
currentSectionFiles.add(first);
currentSectionTotalSize += first.fileSize();
boolean currentSectionHasUnsortedCompactMeta =
ctx.isMarkedForUnsortedCompaction(first);
- BinaryRow sectionMaxBound = first.partitionStats().maxValues();
+ ManifestFileMeta sectionMaxFile = first;
for (int i = 1; i < pickedFiles.size(); i++) {
ManifestFileMeta file = pickedFiles.get(i);
- // Note: Boundary equality (file.min == sectionMaxBound) results
in separate
- // sections. This design choice balances three factors:
- // 1. Avoid merge-sort overhead: Files with non-overlapping
boundaries can be processed
- // independently without merge-sort, improving performance.
- // 2. Maintain partition filtering capability: Each section has a
distinct key range,
- // enabling efficient partition pruning during queries.
- // 3. Preserve ordering invariant: Separating boundary-touching
files into different
- // sections
- // does not break the global sort order, as they are still
processed in ascending
- // order.
- //
- // IMPORTANT: While boundary-touching files are separated into
different Sections here,
- // they may be placed in the same SortedRun during
buildLevelSortedRuns (which uses >= 0
- // comparison). This dual behavior is intentional and documented
in class comments.
- if (fieldComparator.compare(file.partitionStats().minValues(),
sectionMaxBound) >= 0) {
+ // The sort key decides boundary handling. Partition sorting keeps
the historical
+ // boundary-equality behavior, while RowID sorting treats ranges
as inclusive.
+ if (sortKey.isAfterMax(file, sectionMaxFile)) {
sections.add(
new Section(
currentSectionFiles,
@@ -663,7 +639,7 @@ public class ManifestFileSorter {
currentSectionFiles.add(file);
currentSectionTotalSize += file.fileSize();
currentSectionHasUnsortedCompactMeta =
ctx.isMarkedForUnsortedCompaction(file);
- sectionMaxBound = file.partitionStats().maxValues();
+ sectionMaxFile = file;
} else {
currentSectionFiles.add(file);
currentSectionTotalSize += file.fileSize();
@@ -671,9 +647,8 @@ public class ManifestFileSorter {
&& ctx.isMarkedForUnsortedCompaction(file)) {
currentSectionHasUnsortedCompactMeta = true;
}
- if (fieldComparator.compare(file.partitionStats().maxValues(),
sectionMaxBound)
- > 0) {
- sectionMaxBound = file.partitionStats().maxValues();
+ if (sortKey.compareMax(file, sectionMaxFile) > 0) {
+ sectionMaxFile = file;
}
}
}
@@ -1010,8 +985,7 @@ public class ManifestFileSorter {
}
if (!entries.isEmpty()) {
- List<ManifestFileMeta> sorted =
- sortAndWriteEntries(entries, ctx.fieldComparator,
manifestFile);
+ List<ManifestFileMeta> sorted = sortAndWriteEntries(entries,
ctx.sortKey, manifestFile);
output.addSortedFiles(sorted);
sortNewFiles.addAll(sorted);
}
@@ -1066,14 +1040,14 @@ public class ManifestFileSorter {
if (!addEntries.isEmpty()) {
List<ManifestFileMeta> sorted =
- sortAndWriteEntries(addEntries, ctx.fieldComparator,
manifestFile);
+ sortAndWriteEntries(addEntries, ctx.sortKey, manifestFile);
output.addSortedFiles(sorted);
sortNewFiles.addAll(sorted);
}
if (!minorDeleteEntries.isEmpty()) {
List<ManifestFileMeta> sorted =
- sortAndWriteEntries(minorDeleteEntries,
ctx.fieldComparator, manifestFile);
+ sortAndWriteEntries(minorDeleteEntries, ctx.sortKey,
manifestFile);
output.addDeleteFiles(sorted);
sortNewFiles.addAll(sorted);
}
@@ -1081,11 +1055,9 @@ public class ManifestFileSorter {
/** Sort entries and write them to a new manifest file with proper error
handling. */
private static List<ManifestFileMeta> sortAndWriteEntries(
- List<ManifestEntry> entries,
- RecordComparator fieldComparator,
- ManifestFile manifestFile)
+ List<ManifestEntry> entries, ManifestSortKey sortKey, ManifestFile
manifestFile)
throws Exception {
- entries.sort((a, b) -> compareSortKey(a, b, fieldComparator));
+ entries.sort((a, b) -> compareSortKey(a, b, sortKey));
RollingFileWriter<ManifestEntry, ManifestFileMeta> writer =
manifestFile.createRollingWriter();
Exception exception = null;
@@ -1104,12 +1076,12 @@ public class ManifestFileSorter {
}
/**
- * Compare two {@link ManifestEntry}s by the composite key {@code
(sort-field, kind, fileName)}.
+ * Compare two {@link ManifestEntry}s by the composite key {@code
(sort-key, kind, fileName)}.
* {@code fileName} is used as the tie-breaker so that all entries sharing
the same sort-field
* value AND the same data file are emitted contiguously.
*/
- static int compareSortKey(ManifestEntry a, ManifestEntry b,
RecordComparator fieldComparator) {
- int c = fieldComparator.compare(a.partition(), b.partition());
+ private static int compareSortKey(ManifestEntry a, ManifestEntry b,
ManifestSortKey sortKey) {
+ int c = sortKey.compareEntry(a, b);
if (c != 0) {
return c;
}
@@ -1121,6 +1093,185 @@ public class ManifestFileSorter {
return a.file().fileName().compareTo(b.file().fileName());
}
+ private static ManifestSortKey createSortKey(
+ boolean dataEvolutionEnabled,
+ List<ManifestFileMeta> input,
+ 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.
+ RecordComparator partitionComparator =
+ partitionType.getFieldCount() == 0
+ ? null
+ :
CodeGenUtils.newRecordComparator(partitionType.getFieldTypes());
+ return new RowIdSortKey(partitionComparator);
+ }
+
+ if (partitionType.getFieldCount() == 0) {
+ throw new IllegalArgumentException(
+ "Cannot resolve sort key for manifest sort rewrite.");
+ }
+
+ String sortField = resolveSortField(sortPartitionField, partitionType);
+ int sortFieldIndex = partitionType.getFieldNames().indexOf(sortField);
+ if (sortFieldIndex < 0) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Cannot resolve sort field '%s' for manifest sort
rewrite.",
+ sortField));
+ }
+
+ RecordComparator fieldComparator =
+ CodeGenUtils.newRecordComparator(
+ partitionType.getFieldTypes(), new int[]
{sortFieldIndex});
+ return new PartitionSortKey(fieldComparator);
+ }
+
+ interface ManifestSortKey {
+
+ int compareMin(ManifestFileMeta a, ManifestFileMeta b);
+
+ int compareMax(ManifestFileMeta a, ManifestFileMeta b);
+
+ boolean isAfterMax(ManifestFileMeta file, ManifestFileMeta maxFile);
+
+ int compareEntry(ManifestEntry a, ManifestEntry b);
+ }
+
+ private static class PartitionSortKey implements ManifestSortKey {
+
+ private final RecordComparator fieldComparator;
+
+ private PartitionSortKey(RecordComparator fieldComparator) {
+ this.fieldComparator = fieldComparator;
+ }
+
+ @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 int compareEntry(ManifestEntry a, ManifestEntry b) {
+ return fieldComparator.compare(a.partition(), b.partition());
+ }
+ }
+
+ private static class RowIdSortKey implements ManifestSortKey {
+
+ @Nullable private final RecordComparator partitionComparator;
+
+ private RowIdSortKey(@Nullable RecordComparator partitionComparator) {
+ this.partitionComparator = partitionComparator;
+ }
+
+ @Override
+ public int compareMin(ManifestFileMeta a, ManifestFileMeta b) {
+ int c = comparePartitionMin(a, b);
+ if (c != 0) {
+ return c;
+ }
+ return Long.compare(nonNullMinRowId(a), nonNullMinRowId(b));
+ }
+
+ @Override
+ public int compareMax(ManifestFileMeta a, ManifestFileMeta b) {
+ int c = comparePartitionMax(a, b);
+ if (c != 0) {
+ return c;
+ }
+ return Long.compare(nonNullMaxRowId(a), nonNullMaxRowId(b));
+ }
+
+ @Override
+ public boolean isAfterMax(ManifestFileMeta file, ManifestFileMeta
maxFile) {
+ if (partitionComparator != null) {
+ int c =
+ partitionComparator.compare(
+ file.partitionStats().minValues(),
+ maxFile.partitionStats().maxValues());
+ if (c != 0) {
+ return c > 0;
+ }
+ }
+ return Long.compare(nonNullMinRowId(file),
nonNullMaxRowId(maxFile)) > 0;
+ }
+
+ @Override
+ public int compareEntry(ManifestEntry a, ManifestEntry b) {
+ int c = 0;
+ if (partitionComparator != null) {
+ c = partitionComparator.compare(a.partition(), b.partition());
+ if (c != 0) {
+ return c;
+ }
+ }
+ c = Long.compare(a.file().nonNullFirstRowId(),
b.file().nonNullFirstRowId());
+ if (c != 0) {
+ return c;
+ }
+ c = Long.compare(rowIdRangeEnd(a), rowIdRangeEnd(b));
+ if (c != 0) {
+ return c;
+ }
+ return Long.compare(b.file().maxSequenceNumber(),
a.file().maxSequenceNumber());
+ }
+
+ private int comparePartitionMin(ManifestFileMeta a, ManifestFileMeta
b) {
+ if (partitionComparator == null) {
+ return 0;
+ }
+ return partitionComparator.compare(
+ a.partitionStats().minValues(),
b.partitionStats().minValues());
+ }
+
+ private int comparePartitionMax(ManifestFileMeta a, ManifestFileMeta
b) {
+ if (partitionComparator == null) {
+ return 0;
+ }
+ return partitionComparator.compare(
+ a.partitionStats().maxValues(),
b.partitionStats().maxValues());
+ }
+
+ private static long nonNullMinRowId(ManifestFileMeta meta) {
+ Long minRowId = meta.minRowId();
+ if (minRowId == null) {
+ throw new IllegalArgumentException(
+ String.format("Manifest file '%s' has no min RowID.",
meta.fileName()));
+ }
+ return minRowId;
+ }
+
+ private static long nonNullMaxRowId(ManifestFileMeta meta) {
+ Long maxRowId = meta.maxRowId();
+ if (maxRowId == null) {
+ throw new IllegalArgumentException(
+ String.format("Manifest file '%s' has no max RowID.",
meta.fileName()));
+ }
+ return maxRowId;
+ }
+
+ private static long rowIdRangeEnd(ManifestEntry entry) {
+ return entry.file().nonNullFirstRowId() + entry.file().rowCount()
- 1;
+ }
+ }
+
/**
* Resolve the partition field to sort manifests by.
*
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 fedfecaafb..fb4f586268 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
@@ -1164,10 +1164,12 @@ public class SchemaValidation {
private static void validateManifestSort(TableSchema schema, CoreOptions
options) {
if (options.manifestSortEnabled()) {
- checkArgument(
- !schema.partitionKeys().isEmpty(),
- "Cannot enable '%s' for non-partition table.",
- CoreOptions.MANIFEST_SORT_ENABLED.key());
+ if (!options.dataEvolutionEnabled()) {
+ checkArgument(
+ !schema.partitionKeys().isEmpty(),
+ "Cannot enable '%s' for non-partition table.",
+ CoreOptions.MANIFEST_SORT_ENABLED.key());
+ }
String sortPartitionField = options.manifestSortPartitionField();
if (sortPartitionField != null && !sortPartitionField.isEmpty()) {
checkArgument(
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 2b91824d9a..0357a26e52 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
@@ -1064,6 +1064,128 @@ public class ManifestFileMetaTest extends
ManifestFileMetaTestBase {
}
}
+ @Test
+ public void testDataEvolutionManifestSortByPartitionAndRowId() {
+ List<ManifestFileMeta> input = new ArrayList<>();
+
+ input.add(
+ makeManifest(
+ makeRowIdEntry(true, "A-row30", 0, 30, 5),
+ makeRowIdEntry(true, "A-row10-seq3", 3, 10, 5, 3)));
+ input.add(
+ makeManifest(
+ makeRowIdEntry(true, "B-row20", 1, 20, 5),
+ makeRowIdEntry(true, "B-row0", 2, 0, 5)));
+ input.add(makeManifest(makeRowIdEntry(true, "C-row10-seq5", 3, 10, 5,
5)));
+
+ Options testOptions = new Options();
+ testOptions.set("manifest-sort.enabled", "true");
+ testOptions.set("data-evolution.enabled", "true");
+
+ List<ManifestFileMeta> merged =
+ ManifestFileMerger.merge(
+ input,
+ manifestFile,
+ getPartitionType(),
+ CoreOptions.fromMap(testOptions.toMap()));
+
+ assertEquivalentEntries(input, merged);
+
+ List<ManifestEntry> outputEntries = new ArrayList<>();
+ for (ManifestFileMeta meta : merged) {
+ outputEntries.addAll(manifestFile.read(meta.fileName(),
meta.fileSize()));
+ }
+
+ assertThat(
+ outputEntries.stream()
+ .map(entry -> entry.file().fileName())
+ .collect(Collectors.toList()))
+ .containsExactly("A-row30", "B-row20", "B-row0",
"C-row10-seq5", "A-row10-seq3");
+
+ for (int i = 1; i < outputEntries.size(); i++) {
+ int previousPartition = outputEntries.get(i -
1).partition().getInt(0);
+ int currentPartition = outputEntries.get(i).partition().getInt(0);
+ long previousRowId = outputEntries.get(i -
1).file().nonNullFirstRowId();
+ long currentRowId =
outputEntries.get(i).file().nonNullFirstRowId();
+ assertThat(currentPartition)
+ .as("Data evolution manifest entries should be sorted by
partition first")
+ .isGreaterThanOrEqualTo(previousPartition);
+ if (currentPartition == previousPartition) {
+ assertThat(currentRowId)
+ .as("Data evolution manifest entries should be sorted
by RowID")
+ .isGreaterThanOrEqualTo(previousRowId);
+ }
+ }
+ }
+
+ @Test
+ public void
testDataEvolutionManifestSortFallsBackToPartitionWhenRowIdStatsMissing() {
+ List<ManifestFileMeta> input = new ArrayList<>();
+
+ input.add(makeManifest(makeRowIdEntry(true, "rowid-p2-row0", 2, 0,
5)));
+ input.add(makeManifest(makeEntry(true, "legacy-p0", 0)));
+ input.add(makeManifest(makeRowIdEntry(true, "rowid-p1-row10", 1, 10,
5)));
+
+ Options testOptions = new Options();
+ testOptions.set("manifest-sort.enabled", "true");
+ testOptions.set("data-evolution.enabled", "true");
+
+ List<ManifestFileMeta> merged =
+ ManifestFileMerger.merge(
+ input,
+ manifestFile,
+ getPartitionType(),
+ CoreOptions.fromMap(testOptions.toMap()));
+
+ assertEquivalentEntries(input, merged);
+
+ assertThat(
+ readEntries(merged).stream()
+ .map(entry -> entry.file().fileName())
+ .collect(Collectors.toList()))
+ .containsExactly("legacy-p0", "rowid-p1-row10",
"rowid-p2-row0");
+ }
+
+ @Test
+ public void
testDataEvolutionMinorManifestSortPreservesUnmatchedDeleteEntries() {
+ List<ManifestFileMeta> input = new ArrayList<>();
+
+ input.add(
+ makeManifest(
+ makeRowIdEntry(true, "base-row0", 0, 0, 5, 1),
+ makeRowIdEntry(true, "survivor-row30", 0, 30, 5, 1)));
+ input.add(
+ makeManifest(
+ makeRowIdEntry(false, "base-row0", 0, 0, 5, 1),
+ makeRowIdEntry(false, "old-row10", 0, 10, 5, 1),
+ makeRowIdEntry(true, "new-row20", 0, 20, 5, 2)));
+
+ Options testOptions = new Options();
+ testOptions.set("manifest-sort.enabled", "true");
+ testOptions.set("data-evolution.enabled", "true");
+ testOptions.set("manifest.full-compaction-threshold-size",
Long.MAX_VALUE + "B");
+
+ List<ManifestFileMeta> merged =
+ ManifestFileMerger.merge(
+ input,
+ manifestFile,
+ getPartitionType(),
+ CoreOptions.fromMap(testOptions.toMap()));
+
+ List<ManifestEntry> outputEntries = readEntries(merged);
+ assertThat(
+ outputEntries.stream()
+ .map(entry -> entry.kind() + "-" +
entry.file().fileName())
+ .collect(Collectors.toList()))
+ .containsExactly("ADD-new-row20", "ADD-survivor-row30",
"DELETE-old-row10");
+
+ assertThat(
+ FileEntry.mergeEntries(outputEntries).stream()
+ .map(entry -> entry.kind() + "-" +
entry.file().fileName())
+ .collect(Collectors.toList()))
+ .containsExactly("ADD-new-row20", "ADD-survivor-row30",
"DELETE-old-row10");
+ }
+
/**
* Test manifest sort with a multi-field partition type.
*
@@ -1498,4 +1620,58 @@ public class ManifestFileMetaTest extends
ManifestFileMetaTestBase {
null,
null));
}
+
+ /** 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) {
+ return makeRowIdEntry(isAdd, fileName, partition, firstRowId,
rowCount, 0);
+ }
+
+ private ManifestEntry makeRowIdEntry(
+ boolean isAdd,
+ String fileName,
+ int partition,
+ long firstRowId,
+ long rowCount,
+ long sequenceNumber) {
+ BinaryRow binaryRow = new BinaryRow(1);
+ BinaryRowWriter writer = new BinaryRowWriter(binaryRow);
+ writer.writeInt(0, partition);
+ writer.complete();
+
+ return ManifestEntry.create(
+ isAdd ? FileKind.ADD : FileKind.DELETE,
+ binaryRow,
+ 0,
+ 0,
+ DataFileMeta.create(
+ fileName,
+ 0,
+ rowCount,
+ binaryRow,
+ binaryRow,
+ StatsTestUtils.newEmptySimpleStats(),
+ StatsTestUtils.newEmptySimpleStats(),
+ sequenceNumber,
+ sequenceNumber,
+ 0,
+ 0,
+ Collections.emptyList(),
+ Timestamp.fromEpochMillis(200000),
+ 0L,
+ null,
+ FileSource.APPEND,
+ null,
+ null,
+ firstRowId,
+ Collections.singletonList("f0")));
+ }
+
+ private List<ManifestEntry> readEntries(List<ManifestFileMeta>
manifestMetas) {
+ List<ManifestEntry> entries = new ArrayList<>();
+ for (ManifestFileMeta meta : manifestMetas) {
+ entries.addAll(manifestFile.read(meta.fileName(),
meta.fileSize()));
+ }
+ return entries;
+ }
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/manifest/NoPartitionManifestFileMetaTest.java
b/paimon-core/src/test/java/org/apache/paimon/manifest/NoPartitionManifestFileMetaTest.java
index 66465f1e75..52ac56608b 100644
---
a/paimon-core/src/test/java/org/apache/paimon/manifest/NoPartitionManifestFileMetaTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/manifest/NoPartitionManifestFileMetaTest.java
@@ -19,8 +19,12 @@
package org.apache.paimon.manifest;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.Timestamp;
+import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.operation.ManifestFileMerger;
import org.apache.paimon.options.Options;
+import org.apache.paimon.stats.StatsTestUtils;
import org.apache.paimon.types.RowType;
import org.junit.jupiter.api.BeforeEach;
@@ -28,6 +32,7 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;
@@ -117,6 +122,38 @@ public class NoPartitionManifestFileMetaTest extends
ManifestFileMetaTestBase {
.collect(Collectors.toList()));
}
+ @Test
+ public void testDataEvolutionManifestSortByRowId() {
+ List<ManifestFileMeta> input = new ArrayList<>();
+ input.add(makeManifest(makeRowIdEntry("row20", 20, 5, 0),
makeRowIdEntry("row0", 0, 5, 0)));
+ input.add(
+ makeManifest(
+ makeRowIdEntry("row10-seq1", 10, 5, 1),
+ makeRowIdEntry("row10-seq3", 10, 5, 3)));
+
+ Options testOptions = new Options();
+ testOptions.set("manifest-sort.enabled", "true");
+ testOptions.set("row-tracking.enabled", "true");
+ testOptions.set("data-evolution.enabled", "true");
+
+ List<ManifestFileMeta> merged =
+ ManifestFileMerger.merge(
+ input,
+ manifestFile,
+ getPartitionType(),
+ CoreOptions.fromMap(testOptions.toMap()));
+
+ assertEquivalentEntries(input, merged);
+
+ List<String> outputFileNames = new ArrayList<>();
+ for (ManifestFileMeta meta : merged) {
+ for (ManifestEntry entry : manifestFile.read(meta.fileName(),
meta.fileSize())) {
+ outputFileNames.add(entry.file().fileName());
+ }
+ }
+ assertThat(outputFileNames).containsExactly("row0", "row10-seq3",
"row10-seq1", "row20");
+ }
+
@Override
public ManifestFile getManifestFile() {
return manifestFile;
@@ -126,4 +163,34 @@ public class NoPartitionManifestFileMetaTest extends
ManifestFileMetaTestBase {
public RowType getPartitionType() {
return noPartitionType;
}
+
+ private ManifestEntry makeRowIdEntry(
+ String fileName, long firstRowId, long rowCount, long
sequenceNumber) {
+ return ManifestEntry.create(
+ FileKind.ADD,
+ BinaryRow.EMPTY_ROW,
+ 0,
+ 0,
+ DataFileMeta.create(
+ fileName,
+ 0,
+ rowCount,
+ BinaryRow.EMPTY_ROW,
+ BinaryRow.EMPTY_ROW,
+ StatsTestUtils.newEmptySimpleStats(),
+ StatsTestUtils.newEmptySimpleStats(),
+ sequenceNumber,
+ sequenceNumber,
+ 0,
+ 0,
+ Collections.emptyList(),
+ Timestamp.fromEpochMillis(200000),
+ 0L,
+ null,
+ FileSource.APPEND,
+ null,
+ null,
+ firstRowId,
+ Collections.singletonList("f0")));
+ }
}
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 ee3e25846a..ce714e2707 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
@@ -820,6 +820,65 @@ class SchemaValidationTest {
emptyList(),
options3,
"")));
+
+ // Test 4: data evolution tables can sort manifests by RowID without
partition keys
+ Map<String, String> options4 = new HashMap<>();
+ options4.put(CoreOptions.MANIFEST_SORT_ENABLED.key(), "true");
+ options4.put(CoreOptions.ROW_TRACKING_ENABLED.key(), "true");
+ options4.put(DATA_EVOLUTION_ENABLED.key(), "true");
+ options4.put(BUCKET.key(), String.valueOf(-1));
+ assertThatNoException()
+ .isThrownBy(
+ () ->
+ validateTableSchema(
+ new TableSchema(
+ 1,
+ fields,
+ 10,
+ emptyList(),
+ emptyList(),
+ options4,
+ "")));
+
+ // Test 5: data evolution tables should still validate configured
partition field
+ Map<String, String> options5 = new HashMap<>();
+ options5.put(CoreOptions.MANIFEST_SORT_ENABLED.key(), "true");
+ options5.put(CoreOptions.MANIFEST_SORT_PARTITION_FIELD.key(), "f1");
+ options5.put(CoreOptions.ROW_TRACKING_ENABLED.key(), "true");
+ options5.put(DATA_EVOLUTION_ENABLED.key(), "true");
+ options5.put(BUCKET.key(), String.valueOf(-1));
+ assertThatThrownBy(
+ () ->
+ validateTableSchema(
+ new TableSchema(
+ 1,
+ fields,
+ 10,
+ singletonList("f0"),
+ emptyList(),
+ options5,
+ "")))
+ .hasMessageContaining("is not a partition field");
+
+ // Test 6: data evolution non-partition tables cannot configure a
partition field
+ Map<String, String> options6 = new HashMap<>();
+ options6.put(CoreOptions.MANIFEST_SORT_ENABLED.key(), "true");
+ options6.put(CoreOptions.MANIFEST_SORT_PARTITION_FIELD.key(), "f0");
+ options6.put(CoreOptions.ROW_TRACKING_ENABLED.key(), "true");
+ options6.put(DATA_EVOLUTION_ENABLED.key(), "true");
+ options6.put(BUCKET.key(), String.valueOf(-1));
+ assertThatThrownBy(
+ () ->
+ validateTableSchema(
+ new TableSchema(
+ 1,
+ fields,
+ 10,
+ emptyList(),
+ emptyList(),
+ options6,
+ "")))
+ .hasMessageContaining("is not a partition field");
}
@Test
