Repository navigation
Updates Delta CDC read to fail when schemas are different at... - #40868
Conversation
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
There was a problem hiding this comment.
📋 Review Summary
This PR adds data-schema compatibility validation between the resolved start and end versions in Delta Lake CDC reads (DeltaIO.ReadChanges and CreateCDCReadTasksDoFn). The implementation properly distinguishes actual data and column-mapping changes from benign metadata-only updates (such as column comments), and includes thorough unit and integration tests.
🔍 General Feedback
- The recursive comparison in
DeltaIO.hasSameDataSchemaandhasSameColumnMappingaccurately accounts for field names, ordering, nullability, nested types (StructType,ArrayType,MapType), and Delta column mapping metadata (physicalNameandid). - Performing the check during both pipeline expansion (
DeltaIO.ReadChanges.expand) and task planning (CreateCDCReadTasksDoFn.processElement) ensures fast failure at graph construction time as well as runtime safety if the table evolves when the end boundary is open.
💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.
ahmedabu98
left a comment
ahmedabu98
left a comment
There was a problem hiding this comment.
Isn't it fine to read a range where a nullable column was added? older snapshots should assign null values to missing columns
| * example, column comments, column default values, or identity column high-water marks (which | ||
| * writers update whenever they insert rows into the table) may differ between the two schemas. | ||
| */ | ||
| static boolean hasSameDataSchema(DataType left, DataType right) { |
There was a problem hiding this comment.
Can we create a separate util class for this?
| if (startSchema == null || endSchema == null) { | ||
| throw new IllegalStateException("Table schema is null."); | ||
| } | ||
| if (!DeltaIO.hasSameDataSchema(startSchema, endSchema)) { |
There was a problem hiding this comment.
Do we need to check for schema changes within the range? e.g. if a column was added then dropped, this comparison wouldn't catch that
There was a problem hiding this comment.
Done. Also seems like this has negligible performance impact since there's no additional I/O operations needed to perform this check for the commit range.
|
May also need to check that partition spec hasn't changed (can do in a future PR) |
|
Assigning reviewers: R: @ahmedabu98 for label delta. Note: If you would like to opt out of this review, comment Available commands:
The PR bot will only process comments in the main thread (not review comments). |
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #40868 +/- ##
===============================
===============================
☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
github-actions
Bot
added
size/l
chamikaramj
left a comment
chamikaramj
left a comment
There was a problem hiding this comment.
Thanks. PTAL.
Isn't it fine to read a range where a nullable column was added? older snapshots should assign null values to missing columns.
Yeah, updated to allow column additions for nullable fields.
May also need to check that partition spec hasn't changed (can do in a future PR)
Will do this in a future PR.
| if (startSchema == null || endSchema == null) { | ||
| throw new IllegalStateException("Table schema is null."); | ||
| } | ||
| if (!DeltaIO.hasSameDataSchema(startSchema, endSchema)) { |
There was a problem hiding this comment.
Done. Also seems like this has negligible performance impact since there's no additional I/O operations needed to perform this check for the commit range.
| * example, column comments, column default values, or identity column high-water marks (which | ||
| * writers update whenever they insert rows into the table) may differ between the two schemas. | ||
| */ | ||
| static boolean hasSameDataSchema(DataType left, DataType right) { |
...start and end.
Only fails when schemas are actually different, trivial changes metadata only changes, that does not actually change the schema are ignored.
This fixes #40853.