SERVER-55664 EphemeralForTest supports clustered _id indexes

41 files changed, 393 insertions, 441 deletions

diff --git a/jstests/core/timeseries/clustered_index_options.js b/jstests/core/timeseries/clustered_index_options.js
index 8bdf755c72a..02e7f7f9ff6 100644
--- a/jstests/core/timeseries/clustered_index_options.js
+++ b/jstests/core/timeseries/clustered_index_options.js
@@ -8,7 +8,6 @@
  *     assumes_no_implicit_collection_creation_after_drop,
  *     does_not_support_stepdowns,
  *     requires_fcv_49,
- *     requires_wiredtiger,
  * ]
  */
 (function() {
@@ -21,12 +20,6 @@ if (!TimeseriesTest.timeseriesCollectionsEnabled(db.getMongo())) {
     return;
 }
 
-// Although this test is tagged with 'requires_wiredtiger', this is not sufficient for ensuring that
-// the parallel suite runs this test only on WT configurations.
-if (!TimeseriesTest.supportsClusteredIndexes(db.getMongo())) {
-    return;
-}
-
 const testDB = db.getSiblingDB(jsTestName());
 const tsColl = testDB.clustered_index_options;
 const tsCollName = tsColl.getName();
@@ -80,4 +73,4 @@ assert.commandFailedWithCode(testDB.createCollection('test', {expireAfterSeconds
                              ErrorCodes.InvalidOptions);
 assert.commandFailedWithCode(testDB.createCollection(bucketsCollName, {expireAfterSeconds: 10}),
                              ErrorCodes.InvalidOptions);
-})();
\ No newline at end of file
+})();
diff --git a/jstests/core/timeseries/clustered_index_types.js b/jstests/core/timeseries/clustered_index_types.js
index b5241af6aaf..c812d0f55ec 100644
--- a/jstests/core/timeseries/clustered_index_types.js
+++ b/jstests/core/timeseries/clustered_index_types.js
@@ -7,7 +7,6 @@
  *     does_not_support_stepdowns,
  *     requires_fcv_49,
  *     requires_find_command,
- *     requires_wiredtiger,
  * ]
  */
 
@@ -16,12 +15,6 @@
 
 load("jstests/core/timeseries/libs/timeseries.js");
 
-// Although this test is tagged with 'requires_wiredtiger', this is not sufficient for ensuring that
-// the parallel suite runs this test only on WT configurations.
-if (!TimeseriesTest.supportsClusteredIndexes(db.getMongo())) {
-    return;
-}
-
 const collName = 'system.buckets.test';
 const coll = db[collName];
 coll.drop();
diff --git a/jstests/core/timeseries/libs/timeseries.js b/jstests/core/timeseries/libs/timeseries.js
index 146951aed21..c07536c6a9a 100644
--- a/jstests/core/timeseries/libs/timeseries.js
+++ b/jstests/core/timeseries/libs/timeseries.js
@@ -10,23 +10,6 @@ var TimeseriesTest = class {
             .featureFlagTimeseriesCollection.value;
     }
 
