StackableDB optionally take shared ownership of the underlying DB

include/rocksdb/utilities/stackable_db.h

@@ -4,6 +4,7 @@
 
 #pragma once
 #include <map>
+#include <memory>
 #include <string>
 #include "rocksdb/db.h"
 
@@ -18,11 +19,19 @@ namespace rocksdb {
 // This class contains APIs to stack rocksdb wrappers.Eg. Stack TTL over base d
 class StackableDB : public DB {
  public:
-  // StackableDB is the owner of db now!
+  // StackableDB take sole ownership of the underlying db.
   explicit StackableDB(DB* db) : db_(db) {}
 
+  // StackableDB take shared ownership of the underlying db.
+  explicit StackableDB(std::shared_ptr<DB>& db)
+      : db_(db.get()), shared_db_(db) {}
+
   ~StackableDB() {
-    delete db_;
+    if (shared_db_ == nullptr) {
+      delete db_;
+    } else {
+      assert(shared_db_.get() == db_);
+    }
   }
 
   virtual Status Close() override { return db_->Close(); }
@@ -375,6 +384,7 @@ class StackableDB : public DB {
 
  protected:
   DB* db_;
+  std::shared_ptr<DB> shared_db_;
 };
 
 } //  namespace rocksdb

table/block_based_table_reader.h

@@ -215,14 +215,17 @@ class BlockBasedTable : public TableReader {
  private:
   friend class MockedBlockBasedTable;
   // input_iter: if it is not null, update this one and return it as Iterator
-  static BlockIter* NewDataBlockIterator(
-      Rep* rep, const ReadOptions& ro, const Slice& index_value,
-      BlockIter* input_iter = nullptr, bool is_index = false,
-      GetContext* get_context = nullptr);
-  static BlockIter* NewDataBlockIterator(
-      Rep* rep, const ReadOptions& ro, const BlockHandle& block_hanlde,
-      BlockIter* input_iter = nullptr, bool is_index = false,
-      GetContext* get_context = nullptr, Status s = Status());
+  static BlockIter* NewDataBlockIterator(Rep* rep, const ReadOptions& ro,
+                                         const Slice& index_value,
+                                         BlockIter* input_iter = nullptr,
+                                         bool is_index = false,
+                                         GetContext* get_context = nullptr);
+  static BlockIter* NewDataBlockIterator(Rep* rep, const ReadOptions& ro,
+                                         const BlockHandle& block_hanlde,
+                                         BlockIter* input_iter = nullptr,
+                                         bool is_index = false,
+                                         GetContext* get_context = nullptr,
+                                         Status s = Status());
   // If block cache enabled (compressed or uncompressed), looks for the block
   // identified by handle in (1) uncompressed cache, (2) compressed cache, and
   // then (3) file. If found, inserts into the cache(s) that were searched