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

Optimized performance and added useful utility classes

Browse Source
This commit is contained in:
Andrea Cavalli 2021-05-11 21:59:05 +02:00
parent 29b23152ec
commit 2f404c477e
12 changed files with 187 additions and 44 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
src/main/java/it/cavallium/dbengine/client/CastMapper.java Normal file
View File

@ -0,0 +1,16 @@
package it.cavallium.dbengine.client;
import it.cavallium.dbengine.client.Mapper;
public class CastMapper<T, U> implements Mapper<T, U> {
@Override
public U map(T key) {
return (U) key;
}
@Override
public T unmap(U key) {
return (T) key;
}
}

4
src/main/java/it/cavallium/dbengine/client/LuceneIndexImpl.java
View File

@ -50,7 +50,7 @@ public class LuceneIndexImpl<T, U> implements LuceneIndex<T, U> {
.flatMap(entry -> indicizer
.toDocument(entry.getKey(), entry.getValue())
.map(doc -> Map.entry(indicizer.toIndex(entry.getKey()), doc)))
.groupBy(Entry::getKey, Entry::getValue)
.collectMap(Entry::getKey, Entry::getValue)
);
}
@ -74,7 +74,7 @@ public class LuceneIndexImpl<T, U> implements LuceneIndex<T, U> {
.flatMap(entry -> indicizer
.toDocument(entry.getKey(), entry.getValue())
.map(doc -> Map.entry(indicizer.toIndex(entry.getKey()), doc)))
.groupBy(Entry::getKey, Entry::getValue)
.collectMap(Entry::getKey, Entry::getValue)
);
}

31
src/main/java/it/cavallium/dbengine/client/MappedSerializer.java Normal file
View File

@ -0,0 +1,31 @@
package it.cavallium.dbengine.client;
import io.netty.buffer.ByteBuf;
import it.cavallium.dbengine.database.serialization.Serializer;
import org.jetbrains.annotations.NotNull;
public class MappedSerializer<A, B> implements Serializer<B, ByteBuf> {
private final Serializer<A, ByteBuf> serializer;
private final Mapper<A, B> keyMapper;
public MappedSerializer(Serializer<A, ByteBuf> serializer,
Mapper<A, B> keyMapper) {
this.serializer = serializer;
this.keyMapper = keyMapper;
}
@Override
public @NotNull B deserialize(@NotNull ByteBuf serialized) {
try {
return keyMapper.map(serializer.deserialize(serialized.retain()));
} finally {
serialized.release();
}
}
@Override
public @NotNull ByteBuf serialize(@NotNull B deserialized) {
return serializer.serialize(keyMapper.unmap(deserialized));
}
}

36
src/main/java/it/cavallium/dbengine/client/MappedSerializerFixedLength.java Normal file
View File

@ -0,0 +1,36 @@
package it.cavallium.dbengine.client;
import io.netty.buffer.ByteBuf;
import it.cavallium.dbengine.database.serialization.SerializerFixedBinaryLength;
import org.jetbrains.annotations.NotNull;
public class MappedSerializerFixedLength<A, B> implements SerializerFixedBinaryLength<B, ByteBuf> {
private final SerializerFixedBinaryLength<A, ByteBuf> fixedLengthSerializer;
private final Mapper<A, B> keyMapper;
public MappedSerializerFixedLength(SerializerFixedBinaryLength<A, ByteBuf> fixedLengthSerializer,
Mapper<A, B> keyMapper) {
this.fixedLengthSerializer = fixedLengthSerializer;
this.keyMapper = keyMapper;
}
@Override
public @NotNull B deserialize(@NotNull ByteBuf serialized) {
try {
return keyMapper.map(fixedLengthSerializer.deserialize(serialized.retain()));
} finally {
serialized.release();
}
}
@Override
public @NotNull ByteBuf serialize(@NotNull B deserialized) {
return fixedLengthSerializer.serialize(keyMapper.unmap(deserialized));
}
@Override
public int getSerializedBinaryLength() {
return fixedLengthSerializer.getSerializedBinaryLength();
}
}

8
src/main/java/it/cavallium/dbengine/client/Mapper.java Normal file
View File

@ -0,0 +1,8 @@
package it.cavallium.dbengine.client;
public interface Mapper<T, V> {
V map(T key);
T unmap(V key);
}

