Level threshold of lookup to generate remote lookup files. Level files below this threshold will not generate remote lookup files.
+
+
manifest-sort.bucket-first
+
false
+
Boolean
+
Sort manifest entries by bucket before the configured partition field. This improves manifest pruning for bucket-key point lookups spanning many partitions, at the cost of wider partition ranges in each manifest.
+
manifest-sort.enabled
false
Boolean
Whether to invoke manifest sort rewrite during commit. 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.
+
+
manifest-sort.force-rewrite
+
false
+
Boolean
+
Force an explicit manifest compaction to rewrite already compacted manifest runs using the configured manifest sort order. This should be supplied as a one-shot dynamic option for maintenance, not persisted for routine writes.
+
manifest-sort.max-rewrite-size
256 mb
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 7b0665c50296..a806c041763e 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -589,6 +589,16 @@ public InlineElement getDescription() {
"Partition field name to sort manifest entries by. Validated by"
+ " schema validation, if not configured, defaults to the first partition field.");
+ public static final ConfigOption MANIFEST_SORT_BUCKET_FIRST =
+ key("manifest-sort.bucket-first")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Sort manifest entries by bucket before the configured partition"
+ + " field. This improves manifest pruning for bucket-key point"
+ + " lookups spanning many partitions, at the cost of wider"
+ + " partition ranges in each manifest.");
+
public static final ConfigOption MANIFEST_SORT_MAX_REWRITE_SIZE =
key("manifest-sort.max-rewrite-size")
.memoryType()
@@ -599,6 +609,16 @@ public InlineElement getDescription() {
+ " skipped. Set to a larger value to allow more aggressive"
+ " sort rewriting. The cap only limits the sorted rewrite portion and full/minor cleanup may still happen beyond it.");
+ public static final ConfigOption MANIFEST_SORT_FORCE_REWRITE =
+ key("manifest-sort.force-rewrite")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Force an explicit manifest compaction to rewrite already compacted"
+ + " manifest runs using the configured manifest sort order."
+ + " This should be supplied as a one-shot dynamic option for"
+ + " maintenance, not persisted for routine writes.");
+
public static final ConfigOption MANIFEST_MERGE_OPTIMIZE_ENABLED =
key("manifest.merge-optimize.enabled")
.booleanType()
@@ -3223,10 +3243,18 @@ public String manifestSortPartitionField() {
return options.get(MANIFEST_SORT_PARTITION_FIELD);
}
+ public boolean manifestSortBucketFirst() {
+ return options.get(MANIFEST_SORT_BUCKET_FIRST);
+ }
+
public long manifestSortMaxRewriteSize() {
return options.get(MANIFEST_SORT_MAX_REWRITE_SIZE).getBytes();
}
+ public boolean manifestSortForceRewrite() {
+ return options.get(MANIFEST_SORT_FORCE_REWRITE);
+ }
+
public boolean manifestMergeOptimizeEnabled() {
return options.get(MANIFEST_MERGE_OPTIMIZE_ENABLED);
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/BucketFilter.java b/paimon-core/src/main/java/org/apache/paimon/manifest/BucketFilter.java
index 4662d7dab517..09883f70ba40 100644
--- a/paimon-core/src/main/java/org/apache/paimon/manifest/BucketFilter.java
+++ b/paimon-core/src/main/java/org/apache/paimon/manifest/BucketFilter.java
@@ -27,6 +27,8 @@
/** Filter for bucket. */
public class BucketFilter {
+ private static final int MAX_ENUMERATED_BUCKETS = 10_000;
+
private final boolean onlyReadRealBuckets;
private final @Nullable Integer specifiedBucket;
private final @Nullable Filter bucketFilter;
@@ -77,4 +79,50 @@ public boolean test(BinaryRow partition, int bucket, int totalBucket) {
return totalAwareBucketFilter == null
|| totalAwareBucketFilter.test(partition, bucket, totalBucket);
}
+
+ /** Conservatively tests whether a manifest's bucket ranges can contain a matching entry. */
+ public boolean mayContain(ManifestFileMeta manifest) {
+ Integer minBucket = manifest.minBucket();
+ Integer maxBucket = manifest.maxBucket();
+ if (minBucket == null || maxBucket == null) {
+ return true;
+ }
+ if (onlyReadRealBuckets && maxBucket < 0) {
+ return false;
+ }
+ if (specifiedBucket != null
+ && (specifiedBucket < minBucket || specifiedBucket > maxBucket)) {
+ return false;
+ }
+ if (bucketFilter != null
+ && rangeIsReasonable(minBucket, maxBucket)
+ && !anyBucketMatches(minBucket, maxBucket)) {
+ return false;
+ }
+ if (totalAwareBucketFilter instanceof ManifestBucketFilter) {
+ Integer minTotalBuckets = manifest.minTotalBuckets();
+ Integer maxTotalBuckets = manifest.maxTotalBuckets();
+ if (minTotalBuckets != null && maxTotalBuckets != null) {
+ return ((ManifestBucketFilter) totalAwareBucketFilter)
+ .mayContain(minBucket, maxBucket, minTotalBuckets, maxTotalBuckets);
+ }
+ }
+ return true;
+ }
+
+ private boolean anyBucketMatches(int minBucket, int maxBucket) {
+ for (int bucket = minBucket; ; bucket++) {
+ if (bucketFilter.test(bucket)) {
+ return true;
+ }
+ if (bucket == maxBucket) {
+ break;
+ }
+ }
+ return false;
+ }
+
+ private static boolean rangeIsReasonable(int min, int max) {
+ return max >= min && (long) max - min < MAX_ENUMERATED_BUCKETS;
+ }
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/FileEntry.java b/paimon-core/src/main/java/org/apache/paimon/manifest/FileEntry.java
index 8cc109b6a882..63504c17a380 100644
--- a/paimon-core/src/main/java/org/apache/paimon/manifest/FileEntry.java
+++ b/paimon-core/src/main/java/org/apache/paimon/manifest/FileEntry.java
@@ -23,6 +23,7 @@
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.io.ProjectedDataFileMeta;
import org.apache.paimon.memory.MemorySegmentUtils;
+import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.utils.CloseableIterator;
import org.apache.paimon.utils.FileStorePathFactory;
import org.apache.paimon.utils.Filter;
@@ -361,6 +362,15 @@ static Set readDeletedEntries(
ManifestFile manifestFile,
List manifestFiles,
@Nullable Integer manifestReadParallelism) {
+ return readDeletedEntries(manifestFile, manifestFiles, manifestReadParallelism, null, null);
+ }
+
+ static Set readDeletedEntries(
+ ManifestFile manifestFile,
+ List manifestFiles,
+ @Nullable Integer manifestReadParallelism,
+ @Nullable PartitionPredicate partitionFilter,
+ @Nullable BucketFilter bucketFilter) {
manifestFiles =
manifestFiles.stream()
.filter(file -> file.numDeletedFiles() > 0)
@@ -372,7 +382,9 @@ static Set readDeletedEntries(
try (CloseableIterator entries =
manifestFile.scan(
manifest.fileName(),
- ProjectedManifestEntry.DELETE_ENTRY_PROJECTION)) {
+ ProjectedManifestEntry.DELETE_ENTRY_PROJECTION,
+ partitionFilter,
+ bucketFilter)) {
while (entries.hasNext()) {
ProjectedManifestEntry entry = entries.next();
if (entry.isDelete()) {
diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroWriter.java b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroWriter.java
index e78f273e29f1..980f9839f1e0 100644
--- a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroWriter.java
+++ b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestAvroWriter.java
@@ -296,6 +296,7 @@ public static final class EncodedEntry {
private byte kind;
private BinaryRow partition;
private int bucket;
+ private int totalBuckets;
private int level;
private long schemaId;
private boolean hasRowId;
@@ -306,6 +307,7 @@ public EncodedEntry replace(
byte kind,
BinaryRow partition,
int bucket,
+ int totalBuckets,
int level,
long schemaId,
long firstRowId,
@@ -313,6 +315,7 @@ public EncodedEntry replace(
this.kind = kind;
this.partition = partition;
this.bucket = bucket;
+ this.totalBuckets = totalBuckets;
this.level = level;
this.schemaId = schemaId;
this.hasRowId = true;
@@ -325,12 +328,14 @@ public EncodedEntry replace(
byte kind,
BinaryRow partition,
int bucket,
+ int totalBuckets,
int level,
long schemaId,
long rowCount) {
this.kind = kind;
this.partition = partition;
this.bucket = bucket;
+ this.totalBuckets = totalBuckets;
this.level = level;
this.schemaId = schemaId;
this.hasRowId = false;
@@ -348,6 +353,8 @@ public static final class EncodedBlockMeta {
private final long schemaId;
private final int minBucket;
private final int maxBucket;
+ private final int minTotalBuckets;
+ private final int maxTotalBuckets;
private final int minLevel;
private final int maxLevel;
private final long minRowId;
@@ -360,6 +367,8 @@ public EncodedBlockMeta(
long schemaId,
int minBucket,
int maxBucket,
+ int minTotalBuckets,
+ int maxTotalBuckets,
int minLevel,
int maxLevel,
long minRowId,
@@ -370,6 +379,8 @@ public EncodedBlockMeta(
this.schemaId = schemaId;
this.minBucket = minBucket;
this.maxBucket = maxBucket;
+ this.minTotalBuckets = minTotalBuckets;
+ this.maxTotalBuckets = maxTotalBuckets;
this.minLevel = minLevel;
this.maxLevel = maxLevel;
this.minRowId = minRowId;
@@ -396,9 +407,12 @@ private final class FileWriter {
private long schemaId = Long.MIN_VALUE;
private int minBucket = Integer.MAX_VALUE;
private int maxBucket = Integer.MIN_VALUE;
+ private int minTotalBuckets = Integer.MAX_VALUE;
+ private int maxTotalBuckets = Integer.MIN_VALUE;
private int minLevel = Integer.MAX_VALUE;
private int maxLevel = Integer.MIN_VALUE;
private boolean bucketStatsKnown = true;
+ private boolean totalBucketStatsKnown = true;
private boolean levelStatsKnown = true;
private @Nullable RowIdStats rowIdStats = new RowIdStats();
private boolean closed;
@@ -473,6 +487,8 @@ private void collectStats(ManifestEntry entry) {
schemaId = Math.max(schemaId, entry.file().schemaId());
minBucket = Math.min(minBucket, entry.bucket());
maxBucket = Math.max(maxBucket, entry.bucket());
+ minTotalBuckets = Math.min(minTotalBuckets, entry.totalBuckets());
+ maxTotalBuckets = Math.max(maxTotalBuckets, entry.totalBuckets());
minLevel = Math.min(minLevel, entry.level());
maxLevel = Math.max(maxLevel, entry.level());
if (rowIdStats != null) {
@@ -500,6 +516,8 @@ private void collectStats(EncodedEntry entry) {
schemaId = Math.max(schemaId, entry.schemaId);
minBucket = Math.min(minBucket, entry.bucket);
maxBucket = Math.max(maxBucket, entry.bucket);
+ minTotalBuckets = Math.min(minTotalBuckets, entry.totalBuckets);
+ maxTotalBuckets = Math.max(maxTotalBuckets, entry.totalBuckets);
minLevel = Math.min(minLevel, entry.level);
maxLevel = Math.max(maxLevel, entry.level);
if (rowIdStats != null) {
@@ -517,6 +535,8 @@ private void collectStats(EncodedBlockMeta metadata) {
schemaId = Math.max(schemaId, metadata.schemaId);
minBucket = Math.min(minBucket, metadata.minBucket);
maxBucket = Math.max(maxBucket, metadata.maxBucket);
+ minTotalBuckets = Math.min(minTotalBuckets, metadata.minTotalBuckets);
+ maxTotalBuckets = Math.max(maxTotalBuckets, metadata.maxTotalBuckets);
minLevel = Math.min(minLevel, metadata.minLevel);
maxLevel = Math.max(maxLevel, metadata.maxLevel);
if (rowIdStats != null) {
@@ -538,6 +558,12 @@ private void collectStats(ManifestFileMeta manifest) {
minBucket = Math.min(minBucket, manifest.minBucket());
maxBucket = Math.max(maxBucket, manifest.maxBucket());
}
+ if (manifest.minTotalBuckets() == null || manifest.maxTotalBuckets() == null) {
+ totalBucketStatsKnown = false;
+ } else {
+ minTotalBuckets = Math.min(minTotalBuckets, manifest.minTotalBuckets());
+ maxTotalBuckets = Math.max(maxTotalBuckets, manifest.maxTotalBuckets());
+ }
if (manifest.minLevel() == null || manifest.maxLevel() == null) {
levelStatsKnown = false;
} else {
@@ -709,7 +735,10 @@ private ManifestFileMeta result() {
levelStatsKnown ? minLevel : null,
levelStatsKnown ? maxLevel : null,
rowIdStats == null ? null : rowIdStats.minRowId,
- rowIdStats == null ? null : rowIdStats.maxRowId);
+ rowIdStats == null ? null : rowIdStats.maxRowId,
+ null,
+ totalBucketStatsKnown ? minTotalBuckets : null,
+ totalBucketStatsKnown ? maxTotalBuckets : null);
}
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestBucketFilter.java b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestBucketFilter.java
new file mode 100644
index 000000000000..70615de15a03
--- /dev/null
+++ b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestBucketFilter.java
@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.manifest;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.utils.TriFilter;
+
+/** A total-bucket-aware filter that can conservatively test manifest bucket ranges. */
+public interface ManifestBucketFilter extends TriFilter {
+
+ /** Returns false only when no entry in the supplied ranges can match. */
+ boolean mayContain(int minBucket, int maxBucket, int minTotalBuckets, int maxTotalBuckets);
+}
diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java
index 0dc99a047076..3cd9ae68f961 100644
--- a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java
+++ b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java
@@ -168,14 +168,26 @@ public List read(
* materialized with the complete manifest schema.
*/
public CloseableIterator scan(String fileName, Projection projection) {
+ return scan(fileName, projection, null, null);
+ }
+
+ /**
+ * Scans projected manifest entries and filters partitions and buckets before decoding nested
+ * data-file metadata.
+ */
+ public CloseableIterator scan(
+ String fileName,
+ Projection projection,
+ @Nullable PartitionPredicate partitionFilter,
+ @Nullable BucketFilter bucketFilter) {
try {
CloseableIterator rows =
createManifestIterator(
fileIO,
pathFactory.toPath(fileName),
projection.projectedType(),
- null,
- null);
+ partitionFilter,
+ bucketFilter);
return new CloseableIterator() {
@Override
diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFileMeta.java b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFileMeta.java
index 2123a419a5d9..f12aa0ac30a8 100644
--- a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFileMeta.java
+++ b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFileMeta.java
@@ -62,8 +62,9 @@ public class ManifestFileMeta {
new DataField(
12,
"_EXTRA_FILES",
- new ArrayType(
- true, new VarCharType(false, Integer.MAX_VALUE)))));
+ new ArrayType(true, new VarCharType(false, Integer.MAX_VALUE))),
+ new DataField(13, "_MIN_TOTAL_BUCKETS", new IntType(true)),
+ new DataField(14, "_MAX_TOTAL_BUCKETS", new IntType(true))));
private final String fileName;
private final long fileSize;
@@ -78,6 +79,8 @@ public class ManifestFileMeta {
private final @Nullable Long minRowId;
private final @Nullable Long maxRowId;
private final @Nullable List extraFiles;
+ private final @Nullable Integer minTotalBuckets;
+ private final @Nullable Integer maxTotalBuckets;
public ManifestFileMeta(
String fileName,
@@ -105,6 +108,8 @@ public ManifestFileMeta(
maxLevel,
minRowId,
maxRowId,
+ null,
+ null,
null);
}
@@ -122,6 +127,40 @@ public ManifestFileMeta(
@Nullable Long minRowId,
@Nullable Long maxRowId,
@Nullable List extraFiles) {
+ this(
+ fileName,
+ fileSize,
+ numAddedFiles,
+ numDeletedFiles,
+ partitionStats,
+ schemaId,
+ minBucket,
+ maxBucket,
+ minLevel,
+ maxLevel,
+ minRowId,
+ maxRowId,
+ extraFiles,
+ null,
+ null);
+ }
+
+ public ManifestFileMeta(
+ String fileName,
+ long fileSize,
+ long numAddedFiles,
+ long numDeletedFiles,
+ SimpleStats partitionStats,
+ long schemaId,
+ @Nullable Integer minBucket,
+ @Nullable Integer maxBucket,
+ @Nullable Integer minLevel,
+ @Nullable Integer maxLevel,
+ @Nullable Long minRowId,
+ @Nullable Long maxRowId,
+ @Nullable List extraFiles,
+ @Nullable Integer minTotalBuckets,
+ @Nullable Integer maxTotalBuckets) {
this.fileName = fileName;
this.fileSize = fileSize;
this.numAddedFiles = numAddedFiles;
@@ -135,6 +174,8 @@ public ManifestFileMeta(
this.minRowId = minRowId;
this.maxRowId = maxRowId;
this.extraFiles = extraFiles;
+ this.minTotalBuckets = minTotalBuckets;
+ this.maxTotalBuckets = maxTotalBuckets;
}
public String fileName() {
@@ -189,6 +230,14 @@ public long schemaId() {
return extraFiles;
}
+ public @Nullable Integer minTotalBuckets() {
+ return minTotalBuckets;
+ }
+
+ public @Nullable Integer maxTotalBuckets() {
+ return maxTotalBuckets;
+ }
+
@Override
public boolean equals(Object o) {
if (!(o instanceof ManifestFileMeta)) {
@@ -207,7 +256,9 @@ public boolean equals(Object o) {
&& Objects.equals(maxLevel, that.maxLevel)
&& Objects.equals(minRowId, that.minRowId)
&& Objects.equals(maxRowId, that.maxRowId)
- && Objects.equals(extraFiles, that.extraFiles);
+ && Objects.equals(extraFiles, that.extraFiles)
+ && Objects.equals(minTotalBuckets, that.minTotalBuckets)
+ && Objects.equals(maxTotalBuckets, that.maxTotalBuckets);
}
@Override
@@ -225,13 +276,15 @@ public int hashCode() {
maxLevel,
minRowId,
maxRowId,
- extraFiles);
+ extraFiles,
+ minTotalBuckets,
+ maxTotalBuckets);
}
@Override
public String toString() {
return String.format(
- "{%s, %d, %d, %d, %s, %d, %s, %s, %s, %s, %s, %s, %s}",
+ "{%s, %d, %d, %d, %s, %d, %s, %s, %s, %s, %s, %s, %s, %s, %s}",
fileName,
fileSize,
numAddedFiles,
@@ -244,7 +297,9 @@ public String toString() {
maxLevel,
minRowId,
maxRowId,
- extraFiles);
+ extraFiles,
+ minTotalBuckets,
+ maxTotalBuckets);
}
// ----------------------- Serialization -----------------------------
diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFileMetaSerializer.java b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFileMetaSerializer.java
index 4c2ab90c96ff..a22618fa809f 100644
--- a/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFileMetaSerializer.java
+++ b/paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFileMetaSerializer.java
@@ -60,7 +60,9 @@ public InternalRow toRow(ManifestFileMeta meta) {
meta.maxLevel(),
meta.minRowId(),
meta.maxRowId(),
- toStringArrayData(meta.extraFiles()));
+ toStringArrayData(meta.extraFiles()),
+ meta.minTotalBuckets(),
+ meta.maxTotalBuckets());
}
@Override
@@ -95,6 +97,8 @@ private ManifestFileMeta fromDataRow(InternalRow row) {
row.isNullAt(9) ? null : row.getInt(9),
row.isNullAt(10) ? null : row.getLong(10),
row.isNullAt(11) ? null : row.getLong(11),
- row.isNullAt(12) ? null : fromStringArrayData(row.getArray(12)));
+ row.isNullAt(12) ? null : fromStringArrayData(row.getArray(12)),
+ row.getFieldCount() <= 13 || row.isNullAt(13) ? null : row.getInt(13),
+ row.getFieldCount() <= 14 || row.isNullAt(14) ? null : row.getInt(14));
}
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/manifest/ProjectedManifestEntry.java b/paimon-core/src/main/java/org/apache/paimon/manifest/ProjectedManifestEntry.java
index 1d99c8bd918d..45a872f4e2cc 100644
--- a/paimon-core/src/main/java/org/apache/paimon/manifest/ProjectedManifestEntry.java
+++ b/paimon-core/src/main/java/org/apache/paimon/manifest/ProjectedManifestEntry.java
@@ -105,6 +105,7 @@ private static Projection createDeleteEntryProjection() {
manifestType.getField(ManifestEntry.KIND),
manifestType.getField(ManifestEntry.PARTITION),
manifestType.getField(ManifestEntry.BUCKET),
+ manifestType.getField(ManifestEntry.TOTAL_BUCKETS),
manifestType
.getField(ManifestEntry.FILE)
.newType(
@@ -142,6 +143,7 @@ private static Projection createEntryLayoutProjection() {
manifestType.getField(ManifestEntry.KIND),
manifestType.getField(ManifestEntry.PARTITION),
manifestType.getField(ManifestEntry.BUCKET),
+ manifestType.getField(ManifestEntry.TOTAL_BUCKETS),
manifestType
.getField(ManifestEntry.FILE)
.newType(
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java
index a9ef5902ec9e..dd152c4d0725 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java
@@ -157,6 +157,7 @@ public FileStoreScan withBucket(int bucket) {
@Override
public FileStoreScan withBucketFilter(Filter bucketFilter) {
+ manifestsReader.withBucketFilter(bucketFilter);
this.bucketFilter = bucketFilter;
return this;
}
@@ -164,6 +165,7 @@ public FileStoreScan withBucketFilter(Filter bucketFilter) {
@Override
public FileStoreScan withTotalAwareBucketFilter(
TriFilter totalAwareBucketFilter) {
+ manifestsReader.withTotalAwareBucketFilter(totalAwareBucketFilter);
this.totalAwareBucketFilter = totalAwareBucketFilter;
return this;
}
@@ -415,16 +417,27 @@ private Iterator readAndMergeFileEntries(
List manifests,
Function converter,
boolean useSequential) {
- Set deletedEntries =
- FileEntry.readDeletedEntries(
- manifest ->
- readManifest(
- manifest,
- SimpleFileEntry::from,
- FileEntry.deletedFilter(),
- null),
- manifests,
- parallelism);
+ Set deletedEntries;
+ if (manifestEntryFilter == null) {
+ deletedEntries =
+ FileEntry.readDeletedEntries(
+ manifestFileFactory.create(),
+ manifests,
+ parallelism,
+ manifestsReader.partitionFilter(),
+ createBucketFilter());
+ } else {
+ deletedEntries =
+ FileEntry.readDeletedEntries(
+ manifest ->
+ readManifest(
+ manifest,
+ SimpleFileEntry::from,
+ FileEntry.deletedFilter(),
+ null),
+ manifests,
+ parallelism);
+ }
manifests =
manifests.stream()
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/BucketSelector.java b/paimon-core/src/main/java/org/apache/paimon/operation/BucketSelector.java
index 8c28c3c0c641..5fab5b0f2655 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/BucketSelector.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/BucketSelector.java
@@ -23,6 +23,7 @@
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.data.serializer.InternalRowSerializer;
+import org.apache.paimon.manifest.ManifestBucketFilter;
import org.apache.paimon.predicate.Equal;
import org.apache.paimon.predicate.FieldRef;
import org.apache.paimon.predicate.In;
@@ -31,7 +32,6 @@
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.BiFilter;
-import org.apache.paimon.utils.TriFilter;
import org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableSet;
@@ -55,9 +55,11 @@
/** Selector to select bucket from {@link Predicate}. */
@ThreadSafe
-public class BucketSelector implements TriFilter {
+public class BucketSelector implements ManifestBucketFilter {
public static final int MAX_VALUES = 1000;
+ private static final int MAX_TOTAL_BUCKET_RANGE = 10_000;
+ private static final int MAX_MANIFEST_BUCKET_COMBINATIONS = 10_000;
private final BucketFunctionType bucketFunctionType;
private final RowType rowType;
@@ -65,6 +67,7 @@ public class BucketSelector implements TriFilter {
private final RowType bucketKeyType;
private final Predicate predicate;
private final Map> partitionSelectors;
+ private final Optional manifestSelector;
public BucketSelector(
Predicate predicate,
@@ -78,6 +81,7 @@ public BucketSelector(
this.partitionType = partitionType;
this.bucketKeyType = bucketKeyType;
this.partitionSelectors = new ConcurrentHashMap<>();
+ this.manifestSelector = createPartitionSelectorFromPredicate(predicate);
}
@Override
@@ -88,6 +92,24 @@ public boolean test(BinaryRow partition, Integer bucket, Integer numBucket) {
.orElse(true);
}
+ @Override
+ public boolean mayContain(
+ int minBucket, int maxBucket, int minTotalBuckets, int maxTotalBuckets) {
+ if (minBucket < 0
+ || maxBucket < minBucket
+ || minTotalBuckets <= 0
+ || maxTotalBuckets < minTotalBuckets
+ || (long) maxTotalBuckets - minTotalBuckets >= MAX_TOTAL_BUCKET_RANGE) {
+ return true;
+ }
+ return manifestSelector
+ .map(
+ selector ->
+ selector.mayContain(
+ minBucket, maxBucket, minTotalBuckets, maxTotalBuckets))
+ .orElse(true);
+ }
+
private Optional createPartitionSelector(BinaryRow partition) {
Optional partRemoved =
predicate.visit(
@@ -96,9 +118,14 @@ private Optional createPartitionSelector(BinaryRow partition)
return Optional.empty();
}
+ return createPartitionSelectorFromPredicate(partRemoved.get());
+ }
+
+ private Optional createPartitionSelectorFromPredicate(
+ Predicate sourcePredicate) {
List bucketFilters =
pickTransformFieldMapping(
- splitAnd(partRemoved.get()),
+ splitAnd(sourcePredicate),
rowType.getFieldNames(),
bucketKeyType.getFieldNames());
if (bucketFilters.isEmpty()) {
@@ -226,5 +253,25 @@ private Set createBucketSet(int numBucket) {
}
return builder.build();
}
+
+ private boolean mayContain(
+ int minBucket, int maxBucket, int minTotalBuckets, int maxTotalBuckets) {
+ long combinations = ((long) maxTotalBuckets - minTotalBuckets + 1) * bucketKeys.size();
+ if (combinations > MAX_MANIFEST_BUCKET_COMBINATIONS) {
+ return true;
+ }
+ for (int totalBuckets = minTotalBuckets; ; totalBuckets++) {
+ for (BinaryRow key : bucketKeys) {
+ int bucket = bucketFunction.bucket(key, totalBuckets);
+ if (bucket >= minBucket && bucket <= maxBucket) {
+ return true;
+ }
+ }
+ if (totalBuckets == maxTotalBuckets) {
+ break;
+ }
+ }
+ return false;
+ }
}
}
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 7eb4f66e32c8..8148c7422c39 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
@@ -630,6 +630,8 @@ private static final class BlockMetadataAccumulator {
private long schemaId = Long.MIN_VALUE;
private int minBucket = Integer.MAX_VALUE;
private int maxBucket = Integer.MIN_VALUE;
+ private int minTotalBuckets = Integer.MAX_VALUE;
+ private int maxTotalBuckets = Integer.MIN_VALUE;
private int minLevel = Integer.MAX_VALUE;
private int maxLevel = Integer.MIN_VALUE;
private long minRowId = Long.MAX_VALUE;
@@ -650,6 +652,9 @@ private void collect(ProjectedManifestEntry entry, SortKey key, BinaryRow partit
int bucket = entry.bucket();
minBucket = Math.min(minBucket, bucket);
maxBucket = Math.max(maxBucket, bucket);
+ int totalBuckets = entry.totalBuckets();
+ minTotalBuckets = Math.min(minTotalBuckets, totalBuckets);
+ maxTotalBuckets = Math.max(maxTotalBuckets, totalBuckets);
int level = entry.file().level();
minLevel = Math.min(minLevel, level);
maxLevel = Math.max(maxLevel, level);
@@ -665,6 +670,8 @@ private EncodedBlockMeta finish(SimpleStatsConverter partitionStatsConverter) {
schemaId,
minBucket,
maxBucket,
+ minTotalBuckets,
+ maxTotalBuckets,
minLevel,
maxLevel,
minRowId,
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryRunMergePlan.java b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryRunMergePlan.java
index 618943df180d..1645123f9128 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryRunMergePlan.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestEntryRunMergePlan.java
@@ -483,6 +483,7 @@ public boolean advance() throws Exception {
key.kind,
partitions.partition(key.partitionId),
currentEntry.bucket(),
+ currentEntry.totalBuckets(),
currentEntry.file().level(),
currentEntry.file().schemaId(),
key.firstRowId,
@@ -636,6 +637,7 @@ public void materializeCurrent() throws Exception {
key.kind,
partitions.partition(key.partitionId),
currentEntry.bucket(),
+ currentEntry.totalBuckets(),
currentEntry.file().level(),
currentEntry.file().schemaId(),
key.firstRowId,
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileBlockMerger.java b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileBlockMerger.java
index a3631122156a..fe1881b0f670 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileBlockMerger.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestFileBlockMerger.java
@@ -742,6 +742,7 @@ private static void writeCompactedEntry(
entry.kind().toByteValue(),
partition,
entry.bucket(),
+ entry.totalBuckets(),
file.level(),
file.schemaId(),
file.nonNullFirstRowId(),
@@ -751,6 +752,7 @@ private static void writeCompactedEntry(
entry.kind().toByteValue(),
partition,
entry.bucket(),
+ entry.totalBuckets(),
file.level(),
file.schemaId(),
file.rowCount());
@@ -771,6 +773,8 @@ private static final class CompactionBlock {
private long schemaId = Long.MIN_VALUE;
private int minBucket = Integer.MAX_VALUE;
private int maxBucket = Integer.MIN_VALUE;
+ private int minTotalBuckets = Integer.MAX_VALUE;
+ private int maxTotalBuckets = Integer.MIN_VALUE;
private int minLevel = Integer.MAX_VALUE;
private int maxLevel = Integer.MIN_VALUE;
private long minRowId = Long.MAX_VALUE;
@@ -817,6 +821,9 @@ private boolean collect(
int bucket = entry.bucket();
minBucket = Math.min(minBucket, bucket);
maxBucket = Math.max(maxBucket, bucket);
+ int totalBuckets = entry.totalBuckets();
+ minTotalBuckets = Math.min(minTotalBuckets, totalBuckets);
+ maxTotalBuckets = Math.max(maxTotalBuckets, totalBuckets);
int level = file.level();
minLevel = Math.min(minLevel, level);
maxLevel = Math.max(maxLevel, level);
@@ -864,6 +871,8 @@ private void finish(
schemaId,
minBucket,
maxBucket,
+ minTotalBuckets,
+ maxTotalBuckets,
minLevel,
maxLevel,
hasRowIds ? minRowId : -1,
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..a616d24a647d 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
@@ -70,6 +70,7 @@ public class ManifestFileSorter {
/** Context object that carries shared state across compaction methods. */
static class CompactionContext {
final boolean fullCompaction;
+ final boolean forceRewrite;
final boolean runMergeOptimizeEnabled;
final ManifestSortKey sortKey;
final RowType partitionType;
@@ -90,6 +91,7 @@ static class CompactionContext {
CompactionContext(
boolean fullCompaction,
+ boolean forceRewrite,
boolean runMergeOptimizeEnabled,
ManifestSortKey sortKey,
RowType partitionType,
@@ -99,6 +101,7 @@ static class CompactionContext {
List levelRuns,
List pickedRuns) {
this.fullCompaction = fullCompaction;
+ this.forceRewrite = forceRewrite;
this.runMergeOptimizeEnabled = runMergeOptimizeEnabled;
this.sortKey = sortKey;
this.partitionType = partitionType;
@@ -155,10 +158,12 @@ static List trySortCompaction(
@Nullable IOManager ioManager)
throws Exception {
String sortPartitionField = options.manifestSortPartitionField();
+ boolean sortBucketFirst = options.manifestSortBucketFirst();
boolean runMergeOptimizeEnabled = options.manifestMergeOptimizeEnabled();
long suggestedMetaSize = options.manifestTargetSize().getBytes();
int suggestedMinMetaCount = options.manifestMergeMinCount();
long fullCompactionThreshold = options.manifestFullCompactionThresholdSize().getBytes();
+ boolean forceRewrite = options.manifestSortForceRewrite();
long maxRewriteSize = options.manifestSortMaxRewriteSize();
int maxSizeAmplificationPercent = options.maxSizeAmplificationPercent();
int sortedRunSizeRatio = options.sortedRunSizeRatio();
@@ -173,11 +178,13 @@ static List trySortCompaction(
manifestFile,
partitionType,
sortPartitionField,
+ sortBucketFirst,
options.dataEvolutionEnabled(),
runMergeOptimizeEnabled,
suggestedMetaSize,
suggestedMinMetaCount,
fullCompactionThreshold,
+ forceRewrite,
maxRewriteSize,
maxSizeAmplificationPercent,
sortedRunSizeRatio,
@@ -192,6 +199,7 @@ static List trySortCompaction(
manifestFile,
partitionType,
sortPartitionField,
+ sortBucketFirst,
options.dataEvolutionEnabled(),
runMergeOptimizeEnabled,
suggestedMetaSize,
@@ -215,11 +223,13 @@ private static Optional> tryFullCompaction(
ManifestFile manifestFile,
RowType partitionType,
String sortPartitionField,
+ boolean sortBucketFirst,
boolean dataEvolutionEnabled,
boolean runMergeOptimizeEnabled,
long suggestedMetaSize,
int suggestedMinMetaCount,
long fullCompactionThreshold,
+ boolean forceRewrite,
long maxRewriteSize,
int maxSizeAmplificationPercent,
int sortedRunSizeRatio,
@@ -227,7 +237,9 @@ private static Optional> tryFullCompaction(
@Nullable Integer manifestReadParallelism)
throws Exception {
// Step 1: Check if full compaction threshold is met
- if (!reachesFullCompactionThreshold(input, suggestedMetaSize, fullCompactionThreshold)) {
+ if (!forceRewrite
+ && !reachesFullCompactionThreshold(
+ input, suggestedMetaSize, fullCompactionThreshold)) {
return Optional.empty();
}
// Step 2: Prepare compaction context
@@ -235,9 +247,11 @@ private static Optional> tryFullCompaction(
prepareCompaction(
input,
true,
+ forceRewrite,
manifestFile,
partitionType,
sortPartitionField,
+ sortBucketFirst,
dataEvolutionEnabled,
runMergeOptimizeEnabled,
suggestedMetaSize,
@@ -248,6 +262,9 @@ private static Optional> tryFullCompaction(
try {
List levelRuns = ctx.levelRuns;
List pickedRuns = ctx.pickedRuns;
+ if (forceRewrite) {
+ pickedRuns = new ArrayList<>(levelRuns);
+ }
if (pickedRuns.isEmpty() && ctx.defaultCompactFiles.isEmpty()) {
LOG.debug(
@@ -277,9 +294,24 @@ private static Optional> tryFullCompaction(
}
pickedFiles.addAll(ctx.defaultCompactFiles.keySet());
- // Step 4: Split into sections and merge small adjacent sections
- List sections = splitIntoSections(pickedFiles, ctx);
- sections = mergeSmallAdjacentSections(sections, suggestedMetaSize);
+ // Step 4: Split into sections and merge small adjacent sections. A forced rewrite
+ // intentionally uses one global section so entries with the same partition but from
+ // different already-compacted manifests can be clustered by the secondary sort key.
+ List sections;
+ if (forceRewrite) {
+ long totalSize = 0L;
+ boolean hasDefaultCompactFile = false;
+ for (ManifestFileMeta file : pickedFiles) {
+ totalSize += file.fileSize();
+ hasDefaultCompactFile |= ctx.isMarkedForDefaultCompaction(file);
+ }
+ sections =
+ Collections.singletonList(
+ new Section(pickedFiles, totalSize, hasDefaultCompactFile));
+ } else {
+ sections = splitIntoSections(pickedFiles, ctx);
+ sections = mergeSmallAdjacentSections(sections, suggestedMetaSize);
+ }
LOG.info(
"Manifest sort full compact: pickedFiles={}, sections={}.",
@@ -321,6 +353,7 @@ private static List tryMinorCompaction(
ManifestFile manifestFile,
RowType partitionType,
String sortPartitionField,
+ boolean sortBucketFirst,
boolean dataEvolutionEnabled,
boolean runMergeOptimizeEnabled,
long suggestedMetaSize,
@@ -336,9 +369,11 @@ private static List tryMinorCompaction(
prepareCompaction(
input,
false,
+ false,
manifestFile,
partitionType,
sortPartitionField,
+ sortBucketFirst,
dataEvolutionEnabled,
runMergeOptimizeEnabled,
suggestedMetaSize,
@@ -450,9 +485,11 @@ private static List tryMinorCompaction(
private static CompactionContext prepareCompaction(
List input,
boolean fullCompaction,
+ boolean forceRewrite,
ManifestFile manifestFile,
RowType partitionType,
String sortPartitionField,
+ boolean sortBucketFirst,
boolean dataEvolutionEnabled,
boolean runMergeOptimizeEnabled,
long suggestedMetaSize,
@@ -464,7 +501,8 @@ 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(rowIdSort, sortPartitionField, partitionType, sortBucketFirst);
// Step 2: Classify manifests into LSM files and collect delete entries.
ClassifyResult classification =
@@ -489,6 +527,7 @@ private static CompactionContext prepareCompaction(
return new CompactionContext(
fullCompaction,
+ forceRewrite,
useRunMergeOptimize,
sortKey,
partitionType,
@@ -1069,7 +1108,9 @@ private static void rewriteSection(
@Nullable Integer manifestReadParallelism)
throws Exception {
// Skip rewrite for single file not in delete-range.
- if (section.size() == 1 && !ctx.defaultCompactFiles.getOrDefault(section.get(0), false)) {
+ if (section.size() == 1
+ && !ctx.forceRewrite
+ && !ctx.defaultCompactFiles.getOrDefault(section.get(0), false)) {
output.addUnchanged(section.get(0));
return;
}
@@ -1186,11 +1227,15 @@ static ManifestSortKey createSortKey(
return createSortKey(
dataEvolutionEnabled && ManifestFileMeta.allContainsRowId(input),
sortPartitionField,
- partitionType);
+ partitionType,
+ false);
}
private static ManifestSortKey createSortKey(
- boolean rowIdSort, String sortPartitionField, RowType partitionType) {
+ boolean rowIdSort,
+ String sortPartitionField,
+ RowType partitionType,
+ boolean sortBucketFirst) {
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,7 +1264,8 @@ private static ManifestSortKey createSortKey(
RecordComparator fieldComparator =
CodeGenUtils.newRecordComparator(
partitionType.getFieldTypes(), new int[] {sortFieldIndex});
- return new PartitionSortKey(fieldComparator, partitionType, sortFieldIndex);
+ return new PartitionSortKey(
+ fieldComparator, partitionType, sortFieldIndex, sortBucketFirst);
}
private static int[] createPartitionSortFields(
@@ -1282,36 +1328,61 @@ private static class PartitionSortKey implements ManifestSortKey {
private final RowType externalSortRowType;
private final int[] externalSortKeyFields;
private final int sortFieldNum;
+ private final boolean sortBucketFirst;
private PartitionSortKey(
- RecordComparator fieldComparator, RowType partitionType, int sortFieldIndex) {
+ RecordComparator fieldComparator,
+ RowType partitionType,
+ int sortFieldIndex,
+ boolean sortBucketFirst) {
this.fieldComparator = fieldComparator;
+ this.sortBucketFirst = sortBucketFirst;
DataType sortFieldType = partitionType.getTypeAt(sortFieldIndex);
this.sortFieldGetter = InternalRow.createFieldGetter(sortFieldType, sortFieldIndex);
- this.sortFieldNum = 3;
+ this.sortFieldNum = 4;
this.externalSortRowType =
DataTypes.ROW(
sortFieldType,
+ DataTypes.INT(),
DataTypes.TINYINT(),
DataTypes.STRING(),
ManifestEntry.MANIFEST_ROW_TYPE);
- this.externalSortKeyFields = createSequentialFields(sortFieldNum);
+ this.externalSortKeyFields =
+ sortBucketFirst ? new int[] {1, 0, 2, 3} : createSequentialFields(sortFieldNum);
}
@Override
public int compareMin(ManifestFileMeta a, ManifestFileMeta b) {
+ if (sortBucketFirst && a.minBucket() != null && b.minBucket() != null) {
+ int bucketComparison = Integer.compare(a.minBucket(), b.minBucket());
+ if (bucketComparison != 0) {
+ return bucketComparison;
+ }
+ }
return fieldComparator.compare(
a.partitionStats().minValues(), b.partitionStats().minValues());
}
@Override
public int compareMax(ManifestFileMeta a, ManifestFileMeta b) {
+ if (sortBucketFirst && a.maxBucket() != null && b.maxBucket() != null) {
+ int bucketComparison = Integer.compare(a.maxBucket(), b.maxBucket());
+ if (bucketComparison != 0) {
+ return bucketComparison;
+ }
+ }
return fieldComparator.compare(
a.partitionStats().maxValues(), b.partitionStats().maxValues());
}
@Override
public boolean isAfterMax(ManifestFileMeta file, ManifestFileMeta maxFile) {
+ if (sortBucketFirst && file.minBucket() != null && maxFile.maxBucket() != null) {
+ int bucketComparison = Integer.compare(file.minBucket(), maxFile.maxBucket());
+ if (bucketComparison != 0) {
+ return bucketComparison > 0;
+ }
+ }
return fieldComparator.compare(
file.partitionStats().minValues(), maxFile.partitionStats().maxValues())
>= 0;
@@ -1331,13 +1402,14 @@ public int[] externalSortKeyFields() {
public void replaceExternalSortRow(
GenericRow row, ManifestEntry entry, InternalRow binaryManifestRow) {
row.setField(0, sortFieldGetter.getFieldOrNull(entry.partition()));
- row.setField(1, entry.kind().toByteValue());
+ row.setField(1, entry.bucket());
+ row.setField(2, entry.kind().toByteValue());
row.setField(
- 2,
+ 3,
entry instanceof ProjectedManifestEntry
? ((ProjectedManifestEntry) entry).file().fileNameBinary()
: BinaryString.fromString(entry.file().fileName()));
- row.setField(3, binaryManifestRow);
+ row.setField(4, binaryManifestRow);
}
@Override
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestsReader.java b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestsReader.java
index 46bbbd17f6e3..618eae930d40 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestsReader.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestsReader.java
@@ -20,6 +20,7 @@
import org.apache.paimon.Snapshot;
import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.manifest.BucketFilter;
import org.apache.paimon.manifest.ManifestFileMeta;
import org.apache.paimon.manifest.ManifestList;
import org.apache.paimon.partition.PartitionPredicate;
@@ -28,8 +29,10 @@
import org.apache.paimon.table.source.ScanMode;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.BiFilter;
+import org.apache.paimon.utils.Filter;
import org.apache.paimon.utils.RowRangeIndex;
import org.apache.paimon.utils.SnapshotManager;
+import org.apache.paimon.utils.TriFilter;
import javax.annotation.Nullable;
import javax.annotation.concurrent.ThreadSafe;
@@ -53,6 +56,8 @@ public class ManifestsReader {
private boolean onlyReadRealBuckets = false;
@Nullable private Integer specifiedBucket = null;
+ @Nullable private Filter bucketFilter = null;
+ @Nullable private TriFilter totalAwareBucketFilter = null;
@Nullable private Integer specifiedLevel = null;
@Nullable private PartitionPredicate partitionFilter = null;
// Auth partition filter (ANDed with partitionFilter); kept separate so it can be reset each
@@ -82,6 +87,17 @@ public ManifestsReader withBucket(int bucket) {
return this;
}
+ public ManifestsReader withBucketFilter(Filter bucketFilter) {
+ this.bucketFilter = bucketFilter;
+ return this;
+ }
+
+ public ManifestsReader withTotalAwareBucketFilter(
+ TriFilter totalAwareBucketFilter) {
+ this.totalAwareBucketFilter = totalAwareBucketFilter;
+ return this;
+ }
+
public ManifestsReader withLevel(int level) {
this.specifiedLevel = level;
return this;
@@ -147,9 +163,15 @@ public Result read(@Nullable Snapshot specifiedSnapshot, ScanMode scanMode) {
// Compute the effective partition filter once (it ANDs the base and auth slots) instead of
// rebuilding it per manifest.
PartitionPredicate effectivePartitionFilter = partitionFilter();
+ BucketFilter effectiveBucketFilter =
+ BucketFilter.create(
+ onlyReadRealBuckets, specifiedBucket, bucketFilter, totalAwareBucketFilter);
List filtered =
manifests.stream()
- .filter(m -> filterManifestFileMeta(m, effectivePartitionFilter))
+ .filter(
+ m ->
+ filterManifestFileMeta(
+ m, effectivePartitionFilter, effectiveBucketFilter))
.collect(Collectors.toList());
return new Result(snapshot, manifests, filtered);
}
@@ -183,17 +205,11 @@ private boolean filterManifestByRowRanges(ManifestFileMeta manifest) {
/** Note: Keep this thread-safe. */
private boolean filterManifestFileMeta(
- ManifestFileMeta manifest, @Nullable PartitionPredicate effectivePartitionFilter) {
- Integer minBucket = manifest.minBucket();
- Integer maxBucket = manifest.maxBucket();
- if (minBucket != null && maxBucket != null) {
- if (onlyReadRealBuckets && maxBucket < 0) {
- return false;
- }
- if (specifiedBucket != null
- && (specifiedBucket < minBucket || specifiedBucket > maxBucket)) {
- return false;
- }
+ ManifestFileMeta manifest,
+ @Nullable PartitionPredicate effectivePartitionFilter,
+ @Nullable BucketFilter effectiveBucketFilter) {
+ if (effectiveBucketFilter != null && !effectiveBucketFilter.mayContain(manifest)) {
+ return false;
}
Integer minLevel = manifest.minLevel();
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/ManifestsTable.java b/paimon-core/src/main/java/org/apache/paimon/table/system/ManifestsTable.java
index 7bc8f23b4259..8cd663255d3c 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/system/ManifestsTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/system/ManifestsTable.java
@@ -55,6 +55,7 @@
import org.apache.paimon.table.source.snapshot.TimeTravelUtil;
import org.apache.paimon.types.BigIntType;
import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.IntType;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.FileStorePathFactory;
import org.apache.paimon.utils.IteratorRecordReader;
@@ -105,7 +106,11 @@ public class ManifestsTable implements ReadonlyTable {
"max_partition_stats",
SerializationUtils.newStringType(true)),
new DataField(7, "min_row_id", new BigIntType(true)),
- new DataField(8, "max_row_id", new BigIntType(true))));
+ new DataField(8, "max_row_id", new BigIntType(true)),
+ new DataField(9, "min_bucket", new IntType(true)),
+ new DataField(10, "max_bucket", new IntType(true)),
+ new DataField(11, "min_total_buckets", new IntType(true)),
+ new DataField(12, "max_total_buckets", new IntType(true))));
private final FileStoreTable dataTable;
@@ -347,7 +352,11 @@ private InternalRow toRow(
partitionCastExecutor.cast(manifestFileMeta.partitionStats().minValues()),
partitionCastExecutor.cast(manifestFileMeta.partitionStats().maxValues()),
manifestFileMeta.minRowId(),
- manifestFileMeta.maxRowId());
+ manifestFileMeta.maxRowId(),
+ manifestFileMeta.minBucket(),
+ manifestFileMeta.maxBucket(),
+ manifestFileMeta.minTotalBuckets(),
+ manifestFileMeta.maxTotalBuckets());
}
}
diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaSerializerTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaSerializerTest.java
index 2f4e32cd2759..f20445bfe902 100644
--- a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaSerializerTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileMetaSerializerTest.java
@@ -68,7 +68,9 @@ void testExtraFiles() throws IOException {
original.maxLevel(),
original.minRowId(),
original.maxRowId(),
- extraFiles);
+ extraFiles,
+ original.minTotalBuckets(),
+ original.maxTotalBuckets());
ManifestFileMeta fromRow = serializer.fromRow(serializer.toRow(meta));
ManifestFileMeta fromBytes = serializer.deserializeFromBytes(meta.toBytes());
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 593b625cec73..9eeedc9c31f7 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,119 @@ public void testManifestSortWithOverlappingPartitions() {
}
}
+ @Test
+ public void testManifestSortUsesBucketAsSecondaryKey() {
+ 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(0, 1, 2, 3);
+ }
+
+ @Test
+ public void testManifestSortCanUseBucketAsPrimaryKey() {
+ 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.MANIFEST_SORT_BUCKET_FIRST, 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);
+ 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 testManifestSortForceRewriteAlreadyCompactedRun() {
+ List physical =
+ Arrays.asList(
+ makeManifest(makeBucketEntry("a-3", 0, 3), makeBucketEntry("a-1", 0, 1)),
+ makeManifest(makeBucketEntry("b-2", 0, 2), makeBucketEntry("b-0", 0, 0)));
+ long targetSize = CoreOptions.MANIFEST_TARGET_FILE_SIZE.defaultValue().getBytes();
+ List input =
+ physical.stream()
+ .map(meta -> copyWithFileSize(meta, targetSize))
+ .collect(Collectors.toList());
+
+ Options testOptions = new Options();
+ testOptions.set(CoreOptions.MANIFEST_SORT_ENABLED, true);
+ testOptions.set(CoreOptions.MANIFEST_SORT_FORCE_REWRITE, true);
+ testOptions.set(CoreOptions.MANIFEST_SORT_MAX_REWRITE_SIZE.key(), "1G");
+ List merged =
+ ManifestFileMerger.merge(
+ input,
+ manifestFile,
+ getPartitionType(),
+ CoreOptions.fromMap(testOptions.toMap()));
+
+ assertThat(merged)
+ .extracting(ManifestFileMeta::fileName)
+ .doesNotContainAnyElementsOf(
+ input.stream()
+ .map(ManifestFileMeta::fileName)
+ .collect(Collectors.toList()));
+ assertThat(readEntries(merged))
+ .extracting(ManifestEntry::bucket)
+ .containsExactly(0, 1, 2, 3);
+ }
+
+ @Test
+ public void testManifestBucketFilterFallsBackForLegacyMetadata() {
+ ManifestFileMeta current = makeManifest(makeBucketEntry("bucket-1", 0, 1));
+ ManifestBucketFilter rangeFilter =
+ new ManifestBucketFilter() {
+ @Override
+ public boolean test(BinaryRow partition, Integer bucket, Integer totalBuckets) {
+ return true;
+ }
+
+ @Override
+ public boolean mayContain(
+ int minBucket,
+ int maxBucket,
+ int minTotalBuckets,
+ int maxTotalBuckets) {
+ return false;
+ }
+ };
+ BucketFilter filter = new BucketFilter(false, null, null, rangeFilter);
+
+ assertThat(filter.mayContain(current)).isFalse();
+ assertThat(filter.mayContain(copyWithoutTotalBucketStats(current))).isTrue();
+ }
+
@Test
public void testManifestSortMinorCompactionRespectsMergeMinCount() {
List input = new ArrayList<>();
@@ -2702,6 +2815,50 @@ 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());
+ }
+
+ private ManifestFileMeta copyWithFileSize(ManifestFileMeta meta, long fileSize) {
+ return new ManifestFileMeta(
+ meta.fileName(),
+ fileSize,
+ meta.numAddedFiles(),
+ meta.numDeletedFiles(),
+ meta.partitionStats(),
+ meta.schemaId(),
+ meta.minBucket(),
+ meta.maxBucket(),
+ meta.minLevel(),
+ meta.maxLevel(),
+ meta.minRowId(),
+ meta.maxRowId(),
+ meta.extraFiles(),
+ meta.minTotalBuckets(),
+ meta.maxTotalBuckets());
+ }
+
+ private ManifestFileMeta copyWithoutTotalBucketStats(ManifestFileMeta meta) {
+ return new ManifestFileMeta(
+ meta.fileName(),
+ meta.fileSize(),
+ meta.numAddedFiles(),
+ meta.numDeletedFiles(),
+ meta.partitionStats(),
+ meta.schemaId(),
+ meta.minBucket(),
+ meta.maxBucket(),
+ meta.minLevel(),
+ meta.maxLevel(),
+ meta.minRowId(),
+ meta.maxRowId(),
+ meta.extraFiles(),
+ null,
+ null);
+ }
+
/** 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/manifest/ManifestFileTest.java b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java
index c3a50f4ef1de..89cc2026dd17 100644
--- a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestFileTest.java
@@ -193,6 +193,7 @@ void testWriteEncodedRecords() throws Exception {
entry.kind().toByteValue(),
entry.partition(),
entry.bucket(),
+ entry.totalBuckets(),
entry.level(),
entry.file().schemaId(),
entry.file().firstRowId(),
@@ -210,6 +211,8 @@ void testWriteEncodedRecords() throws Exception {
assertThat(result.schemaId()).isEqualTo(sourceMeta.schemaId());
assertThat(result.minBucket()).isEqualTo(sourceMeta.minBucket());
assertThat(result.maxBucket()).isEqualTo(sourceMeta.maxBucket());
+ assertThat(result.minTotalBuckets()).isEqualTo(sourceMeta.minTotalBuckets());
+ assertThat(result.maxTotalBuckets()).isEqualTo(sourceMeta.maxTotalBuckets());
assertThat(result.minLevel()).isEqualTo(sourceMeta.minLevel());
assertThat(result.maxLevel()).isEqualTo(sourceMeta.maxLevel());
assertThat(result.minRowId()).isEqualTo(sourceMeta.minRowId());
@@ -246,6 +249,7 @@ void testWriteEncodedRecordsFlushesPartitionStatsBuffer() throws Exception {
source.kind().toByteValue(),
source.partition().copy(),
source.bucket(),
+ source.totalBuckets(),
source.level(),
source.file().schemaId(),
source.file().firstRowId(),
@@ -863,6 +867,31 @@ void testReadDeletedEntriesWithProjectedScan() throws Exception {
assertThat(deleted)
.containsExactlyInAnyOrder(firstDelete.identifier(), secondDelete.identifier());
+
+ PartitionPredicate partitionFilter =
+ PartitionPredicate.fromMultiple(
+ DEFAULT_PART_TYPE, Collections.singletonList(first.partition()));
+ BucketFilter bucketFilter = new BucketFilter(false, first.bucket(), null, null);
+ Set filteredDeleted =
+ FileEntry.readDeletedEntries(
+ manifestFile,
+ Arrays.asList(firstManifest, secondManifest),
+ 2,
+ partitionFilter,
+ bucketFilter);
+ List expectedFilteredDeleted =
+ Arrays.asList(firstDelete, secondDelete).stream()
+ .filter(entry -> partitionFilter.test(entry.partition()))
+ .filter(
+ entry ->
+ bucketFilter.test(
+ entry.partition(),
+ entry.bucket(),
+ entry.totalBuckets()))
+ .map(ManifestEntry::identifier)
+ .collect(Collectors.toList());
+
+ assertThat(filteredDeleted).containsExactlyInAnyOrderElementsOf(expectedFilteredDeleted);
}
@Test
@@ -1185,6 +1214,8 @@ private ManifestAvroWriter.EncodedBlockMeta encodedBlockMeta(ManifestFileMeta me
meta.schemaId(),
meta.minBucket(),
meta.maxBucket(),
+ meta.minTotalBuckets(),
+ meta.maxTotalBuckets(),
meta.minLevel(),
meta.maxLevel(),
meta.minRowId() == null ? -1 : meta.minRowId(),
diff --git a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestTestDataGenerator.java b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestTestDataGenerator.java
index 4b576c6bd64c..91b758221bbb 100644
--- a/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestTestDataGenerator.java
+++ b/paimon-core/src/test/java/org/apache/paimon/manifest/ManifestTestDataGenerator.java
@@ -101,6 +101,8 @@ public ManifestFileMeta createManifestFileMeta(List entries) {
long numDeletedFiles = 0;
int minBucket = Integer.MAX_VALUE;
int maxBucket = Integer.MIN_VALUE;
+ int minTotalBuckets = Integer.MAX_VALUE;
+ int maxTotalBuckets = Integer.MIN_VALUE;
int minLevel = Integer.MAX_VALUE;
int maxLevel = Integer.MIN_VALUE;
for (ManifestEntry entry : entries) {
@@ -112,6 +114,8 @@ public ManifestFileMeta createManifestFileMeta(List entries) {
}
minBucket = Math.min(minBucket, entry.bucket());
maxBucket = Math.max(maxBucket, entry.bucket());
+ minTotalBuckets = Math.min(minTotalBuckets, entry.totalBuckets());
+ maxTotalBuckets = Math.max(maxTotalBuckets, entry.totalBuckets());
minLevel = Math.min(minLevel, entry.level());
maxLevel = Math.max(maxLevel, entry.level());
}
@@ -128,7 +132,10 @@ public ManifestFileMeta createManifestFileMeta(List entries) {
minLevel,
maxLevel,
null,
- null);
+ null,
+ null,
+ minTotalBuckets,
+ maxTotalBuckets);
}
private void mergeLevelsIfNeeded(BinaryRow partition, int bucket) {
diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/BucketSelectorTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/BucketSelectorTest.java
index 3128ff20db22..00ccd4d85739 100644
--- a/paimon-core/src/test/java/org/apache/paimon/operation/BucketSelectorTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/operation/BucketSelectorTest.java
@@ -27,8 +27,10 @@
import org.junit.jupiter.api.Test;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
+import java.util.List;
import java.util.Set;
import static org.assertj.core.api.Assertions.assertThat;
@@ -57,6 +59,44 @@ public void testEqualPredicate() {
assertThat(selected).hasSize(1);
}
+ @Test
+ public void testManifestBucketRange() {
+ RowType rowType = DataTypes.ROW(DataTypes.FIELD(0, "k", DataTypes.INT()));
+ RowType partType = RowType.of();
+ RowType bucketKeyType = DataTypes.ROW(DataTypes.FIELD(0, "k", DataTypes.INT()));
+ PredicateBuilder pb = new PredicateBuilder(rowType);
+ BucketSelector selector =
+ new BucketSelector(
+ pb.equal(0, 5),
+ BucketFunctionType.DEFAULT,
+ rowType,
+ partType,
+ bucketKeyType);
+
+ int selected =
+ selectedBuckets(selector, BinaryRow.EMPTY_ROW, NUM_BUCKETS).iterator().next();
+ assertThat(selector.mayContain(selected, selected, NUM_BUCKETS, NUM_BUCKETS)).isTrue();
+ int different = (selected + 1) % NUM_BUCKETS;
+ assertThat(selector.mayContain(different, different, NUM_BUCKETS, NUM_BUCKETS)).isFalse();
+
+ // Unknown or excessively broad total-bucket ranges must fall back conservatively.
+ assertThat(selector.mayContain(0, 0, 0, NUM_BUCKETS)).isTrue();
+ assertThat(selector.mayContain(0, 0, 1, 20_001)).isTrue();
+
+ List