[core] Support bucket pruning for manifest files

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

@@ -77,4 +77,29 @@ public boolean test(BinaryRow partition, int bucket, int totalBucket) {
         return totalAwareBucketFilter == null
                 || totalAwareBucketFilter.test(partition, bucket, totalBucket);
     }
+
+    /** Conservatively tests whether a manifest's bucket metadata 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 (totalAwareBucketFilter instanceof ManifestBucketFilter) {
+            Integer totalBuckets = manifest.totalBuckets();
+            if (minBucket < 0 || totalBuckets == null || totalBuckets <= 0) {
+                return true;
+            }
+            return ((ManifestBucketFilter) totalAwareBucketFilter)
+                    .mayContain(minBucket, maxBucket, totalBuckets);
+        }
+        return true;
+    }
 }

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,7 @@ public static final class EncodedBlockMeta {
         private final long schemaId;
         private final int minBucket;
         private final int maxBucket;
+        private final @Nullable Integer totalBuckets;
         private final int minLevel;
         private final int maxLevel;
         private final long minRowId;
@@ -360,6 +366,7 @@ public EncodedBlockMeta(
                 long schemaId,
                 int minBucket,
                 int maxBucket,
+                @Nullable Integer totalBuckets,
                 int minLevel,
                 int maxLevel,
                 long minRowId,
@@ -370,6 +377,7 @@ public EncodedBlockMeta(
             this.schemaId = schemaId;
             this.minBucket = minBucket;
             this.maxBucket = maxBucket;
+            this.totalBuckets = totalBuckets;
             this.minLevel = minLevel;
             this.maxLevel = maxLevel;
             this.minRowId = minRowId;
@@ -396,9 +404,11 @@ private final class FileWriter {
         private long schemaId = Long.MIN_VALUE;
         private int minBucket = Integer.MAX_VALUE;
         private int maxBucket = Integer.MIN_VALUE;
+        private @Nullable Integer totalBuckets;
         private int minLevel = Integer.MAX_VALUE;
         private int maxLevel = Integer.MIN_VALUE;
         private boolean bucketStatsKnown = true;
+        private boolean totalBucketsKnown = true;
         private boolean levelStatsKnown = true;
         private @Nullable RowIdStats rowIdStats = new RowIdStats();
         private boolean closed;
@@ -473,6 +483,7 @@ private void collectStats(ManifestEntry entry) {
             schemaId = Math.max(schemaId, entry.file().schemaId());
             minBucket = Math.min(minBucket, entry.bucket());
             maxBucket = Math.max(maxBucket, entry.bucket());
+            collectTotalBuckets(entry.totalBuckets());
             minLevel = Math.min(minLevel, entry.level());
             maxLevel = Math.max(maxLevel, entry.level());
             if (rowIdStats != null) {
@@ -500,6 +511,7 @@ private void collectStats(EncodedEntry entry) {
             schemaId = Math.max(schemaId, entry.schemaId);
             minBucket = Math.min(minBucket, entry.bucket);
             maxBucket = Math.max(maxBucket, entry.bucket);
+            collectTotalBuckets(entry.totalBuckets);
             minLevel = Math.min(minLevel, entry.level);
             maxLevel = Math.max(maxLevel, entry.level);
             if (rowIdStats != null) {
@@ -517,6 +529,7 @@ private void collectStats(EncodedBlockMeta metadata) {
             schemaId = Math.max(schemaId, metadata.schemaId);
             minBucket = Math.min(minBucket, metadata.minBucket);
             maxBucket = Math.max(maxBucket, metadata.maxBucket);
+            collectTotalBuckets(metadata.totalBuckets);
             minLevel = Math.min(minLevel, metadata.minLevel);
             maxLevel = Math.max(maxLevel, metadata.maxLevel);
             if (rowIdStats != null) {
@@ -538,6 +551,7 @@ private void collectStats(ManifestFileMeta manifest) {
                 minBucket = Math.min(minBucket, manifest.minBucket());
                 maxBucket = Math.max(maxBucket, manifest.maxBucket());
             }
+            collectTotalBuckets(manifest.totalBuckets());
             if (manifest.minLevel() == null || manifest.maxLevel() == null) {
                 levelStatsKnown = false;
             } else {
@@ -555,6 +569,21 @@ private void collectStats(ManifestFileMeta manifest) {
             collectCopiedPartitionStats(manifest.partitionStats());
         }
 
+        private void collectTotalBuckets(@Nullable Integer candidate) {
+            if (!totalBucketsKnown) {
+                return;
+            }
+            if (candidate == null || candidate <= 0) {
+                totalBucketsKnown = false;
+                totalBuckets = null;
+            } else if (totalBuckets == null) {
+                totalBuckets = candidate;
+            } else if (!totalBuckets.equals(candidate)) {
+                totalBucketsKnown = false;
+                totalBuckets = null;
+            }
+        }
+
         private void collectCopiedPartitionStats(SimpleStats partitionStats) {
             collectCopiedPartitionRepresentative(partitionStats.minValues());
             if (!partitionStats.maxValues().equals(partitionStats.minValues())) {
@@ -709,7 +738,9 @@ 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,
+                    totalBucketsKnown ? totalBuckets : 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-aware bucket filter which can conservatively prune manifest files. */
+public interface ManifestBucketFilter extends TriFilter<BinaryRow, Integer, Integer> {
+
+    /** Returns whether a manifest bucket range may contain a matching bucket. */
+    boolean mayContain(int minBucket, int maxBucket, int totalBuckets);
+}

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

