fix(debezium): backfill Postgres TOAST columns on a copy so the merge result survives - #19749
Conversation
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #19749 +/- ##
============================================
- Coverage 77.96% 77.84% -0.13%
- Complexity 33458 33492 +34
============================================
Files 2539 2539
Lines 140939 141367 +428
Branches 17012 17081 +69
============================================
+ Hits 109890 110050 +160
- Misses 23388 23657 +269
+ Partials 7661 7660 -1
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! This PR removes the updatedRecord == newerAvroRecord identity shortcut in HoodieAvroRecordMerger.merge so in-place payload backfills (e.g. PostgresDebeziumAvroPayload TOAST) survive, and adds a v6 MOR regression test. The fix addresses a real correctness bug; one behavioral side effect of always rebuilding the result is worth double-checking in the inline comment. Please take a look at the inline comment, and this should be ready for a Hudi committer or PMC member to take it from here.
|
@danny0405 could you help check why Flink test fails on this one? Any suggestions on fixing the issue? |
|
The red job is The one real failure is: (both COW parameterizations, all 4 surefire retries) Why it failsIt isn't really Flink-specific - removing the shortcut also drops the
With the shortcut gone we fall into the rebuild, whose In other words, the shortcut was not purely an optimization - it was the only thing propagating Suggested fixKeep the identity branch, but rebuild the data from the merged Avro record while preserving everything else if (updatedRecord == newerAvroRecord) {
// Some payloads (e.g. PostgresDebeziumAvroPayload's TOAST backfill) merge by mutating the
// incoming Avro record in place and returning the same reference, so `newer`'s engine-native
// record can be stale. Refresh the data from the Avro result, but keep everything else that
// `newer` carried - in particular the D/-U operation, which the write path relies on.
return new BufferedRecord<>(newer.getRecordKey(), newer.getOrderingValue(),
recordContext.convertAvroRecord(updatedRecord), newer.getSchemaId(), newer.getHoodieOperation());
}Verified locally on this branch (
Two side notes while you're in there:
|
| return newer; | ||
| } | ||
| // Do not short-circuit to `newer` when updatedRecord == newerAvroRecord. Some payloads | ||
| // (e.g. PostgresDebeziumAvroPayload's TOAST backfill via mergeToastedValuesIfPresent) |
There was a problem hiding this comment.
can we fix the payload instead of the merger, either:
- always make the payload immutable and create new one when backfill modifications are needed;
- impl
#equalsfor all kinds of payloads for more detailed equation check.
so that we avoid perf regression for existing payloads.
danny0405
left a comment
danny0405
left a comment
There was a problem hiding this comment.
blocked a little while for clarification: https://github.com/apache/hudi/pull/19749/changes#diff-cbec5c19fad65e7c6ce81be8e9f73170bf9cc8dcd9ad22349ed89d76cb7f5973R75
|
@lokeshj1703 @danny0405 - I tried Danny's first suggestion (make the payload immutable and return a new record on backfill) locally, and it works: it fixes the TOAST bug, keeps the Flink test green, and leaves the merger hot path untouched so there is no perf regression for existing payloads. Patch below if you want to drop it on the branch. This supersedes the merger-side suggestion in my earlier comment - Danny's approach is better and I withdraw that one. The changeRevert // both combineAndGetUpdateValue overloads
if (insertOrDeleteRecord.isPresent()) {
return Option.of(mergeToastedValuesIfPresent(insertOrDeleteRecord.get(), currentValue));
}
return insertOrDeleteRecord;/**
* Returns the incoming record with any TOASTed column backfilled from {@code currentRecord}, or
* {@code incomingRecord} itself when there is nothing to backfill.
*
* <p>The backfill is applied to a copy rather than in place: record mergers treat "the payload
* handed back the same reference" as "the payload changed nothing" and skip rebuilding the
* engine-native record from the Avro result, which would silently drop the backfill. The copy is
* only allocated once a TOASTed column is actually found, so records without a sentinel are
* unaffected.
*/
private IndexedRecord mergeToastedValuesIfPresent(IndexedRecord incomingRecord, IndexedRecord currentRecord) {
List<Schema.Field> fields = incomingRecord.getSchema().getFields();
GenericRecord incoming = (GenericRecord) incomingRecord;
GenericRecord merged = null;
for (Schema.Field field : fields) {
// There are only four avro data types that have unconstrained sizes, which are
// NON-NULLABLE STRING, NULLABLE STRING, NON-NULLABLE BYTES, NULLABLE BYTES
if (incoming.get(field.name()) != null
&& (containsStringToastedValues(incomingRecord, field) || containsBytesToastedValues(incomingRecord, field))) {
if (merged == null) {
merged = new GenericData.Record(incomingRecord.getSchema());
for (Schema.Field f : fields) {
merged.put(f.pos(), incoming.get(f.pos()));
}
}
merged.put(field.name(), ((GenericRecord) currentRecord).get(field.name()));
}
}
return merged == null ? incomingRecord : merged;
}The copy-on-write bit matters for Danny's perf point: a record with no TOAST sentinel allocates nothing and still takes the merger shortcut, so the only rows that change behaviour are the ones that actually carry Verified locallyJDK 11,
Compiles clean with checkstyle enabled. On the second suggestion (
|
… result survives On a table-version-6 MOR table using PostgresDebeziumAvroPayload, an unchanged Postgres TOAST column (emitted as __debezium_unavailable_value) leaked to readers instead of being backfilled from the prior value. The payload backfilled the sentinel by mutating the incoming Avro record in place and returning the same reference. HoodieAvroRecordMerger.merge has an identity shortcut (updatedRecord == newerAvroRecord returns the engine-native newer record), so the merge result was skipped and the backfill dropped. Backfill onto a copy instead: mergeToastedValuesIfPresent returns a new record when a TOASTed column is filled, and the same reference otherwise. The copy is allocated only when a sentinel is found, so records without one keep the merger fast-path. The merger is left unchanged, so other payloads and the HoodieOperation marker it preserves are unaffected. Add TestPostgresDebeziumToastV6ReadMerge, red without the fix and green with it.
a0bed56 to
b093faf
Compare
lokeshj1703
changed the title
7 participants
Describe the issue this Pull Request addresses
closes #19748
Summary and Changelog
On a table-version-6 MOR table using
PostgresDebeziumAvroPayload, an unchanged Postgres TOAST column (emitted as the sentinel__debezium_unavailable_value) leaked to readers instead of being backfilled from the prior value.The payload backfills by mutating the incoming Avro record in place and returning the same reference.
HoodieAvroRecordMerger.mergehad an identity shortcutif (updatedRecord == newerAvroRecord) return newer;that returned the engine-nativenewerrecord, which never received the in-place mutation, so the merge result was discarded. This removes the shortcut so the result is always rebuilt fromupdatedRecord.There are two ways to fix this. This PR takes the merger-side approach, which also protects any other payload that mutates the incoming record in place and returns the same reference. The alternative, #19280, changes
PostgresDebeziumAvroPayloadto return a new record on backfill (preserving the merger shortcut). Only one of the two is needed; opening this to compare the approaches.Adds
TestPostgresDebeziumToastV6ReadMerge, which is red without the fix (read returns the raw sentinel) and green with it.Impact
Correctness is restored for payloads that backfill by mutating the incoming record in place. Trade-off: the removed shortcut was a general fast-path for the common "newer record wins" case of every CUSTOM merge-mode payload (not just Debezium), so the result is now always rebuilt via
convertAvroRecord+BufferedRecords.fromEngineRecord. That adds a per-record avro-to-engine round-trip on the CUSTOM-mode merge path (v6 MOR and any custom-payload table), used by both snapshot reads and compaction. The payload-side alternative #19280 avoids this by returning a new record only when a backfill occurs; that is the main reason to prefer one approach over the other.Risk Level
low. Behavior change is confined to the record merge path and covered by a new functional test.
Documentation Update
none
Contributor's checklist