rahil-c commented on PR #19749:
URL: https://github.com/apache/hudi/pull/19749#issuecomment-5433249413
The red job is `test-common-and-other-modules` (and the same one on Azure,
`UT FT common & other modules`). `test-utilities` was only cancelled as
collateral, and the `TestPreWriteValidatorUtils#testRunValidatorsInParallel`
blip is an unrelated timing flake that passed on retry.
The one real failure is:
```
[ERROR]
org.apache.hudi.client.TestFlinkWriteClientFunctional.testInsertAndUpsertWriteFilesAndCommitMetadata(HoodieTableType,
boolean)[1]
[ERROR] Run 1..4: ...:168->assertWriteStatuses:450 expected: <2> but was:
<3>
```
(both COW parameterizations, all 4 surefire retries)
### Why it fails
It isn't really Flink-specific - removing the shortcut also drops the
`HoodieOperation` that `newer` was carrying.
`HoodieWriteMergeHandle.write` builds the incoming record with
`BufferedRecords.fromHoodieRecord(..., deleteContext)`, which sets
`hoodieOperation = D` (or `-U`) for a delete. In the COW upsert leg of that
test, `id2` is exactly that: `HoodieOperation.DELETE`, ordering value `12` (so
`isCommitTimeOrderingDelete` is false), non-empty payload, `_hoodie_is_deleted`
unset. So the merger goes down the payload path, and `combineAndGetUpdateValue`
hands back the very same `GenericRecord` it was constructed with
(`BaseAvroPayload` keeps the incoming record in a field; `getRecord(schema)`
returns that reference when the schema matches). That makes `updatedRecord ==
newerAvroRecord` true, and the old shortcut returned `newer` with the delete
marker intact.
With the shortcut gone we fall into the rebuild, whose `isDelete` argument
is `updatedValue.isEmpty()` - which is always `false` inside `if
(updatedValue.isPresent())`. So the result comes back with `hoodieOperation ==
null`, `mergeResult.isDelete()` is false, `HoodieWriteMergeHandle.writeRecord`
takes the write branch instead of the delete branch, and `id2` is written as a
normal record: `numWrites` 2 -> 3 (`numDeletes` 1 -> 0 would have failed on the
next assertion too).
In other words, the shortcut was not purely an optimization - it was the
only thing propagating `newer`'s `HoodieOperation` through the merge.
### Suggested fix
Keep the identity branch, but rebuild the *data* from the merged Avro record
while preserving everything else `newer` carried:
```java
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 (`a0bed56`, JDK 11, `-Dspark3.5 -Dscala-2.12
-Dflink2.2`):
- without the fix: `TestFlinkWriteClientFunctional` -> `Tests run: 9,
Failures: 2` with the same `expected: <2> but was: <3>`
- with the fix: `Tests run: 9, Failures: 0, Errors: 0`
- `TestPostgresDebeziumToastV6ReadMerge` -> `Tests: succeeded 1, failed 0`
(the backfilled record is still rebuilt from `updatedRecord`, so the fix for
this PR is preserved)
Two side notes while you're in there:
1. `updatedValue.isEmpty()` in the remaining rebuild branch is dead-`false`;
a literal `false` would read more honestly.
2. This is the cost the PR description already flags - the identity case now
pays an Avro -> engine conversion per merged record on every CUSTOM-payload
table, not just Debezium ones. If that turns out to matter, the payload-side
alternative (#19280) avoids it, since only an actual TOAST backfill would
allocate.
--
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]
