linliu-code commented on code in PR #19908:
URL: https://github.com/apache/hudi/pull/19908#discussion_r4010332392
##########
hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/hudi/TestHoodieCreateRecordUtils.scala:
##########
@@ -259,6 +275,159 @@ class TestHoodieCreateRecordUtils {
.load(TestHoodieCreateRecordUtils.tempDir +
"/test_null_precombine_commit_time")
assertTrue(result.count() > 0, "Data should have been written successfully
with null precombine using COMMIT_TIME_ORDERING")
}
+ @Test
+ def testOrderingValueIsSetWhenCombineBeforeUpsertIsOff(): Unit = {
+ val orderingValue =
buildRecordOrderingValue(RecordMergeMode.EVENT_TIME_ORDERING, isDelete = false)
+ assertInstanceOf(classOf[java.lang.Long], orderingValue,
+ "the record must carry the ordering field's value, not the payload's
Integer default")
+ assertEquals(TS, orderingValue)
+ }
+
+ /** Commit time ordered tables must keep serving the default, whatever the
ordering fields say. */
+ @Test
+ def testCommitTimeOrderingKeepsTheDefaultOrderingValue(): Unit = {
+ assertEquals(OrderingValues.getDefault(),
+ buildRecordOrderingValue(RecordMergeMode.COMMIT_TIME_ORDERING, isDelete
= false))
+ }
+
+ /**
+ * Deletes carry the ordering field's value as well. A delete left on the
default is treated as
+ * commit time ordered by
BufferedRecordMergerFactory#deltaMergeDeleteRecord, which would let a
+ * stale delete remove a record with a higher ordering value.
+ */
+ @Test
+ def testDeleteAlsoGetsTheOrderingValue(): Unit = {
+ val orderingValue =
buildRecordOrderingValue(RecordMergeMode.EVENT_TIME_ORDERING, isDelete = true)
+ assertInstanceOf(classOf[java.lang.Long], orderingValue,
+ "the delete must carry the ordering field's value, not the payload's
Integer default")
+ assertEquals(TS, orderingValue)
+ }
+
+ /** Commit time ordered tables keep serving the default for deletes too. */
+ @Test
+ def testCommitTimeOrderingDeleteKeepsTheDefaultOrderingValue(): Unit = {
+ assertEquals(OrderingValues.getDefault(),
+ buildRecordOrderingValue(RecordMergeMode.COMMIT_TIME_ORDERING, isDelete
= true))
+ }
+
+ /**
+ * A delete row may carry only its key, with the ordering field left null.
Such a row must fall
+ * back to the default ordering value rather than failing the write. Every
other record
+ * representation already tolerates this:
HoodieSparkRecord#doGetOrderingValue,
+ * HoodieFlinkRecord#doGetOrderingValue and
HoodieAvroIndexedRecord#doGetOrderingValue.
+ */
+ @Test
+ def testDeleteWithNullOrderingFieldKeepsTheDefault(): Unit = {
+ assertEquals(OrderingValues.getDefault(),
+ buildRecordOrderingValue(RecordMergeMode.EVENT_TIME_ORDERING, isDelete =
true, ts = null))
+ }
+
+ /**
+ * End to end on the path this change widens. With de-duplication off, a
delete whose ordering
+ * value is older than the stored record must lose. While deletes were left
on the default the
+ * delete was treated as commit time ordered and removed the row regardless
of its ordering value.
+ */
+ @Test
+ def testStaleDeleteLosesWhenCombineBeforeUpsertIsOff(): Unit = {
+ val spark = TestHoodieCreateRecordUtils.spark
+ val basePath = TestHoodieCreateRecordUtils.tempDir +
"/stale_delete_no_combine"
+ val opts = Map(
+ "hoodie.insert.shuffle.parallelism" -> "1",
+ "hoodie.upsert.shuffle.parallelism" -> "1",
+ DataSourceWriteOptions.TABLE_TYPE.key ->
DataSourceWriteOptions.COW_TABLE_TYPE_OPT_VAL,
+ DataSourceWriteOptions.RECORDKEY_FIELD.key -> "uuid",
+ DataSourceWriteOptions.PARTITIONPATH_FIELD.key -> "partition",
+ HoodieTableConfig.ORDERING_FIELDS.key -> "ts",
+ HoodieTableConfig.RECORD_MERGE_MODE.key ->
RecordMergeMode.EVENT_TIME_ORDERING.name,
+ HoodieTableConfig.PAYLOAD_CLASS_NAME.key ->
classOf[DefaultHoodieRecordPayload].getName,
+ HoodieWriteConfig.TBL_NAME.key -> "test_stale_delete_no_combine",
+ // The trigger: no de-duplication of the incoming batch.
+ HoodieWriteConfig.COMBINE_BEFORE_UPSERT.key -> "false",
+ // Route the record through HoodieAvroRecord, whose payload defaults the
ordering value when
+ // none is set, rather than HoodieAvroIndexedRecord which derives it
lazily from the row.
+ HoodieWriteConfig.RECORD_MERGE_IMPL_CLASSES.key ->
classOf[HoodieAvroRecordMerger].getName,
+ HoodieWriteConfig.MERGE_HANDLE_CLASS_NAME.key ->
classOf[HoodieWriteMergeHandle[_, _, _, _]].getName)
+
+ def write(row: Row, mode: SaveMode): Unit =
+ spark.createDataFrame(spark.sparkContext.parallelize(Seq(row)),
ORDERING_TEST_SCHEMA)
+ .write.format("hudi").options(opts).mode(mode).save(basePath)
+
+ write(Row("id1", TS, "par1", false), SaveMode.Overwrite)
+ // A delete one tick older than the stored record.
+ write(Row("id1", TS - 1, "par1", true), SaveMode.Append)
+
+ assertEquals(1L, spark.read.format("hudi").load(basePath).where("uuid =
'id1'").count(),
+ "a delete older than the stored record must not remove it")
+ }
+
+ private def buildRecordOrderingValue(mergeMode: RecordMergeMode,
+ isDelete: Boolean,
+ ts: java.lang.Long = TS): Comparable[_]
= {
+ val spark = TestHoodieCreateRecordUtils.spark
+ val basePath =
+ TestHoodieCreateRecordUtils.tempDir +
s"/ordering_value_${mergeMode.name}_delete_${isDelete}_ts_$ts"
+
+ val parameters = Map(
+ KeyGeneratorOptions.RECORDKEY_FIELD_NAME.key() -> "uuid",
+ KeyGeneratorOptions.PARTITIONPATH_FIELD_NAME.key() -> "partition",
+ DataSourceWriteOptions.RECORDKEY_FIELD.key() -> "uuid",
+ DataSourceWriteOptions.PARTITIONPATH_FIELD.key() -> "partition",
+ // The trigger: no de-duplication of the incoming batch.
+ HoodieWriteConfig.COMBINE_BEFORE_UPSERT.key() -> "false",
+ DataSourceWriteOptions.INSERT_DROP_DUPS.key() -> "false"
+ )
+
+ val metaClient = HoodieTableMetaClient.newTableBuilder()
+ .setTableType(HoodieTableType.COPY_ON_WRITE)
+ .setTableName(s"test_ordering_value_${mergeMode.name}")
+ .setRecordKeyFields("uuid")
+ .setPartitionFields("partition")
+ .setOrderingFields("ts")
+ .setRecordMergeMode(mergeMode)
+
.initTable(HadoopFSUtils.getStorageConfWithCopy(spark.sparkContext.hadoopConfiguration),
basePath)
+
+ val hoodieSchema =
HoodieSchemaConversionUtils.convertStructTypeToHoodieSchema(
+ ORDERING_TEST_SCHEMA, "record", "org.apache.hudi.test")
+
+ val writeConfig = HoodieWriteConfig.newBuilder()
+ .withPath(basePath)
+ .withSchema(hoodieSchema.toString)
+ // The key generator is built from the write config's props, not from
`parameters`.
+ .withProps(writeProps(mergeMode))
+ .build()
+
+ val df = spark.createDataFrame(
+ spark.sparkContext.parallelize(Seq(Row("id1", ts, "par1", isDelete))),
ORDERING_TEST_SCHEMA)
+
+ val records = HoodieCreateRecordUtils.createHoodieRecordRdd(
+ HoodieCreateRecordUtils.createHoodieRecordRddArgs(
+ df, writeConfig, parameters, "record", "org.apache.hudi.test",
+ hoodieSchema, hoodieSchema, WriteOperationType.UPSERT,
"20260910000000",
+ preppedSparkSqlWrites = false, preppedSparkSqlMergeInto = false,
Review Comment:
I tried this and could not make it work, so I would rather report the
negative result than add a test that proves nothing.
**The Spark SQL `UPDATE` test does not discriminate.** I wrote one and ran
the whole class with the fix reverted to `val shouldComputeOrderingValue =
shouldCombine`:
```
$ mvn -pl hudi-spark-datasource/hudi-spark test -DskipUTs=true
-Dtest=TestHoodieCreateRecordUtils
[ERROR] Tests run: 12, Failures: 3, Errors: 0, Skipped: 0 -- in
org.apache.hudi.TestHoodieCreateRecordUtils
[ERROR] testStaleDeleteLosesWhenCombineBeforeUpsertIsOff
[ERROR] testDeleteAlsoGetsTheOrderingValue
[ERROR] testOrderingValueIsSetWhenCombineBeforeUpsertIsOff
```
`testSparkSqlUpdateOnEventTimeOrderedTable` ran in that same JVM and
**passed**. The three sibling failures are the control that the run was live
and the fix really was out. The test forced the merge handle three ways at once
— a session `set`, the same key in `tblproperties`, and
`hoodie.write.record.merge.custom.implementation.classes` — and still could not
reproduce the failure.
**The reason is reachability, and I had it too narrow in my own head.** The
record only serves the payload's `Integer` default when it takes the
`HoodieAvroRecord` branch:
```java
// HoodieRecordUtils.java:169
if (!requiresPayload && isPayloadClassDeprecated(payloadClass)) {
record = new HoodieAvroIndexedRecord(...); // derives the ordering value
from the record
} else {
record = new HoodieAvroRecord<>(...); // serves it from the payload
}
```
so it needs `requiresPayload || !isPayloadClassDeprecated(payloadClass)`.
Two separate routes:
* `requiresPayload` is `isChangingRecords(operation) &&
!config.isFileGroupReaderBasedMergeHandle`
(`HoodieCreateRecordUtils.scala:138`), and `hoodie.write.merge.handle.class`
**defaults** to `FileGroupReaderBasedMergeHandle`
(`HoodieWriteConfig.java:975`). So this route needs a non-default merge handle.
That is where we hit it, and it is why I could not stand it up from SQL on
master.
* `isPayloadClassDeprecated` is membership in a fixed set of nine built-in
payloads (`HoodieRecordUtils.java:71-80`). **Any custom payload class falls
outside it**, on stock merge-handle configuration. `BaseAvroPayload` keeps an
`orderingVal` field but does not override `getOrderingValue()` — among the
hudi-common model payloads only `OverwriteWithLatestAvroPayload` does — so such
a payload returns the interface default `0` and lands in the same comparison.
I have verified the second route by reading the branch condition and the
missing override, not by a running reproduction, so I am flagging it as the
weaker of the two claims.
So I have not added the functional test, and I have rewritten the Impact
section to state both routes rather than imply a table on stock defaults
reaches the exception. The discriminating tests in this PR are the unit tests,
which drive `HoodieCreateRecordUtils` directly; each fails when the one line it
guards is reverted, as above.
Worth separating, because only one half is configuration-dependent: the
**loud `ClassCastException`** needs one of the two routes above, while the
**gate itself** — `shouldCombine`, a de-duplication flag, deciding whether a
record gets an ordering value at all — is wrong regardless of either.
**On MERGE INTO: no, not through that flag.**
`SPARK_SQL_MERGE_INTO_PREPPED_KEY` is set to `isPrimaryKeylessTable`
(`MergeIntoHoodieTableCommand.scala:900` on spark3, `:899` on spark4), and
`preppedSparkSqlMergeInto` never drives `shouldCombine` to false on its own:
for an upsert the else-chain at `HoodieCreateRecordUtils.scala:96-108` falls
through to `hoodie.combine.before.upsert`, which defaults to true. MERGE INTO
also never sets `SPARK_SQL_WRITES_PREPPED_KEY` — that appears in exactly four
places, the `UpdateHoodieTableCommand` and `DeleteHoodieTableCommand` for
spark3 and spark4. So MERGE INTO reaches the widened branch only the ordinary
way, with `combine.before.upsert=false`.
--
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]
