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 4ab06f52bfdc..6f1e5f8e4076 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 @@ private static List buildLevelSortedRunsForDryRun( 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 76c5b0ef5ca0..51f51dabfe30 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 @@ static List trySortCompaction( @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 @@ static List trySortCompaction( manifestFile, partitionType, sortPartitionField, + bucketed, options.dataEvolutionEnabled(), runMergeOptimizeEnabled, suggestedMetaSize, @@ -192,6 +194,7 @@ static List trySortCompaction( manifestFile, partitionType, sortPartitionField, + bucketed, options.dataEvolutionEnabled(), runMergeOptimizeEnabled, suggestedMetaSize, @@ -215,6 +218,7 @@ private static Optional> tryFullCompaction( ManifestFile manifestFile, RowType partitionType, String sortPartitionField, + boolean bucketed, boolean dataEvolutionEnabled, boolean runMergeOptimizeEnabled, long suggestedMetaSize, @@ -238,6 +242,7 @@ private static Optional> tryFullCompaction( manifestFile, partitionType, sortPartitionField, + bucketed, dataEvolutionEnabled, runMergeOptimizeEnabled, suggestedMetaSize, @@ -321,6 +326,7 @@ private static List tryMinorCompaction( ManifestFile manifestFile, RowType partitionType, String sortPartitionField, + boolean bucketed, boolean dataEvolutionEnabled, boolean runMergeOptimizeEnabled, long suggestedMetaSize, @@ -339,6 +345,7 @@ private static List tryMinorCompaction( manifestFile, partitionType, sortPartitionField, + bucketed, dataEvolutionEnabled, runMergeOptimizeEnabled, suggestedMetaSize, @@ -453,6 +460,7 @@ private static CompactionContext prepareCompaction( ManifestFile manifestFile, RowType partitionType, String sortPartitionField, + boolean bucketed, boolean dataEvolutionEnabled, boolean runMergeOptimizeEnabled, long suggestedMetaSize, @@ -464,7 +472,9 @@ private static CompactionContext prepareCompaction( 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 @@ static ManifestSortKey createSortKey( List 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 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 @@ private static ManifestSortKey createSortKey( 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 InternalRow binaryManifestRow(BinaryRow row) { } } + 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 c27d8c012fb3..898f238231ed 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 void testManifestSortWithOverlappingPartitions() { } } + @Test + public void testManifestSortPreservesExistingOrderForUnawareBucketTable() { + List 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 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 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 merged = + ManifestFileMerger.merge( + input, + manifestFile, + getPartitionType(), + CoreOptions.fromMap(testOptions.toMap())); + + assertEquivalentEntries(input, merged); + List 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 input = new ArrayList<>(); @@ -1392,26 +1446,31 @@ public void testManifestSortMaxRewriteSizeSmallerThanTargetFileSizeStillRewrites .isTrue(); } - @Test - public void testManifestSortWithSpillableExternalSortBuffer() { + @ParameterizedTest + @ValueSource(booleans = {false, true}) + public void testManifestSortWithSpillableExternalSortBuffer(boolean bucketed) { List input = new ArrayList<>(); for (int manifest = 0; manifest < 4; manifest++) { List 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 void testManifestSortWithSpillableExternalSortBuffer() { CoreOptions.fromMap(testOptions.toMap())); assertEquivalentEntries(input, merged); - for (ManifestFileMeta meta : merged) { - List 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 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 void testBoundaryEqualityHandling() { } } + /** 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 2d08a158d5ca..1142f2fa1664 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.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 @@ void testLargeFragmentedManifestsUseRunMerge() throws Exception { .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 @@ private ManifestEntry rowIdEntry(String fileName, long firstRowId) { 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;