txn: fix scan missing endKey in V2 API

- Add scan(backOffer, startKey, endKey, version, keyOnly) overload
- Use codec.encodeRange() to set both startKey and endKey
- Decode response KvPairs via codec.decodeKvPairs()
- Pass rangeEndKey through ScanIterator

Signed-off-by: lilei <lilei592@jd.com>

Author: lilei-jd

src/main/java/org/tikv/common/operation/iterator/ConcreteScanIterator.java

@@ -76,7 +76,7 @@ TiRegion loadCurrentRegionToCache() throws GrpcException {
     try (RegionStoreClient client = builder.build(startKey)) {
       client.setTimeout(conf.getScanTimeout());
       BackOffer backOffer = ConcreteBackOffer.newScannerNextMaxBackOff();
-      currentCache = client.scan(backOffer, startKey, version);
+      currentCache = client.scan(backOffer, startKey, rangeEndKey, version, keyOnly);
       // If we get region before scan, we will use region from cache which
       // may have wrong end key. This may miss some regions that split from old region.
       // Client will get the newest region during scan. So we need to
@@ -115,7 +115,8 @@ public boolean hasNext() {
     // for last batch to be processed, we have to check if
     return !processingLastBatch
         || current == null
-        || (hasEndKey && Key.toRawKey(current.getKey()).compareTo(endKey) < 0);
+        || !hasEndKey
+        || (Key.toRawKey(current.getKey()).compareTo(endKey) < 0);
   }
 
   @Override

src/main/java/org/tikv/common/operation/iterator/ScanIterator.java

@@ -35,6 +35,7 @@ public abstract class ScanIterator implements Iterator<Kvrpcpb.KvPair> {
   protected final RegionStoreClientBuilder builder;
   protected List<Kvrpcpb.KvPair> currentCache;
   protected ByteString startKey;
+  protected ByteString rangeEndKey;
   protected int index = -1;
   protected int limit;
   protected boolean keyOnly;
@@ -52,7 +53,8 @@ public abstract class ScanIterator implements Iterator<Kvrpcpb.KvPair> {
       int limit,
       boolean keyOnly) {
     this.startKey = requireNonNull(startKey, "start key is null");
-    this.endKey = Key.toRawKey(requireNonNull(endKey, "end key is null"));
+    this.rangeEndKey = requireNonNull(endKey, "end key is null");
+    this.endKey = Key.toRawKey(this.rangeEndKey);
     this.hasEndKey = !endKey.isEmpty();
     this.limit = limit;
     this.keyOnly = keyOnly;
@@ -106,9 +108,11 @@ boolean cacheLoadFails() {
       }
       // notify last batch if lastKey is greater than or equal to endKey
       // if startKey is empty, it indicates +∞
-      if (hasEndKey && lastKey.compareTo(endKey) >= 0 || startKey.isEmpty()) {
+      if (hasEndKey && lastKey.compareTo(endKey) >= 0) {
         processingLastBatch = true;
         startKey = null;
+      } else if (startKey.isEmpty()) {
+        processingLastBatch = true;
       }
     } catch (Exception e) {
       throw new TiClientInternalException("Error scanning data from region.", e);

src/main/java/org/tikv/common/region/RegionStoreClient.java

@@ -337,15 +337,23 @@ private List<KvPair> handleBatchGetResponse(
 
   public List<KvPair> scan(
       BackOffer backOffer, ByteString startKey, long version, boolean keyOnly) {
+    return scan(backOffer, startKey, ByteString.EMPTY, version, keyOnly);
+  }
+
+  public List<KvPair> scan(
+      BackOffer backOffer, ByteString startKey, ByteString endKey, long version, boolean keyOnly) {
     boolean forWrite = false;
     while (true) {
+      Pair<ByteString, ByteString> range = codec.encodeRange(startKey, endKey);
+
       Supplier<ScanRequest> request =
           () ->
               ScanRequest.newBuilder()
                   .setContext(
                       makeContext(
                           getResolvedLocks(version), this.storeType, backOffer.getSlowLog()))
-                  .setStartKey(codec.encodeKey(startKey))
+                  .setStartKey(range.first)
+                  .setEndKey(range.second)
                   .setVersion(version)
                   .setKeyOnly(keyOnly)
                   .setLimit(getConf().getScanBatchSize())
@@ -367,7 +375,7 @@ public List<KvPair> scan(
       region = regionManager.getRegionByKey(startKey, backOffer);
 
       if (handleScanResponse(backOffer, resp, version, forWrite)) {
-        return resp.getPairsList();
+        return codec.decodeKvPairs(resp.getPairsList());
       }
     }
   }