feat(scan): prune manifests by bucket metadata

crates/paimon/src/spec/avro/manifest_file_meta_decode.rs

@@ -39,6 +39,7 @@ impl AvroRecordDecode for ManifestFileMeta {
         let mut max_level: Option<i32> = None;
         let mut min_row_id: Option<i64> = None;
         let mut max_row_id: Option<i64> = None;
+        let mut total_buckets: Option<i32> = None;
 
         for field in &writer_schema.fields {
             match field.name.as_str() {
@@ -62,6 +63,7 @@ impl AvroRecordDecode for ManifestFileMeta {
                 "_MAX_LEVEL" => max_level = read_optional_int(cursor, field.nullable)?,
                 "_MIN_ROW_ID" => min_row_id = read_optional_long(cursor, field.nullable)?,
                 "_MAX_ROW_ID" => max_row_id = read_optional_long(cursor, field.nullable)?,
+                "_TOTAL_BUCKETS" => total_buckets = read_optional_int(cursor, field.nullable)?,
                 _ => skip_nullable_field(cursor, &field.schema, field.nullable)?,
             }
         }
@@ -80,6 +82,7 @@ impl AvroRecordDecode for ManifestFileMeta {
             max_level,
             min_row_id,
             max_row_id,
+            total_buckets,
         ))
     }
 }

crates/paimon/src/spec/manifest_file_meta.rs

@@ -103,6 +103,16 @@ pub struct ManifestFileMeta {
         skip_serializing_if = "Option::is_none"
     )]
     max_row_id: Option<i64>,
+
+    /// Common positive bucket count for every entry in this manifest. `None`
+    /// means that the writer could not prove a single usable value, so readers
+    /// must not use it for manifest-level bucket pruning.
+    #[serde(
+        rename = "_TOTAL_BUCKETS",
+        default,
+        skip_serializing_if = "Option::is_none"
+    )]
+    total_buckets: Option<i32>,
 }
 
 impl ManifestFileMeta {
@@ -184,6 +194,12 @@ impl ManifestFileMeta {
         self.max_row_id
     }
 
+    /// Get the common positive bucket count for entries in this manifest.
+    #[inline]
+    pub fn total_buckets(&self) -> Option<i32> {
+        self.total_buckets
+    }
+
     /// Attach bucket / level statistics aggregated from manifest entries.
     ///
     /// Use this in writers that have access to the entries that the manifest covers.
@@ -214,6 +230,15 @@ impl ManifestFileMeta {
         self
     }
 
+    /// Attach a common positive bucket count aggregated from every manifest
+    /// entry. Invalid or mixed values must be represented as `None`.
+    #[inline]
+    #[must_use]
+    pub fn with_total_buckets(mut self, total_buckets: Option<i32>) -> Self {
+        self.total_buckets = total_buckets.filter(|value| *value > 0);
+        self
+    }
+
     #[inline]
     pub fn new(
         file_name: String,
@@ -237,6 +262,7 @@ impl ManifestFileMeta {
             max_level: None,
             min_row_id: None,
             max_row_id: None,
+            total_buckets: None,
         }
     }
 
@@ -256,6 +282,7 @@ impl ManifestFileMeta {
         max_level: Option<i32>,
         min_row_id: Option<i64>,
         max_row_id: Option<i64>,
+        total_buckets: Option<i32>,
     ) -> ManifestFileMeta {
         Self {
             version,
@@ -271,6 +298,7 @@ impl ManifestFileMeta {
             max_level,
             min_row_id,
             max_row_id,
+            total_buckets,
         }
     }
 }
@@ -301,7 +329,8 @@ pub const MANIFEST_FILE_META_SCHEMA: &str = r#"["null", {
         {"name": "_MIN_LEVEL", "type": ["null", "int"], "default": null},
         {"name": "_MAX_LEVEL", "type": ["null", "int"], "default": null},
         {"name": "_MIN_ROW_ID", "type": ["null", "long"], "default": null},
-        {"name": "_MAX_ROW_ID", "type": ["null", "long"], "default": null}
+        {"name": "_MAX_ROW_ID", "type": ["null", "long"], "default": null},
+        {"name": "_TOTAL_BUCKETS", "type": ["null", "int"], "default": null}
     ]
 }]"#;
 

crates/paimon/src/spec/objects_file.rs

@@ -199,6 +199,7 @@ mod tests {
                 "_MAX_LEVEL",
                 "_MIN_ROW_ID",
                 "_MAX_ROW_ID",
+                "_TOTAL_BUCKETS",
             ],
         );
         assert_record_field_order(
@@ -228,12 +229,14 @@ mod tests {
             None,
             Some(100),
             Some(199),
+            Some(8),
         )];
         let bytes = to_avro_bytes(MANIFEST_FILE_META_SCHEMA, &original).unwrap();
         let decoded = from_avro_bytes::<ManifestFileMeta>(&bytes).unwrap();
         assert_eq!(original, decoded);
         assert_eq!(decoded[0].min_row_id(), Some(100));
         assert_eq!(decoded[0].max_row_id(), Some(199));
