From 84415a5056242820493e6df5624c778983a15f3a Mon Sep 17 00:00:00 2001 From: yinzhengsun Date: Mon, 21 Sep 2026 20:05:43 +0800 Subject: [PATCH 1/5] feat(parquet): add statistics-based row group filtering --- src/iceberg/CMakeLists.txt | 1 + src/iceberg/data/file_scan_task_reader.cc | 24 +- src/iceberg/data/file_scan_task_reader.h | 2 +- src/iceberg/file_reader.h | 5 + src/iceberg/parquet/parquet_metrics.cc | 90 +-- .../parquet/parquet_metrics_internal.h | 6 + .../parquet_metrics_row_group_filter.cc | 355 +++++++++++ ...arquet_metrics_row_group_filter_internal.h | 49 ++ src/iceberg/parquet/parquet_reader.cc | 114 ++-- src/iceberg/parquet/parquet_schema_util.cc | 4 - .../parquet/parquet_schema_util_internal.h | 7 + src/iceberg/test/CMakeLists.txt | 1 + .../test/file_scan_task_reader_test.cc | 74 +++ .../test/parquet_row_group_filter_test.cc | 549 ++++++++++++++++++ src/iceberg/test/parquet_test.cc | 13 +- 15 files changed, 1193 insertions(+), 101 deletions(-) create mode 100644 src/iceberg/parquet/parquet_metrics_row_group_filter.cc create mode 100644 src/iceberg/parquet/parquet_metrics_row_group_filter_internal.h create mode 100644 src/iceberg/test/parquet_row_group_filter_test.cc diff --git a/src/iceberg/CMakeLists.txt b/src/iceberg/CMakeLists.txt index 8a98274ff..edf862b16 100644 --- a/src/iceberg/CMakeLists.txt +++ b/src/iceberg/CMakeLists.txt @@ -314,6 +314,7 @@ if(ICEBERG_BUILD_BUNDLE) avro/avro_stream_internal.cc parquet/parquet_data_util.cc parquet/parquet_metrics.cc + parquet/parquet_metrics_row_group_filter.cc parquet/parquet_reader.cc parquet/parquet_register.cc parquet/parquet_schema_util.cc diff --git a/src/iceberg/data/file_scan_task_reader.cc b/src/iceberg/data/file_scan_task_reader.cc index 0c41f25ed..9c923a215 100644 --- a/src/iceberg/data/file_scan_task_reader.cc +++ b/src/iceberg/data/file_scan_task_reader.cc @@ -28,6 +28,7 @@ #include "iceberg/arrow_c_data_guard_internal.h" #include "iceberg/arrow_c_data_util_internal.h" #include "iceberg/data/delete_filter.h" +#include "iceberg/expression/binder.h" #include "iceberg/file_reader.h" #include "iceberg/manifest/manifest_entry.h" #include "iceberg/schema.h" @@ -164,10 +165,21 @@ class FileScanTaskReader::Impl { "Data file size must not be negative: {}", data_file->file_size_in_bytes); + auto filter = task.residual_filter(); + if (filter) { + ICEBERG_ASSIGN_OR_RAISE(auto is_bound, IsBoundVisitor::IsBound(filter)); + if (!is_bound) { + ICEBERG_ASSIGN_OR_RAISE( + filter, + Binder::Bind(*table_schema_, filter, + properties_.Get(ReaderProperties::kFilterCaseSensitive))); + } + } + if (task.delete_files().empty()) { auto options = MakeReaderOptions( - *data_file, io_, projected_schema_, task.residual_filter(), name_mapping_, - properties_, data_file->first_row_id, data_file->data_sequence_number); + *data_file, io_, projected_schema_, filter, name_mapping_, properties_, + data_file->first_row_id, data_file->data_sequence_number); ICEBERG_ASSIGN_OR_RAISE( auto reader, ReaderFactoryRegistry::Open(data_file->file_format, options)); return MakeArrowArrayStream(std::move(reader)); @@ -191,9 +203,9 @@ class FileScanTaskReader::Impl { ProjectionContext::Make(*required_schema, *projected_schema_, project_batch_function)); - auto options = MakeReaderOptions( - *data_file, io_, required_schema, task.residual_filter(), name_mapping_, - properties_, data_file->first_row_id, data_file->data_sequence_number); + auto options = MakeReaderOptions(*data_file, io_, required_schema, filter, + name_mapping_, properties_, data_file->first_row_id, + data_file->data_sequence_number); ICEBERG_ASSIGN_OR_RAISE(auto reader, ReaderFactoryRegistry::Open(data_file->file_format, options)); @@ -207,6 +219,7 @@ class FileScanTaskReader::Impl { Impl(Options options, DeleteFilter::FieldLookup field_lookup, std::shared_ptr delete_counter) : io_(std::move(options.io)), + table_schema_(std::move(options.table_schema)), schemas_(std::move(options.schemas)), projected_schema_(std::move(options.projected_schema)), name_mapping_(std::move(options.name_mapping)), @@ -215,6 +228,7 @@ class FileScanTaskReader::Impl { delete_counter_(std::move(delete_counter)) {} std::shared_ptr io_; + std::shared_ptr table_schema_; std::vector> schemas_; std::shared_ptr projected_schema_; std::shared_ptr name_mapping_; diff --git a/src/iceberg/data/file_scan_task_reader.h b/src/iceberg/data/file_scan_task_reader.h index a71ef5f84..daef86d5c 100644 --- a/src/iceberg/data/file_scan_task_reader.h +++ b/src/iceberg/data/file_scan_task_reader.h @@ -48,7 +48,7 @@ class ICEBERG_DATA_EXPORT FileScanTaskReader { struct Options { /// FileIO instance for reading data and delete files. std::shared_ptr io; - /// The table schema. Used as the primary field lookup for delete file resolution. + /// The table schema used to bind filters and resolve delete-file fields. std::shared_ptr table_schema; /// Optional list of historical table schemas for field lookup. std::vector> schemas; diff --git a/src/iceberg/file_reader.h b/src/iceberg/file_reader.h index e08b2df49..efc9b40e1 100644 --- a/src/iceberg/file_reader.h +++ b/src/iceberg/file_reader.h @@ -80,6 +80,11 @@ class ICEBERG_EXPORT ReaderProperties : public ConfigBase { /// Only the Parquet reader honors this option; other readers ignore it. /// Default: false (use 32-bit offset list). inline static Entry kArrowUseLargeList{"read.arrow.use-large-list", false}; + /// \brief Use footer statistics to prune Parquet row groups. + inline static Entry kParquetRowGroupFilter{ + "read.parquet.row-group-filter.enabled", true}; + /// \brief Case sensitivity when binding unbound filter references. + inline static Entry kFilterCaseSensitive{"read.filter.case-sensitive", true}; /// \brief Skip GenericDatum in Avro reader for better performance. /// When true, decode directly from Avro to Arrow without GenericDatum intermediate. /// Default: true (skip GenericDatum for better performance). diff --git a/src/iceberg/parquet/parquet_metrics.cc b/src/iceberg/parquet/parquet_metrics.cc index 22710c137..bad4545a7 100644 --- a/src/iceberg/parquet/parquet_metrics.cc +++ b/src/iceberg/parquet/parquet_metrics.cc @@ -191,47 +191,6 @@ bool NeedsBoundTruncation(const PrimitiveType& type) { return type.type_id() == TypeId::kString || type.type_id() == TypeId::kBinary; } -Result StatsValueToLiteral(const ::parquet::ColumnDescriptor& column, - const PrimitiveType& iceberg_type, - const ::parquet::Statistics& stats, bool is_min) { - switch (column.physical_type()) { - case ::parquet::Type::BOOLEAN: - return TypedStatsLiteral<::parquet::BoolStatistics>( - stats, is_min, [](bool value) { return Literal::Boolean(value); }); - case ::parquet::Type::INT32: - return TypedStatsLiteral<::parquet::Int32Statistics>( - stats, is_min, - [&](int32_t value) { return Int32StatsLiteral(value, iceberg_type); }); - case ::parquet::Type::INT64: - return TypedStatsLiteral<::parquet::Int64Statistics>( - stats, is_min, - [&](int64_t value) { return Int64StatsLiteral(value, iceberg_type); }); - case ::parquet::Type::FLOAT: - return TypedStatsLiteral<::parquet::FloatStatistics>( - stats, is_min, - [&](float value) { return FloatStatsLiteral(value, iceberg_type); }); - case ::parquet::Type::DOUBLE: - return TypedStatsLiteral<::parquet::DoubleStatistics>( - stats, is_min, [](double value) { return Literal::Double(value); }); - case ::parquet::Type::BYTE_ARRAY: - return TypedStatsLiteral<::parquet::ByteArrayStatistics>( - stats, is_min, [&](const ::parquet::ByteArray& value) { - return BinaryStatsLiteral(BytesFromByteArray(value), iceberg_type); - }); - case ::parquet::Type::FIXED_LEN_BYTE_ARRAY: - return TypedStatsLiteral<::parquet::FLBAStatistics>( - stats, is_min, [&](const ::parquet::FixedLenByteArray& value) { - return BinaryStatsLiteral(BytesFromFLBA(value, column.type_length()), - iceberg_type); - }); - case ::parquet::Type::INT96: - case ::parquet::Type::UNDEFINED: - return NotSupported("Cannot convert Parquet statistics for physical type {}", - static_cast(column.physical_type())); - } - std::unreachable(); -} - /// \brief Collect counts (value count and null count) from footer statistics. /// \param field_id The Iceberg field ID. /// \param metadata The Parquet file metadata. @@ -288,15 +247,15 @@ Result> CollectBounds( value_count += column_chunk->num_values(); if (stats->HasMinMax()) { - ICEBERG_ASSIGN_OR_RAISE(auto min_value, - StatsValueToLiteral(*column_desc, *iceberg_type, *stats, + ICEBERG_ASSIGN_OR_RAISE(auto min_value, ParquetMetrics::StatsValueToLiteral( + *column_desc, *iceberg_type, *stats, /*is_min=*/true)); if (!lower_bound.has_value() || min_value < lower_bound.value()) { lower_bound = std::move(min_value); } - ICEBERG_ASSIGN_OR_RAISE(auto max_value, - StatsValueToLiteral(*column_desc, *iceberg_type, *stats, + ICEBERG_ASSIGN_OR_RAISE(auto max_value, ParquetMetrics::StatsValueToLiteral( + *column_desc, *iceberg_type, *stats, /*is_min=*/false)); if (!upper_bound.has_value() || max_value > upper_bound.value()) { upper_bound = std::move(max_value); @@ -504,6 +463,47 @@ class CollectMetricsVisitor { } // namespace +Result ParquetMetrics::StatsValueToLiteral( + const ::parquet::ColumnDescriptor& column, const PrimitiveType& iceberg_type, + const ::parquet::Statistics& stats, bool is_min) { + switch (column.physical_type()) { + case ::parquet::Type::BOOLEAN: + return TypedStatsLiteral<::parquet::BoolStatistics>( + stats, is_min, [](bool value) { return Literal::Boolean(value); }); + case ::parquet::Type::INT32: + return TypedStatsLiteral<::parquet::Int32Statistics>( + stats, is_min, + [&](int32_t value) { return Int32StatsLiteral(value, iceberg_type); }); + case ::parquet::Type::INT64: + return TypedStatsLiteral<::parquet::Int64Statistics>( + stats, is_min, + [&](int64_t value) { return Int64StatsLiteral(value, iceberg_type); }); + case ::parquet::Type::FLOAT: + return TypedStatsLiteral<::parquet::FloatStatistics>( + stats, is_min, + [&](float value) { return FloatStatsLiteral(value, iceberg_type); }); + case ::parquet::Type::DOUBLE: + return TypedStatsLiteral<::parquet::DoubleStatistics>( + stats, is_min, [](double value) { return Literal::Double(value); }); + case ::parquet::Type::BYTE_ARRAY: + return TypedStatsLiteral<::parquet::ByteArrayStatistics>( + stats, is_min, [&](const ::parquet::ByteArray& value) { + return BinaryStatsLiteral(BytesFromByteArray(value), iceberg_type); + }); + case ::parquet::Type::FIXED_LEN_BYTE_ARRAY: + return TypedStatsLiteral<::parquet::FLBAStatistics>( + stats, is_min, [&](const ::parquet::FixedLenByteArray& value) { + return BinaryStatsLiteral(BytesFromFLBA(value, column.type_length()), + iceberg_type); + }); + case ::parquet::Type::INT96: + case ::parquet::Type::UNDEFINED: + return NotSupported("Cannot convert Parquet statistics for physical type {}", + static_cast(column.physical_type())); + } + std::unreachable(); +} + Result ParquetMetrics::GetMetrics( const Schema& schema, const ::parquet::SchemaDescriptor& parquet_schema, const MetricsConfig& metrics_config, const ::parquet::FileMetaData& metadata, diff --git a/src/iceberg/parquet/parquet_metrics_internal.h b/src/iceberg/parquet/parquet_metrics_internal.h index c78c114ab..18d5c0999 100644 --- a/src/iceberg/parquet/parquet_metrics_internal.h +++ b/src/iceberg/parquet/parquet_metrics_internal.h @@ -38,6 +38,12 @@ class ParquetMetrics { public: ParquetMetrics() = delete; + /// Convert one footer bound to an Iceberg literal, including supported promotions. + static Result StatsValueToLiteral(const ::parquet::ColumnDescriptor& column, + const PrimitiveType& iceberg_type, + const ::parquet::Statistics& stats, + bool is_min); + /// \brief Compute file-level metrics from Parquet file metadata. /// /// This function extracts metrics including row count, column sizes, value counts, diff --git a/src/iceberg/parquet/parquet_metrics_row_group_filter.cc b/src/iceberg/parquet/parquet_metrics_row_group_filter.cc new file mode 100644 index 000000000..735892f2d --- /dev/null +++ b/src/iceberg/parquet/parquet_metrics_row_group_filter.cc @@ -0,0 +1,355 @@ +/* + * 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 "iceberg/expression/binder.h" +#include "iceberg/expression/expression_visitor.h" +#include "iceberg/expression/rewrite_not.h" +#include "iceberg/metadata_columns.h" +#include "iceberg/metrics.h" +#include "iceberg/parquet/parquet_metrics_internal.h" +#include "iceberg/parquet/parquet_metrics_row_group_filter_internal.h" +#include "iceberg/parquet/parquet_schema_util_internal.h" +#include "iceberg/schema.h" +#include "iceberg/type.h" +#include "iceberg/util/macros.h" + +namespace iceberg::parquet { + +namespace { + +constexpr size_t kInPredicateLimit = 200; + +class BindFilter : public Binder { + public: + BindFilter(const Schema& schema, bool case_sensitive, + const ::parquet::arrow::SchemaManifest& manifest) + : Binder(schema, case_sensitive) { + for (const auto& [index, field] : manifest.column_index_to_field) { + const auto* column = manifest.descr->Column(index); + // Repeated-column counts and bounds do not describe individual rows. + if (column->max_repetition_level() == 0) { + fields_.emplace(column->schema_node()->field_id(), field); + } + } + } + + Result> Predicate( + const std::shared_ptr& pred) override { + auto bound = Binder::Predicate(pred); + // Direct format readers may only supply a projection, not the complete schema. + if (!bound) { + return True::Instance(); + } + return Visit>(*bound, *this); + } + + Result> Predicate( + const std::shared_ptr& pred) override { + auto ref = std::dynamic_pointer_cast(pred->term()); + if (!ref || !ref->type()->is_primitive() || + MetadataColumns::IsMetadataColumn(ref->field_id()) || + MetadataColumns::IsRowLineageColumn(ref->field_id())) { + return True::Instance(); + } + auto field = fields_.find(ref->field_id()); + if (field == fields_.end() || + !ValidateParquetTypeCompatibility(*ref->type(), *field->second)) { + return True::Instance(); + } + return pred; + } + + private: + std::unordered_map fields_; +}; + +class MetricsVisitor : public BoundVisitor { + public: + MetricsVisitor(const ::parquet::SchemaDescriptor& schema, + const ::parquet::RowGroupMetaData& row_group) + : schema_(schema), row_group_(row_group) { + for (int i = 0; i < schema.num_columns(); ++i) { + auto id = schema.Column(i)->schema_node()->field_id(); + if (id >= 0) { + columns_.emplace(id, i); + } + } + } + + Result AlwaysTrue() override { return true; } + + Result AlwaysFalse() override { return false; } + + 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 IsNull(const std::shared_ptr& expr) override { + return MayContainNull(GetMetrics(expr, false)); + } + + Result NotNull(const std::shared_ptr& expr) override { + return !ContainsNullsOnly(GetMetrics(expr, false)); + } + + Result IsNaN(const std::shared_ptr& expr) override { + return !ContainsNullsOnly(GetMetrics(expr, false)); + } + + Result NotNaN(const std::shared_ptr&) override { return true; } + + Result Lt(const std::shared_ptr& expr, const Literal& value) override { + const auto metrics = GetMetrics(expr); + if (ContainsNullsOnly(metrics)) { + return false; + } + if (!metrics.lower_bound || !ComparableLiteral(value)) { + return true; + } + return !(*metrics.lower_bound >= value); + } + + Result LtEq(const std::shared_ptr& expr, const Literal& value) override { + const auto metrics = GetMetrics(expr); + if (ContainsNullsOnly(metrics)) { + return false; + } + if (!metrics.lower_bound || !ComparableLiteral(value)) { + return true; + } + return !(*metrics.lower_bound > value); + } + + Result Gt(const std::shared_ptr& expr, const Literal& value) override { + const auto metrics = GetMetrics(expr); + if (ContainsNullsOnly(metrics)) { + return false; + } + if (!metrics.upper_bound || !ComparableLiteral(value)) { + return true; + } + return !(*metrics.upper_bound <= value); + } + + Result GtEq(const std::shared_ptr& expr, const Literal& value) override { + const auto metrics = GetMetrics(expr); + if (ContainsNullsOnly(metrics)) { + return false; + } + if (!metrics.upper_bound || !ComparableLiteral(value)) { + return true; + } + return !(*metrics.upper_bound < value); + } + + Result Eq(const std::shared_ptr& expr, const Literal& value) override { + const auto metrics = GetMetrics(expr); + if (ContainsNullsOnly(metrics)) { + return false; + } + if (!metrics.lower_bound || !metrics.upper_bound || !ComparableLiteral(value)) { + return true; + } + return !(*metrics.lower_bound > value || *metrics.upper_bound < value); + } + + Result NotEq(const std::shared_ptr&, const Literal&) override { + // Like Java, keep negative membership predicates inclusive. + return true; + } + + Result In(const std::shared_ptr& expr, + const BoundSetPredicate::LiteralSet& values) override { + const auto metrics = GetMetrics(expr); + if (ContainsNullsOnly(metrics)) { + return false; + } + if (!metrics.lower_bound || !metrics.upper_bound || + values.size() > kInPredicateLimit) { + return true; + } + for (const auto& value : values) { + if (!ComparableLiteral(value) || + !(value < *metrics.lower_bound || value > *metrics.upper_bound)) { + return true; + } + } + return false; + } + + Result NotIn(const std::shared_ptr&, + const BoundSetPredicate::LiteralSet&) override { + return true; + } + + Result StartsWith(const std::shared_ptr& expr, + const Literal& value) override { + const auto metrics = GetMetrics(expr); + if (ContainsNullsOnly(metrics)) { + return false; + } + if (!metrics.lower_bound || !metrics.upper_bound || !ComparableLiteral(value) || + metrics.lower_bound->type()->type_id() != TypeId::kString) { + return true; + } + const auto& prefix = std::get(value.value()); + const auto& lower = std::get(metrics.lower_bound->value()); + const auto& upper = std::get(metrics.upper_bound->value()); + return !(lower.substr(0, prefix.size()) > prefix || + upper.substr(0, prefix.size()) < prefix); + } + + Result NotStartsWith(const std::shared_ptr& expr, + const Literal& value) override { + const auto metrics = GetMetrics(expr); + if (MayContainNull(metrics) || !metrics.lower_bound || !metrics.upper_bound || + !ComparableLiteral(value) || + metrics.lower_bound->type()->type_id() != TypeId::kString) { + return true; + } + const auto& prefix = std::get(value.value()); + const auto& lower = std::get(metrics.lower_bound->value()); + const auto& upper = std::get(metrics.upper_bound->value()); + return !lower.starts_with(prefix) || !upper.starts_with(prefix); + } + + private: + static bool ContainsNullsOnly(const FieldMetrics& metrics) { + return metrics.null_value_count >= 0 && + metrics.null_value_count == metrics.value_count; + } + + static bool MayContainNull(const FieldMetrics& metrics) { + return metrics.null_value_count != 0; + } + + static bool ComparableLiteral(const Literal& value) { + return !value.IsNaN() && !value.IsNull() && !value.IsAboveMax() && + !value.IsBelowMin(); + } + + FieldMetrics GetMetrics(const std::shared_ptr& expr, + bool read_bounds = true) const { + FieldMetrics metrics; + auto ref = std::dynamic_pointer_cast(expr); + if (!ref) { + return metrics; + } + metrics.field_id = ref->field_id(); + auto column = columns_.find(ref->field_id()); + // Missing columns can have initial defaults, so do not assume all nulls. + if (column == columns_.end()) { + return metrics; + } + const auto& descriptor = *schema_.Column(column->second); + const auto& type = static_cast(*ref->type()); + auto chunk = row_group_.ColumnChunk(column->second); + auto stats = chunk->statistics(); + if (!stats) { + return metrics; + } + metrics.value_count = chunk->num_values(); + if (stats->HasNullCount()) { + metrics.null_value_count = stats->null_count(); + } + if (!read_bounds || ContainsNullsOnly(metrics) || !stats->HasMinMax()) { + return metrics; + } + auto lower_result = + ParquetMetrics::StatsValueToLiteral(descriptor, type, *stats, true); + auto upper_result = + ParquetMetrics::StatsValueToLiteral(descriptor, type, *stats, false); + if (!lower_result || !upper_result) { + return metrics; + } + auto lower = std::move(*lower_result); + auto upper = std::move(*upper_result); + if (lower.IsNaN() || upper.IsNaN()) { + return metrics; + } + if (type.type_id() == TypeId::kFloat) { + if (std::get(lower.value()) == 0) { + lower = Literal::Float(-0.0F); + } + if (std::get(upper.value()) == 0) { + upper = Literal::Float(0.0F); + } + } else if (type.type_id() == TypeId::kDouble) { + if (std::get(lower.value()) == 0) { + lower = Literal::Double(-0.0); + } + if (std::get(upper.value()) == 0) { + upper = Literal::Double(0.0); + } + } + if (lower > upper) { + return metrics; + } + metrics.lower_bound = std::move(lower); + metrics.upper_bound = std::move(upper); + return metrics; + } + + const ::parquet::SchemaDescriptor& schema_; + const ::parquet::RowGroupMetaData& row_group_; + std::unordered_map columns_; +}; + +} // namespace + +Result> ParquetMetricsRowGroupFilter::Make( + const Schema& schema, const std::shared_ptr& filter, + const ::parquet::arrow::SchemaManifest& manifest, bool case_sensitive) { + auto result = + std::unique_ptr(new ParquetMetricsRowGroupFilter()); + result->bound_ = True::Instance(); + if (!filter) { + return result; + } + // Eliminate NOT before unsupported predicates are weakened to true. + ICEBERG_ASSIGN_OR_RAISE(auto rewritten, RewriteNot::Visit(filter)); + BindFilter binder(schema, case_sensitive, manifest); + ICEBERG_ASSIGN_OR_RAISE(result->bound_, + Visit>(rewritten, binder)); + return result; +} + +Result ParquetMetricsRowGroupFilter::ShouldRead( + const ::parquet::SchemaDescriptor& file_schema, + const ::parquet::RowGroupMetaData& row_group) const { + if (row_group.num_rows() <= 0) { + return false; + } + try { + MetricsVisitor visitor(file_schema, row_group); + return Visit(bound_, visitor); + } catch (const ::parquet::ParquetException&) { + // Unusable optional statistics must never turn into false negatives. + return true; + } +} + +} // namespace iceberg::parquet diff --git a/src/iceberg/parquet/parquet_metrics_row_group_filter_internal.h b/src/iceberg/parquet/parquet_metrics_row_group_filter_internal.h new file mode 100644 index 000000000..2907dc35c --- /dev/null +++ b/src/iceberg/parquet/parquet_metrics_row_group_filter_internal.h @@ -0,0 +1,49 @@ +/* + * 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 "iceberg/iceberg_bundle_export.h" +#include "iceberg/result.h" +#include "iceberg/type_fwd.h" + +namespace iceberg::parquet { + +// Inclusive footer-statistics filtering, following Java's +// ParquetMetricsRowGroupFilter. False proves that the row group cannot match; +// true still requires evaluating the residual predicate on the returned rows. +class ICEBERG_BUNDLE_EXPORT ParquetMetricsRowGroupFilter { + public: + static Result> Make( + const Schema& schema, const std::shared_ptr& filter, + const ::parquet::arrow::SchemaManifest& manifest, bool case_sensitive = true); + + Result ShouldRead(const ::parquet::SchemaDescriptor& file_schema, + const ::parquet::RowGroupMetaData& row_group) const; + + private: + ParquetMetricsRowGroupFilter() = default; + std::shared_ptr bound_; +}; + +} // namespace iceberg::parquet diff --git a/src/iceberg/parquet/parquet_reader.cc b/src/iceberg/parquet/parquet_reader.cc index 1f8146107..ac1eb2def 100644 --- a/src/iceberg/parquet/parquet_reader.cc +++ b/src/iceberg/parquet/parquet_reader.cc @@ -20,7 +20,6 @@ #include "iceberg/parquet/parquet_reader.h" #include -#include #include #include @@ -39,6 +38,7 @@ #include "iceberg/arrow/arrow_status_internal.h" #include "iceberg/arrow/metadata_column_util_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" #include "iceberg/parquet/parquet_schema_util_internal.h" #include "iceberg/result.h" @@ -65,13 +65,7 @@ Result BuildProjection(::parquet::arrow::FileReader* reader, return NotImplemented("Applying name mapping to Parquet schema is not implemented"); } - ::parquet::arrow::SchemaManifest schema_manifest; - ICEBERG_ARROW_RETURN_NOT_OK(::parquet::arrow::SchemaManifest::Make( - metadata->schema(), metadata->key_value_metadata(), reader->properties(), - &schema_manifest)); - - // Leverage SchemaManifest to project the schema - ICEBERG_ASSIGN_OR_RAISE(auto projection, Project(read_schema, schema_manifest)); + ICEBERG_ASSIGN_OR_RAISE(auto projection, Project(read_schema, reader->manifest())); return projection; } @@ -262,6 +256,12 @@ std::shared_ptr<::arrow::Schema> AlignOutputSchemaToReaderSchema( // A stateful context to keep track of the reading progress. struct ReadContext { + struct SelectedRowGroup { + int index; + int64_t first_row; + }; + std::vector row_groups_; + size_t current_row_group_ = 0; // The arrow schema to output record batches. It may be different with // the schema of record batches returned by `record_batch_reader_` // when there is any schema evolution. @@ -285,6 +285,7 @@ class ParquetReader::Impl { split_ = options.split; read_schema_ = options.projection; + stats_filter_.reset(); // Prepare reader properties ::parquet::ReaderProperties reader_properties(pool_); @@ -307,6 +308,15 @@ class ParquetReader::Impl { // Project read schema onto the Parquet file schema ICEBERG_ASSIGN_OR_RAISE(projection_, BuildProjection(reader_.get(), *read_schema_)); + if (options.filter && + options.properties.Get(ReaderProperties::kParquetRowGroupFilter)) { + ICEBERG_ASSIGN_OR_RAISE( + stats_filter_, + ParquetMetricsRowGroupFilter::Make( + *options.projection, options.filter, reader_->manifest(), + options.properties.Get(ReaderProperties::kFilterCaseSensitive))); + } + metadata_context_ = {.file_path = options.path, .next_file_pos = 0, .first_row_id = options.first_row_id, @@ -322,18 +332,29 @@ class ParquetReader::Impl { } ICEBERG_ARROW_ASSIGN_OR_RETURN(auto batch, context_->record_batch_reader_->Next()); - if (!batch) { - return std::nullopt; + while (!batch) { + const size_t next_group = context_->current_row_group_ + 1; + if (next_group >= context_->row_groups_.size()) { + return std::nullopt; + } + ICEBERG_ARROW_RETURN_NOT_OK(context_->record_batch_reader_->Close()); + const auto& group = context_->row_groups_[next_group]; + ICEBERG_ARROW_ASSIGN_OR_RETURN( + context_->record_batch_reader_, + reader_->GetRecordBatchReader({group.index}, + SelectedColumnIndices(projection_))); + context_->current_row_group_ = next_group; + metadata_context_.next_file_pos = group.first_row; + ICEBERG_ARROW_ASSIGN_OR_RETURN(batch, context_->record_batch_reader_->Next()); } ICEBERG_ASSIGN_OR_RAISE( batch, ProjectRecordBatch(std::move(batch), context_->output_arrow_schema_, *read_schema_, projection_, metadata_context_, pool_)); - metadata_context_.next_file_pos += batch->num_rows(); - ArrowArray arrow_array; ICEBERG_ARROW_RETURN_NOT_OK(::arrow::ExportRecordBatch(*batch, &arrow_array)); + metadata_context_.next_file_pos += batch->num_rows(); return arrow_array; } @@ -388,38 +409,41 @@ class ParquetReader::Impl { private: Status InitReadContext() { - context_ = std::make_unique(); - - // Row group pruning based on the split - // TODO(gangwu): add row group filtering based on zone map, bloom filter, etc. - std::vector row_group_indices; - if (split_.has_value()) { - auto metadata = reader_->parquet_reader()->metadata(); - for (int i = 0; i < metadata->num_row_groups(); ++i) { - auto row_group_offset = metadata->RowGroup(i)->file_offset(); - if (row_group_offset >= split_->offset && - row_group_offset < split_->offset + split_->length) { - row_group_indices.push_back(i); - } else if (row_group_offset >= split_->offset + split_->length) { - break; - } else { - metadata_context_.next_file_pos += metadata->RowGroup(i)->num_rows(); + auto context = std::make_unique(); + auto metadata = reader_->parquet_reader()->metadata(); + + int64_t next_row_start = 0; + for (int i = 0; i < metadata->num_row_groups(); ++i) { + auto row_group = metadata->RowGroup(i); + const int64_t row_start = next_row_start; + next_row_start += row_group->num_rows(); + if (split_.has_value()) { + auto row_group_offset = row_group->file_offset(); + bool in_split = row_group_offset >= split_->offset && + row_group_offset < split_->offset + split_->length; + if (!in_split) { + continue; } } - } else { - row_group_indices.resize(reader_->parquet_reader()->metadata()->num_row_groups()); - std::iota(row_group_indices.begin(), row_group_indices.end(), 0); // NOLINT + if (stats_filter_) { + ICEBERG_ASSIGN_OR_RAISE( + auto should_read, stats_filter_->ShouldRead(*metadata->schema(), *row_group)); + if (!should_read) { + continue; + } + } + if (row_group->num_rows() == 0) { + continue; + } + context->row_groups_.push_back({i, row_start}); } - - // Create the record batch reader - if (row_group_indices.empty()) { - // None of the row groups are selected, return an empty record batch reader - context_->record_batch_reader_ = std::make_unique(); + if (context->row_groups_.empty()) { + context->record_batch_reader_ = std::make_unique(); } else { - auto column_indices = SelectedColumnIndices(projection_); ICEBERG_ARROW_ASSIGN_OR_RETURN( - context_->record_batch_reader_, - reader_->GetRecordBatchReader(row_group_indices, column_indices)); + context->record_batch_reader_, + reader_->GetRecordBatchReader({context->row_groups_.front().index}, + SelectedColumnIndices(projection_))); } // Build the output Arrow schema from the projected Iceberg schema. This schema is the @@ -427,7 +451,7 @@ class ParquetReader::Impl { // the schema of the file. ArrowSchema arrow_schema; ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(*read_schema_, &arrow_schema)); - ICEBERG_ARROW_ASSIGN_OR_RETURN(context_->output_arrow_schema_, + ICEBERG_ARROW_ASSIGN_OR_RETURN(context->output_arrow_schema_, ::arrow::ImportSchema(&arrow_schema)); // Align the output schema with the arrays the reader actually produces. The reader's @@ -438,10 +462,15 @@ class ParquetReader::Impl { // 3. Mixed list and large_list types in files with stored schemas // For each projected field, we use the reader's actual type. For missing fields // (columns not in the file), we apply the configured use_large_list preference. - context_->output_arrow_schema_ = AlignOutputSchemaToReaderSchema( - context_->output_arrow_schema_, context_->record_batch_reader_->schema(), + context->output_arrow_schema_ = AlignOutputSchemaToReaderSchema( + context->output_arrow_schema_, context->record_batch_reader_->schema(), projection_, use_large_list_); + // Publish the read state only after initialization succeeds. + if (!context->row_groups_.empty()) { + metadata_context_.next_file_pos = context->row_groups_.front().first_row; + } + context_ = std::move(context); return {}; } @@ -454,6 +483,7 @@ class ParquetReader::Impl { bool use_large_list_ = false; // Schema to read from the Parquet file. std::shared_ptr<::iceberg::Schema> read_schema_; + std::unique_ptr stats_filter_; // The projection result to apply to the read schema. SchemaProjection projection_; // The input stream to read Parquet file. diff --git a/src/iceberg/parquet/parquet_schema_util.cc b/src/iceberg/parquet/parquet_schema_util.cc index 0b6abf218..1e8a5c7ee 100644 --- a/src/iceberg/parquet/parquet_schema_util.cc +++ b/src/iceberg/parquet/parquet_schema_util.cc @@ -320,8 +320,6 @@ Status ValidateGeospatialParquetType(const Type& expected_type, } // namespace -namespace { - Status ValidateParquetTypeCompatibility( const Type& expected_type, const ::parquet::arrow::SchemaField& parquet_field) { const auto& arrow_type = parquet_field.field->type(); @@ -477,8 +475,6 @@ Status ValidateParquetTypeCompatibility( expected_type, arrow_type->ToString()); } -} // namespace - namespace { class ProjectionBuilder { diff --git a/src/iceberg/parquet/parquet_schema_util_internal.h b/src/iceberg/parquet/parquet_schema_util_internal.h index 2eafd013b..6d9e5a022 100644 --- a/src/iceberg/parquet/parquet_schema_util_internal.h +++ b/src/iceberg/parquet/parquet_schema_util_internal.h @@ -38,6 +38,13 @@ struct ParquetExtraAttributes : public FieldProjection::ExtraAttributes { std::optional column_id; }; +/// \brief Validate a Parquet field's Arrow type against the expected Iceberg type. +/// Checks supported type promotions, time units, and decimal precision/scale. +/// Nested fields require separate validation of their children; geospatial types +/// require additional checks against the Parquet column descriptor. +Status ValidateParquetTypeCompatibility( + const Type& expected_type, const ::parquet::arrow::SchemaField& parquet_field); + /// \brief Project an Iceberg Schema onto a Parquet Schema. /// /// This function creates a projection from an Iceberg Schema to a Parquet schema. diff --git a/src/iceberg/test/CMakeLists.txt b/src/iceberg/test/CMakeLists.txt index f23bc9181..c6bf5de71 100644 --- a/src/iceberg/test/CMakeLists.txt +++ b/src/iceberg/test/CMakeLists.txt @@ -216,6 +216,7 @@ if(ICEBERG_BUILD_BUNDLE) SOURCES metrics_test_base.cc parquet_data_test.cc + parquet_row_group_filter_test.cc parquet_metrics_test.cc parquet_schema_test.cc parquet_test.cc) diff --git a/src/iceberg/test/file_scan_task_reader_test.cc b/src/iceberg/test/file_scan_task_reader_test.cc index cca13ce0d..c64aa224e 100644 --- a/src/iceberg/test/file_scan_task_reader_test.cc +++ b/src/iceberg/test/file_scan_task_reader_test.cc @@ -43,6 +43,8 @@ #include "iceberg/data/position_delete_writer.h" #include "iceberg/deletes/dv_writer.h" #include "iceberg/deletes/position_delete_index.h" +#include "iceberg/expression/binder.h" +#include "iceberg/expression/expressions.h" #include "iceberg/file_format.h" #include "iceberg/file_io.h" #include "iceberg/file_reader.h" @@ -676,4 +678,76 @@ TEST_F(FileScanTaskReaderTest, OpenWithMixedDeletesSkipsFullyDeletedBatches) { ASSERT_NO_FATAL_FAILURE(VerifyStream(&stream, R"([[3, "Baz"]])")); } +TEST_F(FileScanTaskReaderTest, RowGroupPruningPreservesDeletesAndLineage) { + ICEBERG_UNWRAP_OR_FAIL( + auto data_file, + MakeDataFile(table_schema_, + std::vector{R"([[0,"a","keep"],[1,"b","keep"]])", + R"([[2,"c","skip"],[3,"d","skip"]])", + R"([[4,"e","keep"],[5,"f","keep"]])"}, + 6, 2)); + data_file->first_row_id = 100; + data_file->data_sequence_number = 5; + ICEBERG_UNWRAP_OR_FAIL( + auto pos_delete, MakePositionDeleteFile(CreateNewTempFilePathWithSuffix(".parquet"), + {4}, data_file->file_path)); + ICEBERG_UNWRAP_OR_FAIL( + auto dv, MakeDeletionVectorFile(CreateNewTempFilePathWithSuffix(".puffin"), {4}, + data_file->file_path)); + ICEBERG_UNWRAP_OR_FAIL( + auto eq_delete, MakeEqualityDeleteFile(CreateNewTempFilePathWithSuffix(".parquet"), + table_schema_, R"([[4,"e","keep"]])", {1})); + for (auto delete_file : {pos_delete, dv, eq_delete}) { + FileScanTask task(data_file, {delete_file}, + Expressions::Equal("category", Literal::String("keep"))); + FileScanTaskReader::Options options{ + .io = file_io_, + .table_schema = table_schema_, + .schemas = {table_schema_}, + .projected_schema = RowLineageProjection(), + }; + ICEBERG_UNWRAP_OR_FAIL(auto reader, FileScanTaskReader::Make(std::move(options))); + ICEBERG_UNWRAP_OR_FAIL(auto stream, reader->Open(task)); + auto batches = ::arrow::ImportRecordBatchReader(&stream).ValueOrDie(); + std::vector actual_ids; + std::vector lineage; + while (true) { + auto batch = batches->Next().ValueOrDie(); + if (!batch) break; + auto ids = std::static_pointer_cast<::arrow::Int32Array>(batch->column(0)); + auto row_ids = std::static_pointer_cast<::arrow::Int64Array>(batch->column(1)); + for (int64_t i = 0; i < batch->num_rows(); ++i) { + actual_ids.push_back(ids->Value(i)); + lineage.push_back(row_ids->Value(i)); + } + } + EXPECT_EQ(actual_ids, (std::vector{0, 1, 5})); + EXPECT_EQ(lineage, (std::vector{100, 101, 105})); + } +} + +TEST_F(FileScanTaskReaderTest, RowGroupPruningUsesUnprojectedFilterWithoutDeletes) { + ICEBERG_UNWRAP_OR_FAIL( + auto data_file, + MakeDataFile(table_schema_, + std::vector{R"([[0,"a","skip"],[1,"b","skip"]])", + R"([[2,"c","keep"],[3,"d","keep"]])"}, + 4, 2)); + auto filter = Expressions::Equal("category", Literal::String("keep")); + ICEBERG_UNWRAP_OR_FAIL(auto bound, Binder::Bind(*table_schema_, filter, true)); + for (const auto& predicate : std::vector>{ + filter, bound, Expressions::Equal("CATEGORY", Literal::String("keep"))}) { + FileScanTask task(data_file, {}, predicate); + FileScanTaskReader::Options options{ + .io = file_io_, + .table_schema = table_schema_, + .projected_schema = projected_schema_, + }; + options.properties["read.filter.case-sensitive"] = "false"; + ICEBERG_UNWRAP_OR_FAIL(auto reader, FileScanTaskReader::Make(std::move(options))); + ICEBERG_UNWRAP_OR_FAIL(auto stream, reader->Open(task)); + VerifyStream(&stream, R"([[2,"c"],[3,"d"]])"); + } +} + } // namespace iceberg diff --git a/src/iceberg/test/parquet_row_group_filter_test.cc b/src/iceberg/test/parquet_row_group_filter_test.cc new file mode 100644 index 000000000..f3c81c20f --- /dev/null +++ b/src/iceberg/test/parquet_row_group_filter_test.cc @@ -0,0 +1,549 @@ +/* + * 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 "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/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" + +namespace iceberg::parquet { +namespace { + +class ParquetRowGroupFilterTest : public ::testing::Test { + protected: + void SetUp() override { + parquet::RegisterAll(); + io_ = std::make_shared(); + schema_ = std::make_shared( + std::vector{SchemaField::MakeOptional(1, "key", int32()), + SchemaField::MakeRequired(2, "value", int64())}); + projection_ = std::make_shared( + std::vector{SchemaField::MakeRequired(2, "value", int64()), + MetadataColumns::kRowPosition, MetadataColumns::kRowId}); + } + + Status Write(bool statistics = true, + 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 WriteTable(table, statistics); + } + + Status WriteTable(const std::shared_ptr<::arrow::Table>& table, bool statistics) { + ICEBERG_ASSIGN_OR_RAISE(auto out, arrow::OpenArrowOutputStream(io_, path_)); + ::parquet::WriterProperties::Builder properties; + properties.disable_dictionary(); + if (!statistics) { + properties.disable_statistics(); + } + ICEBERG_ARROW_RETURN_NOT_OK(::parquet::arrow::WriteTable( + *table, ::arrow::default_memory_pool(), out, 2, 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) { + // Direct readers need bound references for columns outside the projection. + // Leave invalid or already-bound expressions intact for their dedicated tests. + if (filter) { + auto bound = Binder::Bind(*schema_, filter, case_sensitive); + if (bound) { + filter = *bound; + } + } + ReaderOptions options{.path = path_, + .io = io_, + .projection = projection_, + .filter = std::move(filter), + .first_row_id = 100}; + options.properties.Set(ReaderProperties::kFilterCaseSensitive, 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) { + ICEBERG_UNWRAP_OR_FAIL(auto actual, Read(options)); + EXPECT_EQ(actual, expected); + } + + std::string path_ = "rg.parquet"; + std::shared_ptr io_; + std::shared_ptr schema_; + std::shared_ptr projection_; + std::shared_ptr<::parquet::FileMetaData> metadata_; +}; + +TEST_F(ParquetRowGroupFilterTest, NonContiguousGroupsKeepPhysicalPositions) { + ASSERT_THAT(Write(), IsOk()); + ASSERT_EQ(metadata_->num_row_groups(), 3); + auto filter = Expressions::Or(Expressions::LessThan("key", Literal::Int(2)), + Expressions::GreaterThanOrEqual("key", Literal::Int(20))); + Check(Options(filter), {0, 1, 4, 5}); + auto options = Options(filter); + options.properties.Set(ReaderProperties::kParquetRowGroupFilter, false); + Check(options, {0, 1, 2, 3, 4, 5}); +} + +TEST_F(ParquetRowGroupFilterTest, BatchSizesAcrossMultiplePhysicalGaps) { + ASSERT_THAT(Write(true, + "[[0,0],[1,1],[10,2],[11,3],[20,4],[21,5]," + "[30,6],[31,7],[40,8],[41,9]]"), + IsOk()); + ASSERT_EQ(metadata_->num_row_groups(), 5); + auto filter = Expressions::Or(Expressions::LessThan("key", Literal::Int(2)), + Expressions::GreaterThanOrEqual("key", Literal::Int(20))); + // Retain RG0 and RG2..4: a physical gap followed by multiple adjacent groups. + for (int64_t batch_size : {1, 2, 3, 8, 32}) { + SCOPED_TRACE(batch_size); + auto options = Options(filter); + options.properties.Set(ReaderProperties::kBatchSize, batch_size); + Check(options, {0, 1, 4, 5, 6, 7, 8, 9}); + + // With RG3 also removed, verify positions across two physical gaps. + options.filter = Expressions::And( + filter, + Expressions::Or(Expressions::LessThan("key", Literal::Int(30)), + Expressions::GreaterThanOrEqual("key", Literal::Int(40)))); + ICEBERG_UNWRAP_OR_FAIL(options.filter, Binder::Bind(*schema_, options.filter, true)); + Check(options, {0, 1, 4, 5, 8, 9}); + } +} + +TEST_F(ParquetRowGroupFilterTest, BatchesStayWithinRowGroups) { + ASSERT_THAT(Write(), IsOk()); + auto options = Options(nullptr); + options.properties.Set(ReaderProperties::kBatchSize, int64_t{32}); + for (bool filtered : {false, true}) { + SCOPED_TRACE(filtered); + if (filtered) { + options.filter = + Expressions::Or(Expressions::LessThan("key", Literal::Int(2)), + Expressions::GreaterThanOrEqual("key", Literal::Int(20))); + ICEBERG_UNWRAP_OR_FAIL(options.filter, + Binder::Bind(*schema_, options.filter, true)); + } + ICEBERG_UNWRAP_OR_FAIL( + auto reader, ReaderFactoryRegistry::Open(FileFormatType::kParquet, options)); + std::vector batch_sizes; + while (true) { + ICEBERG_UNWRAP_OR_FAIL(auto array, reader->Next()); + if (!array) { + break; + } + batch_sizes.push_back(array->length); + array->release(&*array); + } + EXPECT_EQ(batch_sizes, + filtered ? std::vector({2, 2}) : std::vector({2, 2, 2})); + ICEBERG_UNWRAP_OR_FAIL(auto end, reader->Next()); + EXPECT_FALSE(end.has_value()); + ASSERT_THAT(reader->Close(), IsOk()); + } +} + +TEST_F(ParquetRowGroupFilterTest, AllNoneAndResidualRows) { + ASSERT_THAT(Write(), IsOk()); + Check(Options(nullptr), {0, 1, 2, 3, 4, 5}); + Check(Options(True::Instance()), {0, 1, 2, 3, 4, 5}); + Check(Options(False::Instance()), {}); + Check(Options(Expressions::Equal("key", Literal::Int(100))), {}); + // RG-only: key=1 is retained with key=0. This is not an exact row filter. + Check(Options(Expressions::Equal("key", Literal::Int(0))), {0, 1}); + Check(Options(Expressions::And(Expressions::GreaterThan("key", Literal::Int(9)), + Expressions::LessThanOrEqual("key", Literal::Int(11)))), + {2, 3}); +} + +TEST_F(ParquetRowGroupFilterTest, RenameBoundPredicateAndCaseSensitivity) { + ASSERT_THAT(Write(), IsOk()); + schema_ = std::make_shared(std::vector{ + SchemaField::MakeOptional(1, "renamed", int32()), schema_->fields()[1]}); + auto filter = Expressions::Equal("renamed", Literal::Int(20)); + Check(Options(filter), {4, 5}); + ICEBERG_UNWRAP_OR_FAIL(auto bound, Binder::Bind(*schema_, filter, true)); + auto options = Options(bound); + Check(options, {4, 5}); + options = Options(Expressions::Equal("RENAMED", Literal::Int(20)), false); + options.properties.Set(ReaderProperties::kFilterCaseSensitive, false); + Check(options, {4, 5}); +} + +TEST_F(ParquetRowGroupFilterTest, MissingStatsUnknownFieldsAndTypePromotion) { + ASSERT_THAT(Write(false), IsOk()); + Check(Options(Expressions::Equal("key", Literal::Int(100))), {0, 1, 2, 3, 4, 5}); + Check(Options(Expressions::IsNull("key")), {0, 1, 2, 3, 4, 5}); + Check(Options(Expressions::NotNull("key")), {0, 1, 2, 3, 4, 5}); + ASSERT_THAT(Write(), IsOk()); + Check(Options(Expressions::Equal("missing", Literal::Int(0))), {0, 1, 2, 3, 4, 5}); + auto options = Options(Expressions::Equal("key", Literal::Int(0))); + options.filter = Expressions::Equal("key", Literal::Int(0)); + Check(options, {0, 1, 2, 3, 4, 5}); + schema_ = std::make_shared(std::vector{ + SchemaField::MakeOptional(1, "key", int64()), schema_->fields()[1], + SchemaField::MakeOptional(3, "defaulted", int32()) + .WithInitialDefault(std::make_shared(Literal::Int(7)))}); + Check(Options(Expressions::Equal("key", Literal::Long(100))), {}); + Check(Options(Expressions::Equal("key", Literal::Long(10))), {2, 3}); + Check(Options(Expressions::Equal("defaulted", Literal::Int(8))), {0, 1, 2, 3, 4, 5}); +} + +TEST_F(ParquetRowGroupFilterTest, FileAndFilterTypeCompatibility) { + struct Case { + std::shared_ptr file_type; + std::shared_ptr filter_type; + std::string json; + Literal value; + bool compatible; + }; + const std::string numbers = "[[0,0],[1,1],[10,2],[11,3],[20,4],[21,5]]"; + const std::string decimals = + R"([["0.00",0],["1.00",1],["10.00",2],["11.00",3],["20.00",4],["21.00",5]])"; + const std::string timestamps = + R"([["1970-01-01 00:00:00",0],["1970-01-01 00:00:01",1],["1970-01-01 00:00:10",2],["1970-01-01 00:00:11",3],["1970-01-01 00:00:20",4],["1970-01-01 00:00:21",5]])"; + for (const auto& test : std::vector{ + {float32(), float64(), numbers, Literal::Double(10), true}, + {float64(), float32(), numbers, Literal::Float(10), false}, + {int64(), int32(), numbers, Literal::Int(10), false}, + {int32(), date(), numbers, Literal::Date(10), false}, + {date(), int32(), numbers, Literal::Int(10), false}, + {decimal(9, 2), decimal(18, 2), decimals, Literal::Decimal(1000, 18, 2), true}, + {decimal(9, 2), decimal(8, 2), decimals, Literal::Decimal(1000, 8, 2), false}, + {decimal(9, 2), decimal(9, 3), decimals, Literal::Decimal(1000, 9, 3), false}, + {timestamp(), timestamp_ns(), timestamps, Literal::TimestampNs(10000000), + false}, + {timestamp(), timestamp_tz(), timestamps, Literal::TimestampTz(10000000), + false}, + {timestamp_ns(), timestamp_ns(), timestamps, Literal::TimestampNs(10000000000), + true}, + {timestamp_tz(), timestamp_tz(), timestamps, Literal::TimestampTz(10000000), + true}, + {fixed(1), fixed(1), R"([["a",0],["b",1],["m",2],["n",3],["y",4],["z",5]])", + Literal::Fixed({'m'}), true}, + {fixed(1), fixed(2), R"([["a",0],["b",1],["m",2],["n",3],["y",4],["z",5]])", + Literal::Fixed({'m', 'm'}), false}}) { + SCOPED_TRACE(test.file_type->ToString() + " -> " + test.filter_type->ToString()); + schema_ = std::make_shared(std::vector{ + SchemaField::MakeOptional(1, "key", test.file_type), schema_->fields()[1]}); + ASSERT_THAT(Write(true, test.json), IsOk()); + schema_ = std::make_shared(std::vector{ + SchemaField::MakeOptional(1, "key", test.filter_type), schema_->fields()[1]}); + Check(Options(Expressions::Equal("key", test.value)), + test.compatible ? std::vector{2, 3} + : std::vector{0, 1, 2, 3, 4, 5}); + } +} + +TEST_F(ParquetRowGroupFilterTest, NullNegativeAndUnsupportedBooleanBranches) { + ASSERT_THAT(Write(true, "[[null,0],[null,1],[10,2],[11,3],[20,4],[21,5]]"), IsOk()); + Check(Options(Expressions::IsNull("key")), {0, 1}); + Check(Options(Expressions::NotNull("key")), {2, 3, 4, 5}); + Check(Options(Expressions::NotEqual("key", Literal::Int(10))), {0, 1, 2, 3, 4, 5}); + auto unsupported = Expressions::Equal("missing", Literal::Int(0)); + ICEBERG_UNWRAP_OR_FAIL( + auto supported, + Binder::Bind(*schema_, Expressions::Equal("key", Literal::Int(20)), true)); + Check(Options(Expressions::And(supported, unsupported)), {4, 5}); + Check(Options(Expressions::Or(supported, unsupported)), {0, 1, 2, 3, 4, 5}); + Check(Options(Expressions::Not(unsupported)), {0, 1, 2, 3, 4, 5}); + Check(Options(Expressions::Not(supported)), {0, 1, 2, 3, 4, 5}); +} + +TEST_F(ParquetRowGroupFilterTest, RewriteNotBeforeBinding) { + ASSERT_THAT(Write(), IsOk()); + auto less_than = Expressions::LessThan("key", Literal::Int(10)); + auto greater_than = Expressions::GreaterThan("key", Literal::Int(11)); + Check(Options(Expressions::Not(less_than)), {2, 3, 4, 5}); + Check(Options(Expressions::Not(Expressions::Or(less_than, greater_than))), {2, 3}); + Check(Options(Expressions::Not( + Expressions::And(Expressions::GreaterThanOrEqual("key", Literal::Int(10)), + Expressions::LessThanOrEqual("key", Literal::Int(11))))), + {0, 1, 4, 5}); + Check(Options(Expressions::Not(Expressions::Not(less_than))), {0, 1}); + Check(Options(Expressions::Not(Expressions::NotEqual("key", Literal::Int(10)))), + {2, 3}); + Check(Options(Expressions::Not(Expressions::NotIn("key", {Literal::Int(10)}))), {2, 3}); + + ICEBERG_UNWRAP_OR_FAIL(auto bound, Binder::Bind(*schema_, less_than, true)); + Check(Options(Expressions::Not(bound)), {2, 3, 4, 5}); + + // Rewrite first: weakening an unknown predicate before NOT could discard rows. + auto unknown = Expressions::Equal("missing", Literal::Int(0)); + Check(Options(Expressions::Not(Expressions::Or(bound, unknown))), {2, 3, 4, 5}); + Check(Options(Expressions::Not(Expressions::And(bound, unknown))), {0, 1, 2, 3, 4, 5}); + + ASSERT_THAT(Write(true, "[[null,0],[null,1],[10,2],[11,3],[20,4],[21,5]]"), IsOk()); + Check(Options(Expressions::Not(Expressions::IsNull("key"))), {2, 3, 4, 5}); + Check(Options(Expressions::Not(Expressions::NotNull("key"))), {0, 1}); +} + +TEST_F(ParquetRowGroupFilterTest, BoundComparisonBoundariesAndAllNullGroups) { + ASSERT_THAT(Write(true, "[[null,0],[null,1],[10,2],[11,3],[20,4],[21,5]]"), IsOk()); + struct Case { + std::shared_ptr filter; + std::vector expected; + }; + for (const auto& test : std::vector{ + {Expressions::LessThan("key", Literal::Int(10)), {}}, + {Expressions::LessThanOrEqual("key", Literal::Int(10)), {2, 3}}, + {Expressions::GreaterThan("key", Literal::Int(21)), {}}, + {Expressions::GreaterThanOrEqual("key", Literal::Int(21)), {4, 5}}, + {Expressions::Equal("key", Literal::Int(11)), {2, 3}}, + {Expressions::In("key", {Literal::Int(11), Literal::Int(20)}), {2, 3, 4, 5}}, + }) { + ICEBERG_UNWRAP_OR_FAIL(auto bound, Binder::Bind(*schema_, test.filter, true)); + Check(Options(bound), test.expected); + } +} + +TEST_F(ParquetRowGroupFilterTest, InPredicateLimit) { + ASSERT_THAT(Write(), IsOk()); + Check(Options(Expressions::In("key", {Literal::Int(0), Literal::Int(20)})), + {0, 1, 4, 5}); + std::vector values; + for (int i = 100; i < 200; ++i) { + values.push_back(Literal::Int(i)); + } + Check(Options(Expressions::In("key", values)), {}); + for (int i = 200; i < 300; ++i) { + values.push_back(Literal::Int(i)); + } + Check(Options(Expressions::In("key", values)), {}); + // The limit applies to the bound set, not the unbound list with duplicates. + auto duplicate_values = values; + duplicate_values.push_back(values.back()); + auto options = Options(nullptr); + auto fields = projection_->fields(); + std::vector projected_fields(fields.begin(), fields.end()); + projected_fields.push_back(schema_->fields()[0]); + options.projection = std::make_shared(std::move(projected_fields)); + options.filter = Expressions::In("key", duplicate_values); + Check(options, {}); + + values.push_back(Literal::Int(300)); + Check(Options(Expressions::In("key", values)), {0, 1, 2, 3, 4, 5}); + Check(Options(Expressions::And(Expressions::In("key", values), + Expressions::Equal("key", Literal::Int(20)))), + {4, 5}); +} + +TEST_F(ParquetRowGroupFilterTest, LargeExpressionStillPrunes) { + ASSERT_THAT(Write(), IsOk()); + auto filter = Expressions::Equal("key", Literal::Int(100)); + std::shared_ptr large = filter; + for (int i = 0; i < 260; ++i) { + large = Expressions::Or(large, filter); + } + Check(Options(large), {}); +} + +TEST_F(ParquetRowGroupFilterTest, SplitIntersectionAndEmptySplit) { + ASSERT_THAT(Write(), IsOk()); + auto offset = metadata_->RowGroup(1)->file_offset(); + ASSERT_GT(offset, metadata_->RowGroup(0)->file_offset()); + auto options = Options(Expressions::GreaterThanOrEqual("key", Literal::Int(20))); + options.split = Split{static_cast(offset), 100000}; + Check(options, {4, 5}); + ICEBERG_UNWRAP_OR_FAIL( + options.filter, + Binder::Bind(*schema_, Expressions::LessThan("key", Literal::Int(2)), true)); + Check(options, {}); + options.filter = nullptr; + options.split = Split{static_cast(offset), 0}; + Check(options, {}); +} + +TEST_F(ParquetRowGroupFilterTest, PrimitiveComparisonTypes) { + struct Case { + std::shared_ptr type; + std::string json; + Literal literal; + }; + for (const auto& test : std::vector{ + {int64(), "[[0,0],[1,1],[10,2],[11,3],[20,4],[21,5]]", Literal::Long(10)}, + {date(), "[[0,0],[1,1],[10,2],[11,3],[20,4],[21,5]]", Literal::Date(10)}, + {string(), R"([["a",0],["b",1],["m",2],["n",3],["y",4],["z",5]])", + Literal::String("m")}, + {boolean(), "[[false,0],[false,1],[true,2],[true,3],[false,4],[false,5]]", + Literal::Boolean(true)}, + {binary(), R"([["a",0],["b",1],["m",2],["n",3],["y",4],["z",5]])", + Literal::Binary({'m'})}}) { + schema_ = std::make_shared( + std::vector{SchemaField::MakeOptional(1, "key", test.type), + SchemaField::MakeRequired(2, "value", int64())}); + ASSERT_THAT(Write(true, test.json), IsOk()); + Check(Options(Expressions::Equal("key", test.literal)), {2, 3}); + } +} + +TEST_F(ParquetRowGroupFilterTest, FloatAndNestedStatsWithTransformFallback) { + schema_ = std::make_shared( + std::vector{SchemaField::MakeOptional(1, "key", float64()), + SchemaField::MakeRequired(2, "value", int64())}); + ASSERT_THAT(Write(), IsOk()); + Check(Options(Expressions::Equal("key", Literal::Double(100))), {}); + Check(Options(Expressions::Equal("key", Literal::Double(10))), {2, 3}); + Check(Options(Expressions::IsNaN("key")), {0, 1, 2, 3, 4, 5}); + schema_ = std::make_shared(std::vector{ + SchemaField::MakeOptional(1, "key", + std::make_shared(std::vector{ + SchemaField::MakeOptional(3, "nested", int32())})), + SchemaField::MakeRequired(2, "value", int64())}); + ASSERT_THAT(Write(true, "[[[0],0],[[1],1],[[10],2],[[11],3],[[20],4],[[21],5]]"), + IsOk()); + Check(Options(Expressions::Equal("key.nested", Literal::Int(100))), {}); + Check(Options(Expressions::Equal("key.nested", Literal::Int(10))), {2, 3}); + schema_ = std::make_shared( + std::vector{SchemaField::MakeOptional(1, "key", int32()), + SchemaField::MakeRequired(2, "value", int64())}); + ASSERT_THAT(Write(), IsOk()); + Check(Options(Expressions::Equal(Expressions::Bucket("key", 16), + Literal::Int(15))), + {0, 1, 2, 3, 4, 5}); +} + +TEST_F(ParquetRowGroupFilterTest, NegativeAndPrefixPredicates) { + ASSERT_THAT(Write(true, "[[5,0],[5,1],[null,2],[5,3],[6,4],[6,5]]"), IsOk()); + Check(Options(Expressions::NotEqual("key", Literal::Int(5))), {0, 1, 2, 3, 4, 5}); + Check(Options(Expressions::NotIn("key", {Literal::Int(5), Literal::Int(6)})), + {0, 1, 2, 3, 4, 5}); + Check(Options(Expressions::Not(Expressions::LessThan("key", Literal::Int(100)))), {}); + schema_ = std::make_shared(std::vector{ + SchemaField::MakeOptional(1, "key", string()), schema_->fields()[1]}); + ASSERT_THAT( + Write( + true, + R"([["apple",0],["apricot",1],["banana",2],["blueberry",3],[null,4],["apricot",5]])"), + IsOk()); + Check(Options(Expressions::StartsWith("key", "ap")), {0, 1, 4, 5}); + Check(Options(Expressions::NotStartsWith("key", "ap")), {2, 3, 4, 5}); +} + +TEST_F(ParquetRowGroupFilterTest, DecimalAndTemporalStatistics) { + struct Case { + std::shared_ptr type; + std::string json; + Literal value; + }; + for ( + const auto& test : std::vector{ + {decimal(9, 2), + R"([["0.00",0],["1.00",1],["10.00",2],["11.00",3],["20.00",4],["21.00",5]])", + Literal::Decimal(1000, 9, 2)}, + {time(), "[[0,0],[1,1],[10,2],[11,3],[20,4],[21,5]]", Literal::Time(10)}, + {timestamp(), + R"([["1970-01-01 00:00:00",0],["1970-01-01 00:00:01",1],["1970-01-01 00:00:10",2],["1970-01-01 00:00:11",3],["1970-01-01 00:00:20",4],["1970-01-01 00:00:21",5]])", + Literal::Timestamp(10000000)}}) { + schema_ = std::make_shared( + std::vector{SchemaField::MakeOptional(1, "key", test.type), + SchemaField::MakeRequired(2, "value", int64())}); + ASSERT_THAT(Write(true, test.json), IsOk()); + Check(Options(Expressions::Equal("key", test.value)), {2, 3}); + Check(Options(Expressions::LessThan("key", test.value)), {0, 1}); + } +} + +TEST_F(ParquetRowGroupFilterTest, FloatingPointNullNaNAndSignedZero) { + schema_ = std::make_shared(std::vector{ + SchemaField::MakeOptional(1, "key", float64()), schema_->fields()[1]}); + ASSERT_THAT(Write(true, "[[-0.0,0],[0.0,1],[null,2],[null,3],[10.0,4],[11.0,5]]"), + IsOk()); + Check(Options(Expressions::Equal("key", Literal::Double(-0.0))), {0, 1}); + Check(Options(Expressions::Equal("key", Literal::Double(0.0))), {0, 1}); + Check(Options(Expressions::IsNull("key")), {2, 3}); + Check(Options(Expressions::NotNull("key")), {0, 1, 4, 5}); + Check(Options(Expressions::IsNaN("key")), {0, 1, 4, 5}); + Check(Options(Expressions::LessThan("key", Literal::Double(-100))), {}); + ASSERT_THAT(Write(true, "[[NaN,0],[-NaN,1],[10.0,2],[11.0,3],[null,4],[null,5]]"), + IsOk()); + Check(Options(Expressions::Equal("key", Literal::Double(100))), {0, 1}); + Check(Options(Expressions::IsNaN("key")), {0, 1, 2, 3}); + Check(Options(Expressions::LessThan("key", Literal::Double(-100))), {0, 1}); +} + +TEST_F(ParquetRowGroupFilterTest, InvalidFilterFailsDuringOpen) { + ASSERT_THAT(Write(), IsOk()); + auto options = Options(Expressions::Count("key")); + options.filter = Expressions::Count("key"); + EXPECT_THAT(ReaderFactoryRegistry::Open(FileFormatType::kParquet, options), + HasErrorMessage("does not support unbound aggregate")); + + ICEBERG_UNWRAP_OR_FAIL(auto bound, Binder::Bind(*schema_, options.filter, true)); + auto bound_options = Options(bound); + EXPECT_THAT(ReaderFactoryRegistry::Open(FileFormatType::kParquet, bound_options), + HasErrorMessage("does not support bound aggregate")); + + options.properties.Set(ReaderProperties::kParquetRowGroupFilter, false); + Check(options, {0, 1, 2, 3, 4, 5}); +} + +} // namespace +} // namespace iceberg::parquet diff --git a/src/iceberg/test/parquet_test.cc b/src/iceberg/test/parquet_test.cc index ec96167d1..f99ec3a1d 100644 --- a/src/iceberg/test/parquet_test.cc +++ b/src/iceberg/test/parquet_test.cc @@ -848,8 +848,13 @@ TEST_F(ParquetReaderTest, ReadSplit) { {.offset = split_offsets[1] + 1, .length = std::numeric_limits::max()}, {.offset = 0, .length = split_offsets[0]}, }; - std::vector expected_json = { - R"([[1, 0], [2, 1], [3, 2]])", R"([[1, 0], [2, 1]])", R"([[3, 2]])", "", "", + // Each batch stays within one row group, including a split covering the whole file. + std::vector> expected_json = { + {R"([[1, 0], [2, 1]])", R"([[3, 2]])"}, + {R"([[1, 0], [2, 1]])"}, + {R"([[3, 2]])"}, + {}, + {}, }; ReaderProperties reader_properties; @@ -866,8 +871,8 @@ TEST_F(ParquetReaderTest, ReadSplit) { }); ASSERT_THAT(reader_result, IsOk()); auto reader = std::move(reader_result.value()); - if (!expected_json[i].empty()) { - ASSERT_NO_FATAL_FAILURE(VerifyNextBatch(*reader, expected_json[i])); + for (const auto& batch_json : expected_json[i]) { + ASSERT_NO_FATAL_FAILURE(VerifyNextBatch(*reader, batch_json)); } ASSERT_NO_FATAL_FAILURE(VerifyExhausted(*reader)); } From 64cc97cf7d3d97709a62bc11260473470a116844 Mon Sep 17 00:00:00 2001 From: yinzhengsun Date: Mon, 21 Sep 2026 23:20:55 +0800 Subject: [PATCH 2/5] refactor --- src/iceberg/data/file_scan_task_reader.h | 2 +- .../parquet_metrics_row_group_filter.cc | 154 +++++++----------- ...arquet_metrics_row_group_filter_internal.h | 9 +- src/iceberg/parquet/parquet_reader.cc | 45 ++--- .../test/parquet_row_group_filter_test.cc | 41 +++-- 5 files changed, 119 insertions(+), 132 deletions(-) diff --git a/src/iceberg/data/file_scan_task_reader.h b/src/iceberg/data/file_scan_task_reader.h index daef86d5c..a71ef5f84 100644 --- a/src/iceberg/data/file_scan_task_reader.h +++ b/src/iceberg/data/file_scan_task_reader.h @@ -48,7 +48,7 @@ class ICEBERG_DATA_EXPORT FileScanTaskReader { struct Options { /// FileIO instance for reading data and delete files. std::shared_ptr io; - /// The table schema used to bind filters and resolve delete-file fields. + /// The table schema. Used as the primary field lookup for delete file resolution. std::shared_ptr table_schema; /// Optional list of historical table schemas for field lookup. std::vector> schemas; diff --git a/src/iceberg/parquet/parquet_metrics_row_group_filter.cc b/src/iceberg/parquet/parquet_metrics_row_group_filter.cc index 735892f2d..3d245a0e3 100644 --- a/src/iceberg/parquet/parquet_metrics_row_group_filter.cc +++ b/src/iceberg/parquet/parquet_metrics_row_group_filter.cc @@ -22,7 +22,6 @@ #include -#include "iceberg/expression/binder.h" #include "iceberg/expression/expression_visitor.h" #include "iceberg/expression/rewrite_not.h" #include "iceberg/metadata_columns.h" @@ -30,7 +29,6 @@ #include "iceberg/parquet/parquet_metrics_internal.h" #include "iceberg/parquet/parquet_metrics_row_group_filter_internal.h" #include "iceberg/parquet/parquet_schema_util_internal.h" -#include "iceberg/schema.h" #include "iceberg/type.h" #include "iceberg/util/macros.h" @@ -39,69 +37,22 @@ namespace iceberg::parquet { namespace { constexpr size_t kInPredicateLimit = 200; - -class BindFilter : public Binder { - public: - BindFilter(const Schema& schema, bool case_sensitive, - const ::parquet::arrow::SchemaManifest& manifest) - : Binder(schema, case_sensitive) { - for (const auto& [index, field] : manifest.column_index_to_field) { - const auto* column = manifest.descr->Column(index); - // Repeated-column counts and bounds do not describe individual rows. - if (column->max_repetition_level() == 0) { - fields_.emplace(column->schema_node()->field_id(), field); - } - } - } - - Result> Predicate( - const std::shared_ptr& pred) override { - auto bound = Binder::Predicate(pred); - // Direct format readers may only supply a projection, not the complete schema. - if (!bound) { - return True::Instance(); - } - return Visit>(*bound, *this); - } - - Result> Predicate( - const std::shared_ptr& pred) override { - auto ref = std::dynamic_pointer_cast(pred->term()); - if (!ref || !ref->type()->is_primitive() || - MetadataColumns::IsMetadataColumn(ref->field_id()) || - MetadataColumns::IsRowLineageColumn(ref->field_id())) { - return True::Instance(); - } - auto field = fields_.find(ref->field_id()); - if (field == fields_.end() || - !ValidateParquetTypeCompatibility(*ref->type(), *field->second)) { - return True::Instance(); - } - return pred; - } - - private: - std::unordered_map fields_; -}; +// True means a matching row may exist; false means the group can be skipped. +constexpr bool kRowsMightMatch = true; +constexpr bool kRowsCannotMatch = false; class MetricsVisitor : public BoundVisitor { public: - MetricsVisitor(const ::parquet::SchemaDescriptor& schema, - const ::parquet::RowGroupMetaData& row_group) - : schema_(schema), row_group_(row_group) { - for (int i = 0; i < schema.num_columns(); ++i) { - auto id = schema.Column(i)->schema_node()->field_id(); - if (id >= 0) { - columns_.emplace(id, i); - } - } - } + MetricsVisitor(const ::parquet::arrow::SchemaManifest& manifest, + const ::parquet::RowGroupMetaData& row_group, + const std::unordered_map& column_indices) + : manifest_(manifest), row_group_(row_group), column_indices_(column_indices) {} - Result AlwaysTrue() override { return true; } + Result AlwaysTrue() override { return kRowsMightMatch; } - Result AlwaysFalse() override { return false; } + Result AlwaysFalse() override { return kRowsCannotMatch; } - Result Not(bool) override { return true; } + Result Not(bool) override { return kRowsMightMatch; } Result And(bool left, bool right) override { return left && right; } @@ -119,15 +70,15 @@ class MetricsVisitor : public BoundVisitor { return !ContainsNullsOnly(GetMetrics(expr, false)); } - Result NotNaN(const std::shared_ptr&) override { return true; } + Result NotNaN(const std::shared_ptr&) override { return kRowsMightMatch; } Result Lt(const std::shared_ptr& expr, const Literal& value) override { const auto metrics = GetMetrics(expr); if (ContainsNullsOnly(metrics)) { - return false; + return kRowsCannotMatch; } if (!metrics.lower_bound || !ComparableLiteral(value)) { - return true; + return kRowsMightMatch; } return !(*metrics.lower_bound >= value); } @@ -135,10 +86,10 @@ class MetricsVisitor : public BoundVisitor { Result LtEq(const std::shared_ptr& expr, const Literal& value) override { const auto metrics = GetMetrics(expr); if (ContainsNullsOnly(metrics)) { - return false; + return kRowsCannotMatch; } if (!metrics.lower_bound || !ComparableLiteral(value)) { - return true; + return kRowsMightMatch; } return !(*metrics.lower_bound > value); } @@ -146,10 +97,10 @@ class MetricsVisitor : public BoundVisitor { Result Gt(const std::shared_ptr& expr, const Literal& value) override { const auto metrics = GetMetrics(expr); if (ContainsNullsOnly(metrics)) { - return false; + return kRowsCannotMatch; } if (!metrics.upper_bound || !ComparableLiteral(value)) { - return true; + return kRowsMightMatch; } return !(*metrics.upper_bound <= value); } @@ -157,10 +108,10 @@ class MetricsVisitor : public BoundVisitor { Result GtEq(const std::shared_ptr& expr, const Literal& value) override { const auto metrics = GetMetrics(expr); if (ContainsNullsOnly(metrics)) { - return false; + return kRowsCannotMatch; } if (!metrics.upper_bound || !ComparableLiteral(value)) { - return true; + return kRowsMightMatch; } return !(*metrics.upper_bound < value); } @@ -168,52 +119,52 @@ class MetricsVisitor : public BoundVisitor { Result Eq(const std::shared_ptr& expr, const Literal& value) override { const auto metrics = GetMetrics(expr); if (ContainsNullsOnly(metrics)) { - return false; + return kRowsCannotMatch; } if (!metrics.lower_bound || !metrics.upper_bound || !ComparableLiteral(value)) { - return true; + return kRowsMightMatch; } return !(*metrics.lower_bound > value || *metrics.upper_bound < value); } Result NotEq(const std::shared_ptr&, const Literal&) override { // Like Java, keep negative membership predicates inclusive. - return true; + return kRowsMightMatch; } Result In(const std::shared_ptr& expr, const BoundSetPredicate::LiteralSet& values) override { const auto metrics = GetMetrics(expr); if (ContainsNullsOnly(metrics)) { - return false; + return kRowsCannotMatch; } if (!metrics.lower_bound || !metrics.upper_bound || values.size() > kInPredicateLimit) { - return true; + return kRowsMightMatch; } for (const auto& value : values) { if (!ComparableLiteral(value) || !(value < *metrics.lower_bound || value > *metrics.upper_bound)) { - return true; + return kRowsMightMatch; } } - return false; + return kRowsCannotMatch; } Result NotIn(const std::shared_ptr&, const BoundSetPredicate::LiteralSet&) override { - return true; + return kRowsMightMatch; } Result StartsWith(const std::shared_ptr& expr, const Literal& value) override { const auto metrics = GetMetrics(expr); if (ContainsNullsOnly(metrics)) { - return false; + return kRowsCannotMatch; } if (!metrics.lower_bound || !metrics.upper_bound || !ComparableLiteral(value) || metrics.lower_bound->type()->type_id() != TypeId::kString) { - return true; + return kRowsMightMatch; } const auto& prefix = std::get(value.value()); const auto& lower = std::get(metrics.lower_bound->value()); @@ -228,7 +179,7 @@ class MetricsVisitor : public BoundVisitor { if (MayContainNull(metrics) || !metrics.lower_bound || !metrics.upper_bound || !ComparableLiteral(value) || metrics.lower_bound->type()->type_id() != TypeId::kString) { - return true; + return kRowsMightMatch; } const auto& prefix = std::get(value.value()); const auto& lower = std::get(metrics.lower_bound->value()); @@ -255,16 +206,25 @@ class MetricsVisitor : public BoundVisitor { bool read_bounds = true) const { FieldMetrics metrics; auto ref = std::dynamic_pointer_cast(expr); - if (!ref) { + if (!ref || !ref->type()->is_primitive() || + MetadataColumns::IsMetadataColumn(ref->field_id()) || + MetadataColumns::IsRowLineageColumn(ref->field_id())) { return metrics; } metrics.field_id = ref->field_id(); - auto column = columns_.find(ref->field_id()); + auto column = column_indices_.find(ref->field_id()); // Missing columns can have initial defaults, so do not assume all nulls. - if (column == columns_.end()) { + if (column == column_indices_.end()) { + return metrics; + } + const auto& descriptor = *manifest_.descr->Column(column->second); + auto field = manifest_.column_index_to_field.find(column->second); + // Repeated-column statistics describe elements rather than rows. + if (descriptor.max_repetition_level() != 0 || + field == manifest_.column_index_to_field.end() || + !ValidateParquetTypeCompatibility(*ref->type(), *field->second)) { return metrics; } - const auto& descriptor = *schema_.Column(column->second); const auto& type = static_cast(*ref->type()); auto chunk = row_group_.ColumnChunk(column->second); auto stats = chunk->statistics(); @@ -313,42 +273,44 @@ class MetricsVisitor : public BoundVisitor { return metrics; } - const ::parquet::SchemaDescriptor& schema_; + const ::parquet::arrow::SchemaManifest& manifest_; const ::parquet::RowGroupMetaData& row_group_; - std::unordered_map columns_; + const std::unordered_map& column_indices_; }; } // namespace Result> ParquetMetricsRowGroupFilter::Make( - const Schema& schema, const std::shared_ptr& filter, - const ::parquet::arrow::SchemaManifest& manifest, bool case_sensitive) { + const std::shared_ptr& filter, + const ::parquet::SchemaDescriptor& file_schema) { auto result = std::unique_ptr(new ParquetMetricsRowGroupFilter()); + for (int i = 0; i < file_schema.num_columns(); ++i) { + auto id = file_schema.Column(i)->schema_node()->field_id(); + if (id >= 0) { + result->column_indices_.emplace(id, i); + } + } result->bound_ = True::Instance(); if (!filter) { return result; } - // Eliminate NOT before unsupported predicates are weakened to true. - ICEBERG_ASSIGN_OR_RAISE(auto rewritten, RewriteNot::Visit(filter)); - BindFilter binder(schema, case_sensitive, manifest); - ICEBERG_ASSIGN_OR_RAISE(result->bound_, - Visit>(rewritten, binder)); + ICEBERG_ASSIGN_OR_RAISE(result->bound_, RewriteNot::Visit(filter)); return result; } Result ParquetMetricsRowGroupFilter::ShouldRead( - const ::parquet::SchemaDescriptor& file_schema, + const ::parquet::arrow::SchemaManifest& manifest, const ::parquet::RowGroupMetaData& row_group) const { if (row_group.num_rows() <= 0) { - return false; + return kRowsCannotMatch; } try { - MetricsVisitor visitor(file_schema, row_group); + MetricsVisitor visitor(manifest, row_group, column_indices_); return Visit(bound_, visitor); } catch (const ::parquet::ParquetException&) { // Unusable optional statistics must never turn into false negatives. - return true; + return kRowsMightMatch; } } diff --git a/src/iceberg/parquet/parquet_metrics_row_group_filter_internal.h b/src/iceberg/parquet/parquet_metrics_row_group_filter_internal.h index 2907dc35c..96703a838 100644 --- a/src/iceberg/parquet/parquet_metrics_row_group_filter_internal.h +++ b/src/iceberg/parquet/parquet_metrics_row_group_filter_internal.h @@ -19,7 +19,9 @@ #pragma once +#include #include +#include #include @@ -35,15 +37,16 @@ namespace iceberg::parquet { class ICEBERG_BUNDLE_EXPORT ParquetMetricsRowGroupFilter { public: static Result> Make( - const Schema& schema, const std::shared_ptr& filter, - const ::parquet::arrow::SchemaManifest& manifest, bool case_sensitive = true); + const std::shared_ptr& filter, + const ::parquet::SchemaDescriptor& file_schema); - Result ShouldRead(const ::parquet::SchemaDescriptor& file_schema, + Result ShouldRead(const ::parquet::arrow::SchemaManifest& manifest, const ::parquet::RowGroupMetaData& row_group) const; private: ParquetMetricsRowGroupFilter() = default; std::shared_ptr bound_; + std::unordered_map column_indices_; }; } // namespace iceberg::parquet diff --git a/src/iceberg/parquet/parquet_reader.cc b/src/iceberg/parquet/parquet_reader.cc index ac1eb2def..6bea88998 100644 --- a/src/iceberg/parquet/parquet_reader.cc +++ b/src/iceberg/parquet/parquet_reader.cc @@ -37,6 +37,7 @@ #include "iceberg/arrow/arrow_io_internal.h" #include "iceberg/arrow/arrow_status_internal.h" #include "iceberg/arrow/metadata_column_util_internal.h" +#include "iceberg/expression/binder.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" @@ -310,11 +311,19 @@ class ParquetReader::Impl { ICEBERG_ASSIGN_OR_RAISE(projection_, BuildProjection(reader_.get(), *read_schema_)); if (options.filter && options.properties.Get(ReaderProperties::kParquetRowGroupFilter)) { - ICEBERG_ASSIGN_OR_RAISE( - stats_filter_, - ParquetMetricsRowGroupFilter::Make( - *options.projection, options.filter, reader_->manifest(), - options.properties.Get(ReaderProperties::kFilterCaseSensitive))); + auto filter = options.filter; + if (filter->op() != Expression::Operation::kTrue && + filter->op() != Expression::Operation::kFalse) { + ICEBERG_ASSIGN_OR_RAISE(auto is_bound, IsBoundVisitor::IsBound(filter)); + if (!is_bound) { + ICEBERG_ASSIGN_OR_RAISE( + filter, Binder::Bind(*options.projection, filter, + options.properties.Get( + ReaderProperties::kFilterCaseSensitive))); + } + } + ICEBERG_ASSIGN_OR_RAISE(stats_filter_, ParquetMetricsRowGroupFilter::Make( + filter, *reader_->manifest().descr)); } metadata_context_ = {.file_path = options.path, @@ -409,7 +418,7 @@ class ParquetReader::Impl { private: Status InitReadContext() { - auto context = std::make_unique(); + context_ = std::make_unique(); auto metadata = reader_->parquet_reader()->metadata(); int64_t next_row_start = 0; @@ -427,7 +436,7 @@ class ParquetReader::Impl { } if (stats_filter_) { ICEBERG_ASSIGN_OR_RAISE( - auto should_read, stats_filter_->ShouldRead(*metadata->schema(), *row_group)); + auto should_read, stats_filter_->ShouldRead(reader_->manifest(), *row_group)); if (!should_read) { continue; } @@ -435,14 +444,14 @@ class ParquetReader::Impl { if (row_group->num_rows() == 0) { continue; } - context->row_groups_.push_back({i, row_start}); + context_->row_groups_.push_back({i, row_start}); } - if (context->row_groups_.empty()) { - context->record_batch_reader_ = std::make_unique(); + if (context_->row_groups_.empty()) { + context_->record_batch_reader_ = std::make_unique(); } else { ICEBERG_ARROW_ASSIGN_OR_RETURN( - context->record_batch_reader_, - reader_->GetRecordBatchReader({context->row_groups_.front().index}, + context_->record_batch_reader_, + reader_->GetRecordBatchReader({context_->row_groups_.front().index}, SelectedColumnIndices(projection_))); } @@ -451,7 +460,7 @@ class ParquetReader::Impl { // the schema of the file. ArrowSchema arrow_schema; ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(*read_schema_, &arrow_schema)); - ICEBERG_ARROW_ASSIGN_OR_RETURN(context->output_arrow_schema_, + ICEBERG_ARROW_ASSIGN_OR_RETURN(context_->output_arrow_schema_, ::arrow::ImportSchema(&arrow_schema)); // Align the output schema with the arrays the reader actually produces. The reader's @@ -462,15 +471,13 @@ class ParquetReader::Impl { // 3. Mixed list and large_list types in files with stored schemas // For each projected field, we use the reader's actual type. For missing fields // (columns not in the file), we apply the configured use_large_list preference. - context->output_arrow_schema_ = AlignOutputSchemaToReaderSchema( - context->output_arrow_schema_, context->record_batch_reader_->schema(), + context_->output_arrow_schema_ = AlignOutputSchemaToReaderSchema( + context_->output_arrow_schema_, context_->record_batch_reader_->schema(), projection_, use_large_list_); - // Publish the read state only after initialization succeeds. - if (!context->row_groups_.empty()) { - metadata_context_.next_file_pos = context->row_groups_.front().first_row; + if (!context_->row_groups_.empty()) { + metadata_context_.next_file_pos = context_->row_groups_.front().first_row; } - context_ = std::move(context); return {}; } diff --git a/src/iceberg/test/parquet_row_group_filter_test.cc b/src/iceberg/test/parquet_row_group_filter_test.cc index f3c81c20f..53773e4f1 100644 --- a/src/iceberg/test/parquet_row_group_filter_test.cc +++ b/src/iceberg/test/parquet_row_group_filter_test.cc @@ -237,16 +237,30 @@ TEST_F(ParquetRowGroupFilterTest, RenameBoundPredicateAndCaseSensitivity) { Check(options, {4, 5}); } +TEST_F(ParquetRowGroupFilterTest, OpenBindsUnboundProjectedReferences) { + ASSERT_THAT(Write(), IsOk()); + auto options = Options(nullptr); + options.filter = Expressions::GreaterThanOrEqual("VALUE", Literal::Long(4)); + EXPECT_THAT(ReaderFactoryRegistry::Open(FileFormatType::kParquet, options), + HasErrorMessage("Cannot find field 'VALUE'")); + options.properties.Set(ReaderProperties::kFilterCaseSensitive, false); + Check(options, {4, 5}); +} + TEST_F(ParquetRowGroupFilterTest, MissingStatsUnknownFieldsAndTypePromotion) { ASSERT_THAT(Write(false), IsOk()); Check(Options(Expressions::Equal("key", Literal::Int(100))), {0, 1, 2, 3, 4, 5}); Check(Options(Expressions::IsNull("key")), {0, 1, 2, 3, 4, 5}); Check(Options(Expressions::NotNull("key")), {0, 1, 2, 3, 4, 5}); ASSERT_THAT(Write(), IsOk()); - Check(Options(Expressions::Equal("missing", Literal::Int(0))), {0, 1, 2, 3, 4, 5}); + EXPECT_THAT(ReaderFactoryRegistry::Open( + FileFormatType::kParquet, + Options(Expressions::Equal("missing", Literal::Int(0)))), + HasErrorMessage("Cannot find field 'missing'")); auto options = Options(Expressions::Equal("key", Literal::Int(0))); options.filter = Expressions::Equal("key", Literal::Int(0)); - Check(options, {0, 1, 2, 3, 4, 5}); + EXPECT_THAT(ReaderFactoryRegistry::Open(FileFormatType::kParquet, options), + HasErrorMessage("Cannot find field 'key'")); schema_ = std::make_shared(std::vector{ SchemaField::MakeOptional(1, "key", int64()), schema_->fields()[1], SchemaField::MakeOptional(3, "defaulted", int32()) @@ -307,17 +321,16 @@ TEST_F(ParquetRowGroupFilterTest, NullNegativeAndUnsupportedBooleanBranches) { Check(Options(Expressions::IsNull("key")), {0, 1}); Check(Options(Expressions::NotNull("key")), {2, 3, 4, 5}); Check(Options(Expressions::NotEqual("key", Literal::Int(10))), {0, 1, 2, 3, 4, 5}); - auto unsupported = Expressions::Equal("missing", Literal::Int(0)); - ICEBERG_UNWRAP_OR_FAIL( - auto supported, - Binder::Bind(*schema_, Expressions::Equal("key", Literal::Int(20)), true)); + auto unsupported = + Expressions::Equal(Expressions::Bucket("key", 16), Literal::Int(0)); + auto supported = Expressions::Equal("key", Literal::Int(20)); Check(Options(Expressions::And(supported, unsupported)), {4, 5}); Check(Options(Expressions::Or(supported, unsupported)), {0, 1, 2, 3, 4, 5}); Check(Options(Expressions::Not(unsupported)), {0, 1, 2, 3, 4, 5}); Check(Options(Expressions::Not(supported)), {0, 1, 2, 3, 4, 5}); } -TEST_F(ParquetRowGroupFilterTest, RewriteNotBeforeBinding) { +TEST_F(ParquetRowGroupFilterTest, RewriteNotForBoundPredicates) { ASSERT_THAT(Write(), IsOk()); auto less_than = Expressions::LessThan("key", Literal::Int(10)); auto greater_than = Expressions::GreaterThan("key", Literal::Int(11)); @@ -335,10 +348,12 @@ TEST_F(ParquetRowGroupFilterTest, RewriteNotBeforeBinding) { ICEBERG_UNWRAP_OR_FAIL(auto bound, Binder::Bind(*schema_, less_than, true)); Check(Options(Expressions::Not(bound)), {2, 3, 4, 5}); - // Rewrite first: weakening an unknown predicate before NOT could discard rows. - auto unknown = Expressions::Equal("missing", Literal::Int(0)); - Check(Options(Expressions::Not(Expressions::Or(bound, unknown))), {2, 3, 4, 5}); - Check(Options(Expressions::Not(Expressions::And(bound, unknown))), {0, 1, 2, 3, 4, 5}); + // Unsupported transforms must remain conservative under rewritten NOTs. + auto unsupported = + Expressions::Equal(Expressions::Bucket("key", 16), Literal::Int(0)); + Check(Options(Expressions::Not(Expressions::Or(less_than, unsupported))), {2, 3, 4, 5}); + Check(Options(Expressions::Not(Expressions::And(less_than, unsupported))), + {0, 1, 2, 3, 4, 5}); ASSERT_THAT(Write(true, "[[null,0],[null,1],[10,2],[11,3],[20,4],[21,5]]"), IsOk()); Check(Options(Expressions::Not(Expressions::IsNull("key"))), {2, 3, 4, 5}); @@ -532,9 +547,9 @@ TEST_F(ParquetRowGroupFilterTest, FloatingPointNullNaNAndSignedZero) { TEST_F(ParquetRowGroupFilterTest, InvalidFilterFailsDuringOpen) { ASSERT_THAT(Write(), IsOk()); auto options = Options(Expressions::Count("key")); - options.filter = Expressions::Count("key"); + options.filter = Expressions::Count("value"); EXPECT_THAT(ReaderFactoryRegistry::Open(FileFormatType::kParquet, options), - HasErrorMessage("does not support unbound aggregate")); + HasErrorMessage("does not support bound aggregate")); ICEBERG_UNWRAP_OR_FAIL(auto bound, Binder::Bind(*schema_, options.filter, true)); auto bound_options = Options(bound); From 3938983feb80bfefd75d3725c0311656cf9970fa Mon Sep 17 00:00:00 2001 From: yinzhengsun Date: Tue, 22 Sep 2026 14:20:11 +0800 Subject: [PATCH 3/5] fix ci --- src/iceberg/parquet/parquet_reader.cc | 3 +- .../test/parquet_row_group_filter_test.cc | 152 +++++++++++++----- 2 files changed, 110 insertions(+), 45 deletions(-) diff --git a/src/iceberg/parquet/parquet_reader.cc b/src/iceberg/parquet/parquet_reader.cc index 6bea88998..8cb472042 100644 --- a/src/iceberg/parquet/parquet_reader.cc +++ b/src/iceberg/parquet/parquet_reader.cc @@ -20,6 +20,7 @@ #include "iceberg/parquet/parquet_reader.h" #include +#include #include #include @@ -444,7 +445,7 @@ class ParquetReader::Impl { if (row_group->num_rows() == 0) { continue; } - context_->row_groups_.push_back({i, row_start}); + context_->row_groups_.push_back({.index = i, .first_row = row_start}); } if (context_->row_groups_.empty()) { context_->record_batch_reader_ = std::make_unique(); diff --git a/src/iceberg/test/parquet_row_group_filter_test.cc b/src/iceberg/test/parquet_row_group_filter_test.cc index 53773e4f1..57aa4d2aa 100644 --- a/src/iceberg/test/parquet_row_group_filter_test.cc +++ b/src/iceberg/test/parquet_row_group_filter_test.cc @@ -283,27 +283,77 @@ TEST_F(ParquetRowGroupFilterTest, FileAndFilterTypeCompatibility) { R"([["0.00",0],["1.00",1],["10.00",2],["11.00",3],["20.00",4],["21.00",5]])"; const std::string timestamps = R"([["1970-01-01 00:00:00",0],["1970-01-01 00:00:01",1],["1970-01-01 00:00:10",2],["1970-01-01 00:00:11",3],["1970-01-01 00:00:20",4],["1970-01-01 00:00:21",5]])"; - for (const auto& test : std::vector{ - {float32(), float64(), numbers, Literal::Double(10), true}, - {float64(), float32(), numbers, Literal::Float(10), false}, - {int64(), int32(), numbers, Literal::Int(10), false}, - {int32(), date(), numbers, Literal::Date(10), false}, - {date(), int32(), numbers, Literal::Int(10), false}, - {decimal(9, 2), decimal(18, 2), decimals, Literal::Decimal(1000, 18, 2), true}, - {decimal(9, 2), decimal(8, 2), decimals, Literal::Decimal(1000, 8, 2), false}, - {decimal(9, 2), decimal(9, 3), decimals, Literal::Decimal(1000, 9, 3), false}, - {timestamp(), timestamp_ns(), timestamps, Literal::TimestampNs(10000000), - false}, - {timestamp(), timestamp_tz(), timestamps, Literal::TimestampTz(10000000), - false}, - {timestamp_ns(), timestamp_ns(), timestamps, Literal::TimestampNs(10000000000), - true}, - {timestamp_tz(), timestamp_tz(), timestamps, Literal::TimestampTz(10000000), - true}, - {fixed(1), fixed(1), R"([["a",0],["b",1],["m",2],["n",3],["y",4],["z",5]])", - Literal::Fixed({'m'}), true}, - {fixed(1), fixed(2), R"([["a",0],["b",1],["m",2],["n",3],["y",4],["z",5]])", - Literal::Fixed({'m', 'm'}), false}}) { + for (const auto& test : + std::vector{{.file_type = float32(), + .filter_type = float64(), + .json = numbers, + .value = Literal::Double(10), + .compatible = true}, + {.file_type = float64(), + .filter_type = float32(), + .json = numbers, + .value = Literal::Float(10), + .compatible = false}, + {.file_type = int64(), + .filter_type = int32(), + .json = numbers, + .value = Literal::Int(10), + .compatible = false}, + {.file_type = int32(), + .filter_type = date(), + .json = numbers, + .value = Literal::Date(10), + .compatible = false}, + {.file_type = date(), + .filter_type = int32(), + .json = numbers, + .value = Literal::Int(10), + .compatible = false}, + {.file_type = decimal(9, 2), + .filter_type = decimal(18, 2), + .json = decimals, + .value = Literal::Decimal(1000, 18, 2), + .compatible = true}, + {.file_type = decimal(9, 2), + .filter_type = decimal(8, 2), + .json = decimals, + .value = Literal::Decimal(1000, 8, 2), + .compatible = false}, + {.file_type = decimal(9, 2), + .filter_type = decimal(9, 3), + .json = decimals, + .value = Literal::Decimal(1000, 9, 3), + .compatible = false}, + {.file_type = timestamp(), + .filter_type = timestamp_ns(), + .json = timestamps, + .value = Literal::TimestampNs(10000000), + .compatible = false}, + {.file_type = timestamp(), + .filter_type = timestamp_tz(), + .json = timestamps, + .value = Literal::TimestampTz(10000000), + .compatible = false}, + {.file_type = timestamp_ns(), + .filter_type = timestamp_ns(), + .json = timestamps, + .value = Literal::TimestampNs(10000000000), + .compatible = true}, + {.file_type = timestamp_tz(), + .filter_type = timestamp_tz(), + .json = timestamps, + .value = Literal::TimestampTz(10000000), + .compatible = true}, + {.file_type = fixed(1), + .filter_type = fixed(1), + .json = R"([["a",0],["b",1],["m",2],["n",3],["y",4],["z",5]])", + .value = Literal::Fixed({'m'}), + .compatible = true}, + {.file_type = fixed(1), + .filter_type = fixed(2), + .json = R"([["a",0],["b",1],["m",2],["n",3],["y",4],["z",5]])", + .value = Literal::Fixed({'m', 'm'}), + .compatible = false}}) { SCOPED_TRACE(test.file_type->ToString() + " -> " + test.filter_type->ToString()); schema_ = std::make_shared(std::vector{ SchemaField::MakeOptional(1, "key", test.file_type), schema_->fields()[1]}); @@ -367,12 +417,15 @@ TEST_F(ParquetRowGroupFilterTest, BoundComparisonBoundariesAndAllNullGroups) { std::vector expected; }; for (const auto& test : std::vector{ - {Expressions::LessThan("key", Literal::Int(10)), {}}, - {Expressions::LessThanOrEqual("key", Literal::Int(10)), {2, 3}}, - {Expressions::GreaterThan("key", Literal::Int(21)), {}}, - {Expressions::GreaterThanOrEqual("key", Literal::Int(21)), {4, 5}}, - {Expressions::Equal("key", Literal::Int(11)), {2, 3}}, - {Expressions::In("key", {Literal::Int(11), Literal::Int(20)}), {2, 3, 4, 5}}, + {.filter = Expressions::LessThan("key", Literal::Int(10)), .expected = {}}, + {.filter = Expressions::LessThanOrEqual("key", Literal::Int(10)), + .expected = {2, 3}}, + {.filter = Expressions::GreaterThan("key", Literal::Int(21)), .expected = {}}, + {.filter = Expressions::GreaterThanOrEqual("key", Literal::Int(21)), + .expected = {4, 5}}, + {.filter = Expressions::Equal("key", Literal::Int(11)), .expected = {2, 3}}, + {.filter = Expressions::In("key", {Literal::Int(11), Literal::Int(20)}), + .expected = {2, 3, 4, 5}}, }) { ICEBERG_UNWRAP_OR_FAIL(auto bound, Binder::Bind(*schema_, test.filter, true)); Check(Options(bound), test.expected); @@ -425,14 +478,14 @@ TEST_F(ParquetRowGroupFilterTest, SplitIntersectionAndEmptySplit) { auto offset = metadata_->RowGroup(1)->file_offset(); ASSERT_GT(offset, metadata_->RowGroup(0)->file_offset()); auto options = Options(Expressions::GreaterThanOrEqual("key", Literal::Int(20))); - options.split = Split{static_cast(offset), 100000}; + options.split = Split{.offset = static_cast(offset), .length = 100000}; Check(options, {4, 5}); ICEBERG_UNWRAP_OR_FAIL( options.filter, Binder::Bind(*schema_, Expressions::LessThan("key", Literal::Int(2)), true)); Check(options, {}); options.filter = nullptr; - options.split = Split{static_cast(offset), 0}; + options.split = Split{.offset = static_cast(offset), .length = 0}; Check(options, {}); } @@ -443,14 +496,21 @@ TEST_F(ParquetRowGroupFilterTest, PrimitiveComparisonTypes) { Literal literal; }; for (const auto& test : std::vector{ - {int64(), "[[0,0],[1,1],[10,2],[11,3],[20,4],[21,5]]", Literal::Long(10)}, - {date(), "[[0,0],[1,1],[10,2],[11,3],[20,4],[21,5]]", Literal::Date(10)}, - {string(), R"([["a",0],["b",1],["m",2],["n",3],["y",4],["z",5]])", - Literal::String("m")}, - {boolean(), "[[false,0],[false,1],[true,2],[true,3],[false,4],[false,5]]", - Literal::Boolean(true)}, - {binary(), R"([["a",0],["b",1],["m",2],["n",3],["y",4],["z",5]])", - Literal::Binary({'m'})}}) { + {.type = int64(), + .json = "[[0,0],[1,1],[10,2],[11,3],[20,4],[21,5]]", + .literal = Literal::Long(10)}, + {.type = date(), + .json = "[[0,0],[1,1],[10,2],[11,3],[20,4],[21,5]]", + .literal = Literal::Date(10)}, + {.type = string(), + .json = R"([["a",0],["b",1],["m",2],["n",3],["y",4],["z",5]])", + .literal = Literal::String("m")}, + {.type = boolean(), + .json = "[[false,0],[false,1],[true,2],[true,3],[false,4],[false,5]]", + .literal = Literal::Boolean(true)}, + {.type = binary(), + .json = R"([["a",0],["b",1],["m",2],["n",3],["y",4],["z",5]])", + .literal = Literal::Binary({'m'})}}) { schema_ = std::make_shared( std::vector{SchemaField::MakeOptional(1, "key", test.type), SchemaField::MakeRequired(2, "value", int64())}); @@ -510,13 +570,17 @@ TEST_F(ParquetRowGroupFilterTest, DecimalAndTemporalStatistics) { }; for ( const auto& test : std::vector{ - {decimal(9, 2), - R"([["0.00",0],["1.00",1],["10.00",2],["11.00",3],["20.00",4],["21.00",5]])", - Literal::Decimal(1000, 9, 2)}, - {time(), "[[0,0],[1,1],[10,2],[11,3],[20,4],[21,5]]", Literal::Time(10)}, - {timestamp(), - R"([["1970-01-01 00:00:00",0],["1970-01-01 00:00:01",1],["1970-01-01 00:00:10",2],["1970-01-01 00:00:11",3],["1970-01-01 00:00:20",4],["1970-01-01 00:00:21",5]])", - Literal::Timestamp(10000000)}}) { + {.type = decimal(9, 2), + .json = + R"([["0.00",0],["1.00",1],["10.00",2],["11.00",3],["20.00",4],["21.00",5]])", + .value = Literal::Decimal(1000, 9, 2)}, + {.type = time(), + .json = "[[0,0],[1,1],[10,2],[11,3],[20,4],[21,5]]", + .value = Literal::Time(10)}, + {.type = timestamp(), + .json = + R"([["1970-01-01 00:00:00",0],["1970-01-01 00:00:01",1],["1970-01-01 00:00:10",2],["1970-01-01 00:00:11",3],["1970-01-01 00:00:20",4],["1970-01-01 00:00:21",5]])", + .value = Literal::Timestamp(10000000)}}) { schema_ = std::make_shared( std::vector{SchemaField::MakeOptional(1, "key", test.type), SchemaField::MakeRequired(2, "value", int64())}); From 782b5039538f52649bb59cca119f74541b7e5c4e Mon Sep 17 00:00:00 2001 From: yinzhengsun Date: Tue, 22 Sep 2026 14:53:30 +0800 Subject: [PATCH 4/5] fix ci --- src/iceberg/parquet/parquet_reader.cc | 4 +- .../test/parquet_row_group_filter_test.cc | 120 ++++++++++-------- 2 files changed, 70 insertions(+), 54 deletions(-) diff --git a/src/iceberg/parquet/parquet_reader.cc b/src/iceberg/parquet/parquet_reader.cc index 8cb472042..fc72de07e 100644 --- a/src/iceberg/parquet/parquet_reader.cc +++ b/src/iceberg/parquet/parquet_reader.cc @@ -429,8 +429,8 @@ class ParquetReader::Impl { next_row_start += row_group->num_rows(); if (split_.has_value()) { auto row_group_offset = row_group->file_offset(); - bool in_split = row_group_offset >= split_->offset && - row_group_offset < split_->offset + split_->length; + bool in_split = std::cmp_greater_equal(row_group_offset, split_->offset) && + std::cmp_less(row_group_offset, split_->offset + split_->length); if (!in_split) { continue; } diff --git a/src/iceberg/test/parquet_row_group_filter_test.cc b/src/iceberg/test/parquet_row_group_filter_test.cc index 57aa4d2aa..e6c27b373 100644 --- a/src/iceberg/test/parquet_row_group_filter_test.cc +++ b/src/iceberg/test/parquet_row_group_filter_test.cc @@ -47,14 +47,18 @@ class ParquetRowGroupFilterTest : public ::testing::Test { void SetUp() override { parquet::RegisterAll(); io_ = std::make_shared(); - schema_ = std::make_shared( - std::vector{SchemaField::MakeOptional(1, "key", int32()), - SchemaField::MakeRequired(2, "value", int64())}); + 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 = true, const std::string& json = "[[0,0],[1,1],[10,2],[11,3],[20,4],[21,5]]") { ArrowSchema c_schema; @@ -67,10 +71,7 @@ class ParquetRowGroupFilterTest : public ::testing::Test { ::arrow::RecordBatch::FromStructArray(array)); ICEBERG_ARROW_ASSIGN_OR_RETURN(auto table, ::arrow::Table::FromRecordBatches({batch})); - return WriteTable(table, statistics); - } - Status WriteTable(const std::shared_ptr<::arrow::Table>& table, bool statistics) { ICEBERG_ASSIGN_OR_RAISE(auto out, arrow::OpenArrowOutputStream(io_, path_)); ::parquet::WriterProperties::Builder properties; properties.disable_dictionary(); @@ -86,12 +87,18 @@ class ParquetRowGroupFilterTest : public ::testing::Test { } ReaderOptions Options(std::shared_ptr filter, bool case_sensitive = true) { - // Direct readers need bound references for columns outside the projection. - // Leave invalid or already-bound expressions intact for their dedicated tests. - if (filter) { - auto bound = Binder::Bind(*schema_, filter, case_sensitive); - if (bound) { - filter = *bound; + // 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_, @@ -131,6 +138,7 @@ class ParquetRowGroupFilterTest : public ::testing::Test { } 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); } @@ -233,7 +241,6 @@ TEST_F(ParquetRowGroupFilterTest, RenameBoundPredicateAndCaseSensitivity) { auto options = Options(bound); Check(options, {4, 5}); options = Options(Expressions::Equal("RENAMED", Literal::Int(20)), false); - options.properties.Set(ReaderProperties::kFilterCaseSensitive, false); Check(options, {4, 5}); } @@ -247,26 +254,37 @@ TEST_F(ParquetRowGroupFilterTest, OpenBindsUnboundProjectedReferences) { Check(options, {4, 5}); } -TEST_F(ParquetRowGroupFilterTest, MissingStatsUnknownFieldsAndTypePromotion) { +TEST_F(ParquetRowGroupFilterTest, MissingStatisticsRetainGroups) { ASSERT_THAT(Write(false), IsOk()); Check(Options(Expressions::Equal("key", Literal::Int(100))), {0, 1, 2, 3, 4, 5}); Check(Options(Expressions::IsNull("key")), {0, 1, 2, 3, 4, 5}); Check(Options(Expressions::NotNull("key")), {0, 1, 2, 3, 4, 5}); +} + +TEST_F(ParquetRowGroupFilterTest, OpenRejectsUnresolvableReferences) { ASSERT_THAT(Write(), IsOk()); - EXPECT_THAT(ReaderFactoryRegistry::Open( - FileFormatType::kParquet, - Options(Expressions::Equal("missing", Literal::Int(0)))), + auto options = Options(nullptr); + options.filter = Expressions::Equal("missing", Literal::Int(0)); + EXPECT_THAT(ReaderFactoryRegistry::Open(FileFormatType::kParquet, options), HasErrorMessage("Cannot find field 'missing'")); - auto options = Options(Expressions::Equal("key", Literal::Int(0))); options.filter = Expressions::Equal("key", Literal::Int(0)); EXPECT_THAT(ReaderFactoryRegistry::Open(FileFormatType::kParquet, options), HasErrorMessage("Cannot find field 'key'")); +} + +TEST_F(ParquetRowGroupFilterTest, PromotedIntegerStatistics) { + ASSERT_THAT(Write(), IsOk()); + SetKeyType(int64()); + Check(Options(Expressions::Equal("key", Literal::Long(100))), {}); + Check(Options(Expressions::Equal("key", Literal::Long(10))), {2, 3}); +} + +TEST_F(ParquetRowGroupFilterTest, MissingColumnWithDefaultRetainsGroups) { + ASSERT_THAT(Write(), IsOk()); schema_ = std::make_shared(std::vector{ - SchemaField::MakeOptional(1, "key", int64()), schema_->fields()[1], + schema_->fields()[0], schema_->fields()[1], SchemaField::MakeOptional(3, "defaulted", int32()) .WithInitialDefault(std::make_shared(Literal::Int(7)))}); - Check(Options(Expressions::Equal("key", Literal::Long(100))), {}); - Check(Options(Expressions::Equal("key", Literal::Long(10))), {2, 3}); Check(Options(Expressions::Equal("defaulted", Literal::Int(8))), {0, 1, 2, 3, 4, 5}); } @@ -355,11 +373,9 @@ TEST_F(ParquetRowGroupFilterTest, FileAndFilterTypeCompatibility) { .value = Literal::Fixed({'m', 'm'}), .compatible = false}}) { SCOPED_TRACE(test.file_type->ToString() + " -> " + test.filter_type->ToString()); - schema_ = std::make_shared(std::vector{ - SchemaField::MakeOptional(1, "key", test.file_type), schema_->fields()[1]}); + SetKeyType(test.file_type); ASSERT_THAT(Write(true, test.json), IsOk()); - schema_ = std::make_shared(std::vector{ - SchemaField::MakeOptional(1, "key", test.filter_type), schema_->fields()[1]}); + SetKeyType(test.filter_type); Check(Options(Expressions::Equal("key", test.value)), test.compatible ? std::vector{2, 3} : std::vector{0, 1, 2, 3, 4, 5}); @@ -427,8 +443,7 @@ TEST_F(ParquetRowGroupFilterTest, BoundComparisonBoundariesAndAllNullGroups) { {.filter = Expressions::In("key", {Literal::Int(11), Literal::Int(20)}), .expected = {2, 3, 4, 5}}, }) { - ICEBERG_UNWRAP_OR_FAIL(auto bound, Binder::Bind(*schema_, test.filter, true)); - Check(Options(bound), test.expected); + Check(Options(test.filter), test.expected); } } @@ -511,48 +526,47 @@ TEST_F(ParquetRowGroupFilterTest, PrimitiveComparisonTypes) { {.type = binary(), .json = R"([["a",0],["b",1],["m",2],["n",3],["y",4],["z",5]])", .literal = Literal::Binary({'m'})}}) { - schema_ = std::make_shared( - std::vector{SchemaField::MakeOptional(1, "key", test.type), - SchemaField::MakeRequired(2, "value", int64())}); + SCOPED_TRACE(test.type->ToString()); + SetKeyType(test.type); ASSERT_THAT(Write(true, test.json), IsOk()); Check(Options(Expressions::Equal("key", test.literal)), {2, 3}); } } -TEST_F(ParquetRowGroupFilterTest, FloatAndNestedStatsWithTransformFallback) { - schema_ = std::make_shared( - std::vector{SchemaField::MakeOptional(1, "key", float64()), - SchemaField::MakeRequired(2, "value", int64())}); +TEST_F(ParquetRowGroupFilterTest, FloatingPointStatistics) { + SetKeyType(float64()); ASSERT_THAT(Write(), IsOk()); Check(Options(Expressions::Equal("key", Literal::Double(100))), {}); Check(Options(Expressions::Equal("key", Literal::Double(10))), {2, 3}); Check(Options(Expressions::IsNaN("key")), {0, 1, 2, 3, 4, 5}); - schema_ = std::make_shared(std::vector{ - SchemaField::MakeOptional(1, "key", - std::make_shared(std::vector{ - SchemaField::MakeOptional(3, "nested", int32())})), - SchemaField::MakeRequired(2, "value", int64())}); +} + +TEST_F(ParquetRowGroupFilterTest, 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()); Check(Options(Expressions::Equal("key.nested", Literal::Int(100))), {}); Check(Options(Expressions::Equal("key.nested", Literal::Int(10))), {2, 3}); - schema_ = std::make_shared( - std::vector{SchemaField::MakeOptional(1, "key", int32()), - SchemaField::MakeRequired(2, "value", int64())}); +} + +TEST_F(ParquetRowGroupFilterTest, UnsupportedTransformRetainsGroups) { ASSERT_THAT(Write(), IsOk()); Check(Options(Expressions::Equal(Expressions::Bucket("key", 16), Literal::Int(15))), {0, 1, 2, 3, 4, 5}); } -TEST_F(ParquetRowGroupFilterTest, NegativeAndPrefixPredicates) { +TEST_F(ParquetRowGroupFilterTest, NegativePredicates) { ASSERT_THAT(Write(true, "[[5,0],[5,1],[null,2],[5,3],[6,4],[6,5]]"), IsOk()); Check(Options(Expressions::NotEqual("key", Literal::Int(5))), {0, 1, 2, 3, 4, 5}); Check(Options(Expressions::NotIn("key", {Literal::Int(5), Literal::Int(6)})), {0, 1, 2, 3, 4, 5}); Check(Options(Expressions::Not(Expressions::LessThan("key", Literal::Int(100)))), {}); - schema_ = std::make_shared(std::vector{ - SchemaField::MakeOptional(1, "key", string()), schema_->fields()[1]}); +} + +TEST_F(ParquetRowGroupFilterTest, PrefixPredicates) { + SetKeyType(string()); ASSERT_THAT( Write( true, @@ -581,18 +595,16 @@ TEST_F(ParquetRowGroupFilterTest, DecimalAndTemporalStatistics) { .json = R"([["1970-01-01 00:00:00",0],["1970-01-01 00:00:01",1],["1970-01-01 00:00:10",2],["1970-01-01 00:00:11",3],["1970-01-01 00:00:20",4],["1970-01-01 00:00:21",5]])", .value = Literal::Timestamp(10000000)}}) { - schema_ = std::make_shared( - std::vector{SchemaField::MakeOptional(1, "key", test.type), - SchemaField::MakeRequired(2, "value", int64())}); + SCOPED_TRACE(test.type->ToString()); + SetKeyType(test.type); ASSERT_THAT(Write(true, test.json), IsOk()); Check(Options(Expressions::Equal("key", test.value)), {2, 3}); Check(Options(Expressions::LessThan("key", test.value)), {0, 1}); } } -TEST_F(ParquetRowGroupFilterTest, FloatingPointNullNaNAndSignedZero) { - schema_ = std::make_shared(std::vector{ - SchemaField::MakeOptional(1, "key", float64()), schema_->fields()[1]}); +TEST_F(ParquetRowGroupFilterTest, FloatingPointNullAndSignedZero) { + SetKeyType(float64()); ASSERT_THAT(Write(true, "[[-0.0,0],[0.0,1],[null,2],[null,3],[10.0,4],[11.0,5]]"), IsOk()); Check(Options(Expressions::Equal("key", Literal::Double(-0.0))), {0, 1}); @@ -601,6 +613,10 @@ TEST_F(ParquetRowGroupFilterTest, FloatingPointNullNaNAndSignedZero) { Check(Options(Expressions::NotNull("key")), {0, 1, 4, 5}); Check(Options(Expressions::IsNaN("key")), {0, 1, 4, 5}); Check(Options(Expressions::LessThan("key", Literal::Double(-100))), {}); +} + +TEST_F(ParquetRowGroupFilterTest, AllNaNGroupRetainedWithoutComparableBounds) { + SetKeyType(float64()); ASSERT_THAT(Write(true, "[[NaN,0],[-NaN,1],[10.0,2],[11.0,3],[null,4],[null,5]]"), IsOk()); Check(Options(Expressions::Equal("key", Literal::Double(100))), {0, 1}); @@ -610,7 +626,7 @@ TEST_F(ParquetRowGroupFilterTest, FloatingPointNullNaNAndSignedZero) { TEST_F(ParquetRowGroupFilterTest, InvalidFilterFailsDuringOpen) { ASSERT_THAT(Write(), IsOk()); - auto options = Options(Expressions::Count("key")); + auto options = Options(nullptr); options.filter = Expressions::Count("value"); EXPECT_THAT(ReaderFactoryRegistry::Open(FileFormatType::kParquet, options), HasErrorMessage("does not support bound aggregate")); From a1211ccbea27b77942d79efa0295997f97da3c11 Mon Sep 17 00:00:00 2001 From: yinzhengsun Date: Tue, 22 Sep 2026 19:51:03 +0800 Subject: [PATCH 5/5] fix --- .../parquet_metrics_row_group_filter.cc | 241 +++++++++--------- .../test/parquet_row_group_filter_test.cc | 12 + 2 files changed, 136 insertions(+), 117 deletions(-) diff --git a/src/iceberg/parquet/parquet_metrics_row_group_filter.cc b/src/iceberg/parquet/parquet_metrics_row_group_filter.cc index 3d245a0e3..89a12ad87 100644 --- a/src/iceberg/parquet/parquet_metrics_row_group_filter.cc +++ b/src/iceberg/parquet/parquet_metrics_row_group_filter.cc @@ -18,6 +18,8 @@ */ #include +#include +#include #include #include @@ -25,7 +27,6 @@ #include "iceberg/expression/expression_visitor.h" #include "iceberg/expression/rewrite_not.h" #include "iceberg/metadata_columns.h" -#include "iceberg/metrics.h" #include "iceberg/parquet/parquet_metrics_internal.h" #include "iceberg/parquet/parquet_metrics_row_group_filter_internal.h" #include "iceberg/parquet/parquet_schema_util_internal.h" @@ -59,72 +60,62 @@ class MetricsVisitor : public BoundVisitor { Result Or(bool left, bool right) override { return left || right; } Result IsNull(const std::shared_ptr& expr) override { - return MayContainNull(GetMetrics(expr, false)); + return MayContainNull(std::dynamic_pointer_cast(expr)); } Result NotNull(const std::shared_ptr& expr) override { - return !ContainsNullsOnly(GetMetrics(expr, false)); + if (ContainsNullsOnly(std::dynamic_pointer_cast(expr))) { + return kRowsCannotMatch; + } + return kRowsMightMatch; } Result IsNaN(const std::shared_ptr& expr) override { - return !ContainsNullsOnly(GetMetrics(expr, false)); + if (ContainsNullsOnly(std::dynamic_pointer_cast(expr))) { + return kRowsCannotMatch; + } + return kRowsMightMatch; } Result NotNaN(const std::shared_ptr&) override { return kRowsMightMatch; } Result Lt(const std::shared_ptr& expr, const Literal& value) override { - const auto metrics = GetMetrics(expr); - if (ContainsNullsOnly(metrics)) { - return kRowsCannotMatch; - } - if (!metrics.lower_bound || !ComparableLiteral(value)) { - return kRowsMightMatch; - } - return !(*metrics.lower_bound >= value); + return VisitInequality(std::dynamic_pointer_cast(expr), value, + std::less{}, /*use_lower_bound=*/true); } Result LtEq(const std::shared_ptr& expr, const Literal& value) override { - const auto metrics = GetMetrics(expr); - if (ContainsNullsOnly(metrics)) { - return kRowsCannotMatch; - } - if (!metrics.lower_bound || !ComparableLiteral(value)) { - return kRowsMightMatch; - } - return !(*metrics.lower_bound > value); + return VisitInequality(std::dynamic_pointer_cast(expr), value, + std::less_equal{}, /*use_lower_bound=*/true); } Result Gt(const std::shared_ptr& expr, const Literal& value) override { - const auto metrics = GetMetrics(expr); - if (ContainsNullsOnly(metrics)) { - return kRowsCannotMatch; - } - if (!metrics.upper_bound || !ComparableLiteral(value)) { - return kRowsMightMatch; - } - return !(*metrics.upper_bound <= value); + return VisitInequality(std::dynamic_pointer_cast(expr), value, + std::greater{}, /*use_lower_bound=*/false); } Result GtEq(const std::shared_ptr& expr, const Literal& value) override { - const auto metrics = GetMetrics(expr); - if (ContainsNullsOnly(metrics)) { + return VisitInequality(std::dynamic_pointer_cast(expr), value, + std::greater_equal{}, /*use_lower_bound=*/false); + } + + Result Eq(const std::shared_ptr& expr, const Literal& value) override { + const auto ref = std::dynamic_pointer_cast(expr); + if (ContainsNullsOnly(ref)) { return kRowsCannotMatch; } - if (!metrics.upper_bound || !ComparableLiteral(value)) { + const auto lower = MinValue(ref); + if (!lower) { return kRowsMightMatch; } - return !(*metrics.upper_bound < value); - } - - Result Eq(const std::shared_ptr& expr, const Literal& value) override { - const auto metrics = GetMetrics(expr); - if (ContainsNullsOnly(metrics)) { + if (*lower > value) { return kRowsCannotMatch; } - if (!metrics.lower_bound || !metrics.upper_bound || !ComparableLiteral(value)) { + const auto upper = MaxValue(ref); + if (!upper) { return kRowsMightMatch; } - return !(*metrics.lower_bound > value || *metrics.upper_bound < value); + return !(*upper < value); } Result NotEq(const std::shared_ptr&, const Literal&) override { @@ -134,17 +125,27 @@ class MetricsVisitor : public BoundVisitor { Result In(const std::shared_ptr& expr, const BoundSetPredicate::LiteralSet& values) override { - const auto metrics = GetMetrics(expr); - if (ContainsNullsOnly(metrics)) { + const auto ref = std::dynamic_pointer_cast(expr); + if (ContainsNullsOnly(ref)) { + return kRowsCannotMatch; + } + if (values.size() > kInPredicateLimit) { + return kRowsMightMatch; + } + const auto lower = MinValue(ref); + if (!lower) { + return kRowsMightMatch; + } + if (std::ranges::all_of(values, [&](const auto& value) { return value < *lower; })) { return kRowsCannotMatch; } - if (!metrics.lower_bound || !metrics.upper_bound || - values.size() > kInPredicateLimit) { + const auto upper = MaxValue(ref); + if (!upper) { return kRowsMightMatch; } + // Like Java, a single candidate must satisfy both bounds. for (const auto& value : values) { - if (!ComparableLiteral(value) || - !(value < *metrics.lower_bound || value > *metrics.upper_bound)) { + if (!(value < *lower || value > *upper)) { return kRowsMightMatch; } } @@ -158,64 +159,69 @@ class MetricsVisitor : public BoundVisitor { Result StartsWith(const std::shared_ptr& expr, const Literal& value) override { - const auto metrics = GetMetrics(expr); - if (ContainsNullsOnly(metrics)) { + const auto ref = std::dynamic_pointer_cast(expr); + if (ContainsNullsOnly(ref)) { return kRowsCannotMatch; } - if (!metrics.lower_bound || !metrics.upper_bound || !ComparableLiteral(value) || - metrics.lower_bound->type()->type_id() != TypeId::kString) { + const auto lower = MinValue(ref); + if (!lower || lower->type()->type_id() != TypeId::kString) { return kRowsMightMatch; } const auto& prefix = std::get(value.value()); - const auto& lower = std::get(metrics.lower_bound->value()); - const auto& upper = std::get(metrics.upper_bound->value()); - return !(lower.substr(0, prefix.size()) > prefix || - upper.substr(0, prefix.size()) < prefix); + if (std::get(lower->value()).compare(0, prefix.size(), prefix) > 0) { + return kRowsCannotMatch; + } + const auto upper = MaxValue(ref); + if (!upper || upper->type()->type_id() != TypeId::kString) { + return kRowsMightMatch; + } + return std::get(upper->value()).compare(0, prefix.size(), prefix) >= 0; } Result NotStartsWith(const std::shared_ptr& expr, const Literal& value) override { - const auto metrics = GetMetrics(expr); - if (MayContainNull(metrics) || !metrics.lower_bound || !metrics.upper_bound || - !ComparableLiteral(value) || - metrics.lower_bound->type()->type_id() != TypeId::kString) { + const auto ref = std::dynamic_pointer_cast(expr); + if (MayContainNull(ref)) { + return kRowsMightMatch; + } + const auto lower = MinValue(ref); + if (!lower || lower->type()->type_id() != TypeId::kString) { return kRowsMightMatch; } const auto& prefix = std::get(value.value()); - const auto& lower = std::get(metrics.lower_bound->value()); - const auto& upper = std::get(metrics.upper_bound->value()); - return !lower.starts_with(prefix) || !upper.starts_with(prefix); + if (!std::get(lower->value()).starts_with(prefix)) { + return kRowsMightMatch; + } + const auto upper = MaxValue(ref); + if (!upper || upper->type()->type_id() != TypeId::kString) { + return kRowsMightMatch; + } + return !std::get(upper->value()).starts_with(prefix); } private: - static bool ContainsNullsOnly(const FieldMetrics& metrics) { - return metrics.null_value_count >= 0 && - metrics.null_value_count == metrics.value_count; - } - - static bool MayContainNull(const FieldMetrics& metrics) { - return metrics.null_value_count != 0; + bool ContainsNullsOnly(const std::shared_ptr& ref) const { + const auto stats = GetStatistics(ref); + // GetStatistics excludes repeated columns, so each row contributes one value. + return stats && stats->HasNullCount() && stats->null_count() == row_group_.num_rows(); } - static bool ComparableLiteral(const Literal& value) { - return !value.IsNaN() && !value.IsNull() && !value.IsAboveMax() && - !value.IsBelowMin(); + bool MayContainNull(const std::shared_ptr& ref) const { + const auto stats = GetStatistics(ref); + return !stats || !stats->HasNullCount() || stats->null_count() != 0; } - FieldMetrics GetMetrics(const std::shared_ptr& expr, - bool read_bounds = true) const { - FieldMetrics metrics; - auto ref = std::dynamic_pointer_cast(expr); + std::shared_ptr<::parquet::Statistics> GetStatistics( + const std::shared_ptr& ref) const { if (!ref || !ref->type()->is_primitive() || MetadataColumns::IsMetadataColumn(ref->field_id()) || MetadataColumns::IsRowLineageColumn(ref->field_id())) { - return metrics; + return nullptr; } - metrics.field_id = ref->field_id(); auto column = column_indices_.find(ref->field_id()); // Missing columns can have initial defaults, so do not assume all nulls. if (column == column_indices_.end()) { - return metrics; + return nullptr; } const auto& descriptor = *manifest_.descr->Column(column->second); auto field = manifest_.column_index_to_field.find(column->second); @@ -223,54 +229,55 @@ class MetricsVisitor : public BoundVisitor { if (descriptor.max_repetition_level() != 0 || field == manifest_.column_index_to_field.end() || !ValidateParquetTypeCompatibility(*ref->type(), *field->second)) { - return metrics; + return nullptr; } - const auto& type = static_cast(*ref->type()); - auto chunk = row_group_.ColumnChunk(column->second); - auto stats = chunk->statistics(); - if (!stats) { - return metrics; + return row_group_.ColumnChunk(column->second)->statistics(); + } + + std::optional MinValue(const std::shared_ptr& ref) const { + return GetBound(ref, /*is_min=*/true); + } + + std::optional MaxValue(const std::shared_ptr& ref) const { + return GetBound(ref, /*is_min=*/false); + } + + std::optional GetBound(const std::shared_ptr& ref, + bool is_min) const { + const auto stats = GetStatistics(ref); + if (!stats || !stats->HasMinMax()) { + return std::nullopt; } - metrics.value_count = chunk->num_values(); - if (stats->HasNullCount()) { - metrics.null_value_count = stats->null_count(); + const auto& type = static_cast(*ref->type()); + auto result = + ParquetMetrics::StatsValueToLiteral(*stats->descr(), type, *stats, is_min); + if (!result || result->IsNaN()) { + return std::nullopt; } - if (!read_bounds || ContainsNullsOnly(metrics) || !stats->HasMinMax()) { - return metrics; + auto bound = std::move(*result); + if (type.type_id() == TypeId::kFloat && std::get(bound.value()) == 0) { + return Literal::Float(is_min ? -0.0F : 0.0F); } - auto lower_result = - ParquetMetrics::StatsValueToLiteral(descriptor, type, *stats, true); - auto upper_result = - ParquetMetrics::StatsValueToLiteral(descriptor, type, *stats, false); - if (!lower_result || !upper_result) { - return metrics; + if (type.type_id() == TypeId::kDouble && std::get(bound.value()) == 0) { + return Literal::Double(is_min ? -0.0 : 0.0); } - auto lower = std::move(*lower_result); - auto upper = std::move(*upper_result); - if (lower.IsNaN() || upper.IsNaN()) { - return metrics; + return bound; + } + + template + bool VisitInequality(const std::shared_ptr& ref, const Literal& value, + Comparator compare, bool lower_bound) const { + if (ContainsNullsOnly(ref)) { + return kRowsCannotMatch; } - if (type.type_id() == TypeId::kFloat) { - if (std::get(lower.value()) == 0) { - lower = Literal::Float(-0.0F); - } - if (std::get(upper.value()) == 0) { - upper = Literal::Float(0.0F); - } - } else if (type.type_id() == TypeId::kDouble) { - if (std::get(lower.value()) == 0) { - lower = Literal::Double(-0.0); - } - if (std::get(upper.value()) == 0) { - upper = Literal::Double(0.0); - } + if (value.IsNaN()) { + return kRowsMightMatch; } - if (lower > upper) { - return metrics; + const auto bound = lower_bound ? MinValue(ref) : MaxValue(ref); + if (!bound) { + return kRowsMightMatch; } - metrics.lower_bound = std::move(lower); - metrics.upper_bound = std::move(upper); - return metrics; + return compare(*bound, value); } const ::parquet::arrow::SchemaManifest& manifest_; diff --git a/src/iceberg/test/parquet_row_group_filter_test.cc b/src/iceberg/test/parquet_row_group_filter_test.cc index e6c27b373..408b6974e 100644 --- a/src/iceberg/test/parquet_row_group_filter_test.cc +++ b/src/iceberg/test/parquet_row_group_filter_test.cc @@ -224,6 +224,7 @@ TEST_F(ParquetRowGroupFilterTest, AllNoneAndResidualRows) { Check(Options(True::Instance()), {0, 1, 2, 3, 4, 5}); Check(Options(False::Instance()), {}); Check(Options(Expressions::Equal("key", Literal::Int(100))), {}); + Check(Options(Expressions::Equal("key", Literal::Int(-1))), {}); // RG-only: key=1 is retained with key=0. This is not an exact row filter. Check(Options(Expressions::Equal("key", Literal::Int(0))), {0, 1}); Check(Options(Expressions::And(Expressions::GreaterThan("key", Literal::Int(9)), @@ -451,6 +452,7 @@ TEST_F(ParquetRowGroupFilterTest, InPredicateLimit) { ASSERT_THAT(Write(), IsOk()); Check(Options(Expressions::In("key", {Literal::Int(0), Literal::Int(20)})), {0, 1, 4, 5}); + Check(Options(Expressions::In("key", {Literal::Int(5), Literal::Int(25)})), {}); std::vector values; for (int i = 100; i < 200; ++i) { values.push_back(Literal::Int(i)); @@ -574,6 +576,11 @@ TEST_F(ParquetRowGroupFilterTest, PrefixPredicates) { IsOk()); Check(Options(Expressions::StartsWith("key", "ap")), {0, 1, 4, 5}); Check(Options(Expressions::NotStartsWith("key", "ap")), {2, 3, 4, 5}); + Check(Options(Expressions::StartsWith("key", "aa")), {}); + Check(Options(Expressions::StartsWith("key", "z")), {}); + Check(Options(Expressions::StartsWith("key", "apricots")), {}); + Check(Options(Expressions::StartsWith("key", "")), {0, 1, 2, 3, 4, 5}); + Check(Options(Expressions::NotStartsWith("key", "")), {4, 5}); } TEST_F(ParquetRowGroupFilterTest, DecimalAndTemporalStatistics) { @@ -611,7 +618,11 @@ TEST_F(ParquetRowGroupFilterTest, FloatingPointNullAndSignedZero) { Check(Options(Expressions::Equal("key", Literal::Double(0.0))), {0, 1}); Check(Options(Expressions::IsNull("key")), {2, 3}); Check(Options(Expressions::NotNull("key")), {0, 1, 4, 5}); + // Like Java, IS NAN excludes all-null groups; NOT NAN retains them. Check(Options(Expressions::IsNaN("key")), {0, 1, 4, 5}); + Check(Options(Expressions::NotNaN("key")), {0, 1, 2, 3, 4, 5}); + Check(Options(Expressions::Not(Expressions::IsNaN("key"))), {0, 1, 2, 3, 4, 5}); + Check(Options(Expressions::Not(Expressions::NotNaN("key"))), {0, 1, 4, 5}); Check(Options(Expressions::LessThan("key", Literal::Double(-100))), {}); } @@ -621,6 +632,7 @@ TEST_F(ParquetRowGroupFilterTest, AllNaNGroupRetainedWithoutComparableBounds) { IsOk()); Check(Options(Expressions::Equal("key", Literal::Double(100))), {0, 1}); Check(Options(Expressions::IsNaN("key")), {0, 1, 2, 3}); + Check(Options(Expressions::NotNaN("key")), {0, 1, 2, 3, 4, 5}); Check(Options(Expressions::LessThan("key", Literal::Double(-100))), {0, 1}); }