This is an automated email from the ASF dual-hosted git repository.
wchevreuil pushed a commit to branch HBASE-30189
in repository https://gitbox.apache.org/repos/asf/hbase-connectors.git
The following commit(s) were added to refs/heads/HBASE-30189 by this push:
new 6d2198a Port the JSON catalog parsing that maps HBase tables to Spark
SQL schemas (#160)
6d2198a is described below
commit 6d2198ae89ec59fdb3ac706999ae6103afbc9e31
Author: Wellington Ramos Chevreuil <[email protected]>
AuthorDate: Fri Aug 28 10:13:30 2026 +0100
Port the JSON catalog parsing that maps HBase tables to Spark SQL schemas
(#160)
Co-authored-by: Claude Code (claude-opus-4-6) <[email protected]>
Signed-off-by: Peter Somogyi <[email protected]>
---
spark4/hbase-spark4/pom.xml | 4 +
.../hadoop/hbase/spark/SchemaConverters.scala | 445 +++++++++++++++++++++
.../spark/datasources/HBaseTableCatalog.scala | 394 ++++++++++++++++++
.../hadoop/hbase/spark/datasources/SerDes.scala | 35 ++
.../hadoop/hbase/spark/datasources/Utils.scala | 118 ++++++
5 files changed, 996 insertions(+)
diff --git a/spark4/hbase-spark4/pom.xml b/spark4/hbase-spark4/pom.xml
index ee9dff9..ee0c5c5 100644
--- a/spark4/hbase-spark4/pom.xml
+++ b/spark4/hbase-spark4/pom.xml
@@ -124,6 +124,10 @@
<groupId>org.apache.yetus</groupId>
<artifactId>audience-annotations</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.avro</groupId>
+ <artifactId>avro</artifactId>
+ </dependency>
<dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-common</artifactId>
diff --git
a/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/SchemaConverters.scala
b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/SchemaConverters.scala
new file mode 100644
index 0000000..df1491b
--- /dev/null
+++
b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/SchemaConverters.scala
@@ -0,0 +1,445 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.spark
+
+import java.io.ByteArrayInputStream
+import java.nio.ByteBuffer
+import java.sql.Timestamp
+import java.util
+import java.util.HashMap
+import org.apache.avro.{Schema, SchemaBuilder}
+import org.apache.avro.Schema.Type._
+import org.apache.avro.SchemaBuilder.BaseFieldTypeBuilder
+import org.apache.avro.SchemaBuilder.BaseTypeBuilder
+import org.apache.avro.SchemaBuilder.FieldAssembler
+import org.apache.avro.SchemaBuilder.FieldDefault
+import org.apache.avro.SchemaBuilder.RecordBuilder
+import org.apache.avro.generic.{GenericData, GenericDatumReader,
GenericDatumWriter, GenericRecord}
+import org.apache.avro.generic.GenericData.{Fixed, Record}
+import org.apache.avro.io._
+import org.apache.commons.io.output.ByteArrayOutputStream
+import org.apache.hadoop.hbase.util.Bytes
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.types._
+import org.apache.yetus.audience.InterfaceAudience
+import scala.jdk.CollectionConverters._
+
[email protected]
+abstract class AvroException(msg: String) extends Exception(msg)
+
[email protected]
+case class SchemaConversionException(msg: String) extends AvroException(msg)
+
+/**
+ * *
+ * On top level, the converters provide three high level interface.
+ * 1. toSqlType: This function takes an avro schema and returns a sql schema.
+ * 2. createConverterToSQL: Returns a function that is used to convert avro
types to their
+ * corresponding sparkSQL representations.
+ * 3. convertTypeToAvro: This function constructs converter function for a
given sparkSQL
+ * datatype. This is used in writing Avro records out to disk
+ */
[email protected]
+object SchemaConverters {
+
+ case class SchemaType(dataType: DataType, nullable: Boolean)
+
+ /**
+ * This function takes an avro schema and returns a sql schema.
+ */
+ def toSqlType(avroSchema: Schema): SchemaType = {
+ avroSchema.getType match {
+ case INT => SchemaType(IntegerType, nullable = false)
+ case STRING => SchemaType(StringType, nullable = false)
+ case BOOLEAN => SchemaType(BooleanType, nullable = false)
+ case BYTES => SchemaType(BinaryType, nullable = false)
+ case DOUBLE => SchemaType(DoubleType, nullable = false)
+ case FLOAT => SchemaType(FloatType, nullable = false)
+ case LONG => SchemaType(LongType, nullable = false)
+ case FIXED => SchemaType(BinaryType, nullable = false)
+ case ENUM => SchemaType(StringType, nullable = false)
+
+ case RECORD =>
+ val fields = avroSchema.getFields.asScala.map { f =>
+ val schemaType = toSqlType(f.schema())
+ StructField(f.name, schemaType.dataType, schemaType.nullable)
+ }
+ SchemaType(StructType(fields.toSeq), nullable = false)
+
+ case ARRAY =>
+ val schemaType = toSqlType(avroSchema.getElementType)
+ SchemaType(
+ ArrayType(schemaType.dataType, containsNull = schemaType.nullable),
+ nullable = false)
+
+ case MAP =>
+ val schemaType = toSqlType(avroSchema.getValueType)
+ SchemaType(
+ MapType(StringType, schemaType.dataType, valueContainsNull =
schemaType.nullable),
+ nullable = false)
+
+ case UNION =>
+ if (avroSchema.getTypes.asScala.exists(_.getType == NULL)) {
+ // In case of a union with null, eliminate it and make a recursive
call
+ val remainingUnionTypes =
+ avroSchema.getTypes.asScala.filterNot(_.getType == NULL).asJava
+ if (remainingUnionTypes.size == 1) {
+ toSqlType(remainingUnionTypes.get(0)).copy(nullable = true)
+ } else {
+ toSqlType(Schema.createUnion(remainingUnionTypes)).copy(nullable =
true)
+ }
+ } else
+ avroSchema.getTypes.asScala.map(_.getType).toSeq match {
+ case Seq(t1, t2) if Set(t1, t2) == Set(INT, LONG) =>
+ SchemaType(LongType, nullable = false)
+ case Seq(t1, t2) if Set(t1, t2) == Set(FLOAT, DOUBLE) =>
+ SchemaType(DoubleType, nullable = false)
+ case other =>
+ throw new SchemaConversionException(
+ s"This mix of union types is not supported: $other")
+ }
+
+ case other => throw new SchemaConversionException(s"Unsupported type
$other")
+ }
+ }
+
+ /**
+ * This function converts sparkSQL StructType into avro schema. This method
uses two other
+ * converter methods in order to do the conversion.
+ */
+ private def convertStructToAvro[T](
+ structType: StructType,
+ schemaBuilder: RecordBuilder[T],
+ recordNamespace: String): T = {
+ val fieldsAssembler: FieldAssembler[T] = schemaBuilder.fields()
+ structType.fields.foreach { field =>
+ val newField = fieldsAssembler.name(field.name).`type`()
+
+ if (field.nullable) {
+ convertFieldTypeToAvro(
+ field.dataType,
+ newField.nullable(),
+ field.name,
+ recordNamespace).noDefault
+ } else {
+ convertFieldTypeToAvro(field.dataType, newField, field.name,
recordNamespace).noDefault
+ }
+ }
+ fieldsAssembler.endRecord()
+ }
+
+ /**
+ * Returns a function that is used to convert avro types to their
+ * corresponding sparkSQL representations.
+ */
+ def createConverterToSQL(schema: Schema): Any => Any = {
+ schema.getType match {
+ case STRING | ENUM => (item: Any) => if (item == null) null else
item.toString
+ case INT | BOOLEAN | DOUBLE | FLOAT | LONG => identity
+ case FIXED =>
+ (item: Any) =>
+ if (item == null) {
+ null
+ } else {
+ item.asInstanceOf[Fixed].bytes().clone()
+ }
+ case BYTES =>
+ (item: Any) =>
+ if (item == null) {
+ null
+ } else {
+ val bytes = item.asInstanceOf[ByteBuffer]
+ val javaBytes = new Array[Byte](bytes.remaining)
+ bytes.get(javaBytes)
+ javaBytes
+ }
+ case RECORD =>
+ val fieldConverters = schema.getFields.asScala.map(f =>
createConverterToSQL(f.schema))
+ (item: Any) =>
+ if (item == null) {
+ null
+ } else {
+ val record = item.asInstanceOf[GenericRecord]
+ val converted = new Array[Any](fieldConverters.size)
+ var idx = 0
+ while (idx < fieldConverters.size) {
+ converted(idx) = fieldConverters(idx)(record.get(idx))
+ idx += 1
+ }
+ Row.fromSeq(converted.toSeq)
+ }
+ case ARRAY =>
+ val elementConverter = createConverterToSQL(schema.getElementType)
+ (item: Any) =>
+ if (item == null) {
+ null
+ } else {
+ try {
+
item.asInstanceOf[GenericData.Array[Any]].asScala.map(elementConverter).toSeq
+ } catch {
+ case _: Throwable =>
+
item.asInstanceOf[util.ArrayList[Any]].asScala.map(elementConverter).toSeq
+ }
+ }
+ case MAP =>
+ val valueConverter = createConverterToSQL(schema.getValueType)
+ (item: Any) =>
+ if (item == null) {
+ null
+ } else {
+ item
+ .asInstanceOf[HashMap[Any, Any]]
+ .asScala
+ .map(x => (x._1.toString, valueConverter(x._2)))
+ .toMap
+ }
+ case UNION =>
+ if (schema.getTypes.asScala.exists(_.getType == NULL)) {
+ val remainingUnionTypes =
schema.getTypes.asScala.filterNot(_.getType == NULL).asJava
+ if (remainingUnionTypes.size == 1) {
+ createConverterToSQL(remainingUnionTypes.get(0))
+ } else {
+ createConverterToSQL(Schema.createUnion(remainingUnionTypes))
+ }
+ } else
+ schema.getTypes.asScala.map(_.getType).toSeq match {
+ case Seq(t1, t2) if Set(t1, t2) == Set(INT, LONG) =>
+ (item: Any) => {
+ item match {
+ case l: Long => l
+ case i: Int => i.toLong
+ case null => null
+ }
+ }
+ case Seq(t1, t2) if Set(t1, t2) == Set(FLOAT, DOUBLE) =>
+ (item: Any) => {
+ item match {
+ case d: Double => d
+ case f: Float => f.toDouble
+ case null => null
+ }
+ }
+ case other =>
+ throw new SchemaConversionException(
+ s"This mix of union types is not supported (see README):
$other")
+ }
+ case other => throw new SchemaConversionException(s"invalid avro type:
$other")
+ }
+ }
+
+ /**
+ * This function is used to convert some sparkSQL type to avro type. Note
that this function won't
+ * be used to construct fields of avro record (convertFieldTypeToAvro is
used for that).
+ */
+ private def convertTypeToAvro[T](
+ dataType: DataType,
+ schemaBuilder: BaseTypeBuilder[T],
+ structName: String,
+ recordNamespace: String): T = {
+ dataType match {
+ case ByteType => schemaBuilder.intType()
+ case ShortType => schemaBuilder.intType()
+ case IntegerType => schemaBuilder.intType()
+ case LongType => schemaBuilder.longType()
+ case FloatType => schemaBuilder.floatType()
+ case DoubleType => schemaBuilder.doubleType()
+ case _: DecimalType => schemaBuilder.stringType()
+ case StringType => schemaBuilder.stringType()
+ case BinaryType => schemaBuilder.bytesType()
+ case BooleanType => schemaBuilder.booleanType()
+ case TimestampType => schemaBuilder.longType()
+
+ case ArrayType(elementType, _) =>
+ val builder =
getSchemaBuilder(dataType.asInstanceOf[ArrayType].containsNull)
+ val elementSchema = convertTypeToAvro(elementType, builder,
structName, recordNamespace)
+ schemaBuilder.array().items(elementSchema)
+
+ case MapType(StringType, valueType, _) =>
+ val builder =
getSchemaBuilder(dataType.asInstanceOf[MapType].valueContainsNull)
+ val valueSchema = convertTypeToAvro(valueType, builder, structName,
recordNamespace)
+ schemaBuilder.map().values(valueSchema)
+
+ case structType: StructType =>
+ convertStructToAvro(
+ structType,
+ schemaBuilder.record(structName).namespace(recordNamespace),
+ recordNamespace)
+
+ case other => throw new IllegalArgumentException(s"Unexpected type
$dataType.")
+ }
+ }
+
+ /**
+ * This function is used to construct fields of the avro record, where
schema of the field is
+ * specified by avro representation of dataType. Since builders for record
fields are different
+ * from those for everything else, we have to use a separate method.
+ */
+ private def convertFieldTypeToAvro[T](
+ dataType: DataType,
+ newFieldBuilder: BaseFieldTypeBuilder[T],
+ structName: String,
+ recordNamespace: String): FieldDefault[T, _] = {
+ dataType match {
+ case ByteType => newFieldBuilder.intType()
+ case ShortType => newFieldBuilder.intType()
+ case IntegerType => newFieldBuilder.intType()
+ case LongType => newFieldBuilder.longType()
+ case FloatType => newFieldBuilder.floatType()
+ case DoubleType => newFieldBuilder.doubleType()
+ case _: DecimalType => newFieldBuilder.stringType()
+ case StringType => newFieldBuilder.stringType()
+ case BinaryType => newFieldBuilder.bytesType()
+ case BooleanType => newFieldBuilder.booleanType()
+ case TimestampType => newFieldBuilder.longType()
+
+ case ArrayType(elementType, _) =>
+ val builder =
getSchemaBuilder(dataType.asInstanceOf[ArrayType].containsNull)
+ val elementSchema = convertTypeToAvro(elementType, builder,
structName, recordNamespace)
+ newFieldBuilder.array().items(elementSchema)
+
+ case MapType(StringType, valueType, _) =>
+ val builder =
getSchemaBuilder(dataType.asInstanceOf[MapType].valueContainsNull)
+ val valueSchema = convertTypeToAvro(valueType, builder, structName,
recordNamespace)
+ newFieldBuilder.map().values(valueSchema)
+
+ case structType: StructType =>
+ convertStructToAvro(
+ structType,
+ newFieldBuilder.record(structName).namespace(recordNamespace),
+ recordNamespace)
+
+ case other => throw new IllegalArgumentException(s"Unexpected type
$dataType.")
+ }
+ }
+
+ private def getSchemaBuilder(isNullable: Boolean): BaseTypeBuilder[Schema] =
{
+ if (isNullable) {
+ SchemaBuilder.builder().nullable()
+ } else {
+ SchemaBuilder.builder()
+ }
+ }
+
+ /**
+ * This function constructs converter function for a given sparkSQL
datatype. This is used in
+ * writing Avro records out to disk
+ */
+ def createConverterToAvro(
+ 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 | StringType |
+ BooleanType =>
+ identity
+ case _: DecimalType => (item: Any) => if (item == null) null else
item.toString
+ case TimestampType =>
+ (item: Any) => if (item == null) null else
item.asInstanceOf[Timestamp].getTime
+ case ArrayType(elementType, _) =>
+ val elementConverter = createConverterToAvro(elementType, structName,
recordNamespace)
+ (item: Any) => {
+ if (item == null) {
+ null
+ } else {
+ val sourceArray = item.asInstanceOf[Seq[Any]]
+ val sourceArraySize = sourceArray.size
+ val targetArray = new util.ArrayList[Any](sourceArraySize)
+ var idx = 0
+ while (idx < sourceArraySize) {
+ targetArray.add(elementConverter(sourceArray(idx)))
+ idx += 1
+ }
+ targetArray
+ }
+ }
+ case MapType(StringType, valueType, _) =>
+ val valueConverter = createConverterToAvro(valueType, structName,
recordNamespace)
+ (item: Any) => {
+ if (item == null) {
+ null
+ } else {
+ val javaMap = new HashMap[String, Any]()
+ item.asInstanceOf[Map[String, Any]].foreach {
+ case (key, value) =>
+ javaMap.put(key, valueConverter(value))
+ }
+ 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 => createConverterToAvro(field.dataType, field.name,
recordNamespace))
+ (item: Any) => {
+ if (item == null) {
+ null
+ } else {
+ val record = new Record(schema)
+ val convertersIterator = fieldConverters.iterator
+ val fieldNamesIterator =
dataType.asInstanceOf[StructType].fieldNames.iterator
+ val rowIterator = item.asInstanceOf[Row].toSeq.iterator
+
+ while (convertersIterator.hasNext) {
+ val converter = convertersIterator.next()
+ record.put(fieldNamesIterator.next(),
converter(rowIterator.next()))
+ }
+ record
+ }
+ }
+ }
+ }
+}
+
[email protected]
+object AvroSerdes {
+ // We only handle top level is record or primary type now
+ def serialize(input: Any, schema: Schema): Array[Byte] = {
+ schema.getType match {
+ case BOOLEAN => Bytes.toBytes(input.asInstanceOf[Boolean])
+ case BYTES | FIXED => input.asInstanceOf[Array[Byte]]
+ case DOUBLE => Bytes.toBytes(input.asInstanceOf[Double])
+ case FLOAT => Bytes.toBytes(input.asInstanceOf[Float])
+ case INT => Bytes.toBytes(input.asInstanceOf[Int])
+ case LONG => Bytes.toBytes(input.asInstanceOf[Long])
+ case STRING => Bytes.toBytes(input.asInstanceOf[String])
+ case RECORD =>
+ val gr = input.asInstanceOf[GenericRecord]
+ val writer2 = new GenericDatumWriter[GenericRecord](schema)
+ val bao2 = new ByteArrayOutputStream()
+ val encoder2: BinaryEncoder =
EncoderFactory.get().directBinaryEncoder(bao2, null)
+ writer2.write(gr, encoder2)
+ bao2.toByteArray()
+ case _ => throw new Exception(s"unsupported data type ${schema.getType}")
+ }
+ }
+
+ def deserialize(input: Array[Byte], schema: Schema): GenericRecord = {
+ val reader2: DatumReader[GenericRecord] = new
GenericDatumReader[GenericRecord](schema)
+ val bai2 = new ByteArrayInputStream(input)
+ val decoder2: BinaryDecoder =
DecoderFactory.get().directBinaryDecoder(bai2, null)
+ reader2.read(null, decoder2)
+ }
+}
diff --git
a/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBaseTableCatalog.scala
b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBaseTableCatalog.scala
new file mode 100644
index 0000000..ddd9d49
--- /dev/null
+++
b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBaseTableCatalog.scala
@@ -0,0 +1,394 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.spark.datasources
+
+import org.apache.avro.Schema
+import org.apache.hadoop.hbase.spark.{Logging, SchemaConverters}
+import org.apache.hadoop.hbase.util.Bytes
+import org.apache.spark.sql.types._
+import org.apache.yetus.audience.InterfaceAudience
+import org.json4s.DefaultFormats
+import org.json4s.Formats
+import org.json4s.jackson.JsonMethods
+import scala.collection.mutable
+import scala.jdk.CollectionConverters._
+
[email protected]
+case class Field(
+ colName: String,
+ cf: String,
+ col: String,
+ sType: Option[String] = None,
+ avroSchema: Option[String] = None,
+ serdes: Option[SerDes] = None,
+ len: Int = -1)
+ extends Logging {
+ override def toString: String = s"$colName $cf $col"
+ val isRowKey: Boolean = cf == HBaseTableCatalog.rowKey
+ var start: Int = _
+ def schema: Option[Schema] = avroSchema.map { x =>
+ logDebug(s"avro: $x")
+ val p = new Schema.Parser
+ p.parse(x)
+ }
+
+ lazy val exeSchema: Option[Schema] = schema
+
+ lazy val avroToCatalyst: Option[Any => Any] = {
+ schema.map(SchemaConverters.createConverterToSQL(_))
+ }
+
+ lazy val catalystToAvro: (Any) => Any = {
+ SchemaConverters.createConverterToAvro(dt, colName, "recordNamespace")
+ }
+
+ def cfBytes: Array[Byte] = {
+ if (isRowKey) {
+ Bytes.toBytes("")
+ } else {
+ Bytes.toBytes(cf)
+ }
+ }
+ def colBytes: Array[Byte] = {
+ if (isRowKey) {
+ Bytes.toBytes("key")
+ } else {
+ Bytes.toBytes(col)
+ }
+ }
+
+ val dt: DataType = {
+ sType.map(DataTypeParserWrapper.parse(_)).getOrElse {
+ schema.map { x => SchemaConverters.toSqlType(x).dataType }.get
+ }
+ }
+
+ var length: Int = {
+ if (len == -1) {
+ dt match {
+ case BinaryType | StringType => -1
+ case BooleanType => Bytes.SIZEOF_BOOLEAN
+ case ByteType => 1
+ case DoubleType => Bytes.SIZEOF_DOUBLE
+ case FloatType => Bytes.SIZEOF_FLOAT
+ case IntegerType => Bytes.SIZEOF_INT
+ case LongType => Bytes.SIZEOF_LONG
+ case ShortType => Bytes.SIZEOF_SHORT
+ case _ => -1
+ }
+ } else {
+ len
+ }
+ }
+
+ override def equals(other: Any): Boolean = other match {
+ case that: Field =>
+ colName == that.colName && cf == that.cf && col == that.col
+ case _ => false
+ }
+}
+
[email protected]
+case class RowKey(k: String) {
+ val keys: Array[String] = k.split(":")
+ var fields: Seq[Field] = _
+ var varLength = false
+ def length: Int = {
+ if (varLength) {
+ -1
+ } else {
+ fields.foldLeft(0) { case (x, y) => x + y.length }
+ }
+ }
+}
+
[email protected]
+case class SchemaMap(map: mutable.HashMap[String, Field]) {
+ def toFields: Seq[StructField] = map.map {
+ case (name, field) =>
+ StructField(name, field.dt)
+ }.toSeq
+
+ def fields: Iterable[Field] = map.values
+
+ def getField(name: String): Field = map(name)
+}
+
[email protected]
+case class HBaseTableCatalog(
+ namespace: String,
+ name: String,
+ row: RowKey,
+ sMap: SchemaMap,
+ @transient params: Map[String, String])
+ extends Logging {
+ def toDataType: StructType = StructType(sMap.toFields)
+ def getField(name: String): Field = sMap.getField(name)
+ def getRowKey: Seq[Field] = row.fields
+ def getPrimaryKey: String = row.keys(0)
+ def getColumnFamilies: Seq[String] = {
+ sMap.fields.map(_.cf).filter(_ != HBaseTableCatalog.rowKey).toSeq.distinct
+ }
+
+ def get(key: String): Option[String] = params.get(key)
+
+ // Setup the start and length for each dimension of row key at runtime.
+ def dynSetupRowKey(rowKey: Array[Byte]): Unit = {
+ logDebug(s"length: ${rowKey.length}")
+ if (row.varLength) {
+ var start = 0
+ row.fields.foreach { f =>
+ logDebug(s"start: $start")
+ f.start = start
+ f.length = {
+ // If the length is not defined
+ if (f.length == -1) {
+ f.dt match {
+ case StringType =>
+ var pos = rowKey.indexOf(HBaseTableCatalog.delimiter, start)
+ if (pos == -1 || pos > rowKey.length) {
+ pos = rowKey.length
+ }
+ pos - start
+ // We don't know the length, assume it extend to the end of the
rowkey.
+ case _ => rowKey.length - start
+ }
+ } else {
+ f.length
+ }
+ }
+ start += f.length
+ }
+ }
+ }
+
+ def initRowKey: Unit = {
+ val fields = sMap.fields.filter(_.cf == HBaseTableCatalog.rowKey)
+ row.fields = row.keys.flatMap(n => fields.find(_.col == n)).toSeq
+ // The length is determined at run time if it is string or binary and the
length is undefined.
+ if (row.fields.filter(_.length == -1).isEmpty) {
+ var start = 0
+ row.fields.foreach { f =>
+ f.start = start
+ start += f.length
+ }
+ } else {
+ row.varLength = true
+ }
+ }
+ initRowKey
+}
+
+/**
+ * Directly ported from spark3 compatible module, this is intended for
internal use in spark 4.
+ * With DataSource V2, the entry point is always a TableProvider — that's the
class Spark instantiates
+ * when you call .format(...). It's the V2 equivalent of what DefaultSource (a
RelationProvider) was in V1.
+ *
+ * The flow is:
+ * 1. .format("org.apache.hadoop.hbase.spark.datasources.HBaseTableProvider")
— Spark instantiates this class
+ * 2. HBaseTableProvider reads the "catalog" option and parses it via
HBaseTableCatalog(properties) internally
+ * 3. It returns an HBaseTable, which returns an HBaseScanBuilder, and so on
down the chain
+ *
+ * For example, in spark 3, client code looks like:
+ * <code>
+ * val catalog = "..." // The actual json catalog string
+ * ...
+ * val df = spark.read.option(HBaseTableCatalog.tableCatalog, catalog)
+ * .format("org.apache.hadoop.hbase.spark")
+ * .load()
+ * ...
+ * </code>
+ *
+ * Whilst in spark 4, client code looks like:
+ * <code>
+ * val catalog = "..." // The actual json catalog string
+ * ...
+ * val df =
spark.read.format("org.apache.hadoop.hbase.spark.datasources.HBaseTableProvider")
+ * .option("catalog", catalog)
+ * ...
+ * </code>
+ */
[email protected]
+object HBaseTableCatalog {
+ val newTable = "newtable"
+ val regionStart = "regionStart"
+ val defaultRegionStart = "aaaaaaa"
+ val regionEnd = "regionEnd"
+ val defaultRegionEnd = "zzzzzzz"
+ val tableCatalog = "catalog"
+ val rowKey = "rowkey"
+ val table = "table"
+ val nameSpace = "namespace"
+ val tableName = "name"
+ val columns = "columns"
+ val cf = "cf"
+ val col = "col"
+ val `type` = "type"
+ val avro = "avro"
+ val delimiter: Byte = 0
+ val serdes = "serdes"
+ val length = "length"
+
+ /**
+ * User provide table schema definition
+ * {"tablename":"name", "rowkey":"key1:key2",
+ * "columns":{"col1":{"cf":"cf1", "col":"col1", "type":"type1"},
+ * "col2":{"cf":"cf2", "col":"col2", "type":"type2"}}}
+ * Note that any col in the rowKey, there has to be one corresponding col
defined in columns
+ */
+ def apply(params: Map[String, String]): HBaseTableCatalog = {
+ val parameters = convert(params)
+ val jString = parameters(tableCatalog)
+ implicit val formats: Formats = DefaultFormats
+ val map = JsonMethods.parse(jString)
+ val tableMeta = map \ table
+ val nSpace = (tableMeta \ nameSpace).extractOrElse("default")
+ val tName = (tableMeta \ tableName).extract[String]
+ val cIter = (map \ columns).extract[Map[String, Map[String, String]]]
+ val schemaMap = mutable.HashMap.empty[String, Field]
+ cIter.foreach {
+ case (name, column) =>
+ val sd = {
+ column
+ .get(serdes)
+ .asInstanceOf[Option[String]]
+ .map(n =>
Class.forName(n).getDeclaredConstructor().newInstance().asInstanceOf[SerDes])
+ }
+ val len = column.get(length).map(_.toInt).getOrElse(-1)
+ val sAvro = column.get(avro).map(parameters(_))
+ val f = Field(
+ name,
+ column.getOrElse(cf, rowKey),
+ column.get(col).get,
+ column.get(`type`),
+ sAvro,
+ sd,
+ len)
+ schemaMap.+=((name, f))
+ }
+ val rKey = RowKey((map \ rowKey).extract[String])
+ HBaseTableCatalog(nSpace, tName, rKey, SchemaMap(schemaMap), parameters)
+ }
+
+ val TABLE_KEY: String = "hbase.table"
+ val SCHEMA_COLUMNS_MAPPING_KEY: String = "hbase.columns.mapping"
+
+ //Directly ported from spark 3 compatible module, which was marking it as
deprecated, but since this is
+ // used by apply(), had made it private access.
+ private def convert(parameters: Map[String, String]): Map[String, String] = {
+ val nsTableName = parameters.get(TABLE_KEY).orNull
+ // if the hbase.table is not defined, we assume it is json format already.
+ if (nsTableName == null) return parameters
+ val tableParts = nsTableName.trim.split(':')
+ val tableNamespace = if (tableParts.length == 1) {
+ "default"
+ } else if (tableParts.length == 2) {
+ tableParts(0)
+ } else {
+ throw new IllegalArgumentException(
+ "Invalid table name '" + nsTableName +
+ "' should be '<namespace>:<name>' or '<name>' ")
+ }
+ val tblName = tableParts(tableParts.length - 1)
+ val schemaMappingString = parameters.getOrElse(SCHEMA_COLUMNS_MAPPING_KEY,
"")
+ val schemaMap = generateSchemaMappingMap(schemaMappingString).asScala.map(
+ _._2.asInstanceOf[SchemaQualifierDefinition])
+
+ val rowkey = schemaMap
+ .filter {
+ _.columnFamily == "rowkey"
+ }
+ .map(_.columnName)
+ val cols = schemaMap.map { x =>
+ s""""${x.columnName}":{"cf":"${x.columnFamily}", "col":"${x.qualifier}",
"type":"${x.colType}"}""".stripMargin
+ }
+ val jsonCatalog =
+ s"""{
+ |"table":{"namespace":"${tableNamespace}", "name":"${tblName}"},
+ |"rowkey":"${rowkey.mkString(":")}",
+ |"columns":{
+ |${cols.mkString(",")}
+ |}
+ |}
+ """.stripMargin
+ parameters ++ Map(HBaseTableCatalog.tableCatalog -> jsonCatalog)
+ }
+
+ /**
+ * Reads the SCHEMA_COLUMNS_MAPPING_KEY and converts it to a map of
+ * SchemaQualifierDefinitions with the original sql column name as the key
+ *
+ * @param schemaMappingString The schema mapping string from the SparkSQL map
+ * @return A map of definitions keyed by the SparkSQL column name
+ */
+ @InterfaceAudience.Private
+ def generateSchemaMappingMap(
+ schemaMappingString: String): java.util.HashMap[String,
SchemaQualifierDefinition] = {
+ try {
+ val columnDefinitions = schemaMappingString.split(',')
+ val resultingMap = new java.util.HashMap[String,
SchemaQualifierDefinition]()
+ columnDefinitions.map { cd =>
+ val parts = cd.trim.split(' ')
+ // Make sure we get three parts
+ // <ColumnName> <ColumnType> <ColumnFamily:Qualifier>
+ if (parts.length == 3) {
+ val hbaseDefinitionParts = if (parts(2).charAt(0) == ':') {
+ Array[String]("rowkey", parts(0))
+ } else {
+ parts(2).split(':')
+ }
+ resultingMap.put(
+ parts(0),
+ new SchemaQualifierDefinition(
+ parts(0),
+ parts(1),
+ hbaseDefinitionParts(0),
+ hbaseDefinitionParts(1)))
+ } else {
+ throw new IllegalArgumentException(
+ "Invalid value for schema mapping '" + cd +
+ "' should be '<columnName> <columnType>
<columnFamily>:<qualifier>' " +
+ "for columns and '<columnName> <columnType> :<qualifier>' for
rowKeys")
+ }
+ }
+ resultingMap
+ } catch {
+ case e: Exception =>
+ throw new IllegalArgumentException(
+ "Invalid value for " + SCHEMA_COLUMNS_MAPPING_KEY +
+ " '" +
+ schemaMappingString + "'",
+ e)
+ }
+ }
+}
+
+/**
+ * Construct to contains column data that spend SparkSQL and HBase
+ *
+ * @param columnName SparkSQL column name
+ * @param colType SparkSQL column type
+ * @param columnFamily HBase column family
+ * @param qualifier HBase qualifier name
+ */
[email protected]
+case class SchemaQualifierDefinition(
+ columnName: String,
+ colType: String,
+ columnFamily: String,
+ qualifier: String)
diff --git
a/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/SerDes.scala
b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/SerDes.scala
new file mode 100644
index 0000000..bf01901
--- /dev/null
+++
b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/SerDes.scala
@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.spark.datasources
+
+import org.apache.hadoop.hbase.util.Bytes
+import org.apache.yetus.audience.InterfaceAudience
+
[email protected]
+trait SerDes {
+ def serialize(value: Any): Array[Byte]
+ def deserialize(bytes: Array[Byte], start: Int, end: Int): Any
+}
+
[email protected]
+class DoubleSerDes extends SerDes {
+ override def serialize(value: Any): Array[Byte] =
Bytes.toBytes(value.asInstanceOf[Double])
+ override def deserialize(bytes: Array[Byte], start: Int, end: Int): Any = {
+ Bytes.toDouble(bytes, start)
+ }
+}
diff --git
a/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/Utils.scala
b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/Utils.scala
new file mode 100644
index 0000000..ea6f3eb
--- /dev/null
+++
b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/Utils.scala
@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.spark.datasources
+
+import java.sql.{Date, Timestamp}
+import org.apache.hadoop.hbase.spark.AvroSerdes
+import org.apache.hadoop.hbase.util.Bytes
+import org.apache.spark.sql.types._
+import org.apache.yetus.audience.InterfaceAudience
+
+/**
+ * Parses the hbase field to it's corresponding
+ * scala type which can then be put into a Spark GenericRow
+ * which is then automatically converted by Spark.
+ */
[email protected]
+object Utils {
+
+ def hbaseFieldToScalaType(f: Field, src: Array[Byte], offset: Int, length:
Int): Any = {
+ if (f.exeSchema.isDefined) {
+ // If we have avro schema defined, use it to get record, and then
convert them to catalyst data type
+ val m = AvroSerdes.deserialize(src, f.exeSchema.get)
+ val n = f.avroToCatalyst.map(_(m))
+ n.get
+ } else {
+ // Fall back to atomic type
+ f.dt match {
+ case BooleanType => src(offset) != 0
+ case ByteType => src(offset)
+ case ShortType => Bytes.toShort(src, offset)
+ case IntegerType => Bytes.toInt(src, offset)
+ case LongType => Bytes.toLong(src, offset)
+ case FloatType => Bytes.toFloat(src, offset)
+ case DoubleType => Bytes.toDouble(src, offset)
+ case DateType => new Date(Bytes.toLong(src, offset))
+ case TimestampType => new Timestamp(Bytes.toLong(src, offset))
+ case StringType => Bytes.toString(src, offset, length)
+ case BinaryType =>
+ val newArray = new Array[Byte](length)
+ System.arraycopy(src, offset, newArray, 0, length)
+ newArray
+ case _: DecimalType => Bytes.toBigDecimal(src, offset, length)
+ case _ => throw new Exception(s"unsupported data type ${f.dt}")
+ }
+ }
+ }
+
+ // convert input to data type
+ def toBytes(input: Any, field: Field): Array[Byte] = {
+ if (field.schema.isDefined) {
+ // Here we assume the top level type is structType
+ val record = field.catalystToAvro(input)
+ AvroSerdes.serialize(record, field.schema.get)
+ } else {
+ field.dt match {
+ case BooleanType => Bytes.toBytes(input.asInstanceOf[Boolean])
+ case ByteType => Array(input.asInstanceOf[Number].byteValue)
+ case ShortType => Bytes.toBytes(input.asInstanceOf[Number].shortValue)
+ case IntegerType => Bytes.toBytes(input.asInstanceOf[Number].intValue)
+ case LongType => Bytes.toBytes(input.asInstanceOf[Number].longValue)
+ case FloatType => Bytes.toBytes(input.asInstanceOf[Number].floatValue)
+ case DoubleType =>
Bytes.toBytes(input.asInstanceOf[Number].doubleValue)
+ case DateType | TimestampType =>
Bytes.toBytes(input.asInstanceOf[java.util.Date].getTime)
+ case StringType => Bytes.toBytes(input.toString)
+ case BinaryType => input.asInstanceOf[Array[Byte]]
+ case _: DecimalType =>
Bytes.toBytes(input.asInstanceOf[java.math.BigDecimal])
+ case _ => throw new Exception(s"unsupported data type ${field.dt}")
+ }
+ }
+ }
+
+ // increment Byte array's value by 1
+ def incrementByteArray(array: Array[Byte]): Array[Byte] = {
+ if (array.length == 0) {
+ return null
+ }
+ var index = -1
+ var a = array.length - 1
+
+ while (a >= 0) {
+ if (array(a) != (-1).toByte) {
+ index = a
+ a = -1
+ }
+ a = a - 1
+ }
+
+ if (index < 0) {
+ return null
+ }
+ val returnArray = new Array[Byte](array.length)
+
+ for (a <- 0 until index) {
+ returnArray(a) = array(a)
+ }
+ returnArray(index) = (array(index) + 1).toByte
+ for (a <- index + 1 until array.length) {
+ returnArray(a) = 0.toByte
+ }
+
+ returnArray
+ }
+}
