[SUPPORT] Error executing Merge Or Read

[jai20242]

Hello.

I am developing a project using Hudi version 0.14.1 and I have developed my custom payload class. I have a problem using Merge On Read because the data is not consistent. If I use Copy On Write, it works well but with Merge On Read hudi calls the method getInsertValue before calling combineAndGetUpdateValue when the row exists. It is a problem for me because in the method getInsertValue I replace the values @#=BDP_N=#@ by null (because when the row does not exist, this value is null). Then, in the method combineAndGetUpdateValue I use the row before and I replace the columns modified. If the value of a column is null, I get the value before.

I am going to explain better with an example:

1) Insert a first row

spark.createDataFrame(Seq(("1",2L,"value1","value2","2",false))).toDF("key", "sort","field1","field2","partition","_hoodie_is_deleted").
write.format("hudi").
option(OPERATION_OPT_KEY, "upsert").
option(CDC_ENABLED.key(), "true").
option(TABLE_NAME, "pruebasc").
option("hoodie.datasource.write.payload.class","CustomOverwriteWithLatestAvroPayload").
option("hoodie.avro.schema.validate","false").
option("hoodie.datasource.write.recordkey.field","key").
option("hoodie.datasource.write.precombine.field","sort").
option("hoodie.datasource.write.new.columns.nullable", "true").
option("hoodie.datasource.write.reconcile.schema","true").
option("hoodie.metadata.enable","false").
option("hoodie.index.type","SIMPLE").
option("hoodie.datasource.write.table.type","MERGE_ON_READ").
option("hoodie.compact.inline","true").
mode(Overwrite).
save("/tmp/pruebasc")

I have the following row:
key -> 1
sort -> 2
field1 -> value1
field2 -> value2
partition -> 2
_hoodie_is_deleted -> false

2) I update the row. The custom payload replaces the values @#=BDP_N=#@ by null and the columns not modified keep the value before

spark.createDataFrame(Seq(("1",3L,"@#=BDP_N=#@","value3","2",false))).toDF("key", "sort","field1","field2","partition","_hoodie_is_deleted").withColumn("field1", col("field1").cast("String")).
write.format("hudi").
option(OPERATION_OPT_KEY, "upsert").
option(CDC_ENABLED.key(), "true").
option(TABLE_NAME, "pruebasc").
option("hoodie.datasource.write.payload.class","CustomOverwriteWithLatestAvroPayload").
option("hoodie.avro.schema.validate","false").
option("hoodie.datasource.write.recordkey.field","key").
option("hoodie.datasource.write.precombine.field","sort").
option("hoodie.datasource.write.new.columns.nullable", "true"). //Las columnas nuevas insertadas las pone a null
option("hoodie.datasource.write.reconcile.schema","true"). //coge el esquema de lo insertado y no del nuevo registro. Importante que todo sea nulo menos la PK
option("hoodie.metadata.enable","false").
option("hoodie.index.type","SIMPLE").
option("hoodie.datasource.write.table.type","MERGE_ON_READ").
option("hoodie.compact.inline","true").
mode(Append).
save("/tmp/pruebasc")

I have the following row:
key -> 1
sort -> 3
field1 -> value1
field2 -> value3
partition -> 2
_hoodie_is_deleted -> false

But the correct row must be:
key -> 1
sort -> 3
field1 -> null
field2 -> value3
partition -> 2
_hoodie_is_deleted -> false

I have checked my log and the problem is that using Merge On Read, hudi calls the method getInsertValue when the key exists before calling combineAndGetUpdateValue

Regards

[danny0405]

There is no guarantee that #getInsertValue should be invoked after #combineAndGetUpdateValue, actually we need the value returned by #getInsertValue to hand it over to #combineAndGetUpdateValue.

[jai20242]

And why does it only happen with Merge On Read? Also, I have tested the version 1.0.0-beta and it doesn't happen (it works well but we can't use a beta version in production and it works worse than 0 versions). With copy or write and with the version 1.0.0-beta with updates only call the method combineAndGetUpdateValue.
I don't understand why we need to call a function to insert data before updating because then both methods are not isolated.

[danny0405]

And why does it only happen with Merge On Read

My guess is we use the #getInsertValue to deserialize the payload from logs. Maybe I lost some reminisenses and if you already tests it locally you can trace the invocation chain.