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

Configurable direct buffers

Browse Source
This commit is contained in:
Andrea Cavalli 2021-06-29 23:31:02 +02:00
parent 2df2c00e36
commit 3758d06853
12 changed files with 141 additions and 65 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
src/main/java/it/cavallium/dbengine/database/disk/DatabaseOptions.java
View File

@ -11,4 +11,6 @@ public record DatabaseOptions(Map<String, String> extraFlags,
boolean lowMemory, boolean lowMemory,
boolean inMemory, boolean inMemory,
boolean useDirectIO, boolean useDirectIO,
boolean allowMemoryMapping) {} boolean allowMemoryMapping,
boolean allowNettyDirect,
boolean useNettyDirect) {}

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

@ -138,6 +138,7 @@ public class LLLocalDictionary implements LLDictionary {
private final ByteBufAllocator alloc; private final ByteBufAllocator alloc;
private final String getRangeMultiDebugName; private final String getRangeMultiDebugName;
private final String getRangeKeysMultiDebugName; private final String getRangeKeysMultiDebugName;
private final DatabaseOptions databaseOptions;
public LLLocalDictionary( public LLLocalDictionary(
ByteBufAllocator allocator, ByteBufAllocator allocator,
@ -147,7 +148,8 @@ public class LLLocalDictionary implements LLDictionary {
String columnName, String columnName,
Scheduler dbScheduler, Scheduler dbScheduler,
Function<LLSnapshot, Snapshot> snapshotResolver, Function<LLSnapshot, Snapshot> snapshotResolver,
UpdateMode updateMode) { UpdateMode updateMode,
DatabaseOptions databaseOptions) {
Objects.requireNonNull(db); Objects.requireNonNull(db);
this.db = db; this.db = db;
Objects.requireNonNull(columnFamilyHandle); Objects.requireNonNull(columnFamilyHandle);
@ -159,6 +161,7 @@ public class LLLocalDictionary implements LLDictionary {
this.updateMode = updateMode; this.updateMode = updateMode;
this.getRangeMultiDebugName = databaseName + "(" + columnName + ")" + "::getRangeMulti"; this.getRangeMultiDebugName = databaseName + "(" + columnName + ")" + "::getRangeMulti";
this.getRangeKeysMultiDebugName = databaseName + "(" + columnName + ")" + "::getRangeKeysMulti"; this.getRangeKeysMultiDebugName = databaseName + "(" + columnName + ")" + "::getRangeKeysMulti";
this.databaseOptions = databaseOptions;
alloc = allocator; alloc = allocator;
} }
@ -258,7 +261,7 @@ public class LLLocalDictionary implements LLDictionary {
ByteBuf key, ByteBuf key,
boolean existsAlmostCertainly) throws RocksDBException { boolean existsAlmostCertainly) throws RocksDBException {
try { try {
if (key.isDirect()) { if (databaseOptions.allowNettyDirect() && key.isDirect()) {
//todo: implement keyMayExist if existsAlmostCertainly is false. //todo: implement keyMayExist if existsAlmostCertainly is false.
// Unfortunately it's not feasible until RocksDB implements keyMayExist with buffers // Unfortunately it's not feasible until RocksDB implements keyMayExist with buffers
@ -366,7 +369,7 @@ public class LLLocalDictionary implements LLDictionary {
private void dbPut(ColumnFamilyHandle cfh, @Nullable WriteOptions writeOptions, ByteBuf key, ByteBuf value) private void dbPut(ColumnFamilyHandle cfh, @Nullable WriteOptions writeOptions, ByteBuf key, ByteBuf value)
throws RocksDBException { throws RocksDBException {
try { try {
if (key.isDirect() && value.isDirect()) { if (databaseOptions.allowNettyDirect() && key.isDirect() && value.isDirect()) {
if (!key.isDirect()) { if (!key.isDirect()) {
throw new RocksDBException("Key buffer must be direct"); throw new RocksDBException("Key buffer must be direct");
} }
@ -416,7 +419,7 @@ public class LLLocalDictionary implements LLDictionary {
readOpts.setVerifyChecksums(VERIFY_CHECKSUMS_WHEN_NOT_NEEDED); readOpts.setVerifyChecksums(VERIFY_CHECKSUMS_WHEN_NOT_NEEDED);
readOpts.setFillCache(false); readOpts.setFillCache(false);
if (range.hasMin()) { if (range.hasMin()) {
if (range.getMin().isDirect()) { if (databaseOptions.allowNettyDirect() && range.getMin().isDirect()) {
readOpts.setIterateLowerBound(new DirectSlice(Objects.requireNonNull(LLUtils.toDirect(range.getMin()), readOpts.setIterateLowerBound(new DirectSlice(Objects.requireNonNull(LLUtils.toDirect(range.getMin()),
"This range must use direct buffers" "This range must use direct buffers"
))); )));
@ -425,7 +428,7 @@ public class LLLocalDictionary implements LLDictionary {
} }
} }
if (range.hasMax()) { if (range.hasMax()) {
if (range.getMax().isDirect()) { if (databaseOptions.allowNettyDirect() && range.getMax().isDirect()) {
readOpts.setIterateUpperBound(new DirectSlice(Objects.requireNonNull(LLUtils.toDirect(range.getMax()), readOpts.setIterateUpperBound(new DirectSlice(Objects.requireNonNull(LLUtils.toDirect(range.getMax()),
"This range must use direct buffers" "This range must use direct buffers"
))); )));
@ -435,7 +438,7 @@ public class LLLocalDictionary implements LLDictionary {
} }
try (RocksIterator rocksIterator = db.newIterator(cfh, readOpts)) { try (RocksIterator rocksIterator = db.newIterator(cfh, readOpts)) {
if (!LLLocalDictionary.PREFER_SEEK_TO_FIRST && range.hasMin()) { if (!LLLocalDictionary.PREFER_SEEK_TO_FIRST && range.hasMin()) {
if (range.getMin().isDirect()) { if (databaseOptions.allowNettyDirect() && range.getMin().isDirect()) {
rocksIterator.seek(Objects.requireNonNull(LLUtils.toDirect(range.getMin()), rocksIterator.seek(Objects.requireNonNull(LLUtils.toDirect(range.getMin()),
"This range must use direct buffers" "This range must use direct buffers"
)); ));
@ -809,7 +812,7 @@ public class LLLocalDictionary implements LLDictionary {
private void dbDelete(ColumnFamilyHandle cfh, @Nullable WriteOptions writeOptions, ByteBuf key) private void dbDelete(ColumnFamilyHandle cfh, @Nullable WriteOptions writeOptions, ByteBuf key)
throws RocksDBException { throws RocksDBException {
try { try {
if (key.isDirect()) { if (databaseOptions.allowNettyDirect() && key.isDirect()) {
if (!key.isDirect()) { if (!key.isDirect()) {
throw new IllegalArgumentException("Key must be a direct buffer"); throw new IllegalArgumentException("Key must be a direct buffer");
} }
@ -983,7 +986,7 @@ public class LLLocalDictionary implements LLDictionary {
} }
}) })
.subscribeOn(dbScheduler) .subscribeOn(dbScheduler)
.flatMapMany(entries -> Flux.fromIterable(entries)) .flatMapMany(Flux::fromIterable)
.onErrorMap(cause -> new IOException("Failed to read keys " .onErrorMap(cause -> new IOException("Failed to read keys "
+ Arrays.deepToString(keysWindow.toArray(ByteBuf[]::new)), cause)) + Arrays.deepToString(keysWindow.toArray(ByteBuf[]::new)), cause))
.doAfterTerminate(() -> keysWindow.forEach(ReferenceCounted::release)) .doAfterTerminate(() -> keysWindow.forEach(ReferenceCounted::release))
@ -1001,23 +1004,7 @@ public class LLLocalDictionary implements LLDictionary {
public Flux<Entry<ByteBuf, ByteBuf>> putMulti(Flux<Entry<ByteBuf, ByteBuf>> entries, boolean getOldValues) { public Flux<Entry<ByteBuf, ByteBuf>> putMulti(Flux<Entry<ByteBuf, ByteBuf>> entries, boolean getOldValues) {
return entries return entries
.window(Math.min(MULTI_GET_WINDOW, CAPPED_WRITE_BATCH_CAP)) .window(Math.min(MULTI_GET_WINDOW, CAPPED_WRITE_BATCH_CAP))
.doOnDiscard(Entry.class, entry -> {
if (entry.getKey() instanceof ByteBuf && entry.getValue() instanceof ByteBuf) {
//noinspection unchecked
var castedEntry = (Entry<ByteBuf, ByteBuf>) entry;
castedEntry.getKey().release();
castedEntry.getValue().release();
}
})
.flatMap(Flux::collectList) .flatMap(Flux::collectList)
.doOnDiscard(Entry.class, entry -> {
if (entry.getKey() instanceof ByteBuf && entry.getValue() instanceof ByteBuf) {
//noinspection unchecked
var castedEntry = (Entry<ByteBuf, ByteBuf>) entry;
castedEntry.getKey().release();
castedEntry.getValue().release();
}
})
.map(Collections::unmodifiableList) .map(Collections::unmodifiableList)
.flatMap(ew -> Mono .flatMap(ew -> Mono
.using( .using(
@ -1089,6 +1076,14 @@ public class LLLocalDictionary implements LLDictionary {
} }
) )
) )
.doOnDiscard(Entry.class, entry -> {
if (entry.getKey() instanceof ByteBuf && entry.getValue() instanceof ByteBuf) {
//noinspection unchecked
var castedEntry = (Entry<ByteBuf, ByteBuf>) entry;
castedEntry.getKey().release();
castedEntry.getValue().release();
}
})
.doOnDiscard(Collection.class, obj -> { .doOnDiscard(Collection.class, obj -> {
//noinspection unchecked //noinspection unchecked
var castedEntries = (Collection<Entry<ByteBuf, ByteBuf>>) obj; var castedEntries = (Collection<Entry<ByteBuf, ByteBuf>>) obj;
@ -1161,6 +1156,7 @@ public class LLLocalDictionary implements LLDictionary {
alloc, alloc,
cfh, cfh,
range.retain(), range.retain(),
databaseOptions.allowNettyDirect(),
resolveSnapshot(snapshot), resolveSnapshot(snapshot),
getRangeMultiDebugName getRangeMultiDebugName
), ),
@ -1191,6 +1187,7 @@ public class LLLocalDictionary implements LLDictionary {
cfh, cfh,
prefixLength, prefixLength,
range.retain(), range.retain(),
databaseOptions.allowNettyDirect(),
resolveSnapshot(snapshot), resolveSnapshot(snapshot),
"getRangeMultiGrouped" "getRangeMultiGrouped"
), ),
@ -1233,6 +1230,7 @@ public class LLLocalDictionary implements LLDictionary {
cfh, cfh,
prefixLength, prefixLength,
range.retain(), range.retain(),
databaseOptions.allowNettyDirect(),
resolveSnapshot(snapshot), resolveSnapshot(snapshot),
"getRangeKeysGrouped" "getRangeKeysGrouped"
), ),
@ -1257,7 +1255,7 @@ public class LLLocalDictionary implements LLDictionary {
ro.setReadaheadSize(32 * 1024); ro.setReadaheadSize(32 * 1024);
} }
ro.setVerifyChecksums(true); ro.setVerifyChecksums(true);
var rocksIteratorTuple = getRocksIterator(ro, range.retain(), db, cfh); var rocksIteratorTuple = getRocksIterator(databaseOptions.allowNettyDirect(), ro, range.retain(), db, cfh);
try { try {
try (var rocksIterator = rocksIteratorTuple.getT1()) { try (var rocksIterator = rocksIteratorTuple.getT1()) {
rocksIterator.seekToFirst(); rocksIterator.seekToFirst();
@ -1299,6 +1297,7 @@ public class LLLocalDictionary implements LLDictionary {
cfh, cfh,
prefixLength, prefixLength,
range.retain(), range.retain(),
databaseOptions.allowNettyDirect(),
resolveSnapshot(snapshot), resolveSnapshot(snapshot),
true, true,
"getRangeKeysGrouped" "getRangeKeysGrouped"
@ -1343,6 +1342,7 @@ public class LLLocalDictionary implements LLDictionary {
alloc, alloc,
cfh, cfh,
range.retain(), range.retain(),
databaseOptions.allowNettyDirect(),
resolveSnapshot(snapshot), resolveSnapshot(snapshot),
getRangeKeysMultiDebugName getRangeKeysMultiDebugName
), ),
@ -1368,20 +1368,28 @@ public class LLLocalDictionary implements LLDictionary {
try (var opts = new ReadOptions(EMPTY_READ_OPTIONS)) { try (var opts = new ReadOptions(EMPTY_READ_OPTIONS)) {
ReleasableSlice minBound; ReleasableSlice minBound;
if (range.hasMin()) { if (range.hasMin()) {
minBound = setIterateBound(opts, IterateBound.LOWER, range.getMin().retain()); minBound = setIterateBound(databaseOptions.allowNettyDirect(),
opts,
IterateBound.LOWER,
range.getMin().retain()
);
} else { } else {
minBound = emptyReleasableSlice(); minBound = emptyReleasableSlice();
} }
try { try {
ReleasableSlice maxBound; ReleasableSlice maxBound;
if (range.hasMax()) { if (range.hasMax()) {
maxBound = setIterateBound(opts, IterateBound.UPPER, range.getMax().retain()); maxBound = setIterateBound(databaseOptions.allowNettyDirect(),
opts,
IterateBound.UPPER,
range.getMax().retain()
);
} else { } else {
maxBound = emptyReleasableSlice(); maxBound = emptyReleasableSlice();
} }
try (RocksIterator it = db.newIterator(cfh, opts)) { try (RocksIterator it = db.newIterator(cfh, opts)) {
if (!PREFER_SEEK_TO_FIRST && range.hasMin()) { if (!PREFER_SEEK_TO_FIRST && range.hasMin()) {
rocksIterSeekTo(it, range.getMin().retain()); rocksIterSeekTo(databaseOptions.allowNettyDirect(), it, range.getMin().retain());
} else { } else {
it.seekToFirst(); it.seekToFirst();
} }
@ -1521,20 +1529,28 @@ public class LLLocalDictionary implements LLDictionary {
readOpts.setFillCache(false); readOpts.setFillCache(false);
ReleasableSlice minBound; ReleasableSlice minBound;
if (range.hasMin()) { if (range.hasMin()) {
minBound = setIterateBound(readOpts, IterateBound.LOWER, range.getMin().retain()); minBound = setIterateBound(databaseOptions.allowNettyDirect(),
readOpts,
IterateBound.LOWER,
range.getMin().retain()
);
} else { } else {
minBound = emptyReleasableSlice(); minBound = emptyReleasableSlice();
} }
try { try {
ReleasableSlice maxBound; ReleasableSlice maxBound;
if (range.hasMax()) { if (range.hasMax()) {
maxBound = setIterateBound(readOpts, IterateBound.UPPER, range.getMax().retain()); maxBound = setIterateBound(databaseOptions.allowNettyDirect(),
readOpts,
IterateBound.UPPER,
range.getMax().retain()
);
} else { } else {
maxBound = emptyReleasableSlice(); maxBound = emptyReleasableSlice();
} }
try (var rocksIterator = db.newIterator(cfh, readOpts)) { try (var rocksIterator = db.newIterator(cfh, readOpts)) {
if (!LLLocalDictionary.PREFER_SEEK_TO_FIRST && range.hasMin()) { if (!LLLocalDictionary.PREFER_SEEK_TO_FIRST && range.hasMin()) {
rocksIterSeekTo(rocksIterator, range.getMin().retain()); rocksIterSeekTo(databaseOptions.allowNettyDirect(), rocksIterator, range.getMin().retain());
} else { } else {
rocksIterator.seekToFirst(); rocksIterator.seekToFirst();
} }
@ -1561,20 +1577,28 @@ public class LLLocalDictionary implements LLDictionary {
readOpts.setFillCache(false); readOpts.setFillCache(false);
ReleasableSlice minBound; ReleasableSlice minBound;
if (range.hasMin()) { if (range.hasMin()) {
minBound = setIterateBound(readOpts, IterateBound.LOWER, range.getMin().retain()); minBound = setIterateBound(databaseOptions.allowNettyDirect(),
readOpts,
IterateBound.LOWER,
range.getMin().retain()
);
} else { } else {
minBound = emptyReleasableSlice(); minBound = emptyReleasableSlice();
} }
try { try {
ReleasableSlice maxBound; ReleasableSlice maxBound;
if (range.hasMax()) { if (range.hasMax()) {
maxBound = setIterateBound(readOpts, IterateBound.UPPER, range.getMax().retain()); maxBound = setIterateBound(databaseOptions.allowNettyDirect(),
readOpts,
IterateBound.UPPER,
range.getMax().retain()
);
} else { } else {
maxBound = emptyReleasableSlice(); maxBound = emptyReleasableSlice();
} }
try (var rocksIterator = db.newIterator(cfh, readOpts)) { try (var rocksIterator = db.newIterator(cfh, readOpts)) {
if (!LLLocalDictionary.PREFER_SEEK_TO_FIRST && range.hasMin()) { if (!LLLocalDictionary.PREFER_SEEK_TO_FIRST && range.hasMin()) {
rocksIterSeekTo(rocksIterator, range.getMin().retain()); rocksIterSeekTo(databaseOptions.allowNettyDirect(), rocksIterator, range.getMin().retain());
} else { } else {
rocksIterator.seekToFirst(); rocksIterator.seekToFirst();
} }
@ -1595,9 +1619,9 @@ public class LLLocalDictionary implements LLDictionary {
} }
} }
private static void rocksIterSeekTo(RocksIterator rocksIterator, ByteBuf buffer) { private static void rocksIterSeekTo(boolean allowNettyDirect, RocksIterator rocksIterator, ByteBuf buffer) {
try { try {
if (buffer.isDirect()) { if (allowNettyDirect && buffer.isDirect()) {
ByteBuffer nioBuffer = LLUtils.toDirect(buffer); ByteBuffer nioBuffer = LLUtils.toDirect(buffer);
assert nioBuffer.isDirect(); assert nioBuffer.isDirect();
rocksIterator.seek(nioBuffer); rocksIterator.seek(nioBuffer);
@ -1611,11 +1635,11 @@ public class LLLocalDictionary implements LLDictionary {
} }
} }
private static ReleasableSlice setIterateBound(ReadOptions readOpts, IterateBound boundType, ByteBuf buffer) { private static ReleasableSlice setIterateBound(boolean allowNettyDirect, ReadOptions readOpts, IterateBound boundType, ByteBuf buffer) {
try { try {
Objects.requireNonNull(buffer); Objects.requireNonNull(buffer);
AbstractSlice<?> slice; AbstractSlice<?> slice;
if (LLLocalDictionary.USE_DIRECT_BUFFER_BOUNDS && buffer.isDirect()) { if (allowNettyDirect && LLLocalDictionary.USE_DIRECT_BUFFER_BOUNDS && buffer.isDirect()) {
ByteBuffer nioBuffer = LLUtils.toDirect(buffer); ByteBuffer nioBuffer = LLUtils.toDirect(buffer);
assert nioBuffer.isDirect(); assert nioBuffer.isDirect();
slice = new DirectSlice(nioBuffer, buffer.readableBytes()); slice = new DirectSlice(nioBuffer, buffer.readableBytes());
@ -1737,14 +1761,22 @@ public class LLLocalDictionary implements LLDictionary {
readOpts.setVerifyChecksums(VERIFY_CHECKSUMS_WHEN_NOT_NEEDED); readOpts.setVerifyChecksums(VERIFY_CHECKSUMS_WHEN_NOT_NEEDED);
ReleasableSlice minBound; ReleasableSlice minBound;
if (range.hasMin()) { if (range.hasMin()) {
minBound = setIterateBound(readOpts, IterateBound.LOWER, range.getMin().retain()); minBound = setIterateBound(databaseOptions.allowNettyDirect(),
readOpts,
IterateBound.LOWER,
range.getMin().retain()
);
} else { } else {
minBound = emptyReleasableSlice(); minBound = emptyReleasableSlice();
} }
try { try {
ReleasableSlice maxBound; ReleasableSlice maxBound;
if (range.hasMax()) { if (range.hasMax()) {
maxBound = setIterateBound(readOpts, IterateBound.UPPER, range.getMax().retain()); maxBound = setIterateBound(databaseOptions.allowNettyDirect(),
readOpts,
IterateBound.UPPER,
range.getMax().retain()
);
} else { } else {
maxBound = emptyReleasableSlice(); maxBound = emptyReleasableSlice();
} }
@ -1755,7 +1787,10 @@ public class LLLocalDictionary implements LLDictionary {
} }
try (var rocksIterator = db.newIterator(cfh, readOpts)) { try (var rocksIterator = db.newIterator(cfh, readOpts)) {
if (!LLLocalDictionary.PREFER_SEEK_TO_FIRST && range.hasMin()) { if (!LLLocalDictionary.PREFER_SEEK_TO_FIRST && range.hasMin()) {
rocksIterSeekTo(rocksIterator, range.getMin().retain()); rocksIterSeekTo(databaseOptions.allowNettyDirect(),
rocksIterator,
range.getMin().retain()
);
} else { } else {
rocksIterator.seekToFirst(); rocksIterator.seekToFirst();
} }
@ -1796,20 +1831,28 @@ public class LLLocalDictionary implements LLDictionary {
try (var readOpts = new ReadOptions(resolveSnapshot(snapshot))) { try (var readOpts = new ReadOptions(resolveSnapshot(snapshot))) {
ReleasableSlice minBound; ReleasableSlice minBound;
if (range.hasMin()) { if (range.hasMin()) {
minBound = setIterateBound(readOpts, IterateBound.LOWER, range.getMin().retain()); minBound = setIterateBound(databaseOptions.allowNettyDirect(),
readOpts,
IterateBound.LOWER,
range.getMin().retain()
);
} else { } else {
minBound = emptyReleasableSlice(); minBound = emptyReleasableSlice();
} }
try { try {
ReleasableSlice maxBound; ReleasableSlice maxBound;
if (range.hasMax()) { if (range.hasMax()) {
maxBound = setIterateBound(readOpts, IterateBound.UPPER, range.getMax().retain()); maxBound = setIterateBound(databaseOptions.allowNettyDirect(),
readOpts,
IterateBound.UPPER,
range.getMax().retain()
);
} else { } else {
maxBound = emptyReleasableSlice(); maxBound = emptyReleasableSlice();
} }
try (var rocksIterator = db.newIterator(cfh, readOpts)) { try (var rocksIterator = db.newIterator(cfh, readOpts)) {
if (!LLLocalDictionary.PREFER_SEEK_TO_FIRST && range.hasMin()) { if (!LLLocalDictionary.PREFER_SEEK_TO_FIRST && range.hasMin()) {
rocksIterSeekTo(rocksIterator, range.getMin().retain()); rocksIterSeekTo(databaseOptions.allowNettyDirect(), rocksIterator, range.getMin().retain());
} else { } else {
rocksIterator.seekToFirst(); rocksIterator.seekToFirst();
} }
@ -1853,20 +1896,28 @@ public class LLLocalDictionary implements LLDictionary {
try (var readOpts = new ReadOptions(resolveSnapshot(snapshot))) { try (var readOpts = new ReadOptions(resolveSnapshot(snapshot))) {
ReleasableSlice minBound; ReleasableSlice minBound;
if (range.hasMin()) { if (range.hasMin()) {
minBound = setIterateBound(readOpts, IterateBound.LOWER, range.getMin().retain()); minBound = setIterateBound(databaseOptions.allowNettyDirect(),
readOpts,
IterateBound.LOWER,
range.getMin().retain()
);
} else { } else {
minBound = emptyReleasableSlice(); minBound = emptyReleasableSlice();
} }
try { try {
ReleasableSlice maxBound; ReleasableSlice maxBound;
if (range.hasMax()) { if (range.hasMax()) {
maxBound = setIterateBound(readOpts, IterateBound.UPPER, range.getMax().retain()); maxBound = setIterateBound(databaseOptions.allowNettyDirect(),
readOpts,
IterateBound.UPPER,
range.getMax().retain()
);
} else { } else {
maxBound = emptyReleasableSlice(); maxBound = emptyReleasableSlice();
} }
try (var rocksIterator = db.newIterator(cfh, readOpts)) { try (var rocksIterator = db.newIterator(cfh, readOpts)) {
if (!LLLocalDictionary.PREFER_SEEK_TO_FIRST && range.hasMin()) { if (!LLLocalDictionary.PREFER_SEEK_TO_FIRST && range.hasMin()) {
rocksIterSeekTo(rocksIterator, range.getMin().retain()); rocksIterSeekTo(databaseOptions.allowNettyDirect(), rocksIterator, range.getMin().retain());
} else { } else {
rocksIterator.seekToFirst(); rocksIterator.seekToFirst();
} }
@ -2010,20 +2061,28 @@ public class LLLocalDictionary implements LLDictionary {
try (var readOpts = new ReadOptions(getReadOptions(null))) { try (var readOpts = new ReadOptions(getReadOptions(null))) {
ReleasableSlice minBound; ReleasableSlice minBound;
if (range.hasMin()) { if (range.hasMin()) {
minBound = setIterateBound(readOpts, IterateBound.LOWER, range.getMin().retain()); minBound = setIterateBound(databaseOptions.allowNettyDirect(),
readOpts,
IterateBound.LOWER,
range.getMin().retain()
);
} else { } else {
minBound = emptyReleasableSlice(); minBound = emptyReleasableSlice();
} }
try { try {
ReleasableSlice maxBound; ReleasableSlice maxBound;
if (range.hasMax()) { if (range.hasMax()) {
maxBound = setIterateBound(readOpts, IterateBound.UPPER, range.getMax().retain()); maxBound = setIterateBound(databaseOptions.allowNettyDirect(),
readOpts,
IterateBound.UPPER,
range.getMax().retain()
);
} else { } else {
maxBound = emptyReleasableSlice(); maxBound = emptyReleasableSlice();
} }
try (RocksIterator rocksIterator = db.newIterator(cfh, readOpts)) { try (RocksIterator rocksIterator = db.newIterator(cfh, readOpts)) {
if (!LLLocalDictionary.PREFER_SEEK_TO_FIRST && range.hasMin()) { if (!LLLocalDictionary.PREFER_SEEK_TO_FIRST && range.hasMin()) {
rocksIterSeekTo(rocksIterator, range.getMin().retain()); rocksIterSeekTo(databaseOptions.allowNettyDirect(), rocksIterator, range.getMin().retain());
} else { } else {
rocksIterator.seekToFirst(); rocksIterator.seekToFirst();
} }
@ -2053,7 +2112,8 @@ public class LLLocalDictionary implements LLDictionary {
} }
@NotNull @NotNull
public static Tuple3<RocksIterator, ReleasableSlice, ReleasableSlice> getRocksIterator(ReadOptions readOptions, public static Tuple3<RocksIterator, ReleasableSlice, ReleasableSlice> getRocksIterator(boolean allowNettyDirect,
ReadOptions readOptions,
LLRange range, LLRange range,
RocksDB db, RocksDB db,
ColumnFamilyHandle cfh) { ColumnFamilyHandle cfh) {
@ -2061,18 +2121,18 @@ public class LLLocalDictionary implements LLDictionary {
ReleasableSlice sliceMin; ReleasableSlice sliceMin;
ReleasableSlice sliceMax; ReleasableSlice sliceMax;
if (range.hasMin()) { if (range.hasMin()) {
sliceMin = setIterateBound(readOptions, IterateBound.LOWER, range.getMin().retain()); sliceMin = setIterateBound(allowNettyDirect, readOptions, IterateBound.LOWER, range.getMin().retain());
} else { } else {
sliceMin = emptyReleasableSlice(); sliceMin = emptyReleasableSlice();
} }
if (range.hasMax()) { if (range.hasMax()) {
sliceMax = setIterateBound(readOptions, IterateBound.UPPER, range.getMax().retain()); sliceMax = setIterateBound(allowNettyDirect, readOptions, IterateBound.UPPER, range.getMax().retain());
} else { } else {
sliceMax = emptyReleasableSlice(); sliceMax = emptyReleasableSlice();
} }
var rocksIterator = db.newIterator(cfh, readOptions); var rocksIterator = db.newIterator(cfh, readOptions);
if (!PREFER_SEEK_TO_FIRST && range.hasMin()) { if (!PREFER_SEEK_TO_FIRST && range.hasMin()) {
rocksIterSeekTo(rocksIterator, range.getMin().retain()); rocksIterSeekTo(allowNettyDirect, rocksIterator, range.getMin().retain());
} else { } else {
rocksIterator.seekToFirst(); rocksIterator.seekToFirst();
} }

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

@ -15,9 +15,10 @@ public class LLLocalEntryReactiveRocksIterator extends LLLocalReactiveRocksItera
ByteBufAllocator alloc, ByteBufAllocator alloc,
ColumnFamilyHandle cfh, ColumnFamilyHandle cfh,
LLRange range, LLRange range,
boolean allowNettyDirect,
ReadOptions readOptions, ReadOptions readOptions,
String debugName) { String debugName) {
super(db, alloc, cfh, range, readOptions, true, debugName); super(db, alloc, cfh, range, allowNettyDirect, readOptions, true, debugName);
} }
@Override @Override

3
src/main/java/it/cavallium/dbengine/database/disk/LLLocalGroupedEntryReactiveRocksIterator.java
View File

@ -15,9 +15,10 @@ public class LLLocalGroupedEntryReactiveRocksIterator extends
public LLLocalGroupedEntryReactiveRocksIterator(RocksDB db, ByteBufAllocator alloc, ColumnFamilyHandle cfh, public LLLocalGroupedEntryReactiveRocksIterator(RocksDB db, ByteBufAllocator alloc, ColumnFamilyHandle cfh,
int prefixLength, int prefixLength,
LLRange range, LLRange range,
boolean allowNettyDirect,
ReadOptions readOptions, ReadOptions readOptions,
String debugName) { String debugName) {
super(db, alloc, cfh, prefixLength, range, readOptions, false, true); super(db, alloc, cfh, prefixLength, range, allowNettyDirect, readOptions, false, true);
} }
@Override @Override

3
src/main/java/it/cavallium/dbengine/database/disk/LLLocalGroupedKeyReactiveRocksIterator.java
View File

@ -14,9 +14,10 @@ public class LLLocalGroupedKeyReactiveRocksIterator extends LLLocalGroupedReacti
ColumnFamilyHandle cfh, ColumnFamilyHandle cfh,
int prefixLength, int prefixLength,
LLRange range, LLRange range,
boolean allowNettyDirect,
ReadOptions readOptions, ReadOptions readOptions,
String debugName) { String debugName) {
super(db, alloc, cfh, prefixLength, range, readOptions, true, false); super(db, alloc, cfh, prefixLength, range, allowNettyDirect, readOptions, true, false);
} }
@Override @Override

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

@ -24,6 +24,7 @@ public abstract class LLLocalGroupedReactiveRocksIterator<T> {
private final ColumnFamilyHandle cfh; private final ColumnFamilyHandle cfh;
private final int prefixLength; private final int prefixLength;
private final LLRange range; private final LLRange range;
private final boolean allowNettyDirect;
private final ReadOptions readOptions; private final ReadOptions readOptions;
private final boolean canFillCache; private final boolean canFillCache;
private final boolean readValues; private final boolean readValues;
@ -31,6 +32,7 @@ public abstract class LLLocalGroupedReactiveRocksIterator<T> {
public LLLocalGroupedReactiveRocksIterator(RocksDB db, ByteBufAllocator alloc, ColumnFamilyHandle cfh, public LLLocalGroupedReactiveRocksIterator(RocksDB db, ByteBufAllocator alloc, ColumnFamilyHandle cfh,
int prefixLength, int prefixLength,
LLRange range, LLRange range,
boolean allowNettyDirect,
ReadOptions readOptions, ReadOptions readOptions,
boolean canFillCache, boolean canFillCache,
boolean readValues) { boolean readValues) {
@ -39,6 +41,7 @@ public abstract class LLLocalGroupedReactiveRocksIterator<T> {
this.cfh = cfh; this.cfh = cfh;
this.prefixLength = prefixLength; this.prefixLength = prefixLength;
this.range = range; this.range = range;
this.allowNettyDirect = allowNettyDirect;
this.readOptions = readOptions; this.readOptions = readOptions;
this.canFillCache = canFillCache; this.canFillCache = canFillCache;
this.readValues = readValues; this.readValues = readValues;
@ -50,7 +53,7 @@ public abstract class LLLocalGroupedReactiveRocksIterator<T> {
.generate(() -> { .generate(() -> {
var readOptions = new ReadOptions(this.readOptions); var readOptions = new ReadOptions(this.readOptions);
readOptions.setFillCache(canFillCache && range.hasMin() && range.hasMax()); readOptions.setFillCache(canFillCache && range.hasMin() && range.hasMax());
return LLLocalDictionary.getRocksIterator(readOptions, range.retain(), db, cfh); return LLLocalDictionary.getRocksIterator(allowNettyDirect, readOptions, range.retain(), db, cfh);
}, (tuple, sink) -> { }, (tuple, sink) -> {
range.retain(); range.retain();
try { try {

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

@ -21,6 +21,7 @@ public class LLLocalKeyPrefixReactiveRocksIterator {
private final ColumnFamilyHandle cfh; private final ColumnFamilyHandle cfh;
private final int prefixLength; private final int prefixLength;
private final LLRange range; private final LLRange range;
private final boolean allowNettyDirect;
private final ReadOptions readOptions; private final ReadOptions readOptions;
private final boolean canFillCache; private final boolean canFillCache;
private final String debugName; private final String debugName;
@ -28,6 +29,7 @@ public class LLLocalKeyPrefixReactiveRocksIterator {
public LLLocalKeyPrefixReactiveRocksIterator(RocksDB db, ByteBufAllocator alloc, ColumnFamilyHandle cfh, public LLLocalKeyPrefixReactiveRocksIterator(RocksDB db, ByteBufAllocator alloc, ColumnFamilyHandle cfh,
int prefixLength, int prefixLength,
LLRange range, LLRange range,
boolean allowNettyDirect,
ReadOptions readOptions, ReadOptions readOptions,
boolean canFillCache, boolean canFillCache,
String debugName) { String debugName) {
@ -36,6 +38,7 @@ public class LLLocalKeyPrefixReactiveRocksIterator {
this.cfh = cfh; this.cfh = cfh;
this.prefixLength = prefixLength; this.prefixLength = prefixLength;
this.range = range; this.range = range;
this.allowNettyDirect = allowNettyDirect;
this.readOptions = readOptions; this.readOptions = readOptions;
this.canFillCache = canFillCache; this.canFillCache = canFillCache;
this.debugName = debugName; this.debugName = debugName;
@ -50,7 +53,7 @@ public class LLLocalKeyPrefixReactiveRocksIterator {
readOptions.setReadaheadSize(32 * 1024); // 32KiB readOptions.setReadaheadSize(32 * 1024); // 32KiB
readOptions.setFillCache(canFillCache); readOptions.setFillCache(canFillCache);
} }
return LLLocalDictionary.getRocksIterator(readOptions, range.retain(), db, cfh); return LLLocalDictionary.getRocksIterator(allowNettyDirect, readOptions, range.retain(), db, cfh);
}, (tuple, sink) -> { }, (tuple, sink) -> {
range.retain(); range.retain();
try { try {

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

@ -13,9 +13,10 @@ public class LLLocalKeyReactiveRocksIterator extends LLLocalReactiveRocksIterato
ByteBufAllocator alloc, ByteBufAllocator alloc,
ColumnFamilyHandle cfh, ColumnFamilyHandle cfh,
LLRange range, LLRange range,
boolean allowNettyDirect,
ReadOptions readOptions, ReadOptions readOptions,
String debugName) { String debugName) {
super(db, alloc, cfh, range, readOptions, false, debugName); super(db, alloc, cfh, range, allowNettyDirect, readOptions, false, debugName);
} }
@Override @Override

3
src/main/java/it/cavallium/dbengine/database/disk/LLLocalKeyValueDatabase.java
View File

@ -474,7 +474,8 @@ public class LLLocalKeyValueDatabase implements LLKeyValueDatabase {
Column.toString(columnName), Column.toString(columnName),
dbScheduler, dbScheduler,
(snapshot) -> snapshotsHandles.get(snapshot.getSequenceNumber()), (snapshot) -> snapshotsHandles.get(snapshot.getSequenceNumber()),
updateMode updateMode,
databaseOptions
)) ))
.subscribeOn(dbScheduler); .subscribeOn(dbScheduler);
} }

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

@ -29,6 +29,7 @@ public abstract class LLLocalReactiveRocksIterator<T> {
private final ByteBufAllocator alloc; private final ByteBufAllocator alloc;
private final ColumnFamilyHandle cfh; private final ColumnFamilyHandle cfh;
private final LLRange range; private final LLRange range;
private final boolean allowNettyDirect;
private final ReadOptions readOptions; private final ReadOptions readOptions;
private final boolean readValues; private final boolean readValues;
private final String debugName; private final String debugName;
@ -37,6 +38,7 @@ public abstract class LLLocalReactiveRocksIterator<T> {
ByteBufAllocator alloc, ByteBufAllocator alloc,
ColumnFamilyHandle cfh, ColumnFamilyHandle cfh,
LLRange range, LLRange range,
boolean allowNettyDirect,
ReadOptions readOptions, ReadOptions readOptions,
boolean readValues, boolean readValues,
String debugName) { String debugName) {
@ -44,6 +46,7 @@ public abstract class LLLocalReactiveRocksIterator<T> {
this.alloc = alloc; this.alloc = alloc;
this.cfh = cfh; this.cfh = cfh;
this.range = range; this.range = range;
this.allowNettyDirect = allowNettyDirect;
this.readOptions = readOptions; this.readOptions = readOptions;
this.readValues = readValues; this.readValues = readValues;
this.debugName = debugName; this.debugName = debugName;
@ -57,7 +60,7 @@ public abstract class LLLocalReactiveRocksIterator<T> {
readOptions.setReadaheadSize(32 * 1024); // 32KiB readOptions.setReadaheadSize(32 * 1024); // 32KiB
readOptions.setFillCache(false); readOptions.setFillCache(false);
} }
return getRocksIterator(readOptions, range.retain(), db, cfh); return getRocksIterator(allowNettyDirect, readOptions, range.retain(), db, cfh);
}, (tuple, sink) -> { }, (tuple, sink) -> {
range.retain(); range.retain();
try { try {

2
src/test/java/it/cavallium/dbengine/DbTestUtils.java
View File

@ -58,7 +58,7 @@ public class DbTestUtils {
.then(new LLLocalDatabaseConnection(DbTestUtils.ALLOCATOR, wrkspcPath).connect()) .then(new LLLocalDatabaseConnection(DbTestUtils.ALLOCATOR, wrkspcPath).connect())
.flatMap(conn -> conn.getDatabase("testdb", .flatMap(conn -> conn.getDatabase("testdb",
List.of(Column.dictionary("testmap"), Column.special("ints"), Column.special("longs")), List.of(Column.dictionary("testmap"), Column.special("ints"), Column.special("longs")),
new DatabaseOptions(Map.of(), true, false, true, false, true) new DatabaseOptions(Map.of(), true, false, true, false, true, true, true)
)), )),
action, action,
db -> db.close().then(Mono.fromCallable(() -> { db -> db.close().then(Mono.fromCallable(() -> {

2
src/test/java/it/cavallium/dbengine/OldDatabaseTests.java
View File

@ -135,7 +135,7 @@ public class OldDatabaseTests {
.then(new LLLocalDatabaseConnection(PooledByteBufAllocator.DEFAULT, wrkspcPath).connect()) .then(new LLLocalDatabaseConnection(PooledByteBufAllocator.DEFAULT, wrkspcPath).connect())
.flatMap(conn -> conn.getDatabase("testdb", .flatMap(conn -> conn.getDatabase("testdb",
List.of(Column.dictionary("testmap")), List.of(Column.dictionary("testmap")),
new DatabaseOptions(Map.of(), true, false, true, false, true) new DatabaseOptions(Map.of(), true, false, true, false, true, true, true)
)); ));
} }