SERVER-25025 Make startup faster with many WT collections

This makes two major improvements:

(1) WiredTiger "metadata:create" cursors are very expensive; only use
these when absolutely necessary, and use standard "metadata:" cursors
everywhere else. This lowers startup calls to "metadata:create" from 3
to 1 per table.

(2) No longer open a cursor for every WiredTiger collection at startup.
Instead, lazily open cursors to initialize RecordIDs for inserts.

(cherry picked from commit ba9670e4c6d3aa62db20501317457af99049570c)

13 files changed, 203 insertions, 93 deletions

diff --git a/jstests/disk/wt_corrupt_file_errors.js b/jstests/disk/wt_corrupt_file_errors.js
index c45b5ae80df..28fa915002e 100644
--- a/jstests/disk/wt_corrupt_file_errors.js
+++ b/jstests/disk/wt_corrupt_file_errors.js
@@ -16,13 +16,23 @@ const dbpath = MongoRunner.dataPath + baseName + "/";
  * Test 1. Corrupt a collection's .wt file.
  */
 
-assertErrorOnStartupWhenFilesAreCorruptOrMissing(dbpath, baseName, collName, (mongod, testColl) => {
-    const testCollUri = getUriForColl(testColl);
-    const testCollFile = dbpath + testCollUri + ".wt";
-    MongoRunner.stopMongod(mongod);
-    jsTestLog("corrupting collection file: " + testCollFile);
-    corruptFile(testCollFile);
-}, "Fatal Assertion 50882");
+assertErrorOnRequestWhenFilesAreCorruptOrMissing(
+    dbpath,
+    baseName,
+    collName,
+    (mongod, testColl) => {
+        const testCollUri = getUriForColl(testColl);
+        const testCollFile = dbpath + testCollUri + ".wt";
+        MongoRunner.stopMongod(mongod);
+        jsTestLog("corrupting collection file: " + testCollFile);
+        corruptFile(testCollFile);
+    },
+    (testColl) => {
+        assert.throws(() => {
+            testColl.insert({a: 1});
+        });
+    },
+    "Fatal Assertion 50882");
 
 /**
  * Test 2. Corrupt the _mdb_catalog.
diff --git a/jstests/disk/wt_missing_file_errors.js b/jstests/disk/wt_missing_file_errors.js
index 1a28ff17f9c..18bcb46ab93 100644
--- a/jstests/disk/wt_missing_file_errors.js
+++ b/jstests/disk/wt_missing_file_errors.js
@@ -16,13 +16,23 @@ const dbpath = MongoRunner.dataPath + baseName + "/";
  * Test 1. Delete a collection's .wt file.
  */
 
