SERVER-63251 Refactor IndexAccessMethod to support non-SortedData indexes

40 files changed, 731 insertions, 853 deletions

diff --git a/src/mongo/db/catalog/SConscript b/src/mongo/db/catalog/SConscript
index 384775dd7c2..17a173e2a51 100644
--- a/src/mongo/db/catalog/SConscript
+++ b/src/mongo/db/catalog/SConscript
@@ -204,9 +204,8 @@ env.Library(
         '$BUILD_DIR/mongo/db/catalog/collection_query_info',
         '$BUILD_DIR/mongo/db/collection_index_usage_tracker',
         '$BUILD_DIR/mongo/db/common',
-        '$BUILD_DIR/mongo/db/index/index_build_interceptor',
+        '$BUILD_DIR/mongo/db/index/index_access_method',
         '$BUILD_DIR/mongo/db/index/index_descriptor',
-        '$BUILD_DIR/mongo/db/index/skipped_record_tracker',
         '$BUILD_DIR/mongo/db/index_names',
         '$BUILD_DIR/mongo/db/repl/repl_coordinator_interface',
         '$BUILD_DIR/mongo/db/ttl_collection_cache',
@@ -251,7 +250,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/catalog_raii',
         '$BUILD_DIR/mongo/db/concurrency/write_conflict_exception',
         '$BUILD_DIR/mongo/db/curop',
-        '$BUILD_DIR/mongo/db/index/index_build_interceptor',
+        '$BUILD_DIR/mongo/db/index/index_access_method',
         '$BUILD_DIR/mongo/db/repl/repl_coordinator_interface',
         '$BUILD_DIR/mongo/db/resumable_index_builds_idl',
         '$BUILD_DIR/mongo/db/service_context',
@@ -372,7 +371,6 @@ env.Library(
         '$BUILD_DIR/mongo/db/index/index_access_method',
         '$BUILD_DIR/mongo/db/index/index_access_method_factory',
         '$BUILD_DIR/mongo/db/index/index_access_methods',
-        '$BUILD_DIR/mongo/db/index/index_build_interceptor',
         '$BUILD_DIR/mongo/db/multitenancy',
         '$BUILD_DIR/mongo/db/op_observer',
         '$BUILD_DIR/mongo/db/record_id_helpers',
@@ -462,6 +460,7 @@ env.Library(
     ],
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/curop',
+        '$BUILD_DIR/mongo/db/index/index_access_method',
         '$BUILD_DIR/mongo/util/fail_point',
         'validate_idl',
     ],
diff --git a/src/mongo/db/catalog/coll_mod_index.cpp b/src/mongo/db/catalog/coll_mod_index.cpp
index 0009192a4a4..b66a2b50db5 100644
--- a/src/mongo/db/catalog/coll_mod_index.cpp
+++ b/src/mongo/db/catalog/coll_mod_index.cpp
@@ -110,7 +110,7 @@ void _processCollModIndexRequestHidden(OperationContext* opCtx,
  */
 void getKeysForIndex(OperationContext* opCtx,
                      const CollectionPtr& collection,
-                     const IndexAccessMethod* accessMethod,
+                     const SortedDataIndexAccessMethod* accessMethod,
                      const BSONObj& doc,
                      KeyStringSet* keys) {
     SharedBufferFragmentBuilder pooledBuilder(KeyString::HeapBuilder::kHeapAllocatorDefaultBytes);
@@ -119,8 +119,8 @@ void getKeysForIndex(OperationContext* opCtx,
                           collection,
                           pooledBuilder,
                           doc,
-                          IndexAccessMethod::GetKeysMode::kEnforceConstraints,
-                          IndexAccessMethod::GetKeysContext::kAddingKeys,
+                          InsertDeleteOptions::ConstraintEnforcementMode::kEnforceConstraints,
+                          SortedDataIndexAccessMethod::GetKeysContext::kAddingKeys,
                           keys,
                           nullptr,       //  multikeyMetadataKeys
                           nullptr,       //  multikeyPaths
@@ -146,7 +146,7 @@ void _processCollModIndexRequestUnique(OperationContext* opCtx,
     std::list<std::set<RecordId>> duplicateRecordsList;
     if (!mode) {
         auto entry = idx->getEntry();
-        auto accessMethod = entry->accessMethod();
+        auto accessMethod = entry->accessMethod()->asSortedData();
 
         invariant(docsForUniqueIndex,
                   fmt::format("Unique index conversion requires valid set of changed docs from "
@@ -330,7 +330,7 @@ std::list<std::set<RecordId>> scanIndexForDuplicates(
     const IndexDescriptor* idx,
     boost::optional<KeyString::Value> firstKeyString) {
     auto entry = idx->getEntry();
-    auto accessMethod = entry->accessMethod();
+    auto accessMethod = entry->accessMethod()->asSortedData();
     // Only scans for the duplicates on one key if 'firstKeyString' is provided.
     bool scanOneKey = static_cast<bool>(firstKeyString);
 
diff --git a/src/mongo/db/catalog/collection_validation.cpp b/src/mongo/db/catalog/collection_validation.cpp
index 318b682689c..51ebcc3bc37 100644
--- a/src/mongo/db/catalog/collection_validation.cpp
+++ b/src/mongo/db/catalog/collection_validation.cpp
@@ -429,7 +429,7 @@ void _validateCatalogEntry(OperationContext* opCtx,
                             status.reason()));
         }
 
-        if (!indexEntry->isReady(opCtx, collection)) {
+        if (!indexEntry->isReady(opCtx)) {
             continue;
         }
 
diff --git a/src/mongo/db/catalog/drop_indexes.cpp b/src/mongo/db/catalog/drop_indexes.cpp
index 9f062bdeb24..5309bdf69b7 100644
--- a/src/mongo/db/catalog/drop_indexes.cpp
+++ b/src/mongo/db/catalog/drop_indexes.cpp
@@ -231,14 +231,14 @@ Status dropIndexByDescriptor(OperationContext* opCtx,
     // exist in standalone mode.
     auto entry = indexCatalog->getEntry(desc);
     if (entry->isFrozen()) {
-        invariant(!entry->isReady(opCtx, collection));
+        invariant(!entry->isReady(opCtx));
         invariant(getReplSetMemberInStandaloneMode(opCtx->getServiceContext()));
         // Return here. No need to fall through to op observer on standalone.
         return indexCatalog->dropUnfinishedIndex(opCtx, collection, desc);
     }
 
     // Do not allow dropping unfinished indexes that are not frozen.
-    if (!entry->isReady(opCtx, collection)) {
+    if (!entry->isReady(opCtx)) {
         return Status(ErrorCodes::IndexNotFound,
                       str::stream()
                           << "can't drop unfinished index with name: " << desc->indexName());
diff --git a/src/mongo/db/catalog/index_catalog_entry.h b/src/mongo/db/catalog/index_catalog_entry.h
index c5eac391739..9b69dff35d5 100644
--- a/src/mongo/db/catalog/index_catalog_entry.h
+++ b/src/mongo/db/catalog/index_catalog_entry.h
@@ -50,6 +50,7 @@ class CollectionPtr;
 class CollectionCatalogEntry;
 class Ident;
 class IndexAccessMethod;
+class SortedDataIndexAccessMethod;
 class IndexBuildInterceptor;
 class IndexDescriptor;
 class MatchExpression;
@@ -155,7 +156,7 @@ public:
      * not consider whether the index is visible or ready in the current storage snapshot. For
      * that, use isReadyInMySnapshot() or isPresentInMySnapshot().
      */
-    virtual bool isReady(OperationContext* opCtx, const CollectionPtr& collection) const = 0;
+    virtual bool isReady(OperationContext* opCtx) const = 0;
 
     /**
      * Safely check whether this index is visible in the durable catalog in the current storage
diff --git a/src/mongo/db/catalog/index_catalog_entry_impl.cpp b/src/mongo/db/catalog/index_catalog_entry_impl.cpp
index 420fb5cf892..8936da74bca 100644
--- a/src/mongo/db/catalog/index_catalog_entry_impl.cpp
+++ b/src/mongo/db/catalog/index_catalog_entry_impl.cpp
@@ -118,8 +118,7 @@ void IndexCatalogEntryImpl::init(std::unique_ptr<IndexAccessMethod> accessMethod
     _accessMethod = std::move(accessMethod);
 }
 
-bool IndexCatalogEntryImpl::isReady(OperationContext* opCtx,
-                                    const CollectionPtr& collection) const {
+bool IndexCatalogEntryImpl::isReady(OperationContext* opCtx) const {
     // For multi-document transactions, we can open a snapshot prior to checking the
     // minimumSnapshotVersion on a collection.  This means we are unprotected from reading
     // out-of-sync index catalog entries.  To fix this, we uassert if we detect that the
@@ -246,8 +245,8 @@ void IndexCatalogEntryImpl::setMultikey(OperationContext* opCtx,
     // not have to account for potential dupes, since all metadata keys are indexed against a single
     // RecordId. An attempt to write a duplicate key will therefore be ignored.
     if (!multikeyMetadataKeys.empty()) {
-        uassertStatusOK(accessMethod()->insertKeys(
-            opCtx, collection, multikeyMetadataKeys, {}, {}, {}, nullptr));
+        uassertStatusOK(accessMethod()->asSortedData()->insertKeys(
+            opCtx, collection, multikeyMetadataKeys, {}, {}, nullptr));
     }
 
     // Mark the catalog as multikey, and record the multikey paths if applicable.
@@ -346,7 +345,7 @@ Status IndexCatalogEntryImpl::_setMultikeyInMultiDocumentTransaction(
 }
 
 std::shared_ptr<Ident> IndexCatalogEntryImpl::getSharedIdent() const {
-    return {shared_from_this(), _accessMethod->getSortedDataInterface()};  // aliasing constructor
+    return {shared_from_this(), _accessMethod->getIdentPtr()};  // aliasing constructor
 }
 
 // ----
diff --git a/src/mongo/db/catalog/index_catalog_entry_impl.h b/src/mongo/db/catalog/index_catalog_entry_impl.h
index 60093abfeeb..9be2ab730d1 100644
--- a/src/mongo/db/catalog/index_catalog_entry_impl.h
+++ b/src/mongo/db/catalog/index_catalog_entry_impl.h
@@ -162,7 +162,7 @@ public:
                           bool isMultikey,
                           const MultikeyPaths& multikeyPaths) const final;
 
-    bool isReady(OperationContext* opCtx, const CollectionPtr& collection) const final;
+    bool isReady(OperationContext* opCtx) const final;
 
     bool isFrozen() const final;
 
diff --git a/src/mongo/db/catalog/index_catalog_impl.cpp b/src/mongo/db/catalog/index_catalog_impl.cpp
index e8fd618f1b8..c6d64ddf583 100644
--- a/src/mongo/db/catalog/index_catalog_impl.cpp
+++ b/src/mongo/db/catalog/index_catalog_impl.cpp
@@ -93,7 +93,6 @@ MONGO_FAIL_POINT_DEFINE(skipIndexNewRecords);
 // This failpoint causes the check for TTL indexes on capped collections to be ignored.
 MONGO_FAIL_POINT_DEFINE(ignoreTTLIndexCappedCollectionCheck);
 
-using std::endl;
 using std::string;
 using std::unique_ptr;
 using std::vector;
@@ -222,19 +221,19 @@ Status IndexCatalogImpl::init(OperationContext* opCtx, Collection* collection) {
                 auto flags = CreateIndexEntryFlags::kInitFromDisk | CreateIndexEntryFlags::kFrozen;
                 IndexCatalogEntry* entry =
                     createIndexEntry(opCtx, collection, std::move(descriptor), flags);
-                fassert(31433, !entry->isReady(opCtx, collection));
+                fassert(31433, !entry->isReady(opCtx));
             } else {
                 // Initializing with unfinished indexes may occur during rollback or startup.
                 auto flags = CreateIndexEntryFlags::kInitFromDisk;
                 IndexCatalogEntry* entry =
                     createIndexEntry(opCtx, collection, std::move(descriptor), flags);
-                fassert(4505500, !entry->isReady(opCtx, collection));
+                fassert(4505500, !entry->isReady(opCtx));
             }
         } else {
             auto flags = CreateIndexEntryFlags::kInitFromDisk | CreateIndexEntryFlags::kIsReady;
             IndexCatalogEntry* entry =
                 createIndexEntry(opCtx, collection, std::move(descriptor), flags);
-            fassert(17340, entry->isReady(opCtx, collection));
+            fassert(17340, entry->isReady(opCtx));
 
             // When initializing indexes from disk, we conservatively set the minimumVisibleSnapshot
             // to non _id indexes to the recovery timestamp. The _id index is left visible. It's
@@ -1171,7 +1170,7 @@ Status IndexCatalogImpl::dropIndex(OperationContext* opCtx,
     if (!entry)
         return Status(ErrorCodes::InternalError, "cannot find index to delete");
 
-    if (!entry->isReady(opCtx, collection))
+    if (!entry->isReady(opCtx))
         return Status(ErrorCodes::InternalError, "cannot delete not ready index");
 
     return dropIndexEntry(opCtx, collection, entry);
@@ -1185,7 +1184,7 @@ Status IndexCatalogImpl::dropUnfinishedIndex(OperationContext* opCtx,
     if (!entry)
         return Status(ErrorCodes::InternalError, "cannot find index to delete");
 
-    if (entry->isReady(opCtx, collection))
+    if (entry->isReady(opCtx))
         return Status(ErrorCodes::InternalError, "expected unfinished index, but it is ready");
 
     return dropIndexEntry(opCtx, collection, entry);
@@ -1448,7 +1447,7 @@ const IndexDescriptor* IndexCatalogImpl::refreshEntry(OperationContext* opCtx,
     // to the CollectionIndexUsageTrackerDecoration (shared state among Collection instances).
     auto newDesc = std::make_unique<IndexDescriptor>(_getAccessMethodName(keyPattern), spec);
     auto newEntry = createIndexEntry(opCtx, collection, std::move(newDesc), flags);
-    invariant(newEntry->isReady(opCtx, collection));
+    invariant(newEntry->isReady(opCtx));
     newEntry->accessMethod()->setEnforceDuplicateConstraints(enforceDuplicateConstraints);
     auto desc = newEntry->descriptor();
     CollectionIndexUsageTrackerDecoration::get(collection->getSharedDecorations())
@@ -1469,58 +1468,6 @@ const IndexDescriptor* IndexCatalogImpl::refreshEntry(OperationContext* opCtx,
 
 // ---------------------------
 
-Status IndexCatalogImpl::_indexKeys(OperationContext* opCtx,
-                                    const CollectionPtr& coll,
-                                    const IndexCatalogEntry* index,
-                                    const KeyStringSet& keys,
-                                    const KeyStringSet& multikeyMetadataKeys,
-                                    const MultikeyPaths& multikeyPaths,
-                                    const BSONObj& obj,
-                                    RecordId loc,
-                                    const InsertDeleteOptions& options,
-                                    int64_t* keysInsertedOut) const {
-    Status status = Status::OK();
-    if (index->isHybridBuilding()) {
-        // The side table interface accepts only records that meet the criteria for this partial
-        // index.
-        // For non-hybrid builds, the decision to use the filter for the partial index is left to
-        // the IndexAccessMethod. See SERVER-28975 for details.
-        if (auto filter = index->getFilterExpression()) {
-            if (!filter->matchesBSON(obj)) {
-                return Status::OK();
-            }
-        }
-
-        int64_t inserted = 0;
-        status = index->indexBuildInterceptor()->sideWrite(opCtx,
-                                                           keys,
-                                                           multikeyMetadataKeys,
-                                                           multikeyPaths,
-                                                           loc,
-                                                           IndexBuildInterceptor::Op::kInsert,
-                                                           &inserted);
-        if (keysInsertedOut) {
-            *keysInsertedOut += inserted;
-        }
-    } else {
-        int64_t numInserted = 0;
-        status = index->accessMethod()->insertKeysAndUpdateMultikeyPaths(
-            opCtx,
-            coll,
-            keys,
-            {multikeyMetadataKeys.begin(), multikeyMetadataKeys.end()},
-            multikeyPaths,
-            loc,
-            options,
-            nullptr,
-            &numInserted);
-        if (keysInsertedOut) {
-            *keysInsertedOut += numInserted;
-        }
-    }
-
-    return status;
-}
 
 Status IndexCatalogImpl::_indexFilteredRecords(OperationContext* opCtx,
                                                const CollectionPtr& coll,
@@ -1528,51 +1475,12 @@ Status IndexCatalogImpl::_indexFilteredRecords(OperationContext* opCtx,
                                                const std::vector<BsonRecord>& bsonRecords,
                                                int64_t* keysInsertedOut) const {
     SharedBufferFragmentBuilder pooledBuilder(KeyString::HeapBuilder::kHeapAllocatorDefaultBytes);
-    auto& executionCtx = StorageExecutionContext::get(opCtx);
 
     InsertDeleteOptions options;
     prepareInsertDeleteOptions(opCtx, coll->ns(), index->descriptor(), &options);
 
-    for (auto bsonRecord : bsonRecords) {
-        invariant(bsonRecord.id != RecordId());
-
-        if (!bsonRecord.ts.isNull()) {
-            Status status = opCtx->recoveryUnit()->setTimestamp(bsonRecord.ts);
-            if (!status.isOK())
-                return status;
-        }
-
-        auto keys = executionCtx.keys();
-        auto multikeyMetadataKeys = executionCtx.multikeyMetadataKeys();
-        auto multikeyPaths = executionCtx.multikeyPaths();
-
-        index->accessMethod()->getKeys(opCtx,
-                                       coll,
-                                       pooledBuilder,
-                                       *bsonRecord.docPtr,
-                                       options.getKeysMode,
-                                       IndexAccessMethod::GetKeysContext::kAddingKeys,
-                                       keys.get(),
-                                       multikeyMetadataKeys.get(),
-                                       multikeyPaths.get(),
-                                       bsonRecord.id);
-
-        Status status = _indexKeys(opCtx,
-                                   coll,
-                                   index,
-                                   *keys,
-                                   *multikeyMetadataKeys,
-                                   *multikeyPaths,
-                                   *bsonRecord.docPtr,
-                                   bsonRecord.id,
-                                   options,
-                                   keysInsertedOut);
-        if (!status.isOK()) {
-            return status;
-        }
-    }
-
-    return Status::OK();
+    return index->accessMethod()->insert(
+        opCtx, pooledBuilder, coll, bsonRecords, options, keysInsertedOut);
 }
 
 Status IndexCatalogImpl::_indexRecords(OperationContext* opCtx,
@@ -1605,36 +1513,16 @@ Status IndexCatalogImpl::_updateRecord(OperationContext* const opCtx,
                                        const RecordId& recordId,
                                        int64_t* const keysInsertedOut,
                                        int64_t* const keysDeletedOut) const {
-    IndexAccessMethod* iam = index->accessMethod();
+    SharedBufferFragmentBuilder pooledBuilder(KeyString::HeapBuilder::kHeapAllocatorDefaultBytes);
 
     InsertDeleteOptions options;
     prepareInsertDeleteOptions(opCtx, coll->ns(), index->descriptor(), &options);
 
-    UpdateTicket updateTicket;
-
-    iam->prepareUpdate(opCtx, coll, index, oldDoc, newDoc, recordId, options, &updateTicket);
-
     int64_t keysInserted = 0;
     int64_t keysDeleted = 0;
 
-    auto status = Status::OK();
-    if (index->isHybridBuilding() || !index->isReady(opCtx, coll)) {
-        bool logIfError = false;
-        _unindexKeys(
-            opCtx, coll, index, updateTicket.removed, oldDoc, recordId, logIfError, &keysDeleted);
-        status = _indexKeys(opCtx,
-                            coll,
-                            index,
-                            updateTicket.added,
-                            updateTicket.newMultikeyMetadataKeys,
-                            updateTicket.newMultikeyPaths,
-                            newDoc,
-                            recordId,
-                            options,
-                            &keysInserted);
-    } else {
-        status = iam->update(opCtx, coll, updateTicket, &keysInserted, &keysDeleted);
-    }
+    auto status = index->accessMethod()->update(
+        opCtx, pooledBuilder, oldDoc, newDoc, recordId, coll, options, &keysInserted, &keysDeleted);
 
     if (!status.isOK())
         return status;
@@ -1645,69 +1533,6 @@ Status IndexCatalogImpl::_updateRecord(OperationContext* const opCtx,
     return Status::OK();
 }
 
-void IndexCatalogImpl::_unindexKeys(OperationContext* opCtx,
-                                    const CollectionPtr& collection,
-                                    const IndexCatalogEntry* index,
-                                    const KeyStringSet& keys,
-                                    const BSONObj& obj,
-                                    RecordId loc,
-                                    bool logIfError,
-                                    int64_t* const keysDeletedOut,
-                                    CheckRecordId checkRecordId) const {
-    InsertDeleteOptions options;
-    prepareInsertDeleteOptions(opCtx, collection->ns(), index->descriptor(), &options);
-    options.logIfError = logIfError;
-
-    if (index->isHybridBuilding()) {
-        // The side table interface accepts only records that meet the criteria for this partial
-        // index.
-        // For non-hybrid builds, the decision to use the filter for the partial index is left to
-        // the IndexAccessMethod. See SERVER-28975 for details.
-        if (auto filter = index->getFilterExpression()) {
-            if (!filter->matchesBSON(obj)) {
-                return;
-            }
-        }
-
-        int64_t removed = 0;
-        fassert(31155,
-                index->indexBuildInterceptor()->sideWrite(
-                    opCtx, keys, {}, {}, loc, IndexBuildInterceptor::Op::kDelete, &removed));
-        if (keysDeletedOut) {
-            *keysDeletedOut += removed;
-        }
-
-        return;
-    }
-
-    // On WiredTiger, we do blind unindexing of records for efficiency.  However, when duplicates
-    // are allowed in unique indexes, WiredTiger does not do blind unindexing, and instead confirms
-    // that the recordid matches the element we are removing.
-    //
-    // We need to disable blind-deletes if 'checkRecordId' is explicitly set 'On', or for
-    // in-progress indexes, in order to force recordid-matching for unindex operations, since
-    // initial sync can build an index over a collection with duplicates. See SERVER-17487 for more
-    // details.
-    options.dupsAllowed = options.dupsAllowed || !index->isReady(opCtx, collection) ||
-        (checkRecordId == CheckRecordId::On);
-
-    int64_t removed = 0;
-    Status status = index->accessMethod()->removeKeys(opCtx, keys, loc, options, &removed);
-
-    if (!status.isOK()) {
-        LOGV2(20362,
-              "Couldn't unindex record {obj} from collection {namespace}: {error}",
-              "Couldn't unindex record",
-              "record"_attr = redact(obj),
-              "namespace"_attr = collection->ns(),
-              "error"_attr = redact(status));
-    }
-
-    if (keysDeletedOut) {
-        *keysDeletedOut += removed;
-    }
-}
-
 void IndexCatalogImpl::_unindexRecord(OperationContext* opCtx,
                                       const CollectionPtr& collection,
                                       const IndexCatalogEntry* entry,
@@ -1716,24 +1541,6 @@ void IndexCatalogImpl::_unindexRecord(OperationContext* opCtx,
                                       bool logIfError,
                                       int64_t* keysDeletedOut,
                                       CheckRecordId checkRecordId) const {
-    SharedBufferFragmentBuilder pooledBuilder(KeyString::HeapBuilder::kHeapAllocatorDefaultBytes);
-    auto& executionCtx = StorageExecutionContext::get(opCtx);
-
-    // There's no need to compute the prefixes of the indexed fields that cause the index to be
-    // multikey when removing a document since the index metadata isn't updated when keys are
-    // deleted.
-    auto keys = executionCtx.keys();
-    entry->accessMethod()->getKeys(opCtx,
-                                   collection,
-                                   pooledBuilder,
-                                   obj,
-                                   IndexAccessMethod::GetKeysMode::kRelaxConstraintsUnfiltered,
-                                   IndexAccessMethod::GetKeysContext::kRemovingKeys,
-                                   keys.get(),
-                                   nullptr,
-                                   nullptr,
-                                   loc);
-
     // Tests can enable this failpoint to produce index corruption scenarios where an index has
     // extra keys.
     if (auto failpoint = skipUnindexingDocumentWhenDeleted.scoped();
@@ -1743,8 +1550,21 @@ void IndexCatalogImpl::_unindexRecord(OperationContext* opCtx,
             return;
         }
     }
-    _unindexKeys(
-        opCtx, collection, entry, *keys, obj, loc, logIfError, keysDeletedOut, checkRecordId);
+
+    SharedBufferFragmentBuilder pooledBuilder(KeyString::HeapBuilder::kHeapAllocatorDefaultBytes);
+
+    InsertDeleteOptions options;
+    prepareInsertDeleteOptions(opCtx, collection->ns(), entry->descriptor(), &options);
+
+    entry->accessMethod()->remove(opCtx,
+                                  pooledBuilder,
+                                  collection,
+                                  obj,
+                                  loc,
+                                  logIfError,
+                                  options,
+                                  keysDeletedOut,
+                                  checkRecordId);
 }
 
 Status IndexCatalogImpl::indexRecords(OperationContext* opCtx,
@@ -1831,7 +1651,7 @@ void IndexCatalogImpl::unindexRecord(OperationContext* opCtx,
         IndexCatalogEntry* entry = it->get();
 
         // If it's a background index, we DO NOT want to log anything.
-        bool logIfError = entry->isReady(opCtx, collection) ? !noWarn : false;
+        bool logIfError = entry->isReady(opCtx) ? !noWarn : false;
         _unindexRecord(
             opCtx, collection, entry, obj, loc, logIfError, keysDeletedOut, checkRecordId);
     }
@@ -1886,9 +1706,9 @@ void IndexCatalogImpl::prepareInsertDeleteOptions(OperationContext* opCtx,
                                                   InsertDeleteOptions* options) const {
     auto replCoord = repl::ReplicationCoordinator::get(opCtx);
     if (replCoord->shouldRelaxIndexConstraints(opCtx, ns)) {
-        options->getKeysMode = IndexAccessMethod::GetKeysMode::kRelaxConstraints;
+        options->getKeysMode = InsertDeleteOptions::ConstraintEnforcementMode::kRelaxConstraints;
     } else {
-        options->getKeysMode = IndexAccessMethod::GetKeysMode::kEnforceConstraints;
+        options->getKeysMode = InsertDeleteOptions::ConstraintEnforcementMode::kEnforceConstraints;
     }
 
     // Don't allow dups for Id key. Allow dups for non-unique keys or when constraints relaxed.
@@ -1896,7 +1716,8 @@ void IndexCatalogImpl::prepareInsertDeleteOptions(OperationContext* opCtx,
         options->dupsAllowed = false;
     } else {
         options->dupsAllowed = !desc->unique() ||
-            options->getKeysMode == IndexAccessMethod::GetKeysMode::kRelaxConstraints;
+            options->getKeysMode ==
+                InsertDeleteOptions::ConstraintEnforcementMode::kRelaxConstraints;
     }
 }
 
diff --git a/src/mongo/db/catalog/index_catalog_impl.h b/src/mongo/db/catalog/index_catalog_impl.h
index e1579f023ef..bcdfd51fcbb 100644
--- a/src/mongo/db/catalog/index_catalog_impl.h
+++ b/src/mongo/db/catalog/index_catalog_impl.h
@@ -329,17 +329,6 @@ private:
      */
     std::string _getAccessMethodName(const BSONObj& keyPattern) const;
 
-    Status _indexKeys(OperationContext* opCtx,
-                      const CollectionPtr& coll,
-                      const IndexCatalogEntry* index,
-                      const KeyStringSet& keys,
-                      const KeyStringSet& multikeyMetadataKeys,
-                      const MultikeyPaths& multikeyPaths,
-                      const BSONObj& obj,
-                      RecordId loc,
-                      const InsertDeleteOptions& options,
-                      int64_t* keysInsertedOut) const;
-
     Status _indexFilteredRecords(OperationContext* opCtx,
                                  const CollectionPtr& coll,
                                  const IndexCatalogEntry* index,
@@ -361,16 +350,6 @@ private:
                          int64_t* keysInsertedOut,
                          int64_t* keysDeletedOut) const;
 
-    void _unindexKeys(OperationContext* opCtx,
-                      const CollectionPtr& collection,
-                      const IndexCatalogEntry* index,
-                      const KeyStringSet& keys,
-                      const BSONObj& obj,
-                      RecordId loc,
-                      bool logIfError,
-                      int64_t* keysDeletedOut,
-                      CheckRecordId checkRecordId = CheckRecordId::Off) const;
-
     void _unindexRecord(OperationContext* opCtx,
                         const CollectionPtr& collection,
                         const IndexCatalogEntry* entry,
diff --git a/src/mongo/db/catalog/index_consistency.cpp b/src/mongo/db/catalog/index_consistency.cpp
index 141a54cf00b..55e35b744f8 100644
--- a/src/mongo/db/catalog/index_consistency.cpp
+++ b/src/mongo/db/catalog/index_consistency.cpp
@@ -405,8 +405,8 @@ void IndexConsistency::addIndexKey(OperationContext* opCtx,
                 writeConflictRetry(
                     opCtx, "removingExtraIndexEntries", _validateState->nss().ns(), [&] {
                         WriteUnitOfWork wunit(opCtx);
-                        Status status = indexInfo->accessMethod->removeKeys(
-                            opCtx, {ks}, recordId, options, &numDeleted);
+                        Status status = indexInfo->accessMethod->asSortedData()->removeKeys(
+                            opCtx, {ks}, options, &numDeleted);
                         wunit.commit();
                     });
                 auto& indexResults = results->indexResultsMap[indexInfo->indexName];
diff --git a/src/mongo/db/catalog/index_repair.cpp b/src/mongo/db/catalog/index_repair.cpp
index a528d682c0e..6effefcd909 100644
--- a/src/mongo/db/catalog/index_repair.cpp
+++ b/src/mongo/db/catalog/index_repair.cpp
@@ -118,15 +118,7 @@ int repairMissingIndexEntry(OperationContext* opCtx,
                             const NamespaceString& nss,
                             const CollectionPtr& coll,
                             ValidateResults* results) {
-    RecordId rid;
-    if (keyFormat == KeyFormat::Long) {
-        rid = KeyString::decodeRecordIdLongAtEnd(ks.getBuffer(), ks.getSize());
-    } else {
-        invariant(keyFormat == KeyFormat::String);
-        rid = KeyString::decodeRecordIdStrAtEnd(ks.getBuffer(), ks.getSize());
-    }
-
-    IndexAccessMethod* accessMethod = const_cast<IndexAccessMethod*>(index->accessMethod());
+    auto accessMethod = const_cast<IndexAccessMethod*>(index->accessMethod())->asSortedData();
     InsertDeleteOptions options;
     options.dupsAllowed = !index->descriptor()->unique();
     int64_t numInserted = 0;
@@ -136,7 +128,7 @@ int repairMissingIndexEntry(OperationContext* opCtx,
         // Ignore return status because we will use numInserted to verify success.
         accessMethod
             ->insertKeysAndUpdateMultikeyPaths(
-                opCtx, coll, {ks}, {}, {}, rid, options, nullptr, &numInserted)
+                opCtx, coll, {ks}, {}, {}, options, nullptr, &numInserted)
             .ignore();
         wunit.commit();
     });
@@ -151,6 +143,14 @@ int repairMissingIndexEntry(OperationContext* opCtx,
         results->numInsertedMissingIndexEntries += numInserted;
         results->repaired = true;
     } else {
+        RecordId rid;
+        if (keyFormat == KeyFormat::Long) {
+            rid = KeyString::decodeRecordIdLongAtEnd(ks.getBuffer(), ks.getSize());
+        } else {
+            invariant(keyFormat == KeyFormat::String);
+            rid = KeyString::decodeRecordIdStrAtEnd(ks.getBuffer(), ks.getSize());
+        }
+
         // Move the duplicate document of the missing index entry from the record store to the lost
         // and found.
         Snapshotted<BSONObj> doc;
diff --git a/src/mongo/db/catalog/multi_index_block.cpp b/src/mongo/db/catalog/multi_index_block.cpp
index 40a6187312d..84dd6a29067 100644
--- a/src/mongo/db/catalog/multi_index_block.cpp
+++ b/src/mongo/db/catalog/multi_index_block.cpp
@@ -326,7 +326,8 @@ StatusWith<std::vector<BSONObj>> MultiIndexBlock::init(
                 opCtx, collection->ns(), descriptor, &index.options);
 
             // Index builds always relax constraints and check for violations at commit-time.
-            index.options.getKeysMode = IndexAccessMethod::GetKeysMode::kRelaxConstraints;
+            index.options.getKeysMode =
+                InsertDeleteOptions::ConstraintEnforcementMode::kRelaxConstraints;
             index.options.dupsAllowed = true;
             index.options.fromIndexBuilder = true;
 
@@ -782,13 +783,12 @@ Status MultiIndexBlock::dumpInsertsFromBulk(
                     "index"_attr = entry->descriptor()->indexName(),
                     "buildUUID"_attr = _buildUUID);
 
-        // SERVER-41918 This call to commitBulk() results in file I/O that may result in an
+        // SERVER-41918 This call to bulk->commit() results in file I/O that may result in an
         // exception.
         try {
-            Status status = _indexes[i].real->commitBulk(
+            Status status = _indexes[i].bulk->commit(
                 opCtx,
                 collection,
-                _indexes[i].bulk.get(),
                 dupsAllowed,
                 kYieldIterations,
                 [=](const KeyString::Value& duplicateKey) {
@@ -1087,18 +1087,7 @@ BSONObj MultiIndexBlock::_constructStateObject(OperationContext* opCtx,
         if (_phase != IndexBuildPhaseEnum::kDrainWrites) {
             // Persist the data to disk so that we see all of the data that has been inserted into
             // the Sorter.
-            auto state = index.bulk->persistDataForShutdown();
-
-            indexInfo.append("fileName", state.fileName);
-            indexInfo.append("numKeys", index.bulk->getKeysInserted());
-
-            BSONArrayBuilder ranges(indexInfo.subarrayStart("ranges"));
-            for (const auto& rangeInfo : state.ranges) {
-                BSONObjBuilder range(ranges.subobjStart());
-                range.append("startOffset", rangeInfo.getStartOffset());
-                range.append("endOffset", rangeInfo.getEndOffset());
-                range.append("checksum", rangeInfo.getChecksum());
-            }
+            index.bulk->persistDataForShutdown(indexInfo);
         }
 
         auto indexBuildInterceptor =
diff --git a/src/mongo/db/catalog/throttle_cursor.cpp b/src/mongo/db/catalog/throttle_cursor.cpp
index ab6077fd167..f8b7a670bf3 100644
--- a/src/mongo/db/catalog/throttle_cursor.cpp
+++ b/src/mongo/db/catalog/throttle_cursor.cpp
@@ -71,9 +71,8 @@ boost::optional<Record> SeekableRecordThrottleCursor::next(OperationContext* opC
     return record;
 }
 
-SortedDataInterfaceThrottleCursor::SortedDataInterfaceThrottleCursor(OperationContext* opCtx,
-                                                                     const IndexAccessMethod* iam,
-                                                                     DataThrottle* dataThrottle) {
+SortedDataInterfaceThrottleCursor::SortedDataInterfaceThrottleCursor(
+    OperationContext* opCtx, const SortedDataIndexAccessMethod* iam, DataThrottle* dataThrottle) {
     _cursor = iam->newCursor(opCtx, /*forward=*/true);
     _dataThrottle = dataThrottle;
 }
diff --git a/src/mongo/db/catalog/throttle_cursor.h b/src/mongo/db/catalog/throttle_cursor.h
index d1551fb8191..1b995989dcf 100644
--- a/src/mongo/db/catalog/throttle_cursor.h
+++ b/src/mongo/db/catalog/throttle_cursor.h
@@ -85,7 +85,7 @@ private:
 class SortedDataInterfaceThrottleCursor {
 public:
     SortedDataInterfaceThrottleCursor(OperationContext* opCtx,
-                                      const IndexAccessMethod* iam,
+                                      const SortedDataIndexAccessMethod* iam,
                                       DataThrottle* dataThrottle);
 
     boost::optional<IndexKeyEntry> seek(OperationContext* opCtx, const KeyString::Value& key);
diff --git a/src/mongo/db/catalog/throttle_cursor_test.cpp b/src/mongo/db/catalog/throttle_cursor_test.cpp
index 97db1fffad4..2f272e3eeef 100644
--- a/src/mongo/db/catalog/throttle_cursor_test.cpp
+++ b/src/mongo/db/catalog/throttle_cursor_test.cpp
@@ -37,6 +37,7 @@
 #include "mongo/db/catalog/index_catalog_entry.h"
 #include "mongo/db/catalog/validate_gen.h"
 #include "mongo/db/db_raii.h"
+#include "mongo/db/index/index_access_method.h"
 #include "mongo/unittest/unittest.h"
 #include "mongo/util/clock_source_mock.h"
 #include "mongo/util/time_support.h"
@@ -110,7 +111,7 @@ int64_t ThrottleCursorTest::getDifferenceInMillis(Date_t start, Date_t end) {
 SortedDataInterfaceThrottleCursor ThrottleCursorTest::getIdIndex(const CollectionPtr& coll) {
     const IndexDescriptor* idDesc = coll->getIndexCatalog()->findIdIndex(operationContext());
     const IndexCatalogEntry* idEntry = coll->getIndexCatalog()->getEntry(idDesc);
-    const IndexAccessMethod* iam = idEntry->accessMethod();
+    auto iam = idEntry->accessMethod()->asSortedData();
 
     return SortedDataInterfaceThrottleCursor(operationContext(), iam, _dataThrottle.get());
 }
diff --git a/src/mongo/db/catalog/validate_adaptor.cpp b/src/mongo/db/catalog/validate_adaptor.cpp
index 7d1d66ad3c5..5f171fccb94 100644
--- a/src/mongo/db/catalog/validate_adaptor.cpp
+++ b/src/mongo/db/catalog/validate_adaptor.cpp
@@ -133,7 +133,7 @@ Status ValidateAdaptor::validateRecord(OperationContext* opCtx,
 
     for (const auto& index : _validateState->getIndexes()) {
         const IndexDescriptor* descriptor = index->descriptor();
-        const IndexAccessMethod* iam = index->accessMethod();
+        auto iam = index->accessMethod()->asSortedData();
 
         if (descriptor->isPartial() && !index->getFilterExpression()->matchesBSON(recordBson))
             continue;
@@ -146,8 +146,8 @@ Status ValidateAdaptor::validateRecord(OperationContext* opCtx,
                      coll,
                      pool,
                      recordBson,
-                     IndexAccessMethod::GetKeysMode::kEnforceConstraints,
-                     IndexAccessMethod::GetKeysContext::kAddingKeys,
+                     InsertDeleteOptions::ConstraintEnforcementMode::kEnforceConstraints,
+                     SortedDataIndexAccessMethod::GetKeysContext::kAddingKeys,
                      documentKeySet.get(),
                      multikeyMetadataKeys.get(),
                      documentMultikeyPaths.get(),
@@ -317,7 +317,7 @@ void ValidateAdaptor::traverseIndex(OperationContext* opCtx,
     }
 
     const KeyString::Version version =
-        index->accessMethod()->getSortedDataInterface()->getKeyStringVersion();
+        index->accessMethod()->asSortedData()->getSortedDataInterface()->getKeyStringVersion();
 
     KeyString::Builder firstKeyStringBuilder(
         version, BSONObj(), indexInfo.ord, KeyString::Discriminator::kExclusiveBefore);
@@ -344,7 +344,8 @@ void ValidateAdaptor::traverseIndex(OperationContext* opCtx,
         throw;
     }
 
-    const auto keyFormat = index->accessMethod()->getSortedDataInterface()->rsKeyFormat();
+    const auto keyFormat =
+        index->accessMethod()->asSortedData()->getSortedDataInterface()->rsKeyFormat();
     const RecordId kWildcardMultikeyMetadataRecordId = record_id_helpers::reservedIdFor(
         record_id_helpers::ReservationId::kWildcardMultikeyMetadataId, keyFormat);
     while (indexEntry) {
diff --git a/src/mongo/db/catalog/validate_state.cpp b/src/mongo/db/catalog/validate_state.cpp
index 4cf13eb010b..3b1b6db8764 100644
--- a/src/mongo/db/catalog/validate_state.cpp
+++ b/src/mongo/db/catalog/validate_state.cpp
@@ -263,10 +263,13 @@ void ValidateState::initializeCursors(OperationContext* opCtx) {
         const IndexCatalogEntry* entry = it->next();
         const IndexDescriptor* desc = entry->descriptor();
 
-        _indexCursors.emplace(desc->indexName(),
-                              std::make_unique<SortedDataInterfaceThrottleCursor>(
-                                  opCtx, entry->accessMethod(), &_dataThrottle));
+        auto iam = entry->accessMethod()->asSortedData();
+        if (!iam)
+            continue;
 
+        _indexCursors.emplace(
+            desc->indexName(),
+            std::make_unique<SortedDataInterfaceThrottleCursor>(opCtx, iam, &_dataThrottle));
 
         _indexes.push_back(indexCatalog->getEntryShared(desc));
     }
diff --git a/src/mongo/db/dbhelpers.cpp b/src/mongo/db/dbhelpers.cpp
index 1933a6907ef..3da676da060 100644
--- a/src/mongo/db/dbhelpers.cpp
+++ b/src/mongo/db/dbhelpers.cpp
@@ -184,8 +184,8 @@ bool Helpers::findById(OperationContext* opCtx,
         return false;
     }
 
-    auto recordId =
-        catalog->getEntry(desc)->accessMethod()->findSingle(opCtx, collection, query["_id"].wrap());
+    auto recordId = catalog->getEntry(desc)->accessMethod()->asSortedData()->findSingle(
+        opCtx, collection, query["_id"].wrap());
     if (recordId.isNull())
         return false;
     result = collection->docFor(opCtx, recordId).value();
@@ -206,7 +206,7 @@ RecordId Helpers::findById(OperationContext* opCtx,
     }
 
     uassert(13430, "no _id index", desc);
-    return catalog->getEntry(desc)->accessMethod()->findSingle(
+    return catalog->getEntry(desc)->accessMethod()->asSortedData()->findSingle(
         opCtx, collection, idquery["_id"].wrap());
 }
 
diff --git a/src/mongo/db/exec/idhack.cpp b/src/mongo/db/exec/idhack.cpp
index 9d55efa9edf..091a1a5381e 100644
--- a/src/mongo/db/exec/idhack.cpp
+++ b/src/mongo/db/exec/idhack.cpp
@@ -86,7 +86,8 @@ PlanStage::StageState IDHackStage::doWork(WorkingSetID* out) {
     WorkingSetID id = WorkingSet::INVALID_ID;
     try {
         // Look up the key by going directly to the index.
-        auto recordId = indexAccessMethod()->findSingle(opCtx(), collection(), _key);
+        auto recordId =
+            indexAccessMethod()->asSortedData()->findSingle(opCtx(), collection(), _key);
 
         // Key not found.
         if (recordId.isNull()) {
diff --git a/src/mongo/db/exec/requires_index_stage.cpp b/src/mongo/db/exec/requires_index_stage.cpp
index 9b7115b4ccc..c1a0763f9f6 100644
--- a/src/mongo/db/exec/requires_index_stage.cpp
+++ b/src/mongo/db/exec/requires_index_stage.cpp
@@ -42,7 +42,7 @@ RequiresIndexStage::RequiresIndexStage(const char* stageType,
       _weakIndexCatalogEntry(indexDescriptor->getEntry()->shared_from_this()) {
     auto indexCatalogEntry = _weakIndexCatalogEntry.lock();
     _indexDescriptor = indexCatalogEntry->descriptor();
-    _indexAccessMethod = indexCatalogEntry->accessMethod();
+    _indexAccessMethod = indexCatalogEntry->accessMethod()->asSortedData();
     invariant(_indexDescriptor);
     invariant(_indexAccessMethod);
     _indexName = _indexDescriptor->indexName();
@@ -70,7 +70,7 @@ void RequiresIndexStage::doRestoreStateRequiresCollection() {
     // access the catalog entry by raw pointer when the query is active, as its validity is
     // protected by at least MODE_IS collection locks.
     _indexDescriptor = indexCatalogEntry->descriptor();
-    _indexAccessMethod = indexCatalogEntry->accessMethod();
+    _indexAccessMethod = indexCatalogEntry->accessMethod()->asSortedData();
     invariant(_indexDescriptor);
     invariant(_indexAccessMethod);
 
diff --git a/src/mongo/db/exec/requires_index_stage.h b/src/mongo/db/exec/requires_index_stage.h
index 37257e98084..5fc1438e106 100644
--- a/src/mongo/db/exec/requires_index_stage.h
+++ b/src/mongo/db/exec/requires_index_stage.h
@@ -31,6 +31,7 @@
 
 #include "mongo/db/exec/requires_collection_stage.h"
 #include "mongo/db/exec/working_set.h"
+#include "mongo/db/index/index_access_method.h"
 #include "mongo/db/index/index_descriptor.h"
 
 namespace mongo {
@@ -74,7 +75,7 @@ protected:
         return _indexDescriptor;
     }
 
-    const IndexAccessMethod* indexAccessMethod() const {
+    const SortedDataIndexAccessMethod* indexAccessMethod() const {
         return _indexAccessMethod;
     }
 
@@ -93,7 +94,7 @@ private:
     std::weak_ptr<const IndexCatalogEntry> _weakIndexCatalogEntry;
 
     const IndexDescriptor* _indexDescriptor;
-    const IndexAccessMethod* _indexAccessMethod;
+    const SortedDataIndexAccessMethod* _indexAccessMethod;
 
     std::string _indexName;
 
diff --git a/src/mongo/db/exec/sbe/stages/ix_scan.cpp b/src/mongo/db/exec/sbe/stages/ix_scan.cpp
index ebd2efcc5d6..5fdfafd2c05 100644
--- a/src/mongo/db/exec/sbe/stages/ix_scan.cpp
+++ b/src/mongo/db/exec/sbe/stages/ix_scan.cpp
@@ -275,7 +275,7 @@ void IndexScanStage::open(bool reOpen) {
             str::stream() << "expected IndexCatalogEntry for index named: " << _indexName,
             static_cast<bool>(entry));
     if (!_cursor) {
-        _cursor = entry->accessMethod()->getSortedDataInterface()->newCursor(_opCtx, _forward);
+        _cursor = entry->accessMethod()->asSortedData()->newCursor(_opCtx, _forward);
     }
 
     if (_seekKeyLowAccessor && _seekKeyHiAccessor) {
@@ -301,7 +301,7 @@ void IndexScanStage::open(bool reOpen) {
                 tagLow == value::TypeTags::ksValue);
         _seekKeyLowHolder->reset(false, tagLow, valLow);
     } else {
-        auto sdi = entry->accessMethod()->getSortedDataInterface();
+        auto sdi = entry->accessMethod()->asSortedData()->getSortedDataInterface();
         KeyString::Builder kb(sdi->getKeyStringVersion(),
                               sdi->getOrdering(),
                               KeyString::Discriminator::kExclusiveBefore);
diff --git a/src/mongo/db/exec/working_set_common.cpp b/src/mongo/db/exec/working_set_common.cpp
index 98b70ccde37..1a1a04cade6 100644
--- a/src/mongo/db/exec/working_set_common.cpp
+++ b/src/mongo/db/exec/working_set_common.cpp
@@ -145,13 +145,13 @@ bool WorkingSetCommon::fetch(OperationContext* opCtx,
             // index to be multikey when ensuring the keyData is still valid.
             KeyStringSet* multikeyMetadataKeys = nullptr;
             MultikeyPaths* multikeyPaths = nullptr;
-            auto* iam = workingSet->retrieveIndexAccessMethod(memberKey.indexId);
+            auto* iam = workingSet->retrieveIndexAccessMethod(memberKey.indexId)->asSortedData();
             iam->getKeys(opCtx,
                          collection,
                          pool,
                          member->doc.value().toBson(),
-                         IndexAccessMethod::GetKeysMode::kEnforceConstraints,
-                         IndexAccessMethod::GetKeysContext::kValidatingKeys,
+                         InsertDeleteOptions::ConstraintEnforcementMode::kEnforceConstraints,
+                         SortedDataIndexAccessMethod::GetKeysContext::kValidatingKeys,
                          keys.get(),
                          multikeyMetadataKeys,
                          multikeyPaths,
diff --git a/src/mongo/db/index/SConscript b/src/mongo/db/index/SConscript
index 9d3e92da6dd..e8e12550fb6 100644
--- a/src/mongo/db/index/SConscript
+++ b/src/mongo/db/index/SConscript
@@ -20,33 +20,6 @@ env.Library(
 )
 
 env.Library(
-    target='duplicate_key_tracker',
-    source=[
-        'duplicate_key_tracker.cpp',
-    ],
-    LIBDEPS_PRIVATE=[
-        '$BUILD_DIR/mongo/base',
-        '$BUILD_DIR/mongo/db/curop',
-        '$BUILD_DIR/mongo/db/service_context',
-        '$BUILD_DIR/mongo/db/storage/key_string',
-    ],
-)
-
-env.Library(
-    target='skipped_record_tracker',
-    source=[
-        'skipped_record_tracker.cpp',
-    ],
-    LIBDEPS_PRIVATE=[
-        '$BUILD_DIR/mongo/base',
-        '$BUILD_DIR/mongo/db/concurrency/write_conflict_exception',
-        '$BUILD_DIR/mongo/db/curop',
-        '$BUILD_DIR/mongo/db/service_context',
-        '$BUILD_DIR/mongo/db/storage/execution_context',
-    ],
-)
-
-env.Library(
     target='key_generator',
     source=[
         'btree_key_generator.cpp',
@@ -118,14 +91,20 @@ serveronlyEnv.InjectThirdParty(libraries=['snappy'])
 serveronlyEnv.Library(
     target="index_access_method",
     source=[
-        "index_access_method.cpp"
+        'duplicate_key_tracker.cpp',
+        'index_access_method.cpp',
+        'index_build_interceptor.cpp',
+        'index_build_interceptor.idl',
+        'skipped_record_tracker.cpp',
     ],
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/base',
         '$BUILD_DIR/mongo/db/catalog/index_catalog_entry',
         '$BUILD_DIR/mongo/db/concurrency/write_conflict_exception',
         '$BUILD_DIR/mongo/db/curop',
+        '$BUILD_DIR/mongo/db/multi_key_path_tracker',
         '$BUILD_DIR/mongo/db/repl/repl_coordinator_interface',
+        '$BUILD_DIR/mongo/db/service_context',
         '$BUILD_DIR/mongo/db/sorter/sorter_idl',
         '$BUILD_DIR/mongo/db/storage/encryption_hooks',
         '$BUILD_DIR/mongo/db/storage/execution_context',
@@ -134,9 +113,9 @@ serveronlyEnv.Library(
         '$BUILD_DIR/mongo/db/storage/storage_options',
         '$BUILD_DIR/mongo/db/vector_clock',
         '$BUILD_DIR/mongo/idl/server_parameter',
+        '$BUILD_DIR/mongo/util/progress_meter',
         '$BUILD_DIR/third_party/shim_snappy',
         'index_descriptor',
-        'skipped_record_tracker',
     ],
 )
 
@@ -152,36 +131,18 @@ env.Library(
         "s2_bucket_access_method.cpp",
         "wildcard_access_method.cpp",
     ],
+    LIBDEPS=[
+        'index_access_method',
+    ],
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/base',
         '$BUILD_DIR/mongo/db/fts/base_fts',
         '$BUILD_DIR/mongo/db/index_names',
         'expression_params',
-        'index_access_method',
         'key_generator',
     ]
 )
 
-env.Library(
-    target="index_build_interceptor",
-    source=[
-        "index_build_interceptor.cpp",
-        'index_build_interceptor.idl',
-    ],
-    LIBDEPS_PRIVATE=[
-        '$BUILD_DIR/mongo/base',
-        '$BUILD_DIR/mongo/db/concurrency/write_conflict_exception',
-        '$BUILD_DIR/mongo/db/curop',
-        '$BUILD_DIR/mongo/db/multi_key_path_tracker',
-        '$BUILD_DIR/mongo/db/service_context',
-        '$BUILD_DIR/mongo/db/storage/key_string',
-        '$BUILD_DIR/mongo/util/progress_meter',
-        'duplicate_key_tracker',
-        'index_access_methods',
-        'skipped_record_tracker',
-    ],
-)
-
 env.CppUnitTest(
     target='db_index_test',
     source=[
diff --git a/src/mongo/db/index/btree_access_method.cpp b/src/mongo/db/index/btree_access_method.cpp
index 149596739d8..93297808aec 100644
--- a/src/mongo/db/index/btree_access_method.cpp
+++ b/src/mongo/db/index/btree_access_method.cpp
@@ -81,7 +81,7 @@ void BtreeAccessMethod::doGetKeys(OperationContext* opCtx,
                                   KeyStringSet* multikeyMetadataKeys,
                                   MultikeyPaths* multikeyPaths,
                                   boost::optional<RecordId> id) const {
-    const auto skipMultikey = context == IndexAccessMethod::GetKeysContext::kValidatingKeys &&
+    const auto skipMultikey = context == GetKeysContext::kValidatingKeys &&
         !_descriptor->getEntry()->isMultikey(opCtx, collection);
     _keyGenerator->getKeys(pooledBufferBuilder, obj, skipMultikey, keys, multikeyPaths, id);
 }
diff --git a/src/mongo/db/index/duplicate_key_tracker.cpp b/src/mongo/db/index/duplicate_key_tracker.cpp
index 67ccb738c20..3b6709bad75 100644
--- a/src/mongo/db/index/duplicate_key_tracker.cpp
+++ b/src/mongo/db/index/duplicate_key_tracker.cpp
@@ -86,7 +86,10 @@ Status DuplicateKeyTracker::recordKey(OperationContext* opCtx, const KeyString::
     // we exclude it from the serialization.
     BufBuilder builder;
     if (KeyFormat::Long ==
-        _indexCatalogEntry->accessMethod()->getSortedDataInterface()->rsKeyFormat()) {
+        _indexCatalogEntry->accessMethod()
+            ->asSortedData()
+            ->getSortedDataInterface()
+            ->rsKeyFormat()) {
         key.serializeWithoutRecordIdLong(builder);
     } else {
         key.serializeWithoutRecordIdStr(builder);
@@ -116,7 +119,7 @@ Status DuplicateKeyTracker::checkConstraints(OperationContext* opCtx) const {
     auto constraintsCursor = _keyConstraintsTable->rs()->getCursor(opCtx);
     auto record = constraintsCursor->next();
 
-    auto index = _indexCatalogEntry->accessMethod()->getSortedDataInterface();
+    auto index = _indexCatalogEntry->accessMethod()->asSortedData()->getSortedDataInterface();
 
     static const char* curopMessage = "Index Build: checking for duplicate keys";
     ProgressMeterHolder progress;
diff --git a/src/mongo/db/index/expression_keys_private.cpp b/src/mongo/db/index/expression_keys_private.cpp
index 0ca6e27729e..43dadc4adb1 100644
--- a/src/mongo/db/index/expression_keys_private.cpp
+++ b/src/mongo/db/index/expression_keys_private.cpp
@@ -122,11 +122,11 @@ Status S2GetKeysForElement(const BSONElement& element,
 void appendToS2Keys(const std::vector<KeyString::HeapBuilder>& existingKeys,
                     std::vector<KeyString::HeapBuilder>* out,
                     KeyString::Version keyStringVersion,
-                    IndexAccessMethod::GetKeysContext context,
+                    SortedDataIndexAccessMethod::GetKeysContext context,
                     Ordering ordering,
                     size_t maxKeys,
                     const std::function<void(KeyString::HeapBuilder&)>& fn) {
-    if (context == IndexAccessMethod::GetKeysContext::kAddingKeys &&
+    if (context == SortedDataIndexAccessMethod::GetKeysContext::kAddingKeys &&
         existingKeys.size() + out->size() > maxKeys) {
         if (!relaxIndexMaxNumGeneratedKeysPerDocument.shouldFail()) {
             throw MaxKeysExceededException();
@@ -160,7 +160,7 @@ bool getS2GeoKeys(const BSONObj& document,
                   const std::vector<KeyString::HeapBuilder>& keysToAdd,
                   std::vector<KeyString::HeapBuilder>* out,
                   KeyString::Version keyStringVersion,
-                  IndexAccessMethod::GetKeysContext context,
+                  SortedDataIndexAccessMethod::GetKeysContext context,
                   Ordering ordering,
                   size_t maxKeys) {
     bool everGeneratedMultipleCells = false;
@@ -177,7 +177,7 @@ bool getS2GeoKeys(const BSONObj& document,
 
         // We'll be taking the cartesian product of cells and keysToAdd, make sure the output won't
         // be too big.
-        if (context == IndexAccessMethod::GetKeysContext::kAddingKeys &&
+        if (context == SortedDataIndexAccessMethod::GetKeysContext::kAddingKeys &&
             cells.size() * keysToAdd.size() > maxKeys) {
             if (!relaxIndexMaxNumGeneratedKeysPerDocument.shouldFail()) {
                 throw MaxKeysExceededException();
@@ -216,7 +216,7 @@ bool getS2BucketGeoKeys(const BSONObj& document,
                         const std::vector<KeyString::HeapBuilder>& keysToAdd,
                         std::vector<KeyString::HeapBuilder>* out,
                         KeyString::Version keyStringVersion,
-                        IndexAccessMethod::GetKeysContext context,
+                        SortedDataIndexAccessMethod::GetKeysContext context,
                         Ordering ordering,
                         size_t maxKeys) {
     bool generatedMultipleCells = false;
@@ -271,7 +271,7 @@ bool getS2BucketGeoKeys(const BSONObj& document,
 
         // We'll be taking the cartesian product of cells and keysToAdd, make sure the output won't
         // be too big.
-        if (context == IndexAccessMethod::GetKeysContext::kAddingKeys &&
+        if (context == SortedDataIndexAccessMethod::GetKeysContext::kAddingKeys &&
             cells.size() * keysToAdd.size() > maxKeys) {
             if (!relaxIndexMaxNumGeneratedKeysPerDocument.shouldFail()) {
                 throw MaxKeysExceededException();
@@ -307,7 +307,7 @@ void getS2LiteralKeysArray(const BSONObj& obj,
                            const std::vector<KeyString::HeapBuilder>& keysToAdd,
                            std::vector<KeyString::HeapBuilder>* out,
                            KeyString::Version keyStringVersion,
-                           IndexAccessMethod::GetKeysContext context,
+                           SortedDataIndexAccessMethod::GetKeysContext context,
                            Ordering ordering,
                            size_t maxKeys) {
     BSONObjIterator objIt(obj);
@@ -354,7 +354,7 @@ bool getS2OneLiteralKey(const BSONElement& elt,
                         const std::vector<KeyString::HeapBuilder>& keysToAdd,
                         std::vector<KeyString::HeapBuilder>* out,
                         KeyString::Version keyStringVersion,
-                        IndexAccessMethod::GetKeysContext context,
+                        SortedDataIndexAccessMethod::GetKeysContext context,
                         Ordering ordering,
                         size_t maxKeys) {
     if (Array == elt.type()) {
@@ -394,7 +394,7 @@ bool getS2LiteralKeys(const BSONElementSet& elements,
                       const std::vector<KeyString::HeapBuilder>& keysToAdd,
                       std::vector<KeyString::HeapBuilder>* out,
                       KeyString::Version keyStringVersion,
-                      IndexAccessMethod::GetKeysContext context,
+                      SortedDataIndexAccessMethod::GetKeysContext context,
                       Ordering ordering,
                       size_t maxKeys) {
     bool foundIndexedArrayValue = false;
@@ -659,7 +659,7 @@ void ExpressionKeysPrivate::getS2Keys(SharedBufferFragmentBuilder& pooledBufferB
                                       KeyStringSet* keys,
                                       MultikeyPaths* multikeyPaths,
                                       KeyString::Version keyStringVersion,
-                                      IndexAccessMethod::GetKeysContext context,
+                                      SortedDataIndexAccessMethod::GetKeysContext context,
                                       Ordering ordering,
                                       boost::optional<RecordId> id) {
     std::vector<KeyString::HeapBuilder> keysToAdd;
diff --git a/src/mongo/db/index/expression_keys_private.h b/src/mongo/db/index/expression_keys_private.h
index 3c32d5533c4..dae982831ef 100644
--- a/src/mongo/db/index/expression_keys_private.h
+++ b/src/mongo/db/index/expression_keys_private.h
@@ -131,7 +131,7 @@ public:
                           KeyStringSet* keys,
                           MultikeyPaths* multikeyPaths,
                           KeyString::Version keyStringVersion,
-                          IndexAccessMethod::GetKeysContext context,
+                          SortedDataIndexAccessMethod::GetKeysContext context,
                           Ordering ordering,
                           boost::optional<RecordId> id = boost::none);
 };
diff --git a/src/mongo/db/index/index_access_method.cpp b/src/mongo/db/index/index_access_method.cpp
index b75d6711ae9..52e02ccbce8 100644
--- a/src/mongo/db/index/index_access_method.cpp
+++ b/src/mongo/db/index/index_access_method.cpp
@@ -60,7 +60,6 @@
 namespace mongo {
 
 using std::pair;
-using std::set;
 
 using IndexVersion = IndexDescriptor::IndexVersion;
 
@@ -118,43 +117,118 @@ SortedDataIndexAccessMethod::SortedDataIndexAccessMethod(const IndexCatalogEntry
     verify(IndexDescriptor::isIndexVersionSupported(_descriptor->version()));
 }
 
-// Find the keys for obj, put them in the tree pointing to loc.
 Status SortedDataIndexAccessMethod::insert(OperationContext* opCtx,
-                                           SharedBufferFragmentBuilder& pooledBufferBuilder,
+                                           SharedBufferFragmentBuilder& pooledBuilder,
                                            const CollectionPtr& coll,
-                                           const BSONObj& obj,
-                                           const RecordId& loc,
+                                           const std::vector<BsonRecord>& bsonRecords,
                                            const InsertDeleteOptions& options,
-                                           KeyHandlerFn&& onDuplicateKey,
                                            int64_t* numInserted) {
-    invariant(options.fromIndexBuilder || !_indexCatalogEntry->isHybridBuilding());
+    for (auto bsonRecord : bsonRecords) {
+        invariant(bsonRecord.id != RecordId());
+
+        if (!bsonRecord.ts.isNull()) {
+            Status status = opCtx->recoveryUnit()->setTimestamp(bsonRecord.ts);
+            if (!status.isOK())
+                return status;
+        }
+
+        auto& executionCtx = StorageExecutionContext::get(opCtx);
+        auto keys = executionCtx.keys();
+        auto multikeyMetadataKeys = executionCtx.multikeyMetadataKeys();
+        auto multikeyPaths = executionCtx.multikeyPaths();
 
+        getKeys(opCtx,
+                coll,
+                pooledBuilder,
+                *bsonRecord.docPtr,
+                options.getKeysMode,
+                GetKeysContext::kAddingKeys,
+                keys.get(),
+                multikeyMetadataKeys.get(),
+                multikeyPaths.get(),
+                bsonRecord.id);
+
+        Status status = _indexKeysOrWriteToSideTable(opCtx,
+                                                     coll,
+                                                     *keys,
+                                                     *multikeyMetadataKeys,
+                                                     *multikeyPaths,
+                                                     *bsonRecord.docPtr,
+                                                     options,
+                                                     numInserted);
+        if (!status.isOK()) {
+            return status;
+        }
+    }
+
+    return Status::OK();
+}
+
+void SortedDataIndexAccessMethod::remove(OperationContext* opCtx,
+                                         SharedBufferFragmentBuilder& pooledBuilder,
+                                         const CollectionPtr& coll,
+                                         const BSONObj& obj,
+                                         const RecordId& loc,
+                                         bool logIfError,
+                                         const InsertDeleteOptions& options,
+                                         int64_t* numDeleted,
+                                         CheckRecordId checkRecordId) {
     auto& executionCtx = StorageExecutionContext::get(opCtx);
 
+    // There's no need to compute the prefixes of the indexed fields that cause the index to be
+    // multikey when removing a document since the index metadata isn't updated when keys are
+    // deleted.
     auto keys = executionCtx.keys();
-    auto multikeyMetadataKeys = executionCtx.multikeyMetadataKeys();
-    auto multikeyPaths = executionCtx.multikeyPaths();
-
     getKeys(opCtx,
             coll,
-            pooledBufferBuilder,
+            pooledBuilder,
             obj,
-            options.getKeysMode,
-            GetKeysContext::kAddingKeys,
+            InsertDeleteOptions::ConstraintEnforcementMode::kRelaxConstraintsUnfiltered,
+            GetKeysContext::kRemovingKeys,
             keys.get(),
-            multikeyMetadataKeys.get(),
-            multikeyPaths.get(),
+            nullptr,
+            nullptr,
             loc);
 
-    return insertKeysAndUpdateMultikeyPaths(opCtx,
+    _unindexKeysOrWriteToSideTable(
+        opCtx, coll->ns(), *keys, obj, logIfError, numDeleted, options, checkRecordId);
+}
+
+Status SortedDataIndexAccessMethod::update(OperationContext* opCtx,
+                                           SharedBufferFragmentBuilder& pooledBufferBuilder,
+                                           const BSONObj& oldDoc,
+                                           const BSONObj& newDoc,
+                                           const RecordId& loc,
+                                           const CollectionPtr& coll,
+                                           const InsertDeleteOptions& options,
+                                           int64_t* numInserted,
+                                           int64_t* numDeleted) {
+
+    UpdateTicket updateTicket;
+    prepareUpdate(opCtx, coll, oldDoc, newDoc, loc, options, &updateTicket);
+
+    auto status = Status::OK();
+    if (_indexCatalogEntry->isHybridBuilding() || !_indexCatalogEntry->isReady(opCtx)) {
+        bool logIfError = false;
+        _unindexKeysOrWriteToSideTable(opCtx,
+                                       coll->ns(),
+                                       updateTicket.removed,
+                                       oldDoc,
+                                       logIfError,
+                                       numDeleted,
+                                       options,
+                                       CheckRecordId::Off);
+        return _indexKeysOrWriteToSideTable(opCtx,
                                             coll,
-                                            *keys,
-                                            *multikeyMetadataKeys,
-                                            *multikeyPaths,
-                                            loc,
+                                            updateTicket.added,
+                                            updateTicket.newMultikeyMetadataKeys,
+                                            updateTicket.newMultikeyPaths,
+                                            newDoc,
                                             options,
-                                            std::move(onDuplicateKey),
                                             numInserted);
+    } else {
+        return doUpdate(opCtx, coll, updateTicket, numInserted, numDeleted);
+    }
 }
 
 Status SortedDataIndexAccessMethod::insertKeysAndUpdateMultikeyPaths(
@@ -163,13 +237,11 @@ Status SortedDataIndexAccessMethod::insertKeysAndUpdateMultikeyPaths(
     const KeyStringSet& keys,
     const KeyStringSet& multikeyMetadataKeys,
     const MultikeyPaths& multikeyPaths,
-    const RecordId& loc,
     const InsertDeleteOptions& options,
     KeyHandlerFn&& onDuplicateKey,
     int64_t* numInserted) {
     // Insert the specified data keys into the index.
-    auto status =
-        insertKeys(opCtx, coll, keys, loc, options, std::move(onDuplicateKey), numInserted);
+    auto status = insertKeys(opCtx, coll, keys, options, std::move(onDuplicateKey), numInserted);
     if (!status.isOK()) {
         return status;
     }
@@ -188,7 +260,6 @@ Status SortedDataIndexAccessMethod::insertKeysAndUpdateMultikeyPaths(
 Status SortedDataIndexAccessMethod::insertKeys(OperationContext* opCtx,
                                                const CollectionPtr& coll,
                                                const KeyStringSet& keys,
-                                               const RecordId& loc,
                                                const InsertDeleteOptions& options,
                                                KeyHandlerFn&& onDuplicateKey,
                                                int64_t* numInserted) {
@@ -250,7 +321,6 @@ Status SortedDataIndexAccessMethod::insertKeys(OperationContext* opCtx,
 
 void SortedDataIndexAccessMethod::removeOneKey(OperationContext* opCtx,
                                                const KeyString::Value& keyString,
-                                               const RecordId& loc,
                                                bool dupsAllowed) {
 
     try {
@@ -259,11 +329,10 @@ void SortedDataIndexAccessMethod::removeOneKey(OperationContext* opCtx,
         NamespaceString ns = _indexCatalogEntry->getNSSFromCatalog(opCtx);
         LOGV2(20683,
               "Assertion failure: _unindex failed on: {namespace} for index: {indexName}. "
-              "{error}  KeyString:{keyString}  dl:{recordId}",
+              "{error}  KeyString:{keyString}",
               "Assertion failure: _unindex failed",
               "error"_attr = redact(e),
               "keyString"_attr = keyString,
-              "recordId"_attr = loc,
               "namespace"_attr = ns,
               "indexName"_attr = _descriptor->indexName());
         printStackTrace();
@@ -275,19 +344,13 @@ std::unique_ptr<SortedDataInterface::Cursor> SortedDataIndexAccessMethod::newCur
     return _newInterface->newCursor(opCtx, isForward);
 }
 
-std::unique_ptr<SortedDataInterface::Cursor> SortedDataIndexAccessMethod::newCursor(
-    OperationContext* opCtx) const {
-    return newCursor(opCtx, true);
-}
-
 Status SortedDataIndexAccessMethod::removeKeys(OperationContext* opCtx,
                                                const KeyStringSet& keys,
-                                               const RecordId& loc,
                                                const InsertDeleteOptions& options,
                                                int64_t* numDeleted) {
 
     for (const auto& key : keys) {
-        removeOneKey(opCtx, key, loc, options.dupsAllowed);
+        removeOneKey(opCtx, key, options.dupsAllowed);
     }
 
     *numDeleted = keys.size();
@@ -316,7 +379,7 @@ RecordId SortedDataIndexAccessMethod::findSingle(OperationContext* opCtx,
                     collection,
                     pooledBuilder,
                     requestedKey,
-                    GetKeysMode::kEnforceConstraints,
+                    InsertDeleteOptions::ConstraintEnforcementMode::kEnforceConstraints,
                     GetKeysContext::kAddingKeys,
                     keys.get(),
                     multikeyMetadataKeys,
@@ -404,19 +467,18 @@ pair<KeyStringSet, KeyStringSet> SortedDataIndexAccessMethod::setDifference(
 
 void SortedDataIndexAccessMethod::prepareUpdate(OperationContext* opCtx,
                                                 const CollectionPtr& collection,
-                                                const IndexCatalogEntry* index,
                                                 const BSONObj& from,
                                                 const BSONObj& to,
                                                 const RecordId& record,
                                                 const InsertDeleteOptions& options,
                                                 UpdateTicket* ticket) const {
     SharedBufferFragmentBuilder pooledBuilder(KeyString::HeapBuilder::kHeapAllocatorDefaultBytes);
-    const MatchExpression* indexFilter = index->getFilterExpression();
+    const MatchExpression* indexFilter = _indexCatalogEntry->getFilterExpression();
     if (!indexFilter || indexFilter->matchesBSON(from)) {
         // Override key constraints when generating keys for removal. This only applies to keys
         // that do not apply to a partial filter expression.
-        const auto getKeysMode = index->isHybridBuilding()
-            ? IndexAccessMethod::GetKeysMode::kRelaxConstraintsUnfiltered
+        const auto getKeysMode = _indexCatalogEntry->isHybridBuilding()
+            ? InsertDeleteOptions::ConstraintEnforcementMode::kRelaxConstraintsUnfiltered
             : options.getKeysMode;
 
         // There's no need to compute the prefixes of the indexed fields that possibly caused the
@@ -455,11 +517,11 @@ void SortedDataIndexAccessMethod::prepareUpdate(OperationContext* opCtx,
     ticket->_isValid = true;
 }
 
-Status SortedDataIndexAccessMethod::update(OperationContext* opCtx,
-                                           const CollectionPtr& coll,
-                                           const UpdateTicket& ticket,
-                                           int64_t* numInserted,
-                                           int64_t* numDeleted) {
+Status SortedDataIndexAccessMethod::doUpdate(OperationContext* opCtx,
+                                             const CollectionPtr& coll,
+                                             const UpdateTicket& ticket,
+                                             int64_t* numInserted,
+                                             int64_t* numDeleted) {
     invariant(!_indexCatalogEntry->isHybridBuilding());
     invariant(ticket.newKeys.size() ==
               ticket.oldKeys.size() + ticket.added.size() - ticket.removed.size());
@@ -504,13 +566,19 @@ Status SortedDataIndexAccessMethod::compact(OperationContext* opCtx) {
     return this->_newInterface->compact(opCtx);
 }
 
-class SortedDataIndexAccessMethod::BulkBuilderImpl : public IndexAccessMethod::BulkBuilder {
+Ident* SortedDataIndexAccessMethod::getIdentPtr() const {
+    return this->_newInterface.get();
+}
+
+class SortedDataIndexAccessMethod::BulkBuilderImpl final : public IndexAccessMethod::BulkBuilder {
 public:
-    BulkBuilderImpl(const IndexCatalogEntry* indexCatalogEntry,
+    using Sorter = mongo::Sorter<KeyString::Value, mongo::NullValue>;
+
+    BulkBuilderImpl(SortedDataIndexAccessMethod* iam,
                     size_t maxMemoryUsageBytes,
                     StringData dbName);
 
-    BulkBuilderImpl(const IndexCatalogEntry* index,
+    BulkBuilderImpl(SortedDataIndexAccessMethod* iam,
                     size_t maxMemoryUsageBytes,
                     const IndexStateInfo& stateInfo,
                     StringData dbName);
@@ -524,21 +592,23 @@ public:
                   const std::function<void()>& saveCursorBeforeWrite,
                   const std::function<void()>& restoreCursorAfterWrite) final;
 
+    Status commit(OperationContext* opCtx,
+                  const CollectionPtr& collection,
+                  bool dupsAllowed,
+                  int32_t yieldIterations,
+                  const KeyHandlerFn& onDuplicateKeyInserted,
+                  const RecordIdHandlerFn& onDuplicateRecord) final;
+
     const MultikeyPaths& getMultikeyPaths() const final;
 
     bool isMultikey() const final;
 
-    /**
-     * Inserts all multikey metadata keys cached during the BulkBuilder's lifetime into the
-     * underlying Sorter, finalizes it, and returns an iterator over the sorted dataset.
-     */
-    Sorter::Iterator* done() final;
-
-    int64_t getKeysInserted() const final;
-
-    Sorter::PersistedState persistDataForShutdown() final;
+    void persistDataForShutdown(BSONObjBuilder& builder) final;
 
 private:
+    void _yield(OperationContext* opCtx,
+                const Yieldable* yieldable,
+                const NamespaceString& ns) const;
     void _insertMultikeyMetadataKeysIntoSorter();
 
     Sorter* _makeSorter(
@@ -549,7 +619,7 @@ private:
 
     Sorter::Settings _makeSorterSettings() const;
 
-    const IndexCatalogEntry* _indexCatalogEntry;
+    SortedDataIndexAccessMethod* _iam;
     std::unique_ptr<Sorter> _sorter;
     int64_t _keysInserted = 0;
 
@@ -571,21 +641,20 @@ std::unique_ptr<IndexAccessMethod::BulkBuilder> SortedDataIndexAccessMethod::ini
     const boost::optional<IndexStateInfo>& stateInfo,
     StringData dbName) {
     return stateInfo
-        ? std::make_unique<BulkBuilderImpl>(
-              _indexCatalogEntry, maxMemoryUsageBytes, *stateInfo, dbName)
-        : std::make_unique<BulkBuilderImpl>(_indexCatalogEntry, maxMemoryUsageBytes, dbName);
+        ? std::make_unique<BulkBuilderImpl>(this, maxMemoryUsageBytes, *stateInfo, dbName)
+        : std::make_unique<BulkBuilderImpl>(this, maxMemoryUsageBytes, dbName);
 }
 
-SortedDataIndexAccessMethod::BulkBuilderImpl::BulkBuilderImpl(const IndexCatalogEntry* index,
+SortedDataIndexAccessMethod::BulkBuilderImpl::BulkBuilderImpl(SortedDataIndexAccessMethod* iam,
                                                               size_t maxMemoryUsageBytes,
                                                               StringData dbName)
-    : _indexCatalogEntry(index), _sorter(_makeSorter(maxMemoryUsageBytes, dbName)) {}
+    : _iam(iam), _sorter(_makeSorter(maxMemoryUsageBytes, dbName)) {}
 
-SortedDataIndexAccessMethod::BulkBuilderImpl::BulkBuilderImpl(const IndexCatalogEntry* index,
+SortedDataIndexAccessMethod::BulkBuilderImpl::BulkBuilderImpl(SortedDataIndexAccessMethod* iam,
                                                               size_t maxMemoryUsageBytes,
                                                               const IndexStateInfo& stateInfo,
                                                               StringData dbName)
-    : _indexCatalogEntry(index),
+    : _iam(iam),
       _sorter(
           _makeSorter(maxMemoryUsageBytes, dbName, stateInfo.getFileName(), stateInfo.getRanges())),
       _keysInserted(stateInfo.getNumKeys().value_or(0)),
@@ -607,37 +676,38 @@ Status SortedDataIndexAccessMethod::BulkBuilderImpl::insert(
     auto multikeyPaths = executionCtx.multikeyPaths();
 
     try {
-        _indexCatalogEntry->accessMethod()->getKeys(
-            opCtx,
-            collection,
-            pooledBuilder,
-            obj,
-            options.getKeysMode,
-            GetKeysContext::kAddingKeys,
-            keys.get(),
-            &_multikeyMetadataKeys,
-            multikeyPaths.get(),
-            loc,
-            [&](Status status, const BSONObj&, boost::optional<RecordId>) {
-                // If a key generation error was suppressed, record the document as "skipped" so the
-                // index builder can retry at a point when data is consistent.
-                auto interceptor = _indexCatalogEntry->indexBuildInterceptor();
-                if (interceptor && interceptor->getSkippedRecordTracker()) {
-                    LOGV2_DEBUG(20684,
-                                1,
-                                "Recording suppressed key generation error to retry later: "
-                                "{error} on {loc}: {obj}",
-                                "error"_attr = status,
-                                "loc"_attr = loc,
-                                "obj"_attr = redact(obj));
-
-                    // Save and restore the cursor around the write in case it throws a WCE
-                    // internally and causes the cursor to be unpositioned.
-                    saveCursorBeforeWrite();
-                    interceptor->getSkippedRecordTracker()->record(opCtx, loc);
-                    restoreCursorAfterWrite();
-                }
-            });
+        _iam->getKeys(opCtx,
+                      collection,
+                      pooledBuilder,
+                      obj,
+                      options.getKeysMode,
+                      GetKeysContext::kAddingKeys,
+                      keys.get(),
+                      &_multikeyMetadataKeys,
+                      multikeyPaths.get(),
+                      loc,
+                      [&](Status status, const BSONObj&, boost::optional<RecordId>) {
+                          // If a key generation error was suppressed, record the document as
+                          // "skipped" so the index builder can retry at a point when data is
+                          // consistent.
+                          auto interceptor = _iam->_indexCatalogEntry->indexBuildInterceptor();
+                          if (interceptor && interceptor->getSkippedRecordTracker()) {
+                              LOGV2_DEBUG(
+                                  20684,
+                                  1,
+                                  "Recording suppressed key generation error to retry later: "
+                                  "{error} on {loc}: {obj}",
+                                  "error"_attr = status,
+                                  "loc"_attr = loc,
+                                  "obj"_attr = redact(obj));
+
+                              // Save and restore the cursor around the write in case it throws a
+                              // WCE internally and causes the cursor to be unpositioned.
+                              saveCursorBeforeWrite();
+                              interceptor->getSkippedRecordTracker()->record(opCtx, loc);
+                              restoreCursorAfterWrite();
+                          }
+                      });
     } catch (...) {
         return exceptionToStatus();
     }
@@ -661,8 +731,7 @@ Status SortedDataIndexAccessMethod::BulkBuilderImpl::insert(
     }
 
     _isMultiKey = _isMultiKey ||
-        _indexCatalogEntry->accessMethod()->shouldMarkIndexAsMultikey(
-            keys->size(), _multikeyMetadataKeys, *multikeyPaths);
+        _iam->shouldMarkIndexAsMultikey(keys->size(), _multikeyMetadataKeys, *multikeyPaths);
 
     return Status::OK();
 }
@@ -675,20 +744,20 @@ bool SortedDataIndexAccessMethod::BulkBuilderImpl::isMultikey() const {
     return _isMultiKey;
 }
 
-IndexAccessMethod::BulkBuilder::Sorter::Iterator*
-SortedDataIndexAccessMethod::BulkBuilderImpl::done() {
+void SortedDataIndexAccessMethod::BulkBuilderImpl::persistDataForShutdown(BSONObjBuilder& builder) {
     _insertMultikeyMetadataKeysIntoSorter();
-    return _sorter->done();
-}
+    auto state = _sorter->persistDataForShutdown();
 
-int64_t SortedDataIndexAccessMethod::BulkBuilderImpl::getKeysInserted() const {
-    return _keysInserted;
-}
+    builder.append("fileName", state.fileName);
+    builder.append("numKeys", _keysInserted);
 
-SortedDataIndexAccessMethod::BulkBuilder::Sorter::PersistedState
-SortedDataIndexAccessMethod::BulkBuilderImpl::persistDataForShutdown() {
-    _insertMultikeyMetadataKeysIntoSorter();
-    return _sorter->persistDataForShutdown();
+    BSONArrayBuilder ranges(builder.subarrayStart("ranges"));
+    for (const auto& rangeInfo : state.ranges) {
+        BSONObjBuilder range(ranges.subobjStart());
+        range.append("startOffset", rangeInfo.getStartOffset());
+        range.append("endOffset", rangeInfo.getEndOffset());
+        range.append("checksum", rangeInfo.getChecksum());
+    }
 }
 
 void SortedDataIndexAccessMethod::BulkBuilderImpl::_insertMultikeyMetadataKeysIntoSorter() {
@@ -706,7 +775,7 @@ SortedDataIndexAccessMethod::BulkBuilderImpl::Sorter::Settings
 SortedDataIndexAccessMethod::BulkBuilderImpl::_makeSorterSettings() const {
     return std::pair<KeyString::Value::SorterDeserializeSettings,
                      mongo::NullValue::SorterDeserializeSettings>(
-        {_indexCatalogEntry->accessMethod()->getSortedDataInterface()->getKeyStringVersion()}, {});
+        {_iam->getSortedDataInterface()->getKeyStringVersion()}, {});
 }
 
 SortedDataIndexAccessMethod::BulkBuilderImpl::Sorter*
@@ -725,9 +794,9 @@ SortedDataIndexAccessMethod::BulkBuilderImpl::_makeSorter(
                                    _makeSorterSettings());
 }
 
-void SortedDataIndexAccessMethod::_yieldBulkLoad(OperationContext* opCtx,
-                                                 const Yieldable* yieldable,
-                                                 const NamespaceString& ns) const {
+void SortedDataIndexAccessMethod::BulkBuilderImpl::_yield(OperationContext* opCtx,
+                                                          const Yieldable* yieldable,
+                                                          const NamespaceString& ns) const {
     // Releasing locks means a new snapshot should be acquired when restored.
     opCtx->recoveryUnit()->abandonSnapshot();
     yieldable->yield();
@@ -757,35 +826,38 @@ void SortedDataIndexAccessMethod::_yieldBulkLoad(OperationContext* opCtx,
     yieldable->restore();
 }
 
-Status SortedDataIndexAccessMethod::commitBulk(OperationContext* opCtx,
-                                               const CollectionPtr& collection,
-                                               BulkBuilder* bulk,
-                                               bool dupsAllowed,
-                                               int32_t yieldIterations,
-                                               const KeyHandlerFn& onDuplicateKeyInserted,
-                                               const RecordIdHandlerFn& onDuplicateRecord) {
+Status SortedDataIndexAccessMethod::BulkBuilderImpl::commit(
+    OperationContext* opCtx,
+    const CollectionPtr& collection,
+    bool dupsAllowed,
+    int32_t yieldIterations,
+    const KeyHandlerFn& onDuplicateKeyInserted,
+    const RecordIdHandlerFn& onDuplicateRecord) {
+
     Timer timer;
 
-    auto ns = _indexCatalogEntry->getNSSFromCatalog(opCtx);
+    const auto descriptor = _iam->_descriptor;
+    auto ns = _iam->_indexCatalogEntry->getNSSFromCatalog(opCtx);
 
-    std::unique_ptr<BulkBuilder::Sorter::Iterator> it(bulk->done());
+    _insertMultikeyMetadataKeysIntoSorter();
+    std::unique_ptr<Sorter::Iterator> it(_sorter->done());
 
     static constexpr char message[] = "Index Build: inserting keys from external sorter into index";
     ProgressMeterHolder pm;
     {
         stdx::unique_lock<Client> lk(*opCtx->getClient());
-        pm.set(CurOp::get(opCtx)->setProgress_inlock(
-            message, bulk->getKeysInserted(), 3 /* secondsBetween */));
+        pm.set(
+            CurOp::get(opCtx)->setProgress_inlock(message, _keysInserted, 3 /* secondsBetween */));
     }
 
-    auto builder = _newInterface->makeBulkBuilder(opCtx, dupsAllowed);
+    auto builder = _iam->getSortedDataInterface()->makeBulkBuilder(opCtx, dupsAllowed);
 
     KeyString::Value previousKey;
 
     for (int64_t i = 0; it->more(); i++) {
         opCtx->checkForInterrupt();
 
-        auto failPointHang = [opCtx, i, &indexName = _descriptor->indexName()](FailPoint* fp) {
+        auto failPointHang = [opCtx, i, &indexName = descriptor->indexName()](FailPoint* fp) {
             fp->executeIf(
                 [fp, opCtx, i, &indexName](const BSONObj& data) {
                     LOGV2(4924400,
@@ -809,13 +881,13 @@ Status SortedDataIndexAccessMethod::commitBulk(OperationContext* opCtx,
         failPointHang(&hangIndexBuildDuringBulkLoadPhaseSecond);
 
         // Get the next datum and add it to the builder.
-        BulkBuilder::Sorter::Data data = it->next();
+        Sorter::Data data = it->next();
 
         // Assert that keys are retrieved from the sorter in non-decreasing order, but only in debug
         // builds since this check can be expensive.
         int cmpData;
-        if (_descriptor->unique()) {
-            cmpData = (_newInterface->rsKeyFormat() == KeyFormat::Long)
+        if (descriptor->unique()) {
+            cmpData = (_iam->getSortedDataInterface()->rsKeyFormat() == KeyFormat::Long)
                 ? data.first.compareWithoutRecordIdLong(previousKey)
                 : data.first.compareWithoutRecordIdStr(previousKey);
         }
@@ -826,13 +898,13 @@ Status SortedDataIndexAccessMethod::commitBulk(OperationContext* opCtx,
                 "Expected the next key to be greater than or equal to the previous key",
                 "nextKey"_attr = data.first.toString(),
                 "previousKey"_attr = previousKey.toString(),
-                "index"_attr = _descriptor->indexName());
+                "index"_attr = descriptor->indexName());
         }
 
         // Before attempting to insert, perform a duplicate key check.
-        bool isDup = (_descriptor->unique()) ? (cmpData == 0) : false;
+        bool isDup = (descriptor->unique()) ? (cmpData == 0) : false;
         if (isDup && !dupsAllowed) {
-            Status status = _handleDuplicateKey(opCtx, data.first, onDuplicateRecord);
+            Status status = _iam->_handleDuplicateKey(opCtx, data.first, onDuplicateRecord);
             if (!status.isOK()) {
                 return status;
             }
@@ -866,7 +938,7 @@ Status SortedDataIndexAccessMethod::commitBulk(OperationContext* opCtx,
 
         // Starts yielding locks after the first non-zero 'yieldIterations' inserts.
         if (yieldIterations && (i + 1) % yieldIterations == 0) {
-            _yieldBulkLoad(opCtx, &collection, ns);
+            _yield(opCtx, &collection, ns);
         }
 
         // If we're here either it's a dup and we're cool with it or the addKey went just fine.
@@ -880,24 +952,17 @@ Status SortedDataIndexAccessMethod::commitBulk(OperationContext* opCtx,
           "{timer_seconds} seconds",
           "Index build: inserted keys from external sorter into index",
           logAttrs(ns),
-          "index"_attr = _descriptor->indexName(),
-          "keysInserted"_attr = bulk->getKeysInserted(),
+          "index"_attr = descriptor->indexName(),
+          "keysInserted"_attr = _keysInserted,
           "duration"_attr = Milliseconds(Seconds(timer.seconds())));
     return Status::OK();
 }
 
-void SortedDataIndexAccessMethod::setIndexIsMultikey(OperationContext* opCtx,
-                                                     const CollectionPtr& collection,
-                                                     KeyStringSet multikeyMetadataKeys,
-                                                     MultikeyPaths paths) {
-    _indexCatalogEntry->setMultikey(opCtx, collection, multikeyMetadataKeys, paths);
-}
-
 void SortedDataIndexAccessMethod::getKeys(OperationContext* opCtx,
                                           const CollectionPtr& collection,
                                           SharedBufferFragmentBuilder& pooledBufferBuilder,
                                           const BSONObj& obj,
-                                          GetKeysMode mode,
+                                          InsertDeleteOptions::ConstraintEnforcementMode mode,
                                           GetKeysContext context,
                                           KeyStringSet* keys,
                                           KeyStringSet* multikeyMetadataKeys,
@@ -924,7 +989,7 @@ void SortedDataIndexAccessMethod::getKeys(OperationContext* opCtx,
                   id);
     } catch (const AssertionException& ex) {
         // Suppress all indexing errors when mode is kRelaxConstraints.
-        if (mode == GetKeysMode::kEnforceConstraints) {
+        if (mode == InsertDeleteOptions::ConstraintEnforcementMode::kEnforceConstraints) {
             throw;
         }
 
@@ -940,8 +1005,8 @@ void SortedDataIndexAccessMethod::getKeys(OperationContext* opCtx,
         // If the document applies to the filter (which means that it should have never been
         // indexed), do not suppress the error.
         const MatchExpression* filter = _indexCatalogEntry->getFilterExpression();
-        if (mode == GetKeysMode::kRelaxConstraintsUnfiltered && filter &&
-            filter->matchesBSON(obj)) {
+        if (mode == InsertDeleteOptions::ConstraintEnforcementMode::kRelaxConstraintsUnfiltered &&
+            filter && filter->matchesBSON(obj)) {
             throw;
         }
 
@@ -969,10 +1034,6 @@ void SortedDataIndexAccessMethod::validateDocument(const CollectionPtr& collecti
                                                    const BSONObj& obj,
                                                    const BSONObj& keyPattern) const {}
 
-SortedDataInterface* SortedDataIndexAccessMethod::getSortedDataInterface() const {
-    return _newInterface.get();
-}
-
 /**
  * Generates a new file name on each call using a static, atomic and monotonically increasing
  * number. Each name is suffixed with a random number generated at startup, to prevent name
@@ -1008,6 +1069,118 @@ Status SortedDataIndexAccessMethod::_handleDuplicateKey(
                                   _descriptor->keyPattern(),
                                   _descriptor->collation());
 }
+
+Status SortedDataIndexAccessMethod::_indexKeysOrWriteToSideTable(
+    OperationContext* opCtx,
+    const CollectionPtr& coll,
+    const KeyStringSet& keys,
+    const KeyStringSet& multikeyMetadataKeys,
+    const MultikeyPaths& multikeyPaths,
+    const BSONObj& obj,
+    const InsertDeleteOptions& options,
+    int64_t* keysInsertedOut) {
+    Status status = Status::OK();
+    if (_indexCatalogEntry->isHybridBuilding()) {
+        // The side table interface accepts only records that meet the criteria for this partial
+        // index.
+        // See SERVER-28975 and SERVER-39705 for details.
+        if (auto filter = _indexCatalogEntry->getFilterExpression()) {
+            if (!filter->matchesBSON(obj)) {
+                return Status::OK();
+            }
+        }
+
+        int64_t inserted = 0;
+        status = _indexCatalogEntry->indexBuildInterceptor()->sideWrite(
+            opCtx,
+            keys,
+            multikeyMetadataKeys,
+            multikeyPaths,
+            IndexBuildInterceptor::Op::kInsert,
+            &inserted);
+        if (keysInsertedOut) {
+            *keysInsertedOut += inserted;
+        }
+    } else {
+        int64_t numInserted = 0;
+        status = insertKeysAndUpdateMultikeyPaths(
+            opCtx,
+            coll,
+            keys,
+            {multikeyMetadataKeys.begin(), multikeyMetadataKeys.end()},
+            multikeyPaths,
+            options,
+            nullptr,
+            &numInserted);
+        if (keysInsertedOut) {
+            *keysInsertedOut += numInserted;
+        }
+    }
+
+    return status;
+}
+
+void SortedDataIndexAccessMethod::_unindexKeysOrWriteToSideTable(
+    OperationContext* opCtx,
+    const NamespaceString& ns,
+    const KeyStringSet& keys,
+    const BSONObj& obj,
+    bool logIfError,
+    int64_t* const keysDeletedOut,
+    InsertDeleteOptions options,  // copy!
+    CheckRecordId checkRecordId) {
+
+    options.logIfError = logIfError;
+
+    if (_indexCatalogEntry->isHybridBuilding()) {
+        // The side table interface accepts only records that meet the criteria for this partial
+        // index.
+        // See SERVER-28975 and SERVER-39705 for details.
+        if (auto filter = _indexCatalogEntry->getFilterExpression()) {
+            if (!filter->matchesBSON(obj)) {
+                return;
+            }
+        }
+
+        int64_t removed = 0;
+        fassert(31155,
+                _indexCatalogEntry->indexBuildInterceptor()->sideWrite(
+                    opCtx, keys, {}, {}, IndexBuildInterceptor::Op::kDelete, &removed));
+        if (keysDeletedOut) {
+            *keysDeletedOut += removed;
+        }
+
+        return;
+    }
+
+    // On WiredTiger, we do blind unindexing of records for efficiency.  However, when duplicates
+    // are allowed in unique indexes, WiredTiger does not do blind unindexing, and instead confirms
+    // that the recordid matches the element we are removing.
+    //
+    // We need to disable blind-deletes if 'checkRecordId' is explicitly set 'On', or for
+    // in-progress indexes, in order to force recordid-matching for unindex operations, since
+    // initial sync can build an index over a collection with duplicates. See SERVER-17487 for more
+    // details.
+    options.dupsAllowed = options.dupsAllowed || !_indexCatalogEntry->isReady(opCtx) ||
+        (checkRecordId == CheckRecordId::On);
+
+    int64_t removed = 0;
+    Status status = removeKeys(opCtx, keys, options, &removed);
+
+    if (!status.isOK()) {
+        LOGV2(20362,
+              "Couldn't unindex record {obj} from collection {namespace}: {error}",
+              "Couldn't unindex record",
+              "record"_attr = redact(obj),
+              "namespace"_attr = ns,
+              "error"_attr = redact(status));
+    }
+
+    if (keysDeletedOut) {
+        *keysDeletedOut += removed;
+    }
+}
+
 }  // namespace mongo
 
 #include "mongo/db/sorter/sorter.cpp"
diff --git a/src/mongo/db/index/index_access_method.h b/src/mongo/db/index/index_access_method.h
index f9988b7d191..8b911c01603 100644
--- a/src/mongo/db/index/index_access_method.h
+++ b/src/mongo/db/index/index_access_method.h
@@ -51,6 +51,7 @@ class BSONObjBuilder;
 class MatchExpression;
 struct UpdateTicket;
 struct InsertDeleteOptions;
+class SortedDataIndexAccessMethod;
 
 /**
  * An IndexAccessMethod is the interface through which all the mutation, lookup, and
@@ -73,107 +74,50 @@ public:
     IndexAccessMethod() = default;
     virtual ~IndexAccessMethod() = default;
 
+    /**
+     * Equivalent to (but shorter and faster than): dynamic_cast<SortedDataIndexAccessMethod*>(this)
+     */
+    virtual SortedDataIndexAccessMethod* asSortedData() {
+        return nullptr;
+    }
+    virtual const SortedDataIndexAccessMethod* asSortedData() const {
+        return nullptr;
+    }
+
     //
     // Lookup, traversal, and mutation support
     //
 
     /**
-     * Internally generate the keys {k1, ..., kn} for 'obj'.  For each key k, insert (k ->
-     * 'loc') into the index.  'obj' is the object at the location 'loc'.
-     * If 'result' is not null, 'numInserted' will be set to the number of keys added to the index
-     * for the document and the number of duplicate keys will be appended to 'dupsInserted' if this
-     * is a unique index and duplicates are allowed.
-     *
-     * If there is more than one key for 'obj', either all keys will be inserted or none will.
-     *
-     * The behavior of the insertion can be specified through 'options'.
+     * Informs the index of inserts, updates, and deletes of records from the indexed collection.
      */
     virtual Status insert(OperationContext* opCtx,
                           SharedBufferFragmentBuilder& pooledBufferBuilder,
                           const CollectionPtr& coll,
-                          const BSONObj& obj,
-                          const RecordId& loc,
+                          const std::vector<BsonRecord>& bsonRecords,
                           const InsertDeleteOptions& options,
-                          KeyHandlerFn&& onDuplicateKey,
                           int64_t* numInserted) = 0;
 
-    /**
-     * Inserts the specified keys into the index. and determines whether these keys should cause the
-     * index to become multikey. If so, this method also handles the task of marking the index as
-     * multikey in the catalog, and sets the path-level multikey information if applicable.
-     */
-    virtual Status insertKeysAndUpdateMultikeyPaths(OperationContext* opCtx,
-                                                    const CollectionPtr& coll,
-                                                    const KeyStringSet& keys,
-                                                    const KeyStringSet& multikeyMetadataKeys,
-                                                    const MultikeyPaths& multikeyPaths,
-                                                    const RecordId& loc,
-                                                    const InsertDeleteOptions& options,
-                                                    KeyHandlerFn&& onDuplicateKey,
-                                                    int64_t* numInserted) = 0;
+    virtual void remove(OperationContext* opCtx,
+                        SharedBufferFragmentBuilder& pooledBufferBuilder,
+                        const CollectionPtr& coll,
+                        const BSONObj& obj,
+                        const RecordId& loc,
+                        bool logIfError,
+                        const InsertDeleteOptions& options,
+                        int64_t* numDeleted,
+                        CheckRecordId checkRecordId) = 0;
 
-    /**
-     * Inserts the specified keys into the index. Does not attempt to determine whether the
-     * insertion of these keys should cause the index to become multikey. The 'numInserted' output
-     * parameter, if non-nullptr, will be reset to the number of keys inserted by this function
-     * call, or to zero in the case of either a non-OK return Status or an empty 'keys' argument.
-     */
-    virtual Status insertKeys(OperationContext* opCtx,
-                              const CollectionPtr& coll,
-                              const KeyStringSet& keys,
-                              const RecordId& loc,
-                              const InsertDeleteOptions& options,
-                              KeyHandlerFn&& onDuplicateKey,
-                              int64_t* numInserted) = 0;
-
-    /**
-     * Analogous to insertKeys above, but remove the keys instead of inserting them.
-     * 'numDeleted' will be set to the number of keys removed from the index for the provided keys.
-     */
-    virtual Status removeKeys(OperationContext* opCtx,
-                              const KeyStringSet& keys,
-                              const RecordId& loc,
-                              const InsertDeleteOptions& options,
-                              int64_t* numDeleted) = 0;
-
-    /**
-     * Gets the keys of the documents 'from' and 'to' and prepares them for the update.
-     * Provides a ticket for actually performing the update.
-     */
-    virtual void prepareUpdate(OperationContext* opCtx,
-                               const CollectionPtr& collection,
-                               const IndexCatalogEntry* index,
-                               const BSONObj& from,
-                               const BSONObj& to,
-                               const RecordId& loc,
-                               const InsertDeleteOptions& options,
-                               UpdateTicket* ticket) const = 0;
-
-    /**
-     * Perform a validated update.  The keys for the 'from' object will be removed, and the keys
-     * for the object 'to' will be added.  Returns OK if the update succeeded, failure if it did
-     * not.  If an update does not succeed, the index will be unmodified, and the keys for
-     * 'from' will remain.  Assumes that the index has not changed since prepareUpdate was
-     * called.  If the index was changed, we may return an error, as our ticket may have been
-     * invalidated.
-     *
-     * 'numInserted' will be set to the number of keys inserted into the index for the document.
-     * 'numDeleted' will be set to the number of keys removed from the index for the document.
-     */
     virtual Status update(OperationContext* opCtx,
+                          SharedBufferFragmentBuilder& pooledBufferBuilder,
+                          const BSONObj& oldDoc,
+                          const BSONObj& newDoc,
+                          const RecordId& loc,
                           const CollectionPtr& coll,
-                          const UpdateTicket& ticket,
+                          const InsertDeleteOptions& options,
                           int64_t* numInserted,
                           int64_t* numDeleted) = 0;
 
-    /**
-     * Returns an unpositioned cursor over 'this' index.
-     */
-    virtual std::unique_ptr<SortedDataInterface::Cursor> newCursor(OperationContext* opCtx,
-                                                                   bool isForward) const = 0;
-    virtual std::unique_ptr<SortedDataInterface::Cursor> newCursor(
-        OperationContext* opCtx) const = 0;
-
     // ------ index level operations ------
 
 
@@ -214,23 +158,13 @@ public:
      */
     virtual long long getFreeStorageBytes(OperationContext* opCtx) const = 0;
 
-    virtual RecordId findSingle(OperationContext* opCtx,
-                                const CollectionPtr& collection,
-                                const BSONObj& key) const = 0;
-
     /**
      * Attempt compaction to regain disk space if the indexed record store supports
      * compaction-in-place.
      */
     virtual Status compact(OperationContext* opCtx) = 0;
 
-    /**
-     * Sets this index as multikey with the provided paths.
-     */
-    virtual void setIndexIsMultikey(OperationContext* opCtx,
-                                    const CollectionPtr& collection,
-                                    KeyStringSet multikeyMetadataKeys,
-                                    MultikeyPaths paths) = 0;
+    virtual Ident* getIdentPtr() const = 0;
 
     //
     // Bulk operations support
@@ -238,8 +172,6 @@ public:
 
     class BulkBuilder {
     public:
-        using Sorter = mongo::Sorter<KeyString::Value, mongo::NullValue>;
-
         virtual ~BulkBuilder() = default;
 
         /**
@@ -262,31 +194,38 @@ public:
                               const std::function<void()>& saveCursorBeforeWrite,
                               const std::function<void()>& restoreCursorAfterWrite) = 0;
 
-        virtual const MultikeyPaths& getMultikeyPaths() const = 0;
-
-        virtual bool isMultikey() const = 0;
-
         /**
-         * Inserts all multikey metadata keys cached during the BulkBuilder's lifetime into the
-         * underlying Sorter, finalizes it, and returns an iterator over the sorted dataset.
+         * Call this when you are ready to finish your bulk work.
+         * @param dupsAllowed - If false and 'dupRecords' is not null, append with the RecordIds of
+         *                      the uninserted duplicates.
+         * @param yieldIterations - The number of iterations run before each yielding. Will not
+         * yield if zero.
+         * @param onDuplicateKeyInserted - Will be called for each duplicate key inserted into the
+         * index.
+         * @param onDuplicateRecord - If not nullptr, will be called for each RecordId of uninserted
+         * duplicate keys.
          */
-        virtual Sorter::Iterator* done() = 0;
+        virtual Status commit(OperationContext* opCtx,
+                              const CollectionPtr& collection,
+                              bool dupsAllowed,
+                              int32_t yieldIterations,
+                              const KeyHandlerFn& onDuplicateKeyInserted,
+                              const RecordIdHandlerFn& onDuplicateRecord) = 0;
 
-        /**
-         * Returns number of keys inserted using this BulkBuilder.
-         */
-        virtual int64_t getKeysInserted() const = 0;
+        virtual const MultikeyPaths& getMultikeyPaths() const = 0;
+
+        virtual bool isMultikey() const = 0;
 
         /**
-         * Persists on disk the keys that have been inserted using this BulkBuilder. Returns the
-         * state of the underlying Sorter.
+         * Persists on disk the keys that have been inserted using this BulkBuilder.
+         * Appends necessary information to resume building to passed-in builder.
          */
-        virtual Sorter::PersistedState persistDataForShutdown() = 0;
+        virtual void persistDataForShutdown(BSONObjBuilder& builder) = 0;
     };
 
     /**
      * Starts a bulk operation.
-     * You work on the returned BulkBuilder and then call commitBulk.
+     * You work on the returned BulkBuilder and then call bulk->commit().
      * This can return NULL, meaning bulk mode is not available.
      *
      * It is only legal to initiate bulk when the index is new and empty, or when resuming an index
@@ -302,94 +241,6 @@ public:
         const boost::optional<IndexStateInfo>& stateInfo,
         StringData dbName) = 0;
 
-    /**
-     * Call this when you are ready to finish your bulk work.
-     * Pass in the BulkBuilder returned from initiateBulk.
-     * @param bulk - Something created from initiateBulk
-     * @param dupsAllowed - If false and 'dupRecords' is not null, append with the RecordIds of
-     *                      the uninserted duplicates.
-     * @param yieldIterations - The number of iterations run before each yielding. Will not yield if
-     * zero.
-     * @param onDuplicateKeyInserted - Will be called for each duplicate key inserted into the
-     * index.
-     * @param onDuplicateRecord - If not nullptr, will be called for each RecordId of uninserted
-     * duplicate keys.
-     */
-    virtual Status commitBulk(OperationContext* opCtx,
-                              const CollectionPtr& collection,
-                              BulkBuilder* bulk,
-                              bool dupsAllowed,
-                              int32_t yieldIterations,
-                              const KeyHandlerFn& onDuplicateKeyInserted,
-                              const RecordIdHandlerFn& onDuplicateRecord) = 0;
-
-    /**
-     * Specifies whether getKeys should relax the index constraints or not, in order of most
-     * permissive to least permissive.
-     */
-    enum class GetKeysMode {
-        // Relax all constraints.
-        kRelaxConstraints,
-        // Relax all constraints on documents that don't apply to a partial index.
-        kRelaxConstraintsUnfiltered,
-        // Enforce all constraints.
-        kEnforceConstraints
-    };
-
-    /**
-     * Specifies whether getKeys is being used in the context of creating new keys, deleting
-     * or validating existing keys.
-     */
-    enum class GetKeysContext { kRemovingKeys, kAddingKeys, kValidatingKeys };
-
-    /**
-     * Fills 'keys' with the keys that should be generated for 'obj' on this index.
-     * Based on 'mode', it will honor or ignore index constraints, e.g. duplicated key, key too
-     * long, and geo index parsing errors. The ignoring of constraints is for replication due to
-     * idempotency reasons. In those cases, the generated 'keys' will be empty.
-     *
-     * If the 'multikeyPaths' pointer is non-null, then it must point to an empty vector. If this
-     * index type supports tracking path-level multikey information, then this function resizes
-     * 'multikeyPaths' to have the same number of elements as the index key pattern and fills each
-     * element with the prefixes of the indexed field that would cause this index to be multikey as
-     * a result of inserting 'keys'.
-     *
-     * If the 'multikeyMetadataKeys' pointer is non-null, then the function will populate the
-     * BSONObjSet with any multikey metadata keys generated while processing the document. These
-     * keys are not associated with the document itself, but instead represent multi-key path
-     * information that must be stored in a reserved keyspace within the index.
-     *
-     * If any key generation errors are encountered and suppressed due to the provided GetKeysMode,
-     * 'onSuppressedErrorFn' is called.
-     */
-    using OnSuppressedErrorFn =
-        std::function<void(Status status, const BSONObj& obj, boost::optional<RecordId> loc)>;
-    virtual void getKeys(OperationContext* opCtx,
-                         const CollectionPtr& collection,
-                         SharedBufferFragmentBuilder& pooledBufferBuilder,
-                         const BSONObj& obj,
-                         GetKeysMode mode,
-                         GetKeysContext context,
-                         KeyStringSet* keys,
-                         KeyStringSet* multikeyMetadataKeys,
-                         MultikeyPaths* multikeyPaths,
-                         boost::optional<RecordId> id,
-                         OnSuppressedErrorFn&& onSuppressedError = nullptr) const = 0;
-
-    /**
-     * Given the set of keys, multikeyMetadataKeys and multikeyPaths generated by a particular
-     * document, return 'true' if the index should be marked as multikey and 'false' otherwise.
-     */
-    virtual bool shouldMarkIndexAsMultikey(size_t numberOfKeys,
-                                           const KeyStringSet& multikeyMetadataKeys,
-                                           const MultikeyPaths& multikeyPaths) const = 0;
-
-    /**
-     * Provides direct access to the SortedDataInterface. This should not be used to insert
-     * documents into an index, except for testing purposes.
-     */
-    virtual SortedDataInterface* getSortedDataInterface() const = 0;
-
     void setEnforceDuplicateConstraints(bool enforceDuplicateConstraints) {
         _enforceDuplicateConstraints.swap(enforceDuplicateConstraints);
     }
@@ -463,9 +314,21 @@ struct InsertDeleteOptions {
     // index builder should set this to 'true'.
     bool fromIndexBuilder = false;
 
+    /**
+     * Specifies whether getKeys should relax the index constraints or not, in order of most
+     * permissive to least permissive.
+     */
+    enum class ConstraintEnforcementMode {
+        // Relax all constraints.
+        kRelaxConstraints,
+        // Relax all constraints on documents that don't apply to a partial index.
+        kRelaxConstraintsUnfiltered,
+        // Enforce all constraints.
+        kEnforceConstraints
+    };
+
     // Should we relax the index constraints?
-    IndexAccessMethod::GetKeysMode getKeysMode =
-        IndexAccessMethod::GetKeysMode::kEnforceConstraints;
+    ConstraintEnforcementMode getKeysMode = ConstraintEnforcementMode::kEnforceConstraints;
 };
 
 /**
@@ -481,6 +344,10 @@ class SortedDataIndexAccessMethod : public IndexAccessMethod {
     SortedDataIndexAccessMethod& operator=(const SortedDataIndexAccessMethod&) = delete;
 
 public:
+    //
+    // SortedData-specific functions
+    //
+
     /**
      * Splits the sets 'left' and 'right' into two sets, the first containing the elements that
      * only appeared in 'left', and the second containing only elements that appeared in 'right'.
@@ -495,58 +362,177 @@ public:
     SortedDataIndexAccessMethod(const IndexCatalogEntry* btreeState,
                                 std::unique_ptr<SortedDataInterface> btree);
 
-    Status insert(OperationContext* opCtx,
-                  SharedBufferFragmentBuilder& pooledBufferBuilder,
-                  const CollectionPtr& coll,
-                  const BSONObj& obj,
-                  const RecordId& loc,
-                  const InsertDeleteOptions& options,
-                  KeyHandlerFn&& onDuplicateKey,
-                  int64_t* numInserted) final;
+    /**
+     * Specifies whether getKeys is being used in the context of creating new keys, deleting
+     * or validating existing keys.
+     */
+    enum class GetKeysContext { kRemovingKeys, kAddingKeys, kValidatingKeys };
+
+    /**
+     * Fills 'keys' with the keys that should be generated for 'obj' on this index.
+     * Based on 'mode', it will honor or ignore index constraints, e.g. duplicated key, key too
+     * long, and geo index parsing errors. The ignoring of constraints is for replication due to
+     * idempotency reasons. In those cases, the generated 'keys' will be empty.
+     *
+     * If the 'multikeyPaths' pointer is non-null, then it must point to an empty vector. If this
+     * index type supports tracking path-level multikey information, then this function resizes
+     * 'multikeyPaths' to have the same number of elements as the index key pattern and fills each
+     * element with the prefixes of the indexed field that would cause this index to be multikey as
+     * a result of inserting 'keys'.
+     *
+     * If the 'multikeyMetadataKeys' pointer is non-null, then the function will populate the
+     * BSONObjSet with any multikey metadata keys generated while processing the document. These
+     * keys are not associated with the document itself, but instead represent multi-key path
+     * information that must be stored in a reserved keyspace within the index.
+     *
+     * If any key generation errors are encountered and suppressed due to the provided GetKeysMode,
+     * 'onSuppressedErrorFn' is called.
+     */
+    using OnSuppressedErrorFn =
+        std::function<void(Status status, const BSONObj& obj, boost::optional<RecordId> loc)>;
+    void getKeys(OperationContext* opCtx,
+                 const CollectionPtr& collection,
+                 SharedBufferFragmentBuilder& pooledBufferBuilder,
+                 const BSONObj& obj,
+                 InsertDeleteOptions::ConstraintEnforcementMode mode,
+                 GetKeysContext context,
+                 KeyStringSet* keys,
+                 KeyStringSet* multikeyMetadataKeys,
+                 MultikeyPaths* multikeyPaths,
+                 boost::optional<RecordId> id,
+                 OnSuppressedErrorFn&& onSuppressedError = nullptr) const;
 
+    /**
+     * Inserts the specified keys into the index. Does not attempt to determine whether the
+     * insertion of these keys should cause the index to become multikey. The 'numInserted' output
+     * parameter, if non-nullptr, will be reset to the number of keys inserted by this function
+     * call, or to zero in the case of either a non-OK return Status or an empty 'keys' argument.
+     */
     Status insertKeys(OperationContext* opCtx,
                       const CollectionPtr& coll,
                       const KeyStringSet& keys,
-                      const RecordId& loc,
                       const InsertDeleteOptions& options,
                       KeyHandlerFn&& onDuplicateKey,
-                      int64_t* numInserted) final;
+                      int64_t* numInserted);
 
+    /**
+     * Inserts the specified keys into the index. and determines whether these keys should cause the
+     * index to become multikey. If so, this method also handles the task of marking the index as
+     * multikey in the catalog, and sets the path-level multikey information if applicable.
+     */
     Status insertKeysAndUpdateMultikeyPaths(OperationContext* opCtx,
                                             const CollectionPtr& coll,
                                             const KeyStringSet& keys,
                                             const KeyStringSet& multikeyMetadataKeys,
                                             const MultikeyPaths& multikeyPaths,
-                                            const RecordId& loc,
                                             const InsertDeleteOptions& options,
                                             KeyHandlerFn&& onDuplicateKey,
-                                            int64_t* numInserted) final;
+                                            int64_t* numInserted);
 
+    /**
+     * Analogous to insertKeys above, but remove the keys instead of inserting them.
+     * 'numDeleted' will be set to the number of keys removed from the index for the provided keys.
+     */
     Status removeKeys(OperationContext* opCtx,
                       const KeyStringSet& keys,
-                      const RecordId& loc,
                       const InsertDeleteOptions& options,
-                      int64_t* numDeleted) final;
+                      int64_t* numDeleted);
 
+    /**
+     * Gets the keys of the documents 'from' and 'to' and prepares them for the update.
+     * Provides a ticket for actually performing the update.
+     */
     void prepareUpdate(OperationContext* opCtx,
                        const CollectionPtr& collection,
-                       const IndexCatalogEntry* index,
                        const BSONObj& from,
                        const BSONObj& to,
                        const RecordId& loc,
                        const InsertDeleteOptions& options,
-                       UpdateTicket* ticket) const final;
+                       UpdateTicket* ticket) const;
+
+    /**
+     * Perform a validated update.  The keys for the 'from' object will be removed, and the keys
+     * for the object 'to' will be added.  Returns OK if the update succeeded, failure if it did
+     * not.  If an update does not succeed, the index will be unmodified, and the keys for
+     * 'from' will remain.  Assumes that the index has not changed since prepareUpdate was
+     * called.  If the index was changed, we may return an error, as our ticket may have been
+     * invalidated.
+     *
+     * 'numInserted' will be set to the number of keys inserted into the index for the document.
+     * 'numDeleted' will be set to the number of keys removed from the index for the document.
+     */
+    Status doUpdate(OperationContext* opCtx,
+                    const CollectionPtr& coll,
+                    const UpdateTicket& ticket,
+                    int64_t* numInserted,
+                    int64_t* numDeleted);
+
+    RecordId findSingle(OperationContext* opCtx,
+                        const CollectionPtr& collection,
+                        const BSONObj& key) const;
+
+    /**
+     * Returns an unpositioned cursor over 'this' index.
+     */
+    std::unique_ptr<SortedDataInterface::Cursor> newCursor(OperationContext* opCtx,
+                                                           bool isForward = true) const;
+
+
+    /**
+     * Given the set of keys, multikeyMetadataKeys and multikeyPaths generated by a particular
+     * document, return 'true' if the index should be marked as multikey and 'false' otherwise.
+     */
+    virtual bool shouldMarkIndexAsMultikey(size_t numberOfKeys,
+                                           const KeyStringSet& multikeyMetadataKeys,
+                                           const MultikeyPaths& multikeyPaths) const;
+
+    /**
+     * Provides direct access to the SortedDataInterface. This should not be used to insert
+     * documents into an index, except for testing purposes.
+     */
+    SortedDataInterface* getSortedDataInterface() const {
+        return _newInterface.get();
+    }
+
+
+    //
+    // Implementations of general IndexAccessMethod API.
+    //
+
+    SortedDataIndexAccessMethod* asSortedData() final {
+        return this;
+    }
+    const SortedDataIndexAccessMethod* asSortedData() const final {
+        return this;
+    }
+
+    Status insert(OperationContext* opCtx,
+                  SharedBufferFragmentBuilder& pooledBufferBuilder,
+                  const CollectionPtr& coll,
+                  const std::vector<BsonRecord>& bsonRecords,
+                  const InsertDeleteOptions& options,
+                  int64_t* numInserted) final;
+
+    void remove(OperationContext* opCtx,
+                SharedBufferFragmentBuilder& pooledBufferBuilder,
+                const CollectionPtr& coll,
+                const BSONObj& obj,
+                const RecordId& loc,
+                bool logIfError,
+                const InsertDeleteOptions& options,
+                int64_t* numDeleted,
+                CheckRecordId checkRecordId) final;
 
     Status update(OperationContext* opCtx,
+                  SharedBufferFragmentBuilder& pooledBufferBuilder,
+                  const BSONObj& oldDoc,
+                  const BSONObj& newDoc,
+                  const RecordId& loc,
                   const CollectionPtr& coll,
-                  const UpdateTicket& ticket,
+                  const InsertDeleteOptions& options,
                   int64_t* numInserted,
                   int64_t* numDeleted) final;
 
-    std::unique_ptr<SortedDataInterface::Cursor> newCursor(OperationContext* opCtx,
-                                                           bool isForward) const final;
-    std::unique_ptr<SortedDataInterface::Cursor> newCursor(OperationContext* opCtx) const final;
-
     Status initializeAsEmpty(OperationContext* opCtx) final;
 
     void validate(OperationContext* opCtx,
@@ -561,47 +547,14 @@ public:
 
     long long getFreeStorageBytes(OperationContext* opCtx) const final;
 
-    RecordId findSingle(OperationContext* opCtx,
-                        const CollectionPtr& collection,
-                        const BSONObj& key) const final;
-
     Status compact(OperationContext* opCtx) final;
 
-    void setIndexIsMultikey(OperationContext* opCtx,
-                            const CollectionPtr& collection,
-                            KeyStringSet multikeyMetadataKeys,
-                            MultikeyPaths paths) final;
+    Ident* getIdentPtr() const final;
 
     std::unique_ptr<BulkBuilder> initiateBulk(size_t maxMemoryUsageBytes,
                                               const boost::optional<IndexStateInfo>& stateInfo,
                                               StringData dbName) final;
 
-    Status commitBulk(OperationContext* opCtx,
-                      const CollectionPtr& collection,
-                      BulkBuilder* bulk,
-                      bool dupsAllowed,
-                      int32_t yieldIterations,
-                      const KeyHandlerFn& onDuplicateKeyInserted,
-                      const RecordIdHandlerFn& onDuplicateRecord) final;
-
-    void getKeys(OperationContext* opCtx,
-                 const CollectionPtr& collection,
-                 SharedBufferFragmentBuilder& pooledBufferBuilder,
-                 const BSONObj& obj,
-                 GetKeysMode mode,
-                 GetKeysContext context,
-                 KeyStringSet* keys,
-                 KeyStringSet* multikeyMetadataKeys,
-                 MultikeyPaths* multikeyPaths,
-                 boost::optional<RecordId> id,
-                 OnSuppressedErrorFn&& onSuppressedError = nullptr) const final;
-
-    bool shouldMarkIndexAsMultikey(size_t numberOfKeys,
-                                   const KeyStringSet& multikeyMetadataKeys,
-                                   const MultikeyPaths& multikeyPaths) const override;
-
-    SortedDataInterface* getSortedDataInterface() const override final;
-
 protected:
     /**
      * Perform some initial validation on the document to ensure it can be indexed before calling
@@ -646,10 +599,8 @@ private:
      *
      * Used by remove() only.
      */
-    void removeOneKey(OperationContext* opCtx,
-                      const KeyString::Value& keyString,
-                      const RecordId& loc,
-                      bool dupsAllowed);
+    void removeOneKey(OperationContext* opCtx, const KeyString::Value& keyString, bool dupsAllowed);
+
     /**
      * While inserting keys into index (from external sorter), if a duplicate key is detected
      * (when duplicates are not allowed), 'onDuplicateRecord' will be called if passed, otherwise a
@@ -659,9 +610,23 @@ private:
                                const KeyString::Value& dataKey,
                                const RecordIdHandlerFn& onDuplicateRecord);
 
-    void _yieldBulkLoad(OperationContext* opCtx,
-                        const Yieldable* yieldable,
-                        const NamespaceString& ns) const;
+    Status _indexKeysOrWriteToSideTable(OperationContext* opCtx,
+                                        const CollectionPtr& coll,
+                                        const KeyStringSet& keys,
+                                        const KeyStringSet& multikeyMetadataKeys,
+                                        const MultikeyPaths& multikeyPaths,
+                                        const BSONObj& obj,
+                                        const InsertDeleteOptions& options,
+                                        int64_t* keysInsertedOut);
+
+    void _unindexKeysOrWriteToSideTable(OperationContext* opCtx,
+                                        const NamespaceString& ns,
+                                        const KeyStringSet& keys,
+                                        const BSONObj& obj,
+                                        bool logIfError,
+                                        int64_t* keysDeletedOut,
+                                        InsertDeleteOptions options,
+                                        CheckRecordId checkRecordId);
 
     const std::unique_ptr<SortedDataInterface> _newInterface;
 };
diff --git a/src/mongo/db/index/index_build_interceptor.cpp b/src/mongo/db/index/index_build_interceptor.cpp
index a2734b76bad..455496c0c98 100644
--- a/src/mongo/db/index/index_build_interceptor.cpp
+++ b/src/mongo/db/index/index_build_interceptor.cpp
@@ -283,24 +283,13 @@ Status IndexBuildInterceptor::_applyWrite(OperationContext* opCtx,
     int keyLen;
     const char* binKey = operation["key"].binData(keyLen);
     BufReader reader(binKey, keyLen);
+    auto accessMethod = _indexCatalogEntry->accessMethod()->asSortedData();
     const KeyString::Value keyString = KeyString::Value::deserialize(
-        reader,
-        _indexCatalogEntry->accessMethod()->getSortedDataInterface()->getKeyStringVersion());
+        reader, accessMethod->getSortedDataInterface()->getKeyStringVersion());
 
     const Op opType = operation.getStringField("op") == "i"_sd ? Op::kInsert : Op::kDelete;
 
     const KeyStringSet keySet{keyString};
-    const RecordId opRecordId = [&]() {
-        auto keyFormat = coll->getRecordStore()->keyFormat();
-        if (keyFormat == KeyFormat::Long) {
-            return KeyString::decodeRecordIdLongAtEnd(keyString.getBuffer(), keyString.getSize());
-        } else {
-            invariant(keyFormat == KeyFormat::String);
-            return KeyString::decodeRecordIdStrAtEnd(keyString.getBuffer(), keyString.getSize());
-        }
-    }();
-
-    auto accessMethod = _indexCatalogEntry->accessMethod();
     if (opType == Op::kInsert) {
         int64_t numInserted;
         auto status = accessMethod->insertKeysAndUpdateMultikeyPaths(
@@ -309,7 +298,6 @@ Status IndexBuildInterceptor::_applyWrite(OperationContext* opCtx,
             {keySet.begin(), keySet.end()},
             {},
             MultikeyPaths{},
-            opRecordId,
             options,
             [=](const KeyString::Value& duplicateKey) {
                 return trackDups == TrackDuplicates::kTrack
@@ -330,8 +318,8 @@ Status IndexBuildInterceptor::_applyWrite(OperationContext* opCtx,
             invariant(operation.getStringField("op") == "d"_sd);
 
         int64_t numDeleted;
-        Status s = accessMethod->removeKeys(
-            opCtx, {keySet.begin(), keySet.end()}, opRecordId, options, &numDeleted);
+        Status s =
+            accessMethod->removeKeys(opCtx, {keySet.begin(), keySet.end()}, options, &numDeleted);
         if (!s.isOK()) {
             return s;
         }
@@ -426,7 +414,6 @@ Status IndexBuildInterceptor::sideWrite(OperationContext* opCtx,
                                         const KeyStringSet& keys,
                                         const KeyStringSet& multikeyMetadataKeys,
                                         const MultikeyPaths& multikeyPaths,
-                                        RecordId loc,
                                         Op op,
                                         int64_t* const numKeysOut) {
     invariant(opCtx->lockState()->inAWriteUnitOfWork());
@@ -437,7 +424,7 @@ Status IndexBuildInterceptor::sideWrite(OperationContext* opCtx,
 
     // Maintain parity with IndexAccessMethod's handling of whether keys could change the multikey
     // state on the index.
-    bool isMultikey = _indexCatalogEntry->accessMethod()->shouldMarkIndexAsMultikey(
+    bool isMultikey = _indexCatalogEntry->accessMethod()->asSortedData()->shouldMarkIndexAsMultikey(
         keys.size(), multikeyMetadataKeys, multikeyPaths);
 
     // No need to take the multikeyPaths mutex if this would not change any multikey state.
diff --git a/src/mongo/db/index/index_build_interceptor.h b/src/mongo/db/index/index_build_interceptor.h
index 32d8de7c657..07eb25cdba3 100644
--- a/src/mongo/db/index/index_build_interceptor.h
+++ b/src/mongo/db/index/index_build_interceptor.h
@@ -97,7 +97,6 @@ public:
                      const KeyStringSet& keys,
                      const KeyStringSet& multikeyMetadataKeys,
                      const MultikeyPaths& multikeyPaths,
-                     RecordId loc,
                      Op op,
                      int64_t* numKeysOut);
 
diff --git a/src/mongo/db/index/s2_bucket_key_generator_test.cpp b/src/mongo/db/index/s2_bucket_key_generator_test.cpp
index fd20519b132..f501f5cc781 100644
--- a/src/mongo/db/index/s2_bucket_key_generator_test.cpp
+++ b/src/mongo/db/index/s2_bucket_key_generator_test.cpp
@@ -155,7 +155,7 @@ TEST_F(S2BucketKeyGeneratorTest, GetS2BucketKeys) {
                                      &actualKeys,
                                      &actualMultikeyPaths,
                                      KeyString::Version::kLatestVersion,
-                                     IndexAccessMethod::GetKeysContext::kAddingKeys,
+                                     SortedDataIndexAccessMethod::GetKeysContext::kAddingKeys,
                                      Ordering::make(BSONObj()));
 
     PointSet set{{0, 0}, {3, 3}};
@@ -184,7 +184,7 @@ TEST_F(S2BucketKeyGeneratorTest, GetS2BucketKeysSubField) {
                                      &actualKeys,
                                      &actualMultikeyPaths,
                                      KeyString::Version::kLatestVersion,
-                                     IndexAccessMethod::GetKeysContext::kAddingKeys,
+                                     SortedDataIndexAccessMethod::GetKeysContext::kAddingKeys,
                                      Ordering::make(BSONObj()));
 
     PointSet set{{0, 0}, {3, 3}};
@@ -213,7 +213,7 @@ TEST_F(S2BucketKeyGeneratorTest, GetS2BucketKeysDeepSubField) {
                                      &actualKeys,
                                      &actualMultikeyPaths,
                                      KeyString::Version::kLatestVersion,
-                                     IndexAccessMethod::GetKeysContext::kAddingKeys,
+                                     SortedDataIndexAccessMethod::GetKeysContext::kAddingKeys,
                                      Ordering::make(BSONObj()));
 
     PointSet set{{0, 0}, {3, 3}};
@@ -247,7 +247,7 @@ TEST_F(S2BucketKeyGeneratorTest, GetS2BucketKeysSubFieldSomeMissing) {
                                      &actualKeys,
                                      &actualMultikeyPaths,
                                      KeyString::Version::kLatestVersion,
-                                     IndexAccessMethod::GetKeysContext::kAddingKeys,
+                                     SortedDataIndexAccessMethod::GetKeysContext::kAddingKeys,
                                      Ordering::make(BSONObj()));
 
     PointSet set{{0, 0}, {3, 3}, {5, 5}};
diff --git a/src/mongo/db/index/s2_key_generator_test.cpp b/src/mongo/db/index/s2_key_generator_test.cpp
index e5a757f7e02..e3e027049ac 100644
--- a/src/mongo/db/index/s2_key_generator_test.cpp
+++ b/src/mongo/db/index/s2_key_generator_test.cpp
@@ -136,7 +136,7 @@ struct S2KeyGeneratorTest : public unittest::Test {
                                          &keys,
                                          multikeyPaths,
                                          KeyString::Version::kLatestVersion,
-                                         IndexAccessMethod::GetKeysContext::kAddingKeys,
+                                         SortedDataIndexAccessMethod::GetKeysContext::kAddingKeys,
                                          Ordering::make(BSONObj()));
 
         ASSERT_EQUALS(1U, keys.size());
@@ -165,7 +165,7 @@ TEST_F(S2KeyGeneratorTest, GetS2KeysFromSubobjectWithArrayOfGeoAndNonGeoSubobjec
                                      &actualKeys,
                                      &actualMultikeyPaths,
                                      KeyString::Version::kLatestVersion,
-                                     IndexAccessMethod::GetKeysContext::kAddingKeys,
+                                     SortedDataIndexAccessMethod::GetKeysContext::kAddingKeys,
                                      Ordering::make(BSONObj()));
 
     KeyString::HeapBuilder keyString1(KeyString::Version::kLatestVersion,
@@ -207,7 +207,7 @@ TEST_F(S2KeyGeneratorTest, GetS2KeysFromArrayOfNonGeoSubobjectsWithArrayValues)
                                      &actualKeys,
                                      &actualMultikeyPaths,
                                      KeyString::Version::kLatestVersion,
-                                     IndexAccessMethod::GetKeysContext::kAddingKeys,
+                                     SortedDataIndexAccessMethod::GetKeysContext::kAddingKeys,
                                      Ordering::make(BSONObj()));
 
     KeyString::HeapBuilder keyString1(KeyString::Version::kLatestVersion,
@@ -243,7 +243,7 @@ TEST_F(S2KeyGeneratorTest, GetS2KeysFromMultiPointInGeoField) {
                                      &actualKeys,
                                      &actualMultikeyPaths,
                                      KeyString::Version::kLatestVersion,
-                                     IndexAccessMethod::GetKeysContext::kAddingKeys,
+                                     SortedDataIndexAccessMethod::GetKeysContext::kAddingKeys,
                                      Ordering::make(BSONObj()));
 
     const bool multiPoint = true;
@@ -279,7 +279,7 @@ TEST_F(S2KeyGeneratorTest, CollationAppliedToNonGeoStringFieldAfterGeoField) {
                                      &actualKeys,
                                      &actualMultikeyPaths,
                                      KeyString::Version::kLatestVersion,
-                                     IndexAccessMethod::GetKeysContext::kAddingKeys,
+                                     SortedDataIndexAccessMethod::GetKeysContext::kAddingKeys,
                                      Ordering::make(BSONObj()));
 
     KeyString::HeapBuilder keyString(KeyString::Version::kLatestVersion,
@@ -310,7 +310,7 @@ TEST_F(S2KeyGeneratorTest, CollationAppliedToNonGeoStringFieldBeforeGeoField) {
                                      &actualKeys,
                                      &actualMultikeyPaths,
                                      KeyString::Version::kLatestVersion,
-                                     IndexAccessMethod::GetKeysContext::kAddingKeys,
+                                     SortedDataIndexAccessMethod::GetKeysContext::kAddingKeys,
                                      Ordering::make(BSONObj()));
 
     KeyString::HeapBuilder keyString(KeyString::Version::kLatestVersion,
@@ -342,7 +342,7 @@ TEST_F(S2KeyGeneratorTest, CollationAppliedToAllNonGeoStringFields) {
                                      &actualKeys,
                                      &actualMultikeyPaths,
                                      KeyString::Version::kLatestVersion,
-                                     IndexAccessMethod::GetKeysContext::kAddingKeys,
+                                     SortedDataIndexAccessMethod::GetKeysContext::kAddingKeys,
                                      Ordering::make(BSONObj()));
 
     KeyString::HeapBuilder keyString(KeyString::Version::kLatestVersion,
@@ -376,7 +376,7 @@ TEST_F(S2KeyGeneratorTest, CollationAppliedToNonGeoStringFieldWithMultiplePathCo
                                      &actualKeys,
                                      &actualMultikeyPaths,
                                      KeyString::Version::kLatestVersion,
-                                     IndexAccessMethod::GetKeysContext::kAddingKeys,
+                                     SortedDataIndexAccessMethod::GetKeysContext::kAddingKeys,
                                      Ordering::make(BSONObj()));
 
     KeyString::HeapBuilder keyString(KeyString::Version::kLatestVersion,
@@ -407,7 +407,7 @@ TEST_F(S2KeyGeneratorTest, CollationAppliedToStringsInArray) {
                                      &actualKeys,
                                      &actualMultikeyPaths,
                                      KeyString::Version::kLatestVersion,
-                                     IndexAccessMethod::GetKeysContext::kAddingKeys,
+                                     SortedDataIndexAccessMethod::GetKeysContext::kAddingKeys,
                                      Ordering::make(BSONObj()));
 
     KeyString::HeapBuilder keyString1(KeyString::Version::kLatestVersion,
@@ -442,7 +442,7 @@ TEST_F(S2KeyGeneratorTest, CollationAppliedToStringsInAllArrays) {
                                      &actualKeys,
                                      &actualMultikeyPaths,
                                      KeyString::Version::kLatestVersion,
-                                     IndexAccessMethod::GetKeysContext::kAddingKeys,
+                                     SortedDataIndexAccessMethod::GetKeysContext::kAddingKeys,
                                      Ordering::make(BSONObj()));
 
     KeyString::HeapBuilder keyString1(KeyString::Version::kLatestVersion,
@@ -493,7 +493,7 @@ TEST_F(S2KeyGeneratorTest, CollationDoesNotAffectNonStringFields) {
                                      &actualKeys,
                                      &actualMultikeyPaths,
                                      KeyString::Version::kLatestVersion,
-                                     IndexAccessMethod::GetKeysContext::kAddingKeys,
+                                     SortedDataIndexAccessMethod::GetKeysContext::kAddingKeys,
                                      Ordering::make(BSONObj()));
 
     KeyString::HeapBuilder keyString(KeyString::Version::kLatestVersion,
@@ -523,7 +523,7 @@ TEST_F(S2KeyGeneratorTest, CollationAppliedToStringsInNestedObjects) {
                                      &actualKeys,
                                      &actualMultikeyPaths,
                                      KeyString::Version::kLatestVersion,
-                                     IndexAccessMethod::GetKeysContext::kAddingKeys,
+                                     SortedDataIndexAccessMethod::GetKeysContext::kAddingKeys,
                                      Ordering::make(BSONObj()));
 
     KeyString::HeapBuilder keyString(KeyString::Version::kLatestVersion,
@@ -555,7 +555,7 @@ TEST_F(S2KeyGeneratorTest, NoCollation) {
                                      &actualKeys,
                                      &actualMultikeyPaths,
                                      KeyString::Version::kLatestVersion,
-                                     IndexAccessMethod::GetKeysContext::kAddingKeys,
+                                     SortedDataIndexAccessMethod::GetKeysContext::kAddingKeys,
                                      Ordering::make(BSONObj()));
 
     KeyString::HeapBuilder keyString(KeyString::Version::kLatestVersion,
@@ -586,7 +586,7 @@ TEST_F(S2KeyGeneratorTest, EmptyArrayForLeadingFieldIsConsideredMultikey) {
                                      &actualKeys,
                                      &actualMultikeyPaths,
                                      KeyString::Version::kLatestVersion,
-                                     IndexAccessMethod::GetKeysContext::kAddingKeys,
+                                     SortedDataIndexAccessMethod::GetKeysContext::kAddingKeys,
                                      Ordering::make(BSONObj()));
 
     KeyString::HeapBuilder keyString(KeyString::Version::kLatestVersion,
@@ -615,7 +615,7 @@ TEST_F(S2KeyGeneratorTest, EmptyArrayForTrailingFieldIsConsideredMultikey) {
                                      &actualKeys,
                                      &actualMultikeyPaths,
                                      KeyString::Version::kLatestVersion,
-                                     IndexAccessMethod::GetKeysContext::kAddingKeys,
+                                     SortedDataIndexAccessMethod::GetKeysContext::kAddingKeys,
                                      Ordering::make(BSONObj()));
 
     KeyString::HeapBuilder keyString(KeyString::Version::kLatestVersion,
@@ -644,7 +644,7 @@ TEST_F(S2KeyGeneratorTest, SingleElementTrailingArrayIsConsideredMultikey) {
                                      &actualKeys,
                                      &actualMultikeyPaths,
                                      KeyString::Version::kLatestVersion,
-                                     IndexAccessMethod::GetKeysContext::kAddingKeys,
+                                     SortedDataIndexAccessMethod::GetKeysContext::kAddingKeys,
                                      Ordering::make(BSONObj()));
 
     KeyString::HeapBuilder keyString(KeyString::Version::kLatestVersion,
@@ -673,7 +673,7 @@ TEST_F(S2KeyGeneratorTest, MidPathSingleElementArrayIsConsideredMultikey) {
                                      &actualKeys,
                                      &actualMultikeyPaths,
                                      KeyString::Version::kLatestVersion,
-                                     IndexAccessMethod::GetKeysContext::kAddingKeys,
+                                     SortedDataIndexAccessMethod::GetKeysContext::kAddingKeys,
                                      Ordering::make(BSONObj()));
 
     KeyString::HeapBuilder keyString(KeyString::Version::kLatestVersion,
diff --git a/src/mongo/db/index/skipped_record_tracker.cpp b/src/mongo/db/index/skipped_record_tracker.cpp
index 8cc30b16337..cd6ebef448d 100644
--- a/src/mongo/db/index/skipped_record_tracker.cpp
+++ b/src/mongo/db/index/skipped_record_tracker.cpp
@@ -125,7 +125,8 @@ Status SkippedRecordTracker::retrySkippedRecords(OperationContext* opCtx,
 
     // This should only be called when constraints are being enforced, on a primary. It does not
     // make sense, nor is it necessary for this to be called on a secondary.
-    invariant(options.getKeysMode == IndexAccessMethod::GetKeysMode::kEnforceConstraints);
+    invariant(options.getKeysMode ==
+              InsertDeleteOptions::ConstraintEnforcementMode::kEnforceConstraints);
 
     static const char* curopMessage = "Index Build: retrying skipped records";
     ProgressMeterHolder progress;
@@ -162,36 +163,30 @@ Status SkippedRecordTracker::retrySkippedRecords(OperationContext* opCtx,
             auto keys = executionCtx.keys();
             auto multikeyMetadataKeys = executionCtx.multikeyMetadataKeys();
             auto multikeyPaths = executionCtx.multikeyPaths();
+            auto iam = _indexCatalogEntry->accessMethod()->asSortedData();
 
             try {
                 // Because constraint enforcement is set, this will throw if there are any indexing
                 // errors, instead of writing back to the skipped records table, which would
                 // normally happen if constraints were relaxed.
-                _indexCatalogEntry->accessMethod()->getKeys(
-                    opCtx,
-                    collection,
-                    pooledBuilder,
-                    skippedDoc,
-                    options.getKeysMode,
-                    IndexAccessMethod::GetKeysContext::kAddingKeys,
-                    keys.get(),
-                    multikeyMetadataKeys.get(),
-                    multikeyPaths.get(),
-                    skippedRecordId);
-
-                auto status = _indexCatalogEntry->accessMethod()->insertKeys(
-                    opCtx, collection, *keys, skippedRecordId, options, nullptr, nullptr);
+                iam->getKeys(opCtx,
+                             collection,
+                             pooledBuilder,
+                             skippedDoc,
+                             options.getKeysMode,
+                             SortedDataIndexAccessMethod::GetKeysContext::kAddingKeys,
+                             keys.get(),
+                             multikeyMetadataKeys.get(),
+                             multikeyPaths.get(),
+                             skippedRecordId);
+
+                auto status = iam->insertKeys(opCtx, collection, *keys, options, nullptr, nullptr);
                 if (!status.isOK()) {
                     return status;
                 }
 
-                status = _indexCatalogEntry->accessMethod()->insertKeys(opCtx,
-                                                                        collection,
-                                                                        *multikeyMetadataKeys,
-                                                                        skippedRecordId,
-                                                                        options,
-                                                                        nullptr,
-                                                                        nullptr);
+                status = iam->insertKeys(
+                    opCtx, collection, *multikeyMetadataKeys, options, nullptr, nullptr);
                 if (!status.isOK()) {
                     return status;
                 }
@@ -199,7 +194,7 @@ Status SkippedRecordTracker::retrySkippedRecords(OperationContext* opCtx,
                 return ex.toStatus();
             }
 
-            if (_indexCatalogEntry->accessMethod()->shouldMarkIndexAsMultikey(
+            if (iam->shouldMarkIndexAsMultikey(
                     keys->size(), *multikeyMetadataKeys, *multikeyPaths)) {
                 if (!_multikeyPaths) {
                     _multikeyPaths = *multikeyPaths;
diff --git a/src/mongo/db/query/sbe_stage_builder.cpp b/src/mongo/db/query/sbe_stage_builder.cpp
index 51f107bd300..33d066d3143 100644
--- a/src/mongo/db/query/sbe_stage_builder.cpp
+++ b/src/mongo/db/query/sbe_stage_builder.cpp
@@ -579,9 +579,10 @@ bool indexKeyConsistencyCheckCallback(OperationContext* opCtx,
                     str::stream() << "IndexAccessMethod not found for index " << indexId,
                     it != iamTable.end());
 
-            auto iam = it->second;
+            auto iam = it->second->asSortedData();
             tassert(5290709,
-                    str::stream() << "Expected to find IndexAccessMethod for index " << indexId,
+                    str::stream() << "Expected to find SortedDataIndexAccessMethod for index "
+                                  << indexId,
                     iam);
 
             auto& executionCtx = StorageExecutionContext::get(opCtx);
@@ -598,8 +599,8 @@ bool indexKeyConsistencyCheckCallback(OperationContext* opCtx,
                          collection,
                          pooledBuilder,
                          nextRecord.data.toBson(),
-                         IndexAccessMethod::GetKeysMode::kEnforceConstraints,
-                         IndexAccessMethod::GetKeysContext::kValidatingKeys,
+                         InsertDeleteOptions::ConstraintEnforcementMode::kEnforceConstraints,
+                         SortedDataIndexAccessMethod::GetKeysContext::kValidatingKeys,
                          keys.get(),
                          multikeyMetadataKeys,
                          multikeyPaths,
diff --git a/src/mongo/db/query/sbe_stage_builder_index_scan.cpp b/src/mongo/db/query/sbe_stage_builder_index_scan.cpp
index 38b2edc3551..c291bc957b0 100644
--- a/src/mongo/db/query/sbe_stage_builder_index_scan.cpp
+++ b/src/mongo/db/query/sbe_stage_builder_index_scan.cpp
@@ -902,7 +902,8 @@ std::pair<std::unique_ptr<sbe::PlanStage>, PlanStageSlots> generateIndexScan(
     auto keyPattern = descriptor->keyPattern();
 
     // Find the IndexAccessMethod which corresponds to the 'indexName'.
-    auto accessMethod = collection->getIndexCatalog()->getEntry(descriptor)->accessMethod();
+    auto accessMethod =
+        collection->getIndexCatalog()->getEntry(descriptor)->accessMethod()->asSortedData();
     auto intervals =
         makeIntervalsFromIndexBounds(ixn->bounds,
                                      ixn->direction == 1,
diff --git a/src/mongo/db/repl/SConscript b/src/mongo/db/repl/SConscript
index 5a1e311c712..dca5f81eb40 100644
--- a/src/mongo/db/repl/SConscript
+++ b/src/mongo/db/repl/SConscript
@@ -1668,8 +1668,6 @@ if wiredtiger:
             '$BUILD_DIR/mongo/db/commands/txn_cmd_request',
             '$BUILD_DIR/mongo/db/dbdirectclient',
             '$BUILD_DIR/mongo/db/index/index_access_methods',
-            '$BUILD_DIR/mongo/db/index/index_build_interceptor',
-            '$BUILD_DIR/mongo/db/index/skipped_record_tracker',
             '$BUILD_DIR/mongo/db/index_build_entry_helpers',
             '$BUILD_DIR/mongo/db/index_builds_coordinator_mongod',
             '$BUILD_DIR/mongo/db/logical_session_id_helpers',
diff --git a/src/mongo/db/repl/rs_rollback.cpp b/src/mongo/db/repl/rs_rollback.cpp
index 3fc5bd47d41..57bfe5e747c 100644
--- a/src/mongo/db/repl/rs_rollback.cpp
+++ b/src/mongo/db/repl/rs_rollback.cpp
@@ -873,7 +873,7 @@ void dropIndex(OperationContext* opCtx,
     }
 
     auto entry = indexCatalog->getEntry(indexDescriptor);
-    if (entry->isReady(opCtx, collection)) {
+    if (entry->isReady(opCtx)) {
         auto status = indexCatalog->dropIndex(opCtx, collection, indexDescriptor);
         if (!status.isOK()) {
             LOGV2_ERROR(21738,
diff --git a/src/mongo/db/transaction_participant.cpp b/src/mongo/db/transaction_participant.cpp
index 6b656bfd97b..76bd7283a37 100644
--- a/src/mongo/db/transaction_participant.cpp
+++ b/src/mongo/db/transaction_participant.cpp
@@ -376,7 +376,8 @@ void updateSessionEntry(OperationContext* opCtx,
                           << NamespaceString::kSessionTransactionsTableNamespace.ns(),
             idIndex);
 
-    auto indexAccess = collection->getIndexCatalog()->getEntry(idIndex)->accessMethod();
+    auto indexAccess =
+        collection->getIndexCatalog()->getEntry(idIndex)->accessMethod()->asSortedData();
     // Since we are looking up a key inside the _id index, create a key object consisting of only
     // the _id field.
     auto idToFetch = updateRequest.getQuery().firstElement();