wchevreuil commented on code in PR #169:
URL: https://github.com/apache/hbase-connectors/pull/169#discussion_r4231442590
##########
spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/SchemaConverters.scala:
##########
@@ -411,6 +414,95 @@ object SchemaConverters {
}
}
}
+
+ /**
+ * Constructs a converter function from Spark InternalRow types to Avro.
Used by the DS V2 write
+ * path where the writer receives InternalRow instead of Row.
+ */
+ def createConverterToAvroFromInternalRow(
+ dataType: DataType,
+ structName: String,
+ recordNamespace: String): (Any) => Any = {
+ dataType match {
+ case BinaryType =>
+ (item: Any) =>
+ item match {
+ case null => null
+ case bytes: Array[Byte] => ByteBuffer.wrap(bytes)
+ }
+ case ByteType | ShortType | IntegerType | LongType | FloatType |
DoubleType | BooleanType =>
+ identity
+ case StringType =>
+ (item: Any) => if (item == null) null else
item.asInstanceOf[UTF8String].toString
+ case _: DecimalType => (item: Any) => if (item == null) null else
item.toString
+ case TimestampType =>
+ (item: Any) => if (item == null) null else item.asInstanceOf[Long] /
1000
+ case ArrayType(elementType, _) =>
+ val elementConverter =
+ createConverterToAvroFromInternalRow(elementType, structName,
recordNamespace)
+ (item: Any) => {
+ if (item == null) {
+ null
+ } else {
+ val sourceArray = item.asInstanceOf[ArrayData]
+ val targetArray = new
util.ArrayList[Any](sourceArray.numElements())
+ var idx = 0
+ while (idx < sourceArray.numElements()) {
+ targetArray.add(elementConverter(sourceArray.get(idx,
elementType)))
+ idx += 1
+ }
+ targetArray
+ }
+ }
+ case MapType(StringType, valueType, _) =>
+ val valueConverter =
+ createConverterToAvroFromInternalRow(valueType, structName,
recordNamespace)
+ (item: Any) => {
+ if (item == null) {
+ null
+ } else {
+ val mapData = item.asInstanceOf[MapData]
+ val keys = mapData.keyArray()
+ val values = mapData.valueArray()
+ val javaMap = new HashMap[String, Any]()
+ var idx = 0
+ while (idx < mapData.numElements()) {
+ javaMap.put(
+ keys.getUTF8String(idx).toString,
+ valueConverter(values.get(idx, valueType)))
+ idx += 1
+ }
+ javaMap
+ }
+ }
+ case structType: StructType =>
+ val builder =
SchemaBuilder.record(structName).namespace(recordNamespace)
+ val schema: Schema =
+ SchemaConverters.convertStructToAvro(structType, builder,
recordNamespace)
+ val fieldConverters = structType.fields.map(field =>
+ createConverterToAvroFromInternalRow(field.dataType, field.name,
recordNamespace))
Review Comment:
Addressing on next commit. Replaced the overload with one that accepts the
actual Schema object. It resolves nullable unions and recurses with the real
field schemas for RECORD, ARRAY, and MAP types.
--
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]
