CavalliumDBEngine/src/main/java/it/cavallium/dbengine/database/disk/AbstractRocksDBColumn.java

510 lines
16 KiB
Java
Raw Normal View History

2021-10-20 01:51:34 +02:00
package it.cavallium.dbengine.database.disk;
2021-12-12 02:17:36 +01:00
import static io.net5.buffer.api.StandardAllocationTypes.OFF_HEAP;
import static it.cavallium.dbengine.database.LLUtils.INITIAL_DIRECT_READ_BYTE_BUF_SIZE_BYTES;
2021-10-20 01:51:34 +02:00
import static java.util.Objects.requireNonNull;
import io.micrometer.core.instrument.Gauge;
import io.micrometer.core.instrument.MeterRegistry;
2021-12-12 02:17:36 +01:00
import io.net5.buffer.api.AllocationType;
import io.net5.buffer.api.AllocatorControl;
2021-10-20 01:51:34 +02:00
import io.net5.buffer.api.Buffer;
import io.net5.buffer.api.BufferAllocator;
2021-12-12 02:17:36 +01:00
import io.net5.buffer.api.CompositeBuffer;
import io.net5.buffer.api.MemoryManager;
2021-10-20 01:51:34 +02:00
import io.net5.buffer.api.Send;
2021-10-30 12:39:56 +02:00
import io.net5.buffer.api.StandardAllocationTypes;
2021-12-12 02:17:36 +01:00
import io.net5.buffer.api.WritableComponent;
import io.net5.buffer.api.internal.Statics;
import io.net5.buffer.api.unsafe.UnsafeMemoryManager;
2021-10-20 01:51:34 +02:00
import io.net5.util.internal.PlatformDependent;
import it.cavallium.dbengine.client.DatabaseOptions;
import it.cavallium.dbengine.database.LLUtils;
import it.cavallium.dbengine.database.LLUtils.DirectBuffer;
import it.cavallium.dbengine.database.RepeatedElementList;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
2021-10-20 01:51:34 +02:00
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import org.rocksdb.ColumnFamilyHandle;
import org.rocksdb.CompactRangeOptions;
2021-12-12 02:17:36 +01:00
import org.rocksdb.FileOperationInfo;
2021-10-20 01:51:34 +02:00
import org.rocksdb.FlushOptions;
import org.rocksdb.Holder;
2021-12-12 02:17:36 +01:00
import org.rocksdb.KeyMayExistWorkaround;
2021-10-20 01:51:34 +02:00
import org.rocksdb.ReadOptions;
import org.rocksdb.RocksDB;
import org.rocksdb.RocksDBException;
import org.rocksdb.RocksIterator;
import org.rocksdb.Transaction;
import org.rocksdb.WriteBatch;
import org.rocksdb.WriteOptions;
import reactor.core.scheduler.Schedulers;
2021-12-12 02:17:36 +01:00
import sun.misc.Unsafe;
2021-10-20 01:51:34 +02:00
public sealed abstract class AbstractRocksDBColumn<T extends RocksDB> implements RocksDBColumn
permits StandardRocksDBColumn, OptimisticRocksDBColumn, PessimisticRocksDBColumn {
private static final byte[] NO_DATA = new byte[0];
protected static final UpdateAtomicResult RESULT_NOTHING = new UpdateAtomicResultNothing();
protected final Logger logger = LogManager.getLogger(this.getClass());
2021-10-20 01:51:34 +02:00
private final T db;
private final DatabaseOptions opts;
2021-10-30 12:39:56 +02:00
private final boolean nettyDirect;
2021-10-20 01:51:34 +02:00
private final BufferAllocator alloc;
private final ColumnFamilyHandle cfh;
private final MeterRegistry meterRegistry;
private final AtomicInteger lastDataSizeMetric = new AtomicInteger(0);
public AbstractRocksDBColumn(T db,
DatabaseOptions databaseOptions,
BufferAllocator alloc,
ColumnFamilyHandle cfh,
MeterRegistry meterRegistry) {
2021-10-20 01:51:34 +02:00
this.db = db;
this.opts = databaseOptions;
2021-12-12 02:17:36 +01:00
this.nettyDirect = opts.allowNettyDirect() && alloc.getAllocationType() == OFF_HEAP;
2021-10-20 01:51:34 +02:00
this.alloc = alloc;
this.cfh = cfh;
this.meterRegistry = meterRegistry;
Gauge
.builder("it.cavallium.dbengine.database.disk.column.lastdatasize", lastDataSizeMetric::get)
.description("Last data size read using get()")
.register(meterRegistry);
2021-10-20 01:51:34 +02:00
}
protected T getDb() {
return db;
}
protected DatabaseOptions getOpts() {
return opts;
}
protected ColumnFamilyHandle getCfh() {
return cfh;
}
@Override
2021-12-12 02:17:36 +01:00
public @Nullable Buffer get(@NotNull ReadOptions readOptions, Buffer key, boolean existsAlmostCertainly)
throws RocksDBException {
if (Schedulers.isInNonBlockingThread()) {
throw new UnsupportedOperationException("Called dbGet in a nonblocking thread");
}
if (!db.isOwningHandle()) {
throw new IllegalStateException("Database is closed");
}
if (!readOptions.isOwningHandle()) {
throw new IllegalStateException("ReadOptions is closed");
}
if (!cfh.isOwningHandle()) {
throw new IllegalStateException("Column family is closed");
}
if (nettyDirect) {
// Get the key nio buffer to pass to RocksDB
ByteBuffer keyNioBuffer = LLUtils.asReadOnlyDirect(key);
boolean mustCloseKey;
if (keyNioBuffer == null) {
mustCloseKey = true;
// If the nio buffer is not available, copy the netty buffer into a new direct buffer
keyNioBuffer = LLUtils.copyToNewDirectBuffer(key);
} else {
2021-12-13 01:57:37 +01:00
assert keyNioBuffer.isDirect();
2021-12-12 02:17:36 +01:00
mustCloseKey = false;
}
2021-12-12 02:17:36 +01:00
assert keyNioBuffer.limit() == key.readableBytes();
try {
2021-10-20 01:51:34 +02:00
// Create a direct result buffer because RocksDB works only with direct buffers
2021-12-12 02:17:36 +01:00
var resultBuffer = LLUtils.allocateShared(INITIAL_DIRECT_READ_BYTE_BUF_SIZE_BYTES);
try {
assert resultBuffer.readerOffset() == 0;
assert resultBuffer.writerOffset() == 0;
var resultWritable = ((WritableComponent) resultBuffer).writableBuffer();
2021-12-12 02:17:36 +01:00
var keyMayExist = db.keyMayExist(cfh, keyNioBuffer, resultWritable);
var keyMayExistState = KeyMayExistWorkaround.getExistenceState(keyMayExist);
int keyMayExistValueLength = KeyMayExistWorkaround.getValueLength(keyMayExist);
// At the beginning, size reflects the expected size, then it becomes the real data size
int size = keyMayExistState == 2 ? keyMayExistValueLength : -1;
switch (keyMayExistState) {
// kNotExist
case 0: {
resultBuffer.close();
return null;
2021-10-20 01:51:34 +02:00
}
2021-12-12 02:17:36 +01:00
// kExistsWithoutValue
case 1: {
assert keyMayExistValueLength == 0;
resultWritable.clear();
// real data size
2021-12-13 01:57:37 +01:00
size = db.get(cfh, readOptions, keyNioBuffer.position(0), resultWritable);
2021-12-12 02:17:36 +01:00
if (size == RocksDB.NOT_FOUND) {
resultBuffer.close();
2021-10-20 01:51:34 +02:00
return null;
2021-12-12 02:17:36 +01:00
}
}
// kExistsWithValue
case 2: {
// real data size
this.lastDataSizeMetric.set(size);
assert size >= 0;
if (size <= resultWritable.limit()) {
assert size == resultWritable.limit();
return resultBuffer.writerOffset(resultWritable.limit());
2021-10-20 01:51:34 +02:00
} else {
2021-12-12 02:17:36 +01:00
resultBuffer.ensureWritable(size);
resultWritable = ((WritableComponent) resultBuffer).writableBuffer();
assert resultBuffer.readerOffset() == 0;
assert resultBuffer.writerOffset() == 0;
2021-12-13 01:57:37 +01:00
size = db.get(cfh, readOptions, keyNioBuffer.position(0), resultWritable);
2021-12-12 02:17:36 +01:00
if (size == RocksDB.NOT_FOUND) {
resultBuffer.close();
return null;
}
assert size == resultWritable.limit();
return resultBuffer.writerOffset(resultWritable.limit());
2021-10-20 01:51:34 +02:00
}
}
2021-12-12 02:17:36 +01:00
default: {
throw new IllegalStateException();
}
2021-10-20 01:51:34 +02:00
}
2021-12-12 02:17:36 +01:00
} catch (Throwable t) {
resultBuffer.close();
throw t;
}
} finally {
if (mustCloseKey) {
PlatformDependent.freeDirectBuffer(keyNioBuffer);
}
}
} else {
try {
byte[] keyArray = LLUtils.toArray(key);
requireNonNull(keyArray);
Holder<byte[]> data = existsAlmostCertainly ? null : new Holder<>();
if (existsAlmostCertainly || db.keyMayExist(cfh, readOptions, keyArray, data)) {
if (!existsAlmostCertainly && data.getValue() != null) {
return LLUtils.fromByteArray(alloc, data.getValue());
} else {
byte[] result = db.get(cfh, readOptions, keyArray);
if (result == null) {
return null;
} else {
return LLUtils.fromByteArray(alloc, result);
}
2021-10-20 01:51:34 +02:00
}
2021-12-12 02:17:36 +01:00
} else {
return null;
}
} finally {
if (!(readOptions instanceof UnreleasableReadOptions)) {
readOptions.close();
2021-10-20 01:51:34 +02:00
}
}
}
}
@Override
2021-12-12 02:17:36 +01:00
public void put(@NotNull WriteOptions writeOptions, Buffer key, Buffer value) throws RocksDBException {
2021-10-20 01:51:34 +02:00
try {
2021-12-12 02:17:36 +01:00
if (Schedulers.isInNonBlockingThread()) {
throw new UnsupportedOperationException("Called dbPut in a nonblocking thread");
}
if (!db.isOwningHandle()) {
throw new IllegalStateException("Database is closed");
}
if (!writeOptions.isOwningHandle()) {
throw new IllegalStateException("WriteOptions is closed");
}
if (!cfh.isOwningHandle()) {
throw new IllegalStateException("Column family is closed");
}
assert key.isAccessible();
assert value.isAccessible();
if (nettyDirect) {
// Get the key nio buffer to pass to RocksDB
ByteBuffer keyNioBuffer = LLUtils.asReadOnlyDirect(key);
boolean mustCloseKey;
if (keyNioBuffer == null) {
mustCloseKey = true;
// If the nio buffer is not available, copy the netty buffer into a new direct buffer
keyNioBuffer = LLUtils.copyToNewDirectBuffer(key);
} else {
mustCloseKey = false;
}
try {
// Get the value nio buffer to pass to RocksDB
ByteBuffer valueNioBuffer = LLUtils.asReadOnlyDirect(value);
boolean mustCloseValue;
if (valueNioBuffer == null) {
mustCloseValue = true;
// If the nio buffer is not available, copy the netty buffer into a new direct buffer
valueNioBuffer = LLUtils.copyToNewDirectBuffer(value);
} else {
mustCloseValue = false;
}
2021-12-12 02:17:36 +01:00
try {
db.put(cfh, writeOptions, keyNioBuffer, valueNioBuffer);
} finally {
if (mustCloseValue) {
PlatformDependent.freeDirectBuffer(valueNioBuffer);
2021-10-20 01:51:34 +02:00
}
2021-12-12 02:17:36 +01:00
}
} finally {
if (mustCloseKey) {
PlatformDependent.freeDirectBuffer(keyNioBuffer);
2021-10-20 01:51:34 +02:00
}
}
2021-12-12 02:17:36 +01:00
} else {
db.put(cfh, writeOptions, LLUtils.toArray(key), LLUtils.toArray(value));
2021-10-20 01:51:34 +02:00
}
} finally {
if (!(writeOptions instanceof UnreleasableWriteOptions)) {
2021-10-20 01:51:34 +02:00
writeOptions.close();
}
}
}
@Override
2021-12-12 02:17:36 +01:00
public boolean exists(@NotNull ReadOptions readOptions, Buffer key) throws RocksDBException {
if (Schedulers.isInNonBlockingThread()) {
throw new UnsupportedOperationException("Called containsKey in a nonblocking thread");
}
if (!db.isOwningHandle()) {
throw new IllegalStateException("Database is closed");
}
if (!readOptions.isOwningHandle()) {
throw new IllegalStateException("ReadOptions is closed");
}
if (!cfh.isOwningHandle()) {
throw new IllegalStateException("Column family is closed");
}
if (nettyDirect) {
// Get the key nio buffer to pass to RocksDB
ByteBuffer keyNioBuffer = LLUtils.asReadOnlyDirect(key);
boolean mustCloseKey;
if (keyNioBuffer == null) {
mustCloseKey = true;
// If the nio buffer is not available, copy the netty buffer into a new direct buffer
keyNioBuffer = LLUtils.copyToNewDirectBuffer(key);
} else {
mustCloseKey = false;
}
2021-12-12 02:17:36 +01:00
try {
if (db.keyMayExist(cfh, keyNioBuffer)) {
2021-12-13 01:57:37 +01:00
int size = db.get(cfh, readOptions, keyNioBuffer.position(0), LLUtils.EMPTY_BYTE_BUFFER);
2021-12-12 02:17:36 +01:00
return size != RocksDB.NOT_FOUND;
} else {
return false;
}
} finally {
if (mustCloseKey) {
PlatformDependent.freeDirectBuffer(keyNioBuffer);
}
}
2021-12-12 02:17:36 +01:00
} else {
2021-10-20 01:51:34 +02:00
int size = RocksDB.NOT_FOUND;
byte[] keyBytes = LLUtils.toArray(key);
Holder<byte[]> data = new Holder<>();
try {
if (db.keyMayExist(cfh, readOptions, keyBytes, data)) {
if (data.getValue() != null) {
size = data.getValue().length;
} else {
size = db.get(cfh, readOptions, keyBytes, NO_DATA);
}
}
} finally {
if (!(readOptions instanceof UnreleasableReadOptions)) {
2021-10-20 01:51:34 +02:00
readOptions.close();
}
}
return size != RocksDB.NOT_FOUND;
}
}
@Override
2021-12-12 02:17:36 +01:00
public void delete(WriteOptions writeOptions, Buffer key) throws RocksDBException {
if (!db.isOwningHandle()) {
throw new IllegalStateException("Database is closed");
}
if (!writeOptions.isOwningHandle()) {
throw new IllegalStateException("WriteOptions is closed");
}
if (!cfh.isOwningHandle()) {
throw new IllegalStateException("Column family is closed");
}
if (nettyDirect) {
// Get the key nio buffer to pass to RocksDB
ByteBuffer keyNioBuffer = LLUtils.asReadOnlyDirect(key);
boolean mustCloseKey;
if (keyNioBuffer == null) {
mustCloseKey = true;
// If the nio buffer is not available, copy the netty buffer into a new direct buffer
keyNioBuffer = LLUtils.copyToNewDirectBuffer(key);
} else {
mustCloseKey = false;
}
2021-12-12 02:17:36 +01:00
try {
db.delete(cfh, writeOptions, keyNioBuffer);
} finally {
if (mustCloseKey) {
PlatformDependent.freeDirectBuffer(keyNioBuffer);
2021-10-20 01:51:34 +02:00
}
}
2021-12-12 02:17:36 +01:00
} else {
db.delete(cfh, writeOptions, LLUtils.toArray(key));
2021-10-20 01:51:34 +02:00
}
}
@Override
public void delete(WriteOptions writeOptions, byte[] key) throws RocksDBException {
if (!db.isOwningHandle()) {
throw new IllegalStateException("Database is closed");
}
if (!writeOptions.isOwningHandle()) {
throw new IllegalStateException("WriteOptions is closed");
}
if (!cfh.isOwningHandle()) {
throw new IllegalStateException("Column family is closed");
}
2021-10-20 01:51:34 +02:00
db.delete(cfh, writeOptions, key);
}
@Override
public List<byte[]> multiGetAsList(ReadOptions readOptions, List<byte[]> keys) throws RocksDBException {
if (!db.isOwningHandle()) {
throw new IllegalStateException("Database is closed");
}
if (!readOptions.isOwningHandle()) {
throw new IllegalStateException("ReadOptions is closed");
}
if (!cfh.isOwningHandle()) {
throw new IllegalStateException("Column family is closed");
}
2021-10-20 01:51:34 +02:00
var columnFamilyHandles = new RepeatedElementList<>(cfh, keys.size());
return db.multiGetAsList(readOptions, columnFamilyHandles, keys);
}
@Override
public void suggestCompactRange() throws RocksDBException {
if (!db.isOwningHandle()) {
throw new IllegalStateException("Database is closed");
}
if (!cfh.isOwningHandle()) {
throw new IllegalStateException("Column family is closed");
}
2021-10-20 01:51:34 +02:00
db.suggestCompactRange(cfh);
}
@Override
public void compactRange(byte[] begin, byte[] end, CompactRangeOptions options)
throws RocksDBException {
if (!db.isOwningHandle()) {
throw new IllegalStateException("Database is closed");
}
if (!options.isOwningHandle()) {
throw new IllegalStateException("CompactRangeOptions is closed");
}
if (!cfh.isOwningHandle()) {
throw new IllegalStateException("Column family is closed");
}
2021-10-20 01:51:34 +02:00
db.compactRange(cfh, begin, end, options);
}
@Override
public void flush(FlushOptions options) throws RocksDBException {
if (!db.isOwningHandle()) {
throw new IllegalStateException("Database is closed");
}
if (!options.isOwningHandle()) {
throw new IllegalStateException("FlushOptions is closed");
}
if (!cfh.isOwningHandle()) {
throw new IllegalStateException("Column family is closed");
}
2021-10-20 01:51:34 +02:00
db.flush(options, cfh);
}
@Override
public void flushWal(boolean sync) throws RocksDBException {
if (!db.isOwningHandle()) {
throw new IllegalStateException("Database is closed");
}
if (!cfh.isOwningHandle()) {
throw new IllegalStateException("Column family is closed");
}
2021-10-20 01:51:34 +02:00
db.flushWal(sync);
}
@Override
public long getLongProperty(String property) throws RocksDBException {
if (!db.isOwningHandle()) {
throw new IllegalStateException("Database is closed");
}
if (!cfh.isOwningHandle()) {
throw new IllegalStateException("Column family is closed");
}
2021-10-20 01:51:34 +02:00
return db.getLongProperty(cfh, property);
}
@Override
public void write(WriteOptions writeOptions, WriteBatch writeBatch) throws RocksDBException {
if (!db.isOwningHandle()) {
throw new IllegalStateException("Database is closed");
}
if (!writeOptions.isOwningHandle()) {
throw new IllegalStateException("WriteOptions is closed");
}
if (!cfh.isOwningHandle()) {
throw new IllegalStateException("Column family is closed");
}
2021-10-20 01:51:34 +02:00
db.write(writeOptions, writeBatch);
}
/**
* @return true if committed successfully
*/
protected abstract boolean commitOptimistically(Transaction tx) throws RocksDBException;
protected abstract Transaction beginTransaction(@NotNull WriteOptions writeOptions);
@Override
@NotNull
public RocksIterator newIterator(@NotNull ReadOptions readOptions) {
if (!db.isOwningHandle()) {
throw new IllegalStateException("Database is closed");
}
if (!readOptions.isOwningHandle()) {
throw new IllegalStateException("ReadOptions is closed");
}
if (!cfh.isOwningHandle()) {
throw new IllegalStateException("Column family is closed");
}
2021-10-20 01:51:34 +02:00
return db.newIterator(cfh, readOptions);
}
@Override
public ColumnFamilyHandle getColumnFamilyHandle() {
return cfh;
}
@Override
public BufferAllocator getAllocator() {
return alloc;
}
public MeterRegistry getMeterRegistry() {
return meterRegistry;
}
2021-10-20 01:51:34 +02:00
}