Copilot commented on code in PR #169:
URL: https://github.com/apache/hbase-connectors/pull/169#discussion_r4209012370
##########
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:
Reconstructing a schema from `StructType` loses required details from the
supplied Avro schema while serialization still uses `field.exeSchema`. For
example, a nullable nested record is created with the generated
`recordNamespace`/field-name schema, so `GenericDatumWriter` cannot resolve it
against the original named-record union; `FIXED` fields are likewise
reconstructed as `BYTES`. Build datum values recursively from the actual Avro
schema instead.
##########
spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBaseDataWriter.scala:
##########
@@ -128,21 +128,27 @@ 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 struct = row.getStruct(idx, field.dt.asInstanceOf[StructType].length)
+ val record = field.catalystToAvroFromInternalRow(struct)
+ AvroSerdes.serialize(record, field.exeSchema.get)
Review Comment:
An Avro-backed field is not necessarily a record: `AvroSerdes` explicitly
supports top-level primitive schemas, and `toSqlType` maps them to scalar
Catalyst types. Casting every such field's data type to `StructType` makes
writes of primitive Avro columns fail with `ClassCastException`; extract the
value according to `field.dt` and use `getStruct` only for records.
##########
spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBaseTable.scala:
##########
@@ -61,12 +61,12 @@ class HBaseTable(tableSchema: StructType, properties:
Map[String, String])
}
override def newScanBuilder(options: CaseInsensitiveStringMap): ScanBuilder
= {
- val mergedProps = properties ++ options.asScala.toMap
+ val mergedProps = properties ++ options.asCaseSensitiveMap().asScala.toMap
Review Comment:
This makes scan options case-sensitive even though Spark supplies a
`CaseInsensitiveStringMap`; mixed-case forms of established options (for
example `QUERY_CACHEDROWS` or `CATALOG`) will no longer reach the reader under
their canonical keys. Preserve both the normalized aliases and the
original-case keys needed by named Avro schema references.
This issue also appears on line 69 of the same file.
##########
spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBaseTableProvider.scala:
##########
@@ -60,7 +60,7 @@ class HBaseTableProvider extends TableProvider with
DataSourceRegister {
override def shortName(): String = "hbase"
override def inferSchema(options: CaseInsensitiveStringMap): StructType = {
- val params = options.asScala.toMap
+ val params = options.asCaseSensitiveMap().asScala.toMap
Review Comment:
Using only the case-sensitive view removes the case-insensitive behavior
promised by `CaseInsensitiveStringMap`. For example, `.option("CATALOG", ...)`
was previously normalized to `catalog`, but now `HBaseTableCatalog` fails at
`parameters("catalog")`. Keep normalized entries as aliases while retaining
original-case entries so catalog references such as `avroSchema` still resolve.
##########
spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBasePartitionReader.scala:
##########
@@ -282,6 +283,19 @@ class HBasePartitionReader(
DateTimeUtils.fromJavaTimestamp(t)
case dt: DecimalType =>
Decimal(value.asInstanceOf[java.math.BigDecimal], dt.precision,
dt.scale)
+ case structType: StructType =>
+ val row = value.asInstanceOf[Row]
+ val values = new Array[Any](structType.length)
+ var i = 0
+ while (i < structType.length) {
+ if (row.isNullAt(i)) {
+ values(i) = null
+ } else {
+ values(i) = convertToInternalRow(row.get(i),
structType.fields(i).dataType)
Review Comment:
Recursive record conversion still leaves Avro ARRAY and MAP fields as Scala
`Seq`/`Map` values returned by `createConverterToSQL`, but Catalyst struct
slots require `ArrayData`/`MapData`. Accessing an array or map nested in an
Avro record will therefore fail at runtime; add recursive ArrayType and MapType
conversions to Catalyst containers.
--
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]
