SERVER-53990 SortedDataInterface supports RecordIds in the binary string format

This implements secondary index support for time-series buckets collections

31 files changed, 711 insertions, 212 deletions

diff --git a/src/mongo/db/catalog/collection_impl.cpp b/src/mongo/db/catalog/collection_impl.cpp
index c7add8332bf..3eb9a454343 100644
--- a/src/mongo/db/catalog/collection_impl.cpp
+++ b/src/mongo/db/catalog/collection_impl.cpp
@@ -330,7 +330,6 @@ void CollectionImpl::init(OperationContext* opCtx) {
     }
 
     if (collectionOptions.clusteredIndex) {
-        invariant(_shared->_recordStore->keyFormat() == KeyFormat::String);
         _clustered = true;
     }
 
@@ -364,8 +363,8 @@ bool CollectionImpl::requiresIdIndex() const {
         return false;
     }
 
-    if (_ns.isTimeseriesBucketsCollection()) {
-        // Time-series bucket collections have a clustered _id index.
+    if (isClustered()) {
+        // Collections clustered by _id do not have a separate _id index.
         return false;
     }
 
diff --git a/src/mongo/db/catalog/collection_test.cpp b/src/mongo/db/catalog/collection_test.cpp
index db4209776a6..da2ea437d1c 100644
--- a/src/mongo/db/catalog/collection_test.cpp
+++ b/src/mongo/db/catalog/collection_test.cpp
@@ -222,6 +222,8 @@ TEST_F(CollectionTest, CreateTimeseriesBucketCollection) {
                                      << "_id_"
                                      << "key" << BSON("_id" << 1));
 
+    CollectionOptions options;
+    options.clusteredIndex = ClusteredIndexOptions{};
     {
         WriteUnitOfWork wuow(operationContext());
 
@@ -229,7 +231,7 @@ TEST_F(CollectionTest, CreateTimeseriesBucketCollection) {
         // the collection.
         Collection* collection = db->createCollection(operationContext(),
                                                       nss,
-                                                      CollectionOptions(),
+                                                      options,
                                                       /*createIdIndex=*/true,
                                                       /*idIndex=*/
                                                       idxSpec);
@@ -248,8 +250,8 @@ TEST_F(CollectionTest, CreateTimeseriesBucketCollection) {
 
     {
         WriteUnitOfWork wuow(operationContext());
-        auto collection = db->createCollection(
-            operationContext(), nss, CollectionOptions(), /*createIdIndex=*/false);
+        auto collection =
+            db->createCollection(operationContext(), nss, options, /*createIdIndex=*/false);
         ASSERT(collection);
         ASSERT_EQ(0, collection->getIndexCatalog()->numIndexesTotal(operationContext()));
         wuow.commit();
diff --git a/src/mongo/db/catalog/index_catalog_impl.cpp b/src/mongo/db/catalog/index_catalog_impl.cpp
index 5752afa8ea1..e2656edd0a5 100644
--- a/src/mongo/db/catalog/index_catalog_impl.cpp
+++ b/src/mongo/db/catalog/index_catalog_impl.cpp
@@ -451,8 +451,10 @@ IndexCatalogEntry* IndexCatalogImpl::createIndexEntry(OperationContext* opCtx,
         opCtx, _collection->getCatalogId(), ident, std::move(descriptor), frozen);
 
     IndexDescriptor* desc = entry->descriptor();
+    auto collOptions =
+        DurableCatalog::get(opCtx)->getCollectionOptions(opCtx, _collection->getCatalogId());
     std::unique_ptr<SortedDataInterface> sdi =
-        engine->getEngine()->getSortedDataInterface(opCtx, ident, desc);
+        engine->getEngine()->getSortedDataInterface(opCtx, collOptions, ident, desc);
 
     std::unique_ptr<IndexAccessMethod> accessMethod =
         IndexAccessMethodFactory::get(opCtx)->make(entry.get(), std::move(sdi));
diff --git a/src/mongo/db/catalog/index_consistency.cpp b/src/mongo/db/catalog/index_consistency.cpp
index 62624f79c9a..d388033cdf5 100644
--- a/src/mongo/db/catalog/index_consistency.cpp
+++ b/src/mongo/db/catalog/index_consistency.cpp
@@ -134,7 +134,7 @@ void IndexConsistency::repairMissingIndexEntries(OperationContext* opCtx,
     for (auto it = _missingIndexEntries.begin(); it != _missingIndexEntries.end();) {
         const IndexKey& key = it->first;
         const KeyString::Value& ks = it->second.keyString;
-        const RecordId rid = KeyString::decodeRecordIdAtEnd(ks.getBuffer(), ks.getSize());
+        const RecordId rid = KeyString::decodeRecordIdLongAtEnd(ks.getBuffer(), ks.getSize());
 
         const std::string& indexName = key.first;
         if (indexName != index->descriptor()->indexName()) {
diff --git a/src/mongo/db/catalog/validate_adaptor.cpp b/src/mongo/db/catalog/validate_adaptor.cpp
index fd61311da8e..7de776ac533 100644
--- a/src/mongo/db/catalog/validate_adaptor.cpp
+++ b/src/mongo/db/catalog/validate_adaptor.cpp
@@ -238,9 +238,9 @@ void _validateKeyOrder(OperationContext* opCtx,
         if (results && results->valid) {
             auto bsonKey = KeyString::toBson(currKey, Ordering::make(descriptor->keyPattern()));
             auto firstRecordId =
-                KeyString::decodeRecordIdAtEnd(prevKey.getBuffer(), prevKey.getSize());
+                KeyString::decodeRecordIdLongAtEnd(prevKey.getBuffer(), prevKey.getSize());
             auto secondRecordId =
-                KeyString::decodeRecordIdAtEnd(currKey.getBuffer(), currKey.getSize());
+                KeyString::decodeRecordIdLongAtEnd(currKey.getBuffer(), currKey.getSize());
             results->errors.push_back(str::stream() << "Unique index '" << descriptor->indexName()
                                                     << "' has duplicate key: " << bsonKey
                                                     << ", first record: " << firstRecordId
@@ -299,8 +299,18 @@ void ValidateAdaptor::traverseIndex(OperationContext* opCtx,
                 opCtx, index, indexEntry->keyString, prevIndexKeyStringValue, &indexResults);
         }
 
-        const RecordId kWildcardMultikeyMetadataRecordId{
-            RecordId::reservedIdFor<int64_t>(RecordId::Reservation::kWildcardMultikeyMetadataId)};
+
+        const RecordId kWildcardMultikeyMetadataRecordId = [&]() {
+            auto keyFormat = _validateState->getCollection()->getRecordStore()->keyFormat();
+            if (keyFormat == KeyFormat::Long) {
+                return RecordId::reservedIdFor<int64_t>(
+                    RecordId::Reservation::kWildcardMultikeyMetadataId);
+            } else {
+                invariant(keyFormat == KeyFormat::String);
+                return RecordId::reservedIdFor<OID>(
+                    RecordId::Reservation::kWildcardMultikeyMetadataId);
+            }
+        }();
         if (descriptor->getIndexType() == IndexType::INDEX_WILDCARD &&
             indexEntry->loc == kWildcardMultikeyMetadataRecordId) {
             _indexConsistency->removeMultikeyMetadataPath(indexEntry->keyString, &indexInfo);
diff --git a/src/mongo/db/index/index_access_method.cpp b/src/mongo/db/index/index_access_method.cpp
index bc53f99f189..486a6740012 100644
--- a/src/mongo/db/index/index_access_method.cpp
+++ b/src/mongo/db/index/index_access_method.cpp
@@ -872,7 +872,7 @@ std::string nextFileName() {
 Status AbstractIndexAccessMethod::_handleDuplicateKey(OperationContext* opCtx,
                                                       const KeyString::Value& dataKey,
                                                       const RecordIdHandlerFn& onDuplicateRecord) {
-    RecordId recordId = KeyString::decodeRecordIdAtEnd(dataKey.getBuffer(), dataKey.getSize());
+    RecordId recordId = KeyString::decodeRecordIdLongAtEnd(dataKey.getBuffer(), dataKey.getSize());
     if (onDuplicateRecord) {
         return onDuplicateRecord(recordId);
     }
diff --git a/src/mongo/db/index/index_build_interceptor.cpp b/src/mongo/db/index/index_build_interceptor.cpp
index a9227adc1fc..2e79179df2e 100644
--- a/src/mongo/db/index/index_build_interceptor.cpp
+++ b/src/mongo/db/index/index_build_interceptor.cpp
@@ -299,8 +299,15 @@ Status IndexBuildInterceptor::_applyWrite(OperationContext* opCtx,
         (strcmp(operation.getStringField("op"), "i") == 0) ? Op::kInsert : Op::kDelete;
 
     const KeyStringSet keySet{keyString};
-    const RecordId opRecordId =
-        KeyString::decodeRecordIdAtEnd(keyString.getBuffer(), keyString.getSize());
+    const RecordId opRecordId = [&]() {
+        auto keyFormat = coll->getRecordStore()->keyFormat();
+        if (keyFormat == KeyFormat::Long) {
+            return KeyString::decodeRecordIdLongAtEnd(keyString.getBuffer(), keyString.getSize());
+        } else {
+            invariant(keyFormat == KeyFormat::String);
+            return KeyString::decodeRecordIdStrAtEnd(keyString.getBuffer(), keyString.getSize());
+        }
+    }();
 
     auto accessMethod = _indexCatalogEntry->accessMethod();
     if (opType == Op::kInsert) {
diff --git a/src/mongo/db/storage/SConscript b/src/mongo/db/storage/SConscript
index 43f8ee4820b..2574035c6f2 100644
--- a/src/mongo/db/storage/SConscript
+++ b/src/mongo/db/storage/SConscript
@@ -219,6 +219,7 @@ env.Library(
         'sorted_data_interface_test_harness.cpp',
         'sorted_data_interface_test_insert.cpp',
         'sorted_data_interface_test_isempty.cpp',
+        'sorted_data_interface_test_keyformat_string.cpp',
         'sorted_data_interface_test_rollback.cpp',
         'sorted_data_interface_test_spaceused.cpp',
         'sorted_data_interface_test_unindex.cpp',
diff --git a/src/mongo/db/storage/devnull/devnull_kv_engine.cpp b/src/mongo/db/storage/devnull/devnull_kv_engine.cpp
index 0ddc0af6b6e..7f380c8b980 100644
--- a/src/mongo/db/storage/devnull/devnull_kv_engine.cpp
+++ b/src/mongo/db/storage/devnull/devnull_kv_engine.cpp
@@ -244,7 +244,10 @@ std::unique_ptr<RecordStore> DevNullKVEngine::makeTemporaryRecordStore(Operation
 }
 
 std::unique_ptr<SortedDataInterface> DevNullKVEngine::getSortedDataInterface(
-    OperationContext* opCtx, StringData ident, const IndexDescriptor* desc) {
+    OperationContext* opCtx,
+    const CollectionOptions& collOptions,
+    StringData ident,
+    const IndexDescriptor* desc) {
     return std::make_unique<DevNullSortedDataInterface>(ident);
 }
 
diff --git a/src/mongo/db/storage/devnull/devnull_kv_engine.h b/src/mongo/db/storage/devnull/devnull_kv_engine.h
index 47bfdc7f307..192193d3013 100644
--- a/src/mongo/db/storage/devnull/devnull_kv_engine.h
+++ b/src/mongo/db/storage/devnull/devnull_kv_engine.h
@@ -76,7 +76,10 @@ public:
     }
 
     virtual std::unique_ptr<SortedDataInterface> getSortedDataInterface(
-        OperationContext* opCtx, StringData ident, const IndexDescriptor* desc);
+        OperationContext* opCtx,
+        const CollectionOptions& collOptions,
+        StringData ident,
+        const IndexDescriptor* desc);
 
     virtual Status dropIdent(RecoveryUnit* ru,
                              StringData ident,
diff --git a/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_kv_engine.cpp b/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_kv_engine.cpp
index d526dfbc36c..3b3b4308c68 100644
--- a/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_kv_engine.cpp
+++ b/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_kv_engine.cpp
@@ -148,7 +148,11 @@ Status KVEngine::importSortedDataInterface(OperationContext* opCtx,
 }
 
 std::unique_ptr<mongo::SortedDataInterface> KVEngine::getSortedDataInterface(
-    OperationContext* opCtx, StringData ident, const IndexDescriptor* desc) {
+    OperationContext* opCtx,
+    const CollectionOptions& collOptions,
+    StringData ident,
+    const IndexDescriptor* desc) {
+    invariant(!collOptions.clusteredIndex);
     {
         stdx::lock_guard lock(_identsLock);
         _idents[ident.toString()] = false;
diff --git a/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_kv_engine.h b/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_kv_engine.h
index 7f8120c2751..dd77b1ae635 100644
--- a/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_kv_engine.h
+++ b/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_kv_engine.h
@@ -85,7 +85,10 @@ public:
     }
 
     virtual std::unique_ptr<mongo::SortedDataInterface> getSortedDataInterface(
-        OperationContext* opCtx, StringData ident, const IndexDescriptor* desc);
+        OperationContext* opCtx,
+        const CollectionOptions& collOptions,
+        StringData ident,
+        const IndexDescriptor* desc);
 
     virtual Status beginBackup(OperationContext* opCtx) {
         return Status::OK();
diff --git a/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_sorted_impl.cpp b/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_sorted_impl.cpp
index 303740dbeab..16e7bad74e7 100644
--- a/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_sorted_impl.cpp
+++ b/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_sorted_impl.cpp
@@ -1188,9 +1188,10 @@ SortedDataBuilderBase::SortedDataBuilderBase(OperationContext* opCtx,
       _collation(collation) {}
 
 Status SortedDataBuilderUnique::addKey(const KeyString::Value& keyString) {
-    dassert(KeyString::decodeRecordIdAtEnd(keyString.getBuffer(), keyString.getSize()).isValid());
+    dassert(
+        KeyString::decodeRecordIdLongAtEnd(keyString.getBuffer(), keyString.getSize()).isValid());
     StringStore* workingCopy(RecoveryUnit::get(_opCtx)->getHead());
-    RecordId loc = KeyString::decodeRecordIdAtEnd(keyString.getBuffer(), keyString.getSize());
+    RecordId loc = KeyString::decodeRecordIdLongAtEnd(keyString.getBuffer(), keyString.getSize());
 
     std::string key = createRadixKeyWithoutLocFromKS(keyString, _prefix);
     auto it = workingCopy->find(key);
@@ -1285,7 +1286,7 @@ Status SortedDataInterfaceUnique::insert(OperationContext* opCtx,
                                          const KeyString::Value& keyString,
                                          bool dupsAllowed) {
     StringStore* workingCopy(RecoveryUnit::get(opCtx)->getHead());
-    RecordId loc = KeyString::decodeRecordIdAtEnd(keyString.getBuffer(), keyString.getSize());
+    RecordId loc = KeyString::decodeRecordIdLongAtEnd(keyString.getBuffer(), keyString.getSize());
 
     std::string key = createRadixKeyWithoutLocFromKS(keyString, _prefix);
     auto it = workingCopy->find(key);
@@ -1316,7 +1317,7 @@ void SortedDataInterfaceUnique::unindex(OperationContext* opCtx,
                                         const KeyString::Value& keyString,
                                         bool dupsAllowed) {
     StringStore* workingCopy(RecoveryUnit::get(opCtx)->getHead());
-    RecordId loc = KeyString::decodeRecordIdAtEnd(keyString.getBuffer(), keyString.getSize());
+    RecordId loc = KeyString::decodeRecordIdLongAtEnd(keyString.getBuffer(), keyString.getSize());
 
     auto key = createRadixKeyWithoutLocFromKS(keyString, _prefix);
     auto it = workingCopy->find(key);
@@ -1428,9 +1429,10 @@ Status SortedDataInterfaceBase::initAsEmpty(OperationContext* opCtx) {
 }
 
 Status SortedDataBuilderStandard::addKey(const KeyString::Value& keyString) {
-    dassert(KeyString::decodeRecordIdAtEnd(keyString.getBuffer(), keyString.getSize()).isValid());
+    dassert(
+        KeyString::decodeRecordIdLongAtEnd(keyString.getBuffer(), keyString.getSize()).isValid());
     StringStore* workingCopy(RecoveryUnit::get(_opCtx)->getHead());
-    RecordId loc = KeyString::decodeRecordIdAtEnd(keyString.getBuffer(), keyString.getSize());
+    RecordId loc = KeyString::decodeRecordIdLongAtEnd(keyString.getBuffer(), keyString.getSize());
 
     std::string key = createRadixKeyWithLocFromKS(keyString, loc, _prefix);
     bool inserted =
@@ -1483,7 +1485,7 @@ Status SortedDataInterfaceStandard::insert(OperationContext* opCtx,
                                            const KeyString::Value& keyString,
                                            bool dupsAllowed) {
     StringStore* workingCopy(RecoveryUnit::get(opCtx)->getHead());
-    RecordId loc = KeyString::decodeRecordIdAtEnd(keyString.getBuffer(), keyString.getSize());
+    RecordId loc = KeyString::decodeRecordIdLongAtEnd(keyString.getBuffer(), keyString.getSize());
 
     std::string key = createRadixKeyWithLocFromKS(keyString, loc, _prefix);
     bool inserted =
@@ -1498,7 +1500,7 @@ void SortedDataInterfaceStandard::unindex(OperationContext* opCtx,
                                           const KeyString::Value& keyString,
                                           bool dupsAllowed) {
     StringStore* workingCopy(RecoveryUnit::get(opCtx)->getHead());
-    RecordId loc = KeyString::decodeRecordIdAtEnd(keyString.getBuffer(), keyString.getSize());
+    RecordId loc = KeyString::decodeRecordIdLongAtEnd(keyString.getBuffer(), keyString.getSize());
 
     auto key = createRadixKeyWithLocFromKS(keyString, loc, _prefix);
     if (workingCopy->erase(key))
diff --git a/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_sorted_impl_test.cpp b/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_sorted_impl_test.cpp
index be04273b3f2..8a8b1502132 100644
--- a/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_sorted_impl_test.cpp
+++ b/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_sorted_impl_test.cpp
@@ -64,11 +64,17 @@ public:
         IndexDescriptor desc("", spec);
         invariant(desc.isIdIndex());
 
-        return _kvEngine.getSortedDataInterface(&opCtx, "ident"_sd, &desc);
+        return _kvEngine.getSortedDataInterface(&opCtx, CollectionOptions(), "ident"_sd, &desc);
     }
 
     std::unique_ptr<mongo::SortedDataInterface> newSortedDataInterface(bool unique,
-                                                                       bool partial) final {
+                                                                       bool partial,
+                                                                       KeyFormat keyFormat) final {
+        if (keyFormat == KeyFormat::String) {
+            // not supported
+            return nullptr;
+        }
+
         std::string ns = "test.ephemeral_for_test";
         OperationContextNoop opCtx(newRecoveryUnit().release());
 
@@ -85,7 +91,8 @@ public:
 
         auto collection = std::make_unique<CollectionMock>(NamespaceString(ns));
         _descs.emplace_back("", spec);
-        return _kvEngine.getSortedDataInterface(&opCtx, "ident"_sd, &_descs.back());
+        return _kvEngine.getSortedDataInterface(
+            &opCtx, CollectionOptions(), "ident"_sd, &_descs.back());
     }
 
     std::unique_ptr<mongo::RecoveryUnit> newRecoveryUnit() final {
diff --git a/src/mongo/db/storage/index_entry_comparison.h b/src/mongo/db/storage/index_entry_comparison.h
index 082e7a6b385..6a1574c7360 100644
--- a/src/mongo/db/storage/index_entry_comparison.h
+++ b/src/mongo/db/storage/index_entry_comparison.h
@@ -92,7 +92,17 @@ inline bool operator!=(const IndexKeyEntry& lhs, const IndexKeyEntry& rhs) {
  */
 struct KeyStringEntry {
     KeyStringEntry(KeyString::Value ks, RecordId loc) : keyString(ks), loc(loc) {
-        invariant(loc == KeyString::decodeRecordIdAtEnd(ks.getBuffer(), ks.getSize()));
+        if (!kDebugBuild) {
+            return;
+        }
+        loc.withFormat(
+            [](RecordId::Null n) { invariant(false); },
+            [&](int64_t rid) {
+                invariant(loc == KeyString::decodeRecordIdLongAtEnd(ks.getBuffer(), ks.getSize()));
+            },
+            [&](const char* str, int size) {
+                invariant(loc == KeyString::decodeRecordIdStrAtEnd(ks.getBuffer(), ks.getSize()));
+            });
     }
 
     KeyString::Value keyString;
diff --git a/src/mongo/db/storage/key_string.cpp b/src/mongo/db/storage/key_string.cpp
index 58c080dde60..89c4561b46b 100644
--- a/src/mongo/db/storage/key_string.cpp
+++ b/src/mongo/db/storage/key_string.cpp
@@ -452,10 +452,17 @@ template <class BufferT>
 void BuilderBase<BufferT>::appendRecordId(RecordId loc) {
     _doneAppending();
     _transition(BuildState::kAppendedRecordID);
+    loc.withFormat([](RecordId::Null n) { invariant(false); },
+                   [&](int64_t rid) { _appendRecordIdLong(rid); },
+                   [&](const char* str, int size) { _appendRecordIdStr(str, size); });
+}
+
+template <class BufferT>
+void BuilderBase<BufferT>::_appendRecordIdLong(int64_t val) {
     // The RecordId encoding must be able to determine the full length starting from the last
     // byte, without knowing where the first byte is since it is stored at the end of a
     // KeyString, and we need to be able to read the RecordId without decoding the whole thing.
-    //
+
     // This encoding places a number (N) between 0 and 7 in both the high 3 bits of the first
     // byte and the low 3 bits of the last byte. This is the number of bytes between the first
     // and last byte (ie total bytes is N + 2). The remaining bits of the first and last bytes
@@ -463,7 +470,7 @@ void BuilderBase<BufferT>::appendRecordId(RecordId loc) {
     // big-endian order. This does not encode negative RecordIds to give maximum space to
     // positive RecordIds which are the only ones that are allowed to be stored in an index.
 
-    int64_t raw = loc.asLong();
+    int64_t raw = val;
     if (raw < 0) {
         // Note: we encode RecordId::minLong() and RecordId() the same which is ok, as they
         // are never stored so they will never be compared to each other.
@@ -498,6 +505,14 @@ void BuilderBase<BufferT>::appendRecordId(RecordId loc) {
 }
 
 template <class BufferT>
+void BuilderBase<BufferT>::_appendRecordIdStr(const char* str, int size) {
+    // Only 12 byte strings can be encoded as RecordIds.
+    invariant(size == RecordId::kSmallStrSize);
+    const bool invert = false;
+    _appendBytes(str, size, invert);
+}
+
+template <class BufferT>
 void BuilderBase<BufferT>::appendTypeBits(const TypeBits& typeBits) {
     _transition(BuildState::kAppendedTypeBits);
     // As an optimization, encode AllZeros as a single 0 byte.
@@ -2438,15 +2453,15 @@ BSONObj toBson(StringData data, Ordering ord, const TypeBits& typeBits) {
     return toBson(data.rawData(), data.size(), ord, typeBits);
 }
 
-RecordId decodeRecordIdAtEnd(const void* bufferRaw, size_t bufSize) {
-    invariant(bufSize >= 2);  // smallest possible encoding of a RecordId.
+RecordId decodeRecordIdLongAtEnd(const void* bufferRaw, size_t bufSize) {
     const unsigned char* buffer = static_cast<const unsigned char*>(bufferRaw);
+    invariant(bufSize >= 2);  // smallest possible encoding of a RecordId.
     const unsigned char lastByte = *(buffer + bufSize - 1);
     const size_t ridSize = 2 + (lastByte & 0x7);  // stored in low 3 bits.
     invariant(bufSize >= ridSize);
     const unsigned char* firstBytePtr = buffer + bufSize - ridSize;
     BufReader reader(firstBytePtr, ridSize);
-    return decodeRecordId(&reader);
+    return decodeRecordIdLong(&reader);
 }
 
 size_t sizeWithoutRecordIdAtEnd(const void* bufferRaw, size_t bufSize) {
@@ -2458,7 +2473,7 @@ size_t sizeWithoutRecordIdAtEnd(const void* bufferRaw, size_t bufSize) {
     return bufSize - ridSize;
 }
 
-RecordId decodeRecordId(BufReader* reader) {
+RecordId decodeRecordIdLong(BufReader* reader) {
     const uint8_t firstByte = readType<uint8_t>(reader, false);
     const uint8_t numExtraBytes = firstByte >> 5;  // high 3 bits in firstByte
     uint64_t repr = firstByte & 0x1f;              // low 5 bits in firstByte
@@ -2472,6 +2487,22 @@ RecordId decodeRecordId(BufReader* reader) {
     return RecordId(repr);
 }
 
+RecordId decodeRecordIdStrAtEnd(const void* bufferRaw, size_t bufSize) {
+    const uint8_t* buffer = static_cast<const uint8_t*>(bufferRaw);
+    // We currently require all RecordId strings to be 12 bytes.
+    const int ridSize = RecordId::kSmallStrSize;
+    invariant(bufSize >= ridSize);
+    const uint8_t* firstBytePtr = (buffer + bufSize - ridSize);
+    BufReader reader(firstBytePtr, ridSize);
+    return decodeRecordIdStr(&reader);
+}
+
+RecordId decodeRecordIdStr(BufReader* reader) {
+    // We currently require all RecordId strings to be 12 bytes.
+    const int size = RecordId::kSmallStrSize;
+    return RecordId(static_cast<const char*>(reader->skip(size)), size);
+}
+
 int compare(const char* leftBuf, const char* rightBuf, size_t leftSize, size_t rightSize) {
     // memcmp has undefined behavior if either leftBuf or rightBuf is a null pointer.
     if (MONGO_unlikely(leftSize == 0))
@@ -2521,7 +2552,7 @@ bool readSBEValue(BufReader* reader,
 }
 
 void Value::serializeWithoutRecordId(BufBuilder& buf) const {
-    dassert(decodeRecordIdAtEnd(_buffer.get(), _ksSize).isValid());
+    dassert(decodeRecordIdLongAtEnd(_buffer.get(), _ksSize).isValid());
 
     const int32_t sizeWithoutRecordId = sizeWithoutRecordIdAtEnd(_buffer.get(), _ksSize);
     buf.appendNum(sizeWithoutRecordId);                 // Serialize size of KeyString
diff --git a/src/mongo/db/storage/key_string.h b/src/mongo/db/storage/key_string.h
index 9a9fe17a060..d4e45aea650 100644
--- a/src/mongo/db/storage/key_string.h
+++ b/src/mongo/db/storage/key_string.h
@@ -639,6 +639,9 @@ protected:
     void _appendNumberInt(const int num, bool invert);
     void _appendNumberDecimal(const Decimal128 num, bool invert);
 
+    void _appendRecordIdLong(const int64_t val);
+    void _appendRecordIdStr(const char* val, int size);
+
     /**
      * @param name - optional, can be NULL
      *              if NULL, not included in encoding
@@ -940,9 +943,14 @@ BSONObj toBson(const T& keyString, Ordering ord) noexcept {
 }
 
 /**
- * Decodes a RecordId from the end of a buffer.
+ * Decodes a RecordId long from the end of a buffer.
+ */
+RecordId decodeRecordIdLongAtEnd(const void* buf, size_t size);
+
+/**
+ * Decodes a RecordId string from the end of a buffer.
  */
-RecordId decodeRecordIdAtEnd(const void* buf, size_t size);
+RecordId decodeRecordIdStrAtEnd(const void* buf, size_t size);
 
 /**
  * Given a KeyString with a RecordId, returns the length of the section without the RecordId.
@@ -952,7 +960,8 @@ size_t sizeWithoutRecordIdAtEnd(const void* bufferRaw, size_t bufSize);
 /**
  * Decodes a RecordId, consuming all bytes needed from reader.
  */
-RecordId decodeRecordId(BufReader* reader);
+RecordId decodeRecordIdLong(BufReader* reader);
+RecordId decodeRecordIdStr(BufReader* reader);
 
 int compare(const char* leftBuf, const char* rightBuf, size_t leftSize, size_t rightSize);
 
diff --git a/src/mongo/db/storage/key_string_test.cpp b/src/mongo/db/storage/key_string_test.cpp
index 91d740130e7..c967916d160 100644
--- a/src/mongo/db/storage/key_string_test.cpp
+++ b/src/mongo/db/storage/key_string_test.cpp
@@ -306,7 +306,7 @@ TEST_F(KeyStringBuilderTest, ExceededBSONDepth) {
 
     // Construct a KeyString from the invalid BSON, and confirm that it fails to convert back to
     // BSON.
-    ks.resetToKey(nestedObj, ALL_ASCENDING, RecordId());
+    ks.resetToKey(nestedObj, ALL_ASCENDING, RecordId(1));
     ASSERT_THROWS_CODE(
         KeyString::toBsonSafe(ks.getBuffer(), ks.getSize(), ALL_ASCENDING, ks.getTypeBits()),
         AssertionException,
@@ -319,8 +319,8 @@ TEST_F(KeyStringBuilderTest, Simple1) {
 
     ASSERT_BSONOBJ_LT(a, b);
 
-    ASSERT_LESS_THAN(KeyString::Builder(version, a, ALL_ASCENDING, RecordId()),
-                     KeyString::Builder(version, b, ALL_ASCENDING, RecordId()));
+    ASSERT_LESS_THAN(KeyString::Builder(version, a, ALL_ASCENDING, RecordId(1)),
+                     KeyString::Builder(version, b, ALL_ASCENDING, RecordId(1)));
 }
 
 #define ROUNDTRIP_ORDER(version, x, order)                            \
@@ -839,63 +839,19 @@ TEST_F(KeyStringBuilderTest, LotsOfNumbers2) {
     }
 }
 
-TEST_F(KeyStringBuilderTest, LotsOfNumbers3) {
-    std::vector<stdx::future<void>> futures;
-
-    for (double k = 0; k < 8; k++) {
-        futures.push_back(stdx::async(stdx::launch::async, [k, this] {
-            for (double i = -1100; i < 1100; i++) {
-                for (double j = 0; j < 52; j++) {
-                    const auto V1 = KeyString::Version::V1;
-                    Decimal128::RoundingPrecision roundingPrecisions[]{
-                        Decimal128::kRoundTo15Digits, Decimal128::kRoundTo34Digits};
-                    Decimal128::RoundingMode roundingModes[]{Decimal128::kRoundTowardNegative,
-                                                             Decimal128::kRoundTowardPositive};
-                    double x = pow(2, i);
-                    double y = pow(2, i - j);
-                    double z = pow(2, i - 53 + k);
-                    double bin = x + y - z;
-
-                    // In general NaNs don't roundtrip as we only store a single NaN, see the NaNs
-                    // test.
-                    if (std::isnan(bin))
-                        continue;
-
-                    ROUNDTRIP(version, BSON("" << bin));
-                    ROUNDTRIP(version, BSON("" << -bin));
-
-                    if (version < V1)
-                        continue;
-
-                    for (auto precision : roundingPrecisions) {
-                        for (auto mode : roundingModes) {
-                            Decimal128 rounded = Decimal128(bin, precision, mode);
-                            ROUNDTRIP(V1, BSON("" << rounded));
-                            ROUNDTRIP(V1, BSON("" << rounded.negate()));
-                        }
-                    }
-                }
-            }
-        }));
-    }
-    for (auto&& future : futures) {
-        future.get();
-    }
-}
-
 TEST_F(KeyStringBuilderTest, RecordIdOrder1) {
     Ordering ordering = Ordering::make(BSON("a" << 1));
 
     KeyString::Builder a(version, BSON("" << 5), ordering, RecordId::minLong());
     KeyString::Builder b(version, BSON("" << 5), ordering, RecordId(2));
     KeyString::Builder c(version, BSON("" << 5), ordering, RecordId(3));
-    KeyString::Builder d(version, BSON("" << 6), ordering, RecordId());
+    KeyString::Builder d(version, BSON("" << 6), ordering, RecordId(4));
     KeyString::Builder e(version, BSON("" << 6), ordering, RecordId(1));
 
     ASSERT_LESS_THAN(a, b);
     ASSERT_LESS_THAN(b, c);
     ASSERT_LESS_THAN(c, d);
-    ASSERT_LESS_THAN(d, e);
+    ASSERT_LESS_THAN(e, d);
 }
 
 TEST_F(KeyStringBuilderTest, RecordIdOrder2) {
@@ -1456,85 +1412,6 @@ TEST_F(KeyStringBuilderTest, NaNs) {
     ASSERT(toBson(ks3d, ONE_DESCENDING)[""].Decimal().isNaN());
     ASSERT(toBson(ks4d, ONE_DESCENDING)[""].Decimal().isNaN());
 }
-TEST_F(KeyStringBuilderTest, NumberOrderLots) {
-    std::vector<BSONObj> numbers;
-    {
-        numbers.push_back(BSON("" << 0));
-        numbers.push_back(BSON("" << 0.0));
-        numbers.push_back(BSON("" << -0.0));
-
-        numbers.push_back(BSON("" << std::numeric_limits<long long>::min()));
-        numbers.push_back(BSON("" << std::numeric_limits<long long>::max()));
-        numbers.push_back(BSON("" << static_cast<double>(std::numeric_limits<long long>::min())));
-        numbers.push_back(BSON("" << static_cast<double>(std::numeric_limits<long long>::max())));
-        numbers.push_back(BSON("" << std::numeric_limits<double>::min()));
-        numbers.push_back(BSON("" << std::numeric_limits<double>::max()));
-        numbers.push_back(BSON("" << std::numeric_limits<int>::min()));
-        numbers.push_back(BSON("" << std::numeric_limits<int>::max()));
-        numbers.push_back(BSON("" << std::numeric_limits<short>::min()));
-        numbers.push_back(BSON("" << std::numeric_limits<short>::max()));
-
-        for (int i = 0; i < 64; i++) {
-            int64_t x = 1LL << i;
-            numbers.push_back(BSON("" << static_cast<long long>(x)));
-            numbers.push_back(BSON("" << static_cast<int>(x)));
-            numbers.push_back(BSON("" << static_cast<double>(x)));
-            numbers.push_back(BSON("" << (static_cast<double>(x) + .1)));
-
-            numbers.push_back(BSON("" << (static_cast<long long>(x) + 1)));
-            numbers.push_back(BSON("" << (static_cast<int>(x) + 1)));
-            numbers.push_back(BSON("" << (static_cast<double>(x) + 1)));
-            numbers.push_back(BSON("" << (static_cast<double>(x) + 1.1)));
-
-            // Avoid negating signed integral minima
-            if (i < 63)
-                numbers.push_back(BSON("" << -static_cast<long long>(x)));
-
-            if (i < 31)
-                numbers.push_back(BSON("" << -static_cast<int>(x)));
-
-            numbers.push_back(BSON("" << -static_cast<double>(x)));
-            numbers.push_back(BSON("" << -(static_cast<double>(x) + .1)));
-
-            numbers.push_back(BSON("" << -(static_cast<long long>(x) + 1)));
-            numbers.push_back(BSON("" << -(static_cast<int>(x) + 1)));
-            numbers.push_back(BSON("" << -(static_cast<double>(x) + 1)));
-            numbers.push_back(BSON("" << -(static_cast<double>(x) + 1.1)));
-        }
-
-        for (double i = 0; i < 1000; i++) {
-            double x = pow(2.1, i);
-            numbers.push_back(BSON("" << x));
-        }
-    }
-
-    Ordering ordering = Ordering::make(BSON("a" << 1));
-
-    std::vector<std::unique_ptr<KeyString::Builder>> KeyStringBuilders;
-    for (size_t i = 0; i < numbers.size(); i++) {
-        KeyStringBuilders.push_back(
-            std::make_unique<KeyString::Builder>(version, numbers[i], ordering));
-    }
-
-    for (size_t i = 0; i < numbers.size(); i++) {
-        for (size_t j = 0; j < numbers.size(); j++) {
-            const KeyString::Builder& a = *KeyStringBuilders[i];
-            const KeyString::Builder& b = *KeyStringBuilders[j];
-            ASSERT_EQUALS(a.compare(b), -b.compare(a));
-
-            if (a.compare(b) !=
-                compareNumbers(numbers[i].firstElement(), numbers[j].firstElement())) {
-                LOGV2(22235,
-                      "{numbers_i} {numbers_j}",
-                      "numbers_i"_attr = numbers[i],
-                      "numbers_j"_attr = numbers[j]);
-            }
-
-            ASSERT_EQUALS(a.compare(b),
-                          compareNumbers(numbers[i].firstElement(), numbers[j].firstElement()));
-        }
-    }
-}
 
 TEST_F(KeyStringBuilderTest, RecordIds) {
     for (int i = 0; i < 63; i++) {
@@ -1545,16 +1422,16 @@ TEST_F(KeyStringBuilderTest, RecordIds) {
             ASSERT_GTE(ks.getSize(), 2u);
             ASSERT_LTE(ks.getSize(), 10u);
 
-            ASSERT_EQ(KeyString::decodeRecordIdAtEnd(ks.getBuffer(), ks.getSize()), rid);
+            ASSERT_EQ(KeyString::decodeRecordIdLongAtEnd(ks.getBuffer(), ks.getSize()), rid);
 
             {
                 BufReader reader(ks.getBuffer(), ks.getSize());
-                ASSERT_EQ(KeyString::decodeRecordId(&reader), rid);
+                ASSERT_EQ(KeyString::decodeRecordIdLong(&reader), rid);
                 ASSERT(reader.atEof());
             }
 
             if (rid.isValid()) {
-                ASSERT_GT(ks, KeyString::Builder(version, RecordId()));
+                ASSERT_GTE(ks, KeyString::Builder(version, RecordId(1)));
                 ASSERT_GT(ks, KeyString::Builder(version, RecordId::minLong()));
                 ASSERT_LT(ks, KeyString::Builder(version, RecordId::maxLong()));
 
@@ -1587,23 +1464,81 @@ TEST_F(KeyStringBuilderTest, RecordIds) {
                 ks.appendRecordId(rid);
                 ks.appendRecordId(other);
 
-                ASSERT_EQ(KeyString::decodeRecordIdAtEnd(ks.getBuffer(), ks.getSize()), other);
+                ASSERT_EQ(KeyString::decodeRecordIdLongAtEnd(ks.getBuffer(), ks.getSize()), other);
 
                 // forward scan
                 BufReader reader(ks.getBuffer(), ks.getSize());
-                ASSERT_EQ(KeyString::decodeRecordId(&reader), RecordId::maxLong());
-                ASSERT_EQ(KeyString::decodeRecordId(&reader), rid);
-                ASSERT_EQ(KeyString::decodeRecordId(&reader), RecordId(0xDEADBEEF));
-                ASSERT_EQ(KeyString::decodeRecordId(&reader), rid);
-                ASSERT_EQ(KeyString::decodeRecordId(&reader), RecordId(1));
-                ASSERT_EQ(KeyString::decodeRecordId(&reader), rid);
-                ASSERT_EQ(KeyString::decodeRecordId(&reader), other);
+                ASSERT_EQ(KeyString::decodeRecordIdLong(&reader), RecordId::maxLong());
+                ASSERT_EQ(KeyString::decodeRecordIdLong(&reader), rid);
+                ASSERT_EQ(KeyString::decodeRecordIdLong(&reader), RecordId(0xDEADBEEF));
+                ASSERT_EQ(KeyString::decodeRecordIdLong(&reader), rid);
+                ASSERT_EQ(KeyString::decodeRecordIdLong(&reader), RecordId(1));
+                ASSERT_EQ(KeyString::decodeRecordIdLong(&reader), rid);
+                ASSERT_EQ(KeyString::decodeRecordIdLong(&reader), other);
                 ASSERT(reader.atEof());
             }
         }
     }
 }
 
+TEST_F(KeyStringBuilderTest, RecordIdStr) {
+    const int kSize = 12;
+    for (int i = 0; i < kSize; i++) {
+        char buf[kSize];
+        memset(buf, 0x80, kSize);
+        buf[i] = 0xFE;
+        const RecordId rid = RecordId(buf, kSize);
+
+        {  // Test encoding / decoding of single RecordIds
+            const KeyString::Builder ks(version, rid);
+            ASSERT_EQ(ks.getSize(), 12u);
+
+            ASSERT_EQ(KeyString::decodeRecordIdStrAtEnd(ks.getBuffer(), ks.getSize()), rid);
+
+            {
+                BufReader reader(ks.getBuffer(), ks.getSize());
+                ASSERT_EQ(KeyString::decodeRecordIdStr(&reader), rid);
+                ASSERT(reader.atEof());
+            }
+
+            if (rid.isValid()) {
+                ASSERT_GT(ks, KeyString::Builder(version, RecordId(1)));
+                ASSERT_GT(
+                    ks, KeyString::Builder(version, RecordId(OID().view().view(), OID::kOIDSize)));
+                ASSERT_LT(
+                    ks,
+                    KeyString::Builder(version, RecordId(OID::max().view().view(), OID::kOIDSize)));
+
+                char bufLt[kSize];
+                memcpy(bufLt, buf, kSize);
+                bufLt[kSize - 1] -= 1;
+                ASSERT_GT(ks, KeyString::Builder(version, RecordId(bufLt, kSize)));
+
+                char bufGt[kSize];
+                memcpy(bufGt, buf, kSize);
+                bufGt[kSize - 1] += 1;
+                ASSERT_LT(ks, KeyString::Builder(version, RecordId(bufGt, kSize)));
+            }
+        }
+
+        for (int j = 0; j < kSize; j++) {
+            char otherBuf[kSize] = {0};
+            otherBuf[j] = 0xFE;
+            RecordId other = RecordId(otherBuf, kSize);
+
+            if (rid == other) {
+                ASSERT_EQ(KeyString::Builder(version, rid), KeyString::Builder(version, other));
+            }
+            if (rid < other) {
+                ASSERT_LT(KeyString::Builder(version, rid), KeyString::Builder(version, other));
+            }
+            if (rid > other) {
+                ASSERT_GT(KeyString::Builder(version, rid), KeyString::Builder(version, other));
+            }
+        }
+    }
+}
+
 TEST_F(KeyStringBuilderTest, KeyWithLotsOfTypeBits) {
     BSONObj obj;
     {
@@ -1975,3 +1910,129 @@ DEATH_TEST(KeyStringBuilderTest, ToBsonPromotesAssertionsToTerminate, "terminate
     KeyString::TypeBits typeBits(KeyString::Version::V1);
     KeyString::toBson(invalidString, sizeof(invalidString), ALL_ASCENDING, typeBits);
 }
+
+// The following tests run last because they take a very long time.
+
+TEST_F(KeyStringBuilderTest, LotsOfNumbers3) {
+    std::vector<stdx::future<void>> futures;
+
+    for (double k = 0; k < 8; k++) {
+        futures.push_back(stdx::async(stdx::launch::async, [k, this] {
+            for (double i = -1100; i < 1100; i++) {
+                for (double j = 0; j < 52; j++) {
+                    const auto V1 = KeyString::Version::V1;
+                    Decimal128::RoundingPrecision roundingPrecisions[]{
+                        Decimal128::kRoundTo15Digits, Decimal128::kRoundTo34Digits};
+                    Decimal128::RoundingMode roundingModes[]{Decimal128::kRoundTowardNegative,
+                                                             Decimal128::kRoundTowardPositive};
+                    double x = pow(2, i);
+                    double y = pow(2, i - j);
+                    double z = pow(2, i - 53 + k);
+                    double bin = x + y - z;
+
+                    // In general NaNs don't roundtrip as we only store a single NaN, see the NaNs
+                    // test.
+                    if (std::isnan(bin))
+                        continue;
+
+                    ROUNDTRIP(version, BSON("" << bin));
+                    ROUNDTRIP(version, BSON("" << -bin));
+
+                    if (version < V1)
+                        continue;
+
+                    for (auto precision : roundingPrecisions) {
+                        for (auto mode : roundingModes) {
+                            Decimal128 rounded = Decimal128(bin, precision, mode);
+                            ROUNDTRIP(V1, BSON("" << rounded));
+                            ROUNDTRIP(V1, BSON("" << rounded.negate()));
+                        }
+                    }
+                }
+            }
+        }));
+    }
+    for (auto&& future : futures) {
+        future.get();
+    }
+}
+
+TEST_F(KeyStringBuilderTest, NumberOrderLots) {
+    std::vector<BSONObj> numbers;
+    {
+        numbers.push_back(BSON("" << 0));
+        numbers.push_back(BSON("" << 0.0));
+        numbers.push_back(BSON("" << -0.0));
+
+        numbers.push_back(BSON("" << std::numeric_limits<long long>::min()));
+        numbers.push_back(BSON("" << std::numeric_limits<long long>::max()));
+        numbers.push_back(BSON("" << static_cast<double>(std::numeric_limits<long long>::min())));
+        numbers.push_back(BSON("" << static_cast<double>(std::numeric_limits<long long>::max())));
+        numbers.push_back(BSON("" << std::numeric_limits<double>::min()));
+        numbers.push_back(BSON("" << std::numeric_limits<double>::max()));
+        numbers.push_back(BSON("" << std::numeric_limits<int>::min()));
+        numbers.push_back(BSON("" << std::numeric_limits<int>::max()));
+        numbers.push_back(BSON("" << std::numeric_limits<short>::min()));
+        numbers.push_back(BSON("" << std::numeric_limits<short>::max()));
+
+        for (int i = 0; i < 64; i++) {
+            int64_t x = 1LL << i;
+            numbers.push_back(BSON("" << static_cast<long long>(x)));
+            numbers.push_back(BSON("" << static_cast<int>(x)));
+            numbers.push_back(BSON("" << static_cast<double>(x)));
+            numbers.push_back(BSON("" << (static_cast<double>(x) + .1)));
+
+            numbers.push_back(BSON("" << (static_cast<long long>(x) + 1)));
+            numbers.push_back(BSON("" << (static_cast<int>(x) + 1)));
+            numbers.push_back(BSON("" << (static_cast<double>(x) + 1)));
+            numbers.push_back(BSON("" << (static_cast<double>(x) + 1.1)));
+
+            // Avoid negating signed integral minima
+            if (i < 63)
+                numbers.push_back(BSON("" << -static_cast<long long>(x)));
+
+            if (i < 31)
+                numbers.push_back(BSON("" << -static_cast<int>(x)));
+
+            numbers.push_back(BSON("" << -static_cast<double>(x)));
+            numbers.push_back(BSON("" << -(static_cast<double>(x) + .1)));
+
+            numbers.push_back(BSON("" << -(static_cast<long long>(x) + 1)));
+            numbers.push_back(BSON("" << -(static_cast<int>(x) + 1)));
+            numbers.push_back(BSON("" << -(static_cast<double>(x) + 1)));
+            numbers.push_back(BSON("" << -(static_cast<double>(x) + 1.1)));
+        }
+
+        for (double i = 0; i < 1000; i++) {
+            double x = pow(2.1, i);
+            numbers.push_back(BSON("" << x));
+        }
+    }
+
+    Ordering ordering = Ordering::make(BSON("a" << 1));
+
+    std::vector<std::unique_ptr<KeyString::Builder>> KeyStringBuilders;
+    for (size_t i = 0; i < numbers.size(); i++) {
+        KeyStringBuilders.push_back(
+            std::make_unique<KeyString::Builder>(version, numbers[i], ordering));
+    }
+
+    for (size_t i = 0; i < numbers.size(); i++) {
+        for (size_t j = 0; j < numbers.size(); j++) {
+            const KeyString::Builder& a = *KeyStringBuilders[i];
+            const KeyString::Builder& b = *KeyStringBuilders[j];
+            ASSERT_EQUALS(a.compare(b), -b.compare(a));
+
+            if (a.compare(b) !=
+                compareNumbers(numbers[i].firstElement(), numbers[j].firstElement())) {
+                LOGV2(22235,
+                      "{numbers_i} {numbers_j}",
+                      "numbers_i"_attr = numbers[i],
+                      "numbers_j"_attr = numbers[j]);
+            }
+
+            ASSERT_EQUALS(a.compare(b),
+                          compareNumbers(numbers[i].firstElement(), numbers[j].firstElement()));
+        }
+    }
+}
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 1caf424d755..cda8abbd0b2 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
@@ -63,7 +63,10 @@ public:
         return {};
     }
     std::unique_ptr<SortedDataInterface> getSortedDataInterface(
-        OperationContext* opCtx, StringData ident, const IndexDescriptor* desc) override {
+        OperationContext* opCtx,
+        const CollectionOptions& collOptions,
+        StringData ident,
+        const IndexDescriptor* desc) override {
         return nullptr;
     }
     Status createRecordStore(OperationContext* opCtx,
diff --git a/src/mongo/db/storage/kv/kv_engine.h b/src/mongo/db/storage/kv/kv_engine.h
index 7fbfd14e642..7d5266ec468 100644
--- a/src/mongo/db/storage/kv/kv_engine.h
+++ b/src/mongo/db/storage/kv/kv_engine.h
@@ -82,7 +82,10 @@ public:
                                                         const CollectionOptions& options) = 0;
 
     virtual std::unique_ptr<SortedDataInterface> getSortedDataInterface(
-        OperationContext* opCtx, StringData ident, const IndexDescriptor* desc) = 0;
+        OperationContext* opCtx,
+        const CollectionOptions& collOptions,
+        StringData ident,
+        const IndexDescriptor* desc) = 0;
 
     /**
      * The create and drop methods on KVEngine are not transactional. Transactional semantics
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 e7129872570..54e3c8a8e86 100644
--- a/src/mongo/db/storage/kv/kv_engine_test_harness.cpp
+++ b/src/mongo/db/storage/kv/kv_engine_test_harness.cpp
@@ -241,7 +241,7 @@ TEST(KVEngineTestHarness, SimpleSorted1) {
     {
         MyOperationContext opCtx(engine);
         ASSERT_OK(engine->createSortedDataInterface(&opCtx, CollectionOptions(), ident, &desc));
-        sorted = engine->getSortedDataInterface(&opCtx, ident, &desc);
+        sorted = engine->getSortedDataInterface(&opCtx, CollectionOptions(), ident, &desc);
         ASSERT(sorted);
     }
 
@@ -1471,7 +1471,7 @@ DEATH_TEST_REGEX_F(DurableCatalogImplTest,
         auto clientAndCtx = makeClientAndCtx("opCtx");
         auto opCtx = clientAndCtx.opCtx();
         ASSERT_OK(engine->createSortedDataInterface(opCtx, CollectionOptions(), ident, &desc));
-        sorted = engine->getSortedDataInterface(opCtx, ident, &desc);
+        sorted = engine->getSortedDataInterface(opCtx, CollectionOptions(), ident, &desc);
         ASSERT(sorted);
     }
 }
diff --git a/src/mongo/db/storage/sorted_data_interface_test_harness.h b/src/mongo/db/storage/sorted_data_interface_test_harness.h
index c099429a603..7e5b7a06790 100644
--- a/src/mongo/db/storage/sorted_data_interface_test_harness.h
+++ b/src/mongo/db/storage/sorted_data_interface_test_harness.h
@@ -90,7 +90,12 @@ class RecoveryUnit;
 class SortedDataInterfaceHarnessHelper : public virtual HarnessHelper {
 public:
     virtual std::unique_ptr<SortedDataInterface> newSortedDataInterface(bool unique,
-                                                                        bool partial) = 0;
+                                                                        bool partial,
+                                                                        KeyFormat keyFormat) = 0;
+
+    std::unique_ptr<SortedDataInterface> newSortedDataInterface(bool unique, bool partial) {
+        return newSortedDataInterface(unique, partial, KeyFormat::Long);
+    }
 
     virtual std::unique_ptr<SortedDataInterface> newIdIndexSortedDataInterface() = 0;
     /**
diff --git a/src/mongo/db/storage/sorted_data_interface_test_keyformat_string.cpp b/src/mongo/db/storage/sorted_data_interface_test_keyformat_string.cpp
new file mode 100644
index 00000000000..4b6c465e6a6
--- /dev/null
+++ b/src/mongo/db/storage/sorted_data_interface_test_keyformat_string.cpp
@@ -0,0 +1,283 @@
+/**
+ *    Copyright (C) 2020-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#include "mongo/db/storage/sorted_data_interface_test_harness.h"
+
+#include <memory>
+
+#include "mongo/db/storage/key_string.h"
+#include "mongo/db/storage/sorted_data_interface.h"
+#include "mongo/unittest/unittest.h"
+
+namespace mongo {
+namespace {
+
+TEST(SortedDataInterface, KeyFormatStringInsertDuplicates) {
+    const auto harnessHelper(newSortedDataInterfaceHarnessHelper());
+    const std::unique_ptr<SortedDataInterface> sorted(harnessHelper->newSortedDataInterface(
+        /*unique=*/false, /*partial=*/false, KeyFormat::String));
+    if (!sorted) {
+        // Not supported by this storage engine.
+        return;
+    }
+    const ServiceContext::UniqueOperationContext opCtx(harnessHelper->newOperationContext());
+    ASSERT(sorted->isEmpty(opCtx.get()));
+
+    char buf1[12];
+    memset(buf1, 0, 12);
+    char buf2[12];
+    memset(buf2, 1, 12);
+    char buf3[12];
+    memset(buf3, 0xff, 12);
+
+    RecordId rid1(buf1, 12);
+    RecordId rid2(buf2, 12);
+    RecordId rid3(buf3, 12);
+
+    {
+        WriteUnitOfWork uow(opCtx.get());
+        ASSERT_OK(sorted->insert(opCtx.get(),
+                                 makeKeyString(sorted.get(), key1, rid1),
+                                 /*dupsAllowed*/ true));
+        ASSERT_OK(sorted->insert(opCtx.get(),
+                                 makeKeyString(sorted.get(), key1, rid2),
+                                 /*dupsAllowed*/ true));
+        ASSERT_OK(sorted->insert(opCtx.get(),
+                                 makeKeyString(sorted.get(), key1, rid3),
+                                 /*dupsAllowed*/ true));
+        uow.commit();
+    }
+    ASSERT_EQUALS(3, sorted->numEntries(opCtx.get()));
+
+    auto ksSeek = makeKeyStringForSeek(sorted.get(), key1, true, true);
+    {
+        auto cursor = sorted->newCursor(opCtx.get());
+        auto entry = cursor->seek(ksSeek);
+        ASSERT(entry);
+        ASSERT_EQ(*entry, IndexKeyEntry(key1, rid1));
+
+        entry = cursor->next();
+        ASSERT(entry);
+        ASSERT_EQ(*entry, IndexKeyEntry(key1, rid2));
+
+        entry = cursor->next();
+        ASSERT(entry);
+        ASSERT_EQ(*entry, IndexKeyEntry(key1, rid3));
+    }
+
+    {
+        auto cursor = sorted->newCursor(opCtx.get());
+        auto entry = cursor->seekForKeyString(ksSeek);
+        ASSERT(entry);
+        ASSERT_EQ(entry->loc, rid1);
+        auto ks1 = makeKeyString(sorted.get(), key1, rid1);
+        ASSERT_EQ(entry->keyString, ks1);
+
+        entry = cursor->nextKeyString();
+        ASSERT(entry);
+        ASSERT_EQ(entry->loc, rid2);
+        auto ks2 = makeKeyString(sorted.get(), key1, rid2);
+        ASSERT_EQ(entry->keyString, ks2);
+
+        entry = cursor->nextKeyString();
+        ASSERT(entry);
+        ASSERT_EQ(entry->loc, rid3);
+        auto ks3 = makeKeyString(sorted.get(), key1, rid3);
+        ASSERT_EQ(entry->keyString, ks3);
+    }
+}
+
+TEST(SortedDataInterface, KeyFormatStringSetEndPosition) {
+    const auto harnessHelper(newSortedDataInterfaceHarnessHelper());
+    const std::unique_ptr<SortedDataInterface> sorted(harnessHelper->newSortedDataInterface(
+        /*unique=*/false, /*partial=*/false, KeyFormat::String));
+    if (!sorted) {
+        // Not supported by this storage engine.
+        return;
+    }
+    const ServiceContext::UniqueOperationContext opCtx(harnessHelper->newOperationContext());
+    ASSERT(sorted->isEmpty(opCtx.get()));
+
+    char buf1[12];
+    memset(buf1, 0, 12);
+    char buf2[12];
+    memset(buf2, 1, 12);
+    char buf3[12];
+    memset(buf3, 0xff, 12);
+
+    RecordId rid1(buf1, 12);
+    RecordId rid2(buf2, 12);
+    RecordId rid3(buf3, 12);
+
+    {
+        WriteUnitOfWork uow(opCtx.get());
+        ASSERT_OK(sorted->insert(opCtx.get(),
+                                 makeKeyString(sorted.get(), key1, rid1),
+                                 /*dupsAllowed*/ true));
+        ASSERT_OK(sorted->insert(opCtx.get(),
+                                 makeKeyString(sorted.get(), key2, rid2),
+                                 /*dupsAllowed*/ true));
+        ASSERT_OK(sorted->insert(opCtx.get(),
+                                 makeKeyString(sorted.get(), key3, rid3),
+                                 /*dupsAllowed*/ true));
+        uow.commit();
+    }
+    ASSERT_EQUALS(3, sorted->numEntries(opCtx.get()));
+
+    // Seek for first key only
+    {
+        auto ksSeek = makeKeyStringForSeek(sorted.get(), key1, true, true);
+        auto cursor = sorted->newCursor(opCtx.get());
+        cursor->setEndPosition(key1, true /* inclusive */);
+        auto entry = cursor->seek(ksSeek);
+        ASSERT(entry);
+        ASSERT_EQ(*entry, IndexKeyEntry(key1, rid1));
+        ASSERT_FALSE(cursor->next());
+    }
+
+    // Seek for second key from first
+    {
+        auto ksSeek = makeKeyStringForSeek(sorted.get(), key1, true, true);
+        auto cursor = sorted->newCursor(opCtx.get());
+        cursor->setEndPosition(key2, true /* inclusive */);
+        auto entry = cursor->seek(ksSeek);
+        ASSERT(entry);
+        entry = cursor->next();
+        ASSERT(entry);
+        ASSERT_EQ(*entry, IndexKeyEntry(key2, rid2));
+        ASSERT_FALSE(cursor->next());
+    }
+
+    // Seek starting from the second, don't include the last key.
+    {
+        auto ksSeek = makeKeyStringForSeek(sorted.get(), key2, true, true);
+        auto cursor = sorted->newCursor(opCtx.get());
+        cursor->setEndPosition(key3, false /* inclusive */);
+        auto entry = cursor->seek(ksSeek);
+        ASSERT(entry);
+        ASSERT_EQ(*entry, IndexKeyEntry(key2, rid2));
+        ASSERT_FALSE(cursor->next());
+    }
+}
+
+TEST(SortedDataInterface, KeyFormatStringInsertReserved) {
+    const auto harnessHelper(newSortedDataInterfaceHarnessHelper());
+    const std::unique_ptr<SortedDataInterface> sorted(harnessHelper->newSortedDataInterface(
+        /*unique=*/false, /*partial=*/false, KeyFormat::String));
+    if (!sorted) {
+        // Not supported by this storage engine.
+        return;
+    }
+    const ServiceContext::UniqueOperationContext opCtx(harnessHelper->newOperationContext());
+    ASSERT(sorted->isEmpty(opCtx.get()));
+
+    RecordId reservedLoc(
+        RecordId::reservedIdFor<OID>(RecordId::Reservation::kWildcardMultikeyMetadataId));
+    invariant(RecordId::isReserved<OID>(reservedLoc));
+    {
+        WriteUnitOfWork uow(opCtx.get());
+        ASSERT_OK(sorted->insert(opCtx.get(),
+                                 makeKeyString(sorted.get(), key1, reservedLoc),
+                                 /*dupsAllowed*/ true));
+        uow.commit();
+    }
+    ASSERT_EQUALS(1, sorted->numEntries(opCtx.get()));
+
+    auto ksSeek = makeKeyStringForSeek(sorted.get(), key1, true, true);
+    {
+        auto cursor = sorted->newCursor(opCtx.get());
+        auto entry = cursor->seek(ksSeek);
+        ASSERT(entry);
+        ASSERT_EQ(*entry, IndexKeyEntry(key1, reservedLoc));
+    }
+
+    {
+        auto cursor = sorted->newCursor(opCtx.get());
+        auto entry = cursor->seekForKeyString(ksSeek);
+        ASSERT(entry);
+        ASSERT_EQ(entry->loc, reservedLoc);
+        auto ks1 = makeKeyString(sorted.get(), key1, reservedLoc);
+        ASSERT_EQ(entry->keyString, ks1);
+    }
+}
+
+TEST(SortedDataInterface, KeyFormatStringUnindex) {
+    const auto harnessHelper(newSortedDataInterfaceHarnessHelper());
+    const std::unique_ptr<SortedDataInterface> sorted(harnessHelper->newSortedDataInterface(
+        /*unique=*/false, /*partial=*/false, KeyFormat::String));
+    if (!sorted) {
+        // Not supported by this storage engine.
+        return;
+    }
+    const ServiceContext::UniqueOperationContext opCtx(harnessHelper->newOperationContext());
+    ASSERT(sorted->isEmpty(opCtx.get()));
+
+    char buf1[12];
+    memset(buf1, 0, 12);
+    char buf2[12];
+    memset(buf2, 1, 12);
+    char buf3[12];
+    memset(buf3, 0xff, 12);
+
+    RecordId rid1(buf1, 12);
+    RecordId rid2(buf2, 12);
+    RecordId rid3(buf3, 12);
+
+    {
+        WriteUnitOfWork uow(opCtx.get());
+        ASSERT_OK(sorted->insert(opCtx.get(),
+                                 makeKeyString(sorted.get(), key1, rid1),
+                                 /*dupsAllowed*/ true));
+        ASSERT_OK(sorted->insert(opCtx.get(),
+                                 makeKeyString(sorted.get(), key1, rid2),
+                                 /*dupsAllowed*/ true));
+        ASSERT_OK(sorted->insert(opCtx.get(),
+                                 makeKeyString(sorted.get(), key1, rid3),
+                                 /*dupsAllowed*/ true));
+        uow.commit();
+    }
+    ASSERT_EQUALS(3, sorted->numEntries(opCtx.get()));
+
+    {
+        WriteUnitOfWork uow(opCtx.get());
+        sorted->unindex(opCtx.get(),
+                        makeKeyString(sorted.get(), key1, rid1),
+                        /*dupsAllowed*/ true);
+        sorted->unindex(opCtx.get(),
+                        makeKeyString(sorted.get(), key1, rid2),
+                        /*dupsAllowed*/ true);
+        sorted->unindex(opCtx.get(),
+                        makeKeyString(sorted.get(), key1, rid3),
+                        /*dupsAllowed*/ true);
+        uow.commit();
+    }
+    ASSERT_EQUALS(0, sorted->numEntries(opCtx.get()));
+}
+}  // namespace
+}  // namespace mongo
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_index.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_index.cpp
index 61e2bbdcbab..87bda0ace95 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_index.cpp
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_index.cpp
@@ -238,6 +238,7 @@ int WiredTigerIndex::Drop(OperationContext* opCtx, const std::string& uri) {
 WiredTigerIndex::WiredTigerIndex(OperationContext* ctx,
                                  const std::string& uri,
                                  StringData ident,
+                                 KeyFormat rsKeyFormat,
                                  const IndexDescriptor* desc,
                                  bool isReadOnly)
     : SortedDataInterface(ident,
@@ -248,17 +249,34 @@ WiredTigerIndex::WiredTigerIndex(OperationContext* ctx,
       _desc(desc),
       _indexName(desc->indexName()),
       _keyPattern(desc->keyPattern()),
-      _collation(desc->collation()) {}
+      _collation(desc->collation()),
+      _rsKeyFormat(rsKeyFormat) {}
 
 NamespaceString WiredTigerIndex::getCollectionNamespace(OperationContext* opCtx) const {
     return _desc->getEntry()->getNSSFromCatalog(opCtx);
 }
 
+namespace {
+void dassertRecordIdAtEnd(const KeyString::Value& keyString, KeyFormat keyFormat) {
+    if (!kDebugBuild) {
+        return;
+    }
+
+    RecordId rid;
+    if (keyFormat == KeyFormat::Long) {
+        rid = KeyString::decodeRecordIdLongAtEnd(keyString.getBuffer(), keyString.getSize());
+    } else {
+        rid = KeyString::decodeRecordIdStrAtEnd(keyString.getBuffer(), keyString.getSize());
+    }
+    invariant(rid.isValid(), rid.toString());
+}
+}  // namespace
+
 Status WiredTigerIndex::insert(OperationContext* opCtx,
                                const KeyString::Value& keyString,
                                bool dupsAllowed) {
     dassert(opCtx->lockState()->isWriteLocked());
-    dassert(KeyString::decodeRecordIdAtEnd(keyString.getBuffer(), keyString.getSize()).isValid());
+    dassertRecordIdAtEnd(keyString, _rsKeyFormat);
 
     LOGV2_TRACE_INDEX(20093, "KeyString: {keyString}", "keyString"_attr = keyString);
 
@@ -273,7 +291,7 @@ void WiredTigerIndex::unindex(OperationContext* opCtx,
                               const KeyString::Value& keyString,
                               bool dupsAllowed) {
     dassert(opCtx->lockState()->isWriteLocked());
-    dassert(KeyString::decodeRecordIdAtEnd(keyString.getBuffer(), keyString.getSize()).isValid());
+    dassertRecordIdAtEnd(keyString, _rsKeyFormat);
 
     WiredTigerCursor curwrap(_uri, _tableId, false, opCtx);
     curwrap.assertInActiveTxn();
@@ -594,8 +612,7 @@ public:
         : BulkBuilder(idx, opCtx), _idx(idx) {}
 
     Status addKey(const KeyString::Value& keyString) override {
-        dassert(
-            KeyString::decodeRecordIdAtEnd(keyString.getBuffer(), keyString.getSize()).isValid());
+        dassertRecordIdAtEnd(keyString, _idx->rsKeyFormat());
 
         // Can't use WiredTigerCursor since we aren't using the cache.
         WiredTigerItem item(keyString.getBuffer(), keyString.getSize());
@@ -639,8 +656,7 @@ public:
     }
 
     Status addKey(const KeyString::Value& newKeyString) override {
-        dassert(KeyString::decodeRecordIdAtEnd(newKeyString.getBuffer(), newKeyString.getSize())
-                    .isValid());
+        dassertRecordIdAtEnd(newKeyString, KeyFormat::Long);
 
         // Do a duplicate check, but only if dups aren't allowed.
         if (!_dupsAllowed) {
@@ -701,15 +717,14 @@ public:
     }
 
     Status addKey(const KeyString::Value& newKeyString) override {
-        dassert(KeyString::decodeRecordIdAtEnd(newKeyString.getBuffer(), newKeyString.getSize())
-                    .isValid());
+        dassertRecordIdAtEnd(newKeyString, KeyFormat::Long);
 
         const int cmp = newKeyString.compareWithoutRecordId(_previousKeyString);
         // _previousKeyString.isEmpty() is only true on the first call to addKey().
         invariant(_previousKeyString.isEmpty() || cmp > 0);
 
         RecordId id =
-            KeyString::decodeRecordIdAtEnd(newKeyString.getBuffer(), newKeyString.getSize());
+            KeyString::decodeRecordIdLongAtEnd(newKeyString.getBuffer(), newKeyString.getSize());
         KeyString::TypeBits typeBits = newKeyString.getTypeBits();
 
         KeyString::Builder value(_idx->getKeyStringVersion());
@@ -928,7 +943,12 @@ protected:
     // Must not throw WriteConflictException, throwing a WriteConflictException will retry the
     // operation effectively skipping over this key.
     virtual void updateIdAndTypeBits() {
-        _id = KeyString::decodeRecordIdAtEnd(_key.getBuffer(), _key.getSize());
+        if (_idx.rsKeyFormat() == KeyFormat::Long) {
+            _id = KeyString::decodeRecordIdLongAtEnd(_key.getBuffer(), _key.getSize());
+        } else {
+            invariant(_idx.rsKeyFormat() == KeyFormat::String);
+            _id = KeyString::decodeRecordIdStrAtEnd(_key.getBuffer(), _key.getSize());
+        }
 
         WT_CURSOR* c = _cursor->get();
         WT_ITEM item;
@@ -1255,7 +1275,7 @@ private:
         invariantWTOK(ret);
 
         BufReader br(item.data, item.size);
-        _id = KeyString::decodeRecordId(&br);
+        _id = KeyString::decodeRecordIdLong(&br);
         _typeBits.resetFromBuffer(&br);
 
         if (!br.atEof()) {
@@ -1287,7 +1307,7 @@ public:
         invariantWTOK(ret);
 
         BufReader br(item.data, item.size);
-        _id = KeyString::decodeRecordId(&br);
+        _id = KeyString::decodeRecordIdLong(&br);
         _typeBits.resetFromBuffer(&br);
 
         if (!br.atEof()) {
@@ -1307,7 +1327,8 @@ WiredTigerIndexUnique::WiredTigerIndexUnique(OperationContext* ctx,
                                              StringData ident,
                                              const IndexDescriptor* desc,
                                              bool isReadOnly)
-    : WiredTigerIndex(ctx, uri, ident, desc, isReadOnly), _partial(desc->isPartial()) {
+    : WiredTigerIndex(ctx, uri, ident, KeyFormat::Long, desc, isReadOnly),
+      _partial(desc->isPartial()) {
     // _id indexes must use WiredTigerIdIndex
     invariant(!isIdIndex());
     // All unique indexes should be in the timestamp-safe format version as of version 4.2.
@@ -1414,7 +1435,7 @@ WiredTigerIdIndex::WiredTigerIdIndex(OperationContext* ctx,
                                      StringData ident,
                                      const IndexDescriptor* desc,
                                      bool isReadOnly)
-    : WiredTigerIndex(ctx, uri, ident, desc, isReadOnly) {
+    : WiredTigerIndex(ctx, uri, ident, KeyFormat::Long, desc, isReadOnly) {
     invariant(isIdIndex());
 }
 
@@ -1428,7 +1449,8 @@ Status WiredTigerIdIndex::_insert(OperationContext* opCtx,
                                   const KeyString::Value& keyString,
                                   bool dupsAllowed) {
     invariant(!dupsAllowed);
-    const RecordId id = KeyString::decodeRecordIdAtEnd(keyString.getBuffer(), keyString.getSize());
+    const RecordId id =
+        KeyString::decodeRecordIdLongAtEnd(keyString.getBuffer(), keyString.getSize());
     invariant(id.isValid());
 
     auto sizeWithoutRecordId =
@@ -1540,7 +1562,8 @@ void WiredTigerIdIndex::_unindex(OperationContext* opCtx,
                                  WT_CURSOR* c,
                                  const KeyString::Value& keyString,
                                  bool dupsAllowed) {
-    const RecordId id = KeyString::decodeRecordIdAtEnd(keyString.getBuffer(), keyString.getSize());
+    const RecordId id =
+        KeyString::decodeRecordIdLongAtEnd(keyString.getBuffer(), keyString.getSize());
     invariant(id.isValid());
 
     auto sizeWithoutRecordId =
@@ -1580,7 +1603,7 @@ void WiredTigerIdIndex::_unindex(OperationContext* opCtx,
     BufReader br(old.data, old.size);
     invariant(br.remaining());
 
-    RecordId idInIndex = KeyString::decodeRecordId(&br);
+    RecordId idInIndex = KeyString::decodeRecordIdLong(&br);
     KeyString::TypeBits typeBits = KeyString::TypeBits::fromBuffer(getKeyStringVersion(), &br);
     if (!br.atEof()) {
         auto bsonKey = KeyString::toBson(keyString, _ordering);
@@ -1649,9 +1672,10 @@ void WiredTigerIndexUnique::_unindex(OperationContext* opCtx,
 WiredTigerIndexStandard::WiredTigerIndexStandard(OperationContext* ctx,
                                                  const std::string& uri,
                                                  StringData ident,
+                                                 KeyFormat rsKeyFormat,
                                                  const IndexDescriptor* desc,
                                                  bool isReadOnly)
-    : WiredTigerIndex(ctx, uri, ident, desc, isReadOnly) {}
+    : WiredTigerIndex(ctx, uri, ident, rsKeyFormat, desc, isReadOnly) {}
 
 std::unique_ptr<SortedDataInterface::Cursor> WiredTigerIndexStandard::newCursor(
     OperationContext* opCtx, bool forward) const {
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_index.h b/src/mongo/db/storage/wiredtiger/wiredtiger_index.h
index e8058e2350f..7f2f60e9c45 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_index.h
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_index.h
@@ -89,9 +89,13 @@ public:
      */
     static int Drop(OperationContext* opCtx, const std::string& uri);
 
+    /**
+     * Constructs an index. The rsKeyFormat is the RecordId key format of the related RecordStore.
+     */
     WiredTigerIndex(OperationContext* ctx,
                     const std::string& uri,
                     StringData ident,
+                    KeyFormat rsKeyFormat,
                     const IndexDescriptor* desc,
                     bool readOnly);
 
@@ -141,6 +145,10 @@ public:
         return _keyPattern;
     }
 
+    KeyFormat rsKeyFormat() const {
+        return _rsKeyFormat;
+    }
+
     virtual bool isIdIndex() const {
         return false;
     }
@@ -190,6 +198,7 @@ protected:
     const std::string _indexName;
     const BSONObj _keyPattern;
     const BSONObj _collation;
+    const KeyFormat _rsKeyFormat;
 };
 
 class WiredTigerIndexUnique : public WiredTigerIndex {
@@ -282,6 +291,7 @@ public:
     WiredTigerIndexStandard(OperationContext* ctx,
                             const std::string& uri,
                             StringData ident,
+                            KeyFormat rsKeyFormat,
                             const IndexDescriptor* desc,
                             bool readOnly = false);
 
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.cpp
index ae00ec30fab..cc8f4b3ff2b 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.cpp
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.cpp
@@ -1568,15 +1568,22 @@ Status WiredTigerKVEngine::dropSortedDataInterface(OperationContext* opCtx, Stri
 }
 
 std::unique_ptr<SortedDataInterface> WiredTigerKVEngine::getSortedDataInterface(
-    OperationContext* opCtx, StringData ident, const IndexDescriptor* desc) {
+    OperationContext* opCtx,
+    const CollectionOptions& collOptions,
+    StringData ident,
+    const IndexDescriptor* desc) {
     if (desc->isIdIndex()) {
+        invariant(!collOptions.clusteredIndex);
         return std::make_unique<WiredTigerIdIndex>(opCtx, _uri(ident), ident, desc, _readOnly);
     }
     if (desc->unique()) {
+        invariant(!collOptions.clusteredIndex);
         return std::make_unique<WiredTigerIndexUnique>(opCtx, _uri(ident), ident, desc, _readOnly);
     }
 
-    return std::make_unique<WiredTigerIndexStandard>(opCtx, _uri(ident), ident, desc, _readOnly);
+    auto keyFormat = (collOptions.clusteredIndex) ? KeyFormat::String : KeyFormat::Long;
+    return std::make_unique<WiredTigerIndexStandard>(
+        opCtx, _uri(ident), ident, keyFormat, desc, _readOnly);
 }
 
 std::unique_ptr<RecordStore> WiredTigerKVEngine::makeTemporaryRecordStore(OperationContext* opCtx,
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.h b/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.h
index 5619449784d..b9464944cdd 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.h
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.h
@@ -152,7 +152,10 @@ public:
                                      const IndexDescriptor* desc) override;
 
     std::unique_ptr<SortedDataInterface> getSortedDataInterface(
-        OperationContext* opCtx, StringData ident, const IndexDescriptor* desc) override;
+        OperationContext* opCtx,
+        const CollectionOptions& collOptions,
+        StringData ident,
+        const IndexDescriptor* desc) override;
 
     Status importRecordStore(OperationContext* opCtx,
                              StringData ident,
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_record_store.h b/src/mongo/db/storage/wiredtiger/wiredtiger_record_store.h
index 16ebc2bc5e9..2b1d021e626 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_record_store.h
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_record_store.h
@@ -34,7 +34,6 @@
 #include <string>
 #include <wiredtiger.h>
 
-#include "mongo/db/catalog/collection_options.h"
 #include "mongo/db/storage/capped_callback.h"
 #include "mongo/db/storage/record_store.h"
 #include "mongo/db/storage/wiredtiger/wiredtiger_cursor.h"
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_standard_index_test.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_standard_index_test.cpp
index a587089e5c5..f3b2f963b53 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_standard_index_test.cpp
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_standard_index_test.cpp
@@ -92,7 +92,9 @@ public:
         return std::make_unique<WiredTigerIdIndex>(&opCtx, uri, "" /* ident */, &desc);
     }
 
-    std::unique_ptr<SortedDataInterface> newSortedDataInterface(bool unique, bool partial) final {
+    std::unique_ptr<mongo::SortedDataInterface> newSortedDataInterface(bool unique,
+                                                                       bool partial,
+                                                                       KeyFormat keyFormat) final {
         std::string ns = "test.wt";
         OperationContextNoop opCtx(newRecoveryUnit().release());
 
@@ -119,9 +121,12 @@ public:
         string uri = "table:" + ns;
         invariantWTOK(WiredTigerIndex::Create(&opCtx, uri, result.getValue()));
 
-        if (unique)
+        if (unique) {
+            invariant(keyFormat == KeyFormat::Long);
             return std::make_unique<WiredTigerIndexUnique>(&opCtx, uri, "" /* ident */, &desc);
-        return std::make_unique<WiredTigerIndexStandard>(&opCtx, uri, "" /* ident */, &desc);
+        }
+        return std::make_unique<WiredTigerIndexStandard>(
+            &opCtx, uri, "" /* ident */, keyFormat, &desc);
     }
 
     std::unique_ptr<RecoveryUnit> newRecoveryUnit() final {
diff --git a/src/mongo/dbtests/validate_tests.cpp b/src/mongo/dbtests/validate_tests.cpp
index 8ee3a96f51f..fb793a2064c 100644
--- a/src/mongo/dbtests/validate_tests.cpp
+++ b/src/mongo/dbtests/validate_tests.cpp
@@ -78,7 +78,8 @@ public:
             options.clusteredIndex = ClusteredIndexOptions{};
         }
 
-        auto coll = db->createCollection(&_opCtx, _nss, options);
+        const bool createIdIndex = !clustered;
+        auto coll = db->createCollection(&_opCtx, _nss, options, createIdIndex);
         ASSERT_TRUE(coll);
         wuow.commit();
 
diff --git a/src/mongo/shell/data_consistency_checker.js b/src/mongo/shell/data_consistency_checker.js
index 8fb7be3c634..0fcef3cca94 100644
--- a/src/mongo/shell/data_consistency_checker.js
+++ b/src/mongo/shell/data_consistency_checker.js
@@ -320,6 +320,8 @@ var {DataConsistencyChecker} = (function() {
                         // from index specs in 4.4.
                         if (sourceInfo.idIndex) {
                             delete sourceInfo.idIndex.ns;
+                        }
+                        if (syncingInfo.idIndex) {
                             delete syncingInfo.idIndex.ns;
                         }