diff --git a/src/iceberg/CMakeLists.txt b/src/iceberg/CMakeLists.txt
index 5af4fdae5..c70ba05c6 100644
--- a/src/iceberg/CMakeLists.txt
+++ b/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
diff --git a/src/iceberg/data/file_scan_task_reader.cc b/src/iceberg/data/file_scan_task_reader.cc
index 8c55e31b2..bf23c5c36 100644
--- a/src/iceberg/data/file_scan_task_reader.cc
+++ b/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_;
diff --git a/src/iceberg/data/file_scan_task_reader.h b/src/iceberg/data/file_scan_task_reader.h
index 33a4b82ff..591d49c92 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<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.
diff --git a/src/iceberg/expression/literal.cc b/src/iceberg/expression/literal.cc
index 88f4b5740..c8dd74a44 100644
--- a/src/iceberg/expression/literal.cc
+++ b/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:
diff --git a/src/iceberg/expression/predicate.cc b/src/iceberg/expression/predicate.cc
index 383edbc27..f082a3c9a 100644
--- a/src/iceberg/expression/predicate.cc
+++ b/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 {
diff --git a/src/iceberg/expression/predicate.h b/src/iceberg/expression/predicate.h
index 152c6576e..c7afbca6c 100644
--- a/src/iceberg/expression/predicate.h
+++ b/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;
 
diff --git a/src/iceberg/file_reader.h b/src/iceberg/file_reader.h
index e08b2df49..13e8b3ad2 100644
--- a/src/iceberg/file_reader.h
+++ b/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;
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<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,
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<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,
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..eb543e872
--- /dev/null
+++ b/src/iceberg/parquet/parquet_metrics_row_group_filter.cc
@@ -0,0 +1,326 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include <algorithm>
+#include <compare>
+#include <functional>
+#include <optional>
+#include <unordered_map>
+
+#include <parquet/statistics.h>
+
+#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<bool> {
+ public:
+  MetricsVisitor(const ::parquet::arrow::SchemaManifest& manifest,
+                 const ::parquet::RowGroupMetaData& row_group,
+                 const std::unordered_map<int32_t, int>& field_id_to_column_idx)
+      : manifest_(manifest),
+        row_group_(row_group),
+        field_id_to_column_idx_(field_id_to_column_idx) {}
+
+  Result<bool> AlwaysTrue() override { return kRowsMightMatch; }
+
+  Result<bool> AlwaysFalse() override { return kRowsCannotMatch; }
+
+  Result<bool> Not(bool) override {
+    // Make() applies RewriteNot, so NOT should not reach this visitor. If it does,
+    // keep the group: "might match" is not "all rows match", so negating it cannot
+    // prove that no rows match.
+    return kRowsMightMatch;
+  }
+
+  Result<bool> And(bool left, bool right) override { return left && right; }
+
+  Result<bool> Or(bool left, bool right) override { return left || right; }
+
+  Result<bool> IsNull(const std::shared_ptr<Bound>& expr) override {
+    return MayContainNull(std::dynamic_pointer_cast<BoundReference>(expr));
+  }
+
+  Result<bool> NotNull(const std::shared_ptr<Bound>& expr) override {
+    if (ContainsNullsOnly(std::dynamic_pointer_cast<BoundReference>(expr))) {
+      return kRowsCannotMatch;
+    }
+    return kRowsMightMatch;
+  }
+
+  Result<bool> IsNaN(const std::shared_ptr<Bound>& expr) override {
+    if (ContainsNullsOnly(std::dynamic_pointer_cast<BoundReference>(expr))) {
+      return kRowsCannotMatch;
+    }
+    return kRowsMightMatch;
+  }
+
+  Result<bool> NotNaN(const std::shared_ptr<Bound>&) override { return kRowsMightMatch; }
+
+  Result<bool> Lt(const std::shared_ptr<Bound>& expr, const Literal& value) override {
+    return VisitInequality(std::dynamic_pointer_cast<BoundReference>(expr), value,
+                           std::less<Literal>{}, /*use_lower_bound=*/true);
+  }
+
+  Result<bool> LtEq(const std::shared_ptr<Bound>& expr, const Literal& value) override {
+    return VisitInequality(std::dynamic_pointer_cast<BoundReference>(expr), value,
+                           std::less_equal<Literal>{}, /*use_lower_bound=*/true);
+  }
+
+  Result<bool> Gt(const std::shared_ptr<Bound>& expr, const Literal& value) override {
+    return VisitInequality(std::dynamic_pointer_cast<BoundReference>(expr), value,
+                           std::greater<Literal>{}, /*use_lower_bound=*/false);
+  }
+
+  Result<bool> GtEq(const std::shared_ptr<Bound>& expr, const Literal& value) override {
+    return VisitInequality(std::dynamic_pointer_cast<BoundReference>(expr), value,
+                           std::greater_equal<Literal>{}, /*use_lower_bound=*/false);
+  }
+
+  Result<bool> Eq(const std::shared_ptr<Bound>& expr, const Literal& value) override {
+    const auto ref = std::dynamic_pointer_cast<BoundReference>(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<bool> NotEq(const std::shared_ptr<Bound>&, const Literal&) override {
+    // Like Java, keep negative membership predicates inclusive.
+    return kRowsMightMatch;
+  }
+
+  Result<bool> In(const std::shared_ptr<Bound>& expr,
+                  const BoundSetPredicate::LiteralSet& values) override {
+    const auto ref = std::dynamic_pointer_cast<BoundReference>(expr);
+    if (ContainsNullsOnly(ref)) {
+      return kRowsCannotMatch;
+    }
+    if (values.size() > kInPredicateLimit) {
+      return kRowsMightMatch;
+    }
+    const auto lower = MinValue(ref);
+    const auto upper = MaxValue(ref);
+    if (!lower || !upper) {
+      return kRowsMightMatch;
+    }
+    // Equivalent to Java's two-step lower/upper filtering: the group can be
+    // skipped only if no value falls within [lower, upper].
+    const bool any_in_range = std::ranges::any_of(
+        values, [&](const auto& value) { return !(value < *lower || value > *upper); });
+    return any_in_range ? kRowsMightMatch : kRowsCannotMatch;
+  }
+
+  Result<bool> NotIn(const std::shared_ptr<Bound>&,
+                     const BoundSetPredicate::LiteralSet&) override {
+    return kRowsMightMatch;
+  }
+
+  Result<bool> StartsWith(const std::shared_ptr<Bound>& expr,
+                          const Literal& value) override {
+    const auto ref = std::dynamic_pointer_cast<BoundReference>(expr);
+    if (ContainsNullsOnly(ref)) {
+      return kRowsCannotMatch;
+    }
+    const auto lower = MinValue(ref);
+    if (!lower) {
+      return kRowsMightMatch;
+    }
+    const auto& prefix = std::get<std::string>(value.value());
+    if (std::get<std::string>(lower->value()).compare(0, prefix.size(), prefix) > 0) {
+      return kRowsCannotMatch;
+    }
+    const auto upper = MaxValue(ref);
+    if (!upper) {
+      return kRowsMightMatch;
+    }
+    return std::get<std::string>(upper->value()).compare(0, prefix.size(), prefix) >= 0;
+  }
+
+  Result<bool> NotStartsWith(const std::shared_ptr<Bound>& expr,
+                             const Literal& value) override {
+    const auto ref = std::dynamic_pointer_cast<BoundReference>(expr);
+    if (MayContainNull(ref)) {
+      return kRowsMightMatch;
+    }
+    const auto lower = MinValue(ref);
+    if (!lower) {
+      return kRowsMightMatch;
+    }
+    const auto& prefix = std::get<std::string>(value.value());
+    if (!std::get<std::string>(lower->value()).starts_with(prefix)) {
+      return kRowsMightMatch;
+    }
+    const auto upper = MaxValue(ref);
+    if (!upper) {
+      return kRowsMightMatch;
+    }
+    return !std::get<std::string>(upper->value()).starts_with(prefix);
+  }
+
+ private:
+  bool ContainsNullsOnly(const std::shared_ptr<BoundReference>& ref) const {
+    const auto stats = GetStatistics(ref);
+    if (!stats || !stats->HasNullCount()) {
+      return false;
+    }
+    const auto column =
+        row_group_.ColumnChunk(field_id_to_column_idx_.at(ref->field_id()));
+    return stats->null_count() == column->num_values();
+  }
+
+  bool MayContainNull(const std::shared_ptr<BoundReference>& 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<BoundReference>& ref) const {
+    if (!ref || !ref->type()->is_primitive() ||
+        MetadataColumns::IsMetadataColumn(ref->field_id())) {
+      return nullptr;
+    }
+    auto column = field_id_to_column_idx_.find(ref->field_id());
+    // Missing columns can have initial defaults, so do not assume all nulls.
+    if (column == field_id_to_column_idx_.end()) {
+      return nullptr;
+    }
+    auto field = manifest_.column_index_to_field.find(column->second);
+    if (field == manifest_.column_index_to_field.end() ||
+        !ValidateParquetTypeCompatibility(*ref->type(), *field->second)) {
+      return nullptr;
+    }
+    return row_group_.ColumnChunk(column->second)->statistics();
+  }
+
+  std::optional<Literal> MinValue(const std::shared_ptr<BoundReference>& ref) const {
+    return GetBound(ref, /*is_min=*/true);
+  }
+
+  std::optional<Literal> MaxValue(const std::shared_ptr<BoundReference>& ref) const {
+    return GetBound(ref, /*is_min=*/false);
+  }
+
+  std::optional<Literal> GetBound(const std::shared_ptr<BoundReference>& ref,
+                                  bool is_min) const {
+    const auto stats = GetStatistics(ref);
+    if (!stats || !stats->HasMinMax()) {
+      return std::nullopt;
+    }
+    const auto& type = static_cast<const PrimitiveType&>(*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 <typename Comparator>
+  bool VisitInequality(const std::shared_ptr<BoundReference>& 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<int32_t, int>& field_id_to_column_idx_;
+};
+
+}  // namespace
+
+Result<std::unique_ptr<ParquetMetricsRowGroupFilter>> ParquetMetricsRowGroupFilter::Make(
+    const std::shared_ptr<Expression>& filter,
+    const ::parquet::SchemaDescriptor& file_schema) {
+  auto result =
+      std::unique_ptr<ParquetMetricsRowGroupFilter>(new ParquetMetricsRowGroupFilter());
+  for (int i = 0; i < file_schema.num_columns(); ++i) {
+    auto id = file_schema.Column(i)->schema_node()->field_id();
+    if (id >= 0) {
+      result->field_id_to_column_idx_.emplace(id, i);
+    }
+  }
+  result->bound_ = True::Instance();
+  if (!filter) {
+    return result;
+  }
+  ICEBERG_ASSIGN_OR_RAISE(result->bound_, RewriteNot::Visit(filter));
+  return result;
+}
+
+Result<bool> ParquetMetricsRowGroupFilter::ShouldRead(
+    const ::parquet::arrow::SchemaManifest& manifest,
+    const ::parquet::RowGroupMetaData& row_group) const {
+  if (row_group.num_rows() <= 0) {
+    return kRowsCannotMatch;
+  }
+  try {
+    MetricsVisitor visitor(manifest, row_group, field_id_to_column_idx_);
+    return Visit<bool>(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..5d80fe3bf
--- /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 <cstdint>
+#include <memory>
+#include <unordered_map>
+
+#include <parquet/arrow/schema.h>
+
+#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<std::unique_ptr<ParquetMetricsRowGroupFilter>> Make(
+      const std::shared_ptr<Expression>& filter,
+      const ::parquet::SchemaDescriptor& file_schema);
+
+  Result<bool> ShouldRead(const ::parquet::arrow::SchemaManifest& manifest,
+                          const ::parquet::RowGroupMetaData& row_group) const;
+
+ private:
+  ParquetMetricsRowGroupFilter() = default;
+  std::shared_ptr<Expression> bound_;
+  std::unordered_map<int32_t, int> field_id_to_column_idx_;
+};
+
+}  // namespace iceberg::parquet
diff --git a/src/iceberg/parquet/parquet_reader.cc b/src/iceberg/parquet/parquet_reader.cc
index 549a3d3ae..b88dfd000 100644
--- a/src/iceberg/parquet/parquet_reader.cc
+++ b/src/iceberg/parquet/parquet_reader.cc
@@ -41,7 +41,9 @@
 #include "iceberg/arrow/arrow_io_internal.h"
 #include "iceberg/arrow/arrow_status_internal.h"
 #include "iceberg/arrow/metadata_column_util_internal.h"
+#include "iceberg/expression/binder.h"
 #include "iceberg/parquet/parquet_data_util_internal.h"
+#include "iceberg/parquet/parquet_metrics_row_group_filter_internal.h"
 #include "iceberg/parquet/parquet_register.h"
 #include "iceberg/parquet/parquet_schema_util_internal.h"
 #include "iceberg/parquet/row_number_reader_internal.h"
@@ -65,21 +67,20 @@ Result<SchemaProjection> BuildProjection(::parquet::arrow::FileReader* reader,
                                          const NameMapping* name_mapping) {
   auto metadata = reader->parquet_reader()->metadata();
 
-  ::parquet::arrow::SchemaManifest schema_manifest;
-  ICEBERG_ARROW_RETURN_NOT_OK(::parquet::arrow::SchemaManifest::Make(
-      metadata->schema(), metadata->key_value_metadata(), reader->properties(),
-      &schema_manifest));
-
   if (!HasFieldIds(metadata->schema()->schema_root())) {
     if (name_mapping == nullptr) {
       return InvalidSchema(
           "Parquet file schema has no field IDs and no name mapping provided");
     }
+    ::parquet::arrow::SchemaManifest schema_manifest;
+    ICEBERG_ARROW_RETURN_NOT_OK(::parquet::arrow::SchemaManifest::Make(
+        metadata->schema(), metadata->key_value_metadata(), reader->properties(),
+        &schema_manifest));
     ApplyNameMapping(*name_mapping, &schema_manifest);
+    return Project(read_schema, 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;
 }
 
@@ -282,6 +283,7 @@ class ParquetReader::Impl {
     }
 
     read_schema_ = options.projection;
+    stats_filter_.reset();
 
     // Prepare reader properties
     ::parquet::ReaderProperties reader_properties(pool_);
@@ -301,11 +303,24 @@ class ParquetReader::Impl {
     ICEBERG_ARROW_ASSIGN_OR_RETURN(
         reader_, ::parquet::arrow::FileReader::Make(pool_, std::move(file_reader),
                                                     arrow_reader_properties));
-    ICEBERG_RETURN_UNEXPECTED(PlanRowGroups(options));
 
     // Project read schema onto the Parquet file schema
     ICEBERG_ASSIGN_OR_RAISE(projection_, BuildProjection(reader_.get(), *read_schema_,
                                                          options.name_mapping.get()));
+    if (options.filter &&
+        options.properties.Get(ReaderProperties::kParquetRowGroupFilter)) {
+      auto filter = options.filter;
+      ICEBERG_ASSIGN_OR_RAISE(auto is_bound, IsBoundVisitor::IsBound(filter));
+      if (!is_bound) {
+        ICEBERG_ASSIGN_OR_RAISE(
+            filter, Binder::Bind(*options.projection, filter, options.case_sensitive));
+      }
+      ICEBERG_ASSIGN_OR_RAISE(stats_filter_, ParquetMetricsRowGroupFilter::Make(
+                                                 filter, *reader_->manifest().descr));
+    }
+
+    ICEBERG_RETURN_UNEXPECTED(PlanRowGroups(options));
+
     metadata_context_ = {.file_path = options.path,
                          .first_row_id = options.first_row_id,
                          .data_sequence_number = options.data_sequence_number};
@@ -395,7 +410,6 @@ class ParquetReader::Impl {
 
  private:
   Status PlanRowGroups(const ReaderOptions& options) {
-    // TODO(gangwu): add row group filtering based on zone map, bloom filter, etc.
     auto metadata = reader_->parquet_reader()->metadata();
     if (options.split.has_value()) {
       const auto& split = *options.split;
@@ -431,6 +445,19 @@ class ParquetReader::Impl {
       row_group_indices_.resize(metadata->num_row_groups());
       std::iota(row_group_indices_.begin(), row_group_indices_.end(), 0);  // NOLINT
     }
+
+    if (stats_filter_) {
+      size_t selected_count = 0;
+      for (int row_group_index : row_group_indices_) {
+        auto row_group = metadata->RowGroup(row_group_index);
+        ICEBERG_ASSIGN_OR_RAISE(
+            auto should_read, stats_filter_->ShouldRead(reader_->manifest(), *row_group));
+        if (should_read) {
+          row_group_indices_[selected_count++] = row_group_index;
+        }
+      }
+      row_group_indices_.resize(selected_count);
+    }
     return {};
   }
 
@@ -484,6 +511,8 @@ class ParquetReader::Impl {
   bool use_large_list_ = false;
   // Schema to read from the Parquet file.
   std::shared_ptr<::iceberg::Schema> read_schema_;
+  // Footer-statistics filter for row group pruning, if enabled.
+  std::unique_ptr<ParquetMetricsRowGroupFilter> 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 2a230fe4c..d16b8a1ae 100644
--- a/src/iceberg/parquet/parquet_schema_util.cc
+++ b/src/iceberg/parquet/parquet_schema_util.cc
@@ -343,8 +343,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();
@@ -502,8 +500,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 6612de056..49c62ff09 100644
--- a/src/iceberg/parquet/parquet_schema_util_internal.h
+++ b/src/iceberg/parquet/parquet_schema_util_internal.h
@@ -39,6 +39,13 @@ struct ParquetExtraAttributes : public FieldProjection::ExtraAttributes {
   std::optional<int32_t> 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/row/partition_values.cc b/src/iceberg/row/partition_values.cc
index fcd4691d0..edf55a772 100644
--- a/src/iceberg/row/partition_values.cc
+++ b/src/iceberg/row/partition_values.cc
@@ -19,6 +19,10 @@
 
 #include "iceberg/row/partition_values.h"
 
+#include <string_view>
+
+#include "iceberg/util/uuid.h"
+
 namespace iceberg {
 
 PartitionValues& PartitionValues::operator=(const PartitionValues& other) {
@@ -74,6 +78,11 @@ Result<Scalar> PartitionValues::GetField(size_t pos) const {
       return Scalar{
           std::string_view(reinterpret_cast<const char*>(bytes.data()), bytes.size())};
     }
+    case TypeId::kUuid: {
+      const auto bytes = std::get<Uuid>(literal.value()).bytes();
+      return Scalar{
+          std::string_view(reinterpret_cast<const char*>(bytes.data()), bytes.size())};
+    }
     case TypeId::kDecimal:
       return Scalar{std::get<Decimal>(literal.value())};
     default:
diff --git a/src/iceberg/test/CMakeLists.txt b/src/iceberg/test/CMakeLists.txt
index 37e20e339..534cbfc13 100644
--- a/src/iceberg/test/CMakeLists.txt
+++ b/src/iceberg/test/CMakeLists.txt
@@ -224,6 +224,7 @@ if(ICEBERG_BUILD_BUNDLE)
                    SOURCES
                    metrics_test_base.cc
                    parquet_data_test.cc
+                   parquet_row_group_filter_test.cc
                    parquet_metrics_test.cc
                    parquet_schema_test.cc
                    parquet_test.cc
diff --git a/src/iceberg/test/evaluator_test.cc b/src/iceberg/test/evaluator_test.cc
index e365a7d3e..e446bc66c 100644
--- a/src/iceberg/test/evaluator_test.cc
+++ b/src/iceberg/test/evaluator_test.cc
@@ -36,6 +36,7 @@
 #include "iceberg/arrow_c_data_guard_internal.h"
 #include "iceberg/expression/expressions.h"
 #include "iceberg/expression/literal.h"
+#include "iceberg/expression/rewrite_not.h"
 #include "iceberg/result.h"
 #include "iceberg/row/arrow_array_wrapper.h"
 #include "iceberg/schema.h"
@@ -705,6 +706,58 @@ TEST_F(EvaluatorTest, Not) {
       *struct_evaluator, true);
 }
 
+TEST_F(EvaluatorTest, NotRangeComparisonsIncludeNull) {
+  struct Case {
+    std::shared_ptr<Expression> predicate;
+    bool below;
+    bool equal;
+    bool above;
+  };
+  const std::vector<Case> cases = {
+      {.predicate = Expressions::LessThan("z", Literal::Int(7)),
+       .below = false,
+       .equal = true,
+       .above = true},
+      {.predicate = Expressions::LessThanOrEqual("z", Literal::Int(7)),
+       .below = false,
+       .equal = false,
+       .above = true},
+      {.predicate = Expressions::GreaterThan("z", Literal::Int(7)),
+       .below = true,
+       .equal = true,
+       .above = false},
+      {.predicate = Expressions::GreaterThanOrEqual("z", Literal::Int(7)),
+       .below = true,
+       .equal = false,
+       .above = false}};
+  for (const auto& test : cases) {
+    SCOPED_TRACE(test.predicate->ToString());
+    ICEBERG_UNWRAP_OR_FAIL(auto evaluator,
+                           Evaluator::Make(*schema_, Expressions::Not(test.predicate)));
+    TestData(R"([{"x": 0, "y": 0.0, "z": null}])", *evaluator, true);
+    TestData(R"([{"x": 0, "y": 0.0, "z": 6}])", *evaluator, test.below);
+    TestData(R"([{"x": 0, "y": 0.0, "z": 7}])", *evaluator, test.equal);
+    TestData(R"([{"x": 0, "y": 0.0, "z": 8}])", *evaluator, test.above);
+  }
+}
+
+TEST_F(EvaluatorTest, NegatedRangePreservesNullableAncestors) {
+  for (const auto& constant :
+       {Literal::Int(7), Literal::Long(std::numeric_limits<int64_t>::max())}) {
+    auto expr = Expressions::Not(Expressions::LessThan("s1.s2.s3.s4.i", constant));
+    ICEBERG_UNWRAP_OR_FAIL(auto original, Evaluator::Make(*schema_, expr));
+    ICEBERG_UNWRAP_OR_FAIL(auto rewritten, RewriteNot::Visit(expr));
+    ICEBERG_UNWRAP_OR_FAIL(auto normalized, Evaluator::Make(*schema_, rewritten));
+    for (auto* evaluator : {original.get(), normalized.get()}) {
+      TestData(R"([{"x": 0, "y": 0.0, "s1": null}])", *evaluator, true);
+      TestData(R"([{"x": 0, "y": 0.0, "s1": {"s2": {"s3": {"s4": {"i": 6}}}}}])",
+               *evaluator, false);
+      TestData(R"([{"x": 0, "y": 0.0, "s1": {"s2": {"s3": {"s4": {"i": 7}}}}}])",
+               *evaluator, constant == Literal::Int(7));
+    }
+  }
+}
+
 TEST_F(EvaluatorTest, CaseInsensitiveNot) {
   // Use case-insensitive binding (false)
   ICEBERG_UNWRAP_OR_FAIL(
diff --git a/src/iceberg/test/expression_visitor_test.cc b/src/iceberg/test/expression_visitor_test.cc
index d24142cf1..74d8537b9 100644
--- a/src/iceberg/test/expression_visitor_test.cc
+++ b/src/iceberg/test/expression_visitor_test.cc
@@ -549,6 +549,31 @@ TEST_F(RewriteNotTest, ComplexExpression) {
   EXPECT_EQ(rewritten->op(), Expression::Operation::kOr);
 }
 
+TEST_F(RewriteNotTest, NullableRangeComparisons) {
+  const auto value = Literal::Double(7);
+  const std::vector<std::pair<std::shared_ptr<Expression>, std::shared_ptr<Expression>>>
+      cases = {{Expressions::LessThan("salary", value),
+                Expressions::GreaterThanOrEqual("salary", value)},
+               {Expressions::LessThanOrEqual("salary", value),
+                Expressions::GreaterThan("salary", value)},
+               {Expressions::GreaterThan("salary", value),
+                Expressions::LessThanOrEqual("salary", value)},
+               {Expressions::GreaterThanOrEqual("salary", value),
+                Expressions::LessThan("salary", value)}};
+  for (const auto& [predicate, inverse] : cases) {
+    SCOPED_TRACE(predicate->ToString());
+    auto expr = Expressions::Not(predicate);
+    auto expected = Expressions::Or(Expressions::IsNull("salary"), inverse);
+    ICEBERG_UNWRAP_OR_FAIL(auto rewritten, RewriteNot::Visit(expr));
+    EXPECT_TRUE(rewritten->Equals(*expected));
+
+    ICEBERG_UNWRAP_OR_FAIL(auto bound, Bind(expr));
+    ICEBERG_UNWRAP_OR_FAIL(auto rewritten_bound, RewriteNot::Visit(bound));
+    ICEBERG_UNWRAP_OR_FAIL(auto expected_bound, Bind(expected));
+    EXPECT_TRUE(rewritten_bound->Equals(*expected_bound));
+  }
+}
+
 // gtest's MatchesRegex/ContainsRegex falls back to a minimal "simple regex" engine on
 // MSVC (no groups, character classes, repetition, or alternation), unlike the POSIX
 // extended regex used on Linux/macOS. std::regex (ECMAScript grammar) behaves
diff --git a/src/iceberg/test/file_scan_task_reader_test.cc b/src/iceberg/test/file_scan_task_reader_test.cc
index 05416bd7d..8b0d79d15 100644
--- a/src/iceberg/test/file_scan_task_reader_test.cc
+++ b/src/iceberg/test/file_scan_task_reader_test.cc
@@ -46,8 +46,11 @@
 #include "iceberg/arrow_c_data_util_internal.h"
 #include "iceberg/data/equality_delete_writer.h"
 #include "iceberg/data/position_delete_writer.h"
+#include "iceberg/data_table_scan.h"
 #include "iceberg/deletes/dv_writer.h"
 #include "iceberg/deletes/position_delete_index.h"
+#include "iceberg/expression/binder.h"
+#include "iceberg/expression/expressions.h"
 #include "iceberg/file_format.h"
 #include "iceberg/file_io.h"
 #include "iceberg/file_reader.h"
@@ -862,4 +865,196 @@ TEST_F(FileScanTaskReaderTest, OpenWithMixedDeletesSkipsFullyDeletedBatches) {
   ASSERT_NO_FATAL_FAILURE(VerifyStream(&stream, R"([[3, "Baz"]])"));
 }
 
+TEST_F(FileScanTaskReaderTest, RowGroupPruningPreservesDeletesAndLineage) {
+  auto lineage_schema = RowLineageProjection();
+  std::vector<SchemaField> read_fields(lineage_schema->fields().begin(),
+                                       lineage_schema->fields().end());
+  read_fields.push_back(table_schema_->fields()[2]);
+  auto read_schema = std::make_shared<Schema>(std::move(read_fields));
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto data_file,
+      MakeDataFile(table_schema_,
+                   std::vector<std::string>{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<int32_t> actual_ids;
+    std::vector<int64_t> 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<int32_t>{0, 1, 5}));
+    EXPECT_EQ(lineage, (std::vector<int64_t>{100, 101, 105}));
+  }
+}
+
+TEST_F(FileScanTaskReaderTest, RowGroupPruningUsesCallerReadSchema) {
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto data_file,
+      MakeDataFile(table_schema_,
+                   std::vector<std::string>{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<std::shared_ptr<Expression>>{
+           filter, bound, Expressions::Equal("CATEGORY", Literal::String("keep"))}) {
+    FileScanTask task(data_file, {}, predicate);
+    FileScanTaskReader::Options options{
+        .io = file_io_,
+        .table_schema = table_schema_,
+        .projected_schema = table_schema_,
+    };
+    options.case_sensitive = false;
+    ICEBERG_UNWRAP_OR_FAIL(auto reader, FileScanTaskReader::Make(std::move(options)));
+    ICEBERG_UNWRAP_OR_FAIL(auto stream, reader->Open(task));
+    VerifyStream(&stream, R"([[2,"c","keep"],[3,"d","keep"]])");
+  }
+}
+
+TEST_F(FileScanTaskReaderTest, ConstantResidualsAndNestedConstants) {
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto data_file, MakeDataFile(table_schema_, R"([[0,"a","keep"],[1,"b","keep"]])"));
+  auto predicate = Expressions::Equal("name", Literal::String("a"));
+  ICEBERG_UNWRAP_OR_FAIL(std::shared_ptr<Expression> conjunction,
+                         And::Make(Expressions::AlwaysTrue(), predicate));
+  ICEBERG_UNWRAP_OR_FAIL(std::shared_ptr<Expression> 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<std::shared_ptr<Expression>>{
+           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<std::string>{"id", "name"};
+    } else {
+      // Explicit projections are prepared by the caller, including filter columns.
+      context.projected_schema = table_schema_;
+    }
+    auto metadata = std::make_shared<TableMetadata>();
+    metadata->schemas = {table_schema_};
+    metadata->current_schema_id = table_schema_->schema_id();
+    ICEBERG_UNWRAP_OR_FAIL(
+        auto scan, DataTableScan::Make(metadata, table_schema_, file_io_, context));
+    ICEBERG_UNWRAP_OR_FAIL(auto projection, scan->schema());
+    FileScanTaskReader::Options options{
+        .io = scan->io(),
+        .table_schema = table_schema_,
+        .projected_schema = projection,
+        .case_sensitive = scan->is_case_sensitive(),
+    };
+    FileScanTask task(data_file, {}, scan->filter());
+    ICEBERG_UNWRAP_OR_FAIL(auto reader, FileScanTaskReader::Make(options));
+    ICEBERG_UNWRAP_OR_FAIL(auto stream, reader->Open(task));
+    VerifyStream(&stream, R"([[0,"a","keep"],[1,"b","keep"]])");
+    options.case_sensitive = true;
+    ICEBERG_UNWRAP_OR_FAIL(auto case_sensitive_reader, FileScanTaskReader::Make(options));
+    EXPECT_THAT(case_sensitive_reader->Open(task),
+                IsError(ErrorKind::kInvalidExpression));
+  }
+}
+
+TEST_F(FileScanTaskReaderTest, UnboundFilterRequiresProjectedField) {
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto data_file, MakeDataFile(table_schema_, R"([[0,"a","keep"],[1,"b","keep"]])"));
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto reader, FileScanTaskReader::Make({.io = file_io_,
+                                             .table_schema = table_schema_,
+                                             .projected_schema = projected_schema_}));
+  FileScanTask task(data_file, {},
+                    Expressions::Equal("category", Literal::String("keep")));
+  EXPECT_THAT(reader->Open(task), IsError(ErrorKind::kInvalidExpression));
+}
+
+TEST_F(FileScanTaskReaderTest, NestedFilterFieldsComeFromScanSchema) {
+  table_schema_ = std::make_shared<Schema>(std::vector<SchemaField>{
+      SchemaField::MakeRequired(1, "id", int32()),
+      SchemaField::MakeOptional(2, "record",
+                                std::make_shared<StructType>(std::vector<SchemaField>{
+                                    SchemaField::MakeOptional(3, "value", string()),
+                                    SchemaField::MakeOptional(4, "key", int32())}))});
+  internal::TableScanContext context;
+  context.selected_columns = std::vector<std::string>{"id", "record.value"};
+  context.filter = Expressions::GreaterThan("record.key", Literal::Int(5));
+  auto metadata = std::make_shared<TableMetadata>();
+  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<std::string>{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<std::shared_ptr<DataFile>>{},
+        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/literal_test.cc b/src/iceberg/test/literal_test.cc
index 6f515cdf7..60bd8f3ec 100644
--- a/src/iceberg/test/literal_test.cc
+++ b/src/iceberg/test/literal_test.cc
@@ -414,8 +414,15 @@ TEST(LiteralTest, UuidComparison) {
   auto literal3 = Literal::UUID(uuid3);
 
   EXPECT_EQ(literal1 <=> literal3, std::partial_ordering::equivalent);
-  EXPECT_EQ(literal1 <=> literal2, std::partial_ordering::unordered);
-  EXPECT_EQ(literal2 <=> literal1, std::partial_ordering::unordered);
+  EXPECT_EQ(literal1 <=> literal2, std::partial_ordering::less);
+  EXPECT_EQ(literal2 <=> literal1, std::partial_ordering::greater);
+
+  auto low =
+      Literal::UUID(Uuid::FromString("7fffffff-ffff-ffff-ffff-ffffffffffff").value());
+  auto high =
+      Literal::UUID(Uuid::FromString("80000000-0000-0000-0000-000000000000").value());
+  EXPECT_EQ(low <=> high, std::partial_ordering::less);
+  EXPECT_EQ(high <=> low, std::partial_ordering::greater);
 }
 
 // Parameter struct for literal serialization and deserialization tests
diff --git a/src/iceberg/test/manifest_reader_test.cc b/src/iceberg/test/manifest_reader_test.cc
index 80a8acd72..6a201bd56 100644
--- a/src/iceberg/test/manifest_reader_test.cc
+++ b/src/iceberg/test/manifest_reader_test.cc
@@ -243,6 +243,31 @@ TEST_P(TestManifestReader, TestManifestReaderWithEmptyInheritableMetadata) {
   EXPECT_EQ(read_entry.snapshot_id, 1000L);
 }
 
+TEST_P(TestManifestReader, NegatedRowRangeRetainsNullPartitions) {
+  schema_ = std::make_shared<Schema>(
+      std::vector<SchemaField>{SchemaField::MakeOptional(3, "key", int32())});
+  ICEBERG_UNWRAP_OR_FAIL(
+      spec_, PartitionSpec::Make(
+                 *schema_, 0,
+                 {PartitionField(3, 1000, "key_part", Transform::Identity())}, false));
+  std::vector<ManifestEntry> entries;
+  for (const auto& value :
+       {Literal::Null(int32()), Literal::Int(50), Literal::Int(150)}) {
+    entries.push_back(MakeEntry(ManifestStatus::kAdded, 1000L,
+                                MakeDataFile("/path/to/" + value.ToString() + ".parquet",
+                                             PartitionValues(value))));
+  }
+  auto manifest = WriteManifest(GetParam(), 1000L, entries);
+  auto filter = Expressions::Not(Expressions::GreaterThan("key", Literal::Int(100)));
+  ICEBERG_UNWRAP_OR_FAIL(auto reader,
+                         ManifestReader::Make(manifest, file_io_, schema_, spec_));
+  reader->FilterRows(filter);
+  ICEBERG_UNWRAP_OR_FAIL(auto actual, reader->Entries());
+  ASSERT_EQ(actual.size(), 2);
+  EXPECT_TRUE(actual[0].data_file->partition.values()[0].IsNull());
+  EXPECT_EQ(actual[1].data_file->partition.values()[0], Literal::Int(50));
+}
+
 TEST_P(TestManifestReader, EntriesStreamOwnsReaderResources) {
   auto version = GetParam();
   auto file_a =
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..cf44af16c
--- /dev/null
+++ b/src/iceberg/test/parquet_row_group_filter_test.cc
@@ -0,0 +1,859 @@
+/*
+ * 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 <arrow/array.h>
+#include <arrow/builder.h>
+#include <arrow/c/bridge.h>
+#include <arrow/extension/uuid.h>
+#include <arrow/json/from_string.h>
+#include <arrow/record_batch.h>
+#include <arrow/table.h>
+#include <parquet/arrow/writer.h>
+#include <parquet/file_reader.h>
+#include <parquet/metadata.h>
+#include <parquet/statistics.h>
+
+#include "iceberg/arrow/arrow_io_internal.h"
+#include "iceberg/arrow/arrow_status_internal.h"
+#include "iceberg/expression/binder.h"
+#include "iceberg/expression/expressions.h"
+#include "iceberg/file_reader.h"
+#include "iceberg/metadata_columns.h"
+#include "iceberg/name_mapping.h"
+#include "iceberg/parquet/parquet_data_util_internal.h"
+#include "iceberg/parquet/parquet_metrics_row_group_filter_internal.h"
+#include "iceberg/parquet/parquet_register.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_internal.h"
+#include "iceberg/test/matchers.h"
+#include "iceberg/test/mock_io.h"
+#include "iceberg/type.h"
+#include "iceberg/util/uuid.h"
+
+namespace iceberg::parquet {
+namespace {
+
+class ParquetRowGroupFilterTest : public ::testing::Test {
+ protected:
+  void SetUp() override {
+    parquet::RegisterAll();
+    io_ = std::make_shared<MockFileIO>();
+    SetKeyType(int32());
+    projection_ = std::make_shared<Schema>(
+        std::vector<SchemaField>{SchemaField::MakeRequired(2, "value", int64()),
+                                 MetadataColumns::kRowPosition, MetadataColumns::kRowId});
+  }
+
+  void SetKeyType(std::shared_ptr<Type> type) {
+    schema_ = std::make_shared<Schema>(
+        std::vector<SchemaField>{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<Expression> filter, bool case_sensitive = true) {
+    // Bind against the table schema: key is intentionally outside the projection.
+    // Tests of Open's binding behavior set options.filter directly instead.
+    if (filter && filter->op() != Expression::Operation::kTrue &&
+        filter->op() != Expression::Operation::kFalse) {
+      auto is_bound = IsBoundVisitor::IsBound(filter);
+      EXPECT_THAT(is_bound, IsOk());
+      if (is_bound && !*is_bound) {
+        auto bound = Binder::Bind(*schema_, filter, case_sensitive);
+        EXPECT_THAT(bound, IsOk());
+        if (bound) {
+          filter = *bound;
+        }
+      }
+    }
+    ReaderOptions options{.path = path_,
+                          .io = io_,
+                          .projection = projection_,
+                          .filter = std::move(filter),
+                          .first_row_id = 100};
+    options.case_sensitive = case_sensitive;
+    options.properties.Set(ReaderProperties::kBatchSize, int64_t{3});
+    return options;
+  }
+
+  Result<std::vector<int64_t>> 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<int64_t> 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<int64_t>& 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<FileIO> io_;
+  std::shared_ptr<Schema> schema_;
+  std::shared_ptr<Schema> 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<Schema>(std::vector<SchemaField>{
+      SchemaField::MakeOptional(1, "renamed", int32()), schema_->fields()[1]});
+  auto filter = Expressions::Equal("renamed", Literal::Int(20));
+  Check(Options(filter), {4, 5});
+  ICEBERG_UNWRAP_OR_FAIL(auto bound, Binder::Bind(*schema_, filter, true));
+  auto options = Options(bound);
+  Check(options, {4, 5});
+  options = Options(Expressions::Equal("RENAMED", Literal::Int(20)), false);
+  Check(options, {4, 5});
+}
+
+TEST_F(ParquetRowGroupFilterTest, OpenBindsUnboundProjectedReferences) {
+  ASSERT_THAT(Write(), IsOk());
+  auto options = Options(nullptr);
+  options.filter = Expressions::GreaterThanOrEqual("VALUE", Literal::Long(4));
+  EXPECT_THAT(ReaderFactoryRegistry::Open(FileFormatType::kParquet, options),
+              HasErrorMessage("Cannot find field 'VALUE'"));
+  options.case_sensitive = false;
+  Check(options, {4, 5});
+}
+
+TEST_F(ParquetRowGroupFilterTest, MissingStatisticsRetainGroups) {
+  ASSERT_THAT(Write(false), IsOk());
+  Check(Options(Expressions::Equal("key", Literal::Int(100))), {0, 1, 2, 3, 4, 5});
+  Check(Options(Expressions::IsNull("key")), {0, 1, 2, 3, 4, 5});
+  Check(Options(Expressions::NotNull("key")), {0, 1, 2, 3, 4, 5});
+}
+
+TEST_F(ParquetRowGroupFilterTest, MissingNullCountRetainsNullBranches) {
+  ASSERT_THAT(Write(true, "[[null,0],[150,1]]"), IsOk());
+  auto original = metadata_->RowGroup(0);
+  auto builder = ::parquet::FileMetaDataBuilder::Make(
+      metadata_->schema(), ::parquet::default_writer_properties());
+  auto row_group = builder->AppendRowGroup();
+  row_group->set_num_rows(original->num_rows());
+  for (int i = 0; i < original->num_columns(); ++i) {
+    auto column = original->ColumnChunk(i);
+    auto stats = column->statistics()->Encode();
+    if (i == 0) {
+      stats.has_null_count = false;
+    }
+    auto chunk = row_group->NextColumnChunk();
+    chunk->SetStatistics(stats);
+    chunk->Finish(column->num_values(), column->dictionary_page_offset(),
+                  /*index_page_offset=*/0, column->data_page_offset(),
+                  column->total_compressed_size(), column->total_uncompressed_size(),
+                  column->has_dictionary_page(), /*dictionary_fallback=*/false, {}, {});
+  }
+  row_group->Finish(original->total_byte_size());
+  auto rebuilt = builder->Finish();
+  auto group = rebuilt->RowGroup(0);
+  auto stats = group->ColumnChunk(0)->statistics();
+  ASSERT_NE(stats, nullptr);
+  ASSERT_TRUE(stats->HasMinMax());
+  ASSERT_FALSE(stats->HasNullCount());
+
+  ::parquet::arrow::SchemaManifest manifest;
+  ASSERT_TRUE(::parquet::arrow::SchemaManifest::Make(
+                  rebuilt->schema(), /*key_value_metadata=*/nullptr,
+                  ::parquet::default_arrow_reader_properties(), &manifest)
+                  .ok());
+  struct Case {
+    std::shared_ptr<Expression> filter;
+    bool should_read;
+  };
+  for (const auto& test : std::vector<Case>{
+           {.filter = Expressions::IsNull("key"), .should_read = true},
+           {.filter = Expressions::NotNull("key"), .should_read = true},
+           {.filter =
+                Expressions::Not(Expressions::GreaterThan("key", Literal::Int(100))),
+            .should_read = true},
+           {.filter = Expressions::LessThan("key", Literal::Int(100)),
+            .should_read = false}}) {
+    SCOPED_TRACE(test.filter->ToString());
+    ICEBERG_UNWRAP_OR_FAIL(auto filter,
+                           ParquetMetricsRowGroupFilter::Make(Options(test.filter).filter,
+                                                              *rebuilt->schema()));
+    ICEBERG_UNWRAP_OR_FAIL(auto should_read, filter->ShouldRead(manifest, *group));
+    EXPECT_EQ(should_read, test.should_read);
+  }
+}
+
+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<Schema>(std::vector<SchemaField>{
+      schema_->fields()[0], schema_->fields()[1],
+      SchemaField::MakeOptional(3, "defaulted", int32())
+          .WithInitialDefault(std::make_shared<Literal>(Literal::Int(7)))});
+  Check(Options(Expressions::Equal("defaulted", Literal::Int(7))), {0, 1, 2, 3, 4, 5});
+}
+
+TEST_F(ParquetRowGroupFilterTest, FileAndFilterTypeCompatibility) {
+  struct Case {
+    std::shared_ptr<Type> file_type;
+    std::shared_ptr<Type> 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<Case>{{.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);
+    const bool floating = test.filter_type->type_id() == TypeId::kFloat ||
+                          test.filter_type->type_id() == TypeId::kDouble;
+    Check(Options(Expressions::Equal("key", test.value)),
+          !floating && test.compatible ? std::vector<int64_t>{2, 3}
+                                       : std::vector<int64_t>{0, 1, 2, 3, 4, 5});
+    if (floating) {
+      // Null counts still distinguish compatible types when bounds are disabled.
+      SetKeyType(test.file_type);
+      ASSERT_THAT(Write(true, "[[null,0],[null,1],[10,2],[11,3]]"), IsOk());
+      SetKeyType(test.filter_type);
+      Check(Options(Expressions::IsNull("key")), test.compatible
+                                                     ? std::vector<int64_t>{0, 1}
+                                                     : std::vector<int64_t>{0, 1, 2, 3});
+      Check(Options(Expressions::NotNull("key")), test.compatible
+                                                      ? std::vector<int64_t>{2, 3}
+                                                      : std::vector<int64_t>{0, 1, 2, 3});
+    }
+  }
+}
+
+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<BoundTransform>(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<BoundTransform>(Expressions::Bucket("key", 16), Literal::Int(0));
+  Check(Options(Expressions::Not(Expressions::Or(less_than, unsupported))), {2, 3, 4, 5});
+  Check(Options(Expressions::Not(Expressions::And(less_than, unsupported))),
+        {0, 1, 2, 3, 4, 5});
+
+  ASSERT_THAT(Write(true, "[[null,0],[null,1],[10,2],[11,3],[20,4],[21,5]]"), IsOk());
+  Check(Options(Expressions::Not(Expressions::IsNull("key"))), {2, 3, 4, 5});
+  Check(Options(Expressions::Not(Expressions::NotNull("key"))), {0, 1});
+}
+
+TEST_F(ParquetRowGroupFilterTest, RewrittenNotPrunesNullableGroups) {
+  ASSERT_THAT(Write(true, "[[null,0],[null,1],[null,2],[50,3],[100,4],[150,5]]"), IsOk());
+  // Negated ranges match nulls; non-null groups still use range pruning.
+  Check(Options(Expressions::Not(Expressions::LessThan("key", Literal::Int(100)))),
+        {0, 1, 2, 3, 4, 5});
+  Check(Options(Expressions::Not(Expressions::LessThanOrEqual("key", Literal::Int(100)))),
+        {0, 1, 2, 3, 4, 5});
+  Check(Options(Expressions::Not(Expressions::GreaterThan("key", Literal::Int(100)))),
+        {0, 1, 2, 3, 4, 5});
+  Check(Options(
+            Expressions::Not(Expressions::GreaterThanOrEqual("key", Literal::Int(100)))),
+        {0, 1, 2, 3});
+}
+
+TEST_F(ParquetRowGroupFilterTest, BoundComparisonBoundariesAndAllNullGroups) {
+  ASSERT_THAT(Write(true, "[[null,0],[null,1],[10,2],[11,3],[20,4],[21,5]]"), IsOk());
+  struct Case {
+    std::shared_ptr<Expression> filter;
+    std::vector<int64_t> expected;
+  };
+  for (const auto& test : std::vector<Case>{
+           {.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<Literal> 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<SchemaField> projected_fields(fields.begin(), fields.end());
+  projected_fields.push_back(schema_->fields()[0]);
+  options.projection = std::make_shared<Schema>(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<Expression> 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());
+  ICEBERG_UNWRAP_OR_FAIL(auto input, arrow::OpenArrowInputStream(io_, path_));
+  auto file_size = input->GetSize().ValueOrDie();
+  ASSERT_TRUE(input->Close().ok());
+  ICEBERG_UNWRAP_OR_FAIL(auto midpoint,
+                         RowGroupMidpoint(*metadata_->RowGroup(1), file_size));
+  auto options = Options(Expressions::GreaterThanOrEqual("key", Literal::Int(20)));
+  options.split = Split{.offset = static_cast<size_t>(midpoint), .length = 100000};
+  Check(options, {4, 5});
+  ICEBERG_UNWRAP_OR_FAIL(
+      options.filter,
+      Binder::Bind(*schema_, Expressions::LessThan("key", Literal::Int(2)), true));
+  Check(options, {});
+
+  // A one-byte split selects only the group with this midpoint. Statistics must
+  // refine that selection without adding groups from outside the split.
+  options.split->length = 1;
+  options.filter = Options(Expressions::Equal("key", Literal::Int(10))).filter;
+  Check(options, {2, 3});
+  options.filter = Options(Expressions::Equal("key", Literal::Int(20))).filter;
+  Check(options, {});
+  options.properties.Set(ReaderProperties::kParquetRowGroupFilter, false);
+  Check(options, {2, 3});
+  options.filter = nullptr;
+  options.split->length = 0;
+  Check(options, {});
+}
+
+TEST_F(ParquetRowGroupFilterTest, NameMappingWithoutFieldIdsRetainsGroups) {
+  auto keys = ::arrow::json::ArrayFromJSONString(::arrow::int32(), "[0,1,10,11,20,21]")
+                  .ValueOrDie();
+  auto values =
+      ::arrow::json::ArrayFromJSONString(::arrow::int64(), "[0,1,2,3,4,5]").ValueOrDie();
+  auto file_schema = ::arrow::schema({::arrow::field("stored_key", ::arrow::int32()),
+                                      ::arrow::field("stored_value", ::arrow::int64())});
+  ASSERT_THAT(Write(::arrow::Table::Make(file_schema, {keys, values})), IsOk());
+  ASSERT_EQ(metadata_->schema()->Column(0)->schema_node()->field_id(), -1);
+  auto mapping =
+      NameMapping::Make({MappedField{.names = {"stored_key"}, .field_id = 1},
+                         MappedField{.names = {"stored_value"}, .field_id = 2}});
+  auto options = Options(Expressions::Equal("key", Literal::Int(100)));
+  options.name_mapping = std::move(mapping);
+  // Name mapping resolves the projection. Footer columns without IDs remain
+  // unavailable to statistics filtering, so the reader must keep their groups.
+  Check(options, {0, 1, 2, 3, 4, 5});
+}
+
+TEST_F(ParquetRowGroupFilterTest, PrimitiveComparisonTypes) {
+  struct Case {
+    std::shared_ptr<Type> type;
+    std::string json;
+    Literal literal;
+  };
+  for (const auto& test : std::vector<Case>{
+           {.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, UuidBoundsUseUnsignedByteOrder) {
+  SetKeyType(uuid());
+  // Group 0 holds UUIDs below 0x80..., group 1 holds UUIDs at or above it. A signed
+  // byte comparison would order group 1 before group 0.
+  std::vector<Uuid> uuids;
+  for (const auto* str :
+       {"00000000-0000-0000-0000-000000000001", "7fffffff-ffff-ffff-ffff-ffffffffffff",
+        "80000000-0000-0000-0000-000000000000", "ffffffff-ffff-ffff-ffff-fffffffffffe"}) {
+    ICEBERG_UNWRAP_OR_FAIL(auto uuid, Uuid::FromString(str));
+    uuids.push_back(uuid);
+  }
+  ::arrow::FixedSizeBinaryBuilder keys(::arrow::fixed_size_binary(Uuid::kLength));
+  for (const auto& uuid : uuids) {
+    ASSERT_TRUE(keys.Append(uuid.bytes().data()).ok());
+  }
+  auto key_array = ::arrow::ExtensionType::WrapArray(::arrow::extension::uuid(),
+                                                     keys.Finish().ValueOrDie());
+  auto values =
+      ::arrow::json::ArrayFromJSONString(::arrow::int64(), "[0,1,2,3]").ValueOrDie();
+  ArrowSchema c_schema;
+  ASSERT_THAT(ToArrowSchema(*schema_, &c_schema), IsOk());
+  auto arrow_schema = ::arrow::ImportSchema(&c_schema).ValueOrDie();
+  ASSERT_THAT(Write(::arrow::Table::Make(arrow_schema, {key_array, values})), IsOk());
+  ASSERT_EQ(metadata_->num_row_groups(), 2);
+  ASSERT_TRUE(metadata_->RowGroup(0)->ColumnChunk(0)->statistics()->HasMinMax());
+
+  const auto& group0_max = uuids[1];
+  const auto& group1_min = uuids[2];
+  Check(Options(Expressions::LessThan("key", Literal::UUID(group1_min))), {0, 1});
+  Check(Options(Expressions::LessThanOrEqual("key", Literal::UUID(group0_max))), {0, 1});
+  Check(Options(Expressions::GreaterThan("key", Literal::UUID(group0_max))), {2, 3});
+  Check(Options(Expressions::GreaterThanOrEqual("key", Literal::UUID(group1_min))),
+        {2, 3});
+  Check(Options(Expressions::Equal("key", Literal::UUID(group1_min))), {2, 3});
+  Check(
+      Options(Expressions::Not(Expressions::LessThan("key", Literal::UUID(group1_min)))),
+      {2, 3});
+}
+
+TEST_F(ParquetRowGroupFilterTest, FloatingPointStatistics) {
+  // TODO(YinZheng-Sun): remove this once row group pruning is applied to
+  // float/double columns.
+  SetKeyType(float64());
+  ASSERT_THAT(Write(), IsOk());
+  Check(Options(Expressions::Equal("key", Literal::Double(100))), {0, 1, 2, 3, 4, 5});
+  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<StructType>(
+      std::vector<SchemaField>{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<StructType>(
+      std::vector<SchemaField>{SchemaField::MakeRequired(3, "nested", int32())}));
+  ASSERT_THAT(Write(true, "[[null,0],[null,1],[[10],2],[[11],3],[null,4],[null,5]]"),
+              IsOk());
+  auto is_null = Expressions::IsNull("key.nested");
+  auto not_null = Expressions::NotNull("key.nested");
+  Check(Options(is_null), {0, 1, 4, 5});
+  Check(Options(not_null), {2, 3});
+  Check(Options(Expressions::Not(is_null)), {2, 3});
+  Check(Options(Expressions::Not(not_null)), {0, 1, 4, 5});
+  Check(Options(Expressions::Not(Expressions::LessThan("key.nested", Literal::Int(100)))),
+        {0, 1, 4, 5});
+}
+
+TEST_F(ParquetRowGroupFilterTest, UnsupportedTransformRetainsGroups) {
+  ASSERT_THAT(Write(), IsOk());
+  Check(Options(Expressions::Equal<BoundTransform>(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)))),
+        {2, 3});
+}
+
+TEST_F(ParquetRowGroupFilterTest, PrefixPredicates) {
+  SetKeyType(string());
+  ASSERT_THAT(
+      Write(
+          true,
+          R"([["apple",0],["apricot",1],["banana",2],["blueberry",3],[null,4],["apricot",5]])"),
+      IsOk());
+  Check(Options(Expressions::StartsWith("key", "ap")), {0, 1, 4, 5});
+  Check(Options(Expressions::NotStartsWith("key", "ap")), {2, 3, 4, 5});
+  Check(Options(Expressions::StartsWith("key", "aa")), {});
+  Check(Options(Expressions::StartsWith("key", "z")), {});
+  Check(Options(Expressions::StartsWith("key", "apricots")), {});
+  Check(Options(Expressions::StartsWith("key", "")), {0, 1, 2, 3, 4, 5});
+  Check(Options(Expressions::NotStartsWith("key", "")), {4, 5});
+}
+
+TEST_F(ParquetRowGroupFilterTest, Utf8StringBoundsUseUnsignedByteOrder) {
+  SetKeyType(string());
+  // RG2 holds multi-byte UTF-8 values (U+00E9, U+00FC) whose leading byte 0xC3
+  // sorts after ASCII only under unsigned byte comparison.
+  ASSERT_THAT(
+      Write(true, R"([["a",0],["b",1],["y",2],["z",3],["\u00e9",4],["\u00fc",5]])"),
+      IsOk());
+  ASSERT_EQ(metadata_->num_row_groups(), 3);
+  const std::string e_acute = "\xC3\xA9";
+  const std::string u_umlaut = "\xC3\xBC";
+  Check(Options(Expressions::GreaterThan("key", Literal::String("z"))), {4, 5});
+  Check(Options(Expressions::LessThan("key", Literal::String("c"))), {0, 1});
+  Check(Options(Expressions::Equal("key", Literal::String(e_acute))), {4, 5});
+  Check(Options(Expressions::StartsWith("key", e_acute)), {4, 5});
+  Check(Options(Expressions::StartsWith("key", u_umlaut)), {4, 5});
+}
+
+TEST_F(ParquetRowGroupFilterTest, DecimalAndTemporalStatistics) {
+  struct Case {
+    std::shared_ptr<Type> type;
+    std::string json;
+    Literal value;
+  };
+  for (
+      const auto& test : std::vector<Case>{
+          {.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/predicate_test.cc b/src/iceberg/test/predicate_test.cc
index 13423c664..0c4a43681 100644
--- a/src/iceberg/test/predicate_test.cc
+++ b/src/iceberg/test/predicate_test.cc
@@ -571,14 +571,26 @@ TEST_F(PredicateTest, BoundLiteralPredicateNegate) {
   ASSERT_THAT(negated_result, IsOk());
 
   auto negated = negated_result.value();
-  EXPECT_EQ(negated->op(), Expression::Operation::kNotEq);
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto expected_not_eq,
+      Expressions::NotEqual("age", Literal::Int(25))->Bind(*schema_, true));
+  EXPECT_TRUE(negated->Equals(*expected_not_eq));
 
   // Test less than negation
   auto lt_pred = Expressions::LessThan("age", Literal::Int(30));
   auto bound_lt = lt_pred->Bind(*schema_, true).value();
   auto neg_lt_result = bound_lt->Negate();
   ASSERT_THAT(neg_lt_result, IsOk());
-  EXPECT_EQ(neg_lt_result.value()->op(), Expression::Operation::kGtEq);
+  auto original = std::dynamic_pointer_cast<BoundLiteralPredicate>(bound_lt);
+  ASSERT_NE(original, nullptr);
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto null_guard,
+      BoundUnaryPredicate::Make(Expression::Operation::kIsNull, original->term()));
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto inverse,
+      Expressions::GreaterThanOrEqual("age", Literal::Int(30))->Bind(*schema_, true));
+  auto expected = Expressions::Or(std::move(null_guard), std::move(inverse));
+  EXPECT_TRUE(neg_lt_result.value()->Equals(*expected));
 }
 
 TEST_F(PredicateTest, BoundLiteralPredicateEquals) {
diff --git a/src/iceberg/test/residual_evaluator_test.cc b/src/iceberg/test/residual_evaluator_test.cc
index d2aa2aeb0..6e4ee17f8 100644
--- a/src/iceberg/test/residual_evaluator_test.cc
+++ b/src/iceberg/test/residual_evaluator_test.cc
@@ -35,6 +35,7 @@
 #include "iceberg/test/matchers.h"
 #include "iceberg/transform.h"
 #include "iceberg/type.h"
+#include "iceberg/util/uuid.h"
 
 namespace iceberg {
 
@@ -219,6 +220,81 @@ TEST_F(ResidualEvaluatorTest, CaseInsensitiveIdentityTransformResiduals) {
   EXPECT_EQ(residual4->op(), Expression::Operation::kFalse);
 }
 
+TEST_F(ResidualEvaluatorTest, ComparisonWithNullValue) {
+  auto schema = std::make_shared<Schema>(
+      std::vector<SchemaField>{SchemaField::MakeOptional(50, "key", int32()),
+                               SchemaField::MakeOptional(51, "value", int32())});
+
+  PartitionField pt_field(50, 1000, "key", Transform::Identity());
+  ICEBERG_UNWRAP_OR_FAIL(auto spec_unique,
+                         PartitionSpec::Make(*schema, 0, {pt_field}, false));
+  auto spec = std::shared_ptr<PartitionSpec>(spec_unique.release());
+
+  // NOT(key < 100) OR value = 5
+  auto expr =
+      Expressions::Or(Expressions::Not(Expressions::LessThan("key", Literal::Int(100))),
+                      Expressions::Equal("value", Literal::Int(5)));
+  ICEBERG_UNWRAP_OR_FAIL(auto evaluator,
+                         ResidualEvaluator::Make(expr, *spec, *schema, true));
+
+  // key < 100 is false for null, so its boolean NOT is true for the partition.
+  PartitionValues null_partition(Literal::Null(int32()));
+  ICEBERG_UNWRAP_OR_FAIL(auto null_residual, evaluator->ResidualFor(null_partition));
+  EXPECT_EQ(null_residual->op(), Expression::Operation::kTrue);
+
+  PartitionValues below(Literal::Int(50));
+  ICEBERG_UNWRAP_OR_FAIL(auto below_residual, evaluator->ResidualFor(below));
+  EXPECT_TRUE(below_residual->Equals(*Expressions::Equal("value", Literal::Int(5))));
+
+  PartitionValues above(Literal::Int(150));
+  ICEBERG_UNWRAP_OR_FAIL(auto above_residual, evaluator->ResidualFor(above));
+  EXPECT_EQ(above_residual->op(), Expression::Operation::kTrue);
+
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto unpartitioned,
+      ResidualEvaluator::Make(expr, *PartitionSpec::Unpartitioned(), *schema, true));
+  PartitionValues empty_partition;
+  ICEBERG_UNWRAP_OR_FAIL(auto unpartitioned_residual,
+                         unpartitioned->ResidualFor(empty_partition));
+  EXPECT_TRUE(unpartitioned_residual->Equals(*expr));
+}
+
+TEST_F(ResidualEvaluatorTest, NotUuidComparison) {
+  auto schema = std::make_shared<Schema>(
+      std::vector<SchemaField>{SchemaField::MakeOptional(50, "key", uuid()),
+                               SchemaField::MakeOptional(51, "value", int32())});
+  PartitionField pt_field(50, 1000, "key", Transform::Identity());
+  ICEBERG_UNWRAP_OR_FAIL(auto spec_unique,
+                         PartitionSpec::Make(*schema, 0, {pt_field}, false));
+  auto spec = std::shared_ptr<PartitionSpec>(spec_unique.release());
+
+  ICEBERG_UNWRAP_OR_FAIL(auto lower,
+                         Uuid::FromString("7fffffff-ffff-ffff-ffff-ffffffffffff"));
+  ICEBERG_UNWRAP_OR_FAIL(auto upper,
+                         Uuid::FromString("80000000-0000-0000-0000-000000000000"));
+  auto expr = Expressions::Not(Expressions::LessThan("key", Literal::UUID(lower)));
+  ICEBERG_UNWRAP_OR_FAIL(auto evaluator,
+                         ResidualEvaluator::Make(expr, *spec, *schema, true));
+
+  PartitionValues above(Literal::UUID(upper));
+  ICEBERG_UNWRAP_OR_FAIL(auto above_residual, evaluator->ResidualFor(above));
+  EXPECT_EQ(above_residual->op(), Expression::Operation::kTrue);
+
+  PartitionValues equal(Literal::UUID(lower));
+  ICEBERG_UNWRAP_OR_FAIL(auto equal_residual, evaluator->ResidualFor(equal));
+  EXPECT_EQ(equal_residual->op(), Expression::Operation::kTrue);
+
+  ICEBERG_UNWRAP_OR_FAIL(auto smaller,
+                         Uuid::FromString("00000000-0000-0000-0000-000000000001"));
+  PartitionValues below(Literal::UUID(smaller));
+  ICEBERG_UNWRAP_OR_FAIL(auto below_residual, evaluator->ResidualFor(below));
+  EXPECT_EQ(below_residual->op(), Expression::Operation::kFalse);
+
+  PartitionValues null_partition(Literal::Null(uuid()));
+  ICEBERG_UNWRAP_OR_FAIL(auto null_residual, evaluator->ResidualFor(null_partition));
+  EXPECT_EQ(null_residual->op(), Expression::Operation::kTrue);
+}
+
 TEST_F(ResidualEvaluatorTest, UnpartitionedResiduals) {
   std::vector<std::shared_ptr<Expression>> expressions = {
       Expressions::AlwaysTrue(),