From ff4f98596fad13a6547ff1c88583fb360f367e34 Mon Sep 17 00:00:00 2001 From: Junwang Zhao Date: Thu, 12 Nov 2026 11:38:18 +0800 Subject: [PATCH] feat(parquet): add Bloom row-group pruning Evaluate EQ and IN predicates with Arrow's Bloom reader after footer statistics, following Java Iceberg's inclusive filtering rules. Keep other predicates, missing columns and files without usable field IDs or Bloom metadata inclusive. Hash literals using the file's physical representation, including numeric promotions, temporal types, decimal precision and width, binary/fixed and UUID values. Cache each column's Bloom filter per row group and load only Bloom payloads during planning. Preserve split selection, file-relative row positions and inherited row IDs after pruning. --- src/iceberg/CMakeLists.txt | 1 + src/iceberg/file_reader.h | 2 +- .../parquet/parquet_bloom_row_group_filter.cc | 317 +++++++ .../parquet_bloom_row_group_filter_internal.h | 54 ++ src/iceberg/parquet/parquet_reader.cc | 13 + src/iceberg/test/CMakeLists.txt | 1 + .../parquet_bloom_row_group_filter_test.cc | 854 ++++++++++++++++++ 7 files changed, 1241 insertions(+), 1 deletion(-) create mode 100644 src/iceberg/parquet/parquet_bloom_row_group_filter.cc create mode 100644 src/iceberg/parquet/parquet_bloom_row_group_filter_internal.h create mode 100644 src/iceberg/test/parquet_bloom_row_group_filter_test.cc diff --git a/src/iceberg/CMakeLists.txt b/src/iceberg/CMakeLists.txt index ad9fd54dbf..b054662574 100644 --- a/src/iceberg/CMakeLists.txt +++ b/src/iceberg/CMakeLists.txt @@ -324,6 +324,7 @@ if(ICEBERG_BUILD_BUNDLE) avro/avro_register.cc avro/avro_schema_util.cc avro/avro_stream_internal.cc + parquet/parquet_bloom_row_group_filter.cc parquet/parquet_data_util.cc parquet/parquet_metrics.cc parquet/parquet_metrics_row_group_filter.cc diff --git a/src/iceberg/file_reader.h b/src/iceberg/file_reader.h index eded9191c6..3d07060ff5 100644 --- a/src/iceberg/file_reader.h +++ b/src/iceberg/file_reader.h @@ -83,7 +83,7 @@ class ICEBERG_EXPORT ReaderProperties : public ConfigBase { /// \brief Validate Arrow batches before merge-on-read projection. /// Default: true. Disable only when the reader is trusted to produce valid arrays. inline static Entry kArrowValidateBatch{"read.arrow.validate-batch", true}; - /// \brief Use footer statistics to prune Parquet row groups. + /// \brief Use footer statistics and Bloom filters to prune Parquet row groups. inline static Entry kParquetRowGroupFilter{ "read.parquet.row-group-filter.enabled", true}; /// \brief Skip GenericDatum in Avro reader for better performance. diff --git a/src/iceberg/parquet/parquet_bloom_row_group_filter.cc b/src/iceberg/parquet/parquet_bloom_row_group_filter.cc new file mode 100644 index 0000000000..ce7b5c230d --- /dev/null +++ b/src/iceberg/parquet/parquet_bloom_row_group_filter.cc @@ -0,0 +1,317 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include +#include +#include +#include +#include + +#include + +#include "iceberg/expression/expression_visitor.h" +#include "iceberg/expression/rewrite_not.h" +#include "iceberg/metadata_columns.h" +#include "iceberg/parquet/parquet_bloom_row_group_filter_internal.h" +#include "iceberg/parquet/parquet_schema_util_internal.h" +#include "iceberg/util/macros.h" + +namespace iceberg::parquet { +namespace { + +bool FindBytes(const ::parquet::BloomFilter& bloom, std::span bytes) { + ::parquet::ByteArray value{static_cast(bytes.size()), bytes.data()}; + return bloom.FindHash(bloom.Hash(value)); +} + +// Hash the file's physical representation, not Iceberg's single-value encoding. +// Java's BloomEvalVisitor.shouldRead uses the same physical-type dispatch. +bool MightContain(const ::parquet::ColumnDescriptor& column, const Literal& value, + const ::parquet::BloomFilter& bloom) { + if (value.IsNull() || value.IsAboveMax() || value.IsBelowMin() || value.IsNaN()) { + // Bloom hashes floating-point bits, whereas equality treats all NaN payloads + // alike. Neither a NaN predicate nor a NaN literal can safely exclude rows. + return true; + } + auto type = value.type()->type_id(); + switch (column.physical_type()) { + case ::parquet::Type::INT32: { + int128_t number; + switch (type) { + case TypeId::kInt: + case TypeId::kDate: + number = std::get(value.value()); + break; + case TypeId::kLong: + number = std::get(value.value()); + break; + case TypeId::kDecimal: + number = std::get(value.value()).value(); + break; + default: + return true; + } + if (number < std::numeric_limits::min() || + number > std::numeric_limits::max()) { + return false; + } + return bloom.FindHash(bloom.Hash(static_cast(number))); + } + case ::parquet::Type::INT64: { + int128_t number; + switch (type) { + case TypeId::kLong: + case TypeId::kTime: + case TypeId::kTimestamp: + case TypeId::kTimestampTz: + case TypeId::kTimestampNs: + case TypeId::kTimestampTzNs: + number = std::get(value.value()); + break; + case TypeId::kDecimal: + number = std::get(value.value()).value(); + break; + default: + return true; + } + if (number < std::numeric_limits::min() || + number > std::numeric_limits::max()) { + return false; + } + return bloom.FindHash(bloom.Hash(static_cast(number))); + } + case ::parquet::Type::FLOAT: { + float number; + if (type == TypeId::kFloat) { + number = std::get(value.value()); + } else if (type == TypeId::kDouble) { + auto promoted = std::get(value.value()); + if (std::isfinite(promoted) && + std::abs(promoted) > std::numeric_limits::max()) { + return false; + } + number = static_cast(promoted); + } else { + return true; + } + return bloom.FindHash(bloom.Hash(number)); + } + case ::parquet::Type::DOUBLE: + return type != TypeId::kDouble || + bloom.FindHash(bloom.Hash(std::get(value.value()))); + case ::parquet::Type::BYTE_ARRAY: + case ::parquet::Type::FIXED_LEN_BYTE_ARRAY: + switch (type) { + case TypeId::kString: + return bloom.FindHash( + bloom.Hash(std::string_view(std::get(value.value())))); + case TypeId::kBinary: + case TypeId::kFixed: + return FindBytes(bloom, std::get>(value.value())); + case TypeId::kUuid: + return FindBytes(bloom, std::get(value.value()).bytes()); + case TypeId::kDecimal: { + // Iceberg's Parquet decimal representation is precision-sized, + // two's-complement big endian, including sign extension. Use the + // file's precision, since the table decimal may have been promoted. + const auto& decimal = std::get(value.value()); + if (!decimal.FitsInPrecision(column.type_precision())) { + return false; + } + auto bytes = decimal.ToBigEndian(); + if (column.physical_type() == ::parquet::Type::BYTE_ARRAY) { + // Unlike FIXED, binary metadata does not specify the stored width. + // Include Java's precision-sized encoding and all other supported + // sign-extended representations to avoid false negatives. + for (size_t width = bytes.size(); width <= Decimal::kByteWidth; ++width) { + if (FindBytes(bloom, bytes)) { + return true; + } + bytes.insert(bytes.begin(), decimal.value() < 0 ? 0xff : 0x00); + } + return false; + } + int width = column.type_length(); + if (std::cmp_less(width, bytes.size())) { + return true; + } + bytes.insert(bytes.begin(), width - bytes.size(), + decimal.value() < 0 ? 0xff : 0x00); + return FindBytes(bloom, bytes); + } + default: + return true; + } + default: + // Boolean and legacy INT96 have no Iceberg Bloom conversion here. + return true; + } +} + +class BloomVisitor : public BoundVisitor { + public: + BloomVisitor(const ::parquet::arrow::SchemaManifest& manifest, + const ::parquet::RowGroupMetaData& row_group, + ::parquet::RowGroupBloomFilterReader* bloom_reader, + const std::unordered_map& columns) + : manifest_(manifest), + row_group_(row_group), + bloom_reader_(bloom_reader), + columns_(columns) {} + + Result AlwaysTrue() override { return true; } + Result AlwaysFalse() override { return false; } + // Make rewrites NOT. Negating an inclusive result would be unsafe. + Result Not(bool) override { return true; } + Result And(bool left, bool right) override { return left && right; } + Result Or(bool left, bool right) override { return left || right; } + + Result Eq(const std::shared_ptr& expr, const Literal& value) override { + if (value.IsNaN()) { + return true; + } + auto ref = std::dynamic_pointer_cast(expr); + ICEBERG_ASSIGN_OR_RAISE(auto bloom, GetBloomFilter(ref)); + return !bloom || MightContain(*manifest_.descr->Column(columns_.at(ref->field_id())), + value, *bloom); + } + + Result In(const std::shared_ptr& expr, + const BoundSetPredicate::LiteralSet& values) override { + if (std::ranges::any_of(values, [](const auto& value) { return value.IsNaN(); })) { + return true; + } + auto ref = std::dynamic_pointer_cast(expr); + ICEBERG_ASSIGN_OR_RAISE(auto bloom, GetBloomFilter(ref)); + if (!bloom) { + return true; + } + auto column = manifest_.descr->Column(columns_.at(ref->field_id())); + return std::ranges::any_of( + values, [&](const auto& value) { return MightContain(*column, value, *bloom); }); + } + Result IsNull(const std::shared_ptr&) override { return true; } + Result NotNull(const std::shared_ptr&) override { return true; } + Result IsNaN(const std::shared_ptr&) override { return true; } + Result NotNaN(const std::shared_ptr&) override { return true; } + Result Lt(const std::shared_ptr&, const Literal&) override { return true; } + Result LtEq(const std::shared_ptr&, const Literal&) override { + return true; + } + Result Gt(const std::shared_ptr&, const Literal&) override { return true; } + Result GtEq(const std::shared_ptr&, const Literal&) override { + return true; + } + Result NotEq(const std::shared_ptr&, const Literal&) override { + return true; + } + Result StartsWith(const std::shared_ptr&, const Literal&) override { + return true; + } + Result NotStartsWith(const std::shared_ptr&, const Literal&) override { + return true; + } + Result NotIn(const std::shared_ptr&, + const BoundSetPredicate::LiteralSet&) override { + return true; + } + + private: + Result GetBloomFilter( + const std::shared_ptr& ref) { + if (!ref || !ref->type()->is_primitive() || + MetadataColumns::IsMetadataColumn(ref->field_id())) { + return nullptr; + } + auto index = columns_.find(ref->field_id()); + if (index == columns_.end()) { + // Missing fields may have initial defaults; never infer all nulls here. + return nullptr; + } + auto field = manifest_.column_index_to_field.find(index->second); + if (field == manifest_.column_index_to_field.end() || + !ValidateParquetTypeCompatibility(*ref->type(), *field->second) || + manifest_.descr->Column(index->second)->max_repetition_level() > 0) { + return nullptr; + } + auto physical_type = manifest_.descr->Column(index->second)->physical_type(); + if (physical_type == ::parquet::Type::BOOLEAN || + physical_type == ::parquet::Type::INT96) { + return nullptr; + } + auto column = row_group_.ColumnChunk(index->second); + auto offset = column->bloom_filter_offset(); + if (!offset || *offset <= 0) { + return nullptr; + } + auto [entry, inserted] = cache_.try_emplace(ref->field_id()); + if (inserted) { + if (bloom_reader_) { + entry->second = bloom_reader_->GetColumnBloomFilter(index->second); + } + if (!entry->second) { + return Invalid("Missing Parquet Bloom filter for field {}", ref->field_id()); + } + } + return entry->second.get(); + } + + const ::parquet::arrow::SchemaManifest& manifest_; + const ::parquet::RowGroupMetaData& row_group_; + ::parquet::RowGroupBloomFilterReader* bloom_reader_; + const std::unordered_map& columns_; + std::unordered_map> cache_; +}; + +} // namespace + +Result> ParquetBloomRowGroupFilter::Make( + const std::shared_ptr& filter, + const ::parquet::SchemaDescriptor& file_schema) { + auto result = + std::unique_ptr(new ParquetBloomRowGroupFilter()); + for (int i = 0; i < file_schema.num_columns(); ++i) { + auto id = file_schema.Column(i)->schema_node()->field_id(); + if (id >= 0) { + result->field_id_to_column_idx_.emplace(id, i); + } + } + result->bound_ = True::Instance(); + if (filter) { + ICEBERG_ASSIGN_OR_RAISE(result->bound_, RewriteNot::Visit(filter)); + } + return result; +} + +Result ParquetBloomRowGroupFilter::ShouldRead( + const ::parquet::arrow::SchemaManifest& manifest, + const ::parquet::RowGroupMetaData& row_group, + ::parquet::RowGroupBloomFilterReader* bloom_reader) const { + if (row_group.num_rows() <= 0) { + return false; + } + try { + BloomVisitor visitor(manifest, row_group, bloom_reader, field_id_to_column_idx_); + return Visit(bound_, visitor); + } catch (const ::parquet::ParquetException& e) { + return Invalid("Cannot read Parquet Bloom filter: {}", e.what()); + } +} + +} // namespace iceberg::parquet diff --git a/src/iceberg/parquet/parquet_bloom_row_group_filter_internal.h b/src/iceberg/parquet/parquet_bloom_row_group_filter_internal.h new file mode 100644 index 0000000000..42adfcde05 --- /dev/null +++ b/src/iceberg/parquet/parquet_bloom_row_group_filter_internal.h @@ -0,0 +1,54 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#pragma once + +#include +#include +#include + +#include +#include +#include + +#include "iceberg/iceberg_bundle_export.h" +#include "iceberg/result.h" +#include "iceberg/type_fwd.h" + +namespace iceberg::parquet { + +// Inclusive Bloom filtering, following Java's ParquetBloomRowGroupFilter. +// False proves that no row can match; true still requires residual evaluation. +class ICEBERG_BUNDLE_EXPORT ParquetBloomRowGroupFilter { + public: + static Result> Make( + const std::shared_ptr& filter, + const ::parquet::SchemaDescriptor& file_schema); + + Result ShouldRead(const ::parquet::arrow::SchemaManifest& manifest, + const ::parquet::RowGroupMetaData& row_group, + ::parquet::RowGroupBloomFilterReader* bloom_reader) const; + + private: + ParquetBloomRowGroupFilter() = default; + std::shared_ptr bound_; + std::unordered_map field_id_to_column_idx_; +}; + +} // namespace iceberg::parquet diff --git a/src/iceberg/parquet/parquet_reader.cc b/src/iceberg/parquet/parquet_reader.cc index b88dfd000e..a45ca3a808 100644 --- a/src/iceberg/parquet/parquet_reader.cc +++ b/src/iceberg/parquet/parquet_reader.cc @@ -35,6 +35,7 @@ #include #include #include +#include #include #include @@ -42,6 +43,7 @@ #include "iceberg/arrow/arrow_status_internal.h" #include "iceberg/arrow/metadata_column_util_internal.h" #include "iceberg/expression/binder.h" +#include "iceberg/parquet/parquet_bloom_row_group_filter_internal.h" #include "iceberg/parquet/parquet_data_util_internal.h" #include "iceberg/parquet/parquet_metrics_row_group_filter_internal.h" #include "iceberg/parquet/parquet_register.h" @@ -284,6 +286,7 @@ class ParquetReader::Impl { read_schema_ = options.projection; stats_filter_.reset(); + bloom_filter_.reset(); // Prepare reader properties ::parquet::ReaderProperties reader_properties(pool_); @@ -317,6 +320,8 @@ class ParquetReader::Impl { } ICEBERG_ASSIGN_OR_RAISE(stats_filter_, ParquetMetricsRowGroupFilter::Make( filter, *reader_->manifest().descr)); + ICEBERG_ASSIGN_OR_RAISE(bloom_filter_, ParquetBloomRowGroupFilter::Make( + filter, *reader_->manifest().descr)); } ICEBERG_RETURN_UNEXPECTED(PlanRowGroups(options)); @@ -452,6 +457,13 @@ class ParquetReader::Impl { auto row_group = metadata->RowGroup(row_group_index); ICEBERG_ASSIGN_OR_RAISE( auto should_read, stats_filter_->ShouldRead(reader_->manifest(), *row_group)); + if (should_read && bloom_filter_) { + auto bloom_reader = + reader_->parquet_reader()->GetBloomFilterReader().RowGroup(row_group_index); + ICEBERG_ASSIGN_OR_RAISE( + should_read, bloom_filter_->ShouldRead(reader_->manifest(), *row_group, + bloom_reader.get())); + } if (should_read) { row_group_indices_[selected_count++] = row_group_index; } @@ -513,6 +525,7 @@ class ParquetReader::Impl { std::shared_ptr<::iceberg::Schema> read_schema_; // Footer-statistics filter for row group pruning, if enabled. std::unique_ptr stats_filter_; + std::unique_ptr bloom_filter_; // The projection result to apply to the read schema. SchemaProjection projection_; // The input stream to read Parquet file. diff --git a/src/iceberg/test/CMakeLists.txt b/src/iceberg/test/CMakeLists.txt index 50c336739e..5350ff28e0 100644 --- a/src/iceberg/test/CMakeLists.txt +++ b/src/iceberg/test/CMakeLists.txt @@ -232,6 +232,7 @@ if(ICEBERG_BUILD_BUNDLE) USE_BUNDLE SOURCES metrics_test_base.cc + parquet_bloom_row_group_filter_test.cc parquet_data_test.cc parquet_row_group_filter_test.cc parquet_metrics_test.cc diff --git a/src/iceberg/test/parquet_bloom_row_group_filter_test.cc b/src/iceberg/test/parquet_bloom_row_group_filter_test.cc new file mode 100644 index 0000000000..a0ae441ce6 --- /dev/null +++ b/src/iceberg/test/parquet_bloom_row_group_filter_test.cc @@ -0,0 +1,854 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "iceberg/arrow/arrow_io_internal.h" +#include "iceberg/arrow/arrow_status_internal.h" +#include "iceberg/expression/binder.h" +#include "iceberg/expression/expressions.h" +#include "iceberg/file_reader.h" +#include "iceberg/metadata_columns.h" +#include "iceberg/name_mapping.h" +#include "iceberg/parquet/parquet_bloom_row_group_filter_internal.h" +#include "iceberg/parquet/parquet_data_util_internal.h" +#include "iceberg/parquet/parquet_register.h" +#include "iceberg/schema.h" +#include "iceberg/schema_internal.h" +#include "iceberg/test/matchers.h" +#include "iceberg/test/mock_io.h" +#include "iceberg/type.h" +#include "iceberg/util/uuid.h" + +namespace iceberg::parquet { +namespace { + +struct BloomReadRange { + int64_t offset; + int64_t length; + bool operator==(const BloomReadRange&) const = default; +}; + +class BloomRecordingStream : public SeekableInputStream { + public: + BloomRecordingStream(std::unique_ptr stream, + std::vector& reads) + : stream_(std::move(stream)), reads_(reads) {} + Result Position() const override { return stream_->Position(); } + Status Seek(int64_t position) override { return stream_->Seek(position); } + Result Read(std::span out) override { + ICEBERG_ASSIGN_OR_RAISE(auto position, stream_->Position()); + reads_.push_back({.offset = position, .length = static_cast(out.size())}); + return stream_->Read(out); + } + Status ReadFully(int64_t position, std::span out) override { + reads_.push_back({.offset = position, .length = static_cast(out.size())}); + return stream_->ReadFully(position, out); + } + Status Close() override { return stream_->Close(); } + + private: + std::unique_ptr stream_; + std::vector& reads_; +}; + +class BloomRecordingInput : public InputFile { + public: + BloomRecordingInput(std::unique_ptr input, + std::vector& reads) + : input_(std::move(input)), reads_(reads) {} + std::string_view location() const override { return input_->location(); } + Result Size() const override { return input_->Size(); } + Result> Open() override { + ICEBERG_ASSIGN_OR_RAISE(auto stream, input_->Open()); + return std::make_unique(std::move(stream), reads_); + } + + private: + std::unique_ptr input_; + std::vector& reads_; +}; + +class BloomRecordingFileIO : public MockFileIO { + public: + Result> NewInputFile(std::string path) override { + ICEBERG_ASSIGN_OR_RAISE(auto file, MockFileIO::NewInputFile(std::move(path))); + return std::make_unique(std::move(file), reads); + } + Result> NewInputFile(std::string path, + size_t length) override { + ICEBERG_ASSIGN_OR_RAISE(auto file, MockFileIO::NewInputFile(std::move(path), length)); + return std::make_unique(std::move(file), reads); + } + std::vector reads; +}; + +class ParquetBloomFilterTest : public ::testing::Test { + protected: + void SetUp() override { + parquet::RegisterAll(); + io_ = std::make_shared(); + SetKeyType(int32()); + projection_ = std::make_shared( + std::vector{SchemaField::MakeRequired(2, "value", int64()), + MetadataColumns::kRowPosition, MetadataColumns::kRowId}); + } + + void SetKeyType(std::shared_ptr type) { + schema_ = std::make_shared( + std::vector{SchemaField::MakeOptional(1, "key", std::move(type)), + SchemaField::MakeRequired(2, "value", int64())}); + } + + Status Write(bool statistics = false, + const std::string& json = "[[0,0],[1,1],[10,2],[11,3],[20,4],[21,5]]") { + ArrowSchema c_schema; + ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(*schema_, &c_schema)); + ICEBERG_ARROW_ASSIGN_OR_RETURN(auto arrow_schema, ::arrow::ImportSchema(&c_schema)); + ICEBERG_ARROW_ASSIGN_OR_RETURN(auto array, + ::arrow::json::ArrayFromJSONString( + ::arrow::struct_(arrow_schema->fields()), json)); + ICEBERG_ARROW_ASSIGN_OR_RETURN(auto batch, + ::arrow::RecordBatch::FromStructArray(array)); + ICEBERG_ARROW_ASSIGN_OR_RETURN(auto table, + ::arrow::Table::FromRecordBatches({batch})); + return Write(table, statistics); + } + + Status Write(const std::shared_ptr<::arrow::Table>& table, bool statistics = true) { + ICEBERG_ASSIGN_OR_RAISE(auto out, arrow::OpenArrowOutputStream(io_, path_)); + ::parquet::WriterProperties::Builder properties; + properties.disable_dictionary(); + properties.enable_store_decimal_as_integer(); + if (write_bloom_) { + properties.enable_bloom_filter("key", {.ndv = 1000, .fpp = 0.0001}); + properties.enable_bloom_filter("key.nested", {.ndv = 1000, .fpp = 0.0001}); + } + if (writer_properties_) { + properties = ::parquet::WriterProperties::Builder(*writer_properties_); + } + if (!statistics) { + properties.disable_statistics(); + } + ICEBERG_ARROW_RETURN_NOT_OK( + ::parquet::arrow::WriteTable(*table, ::arrow::default_memory_pool(), out, + row_group_size_, properties.build())); + ICEBERG_ARROW_RETURN_NOT_OK(out->Close()); + ICEBERG_ASSIGN_OR_RAISE(auto input, arrow::OpenArrowInputStream(io_, path_)); + metadata_ = ::parquet::ParquetFileReader::Open(input)->metadata(); + return {}; + } + + ReaderOptions Options(std::shared_ptr filter, bool case_sensitive = true) { + // Bind against the table schema: key is intentionally outside the projection. + // Tests of Open's binding behavior set options.filter directly instead. + if (filter && filter->op() != Expression::Operation::kTrue && + filter->op() != Expression::Operation::kFalse) { + auto is_bound = IsBoundVisitor::IsBound(filter); + EXPECT_THAT(is_bound, IsOk()); + if (is_bound && !*is_bound) { + auto bound = Binder::Bind(*schema_, filter, case_sensitive); + EXPECT_THAT(bound, IsOk()); + if (bound) { + filter = *bound; + } + } + } + ReaderOptions options{.path = path_, + .io = io_, + .projection = projection_, + .filter = std::move(filter), + .first_row_id = 100}; + options.case_sensitive = case_sensitive; + options.properties.Set(ReaderProperties::kBatchSize, int64_t{3}); + return options; + } + + Result> Read(ReaderOptions options) { + ICEBERG_ASSIGN_OR_RAISE( + auto reader, ReaderFactoryRegistry::Open(FileFormatType::kParquet, options)); + ICEBERG_ASSIGN_OR_RAISE(auto c_schema, reader->Schema()); + ICEBERG_ARROW_ASSIGN_OR_RETURN(auto schema, ::arrow::ImportSchema(&c_schema)); + std::vector positions; + while (true) { + ICEBERG_ASSIGN_OR_RAISE(auto array, reader->Next()); + if (!array) { + break; + } + ICEBERG_ARROW_ASSIGN_OR_RETURN(auto batch, + ::arrow::ImportRecordBatch(&*array, schema)); + auto values = std::static_pointer_cast<::arrow::Int64Array>(batch->column(0)); + auto pos = std::static_pointer_cast<::arrow::Int64Array>(batch->column(1)); + auto ids = std::static_pointer_cast<::arrow::Int64Array>(batch->column(2)); + for (int64_t i = 0; i < batch->num_rows(); ++i) { + EXPECT_EQ(values->Value(i), pos->Value(i)); + EXPECT_EQ(ids->Value(i), 100 + pos->Value(i)); + positions.push_back(pos->Value(i)); + } + } + ICEBERG_RETURN_UNEXPECTED(reader->Close()); + return positions; + } + + void Check(const ReaderOptions& options, const std::vector& expected) { + SCOPED_TRACE(options.filter ? options.filter->ToString() : "no filter"); + ICEBERG_UNWRAP_OR_FAIL(auto actual, Read(options)); + EXPECT_EQ(actual, expected); + } + + bool write_bloom_ = true; + int64_t row_group_size_ = 2; + std::shared_ptr<::parquet::WriterProperties> writer_properties_; + std::string path_ = "bloom.parquet"; + std::shared_ptr io_; + std::shared_ptr schema_; + std::shared_ptr projection_; + std::shared_ptr<::parquet::FileMetaData> metadata_; + + void CheckBloom(const std::shared_ptr& expression, + const std::vector& expected, bool case_sensitive = true) { + ICEBERG_UNWRAP_OR_FAIL(auto input, arrow::OpenArrowInputStream(io_, path_)); + auto reader = ::parquet::ParquetFileReader::Open(input); + ::parquet::arrow::SchemaManifest manifest; + ::parquet::ArrowReaderProperties properties; + properties.set_arrow_extensions_enabled(true); + ASSERT_TRUE(::parquet::arrow::SchemaManifest::Make(metadata_->schema(), + metadata_->key_value_metadata(), + properties, &manifest) + .ok()); + ICEBERG_UNWRAP_OR_FAIL(auto filter, ParquetBloomRowGroupFilter::Make( + Options(expression, case_sensitive).filter, + *metadata_->schema())); + ASSERT_EQ(expected.size(), metadata_->num_row_groups()); + for (int i = 0; i < metadata_->num_row_groups(); ++i) { + SCOPED_TRACE(expression->ToString() + " group " + std::to_string(i)); + auto bloom_reader = reader->GetBloomFilterReader().RowGroup(i); + ICEBERG_UNWRAP_OR_FAIL( + auto actual, + filter->ShouldRead(manifest, *metadata_->RowGroup(i), bloom_reader.get())); + EXPECT_EQ(actual, expected[i]); + } + } +}; + +TEST_F(ParquetBloomFilterTest, EqualityMembershipAndInclusiveBooleanRules) { + ASSERT_THAT(Write(true, "[[0,0],[10,1],[5,2],[15,3],[0,4],[10,5]]"), IsOk()); + CheckBloom(Expressions::Equal("key", Literal::Int(5)), {false, true, false}); + CheckBloom(Expressions::In("key", {Literal::Int(5), Literal::Int(7)}), + {false, true, false}); + auto supported = Expressions::Equal("key", Literal::Int(5)); + auto unsupported = Expressions::LessThan("key", Literal::Int(100)); + CheckBloom(Expressions::And(supported, unsupported), {false, true, false}); + CheckBloom(Expressions::Or(supported, unsupported), {true, true, true}); + CheckBloom(Expressions::Not(Expressions::NotEqual("key", Literal::Int(5))), + {false, true, false}); + CheckBloom( + Expressions::Not(Expressions::NotIn("key", {Literal::Int(5), Literal::Int(7)})), + {false, true, false}); + CheckBloom(Expressions::AlwaysTrue(), {true, true, true}); + CheckBloom(Expressions::AlwaysFalse(), {false, false, false}); + Check(Options(supported), {2, 3}); + Check(Options(Expressions::In("key", {Literal::Int(5), Literal::Int(7)})), {2, 3}); + auto options = Options(supported); + options.properties.Set(ReaderProperties::kParquetRowGroupFilter, false); + Check(options, {0, 1, 2, 3, 4, 5}); + Check(Options(nullptr), {0, 1, 2, 3, 4, 5}); +} + +TEST_F(ParquetBloomFilterTest, UnsupportedPredicatesAndNullsRemainInclusive) { + ASSERT_THAT(Write(true, "[[null,0],[null,1],[5,2],[5,3],[10,4],[10,5]]"), IsOk()); + for (const auto& expression : std::vector>{ + Expressions::IsNull("key"), Expressions::NotNull("key"), + Expressions::LessThan("key", Literal::Int(1)), + Expressions::LessThanOrEqual("key", Literal::Int(1)), + Expressions::GreaterThan("key", Literal::Int(100)), + Expressions::GreaterThanOrEqual("key", Literal::Int(100)), + Expressions::NotEqual("key", Literal::Int(5)), + Expressions::NotIn("key", {Literal::Int(5), Literal::Int(10)}), + Expressions::Not(Expressions::LessThan("key", Literal::Int(5)))}) { + CheckBloom(expression, {true, true, true}); + } + SetKeyType(string()); + ASSERT_THAT(Write(true, R"([["abc",0],["abc",1],[null,2],["abc",3]])"), IsOk()); + CheckBloom(Expressions::StartsWith("key", "z"), {true, true}); + CheckBloom(Expressions::NotStartsWith("key", "a"), {true, true}); +} + +TEST_F(ParquetBloomFilterTest, PhysicalTypesAndSchemaPromotions) { + struct Case { + std::shared_ptr file_type; + std::shared_ptr read_type; + std::string json; + Literal present; + Literal absent; + }; + const std::string numbers = "[[0,0],[10,1],[5,2],[15,3],[0,4],[10,5]]"; + const std::string decimals = + R"([["0.00",0],["10.00",1],["5.00",2],["15.00",3],["0.00",4],["10.00",5]])"; + const std::string timestamps = + R"([["1970-01-01 00:00:00",0],["1970-01-01 00:00:10",1],["1970-01-01 00:00:05",2],["1970-01-01 00:00:15",3],["1970-01-01 00:00:00",4],["1970-01-01 00:00:10",5]])"; + const std::string strings = R"([["a",0],["z",1],["m",2],["z",3],["a",4],["z",5]])"; + for (const auto& test : + std::vector{{.file_type = int32(), + .read_type = int32(), + .json = numbers, + .present = Literal::Int(5), + .absent = Literal::Int(7)}, + {.file_type = int32(), + .read_type = int64(), + .json = numbers, + .present = Literal::Long(5), + .absent = Literal::Long(7)}, + {.file_type = int64(), + .read_type = int64(), + .json = numbers, + .present = Literal::Long(5), + .absent = Literal::Long(7)}, + {.file_type = float32(), + .read_type = float32(), + .json = numbers, + .present = Literal::Float(5), + .absent = Literal::Float(7)}, + {.file_type = float32(), + .read_type = float64(), + .json = numbers, + .present = Literal::Double(5), + .absent = Literal::Double(7)}, + {.file_type = float64(), + .read_type = float64(), + .json = numbers, + .present = Literal::Double(5), + .absent = Literal::Double(7)}, + {.file_type = date(), + .read_type = date(), + .json = numbers, + .present = Literal::Date(5), + .absent = Literal::Date(7)}, + {.file_type = time(), + .read_type = time(), + .json = numbers, + .present = Literal::Time(5), + .absent = Literal::Time(7)}, + {.file_type = timestamp(), + .read_type = timestamp(), + .json = timestamps, + .present = Literal::Timestamp(5000000), + .absent = Literal::Timestamp(7000000)}, + {.file_type = timestamp_tz(), + .read_type = timestamp_tz(), + .json = timestamps, + .present = Literal::TimestampTz(5000000), + .absent = Literal::TimestampTz(7000000)}, + {.file_type = timestamp_ns(), + .read_type = timestamp_ns(), + .json = timestamps, + .present = Literal::TimestampNs(5000000000), + .absent = Literal::TimestampNs(7000000000)}, + {.file_type = timestamptz_ns(), + .read_type = timestamptz_ns(), + .json = timestamps, + .present = Literal::TimestampTzNs(5000000000), + .absent = Literal::TimestampTzNs(7000000000)}, + {.file_type = string(), + .read_type = string(), + .json = strings, + .present = Literal::String("m"), + .absent = Literal::String("x")}, + {.file_type = binary(), + .read_type = binary(), + .json = strings, + .present = Literal::Binary({'m'}), + .absent = Literal::Binary({'x'})}, + {.file_type = fixed(1), + .read_type = fixed(1), + .json = strings, + .present = Literal::Fixed({'m'}), + .absent = Literal::Fixed({'x'})}, + {.file_type = decimal(9, 2), + .read_type = decimal(9, 2), + .json = decimals, + .present = Literal::Decimal(500, 9, 2), + .absent = Literal::Decimal(700, 9, 2)}, + {.file_type = decimal(9, 2), + .read_type = decimal(18, 2), + .json = decimals, + .present = Literal::Decimal(500, 18, 2), + .absent = Literal::Decimal(700, 18, 2)}, + {.file_type = decimal(18, 2), + .read_type = decimal(18, 2), + .json = decimals, + .present = Literal::Decimal(500, 18, 2), + .absent = Literal::Decimal(700, 18, 2)}, + {.file_type = decimal(30, 2), + .read_type = decimal(38, 2), + .json = decimals, + .present = Literal::Decimal(500, 38, 2), + .absent = Literal::Decimal(700, 38, 2)}}) { + SCOPED_TRACE(test.file_type->ToString() + " -> " + test.read_type->ToString()); + SetKeyType(test.file_type); + ASSERT_THAT(Write(false, test.json), IsOk()); + SetKeyType(test.read_type); + CheckBloom(Expressions::Equal("key", test.present), {false, true, false}); + CheckBloom(Expressions::Equal("key", test.absent), {false, false, false}); + CheckBloom(Expressions::In("key", {test.present, test.absent}), {false, true, false}); + Check(Options(Expressions::Equal("key", test.present)), {2, 3}); + } +} + +TEST_F(ParquetBloomFilterTest, MissingBloomFiltersAndFields) { + write_bloom_ = false; + ASSERT_THAT(Write(), IsOk()); + CheckBloom(Expressions::Equal("key", Literal::Int(100)), {true, true, true}); + Check(Options(Expressions::Equal("key", Literal::Int(100))), {0, 1, 2, 3, 4, 5}); + write_bloom_ = true; + ASSERT_THAT(Write(), IsOk()); + schema_ = std::make_shared(std::vector{ + SchemaField::MakeOptional(3, "missing", int32()) + .WithInitialDefault(std::make_shared(Literal::Int(100)))}); + // Initial-default fields are absent from the physical schema. + CheckBloom(Expressions::Equal("missing", Literal::Int(100)), {true, true, true}); + Check(Options(Expressions::Equal("missing", Literal::Int(100))), {0, 1, 2, 3, 4, 5}); +} + +TEST_F(ParquetBloomFilterTest, IncompatibleTypesRetainGroups) { + ASSERT_THAT(Write(), IsOk()); + SetKeyType(date()); + CheckBloom(Expressions::Equal("key", Literal::Date(100)), {true, true, true}); + SetKeyType(int64()); + CheckBloom( + Expressions::Equal("key", Literal::Long(std::numeric_limits::max())), + {false, false, false}); + SetKeyType(float64()); + CheckBloom(Expressions::Equal("key", Literal::Double(100)), {true, true, true}); +} + +TEST_F(ParquetBloomFilterTest, RenameCaseSensitivityAndUnboundReferences) { + ASSERT_THAT(Write(), IsOk()); + schema_ = std::make_shared( + std::vector{SchemaField::MakeOptional(1, "renamed", int32())}); + CheckBloom(Expressions::Equal("RENAMED", Literal::Int(10)), {false, true, false}, + false); + auto options = Options(Expressions::Equal("RENAMED", Literal::Int(10)), false); + Check(options, {2, 3}); + options.filter = + Expressions::Equal("value", Literal::Long(2)); // Projected but has no Bloom. + Check(options, {0, 1, 2, 3, 4, 5}); + options.filter = Expressions::Equal("unresolved", Literal::Int(2)); + EXPECT_THAT(ReaderFactoryRegistry::Open(FileFormatType::kParquet, options), + HasErrorMessage("Cannot find field")); +} + +TEST_F(ParquetBloomFilterTest, NonContiguousGroupsSplitsAndLineage) { + ASSERT_THAT(Write(false, "[[5,0],[15,1],[0,2],[10,3],[5,4],[15,5]]"), IsOk()); + auto options = Options(Expressions::Equal("key", Literal::Int(5))); + for (int64_t batch_size : {1, 2, 3, 8}) { + options.properties.Set(ReaderProperties::kBatchSize, batch_size); + Check(options, {0, 1, 4, 5}); + } + ICEBERG_UNWRAP_OR_FAIL(auto input, arrow::OpenArrowInputStream(io_, path_)); + ICEBERG_UNWRAP_OR_FAIL(auto midpoint, RowGroupMidpoint(*metadata_->RowGroup(1), + input->GetSize().ValueOrDie())); + options.split = Split{.offset = static_cast(midpoint), .length = 1}; + Check(options, {}); + options.filter = Options(Expressions::Equal("key", Literal::Int(0))).filter; + Check(options, {2, 3}); + options.split->length = 0; + Check(options, {}); +} + +TEST_F(ParquetBloomFilterTest, NaNsAndSignedZeros) { + for (const auto& type : std::vector>{float32(), float64()}) { + SetKeyType(type); + ASSERT_THAT(Write(false, "[[NaN,0],[NaN,1],[null,2],[0,3],[-0.0,4],[-0.0,5]]"), + IsOk()); + CheckBloom(Expressions::IsNaN("key"), {true, true, true}); + CheckBloom(Expressions::NotNaN("key"), {true, true, true}); + auto positive = + type->type_id() == TypeId::kFloat ? Literal::Float(0.0F) : Literal::Double(0.0); + auto negative = + type->type_id() == TypeId::kFloat ? Literal::Float(-0.0F) : Literal::Double(-0.0); + CheckBloom(Expressions::Equal("key", positive), {false, true, false}); + CheckBloom(Expressions::Equal("key", negative), {false, false, true}); + } +} +TEST_F(ParquetBloomFilterTest, NameMappingWithoutFieldIdsRetainsGroups) { + ::parquet::WriterProperties::Builder properties; + writer_properties_ = + properties.disable_dictionary() + ->enable_bloom_filter("stored_key", {.ndv = 1000, .fpp = 0.0001}) + ->build(); + auto keys = ::arrow::json::ArrayFromJSONString(::arrow::int32(), "[0,1,10,11,20,21]") + .ValueOrDie(); + auto values = + ::arrow::json::ArrayFromJSONString(::arrow::int64(), "[0,1,2,3,4,5]").ValueOrDie(); + auto file_schema = ::arrow::schema({::arrow::field("stored_key", ::arrow::int32()), + ::arrow::field("stored_value", ::arrow::int64())}); + ASSERT_THAT(Write(::arrow::Table::Make(file_schema, {keys, values})), IsOk()); + ASSERT_EQ(metadata_->schema()->Column(0)->schema_node()->field_id(), -1); + ASSERT_TRUE(metadata_->RowGroup(0)->ColumnChunk(0)->bloom_filter_offset().has_value()); + auto mapping = + NameMapping::Make({MappedField{.names = {"stored_key"}, .field_id = 1}, + MappedField{.names = {"stored_value"}, .field_id = 2}}); + auto options = Options(Expressions::Equal("key", Literal::Int(100))); + options.name_mapping = std::move(mapping); + // Name mapping resolves the projection. Footer columns without IDs remain + // unavailable to both filters, even though a Bloom payload exists. + Check(options, {0, 1, 2, 3, 4, 5}); +} + +TEST_F(ParquetBloomFilterTest, UnsupportedTransformRetainsGroups) { + ASSERT_THAT(Write(), IsOk()); + Check(Options(Expressions::Equal(Expressions::Bucket("key", 16), + Literal::Int(15))), + {0, 1, 2, 3, 4, 5}); +} + +TEST_F(ParquetBloomFilterTest, RequiredLeafWithNullableParent) { + SetKeyType(std::make_shared( + std::vector{SchemaField::MakeRequired(3, "nested", int32())})); + ASSERT_THAT(Write(true, "[[null,0],[null,1],[[10],2],[[11],3],[null,4],[null,5]]"), + IsOk()); + auto is_null = Expressions::IsNull("key.nested"); + auto not_null = Expressions::NotNull("key.nested"); + Check(Options(is_null), {0, 1, 4, 5}); + Check(Options(not_null), {2, 3}); + Check(Options(Expressions::Not(is_null)), {2, 3}); + Check(Options(Expressions::Not(not_null)), {0, 1, 4, 5}); + Check(Options(Expressions::Not(Expressions::LessThan("key.nested", Literal::Int(100)))), + {0, 1, 4, 5}); +} + +TEST_F(ParquetBloomFilterTest, NestedPrimitiveStatistics) { + SetKeyType(std::make_shared( + std::vector{SchemaField::MakeOptional(3, "nested", int32())})); + ASSERT_THAT(Write(true, "[[[0],0],[[1],1],[[10],2],[[11],3],[[20],4],[[21],5]]"), + IsOk()); + CheckBloom(Expressions::Equal("key.nested", Literal::Int(100)), {false, false, false}); + CheckBloom(Expressions::Equal("key.nested", Literal::Int(10)), {false, true, false}); + Check(Options(Expressions::Equal("key.nested", Literal::Int(100))), {}); + Check(Options(Expressions::Equal("key.nested", Literal::Int(10))), {2, 3}); +} + +TEST_F(ParquetBloomFilterTest, UuidBoundsUseUnsignedByteOrder) { + SetKeyType(uuid()); + // Group 0 holds UUIDs below 0x80..., group 1 holds UUIDs at or above it. A signed + // byte comparison would order group 1 before group 0. + std::vector uuids; + for (const auto* str : + {"00000000-0000-0000-0000-000000000001", "7fffffff-ffff-ffff-ffff-ffffffffffff", + "80000000-0000-0000-0000-000000000000", "ffffffff-ffff-ffff-ffff-fffffffffffe"}) { + ICEBERG_UNWRAP_OR_FAIL(auto uuid, Uuid::FromString(str)); + uuids.push_back(uuid); + } + ::arrow::FixedSizeBinaryBuilder keys(::arrow::fixed_size_binary(Uuid::kLength)); + for (const auto& uuid : uuids) { + ASSERT_TRUE(keys.Append(uuid.bytes().data()).ok()); + } + auto key_array = ::arrow::ExtensionType::WrapArray(::arrow::extension::uuid(), + keys.Finish().ValueOrDie()); + auto values = + ::arrow::json::ArrayFromJSONString(::arrow::int64(), "[0,1,2,3]").ValueOrDie(); + ArrowSchema c_schema; + ASSERT_THAT(ToArrowSchema(*schema_, &c_schema), IsOk()); + auto arrow_schema = ::arrow::ImportSchema(&c_schema).ValueOrDie(); + ASSERT_THAT(Write(::arrow::Table::Make(arrow_schema, {key_array, values})), IsOk()); + ASSERT_EQ(metadata_->num_row_groups(), 2); + ASSERT_TRUE(metadata_->RowGroup(0)->ColumnChunk(0)->statistics()->HasMinMax()); + + CheckBloom(Expressions::Equal("key", Literal::UUID(uuids[2])), {false, true}); + const auto& group0_max = uuids[1]; + const auto& group1_min = uuids[2]; + Check(Options(Expressions::LessThan("key", Literal::UUID(group1_min))), {0, 1}); + Check(Options(Expressions::LessThanOrEqual("key", Literal::UUID(group0_max))), {0, 1}); + Check(Options(Expressions::GreaterThan("key", Literal::UUID(group0_max))), {2, 3}); + Check(Options(Expressions::GreaterThanOrEqual("key", Literal::UUID(group1_min))), + {2, 3}); + Check(Options(Expressions::Equal("key", Literal::UUID(group1_min))), {2, 3}); + Check( + Options(Expressions::Not(Expressions::LessThan("key", Literal::UUID(group1_min)))), + {2, 3}); +} + +class CountingBloomReader : public ::parquet::RowGroupBloomFilterReader { + public: + enum class Mode { kNormal, kMissing, kThrow, kSaturated }; + CountingBloomReader(std::shared_ptr<::parquet::RowGroupBloomFilterReader> delegate, + Mode mode) + : delegate_(std::move(delegate)), mode_(mode) {} + std::unique_ptr<::parquet::BloomFilter> GetColumnBloomFilter(int i) override { + ++calls; + switch (mode_) { + case Mode::kMissing: + return nullptr; + case Mode::kThrow: + throw ::parquet::ParquetException("test Bloom read failure"); + case Mode::kSaturated: { + std::array bits; + bits.fill(0xff); + auto bloom = std::make_unique<::parquet::BlockSplitBloomFilter>(); + bloom->Init(bits.data(), static_cast(bits.size())); + return bloom; + } + default: + return delegate_->GetColumnBloomFilter(i); + } + } + int calls = 0; + + private: + std::shared_ptr<::parquet::RowGroupBloomFilterReader> delegate_; + Mode mode_; +}; + +TEST_F(ParquetBloomFilterTest, CacheReadErrorsAndFalsePositives) { + ASSERT_THAT(Write(), IsOk()); + ICEBERG_UNWRAP_OR_FAIL(auto input, arrow::OpenArrowInputStream(io_, path_)); + auto reader = ::parquet::ParquetFileReader::Open(input); + ::parquet::arrow::SchemaManifest manifest; + ASSERT_TRUE(::parquet::arrow::SchemaManifest::Make( + metadata_->schema(), nullptr, + ::parquet::default_arrow_reader_properties(), &manifest) + .ok()); + auto expr = + Expressions::And(Expressions::Equal("key", Literal::Int(100)), + Expressions::In("key", {Literal::Int(101), Literal::Int(102)})); + ICEBERG_UNWRAP_OR_FAIL(auto filter, ParquetBloomRowGroupFilter::Make( + Options(expr).filter, *metadata_->schema())); + auto delegate = reader->GetBloomFilterReader().RowGroup(0); + for (auto mode : + {CountingBloomReader::Mode::kNormal, CountingBloomReader::Mode::kMissing, + CountingBloomReader::Mode::kThrow, CountingBloomReader::Mode::kSaturated}) { + CountingBloomReader tracked(delegate, mode); + auto result = filter->ShouldRead(manifest, *metadata_->RowGroup(0), &tracked); + if (mode == CountingBloomReader::Mode::kMissing || + mode == CountingBloomReader::Mode::kThrow) { + EXPECT_FALSE(result.has_value()); + } else { + ASSERT_THAT(result, IsOk()); + EXPECT_EQ(*result, mode == CountingBloomReader::Mode::kSaturated); + } + EXPECT_EQ(tracked.calls, 1); + } + // Unsupported predicates must not load any Bloom payload. + for (const auto& unsupported : std::vector>{ + Expressions::LessThan("key", Literal::Int(100)), + Expressions::NotEqual("key", Literal::Int(100)), Expressions::IsNull("key"), + Expressions::NotNull("key"), Expressions::NotIn("key", {Literal::Int(100)})}) { + ICEBERG_UNWRAP_OR_FAIL(auto unsupported_filter, + ParquetBloomRowGroupFilter::Make(Options(unsupported).filter, + *metadata_->schema())); + CountingBloomReader tracked(delegate, CountingBloomReader::Mode::kThrow); + ICEBERG_UNWRAP_OR_FAIL(auto result, unsupported_filter->ShouldRead( + manifest, *metadata_->RowGroup(0), &tracked)); + EXPECT_TRUE(result); + EXPECT_EQ(tracked.calls, 0); + } +} + +TEST_F(ParquetBloomFilterTest, CorruptBloomPayloadFailsOpen) { + ASSERT_THAT(Write(), IsOk()); + auto column = metadata_->RowGroup(0)->ColumnChunk(0); + ASSERT_TRUE(column->bloom_filter_offset().has_value()); + ASSERT_TRUE(column->bloom_filter_length().has_value()); + auto& data = std::static_pointer_cast(io_)->FileData(path_); + std::ranges::fill(std::span(data).subspan(*column->bloom_filter_offset(), + *column->bloom_filter_length()), + std::byte{0}); + EXPECT_THAT( + ReaderFactoryRegistry::Open(FileFormatType::kParquet, + Options(Expressions::Equal("key", Literal::Int(100)))), + HasErrorMessage("Cannot read Parquet Bloom filter")); +} + +TEST_F(ParquetBloomFilterTest, FooterPruningAndDisabledOptionAvoidBloomReads) { + ASSERT_THAT(Write(true), IsOk()); + auto io = std::static_pointer_cast(io_); + io->reads.clear(); + ICEBERG_UNWRAP_OR_FAIL( + auto unfiltered, + ReaderFactoryRegistry::Open(FileFormatType::kParquet, Options(nullptr))); + auto footer_reads = io->reads; + ASSERT_THAT(unfiltered->Close(), IsOk()); + auto pruned = Options(Expressions::Equal("key", Literal::Int(100))); + auto disabled = Options(Expressions::Equal("key", Literal::Int(10))); + disabled.properties.Set(ReaderProperties::kParquetRowGroupFilter, false); + for (const auto& options : {pruned, disabled}) { + io->reads.clear(); + ICEBERG_UNWRAP_OR_FAIL( + auto reader, ReaderFactoryRegistry::Open(FileFormatType::kParquet, options)); + EXPECT_EQ(io->reads, footer_reads); + ASSERT_THAT(reader->Close(), IsOk()); + } +} + +TEST_F(ParquetBloomFilterTest, OpenReadsOnlyBloomPayload) { + row_group_size_ = 768 * 1024; + ::parquet::WriterProperties::Builder properties; + writer_properties_ = properties.disable_dictionary() + ->enable_bloom_filter("key", {.ndv = 1000, .fpp = 0.0001}) + ->compression(::parquet::Compression::UNCOMPRESSED) + ->data_pagesize(8 * 1024 * 1024) + ->max_row_group_length(row_group_size_) + ->max_rows_per_page(row_group_size_) + ->build(); + ::arrow::Int32Builder keys; + ::arrow::Int64Builder positions; + ASSERT_TRUE(keys.Reserve(row_group_size_).ok()); + ASSERT_TRUE(positions.Reserve(row_group_size_).ok()); + for (int64_t i = 0; i < row_group_size_; ++i) { + keys.UnsafeAppend(static_cast(2 * (i % 256))); + positions.UnsafeAppend(i); + } + ArrowSchema c_schema; + ASSERT_THAT(ToArrowSchema(*schema_, &c_schema), IsOk()); + auto arrow_schema = ::arrow::ImportSchema(&c_schema).ValueOrDie(); + ASSERT_THAT(Write(::arrow::Table::Make(arrow_schema, {keys.Finish().ValueOrDie(), + positions.Finish().ValueOrDie()}), + false), + IsOk()); + ASSERT_EQ(metadata_->num_row_groups(), 1); + auto column = metadata_->RowGroup(0)->ColumnChunk(0); + ASSERT_GT(column->total_compressed_size(), 2 * 1024 * 1024); + ASSERT_TRUE(column->bloom_filter_offset().has_value()); + ASSERT_TRUE(column->bloom_filter_length().has_value()); + auto io = std::static_pointer_cast(io_); + io->reads.clear(); + ICEBERG_UNWRAP_OR_FAIL( + auto unfiltered, + ReaderFactoryRegistry::Open(FileFormatType::kParquet, Options(nullptr))); + auto footer_reads = io->reads; + ASSERT_THAT(unfiltered->Close(), IsOk()); + for (int value : {1, 2}) { + io->reads.clear(); + ICEBERG_UNWRAP_OR_FAIL(auto reader, + ReaderFactoryRegistry::Open( + FileFormatType::kParquet, + Options(Expressions::Equal("key", Literal::Int(value))))); + auto expected = footer_reads; + expected.push_back({.offset = *column->bloom_filter_offset(), + .length = *column->bloom_filter_length()}); + EXPECT_EQ(io->reads, expected); + ASSERT_THAT(reader->Close(), IsOk()); + } +} + +TEST_F(ParquetBloomFilterTest, ReaderReuseResetsBloomFilter) { + ASSERT_THAT(Write(), IsOk()); + ICEBERG_UNWRAP_OR_FAIL( + auto reader, + ReaderFactoryRegistry::Open(FileFormatType::kParquet, + Options(Expressions::Equal("key", Literal::Int(100))))); + ICEBERG_UNWRAP_OR_FAIL(auto empty, reader->Next()); + EXPECT_FALSE(empty.has_value()); + ASSERT_THAT(reader->Close(), IsOk()); + ASSERT_THAT(reader->Open(Options(nullptr)), IsOk()); + ICEBERG_UNWRAP_OR_FAIL(auto batch, reader->Next()); + ASSERT_TRUE(batch.has_value()); + EXPECT_GT(batch->length, 0); + batch->release(&*batch); + ASSERT_THAT(reader->Close(), IsOk()); +} + +TEST_F(ParquetBloomFilterTest, NegativeAndPromotedDecimalsUseFileWidth) { + for (int precision : {9, 18, 30}) { + SetKeyType(decimal(precision, 2)); + ASSERT_THAT( + Write( + false, + R"([["0.00",0],["10.00",1],["-5.00",2],["15.00",3],["0.00",4],["10.00",5]])"), + IsOk()); + SetKeyType(decimal(38, 2)); + CheckBloom(Expressions::Equal("key", Literal::Decimal(-500, 38, 2)), + {false, true, false}); + CheckBloom(Expressions::Equal("key", Literal::Decimal(-700, 38, 2)), + {false, false, false}); + CheckBloom(Expressions::Equal("key", Literal::Decimal(100000000000LL, 38, 2)), + {false, false, false}); + Check(Options(Expressions::Equal("key", Literal::Decimal(-500, 38, 2))), {2, 3}); + } +} + +TEST_F(ParquetBloomFilterTest, BinaryDecimalsWithDifferentSignExtensionWidths) { + SetKeyType(decimal(18, 2)); + for (int width : {2, 4, 8}) { + // Decimal precision 8 normally uses four bytes in Java. A variable-length + // binary can also use minimal bytes or additional sign extension. + auto key_node = ::parquet::schema::PrimitiveNode::Make( + "key", ::parquet::Repetition::REQUIRED, ::parquet::LogicalType::Decimal(8, 2), + ::parquet::Type::BYTE_ARRAY, -1, 1); + auto value_node = ::parquet::schema::PrimitiveNode::Make( + "value", ::parquet::Repetition::REQUIRED, ::parquet::Type::INT64, + ::parquet::ConvertedType::NONE, -1, -1, -1, 2); + auto file_schema = std::static_pointer_cast<::parquet::schema::GroupNode>( + ::parquet::schema::GroupNode::Make("schema", ::parquet::Repetition::REQUIRED, + {key_node, value_node})); + ::parquet::WriterProperties::Builder properties; + auto props = properties.disable_dictionary() + ->disable_statistics() + ->enable_bloom_filter("key", {.ndv = 1000, .fpp = 0.0001}) + ->build(); + ICEBERG_UNWRAP_OR_FAIL(auto out, arrow::OpenArrowOutputStream(io_, path_)); + auto writer = ::parquet::ParquetFileWriter::Open(out, file_schema, props); + auto group = writer->AppendRowGroup(); + auto positive = Decimal(500).ToBigEndian(); + auto negative = Decimal(-500).ToBigEndian(); + positive.insert(positive.begin(), width - positive.size(), 0x00); + negative.insert(negative.begin(), width - negative.size(), 0xff); + std::array<::parquet::ByteArray, 2> values{ + {{static_cast(positive.size()), positive.data()}, + {static_cast(negative.size()), negative.data()}}}; + auto keys = static_cast<::parquet::ByteArrayWriter*>(group->NextColumn()); + keys->WriteBatch(2, nullptr, nullptr, values.data()); + keys->Close(); + std::array positions{0, 1}; + auto data = static_cast<::parquet::Int64Writer*>(group->NextColumn()); + data->WriteBatch(2, nullptr, nullptr, positions.data()); + data->Close(); + group->Close(); + writer->Close(); + ASSERT_TRUE(out->Close().ok()); + ICEBERG_UNWRAP_OR_FAIL(auto input, arrow::OpenArrowInputStream(io_, path_)); + metadata_ = ::parquet::ParquetFileReader::Open(input)->metadata(); + CheckBloom(Expressions::Equal("key", Literal::Decimal(500, 18, 2)), {true}); + CheckBloom(Expressions::Equal("key", Literal::Decimal(-500, 18, 2)), {true}); + CheckBloom(Expressions::Equal("key", Literal::Decimal(700, 18, 2)), {false}); + Check(Options(Expressions::Equal("key", Literal::Decimal(-500, 18, 2))), {0, 1}); + } +} + +} // namespace +} // namespace iceberg::parquet