-assertErrorOnStartupWhenFilesAreCorruptOrMissing(dbpath, baseName, collName, (mongod, testColl) => {
-    const testCollUri = getUriForColl(testColl);
-    const testCollFile = dbpath + testCollUri + ".wt";
-    MongoRunner.stopMongod(mongod);
-    jsTestLog("deleting collection file: " + testCollFile);
-    removeFile(testCollFile);
-}, "Fatal Assertion 50882");
+assertErrorOnRequestWhenFilesAreCorruptOrMissing(
+    dbpath,
+    baseName,
+    collName,
+    (mongod, testColl) => {
+        const testCollUri = getUriForColl(testColl);
+        const testCollFile = dbpath + testCollUri + ".wt";
+        MongoRunner.stopMongod(mongod);
+        jsTestLog("deleting collection file: " + testCollFile);
+        removeFile(testCollFile);
+    },
+    (testColl) => {
+        assert.throws(() => {
+            testColl.insert({a: 1});
+        });
+    },
+    "Fatal Assertion 50882");
 
 /**
  * Test 2. Delete the _mdb_catalog.
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_index.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_index.cpp
index ac0a40b1958..949f63263e6 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_index.cpp
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_index.cpp
@@ -383,7 +383,7 @@ bool WiredTigerIndex::appendCustomStats(OperationContext* opCtx,
     }
     std::string type, sourceURI;
     WiredTigerUtil::fetchTypeAndSourceURI(opCtx, _uri, &type, &sourceURI);
-    StatusWith<std::string> metadataResult = WiredTigerUtil::getMetadata(opCtx, sourceURI);
+    StatusWith<std::string> metadataResult = WiredTigerUtil::getMetadataCreate(opCtx, sourceURI);
     StringData creationStringName("creationString");
     if (!metadataResult.isOK()) {
         BSONObjBuilder creationString(output->subobjStart(creationStringName));
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.cpp
index 6f280b53c0b..1695bced3fc 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.cpp
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.cpp
@@ -63,6 +63,7 @@
 #include "mongo/db/repl/repl_settings.h"
 #include "mongo/db/repl/replication_coordinator.h"
 #include "mongo/db/server_options.h"
+#include "mongo/db/server_recovery.h"
 #include "mongo/db/service_context.h"
 #include "mongo/db/snapshot_window_options.h"
 #include "mongo/db/storage/journal_listener.h"
@@ -956,7 +957,7 @@ Status WiredTigerKVEngine::_rebuildIdent(WT_SESSION* session, const char* uri) {
 
     // This is safe to call after moving the file because it only reads from the metadata, and not
     // the data file itself.
-    auto swMetadata = WiredTigerUtil::getMetadataRaw(session, uri);
+    auto swMetadata = WiredTigerUtil::getMetadataCreate(session, uri);
     if (!swMetadata.isOK()) {
         error() << "Failed to get metadata for " << uri;
         return swMetadata.getStatus();
@@ -1248,6 +1249,16 @@ std::unique_ptr<RecordStore> WiredTigerKVEngine::getGroupedRecordStore(
     }
     ret->postConstructorInit(opCtx);
 
+    // Sizes should always be checked when creating a collection during rollback or replication
+    // recovery. This is in case the size storer information is no longer accurate. This may be
+    // necessary if capped deletes are rolled-back, if rollback occurs across a collection rename,
+    // or when collection creation is not part of a stable checkpoint.
+    const auto replCoord = repl::ReplicationCoordinator::get(getGlobalServiceContext());
+    const bool inRollback = replCoord && replCoord->getMemberState().rollback();
+    if (inRollback || inReplicationRecovery(getGlobalServiceContext())) {
+        ret->checkSize(opCtx);
+    }
+
     return std::move(ret);
 }
 
@@ -1494,8 +1505,9 @@ bool WiredTigerKVEngine::hasIdent(OperationContext* opCtx, StringData ident) con
 
 bool WiredTigerKVEngine::_hasUri(WT_SESSION* session, const std::string& uri) const {
     // can't use WiredTigerCursor since this is called from constructor.
-    WT_CURSOR* c = NULL;
-    int ret = session->open_cursor(session, "metadata:create", NULL, NULL, &c);
+    WT_CURSOR* c = nullptr;
+    // No need for a metadata:create cursor, since it gathers extra information and is slower.
+    int ret = session->open_cursor(session, "metadata:", nullptr, nullptr, &c);
     if (ret == ENOENT)
         return false;
     invariantWTOK(ret);
@@ -1508,7 +1520,8 @@ bool WiredTigerKVEngine::_hasUri(WT_SESSION* session, const std::string& uri) co
 std::vector<std::string> WiredTigerKVEngine::getAllIdents(OperationContext* opCtx) const {
     std::vector<std::string> all;
     int ret;
-    WiredTigerCursor cursor("metadata:create", WiredTigerSession::kMetadataTableId, false, opCtx);
+    // No need for a metadata:create cursor, since it gathers extra information and is slower.
+    WiredTigerCursor cursor("metadata:", WiredTigerSession::kMetadataTableId, false, opCtx);
     WT_CURSOR* c = cursor.get();
     if (!c)
         return all;
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_record_store.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_record_store.cpp
index 784a472af65..16358b35029 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_record_store.cpp
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_record_store.cpp
@@ -694,6 +694,13 @@ WiredTigerRecordStore::WiredTigerRecordStore(WiredTigerKVEngine* kvEngine,
         sizeRecoveryState(getGlobalServiceContext())
             .markCollectionAsAlwaysNeedsSizeAdjustment(_ident);
     }
+
+    // If no SizeStorer is in use, start counting at zero. In practice, this will only ever be the
+    // the case for temporary RecordStores (those not associated with any collection) and in unit
+    // tests. Persistent size information is not required in either case. If a RecordStore needs
+    // persistent size information, we require it to use a SizeStorer.
+    _sizeInfo = _sizeStorer ? _sizeStorer->load(_uri)
+                            : std::make_shared<WiredTigerSizeStorer::SizeInfo>(0, 0);
 }
 
 WiredTigerRecordStore::~WiredTigerRecordStore() {
@@ -718,29 +725,9 @@ WiredTigerRecordStore::~WiredTigerRecordStore() {
     }
 }
 
-void WiredTigerRecordStore::postConstructorInit(OperationContext* opCtx) {
-    // Find the largest RecordId currently in use and estimate the number of records.
-    std::unique_ptr<SeekableRecordCursor> cursor = getCursor(opCtx, /*forward=*/false);
-    _sizeInfo =
-        _sizeStorer ? _sizeStorer->load(_uri) : std::make_shared<WiredTigerSizeStorer::SizeInfo>();
-
-    if (auto record = cursor->next()) {
-        int64_t max = record->id.repr();
-        _nextIdNum.store(1 + max);
-
-        if (!_sizeStorer) {
-            LOG(1) << "Doing scan of collection " << ns() << " to get size and count info";
-
-            int64_t numRecords = 0;
-            int64_t dataSize = 0;
-            do {
-                numRecords++;
-                dataSize += record->data.size();
-            } while ((record = cursor->next()));
-            _sizeInfo->numRecords.store(numRecords);
-            _sizeInfo->dataSize.store(dataSize);
-        }
-    } else {
+void WiredTigerRecordStore::checkSize(OperationContext* opCtx) {
+    std::unique_ptr<SeekableRecordCursor> cursor = getCursor(opCtx, /*forward=*/true);
+    if (!cursor->next()) {
         // We found no records in this collection; however, there may actually be documents present
         // if writes to this collection were not included in the stable checkpoint the last time
         // this node shut down. We set the data size and the record count to zero, but will adjust
@@ -759,14 +746,13 @@ void WiredTigerRecordStore::postConstructorInit(OperationContext* opCtx) {
             .markCollectionAsAlwaysNeedsSizeAdjustment(_ident);
         _sizeInfo->dataSize.store(0);
         _sizeInfo->numRecords.store(0);
-
-        // Need to start at 1 so we are always higher than RecordId::min()
-        _nextIdNum.store(1);
     }
 
     if (_sizeStorer)
         _sizeStorer->store(_uri, _sizeInfo);
+}
 
+void WiredTigerRecordStore::postConstructorInit(OperationContext* opCtx) {
     if (WiredTigerKVEngine::initRsOplogBackgroundThread(ns())) {
         _oplogStones = std::make_shared<OplogStones>(opCtx, this);
     }
@@ -974,10 +960,10 @@ void WiredTigerRecordStore::_positionAtFirstRecordId(OperationContext* opCtx,
                                                      WT_CURSOR* cursor,
                                                      const RecordId& firstRecordId,
                                                      bool forTruncate) const {
-    // Use the previous first RecordId, if available, to navigate to the current first
-    // RecordId. The straightforward algorithm of resetting the cursor and advancing to the first
-    // element will be slow for capped collections since there may be many tombstones to traverse
-    // at the beginning of the table.
+    // Use the previous first RecordId, if available, to navigate to the current first RecordId. The
+    // straightforward algorithm of resetting the cursor and advancing to the first element will be
+    // slow for capped collections since there may be many tombstones to traverse at the beginning
+    // of the table.
     if (!firstRecordId.isNull()) {
         setKey(cursor, firstRecordId);
         // Truncate does not require its cursor to be explicitly positioned.
@@ -1274,10 +1260,8 @@ Status WiredTigerRecordStore::_insertRecords(OperationContext* opCtx,
             if (!status.isOK())
                 return status.getStatus();
             record.id = status.getValue();
-        } else if (_isCapped) {
-            record.id = _nextId();
         } else {
-            record.id = _nextId();
+            record.id = _nextId(opCtx);
         }
         dassert(record.id > highestId);
         highestId = record.id;
@@ -1595,7 +1579,7 @@ void WiredTigerRecordStore::appendCustomStats(OperationContext* opCtx,
 
     std::string type, sourceURI;
     WiredTigerUtil::fetchTypeAndSourceURI(opCtx, _uri, &type, &sourceURI);
-    StatusWith<std::string> metadataResult = WiredTigerUtil::getMetadata(opCtx, sourceURI);
+    StatusWith<std::string> metadataResult = WiredTigerUtil::getMetadataCreate(opCtx, sourceURI);
     StringData creationStringName("creationString");
     if (!metadataResult.isOK()) {
         BSONObjBuilder creationString(bob.subobjStart(creationStringName));
@@ -1683,8 +1667,35 @@ void WiredTigerRecordStore::updateStatsAfterRepair(OperationContext* opCtx,
         _sizeStorer->store(_uri, _sizeInfo);
 }
 
-RecordId WiredTigerRecordStore::_nextId() {
+void WiredTigerRecordStore::_initNextIdIfNeeded(OperationContext* opCtx) {
+    // In the normal case, this will already be initialized, so use a weak load. Since this value
+    // will only change from 0 to a positive integer, the only risk is reading an outdated value, 0,
+    // and having to take the mutex.
+    if (_nextIdNum.loadRelaxed() > 0) {
+        return;
+    }
+
+    // Only one thread needs to do this.
+    stdx::lock_guard<stdx::mutex> lk(_initNextIdMutex);
+    if (_nextIdNum.load() > 0) {
+        return;
+    }
+
+    // Need to start at 1 so we are always higher than RecordId::min()
+    int64_t nextId = 1;
+
+    // Find the largest RecordId currently in use.
+    std::unique_ptr<SeekableRecordCursor> cursor = getCursor(opCtx, /*forward=*/false);
+    if (auto record = cursor->next()) {
+        nextId = record->id.repr() + 1;
+    }
+
+    _nextIdNum.store(nextId);
+}
+
+RecordId WiredTigerRecordStore::_nextId(OperationContext* opCtx) {
     invariant(!_isOplog);
+    _initNextIdIfNeeded(opCtx);
     RecordId out = RecordId(_nextIdNum.fetchAndAdd(1));
     invariant(out.isNormal());
     return out;
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_record_store.h b/src/mongo/db/storage/wiredtiger/wiredtiger_record_store.h
index 49786746884..a3d5870a489 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_record_store.h
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_record_store.h
@@ -237,6 +237,13 @@ public:
         return _tableId;
     }
 
+    /*
+     * Check the size information for this RecordStore. This function opens a cursor on the
+     * RecordStore to determine if it is empty. If it is empty, it will mark the collection as
+     * needing size adjustment as a result of a rollback or storage recovery event.
+     */
+    void checkSize(OperationContext* opCtx);
+
     void setSizeStorer(WiredTigerSizeStorer* ss) {
         _sizeStorer = ss;
     }
@@ -286,11 +293,18 @@ private:
                           const Timestamp* timestamps,
                           size_t nRecords);
 
-    RecordId _nextId();
-    void _setId(RecordId id);
+    RecordId _nextId(OperationContext* opCtx);
     bool cappedAndNeedDelete() const;
     RecordData _getData(const WiredTigerCursor& cursor) const;
 
+
+    /**
+     * Initialize the largest known RecordId if it is not already. This is designed to be called
+     * immediately before operations that may need this Recordid. This is to support lazily
+     * initializing the value instead of all at once during startup.
+     */
+    void _initNextIdIfNeeded(OperationContext* opCtx);
+
     /**
      * Position the cursor at the first key. The previously known first key is
      * provided, as well as an indicator that this is being positioned for
@@ -358,7 +372,9 @@ private:
     int _cappedDeleteCheckCount;
     mutable stdx::timed_mutex _cappedDeleterMutex;
 
-    AtomicWord<long long> _nextIdNum;
+    // Protects initialization of the _nextIdNum.
+    mutable stdx::mutex _initNextIdMutex;
+    AtomicWord<long long> _nextIdNum{0};
 
     WiredTigerSizeStorer* _sizeStorer;  // not owned, can be NULL
     std::shared_ptr<WiredTigerSizeStorer::SizeInfo> _sizeInfo;
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_session_cache.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_session_cache.cpp
index fb41f7c2b3b..1fee696bc79 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_session_cache.cpp
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_session_cache.cpp
@@ -187,7 +187,7 @@ void WiredTigerSession::closeCursorsForQueuedDrops(WiredTigerKVEngine* engine) {
 }
 
 namespace {
-AtomicWord<unsigned long long> nextTableId(1);
+AtomicWord<unsigned long long> nextTableId(WiredTigerSession::kLastTableId);
 }
 // static
 uint64_t WiredTigerSession::genTableId() {
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_session_cache.h b/src/mongo/db/storage/wiredtiger/wiredtiger_session_cache.h
index afb2da1fbed..aff9ef8d874 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_session_cache.h
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_session_cache.h
@@ -149,9 +149,16 @@ public:
     static uint64_t genTableId();
 
     /**
-     * For "metadata:" cursors. Guaranteed never to collide with genTableId() ids.
+     * For special cursors. Guaranteed never to collide with genTableId() ids.
      */
-    static const uint64_t kMetadataTableId = 0;
+    enum TableId {
+        /* For "metadata:" cursors */
+        kMetadataTableId,
+        /* For "metadata:create" cursors */
+        kMetadataCreateTableId,
+        /* The start of non-special table ids for genTableId() */
+        kLastTableId
+    };
 
     void setIdleExpireTime(Date_t idleExpireTime) {
         _idleExpireTime = idleExpireTime;
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_size_storer.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_size_storer.cpp
index 56c8161d134..1cf95a4ad7d 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_size_storer.cpp
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_size_storer.cpp
@@ -116,10 +116,8 @@ std::shared_ptr<WiredTigerSizeStorer::SizeInfo> WiredTigerSizeStorer::load(Strin
     BSONObj data(reinterpret_cast<const char*>(value.data));
 
     LOG(2) << "WiredTigerSizeStorer::load " << uri << " -> " << redact(data);
-    auto result = std::make_shared<SizeInfo>();
-    result->numRecords.store(data["numRecords"].safeNumberLong());
-    result->dataSize.store(data["dataSize"].safeNumberLong());
-    return result;
+    return std::make_shared<SizeInfo>(data["numRecords"].safeNumberLong(),
+                                      data["dataSize"].safeNumberLong());
 }
 
 void WiredTigerSizeStorer::flush(bool syncToDisk) {
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_size_storer.h b/src/mongo/db/storage/wiredtiger/wiredtiger_size_storer.h
index 5db2a4e72bc..0063b2b6cfe 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_size_storer.h
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_size_storer.h
@@ -62,6 +62,9 @@ public:
      * SizeInfo.
      */
     struct SizeInfo {
+        SizeInfo() = default;
+        SizeInfo(long long records, long long size) : numRecords(records), dataSize(size) {}
+
         ~SizeInfo() {
             invariant(!_dirty.load());
         }
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_util.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_util.cpp
index 9c2b1155483..d9e327163ff 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_util.cpp
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_util.cpp
@@ -89,7 +89,7 @@ void WiredTigerUtil::fetchTypeAndSourceURI(OperationContext* opCtx,
     const size_t colon = tableUri.find(':');
     invariant(colon != string::npos);
     colgroupUri += tableUri.substr(colon);
-    StatusWith<std::string> colgroupResult = getMetadata(opCtx, colgroupUri);
+    StatusWith<std::string> colgroupResult = getMetadataCreate(opCtx, colgroupUri);
     invariant(colgroupResult.getStatus());
     WiredTigerConfigParser parser(colgroupResult.getValue());
 
@@ -104,12 +104,8 @@ void WiredTigerUtil::fetchTypeAndSourceURI(OperationContext* opCtx,
     *source = std::string(sourceItem.str, sourceItem.len);
 }
 
-StatusWith<std::string> WiredTigerUtil::getMetadataRaw(WT_SESSION* session, StringData uri) {
-    WT_CURSOR* cursor;
-    invariantWTOK(session->open_cursor(session, "metadata:create", nullptr, "", &cursor));
-    invariant(cursor);
-    ON_BLOCK_EXIT([cursor] { invariantWTOK(cursor->close(cursor)); });
-
+namespace {
+StatusWith<std::string> _getMetadata(WT_CURSOR* cursor, StringData uri) {
     std::string strUri = uri.toString();
     cursor->set_key(cursor, strUri.c_str());
     int ret = cursor->search(cursor);
@@ -127,33 +123,48 @@ StatusWith<std::string> WiredTigerUtil::getMetadataRaw(WT_SESSION* session, Stri
     invariant(metadata);
     return StatusWith<std::string>(metadata);
 }
+}  // namespace
 
-StatusWith<std::string> WiredTigerUtil::getMetadata(OperationContext* opCtx, StringData uri) {
+StatusWith<std::string> WiredTigerUtil::getMetadataCreate(WT_SESSION* session, StringData uri) {
+    WT_CURSOR* cursor;
+    invariantWTOK(session->open_cursor(session, "metadata:create", nullptr, "", &cursor));
+    invariant(cursor);
+    ON_BLOCK_EXIT([cursor] { invariantWTOK(cursor->close(cursor)); });
+
+    return _getMetadata(cursor, uri);
+}
+
+StatusWith<std::string> WiredTigerUtil::getMetadataCreate(OperationContext* opCtx, StringData uri) {
     invariant(opCtx);
 
     auto session = WiredTigerRecoveryUnit::get(opCtx)->getSessionNoTxn();
     WT_CURSOR* cursor =
-        session->getCursor("metadata:create", WiredTigerSession::kMetadataTableId, false);
+        session->getCursor("metadata:create", WiredTigerSession::kMetadataCreateTableId, false);
+
+    auto releaser = makeGuard(
+        [&] { session->releaseCursor(WiredTigerSession::kMetadataCreateTableId, cursor); });
+
+    return _getMetadata(cursor, uri);
+}
+
+StatusWith<std::string> WiredTigerUtil::getMetadata(WT_SESSION* session, StringData uri) {
+    WT_CURSOR* cursor;
+    invariantWTOK(session->open_cursor(session, "metadata:", nullptr, "", &cursor));
     invariant(cursor);
+    ON_BLOCK_EXIT([cursor] { invariantWTOK(cursor->close(cursor)); });
+
+    return _getMetadata(cursor, uri);
+}
+
+StatusWith<std::string> WiredTigerUtil::getMetadata(OperationContext* opCtx, StringData uri) {
+    invariant(opCtx);
+
+    auto session = WiredTigerRecoveryUnit::get(opCtx)->getSessionNoTxn();
+    WT_CURSOR* cursor = session->getCursor("metadata:", WiredTigerSession::kMetadataTableId, false);
     auto releaser =
         makeGuard([&] { session->releaseCursor(WiredTigerSession::kMetadataTableId, cursor); });
 
-    std::string strUri = uri.toString();
-    cursor->set_key(cursor, strUri.c_str());
-    int ret = cursor->search(cursor);
-    if (ret == WT_NOTFOUND) {
-        return StatusWith<std::string>(ErrorCodes::NoSuchKey,
-                                       str::stream() << "Unable to find metadata for " << uri);
-    } else if (ret != 0) {
-        return StatusWith<std::string>(wtRCToStatus(ret));
-    }
-    const char* metadata = NULL;
-    ret = cursor->get_value(cursor, &metadata);
-    if (ret != 0) {
-        return StatusWith<std::string>(wtRCToStatus(ret));
-    }
-    invariant(metadata);
-    return StatusWith<std::string>(metadata);
+    return _getMetadata(cursor, uri);
 }
 
 Status WiredTigerUtil::getApplicationMetadata(OperationContext* opCtx,
@@ -564,7 +575,7 @@ Status WiredTigerUtil::setTableLogging(WT_SESSION* session, const std::string& u
     //
     // If the settings need to be changed (only expected at startup), the alter table call must
     // succeed.
-    std::string existingMetadata = getMetadataRaw(session, uri).getValue();
+    std::string existingMetadata = getMetadataCreate(session, uri).getValue();
     if (existingMetadata.find("log=(enabled=true)") != std::string::npos &&
         existingMetadata.find("log=(enabled=false)") != std::string::npos) {
         // Sanity check against a table having multiple logging specifications.
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_util.h b/src/mongo/db/storage/wiredtiger/wiredtiger_util.h
index ced08bc3753..1e7e5d9c113 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_util.h
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_util.h
@@ -173,14 +173,20 @@ public:
                                              BSONObjBuilder* bob);
 
     /**
-     * Gets entire metadata string for collection/index at URI with the provided session.
+     * Gets the creation metadata string for a collection or index at a given URI. Accepts an
+     * OperationContext or session.
+     *
+     * This returns more information, but is slower than getMetadata().
      */
-    static StatusWith<std::string> getMetadataRaw(WT_SESSION* session, StringData uri);
+    static StatusWith<std::string> getMetadataCreate(OperationContext* opCtx, StringData uri);
+    static StatusWith<std::string> getMetadataCreate(WT_SESSION* session, StringData uri);
 
     /**
-     * Gets entire metadata string for collection/index at URI.
+     * Gets the entire metadata string for collection or index at URI. Accepts an OperationContext
+     * or session.
      */
     static StatusWith<std::string> getMetadata(OperationContext* opCtx, StringData uri);
+    static StatusWith<std::string> getMetadata(WT_SESSION* session, StringData uri);
 
     /**
      * Reads app_metadata for collection/index at URI as a BSON document.
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_util_test.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_util_test.cpp
index e36b625b18d..0f259aea36c 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_util_test.cpp
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_util_test.cpp
@@ -136,6 +136,31 @@ private:
     std::unique_ptr<OperationContext> _opCtx;
 };
 
+TEST_F(WiredTigerUtilMetadataTest, GetMetadataCreateInvalid) {
+    StatusWith<std::string> result =
+        WiredTigerUtil::getMetadataCreate(getOperationContext(), getURI());
+    ASSERT_NOT_OK(result.getStatus());
+    ASSERT_EQUALS(ErrorCodes::NoSuchKey, result.getStatus().code());
+}
+
+TEST_F(WiredTigerUtilMetadataTest, GetMetadataCreateNull) {
+    const char* config = nullptr;
+    createSession(config);
+    StatusWith<std::string> result =
+        WiredTigerUtil::getMetadataCreate(getOperationContext(), getURI());
+    ASSERT_OK(result.getStatus());
+    ASSERT_FALSE(result.getValue().empty());
+}
+
+TEST_F(WiredTigerUtilMetadataTest, GetMetadataCreateStringSimple) {
+    const char* config = "app_metadata=(abc=123)";
+    createSession(config);
+    StatusWith<std::string> result =
+        WiredTigerUtil::getMetadataCreate(getOperationContext(), getURI());
+    ASSERT_OK(result.getStatus());
+    ASSERT_STRING_CONTAINS(result.getValue(), config);
+}
+
 TEST_F(WiredTigerUtilMetadataTest, GetConfigurationStringInvalidURI) {
     StatusWith<std::string> result = WiredTigerUtil::getMetadata(getOperationContext(), getURI());
     ASSERT_NOT_OK(result.getStatus());