feat(data): enforce residual filters in the task reader

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 {}",

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,36 +284,46 @@ 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,
                                      data_file->data_sequence_number);
     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_;
 };

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

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();
   }

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));