xiarixiaoyao commented on code in PR #5376:
URL: https://github.com/apache/hudi/pull/5376#discussion_r855809877
##########
hudi-common/src/main/java/org/apache/hudi/common/table/log/AbstractHoodieLogRecordReader.java:
##########
@@ -379,7 +380,7 @@ private void processDataBlock(HoodieDataBlock dataBlock,
Option<KeySpec> keySpec
Option<Schema> schemaOption = getMergedSchema(dataBlock);
while (recordIterator.hasNext()) {
IndexedRecord currentRecord = recordIterator.next();
- IndexedRecord record = schemaOption.isPresent() ?
HoodieAvroUtils.rewriteRecordWithNewSchema(currentRecord, schemaOption.get()) :
currentRecord;
+ IndexedRecord record = schemaOption.isPresent() ?
HoodieAvroUtils.rewriteRecordWithNewSchema(currentRecord, schemaOption.get(),
new HashMap<>()) : currentRecord;
Review Comment:
yes, fixed
##########
hudi-common/src/main/java/org/apache/hudi/avro/HoodieAvroUtils.java:
##########
@@ -765,27 +802,41 @@ private static Object rewriteRecordWithNewSchema(Object
oldRecord, Schema oldSch
}
Collection array = (Collection)oldRecord;
List<Object> newArray = new ArrayList();
+ fieldNames.push("element");
for (Object element : array) {
- newArray.add(rewriteRecordWithNewSchema(element,
oldSchema.getElementType(), newSchema.getElementType()));
+ newArray.add(rewriteRecordWithNewSchema(element,
oldSchema.getElementType(), newSchema.getElementType(), renameCols,
fieldNames));
}
+ fieldNames.pop();
return newArray;
case MAP:
if (!(oldRecord instanceof Map)) {
throw new IllegalArgumentException("cannot rewrite record with
different type");
}
Map<Object, Object> map = (Map<Object, Object>) oldRecord;
Map<Object, Object> newMap = new HashMap<>();
+ fieldNames.push("value");
for (Map.Entry<Object, Object> entry : map.entrySet()) {
- newMap.put(entry.getKey(),
rewriteRecordWithNewSchema(entry.getValue(), oldSchema.getValueType(),
newSchema.getValueType()));
+ newMap.put(entry.getKey(),
rewriteRecordWithNewSchema(entry.getValue(), oldSchema.getValueType(),
newSchema.getValueType(), renameCols, fieldNames));
}
+ fieldNames.pop();
return newMap;
case UNION:
- return rewriteRecordWithNewSchema(oldRecord,
getActualSchemaFromUnion(oldSchema, oldRecord),
getActualSchemaFromUnion(newSchema, oldRecord));
+ return rewriteRecordWithNewSchema(oldRecord,
getActualSchemaFromUnion(oldSchema, oldRecord),
getActualSchemaFromUnion(newSchema, oldRecord), renameCols, fieldNames);
default:
return rewritePrimaryType(oldRecord, oldSchema, newSchema);
}
}
+ private static String createFullName(Deque<String> fieldNames) {
+ String result = "";
+ if (!fieldNames.isEmpty()) {
+ List<String> parentNames = new ArrayList<>();
+ fieldNames.descendingIterator().forEachRemaining(parentNames::add);
Review Comment:
fixed
--
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]
