diff --git a/dingo-raft-kv/src/main/java/io/dingodb/raft/kv/storage/ReadIndexRunner.java b/dingo-raft-kv/src/main/java/io/dingodb/raft/kv/storage/ReadIndexRunner.java index 0ed2b1de37..8e016fac23 100644 --- a/dingo-raft-kv/src/main/java/io/dingodb/raft/kv/storage/ReadIndexRunner.java +++ b/dingo-raft-kv/src/main/java/io/dingodb/raft/kv/storage/ReadIndexRunner.java @@ -55,20 +55,20 @@ private class ReadIndexClosure extends io.dingodb.raft.closure.ReadIndexClosu @Override public void run(Status status, long index, byte[] reqCtx) { if (status.isOk()) { - future.complete((T) executeFunc.apply(operation)); + Executors.execute("read-index-exec", () -> future.complete((T) executeFunc.apply(operation))); return; } executor.execute(() -> { if (node.isLeader()) { log.warn("Fail to [get] with 'ReadIndex': {}, try to applying to the state machine.", status); // If 'read index' read fails, try to applying to the state machine at the leader node - RaftRawKVOperation.sync().applyOnNode(node).whenComplete((r, e) -> { + RaftRawKVOperation.sync().applyOnNode(node).whenCompleteAsync((r, e) -> { if (e == null) { future.complete((T) executeFunc.apply(operation)); } else { future.completeExceptionally(e); } - }); + }, Executors.executor("read-index-exec")); } else { log.warn("Fail to [get] with 'ReadIndex': {}.", status); // Client will retry to leader node diff --git a/dingo-raft-kv/src/main/java/io/dingodb/raft/kv/storage/RocksRawKVStore.java b/dingo-raft-kv/src/main/java/io/dingodb/raft/kv/storage/RocksRawKVStore.java index dcb06aaa17..fd5c89a65a 100644 --- a/dingo-raft-kv/src/main/java/io/dingodb/raft/kv/storage/RocksRawKVStore.java +++ b/dingo-raft-kv/src/main/java/io/dingodb/raft/kv/storage/RocksRawKVStore.java @@ -40,6 +40,7 @@ import java.nio.file.Path; import java.nio.file.Paths; import java.util.ArrayList; +import java.util.Arrays; import java.util.Iterator; import java.util.List; import java.util.concurrent.CompletableFuture; @@ -170,7 +171,7 @@ public boolean delete(List keys) { public boolean delete(byte[] startKey, byte[] endKey) { try { if (endKey == null) { - try (RocksIterator iterator = this.db.newIterator()) { + try (RocksIterator iterator = this.db.newIterator();) { iterator.seekToLast(); if (iterator.isValid()) { try (final WriteBatch batch = new WriteBatch()) { @@ -178,6 +179,8 @@ public boolean delete(byte[] startKey, byte[] endKey) { batch.delete(endKey); batch.deleteRange(startKey, endKey); this.db.write(this.writeOptions, batch); + this.db.deleteFilesInRanges(this.db.getDefaultColumnFamily(), Arrays.asList(startKey, endKey), true); + this.db.compactRange(startKey, endKey); return true; } } else { @@ -186,7 +189,9 @@ public boolean delete(byte[] startKey, byte[] endKey) { } } } else { + this.db.deleteFilesInRanges(this.db.getDefaultColumnFamily(), Arrays.asList(startKey, endKey), true); this.db.deleteRange(this.writeOptions, startKey, endKey); + this.db.compactRange(startKey, endKey); return true; } } catch (final Exception e) { diff --git a/dingo-store-raft/src/main/java/io/dingodb/store/raft/PartStateMachine.java b/dingo-store-raft/src/main/java/io/dingodb/store/raft/PartStateMachine.java index 61fad330ea..48c4093149 100644 --- a/dingo-store-raft/src/main/java/io/dingodb/store/raft/PartStateMachine.java +++ b/dingo-store-raft/src/main/java/io/dingodb/store/raft/PartStateMachine.java @@ -230,9 +230,9 @@ private void sendStats() { log.warn("Report stats but current node not leader."); execute("cancel-report-stats", () -> scheduledFuture.cancel(true)); } - SeekableIterator iterator = store.scan(part.getStart(), part.getEnd()).join(); List approximateStats = new ArrayList<>(); if (collectStats) { + SeekableIterator iterator = store.scan(part.getStart(), part.getEnd()).join(); long count = 0; long size = 0; byte[] startKey = null;