HBASE-30187 Port hbase-spark bulk load functionality to Spark 4 - #163
wchevreuil merged 2 commits into
Conversation
There was a problem hiding this comment.
Warning
Copilot couldn't run its full agentic review because it didn't start before the timeout. Make sure your repository has a runner available, or add a copilot-code-review.yml file specifying one with the runs-on attribute. See the docs for more details.
Pull request overview
Ports HBase Spark bulk-load APIs and supporting utilities from the Spark 3 connector into the Spark 4 module, including Scala 2.13/Spark 4 test coverage.
Changes:
- Added Spark 4 bulk-load APIs (
bulkLoad,bulkLoadThinRows) and RDD extension methods. - Added supporting helper classes for partitioning/sorting and HFile writing.
- Ported bulk-load test suite(s) to Spark 4 / Scala 2.13 and reformatted Spark 4 sources.
Reviewed changes
Copilot reviewed 10 out of 10 changed files in this pull request and generated 10 comments.
Show a summary per file
| File | Description |
|---|---|
| spark4/hbase-spark4/src/test/scala/org/apache/hadoop/hbase/spark/HBaseContextSuite.scala | Minor formatting updates in existing tests. |
| spark4/hbase-spark4/src/test/scala/org/apache/hadoop/hbase/spark/HBaseConnectionCacheSuite.scala | Formatting updates (imports/indentation) in existing tests. |
| spark4/hbase-spark4/src/test/scala/org/apache/hadoop/hbase/spark/BulkLoadSuite.scala | Added Spark 4 bulk-load test suite covering wide/thin row bulk load behaviors. |
| spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/HBaseRDDFunctions.scala | Added RDD extension methods for bulk-load APIs. |
| spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/HBaseContext.scala | Implemented bulk-load and thin-row bulk-load execution + HFile writing helpers. |
| spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/BulkLoadPartitioner.scala | Added region-split-aware Spark partitioner used by bulk-load shuffles. |
| spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/KeyFamilyQualifier.scala | Added sortable key type for wide-row bulk-load shuffles. |
| spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/ByteArrayWrapper.scala | Added byte-array wrapper to support map keys / ordering. |
| spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/FamiliesQualifiersValues.scala | Added container for thin-row in-memory family/qualifier/value sorting. |
| spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/FamilyHFileWriteOptions.scala | Added per-family HFile writer options model. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
| val comparator: Comparator[Array[Byte]] = new Comparator[Array[Byte]] { | ||
| override def compare(o1: Array[Byte], o2: Array[Byte]): Int = { | ||
| Bytes.compareTo(o1, o2) | ||
| } | ||
| } |
wchevreuil
left a comment
wchevreuil
left a comment
There was a problem hiding this comment.
Apologies for the delayed review, @Sigma-Ma . Overall, looks good. Can you please address the review comments? I have resolved copilot ones I thought were not relevant.
| def bulkDelete[T]( | ||
| rdd: RDD[T], | ||
| tableName: TableName, | ||
| f: (T) => Delete, | ||
| batchSize: Integer): Unit = { | ||
| bulkMutation(rdd, tableName, f, batchSize) | ||
| } |
There was a problem hiding this comment.
can we revert these non-logic changes?
| iterator.foreach { | ||
| t => | ||
| mutationList.add(f(t)) | ||
| if (mutationList.size >= batchSize) { | ||
| table.batch(mutationList, null) | ||
| mutationList.clear() | ||
| } |
There was a problem hiding this comment.
can we revert these non-logic changes?
There was a problem hiding this comment.
Done. I reverted the unrelated formatting changes and restored the formatting-only test changes to match master.
|
Thanks for the review. I addressed the remaining comments:
I left the staging-directory test and Partitioner equals/hashCode unchanged, as they are not needed for this port. |
Can you push these changes? I don't see any commit in the PR branch. |
Apologies, I forgot to push the changes due to work commitments and only left the comment. The changes have now been pushed. Thanks for the reminder! |
There was a problem hiding this comment.
🟡 Changes recommended
One or more issues must be addressed before approval.
Once you've addressed the issues Copilot identified, you can request another Copilot review.
Review details
Suppressed comments (5)
Previously missed (3) — in code that hasn't changed since the last review.
spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/HBaseContext.scala:427
getRegionLocatorreturns a closeable locator, but this method only reads its start keys and never closes it. Sinceconn.close()only releases the cached connection reference, repeated bulk-load calls can retain locator resources; close the locator in afinallyafter obtaining the keys, and apply the same fix tobulkLoadThinRows.
This issue also appears on line 557 of the same file.
spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/HBaseContext.scala:503
- This final
rollWriterscall only runs after the partition iterator completes normally. If HFile creation/append, filesystem I/O, or the userflatMapthrows, the open writers and underscore-prefixed partial files are left behind; Spark retries can accumulate them, and the method's existing-path check then prevents reusing the staging directory. Add failure cleanup that closes/removes unfinished files without renaming them, then rethrow the original error.
spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/HBaseContext.scala:789 - A new
RegionLocatoris created for every executor partition/family here and is never closed. With many partitions or families this accumulates locator resources on the shared HBase connection; close it in a nestedfinallyaftergetRegionLocation(or create one locator per partition and close it after the partition).
spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/HBaseContext.scala:558
- The thin-row implementation has the same unclosed
RegionLocatorproblem here: every bulk-load call acquires one to read the region start keys, but only decrements the cached connection reference. Close this locator aftergetStartKeysso repeated calls do not accumulate client resources.
val regionLocator = conn.getRegionLocator(tableName)
val startKeys = regionLocator.getStartKeys
spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/HBaseContext.scala:670
- The thin-row callback has the same failure path: any exception, including the intentional duplicate-row validation error, skips this final close. That leaves open writers and partial underscore files in the staging directory, so retries can leak resources and cannot reuse the path; add exception cleanup that closes/removes unfinished files without renaming them.
// This will get a writer for the column family
// If there is no writer for a given column family then
// it will get created here.
// We have finished all the data so lets close up the writers
rollWriters(fs, writerMap, regionSplitPartitioner, previousRow, compactionExclude)
rollOverRequested = false
})
- Files reviewed: 8/8 changed files
- Comments generated: 1
- Review effort level: Lite
JIRA: https://issues.apache.org/jira/browse/HBASE-30187
What changes were proposed in this pull request?
Port the Spark bulk-load APIs from the Spark 3 connector to the Spark 4 module. This adds
bulkLoad,bulkLoadThinRows, the related RDD extension methods, and the helper classes used to partition rows and write HFiles. It also ports the existing bulk-load tests to Scala 2.13 and Spark 4.The Spark 4 sources are formatted with the project Spotless configuration.
Why are the changes needed?
The Spark 4 connector does not currently provide the HFile bulk-load functionality available in the Spark 3 connector. This change provides the same wide-row and thin-row bulk-load support for Spark 4 users.
How was this patch tested?