Copilot commented on code in PR #169:
URL: https://github.com/apache/hbase-connectors/pull/169#discussion_r4231811313
##########
spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/SchemaConverters.scala:
##########
@@ -411,6 +414,106 @@ 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)
Review Comment:
`createConverterToAvroFromInternalRow` returns a `ByteBuffer` for
`BinaryType`, but `AvroSerdes.serialize` (BYTES/FIXED branch) expects an
`Array[Byte]` and will `ClassCastException` when the *top-level* Avro schema
type is BYTES/FIXED. Fix by returning `Array[Byte]` here (or updating
`AvroSerdes.serialize` to accept `ByteBuffer` for BYTES/FIXED).
##########
spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/SchemaConverters.scala:
##########
@@ -411,6 +414,106 @@ 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 =>
+ 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
Review Comment:
This pattern match is non-exhaustive for Spark SQL types (e.g., `DateType`,
`MapType` with non-String keys, etc.), which will fail with a runtime
`MatchError`. Add an explicit default case that throws
`UnsupportedOperationException` (and implement any additional types you intend
to support). Also, converting `DecimalType` via `toString` is unlikely to be
compatible with standard Avro decimal logical types (bytes/fixed); if decimal
support is expected, it should follow Avro's decimal encoding instead of
stringifying.
##########
spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/SchemaConverters.scala:
##########
@@ -411,6 +414,106 @@ 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 =>
+ 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,
resolvedSchema.getElementType)
+ (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,
resolvedSchema.getValueType)
+ (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 avroFields = resolvedSchema.getFields.asScala
+ val fieldConverters = structType.fields.zipWithIndex.map { case
(field, i) =>
+ createConverterToAvroFromInternalRow(field.dataType,
avroFields(i).schema())
+ }
+ (item: Any) => {
+ if (item == null) {
+ null
+ } else {
+ val record = new Record(resolvedSchema)
+ val row = item.asInstanceOf[InternalRow]
+ var i = 0
+ while (i < structType.length) {
+ if (row.isNullAt(i)) {
+ record.put(structType.fields(i).name, null)
+ } else {
+ record.put(
+ structType.fields(i).name,
+ fieldConverters(i)(row.get(i,
structType.fields(i).dataType)))
+ }
+ i += 1
+ }
+ record
+ }
+ }
Review Comment:
The struct conversion assumes Spark struct field order/length matches the
Avro schema field order (`avroFields(i)`), and it uses Spark field names when
populating the Avro record. If there is any schema evolution/reordering (or
case differences), this can produce incorrect conversions or
`IndexOutOfBoundsException`/`AvroRuntimeException`. Prefer building converters
by matching Avro fields by name (using `resolvedSchema.getField(name)`), and
validate field presence to provide a clearer failure mode.
##########
spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBasePartitionReader.scala:
##########
@@ -282,6 +283,27 @@ class HBasePartitionReader(
DateTimeUtils.fromJavaTimestamp(t)
case dt: DecimalType =>
Decimal(value.asInstanceOf[java.math.BigDecimal], dt.precision,
dt.scale)
+ case ArrayType(elementType, _) =>
+ val seq = value.asInstanceOf[Seq[Any]]
+ new GenericArrayData(seq.map(convertToInternalRow(_,
elementType)).toArray)
+ case MapType(keyType, valueType, _) =>
+ val map = value.asInstanceOf[Map[Any, Any]]
+ val keys = map.keys.map(convertToInternalRow(_, keyType)).toArray
+ val values = map.values.map(convertToInternalRow(_, valueType)).toArray
Review Comment:
Building `keys` and `values` arrays via separate `map.keys` and `map.values`
iterations can misalign keys and values for map implementations where iteration
order is not guaranteed to be consistent across the two traversals. Iterate
entries once (e.g., over `map.iterator`) and append key/value pairs in lockstep
to ensure correct mapping.
##########
spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/SchemaConverters.scala:
##########
@@ -411,6 +414,106 @@ 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 =>
+ 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,
resolvedSchema.getElementType)
+ (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,
resolvedSchema.getValueType)
+ (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 avroFields = resolvedSchema.getFields.asScala
+ val fieldConverters = structType.fields.zipWithIndex.map { case
(field, i) =>
+ createConverterToAvroFromInternalRow(field.dataType,
avroFields(i).schema())
+ }
+ (item: Any) => {
+ if (item == null) {
+ null
+ } else {
+ val record = new Record(resolvedSchema)
+ val row = item.asInstanceOf[InternalRow]
+ var i = 0
+ while (i < structType.length) {
+ if (row.isNullAt(i)) {
+ record.put(structType.fields(i).name, null)
+ } else {
+ record.put(
+ structType.fields(i).name,
+ fieldConverters(i)(row.get(i,
structType.fields(i).dataType)))
+ }
+ i += 1
+ }
+ record
+ }
+ }
+ }
Review Comment:
This pattern match is non-exhaustive for Spark SQL types (e.g., `DateType`,
`MapType` with non-String keys, etc.), which will fail with a runtime
`MatchError`. Add an explicit default case that throws
`UnsupportedOperationException` (and implement any additional types you intend
to support). Also, converting `DecimalType` via `toString` is unlikely to be
compatible with standard Avro decimal logical types (bytes/fixed); if decimal
support is expected, it should follow Avro's decimal encoding instead of
stringifying.
##########
spark4/hbase-spark4/src/test/scala/org/apache/hadoop/hbase/spark/datasources/HBaseTableProviderSuite.scala:
##########
@@ -555,4 +566,252 @@ class HBaseTableProviderSuite extends AnyFunSuite with
BeforeAndAfterAll with Lo
assert(result(0).getAs[String]("name") == "Nora-Updated")
assert(result(0).getAs[String]("age") == "31")
}
+
+ // --- Avro schema field tests ---
+
+ val avroTableName = "test_avro"
+ val avroWriteTableName = "test_avro_write"
+
+ val avroSchemaString: String =
+ s"""{"namespace": "example.avro",
+ | "type": "record", "name": "User",
+ | "fields": [
+ | {"name": "name", "type": "string"},
+ | {"name": "favorite_number", "type": ["int", "null"]},
+ | {"name": "favorite_color", "type": ["string", "null"]}
+ | ]
+ |}""".stripMargin
+
+ val avroSchema: Schema = new Schema.Parser().parse(avroSchemaString)
+
+ val avroCatalog: String = s"""{
+ |"table":{"namespace":"default", "name":"$avroTableName"},
+ |"rowkey":"key",
+ |"columns":{
+ |"key":{"cf":"rowkey", "col":"key", "type":"string"},
+ |"col1":{"cf":"$columnFamily", "col":"col1", "avro":"avroSchema"}
+ |}
+ |}""".stripMargin
+
+ val avroWriteCatalog: String = s"""{
+ |"table":{"namespace":"default", "name":"$avroWriteTableName"},
+ |"rowkey":"key",
+ |"columns":{
+ |"key":{"cf":"rowkey", "col":"key", "type":"string"},
+ |"col1":{"cf":"$columnFamily", "col":"col1", "avro":"avroSchema"}
+ |}
+ |}""".stripMargin
+
+ private def populateAvroTestData(): Unit = {
+ val connection =
ConnectionFactory.createConnection(TEST_UTIL.getConfiguration)
+ val table = connection.getTable(TableName.valueOf(avroTableName))
+ try {
+ for (i <- 0 until 10) {
+ val key = f"arow$i%03d"
+ val user = new GenericData.Record(avroSchema)
+ user.put("name", s"name${"%03d".format(i)}")
+ user.put("favorite_number", i)
+ user.put("favorite_color", s"color${"%03d".format(i)}")
+ val avroBytes = AvroSerdes.serialize(user, avroSchema)
+ val put = new Put(Bytes.toBytes(key))
+ put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes("col1"),
avroBytes)
+ table.put(put)
+ }
+ } finally {
+ table.close()
+ connection.close()
+ }
+ }
+
+ private def loadAvroTable() = {
+ spark.read
+ .format("org.apache.hadoop.hbase.spark.datasources.HBaseTableProvider")
+ .options(Map(
+ "catalog" -> avroCatalog,
+ "avroSchema" -> avroSchemaString,
+ HBaseSparkConf.HBASE_CONFIG_LOCATION -> configFile.getAbsolutePath))
+ .load()
+ }
+
+ test("read Avro fields from HBase") {
+ val df = loadAvroTable()
+ assert(df.count() == 10)
+ df.printSchema()
Review Comment:
Avoid `printSchema()` in tests since it adds noisy stdout output and can
make CI logs harder to read. If you want schema validation, assert on the
expected schema (or log via the test logger only when debugging).
--
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]
