voonhous commented on code in PR #17535:
URL: https://github.com/apache/hudi/pull/17535#discussion_r2615036323
##########
hudi-hadoop-mr/src/main/java/org/apache/hudi/hadoop/utils/HoodieArrayWritableAvroUtils.java:
##########
@@ -65,65 +65,66 @@
public class HoodieArrayWritableAvroUtils {
- public static ArrayWritable rewriteRecordWithNewSchema(ArrayWritable
writable, Schema oldSchema, Schema newSchema, Map<String, String> renameCols) {
+ public static ArrayWritable rewriteRecordWithNewSchema(ArrayWritable
writable, HoodieSchema oldSchema, HoodieSchema newSchema, Map<String, String>
renameCols) {
return (ArrayWritable) rewriteRecordWithNewSchema(writable, oldSchema,
newSchema, renameCols, new LinkedList<>());
}
- private static Writable rewriteRecordWithNewSchema(Writable writable, Schema
oldAvroSchema, Schema newAvroSchema, Map<String, String> renameCols,
Deque<String> fieldNames) {
+ private static Writable rewriteRecordWithNewSchema(Writable writable,
HoodieSchema oldAvroSchema, HoodieSchema newAvroSchema, Map<String, String>
renameCols, Deque<String> fieldNames) {
if (writable == null) {
return null;
}
- Schema oldSchema = AvroSchemaUtils.getNonNullTypeFromUnion(oldAvroSchema);
- Schema newSchema = AvroSchemaUtils.getNonNullTypeFromUnion(newAvroSchema);
- if (areSchemasProjectionEquivalent(oldSchema, newSchema)) {
+ HoodieSchema oldSchema =
HoodieSchemaUtils.getNonNullTypeFromUnion(oldAvroSchema);
+ HoodieSchema newSchema =
HoodieSchemaUtils.getNonNullTypeFromUnion(newAvroSchema);
+ if (HoodieSchemaCompatibility.areSchemasProjectionEquivalent(oldSchema,
newSchema)) {
return writable;
}
return rewriteRecordWithNewSchemaInternal(writable, oldSchema, newSchema,
renameCols, fieldNames);
}
- private static Writable rewriteRecordWithNewSchemaInternal(Writable
writable, Schema oldSchema, Schema newSchema, Map<String, String> renameCols,
Deque<String> fieldNames) {
+ private static Writable rewriteRecordWithNewSchemaInternal(Writable
writable, HoodieSchema oldSchema, HoodieSchema newSchema, Map<String, String>
renameCols, Deque<String> fieldNames) {
switch (newSchema.getType()) {
case RECORD:
if (!(writable instanceof ArrayWritable)) {
throw new SchemaCompatibilityException(String.format("Cannot rewrite
%s as a record", writable.getClass().getName()));
}
ArrayWritable arrayWritable = (ArrayWritable) writable;
- List<Schema.Field> fields = newSchema.getFields();
+ List<HoodieSchemaField> fields = newSchema.getFields();
// projection will keep the size from the "from" schema because it
gets recycled
// and if the size changes the reader will fail
boolean noFieldsRenaming = renameCols.isEmpty();
String namePrefix = createNamePrefix(noFieldsRenaming, fieldNames);
Writable[] values = new Writable[Math.max(fields.size(),
arrayWritable.get().length)];
for (int i = 0; i < fields.size(); i++) {
- Schema.Field newField = newSchema.getFields().get(i);
+ HoodieSchemaField newField = newSchema.getFields().get(i);
String newFieldName = newField.name();
fieldNames.push(newFieldName);
- Schema.Field oldField = noFieldsRenaming
+ Option<HoodieSchemaField> oldFieldOpt = noFieldsRenaming
? oldSchema.getField(newFieldName)
: oldSchema.getField(getOldFieldNameWithRenaming(namePrefix,
newFieldName, renameCols));
- if (oldField != null) {
+ if (oldFieldOpt.isPresent()) {
+ HoodieSchemaField oldField = oldFieldOpt.get();
values[i] =
rewriteRecordWithNewSchema(arrayWritable.get()[oldField.pos()],
oldField.schema(), newField.schema(), renameCols, fieldNames);
- } else if (newField.defaultVal() instanceof JsonProperties.Null) {
+ } else if (newField.defaultVal().isPresent() &&
newField.defaultVal().get() instanceof JsonProperties.Null) {
Review Comment:
Changing this to `HoodieSchema.NULL_VALUE` they are the same underlying
value.
--
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]
