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..959d56322 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" @@ -42,6 +41,7 @@ namespace { ReaderOptions MakeReaderOptions(const DataFile& data_file, std::shared_ptr io, std::shared_ptr projection, std::shared_ptr filter, + bool filter_case_sensitive, std::shared_ptr name_mapping, ReaderProperties properties, std::optional first_row_id, @@ -52,6 +52,7 @@ ReaderOptions MakeReaderOptions(const DataFile& data_file, std::shared_ptrfile_size_in_bytes); + auto filter = task.residual_filter(); + 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); + auto options = + MakeReaderOptions(*data_file, io_, projected_schema_, filter, + 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)); return MakeArrowArrayStream(std::move(reader)); @@ -192,7 +196,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, 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)); @@ -207,16 +211,16 @@ class FileScanTaskReader::Impl { Impl(Options options, DeleteFilter::FieldLookup field_lookup, std::shared_ptr delete_counter) : io_(std::move(options.io)), - schemas_(std::move(options.schemas)), projected_schema_(std::move(options.projected_schema)), + filter_case_sensitive_(options.filter_case_sensitive), name_mapping_(std::move(options.name_mapping)), properties_(ReaderProperties::FromMap(options.properties)), field_lookup_(std::move(field_lookup)), delete_counter_(std::move(delete_counter)) {} std::shared_ptr io_; - std::vector> schemas_; std::shared_ptr projected_schema_; + bool filter_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 a71ef5f84..3b0232418 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 filter binding. Pass TableScan::is_case_sensitive(). + bool filter_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/binder.cc b/src/iceberg/expression/binder.cc index 9474151d7..ecc1f1f2a 100644 --- a/src/iceberg/expression/binder.cc +++ b/src/iceberg/expression/binder.cc @@ -24,6 +24,21 @@ namespace iceberg { +namespace { + +Result> CombineResults(const std::optional& left_result, + const std::optional& right_result) { + if (!left_result.has_value()) { + return right_result; + } + if (right_result.has_value() && left_result != right_result) { + return InvalidExpression("Found partially bound expression"); + } + return left_result; +} + +} // namespace + Binder::Binder(const Schema& schema, bool case_sensitive) : schema_(schema), case_sensitive_(case_sensitive) {} @@ -82,40 +97,44 @@ Result> Binder::Aggregate( Result IsBoundVisitor::IsBound(const std::shared_ptr& expr) { ICEBERG_PRECHECK(expr != nullptr, "Expression cannot be null"); IsBoundVisitor visitor; - return Visit(expr, visitor); + ICEBERG_ASSIGN_OR_RAISE(auto is_bound, Visit>(expr, visitor)); + return is_bound.value_or(false); } -Result IsBoundVisitor::AlwaysTrue() { - return InvalidExpression("IsBoundVisitor does not support AlwaysTrue expression"); -} +Result> IsBoundVisitor::AlwaysTrue() { return std::nullopt; } -Result IsBoundVisitor::AlwaysFalse() { - return InvalidExpression("IsBoundVisitor does not support AlwaysFalse expression"); -} +Result> IsBoundVisitor::AlwaysFalse() { return std::nullopt; } -Result IsBoundVisitor::Not(bool child_result) { return child_result; } +Result> IsBoundVisitor::Not(const std::optional& child_result) { + return child_result; +} -Result IsBoundVisitor::And(bool left_result, bool right_result) { - return left_result && right_result; +Result> IsBoundVisitor::And(const std::optional& left_result, + const std::optional& right_result) { + return CombineResults(left_result, right_result); } -Result IsBoundVisitor::Or(bool left_result, bool right_result) { - return left_result && right_result; +Result> IsBoundVisitor::Or(const std::optional& left_result, + const std::optional& right_result) { + return CombineResults(left_result, right_result); } -Result IsBoundVisitor::Predicate(const std::shared_ptr& pred) { +Result> IsBoundVisitor::Predicate( + const std::shared_ptr& pred) { return true; } -Result IsBoundVisitor::Predicate(const std::shared_ptr& pred) { +Result> IsBoundVisitor::Predicate( + const std::shared_ptr& pred) { return false; } -Result IsBoundVisitor::Aggregate(const std::shared_ptr& aggregate) { +Result> IsBoundVisitor::Aggregate( + const std::shared_ptr& aggregate) { return true; } -Result IsBoundVisitor::Aggregate( +Result> IsBoundVisitor::Aggregate( const std::shared_ptr& aggregate) { return false; } diff --git a/src/iceberg/expression/binder.h b/src/iceberg/expression/binder.h index 276ab0760..9aac5b752 100644 --- a/src/iceberg/expression/binder.h +++ b/src/iceberg/expression/binder.h @@ -23,6 +23,7 @@ /// Bind an expression to a schema. #include +#include #include #include "iceberg/expression/expression_visitor.h" @@ -61,19 +62,25 @@ class ICEBERG_EXPORT Binder : public ExpressionVisitor { +class ICEBERG_EXPORT IsBoundVisitor : public ExpressionVisitor> { public: static Result IsBound(const std::shared_ptr& expr); - Result AlwaysTrue() override; - Result AlwaysFalse() override; - Result Not(bool child_result) override; - Result And(bool left_result, bool right_result) override; - Result Or(bool left_result, bool right_result) override; - Result Predicate(const std::shared_ptr& pred) override; - Result Predicate(const std::shared_ptr& pred) override; - Result Aggregate(const std::shared_ptr& aggregate) override; - Result Aggregate(const std::shared_ptr& aggregate) override; + Result> AlwaysTrue() override; + Result> AlwaysFalse() override; + Result> Not(const std::optional& child_result) override; + Result> And(const std::optional& left_result, + const std::optional& right_result) override; + Result> Or(const std::optional& left_result, + const std::optional& right_result) override; + Result> Predicate( + const std::shared_ptr& pred) override; + Result> Predicate( + const std::shared_ptr& pred) override; + Result> Aggregate( + const std::shared_ptr& aggregate) override; + Result> Aggregate( + const std::shared_ptr& aggregate) override; }; using FieldIdsSetRef = std::reference_wrapper>; diff --git a/src/iceberg/expression/predicate.cc b/src/iceberg/expression/predicate.cc index 44c4a3a70..8da9c2308 100644 --- a/src/iceberg/expression/predicate.cc +++ b/src/iceberg/expression/predicate.cc @@ -25,6 +25,8 @@ #include "iceberg/expression/expressions.h" #include "iceberg/expression/literal.h" #include "iceberg/result.h" +#include "iceberg/row/struct_like.h" +#include "iceberg/schema.h" #include "iceberg/transform.h" #include "iceberg/type.h" #include "iceberg/util/checked_cast.h" @@ -230,7 +232,7 @@ Result> UnboundPredicateImpl::Bind( ICEBERG_ASSIGN_OR_RAISE(auto bound_term, BASE::term()->Bind(schema, case_sensitive)); if (values_.empty()) { - return BindUnaryOperation(std::move(bound_term)); + return BindUnaryOperation(schema, std::move(bound_term)); } if (BASE::op() == Expression::Operation::kIn || @@ -247,6 +249,19 @@ bool IsFloatingType(TypeId type) { return type == TypeId::kFloat || type == TypeId::kDouble; } +bool AllAncestorsRequired(const StructType& struct_type, const BoundReference& ref) { + const auto& path = ref.accessor().position_path(); + const StructType* parent = &struct_type; + for (size_t i = 0; i + 1 < path.size(); ++i) { + const auto& field = parent->fields()[path[i]]; + if (field.optional()) { + return false; + } + parent = &internal::checked_cast(*field.type()); + } + return true; +} + bool StartsWith(const Literal& lhs, const Literal& rhs) { const auto& lhs_value = lhs.value(); const auto& rhs_value = rhs.value(); @@ -261,17 +276,19 @@ bool StartsWith(const Literal& lhs, const Literal& rhs) { template Result> UnboundPredicateImpl::BindUnaryOperation( - std::shared_ptr bound_term) const { + const StructType& struct_type, std::shared_ptr bound_term) const { switch (BASE::op()) { case Expression::Operation::kIsNull: - if (!bound_term->MayProduceNull()) { + if (!bound_term->MayProduceNull() && + AllAncestorsRequired(struct_type, *bound_term->reference())) { return Expressions::AlwaysFalse(); } // TODO(gangwu): deal with UnknownType return BoundUnaryPredicate::Make(Expression::Operation::kIsNull, std::move(bound_term)); case Expression::Operation::kNotNull: - if (!bound_term->MayProduceNull()) { + if (!bound_term->MayProduceNull() && + AllAncestorsRequired(struct_type, *bound_term->reference())) { return Expressions::AlwaysTrue(); } return BoundUnaryPredicate::Make(Expression::Operation::kNotNull, diff --git a/src/iceberg/expression/predicate.h b/src/iceberg/expression/predicate.h index af883fe19..4fc112403 100644 --- a/src/iceberg/expression/predicate.h +++ b/src/iceberg/expression/predicate.h @@ -150,7 +150,7 @@ class ICEBERG_EXPORT UnboundPredicateImpl : public UnboundPredicate, std::vector values); Result> BindUnaryOperation( - std::shared_ptr bound_term) const; + const StructType& struct_type, std::shared_ptr bound_term) const; Result> BindLiteralOperation( std::shared_ptr bound_term) const; Result> BindInOperation( diff --git a/src/iceberg/file_reader.h b/src/iceberg/file_reader.h index e08b2df49..3acf2f16f 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). @@ -103,10 +106,13 @@ struct ICEBERG_EXPORT ReaderOptions { /// \brief FileIO instance to open the file. std::shared_ptr io; /// \brief The projection schema to read from the file. This field is required. + /// Include columns referenced by unbound filters; readers bind them to this schema. std::shared_ptr projection; /// \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 binding filter references. + bool filter_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 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..24243edcc --- /dev/null +++ b/src/iceberg/parquet/parquet_metrics_row_group_filter.cc @@ -0,0 +1,327 @@ +/* + * 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& column_indices) + : manifest_(manifest), row_group_(row_group), column_indices_(column_indices) {} + + Result AlwaysTrue() override { return kRowsMightMatch; } + + Result AlwaysFalse() override { return kRowsCannotMatch; } + + Result Not(bool) override { 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); + if (!lower) { + return kRowsMightMatch; + } + if (std::ranges::all_of(values, [&](const auto& value) { return value < *lower; })) { + return kRowsCannotMatch; + } + const auto upper = MaxValue(ref); + if (!upper) { + return kRowsMightMatch; + } + // Like Java, a single candidate must satisfy both bounds. + for (const auto& value : values) { + if (!(value < *lower || value > *upper)) { + return kRowsMightMatch; + } + } + return 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 || lower->type()->type_id() != TypeId::kString) { + 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 || 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 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()); + 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: + 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(); + } + + 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()) || + MetadataColumns::IsRowLineageColumn(ref->field_id())) { + return nullptr; + } + 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 nullptr; + } + 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 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& column_indices_; +}; + +} // 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->column_indices_.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, column_indices_); + 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 000000000..96703a838 --- /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 column_indices_; +}; + +} // namespace iceberg::parquet diff --git a/src/iceberg/parquet/parquet_reader.cc b/src/iceberg/parquet/parquet_reader.cc index 1f8146107..2848970e6 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 @@ -38,7 +38,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/result.h" @@ -65,13 +67,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 +258,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 next_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 +287,7 @@ class ParquetReader::Impl { split_ = options.split; read_schema_ = options.projection; + stats_filter_.reset(); // Prepare reader properties ::parquet::ReaderProperties reader_properties(pool_); @@ -307,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.filter_case_sensitive)); + } + ICEBERG_ASSIGN_OR_RAISE(stats_filter_, ParquetMetricsRowGroupFilter::Make( + filter, *reader_->manifest().descr)); + } + metadata_context_ = {.file_path = options.path, .next_file_pos = 0, .first_row_id = options.first_row_id, @@ -322,18 +337,22 @@ class ParquetReader::Impl { } ICEBERG_ARROW_ASSIGN_OR_RETURN(auto batch, context_->record_batch_reader_->Next()); - if (!batch) { - return std::nullopt; + while (!batch) { + if (context_->next_row_group_ >= context_->row_groups_.size()) { + return std::nullopt; + } + ICEBERG_ARROW_RETURN_NOT_OK(context_->record_batch_reader_->Close()); + ICEBERG_RETURN_UNEXPECTED(OpenNextRowGroupRange()); + 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; } @@ -387,39 +406,58 @@ class ParquetReader::Impl { } private: + Status OpenNextRowGroupRange() { + const size_t begin = context_->next_row_group_; + size_t end = begin + 1; + std::vector row_groups{context_->row_groups_[begin].index}; + // Share a reader across adjacent row groups so Arrow can coalesce pre-buffered + // reads. Stop at gaps to keep physical row positions contiguous within batches. + while (end < context_->row_groups_.size() && + context_->row_groups_[end].index == context_->row_groups_[end - 1].index + 1) { + row_groups.push_back(context_->row_groups_[end].index); + ++end; + } + ICEBERG_ARROW_ASSIGN_OR_RETURN( + context_->record_batch_reader_, + reader_->GetRecordBatchReader(row_groups, SelectedColumnIndices(projection_))); + context_->next_row_group_ = end; + metadata_context_.next_file_pos = context_->row_groups_[begin].first_row; + return {}; + } + Status InitReadContext() { context_ = std::make_unique(); + auto metadata = reader_->parquet_reader()->metadata(); - // 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(); + 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 = std::cmp_greater_equal(row_group_offset, split_->offset) && + std::cmp_less(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(reader_->manifest(), *row_group)); + if (!should_read) { + continue; + } + } + if (row_group->num_rows() == 0) { + continue; + } + context_->row_groups_.push_back({.index = i, .first_row = 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 + 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)); + ICEBERG_RETURN_UNEXPECTED(OpenNextRowGroupRange()); } // Build the output Arrow schema from the projected Iceberg schema. This schema is the @@ -454,6 +492,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/expression_visitor_test.cc b/src/iceberg/test/expression_visitor_test.cc index a3b2c4cad..e58902dca 100644 --- a/src/iceberg/test/expression_visitor_test.cc +++ b/src/iceberg/test/expression_visitor_test.cc @@ -305,14 +305,15 @@ TEST_F(BinderTest, ErrorNestedUnboundField) { class IsBoundVisitorTest : public ExpressionVisitorTest {}; TEST_F(IsBoundVisitorTest, Constants) { - // True and False should error out auto true_expr = Expressions::AlwaysTrue(); auto result_true = IsBoundVisitor::IsBound(true_expr); - EXPECT_THAT(result_true, IsError(ErrorKind::kInvalidExpression)); + ASSERT_THAT(result_true, IsOk()); + EXPECT_FALSE(*result_true); auto false_expr = Expressions::AlwaysFalse(); auto result_false = IsBoundVisitor::IsBound(false_expr); - EXPECT_THAT(result_false, IsError(ErrorKind::kInvalidExpression)); + ASSERT_THAT(result_false, IsOk()); + EXPECT_FALSE(*result_false); } TEST_F(IsBoundVisitorTest, UnboundPredicate) { @@ -322,6 +323,39 @@ TEST_F(IsBoundVisitorTest, UnboundPredicate) { EXPECT_FALSE(is_bound); } +TEST_F(IsBoundVisitorTest, AndWithConstants) { + auto unbound = Expressions::Equal("name", Literal::String("Alice")); + ICEBERG_UNWRAP_OR_FAIL(auto bound, Bind(unbound)); + + // Use Make to keep constants in the expression tree instead of folding them. + ICEBERG_UNWRAP_OR_FAIL(std::shared_ptr unbound_and, + And::Make(Expressions::AlwaysTrue(), unbound)); + ICEBERG_UNWRAP_OR_FAIL(auto unbound_result, IsBoundVisitor::IsBound(unbound_and)); + EXPECT_FALSE(unbound_result); + EXPECT_THAT(Bind(unbound_and), IsOk()); + + ICEBERG_UNWRAP_OR_FAIL(std::shared_ptr bound_and, + And::Make(bound, Expressions::AlwaysTrue())); + ICEBERG_UNWRAP_OR_FAIL(auto bound_result, IsBoundVisitor::IsBound(bound_and)); + EXPECT_TRUE(bound_result); +} + +TEST_F(IsBoundVisitorTest, OrWithConstants) { + auto unbound = Expressions::Equal("name", Literal::String("Alice")); + ICEBERG_UNWRAP_OR_FAIL(auto bound, Bind(unbound)); + + ICEBERG_UNWRAP_OR_FAIL(std::shared_ptr unbound_or, + Or::Make(Expressions::AlwaysFalse(), unbound)); + ICEBERG_UNWRAP_OR_FAIL(auto unbound_result, IsBoundVisitor::IsBound(unbound_or)); + EXPECT_FALSE(unbound_result); + EXPECT_THAT(Bind(unbound_or), IsOk()); + + ICEBERG_UNWRAP_OR_FAIL(std::shared_ptr bound_or, + Or::Make(bound, Expressions::AlwaysFalse())); + ICEBERG_UNWRAP_OR_FAIL(auto bound_result, IsBoundVisitor::IsBound(bound_or)); + EXPECT_TRUE(bound_result); +} + TEST_F(IsBoundVisitorTest, BoundPredicate) { // Bound predicates should return true auto unbound_pred = Expressions::Equal("name", Literal::String("Alice")); @@ -342,14 +376,14 @@ TEST_F(IsBoundVisitorTest, AndWithBoundChildren) { } TEST_F(IsBoundVisitorTest, AndWithUnboundChild) { - // AND with any unbound child should return false + // Mixing bound and unbound predicates is invalid. auto bound_pred = Expressions::Equal("name", Literal::String("Alice")); ICEBERG_UNWRAP_OR_FAIL(auto pred1, Bind(bound_pred)); auto pred2 = Expressions::Equal("age", Literal::Int(25)); // unbound auto mixed_and = Expressions::And(pred1, pred2); - ICEBERG_UNWRAP_OR_FAIL(auto is_bound, IsBoundVisitor::IsBound(mixed_and)); - EXPECT_FALSE(is_bound); + EXPECT_THAT(IsBoundVisitor::IsBound(mixed_and), + HasErrorMessage("Found partially bound expression")); } TEST_F(IsBoundVisitorTest, OrWithBoundChildren) { @@ -364,14 +398,14 @@ TEST_F(IsBoundVisitorTest, OrWithBoundChildren) { } TEST_F(IsBoundVisitorTest, OrWithUnboundChild) { - // OR with any unbound child should return false + // Mixing bound and unbound predicates is invalid. auto pred1 = Expressions::IsNull("name"); // unbound auto bound_pred2 = Expressions::Equal("age", Literal::Int(25)); ICEBERG_UNWRAP_OR_FAIL(auto pred2, Bind(bound_pred2)); auto mixed_or = Expressions::Or(pred1, pred2); - ICEBERG_UNWRAP_OR_FAIL(auto is_bound, IsBoundVisitor::IsBound(mixed_or)); - EXPECT_FALSE(is_bound); + EXPECT_THAT(IsBoundVisitor::IsBound(mixed_or), + HasErrorMessage("Found partially bound expression")); } TEST_F(IsBoundVisitorTest, NotWithBoundChild) { @@ -405,15 +439,15 @@ TEST_F(IsBoundVisitorTest, ComplexExpression) { ICEBERG_UNWRAP_OR_FAIL(auto is_bound, IsBoundVisitor::IsBound(bound_complex)); EXPECT_TRUE(is_bound); - // Complex expression: one unbound should return false + // Mixed binding state is rejected even in a nested expression. auto unbound_pred = Expressions::Equal("name", Literal::String("Alice")); ICEBERG_UNWRAP_OR_FAIL(auto bound_pred2, Bind(pred2)); ICEBERG_UNWRAP_OR_FAIL(auto bound_pred3, Bind(pred3)); auto mixed_and = Expressions::And(unbound_pred, bound_pred2); auto mixed_complex = Expressions::Or(mixed_and, bound_pred3); - ICEBERG_UNWRAP_OR_FAIL(auto is_bound_mixed, IsBoundVisitor::IsBound(mixed_complex)); - EXPECT_FALSE(is_bound_mixed); + EXPECT_THAT(IsBoundVisitor::IsBound(mixed_complex), + HasErrorMessage("Found partially bound expression")); } class RewriteNotTest : public ExpressionVisitorTest {}; diff --git a/src/iceberg/test/file_scan_task_reader_test.cc b/src/iceberg/test/file_scan_task_reader_test.cc index cca13ce0d..8f09140d3 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,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.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","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, + .filter_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.filter_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 000000000..85ca80b3b --- /dev/null +++ b/src/iceberg/test/parquet_row_group_filter_test.cc @@ -0,0 +1,712 @@ +/* + * 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.filter_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.filter_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, 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}); +} + +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, 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/parquet_test.cc b/src/iceberg/test/parquet_test.cc index ec96167d1..684b5b625 100644 --- a/src/iceberg/test/parquet_test.cc +++ b/src/iceberg/test/parquet_test.cc @@ -848,8 +848,8 @@ 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]])", "", "", + std::vector> expected_json = { + {R"([[1, 0], [2, 1], [3, 2]])"}, {R"([[1, 0], [2, 1]])"}, {R"([[3, 2]])"}, {}, {}, }; ReaderProperties reader_properties; @@ -866,8 +866,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)); } diff --git a/src/iceberg/test/predicate_test.cc b/src/iceberg/test/predicate_test.cc index 6b990462a..fe9df4963 100644 --- a/src/iceberg/test/predicate_test.cc +++ b/src/iceberg/test/predicate_test.cc @@ -21,6 +21,7 @@ #include #include +#include #include "iceberg/expression/expressions.h" #include "iceberg/schema.h" @@ -365,6 +366,49 @@ TEST_F(PredicateTest, UnboundPredicateBindUnary) { EXPECT_EQ(bound_is_null->op(), Expression::Operation::kFalse); } +TEST_F(PredicateTest, BindNestedNullPredicates) { + for (auto [outer_optional, inner_optional] : + {std::pair{false, false}, std::pair{true, false}, std::pair{false, true}}) { + auto inner = std::make_shared( + std::vector{SchemaField::MakeRequired(4, "leaf", int32())}); + auto outer = std::make_shared( + std::vector{SchemaField(3, "middle", inner, inner_optional)}); + Schema schema( + std::vector{SchemaField::MakeOptional(1, "unrelated", int32()), + SchemaField(2, "key", outer, outer_optional)}); + SCOPED_TRACE(schema.ToString()); + bool nullable = outer_optional || inner_optional; + ICEBERG_UNWRAP_OR_FAIL(auto is_null, + Expressions::IsNull("key.middle.leaf")->Bind(schema, true)); + ICEBERG_UNWRAP_OR_FAIL(auto not_null, + Expressions::NotNull("key.middle.leaf")->Bind(schema, true)); + EXPECT_EQ(is_null->op(), + nullable ? Expression::Operation::kIsNull : Expression::Operation::kFalse); + EXPECT_EQ(not_null->op(), + nullable ? Expression::Operation::kNotNull : Expression::Operation::kTrue); + } +} + +TEST_F(PredicateTest, BindTransformedNestedNullPredicates) { + for (bool parent_optional : {false, true}) { + Schema schema(std::vector{SchemaField( + 1, "key", + std::make_shared( + std::vector{SchemaField::MakeRequired(2, "nested", int32())}), + parent_optional)}); + auto truncate = Expressions::Truncate("key.nested", 10); + ICEBERG_UNWRAP_OR_FAIL( + auto is_null, Expressions::IsNull(truncate)->Bind(schema, true)); + ICEBERG_UNWRAP_OR_FAIL( + auto not_null, + Expressions::NotNull(truncate)->Bind(schema, true)); + EXPECT_EQ(is_null->op(), parent_optional ? Expression::Operation::kIsNull + : Expression::Operation::kFalse); + EXPECT_EQ(not_null->op(), parent_optional ? Expression::Operation::kNotNull + : Expression::Operation::kTrue); + } +} + TEST_F(PredicateTest, UnboundPredicateBindLiteral) { auto equal_pred = Expressions::Equal("age", Literal::Int(25)); auto bound_result = equal_pred->Bind(*schema_, /*case_sensitive=*/true);