taklwu commented on code in PR #169:
URL: https://github.com/apache/hbase-connectors/pull/169#discussion_r4340587215
##########
spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBaseDataWriter.scala:
##########
@@ -128,21 +128,30 @@ class HBaseDataWriter(
private val MILLIS_PER_DAY = 86400000L
private def getValueBytes(row: InternalRow, idx: Int, field: Field):
Array[Byte] = {
- field.dt match {
- case BooleanType => Bytes.toBytes(row.getBoolean(idx))
- case ByteType => Array(row.getByte(idx))
- case ShortType => Bytes.toBytes(row.getShort(idx))
- case IntegerType => Bytes.toBytes(row.getInt(idx))
- case LongType => Bytes.toBytes(row.getLong(idx))
- case FloatType => Bytes.toBytes(row.getFloat(idx))
- case DoubleType => Bytes.toBytes(row.getDouble(idx))
- case StringType => Bytes.toBytes(row.getUTF8String(idx).toString)
- case BinaryType => row.getBinary(idx)
- case DateType => Bytes.toBytes(row.getInt(idx).toLong * MILLIS_PER_DAY)
- case TimestampType => Bytes.toBytes(row.getLong(idx) / 1000)
- case dt: DecimalType =>
- Bytes.toBytes(row.getDecimal(idx, dt.precision,
dt.scale).toJavaBigDecimal)
- case _ => throw new UnsupportedOperationException(s"unsupported data
type ${field.dt}")
+ if (field.exeSchema.isDefined) {
+ val value = field.dt match {
+ case st: StructType => row.getStruct(idx, st.length)
+ case _ => row.get(idx, field.dt)
+ }
+ val converted = field.catalystToAvroFromInternalRow(value)
+ AvroSerdes.serialize(converted, field.exeSchema.get)
Review Comment:
```suggestion
try {
val converted = field.catalystToAvroFromInternalRow(value)
AvroSerdes.serialize(converted, field.exeSchema.get)
} catch {
case e: IllegalArgumentException =>
throw new IllegalArgumentException(
s"Cannot encode column '${field.colName}' with its Avro schema:
${e.getMessage}", e)
}
```
##########
spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/SchemaConverters.scala:
##########
@@ -411,15 +414,143 @@ 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 =>
+ resolvedSchema.getType match {
+ case FIXED =>
+ (item: Any) =>
+ if (item == null) null
+ else new Fixed(resolvedSchema, item.asInstanceOf[Array[Byte]])
Review Comment:
```suggestion
case FIXED =>
val expected = resolvedSchema.getFixedSize
(item: Any) =>
if (item == null) null
else {
val bytes = item.asInstanceOf[Array[Byte]]
if (bytes.length != expected) {
throw new IllegalArgumentException(
s"Value of ${bytes.length} bytes does not match Avro fixed
type " +
s"'${resolvedSchema.getFullName}' of size $expected")
}
new Fixed(resolvedSchema, bytes)
}
```
--
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]
