[ahmedabu98]

Fixed some bugs in the batch and CDC sources:

• CDC reads used to crash when filtering on a column that isn't selected, when the range has merge-on-read deletes (e.g. keep: [id] and filter: "amount > 10")
• regular batch reads used to crash for tables with equality delete files on unselected columns (e.g. equality delete on id = 123 and keep: [amount]).
• CDC reads for table with equality deletes on a nested field, when that field isn't selected, used to return every row as deleted (e.g. equality delete on top.nested = 'xyz' with keep: [id] used to return every row as deleted)
• CDC equality-delete pushdown on nested fields used to crash and fall back to slower path

[github-actions]

📋 Review Summary

This PR adds constant value maps (idToConstant) for partition identity columns and v3 row-lineage metadata (_row_id and _last_updated_sequence_number) to ReadUtils and wires them through ScanTaskReader and CdcReadUtils. It also fixes BeamBaseIncrementalChangelogScan to restore firstRowId from manifest entries when planning changelog tasks.

🔍 General Feedback

• The core row-lineage and partition constant wiring is clean and well-tested across Parquet, Avro, and ORC.
• However, there are a few edge cases around row-lineage synthesis on deleted rows, row-metadata ordering in CdcOutputUtils, and ScanTaskReader partition constants that should be addressed.

---

💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.

💬 Inline Findings

• sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/CdcReadUtils.java (line 225): 🟠 High: CdcOutputUtils.outputRow short-circuits and returns dataAndRowMetadata unchanged when metadataColumns contains only row metadata columns (_row_id, _last_updated_sequence_number): if (metadataColumns.isEmpty() || metadataColumns.stream().allMatch(IcebergCdcMetadataColumns::isRowMetadataColumn)) { return dataAndRowMetadata; }.

When a user configures withMetadataColumns(ImmutableList.of("_last_updated_sequence_number", "_row_id")) (without any commit/change metadata columns), outputSchema has fields [..., _last_updated_sequence_number, _row_id]. However:

1. In the unidirectional path (applyColumnPruning = true), DeleteReader.fileProjection appends missing _pos after the requested columns, and CdcReadUtils line 229 then appends _row_id after _pos (or after _last_updated_sequence_number even if _row_id was already requested first), producing a Record whose schema does not match projectedBeamRowSchema or outputBeamRowSchema.
2. In the bidirectional path (applyColumnPruning = false) or when _last_updated_sequence_number is requested before _row_id, CdcOutputUtils.outputRow returns row directly with schema [..., _last_updated_sequence_number, _row_id] (from projectedBeamRowSchema), or if CdcReadUtils reorders them, StructProjection / icebergRecordToBeamRow may mismatch with outputSchema.

Ensure that when _row_id is added solely as an internal dependency for computing _last_updated_sequence_number, it is stripped out when projecting to projectedBeamRowSchema / outputBeamRowSchema, and that CdcOutputUtils.outputRow preserves user-configured column ordering even when only row metadata columns are requested.

• sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/CdcReadUtils.java (line 236): 🟠 High: When reading deleted rows (DELETED_ROWS or DELETED_FILE) with _row_id or _last_updated_sequence_number requested, deleteReader filters the base DataFile records before RowLineage.synthesizeRowLineage wraps the CloseableIterable at line 255.

Because DeleteReader wraps the underlying reader iterable and filters out non-deleted rows, SynthesizeRowLineageIterator only sees the surviving (deleted) records and increments rowPosition as 0, 1, 2, ... for each emitted deleted record rather than using the record's actual physical row position (_pos) in the data file. As a result, for any row deleted at physical position p > 0 (via position or equality deletes), _row_id will be synthesized as firstRowId + emittedIndex instead of firstRowId + _pos.

Either apply RowLineage.synthesizeRowLineage to records before passing records to deleteReader.read(records) (for the non-Parquet path), or read _pos from the record when available instead of using an invocation counter.

• sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ScanTaskReader.java (line 117): 🟡 Medium: ReadUtils.idToConstant(table, currentTask, project) populates partition identity constants (PartitionUtil.constantsMap(task, ...)). However, ScanTaskReader passes idToConstant only to the Parquet reader builder (.reuseContainers().withNameMapping(...).idToConstant(idToConstant)) and does not pass .constants(idToConstant) to ORC.read(...) or Avro.read(...). Additionally, ScanTaskReader still manually overwrites partition columns after reading in advance() (lines 169-175), whereas RowLineage.synthesizeRowLineage is not applied for Avro/ORC in ScanTaskReader if row-lineage columns are ever projected there. Passing .constants(idToConstant) to ORC and Avro builders keeps format handling consistent with CdcReadUtils.

• sdks/java/io/iceberg/src/main/java/org/apache/iceberg/BeamBaseIncrementalChangelogScan.java (line 175): 🟡 Medium: restoreFirstRowId uses DataFiles.builder(entry.file().spec()).copy(entry.file()).withFirstRowId(entry.file().firstRowId()).build() when entry.file().firstRowId() != null. Note that DataFiles.Builder.copy(DataFile) in Iceberg does not copy dataSequenceNumber or fileSequenceNumber (which are set on BaseFile by the manifest reader), so rebuilding the DataFile via DataFiles.builder(...).copy(...) will reset dataSequenceNumber() and fileSequenceNumber() to null (or -1), breaking _last_updated_sequence_number synthesis in ReadUtils.idToConstant (task.file().dataSequenceNumber()).

Please verify whether DataFiles.Builder.copy preserves dataSequenceNumber and fileSequenceNumber in the Iceberg version used here; if entry.file() already has firstRowId() != null, why does entry.file() need to be rebuilt at all, or if copy() drops firstRowId elsewhere in ManifestGroup, ensure dataSequenceNumber is preserved.

[github-actions]

Assigning reviewers:

R: @chamikaramj for label iceberg.
Backup reviewer: Abacn (if the main reviewer is unresponsive, you can escalate to them by tagging their username)

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

[ahmedabu98]

AI review comment is hallucinating.. there's no such synthesizeRowLineage or restoreFirstRowId or ReadUtils.idToConstant methods in this change