Implementing DataSource V2 Read Path - #162
Conversation
There was a problem hiding this comment.
Pull request overview
Implements the Spark 4 DataSource V2 read path for HBase, including filter pushdown, partition planning, and row conversion.
Changes:
- Adds the V2 provider, table, scan, batch, partition, and reader pipeline.
- Ports row-key range and predicate handling.
- Adds unit and mini-cluster integration tests.
Reviewed changes
Copilot reviewed 13 out of 13 changed files in this pull request and generated 9 comments.
Show a summary per file
| File | Description |
|---|---|
ScanRange.scala |
Adds row-key range operations. |
HBaseTableProvider.scala |
Adds the V2 provider entry point. |
HBaseTable.scala |
Declares table schema and read capability. |
HBaseScanBuilder.scala |
Negotiates filters and projections. |
HBaseScan.scala |
Builds the logical HBase scan. |
HBaseBatch.scala |
Plans region-based input partitions. |
HBaseInputPartition.scala |
Defines serializable partitions. |
HBasePartitionReaderFactory.scala |
Creates partition readers. |
HBasePartitionReader.scala |
Executes scans and creates rows. |
ScanRangeSuite.scala |
Tests range operations. |
HBaseScanBuilderSuite.scala |
Tests filter and column pushdown. |
HBaseTableCatalogSuite.scala |
Tests catalog parsing. |
HBaseTableProviderSuite.scala |
Tests end-to-end reads. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
| case f @ Or(_, _) => supported += f | ||
| case f @ And(_, _) => supported += f |
There was a problem hiding this comment.
recheck with Cursor, they found the same with below comments
Or/And are pushed even when nested predicates reference unknown columns. V1 effectively only pushed filters it could interpret. Consider recursively validating leaf attributes, or only pushing compounds whose children are all supported.
| private var _pushedFilters: Array[Filter] = Array.empty | ||
| private var requiredSchema: StructType = schema | ||
|
|
||
| override def pushFilters(filters: Array[Filter]): Array[Filter] = { |
| val startRow = intersectedRanges.headOption.flatMap(_.lower).map(_.b) | ||
| .orElse(intersectedPoints.headOption) | ||
| .orElse(region.start) | ||
| .orNull | ||
| val stopRow = intersectedRanges.lastOption.flatMap(_.upper).map(_.b) | ||
| .orElse(intersectedPoints.lastOption.map(Utils.incrementByteArray)) | ||
| .orElse(region.end) |
| .orElse(intersectedPoints.headOption) | ||
| .orElse(region.start) | ||
| .orNull | ||
| val stopRow = intersectedRanges.lastOption.flatMap(_.upper).map(_.b) |
There was a problem hiding this comment.
this is the bigger gap I checked again with cursor as well.
| val intersectedRanges = Ranges.and(regionRange, scanRanges.map { sr => | ||
| Range( | ||
| Option(sr.lowerBound).filter(_.nonEmpty).map(Bound(_, sr.isLowerBoundEqualTo)), | ||
| Option(sr.upperBound).map(Bound(_, sr.isUpperBoundEqualTo))) | ||
| }) |
| catalog.dynSetupRowKey(rowKey) | ||
| val keyFields = catalog.getRowKey |
| private def convertToInternalRow(value: Any, dataType: DataType): Any = { | ||
| if (value == null) return null | ||
| dataType match { | ||
| case StringType => UTF8String.fromString(value.asInstanceOf[String]) | ||
| case _ => value | ||
| } |
| val blockCacheEnable = properties | ||
| .get(HBaseSparkConf.QUERY_CACHEBLOCKS) | ||
| .map(_.toBoolean) | ||
| .getOrElse(HBaseSparkConf.DEFAULT_QUERY_CACHEBLOCKS) | ||
| scan.setCacheBlocks(blockCacheEnable) |
| * as a short name for format("org.apache.hadoop.hbase.spark.datasources.HBaseTableProvider"). | ||
| * @return | ||
| */ | ||
| override def shortName(): String = "hbase" |
taklwu
left a comment
taklwu
left a comment
There was a problem hiding this comment.
may you fix the commit title with JIRA prefix HBASE-30189 ?
| .orElse(intersectedPoints.headOption) | ||
| .orElse(region.start) | ||
| .orNull | ||
| val stopRow = intersectedRanges.lastOption.flatMap(_.upper).map(_.b) |
There was a problem hiding this comment.
this is the bigger gap I checked again with cursor as well.
| .orElse(intersectedPoints.lastOption.map(Utils.incrementByteArray)) | ||
| .orElse(region.end) | ||
| .orNull | ||
| Some(HBaseInputPartition(region.index, startRow, stopRow): InputPartition) |
There was a problem hiding this comment.
seems like a behavior difference from V1 that HBaseInputPartition that uses HBasePartitionReader.scala are scanning instead of Get , do you think this is good ? if not , please try to align what Spark3 does with Get.
| case f @ Or(_, _) => supported += f | ||
| case f @ And(_, _) => supported += f |
There was a problem hiding this comment.
recheck with Cursor, they found the same with below comments
Or/And are pushed even when nested predicates reference unknown columns. V1 effectively only pushed filters it could interpret. Consider recursively validating leaf attributes, or only pushing compounds whose children are all supported.
| val blockCacheEnable = properties | ||
| .get(HBaseSparkConf.QUERY_CACHEBLOCKS) | ||
| .map(_.toBoolean) | ||
| .getOrElse(HBaseSparkConf.DEFAULT_QUERY_CACHEBLOCKS) | ||
| scan.setCacheBlocks(blockCacheEnable) |
| extends Batch | ||
| with Logging { | ||
|
|
||
| override def planInputPartitions(): Array[InputPartition] = { |
There was a problem hiding this comment.
we may need to support getPreferredLocations like V1 does , you can have it in the future PR.
No description provided.