Rocksdb plugin to support OptimisticTransactionDb and TransactionDb

plugins/rocksdb/build.gradle

@@ -47,6 +47,8 @@ dependencies {
   implementation 'io.prometheus:simpleclient'
   implementation 'org.apache.tuweni:tuweni-bytes'
   implementation 'org.rocksdb:rocksdbjni'
+  implementation project(path: ':ethereum:core')
+  implementation project(path: ':ethereum:core')
 
   testImplementation project(':testutil')
 

plugins/rocksdb/src/main/java/org/hyperledger/besu/plugin/services/storage/rocksdb/RocksDBKeyValueStorageFactory.java

@@ -16,6 +16,7 @@
 
 import static com.google.common.base.Preconditions.checkNotNull;
 
+import org.hyperledger.besu.ethereum.worldstate.DataStorageFormat;
 import org.hyperledger.besu.plugin.services.BesuConfiguration;
 import org.hyperledger.besu.plugin.services.MetricsSystem;
 import org.hyperledger.besu.plugin.services.exception.StorageException;
@@ -26,6 +27,8 @@
 import org.hyperledger.besu.plugin.services.storage.rocksdb.configuration.RocksDBConfiguration;
 import org.hyperledger.besu.plugin.services.storage.rocksdb.configuration.RocksDBConfigurationBuilder;
 import org.hyperledger.besu.plugin.services.storage.rocksdb.configuration.RocksDBFactoryConfiguration;
+import org.hyperledger.besu.plugin.services.storage.rocksdb.segmented.OptimisticRocksDBColumnarKeyValueStorage;
+import org.hyperledger.besu.plugin.services.storage.rocksdb.segmented.PessimisticRocksDBColumnarKeyValueStorage;
 import org.hyperledger.besu.plugin.services.storage.rocksdb.segmented.RocksDBColumnarKeyValueStorage;
 import org.hyperledger.besu.plugin.services.storage.rocksdb.unsegmented.RocksDBKeyValueStorage;
 import org.hyperledger.besu.services.kvstore.SegmentedKeyValueStorageAdapter;
@@ -149,7 +152,8 @@ public KeyValueStorage create(
       final BesuConfiguration commonConfiguration,
       final MetricsSystem metricsSystem)
       throws StorageException {
-
+    final boolean isForestStorageFormat =
+        DataStorageFormat.FOREST.getDatabaseVersion() == commonConfiguration.getDatabaseVersion();
     if (requiresInit()) {
       init(commonConfiguration);
     }
@@ -177,19 +181,35 @@ public KeyValueStorage create(
                 segments.stream()
                     .filter(segmentId -> segmentId.includeInDatabaseVersion(databaseVersion))
                     .collect(Collectors.toList());
-
-            segmentedStorage =
-                new RocksDBColumnarKeyValueStorage(
-                    rocksDBConfiguration,
-                    segmentsForVersion,
-                    ignorableSegments,
-                    metricsSystem,
-                    rocksDBMetricsFactory);
+            if (isForestStorageFormat) {
+              LOG.info("FOREST mode detected, using pessimistic DB.");
+              segmentedStorage =
+                  new PessimisticRocksDBColumnarKeyValueStorage(
+                      rocksDBConfiguration,
+                      segmentsForVersion,
+                      ignorableSegments,
+                      metricsSystem,
+                      rocksDBMetricsFactory);
+            } else {
+              LOG.info("Using Optimistic DB.");
+              segmentedStorage =
+                  new OptimisticRocksDBColumnarKeyValueStorage(
+                      rocksDBConfiguration,
+                      segmentsForVersion,
+                      ignorableSegments,
+                      metricsSystem,
+                      rocksDBMetricsFactory);
+            }
           }
           final RocksDbSegmentIdentifier rocksSegment =
               segmentedStorage.getSegmentIdentifierByName(segment);
-          return new SegmentedKeyValueStorageAdapter<>(
-              segment, segmentedStorage, () -> segmentedStorage.takeSnapshot(rocksSegment));
+
+          if (isForestStorageFormat) {
+            return new SegmentedKeyValueStorageAdapter<>(segment, segmentedStorage);
+          } else {
+            return new SegmentedKeyValueStorageAdapter<>(
+                segment, segmentedStorage, () -> segmentedStorage.takeSnapshot(rocksSegment));
+          }
         }
       default:
         {

plugins/rocksdb/src/main/java/org/hyperledger/besu/plugin/services/storage/rocksdb/RocksDBMetricsFactory.java

@@ -23,7 +23,7 @@
 import org.hyperledger.besu.plugin.services.metrics.OperationTimer;
 import org.hyperledger.besu.plugin.services.storage.rocksdb.configuration.RocksDBConfiguration;
 
-import org.rocksdb.OptimisticTransactionDB;
+import org.rocksdb.RocksDB;
 import org.rocksdb.RocksDBException;
 import org.rocksdb.Statistics;
 import org.slf4j.Logger;
@@ -72,7 +72,7 @@ public RocksDBMetricsFactory(
   public RocksDBMetrics create(
       final MetricsSystem metricsSystem,
       final RocksDBConfiguration rocksDbConfiguration,
-      final OptimisticTransactionDB db,
+      final RocksDB db,
       final Statistics stats) {
     final OperationTimer readLatency =
         metricsSystem

plugins/rocksdb/src/main/java/org/hyperledger/besu/plugin/services/storage/rocksdb/RocksDbSegmentIdentifier.java

@@ -21,13 +21,13 @@
 
 import org.rocksdb.ColumnFamilyDescriptor;
 import org.rocksdb.ColumnFamilyHandle;
-import org.rocksdb.OptimisticTransactionDB;
+import org.rocksdb.RocksDB;
 import org.rocksdb.RocksDBException;
 
 /** The RocksDb segment identifier. */
 public class RocksDbSegmentIdentifier {
 
-  private final OptimisticTransactionDB db;
+  private final RocksDB db;
   private final AtomicReference<ColumnFamilyHandle> reference;
 
   /**
@@ -36,8 +36,7 @@ public class RocksDbSegmentIdentifier {
    * @param db the db
    * @param columnFamilyHandle the column family handle
    */
-  public RocksDbSegmentIdentifier(
-      final OptimisticTransactionDB db, final ColumnFamilyHandle columnFamilyHandle) {
+  public RocksDbSegmentIdentifier(final RocksDB db, final ColumnFamilyHandle columnFamilyHandle) {
     this.db = db;
     this.reference = new AtomicReference<>(columnFamilyHandle);
   }

plugins/rocksdb/src/main/java/org/hyperledger/besu/plugin/services/storage/rocksdb/segmented/OptimisticRocksDBColumnarKeyValueStorage.java

@@ -0,0 +1,103 @@
+/*
+ * Copyright Hyperledger Besu Contributors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package org.hyperledger.besu.plugin.services.storage.rocksdb.segmented;
+
+import org.hyperledger.besu.plugin.services.MetricsSystem;
+import org.hyperledger.besu.plugin.services.exception.StorageException;
+import org.hyperledger.besu.plugin.services.storage.SegmentIdentifier;
+import org.hyperledger.besu.plugin.services.storage.rocksdb.RocksDBMetricsFactory;
+import org.hyperledger.besu.plugin.services.storage.rocksdb.RocksDbSegmentIdentifier;
+import org.hyperledger.besu.plugin.services.storage.rocksdb.configuration.RocksDBConfiguration;
+import org.hyperledger.besu.services.kvstore.SegmentedKeyValueStorage;
+import org.hyperledger.besu.services.kvstore.SegmentedKeyValueStorageTransactionTransitionValidatorDecorator;
+
+import java.util.List;
+
+import org.rocksdb.OptimisticTransactionDB;
+import org.rocksdb.RocksDB;
+import org.rocksdb.RocksDBException;
+import org.rocksdb.WriteOptions;
+
+/** Optimistic RocksDB Columnar key value storage */
+public class OptimisticRocksDBColumnarKeyValueStorage extends RocksDBColumnarKeyValueStorage {
+  private final OptimisticTransactionDB db;
+
+  /**
+   * Instantiates a new Rocks db columnar key value optimistic storage.
+   *
+   * @param configuration the configuration
+   * @param segments the segments
+   * @param ignorableSegments the ignorable segments
+   * @param metricsSystem the metrics system
+   * @param rocksDBMetricsFactory the rocks db metrics factory
+   * @throws StorageException the storage exception
+   */
+  public OptimisticRocksDBColumnarKeyValueStorage(
+      final RocksDBConfiguration configuration,
+      final List<SegmentIdentifier> segments,
+      final List<SegmentIdentifier> ignorableSegments,
+      final MetricsSystem metricsSystem,
+      final RocksDBMetricsFactory rocksDBMetricsFactory)
+      throws StorageException {
+    super(configuration, segments, ignorableSegments, metricsSystem, rocksDBMetricsFactory);
+    try {
+
+      db =
+          OptimisticTransactionDB.open(
+              options, configuration.getDatabaseDir().toString(), columnDescriptors, columnHandles);
+      initMetrics();
+      initColumnHandler();
+
+    } catch (final RocksDBException e) {
+      throw new StorageException(e);
+    }
+  }
+
+  @Override
+  RocksDB getDB() {
+    return db;
+  }
+
+  /**
+   * Start a transaction
+   *
+   * @return the new transaction started
+   * @throws StorageException the storage exception
+   */
+  @Override
+  public SegmentedKeyValueStorage.Transaction<RocksDbSegmentIdentifier> startTransaction()
+      throws StorageException {
+    throwIfClosed();
+    final WriteOptions writeOptions = new WriteOptions();
+    writeOptions.setIgnoreMissingColumnFamilies(true);
+    return new SegmentedKeyValueStorageTransactionTransitionValidatorDecorator<>(
+        new RocksDBColumnarKeyValueStorage.RocksDbTransaction(
+            db.beginTransaction(writeOptions), writeOptions));
+  }
+
+  /**
+   * Take snapshot RocksDb columnar key value snapshot.
+   *
+   * @param segment the segment
+   * @return the RocksDb columnar key value snapshot
+   * @throws StorageException the storage exception
+   */
+  @Override
+  public RocksDBColumnarKeyValueSnapshot takeSnapshot(final RocksDbSegmentIdentifier segment)
+      throws StorageException {
+    throwIfClosed();
+    return new RocksDBColumnarKeyValueSnapshot(db, segment, metrics);
+  }
+}

plugins/rocksdb/src/main/java/org/hyperledger/besu/plugin/services/storage/rocksdb/segmented/PessimisticRocksDBColumnarKeyValueStorage.java

@@ -0,0 +1,103 @@
+/*
+ * Copyright Hyperledger Besu Contributors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package org.hyperledger.besu.plugin.services.storage.rocksdb.segmented;
+
+import org.hyperledger.besu.plugin.services.MetricsSystem;
+import org.hyperledger.besu.plugin.services.exception.StorageException;
+import org.hyperledger.besu.plugin.services.storage.SegmentIdentifier;
+import org.hyperledger.besu.plugin.services.storage.rocksdb.RocksDBMetricsFactory;
+import org.hyperledger.besu.plugin.services.storage.rocksdb.RocksDbSegmentIdentifier;
+import org.hyperledger.besu.plugin.services.storage.rocksdb.configuration.RocksDBConfiguration;
+import org.hyperledger.besu.services.kvstore.SegmentedKeyValueStorageTransactionTransitionValidatorDecorator;
+
+import java.util.List;
+
+import org.rocksdb.RocksDB;
+import org.rocksdb.RocksDBException;
+import org.rocksdb.TransactionDB;
+import org.rocksdb.WriteOptions;
+
+/** Pessimistic RocksDB Columnar key value storage */
+public class PessimisticRocksDBColumnarKeyValueStorage extends RocksDBColumnarKeyValueStorage {
+
+  private final TransactionDB db;
+
+  /**
+   * The constructor of PessimisticRocksDBColumnarKeyValueStorage
+   *
+   * @param configuration the RocksDB configuration
+   * @param segments the segments
+   * @param ignorableSegments the ignorable segments
+   * @param metricsSystem the metrics system
+   * @param rocksDBMetricsFactory the rocksdb metrics factory
+   * @throws StorageException the storage exception
+   */
+  public PessimisticRocksDBColumnarKeyValueStorage(
+      final RocksDBConfiguration configuration,
+      final List<SegmentIdentifier> segments,
+      final List<SegmentIdentifier> ignorableSegments,
+      final MetricsSystem metricsSystem,
+      final RocksDBMetricsFactory rocksDBMetricsFactory)
+      throws StorageException {
+    super(configuration, segments, ignorableSegments, metricsSystem, rocksDBMetricsFactory);
+    try {
+
+      db =
+          TransactionDB.open(
+              options,
+              txOptions,
+              configuration.getDatabaseDir().toString(),
+              columnDescriptors,
+              columnHandles);
+      initMetrics();
+      initColumnHandler();
+
+    } catch (final RocksDBException e) {
+      throw new StorageException(e);
+    }
+  }
+
+  @Override
+  RocksDB getDB() {
+    return db;
+  }
+
+  /**
+   * Start a transaction
+   *
+   * @return the new transaction started
+   * @throws StorageException the storage exception
+   */
+  @Override
+  public Transaction<RocksDbSegmentIdentifier> startTransaction() throws StorageException {
+    throwIfClosed();
+    final WriteOptions writeOptions = new WriteOptions();
+    writeOptions.setIgnoreMissingColumnFamilies(true);
+    return new SegmentedKeyValueStorageTransactionTransitionValidatorDecorator<>(
+        new RocksDbTransaction(db.beginTransaction(writeOptions), writeOptions));
+  }
+
+  /**
+   * Not supported for Pessimistic.
+   *
+   * @param segment the segment
+   * @return the RocksDb columnar key value snapshot
+   * @throws StorageException the storage exception
+   */
+  @Override
+  public RocksDBColumnarKeyValueSnapshot takeSnapshot(final RocksDbSegmentIdentifier segment) {
+    throw new UnsupportedOperationException("Not supported");
+  }
+}