Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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);
Expand Down
12 changes: 10 additions & 2 deletions src/main/java/org/tikv/common/region/RegionStoreClient.java
Original file line number Diff line number Diff line change
Expand Up @@ -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())
Expand All @@ -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());
}
}
}
Expand Down
Loading