wombatu-kun opened a new issue, #20372:
URL: https://github.com/apache/hudi/issues/20372
### Bug Description
**What happened:**
On master, MERGE_ON_READ tables at table version 10, the default, write
deletes into native delete log files. With the Spark record type
(`DefaultSparkRecordMerger`) and a string ordering field, a write that sends a
delete to a log file fails with `HoodieException: Failed to write to Hudi`. The
executor logs `java.lang.ClassCastException: class
org.apache.hudi.Spark3HoodieUTF8String cannot be cast to class
org.apache.spark.unsafe.types.UTF8String`.
`HoodieNativeLogFormatWriter.appendDeleteRecord` builds the delete row from
`record.getOrderingValue(...)`, which is the comparable form of the value.
Spark wraps a string into `HoodieUTF8String` there, because `UTF8String` no
longer supports `compareTo` (SPARK-46832). The wrapper goes into the
`GenericInternalRow` unchanged, and the Spark Parquet writer fails when it
reads the column with `getUTF8String`. The Spark record context builds these
rows for the Spark record type since #19283
(`HoodieSparkTable.getRecordContextForWrite`).
Native log compaction with the Spark record type writes the deletes it reads
back through the same method, again with a `HoodieUTF8String` from the Spark
reader context, whoever wrote the deletes. There the exception is swallowed:
`HoodieAppendHandle` only marks the record as failed
(`hoodie.write.ignore.failed` defaults to true), the log compaction commits,
and the delete is gone. A table written with the default Avro record type
therefore loses its string-ordered deletes once a Spark-typed job log-compacts
it.
With several ordering fields, a null field fails the delete write too: the
writer stores the default ordering value, an `Integer` 0, in that field's
column instead of NULL, and a string column rejects it.
**What you expected:**
The delete row holds the values the engine stores, with NULL for a null
ordering field. The write and the log compaction succeed, and later reads order
the delete correctly against older and newer updates.
**Steps to reproduce:**
1. On master, create a MERGE_ON_READ table (table version 10 is the default)
with
`hoodie.write.record.merge.custom.implementation.classes=org.apache.hudi.DefaultSparkRecordMerger`,
a string ordering field `ts` and `EVENT_TIME_ORDERING`.
2. Insert `id1` and `id2`.
3. Upsert `id1` with `_hoodie_is_deleted=true`. The write fails.
For the log compaction case, write the same rows with the default Avro
record type, run log compaction with `DefaultSparkRecordMerger`, then upsert
`id1` with an ordering value older than the delete's: `id1` comes back.
A test in `TestMORDataSource` reproduces both. On master, rows
`SPARK;SPARK;string;ts` and `SPARK;SPARK;string;ts,seq` fail at the delete
write with `Failed to write to Hudi`, and row `AVRO;SPARK;string;ts` fails
after log compaction with `expected: <Map(id2 -> 120)> but was: <Map(id2 ->
120, id1 -> 105)>`:
```scala
@ParameterizedTest
@CsvSource(value = Array(
"AVRO;AVRO;string;ts", "SPARK;SPARK;string;ts",
"SPARK;SPARK;decimal(10,2);ts", "SPARK;SPARK;string;ts,seq",
"AVRO;SPARK;string;ts"), delimiter = ';')
def testNativeDeleteLogWithEngineOrdering(recordType: HoodieRecordType,
logCompactionRecordType: HoodieRecordType,
orderingType: String,
orderingFields: String): 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 -> orderingFields,
HoodieWriteConfig.RECORD_MERGE_MODE.key ->
RecordMergeMode.EVENT_TIME_ORDERING.name(),
// Table version 10 writes deletes to native delete log files.
HoodieWriteConfig.WRITE_TABLE_VERSION.key ->
HoodieTableVersion.TEN.versionCode().toString,
HoodieCompactionConfig.INLINE_COMPACT.key -> "false",
HoodieCompactionConfig.ENABLE_LOG_COMPACTION.key -> "true",
HoodieCompactionConfig.LOG_COMPACTION_BLOCKS_THRESHOLD.key -> "1")
def write(rows: Seq[(String, String, Boolean)], mode: SaveMode): Unit = {
val localSpark = spark
import localSpark.implicits._
rows.toDF("id", "ts", "_hoodie_is_deleted")
.withColumn("ts", col("ts").cast(orderingType))
.withColumn("seq", when(col("_hoodie_is_deleted"),
lit(null)).otherwise(lit("1")).cast(orderingType))
.coalesce(1).write.format("hudi").options(opts).mode(mode).save(basePath)
}
write(Seq(("id1", "100", false), ("id2", "100", false)),
SaveMode.Overwrite)
// The delete leaves seq null, so with two ordering fields its second
value is missing.
write(Seq(("id1", "110", true)), SaveMode.Append)
write(Seq(("id2", "120", false)), SaveMode.Append)
// Log compaction reads the delete back and writes it into a new native
delete log. Its record type can differ
// from the writers', e.g. when a Spark-typed job compacts deletes that
an Avro-typed one wrote.
val logCompactionOpts = if (logCompactionRecordType ==
HoodieRecordType.SPARK) opts ++ sparkOpts else opts
val client = DataSourceUtils.createHoodieClient(spark.sparkContext, "",
basePath, tableName, logCompactionOpts.asJava)
.asInstanceOf[SparkRDDWriteClient[HoodieRecordPayload[Nothing]]]
try {
val logCompactionInstant =
client.scheduleLogCompaction(Option.empty()).get()
client.logCompact(logCompactionInstant, true)
} finally {
client.close()
}
def readOrdering(): Map[String, Int] =
spark.read.format("hudi").options(readOpts).load(basePath)
.select("id", "ts").collect().map(row => row.getString(0) ->
BigDecimal(row.get(1).toString).toInt).toMap
// The delete keeps its ordering value: an older update leaves id1
deleted, a newer one brings it back.
write(Seq(("id1", "105", false)), SaveMode.Append)
assertEquals(Map("id2" -> 120), readOrdering())
write(Seq(("id1", "115", false)), SaveMode.Append)
assertEquals(Map("id1" -> 115, "id2" -> 120), readOrdering())
}
```
### Environment
**Hudi version:** master (table version 10 and native log files are not
released yet)
**Query engine:** Spark
**Relevant configs:**
`hoodie.write.record.merge.custom.implementation.classes=org.apache.hudi.DefaultSparkRecordMerger`,
a string ordering field, table version 10
### Logs and Stack Trace
The delete write in the reproducing test on master:
```
ERROR org.apache.hudi.io.HoodieAppendHandle [] - Error writing record
HoodieRecord{key=HoodieKey(recordKey=id1, partitionPath=), ...}
java.lang.ClassCastException: class org.apache.hudi.Spark3HoodieUTF8String
cannot be cast to class org.apache.spark.unsafe.types.UTF8String
at
org.apache.spark.sql.catalyst.expressions.GenericInternalRow.getUTF8String(rows.scala:165)
at
org.apache.hudi.io.storage.row.HoodieRowParquetWriteSupport.lambda$makeWriter$23(HoodieRowParquetWriteSupport.java:569)
...
at
org.apache.hudi.io.storage.HoodieSparkParquetWriter.writeRow(HoodieSparkParquetWriter.java:100)
at
org.apache.hudi.io.cdc.HoodieNativeLogFormatWriter.appendDeleteRecord(HoodieNativeLogFormatWriter.java:178)
at
org.apache.hudi.io.HoodieNativeLogAppendHandle.writeDelete(HoodieNativeLogAppendHandle.java:152)
at
org.apache.hudi.io.HoodieAppendHandle.writeRecord(HoodieAppendHandle.java:397)
org.apache.hudi.exception.HoodieException: Failed to write to Hudi
at org.apache.hudi.DefaultSource.createRelation(DefaultSource.scala:189)
```
Log compaction of the Avro-written table logs the same `ClassCastException`
from `HoodieNativeLogFormatWriter.appendDeleteRecord`, and then commits without
the delete.
--
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]
