linliu-code commented on code in PR #19908:
URL: https://github.com/apache/hudi/pull/19908#discussion_r3999876775
##########
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:
these prepped* arguments are always false. However, our initial finding of
the issue is from the spark SQL update command, which would cause these
preppedSparkSqlWrites as true.
Therefore, I think we should add functional using Spark SQL to see if the
issue exists before the fix and is gone after the fix.
preppedSparkSqlWrites vs preppedSparkSqlMergeInto, does it mean the issue
could also happen for MIT statements?
--
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]
