andreacavalli/CavalliumDBEngine
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Allow reverse iteration

Browse Source
This commit is contained in:
Andrea Cavalli 2022-03-24 21:14:17 +01:00
parent 8e88c78ce7
commit 388b79c6d1
15 changed files with 239 additions and 156 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
src/main/java/it/cavallium/dbengine/database/LLDictionary.java
View File

@ -20,11 +20,7 @@ public interface LLDictionary extends LLKeyValueDatabaseStructure {
BufferAllocator getAllocator();
Mono<Send<Buffer>> get(@Nullable LLSnapshot snapshot, Mono<Send<Buffer>> key, boolean existsAlmostCertainly);
default Mono<Send<Buffer>> get(@Nullable LLSnapshot snapshot, Mono<Send<Buffer>> key) {
return get(snapshot, key, false);
}
Mono<Send<Buffer>> get(@Nullable LLSnapshot snapshot, Mono<Send<Buffer>> key);
Mono<Send<Buffer>> put(Mono<Send<Buffer>> key, Mono<Send<Buffer>> value, LLDictionaryResultType resultType);
@ -45,38 +41,18 @@ public interface LLDictionary extends LLKeyValueDatabaseStructure {
Mono<Send<Buffer>> remove(Mono<Send<Buffer>> key, LLDictionaryResultType resultType);
Flux<Optional<Buffer>> getMulti(@Nullable LLSnapshot snapshot,
Flux<Send<Buffer>> keys,
boolean existsAlmostCertainly);
default Flux<Optional<Buffer>> getMulti(@Nullable LLSnapshot snapshot,
Flux<Send<Buffer>> keys) {
return getMulti(snapshot, keys, false);
}
Flux<Optional<Buffer>> getMulti(@Nullable LLSnapshot snapshot, Flux<Send<Buffer>> keys);
Mono<Void> putMulti(Flux<Send<LLEntry>> entries);
<K> Flux<Boolean> updateMulti(Flux<K> keys, Flux<Send<Buffer>> serializedKeys,
KVSerializationFunction<K, @Nullable Send<Buffer>, @Nullable Buffer> updateFunction);
Flux<Send<LLEntry>> getRange(@Nullable LLSnapshot snapshot, Mono<Send<LLRange>> range, boolean existsAlmostCertainly);
Flux<Send<LLEntry>> getRange(@Nullable LLSnapshot snapshot, Mono<Send<LLRange>> range, boolean reverse);
default Flux<Send<LLEntry>> getRange(@Nullable LLSnapshot snapshot, Mono<Send<LLRange>> range) {
return getRange(snapshot, range, false);
}
Flux<List<Send<LLEntry>>> getRangeGrouped(@Nullable LLSnapshot snapshot, Mono<Send<LLRange>> range, int prefixLength);
Flux<List<Send<LLEntry>>> getRangeGrouped(@Nullable LLSnapshot snapshot,
Mono<Send<LLRange>> range,
int prefixLength,
boolean existsAlmostCertainly);
default Flux<List<Send<LLEntry>>> getRangeGrouped(@Nullable LLSnapshot snapshot,
Mono<Send<LLRange>> range,
int prefixLength) {
return getRangeGrouped(snapshot, range, prefixLength, false);
}
Flux<Send<Buffer>> getRangeKeys(@Nullable LLSnapshot snapshot, Mono<Send<LLRange>> range);
Flux<Send<Buffer>> getRangeKeys(@Nullable LLSnapshot snapshot, Mono<Send<LLRange>> range, boolean reverse);
Flux<List<Send<Buffer>>> getRangeKeysGrouped(@Nullable LLSnapshot snapshot, Mono<Send<LLRange>> range, int prefixLength);
@ -94,11 +70,11 @@ public interface LLDictionary extends LLKeyValueDatabaseStructure {
if (canKeysChange) {
return this
.setRange(range, this
.getRange(null, range, existsAlmostCertainly)
.getRange(null, range, false)
.flatMap(entriesReplacer)
);
} else {
return this.putMulti(this.getRange(null, range, existsAlmostCertainly).flatMap(entriesReplacer));
return this.putMulti(this.getRange(null, range, false).flatMap(entriesReplacer));
}
});
}

8
src/main/java/it/cavallium/dbengine/database/LLRange.java
View File

