This is an automated email from the ASF dual-hosted git repository. Gabriel39 pushed a commit to branch codex/fix-parquet-row-group-bloom-cache in repository https://gitbox.apache.org/repos/asf/doris.git
commit 33457142f780913633b2f8bdfbc4fad4e1d9cf06 Author: Gabriel <[email protected]> AuthorDate: Sun Jul 12 11:40:03 2026 +0800 [fix](be) Scope Parquet bloom cache to row groups ### What problem does this PR solve? Issue Number: N/A Related PR: N/A Problem Summary: Parquet row-group pruning cached Bloom filters only by leaf column id. After the first row group loaded a column Bloom filter, later row groups reused that filter and could be incorrectly pruned when a matching value existed only outside the first row group. Scope cache entries by both row group and leaf column so absence checks always use the current row group's Bloom filter. Add a parquet-mr fixture with two row groups and real Bloom data plus an enabled/disabled di [...] ### Release note Fix incorrect Parquet Bloom filter pruning that could silently omit rows from later row groups. ### Check List (For Author) - Test: Unit Test - ParquetStatisticsPruningTest.BloomFilterCacheIsScopedToRowGroupAndColumn - Behavior changed: Yes, Parquet Bloom pruning now uses each row group's own filter - Does this need documentation: No --- be/src/format_v2/parquet/parquet_statistics.cpp | 17 ++++---- .../multi_row_group_bloom_filter.parquet | Bin 0 -> 534 bytes .../format_v2/parquet/parquet_statistics_test.cpp | 44 +++++++++++++++++++++ 3 files changed, 54 insertions(+), 7 deletions(-) diff --git a/be/src/format_v2/parquet/parquet_statistics.cpp b/be/src/format_v2/parquet/parquet_statistics.cpp index d0692a80205..41eb8ee553e 100644 --- a/be/src/format_v2/parquet/parquet_statistics.cpp +++ b/be/src/format_v2/parquet/parquet_statistics.cpp @@ -397,30 +397,33 @@ bool bloom_filter_excludes(const ParquetColumnSchema& column_schema, int slot_in } struct RowGroupBloomFilterCache { + using CacheKey = std::pair<int, int>; + ::parquet::BloomFilterReader* bloom_filter_reader = nullptr; - std::map<int, std::unique_ptr<::parquet::BloomFilter>> column_bloom_filters; - std::set<int> loaded_columns; + std::map<CacheKey, std::unique_ptr<::parquet::BloomFilter>> column_bloom_filters; + std::set<CacheKey> loaded_columns; ::parquet::BloomFilter* get(int row_group_idx, int leaf_column_id, ParquetPruningStats* pruning_stats) { if (bloom_filter_reader == nullptr || leaf_column_id < 0) { return nullptr; } - if (loaded_columns.find(leaf_column_id) == loaded_columns.end()) { - loaded_columns.insert(leaf_column_id); + const CacheKey cache_key {row_group_idx, leaf_column_id}; + if (loaded_columns.find(cache_key) == loaded_columns.end()) { + loaded_columns.insert(cache_key); try { std::shared_ptr<::parquet::RowGroupBloomFilterReader> row_group_reader; if (pruning_stats != nullptr) { SCOPED_RAW_TIMER(&pruning_stats->bloom_filter_read_time); row_group_reader = bloom_filter_reader->RowGroup(row_group_idx); if (row_group_reader != nullptr) { - column_bloom_filters[leaf_column_id] = + column_bloom_filters[cache_key] = row_group_reader->GetColumnBloomFilter(leaf_column_id); } } else { row_group_reader = bloom_filter_reader->RowGroup(row_group_idx); if (row_group_reader != nullptr) { - column_bloom_filters[leaf_column_id] = + column_bloom_filters[cache_key] = row_group_reader->GetColumnBloomFilter(leaf_column_id); } } @@ -430,7 +433,7 @@ struct RowGroupBloomFilterCache { return nullptr; } } - auto it = column_bloom_filters.find(leaf_column_id); + auto it = column_bloom_filters.find(cache_key); return it == column_bloom_filters.end() ? nullptr : it->second.get(); } }; diff --git a/be/test/exec/test_data/parquet_scanner/multi_row_group_bloom_filter.parquet b/be/test/exec/test_data/parquet_scanner/multi_row_group_bloom_filter.parquet new file mode 100644 index 00000000000..6b9c7784258 Binary files /dev/null and b/be/test/exec/test_data/parquet_scanner/multi_row_group_bloom_filter.parquet differ diff --git a/be/test/format_v2/parquet/parquet_statistics_test.cpp b/be/test/format_v2/parquet/parquet_statistics_test.cpp index e5cdd8a8745..404b50c7c70 100644 --- a/be/test/format_v2/parquet/parquet_statistics_test.cpp +++ b/be/test/format_v2/parquet/parquet_statistics_test.cpp @@ -24,6 +24,7 @@ #include <parquet/api/reader.h> #include <parquet/arrow/writer.h> #include <parquet/bloom_filter.h> +#include <parquet/bloom_filter_reader.h> #include <cstdint> #include <cstring> @@ -270,6 +271,14 @@ VExprContextSPtrs bloom_conjuncts(DataTypePtr data_type, std::vector<Field> valu std::make_shared<BloomInExpr>(0, std::move(data_type), std::move(values)))}; } +format::FileScanRequest request_with_bloom_conjunct(DataTypePtr data_type, + std::vector<Field> values) { + format::FileScanRequest request; + request.local_positions.emplace(format::LocalColumnId(0), format::LocalIndex(0)); + request.conjuncts = bloom_conjuncts(std::move(data_type), std::move(values)); + return request; +} + void add_bloom_field(segment_v2::BlockSplitBloomFilter* bloom_filter, const Field& value, PrimitiveType type) { DORIS_CHECK(bloom_filter != nullptr); @@ -522,6 +531,41 @@ TEST(ParquetStatisticsPruningTest, VExprUsesDictionaryAndMissingBloomKeepsRows) EXPECT_EQ(pruning_stats.filtered_row_groups_by_bloom_filter, 0); } +TEST(ParquetStatisticsPruningTest, BloomFilterCacheIsScopedToRowGroupAndColumn) { + auto input = arrow::io::ReadableFile::Open( + "./be/test/exec/test_data/parquet_scanner/multi_row_group_bloom_filter.parquet"); + ASSERT_TRUE(input.ok()); + auto reader = ::parquet::ParquetFileReader::Open(*input); + ASSERT_EQ(reader->metadata()->num_row_groups(), 2); + auto& bloom_filter_reader = reader->GetBloomFilterReader(); + for (int row_group_idx = 0; row_group_idx < 2; ++row_group_idx) { + auto row_group_reader = bloom_filter_reader.RowGroup(row_group_idx); + ASSERT_NE(row_group_reader, nullptr); + ASSERT_NE(row_group_reader->GetColumnBloomFilter(0), nullptr); + } + auto schema = build_file_schema(*reader); + + std::vector<int> selected; + format::parquet::ParquetPruningStats pruning_stats; + auto request = request_with_bloom_conjunct(std::make_shared<DataTypeInt32>(), + {Field::create_field<TYPE_INT>(12345)}); + ASSERT_TRUE(format::parquet::select_row_groups_by_metadata(*reader->metadata(), reader.get(), + schema, request, nullptr, &selected, + false, &pruning_stats) + .ok()); + EXPECT_EQ(selected, std::vector<int>({0, 1})); + EXPECT_EQ(pruning_stats.filtered_row_groups_by_bloom_filter, 0); + + selected.clear(); + pruning_stats = {}; + ASSERT_TRUE(format::parquet::select_row_groups_by_metadata(*reader->metadata(), reader.get(), + schema, request, nullptr, &selected, + true, &pruning_stats) + .ok()); + EXPECT_EQ(selected, std::vector<int>({1})); + EXPECT_EQ(pruning_stats.filtered_row_groups_by_bloom_filter, 1); +} + TEST(ParquetBloomFilterPruningTest, VExprEqPrunesAbsentIntValue) { auto data_type = std::make_shared<DataTypeInt32>(); auto bloom_filter = bloom_filter_for_fields( --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
