From 191f76da546a9d10fb5236a7af2282e2fc08b383 Mon Sep 17 00:00:00 2001 From: Kam Cheung Ting Date: Mon, 26 Oct 2026 10:26:01 +0000 Subject: [PATCH 1/3] feat(data): enforce residual filters in the task reader FileScanTaskReader gains two Options: case_sensitive (matching the planning scan's setting, because task residuals are unbound) and apply_residuals (default false, keeping the serial reader's results unchanged; the parallel reader will always enable it). When enabled, each task's residual is enforced row by row after deletes and before the final projection with the existing Evaluator. Residual-only columns are appended to the read schema and projected away again, so a filter can reference columns absent from the output projection. Residuals are bound against the table schema with the configured case sensitivity; unresolvable or nested references fail with explicit errors instead of silently returning unfiltered rows. ReaderOptions.filter is still forwarded to format readers as a pruning hint, but nothing relies on it. The merge-on-read stream source now also serves tasks without deletes when a residual must be enforced: delete filtering and residual evaluation are both optional, and the pass-through fast path only triggers when no filter removed a row. Tests cover residuals with position and equality deletes, filter columns absent from the projection, case-insensitive residuals, null values, always-false residuals, kept row identities, apply_residuals disabled, and explicit errors for unknown, case-mismatched, and nested references. --- src/iceberg/data/file_scan_task_reader.cc | 230 +++++++++++++++--- src/iceberg/data/file_scan_task_reader.h | 18 +- .../test/file_scan_task_reader_test.cc | 193 +++++++++++++++ 3 files changed, 400 insertions(+), 41 deletions(-) diff --git a/src/iceberg/data/file_scan_task_reader.cc b/src/iceberg/data/file_scan_task_reader.cc index 8d3c73281..99793990d 100644 --- a/src/iceberg/data/file_scan_task_reader.cc +++ b/src/iceberg/data/file_scan_task_reader.cc @@ -21,14 +21,19 @@ #include #include +#include #include #include +#include #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,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 +130,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 (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 (alive.alive_count() == input_batch.length && project_all_rows_) { + 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}); + } + return ProjectBatch(&input_batch, selection, projection_context_); } } @@ -131,9 +186,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 +251,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::sort(field_ids.begin(), field_ids.end()); + + 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"); @@ -189,8 +317,9 @@ class FileScanTaskReader::Impl { } auto filter = task.residual_filter(); + ICEBERG_ASSIGN_OR_RAISE(auto residual_columns, ResidualReadColumns(task)); - if (task.delete_files().empty()) { + if (task.delete_files().empty() && !residual_columns.has_value()) { 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 +328,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, - properties_.Get(ReaderProperties::kArrowValidateBatch))); + 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, filter, case_sensitive_, name_mapping_, properties_, data_file->first_row_id, @@ -225,30 +360,49 @@ 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, + properties_.Get(ReaderProperties::kArrowValidateBatch))); + + 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)), 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)) {} std::shared_ptr io_; + std::shared_ptr table_schema_; std::vector> schemas_; std::shared_ptr projected_schema_; bool case_sensitive_; std::shared_ptr name_mapping_; ReaderProperties properties_; + 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 591d49c92..2e9f6b263 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. @@ -55,12 +55,20 @@ class ICEBERG_DATA_EXPORT FileScanTaskReader { /// The output schema for the returned ArrowArrayStream. Must be a /// projection of table_schema. std::shared_ptr 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 name_mapping; /// Format-specific or implementation-specific options for data readers. std::unordered_map properties; + /// 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. @@ -70,6 +78,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 596ec488a..e6351375e 100644 --- a/src/iceberg/test/file_scan_task_reader_test.cc +++ b/src/iceberg/test/file_scan_task_reader_test.cc @@ -51,6 +51,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 +99,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 +393,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 = {}) { @@ -647,6 +685,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, From d176c6705889bba75570c9160bb7ba0b533d1e2f Mon Sep 17 00:00:00 2001 From: Kam Cheung Ting Date: Tue, 27 Oct 2026 01:29:55 +0000 Subject: [PATCH 2/3] style(data): satisfy clang-tidy modernize-use-ranges Use std::ranges::sort for the residual field ids, and keep std::iota for the row-index selection with a NOLINT: ranges::iota requires C++23, but the repo builds with C++20 compatibility (same pattern as parquet_reader.cc). --- src/iceberg/data/file_scan_task_reader.cc | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/iceberg/data/file_scan_task_reader.cc b/src/iceberg/data/file_scan_task_reader.cc index 99793990d..69b30fa02 100644 --- a/src/iceberg/data/file_scan_task_reader.cc +++ b/src/iceberg/data/file_scan_task_reader.cc @@ -173,7 +173,7 @@ class TaskStreamSource { // 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}); + std::iota(selection.begin(), selection.end(), int32_t{0}); // NOLINT } return ProjectBatch(&input_batch, selection, projection_context_); } @@ -271,7 +271,7 @@ class FileScanTaskReader::Impl { ICEBERG_ASSIGN_OR_RAISE(auto field_id_set, ReferenceVisitor::GetReferencedFieldIds(bound)); std::vector field_ids(field_id_set.begin(), field_id_set.end()); - std::sort(field_ids.begin(), field_ids.end()); + std::ranges::sort(field_ids); std::vector read_columns; for (int32_t field_id : field_ids) { From 87764afd88cc175997b19df37a4aec4bf08a6d4a Mon Sep 17 00:00:00 2001 From: Kam Cheung Ting Date: Fri, 6 Nov 2026 08:21:45 +0000 Subject: [PATCH 3/3] fix(data): preserve residual field identity and null semantics --- src/iceberg/arrow_c_data_util.cc | 53 ++-- src/iceberg/arrow_c_data_util_internal.h | 25 +- src/iceberg/data/file_scan_task_reader.cc | 179 ++++++++---- src/iceberg/data/file_scan_task_reader.h | 11 +- src/iceberg/expression/evaluator.cc | 8 + src/iceberg/expression/predicate.cc | 5 + src/iceberg/test/arrow_c_data_util_test.cc | 160 +++++++++-- src/iceberg/test/evaluator_test.cc | 51 ++++ .../test/file_scan_task_reader_test.cc | 254 ++++++++++++++++++ src/iceberg/test/predicate_test.cc | 20 ++ src/iceberg/util/type_util.cc | 28 +- src/iceberg/util/type_util.h | 2 + 12 files changed, 678 insertions(+), 118 deletions(-) diff --git a/src/iceberg/arrow_c_data_util.cc b/src/iceberg/arrow_c_data_util.cc index 915eeabcc..f4263504a 100644 --- a/src/iceberg/arrow_c_data_util.cc +++ b/src/iceberg/arrow_c_data_util.cc @@ -232,9 +232,7 @@ Status AppendValue(const ArrowSchema& input_schema, const ArrowArray& input_arra } // namespace ProjectionContext::ProjectionContext(ProjectionContext&& other) noexcept - : input_schema_(std::exchange(other.input_schema_, nullptr)), - output_schema_(std::exchange(other.output_schema_, nullptr)), - selected_field_indices_(std::move(other.selected_field_indices_)), + : selected_field_indices_(std::move(other.selected_field_indices_)), input_arrow_schema_(other.input_arrow_schema_), output_arrow_schema_(other.output_arrow_schema_), project_batch_function_(std::exchange(other.project_batch_function_, nullptr)), @@ -256,8 +254,6 @@ ProjectionContext& ProjectionContext::operator=(ProjectionContext&& other) noexc output_arrow_schema_.release(&output_arrow_schema_); } - input_schema_ = std::exchange(other.input_schema_, nullptr); - output_schema_ = std::exchange(other.output_schema_, nullptr); selected_field_indices_ = std::move(other.selected_field_indices_); input_arrow_schema_ = other.input_arrow_schema_; other.input_arrow_schema_.release = nullptr; @@ -280,27 +276,54 @@ ProjectionContext::~ProjectionContext() { Result ProjectionContext::Make( const Schema& input_schema, const Schema& output_schema, + const ArrowSchema& physical_input_schema, ProjectionContext::ProjectBatchFunction project_batch_function, bool validate_batch) { + ArrowSchemaView schema_view; + ArrowError error; + ArrowErrorInit(&error); + ICEBERG_NANOARROW_RETURN_UNEXPECTED_WITH_ERROR( + ArrowSchemaViewInit(&schema_view, &physical_input_schema, &error), error); + ICEBERG_PRECHECK(schema_view.type == NANOARROW_TYPE_STRUCT, + "Physical input schema must be a struct, but is {}", + ArrowTypeString(schema_view.type)); + ICEBERG_PRECHECK( + std::cmp_equal(physical_input_schema.n_children, input_schema.fields().size()), + "Physical input schema must have {} fields to match the logical schema, but has {}", + input_schema.fields().size(), physical_input_schema.n_children); + ICEBERG_ASSIGN_OR_RAISE( auto selected_field_indices, BuildSelectedFieldIndices(input_schema.fields(), output_schema.fields())); - ProjectionContext context; - context.input_schema_ = &input_schema; - context.output_schema_ = &output_schema; - context.selected_field_indices_ = std::move(selected_field_indices); - ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(input_schema, &context.input_arrow_schema_)); - ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(output_schema, &context.output_arrow_schema_)); context.project_batch_function_ = project_batch_function; context.validate_batch_ = validate_batch; + ICEBERG_NANOARROW_RETURN_UNEXPECTED( + ArrowSchemaDeepCopy(&physical_input_schema, &context.input_arrow_schema_)); + ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(output_schema, &context.output_arrow_schema_)); + // Replace each logical output field with the physical layout of its selected input + // field. The deep copy also carries the physical metadata (e.g. field ids and + // extension annotations); only the top-level name and nullability come from + // output_schema. + const auto& output_fields = output_schema.fields(); + for (size_t i = 0; i < output_fields.size(); ++i) { + auto* output_field = context.output_arrow_schema_.children[i]; + output_field->release(output_field); + ICEBERG_NANOARROW_RETURN_UNEXPECTED(ArrowSchemaDeepCopy( + physical_input_schema.children[selected_field_indices[i]], output_field)); + ICEBERG_NANOARROW_RETURN_UNEXPECTED( + ArrowSchemaSetName(output_field, std::string(output_fields[i].name()).c_str())); + // Preserve type-specific flags such as sorted map keys when setting nullability. + if (output_fields[i].optional()) { + output_field->flags |= ARROW_FLAG_NULLABLE; + } else { + output_field->flags &= ~ARROW_FLAG_NULLABLE; + } + } + context.selected_field_indices_ = std::move(selected_field_indices); return context; } -const Schema& ProjectionContext::input_schema() const { return *input_schema_; } - -const Schema& ProjectionContext::output_schema() const { return *output_schema_; } - const ArrowSchema& ProjectionContext::input_arrow_schema() const { return input_arrow_schema_; } diff --git a/src/iceberg/arrow_c_data_util_internal.h b/src/iceberg/arrow_c_data_util_internal.h index c5d203973..5cbb8a8d6 100644 --- a/src/iceberg/arrow_c_data_util_internal.h +++ b/src/iceberg/arrow_c_data_util_internal.h @@ -61,21 +61,28 @@ class ICEBERG_EXPORT ProjectionContext { /// \brief Resolve the registered ProjectBatch implementation. static auto ResolveProjectBatchFunction() -> ProjectBatchFunction; - /// \brief Build reusable projection state for a validated schema pair. + /// \brief Build reusable projection state using the reader's physical Arrow layout. /// /// \param input_schema Schema that describes every input batch. /// \param output_schema Final schema and column order requested by the caller. + /// \param physical_input_schema Physical fields corresponding to input_schema in + /// the same order. Each selected output field becomes a deep copy of its physical + /// counterpart, preserving the physical layout, nested list offset widths, + /// metadata, and type-specific flags; only the top-level name and nullability come + /// from output_schema. /// \param project_batch_function Optional implementation returned by /// ProjectionContext::ResolveProjectBatchFunction, or nullptr to use the nanoarrow /// path. /// \param validate_batch Whether to perform full content validation. Disable only for /// trusted input. /// \note It validates that output_schema selects or reorders complete top-level fields - /// by field id. Nested pruning and type changes are rejected. The input_schema and - /// output_schema passed to Make must outlive the context. ProjectBatch may lazily - /// initialize backend cache; do not share one context across concurrent calls. + /// by field id. Nested pruning and type changes are rejected. Schema inputs are only + /// used during construction; the context owns its physical Arrow schemas. + /// ProjectBatch may lazily initialize backend cache; do not share one context across + /// concurrent calls. static Result Make(const Schema& input_schema, const Schema& output_schema, + const ArrowSchema& physical_input_schema, ProjectBatchFunction project_batch_function, bool validate_batch = true); @@ -86,10 +93,6 @@ class ICEBERG_EXPORT ProjectionContext { ProjectionContext(const ProjectionContext&) = delete; ProjectionContext& operator=(const ProjectionContext&) = delete; - const Schema& input_schema() const; - - const Schema& output_schema() const; - const ArrowSchema& input_arrow_schema() const; const ArrowSchema& output_arrow_schema() const; @@ -105,10 +108,6 @@ class ICEBERG_EXPORT ProjectionContext { private: ProjectionContext() = default; - // Raw schema pointers are borrowed from caller-owned schemas. FileScanTaskReader - // keeps those schema objects alive in the same stream source that owns this context. - const Schema* input_schema_ = nullptr; - const Schema* output_schema_ = nullptr; std::vector selected_field_indices_; ArrowSchema input_arrow_schema_{}; ArrowSchema output_arrow_schema_{}; @@ -244,7 +243,7 @@ ICEBERG_EXPORT Result MakeArrowArrayStream( /// \param row_indices Zero-based row positions to copy from `input_batch`. /// \param projection Reusable schema/projection state created by /// ProjectionContext::Make. -/// \return A newly owned ArrowArray matching `projection.output_schema()`. +/// \return A newly owned ArrowArray matching `projection.output_arrow_schema()`. ICEBERG_EXPORT Result ProjectBatch(ArrowArray* input_batch, std::span row_indices, ProjectionContext& projection); diff --git a/src/iceberg/data/file_scan_task_reader.cc b/src/iceberg/data/file_scan_task_reader.cc index 69b30fa02..1b4dee79e 100644 --- a/src/iceberg/data/file_scan_task_reader.cc +++ b/src/iceberg/data/file_scan_task_reader.cc @@ -23,9 +23,13 @@ #include #include #include +#include +#include #include #include +#include + #include "iceberg/arrow_c_data_guard_internal.h" #include "iceberg/arrow_c_data_util_internal.h" #include "iceberg/data/delete_filter.h" @@ -33,9 +37,9 @@ #include "iceberg/expression/evaluator.h" #include "iceberg/file_reader.h" #include "iceberg/manifest/manifest_entry.h" +#include "iceberg/nanoarrow_status_internal.h" #include "iceberg/row/arrow_array_wrapper.h" #include "iceberg/schema.h" -#include "iceberg/schema_internal.h" #include "iceberg/table_scan.h" #include "iceberg/util/macros.h" @@ -80,6 +84,69 @@ const SchemaField* FindTopLevelField(const Schema& schema, int32_t field_id) { return nullptr; } +// Collect original references before binding can fold predicates away. The evaluator +// binds the original residual again against its own schema, so it needs these fields +// even when their predicates are constant under the table schema. +class ResidualBinder : public Binder { + public: + ResidualBinder(const Schema& schema, bool case_sensitive) + : Binder(schema, case_sensitive), + schema_(schema), + case_sensitive_(case_sensitive) {} + + using Binder::Predicate; + + Result> Predicate( + const std::shared_ptr& pred) override { + ICEBERG_ASSIGN_OR_RAISE( + auto field, schema_.FindFieldByName(pred->reference()->name(), case_sensitive_)); + if (field.has_value()) { + const auto& schema_field = field->get(); + const bool composite_null_check = + schema_field.type()->is_nested() && + pred->unbound_term().kind() == Term::Kind::kReference && + (pred->op() == Expression::Operation::kIsNull || + pred->op() == Expression::Operation::kNotNull); + if (!schema_field.type()->is_primitive() && !composite_null_check) { + return NotSupported( + "Residual filter references non-primitive column '{}' of type {}; " + "only null checks are supported for composite residual columns", + pred->reference()->name(), schema_field.type()->ToString()); + } + field_ids_.insert(schema_field.field_id()); + } + return Binder::Predicate(pred); + } + + const std::unordered_set& field_ids() const { return field_ids_; } + + private: + const Schema& schema_; + bool case_sensitive_; + std::unordered_set field_ids_; +}; + +// Exposes residual columns in the order of their planning-schema fields. The read +// schema can include historical delete columns with colliding names, so positions +// are resolved by field id once when opening the task. +class ResidualRow : public StructLike { + public: + ResidualRow(const StructLike& row, std::span positions) + : row_(row), positions_(positions) {} + + Result GetField(size_t pos) const override { + ICEBERG_PRECHECK(pos < positions_.size(), "Residual field position {} out of range", + pos); + return row_.GetField(positions_[pos]); + } + + size_t num_fields() const override { return positions_.size(); } + + private: + const StructLike& row_; + std::span positions_; +}; + // 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 both read a possibly wider schema @@ -91,22 +158,15 @@ class TaskStreamSource { 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, + std::vector residual_positions, bool project_all_columns, ProjectionContext projection_context) : reader_(std::move(reader)), delete_filter_(std::move(delete_filter)), residual_evaluator_(std::move(residual_evaluator)), - projected_schema_(std::move(projected_schema)), - project_all_rows_(required_schema->SameSchema(*projected_schema_)), + residual_positions_(std::move(residual_positions)), + project_all_columns_(project_all_columns), projection_context_(std::move(projection_context)) {} - ~TaskStreamSource() { - if (cached_schema_.has_value() && cached_schema_->release != nullptr) { - cached_schema_->release(&cached_schema_.value()); - } - } - Status Close() { if (reader_ == nullptr) { return {}; @@ -115,11 +175,7 @@ class TaskStreamSource { } Result> Next() { - if (!cached_schema_.has_value()) { - // 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 auto& input_arrow_schema = projection_context_.input_arrow_schema(); while (true) { ICEBERG_ASSIGN_OR_RAISE(auto next_batch, reader_->Next()); @@ -162,7 +218,7 @@ class TaskStreamSource { } } - if (!filtered && 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; @@ -180,8 +236,9 @@ class TaskStreamSource { } Result Schema() { - ArrowSchema schema; - ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(*projected_schema_, &schema)); + ArrowSchema schema{}; + ICEBERG_NANOARROW_RETURN_UNEXPECTED( + ArrowSchemaDeepCopy(&projection_context_.output_arrow_schema(), &schema)); return schema; } @@ -197,10 +254,11 @@ class TaskStreamSource { input_batch.length); ICEBERG_ASSIGN_OR_RAISE(auto row, ArrowArrayStructLike::Make(input_arrow_schema, input_batch)); + ResidualRow residual_row(*row, residual_positions_); 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)); + ICEBERG_ASSIGN_OR_RAISE(bool matches, residual_evaluator_->Evaluate(residual_row)); if (matches) { kept.push_back(static_cast(index)); } @@ -223,10 +281,9 @@ class TaskStreamSource { std::unique_ptr reader_; std::unique_ptr delete_filter_; std::unique_ptr residual_evaluator_; - std::shared_ptr<::iceberg::Schema> projected_schema_; - bool project_all_rows_ = false; + std::vector residual_positions_; + bool project_all_columns_ = false; ProjectionContext projection_context_; - std::optional cached_schema_; }; } // namespace @@ -251,33 +308,26 @@ 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 { + // Planning-schema fields used by the residual, or nullptr when it is not enforced. + // Keep this schema independent of projection aliases and historical delete fields. + Result> ResidualSchema(const FileScanTask& task) const { if (!apply_residuals_) { - return std::nullopt; + return nullptr; } const auto& residual = task.residual_filter(); if (residual == nullptr || residual->op() == Expression::Operation::kTrue) { - return std::nullopt; + return nullptr; } // 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()); + ResidualBinder binder(*table_schema_, case_sensitive_); + ICEBERG_RETURN_UNEXPECTED(Visit>(residual, binder)); + std::vector field_ids(binder.field_ids().begin(), binder.field_ids().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( @@ -287,7 +337,7 @@ class FileScanTaskReader::Impl { } read_columns.push_back(*field); } - return read_columns; + return std::make_shared(std::move(read_columns)); } Result Open(const FileScanTask& task) { @@ -316,10 +366,16 @@ class FileScanTaskReader::Impl { } } + ICEBERG_ASSIGN_OR_RAISE(auto residual_schema, ResidualSchema(task)); auto filter = task.residual_filter(); - ICEBERG_ASSIGN_OR_RAISE(auto residual_columns, ResidualReadColumns(task)); + if (residual_schema != nullptr) { + // Pruning readers also need field-id references, independent of projected + // names and historical delete columns in the physical read schema. + ICEBERG_ASSIGN_OR_RAISE(filter, + Binder::Bind(*residual_schema, filter, case_sensitive_)); + } - if (task.delete_files().empty() && !residual_columns.has_value()) { + if (task.delete_files().empty() && residual_schema == nullptr) { auto options = MakeReaderOptions( task, io_, projected_schema_, filter, case_sensitive_, name_mapping_, properties_, data_file->first_row_id, data_file->data_sequence_number); @@ -331,10 +387,14 @@ class FileScanTaskReader::Impl { // 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()) { + if (residual_schema != nullptr) { std::vector fields(projected_schema_->fields().begin(), projected_schema_->fields().end()); - fields.insert(fields.end(), residual_columns->begin(), residual_columns->end()); + for (const auto& field : residual_schema->fields()) { + if (FindTopLevelField(*projected_schema_, field.field_id()) == nullptr) { + fields.push_back(field); + } + } requested_schema = std::make_shared(std::move(fields)); } @@ -360,24 +420,37 @@ class FileScanTaskReader::Impl { ICEBERG_ASSIGN_OR_RAISE(auto reader, ReaderFactoryRegistry::Open(data_file->file_format, options)); - 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, - properties_.Get(ReaderProperties::kArrowValidateBatch))); + // Use the reader's physical layout for both pass-through and projected batches. + ICEBERG_ASSIGN_OR_RAISE(auto input_arrow_schema, reader->Schema()); + internal::ArrowSchemaGuard input_schema_guard(&input_arrow_schema); + ICEBERG_ASSIGN_OR_RAISE( + auto projection_context, + ProjectionContext::Make(*required_schema, *projected_schema_, input_arrow_schema, + ProjectionContext::ResolveProjectBatchFunction(), + properties_.Get(ReaderProperties::kArrowValidateBatch))); std::unique_ptr residual_evaluator; - if (residual_columns.has_value()) { - // Bind to the read schema: its field positions describe every batch. + std::vector residual_positions; + if (residual_schema != nullptr) { + // Bind names only against planning-schema fields, then map to batch positions + // by id. Historical equality-delete fields may reuse a current column's name. ICEBERG_ASSIGN_OR_RAISE( residual_evaluator, - Evaluator::Make(*required_schema, task.residual_filter(), case_sensitive_)); + Evaluator::Make(*residual_schema, task.residual_filter(), case_sensitive_)); + for (const auto& field : residual_schema->fields()) { + auto fields = required_schema->fields(); + auto it = std::ranges::find(fields, field.field_id(), &SchemaField::field_id); + ICEBERG_PRECHECK(it != fields.end(), "Cannot find residual field id {}", + field.field_id()); + residual_positions.push_back(static_cast(it - fields.begin())); + } } + const bool project_all_columns = required_schema->SameSchema(*projected_schema_); 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)); + std::move(residual_positions), project_all_columns, + std::move(projection_context)); return MakeArrowArrayStream(std::move(task_source)); } diff --git a/src/iceberg/data/file_scan_task_reader.h b/src/iceberg/data/file_scan_task_reader.h index 2e9f6b263..5f7c5cac8 100644 --- a/src/iceberg/data/file_scan_task_reader.h +++ b/src/iceberg/data/file_scan_task_reader.h @@ -38,8 +38,11 @@ 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`. When `apply_residuals` is enabled, each -/// task's residual filter is enforced after deletes and before final projection. +/// stream exposes the fields of `projected_schema` in one stable physical Arrow layout, +/// which may depend on the file format and reader properties (for example, `large_list` +/// instead of `list`). Every batch matches the schema reported by the stream. 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. class ICEBERG_DATA_EXPORT FileScanTaskReader { @@ -67,7 +70,9 @@ class ICEBERG_DATA_EXPORT FileScanTaskReader { /// 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. + /// the only guaranteed enforcement path. Residual references must be top-level + /// primitive columns or null checks on top-level struct, list, or map columns; + /// other references fail when opening the task. bool apply_residuals{false}; }; 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 { } Result IsNull(const std::shared_ptr& 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(expr)) { + ICEBERG_ASSIGN_OR_RAISE(auto value, ref->accessor().Get(row_)); + return std::holds_alternative(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> UnboundPredicateImpl::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/arrow_c_data_util_test.cc b/src/iceberg/test/arrow_c_data_util_test.cc index a21f88c43..c5a9d94c3 100644 --- a/src/iceberg/test/arrow_c_data_util_test.cc +++ b/src/iceberg/test/arrow_c_data_util_test.cc @@ -54,12 +54,8 @@ namespace iceberg::internal { namespace { -std::shared_ptr<::arrow::RecordBatch> MakeBatch(const Schema& schema, - std::string_view json) { - ArrowSchema c_schema; - EXPECT_THAT(ToArrowSchema(schema, &c_schema), IsOk()); - // ImportSchema takes ownership of c_schema and calls release. - auto arrow_schema = ::arrow::ImportSchema(&c_schema).ValueOrDie(); +std::shared_ptr<::arrow::RecordBatch> MakeBatch( + const std::shared_ptr<::arrow::Schema>& arrow_schema, std::string_view json) { auto struct_type = ::arrow::struct_(arrow_schema->fields()); return ::arrow::RecordBatch::FromStructArray( ::arrow::json::ArrayFromJSONString(struct_type, std::string(json)) @@ -67,6 +63,13 @@ std::shared_ptr<::arrow::RecordBatch> MakeBatch(const Schema& schema, .ValueOrDie(); } +std::shared_ptr<::arrow::RecordBatch> MakeBatch(const Schema& schema, + std::string_view json) { + ArrowSchema c_schema; + EXPECT_THAT(ToArrowSchema(schema, &c_schema), IsOk()); + return MakeBatch(::arrow::ImportSchema(&c_schema).ValueOrDie(), json); +} + ProjectionContext::ProjectBatchFunction ArrowComputeFunction() { arrow::RegisterAll(); auto function = ProjectionContext::ResolveProjectBatchFunction(); @@ -77,22 +80,25 @@ ProjectionContext::ProjectBatchFunction ArrowComputeFunction() { std::shared_ptr<::arrow::RecordBatch> RunProjectBatch( const ::arrow::RecordBatch& batch, const std::vector& alive_indices, const Schema& required_schema, const Schema& projected_schema, - ProjectionContext::ProjectBatchFunction project_batch_function, bool validate_batch) { + ProjectionContext::ProjectBatchFunction project_batch_function, + bool validate_batch = true) { ArrowSchema c_schema; ArrowArray c_array; EXPECT_TRUE(::arrow::ExportRecordBatch(batch, &c_array, &c_schema).ok()); ArrowSchemaGuard schema_guard(&c_schema); ArrowArrayGuard array_guard(&c_array); - auto projection = ProjectionContext::Make(required_schema, projected_schema, + auto projection = ProjectionContext::Make(required_schema, projected_schema, c_schema, project_batch_function, validate_batch); EXPECT_THAT(projection, IsOk()); + // The context must own its Arrow schemas independently of the caller. + c_schema.release(&c_schema); auto result = ProjectBatch(&c_array, alive_indices, projection.value()); EXPECT_THAT(result, IsOk()); ArrowSchema out_c_schema; - EXPECT_THAT(ToArrowSchema(projected_schema, &out_c_schema), IsOk()); + EXPECT_EQ(ArrowSchemaDeepCopy(&projection->output_arrow_schema(), &out_c_schema), 0); auto arrow_out_schema = ::arrow::ImportSchema(&out_c_schema).ValueOrDie(); ArrowArray out_array = std::exchange(result.value(), ArrowArray{}); @@ -205,6 +211,87 @@ TEST(ProjectBatchTest, EmptyRowSelection) { ExpectProjectBatch(*batch, alive, *schema, *schema, R"([])"); } +TEST(ProjectBatchTest, EmptyProjectionPreservesRowCount) { + auto required = MakeFullSchema(); + Schema projected(std::vector{}); + auto batch = MakeBatch(*required, R"([[1,"a",1.0],[2,"b",2.0],[3,"c",3.0]])"); + + for (auto backend : + {ProjectionContext::ProjectBatchFunction{nullptr}, ArrowComputeFunction()}) { + SCOPED_TRACE(backend == nullptr ? "nanoarrow" : "Arrow compute"); + for (const auto& alive : {std::vector{}, std::vector{0, 2}}) { + auto actual = RunProjectBatch(*batch, alive, *required, projected, backend); + EXPECT_EQ(actual->num_columns(), 0); + EXPECT_EQ(actual->num_rows(), alive.size()); + EXPECT_TRUE(actual->ValidateFull().ok()) << actual->ValidateFull().ToString(); + } + } +} + +TEST(ProjectBatchTest, PreservesPhysicalListLayoutsWithBothBackends) { + auto nested_list = std::make_shared(SchemaField::MakeOptional( + 3, "element", + std::make_shared(SchemaField::MakeOptional(4, "element", int32())))); + Schema required( + std::vector{SchemaField::MakeRequired(1, "id", int32()), + SchemaField::MakeOptional(2, "numbers", nested_list), + SchemaField::MakeOptional(5, "unused", string())}); + Schema projected(std::vector{required.fields()[1].WithName("renamed"), + required.fields()[0]}); + ArrowSchema c_schema; + ASSERT_THAT(ToArrowSchema(required, &c_schema), IsOk()); + auto physical_schema = ::arrow::ImportSchema(&c_schema).ValueOrDie(); + auto fields = physical_schema->fields(); + // Keep the inner list at 32-bit offsets while the outer list uses 64-bit offsets. + fields[1] = fields[1]->WithType(::arrow::large_list(fields[1]->type()->field(0))); + physical_schema = ::arrow::schema(std::move(fields)); + auto batch = MakeBatch(physical_schema, + R"([[1, [[10, null], []], "a"], [2, [], "b"], + [3, [[30]], "c"], [4, null, "d"]])"); + auto expected_schema = ::arrow::schema( + {physical_schema->field(1)->WithName("renamed"), physical_schema->field(0)}); + auto expected = + MakeBatch(expected_schema, R"([[[[10, null], []], 1], [[], 2], [null, 4]])"); + + for (auto backend : + {ProjectionContext::ProjectBatchFunction{nullptr}, ArrowComputeFunction()}) { + SCOPED_TRACE(backend == nullptr ? "nanoarrow" : "Arrow compute"); + const std::vector alive{0, 1, 3}; + auto actual = RunProjectBatch(*batch, alive, required, projected, backend); + ASSERT_TRUE(actual->schema()->Equals(*expected_schema)); + ASSERT_TRUE(actual->ValidateFull().ok()) << actual->ValidateFull().ToString(); + EXPECT_TRUE(actual->Equals(*expected)) << actual->ToString(); + } +} + +TEST(ProjectBatchTest, PreservesPhysicalMapFlagsWhenChangingNullability) { + auto map_type = map(SchemaField::MakeRequired(2, "key", int32()), + SchemaField::MakeOptional(3, "value", string())); + Schema required( + std::vector{SchemaField::MakeRequired(1, "values", map_type)}); + Schema projected( + std::vector{SchemaField::MakeOptional(1, "renamed", map_type)}); + auto canonical = MakeBatch(required, "[]")->schema(); + const auto& canonical_map = + static_cast(*canonical->field(0)->type()); + auto physical_field = canonical->field(0)->WithType(std::make_shared<::arrow::MapType>( + canonical_map.key_field(), canonical_map.item_field(), /*keys_sorted=*/true)); + auto batch = + MakeBatch(::arrow::schema({physical_field}), R"([[[[1, "a"], [2, "b"]]], [[]]])"); + auto expected_schema = + ::arrow::schema({physical_field->WithName("renamed")->WithNullable(true)}); + auto expected = MakeBatch(expected_schema, R"([[[[1, "a"], [2, "b"]]]])"); + + for (auto backend : + {ProjectionContext::ProjectBatchFunction{nullptr}, ArrowComputeFunction()}) { + SCOPED_TRACE(backend == nullptr ? "nanoarrow" : "Arrow compute"); + auto actual = RunProjectBatch(*batch, {0}, required, projected, backend); + EXPECT_TRUE(actual->schema()->Equals(*expected_schema)); + EXPECT_TRUE(actual->ValidateFull().ok()) << actual->ValidateFull().ToString(); + EXPECT_TRUE(actual->Equals(*expected)) << actual->ToString(); + } +} + TEST(ProjectBatchTest, ProjectionRejectsNestedPruning) { auto input_schema = Schema(std::vector{ SchemaField::MakeOptional(1, "person", @@ -220,9 +307,33 @@ TEST(ProjectBatchTest, ProjectionRejectsNestedPruning) { })), }); - auto projection = ProjectionContext::Make(input_schema, output_schema, nullptr); + ArrowSchema physical_schema; + ASSERT_THAT(ToArrowSchema(input_schema, &physical_schema), IsOk()); + ArrowSchemaGuard schema_guard(&physical_schema); + EXPECT_THAT( + ProjectionContext::Make(input_schema, output_schema, physical_schema, nullptr), + IsError(ErrorKind::kInvalidArgument)); +} + +TEST(ProjectBatchTest, PhysicalSchemaMustMatchLogicalFieldCount) { + auto required = MakeFullSchema(); + Schema projected(std::vector{required->fields()[0]}); + ArrowSchema physical_schema; + ASSERT_THAT(ToArrowSchema(projected, &physical_schema), IsOk()); + ArrowSchemaGuard schema_guard(&physical_schema); + + EXPECT_THAT(ProjectionContext::Make(*required, projected, physical_schema, nullptr), + HasErrorMessage("Physical input schema must have 3 fields")); +} + +TEST(ProjectBatchTest, PhysicalSchemaMustBeAStruct) { + Schema schema(std::vector{}); + ArrowSchema physical_schema; + ASSERT_EQ(ArrowSchemaInitFromType(&physical_schema, NANOARROW_TYPE_INT32), 0); + ArrowSchemaGuard schema_guard(&physical_schema); - EXPECT_THAT(projection, IsError(ErrorKind::kInvalidArgument)); + EXPECT_THAT(ProjectionContext::Make(schema, schema, physical_schema, nullptr), + HasErrorMessage("Physical input schema must be a struct")); } TEST(ProjectBatchTest, ValidateBatchControlsNonMonotonicStringOffsets) { @@ -253,8 +364,11 @@ TEST(ProjectBatchTest, ValidateBatchControlsNonMonotonicStringOffsets) { root.null_count = 0; ASSERT_EQ(ArrowArrayFinishBuildingDefault(&root, nullptr), NANOARROW_OK); - auto projection = - ProjectionContext::Make(*schema, *schema, function, validate_batch); + ArrowSchema physical_schema{}; + ASSERT_THAT(ToArrowSchema(*schema, &physical_schema), IsOk()); + ArrowSchemaGuard schema_guard(&physical_schema); + auto projection = ProjectionContext::Make(*schema, *schema, physical_schema, + function, validate_batch); ASSERT_THAT(projection, IsOk()); // Empty selection skips the content check without reading corrupt data. @@ -282,9 +396,13 @@ TEST(ProjectBatchTest, SkippedContentValidationStillRejectsInvalidStructure) { ArrowArray array; ASSERT_TRUE(::arrow::ExportRecordBatch(*batch, &array).ok()); array.length = -1; + ArrowSchema physical_schema{}; + ASSERT_THAT(ToArrowSchema(*schema, &physical_schema), IsOk()); + ArrowSchemaGuard schema_guard(&physical_schema); ICEBERG_UNWRAP_OR_FAIL( auto projection, - ProjectionContext::Make(*schema, *schema, function, /*validate_batch=*/false)); + ProjectionContext::Make(*schema, *schema, physical_schema, function, + /*validate_batch=*/false)); auto result = ProjectBatch(&array, {}, projection); EXPECT_THAT(result, IsError(ErrorKind::kInvalidArrowData)); EXPECT_THAT(result, HasErrorMessage("length")); @@ -294,13 +412,17 @@ TEST(ProjectBatchTest, SkippedContentValidationStillRejectsInvalidStructure) { TEST(ProjectBatchTest, MovePreservesValidateBatch) { auto schema = MakeFullSchema(); - ICEBERG_UNWRAP_OR_FAIL( - auto without_validation, - ProjectionContext::Make(*schema, *schema, nullptr, /*validate_batch=*/false)); + ArrowSchema physical_schema{}; + ASSERT_THAT(ToArrowSchema(*schema, &physical_schema), IsOk()); + ArrowSchemaGuard schema_guard(&physical_schema); + ICEBERG_UNWRAP_OR_FAIL(auto without_validation, + ProjectionContext::Make(*schema, *schema, physical_schema, + nullptr, /*validate_batch=*/false)); auto moved = std::move(without_validation); EXPECT_FALSE(moved.validate_batch()); - ICEBERG_UNWRAP_OR_FAIL(auto validation, - ProjectionContext::Make(*schema, *schema, nullptr)); + ICEBERG_UNWRAP_OR_FAIL( + auto validation, + ProjectionContext::Make(*schema, *schema, physical_schema, nullptr)); EXPECT_TRUE(validation.validate_batch()); validation = std::move(moved); EXPECT_FALSE(validation.validate_batch()); 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::max()} + 1; + const int64_t below = int64_t{std::numeric_limits::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(term))); + ICEBERG_UNWRAP_OR_FAIL( + auto not_null, + Evaluator::Make(*schema_, Expressions::NotNull(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 e6351375e..f3ee56069 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 #include +#include #include #include #include @@ -29,6 +30,8 @@ #include #include +#include +#include #include #include #include @@ -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" @@ -690,6 +694,134 @@ TEST_F(FileScanTaskReaderTest, OpenAppliesResidualFilterWithoutDeletes) { Expressions::GreaterThan("id", Literal::Int(1)), R"([[2, "Bar"], [3, "Baz"]])")); } +TEST_F(FileScanTaskReaderTest, OpenAppliesFoldedResidualsOnUnprojectedRequiredColumn) { + table_schema_ = std::make_shared( + std::vector{SchemaField::MakeRequired(1, "flag", boolean()), + SchemaField::MakeRequired(2, "id", int32())}); + projected_schema_ = std::make_shared( + std::vector{SchemaField::MakeRequired(2, "id", int32())}); + constexpr std::string_view data = "[[true, 1], [false, 2], [true, 3]]"; + + ASSERT_NO_FATAL_FAILURE(VerifyResidualStream(Expressions::NotNull("flag"), + "[[1], [2], [3]]", {}, true, data)); + // Binding against the table folds away flag, but id's position also changes in + // the read schema. The evaluator must bind using the final read positions. + ASSERT_NO_FATAL_FAILURE(VerifyResidualStream( + Expressions::And(Expressions::NotNull("flag"), + Expressions::GreaterThan("id", Literal::Int(1))), + "[[2], [3]]", {}, true, data)); + ASSERT_NO_FATAL_FAILURE(VerifyResidualStream( + Expressions::Or(Expressions::IsNull("flag"), + Expressions::GreaterThan("id", Literal::Int(1))), + "[[2], [3]]", {}, 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( + std::vector{SchemaField::MakeRequired(1, "id", int32()), + SchemaField::MakeOptional(2, "n", int32())}); + projected_schema_ = + std::make_shared(std::vector{table_schema_->fields()[0]}); + const int64_t above = int64_t{std::numeric_limits::max()} + 1; + const int64_t below = int64_t{std::numeric_limits::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]]", {}, true, "[[1, 1], [2, null]]"); + } +} + +class FileScanTaskResidualPhysicalSchemaTest + : public FileScanTaskReaderTest, + public ::testing::WithParamInterface> {}; + +TEST_P(FileScanTaskResidualPhysicalSchemaTest, PreservesLargeListAcrossResidualBatches) { + const auto [filter_rows, prune_columns, with_deletes] = GetParam(); + table_schema_ = std::make_shared(std::vector{ + SchemaField::MakeRequired(1, "id", int32()), + SchemaField::MakeOptional( + 2, "values", + std::make_shared(SchemaField::MakeOptional(3, "element", int32())))}); + projected_schema_ = + prune_columns + ? std::make_shared(std::vector{table_schema_->fields()[1]}) + : table_schema_; + ICEBERG_UNWRAP_OR_FAIL(auto data_file, + MakeDataFile(table_schema_, + std::vector{"[[1, [10, 20]], [2, []]]", + "[[3, [30]], [4, null]]"}, + /*record_count=*/4, /*max_row_group_length=*/2)); + std::vector> deletes; + if (with_deletes) { + ICEBERG_UNWRAP_OR_FAIL( + auto deletion, MakePositionDeleteFile(CreateNewTempFilePathWithSuffix(".parquet"), + {99}, data_file->file_path)); + deletes.push_back(std::move(deletion)); + } + // The first batch always passes unchanged. The second is either all-matching + // or partially filtered, so one stream exercises both paths with the same schema. + auto residual = filter_rows ? Expressions::NotEqual("id", Literal::Int(3)) + : Expressions::GreaterThan("id", Literal::Int(0)); + FileScanTask task(data_file, std::move(deletes), std::move(residual)); + ICEBERG_UNWRAP_OR_FAIL( + auto reader, + FileScanTaskReader::Make({ + .io = file_io_, + .table_schema = table_schema_, + .projected_schema = projected_schema_, + .properties = {{"read.batch-size", "2"}, {"read.arrow.use-large-list", "true"}}, + .apply_residuals = true, + })); + ICEBERG_UNWRAP_OR_FAIL(auto stream, reader->Open(task)); + reader.reset(); + ArrowSchema declared_schema{}; + internal::ArrowSchemaGuard schema_guard(&declared_schema); + ASSERT_EQ(stream.get_schema(&stream, &declared_schema), 0); + const int list_index = prune_columns ? 0 : 1; + EXPECT_STREQ(declared_schema.children[list_index]->format, "+L"); + auto batches = ::arrow::ImportRecordBatchReader(&stream).ValueOrDie(); + std::vector> retained_batches; + const std::array expected_json = + prune_columns + ? std::array{"[[[10, 20]], [[]]]", + filter_rows ? "[[null]]" : "[[[30]], [null]]"} + : std::array{ + "[[1, [10, 20]], [2, []]]", + filter_rows ? "[[4, null]]" : "[[3, [30]], [4, null]]"}; + for (auto json : expected_json) { + auto result = batches->Next(); + ASSERT_TRUE(result.ok()) << result.status(); + auto batch = result.ValueOrDie(); + ASSERT_NE(batch, nullptr); + ASSERT_TRUE(batch->ValidateFull().ok()) << batch->ValidateFull(); + ASSERT_EQ(batch->column(list_index)->type_id(), ::arrow::Type::LARGE_LIST); + ICEBERG_UNWRAP_OR_FAIL(auto expected, + MakeRecordBatch(batches->schema(), std::string(json))); + EXPECT_TRUE(batch->Equals(*expected)) << batch->ToString(); + retained_batches.push_back(std::move(batch)); + } + auto next = batches->Next(); + ASSERT_TRUE(next.ok()) << next.status(); + EXPECT_EQ(next.ValueOrDie(), nullptr); + batches.reset(); + for (const auto& batch : retained_batches) { + EXPECT_TRUE(batch->ValidateFull().ok()) << batch->ValidateFull(); + } +} + +INSTANTIATE_TEST_SUITE_P(ResidualsAndProjections, FileScanTaskResidualPhysicalSchemaTest, + ::testing::Combine(::testing::Bool(), ::testing::Bool(), + ::testing::Bool())); + TEST_F(FileScanTaskReaderTest, OpenAppliesResidualFilterWithPositionDeletes) { ICEBERG_UNWRAP_OR_FAIL( auto data_file, @@ -840,6 +972,84 @@ TEST_F(FileScanTaskReaderTest, OpenRejectsResidualOnNestedColumn) { HasErrorMessage("which is not a top-level column")); } +TEST_F(FileScanTaskReaderTest, OpenAppliesNullPredicatesOnCompositeColumns) { + const std::vector, std::string>> cases = { + {struct_({SchemaField::MakeOptional(3, "value", int32())}), + R"([[1, null], [2, {"value": null}], [3, {"value": 10}], [4, null]])"}, + {list(SchemaField::MakeOptional(3, "element", int32())), + "[[1, null], [2, []], [3, [10]], [4, null]]"}, + {map(SchemaField::MakeRequired(3, "key", string()), + SchemaField::MakeOptional(4, "value", int32())), + R"([[1, null], [2, []], [3, [["a", 10]]], [4, null]])"}, + }; + for (const auto& [type, data] : cases) { + SCOPED_TRACE(type->ToString()); + table_schema_ = std::make_shared( + std::vector{SchemaField::MakeRequired(1, "id", int32()), + SchemaField::MakeOptional(2, "value", type)}); + projected_schema_ = + std::make_shared(std::vector{table_schema_->fields()[0]}); + VerifyResidualStream(Expressions::IsNull("value"), "[[1], [4]]", {}, true, data); + VerifyResidualStream(Expressions::NotNull("value"), "[[2], [3]]", {}, true, data); + VerifyResidualStream( + Expressions::And(Expressions::NotNull("value"), + Expressions::GreaterThan("id", Literal::Int(2))), + "[[3]]", {}, true, data); + } +} + +TEST_F(FileScanTaskReaderTest, OpenAppliesGeospatialNullPredicates) { + // Valid little-endian WKB POINT(1 2). + const std::array 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>{geometry(), geography()}) { + SCOPED_TRACE(type->ToString()); + table_schema_ = std::make_shared( + std::vector{SchemaField::MakeRequired(1, "id", int32()), + SchemaField::MakeOptional(2, "geom", type)}); + projected_schema_ = + std::make_shared(std::vector{table_schema_->fields()[0]}); + ::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)); + VerifyStream(&stream, is_null ? "[[2]]" : "[[1]]"); + } + } +} + TEST_F(FileScanTaskReaderTest, OpenWithDeletionVectorFiltersRows) { ICEBERG_UNWRAP_OR_FAIL( auto data_file, @@ -1066,6 +1276,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( + std::vector{SchemaField::MakeRequired(1, "id", int32()), + SchemaField::MakeOptional(4, "name", string())}, + 2); + auto old_schema = std::make_shared( + std::vector{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(std::vector{ + 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::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> 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> Finish(); private: + Status RecordPosition(const Type& type); + constexpr static int32_t kUnassignedFieldId = -1; int32_t current_field_id_ = kUnassignedFieldId; std::vector current_path_;