SERVER-41819 Move methods from KVCollectionCatalogEntry to KVCatalog

(cherry picked from commit 77ccfbefcd740051597d1cb4eb36d25776926f72)

14 files changed, 731 insertions, 592 deletions

diff --git a/src/mongo/db/storage/kv/SConscript b/src/mongo/db/storage/kv/SConscript
index 69bba58c8fc..799904611c0 100644
--- a/src/mongo/db/storage/kv/SConscript
+++ b/src/mongo/db/storage/kv/SConscript
@@ -18,7 +18,7 @@ env.Library(
     source=[
         'kv_catalog.cpp',
         'kv_collection_catalog_entry.cpp',
-        ],
+    ],
     LIBDEPS=[
         '$BUILD_DIR/mongo/bson/util/bson_extract',
         '$BUILD_DIR/mongo/db/concurrency/lock_manager',
@@ -29,8 +29,11 @@ env.Library(
         '$BUILD_DIR/mongo/db/catalog/collection_catalog',
         'kv_drop_pending_ident_reaper',
         'kv_prefix',
-        ],
-    )
+    ],
+    LIBDEPS_PRIVATE=[
+        '$BUILD_DIR/mongo/db/catalog/disable_index_spec_namespace_generation',
+    ],
+)
 
 # Should not be referenced outside this SConscript file.
 env.Library(
diff --git a/src/mongo/db/storage/kv/kv_catalog.cpp b/src/mongo/db/storage/kv/kv_catalog.cpp
index d0dad7d8657..262a65eaedc 100644
--- a/src/mongo/db/storage/kv/kv_catalog.cpp
+++ b/src/mongo/db/storage/kv/kv_catalog.cpp
@@ -35,7 +35,9 @@
 
 #include "mongo/bson/util/bson_extract.h"
 #include "mongo/bson/util/builder.h"
+#include "mongo/db/catalog/disable_index_spec_namespace_generation_gen.h"
 #include "mongo/db/concurrency/d_concurrency.h"
+#include "mongo/db/index/index_descriptor.h"
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/operation_context.h"
 #include "mongo/db/storage/kv/kv_catalog_feature_tracker.h"
@@ -132,6 +134,10 @@ std::string escapeDbName(StringData dbname) {
     return escaped;
 }
 
+bool indexTypeSupportsPathLevelMultikeyTracking(StringData accessMethod) {
+    return accessMethod == IndexNames::BTREE || accessMethod == IndexNames::GEO_2DSPHERE;
+}
+
 }  // namespace
 
 using std::unique_ptr;
@@ -168,6 +174,61 @@ public:
     const Entry _entry;
 };
 
