in paimon-core/src/main/java/org/apache/paimon/schema/SchemaMergingUtils.java [84:210]
public static DataType merge(
DataType base0,
DataType update0,
AtomicInteger highestFieldId,
boolean allowExplicitCast) {
// Here we try t0 merge the base0 and update0 without regard to the nullability,
// and set the base0's nullability to the return's.
DataType base = base0.copy(true);
DataType update = update0.copy(true);
if (base.equals(update)) {
return base0;
} else if (base instanceof RowType && update instanceof RowType) {
List<DataField> baseFields = ((RowType) base).getFields();
List<DataField> updateFields = ((RowType) update).getFields();
Map<String, DataField> updateFieldMap =
updateFields.stream()
.collect(Collectors.toMap(DataField::name, Function.identity()));
List<DataField> updatedFields =
baseFields.stream()
.map(
baseField -> {
if (updateFieldMap.containsKey(baseField.name())) {
DataField updateField =
updateFieldMap.get(baseField.name());
DataType updatedDataType =
merge(
baseField.type(),
updateField.type(),
highestFieldId,
allowExplicitCast);
return new DataField(
baseField.id(),
baseField.name(),
updatedDataType,
baseField.description());
} else {
return baseField;
}
})
.collect(Collectors.toList());
Map<String, DataField> baseFieldMap =
baseFields.stream()
.collect(Collectors.toMap(DataField::name, Function.identity()));
List<DataField> newFields =
updateFields.stream()
.filter(field -> !baseFieldMap.containsKey(field.name()))
.map(field -> assignIdForNewField(field, highestFieldId))
.collect(Collectors.toList());
updatedFields.addAll(newFields);
return new RowType(base.isNullable(), updatedFields);
} else if (base instanceof MapType && update instanceof MapType) {
return new MapType(
base.isNullable(),
merge(
((MapType) base).getKeyType(),
((MapType) update).getKeyType(),
highestFieldId,
allowExplicitCast),
merge(
((MapType) base).getValueType(),
((MapType) update).getValueType(),
highestFieldId,
allowExplicitCast));
} else if (base instanceof ArrayType && update instanceof ArrayType) {
return new ArrayType(
base.isNullable(),
merge(
((ArrayType) base).getElementType(),
((ArrayType) update).getElementType(),
highestFieldId,
allowExplicitCast));
} else if (base instanceof MultisetType && update instanceof MultisetType) {
return new MultisetType(
base.isNullable(),
merge(
((MultisetType) base).getElementType(),
((MultisetType) update).getElementType(),
highestFieldId,
allowExplicitCast));
} else if (base instanceof DecimalType && update instanceof DecimalType) {
if (base.equals(update)) {
return base0;
} else {
throw new UnsupportedOperationException(
String.format(
"Failed to merge decimal types with different precision or scale: %s and %s",
base, update));
}
} else if (supportsDataTypesCast(base, update, allowExplicitCast)) {
if (DataTypes.getLength(base).isPresent() && DataTypes.getLength(update).isPresent()) {
// this will check and merge types which has a `length` attribute, like BinaryType,
// CharType, VarBinaryType, VarCharType.
if (allowExplicitCast
|| DataTypes.getLength(base).getAsInt()
<= DataTypes.getLength(update).getAsInt()) {
return update.copy(base0.isNullable());
} else {
throw new UnsupportedOperationException(
String.format(
"Failed to merge the target type that has a smaller length: %s and %s",
base, update));
}
} else if (DataTypes.getPrecision(base).isPresent()
&& DataTypes.getPrecision(update).isPresent()) {
// this will check and merge types which has a `precision` attribute, like
// LocalZonedTimestampType, TimeType, TimestampType.
if (allowExplicitCast
|| DataTypes.getPrecision(base).getAsInt()
<= DataTypes.getPrecision(update).getAsInt()) {
return update.copy(base0.isNullable());
} else {
throw new UnsupportedOperationException(
String.format(
"Failed to merge the target type that has a lower precision: %s and %s",
base, update));
}
} else {
return update.copy(base0.isNullable());
}
} else {
throw new UnsupportedOperationException(
String.format("Failed to merge data types %s and %s", base, update));
}
}