fix(scan): preserve decimal and NaN bucket matches

Author: wangyong.alen

docs/source/api/scan.rst

@@ -32,6 +32,14 @@ all bucket keys with equality, and bucket-unaware tables, keep the existing scan
 behavior. Inferred pruning applies only to files matching the scan schema and
 bucket count; older layouts retain the existing filtering behavior.
 
+This inference prunes data files at the manifest-entry level. It does not enable
+manifest min/max-bucket skipping or the bucket-specific live-manifest-entry cache,
+which require an explicit bucket filter.
+
+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
 =========
 

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"
@@ -79,7 +81,21 @@ 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::optional<int32_t>(std::nullopt);
+        }
+        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::optional<int32_t>(std::nullopt);
+            }
+            literal = std::move(scaled);
+        }
         PAIMON_RETURN_NOT_OK(
             WriteLiteralToRow(i, literal, bucket_key_types[i], bucket_key_arrow_types[i], &writer));
     }

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>
@@ -246,6 +247,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_test.cpp

@@ -20,6 +20,7 @@
 
 #include <algorithm>
 #include <cstdint>
+#include <cstring>
 #include <map>
 #include <optional>
 #include <string>
@@ -61,7 +62,7 @@ class AppendBucketPruningTest : public testing::Test {
         const std::shared_ptr<Predicate>& predicate,
         const std::optional<int32_t>& bucket = std::nullopt) const {
         auto arrow_schema = DataField::ConvertDataFieldsToArrowSchema(
-            {DataField(0, arrow::field("rowkey", arrow::utf8())),
+            {DataField(0, arrow::field("rowkey", rowkey_type_)),
              DataField(1, arrow::field("value", arrow::int32()))});
         PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<TableSchema> schema,
                                TableSchema::Create(schema_id_, arrow_schema, {}, {}, options_));
@@ -101,6 +102,33 @@ class AppendBucketPruningTest : public testing::Test {
                                        Literal(FieldType::STRING, "key", 3));
     }
 
+    template <typename T>
+    void CheckMatchingValue(FieldType field_type, const T& query_value, const T& stored_value) {
+        options_[Options::BUCKET] = "17";
+        auto predicate = PredicateBuilder::Equal(0, "rowkey", field_type, Literal(query_value));
+        ASSERT_OK_AND_ASSIGN(auto comparison,
+                             Literal(query_value).CompareTo(Literal(stored_value)));
+        ASSERT_EQ(comparison, 0);
+        BinaryRow stored_key = BinaryRowGenerator::GenerateRow({stored_value}, pool_.get());
+        BinaryRow query_key = BinaryRowGenerator::GenerateRow({query_value}, pool_.get());
+        int32_t bucket = DefaultBucketFunction().Bucket(stored_key, 17);
+        ASSERT_NE(bucket, DefaultBucketFunction().Bucket(query_key, 17));
+        SimpleStats stats = BinaryRowGenerator::GenerateStats(
+            {stored_value, 0}, {stored_value, 100}, {0, 0}, pool_.get());
+        ASSERT_OK_AND_ASSIGN(
+            auto file,
+            DataFileMeta::ForAppend("data.parquet", 100, 10, stats, 0, 9, 0, std::nullopt,
+                                    std::nullopt, std::nullopt, std::nullopt, std::nullopt));
+        ManifestEntry entry(FileKind::Add(), BinaryRow::EmptyRow(), bucket, 17, file);
+        ASSERT_OK_AND_ASSIGN(auto stats_scan, CreateScan(predicate, bucket));
+        ASSERT_OK_AND_ASSIGN(bool stats_match, stats_scan->FilterByStats(entry));
+        ASSERT_TRUE(stats_match);
+        ASSERT_OK_AND_ASSIGN(auto scan, CreateScan(predicate));
+        ASSERT_OK_AND_ASSIGN(bool keep, scan->FilterManifestEntry(entry));
+        ASSERT_TRUE(keep);
+    }
+
+    std::shared_ptr<arrow::DataType> rowkey_type_ = arrow::utf8();
     static constexpr int32_t kNumBuckets = 4;
     int64_t schema_id_ = 0;
     std::shared_ptr<SchemaManager> schema_manager_;
@@ -149,6 +177,23 @@ TEST_F(AppendBucketPruningTest, DoesNotPruneDifferentSchema) {
     CheckBuckets(KeyEquals(), std::nullopt);
 }
 
+TEST_F(AppendBucketPruningTest, PreservesCrossScaleDecimalMatch) {
+    rowkey_type_ = arrow::decimal128(10, 2);
+    CheckMatchingValue(FieldType::DECIMAL, Decimal::FromUnscaledLong(12, 10, 1),
+                       Decimal::FromUnscaledLong(120, 10, 2));
+}
+
+TEST_F(AppendBucketPruningTest, PreservesDifferentNaNPayloadMatch) {
+    rowkey_type_ = arrow::float64();
+    uint64_t query_bits = 0x7ff8000000000000ULL;
+    uint64_t stored_bits = 0x7ff8000000000001ULL;
+    double query_value;
+    double stored_value;
+    std::memcpy(&query_value, &query_bits, sizeof(query_value));
+    std::memcpy(&stored_value, &stored_bits, sizeof(stored_value));
+    CheckMatchingValue(FieldType::DOUBLE, query_value, stored_value);
+}
+
 TEST(AppendOnlyFileStoreScanTest, TestReconstructPredicateWithNonCastedFields) {
     std::string table_root =
         paimon::test::GetDataDir() +