hudi-agent commented on code in PR #19660:
URL: https://github.com/apache/hudi/pull/19660#discussion_r3806194442
##########
hudi-common/src/main/java/org/apache/hudi/common/model/debezium/AbstractDebeziumAvroPayload.java:
##########
@@ -72,15 +75,37 @@ public Option<IndexedRecord> getInsertValue(Schema schema)
throws IOException {
return insertValue.isPresent() ? handleDeleteOperation(insertValue.get())
: Option.empty();
}
+ @Override
+ public Option<IndexedRecord> getInsertValue(Schema schema, Properties
properties) throws IOException {
+ // Pin to the Debezium delete-op handling; DefaultHoodieRecordPayload's
properties-aware variant
+ // (event-time tracking, DELETE_KEY/DELETE_MARKER) must not replace it
+ return getInsertValue(schema);
+ }
+
@Override
public Option<IndexedRecord> combineAndGetUpdateValue(IndexedRecord
currentValue, Schema schema) throws IOException {
+ return combineAndGetUpdateValue(currentValue, schema, new Properties());
+ }
+
+ @Override
+ public Option<IndexedRecord> combineAndGetUpdateValue(IndexedRecord
currentValue, Schema schema, Properties properties) throws IOException {
// Step 1: If the time occurrence of the current record in storage is
higher than the time occurrence of the
// insert record (including a delete record), pick the current record.
Option<IndexedRecord> insertValue = getRecord(schema);
if (!insertValue.isPresent()) {
return Option.empty();
}
- if (shouldPickCurrentRecord(currentValue, insertValue.get(), schema)) {
+ String[] orderingFields = ConfigUtils.getOrderingFields(properties);
+ boolean pickCurrentRecord;
+ if (orderingFields == null || orderingFields.length != 1 ||
orderingFields[0].equals(getConnectorOrderingField())) {
+ // No ordering field configured, a composite ordering (not supported
yet), or the connector's own column:
+ // use the connector-specific comparison (MySQL's "file.pos" seq needs
segment-wise numeric compare;
+ // a plain Comparable is lexicographic)
+ pickCurrentRecord = shouldPickCurrentRecord(currentValue,
insertValue.get(), schema);
+ } else {
+ pickCurrentRecord = !needUpdatingPersistedRecord(currentValue,
insertValue, properties);
Review Comment:
🤖 On this configured-ordering path, `needUpdatingPersistedRecord` ->
`compareOrderingVal` does
`persistedOrderingVal.compareTo(incomingOrderingVal)`. If the incoming record's
configured ordering field is null (while the stored value is non-null), that's
`nonNull.compareTo(null)` -> NPE. The legacy `shouldPickCurrentRecord` path
guarded these bootstrap/null cases explicitly. Could a Debezium delete event
(op=d, where the ordering column may not be populated in the flattened
after-image) hit this? The added tests only exercise a null value on the stored
side, not the incoming side — might be worth a delete-op test through the else
branch. @nsivabalan
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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]
