wombatu-kun opened a new issue, #20365:
URL: https://github.com/apache/hudi/issues/20365
### Bug Description
**What happened:**
On a MERGE_ON_READ table at table version 8 or 9, Spark log compaction fails
when the file slice it compacts holds a delete block and the ordering field is
a string or a decimal. The error is `java.lang.UnsupportedOperationException:
Unsupported type of the value (class org.apache.hudi.Spark3HoodieUTF8String)`,
or `(class org.apache.spark.sql.types.Decimal)` for a decimal ordering field.
Log compaction reads the file slice with the file group reader. That reader
converts the ordering value of every delete-block entry to the Spark engine
type, so it can be compared with the data records
(`BufferedRecords.fromDeleteRecord` ->
`RecordContext.getOrderingValue(DeleteRecord)`): a string becomes a
`HoodieUTF8String` and a decimal becomes a Spark `Decimal`.
`FileGroupReaderBasedInlineLogAppendHandle` writes these deletes back as
`HoodieEmptyRecord`s that keep the engine value, but `HoodieDeleteBlock` can
only serialize Java types.
Log compaction of a data table always reads through the Spark reader
context, so the default Avro record merger is affected as well as the Spark
one. With `hoodie.log.compaction.inline=true`, the failure surfaces in the
write that triggers log compaction
(`BaseHoodieTableServiceClient.inlineLogCompact`).
**What you expected:**
Log compaction writes the deletes back with the Java ordering values they
had in the delete block. Later reads should still order the deletes correctly
against older and newer updates.
**Steps to reproduce:**
1. Create a MERGE_ON_READ table at table version 9 (the default of 1.2.x)
with a string ordering field `ts`, `EVENT_TIME_ORDERING` and
`hoodie.log.compaction.enable=true`.
2. Insert `id1` and `id2`.
3. Delete `id1` with `_hoodie_is_deleted=true`, which writes a delete block.
4. Update `id2`, which writes a data block in the same file slice.
5. Run log compaction, either with
`SparkRDDWriteClient.scheduleLogCompaction` and `logCompact` or through
`hoodie.log.compaction.inline=true` on a later write.
The reproducing test in `TestMORDataSource` fails on master for every row:
```scala
@ParameterizedTest
@CsvSource(value = Array("AVRO;string;8", "AVRO;decimal(10,2);9",
"SPARK;string;9", "SPARK;decimal(10,2);8"), delimiter = ';')
def testLogCompactionRewritesDeletesWithEngineOrdering(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 -> "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",
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))
.coalesce(1).write.format("hudi").options(opts).mode(mode).save(basePath)
}
write(Seq(("id1", "100", false), ("id2", "100", false)),
SaveMode.Overwrite)
// The delete lands in a delete block, the update in a data block of the
same file slice.
write(Seq(("id1", "110", true)), SaveMode.Append)
write(Seq(("id2", "120", false)), SaveMode.Append)
val client = DataSourceUtils.createHoodieClient(spark.sparkContext, "",
basePath, tableName, opts.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 compacted 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:** 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
**Relevant configs:** `hoodie.log.compaction.enable=true` or
`hoodie.log.compaction.inline=true`, a string or decimal ordering field, table
version 8 or 9
### Logs and Stack Trace
Log compaction in the reproducing test on master:
```
org.apache.hudi.exception.HoodieAppendException: Failed while appending
records to <table>/.<fileId>-0_<instant>.log.1_0-79-148
at
org.apache.hudi.io.HoodieInlineLogAppendHandle.appendDataAndDeleteBlocks(HoodieInlineLogAppendHandle.java:243)
at
org.apache.hudi.io.HoodieInlineLogAppendHandle.flushAppend(HoodieInlineLogAppendHandle.java:163)
at
org.apache.hudi.io.HoodieAppendHandle.doAppend(HoodieAppendHandle.java:161)
at
org.apache.hudi.io.FileGroupReaderBasedInlineLogAppendHandle.doAppend(FileGroupReaderBasedInlineLogAppendHandle.java:112)
at
org.apache.hudi.table.action.compact.HoodieCompactor.logCompact(HoodieCompactor.java:185)
...
Caused by: java.lang.UnsupportedOperationException: Unsupported type of the
value (class org.apache.hudi.Spark3HoodieUTF8String)
at
org.apache.hudi.common.avro.HoodieAvroWrapperUtils.wrapValueIntoAvro(HoodieAvroWrapperUtils.java:159)
at
org.apache.hudi.common.table.log.block.HoodieDeleteBlock.lambda$serializeV3$2(HoodieDeleteBlock.java:162)
...
at
org.apache.hudi.common.table.log.block.HoodieDeleteBlock.serializeV3(HoodieDeleteBlock.java:164)
at
org.apache.hudi.common.table.log.block.HoodieDeleteBlock.getContentBytes(HoodieDeleteBlock.java:109)
at
org.apache.hudi.common.table.log.HoodieLogFormatWriter.appendBlocks(HoodieLogFormatWriter.java:147)
at
org.apache.hudi.io.HoodieInlineLogAppendHandle.appendDataAndDeleteBlocks(HoodieInlineLogAppendHandle.java:230)
```
The decimal rows fail the same way with `Unsupported type of the value
(class org.apache.spark.sql.types.Decimal)`.
--
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]
