fix(debezium): keep Postgres toasted columns on MOR read (v6/v8)#19280
fix(debezium): keep Postgres toasted columns on MOR read (v6/v8)#19280linliu-code wants to merge 1 commit into
Conversation
linliu-code
changed the title
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! The PR removes an incorrect short-circuit in HoodieAvroRecordMerger#merge so that a payload's in-place Avro mutation (e.g. PostgresDebeziumAvroPayload filling TOASTed columns from the base value) survives the merge on MoR table versions 6/8, with a new functional test covering versions 6, 8 and 9. The correctness reasoning holds up; one behavioral edge case around record operation metadata is worth double-checking in the inline comment. Please take a look at any inline comments, and this should be ready for a Hudi committer or PMC member to take it from here. A couple of minor consistency nits in the new test file; the merger fix and its comment look clean.
2b9be8a to
ece2039
Compare
linliu-code
changed the title
ece2039 to
d30ce02
Compare
linliu-code
changed the title
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! The PR makes PostgresDebeziumAvroPayload return a fresh (shallow-copied) record only when it fills a TOASTed column, so the merger's identity short-circuit no longer discards the fill on MOR reads for table versions 6/8. The core mechanism looks sound; one edge case around the record's concrete type is worth double-checking in the inline comment. Please take a look at any inline comments, and this should be ready for a Hudi committer or PMC member to take it from here. One minor naming nit in the test; the production change and its Javadoc are clean.
| && (containsStringToastedValues(incomingRecord, field) || containsBytesToastedValues(incomingRecord, field))) { | ||
| ((GenericRecord) incomingRecord).put(field.name(), ((GenericData.Record) currentRecord).get(field.name())); | ||
| if (filled == null) { | ||
| filled = new GenericData.Record((GenericData.Record) incomingRecord, false); |
There was a problem hiding this comment.
🤖 Could incomingRecord here ever be a SerializableIndexedRecord rather than a GenericData.Record? BaseAvroPayload.getRecord() returns a SerializableIndexedRecord when the stored record's schema isn't reference-equal to the requested schema, and in the avro file-group-reader path AvroRecordContext.convertToAvroRecord just casts newer.getRecord(). If so, new GenericData.Record((GenericData.Record) incomingRecord, false) would throw a ClassCastException on a toasted fill. The old in-place code used only the GenericRecord interface for STRING columns, so it was immune — building the copy from incomingRecord.getSchema() and copying via the interface might be safer.
| // 2. Newer log records at LSN 20, each carrying the toasted sentinel in one column: | ||
| // id=1 -> name updated, `city` toasted (must keep base "NYC") | ||
| // id=2 -> name toasted (must keep base "bob"), `city` updated | ||
| val update = Seq( |
There was a problem hiding this comment.
🤖 nit: val update is a singular/verb form for what is actually a sequence of records — could you rename it to val updates or val logRecords to make it read as a collection?
On a Merge-on-Read table written with PostgresDebeziumAvroPayload, a snapshot read could return the raw sentinel string `__debezium_unavailable_value` for TOASTed columns (large STRING/BYTES columns Debezium could not capture in an UPDATE) instead of the real value. This affected pre-9 table versions (6 and 8); version 9 resolves toasted values through the FILL_UNAVAILABLE partial-update handler and was unaffected. The payload fills toasted columns from the base value by mutating the incoming avro record in place and returning that same reference. HoodieAvroRecordMerger has an identity short-circuit: when the payload returns the same object it was handed, the merger returns the engine-backed `newer` record rather than rebuilding. Since `newer` is a separate avro conversion of the same row, the in-place fill was dropped and the sentinel survived into the merged result. Make the payload return a new record only when it actually fills a toasted column (a lazy shallow copy), leaving the record untouched otherwise. The merger's identity short-circuit then still fires for every normal merge, and only the toasted case falls through to rebuild, so the fill survives. The merger is unchanged. TestDebeziumToastedValue writes a base row then a toasted-update delta on a MOR table and asserts the snapshot read keeps the prior base value, parameterized over table versions 6, 8 and 9.
d30ce02 to
02e915d
Compare
3 participants
Describe the issue this Pull Request addresses
On a Merge-on-Read table written with
PostgresDebeziumAvroPayload, a snapshot read could return the raw sentinel string__debezium_unavailable_valuefor TOASTed columns (large STRING/BYTES columns Debezium could not capture in an UPDATE) instead of the real value. This affected pre-9 table versions (6 and 8); version 9 resolves toasted values through theFILL_UNAVAILABLEpartial-update handler and was unaffected.Root cause: the payload fills toasted columns from the base value by mutating the incoming avro record in place and returning that same reference.
HoodieAvroRecordMergerhas an identity short-circuit — when the payload returns the same object it was handed, the merger returns the engine-backednewerrecord rather than rebuilding. Sinceneweris a separate avro conversion of the same row, the in-place fill was dropped and the sentinel survived into the merged result.Summary and Changelog
Make the payload return a new record only when it actually fills a toasted column (a lazy shallow copy), leaving the record untouched otherwise. The merger's identity short-circuit then still fires for every normal merge, and only the toasted case falls through to rebuild, so the fill survives.
HoodieAvroRecordMergeris unchanged.PostgresDebeziumAvroPayload.mergeToastedValuesIfPresentnow returns anIndexedRecord(a distinct shallow copy when it fills a toasted column, else the original) instead of mutating in place; bothcombineAndGetUpdateValueoverloads return that record.TestDebeziumToastedValue, a parameterized functional test over table versions 6, 8 and 9.Impact
Fixes wrong data (sentinel placeholder leaking into query results) for Postgres Debezium toasted columns on MOR snapshot reads for table versions 6 and 8. No public API change. The common merge path is unchanged (the merger's zero-copy short-circuit still applies); only a toasted-column merge does one extra shallow record copy.
Risk Level
low
The change is confined to
PostgresDebeziumAvroPayload; the record merger is untouched. Covered by the new parameterized functional test across table versions 6, 8 and 9.Documentation Update
none
Contributor's checklist