fix(scan): share bucket pruning across append and key-value scans

docs/source/api/scan.rst

@@ -21,6 +21,32 @@ Scan
 
 .. _cpp-api-scan:
 
+Bucket pruning
+==============
+
+For fixed-bucket append and primary-key tables, an equality predicate on every bucket key lets
+the scan derive the target bucket using the table's bucket function. Other buckets
+are excluded from the scan plan without requiring an explicit bucket ID from the
+caller. An explicit bucket filter takes precedence. Queries that do not constrain
+all bucket keys with equality, and bucket-unaware tables, keep the existing scan
+behavior. Both scan types use a shared selector that computes the bucket with
+each manifest entry's total bucket count, so rescaled files are not filtered using
+the current table's bucket count. Historical schemas remain eligible when their ordered bucket-key field IDs,
+types and bucket function match. Changes to unrelated columns do not disable
+pruning. Incompatible bucket schemas and nonpositive total bucket counts retain
+the existing filtering behavior.
+
+This inference prunes data files at the manifest-entry level. Manifest min/max-bucket
+skipping requires an explicit bucket filter. When the snapshot live-manifest-entry
+cache is enabled, inferred scans cache candidates by bucket, current bucket count
+and current schema ID. Files with other bucket counts or schema IDs remain in the
+cached candidates and are filtered after lookup. This permits cache reuse without
+discarding files that require a different bucket calculation or schema fallback.
+
+Decimal literals are rescaled to the bucket field's type only when the conversion
+is exact. NaN literals and decimals that cannot be represented exactly disable
+inferred bucket pruning.
+
 Interface
 =========
 

include/paimon/cache/cache.h

@@ -48,6 +48,16 @@ class PAIMON_EXPORT CacheKey {
                                                                     const std::string& branch,
                                                                     int32_t bucket);
 
+    /// Cache candidates for an inferred bucket, including files with other bucket counts or
+    /// schemas.
+    /// @param total_buckets Positive bucket count used to compute the inferred bucket.
+    /// @param schema_id Schema used to build the selector.
+    static std::shared_ptr<CacheKey> ForSnapshotLiveManifestEntries(const std::string& table_path,
+                                                                    const std::string& branch,
+                                                                    int32_t bucket,
+                                                                    int32_t total_buckets,
+                                                                    int64_t schema_id);
+
  public:
     virtual ~CacheKey() = default;
 

src/paimon/common/io/cache/cache_key.cpp

