HBASE-26863 fix incorrect rowkey pushdown

Author: morokosi

spark/hbase-spark/src/main/scala/org/apache/hadoop/hbase/spark/DefaultSource.scala

@@ -593,11 +593,11 @@ class ScanRange(var upperBound:Array[Byte], var isUpperBoundEqualTo:Boolean,
 
     isLowerBoundEqualTo = if (lowerBoundCompare == 0)
       isLowerBoundEqualTo && other.isLowerBoundEqualTo
-    else isLowerBoundEqualTo
+    else if (lowerBoundCompare < 0) other.isLowerBoundEqualTo else isLowerBoundEqualTo
 
     isUpperBoundEqualTo = if (upperBoundCompare == 0)
       isUpperBoundEqualTo && other.isUpperBoundEqualTo
-    else isUpperBoundEqualTo
+    else if (upperBoundCompare < 0) isUpperBoundEqualTo else other.isUpperBoundEqualTo
   }
 
   /**
@@ -643,7 +643,6 @@ class ScanRange(var upperBound:Array[Byte], var isUpperBoundEqualTo:Boolean,
    * @return      True is overlap false is not overlap
    */
   def getOverLapScanRange(other:ScanRange): ScanRange = {
-
     var leftRange:ScanRange = null
     var rightRange:ScanRange = null
 
@@ -659,14 +658,9 @@ class ScanRange(var upperBound:Array[Byte], var isUpperBoundEqualTo:Boolean,
     }
 
     if (hasOverlap(leftRange, rightRange)) {
-      // Find the upper bound and lower bound
-      if (compareRange(leftRange.upperBound, rightRange.upperBound) >= 0) {
-        new ScanRange(rightRange.upperBound, rightRange.isUpperBoundEqualTo,
-          rightRange.lowerBound, rightRange.isLowerBoundEqualTo)
-      } else {
-        new ScanRange(leftRange.upperBound, leftRange.isUpperBoundEqualTo,
-          rightRange.lowerBound, rightRange.isLowerBoundEqualTo)
-      }
+      val result = new ScanRange(upperBound, isUpperBoundEqualTo, lowerBound, isLowerBoundEqualTo)
+      result.mergeIntersect(other)
+      result
     } else {
       null
     }

spark/hbase-spark/src/test/scala/org/apache/hadoop/hbase/spark/DefaultSourceSuite.scala

@@ -611,6 +611,26 @@ BeforeAndAfterEach with BeforeAndAfterAll with Logging {
     assert(scanRange1.isUpperBoundEqualTo)
   }
 
+  test("Test Rowkey And with complex logic (HBASE-26863)") {
+    val results = sqlContext.sql("SELECT KEY_FIELD, B_FIELD, A_FIELD FROM hbaseTable1 " +
+      "WHERE " +
+      "( KEY_FIELD >= 'get1' AND KEY_FIELD <= 'get3' ) AND (A_FIELD = 'foo1' OR B_FIELD = '8')").take(10)
+    val executionRules = DefaultSourceStaticUtils.lastFiveExecutionRules.poll()
+    assert(results.length == 2)
+
+    assert(executionRules.dynamicLogicExpression.toExpressionString
+      == "( ( ( KEY_FIELD isNotNull AND KEY_FIELD >= 0 ) AND KEY_FIELD <= 1 ) AND ( A_FIELD == 2 OR B_FIELD == 3 ) )")
+
+    assert(executionRules.rowKeyFilter.points.size == 0)
+    assert(executionRules.rowKeyFilter.ranges.size == 1)
+
+    val scanRange1 = executionRules.rowKeyFilter.ranges.get(0).get
+    assert(Bytes.equals(scanRange1.lowerBound,Bytes.toBytes("get1")))
+    assert(Bytes.equals(scanRange1.upperBound, Bytes.toBytes("get3")))
+    assert(scanRange1.isLowerBoundEqualTo)
+    assert(scanRange1.isUpperBoundEqualTo)
+  }
+
   test("Test table that doesn't exist") {
     val catalog = s"""{
             |"table":{"namespace":"default", "name":"t1NotThere"},