SERVER-40478 Merge StorageEngine class with KVStorageEngine

30 files changed, 200 insertions, 207 deletions

diff --git a/src/mongo/db/catalog/SConscript b/src/mongo/db/catalog/SConscript
index af02608ff4b..4de2692e67a 100644
--- a/src/mongo/db/catalog/SConscript
+++ b/src/mongo/db/catalog/SConscript
@@ -425,7 +425,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/service_context',
         '$BUILD_DIR/mongo/db/storage/key_string',
         '$BUILD_DIR/mongo/db/storage/kv/kv_engine_core',
-        '$BUILD_DIR/mongo/db/storage/kv/kv_storage_engine',
+        '$BUILD_DIR/mongo/db/storage/kv/storage_engine',
         '$BUILD_DIR/mongo/db/system_index',
         '$BUILD_DIR/mongo/db/ttl_collection_cache',
         '$BUILD_DIR/mongo/db/views/views_mongod',
diff --git a/src/mongo/db/catalog/catalog_control_test.cpp b/src/mongo/db/catalog/catalog_control_test.cpp
index c487801524e..961e1a83997 100644
--- a/src/mongo/db/catalog/catalog_control_test.cpp
+++ b/src/mongo/db/catalog/catalog_control_test.cpp
@@ -102,6 +102,18 @@ public:
     int64_t sizeOnDiskForDb(OperationContext* opCtx, StringData dbName) {
         return 0;
     }
+    KVEngine* getEngine() {
+        return nullptr;
+    }
+    const KVEngine* getEngine() const {
+        return nullptr;
+    }
+    KVCatalog* getCatalog() {
+        return nullptr;
+    }
+    const KVCatalog* getCatalog() const {
+        return nullptr;
+    }
 };
 
 /**
diff --git a/src/mongo/db/catalog/database_impl.cpp b/src/mongo/db/catalog/database_impl.cpp
index 857a8d0a9b9..3e981742a6e 100644
--- a/src/mongo/db/catalog/database_impl.cpp
+++ b/src/mongo/db/catalog/database_impl.cpp
@@ -65,7 +65,7 @@
 #include "mongo/db/server_options.h"
 #include "mongo/db/service_context.h"
 #include "mongo/db/stats/top.h"
-#include "mongo/db/storage/kv/kv_storage_engine.h"
+#include "mongo/db/storage/kv/kv_catalog.h"
 #include "mongo/db/storage/recovery_unit.h"
 #include "mongo/db/storage/storage_engine.h"
 #include "mongo/db/storage/storage_engine_init.h"
@@ -474,8 +474,7 @@ Status DatabaseImpl::_finishDropCollection(OperationContext* opCtx,
     UUID uuid = *collection->uuid();
     log() << "Finishing collection drop for " << nss << " (" << uuid << ").";
 
-    auto storageEngine =
-        checked_cast<KVStorageEngine*>(opCtx->getServiceContext()->getStorageEngine());
+    auto storageEngine = opCtx->getServiceContext()->getStorageEngine();
     auto status = storageEngine->getCatalog()->dropCollection(opCtx, nss);
     if (!status.isOK())
         return status;
@@ -533,8 +532,7 @@ Status DatabaseImpl::renameCollection(OperationContext* opCtx,
 
     Top::get(opCtx->getServiceContext()).collectionDropped(fromNss);
 
-    auto storageEngine =
-        checked_cast<KVStorageEngine*>(opCtx->getServiceContext()->getStorageEngine());
+    auto storageEngine = opCtx->getServiceContext()->getStorageEngine();
     Status status = storageEngine->getCatalog()->renameCollection(opCtx, fromNss, toNss, stayTemp);
 
     // Set the namespace of 'collToRename' from within the CollectionCatalog. This is necessary
@@ -663,8 +661,7 @@ Collection* DatabaseImpl::createCollection(OperationContext* opCtx,
           << " UUID: " << optionsWithUUID.uuid.get() << " and options: " << options.toBSON();
 
     // Create CollectionCatalogEntry
-    auto storageEngine =
-        checked_cast<KVStorageEngine*>(opCtx->getServiceContext()->getStorageEngine());
+    auto storageEngine = opCtx->getServiceContext()->getStorageEngine();
     auto statusWithCatalogEntry = storageEngine->getCatalog()->createCollection(
         opCtx, nss, optionsWithUUID, true /*allocateDefaultSpace*/);
     massertStatusOK(statusWithCatalogEntry.getStatus());
diff --git a/src/mongo/db/catalog/index_catalog_impl.cpp b/src/mongo/db/catalog/index_catalog_impl.cpp
index 0d3ed9abda6..a0e45df326c 100644
--- a/src/mongo/db/catalog/index_catalog_impl.cpp
+++ b/src/mongo/db/catalog/index_catalog_impl.cpp
@@ -66,7 +66,7 @@
 #include "mongo/db/server_options.h"
 #include "mongo/db/service_context.h"
 #include "mongo/db/storage/kv/kv_catalog.h"
-#include "mongo/db/storage/kv/kv_storage_engine.h"
+#include "mongo/db/storage/kv/kv_engine.h"
 #include "mongo/db/storage/storage_engine_init.h"
 #include "mongo/util/assert_util.h"
 #include "mongo/util/log.h"
