feat(parquet): add statistics-based row group filtering

src/iceberg/CMakeLists.txt

@@ -321,6 +321,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

src/iceberg/data/file_scan_task_reader.cc

@@ -23,7 +23,6 @@
 #include <memory>
 #include <optional>
 #include <utility>
-#include <vector>
 
 #include "iceberg/arrow_c_data_guard_internal.h"
 #include "iceberg/arrow_c_data_util_internal.h"
@@ -41,7 +40,7 @@ namespace {
 
 ReaderOptions MakeReaderOptions(const FileScanTask& task, std::shared_ptr<FileIO> io,
                                 std::shared_ptr<Schema> projection,
-                                std::shared_ptr<Expression> filter,
+                                std::shared_ptr<Expression> filter, bool case_sensitive,
                                 std::shared_ptr<NameMapping> name_mapping,
                                 ReaderProperties properties,
                                 std::optional<int64_t> first_row_id,
@@ -57,6 +56,7 @@ ReaderOptions MakeReaderOptions(const FileScanTask& task, std::shared_ptr<FileIO
       .io = std::move(io),
       .projection = std::move(projection),
       .filter = std::move(filter),
+      .case_sensitive = case_sensitive,
       .name_mapping = std::move(name_mapping),
       .first_row_id = first_row_id,
       .data_sequence_number = data_sequence_number,
@@ -188,9 +188,11 @@ class FileScanTaskReader::Impl {
       }
     }
 
+    auto filter = task.residual_filter();
+
     if (task.delete_files().empty()) {
       auto options = MakeReaderOptions(
-          task, io_, projected_schema_, task.residual_filter(), name_mapping_,
+          task, 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));
@@ -215,7 +217,7 @@ class FileScanTaskReader::Impl {
                             ProjectionContext::Make(*required_schema, *projected_schema_,
                                                     project_batch_function));
 
-    auto options = MakeReaderOptions(task, io_, required_schema, task.residual_filter(),
+    auto options = MakeReaderOptions(task, 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,
@@ -233,6 +235,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)),
@@ -241,6 +244,7 @@ class FileScanTaskReader::Impl {
   std::shared_ptr<FileIO> io_;
   std::vector<std::shared_ptr<Schema>> schemas_;
   std::shared_ptr<Schema> projected_schema_;
+  bool case_sensitive_;
   std::shared_ptr<NameMapping> name_mapping_;
   ReaderProperties properties_;
   DeleteFilter::FieldLookup field_lookup_;

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<Schema> 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<NameMapping> name_mapping;
     /// Format-specific or implementation-specific options for data readers.

src/iceberg/expression/literal.cc

@@ -24,6 +24,7 @@
 #include <cmath>
 #include <concepts>
 #include <cstdint>
+#include <cstring>
 #include <format>
 #include <string>
 #include <string_view>
@@ -710,12 +711,9 @@ std::partial_ordering Literal::operator<=>(const Literal& other) const {
     }
 
     case TypeId::kUuid: {
-      auto& this_val = std::get<Uuid>(value_);
-      auto& other_val = std::get<Uuid>(other.value_);
-      if (this_val == other_val) {
-        return std::partial_ordering::equivalent;
-      }
-      return std::partial_ordering::unordered;
+      auto this_bytes = std::get<Uuid>(value_).bytes();
+      auto other_bytes = std::get<Uuid>(other.value_).bytes();
+      return std::memcmp(this_bytes.data(), other_bytes.data(), Uuid::kLength) <=> 0;
     }
 
     case TypeId::kBinary:

src/iceberg/expression/predicate.cc

@@ -35,6 +35,15 @@
 
 namespace iceberg {
 
+namespace {
+
+bool IsRangeComparison(Expression::Operation op) {
+  return op == Expression::Operation::kLt || op == Expression::Operation::kLtEq ||
+         op == Expression::Operation::kGt || op == Expression::Operation::kGtEq;
+}
+
+}  // namespace
+
 // Predicate template implementations
 template <TermType T>
 Predicate<T>::Predicate(Expression::Operation op, std::shared_ptr<T> term)
@@ -223,7 +232,16 @@ bool UnboundPredicateImpl<B>::Equals(const Expression& other) const {
 template <typename B>
 Result<std::shared_ptr<Expression>> UnboundPredicateImpl<B>::Negate() const {
   ICEBERG_ASSIGN_OR_RAISE(auto negated_op, ::iceberg::Negate(BASE::op()));
-  return UnboundPredicateImpl::Make(negated_op, BASE::term(), values_);
+  ICEBERG_ASSIGN_OR_RAISE(auto inverse,
+                          UnboundPredicateImpl::Make(negated_op, BASE::term(), values_));
+  if (!IsRangeComparison(BASE::op())) {
+    return inverse;
+  }
+  // Range comparisons reject null, but their boolean negation accepts it.
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto is_null,
+      UnboundPredicateImpl::Make(Expression::Operation::kIsNull, BASE::term()));
+  return Or::MakeFolded(std::move(is_null), std::move(inverse));
 }
 
 template <typename B>
@@ -240,7 +258,7 @@ Result<std::shared_ptr<Expression>> UnboundPredicateImpl<B>::Bind(
     return BindInOperation(std::move(bound_term));
   }
 
-  return BindLiteralOperation(std::move(bound_term));
+  return BindLiteralOperation(schema, std::move(bound_term));
 }
 
 namespace {
@@ -280,6 +298,23 @@ Result<bool> AllAncestorsAreRequired(const Schema& schema, int32_t field_id) {
   return true;
 }
 
+Result<std::shared_ptr<Expression>> BindNotNull(const Schema& schema,
+                                                std::shared_ptr<BoundTerm> term) {
+  // A required field can still be null if a parent is optional.
+  if (!term->MayProduceNull()) {
+    ICEBERG_ASSIGN_OR_RAISE(
+        bool all_ancestors_required,
+        AllAncestorsAreRequired(schema, term->reference()->field_id()));
+    if (all_ancestors_required) {
+      return Expressions::AlwaysTrue();
+    }
+  }
+  if (term->type()->type_id() == TypeId::kUnknown) {
+    return Expressions::AlwaysFalse();
+  }
+  return BoundUnaryPredicate::Make(Expression::Operation::kNotNull, std::move(term));
+}
+
 }  // namespace
 
 template <typename B>
@@ -301,21 +336,8 @@ Result<std::shared_ptr<Expression>> UnboundPredicateImpl<B>::BindUnaryOperation(
       return BoundUnaryPredicate::Make(Expression::Operation::kIsNull,
                                        std::move(bound_term));
     }
-    case Expression::Operation::kNotNull: {
-      if (!bound_term->MayProduceNull()) {
-        ICEBERG_ASSIGN_OR_RAISE(
-            bool all_ancestors_required,
-            AllAncestorsAreRequired(schema, bound_term->reference()->field_id()));
-        if (all_ancestors_required) {
-          return Expressions::AlwaysTrue();
-        }
-      }
-      if (bound_term->type()->type_id() == TypeId::kUnknown) {
-        return Expressions::AlwaysFalse();
-      }
-      return BoundUnaryPredicate::Make(Expression::Operation::kNotNull,
-                                       std::move(bound_term));
-    }
+    case Expression::Operation::kNotNull:
+      return BindNotNull(schema, std::move(bound_term));
     case Expression::Operation::kIsNan:
     case Expression::Operation::kNotNan:
       if (!IsFloatingType(bound_term->type()->type_id())) {
@@ -331,7 +353,7 @@ Result<std::shared_ptr<Expression>> UnboundPredicateImpl<B>::BindUnaryOperation(
 
 template <typename B>
 Result<std::shared_ptr<Expression>> UnboundPredicateImpl<B>::BindLiteralOperation(
-    std::shared_ptr<B> bound_term) const {
+    const Schema& schema, std::shared_ptr<B> bound_term) const {
   if (BASE::op() == Expression::Operation::kStartsWith ||
       BASE::op() == Expression::Operation::kNotStartsWith) {
     if (bound_term->type()->type_id() != TypeId::kString) {
@@ -357,6 +379,8 @@ Result<std::shared_ptr<Expression>> UnboundPredicateImpl<B>::BindLiteralOperatio
     switch (BASE::op()) {
       case Expression::Operation::kLt:
       case Expression::Operation::kLtEq:
+        // All non-null values match, but null still fails the comparison.
+        return BindNotNull(schema, std::move(bound_term));
       case Expression::Operation::kNotEq:
         return Expressions::AlwaysTrue();
       case Expression::Operation::kGt:
@@ -370,6 +394,8 @@ Result<std::shared_ptr<Expression>> UnboundPredicateImpl<B>::BindLiteralOperatio
     switch (BASE::op()) {
       case Expression::Operation::kGt:
       case Expression::Operation::kGtEq:
+        // All non-null values match, but null still fails the comparison.
+        return BindNotNull(schema, std::move(bound_term));
       case Expression::Operation::kNotEq:
         return Expressions::AlwaysTrue();
       case Expression::Operation::kLt:
@@ -562,7 +588,15 @@ Result<bool> BoundLiteralPredicate::Test(const Literal& value) const {
 
 Result<std::shared_ptr<Expression>> BoundLiteralPredicate::Negate() const {
   ICEBERG_ASSIGN_OR_RAISE(auto negated_op, ::iceberg::Negate(op()));
-  return BoundLiteralPredicate::Make(negated_op, term_, literal_);
+  ICEBERG_ASSIGN_OR_RAISE(auto inverse,
+                          BoundLiteralPredicate::Make(negated_op, term_, literal_));
+  if (!IsRangeComparison(op())) {
+    return inverse;
+  }
+  // A required leaf can still produce null through an optional ancestor.
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto is_null, BoundUnaryPredicate::Make(Expression::Operation::kIsNull, term_));
+  return Or::MakeFolded(std::move(is_null), std::move(inverse));
 }
 
 bool BoundLiteralPredicate::Equals(const Expression& other) const {

src/iceberg/expression/predicate.h

@@ -152,7 +152,7 @@ class ICEBERG_EXPORT UnboundPredicateImpl : public UnboundPredicate,
   Result<std::shared_ptr<Expression>> BindUnaryOperation(
       const Schema& schema, std::shared_ptr<B> bound_term) const;
   Result<std::shared_ptr<Expression>> BindLiteralOperation(
-      std::shared_ptr<B> bound_term) const;
+      const Schema& schema, std::shared_ptr<B> bound_term) const;
   Result<std::shared_ptr<Expression>> BindInOperation(
       std::shared_ptr<B> bound_term) const;
 

src/iceberg/file_reader.h

@@ -80,6 +80,9 @@ class ICEBERG_EXPORT ReaderProperties : public ConfigBase<ReaderProperties> {
   /// Only the Parquet reader honors this option; other readers ignore it.
   /// Default: false (use 32-bit offset list).
   inline static Entry<bool> kArrowUseLargeList{"read.arrow.use-large-list", false};
+  /// \brief Use footer statistics to prune Parquet row groups.
+  inline static Entry<bool> 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<class Expression> 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<class NameMapping> name_mapping;

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<Literal> 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<int>(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<std::optional<FieldMetrics>> 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<Literal> 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<int>(column.physical_type()));
+  }
+  std::unreachable();
+}
+
 Result<Metrics> ParquetMetrics::GetMetrics(
     const Schema& schema, const ::parquet::SchemaDescriptor& parquet_schema,
     const MetricsConfig& metrics_config, const ::parquet::FileMetaData& metadata,

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<Literal> 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,