Repository navigation
fix(iceberg): Apply row lineage filters after synthesis - #38
Closed
Joe-Abraham wants to merge 3 commits into
Closed
Joe-Abraham wants to merge 3 commits into
Joe-Abraham wants to merge 3 commits into
Conversation
Reviewer's GuideThe PR fixes incorrect Iceberg lineage predicates by preventing scan-time evaluation against pre-synthesis values, evaluating them after row lineage is filled in, and combining that filtering with delete handling. It also makes file-position tracking robust across dropped rows and dynamic filters, supports internally appended columns and extraction pushdown, corrects row-count accounting, and adds broad regression coverage. Sequence diagram for post-synthesis Iceberg lineage filteringsequenceDiagram
participant Scan as IcebergSplitReader
participant Reader as FileReader
participant Filter as ScanSpec
participant Deletes as EqualityDeleteFileReader
participant Output as RowVector
Scan->>Filter: setFilterEnabled(false)
Scan->>Scan: projectFilterOnlyFilledColumns()
Scan->>Reader: read batch with row-number column
Reader-->>Output: raw rows
Scan->>Scan: fill _row_id and lineage columns
Scan->>Filter: applyFilter(lineage column)
Scan->>Deletes: applyDeletes(output, rowsToRemove)
Deletes-->>Scan: mark deleted rows
Scan->>Output: removeRows(output, rowsToRemove)
Scan-->>Scan: return rowsScanned
Flow diagram for Iceberg row lineage and position trackingflowchart TD
A["Read split batch"] --> B["Append row-number column when lineage is projected"]
B --> C["Synthesize _row_id or $target_table_row_id from file positions"]
C --> D["Apply post-read lineage filters"]
D --> E["Apply equality deletes to the shared removal bitmap"]
E --> F["Compact surviving rows"]
F --> G["Report rows read, including filtered and deleted rows"]
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey - I've found 2 issues
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location path="velox/connectors/hive/iceberg/tests/IcebergReadTest.cpp" line_range="1880-1889" />
<code_context>
+}
+
+// A deferred filter runs per output batch, on rows the reader has already read.
+TEST_F(IcebergReadTest, rowLineageFilterOverBatches) {
+ // The first two batches come back empty. An empty batch must not be reported
+ // as the end of the split, or the rows after it are lost.
+ assertRowLineageFilterOverBatches(
+ /*firstSelected=*/2 * kMultiBatchRowsPerBatch,
+ /*lastSelected=*/kMultiBatchNumRows - 1);
+
+ // A range ending inside the second batch leaves that one partially compacted.
+ assertRowLineageFilterOverBatches(
+ /*firstSelected=*/0,
+ /*lastSelected=*/kMultiBatchRowsPerBatch + kMultiBatchRowsPerBatch / 2 -
+ 1);
+}
</code_context>
<issue_to_address>
**suggestion (testing):** Add a batch-range scenario where the lineage filter drops all rows to validate end-of-split behavior under full post-read filtering.
To complete coverage of deferred filters and equality deletes across batches, please add a third variant where the `_row_id` range selects no rows (e.g., fully outside `[kMultiBatchFirstRowId, kMultiBatchFirstRowId + kMultiBatchNumRows - 1]`). This should assert that `rawInputRows` still counts all scanned rows, `outputRows` is zero, and that a fully empty batch neither terminates the split early nor skips subsequent batches.
Suggested implementation:
```cpp
// A deferred filter runs per output batch, on rows the reader has already read.
TEST_F(IcebergReadTest, rowLineageFilterOverBatches) {
// The first two batches come back empty. An empty batch must not be reported
// as the end of the split, or the rows after it are lost.
assertRowLineageFilterOverBatches(
/*firstSelected=*/2 * kMultiBatchRowsPerBatch,
/*lastSelected=*/kMultiBatchNumRows - 1);
// A range ending inside the second batch leaves that one partially compacted.
assertRowLineageFilterOverBatches(
/*firstSelected=*/0,
/*lastSelected=*/kMultiBatchRowsPerBatch + kMultiBatchRowsPerBatch / 2 -
1);
// A range fully outside the multi-batch row-id span selects no rows. Even
// with full post-read filtering, the reader must still scan all input rows
// and not terminate the split early on an empty batch.
assertRowLineageFilterOverBatches(
/*firstSelected=*/kMultiBatchNumRows,
/*lastSelected=*/kMultiBatchNumRows + kMultiBatchRowsPerBatch - 1);
}
```
If `assertRowLineageFilterOverBatches` does not yet validate the empty-selection case, it should be updated to:
1. Treat `firstSelected > (kMultiBatchNumRows - 1)` (or a non-overlapping [firstSelected, lastSelected] range) as "no rows selected".
2. In that case assert that:
- `rawInputRows` equals the total number of scanned rows across all batches.
- `outputRows` is zero.
- All batches are processed and the split is not terminated early due to an empty output batch.
You may need to adjust how `firstSelected`/`lastSelected` are mapped to `_row_id` (e.g., offset by `kMultiBatchFirstRowId`) to keep this new scenario fully outside the valid `[kMultiBatchFirstRowId, kMultiBatchFirstRowId + kMultiBatchNumRows - 1]` range if your helper uses absolute row ids rather than indices.
</issue_to_address>
### Comment 2
<location path="velox/connectors/hive/tests/FileConnectorUtilTest.cpp" line_range="523-516" />
<code_context>
+TEST_F(FileConnectorUtilTest, testFiltersConstantOverridingFileColumn) {
</code_context>
<issue_to_address>
**suggestion (testing):** Extend constant-over-file-column tests to include a NULL constant and corresponding filters.
To fully exercise this path, please add a variation where `c0` is a NULL constant with both `IsNull` and `IsNotNull` filters. This will verify that `testFilters` accepts/rejects splits based solely on the constant (ignoring file stats) in the NULL case, and that the "constant decides the column" rule holds consistently for both null and non-null constants.
Suggested implementation:
```cpp
// A constant on the scan spec decides the filter even when the data file
// carries the column: the constant is what the scan returns. Building the
// reader with the scan spec also leaves the constant's subtree out of
// 'typeWithId()', so reading its statistics would dereference null.
TEST_F(FileConnectorUtilTest, testFiltersConstantOverridingFileColumn) {
auto batch =
makeRowVector({"c0"}, {makeFlatVector<int64_t>(100, folly::identity)});
auto filePath = writeDataFile(batch);
// Non-null constant: file stats would keep the split, but the constant
// alone must determine acceptance/rejection.
auto scanSpec = std::make_shared<common::ScanSpec>("<root>");
// The file holds 0 through 99, so a filter tested against the statistics
// would keep the split for both constants below.
scanSpec->addField("c0", 0)->setConstantValue<int64_t>(
1'000, BIGINT(), pool_.get());
// Validate that filters are evaluated against the constant value only.
{
auto filterMap = common::SubfieldFilters{
{common::Subfield("c0"),
std::make_unique<common::BigintRange>(
/*lower=*/1'000,
/*upper=*/1'000,
/*lowerExclusive=*/false,
/*upperExclusive=*/false)}};
// For a constant of 1000 and a range [1000, 1000], the split must be
// accepted solely based on the constant (ignoring file stats).
testFilters(filePath, scanSpec, filterMap, /*expectAccepted=*/true);
}
{
auto filterMap = common::SubfieldFilters{
{common::Subfield("c0"),
std::make_unique<common::BigintRange>(
/*lower=*/1'001,
/*upper=*/1'001,
/*lowerExclusive=*/false,
/*upperExclusive=*/false)}};
// For a constant of 1000 and a range [1001, 1001], the split must be
// rejected solely based on the constant.
testFilters(filePath, scanSpec, filterMap, /*expectAccepted=*/false);
}
// NULL constant: ensure that "constant decides the column" holds for null
// as well, and that file stats are ignored.
auto nullConstantScanSpec = std::make_shared<common::ScanSpec>("<root>");
nullConstantScanSpec->addField("c0", 0)
->setConstantValue<int64_t>(std::nullopt, BIGINT(), pool_.get());
// IsNull filter must accept the split when the constant is NULL.
{
auto filterMap = common::SubfieldFilters{
{common::Subfield("c0"), common::isNull()}};
testFilters(
filePath, nullConstantScanSpec, filterMap, /*expectAccepted=*/true);
}
// IsNotNull filter must reject the split when the constant is NULL.
{
auto filterMap = common::SubfieldFilters{
{common::Subfield("c0"), common::isNotNull()}};
testFilters(
filePath, nullConstantScanSpec, filterMap, /*expectAccepted=*/false);
}
}
```
The above edit assumes:
1. `testFilters` is available in this test file with signature similar to `testFilters(filePath, scanSpec, filterMap, expectAccepted)` or close; adjust arguments/order to match the existing helper.
2. `SubfieldFilters` is the filter map type used elsewhere in this file (likely `std::unordered_map<common::Subfield, std::unique_ptr<common::Filter>>`); if the existing tests use a different alias or construction pattern, adapt the `filterMap` initialization accordingly.
3. `BigintRange`, `isNull()`, and `isNotNull()` exist in `common` namespace; if other tests in this file create filters via factory helpers (e.g. `common::between`, `common::nullFilter`), mirror that convention instead of using `BigintRange` directly.
4. The API for setting a NULL constant on `ScanSpec::Field` might differ (e.g. an overload taking `VectorPtr` or `Variant`); update the `setConstantValue` call for the NULL case to match how other tests in this file express a NULL constant.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
There was a problem hiding this comment.
Pull request overview
This PR fixes incorrect filtering and row-position metadata for Iceberg synthetic/lineage columns by deferring those filters until after the split reader synthesizes the final values (e.g. _row_id, _last_updated_sequence_number, and $target_table_row_id). It also improves split pruning correctness for constants and partition-key filters, and adds targeted tests to prevent regressions.
Changes:
- Add
ScanSpec::setFilterEnabled()to disable reader-side filtering while keeping filters available for pruning/pushdown, enabling correct post-synthesis filtering. - Update
IcebergSplitReaderto (a) project filter-only synthesized columns, (b) defer filters on synthesized columns tonext(), and (c) fix row-position and$target_table_row_idsizing under filtering/deletes. - Fix split-pruning behavior in
hive::testFilters()(constants override file columns; evaluate all partition-key filters) and add comprehensive tests across Iceberg and DWIO utilities.
Reviewed changes
Copilot reviewed 11 out of 11 changed files in this pull request and generated 1 comment.
Show a summary per file
| File | Description |
|---|---|
| velox/dwio/common/tests/TypeTests.cpp | Adds regression test for TypeWithId::create(type, scanSpec) preserving type shape/IDs when scan spec prunes file-read children. |
| velox/dwio/common/tests/ScanSpecTest.cpp | Adds coverage for moveAdaptationFrom() behavior when filters are deferred (reader filtering disabled). |
| velox/dwio/common/ScanSpec.h | Introduces setFilterEnabled() and related tree-cached-value reset plumbing; adds parent_ for root traversal. |
| velox/dwio/common/ScanSpec.cpp | Implements deferred-filter enable/disable propagation, cached-value resets, and delta-update reset behavior. |
| velox/connectors/hive/tests/FileConnectorUtilTest.cpp | Adds tests for constant-over-file-column pruning and for evaluating multiple partition-key filters. |
| velox/connectors/hive/iceberg/tests/IcebergReadTest.cpp | Adds extensive Iceberg lineage/deferred-filter correctness tests (including dynamic filters, multi-split, multi-batch, deletes). |
| velox/connectors/hive/iceberg/tests/IcebergDeletionVectorReadTest.cpp | Adds include needed for updated Iceberg test wiring. |
| velox/connectors/hive/iceberg/tests/CMakeLists.txt | Ensures IcebergDeletionVectorReadTest.cpp is built into the Iceberg test target. |
| velox/connectors/hive/iceberg/IcebergSplitReader.h | Adds deferred-filter machinery and helper declarations for projecting filter-only synthesized columns. |
| velox/connectors/hive/iceberg/IcebergSplitReader.cpp | Implements deferral and post-synthesis filtering, row-number injection changes, shared compaction for deferred filters + equality deletes, and target row-id fixes. |
| velox/connectors/hive/FileConnectorUtil.cpp | Fixes pruning semantics for constants and ensures all partition-key filters are evaluated. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
Joe-Abraham
force-pushed
the
fix/row_lineage_pushdown
branch
from
August 18, 2026 06:27
1b79daa to
193a225
Compare
Owner
Author
|
@soucery-ai summary |
Owner
Author
|
@soucery-ai review |
Owner
Author
|
@sourcery-ai review |
Owner
Author
|
@sourcery-ai summary |
There was a problem hiding this comment.
Hey - I've found 2 issues
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location path="velox/dwio/common/tests/ScanSpecTest.cpp" line_range="136-145" />
<code_context>
+TEST_F(ScanSpecTest, moveAdaptationFromDeferredFilter) {
</code_context>
<issue_to_address>
**suggestion (testing):** Consider adding coverage for resetDeltaUpdates() to ensure filters are re-enabled and cached hasFilter() state is updated correctly.
Currently only `moveAdaptationFromDeferredFilter` exercises the new behavior around filter enabling/disabling and cached `hasFilter()`. A targeted test for `resetDeltaUpdates()` would better cover this interaction: e.g., construct a ScanSpec with a child that has a delta update and a filter, call `setDeltaUpdate` and confirm `hasFilter()` becomes false, then call `resetDeltaUpdates()` and assert `hasFilter()` is true again and `applyFilter()` evaluates the filter. This helps catch regressions in how delta updates, `setFilterEnabled`, and cached `hasFilter()` interact.
Suggested implementation:
```cpp
// moveAdaptationFrom() skips a child that is constant on either side, because a
// filter on a constant was evaluated at split start. Not so for a column whose
// filtering the reader was told to skip: nothing evaluated its filter.
TEST_F(ScanSpecTest, moveAdaptationFromDeferredFilter) {
auto makeSpec = [] {
auto spec = std::make_shared<ScanSpec>("<root>");
spec->addField("c0", 0);
spec->addField("c1", 1);
return spec;
};
auto from = makeSpec();
from->childByName("c0")->setFilter(
std::make_shared<BigintRange>(10, 20, false));
}
TEST_F(ScanSpecTest, resetDeltaUpdatesReEnablesFilters) {
// Construct a ScanSpec with a single child that has a filter.
ScanSpec scanSpec("<root>");
auto* child = scanSpec.addField("c0", 0);
child->setFilter(std::make_shared<BigintRange>(10, 20, false));
scanSpec.resetCachedValues(true);
// Initially, the filter should be visible through hasFilter().
EXPECT_TRUE(scanSpec.hasFilter());
// Mark the child as a delta update; this should disable the filter and clear
// the cached hasFilter() state.
child->setDeltaUpdate(true);
scanSpec.resetCachedValues(true);
EXPECT_FALSE(scanSpec.hasFilter());
// After resetDeltaUpdates(), filters should be re-enabled and hasFilter()
// should be updated accordingly.
scanSpec.resetDeltaUpdates();
scanSpec.resetCachedValues(true);
EXPECT_TRUE(scanSpec.hasFilter());
// Verify that the filter is actually evaluated again via applyFilter().
// A value inside the range [10, 20] should pass the filter; a value outside
// the range should fail.
EXPECT_TRUE(child->applyFilter(15));
EXPECT_FALSE(child->applyFilter(25));
```
The above patch assumes the following APIs and behaviors on `ScanSpec` and its children:
1. `ScanSpec::addField(const std::string& name, int32_t index)` returns a pointer to a child `ScanSpec`.
2. `ScanSpec::hasFilter()` reflects whether any enabled filter is present in the subtree, and it is cached until `resetCachedValues(true)` is called.
3. `ScanSpec::resetDeltaUpdates()` clears any `setDeltaUpdate(true)` state and re-enables filters that were disabled as a result.
4. Child `ScanSpec` exposes:
- `void setDeltaUpdate(bool)`;
- `bool applyFilter(T value)` (or a similar API) that evaluates the installed `Filter` (`BigintRange` here) against a scalar value.
If the actual APIs differ in your codebase, adjust the test as follows:
- Replace `child->setDeltaUpdate(true);` with the correct way to mark a field as a delta update (e.g., `scanSpec.setDeltaUpdate(child, true);` or similar).
- Replace `child->applyFilter(15)` / `child->applyFilter(25)` with the appropriate call that forces evaluation of the filter (e.g., `child->filter()->testInt64(15, ...)`, or a helper method in `ScanSpecTest` that exercises the filter).
- If `resetCachedValues(true)` is not required to update the cached `hasFilter()` state after `setDeltaUpdate` / `resetDeltaUpdates`, you can remove or adjust those calls in the test to match the actual invalidation behavior.
</issue_to_address>
### Comment 2
<location path="velox/connectors/hive/iceberg/tests/IcebergReadTest.cpp" line_range="1427-1436" />
<code_context>
+TEST_F(IcebergReadTest, targetTableRowIdSynthesis) {
</code_context>
<issue_to_address>
**suggestion (testing):** Add a test where a filter on $target_table_row_id selects by child fields (e.g., row_position) to prove deferred filtering uses synthesized composite values, including with positional deletes.
The existing `targetTableRowIdSynthesis` test only checks that `$target_table_row_id` is synthesized and that an `IsNotNull` filter doesn’t prune splits too early. It doesn’t cover filters that actually constrain child fields of the composite (e.g., `row_position` or `spec_id`) and return a subset of rows.
Please add a scenario that:
- Projects `c0` and `$target_table_row_id`.
- Uses `assertTargetTableRowId` (or a similar helper) with a non-null filter that applies a `BigintRange` on `row_position` or `spec_id` via a `common::Filter` on the composite.
- Includes positional deletes so some row positions are skipped, and verifies the filter is evaluated against the synthesized row-id values, not pre-synthesis placeholders.
This will explicitly validate that deferred filters on `$target_table_row_id` work correctly with positional deletes and compacted batches.
Suggested implementation:
```cpp
TEST_F(IcebergReadTest, targetTableRowIdChildFieldFilterWithPosDeletes) {
// This test complements targetTableRowIdSynthesis by asserting that a filter
// on a child field of $target_table_row_id (row_position) is evaluated
// against the synthesized composite row-id values, including in the presence
// of positional deletes and compacted batches.
// Build a simple schema with a single data column and the synthesized
// $target_table_row_id lineage column.
const auto rowType = ROW(
{"c0"},
{BIGINT()});
// Create a small dataset with known row positions. We will later apply
// positional deletes to skip some positions:
//
// logical row index: 0 1 2 3 4
// c0 : 10 20 30 40 50
//
// We'll delete positions 1 and 3 (c0 = 20 and 40) via positional deletes.
auto data = makeRowVector(
{makeFlatVector<int64_t>({10, 20, 30, 40, 50})});
auto dataFilePath = TempFilePath::create();
writeIcebergDataFile(
*dataFilePath,
rowType,
{data},
/*format*/ IcebergFileFormat::kParquet);
// Create a positional delete file that removes row positions 1 and 3 from
// the data file.
auto deleteFilePath = TempFilePath::create();
writeIcebergPositionalDeleteFile(
*deleteFilePath,
*dataFilePath,
/*rowPositions*/ {1, 3});
// Build an Iceberg table/manifest that references the data file and the
// positional delete file. Reuse the standard helpers to ensure we exercise
// the normal Iceberg split + delete wiring.
auto tableMetadata = makeIcebergTableWithDeletes(
rowType,
{*dataFilePath},
{*deleteFilePath});
// Build a scan spec that:
// - projects c0 and $target_table_row_id
// - applies IsNotNull on $target_table_row_id
// - applies a BigintRange filter on the row_position child field so that we
// select only a subset of live rows.
auto scanSpec = std::make_shared<common::ScanSpec>("root");
// Project c0.
scanSpec->addProjectedField("c0");
// Project and filter on $target_table_row_id.row_position.
auto* rowIdSpec =
scanSpec->getOrCreateChild(IcebergMetadataColumnPaths::kTargetTableRowId);
scanSpec->addProjectedField(
IcebergMetadataColumnPaths::kTargetTableRowId);
// Add a non-null filter on the composite row-id so that synthesis is
// triggered and the composite value must be fully materialized.
rowIdSpec->setFilter(std::make_unique<common::IsNotNull>());
// Add a filter on the child field "row_position". We want to select only
// rows whose synthesized row_position is in [1, 3]. Because positions 1 and
// 3 are deleted, the filter should still be evaluated on the synthesized
// row-id values and only the remaining live rows in that range should be
// returned (positions 2 only).
auto* rowPositionSpec = rowIdSpec->getOrCreateChild("row_position");
rowPositionSpec->setFilter(
std::make_unique<common::BigintRange>(1, 3));
// Create a reader over the Iceberg table using the scan spec above.
auto reader =
makeIcebergReader(tableMetadata, rowType, scanSpec);
// Collect all output rows. We expect only the live rows whose synthesized
// row_position satisfies [1, 3]. Given our deletes at positions 1 and 3,
// only logical row with c0 = 30 (row_position = 2) should survive.
std::vector<RowVectorPtr> results;
readAll(reader.get(), results);
ASSERT_EQ(results.size(), 1);
auto result = results.front();
ASSERT_EQ(result->size(), 1);
// Verify that c0 matches the expected surviving value.
auto c0 = result->childAt(0)->asFlatVector<int64_t>();
ASSERT_EQ(c0->valueAt(0), 30);
// Verify that the synthesized $target_table_row_id values correspond to the
// expected row_position and that filtering was applied post synthesis.
//
// We reuse the existing helper (or a similar one) that validates both the
// structure and contents of the synthesized row-id.
assertTargetTableRowId(
result,
/*expectedRowPositions*/ {2},
/*expectedC0*/ {30});
}
// A filter on a lineage column the query does not select must still see the
// synthesized value. Such a column is left out of the reader output, so there
// is no slot to synthesize into until the split reader adds one back.
TEST_F(IcebergReadTest, rowLineageFilterOnUnprojectedColumn) {
auto dataFilePath = TempFilePath::create();
```
The above test uses helpers and constants that should be aligned with your existing codebase. You will likely need to:
1. **Reuse / adjust table + delete helpers:**
- Replace `writeIcebergDataFile`, `writeIcebergPositionalDeleteFile`, and `makeIcebergTableWithDeletes` with the actual helpers used in `targetTableRowIdSynthesis` (or similar tests) to construct data and positional delete files and to wire them into an Iceberg table/manifest.
- If `targetTableRowIdSynthesis` builds its table differently (e.g., via `makeIcebergSplit` or `makeIcebergTable`), mirror that pattern so row-id synthesis follows the same path.
2. **Align the metadata column path and row-id assertion helper:**
- Replace `IcebergMetadataColumnPaths::kTargetTableRowId` with whatever constant/string is used elsewhere in the file for `$target_table_row_id`.
- Adjust `assertTargetTableRowId(...)`’s arguments to match the existing helper’s signature (e.g., it might only take the `RowVectorPtr` and internally inspect `row_position` and, possibly, `spec_id`).
3. **Adjust reader construction and result collection:**
- Replace `makeIcebergReader(..., scanSpec)` and `readAll(reader.get(), results)` with the actual reader/convenience APIs used in the surrounding tests (for example, `assertRead`, `readIceberg`, or a helper that consumes all batches).
- Ensure the projected columns and scan spec wiring matches the rest of the file. In some setups, projection is configured at the reader factory rather than by calling `scanSpec->addProjectedField`.
4. **Row-position range and expectations:**
- If your positional delete writer or row-id synthesis logic uses zero-based versus one-based positions differently, adjust the `BigintRange(1, 3)` and the expected surviving positions in `assertTargetTableRowId` accordingly so that the test aligns with the semantics exercised in `targetTableRowIdSynthesis`.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Joe-Abraham
force-pushed
the
fix/row_lineage_pushdown
branch
from
August 18, 2026 07:42
193a225 to
6120394
Compare
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 12 out of 12 changed files in this pull request and generated no new comments.
Suppressed comments (1)
velox/dwio/common/ScanSpec.cpp:44
children_is astd::vector<std::shared_ptr<ScanSpec>>(seeScanSpec.h:539), but this code pushes astd::unique_ptr. This won’t compile; usestd::make_sharedhere to match the container type.
this->children_.push_back(std::make_unique<ScanSpec>(name));
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 12 out of 12 changed files in this pull request and generated no new comments.
Suppressed comments (1)
velox/connectors/hive/iceberg/tests/IcebergDeletionVectorReadTest.cpp:18
- This header is already included later in the same file, so this addition creates a duplicate include. Please keep only one
IcebergMetadataColumns.hinclude.
#include "velox/connectors/hive/iceberg/IcebergMetadataColumns.h"
Owner
Author
|
@sourcery-ai REVIEW |
Joe-Abraham
force-pushed
the
fix/row_lineage_pushdown
branch
from
August 19, 2026 05:13
6120394 to
738ebd8
Compare
Owner
Author
|
@sourcery-ai REVIEW |
Owner
Author
|
@sourcery-ai summary |
Joe-Abraham
force-pushed
the
fix/row_lineage_pushdown
branch
from
August 21, 2026 06:06
738ebd8 to
c1f794f
Compare
Joe-Abraham
force-pushed
the
fix/row_lineage_pushdown
branch
from
September 19, 2026 03:33
f2cb0c3 to
00b141e
Compare
Owner
Author
|
@sourcery-ai review |
There was a problem hiding this comment.
Hey - I've found 1 issue
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location path="velox/connectors/hive/iceberg/IcebergSplitReader.cpp" line_range="742" />
<code_context>
+ }
+ // A previous split may have deferred a column this one does not synthesize,
+ // where the reader has to filter it again.
+ childSpec->setFilterEnabled(!outputIndex.has_value());
+ if (outputIndex.has_value()) {
+ deferredFilters_.push_back({childSpec, *outputIndex});
</code_context>
<issue_to_address>
**issue (bug_risk):** Concurrent split readers sharing the same ScanSpec race while toggling `filterDisabled_` during `prepareSplit()`. One split can re-enable a lineage filter after another split has deferred it, causing the latter's reader to filter synthesized columns using stored or null values and drop or retain the wrong rows.
**Triggers:** When multiple splits are prepared concurrently, such as with multiple drivers or split preloading, and some splits synthesize a lineage column while others do not.
**Suggested fix:** Keep filter enablement state per reader or construct an ownership-preserving ScanSpec snapshot for each split instead of mutating the shared ScanSpec during reader preparation.
</issue_to_address>Sourcery assessment
Needs a human reviewer. 1 finding to address first, and the change substantially alters Iceberg scan filtering, row compaction, deletion handling, and synthesized lineage values. If it is wrong, rows could be incorrectly omitted or retained and a MERGE path could use incorrect synthesized row IDs to update or delete records; reverting would not undo those writes.
Blocking findings: velox/connectors/hive/iceberg/IcebergSplitReader.cpp:742
| } | ||
| // A previous split may have deferred a column this one does not synthesize, | ||
| // where the reader has to filter it again. | ||
| childSpec->setFilterEnabled(!outputIndex.has_value()); |
There was a problem hiding this comment.
issue (bug_risk): Concurrent split readers sharing the same ScanSpec race while toggling filterDisabled_ during prepareSplit(). One split can re-enable a lineage filter after another split has deferred it, causing the latter's reader to filter synthesized columns using stored or null values and drop or retain the wrong rows.
Triggers: When multiple splits are prepared concurrently, such as with multiple drivers or split preloading, and some splits synthesize a lineage column while others do not.
Suggested fix: Keep filter enablement state per reader or construct an ownership-preserving ScanSpec snapshot for each split instead of mutating the shared ScanSpec during reader preparation.
Joe-Abraham
force-pushed
the
fix/row_lineage_pushdown
branch
6 times, most recently
from
September 28, 2026 06:57
1721212 to
77788ad
Compare
Owner
Author
|
@sourcery-ai review |
Joe-Abraham
force-pushed
the
fix/row_lineage_pushdown
branch
5 times, most recently
from
October 1, 2026 13:00
2a00892 to
867670d
Compare
Joe-Abraham
force-pushed
the
fix/row_lineage_pushdown
branch
from
October 5, 2026 05:44
867670d to
dcbf7ec
Compare
Owner
Author
|
@sourcery-ai review |
Owner
Author
|
@sourcery-ai summary |
Owner
Author
|
@sourcery-ai guide |
Comment on lines
+431
to
+444
| if (readerProducedType_ != nullptr && | ||
| splitReaderOutputType->size() > readerProducedType_->size()) { | ||
| // The split reader appended columns, e.g. an unselected Iceberg | ||
| // equality-delete or lineage column. next() allocates the output from | ||
| // 'readerProducedType_', so it has to grow by the same columns. | ||
| auto names = readerProducedType_->names(); | ||
| auto types = readerProducedType_->children(); | ||
| for (auto i = readerProducedType_->size(); | ||
| i < splitReaderOutputType->size(); | ||
| ++i) { | ||
| names.push_back(splitReaderOutputType->nameOf(i)); | ||
| types.push_back(splitReaderOutputType->childAt(i)); | ||
| } | ||
| readerProducedType_ = ROW(std::move(names), std::move(types)); |
| const folly::F14FastMap<std::string, std::string>& fileReadOps) { | ||
| // The file schema extension and createReader() below read | ||
| // 'readerOutputType_'. | ||
| projectFilterOnlyFilledColumns(); |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
A filter on
_row_idor_last_updated_sequence_numberreturned the wrong rows when the engine pushed it into the scan, which Presto does withiceberg.pushdown_filter_enabled. For a three-row data file whose manifest entry carriesfirst_row_id = 100:returned no rows instead of two. The reader evaluated the filter on the values stored in the file, and
_row_idis synthesized after the read. A file that stores no lineage column reads as a null constant, and split pruning skipped it. A file an UPDATE rewrote stores the column only for the rows it carried over, and the reader kept or dropped rows by the stored nulls.The split reader now applies these filters after it fills in the values, in the same pass as equality deletes. The filter stays on the scan spec; the new
ScanSpec::setFilterEnabled()switches it off in the reader for a split that synthesizes the column, so dynamic filters still land on it. A lineage column the query filters on but does not select gets an appended reader output slot, the way unselected equality-delete columns already do. Appending a column with extraction pushdown on used to crash withTrying to access non-existing child in RowVector; the output vector is now allocated from the appended type.With equality deletes, the split reader's
next()now returns the number of rows it read, not the number left after the deletes, socompletedRows_andrawInputRowsinclude the deleted rows. Reader-side filters already count rows this way.For a split that synthesizes a lineage column, a filter on it prunes no files or row groups. A dynamic filter that arrives after a split was preloaded does not reach that split when its lineage column is a constant; the join still applies it, so only performance is affected.
The reader now adds the row-number column whenever
_row_idor$target_table_row_idis projected, and takes every file position from it. Previously it added the column only when a filter, random skip, or positional delete was present at split start, and otherwise counted positions on from the last batch. A dynamic filter on another column arriving mid-split made the reader drop rows while that count kept going, so_row_idcame out wrong. The bug predates this PR, but a filter on_row_idused to force the column and no longer does. The cost is one extra row-number column for every scan that projects either lineage column.Test plan: each lineage predicate runs as a scan subfield filter, as a remaining filter, and as a filter node above an unfiltered scan, and must return the same rows.
rowLineageDynamicFilterMidSplitadds a dynamic filter on a data column between two batches of one split and checks that_row_idstill matches the file positions.End-to-end tests run on the Presto native worker in prestodb/presto#28287 (
TestIcebergV3RowLineage), each withiceberg.pushdown_filter_enabledon and off. They cover lineage predicates on inherited, stored, and mixed stored-and-inherited values, before and after compaction, with deleted rows, combined with regular-column predicates, and on V2 tables and V2-to-V3 upgrades, plus split pruning.Summary by Sourcery
Apply Iceberg lineage filters after row lineage synthesis while preserving correct file positions, delete handling, and scan statistics.
Bug Fixes:
_row_id,_last_updated_sequence_number, and$target_table_row_idreturn the correct rows.Enhancements:
Tests: