diff --git a/src/iceberg/CMakeLists.txt b/src/iceberg/CMakeLists.txt
index 8a98274ff..edf862b16 100644
--- a/src/iceberg/CMakeLists.txt
+++ b/src/iceberg/CMakeLists.txt
@@ -314,6 +314,7 @@ if(ICEBERG_BUILD_BUNDLE)
       avro/avro_stream_internal.cc
       parquet/parquet_data_util.cc
       parquet/parquet_metrics.cc
+      parquet/parquet_metrics_row_group_filter.cc
       parquet/parquet_reader.cc
       parquet/parquet_register.cc
       parquet/parquet_schema_util.cc
diff --git a/src/iceberg/data/file_scan_task_reader.cc b/src/iceberg/data/file_scan_task_reader.cc
index 0c41f25ed..9c923a215 100644
--- a/src/iceberg/data/file_scan_task_reader.cc
+++ b/src/iceberg/data/file_scan_task_reader.cc
@@ -28,6 +28,7 @@
 #include "iceberg/arrow_c_data_guard_internal.h"
 #include "iceberg/arrow_c_data_util_internal.h"
 #include "iceberg/data/delete_filter.h"
+#include "iceberg/expression/binder.h"
 #include "iceberg/file_reader.h"
 #include "iceberg/manifest/manifest_entry.h"
 #include "iceberg/schema.h"
@@ -164,10 +165,21 @@ class FileScanTaskReader::Impl {
                      "Data file size must not be negative: {}",
                      data_file->file_size_in_bytes);
 
+    auto filter = task.residual_filter();
+    if (filter) {
+      ICEBERG_ASSIGN_OR_RAISE(auto is_bound, IsBoundVisitor::IsBound(filter));
+      if (!is_bound) {
+        ICEBERG_ASSIGN_OR_RAISE(
+            filter,
+            Binder::Bind(*table_schema_, filter,
+                         properties_.Get(ReaderProperties::kFilterCaseSensitive)));
+      }
+    }
+
     if (task.delete_files().empty()) {
       auto options = MakeReaderOptions(
-          *data_file, io_, projected_schema_, task.residual_filter(), name_mapping_,
-          properties_, data_file->first_row_id, data_file->data_sequence_number);
+          *data_file, io_, projected_schema_, filter, name_mapping_, properties_,
+          data_file->first_row_id, data_file->data_sequence_number);
       ICEBERG_ASSIGN_OR_RAISE(
           auto reader, ReaderFactoryRegistry::Open(data_file->file_format, options));
       return MakeArrowArrayStream(std::move(reader));
@@ -191,9 +203,9 @@ class FileScanTaskReader::Impl {
                             ProjectionContext::Make(*required_schema, *projected_schema_,
                                                     project_batch_function));
 
-    auto options = MakeReaderOptions(
-        *data_file, io_, required_schema, task.residual_filter(), name_mapping_,
-        properties_, data_file->first_row_id, data_file->data_sequence_number);
+    auto options = MakeReaderOptions(*data_file, io_, required_schema, filter,
+                                     name_mapping_, properties_, data_file->first_row_id,
+                                     data_file->data_sequence_number);
     ICEBERG_ASSIGN_OR_RAISE(auto reader,
                             ReaderFactoryRegistry::Open(data_file->file_format, options));
 
