You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by wc...@apache.org on 2026/09/16 13:35:37 UTC
(hbase-connectors) branch HBASE-30189 updated: Implementing DataSource V2 Read Path (#162)
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 3778895 Implementing DataSource V2 Read Path (#162)
3778895 is described below
commit 3778895e1ae9e7fa81052b53350a9b14e0f3ba1a
Author: Wellington Ramos Chevreuil <wc...@cloudera.com>
AuthorDate: Wed Sep 16 14:35:32 2026 +0100
Implementing DataSource V2 Read Path (#162)
Signed-off-by: Tak Lon (Stephen) Wu <ta...@apache.org>
Signed-off-by: Dávid Paksy <pa...@apache.org>Signed-off-by: Peter Somogyi <ps...@apache.org>
Co-authored-by: Claude Code (claude-opus-4-6) <no...@anthropic.com>
---
dev-support/jenkins/Dockerfile | 2 +-
dev-support/jenkins/Jenkinsfile | 2 +-
...org.apache.spark.sql.sources.DataSourceRegister | 1 +
.../hadoop/hbase/spark/datasources/Bound.scala | 2 +-
.../hbase/spark/datasources/HBaseBatch.scala | 123 ++++++
.../{SerDes.scala => HBaseInputPartition.scala} | 28 +-
.../spark/datasources/HBasePartitionReader.scala | 435 +++++++++++++++++++++
.../datasources/HBasePartitionReaderFactory.scala | 64 +++
.../hadoop/hbase/spark/datasources/HBaseScan.scala | 151 +++++++
.../hbase/spark/datasources/HBaseScanBuilder.scala | 115 ++++++
.../hbase/spark/datasources/HBaseTable.scala | 56 +++
.../spark/datasources/HBaseTableProvider.scala | 74 ++++
.../hadoop/hbase/spark/datasources/ScanRange.scala | 222 +++++++++++
.../hadoop/hbase/spark/datasources/SerDes.scala | 2 +-
.../spark/datasources/HBaseScanBuilderSuite.scala | 161 ++++++++
.../spark/datasources/HBaseTableCatalogSuite.scala | 189 +++++++++
.../datasources/HBaseTableProviderSuite.scala | 219 +++++++++++
.../hbase/spark/datasources/ScanRangeSuite.scala | 162 ++++++++
18 files changed, 1992 insertions(+), 16 deletions(-)
diff --git a/dev-support/jenkins/Dockerfile b/dev-support/jenkins/Dockerfile
index 45551b9..347d313 100644
--- a/dev-support/jenkins/Dockerfile
+++ b/dev-support/jenkins/Dockerfile
@@ -17,7 +17,7 @@
# Dockerfile for hbase-connectors pre-commit build.
# https://builds.apache.org/job/PreCommit-HBASE-CONNECTORS-Build
-FROM maven:3.8-jdk-8
+FROM maven:3.9-eclipse-temurin-8
# hadolint ignore=DL3008
RUN apt-get -q update && apt-get -q install --no-install-recommends -y \
diff --git a/dev-support/jenkins/Jenkinsfile b/dev-support/jenkins/Jenkinsfile
index d788de9..df98643 100644
--- a/dev-support/jenkins/Jenkinsfile
+++ b/dev-support/jenkins/Jenkinsfile
@@ -40,7 +40,7 @@ pipeline {
YETUS_DRIVER_REL = "${SRC_REL}/dev-support/jenkins/jenkins_precommit_github_yetus.sh"
ARCHIVE_PATTERN_LIST = 'TEST-*.xml,org.apache.h*.txt,*.dumpstream,*.dump'
BUILD_URL_ARTIFACTS = "artifact/${WORKDIR_REL}/${PATCH_REL}"
- SET_JAVA_HOME = '/usr/local/openjdk-8'
+ SET_JAVA_HOME = '/opt/java/openjdk'
WORKDIR_REL = 'yetus-precommit-check'
WORKDIR = "${WORKSPACE}/${WORKDIR_REL}"
SOURCEDIR = "${WORKDIR}/${SRC_REL}"
diff --git a/spark4/hbase-spark4/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister b/spark4/hbase-spark4/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister
new file mode 100644
index 0000000..be8c3b1
--- /dev/null
+++ b/spark4/hbase-spark4/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister
@@ -0,0 +1 @@
+org.apache.hadoop.hbase.spark.datasources.HBaseTableProvider
diff --git a/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/Bound.scala b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/Bound.scala
index 7b12982..c00dd8f 100644
--- a/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/Bound.scala
+++ b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/Bound.scala
@@ -85,7 +85,7 @@ object Ranges {
.getOrElse(-1)
}
.getOrElse(-1)
- if (c < 0) {
+ if (c < 0 || (c == 0 && lower.exists(_.inc) && upper.exists(_.inc))) {
Some(Range(lower, upper))
} else {
None
diff --git a/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBaseBatch.scala b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBaseBatch.scala
new file mode 100644
index 0000000..b6fc7aa
--- /dev/null
+++ b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBaseBatch.scala
@@ -0,0 +1,123 @@
+/*
+ * 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.fs.Path
+import org.apache.hadoop.hbase.{HBaseConfiguration, TableName}
+import org.apache.hadoop.hbase.spark.{HBaseConnectionCache, Logging}
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.connector.read.{Batch, InputPartition, PartitionReaderFactory}
+import org.apache.spark.sql.sources._
+import org.apache.spark.sql.types.StructType
+import org.apache.yetus.audience.InterfaceAudience
+
+/**
+ * This is a new class in the spark4 module. Implements Batch.
+ * Responsible for physical planning: splits the read into partitions.
+ * Calls RegionLocator.getStartEndKeys() to discover HBase regions,
+ * intersects them with the row key filter's scan ranges, and produces an array of InputPartition objects.
+ *
+ * In the spark 3 DS V1 model, this logic was inside HBaseTableScanRDD.getPartitions().
+ *
+ * @param requiredSchema
+ * @param properties
+ * @param catalog
+ * @param rowKeyFilter
+ * @param pushedFilters
+ * @param encoderClsName
+ */
+@InterfaceAudience.Private
+class HBaseBatch(
+ requiredSchema: StructType,
+ properties: Map[String, String],
+ catalog: HBaseTableCatalog,
+ rowKeyFilter: RowKeyFilter,
+ pushedFilters: Array[Filter],
+ encoderClsName: String)
+ extends Batch
+ with Logging {
+
+ override def planInputPartitions(): Array[InputPartition] = {
+ val hadoopConf = SparkSession.active.sparkContext.hadoopConfiguration
+ val conf = HBaseConfiguration.create(hadoopConf)
+ val configResources = properties.get(HBaseSparkConf.HBASE_CONFIG_LOCATION)
+ configResources.foreach(_.split(",").foreach(r => conf.addResource(new Path(r))))
+
+ val connection = HBaseConnectionCache.getConnection(conf)
+ try {
+ val tableName = s"${catalog.namespace}:${catalog.name}"
+ val regionLocator = connection.getRegionLocator(TableName.valueOf(tableName))
+ try {
+ val keys = regionLocator.getStartEndKeys
+ val startKeys = keys.getFirst
+ val endKeys = keys.getSecond
+
+ val regions = startKeys.zip(endKeys).zipWithIndex.map { case ((start, end), idx) =>
+ HBaseRegion(idx, Some(start), Some(end),
+ Some(regionLocator.getRegionLocation(start).getHostname))
+ }
+
+ val scanRanges = rowKeyFilter.ranges.toSeq.map { sr =>
+ Range(
+ Option(sr.lowerBound).filter(_.nonEmpty).map(Bound(_, sr.isLowerBoundEqualTo)),
+ Option(sr.upperBound).map(Bound(_, sr.isUpperBoundEqualTo)))
+ }
+ val points = rowKeyFilter.points.toSeq
+
+ regions.flatMap { region =>
+ val regionRange = Range(region)
+ val intersectedRanges = Ranges.and(regionRange, scanRanges)
+ val intersectedPoints = Points.and(regionRange, points)
+
+ if (intersectedRanges.nonEmpty || intersectedPoints.nonEmpty) {
+ Some(HBaseInputPartition(
+ region.index, intersectedRanges, intersectedPoints, region.server): InputPartition)
+ } else {
+ None
+ }
+ }
+ } finally {
+ regionLocator.close()
+ }
+ } finally {
+ connection.close()
+ }
+ }
+
+ override def createReaderFactory(): PartitionReaderFactory = {
+ val usePushDownColumnFilter = properties
+ .get(HBaseSparkConf.PUSHDOWN_COLUMN_FILTER)
+ .map(_.toBoolean)
+ .getOrElse(HBaseSparkConf.DEFAULT_PUSHDOWN_COLUMN_FILTER)
+
+ val hadoopConf = SparkSession.active.sparkContext.hadoopConfiguration
+ val hbaseConf = HBaseConfiguration.create(hadoopConf)
+ properties.get(HBaseSparkConf.HBASE_CONFIG_LOCATION)
+ .foreach(_.split(",").foreach(r => hbaseConf.addResource(new Path(r))))
+ val wrappedConf = new SerializableConfiguration(hbaseConf)
+
+ new HBasePartitionReaderFactory(
+ requiredSchema,
+ properties,
+ catalog,
+ pushedFilters,
+ encoderClsName,
+ usePushDownColumnFilter,
+ wrappedConf)
+ }
+}
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/HBaseInputPartition.scala
similarity index 60%
copy from spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/SerDes.scala
copy to spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBaseInputPartition.scala
index bf01901..b8791d2 100644
--- 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/HBaseInputPartition.scala
@@ -17,19 +17,23 @@
*/
package org.apache.hadoop.hbase.spark.datasources
-import org.apache.hadoop.hbase.util.Bytes
+import org.apache.spark.sql.connector.read.InputPartition
import org.apache.yetus.audience.InterfaceAudience
-@InterfaceAudience.Public
-trait SerDes {
- def serialize(value: Any): Array[Byte]
- def deserialize(bytes: Array[Byte], start: Int, end: Int): Any
-}
-
+/**
+ * This is a new class in the spark4 module. Implements InputPartition for serialization of the partition
+ * information to be sent to executors.
+ *
+ * Ranges are executed as HBase Scan operations; points as batched Get operations.
+ * This mirrors the spark3 HBaseScanPartition behavior.
+ */
@InterfaceAudience.Private
-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)
- }
+case class HBaseInputPartition(
+ index: Int,
+ scanRanges: Seq[Range],
+ points: Seq[Array[Byte]],
+ serverHostname: Option[String] = None)
+ extends InputPartition {
+ override def preferredLocations(): Array[String] =
+ serverHostname.toArray
}
diff --git a/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBasePartitionReader.scala b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBasePartitionReader.scala
new file mode 100644
index 0000000..5d9fbd4
--- /dev/null
+++ b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBasePartitionReader.scala
@@ -0,0 +1,435 @@
+/*
+ * 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.util.ArrayList
+import org.apache.hadoop.hbase.{CellUtil, TableName}
+import org.apache.hadoop.hbase.client.{Get, Query, Result, ResultScanner, Scan, Table}
+import org.apache.hadoop.hbase.spark.{AndLogicExpression, DynamicLogicExpression,
+ EqualLogicExpression, GreaterThanLogicExpression, GreaterThanOrEqualLogicExpression,
+ HBaseConnectionCache, IsNullLogicExpression, LessThanLogicExpression,
+ LessThanOrEqualLogicExpression, Logging, OrLogicExpression, PassThroughLogicExpression,
+ PushdownMappedField, SmartConnection, SparkSQLPushDownFilter, StartsWithLogicExpression}
+import org.apache.hadoop.hbase.util.Bytes
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.GenericInternalRow
+import org.apache.spark.sql.catalyst.util.DateTimeUtils
+import org.apache.spark.sql.types.Decimal
+import org.apache.spark.sql.connector.read.PartitionReader
+import org.apache.spark.sql.sources._
+import org.apache.spark.sql.types._
+import org.apache.spark.unsafe.types.UTF8String
+import org.apache.yetus.audience.InterfaceAudience
+import scala.collection.mutable.ListBuffer
+import scala.jdk.CollectionConverters._
+
+/**
+ * This is a new class in the spark4 module. Extends PartitionReader[InternalRow] for reading data from HBase regions.
+ * The actual execution: opens an HBase scanner on the partition's range, attaches the SparkSQLPushDownFilter,
+ * reads Result objects, and converts them to InternalRow. Implements next()/get()/close().
+ *
+ *
+ * In the spark 3 DS V1 model, this logic was inside DefaultSource.buildScan()
+ * which returned an RDD[Row] with its own compute() method.
+ *
+ * Ranges are executed as Scan operations, whilst points are executed as batched Get operations. This mirrors the spark3
+ * HBaseTableScanRDD.compute() behavior.
+ */
+@InterfaceAudience.Private
+class HBasePartitionReader(
+ partition: HBaseInputPartition,
+ requiredSchema: StructType,
+ properties: Map[String, String],
+ catalog: HBaseTableCatalog,
+ pushedFilters: Array[Filter],
+ encoderClsName: String,
+ usePushDownColumnFilter: Boolean,
+ wrappedConf: SerializableConfiguration)
+ extends PartitionReader[InternalRow]
+ with Logging {
+
+ private val conf = wrappedConf.value
+
+ private val connection: SmartConnection = HBaseConnectionCache.getConnection(conf)
+ private val tableName = s"${catalog.namespace}:${catalog.name}"
+ private val table: Table = connection.getTable(TableName.valueOf(tableName))
+
+ private val requiredFields = requiredSchema.fieldNames.map(catalog.sMap.getField(_))
+ private val filterFields = extractFilterFields(pushedFilters)
+ private val scanFields = (requiredFields ++ filterFields).distinct.filterNot(_.isRowKey)
+ private val hasNullCheck = pushedFilters.exists(containsNullCheck)
+ private val pushDownFilter: Option[SparkSQLPushDownFilter] = buildPushDownFilter()
+
+ private val bulkGetSize = properties
+ .get(HBaseSparkConf.BULKGET_SIZE)
+ .map(_.toInt)
+ .getOrElse(HBaseSparkConf.DEFAULT_BULKGET_SIZE)
+
+ private val blockCacheEnable = properties
+ .get(HBaseSparkConf.QUERY_CACHEBLOCKS)
+ .map(_.toBoolean)
+ .getOrElse(HBaseSparkConf.DEFAULT_QUERY_CACHEBLOCKS)
+
+ private val scanners = new ListBuffer[ResultScanner]()
+
+ private val resultIterator: Iterator[Result] = {
+ val scanIterators = partition.scanRanges.map { range =>
+ val scanner = buildScanner(range)
+ scanners += scanner
+ scannerToIterator(scanner)
+ }
+ val getIterator = if (partition.points.nonEmpty) {
+ buildGets(partition.points)
+ } else {
+ Iterator.empty
+ }
+ scanIterators.foldLeft(Iterator.empty: Iterator[Result])(_ ++ _) ++ getIterator
+ }
+
+ private var currentResult: Result = _
+
+ override def next(): Boolean = {
+ if (resultIterator.hasNext) {
+ currentResult = resultIterator.next()
+ true
+ } else {
+ false
+ }
+ }
+
+ private val keyFields = catalog.getRowKey
+
+ override def get(): InternalRow = {
+ val rowKey = currentResult.getRow
+
+ val keyValues = parseRowKey(rowKey, keyFields)
+ val values = new Array[Any](requiredFields.length)
+
+ requiredFields.zipWithIndex.foreach { case (field, idx) =>
+ if (field.isRowKey) {
+ values(idx) = convertToInternalRow(keyValues.get(field).orNull, field.dt)
+ } else {
+ val cell = currentResult.getColumnLatestCell(
+ Bytes.toBytes(field.cf), Bytes.toBytes(field.col))
+ if (cell == null || cell.getValueLength == 0) {
+ values(idx) = null
+ } else {
+ val v = CellUtil.cloneValue(cell)
+ val scalaValue = field.dt match {
+ case BinaryType => v
+ case _ => Utils.hbaseFieldToScalaType(field, v, 0, v.length)
+ }
+ values(idx) = convertToInternalRow(scalaValue, field.dt)
+ }
+ }
+ }
+ new GenericInternalRow(values)
+ }
+
+ override def close(): Unit = {
+ scanners.foreach(s => if (s != null) s.close())
+ if (table != null) table.close()
+ if (connection != null) connection.close()
+ }
+
+ private def setStopRow(scan: Scan, bound: Bound): Scan = {
+ if (bound.inc) {
+ val newArray = new Array[Byte](bound.b.length + 1)
+ System.arraycopy(bound.b, 0, newArray, 0, bound.b.length)
+ scan.withStopRow(newArray)
+ } else {
+ scan.withStopRow(bound.b)
+ }
+ }
+
+ private def buildScanner(range: Range): ResultScanner = {
+ val scan = (range.lower, range.upper) match {
+ case (Some(Bound(a, _)), Some(upper)) =>
+ setStopRow(new Scan().withStartRow(a), upper)
+ case (None, Some(upper)) =>
+ setStopRow(new Scan(), upper)
+ case (Some(Bound(a, _)), None) =>
+ new Scan().withStartRow(a)
+ case (None, None) =>
+ new Scan()
+ }
+
+ scan.setCacheBlocks(blockCacheEnable)
+ properties.get(HBaseSparkConf.QUERY_CACHEDROWS).map(_.toInt).foreach { rows =>
+ if (rows > 0) scan.setCaching(rows)
+ }
+ properties.get(HBaseSparkConf.QUERY_BATCHSIZE).map(_.toInt).foreach { batch =>
+ if (batch > 0) scan.setBatch(batch)
+ }
+ handleTimeSemantics(scan)
+
+ if (!hasNullCheck) {
+ scanFields.foreach { f =>
+ scan.addColumn(f.cfBytes, f.colBytes)
+ }
+ }
+ pushDownFilter.foreach(scan.setFilter(_))
+
+ table.getScanner(scan)
+ }
+
+ private def buildGets(points: Seq[Array[Byte]]): Iterator[Result] = {
+ points.grouped(bulkGetSize).flatMap { batch =>
+ val gets = new ArrayList[Get](batch.size)
+ batch.foreach { point =>
+ val g = new Get(point)
+ handleTimeSemantics(g)
+ if (!hasNullCheck) {
+ scanFields.foreach { f =>
+ g.addColumn(f.cfBytes, f.colBytes)
+ }
+ }
+ pushDownFilter.foreach(g.setFilter(_))
+ gets.add(g)
+ }
+ table.get(gets).toSeq.iterator.filter(r => r != null && !r.isEmpty)
+ }
+ }
+
+ private def scannerToIterator(scanner: ResultScanner): Iterator[Result] = {
+ new Iterator[Result] {
+ var cur: Option[Result] = None
+ override def hasNext: Boolean = {
+ if (cur.isEmpty) {
+ val r = scanner.next()
+ if (r != null) cur = Some(r)
+ }
+ cur.isDefined
+ }
+ override def next(): Result = {
+ hasNext
+ val ret = cur.get
+ cur = None
+ ret
+ }
+ }
+ }
+
+ private def handleTimeSemantics(query: Query): Unit = {
+ val timestamp = properties.get(HBaseSparkConf.TIMESTAMP).map(_.toLong)
+ val minTs = properties.get(HBaseSparkConf.TIMERANGE_START).map(_.toLong)
+ val maxTs = properties.get(HBaseSparkConf.TIMERANGE_END).map(_.toLong)
+ (query, timestamp, minTs, maxTs) match {
+ case (q: Scan, Some(ts), None, None) => q.setTimestamp(ts)
+ case (q: Get, Some(ts), None, None) => q.setTimestamp(ts)
+ case (q: Scan, None, Some(min), Some(max)) => q.setTimeRange(min, max)
+ case (q: Get, None, Some(min), Some(max)) => q.setTimeRange(min, max)
+ case (_, None, None, None) =>
+ case _ =>
+ throw new IllegalArgumentException(
+ "Invalid combination of timestamp/time range provided.")
+ }
+ val maxVersions = properties.get(HBaseSparkConf.MAX_VERSIONS).map(_.toInt)
+ maxVersions.foreach { mv =>
+ query match {
+ case q: Scan => q.readVersions(mv)
+ case q: Get => q.readVersions(mv)
+ case _ =>
+ }
+ }
+ }
+
+ private def buildPushDownFilter(): Option[SparkSQLPushDownFilter] = {
+ if (!usePushDownColumnFilter || pushedFilters.isEmpty) return None
+ val valueArray = buildValueArray()
+ val dynamicLogicExpression = buildDynamicLogicExpression()
+ if (dynamicLogicExpression == null) return None
+
+ val allFilterFields = (requiredFields ++ filterFields).distinct
+ val columnMappings = allFilterFields.map { field =>
+ new PushdownMappedField {
+ override def colName(): String = field.colName
+ override def cfBytes(): Array[Byte] = field.cfBytes
+ override def colBytes(): Array[Byte] = field.colBytes
+ }
+ }
+ Some(new SparkSQLPushDownFilter(
+ dynamicLogicExpression,
+ valueArray,
+ columnMappings.toList.asJava,
+ encoderClsName))
+ }
+
+ private def convertToInternalRow(value: Any, dataType: DataType): Any = {
+ if (value == null) return null
+ dataType match {
+ case StringType => UTF8String.fromString(value.asInstanceOf[String])
+ case DateType =>
+ val d = value.asInstanceOf[java.sql.Date]
+ DateTimeUtils.fromJavaDate(d)
+ case TimestampType =>
+ val t = value.asInstanceOf[java.sql.Timestamp]
+ DateTimeUtils.fromJavaTimestamp(t)
+ case dt: DecimalType =>
+ Decimal(value.asInstanceOf[java.math.BigDecimal], dt.precision, dt.scale)
+ case _ => value
+ }
+ }
+
+ private def parseRowKey(row: Array[Byte], keyFields: Seq[Field]): Map[Field, Any] = {
+ keyFields
+ .foldLeft((0, Seq[(Field, Any)]())) { (state, field) =>
+ val idx = state._1
+ val parsed = state._2
+ if (field.length != -1) {
+ val value = Utils.hbaseFieldToScalaType(field, row, idx, field.length)
+ (idx + field.length, parsed :+ (field, value))
+ } else {
+ field.dt match {
+ case StringType =>
+ val pos = row.indexOf(HBaseTableCatalog.delimiter, idx)
+ if (pos == -1 || pos > row.length) {
+ val value = Utils.hbaseFieldToScalaType(field, row, idx, row.length - idx)
+ (row.length + 1, parsed :+ (field, value))
+ } else {
+ val value = Utils.hbaseFieldToScalaType(field, row, idx, pos - idx)
+ (pos + 1, parsed :+ (field, value))
+ }
+ case _ =>
+ (
+ row.length + 1,
+ parsed :+ (field, Utils.hbaseFieldToScalaType(field, row, idx, row.length - idx)))
+ }
+ }
+ }
+ ._2
+ .toMap
+ }
+
+ private def extractFilterFields(filters: Array[Filter]): Array[Field] = {
+ val fields = new ListBuffer[Field]()
+ def extract(f: Filter): Unit = f match {
+ case EqualTo(attr, _) => catalog.sMap.map.get(attr).foreach(fields += _)
+ case LessThan(attr, _) => catalog.sMap.map.get(attr).foreach(fields += _)
+ case GreaterThan(attr, _) => catalog.sMap.map.get(attr).foreach(fields += _)
+ case LessThanOrEqual(attr, _) => catalog.sMap.map.get(attr).foreach(fields += _)
+ case GreaterThanOrEqual(attr, _) => catalog.sMap.map.get(attr).foreach(fields += _)
+ case StringStartsWith(attr, _) => catalog.sMap.map.get(attr).foreach(fields += _)
+ case IsNull(attr) => catalog.sMap.map.get(attr).foreach(fields += _)
+ case IsNotNull(attr) => catalog.sMap.map.get(attr).foreach(fields += _)
+ case Or(left, right) => extract(left); extract(right)
+ case And(left, right) => extract(left); extract(right)
+ case _ =>
+ }
+ filters.foreach(extract)
+ fields.toArray
+ }
+
+ private def buildValueArray(): Array[Array[Byte]] = {
+ val values = new ListBuffer[Array[Byte]]()
+ val encoder = JavaBytesEncoder.create(encoderClsName)
+ pushedFilters.foreach(f => collectFilterValues(values, f, encoder))
+ values.toArray
+ }
+
+ private def collectFilterValues(
+ values: ListBuffer[Array[Byte]],
+ filter: Filter,
+ encoder: BytesEncoder): Unit = {
+ filter match {
+ case EqualTo(attr, value) =>
+ val field = catalog.sMap.map.get(attr).orNull
+ if (field != null) values += Utils.toBytes(value, field)
+ case LessThan(attr, value) =>
+ val field = catalog.sMap.map.get(attr).orNull
+ if (field != null) values += encoder.encode(field.dt, value)
+ case GreaterThan(attr, value) =>
+ val field = catalog.sMap.map.get(attr).orNull
+ if (field != null) values += encoder.encode(field.dt, value)
+ case LessThanOrEqual(attr, value) =>
+ val field = catalog.sMap.map.get(attr).orNull
+ if (field != null) values += encoder.encode(field.dt, value)
+ case GreaterThanOrEqual(attr, value) =>
+ val field = catalog.sMap.map.get(attr).orNull
+ if (field != null) values += encoder.encode(field.dt, value)
+ case StringStartsWith(attr, value) =>
+ val field = catalog.sMap.map.get(attr).orNull
+ if (field != null) values += Utils.toBytes(value, field)
+ case Or(left, right) =>
+ collectFilterValues(values, left, encoder)
+ collectFilterValues(values, right, encoder)
+ case And(left, right) =>
+ collectFilterValues(values, left, encoder)
+ collectFilterValues(values, right, encoder)
+ case _ =>
+ }
+ }
+
+ private def buildDynamicLogicExpression(): DynamicLogicExpression = {
+ var idx = 0
+ def buildExpression(filter: Filter): DynamicLogicExpression = {
+ filter match {
+ case EqualTo(attr, _) =>
+ val field = catalog.sMap.map.get(attr).orNull
+ if (field != null) { val i = idx; idx += 1; new EqualLogicExpression(attr, i, false) }
+ else new PassThroughLogicExpression
+ case LessThan(attr, _) =>
+ val field = catalog.sMap.map.get(attr).orNull
+ if (field != null) { val i = idx; idx += 1; new LessThanLogicExpression(attr, i) }
+ else new PassThroughLogicExpression
+ case GreaterThan(attr, _) =>
+ val field = catalog.sMap.map.get(attr).orNull
+ if (field != null) { val i = idx; idx += 1; new GreaterThanLogicExpression(attr, i) }
+ else new PassThroughLogicExpression
+ case LessThanOrEqual(attr, _) =>
+ val field = catalog.sMap.map.get(attr).orNull
+ if (field != null) { val i = idx; idx += 1; new LessThanOrEqualLogicExpression(attr, i) }
+ else new PassThroughLogicExpression
+ case GreaterThanOrEqual(attr, _) =>
+ val field = catalog.sMap.map.get(attr).orNull
+ if (field != null) {
+ val i = idx; idx += 1; new GreaterThanOrEqualLogicExpression(attr, i)
+ }
+ else new PassThroughLogicExpression
+ case StringStartsWith(attr, _) =>
+ val field = catalog.sMap.map.get(attr).orNull
+ if (field != null) { val i = idx; idx += 1; new StartsWithLogicExpression(attr, i) }
+ else new PassThroughLogicExpression
+ case IsNull(attr) => new IsNullLogicExpression(attr, false)
+ case IsNotNull(attr) => new IsNullLogicExpression(attr, true)
+ case Or(left, right) =>
+ new OrLogicExpression(buildExpression(left), buildExpression(right))
+ case And(left, right) =>
+ new AndLogicExpression(buildExpression(left), buildExpression(right))
+ case _ => new PassThroughLogicExpression
+ }
+ }
+
+ if (pushedFilters.isEmpty) return null
+
+ var result: DynamicLogicExpression = null
+ pushedFilters.foreach { f =>
+ val expr = buildExpression(f)
+ result = if (result == null) expr else new AndLogicExpression(result, expr)
+ }
+ result
+ }
+
+ private def containsNullCheck(f: Filter): Boolean = f match {
+ case IsNull(_) => true
+ case IsNotNull(_) => true
+ case And(left, right) => containsNullCheck(left) || containsNullCheck(right)
+ case Or(left, right) => containsNullCheck(left) || containsNullCheck(right)
+ case _ => false
+ }
+}
diff --git a/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBasePartitionReaderFactory.scala b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBasePartitionReaderFactory.scala
new file mode 100644
index 0000000..1a85e7f
--- /dev/null
+++ b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBasePartitionReaderFactory.scala
@@ -0,0 +1,64 @@
+/*
+ * 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.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.connector.read.{InputPartition, PartitionReader, PartitionReaderFactory}
+import org.apache.spark.sql.sources.Filter
+import org.apache.spark.sql.types.StructType
+import org.apache.yetus.audience.InterfaceAudience
+
+/**
+ * This is a new class in the spark4 module. Implements PartitionReaderFactory for
+ * creating HBasePartitionReader instances on executors.
+ *
+ * In the spark 3 DS V1 model, there was no factory, the RDD.compute(partition) call directly created the iterator.
+ *
+ *
+ * @param requiredSchema
+ * @param properties
+ * @param catalog
+ * @param pushedFilters
+ * @param encoderClsName
+ * @param usePushDownColumnFilter
+ */
+@InterfaceAudience.Private
+class HBasePartitionReaderFactory(
+ requiredSchema: StructType,
+ properties: Map[String, String],
+ catalog: HBaseTableCatalog,
+ pushedFilters: Array[Filter],
+ encoderClsName: String,
+ usePushDownColumnFilter: Boolean,
+ wrappedConf: SerializableConfiguration)
+ extends PartitionReaderFactory
+ with Serializable {
+
+ override def createReader(partition: InputPartition): PartitionReader[InternalRow] = {
+ val hbasePartition = partition.asInstanceOf[HBaseInputPartition]
+ new HBasePartitionReader(
+ hbasePartition,
+ requiredSchema,
+ properties,
+ catalog,
+ pushedFilters,
+ encoderClsName,
+ usePushDownColumnFilter,
+ wrappedConf)
+ }
+}
diff --git a/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBaseScan.scala b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBaseScan.scala
new file mode 100644
index 0000000..f152d6b
--- /dev/null
+++ b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBaseScan.scala
@@ -0,0 +1,151 @@
+/*
+ * 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.spark.Logging
+import org.apache.spark.sql.connector.read.{Batch, Scan}
+import org.apache.spark.sql.sources._
+import org.apache.spark.sql.types.StructType
+import org.apache.yetus.audience.InterfaceAudience
+
+/**
+ * This is a new class in the spark4 module. Implements Scan.
+ *
+ * An immutable snapshot of the scan plan after negotiation is complete.
+ * Holds the pushed filters, required schema, and encoder.
+ * Builds the RowKeyFilter (for scan range narrowing from row key predicates) and produces the Batch.
+ *
+ * In the spark 3 V1 model, there was no separation between "plan" and "execution", the buildScan() method did both.
+ *
+ * @param requiredSchema
+ * @param properties
+ * @param catalog
+ * @param pushedFilters
+ * @param encoderClsName
+ * @param encoder
+ */
+@InterfaceAudience.Private
+class HBaseScan(
+ requiredSchema: StructType,
+ properties: Map[String, String],
+ catalog: HBaseTableCatalog,
+ pushedFilters: Array[Filter],
+ rowKeyFilters: Array[Filter],
+ encoderClsName: String,
+ @transient encoder: BytesEncoder)
+ extends Scan
+ with Logging {
+
+ override def readSchema(): StructType = requiredSchema
+
+ override def toBatch: Batch = {
+ val rowKeyFilter = buildRowKeyFilter()
+ new HBaseBatch(requiredSchema, properties, catalog, rowKeyFilter, pushedFilters, encoderClsName)
+ }
+
+ private def buildRowKeyFilter(): RowKeyFilter = {
+ var superRowKeyFilter: RowKeyFilter = null
+
+ rowKeyFilters.foreach { f =>
+ val rowKeyFilter = new RowKeyFilter()
+ traverseFilterTree(rowKeyFilter, f)
+ if (superRowKeyFilter == null) {
+ superRowKeyFilter = rowKeyFilter
+ } else {
+ superRowKeyFilter.mergeIntersect(rowKeyFilter)
+ }
+ }
+
+ if (superRowKeyFilter == null) {
+ superRowKeyFilter = new RowKeyFilter
+ }
+ superRowKeyFilter
+ }
+
+ private def traverseFilterTree(
+ parentRowKeyFilter: RowKeyFilter,
+ filter: Filter): Unit = {
+ filter match {
+ case EqualTo(attr, value) =>
+ val field = catalog.sMap.map.get(attr).orNull
+ if (field != null && field.isRowKey) {
+ parentRowKeyFilter.mergeIntersect(new RowKeyFilter(Utils.toBytes(value, field), null))
+ }
+ case LessThan(attr, value) =>
+ val field = catalog.sMap.map.get(attr).orNull
+ if (field != null && field.isRowKey) {
+ val b = encoder.ranges(value)
+ var inc = false
+ b.map(_.less.map { x =>
+ val r = new RowKeyFilter(null, new ScanRange(x.upper, inc, x.low, true))
+ inc = true
+ r
+ }).map(x => x.reduce((i, j) => i.mergeUnion(j)))
+ .map(parentRowKeyFilter.mergeIntersect(_))
+ }
+ case GreaterThan(attr, value) =>
+ val field = catalog.sMap.map.get(attr).orNull
+ if (field != null && field.isRowKey) {
+ val b = encoder.ranges(value)
+ var inc = false
+ b.map(_.greater.map { x =>
+ val r = new RowKeyFilter(null, new ScanRange(x.upper, true, x.low, inc))
+ inc = true
+ r
+ }).map(x => x.reduce((i, j) => i.mergeUnion(j)))
+ .map(parentRowKeyFilter.mergeIntersect(_))
+ }
+ case LessThanOrEqual(attr, value) =>
+ val field = catalog.sMap.map.get(attr).orNull
+ if (field != null && field.isRowKey) {
+ val b = encoder.ranges(value)
+ b.map(_.less.map(x => new RowKeyFilter(null, new ScanRange(x.upper, true, x.low, true))))
+ .map(x => x.reduce((i, j) => i.mergeUnion(j)))
+ .map(parentRowKeyFilter.mergeIntersect(_))
+ }
+ case GreaterThanOrEqual(attr, value) =>
+ val field = catalog.sMap.map.get(attr).orNull
+ if (field != null && field.isRowKey) {
+ val b = encoder.ranges(value)
+ b.map(
+ _.greater.map(x => new RowKeyFilter(null, new ScanRange(x.upper, true, x.low, true))))
+ .map(x => x.reduce((i, j) => i.mergeUnion(j)))
+ .map(parentRowKeyFilter.mergeIntersect(_))
+ }
+ case StringStartsWith(attr, value) =>
+ val field = catalog.sMap.map.get(attr).orNull
+ if (field != null && field.isRowKey) {
+ val p = Utils.toBytes(value, field)
+ val endRange = Utils.incrementByteArray(p)
+ parentRowKeyFilter.mergeIntersect(
+ new RowKeyFilter(null, new ScanRange(endRange, false, p, true)))
+ }
+ case Or(left, right) =>
+ traverseFilterTree(parentRowKeyFilter, left)
+ val rightSideRowKeyFilter = new RowKeyFilter
+ traverseFilterTree(rightSideRowKeyFilter, right)
+ parentRowKeyFilter.mergeUnion(rightSideRowKeyFilter)
+ case And(left, right) =>
+ traverseFilterTree(parentRowKeyFilter, left)
+ val rightSideRowKeyFilter = new RowKeyFilter
+ traverseFilterTree(rightSideRowKeyFilter, right)
+ parentRowKeyFilter.mergeIntersect(rightSideRowKeyFilter)
+ case _ =>
+ }
+ }
+}
diff --git a/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBaseScanBuilder.scala b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBaseScanBuilder.scala
new file mode 100644
index 0000000..6de7da5
--- /dev/null
+++ b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBaseScanBuilder.scala
@@ -0,0 +1,115 @@
+/*
+ * 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.spark.Logging
+import org.apache.spark.sql.connector.read.{Scan, ScanBuilder, SupportsPushDownFilters, SupportsPushDownRequiredColumns}
+import org.apache.spark.sql.sources._
+import org.apache.spark.sql.types.StructType
+import org.apache.yetus.audience.InterfaceAudience
+import scala.collection.mutable.ListBuffer
+
+/**
+ * This is a new class in the spark4 module.
+ * Implements ScanBuilder, SupportsPushDownFilters, and SupportsPushDownRequiredColumns.
+ * This is where Catalyst negotiates with the connector. Spark calls pushFilters() with
+ * candidate predicates and the builder accepts what it can handle and returns the rest.
+ * Spark calls pruneColumns() to say which columns it actually needs.
+ * Then build() produces the final scan plan.
+ *
+ * In the spark 3 V1 model, this negotiation happened implicitly via
+ * PrunedFilteredScan.buildScan(requiredColumns, filters) as a single method call with no back-and-forth.
+ *
+ * @param schema
+ * @param properties
+ */
+@InterfaceAudience.Private
+class HBaseScanBuilder(schema: StructType, properties: Map[String, String])
+ extends ScanBuilder
+ with SupportsPushDownFilters
+ with SupportsPushDownRequiredColumns
+ with Logging {
+
+ private val catalog = HBaseTableCatalog(properties)
+ private val encoderClsName =
+ properties.getOrElse(HBaseSparkConf.QUERY_ENCODER, HBaseSparkConf.DEFAULT_QUERY_ENCODER)
+ @transient private val encoder = JavaBytesEncoder.create(encoderClsName)
+
+ private var _pushedFilters: Array[Filter] = Array.empty
+ private var _rowKeyFilters: Array[Filter] = Array.empty
+ private var requiredSchema: StructType = schema
+
+ override def pushFilters(filters: Array[Filter]): Array[Filter] = {
+ val hasCompositeRowKey = catalog.getRowKey.size > 1
+
+ def isSupported(f: Filter): Boolean = f match {
+ case EqualTo(attr, _) => isSupportedField(attr)
+ case LessThan(attr, _) => isSupportedField(attr)
+ case GreaterThan(attr, _) => isSupportedField(attr)
+ case LessThanOrEqual(attr, _) => isSupportedField(attr)
+ case GreaterThanOrEqual(attr, _) => isSupportedField(attr)
+ case StringStartsWith(attr, _) => isSupportedField(attr)
+ case IsNull(attr) => isSupportedField(attr)
+ case IsNotNull(attr) => isSupportedField(attr)
+ case Or(left, right) => isSupported(left) && isSupported(right)
+ case And(left, right) => isSupported(left) && isSupported(right)
+ case _ => false
+ }
+
+ def isSupportedField(attr: String): Boolean = {
+ catalog.sMap.map.get(attr) match {
+ case Some(field) => !(hasCompositeRowKey && field.isRowKey)
+ case None => false
+ }
+ }
+
+ val supported = new ListBuffer[Filter]()
+ val unsupported = new ListBuffer[Filter]()
+
+ filters.foreach { f =>
+ if (isSupported(f)) supported += f
+ else unsupported += f
+ }
+
+ _rowKeyFilters = supported.toArray
+
+ val usePushDown = properties
+ .get(HBaseSparkConf.PUSHDOWN_COLUMN_FILTER)
+ .map(_.toBoolean)
+ .getOrElse(HBaseSparkConf.DEFAULT_PUSHDOWN_COLUMN_FILTER)
+
+ if (!usePushDown) {
+ _pushedFilters = Array.empty
+ return filters
+ }
+
+ _pushedFilters = supported.toArray
+ unsupported.toArray
+ }
+
+ override def pushedFilters(): Array[Filter] = _pushedFilters
+
+ override def pruneColumns(requiredSchema: StructType): Unit = {
+ this.requiredSchema = requiredSchema
+ }
+
+ override def build(): Scan = {
+ new HBaseScan(requiredSchema, properties, catalog, _pushedFilters, _rowKeyFilters,
+ encoderClsName, encoder)
+ }
+}
diff --git a/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBaseTable.scala b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBaseTable.scala
new file mode 100644
index 0000000..c2fe3ff
--- /dev/null
+++ b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBaseTable.scala
@@ -0,0 +1,56 @@
+/*
+ * 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.util
+import org.apache.spark.sql.connector.catalog.{SupportsRead, Table, TableCapability}
+import org.apache.spark.sql.connector.read.ScanBuilder
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+import org.apache.yetus.audience.InterfaceAudience
+import scala.jdk.CollectionConverters._
+
+/**
+ * This is a new class in the spark4 module. Implements Table and SupportsRead (TODO: modify once write ops are added).
+ * Represents the HBase table as a Spark entity. Declares its schema (from the catalog) and capabilities (BATCH_READ).
+ * When Spark wants to read, it calls newScanBuilder(). In the spark 3 DS V1 model,
+ * there was no separate "table" concept, BaseRelation bundled schema and read logic together.
+ * @param tableSchema
+ * @param properties
+ */
+@InterfaceAudience.Private
+class HBaseTable(tableSchema: StructType, properties: Map[String, String])
+ extends Table
+ with SupportsRead {
+
+ override def name(): String = {
+ val catalog = HBaseTableCatalog(properties)
+ s"${catalog.namespace}:${catalog.name}"
+ }
+
+ override def schema(): StructType = tableSchema
+
+ override def capabilities(): util.Set[TableCapability] = {
+ Set(TableCapability.BATCH_READ).asJava
+ }
+
+ override def newScanBuilder(options: CaseInsensitiveStringMap): ScanBuilder = {
+ val mergedProps = properties ++ options.asScala.toMap
+ new HBaseScanBuilder(tableSchema, mergedProps)
+ }
+}
diff --git a/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBaseTableProvider.scala b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBaseTableProvider.scala
new file mode 100644
index 0000000..0710158
--- /dev/null
+++ b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBaseTableProvider.scala
@@ -0,0 +1,74 @@
+/*
+ * 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.util
+import org.apache.spark.sql.connector.catalog.{Table, TableProvider}
+import org.apache.spark.sql.connector.expressions.Transform
+import org.apache.spark.sql.sources.DataSourceRegister
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
+import org.apache.yetus.audience.InterfaceAudience
+import scala.jdk.CollectionConverters._
+
+/**
+ * This is a new class in the spark4 module, and is the entry point for sparkSQL in spark 4 DataSource V2.
+ * It's the equivalent to what DefaultSource (a RelationProvider) was in spark 3 DataSource V1.
+ *
+ * Implements DS V2 TableProvider and DataSourceRegister, so when you call:
+ * <code>
+ * ...
+ * spark.read.format("org.apache.hadoop.hbase.spark.datasources.HBaseTableProvider")
+ * ...
+ * </code>
+ *
+ * Spark instantiates this class and calls its getTable() method. In V1, DefaultSource.createRelation() returned a
+ * BaseRelation directly, but now in V2, it returns an HBaseTable object that describes the table's capabilities.
+ *
+ * Alternatively, it overrides shortName() to provide a short name "hbase" for this data source, so you can also call:
+ * <code>
+ * ...
+ * spark.read.format("hbase")
+ * ...
+ * </code>
+ *
+ *
+ */
+@InterfaceAudience.Public
+class HBaseTableProvider extends TableProvider with DataSourceRegister {
+
+ /**
+ * Short name of the data source, used to allow users to specify format("hbase")
+ * as a short name for format("org.apache.hadoop.hbase.spark.datasources.HBaseTableProvider").
+ * @return
+ */
+ override def shortName(): String = "hbase"
+
+ override def inferSchema(options: CaseInsensitiveStringMap): StructType = {
+ val params = options.asScala.toMap
+ HBaseTableCatalog(params).toDataType
+ }
+
+ override def getTable(
+ schema: StructType,
+ partitioning: Array[Transform],
+ properties: util.Map[String, String]): Table = {
+ val params = properties.asScala.toMap
+ new HBaseTable(schema, params)
+ }
+}
diff --git a/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/ScanRange.scala b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/ScanRange.scala
new file mode 100644
index 0000000..f1e5a7a
--- /dev/null
+++ b/spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/ScanRange.scala
@@ -0,0 +1,222 @@
+/*
+ * 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
+import scala.collection.mutable.ListBuffer
+
+/**
+ * This is a new class in the spark4 module. Wraps ScanRange and RowKeyFilter.
+ *
+ * Extracted from DefaultSource.scala in spark3 where it was an inner class. Handles merging scan ranges
+ * from row key predicates (union/intersect). Same logic, just in its own file now for clarity.
+ *
+ * @param upperBound
+ * @param isUpperBoundEqualTo
+ * @param lowerBound
+ * @param isLowerBoundEqualTo
+ */
+
+@InterfaceAudience.Private
+class ScanRange(
+ var upperBound: Array[Byte],
+ var isUpperBoundEqualTo: Boolean,
+ var lowerBound: Array[Byte],
+ var isLowerBoundEqualTo: Boolean)
+ extends Serializable {
+
+ def mergeIntersect(other: ScanRange): Unit = {
+ val upperBoundCompare = compareRange(upperBound, other.upperBound)
+ val lowerBoundCompare = compareRange(lowerBound, other.lowerBound)
+
+ upperBound = if (upperBoundCompare < 0) upperBound else other.upperBound
+ lowerBound = if (lowerBoundCompare > 0) lowerBound else other.lowerBound
+
+ isLowerBoundEqualTo =
+ if (lowerBoundCompare == 0)
+ isLowerBoundEqualTo && other.isLowerBoundEqualTo
+ else if (lowerBoundCompare < 0) other.isLowerBoundEqualTo
+ else isLowerBoundEqualTo
+
+ isUpperBoundEqualTo =
+ if (upperBoundCompare == 0)
+ isUpperBoundEqualTo && other.isUpperBoundEqualTo
+ else if (upperBoundCompare < 0) isUpperBoundEqualTo
+ else other.isUpperBoundEqualTo
+ }
+
+ def mergeUnion(other: ScanRange): Unit = {
+ val upperBoundCompare = compareRange(upperBound, other.upperBound)
+ val lowerBoundCompare = compareRange(lowerBound, other.lowerBound)
+
+ upperBound = if (upperBoundCompare > 0) upperBound else other.upperBound
+ lowerBound = if (lowerBoundCompare < 0) lowerBound else other.lowerBound
+
+ isLowerBoundEqualTo =
+ if (lowerBoundCompare == 0)
+ isLowerBoundEqualTo || other.isLowerBoundEqualTo
+ else if (lowerBoundCompare < 0) isLowerBoundEqualTo
+ else other.isLowerBoundEqualTo
+
+ isUpperBoundEqualTo =
+ if (upperBoundCompare == 0)
+ isUpperBoundEqualTo || other.isUpperBoundEqualTo
+ else if (upperBoundCompare < 0) other.isUpperBoundEqualTo
+ else isUpperBoundEqualTo
+ }
+
+ def getOverLapScanRange(other: ScanRange): ScanRange = {
+ var leftRange: ScanRange = null
+ var rightRange: ScanRange = null
+
+ if (compareRange(lowerBound, other.lowerBound) < 0 ||
+ compareRange(upperBound, other.upperBound) < 0) {
+ leftRange = this
+ rightRange = other
+ } else {
+ leftRange = other
+ rightRange = this
+ }
+
+ if (hasOverlap(leftRange, rightRange)) {
+ val result = new ScanRange(upperBound, isUpperBoundEqualTo, lowerBound, isLowerBoundEqualTo)
+ result.mergeIntersect(other)
+ result
+ } else {
+ null
+ }
+ }
+
+ def hasOverlap(left: ScanRange, right: ScanRange): Boolean = {
+ val cmp = compareRange(left.upperBound, right.lowerBound)
+ if (cmp > 0) true
+ else if (cmp == 0) left.isUpperBoundEqualTo && right.isLowerBoundEqualTo
+ else false
+ }
+
+ def compareRange(left: Array[Byte], right: Array[Byte]): Int = {
+ if (left == null && right == null) 0
+ else if (left == null && right != null) 1
+ else if (left != null && right == null) -1
+ else Bytes.compareTo(left, right)
+ }
+
+ def containsPoint(point: Array[Byte]): Boolean = {
+ val lowerCompare = compareRange(point, lowerBound)
+ val upperCompare = compareRange(point, upperBound)
+
+ ((isLowerBoundEqualTo && lowerCompare >= 0) ||
+ (!isLowerBoundEqualTo && lowerCompare > 0)) &&
+ ((isUpperBoundEqualTo && upperCompare <= 0) ||
+ (!isUpperBoundEqualTo && upperCompare < 0))
+ }
+
+ override def toString: String = {
+ "ScanRange:(upperBound:" + Bytes.toString(upperBound) +
+ ",isUpperBoundEqualTo:" + isUpperBoundEqualTo + ",lowerBound:" +
+ Bytes.toString(lowerBound) + ",isLowerBoundEqualTo:" + isLowerBoundEqualTo + ")"
+ }
+}
+
+@InterfaceAudience.Private
+class RowKeyFilter(
+ currentPoint: Array[Byte] = null,
+ currentRange: ScanRange = new ScanRange(null, true, new Array[Byte](0), true),
+ var points: ListBuffer[Array[Byte]] = new ListBuffer[Array[Byte]](),
+ var ranges: ListBuffer[ScanRange] = new ListBuffer[ScanRange]())
+ extends Serializable {
+
+ if (currentRange != null) ranges += currentRange
+ if (currentPoint != null) points += currentPoint
+
+ def mergeUnion(other: RowKeyFilter): RowKeyFilter = {
+ other.points.foreach(p => points += p)
+
+ other.ranges.foreach { otherR =>
+ var doesOverLap = false
+ ranges.foreach { r =>
+ if (r.getOverLapScanRange(otherR) != null) {
+ r.mergeUnion(otherR)
+ doesOverLap = true
+ }
+ }
+ if (!doesOverLap) ranges += otherR
+ }
+ this
+ }
+
+ def mergeIntersect(other: RowKeyFilter): RowKeyFilter = {
+ val survivingPoints = new ListBuffer[Array[Byte]]()
+ val didntSurviveFirstPassPoints = new ListBuffer[Array[Byte]]()
+ if (points == null || points.isEmpty) {
+ other.points.foreach(otherP => didntSurviveFirstPassPoints += otherP)
+ } else {
+ points.foreach { p =>
+ if (other.points.exists(Bytes.equals(p, _))) {
+ survivingPoints += p
+ } else {
+ didntSurviveFirstPassPoints += p
+ }
+ }
+ }
+
+ val survivingRanges = new ListBuffer[ScanRange]()
+
+ if (ranges.isEmpty) {
+ didntSurviveFirstPassPoints.foreach(p => survivingPoints += p)
+ } else {
+ ranges.foreach { r =>
+ other.ranges.foreach { otherR =>
+ val overLapScanRange = r.getOverLapScanRange(otherR)
+ if (overLapScanRange != null) {
+ survivingRanges += overLapScanRange
+ }
+ }
+ didntSurviveFirstPassPoints.foreach { p =>
+ if (r.containsPoint(p)) {
+ survivingPoints += p
+ }
+ }
+ }
+ }
+ points = survivingPoints
+ ranges = survivingRanges
+ this
+ }
+
+ override def toString: String = {
+ val strBuilder = new StringBuilder
+ strBuilder.append("(points:(")
+ var isFirst = true
+ points.foreach { p =>
+ if (isFirst) isFirst = false
+ else strBuilder.append(",")
+ strBuilder.append(Bytes.toString(p))
+ }
+ strBuilder.append("),ranges:")
+ isFirst = true
+ ranges.foreach { r =>
+ if (isFirst) isFirst = false
+ else strBuilder.append(",")
+ strBuilder.append(r)
+ }
+ strBuilder.append("))")
+ strBuilder.toString()
+ }
+}
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
index bf01901..a2cdfc9 100644
--- 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
@@ -21,7 +21,7 @@ import org.apache.hadoop.hbase.util.Bytes
import org.apache.yetus.audience.InterfaceAudience
@InterfaceAudience.Public
-trait SerDes {
+trait SerDes extends Serializable {
def serialize(value: Any): Array[Byte]
def deserialize(bytes: Array[Byte], start: Int, end: Int): Any
}
diff --git a/spark4/hbase-spark4/src/test/scala/org/apache/hadoop/hbase/spark/datasources/HBaseScanBuilderSuite.scala b/spark4/hbase-spark4/src/test/scala/org/apache/hadoop/hbase/spark/datasources/HBaseScanBuilderSuite.scala
new file mode 100644
index 0000000..5dcc13d
--- /dev/null
+++ b/spark4/hbase-spark4/src/test/scala/org/apache/hadoop/hbase/spark/datasources/HBaseScanBuilderSuite.scala
@@ -0,0 +1,161 @@
+/*
+ * 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.spark.sql.sources._
+import org.apache.spark.sql.types._
+import org.scalatest.funsuite.AnyFunSuite
+
+class HBaseScanBuilderSuite extends AnyFunSuite {
+
+ val catalog = s"""{
+ |"table":{"namespace":"default", "name":"testtable"},
+ |"rowkey":"key",
+ |"columns":{
+ |"key":{"cf":"rowkey", "col":"key", "type":"string"},
+ |"name":{"cf":"cf", "col":"name", "type":"string"},
+ |"age":{"cf":"cf", "col":"age", "type":"int"},
+ |"salary":{"cf":"cf", "col":"salary", "type":"double"}
+ |}
+ |}""".stripMargin
+
+ val schema = StructType(Seq(
+ StructField("key", StringType),
+ StructField("name", StringType),
+ StructField("age", IntegerType),
+ StructField("salary", DoubleType)))
+
+ val properties = Map(HBaseTableCatalog.tableCatalog -> catalog)
+
+ private def newBuilder(): HBaseScanBuilder = new HBaseScanBuilder(schema, properties)
+
+ test("EqualTo on known column is pushed") {
+ val builder = newBuilder()
+ val unsupported = builder.pushFilters(Array(EqualTo("name", "Alice")))
+ assert(unsupported.isEmpty)
+ assert(builder.pushedFilters().length == 1)
+ assert(builder.pushedFilters()(0) == EqualTo("name", "Alice"))
+ }
+
+ test("LessThan on known column is pushed") {
+ val builder = newBuilder()
+ val unsupported = builder.pushFilters(Array(LessThan("age", 30)))
+ assert(unsupported.isEmpty)
+ assert(builder.pushedFilters().length == 1)
+ }
+
+ test("GreaterThan on known column is pushed") {
+ val builder = newBuilder()
+ val unsupported = builder.pushFilters(Array(GreaterThan("salary", 50000.0)))
+ assert(unsupported.isEmpty)
+ assert(builder.pushedFilters().length == 1)
+ }
+
+ test("LessThanOrEqual on known column is pushed") {
+ val builder = newBuilder()
+ val unsupported = builder.pushFilters(Array(LessThanOrEqual("age", 25)))
+ assert(unsupported.isEmpty)
+ assert(builder.pushedFilters().length == 1)
+ }
+
+ test("GreaterThanOrEqual on known column is pushed") {
+ val builder = newBuilder()
+ val unsupported = builder.pushFilters(Array(GreaterThanOrEqual("age", 18)))
+ assert(unsupported.isEmpty)
+ assert(builder.pushedFilters().length == 1)
+ }
+
+ test("StringStartsWith on known column is pushed") {
+ val builder = newBuilder()
+ val unsupported = builder.pushFilters(Array(StringStartsWith("name", "A")))
+ assert(unsupported.isEmpty)
+ assert(builder.pushedFilters().length == 1)
+ }
+
+ test("IsNull on known column is pushed") {
+ val builder = newBuilder()
+ val unsupported = builder.pushFilters(Array(IsNull("name")))
+ assert(unsupported.isEmpty)
+ assert(builder.pushedFilters().length == 1)
+ }
+
+ test("IsNotNull on known column is pushed") {
+ val builder = newBuilder()
+ val unsupported = builder.pushFilters(Array(IsNotNull("name")))
+ assert(unsupported.isEmpty)
+ assert(builder.pushedFilters().length == 1)
+ }
+
+ test("filter on unknown column is not pushed") {
+ val builder = newBuilder()
+ val unsupported = builder.pushFilters(Array(EqualTo("unknown_col", "value")))
+ assert(unsupported.length == 1)
+ assert(builder.pushedFilters().isEmpty)
+ }
+
+ test("Or compound filter is pushed") {
+ val builder = newBuilder()
+ val filter = Or(EqualTo("name", "Alice"), EqualTo("name", "Bob"))
+ val unsupported = builder.pushFilters(Array(filter))
+ assert(unsupported.isEmpty)
+ assert(builder.pushedFilters().length == 1)
+ }
+
+ test("And compound filter is pushed") {
+ val builder = newBuilder()
+ val filter = And(GreaterThan("age", 20), LessThan("age", 40))
+ val unsupported = builder.pushFilters(Array(filter))
+ assert(unsupported.isEmpty)
+ assert(builder.pushedFilters().length == 1)
+ }
+
+ test("mixed supported and unsupported filters are separated") {
+ val builder = newBuilder()
+ val filters = Array[Filter](
+ EqualTo("name", "Alice"),
+ EqualTo("unknown", "value"),
+ GreaterThan("age", 20))
+ val unsupported = builder.pushFilters(filters)
+ assert(unsupported.length == 1)
+ assert(unsupported(0) == EqualTo("unknown", "value"))
+ assert(builder.pushedFilters().length == 2)
+ }
+
+ test("filter on row key column is pushed") {
+ val builder = newBuilder()
+ val unsupported = builder.pushFilters(Array(EqualTo("key", "row001")))
+ assert(unsupported.isEmpty)
+ assert(builder.pushedFilters().length == 1)
+ }
+
+ test("pruneColumns reduces required schema") {
+ val builder = newBuilder()
+ val pruned = StructType(Seq(StructField("key", StringType), StructField("name", StringType)))
+ builder.pruneColumns(pruned)
+ val scan = builder.build()
+ assert(scan.readSchema() == pruned)
+ }
+
+ test("multiple pushFilters calls replace previous filters") {
+ val builder = newBuilder()
+ builder.pushFilters(Array(EqualTo("name", "Alice")))
+ assert(builder.pushedFilters().length == 1)
+ builder.pushFilters(Array(GreaterThan("age", 20), LessThan("age", 40)))
+ assert(builder.pushedFilters().length == 2)
+ }
+}
diff --git a/spark4/hbase-spark4/src/test/scala/org/apache/hadoop/hbase/spark/datasources/HBaseTableCatalogSuite.scala b/spark4/hbase-spark4/src/test/scala/org/apache/hadoop/hbase/spark/datasources/HBaseTableCatalogSuite.scala
new file mode 100644
index 0000000..2b77ae5
--- /dev/null
+++ b/spark4/hbase-spark4/src/test/scala/org/apache/hadoop/hbase/spark/datasources/HBaseTableCatalogSuite.scala
@@ -0,0 +1,189 @@
+/*
+ * 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.spark.sql.types._
+import org.scalatest.funsuite.AnyFunSuite
+
+class HBaseTableCatalogSuite extends AnyFunSuite {
+
+ val map = s"""MAP<int, struct<varchar:string>>"""
+ val array = s"""array<struct<tinYint:tinyint>>"""
+ val arrayMap = s"""MAp<int, ARRAY<double>>"""
+
+ val catalog = s"""{
+ |"table":{"namespace":"default", "name":"htable"},
+ |"rowkey":"key1:key2",
+ |"columns":{
+ |"col1":{"cf":"rowkey", "col":"key1", "type":"string"},
+ |"col2":{"cf":"rowkey", "col":"key2", "type":"double"},
+ |"col3":{"cf":"cf1", "col":"col2", "type":"binary"},
+ |"col4":{"cf":"cf1", "col":"col3", "type":"timestamp"},
+ |"col5":{"cf":"cf1", "col":"col4", "type":"double", "serdes":"${classOf[
+ DoubleSerDes].getName}"},
+ |"col6":{"cf":"cf1", "col":"col5", "type":"$map"},
+ |"col7":{"cf":"cf1", "col":"col6", "type":"$array"},
+ |"col8":{"cf":"cf1", "col":"col7", "type":"$arrayMap"},
+ |"col9":{"cf":"cf1", "col":"col8", "type":"date"},
+ |"col10":{"cf":"cf1", "col":"col9", "type":"timestamp"}
+ |}
+ |}""".stripMargin
+
+ val parameters = Map(HBaseTableCatalog.tableCatalog -> catalog)
+ val t = HBaseTableCatalog(parameters)
+
+ test("basic field properties") {
+ assert(t.getField("col1").isRowKey)
+ assert(t.getPrimaryKey == "key1")
+ assert(t.getField("col3").dt == BinaryType)
+ assert(t.getField("col4").dt == TimestampType)
+ assert(t.getField("col5").dt == DoubleType)
+ assert(t.getField("col5").serdes.isDefined)
+ assert(t.getField("col4").serdes.isEmpty)
+ assert(t.getField("col1").isRowKey)
+ assert(t.getField("col2").isRowKey)
+ assert(!t.getField("col3").isRowKey)
+ assert(t.getField("col2").length == Bytes.SIZEOF_DOUBLE)
+ assert(t.getField("col1").length == -1)
+ assert(t.getField("col8").length == -1)
+ assert(t.getField("col9").dt == DateType)
+ assert(t.getField("col10").dt == TimestampType)
+ }
+
+ test("composite row key fields") {
+ val rowKey = t.getRowKey
+ assert(rowKey.length == 2)
+ assert(rowKey.head.col == "key1")
+ assert(rowKey(1).col == "key2")
+ }
+
+ test("namespace and table name") {
+ assert(t.namespace == "default")
+ assert(t.name == "htable")
+ }
+
+ test("column families") {
+ val cfs = t.getColumnFamilies
+ assert(cfs.contains("cf1"))
+ assert(!cfs.contains("rowkey"))
+ }
+
+ test("parse MAP type") {
+ assert(DataTypeParserWrapper.parse(map) === t.getField("col6").dt)
+ }
+
+ test("parse array type") {
+ assert(DataTypeParserWrapper.parse(array) === t.getField("col7").dt)
+ }
+
+ test("parse array of map type") {
+ assert(DataTypeParserWrapper.parse(arrayMap) === t.getField("col8").dt)
+ }
+
+ test("convert legacy hbase.columns.mapping format") {
+ val m = Map(
+ "hbase.columns.mapping" ->
+ "KEY_FIELD STRING :key, A_FIELD STRING c:a, B_FIELD DOUBLE c:b, C_FIELD BINARY c:c,",
+ "hbase.table" -> "NAMESPACE:TABLE")
+ val converted = HBaseTableCatalog(m)
+ assert(converted.namespace === "NAMESPACE")
+ assert(converted.name == "TABLE")
+ assert(converted.getField("KEY_FIELD").isRowKey)
+ assert(DataTypeParserWrapper.parse("STRING") === converted.getField("A_FIELD").dt)
+ assert(!converted.getField("A_FIELD").isRowKey)
+ assert(DataTypeParserWrapper.parse("DOUBLE") === converted.getField("B_FIELD").dt)
+ assert(DataTypeParserWrapper.parse("BINARY") === converted.getField("C_FIELD").dt)
+ }
+
+ test("convert legacy format with default namespace") {
+ val m = Map(
+ "hbase.columns.mapping" ->
+ "KEY_FIELD STRING :key, A_FIELD STRING c:a, B_FIELD DOUBLE c:b, C_FIELD BINARY c:c,",
+ "hbase.table" -> "t1")
+ val converted = HBaseTableCatalog(m)
+ assert(converted.namespace === "default")
+ assert(converted.name == "t1")
+ assert(converted.getField("KEY_FIELD").isRowKey)
+ }
+
+ test("single row key field") {
+ val simpleCatalog = s"""{
+ |"table":{"namespace":"default", "name":"simple"},
+ |"rowkey":"id",
+ |"columns":{
+ |"id":{"cf":"rowkey", "col":"id", "type":"string"},
+ |"name":{"cf":"cf", "col":"name", "type":"string"}
+ |}
+ |}""".stripMargin
+ val params = Map(HBaseTableCatalog.tableCatalog -> simpleCatalog)
+ val cat = HBaseTableCatalog(params)
+ assert(cat.getRowKey.length == 1)
+ assert(cat.getRowKey.head.col == "id")
+ assert(cat.getField("id").isRowKey)
+ assert(!cat.getField("name").isRowKey)
+ assert(cat.getField("name").cf == "cf")
+ assert(cat.getField("name").col == "name")
+ }
+
+ test("fixed-length row key field has correct length") {
+ val intKeyCatalog = s"""{
+ |"table":{"namespace":"default", "name":"intkey"},
+ |"rowkey":"id",
+ |"columns":{
+ |"id":{"cf":"rowkey", "col":"id", "type":"int"},
+ |"value":{"cf":"cf", "col":"v", "type":"string"}
+ |}
+ |}""".stripMargin
+ val params = Map(HBaseTableCatalog.tableCatalog -> intKeyCatalog)
+ val cat = HBaseTableCatalog(params)
+ assert(cat.getRowKey.head.length == Bytes.SIZEOF_INT)
+ assert(cat.row.varLength == false)
+ }
+
+ test("variable-length row key sets varLength") {
+ assert(t.row.varLength == true)
+ }
+
+ test("cfBytes and colBytes for row key field") {
+ val field = t.getField("col1")
+ assert(field.cfBytes.sameElements(Bytes.toBytes("")))
+ assert(field.colBytes.sameElements(Bytes.toBytes("key")))
+ }
+
+ test("cfBytes and colBytes for regular field") {
+ val field = t.getField("col3")
+ assert(field.cfBytes.sameElements(Bytes.toBytes("cf1")))
+ assert(field.colBytes.sameElements(Bytes.toBytes("col2")))
+ }
+
+ test("custom namespace") {
+ val nsCatalog = s"""{
+ |"table":{"namespace":"myns", "name":"mytable"},
+ |"rowkey":"k",
+ |"columns":{
+ |"k":{"cf":"rowkey", "col":"k", "type":"string"},
+ |"v":{"cf":"data", "col":"val", "type":"string"}
+ |}
+ |}""".stripMargin
+ val params = Map(HBaseTableCatalog.tableCatalog -> nsCatalog)
+ val cat = HBaseTableCatalog(params)
+ assert(cat.namespace == "myns")
+ assert(cat.name == "mytable")
+ }
+}
diff --git a/spark4/hbase-spark4/src/test/scala/org/apache/hadoop/hbase/spark/datasources/HBaseTableProviderSuite.scala b/spark4/hbase-spark4/src/test/scala/org/apache/hadoop/hbase/spark/datasources/HBaseTableProviderSuite.scala
new file mode 100644
index 0000000..7268ce6
--- /dev/null
+++ b/spark4/hbase-spark4/src/test/scala/org/apache/hadoop/hbase/spark/datasources/HBaseTableProviderSuite.scala
@@ -0,0 +1,219 @@
+/*
+ * 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.io.{File, FileOutputStream}
+import org.apache.hadoop.hbase.{HBaseTestingUtility, TableName}
+import org.apache.hadoop.hbase.client.{ConnectionFactory, Put}
+import org.apache.hadoop.hbase.spark.Logging
+import org.apache.hadoop.hbase.util.Bytes
+import org.apache.spark.sql.SparkSession
+import org.scalatest.BeforeAndAfterAll
+import org.scalatest.funsuite.AnyFunSuite
+
+class HBaseTableProviderSuite extends AnyFunSuite with BeforeAndAfterAll with Logging {
+
+ val TEST_UTIL = new HBaseTestingUtility
+ var spark: SparkSession = _
+ var configFile: File = _
+
+ val tableName = "test_provider"
+ val columnFamily = "cf"
+ val numRows = 20
+ val numRowsWithoutName = 3
+
+ val catalog: String = s"""{
+ |"table":{"namespace":"default", "name":"$tableName"},
+ |"rowkey":"key",
+ |"columns":{
+ |"key":{"cf":"rowkey", "col":"key", "type":"string"},
+ |"name":{"cf":"$columnFamily", "col":"name", "type":"string"},
+ |"age":{"cf":"$columnFamily", "col":"age", "type":"string"},
+ |"salary":{"cf":"$columnFamily", "col":"salary", "type":"string"}
+ |}
+ |}""".stripMargin
+
+ override def beforeAll(): Unit = {
+ TEST_UTIL.startMiniCluster()
+ logInfo(" - minicluster started")
+
+ TEST_UTIL.createTable(TableName.valueOf(tableName), Bytes.toBytes(columnFamily))
+ logInfo(s" - created table $tableName")
+
+ populateTestData()
+
+ val tmpDir = new File("target", "test-tmp")
+ tmpDir.mkdirs()
+ configFile = File.createTempFile("hbase-site", ".xml", tmpDir)
+ configFile.deleteOnExit()
+ val out = new FileOutputStream(configFile)
+ try {
+ TEST_UTIL.getConfiguration.writeXml(out)
+ } finally {
+ out.close()
+ }
+
+ spark = SparkSession.builder()
+ .master("local[2]")
+ .appName("HBaseTableProviderSuite")
+ .config("spark.hadoopRDD.ignoreEmptySplits", "false")
+ .getOrCreate()
+ }
+
+ override def afterAll(): Unit = {
+ if (spark != null) {
+ spark.stop()
+ }
+ if (configFile != null) {
+ configFile.delete()
+ }
+ TEST_UTIL.shutdownMiniCluster()
+ TEST_UTIL.cleanupTestDir()
+ }
+
+ private def populateTestData(): Unit = {
+ val connection = ConnectionFactory.createConnection(TEST_UTIL.getConfiguration)
+ val table = connection.getTable(TableName.valueOf(tableName))
+ try {
+ for (i <- 0 until numRows) {
+ val key = f"row$i%03d"
+ val put = new Put(Bytes.toBytes(key))
+ put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes("name"), Bytes.toBytes(s"Name$i"))
+ put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes("age"), Bytes.toBytes(s"${20 + i}"))
+ put.addColumn(
+ Bytes.toBytes(columnFamily),
+ Bytes.toBytes("salary"),
+ Bytes.toBytes(s"${30000 + i * 1000}"))
+ table.put(put)
+ }
+ for (i <- 0 until numRowsWithoutName) {
+ val key = f"noname$i%03d"
+ val put = new Put(Bytes.toBytes(key))
+ put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes("age"), Bytes.toBytes(s"${50 + i}"))
+ put.addColumn(
+ Bytes.toBytes(columnFamily),
+ Bytes.toBytes("salary"),
+ Bytes.toBytes(s"${60000 + i * 1000}"))
+ table.put(put)
+ }
+ } finally {
+ table.close()
+ connection.close()
+ }
+ }
+
+ private def loadTable() = {
+ spark.read
+ .format("org.apache.hadoop.hbase.spark.datasources.HBaseTableProvider")
+ .option("catalog", catalog)
+ .option(HBaseSparkConf.HBASE_CONFIG_LOCATION, configFile.getAbsolutePath)
+ .load()
+ }
+
+ test("full table scan returns all rows") {
+ val df = loadTable()
+ assert(df.count() == numRows + numRowsWithoutName)
+ }
+
+ test("select subset of columns") {
+ val df = loadTable().select("key", "name").filter("key LIKE 'row%'")
+ assert(df.columns.length == 2)
+ assert(df.count() == numRows)
+ val firstRow = df.orderBy("key").first()
+ assert(firstRow.getString(0) == "row000")
+ assert(firstRow.getString(1) == "Name0")
+ }
+
+ test("filter with EqualTo on non-row-key column") {
+ val df = loadTable().filter("name = 'Name5'")
+ assert(df.count() == 1)
+ val row = df.first()
+ assert(row.getAs[String]("key") == "row005")
+ }
+
+ test("filter with row key EqualTo narrows scan") {
+ val df = loadTable().filter("key = 'row010'")
+ assert(df.count() == 1)
+ val row = df.first()
+ assert(row.getAs[String]("name") == "Name10")
+ }
+
+ test("filter with row key range predicates") {
+ val df = loadTable().filter("key >= 'row005' AND key < 'row010'")
+ assert(df.count() == 5)
+ }
+
+ test("filter with inclusive row key range includes upper bound") {
+ val df = loadTable().filter("key >= 'row005' AND key <= 'row010'")
+ assert(df.count() == 6)
+ }
+
+ test("filter with row key GTE AND LTE on same value returns one row") {
+ val df = loadTable().filter("key >= 'row010' AND key <= 'row010'")
+ assert(df.count() == 1)
+ assert(df.first().getAs[String]("name") == "Name10")
+ }
+
+ test("count with filter on non-output columns") {
+ val df = loadTable().filter("name = 'Name3' AND age = '23'")
+ assert(df.count() == 1)
+ }
+
+ test("StringStartsWith filter") {
+ val df = loadTable().filter("name LIKE 'Name1%'")
+ assert(df.count() == 11)
+ }
+
+ test("SQL query via temp view") {
+ val df = loadTable()
+ df.createOrReplaceTempView("hbase_test")
+ val result = spark.sql("SELECT COUNT(*) as cnt FROM hbase_test WHERE age = '25'")
+ assert(result.first().getLong(0) == 1)
+ }
+
+ test("select all columns preserves schema") {
+ val df = loadTable()
+ assert(df.schema.fieldNames.sorted.sameElements(Array("age", "key", "name", "salary")))
+ }
+
+ test("empty result for non-matching filter") {
+ val df = loadTable().filter("name = 'NonExistent'")
+ assert(df.count() == 0)
+ }
+
+ test("IS NULL filter returns rows missing the column") {
+ val df = loadTable().select("key").filter("name IS NULL")
+ assert(df.count() == numRowsWithoutName)
+ val keys = df.orderBy("key").collect().map(_.getString(0))
+ assert(keys.sameElements(Array("noname000", "noname001", "noname002")))
+ }
+
+ test("IS NOT NULL filter excludes rows missing the column") {
+ val df = loadTable().select("key").filter("name IS NOT NULL")
+ assert(df.count() == numRows)
+ }
+
+ test("short name 'hbase' alias resolves via ServiceLoader") {
+ val df = spark.read
+ .format("hbase")
+ .option("catalog", catalog)
+ .option(HBaseSparkConf.HBASE_CONFIG_LOCATION, configFile.getAbsolutePath)
+ .load()
+ assert(df.count() == numRows + numRowsWithoutName)
+ }
+}
diff --git a/spark4/hbase-spark4/src/test/scala/org/apache/hadoop/hbase/spark/datasources/ScanRangeSuite.scala b/spark4/hbase-spark4/src/test/scala/org/apache/hadoop/hbase/spark/datasources/ScanRangeSuite.scala
new file mode 100644
index 0000000..7dea2f8
--- /dev/null
+++ b/spark4/hbase-spark4/src/test/scala/org/apache/hadoop/hbase/spark/datasources/ScanRangeSuite.scala
@@ -0,0 +1,162 @@
+/*
+ * 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.scalatest.funsuite.AnyFunSuite
+
+class ScanRangeSuite extends AnyFunSuite {
+
+ private def toBytes(s: String): Array[Byte] = Bytes.toBytes(s)
+
+ test("ScanRange mergeIntersect narrows bounds") {
+ val r1 = new ScanRange(toBytes("F"), true, toBytes("A"), true)
+ val r2 = new ScanRange(toBytes("D"), true, toBytes("C"), true)
+ r1.mergeIntersect(r2)
+ assert(Bytes.toString(r1.upperBound) == "D")
+ assert(Bytes.toString(r1.lowerBound) == "C")
+ assert(r1.isUpperBoundEqualTo)
+ assert(r1.isLowerBoundEqualTo)
+ }
+
+ test("ScanRange mergeIntersect with exclusive bounds") {
+ val r1 = new ScanRange(toBytes("F"), true, toBytes("A"), true)
+ val r2 = new ScanRange(toBytes("F"), false, toBytes("A"), false)
+ r1.mergeIntersect(r2)
+ assert(!r1.isUpperBoundEqualTo)
+ assert(!r1.isLowerBoundEqualTo)
+ }
+
+ test("ScanRange mergeUnion widens bounds") {
+ val r1 = new ScanRange(toBytes("D"), true, toBytes("C"), true)
+ val r2 = new ScanRange(toBytes("F"), true, toBytes("A"), true)
+ r1.mergeUnion(r2)
+ assert(Bytes.toString(r1.upperBound) == "F")
+ assert(Bytes.toString(r1.lowerBound) == "A")
+ assert(r1.isUpperBoundEqualTo)
+ assert(r1.isLowerBoundEqualTo)
+ }
+
+ test("ScanRange mergeUnion with same bounds picks inclusive") {
+ val r1 = new ScanRange(toBytes("D"), false, toBytes("A"), false)
+ val r2 = new ScanRange(toBytes("D"), true, toBytes("A"), true)
+ r1.mergeUnion(r2)
+ assert(r1.isUpperBoundEqualTo)
+ assert(r1.isLowerBoundEqualTo)
+ }
+
+ test("ScanRange containsPoint with inclusive bounds") {
+ val r = new ScanRange(toBytes("D"), true, toBytes("A"), true)
+ assert(r.containsPoint(toBytes("A")))
+ assert(r.containsPoint(toBytes("B")))
+ assert(r.containsPoint(toBytes("D")))
+ assert(!r.containsPoint(toBytes("E")))
+ }
+
+ test("ScanRange containsPoint with exclusive bounds") {
+ val r = new ScanRange(toBytes("D"), false, toBytes("A"), false)
+ assert(!r.containsPoint(toBytes("A")))
+ assert(r.containsPoint(toBytes("B")))
+ assert(!r.containsPoint(toBytes("D")))
+ }
+
+ test("ScanRange getOverLapScanRange returns overlap") {
+ val r1 = new ScanRange(toBytes("D"), true, toBytes("A"), true)
+ val r2 = new ScanRange(toBytes("F"), true, toBytes("C"), true)
+ val overlap = r1.getOverLapScanRange(r2)
+ assert(overlap != null)
+ assert(Bytes.toString(overlap.lowerBound) == "C")
+ assert(Bytes.toString(overlap.upperBound) == "D")
+ }
+
+ test("ScanRange getOverLapScanRange returns null for no overlap") {
+ val r1 = new ScanRange(toBytes("B"), true, toBytes("A"), true)
+ val r2 = new ScanRange(toBytes("F"), true, toBytes("D"), true)
+ val overlap = r1.getOverLapScanRange(r2)
+ assert(overlap == null)
+ }
+
+ test("ScanRange with null upper bound treated as unbounded") {
+ val r1 = new ScanRange(null, true, toBytes("A"), true)
+ val r2 = new ScanRange(toBytes("Z"), true, toBytes("A"), true)
+ r1.mergeIntersect(r2)
+ assert(Bytes.toString(r1.upperBound) == "Z")
+ }
+
+ test("ScanRange with empty lower bound treated as unbounded") {
+ val r1 = new ScanRange(toBytes("Z"), true, new Array[Byte](0), true)
+ val r2 = new ScanRange(toBytes("Z"), true, toBytes("A"), true)
+ r1.mergeIntersect(r2)
+ assert(Bytes.toString(r1.lowerBound) == "A")
+ }
+
+ test("RowKeyFilter mergeIntersect combines ranges") {
+ val f1 = new RowKeyFilter(null, new ScanRange(toBytes("F"), true, toBytes("A"), true))
+ val f2 = new RowKeyFilter(null, new ScanRange(toBytes("D"), true, toBytes("C"), true))
+ f1.mergeIntersect(f2)
+ assert(f1.ranges.nonEmpty)
+ }
+
+ test("RowKeyFilter mergeUnion accumulates points") {
+ val f1 = new RowKeyFilter(toBytes("A"), null)
+ val f2 = new RowKeyFilter(toBytes("B"), null)
+ f1.mergeUnion(f2)
+ assert(f1.points.size == 2)
+ assert(f1.points.exists(Bytes.equals(_, toBytes("A"))))
+ assert(f1.points.exists(Bytes.equals(_, toBytes("B"))))
+ }
+
+ test("RowKeyFilter with point only") {
+ val f = new RowKeyFilter(toBytes("key1"), null)
+ assert(f.points.size == 1)
+ assert(f.ranges.isEmpty)
+ }
+
+ test("RowKeyFilter with range only") {
+ val f = new RowKeyFilter(null, new ScanRange(toBytes("Z"), true, toBytes("A"), true))
+ assert(f.points.isEmpty)
+ assert(f.ranges.size == 1)
+ }
+
+ test("RowKeyFilter default has empty range covering all") {
+ val f = new RowKeyFilter()
+ assert(f.ranges.size == 1)
+ assert(f.ranges.head.lowerBound.sameElements(new Array[Byte](0)))
+ assert(f.ranges.head.upperBound == null)
+ }
+
+ test("Ranges.and retains singleton range with both bounds inclusive") {
+ val region = Range(Some(Bound(new Array[Byte](0), true)), None)
+ val singleton = Range(
+ Some(Bound(toBytes("K"), true)),
+ Some(Bound(toBytes("K"), true)))
+ val result = Ranges.and(region, Seq(singleton))
+ assert(result.size == 1)
+ assert(result.head.lower.get.b.sameElements(toBytes("K")))
+ assert(result.head.upper.get.b.sameElements(toBytes("K")))
+ }
+
+ test("Ranges.and discards singleton range with exclusive bound") {
+ val region = Range(Some(Bound(new Array[Byte](0), true)), None)
+ val exclusive = Range(
+ Some(Bound(toBytes("K"), true)),
+ Some(Bound(toBytes("K"), false)))
+ val result = Ranges.and(region, Seq(exclusive))
+ assert(result.isEmpty)
+ }
+}