Repository navigation
feat(data): enforce residual filters in the task reader - #1065
kamcheungting-db wants to merge 3 commits into
Conversation
| /// 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}; |
There was a problem hiding this comment.
Defaulting apply_residuals to false for the serial reader while the parallel reader always enables it means a caller who switches readers, or a future serial-path caller who forgets to set this, silently gets unfiltered rows instead of an error.
| ICEBERG_ASSIGN_OR_RAISE(auto bound, | ||
| Binder::Bind(*table_schema_, residual, case_sensitive_)); | ||
| ICEBERG_ASSIGN_OR_RAISE(auto field_id_set, | ||
| ReferenceVisitor::GetReferencedFieldIds(bound)); |
There was a problem hiding this comment.
Binding can fold away references, but line 372 binds the original residual again. With an unprojected required flag, notNull(flag) drops its ID here and then fails to bind. Preserve the required references or carry the simplified expression through; please also cover a partially folded AND.
| ICEBERG_ASSIGN_OR_RAISE(auto projection_context, | ||
| ProjectionContext::Make(*required_schema, *projected_schema_, | ||
| project_batch_function)); |
There was a problem hiding this comment.
This new no-delete residual path still builds projection state from the logical schema. With large_list output, the stream declares list while the all-matching path returns the reader's physical array. Please use the reader's physical schema, as in #1062, and test both all-matching and filtered residuals.
| std::vector<int32_t> kept; | ||
| auto keep = [&](int64_t index) -> Status { | ||
| ICEBERG_RETURN_UNEXPECTED(row->Reset(index)); | ||
| ICEBERG_ASSIGN_OR_RAISE(bool matches, residual_evaluator_->Evaluate(*row)); |
There was a problem hiding this comment.
Top-level isNull(s) and notNull(s) are accepted for an optional struct, but evaluation calls StructLikeAccessor::GetLiteral, which rejects non-primitive types before checking null. Please evaluate null predicates from the scalar/validity and add a composite-column case.
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.
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).
6a278cc to
87764af
Compare
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.