nsivabalan commented on code in PR #17473:
URL: https://github.com/apache/hudi/pull/17473#discussion_r2586126216
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/avro/SchemaConverters.scala:
##########
@@ -183,39 +190,40 @@ private[sql] object SchemaConverters {
}
avroType.addToSchema(SchemaBuilder.fixed(name).size(fixedSize))
- case BinaryType => builder.bytesType()
+ case BinaryType => Schema.create(Schema.Type.BYTES)
case ArrayType(et, containsNull) =>
- builder.array()
- .items(toAvroType(et, containsNull, recordName, nameSpace))
+ Schema.createArray(toAvroType(et, containsNull, recordName, nameSpace))
case MapType(StringType, vt, valueContainsNull) =>
- builder.map()
- .values(toAvroType(vt, valueContainsNull, recordName, nameSpace))
- case st: StructType =>
+ Schema.createMap(toAvroType(vt, valueContainsNull, recordName,
nameSpace))
+ case struct: StructType =>
val childNameSpace = if (nameSpace != "") s"$nameSpace.$recordName"
else recordName
- if (canBeUnion(st)) {
- val nonNullUnionFieldTypes = st.map(f => toAvroType(f.dataType,
nullable = false, f.name, childNameSpace))
+ if (canBeUnion(struct)) {
+ val nonNullUnionFieldTypes = struct.map(f => toAvroType(f.dataType,
nullable = false, f.name, childNameSpace))
val unionFieldTypes = if (nullable) {
nullSchema +: nonNullUnionFieldTypes
} else {
nonNullUnionFieldTypes
}
Schema.createUnion(unionFieldTypes:_*)
} else {
- val fieldsAssembler =
builder.record(recordName).namespace(nameSpace).fields()
- st.foreach { f =>
- val fieldAvroType =
- toAvroType(f.dataType, f.nullable, f.name, childNameSpace)
- fieldsAssembler.name(f.name).`type`(fieldAvroType).noDefault()
+ val fields = struct.map { field =>
+ val doc = field.getComment().orNull
+ val fieldAvroType = toAvroType(field.dataType, field.nullable,
field.name, childNameSpace)
+ if (AvroSchemaUtils.isNullable(fieldAvroType)) {
+ HoodieAvroUtils.createNewSchemaField(field.name, fieldAvroType,
doc, JsonProperties.NULL_VALUE)
+ } else {
+ HoodieAvroUtils.createNewSchemaField(field.name, fieldAvroType,
doc, null)
+ }
}
- fieldsAssembler.endRecord()
+ Schema.createRecord(recordName, null, nameSpace, false,
fields.asJava)
}
// This should never happen.
case other => throw new IncompatibleSchemaException(s"Unexpected type
$other.")
}
if (nullable && catalystType != NullType && schema.getType !=
Schema.Type.UNION) {
- Schema.createUnion(schema, nullSchema)
+ Schema.createUnion(nullSchema, schema)
Review Comment:
are we changing the order of null w/ the union w/ this change?
anything to consider on b/w compatibilty.
--
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]