@ -13,7 +13,7 @@ import org.apache.logging.log4j.Logger;
import org.jetbrains.annotations.Nullable;
/**
* Range of data, from min (inclusive),to max (exclusive)
* Range of data, from min (inclusive), to max (exclusive)
*/
public class LLRange extends ResourceSupport<LLRange, LLRange> {
@ -137,8 +137,10 @@ public class LLRange extends ResourceSupport<LLRange, LLRange> {
public Send<Buffer> getMin() {
ensureOwned();
if (min != null) {
// todo: use a read-only copy
return min.copy().send();
} else if (single != null) {
// todo: use a read-only copy
return single.copy().send();
} else {
return null;
@ -164,8 +166,10 @@ public class LLRange extends ResourceSupport<LLRange, LLRange> {
public Send<Buffer> getMax() {
ensureOwned();
if (max != null) {
// todo: use a read-only copy
return max.copy().send();
} else if (single != null) {
// todo: use a read-only copy
return single.copy().send();
} else {
return null;
@ -186,6 +190,7 @@ public class LLRange extends ResourceSupport<LLRange, LLRange> {
public Send<Buffer> getSingle() {
ensureOwned();
assert isSingle();
// todo: use a read-only copy
return single != null ? single.copy().send() : null;
}
@ -235,6 +240,7 @@ public class LLRange extends ResourceSupport<LLRange, LLRange> {
public LLRange copy() {
ensureOwned();
// todo: use a read-only copy
return new LLRange(min != null ? min.copy().send() : null,
max != null ? max.copy().send() : null,
single != null ? single.copy().send(): null

109
src/main/java/it/cavallium/dbengine/database/collections/DatabaseMapDictionary.java
View File

@ -72,18 +72,43 @@ public class DatabaseMapDictionary<T, U> extends DatabaseMapDictionaryDeep<T, U,
public static <K, V> Flux<Entry<K, V>> getLeavesFrom(DatabaseMapDictionary<K, V> databaseMapDictionary,
CompositeSnapshot snapshot,
Mono<K> prevMono) {
Mono<Optional<K>> prevOptMono = prevMono.map(Optional::of).defaultIfEmpty(Optional.empty());
Mono<K> key,
boolean reverse) {
Mono<Optional<K>> keyOptMono = key.map(Optional::of).defaultIfEmpty(Optional.empty());
return prevOptMono.flatMapMany(prevOpt -> {
if (prevOpt.isPresent()) {
return databaseMapDictionary.getAllValues(snapshot, prevOpt.get());
return keyOptMono.flatMapMany(keyOpt -> {
if (keyOpt.isPresent()) {
return databaseMapDictionary.getAllValues(snapshot, keyOpt.get(), reverse);
} else {
return databaseMapDictionary.getAllValues(snapshot);
}
});
}
public static <K> Flux<K> getKeyLeavesFrom(DatabaseMapDictionary<K, ?> databaseMapDictionary,
CompositeSnapshot snapshot,
Mono<K> key,
boolean reverse) {
Mono<Optional<K>> keyOptMono = key.map(Optional::of).defaultIfEmpty(Optional.empty());
return keyOptMono.flatMapMany(keyOpt -> {
Flux<? extends Entry<K, ? extends DatabaseStageEntry<?>>> stagesFlux;
if (keyOpt.isPresent()) {
stagesFlux = databaseMapDictionary
.getAllStages(snapshot, keyOpt.get(), reverse);
} else {
stagesFlux = databaseMapDictionary.getAllStages(snapshot);
}
return stagesFlux.doOnNext(e -> e.getValue().close())
.doOnDiscard(Entry.class, e -> {
if (e.getValue() instanceof DatabaseStageEntry<?> resource) {
resource.close();
}
})
.map(Entry::getKey);
});
}
private void deserializeValue(T keySuffix, Send<Buffer> valueToReceive, SynchronousSink<U> sink) {
try (var value = valueToReceive.receive()) {
try {
@ -170,7 +195,7 @@ public class DatabaseMapDictionary<T, U> extends DatabaseMapDictionaryDeep<T, U,
@Override
public Mono<Object2ObjectSortedMap<T, U>> get(@Nullable CompositeSnapshot snapshot, boolean existsAlmostCertainly) {
return dictionary
.getRange(resolveSnapshot(snapshot), rangeMono, existsAlmostCertainly)
.getRange(resolveSnapshot(snapshot), rangeMono, false)
.<Entry<T, U>>handle((entrySend, sink) -> {
Entry<T, U> deserializedEntry;
try {
@ -244,8 +269,7 @@ public class DatabaseMapDictionary<T, U> extends DatabaseMapDictionaryDeep<T, U,
public Mono<U> getValue(@Nullable CompositeSnapshot snapshot, T keySuffix, boolean existsAlmostCertainly) {
return dictionary
.get(resolveSnapshot(snapshot),
Mono.fromCallable(() -> serializeKeySuffixToKey(keySuffix).send()),
existsAlmostCertainly
Mono.fromCallable(() -> serializeKeySuffixToKey(keySuffix).send())
)
.handle((valueToReceive, sink) -> deserializeValue(keySuffix, valueToReceive, sink));
}
@ -389,7 +413,7 @@ public class DatabaseMapDictionary<T, U> extends DatabaseMapDictionaryDeep<T, U,
}
});
return dictionary
.getMulti(resolveSnapshot(snapshot), mappedKeys, existsAlmostCertainly)
.getMulti(resolveSnapshot(snapshot), mappedKeys)
.<Optional<U>>handle((valueBufOpt, sink) -> {
try {
Optional<U> valueOpt;
@ -457,8 +481,44 @@ public class DatabaseMapDictionary<T, U> extends DatabaseMapDictionaryDeep<T, U,
@Override
public Flux<Entry<T, DatabaseStageEntry<U>>> getAllStages(@Nullable CompositeSnapshot snapshot) {
return getAllStages(snapshot, rangeMono, false);
}
/**
* Get all stages
* @param key from/to the specified key, if not null
* @param reverse if true, the results will go backwards from the specified key (inclusive)
* if false, the results will go forward from the specified key (inclusive)
*/
public Flux<Entry<T, DatabaseStageEntry<U>>> getAllStages(@Nullable CompositeSnapshot snapshot,
@Nullable T key,
boolean reverse) {
if (key == null) {
return getAllStages(snapshot);
} else {
Mono<Send<LLRange>> boundedRangeMono = rangeMono.flatMap(fullRangeSend -> Mono.fromCallable(() -> {
try (var fullRange = fullRangeSend.receive()) {
try (var keyWithoutExtBuf = keyPrefix == null ? alloc.allocate(keySuffixLength + keyExtLength)
// todo: use a read-only copy
: keyPrefix.copy()) {
keyWithoutExtBuf.ensureWritable(keySuffixLength + keyExtLength);
serializeSuffix(key, keyWithoutExtBuf);
if (reverse) {
return LLRange.of(fullRange.getMin(), keyWithoutExtBuf.send()).send();
} else {
return LLRange.of(keyWithoutExtBuf.send(), fullRange.getMax()).send();
}
}
}
}));
return getAllStages(snapshot, boundedRangeMono, reverse);
}
}
private Flux<Entry<T, DatabaseStageEntry<U>>> getAllStages(@Nullable CompositeSnapshot snapshot,
Mono<Send<LLRange>> sliceRangeMono, boolean reverse) {
return dictionary
.getRangeKeys(resolveSnapshot(snapshot), rangeMono)
.getRangeKeys(resolveSnapshot(snapshot), sliceRangeMono, reverse)
.handle((keyBufToReceive, sink) -> {
var keyBuf = keyBufToReceive.receive();
try {
@ -481,30 +541,43 @@ public class DatabaseMapDictionary<T, U> extends DatabaseMapDictionaryDeep<T, U,
@Override
public Flux<Entry<T, U>> getAllValues(@Nullable CompositeSnapshot snapshot) {
return getAllValues(snapshot, rangeMono);
return getAllValues(snapshot, rangeMono, false);
}
public Flux<Entry<T, U>> getAllValues(@Nullable CompositeSnapshot snapshot, @Nullable T from) {
if (from == null) {
/**
* Get all values
* @param key from/to the specified key, if not null
* @param reverse if true, the results will go backwards from the specified key (inclusive)
* if false, the results will go forward from the specified key (inclusive)
*/
public Flux<Entry<T, U>> getAllValues(@Nullable CompositeSnapshot snapshot, @Nullable T key, boolean reverse) {
if (key == null) {
return getAllValues(snapshot);
} else {
Mono<Send<LLRange>> boundedRangeMono = rangeMono.flatMap(fullRangeSend -> Mono.fromCallable(() -> {
try (var fullRange = fullRangeSend.receive()) {
try (var keyWithoutExtBuf = keyPrefix == null ? alloc.allocate(keySuffixLength + keyExtLength)
// todo: use a read-only copy
: keyPrefix.copy()) {
keyWithoutExtBuf.ensureWritable(keySuffixLength + keyExtLength);
serializeSuffix(from, keyWithoutExtBuf);
return LLRange.of(keyWithoutExtBuf.send(), fullRange.getMax()).send();
serializeSuffix(key, keyWithoutExtBuf);
if (reverse) {
return LLRange.of(fullRange.getMin(), keyWithoutExtBuf.send()).send();
} else {
return LLRange.of(keyWithoutExtBuf.send(), fullRange.getMax()).send();
}
}
}
}));
return getAllValues(snapshot, boundedRangeMono);
return getAllValues(snapshot, boundedRangeMono, reverse);
}
}
private Flux<Entry<T, U>> getAllValues(@Nullable CompositeSnapshot snapshot, Mono<Send<LLRange>> sliceRangeMono) {
private Flux<Entry<T, U>> getAllValues(@Nullable CompositeSnapshot snapshot,
Mono<Send<LLRange>> sliceRangeMono,
boolean reverse) {
return dictionary
.getRange(resolveSnapshot(snapshot), sliceRangeMono)
.getRange(resolveSnapshot(snapshot), sliceRangeMono, reverse)
.<Entry<T, U>>handle((serializedEntryToReceive, sink) -> {
try {
Entry<T, U> entry;

2
src/main/java/it/cavallium/dbengine/database/collections/DatabaseMapDictionaryDeep.java
View File

@ -551,7 +551,7 @@ public class DatabaseMapDictionaryDeep<T, U, US extends DatabaseStage<U>> extend
sink.next(fullRange.send());
}
}
}))
}), false)
.concatMapIterable(entrySend -> {
K1 key1 = null;
Object key2 = null;

2
src/main/java/it/cavallium/dbengine/database/collections/DatabaseMapSingle.java
View File

@ -114,7 +114,7 @@ public class DatabaseMapSingle<U> extends ResourceSupport<DatabaseStage<U>, Data
@Override
public Mono<U> get(@Nullable CompositeSnapshot snapshot, boolean existsAlmostCertainly) {
return dictionary
.get(resolveSnapshot(snapshot), keyMono, existsAlmostCertainly)
.get(resolveSnapshot(snapshot), keyMono)
.handle(this::deserializeValue);
}

118
src/main/java/it/cavallium/dbengine/database/disk/LLLocalDictionary.java
View File

@ -31,7 +31,6 @@ import it.cavallium.dbengine.database.serialization.SerializationFunction;
import it.cavallium.dbengine.rpc.current.data.DatabaseOptions;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@ -80,7 +79,7 @@ public class LLLocalDictionary implements LLDictionary {
static final ReadOptions EMPTY_READ_OPTIONS = new UnreleasableReadOptions(new UnmodifiableReadOptions());
static final WriteOptions EMPTY_WRITE_OPTIONS = new UnreleasableWriteOptions(new UnmodifiableWriteOptions());
static final WriteOptions BATCH_WRITE_OPTIONS = new UnreleasableWriteOptions(new UnmodifiableWriteOptions());
static final boolean PREFER_SEEK_TO_FIRST = false;
static final boolean PREFER_AUTO_SEEK_BOUND = false;
/**
* It used to be false,
* now it's true to avoid crashes during iterations on completely corrupted files
@ -251,9 +250,7 @@ public class LLLocalDictionary implements LLDictionary {
}
@Override
public Mono<Send<Buffer>> get(@Nullable LLSnapshot snapshot,
Mono<Send<Buffer>> keyMono,
boolean existsAlmostCertainly) {
public Mono<Send<Buffer>> get(@Nullable LLSnapshot snapshot, Mono<Send<Buffer>> keyMono) {
return keyMono
.publishOn(dbScheduler)
.<Send<Buffer>>handle((keySend, sink) -> {
@ -322,7 +319,7 @@ public class LLLocalDictionary implements LLDictionary {
}
}
try (RocksIterator rocksIterator = db.newIterator(readOpts)) {
if (!LLLocalDictionary.PREFER_SEEK_TO_FIRST && range.hasMin()) {
if (!LLLocalDictionary.PREFER_AUTO_SEEK_BOUND && range.hasMin()) {
if (nettyDirect && isReadOnlyDirect(range.getMinUnsafe())) {
rocksIterator.seek(((ReadableComponent) range.getMinUnsafe()).readableBuffer());
} else {
@ -382,7 +379,7 @@ public class LLLocalDictionary implements LLDictionary {
// Zip the entry to write to the database
var entryMono = Mono.zip(keyMono, valueMono, Map::entry);
// Obtain the previous value from the database
var previousDataMono = this.getPreviousData(keyMono, resultType, false);
var previousDataMono = this.getPreviousData(keyMono, resultType);
// Write the new entry to the database
Mono<Send<Buffer>> putMono = entryMono
.publishOn(dbScheduler)
@ -511,7 +508,7 @@ public class LLLocalDictionary implements LLDictionary {
@Override
public Mono<Send<Buffer>> remove(Mono<Send<Buffer>> keyMono, LLDictionaryResultType resultType) {
// Obtain the previous value from the database
Mono<Send<Buffer>> previousDataMono = this.getPreviousData(keyMono, resultType, true);
Mono<Send<Buffer>> previousDataMono = this.getPreviousData(keyMono, resultType);
// Delete the value from the database
Mono<Send<Buffer>> removeMono = keyMono
.publishOn(dbScheduler)
@ -538,8 +535,7 @@ public class LLLocalDictionary implements LLDictionary {
return Flux.concat(previousDataMono, removeMono).singleOrEmpty();
}
private Mono<Send<Buffer>> getPreviousData(Mono<Send<Buffer>> keyMono, LLDictionaryResultType resultType,
boolean existsAlmostCertainly) {
private Mono<Send<Buffer>> getPreviousData(Mono<Send<Buffer>> keyMono, LLDictionaryResultType resultType) {
return switch (resultType) {
case PREVIOUS_VALUE_EXISTENCE -> keyMono
.publishOn(dbScheduler)
@ -572,9 +568,7 @@ public class LLLocalDictionary implements LLDictionary {
}
@Override
public Flux<Optional<Buffer>> getMulti(@Nullable LLSnapshot snapshot,
Flux<Send<Buffer>> keys,
boolean existsAlmostCertainly) {
public Flux<Optional<Buffer>> getMulti(@Nullable LLSnapshot snapshot, Flux<Send<Buffer>> keys) {
return keys
.buffer(MULTI_GET_WINDOW)
.publishOn(dbScheduler)
@ -774,16 +768,14 @@ public class LLLocalDictionary implements LLDictionary {
}
@Override
public Flux<Send<LLEntry>> getRange(@Nullable LLSnapshot snapshot,
Mono<Send<LLRange>> rangeMono,
boolean existsAlmostCertainly) {
public Flux<Send<LLEntry>> getRange(@Nullable LLSnapshot snapshot, Mono<Send<LLRange>> rangeMono, boolean reverse) {
return rangeMono.flatMapMany(rangeSend -> {
try (var range = rangeSend.receive()) {
if (range.isSingle()) {
var rangeSingleMono = rangeMono.map(r -> r.receive().getSingle());
return getRangeSingle(snapshot, rangeSingleMono, existsAlmostCertainly);
return getRangeSingle(snapshot, rangeSingleMono);
} else {
return getRangeMulti(snapshot, rangeMono);
return getRangeMulti(snapshot, rangeMono, reverse);
}
}
});
@ -792,13 +784,12 @@ public class LLLocalDictionary implements LLDictionary {
@Override
public Flux<List<Send<LLEntry>>> getRangeGrouped(@Nullable LLSnapshot snapshot,
Mono<Send<LLRange>> rangeMono,
int prefixLength,
boolean existsAlmostCertainly) {
int prefixLength) {
return rangeMono.flatMapMany(rangeSend -> {
try (var range = rangeSend.receive()) {
if (range.isSingle()) {
var rangeSingleMono = rangeMono.map(r -> r.receive().getSingle());
return getRangeSingle(snapshot, rangeSingleMono, existsAlmostCertainly).map(List::of);
return getRangeSingle(snapshot, rangeSingleMono).map(List::of);
} else {
return getRangeMultiGrouped(snapshot, rangeMono, prefixLength);
}
@ -806,19 +797,17 @@ public class LLLocalDictionary implements LLDictionary {
});
}
private Flux<Send<LLEntry>> getRangeSingle(LLSnapshot snapshot,
Mono<Send<Buffer>> keyMono,
boolean existsAlmostCertainly) {
private Flux<Send<LLEntry>> getRangeSingle(LLSnapshot snapshot, Mono<Send<Buffer>> keyMono) {
return Mono
.zip(keyMono, this.get(snapshot, keyMono, existsAlmostCertainly))
.zip(keyMono, this.get(snapshot, keyMono))
.map(result -> LLEntry.of(result.getT1(), result.getT2()).send())
.flux();
}
private Flux<Send<LLEntry>> getRangeMulti(LLSnapshot snapshot, Mono<Send<LLRange>> rangeMono) {
private Flux<Send<LLEntry>> getRangeMulti(LLSnapshot snapshot, Mono<Send<LLRange>> rangeMono, boolean reverse) {
Mono<LLLocalEntryReactiveRocksIterator> iteratorMono = rangeMono.map(rangeSend -> {
ReadOptions resolvedSnapshot = resolveSnapshot(snapshot);
return new LLLocalEntryReactiveRocksIterator(db, rangeSend, nettyDirect, resolvedSnapshot);
return new LLLocalEntryReactiveRocksIterator(db, rangeSend, nettyDirect, resolvedSnapshot, reverse);
});
return Flux.usingWhen(iteratorMono,
iterator -> iterator.flux().subscribeOn(dbScheduler, false),
@ -840,13 +829,13 @@ public class LLLocalDictionary implements LLDictionary {
}
@Override
public Flux<Send<Buffer>> getRangeKeys(@Nullable LLSnapshot snapshot, Mono<Send<LLRange>> rangeMono) {
public Flux<Send<Buffer>> getRangeKeys(@Nullable LLSnapshot snapshot, Mono<Send<LLRange>> rangeMono, boolean reverse) {
return rangeMono.flatMapMany(rangeSend -> {
try (var range = rangeSend.receive()) {
if (range.isSingle()) {
return this.getRangeKeysSingle(snapshot, rangeMono.map(r -> r.receive().getSingle()));
} else {
return this.getRangeKeysMulti(snapshot, rangeMono);
return this.getRangeKeysMulti(snapshot, rangeMono, reverse);
}
}
});
@ -881,7 +870,7 @@ public class LLLocalDictionary implements LLDictionary {
}
}
ro.setVerifyChecksums(true);
try (var rocksIteratorTuple = getRocksIterator(nettyDirect, ro, range, db)) {
try (var rocksIteratorTuple = getRocksIterator(nettyDirect, ro, range, db, false)) {
var rocksIterator = rocksIteratorTuple.iterator();
rocksIterator.seekToFirst();
rocksIterator.status();
@ -938,10 +927,10 @@ public class LLLocalDictionary implements LLDictionary {
.flux();
}
private Flux<Send<Buffer>> getRangeKeysMulti(LLSnapshot snapshot, Mono<Send<LLRange>> rangeMono) {
private Flux<Send<Buffer>> getRangeKeysMulti(LLSnapshot snapshot, Mono<Send<LLRange>> rangeMono, boolean reverse) {
Mono<LLLocalKeyReactiveRocksIterator> iteratorMono = rangeMono.map(range -> {
ReadOptions resolvedSnapshot = resolveSnapshot(snapshot);
return new LLLocalKeyReactiveRocksIterator(db, range, nettyDirect, resolvedSnapshot);
return new LLLocalKeyReactiveRocksIterator(db, range, nettyDirect, resolvedSnapshot, reverse);
});
return Flux.usingWhen(iteratorMono,
iterator -> iterator.flux().subscribeOn(dbScheduler, false),
@ -977,7 +966,7 @@ public class LLLocalDictionary implements LLDictionary {
assert opts.isOwningHandle();
SafeCloseable seekTo;
try (RocksIterator it = db.newIterator(opts)) {
if (!PREFER_SEEK_TO_FIRST && range.hasMin()) {
if (!PREFER_AUTO_SEEK_BOUND && range.hasMin()) {
seekTo = rocksIterSeekTo(nettyDirect, it, range.getMinUnsafe());
} else {
seekTo = null;
@ -1149,7 +1138,7 @@ public class LLLocalDictionary implements LLDictionary {
}
try (var rocksIterator = db.newIterator(readOpts)) {
SafeCloseable seekTo;
if (!LLLocalDictionary.PREFER_SEEK_TO_FIRST && range.hasMin()) {
if (!LLLocalDictionary.PREFER_AUTO_SEEK_BOUND && range.hasMin()) {
seekTo = rocksIterSeekTo(nettyDirect, rocksIterator, range.getMinUnsafe());
} else {
seekTo = null;
@ -1199,7 +1188,7 @@ public class LLLocalDictionary implements LLDictionary {
}
try (var rocksIterator = db.newIterator(readOpts)) {
SafeCloseable seekTo;
if (!LLLocalDictionary.PREFER_SEEK_TO_FIRST && range.hasMin()) {
if (!LLLocalDictionary.PREFER_AUTO_SEEK_BOUND && range.hasMin()) {
seekTo = rocksIterSeekTo(nettyDirect, rocksIterator, range.getMinUnsafe());
} else {
seekTo = null;
@ -1227,6 +1216,27 @@ public class LLLocalDictionary implements LLDictionary {
}
}
/**
* Useful for reverse iterations
*/
@Nullable
private static SafeCloseable rocksIterSeekFrom(boolean allowNettyDirect,
RocksIterator rocksIterator, Buffer key) {
if (allowNettyDirect && isReadOnlyDirect(key)) {
ByteBuffer keyInternalByteBuffer = ((ReadableComponent) key).readableBuffer();
assert keyInternalByteBuffer.position() == 0;
rocksIterator.seekForPrev(keyInternalByteBuffer);
// This is useful to retain the key buffer in memory and avoid deallocations
return key::isAccessible;
} else {
rocksIterator.seekForPrev(LLUtils.toArray(key));
return null;
}
}
/**
* Useful for forward iterations
*/
@Nullable
private static SafeCloseable rocksIterSeekTo(boolean allowNettyDirect,
RocksIterator rocksIterator, Buffer key) {
@ -1391,7 +1401,7 @@ public class LLLocalDictionary implements LLDictionary {
}
try (var rocksIterator = db.newIterator(readOpts)) {
SafeCloseable seekTo;
if (!LLLocalDictionary.PREFER_SEEK_TO_FIRST && range.hasMin()) {
if (!LLLocalDictionary.PREFER_AUTO_SEEK_BOUND && range.hasMin()) {
seekTo = rocksIterSeekTo(nettyDirect, rocksIterator, range.getMinUnsafe());
} else {
seekTo = null;
@ -1447,7 +1457,7 @@ public class LLLocalDictionary implements LLDictionary {
}
try (var rocksIterator = db.newIterator(readOpts)) {
SafeCloseable seekTo;
if (!LLLocalDictionary.PREFER_SEEK_TO_FIRST && range.hasMin()) {
if (!LLLocalDictionary.PREFER_AUTO_SEEK_BOUND && range.hasMin()) {
seekTo = rocksIterSeekTo(nettyDirect, rocksIterator, range.getMinUnsafe());
} else {
seekTo = null;
@ -1503,7 +1513,7 @@ public class LLLocalDictionary implements LLDictionary {
}
try (var rocksIterator = db.newIterator(readOpts)) {
SafeCloseable seekTo;
if (!LLLocalDictionary.PREFER_SEEK_TO_FIRST && range.hasMin()) {
if (!LLLocalDictionary.PREFER_AUTO_SEEK_BOUND && range.hasMin()) {
seekTo = rocksIterSeekTo(nettyDirect, rocksIterator, range.getMinUnsafe());
} else {
seekTo = null;
@ -1668,7 +1678,7 @@ public class LLLocalDictionary implements LLDictionary {
}
try (RocksIterator rocksIterator = db.newIterator(readOpts)) {
SafeCloseable seekTo;
if (!LLLocalDictionary.PREFER_SEEK_TO_FIRST && range.hasMin()) {
if (!LLLocalDictionary.PREFER_AUTO_SEEK_BOUND && range.hasMin()) {
seekTo = rocksIterSeekTo(nettyDirect, rocksIterator, range.getMinUnsafe());
} else {
seekTo = null;
@ -1706,11 +1716,11 @@ public class LLLocalDictionary implements LLDictionary {
* This method should not modify or move the writerIndex/readerIndex of the buffers inside the range
*/
@NotNull
public static RocksIteratorTuple getRocksIterator(
boolean allowNettyDirect,
public static RocksIteratorTuple getRocksIterator(boolean allowNettyDirect,
ReadOptions readOptions,
LLRange range,
RocksDBColumn db) {
RocksDBColumn db,
boolean reverse) {
assert !Schedulers.isInNonBlockingThread() : "Called getRocksIterator in a nonblocking thread";
ReleasableSlice sliceMin;
ReleasableSlice sliceMax;
@ -1725,14 +1735,24 @@ public class LLLocalDictionary implements LLDictionary {
sliceMax = emptyReleasableSlice();
}
var rocksIterator = db.newIterator(readOptions);
SafeCloseable seekTo;
if (!PREFER_SEEK_TO_FIRST && range.hasMin()) {
seekTo = Objects.requireNonNullElseGet(rocksIterSeekTo(allowNettyDirect, rocksIterator, range.getMinUnsafe()),
() -> ((SafeCloseable) () -> {}));
SafeCloseable seekFromOrTo;
if (reverse) {
if (!PREFER_AUTO_SEEK_BOUND && range.hasMax()) {
seekFromOrTo = Objects.requireNonNullElseGet(rocksIterSeekFrom(allowNettyDirect, rocksIterator, range.getMaxUnsafe()),
() -> ((SafeCloseable) () -> {}));
} else {
seekFromOrTo = () -> {};
rocksIterator.seekToLast();
}
} else {
seekTo = () -> {};
rocksIterator.seekToFirst();
if (!PREFER_AUTO_SEEK_BOUND && range.hasMin()) {
seekFromOrTo = Objects.requireNonNullElseGet(rocksIterSeekTo(allowNettyDirect, rocksIterator, range.getMinUnsafe()),
() -> ((SafeCloseable) () -> {}));
} else {
seekFromOrTo = () -> {};
rocksIterator.seekToFirst();
}
}
return new RocksIteratorTuple(List.of(readOptions), rocksIterator, sliceMin, sliceMax, seekTo);
return new RocksIteratorTuple(List.of(readOptions), rocksIterator, sliceMin, sliceMax, seekFromOrTo);
}
}

9
src/main/java/it/cavallium/dbengine/database/disk/LLLocalEntryReactiveRocksIterator.java
View File

@ -1,23 +1,18 @@
package it.cavallium.dbengine.database.disk;
import io.netty5.buffer.api.Buffer;
import io.netty5.buffer.api.BufferAllocator;
import io.netty5.buffer.api.Owned;
import io.netty5.buffer.api.Send;
import it.cavallium.dbengine.database.LLEntry;
import it.cavallium.dbengine.database.LLRange;
import it.cavallium.dbengine.netty.NullableBuffer;
import org.rocksdb.ColumnFamilyHandle;
import org.rocksdb.ReadOptions;
import org.rocksdb.RocksDB;
public class LLLocalEntryReactiveRocksIterator extends LLLocalReactiveRocksIterator<Send<LLEntry>> {
public LLLocalEntryReactiveRocksIterator(RocksDBColumn db,
Send<LLRange> range,
boolean allowNettyDirect,
ReadOptions readOptions) {
super(db, range, allowNettyDirect, readOptions, true);
ReadOptions readOptions, boolean reverse) {
super(db, range, allowNettyDirect, readOptions, true, reverse);
}
@Override

2
src/main/java/it/cavallium/dbengine/database/disk/LLLocalGroupedReactiveRocksIterator.java
View File

@ -92,7 +92,7 @@ public abstract class LLLocalGroupedReactiveRocksIterator<T> extends
if (logger.isTraceEnabled()) {
logger.trace(MARKER_ROCKSDB, "Range {} started", LLUtils.toStringSafe(range));
}
return LLLocalDictionary.getRocksIterator(allowNettyDirect, readOptions, range, db);
return LLLocalDictionary.getRocksIterator(allowNettyDirect, readOptions, range, db, false);
}, (tuple, sink) -> {
try {
var rocksIterator = tuple.iterator();

2
src/main/java/it/cavallium/dbengine/database/disk/LLLocalKeyPrefixReactiveRocksIterator.java
View File

@ -84,7 +84,7 @@ public class LLLocalKeyPrefixReactiveRocksIterator extends
if (logger.isTraceEnabled()) {
logger.trace(MARKER_ROCKSDB, "Range {} started", LLUtils.toStringSafe(rangeShared));
}
return LLLocalDictionary.getRocksIterator(allowNettyDirect, readOptions, rangeShared, db);
return LLLocalDictionary.getRocksIterator(allowNettyDirect, readOptions, rangeShared, db, false);
}, (tuple, sink) -> {
try {
var rocksIterator = tuple.iterator();

8
src/main/java/it/cavallium/dbengine/database/disk/LLLocalKeyReactiveRocksIterator.java
View File

@ -1,20 +1,18 @@
package it.cavallium.dbengine.database.disk;
import io.netty5.buffer.api.Buffer;
import io.netty5.buffer.api.BufferAllocator;
import io.netty5.buffer.api.Send;
import it.cavallium.dbengine.database.LLRange;
import org.rocksdb.ColumnFamilyHandle;
import org.rocksdb.ReadOptions;
import org.rocksdb.RocksDB;
public class LLLocalKeyReactiveRocksIterator extends LLLocalReactiveRocksIterator<Send<Buffer>> {
public LLLocalKeyReactiveRocksIterator(RocksDBColumn db,
Send<LLRange> range,
boolean allowNettyDirect,
ReadOptions readOptions) {
super(db, range, allowNettyDirect, readOptions, false);
ReadOptions readOptions,
boolean reverse) {
super(db, range, allowNettyDirect, readOptions, false, reverse);
}
@Override

15
src/main/java/it/cavallium/dbengine/database/disk/LLLocalReactiveRocksIterator.java
View File

@ -60,13 +60,15 @@ public abstract class LLLocalReactiveRocksIterator<T> extends
private final boolean allowNettyDirect;
private ReadOptions readOptions;
private final boolean readValues;
private final boolean reverse;
@SuppressWarnings({"unchecked", "rawtypes"})
public LLLocalReactiveRocksIterator(RocksDBColumn db,
Send<LLRange> range,
boolean allowNettyDirect,
ReadOptions readOptions,
boolean readValues) {
boolean readValues,
boolean reverse) {
super((Drop<LLLocalReactiveRocksIterator<T>>) (Drop) DROP);
try (range) {
this.db = db;
@ -74,6 +76,7 @@ public abstract class LLLocalReactiveRocksIterator<T> extends
this.allowNettyDirect = allowNettyDirect;
this.readOptions = readOptions;
this.readValues = readValues;
this.reverse = reverse;
}
}
@ -83,7 +86,7 @@ public abstract class LLLocalReactiveRocksIterator<T> extends
if (logger.isTraceEnabled()) {
logger.trace(MARKER_ROCKSDB, "Range {} started", LLUtils.toStringSafe(rangeShared));
}
return getRocksIterator(allowNettyDirect, readOptions, rangeShared, db);
return getRocksIterator(allowNettyDirect, readOptions, rangeShared, db, reverse);
}, (tuple, sink) -> {
try {
var rocksIterator = tuple.iterator();
@ -117,7 +120,11 @@ public abstract class LLLocalReactiveRocksIterator<T> extends
}
try {
rocksIterator.next();
if (reverse) {
rocksIterator.prev();
} else {
rocksIterator.next();
}
rocksIterator.status();
sink.next(getEntry(key.send(), value == null ? null : value.send()));
} finally {
@ -153,7 +160,7 @@ public abstract class LLLocalReactiveRocksIterator<T> extends
protected Owned<LLLocalReactiveRocksIterator<T>> prepareSend() {
var range = this.rangeShared.send();
var readOptions = this.readOptions;
return drop -> new LLLocalReactiveRocksIterator<>(db, range, allowNettyDirect, readOptions, readValues) {
return drop -> new LLLocalReactiveRocksIterator<>(db, range, allowNettyDirect, readOptions, readValues, reverse) {
@Override
public T getEntry(@Nullable Send<Buffer> key, @Nullable Send<Buffer> value) {
return LLLocalReactiveRocksIterator.this.getEntry(key, value);

59
src/main/java/it/cavallium/dbengine/database/memory/LLMemoryDictionary.java
View File

@ -2,7 +2,6 @@ package it.cavallium.dbengine.database.memory;
import io.netty5.buffer.api.Buffer;
import io.netty5.buffer.api.BufferAllocator;
import io.netty5.buffer.api.Resource;
import io.netty5.buffer.api.Send;
import it.cavallium.dbengine.client.BadBlock;
import it.cavallium.dbengine.database.LLDelta;
@ -18,19 +17,19 @@ import it.cavallium.dbengine.database.serialization.SerializationException;
import it.cavallium.dbengine.database.serialization.SerializationFunction;
import it.unimi.dsi.fastutil.bytes.ByteList;
import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.SortedMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.atomic.AtomicReference;
import org.jetbrains.annotations.Nullable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.util.function.Tuple2;
import reactor.util.function.Tuple3;
import reactor.util.function.Tuples;
public class LLMemoryDictionary implements LLDictionary {
@ -133,7 +132,7 @@ public class LLMemoryDictionary implements LLDictionary {
}
}
private Map<ByteList, ByteList> mapSlice(LLSnapshot snapshot, Send<LLRange> rangeToReceive) {
private ConcurrentNavigableMap<ByteList, ByteList> mapSlice(LLSnapshot snapshot, Send<LLRange> rangeToReceive) {
try (var range = rangeToReceive.receive()) {
if (range.isAll()) {
return snapshots.get(resolveSnapshot(snapshot));
@ -143,15 +142,15 @@ public class LLMemoryDictionary implements LLDictionary {
.get(resolveSnapshot(snapshot))
.get(key);
if (value != null) {
return Map.of(key, value);
return new ConcurrentSkipListMap<>(Map.of(key, value));
} else {
return Map.of();
return new ConcurrentSkipListMap<>(Map.of());
}
} else if (range.hasMin() && range.hasMax()) {
var min = k(range.getMin());
var max = k(range.getMax());
if (min.compareTo(max) > 0) {
return Map.of();
return new ConcurrentSkipListMap<>(Map.of());
}
return snapshots
.get(resolveSnapshot(snapshot))
@ -169,7 +168,7 @@ public class LLMemoryDictionary implements LLDictionary {
}
@Override
public Mono<Send<Buffer>> get(@Nullable LLSnapshot snapshot, Mono<Send<Buffer>> keyMono, boolean existsAlmostCertainly) {
public Mono<Send<Buffer>> get(@Nullable LLSnapshot snapshot, Mono<Send<Buffer>> keyMono) {
return Mono.usingWhen(keyMono,
key -> Mono
.fromCallable(() -> snapshots.get(resolveSnapshot(snapshot)).get(k(key)))
@ -263,8 +262,7 @@ public class LLMemoryDictionary implements LLDictionary {
}
@Override
public Flux<Optional<Buffer>> getMulti(@Nullable LLSnapshot snapshot, Flux<Send<Buffer>> keys,
boolean existsAlmostCertainly) {
public Flux<Optional<Buffer>> getMulti(@Nullable LLSnapshot snapshot, Flux<Send<Buffer>> keys) {
return keys.map(key -> {
try (var t2 = key.receive()) {
ByteList v = snapshots.get(resolveSnapshot(snapshot)).get(k(t2.copy().send()));
@ -298,9 +296,7 @@ public class LLMemoryDictionary implements LLDictionary {
}
@Override
public Flux<Send<LLEntry>> getRange(@Nullable LLSnapshot snapshot,
Mono<Send<LLRange>> rangeMono,
boolean existsAlmostCertainly) {
public Flux<Send<LLEntry>> getRange(@Nullable LLSnapshot snapshot, Mono<Send<LLRange>> rangeMono, boolean reverse) {
return Flux.usingWhen(rangeMono, rangeToReceive -> {
try (var range = rangeToReceive.receive()) {
if (range.isSingle()) {
@ -319,7 +315,13 @@ public class LLMemoryDictionary implements LLDictionary {
var rangeToReceive2 = range.send();
return Mono
.fromCallable(() -> mapSlice(snapshot, rangeToReceive2))
.flatMapMany(map -> Flux.fromIterable(map.entrySet()))
.flatMapIterable(map -> {
if (reverse) {
return map.descendingMap().entrySet();
} else {
return map.entrySet();
}
})
.map(entry -> LLEntry.of(kk(entry.getKey()), kk(entry.getValue())).send());
}
}
@ -329,8 +331,7 @@ public class LLMemoryDictionary implements LLDictionary {
@Override
public Flux<List<Send<LLEntry>>> getRangeGrouped(@Nullable LLSnapshot snapshot,
Mono<Send<LLRange>> rangeMono,
int prefixLength,
boolean existsAlmostCertainly) {
int prefixLength) {
return Flux.usingWhen(rangeMono, rangeToReceive -> {
try (var range = rangeToReceive.receive()) {
if (range.isSingle()) {
@ -349,7 +350,7 @@ public class LLMemoryDictionary implements LLDictionary {
var rangeToReceive2 = range.send();
return Mono
.fromCallable(() -> mapSlice(snapshot, rangeToReceive2))
.flatMapMany(map -> Flux.fromIterable(map.entrySet()))
.flatMapIterable(SortedMap::entrySet)
.groupBy(k -> k.getKey().subList(0, prefixLength))
.flatMap(groupedFlux -> groupedFlux
.map(entry -> LLEntry.of(kk(entry.getKey()), kk(entry.getValue())).send())
@ -361,7 +362,7 @@ public class LLMemoryDictionary implements LLDictionary {
}
@Override
public Flux<Send<Buffer>> getRangeKeys(@Nullable LLSnapshot snapshot, Mono<Send<LLRange>> rangeMono) {
public Flux<Send<Buffer>> getRangeKeys(@Nullable LLSnapshot snapshot, Mono<Send<LLRange>> rangeMono, boolean reverse) {
return Flux.usingWhen(rangeMono,
rangeToReceive -> {
try (var range = rangeToReceive.receive()) {
@ -377,8 +378,14 @@ public class LLMemoryDictionary implements LLDictionary {
var rangeToReceive2 = range.send();
return Mono
.fromCallable(() -> mapSlice(snapshot, rangeToReceive2))
.flatMapMany(map -> Flux.fromIterable(map.entrySet()))
.map(entry -> kk(entry.getKey()));
.<ByteList>flatMapIterable(map -> {
if (reverse) {
return map.descendingMap().keySet();
} else {
return map.keySet();
}
})
.map(this::kk);
}
}
},
@ -408,7 +415,7 @@ public class LLMemoryDictionary implements LLDictionary {
var rangeToReceive2 = range.send();
return Mono
.fromCallable(() -> mapSlice(snapshot, rangeToReceive2))
.flatMapMany(map -> Flux.fromIterable(map.entrySet()))
.flatMapIterable(SortedMap::entrySet)
.groupBy(k -> k.getKey().subList(0, prefixLength))
.flatMap(groupedFlux -> groupedFlux
.map(entry -> kk(entry.getKey()))
@ -443,7 +450,7 @@ public class LLMemoryDictionary implements LLDictionary {
var rangeToReceive2 = range.send();
return Mono
.fromCallable(() -> mapSlice(snapshot, rangeToReceive2))
.flatMapMany(map -> Flux.fromIterable(map.entrySet()))
.flatMapIterable(SortedMap::entrySet)
.map(k -> (ByteList) k.getKey().subList(0, prefixLength))
.distinctUntilChanged()
.map(this::kk);
@ -513,7 +520,7 @@ public class LLMemoryDictionary implements LLDictionary {
@Override
public Mono<Boolean> isRangeEmpty(@Nullable LLSnapshot snapshot, Mono<Send<LLRange>> rangeMono, boolean fillCache) {
return getRangeKeys(snapshot, rangeMono)
return getRangeKeys(snapshot, rangeMono, false)
.doOnNext(buf -> buf.receive().close())
.count()
.map(count -> count == 0);
@ -529,14 +536,14 @@ public class LLMemoryDictionary implements LLDictionary {
@Override
public Mono<Send<LLEntry>> getOne(@Nullable LLSnapshot snapshot, Mono<Send<LLRange>> rangeMono) {
return getRange(snapshot, rangeMono)
return getRange(snapshot, rangeMono, false)
.take(1, true)
.singleOrEmpty();
}
@Override
public Mono<Send<Buffer>> getOneKey(@Nullable LLSnapshot snapshot, Mono<Send<LLRange>> rangeMono) {
return getRangeKeys(snapshot, rangeMono)
return getRangeKeys(snapshot, rangeMono, false)
.take(1, true)
.singleOrEmpty();
}

3
src/main/java/it/cavallium/dbengine/database/memory/LLMemorySingleton.java
View File

@ -7,7 +7,6 @@ import it.cavallium.dbengine.database.LLDelta;
import it.cavallium.dbengine.database.LLDictionaryResultType;
import it.cavallium.dbengine.database.LLSingleton;
import it.cavallium.dbengine.database.LLSnapshot;
import it.cavallium.dbengine.database.LLUtils;
import it.cavallium.dbengine.database.UpdateReturnMode;
import it.cavallium.dbengine.database.serialization.SerializationFunction;
import org.jetbrains.annotations.Nullable;
@ -43,7 +42,7 @@ public class LLMemorySingleton implements LLSingleton {
@Override
public Mono<Send<Buffer>> get(@Nullable LLSnapshot snapshot) {
return dict.get(snapshot, singletonNameBufMono, false);
return dict.get(snapshot, singletonNameBufMono);
}
@Override

15
src/test/java/it/cavallium/dbengine/TestLLDictionary.java
View File

@ -156,11 +156,11 @@ public abstract class TestLLDictionary {
var keyEx = Mono.fromCallable(() -> fromString("test-key-1").send());
var keyNonEx = Mono.fromCallable(() -> fromString("test-nonexistent").send());
Assertions.assertEquals("test-value", run(dict.get(null, keyEx).map(this::toString)));
Assertions.assertEquals("test-value", run(dict.get(null, keyEx, true).map(this::toString)));
Assertions.assertEquals("test-value", run(dict.get(null, keyEx, false).map(this::toString)));
Assertions.assertEquals("test-value", run(dict.get(null, keyEx).map(this::toString)));
Assertions.assertEquals("test-value", run(dict.get(null, keyEx).map(this::toString)));
Assertions.assertEquals((String) null, run(dict.get(null, keyNonEx).map(this::toString)));
Assertions.assertEquals((String) null, run(dict.get(null, keyNonEx).map(this::toString)));
Assertions.assertEquals((String) null, run(dict.get(null, keyNonEx).map(this::toString)));
Assertions.assertEquals((String) null, run(dict.get(null, keyNonEx, true).map(this::toString)));
Assertions.assertEquals((String) null, run(dict.get(null, keyNonEx, false).map(this::toString)));
}
@ParameterizedTest
@ -192,7 +192,8 @@ public abstract class TestLLDictionary {
var afterSize = run(dict.sizeRange(null, Mono.fromCallable(() -> LLRange.all().send()), false));
Assertions.assertEquals(1, afterSize - beforeSize);
Assertions.assertTrue(run(dict.getRangeKeys(null, RANGE_ALL).map(this::toString).collectList()).contains("test-nonexistent"));
Assertions.assertTrue(run(dict.getRangeKeys(null, RANGE_ALL, false).map(this::toString).collectList()).contains("test-nonexistent"));
Assertions.assertTrue(run(dict.getRangeKeys(null, RANGE_ALL, true).map(this::toString).collectList()).contains("test-nonexistent"));
}
@ParameterizedTest
@ -251,7 +252,9 @@ public abstract class TestLLDictionary {
assertEquals(expected, afterSize - beforeSize);
if (updateMode != UpdateMode.DISALLOW) {
Assertions.assertTrue(run(dict.getRangeKeys(null, RANGE_ALL).map(this::toString).collectList()).contains(
Assertions.assertTrue(run(dict.getRangeKeys(null, RANGE_ALL, false).map(this::toString).collectList()).contains(
"test-nonexistent"));
Assertions.assertTrue(run(dict.getRangeKeys(null, RANGE_ALL, true).map(this::toString).collectList()).contains(
"test-nonexistent"));
}
}

5
src/test/java/it/cavallium/dbengine/TestLLDictionaryLeaks.java
View File

@ -14,7 +14,6 @@ import it.cavallium.dbengine.DbTestUtils.TestAllocator;
import it.cavallium.dbengine.database.LLDictionary;
import it.cavallium.dbengine.database.LLDictionaryResultType;
import it.cavallium.dbengine.database.LLKeyValueDatabase;
import it.cavallium.dbengine.database.LLUtils;
import it.cavallium.dbengine.database.UpdateMode;
import it.cavallium.dbengine.database.UpdateReturnMode;
import java.nio.charset.StandardCharsets;
@ -134,8 +133,8 @@ public abstract class TestLLDictionaryLeaks {
var dict = getDict(updateMode);
var key = Mono.fromCallable(() -> fromString("test"));
runVoid(dict.get(null, key).then());
runVoid(dict.get(null, key, true).then());
runVoid(dict.get(null, key, false).then());
runVoid(dict.get(null, key).then());
runVoid(dict.get(null, key).then());
}
@ParameterizedTest