SERVER-51747 SERVER-51813 Tenant collection cloner calls write_ops_exec::performInserts() to insert documents for 2 reasons. 1) To comply with "multi-timestamp" rules. 2) To have limit on the size of the batch that gets inserted in a single WUOW.

SERVER-51845 StorageInterfaceImpl::createIndexesOnEmptyCollection() acquires stronger collection lock (X) mode while creating indexes on empty committed collection.

SERVER-51869 StorageInterfaceImpl::createIndexesOnEmptyCollection() generates oplog entry on creating indexes.

SERVER-51844 Added sanity check in tenant collection cloner to catch case like,  tenant collection not having _id index and autoIdIndex is true.

1 files changed, 28 insertions, 3 deletions

diff --git a/src/mongo/db/repl/tenant_collection_cloner.cpp b/src/mongo/db/repl/tenant_collection_cloner.cpp
index 1c1bbe88edc..02baa2de8c1 100644
--- a/src/mongo/db/repl/tenant_collection_cloner.cpp
+++ b/src/mongo/db/repl/tenant_collection_cloner.cpp
@@ -34,6 +34,7 @@
 #include "mongo/base/string_data.h"
 #include "mongo/db/commands/list_collections_filter.h"
 #include "mongo/db/db_raii.h"
+#include "mongo/db/ops/write_ops_exec.h"
 #include "mongo/db/repl/cloner_utils.h"
 #include "mongo/db/repl/database_cloner_gen.h"
 #include "mongo/db/repl/repl_server_parameters_gen.h"
@@ -218,6 +219,16 @@ BaseCloner::AfterStageBehavior TenantCollectionCloner::listIndexesStage() {
         _stats.indexes = _readyIndexSpecs.size() + (_idIndexSpec.isEmpty() ? 0 : 1);
     };
 
+    // Tenant collections are replicated collections and it's impossible to have an empty _id index
+    // and collection options 'autoIndexId' as false. These are extra sanity checks made on the
+    // response received from the remote node.
+    uassert(
+        ErrorCodes::IllegalOperation,
+        str::stream() << "Found empty '_id' index spec but the collection is not specified with "
+                         "'autoIndexId' as false, tenantId: "
+                      << _tenantId << ", namespace: " << this->_sourceNss,
+        !_idIndexSpec.isEmpty() || _collectionOptions.autoIndexId == CollectionOptions::NO);
+
     if (!_idIndexSpec.isEmpty() && _collectionOptions.autoIndexId == CollectionOptions::NO) {
         LOGV2_WARNING(4884504,
                       "Found the _id index spec but the collection specified autoIndexId of false",
@@ -293,7 +304,7 @@ void TenantCollectionCloner::handleNextBatch(DBClientCursorBatchIterator& iter)
         stdx::lock_guard<Latch> lk(_mutex);
         _stats.receivedBatches++;
         while (iter.moreInCurrentBatch()) {
-            _documentsToInsert.emplace_back(InsertStatement(iter.nextSafe()));
+            _documentsToInsert.emplace_back(iter.nextSafe());
         }
     }
 
@@ -333,7 +344,7 @@ void TenantCollectionCloner::handleNextBatch(DBClientCursorBatchIterator& iter)
 void TenantCollectionCloner::insertDocumentsCallback(
     const executor::TaskExecutor::CallbackArgs& cbd) {
     uassertStatusOK(cbd.status);
-    std::vector<InsertStatement> docs;
+    std::vector<BSONObj> docs;
 
     {
         stdx::lock_guard<Latch> lk(_mutex);
@@ -350,7 +361,21 @@ void TenantCollectionCloner::insertDocumentsCallback(
         _progressMeter.hit(int(docs.size()));
     }
 
-    uassertStatusOK(getStorageInterface()->insertDocuments(cbd.opCtx, _sourceDbAndUuid, docs));
+    write_ops::Insert insertOp(_sourceNss);
+    insertOp.setDocuments(std::move(docs));
+    insertOp.setWriteCommandBase([] {
+        write_ops::WriteCommandBase wcb;
+        wcb.setOrdered(true);
+        wcb.setBypassDocumentValidation(true);
+        return wcb;
+    }());
+
+    // write_ops_exec::PerformInserts() will handle limiting the batch size
+    // that gets inserted in a single WUOW.
+    auto writeResults = write_ops_exec::performInserts(cbd.opCtx, insertOp);
+    // Since the writes are ordered, it's ok to check just the last writeOp result.
+    uassertStatusOKWithContext(writeResults.results.back(),
+                               "Tenant collection cloner: insert documents");
 
     tenantMigrationHangDuringCollectionClone.executeIf(
         [&](const BSONObj&) {