fix(file-index): address BSI and Bloom writer review issues

Author: zjw1111

src/paimon/common/file_index/bloomfilter/bloom_filter_file_index.cpp

@@ -119,7 +119,7 @@ Status BloomFilterFileIndexWriter::AddBatch(::ArrowArray* batch) {
     }
     PAIMON_ASSIGN_OR_RAISE(
         std::vector<Literal> values,
-        LiteralConverter::ConvertLiteralsFromArray(*struct_array->field(0), /*own_data=*/true));
+        LiteralConverter::ConvertLiteralsFromArray(*struct_array->field(0), /*own_data=*/false));
     for (const Literal& value : values) {
         if (!value.IsNull()) {
             filter_.AddHash(hash_function_(value));
@@ -130,7 +130,7 @@ Status BloomFilterFileIndexWriter::AddBatch(::ArrowArray* batch) {
 
 Result<PAIMON_UNIQUE_PTR<Bytes>> BloomFilterFileIndexWriter::SerializedBytes() const {
     constexpr int32_t kHeaderLength = sizeof(int32_t);
-    const int32_t bit_set_length = filter_.GetBitSet().BitSize() / BloomFilter64::BYTE_SIZE;
+    const int32_t bit_set_length = filter_.GetBitSet().ByteLength();
     PAIMON_UNIQUE_PTR<Bytes> bytes =
         Bytes::AllocateBytes(kHeaderLength + bit_set_length, pool_.get());
     const auto num_hash_functions = static_cast<uint32_t>(filter_.GetNumHashFunctions());

src/paimon/common/file_index/bsi/bit_slice_index_bitmap_file_index.cpp

@@ -81,7 +81,10 @@ BitSliceIndexBitmapFileIndexWriter::BitSliceIndexBitmapFileIndexWriter(
     const std::shared_ptr<arrow::Field>& field,
     const BitSliceIndexBitmapFileIndex::ValueMapperType& value_mapper,
     const std::shared_ptr<MemoryPool>& pool)
-    : struct_type_(arrow::struct_({field})), value_mapper_(value_mapper), pool_(pool) {}
+    : struct_type_(arrow::struct_({field})),
+      field_name_(field->name()),
+      value_mapper_(value_mapper),
+      pool_(pool) {}
 
 Status BitSliceIndexBitmapFileIndexWriter::AddBatch(::ArrowArray* batch) {
     PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Array> array,
@@ -102,14 +105,18 @@ Status BitSliceIndexBitmapFileIndexWriter::AddBatch(::ArrowArray* batch) {
     }
     PAIMON_ASSIGN_OR_RAISE(
         std::vector<Literal> literals,
-        LiteralConverter::ConvertLiteralsFromArray(*struct_array->field(0), /*own_data=*/true));
+        LiteralConverter::ConvertLiteralsFromArray(*struct_array->field(0), /*own_data=*/false));
     values_.reserve(values_.size() + literals.size());
     for (const Literal& literal : literals) {
         if (literal.IsNull()) {
             values_.emplace_back(std::nullopt);
             continue;
         }
         PAIMON_ASSIGN_OR_RAISE(int64_t value, value_mapper_(literal));
+        if (value == std::numeric_limits<int64_t>::min()) {
+            return Status::Invalid(
+                fmt::format("bsi index does not support INT64_MIN for field '{}'", field_name_));
+        }
         values_.emplace_back(value);
         if (value < 0) {
             const int64_t absolute_value = SafeAbs(value);

src/paimon/common/file_index/bsi/bit_slice_index_bitmap_file_index.h

@@ -87,6 +87,7 @@ class BitSliceIndexBitmapFileIndexWriter : public FileIndexWriter {
 
  private:
     std::shared_ptr<arrow::DataType> struct_type_;
+    std::string field_name_;
     BitSliceIndexBitmapFileIndex::ValueMapperType value_mapper_;
     std::vector<std::optional<int64_t>> values_;
     int64_t positive_min_ = 0;

src/paimon/common/file_index/bsi/bit_slice_index_bitmap_file_index_test.cpp

@@ -161,7 +161,7 @@ TEST_F(BitSliceIndexBitmapIndexReaderTest, TestMix) {
 
 TEST_F(BitSliceIndexBitmapIndexReaderTest, TestWriterRejectsInt64MinAndUnsupportedType) {
     ASSERT_NOK_WITH_MSG(WriteIndex(arrow::int64(), R"([[-9223372036854775808]])"),
-                        "values should be non-negative");
+                        "bsi index does not support INT64_MIN for field 'f0'");
 
     BitSliceIndexBitmapFileIndex file_index({});
     ASSERT_NOK_WITH_MSG(file_index.CreateWriter(CreateArrowSchema(arrow::boolean()).get(), pool_),

src/paimon/common/utils/bloom_filter64.cpp

@@ -49,7 +49,11 @@ bool BloomFilter64::BitSet::Get(int32_t index) const {
 }
 
 int32_t BloomFilter64::BitSet::BitSize() const {
-    return (bytes_->size() - offset_) * BloomFilter64::BYTE_SIZE;
+    return ByteLength() * BloomFilter64::BYTE_SIZE;
+}
+
+int32_t BloomFilter64::BitSet::ByteLength() const {
+    return static_cast<int32_t>(bytes_->size() - offset_);
 }
 
 void BloomFilter64::BitSet::ToByteArray(int32_t offset, int32_t length, char* bytes) const {
@@ -80,14 +84,21 @@ Result<BloomFilter64> BloomFilter64::Create(int64_t items, double fpp,
         1, static_cast<int32_t>(std::round(static_cast<double>(num_bits) / items * log_two)));
     auto bytes = std::make_shared<Bytes>(num_bits / BYTE_SIZE, pool.get());
     auto bit_set = std::make_unique<BitSet>(bytes, /*offset=*/0);
-    return BloomFilter64(num_hash_functions, std::move(bit_set));
+    return BloomFilter64(num_hash_functions, std::move(bit_set), pool);
 }
 
 BloomFilter64::BloomFilter64(int32_t num_hash_functions, std::unique_ptr<BitSet>&& bit_set)
     : num_bits_(bit_set->BitSize()),
       num_hash_functions_(num_hash_functions),
       bit_set_(std::move(bit_set)) {}
 
+BloomFilter64::BloomFilter64(int32_t num_hash_functions, std::unique_ptr<BitSet>&& bit_set,
+                             const std::shared_ptr<MemoryPool>& pool)
+    : num_bits_(bit_set->BitSize()),
+      num_hash_functions_(num_hash_functions),
+      pool_(pool),
+      bit_set_(std::move(bit_set)) {}
+
 void BloomFilter64::AddHash(int64_t hash64) {
     auto hash1 = static_cast<int32_t>(hash64);
     auto hash2 = static_cast<int32_t>(static_cast<uint64_t>(hash64) >> 32);

src/paimon/common/utils/bloom_filter64.h

@@ -57,6 +57,7 @@ class PAIMON_EXPORT BloomFilter64 {
         void Set(int32_t index);
         bool Get(int32_t index) const;
         int32_t BitSize() const;
+        int32_t ByteLength() const;
         void ToByteArray(int32_t offset, int32_t length, char* bytes) const;
 
      private:
@@ -67,11 +68,15 @@ class PAIMON_EXPORT BloomFilter64 {
         std::shared_ptr<Bytes> bytes_;
     };
 
+ private:
+    BloomFilter64(int32_t num_hash_functions, std::unique_ptr<BitSet>&& bit_set,
+                  const std::shared_ptr<MemoryPool>& pool);
+
     static constexpr int32_t BYTE_SIZE = 8;
 
- private:
     int32_t num_bits_ = -1;
     int32_t num_hash_functions_ = -1;
+    std::shared_ptr<MemoryPool> pool_;
     std::unique_ptr<BitSet> bit_set_;
 };
 }  // namespace paimon

src/paimon/common/utils/bloom_filter64_test.cpp

@@ -82,6 +82,19 @@ TEST(BloomFilter64Test, TestInvalidItemsAndFpp) {
         "bloom filter size exceeds the supported range");
 }
 
+TEST(BloomFilter64Test, TestKeepMemoryPoolAlive) {
+    std::weak_ptr<MemoryPool> weak_pool;
+    {
+        std::shared_ptr<MemoryPool> pool(GetMemoryPool());
+        weak_pool = pool;
+        ASSERT_OK_AND_ASSIGN(BloomFilter64 bloom_filter,
+                             BloomFilter64::Create(/*items=*/100, /*fpp=*/0.1, pool));
+        pool.reset();
+        ASSERT_FALSE(weak_pool.expired());
+    }
+    ASSERT_TRUE(weak_pool.expired());
+}
+
 TEST(BloomFilter64Test, TestCompatibleWithJava) {
     // data: -10, -5, 0, 13, 100, 200, 500
     std::vector<uint8_t> se_bytes = {241, 255, 17, 0, 0, 0, 0, 0, 0, 0, 0, 0};
@@ -97,8 +110,8 @@ TEST(BloomFilter64Test, TestCompatibleWithJava) {
 
     ASSERT_OK_AND_ASSIGN(BloomFilter64 bloom_filter2, BloomFilter64::Create(10, 0.01, pool));
     ASSERT_EQ(7, bloom_filter2.GetNumHashFunctions());
-    ASSERT_EQ(se_bytes.size() * BloomFilter64::BYTE_SIZE, bloom_filter2.num_bits_);
-    ASSERT_EQ(se_bytes.size(), bloom_filter2.GetBitSet().bytes_->size());
+    ASSERT_EQ(se_bytes.size() * 8, bloom_filter2.GetBitSet().BitSize());
+    ASSERT_EQ(se_bytes.size(), bloom_filter2.GetBitSet().ByteLength());
 }
 
 }  // namespace paimon::test