rahil-c commented on PR #19749:
URL: https://github.com/apache/hudi/pull/19749#issuecomment-5434203831
@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 change
Revert `HoodieAvroRecordMerger` to master (keep the `updatedRecord ==
newerAvroRecord` shortcut) and change only `PostgresDebeziumAvroPayload`:
```java
// both combineAndGetUpdateValue overloads
if (insertOrDeleteRecord.isPresent()) {
return Option.of(mergeToastedValuesIfPresent(insertOrDeleteRecord.get(),
currentValue));
}
return insertOrDeleteRecord;
```
```java
/**
* 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
`__debezium_unavailable_value`.
### Verified locally
JDK 11, `-Dspark3.5 -Dscala-2.12 -Dflink2.2`, merger reverted to master:
| Test | Result |
|---|---|
| `TestPostgresDebeziumToastV6ReadMerge` (this PR's test) | `Tests:
succeeded 1, failed 0` |
| `TestFlinkWriteClientFunctional` | `Tests run: 9, Failures: 0, Errors: 0` |
| `TestPostgresDebeziumAvroPayload` | 8/8 |
| `TestMySqlDebeziumAvroPayload` | 13/13 |
| `TestBufferedRecordMerger` | 18/18 |
Compiles clean with checkstyle enabled.
### On the second suggestion (`#equals` on payloads)
I don't think that one can work here. The merger's check is `updatedRecord
== newerAvroRecord` on the **Avro records**, not on the payloads - and after an
in-place backfill those two are literally the same object, so any `equals`,
reference or deep, returns true and the shortcut still fires. Detecting the
mutation by comparison would mean snapshotting every incoming record before
calling the payload, which costs strictly more than making the payload
immutable.
### Why the Flink test was failing, for the record
Removing the shortcut also dropped the `HoodieOperation` that `newer`
carried, because the rebuild passes `updatedValue.isEmpty()` as `isDelete` and
that is always `false` inside `if (updatedValue.isPresent())`. In the COW
upsert leg of `TestFlinkWriteClientFunctional`, `id2` is a
`HoodieOperation.DELETE` record whose delete-ness lives only in the operation
field (no `_hoodie_is_deleted`, non-empty payload), so
`combineAndGetUpdateValue` returned the same reference, the shortcut used to
preserve the marker, and without it the row got written instead of deleted -
`numWrites` 2 -> 3. The payload-side fix avoids this entirely, since the
shortcut stays in place for every record that isn't being backfilled.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
