Copilot commented on code in PR #172:
URL: https://github.com/apache/hbase-connectors/pull/172#discussion_r4444964994
##########
spark/hbase-spark/src/main/scala/org/apache/hadoop/hbase/spark/DefaultSource.scala:
##########
@@ -1142,6 +1142,31 @@ class RowKeyFilter(
result
}
+ // Normalize after the complete filter tree is built, before region pruning
and Get batching.
+ def normalize(): RowKeyFilter = {
+ val normalizedRanges = new mutable.MutableList[ScanRange]()
Review Comment:
`mutable.MutableList` is deprecated in modern Scala and is generally
discouraged. Since this method is newly added/updated, prefer a non-deprecated
collection (e.g., `scala.collection.mutable.ListBuffer`) for `normalizedRanges`
to avoid deprecation churn and align with the Spark4 connector implementation.
##########
spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/ScanRange.scala:
##########
@@ -145,6 +145,31 @@ class RowKeyFilter(
if (currentRange != null) ranges += currentRange
if (currentPoint != null) points += currentPoint
+ // Normalize after the complete filter tree is built, before region pruning
and Get batching.
+ def normalize(): RowKeyFilter = {
+ val normalizedRanges = new ListBuffer[ScanRange]()
+ ranges
+ .sortWith((left, right) => Bytes.compareTo(left.lowerBound,
right.lowerBound) < 0)
Review Comment:
`Bytes.compareTo(left.lowerBound, right.lowerBound)` will throw if either
`lowerBound` is `null`. Elsewhere in the codebase (e.g., scan planning)
`lowerBound` can be `null` to represent an unbounded/empty start. Update the
sort comparator to handle `null` safely (e.g., treat `null` as
`Array.emptyByteArray` or explicitly order `null` first), so `normalize()`
cannot NPE on full-scan / unbounded-range filters.
##########
spark/hbase-spark/src/main/scala/org/apache/hadoop/hbase/spark/datasources/HBaseTableScanRDD.scala:
##########
@@ -182,9 +183,10 @@ class HBaseTableScanRDD(
filter: Option[SparkSQLPushDownFilter],
columns: Seq[Field]): Scan = {
val scan = (range.lower, range.upper) match {
- case (Some(Bound(a, b)), Some(Bound(c, d))) => new Scan(a, c)
+ case (Some(Bound(a, b)), Some(Bound(c, d))) =>
+ new Scan().withStartRow(a, b).withStopRow(c, d)
case (None, Some(Bound(c, d))) => new Scan(Array[Byte](), c)
Review Comment:
The `(None, Some(Bound(c, d)))` case ignores the stop-row inclusivity flag
`d`, which can change query semantics for `<=` / inclusive upper bounds. Use
the inclusive-aware API for stop rows in this branch as well (e.g., construct a
`Scan` and apply `withStopRow(c, d)`), to keep behavior consistent with the
other cases.
##########
spark4/hbase-spark4/src/main/scala/org/apache/hadoop/hbase/spark/datasources/ScanRange.scala:
##########
@@ -145,6 +145,31 @@ class RowKeyFilter(
if (currentRange != null) ranges += currentRange
if (currentPoint != null) points += currentPoint
+ // Normalize after the complete filter tree is built, before region pruning
and Get batching.
+ def normalize(): RowKeyFilter = {
+ val normalizedRanges = new ListBuffer[ScanRange]()
+ ranges
+ .sortWith((left, right) => Bytes.compareTo(left.lowerBound,
right.lowerBound) < 0)
+ .foreach {
+ range =>
+ normalizedRanges.lastOption match {
+ case Some(previous) =>
+ val cmp = previous.compareRange(previous.upperBound,
range.lowerBound)
+ if (cmp > 0 ||
+ (cmp == 0 && (previous.isUpperBoundEqualTo ||
range.isLowerBoundEqualTo))) {
+ previous.mergeUnion(range)
+ } else {
+ normalizedRanges += range
+ }
+ case None => normalizedRanges += range
+ }
+ }
+ ranges = normalizedRanges
+ val seen = new java.util.TreeSet[Array[Byte]](Bytes.BYTES_COMPARATOR)
+ points = points.filter(p => seen.add(p) &&
!ranges.exists(_.containsPoint(p)))
Review Comment:
After ranges are normalized and made disjoint, pruning points with
`ranges.exists(_.containsPoint(p))` is O(#points × #ranges). If row-key
predicates can produce many points/ranges (e.g., large IN-lists combined with
unions), this may become a hot path. Consider optimizing membership checks
using the fact that `ranges` are sorted/disjoint (e.g., binary search over
range boundaries, or a single pass if you also process points in sorted order),
while preserving the desired point ordering semantics.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
