This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 81e9d87a376d9191df32356d79624ea898fc07ec Author: JingsongLi <[email protected]> AuthorDate: Mon Sep 14 10:15:19 2026 +0800 [core] Fix bucket-first manifest sorting for postpone tables --- .../paimon/operation/ManifestCompactDryRun.java | 3 +- .../paimon/operation/ManifestEntryRunMerge.java | 6 +- .../paimon/operation/ManifestFileSorter.java | 25 ++------ .../paimon/manifest/ManifestFileMetaTest.java | 73 ++++++++++++++++++---- .../operation/ManifestEntryRunMergeTest.java | 7 ++- .../operation/ManifestRewriteCleanupTest.java | 2 +- 6 files changed, 76 insertions(+), 40 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 6f1e5f8e40..f80554ce6e 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 @@ -24,6 +24,7 @@ import org.apache.paimon.manifest.ManifestFile; import org.apache.paimon.manifest.ManifestFileMeta; import org.apache.paimon.manifest.ManifestList; import org.apache.paimon.options.MemorySize; +import org.apache.paimon.table.BucketMode; import org.apache.paimon.table.FileStoreTable; import org.apache.paimon.types.RowType; @@ -120,7 +121,7 @@ public class ManifestCompactDryRun { manifests, options.manifestSortPartitionField(), partitionType, - options.bucket() > 0); + options.bucket() > 0 || options.bucket() == BucketMode.POSTPONE_BUCKET); ManifestFileSorter.ClassifyResult classifyResult = ManifestFileSorter.classifyManifests( manifests, diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryRunMerge.java b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryRunMerge.java index 740b23f58d..aeb8e6ef62 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryRunMerge.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryRunMerge.java @@ -71,7 +71,7 @@ final class ManifestEntryRunMerge { @Nullable static List<ManifestFileMeta> sortAndWriteFullEntries( List<ManifestFileMeta> section, - ManifestFileSorter.RowIdEntrySortKey sortKey, + ManifestFileSorter.RowIdSortKey sortKey, RowType partitionType, ManifestFile manifestFile, List<ManifestFileMeta> newFilesForAbort, @@ -102,7 +102,7 @@ final class ManifestEntryRunMerge { @Nullable static Pair<List<ManifestFileMeta>, List<ManifestFileMeta>> sortAndWriteMinorEntries( List<ManifestFileMeta> section, - ManifestFileSorter.RowIdEntrySortKey sortKey, + ManifestFileSorter.RowIdSortKey sortKey, RowType partitionType, ManifestFile manifestFile, List<ManifestFileMeta> newFilesForAbort, @@ -133,7 +133,7 @@ final class ManifestEntryRunMerge { @Nullable private static ManifestEntryRunMergePlan discoverRuns( List<ManifestFileMeta> section, - ManifestFileSorter.RowIdEntrySortKey sortKey, + ManifestFileSorter.RowIdSortKey sortKey, RowType partitionType, ManifestFile manifestFile, CollectedDeletes deletes, 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 51f51dabfe..87fd4611e7 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 @@ -32,6 +32,7 @@ import org.apache.paimon.manifest.ManifestFile; import org.apache.paimon.manifest.ManifestFileMeta; import org.apache.paimon.manifest.ProjectedManifestEntry; import org.apache.paimon.partition.PartitionPredicate; +import org.apache.paimon.table.BucketMode; import org.apache.paimon.types.DataType; import org.apache.paimon.types.DataTypes; import org.apache.paimon.types.RowType; @@ -155,7 +156,7 @@ public class ManifestFileSorter { @Nullable IOManager ioManager) throws Exception { String sortPartitionField = options.manifestSortPartitionField(); - boolean bucketed = options.bucket() > 0; + boolean bucketed = options.bucket() > 0 || options.bucket() == BucketMode.POSTPONE_BUCKET; boolean runMergeOptimizeEnabled = options.manifestMergeOptimizeEnabled(); long suggestedMetaSize = options.manifestTargetSize().getBytes(); int suggestedMinMetaCount = options.manifestMergeMinCount(); @@ -1110,7 +1111,7 @@ public class ManifestFileSorter { sorted = ManifestEntryRunMerge.sortAndWriteFullEntries( section, - (RowIdEntrySortKey) ctx.sortKey, + (RowIdSortKey) ctx.sortKey, ctx.partitionType, manifestFile, sortNewFiles, @@ -1152,7 +1153,7 @@ public class ManifestFileSorter { sorted = ManifestEntryRunMerge.sortAndWriteMinorEntries( section, - (RowIdEntrySortKey) ctx.sortKey, + (RowIdSortKey) ctx.sortKey, ctx.partitionType, manifestFile, sortNewFiles, @@ -1188,14 +1189,6 @@ public class ManifestFileSorter { return true; } - static ManifestSortKey createSortKey( - boolean dataEvolutionEnabled, - List<ManifestFileMeta> input, - String sortPartitionField, - RowType partitionType) { - return createSortKey(dataEvolutionEnabled, input, sortPartitionField, partitionType, false); - } - static ManifestSortKey createSortKey( boolean dataEvolutionEnabled, List<ManifestFileMeta> input, @@ -1289,11 +1282,6 @@ public class ManifestFileSorter { InternalRow binaryManifestRow(BinaryRow row); } - interface RowIdEntrySortKey extends ManifestSortKey { - - int comparePartitions(BinaryRow left, BinaryRow right); - } - private static class PartitionSortKey implements ManifestSortKey { private final RecordComparator fieldComparator; @@ -1458,7 +1446,7 @@ public class ManifestFileSorter { } } - private static class RowIdSortKey implements RowIdEntrySortKey { + static class RowIdSortKey implements ManifestSortKey { @Nullable private final RecordComparator partitionComparator; private final InternalRow.FieldGetter[] partitionFieldGetters; @@ -1545,8 +1533,7 @@ public class ManifestFileSorter { return row.getRow(sortFieldNum, ManifestEntry.MANIFEST_ROW_TYPE.getFieldCount()); } - @Override - public int comparePartitions(BinaryRow left, BinaryRow right) { + int comparePartitions(BinaryRow left, BinaryRow right) { return partitionComparator == null ? 0 : partitionComparator.compare(left, right); } 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 898f238231..6d67eddfbd 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 @@ -19,6 +19,7 @@ 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.BinaryRowWriter; import org.apache.paimon.data.Timestamp; @@ -30,11 +31,13 @@ import org.apache.paimon.fs.SeekableInputStream; import org.apache.paimon.fs.SeekableInputStreamWrapper; import org.apache.paimon.fs.local.LocalFileIO; 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.partition.PartitionPredicate; import org.apache.paimon.schema.FileSystemSchemaManager; import org.apache.paimon.stats.StatsTestUtils; +import org.apache.paimon.table.FileStoreTable; import org.apache.paimon.types.IntType; import org.apache.paimon.types.RowType; import org.apache.paimon.utils.FailingFileIO; @@ -48,6 +51,7 @@ import org.junit.jupiter.api.RepeatedTest; 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 org.junit.jupiter.params.provider.NullSource; import org.junit.jupiter.params.provider.ValueSource; @@ -80,6 +84,9 @@ import java.util.stream.LongStream; import static org.assertj.core.api.Assertions.assertThat; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assumptions.assumeTrue; +import static org.mockito.Mockito.RETURNS_DEEP_STUBS; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; /** Tests for {@link ManifestFileMeta}. */ public class ManifestFileMetaTest extends ManifestFileMetaTestBase { @@ -1310,19 +1317,22 @@ public class ManifestFileMetaTest extends ManifestFileMetaTestBase { .containsExactly(1, 3, 0, 2); } - @Test - public void testManifestSortUsesBucketAsPrimaryKeyForBucketedTable() { + @ParameterizedTest + @ValueSource(ints = {4, -2}) + public void testManifestSortUsesBucketAsPrimaryKeyForBucketedTable(int bucket) { + int firstBucket = bucket == -2 ? -2 : 0; List<ManifestFileMeta> input = Arrays.asList( makeManifest( - makeBucketEntry("a-b1-p1", 1, 1), makeBucketEntry("a-b0-p0", 0, 0)), + makeBucketEntry("a-b1-p1", 1, 1), + makeBucketEntry("a-first-p0", 0, firstBucket)), makeManifest( makeBucketEntry("b-b1-p0", 0, 1), - makeBucketEntry("b-b0-p1", 1, 0))); + makeBucketEntry("b-first-p1", 1, firstBucket))); Options testOptions = new Options(); testOptions.set(CoreOptions.MANIFEST_SORT_ENABLED, true); - testOptions.set(CoreOptions.BUCKET, 4); + testOptions.set(CoreOptions.BUCKET, bucket); testOptions.set(CoreOptions.MANIFEST_TARGET_FILE_SIZE.key(), "1G"); testOptions.set(CoreOptions.MANIFEST_FULL_COMPACTION_FILE_SIZE.key(), "1B"); List<ManifestFileMeta> merged = @@ -1334,12 +1344,48 @@ public class ManifestFileMetaTest extends ManifestFileMetaTestBase { assertEquivalentEntries(input, merged); List<ManifestEntry> entries = readEntries(merged); - assertThat(entries).extracting(ManifestEntry::bucket).containsExactly(0, 0, 1, 1); + assertThat(entries) + .extracting(ManifestEntry::bucket) + .containsExactly(firstBucket, firstBucket, 1, 1); assertThat(entries) .extracting(entry -> entry.partition().getInt(0)) .containsExactly(0, 1, 0, 1); } + @ParameterizedTest + @ValueSource(ints = {-1, 4, -2}) + public void testManifestSortDryRunUsesBucketRangesForBucketedTable(int bucket) { + int firstBucket = bucket == -2 ? -2 : 0; + List<ManifestFileMeta> input = + Arrays.asList( + makeManifest( + makeBucketEntry("a-p0", 0, firstBucket), + makeBucketEntry("a-p1", 1, firstBucket)), + makeManifest(makeBucketEntry("b-p0", 0, 1), makeBucketEntry("b-p1", 1, 1))); + + Options testOptions = new Options(); + testOptions.set(CoreOptions.MANIFEST_SORT_ENABLED, true); + testOptions.set(CoreOptions.BUCKET, bucket); + testOptions.set(CoreOptions.MANIFEST_TARGET_FILE_SIZE.key(), "1B"); + testOptions.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(testOptions.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()); + + // Overlapping partition ranges form one run only when buckets take precedence. + assertThat(ManifestCompactDryRun.execute(table)) + .endsWith( + bucket == -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."); + } + @Test public void testManifestSortMinorCompactionRespectsMergeMinCount() { List<ManifestFileMeta> input = new ArrayList<>(); @@ -1447,30 +1493,31 @@ public class ManifestFileMetaTest extends ManifestFileMetaTestBase { } @ParameterizedTest - @ValueSource(booleans = {false, true}) - public void testManifestSortWithSpillableExternalSortBuffer(boolean bucketed) { + @CsvSource({"-1, false", "4, true", "-2, true"}) + public void testManifestSortWithSpillableExternalSortBuffer(int bucket, 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); + int entryBucket = Math.floorMod(manifest * 31 + i * 17, 4); + if (bucket == -2 && entryBucket == 3) { + entryBucket = -2; + } entries.add( makeBucketEntry( String.format( "spill-manifest-%02d-entry-%03d-payload-padding-%040d", manifest, i, i), partition, - bucket)); + entryBucket)); } 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(CoreOptions.BUCKET, bucket); testOptions.set("manifest.full-compaction-threshold-size", "1B"); testOptions.set("page-size", "1kb"); testOptions.set("sort-spill-buffer-size", "4kb"); 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 1142f2fa16..dbd8ef66a1 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 @@ -76,9 +76,10 @@ class ManifestEntryRunMergeTest extends ManifestFileMetaTestBase { List<ManifestFileMeta> manifests = new ArrayList<>(); manifests.add(makeManifest(firstManifest.toArray(new ManifestEntry[0]))); manifests.add(makeManifest(secondManifest.toArray(new ManifestEntry[0]))); - ManifestFileSorter.RowIdEntrySortKey sortKey = - (ManifestFileSorter.RowIdEntrySortKey) - ManifestFileSorter.createSortKey(true, manifests, null, partitionType); + ManifestFileSorter.RowIdSortKey sortKey = + (ManifestFileSorter.RowIdSortKey) + ManifestFileSorter.createSortKey( + true, manifests, null, partitionType, false); CollectedDeletes deletes = new CollectedDeletes(true); List<ManifestFileMeta> output; diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/ManifestRewriteCleanupTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/ManifestRewriteCleanupTest.java index 2049011bc8..8247d23262 100644 --- a/paimon-core/src/test/java/org/apache/paimon/operation/ManifestRewriteCleanupTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/operation/ManifestRewriteCleanupTest.java @@ -598,7 +598,7 @@ class ManifestRewriteCleanupTest extends ManifestFileMetaTestBase { private ManifestFileSorter.ManifestSortKey failingSortKey( List<ManifestFileMeta> input, int successfulRowsBeforeFailure, AssertionError failure) { return new FailingManifestSortKey( - ManifestFileSorter.createSortKey(true, input, null, PARTITION_TYPE), + ManifestFileSorter.createSortKey(true, input, null, PARTITION_TYPE, false), successfulRowsBeforeFailure, failure); }
