hudi-agent commented on code in PR #20049:
URL: https://github.com/apache/hudi/pull/20049#discussion_r4235923053
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieCreateRecordUtils.scala:
##########
@@ -290,27 +291,41 @@ object HoodieCreateRecordUtils {
* null values are allowed and a default ordering value is used.
* Otherwise, throws IllegalArgumentException if any ordering field has a
null value.
*/
- private def getOrderingValue(orderingFields: java.util.List[String],
- avroRec: GenericRecord,
- recordKey: String,
- consistentLogicalTimestampEnabled: Boolean,
- requiresOrderingValue: Boolean): Comparable[_]
= {
- OrderingValues.create(
- orderingFields,
- JFunction.toJavaFunction[String, Comparable[_]](field => {
- val fieldVal = HoodieAvroUtils.getNestedFieldVal(avroRec, field,
false, consistentLogicalTimestampEnabled)
- if (fieldVal == null) {
- if (requiresOrderingValue) {
- throw new IllegalArgumentException(
- s"Ordering field '$field' has null value for record key
'$recordKey'. " +
- s"Please ensure all records have non-null values for the
ordering field, " +
- s"or use a payload class that doesn't require ordering (e.g.,
OverwriteWithLatestAvroPayload).")
- }
- // Return default ordering value for payloads that don't require
ordering
- OrderingValues.getDefault.asInstanceOf[Comparable[_]]
- } else {
- fieldVal.asInstanceOf[Comparable[_]]
- }
- }))
+ /**
+ * Returns the function that reads the ordering value off a record, chosen
once for the write
+ * rather than per record: when an ordering value is required a missing one
is an error, and when
+ * it is not (COMMIT_TIME_ORDERING or OverwriteWithLatestAvroPayload) it is
replaced by the
+ * default ordering value.
+ */
+ private def orderingValueExtractor(orderingFields: java.util.List[String],
+ consistentLogicalTimestampEnabled:
Boolean,
+ requiresOrderingValue: Boolean):
(GenericRecord, String) => Comparable[_] = {
+ if (requiresOrderingValue) {
+ (avroRec, recordKey) =>
+ OrderingValues.create(
Review Comment:
🤖 nit: would it make sense to share the field-reading and
`OrderingValues.create` code between the two branches, so only the
missing-value handling differs?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieStreamerUtils.java:
##########
@@ -94,9 +96,20 @@ public static Option<JavaRDD<HoodieRecord>>
createHoodieRecords(HoodieStreamer.C
boolean useConsistentLogicalTimestamp = ConfigUtils.getBooleanWithAltKeys(
props,
KeyGeneratorOptions.KEYGENERATOR_CONSISTENT_LOGICAL_TIMESTAMP_ENABLED);
Set<String> partitionColumns = getPartitionColumns(props);
- String payloadClassName = StringUtils.isNullOrEmpty(cfg.payloadClassName)
- ? HoodieRecordPayload.getAvroPayloadForMergeMode(cfg.recordMergeMode,
cfg.payloadClassName)
- : cfg.payloadClassName;
+ // Prefer the table, like orderingFieldsStr above. cfg.recordMergeMode is
inferred from the
+ // streamer's own arguments and never reconciled with the table, so a run
that omits
+ // --source-ordering-fields on an event-time table reads
COMMIT_TIME_ORDERING here and leaves the
+ // ordering value unchecked. The payload has to follow the same mode, or a
stale
+ // COMMIT_TIME_ORDERING resolves it to OverwriteWithLatestAvroPayload and
skips the check anyway.
+ RecordMergeMode mergeMode = tableConfig.getRecordMergeMode() != null
+ ? tableConfig.getRecordMergeMode() : cfg.recordMergeMode;
+ String payloadClassName = tableConfig.getPayloadClassIfPresent()
Review Comment:
🤖 I think `payloadClassName` can now resolve to null. Take a v9 table with
`CUSTOM` merge mode backed by a custom merger, so no payload class is
persisted, and a run without `--payload-class`. `mergeMode` becomes CUSTOM,
`getAvroPayloadForMergeMode(CUSTOM, null)` returns null, and every record then
NPEs in `HoodieRecordUtils.loadPayload(null, ...)`. With an error table each
record is quarantined; without one the batch fails. Before this change, the
mode inferred from the CLI resolved to a built-in payload. Could you fall back
to a non-null payload here (e.g.
`HoodieTableConfig.getDefaultPayloadClassName()`), or use the table-preferred
payload only for the `requiresOrderingValue` gate?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieCreateRecordUtils.scala:
##########
@@ -290,27 +291,41 @@ object HoodieCreateRecordUtils {
* null values are allowed and a default ordering value is used.
* Otherwise, throws IllegalArgumentException if any ordering field has a
null value.
*/
- private def getOrderingValue(orderingFields: java.util.List[String],
- avroRec: GenericRecord,
- recordKey: String,
- consistentLogicalTimestampEnabled: Boolean,
- requiresOrderingValue: Boolean): Comparable[_]
= {
- OrderingValues.create(
- orderingFields,
- JFunction.toJavaFunction[String, Comparable[_]](field => {
- val fieldVal = HoodieAvroUtils.getNestedFieldVal(avroRec, field,
false, consistentLogicalTimestampEnabled)
- if (fieldVal == null) {
- if (requiresOrderingValue) {
- throw new IllegalArgumentException(
- s"Ordering field '$field' has null value for record key
'$recordKey'. " +
- s"Please ensure all records have non-null values for the
ordering field, " +
- s"or use a payload class that doesn't require ordering (e.g.,
OverwriteWithLatestAvroPayload).")
- }
- // Return default ordering value for payloads that don't require
ordering
- OrderingValues.getDefault.asInstanceOf[Comparable[_]]
- } else {
- fieldVal.asInstanceOf[Comparable[_]]
- }
- }))
+ /**
Review Comment:
🤖 nit: could you drop the old Javadoc above and keep only the new one? The
stale `getOrderingValue` comment is now stacked directly above it.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/streamer/HoodieStreamerUtils.java:
##########
@@ -130,6 +135,13 @@ public static Option<JavaRDD<HoodieRecord>>
createHoodieRecords(HoodieStreamer.C
? OrderingValues.create(orderingFieldsStr.split(","),
field -> (Comparable)
HoodieAvroUtils.getNestedFieldVal(gr, field, false,
useConsistentLogicalTimestamp))
: null;
+ if (requiresOrderingValue &&
OrderingValues.isMissing(orderingValue)) {
+ throw new IllegalArgumentException(
+ "Ordering fields '" + orderingFieldsStr + "' resolved
to a null value for record key '"
+ + hoodieKey.getRecordKey() + "'. Please ensure all
records carry non-null values for "
+ + "the ordering fields, or use a merge mode or
payload class that does not order "
+ + "(e.g., COMMIT_TIME_ORDERING or
OverwriteWithLatestAvroPayload).");
Review Comment:
🤖 Exempting deletes here would make the streamer differ from the Spark
datasource write path, which already throws for a `_hoodie_is_deleted` record
with a null ordering field when `requiresOrderingValue` holds
(`HoodieCreateRecordUtils` builds the ordering value regardless of `isDelete`),
so the quarantine looks consistent; the PR description's MOR note now calls
this out too.
--
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]