@@ -207,6 +219,7 @@ class FileScanTaskReader::Impl {
   Impl(Options options, DeleteFilter::FieldLookup field_lookup,
        std::shared_ptr<DeleteCounter> delete_counter)
       : io_(std::move(options.io)),
+        table_schema_(std::move(options.table_schema)),
         schemas_(std::move(options.schemas)),
         projected_schema_(std::move(options.projected_schema)),
         name_mapping_(std::move(options.name_mapping)),
@@ -215,6 +228,7 @@ class FileScanTaskReader::Impl {
         delete_counter_(std::move(delete_counter)) {}
 
   std::shared_ptr<FileIO> io_;
+  std::shared_ptr<Schema> table_schema_;
   std::vector<std::shared_ptr<Schema>> schemas_;
   std::shared_ptr<Schema> projected_schema_;
   std::shared_ptr<NameMapping> name_mapping_;
diff --git a/src/iceberg/file_reader.h b/src/iceberg/file_reader.h
index e08b2df49..efc9b40e1 100644
--- a/src/iceberg/file_reader.h
+++ b/src/iceberg/file_reader.h
@@ -80,6 +80,11 @@ class ICEBERG_EXPORT ReaderProperties : public ConfigBase<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 Case sensitivity when binding unbound filter references.
+  inline static Entry<bool> kFilterCaseSensitive{"read.filter.case-sensitive", true};
   /// \brief Skip GenericDatum in Avro reader for better performance.
   /// When true, decode directly from Avro to Arrow without GenericDatum intermediate.
   /// Default: true (skip GenericDatum for better performance).
diff --git a/src/iceberg/parquet/parquet_metrics.cc b/src/iceberg/parquet/parquet_metrics.cc
index 22710c137..bad4545a7 100644
--- a/src/iceberg/parquet/parquet_metrics.cc
+++ b/src/iceberg/parquet/parquet_metrics.cc
@@ -191,47 +191,6 @@ bool NeedsBoundTruncation(const PrimitiveType& type) {
   return type.type_id() == TypeId::kString || type.type_id() == TypeId::kBinary;
 }
 
-Result<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..89a12ad87
--- /dev/null
+++ b/src/iceberg/parquet/parquet_metrics_row_group_filter.cc
@@ -0,0 +1,324 @@
+/*
+ * 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 <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>& column_indices)
+      : manifest_(manifest), row_group_(row_group), column_indices_(column_indices) {}
+
+  Result<bool> AlwaysTrue() override { return kRowsMightMatch; }
+
+  Result<bool> AlwaysFalse() override { return kRowsCannotMatch; }
+
+  Result<bool> Not(bool) override { 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);
+    if (!lower) {
+      return kRowsMightMatch;
+    }
+    if (std::ranges::all_of(values, [&](const auto& value) { return value < *lower; })) {
+      return kRowsCannotMatch;
+    }
+    const auto upper = MaxValue(ref);
+    if (!upper) {
+      return kRowsMightMatch;
+    }
+    // Like Java, a single candidate must satisfy both bounds.
+    for (const auto& value : values) {
+      if (!(value < *lower || value > *upper)) {
+        return kRowsMightMatch;
+      }
+    }
+    return kRowsCannotMatch;
+  }
+
+  Result<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 || lower->type()->type_id() != TypeId::kString) {
+      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 || upper->type()->type_id() != TypeId::kString) {
+      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 || lower->type()->type_id() != TypeId::kString) {
+      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 || upper->type()->type_id() != TypeId::kString) {
+      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);
+    // GetStatistics excludes repeated columns, so each row contributes one value.
+    return stats && stats->HasNullCount() && stats->null_count() == row_group_.num_rows();
+  }
+
+  bool MayContainNull(const std::shared_ptr<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()) ||
+        MetadataColumns::IsRowLineageColumn(ref->field_id())) {
+      return nullptr;
+    }
+    auto column = column_indices_.find(ref->field_id());
+    // Missing columns can have initial defaults, so do not assume all nulls.
+    if (column == column_indices_.end()) {
+      return nullptr;
+    }
+    const auto& descriptor = *manifest_.descr->Column(column->second);
+    auto field = manifest_.column_index_to_field.find(column->second);
+    // Repeated-column statistics describe elements rather than rows.
+    if (descriptor.max_repetition_level() != 0 ||
+        field == manifest_.column_index_to_field.end() ||
+        !ValidateParquetTypeCompatibility(*ref->type(), *field->second)) {
+      return nullptr;
+    }
+    return row_group_.ColumnChunk(column->second)->statistics();
+  }
+
+  std::optional<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());
+    auto result =
+        ParquetMetrics::StatsValueToLiteral(*stats->descr(), type, *stats, is_min);
+    if (!result || result->IsNaN()) {
+      return std::nullopt;
+    }
+    auto bound = std::move(*result);
+    if (type.type_id() == TypeId::kFloat && std::get<float>(bound.value()) == 0) {
+      return Literal::Float(is_min ? -0.0F : 0.0F);
+    }
+    if (type.type_id() == TypeId::kDouble && std::get<double>(bound.value()) == 0) {
+      return Literal::Double(is_min ? -0.0 : 0.0);
+    }
+    return bound;
+  }
+
+  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;
+    }
+    return compare(*bound, value);
+  }
+
+  const ::parquet::arrow::SchemaManifest& manifest_;
+  const ::parquet::RowGroupMetaData& row_group_;
+  const std::unordered_map<int32_t, int>& column_indices_;
+};
+
+}  // 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->column_indices_.emplace(id, i);
+    }
+  }
+  result->bound_ = True::Instance();
+  if (!filter) {
+    return result;
+  }
+  ICEBERG_ASSIGN_OR_RAISE(result->bound_, RewriteNot::Visit(filter));
+  return result;
+}
+
+Result<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, column_indices_);
+    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..96703a838
--- /dev/null
+++ b/src/iceberg/parquet/parquet_metrics_row_group_filter_internal.h
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#pragma once
+
+#include <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> column_indices_;
+};
+
+}  // namespace iceberg::parquet
diff --git a/src/iceberg/parquet/parquet_reader.cc b/src/iceberg/parquet/parquet_reader.cc
index 1f8146107..fc72de07e 100644
--- a/src/iceberg/parquet/parquet_reader.cc
+++ b/src/iceberg/parquet/parquet_reader.cc
@@ -20,7 +20,7 @@
 #include "iceberg/parquet/parquet_reader.h"
 
 #include <algorithm>
-#include <numeric>
+#include <utility>
 #include <variant>
 #include <vector>
 
@@ -38,7 +38,9 @@
 #include "iceberg/arrow/arrow_io_internal.h"
 #include "iceberg/arrow/arrow_status_internal.h"
 #include "iceberg/arrow/metadata_column_util_internal.h"
+#include "iceberg/expression/binder.h"
 #include "iceberg/parquet/parquet_data_util_internal.h"
+#include "iceberg/parquet/parquet_metrics_row_group_filter_internal.h"
 #include "iceberg/parquet/parquet_register.h"
 #include "iceberg/parquet/parquet_schema_util_internal.h"
 #include "iceberg/result.h"
@@ -65,13 +67,7 @@ Result<SchemaProjection> BuildProjection(::parquet::arrow::FileReader* reader,
     return NotImplemented("Applying name mapping to Parquet schema is not implemented");
   }
 
-  ::parquet::arrow::SchemaManifest schema_manifest;
-  ICEBERG_ARROW_RETURN_NOT_OK(::parquet::arrow::SchemaManifest::Make(
-      metadata->schema(), metadata->key_value_metadata(), reader->properties(),
-      &schema_manifest));
-
-  // Leverage SchemaManifest to project the schema
-  ICEBERG_ASSIGN_OR_RAISE(auto projection, Project(read_schema, schema_manifest));
+  ICEBERG_ASSIGN_OR_RAISE(auto projection, Project(read_schema, reader->manifest()));
   return projection;
 }
 
@@ -262,6 +258,12 @@ std::shared_ptr<::arrow::Schema> AlignOutputSchemaToReaderSchema(
 
 // A stateful context to keep track of the reading progress.
 struct ReadContext {
+  struct SelectedRowGroup {
+    int index;
+    int64_t first_row;
+  };
+  std::vector<SelectedRowGroup> row_groups_;
+  size_t current_row_group_ = 0;
   // The arrow schema to output record batches. It may be different with
   // the schema of record batches returned by `record_batch_reader_`
   // when there is any schema evolution.
@@ -285,6 +287,7 @@ class ParquetReader::Impl {
 
     split_ = options.split;
     read_schema_ = options.projection;
+    stats_filter_.reset();
 
     // Prepare reader properties
     ::parquet::ReaderProperties reader_properties(pool_);
@@ -307,6 +310,23 @@ class ParquetReader::Impl {
 
     // Project read schema onto the Parquet file schema
     ICEBERG_ASSIGN_OR_RAISE(projection_, BuildProjection(reader_.get(), *read_schema_));
+    if (options.filter &&
+        options.properties.Get(ReaderProperties::kParquetRowGroupFilter)) {
+      auto filter = options.filter;
+      if (filter->op() != Expression::Operation::kTrue &&
+          filter->op() != Expression::Operation::kFalse) {
+        ICEBERG_ASSIGN_OR_RAISE(auto is_bound, IsBoundVisitor::IsBound(filter));
+        if (!is_bound) {
+          ICEBERG_ASSIGN_OR_RAISE(
+              filter, Binder::Bind(*options.projection, filter,
+                                   options.properties.Get(
+                                       ReaderProperties::kFilterCaseSensitive)));
+        }
+      }
+      ICEBERG_ASSIGN_OR_RAISE(stats_filter_, ParquetMetricsRowGroupFilter::Make(
+                                                 filter, *reader_->manifest().descr));
+    }
+
     metadata_context_ = {.file_path = options.path,
                          .next_file_pos = 0,
                          .first_row_id = options.first_row_id,
@@ -322,18 +342,29 @@ class ParquetReader::Impl {
     }
 
     ICEBERG_ARROW_ASSIGN_OR_RETURN(auto batch, context_->record_batch_reader_->Next());
-    if (!batch) {
-      return std::nullopt;
+    while (!batch) {
+      const size_t next_group = context_->current_row_group_ + 1;
+      if (next_group >= context_->row_groups_.size()) {
+        return std::nullopt;
+      }
+      ICEBERG_ARROW_RETURN_NOT_OK(context_->record_batch_reader_->Close());
+      const auto& group = context_->row_groups_[next_group];
+      ICEBERG_ARROW_ASSIGN_OR_RETURN(
+          context_->record_batch_reader_,
+          reader_->GetRecordBatchReader({group.index},
+                                        SelectedColumnIndices(projection_)));
+      context_->current_row_group_ = next_group;
+      metadata_context_.next_file_pos = group.first_row;
+      ICEBERG_ARROW_ASSIGN_OR_RETURN(batch, context_->record_batch_reader_->Next());
     }
 
     ICEBERG_ASSIGN_OR_RAISE(
         batch, ProjectRecordBatch(std::move(batch), context_->output_arrow_schema_,
                                   *read_schema_, projection_, metadata_context_, pool_));
 
-    metadata_context_.next_file_pos += batch->num_rows();
-
     ArrowArray arrow_array;
     ICEBERG_ARROW_RETURN_NOT_OK(::arrow::ExportRecordBatch(*batch, &arrow_array));
+    metadata_context_.next_file_pos += batch->num_rows();
     return arrow_array;
   }
 
@@ -389,37 +420,40 @@ class ParquetReader::Impl {
  private:
   Status InitReadContext() {
     context_ = std::make_unique<ReadContext>();
+    auto metadata = reader_->parquet_reader()->metadata();
 
-    // Row group pruning based on the split
-    // TODO(gangwu): add row group filtering based on zone map, bloom filter, etc.
-    std::vector<int> row_group_indices;
-    if (split_.has_value()) {
-      auto metadata = reader_->parquet_reader()->metadata();
-      for (int i = 0; i < metadata->num_row_groups(); ++i) {
-        auto row_group_offset = metadata->RowGroup(i)->file_offset();
-        if (row_group_offset >= split_->offset &&
-            row_group_offset < split_->offset + split_->length) {
-          row_group_indices.push_back(i);
-        } else if (row_group_offset >= split_->offset + split_->length) {
-          break;
-        } else {
-          metadata_context_.next_file_pos += metadata->RowGroup(i)->num_rows();
+    int64_t next_row_start = 0;
+    for (int i = 0; i < metadata->num_row_groups(); ++i) {
+      auto row_group = metadata->RowGroup(i);
+      const int64_t row_start = next_row_start;
+      next_row_start += row_group->num_rows();
+      if (split_.has_value()) {
+        auto row_group_offset = row_group->file_offset();
+        bool in_split = std::cmp_greater_equal(row_group_offset, split_->offset) &&
+                        std::cmp_less(row_group_offset, split_->offset + split_->length);
+        if (!in_split) {
+          continue;
         }
       }
-    } else {
-      row_group_indices.resize(reader_->parquet_reader()->metadata()->num_row_groups());
-      std::iota(row_group_indices.begin(), row_group_indices.end(), 0);  // NOLINT
+      if (stats_filter_) {
+        ICEBERG_ASSIGN_OR_RAISE(
+            auto should_read, stats_filter_->ShouldRead(reader_->manifest(), *row_group));
+        if (!should_read) {
+          continue;
+        }
+      }
+      if (row_group->num_rows() == 0) {
+        continue;
+      }
+      context_->row_groups_.push_back({.index = i, .first_row = row_start});
     }
-
-    // Create the record batch reader
-    if (row_group_indices.empty()) {
-      // None of the row groups are selected, return an empty record batch reader
+    if (context_->row_groups_.empty()) {
       context_->record_batch_reader_ = std::make_unique<EmptyRecordBatchReader>();
     } else {
-      auto column_indices = SelectedColumnIndices(projection_);
       ICEBERG_ARROW_ASSIGN_OR_RETURN(
           context_->record_batch_reader_,
-          reader_->GetRecordBatchReader(row_group_indices, column_indices));
+          reader_->GetRecordBatchReader({context_->row_groups_.front().index},
+                                        SelectedColumnIndices(projection_)));
     }
 
     // Build the output Arrow schema from the projected Iceberg schema. This schema is the
@@ -442,6 +476,9 @@ class ParquetReader::Impl {
         context_->output_arrow_schema_, context_->record_batch_reader_->schema(),
         projection_, use_large_list_);
 
+    if (!context_->row_groups_.empty()) {
+      metadata_context_.next_file_pos = context_->row_groups_.front().first_row;
+    }
     return {};
   }
 
@@ -454,6 +491,7 @@ class ParquetReader::Impl {
   bool use_large_list_ = false;
   // Schema to read from the Parquet file.
   std::shared_ptr<::iceberg::Schema> read_schema_;
+  std::unique_ptr<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 0b6abf218..1e8a5c7ee 100644
--- a/src/iceberg/parquet/parquet_schema_util.cc
+++ b/src/iceberg/parquet/parquet_schema_util.cc
@@ -320,8 +320,6 @@ Status ValidateGeospatialParquetType(const Type& expected_type,
 
 }  // namespace
 
-namespace {
-
 Status ValidateParquetTypeCompatibility(
     const Type& expected_type, const ::parquet::arrow::SchemaField& parquet_field) {
   const auto& arrow_type = parquet_field.field->type();
@@ -477,8 +475,6 @@ Status ValidateParquetTypeCompatibility(
                        expected_type, arrow_type->ToString());
 }
 
-}  // namespace
-
 namespace {
 
 class ProjectionBuilder {
diff --git a/src/iceberg/parquet/parquet_schema_util_internal.h b/src/iceberg/parquet/parquet_schema_util_internal.h
index 2eafd013b..6d9e5a022 100644
--- a/src/iceberg/parquet/parquet_schema_util_internal.h
+++ b/src/iceberg/parquet/parquet_schema_util_internal.h
@@ -38,6 +38,13 @@ struct ParquetExtraAttributes : public FieldProjection::ExtraAttributes {
   std::optional<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/test/CMakeLists.txt b/src/iceberg/test/CMakeLists.txt
index f23bc9181..c6bf5de71 100644
--- a/src/iceberg/test/CMakeLists.txt
+++ b/src/iceberg/test/CMakeLists.txt
@@ -216,6 +216,7 @@ if(ICEBERG_BUILD_BUNDLE)
                    SOURCES
                    metrics_test_base.cc
                    parquet_data_test.cc
+                   parquet_row_group_filter_test.cc
                    parquet_metrics_test.cc
                    parquet_schema_test.cc
                    parquet_test.cc)
diff --git a/src/iceberg/test/file_scan_task_reader_test.cc b/src/iceberg/test/file_scan_task_reader_test.cc
index cca13ce0d..c64aa224e 100644
--- a/src/iceberg/test/file_scan_task_reader_test.cc
+++ b/src/iceberg/test/file_scan_task_reader_test.cc
@@ -43,6 +43,8 @@
 #include "iceberg/data/position_delete_writer.h"
 #include "iceberg/deletes/dv_writer.h"
 #include "iceberg/deletes/position_delete_index.h"
+#include "iceberg/expression/binder.h"
+#include "iceberg/expression/expressions.h"
 #include "iceberg/file_format.h"
 #include "iceberg/file_io.h"
 #include "iceberg/file_reader.h"
@@ -676,4 +678,76 @@ TEST_F(FileScanTaskReaderTest, OpenWithMixedDeletesSkipsFullyDeletedBatches) {
   ASSERT_NO_FATAL_FAILURE(VerifyStream(&stream, R"([[3, "Baz"]])"));
 }
 
+TEST_F(FileScanTaskReaderTest, RowGroupPruningPreservesDeletesAndLineage) {
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto data_file,
+      MakeDataFile(table_schema_,
+                   std::vector<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 = RowLineageProjection(),
+    };
+    ICEBERG_UNWRAP_OR_FAIL(auto reader, FileScanTaskReader::Make(std::move(options)));
+    ICEBERG_UNWRAP_OR_FAIL(auto stream, reader->Open(task));
+    auto batches = ::arrow::ImportRecordBatchReader(&stream).ValueOrDie();
+    std::vector<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, RowGroupPruningUsesUnprojectedFilterWithoutDeletes) {
+  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 = projected_schema_,
+    };
+    options.properties["read.filter.case-sensitive"] = "false";
+    ICEBERG_UNWRAP_OR_FAIL(auto reader, FileScanTaskReader::Make(std::move(options)));
+    ICEBERG_UNWRAP_OR_FAIL(auto stream, reader->Open(task));
+    VerifyStream(&stream, R"([[2,"c"],[3,"d"]])");
+  }
+}
+
 }  // namespace iceberg
diff --git a/src/iceberg/test/parquet_row_group_filter_test.cc b/src/iceberg/test/parquet_row_group_filter_test.cc
new file mode 100644
index 000000000..408b6974e
--- /dev/null
+++ b/src/iceberg/test/parquet_row_group_filter_test.cc
@@ -0,0 +1,656 @@
+/*
+ * 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/c/bridge.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 "iceberg/arrow/arrow_io_internal.h"
+#include "iceberg/arrow/arrow_status_internal.h"
+#include "iceberg/expression/binder.h"
+#include "iceberg/expression/expressions.h"
+#include "iceberg/file_reader.h"
+#include "iceberg/metadata_columns.h"
+#include "iceberg/parquet/parquet_register.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_internal.h"
+#include "iceberg/test/matchers.h"
+#include "iceberg/test/mock_io.h"
+#include "iceberg/type.h"
+
+namespace iceberg::parquet {
+namespace {
+
+class ParquetRowGroupFilterTest : public ::testing::Test {
+ protected:
+  void SetUp() override {
+    parquet::RegisterAll();
+    io_ = std::make_shared<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}));
+
+    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.properties.Set(ReaderProperties::kFilterCaseSensitive, 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, BatchesStayWithinRowGroups) {
+  ASSERT_THAT(Write(), IsOk());
+  auto options = Options(nullptr);
+  options.properties.Set(ReaderProperties::kBatchSize, int64_t{32});
+  for (bool filtered : {false, true}) {
+    SCOPED_TRACE(filtered);
+    if (filtered) {
+      options.filter =
+          Expressions::Or(Expressions::LessThan("key", Literal::Int(2)),
+                          Expressions::GreaterThanOrEqual("key", Literal::Int(20)));
+      ICEBERG_UNWRAP_OR_FAIL(options.filter,
+                             Binder::Bind(*schema_, options.filter, true));
+    }
+    ICEBERG_UNWRAP_OR_FAIL(
+        auto reader, ReaderFactoryRegistry::Open(FileFormatType::kParquet, options));
+    std::vector<int64_t> batch_sizes;
+    while (true) {
+      ICEBERG_UNWRAP_OR_FAIL(auto array, reader->Next());
+      if (!array) {
+        break;
+      }
+      batch_sizes.push_back(array->length);
+      array->release(&*array);
+    }
+    EXPECT_EQ(batch_sizes,
+              filtered ? std::vector<int64_t>({2, 2}) : std::vector<int64_t>({2, 2, 2}));
+    ICEBERG_UNWRAP_OR_FAIL(auto end, reader->Next());
+    EXPECT_FALSE(end.has_value());
+    ASSERT_THAT(reader->Close(), IsOk());
+  }
+}
+
+TEST_F(ParquetRowGroupFilterTest, AllNoneAndResidualRows) {
+  ASSERT_THAT(Write(), IsOk());
+  Check(Options(nullptr), {0, 1, 2, 3, 4, 5});
+  Check(Options(True::Instance()), {0, 1, 2, 3, 4, 5});
+  Check(Options(False::Instance()), {});
+  Check(Options(Expressions::Equal("key", Literal::Int(100))), {});
+  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.properties.Set(ReaderProperties::kFilterCaseSensitive, false);
+  Check(options, {4, 5});
+}
+
+TEST_F(ParquetRowGroupFilterTest, MissingStatisticsRetainGroups) {
+  ASSERT_THAT(Write(false), IsOk());
+  Check(Options(Expressions::Equal("key", Literal::Int(100))), {0, 1, 2, 3, 4, 5});
+  Check(Options(Expressions::IsNull("key")), {0, 1, 2, 3, 4, 5});
+  Check(Options(Expressions::NotNull("key")), {0, 1, 2, 3, 4, 5});
+}
+
+TEST_F(ParquetRowGroupFilterTest, OpenRejectsUnresolvableReferences) {
+  ASSERT_THAT(Write(), IsOk());
+  auto options = Options(nullptr);
+  options.filter = Expressions::Equal("missing", Literal::Int(0));
+  EXPECT_THAT(ReaderFactoryRegistry::Open(FileFormatType::kParquet, options),
+              HasErrorMessage("Cannot find field 'missing'"));
+  options.filter = Expressions::Equal("key", Literal::Int(0));
+  EXPECT_THAT(ReaderFactoryRegistry::Open(FileFormatType::kParquet, options),
+              HasErrorMessage("Cannot find field 'key'"));
+}
+
+TEST_F(ParquetRowGroupFilterTest, PromotedIntegerStatistics) {
+  ASSERT_THAT(Write(), IsOk());
+  SetKeyType(int64());
+  Check(Options(Expressions::Equal("key", Literal::Long(100))), {});
+  Check(Options(Expressions::Equal("key", Literal::Long(10))), {2, 3});
+}
+
+TEST_F(ParquetRowGroupFilterTest, MissingColumnWithDefaultRetainsGroups) {
+  ASSERT_THAT(Write(), IsOk());
+  schema_ = std::make_shared<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(8))), {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);
+    Check(Options(Expressions::Equal("key", test.value)),
+          test.compatible ? std::vector<int64_t>{2, 3}
+                          : std::vector<int64_t>{0, 1, 2, 3, 4, 5});
+  }
+}
+
+TEST_F(ParquetRowGroupFilterTest, NullNegativeAndUnsupportedBooleanBranches) {
+  ASSERT_THAT(Write(true, "[[null,0],[null,1],[10,2],[11,3],[20,4],[21,5]]"), IsOk());
+  Check(Options(Expressions::IsNull("key")), {0, 1});
+  Check(Options(Expressions::NotNull("key")), {2, 3, 4, 5});
+  Check(Options(Expressions::NotEqual("key", Literal::Int(10))), {0, 1, 2, 3, 4, 5});
+  auto unsupported =
+      Expressions::Equal<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, 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());
+  auto offset = metadata_->RowGroup(1)->file_offset();
+  ASSERT_GT(offset, metadata_->RowGroup(0)->file_offset());
+  auto options = Options(Expressions::GreaterThanOrEqual("key", Literal::Int(20)));
+  options.split = Split{.offset = static_cast<size_t>(offset), .length = 100000};
+  Check(options, {4, 5});
+  ICEBERG_UNWRAP_OR_FAIL(
+      options.filter,
+      Binder::Bind(*schema_, Expressions::LessThan("key", Literal::Int(2)), true));
+  Check(options, {});
+  options.filter = nullptr;
+  options.split = Split{.offset = static_cast<size_t>(offset), .length = 0};
+  Check(options, {});
+}
+
+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, FloatingPointStatistics) {
+  SetKeyType(float64());
+  ASSERT_THAT(Write(), IsOk());
+  Check(Options(Expressions::Equal("key", Literal::Double(100))), {});
+  Check(Options(Expressions::Equal("key", Literal::Double(10))), {2, 3});
+  Check(Options(Expressions::IsNaN("key")), {0, 1, 2, 3, 4, 5});
+}
+
+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, 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)))), {});
+}
+
+TEST_F(ParquetRowGroupFilterTest, PrefixPredicates) {
+  SetKeyType(string());
+  ASSERT_THAT(
+      Write(
+          true,
+          R"([["apple",0],["apricot",1],["banana",2],["blueberry",3],[null,4],["apricot",5]])"),
+      IsOk());
+  Check(Options(Expressions::StartsWith("key", "ap")), {0, 1, 4, 5});
+  Check(Options(Expressions::NotStartsWith("key", "ap")), {2, 3, 4, 5});
+  Check(Options(Expressions::StartsWith("key", "aa")), {});
+  Check(Options(Expressions::StartsWith("key", "z")), {});
+  Check(Options(Expressions::StartsWith("key", "apricots")), {});
+  Check(Options(Expressions::StartsWith("key", "")), {0, 1, 2, 3, 4, 5});
+  Check(Options(Expressions::NotStartsWith("key", "")), {4, 5});
+}
+
+TEST_F(ParquetRowGroupFilterTest, DecimalAndTemporalStatistics) {
+  struct Case {
+    std::shared_ptr<Type> 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});
+  Check(Options(Expressions::Equal("key", Literal::Double(0.0))), {0, 1});
+  Check(Options(Expressions::IsNull("key")), {2, 3});
+  Check(Options(Expressions::NotNull("key")), {0, 1, 4, 5});
+  // Like Java, IS NAN excludes all-null groups; NOT NAN retains them.
+  Check(Options(Expressions::IsNaN("key")), {0, 1, 4, 5});
+  Check(Options(Expressions::NotNaN("key")), {0, 1, 2, 3, 4, 5});
+  Check(Options(Expressions::Not(Expressions::IsNaN("key"))), {0, 1, 2, 3, 4, 5});
+  Check(Options(Expressions::Not(Expressions::NotNaN("key"))), {0, 1, 4, 5});
+  Check(Options(Expressions::LessThan("key", Literal::Double(-100))), {});
+}
+
+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});
+  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});
+}
+
+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});
+}
+
+}  // namespace
+}  // namespace iceberg::parquet
diff --git a/src/iceberg/test/parquet_test.cc b/src/iceberg/test/parquet_test.cc
index ec96167d1..f99ec3a1d 100644
--- a/src/iceberg/test/parquet_test.cc
+++ b/src/iceberg/test/parquet_test.cc
@@ -848,8 +848,13 @@ TEST_F(ParquetReaderTest, ReadSplit) {
       {.offset = split_offsets[1] + 1, .length = std::numeric_limits<size_t>::max()},
       {.offset = 0, .length = split_offsets[0]},
   };
-  std::vector<std::string> expected_json = {
-      R"([[1, 0], [2, 1], [3, 2]])", R"([[1, 0], [2, 1]])", R"([[3, 2]])", "", "",
+  // Each batch stays within one row group, including a split covering the whole file.
+  std::vector<std::vector<std::string>> expected_json = {
+      {R"([[1, 0], [2, 1]])", R"([[3, 2]])"},
+      {R"([[1, 0], [2, 1]])"},
+      {R"([[3, 2]])"},
+      {},
+      {},
   };
 
   ReaderProperties reader_properties;
@@ -866,8 +871,8 @@ TEST_F(ParquetReaderTest, ReadSplit) {
                                                      });
     ASSERT_THAT(reader_result, IsOk());
     auto reader = std::move(reader_result.value());
-    if (!expected_json[i].empty()) {
-      ASSERT_NO_FATAL_FAILURE(VerifyNextBatch(*reader, expected_json[i]));
+    for (const auto& batch_json : expected_json[i]) {
+      ASSERT_NO_FATAL_FAILURE(VerifyNextBatch(*reader, batch_json));
     }
     ASSERT_NO_FATAL_FAILURE(VerifyExhausted(*reader));
   }