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

Make buckets unmodifiable

Browse Source
This commit is contained in:
Andrea Cavalli 2021-07-13 22:58:08 +02:00
parent fe5f50be10
commit 2489c50022
3 changed files with 42 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

37
src/main/java/it/cavallium/dbengine/database/collections/DatabaseMapDictionaryHashed.java
View File

@ -12,6 +12,7 @@ import it.cavallium.dbengine.database.collections.JoinerBlocking.ValueGetterBloc
import it.cavallium.dbengine.database.serialization.Serializer; import it.cavallium.dbengine.database.serialization.Serializer;
import it.cavallium.dbengine.database.serialization.SerializerFixedBinaryLength; import it.cavallium.dbengine.database.serialization.SerializerFixedBinaryLength;
import it.unimi.dsi.fastutil.objects.ObjectArraySet; import it.unimi.dsi.fastutil.objects.ObjectArraySet;
import it.unimi.dsi.fastutil.objects.ObjectOpenHashSet;
import it.unimi.dsi.fastutil.objects.ObjectSets; import it.unimi.dsi.fastutil.objects.ObjectSets;
import java.util.Collections; import java.util.Collections;
import java.util.HashMap; import java.util.HashMap;
@ -30,7 +31,7 @@ import reactor.core.publisher.Mono;
public class DatabaseMapDictionaryHashed<T, U, TH> implements DatabaseStageMap<T, U, DatabaseStageEntry<U>> { public class DatabaseMapDictionaryHashed<T, U, TH> implements DatabaseStageMap<T, U, DatabaseStageEntry<U>> {
private final ByteBufAllocator alloc; private final ByteBufAllocator alloc;
private final DatabaseMapDictionary<TH, Set<Entry<T, U>>> subDictionary; private final DatabaseMapDictionary<TH, ObjectArraySet<Entry<T, U>>> subDictionary;
private final Function<T, TH> keySuffixHashFunction; private final Function<T, TH> keySuffixHashFunction;
protected DatabaseMapDictionaryHashed(LLDictionary dictionary, protected DatabaseMapDictionaryHashed(LLDictionary dictionary,
@ -89,11 +90,11 @@ public class DatabaseMapDictionaryHashed<T, U, TH> implements DatabaseStageMap<T
); );
} }
private Map<TH, Set<Entry<T, U>>> serializeMap(Map<T, U> map) { private Map<TH, ObjectArraySet<Entry<T, U>>> serializeMap(Map<T, U> map) {
var newMap = new HashMap<TH, Set<Entry<T, U>>>(map.size()); var newMap = new HashMap<TH, ObjectArraySet<Entry<T, U>>>(map.size());
map.forEach((key, value) -> newMap.compute(keySuffixHashFunction.apply(key), (hash, prev) -> { map.forEach((key, value) -> newMap.compute(keySuffixHashFunction.apply(key), (hash, prev) -> {
if (prev == null) { if (prev == null) {
prev = new HashSet<>(); prev = new ObjectArraySet<>();
} }
prev.add(Map.entry(key, value)); prev.add(Map.entry(key, value));
return prev; return prev;
@ -101,7 +102,7 @@ public class DatabaseMapDictionaryHashed<T, U, TH> implements DatabaseStageMap<T
return newMap; return newMap;
} }
private Map<T, U> deserializeMap(Map<TH, Set<Entry<T, U>>> map) { private Map<T, U> deserializeMap(Map<TH, ObjectArraySet<Entry<T, U>>> map) {
var newMap = new HashMap<T, U>(map.size()); var newMap = new HashMap<T, U>(map.size());
map.forEach((hash, set) -> set.forEach(entry -> newMap.put(entry.getKey(), entry.getValue()))); map.forEach((hash, set) -> set.forEach(entry -> newMap.put(entry.getKey(), entry.getValue())));
return newMap; return newMap;
@ -246,24 +247,24 @@ public class DatabaseMapDictionaryHashed<T, U, TH> implements DatabaseStageMap<T
@Override @Override
public ValueGetterBlocking<T, U> getDbValueGetter(@Nullable CompositeSnapshot snapshot) { public ValueGetterBlocking<T, U> getDbValueGetter(@Nullable CompositeSnapshot snapshot) {
ValueGetterBlocking<TH, Set<Entry<T, U>>> getter = subDictionary.getDbValueGetter(snapshot); ValueGetterBlocking<TH, ObjectArraySet<Entry<T, U>>> getter = subDictionary.getDbValueGetter(snapshot);
return key -> extractValue(getter.get(keySuffixHashFunction.apply(key)), key); return key -> extractValue(getter.get(keySuffixHashFunction.apply(key)), key);
} }
@Override @Override
public ValueGetter<T, U> getAsyncDbValueGetter(@Nullable CompositeSnapshot snapshot) { public ValueGetter<T, U> getAsyncDbValueGetter(@Nullable CompositeSnapshot snapshot) {
ValueGetter<TH, Set<Entry<T, U>>> getter = subDictionary.getAsyncDbValueGetter(snapshot); ValueGetter<TH, ObjectArraySet<Entry<T, U>>> getter = subDictionary.getAsyncDbValueGetter(snapshot);
return key -> getter return key -> getter
.get(keySuffixHashFunction.apply(key)) .get(keySuffixHashFunction.apply(key))
.flatMap(set -> this.extractValueTransformation(set, key)); .flatMap(set -> this.extractValueTransformation(set, key));
} }
private Mono<U> extractValueTransformation(Set<Entry<T, U>> entries, T key) { private Mono<U> extractValueTransformation(ObjectArraySet<Entry<T, U>> entries, T key) {
return Mono.fromCallable(() -> extractValue(entries, key)); return Mono.fromCallable(() -> extractValue(entries, key));
} }
@Nullable @Nullable
private U extractValue(Set<Entry<T, U>> entries, T key) { private U extractValue(ObjectArraySet<Entry<T, U>> entries, T key) {
if (entries == null) return null; if (entries == null) return null;
for (Entry<T, U> entry : entries) { for (Entry<T, U> entry : entries) {
if (Objects.equals(entry.getKey(), key)) { if (Objects.equals(entry.getKey(), key)) {
@ -274,21 +275,23 @@ public class DatabaseMapDictionaryHashed<T, U, TH> implements DatabaseStageMap<T
} }
@NotNull @NotNull
private Set<Entry<T, U>> insertValueOrCreate(@Nullable Set<Entry<T, U>> entries, T key, U value) { private ObjectArraySet<Entry<T, U>> insertValueOrCreate(@Nullable ObjectArraySet<Entry<T, U>> entries, T key, U value) {
if (entries != null) { if (entries != null) {
entries.add(Map.entry(key, value)); var clonedEntries = entries.clone();
return entries; clonedEntries.add(Map.entry(key, value));
return clonedEntries;
} else { } else {
var oas = new HashSet<Entry<T, U>>(1); var oas = new ObjectArraySet<Entry<T, U>>(1);
oas.add(Map.entry(key, value)); oas.add(Map.entry(key, value));
return oas; return oas;
} }
} }
@Nullable @Nullable
private Set<Entry<T, U>> removeValueOrDelete(@Nullable Set<Entry<T, U>> entries, T key) { private Set<Entry<T, U>> removeValueOrDelete(@Nullable ObjectArraySet<Entry<T, U>> entries, T key) {
if (entries != null) { if (entries != null) {
var it = entries.iterator(); var clonedEntries = entries.clone();
var it = clonedEntries.iterator();
while (it.hasNext()) { while (it.hasNext()) {
var entry = it.next(); var entry = it.next();
if (Objects.equals(entry.getKey(), key)) { if (Objects.equals(entry.getKey(), key)) {
@ -296,10 +299,10 @@ public class DatabaseMapDictionaryHashed<T, U, TH> implements DatabaseStageMap<T
break; break;
} }
} }
if (entries.size() == 0) { if (clonedEntries.size() == 0) {
return null; return null;
} else { } else {
return entries; return clonedEntries;
} }
} else { } else {
return null; return null;

24
src/main/java/it/cavallium/dbengine/database/collections/DatabaseSingleBucket.java
View File

@ -24,10 +24,10 @@ import reactor.core.publisher.Mono;
@SuppressWarnings("unused") @SuppressWarnings("unused")
public class DatabaseSingleBucket<K, V, TH> implements DatabaseStageEntry<V> { public class DatabaseSingleBucket<K, V, TH> implements DatabaseStageEntry<V> {
private final DatabaseStageEntry<Set<Entry<K, V>>> bucketStage; private final DatabaseStageEntry<ObjectArraySet<Entry<K, V>>> bucketStage;
private final K key; private final K key;
public DatabaseSingleBucket(DatabaseStageEntry<Set<Entry<K, V>>> bucketStage, K key) { public DatabaseSingleBucket(DatabaseStageEntry<ObjectArraySet<Entry<K, V>>> bucketStage, K key) {
this.bucketStage = bucketStage; this.bucketStage = bucketStage;
this.key = key; this.key = key;
} }
@ -151,9 +151,10 @@ public class DatabaseSingleBucket<K, V, TH> implements DatabaseStageEntry<V> {
} }
@NotNull @NotNull
private Set<Entry<K, V>> insertValueOrCreate(@Nullable Set<Entry<K, V>> entries, V value) { private ObjectArraySet<Entry<K, V>> insertValueOrCreate(@Nullable ObjectArraySet<Entry<K, V>> entries, V value) {
if (entries != null) { if (entries != null) {
var it = entries.iterator(); var clonedEntries = entries.clone();
var it = clonedEntries.iterator();
while (it.hasNext()) { while (it.hasNext()) {
var entry = it.next(); var entry = it.next();
if (Objects.equals(entry.getKey(), key)) { if (Objects.equals(entry.getKey(), key)) {
@ -161,19 +162,20 @@ public class DatabaseSingleBucket<K, V, TH> implements DatabaseStageEntry<V> {
break; break;
} }
} }
entries.add(Map.entry(key, value)); clonedEntries.add(Map.entry(key, value));
return entries; return clonedEntries;
} else { } else {
var oas = new HashSet<Entry<K, V>>(1); var oas = new ObjectArraySet<Entry<K, V>>(1);
oas.add(Map.entry(key, value)); oas.add(Map.entry(key, value));
return oas; return oas;
} }
} }
@Nullable @Nullable
private Set<Entry<K, V>> removeValueOrDelete(@Nullable Set<Entry<K, V>> entries) { private ObjectArraySet<Entry<K, V>> removeValueOrDelete(@Nullable ObjectArraySet<Entry<K, V>> entries) {
if (entries != null) { if (entries != null) {
var it = entries.iterator(); var clonedEntries = entries.clone();
var it = clonedEntries.iterator();
while (it.hasNext()) { while (it.hasNext()) {
var entry = it.next(); var entry = it.next();
if (Objects.equals(entry.getKey(), key)) { if (Objects.equals(entry.getKey(), key)) {
@ -181,10 +183,10 @@ public class DatabaseSingleBucket<K, V, TH> implements DatabaseStageEntry<V> {
break; break;
} }
} }
if (entries.size() == 0) { if (clonedEntries.size() == 0) {
return null; return null;
} else { } else {
return entries; return clonedEntries;
} }
} else { } else {
return null; return null;

15
src/main/java/it/cavallium/dbengine/database/collections/ValuesSetSerializer.java
View File

@ -4,12 +4,15 @@ import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator; import io.netty.buffer.ByteBufAllocator;
import it.cavallium.dbengine.database.serialization.Serializer; import it.cavallium.dbengine.database.serialization.Serializer;
import it.unimi.dsi.fastutil.objects.ObjectArraySet; import it.unimi.dsi.fastutil.objects.ObjectArraySet;
import it.unimi.dsi.fastutil.objects.ObjectOpenHashSet;
import it.unimi.dsi.fastutil.objects.ObjectSets; import it.unimi.dsi.fastutil.objects.ObjectSets;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet; import java.util.HashSet;
import java.util.Set; import java.util.Set;
import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.NotNull;
class ValuesSetSerializer<X> implements Serializer<Set<X>, ByteBuf> { class ValuesSetSerializer<X> implements Serializer<ObjectArraySet<X>, ByteBuf> {
private final ByteBufAllocator allocator; private final ByteBufAllocator allocator;
private final Serializer<X, ByteBuf> entrySerializer; private final Serializer<X, ByteBuf> entrySerializer;
@ -20,22 +23,22 @@ class ValuesSetSerializer<X> implements Serializer<Set<X>, ByteBuf> {
} }
@Override @Override
public @NotNull Set<X> deserialize(@NotNull ByteBuf serialized) { public @NotNull ObjectArraySet<X> deserialize(@NotNull ByteBuf serialized) {
try { try {
int entriesLength = serialized.readInt(); int entriesLength = serialized.readInt();
var set = new HashSet<X>(); ArrayList<X> deserializedElements = new ArrayList<>(entriesLength);
for (int i = 0; i < entriesLength; i++) { for (int i = 0; i < entriesLength; i++) {
X entry = entrySerializer.deserialize(serialized.retain()); X entry = entrySerializer.deserialize(serialized.retain());
set.add(entry); deserializedElements.add(entry);
} }
return set; return new ObjectArraySet<>(deserializedElements);
} finally { } finally {
serialized.release(); serialized.release();
} }
} }
@Override @Override
public @NotNull ByteBuf serialize(@NotNull Set<X> deserialized) { public @NotNull ByteBuf serialize(@NotNull ObjectArraySet<X> deserialized) {
ByteBuf output = allocator.buffer(); ByteBuf output = allocator.buffer();
try { try {
output.writeInt(deserialized.size()); output.writeInt(deserialized.size());