hudi-agent commented on code in PR #19908:
URL: https://github.com/apache/hudi/pull/19908#discussion_r3998624175
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieCreateRecordUtils.scala:
##########
@@ -155,9 +155,20 @@ 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 take the ordering value too. A delete left on the
default is treated as commit
+ // time ordered by
BufferedRecordMergerFactory#shouldKeepNewerRecord, so excluding them
Review Comment:
🤖 nit: this comment points at
`BufferedRecordMergerFactory#shouldKeepNewerRecord` while the test javadoc
(line 295 of the test) points at `#deltaMergeDeleteRecord` for the same
behaviour — could you make both reference the same method so a reader chasing
the default-means-commit-time rule lands in one place?
<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,9 +155,20 @@ 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 take the ordering value too. A delete left on the
default is treated as commit
+ // time ordered by
BufferedRecordMergerFactory#shouldKeepNewerRecord, so excluding them
+ // here would let a stale delete remove a record with a higher
ordering value, while the
+ // same delete written through a path where `shouldCombine` is
true would correctly lose.
+ // A null ordering field stays tolerated for a delete, which may
carry only its key: the
+ // value falls back to the default rather than failing the write.
+ val computeOrderingValue = shouldCombine || requiresOrderingValue
+ val hoodieRecord = if (computeOrderingValue &&
!orderingFields.isEmpty) {
val orderingVal = getOrderingValue(orderingFields, avroRec,
hoodieKey.getRecordKey,
- consistentLogicalTimestampEnabled, requiresOrderingValue)
+ consistentLogicalTimestampEnabled, requiresOrderingValue &&
!isDelete)
Review Comment:
🤖 The PR description still says deletes are deliberately excluded from the
widened branch and stay commit-time ordered, but the code (and the comment
above) now gives deletes a real ordering value too. Could you update the
description so the semantics change for deletes on event-time tables is visible
to reviewers?
<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]
