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

Close iterators

This commit is contained in:
Andrea Cavalli 2023-05-23 01:05:03 +02:00
parent 0062a36ed5
commit bee2fe1bf5
6 changed files with 106 additions and 201 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
src/main/java/it/cavallium/dbengine/database/LLUtils.java
View File

@ -74,7 +74,6 @@ public class LLUtils {
private static final byte[] RESPONSE_FALSE_BUF = new byte[]{0}; private static final byte[] RESPONSE_FALSE_BUF = new byte[]{0};
public static final byte[][] LEXICONOGRAPHIC_ITERATION_SEEKS = new byte[256][1]; public static final byte[][] LEXICONOGRAPHIC_ITERATION_SEEKS = new byte[256][1];
public static final boolean MANUAL_READAHEAD = false; public static final boolean MANUAL_READAHEAD = false;
public static final boolean ALLOW_STATIC_OPTIONS = false;
public static final boolean FORCE_DISABLE_CHECKSUM_VERIFICATION public static final boolean FORCE_DISABLE_CHECKSUM_VERIFICATION
= Boolean.parseBoolean(System.getProperty("it.cavallium.dbengine.checksum.disable.force", "false")); = Boolean.parseBoolean(System.getProperty("it.cavallium.dbengine.checksum.disable.force", "false"));

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

@ -1,6 +1,5 @@
package it.cavallium.dbengine.database.disk; package it.cavallium.dbengine.database.disk;
import static it.cavallium.dbengine.database.LLUtils.ALLOW_STATIC_OPTIONS;
import static it.cavallium.dbengine.database.LLUtils.MARKER_ROCKSDB; import static it.cavallium.dbengine.database.LLUtils.MARKER_ROCKSDB;
import static it.cavallium.dbengine.database.LLUtils.isBoundedRange; import static it.cavallium.dbengine.database.LLUtils.isBoundedRange;
import static it.cavallium.dbengine.database.LLUtils.mapList; import static it.cavallium.dbengine.database.LLUtils.mapList;
@ -32,7 +31,6 @@ import it.cavallium.dbengine.database.SerializedKey;
import it.cavallium.dbengine.database.UpdateMode; import it.cavallium.dbengine.database.UpdateMode;
import it.cavallium.dbengine.database.UpdateReturnMode; import it.cavallium.dbengine.database.UpdateReturnMode;
import it.cavallium.dbengine.database.disk.rocksdb.LLReadOptions; import it.cavallium.dbengine.database.disk.rocksdb.LLReadOptions;
import it.cavallium.dbengine.database.disk.rocksdb.LLSlice;
import it.cavallium.dbengine.database.disk.rocksdb.LLWriteOptions; import it.cavallium.dbengine.database.disk.rocksdb.LLWriteOptions;
import it.cavallium.dbengine.database.serialization.KVSerializationFunction; import it.cavallium.dbengine.database.serialization.KVSerializationFunction;
import it.cavallium.dbengine.database.serialization.SerializationFunction; import it.cavallium.dbengine.database.serialization.SerializationFunction;
@ -57,12 +55,9 @@ import org.jetbrains.annotations.Nullable;
import org.rocksdb.ColumnFamilyHandle; import org.rocksdb.ColumnFamilyHandle;
import org.rocksdb.CompactRangeOptions; import org.rocksdb.CompactRangeOptions;
import org.rocksdb.FlushOptions; import org.rocksdb.FlushOptions;
import org.rocksdb.ReadOptions;
import org.rocksdb.RocksDBException; import org.rocksdb.RocksDBException;
import org.rocksdb.Slice;
import org.rocksdb.Snapshot; import org.rocksdb.Snapshot;
import org.rocksdb.WriteBatch; import org.rocksdb.WriteBatch;
import org.rocksdb.WriteOptions;
public class LLLocalDictionary implements LLDictionary { public class LLLocalDictionary implements LLDictionary {
@ -72,7 +67,6 @@ public class LLLocalDictionary implements LLDictionary {
static final long MAX_WRITE_BATCH_SIZE = 1024L * 1024L * 1024L; // 1GiB static final long MAX_WRITE_BATCH_SIZE = 1024L * 1024L * 1024L; // 1GiB
static final int CAPPED_WRITE_BATCH_CAP = 50000; // 50K operations static final int CAPPED_WRITE_BATCH_CAP = 50000; // 50K operations
static final int MULTI_GET_WINDOW = 16; static final int MULTI_GET_WINDOW = 16;
private static final LLReadOptions EMPTY_READ_OPTIONS = LLUtils.ALLOW_STATIC_OPTIONS ? new LLReadOptions() : null;
static final boolean PREFER_AUTO_SEEK_BOUND = false; static final boolean PREFER_AUTO_SEEK_BOUND = false;
/** /**
* It used to be false, * It used to be false,
@ -103,10 +97,6 @@ public class LLLocalDictionary implements LLDictionary {
private static final boolean USE_NUM_ENTRIES_PRECISE_COUNTER = true; private static final boolean USE_NUM_ENTRIES_PRECISE_COUNTER = true;
private static final byte[] FIRST_KEY = new byte[]{}; private static final byte[] FIRST_KEY = new byte[]{};
/**
* 1KiB dummy buffer, write only, used for debugging purposes
*/
private static final ByteBuffer DUMMY_WRITE_ONLY_BYTE_BUFFER = ByteBuffer.allocateDirect(1024);
private final RocksDBColumn db; private final RocksDBColumn db;
private final ColumnFamilyHandle cfh; private final ColumnFamilyHandle cfh;
@ -198,38 +188,16 @@ public class LLLocalDictionary implements LLDictionary {
return columnName; return columnName;
} }
@NotNull
private LLReadOptions generateReadOptionsOrStatic(LLSnapshot snapshot) {
var resolved = generateReadOptions(snapshot != null ? snapshotResolver.apply(snapshot) : null, true);
if (resolved != null) {
return resolved;
} else {
return new LLReadOptions();
}
}
@Nullable
private LLReadOptions generateReadOptionsOrNull(LLSnapshot snapshot) {
return generateReadOptions(snapshot != null ? snapshotResolver.apply(snapshot) : null, false);
}
@NotNull @NotNull
private LLReadOptions generateReadOptionsOrNew(LLSnapshot snapshot) { private LLReadOptions generateReadOptionsOrNew(LLSnapshot snapshot) {
var result = generateReadOptions(snapshot != null ? snapshotResolver.apply(snapshot) : null, false); return generateReadOptions(snapshot != null ? snapshotResolver.apply(snapshot) : null);
if (result != null) {
return result;
} else {
return new LLReadOptions();
}
} }
private LLReadOptions generateReadOptions(Snapshot snapshot, boolean orStaticOpts) { private LLReadOptions generateReadOptions(Snapshot snapshot) {
if (snapshot != null) { if (snapshot != null) {
return new LLReadOptions().setSnapshot(snapshot); return new LLReadOptions().setSnapshot(snapshot);
} else if (ALLOW_STATIC_OPTIONS && orStaticOpts) {
return EMPTY_READ_OPTIONS;
} else { } else {
return null; return new LLReadOptions();
} }
} }
@ -241,18 +209,14 @@ public class LLLocalDictionary implements LLDictionary {
private Buf getSync(LLSnapshot snapshot, Buf key) { private Buf getSync(LLSnapshot snapshot, Buf key) {
logger.trace(MARKER_ROCKSDB, "Reading {}", () -> toStringSafe(key)); logger.trace(MARKER_ROCKSDB, "Reading {}", () -> toStringSafe(key));
try { try {
var readOptions = generateReadOptionsOrStatic(snapshot);
Buf result; Buf result;
startedGet.increment(); startedGet.increment();
try { try (var readOptions = generateReadOptionsOrNew(snapshot)) {
var initTime = System.nanoTime(); var initTime = System.nanoTime();
result = db.get(readOptions, key); result = db.get(readOptions, key);
getTime.record(Duration.ofNanos(System.nanoTime() - initTime)); getTime.record(Duration.ofNanos(System.nanoTime() - initTime));
} finally { } finally {
endedGet.increment(); endedGet.increment();
if (readOptions != EMPTY_READ_OPTIONS) {
readOptions.close();
}
} }
logger.trace(MARKER_ROCKSDB, "Read {}: {}", () -> toStringSafe(key), () -> toStringSafe(result)); logger.trace(MARKER_ROCKSDB, "Read {}: {}", () -> toStringSafe(key), () -> toStringSafe(result));
return result; return result;
@ -272,7 +236,7 @@ public class LLLocalDictionary implements LLDictionary {
} else { } else {
// Temporary resources to release after finished // Temporary resources to release after finished
try (var readOpts = LLUtils.generateCustomReadOptions(generateReadOptionsOrNull(snapshot), try (var readOpts = LLUtils.generateCustomReadOptions(generateReadOptionsOrNew(snapshot),
true, true,
isBoundedRange(range), isBoundedRange(range),
true true
@ -304,13 +268,8 @@ public class LLLocalDictionary implements LLDictionary {
startedContains.increment(); startedContains.increment();
try { try {
var result = containsTime.recordCallable(() -> { var result = containsTime.recordCallable(() -> {
var readOptions = generateReadOptionsOrStatic(snapshot); try (var readOptions = generateReadOptionsOrNew(snapshot)) {
try {
return db.exists(readOptions, key); return db.exists(readOptions, key);
} finally {
if (readOptions != EMPTY_READ_OPTIONS) {
readOptions.close();
}
} }
}); });
assert result != null; assert result != null;
@ -375,15 +334,12 @@ public class LLLocalDictionary implements LLDictionary {
}; };
UpdateAtomicResult result = null; UpdateAtomicResult result = null;
try { try {
var readOptions = generateReadOptionsOrStatic(null);
startedUpdates.increment(); startedUpdates.increment();
try (var writeOptions = new LLWriteOptions()) { try (var readOptions = generateReadOptionsOrNew(null);
var writeOptions = new LLWriteOptions()) {
result = updateTime.recordCallable(() -> db.updateAtomic(readOptions, writeOptions, key, updater, returnMode)); result = updateTime.recordCallable(() -> db.updateAtomic(readOptions, writeOptions, key, updater, returnMode));
} finally { } finally {
endedUpdates.increment(); endedUpdates.increment();
if (readOptions != EMPTY_READ_OPTIONS) {
readOptions.close();
}
} }
assert result != null; assert result != null;
return switch (updateReturnMode) { return switch (updateReturnMode) {
@ -410,18 +366,19 @@ public class LLLocalDictionary implements LLDictionary {
+ "safe atomic operations"); + "safe atomic operations");
} }
UpdateAtomicResultDelta result = null; UpdateAtomicResultDelta result;
try { try {
var readOptions = generateReadOptionsOrStatic(null);
startedUpdates.increment(); startedUpdates.increment();
try (var writeOptions = new LLWriteOptions()) { try (var readOptions = generateReadOptionsOrNew(null);
result = updateTime.recordCallable(() -> var writeOptions = new LLWriteOptions()) {
(UpdateAtomicResultDelta) db.updateAtomic(readOptions, writeOptions, key, updater, DELTA)); result = updateTime.recordCallable(() -> (UpdateAtomicResultDelta) db.updateAtomic(readOptions,
writeOptions,
key,
updater,
DELTA
));
} finally { } finally {
endedUpdates.increment(); endedUpdates.increment();
if (readOptions != EMPTY_READ_OPTIONS) {
readOptions.close();
}
} }
assert result != null; assert result != null;
return result.delta(); return result.delta();
@ -462,13 +419,8 @@ public class LLLocalDictionary implements LLDictionary {
case PREVIOUS_VALUE -> { case PREVIOUS_VALUE -> {
assert !LLUtils.isInNonBlockingThread() : "Called getPreviousData in a nonblocking thread"; assert !LLUtils.isInNonBlockingThread() : "Called getPreviousData in a nonblocking thread";
Buf result; Buf result;
var readOptions = generateReadOptionsOrStatic(null); try (var readOptions = generateReadOptionsOrNew(null)) {
try {
result = db.get(readOptions, key); result = db.get(readOptions, key);
} finally {
if (readOptions != EMPTY_READ_OPTIONS) {
readOptions.close();
}
} }
logger.trace(MARKER_ROCKSDB, "Read {}: {}", () -> toStringSafe(key), () -> toStringSafe(result)); logger.trace(MARKER_ROCKSDB, "Read {}: {}", () -> toStringSafe(key), () -> toStringSafe(result));
yield result; yield result;
@ -532,31 +484,24 @@ public class LLLocalDictionary implements LLDictionary {
} }
ArrayList<MappedInput<K>> mappedInputs; ArrayList<MappedInput<K>> mappedInputs;
{ {
var readOptions = generateReadOptionsOrStatic(null); try (var readOptions = generateReadOptionsOrNew(null)) {
try {
var inputs = db.multiGetAsList(readOptions, mapList(keyBufsWindow, Buf::asArray)); var inputs = db.multiGetAsList(readOptions, mapList(keyBufsWindow, Buf::asArray));
mappedInputs = new ArrayList<>(inputs.size()); mappedInputs = new ArrayList<>(inputs.size());
for (int i = 0; i < inputs.size(); i++) { for (int i = 0; i < inputs.size(); i++) {
var val = inputs.get(i); var val = inputs.get(i);
if (val != null) { if (val != null) {
inputs.set(i, null); inputs.set(i, null);
mappedInputs.add(new MappedInput<>( mappedInputs.add(new MappedInput<>(entriesWindow.get(i).key(),
entriesWindow.get(i).key(),
keyBufsWindow.get(i), keyBufsWindow.get(i),
OptionalBuf.of(Buf.wrap(val)) OptionalBuf.of(Buf.wrap(val))
)); ));
} else { } else {
mappedInputs.add(new MappedInput<>( mappedInputs.add(new MappedInput<>(entriesWindow.get(i).key(),
entriesWindow.get(i).key(),
keyBufsWindow.get(i), keyBufsWindow.get(i),
OptionalBuf.empty() OptionalBuf.empty()
)); ));
} }
} }
} finally {
if (readOptions != EMPTY_READ_OPTIONS) {
readOptions.close();
}
} }
} }
var updatedValuesToWrite = new ArrayList<Buf>(mappedInputs.size()); var updatedValuesToWrite = new ArrayList<Buf>(mappedInputs.size());
@ -640,7 +585,7 @@ public class LLLocalDictionary implements LLDictionary {
boolean smallRange) { boolean smallRange) {
return new LLLocalEntryReactiveRocksIterator(db, return new LLLocalEntryReactiveRocksIterator(db,
range, range,
() -> generateReadOptionsOrNull(snapshot), () -> generateReadOptionsOrNew(snapshot),
reverse, reverse,
smallRange smallRange
).stream(); ).stream();
@ -651,7 +596,7 @@ public class LLLocalDictionary implements LLDictionary {
return new LLLocalGroupedEntryReactiveRocksIterator(db, return new LLLocalGroupedEntryReactiveRocksIterator(db,
prefixLength, prefixLength,
range, range,
() -> generateReadOptionsOrNull(snapshot), () -> generateReadOptionsOrNew(snapshot),
smallRange smallRange
).stream(); ).stream();
} }
@ -676,7 +621,7 @@ public class LLLocalDictionary implements LLDictionary {
return new LLLocalGroupedKeyReactiveRocksIterator(db, return new LLLocalGroupedKeyReactiveRocksIterator(db,
prefixLength, prefixLength,
range, range,
() -> generateReadOptionsOrNull(snapshot), () -> generateReadOptionsOrNew(snapshot),
smallRange smallRange
).stream(); ).stream();
} }
@ -729,7 +674,7 @@ public class LLLocalDictionary implements LLDictionary {
return new LLLocalKeyPrefixReactiveRocksIterator(db, return new LLLocalKeyPrefixReactiveRocksIterator(db,
prefixLength, prefixLength,
range, range,
() -> generateReadOptionsOrNull(snapshot), () -> generateReadOptionsOrNew(snapshot),
true, true,
smallRange smallRange
).stream(); ).stream();
@ -753,7 +698,7 @@ public class LLLocalDictionary implements LLDictionary {
boolean smallRange) { boolean smallRange) {
return new LLLocalKeyReactiveRocksIterator(db, return new LLLocalKeyReactiveRocksIterator(db,
range, range,
() -> generateReadOptionsOrNull(snapshot), () -> generateReadOptionsOrNew(snapshot),
reverse, reverse,
smallRange smallRange
).stream(); ).stream();
@ -950,7 +895,7 @@ public class LLLocalDictionary implements LLDictionary {
if (range.isAll()) { if (range.isAll()) {
return fast ? fastSizeAll(snapshot) : exactSizeAll(snapshot); return fast ? fastSizeAll(snapshot) : exactSizeAll(snapshot);
} else { } else {
try (var readOpts = LLUtils.generateCustomReadOptions(generateReadOptionsOrNull(snapshot), try (var readOpts = LLUtils.generateCustomReadOptions(generateReadOptionsOrNew(snapshot),
false, false,
isBoundedRange(range), isBoundedRange(range),
false false
@ -985,7 +930,7 @@ public class LLLocalDictionary implements LLDictionary {
public LLEntry getOne(@Nullable LLSnapshot snapshot, LLRange range) { public LLEntry getOne(@Nullable LLSnapshot snapshot, LLRange range) {
try { try {
assert !LLUtils.isInNonBlockingThread() : "Called getOne in a nonblocking thread"; assert !LLUtils.isInNonBlockingThread() : "Called getOne in a nonblocking thread";
try (var readOpts = LLUtils.generateCustomReadOptions(generateReadOptionsOrNull(snapshot), true, true, true)) { try (var readOpts = LLUtils.generateCustomReadOptions(generateReadOptionsOrNew(snapshot), true, true, true)) {
try (var rocksIterator = db.newIterator(readOpts, range.getMin(), range.getMax())) { try (var rocksIterator = db.newIterator(readOpts, range.getMin(), range.getMax())) {
if (!LLLocalDictionary.PREFER_AUTO_SEEK_BOUND && range.hasMin()) { if (!LLLocalDictionary.PREFER_AUTO_SEEK_BOUND && range.hasMin()) {
rocksIterator.seekTo(range.getMin()); rocksIterator.seekTo(range.getMin());
@ -1010,7 +955,7 @@ public class LLLocalDictionary implements LLDictionary {
public Buf getOneKey(@Nullable LLSnapshot snapshot, LLRange range) { public Buf getOneKey(@Nullable LLSnapshot snapshot, LLRange range) {
try { try {
assert !LLUtils.isInNonBlockingThread() : "Called getOneKey in a nonblocking thread"; assert !LLUtils.isInNonBlockingThread() : "Called getOneKey in a nonblocking thread";
try (var readOpts = LLUtils.generateCustomReadOptions(generateReadOptionsOrNull(snapshot), true, true, true)) { try (var readOpts = LLUtils.generateCustomReadOptions(generateReadOptionsOrNew(snapshot), true, true, true)) {
try (var rocksIterator = db.newIterator(readOpts, range.getMin(), range.getMax())) { try (var rocksIterator = db.newIterator(readOpts, range.getMin(), range.getMax())) {
if (!LLLocalDictionary.PREFER_AUTO_SEEK_BOUND && range.hasMin()) { if (!LLLocalDictionary.PREFER_AUTO_SEEK_BOUND && range.hasMin()) {
rocksIterator.seekTo(range.getMin()); rocksIterator.seekTo(range.getMin());
@ -1075,7 +1020,7 @@ public class LLLocalDictionary implements LLDictionary {
if (LLUtils.isInNonBlockingThread()) { if (LLUtils.isInNonBlockingThread()) {
throw new UnsupportedOperationException("Called exactSizeAll in a nonblocking thread"); throw new UnsupportedOperationException("Called exactSizeAll in a nonblocking thread");
} }
try (var readOpts = LLUtils.generateCustomReadOptions(generateReadOptionsOrNull(snapshot), false, false, false)) { try (var readOpts = LLUtils.generateCustomReadOptions(generateReadOptionsOrNew(snapshot), false, false, false)) {
if (LLUtils.MANUAL_READAHEAD) { if (LLUtils.MANUAL_READAHEAD) {
readOpts.setReadaheadSize(128 * 1024); // 128KiB readOpts.setReadaheadSize(128 * 1024); // 128KiB
} }

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

@ -53,21 +53,12 @@ public abstract class LLLocalGroupedReactiveRocksIterator<T> {
} }
public final Stream<List<T>> stream() { public final Stream<List<T>> stream() {
var readOptions = generateCustomReadOptions(this.readOptions.get(), canFillCache, isBoundedRange(range), smallRange);
if (logger.isTraceEnabled()) {
logger.trace(MARKER_ROCKSDB, "Range {} started", LLUtils.toStringSafe(range));
}
RocksIteratorObj rocksIterator;
try {
rocksIterator = db.newRocksIterator(readOptions, range, false);
} catch (RocksDBException e) {
readOptions.close();
throw new DBException("Failed to iterate the range", e);
}
return StreamUtils.<List<T>>streamWhileNonNull(() -> { return StreamUtils.<List<T>>streamWhileNonNull(() -> {
try { try (var readOptions = generateCustomReadOptions(this.readOptions.get(), canFillCache, isBoundedRange(range), smallRange);
var rocksIterator = db.newRocksIterator(readOptions, range, false)) {
if (logger.isTraceEnabled()) {
logger.trace(MARKER_ROCKSDB, "Range {} started", LLUtils.toStringSafe(range));
}
ObjectArrayList<T> values = new ObjectArrayList<>(); ObjectArrayList<T> values = new ObjectArrayList<>();
Buf firstGroupKey = null; Buf firstGroupKey = null;
while (rocksIterator.isValid()) { while (rocksIterator.isValid()) {
@ -113,9 +104,6 @@ public abstract class LLLocalGroupedReactiveRocksIterator<T> {
} }
throw new CompletionException(new DBException("Range failed", ex)); throw new CompletionException(new DBException("Range failed", ex));
} }
}).onClose(() -> {
rocksIterator.close();
readOptions.close();
}); });
} }

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

@ -50,71 +50,61 @@ public class LLLocalKeyPrefixReactiveRocksIterator {
public Stream<Buf> stream() { public Stream<Buf> stream() {
try { return streamWhileNonNull(() -> {
var readOptions try (var readOptions = generateCustomReadOptions(this.readOptions.get(), canFillCache, isBoundedRange(range), smallRange);
= generateCustomReadOptions(this.readOptions.get(), canFillCache, isBoundedRange(range), smallRange); var rocksIterator = db.newRocksIterator(readOptions, range, false)) {
if (logger.isTraceEnabled()) { if (logger.isTraceEnabled()) {
logger.trace(MARKER_ROCKSDB, "Range {} started", LLUtils.toStringSafe(range)); logger.trace(MARKER_ROCKSDB, "Range {} started", LLUtils.toStringSafe(range));
}
var rocksIterator = db.newRocksIterator(readOptions, range, false);
return streamWhileNonNull(() -> {
try {
Buf firstGroupKey = null;
while (rocksIterator.isValid()) {
// Note that the underlying array is subject to changes!
Buf key = rocksIterator.keyBuf();
var keyLen = key.size();
if (keyLen >= prefixLength) {
if (firstGroupKey == null) {
firstGroupKey = key.copy();
assert firstGroupKey == null || firstGroupKey.size() >= prefixLength;
} else if (!LLUtils.equals(firstGroupKey,
0,
key,
0,
prefixLength
)) {
break;
}
} else {
logger.error("Skipped a key with length {}, the expected minimum prefix key length is {}!"
+ " This key will be dropped", key.size(), prefixLength);
}
rocksIterator.next();
}
if (firstGroupKey != null) {
var groupKeyPrefix = firstGroupKey.subList(0, prefixLength);
if (logger.isTraceEnabled()) {
logger.trace(MARKER_ROCKSDB,
"Range {} is reading prefix {}",
LLUtils.toStringSafe(range),
LLUtils.toStringSafe(groupKeyPrefix)
);
}
return groupKeyPrefix;
} else {
if (logger.isTraceEnabled()) {
logger.trace(MARKER_ROCKSDB, "Range {} ended", LLUtils.toStringSafe(range));
}
return null;
}
} catch (RocksDBException ex) {
if (logger.isTraceEnabled()) {
logger.trace(MARKER_ROCKSDB, "Range {} failed", LLUtils.toStringSafe(range));
}
throw new CompletionException(new DBException("Range failed", ex));
} }
}).onClose(() -> { Buf firstGroupKey = null;
rocksIterator.close(); while (rocksIterator.isValid()) {
readOptions.close(); // Note that the underlying array is subject to changes!
}); Buf key = rocksIterator.keyBuf();
} catch (RocksDBException e) { var keyLen = key.size();
throw new DBException("Failed to open stream", e); if (keyLen >= prefixLength) {
} if (firstGroupKey == null) {
firstGroupKey = key.copy();
assert firstGroupKey == null || firstGroupKey.size() >= prefixLength;
} else if (!LLUtils.equals(firstGroupKey,
0,
key,
0,
prefixLength
)) {
break;
}
} else {
logger.error("Skipped a key with length {}, the expected minimum prefix key length is {}!"
+ " This key will be dropped", key.size(), prefixLength);
}
rocksIterator.next();
}
if (firstGroupKey != null) {
var groupKeyPrefix = firstGroupKey.subList(0, prefixLength);
if (logger.isTraceEnabled()) {
logger.trace(MARKER_ROCKSDB,
"Range {} is reading prefix {}",
LLUtils.toStringSafe(range),
LLUtils.toStringSafe(groupKeyPrefix)
);
}
return groupKeyPrefix;
} else {
if (logger.isTraceEnabled()) {
logger.trace(MARKER_ROCKSDB, "Range {} ended", LLUtils.toStringSafe(range));
}
return null;
}
} catch (RocksDBException ex) {
if (logger.isTraceEnabled()) {
logger.trace(MARKER_ROCKSDB, "Range {} failed", LLUtils.toStringSafe(range));
}
throw new CompletionException(new DBException("Range failed", ex));
}
});
} }
} }

29
src/main/java/it/cavallium/dbengine/database/disk/LLLocalMigrationReactiveRocksIterator.java
View File

@ -34,25 +34,20 @@ public final class LLLocalMigrationReactiveRocksIterator {
} }
public Stream<LLEntry> stream() { public Stream<LLEntry> stream() {
var readOptions = generateCustomReadOptions(this.readOptions.get(), false, false, false);
RocksIteratorObj rocksIterator;
try {
rocksIterator = db.newRocksIterator(readOptions, range, false);
} catch (RocksDBException e) {
throw new DBException("Failed to open iterator", e);
}
return streamWhileNonNull(() -> { return streamWhileNonNull(() -> {
if (rocksIterator.isValid()) { try (var readOptions = generateCustomReadOptions(this.readOptions.get(), false, false, false);
var key = rocksIterator.keyBuf().copy(); var rocksIterator = db.newRocksIterator(readOptions, range, false)) {
var value = rocksIterator.valueBuf().copy(); if (rocksIterator.isValid()) {
rocksIterator.next(false); var key = rocksIterator.keyBuf().copy();
return LLEntry.of(key, value); var value = rocksIterator.valueBuf().copy();
} else { rocksIterator.next(false);
return null; return LLEntry.of(key, value);
} else {
return null;
}
} catch (RocksDBException e) {
throw new DBException("Failed to open iterator", e);
} }
}).onClose(() -> {
rocksIterator.close();
readOptions.close();
}); });
} }
} }

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

@ -48,21 +48,12 @@ public abstract class LLLocalReactiveRocksIterator<T> {
} }
public final Stream<T> stream() { public final Stream<T> stream() {
var readOptions = generateCustomReadOptions(this.readOptions.get(), true, isBoundedRange(range), smallRange);
if (logger.isTraceEnabled()) {
logger.trace(MARKER_ROCKSDB, "Range {} started", LLUtils.toStringSafe(range));
}
RocksIteratorObj rocksIterator;
try {
rocksIterator = db.newRocksIterator(readOptions, range, reverse);
} catch (RocksDBException e) {
readOptions.close();
throw new DBException("Failed to iterate the range", e);
}
return streamWhileNonNull(() -> { return streamWhileNonNull(() -> {
try { try (var readOptions = generateCustomReadOptions(this.readOptions.get(), true, isBoundedRange(range), smallRange);
var rocksIterator = db.newRocksIterator(readOptions, range, reverse)) {
if (logger.isTraceEnabled()) {
logger.trace(MARKER_ROCKSDB, "Range {} started", LLUtils.toStringSafe(range));
}
if (rocksIterator.isValid()) { if (rocksIterator.isValid()) {
// Note that the underlying array is subject to changes! // Note that the underlying array is subject to changes!
Buf key; Buf key;
@ -102,9 +93,6 @@ public abstract class LLLocalReactiveRocksIterator<T> {
} }
throw new CompletionException(ex); throw new CompletionException(ex);
} }
}).onClose(() -> {
rocksIterator.close();
readOptions.close();
}); });
} }