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

551 lines
18 KiB
Java
Raw Normal View History

2021-10-20 01:51:34 +02:00
package it.cavallium.dbengine.database.disk;
2022-03-16 13:47:56 +01:00
import static io.netty5.buffer.api.StandardAllocationTypes.OFF_HEAP;
2021-12-12 02:17:36 +01:00
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;
2022-02-11 12:19:32 +01:00
import static org.rocksdb.KeyMayExist.KeyMayExistEnum.kExistsWithValue;
import static org.rocksdb.KeyMayExist.KeyMayExistEnum.kExistsWithoutValue;
2021-10-20 01:51:34 +02:00
import io.micrometer.core.instrument.Gauge;
import io.micrometer.core.instrument.MeterRegistry;
2022-03-16 13:47:56 +01:00
import io.netty5.buffer.api.Buffer;
import io.netty5.buffer.api.BufferAllocator;
2022-03-16 19:19:26 +01:00
import io.netty5.buffer.api.DefaultBufferAllocators;
import io.netty5.buffer.api.MemoryManager;
import io.netty5.buffer.api.ReadableComponent;
2022-03-16 13:47:56 +01:00
import io.netty5.buffer.api.WritableComponent;
import io.netty5.util.internal.PlatformDependent;
2021-10-20 01:51:34 +02:00
import it.cavallium.dbengine.database.LLUtils;
import it.cavallium.dbengine.database.RepeatedElementList;
2022-03-16 19:19:26 +01:00
import it.cavallium.dbengine.lucene.DirectNIOFSDirectory;
2022-03-02 12:34:30 +01:00
import it.cavallium.dbengine.rpc.current.data.DatabaseOptions;
2021-10-20 01:51:34 +02:00
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;
import org.rocksdb.FlushOptions;
import org.rocksdb.Holder;
2022-02-11 12:19:32 +01:00
import org.rocksdb.KeyMayExist.KeyMayExistEnum;
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;
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
2022-03-02 18:33:58 +01:00
public @Nullable Buffer get(@NotNull ReadOptions readOptions, Buffer key)
2021-12-12 02:17:36 +01:00
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
2022-03-16 19:19:26 +01:00
ByteBuffer keyNioBuffer;
2021-12-12 02:17:36 +01:00
boolean mustCloseKey;
2022-03-16 19:19:26 +01:00
{
if (!LLUtils.isReadOnlyDirect(key)) {
// If the nio buffer is not available, copy the netty buffer into a new direct buffer
mustCloseKey = true;
var directKey = DefaultBufferAllocators.offHeapAllocator().allocate(key.readableBytes());
key.copyInto(key.readerOffset(), directKey, 0, key.readableBytes());
key = directKey;
} else {
mustCloseKey = false;
}
keyNioBuffer = ((ReadableComponent) key).readableBuffer();
2021-12-13 01:57:37 +01:00
assert keyNioBuffer.isDirect();
2022-03-16 19:19:26 +01:00
assert keyNioBuffer.limit() == key.readableBytes();
}
2022-03-16 19:19:26 +01:00
2021-12-12 02:17:36 +01:00
try {
2021-10-20 01:51:34 +02:00
// Create a direct result buffer because RocksDB works only with direct buffers
2022-03-16 19:19:26 +01:00
var resultBuffer = alloc.allocate(INITIAL_DIRECT_READ_BYTE_BUF_SIZE_BYTES);
2021-12-12 02:17:36 +01:00
try {
assert resultBuffer.readerOffset() == 0;
assert resultBuffer.writerOffset() == 0;
var resultWritable = ((WritableComponent) resultBuffer).writableBuffer();
2021-12-30 22:29:17 +01:00
var keyMayExist = db.keyMayExist(cfh, readOptions, keyNioBuffer.rewind(), resultWritable.clear());
2022-02-11 12:19:32 +01:00
KeyMayExistEnum keyMayExistState = keyMayExist.exists;
int keyMayExistValueLength = keyMayExist.valueLength;
2021-12-12 02:17:36 +01:00
// At the beginning, size reflects the expected size, then it becomes the real data size
2022-02-11 12:19:32 +01:00
int size = keyMayExistState == kExistsWithValue ? keyMayExistValueLength : -1;
2021-12-12 02:17:36 +01:00
switch (keyMayExistState) {
2022-02-11 12:19:32 +01:00
case kNotExist: {
2021-12-12 02:17:36 +01:00
resultBuffer.close();
return null;
2021-10-20 01:51:34 +02:00
}
// todo: kExistsWithValue is not reliable (read below),
// in some cases it should be treated as kExistsWithoutValue
case kExistsWithValue:
2022-02-11 12:19:32 +01:00
case kExistsWithoutValue: {
boolean isKExistsWithoutValue = false;
if (keyMayExistState == kExistsWithoutValue) {
isKExistsWithoutValue = true;
} else {
// todo: "size == 0 || resultWritable.limit() == 0" is checked because keyMayExist is broken,
// and sometimes it returns an empty array, as if it exists
if (size == 0 || resultWritable.limit() == 0) {
isKExistsWithoutValue = true;
}
}
if (isKExistsWithoutValue) {
assert keyMayExistValueLength == 0;
resultWritable.clear();
// real data size
size = db.get(cfh, readOptions, keyNioBuffer.rewind(), resultWritable.clear());
if (size == RocksDB.NOT_FOUND) {
resultBuffer.close();
return null;
}
2021-12-12 02:17:36 +01:00
}
}
default: {
2021-12-12 02:17:36 +01:00
// 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-30 22:29:17 +01:00
size = db.get(cfh, readOptions, keyNioBuffer.rewind(), resultWritable.clear());
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
} catch (Throwable t) {
resultBuffer.close();
throw t;
}
} finally {
if (mustCloseKey) {
2022-03-16 19:19:26 +01:00
key.close();
2021-12-12 02:17:36 +01:00
}
}
} else {
try {
byte[] keyArray = LLUtils.toArray(key);
requireNonNull(keyArray);
2022-03-02 18:33:58 +01:00
Holder<byte[]> data = new Holder<>();
if (db.keyMayExist(cfh, readOptions, keyArray, data)) {
// todo: "data.getValue().length > 0" is checked because keyMayExist is broken, and sometimes it
// returns an empty array, as if it exists
if (data.getValue() != null && data.getValue().length > 0) {
2021-12-12 02:17:36 +01:00
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
2022-03-16 19:19:26 +01:00
ByteBuffer keyNioBuffer;
2021-12-12 02:17:36 +01:00
boolean mustCloseKey;
2022-03-16 19:19:26 +01:00
{
if (!LLUtils.isReadOnlyDirect(key)) {
// If the nio buffer is not available, copy the netty buffer into a new direct buffer
mustCloseKey = true;
var directKey = DefaultBufferAllocators.offHeapAllocator().allocate(key.readableBytes());
key.copyInto(key.readerOffset(), directKey, 0, key.readableBytes());
key = directKey;
} else {
mustCloseKey = false;
}
keyNioBuffer = ((ReadableComponent) key).readableBuffer();
assert keyNioBuffer.isDirect();
assert keyNioBuffer.limit() == key.readableBytes();
2021-12-12 02:17:36 +01:00
}
try {
// Get the value nio buffer to pass to RocksDB
2022-03-16 19:19:26 +01:00
ByteBuffer valueNioBuffer;
2021-12-12 02:17:36 +01:00
boolean mustCloseValue;
2022-03-16 19:19:26 +01:00
{
if (!LLUtils.isReadOnlyDirect(value)) {
// If the nio buffer is not available, copy the netty buffer into a new direct buffer
mustCloseValue = true;
var directValue = DefaultBufferAllocators.offHeapAllocator().allocate(value.readableBytes());
value.copyInto(value.readerOffset(), directValue, 0, value.readableBytes());
value = directValue;
} else {
mustCloseValue = false;
}
valueNioBuffer = ((ReadableComponent) value).readableBuffer();
assert valueNioBuffer.isDirect();
assert valueNioBuffer.limit() == value.readableBytes();
}
2022-03-16 19:19:26 +01:00
2021-12-12 02:17:36 +01:00
try {
db.put(cfh, writeOptions, keyNioBuffer, valueNioBuffer);
} finally {
if (mustCloseValue) {
2022-03-16 19:19:26 +01:00
value.close();
2021-10-20 01:51:34 +02:00
}
2021-12-12 02:17:36 +01:00
}
} finally {
if (mustCloseKey) {
2022-03-16 19:19:26 +01:00
key.close();
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
2022-03-16 19:19:26 +01:00
ByteBuffer keyNioBuffer;
2021-12-12 02:17:36 +01:00
boolean mustCloseKey;
2022-03-16 19:19:26 +01:00
{
if (!LLUtils.isReadOnlyDirect(key)) {
// If the nio buffer is not available, copy the netty buffer into a new direct buffer
mustCloseKey = true;
var directKey = DefaultBufferAllocators.offHeapAllocator().allocate(key.readableBytes());
key.copyInto(key.readerOffset(), directKey, 0, key.readableBytes());
key = directKey;
} else {
mustCloseKey = false;
}
keyNioBuffer = ((ReadableComponent) key).readableBuffer();
assert keyNioBuffer.isDirect();
assert keyNioBuffer.limit() == key.readableBytes();
}
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) {
2022-03-16 19:19:26 +01:00
key.close();
2021-12-12 02:17:36 +01:00
}
}
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
2022-03-16 19:19:26 +01:00
ByteBuffer keyNioBuffer;
2021-12-12 02:17:36 +01:00
boolean mustCloseKey;
2022-03-16 19:19:26 +01:00
{
if (!LLUtils.isReadOnlyDirect(key)) {
// If the nio buffer is not available, copy the netty buffer into a new direct buffer
mustCloseKey = true;
var directKey = DefaultBufferAllocators.offHeapAllocator().allocate(key.readableBytes());
key.copyInto(key.readerOffset(), directKey, 0, key.readableBytes());
key = directKey;
} else {
mustCloseKey = false;
}
keyNioBuffer = ((ReadableComponent) key).readableBuffer();
assert keyNioBuffer.isDirect();
assert keyNioBuffer.limit() == key.readableBytes();
}
2021-12-12 02:17:36 +01:00
try {
db.delete(cfh, writeOptions, keyNioBuffer);
} finally {
if (mustCloseKey) {
2022-03-16 19:19:26 +01:00
key.close();
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
}