Do not create ignorable segments on revert storage-variables

besu/src/main/java/org/hyperledger/besu/cli/BesuCommand.java

@@ -3489,7 +3489,8 @@ private void setMergeConfigOptions() {
         getActualGenesisConfigOptions().getTerminalTotalDifficulty().isPresent());
   }
 
-  private void setIgnorableStorageSegments() {
+  /** Set ignorable segments in RocksDB Storage Provider plugin. */
+  public void setIgnorableStorageSegments() {
     if (!unstableChainPruningOptions.getChainDataPruningEnabled()) {
       rocksDBPlugin.addIgnorableSegmentIdentifier(KeyValueSegmentIdentifier.CHAIN_PRUNER_STATE);
     }

besu/src/main/java/org/hyperledger/besu/cli/subcommands/storage/StorageSubCommand.java

@@ -99,6 +99,8 @@ public void run() {
     }
 
     private StorageProvider getStorageProvider() {
+      // init collection of ignorable segments
+      parentCommand.parentCommand.setIgnorableStorageSegments();
       return parentCommand.parentCommand.getStorageProvider();
     }
 

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

@@ -63,7 +63,7 @@ public OptimisticRocksDBColumnarKeyValueStorage(
       initColumnHandles();
 
     } catch (final RocksDBException e) {
-      throw new StorageException(e);
+      throw parseRocksDBException(e, segments, ignorableSegments);
     }
   }
 

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

@@ -40,7 +40,10 @@
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
+import com.google.common.base.Splitter;
+import com.google.common.collect.Streams;
 import org.apache.commons.lang3.tuple.Pair;
+import org.apache.tuweni.bytes.Bytes;
 import org.rocksdb.BlockBasedTableConfig;
 import org.rocksdb.BloomFilter;
 import org.rocksdb.ColumnFamilyDescriptor;
@@ -161,7 +164,52 @@ public RocksDBColumnarKeyValueStorage(
       txOptions = new TransactionDBOptions();
       columnHandles = new ArrayList<>(columnDescriptors.size());
     } catch (RocksDBException e) {
-      throw new StorageException(e);
+      throw parseRocksDBException(e, defaultSegments, ignorableSegments);
+    }
+  }
+
+  /**
+   * Parse RocksDBException and wrap in StorageException
+   *
+   * @param ex RocksDBException
+   * @param defaultSegments segments requested to open
+   * @param ignorableSegments segments which are ignorable if not present
+   * @return StorageException wrapping the RocksDB Exception
+   */
+  protected static StorageException parseRocksDBException(
+      final RocksDBException ex,
+      final List<SegmentIdentifier> defaultSegments,
+      final List<SegmentIdentifier> ignorableSegments) {
+    String message = ex.getMessage();
+    List<SegmentIdentifier> knownSegments =
+        Streams.concat(defaultSegments.stream(), ignorableSegments.stream()).distinct().toList();
+
+    // parse out unprintable segment names for a more useful exception:
+    String columnExceptionMessagePrefix = "Column families not opened: ";
+    if (message.contains(columnExceptionMessagePrefix)) {
+      String substring = message.substring(message.indexOf(": ") + 2);
+
+      List<String> unHandledSegments = new ArrayList<>();
+      Splitter.on(", ")
+          .splitToStream(substring)
+          .forEach(
+              part -> {
+                byte[] bytes = part.getBytes(StandardCharsets.UTF_8);
+                unHandledSegments.add(
+                    knownSegments.stream()
+                        .filter(seg -> Arrays.equals(seg.getId(), bytes))
+                        .findFirst()
+                        .map(seg -> new SegmentRecord(seg.getName(), seg.getId()))
+                        .orElse(new SegmentRecord(part, bytes))
+                        .forDisplay());
+              });
+
+      return new StorageException(
+          "RocksDBException: Unhandled column families: ["
+              + unHandledSegments.stream().collect(Collectors.joining(", "))
+              + "]");
+    } else {
+      return new StorageException(ex);
     }
   }
 
@@ -356,4 +404,10 @@ void throwIfClosed() {
   }
 
   abstract RocksDB getDB();
+
+  record SegmentRecord(String name, byte[] id) {
+    public String forDisplay() {
+      return String.format("'%s'(%s)", name, Bytes.of(id).toHexString());
+    }
+  }
 }

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

@@ -66,7 +66,7 @@ public TransactionDBRocksDBColumnarKeyValueStorage(
       initColumnHandles();
 
     } catch (final RocksDBException e) {
-      throw new StorageException(e);
+      throw parseRocksDBException(e, segments, ignorableSegments);
     }
   }
 

plugins/rocksdb/src/test/java/org/hyperledger/besu/plugin/services/storage/rocksdb/segmented/RocksDBColumnarKeyValueStorageTest.java

@@ -222,7 +222,7 @@ public void dbShouldNotIgnoreExperimentalSegmentsIfExisted(@TempDir final Path t
       createSegmentedStore(testPath, Arrays.asList(TestSegment.FOO, TestSegment.BAR), List.of());
       fail("DB without knowledge of experimental column family should fail");
     } catch (StorageException e) {
-      assertThat(e.getMessage()).contains("Column families not opened");
+      assertThat(e.getMessage()).contains("Unhandled column families");
     }
 
     // Even if the column family is marked as ignored, as long as it exists, it will not be ignored
@@ -265,7 +265,7 @@ public void dbWillBeBackwardIncompatibleAfterExperimentalSegmentsAreAdded(
       createSegmentedStore(testPath, Arrays.asList(TestSegment.FOO, TestSegment.BAR), List.of());
       fail("DB without knowledge of experimental column family should fail");
     } catch (StorageException e) {
-      assertThat(e.getMessage()).contains("Column families not opened");
+      assertThat(e.getMessage()).contains("Unhandled column families");
     }
   }