Different behavior between RealtimeCompactedRecordReader and HoodieMergeOnReadRDD

Hi, We were writing our internal payload class and found that Hudi's RealtimeCompactedRecordReader's behavior doesn't match HoodieMergeOnReadRDD.

 

To be specifically, when reading a delta record with merging log and base enabled, the expected behavior would be merge base + delta record. It's correctly handled by `HoodieMergeOnReadRDD`'s `mergeRowWithLog` method
{code:java}
private def mergeRowWithLog(curRow: InternalRow, curKey: String) = {
val historyAvroRecord = serializer.serialize(curRow).asInstanceOf[GenericRecord]
logRecords.get(curKey).getData.combineAndGetUpdateValue(historyAvroRecord, tableAvroSchema)
}
{code}
However the similar logic cannot be found in `RealtimeCompactedRecordReader's`, it just assumes the record in delta log is the latest.

 

cc [~garyli1019] since you wrote the `HoodieMergeOnReadRDD` code. It would be wonderful to merge base and delta record in the `RealtimeCompactedRecordReader`

 

Also cc [~wayblink].

## JIRA info

- Link: https://issues.apache.org/jira/browse/HUDI-1397
- Type: Bug

---

## Comments

16/Nov/20 16:45;garyli;[~advancedxy] thanks for reporting. We need a serializer here to convert arraywritable to avro and then merge them, just like the Spark implementation.

We already have avroToArrayWritable and we need to implement arrayWritableToAvro. 

I am a bit busy recently but I can take a look when I have some time. This is also a good task if you are interested in contributing.;;;

---

17/Nov/20 15:32;advancedxy;[~garyli] Thanks for your reply. [~wayblink] did a quick fix, the pr should be submitted in this week.;;;

---

18/Nov/20 02:24;garyli;Awesome!;;;

Compare RealtimeCompactedRecordReader with HoodieMergeOnReadRDD and its mergeRowWithLog method. Start with the existing avroToArrayWritable conversion and the missing arrayWritableToAvro path described in the discussion; done means base and delta records are merged consistently when reading compacted logs.

Written by the indexing model from the issue text.