@@ -148,8 +148,7 @@ IndexCatalogEntry* IndexCatalogImpl::_setupInMemoryStructures(
 
     IndexDescriptor* desc = entry->descriptor();
 
-    KVStorageEngine* engine =
-        checked_cast<KVStorageEngine*>(opCtx->getServiceContext()->getStorageEngine());
+    auto engine = opCtx->getServiceContext()->getStorageEngine();
     std::string ident =
         engine->getCatalog()->getIndexIdent(opCtx, _collection->ns(), desc->indexName());
 
@@ -1226,20 +1225,17 @@ const IndexDescriptor* IndexCatalogImpl::refreshEntry(OperationContext* opCtx,
     invariant(_collection->getCatalogEntry()->isIndexReady(opCtx, indexName));
 
     // Delete the IndexCatalogEntry that owns this descriptor.  After deletion, 'oldDesc' is
-    // invalid and should not be dereferenced. Also, invalidate the index from the
-    // CollectionInfoCache.
+    // invalid and should not be dereferenced.
     auto oldEntry = _readyIndexes.release(oldDesc);
     invariant(oldEntry);
     opCtx->recoveryUnit()->registerChange(
         new IndexRemoveChange(opCtx, _collection, &_readyIndexes, std::move(oldEntry)));
-    _collection->infoCache()->droppedIndex(opCtx, indexName);
 
     // Ask the CollectionCatalogEntry for the new index spec.
     BSONObj spec = _collection->getCatalogEntry()->getIndexSpec(opCtx, indexName).getOwned();
     BSONObj keyPattern = spec.getObjectField("key");
 
-    // Re-register this index in the index catalog with the new spec. Also, add the new index
-    // to the CollectionInfoCache.
+    // Re-register this index in the index catalog with the new spec.
     auto newDesc =
         std::make_unique<IndexDescriptor>(_collection, _getAccessMethodName(keyPattern), spec);
     const bool initFromDisk = false;
@@ -1247,7 +1243,6 @@ const IndexDescriptor* IndexCatalogImpl::refreshEntry(OperationContext* opCtx,
     const IndexCatalogEntry* newEntry =
         _setupInMemoryStructures(opCtx, std::move(newDesc), initFromDisk, isReadyIndex);
     invariant(newEntry->isReady(opCtx));
-    _collection->infoCache()->addedIndex(opCtx, newEntry->descriptor());
 
     // Return the new descriptor.
     return newEntry->descriptor();
@@ -1266,16 +1261,6 @@ Status IndexCatalogImpl::_indexKeys(OperationContext* opCtx,
                                     int64_t* keysInsertedOut) {
     Status status = Status::OK();
     if (index->isHybridBuilding()) {
-        // The side table interface accepts only records that meet the criteria for this partial
-        // index.
-        // For non-hybrid builds, the decision to use the filter for the partial index is left to
-        // the IndexAccessMethod. See SERVER-28975 for details.
-        if (auto filter = index->getFilterExpression()) {
-            if (!filter->matchesBSON(obj)) {
-                return Status::OK();
-            }
-        }
-
         int64_t inserted;
         status = index->indexBuildInterceptor()->sideWrite(opCtx,
                                                            keys,
diff --git a/src/mongo/db/storage/README.md b/src/mongo/db/storage/README.md
index 37699575ded..17d1af8f2e0 100644
--- a/src/mongo/db/storage/README.md
+++ b/src/mongo/db/storage/README.md
@@ -28,7 +28,7 @@ Concepts
 A database contains one or more collections, each with a number of indexes, and a catalog listing
 them. All MongoDB collections are implemented with record stores: one for the documents themselves,
 and one for each index. By using the KVEngine class, you only have to deal with the abstraction, as
-the KVStorageEngine implements the StorageEngine interface, using record stores for catalogs and
+the StorageEngineImpl implements the StorageEngine interface, using record stores for catalogs and
 indexes.
 
 #### Record Identities
diff --git a/src/mongo/db/storage/biggie/SConscript b/src/mongo/db/storage/biggie/SConscript
index 09526f74044..a86d5a8cbfd 100644
--- a/src/mongo/db/storage/biggie/SConscript
+++ b/src/mongo/db/storage/biggie/SConscript
@@ -34,7 +34,7 @@ env.Library(
     ],
     LIBDEPS=[
         '$BUILD_DIR/mongo/db/storage/kv/kv_engine_core',
-        '$BUILD_DIR/mongo/db/storage/kv/kv_storage_engine',
+        '$BUILD_DIR/mongo/db/storage/kv/storage_engine',
         'storage_biggie_core',
     ],
     LIBDEPS_PRIVATE=[
diff --git a/src/mongo/db/storage/biggie/biggie_init.cpp b/src/mongo/db/storage/biggie/biggie_init.cpp
index ec8912a3d61..5a006a7342c 100644
--- a/src/mongo/db/storage/biggie/biggie_init.cpp
+++ b/src/mongo/db/storage/biggie/biggie_init.cpp
@@ -32,7 +32,7 @@
 #include "mongo/base/init.h"
 #include "mongo/db/service_context.h"
 #include "mongo/db/storage/biggie/biggie_kv_engine.h"
-#include "mongo/db/storage/kv/kv_storage_engine.h"
+#include "mongo/db/storage/kv/storage_engine_impl.h"
 #include "mongo/db/storage/storage_engine_init.h"
 #include "mongo/db/storage/storage_options.h"
 
@@ -44,10 +44,10 @@ class BiggieStorageEngineFactory : public StorageEngine::Factory {
 public:
     virtual StorageEngine* create(const StorageGlobalParams& params,
                                   const StorageEngineLockFile* lockFile) const {
-        KVStorageEngineOptions options;
+        StorageEngineOptions options;
         options.directoryPerDB = params.directoryperdb;
         options.forRepair = params.repair;
-        return new KVStorageEngine(new KVEngine(), options);
+        return new StorageEngineImpl(new KVEngine(), options);
     }
 
     virtual StringData getCanonicalName() const {
diff --git a/src/mongo/db/storage/devnull/SConscript b/src/mongo/db/storage/devnull/SConscript
index 24f10a1b7f1..795b5992c34 100644
--- a/src/mongo/db/storage/devnull/SConscript
+++ b/src/mongo/db/storage/devnull/SConscript
@@ -26,7 +26,7 @@ env.Library(
     LIBDEPS=[
         'storage_devnull_core',
         '$BUILD_DIR/mongo/db/storage/kv/kv_engine_core',
-        '$BUILD_DIR/mongo/db/storage/kv/kv_storage_engine',
+        '$BUILD_DIR/mongo/db/storage/kv/storage_engine',
     ],
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/storage/storage_engine_common',
diff --git a/src/mongo/db/storage/devnull/devnull_init.cpp b/src/mongo/db/storage/devnull/devnull_init.cpp
index 64eee680fd2..e84b3767899 100644
--- a/src/mongo/db/storage/devnull/devnull_init.cpp
+++ b/src/mongo/db/storage/devnull/devnull_init.cpp
@@ -32,7 +32,7 @@
 #include "mongo/base/init.h"
 #include "mongo/db/service_context.h"
 #include "mongo/db/storage/devnull/devnull_kv_engine.h"
-#include "mongo/db/storage/kv/kv_storage_engine.h"
+#include "mongo/db/storage/kv/storage_engine_impl.h"
 #include "mongo/db/storage/storage_engine_init.h"
 #include "mongo/db/storage/storage_options.h"
 
@@ -43,10 +43,10 @@ class DevNullStorageEngineFactory : public StorageEngine::Factory {
 public:
     virtual StorageEngine* create(const StorageGlobalParams& params,
                                   const StorageEngineLockFile* lockFile) const {
-        KVStorageEngineOptions options;
+        StorageEngineOptions options;
         options.directoryPerDB = params.directoryperdb;
         options.forRepair = params.repair;
-        return new KVStorageEngine(new DevNullKVEngine(), options);
+        return new StorageEngineImpl(new DevNullKVEngine(), options);
     }
 
     virtual StringData getCanonicalName() const {
diff --git a/src/mongo/db/storage/ephemeral_for_test/SConscript b/src/mongo/db/storage/ephemeral_for_test/SConscript
index 64a6f1f520e..cf015f545e4 100644
--- a/src/mongo/db/storage/ephemeral_for_test/SConscript
+++ b/src/mongo/db/storage/ephemeral_for_test/SConscript
@@ -43,7 +43,7 @@ env.Library(
     LIBDEPS= [
         'storage_ephemeral_for_test_core',
         '$BUILD_DIR/mongo/db/storage/kv/kv_engine_core',
-        '$BUILD_DIR/mongo/db/storage/kv/kv_storage_engine',
+        '$BUILD_DIR/mongo/db/storage/kv/storage_engine',
     ],
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/storage/storage_engine_common',
diff --git a/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_init.cpp b/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_init.cpp
index 622adf66051..1636dd3be63 100644
--- a/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_init.cpp
+++ b/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_init.cpp
@@ -31,7 +31,7 @@
 
 #include "mongo/db/service_context.h"
 #include "mongo/db/storage/ephemeral_for_test/ephemeral_for_test_engine.h"
-#include "mongo/db/storage/kv/kv_storage_engine.h"
+#include "mongo/db/storage/kv/storage_engine_impl.h"
 #include "mongo/db/storage/storage_engine_init.h"
 #include "mongo/db/storage/storage_options.h"
 
@@ -48,10 +48,10 @@ public:
                 "ephemeralForTest does not support --groupCollections",
                 !params.groupCollections);
 
-        KVStorageEngineOptions options;
+        StorageEngineOptions options;
         options.directoryPerDB = params.directoryperdb;
         options.forRepair = params.repair;
-        return new KVStorageEngine(new EphemeralForTestEngine(), options);
+        return new StorageEngineImpl(new EphemeralForTestEngine(), options);
     }
 
     virtual StringData getCanonicalName() const {
diff --git a/src/mongo/db/storage/kv/SConscript b/src/mongo/db/storage/kv/SConscript
index 74b955d1f7b..69bba58c8fc 100644
--- a/src/mongo/db/storage/kv/SConscript
+++ b/src/mongo/db/storage/kv/SConscript
@@ -45,9 +45,9 @@ env.Library(
 
 # Should not be referenced outside this SConscript file.
 env.Library(
-    target='kv_storage_engine',
+    target='storage_engine',
     source=[
-        'kv_storage_engine.cpp',
+        'storage_engine_impl.cpp',
         'temporary_kv_record_store.cpp',
     ],
     LIBDEPS=[
@@ -107,7 +107,7 @@ env.CppUnitTest(
         '$BUILD_DIR/mongo/db/storage/devnull/storage_devnull_core',
         '$BUILD_DIR/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_record_store',
         'kv_engine_core',
-        'kv_storage_engine',
+        'storage_engine',
     ],
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/auth/authmocks'
@@ -115,9 +115,9 @@ env.CppUnitTest(
 )
 
 env.CppUnitTest(
-    target='kv_storage_engine_test',
+    target='storage_engine_test',
     source=[
-        'kv_storage_engine_test.cpp',
+        'storage_engine_test.cpp',
     ],
     LIBDEPS=[
         '$BUILD_DIR/mongo/db/catalog_raii',
@@ -130,6 +130,6 @@ env.CppUnitTest(
         '$BUILD_DIR/mongo/db/storage/ephemeral_for_test/storage_ephemeral_for_test_core',
         '$BUILD_DIR/mongo/db/storage/storage_repair_observer',
         'kv_engine_core',
-        'kv_storage_engine',
+        'storage_engine',
     ],
 )
diff --git a/src/mongo/db/storage/kv/kv_catalog.cpp b/src/mongo/db/storage/kv/kv_catalog.cpp
index 1df1227de31..dcecb8c429c 100644
--- a/src/mongo/db/storage/kv/kv_catalog.cpp
+++ b/src/mongo/db/storage/kv/kv_catalog.cpp
@@ -42,7 +42,7 @@
 #include "mongo/db/storage/kv/kv_catalog_feature_tracker.h"
 #include "mongo/db/storage/kv/kv_collection_catalog_entry.h"
 #include "mongo/db/storage/kv/kv_engine.h"
-#include "mongo/db/storage/kv/kv_storage_engine_interface.h"
+#include "mongo/db/storage/kv/storage_engine_interface.h"
 #include "mongo/db/storage/record_store.h"
 #include "mongo/db/storage/recovery_unit.h"
 #include "mongo/platform/bits.h"
@@ -321,7 +321,7 @@ void KVCatalog::FeatureTracker::putInfo(OperationContext* opCtx, const FeatureBi
 KVCatalog::KVCatalog(RecordStore* rs,
                      bool directoryPerDb,
                      bool directoryForIndexes,
-                     KVStorageEngineInterface* engine)
+                     StorageEngineInterface* engine)
     : _rs(rs),
       _directoryPerDb(directoryPerDb),
       _directoryForIndexes(directoryForIndexes),
@@ -838,7 +838,7 @@ Status KVCatalog::dropCollection(OperationContext* opCtx, const NamespaceString&
     // drop the collection only on WUOW::commit().
     opCtx->recoveryUnit()->onCommit(
         [ opCtx, catalog = this, nss, uuid, ident ](boost::optional<Timestamp> commitTimestamp) {
-            KVStorageEngineInterface* engine = catalog->_engine;
+            StorageEngineInterface* engine = catalog->_engine;
             auto storageEngine = engine->getStorageEngine();
             if (storageEngine->supportsPendingDrops() && commitTimestamp) {
                 log() << "Deferring table drop for collection '" << nss << "' (" << uuid << ")"
diff --git a/src/mongo/db/storage/kv/kv_catalog.h b/src/mongo/db/storage/kv/kv_catalog.h
index a173209b53d..957b12c38e1 100644
--- a/src/mongo/db/storage/kv/kv_catalog.h
+++ b/src/mongo/db/storage/kv/kv_catalog.h
@@ -45,7 +45,7 @@ namespace mongo {
 
 class OperationContext;
 class RecordStore;
-class KVStorageEngineInterface;
+class StorageEngineInterface;
 
 class KVCatalog {
 public:
@@ -59,7 +59,7 @@ public:
     KVCatalog(RecordStore* rs,
               bool directoryPerDb,
               bool directoryForIndexes,
-              KVStorageEngineInterface* engine);
+              StorageEngineInterface* engine);
     ~KVCatalog();
 
     void init(OperationContext* opCtx);
@@ -132,9 +132,9 @@ private:
     class AddIdentChange;
     class RemoveIdentChange;
 
-    friend class KVStorageEngine;
+    friend class StorageEngineImpl;
     friend class KVCatalogTest;
-    friend class KVStorageEngineTest;
+    friend class StorageEngineTest;
 
     BSONObj _findEntry(OperationContext* opCtx,
                        const NamespaceString& nss,
@@ -182,6 +182,6 @@ private:
     // guaranteed to be non-null after KVCatalog::init() is called.
     std::unique_ptr<FeatureTracker> _featureTracker;
 
-    KVStorageEngineInterface* const _engine;
+    StorageEngineInterface* const _engine;
 };
 }
diff --git a/src/mongo/db/storage/kv/kv_collection_catalog_entry.cpp b/src/mongo/db/storage/kv/kv_collection_catalog_entry.cpp
index a773e244e70..43150b4d6b3 100644
--- a/src/mongo/db/storage/kv/kv_collection_catalog_entry.cpp
+++ b/src/mongo/db/storage/kv/kv_collection_catalog_entry.cpp
@@ -40,7 +40,7 @@
 #include "mongo/db/storage/kv/kv_catalog.h"
 #include "mongo/db/storage/kv/kv_catalog_feature_tracker.h"
 #include "mongo/db/storage/kv/kv_engine.h"
-#include "mongo/db/storage/kv/kv_storage_engine.h"
+#include "mongo/db/storage/kv/storage_engine_impl.h"
 #include "mongo/util/log.h"
 
 namespace mongo {
@@ -113,7 +113,7 @@ public:
 };
 
 
-KVCollectionCatalogEntry::KVCollectionCatalogEntry(KVStorageEngineInterface* engine,
+KVCollectionCatalogEntry::KVCollectionCatalogEntry(StorageEngineInterface* engine,
                                                    KVCatalog* catalog,
                                                    StringData ns,
                                                    StringData ident,
diff --git a/src/mongo/db/storage/kv/kv_collection_catalog_entry.h b/src/mongo/db/storage/kv/kv_collection_catalog_entry.h
index cd8643bbcb7..644a00f8943 100644
--- a/src/mongo/db/storage/kv/kv_collection_catalog_entry.h
+++ b/src/mongo/db/storage/kv/kv_collection_catalog_entry.h
@@ -39,11 +39,11 @@
 namespace mongo {
 
 class KVCatalog;
-class KVStorageEngineInterface;
+class StorageEngineInterface;
 
 class KVCollectionCatalogEntry final : public BSONCollectionCatalogEntry {
 public:
-    KVCollectionCatalogEntry(KVStorageEngineInterface* engine,
+    KVCollectionCatalogEntry(StorageEngineInterface* engine,
                              KVCatalog* catalog,
                              StringData ns,
                              StringData ident,
@@ -123,8 +123,8 @@ private:
     class AddIndexChange;
     class RemoveIndexChange;
 
-    KVStorageEngineInterface* const _engine;  // not owned
-    KVCatalog* _catalog;                      // not owned
+    StorageEngineInterface* const _engine;  // not owned
+    KVCatalog* _catalog;                    // not owned
     std::string _ident;
     std::unique_ptr<RecordStore> _recordStore;  // owned
 };
diff --git a/src/mongo/db/storage/kv/kv_collection_catalog_entry_test.cpp b/src/mongo/db/storage/kv/kv_collection_catalog_entry_test.cpp
index 34542493c2e..5c1aa53aef3 100644
--- a/src/mongo/db/storage/kv/kv_collection_catalog_entry_test.cpp
+++ b/src/mongo/db/storage/kv/kv_collection_catalog_entry_test.cpp
@@ -43,7 +43,7 @@
 #include "mongo/db/service_context_test_fixture.h"
 #include "mongo/db/storage/devnull/devnull_kv_engine.h"
 #include "mongo/db/storage/kv/kv_engine.h"
-#include "mongo/db/storage/kv/kv_storage_engine.h"
+#include "mongo/db/storage/kv/storage_engine_impl.h"
 #include "mongo/unittest/death_test.h"
 #include "mongo/unittest/unittest.h"
 #include "mongo/util/str.h"
@@ -61,7 +61,7 @@ class KVCollectionCatalogEntryTest : public ServiceContextTest {
 public:
     KVCollectionCatalogEntryTest()
         : _nss("unittests.kv_collection_catalog_entry"),
-          _storageEngine(new DevNullKVEngine(), KVStorageEngineOptions()) {
+          _storageEngine(new DevNullKVEngine(), StorageEngineOptions()) {
         _storageEngine.finishInit();
     }
 
@@ -153,7 +153,7 @@ private:
     }
 
     const NamespaceString _nss;
-    KVStorageEngine _storageEngine;
+    StorageEngineImpl _storageEngine;
     size_t numIndexesCreated = 0;
 };
 
diff --git a/src/mongo/db/storage/kv/kv_engine.h b/src/mongo/db/storage/kv/kv_engine.h
index e6801dab6c7..5a6d6c80791 100644
--- a/src/mongo/db/storage/kv/kv_engine.h
+++ b/src/mongo/db/storage/kv/kv_engine.h
@@ -107,7 +107,7 @@ public:
 
     /**
      * The create and drop methods on KVEngine are not transactional. Transactional semantics
-     * are provided by the KVStorageEngine code that calls these. For example, drop will be
+     * are provided by the StorageEngine code that calls these. For example, drop will be
      * called if a create is rolled back. A higher-level drop operation will only propagate to a
      * drop call on the KVEngine once the WUOW commits. Therefore drops will never be rolled
      * back and it is safe to immediately reclaim storage.
diff --git a/src/mongo/db/storage/kv/kv_storage_engine.cpp b/src/mongo/db/storage/kv/storage_engine_impl.cpp
index 328fa7cd1c5..3aa39b1c123 100644
--- a/src/mongo/db/storage/kv/kv_storage_engine.cpp
+++ b/src/mongo/db/storage/kv/storage_engine_impl.cpp
@@ -31,7 +31,7 @@
 #define LOG_FOR_RECOVERY(level) \
     MONGO_LOG_COMPONENT(level, ::mongo::logger::LogComponent::kStorageRecovery)
 
-#include "mongo/db/storage/kv/kv_storage_engine.h"
+#include "mongo/db/storage/kv/storage_engine_impl.h"
 
 #include <algorithm>
 
@@ -63,7 +63,7 @@ const std::string catalogInfo = "_mdb_catalog";
 const auto kCatalogLogLevel = logger::LogSeverity::Debug(2);
 }
 
-KVStorageEngine::KVStorageEngine(KVEngine* engine, KVStorageEngineOptions options)
+StorageEngineImpl::StorageEngineImpl(KVEngine* engine, StorageEngineOptions options)
     : _engine(engine),
       _options(std::move(options)),
       _dropPendingIdentReaper(engine),
@@ -81,7 +81,7 @@ KVStorageEngine::KVStorageEngine(KVEngine* engine, KVStorageEngineOptions option
     loadCatalog(&opCtx);
 }
 
-void KVStorageEngine::loadCatalog(OperationContext* opCtx) {
+void StorageEngineImpl::loadCatalog(OperationContext* opCtx) {
     bool catalogExists = _engine->hasIdent(opCtx, catalogInfo);
     if (_options.forRepair && catalogExists) {
         auto repairObserver = StorageRepairObserver::get(getGlobalServiceContext());
@@ -235,9 +235,9 @@ void KVStorageEngine::loadCatalog(OperationContext* opCtx) {
     startingAfterUncleanShutdown(getGlobalServiceContext()) = false;
 }
 
-void KVStorageEngine::_initCollection(OperationContext* opCtx,
-                                      const NamespaceString& nss,
-                                      bool forRepair) {
+void StorageEngineImpl::_initCollection(OperationContext* opCtx,
+                                        const NamespaceString& nss,
+                                        bool forRepair) {
     auto catalogEntry = _catalog->makeCollectionCatalogEntry(opCtx, nss, forRepair);
     auto uuid = catalogEntry->getCollectionOptions(opCtx).uuid.get();
 
@@ -248,7 +248,7 @@ void KVStorageEngine::_initCollection(OperationContext* opCtx,
     collectionCatalog.registerCollection(uuid, std::move(catalogEntry), std::move(collection));
 }
 
-void KVStorageEngine::closeCatalog(OperationContext* opCtx) {
+void StorageEngineImpl::closeCatalog(OperationContext* opCtx) {
     dassert(opCtx->lockState()->isLocked());
     if (shouldLog(::mongo::logger::LogComponent::kStorageRecovery, kCatalogLogLevel)) {
         LOG_FOR_RECOVERY(kCatalogLogLevel) << "loadCatalog:";
@@ -260,9 +260,9 @@ void KVStorageEngine::closeCatalog(OperationContext* opCtx) {
     _catalogRecordStore.reset();
 }
 
-Status KVStorageEngine::_recoverOrphanedCollection(OperationContext* opCtx,
-                                                   const NamespaceString& collectionName,
-                                                   StringData collectionIdent) {
+Status StorageEngineImpl::_recoverOrphanedCollection(OperationContext* opCtx,
+                                                     const NamespaceString& collectionName,
+                                                     StringData collectionIdent) {
     if (!_options.forRepair) {
         return {ErrorCodes::IllegalOperation, "Orphan recovery only supported in repair"};
     }
@@ -303,7 +303,7 @@ Status KVStorageEngine::_recoverOrphanedCollection(OperationContext* opCtx,
  * rebuild the index.
  */
 StatusWith<std::vector<StorageEngine::CollectionIndexNamePair>>
-KVStorageEngine::reconcileCatalogAndIdents(OperationContext* opCtx) {
+StorageEngineImpl::reconcileCatalogAndIdents(OperationContext* opCtx) {
     // Gather all tables known to the storage engine and drop those that aren't cross-referenced
     // in the _mdb_catalog. This can happen for two reasons.
     //
@@ -498,11 +498,11 @@ KVStorageEngine::reconcileCatalogAndIdents(OperationContext* opCtx) {
     return ret;
 }
 
-std::string KVStorageEngine::getFilesystemPathForDb(const std::string& dbName) const {
+std::string StorageEngineImpl::getFilesystemPathForDb(const std::string& dbName) const {
     return _catalog->getFilesystemPathForDb(dbName);
 }
 
-void KVStorageEngine::cleanShutdown() {
+void StorageEngineImpl::cleanShutdown() {
     if (_timestampMonitor) {
         _timestampMonitor->removeListener(&_minOfCheckpointAndOldestTimestampListener);
     }
@@ -519,9 +519,9 @@ void KVStorageEngine::cleanShutdown() {
     // intentionally not deleting _engine
 }
 
-KVStorageEngine::~KVStorageEngine() {}
+StorageEngineImpl::~StorageEngineImpl() {}
 
-void KVStorageEngine::finishInit() {
+void StorageEngineImpl::finishInit() {
     if (_engine->supportsRecoveryTimestamp()) {
         _timestampMonitor = std::make_unique<TimestampMonitor>(
             _engine.get(), getGlobalServiceContext()->getPeriodicRunner());
@@ -530,7 +530,7 @@ void KVStorageEngine::finishInit() {
     }
 }
 
-RecoveryUnit* KVStorageEngine::newRecoveryUnit() {
+RecoveryUnit* StorageEngineImpl::newRecoveryUnit() {
     if (!_engine) {
         // shutdown
         return nullptr;
@@ -538,16 +538,16 @@ RecoveryUnit* KVStorageEngine::newRecoveryUnit() {
     return _engine->newRecoveryUnit();
 }
 
-std::vector<std::string> KVStorageEngine::listDatabases() const {
+std::vector<std::string> StorageEngineImpl::listDatabases() const {
     return CollectionCatalog::get(getGlobalServiceContext()).getAllDbNames();
 }
 
-Status KVStorageEngine::closeDatabase(OperationContext* opCtx, StringData db) {
+Status StorageEngineImpl::closeDatabase(OperationContext* opCtx, StringData db) {
     // This is ok to be a no-op as there is no database layer in kv.
     return Status::OK();
 }
 
-Status KVStorageEngine::dropDatabase(OperationContext* opCtx, StringData db) {
+Status StorageEngineImpl::dropDatabase(OperationContext* opCtx, StringData db) {
     {
         auto dbs = CollectionCatalog::get(opCtx).getAllDbNames();
         if (std::count(dbs.begin(), dbs.end(), db.toString()) == 0) {
@@ -568,8 +568,8 @@ Status KVStorageEngine::dropDatabase(OperationContext* opCtx, StringData db) {
  * Returns the first `dropCollection` error that this method encounters. This method will attempt
  * to drop all collections, regardless of the error status.
  */
-Status KVStorageEngine::_dropCollectionsNoTimestamp(OperationContext* opCtx,
-                                                    std::vector<NamespaceString>& toDrop) {
+Status StorageEngineImpl::_dropCollectionsNoTimestamp(OperationContext* opCtx,
+                                                      std::vector<NamespaceString>& toDrop) {
     // On primaries, this method will be called outside of any `TimestampBlock` state meaning the
     // "commit timestamp" will not be set. For this case, this method needs no special logic to
     // avoid timestamping the upcoming writes.
@@ -611,11 +611,11 @@ Status KVStorageEngine::_dropCollectionsNoTimestamp(OperationContext* opCtx,
     return firstError;
 }
 
-int KVStorageEngine::flushAllFiles(OperationContext* opCtx, bool sync) {
+int StorageEngineImpl::flushAllFiles(OperationContext* opCtx, bool sync) {
     return _engine->flushAllFiles(opCtx, sync);
 }
 
-Status KVStorageEngine::beginBackup(OperationContext* opCtx) {
+Status StorageEngineImpl::beginBackup(OperationContext* opCtx) {
     // We should not proceed if we are already in backup mode
     if (_inBackupMode)
         return Status(ErrorCodes::BadValue, "Already in Backup Mode");
@@ -625,39 +625,40 @@ Status KVStorageEngine::beginBackup(OperationContext* opCtx) {
     return status;
 }
 
-void KVStorageEngine::endBackup(OperationContext* opCtx) {
+void StorageEngineImpl::endBackup(OperationContext* opCtx) {
     // We should never reach here if we aren't already in backup mode
     invariant(_inBackupMode);
     _engine->endBackup(opCtx);
     _inBackupMode = false;
 }
 
-StatusWith<std::vector<std::string>> KVStorageEngine::beginNonBlockingBackup(
+StatusWith<std::vector<std::string>> StorageEngineImpl::beginNonBlockingBackup(
     OperationContext* opCtx) {
     return _engine->beginNonBlockingBackup(opCtx);
 }
 
-void KVStorageEngine::endNonBlockingBackup(OperationContext* opCtx) {
+void StorageEngineImpl::endNonBlockingBackup(OperationContext* opCtx) {
     return _engine->endNonBlockingBackup(opCtx);
 }
 
-StatusWith<std::vector<std::string>> KVStorageEngine::extendBackupCursor(OperationContext* opCtx) {
+StatusWith<std::vector<std::string>> StorageEngineImpl::extendBackupCursor(
+    OperationContext* opCtx) {
     return _engine->extendBackupCursor(opCtx);
 }
 
-bool KVStorageEngine::isDurable() const {
+bool StorageEngineImpl::isDurable() const {
     return _engine->isDurable();
 }
 
-bool KVStorageEngine::isEphemeral() const {
+bool StorageEngineImpl::isEphemeral() const {
     return _engine->isEphemeral();
 }
 
-SnapshotManager* KVStorageEngine::getSnapshotManager() const {
+SnapshotManager* StorageEngineImpl::getSnapshotManager() const {
     return _engine->getSnapshotManager();
 }
 
-Status KVStorageEngine::repairRecordStore(OperationContext* opCtx, const NamespaceString& nss) {
+Status StorageEngineImpl::repairRecordStore(OperationContext* opCtx, const NamespaceString& nss) {
     auto repairObserver = StorageRepairObserver::get(getGlobalServiceContext());
     invariant(repairObserver->isIncomplete());
 
@@ -680,7 +681,7 @@ Status KVStorageEngine::repairRecordStore(OperationContext* opCtx, const Namespa
     return Status::OK();
 }
 
-std::unique_ptr<TemporaryRecordStore> KVStorageEngine::makeTemporaryRecordStore(
+std::unique_ptr<TemporaryRecordStore> StorageEngineImpl::makeTemporaryRecordStore(
     OperationContext* opCtx) {
     std::unique_ptr<RecordStore> rs =
         _engine->makeTemporaryRecordStore(opCtx, _catalog->newInternalIdent());
@@ -688,50 +689,50 @@ std::unique_ptr<TemporaryRecordStore> KVStorageEngine::makeTemporaryRecordStore(
     return std::make_unique<TemporaryKVRecordStore>(getEngine(), std::move(rs));
 }
 
-void KVStorageEngine::setJournalListener(JournalListener* jl) {
+void StorageEngineImpl::setJournalListener(JournalListener* jl) {
     _engine->setJournalListener(jl);
 }
 
-void KVStorageEngine::setStableTimestamp(Timestamp stableTimestamp, bool force) {
+void StorageEngineImpl::setStableTimestamp(Timestamp stableTimestamp, bool force) {
     _engine->setStableTimestamp(stableTimestamp, force);
 }
 
-void KVStorageEngine::setInitialDataTimestamp(Timestamp initialDataTimestamp) {
+void StorageEngineImpl::setInitialDataTimestamp(Timestamp initialDataTimestamp) {
     _initialDataTimestamp = initialDataTimestamp;
     _engine->setInitialDataTimestamp(initialDataTimestamp);
 }
 
-void KVStorageEngine::setOldestTimestampFromStable() {
+void StorageEngineImpl::setOldestTimestampFromStable() {
     _engine->setOldestTimestampFromStable();
 }
 
-void KVStorageEngine::setOldestTimestamp(Timestamp newOldestTimestamp) {
+void StorageEngineImpl::setOldestTimestamp(Timestamp newOldestTimestamp) {
     const bool force = true;
     _engine->setOldestTimestamp(newOldestTimestamp, force);
 }
 
-void KVStorageEngine::setOldestActiveTransactionTimestampCallback(
+void StorageEngineImpl::setOldestActiveTransactionTimestampCallback(
     StorageEngine::OldestActiveTransactionTimestampCallback callback) {
     _engine->setOldestActiveTransactionTimestampCallback(callback);
 }
 
-int64_t KVStorageEngine::getCacheOverflowTableInsertCount(OperationContext* opCtx) const {
+int64_t StorageEngineImpl::getCacheOverflowTableInsertCount(OperationContext* opCtx) const {
     return _engine->getCacheOverflowTableInsertCount(opCtx);
 }
 
-void KVStorageEngine::setCacheOverflowTableInsertCountForTest(int insertCount) {
+void StorageEngineImpl::setCacheOverflowTableInsertCountForTest(int insertCount) {
     return _engine->setCacheOverflowTableInsertCountForTest(insertCount);
 }
 
-bool KVStorageEngine::supportsRecoverToStableTimestamp() const {
+bool StorageEngineImpl::supportsRecoverToStableTimestamp() const {
     return _engine->supportsRecoverToStableTimestamp();
 }
 
-bool KVStorageEngine::supportsRecoveryTimestamp() const {
+bool StorageEngineImpl::supportsRecoveryTimestamp() const {
     return _engine->supportsRecoveryTimestamp();
 }
 
-StatusWith<Timestamp> KVStorageEngine::recoverToStableTimestamp(OperationContext* opCtx) {
+StatusWith<Timestamp> StorageEngineImpl::recoverToStableTimestamp(OperationContext* opCtx) {
     invariant(opCtx->lockState()->isW());
 
     // The "feature document" should not be rolled back. Perform a non-timestamped update to the
@@ -757,47 +758,47 @@ StatusWith<Timestamp> KVStorageEngine::recoverToStableTimestamp(OperationContext
     return {swTimestamp.getValue()};
 }
 
-boost::optional<Timestamp> KVStorageEngine::getRecoveryTimestamp() const {
+boost::optional<Timestamp> StorageEngineImpl::getRecoveryTimestamp() const {
     return _engine->getRecoveryTimestamp();
 }
 
-boost::optional<Timestamp> KVStorageEngine::getLastStableRecoveryTimestamp() const {
+boost::optional<Timestamp> StorageEngineImpl::getLastStableRecoveryTimestamp() const {
     return _engine->getLastStableRecoveryTimestamp();
 }
 
-bool KVStorageEngine::supportsReadConcernSnapshot() const {
+bool StorageEngineImpl::supportsReadConcernSnapshot() const {
     return _engine->supportsReadConcernSnapshot();
 }
 
-bool KVStorageEngine::supportsReadConcernMajority() const {
+bool StorageEngineImpl::supportsReadConcernMajority() const {
     return _engine->supportsReadConcernMajority();
 }
 
-bool KVStorageEngine::supportsPendingDrops() const {
+bool StorageEngineImpl::supportsPendingDrops() const {
     return supportsReadConcernMajority();
 }
 
-void KVStorageEngine::clearDropPendingState() {
+void StorageEngineImpl::clearDropPendingState() {
     _dropPendingIdentReaper.clearDropPendingState();
 }
 
-void KVStorageEngine::replicationBatchIsComplete() const {
+void StorageEngineImpl::replicationBatchIsComplete() const {
     return _engine->replicationBatchIsComplete();
 }
 
-Timestamp KVStorageEngine::getAllCommittedTimestamp() const {
+Timestamp StorageEngineImpl::getAllCommittedTimestamp() const {
     return _engine->getAllCommittedTimestamp();
 }
 
-Timestamp KVStorageEngine::getOldestOpenReadTimestamp() const {
+Timestamp StorageEngineImpl::getOldestOpenReadTimestamp() const {
     return _engine->getOldestOpenReadTimestamp();
 }
 
-boost::optional<Timestamp> KVStorageEngine::getOplogNeededForCrashRecovery() const {
+boost::optional<Timestamp> StorageEngineImpl::getOplogNeededForCrashRecovery() const {
     return _engine->getOplogNeededForCrashRecovery();
 }
 
-void KVStorageEngine::_dumpCatalog(OperationContext* opCtx) {
+void StorageEngineImpl::_dumpCatalog(OperationContext* opCtx) {
     auto catalogRs = _catalogRecordStore.get();
     auto cursor = catalogRs->getCursor(opCtx);
     boost::optional<Record> rec = cursor->next();
@@ -811,13 +812,13 @@ void KVStorageEngine::_dumpCatalog(OperationContext* opCtx) {
     opCtx->recoveryUnit()->abandonSnapshot();
 }
 
-void KVStorageEngine::addDropPendingIdent(const Timestamp& dropTimestamp,
-                                          const NamespaceString& nss,
-                                          StringData ident) {
+void StorageEngineImpl::addDropPendingIdent(const Timestamp& dropTimestamp,
+                                            const NamespaceString& nss,
+                                            StringData ident) {
     _dropPendingIdentReaper.addDropPendingIdent(dropTimestamp, nss, ident);
 }
 
-void KVStorageEngine::_onMinOfCheckpointAndOldestTimestampChanged(const Timestamp& timestamp) {
+void StorageEngineImpl::_onMinOfCheckpointAndOldestTimestampChanged(const Timestamp& timestamp) {
     if (timestamp.isNull()) {
         return;
     }
@@ -838,7 +839,7 @@ void KVStorageEngine::_onMinOfCheckpointAndOldestTimestampChanged(const Timestam
     }
 }
 
-KVStorageEngine::TimestampMonitor::TimestampMonitor(KVEngine* engine, PeriodicRunner* runner)
+StorageEngineImpl::TimestampMonitor::TimestampMonitor(KVEngine* engine, PeriodicRunner* runner)
     : _engine(engine), _running(false), _periodicRunner(runner) {
     _currentTimestamps.checkpoint = _engine->getCheckpointTimestamp();
     _currentTimestamps.oldest = _engine->getOldestTimestamp();
@@ -850,13 +851,13 @@ KVStorageEngine::TimestampMonitor::TimestampMonitor(KVEngine* engine, PeriodicRu
         : _currentTimestamps.checkpoint;
 }
 
-KVStorageEngine::TimestampMonitor::~TimestampMonitor() {
+StorageEngineImpl::TimestampMonitor::~TimestampMonitor() {
     log() << "Timestamp monitor shutting down";
     stdx::lock_guard<stdx::mutex> lock(_monitorMutex);
     invariant(_listeners.empty());
 }
 
-void KVStorageEngine::TimestampMonitor::startup() {
+void StorageEngineImpl::TimestampMonitor::startup() {
     invariant(!_running);
 
     log() << "Timestamp monitor starting";
@@ -925,7 +926,7 @@ void KVStorageEngine::TimestampMonitor::startup() {
     _running = true;
 }
 
-void KVStorageEngine::TimestampMonitor::notifyAll(TimestampType type, Timestamp newTimestamp) {
+void StorageEngineImpl::TimestampMonitor::notifyAll(TimestampType type, Timestamp newTimestamp) {
     stdx::lock_guard<stdx::mutex> lock(_monitorMutex);
     for (auto& listener : _listeners) {
         if (listener->getType() == type) {
@@ -934,7 +935,7 @@ void KVStorageEngine::TimestampMonitor::notifyAll(TimestampType type, Timestamp
     }
 }
 
-void KVStorageEngine::TimestampMonitor::addListener(TimestampListener* listener) {
+void StorageEngineImpl::TimestampMonitor::addListener(TimestampListener* listener) {
     stdx::lock_guard<stdx::mutex> lock(_monitorMutex);
     if (std::find(_listeners.begin(), _listeners.end(), listener) != _listeners.end()) {
         bool listenerAlreadyRegistered = true;
@@ -943,7 +944,7 @@ void KVStorageEngine::TimestampMonitor::addListener(TimestampListener* listener)
     _listeners.push_back(listener);
 }
 
-void KVStorageEngine::TimestampMonitor::removeListener(TimestampListener* listener) {
+void StorageEngineImpl::TimestampMonitor::removeListener(TimestampListener* listener) {
     stdx::lock_guard<stdx::mutex> lock(_monitorMutex);
     if (std::find(_listeners.begin(), _listeners.end(), listener) == _listeners.end()) {
         bool listenerNotRegistered = true;
@@ -952,7 +953,7 @@ void KVStorageEngine::TimestampMonitor::removeListener(TimestampListener* listen
     _listeners.erase(std::remove(_listeners.begin(), _listeners.end(), listener));
 }
 
-int64_t KVStorageEngine::sizeOnDiskForDb(OperationContext* opCtx, StringData dbName) {
+int64_t StorageEngineImpl::sizeOnDiskForDb(OperationContext* opCtx, StringData dbName) {
     int64_t size = 0;
 
     catalog::forEachCollectionFromDb(
diff --git a/src/mongo/db/storage/kv/kv_storage_engine.h b/src/mongo/db/storage/kv/storage_engine_impl.h
index e75f5271f8b..4d060cdcd4b 100644
--- a/src/mongo/db/storage/kv/kv_storage_engine.h
+++ b/src/mongo/db/storage/kv/storage_engine_impl.h
@@ -42,7 +42,7 @@
 #include "mongo/db/storage/kv/kv_catalog.h"
 #include "mongo/db/storage/kv/kv_catalog_feature_tracker.h"
 #include "mongo/db/storage/kv/kv_drop_pending_ident_reaper.h"
-#include "mongo/db/storage/kv/kv_storage_engine_interface.h"
+#include "mongo/db/storage/kv/storage_engine_interface.h"
 #include "mongo/db/storage/record_store.h"
 #include "mongo/db/storage/storage_engine.h"
 #include "mongo/db/storage/temporary_record_store.h"
@@ -54,20 +54,20 @@ namespace mongo {
 class KVCatalog;
 class KVEngine;
 
-struct KVStorageEngineOptions {
+struct StorageEngineOptions {
     bool directoryPerDB = false;
     bool directoryForIndexes = false;
     bool forRepair = false;
 };
 
-class KVStorageEngine final : public KVStorageEngineInterface, public StorageEngine {
+class StorageEngineImpl final : public StorageEngineInterface, public StorageEngine {
 public:
     /**
      * @param engine - ownership passes to me
      */
-    KVStorageEngine(KVEngine* engine, KVStorageEngineOptions options = KVStorageEngineOptions());
+    StorageEngineImpl(KVEngine* engine, StorageEngineOptions options = StorageEngineOptions());
 
-    virtual ~KVStorageEngine();
+    virtual ~StorageEngineImpl();
 
     virtual void finishInit();
 
@@ -380,7 +380,7 @@ private:
     // This must be the first member so it is destroyed last.
     std::unique_ptr<KVEngine> _engine;
 
-    const KVStorageEngineOptions _options;
+    const StorageEngineOptions _options;
 
     // Manages drop-pending idents. Requires access to '_engine'.
     KVDropPendingIdentReaper _dropPendingIdentReaper;
diff --git a/src/mongo/db/storage/kv/kv_storage_engine_interface.h b/src/mongo/db/storage/kv/storage_engine_interface.h
index 09dae7d2a7f..89ab71b83b2 100644
--- a/src/mongo/db/storage/kv/kv_storage_engine_interface.h
+++ b/src/mongo/db/storage/kv/storage_engine_interface.h
@@ -35,10 +35,10 @@ class KVEngine;
 class KVCatalog;
 class StorageEngine;
 
-class KVStorageEngineInterface {
+class StorageEngineInterface {
 public:
-    KVStorageEngineInterface() = default;
-    virtual ~KVStorageEngineInterface() = default;
+    StorageEngineInterface() = default;
+    virtual ~StorageEngineInterface() = default;
     virtual StorageEngine* getStorageEngine() = 0;
     virtual KVEngine* getEngine() = 0;
     virtual void addDropPendingIdent(const Timestamp& dropTimestamp,
diff --git a/src/mongo/db/storage/kv/kv_storage_engine_test.cpp b/src/mongo/db/storage/kv/storage_engine_test.cpp
index 554793f9f79..d424f64f3c1 100644
--- a/src/mongo/db/storage/kv/kv_storage_engine_test.cpp
+++ b/src/mongo/db/storage/kv/storage_engine_test.cpp
@@ -46,8 +46,8 @@
 #include "mongo/db/storage/ephemeral_for_test/ephemeral_for_test_engine.h"
 #include "mongo/db/storage/kv/kv_catalog.h"
 #include "mongo/db/storage/kv/kv_engine.h"
-#include "mongo/db/storage/kv/kv_storage_engine.h"
-#include "mongo/db/storage/kv/kv_storage_engine_test_fixture.h"
+#include "mongo/db/storage/kv/storage_engine_impl.h"
+#include "mongo/db/storage/kv/storage_engine_test_fixture.h"
 #include "mongo/db/storage/storage_repair_observer.h"
 #include "mongo/db/unclean_shutdown.h"
 #include "mongo/unittest/barrier.h"
@@ -57,7 +57,7 @@
 namespace mongo {
 namespace {
 
-TEST_F(KVStorageEngineTest, ReconcileIdentsTest) {
+TEST_F(StorageEngineTest, ReconcileIdentsTest) {
     auto opCtx = cc().makeOperationContext();
 
     // Add a collection, `db.coll1` to both the KVCatalog and KVEngine. The returned value is the
@@ -97,7 +97,7 @@ TEST_F(KVStorageEngineTest, ReconcileIdentsTest) {
     ASSERT_EQUALS(ErrorCodes::UnrecoverableRollbackError, reconcileStatus.getStatus());
 }
 
-TEST_F(KVStorageEngineTest, LoadCatalogDropsOrphansAfterUncleanShutdown) {
+TEST_F(StorageEngineTest, LoadCatalogDropsOrphansAfterUncleanShutdown) {
     auto opCtx = cc().makeOperationContext();
 
     const NamespaceString collNs("db.coll1");
@@ -120,7 +120,7 @@ TEST_F(KVStorageEngineTest, LoadCatalogDropsOrphansAfterUncleanShutdown) {
     ASSERT(!collectionExists(opCtx.get(), collNs));
 }
 
-TEST_F(KVStorageEngineTest, ReconcileDropsTemporary) {
+TEST_F(StorageEngineTest, ReconcileDropsTemporary) {
     auto opCtx = cc().makeOperationContext();
 
     Lock::GlobalLock lk(&*opCtx, MODE_IS);
@@ -139,7 +139,7 @@ TEST_F(KVStorageEngineTest, ReconcileDropsTemporary) {
     rs->deleteTemporaryTable(opCtx.get());
 }
 
-TEST_F(KVStorageEngineTest, TemporaryDropsItself) {
+TEST_F(StorageEngineTest, TemporaryDropsItself) {
     auto opCtx = cc().makeOperationContext();
 
     Lock::GlobalLock lk(&*opCtx, MODE_IS);
@@ -159,7 +159,7 @@ TEST_F(KVStorageEngineTest, TemporaryDropsItself) {
     ASSERT(!identExists(opCtx.get(), ident));
 }
 
-TEST_F(KVStorageEngineTest, ReconcileDoesNotDropIndexBuildTempTables) {
+TEST_F(StorageEngineTest, ReconcileDoesNotDropIndexBuildTempTables) {
     auto opCtx = cc().makeOperationContext();
 
     Lock::GlobalLock lk(&*opCtx, MODE_IS);
@@ -200,7 +200,7 @@ TEST_F(KVStorageEngineTest, ReconcileDoesNotDropIndexBuildTempTables) {
     constraintViolations->deleteTemporaryTable(opCtx.get());
 }
 
-TEST_F(KVStorageEngineTest, ReconcileDoesNotDropIndexBuildTempTablesBackgroundSecondary) {
+TEST_F(StorageEngineTest, ReconcileDoesNotDropIndexBuildTempTablesBackgroundSecondary) {
     auto opCtx = cc().makeOperationContext();
 
     Lock::GlobalLock lk(&*opCtx, MODE_IS);
@@ -246,7 +246,7 @@ TEST_F(KVStorageEngineTest, ReconcileDoesNotDropIndexBuildTempTablesBackgroundSe
     constraintViolations->deleteTemporaryTable(opCtx.get());
 }
 
-TEST_F(KVStorageEngineRepairTest, LoadCatalogRecoversOrphans) {
+TEST_F(StorageEngineRepairTest, LoadCatalogRecoversOrphans) {
     auto opCtx = cc().makeOperationContext();
 
     const NamespaceString collNs("db.coll1");
@@ -270,7 +270,7 @@ TEST_F(KVStorageEngineRepairTest, LoadCatalogRecoversOrphans) {
     ASSERT_EQ(1U, StorageRepairObserver::get(getGlobalServiceContext())->getModifications().size());
 }
 
-TEST_F(KVStorageEngineRepairTest, ReconcileSucceeds) {
+TEST_F(StorageEngineRepairTest, ReconcileSucceeds) {
     auto opCtx = cc().makeOperationContext();
 
     const NamespaceString collNs("db.coll1");
@@ -290,7 +290,7 @@ TEST_F(KVStorageEngineRepairTest, ReconcileSucceeds) {
     ASSERT_EQ(0U, StorageRepairObserver::get(getGlobalServiceContext())->getModifications().size());
 }
 
-TEST_F(KVStorageEngineRepairTest, LoadCatalogRecoversOrphansInCatalog) {
+TEST_F(StorageEngineRepairTest, LoadCatalogRecoversOrphansInCatalog) {
     auto opCtx = cc().makeOperationContext();
 
     const NamespaceString collNs("db.coll1");
@@ -318,7 +318,7 @@ TEST_F(KVStorageEngineRepairTest, LoadCatalogRecoversOrphansInCatalog) {
     ASSERT_EQ(1U, StorageRepairObserver::get(getGlobalServiceContext())->getModifications().size());
 }
 
-TEST_F(KVStorageEngineTest, LoadCatalogDropsOrphans) {
+TEST_F(StorageEngineTest, LoadCatalogDropsOrphans) {
     auto opCtx = cc().makeOperationContext();
 
     const NamespaceString collNs("db.coll1");
@@ -377,16 +377,16 @@ public:
 
 class TimestampKVEngineTest : public ServiceContextMongoDTest {
 public:
-    using TimestampType = KVStorageEngine::TimestampMonitor::TimestampType;
-    using TimestampListener = KVStorageEngine::TimestampMonitor::TimestampListener;
+    using TimestampType = StorageEngineImpl::TimestampMonitor::TimestampType;
+    using TimestampListener = StorageEngineImpl::TimestampMonitor::TimestampListener;
 
     /**
      * Create an instance of the KV Storage Engine so that we have a timestamp monitor operating.
      */
     TimestampKVEngineTest() {
-        KVStorageEngineOptions options{
+        StorageEngineOptions options{
             /*directoryPerDB=*/false, /*directoryForIndexes=*/false, /*forRepair=*/false};
-        _storageEngine = std::make_unique<KVStorageEngine>(new TimestampMockKVEngine, options);
+        _storageEngine = std::make_unique<StorageEngineImpl>(new TimestampMockKVEngine, options);
         _storageEngine->finishInit();
     }
 
@@ -399,7 +399,7 @@ public:
         _storageEngine.reset();
     }
 
-    std::unique_ptr<KVStorageEngine> _storageEngine;
+    std::unique_ptr<StorageEngineImpl> _storageEngine;
 
     TimestampType checkpoint = TimestampType::kCheckpoint;
     TimestampType oldest = TimestampType::kOldest;
diff --git a/src/mongo/db/storage/kv/kv_storage_engine_test_fixture.h b/src/mongo/db/storage/kv/storage_engine_test_fixture.h
index c9cfbafe911..7f3bddb4bc8 100644
--- a/src/mongo/db/storage/kv/kv_storage_engine_test_fixture.h
+++ b/src/mongo/db/storage/kv/storage_engine_test_fixture.h
@@ -35,18 +35,18 @@
 #include "mongo/db/service_context_d_test_fixture.h"
 #include "mongo/db/storage/kv/kv_catalog.h"
 #include "mongo/db/storage/kv/kv_engine.h"
-#include "mongo/db/storage/kv/kv_storage_engine.h"
+#include "mongo/db/storage/kv/storage_engine_impl.h"
 #include "mongo/db/storage/storage_repair_observer.h"
 
 namespace mongo {
 
-class KVStorageEngineTest : public ServiceContextMongoDTest {
+class StorageEngineTest : public ServiceContextMongoDTest {
 public:
-    KVStorageEngineTest(RepairAction repair)
+    StorageEngineTest(RepairAction repair)
         : ServiceContextMongoDTest("ephemeralForTest", repair),
-          _storageEngine(checked_cast<KVStorageEngine*>(getServiceContext()->getStorageEngine())) {}
+          _storageEngine(getServiceContext()->getStorageEngine()) {}
 
-    KVStorageEngineTest() : KVStorageEngineTest(RepairAction::kNoRepair) {}
+    StorageEngineTest() : StorageEngineTest(RepairAction::kNoRepair) {}
 
     /**
      * Create a collection in the catalog and in the KVEngine. Return the storage engine's `ident`.
@@ -168,12 +168,12 @@ public:
         return catalog->_removeEntry(opCtx, NamespaceString(ns));
     }
 
-    KVStorageEngine* _storageEngine;
+    StorageEngine* _storageEngine;
 };
 
-class KVStorageEngineRepairTest : public KVStorageEngineTest {
+class StorageEngineRepairTest : public StorageEngineTest {
 public:
-    KVStorageEngineRepairTest() : KVStorageEngineTest(RepairAction::kRepair) {}
+    StorageEngineRepairTest() : StorageEngineTest(RepairAction::kRepair) {}
 
     void tearDown() {
         auto repairObserver = StorageRepairObserver::get(getGlobalServiceContext());
diff --git a/src/mongo/db/storage/mobile/SConscript b/src/mongo/db/storage/mobile/SConscript
index 03e951da946..8c7d1012d99 100644
--- a/src/mongo/db/storage/mobile/SConscript
+++ b/src/mongo/db/storage/mobile/SConscript
@@ -69,7 +69,7 @@ env.Library(
     LIBDEPS=[
         'storage_mobile_core',
         '$BUILD_DIR/mongo/db/storage/kv/kv_engine_core',
-        '$BUILD_DIR/mongo/db/storage/kv/kv_storage_engine',
+        '$BUILD_DIR/mongo/db/storage/kv/storage_engine',
         ],
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/storage/storage_engine_common',
diff --git a/src/mongo/db/storage/mobile/mobile_init.cpp b/src/mongo/db/storage/mobile/mobile_init.cpp
index e504d40cb7e..dd333cc5667 100644
--- a/src/mongo/db/storage/mobile/mobile_init.cpp
+++ b/src/mongo/db/storage/mobile/mobile_init.cpp
@@ -33,7 +33,7 @@
 
 #include "mongo/base/init.h"
 #include "mongo/db/service_context.h"
-#include "mongo/db/storage/kv/kv_storage_engine.h"
+#include "mongo/db/storage/kv/storage_engine_impl.h"
 #include "mongo/db/storage/mobile/mobile_kv_engine.h"
 #include "mongo/db/storage/mobile/mobile_options.h"
 #include "mongo/db/storage/storage_engine_init.h"
@@ -50,14 +50,14 @@ public:
                 "mobile does not support --groupCollections",
                 !params.groupCollections);
 
-        KVStorageEngineOptions options;
+        StorageEngineOptions options;
         options.directoryPerDB = params.directoryperdb;
         options.forRepair = params.repair;
 
         MobileKVEngine* kvEngine = new MobileKVEngine(
             params.dbpath, embedded::mobileGlobalOptions, getGlobalServiceContext());
 
-        return new KVStorageEngine(kvEngine, options);
+        return new StorageEngineImpl(kvEngine, options);
     }
 
     StringData getCanonicalName() const override {
diff --git a/src/mongo/db/storage/storage_engine.h b/src/mongo/db/storage/storage_engine.h
index ad3abc4f1f0..38b3d19ebd9 100644
--- a/src/mongo/db/storage/storage_engine.h
+++ b/src/mongo/db/storage/storage_engine.h
@@ -42,17 +42,20 @@
 namespace mongo {
 
 class JournalListener;
+class KVCatalog;
+class KVEngine;
 class OperationContext;
 class RecoveryUnit;
 class SnapshotManager;
-struct StorageGlobalParams;
 class StorageEngineLockFile;
 class StorageEngineMetadata;
 
+struct StorageGlobalParams;
+
 /**
- * The StorageEngine class is the top level interface for creating a new storage
- * engine.  All StorageEngine(s) must be registered by calling registerFactory in order
- * to possibly be activated.
+ * The StorageEngine class is the top level interface for creating a new storage engine. All
+ * StorageEngine(s) must be registered by calling registerFactory in order to possibly be
+ * activated.
  */
 class StorageEngine {
 public:
@@ -528,6 +531,11 @@ public:
     virtual Status currentFilesCompatible(OperationContext* opCtx) const = 0;
 
     virtual int64_t sizeOnDiskForDb(OperationContext* opCtx, StringData dbName) = 0;
+
+    virtual KVEngine* getEngine() = 0;
+    virtual const KVEngine* getEngine() const = 0;
+    virtual KVCatalog* getCatalog() = 0;
+    virtual const KVCatalog* getCatalog() const = 0;
 };
 
 }  // namespace mongo
diff --git a/src/mongo/db/storage/wiredtiger/SConscript b/src/mongo/db/storage/wiredtiger/SConscript
index e0aa028a938..d1ec377be9c 100644
--- a/src/mongo/db/storage/wiredtiger/SConscript
+++ b/src/mongo/db/storage/wiredtiger/SConscript
@@ -102,7 +102,7 @@ if wiredtiger:
             'storage_wiredtiger_core',
             'storage_wiredtiger_customization_hooks',
             '$BUILD_DIR/mongo/db/db_raii',
-            '$BUILD_DIR/mongo/db/storage/kv/kv_storage_engine',
+            '$BUILD_DIR/mongo/db/storage/kv/storage_engine',
             '$BUILD_DIR/mongo/db/storage/storage_engine_lock_file',
             '$BUILD_DIR/mongo/db/storage/storage_engine_metadata',
         ],
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_init.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_init.cpp
index 7ea75f3a47a..8aa6fef42c0 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_init.cpp
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_init.cpp
@@ -39,7 +39,7 @@
 #include "mongo/db/catalog/collection_options.h"
 #include "mongo/db/jsobj.h"
 #include "mongo/db/service_context.h"
-#include "mongo/db/storage/kv/kv_storage_engine.h"
+#include "mongo/db/storage/kv/storage_engine_impl.h"
 #include "mongo/db/storage/storage_engine_init.h"
 #include "mongo/db/storage/storage_engine_lock_file.h"
 #include "mongo/db/storage/storage_engine_metadata.h"
@@ -126,11 +126,11 @@ public:
             "wiredTigerMaxCacheOverflowSizeGB", ServerParameterType::kRuntimeOnly);
         maxCacheOverflowParam->_data = {wiredTigerGlobalOptions.maxCacheOverflowFileSizeGB, kv};
 
-        KVStorageEngineOptions options;
+        StorageEngineOptions options;
         options.directoryPerDB = params.directoryperdb;
         options.directoryForIndexes = wiredTigerGlobalOptions.directoryForIndexes;
         options.forRepair = params.repair;
-        return new KVStorageEngine(kv, options);
+        return new StorageEngineImpl(kv, options);
     }
 
     virtual StringData getCanonicalName() const {
diff --git a/src/mongo/dbtests/SConscript b/src/mongo/dbtests/SConscript
index 1b887519b33..c7ccaa7decd 100644
--- a/src/mongo/dbtests/SConscript
+++ b/src/mongo/dbtests/SConscript
@@ -156,7 +156,7 @@ if not has_option('noshell') and usemozjs:
             "$BUILD_DIR/mongo/db/storage/biggie/storage_biggie",
             "$BUILD_DIR/mongo/db/storage/ephemeral_for_test/storage_ephemeral_for_test",
             "$BUILD_DIR/mongo/db/storage/kv/kv_engine_core",
-            "$BUILD_DIR/mongo/db/storage/kv/kv_storage_engine",
+            "$BUILD_DIR/mongo/db/storage/kv/storage_engine",
             "$BUILD_DIR/mongo/db/storage/wiredtiger/storage_wiredtiger" if wiredtiger else [],
             "$BUILD_DIR/mongo/transport/transport_layer_manager",
             "$BUILD_DIR/mongo/transport/message_compressor_options_server",
diff --git a/src/mongo/dbtests/storage_timestamp_tests.cpp b/src/mongo/dbtests/storage_timestamp_tests.cpp
index 6208fe8944d..0a71bdea88a 100644
--- a/src/mongo/dbtests/storage_timestamp_tests.cpp
+++ b/src/mongo/dbtests/storage_timestamp_tests.cpp
@@ -74,7 +74,7 @@
 #include "mongo/db/service_context.h"
 #include "mongo/db/session.h"
 #include "mongo/db/session_catalog_mongod.h"
-#include "mongo/db/storage/kv/kv_storage_engine.h"
+#include "mongo/db/storage/kv/storage_engine_impl.h"
 #include "mongo/db/storage/snapshot_manager.h"
 #include "mongo/db/transaction_participant.h"
 #include "mongo/db/transaction_participant_gen.h"
@@ -460,9 +460,7 @@ public:
      */
     void assertNamespaceInIdents(NamespaceString nss, Timestamp ts, bool shouldExpect) {
         OneOffRead oor(_opCtx, ts);
-        KVCatalog* kvCatalog =
-            static_cast<KVStorageEngine*>(_opCtx->getServiceContext()->getStorageEngine())
-                ->getCatalog();
+        auto kvCatalog = _opCtx->getServiceContext()->getStorageEngine()->getCatalog();
 
         AutoGetCollection autoColl(_opCtx, nss, LockMode::MODE_IS);
 
@@ -1736,7 +1734,7 @@ public:
 /**
  * This KVDropDatabase test only exists in this file for historical reasons, the final phase of
  * timestamping `dropDatabase` side-effects no longer applies. The purpose of this test is to
- * exercise the `KVStorageEngine::dropDatabase` method.
+ * exercise the `StorageEngine::dropDatabase` method.
  */
 template <bool SimulatePrimary>
 class KVDropDatabase : public StorageTimestampTest {
@@ -1747,14 +1745,13 @@ public:
             _opCtx->getServiceContext(),
             std::make_unique<repl::DropPendingCollectionReaper>(storageInterface));
 
-        auto kvStorageEngine =
-            dynamic_cast<KVStorageEngine*>(_opCtx->getServiceContext()->getStorageEngine());
-        KVCatalog* kvCatalog = kvStorageEngine->getCatalog();
+        auto storageEngine = _opCtx->getServiceContext()->getStorageEngine();
+        auto kvCatalog = storageEngine->getCatalog();
 
         // Declare the database to be in a "synced" state, i.e: in steady-state replication.
         Timestamp syncTime = _clock->reserveTicks(1).asTimestamp();
         invariant(!syncTime.isNull());
-        kvStorageEngine->setInitialDataTimestamp(syncTime);
+        storageEngine->setInitialDataTimestamp(syncTime);
 
         // This test drops collections piece-wise instead of having the "drop database" algorithm
         // perform this walk. Defensively operate on a separate DB from the other tests to ensure
@@ -1808,7 +1805,7 @@ public:
 
         // If the storage engine is managing drops internally, the ident should not be visible after
         // a drop.
-        if (kvStorageEngine->supportsPendingDrops()) {
+        if (storageEngine->supportsPendingDrops()) {
             assertIdentsMissingAtTimestamp(kvCatalog, collIdent, indexIdent, postRenameTime);
         } else {
             // The namespace has changed, but the ident still exists as-is after the rename.
@@ -1860,9 +1857,8 @@ public:
             ASSERT_OK(_coordinatorMock->setFollowerMode({repl::MemberState::MS::RS_SECONDARY}));
         }
 
-        auto kvStorageEngine =
-            dynamic_cast<KVStorageEngine*>(_opCtx->getServiceContext()->getStorageEngine());
-        KVCatalog* kvCatalog = kvStorageEngine->getCatalog();
+        auto storageEngine = _opCtx->getServiceContext()->getStorageEngine();
+        auto kvCatalog = storageEngine->getCatalog();
 
         NamespaceString nss("unittests.timestampIndexBuilds");
         reset(nss);
@@ -2129,9 +2125,8 @@ public:
 class TimestampMultiIndexBuilds : public StorageTimestampTest {
 public:
     void run() {
-        auto kvStorageEngine =
-            dynamic_cast<KVStorageEngine*>(_opCtx->getServiceContext()->getStorageEngine());
-        KVCatalog* kvCatalog = kvStorageEngine->getCatalog();
+        auto storageEngine = _opCtx->getServiceContext()->getStorageEngine();
+        auto kvCatalog = storageEngine->getCatalog();
 
         NamespaceString nss("unittests.timestampMultiIndexBuilds");
         reset(nss);
@@ -2214,9 +2209,8 @@ public:
 class TimestampMultiIndexBuildsDuringRename : public StorageTimestampTest {
 public:
     void run() {
-        auto kvStorageEngine =
-            dynamic_cast<KVStorageEngine*>(_opCtx->getServiceContext()->getStorageEngine());
-        KVCatalog* kvCatalog = kvStorageEngine->getCatalog();
+        auto storageEngine = _opCtx->getServiceContext()->getStorageEngine();
+        auto kvCatalog = storageEngine->getCatalog();
 
         NamespaceString nss("unittests.timestampMultiIndexBuildsDuringRename");
         reset(nss);
@@ -2323,9 +2317,8 @@ public:
 class TimestampIndexDrops : public StorageTimestampTest {
 public:
     void run() {
-        auto kvStorageEngine =
-            dynamic_cast<KVStorageEngine*>(_opCtx->getServiceContext()->getStorageEngine());
-        KVCatalog* kvCatalog = kvStorageEngine->getCatalog();
+        auto storageEngine = _opCtx->getServiceContext()->getStorageEngine();
+        auto kvCatalog = storageEngine->getCatalog();
 
         NamespaceString nss("unittests.timestampIndexDrops");
         reset(nss);
@@ -2541,9 +2534,8 @@ public:
             const auto startBuildTs = beforeBuildTime.addTicks(1).asTimestamp();
 
             // Grab the existing idents to identify the ident created by the index build.
-            auto kvStorageEngine =
-                dynamic_cast<KVStorageEngine*>(_opCtx->getServiceContext()->getStorageEngine());
-            KVCatalog* kvCatalog = kvStorageEngine->getCatalog();
+            auto storageEngine = _opCtx->getServiceContext()->getStorageEngine();
+            auto kvCatalog = storageEngine->getCatalog();
             std::vector<std::string> origIdents;
             {
                 AutoGetCollection autoColl(_opCtx, nss, LockMode::MODE_IS);
@@ -2586,9 +2578,8 @@ public:
 class ViewCreationSeparateTransaction : public StorageTimestampTest {
 public:
     void run() {
-        auto kvStorageEngine =
-            dynamic_cast<KVStorageEngine*>(_opCtx->getServiceContext()->getStorageEngine());
-        KVCatalog* kvCatalog = kvStorageEngine->getCatalog();
+        auto storageEngine = _opCtx->getServiceContext()->getStorageEngine();
+        auto kvCatalog = storageEngine->getCatalog();
 
         const NamespaceString backingCollNss("unittests.backingColl");
         reset(backingCollNss);
@@ -2687,9 +2678,8 @@ public:
         ASSERT_EQ(indexOp.getObject()["name"].str(), "user_1_db_1");
         ASSERT_GT(indexOp.getTimestamp(), futureTs) << op.toBSON();
         AutoGetCollection autoColl(_opCtx, nss, LockMode::MODE_IS);
-        auto kvStorageEngine =
-            dynamic_cast<KVStorageEngine*>(_opCtx->getServiceContext()->getStorageEngine());
-        KVCatalog* kvCatalog = kvStorageEngine->getCatalog();
+        auto storageEngine = _opCtx->getServiceContext()->getStorageEngine();
+        auto kvCatalog = storageEngine->getCatalog();
         auto indexIdent = kvCatalog->getIndexIdent(_opCtx, nss, "user_1_db_1");
         assertIdentsMissingAtTimestamp(kvCatalog, "", indexIdent, pastTs);
         assertIdentsMissingAtTimestamp(kvCatalog, "", indexIdent, presentTs);