wombatu-kun opened a new issue, #20367:
URL: https://github.com/apache/hudi/issues/20367
### Bug Description
**What happened:**
This affects the Spark record type
(`hoodie.write.record.merge.custom.implementation.classes=org.apache.hudi.DefaultSparkRecordMerger`)
on a MERGE_ON_READ table at table version 8 or 9 whose ordering field is
nested, for example `meta.ts`. Every delete is written into the delete block
with the default ordering value instead of its own.
`HoodieInlineLogAppendHandle.writeDelete` takes the value from
`HoodieSparkRecord.getOrderingValueAsJava`. That method checks the field with
the top-level-only `HoodieSchema.getField`, finds no `meta.ts` and falls back
to the default. Readers treat a delete with the default ordering value as a
commit-time delete. So under `EVENT_TIME_ORDERING`, a delete that is older than
the stored row still removes it.
`HoodieSparkRecord.doGetOrderingValue` resolves the same field by path, and
the Avro record type keeps the value.
**What you expected:**
The stale delete is ignored and the newer row stays. That is what happens
with the Avro record type or a top-level ordering field.
**Steps to reproduce:**
1. Create a MERGE_ON_READ table at table version 9 (the default of 1.2.x)
with the Spark record merger, a nested ordering field `meta.ts` and
`EVENT_TIME_ORDERING`.
2. Insert `id1` with `meta.ts = 100`.
3. Upsert `id1` with `meta.ts = 50` and `_hoodie_is_deleted = true`.
4. A snapshot read no longer returns `id1`; it should still return it with
`meta.ts = 100`.
The reproducing test in `TestMORDataSource` fails on master for the Spark
rows. The Avro row shows the expected result:
```scala
@ParameterizedTest
@CsvSource(value = Array("AVRO;bigint;8", "SPARK;bigint;9",
"SPARK;string;8", "SPARK;decimal(10,2);9"), delimiter = ';')
def testDeleteWithNestedOrderingField(recordType: HoodieRecordType,
orderingType: String, tableVersion: Int): Unit = {
val (writeOpts, readOpts) = getWriterReaderOpts(recordType)
val opts = writeOpts ++ Map(
TABLE_TYPE.key -> MOR_TABLE_TYPE_OPT_VAL,
RECORDKEY_FIELD.key -> "id",
PARTITIONPATH_FIELD.key -> "",
HoodieTableConfig.ORDERING_FIELDS.key -> "meta.ts",
HoodieWriteConfig.RECORD_MERGE_MODE.key ->
RecordMergeMode.EVENT_TIME_ORDERING.name(),
HoodieWriteConfig.WRITE_TABLE_VERSION.key -> tableVersion.toString,
HoodieTableConfig.VERSION.key -> tableVersion.toString,
HoodieCompactionConfig.INLINE_COMPACT.key -> "false")
def write(rows: Seq[(String, String, Boolean)], mode: SaveMode): Unit = {
val localSpark = spark
import localSpark.implicits._
rows.toDF("id", "ts", "_hoodie_is_deleted")
.select(col("id"),
struct(col("ts").cast(orderingType).as("ts")).as("meta"),
col("_hoodie_is_deleted"))
.coalesce(1).write.format("hudi").options(opts).mode(mode).save(basePath)
}
write(Seq(("id1", "100", false), ("id2", "100", false)),
SaveMode.Overwrite)
// The delete of id1 is older than the stored row and must not remove
it; the delete of id2 is newer.
write(Seq(("id1", "050", true), ("id2", "150", true)), SaveMode.Append)
val rows =
spark.read.format("hudi").options(readOpts).load(basePath).select("id",
"meta.ts").collect()
assertEquals(Seq("id1"), rows.map(_.getString(0)).toSeq)
assertEquals(100, BigDecimal(rows.head.get(1).toString).toInt)
}
```
### Environment
**Hudi version:** 1.2.0 and 1.2.1 (table version 9 by default), and master
with `hoodie.write.table.version` 8 or 9
**Query engine:** Spark, with the Spark record merger
**Relevant configs:** a nested ordering field, `EVENT_TIME_ORDERING`,
`hoodie.write.record.merge.custom.implementation.classes=org.apache.hudi.DefaultSparkRecordMerger`
### Logs and Stack Trace
The reproducing test on master:
```
[ERROR]
org.apache.hudi.functional.TestMORDataSource.testDeleteWithNestedOrderingField(HoodieRecordType,
String, int)[2] -- Time elapsed: 5.606 s <<< FAILURE!
org.opentest4j.AssertionFailedError: expected: <List(id1)> but was:
<WrappedArray()>
[ERROR]
org.apache.hudi.functional.TestMORDataSource.testDeleteWithNestedOrderingField(HoodieRecordType,
String, int)[3] -- Time elapsed: 4.644 s <<< FAILURE!
org.opentest4j.AssertionFailedError: expected: <List(id1)> but was:
<WrappedArray()>
[ERROR]
org.apache.hudi.functional.TestMORDataSource.testDeleteWithNestedOrderingField(HoodieRecordType,
String, int)[4] -- Time elapsed: 4.556 s <<< FAILURE!
org.opentest4j.AssertionFailedError: expected: <List(id1)> but was:
<WrappedArray()>
```
--
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]
