diff --git a/src/iceberg/data/file_scan_task_reader.cc b/src/iceberg/data/file_scan_task_reader.cc index 8c55e31b2..64533195e 100644 --- a/src/iceberg/data/file_scan_task_reader.cc +++ b/src/iceberg/data/file_scan_task_reader.cc @@ -21,6 +21,7 @@ #include #include +#include #include #include #include @@ -28,8 +29,11 @@ #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 +68,39 @@ ReaderOptions MakeReaderOptions(const FileScanTask& task, std::shared_ptr reader, - std::unique_ptr delete_filter, - std::shared_ptr<::iceberg::Schema> required_schema, - std::shared_ptr<::iceberg::Schema> projected_schema, - ProjectionContext projection_context) + // `delete_filter` is null for tasks without delete files; `residual_evaluator` + // is null when residual enforcement is disabled or not needed for the task. + TaskStreamSource(std::unique_ptr reader, + std::unique_ptr delete_filter, + std::unique_ptr 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)), - required_schema_(std::move(required_schema)), + residual_evaluator_(std::move(residual_evaluator)), projected_schema_(std::move(projected_schema)), - project_all_rows_(required_schema_->SameSchema(*projected_schema_)), + project_all_rows_(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()); } @@ -107,20 +129,52 @@ 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 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 (alive.alive_count() == input_batch.length && project_all_rows_) { + if (residual_evaluator_ != nullptr) { + const int64_t candidate_count = + filtered ? static_cast(selection.size()) : input_batch.length; + ICEBERG_ASSIGN_OR_RAISE( + auto kept, KeepResidualMatches(input_arrow_schema, input_batch, selection, + /*all_rows_selected=*/!filtered)); + filtered = filtered || kept.size() != candidate_count; + if (filtered) { + selection = std::move(kept); + if (selection.empty()) { + continue; + } + } + } + + if (!filtered && project_all_rows_) { // 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(input_batch.length)); + std::iota(selection.begin(), selection.end(), int32_t{0}); // NOLINT + } + return ProjectBatch(&input_batch, selection, projection_context_); } } @@ -131,9 +185,43 @@ class MergeOnReadStreamSource { } 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> KeepResidualMatches(const ArrowSchema& input_arrow_schema, + const ArrowArray& input_batch, + const std::vector& selection, + bool all_rows_selected) { + ICEBERG_PRECHECK(std::in_range(input_batch.length), + "Batch length {} exceeds int32_t row index capacity", + input_batch.length); + ICEBERG_ASSIGN_OR_RAISE(auto row, + ArrowArrayStructLike::Make(input_arrow_schema, input_batch)); + std::vector 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(index)); + } + return {}; + }; + if (all_rows_selected) { + kept.reserve(static_cast(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_; std::unique_ptr delete_filter_; - std::shared_ptr<::iceberg::Schema> required_schema_; + std::unique_ptr residual_evaluator_; std::shared_ptr<::iceberg::Schema> projected_schema_; bool project_all_rows_ = false; ProjectionContext projection_context_; @@ -162,6 +250,45 @@ class FileScanTaskReader::Impl { new Impl(std::move(options), std::move(field_lookup), std::move(delete_counter))); } + // Columns that must be added to the read schema so the task's residual filter + // can be evaluated, or nullopt when the residual must not be enforced. An + // empty vector means every referenced column is already projected. + Result>> ResidualReadColumns( + const FileScanTask& task) const { + if (!apply_residuals_) { + return std::nullopt; + } + const auto& residual = task.residual_filter(); + if (residual == nullptr || residual->op() == Expression::Operation::kTrue) { + return std::nullopt; + } + + // Residuals are unbound and reference the planning scan's schema: bind them + // against the table schema with the scan's case sensitivity. + ICEBERG_ASSIGN_OR_RAISE(auto bound, + Binder::Bind(*table_schema_, residual, case_sensitive_)); + ICEBERG_ASSIGN_OR_RAISE(auto field_id_set, + ReferenceVisitor::GetReferencedFieldIds(bound)); + std::vector field_ids(field_id_set.begin(), field_id_set.end()); + std::ranges::sort(field_ids); + + std::vector read_columns; + for (int32_t field_id : field_ids) { + if (FindTopLevelField(*projected_schema_, field_id) != nullptr) { + continue; + } + const SchemaField* field = FindTopLevelField(*table_schema_, field_id); + if (field == nullptr) { + return InvalidArgument( + "Residual filter references field id {}, which is not a top-level " + "column; nested residual references are not supported", + field_id); + } + read_columns.push_back(*field); + } + return read_columns; + } + Result Open(const FileScanTask& task) { const auto& data_file = task.data_file(); ICEBERG_PRECHECK(data_file != nullptr, "Data file must not be null"); @@ -188,7 +315,9 @@ class FileScanTaskReader::Impl { } } - if (task.delete_files().empty()) { + ICEBERG_ASSIGN_OR_RAISE(auto residual_columns, ResidualReadColumns(task)); + + if (task.delete_files().empty() && !residual_columns.has_value()) { auto options = MakeReaderOptions( task, io_, projected_schema_, task.residual_filter(), name_mapping_, properties_, data_file->first_row_id, data_file->data_sequence_number); @@ -197,23 +326,31 @@ 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& f) { - return f->content == DataFile::Content::kPositionDeletes; - }); + // The read schema covers the projection plus any columns needed to enforce + // the residual filter; the stream projects back to the caller's schema. + std::shared_ptr requested_schema = projected_schema_; + if (residual_columns.has_value() && !residual_columns->empty()) { + std::vector fields(projected_schema_->fields().begin(), + projected_schema_->fields().end()); + fields.insert(fields.end(), residual_columns->begin(), residual_columns->end()); + requested_schema = std::make_shared(std::move(fields)); + } - 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 delete_filter; + std::shared_ptr required_schema = requested_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& 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)); + ICEBERG_ASSIGN_OR_RAISE( + delete_filter, + DeleteFilter::Make(data_file->file_path, task.delete_files(), requested_schema, + io_, field_lookup_, has_position_deletes, delete_counter_)); + required_schema = delete_filter->RequiredSchema(); + } auto options = MakeReaderOptions(task, io_, required_schema, task.residual_filter(), name_mapping_, properties_, data_file->first_row_id, @@ -221,28 +358,48 @@ class FileScanTaskReader::Impl { ICEBERG_ASSIGN_OR_RAISE(auto reader, ReaderFactoryRegistry::Open(data_file->file_format, options)); - auto mor_reader = std::make_unique( - std::move(reader), std::move(delete_filter), std::move(required_schema), - projected_schema_, std::move(projection_context)); - return MakeArrowArrayStream(std::move(mor_reader)); + auto project_batch_function = ProjectionContext::ResolveProjectBatchFunction(); + // ProjectionContext borrows schemas that are kept in TaskStreamSource. + ICEBERG_ASSIGN_OR_RAISE(auto projection_context, + ProjectionContext::Make(*required_schema, *projected_schema_, + project_batch_function)); + + std::unique_ptr residual_evaluator; + if (residual_columns.has_value()) { + // Bind to the read schema: its field positions describe every batch. + ICEBERG_ASSIGN_OR_RAISE( + residual_evaluator, + Evaluator::Make(*required_schema, task.residual_filter(), case_sensitive_)); + } + + auto task_source = std::make_unique( + 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: Impl(Options options, DeleteFilter::FieldLookup field_lookup, std::shared_ptr 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)), properties_(ReaderProperties::FromMap(options.properties)), + case_sensitive_(options.case_sensitive), + apply_residuals_(options.apply_residuals), field_lookup_(std::move(field_lookup)), delete_counter_(std::move(delete_counter)) {} std::shared_ptr io_; + std::shared_ptr table_schema_; std::vector> schemas_; std::shared_ptr projected_schema_; std::shared_ptr name_mapping_; ReaderProperties properties_; + bool case_sensitive_ = true; + bool apply_residuals_ = false; DeleteFilter::FieldLookup field_lookup_; std::shared_ptr delete_counter_; }; diff --git a/src/iceberg/data/file_scan_task_reader.h b/src/iceberg/data/file_scan_task_reader.h index 33a4b82ff..ea0379dbe 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. @@ -59,6 +59,17 @@ class ICEBERG_DATA_EXPORT FileScanTaskReader { std::shared_ptr name_mapping; /// Format-specific or implementation-specific options for data readers. std::unordered_map properties; + /// Case sensitivity used to bind residual filters. Must match the planning + /// scan's case sensitivity because task residuals are unbound and tasks do + /// not carry this setting. + bool case_sensitive{true}; + /// Whether to enforce each task's residual filter row by row, after deletes + /// and before final projection. The default keeps the serial reader's + /// results unchanged; the parallel reader always enables it. Columns used + /// only by the residual are read for evaluation and removed by the final + /// projection. Format readers may ignore `ReaderOptions.filter`, so this is + /// the only guaranteed enforcement path. + bool apply_residuals{false}; }; /// \brief Create a reusable task reader from shared read context. @@ -68,6 +79,10 @@ 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, the task's residual filter is enforced + /// row by row after deletes and before the final projection; unsupported + /// residual references fail with an explicit error. + /// /// 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/test/file_scan_task_reader_test.cc b/src/iceberg/test/file_scan_task_reader_test.cc index 05416bd7d..99e0034e7 100644 --- a/src/iceberg/test/file_scan_task_reader_test.cc +++ b/src/iceberg/test/file_scan_task_reader_test.cc @@ -48,6 +48,8 @@ #include "iceberg/data/position_delete_writer.h" #include "iceberg/deletes/dv_writer.h" #include "iceberg/deletes/position_delete_index.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" @@ -95,6 +97,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 { @@ -385,6 +391,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 OpenResidualTask( + std::shared_ptr residual, + std::vector> 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 residual, + std::string_view expected_json, + std::vector> 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& expected_lengths, ReaderProperties properties = {}) { @@ -644,6 +683,161 @@ 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, 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) { + 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, "Baz"]])", {eq_delete})); +} + +TEST_F(FileScanTaskReaderTest, OpenAppliesResidualFilterOnColumnAbsentFromProjection) { + ICEBERG_UNWRAP_OR_FAIL(auto stream, OpenResidualTask(Expressions::Equal( + "category", Literal::String("blue")))); + + // The filter column is read for evaluation but must not leak into the output. + auto record_batch_reader = ::arrow::ImportRecordBatchReader(&stream).ValueOrDie(); + auto result = record_batch_reader->Next(); + ASSERT_TRUE(result.ok()) << result.status().message(); + auto actual_batch = result.ValueOrDie(); + ASSERT_NE(actual_batch, nullptr); + ICEBERG_UNWRAP_OR_FAIL(auto expected_schema, MakeArrowSchema(*projected_schema_)); + ASSERT_TRUE(actual_batch->schema()->Equals(*expected_schema)) + << actual_batch->schema()->ToString(); + ICEBERG_UNWRAP_OR_FAIL(auto expected_batch, + MakeRecordBatch(expected_schema, R"([[1, "Foo"]])")); + ASSERT_TRUE(actual_batch->Equals(*expected_batch)) + << actual_batch->ToString() << "\nExpected:\n" + << expected_batch->ToString(); + + result = record_batch_reader->Next(); + ASSERT_TRUE(result.ok()) << result.status().message(); + ASSERT_EQ(result.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, OpenRejectsResidualOnNestedColumn) { + table_schema_ = std::make_shared(std::vector{ + SchemaField::MakeRequired(1, "id", int32()), + SchemaField::MakeOptional(2, "s", + std::make_shared(std::vector{ + SchemaField::MakeOptional(3, "tags", string())}))}); + projected_schema_ = table_schema_; + EXPECT_THAT(OpenResidualTask(Expressions::NotNull("s.tags"), {}, + /*case_sensitive=*/true, /*apply_residuals=*/true, + R"([[1, {"tags": "a"}]])"), + HasErrorMessage("which is not a top-level column")); +} + TEST_F(FileScanTaskReaderTest, OpenWithDeletionVectorFiltersRows) { ICEBERG_UNWRAP_OR_FAIL( auto data_file,