16
src/main/java/it/cavallium/dbengine/client/NoMapper.java Normal file
View File

@ -0,0 +1,16 @@
package it.cavallium.dbengine.client;
import it.cavallium.dbengine.client.Mapper;
public class NoMapper<T> implements Mapper<T, T> {
@Override
public T map(T key) {
return key;
}
@Override
public T unmap(T key) {
return key;
}
}

6
src/main/java/it/cavallium/dbengine/database/LLLuceneIndex.java
View File

@ -6,6 +6,8 @@ import it.cavallium.dbengine.client.query.current.data.Query;
import it.cavallium.dbengine.client.query.current.data.QueryParams;
import it.cavallium.dbengine.client.query.current.data.ScoreMode;
import it.cavallium.dbengine.lucene.LuceneUtils;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import org.jetbrains.annotations.Nullable;
import reactor.core.publisher.Flux;
@ -19,13 +21,13 @@ public interface LLLuceneIndex extends LLSnapshottable {
Mono<Void> addDocument(LLTerm id, LLDocument doc);
Mono<Void> addDocuments(Flux<GroupedFlux<LLTerm, LLDocument>> documents);
Mono<Void> addDocuments(Mono<Map<LLTerm, LLDocument>> documents);
Mono<Void> deleteDocument(LLTerm id);
Mono<Void> updateDocument(LLTerm id, LLDocument document);
Mono<Void> updateDocuments(Flux<GroupedFlux<LLTerm, LLDocument>> documents);
Mono<Void> updateDocuments(Mono<Map<LLTerm, LLDocument>> documents);
Mono<Void> deleteAll();

4
src/main/java/it/cavallium/dbengine/database/collections/DatabaseSetDictionary.java
View File

@ -12,12 +12,12 @@ import org.jetbrains.annotations.Nullable;
import reactor.core.publisher.Mono;
@SuppressWarnings("unused")
public class DatabaseSetDictionary<T> extends DatabaseMapDictionaryDeep<T, Nothing, DatabaseStageEntry<Nothing>> {
public class DatabaseSetDictionary<T> extends DatabaseMapDictionary<T, Nothing> {
protected DatabaseSetDictionary(LLDictionary dictionary,
ByteBuf prefixKey,
SerializerFixedBinaryLength<T, ByteBuf> keySuffixSerializer) {
super(dictionary, prefixKey, keySuffixSerializer, DatabaseEmpty.createSubStageGetter(), 0);
super(dictionary, prefixKey, keySuffixSerializer, DatabaseEmpty.NOTHING_SERIALIZER);
}
public static <T> DatabaseSetDictionary<T> simple(LLDictionary dictionary,

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

@ -72,7 +72,7 @@ public class LLLocalDictionary implements LLDictionary {
static final int MULTI_GET_WINDOW = 500;
static final ReadOptions EMPTY_READ_OPTIONS = new ReadOptions();
static final WriteOptions EMPTY_WRITE_OPTIONS = new WriteOptions();
static final WriteOptions BATCH_WRITE_OPTIONS = new WriteOptions().setLowPri(true);
static final WriteOptions BATCH_WRITE_OPTIONS = new WriteOptions();
static final boolean PREFER_SEEK_TO_FIRST = false;
static final boolean VERIFY_CHECKSUMS_WHEN_NOT_NEEDED = false;
public static final boolean DEBUG_PREFIXES_WHEN_ASSERTIONS_ARE_ENABLED = true;
@ -982,6 +982,7 @@ public class LLLocalDictionary implements LLDictionary {
}
}
})
.subscribeOn(dbScheduler)
// Prepend everything to get previous elements
.transformDeferred(transformer -> {

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

@ -38,6 +38,7 @@ import org.rocksdb.DbPath;
import org.rocksdb.FlushOptions;
import org.rocksdb.LRUCache;
import org.rocksdb.Options;
import org.rocksdb.RateLimiter;
import org.rocksdb.RocksDB;
import org.rocksdb.RocksDBException;
import org.rocksdb.Snapshot;
@ -200,8 +201,12 @@ public class LLLocalKeyValueDatabase implements LLKeyValueDatabase {
options.setCompactionStyle(CompactionStyle.LEVEL);
options.setLevelCompactionDynamicLevelBytes(true);
options.setTargetFileSizeBase(64 * 1024 * 1024); // 64MiB sst file
options.setMaxBytesForLevelBase(4 * 256 * 1024 * 1024); // 4 times the sst file
options.setCompressionType(CompressionType.SNAPPY_COMPRESSION);
options.setTargetFileSizeMultiplier(2); // Each level is 2 times the previous level
options.setCompressionPerLevel(List.of(CompressionType.NO_COMPRESSION,
CompressionType.SNAPPY_COMPRESSION,
CompressionType.SNAPPY_COMPRESSION
));
//options.setMaxBytesForLevelBase(4 * 256 * 1024 * 1024); // 4 times the sst file
options.setManualWalFlush(false);
options.setMinWriteBufferNumberToMerge(3);
options.setMaxWriteBufferNumber(4);
@ -217,6 +222,13 @@ public class LLLocalKeyValueDatabase implements LLKeyValueDatabase {
options.setAllowMmapReads(true);
options.setAllowMmapWrites(true);
options.setAllowFAllocate(true);
options.setRateLimiter(new RateLimiter(10L * 1024L * 1024L)); // 10MiB/s max compaction write speed
options.setDbPaths(List.of(new DbPath(databasesDirPath.resolve(path.getFileName() + "_hot"),
50L * 1024L * 1024L * 1024L), // 50GiB
new DbPath(databasesDirPath.resolve(path.getFileName() + "_cold"),
400L * 1024L * 1024L * 1024L), // 400GiB
new DbPath(databasesDirPath.resolve(path.getFileName() + "_colder"),
600L * 1024L * 1024L * 1024L))); // 600GiB
// Direct I/O parameters. Removed because they use too much disk.
//options.setUseDirectReads(true);
//options.setUseDirectIoForFlushAndCompaction(true);
@ -234,10 +246,6 @@ public class LLLocalKeyValueDatabase implements LLKeyValueDatabase {
.setWriteBufferSize(1024 * 1024) // 1MB
.setWalSizeLimitMB(16) // 16MB
.setMaxTotalWalSize(1024L * 1024L * 1024L) // 1GiB max wal directory size
.setDbPaths(List.of(new DbPath(databasesDirPath.resolve(path.getFileName() + "_hot"),
400L * 1024L * 1024L * 1024L), // 400GiB
new DbPath(databasesDirPath.resolve(path.getFileName() + "_cold"),
600L * 1024L * 1024L * 1024L))) // 600GiB
;
tableOptions.setBlockCache(new LRUCache(8L * 1024L * 1024L)); // 8MiB
options.setWriteBufferManager(new WriteBufferManager(8L * 1024L * 1024L, new LRUCache(8L * 1024L * 1024L))); // 8MiB
@ -255,10 +263,6 @@ public class LLLocalKeyValueDatabase implements LLKeyValueDatabase {
.setWriteBufferSize(64 * 1024 * 1024) // 64MB
.setWalSizeLimitMB(1024) // 1024MB
.setMaxTotalWalSize(2L * 1024L * 1024L * 1024L) // 2GiB max wal directory size
.setDbPaths(List.of(new DbPath(databasesDirPath.resolve(path.getFileName() + "_hot"),
400L * 1024L * 1024L * 1024L), // 400GiB
new DbPath(databasesDirPath.resolve(path.getFileName() + "_cold"),
600L * 1024L * 1024L * 1024L))) // 600GiB
;
tableOptions.setBlockCache(new LRUCache(128L * 1024L * 1024L)); // 128MiB
options.setWriteBufferManager(new WriteBufferManager(256L * 1024L * 1024L, new LRUCache(128L * 1024L * 1024L))); // 128MiB

44
src/main/java/it/cavallium/dbengine/database/disk/LLLocalLuceneIndex.java
View File

@ -26,6 +26,7 @@ import java.nio.file.Path;
import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
@ -234,18 +235,15 @@ public class LLLocalLuceneIndex implements LLLuceneIndex {
}
@Override
public Mono<Void> addDocuments(Flux<GroupedFlux<LLTerm, LLDocument>> documents) {
public Mono<Void> addDocuments(Mono<Map<LLTerm, LLDocument>> documents) {
return documents
.flatMap(group -> group
.collectList()
.flatMap(docs -> Mono
.<Void>fromCallable(() -> {
indexWriter.addDocuments(LLUtils.toDocuments(docs));
return null;
})
.subscribeOn(Schedulers.boundedElastic()))
)
.then();
.flatMap(documentsMap -> Mono
.<Void>fromCallable(() -> {
indexWriter.addDocuments(LLUtils.toDocuments(documentsMap.values()));
return null;
})
.subscribeOn(Schedulers.boundedElastic())
);
}
@ -266,21 +264,21 @@ public class LLLocalLuceneIndex implements LLLuceneIndex {
}
@Override
public Mono<Void> updateDocuments(Flux<GroupedFlux<LLTerm, LLDocument>> documents) {
public Mono<Void> updateDocuments(Mono<Map<LLTerm, LLDocument>> documents) {
return documents.flatMap(this::updateDocuments).then();
}
private Mono<Void> updateDocuments(GroupedFlux<LLTerm, LLDocument> documents) {
return documents
.map(LLUtils::toDocument)
.collectList()
.flatMap(luceneDocuments -> Mono
.<Void>fromCallable(() -> {
indexWriter.updateDocuments(LLUtils.toTerm(documents.key()), luceneDocuments);
return null;
})
.subscribeOn(Schedulers.boundedElastic())
);
private Mono<Void> updateDocuments(Map<LLTerm, LLDocument> documentsMap) {
return Mono
.<Void>fromCallable(() -> {
for (Entry<LLTerm, LLDocument> entry : documentsMap.entrySet()) {
LLTerm key = entry.getKey();
LLDocument value = entry.getValue();
indexWriter.updateDocument(LLUtils.toTerm(key), LLUtils.toDocument(value));
}
return null;
})
.subscribeOn(Schedulers.boundedElastic());
}
@Override

39
src/main/java/it/cavallium/dbengine/database/disk/LLLocalMultiLuceneIndex.java
View File

@ -17,6 +17,9 @@ import it.unimi.dsi.fastutil.longs.Long2ObjectOpenHashMap;
import java.io.IOException;
import java.nio.file.Path;
import java.time.Duration;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
@ -168,8 +171,22 @@ public class LLLocalMultiLuceneIndex implements LLLuceneIndex {
}
@Override
public Mono<Void> addDocuments(Flux<GroupedFlux<LLTerm, LLDocument>> documents) {
return documents.flatMap(docs -> getLuceneIndex(docs.key()).addDocuments(documents)).then();
public Mono<Void> addDocuments(Mono<Map<LLTerm, LLDocument>> documents) {
return documents
.flatMapMany(map -> {
var sortedMap = new HashMap<LLLocalLuceneIndex, Map<LLTerm, LLDocument>>();
map.forEach((key, value) -> sortedMap
.computeIfAbsent(getLuceneIndex(key), _unused -> new HashMap<>())
.put(key, value)
);
return Flux.fromIterable(sortedMap.entrySet());
})
.flatMap(luceneIndexWithNewDocuments -> {
var luceneIndex = luceneIndexWithNewDocuments.getKey();
var docs = luceneIndexWithNewDocuments.getValue();
return luceneIndex.addDocuments(Mono.just(docs));
})
.then();
}
@Override
@ -183,8 +200,22 @@ public class LLLocalMultiLuceneIndex implements LLLuceneIndex {
}
@Override
public Mono<Void> updateDocuments(Flux<GroupedFlux<LLTerm, LLDocument>> documents) {
return documents.flatMap(docs -> getLuceneIndex(docs.key()).updateDocuments(documents)).then();
public Mono<Void> updateDocuments(Mono<Map<LLTerm, LLDocument>> documents) {
return documents
.flatMapMany(map -> {
var sortedMap = new HashMap<LLLocalLuceneIndex, Map<LLTerm, LLDocument>>();
map.forEach((key, value) -> sortedMap
.computeIfAbsent(getLuceneIndex(key), _unused -> new HashMap<>())
.put(key, value)
);
return Flux.fromIterable(sortedMap.entrySet());
})
.flatMap(luceneIndexWithNewDocuments -> {
var luceneIndex = luceneIndexWithNewDocuments.getKey();
var docs = luceneIndexWithNewDocuments.getValue();
return luceneIndex.updateDocuments(Mono.just(docs));
})
.then();
}
@Override