diff --git a/src/iceberg/arrow_c_data_util.cc b/src/iceberg/arrow_c_data_util.cc
index 915eeabcc..d6eb2303e 100644
--- a/src/iceberg/arrow_c_data_util.cc
+++ b/src/iceberg/arrow_c_data_util.cc
@@ -85,6 +85,7 @@ Result<std::vector<int32_t>> BuildSelectedFieldIndices(
                             FindFieldIndexById(input_fields, output_field.field_id()));
     const auto& input_field = input_fields[input_index];
     if (*input_field.type() != *output_field.type()) {
+      // TODO(gangwu): Support nested pruning when input and output structs differ.
       return InvalidArgument(
           "ProjectBatch only supports complete top-level fields, but field id "
           "{} changes type from {} to {}",
diff --git a/src/iceberg/data/file_scan_task_reader.cc b/src/iceberg/data/file_scan_task_reader.cc
index 8d3c73281..f52a42144 100644
--- a/src/iceberg/data/file_scan_task_reader.cc
+++ b/src/iceberg/data/file_scan_task_reader.cc
@@ -21,14 +21,19 @@
 
 #include <algorithm>
 #include <memory>
+#include <numeric>
 #include <optional>
 #include <utility>
+#include <vector>
 
 #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/expression/evaluator.h"
 #include "iceberg/file_reader.h"
 #include "iceberg/manifest/manifest_entry.h"
+#include "iceberg/row/arrow_array_wrapper.h"
 #include "iceberg/schema.h"
 #include "iceberg/schema_internal.h"
 #include "iceberg/table_scan.h"
@@ -64,21 +69,27 @@ ReaderOptions MakeReaderOptions(const FileScanTask& task, std::shared_ptr<FileIO
   };
 }
 
-class MergeOnReadStreamSource {
+// Streams one task's rows through optional delete filtering, optional residual
+// evaluation, and projection to the caller's schema. Tasks with deletes and
+// tasks whose residual filter is enforced share the same output projection.
+// Delete-only columns and row positions are pruned before returning batches.
+class TaskStreamSource {
  public:
-  MergeOnReadStreamSource(std::unique_ptr<Reader> reader,
-                          std::unique_ptr<DeleteFilter> delete_filter,
-                          std::shared_ptr<::iceberg::Schema> required_schema,
-                          std::shared_ptr<::iceberg::Schema> projected_schema,
-                          ProjectionContext projection_context)
+  TaskStreamSource(std::unique_ptr<Reader> reader,
+                   std::unique_ptr<DeleteFilter> delete_filter,
+                   std::unique_ptr<Evaluator> residual_evaluator,
+                   std::shared_ptr<::iceberg::Schema> required_schema,
+                   std::shared_ptr<::iceberg::Schema> projected_schema,
+                   ProjectionContext projection_context)
       : reader_(std::move(reader)),
         delete_filter_(std::move(delete_filter)),
+        residual_evaluator_(std::move(residual_evaluator)),
         required_schema_(std::move(required_schema)),
         projected_schema_(std::move(projected_schema)),
-        project_all_rows_(required_schema_->SameSchema(*projected_schema_)),
+        project_all_columns_(required_schema_->SameSchema(*projected_schema_)),
         projection_context_(std::move(projection_context)) {}
 
-  ~MergeOnReadStreamSource() {
+  ~TaskStreamSource() {
     if (cached_schema_.has_value() && cached_schema_->release != nullptr) {
       cached_schema_->release(&cached_schema_.value());
     }
@@ -96,7 +107,7 @@ class MergeOnReadStreamSource {
       // File readers expose one stable Arrow schema for every batch in the stream.
       ICEBERG_ASSIGN_OR_RAISE(cached_schema_, reader_->Schema());
     }
-    ArrowSchema& input_arrow_schema = cached_schema_.value();
+    const ArrowSchema& input_arrow_schema = cached_schema_.value();
 
     while (true) {
       ICEBERG_ASSIGN_OR_RAISE(auto next_batch, reader_->Next());
@@ -107,35 +118,103 @@ class MergeOnReadStreamSource {
       ArrowArray input_batch = std::move(next_batch.value());
       internal::ArrowArrayGuard input_batch_guard(&input_batch);
 
-      ICEBERG_ASSIGN_OR_RAISE(
-          auto alive, delete_filter_->ComputeAliveRows(input_arrow_schema, input_batch));
-      if (alive.empty()) {
-        continue;
+      // Rows still selected. While `filtered` is false, every row of the batch
+      // is still a candidate and `selection` has not been materialized.
+      std::vector<int32_t> selection;
+      bool filtered = false;
+
+      if (delete_filter_ != nullptr) {
+        ICEBERG_ASSIGN_OR_RAISE(auto alive, delete_filter_->ComputeAliveRows(
+                                                input_arrow_schema, input_batch));
+        if (alive.empty()) {
+          continue;
+        }
+        filtered = alive.alive_count() != input_batch.length;
+        if (filtered) {
+          selection = std::move(alive.indices);
+        }
+      }
+
+      if (residual_evaluator_ != nullptr) {
+        const int64_t candidate_count =
+            filtered ? static_cast<int64_t>(selection.size()) : input_batch.length;
+        ICEBERG_ASSIGN_OR_RAISE(
+            auto kept, KeepResidualMatches(input_arrow_schema, input_batch, selection,
+                                           /*all_rows_selected=*/!filtered));
+        filtered = filtered || std::cmp_not_equal(kept.size(), candidate_count);
+        if (filtered) {
+          selection = std::move(kept);
+          if (selection.empty()) {
+            continue;
+          }
+        }
       }
 
-      if (alive.alive_count() == input_batch.length && project_all_rows_) {
+      if (!filtered && project_all_columns_) {
         // Transfer ownership to the stream result; the local guard must not release it.
         ArrowArray output_batch = input_batch;
         input_batch.release = nullptr;
         return output_batch;
       }
 
-      return ProjectBatch(&input_batch, alive.indices, projection_context_);
+      if (!filtered) {
+        // Project a fully selected batch, e.g. because columns were pruned or
+        // reordered for evaluation.
+        selection.resize(static_cast<size_t>(input_batch.length));
+        std::iota(selection.begin(), selection.end(), int32_t{0});  // NOLINT
+      }
+      return ProjectBatch(&input_batch, selection, projection_context_);
     }
   }
 
   Result<ArrowSchema> Schema() {
-    ArrowSchema schema;
+    ArrowSchema schema{};
     ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(*projected_schema_, &schema));
     return schema;
   }
 
  private:
+  // Returns the candidate rows that match the residual filter: either the
+  // previously selected rows, or every row when no filter has removed a row yet.
+  Result<std::vector<int32_t>> KeepResidualMatches(const ArrowSchema& input_arrow_schema,
+                                                   const ArrowArray& input_batch,
+                                                   const std::vector<int32_t>& selection,
+                                                   bool all_rows_selected) {
+    ICEBERG_PRECHECK(std::in_range<int32_t>(input_batch.length),
+                     "Batch length {} exceeds int32_t row index capacity",
+                     input_batch.length);
+    // DeleteFilter preserves existing field positions when adding delete keys.
+    ICEBERG_ASSIGN_OR_RAISE(auto row,
+                            ArrowArrayStructLike::Make(input_arrow_schema, input_batch));
+    std::vector<int32_t> kept;
+    auto keep = [&](int64_t index) -> Status {
+      ICEBERG_RETURN_UNEXPECTED(row->Reset(index));
+      ICEBERG_ASSIGN_OR_RAISE(bool matches, residual_evaluator_->Evaluate(*row));
+      if (matches) {
+        kept.push_back(static_cast<int32_t>(index));
+      }
+      return {};
+    };
+    if (all_rows_selected) {
+      kept.reserve(static_cast<size_t>(input_batch.length));
+      for (int64_t index = 0; index < input_batch.length; ++index) {
+        ICEBERG_RETURN_UNEXPECTED(keep(index));
+      }
+    } else {
+      kept.reserve(selection.size());
+      for (int32_t index : selection) {
+        ICEBERG_RETURN_UNEXPECTED(keep(index));
+      }
+    }
+    return kept;
+  }
+
   std::unique_ptr<Reader> reader_;
   std::unique_ptr<DeleteFilter> delete_filter_;
+  std::unique_ptr<Evaluator> residual_evaluator_;
   std::shared_ptr<::iceberg::Schema> required_schema_;
   std::shared_ptr<::iceberg::Schema> projected_schema_;
-  bool project_all_rows_ = false;
+  bool project_all_columns_ = false;
   ProjectionContext projection_context_;
   std::optional<ArrowSchema> cached_schema_;
 };
@@ -189,8 +268,14 @@ class FileScanTaskReader::Impl {
     }
 
     auto filter = task.residual_filter();
+    const bool apply_residual = apply_residuals_ && filter != nullptr &&
+                                filter->op() != Expression::Operation::kTrue;
+    if (apply_residual) {
+      ICEBERG_ASSIGN_OR_RAISE(filter,
+                              Binder::Bind(*projected_schema_, filter, case_sensitive_));
+    }
 
-    if (task.delete_files().empty()) {
+    if (task.delete_files().empty() && !apply_residual) {
       auto options = MakeReaderOptions(
           task, io_, projected_schema_, filter, case_sensitive_, name_mapping_,
           properties_, data_file->first_row_id, data_file->data_sequence_number);
@@ -199,25 +284,21 @@ class FileScanTaskReader::Impl {
       return MakeArrowArrayStream(std::move(reader));
     }
 
-    const bool has_position_deletes =
-        std::any_of(task.delete_files().begin(), task.delete_files().end(),
-                    [](const std::shared_ptr<DataFile>& f) {
-                      return f->content == DataFile::Content::kPositionDeletes;
-                    });
-
-    ICEBERG_ASSIGN_OR_RAISE(
-        auto delete_filter,
-        DeleteFilter::Make(data_file->file_path, task.delete_files(), projected_schema_,
-                           io_, field_lookup_, has_position_deletes, delete_counter_));
+    std::unique_ptr<DeleteFilter> delete_filter;
+    std::shared_ptr<Schema> required_schema = projected_schema_;
+    if (!task.delete_files().empty()) {
+      const bool has_position_deletes =
+          std::any_of(task.delete_files().begin(), task.delete_files().end(),
+                      [](const std::shared_ptr<DataFile>& f) {
+                        return f->content == DataFile::Content::kPositionDeletes;
+                      });
 
-    auto required_schema = delete_filter->RequiredSchema();
-    auto project_batch_function = ProjectionContext::ResolveProjectBatchFunction();
-    // ProjectionContext borrows schemas that are kept in MergeOnReadStreamSource.
-    ICEBERG_ASSIGN_OR_RAISE(
-        auto projection_context,
-        ProjectionContext::Make(*required_schema, *projected_schema_,
-                                project_batch_function,
-                                properties_.Get(ReaderProperties::kArrowValidateBatch)));
+      ICEBERG_ASSIGN_OR_RAISE(
+          delete_filter,
+          DeleteFilter::Make(data_file->file_path, task.delete_files(), projected_schema_,
+                             io_, field_lookup_, has_position_deletes, delete_counter_));
+      required_schema = delete_filter->RequiredSchema();
+    }
 
     auto options = MakeReaderOptions(task, io_, required_schema, filter, case_sensitive_,
                                      name_mapping_, properties_, data_file->first_row_id,
@@ -225,10 +306,24 @@ class FileScanTaskReader::Impl {
     ICEBERG_ASSIGN_OR_RAISE(auto reader,
                             ReaderFactoryRegistry::Open(data_file->file_format, options));
 
-    auto mor_reader = std::make_unique<MergeOnReadStreamSource>(
-        std::move(reader), std::move(delete_filter), std::move(required_schema),
-        projected_schema_, std::move(projection_context));
-    return MakeArrowArrayStream(std::move(mor_reader));
+    // ProjectionContext borrows schemas that are kept in TaskStreamSource.
+    ICEBERG_ASSIGN_OR_RAISE(
+        auto projection_context,
+        ProjectionContext::Make(*required_schema, *projected_schema_,
+                                ProjectionContext::ResolveProjectBatchFunction(),
+                                properties_.Get(ReaderProperties::kArrowValidateBatch)));
+
+    std::unique_ptr<Evaluator> residual_evaluator;
+    if (apply_residual) {
+      ICEBERG_ASSIGN_OR_RAISE(
+          residual_evaluator,
+          Evaluator::Make(*projected_schema_, task.residual_filter(), case_sensitive_));
+    }
+
+    auto task_source = std::make_unique<TaskStreamSource>(
+        std::move(reader), std::move(delete_filter), std::move(residual_evaluator),
+        std::move(required_schema), projected_schema_, std::move(projection_context));
+    return MakeArrowArrayStream(std::move(task_source));
   }
 
  private:
@@ -240,6 +335,7 @@ class FileScanTaskReader::Impl {
         case_sensitive_(options.case_sensitive),
         name_mapping_(std::move(options.name_mapping)),
         properties_(ReaderProperties::FromMap(options.properties)),
+        apply_residuals_(options.apply_residuals),
         field_lookup_(std::move(field_lookup)),
         delete_counter_(std::move(delete_counter)) {}
 
@@ -249,6 +345,7 @@ class FileScanTaskReader::Impl {
   bool case_sensitive_;
   std::shared_ptr<NameMapping> name_mapping_;
   ReaderProperties properties_;
+  bool apply_residuals_ = false;
   DeleteFilter::FieldLookup field_lookup_;
   std::shared_ptr<DeleteCounter> delete_counter_;
 };
diff --git a/src/iceberg/data/file_scan_task_reader.h b/src/iceberg/data/file_scan_task_reader.h
index 591d49c92..671fffedb 100644
--- a/src/iceberg/data/file_scan_task_reader.h
+++ b/src/iceberg/data/file_scan_task_reader.h
@@ -38,10 +38,10 @@ namespace iceberg {
 ///
 /// FileScanTaskReader chooses the copy-on-write path for tasks without deletes and
 /// the merge-on-read path for tasks with v2 position or equality deletes. The returned
-/// stream always exposes `projected_schema`.
+/// stream always exposes `projected_schema`. When `apply_residuals` is enabled, each
+/// task's residual filter is enforced after deletes and before final projection.
 ///
 /// TODO(gangwu): Add a mode that emits a `_deleted` column instead of filtering rows.
-/// TODO(gangwu): Use evaluator to apply residual expression filters.
 class ICEBERG_DATA_EXPORT FileScanTaskReader {
  public:
   /// \brief Options shared by all tasks opened by this reader.
@@ -52,15 +52,25 @@ class ICEBERG_DATA_EXPORT FileScanTaskReader {
     std::shared_ptr<Schema> table_schema;
     /// Optional list of historical table schemas for field lookup.
     std::vector<std::shared_ptr<Schema>> schemas;
-    /// The output schema for the returned ArrowArrayStream. Must be a
-    /// projection of table_schema.
+    /// The read and output projection; this need not be the full table schema.
+    /// Must be a projection of table_schema. With apply_residuals enabled, callers
+    /// must include every field referenced by each task's original residual, even
+    /// if binding would fold its predicate to a constant. Open() fails on missing
+    /// references rather than adding fields from table_schema. Filter-only columns
+    /// remain in the output; delete-only columns are added internally as needed
+    /// and removed before returning batches.
     std::shared_ptr<Schema> projected_schema;
-    /// Case sensitivity for field name resolution. Pass TableScan::is_case_sensitive().
+    /// Case sensitivity for residual binding and reader field name resolution.
+    /// Must match TableScan::is_case_sensitive(); tasks do not carry this setting.
     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.
     std::unordered_map<std::string, std::string> properties;
+    /// Whether to enforce each task's residual filter row by row, after deletes
+    /// and before final projection. Residuals are bound against projected_schema;
+    /// invalid references and unsupported expressions return errors.
+    bool apply_residuals{false};
   };
 
   /// \brief Create a reusable task reader from shared read context.
@@ -70,6 +80,11 @@ class ICEBERG_DATA_EXPORT FileScanTaskReader {
 
   /// \brief Open a task and return an Arrow C stream for its projected live rows.
   ///
+  /// When Options::apply_residuals is set, Options::projected_schema must contain
+  /// all fields referenced by the task's original residual. Missing references
+  /// cause Open() to fail; they are not added from the table schema. Rows are
+  /// filtered after deletes and before final projection.
+  ///
   /// Partial tasks are supported for Parquet, with each row group assigned to the
   /// half-open range containing its midpoint, as in Java Iceberg. Row positions and
   /// inherited row IDs remain relative to the whole file. Other formats require
diff --git a/src/iceberg/expression/evaluator.cc b/src/iceberg/expression/evaluator.cc
index 02438e3f5..417e6712b 100644
--- a/src/iceberg/expression/evaluator.cc
+++ b/src/iceberg/expression/evaluator.cc
@@ -21,6 +21,7 @@
 
 #include "iceberg/expression/binder.h"
 #include "iceberg/expression/expression_visitor.h"
+#include "iceberg/row/struct_like.h"
 #include "iceberg/schema.h"
 #include "iceberg/util/macros.h"
 
@@ -45,6 +46,13 @@ class EvalVisitor : public BoundVisitor<bool> {
   }
 
   Result<bool> IsNull(const std::shared_ptr<Bound>& expr) override {
+    // References can expose validity without converting their value to a literal.
+    // This also supports composite and geospatial fields. Transforms must still
+    // evaluate their result, which may be null even when the input is not.
+    if (auto ref = std::dynamic_pointer_cast<BoundReference>(expr)) {
+      ICEBERG_ASSIGN_OR_RAISE(auto value, ref->accessor().Get(row_));
+      return std::holds_alternative<std::monostate>(value);
+    }
     ICEBERG_ASSIGN_OR_RAISE(auto value, expr->Evaluate(row_));
     return value.IsNull();
   }
diff --git a/src/iceberg/expression/predicate.cc b/src/iceberg/expression/predicate.cc
index f082a3c9a..d8c0e02b7 100644
--- a/src/iceberg/expression/predicate.cc
+++ b/src/iceberg/expression/predicate.cc
@@ -253,6 +253,11 @@ Result<std::shared_ptr<Expression>> UnboundPredicateImpl<B>::Bind(
     return BindUnaryOperation(schema, std::move(bound_term));
   }
 
+  if (!bound_term->type()->is_primitive()) {
+    return InvalidExpression("Cannot bind {} predicate to non-primitive type {}",
+                             BASE::op(), *bound_term->type());
+  }
+
   if (BASE::op() == Expression::Operation::kIn ||
       BASE::op() == Expression::Operation::kNotIn) {
     return BindInOperation(std::move(bound_term));
diff --git a/src/iceberg/test/evaluator_test.cc b/src/iceberg/test/evaluator_test.cc
index e446bc66c..beda8085f 100644
--- a/src/iceberg/test/evaluator_test.cc
+++ b/src/iceberg/test/evaluator_test.cc
@@ -41,6 +41,7 @@
 #include "iceberg/row/arrow_array_wrapper.h"
 #include "iceberg/schema.h"
 #include "iceberg/test/matchers.h"
+#include "iceberg/transform.h"
 #include "iceberg/type.h"
 
 namespace iceberg {
@@ -159,6 +160,56 @@ TEST_F(EvaluatorTest, LessThan) {
       *struct_evaluator, true);
 }
 
+TEST_F(EvaluatorTest, OutOfRangeComparisonsPreserveNulls) {
+  using Op = Expression::Operation;
+  const int64_t above = int64_t{std::numeric_limits<int32_t>::max()} + 1;
+  const int64_t below = int64_t{std::numeric_limits<int32_t>::min()} - 1;
+  const std::string present =
+      R"([{"x": 1, "y": 0.0, "z": 1, "s1": {"s2": {"s3": {"s4": {"i": 1}}}}}])";
+  const std::string nulls = R"([{"x": 1, "y": 0.0, "z": null, "s1": null}])";
+  for (auto op : {Op::kLt, Op::kLtEq, Op::kGt, Op::kGtEq, Op::kNotEq}) {
+    for (const std::string name : {"x", "z", "s1.s2.s3.s4.i"}) {
+      const int64_t bound = op == Op::kGt || op == Op::kGtEq ? below : above;
+      auto predicate = Expressions::Predicate(op, name, Literal::Long(bound));
+      SCOPED_TRACE(predicate->ToString());
+      ICEBERG_UNWRAP_OR_FAIL(auto evaluator, Evaluator::Make(*schema_, predicate));
+      TestData(present, *evaluator, true);
+      // A required leaf under an optional parent can still be null. NotEq keeps
+      // its existing semantics: null is unequal to the out-of-range literal.
+      TestData(nulls, *evaluator, name == "x" || op == Op::kNotEq);
+    }
+  }
+}
+
+TEST_F(EvaluatorTest, NullPredicatesOnStructsRespectParentValidity) {
+  for (const std::string name : {"s1", "s1.s2", "s1.s2.s3.s4"}) {
+    SCOPED_TRACE(name);
+    ICEBERG_UNWRAP_OR_FAIL(auto is_null,
+                           Evaluator::Make(*schema_, Expressions::IsNull(name)));
+    ICEBERG_UNWRAP_OR_FAIL(auto not_null,
+                           Evaluator::Make(*schema_, Expressions::NotNull(name)));
+    const std::string absent = R"([{"x": 1, "y": 0.0, "s1": null}])";
+    const std::string present =
+        R"([{"x": 1, "y": 0.0, "s1": {"s2": {"s3": {"s4": {"i": 1}}}}}])";
+    TestData(absent, *is_null, true);
+    TestData(absent, *not_null, false);
+    TestData(present, *is_null, false);
+    TestData(present, *not_null, true);
+  }
+}
+
+TEST_F(EvaluatorTest, NullPredicatesEvaluateTransformResults) {
+  auto term = Expressions::Transform("x", Transform::Void());
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto is_null, Evaluator::Make(*schema_, Expressions::IsNull<BoundTransform>(term)));
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto not_null,
+      Evaluator::Make(*schema_, Expressions::NotNull<BoundTransform>(term)));
+  // The source is non-null but the transform result is null.
+  TestData(R"([{"x": 1, "y": 0.0}])", *is_null, true);
+  TestData(R"([{"x": 1, "y": 0.0}])", *not_null, false);
+}
+
 TEST_F(EvaluatorTest, LessThanOrEqual) {
   ICEBERG_UNWRAP_OR_FAIL(
       auto evaluator,
diff --git a/src/iceberg/test/file_scan_task_reader_test.cc b/src/iceberg/test/file_scan_task_reader_test.cc
index 596ec488a..69e2dbf2c 100644
--- a/src/iceberg/test/file_scan_task_reader_test.cc
+++ b/src/iceberg/test/file_scan_task_reader_test.cc
@@ -21,6 +21,7 @@
 
 #include <algorithm>
 #include <array>
+#include <limits>
 #include <memory>
 #include <optional>
 #include <string>
@@ -29,6 +30,8 @@
 #include <vector>
 
 #include <arrow/array.h>
+#include <arrow/array/builder_binary.h>
+#include <arrow/array/builder_primitive.h>
 #include <arrow/c/bridge.h>
 #include <arrow/json/from_string.h>
 #include <arrow/record_batch.h>
@@ -44,6 +47,7 @@
 #include "iceberg/arrow_c_data.h"
 #include "iceberg/arrow_c_data_guard_internal.h"
 #include "iceberg/arrow_c_data_util_internal.h"
+#include "iceberg/data/data_writer.h"
 #include "iceberg/data/equality_delete_writer.h"
 #include "iceberg/data/position_delete_writer.h"
 #include "iceberg/data_table_scan.h"
@@ -51,6 +55,7 @@
 #include "iceberg/deletes/position_delete_index.h"
 #include "iceberg/expression/binder.h"
 #include "iceberg/expression/expressions.h"
+#include "iceberg/expression/literal.h"
 #include "iceberg/file_format.h"
 #include "iceberg/file_io.h"
 #include "iceberg/file_reader.h"
@@ -98,6 +103,10 @@ struct ExportedBatch {
   ExportedBatch& operator=(ExportedBatch&& other) noexcept = delete;
 };
 
+// Default rows read by the residual-filter tests.
+constexpr std::string_view kResidualDataJson =
+    R"([[1, "Foo", "blue"], [2, "Bar", "red"], [3, "Baz", "green"]])";
+
 }  // namespace
 
 class FileScanTaskReaderTest : public TempFileTestBase {
@@ -388,6 +397,39 @@ class FileScanTaskReaderTest : public TempFileTestBase {
     ASSERT_EQ(result.ValueOrDie(), nullptr) << "Reader returned an extra batch.";
   }
 
+  // Opens a task with a residual filter and the fixture's schemas.
+  Result<ArrowArrayStream> OpenResidualTask(
+      std::shared_ptr<Expression> residual,
+      std::vector<std::shared_ptr<DataFile>> delete_files = {},
+      bool case_sensitive = true, bool apply_residuals = true,
+      std::string_view data_json = kResidualDataJson) {
+    ICEBERG_ASSIGN_OR_RAISE(auto data_file,
+                            MakeDataFile(table_schema_, std::string(data_json)));
+    FileScanTask task(data_file, std::move(delete_files), std::move(residual));
+    FileScanTaskReader::Options options{
+        .io = file_io_,
+        .table_schema = table_schema_,
+        .schemas = {table_schema_},
+        .projected_schema = projected_schema_,
+        .case_sensitive = case_sensitive,
+        .apply_residuals = apply_residuals,
+    };
+    ICEBERG_ASSIGN_OR_RAISE(auto reader, FileScanTaskReader::Make(std::move(options)));
+    return reader->Open(task);
+  }
+
+  // Opens a task with `residual` enforced and verifies the streamed rows.
+  void VerifyResidualStream(std::shared_ptr<Expression> residual,
+                            std::string_view expected_json,
+                            std::vector<std::shared_ptr<DataFile>> delete_files = {},
+                            bool case_sensitive = true,
+                            std::string_view data_json = kResidualDataJson) {
+    ICEBERG_UNWRAP_OR_FAIL(
+        auto stream, OpenResidualTask(std::move(residual), delete_files, case_sensitive,
+                                      /*apply_residuals=*/true, data_json));
+    ASSERT_NO_FATAL_FAILURE(VerifyStream(&stream, expected_json));
+  }
+
   void VerifyDataReaderBatchLengths(const DataFile& data_file,
                                     const std::vector<int64_t>& expected_lengths,
                                     ReaderProperties properties = {}) {
@@ -647,6 +689,436 @@ TEST_F(FileScanTaskReaderTest, OpenWithPositionDeletesFiltersRowsAndPrunesPos) {
   ASSERT_NO_FATAL_FAILURE(VerifyStream(&stream, R"([[1, "Foo"], [3, "Baz"]])"));
 }
 
+TEST_F(FileScanTaskReaderTest, OpenAppliesResidualFilterWithoutDeletes) {
+  ASSERT_NO_FATAL_FAILURE(VerifyResidualStream(
+      Expressions::GreaterThan("id", Literal::Int(1)), R"([[2, "Bar"], [3, "Baz"]])"));
+}
+
+TEST_F(FileScanTaskReaderTest, OpenAppliesFoldedResidualsOnProjectedRequiredColumn) {
+  table_schema_ = std::make_shared<Schema>(
+      std::vector<SchemaField>{SchemaField::MakeRequired(1, "flag", boolean()),
+                               SchemaField::MakeRequired(2, "id", int32())});
+  projected_schema_ = std::make_shared<Schema>(
+      std::vector<SchemaField>{table_schema_->fields()[1], table_schema_->fields()[0]});
+  constexpr std::string_view data = "[[true, 1], [false, 2], [true, 3]]";
+
+  ASSERT_NO_FATAL_FAILURE(VerifyResidualStream(Expressions::NotNull("flag"),
+                                               "[[1, true], [2, false], [3, true]]", {},
+                                               true, data));
+  // The projection reorders the fields; folding must preserve the read positions.
+  ASSERT_NO_FATAL_FAILURE(VerifyResidualStream(
+      Expressions::And(Expressions::NotNull("flag"),
+                       Expressions::GreaterThan("id", Literal::Int(1))),
+      "[[2, false], [3, true]]", {}, true, data));
+  ASSERT_NO_FATAL_FAILURE(VerifyResidualStream(
+      Expressions::Or(Expressions::IsNull("flag"),
+                      Expressions::GreaterThan("id", Literal::Int(1))),
+      "[[2, false], [3, true]]", {}, true, data));
+
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto stream, OpenResidualTask(Expressions::IsNull("flag"), {}, true, true, data));
+  auto batches = ::arrow::ImportRecordBatchReader(&stream).ValueOrDie();
+  auto next = batches->Next();
+  ASSERT_TRUE(next.ok()) << next.status();
+  EXPECT_EQ(next.ValueOrDie(), nullptr);
+}
+
+TEST_F(FileScanTaskReaderTest, OpenAppliesNullableOutOfRangeResiduals) {
+  table_schema_ = std::make_shared<Schema>(
+      std::vector<SchemaField>{SchemaField::MakeRequired(1, "id", int32()),
+                               SchemaField::MakeOptional(2, "n", int32())});
+  projected_schema_ = table_schema_;
+  const int64_t above = int64_t{std::numeric_limits<int32_t>::max()} + 1;
+  const int64_t below = int64_t{std::numeric_limits<int32_t>::min()} - 1;
+  for (auto residual : {Expressions::LessThan("n", Literal::Long(above)),
+                        Expressions::LessThanOrEqual("n", Literal::Long(above)),
+                        Expressions::GreaterThan("n", Literal::Long(below)),
+                        Expressions::GreaterThanOrEqual("n", Literal::Long(below))}) {
+    SCOPED_TRACE(residual->ToString());
+    VerifyResidualStream(residual, "[[1, 1]]", {}, true, "[[1, 1], [2, null]]");
+  }
+}
+
+TEST_F(FileScanTaskReaderTest, OpenAppliesResidualFilterWithPositionDeletes) {
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto data_file,
+      MakeDataFile(table_schema_,
+                   R"([[1, "Foo", "blue"], [2, "Bar", "red"], [3, "Baz", "green"]])"));
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto pos_delete, MakePositionDeleteFile(CreateNewTempFilePathWithSuffix(".parquet"),
+                                              {1}, data_file->file_path));
+  // The position delete removes "Bar" and the residual removes "Baz".
+  FileScanTask task(data_file, {pos_delete},
+                    Expressions::LessThan("id", Literal::Int(3)));
+
+  FileScanTaskReader::Options options{
+      .io = file_io_,
+      .table_schema = table_schema_,
+      .schemas = {table_schema_},
+      .projected_schema = projected_schema_,
+      .apply_residuals = true,
+  };
+  ICEBERG_UNWRAP_OR_FAIL(auto reader, FileScanTaskReader::Make(std::move(options)));
+  auto stream_result = reader->Open(task);
+  ASSERT_THAT(stream_result, IsOk());
+  auto stream = std::move(stream_result.value());
+
+  ASSERT_NO_FATAL_FAILURE(VerifyStream(&stream, R"([[1, "Foo"]])"));
+}
+
+TEST_F(FileScanTaskReaderTest, OpenAppliesResidualFilterWithEqualityDeletes) {
+  projected_schema_ = std::make_shared<Schema>(
+      std::vector<SchemaField>{table_schema_->fields()[0], table_schema_->fields()[2]});
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto eq_delete,
+      MakeEqualityDeleteFile(CreateNewTempFilePathWithSuffix(".parquet"), table_schema_,
+                             R"([[0, "Bar", "unused"]])", {2}));
+
+  // The residual removes "Foo"; the equality delete removes "Bar".
+  ASSERT_NO_FATAL_FAILURE(
+      VerifyResidualStream(Expressions::NotEqual("category", Literal::String("blue")),
+                           R"([[3, "green"]])", {eq_delete}));
+}
+
+TEST_F(FileScanTaskReaderTest, OpenRejectsResidualOnColumnAbsentFromProjection) {
+  EXPECT_THAT(OpenResidualTask(Expressions::Equal("category", Literal::String("blue"))),
+              HasErrorMessage("Cannot find field 'category'"));
+  // Missing references are rejected even if the table schema would fold them away.
+  projected_schema_ =
+      std::make_shared<Schema>(std::vector<SchemaField>{table_schema_->fields()[1]});
+  EXPECT_THAT(OpenResidualTask(Expressions::NotNull("id")),
+              HasErrorMessage("Cannot find field 'id'"));
+}
+
+TEST_F(FileScanTaskReaderTest, ScanProjectionKeepsFilterColumnsInOutput) {
+  internal::TableScanContext context;
+  context.selected_columns = std::vector<std::string>{"id"};
+  context.filter = Expressions::Equal("category", Literal::String("blue"));
+  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());
+  ASSERT_EQ(projection->fields().size(), 2);
+  EXPECT_EQ(projection->fields()[0].field_id(), 1);
+  EXPECT_EQ(projection->fields()[1].field_id(), 3);
+
+  ICEBERG_UNWRAP_OR_FAIL(auto data_file,
+                         MakeDataFile(table_schema_, std::string(kResidualDataJson)));
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto pos_delete, MakePositionDeleteFile(CreateNewTempFilePathWithSuffix(".parquet"),
+                                              {1}, data_file->file_path));
+  for (bool with_deletes : {false, true}) {
+    SCOPED_TRACE(with_deletes);
+    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 = scan->io(),
+                                            .table_schema = table_schema_,
+                                            .projected_schema = projection,
+                                            .case_sensitive = scan->is_case_sensitive(),
+                                            .apply_residuals = true,
+                                        }));
+    ICEBERG_UNWRAP_OR_FAIL(auto stream, reader->Open(task));
+    auto batches = ::arrow::ImportRecordBatchReader(&stream).ValueOrDie();
+    ICEBERG_UNWRAP_OR_FAIL(auto expected_schema, MakeArrowSchema(*projection));
+    ASSERT_TRUE(batches->schema()->Equals(*expected_schema));
+    auto batch = batches->Next().ValueOrDie();
+    ASSERT_NE(batch, nullptr);
+    ICEBERG_UNWRAP_OR_FAIL(auto expected,
+                           MakeRecordBatch(expected_schema, R"([[1, "blue"]])"));
+    EXPECT_TRUE(batch->Equals(*expected)) << batch->ToString();
+    EXPECT_EQ(batches->Next().ValueOrDie(), nullptr);
+  }
+}
+
+TEST_F(FileScanTaskReaderTest, OpenAppliesCaseInsensitiveResidualFilter) {
+  ASSERT_NO_FATAL_FAILURE(
+      VerifyResidualStream(Expressions::GreaterThan("ID", Literal::Int(1)),
+                           R"([[2, "Bar"], [3, "Baz"]])", {}, /*case_sensitive=*/false));
+}
+
+TEST_F(FileScanTaskReaderTest, OpenRejectsCaseSensitiveResidualWithDifferentCase) {
+  EXPECT_THAT(OpenResidualTask(Expressions::GreaterThan("ID", Literal::Int(1))),
+              HasErrorMessage("Cannot find field 'ID' in struct"));
+}
+
+TEST_F(FileScanTaskReaderTest, OpenAppliesResidualFilterOnNullValues) {
+  // Null values never satisfy a comparison residual.
+  ASSERT_NO_FATAL_FAILURE(VerifyResidualStream(
+      Expressions::NotNull("name"), R"([[1, "Foo"], [3, "Baz"]])", {}, true,
+      R"([[1, "Foo", "blue"], [2, null, "red"], [3, "Baz", "green"]])"));
+}
+
+TEST_F(FileScanTaskReaderTest, OpenAppliesResidualMatchingAllRows) {
+  ASSERT_NO_FATAL_FAILURE(
+      VerifyResidualStream(Expressions::GreaterThanOrEqual("id", Literal::Int(1)),
+                           R"([[1, "Foo"], [2, "Bar"], [3, "Baz"]])"));
+}
+
+TEST_F(FileScanTaskReaderTest, OpenAppliesAlwaysFalseResidualReturningNoRows) {
+  ICEBERG_UNWRAP_OR_FAIL(auto stream, OpenResidualTask(Expressions::AlwaysFalse()));
+
+  auto record_batch_reader = ::arrow::ImportRecordBatchReader(&stream).ValueOrDie();
+  auto result = record_batch_reader->Next();
+  ASSERT_TRUE(result.ok()) << result.status().message();
+  ASSERT_EQ(result.ValueOrDie(), nullptr);
+}
+
+TEST_F(FileScanTaskReaderTest, OpenAppliesResidualFilterKeepingRowIdentities) {
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto data_file,
+      MakeDataFile(table_schema_,
+                   R"([[1, "Foo", "blue"], [2, "Bar", "red"], [3, "Baz", "green"]])"));
+  data_file->first_row_id = 100L;
+  data_file->data_sequence_number = 5L;
+  FileScanTask task(data_file, {}, Expressions::GreaterThan("id", Literal::Int(1)));
+
+  FileScanTaskReader::Options options{
+      .io = file_io_,
+      .table_schema = table_schema_,
+      .schemas = {table_schema_},
+      .projected_schema = RowLineageProjection(),
+      .apply_residuals = true,
+  };
+  ICEBERG_UNWRAP_OR_FAIL(auto reader, FileScanTaskReader::Make(std::move(options)));
+  auto stream_result = reader->Open(task);
+  ASSERT_THAT(stream_result, IsOk());
+  auto stream = std::move(stream_result.value());
+
+  // Row identities of the surviving rows only: dropped rows must not shift ids.
+  ASSERT_NO_FATAL_FAILURE(VerifyStream(&stream, R"([[2, 101, 5],
+                                                   [3, 102, 5]])"));
+}
+
+TEST_F(FileScanTaskReaderTest, OpenWithoutResidualEnforcementKeepsAllRows) {
+  // Residual enforcement is off; rows are returned unfiltered.
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto stream, OpenResidualTask(Expressions::GreaterThan("id", Literal::Int(1)), {},
+                                    /*case_sensitive=*/true, /*apply_residuals=*/false));
+
+  ASSERT_NO_FATAL_FAILURE(
+      VerifyStream(&stream, R"([[1, "Foo"], [2, "Bar"], [3, "Baz"]])"));
+}
+
+TEST_F(FileScanTaskReaderTest, OpenRejectsResidualWithUnknownColumn) {
+  EXPECT_THAT(OpenResidualTask(Expressions::Equal("missing", Literal::Int(1))),
+              HasErrorMessage("Cannot find field 'missing'"));
+}
+
+TEST_F(FileScanTaskReaderTest, OpenAppliesResidualOnNestedColumn) {
+  table_schema_ = std::make_shared<Schema>(std::vector<SchemaField>{
+      SchemaField::MakeRequired(1, "id", int32()),
+      SchemaField::MakeOptional(2, "s",
+                                std::make_shared<StructType>(std::vector<SchemaField>{
+                                    SchemaField::MakeOptional(3, "tags", string())}))});
+  projected_schema_ = table_schema_;
+  constexpr std::string_view data =
+      R"([[1, {"tags": "a"}], [2, null], [3, {"tags": null}]])";
+  VerifyResidualStream(Expressions::NotNull("s.tags"), R"([[1, {"tags": "a"}]])", {},
+                       true, data);
+  VerifyResidualStream(Expressions::IsNull("s.tags"),
+                       R"([[2, null], [3, {"tags": null}]])", {}, true, data);
+}
+
+TEST_F(FileScanTaskReaderTest, NestedResidualsPreserveReadPositionsWithDeletes) {
+  auto value = SchemaField::MakeOptional(3, "value", string());
+  auto key = SchemaField::MakeRequired(4, "key", int32());
+  table_schema_ = std::make_shared<Schema>(std::vector<SchemaField>{
+      SchemaField::MakeRequired(1, "id", int32()),
+      SchemaField::MakeOptional(2, "record", struct_({value, key}))});
+  // Reorder both the top-level fields and the nested children in the read schema.
+  projected_schema_ = std::make_shared<Schema>(
+      std::vector<SchemaField>{table_schema_->fields()[1].WithType(struct_({key, value})),
+                               table_schema_->fields()[0]});
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto data_file,
+      MakeDataFile(table_schema_,
+                   R"([[1, {"value": "a", "key": 1}], [2, {"value": "b", "key": 2}],
+                       [3, null], [4, {"value": "d", "key": 3}],
+                       [5, {"value": "e", "key": 4}]])"));
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto pos_delete, MakePositionDeleteFile(CreateNewTempFilePathWithSuffix(".parquet"),
+                                              {1}, data_file->file_path));
+  auto delete_schema =
+      std::make_shared<Schema>(std::vector<SchemaField>{table_schema_->fields()[0]});
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto eq_delete, MakeEqualityDeleteFile(CreateNewTempFilePathWithSuffix(".parquet"),
+                                             delete_schema, "[[4]]", {1}));
+  for (bool case_sensitive : {true, false}) {
+    for (const std::string deletes : {"none", "position", "equality", "mixed"}) {
+      SCOPED_TRACE(case_sensitive);
+      SCOPED_TRACE(deletes);
+      std::vector<std::shared_ptr<DataFile>> delete_files;
+      if (deletes == "position" || deletes == "mixed") {
+        delete_files.push_back(pos_delete);
+      }
+      if (deletes == "equality" || deletes == "mixed") {
+        delete_files.push_back(eq_delete);
+      }
+      FileScanTask task(
+          data_file, std::move(delete_files),
+          Expressions::GreaterThan(case_sensitive ? "record.key" : "RECORD.KEY",
+                                   Literal::Int(1)));
+      ICEBERG_UNWRAP_OR_FAIL(auto reader, FileScanTaskReader::Make({
+                                              .io = file_io_,
+                                              .table_schema = table_schema_,
+                                              .projected_schema = projected_schema_,
+                                              .case_sensitive = case_sensitive,
+                                              .apply_residuals = true,
+                                          }));
+      ICEBERG_UNWRAP_OR_FAIL(auto stream, reader->Open(task));
+      const std::string_view expected =
+          deletes == "none"
+              ? R"([[{"key": 2, "value": "b"}, 2], [{"key": 3, "value": "d"}, 4],
+                     [{"key": 4, "value": "e"}, 5]])"
+          : deletes == "position"
+              ? R"([[{"key": 3, "value": "d"}, 4], [{"key": 4, "value": "e"}, 5]])"
+          : deletes == "equality"
+              ? R"([[{"key": 2, "value": "b"}, 2], [{"key": 4, "value": "e"}, 5]])"
+              : R"([[{"key": 4, "value": "e"}, 5]])";
+      VerifyStream(&stream, expected);
+    }
+  }
+}
+
+TEST_F(FileScanTaskReaderTest, OpenRejectsResidualOnListElement) {
+  table_schema_ = std::make_shared<Schema>(std::vector<SchemaField>{
+      SchemaField::MakeRequired(1, "id", int32()),
+      SchemaField::MakeOptional(
+          2, "items",
+          list(SchemaField::MakeOptional(
+              3, "element",
+              struct_({SchemaField::MakeOptional(4, "value", int32())}))))});
+  projected_schema_ = table_schema_;
+  EXPECT_THAT(
+      OpenResidualTask(Expressions::GreaterThan("items.element.value", Literal::Int(0)),
+                       {}, true, true, R"([[1, [{"value": 1}]]])"),
+      IsError(ErrorKind::kNotFound));
+}
+
+TEST_F(FileScanTaskReaderTest, OpenRejectsResidualComparisonOnCompositeColumn) {
+  table_schema_ = std::make_shared<Schema>(std::vector<SchemaField>{
+      SchemaField::MakeRequired(1, "id", int32()),
+      SchemaField::MakeOptional(
+          2, "s", struct_({SchemaField::MakeOptional(3, "value", int32())}))});
+  projected_schema_ = table_schema_;
+  EXPECT_THAT(OpenResidualTask(Expressions::Equal("s", Literal::Int(1)), {}, true, true,
+                               R"([[1, {"value": 1}]])"),
+              IsError(ErrorKind::kInvalidExpression));
+}
+
+TEST_F(FileScanTaskReaderTest, OpenAppliesNullPredicatesOnCompositeColumns) {
+  struct Case {
+    std::shared_ptr<Type> type;
+    std::string data;
+    std::string non_null_rows;
+    std::string matching_row;
+  };
+  const std::vector<Case> cases = {
+      {.type = struct_({SchemaField::MakeOptional(3, "value", int32())}),
+       .data = R"([[1, null], [2, {"value": null}], [3, {"value": 10}], [4, null]])",
+       .non_null_rows = R"([[2, {"value": null}], [3, {"value": 10}]])",
+       .matching_row = R"([[3, {"value": 10}]])"},
+      {.type = list(SchemaField::MakeOptional(3, "element", int32())),
+       .data = "[[1, null], [2, []], [3, [10]], [4, null]]",
+       .non_null_rows = "[[2, []], [3, [10]]]",
+       .matching_row = "[[3, [10]]]"},
+      {.type = map(SchemaField::MakeRequired(3, "key", string()),
+                   SchemaField::MakeOptional(4, "value", int32())),
+       .data = R"([[1, null], [2, []], [3, [["a", 10]]], [4, null]])",
+       .non_null_rows = R"([[2, []], [3, [["a", 10]]]])",
+       .matching_row = R"([[3, [["a", 10]]]])"},
+  };
+  for (const auto& [type, data, non_null_rows, matching_row] : cases) {
+    SCOPED_TRACE(type->ToString());
+    table_schema_ = std::make_shared<Schema>(
+        std::vector<SchemaField>{SchemaField::MakeRequired(1, "id", int32()),
+                                 SchemaField::MakeOptional(2, "value", type)});
+    projected_schema_ = table_schema_;
+    VerifyResidualStream(Expressions::IsNull("value"), "[[1, null], [4, null]]", {}, true,
+                         data);
+    VerifyResidualStream(Expressions::NotNull("value"), non_null_rows, {}, true, data);
+    VerifyResidualStream(
+        Expressions::And(Expressions::NotNull("value"),
+                         Expressions::GreaterThan("id", Literal::Int(2))),
+        matching_row, {}, true, data);
+  }
+}
+
+TEST_F(FileScanTaskReaderTest, OpenAppliesGeospatialNullPredicates) {
+  // Valid little-endian WKB POINT(1 2).
+  const std::array<uint8_t, 21> point = {0x01, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00,
+                                         0x00, 0x00, 0x00, 0x00, 0xf0, 0x3f, 0x00,
+                                         0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x40};
+  for (auto type : std::vector<std::shared_ptr<PrimitiveType>>{geometry(), geography()}) {
+    SCOPED_TRACE(type->ToString());
+    table_schema_ = std::make_shared<Schema>(
+        std::vector<SchemaField>{SchemaField::MakeRequired(1, "id", int32()),
+                                 SchemaField::MakeOptional(2, "geom", type)});
+    projected_schema_ = table_schema_;
+    ::arrow::Int32Builder ids;
+    ASSERT_TRUE(ids.AppendValues({1, 2}).ok());
+    ::arrow::BinaryBuilder values;
+    ASSERT_TRUE(values.Append(point.data(), point.size()).ok());
+    ASSERT_TRUE(values.AppendNull().ok());
+    auto array = ::arrow::StructArray::Make(
+                     {ids.Finish().ValueOrDie(), values.Finish().ValueOrDie()},
+                     {::arrow::field("id", ::arrow::int32(), false),
+                      ::arrow::field("geom", ::arrow::binary())})
+                     .ValueOrDie();
+    ExportedBatch batch;
+    ASSERT_TRUE(::arrow::ExportArray(*array, &batch.array).ok());
+    ICEBERG_UNWRAP_OR_FAIL(auto writer,
+                           DataWriter::Make({
+                               .path = CreateNewTempFilePathWithSuffix(".parquet"),
+                               .schema = table_schema_,
+                               .spec = partition_spec_,
+                               .partition = PartitionValues{},
+                               .format = FileFormatType::kParquet,
+                               .io = file_io_,
+                           }));
+    ASSERT_THAT(writer->Write(&batch.array), IsOk());
+    ASSERT_THAT(writer->Close(), IsOk());
+    ICEBERG_UNWRAP_OR_FAIL(auto metadata, writer->Metadata());
+    for (bool is_null : {false, true}) {
+      FileScanTask task(
+          metadata.data_files[0], {},
+          is_null ? Expressions::IsNull("geom") : Expressions::NotNull("geom"));
+      ICEBERG_UNWRAP_OR_FAIL(auto reader, FileScanTaskReader::Make({
+                                              .io = file_io_,
+                                              .table_schema = table_schema_,
+                                              .projected_schema = projected_schema_,
+                                              .apply_residuals = true,
+                                          }));
+      ICEBERG_UNWRAP_OR_FAIL(auto stream, reader->Open(task));
+      auto batches = ::arrow::ImportRecordBatchReader(&stream).ValueOrDie();
+      ICEBERG_UNWRAP_OR_FAIL(auto expected_schema, MakeArrowSchema(*projected_schema_));
+      ASSERT_TRUE(batches->schema()->Equals(*expected_schema));
+      auto result = batches->Next().ValueOrDie();
+      ASSERT_NE(result, nullptr);
+      ASSERT_EQ(result->num_rows(), 1);
+      ASSERT_EQ(result->num_columns(), 2);
+      auto result_ids = std::static_pointer_cast<::arrow::Int32Array>(result->column(0));
+      EXPECT_EQ(result_ids->Value(0), is_null ? 2 : 1);
+      auto result_values =
+          std::static_pointer_cast<::arrow::BinaryArray>(result->column(1));
+      if (is_null) {
+        EXPECT_TRUE(result_values->IsNull(0));
+      } else {
+        EXPECT_FALSE(result_values->IsNull(0));
+        EXPECT_EQ(
+            result_values->GetView(0),
+            std::string_view(reinterpret_cast<const char*>(point.data()), point.size()));
+      }
+      EXPECT_EQ(batches->Next().ValueOrDie(), nullptr);
+    }
+  }
+}
+
 TEST_F(FileScanTaskReaderTest, OpenWithDeletionVectorFiltersRows) {
   ICEBERG_UNWRAP_OR_FAIL(
       auto data_file,
@@ -873,6 +1345,50 @@ TEST_F(FileScanTaskReaderTest, OpenWithSchemasResolvesDroppedEqualityField) {
   ASSERT_NO_FATAL_FAILURE(VerifyStream(&stream, R"([[1, "Foo"], [3, "Baz"]])"));
 }
 
+TEST_F(FileScanTaskReaderTest, ResidualsWithReusedEqualityDeleteNames) {
+  for (const std::string old_name : {"name", "NAME"}) {
+    auto current_schema = std::make_shared<Schema>(
+        std::vector<SchemaField>{SchemaField::MakeRequired(1, "id", int32()),
+                                 SchemaField::MakeOptional(4, "name", string())},
+        2);
+    auto old_schema = std::make_shared<Schema>(
+        std::vector<SchemaField>{SchemaField::MakeRequired(1, "id", int32()),
+                                 SchemaField::MakeOptional(2, old_name, string())},
+        1);
+    // Reorder output columns so residual field positions must be mapped by id.
+    auto projection = std::make_shared<Schema>(std::vector<SchemaField>{
+        current_schema->fields()[1], current_schema->fields()[0]});
+    ICEBERG_UNWRAP_OR_FAIL(
+        auto data_file,
+        MakeDataFile(old_schema, R"([[1, "keep"], [2, "gone"], [3, "keep"]])"));
+    ICEBERG_UNWRAP_OR_FAIL(
+        auto deletion, MakeEqualityDeleteFile(CreateNewTempFilePathWithSuffix(".parquet"),
+                                              old_schema, R"([[0, "gone"]])", {2}));
+    for (bool case_sensitive : {false, true}) {
+      auto is_null = Expressions::IsNull(case_sensitive ? "name" : "NAME");
+      auto greater = Expressions::GreaterThan("id", Literal::Int(1));
+      for (const auto& [residual, expected] :
+           std::vector<std::pair<std::shared_ptr<Expression>, std::string>>{
+               {is_null, "[[null, 1], [null, 3]]"},
+               {greater, "[[null, 3]]"},
+               {Expressions::And(is_null, greater), "[[null, 3]]"}}) {
+        SCOPED_TRACE(old_name + ": " + residual->ToString());
+        ICEBERG_UNWRAP_OR_FAIL(auto reader, FileScanTaskReader::Make({
+                                                .io = file_io_,
+                                                .table_schema = current_schema,
+                                                .schemas = {current_schema, old_schema},
+                                                .projected_schema = projection,
+                                                .case_sensitive = case_sensitive,
+                                                .apply_residuals = true,
+                                            }));
+        FileScanTask task(data_file, {deletion}, residual);
+        ICEBERG_UNWRAP_OR_FAIL(auto stream, reader->Open(task));
+        VerifyStream(&stream, expected);
+      }
+    }
+  }
+}
+
 TEST_F(FileScanTaskReaderTest, OpenWithMixedDeletesSkipsFullyDeletedBatches) {
   ICEBERG_UNWRAP_OR_FAIL(
       auto data_file,
diff --git a/src/iceberg/test/predicate_test.cc b/src/iceberg/test/predicate_test.cc
index 0c4a43681..9332842ba 100644
--- a/src/iceberg/test/predicate_test.cc
+++ b/src/iceberg/test/predicate_test.cc
@@ -503,6 +503,26 @@ TEST_F(PredicateTest, LiteralConversionEdgeCases) {
   EXPECT_EQ(bound_large->op(), Expression::Operation::kTrue);
 }
 
+TEST_F(PredicateTest, RejectsComparisonsOnCompositeFields) {
+  Schema schema({
+      SchemaField::MakeOptional(
+          1, "s", struct_({SchemaField::MakeOptional(2, "value", int32())})),
+      SchemaField::MakeOptional(3, "l",
+                                list(SchemaField::MakeOptional(4, "element", int32()))),
+      SchemaField::MakeOptional(5, "m",
+                                map(SchemaField::MakeRequired(6, "key", string()),
+                                    SchemaField::MakeOptional(7, "value", int32()))),
+  });
+  for (const std::string name : {"s", "l", "m"}) {
+    SCOPED_TRACE(name);
+    EXPECT_THAT(Expressions::Equal(name, Literal::Int(1))->Bind(schema, true),
+                IsError(ErrorKind::kInvalidExpression));
+    EXPECT_THAT(
+        Expressions::In(name, {Literal::Int(1), Literal::Int(2)})->Bind(schema, true),
+        IsError(ErrorKind::kInvalidExpression));
+  }
+}
+
 TEST_F(PredicateTest, ComplexExpressionCombinations) {
   auto eq_pred = Expressions::Equal("age", Literal::Int(25));
   auto null_pred = Expressions::IsNull("name");
diff --git a/src/iceberg/util/type_util.cc b/src/iceberg/util/type_util.cc
index 83b41bc1c..5f96a28fe 100644
--- a/src/iceberg/util/type_util.cc
+++ b/src/iceberg/util/type_util.cc
@@ -161,7 +161,7 @@ void NameToIdVisitor::Finish() {
   }
 }
 
-Status PositionPathVisitor::Visit(const PrimitiveType& type) {
+Status PositionPathVisitor::RecordPosition(const Type& type) {
   if (current_field_id_ == kUnassignedFieldId) {
     return InvalidSchema("Current field id is not assigned, type: {}", type.ToString());
   }
@@ -175,21 +175,19 @@ Status PositionPathVisitor::Visit(const PrimitiveType& type) {
   return {};
 }
 
-Status PositionPathVisitor::Visit(const VariantType& type) {
-  if (current_field_id_ == kUnassignedFieldId) {
-    return InvalidSchema("Current field id is not assigned, type: {}", type.ToString());
-  }
-
-  if (auto ret = position_path_.try_emplace(current_field_id_, current_path_);
-      !ret.second) {
-    return InvalidSchema("Duplicate field id found: {}, prev path: {}, curr path: {}",
-                         current_field_id_, ret.first->second, current_path_);
-  }
+Status PositionPathVisitor::Visit(const PrimitiveType& type) {
+  return RecordPosition(type);
+}
 
-  return {};
+Status PositionPathVisitor::Visit(const VariantType& type) {
+  return RecordPosition(type);
 }
 
 Status PositionPathVisitor::Visit(const StructType& type) {
+  // The root schema has no field id; nested structs do and can be tested for null.
+  if (!current_path_.empty()) {
+    ICEBERG_RETURN_UNEXPECTED(RecordPosition(type));
+  }
   for (size_t i = 0; i < type.fields().size(); ++i) {
     const auto& field = type.fields()[i];
     current_field_id_ = field.field_id();
@@ -200,10 +198,10 @@ Status PositionPathVisitor::Visit(const StructType& type) {
   return {};
 }
 
-// Non-struct types are not supported yet, but it is not an error.
-Status PositionPathVisitor::Visit(const ListType& type) { return {}; }
+// Containers are accessible as fields, but their elements are not scalar row fields.
+Status PositionPathVisitor::Visit(const ListType& type) { return RecordPosition(type); }
 
-Status PositionPathVisitor::Visit(const MapType& type) { return {}; }
+Status PositionPathVisitor::Visit(const MapType& type) { return RecordPosition(type); }
 
 std::unordered_map<int32_t, std::vector<size_t>> PositionPathVisitor::Finish() {
   return std::move(position_path_);
diff --git a/src/iceberg/util/type_util.h b/src/iceberg/util/type_util.h
index 72e6bacee..49042f189 100644
--- a/src/iceberg/util/type_util.h
+++ b/src/iceberg/util/type_util.h
@@ -95,6 +95,8 @@ class PositionPathVisitor {
   std::unordered_map<int32_t, std::vector<size_t>> Finish();
 
  private:
+  Status RecordPosition(const Type& type);
+
   constexpr static int32_t kUnassignedFieldId = -1;
   int32_t current_field_id_ = kUnassignedFieldId;
   std::vector<size_t> current_path_;