hudi-agent commented on code in PR #19900:
URL: https://github.com/apache/hudi/pull/19900#discussion_r3991469855
##########
hudi-common/src/main/java/org/apache/hudi/common/model/HoodieAvroRecord.java:
##########
@@ -104,7 +106,58 @@ public T getData() {
@Override
public Comparable<?> doGetOrderingValue(HoodieSchema recordSchema,
Properties props, String[] orderingFields) {
- return this.getData().getOrderingValue();
+ Comparable<?> payloadOrderingValue = getData().getOrderingValue();
+ if (!canReadOrderingFieldsFromRecord(payloadOrderingValue, recordSchema,
props, orderingFields)) {
+ return payloadOrderingValue;
+ }
+ // The payload carries no ordering value: payloads constructed from the
record alone
+ // (HoodieRecordUtils#loadPayload(String, GenericRecord)) default to
OrderingValues#getDefault,
+ // an Integer. Read the ordering fields off the record instead, the way
+ // HoodieAvroIndexedRecord#doGetOrderingValue does, so that both Avro
record representations
+ // yield an ordering value of the type the ordering field declares. This
is best effort: any
+ // record we cannot read leaves the payload's value in place, which is
what callers saw before.
+ try {
+ Option<IndexedRecord> avroData =
getData().getIndexedRecord(recordSchema.toAvroSchema(), props);
+ if (!avroData.isPresent()) {
+ return payloadOrderingValue;
+ }
+ boolean consistentLogicalTimestampEnabled =
Boolean.parseBoolean(props.getProperty(
+
KeyGeneratorOptions.KEYGENERATOR_CONSISTENT_LOGICAL_TIMESTAMP_ENABLED.key(),
+
KeyGeneratorOptions.KEYGENERATOR_CONSISTENT_LOGICAL_TIMESTAMP_ENABLED.defaultValue()));
+ Comparable<?> recordOrderingValue = OrderingValues.create(
+ orderingFields,
+ field -> (Comparable<?>) HoodieAvroUtils.getNestedFieldVal(
+ (GenericRecord) avroData.get(), field, true,
consistentLogicalTimestampEnabled));
+ // A nullable ordering field holding null reads back as null, which no
caller expects here.
+ return recordOrderingValue == null ? payloadOrderingValue :
recordOrderingValue;
+ } catch (Exception e) {
+ // Reading the record is an optimization over the payload's default,
never a new failure
+ // mode: a schema the payload cannot decode, or props an exotic payload
requires and does
+ // not get here, must not turn an ordering value lookup into a write
failure.
+ return payloadOrderingValue;
+ }
+ }
+
+ /**
+ * Whether the ordering value is worth reading off the record rather than
taking the payload's.
+ * Only when the payload has none, the table declares ordering fields, and
the record is not a
+ * delete: BufferedRecord#isCommitTimeOrderingDelete treats a delete
carrying the default
Review Comment:
🤖 Keeping deletes at the default is a deliberate choice here, but it leaves
the two representations disagreeing for deletes: a `_hoodie_is_deleted=true`
record with `ts=100` is an event-time delete (loses to an existing `ts=200`) as
`HoodieAvroIndexedRecord`, yet a commit-time delete (always wins via
`isCommitTimeOrderingDelete`) as `HoodieAvroRecord`. Is that the intended
long-term contract, or should this be tracked as a follow-up? @nsivabalan would
be good to get your read on which delete semantics is the desired one for
payload-based records.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-common/src/main/java/org/apache/hudi/common/model/HoodieAvroRecord.java:
##########
@@ -104,7 +106,58 @@ public T getData() {
@Override
public Comparable<?> doGetOrderingValue(HoodieSchema recordSchema,
Properties props, String[] orderingFields) {
- return this.getData().getOrderingValue();
+ Comparable<?> payloadOrderingValue = getData().getOrderingValue();
+ if (!canReadOrderingFieldsFromRecord(payloadOrderingValue, recordSchema,
props, orderingFields)) {
+ return payloadOrderingValue;
+ }
+ // The payload carries no ordering value: payloads constructed from the
record alone
+ // (HoodieRecordUtils#loadPayload(String, GenericRecord)) default to
OrderingValues#getDefault,
+ // an Integer. Read the ordering fields off the record instead, the way
+ // HoodieAvroIndexedRecord#doGetOrderingValue does, so that both Avro
record representations
+ // yield an ordering value of the type the ordering field declares. This
is best effort: any
+ // record we cannot read leaves the payload's value in place, which is
what callers saw before.
+ try {
+ Option<IndexedRecord> avroData =
getData().getIndexedRecord(recordSchema.toAvroSchema(), props);
+ if (!avroData.isPresent()) {
+ return payloadOrderingValue;
+ }
+ boolean consistentLogicalTimestampEnabled =
Boolean.parseBoolean(props.getProperty(
+
KeyGeneratorOptions.KEYGENERATOR_CONSISTENT_LOGICAL_TIMESTAMP_ENABLED.key(),
+
KeyGeneratorOptions.KEYGENERATOR_CONSISTENT_LOGICAL_TIMESTAMP_ENABLED.defaultValue()));
+ Comparable<?> recordOrderingValue = OrderingValues.create(
+ orderingFields,
+ field -> (Comparable<?>) HoodieAvroUtils.getNestedFieldVal(
+ (GenericRecord) avroData.get(), field, true,
consistentLogicalTimestampEnabled));
+ // A nullable ordering field holding null reads back as null, which no
caller expects here.
+ return recordOrderingValue == null ? payloadOrderingValue :
recordOrderingValue;
+ } catch (Exception e) {
Review Comment:
🤖 Catching `Exception` here silently hands back the `Integer` default, so if
the record read fails for only some records in a batch the merge ends up
comparing `Integer` vs `Long` again (or silently picks the newer record) far
from the root cause. Would narrowing this to `IOException`/`HoodieException`,
or at least logging at debug/warn, be reasonable so a real decode problem is
still visible?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-common/src/main/java/org/apache/hudi/common/model/HoodieAvroRecord.java:
##########
@@ -104,7 +106,58 @@ public T getData() {
@Override
public Comparable<?> doGetOrderingValue(HoodieSchema recordSchema,
Properties props, String[] orderingFields) {
- return this.getData().getOrderingValue();
+ Comparable<?> payloadOrderingValue = getData().getOrderingValue();
+ if (!canReadOrderingFieldsFromRecord(payloadOrderingValue, recordSchema,
props, orderingFields)) {
+ return payloadOrderingValue;
+ }
+ // The payload carries no ordering value: payloads constructed from the
record alone
+ // (HoodieRecordUtils#loadPayload(String, GenericRecord)) default to
OrderingValues#getDefault,
+ // an Integer. Read the ordering fields off the record instead, the way
+ // HoodieAvroIndexedRecord#doGetOrderingValue does, so that both Avro
record representations
+ // yield an ordering value of the type the ordering field declares. This
is best effort: any
+ // record we cannot read leaves the payload's value in place, which is
what callers saw before.
+ try {
+ Option<IndexedRecord> avroData =
getData().getIndexedRecord(recordSchema.toAvroSchema(), props);
+ if (!avroData.isPresent()) {
+ return payloadOrderingValue;
+ }
+ boolean consistentLogicalTimestampEnabled =
Boolean.parseBoolean(props.getProperty(
+
KeyGeneratorOptions.KEYGENERATOR_CONSISTENT_LOGICAL_TIMESTAMP_ENABLED.key(),
+
KeyGeneratorOptions.KEYGENERATOR_CONSISTENT_LOGICAL_TIMESTAMP_ENABLED.defaultValue()));
+ Comparable<?> recordOrderingValue = OrderingValues.create(
+ orderingFields,
+ field -> (Comparable<?>) HoodieAvroUtils.getNestedFieldVal(
+ (GenericRecord) avroData.get(), field, true,
consistentLogicalTimestampEnabled));
+ // A nullable ordering field holding null reads back as null, which no
caller expects here.
+ return recordOrderingValue == null ? payloadOrderingValue :
recordOrderingValue;
+ } catch (Exception e) {
+ // Reading the record is an optimization over the payload's default,
never a new failure
+ // mode: a schema the payload cannot decode, or props an exotic payload
requires and does
+ // not get here, must not turn an ordering value lookup into a write
failure.
+ return payloadOrderingValue;
+ }
+ }
+
+ /**
+ * Whether the ordering value is worth reading off the record rather than
taking the payload's.
+ * Only when the payload has none, the table declares ordering fields, and
the record is not a
+ * delete: BufferedRecord#isCommitTimeOrderingDelete treats a delete
carrying the default
+ * ordering value as commit time ordered, so giving deletes a real value
here would silently
+ * change which delete wins.
+ */
+ private boolean canReadOrderingFieldsFromRecord(Comparable<?>
payloadOrderingValue,
+ HoodieSchema recordSchema, Properties props, String[] orderingFields) {
+ if (orderingFields == null || orderingFields.length == 0 || props == null)
{
+ return false;
+ }
+ if (!OrderingValues.isDefault(payloadOrderingValue)) {
+ return false;
+ }
+ if (Boolean.TRUE.equals(isDelete) ||
HoodieOperation.isDelete(getOperation())) {
+ return false;
+ }
+ return !(this.data instanceof BaseAvroPayload)
+ || !((BaseAvroPayload)
this.data).isDeleted(recordSchema.toAvroSchema(), props);
Review Comment:
🤖 This `isDeleted(...)` call runs outside the try/catch, and for
`DefaultHoodieRecordPayload` it decodes the record with `recordSchema` when the
schema has `_hoodie_is_deleted`. I reproduced it locally: the same mismatched
schema as `testUndecodableRecordSchemaKeepsPayloadDefault` plus a
`_hoodie_is_deleted` field throws `UnsupportedOperationException` out of
`getOrderingValue` (via `SerializableIndexedRecord.getData` -> `isDeleteRecord`
-> `isDeleted` -> line 160). Could the delete check move inside the try so the
"never a new failure mode" contract actually holds?
<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]
