From 2e387ad59b1757c5df6c8389c0688fb6107900c8 Mon Sep 17 00:00:00 2001 From: yinzhengsun Date: Thu, 15 Oct 2026 14:48:02 +0800 Subject: [PATCH 1/5] feat(parquet): prune row groups with footer statistics Add ParquetMetricsRowGroupFilter and apply it when opening a Parquet reader. Row groups are first selected by split and then pruned by footer statistics; RowNumberReader keeps _pos and _row_id correct across pruned row groups. Also apply RewriteNot in Evaluator and ResidualEvaluator so NOT range comparisons exclude nulls. --- src/iceberg/CMakeLists.txt | 1 + src/iceberg/data/file_scan_task_reader.cc | 12 +- src/iceberg/data/file_scan_task_reader.h | 2 + src/iceberg/expression/evaluator.cc | 2 + src/iceberg/expression/residual_evaluator.cc | 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 | 326 ++++++++ ...arquet_metrics_row_group_filter_internal.h | 52 ++ src/iceberg/parquet/parquet_reader.cc | 56 +- src/iceberg/parquet/parquet_schema_util.cc | 4 - .../parquet/parquet_schema_util_internal.h | 7 + src/iceberg/test/CMakeLists.txt | 1 + src/iceberg/test/evaluator_test.cc | 35 + .../test/file_scan_task_reader_test.cc | 195 +++++ .../test/parquet_row_group_filter_test.cc | 745 ++++++++++++++++++ src/iceberg/test/residual_evaluator_test.cc | 48 ++ 18 files changed, 1516 insertions(+), 73 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 717f1da4b2..cf88e20c98 100644 --- a/src/iceberg/CMakeLists.txt +++ b/src/iceberg/CMakeLists.txt @@ -319,6 +319,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 7d3a8bc7e3..292d7ac5c5 100644 --- a/src/iceberg/data/file_scan_task_reader.cc +++ b/src/iceberg/data/file_scan_task_reader.cc @@ -23,7 +23,6 @@ #include #include #include -#include #include "iceberg/arrow_c_data_guard_internal.h" #include "iceberg/arrow_c_data_util_internal.h" @@ -41,7 +40,7 @@ namespace { ReaderOptions MakeReaderOptions(const DataFile& data_file, std::shared_ptr io, std::shared_ptr projection, - std::shared_ptr filter, + std::shared_ptr filter, bool case_sensitive, std::shared_ptr name_mapping, ReaderProperties properties, std::optional first_row_id, @@ -52,6 +51,7 @@ ReaderOptions MakeReaderOptions(const DataFile& data_file, std::shared_ptrfile_path, task.start(), task.length()); } + auto filter = task.residual_filter(); + if (task.delete_files().empty()) { auto options = MakeReaderOptions( - *data_file, io_, projected_schema_, task.residual_filter(), name_mapping_, + *data_file, io_, projected_schema_, filter, case_sensitive_, 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)); @@ -198,7 +200,7 @@ class FileScanTaskReader::Impl { project_batch_function)); auto options = MakeReaderOptions( - *data_file, io_, required_schema, task.residual_filter(), name_mapping_, + *data_file, io_, required_schema, filter, case_sensitive_, 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)); @@ -215,6 +217,7 @@ class FileScanTaskReader::Impl { : io_(std::move(options.io)), schemas_(std::move(options.schemas)), projected_schema_(std::move(options.projected_schema)), + case_sensitive_(options.case_sensitive), name_mapping_(std::move(options.name_mapping)), properties_(ReaderProperties::FromMap(options.properties)), field_lookup_(std::move(field_lookup)), @@ -223,6 +226,7 @@ class FileScanTaskReader::Impl { std::shared_ptr io_; std::vector> schemas_; std::shared_ptr projected_schema_; + bool case_sensitive_; std::shared_ptr name_mapping_; ReaderProperties properties_; DeleteFilter::FieldLookup field_lookup_; diff --git a/src/iceberg/data/file_scan_task_reader.h b/src/iceberg/data/file_scan_task_reader.h index c34f3e7aa8..d88dcae79a 100644 --- a/src/iceberg/data/file_scan_task_reader.h +++ b/src/iceberg/data/file_scan_task_reader.h @@ -55,6 +55,8 @@ class ICEBERG_DATA_EXPORT FileScanTaskReader { /// The output schema for the returned ArrowArrayStream. Must be a /// projection of table_schema. std::shared_ptr projected_schema; + /// Case sensitivity for field name resolution. Pass TableScan::is_case_sensitive(). + bool case_sensitive = true; /// Optional name mapping for files written without field IDs. std::shared_ptr name_mapping; /// Format-specific or implementation-specific options for data readers. diff --git a/src/iceberg/expression/evaluator.cc b/src/iceberg/expression/evaluator.cc index 02438e3f50..0183097350 100644 --- a/src/iceberg/expression/evaluator.cc +++ b/src/iceberg/expression/evaluator.cc @@ -21,6 +21,7 @@ #include "iceberg/expression/binder.h" #include "iceberg/expression/expression_visitor.h" +#include "iceberg/expression/rewrite_not.h" #include "iceberg/schema.h" #include "iceberg/util/macros.h" @@ -140,6 +141,7 @@ Result> Evaluator::Make(const Schema& schema, std::shared_ptr unbound, bool case_sensitive) { ICEBERG_ASSIGN_OR_RAISE(auto bound_expr, Binder::Bind(schema, unbound, case_sensitive)); + ICEBERG_ASSIGN_OR_RAISE(bound_expr, RewriteNot::Visit(bound_expr)); return std::unique_ptr(new Evaluator(std::move(bound_expr))); } diff --git a/src/iceberg/expression/residual_evaluator.cc b/src/iceberg/expression/residual_evaluator.cc index 821213eeaa..1f4abd488d 100644 --- a/src/iceberg/expression/residual_evaluator.cc +++ b/src/iceberg/expression/residual_evaluator.cc @@ -22,6 +22,7 @@ #include "iceberg/expression/expression.h" #include "iceberg/expression/expression_visitor.h" #include "iceberg/expression/predicate.h" +#include "iceberg/expression/rewrite_not.h" #include "iceberg/partition_spec.h" #include "iceberg/row/struct_like.h" #include "iceberg/schema.h" @@ -328,6 +329,7 @@ Result> ResidualEvaluator::Unpartitioned( Result> ResidualEvaluator::Make( std::shared_ptr expr, const PartitionSpec& spec, const Schema& schema, bool case_sensitive) { + ICEBERG_ASSIGN_OR_RAISE(expr, RewriteNot::Visit(std::move(expr))); if (spec.fields().empty()) { return Unpartitioned(std::move(expr)); } diff --git a/src/iceberg/file_reader.h b/src/iceberg/file_reader.h index e08b2df494..13e8b3ad20 100644 --- a/src/iceberg/file_reader.h +++ b/src/iceberg/file_reader.h @@ -80,6 +80,9 @@ 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 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). @@ -107,6 +110,8 @@ struct ICEBERG_EXPORT ReaderOptions { /// \brief The filter to apply to the data. Reader implementations may ignore this if /// the file format does not support filtering. std::shared_ptr filter; + /// \brief Case sensitivity inherited from the scan when resolving field names. + bool case_sensitive = true; /// \brief Name mapping for schema evolution compatibility. Used when reading files /// that may have different field names than the current schema. std::shared_ptr name_mapping; diff --git a/src/iceberg/parquet/parquet_metrics.cc b/src/iceberg/parquet/parquet_metrics.cc index 22710c1372..bad4545a78 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 c78c114ab1..18d5c0999f 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 0000000000..eb543e8723 --- /dev/null +++ b/src/iceberg/parquet/parquet_metrics_row_group_filter.cc @@ -0,0 +1,326 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include +#include +#include +#include +#include + +#include + +#include "iceberg/expression/expression_visitor.h" +#include "iceberg/expression/rewrite_not.h" +#include "iceberg/metadata_columns.h" +#include "iceberg/parquet/parquet_metrics_internal.h" +#include "iceberg/parquet/parquet_metrics_row_group_filter_internal.h" +#include "iceberg/parquet/parquet_schema_util_internal.h" +#include "iceberg/type.h" +#include "iceberg/util/macros.h" + +namespace iceberg::parquet { + +namespace { + +constexpr size_t kInPredicateLimit = 200; +// 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::arrow::SchemaManifest& manifest, + const ::parquet::RowGroupMetaData& row_group, + const std::unordered_map& field_id_to_column_idx) + : manifest_(manifest), + row_group_(row_group), + field_id_to_column_idx_(field_id_to_column_idx) {} + + Result AlwaysTrue() override { return kRowsMightMatch; } + + Result AlwaysFalse() override { return kRowsCannotMatch; } + + Result Not(bool) override { + // Make() applies RewriteNot, so NOT should not reach this visitor. If it does, + // keep the group: "might match" is not "all rows match", so negating it cannot + // prove that no rows match. + return kRowsMightMatch; + } + + 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(std::dynamic_pointer_cast(expr)); + } + + Result NotNull(const std::shared_ptr& expr) override { + if (ContainsNullsOnly(std::dynamic_pointer_cast(expr))) { + return kRowsCannotMatch; + } + return kRowsMightMatch; + } + + Result IsNaN(const std::shared_ptr& expr) override { + 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 { + 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 { + 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 { + 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 { + 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; + } + const auto lower = MinValue(ref); + if (!lower) { + return kRowsMightMatch; + } + if (*lower > value) { + return kRowsCannotMatch; + } + const auto upper = MaxValue(ref); + if (!upper) { + return kRowsMightMatch; + } + return !(*upper < value); + } + + Result NotEq(const std::shared_ptr&, const Literal&) override { + // Like Java, keep negative membership predicates inclusive. + return kRowsMightMatch; + } + + Result In(const std::shared_ptr& expr, + const BoundSetPredicate::LiteralSet& values) override { + const auto ref = std::dynamic_pointer_cast(expr); + if (ContainsNullsOnly(ref)) { + return kRowsCannotMatch; + } + if (values.size() > kInPredicateLimit) { + return kRowsMightMatch; + } + const auto lower = MinValue(ref); + const auto upper = MaxValue(ref); + if (!lower || !upper) { + return kRowsMightMatch; + } + // Equivalent to Java's two-step lower/upper filtering: the group can be + // skipped only if no value falls within [lower, upper]. + const bool any_in_range = std::ranges::any_of( + values, [&](const auto& value) { return !(value < *lower || value > *upper); }); + return any_in_range ? kRowsMightMatch : kRowsCannotMatch; + } + + Result NotIn(const std::shared_ptr&, + const BoundSetPredicate::LiteralSet&) override { + return kRowsMightMatch; + } + + Result StartsWith(const std::shared_ptr& expr, + const Literal& value) override { + const auto ref = std::dynamic_pointer_cast(expr); + if (ContainsNullsOnly(ref)) { + return kRowsCannotMatch; + } + const auto lower = MinValue(ref); + if (!lower) { + return kRowsMightMatch; + } + const auto& prefix = std::get(value.value()); + if (std::get(lower->value()).compare(0, prefix.size(), prefix) > 0) { + return kRowsCannotMatch; + } + const auto upper = MaxValue(ref); + if (!upper) { + 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 ref = std::dynamic_pointer_cast(expr); + if (MayContainNull(ref)) { + return kRowsMightMatch; + } + const auto lower = MinValue(ref); + if (!lower) { + return kRowsMightMatch; + } + const auto& prefix = std::get(value.value()); + if (!std::get(lower->value()).starts_with(prefix)) { + return kRowsMightMatch; + } + const auto upper = MaxValue(ref); + if (!upper) { + return kRowsMightMatch; + } + return !std::get(upper->value()).starts_with(prefix); + } + + private: + bool ContainsNullsOnly(const std::shared_ptr& ref) const { + const auto stats = GetStatistics(ref); + if (!stats || !stats->HasNullCount()) { + return false; + } + const auto column = + row_group_.ColumnChunk(field_id_to_column_idx_.at(ref->field_id())); + return stats->null_count() == column->num_values(); + } + + bool MayContainNull(const std::shared_ptr& ref) const { + const auto stats = GetStatistics(ref); + return !stats || !stats->HasNullCount() || stats->null_count() != 0; + } + + std::shared_ptr<::parquet::Statistics> GetStatistics( + const std::shared_ptr& ref) const { + if (!ref || !ref->type()->is_primitive() || + MetadataColumns::IsMetadataColumn(ref->field_id())) { + return nullptr; + } + auto column = field_id_to_column_idx_.find(ref->field_id()); + // Missing columns can have initial defaults, so do not assume all nulls. + if (column == field_id_to_column_idx_.end()) { + return nullptr; + } + auto field = manifest_.column_index_to_field.find(column->second); + if (field == manifest_.column_index_to_field.end() || + !ValidateParquetTypeCompatibility(*ref->type(), *field->second)) { + return nullptr; + } + 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; + } + const auto& type = static_cast(*ref->type()); + // Parquet floating-point bounds can exclude NaNs. + // Treat both bounds as undefined, preserving null-count pruning. + if (type.type_id() == TypeId::kFloat || type.type_id() == TypeId::kDouble) { + return std::nullopt; + } + auto result = + ParquetMetrics::StatsValueToLiteral(*stats->descr(), type, *stats, is_min); + if (!result || result->IsNaN()) { + return std::nullopt; + } + return std::move(*result); + } + + template + bool VisitInequality(const std::shared_ptr& ref, const Literal& value, + Comparator compare, bool lower_bound) const { + if (ContainsNullsOnly(ref)) { + return kRowsCannotMatch; + } + if (value.IsNaN()) { + return kRowsMightMatch; + } + const auto bound = lower_bound ? MinValue(ref) : MaxValue(ref); + if (!bound) { + return kRowsMightMatch; + } + // An unordered comparison cannot rule out matching rows. + if ((*bound <=> value) == std::partial_ordering::unordered) { + return kRowsMightMatch; + } + return compare(*bound, value); + } + + const ::parquet::arrow::SchemaManifest& manifest_; + const ::parquet::RowGroupMetaData& row_group_; + const std::unordered_map& field_id_to_column_idx_; +}; + +} // namespace + +Result> ParquetMetricsRowGroupFilter::Make( + 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->field_id_to_column_idx_.emplace(id, i); + } + } + result->bound_ = True::Instance(); + if (!filter) { + return result; + } + ICEBERG_ASSIGN_OR_RAISE(result->bound_, RewriteNot::Visit(filter)); + return result; +} + +Result ParquetMetricsRowGroupFilter::ShouldRead( + const ::parquet::arrow::SchemaManifest& manifest, + const ::parquet::RowGroupMetaData& row_group) const { + if (row_group.num_rows() <= 0) { + return kRowsCannotMatch; + } + try { + MetricsVisitor visitor(manifest, row_group, field_id_to_column_idx_); + return Visit(bound_, visitor); + } catch (const ::parquet::ParquetException&) { + // Unusable optional statistics must never turn into false negatives. + return kRowsMightMatch; + } +} + +} // 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 0000000000..5d80fe3bfe --- /dev/null +++ b/src/iceberg/parquet/parquet_metrics_row_group_filter_internal.h @@ -0,0 +1,52 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#pragma once + +#include +#include +#include + +#include + +#include "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 std::shared_ptr& filter, + 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 field_id_to_column_idx_; +}; + +} // namespace iceberg::parquet diff --git a/src/iceberg/parquet/parquet_reader.cc b/src/iceberg/parquet/parquet_reader.cc index 3f15ab7a55..2507d39077 100644 --- a/src/iceberg/parquet/parquet_reader.cc +++ b/src/iceberg/parquet/parquet_reader.cc @@ -20,7 +20,7 @@ #include "iceberg/parquet/parquet_reader.h" #include -#include +#include #include #include @@ -39,7 +39,9 @@ #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" #include "iceberg/parquet/parquet_schema_util_internal.h" #include "iceberg/parquet/row_number_reader_internal.h" @@ -67,13 +69,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; } @@ -291,6 +287,7 @@ class ParquetReader::Impl { split_ = options.split; read_schema_ = options.projection; + stats_filter_.reset(); // Prepare reader properties ::parquet::ReaderProperties reader_properties(pool_); @@ -313,6 +310,18 @@ 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)) { + auto filter = options.filter; + ICEBERG_ASSIGN_OR_RAISE(auto is_bound, IsBoundVisitor::IsBound(filter)); + if (!is_bound) { + ICEBERG_ASSIGN_OR_RAISE( + filter, Binder::Bind(*options.projection, filter, options.case_sensitive)); + } + ICEBERG_ASSIGN_OR_RAISE(stats_filter_, ParquetMetricsRowGroupFilter::Make( + filter, *reader_->manifest().descr)); + } + metadata_context_ = {.file_path = options.path, .first_row_id = options.first_row_id, .data_sequence_number = options.data_sequence_number}; @@ -404,23 +413,28 @@ class ParquetReader::Impl { 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. + // Row group pruning based on the split and zone map auto metadata = reader_->parquet_reader()->metadata(); std::vector row_group_indices; - if (split_.has_value()) { - 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) { + for (int i = 0; i < metadata->num_row_groups(); ++i) { + auto row_group = metadata->RowGroup(i); + if (split_.has_value()) { + auto row_group_offset = row_group->file_offset(); + if (std::cmp_greater_equal(row_group_offset, split_->offset + split_->length)) { break; } + if (std::cmp_less(row_group_offset, split_->offset)) { + continue; + } } - } else { - row_group_indices.resize(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(reader_->manifest(), *row_group)); + if (!should_read) { + continue; + } + } + row_group_indices.push_back(i); } // Create the record batch reader @@ -471,6 +485,8 @@ class ParquetReader::Impl { bool use_large_list_ = false; // Schema to read from the Parquet file. std::shared_ptr<::iceberg::Schema> read_schema_; + // Footer-statistics filter for row group pruning, if enabled. + 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 05f038c133..9ab1251f00 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(); @@ -479,8 +477,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 2eafd013b5..6d9e5a0222 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 292132bb6d..ba230cfc27 100644 --- a/src/iceberg/test/CMakeLists.txt +++ b/src/iceberg/test/CMakeLists.txt @@ -218,6 +218,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/evaluator_test.cc b/src/iceberg/test/evaluator_test.cc index e365a7d3eb..f6d0c5a072 100644 --- a/src/iceberg/test/evaluator_test.cc +++ b/src/iceberg/test/evaluator_test.cc @@ -705,6 +705,41 @@ TEST_F(EvaluatorTest, Not) { *struct_evaluator, true); } +TEST_F(EvaluatorTest, NotRangeComparisonsExcludeNull) { + struct Case { + std::shared_ptr predicate; + bool below; + bool equal; + bool above; + }; + const std::vector cases = { + {.predicate = Expressions::LessThan("z", Literal::Int(7)), + .below = false, + .equal = true, + .above = true}, + {.predicate = Expressions::LessThanOrEqual("z", Literal::Int(7)), + .below = false, + .equal = false, + .above = true}, + {.predicate = Expressions::GreaterThan("z", Literal::Int(7)), + .below = true, + .equal = true, + .above = false}, + {.predicate = Expressions::GreaterThanOrEqual("z", Literal::Int(7)), + .below = true, + .equal = false, + .above = false}}; + for (const auto& test : cases) { + SCOPED_TRACE(test.predicate->ToString()); + ICEBERG_UNWRAP_OR_FAIL(auto evaluator, + Evaluator::Make(*schema_, Expressions::Not(test.predicate))); + TestData(R"([{"x": 0, "y": 0.0, "z": null}])", *evaluator, false); + TestData(R"([{"x": 0, "y": 0.0, "z": 6}])", *evaluator, test.below); + TestData(R"([{"x": 0, "y": 0.0, "z": 7}])", *evaluator, test.equal); + TestData(R"([{"x": 0, "y": 0.0, "z": 8}])", *evaluator, test.above); + } +} + TEST_F(EvaluatorTest, CaseInsensitiveNot) { // Use case-insensitive binding (false) ICEBERG_UNWRAP_OR_FAIL( diff --git a/src/iceberg/test/file_scan_task_reader_test.cc b/src/iceberg/test/file_scan_task_reader_test.cc index 11366f72d9..2832ceba72 100644 --- a/src/iceberg/test/file_scan_task_reader_test.cc +++ b/src/iceberg/test/file_scan_task_reader_test.cc @@ -41,8 +41,11 @@ #include "iceberg/arrow_c_data_util_internal.h" #include "iceberg/data/equality_delete_writer.h" #include "iceberg/data/position_delete_writer.h" +#include "iceberg/data_table_scan.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" @@ -695,4 +698,196 @@ TEST_F(FileScanTaskReaderTest, OpenWithMixedDeletesSkipsFullyDeletedBatches) { ASSERT_NO_FATAL_FAILURE(VerifyStream(&stream, R"([[3, "Baz"]])")); } +TEST_F(FileScanTaskReaderTest, RowGroupPruningPreservesDeletesAndLineage) { + auto lineage_schema = RowLineageProjection(); + std::vector read_fields(lineage_schema->fields().begin(), + lineage_schema->fields().end()); + read_fields.push_back(table_schema_->fields()[2]); + auto read_schema = std::make_shared(std::move(read_fields)); + 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 = read_schema, + }; + 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, RowGroupPruningUsesCallerReadSchema) { + 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 = table_schema_, + }; + options.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","keep"],[3,"d","keep"]])"); + } +} + +TEST_F(FileScanTaskReaderTest, ConstantResidualsAndNestedConstants) { + ICEBERG_UNWRAP_OR_FAIL( + auto data_file, MakeDataFile(table_schema_, R"([[0,"a","keep"],[1,"b","keep"]])")); + auto predicate = Expressions::Equal("name", Literal::String("a")); + ICEBERG_UNWRAP_OR_FAIL(std::shared_ptr conjunction, + And::Make(Expressions::AlwaysTrue(), predicate)); + ICEBERG_UNWRAP_OR_FAIL(std::shared_ptr disjunction, + Or::Make(Expressions::AlwaysFalse(), predicate)); + ICEBERG_UNWRAP_OR_FAIL( + auto reader, FileScanTaskReader::Make({.io = file_io_, + .table_schema = table_schema_, + .projected_schema = projected_schema_})); + for (const auto& filter : std::vector>{ + Expressions::AlwaysTrue(), conjunction, disjunction}) { + SCOPED_TRACE(filter->ToString()); + FileScanTask task(data_file, {}, filter); + ICEBERG_UNWRAP_OR_FAIL(auto stream, reader->Open(task)); + VerifyStream(&stream, R"([[0,"a"],[1,"b"]])"); + } + FileScanTask empty_task(data_file, {}, Expressions::AlwaysFalse()); + ICEBERG_UNWRAP_OR_FAIL(auto stream, reader->Open(empty_task)); + auto batches = ::arrow::ImportRecordBatchReader(&stream).ValueOrDie(); + EXPECT_EQ(batches->schema()->num_fields(), 2); + EXPECT_EQ(batches->Next().ValueOrDie(), nullptr); +} + +TEST_F(FileScanTaskReaderTest, ScanCaseSensitivityAndProjection) { + ICEBERG_UNWRAP_OR_FAIL( + auto data_file, MakeDataFile(table_schema_, R"([[0,"a","keep"],[1,"b","keep"]])")); + auto filter = Expressions::Equal("CATEGORY", Literal::String("keep")); + for (bool select_columns : {false, true}) { + internal::TableScanContext context; + context.filter = filter; + context.case_sensitive = false; + if (select_columns) { + context.selected_columns = std::vector{"id", "name"}; + } else { + // Explicit projections are prepared by the caller, including filter columns. + context.projected_schema = table_schema_; + } + auto metadata = std::make_shared(); + metadata->schemas = {table_schema_}; + metadata->current_schema_id = table_schema_->schema_id(); + ICEBERG_UNWRAP_OR_FAIL( + auto scan, DataTableScan::Make(metadata, table_schema_, file_io_, context)); + ICEBERG_UNWRAP_OR_FAIL(auto projection, scan->schema()); + FileScanTaskReader::Options options{ + .io = scan->io(), + .table_schema = table_schema_, + .projected_schema = projection, + .case_sensitive = scan->is_case_sensitive(), + }; + FileScanTask task(data_file, {}, scan->filter()); + ICEBERG_UNWRAP_OR_FAIL(auto reader, FileScanTaskReader::Make(options)); + ICEBERG_UNWRAP_OR_FAIL(auto stream, reader->Open(task)); + VerifyStream(&stream, R"([[0,"a","keep"],[1,"b","keep"]])"); + options.case_sensitive = true; + ICEBERG_UNWRAP_OR_FAIL(auto case_sensitive_reader, FileScanTaskReader::Make(options)); + EXPECT_THAT(case_sensitive_reader->Open(task), + IsError(ErrorKind::kInvalidExpression)); + } +} + +TEST_F(FileScanTaskReaderTest, UnboundFilterRequiresProjectedField) { + ICEBERG_UNWRAP_OR_FAIL( + auto data_file, MakeDataFile(table_schema_, R"([[0,"a","keep"],[1,"b","keep"]])")); + ICEBERG_UNWRAP_OR_FAIL( + auto reader, FileScanTaskReader::Make({.io = file_io_, + .table_schema = table_schema_, + .projected_schema = projected_schema_})); + FileScanTask task(data_file, {}, + Expressions::Equal("category", Literal::String("keep"))); + EXPECT_THAT(reader->Open(task), IsError(ErrorKind::kInvalidExpression)); +} + +TEST_F(FileScanTaskReaderTest, NestedFilterFieldsComeFromScanSchema) { + table_schema_ = std::make_shared(std::vector{ + SchemaField::MakeRequired(1, "id", int32()), + SchemaField::MakeOptional(2, "record", + std::make_shared(std::vector{ + SchemaField::MakeOptional(3, "value", string()), + SchemaField::MakeOptional(4, "key", int32())}))}); + internal::TableScanContext context; + context.selected_columns = std::vector{"id", "record.value"}; + context.filter = Expressions::GreaterThan("record.key", Literal::Int(5)); + auto metadata = std::make_shared(); + metadata->schemas = {table_schema_}; + metadata->current_schema_id = table_schema_->schema_id(); + ICEBERG_UNWRAP_OR_FAIL(auto scan, + DataTableScan::Make(metadata, table_schema_, file_io_, context)); + ICEBERG_UNWRAP_OR_FAIL(projected_schema_, scan->schema()); + ICEBERG_UNWRAP_OR_FAIL( + auto data_file, + MakeDataFile(table_schema_, + std::vector{R"([[0,["a",0]],[1,["b",1]]])", + R"([[2,["c",10]],[3,["d",11]]])"}, + 4, 2)); + ICEBERG_UNWRAP_OR_FAIL( + auto pos_delete, MakePositionDeleteFile(CreateNewTempFilePathWithSuffix(".parquet"), + {2}, data_file->file_path)); + for (bool with_deletes : {false, true}) { + FileScanTask task( + data_file, + with_deletes ? std::vector{pos_delete} : std::vector>{}, + scan->filter()); + ICEBERG_UNWRAP_OR_FAIL( + auto reader, FileScanTaskReader::Make({.io = file_io_, + .table_schema = table_schema_, + .projected_schema = projected_schema_})); + ICEBERG_UNWRAP_OR_FAIL(auto stream, reader->Open(task)); + VerifyStream(&stream, + with_deletes ? R"([[3,["d",11]]])" : R"([[2,["c",10]],[3,["d",11]]])"); + } +} + } // 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 0000000000..9c14cfdada --- /dev/null +++ b/src/iceberg/test/parquet_row_group_filter_test.cc @@ -0,0 +1,745 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "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" +#include "iceberg/util/uuid.h" + +namespace iceberg::parquet { +namespace { + +class ParquetRowGroupFilterTest : public ::testing::Test { + protected: + void SetUp() override { + parquet::RegisterAll(); + io_ = std::make_shared(); + SetKeyType(int32()); + projection_ = std::make_shared( + std::vector{SchemaField::MakeRequired(2, "value", int64()), + MetadataColumns::kRowPosition, MetadataColumns::kRowId}); + } + + void SetKeyType(std::shared_ptr type) { + schema_ = std::make_shared( + std::vector{SchemaField::MakeOptional(1, "key", std::move(type)), + SchemaField::MakeRequired(2, "value", int64())}); + } + + Status Write(bool statistics = 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 Write(table, statistics); + } + + Status Write(const std::shared_ptr<::arrow::Table>& table, bool statistics = true) { + ICEBERG_ASSIGN_OR_RAISE(auto out, arrow::OpenArrowOutputStream(io_, path_)); + ::parquet::WriterProperties::Builder properties; + properties.disable_dictionary(); + 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) { + // Bind against the table schema: key is intentionally outside the projection. + // Tests of Open's binding behavior set options.filter directly instead. + if (filter && filter->op() != Expression::Operation::kTrue && + filter->op() != Expression::Operation::kFalse) { + auto is_bound = IsBoundVisitor::IsBound(filter); + EXPECT_THAT(is_bound, IsOk()); + if (is_bound && !*is_bound) { + auto bound = Binder::Bind(*schema_, filter, case_sensitive); + EXPECT_THAT(bound, IsOk()); + if (bound) { + filter = *bound; + } + } + } + ReaderOptions options{.path = path_, + .io = io_, + .projection = projection_, + .filter = std::move(filter), + .first_row_id = 100}; + options.case_sensitive = case_sensitive; + options.properties.Set(ReaderProperties::kBatchSize, int64_t{3}); + return options; + } + + Result> Read(ReaderOptions options) { + ICEBERG_ASSIGN_OR_RAISE( + auto reader, ReaderFactoryRegistry::Open(FileFormatType::kParquet, options)); + ICEBERG_ASSIGN_OR_RAISE(auto c_schema, reader->Schema()); + ICEBERG_ARROW_ASSIGN_OR_RETURN(auto schema, ::arrow::ImportSchema(&c_schema)); + std::vector positions; + while (true) { + ICEBERG_ASSIGN_OR_RAISE(auto array, reader->Next()); + if (!array) { + break; + } + ICEBERG_ARROW_ASSIGN_OR_RETURN(auto batch, + ::arrow::ImportRecordBatch(&*array, schema)); + auto values = std::static_pointer_cast<::arrow::Int64Array>(batch->column(0)); + auto pos = std::static_pointer_cast<::arrow::Int64Array>(batch->column(1)); + auto ids = std::static_pointer_cast<::arrow::Int64Array>(batch->column(2)); + for (int64_t i = 0; i < batch->num_rows(); ++i) { + EXPECT_EQ(values->Value(i), pos->Value(i)); + EXPECT_EQ(ids->Value(i), 100 + pos->Value(i)); + positions.push_back(pos->Value(i)); + } + } + ICEBERG_RETURN_UNEXPECTED(reader->Close()); + return positions; + } + + void Check(const ReaderOptions& options, const std::vector& expected) { + SCOPED_TRACE(options.filter ? options.filter->ToString() : "no filter"); + ICEBERG_UNWRAP_OR_FAIL(auto actual, Read(options)); + EXPECT_EQ(actual, expected); + } + + 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, 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))), {}); + 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)), + 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); + 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.case_sensitive = false; + Check(options, {4, 5}); +} + +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()); + auto options = Options(nullptr); + options.filter = Expressions::Equal("missing", Literal::Int(0)); + EXPECT_THAT(ReaderFactoryRegistry::Open(FileFormatType::kParquet, options), + HasErrorMessage("Cannot find field 'missing'")); + 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{ + schema_->fields()[0], schema_->fields()[1], + SchemaField::MakeOptional(3, "defaulted", int32()) + .WithInitialDefault(std::make_shared(Literal::Int(7)))}); + 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{{.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()); + SetKeyType(test.file_type); + ASSERT_THAT(Write(true, test.json), IsOk()); + SetKeyType(test.filter_type); + Check(Options(Expressions::Equal("key", test.value)), + test.filter_type->type_id() != TypeId::kFloat && + test.filter_type->type_id() != TypeId::kDouble && 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(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, RewriteNotForBoundPredicates) { + 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}); + + // 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}); + Check(Options(Expressions::Not(Expressions::NotNull("key"))), {0, 1}); +} + +TEST_F(ParquetRowGroupFilterTest, RewrittenNotPrunesNullableGroups) { + ASSERT_THAT(Write(true, "[[null,0],[null,1],[null,2],[50,3],[100,4],[150,5]]"), IsOk()); + // Match Evaluator's normalized predicates: null does not match these ranges. + Check(Options(Expressions::Not(Expressions::LessThan("key", Literal::Int(100)))), + {4, 5}); + Check(Options(Expressions::Not(Expressions::LessThanOrEqual("key", Literal::Int(100)))), + {4, 5}); + Check(Options(Expressions::Not(Expressions::GreaterThan("key", Literal::Int(100)))), + {2, 3, 4, 5}); + Check(Options( + Expressions::Not(Expressions::GreaterThanOrEqual("key", Literal::Int(100)))), + {2, 3}); +} + +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{ + {.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}}, + }) { + Check(Options(test.filter), 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}); + 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)); + } + 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{.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{.offset = static_cast(offset), .length = 0}; + Check(options, {}); +} + +TEST_F(ParquetRowGroupFilterTest, PrimitiveComparisonTypes) { + struct Case { + std::shared_ptr type; + std::string json; + Literal literal; + }; + for (const auto& test : std::vector{ + {.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'})}}) { + 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, UnorderedUuidBoundsRetainGroups) { + SetKeyType(uuid()); + ICEBERG_UNWRAP_OR_FAIL(auto lower, + Uuid::FromString("00000000-0000-0000-0000-000000000001")); + ICEBERG_UNWRAP_OR_FAIL(auto upper, + Uuid::FromString("00000000-0000-0000-0000-000000000002")); + ::arrow::FixedSizeBinaryBuilder keys(::arrow::fixed_size_binary(Uuid::kLength)); + ASSERT_TRUE(keys.Append(lower.bytes().data()).ok()); + ASSERT_TRUE(keys.Append(upper.bytes().data()).ok()); + auto key_array = ::arrow::ExtensionType::WrapArray(::arrow::extension::uuid(), + keys.Finish().ValueOrDie()); + auto values = + ::arrow::json::ArrayFromJSONString(::arrow::int64(), "[0,1]").ValueOrDie(); + ArrowSchema c_schema; + ASSERT_THAT(ToArrowSchema(*schema_, &c_schema), IsOk()); + auto arrow_schema = ::arrow::ImportSchema(&c_schema).ValueOrDie(); + ASSERT_THAT(Write(::arrow::Table::Make(arrow_schema, {key_array, values})), IsOk()); + ASSERT_TRUE(metadata_->RowGroup(0)->ColumnChunk(0)->statistics()->HasMinMax()); + + auto options = Options(Expressions::LessThanOrEqual("key", Literal::UUID(upper))); + auto predicate = std::dynamic_pointer_cast(options.filter); + ASSERT_NE(predicate, nullptr); + EXPECT_THAT(predicate->Test(Literal::UUID(upper)), HasValue(testing::Eq(true))); + Check(options, {0, 1}); + Check(Options(Expressions::GreaterThanOrEqual("key", Literal::UUID(lower))), {0, 1}); + Check(Options(Expressions::LessThan("key", Literal::UUID(upper))), {0, 1}); + Check(Options(Expressions::GreaterThan("key", Literal::UUID(lower))), {0, 1}); +} + +TEST_F(ParquetRowGroupFilterTest, FloatingPointStatistics) { + SetKeyType(float64()); + ASSERT_THAT(Write(), IsOk()); + Check(Options(Expressions::Equal("key", Literal::Double(100))), {0, 1, 2, 3, 4, 5}); + Check(Options(Expressions::Equal("key", Literal::Double(10))), {0, 1, 2, 3, 4, 5}); + Check(Options(Expressions::IsNaN("key")), {0, 1, 2, 3, 4, 5}); +} + +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}); +} + +TEST_F(ParquetRowGroupFilterTest, RequiredLeafWithNullableParent) { + SetKeyType(std::make_shared( + std::vector{SchemaField::MakeRequired(3, "nested", int32())})); + ASSERT_THAT(Write(true, "[[null,0],[null,1],[[10],2],[[11],3],[null,4],[null,5]]"), + IsOk()); + auto is_null = Expressions::IsNull("key.nested"); + auto not_null = Expressions::NotNull("key.nested"); + Check(Options(is_null), {0, 1, 4, 5}); + Check(Options(not_null), {2, 3}); + Check(Options(Expressions::Not(is_null)), {2, 3}); + Check(Options(Expressions::Not(not_null)), {0, 1, 4, 5}); + Check(Options(Expressions::Not(Expressions::LessThan("key.nested", Literal::Int(100)))), + {}); +} + +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, 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)))), {}); +} + +TEST_F(ParquetRowGroupFilterTest, PrefixPredicates) { + SetKeyType(string()); + 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}); + 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, Utf8StringBoundsUseUnsignedByteOrder) { + SetKeyType(string()); + // RG2 holds multi-byte UTF-8 values (U+00E9, U+00FC) whose leading byte 0xC3 + // sorts after ASCII only under unsigned byte comparison. + ASSERT_THAT( + Write(true, R"([["a",0],["b",1],["y",2],["z",3],["\u00e9",4],["\u00fc",5]])"), + IsOk()); + ASSERT_EQ(metadata_->num_row_groups(), 3); + const std::string e_acute = "\xC3\xA9"; + const std::string u_umlaut = "\xC3\xBC"; + Check(Options(Expressions::GreaterThan("key", Literal::String("z"))), {4, 5}); + Check(Options(Expressions::LessThan("key", Literal::String("c"))), {0, 1}); + Check(Options(Expressions::Equal("key", Literal::String(e_acute))), {4, 5}); + Check(Options(Expressions::StartsWith("key", e_acute)), {4, 5}); + Check(Options(Expressions::StartsWith("key", u_umlaut)), {4, 5}); +} + +TEST_F(ParquetRowGroupFilterTest, DecimalAndTemporalStatistics) { + struct Case { + std::shared_ptr type; + std::string json; + Literal value; + }; + for ( + const auto& test : std::vector{ + {.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)}}) { + 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, 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, 4, 5}); + Check(Options(Expressions::Equal("key", Literal::Double(0.0))), {0, 1, 4, 5}); + 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))), {0, 1, 4, 5}); +} + +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, 2, 3}); + 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, 2, 3}); +} + +TEST_F(ParquetRowGroupFilterTest, InvalidFilterFailsDuringOpen) { + ASSERT_THAT(Write(), IsOk()); + auto options = Options(nullptr); + options.filter = Expressions::Count("value"); + EXPECT_THAT(ReaderFactoryRegistry::Open(FileFormatType::kParquet, options), + HasErrorMessage("does not support bound 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}); +} + +TEST_F(ParquetRowGroupFilterTest, FloatBoundsMustNotHideNaNs) { + SetKeyType(float32()); + ASSERT_THAT(Write(true, "[[NaN,0],[2.0,1],[-NaN,2],[2.0,3]]"), IsOk()); + ASSERT_EQ(metadata_->num_row_groups(), 2); + + // Both groups contain NaN, but Parquet records min = max = 2. + for (int group : {0, 1}) { + auto stats = std::static_pointer_cast<::parquet::FloatStatistics>( + metadata_->RowGroup(group)->ColumnChunk(0)->statistics()); + ASSERT_NE(stats, nullptr); + ASSERT_TRUE(stats->HasMinMax()); + EXPECT_EQ(stats->min(), 2.0F); + EXPECT_EQ(stats->max(), 2.0F); + } + + Check(Options(Expressions::GreaterThan("key", Literal::Float(3))), {0, 1, 2, 3}); + Check(Options(Expressions::LessThan("key", Literal::Float(-3))), {0, 1, 2, 3}); +} + +TEST_F(ParquetRowGroupFilterTest, DoubleBoundsMustNotHideNaNs) { + SetKeyType(float64()); + ASSERT_THAT(Write(true, "[[NaN,0],[2.0,1],[-NaN,2],[2.0,3]]"), IsOk()); + ASSERT_EQ(metadata_->num_row_groups(), 2); + + for (int group : {0, 1}) { + auto stats = std::static_pointer_cast<::parquet::DoubleStatistics>( + metadata_->RowGroup(group)->ColumnChunk(0)->statistics()); + ASSERT_NE(stats, nullptr); + ASSERT_TRUE(stats->HasMinMax()); + EXPECT_EQ(stats->min(), 2.0); + EXPECT_EQ(stats->max(), 2.0); + } + + Check(Options(Expressions::GreaterThan("key", Literal::Double(3))), {0, 1, 2, 3}); + Check(Options(Expressions::LessThan("key", Literal::Double(-3))), {0, 1, 2, 3}); +} + +} // namespace +} // namespace iceberg::parquet diff --git a/src/iceberg/test/residual_evaluator_test.cc b/src/iceberg/test/residual_evaluator_test.cc index d2aa2aeb0f..5e804f2fc7 100644 --- a/src/iceberg/test/residual_evaluator_test.cc +++ b/src/iceberg/test/residual_evaluator_test.cc @@ -219,6 +219,54 @@ TEST_F(ResidualEvaluatorTest, CaseInsensitiveIdentityTransformResiduals) { EXPECT_EQ(residual4->op(), Expression::Operation::kFalse); } +TEST_F(ResidualEvaluatorTest, ComparisonWithNullValue) { + auto schema = std::make_shared( + std::vector{SchemaField::MakeOptional(50, "key", int32()), + SchemaField::MakeOptional(51, "value", int32())}); + + PartitionField pt_field(50, 1000, "key", Transform::Identity()); + ICEBERG_UNWRAP_OR_FAIL(auto spec_unique, + PartitionSpec::Make(*schema, 0, {pt_field}, false)); + auto spec = std::shared_ptr(spec_unique.release()); + + // NOT(key < 100) OR value = 5 + auto expr = + Expressions::Or(Expressions::Not(Expressions::LessThan("key", Literal::Int(100))), + Expressions::Equal("value", Literal::Int(5))); + ICEBERG_UNWRAP_OR_FAIL(auto evaluator, + ResidualEvaluator::Make(expr, *spec, *schema, true)); + + // A null partition value must not satisfy NOT(key < 100); otherwise the residual + // folds to true and drops value = 5, disagreeing with Evaluator. + PartitionValues null_partition(Literal::Null(int32())); + ICEBERG_UNWRAP_OR_FAIL(auto null_residual, evaluator->ResidualFor(null_partition)); + ASSERT_TRUE(null_residual->is_unbound_predicate()); + auto unbound = std::dynamic_pointer_cast(null_residual); + ASSERT_NE(unbound, nullptr); + EXPECT_EQ(unbound->op(), Expression::Operation::kEq); + EXPECT_EQ(unbound->reference()->name(), "value"); + + PartitionValues below(Literal::Int(50)); + ICEBERG_UNWRAP_OR_FAIL(auto below_residual, evaluator->ResidualFor(below)); + ASSERT_TRUE(below_residual->is_unbound_predicate()); + EXPECT_EQ(below_residual->op(), Expression::Operation::kEq); + + PartitionValues above(Literal::Int(150)); + ICEBERG_UNWRAP_OR_FAIL(auto above_residual, evaluator->ResidualFor(above)); + EXPECT_EQ(above_residual->op(), Expression::Operation::kTrue); + + ICEBERG_UNWRAP_OR_FAIL( + auto unpartitioned, + ResidualEvaluator::Make(expr, *PartitionSpec::Unpartitioned(), *schema, true)); + PartitionValues empty_partition; + ICEBERG_UNWRAP_OR_FAIL(auto unpartitioned_residual, + unpartitioned->ResidualFor(empty_partition)); + ASSERT_EQ(unpartitioned_residual->op(), Expression::Operation::kOr); + auto unpartitioned_or = std::dynamic_pointer_cast(unpartitioned_residual); + ASSERT_NE(unpartitioned_or, nullptr); + EXPECT_EQ(unpartitioned_or->left()->op(), Expression::Operation::kGtEq); +} + TEST_F(ResidualEvaluatorTest, UnpartitionedResiduals) { std::vector> expressions = { Expressions::AlwaysTrue(), From 3835dfeeb92bbe71bbd8a7af0ffdd52ed52254af Mon Sep 17 00:00:00 2001 From: yinzhengsun Date: Fri, 16 Oct 2026 16:34:49 +0800 Subject: [PATCH 2/5] refactor(parquet): simplify split row group selection --- src/iceberg/parquet/parquet_reader.cc | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/src/iceberg/parquet/parquet_reader.cc b/src/iceberg/parquet/parquet_reader.cc index 2507d39077..fde3e57ac6 100644 --- a/src/iceberg/parquet/parquet_reader.cc +++ b/src/iceberg/parquet/parquet_reader.cc @@ -420,10 +420,9 @@ class ParquetReader::Impl { auto row_group = metadata->RowGroup(i); if (split_.has_value()) { auto row_group_offset = row_group->file_offset(); - if (std::cmp_greater_equal(row_group_offset, split_->offset + split_->length)) { - break; - } - if (std::cmp_less(row_group_offset, split_->offset)) { + bool selected = std::cmp_greater_equal(row_group_offset, split_->offset) && + std::cmp_less(row_group_offset, split_->offset + split_->length); + if (!selected) { continue; } } From ab0fbd17fea77648c4b9d1ab03eb4cafd0997082 Mon Sep 17 00:00:00 2001 From: yinzhengsun Date: Mon, 19 Oct 2026 16:48:15 +0800 Subject: [PATCH 3/5] fix comments --- src/iceberg/expression/literal.cc | 10 ++--- src/iceberg/row/partition_values.cc | 9 ++++ src/iceberg/test/literal_test.cc | 11 ++++- .../test/parquet_row_group_filter_test.cc | 43 ++++++++++++------- src/iceberg/test/residual_evaluator_test.cc | 27 ++++++++++++ 5 files changed, 76 insertions(+), 24 deletions(-) diff --git a/src/iceberg/expression/literal.cc b/src/iceberg/expression/literal.cc index 88f4b57404..c8dd74a448 100644 --- a/src/iceberg/expression/literal.cc +++ b/src/iceberg/expression/literal.cc @@ -24,6 +24,7 @@ #include #include #include +#include #include #include #include @@ -710,12 +711,9 @@ std::partial_ordering Literal::operator<=>(const Literal& other) const { } case TypeId::kUuid: { - auto& this_val = std::get(value_); - auto& other_val = std::get(other.value_); - if (this_val == other_val) { - return std::partial_ordering::equivalent; - } - return std::partial_ordering::unordered; + auto this_bytes = std::get(value_).bytes(); + auto other_bytes = std::get(other.value_).bytes(); + return std::memcmp(this_bytes.data(), other_bytes.data(), Uuid::kLength) <=> 0; } case TypeId::kBinary: diff --git a/src/iceberg/row/partition_values.cc b/src/iceberg/row/partition_values.cc index fcd4691d01..edf55a7725 100644 --- a/src/iceberg/row/partition_values.cc +++ b/src/iceberg/row/partition_values.cc @@ -19,6 +19,10 @@ #include "iceberg/row/partition_values.h" +#include + +#include "iceberg/util/uuid.h" + namespace iceberg { PartitionValues& PartitionValues::operator=(const PartitionValues& other) { @@ -74,6 +78,11 @@ Result PartitionValues::GetField(size_t pos) const { return Scalar{ std::string_view(reinterpret_cast(bytes.data()), bytes.size())}; } + case TypeId::kUuid: { + const auto bytes = std::get(literal.value()).bytes(); + return Scalar{ + std::string_view(reinterpret_cast(bytes.data()), bytes.size())}; + } case TypeId::kDecimal: return Scalar{std::get(literal.value())}; default: diff --git a/src/iceberg/test/literal_test.cc b/src/iceberg/test/literal_test.cc index 6f515cdf71..60bd8f3ec5 100644 --- a/src/iceberg/test/literal_test.cc +++ b/src/iceberg/test/literal_test.cc @@ -414,8 +414,15 @@ TEST(LiteralTest, UuidComparison) { auto literal3 = Literal::UUID(uuid3); EXPECT_EQ(literal1 <=> literal3, std::partial_ordering::equivalent); - EXPECT_EQ(literal1 <=> literal2, std::partial_ordering::unordered); - EXPECT_EQ(literal2 <=> literal1, std::partial_ordering::unordered); + EXPECT_EQ(literal1 <=> literal2, std::partial_ordering::less); + EXPECT_EQ(literal2 <=> literal1, std::partial_ordering::greater); + + auto low = + Literal::UUID(Uuid::FromString("7fffffff-ffff-ffff-ffff-ffffffffffff").value()); + auto high = + Literal::UUID(Uuid::FromString("80000000-0000-0000-0000-000000000000").value()); + EXPECT_EQ(low <=> high, std::partial_ordering::less); + EXPECT_EQ(high <=> low, std::partial_ordering::greater); } // Parameter struct for literal serialization and deserialization tests diff --git a/src/iceberg/test/parquet_row_group_filter_test.cc b/src/iceberg/test/parquet_row_group_filter_test.cc index 9c14cfdada..002c25365d 100644 --- a/src/iceberg/test/parquet_row_group_filter_test.cc +++ b/src/iceberg/test/parquet_row_group_filter_test.cc @@ -526,36 +526,47 @@ TEST_F(ParquetRowGroupFilterTest, PrimitiveComparisonTypes) { } } -TEST_F(ParquetRowGroupFilterTest, UnorderedUuidBoundsRetainGroups) { +TEST_F(ParquetRowGroupFilterTest, UuidBoundsUseUnsignedByteOrder) { SetKeyType(uuid()); - ICEBERG_UNWRAP_OR_FAIL(auto lower, - Uuid::FromString("00000000-0000-0000-0000-000000000001")); - ICEBERG_UNWRAP_OR_FAIL(auto upper, - Uuid::FromString("00000000-0000-0000-0000-000000000002")); + // Group 0 holds UUIDs below 0x80..., group 1 holds UUIDs at or above it. A signed + // byte comparison would order group 1 before group 0. + std::vector uuids; + for (const auto* str : + {"00000000-0000-0000-0000-000000000001", "7fffffff-ffff-ffff-ffff-ffffffffffff", + "80000000-0000-0000-0000-000000000000", "ffffffff-ffff-ffff-ffff-fffffffffffe"}) { + ICEBERG_UNWRAP_OR_FAIL(auto uuid, Uuid::FromString(str)); + uuids.push_back(uuid); + } ::arrow::FixedSizeBinaryBuilder keys(::arrow::fixed_size_binary(Uuid::kLength)); - ASSERT_TRUE(keys.Append(lower.bytes().data()).ok()); - ASSERT_TRUE(keys.Append(upper.bytes().data()).ok()); + for (const auto& uuid : uuids) { + ASSERT_TRUE(keys.Append(uuid.bytes().data()).ok()); + } auto key_array = ::arrow::ExtensionType::WrapArray(::arrow::extension::uuid(), keys.Finish().ValueOrDie()); auto values = - ::arrow::json::ArrayFromJSONString(::arrow::int64(), "[0,1]").ValueOrDie(); + ::arrow::json::ArrayFromJSONString(::arrow::int64(), "[0,1,2,3]").ValueOrDie(); ArrowSchema c_schema; ASSERT_THAT(ToArrowSchema(*schema_, &c_schema), IsOk()); auto arrow_schema = ::arrow::ImportSchema(&c_schema).ValueOrDie(); ASSERT_THAT(Write(::arrow::Table::Make(arrow_schema, {key_array, values})), IsOk()); + ASSERT_EQ(metadata_->num_row_groups(), 2); ASSERT_TRUE(metadata_->RowGroup(0)->ColumnChunk(0)->statistics()->HasMinMax()); - auto options = Options(Expressions::LessThanOrEqual("key", Literal::UUID(upper))); - auto predicate = std::dynamic_pointer_cast(options.filter); - ASSERT_NE(predicate, nullptr); - EXPECT_THAT(predicate->Test(Literal::UUID(upper)), HasValue(testing::Eq(true))); - Check(options, {0, 1}); - Check(Options(Expressions::GreaterThanOrEqual("key", Literal::UUID(lower))), {0, 1}); - Check(Options(Expressions::LessThan("key", Literal::UUID(upper))), {0, 1}); - Check(Options(Expressions::GreaterThan("key", Literal::UUID(lower))), {0, 1}); + const auto& group0_max = uuids[1]; + const auto& group1_min = uuids[2]; + Check(Options(Expressions::LessThan("key", Literal::UUID(group1_min))), {0, 1}); + Check(Options(Expressions::LessThanOrEqual("key", Literal::UUID(group0_max))), {0, 1}); + Check(Options(Expressions::GreaterThan("key", Literal::UUID(group0_max))), {2, 3}); + Check(Options(Expressions::GreaterThanOrEqual("key", Literal::UUID(group1_min))), + {2, 3}); + Check(Options(Expressions::Equal("key", Literal::UUID(group1_min))), {2, 3}); + Check( + Options(Expressions::Not(Expressions::LessThan("key", Literal::UUID(group1_min)))), + {2, 3}); } TEST_F(ParquetRowGroupFilterTest, FloatingPointStatistics) { + // TODO: remove this once row group pruning is applied to float/double columns. SetKeyType(float64()); ASSERT_THAT(Write(), IsOk()); Check(Options(Expressions::Equal("key", Literal::Double(100))), {0, 1, 2, 3, 4, 5}); diff --git a/src/iceberg/test/residual_evaluator_test.cc b/src/iceberg/test/residual_evaluator_test.cc index 5e804f2fc7..4ca9f00a66 100644 --- a/src/iceberg/test/residual_evaluator_test.cc +++ b/src/iceberg/test/residual_evaluator_test.cc @@ -35,6 +35,7 @@ #include "iceberg/test/matchers.h" #include "iceberg/transform.h" #include "iceberg/type.h" +#include "iceberg/util/uuid.h" namespace iceberg { @@ -267,6 +268,32 @@ TEST_F(ResidualEvaluatorTest, ComparisonWithNullValue) { EXPECT_EQ(unpartitioned_or->left()->op(), Expression::Operation::kGtEq); } +TEST_F(ResidualEvaluatorTest, NotUuidComparison) { + auto schema = std::make_shared( + std::vector{SchemaField::MakeOptional(50, "key", uuid()), + SchemaField::MakeOptional(51, "value", int32())}); + PartitionField pt_field(50, 1000, "key", Transform::Identity()); + ICEBERG_UNWRAP_OR_FAIL(auto spec_unique, + PartitionSpec::Make(*schema, 0, {pt_field}, false)); + auto spec = std::shared_ptr(spec_unique.release()); + + ICEBERG_UNWRAP_OR_FAIL(auto lower, + Uuid::FromString("7fffffff-ffff-ffff-ffff-ffffffffffff")); + ICEBERG_UNWRAP_OR_FAIL(auto upper, + Uuid::FromString("80000000-0000-0000-0000-000000000000")); + auto expr = Expressions::Not(Expressions::LessThan("key", Literal::UUID(lower))); + ICEBERG_UNWRAP_OR_FAIL(auto evaluator, + ResidualEvaluator::Make(expr, *spec, *schema, true)); + + PartitionValues above(Literal::UUID(upper)); + ICEBERG_UNWRAP_OR_FAIL(auto above_residual, evaluator->ResidualFor(above)); + EXPECT_EQ(above_residual->op(), Expression::Operation::kTrue); + + PartitionValues equal(Literal::UUID(lower)); + ICEBERG_UNWRAP_OR_FAIL(auto equal_residual, evaluator->ResidualFor(equal)); + EXPECT_EQ(equal_residual->op(), Expression::Operation::kTrue); +} + TEST_F(ResidualEvaluatorTest, UnpartitionedResiduals) { std::vector> expressions = { Expressions::AlwaysTrue(), From 9e63f0aa27e6a9fb784ae5d94bbd0a49c5c7651d Mon Sep 17 00:00:00 2001 From: yinzhengsun Date: Mon, 19 Oct 2026 17:37:39 +0800 Subject: [PATCH 4/5] style: add owner to TODO --- src/iceberg/test/parquet_row_group_filter_test.cc | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/iceberg/test/parquet_row_group_filter_test.cc b/src/iceberg/test/parquet_row_group_filter_test.cc index 002c25365d..96f4f644c7 100644 --- a/src/iceberg/test/parquet_row_group_filter_test.cc +++ b/src/iceberg/test/parquet_row_group_filter_test.cc @@ -566,7 +566,8 @@ TEST_F(ParquetRowGroupFilterTest, UuidBoundsUseUnsignedByteOrder) { } TEST_F(ParquetRowGroupFilterTest, FloatingPointStatistics) { - // TODO: remove this once row group pruning is applied to float/double columns. + // TODO(YinZheng-Sun): remove this once row group pruning is applied to + // float/double columns. SetKeyType(float64()); ASSERT_THAT(Write(), IsOk()); Check(Options(Expressions::Equal("key", Literal::Double(100))), {0, 1, 2, 3, 4, 5}); From 80199d2ce09911d12b79cb947c5cdf21444183d1 Mon Sep 17 00:00:00 2001 From: yinzhengsun Date: Tue, 20 Oct 2026 00:16:45 +0800 Subject: [PATCH 5/5] chore: retrigger CI