[core] Prune manifests by bucket metadata

docs/generated/core_configuration.html

@@ -1059,12 +1059,24 @@
             <td>Integer</td>
             <td>Level threshold of lookup to generate remote lookup files. Level files below this threshold will not generate remote lookup files.</td>
         </tr>
+        <tr>
+            <td><h5>manifest-sort.bucket-first</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>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.</td>
+        </tr>
         <tr>
             <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>
         </tr>
+        <tr>
+            <td><h5>manifest-sort.force-rewrite</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>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.</td>
+        </tr>
         <tr>
             <td><h5>manifest-sort.max-rewrite-size</h5></td>
             <td style="word-wrap: break-word;">256 mb</td>

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<Boolean> 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<MemorySize> 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<Boolean> 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<Boolean> 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);
     }

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<Integer> 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;
+    }
 }

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<Identifier> readDeletedEntries(
             ManifestFile manifestFile,
             List<ManifestFileMeta> manifestFiles,
             @Nullable Integer manifestReadParallelism) {
+        return readDeletedEntries(manifestFile, manifestFiles, manifestReadParallelism, null, null);
+    }
+
+    static Set<Identifier> readDeletedEntries(
+            ManifestFile manifestFile,
+            List<ManifestFileMeta> manifestFiles,
+            @Nullable Integer manifestReadParallelism,
+            @Nullable PartitionPredicate partitionFilter,
+            @Nullable BucketFilter bucketFilter) {
         manifestFiles =
                 manifestFiles.stream()
                         .filter(file -> file.numDeletedFiles() > 0)
@@ -372,7 +382,9 @@ static Set<Identifier> readDeletedEntries(
                     try (CloseableIterator<ProjectedManifestEntry> 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()) {

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,13 +307,15 @@ public EncodedEntry replace(
                 byte kind,
                 BinaryRow partition,
                 int bucket,
+                int totalBuckets,
                 int level,
                 long schemaId,
                 long firstRowId,
                 long rowCount) {
             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);
         }
     }
 

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<BinaryRow, Integer, Integer> {
+
+    /** Returns false only when no entry in the supplied ranges can match. */
+    boolean mayContain(int minBucket, int maxBucket, int minTotalBuckets, int maxTotalBuckets);
+}

paimon-core/src/main/java/org/apache/paimon/manifest/ManifestFile.java

@@ -168,14 +168,26 @@ public <T> List<T> read(
      * materialized with the complete manifest schema.
      */
     public CloseableIterator<ProjectedManifestEntry> 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<ProjectedManifestEntry> scan(
+            String fileName,
+            Projection projection,
+            @Nullable PartitionPredicate partitionFilter,
+            @Nullable BucketFilter bucketFilter) {
         try {
             CloseableIterator<InternalRow> rows =
                     createManifestIterator(
                             fileIO,
                             pathFactory.toPath(fileName),
                             projection.projectedType(),
-                            null,
-                            null);
+                            partitionFilter,
+                            bucketFilter);
             return new CloseableIterator<ProjectedManifestEntry>() {
 
                 @Override