HBASE-30186 Port basic HBase operations to spark4 HBaseContext - #158
Open
wchevreuil wants to merge 1 commit into
Open
HBASE-30186 Port basic HBase operations to spark4 HBaseContext#158wchevreuil wants to merge 1 commit into
wchevreuil wants to merge 1 commit into
Conversation
Co-authored-by: Claude Code (claude-opus-4-6) <no-reply@anthropic.com>
Contributor
There was a problem hiding this comment.
Pull request overview
This PR ports core HBase interaction utilities into the Spark 4 connector by expanding HBaseContext beyond scan-only support to include bulk put/get/delete and connection-aware partition helpers, plus the supporting connection cache and RDD implicit wrappers.
Changes:
- Add Spark-4 implementations of
HBaseConnectionCacheandHBaseRDDFunctions(implicit RDD helpers). - Extend
HBaseContextwithforeachPartition,mapPartitions,bulkPut,bulkGet, andbulkDeleteimplementations backed by the connection cache. - Expand/introduce test suites covering bulk operations and the connection cache.
Reviewed changes
Copilot reviewed 5 out of 5 changed files in this pull request and generated 9 comments.
Show a summary per file
| File | Description |
|---|---|
| spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/HBaseContext.scala | Adds bulk put/get/delete and connection-aware partition helpers for Spark 4. |
| spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/HBaseConnectionCache.scala | Introduces a cached-connection layer with housekeeping cleanup for Spark 4. |
| spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/HBaseRDDFunctions.scala | Adds implicit RDD extension methods that delegate to HBaseContext bulk/partition APIs. |
| spark4/hbase-spark4/src/test/scala/org/apache/hadoop/hbase/spark/HBaseContextSuite.scala | Adds tests for the newly ported bulk operations and partition helpers. |
| spark4/hbase-spark4/src/test/scala/org/apache/hadoop/hbase/spark/HBaseConnectionCacheSuite.scala | Adds Spark 4 tests validating the connection cache behavior under load. |
Comments suppressed due to low confidence (3)
spark4/hbase-spark4/src/test/scala/org/apache/hadoop/hbase/spark/HBaseConnectionCacheSuite.scala:224
- Threads are created but executed with Thread.run(), which runs sequentially on the current thread. To validate connection-cache behavior under concurrent pressure, start the threads instead.
for (i <- threads.indices) {
threads.update(i, new Thread(new TestThread()))
threads(i).run()
}
spark4/hbase-spark4/src/test/scala/org/apache/hadoop/hbase/spark/HBaseConnectionCacheSuite.scala:229
- The InterruptedException is swallowed via println, which clears the interrupt status and can hide test failures. Re-interrupt the thread and rethrow so the test fails appropriately.
try {
threads.foreach { x => x.join() }
} catch {
case e: InterruptedException => println(e.getMessage)
}
spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/HBaseContext.scala:286
- smartConn can be null when HBaseConnectionCache is closed, but smartConn.connection is dereferenced unconditionally, which will throw a NullPointerException. Fail fast with a clear exception before dereferencing, and always close the SmartConnection in finally.
val config = getConf(configBroadcast)
val smartConn = HBaseConnectionCache.getConnection(config)
try {
mp(it, smartConn.connection)
} finally {
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+170
to
+173
| for (i <- 0 to 99) { | ||
| threads.update(i, new Thread(new TestThread())) | ||
| threads(i).run() | ||
| } |
Comment on lines
+174
to
+178
| try { | ||
| threads.foreach { x => x.join() } | ||
| } catch { | ||
| case e: InterruptedException => println(e.getMessage) | ||
| } |
Comment on lines
+90
to
+108
| connectionMap.synchronized { | ||
| connectionMap.foreach { | ||
| x => | ||
| { | ||
| if (x._2.refCount < 0) { | ||
| logError(s"Bug to be fixed: negative refCount of connection ${x._2}") | ||
| } | ||
|
|
||
| if (forceClean || ((x._2.refCount <= 0) && (tsNow - x._2.timestamp > timeout))) { | ||
| try { | ||
| x._2.connection.close() | ||
| } catch { | ||
| case e: IOException => logWarning(s"Fail to close connection ${x._2}", e) | ||
| } | ||
| connectionMap.remove(x._1) | ||
| } | ||
| } | ||
| } | ||
| } |
Comment on lines
+268
to
+274
| val config = getConf(configBroadcast) | ||
| val smartConn = HBaseConnectionCache.getConnection(config) | ||
| try { | ||
| f(it, smartConn.connection) | ||
| } finally { | ||
| if (smartConn != null) smartConn.close() | ||
| } |
Comment on lines
+117
to
+120
| val m = connection.getBufferedMutator(TableName.valueOf(tName)) | ||
| iterator.foreach(T => m.mutate(f(T))) | ||
| m.flush() | ||
| m.close() |
Comment on lines
+304
to
+318
| val table = connection.getTable(TableName.valueOf(tName)) | ||
| val mutationList = new java.util.ArrayList[Mutation] | ||
| iterator.foreach( | ||
| T => { | ||
| mutationList.add(f(T)) | ||
| if (mutationList.size >= batchSize) { | ||
| table.batch(mutationList, null) | ||
| mutationList.clear() | ||
| } | ||
| }) | ||
| if (mutationList.size() > 0) { | ||
| table.batch(mutationList, null) | ||
| mutationList.clear() | ||
| } | ||
| table.close() |
Comment on lines
+331
to
+353
| def run(iterator: Iterator[T], connection: Connection): Iterator[U] = { | ||
| val table = connection.getTable(TableName.valueOf(tName)) | ||
|
|
||
| val gets = new java.util.ArrayList[Get]() | ||
| var res = List[U]() | ||
|
|
||
| while (iterator.hasNext) { | ||
| gets.add(makeGet(iterator.next())) | ||
|
|
||
| if (gets.size() == batchSize) { | ||
| val results = table.get(gets) | ||
| res = res ++ results.map(convertResult) | ||
| gets.clear() | ||
| } | ||
| } | ||
| if (gets.size() > 0) { | ||
| val results = table.get(gets) | ||
| res = res ++ results.map(convertResult) | ||
| gets.clear() | ||
| } | ||
| table.close() | ||
| res.iterator | ||
| } |
Comment on lines
+104
to
+115
| hc.bulkGet[T, (ImmutableBytesWritable, Result)]( | ||
| tableName, | ||
| batchSize, | ||
| rdd, | ||
| f, | ||
| result => | ||
| if (result != null && result.getRow != null) { | ||
| (new ImmutableBytesWritable(result.getRow), result) | ||
| } else { | ||
| null | ||
| }) | ||
| } |
Comment on lines
+401
to
+411
| val tbl = conn.getTable(TableName.valueOf(tName)) | ||
| val res = new ListBuffer[String]() | ||
| it.foreach { rowKey => | ||
| val result = tbl.get(new Get(rowKey)) | ||
| val cell = result.getColumnLatestCell(Bytes.toBytes(cf), Bytes.toBytes("a")) | ||
| if (cell != null) { | ||
| res += Bytes.toString(result.getRow) + "=" + Bytes.toString(CellUtil.cloneValue(cell)) | ||
| } | ||
| } | ||
| tbl.close() | ||
| res.iterator |
taklwu
reviewed
Jul 28, 2026
taklwu
left a comment
taklwu
left a comment
Contributor
There was a problem hiding this comment.
please check those minor optimizations suggested by copilot, otherwise LGTM
| } | ||
| } | ||
|
|
||
| test("bulkGet to test HBase client") { |
Contributor
There was a problem hiding this comment.
nit: a failure test for bulkGet (bad table) in HBaseContextSuite of hbase-spark was missing but it should be fine
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
No description provided.