feat: rheakv support scan in reverse

jraft-example/src/main/java/com/alipay/sofa/jraft/example/rheakv/ReverseScanExample.java

@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.alipay.sofa.jraft.example.rheakv;
+
+import com.alipay.sofa.jraft.rhea.client.RheaKVStore;
+import com.alipay.sofa.jraft.rhea.storage.KVEntry;
+import com.alipay.sofa.jraft.rhea.util.Lists;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
+import static com.alipay.sofa.jraft.util.BytesUtil.readUtf8;
+import static com.alipay.sofa.jraft.util.BytesUtil.writeUtf8;
+
+/**
+ *
+ * @author baozi
+ */
+public class ReverseScanExample {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ReverseScanExample.class);
+
+ public static void main(final String[] args) throws Exception {
+ final Client client = new Client();
+ client.init();
+ scan(client.getRheaKVStore());
+ client.shutdown();
+ }
+
+ @SuppressWarnings("unchecked")
+ public static void scan(final RheaKVStore rheaKVStore) {
+ final List<byte[]> keys = Lists.newArrayList();
+ for (int i = 0; i < 10; i++) {
+ final byte[] bytes = writeUtf8("scan_demo_" + i);
+ keys.add(bytes);
+ rheaKVStore.bPut(bytes, bytes);
+ }
+
+ final byte[] firstKey = keys.get(keys.size() - 1);
+ final byte[] lastKey = keys.get(0);
+ final String firstKeyString = readUtf8(firstKey);
+ final String lastKeyString = readUtf8(lastKey);
+
+ // async scan
+ final CompletableFuture<List<KVEntry>> f1 = rheaKVStore.reverseScan(firstKey, lastKey);
+ final CompletableFuture<List<KVEntry>> f2 = rheaKVStore.reverseScan(firstKey, lastKey, false);
+ final CompletableFuture<List<KVEntry>> f3 = rheaKVStore.reverseScan(firstKeyString, lastKeyString);
+ final CompletableFuture<List<KVEntry>> f4 = rheaKVStore.reverseScan(firstKeyString, lastKeyString, false);
+ CompletableFuture.allOf(f1, f2, f3, f4).join();
+ for (final CompletableFuture<List<KVEntry>> f : new CompletableFuture[] { f1, f2, f3, f4 }) {
+ for (final KVEntry kv : f.join()) {
+ LOG.info("Async reverseScan: key={}, value={}", readUtf8(kv.getKey()), readUtf8(kv.getValue()));
+ }
+ }
+
+ // sync scan
+ final List<KVEntry> l1 = rheaKVStore.bReverseScan(firstKey, lastKey);
+ final List<KVEntry> l2 = rheaKVStore.bReverseScan(firstKey, lastKey, false);
+ final List<KVEntry> l3 = rheaKVStore.bReverseScan(firstKeyString, lastKeyString);
+ final List<KVEntry> l4 = rheaKVStore.bReverseScan(firstKeyString, lastKeyString, false);
+ for (final List<KVEntry> l : new List[] { l1, l2, l3, l4 }) {
+ for (final KVEntry kv : l) {
+ LOG.info("sync reverseScan: key={}, value={}", readUtf8(kv.getKey()), readUtf8(kv.getValue()));
+ }
+ }
+ }
+}

jraft-example/src/main/java/com/alipay/sofa/jraft/example/rheakv/ScanExample.java

@@ -77,7 +77,7 @@ public static void scan(final RheaKVStore rheaKVStore) {
  final List<KVEntry> l4 = rheaKVStore.bScan(firstKeyString, lastKeyString, false);
  for (final List<KVEntry> l : new List[] { l1, l2, l3, l4 }) {
  for (final KVEntry kv : l) {
- LOG.info("Async scan: key={}, value={}", readUtf8(kv.getKey()), readUtf8(kv.getValue()));
+ LOG.info("sync scan: key={}, value={}", readUtf8(kv.getKey()), readUtf8(kv.getValue()));
  }
  }
  }

