linliu-code commented on code in PR #19908:
URL: https://github.com/apache/hudi/pull/19908#discussion_r4007918544
##########
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:
Good catch, and yes it is intended. Added to Impact: a non-delete row with a
null value in the ordering column now fails the write with
`IllegalArgumentException: Ordering field '<field>' has null value`, where such
rows previously passed through carrying the default.
Worth noting it is not a new rule so much as an alignment:
`getOrderingValue` already threw that for `requiresOrderingValue` writes, and
the `shouldCombine` path already reached it. This change makes the prepped and
non-prepped paths agree. Deletes are exempt, since a delete may carry only its
key.
##########
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:
Done — it reads `shouldComputeOrderingValue` now.
##########
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:
Description updated. The code and the body now agree: deletes take the
ordering value, and a delete left on the default would be treated as commit
time ordered by `shouldKeepNewerRecord`, which is precisely why excluding them
would let a stale delete remove a newer record. Both the Change Logs and Impact
sections say so.
##########
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:
Aligned — both the inline comment and the test javadoc now reference
`BufferedRecordMergerFactory#shouldKeepNewerRecord`, so the
default-means-commit-time rule has one landing place.
--
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]