-    static supportsClusteredIndexes(conn) {
-        if (!assert
-                 .commandWorked(conn.adminCommand(
-                     {getParameter: 1, timeseriesBucketsCollectionClusterById: 1}))
-                 .timeseriesBucketsCollectionClusterById) {
-            jsTestLog('Time-series buckets collection not clustered by _id');
-            return false;
-        }
-        if (jsTest.options().storageEngine && jsTest.options().storageEngine !== 'wiredTiger' &&
-            jsTest.options().storageEngine !== 'inMemory') {
-            jsTestLog('Time-series test running on non-WT storage engine: ' +
-                      jsTest.options().storageEngine);
-            return false;
-        }
-        return true;
-    }
-
     /**
      * Adjusts the values in 'fields' by a random amount.
      * Ensures that the new values stay in the range [0, 100].
diff --git a/jstests/core/timeseries/timeseries_bucket_index.js b/jstests/core/timeseries/timeseries_bucket_index.js
index 1847da37332..8ac8d28dea2 100644
--- a/jstests/core/timeseries/timeseries_bucket_index.js
+++ b/jstests/core/timeseries/timeseries_bucket_index.js
@@ -43,11 +43,7 @@ TimeseriesTest.run((insert) => {
 
     assert.docEq(buckets, bucketsColl.find({_id: bucketId}).toArray());
     let explain = bucketsColl.find({_id: bucketId}).explain();
-    assert(planHasStage(
-               db,
-               explain,
-               TimeseriesTest.supportsClusteredIndexes(db.getMongo()) ? "COLLSCAN" : "IDHACK"),
-           explain);
+    assert(planHasStage(db, explain, "COLLSCAN"), explain);
 
     assert.docEq(buckets, bucketsColl.find({"control.max.time": maxTime}).toArray());
     explain = bucketsColl.find({"control.max.time": minTime}).explain();
diff --git a/jstests/core/timeseries/timeseries_expire_collmod.js b/jstests/core/timeseries/timeseries_expire_collmod.js
index 75bf199e0b1..6f585959eb8 100644
--- a/jstests/core/timeseries/timeseries_expire_collmod.js
+++ b/jstests/core/timeseries/timeseries_expire_collmod.js
@@ -7,7 +7,6 @@
  *     does_not_support_stepdowns,
  *     does_not_support_transactions,
  *     requires_fcv_49,
- *     requires_wiredtiger,
  * ]
  */
 (function() {
@@ -20,12 +19,6 @@ if (!TimeseriesTest.timeseriesCollectionsEnabled(db.getMongo())) {
     return;
 }
 
-// Although this test is tagged with 'requires_wiredtiger', this is not sufficient for ensuring that
-// the parallel suite runs this test only on WT configurations.
-if (!TimeseriesTest.supportsClusteredIndexes(db.getMongo())) {
-    return;
-}
-
 const coll = db.timeseries_expire_collmod;
 coll.drop();
 
diff --git a/jstests/core/timeseries/timeseries_id_index.js b/jstests/core/timeseries/timeseries_id_index.js
index 92687c480f6..596f7ba2570 100644
--- a/jstests/core/timeseries/timeseries_id_index.js
+++ b/jstests/core/timeseries/timeseries_id_index.js
@@ -8,7 +8,6 @@
  *     requires_fcv_49,
  *     requires_find_command,
  *     requires_getmore,
- *     requires_wiredtiger,
  * ]
  */
 (function() {
@@ -21,12 +20,6 @@ if (!TimeseriesTest.timeseriesCollectionsEnabled(db.getMongo())) {
     return;
 }
 
-// Although this test is tagged with 'requires_wiredtiger', this is not sufficient for ensuring that
-// the parallel suite runs this test only on WT configurations.
-if (!TimeseriesTest.supportsClusteredIndexes(db.getMongo())) {
-    return;
-}
-
 const coll = db.timeseries_id_index;
 coll.drop();
 
diff --git a/jstests/core/timeseries/timeseries_id_range.js b/jstests/core/timeseries/timeseries_id_range.js
index 692c0fa265b..9950a81d55d 100644
--- a/jstests/core/timeseries/timeseries_id_range.js
+++ b/jstests/core/timeseries/timeseries_id_range.js
@@ -8,7 +8,6 @@
  *     requires_fcv_49,
  *     requires_find_command,
  *     requires_getmore,
- *     requires_wiredtiger,
  * ]
  */
 (function() {
@@ -17,12 +16,6 @@
 load('jstests/libs/analyze_plan.js');
 load("jstests/core/timeseries/libs/timeseries.js");
 
-// Although this test is tagged with 'requires_wiredtiger', this is not sufficient for ensuring that
-// the parallel suite runs this test only on WT configurations.
-if (!TimeseriesTest.supportsClusteredIndexes(db.getMongo())) {
-    return;
-}
-
 TimeseriesTest.run((insert) => {
     // These dates will all be inserted into individual buckets.
     const dates = [
diff --git a/jstests/core/timeseries/timeseries_index.js b/jstests/core/timeseries/timeseries_index.js
index 1de627a9645..513595d1115 100644
--- a/jstests/core/timeseries/timeseries_index.js
+++ b/jstests/core/timeseries/timeseries_index.js
@@ -226,12 +226,7 @@ TimeseriesTest.run((insert) => {
     const testCreateIndex = function(spec, options = {}) {
         const indexName = 'testCreateIndex';
         const res = coll.createIndex(spec, Object.extend({name: indexName}, options));
-        if (TimeseriesTest.supportsClusteredIndexes(db.getMongo())) {
-            assert.commandFailedWithCode(res, ErrorCodes.InvalidOptions);
-        } else {
-            assert.commandWorked(res);
-            assert.commandWorked(coll.dropIndex(indexName));
-        }
+        assert.commandFailedWithCode(res, ErrorCodes.InvalidOptions);
     };
 
     // Partial indexes are not supported on clustered time-series bucket collections.
@@ -252,9 +247,7 @@ TimeseriesTest.run((insert) => {
     const bucketsColl = db.getCollection('system.buckets.' + coll.getName());
     assert.commandWorked(bucketsColl.createIndex({not_metadata: 1}),
                          'failed to create index: ' + tojson({not_metadata: 1}));
-    assert.eq(TimeseriesTest.supportsClusteredIndexes(db.getMongo()) ? 1 : 2,
-              bucketsColl.getIndexes().length,
-              tojson(bucketsColl.getIndexes()));
+    assert.eq(1, bucketsColl.getIndexes().length, tojson(bucketsColl.getIndexes()));
     assert.eq(0, coll.getIndexes().length, tojson(coll.getIndexes()));
 });
 })();
diff --git a/jstests/core/timeseries/timeseries_index_stats.js b/jstests/core/timeseries/timeseries_index_stats.js
index b848b5f2ef7..2e24d67aec9 100644
--- a/jstests/core/timeseries/timeseries_index_stats.js
+++ b/jstests/core/timeseries/timeseries_index_stats.js
@@ -78,8 +78,7 @@ TimeseriesTest.run((insert) => {
     // Confirm that that $indexStats is indeed ignoring one index in schema translation by checking
     // $indexStats on the buckets collection.
     const bucketIndexStatsDocs = bucketsColl.aggregate([{$indexStats: {}}]).toArray();
-    assert.eq(Object.keys(indexKeys).length +
-                  (TimeseriesTest.supportsClusteredIndexes(db.getMongo()) ? 1 : 2),
+    assert.eq(Object.keys(indexKeys).length + 1,
               bucketIndexStatsDocs.length,
               tojson(bucketIndexStatsDocs));
 
diff --git a/jstests/core/timeseries/timeseries_index_use.js b/jstests/core/timeseries/timeseries_index_use.js
index 66b47fc0aee..d48a50e68dd 100644
--- a/jstests/core/timeseries/timeseries_index_use.js
+++ b/jstests/core/timeseries/timeseries_index_use.js
@@ -64,15 +64,7 @@ TimeseriesTest.run((insert) => {
         const explain = query.explain();
         const ixscan = getAggPlanStage(explain, "IXSCAN");
         assert.neq(null, ixscan, tojson(explain));
-        // TODO (SERVER-56238): Remove conditional once queries always use the 'testIndexName'
-        // index.
-        assert.eq(TimeseriesTest.supportsClusteredIndexes(db.getMongo()) ||
-                          !filter.hasOwnProperty(timeFieldName) ||
-                          !indexSpec.hasOwnProperty(timeFieldName)
-                      ? "testIndexName"
-                      : "_id_",
-                  ixscan.indexName,
-                  tojson(ixscan));
+        assert.eq("testIndexName", ixscan.indexName, tojson(ixscan));
 
         assert.commandWorked(coll.dropIndex("testIndexName"));
     };
diff --git a/jstests/core/timeseries/timeseries_show_record_id.js b/jstests/core/timeseries/timeseries_show_record_id.js
index 00ffee6d99c..4aab2d2901c 100644
--- a/jstests/core/timeseries/timeseries_show_record_id.js
+++ b/jstests/core/timeseries/timeseries_show_record_id.js
@@ -44,9 +44,7 @@ TimeseriesTest.run((insert) => {
     function checkRecordId(documents) {
         for (const document of documents) {
             assert(document.hasOwnProperty("$recordId"));
-            if (TimeseriesTest.supportsClusteredIndexes(db.getMongo())) {
-                assert(isString(document["$recordId"]));
-            }
+            assert(isString(document["$recordId"]));
         }
     }
 
diff --git a/jstests/core/timeseries/timeseries_text_geonear_disallowed.js b/jstests/core/timeseries/timeseries_text_geonear_disallowed.js
index be0eb8bcd7c..c6bf52d5aad 100644
--- a/jstests/core/timeseries/timeseries_text_geonear_disallowed.js
+++ b/jstests/core/timeseries/timeseries_text_geonear_disallowed.js
@@ -78,12 +78,8 @@ assert.commandFailedWithCode(
 
 // $text
 // Text indices are disallowed on collections clustered by _id.
-if (TimeseriesTest.supportsClusteredIndexes(db.getMongo())) {
-    assert.commandFailedWithCode(tsColl.createIndex({"tags.descr": "text"}),
-                                 ErrorCodes.InvalidOptions);
-    // Since a Text index can't be created, a $text query should fail due to a missing index.
-    assert.commandFailedWithCode(
-        assert.throws(() => tsColl.find({$text: {$search: "1"}}).itcount()),
-                     ErrorCodes.IndexNotFound);
-}
+assert.commandFailedWithCode(tsColl.createIndex({"tags.descr": "text"}), ErrorCodes.InvalidOptions);
+// Since a Text index can't be created, a $text query should fail due to a missing index.
+assert.commandFailedWithCode(assert.throws(() => tsColl.find({$text: {$search: "1"}}).itcount()),
+                                          ErrorCodes.IndexNotFound);
 })();
diff --git a/jstests/noPassthrough/timeseries_create.js b/jstests/noPassthrough/timeseries_create.js
index 182e8efba5f..3a74bff2fb7 100644
--- a/jstests/noPassthrough/timeseries_create.js
+++ b/jstests/noPassthrough/timeseries_create.js
@@ -54,22 +54,11 @@ const testOptions = function(allowed,
         const bucketsColl = collections.find(coll => coll.name == bucketsCollName);
         assert(bucketsColl, collections);
         assert.eq(bucketsColl.type, "collection", bucketsColl);
-        if (TimeseriesTest.supportsClusteredIndexes(conn)) {
-            assert(bucketsColl.options.clusteredIndex, bucketsColl);
-        }
+        assert(bucketsColl.options.clusteredIndex, bucketsColl);
         if (createOptions.expireAfterSeconds) {
-            if (TimeseriesTest.supportsClusteredIndexes(conn)) {
-                assert.eq(bucketsColl.options.expireAfterSeconds,
-                          createOptions.expireAfterSeconds,
-                          bucketsColl);
-            } else {
-                assert.docEq(testDB[collName].getIndexes(), [{
-                                 v: 2,
-                                 key: {time: 1},
-                                 name: 'control.min.time_1',
-                                 expireAfterSeconds: createOptions.expireAfterSeconds.valueOf(),
-                             }]);
-            }
+            assert.eq(bucketsColl.options.expireAfterSeconds,
+                      createOptions.expireAfterSeconds,
+                      bucketsColl);
         }
 
         assert.commandWorked(testDB.runCommand({drop: collName, writeConcern: {w: "majority"}}));
@@ -153,13 +142,9 @@ testCompatibleCreateOptions({collation: {locale: "ja"}});
 testCompatibleCreateOptions({writeConcern: {}});
 testCompatibleCreateOptions({comment: ""});
 
-const errorCodeForInvalidExpireAfterSecondsValue = TimeseriesTest.supportsClusteredIndexes(conn)
-    ? ErrorCodes.InvalidOptions
-    : ErrorCodes.CannotCreateIndex;
-testIncompatibleCreateOptions({expireAfterSeconds: NumberLong(-10)},
-                              errorCodeForInvalidExpireAfterSecondsValue);
+testIncompatibleCreateOptions({expireAfterSeconds: NumberLong(-10)}, ErrorCodes.InvalidOptions);
 testIncompatibleCreateOptions({expireAfterSeconds: NumberLong("4611686018427387904")},
-                              errorCodeForInvalidExpireAfterSecondsValue);
+                              ErrorCodes.InvalidOptions);
 testIncompatibleCreateOptions({expireAfterSeconds: ""}, ErrorCodes.TypeMismatch);
 testIncompatibleCreateOptions({capped: true, size: 100});
 testIncompatibleCreateOptions({capped: true, max: 100});
diff --git a/jstests/noPassthrough/timeseries_ttl.js b/jstests/noPassthrough/timeseries_ttl.js
index 97842c8f120..c43b2f2a034 100644
--- a/jstests/noPassthrough/timeseries_ttl.js
+++ b/jstests/noPassthrough/timeseries_ttl.js
@@ -163,15 +163,10 @@ testCase((coll, bucketsColl) => {
 
     // Make the collection TTL and expect the data to be deleted because the bucket minimum is past
     // the expiry plus the maximum bucket range.
-    if (TimeseriesTest.supportsClusteredIndexes(conn)) {
-        assert.commandWorked(testDB.runCommand({
-            collMod: 'system.buckets.ts',
-            expireAfterSeconds: expireAfterSeconds,
-        }));
-    } else {
-        assert.commandWorked(bucketsColl.createIndex({['control.min.' + timeFieldName]: 1},
-                                                     {expireAfterSeconds: expireAfterSeconds}));
-    }
+    assert.commandWorked(testDB.runCommand({
+        collMod: 'system.buckets.ts',
+        expireAfterSeconds: expireAfterSeconds,
+    }));
 
     waitForTTL();
     assert.eq(0, coll.find().itcount());
diff --git a/src/mongo/db/catalog/create_collection.cpp b/src/mongo/db/catalog/create_collection.cpp
index fa49d083b25..b59b23bb22d 100644
--- a/src/mongo/db/catalog/create_collection.cpp
+++ b/src/mongo/db/catalog/create_collection.cpp
@@ -226,8 +226,7 @@ Status _createTimeseries(OperationContext* opCtx,
             bucketsOptions.validator = validatorObj;
 
             // If possible, cluster time-series buckets collections by _id.
-            const bool useClusteredIdIndex = gTimeseriesBucketsCollectionClusterById &&
-                opCtx->getServiceContext()->getStorageEngine()->supportsClusteredIdIndex();
+            const bool useClusteredIdIndex = gTimeseriesBucketsCollectionClusterById;
             auto expireAfterSeconds = options.expireAfterSeconds;
             if (useClusteredIdIndex) {
                 if (expireAfterSeconds) {
diff --git a/src/mongo/db/catalog/validate_adaptor.cpp b/src/mongo/db/catalog/validate_adaptor.cpp
index 63b2d3aaa05..36ef648708f 100644
--- a/src/mongo/db/catalog/validate_adaptor.cpp
+++ b/src/mongo/db/catalog/validate_adaptor.cpp
@@ -234,7 +234,7 @@ void _validateKeyOrder(OperationContext* opCtx,
 
     if (unique) {
         // Unique indexes must not have duplicate keys.
-        int cmp = currKey.compareWithoutRecordId(prevKey);
+        int cmp = currKey.compareWithoutRecordIdLong(prevKey);
         if (cmp != 0) {
             return;
         }
diff --git a/src/mongo/db/index/duplicate_key_tracker.cpp b/src/mongo/db/index/duplicate_key_tracker.cpp
index 4d3d88bd075..da9f931c079 100644
--- a/src/mongo/db/index/duplicate_key_tracker.cpp
+++ b/src/mongo/db/index/duplicate_key_tracker.cpp
@@ -86,7 +86,7 @@ Status DuplicateKeyTracker::recordKey(OperationContext* opCtx, const KeyString::
     // store the TypeBits for error reporting later on. The RecordId does not need to be stored, so
     // we exclude it from the serialization.
     BufBuilder builder;
-    key.serializeWithoutRecordId(builder);
+    key.serializeWithoutRecordIdLong(builder);
 
     auto status =
         _keyConstraintsTable->rs()->insertRecord(opCtx, builder.buf(), builder.len(), Timestamp());
diff --git a/src/mongo/db/index/index_access_method.cpp b/src/mongo/db/index/index_access_method.cpp
index 2bc543f4d64..b0caada34ee 100644
--- a/src/mongo/db/index/index_access_method.cpp
+++ b/src/mongo/db/index/index_access_method.cpp
@@ -742,7 +742,7 @@ Status AbstractIndexAccessMethod::commitBulk(OperationContext* opCtx,
         // builds since this check can be expensive.
         int cmpData;
         if (_descriptor->unique()) {
-            cmpData = data.first.compareWithoutRecordId(previousKey);
+            cmpData = data.first.compareWithoutRecordIdLong(previousKey);
         }
 
         if (kDebugBuild && data.first.compare(previousKey) < 0) {
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 bf7f2e5a943..f51e580fdc6 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
@@ -70,7 +70,6 @@ Status KVEngine::createRecordStore(OperationContext* opCtx,
                                    StringData ns,
                                    StringData ident,
                                    const CollectionOptions& options) {
-    uassert(5555900, "The 'clusteredIndex' option is not supported", !options.clusteredIndex);
     stdx::lock_guard lock(_identsLock);
     _idents[ident.toString()] = true;
     return Status::OK();
@@ -87,7 +86,7 @@ Status KVEngine::importRecordStore(OperationContext* opCtx,
 std::unique_ptr<mongo::RecordStore> KVEngine::makeTemporaryRecordStore(OperationContext* opCtx,
                                                                        StringData ident) {
     std::unique_ptr<mongo::RecordStore> recordStore =
-        std::make_unique<RecordStore>("", ident, false);
+        std::make_unique<RecordStore>("", ident, KeyFormat::Long, false);
     stdx::lock_guard lock(_identsLock);
     _idents[ident.toString()] = true;
     return recordStore;
@@ -99,14 +98,16 @@ std::unique_ptr<mongo::RecordStore> KVEngine::getRecordStore(OperationContext* u
                                                              StringData ident,
                                                              const CollectionOptions& options) {
     std::unique_ptr<mongo::RecordStore> recordStore;
+    const auto keyFormat = options.clusteredIndex ? KeyFormat::String : KeyFormat::Long;
     if (options.capped) {
         recordStore = std::make_unique<RecordStore>(ns,
                                                     ident,
+                                                    keyFormat,
                                                     options.capped,
                                                     /*cappedCallback*/ nullptr,
                                                     _visibilityManager.get());
     } else {
-        recordStore = std::make_unique<RecordStore>(ns, ident, options.capped);
+        recordStore = std::make_unique<RecordStore>(ns, ident, keyFormat, options.capped);
     }
     stdx::lock_guard lock(_identsLock);
     _idents[ident.toString()] = true;
@@ -150,15 +151,20 @@ std::unique_ptr<mongo::SortedDataInterface> KVEngine::getSortedDataInterface(
     const CollectionOptions& collOptions,
     StringData ident,
     const IndexDescriptor* desc) {
-    invariant(!collOptions.clusteredIndex);
+    auto rsKeyFormat = collOptions.clusteredIndex ? KeyFormat::String : KeyFormat::Long;
+    return getSortedDataInterface(opCtx, rsKeyFormat, ident, desc);
+}
+
+std::unique_ptr<mongo::SortedDataInterface> KVEngine::getSortedDataInterface(
+    OperationContext* opCtx, KeyFormat rsKeyFormat, StringData ident, const IndexDescriptor* desc) {
     {
         stdx::lock_guard lock(_identsLock);
         _idents[ident.toString()] = false;
     }
     if (desc->unique())
-        return std::make_unique<SortedDataInterfaceUnique>(opCtx, ident, desc);
+        return std::make_unique<SortedDataInterfaceUnique>(opCtx, ident, rsKeyFormat, desc);
     else
-        return std::make_unique<SortedDataInterfaceStandard>(opCtx, ident, desc);
+        return std::make_unique<SortedDataInterfaceStandard>(opCtx, ident, rsKeyFormat, desc);
 }
 
 Status KVEngine::dropIdent(mongo::RecoveryUnit* ru,
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 0b6c9560c10..11b88951580 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
@@ -86,6 +86,12 @@ public:
 
     virtual std::unique_ptr<mongo::SortedDataInterface> getSortedDataInterface(
         OperationContext* opCtx,
+        KeyFormat rsKeyFormat,
+        StringData ident,
+        const IndexDescriptor* desc);
+
+    virtual std::unique_ptr<mongo::SortedDataInterface> getSortedDataInterface(
+        OperationContext* opCtx,
         const CollectionOptions& collOptions,
         StringData ident,
         const IndexDescriptor* desc);
diff --git a/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_kv_engine_test.cpp b/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_kv_engine_test.cpp
index 9ba140533fd..7eee134ed39 100644
--- a/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_kv_engine_test.cpp
+++ b/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_kv_engine_test.cpp
@@ -329,7 +329,7 @@ TEST_F(EphemeralForTestKVEngineTest, ReadOlderSnapshotsSimple) {
 
     // Pin oldest timestamp with a read transaction.
     OperationContextFromKVEngine pinningOldest(_engine);
-    ASSERT(!rs->findRecord(&pinningOldest, RecordId(), nullptr));
+    ASSERT(!rs->findRecord(&pinningOldest, RecordId(1), nullptr));
 
     // Set readFrom to timestamp with no committed transactions.
     Timestamp readFrom = _engine->getHistory_forTest().rbegin()->first;
diff --git a/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_record_store.cpp b/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_record_store.cpp
index 7528a4feba2..dc2ba7b4e3b 100644
--- a/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_record_store.cpp
+++ b/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_record_store.cpp
@@ -57,31 +57,45 @@ BSONObj const sample = BSON(""
                             << "s"
                             << "" << (int64_t)0);
 
-std::string createKey(StringData ident, int64_t recordId) {
-    KeyString::Builder ks(version, BSON("" << ident << "" << recordId), allAscending);
+std::string createKey(StringData ident, RecordId recordId) {
+    KeyString::Builder ks(version, BSON("" << ident), allAscending, recordId);
     return std::string(ks.getBuffer(), ks.getSize());
 }
 
-RecordId extractRecordId(const std::string& keyStr) {
-    KeyString::Builder ks(version, sample, allAscending);
-    ks.resetFromBuffer(keyStr.c_str(), keyStr.size());
-    BSONObj obj = KeyString::toBson(keyStr.c_str(), keyStr.size(), allAscending, ks.getTypeBits());
-    auto it = BSONObjIterator(obj);
-    ++it;
-    return RecordId((*it).Long());
+std::string createPrefix(StringData ident) {
+    KeyString::Builder ks(
+        version, BSON("" << ident), allAscending, KeyString::Discriminator::kExclusiveBefore);
+    return std::string(ks.getBuffer(), ks.getSize());
+}
+
+std::string createPostfix(StringData ident) {
+    KeyString::Builder ks(
+        version, BSON("" << ident), allAscending, KeyString::Discriminator::kExclusiveAfter);
+    return std::string(ks.getBuffer(), ks.getSize());
+}
+
+RecordId extractRecordId(const std::string& keyStr, KeyFormat keyFormat) {
+    if (KeyFormat::Long == keyFormat) {
+        return KeyString::decodeRecordIdLongAtEnd(keyStr.c_str(), keyStr.size());
+    } else {
+        invariant(KeyFormat::String == keyFormat);
+        return KeyString::decodeRecordIdStrAtEnd(keyStr.c_str(), keyStr.size());
+    }
 }
 }  // namespace
 
 RecordStore::RecordStore(StringData ns,
                          StringData ident,
+                         KeyFormat keyFormat,
                          bool isCapped,
                          CappedCallback* cappedCallback,
                          VisibilityManager* visibilityManager)
     : mongo::RecordStore(ns, ident),
+      _keyFormat(keyFormat),
       _isCapped(isCapped),
       _ident(getIdent().data(), getIdent().size()),
-      _prefix(createKey(_ident, std::numeric_limits<int64_t>::min())),
-      _postfix(createKey(_ident, std::numeric_limits<int64_t>::max())),
+      _prefix(createPrefix(_ident)),
+      _postfix(createPostfix(_ident)),
       _cappedCallback(cappedCallback),
       _isOplog(NamespaceString::oplog(ns)),
       _visibilityManager(visibilityManager) {}
@@ -111,7 +125,7 @@ int64_t RecordStore::storageSize(OperationContext* opCtx,
 
 bool RecordStore::findRecord(OperationContext* opCtx, const RecordId& loc, RecordData* rd) const {
     StringStore* workingCopy(RecoveryUnit::get(opCtx)->getHead());
-    auto it = workingCopy->find(createKey(_ident, loc.getLong()));
+    auto it = workingCopy->find(createKey(_ident, loc));
     if (it == workingCopy->end()) {
         return false;
     }
@@ -120,11 +134,13 @@ bool RecordStore::findRecord(OperationContext* opCtx, const RecordId& loc, Recor
 }
 
 void RecordStore::deleteRecord(OperationContext* opCtx, const RecordId& dl) {
-    _initHighestIdIfNeeded(opCtx);
+    if (KeyFormat::Long == _keyFormat) {
+        _initHighestIdIfNeeded(opCtx);
+    }
     auto ru = RecoveryUnit::get(opCtx);
     StringStore* workingCopy(ru->getHead());
     SizeAdjuster adjuster(opCtx, this);
-    invariant(workingCopy->erase(createKey(_ident, dl.getLong())));
+    invariant(workingCopy->erase(createKey(_ident, dl)));
     ru->makeDirty();
 }
 
@@ -136,23 +152,36 @@ Status RecordStore::insertRecords(OperationContext* opCtx,
     {
         SizeAdjuster adjuster(opCtx, this);
         for (auto& record : *inOutRecords) {
-            int64_t thisRecordId = record.id.getLong();
+            auto thisRecordId = record.id;
             if (_isOplog) {
                 StatusWith<RecordId> status =
                     record_id_helpers::extractKeyOptime(record.data.data(), record.data.size());
                 if (!status.isOK())
                     return status.getStatus();
-                thisRecordId = status.getValue().getLong();
-                _visibilityManager->addUncommittedRecord(opCtx, this, RecordId(thisRecordId));
+                thisRecordId = status.getValue();
+                _visibilityManager->addUncommittedRecord(opCtx, this, thisRecordId);
             } else {
                 // If the record had an id already, keep it.
-                if (thisRecordId == 0) {
-                    thisRecordId = _nextRecordId(opCtx);
+                if (KeyFormat::Long == _keyFormat && thisRecordId.isNull()) {
+                    thisRecordId = RecordId(_nextRecordId(opCtx));
                 }
             }
+            auto key = createKey(_ident, thisRecordId);
+            auto it = workingCopy->find(key);
+            if (keyFormat() == KeyFormat::String && it != workingCopy->end()) {
+                // RecordStores with the string KeyFormat implicitly expect enforcement of _id
+                // uniqueness. Generate a useful error message that is consistent with duplicate key
+                // error messages on indexes.
+                BSONObj obj = record_id_helpers::toBSONAs(thisRecordId, "");
+                return buildDupKeyErrorStatus(obj,
+                                              NamespaceString(ns()),
+                                              "" /* indexName */,
+                                              BSON("_id" << 1),
+                                              BSONObj() /* collation */);
+            }
+
             workingCopy->insert(
-                StringStore::value_type{createKey(_ident, thisRecordId),
-                                        std::string(record.data.data(), record.data.size())});
+                StringStore::value_type{key, std::string(record.data.data(), record.data.size())});
             record.id = RecordId(thisRecordId);
         }
     }
@@ -167,7 +196,7 @@ Status RecordStore::updateRecord(OperationContext* opCtx,
     StringStore* workingCopy(RecoveryUnit::get(opCtx)->getHead());
     SizeAdjuster adjuster(opCtx, this);
     {
-        std::string key = createKey(_ident, oldLocation.getLong());
+        std::string key = createKey(_ident, oldLocation);
         StringStore::const_iterator it = workingCopy->find(key);
         invariant(it != workingCopy->end());
         workingCopy->update(StringStore::value_type{key, std::string(data, len)});
@@ -232,7 +261,7 @@ void RecordStore::cappedTruncateAfter(OperationContext* opCtx, RecordId end, boo
     auto ru = RecoveryUnit::get(opCtx);
     StringStore* workingCopy(ru->getHead());
     WriteUnitOfWork wuow(opCtx);
-    const auto recordKey = createKey(_ident, end.getLong());
+    const auto recordKey = createKey(_ident, end);
     auto recordIt =
         inclusive ? workingCopy->lower_bound(recordKey) : workingCopy->upper_bound(recordKey);
     auto endIt = workingCopy->upper_bound(_postfix);
@@ -242,7 +271,7 @@ void RecordStore::cappedTruncateAfter(OperationContext* opCtx, RecordId end, boo
         if (_cappedCallback) {
             // Documents are guaranteed to have a RecordId at the end of the KeyString, unlike
             // unique indexes.
-            RecordId rid = extractRecordId(recordIt->first);
+            RecordId rid = extractRecordId(recordIt->first, _keyFormat);
             RecordData rd = RecordData(recordIt->second.c_str(), recordIt->second.length());
             uassertStatusOK(_cappedCallback->aboutToDeleteCapped(opCtx, rid, rd));
         }
@@ -342,7 +371,7 @@ boost::optional<Record> RecordStore::Cursor::next() {
     if (it != workingCopy->end() && inPrefix(it->first)) {
         _savedPosition = it->first;
         Record nextRecord;
-        nextRecord.id = RecordId(extractRecordId(it->first));
+        nextRecord.id = RecordId(extractRecordId(it->first, _rs._keyFormat));
         nextRecord.data = RecordData(it->second.c_str(), it->second.length());
 
         if (_rs._isOplog && nextRecord.id > _oplogVisibility) {
@@ -358,7 +387,7 @@ boost::optional<Record> RecordStore::Cursor::seekExact(const RecordId& id) {
     _savedPosition = boost::none;
     _lastMoveWasRestore = false;
     StringStore* workingCopy(RecoveryUnit::get(opCtx)->getHead());
-    std::string key = createKey(_rs._ident, id.getLong());
+    std::string key = createKey(_rs._ident, id);
     it = workingCopy->find(key);
 
     if (it == workingCopy->end() || !inPrefix(it->first))
@@ -387,7 +416,7 @@ boost::optional<Record> RecordStore::Cursor::seekNear(const RecordId& id) {
         return boost::none;
 
     StringStore* workingCopy{RecoveryUnit::get(opCtx)->getHead()};
-    std::string key = createKey(_rs._ident, search.getLong());
+    std::string key = createKey(_rs._ident, search);
     // We may land higher and that is fine per the API contract.
     it = workingCopy->lower_bound(key);
 
@@ -403,13 +432,13 @@ boost::optional<Record> RecordStore::Cursor::seekNear(const RecordId& id) {
     }
 
     // If we landed one higher, then per the API contract, we need to return the previous record.
-    RecordId rid = extractRecordId(it->first);
+    RecordId rid = extractRecordId(it->first, _rs._keyFormat);
     if (rid > search) {
         StringStore::const_reverse_iterator revIt(it);
         // The reverse iterator constructor positions on the next record automatically.
         if (revIt != workingCopy->rend() && inPrefix(revIt->first)) {
             it = workingCopy->lower_bound(revIt->first);
-            rid = RecordId(extractRecordId(it->first));
+            rid = RecordId(extractRecordId(it->first, _rs._keyFormat));
         }
         // Otherwise, we hit the beginning of this record store, then there is only one record and
         // we should return that.
@@ -490,7 +519,7 @@ boost::optional<Record> RecordStore::ReverseCursor::next() {
     if (it != workingCopy->rend() && inPrefix(it->first)) {
         _savedPosition = it->first;
         Record nextRecord;
-        nextRecord.id = RecordId(extractRecordId(it->first));
+        nextRecord.id = RecordId(extractRecordId(it->first, _rs._keyFormat));
         nextRecord.data = RecordData(it->second.c_str(), it->second.length());
 
         return nextRecord;
@@ -502,7 +531,7 @@ boost::optional<Record> RecordStore::ReverseCursor::seekExact(const RecordId& id
     _needFirstSeek = false;
     _savedPosition = boost::none;
     StringStore* workingCopy(RecoveryUnit::get(opCtx)->getHead());
-    std::string key = createKey(_rs._ident, id.getLong());
+    std::string key = createKey(_rs._ident, id);
     StringStore::const_iterator canFind = workingCopy->find(key);
     if (canFind == workingCopy->end() || !inPrefix(canFind->first)) {
         it = workingCopy->rend();
@@ -523,7 +552,7 @@ boost::optional<Record> RecordStore::ReverseCursor::seekNear(const RecordId& id)
         return boost::none;
 
     StringStore* workingCopy{RecoveryUnit::get(opCtx)->getHead()};
-    std::string key = createKey(_rs._ident, id.getLong());
+    std::string key = createKey(_rs._ident, id);
     it = StringStore::const_reverse_iterator(workingCopy->upper_bound(key));
 
     // Since there is at least 1 record, if we hit the beginning we need to return the only record.
@@ -537,7 +566,7 @@ boost::optional<Record> RecordStore::ReverseCursor::seekNear(const RecordId& id)
     }
 
     // If we landed lower, then per the API contract, we need to return the previous record.
-    RecordId rid = extractRecordId(it->first);
+    RecordId rid = extractRecordId(it->first, _rs._keyFormat);
     if (rid < id) {
         // This lands on the next key.
         auto fwdIt = workingCopy->upper_bound(key);
@@ -548,7 +577,7 @@ boost::optional<Record> RecordStore::ReverseCursor::seekNear(const RecordId& id)
         // we should return that.
     }
 
-    rid = RecordId(extractRecordId(it->first));
+    rid = RecordId(extractRecordId(it->first, _rs._keyFormat));
     _needFirstSeek = false;
     _savedPosition = it->first;
     return Record{rid, RecordData(it->second.c_str(), it->second.length())};
diff --git a/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_record_store.h b/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_record_store.h
index 2c02713f88a..02dd90fe4da 100644
--- a/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_record_store.h
+++ b/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_record_store.h
@@ -50,6 +50,7 @@ class RecordStore final : public ::mongo::RecordStore {
 public:
     explicit RecordStore(StringData ns,
                          StringData ident,
+                         KeyFormat keyFormat,
                          bool isCapped = false,
                          CappedCallback* cappedCallback = nullptr,
                          VisibilityManager* visibilityManager = nullptr);
@@ -57,7 +58,7 @@ public:
 
     virtual const char* name() const;
     virtual KeyFormat keyFormat() const {
-        return KeyFormat::Long;
+        return _keyFormat;
     }
     virtual long long dataSize(OperationContext* opCtx) const;
     virtual long long numRecords(OperationContext* opCtx) const;
@@ -119,6 +120,7 @@ private:
      */
     int64_t _nextRecordId(OperationContext* opCtx);
 
+    const KeyFormat _keyFormat;
     const bool _isCapped;
 
     StringData _ident;
diff --git a/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_record_store_test.cpp b/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_record_store_test.cpp
index 5411630d8d5..f37b95b9665 100644
--- a/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_record_store_test.cpp
+++ b/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_record_store_test.cpp
@@ -59,6 +59,8 @@ public:
         const std::string& ns, const CollectionOptions& collOptions) {
         return std::make_unique<RecordStore>(ns,
                                              "ident"_sd /* ident */,
+                                             collOptions.clusteredIndex ? KeyFormat::String
+                                                                        : KeyFormat::Long,
                                              false /* isCapped */,
                                              nullptr /* cappedCallback */,
                                              nullptr /* visibilityManager */);
@@ -67,6 +69,7 @@ public:
     virtual std::unique_ptr<mongo::RecordStore> newOplogRecordStore() final {
         return std::make_unique<RecordStore>(NamespaceString::kRsOplogNamespace.toString(),
                                              "ident"_sd,
+                                             KeyFormat::Long,
                                              /*isCapped*/ true,
                                              /*cappedCallback*/ nullptr,
                                              &_visibilityManager);
diff --git a/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_recovery_unit_test.cpp b/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_recovery_unit_test.cpp
index 6dd477338f6..2b91c71100b 100644
--- a/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_recovery_unit_test.cpp
+++ b/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_recovery_unit_test.cpp
@@ -51,8 +51,11 @@ public:
 
     virtual std::unique_ptr<mongo::RecordStore> createRecordStore(OperationContext* opCtx,
                                                                   const std::string& ns) {
-        return std::make_unique<RecordStore>(
-            ns, "ident"_sd /* ident */, false /* isCapped */, nullptr /* cappedCallback */);
+        return std::make_unique<RecordStore>(ns,
+                                             "ident"_sd /* ident */,
+                                             KeyFormat::Long,
+                                             false /* isCapped */,
+                                             nullptr /* cappedCallback */);
     }
 
 private:
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 9203972b62c..5159920d584 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
@@ -59,9 +59,9 @@ namespace {
 // Helper to interpret index data buffer
 class IndexDataEntry {
 public:
-    IndexDataEntry() : _buffer(nullptr) {}
-    IndexDataEntry(const uint8_t* buffer);
-    IndexDataEntry(const std::string& indexDataEntry);
+    IndexDataEntry() : _buffer(nullptr), _keyFormat(KeyFormat::Long) {}
+    IndexDataEntry(const uint8_t* buffer, KeyFormat keyFormat);
+    IndexDataEntry(const std::string& indexDataEntry, KeyFormat keyFormat);
 
     static std::string create(RecordId loc, const KeyString::TypeBits& typeBits);
 
@@ -69,9 +69,11 @@ public:
     size_t size() const;  // returns buffer size
     RecordId loc() const;
     KeyString::TypeBits typeBits() const;
+    KeyFormat keyFormat() const;
 
 private:
     const uint8_t* _buffer;
+    KeyFormat _keyFormat;
 };
 
 // Forward iterator for IndexDataEntry with contigous memory layout
@@ -80,7 +82,7 @@ class IndexDataEntryIterator : public boost::iterator_facade<IndexDataEntryItera
                                                              boost::forward_traversal_tag> {
 public:
     IndexDataEntryIterator() = default;
-    IndexDataEntryIterator(const uint8_t* entryData);
+    IndexDataEntryIterator(const uint8_t* entryData, KeyFormat keyFormat);
 
 private:
     friend class boost::iterator_core_access;
@@ -96,10 +98,11 @@ private:
 class UniqueIndexData {
 public:
     UniqueIndexData() : _size(0), _begin(nullptr), _end(nullptr) {}
-    UniqueIndexData(const std::string& indexData) {
+    UniqueIndexData(const std::string& indexData, KeyFormat keyFormat) {
         std::memcpy(&_size, indexData.data(), sizeof(uint64_t));
         _begin = reinterpret_cast<const uint8_t*>(indexData.data() + sizeof(uint64_t));
         _end = reinterpret_cast<const uint8_t*>(indexData.data() + indexData.size());
+        _keyFormat = keyFormat;
     }
 
     using const_iterator = IndexDataEntryIterator;
@@ -111,10 +114,10 @@ public:
         return _size == 0;
     }
     const_iterator begin() const {
-        return IndexDataEntryIterator(_begin);
+        return IndexDataEntryIterator(_begin, _keyFormat);
     }
     const_iterator end() const {
-        return IndexDataEntryIterator(_end);
+        return IndexDataEntryIterator(_end, _keyFormat);
     }
     const_iterator lower_bound(RecordId loc) const;
     const_iterator upper_bound(RecordId loc) const;
@@ -133,60 +136,84 @@ private:
     size_t _size;
     const uint8_t* _begin;
     const uint8_t* _end;
+    KeyFormat _keyFormat;
 };
 
-IndexDataEntry::IndexDataEntry(const uint8_t* buffer) : _buffer(buffer) {}
-IndexDataEntry::IndexDataEntry(const std::string& indexDataEntry)
-    : _buffer(reinterpret_cast<const uint8_t*>(indexDataEntry.data())) {}
+IndexDataEntry::IndexDataEntry(const uint8_t* buffer, KeyFormat keyFormat)
+    : _buffer(buffer), _keyFormat(keyFormat) {}
+IndexDataEntry::IndexDataEntry(const std::string& indexDataEntry, KeyFormat keyFormat)
+    : _buffer(reinterpret_cast<const uint8_t*>(indexDataEntry.data())), _keyFormat(keyFormat) {}
 
 std::string IndexDataEntry::create(RecordId loc, const KeyString::TypeBits& typeBits) {
-    uint64_t repr = loc.getLong();
-    uint64_t typebitsSize = typeBits.getSize();
-    std::string output(sizeof(repr) + sizeof(typebitsSize) + typebitsSize, '\0');
-
-    // RecordId
-    std::memcpy(output.data(), &repr, sizeof(repr));
-
-    // TypeBits size
-    std::memcpy(output.data() + sizeof(repr), &typebitsSize, sizeof(typebitsSize));
-
-    // TypeBits data
-    std::memcpy(
-        output.data() + sizeof(repr) + sizeof(typebitsSize), typeBits.getBuffer(), typebitsSize);
-
-    return output;
+    // [RecordId size, RecordId, TypeBits size, TypeBits]
+    StackBufBuilder builder;
+    if (loc.isLong()) {
+        builder.appendNum(sizeof(int64_t));
+        builder.appendNum(loc.getLong());
+    } else {
+        auto str = loc.getStr();
+        builder.appendNum(str.size());
+        builder.appendBuf(str.rawData(), str.size());
+    }
+    builder.appendNum(typeBits.getSize());
+    builder.appendBuf(typeBits.getBuffer(), typeBits.getSize());
+    return std::string(builder.buf(), builder.len());
 }
 
 const uint8_t* IndexDataEntry::buffer() const {
     return _buffer;
 }
 
+KeyFormat IndexDataEntry::keyFormat() const {
+    return _keyFormat;
+}
+
 size_t IndexDataEntry::size() const {
-    uint64_t typeBitsSize;
-    std::memcpy(&typeBitsSize, _buffer + sizeof(uint64_t), sizeof(uint64_t));
+    // [RecordId size, RecordId, TypeBits size, TypeBits]
+    size_t ridSize;
+    std::memcpy(&ridSize, _buffer, sizeof(size_t));
 
-    // RecordId + TypeBits size + TypeBits buffer
-    return sizeof(uint64_t) * 2 + typeBitsSize;
+    int len = sizeof(size_t) + ridSize;
+    size_t typeBitsSize;
+    std::memcpy(&typeBitsSize, _buffer + len, sizeof(size_t));
+
+    len += sizeof(size_t) + typeBitsSize;
+    return len;
 }
 
 RecordId IndexDataEntry::loc() const {
-    uint64_t repr;
-    std::memcpy(&repr, _buffer, sizeof(uint64_t));
-    return RecordId(repr);
+    // [RecordId size, RecordId, TypeBits size, TypeBits]
+    size_t ridSize;
+    std::memcpy(&ridSize, _buffer, sizeof(size_t));
+    const uint8_t* ridStart = _buffer + sizeof(size_t);
+    if (KeyFormat::Long == _keyFormat) {
+        int64_t repr;
+        std::memcpy(&repr, ridStart, ridSize);
+        return RecordId(repr);
+    } else {
+        return RecordId(reinterpret_cast<const char*>(ridStart), ridSize);
+    }
 }
 
 KeyString::TypeBits IndexDataEntry::typeBits() const {
-    uint64_t size;
-    std::memcpy(&size, _buffer + sizeof(uint64_t), sizeof(uint64_t));
+    // [RecordId size, RecordId, TypeBits size, TypeBits]
+    size_t ridSize;
+    std::memcpy(&ridSize, _buffer, sizeof(size_t));
+
+    int len = sizeof(size_t) + ridSize;
+    size_t typeBitsSize;
+    std::memcpy(&typeBitsSize, _buffer + len, sizeof(size_t));
 
-    BufReader reader(_buffer + sizeof(uint64_t) * 2, size);
+    len += sizeof(size_t);
+
+    BufReader reader(_buffer + len, typeBitsSize);
     return KeyString::TypeBits::fromBuffer(KeyString::Version::kLatestVersion, &reader);
 }
 
-IndexDataEntryIterator::IndexDataEntryIterator(const uint8_t* entryData)
-    : _entry(IndexDataEntry(entryData)) {}
+IndexDataEntryIterator::IndexDataEntryIterator(const uint8_t* entryData, KeyFormat keyFormat)
+    : _entry(IndexDataEntry(entryData, keyFormat)) {}
 void IndexDataEntryIterator::increment() {
-    _entry = IndexDataEntry(_entry.buffer() + _entry.size());
+    _entry = IndexDataEntry(_entry.buffer() + _entry.size(), _entry.keyFormat());
 }
 bool IndexDataEntryIterator::equal(IndexDataEntryIterator const& other) const {
     return _entry.buffer() == other._entry.buffer();
@@ -293,6 +320,22 @@ void prefixKeyStringWithLoc(KeyString::Builder* keyString,
     keyString->resetToKey(b.obj(), allAscending, loc);
 }
 
+void prefixMinKeyString(KeyString::Builder* keyString, const std::string& prefixToUse) {
+    BSONObjBuilder b;
+    b.append("", prefixToUse);                                               // prefix
+    b.append("", StringData(keyString->getBuffer(), keyString->getSize()));  // key
+
+    keyString->resetToKey(b.obj(), allAscending, KeyString::Discriminator::kExclusiveBefore);
+}
+
+void prefixMaxKeyString(KeyString::Builder* keyString, const std::string& prefixToUse) {
+    BSONObjBuilder b;
+    b.append("", prefixToUse);                                               // prefix
+    b.append("", StringData(keyString->getBuffer(), keyString->getSize()));  // key
+
+    keyString->resetToKey(b.obj(), allAscending, KeyString::Discriminator::kExclusiveAfter);
+}
+
 std::string createRadixKeyWithoutLocFromObj(const BSONObj& key,
                                             const std::string& prefixToUse,
                                             Ordering order) {
@@ -304,12 +347,19 @@ std::string createRadixKeyWithoutLocFromObj(const BSONObj& key,
 }
 
 std::string createRadixKeyWithoutLocFromKS(const KeyString::Value& keyString,
-                                           const std::string& prefixToUse) {
+                                           const std::string& prefixToUse,
+                                           KeyFormat keyFormat) {
     KeyString::Builder ks(KeyString::Version::kLatestVersion);
     auto ksBuffer = keyString.getBuffer();
-    if (ksBuffer)
-        ks.resetFromBuffer(ksBuffer,
-                           KeyString::sizeWithoutRecordIdAtEnd(ksBuffer, keyString.getSize()));
+    if (ksBuffer) {
+        if (KeyFormat::Long == keyFormat) {
+            ks.resetFromBuffer(
+                ksBuffer, KeyString::sizeWithoutRecordIdLongAtEnd(ksBuffer, keyString.getSize()));
+        } else {
+            ks.resetFromBuffer(
+                ksBuffer, KeyString::sizeWithoutRecordIdStrAtEnd(ksBuffer, keyString.getSize()));
+        }
+    }
     prefixKeyStringWithoutLoc(&ks, prefixToUse);
     return std::string(ks.getBuffer(), ks.getSize());
 }
@@ -324,37 +374,62 @@ std::string createRadixKeyWithoutLocFromKSWithoutRecordId(const KeyString::Value
     return std::string(ks.getBuffer(), ks.getSize());
 }
 
-std::string createRadixKeyWithLocFromObj(const BSONObj& key,
-                                         RecordId loc,
-                                         const std::string& prefixToUse,
-                                         Ordering order) {
+std::string createMinRadixKeyFromObj(const BSONObj& key,
+                                     const std::string& prefixToUse,
+                                     Ordering order) {
     KeyString::Version version = KeyString::Version::kLatestVersion;
     KeyString::Builder ks(version, BSONObj::stripFieldNames(key), order);
 
-    prefixKeyStringWithLoc(&ks, loc, prefixToUse);
+    prefixMinKeyString(&ks, prefixToUse);
+    return std::string(ks.getBuffer(), ks.getSize());
+}
+
+std::string createMaxRadixKeyFromObj(const BSONObj& key,
+                                     const std::string& prefixToUse,
+                                     Ordering order) {
+    KeyString::Version version = KeyString::Version::kLatestVersion;
+    KeyString::Builder ks(version, BSONObj::stripFieldNames(key), order);
+
+    prefixMaxKeyString(&ks, prefixToUse);
     return std::string(ks.getBuffer(), ks.getSize());
 }
 
+
 std::string createRadixKeyWithLocFromKS(const KeyString::Value& keyString,
                                         RecordId loc,
                                         const std::string& prefixToUse) {
     KeyString::Builder ks(KeyString::Version::kLatestVersion);
     auto ksBuffer = keyString.getBuffer();
-    if (ksBuffer)
-        ks.resetFromBuffer(ksBuffer,
-                           KeyString::sizeWithoutRecordIdAtEnd(ksBuffer, keyString.getSize()));
+    if (ksBuffer) {
+        if (loc.isLong()) {
+            ks.resetFromBuffer(
+                ksBuffer, KeyString::sizeWithoutRecordIdLongAtEnd(ksBuffer, keyString.getSize()));
+        } else {
+            ks.resetFromBuffer(
+                ksBuffer, KeyString::sizeWithoutRecordIdStrAtEnd(ksBuffer, keyString.getSize()));
+        }
+    }
     prefixKeyStringWithLoc(&ks, loc, prefixToUse);
     return std::string(ks.getBuffer(), ks.getSize());
 }
 
-std::string createRadixKeyWithLocFromKSWithoutRecordId(const KeyString::Value& keyString,
-                                                       RecordId loc,
-                                                       const std::string& prefixToUse) {
+std::string createMinRadixKeyFromKSWithoutRecordId(const KeyString::Value& keyString,
+                                                   const std::string& prefixToUse) {
     KeyString::Builder ks(KeyString::Version::kLatestVersion);
     auto ksBuffer = keyString.getBuffer();
     if (ksBuffer)
         ks.resetFromBuffer(ksBuffer, keyString.getSize());
-    prefixKeyStringWithLoc(&ks, loc, prefixToUse);
+    prefixMinKeyString(&ks, prefixToUse);
+    return std::string(ks.getBuffer(), ks.getSize());
+}
+
+std::string createMaxRadixKeyFromKSWithoutRecordId(const KeyString::Value& keyString,
+                                                   const std::string& prefixToUse) {
+    KeyString::Builder ks(KeyString::Version::kLatestVersion);
+    auto ksBuffer = keyString.getBuffer();
+    if (ksBuffer)
+        ks.resetFromBuffer(ksBuffer, keyString.getSize());
+    prefixMaxKeyString(&ks, prefixToUse);
     return std::string(ks.getBuffer(), ks.getSize());
 }
 
@@ -384,8 +459,9 @@ IndexKeyEntry createIndexKeyEntryFromRadixKey(const std::string& radixKey,
 
 IndexKeyEntry createIndexKeyEntryFromRadixKey(const std::string& radixKey,
                                               const std::string& indexDataEntry,
-                                              const Ordering order) {
-    IndexDataEntry data(indexDataEntry);
+                                              const Ordering order,
+                                              KeyFormat keyFormat) {
+    IndexDataEntry data(indexDataEntry, keyFormat);
     return IndexKeyEntry(createObjFromRadixKey(radixKey, data.typeBits(), order), data.loc());
 }
 
@@ -402,8 +478,9 @@ boost::optional<KeyStringEntry> createKeyStringEntryFromRadixKey(
 
 boost::optional<KeyStringEntry> createKeyStringEntryFromRadixKey(const std::string& radixKey,
                                                                  const std::string& indexDataEntry,
-                                                                 const Ordering& order) {
-    IndexDataEntry data(indexDataEntry);
+                                                                 const Ordering& order,
+                                                                 KeyFormat keyFormat) {
+    IndexDataEntry data(indexDataEntry, keyFormat);
     RecordId loc = data.loc();
     auto key = createObjFromRadixKey(radixKey, data.typeBits(), order);
     KeyString::Builder ksFinal(KeyString::Version::kLatestVersion, key, order);
@@ -428,6 +505,7 @@ public:
                std::string _identEnd,
                StringStore* workingCopy,
                Ordering order,
+               KeyFormat keyFormat,
                std::string prefixBSON,
                std::string KSForIdentEnd);
     virtual void setEndPosition(const BSONObj& key, bool inclusive) override;
@@ -446,18 +524,6 @@ public:
 
 private:
     // CRTP Interface
-    std::string createRadixKeyFromObj(const BSONObj& key,
-                                      RecordId loc,
-                                      const std::string& prefixToUse,
-                                      Ordering order) {
-        return static_cast<CursorImpl*>(this)->createRadixKeyFromObj(key, loc, prefixToUse, order);
-    }
-    std::string createRadixKeyFromKSWithoutRecordId(const KeyString::Value& keyString,
-                                                    RecordId loc,
-                                                    const std::string& prefixToUse) {
-        return static_cast<CursorImpl*>(this)->createRadixKeyFromKSWithoutRecordId(
-            keyString, loc, prefixToUse);
-    }
     boost::optional<KeyStringEntry> finishSeekAfterProcessing() {
         return static_cast<CursorImpl*>(this)->finishSeekAfterProcessing();
     }
@@ -514,6 +580,7 @@ protected:
     StringStore::const_reverse_iterator _reverseIt;
     // This is the ordering for the key's values for multi-field keys.
     Ordering _order;
+    KeyFormat _keyFormat;
     // This stores whether or not the end position is inclusive for restore.
     bool _endPosIncl;
     // This stores the key for the end position.
@@ -531,6 +598,7 @@ CursorBase<CursorImpl>::CursorBase(OperationContext* opCtx,
                                    std::string _identEnd,
                                    StringStore* workingCopy,
                                    Ordering order,
+                                   KeyFormat keyFormat,
                                    std::string _KSForIdentStart,
                                    std::string identEndBSON)
     : _opCtx(opCtx),
@@ -545,6 +613,7 @@ CursorBase<CursorImpl>::CursorBase(OperationContext* opCtx,
       _forwardIt(workingCopy->begin()),
       _reverseIt(workingCopy->rbegin()),
       _order(order),
+      _keyFormat(keyFormat),
       _endPosIncl(false),
       _KSForIdentStart(_KSForIdentStart),
       _KSForIdentEnd(identEndBSON) {}
@@ -605,11 +674,9 @@ void CursorBase<CursorImpl>::setEndPosition(const BSONObj& key, bool inclusive)
     // If forward and inclusive or reverse and not inclusive, then we use the last element in this
     // ident. Otherwise, we use the first as our bound.
     if (_forward == inclusive)
-        it = workingCopy->upper_bound(
-            createRadixKeyFromObj(key, RecordId::maxLong(), _prefix, _order));
+        it = workingCopy->upper_bound(createMaxRadixKeyFromObj(key, _prefix, _order));
     else
-        it = workingCopy->lower_bound(
-            createRadixKeyFromObj(key, RecordId::minLong(), _prefix, _order));
+        it = workingCopy->lower_bound(createMinRadixKeyFromObj(key, _prefix, _order));
     if (_forward)
         _endPos = it;
     else
@@ -661,10 +728,10 @@ boost::optional<KeyStringEntry> CursorBase<CursorImpl>::seekAfterProcessing(
     // is also reversed.
     if (_forward == inclusive)
         it = _workingCopy->lower_bound(
-            createRadixKeyFromKSWithoutRecordId(keyStringVal, RecordId::minLong(), _prefix));
+            createMinRadixKeyFromKSWithoutRecordId(keyStringVal, _prefix));
     else
         it = _workingCopy->upper_bound(
-            createRadixKeyFromKSWithoutRecordId(keyStringVal, RecordId::maxLong(), _prefix));
+            createMaxRadixKeyFromKSWithoutRecordId(keyStringVal, _prefix));
     if (_forward)
         _forwardIt = it;
     else
@@ -715,10 +782,14 @@ boost::optional<KeyStringEntry> CursorBase<CursorImpl>::seekExactForKeyString(
     if (!ksEntry) {
         return {};
     }
+    auto sizeWithoutRecordId = KeyFormat::Long == _keyFormat
+        ? KeyString::sizeWithoutRecordIdLongAtEnd(ksEntry->keyString.getBuffer(),
+                                                  ksEntry->keyString.getSize())
+        : KeyString::sizeWithoutRecordIdStrAtEnd(ksEntry->keyString.getBuffer(),
+                                                 ksEntry->keyString.getSize());
     if (KeyString::compare(ksEntry->keyString.getBuffer(),
                            keyStringValue.getBuffer(),
-                           KeyString::sizeWithoutRecordIdAtEnd(ksEntry->keyString.getBuffer(),
-                                                               ksEntry->keyString.getSize()),
+                           sizeWithoutRecordId,
                            keyStringValue.getSize()) == 0) {
         return KeyStringEntry(ksEntry->keyString, ksEntry->loc);
     }
@@ -818,17 +889,6 @@ private:
 
     bool advanceNextInternal();
     void finishAdvanceNext();
-    std::string createRadixKeyFromObj(const BSONObj& key,
-                                      RecordId loc,
-                                      const std::string& prefixToUse,
-                                      Ordering order) {
-        return createRadixKeyWithoutLocFromObj(key, prefixToUse, order);
-    }
-    std::string createRadixKeyFromKSWithoutRecordId(const KeyString::Value& keyString,
-                                                    RecordId loc,
-                                                    const std::string& prefixToUse) {
-        return createRadixKeyWithoutLocFromKSWithoutRecordId(keyString, prefixToUse);
-    }
     boost::optional<KeyStringEntry> finishSeekAfterProcessing();
 
     void saveForward();
@@ -869,11 +929,11 @@ void CursorUnique::finishAdvanceNext() {
     // We have moved to a new position in the tree, initialize index data for iterating over
     // duplicates
     if (_forward) {
-        _indexData = UniqueIndexData(_forwardIt->second);
+        _indexData = UniqueIndexData(_forwardIt->second, _keyFormat);
         _indexDataIt = _indexData.begin();
         _indexDataEnd = _indexData.end();
     } else {
-        _indexData = UniqueIndexData(_reverseIt->second);
+        _indexData = UniqueIndexData(_reverseIt->second, _keyFormat);
         _reversePos = 0;
         initReverseDataIterators();
     }
@@ -901,7 +961,7 @@ bool CursorUnique::checkCursorValid() {
                 // check the equal side of things. This assumption doesn't hold for unique index
                 // KeyStrings.
                 std::string endPosKeyString =
-                    createRadixKeyFromObj(*_endPosKey, RecordId::minLong(), _prefix, _order);
+                    createMaxRadixKeyFromObj(*_endPosKey, _prefix, _order);
 
                 if (_forwardIt->first.compare(endPosKeyString) <= 0)
                     return true;
@@ -923,7 +983,7 @@ bool CursorUnique::checkCursorValid() {
                     return true;
 
                 std::string endPosKeyString =
-                    createRadixKeyFromObj(*_endPosKey, RecordId::minLong(), _prefix, _order);
+                    createMinRadixKeyFromObj(*_endPosKey, _prefix, _order);
 
                 if (_reverseIt->first.compare(endPosKeyString) >= 0)
                     return true;
@@ -974,13 +1034,13 @@ boost::optional<KeyStringEntry> CursorUnique::finishSeekAfterProcessing() {
     // We have seeked to an entry in the tree. Now unpack the data and initialize iterators to point
     // to the first entry if this index contains duplicates
     if (_forward) {
-        _indexData = UniqueIndexData(_forwardIt->second);
+        _indexData = UniqueIndexData(_forwardIt->second, _keyFormat);
         _indexDataIt = _indexData.begin();
         _indexDataEnd = _indexData.end();
         return createKeyStringEntryFromRadixKey(
             _forwardIt->first, _indexDataIt->loc(), _indexDataIt->typeBits(), _order);
     } else {
-        _indexData = UniqueIndexData(_reverseIt->second);
+        _indexData = UniqueIndexData(_reverseIt->second, _keyFormat);
         _reversePos = 0;
         initReverseDataIterators();
         return createKeyStringEntryFromRadixKey(
@@ -1004,7 +1064,7 @@ void CursorUnique::restoreForward() {
     _lastMoveWasRestore = true;
     if (_saveLoc != RecordId() && _forwardIt != _workingCopy->end() &&
         _forwardIt->first == _saveKey) {
-        _indexData = UniqueIndexData(_forwardIt->second);
+        _indexData = UniqueIndexData(_forwardIt->second, _keyFormat);
         _indexDataIt = _indexData.lower_bound(_saveLoc);
         _indexDataEnd = _indexData.end();
         if (_indexDataIt == _indexDataEnd) {
@@ -1012,7 +1072,7 @@ void CursorUnique::restoreForward() {
             // radix tree to be positioned on a valid item
             ++_forwardIt;
             if (checkCursorValid()) {
-                _indexData = UniqueIndexData(_forwardIt->second);
+                _indexData = UniqueIndexData(_forwardIt->second, _keyFormat);
                 _indexDataIt = _indexData.begin();
                 _indexDataEnd = _indexData.end();
             }
@@ -1031,13 +1091,13 @@ void CursorUnique::restoreReverse() {
     _lastMoveWasRestore = true;
     if (_saveLoc != RecordId() && _reverseIt != _workingCopy->rend() &&
         _reverseIt->first == _saveKey) {
-        _indexData = UniqueIndexData(_reverseIt->second);
+        _indexData = UniqueIndexData(_reverseIt->second, _keyFormat);
         _indexDataIt = _indexData.upper_bound(_saveLoc);
         _indexDataEnd = _indexData.end();
         if (_indexDataIt == _indexDataEnd) {
             ++_reverseIt;
             if (checkCursorValid()) {
-                _indexData = UniqueIndexData(_reverseIt->second);
+                _indexData = UniqueIndexData(_reverseIt->second, _keyFormat);
                 _reversePos = 0;
                 initReverseDataIterators();
             }
@@ -1069,17 +1129,6 @@ protected:
         return false;
     }
     void finishAdvanceNext() {}
-    std::string createRadixKeyFromObj(const BSONObj& key,
-                                      RecordId loc,
-                                      const std::string& prefixToUse,
-                                      Ordering order) {
-        return createRadixKeyWithLocFromObj(key, loc, prefixToUse, order);
-    }
-    std::string createRadixKeyFromKSWithoutRecordId(const KeyString::Value& keyString,
-                                                    RecordId loc,
-                                                    const std::string& prefixToUse) {
-        return createRadixKeyWithLocFromKSWithoutRecordId(keyString, loc, prefixToUse);
-    }
     boost::optional<KeyStringEntry> finishSeekAfterProcessing();
     void saveForward() {}
     void saveReverse() {}
@@ -1124,9 +1173,11 @@ boost::optional<IndexKeyEntry> CursorStandard::next(RequestedInfo parts) {
     }
 
     if (_forward) {
-        return createIndexKeyEntryFromRadixKey(_forwardIt->first, _forwardIt->second, _order);
+        return createIndexKeyEntryFromRadixKey(
+            _forwardIt->first, _forwardIt->second, _order, _keyFormat);
     }
-    return createIndexKeyEntryFromRadixKey(_reverseIt->first, _reverseIt->second, _order);
+    return createIndexKeyEntryFromRadixKey(
+        _reverseIt->first, _reverseIt->second, _order, _keyFormat);
 }
 
 boost::optional<KeyStringEntry> CursorStandard::nextKeyString() {
@@ -1135,17 +1186,21 @@ boost::optional<KeyStringEntry> CursorStandard::nextKeyString() {
     }
 
     if (_forward) {
-        return createKeyStringEntryFromRadixKey(_forwardIt->first, _forwardIt->second, _order);
+        return createKeyStringEntryFromRadixKey(
+            _forwardIt->first, _forwardIt->second, _order, _keyFormat);
     }
-    return createKeyStringEntryFromRadixKey(_reverseIt->first, _reverseIt->second, _order);
+    return createKeyStringEntryFromRadixKey(
+        _reverseIt->first, _reverseIt->second, _order, _keyFormat);
 }
 
 boost::optional<KeyStringEntry> CursorStandard::finishSeekAfterProcessing() {
     // We have seeked to an entry in the tree.
     if (_forward) {
-        return createKeyStringEntryFromRadixKey(_forwardIt->first, _forwardIt->second, _order);
+        return createKeyStringEntryFromRadixKey(
+            _forwardIt->first, _forwardIt->second, _order, _keyFormat);
     } else {
-        return createKeyStringEntryFromRadixKey(_reverseIt->first, _reverseIt->second, _order);
+        return createKeyStringEntryFromRadixKey(
+            _reverseIt->first, _reverseIt->second, _order, _keyFormat);
     }
 }
 
@@ -1166,11 +1221,23 @@ void CursorStandard::restoreReverse() {
     _lastMoveWasRestore = (_reverseIt->first.compare(_saveKey) != 0);
 }
 
+RecordId decodeRecordId(const KeyString::Value& keyString, KeyFormat keyFormat) {
+    RecordId loc;
+    if (keyFormat == KeyFormat::Long) {
+        loc = KeyString::decodeRecordIdLongAtEnd(keyString.getBuffer(), keyString.getSize());
+    } else {
+        loc = KeyString::decodeRecordIdStrAtEnd(keyString.getBuffer(), keyString.getSize());
+    }
+    invariant(loc.isValid(), loc.toString());
+    return loc;
+}
+
 }  // namespace
 
 SortedDataBuilderBase::SortedDataBuilderBase(OperationContext* opCtx,
                                              bool dupsAllowed,
                                              Ordering order,
+                                             KeyFormat rsKeyFormat,
                                              const std::string& prefix,
                                              const std::string& identEnd,
                                              const IndexDescriptor* desc,
@@ -1180,6 +1247,7 @@ SortedDataBuilderBase::SortedDataBuilderBase(OperationContext* opCtx,
     : _opCtx(opCtx),
       _dupsAllowed(dupsAllowed),
       _order(order),
+      _rsKeyFormat(rsKeyFormat),
       _prefix(prefix),
       _identEnd(identEnd),
       _desc(desc),
@@ -1188,12 +1256,10 @@ SortedDataBuilderBase::SortedDataBuilderBase(OperationContext* opCtx,
       _collation(collation) {}
 
 Status SortedDataBuilderUnique::addKey(const KeyString::Value& keyString) {
-    dassert(
-        KeyString::decodeRecordIdLongAtEnd(keyString.getBuffer(), keyString.getSize()).isValid());
+    RecordId loc = decodeRecordId(keyString, _rsKeyFormat);
     StringStore* workingCopy(RecoveryUnit::get(_opCtx)->getHead());
-    RecordId loc = KeyString::decodeRecordIdLongAtEnd(keyString.getBuffer(), keyString.getSize());
 
-    std::string key = createRadixKeyWithoutLocFromKS(keyString, _prefix);
+    std::string key = createRadixKeyWithoutLocFromKS(keyString, _prefix, _rsKeyFormat);
     auto it = workingCopy->find(key);
     if (it != workingCopy->end()) {
         if (!_dupsAllowed) {
@@ -1203,7 +1269,7 @@ Status SortedDataBuilderUnique::addKey(const KeyString::Value& keyString) {
             return buildDupKeyErrorStatus(_opCtx, keyString, _order, _desc);
         }
 
-        UniqueIndexData data(it->second);
+        UniqueIndexData data(it->second, _rsKeyFormat);
         // Bulk builder add keys in ascending order so we should insert at the end
         auto added = data.add(loc, keyString.getTypeBits());
         if (!added) {
@@ -1226,11 +1292,12 @@ Status SortedDataBuilderUnique::addKey(const KeyString::Value& keyString) {
 // this ident.
 SortedDataInterfaceBase::SortedDataInterfaceBase(OperationContext* opCtx,
                                                  StringData ident,
+                                                 KeyFormat rsKeyFormat,
                                                  const IndexDescriptor* desc)
     : ::mongo::SortedDataInterface(ident,
                                    KeyString::Version::kLatestVersion,
                                    Ordering::make(desc->keyPattern()),
-                                   KeyFormat::Long),
+                                   rsKeyFormat),
       // All entries in this ident will have a prefix of ident + \1.
       _prefix(ident.toString().append(1, '\1')),
       // Therefore, the string ident + \2 will be greater than all elements in this ident.
@@ -1254,6 +1321,7 @@ std::unique_ptr<SortedDataBuilderInterface> SortedDataInterfaceUnique::makeBulkB
     return std::make_unique<SortedDataBuilderUnique>(opCtx,
                                                      dupsAllowed,
                                                      _ordering,
+                                                     _rsKeyFormat,
                                                      _prefix,
                                                      _identEnd,
                                                      _desc,
@@ -1267,8 +1335,9 @@ std::unique_ptr<SortedDataBuilderInterface> SortedDataInterfaceUnique::makeBulkB
 // this ident.
 SortedDataInterfaceUnique::SortedDataInterfaceUnique(OperationContext* opCtx,
                                                      StringData ident,
+                                                     KeyFormat rsKeyFormat,
                                                      const IndexDescriptor* desc)
-    : SortedDataInterfaceBase(opCtx, ident, desc) {
+    : SortedDataInterfaceBase(opCtx, ident, rsKeyFormat, desc) {
     // This is the string representation of the KeyString before elements in this ident, which is
     // ident + \0. This is before all elements in this ident.
     _KSForIdentStart =
@@ -1289,9 +1358,9 @@ Status SortedDataInterfaceUnique::insert(OperationContext* opCtx,
                                          const KeyString::Value& keyString,
                                          bool dupsAllowed) {
     StringStore* workingCopy(RecoveryUnit::get(opCtx)->getHead());
-    RecordId loc = KeyString::decodeRecordIdLongAtEnd(keyString.getBuffer(), keyString.getSize());
+    RecordId loc = decodeRecordId(keyString, rsKeyFormat());
 
-    std::string key = createRadixKeyWithoutLocFromKS(keyString, _prefix);
+    std::string key = createRadixKeyWithoutLocFromKS(keyString, _prefix, _rsKeyFormat);
     auto it = workingCopy->find(key);
     if (it != workingCopy->end()) {
         if (!dupsAllowed) {
@@ -1300,7 +1369,7 @@ Status SortedDataInterfaceUnique::insert(OperationContext* opCtx,
             return buildDupKeyErrorStatus(opCtx, keyString, _ordering, _desc);
         }
 
-        UniqueIndexData data(it->second);
+        UniqueIndexData data(it->second, _rsKeyFormat);
         auto added = data.add(loc, keyString.getTypeBits());
         if (!added) {
             // Already indexed
@@ -1320,17 +1389,17 @@ void SortedDataInterfaceUnique::unindex(OperationContext* opCtx,
                                         const KeyString::Value& keyString,
                                         bool dupsAllowed) {
     StringStore* workingCopy(RecoveryUnit::get(opCtx)->getHead());
-    RecordId loc = KeyString::decodeRecordIdLongAtEnd(keyString.getBuffer(), keyString.getSize());
+    RecordId loc = decodeRecordId(keyString, rsKeyFormat());
 
-    auto key = createRadixKeyWithoutLocFromKS(keyString, _prefix);
+    auto key = createRadixKeyWithoutLocFromKS(keyString, _prefix, _rsKeyFormat);
     auto it = workingCopy->find(key);
     if (it != workingCopy->end()) {
-        UniqueIndexData data(it->second);
+        UniqueIndexData data(it->second, _rsKeyFormat);
         auto removed = data.remove(loc);
         if (!removed)
             return;  // loc not found, nothing to unindex
 
-        if (UniqueIndexData(*removed).empty()) {
+        if (UniqueIndexData(*removed, _rsKeyFormat).empty()) {
             workingCopy->erase(key);
         } else {
             workingCopy->update({std::move(key), *removed});
@@ -1367,7 +1436,7 @@ Status SortedDataInterfaceUnique::dupKeyCheck(OperationContext* opCtx,
     if (it == workingCopy->end())
         return Status::OK();
 
-    UniqueIndexData data(it->second);
+    UniqueIndexData data(it->second, _rsKeyFormat);
     if (data.size() > 1) {
         return buildDupKeyErrorStatus(opCtx, key, _ordering, _desc);
     }
@@ -1382,7 +1451,7 @@ void SortedDataInterfaceUnique::fullValidate(OperationContext* opCtx,
     long long numKeys = 0;
     auto it = workingCopy->lower_bound(_KSForIdentStart);
     while (it != workingCopy->end() && it->first.compare(_KSForIdentEnd) < 0) {
-        numKeys += UniqueIndexData(it->second).size();
+        numKeys += UniqueIndexData(it->second, _rsKeyFormat).size();
         ++it;
     }
     *numKeysOut = numKeys;
@@ -1423,6 +1492,7 @@ std::unique_ptr<mongo::SortedDataInterface::Cursor> SortedDataInterfaceUnique::n
                                           _identEnd,
                                           workingCopy,
                                           _ordering,
+                                          _rsKeyFormat,
                                           _KSForIdentStart,
                                           _KSForIdentEnd);
 }
@@ -1432,10 +1502,8 @@ Status SortedDataInterfaceBase::initAsEmpty(OperationContext* opCtx) {
 }
 
 Status SortedDataBuilderStandard::addKey(const KeyString::Value& keyString) {
-    dassert(
-        KeyString::decodeRecordIdLongAtEnd(keyString.getBuffer(), keyString.getSize()).isValid());
     StringStore* workingCopy(RecoveryUnit::get(_opCtx)->getHead());
-    RecordId loc = KeyString::decodeRecordIdLongAtEnd(keyString.getBuffer(), keyString.getSize());
+    RecordId loc = decodeRecordId(keyString, _rsKeyFormat);
 
     std::string key = createRadixKeyWithLocFromKS(keyString, loc, _prefix);
     bool inserted =
@@ -1451,6 +1519,7 @@ std::unique_ptr<SortedDataBuilderInterface> SortedDataInterfaceStandard::makeBul
     return std::make_unique<SortedDataBuilderStandard>(opCtx,
                                                        dupsAllowed,
                                                        _ordering,
+                                                       _rsKeyFormat,
                                                        _prefix,
                                                        _identEnd,
                                                        _desc,
@@ -1464,31 +1533,30 @@ std::unique_ptr<SortedDataBuilderInterface> SortedDataInterfaceStandard::makeBul
 // this ident.
 SortedDataInterfaceStandard::SortedDataInterfaceStandard(OperationContext* opCtx,
                                                          StringData ident,
+                                                         KeyFormat rsKeyFormat,
                                                          const IndexDescriptor* desc)
-    : SortedDataInterfaceBase(opCtx, ident, desc) {
+    : SortedDataInterfaceBase(opCtx, ident, rsKeyFormat, desc) {
     // This is the string representation of the KeyString before elements in this ident, which is
     // ident + \0. This is before all elements in this ident.
-    _KSForIdentStart = createRadixKeyWithLocFromObj(
-        BSONObj(), RecordId::minLong(), ident.toString().append(1, '\0'), _ordering);
+    _KSForIdentStart =
+        createMinRadixKeyFromObj(BSONObj(), ident.toString().append(1, '\0'), _ordering);
     // Similarly, this is the string representation of the KeyString for something greater than
     // all other elements in this ident.
-    _KSForIdentEnd =
-        createRadixKeyWithLocFromObj(BSONObj(), RecordId::minLong(), _identEnd, _ordering);
+    _KSForIdentEnd = createMinRadixKeyFromObj(BSONObj(), _identEnd, _ordering);
 }
 
 SortedDataInterfaceStandard::SortedDataInterfaceStandard(const Ordering& ordering, StringData ident)
     : SortedDataInterfaceBase(ordering, ident) {
-    _KSForIdentStart = createRadixKeyWithLocFromObj(
-        BSONObj(), RecordId::minLong(), ident.toString().append(1, '\0'), _ordering);
-    _KSForIdentEnd =
-        createRadixKeyWithLocFromObj(BSONObj(), RecordId::minLong(), _identEnd, _ordering);
+    _KSForIdentStart =
+        createMinRadixKeyFromObj(BSONObj(), ident.toString().append(1, '\0'), _ordering);
+    _KSForIdentEnd = createMinRadixKeyFromObj(BSONObj(), _identEnd, _ordering);
 }
 
 Status SortedDataInterfaceStandard::insert(OperationContext* opCtx,
                                            const KeyString::Value& keyString,
                                            bool dupsAllowed) {
     StringStore* workingCopy(RecoveryUnit::get(opCtx)->getHead());
-    RecordId loc = KeyString::decodeRecordIdLongAtEnd(keyString.getBuffer(), keyString.getSize());
+    RecordId loc = decodeRecordId(keyString, rsKeyFormat());
 
     std::string key = createRadixKeyWithLocFromKS(keyString, loc, _prefix);
     bool inserted =
@@ -1503,7 +1571,7 @@ void SortedDataInterfaceStandard::unindex(OperationContext* opCtx,
                                           const KeyString::Value& keyString,
                                           bool dupsAllowed) {
     StringStore* workingCopy(RecoveryUnit::get(opCtx)->getHead());
-    RecordId loc = KeyString::decodeRecordIdLongAtEnd(keyString.getBuffer(), keyString.getSize());
+    RecordId loc = decodeRecordId(keyString, rsKeyFormat());
 
     auto key = createRadixKeyWithLocFromKS(keyString, loc, _prefix);
     if (workingCopy->erase(key))
@@ -1539,6 +1607,7 @@ std::unique_ptr<mongo::SortedDataInterface::Cursor> SortedDataInterfaceStandard:
                                             _identEnd,
                                             workingCopy,
                                             _ordering,
+                                            _rsKeyFormat,
                                             _KSForIdentStart,
                                             _KSForIdentEnd);
 }
diff --git a/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_sorted_impl.h b/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_sorted_impl.h
index a8275188cd4..7a497c9ea19 100644
--- a/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_sorted_impl.h
+++ b/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_sorted_impl.h
@@ -42,6 +42,7 @@ public:
     SortedDataBuilderBase(OperationContext* opCtx,
                           bool dupsAllowed,
                           Ordering order,
+                          KeyFormat rsKeyFormat,
                           const std::string& prefix,
                           const std::string& identEnd,
                           const IndexDescriptor* desc,
@@ -54,6 +55,8 @@ protected:
     bool _dupsAllowed;
     // Order of the keys.
     Ordering _order;
+    // RecordId format of the related record store
+    KeyFormat _rsKeyFormat;
     // Prefix and identEnd for the ident.
     std::string _prefix;
     std::string _identEnd;
@@ -75,7 +78,10 @@ public:
     // Truncate is not required at the time of writing but will be when the truncate command is
     // created
     Status truncate(RecoveryUnit* ru);
-    SortedDataInterfaceBase(OperationContext* opCtx, StringData ident, const IndexDescriptor* desc);
+    SortedDataInterfaceBase(OperationContext* opCtx,
+                            StringData ident,
+                            KeyFormat rsKeyFormat,
+                            const IndexDescriptor* desc);
     SortedDataInterfaceBase(const Ordering& ordering, StringData ident);
     bool appendCustomStats(OperationContext* opCtx,
                            BSONObjBuilder* output,
@@ -107,6 +113,7 @@ class SortedDataInterfaceUnique : public SortedDataInterfaceBase {
 public:
     SortedDataInterfaceUnique(OperationContext* opCtx,
                               StringData ident,
+                              KeyFormat rsKeyFormat,
                               const IndexDescriptor* desc);
     SortedDataInterfaceUnique(const Ordering& ordering, StringData ident);
     std::unique_ptr<SortedDataBuilderInterface> makeBulkBuilder(OperationContext* opCtx,
@@ -135,6 +142,7 @@ class SortedDataInterfaceStandard : public SortedDataInterfaceBase {
 public:
     SortedDataInterfaceStandard(OperationContext* opCtx,
                                 StringData ident,
+                                KeyFormat rsKeyFormat,
                                 const IndexDescriptor* desc);
     SortedDataInterfaceStandard(const Ordering& ordering, StringData ident);
     std::unique_ptr<SortedDataBuilderInterface> makeBulkBuilder(OperationContext* opCtx,
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 8a8b1502132..4798342147b 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
@@ -70,11 +70,6 @@ public:
     std::unique_ptr<mongo::SortedDataInterface> newSortedDataInterface(bool unique,
                                                                        bool partial,
                                                                        KeyFormat keyFormat) final {
-        if (keyFormat == KeyFormat::String) {
-            // not supported
-            return nullptr;
-        }
-
         std::string ns = "test.ephemeral_for_test";
         OperationContextNoop opCtx(newRecoveryUnit().release());
 
@@ -91,8 +86,7 @@ public:
 
         auto collection = std::make_unique<CollectionMock>(NamespaceString(ns));
         _descs.emplace_back("", spec);
-        return _kvEngine.getSortedDataInterface(
-            &opCtx, CollectionOptions(), "ident"_sd, &_descs.back());
+        return _kvEngine.getSortedDataInterface(&opCtx, keyFormat, "ident"_sd, &_descs.back());
     }
 
     std::unique_ptr<mongo::RecoveryUnit> newRecoveryUnit() final {
diff --git a/src/mongo/db/storage/key_string.cpp b/src/mongo/db/storage/key_string.cpp
index fd79e6a9ada..0f8dfc2c2d7 100644
--- a/src/mongo/db/storage/key_string.cpp
+++ b/src/mongo/db/storage/key_string.cpp
@@ -2492,7 +2492,7 @@ RecordId decodeRecordIdLongAtEnd(const void* bufferRaw, size_t bufSize) {
     return decodeRecordIdLong(&reader);
 }
 
-size_t sizeWithoutRecordIdAtEnd(const void* bufferRaw, size_t bufSize) {
+size_t sizeWithoutRecordIdLongAtEnd(const void* bufferRaw, size_t bufSize) {
     invariant(bufSize >= 2);  // smallest possible encoding of a RecordId.
     const unsigned char* buffer = static_cast<const unsigned char*>(bufferRaw);
     const unsigned char lastByte = *(buffer + bufSize - 1);
@@ -2501,6 +2501,22 @@ size_t sizeWithoutRecordIdAtEnd(const void* bufferRaw, size_t bufSize) {
     return bufSize - ridSize;
 }
 
+size_t sizeWithoutRecordIdStrAtEnd(const void* bufferRaw, size_t bufSize) {
+    invariant(bufSize > 0);
+    const uint8_t* buffer = static_cast<const uint8_t*>(bufferRaw);
+
+    // The current encoding for strings supports strings up to 128 bytes. The high bit is reserved
+    // for future usage.
+    uint8_t len = buffer[bufSize - 1];
+    keyStringAssert(5566400,
+                    fmt::format("Cannot decode record id string longer than {} bytes; size is {}",
+                                kMaxRecordIdStrLen,
+                                len),
+                    len <= kMaxRecordIdStrLen);
+    invariant(bufSize > static_cast<size_t>(len + 1));
+    return bufSize - len - 1;
+}
+
 RecordId decodeRecordIdLong(BufReader* reader) {
     const uint8_t firstByte = readType<uint8_t>(reader, false);
     const uint8_t numExtraBytes = firstByte >> 5;  // high 3 bits in firstByte
@@ -2615,10 +2631,10 @@ void appendToBSONArray(const char* buf, int len, BSONArrayBuilder* builder, Vers
     toBsonValue(ctype, &reader, &typeBitsReader, inverted, version, builder, depth);
 }
 
-void Value::serializeWithoutRecordId(BufBuilder& buf) const {
+void Value::serializeWithoutRecordIdLong(BufBuilder& buf) const {
     dassert(decodeRecordIdLongAtEnd(_buffer.get(), _ksSize).isValid());
 
-    const int32_t sizeWithoutRecordId = sizeWithoutRecordIdAtEnd(_buffer.get(), _ksSize);
+    const int32_t sizeWithoutRecordId = sizeWithoutRecordIdLongAtEnd(_buffer.get(), _ksSize);
     buf.appendNum(sizeWithoutRecordId);                 // Serialize size of KeyString
     buf.appendBuf(_buffer.get(), sizeWithoutRecordId);  // Serialize KeyString
     buf.appendBuf(_buffer.get() + _ksSize, _buffer.size() - _ksSize);  // Serialize TypeBits
diff --git a/src/mongo/db/storage/key_string.h b/src/mongo/db/storage/key_string.h
index 68fc1cdeb53..2b846e55270 100644
--- a/src/mongo/db/storage/key_string.h
+++ b/src/mongo/db/storage/key_string.h
@@ -335,7 +335,7 @@ public:
     int compareWithTypeBits(const Value& other) const;
 
     template <class T>
-    int compareWithoutRecordId(const T& other) const;
+    int compareWithoutRecordIdLong(const T& other) const;
 
     // Returns the size of the stored KeyString.
     size_t getSize() const {
@@ -381,7 +381,7 @@ public:
      * information. The serialized format takes the following form:
      *   [keystring size][keystring encoding][typebits encoding]
      */
-    void serializeWithoutRecordId(BufBuilder& buf) const;
+    void serializeWithoutRecordIdLong(BufBuilder& buf) const;
 
     // Deserialize the Value from a serialized format.
     static Value deserialize(BufReader& buf, KeyString::Version version) {
@@ -605,7 +605,7 @@ public:
     int compare(const T& other) const;
 
     template <class T>
-    int compareWithoutRecordId(const T& other) const;
+    int compareWithoutRecordIdLong(const T& other) const;
 
     /**
      * @return a hex encoding of this key
@@ -919,7 +919,7 @@ inline typename std::enable_if<isKeyString<T>::value, std::ostream&>::type opera
 
 /**
  * Given a KeyString which may or may not have a RecordId, returns the length of the section without
- * the RecordId. More expensive than sizeWithoutRecordIdAtEnd
+ * the RecordId. More expensive than sizeWithoutRecordId(Long|Str)AtEnd
  */
 size_t getKeySize(const char* buffer, size_t len, Ordering ord, const TypeBits& typeBits);
 
@@ -957,9 +957,16 @@ RecordId decodeRecordIdLongAtEnd(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.
+ * Given a KeyString with a RecordId in the long format, returns the length of the section without
+ * the RecordId.
  */
-size_t sizeWithoutRecordIdAtEnd(const void* bufferRaw, size_t bufSize);
+size_t sizeWithoutRecordIdLongAtEnd(const void* bufferRaw, size_t bufSize);
+
+/**
+ * Given a KeyString with a RecordId in the string format, returns the length of the section without
+ * the RecordId.
+ */
+size_t sizeWithoutRecordIdStrAtEnd(const void* bufferRaw, size_t bufSize);
 
 /**
  * Decodes a RecordId, consuming all bytes needed from reader.
@@ -998,12 +1005,12 @@ int BuilderBase<BufferT>::compare(const T& other) const {
 
 template <class BufferT>
 template <class T>
-int BuilderBase<BufferT>::compareWithoutRecordId(const T& other) const {
+int BuilderBase<BufferT>::compareWithoutRecordIdLong(const T& other) const {
     return KeyString::compare(
         getBuffer(),
         other.getBuffer(),
-        !isEmpty() ? sizeWithoutRecordIdAtEnd(getBuffer(), getSize()) : 0,
-        !other.isEmpty() ? sizeWithoutRecordIdAtEnd(other.getBuffer(), other.getSize()) : 0);
+        !isEmpty() ? sizeWithoutRecordIdLongAtEnd(getBuffer(), getSize()) : 0,
+        !other.isEmpty() ? sizeWithoutRecordIdLongAtEnd(other.getBuffer(), other.getSize()) : 0);
 }
 
 template <class T>
@@ -1012,12 +1019,12 @@ int Value::compare(const T& other) const {
 }
 
 template <class T>
-int Value::compareWithoutRecordId(const T& other) const {
+int Value::compareWithoutRecordIdLong(const T& other) const {
     return KeyString::compare(
         getBuffer(),
         other.getBuffer(),
-        !isEmpty() ? sizeWithoutRecordIdAtEnd(getBuffer(), getSize()) : 0,
-        !other.isEmpty() ? sizeWithoutRecordIdAtEnd(other.getBuffer(), other.getSize()) : 0);
+        !isEmpty() ? sizeWithoutRecordIdLongAtEnd(getBuffer(), getSize()) : 0,
+        !other.isEmpty() ? sizeWithoutRecordIdLongAtEnd(other.getBuffer(), other.getSize()) : 0);
 }
 
 }  // namespace KeyString
diff --git a/src/mongo/db/storage/kv/kv_engine.h b/src/mongo/db/storage/kv/kv_engine.h
index 08152c0dc9c..1398e2b6d1c 100644
--- a/src/mongo/db/storage/kv/kv_engine.h
+++ b/src/mongo/db/storage/kv/kv_engine.h
@@ -231,14 +231,6 @@ public:
     }
 
     /**
-     * Returns true if the storage engine supports collections clustered on _id. That is,
-     * collections will use _id values as their RecordId and do not need a separate _id index.
-     */
-    virtual bool supportsClusteredIdIndex() const {
-        return false;
-    }
-
-    /**
      * Returns true if storage engine supports --directoryperdb.
      * See:
      *     http://docs.mongodb.org/manual/reference/program/mongod/#cmdoption--directoryperdb
diff --git a/src/mongo/db/storage/record_store_test_harness.cpp b/src/mongo/db/storage/record_store_test_harness.cpp
index 12c5cb12776..495cdd32a60 100644
--- a/src/mongo/db/storage/record_store_test_harness.cpp
+++ b/src/mongo/db/storage/record_store_test_harness.cpp
@@ -409,11 +409,6 @@ TEST(RecordStoreTestHarness, Cursor1) {
 
 TEST(RecordStoreTestHarness, ClusteredRecordStore) {
     const auto harnessHelper = newRecordStoreHarnessHelper();
-    if (!harnessHelper->getEngine()->supportsClusteredIdIndex()) {
-        // Only WiredTiger supports clustered indexes on _id.
-        return;
-    }
-
     const std::string ns = "test.system.buckets.a";
     CollectionOptions options;
     options.clusteredIndex = true;
@@ -453,9 +448,8 @@ TEST(RecordStoreTestHarness, ClusteredRecordStore) {
         ASSERT_EQ(numRecords, currRecord);
     }
 
-    {
+    if (auto cursor = rs->getRandomCursor(opCtx.get())) {
         // Verify random cursors work on ObjectId's.
-        auto cursor = rs->getRandomCursor(opCtx.get());
         auto record = cursor->next();
         ASSERT(record);
 
@@ -520,11 +514,6 @@ TEST(RecordStoreTestHarness, ClusteredRecordStore) {
 
 TEST(RecordStoreTestHarness, ClusteredRecordStoreSeekNear) {
     const auto harnessHelper = newRecordStoreHarnessHelper();
-    if (!harnessHelper->getEngine()->supportsClusteredIdIndex()) {
-        // Only WiredTiger supports clustered indexes on _id.
-        return;
-    }
-
     const std::string ns = "test.system.buckets.a";
     CollectionOptions options;
     options.clusteredIndex = true;
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
index 61379236433..604c4ae0168 100644
--- a/src/mongo/db/storage/sorted_data_interface_test_keyformat_string.cpp
+++ b/src/mongo/db/storage/sorted_data_interface_test_keyformat_string.cpp
@@ -43,10 +43,6 @@ 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()));
 
@@ -118,10 +114,6 @@ 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()));
 
@@ -191,10 +183,6 @@ 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()));
 
@@ -244,10 +232,6 @@ TEST(SortedDataInterface, InsertReservedRecordIdStr) {
     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()));
     WriteUnitOfWork uow(opCtx.get());
@@ -265,10 +249,6 @@ TEST(SortedDataInterface, BuilderAddKeyWithReservedRecordIdStr) {
     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()));
diff --git a/src/mongo/db/storage/storage_engine.h b/src/mongo/db/storage/storage_engine.h
index 1fd8922399e..aed7c3978d4 100644
--- a/src/mongo/db/storage/storage_engine.h
+++ b/src/mongo/db/storage/storage_engine.h
@@ -422,12 +422,6 @@ public:
     virtual void setJournalListener(JournalListener* jl) = 0;
 
     /**
-     * Returns true if the storage engine supports collections clustered on _id. That is,
-     * collections will use _id values as their RecordId and do not need a separate _id index.
-     */
-    virtual bool supportsClusteredIdIndex() const = 0;
-
-    /**
      * Returns whether the storage engine supports "recover to stable timestamp". Returns true
      * if the storage engine supports "recover to stable timestamp" but does not currently have
      * a stable timestamp. In that case StorageEngine::recoverToStableTimestamp() will return
diff --git a/src/mongo/db/storage/storage_engine_impl.cpp b/src/mongo/db/storage/storage_engine_impl.cpp
index ab79ae046d0..822bc3dbc32 100644
--- a/src/mongo/db/storage/storage_engine_impl.cpp
+++ b/src/mongo/db/storage/storage_engine_impl.cpp
@@ -1025,10 +1025,6 @@ boost::optional<Timestamp> StorageEngineImpl::getLastStableRecoveryTimestamp() c
     return _engine->getLastStableRecoveryTimestamp();
 }
 
-bool StorageEngineImpl::supportsClusteredIdIndex() const {
-    return _engine->supportsClusteredIdIndex();
-}
-
 bool StorageEngineImpl::supportsReadConcernSnapshot() const {
     return _engine->supportsReadConcernSnapshot();
 }
diff --git a/src/mongo/db/storage/storage_engine_impl.h b/src/mongo/db/storage/storage_engine_impl.h
index 30b9546374a..119de49c4bf 100644
--- a/src/mongo/db/storage/storage_engine_impl.h
+++ b/src/mongo/db/storage/storage_engine_impl.h
@@ -151,8 +151,6 @@ public:
 
     boost::optional<Timestamp> getOplogNeededForCrashRecovery() const final;
 
-    bool supportsClusteredIdIndex() const final;
-
     bool supportsReadConcernSnapshot() const final;
 
     bool supportsReadConcernMajority() const final;
diff --git a/src/mongo/db/storage/storage_engine_mock.h b/src/mongo/db/storage/storage_engine_mock.h
index bc86f13b61a..3d69c854a5b 100644
--- a/src/mongo/db/storage/storage_engine_mock.h
+++ b/src/mongo/db/storage/storage_engine_mock.h
@@ -103,9 +103,6 @@ public:
         return nullptr;
     }
     void setJournalListener(JournalListener* jl) final {}
-    bool supportsClusteredIdIndex() const final {
-        return false;
-    }
     bool supportsRecoverToStableTimestamp() const final {
         return false;
     }
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_index.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_index.cpp
index 6c3dfd4414f..62b88a3f349 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_index.cpp
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_index.cpp
@@ -660,7 +660,7 @@ public:
 
         // Do a duplicate check, but only if dups aren't allowed.
         if (!_dupsAllowed) {
-            const int cmp = newKeyString.compareWithoutRecordId(_previousKeyString);
+            const int cmp = newKeyString.compareWithoutRecordIdLong(_previousKeyString);
             if (cmp == 0) {
                 // Duplicate found!
                 auto newKey = KeyString::toBson(newKeyString, _idx->_ordering);
@@ -719,7 +719,7 @@ public:
     Status addKey(const KeyString::Value& newKeyString) override {
         dassertRecordIdAtEnd(newKeyString, KeyFormat::Long);
 
-        const int cmp = newKeyString.compareWithoutRecordId(_previousKeyString);
+        const int cmp = newKeyString.compareWithoutRecordIdLong(_previousKeyString);
         // _previousKeyString.isEmpty() is only true on the first call to addKey().
         invariant(_previousKeyString.isEmpty() || cmp > 0);
 
@@ -735,8 +735,8 @@ public:
             value.appendTypeBits(typeBits);
         }
 
-        auto sizeWithoutRecordId =
-            KeyString::sizeWithoutRecordIdAtEnd(newKeyString.getBuffer(), newKeyString.getSize());
+        auto sizeWithoutRecordId = KeyString::sizeWithoutRecordIdLongAtEnd(newKeyString.getBuffer(),
+                                                                           newKeyString.getSize());
         WiredTigerItem keyItem(newKeyString.getBuffer(), sizeWithoutRecordId);
         WiredTigerItem valueItem(value.getBuffer(), value.getSize());
 
@@ -869,8 +869,8 @@ public:
 
         if (KeyString::compare(ksEntry->keyString.getBuffer(),
                                key.getBuffer(),
-                               KeyString::sizeWithoutRecordIdAtEnd(ksEntry->keyString.getBuffer(),
-                                                                   ksEntry->keyString.getSize()),
+                               KeyString::sizeWithoutRecordIdLongAtEnd(
+                                   ksEntry->keyString.getBuffer(), ksEntry->keyString.getSize()),
                                key.getSize()) == 0) {
             return KeyStringEntry(ksEntry->keyString, ksEntry->loc);
         }
@@ -1496,7 +1496,7 @@ Status WiredTigerIdIndex::_insert(OperationContext* opCtx,
     invariant(id.isValid());
 
     auto sizeWithoutRecordId =
-        KeyString::sizeWithoutRecordIdAtEnd(keyString.getBuffer(), keyString.getSize());
+        KeyString::sizeWithoutRecordIdLongAtEnd(keyString.getBuffer(), keyString.getSize());
     WiredTigerItem keyItem(keyString.getBuffer(), sizeWithoutRecordId);
 
     KeyString::Builder value(getKeyStringVersion(), id);
@@ -1535,7 +1535,7 @@ Status WiredTigerIndexUnique::_insert(OperationContext* opCtx,
         // A prefix key is KeyString of index key. It is the component of the index entry that
         // should be unique.
         auto sizeWithoutRecordId =
-            KeyString::sizeWithoutRecordIdAtEnd(keyString.getBuffer(), keyString.getSize());
+            KeyString::sizeWithoutRecordIdLongAtEnd(keyString.getBuffer(), keyString.getSize());
         WiredTigerItem prefixKeyItem(keyString.getBuffer(), sizeWithoutRecordId);
 
         // First phase inserts the prefix key to prohibit concurrent insertions of same key
@@ -1614,7 +1614,7 @@ void WiredTigerIdIndex::_unindex(OperationContext* opCtx,
     invariant(id.isValid());
 
     auto sizeWithoutRecordId =
-        KeyString::sizeWithoutRecordIdAtEnd(keyString.getBuffer(), keyString.getSize());
+        KeyString::sizeWithoutRecordIdLongAtEnd(keyString.getBuffer(), keyString.getSize());
     WiredTigerItem keyItem(keyString.getBuffer(), sizeWithoutRecordId);
     setKey(c, keyItem.Get());
 
@@ -1704,7 +1704,7 @@ void WiredTigerIndexUnique::_unindex(OperationContext* opCtx,
     // format key has index key + Record id. WT_NOTFOUND is possible if index key is in old format.
     // Retry removal of key using old format.
     auto sizeWithoutRecordId =
-        KeyString::sizeWithoutRecordIdAtEnd(keyString.getBuffer(), keyString.getSize());
+        KeyString::sizeWithoutRecordIdLongAtEnd(keyString.getBuffer(), keyString.getSize());
     WiredTigerItem keyItem(keyString.getBuffer(), sizeWithoutRecordId);
     setKey(c, keyItem.Get());
 
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.h b/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.h
index daa8166e56f..fe75e451a36 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.h
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.h
@@ -251,10 +251,6 @@ public:
 
     Timestamp getAllDurableTimestamp() const override;
 
-    bool supportsClusteredIdIndex() const final override {
-        return true;
-    }
-
     bool supportsReadConcernSnapshot() const final override;
 
     bool supportsOplogStones() const final override;
diff --git a/src/mongo/dbtests/query_stage_collscan.cpp b/src/mongo/dbtests/query_stage_collscan.cpp
index cb9a8038509..4fafd1fc46e 100644
--- a/src/mongo/dbtests/query_stage_collscan.cpp
+++ b/src/mongo/dbtests/query_stage_collscan.cpp
@@ -490,9 +490,6 @@ TEST_F(QueryStageCollectionScanTest, QueryTestCollscanResumeAfterRecordIdSeekFai
 }
 
 TEST_F(QueryStageCollectionScanTest, QueryTestCollscanClusteredMinMax) {
-    if (!(&_opCtx)->getServiceContext()->getStorageEngine()->supportsClusteredIdIndex()) {
-        return;
-    }
     auto ns = NamespaceString("a.b");
     auto collDeleter = makeCollectionClustered(ns);
     AutoGetCollectionForRead autoColl(&_opCtx, ns);
@@ -535,9 +532,6 @@ TEST_F(QueryStageCollectionScanTest, QueryTestCollscanClusteredMinMax) {
 }
 
 TEST_F(QueryStageCollectionScanTest, QueryTestCollscanClusteredReverse) {
-    if (!(&_opCtx)->getServiceContext()->getStorageEngine()->supportsClusteredIdIndex()) {
-        return;
-    }
     auto ns = NamespaceString("a.b");
     auto collDeleter = makeCollectionClustered(ns);
     AutoGetCollectionForRead autoColl(&_opCtx, ns);
@@ -582,9 +576,6 @@ TEST_F(QueryStageCollectionScanTest, QueryTestCollscanClusteredReverse) {
 }
 
 TEST_F(QueryStageCollectionScanTest, QueryTestCollscanClusteredNonExistentRecordIds) {
-    if (!(&_opCtx)->getServiceContext()->getStorageEngine()->supportsClusteredIdIndex()) {
-        return;
-    }
     auto ns = NamespaceString("a.b");
     auto collDeleter = makeCollectionClustered(ns);
     AutoGetCollectionForRead autoColl(&_opCtx, ns);
@@ -629,9 +620,6 @@ TEST_F(QueryStageCollectionScanTest, QueryTestCollscanClusteredNonExistentRecord
 }
 
 TEST_F(QueryStageCollectionScanTest, QueryTestCollscanClusteredInnerRange) {
-    if (!(&_opCtx)->getServiceContext()->getStorageEngine()->supportsClusteredIdIndex()) {
-        return;
-    }
     auto ns = NamespaceString("a.b");
     auto collDeleter = makeCollectionClustered(ns);
     AutoGetCollectionForRead autoColl(&_opCtx, ns);
@@ -681,9 +669,6 @@ TEST_F(QueryStageCollectionScanTest, QueryTestCollscanClusteredInnerRange) {
 }
 
 TEST_F(QueryStageCollectionScanTest, QueryTestCollscanClusteredInnerRangeExclusive) {
-    if (!(&_opCtx)->getServiceContext()->getStorageEngine()->supportsClusteredIdIndex()) {
-        return;
-    }
     auto ns = NamespaceString("a.b");
     auto collDeleter = makeCollectionClustered(ns);
     AutoGetCollectionForRead autoColl(&_opCtx, ns);
@@ -744,9 +729,6 @@ TEST_F(QueryStageCollectionScanTest, QueryTestCollscanClusteredInnerRangeExclusi
 }
 
 TEST_F(QueryStageCollectionScanTest, QueryTestCollscanClusteredInnerRangeExclusiveReverse) {
-    if (!(&_opCtx)->getServiceContext()->getStorageEngine()->supportsClusteredIdIndex()) {
-        return;
-    }
     auto ns = NamespaceString("a.b");
     auto collDeleter = makeCollectionClustered(ns);
     AutoGetCollectionForRead autoColl(&_opCtx, ns);
diff --git a/src/mongo/dbtests/validate_tests.cpp b/src/mongo/dbtests/validate_tests.cpp
index 418146ff05d..92e83b8f465 100644
--- a/src/mongo/dbtests/validate_tests.cpp
+++ b/src/mongo/dbtests/validate_tests.cpp
@@ -71,11 +71,8 @@ public:
 
         _supportsBackgroundValidation = storageGlobalParams.engine != "ephemeralForTest";
 
-        _supportsClusteredIdIndex =
-            _opCtx.getServiceContext()->getStorageEngine()->supportsClusteredIdIndex();
-
         CollectionOptions options;
-        if (clustered && _supportsClusteredIdIndex) {
+        if (clustered) {
             options.clusteredIndex = true;
         }
 
@@ -193,7 +190,6 @@ protected:
     unique_ptr<AutoGetDb> _autoDb;
     Database* _db;
     bool _supportsBackgroundValidation;
-    bool _supportsClusteredIdIndex;
 };
 
 template <bool full, bool background>
@@ -3489,10 +3485,6 @@ public:
         : ValidateBase(/*full=*/false, background, /*clustered=*/true) {}
 
     void run() {
-        if (!_supportsClusteredIdIndex) {
-            return;
-        }
-
         if (_background && !_supportsBackgroundValidation) {
             return;
         }
@@ -3555,10 +3547,6 @@ public:
         : ValidateBase(/*full=*/false, background, /*clustered=*/true) {}
 
     void run() {
-        if (!_supportsClusteredIdIndex) {
-            return;
-        }
-
         if (_background && !_supportsBackgroundValidation) {
             return;
         }
@@ -3656,10 +3644,6 @@ public:
         : ValidateBase(/*full=*/false, /*background=*/false, /*clustered=*/true) {}
 
     void run() {
-        if (!_supportsClusteredIdIndex) {
-            return;
-        }
-
         if (_background && !_supportsBackgroundValidation) {
             return;
         }