feat(global-index): align row filtering and BTree Bloom support

docs/source/user_guide/global_index.rst

@@ -60,6 +60,12 @@ BTree is an efficient index based on multi-level SST files, supporting rich pred
 
 **Special Configuration:**
 
+- **Option**: ``btree-index.bloom-filter.enabled``
+
+  - **Default**: ``false``
+  - **Description**: Enables a Bloom filter for BTree point lookups such as equality and ``IN``
+    predicates. This can avoid reading data blocks for missing keys.
+
 - **Option**: ``btree-index.read-buffer-size``
 
   - **Description**: Optional. Specifies the read buffer size for the B-tree index. This setting can be tuned based on query patterns:

include/paimon/global_index/global_index_io_meta.h

@@ -24,16 +24,22 @@
 #include <string>
 
 #include "paimon/memory/bytes.h"
+#include "paimon/visibility.h"
 
 namespace paimon {
 /// Metadata describing a single file entry in a global index.
 struct PAIMON_EXPORT GlobalIndexIOMeta {
-    GlobalIndexIOMeta(const std::string& _file_path, int64_t _file_size,
+    GlobalIndexIOMeta(const std::string& _file_path, int64_t _file_size, int64_t _row_count,
                       const std::shared_ptr<Bytes>& _metadata)
-        : file_path(_file_path), file_size(_file_size), metadata(_metadata) {}
+        : file_path(_file_path),
+          file_size(_file_size),
+          row_count(_row_count),
+          metadata(_metadata) {}
 
     std::string file_path;
     int64_t file_size;
+    /// Total logical rows processed by the writer, including null or skipped rows.
+    int64_t row_count;
     /// Optional binary metadata associated with the file, such as serialized
     /// secondary index structures or inline index bytes.
     /// May be null if no additional metadata is available.

include/paimon/global_index/global_index_scan.h

@@ -88,8 +88,9 @@ class PAIMON_EXPORT GlobalIndexScan {
 
     /// Creates several `GlobalIndexReader`s for a specific field.
     /// @param field_name       Name of the indexed column.
-    /// @param row_range_index  Optional row range that limits the scan to a sub-range of row ids.
-    ///                         If not provided, the entire row range is considered.
+    /// @param row_range_index  Optional row ranges used to skip non-intersecting index shards and
+    ///                         provide pruning hints. Returned candidates are not guaranteed to be
+    ///                         clipped; callers must intersect them with the requested ranges.
     /// @return A `Result` that is:
     ///         - Successful with several readers(with global row id) if the indexes exist and load
     ///         correctly;
@@ -103,8 +104,9 @@ class PAIMON_EXPORT GlobalIndexScan {
     /// Creates a `GlobalIndexReader` for a specific field with a specific index type.
     /// @param field_name       Name of the indexed column.
     /// @param index_type       Name of index type.
-    /// @param row_range_index  Optional row range that limits the scan to a sub-range of row ids.
-    ///                         If not provided, the entire row range is considered.
+    /// @param row_range_index  Optional row ranges used to skip non-intersecting index shards and
+    ///                         provide pruning hints. Returned candidates are not guaranteed to be
+    ///                         clipped; callers must intersect them with the requested ranges.
     /// @return A `Result` that is:
     ///         - Successful with a reader(with global row id) if the index exists and loads
     ///         correctly;
@@ -118,8 +120,9 @@ class PAIMON_EXPORT GlobalIndexScan {
 
     /// Creates several `GlobalIndexReader`s for a specific field (looked up by id),
     /// @param field_id         Field id of the indexed column.
-    /// @param row_range_index  Optional row range that limits the scan to a sub-range of row ids.
-    ///                         If not provided, the entire row range is considered.
+    /// @param row_range_index  Optional row ranges used to skip non-intersecting index shards and
+    ///                         provide pruning hints. Returned candidates are not guaranteed to be
+    ///                         clipped; callers must intersect them with the requested ranges.
     /// @return A `Result` that is:
     ///         - Successful with several readers(with global row id) if the indexes exist and load
     ///         correctly;

include/paimon/global_index/global_indexer.h

@@ -19,6 +19,7 @@
 
 #pragma once
 
+#include <cstdint>
 #include <memory>
 #include <optional>
 #include <string>
@@ -30,6 +31,7 @@
 #include "paimon/global_index/io/global_index_file_reader.h"
 #include "paimon/global_index/io/global_index_file_writer.h"
 #include "paimon/memory/memory_pool.h"
+#include "paimon/utils/range.h"
 #include "paimon/visibility.h"
 
 struct ArrowSchema;
@@ -58,17 +60,23 @@ class PAIMON_EXPORT GlobalIndexer {
         const std::shared_ptr<GlobalIndexFileWriter>& file_writer,
         const std::shared_ptr<MemoryPool>& pool) const = 0;
 
-    /// Creates a reader for querying a pre-built global index.
+    /// Creates a reader for querying a pre-built global index whose relative row IDs cover
+    /// [0, total_row_count).
     ///
-    /// @param arrow_schema   Schema of the indexed data; used to interpret predicate literals.
-    /// @param file_reader    I/O handler for reading index artifacts from storage.
-    /// @param files          List of index file metadata entries produced during writing.
-    /// @param pool           Memory pool for temporary allocations; if nullptr, uses default.
+    /// @param arrow_schema    Schema of the indexed data; used to interpret predicate literals.
+    /// @param file_reader     I/O handler for reading index artifacts from storage.
+    /// @param files           List of index file metadata entries produced during writing.
+    /// @param total_row_count Complete row count for evaluating negative predicates by complement.
+    /// @param row_ranges      Optional local row ranges used only as a pruning hint.
+    ///                        Implementations may ignore them; callers must still clip returned
+    ///                        candidates to the requested ranges.
+    /// @param pool            Memory pool for temporary allocations; if nullptr, uses default.
     /// @return A `Result` containing a shared pointer to the created `GlobalIndexReader`,
     ///         or an error if the index cannot be loaded or is incompatible, etc.
     virtual Result<std::shared_ptr<GlobalIndexReader>> CreateReader(
         ::ArrowSchema* arrow_schema, const std::shared_ptr<GlobalIndexFileReader>& file_reader,
-        const std::vector<GlobalIndexIOMeta>& files,
+        const std::vector<GlobalIndexIOMeta>& files, int64_t total_row_count,
+        const std::optional<std::vector<Range>>& row_ranges,
         const std::shared_ptr<MemoryPool>& pool) const = 0;
 
     /// Whether this indexer supports checkpointing an index build.

src/paimon/common/global_index/bitmap/bitmap_global_index_writer.cpp

@@ -143,8 +143,8 @@ Result<std::vector<GlobalIndexIOMeta>> BitmapGlobalIndexWriter::Finish() {
     SortedIndexFileMeta index_meta(first_key_bytes, last_key_bytes, !null_rows_.IsEmpty());
     std::shared_ptr<Bytes> metadata = index_meta.Serialize(pool_.get());
     PAIMON_ASSIGN_OR_RAISE(int64_t file_size, file_writer_->GetFileSize(file_name_));
-    return std::vector<GlobalIndexIOMeta>{
-        GlobalIndexIOMeta(file_writer_->ToPath(file_name_), file_size, std::move(metadata))};
+    return std::vector<GlobalIndexIOMeta>{GlobalIndexIOMeta(
+        file_writer_->ToPath(file_name_), file_size, row_count_, std::move(metadata))};
 }
 
 Status BitmapGlobalIndexWriter::FlushCurrentBitmap() {

src/paimon/common/global_index/bitmap/bitmap_global_indexer.cpp

@@ -98,7 +98,9 @@ Result<std::shared_ptr<GlobalIndexWriter>> BitmapGlobalIndexer::CreateWriter(
 
 Result<std::shared_ptr<GlobalIndexReader>> BitmapGlobalIndexer::CreateReader(
     ::ArrowSchema* arrow_schema, const std::shared_ptr<GlobalIndexFileReader>& file_reader,
-    const std::vector<GlobalIndexIOMeta>& files, const std::shared_ptr<MemoryPool>& pool) const {
+    const std::vector<GlobalIndexIOMeta>& files, int64_t total_row_count,
+    const std::optional<std::vector<Range>>& /*row_ranges*/,
+    const std::shared_ptr<MemoryPool>& pool) const {
     PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Schema> schema,
                                       arrow::ImportSchema(arrow_schema));
     if (schema->num_fields() != 1) {
@@ -113,7 +115,7 @@ Result<std::shared_ptr<GlobalIndexReader>> BitmapGlobalIndexer::CreateReader(
     PAIMON_ASSIGN_OR_RAISE(
         std::shared_ptr<LazyFilteredBitmapReader> reader,
         LazyFilteredBitmapReader::Create(file_reader, files, schema->field(0)->type(),
-                                         fallback_scan_max_size_, pool, executor));
+                                         fallback_scan_max_size_, total_row_count, pool, executor));
     return std::shared_ptr<GlobalIndexReader>(std::move(reader));
 }
 

src/paimon/common/global_index/bitmap/bitmap_global_indexer.h

@@ -45,7 +45,8 @@ class BitmapGlobalIndexer : public GlobalIndexer {
 
     Result<std::shared_ptr<GlobalIndexReader>> CreateReader(
         ::ArrowSchema* arrow_schema, const std::shared_ptr<GlobalIndexFileReader>& file_reader,
-        const std::vector<GlobalIndexIOMeta>& files,
+        const std::vector<GlobalIndexIOMeta>& files, int64_t total_row_count,
+        const std::optional<std::vector<Range>>& row_ranges,
         const std::shared_ptr<MemoryPool>& pool) const override;
 
  private:

src/paimon/common/global_index/bitmap/lazy_filtered_bitmap_reader.cpp

@@ -29,28 +29,33 @@ namespace paimon {
 Result<std::shared_ptr<LazyFilteredBitmapReader>> LazyFilteredBitmapReader::Create(
     const std::shared_ptr<GlobalIndexFileReader>& file_reader,
     const std::vector<GlobalIndexIOMeta>& files, const std::shared_ptr<arrow::DataType>& key_type,
-    int64_t fallback_scan_max_size, const std::shared_ptr<MemoryPool>& pool,
-    const std::shared_ptr<Executor>& executor) {
+    int64_t fallback_scan_max_size, int64_t total_row_count,
+    const std::shared_ptr<MemoryPool>& pool, const std::shared_ptr<Executor>& executor) {
     if (file_reader == nullptr || key_type == nullptr || pool == nullptr) {
         return Status::Invalid(
             "Cannot create LazyFilteredBitmapReader without file reader, key type, and memory "
             "pool.");
     }
+    if (total_row_count < 0) {
+        return Status::Invalid("Total row count must be non-negative.");
+    }
     PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<KeySerializer> key_serializer,
                            KeySerializer::Create(key_type, pool));
     PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<SortedFileMetaSelector> file_selector,
                            SortedFileMetaSelector::Create(files, key_serializer));
     return std::shared_ptr<LazyFilteredBitmapReader>(
         new LazyFilteredBitmapReader(file_reader, std::move(file_selector), key_serializer,
-                                     fallback_scan_max_size, pool, executor));
+                                     fallback_scan_max_size, total_row_count, pool, executor));
 }
 
 LazyFilteredBitmapReader::LazyFilteredBitmapReader(
     const std::shared_ptr<GlobalIndexFileReader>& file_reader,
     std::unique_ptr<SortedFileMetaSelector> file_selector,
     const std::shared_ptr<KeySerializer>& key_serializer, int64_t fallback_scan_max_size,
-    const std::shared_ptr<MemoryPool>& pool, const std::shared_ptr<Executor>& executor)
-    : SortedFileGlobalIndexReader(std::move(file_selector), fallback_scan_max_size, executor),
+    int64_t total_row_count, const std::shared_ptr<MemoryPool>& pool,
+    const std::shared_ptr<Executor>& executor)
+    : SortedFileGlobalIndexReader(std::move(file_selector), fallback_scan_max_size, total_row_count,
+                                  executor),
       file_reader_(file_reader),
       key_serializer_(key_serializer),
       pool_(pool) {}

src/paimon/common/global_index/bitmap/lazy_filtered_bitmap_reader.h

@@ -39,7 +39,8 @@ class LazyFilteredBitmapReader : public SortedFileGlobalIndexReader {
         const std::shared_ptr<GlobalIndexFileReader>& file_reader,
         const std::vector<GlobalIndexIOMeta>& files,
         const std::shared_ptr<arrow::DataType>& key_type, int64_t fallback_scan_max_size,
-        const std::shared_ptr<MemoryPool>& pool, const std::shared_ptr<Executor>& executor);
+        int64_t total_row_count, const std::shared_ptr<MemoryPool>& pool,
+        const std::shared_ptr<Executor>& executor);
 
     Result<std::shared_ptr<ScoredGlobalIndexResult>> VisitVectorSearch(
         const std::shared_ptr<VectorSearch>& vector_search) override;
@@ -54,7 +55,7 @@ class LazyFilteredBitmapReader : public SortedFileGlobalIndexReader {
     LazyFilteredBitmapReader(const std::shared_ptr<GlobalIndexFileReader>& file_reader,
                              std::unique_ptr<SortedFileMetaSelector> file_selector,
                              const std::shared_ptr<KeySerializer>& key_serializer,
-                             int64_t fallback_scan_max_size,
+                             int64_t fallback_scan_max_size, int64_t total_row_count,
                              const std::shared_ptr<MemoryPool>& pool,
                              const std::shared_ptr<Executor>& executor);
 

src/paimon/common/global_index/bitmap/lazy_filtered_bitmap_reader_test.cpp

@@ -100,11 +100,15 @@ class LazyFilteredBitmapReaderTest : public ::testing::Test {
     std::shared_ptr<LazyFilteredBitmapReader> CreateReader(
         const std::vector<GlobalIndexIOMeta>& metas, int64_t fallback_scan_max_size,
         const std::shared_ptr<GlobalIndexFileReader>& file_reader = nullptr) {
-        EXPECT_OK_AND_ASSIGN(
-            std::shared_ptr<LazyFilteredBitmapReader> reader,
-            LazyFilteredBitmapReader::Create(file_reader == nullptr ? file_manager_ : file_reader,
-                                             metas, arrow::utf8(), fallback_scan_max_size, pool_,
-                                             /*executor=*/nullptr));
+        int64_t total_row_count = 0;
+        for (const GlobalIndexIOMeta& meta : metas) {
+            total_row_count += meta.row_count;
+        }
+        EXPECT_OK_AND_ASSIGN(std::shared_ptr<LazyFilteredBitmapReader> reader,
+                             LazyFilteredBitmapReader::Create(
+                                 file_reader == nullptr ? file_manager_ : file_reader, metas,
+                                 arrow::utf8(), fallback_scan_max_size, total_row_count, pool_,
+                                 /*executor=*/nullptr));
         return reader;
     }
 

src/paimon/common/global_index/btree/btree_compatibility_test.cpp

@@ -112,13 +112,13 @@ class BTreeCompatibilityTest : public ::testing::Test {
 
     Result<std::shared_ptr<GlobalIndexReader>> CreateReaderFromFiles(
         const std::string& bin_path, const std::string& meta_path,
-        const std::shared_ptr<arrow::DataType>& arrow_type) const {
+        const std::shared_ptr<arrow::DataType>& arrow_type, int64_t total_row_count) const {
         auto meta_str = ReadFileAsString(meta_path);
         std::shared_ptr<Bytes> meta_bytes = Bytes::AllocateBytes(meta_str, pool_.get());
         PAIMON_ASSIGN_OR_RAISE(FileStatus file_status, fs_->GetFileStatus(bin_path));
         auto file_size = file_status.GetLen();
 
-        GlobalIndexIOMeta io_meta(bin_path, file_size, meta_bytes);
+        GlobalIndexIOMeta io_meta(bin_path, file_size, total_row_count, meta_bytes);
         std::vector<GlobalIndexIOMeta> metas = {io_meta};
 
         auto schema = arrow::schema({arrow::field("testField", arrow_type)});
@@ -128,7 +128,8 @@ class BTreeCompatibilityTest : public ::testing::Test {
         auto file_reader = std::make_shared<LocalGlobalIndexFileReader>(fs_);
         std::map<std::string, std::string> options;
         PAIMON_ASSIGN_OR_RAISE(auto indexer, BTreeGlobalIndexer::Create(options));
-        return indexer->CreateReader(c_schema.get(), file_reader, metas, pool_);
+        return indexer->CreateReader(c_schema.get(), file_reader, metas, total_row_count,
+                                     /*row_ranges=*/std::nullopt, pool_);
     }
 
     void RunIntQueries(const std::shared_ptr<GlobalIndexReader>& reader,
@@ -569,8 +570,8 @@ TEST_F(BTreeCompatibilityTest, ReadAndQueryIntData) {
         auto records = ParseCsvFile(csv_path);
         ASSERT_EQ(static_cast<int32_t>(records.size()), count);
 
-        ASSERT_OK_AND_ASSIGN(auto reader,
-                             CreateReaderFromFiles(bin_path, meta_path, arrow::int32()));
+        ASSERT_OK_AND_ASSIGN(auto reader, CreateReaderFromFiles(bin_path, meta_path, arrow::int32(),
+                                                                records.size()));
         RunIntQueries(reader, records);
     }
 }
@@ -585,8 +586,8 @@ TEST_F(BTreeCompatibilityTest, ReadAndQueryStringData) {
         auto records = ParseCsvFile(csv_path);
         ASSERT_EQ(static_cast<int32_t>(records.size()), count);
 
-        ASSERT_OK_AND_ASSIGN(auto reader,
-                             CreateReaderFromFiles(bin_path, meta_path, arrow::utf8()));
+        ASSERT_OK_AND_ASSIGN(
+            auto reader, CreateReaderFromFiles(bin_path, meta_path, arrow::utf8(), records.size()));
         RunStringQueries(reader, records);
     }
 }
@@ -601,7 +602,8 @@ TEST_F(BTreeCompatibilityTest, ReadAndQueryFloatData) {
     auto records = ParseCsvFile(csv_path);
     ASSERT_EQ(static_cast<int32_t>(records.size()), count);
 
-    ASSERT_OK_AND_ASSIGN(auto reader, CreateReaderFromFiles(bin_path, meta_path, arrow::float32()));
+    ASSERT_OK_AND_ASSIGN(
+        auto reader, CreateReaderFromFiles(bin_path, meta_path, arrow::float32(), records.size()));
     RunFloatQueries(reader, records);
 }
 
@@ -615,7 +617,8 @@ TEST_F(BTreeCompatibilityTest, AllNulls) {
     ASSERT_FALSE(records.empty());
     auto count = static_cast<int32_t>(records.size());
 
-    ASSERT_OK_AND_ASSIGN(auto reader, CreateReaderFromFiles(bin_path, meta_path, arrow::int32()));
+    ASSERT_OK_AND_ASSIGN(
+        auto reader, CreateReaderFromFiles(bin_path, meta_path, arrow::int32(), records.size()));
 
     // All rows should be null
     {
@@ -653,7 +656,8 @@ TEST_F(BTreeCompatibilityTest, NoNulls) {
     ASSERT_FALSE(records.empty());
     auto count = static_cast<int32_t>(records.size());
 
-    ASSERT_OK_AND_ASSIGN(auto reader, CreateReaderFromFiles(bin_path, meta_path, arrow::int32()));
+    ASSERT_OK_AND_ASSIGN(
+        auto reader, CreateReaderFromFiles(bin_path, meta_path, arrow::int32(), records.size()));
 
     // No rows should be null
     {
@@ -720,7 +724,8 @@ TEST_F(BTreeCompatibilityTest, DuplicateKeys) {
     auto records = ParseCsvFile(csv_path);
     ASSERT_FALSE(records.empty());
 
-    ASSERT_OK_AND_ASSIGN(auto reader, CreateReaderFromFiles(bin_path, meta_path, arrow::int32()));
+    ASSERT_OK_AND_ASSIGN(
+        auto reader, CreateReaderFromFiles(bin_path, meta_path, arrow::int32(), records.size()));
 
     // VisitIsNull
     {
@@ -914,7 +919,8 @@ TEST_F(BTreeCompatibilityTest, RowCountConsistency) {
         ASSERT_FALSE(records.empty());
         auto count = static_cast<int32_t>(records.size());
 
-        ASSERT_OK_AND_ASSIGN(auto reader, CreateReaderFromFiles(bin_path, meta_path, arrow_type));
+        ASSERT_OK_AND_ASSIGN(
+            auto reader, CreateReaderFromFiles(bin_path, meta_path, arrow_type, records.size()));
 
         ASSERT_OK_AND_ASSIGN(auto null_result, reader->VisitIsNull());
         auto null_ids = CollectRowIds(null_result);

src/paimon/common/global_index/btree/btree_defs.h

@@ -36,6 +36,9 @@ struct BtreeDefs {
     /// "btree-index.block-size" - The block size to use for BTreeIndex.
     /// Default value is 64 KB.
     static inline const char kBtreeIndexBlockSize[] = "btree-index.block-size";
+    /// "btree-index.bloom-filter.enabled" - Whether to enable the Bloom filter for BTree index
+    /// point lookups. Default value is false.
+    static inline const char kBtreeIndexBloomFilterEnabled[] = "btree-index.bloom-filter.enabled";
     /// "btree-index.cache-size" - The cache size to use for BTreeIndex.
     /// Default value is 128 MB.
     static inline const char kBtreeIndexCacheSize[] = "btree-index.cache-size";
@@ -66,5 +69,6 @@ struct BtreeDefs {
     static inline const char kDefaultBtreeIndexFallbackScanMaxSize[] = "256MB";
     static inline const int32_t kDefaultBtreeIndexCompressionLevel = 1;
     static inline const double kDefaultBtreeIndexHighPriorityPoolRatio = 0.1;
+    static inline const bool kDefaultBtreeIndexBloomFilterEnabled = false;
 };
 }  // namespace paimon