hudi-agent commented on code in PR #19934:
URL: https://github.com/apache/hudi/pull/19934#discussion_r4000399387
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/hudi/command/payload/ExpressionPayload.scala:
##########
@@ -163,6 +163,37 @@ class ExpressionPayload(@transient record: GenericRecord,
}
}
+ /**
+ * Weighs a matched delete against the record already in storage.
+ *
+ * A matched delete carries the source row's ordering value, so on a table
that orders by event
+ * time it has to lose to a stored record with a higher ordering value,
exactly as the update
+ * branch does through [[doRecordMerge]]. Returning the stored record leaves
it untouched.
+ *
+ * The comparison needs the stored record, which only
`combineAndGetUpdateValue` supplies. The
+ * `getInsertValue` entry point passes none and the delete is written into a
delete block with
+ * its ordering value, so the reader weighs it there instead.
+ *
+ * No incoming record is handed to `needUpdatingPersistedRecord`: it would
look the ordering
+ * fields up by name on the joined record, whose fields are all renamed by
[[mergeSchema]], so
+ * every lookup misses. Passing none makes it fall back to this payload's
own ordering value,
+ * which was read from the source row when the record was created.
+ */
+ private def processDelete(targetRecord: Option[IndexedRecord],
+ properties: Properties): HOption[IndexedRecord] = {
+ val originalPayload = properties.getProperty(PAYLOAD_ORIGINAL_AVRO_PAYLOAD)
+ // Commit time ordering resolves to OverwriteWithLatestAvroPayload, where
the delete always
+ // wins. A custom payload keeps its existing semantics rather than having
one guessed for it.
+ // TODO(HUDI-8915): EventTimeAvroPayload also infers EVENT_TIME_ORDERING
and is not covered.
+ if (targetRecord.isEmpty ||
!classOf[DefaultHoodieRecordPayload].getName.equals(originalPayload)) {
+ HOption.empty()
+ } else if (needUpdatingPersistedRecord(targetRecord.get, HOption.empty(),
properties)) {
+ HOption.empty()
+ } else {
+ HOption.of(targetRecord.get)
Review Comment:
🤖 When the delete loses, this returns the stored record, which the merge
handles treat as an update and rewrite with the new `_hoodie_commit_time` (vs.
`SENTINEL`, which copies the old row untouched and is what
`DefaultHoodieRecordPayload` does for the same case). I see the update branch
already behaves this way so this is consistent, but is the commit-time bump for
a no-op delete intended? It would surface the row in incremental queries even
though nothing changed.
<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]