+        assert_eq!(decoded[0].total_buckets(), Some(8));
     }
 
     #[test]

crates/paimon/src/table/table_commit.rs

@@ -52,6 +52,15 @@ type PartitionBucketKey = (Vec<u8>, i32);
 type RowIdRange = (i64, i64);
 type ExistingRowIdRanges = HashMap<PartitionBucketKey, Vec<RowIdRange>>;
 
+fn common_positive_total_buckets(entries: &[ManifestEntry]) -> Option<i32> {
+    let total_buckets = entries.first()?.total_buckets();
+    (total_buckets > 0
+        && entries
+            .iter()
+            .all(|entry| entry.total_buckets() == total_buckets))
+    .then_some(total_buckets)
+}
+
 fn validate_bucket_ownership(messages: &[CommitMessage]) -> Result<()> {
     let mut owners = HashSet::new();
     for message in messages {
@@ -1228,6 +1237,7 @@ impl TableCommit {
         let mut max_level: Option<i32> = None;
         let mut min_row_id: Option<i64> = None;
         let mut max_row_id: Option<i64> = None;
+        let total_buckets = common_positive_total_buckets(entries);
         let mut all_entries_have_row_id = !entries.is_empty();
         let mut schema_id = self.table.schema().id();
         for entry in entries {
@@ -1265,7 +1275,8 @@ impl TableCommit {
             schema_id,
         )
         .with_bucket_level_stats(min_bucket, max_bucket, min_level, max_level)
-        .with_row_id_stats(min_row_id, max_row_id))
+        .with_row_id_stats(min_row_id, max_row_id)
+        .with_total_buckets(total_buckets))
     }
 
     /// Check if this commit was already completed (idempotency).
@@ -5796,6 +5807,29 @@ mod tests {
         assert!(stats.null_counts().is_empty());
     }
 
+    #[test]
+    fn manifest_total_buckets_requires_one_consistent_positive_value() {
+        let entry = |total_buckets| {
+            ManifestEntry::new(
+                FileKind::Add,
+                vec![],
+                0,
+                total_buckets,
+                test_data_file("data.parquet", 1),
+                2,
+            )
+        };
+
+        assert_eq!(common_positive_total_buckets(&[]), None);
+        assert_eq!(
+            common_positive_total_buckets(&[entry(8), entry(8)]),
+            Some(8)
+        );
+        assert_eq!(common_positive_total_buckets(&[entry(8), entry(4)]), None);
+        assert_eq!(common_positive_total_buckets(&[entry(0)]), None);
+        assert_eq!(common_positive_total_buckets(&[entry(-1)]), None);
+    }
+
     /// Regression: when there are no entries at all, the empty stats we return must also
     /// satisfy the protocol — same Java reader path runs on it.
     #[test]

crates/paimon/src/table/table_scan.rs

@@ -212,6 +212,14 @@ async fn read_all_manifest_entries(
         trace.manifest_files_after_partition_pruning = manifest_files.len();
     }
 
+    retain_manifest_buckets(
+        &mut manifest_files,
+        has_primary_keys && !scan_all_files,
+        bucket_predicate,
+        bucket_key_fields,
+        bucket_function_type,
+    );
+
     if let Some(index) = row_range_index {
         let before = manifest_files.len();
         retain_manifest_row_ranges(&mut manifest_files, index);
@@ -364,6 +372,58 @@ fn retain_manifest_row_ranges(
     manifests.retain(|manifest| manifest_file_overlaps_row_range_index(manifest, row_range_index));
 }
 
+/// Conservatively prune manifests using their bucket envelope and a common
+/// positive `_TOTAL_BUCKETS` value. Missing, invalid, mixed, or legacy metadata
+/// always fails open. Tight envelopes become especially effective when
+/// manifests are bucket-sorted.
+fn retain_manifest_buckets(
+    manifests: &mut Vec<crate::spec::ManifestFileMeta>,
+    only_real_buckets: bool,
+    bucket_predicate: Option<&Predicate>,
+    bucket_key_fields: &[DataField],
+    bucket_function_type: BucketFunctionType,
+) {
+    let mut target_cache: HashMap<i32, Option<HashSet<i32>>> = HashMap::new();
+    manifests.retain(|manifest| {
+        let (Some(min_bucket), Some(max_bucket)) = (manifest.min_bucket(), manifest.max_bucket())
+        else {
+            return true;
+        };
+        if min_bucket > max_bucket {
+            return true;
+        }
+        if only_real_buckets && max_bucket < 0 {
+            return false;
+        }
+
+        let Some(predicate) = bucket_predicate else {
+            return true;
+        };
+        let Some(total_buckets) = manifest.total_buckets().filter(|value| *value > 0) else {
+            return true;
+        };
+        // A mixed range containing the unassigned bucket is not a safe
+        // representation of the real-bucket subset.
+        if min_bucket < 0 || max_bucket >= total_buckets {
+            return true;
+        }
+
+        let targets = target_cache.entry(total_buckets).or_insert_with(|| {
+            compute_target_buckets(
+                predicate,
+                bucket_key_fields,
+                bucket_function_type,
+                total_buckets,
+            )
+        });
+        targets.as_ref().is_none_or(|targets| {
+            targets
+                .iter()
+                .any(|bucket| *bucket >= min_bucket && *bucket <= max_bucket)
+        })
+    });
+}
+
 fn data_file_overlaps_row_range_index(
     file: &DataFileMeta,
     row_range_index: &RowRangeIndex,
@@ -1830,6 +1890,14 @@ impl<'a> PaimonTableScan<'a> {
         };
         let bucket_function_type = core_options.bucket_function_type()?;
 
+        retain_manifest_buckets(
+            &mut manifest_metas,
+            has_primary_keys && !self.scan_all_files,
+            self.bucket_predicate.as_ref(),
+            &bucket_key_fields,
+            bucket_function_type,
+        );
+
         let base_path = format!("{}/{}", table_path.trim_end_matches('/'), MANIFEST_DIR);
         let shared_cache = SharedSchemaCache::new();
         let partition_filter = self.partition_filter.as_ref();
@@ -2374,10 +2442,10 @@ mod tests {
         data_evolution_row_range_groups, data_file_overlaps_row_range_index,
         group_data_files_by_partition_bucket, manifest_file_overlaps_row_range_index,
         prune_data_evolution_group_by_read_fields, retain_index_manifest_entry,
-        retain_index_manifest_entry_for_scan, retain_manifest_entry_row_ranges,
-        retain_manifest_row_ranges, scan_predicate_field_ids, should_skip_level_zero_for_scan,
-        split_row_ranges_for_files, LimitPushdownAccumulator, PaimonTableScan, RowRangeIndex,
-        TableScan,
+        retain_index_manifest_entry_for_scan, retain_manifest_buckets,
+        retain_manifest_entry_row_ranges, retain_manifest_row_ranges, scan_predicate_field_ids,
+        should_skip_level_zero_for_scan, split_row_ranges_for_files, LimitPushdownAccumulator,
+        PaimonTableScan, RowRangeIndex, TableScan,
     };
     use crate::catalog::Identifier;
     use crate::io::FileIOBuilder;
@@ -2480,6 +2548,50 @@ mod tests {
         ));
     }
 
+    #[test]
+    fn test_manifest_bucket_pruning_is_exact_and_fails_open() {
+        let fields = int_field();
+        let predicate = PredicateBuilder::new(&fields)
+            .equal("id", Datum::Int(42))
+            .unwrap();
+        let target = *compute_target_buckets(&predicate, &fields, BucketFunctionType::Default, 8)
+            .unwrap()
+            .iter()
+            .next()
+            .unwrap();
+        let other = (target + 1) % 8;
+        let stats = BinaryTableStats::new(Vec::new(), Vec::new(), Vec::new());
+        let manifest = |name: &str, bucket: i32, total_buckets| {
+            ManifestFileMeta::new(name.to_string(), 1, 1, 0, stats.clone(), 0)
+                .with_bucket_level_stats(Some(bucket), Some(bucket), Some(0), Some(0))
+                .with_total_buckets(total_buckets)
+        };
+        let mut manifests = vec![
+            manifest("target", target, Some(8)),
+            manifest("other", other, Some(8)),
+            manifest("legacy", other, None),
+            manifest("unassigned", -1, Some(8)),
+            manifest("at-total", 8, Some(8)),
+            manifest("above-total", 9, Some(8)),
+        ];
+
+        retain_manifest_buckets(
+            &mut manifests,
+            true,
+            Some(&predicate),
+            &fields,
+            BucketFunctionType::Default,
+        );
+
+        assert_eq!(
+            manifests
+                .iter()
+                .map(ManifestFileMeta::file_name)
+                .collect::<Vec<_>>(),
+            vec!["target", "legacy", "at-total", "above-total"]
+        );
+    }
+
     #[test]
     fn test_manifest_row_range_pruning_uses_each_envelope() {
         let index = RowRangeIndex::create(vec![RowRange::new(2, 2)]);