jraft-rheakv/rheakv-core/src/main/java/com/alipay/sofa/jraft/rhea/DefaultRegionKVService.java

@@ -455,20 +455,26 @@ public void handleScanRequest(final ScanRequest request,
  response.setRegionEpoch(getRegionEpoch());
  try {
  KVParameterRequires.requireSameEpoch(request, getRegionEpoch());
- this.rawKVStore.scan(request.getStartKey(), request.getEndKey(), request.getLimit(),
- request.isReadOnlySafe(), request.isReturnValue(), new BaseKVStoreClosure() {
-
- @SuppressWarnings("unchecked")
- @Override
- public void run(final Status status) {
- if (status.isOk()) {
- response.setValue((List<KVEntry>) getData());
- } else {
- setFailure(request, response, status, getError());
- }
- closure.sendResponse(response);
+ BaseKVStoreClosure KVStoreClosure = new BaseKVStoreClosure() {
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public void run(final Status status) {
+ if (status.isOk()) {
+ response.setValue((List<KVEntry>) getData());
+ } else {
+ setFailure(request, response, status, getError());
  }
- });
+ closure.sendResponse(response);
+ }
+ };
+ if (request.isReverse()) {
+ this.rawKVStore.reverseScan(request.getStartKey(), request.getEndKey(), request.getLimit(),
+ request.isReadOnlySafe(), request.isReturnValue(), KVStoreClosure);
+ } else {
+ this.rawKVStore.scan(request.getStartKey(), request.getEndKey(), request.getLimit(),
+ request.isReadOnlySafe(), request.isReturnValue(), KVStoreClosure);
+ }
  } catch (final Throwable t) {
  LOG.error("Failed to handle: {}, {}.", request, StackTraceUtil.stackTrace(t));
  response.setError(Errors.forException(t));

jraft-rheakv/rheakv-core/src/main/java/com/alipay/sofa/jraft/rhea/client/DefaultRheaKVStore.java

@@ -16,6 +16,7 @@
  */
 package com.alipay.sofa.jraft.rhea.client;
 
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
@@ -610,8 +611,7 @@ private FutureGroup<List<KVEntry>> internalScan(final byte[] startKey, final byt
  final boolean readOnlySafe, final boolean returnValue,
  final int retriesLeft, final Throwable lastCause) {
  Requires.requireNonNull(startKey, "startKey");
- final List<Region> regionList = this.pdClient
- .findRegionsByKeyRange(startKey, endKey, ApiExceptionHelper.isInvalidEpoch(lastCause));
+ final List<Region> regionList = this.pdClient.findRegionsByKeyRange(startKey, endKey, ApiExceptionHelper.isInvalidEpoch(lastCause));
  final List<CompletableFuture<List<KVEntry>>> futures = Lists.newArrayListWithCapacity(regionList.size());
  final Errors lastError = lastCause == null ? null : Errors.forException(lastCause);
  for (final Region region : regionList) {
@@ -623,31 +623,40 @@ private FutureGroup<List<KVEntry>> internalScan(final byte[] startKey, final byt
  final ListRetryCallable<KVEntry> retryCallable = retryCause -> internalScan(subStartKey, subEndKey,
  readOnlySafe, returnValue, retriesLeft - 1, retryCause);
  final ListFailoverFuture<KVEntry> future = new ListFailoverFuture<>(retriesLeft, retryCallable);
- internalRegionScan(region, subStartKey, subEndKey, readOnlySafe, returnValue, future, retriesLeft, lastError,
+ internalRegionScan(region, subStartKey, subEndKey, false, readOnlySafe, returnValue, future, retriesLeft, lastError,
  this.onlyLeaderRead);
  futures.add(future);
  }
  return new FutureGroup<>(futures);
  }
 
  private void internalRegionScan(final Region region, final byte[] subStartKey, final byte[] subEndKey,
- final boolean readOnlySafe, final boolean returnValue,
+ final boolean reverse, final boolean readOnlySafe, final boolean returnValue,
  final CompletableFuture<List<KVEntry>> future, final int retriesLeft,
  final Errors lastCause, final boolean requireLeader) {
  final RegionEngine regionEngine = getRegionEngine(region.getId(), requireLeader);
  // require leader on retry
- final RetryRunner retryRunner = retryCause -> internalRegionScan(region, subStartKey, subEndKey, readOnlySafe,
+ final RetryRunner retryRunner = retryCause -> internalRegionScan(region, subStartKey, subEndKey, reverse, readOnlySafe,
  returnValue, future, retriesLeft - 1, retryCause, true);
  final FailoverClosure<List<KVEntry>> closure = new FailoverClosureImpl<>(future, false,
  retriesLeft, retryRunner);
  if (regionEngine != null) {
  if (ensureOnValidEpoch(region, regionEngine, closure)) {
  final RawKVStore rawKVStore = getRawKVStore(regionEngine);
- if (this.kvDispatcher == null) {
- rawKVStore.scan(subStartKey, subEndKey, readOnlySafe, returnValue, closure);
+ if (reverse) {
+ if (this.kvDispatcher == null) {
+ rawKVStore.reverseScan(subStartKey, subEndKey, readOnlySafe, returnValue, closure);
+ } else {
+ this.kvDispatcher.execute(
+ () -> rawKVStore.reverseScan(subStartKey, subEndKey, readOnlySafe, returnValue, closure));
+ }
  } else {
- this.kvDispatcher.execute(
- () -> rawKVStore.scan(subStartKey, subEndKey, readOnlySafe, returnValue, closure));
+ if (this.kvDispatcher == null) {
+ rawKVStore.scan(subStartKey, subEndKey, readOnlySafe, returnValue, closure);
+ } else {
+ this.kvDispatcher.execute(
+ () -> rawKVStore.scan(subStartKey, subEndKey, readOnlySafe, returnValue, closure));
+ }
  }
  }
  } else {
@@ -658,10 +667,107 @@ private void internalRegionScan(final Region region, final byte[] subStartKey, f
  request.setReturnValue(returnValue);
  request.setRegionId(region.getId());
  request.setRegionEpoch(region.getRegionEpoch());
+ request.setReverse(reverse);
  this.rheaKVRpcService.callAsyncWithRpc(request, closure, lastCause, requireLeader);
  }
  }
 
+ @Override
+ public CompletableFuture<List<KVEntry>> reverseScan(final byte[] startKey, final byte[] endKey) {
+ return reverseScan(startKey, endKey, true);
+ }
+
+ @Override
+ public CompletableFuture<List<KVEntry>> reverseScan(final String startKey, final String endKey) {
+ return reverseScan(BytesUtil.writeUtf8(startKey), BytesUtil.writeUtf8(endKey));
+ }
+
+ @Override
+ public CompletableFuture<List<KVEntry>> reverseScan(final byte[] startKey, final byte[] endKey,
+ final boolean readOnlySafe) {
+ return reverseScan(startKey, endKey, readOnlySafe, true);
+ }
+
+ @Override
+ public CompletableFuture<List<KVEntry>> reverseScan(final String startKey, final String endKey,
+ final boolean readOnlySafe) {
+ return reverseScan(BytesUtil.writeUtf8(startKey), BytesUtil.writeUtf8(endKey), readOnlySafe);
+ }
+
+ @Override
+ public CompletableFuture<List<KVEntry>> reverseScan(final byte[] startKey, final byte[] endKey,
+ final boolean readOnlySafe, final boolean returnValue) {
+ checkState();
+ final byte[] realEndKey = BytesUtil.nullToEmpty(endKey);
+ if (startKey != null) {
+ Requires.requireTrue(BytesUtil.compare(startKey, realEndKey) > 0, "startKey must > endKey");
+ }
+ final FutureGroup<List<KVEntry>> futureGroup = internalReverseScan(startKey, realEndKey, readOnlySafe,
+ returnValue, this.failoverRetries, null);
+ return FutureHelper.joinList(futureGroup);
+ }
+
+ @Override
+ public CompletableFuture<List<KVEntry>> reverseScan(final String startKey, final String endKey,
+ final boolean readOnlySafe, final boolean returnValue) {
+ return reverseScan(BytesUtil.writeUtf8(startKey), BytesUtil.writeUtf8(endKey), readOnlySafe, returnValue);
+ }
+
+ @Override
+ public List<KVEntry> bReverseScan(final byte[] startKey, final byte[] endKey) {
+ return FutureHelper.get(reverseScan(startKey, endKey), this.futureTimeoutMillis);
+ }
+
+ @Override
+ public List<KVEntry> bReverseScan(final String startKey, final String endKey) {
+ return FutureHelper.get(reverseScan(startKey, endKey), this.futureTimeoutMillis);
+ }
+
+ @Override
+ public List<KVEntry> bReverseScan(final byte[] startKey, final byte[] endKey, final boolean readOnlySafe) {
+ return FutureHelper.get(reverseScan(startKey, endKey, readOnlySafe), this.futureTimeoutMillis);
+ }
+
+ @Override
+ public List<KVEntry> bReverseScan(final String startKey, final String endKey, final boolean readOnlySafe) {
+ return FutureHelper.get(reverseScan(startKey, endKey, readOnlySafe), this.futureTimeoutMillis);
+ }
+
+ @Override
+ public List<KVEntry> bReverseScan(final byte[] startKey, final byte[] endKey, final boolean readOnlySafe,
+ final boolean returnValue) {
+ return FutureHelper.get(reverseScan(startKey, endKey, readOnlySafe, returnValue), this.futureTimeoutMillis);
+ }
+
+ @Override
+ public List<KVEntry> bReverseScan(final String startKey, final String endKey, final boolean readOnlySafe,
+ final boolean returnValue) {
+ return FutureHelper.get(reverseScan(startKey, endKey, readOnlySafe, returnValue), this.futureTimeoutMillis);
+ }
+
+ private FutureGroup<List<KVEntry>> internalReverseScan(final byte[] startKey, final byte[] endKey,
+ final boolean readOnlySafe, final boolean returnValue,
+ final int retriesLeft, final Throwable lastCause) {
+ Requires.requireNonNull(endKey, "endKey");
+ final List<Region> regionList = this.pdClient.findRegionsByKeyRange(endKey, startKey, ApiExceptionHelper.isInvalidEpoch(lastCause));
+ Collections.reverse(regionList);
+ final List<CompletableFuture<List<KVEntry>>> futures = Lists.newArrayListWithCapacity(regionList.size());
+ final Errors lastError = lastCause == null ? null : Errors.forException(lastCause);
+ for (final Region region : regionList) {
+ final byte[] regionEndKey = region.getEndKey();
+ final byte[] regionStartKey = region.getStartKey();
+ final byte[] subStartKey = regionEndKey == null ? startKey : (startKey == null ? regionEndKey : BytesUtil.min(regionEndKey, startKey));
+ final byte[] subEndKey = regionStartKey == null ? endKey : BytesUtil.max(regionStartKey, endKey);
+ final ListRetryCallable<KVEntry> retryCallable = retryCause -> internalReverseScan(subStartKey, subEndKey,
+ readOnlySafe, returnValue, retriesLeft - 1, retryCause);
+ final ListFailoverFuture<KVEntry> future = new ListFailoverFuture<>(retriesLeft, retryCallable);
+ internalRegionScan(region, subStartKey, subEndKey, true, readOnlySafe, returnValue, future, retriesLeft, lastError,
+ this.onlyLeaderRead );
+ futures.add(future);
+ }
+ return new FutureGroup<>(futures);
+ }
+
  public List<KVEntry> singleRegionScan(final byte[] startKey, final byte[] endKey, final int limit,
  final boolean readOnlySafe, final boolean returnValue) {
  checkState();

jraft-rheakv/rheakv-core/src/main/java/com/alipay/sofa/jraft/rhea/client/RheaKVStore.java

@@ -254,6 +254,86 @@ List<KVEntry> bScan(final byte[] startKey, final byte[] endKey, final boolean re
  List<KVEntry> bScan(final String startKey, final String endKey, final boolean readOnlySafe,
  final boolean returnValue);
 
+ /**
+ * Equivalent to {@code reverseScan(startKey, endKey, true)}.
+ */
+ CompletableFuture<List<KVEntry>> reverseScan(final byte[] startKey, final byte[] endKey);
+
+ /**
+ * @see #reverseScan(byte[], byte[])
+ */
+ CompletableFuture<List<KVEntry>> reverseScan(final String startKey, final String endKey);
+
+ /**
+ * Equivalent to {@code reverseScan(startKey, endKey, readOnlySafe, true)}.
+ */
+ CompletableFuture<List<KVEntry>> reverseScan(final byte[] startKey, final byte[] endKey, final boolean readOnlySafe);
+
+ /**
+ * @see #reverseScan(byte[], byte[], boolean)
+ */
+ CompletableFuture<List<KVEntry>> reverseScan(final String startKey, final String endKey, final boolean readOnlySafe);
+
+ /**
+ * Reverse query all data in the key of range [startKey, endKey).
+ * <p>
+ * Provide consistent reading if {@code readOnlySafe} is true.
+ *
+ * Reverse scanning is usually much worse than forward scanning.
+ *
+ * Reverse scanning across multi regions maybe slower and devastating.
+ *
+ * @param startKey first key to reverse scan within database (included),
+ * null means 'max-key' in the database.
+ * @param endKey last key to reverse scan within database (excluded).
+ * null means 'min-key' in the database.
+ * @param readOnlySafe provide consistent reading if {@code readOnlySafe}
+ * is true.
+ * @param returnValue whether to return value.
+ * @return a list where the key of range [startKey, endKey) passed by user
+ * and value for {@code KVEntry}
+ */
+ CompletableFuture<List<KVEntry>> reverseScan(final byte[] startKey, final byte[] endKey,
+ final boolean readOnlySafe, final boolean returnValue);
+
+ /**
+ * @see #reverseScan(byte[], byte[], boolean, boolean)
+ */
+ CompletableFuture<List<KVEntry>> reverseScan(final String startKey, final String endKey,
+ final boolean readOnlySafe, final boolean returnValue);
+
+ /**
+ * @see #reverseScan(byte[], byte[])
+ */
+ List<KVEntry> bReverseScan(final byte[] startKey, final byte[] endKey);
+
+ /**
+ * @see #scan(String, String)
+ */
+ List<KVEntry> bReverseScan(final String startKey, final String endKey);
+
+ /**
+ * @see #scan(String, String, boolean)
+ */
+ List<KVEntry> bReverseScan(final byte[] startKey, final byte[] endKey, final boolean readOnlySafe);
+
+ /**
+ * @see #scan(String, String, boolean)
+ */
+ List<KVEntry> bReverseScan(final String startKey, final String endKey, final boolean readOnlySafe);
+
+ /**
+ * @see #reverseScan(String, String, boolean, boolean)
+ */
+ List<KVEntry> bReverseScan(final byte[] startKey, final byte[] endKey, final boolean readOnlySafe,
+ final boolean returnValue);
+
+ /**
+ * @see #reverseScan(String, String, boolean, boolean)
+ */
+ List<KVEntry> bReverseScan(final String startKey, final String endKey, final boolean readOnlySafe,
+ final boolean returnValue);
+
  /**
  * Equivalent to {@code iterator(startKey, endKey, bufSize, true)}.
  */