hudi-agent commented on code in PR #19908:
URL: https://github.com/apache/hudi/pull/19908#discussion_r3991125593
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieCreateRecordUtils.scala:
##########
@@ -155,7 +155,17 @@ object HoodieCreateRecordUtils {
} else {
avroRecWithoutMeta
}
- val hoodieRecord = if (shouldCombine && !orderingFields.isEmpty) {
+ // `shouldCombine` says whether the incoming batch needs
de-duplicating, which is not the
+ // same question as whether this record needs an ordering value.
Prepped Spark SQL writes
+ // emit one row per key and so set it false, which used to leave
the record with no
+ // ordering value on tables that order by event time.
+ // Deletes are excluded from the widened branch so their semantics
do not change: a
+ // delete carrying the default ordering value is treated as commit
time ordered by
+ // BufferedRecordMergerFactory#deltaMergeDeleteRecord, and giving
it a real value would
+ // make a delete lose to a stored record with a higher ordering
value. Deletes still get
Review Comment:
🤖 nit: since this is a boolean, `shouldComputeOrderingValue` (matching the
existing `shouldCombine` style) would read more clearly as a predicate than
`computeOrderingValue`.
<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:
##########
@@ -155,7 +155,17 @@ object HoodieCreateRecordUtils {
} else {
avroRecWithoutMeta
}
- val hoodieRecord = if (shouldCombine && !orderingFields.isEmpty) {
+ // `shouldCombine` says whether the incoming batch needs
de-duplicating, which is not the
+ // same question as whether this record needs an ordering value.
Prepped Spark SQL writes
+ // emit one row per key and so set it false, which used to leave
the record with no
+ // ordering value on tables that order by event time.
+ // Deletes are excluded from the widened branch so their semantics
do not change: a
+ // delete carrying the default ordering value is treated as commit
time ordered by
+ // BufferedRecordMergerFactory#deltaMergeDeleteRecord, and giving
it a real value would
+ // make a delete lose to a stored record with a higher ordering
value. Deletes still get
+ // an ordering value when `shouldCombine` is true, exactly as
before.
+ val computeOrderingValue = shouldCombine || (requiresOrderingValue
&& !isDelete)
Review Comment:
🤖 One side effect worth confirming is intended: `getOrderingValue` throws
`IllegalArgumentException` on a null ordering field when
`requiresOrderingValue` is true, and this branch is now reached for prepped
Spark SQL writes (UPDATE / MERGE INTO) and `combine.before.upsert=false`
upserts that previously passed such rows through with the default. So an
event-time-ordered table with nulls in the ordering column that used to accept
those writes will now fail them. That seems like the right call (and consistent
with `testNullPrecombineFieldThrowsClearError`), but could you call it out in
the PR description since it's a user-visible change beyond the
`ClassCastException` fix?
<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]
