lanyuanxiaoyao/hudi
1
0
Fork 0
Code Activity

[HUDI-1720] Fix RealtimeCompactedRecordReader StackOverflowError (#2721)

Browse Source
This commit is contained in:
xiarixiaoyao
2021-04-13 18:23:26 +08:00
committed by GitHub
parent e16d31dce2
commit 65844a8d29
1 changed files with 24 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/realtime/RealtimeCompactedRecordReader.java
View File

@@ -77,6 +77,14 @@ class RealtimeCompactedRecordReader extends AbstractRealtimeRecordReader
.build();
}
private Option<GenericRecord> buildGenericRecordwithCustomPayload(HoodieRecord record) throws IOException {
if (usesCustomPayload) {
return record.getData().getInsertValue(getWriterSchema());
} else {
return record.getData().getInsertValue(getReaderSchema());
}
}
@Override
public boolean next(NullWritable aVoid, ArrayWritable arrayWritable) throws IOException {
// Call the underlying parquetReader.next - which may replace the passed in ArrayWritable
@@ -95,15 +103,24 @@ class RealtimeCompactedRecordReader extends AbstractRealtimeRecordReader
// TODO(NA): Invoke preCombine here by converting arrayWritable to Avro. This is required since the
// deltaRecord may not be a full record and needs values of columns from the parquet
Option<GenericRecord> rec;
if (usesCustomPayload) {
rec = deltaRecordMap.get(key).getData().getInsertValue(getWriterSchema());
} else {
rec = deltaRecordMap.get(key).getData().getInsertValue(getReaderSchema());
rec = buildGenericRecordwithCustomPayload(deltaRecordMap.get(key));
// If the record is not present, this is a delete record using an empty payload so skip this base record
// and move to the next record
while (!rec.isPresent()) {
// if current parquet reader has no record, return false
if (!this.parquetReader.next(aVoid, arrayWritable)) {
return false;
}
String tempKey = arrayWritable.get()[HoodieInputFormatUtils.HOODIE_RECORD_KEY_COL_POS].toString();
if (deltaRecordMap.containsKey(tempKey)) {
rec = buildGenericRecordwithCustomPayload(deltaRecordMap.get(tempKey));
} else {
// need to return true, since now log file does not contain tempKey, but parquet file contains tempKey
return true;
}
}
if (!rec.isPresent()) {
// If the record is not present, this is a delete record using an empty payload so skip this base record
// and move to the next record
return next(aVoid, arrayWritable);
return false;
}
GenericRecord recordToReturn = rec.get();
if (usesCustomPayload) {