+class KVCatalog::AddIndexChange : public RecoveryUnit::Change {
+public:
+    AddIndexChange(OperationContext* opCtx, StorageEngineInterface* engine, StringData ident)
+        : _opCtx(opCtx), _engine(engine), _ident(ident.toString()) {}
+
+    virtual void commit(boost::optional<Timestamp>) {}
+    virtual void rollback() {
+        // Intentionally ignoring failure.
+        auto kvEngine = _engine->getEngine();
+        MONGO_COMPILER_VARIABLE_UNUSED auto status = kvEngine->dropIdent(_opCtx, _ident);
+    }
+
+    OperationContext* const _opCtx;
+    StorageEngineInterface* _engine;
+    const std::string _ident;
+};
+
+class KVCatalog::RemoveIndexChange : public RecoveryUnit::Change {
+public:
+    RemoveIndexChange(OperationContext* opCtx,
+                      StorageEngineInterface* engine,
+                      OptionalCollectionUUID uuid,
+                      const NamespaceString& indexNss,
+                      StringData indexName,
+                      StringData ident)
+        : _opCtx(opCtx),
+          _engine(engine),
+          _uuid(uuid),
+          _indexNss(indexNss),
+          _indexName(indexName),
+          _ident(ident.toString()) {}
+
+    virtual void rollback() {}
+    virtual void commit(boost::optional<Timestamp> commitTimestamp) {
+        // Intentionally ignoring failure here. Since we've removed the metadata pointing to the
+        // index, we should never see it again anyway.
+        if (_engine->getStorageEngine()->supportsPendingDrops() && commitTimestamp) {
+            log() << "Deferring table drop for index '" << _indexName << "' on collection '"
+                  << _indexNss << (_uuid ? " (" + _uuid->toString() + ")'" : "") << ". Ident: '"
+                  << _ident << "', commit timestamp: '" << commitTimestamp << "'";
+            _engine->addDropPendingIdent(*commitTimestamp, _indexNss, _ident);
+        } else {
+            auto kvEngine = _engine->getEngine();
+            MONGO_COMPILER_VARIABLE_UNUSED auto status = kvEngine->dropIdent(_opCtx, _ident);
+        }
+    }
+
+    OperationContext* const _opCtx;
+    StorageEngineInterface* _engine;
+    OptionalCollectionUUID _uuid;
+    const NamespaceString _indexNss;
+    const std::string _indexName;
+    const std::string _ident;
+};
+
 bool KVCatalog::FeatureTracker::isFeatureDocument(BSONObj obj) {
     BSONElement firstElem = obj.firstElement();
     if (firstElem.fieldNameStringData() == kIsFeatureDocumentFieldName) {
@@ -726,8 +787,7 @@ std::unique_ptr<CollectionCatalogEntry> KVCatalog::makeCollectionCatalogEntry(
         invariant(rs);
     }
 
-    return std::make_unique<KVCollectionCatalogEntry>(
-        _engine, this, nss.ns(), ident, std::move(rs));
+    return std::make_unique<KVCollectionCatalogEntry>(_engine, nss.ns(), ident, std::move(rs));
 }
 
 StatusWith<std::unique_ptr<CollectionCatalogEntry>> KVCatalog::createCollection(
@@ -774,8 +834,7 @@ StatusWith<std::unique_ptr<CollectionCatalogEntry>> KVCatalog::createCollection(
     auto rs = _engine->getEngine()->getGroupedRecordStore(opCtx, nss.ns(), ident, options, prefix);
     invariant(rs);
 
-    return std::make_unique<KVCollectionCatalogEntry>(
-        _engine, this, nss.ns(), ident, std::move(rs));
+    return std::make_unique<KVCollectionCatalogEntry>(_engine, nss.ns(), ident, std::move(rs));
 }
 
 Status KVCatalog::renameCollection(OperationContext* opCtx,
@@ -814,17 +873,17 @@ Status KVCatalog::dropCollection(OperationContext* opCtx, const NamespaceString&
     auto& catalog = CollectionCatalog::get(opCtx);
     auto uuid = catalog.lookupUUIDByNSS(nss);
 
-    invariant(entry->getTotalIndexCount(opCtx) == entry->getCompletedIndexCount(opCtx));
+    invariant(getTotalIndexCount(opCtx, nss) == getCompletedIndexCount(opCtx, nss));
 
     {
         std::vector<std::string> indexNames;
-        entry->getAllIndexes(opCtx, &indexNames);
+        getAllIndexes(opCtx, nss, &indexNames);
         for (size_t i = 0; i < indexNames.size(); i++) {
-            entry->removeIndex(opCtx, indexNames[i]).transitional_ignore();
+            Status status = removeIndex(opCtx, nss, indexNames[i]);
         }
     }
 
-    invariant(entry->getTotalIndexCount(opCtx) == 0);
+    invariant(getTotalIndexCount(opCtx, nss) == 0);
 
     const std::string ident = getCollectionIdent(nss);
 
@@ -854,4 +913,402 @@ Status KVCatalog::dropCollection(OperationContext* opCtx, const NamespaceString&
 
     return Status::OK();
 }
+
+void KVCatalog::updateCappedSize(OperationContext* opCtx, NamespaceString ns, long long size) {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+    md.options.cappedSize = size;
+    putMetaData(opCtx, ns, md);
+}
+
+void KVCatalog::updateTTLSetting(OperationContext* opCtx,
+                                 NamespaceString ns,
+                                 StringData idxName,
+                                 long long newExpireSeconds) {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+    int offset = md.findIndexOffset(idxName);
+    invariant(offset >= 0);
+    md.indexes[offset].updateTTLSetting(newExpireSeconds);
+    putMetaData(opCtx, ns, md);
+}
+
+bool KVCatalog::isEqualToMetadataUUID(OperationContext* opCtx,
+                                      NamespaceString ns,
+                                      OptionalCollectionUUID uuid) {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+    return md.options.uuid && md.options.uuid == uuid;
+}
+
+void KVCatalog::setIsTemp(OperationContext* opCtx, NamespaceString ns, bool isTemp) {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+    md.options.temp = isTemp;
+    putMetaData(opCtx, ns, md);
+}
+
+boost::optional<std::string> KVCatalog::getSideWritesIdent(OperationContext* opCtx,
+                                                           NamespaceString ns,
+                                                           StringData indexName) const {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+    int offset = md.findIndexOffset(indexName);
+    invariant(offset >= 0);
+    return md.indexes[offset].sideWritesIdent;
+}
+
+void KVCatalog::setIndexKeyStringWithLongTypeBitsExistsOnDisk(OperationContext* opCtx) {
+    const auto feature = FeatureTracker::RepairableFeature::kIndexKeyStringWithLongTypeBits;
+    if (!getFeatureTracker()->isRepairableFeatureInUse(opCtx, feature)) {
+        getFeatureTracker()->markRepairableFeatureAsInUse(opCtx, feature);
+    }
+}
+
+void KVCatalog::updateValidator(OperationContext* opCtx,
+                                NamespaceString ns,
+                                const BSONObj& validator,
+                                StringData validationLevel,
+                                StringData validationAction) {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+    md.options.validator = validator;
+    md.options.validationLevel = validationLevel.toString();
+    md.options.validationAction = validationAction.toString();
+    putMetaData(opCtx, ns, md);
+}
+
+void KVCatalog::updateIndexMetadata(OperationContext* opCtx,
+                                    NamespaceString ns,
+                                    const IndexDescriptor* desc) {
+    // Update any metadata Ident has for this index
+    const string ident = getIndexIdent(opCtx, ns, desc->indexName());
+    auto kvEngine = _engine->getEngine();
+    kvEngine->alterIdentMetadata(opCtx, ident, desc);
+}
+
+Status KVCatalog::removeIndex(OperationContext* opCtx, NamespaceString ns, StringData indexName) {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+
+    if (md.findIndexOffset(indexName) < 0)
+        return Status::OK();  // never had the index so nothing to do.
+
+    const string ident = getIndexIdent(opCtx, ns, indexName);
+
+    md.eraseIndex(indexName);
+    putMetaData(opCtx, ns, md);
+
+    // Lazily remove to isolate underlying engine from rollback.
+    opCtx->recoveryUnit()->registerChange(new RemoveIndexChange(
+        opCtx, _engine, md.options.uuid, ns.makeIndexNamespace(indexName), indexName, ident));
+    return Status::OK();
+}
+
+Status KVCatalog::prepareForIndexBuild(OperationContext* opCtx,
+                                       NamespaceString ns,
+                                       const IndexDescriptor* spec,
+                                       IndexBuildProtocol indexBuildProtocol,
+                                       bool isBackgroundSecondaryBuild) {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+
+    KVPrefix prefix = KVPrefix::getNextPrefix(ns);
+    BSONCollectionCatalogEntry::IndexMetaData imd;
+    imd.spec = spec->infoObj();
+    imd.ready = false;
+    imd.multikey = false;
+    imd.prefix = prefix;
+    imd.isBackgroundSecondaryBuild = isBackgroundSecondaryBuild;
+    imd.runTwoPhaseBuild = indexBuildProtocol == IndexBuildProtocol::kTwoPhase;
+
+    if (indexTypeSupportsPathLevelMultikeyTracking(spec->getAccessMethodName())) {
+        const auto feature = FeatureTracker::RepairableFeature::kPathLevelMultikeyTracking;
+        if (!getFeatureTracker()->isRepairableFeatureInUse(opCtx, feature)) {
+            getFeatureTracker()->markRepairableFeatureAsInUse(opCtx, feature);
+        }
+        imd.multikeyPaths = MultikeyPaths{static_cast<size_t>(spec->keyPattern().nFields())};
+    }
+
+    // Mark collation feature as in use if the index has a non-simple collation.
+    if (imd.spec["collation"]) {
+        const auto feature = KVCatalog::FeatureTracker::NonRepairableFeature::kCollation;
+        if (!getFeatureTracker()->isNonRepairableFeatureInUse(opCtx, feature)) {
+            getFeatureTracker()->markNonRepairableFeatureAsInUse(opCtx, feature);
+        }
+    }
+
+    md.indexes.push_back(imd);
+    putMetaData(opCtx, ns, md);
+
+    string ident = getIndexIdent(opCtx, ns, spec->indexName());
+
+    auto kvEngine = _engine->getEngine();
+    const Status status = kvEngine->createGroupedSortedDataInterface(
+        opCtx, getCollectionOptions(opCtx, ns), ident, spec, prefix);
+    if (status.isOK()) {
+        opCtx->recoveryUnit()->registerChange(new AddIndexChange(opCtx, _engine, ident));
+    }
+
+    return status;
+}
+
+bool KVCatalog::isTwoPhaseIndexBuild(OperationContext* opCtx,
+                                     NamespaceString ns,
+                                     StringData indexName) const {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+    int offset = md.findIndexOffset(indexName);
+    invariant(offset >= 0);
+    return md.indexes[offset].runTwoPhaseBuild;
+}
+
+void KVCatalog::setIndexBuildScanning(OperationContext* opCtx,
+                                      NamespaceString ns,
+                                      StringData indexName,
+                                      std::string sideWritesIdent,
+                                      boost::optional<std::string> constraintViolationsIdent) {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+    int offset = md.findIndexOffset(indexName);
+    invariant(offset >= 0);
+    invariant(!md.indexes[offset].ready);
+    invariant(!md.indexes[offset].buildPhase);
+    invariant(md.indexes[offset].runTwoPhaseBuild);
+
+    md.indexes[offset].buildPhase = BSONCollectionCatalogEntry::kIndexBuildScanning.toString();
+    md.indexes[offset].sideWritesIdent = sideWritesIdent;
+    md.indexes[offset].constraintViolationsIdent = constraintViolationsIdent;
+    putMetaData(opCtx, ns, md);
+}
+
+bool KVCatalog::isIndexBuildScanning(OperationContext* opCtx,
+                                     NamespaceString ns,
+                                     StringData indexName) const {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+    int offset = md.findIndexOffset(indexName);
+    invariant(offset >= 0);
+    return md.indexes[offset].buildPhase ==
+        BSONCollectionCatalogEntry::kIndexBuildScanning.toString();
+}
+
+void KVCatalog::setIndexBuildDraining(OperationContext* opCtx,
+                                      NamespaceString ns,
+                                      StringData indexName) {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+    int offset = md.findIndexOffset(indexName);
+    invariant(offset >= 0);
+    invariant(!md.indexes[offset].ready);
+    invariant(md.indexes[offset].runTwoPhaseBuild);
+    invariant(md.indexes[offset].buildPhase ==
+              BSONCollectionCatalogEntry::kIndexBuildScanning.toString());
+
+    md.indexes[offset].buildPhase = BSONCollectionCatalogEntry::kIndexBuildDraining.toString();
+    putMetaData(opCtx, ns, md);
+}
+
+bool KVCatalog::isIndexBuildDraining(OperationContext* opCtx,
+                                     NamespaceString ns,
+                                     StringData indexName) const {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+    int offset = md.findIndexOffset(indexName);
+    invariant(offset >= 0);
+    return md.indexes[offset].buildPhase ==
+        BSONCollectionCatalogEntry::kIndexBuildDraining.toString();
+}
+
+void KVCatalog::indexBuildSuccess(OperationContext* opCtx,
+                                  NamespaceString ns,
+                                  StringData indexName) {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+    int offset = md.findIndexOffset(indexName);
+    invariant(offset >= 0);
+    md.indexes[offset].ready = true;
+    md.indexes[offset].runTwoPhaseBuild = false;
+    md.indexes[offset].buildPhase = boost::none;
+    md.indexes[offset].sideWritesIdent = boost::none;
+    md.indexes[offset].constraintViolationsIdent = boost::none;
+    putMetaData(opCtx, ns, md);
+}
+
+bool KVCatalog::isIndexMultikey(OperationContext* opCtx,
+                                NamespaceString ns,
+                                StringData indexName,
+                                MultikeyPaths* multikeyPaths) const {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+
+    int offset = md.findIndexOffset(indexName);
+    invariant(offset >= 0);
+
+    if (multikeyPaths && !md.indexes[offset].multikeyPaths.empty()) {
+        *multikeyPaths = md.indexes[offset].multikeyPaths;
+    }
+
+    return md.indexes[offset].multikey;
+}
+
+bool KVCatalog::setIndexIsMultikey(OperationContext* opCtx,
+                                   NamespaceString ns,
+                                   StringData indexName,
+                                   const MultikeyPaths& multikeyPaths) {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+
+    int offset = md.findIndexOffset(indexName);
+    invariant(offset >= 0);
+
+    const bool tracksPathLevelMultikeyInfo = !md.indexes[offset].multikeyPaths.empty();
+    if (tracksPathLevelMultikeyInfo) {
+        invariant(!multikeyPaths.empty());
+        invariant(multikeyPaths.size() == md.indexes[offset].multikeyPaths.size());
+    } else {
+        invariant(multikeyPaths.empty());
+
+        if (md.indexes[offset].multikey) {
+            // The index is already set as multikey and we aren't tracking path-level multikey
+            // information for it. We return false to indicate that the index metadata is unchanged.
+            return false;
+        }
+    }
+
+    md.indexes[offset].multikey = true;
+
+    if (tracksPathLevelMultikeyInfo) {
+        bool newPathIsMultikey = false;
+        bool somePathIsMultikey = false;
+
+        // Store new path components that cause this index to be multikey in catalog's index
+        // metadata.
+        for (size_t i = 0; i < multikeyPaths.size(); ++i) {
+            std::set<size_t>& indexMultikeyComponents = md.indexes[offset].multikeyPaths[i];
+            for (const auto multikeyComponent : multikeyPaths[i]) {
+                auto result = indexMultikeyComponents.insert(multikeyComponent);
+                newPathIsMultikey = newPathIsMultikey || result.second;
+                somePathIsMultikey = true;
+            }
+        }
+
+        // If all of the sets in the multikey paths vector were empty, then no component of any
+        // indexed field caused the index to be multikey. setIndexIsMultikey() therefore shouldn't
+        // have been called.
+        invariant(somePathIsMultikey);
+
+        if (!newPathIsMultikey) {
+            // We return false to indicate that the index metadata is unchanged.
+            return false;
+        }
+    }
+
+    putMetaData(opCtx, ns, md);
+    return true;
+}
+
+boost::optional<std::string> KVCatalog::getConstraintViolationsIdent(OperationContext* opCtx,
+                                                                     NamespaceString ns,
+                                                                     StringData indexName) const {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+    int offset = md.findIndexOffset(indexName);
+    invariant(offset >= 0);
+    return md.indexes[offset].constraintViolationsIdent;
+}
+
+long KVCatalog::getIndexBuildVersion(OperationContext* opCtx,
+                                     NamespaceString ns,
+                                     StringData indexName) const {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+    int offset = md.findIndexOffset(indexName);
+    invariant(offset >= 0);
+    return md.indexes[offset].versionOfBuild;
+}
+
+CollectionOptions KVCatalog::getCollectionOptions(OperationContext* opCtx,
+                                                  NamespaceString ns) const {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+    return md.options;
+}
+
+int KVCatalog::getTotalIndexCount(OperationContext* opCtx, NamespaceString ns) const {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+
+    return static_cast<int>(md.indexes.size());
+}
+
+int KVCatalog::getCompletedIndexCount(OperationContext* opCtx, NamespaceString ns) const {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+
+    int num = 0;
+    for (unsigned i = 0; i < md.indexes.size(); i++) {
+        if (md.indexes[i].ready)
+            num++;
+    }
+    return num;
+}
+
+BSONObj KVCatalog::getIndexSpec(OperationContext* opCtx,
+                                NamespaceString ns,
+                                StringData indexName) const {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+
+    int offset = md.findIndexOffset(indexName);
+    invariant(offset >= 0);
+
+    BSONObj spec = md.indexes[offset].spec.getOwned();
+    if (spec.hasField("ns") || disableIndexSpecNamespaceGeneration.load()) {
+        return spec;
+    }
+
+    BSONObj nsObj = BSON("ns" << ns.ns());
+    spec = spec.addField(nsObj.firstElement());
+    return spec;
+}
+
+void KVCatalog::getAllIndexes(OperationContext* opCtx,
+                              NamespaceString ns,
+                              std::vector<std::string>* names) const {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+
+    for (unsigned i = 0; i < md.indexes.size(); i++) {
+        names->push_back(md.indexes[i].spec["name"].String());
+    }
+}
+
+void KVCatalog::getReadyIndexes(OperationContext* opCtx,
+                                NamespaceString ns,
+                                std::vector<std::string>* names) const {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+
+    for (unsigned i = 0; i < md.indexes.size(); i++) {
+        if (md.indexes[i].ready)
+            names->push_back(md.indexes[i].spec["name"].String());
+    }
+}
+
+void KVCatalog::getAllUniqueIndexes(OperationContext* opCtx,
+                                    NamespaceString ns,
+                                    std::vector<std::string>* names) const {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+
+    for (unsigned i = 0; i < md.indexes.size(); i++) {
+        if (md.indexes[i].spec["unique"]) {
+            std::string indexName = md.indexes[i].spec["name"].String();
+            names->push_back(indexName);
+        }
+    }
+}
+
+bool KVCatalog::isIndexPresent(OperationContext* opCtx,
+                               NamespaceString ns,
+                               StringData indexName) const {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+    int offset = md.findIndexOffset(indexName);
+    return offset >= 0;
+}
+
+bool KVCatalog::isIndexReady(OperationContext* opCtx,
+                             NamespaceString ns,
+                             StringData indexName) const {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+
+    int offset = md.findIndexOffset(indexName);
+    invariant(offset >= 0);
+    return md.indexes[offset].ready;
+}
+
+KVPrefix KVCatalog::getIndexPrefix(OperationContext* opCtx,
+                                   NamespaceString ns,
+                                   StringData indexName) const {
+    BSONCollectionCatalogEntry::MetaData md = getMetaData(opCtx, ns);
+    int offset = md.findIndexOffset(indexName);
+    invariant(offset >= 0);
+    return md.indexes[offset].prefix;
+}
 }
diff --git a/src/mongo/db/storage/kv/kv_catalog.h b/src/mongo/db/storage/kv/kv_catalog.h
index 957b12c38e1..6d46dcc482b 100644
--- a/src/mongo/db/storage/kv/kv_catalog.h
+++ b/src/mongo/db/storage/kv/kv_catalog.h
@@ -38,6 +38,7 @@
 #include "mongo/db/catalog/collection_options.h"
 #include "mongo/db/record_id.h"
 #include "mongo/db/storage/bson_collection_catalog_entry.h"
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/db/storage/kv/kv_prefix.h"
 #include "mongo/stdx/mutex.h"
 
@@ -47,7 +48,7 @@ class OperationContext;
 class RecordStore;
 class StorageEngineInterface;
 
-class KVCatalog {
+class KVCatalog : public DurableCatalog {
 public:
     class FeatureTracker;
 
@@ -96,19 +97,10 @@ public:
         return _rs;
     }
 
-    /**
-     * Create an entry in the catalog for an orphaned collection found in the
-     * storage engine. Return the generated ns of the collection.
-     * Note that this function does not recreate the _id index on the collection because it does not
-     * have access to index catalog.
-     */
     StatusWith<std::string> newOrphanedIdent(OperationContext* opCtx, std::string ident);
 
     std::string getFilesystemPathForDb(const std::string& dbName) const;
 
-    /**
-     * Generate an internal ident name.
-     */
     std::string newInternalIdent();
 
     std::unique_ptr<CollectionCatalogEntry> makeCollectionCatalogEntry(OperationContext* opCtx,
@@ -128,9 +120,118 @@ public:
 
     Status dropCollection(OperationContext* opCtx, const NamespaceString& nss);
 
+    void updateCappedSize(OperationContext* opCtx, NamespaceString ns, long long size);
+
+    void updateTTLSetting(OperationContext* opCtx,
+                          NamespaceString ns,
+                          StringData idxName,
+                          long long newExpireSeconds);
+
+    bool isEqualToMetadataUUID(OperationContext* opCtx,
+                               NamespaceString ns,
+                               OptionalCollectionUUID uuid);
+
+    void setIsTemp(OperationContext* opCtx, NamespaceString ns, bool isTemp);
+
+    boost::optional<std::string> getSideWritesIdent(OperationContext* opCtx,
+                                                    NamespaceString ns,
+                                                    StringData indexName) const;
+
+    // TODO SERVER-36385 Remove this function: we don't set the feature tracker bit in 4.4 because
+    // 4.4 can only downgrade to 4.2 which can read long TypeBits.
+    void setIndexKeyStringWithLongTypeBitsExistsOnDisk(OperationContext* opCtx);
+
+    void updateValidator(OperationContext* opCtx,
+                         NamespaceString ns,
+                         const BSONObj& validator,
+                         StringData validationLevel,
+                         StringData validationAction);
+
+    void updateIndexMetadata(OperationContext* opCtx,
+                             NamespaceString ns,
+                             const IndexDescriptor* desc);
+
+    Status removeIndex(OperationContext* opCtx, NamespaceString ns, StringData indexName);
+
+    Status prepareForIndexBuild(OperationContext* opCtx,
+                                NamespaceString ns,
+                                const IndexDescriptor* spec,
+                                IndexBuildProtocol indexBuildProtocol,
+                                bool isBackgroundSecondaryBuild);
+
+    bool isTwoPhaseIndexBuild(OperationContext* opCtx,
+                              NamespaceString ns,
+                              StringData indexName) const;
+
+    void setIndexBuildScanning(OperationContext* opCtx,
+                               NamespaceString ns,
+                               StringData indexName,
+                               std::string sideWritesIdent,
+                               boost::optional<std::string> constraintViolationsIdent);
+
+
+    bool isIndexBuildScanning(OperationContext* opCtx,
+                              NamespaceString ns,
+                              StringData indexName) const;
+
+    void setIndexBuildDraining(OperationContext* opCtx, NamespaceString ns, StringData indexName);
+
+    bool isIndexBuildDraining(OperationContext* opCtx,
+                              NamespaceString ns,
+                              StringData indexName) const;
+
+    void indexBuildSuccess(OperationContext* opCtx, NamespaceString ns, StringData indexName);
+
+    bool isIndexMultikey(OperationContext* opCtx,
+                         NamespaceString ns,
+                         StringData indexName,
+                         MultikeyPaths* multikeyPaths) const;
+
+    bool setIndexIsMultikey(OperationContext* opCtx,
+                            NamespaceString ns,
+                            StringData indexName,
+                            const MultikeyPaths& multikeyPaths);
+
+    boost::optional<std::string> getConstraintViolationsIdent(OperationContext* opCtx,
+                                                              NamespaceString ns,
+                                                              StringData indexName) const;
+
+    long getIndexBuildVersion(OperationContext* opCtx,
+                              NamespaceString ns,
+                              StringData indexName) const;
+
+    CollectionOptions getCollectionOptions(OperationContext* opCtx, NamespaceString ns) const;
+
+    int getTotalIndexCount(OperationContext* opCtx, NamespaceString ns) const;
+
+    int getCompletedIndexCount(OperationContext* opCtx, NamespaceString ns) const;
+
+    BSONObj getIndexSpec(OperationContext* opCtx, NamespaceString ns, StringData indexName) const;
+
+    void getAllIndexes(OperationContext* opCtx,
+                       NamespaceString ns,
+                       std::vector<std::string>* names) const;
+
+    void getReadyIndexes(OperationContext* opCtx,
+                         NamespaceString ns,
+                         std::vector<std::string>* names) const;
+    void getAllUniqueIndexes(OperationContext* opCtx,
+                             NamespaceString ns,
+                             std::vector<std::string>* names) const;
+
+    bool isIndexPresent(OperationContext* opCtx, NamespaceString ns, StringData indexName) const;
+
+    bool isIndexReady(OperationContext* opCtx, NamespaceString ns, StringData indexName) const;
+
+    KVPrefix getIndexPrefix(OperationContext* opCtx,
+                            NamespaceString ns,
+                            StringData indexName) const;
+
 private:
     class AddIdentChange;
     class RemoveIdentChange;
+    class AddIndexChange;
+    class RemoveIndexChange;
 
     friend class StorageEngineImpl;
     friend class KVCatalogTest;
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 43150b4d6b3..d971924cadd 100644
--- a/src/mongo/db/storage/kv/kv_collection_catalog_entry.cpp
+++ b/src/mongo/db/storage/kv/kv_collection_catalog_entry.cpp
@@ -37,7 +37,7 @@
 
 #include "mongo/db/catalog/collection_catalog.h"
 #include "mongo/db/index/index_descriptor.h"
-#include "mongo/db/storage/kv/kv_catalog.h"
+#include "mongo/db/storage/durable_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/storage_engine_impl.h"
@@ -47,351 +47,12 @@ namespace mongo {
 
 using std::string;
 
-namespace {
-
-bool indexTypeSupportsPathLevelMultikeyTracking(StringData accessMethod) {
-    return accessMethod == IndexNames::BTREE || accessMethod == IndexNames::GEO_2DSPHERE;
-}
-
-}  // namespace
-
-class KVCollectionCatalogEntry::AddIndexChange : public RecoveryUnit::Change {
-public:
-    AddIndexChange(OperationContext* opCtx, KVCollectionCatalogEntry* cce, StringData ident)
-        : _opCtx(opCtx), _cce(cce), _ident(ident.toString()) {}
-
-    virtual void commit(boost::optional<Timestamp>) {}
-    virtual void rollback() {
-        // Intentionally ignoring failure.
-        auto kvEngine = _cce->_engine->getEngine();
-        MONGO_COMPILER_VARIABLE_UNUSED auto status = kvEngine->dropIdent(_opCtx, _ident);
-    }
-
-    OperationContext* const _opCtx;
-    KVCollectionCatalogEntry* const _cce;
-    const std::string _ident;
-};
-
-class KVCollectionCatalogEntry::RemoveIndexChange : public RecoveryUnit::Change {
-public:
-    RemoveIndexChange(OperationContext* opCtx,
-                      KVCollectionCatalogEntry* cce,
-                      OptionalCollectionUUID uuid,
-                      const NamespaceString& indexNss,
-                      StringData indexName,
-                      StringData ident)
-        : _opCtx(opCtx),
-          _cce(cce),
-          _uuid(uuid),
-          _indexNss(indexNss),
-          _indexName(indexName),
-          _ident(ident.toString()) {}
-
-    virtual void rollback() {}
-    virtual void commit(boost::optional<Timestamp> commitTimestamp) {
-        // Intentionally ignoring failure here. Since we've removed the metadata pointing to the
-        // index, we should never see it again anyway.
-        auto engine = _cce->_engine;
-        auto storageEngine = engine->getStorageEngine();
-        if (storageEngine->supportsPendingDrops() && commitTimestamp) {
-            log() << "Deferring table drop for index '" << _indexName << "' on collection '"
-                  << _indexNss << (_uuid ? " (" + _uuid->toString() + ")'" : "") << ". Ident: '"
-                  << _ident << "', commit timestamp: '" << commitTimestamp << "'";
-            engine->addDropPendingIdent(*commitTimestamp, _indexNss, _ident);
-        } else {
-            auto kvEngine = engine->getEngine();
-            MONGO_COMPILER_VARIABLE_UNUSED auto status = kvEngine->dropIdent(_opCtx, _ident);
-        }
-    }
-
-    OperationContext* const _opCtx;
-    KVCollectionCatalogEntry* const _cce;
-    OptionalCollectionUUID _uuid;
-    const NamespaceString _indexNss;
-    const std::string _indexName;
-    const std::string _ident;
-};
-
-
 KVCollectionCatalogEntry::KVCollectionCatalogEntry(StorageEngineInterface* engine,
-                                                   KVCatalog* catalog,
                                                    StringData ns,
                                                    StringData ident,
                                                    std::unique_ptr<RecordStore> rs)
     : BSONCollectionCatalogEntry(ns),
       _engine(engine),
-      _catalog(catalog),
       _ident(ident.toString()),
       _recordStore(std::move(rs)) {}
-
-KVCollectionCatalogEntry::~KVCollectionCatalogEntry() {}
-
-bool KVCollectionCatalogEntry::setIndexIsMultikey(OperationContext* opCtx,
-                                                  StringData indexName,
-                                                  const MultikeyPaths& multikeyPaths) {
-    MetaData md = _getMetaData(opCtx);
-
-    int offset = md.findIndexOffset(indexName);
-    invariant(offset >= 0);
-
-    const bool tracksPathLevelMultikeyInfo = !md.indexes[offset].multikeyPaths.empty();
-    if (tracksPathLevelMultikeyInfo) {
-        invariant(!multikeyPaths.empty());
-        invariant(multikeyPaths.size() == md.indexes[offset].multikeyPaths.size());
-    } else {
-        invariant(multikeyPaths.empty());
-
-        if (md.indexes[offset].multikey) {
-            // The index is already set as multikey and we aren't tracking path-level multikey
-            // information for it. We return false to indicate that the index metadata is unchanged.
-            return false;
-        }
-    }
-
-    md.indexes[offset].multikey = true;
-
-    if (tracksPathLevelMultikeyInfo) {
-        bool newPathIsMultikey = false;
-        bool somePathIsMultikey = false;
-
-        // Store new path components that cause this index to be multikey in catalog's index
-        // metadata.
-        for (size_t i = 0; i < multikeyPaths.size(); ++i) {
-            std::set<size_t>& indexMultikeyComponents = md.indexes[offset].multikeyPaths[i];
-            for (const auto multikeyComponent : multikeyPaths[i]) {
-                auto result = indexMultikeyComponents.insert(multikeyComponent);
-                newPathIsMultikey = newPathIsMultikey || result.second;
-                somePathIsMultikey = true;
-            }
-        }
-
-        // If all of the sets in the multikey paths vector were empty, then no component of any
-        // indexed field caused the index to be multikey. setIndexIsMultikey() therefore shouldn't
-        // have been called.
-        invariant(somePathIsMultikey);
-
-        if (!newPathIsMultikey) {
-            // We return false to indicate that the index metadata is unchanged.
-            return false;
-        }
-    }
-
-    _catalog->putMetaData(opCtx, ns(), md);
-    return true;
-}
-
-void KVCollectionCatalogEntry::setIndexKeyStringWithLongTypeBitsExistsOnDisk(
-    OperationContext* opCtx) {
-    const auto feature =
-        KVCatalog::FeatureTracker::RepairableFeature::kIndexKeyStringWithLongTypeBits;
-    if (!_catalog->getFeatureTracker()->isRepairableFeatureInUse(opCtx, feature)) {
-        _catalog->getFeatureTracker()->markRepairableFeatureAsInUse(opCtx, feature);
-    }
-}
-
-Status KVCollectionCatalogEntry::removeIndex(OperationContext* opCtx, StringData indexName) {
-    MetaData md = _getMetaData(opCtx);
-
-    if (md.findIndexOffset(indexName) < 0)
-        return Status::OK();  // never had the index so nothing to do.
-
-    const string ident = _catalog->getIndexIdent(opCtx, ns(), indexName);
-
-    md.eraseIndex(indexName);
-    _catalog->putMetaData(opCtx, ns(), md);
-
-    // Lazily remove to isolate underlying engine from rollback.
-    opCtx->recoveryUnit()->registerChange(new RemoveIndexChange(
-        opCtx, this, md.options.uuid, ns().makeIndexNamespace(indexName), indexName, ident));
-    return Status::OK();
-}
-
-Status KVCollectionCatalogEntry::prepareForIndexBuild(OperationContext* opCtx,
-                                                      const IndexDescriptor* spec,
-                                                      IndexBuildProtocol indexBuildProtocol,
-                                                      bool isBackgroundSecondaryBuild) {
-    MetaData md = _getMetaData(opCtx);
-
-    KVPrefix prefix = KVPrefix::getNextPrefix(ns());
-    IndexMetaData imd;
-    imd.spec = spec->infoObj();
-    imd.ready = false;
-    imd.multikey = false;
-    imd.prefix = prefix;
-    imd.isBackgroundSecondaryBuild = isBackgroundSecondaryBuild;
-    imd.runTwoPhaseBuild = indexBuildProtocol == IndexBuildProtocol::kTwoPhase;
-
-    if (indexTypeSupportsPathLevelMultikeyTracking(spec->getAccessMethodName())) {
-        const auto feature =
-            KVCatalog::FeatureTracker::RepairableFeature::kPathLevelMultikeyTracking;
-        if (!_catalog->getFeatureTracker()->isRepairableFeatureInUse(opCtx, feature)) {
-            _catalog->getFeatureTracker()->markRepairableFeatureAsInUse(opCtx, feature);
-        }
-        imd.multikeyPaths = MultikeyPaths{static_cast<size_t>(spec->keyPattern().nFields())};
-    }
-
-    // Mark collation feature as in use if the index has a non-simple collation.
-    if (imd.spec["collation"]) {
-        const auto feature = KVCatalog::FeatureTracker::NonRepairableFeature::kCollation;
-        if (!_catalog->getFeatureTracker()->isNonRepairableFeatureInUse(opCtx, feature)) {
-            _catalog->getFeatureTracker()->markNonRepairableFeatureAsInUse(opCtx, feature);
-        }
-    }
-
-    md.indexes.push_back(imd);
-    _catalog->putMetaData(opCtx, ns(), md);
-
-    string ident = _catalog->getIndexIdent(opCtx, ns(), spec->indexName());
-
-    auto kvEngine = _engine->getEngine();
-    const Status status = kvEngine->createGroupedSortedDataInterface(opCtx, ident, spec, prefix);
-    if (status.isOK()) {
-        opCtx->recoveryUnit()->registerChange(new AddIndexChange(opCtx, this, ident));
-    }
-
-    return status;
-}
-
-bool KVCollectionCatalogEntry::isTwoPhaseIndexBuild(OperationContext* opCtx,
-                                                    StringData indexName) const {
-    MetaData md = _getMetaData(opCtx);
-    int offset = md.findIndexOffset(indexName);
-    invariant(offset >= 0);
-    return md.indexes[offset].runTwoPhaseBuild;
-}
-
-long KVCollectionCatalogEntry::getIndexBuildVersion(OperationContext* opCtx,
-                                                    StringData indexName) const {
-    MetaData md = _getMetaData(opCtx);
-    int offset = md.findIndexOffset(indexName);
-    invariant(offset >= 0);
-    return md.indexes[offset].versionOfBuild;
-}
-
-void KVCollectionCatalogEntry::setIndexBuildScanning(
-    OperationContext* opCtx,
-    StringData indexName,
-    std::string sideWritesIdent,
-    boost::optional<std::string> constraintViolationsIdent) {
-    MetaData md = _getMetaData(opCtx);
-    int offset = md.findIndexOffset(indexName);
-    invariant(offset >= 0);
-    invariant(!md.indexes[offset].ready);
-    invariant(!md.indexes[offset].buildPhase);
-    invariant(md.indexes[offset].runTwoPhaseBuild);
-
-    md.indexes[offset].buildPhase = kIndexBuildScanning.toString();
-    md.indexes[offset].sideWritesIdent = sideWritesIdent;
-    md.indexes[offset].constraintViolationsIdent = constraintViolationsIdent;
-    _catalog->putMetaData(opCtx, ns(), md);
-}
-
-bool KVCollectionCatalogEntry::isIndexBuildScanning(OperationContext* opCtx,
-                                                    StringData indexName) const {
-    MetaData md = _getMetaData(opCtx);
-    int offset = md.findIndexOffset(indexName);
-    invariant(offset >= 0);
-    return md.indexes[offset].buildPhase == kIndexBuildScanning.toString();
-}
-
-void KVCollectionCatalogEntry::setIndexBuildDraining(OperationContext* opCtx,
-                                                     StringData indexName) {
-    MetaData md = _getMetaData(opCtx);
-    int offset = md.findIndexOffset(indexName);
-    invariant(offset >= 0);
-    invariant(!md.indexes[offset].ready);
-    invariant(md.indexes[offset].runTwoPhaseBuild);
-    invariant(md.indexes[offset].buildPhase == kIndexBuildScanning.toString());
-
-    md.indexes[offset].buildPhase = kIndexBuildDraining.toString();
-    _catalog->putMetaData(opCtx, ns(), md);
-}
-
-bool KVCollectionCatalogEntry::isIndexBuildDraining(OperationContext* opCtx,
-                                                    StringData indexName) const {
-    MetaData md = _getMetaData(opCtx);
-    int offset = md.findIndexOffset(indexName);
-    invariant(offset >= 0);
-    return md.indexes[offset].buildPhase == kIndexBuildDraining.toString();
-}
-
-void KVCollectionCatalogEntry::indexBuildSuccess(OperationContext* opCtx, StringData indexName) {
-    MetaData md = _getMetaData(opCtx);
-    int offset = md.findIndexOffset(indexName);
-    invariant(offset >= 0);
-    md.indexes[offset].ready = true;
-    md.indexes[offset].runTwoPhaseBuild = false;
-    md.indexes[offset].buildPhase = boost::none;
-    md.indexes[offset].sideWritesIdent = boost::none;
-    md.indexes[offset].constraintViolationsIdent = boost::none;
-    _catalog->putMetaData(opCtx, ns(), md);
-}
-
-boost::optional<std::string> KVCollectionCatalogEntry::getSideWritesIdent(
-    OperationContext* opCtx, StringData indexName) const {
-    MetaData md = _getMetaData(opCtx);
-    int offset = md.findIndexOffset(indexName);
-    invariant(offset >= 0);
-    return md.indexes[offset].sideWritesIdent;
-}
-
-boost::optional<std::string> KVCollectionCatalogEntry::getConstraintViolationsIdent(
-    OperationContext* opCtx, StringData indexName) const {
-    MetaData md = _getMetaData(opCtx);
-    int offset = md.findIndexOffset(indexName);
-    invariant(offset >= 0);
-    return md.indexes[offset].constraintViolationsIdent;
-}
-
-void KVCollectionCatalogEntry::updateTTLSetting(OperationContext* opCtx,
-                                                StringData idxName,
-                                                long long newExpireSeconds) {
-    MetaData md = _getMetaData(opCtx);
-    int offset = md.findIndexOffset(idxName);
-    invariant(offset >= 0);
-    md.indexes[offset].updateTTLSetting(newExpireSeconds);
-    _catalog->putMetaData(opCtx, ns(), md);
-}
-
-void KVCollectionCatalogEntry::updateIndexMetadata(OperationContext* opCtx,
-                                                   const IndexDescriptor* desc) {
-    // Update any metadata Ident has for this index
-    const string ident = _catalog->getIndexIdent(opCtx, ns(), desc->indexName());
-    auto kvEngine = _engine->getEngine();
-    kvEngine->alterIdentMetadata(opCtx, ident, desc);
-}
-
-bool KVCollectionCatalogEntry::isEqualToMetadataUUID(OperationContext* opCtx,
-                                                     OptionalCollectionUUID uuid) {
-    MetaData md = _getMetaData(opCtx);
-    return md.options.uuid && md.options.uuid == uuid;
-}
-
-void KVCollectionCatalogEntry::updateValidator(OperationContext* opCtx,
-                                               const BSONObj& validator,
-                                               StringData validationLevel,
-                                               StringData validationAction) {
-    MetaData md = _getMetaData(opCtx);
-    md.options.validator = validator;
-    md.options.validationLevel = validationLevel.toString();
-    md.options.validationAction = validationAction.toString();
-    _catalog->putMetaData(opCtx, ns(), md);
-}
-
-void KVCollectionCatalogEntry::setIsTemp(OperationContext* opCtx, bool isTemp) {
-    MetaData md = _getMetaData(opCtx);
-    md.options.temp = isTemp;
-    _catalog->putMetaData(opCtx, ns(), md);
-}
-
-void KVCollectionCatalogEntry::updateCappedSize(OperationContext* opCtx, long long size) {
-    MetaData md = _getMetaData(opCtx);
-    md.options.cappedSize = size;
-    _catalog->putMetaData(opCtx, ns(), md);
-}
-
-BSONCollectionCatalogEntry::MetaData KVCollectionCatalogEntry::_getMetaData(
-    OperationContext* opCtx) const {
-    return _catalog->getMetaData(opCtx, ns());
-}
 }
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 644a00f8943..e9133a607f5 100644
--- a/src/mongo/db/storage/kv/kv_collection_catalog_entry.h
+++ b/src/mongo/db/storage/kv/kv_collection_catalog_entry.h
@@ -38,76 +38,16 @@
 
 namespace mongo {
 
-class KVCatalog;
 class StorageEngineInterface;
 
 class KVCollectionCatalogEntry final : public BSONCollectionCatalogEntry {
 public:
     KVCollectionCatalogEntry(StorageEngineInterface* engine,
-                             KVCatalog* catalog,
                              StringData ns,
                              StringData ident,
                              std::unique_ptr<RecordStore> rs);
 
-    ~KVCollectionCatalogEntry() final;
-
-    int getMaxAllowedIndexes() const final {
-        return 64;
-    };
-
-    bool setIndexIsMultikey(OperationContext* opCtx,
-                            StringData indexName,
-                            const MultikeyPaths& multikeyPaths) final;
-
-    // TODO SERVER-36385 Remove this function: we don't set the feature tracker bit in 4.4 because
-    // 4.4 can only downgrade to 4.2 which can read long TypeBits.
-    void setIndexKeyStringWithLongTypeBitsExistsOnDisk(OperationContext* opCtx) final;
-
-    Status removeIndex(OperationContext* opCtx, StringData indexName) final;
-
-    Status prepareForIndexBuild(OperationContext* opCtx,
-                                const IndexDescriptor* spec,
-                                IndexBuildProtocol indexBuildProtocol,
-                                bool isBackgroundSecondaryBuild) final;
-
-    bool isTwoPhaseIndexBuild(OperationContext* opCtx, StringData indexName) const final;
-
-    long getIndexBuildVersion(OperationContext* opCtx, StringData indexName) const final;
-
-    void setIndexBuildScanning(OperationContext* opCtx,
-                               StringData indexName,
-                               std::string sideWritesIdent,
-                               boost::optional<std::string> constraintViolationsIdent) final;
-
-    bool isIndexBuildScanning(OperationContext* opCtx, StringData indexName) const final;
-
-    void setIndexBuildDraining(OperationContext* opCtx, StringData indexName) final;
-
-    bool isIndexBuildDraining(OperationContext* opCtx, StringData indexName) const final;
-
-    void indexBuildSuccess(OperationContext* opCtx, StringData indexName) final;
-
-    boost::optional<std::string> getSideWritesIdent(OperationContext* opCtx,
-                                                    StringData indexName) const final;
-
-    boost::optional<std::string> getConstraintViolationsIdent(OperationContext* opCtx,
-                                                              StringData indexName) const final;
-    void updateTTLSetting(OperationContext* opCtx,
-                          StringData idxName,
-                          long long newExpireSeconds) final;
-
-    void updateIndexMetadata(OperationContext* opCtx, const IndexDescriptor* desc) final;
-
-    void updateValidator(OperationContext* opCtx,
-                         const BSONObj& validator,
-                         StringData validationLevel,
-                         StringData validationAction) final;
-
-    void setIsTemp(OperationContext* opCtx, bool isTemp);
-
-    void updateCappedSize(OperationContext*, long long int) final;
-
-    bool isEqualToMetadataUUID(OperationContext* opCtx, OptionalCollectionUUID uuid) final;
+    ~KVCollectionCatalogEntry() final {}
 
     RecordStore* getRecordStore() {
         return _recordStore.get();
@@ -116,15 +56,8 @@ public:
         return _recordStore.get();
     }
 
-protected:
-    MetaData _getMetaData(OperationContext* opCtx) const final;
-
 private:
-    class AddIndexChange;
-    class RemoveIndexChange;
-
     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 14b70f282fc..c521b880e3e 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
@@ -96,9 +96,12 @@ public:
         }
     }
 
-    CollectionCatalogEntry* getCollectionCatalogEntry() {
-        return CollectionCatalog::get(getGlobalServiceContext())
-            .lookupCollectionCatalogEntryByNamespace(_nss);
+    NamespaceString ns() {
+        return _nss;
+    }
+
+    DurableCatalog* getCatalog() {
+        return _storageEngine.getCatalog();
     }
 
     std::string createIndex(BSONObj keyPattern,
@@ -116,8 +119,8 @@ public:
         {
             WriteUnitOfWork wuow(opCtx.get());
             const bool isSecondaryBackgroundIndexBuild = false;
-            ASSERT_OK(getCollectionCatalogEntry()->prepareForIndexBuild(
-                opCtx.get(), &desc, protocol, isSecondaryBackgroundIndexBuild));
+            ASSERT_OK(_storageEngine.getCatalog()->prepareForIndexBuild(
+                opCtx.get(), _nss, &desc, protocol, isSecondaryBackgroundIndexBuild));
             wuow.commit();
         }
 
@@ -159,106 +162,100 @@ private:
 
 TEST_F(KVCollectionCatalogEntryTest, MultikeyPathsForBtreeIndexInitializedToVectorOfEmptySets) {
     std::string indexName = createIndex(BSON("a" << 1 << "b" << 1));
-    CollectionCatalogEntry* collEntry = getCollectionCatalogEntry();
-
     auto opCtx = newOperationContext();
+    DurableCatalog* catalog = getCatalog();
     {
         MultikeyPaths multikeyPaths;
-        ASSERT(!collEntry->isIndexMultikey(opCtx.get(), indexName, &multikeyPaths));
+        ASSERT(!catalog->isIndexMultikey(opCtx.get(), ns(), indexName, &multikeyPaths));
         assertMultikeyPathsAreEqual(multikeyPaths, {std::set<size_t>{}, std::set<size_t>{}});
     }
 }
 
 TEST_F(KVCollectionCatalogEntryTest, CanSetIndividualPathComponentOfBtreeIndexAsMultikey) {
     std::string indexName = createIndex(BSON("a" << 1 << "b" << 1));
-    CollectionCatalogEntry* collEntry = getCollectionCatalogEntry();
-
     auto opCtx = newOperationContext();
-    ASSERT(collEntry->setIndexIsMultikey(opCtx.get(), indexName, {std::set<size_t>{}, {0U}}));
+    DurableCatalog* catalog = getCatalog();
+    ASSERT(catalog->setIndexIsMultikey(opCtx.get(), ns(), indexName, {std::set<size_t>{}, {0U}}));
 
     {
         MultikeyPaths multikeyPaths;
-        ASSERT(collEntry->isIndexMultikey(opCtx.get(), indexName, &multikeyPaths));
+        ASSERT(catalog->isIndexMultikey(opCtx.get(), ns(), indexName, &multikeyPaths));
         assertMultikeyPathsAreEqual(multikeyPaths, {std::set<size_t>{}, {0U}});
     }
 }
 
 TEST_F(KVCollectionCatalogEntryTest, MultikeyPathsAccumulateOnDifferentFields) {
     std::string indexName = createIndex(BSON("a" << 1 << "b" << 1));
-    CollectionCatalogEntry* collEntry = getCollectionCatalogEntry();
-
     auto opCtx = newOperationContext();
-    ASSERT(collEntry->setIndexIsMultikey(opCtx.get(), indexName, {std::set<size_t>{}, {0U}}));
+    DurableCatalog* catalog = getCatalog();
+    ASSERT(catalog->setIndexIsMultikey(opCtx.get(), ns(), indexName, {std::set<size_t>{}, {0U}}));
 
     {
         MultikeyPaths multikeyPaths;
-        ASSERT(collEntry->isIndexMultikey(opCtx.get(), indexName, &multikeyPaths));
+        ASSERT(catalog->isIndexMultikey(opCtx.get(), ns(), indexName, &multikeyPaths));
         assertMultikeyPathsAreEqual(multikeyPaths, {std::set<size_t>{}, {0U}});
     }
 
-    ASSERT(collEntry->setIndexIsMultikey(opCtx.get(), indexName, {{0U}, std::set<size_t>{}}));
+    ASSERT(catalog->setIndexIsMultikey(opCtx.get(), ns(), indexName, {{0U}, std::set<size_t>{}}));
 
     {
         MultikeyPaths multikeyPaths;
-        ASSERT(collEntry->isIndexMultikey(opCtx.get(), indexName, &multikeyPaths));
+        ASSERT(catalog->isIndexMultikey(opCtx.get(), ns(), indexName, &multikeyPaths));
         assertMultikeyPathsAreEqual(multikeyPaths, {{0U}, {0U}});
     }
 }
 
 TEST_F(KVCollectionCatalogEntryTest, MultikeyPathsAccumulateOnDifferentComponentsOfTheSameField) {
     std::string indexName = createIndex(BSON("a.b" << 1));
-    CollectionCatalogEntry* collEntry = getCollectionCatalogEntry();
-
     auto opCtx = newOperationContext();
-    ASSERT(collEntry->setIndexIsMultikey(opCtx.get(), indexName, {{0U}}));
+    DurableCatalog* catalog = getCatalog();
+    ASSERT(catalog->setIndexIsMultikey(opCtx.get(), ns(), indexName, {{0U}}));
 
     {
         MultikeyPaths multikeyPaths;
-        ASSERT(collEntry->isIndexMultikey(opCtx.get(), indexName, &multikeyPaths));
+        ASSERT(catalog->isIndexMultikey(opCtx.get(), ns(), indexName, &multikeyPaths));
         assertMultikeyPathsAreEqual(multikeyPaths, {{0U}});
     }
 
-    ASSERT(collEntry->setIndexIsMultikey(opCtx.get(), indexName, {{1U}}));
+    ASSERT(catalog->setIndexIsMultikey(opCtx.get(), ns(), indexName, {{1U}}));
 
     {
         MultikeyPaths multikeyPaths;
-        ASSERT(collEntry->isIndexMultikey(opCtx.get(), indexName, &multikeyPaths));
+        ASSERT(catalog->isIndexMultikey(opCtx.get(), ns(), indexName, &multikeyPaths));
         assertMultikeyPathsAreEqual(multikeyPaths, {{0U, 1U}});
     }
 }
 
 TEST_F(KVCollectionCatalogEntryTest, NoOpWhenSpecifiedPathComponentsAlreadySetAsMultikey) {
     std::string indexName = createIndex(BSON("a" << 1));
-    CollectionCatalogEntry* collEntry = getCollectionCatalogEntry();
-
     auto opCtx = newOperationContext();
-    ASSERT(collEntry->setIndexIsMultikey(opCtx.get(), indexName, {{0U}}));
+    DurableCatalog* catalog = getCatalog();
+    ASSERT(catalog->setIndexIsMultikey(opCtx.get(), ns(), indexName, {{0U}}));
 
     {
         MultikeyPaths multikeyPaths;
-        ASSERT(collEntry->isIndexMultikey(opCtx.get(), indexName, &multikeyPaths));
+        ASSERT(catalog->isIndexMultikey(opCtx.get(), ns(), indexName, &multikeyPaths));
         assertMultikeyPathsAreEqual(multikeyPaths, {{0U}});
     }
 
-    ASSERT(!collEntry->setIndexIsMultikey(opCtx.get(), indexName, {{0U}}));
+    ASSERT(!catalog->setIndexIsMultikey(opCtx.get(), ns(), indexName, {{0U}}));
 
     {
         MultikeyPaths multikeyPaths;
-        ASSERT(collEntry->isIndexMultikey(opCtx.get(), indexName, &multikeyPaths));
+        ASSERT(catalog->isIndexMultikey(opCtx.get(), ns(), indexName, &multikeyPaths));
         assertMultikeyPathsAreEqual(multikeyPaths, {{0U}});
     }
 }
 
 TEST_F(KVCollectionCatalogEntryTest, CanSetMultipleFieldsAndComponentsAsMultikey) {
     std::string indexName = createIndex(BSON("a.b.c" << 1 << "a.b.d" << 1));
-    CollectionCatalogEntry* collEntry = getCollectionCatalogEntry();
-
     auto opCtx = newOperationContext();
-    ASSERT(collEntry->setIndexIsMultikey(opCtx.get(), indexName, {{0U, 1U}, {0U, 1U}}));
+    DurableCatalog* catalog = getCatalog();
+    ASSERT(catalog->setIndexIsMultikey(opCtx.get(), ns(), indexName, {{0U, 1U}, {0U, 1U}}));
 
     {
         MultikeyPaths multikeyPaths;
-        ASSERT(collEntry->isIndexMultikey(opCtx.get(), indexName, &multikeyPaths));
+        ASSERT(catalog->isIndexMultikey(opCtx.get(), ns(), indexName, &multikeyPaths));
         assertMultikeyPathsAreEqual(multikeyPaths, {{0U, 1U}, {0U, 1U}});
     }
 }
@@ -267,31 +264,29 @@ DEATH_TEST_F(KVCollectionCatalogEntryTest,
              CannotOmitPathLevelMultikeyInfoWithBtreeIndex,
              "Invariant failure !multikeyPaths.empty()") {
     std::string indexName = createIndex(BSON("a" << 1 << "b" << 1));
-    CollectionCatalogEntry* collEntry = getCollectionCatalogEntry();
-
     auto opCtx = newOperationContext();
-    collEntry->setIndexIsMultikey(opCtx.get(), indexName, MultikeyPaths{});
+    DurableCatalog* catalog = getCatalog();
+    catalog->setIndexIsMultikey(opCtx.get(), ns(), indexName, MultikeyPaths{});
 }
 
 DEATH_TEST_F(KVCollectionCatalogEntryTest,
              AtLeastOnePathComponentMustCauseIndexToBeMultikey,
              "Invariant failure somePathIsMultikey") {
     std::string indexName = createIndex(BSON("a" << 1 << "b" << 1));
-    CollectionCatalogEntry* collEntry = getCollectionCatalogEntry();
-
     auto opCtx = newOperationContext();
-    collEntry->setIndexIsMultikey(opCtx.get(), indexName, {std::set<size_t>{}, std::set<size_t>{}});
+    DurableCatalog* catalog = getCatalog();
+    catalog->setIndexIsMultikey(
+        opCtx.get(), ns(), indexName, {std::set<size_t>{}, std::set<size_t>{}});
 }
 
 TEST_F(KVCollectionCatalogEntryTest, PathLevelMultikeyTrackingIsSupportedBy2dsphereIndexes) {
     std::string indexType = IndexNames::GEO_2DSPHERE;
     std::string indexName = createIndex(BSON("a" << indexType << "b" << 1), indexType);
-    CollectionCatalogEntry* collEntry = getCollectionCatalogEntry();
-
     auto opCtx = newOperationContext();
+    DurableCatalog* catalog = getCatalog();
     {
         MultikeyPaths multikeyPaths;
-        ASSERT(!collEntry->isIndexMultikey(opCtx.get(), indexName, &multikeyPaths));
+        ASSERT(!catalog->isIndexMultikey(opCtx.get(), ns(), indexName, &multikeyPaths));
         assertMultikeyPathsAreEqual(multikeyPaths, {std::set<size_t>{}, std::set<size_t>{}});
     }
 }
@@ -302,12 +297,11 @@ TEST_F(KVCollectionCatalogEntryTest, PathLevelMultikeyTrackingIsNotSupportedByAl
 
     for (auto&& indexType : indexTypes) {
         std::string indexName = createIndex(BSON("a" << indexType << "b" << 1), indexType);
-        CollectionCatalogEntry* collEntry = getCollectionCatalogEntry();
-
         auto opCtx = newOperationContext();
+        DurableCatalog* catalog = getCatalog();
         {
             MultikeyPaths multikeyPaths;
-            ASSERT(!collEntry->isIndexMultikey(opCtx.get(), indexName, &multikeyPaths));
+            ASSERT(!catalog->isIndexMultikey(opCtx.get(), ns(), indexName, &multikeyPaths));
             ASSERT(multikeyPaths.empty());
         }
     }
@@ -316,14 +310,13 @@ TEST_F(KVCollectionCatalogEntryTest, PathLevelMultikeyTrackingIsNotSupportedByAl
 TEST_F(KVCollectionCatalogEntryTest, CanSetEntireTextIndexAsMultikey) {
     std::string indexType = IndexNames::TEXT;
     std::string indexName = createIndex(BSON("a" << indexType << "b" << 1), indexType);
-    CollectionCatalogEntry* collEntry = getCollectionCatalogEntry();
-
     auto opCtx = newOperationContext();
-    ASSERT(collEntry->setIndexIsMultikey(opCtx.get(), indexName, MultikeyPaths{}));
+    DurableCatalog* catalog = getCatalog();
+    ASSERT(catalog->setIndexIsMultikey(opCtx.get(), ns(), indexName, MultikeyPaths{}));
 
     {
         MultikeyPaths multikeyPaths;
-        ASSERT(collEntry->isIndexMultikey(opCtx.get(), indexName, &multikeyPaths));
+        ASSERT(catalog->isIndexMultikey(opCtx.get(), ns(), indexName, &multikeyPaths));
         ASSERT(multikeyPaths.empty());
     }
 }
@@ -331,119 +324,115 @@ TEST_F(KVCollectionCatalogEntryTest, CanSetEntireTextIndexAsMultikey) {
 TEST_F(KVCollectionCatalogEntryTest, NoOpWhenEntireIndexAlreadySetAsMultikey) {
     std::string indexType = IndexNames::TEXT;
     std::string indexName = createIndex(BSON("a" << indexType << "b" << 1), indexType);
-    CollectionCatalogEntry* collEntry = getCollectionCatalogEntry();
-
     auto opCtx = newOperationContext();
-    ASSERT(collEntry->setIndexIsMultikey(opCtx.get(), indexName, MultikeyPaths{}));
+    DurableCatalog* catalog = getCatalog();
+    ASSERT(catalog->setIndexIsMultikey(opCtx.get(), ns(), indexName, MultikeyPaths{}));
 
     {
         MultikeyPaths multikeyPaths;
-        ASSERT(collEntry->isIndexMultikey(opCtx.get(), indexName, &multikeyPaths));
+        ASSERT(catalog->isIndexMultikey(opCtx.get(), ns(), indexName, &multikeyPaths));
         ASSERT(multikeyPaths.empty());
     }
 
-    ASSERT(!collEntry->setIndexIsMultikey(opCtx.get(), indexName, MultikeyPaths{}));
+    ASSERT(!catalog->setIndexIsMultikey(opCtx.get(), ns(), indexName, MultikeyPaths{}));
 
     {
         MultikeyPaths multikeyPaths;
-        ASSERT(collEntry->isIndexMultikey(opCtx.get(), indexName, &multikeyPaths));
+        ASSERT(catalog->isIndexMultikey(opCtx.get(), ns(), indexName, &multikeyPaths));
         ASSERT(multikeyPaths.empty());
     }
 }
 
 TEST_F(KVCollectionCatalogEntryTest, SinglePhaseIndexBuild) {
     std::string indexName = createIndex(BSON("a" << 1));
-    CollectionCatalogEntry* collEntry = getCollectionCatalogEntry();
-
     auto opCtx = newOperationContext();
-
-    ASSERT_EQ(kExpectedVersion, collEntry->getIndexBuildVersion(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->isIndexReady(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->isTwoPhaseIndexBuild(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->isIndexBuildScanning(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->isIndexBuildDraining(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->getSideWritesIdent(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->getConstraintViolationsIdent(opCtx.get(), indexName));
-
-    collEntry->indexBuildSuccess(opCtx.get(), indexName);
-
-    ASSERT_EQ(kExpectedVersion, collEntry->getIndexBuildVersion(opCtx.get(), indexName));
-    ASSERT_TRUE(collEntry->isIndexReady(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->isTwoPhaseIndexBuild(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->isIndexBuildScanning(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->isIndexBuildDraining(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->getSideWritesIdent(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->getConstraintViolationsIdent(opCtx.get(), indexName));
+    DurableCatalog* catalog = getCatalog();
+
+    ASSERT_EQ(kExpectedVersion, catalog->getIndexBuildVersion(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->isIndexReady(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->isTwoPhaseIndexBuild(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->isIndexBuildScanning(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->isIndexBuildDraining(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->getSideWritesIdent(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->getConstraintViolationsIdent(opCtx.get(), ns(), indexName));
+
+    catalog->indexBuildSuccess(opCtx.get(), ns(), indexName);
+
+    ASSERT_EQ(kExpectedVersion, catalog->getIndexBuildVersion(opCtx.get(), ns(), indexName));
+    ASSERT_TRUE(catalog->isIndexReady(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->isTwoPhaseIndexBuild(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->isIndexBuildScanning(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->isIndexBuildDraining(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->getSideWritesIdent(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->getConstraintViolationsIdent(opCtx.get(), ns(), indexName));
 }
 
 TEST_F(KVCollectionCatalogEntryTest, TwoPhaseIndexBuild) {
     std::string indexName =
         createIndex(BSON("a" << 1), IndexNames::BTREE, IndexBuildProtocol::kTwoPhase);
-    CollectionCatalogEntry* collEntry = getCollectionCatalogEntry();
-
     auto opCtx = newOperationContext();
-
-    ASSERT_EQ(kExpectedVersion, collEntry->getIndexBuildVersion(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->isIndexReady(opCtx.get(), indexName));
-    ASSERT_TRUE(collEntry->isTwoPhaseIndexBuild(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->isIndexBuildScanning(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->isIndexBuildDraining(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->getSideWritesIdent(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->getConstraintViolationsIdent(opCtx.get(), indexName));
-
-    collEntry->setIndexBuildScanning(
-        opCtx.get(), indexName, kSideWritesTableIdent, kConstraintViolationsTableIdent);
-
-    ASSERT_EQ(kExpectedVersion, collEntry->getIndexBuildVersion(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->isIndexReady(opCtx.get(), indexName));
-    ASSERT_TRUE(collEntry->isTwoPhaseIndexBuild(opCtx.get(), indexName));
-    ASSERT_TRUE(collEntry->isIndexBuildScanning(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->isIndexBuildDraining(opCtx.get(), indexName));
-    ASSERT_EQ(kSideWritesTableIdent, collEntry->getSideWritesIdent(opCtx.get(), indexName));
+    DurableCatalog* catalog = getCatalog();
+
+    ASSERT_EQ(kExpectedVersion, catalog->getIndexBuildVersion(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->isIndexReady(opCtx.get(), ns(), indexName));
+    ASSERT_TRUE(catalog->isTwoPhaseIndexBuild(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->isIndexBuildScanning(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->isIndexBuildDraining(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->getSideWritesIdent(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->getConstraintViolationsIdent(opCtx.get(), ns(), indexName));
+
+    catalog->setIndexBuildScanning(
+        opCtx.get(), ns(), indexName, kSideWritesTableIdent, kConstraintViolationsTableIdent);
+
+    ASSERT_EQ(kExpectedVersion, catalog->getIndexBuildVersion(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->isIndexReady(opCtx.get(), ns(), indexName));
+    ASSERT_TRUE(catalog->isTwoPhaseIndexBuild(opCtx.get(), ns(), indexName));
+    ASSERT_TRUE(catalog->isIndexBuildScanning(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->isIndexBuildDraining(opCtx.get(), ns(), indexName));
+    ASSERT_EQ(kSideWritesTableIdent, catalog->getSideWritesIdent(opCtx.get(), ns(), indexName));
     ASSERT_EQ(kConstraintViolationsTableIdent,
-              collEntry->getConstraintViolationsIdent(opCtx.get(), indexName));
+              catalog->getConstraintViolationsIdent(opCtx.get(), ns(), indexName));
 
-    collEntry->setIndexBuildDraining(opCtx.get(), indexName);
+    catalog->setIndexBuildDraining(opCtx.get(), ns(), indexName);
 
-    ASSERT_EQ(kExpectedVersion, collEntry->getIndexBuildVersion(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->isIndexReady(opCtx.get(), indexName));
-    ASSERT_TRUE(collEntry->isTwoPhaseIndexBuild(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->isIndexBuildScanning(opCtx.get(), indexName));
-    ASSERT_TRUE(collEntry->isIndexBuildDraining(opCtx.get(), indexName));
-    ASSERT_EQ(kSideWritesTableIdent, collEntry->getSideWritesIdent(opCtx.get(), indexName));
+    ASSERT_EQ(kExpectedVersion, catalog->getIndexBuildVersion(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->isIndexReady(opCtx.get(), ns(), indexName));
+    ASSERT_TRUE(catalog->isTwoPhaseIndexBuild(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->isIndexBuildScanning(opCtx.get(), ns(), indexName));
+    ASSERT_TRUE(catalog->isIndexBuildDraining(opCtx.get(), ns(), indexName));
+    ASSERT_EQ(kSideWritesTableIdent, catalog->getSideWritesIdent(opCtx.get(), ns(), indexName));
     ASSERT_EQ(kConstraintViolationsTableIdent,
-              collEntry->getConstraintViolationsIdent(opCtx.get(), indexName));
+              catalog->getConstraintViolationsIdent(opCtx.get(), ns(), indexName));
 
-    collEntry->indexBuildSuccess(opCtx.get(), indexName);
+    catalog->indexBuildSuccess(opCtx.get(), ns(), indexName);
 
-    ASSERT_EQ(kExpectedVersion, collEntry->getIndexBuildVersion(opCtx.get(), indexName));
-    ASSERT(collEntry->isIndexReady(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->isIndexBuildScanning(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->isIndexBuildDraining(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->isTwoPhaseIndexBuild(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->getSideWritesIdent(opCtx.get(), indexName));
-    ASSERT_FALSE(collEntry->getConstraintViolationsIdent(opCtx.get(), indexName));
+    ASSERT_EQ(kExpectedVersion, catalog->getIndexBuildVersion(opCtx.get(), ns(), indexName));
+    ASSERT(catalog->isIndexReady(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->isIndexBuildScanning(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->isIndexBuildDraining(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->isTwoPhaseIndexBuild(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->getSideWritesIdent(opCtx.get(), ns(), indexName));
+    ASSERT_FALSE(catalog->getConstraintViolationsIdent(opCtx.get(), ns(), indexName));
 }
 
 DEATH_TEST_F(KVCollectionCatalogEntryTest,
              SinglePhaseIllegalScanPhase,
              "Invariant failure md.indexes[offset].runTwoPhaseBuild") {
     std::string indexName = createIndex(BSON("a" << 1));
-    CollectionCatalogEntry* collEntry = getCollectionCatalogEntry();
-
     auto opCtx = newOperationContext();
-    collEntry->setIndexBuildScanning(
-        opCtx.get(), indexName, kSideWritesTableIdent, kConstraintViolationsTableIdent);
+    DurableCatalog* catalog = getCatalog();
+
+    catalog->setIndexBuildScanning(
+        opCtx.get(), ns(), indexName, kSideWritesTableIdent, kConstraintViolationsTableIdent);
 }
 
 DEATH_TEST_F(KVCollectionCatalogEntryTest,
              SinglePhaseIllegalDrainPhase,
              "Invariant failure md.indexes[offset].runTwoPhaseBuild") {
     std::string indexName = createIndex(BSON("a" << 1));
-    CollectionCatalogEntry* collEntry = getCollectionCatalogEntry();
-
     auto opCtx = newOperationContext();
-    collEntry->setIndexBuildDraining(opCtx.get(), indexName);
+    DurableCatalog* catalog = getCatalog();
+    catalog->setIndexBuildDraining(opCtx.get(), ns(), indexName);
 }
 
 DEATH_TEST_F(KVCollectionCatalogEntryTest,
@@ -451,10 +440,9 @@ DEATH_TEST_F(KVCollectionCatalogEntryTest,
              "Invariant failure multikeyPaths.empty()") {
     std::string indexType = IndexNames::TEXT;
     std::string indexName = createIndex(BSON("a" << indexType << "b" << 1), indexType);
-    CollectionCatalogEntry* collEntry = getCollectionCatalogEntry();
-
     auto opCtx = newOperationContext();
-    collEntry->setIndexIsMultikey(opCtx.get(), indexName, {{0U}, {0U}});
+    DurableCatalog* catalog = getCatalog();
+    catalog->setIndexIsMultikey(opCtx.get(), ns(), indexName, {{0U}, {0U}});
 }
 
 }  // namespace
diff --git a/src/mongo/db/storage/kv/kv_drop_pending_ident_reaper_test.cpp b/src/mongo/db/storage/kv/kv_drop_pending_ident_reaper_test.cpp
index 6cbb5572554..309d5825266 100644
--- a/src/mongo/db/storage/kv/kv_drop_pending_ident_reaper_test.cpp
+++ b/src/mongo/db/storage/kv/kv_drop_pending_ident_reaper_test.cpp
@@ -88,6 +88,7 @@ public:
         return {};
     }
     Status createSortedDataInterface(OperationContext* opCtx,
+                                     const CollectionOptions& collOptions,
                                      StringData ident,
                                      const IndexDescriptor* desc) override {
         return Status::OK();
diff --git a/src/mongo/db/storage/kv/kv_engine.h b/src/mongo/db/storage/kv/kv_engine.h
index b9e65bfd24f..2c0e56a0c33 100644
--- a/src/mongo/db/storage/kv/kv_engine.h
+++ b/src/mongo/db/storage/kv/kv_engine.h
@@ -144,6 +144,7 @@ public:
     }
 
     virtual Status createSortedDataInterface(OperationContext* opCtx,
+                                             const CollectionOptions& collOptions,
                                              StringData ident,
                                              const IndexDescriptor* desc) = 0;
 
@@ -158,11 +159,12 @@ public:
      *        share a table. Sharing indexes belonging to different databases is forbidden.
      */
     virtual Status createGroupedSortedDataInterface(OperationContext* opCtx,
+                                                    const CollectionOptions& collOptions,
                                                     StringData ident,
                                                     const IndexDescriptor* desc,
                                                     KVPrefix prefix) {
         invariant(prefix == KVPrefix::kNotPrefixed);
-        return createSortedDataInterface(opCtx, ident, desc);
+        return createSortedDataInterface(opCtx, collOptions, ident, desc);
     }
 
     virtual int64_t getIdentSize(OperationContext* opCtx, StringData ident) = 0;
diff --git a/src/mongo/db/storage/kv/kv_engine_test_harness.cpp b/src/mongo/db/storage/kv/kv_engine_test_harness.cpp
index 72935625b70..14c91e85938 100644
--- a/src/mongo/db/storage/kv/kv_engine_test_harness.cpp
+++ b/src/mongo/db/storage/kv/kv_engine_test_harness.cpp
@@ -33,7 +33,6 @@
 #include "mongo/db/catalog/collection_impl.h"
 #include "mongo/db/index/index_descriptor.h"
 #include "mongo/db/operation_context_noop.h"
-#include "mongo/db/storage/kv/kv_catalog.h"
 #include "mongo/db/storage/kv/kv_catalog_test_fixture.h"
 #include "mongo/db/storage/kv/kv_engine.h"
 #include "mongo/db/storage/kv/kv_prefix.h"
@@ -181,7 +180,7 @@ TEST(KVEngineTestHarness, SimpleSorted1) {
     unique_ptr<SortedDataInterface> sorted;
     {
         MyOperationContext opCtx(engine);
-        ASSERT_OK(engine->createSortedDataInterface(&opCtx, ident, &desc));
+        ASSERT_OK(engine->createSortedDataInterface(&opCtx, CollectionOptions(), ident, &desc));
         sorted.reset(engine->getSortedDataInterface(&opCtx, ident, &desc));
         ASSERT(sorted);
     }
@@ -724,7 +723,7 @@ DEATH_TEST_F(KVCatalogTest, TerminateOnNonNumericIndexVersion, "Fatal Assertion
     unique_ptr<SortedDataInterface> sorted;
     {
         MyOperationContext opCtx(engine);
-        ASSERT_OK(engine->createSortedDataInterface(&opCtx, ident, &desc));
+        ASSERT_OK(engine->createSortedDataInterface(&opCtx, CollectionOptions(), ident, &desc));
         sorted.reset(engine->getSortedDataInterface(&opCtx, ident, &desc));
         ASSERT(sorted);
     }
diff --git a/src/mongo/db/storage/kv/storage_engine_impl.cpp b/src/mongo/db/storage/kv/storage_engine_impl.cpp
index 3aa39b1c123..cd23a9208bb 100644
--- a/src/mongo/db/storage/kv/storage_engine_impl.cpp
+++ b/src/mongo/db/storage/kv/storage_engine_impl.cpp
@@ -92,7 +92,7 @@ void StorageEngineImpl::loadCatalog(OperationContext* opCtx) {
 
         if (status.code() == ErrorCodes::DataModifiedByRepair) {
             warning() << "Catalog data modified by repair: " << status.reason();
-            repairObserver->onModification(str::stream() << "KVCatalog repaired: "
+            repairObserver->onModification(str::stream() << "DurableCatalog repaired: "
                                                          << status.reason());
         } else {
             fassertNoTrace(50926, status);
@@ -239,10 +239,10 @@ void StorageEngineImpl::_initCollection(OperationContext* opCtx,
                                         const NamespaceString& nss,
                                         bool forRepair) {
     auto catalogEntry = _catalog->makeCollectionCatalogEntry(opCtx, nss, forRepair);
-    auto uuid = catalogEntry->getCollectionOptions(opCtx).uuid.get();
+    auto uuid = _catalog->getCollectionOptions(opCtx, nss).uuid.get();
 
     auto collectionFactory = Collection::Factory::get(getGlobalServiceContext());
-    auto collection = collectionFactory->make(opCtx, catalogEntry.get());
+    auto collection = collectionFactory->make(opCtx, uuid, catalogEntry.get());
 
     auto& collectionCatalog = CollectionCatalog::get(getGlobalServiceContext());
     collectionCatalog.registerCollection(uuid, std::move(catalogEntry), std::move(collection));
@@ -291,15 +291,15 @@ Status StorageEngineImpl::_recoverOrphanedCollection(OperationContext* opCtx,
 
 /**
  * This method reconciles differences between idents the KVEngine is aware of and the
- * KVCatalog. There are three differences to consider:
+ * DurableCatalog. There are three differences to consider:
  *
- * First, a KVEngine may know of an ident that the KVCatalog does not. This method will drop
+ * First, a KVEngine may know of an ident that the DurableCatalog does not. This method will drop
  * the ident from the KVEngine.
  *
- * Second, a KVCatalog may have a collection ident that the KVEngine does not. This is an
+ * Second, a DurableCatalog may have a collection ident that the KVEngine does not. This is an
  * illegal state and this method fasserts.
  *
- * Third, a KVCatalog may have an index ident that the KVEngine does not. This method will
+ * Third, a DurableCatalog may have an index ident that the KVEngine does not. This method will
  * rebuild the index.
  */
 StatusWith<std::vector<StorageEngine::CollectionIndexNamePair>>
@@ -964,7 +964,7 @@ int64_t StorageEngineImpl::sizeOnDiskForDb(OperationContext* opCtx, StringData d
             size += catalogEntry->getRecordStore()->storageSize(opCtx);
 
             std::vector<std::string> indexNames;
-            catalogEntry->getAllIndexes(opCtx, &indexNames);
+            _catalog->getAllIndexes(opCtx, collection->ns(), &indexNames);
 
             for (size_t i = 0; i < indexNames.size(); i++) {
                 std::string ident =
diff --git a/src/mongo/db/storage/kv/storage_engine_impl.h b/src/mongo/db/storage/kv/storage_engine_impl.h
index 81a2ad3f476..c8084559b80 100644
--- a/src/mongo/db/storage/kv/storage_engine_impl.h
+++ b/src/mongo/db/storage/kv/storage_engine_impl.h
@@ -36,8 +36,9 @@
 #include "mongo/base/string_data.h"
 #include "mongo/bson/timestamp.h"
 #include "mongo/db/namespace_string.h"
+#include "mongo/db/storage/durable_catalog.h"
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/db/storage/journal_listener.h"
-#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/storage_engine_interface.h"
@@ -307,10 +308,10 @@ public:
                              const NamespaceString& nss,
                              StringData ident) override;
 
-    KVCatalog* getCatalog() {
+    DurableCatalog* getCatalog() {
         return _catalog.get();
     }
-    const KVCatalog* getCatalog() const {
+    const DurableCatalog* getCatalog() const {
         return _catalog.get();
     }
 
diff --git a/src/mongo/db/storage/kv/storage_engine_interface.h b/src/mongo/db/storage/kv/storage_engine_interface.h
index 89ab71b83b2..db7201c1492 100644
--- a/src/mongo/db/storage/kv/storage_engine_interface.h
+++ b/src/mongo/db/storage/kv/storage_engine_interface.h
@@ -32,7 +32,7 @@
 namespace mongo {
 
 class KVEngine;
-class KVCatalog;
+class DurableCatalog;
 class StorageEngine;
 
 class StorageEngineInterface {
@@ -44,6 +44,6 @@ public:
     virtual void addDropPendingIdent(const Timestamp& dropTimestamp,
                                      const NamespaceString& nss,
                                      StringData ident) = 0;
-    virtual KVCatalog* getCatalog() = 0;
+    virtual DurableCatalog* getCatalog() = 0;
 };
 }
diff --git a/src/mongo/db/storage/kv/storage_engine_test.cpp b/src/mongo/db/storage/kv/storage_engine_test.cpp
index 5465a5ff31f..d80d9da9043 100644
--- a/src/mongo/db/storage/kv/storage_engine_test.cpp
+++ b/src/mongo/db/storage/kv/storage_engine_test.cpp
@@ -41,8 +41,8 @@
 #include "mongo/db/repl/replication_coordinator_mock.h"
 #include "mongo/db/service_context_d_test_fixture.h"
 #include "mongo/db/storage/devnull/devnull_kv_engine.h"
+#include "mongo/db/storage/durable_catalog.h"
 #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/storage_engine_impl.h"
 #include "mongo/db/storage/kv/storage_engine_test_fixture.h"
@@ -59,12 +59,12 @@ namespace {
 TEST_F(StorageEngineTest, ReconcileIdentsTest) {
     auto opCtx = cc().makeOperationContext();
 
-    // Add a collection, `db.coll1` to both the KVCatalog and KVEngine. The returned value is the
-    // `ident` name given to the collection.
+    // Add a collection, `db.coll1` to both the DurableCatalog and KVEngine. The returned value is
+    // the `ident` name given to the collection.
     auto swIdentName = createCollection(opCtx.get(), NamespaceString("db.coll1"));
     ASSERT_OK(swIdentName);
-    // Create a table in the KVEngine not reflected in the KVCatalog. This should be dropped when
-    // reconciling.
+    // Create a table in the KVEngine not reflected in the DurableCatalog. This should be dropped
+    // when reconciling.
     ASSERT_OK(createCollTable(opCtx.get(), NamespaceString("db.coll2")));
     ASSERT_OK(reconcile(opCtx.get()).getStatus());
     auto identsVec = getAllKVEngineIdents(opCtx.get());
@@ -86,7 +86,7 @@ TEST_F(StorageEngineTest, ReconcileIdentsTest) {
     ASSERT_EQUALS("db.coll1", toRebuild.first);
     ASSERT_EQUALS("_id", toRebuild.second);
 
-    // Now drop the `db.coll1` table, while leaving the KVCatalog entry.
+    // Now drop the `db.coll1` table, while leaving the DurableCatalog entry.
     ASSERT_OK(dropIdent(opCtx.get(), swIdentName.getValue()));
     ASSERT_EQUALS(static_cast<const unsigned long>(1), getAllKVEngineIdents(opCtx.get()).size());
 
diff --git a/src/mongo/db/storage/kv/storage_engine_test_fixture.h b/src/mongo/db/storage/kv/storage_engine_test_fixture.h
index 7f3bddb4bc8..7c588c991cd 100644
--- a/src/mongo/db/storage/kv/storage_engine_test_fixture.h
+++ b/src/mongo/db/storage/kv/storage_engine_test_fixture.h
@@ -33,7 +33,7 @@
 #include "mongo/db/catalog/collection_mock.h"
 #include "mongo/db/catalog_raii.h"
 #include "mongo/db/service_context_d_test_fixture.h"
-#include "mongo/db/storage/kv/kv_catalog.h"
+#include "mongo/db/storage/durable_catalog.h"
 #include "mongo/db/storage/kv/kv_engine.h"
 #include "mongo/db/storage/kv/storage_engine_impl.h"
 #include "mongo/db/storage/storage_repair_observer.h"
@@ -68,7 +68,7 @@ public:
     }
 
     /**
-     * Create a collection table in the KVEngine not reflected in the KVCatalog.
+     * Create a collection table in the KVEngine not reflected in the DurableCatalog.
      */
     Status createCollTable(OperationContext* opCtx, NamespaceString collName) {
         const std::string identName = "collection-" + collName.ns();
@@ -134,11 +134,9 @@ public:
         auto descriptor = std::make_unique<IndexDescriptor>(
             collection.get(), IndexNames::findPluginName(spec), spec);
 
-        CollectionCatalogEntry* cce =
-            CollectionCatalog::get(opCtx).lookupCollectionCatalogEntryByNamespace(collNs);
         const auto protocol = IndexBuildProtocol::kTwoPhase;
-        auto ret = cce->prepareForIndexBuild(
-            opCtx, descriptor.get(), protocol, isBackgroundSecondaryBuild);
+        auto ret = DurableCatalog::get(opCtx)->prepareForIndexBuild(
+            opCtx, collNs, descriptor.get(), protocol, isBackgroundSecondaryBuild);
         return ret;
     }
 
@@ -147,25 +145,20 @@ public:
                         std::string key,
                         std::string sideWritesIdent,
                         std::string constraintViolationsIdent) {
-        CollectionCatalogEntry* cce =
-            CollectionCatalog::get(opCtx).lookupCollectionCatalogEntryByNamespace(collNs);
-        cce->setIndexBuildScanning(opCtx, key, sideWritesIdent, constraintViolationsIdent);
+        DurableCatalog::get(opCtx)->setIndexBuildScanning(
+            opCtx, collNs, key, sideWritesIdent, constraintViolationsIdent);
     }
 
     void indexBuildDrain(OperationContext* opCtx, NamespaceString collNs, std::string key) {
-        CollectionCatalogEntry* cce =
-            CollectionCatalog::get(opCtx).lookupCollectionCatalogEntryByNamespace(collNs);
-        cce->setIndexBuildDraining(opCtx, key);
+        DurableCatalog::get(opCtx)->setIndexBuildDraining(opCtx, collNs, key);
     }
 
     void indexBuildSuccess(OperationContext* opCtx, NamespaceString collNs, std::string key) {
-        CollectionCatalogEntry* cce =
-            CollectionCatalog::get(opCtx).lookupCollectionCatalogEntryByNamespace(collNs);
-        cce->indexBuildSuccess(opCtx, key);
+        DurableCatalog::get(opCtx)->indexBuildSuccess(opCtx, collNs, key);
     }
 
-    Status removeEntry(OperationContext* opCtx, StringData ns, KVCatalog* catalog) {
-        return catalog->_removeEntry(opCtx, NamespaceString(ns));
+    Status removeEntry(OperationContext* opCtx, StringData ns, DurableCatalog* catalog) {
+        return dynamic_cast<KVCatalog*>(catalog)->_removeEntry(opCtx, NamespaceString(ns));
     }
 
     StorageEngine* _storageEngine;