@@ -24,11 +24,14 @@ namespace {
 class SnapshotLiveManifestEntriesCacheKey : public CacheKey {
  public:
     SnapshotLiveManifestEntriesCacheKey(const std::string& table_path, const std::string& branch,
-                                        int32_t bucket)
+                                        int32_t bucket, int32_t total_buckets = 0,
+                                        int64_t schema_id = 0)
         : CacheKey(CacheKind::SNAPSHOT_LIVE_MANIFEST),
           table_path_(table_path),
           branch_(branch),
-          bucket_(bucket) {}
+          bucket_(bucket),
+          total_buckets_(total_buckets),
+          schema_id_(schema_id) {}
 
     bool IsIndex() const override {
         return false;
@@ -40,7 +43,8 @@ class SnapshotLiveManifestEntriesCacheKey : public CacheKey {
             return false;
         }
         return table_path_ == rhs->table_path_ && branch_ == rhs->branch_ &&
-               bucket_ == rhs->bucket_ && GetKind() == rhs->GetKind();
+               bucket_ == rhs->bucket_ && total_buckets_ == rhs->total_buckets_ &&
+               schema_id_ == rhs->schema_id_ && GetKind() == rhs->GetKind();
     }
 
     size_t HashCode() const override {
@@ -50,6 +54,8 @@ class SnapshotLiveManifestEntriesCacheKey : public CacheKey {
         seed ^= std::hash<int32_t>{}(bucket_) + HASH_CONSTANT + (seed << 6) + (seed >> 2);
         seed ^= std::hash<int32_t>{}(static_cast<int32_t>(GetKind())) + HASH_CONSTANT +
                 (seed << 6) + (seed >> 2);
+        seed ^= std::hash<int32_t>{}(total_buckets_) + HASH_CONSTANT + (seed << 6) + (seed >> 2);
+        seed ^= std::hash<int64_t>{}(schema_id_) + HASH_CONSTANT + (seed << 6) + (seed >> 2);
         return seed;
     }
 
@@ -59,6 +65,8 @@ class SnapshotLiveManifestEntriesCacheKey : public CacheKey {
     const std::string table_path_;
     const std::string branch_;
     const int32_t bucket_;
+    const int32_t total_buckets_;
+    const int64_t schema_id_;
 };
 
 }  // namespace
@@ -82,6 +90,15 @@ std::shared_ptr<CacheKey> CacheKey::ForSnapshotLiveManifestEntries(const std::st
     return std::make_shared<SnapshotLiveManifestEntriesCacheKey>(table_path, branch, bucket);
 }
 
+std::shared_ptr<CacheKey> CacheKey::ForSnapshotLiveManifestEntries(const std::string& table_path,
+                                                                   const std::string& branch,
+                                                                   int32_t bucket,
+                                                                   int32_t total_buckets,
+                                                                   int64_t schema_id) {
+    return std::make_shared<SnapshotLiveManifestEntriesCacheKey>(table_path, branch, bucket,
+                                                                 total_buckets, schema_id);
+}
+
 bool PositionCacheKey::IsIndex() const {
     return is_index_;
 }

src/paimon/common/io/cache/lru_cache_test.cpp

@@ -383,6 +383,16 @@ TEST_F(LruCacheTest, TestForKindSetsKeyKind) {
     ASSERT_EQ(CacheKind::MANIFEST, put_key->GetKind());
 }
 
+TEST_F(LruCacheTest, InferredManifestCacheKeysIncludeBucketCountAndSchema) {
+    auto key = CacheKey::ForSnapshotLiveManifestEntries("table", "main", 1, 4, 0);
+    auto same = CacheKey::ForSnapshotLiveManifestEntries("table", "main", 1, 4, 0);
+    ASSERT_TRUE(key->Equals(*same));
+    ASSERT_EQ(key->HashCode(), same->HashCode());
+    ASSERT_FALSE(key->Equals(*CacheKey::ForSnapshotLiveManifestEntries("table", "main", 1)));
+    ASSERT_FALSE(key->Equals(*CacheKey::ForSnapshotLiveManifestEntries("table", "main", 1, 8, 0)));
+    ASSERT_FALSE(key->Equals(*CacheKey::ForSnapshotLiveManifestEntries("table", "main", 1, 4, 1)));
+}
+
 TEST_F(LruCacheTest, TestForSnapshotLiveManifestEntries) {
     auto main_key = CacheKey::ForSnapshotLiveManifestEntries("table_path", "main", 0);
     auto same_key = CacheKey::ForSnapshotLiveManifestEntries("table_path", "main", 0);

src/paimon/core/bucket/bucket_select_converter.cpp

@@ -19,6 +19,7 @@
 #include "paimon/core/bucket/bucket_select_converter.h"
 
 #include <cassert>
+#include <cmath>
 #include <set>
 #include <string>
 #include <utility>
@@ -34,6 +35,7 @@
 #include "paimon/core/bucket/default_bucket_function.h"
 #include "paimon/core/bucket/hive_bucket_function.h"
 #include "paimon/core/bucket/mod_bucket_function.h"
+#include "paimon/core/casting/decimal_to_decimal_cast_executor.h"
 #include "paimon/data/timestamp.h"
 #include "paimon/memory/memory_pool.h"
 #include "paimon/predicate/leaf_predicate.h"
@@ -46,9 +48,25 @@ Result<std::optional<int32_t>> BucketSelectConverter::Convert(
     const std::shared_ptr<Predicate>& predicate, const std::vector<std::string>& bucket_key_names,
     const std::vector<std::shared_ptr<arrow::DataType>>& bucket_key_arrow_types,
     BucketFunctionType bucket_function_type, int32_t num_buckets, MemoryPool* pool) {
+    if (num_buckets <= 0) {
+        return std::optional<int32_t>();
+    }
+    PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<BucketSelector> selector,
+                           ConvertToSelector(predicate, bucket_key_names, bucket_key_arrow_types,
+                                             bucket_function_type, pool));
+    if (!selector) {
+        return std::optional<int32_t>();
+    }
+    return std::optional<int32_t>(selector->Bucket(num_buckets));
+}
+
+Result<std::unique_ptr<BucketSelector>> BucketSelectConverter::ConvertToSelector(
+    const std::shared_ptr<Predicate>& predicate, const std::vector<std::string>& bucket_key_names,
+    const std::vector<std::shared_ptr<arrow::DataType>>& bucket_key_arrow_types,
+    BucketFunctionType bucket_function_type, MemoryPool* pool) {
     assert(pool);
-    if (!predicate || bucket_key_names.empty() || num_buckets <= 0) {
-        return std::optional<int32_t>(std::nullopt);
+    if (!predicate || bucket_key_names.empty()) {
+        return std::unique_ptr<BucketSelector>();
     }
 
     if (bucket_key_names.size() != bucket_key_arrow_types.size()) {
@@ -66,7 +84,7 @@ Result<std::optional<int32_t>> BucketSelectConverter::Convert(
 
     auto literals_opt = ExtractEqualLiterals(predicate, bucket_key_names);
     if (!literals_opt.has_value()) {
-        return std::optional<int32_t>(std::nullopt);
+        return std::unique_ptr<BucketSelector>();
     }
 
     const auto& literals_map = literals_opt.value();
@@ -79,18 +97,31 @@ Result<std::optional<int32_t>> BucketSelectConverter::Convert(
 
     for (int32_t i = 0; i < num_fields; i++) {
         const auto& field_name = bucket_key_names[i];
-        const auto& literal = literals_map.at(field_name);
+        Literal literal = literals_map.at(field_name);
+        // Equal NaNs can have different stored bit patterns and therefore different buckets.
+        if ((bucket_key_types[i] == FieldType::FLOAT && std::isnan(literal.GetValue<float>())) ||
+            (bucket_key_types[i] == FieldType::DOUBLE && std::isnan(literal.GetValue<double>()))) {
+            return std::unique_ptr<BucketSelector>();
+        }
+        if (bucket_key_types[i] == FieldType::DECIMAL) {
+            PAIMON_ASSIGN_OR_RAISE(Literal scaled, DecimalToDecimalCastExecutor().Cast(
+                                                       literal, bucket_key_arrow_types[i]));
+            // Hash the field's representation only when rescaling preserves the exact value.
+            if (scaled.IsNull() || !(scaled.GetValue<Decimal>() == literal.GetValue<Decimal>())) {
+                return std::unique_ptr<BucketSelector>();
+            }
+            literal = std::move(scaled);
+        }
         PAIMON_RETURN_NOT_OK(
             WriteLiteralToRow(i, literal, bucket_key_types[i], bucket_key_arrow_types[i], &writer));
     }
     writer.Complete();
 
-    // Create the bucket function and compute the bucket
+    // Retain the key and function; the bucket count belongs to each manifest entry.
     PAIMON_ASSIGN_OR_RAISE(
         std::unique_ptr<BucketFunction> bucket_function,
         CreateBucketFunction(bucket_function_type, bucket_key_types, bucket_key_arrow_types));
-    int32_t bucket = bucket_function->Bucket(row, num_buckets);
-    return std::optional<int32_t>(bucket);
+    return std::make_unique<BucketSelector>(std::move(row), std::move(bucket_function));
 }
 
 std::optional<std::map<std::string, Literal>> BucketSelectConverter::ExtractEqualLiterals(

src/paimon/core/bucket/bucket_select_converter.h

@@ -23,10 +23,13 @@
 #include <memory>
 #include <optional>
 #include <string>
+#include <utility>
 #include <vector>
 
 #include "arrow/type_fwd.h"
 #include "paimon/bucket/bucket_function_type.h"
+#include "paimon/common/data/binary_row.h"
+#include "paimon/core/bucket/bucket_function.h"
 #include "paimon/defs.h"
 #include "paimon/predicate/literal.h"
 #include "paimon/result.h"
@@ -38,6 +41,22 @@ class BucketFunction;
 class MemoryPool;
 class Predicate;
 
+/// Selects a bucket using the bucket count recorded in each manifest entry.
+class BucketSelector {
+ public:
+    BucketSelector(BinaryRow key, std::unique_ptr<BucketFunction> function)
+        : key_(std::move(key)), function_(std::move(function)) {}
+
+    /// Compute the bucket for a positive total bucket count.
+    int32_t Bucket(int32_t num_buckets) const {
+        return function_->Bucket(key_, num_buckets);
+    }
+
+ private:
+    BinaryRow key_;
+    std::unique_ptr<BucketFunction> function_;
+};
+
 /// Converts predicates on bucket key fields to a target bucket ID.
 /// When all bucket key fields have EQUAL predicates, the converter computes
 /// which bucket the data must reside in, enabling bucket pruning during scan.
@@ -62,6 +81,14 @@ class BucketSelectConverter {
         const std::vector<std::shared_ptr<arrow::DataType>>& bucket_key_arrow_types,
         BucketFunctionType bucket_function_type, int32_t num_buckets, MemoryPool* pool);
 
+    /// Build a selector once, then evaluate it with each file's bucket count.
+    /// Returns nullptr when the predicate cannot safely constrain all bucket keys.
+    static Result<std::unique_ptr<BucketSelector>> ConvertToSelector(
+        const std::shared_ptr<Predicate>& predicate,
+        const std::vector<std::string>& bucket_key_names,
+        const std::vector<std::shared_ptr<arrow::DataType>>& bucket_key_arrow_types,
+        BucketFunctionType bucket_function_type, MemoryPool* pool);
+
  private:
     /// Extract single literal per bucket key field from EQUAL predicates.
     /// Splits the predicate by AND and looks for EQUAL leaf predicates on bucket key fields.

src/paimon/core/bucket/bucket_select_converter_test.cpp

@@ -18,6 +18,7 @@
 
 #include "paimon/core/bucket/bucket_select_converter.h"
 
+#include <limits>
 #include <optional>
 #include <string>
 #include <vector>
@@ -60,6 +61,30 @@ class BucketSelectConverterTest : public ::testing::Test {
     std::shared_ptr<MemoryPool> pool_ = GetDefaultPool();
 };
 
+TEST_F(BucketSelectConverterTest, SelectorUsesEachBucketCount) {
+    auto pool = GetDefaultPool();
+    auto predicate =
+        PredicateBuilder::Equal(0, "key", FieldType::INT, Literal(static_cast<int32_t>(-23)));
+    BinaryRow row = BinaryRowGenerator::GenerateRow({static_cast<int32_t>(-23)}, pool.get());
+    ASSERT_OK_AND_ASSIGN(auto mod_function, ModBucketFunction::Create(FieldType::INT));
+    ASSERT_OK_AND_ASSIGN(auto hive_function,
+                         HiveBucketFunction::Create({HiveFieldInfo(FieldType::INT)}));
+    DefaultBucketFunction default_function;
+    const std::map<BucketFunctionType, const BucketFunction*> functions = {
+        {BucketFunctionType::DEFAULT, &default_function},
+        {BucketFunctionType::MOD, mod_function.get()},
+        {BucketFunctionType::HIVE, hive_function.get()}};
+    for (const auto& [type, function] : functions) {
+        ASSERT_OK_AND_ASSIGN(auto selector,
+                             BucketSelectConverter::ConvertToSelector(
+                                 predicate, {"key"}, {arrow::int32()}, type, pool.get()));
+        ASSERT_TRUE(selector);
+        for (int32_t total_buckets : {2, 4, 8, 17, 2}) {
+            ASSERT_EQ(selector->Bucket(total_buckets), function->Bucket(row, total_buckets));
+        }
+    }
+}
+
 TEST_F(BucketSelectConverterTest, SingleStringEqualDefault) {
     std::string value = "hello_world";
     AssertDefaultBucket(FieldType::STRING, Literal(FieldType::STRING, value.c_str(), value.size()),
@@ -246,6 +271,55 @@ TEST_F(BucketSelectConverterTest, HiveBucketFunctionWithDecimal) {
     ASSERT_EQ(function->Bucket(row, num_buckets), selected_bucket.value());
 }
 
+TEST_F(BucketSelectConverterTest, RescalesDecimalLiteralsExactly) {
+    for (int32_t precision : {10, 20}) {
+        for (BucketFunctionType function_type :
+             {BucketFunctionType::DEFAULT, BucketFunctionType::HIVE}) {
+            Decimal stored = Decimal::FromUnscaledLong(120, precision, 2);
+            auto stored_predicate =
+                PredicateBuilder::Equal(0, "amount", FieldType::DECIMAL, Literal(stored));
+            ASSERT_OK_AND_ASSIGN(auto expected,
+                                 BucketSelectConverter::Convert(stored_predicate, {"amount"},
+                                                                {arrow::decimal128(precision, 2)},
+                                                                function_type, 17, pool_.get()));
+            ASSERT_TRUE(expected.has_value());
+            for (const auto& query : {Decimal::FromUnscaledLong(12, precision, 1),
+                                      Decimal::FromUnscaledLong(1200, precision, 3)}) {
+                auto predicate =
+                    PredicateBuilder::Equal(0, "amount", FieldType::DECIMAL, Literal(query));
+                ASSERT_OK_AND_ASSIGN(
+                    auto result, BucketSelectConverter::Convert(predicate, {"amount"},
+                                                                {arrow::decimal128(precision, 2)},
+                                                                function_type, 17, pool_.get()));
+                ASSERT_EQ(result, expected);
+            }
+        }
+    }
+}
+
+TEST_F(BucketSelectConverterTest, InexactDecimalConversionReturnsNullopt) {
+    for (const auto& value :
+         {Decimal::FromUnscaledLong(123, 10, 3), Decimal::FromUnscaledLong(9999999999LL, 10, 0)}) {
+        auto predicate = PredicateBuilder::Equal(0, "amount", FieldType::DECIMAL, Literal(value));
+        ASSERT_OK_AND_ASSIGN(auto result, BucketSelectConverter::Convert(
+                                              predicate, {"amount"}, {arrow::decimal128(10, 2)},
+                                              BucketFunctionType::DEFAULT, 17, pool_.get()));
+        ASSERT_FALSE(result.has_value());
+    }
+}
+
+TEST_F(BucketSelectConverterTest, NaNReturnsNullopt) {
+    for (const auto& value : {Literal(std::numeric_limits<float>::quiet_NaN()),
+                              Literal(std::numeric_limits<double>::quiet_NaN())}) {
+        auto type = value.GetType() == FieldType::FLOAT ? arrow::float32() : arrow::float64();
+        auto predicate = PredicateBuilder::Equal(0, "key", value.GetType(), value);
+        ASSERT_OK_AND_ASSIGN(auto result, BucketSelectConverter::Convert(
+                                              predicate, {"key"}, {type},
+                                              BucketFunctionType::DEFAULT, 17, pool_.get()));
+        ASSERT_FALSE(result.has_value());
+    }
+}
+
 TEST_F(BucketSelectConverterTest, UnsupportedFieldTypeReturnsError) {
     auto predicate =
         PredicateBuilder::Equal(0, "items", FieldType::ARRAY, Literal(static_cast<int32_t>(42)));

src/paimon/core/operation/append_only_file_store_scan.cpp

@@ -26,6 +26,7 @@
 #include <set>
 #include <string>
 #include <utility>
+#include <vector>
 
 #include "arrow/type.h"
 #include "fmt/format.h"
@@ -42,6 +43,7 @@
 #include "paimon/core/utils/field_mapping.h"
 #include "paimon/file_index/file_index_result.h"
 #include "paimon/predicate/predicate_utils.h"
+#include "paimon/scan_context.h"
 #include "paimon/status.h"
 
 namespace paimon {

src/paimon/core/operation/append_only_file_store_scan.h

@@ -18,6 +18,7 @@
 
 #pragma once
 
+#include <cstdint>
 #include <memory>
 #include <vector>