taklwu commented on code in PR #169:
URL: https://github.com/apache/hbase-connectors/pull/169#discussion_r4333258283
##########
spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/SchemaConverters.scala:
##########
@@ -411,6 +414,115 @@ 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.
+ *
+ * This overload accepts the actual Avro Schema so that GenericRecord
instances are built against
+ * the original schema rather than one reconstructed from the Spark
StructType. The round-trip
+ * through convertStructToAvro loses nullable unions, FIXED types, and
named-record identities,
+ * which causes GenericDatumWriter to reject the record when serializing
against the original
+ * schema.
+ */
+ def createConverterToAvroFromInternalRow(
+ dataType: DataType,
+ avroSchema: Schema): (Any) => Any = {
+ val resolvedSchema = avroSchema.getType match {
+ case UNION =>
+ val nonNull = avroSchema.getTypes.asScala.filterNot(_.getType == NULL)
+ if (nonNull.size == 1) nonNull.head else avroSchema
+ case _ => avroSchema
+ }
+ dataType match {
+ case BinaryType =>
Review Comment:
nit: [optional] the converter turns BinaryType into a ByteBuffer, which
fails with ClassCastException ... cannot be cast to expected type F. The new
top-level ByteBuffer branch in AvroSerdes.serialize only helps a top-level
BYTES/FIXED column. It does not help FIXED nested in a record. The javadoc says
FIXED types are now handled, which overstates it.
##########
spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/SchemaConverters.scala:
##########
@@ -411,6 +414,115 @@ 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.
+ *
+ * This overload accepts the actual Avro Schema so that GenericRecord
instances are built against
+ * the original schema rather than one reconstructed from the Spark
StructType. The round-trip
+ * through convertStructToAvro loses nullable unions, FIXED types, and
named-record identities,
+ * which causes GenericDatumWriter to reject the record when serializing
against the original
+ * schema.
+ */
+ def createConverterToAvroFromInternalRow(
+ dataType: DataType,
+ avroSchema: Schema): (Any) => Any = {
+ val resolvedSchema = avroSchema.getType match {
+ case UNION =>
+ val nonNull = avroSchema.getTypes.asScala.filterNot(_.getType == NULL)
+ if (nonNull.size == 1) nonNull.head else avroSchema
+ case _ => avroSchema
+ }
+ 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 |
+ DateType =>
+ identity
+ case StringType =>
+ (item: Any) => if (item == null) null else
item.asInstanceOf[UTF8String].toString
Review Comment:
ENUM: the read side maps ENUM to StringType, so enum columns can be read but
not written. The writer gets a String and fails with AvroTypeException: value A
(a java.lang.String) is not a E
##########
spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/SchemaConverters.scala:
##########
Review Comment:
nit: claude suggested to add a Top-level nullable union for column like
["null","int"] or ["null", record], or fail with message like `Top-level union
with more than one non-null branch is not supported:`
if we fix this , you may also need to fix the read path in Utils.scala #
hbaseFieldToScalaType at lines 34–39. The write side stores top-level
primitives raw (Bytes.toBytes), so the read side decodes them the same way.
--
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]
