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 11db5d1de7 [core] Support manifest sorting for non-partitioned
bucketed tables (#9808)
11db5d1de7 is described below
commit 11db5d1de7b9e7183dea1456dad0e978678d7337
Author: Jingsong Lee <[email protected]>
AuthorDate: Mon Sep 14 16:44:25 2026 +0800
[core] Support manifest sorting for non-partitioned bucketed tables (#9808)
---
docs/generated/core_configuration.html | 4 +-
.../main/java/org/apache/paimon/CoreOptions.java | 11 +-
.../operation/ManifestAdjacentSortedRun.java | 8 +-
.../paimon/operation/ManifestCompactDryRun.java | 4 +-
.../paimon/operation/ManifestFileMerger.java | 8 +-
.../paimon/operation/ManifestFileSorter.java | 95 ++++++++-------
.../org/apache/paimon/schema/SchemaValidation.java | 6 +-
.../manifest/NoPartitionManifestFileMetaTest.java | 128 +++++++++++++++++++++
.../apache/paimon/schema/SchemaValidationTest.java | 29 ++++-
9 files changed, 236 insertions(+), 57 deletions(-)
diff --git a/docs/generated/core_configuration.html
b/docs/generated/core_configuration.html
index ac57300175..16d3339f41 100644
--- a/docs/generated/core_configuration.html
+++ b/docs/generated/core_configuration.html
@@ -1063,7 +1063,7 @@ Mainly to resolve data skew on primary keys. We recommend
starting with 64 mb wh
<td><h5>manifest-sort.enabled</h5></td>
<td style="word-wrap: break-word;">false</td>
<td>Boolean</td>
- <td>Whether to invoke manifest sort rewrite during commit.<br
/>Note: enabling this changes the semantics of 'manifest.merge-min-count'. In
the sort rewrite path, small manifest files within the rewrite budget are
sorted and merged directly, so the minimum-count gate no longer prevents
merging a small number of under-budget manifest files when full compaction is
not triggered.</td>
+ <td>Whether to invoke manifest sort rewrite during commit.
Non-partitioned tables can sort by bucket with fixed or postponed buckets, or
by RowID for data evolution tables when all input manifests contain RowID
ranges.<br />Note: enabling this changes the semantics of
'manifest.merge-min-count'. In the sort rewrite path, small manifest files
within the rewrite budget are sorted and merged directly, so the minimum-count
gate no longer prevents merging a small number of under-b [...]
</tr>
<tr>
<td><h5>manifest-sort.max-rewrite-size</h5></td>
@@ -1075,7 +1075,7 @@ Mainly to resolve data skew on primary keys. We recommend
starting with 64 mb wh
<td><h5>manifest-sort.partition-field</h5></td>
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
- <td>Partition field name to sort manifest entries by. Validated by
schema validation, if not configured, defaults to the first partition
field.</td>
+ <td>Partition field name to sort manifest entries by. Validated by
schema validation; must be unset for non-partitioned tables. If not configured,
defaults to the first partition field, or all partition fields for data
evolution RowID sorting.</td>
</tr>
<tr>
<td><h5>manifest.compression</h5></td>
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 7b0665c502..68215e18e4 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -568,7 +568,12 @@ public class CoreOptions implements Serializable {
.defaultValue(false)
.withDescription(
Description.builder()
- .text("Whether to invoke manifest sort
rewrite during commit.")
+ .text(
+ "Whether to invoke manifest sort
rewrite during commit."
+ + " Non-partitioned tables
can sort by bucket"
+ + " with fixed or
postponed buckets, or by RowID"
+ + " for data evolution
tables when all input"
+ + " manifests contain
RowID ranges.")
.linebreak()
.text(
"Note: enabling this changes the
semantics of '"
@@ -587,7 +592,9 @@ public class CoreOptions implements Serializable {
.noDefaultValue()
.withDescription(
"Partition field name to sort manifest entries by.
Validated by"
- + " schema validation, if not configured,
defaults to the first partition field.");
+ + " schema validation; must be unset for
non-partitioned tables."
+ + " If not configured, defaults to the
first partition field,"
+ + " or all partition fields for data
evolution RowID sorting.");
public static final ConfigOption<MemorySize>
MANIFEST_SORT_MAX_REWRITE_SIZE =
key("manifest-sort.max-rewrite-size")
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 b076387ab2..ceafa1ad5b 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
@@ -29,10 +29,10 @@ import java.util.stream.Collectors;
* 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> 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.
+ * <p><b>Boundary Equality:</b> Bucket and partition-field sorting treat
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/ManifestCompactDryRun.java
b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestCompactDryRun.java
index f80554ce6e..1b04912ad8 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
@@ -84,9 +84,7 @@ public class ManifestCompactDryRun {
}
RowType partitionType = table.schema().logicalPartitionType();
- if (partitionType.getFieldCount() == 0
- && !(options.dataEvolutionEnabled()
- && ManifestFileMeta.allContainsRowId(manifests))) {
+ if (!ManifestFileMerger.canUseManifestSort(manifests, partitionType,
options)) {
return summary + " Manifest sort level files: unavailable (no
sortable field).";
}
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 e3f8c7af76..688d59f492 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
@@ -23,6 +23,7 @@ import org.apache.paimon.disk.IOManager;
import org.apache.paimon.manifest.ManifestEntry;
import org.apache.paimon.manifest.ManifestFile;
import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.table.BucketMode;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.ExceptionUtils;
@@ -61,9 +62,8 @@ public class ManifestFileMerger {
List<ManifestFileMeta> newFilesForAbort = new ArrayList<>();
try {
- // 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.
+ // Bucketed tables and data evolution tables with complete RowID
ranges do not require
+ // partition fields for manifest sort rewrite.
if (canUseManifestSort(input, partitionType, options)) {
return ManifestFileSorter.trySortCompaction(
input, newFilesForAbort, manifestFile, partitionType,
options, ioManager);
@@ -94,6 +94,8 @@ public class ManifestFileMerger {
List<ManifestFileMeta> input, RowType partitionType, CoreOptions
options) {
return options.manifestSortEnabled()
&& (partitionType.getFieldCount() > 0
+ || options.bucket() > 0
+ || options.bucket() == BucketMode.POSTPONE_BUCKET
|| (options.dataEvolutionEnabled() &&
allContainsRowId(input)));
}
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 87fd4611e7..94255953fc 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
@@ -62,7 +62,7 @@ import java.util.function.Function;
import static
org.apache.paimon.utils.ManifestReadThreadPool.sequentialBatchedExecute;
/**
- * Manifest file sorter that sorts and rewrites manifest files by a configured
partition field, or
+ * Manifest file sorter that sorts and rewrites manifest files by bucket
and/or partition fields, or
* by RowID for data evolution tables.
*/
public class ManifestFileSorter {
@@ -697,7 +697,7 @@ public class ManifestFileSorter {
} 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 for partition sort and can be placed in the
same SortedRun.
+ // non-overlapping for bucket/partition sort and can share the
same SortedRun.
// RowID sort uses inclusive ranges, so boundary equality is
treated as overlap.
//
// See ManifestAdjacentSortedRun class comment for the full
boundary equality
@@ -1207,23 +1207,22 @@ public class ManifestFileSorter {
return new RowIdSortKey(partitionComparator, partitionType,
partitionSortFields);
}
- 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));
+ int sortFieldIndex = -1;
+ RecordComparator fieldComparator = null;
+ if (partitionType.getFieldCount() > 0) {
+ String sortField = resolveSortField(sortPartitionField,
partitionType);
+ sortFieldIndex = partitionType.getFieldNames().indexOf(sortField);
+ if (sortFieldIndex < 0) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Cannot resolve sort field '%s' for manifest
sort rewrite.",
+ sortField));
+ }
+ fieldComparator =
+ CodeGenUtils.newRecordComparator(
+ partitionType.getFieldTypes(), new int[]
{sortFieldIndex});
}
- RecordComparator fieldComparator =
- CodeGenUtils.newRecordComparator(
- partitionType.getFieldTypes(), new int[]
{sortFieldIndex});
if (bucketed) {
boolean compareManifestBuckets =
input.stream()
@@ -1231,6 +1230,10 @@ public class ManifestFileSorter {
return new BucketSortKey(
fieldComparator, partitionType, sortFieldIndex,
compareManifestBuckets);
}
+ if (fieldComparator == null) {
+ throw new IllegalArgumentException(
+ "Cannot resolve sort key for manifest sort rewrite.");
+ }
return new PartitionSortKey(fieldComparator, partitionType,
sortFieldIndex);
}
@@ -1355,31 +1358,38 @@ public class ManifestFileSorter {
private static class BucketSortKey implements ManifestSortKey {
- private final PartitionSortKey partitionSortKey;
- private final InternalRow.FieldGetter sortFieldGetter;
+ @Nullable private final PartitionSortKey partitionSortKey;
+ @Nullable 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,
+ @Nullable 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);
+ List<DataType> fieldTypes = new ArrayList<>();
+ fieldTypes.add(DataTypes.INT());
+ if (fieldComparator == null) {
+ this.partitionSortKey = null;
+ this.sortFieldGetter = null;
+ } else {
+ this.partitionSortKey =
+ new PartitionSortKey(fieldComparator, partitionType,
sortFieldIndex);
+ DataType sortFieldType =
partitionType.getTypeAt(sortFieldIndex);
+ this.sortFieldGetter =
InternalRow.createFieldGetter(sortFieldType, sortFieldIndex);
+ fieldTypes.add(sortFieldType);
+ }
+ Collections.addAll(
+ fieldTypes,
+ DataTypes.TINYINT(),
+ DataTypes.STRING(),
+ ManifestEntry.MANIFEST_ROW_TYPE);
+ this.sortFieldNum = fieldTypes.size() - 1;
+ this.externalSortRowType = DataTypes.ROW(fieldTypes.toArray(new
DataType[0]));
this.externalSortKeyFields = createSequentialFields(sortFieldNum);
}
@@ -1391,7 +1401,7 @@ public class ManifestFileSorter {
return bucketComparison;
}
}
- return partitionSortKey.compareMin(a, b);
+ return partitionSortKey == null ? 0 :
partitionSortKey.compareMin(a, b);
}
@Override
@@ -1402,7 +1412,7 @@ public class ManifestFileSorter {
return bucketComparison;
}
}
- return partitionSortKey.compareMax(a, b);
+ return partitionSortKey == null ? 0 :
partitionSortKey.compareMax(a, b);
}
@Override
@@ -1413,7 +1423,11 @@ public class ManifestFileSorter {
return bucketComparison > 0;
}
}
- return partitionSortKey.isAfterMax(file, maxFile);
+ // Without partition fields, equal bucket boundaries can share a
run. Missing bucket
+ // statistics must conservatively be treated as overlapping ranges.
+ return partitionSortKey == null
+ ? compareManifestBuckets
+ : partitionSortKey.isAfterMax(file, maxFile);
}
@Override
@@ -1430,14 +1444,17 @@ public class ManifestFileSorter {
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());
+ int pos = 1;
+ if (sortFieldGetter != null) {
+ row.setField(pos++,
sortFieldGetter.getFieldOrNull(entry.partition()));
+ }
+ row.setField(pos++, entry.kind().toByteValue());
row.setField(
- 3,
+ pos++,
entry instanceof ProjectedManifestEntry
? ((ProjectedManifestEntry)
entry).file().fileNameBinary()
:
BinaryString.fromString(entry.file().fileName()));
- row.setField(4, binaryManifestRow);
+ row.setField(pos, 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 996fceaa97..3a31f7dbba 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
@@ -2023,8 +2023,10 @@ public class SchemaValidation {
if (options.manifestSortEnabled()) {
if (!options.dataEvolutionEnabled()) {
checkArgument(
- !schema.partitionKeys().isEmpty(),
- "Cannot enable '%s' for non-partition table.",
+ !schema.partitionKeys().isEmpty()
+ || options.bucket() > 0
+ || options.bucket() ==
BucketMode.POSTPONE_BUCKET,
+ "Cannot enable '%s' for non-partition table without
fixed or postponed buckets or data evolution.",
CoreOptions.MANIFEST_SORT_ENABLED.key());
}
String sortPartitionField = options.manifestSortPartitionField();
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 3170c08d02..af6aca173f 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,25 +19,34 @@
package org.apache.paimon.manifest;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.Snapshot;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.Timestamp;
import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.operation.ManifestCompactDryRun;
import org.apache.paimon.operation.ManifestFileMerger;
import org.apache.paimon.options.Options;
import org.apache.paimon.stats.StatsTestUtils;
+import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.types.RowType;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
/** Test {@link ManifestFile}. for table without partition */
public class NoPartitionManifestFileMetaTest extends ManifestFileMetaTestBase {
@@ -154,6 +163,125 @@ public class NoPartitionManifestFileMetaTest extends
ManifestFileMetaTestBase {
assertThat(outputFileNames).containsExactly("row0", "row10-seq3",
"row10-seq1", "row20");
}
+ @ParameterizedTest
+ @CsvSource({
+ "4, false, false", "4, true, false", "-2, false, false", "-2, true,
false",
+ "4, false, true", "4, true, true", "-2, false, true", "-2, true, true"
+ })
+ public void testManifestSortByBucket(
+ int bucket, boolean fullCompaction, boolean missingBucketStats) {
+ int firstBucket = bucket == -2 ? -2 : 0;
+ List<ManifestFileMeta> input =
+ Arrays.asList(
+ makeManifest(
+ makeBucketEntry(true, "a-high", 3),
+ makeBucketEntry(true, "same-name", 1),
+ makeBucketEntry(true, "z-low", firstBucket)),
+ makeManifest(
+ makeBucketEntry(true, "same-name", 2),
+ makeBucketEntry(false, "same-name", 1),
+ makeBucketEntry(true, "b-high", 3)));
+ if (missingBucketStats) {
+ input.set(0, withoutBucketStats(input.get(0)));
+ }
+
+ Options options = new Options();
+ options.set(CoreOptions.MANIFEST_SORT_ENABLED, true);
+ options.set(CoreOptions.BUCKET, bucket);
+ options.set(CoreOptions.MANIFEST_TARGET_FILE_SIZE.key(), "1G");
+ options.set(CoreOptions.MANIFEST_MERGE_MIN_COUNT, 100);
+ options.set(
+ CoreOptions.MANIFEST_FULL_COMPACTION_FILE_SIZE.key(),
+ fullCompaction ? "1B" : Long.MAX_VALUE + "B");
+
+ List<ManifestFileMeta> merged =
+ ManifestFileMerger.merge(
+ input, manifestFile, getPartitionType(), new
CoreOptions(options));
+ List<ManifestEntry> entries =
+ merged.stream()
+ .flatMap(
+ meta ->
+ manifestFile.read(meta.fileName(),
meta.fileSize())
+ .stream())
+ .collect(Collectors.toList());
+
+ assertThat(entries)
+ .containsExactly(
+ makeBucketEntry(true, "z-low", firstBucket),
+ makeBucketEntry(true, "same-name", 2),
+ makeBucketEntry(true, "a-high", 3),
+ makeBucketEntry(true, "b-high", 3));
+ assertThat(merged).hasSize(1);
+ assertThat(merged.get(0).minBucket()).isEqualTo(firstBucket);
+ assertThat(merged.get(0).maxBucket()).isEqualTo(3);
+ }
+
+ @ParameterizedTest
+ @CsvSource({
+ "4, 0, false", "4, 1, false", "4, 2, false",
+ "-2, 0, false", "-2, 1, false", "-2, 2, false",
+ "4, 0, true", "-2, 0, true"
+ })
+ public void testManifestSortDryRunByBucket(
+ int bucket, int lowMaxBucket, boolean missingBucketStats) {
+ int firstBucket = bucket == -2 ? -2 : 0;
+ List<ManifestFileMeta> input =
+ Arrays.asList(
+ makeManifest(
+ makeBucketEntry(true, "high-start", 1),
+ makeBucketEntry(true, "high-end", 3)),
+ makeManifest(
+ makeBucketEntry(true, "low-start",
firstBucket),
+ makeBucketEntry(true, "low-end",
lowMaxBucket)));
+ if (missingBucketStats) {
+ input.set(0, withoutBucketStats(input.get(0)));
+ }
+
+ Options options = new Options();
+ options.set(CoreOptions.MANIFEST_SORT_ENABLED, true);
+ options.set(CoreOptions.BUCKET, bucket);
+ options.set(CoreOptions.MANIFEST_TARGET_FILE_SIZE.key(), "1B");
+ options.set(CoreOptions.MANIFEST_FULL_COMPACTION_FILE_SIZE.key(),
Long.MAX_VALUE + "B");
+
+ FileStoreTable table = mock(FileStoreTable.class, RETURNS_DEEP_STUBS);
+ Snapshot snapshot = mock(Snapshot.class);
+ when(table.options()).thenReturn(options.toMap());
+
when(table.store().snapshotManager().latestSnapshot()).thenReturn(snapshot);
+
when(table.store().manifestListFactory().create().readDataManifests(snapshot))
+ .thenReturn(input);
+
when(table.store().manifestFileFactory().create()).thenReturn(manifestFile);
+
when(table.schema().logicalPartitionType()).thenReturn(getPartitionType());
+
+ assertThat(ManifestCompactDryRun.execute(table))
+ .endsWith(
+ missingBucketStats || lowMaxBucket > 1
+ ? "Manifest sort level files: L0=0, L1=0,
L2=0, L3=1, L4=1."
+ : "Manifest sort level files: L0=0, L1=0,
L2=0, L3=0, L4=2.");
+ }
+
+ private ManifestEntry makeBucketEntry(boolean isAdd, String fileName, int
bucket) {
+ ManifestEntry entry = makeEntry(isAdd, fileName, null);
+ return ManifestEntry.create(entry.kind(), entry.partition(), bucket,
4, entry.file());
+ }
+
+ private ManifestFileMeta withoutBucketStats(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
public 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 c9588f4552..238dbef999 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
@@ -27,6 +27,8 @@ import org.apache.paimon.types.DataTypes;
import org.assertj.core.api.ThrowableAssert;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
@@ -1928,6 +1930,29 @@ class SchemaValidationTest {
1, fields, 10, partitionKeys, primaryKeys, options,
"geospatial test");
}
+ @ParameterizedTest
+ @ValueSource(ints = {1, 4, -2})
+ void testManifestSortForNonPartitionBucketedTable(int bucket) {
+ Map<String, String> options = new HashMap<>();
+ options.put(CoreOptions.MANIFEST_SORT_ENABLED.key(), "true");
+ options.put(BUCKET.key(), String.valueOf(bucket));
+ TableSchema schema =
+ new TableSchema(
+ 1,
+ singletonList(new DataField(0, "f0", DataTypes.INT())),
+ 10,
+ emptyList(),
+ singletonList("f0"),
+ options,
+ "");
+ assertThatNoException().isThrownBy(() -> validateTableSchema(schema));
+
+ options.put(CoreOptions.MANIFEST_SORT_PARTITION_FIELD.key(), "f0");
+ assertThatThrownBy(() -> validateTableSchema(schema.copy(options)))
+ .hasMessageContaining(
+ "'manifest-sort.partition-field' = 'f0' is not a
partition field");
+ }
+
@Test
void testManifestSortValidation() {
List<DataField> fields =
@@ -1935,7 +1960,7 @@ class SchemaValidationTest {
new DataField(0, "f0", DataTypes.INT()),
new DataField(1, "f1", DataTypes.INT()));
- // Test 1: manifest-sort.enabled on non-partition table should fail
+ // Test 1: non-partition tables without bucket or RowID sorting should
fail
Map<String, String> options1 = new HashMap<>();
options1.put(CoreOptions.MANIFEST_SORT_ENABLED.key(), "true");
options1.put(BUCKET.key(), String.valueOf(-1));
@@ -1951,7 +1976,7 @@ class SchemaValidationTest {
options1,
"")))
.hasMessageContaining(
- "Cannot enable 'manifest-sort.enabled' for
non-partition table.");
+ "Cannot enable 'manifest-sort.enabled' for
non-partition table without fixed or postponed buckets or data evolution.");
// Test 2: manifest-sort-partition-field not in partition keys should
fail
Map<String, String> options2 = new HashMap<>();