@@ -62,8 +62,8 @@ 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, "_TOTAL_BUCKETS", new IntType(true))));
 
     private final String fileName;
     private final long fileSize;
@@ -78,6 +78,7 @@ public class ManifestFileMeta {
     private final @Nullable Long minRowId;
     private final @Nullable Long maxRowId;
     private final @Nullable List<String> extraFiles;
+    private final @Nullable Integer totalBuckets;
 
     public ManifestFileMeta(
             String fileName,
@@ -105,6 +106,7 @@ public ManifestFileMeta(
                 maxLevel,
                 minRowId,
                 maxRowId,
+                null,
                 null);
     }
 
@@ -122,6 +124,38 @@ public ManifestFileMeta(
             @Nullable Long minRowId,
             @Nullable Long maxRowId,
             @Nullable List<String> extraFiles) {
+        this(
+                fileName,
+                fileSize,
+                numAddedFiles,
+                numDeletedFiles,
+                partitionStats,
+                schemaId,
+                minBucket,
+                maxBucket,
+                minLevel,
+                maxLevel,
+                minRowId,
+                maxRowId,
+                extraFiles,
+                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<String> extraFiles,
+            @Nullable Integer totalBuckets) {
         this.fileName = fileName;
         this.fileSize = fileSize;
         this.numAddedFiles = numAddedFiles;
@@ -135,6 +169,7 @@ public ManifestFileMeta(
         this.minRowId = minRowId;
         this.maxRowId = maxRowId;
         this.extraFiles = extraFiles;
+        this.totalBuckets = totalBuckets;
     }
 
     public String fileName() {
@@ -189,6 +224,10 @@ public long schemaId() {
         return extraFiles;
     }
 
+    public @Nullable Integer totalBuckets() {
+        return totalBuckets;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (!(o instanceof ManifestFileMeta)) {
@@ -207,7 +246,8 @@ 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(totalBuckets, that.totalBuckets);
     }
 
     @Override
@@ -225,13 +265,14 @@ public int hashCode() {
                 maxLevel,
                 minRowId,
                 maxRowId,
-                extraFiles);
+                extraFiles,
+                totalBuckets);
     }
 
     @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}",
                 fileName,
                 fileSize,
                 numAddedFiles,
@@ -244,7 +285,8 @@ public String toString() {
                 maxLevel,
                 minRowId,
                 maxRowId,
-                extraFiles);
+                extraFiles,
+                totalBuckets);
     }
 
     // ----------------------- Serialization -----------------------------

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

@@ -60,7 +60,8 @@ public InternalRow toRow(ManifestFileMeta meta) {
                 meta.maxLevel(),
                 meta.minRowId(),
                 meta.maxRowId(),
-                toStringArrayData(meta.extraFiles()));
+                toStringArrayData(meta.extraFiles()),
+                meta.totalBuckets());
     }
 
     @Override
@@ -95,6 +96,7 @@ 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));
     }
 }

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

@@ -142,6 +142,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(

paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java

@@ -157,13 +157,15 @@ public FileStoreScan withBucket(int bucket) {
 
     @Override
     public FileStoreScan withBucketFilter(Filter<Integer> bucketFilter) {
+        manifestsReader.withBucketFilter(bucketFilter);
         this.bucketFilter = bucketFilter;
         return this;
     }
 
     @Override
     public FileStoreScan withTotalAwareBucketFilter(
             TriFilter<BinaryRow, Integer, Integer> totalAwareBucketFilter) {
+        manifestsReader.withTotalAwareBucketFilter(totalAwareBucketFilter);
         this.totalAwareBucketFilter = totalAwareBucketFilter;
         return this;
     }