SERVER-43323 All replicated index builds should write startIndexBuild

and commitIndexBuild oplog entries

This emits the two-phase oplog entries in cloner.cpp, system_index.cpp and
index_builder.cpp.

5 files changed, 127 insertions, 30 deletions

diff --git a/src/mongo/db/SConscript b/src/mongo/db/SConscript
index 12f2f55c5b6..5e35d6f415b 100644
--- a/src/mongo/db/SConscript
+++ b/src/mongo/db/SConscript
@@ -839,6 +839,7 @@ env.Library(
         '$BUILD_DIR/mongo/client/authentication',
         'catalog/database_holder',
         'commands/list_collections_filter',
+        'index_builds_coordinator_interface',
     ],
 )
 
diff --git a/src/mongo/db/cloner.cpp b/src/mongo/db/cloner.cpp
index 06d594d1afd..16dfcd0cd8f 100644
--- a/src/mongo/db/cloner.cpp
+++ b/src/mongo/db/cloner.cpp
@@ -52,6 +52,7 @@
 #include "mongo/db/curop.h"
 #include "mongo/db/dbdirectclient.h"
 #include "mongo/db/index/index_descriptor.h"
+#include "mongo/db/index_builds_coordinator.h"
 #include "mongo/db/jsobj.h"
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/op_observer.h"
@@ -369,20 +370,53 @@ void Cloner::copyIndexes(OperationContext* opCtx,
     // The code below throws, so ensure build cleanup occurs.
     ON_BLOCK_EXIT([&] { indexer.cleanUpAfterBuild(opCtx, collection); });
 
-    auto indexInfoObjs = uassertStatusOK(
-        indexer.init(opCtx, collection, indexesToBuild, MultiIndexBlock::kNoopOnInitFn));
+    // Emit startIndexBuild and commitIndexBuild oplog entries if supported by the current FCV.
+    auto opObserver = opCtx->getServiceContext()->getOpObserver();
+    auto fromMigrate = false;
+    auto buildUUID = serverGlobalParams.featureCompatibility.isVersionInitialized() &&
+            serverGlobalParams.featureCompatibility.getVersion() ==
+                ServerGlobalParams::FeatureCompatibility::Version::kFullyUpgradedTo44
+        ? boost::make_optional(UUID::gen())
+        : boost::none;
+
+    MultiIndexBlock::OnInitFn onInitFn;
+    if (opCtx->writesAreReplicated() && buildUUID) {
+        onInitFn = [&](std::vector<BSONObj>& specs) {
+            opObserver->onStartIndexBuild(
+                opCtx, to_collection, collection->uuid(), *buildUUID, specs, fromMigrate);
+            return Status::OK();
+        };
+    } else {
+        onInitFn = MultiIndexBlock::kNoopOnInitFn;
+    }
+
+    auto indexInfoObjs = uassertStatusOK(indexer.init(opCtx, collection, indexesToBuild, onInitFn));
     uassertStatusOK(indexer.insertAllDocumentsInCollection(opCtx, collection));
     uassertStatusOK(indexer.checkConstraints(opCtx));
 
     WriteUnitOfWork wunit(opCtx);
-    uassertStatusOK(indexer.commit(
-        opCtx, collection, MultiIndexBlock::kNoopOnCreateEachFn, MultiIndexBlock::kNoopOnCommitFn));
-    if (opCtx->writesAreReplicated()) {
-        for (auto&& infoObj : indexInfoObjs) {
-            getGlobalServiceContext()->getOpObserver()->onCreateIndex(
-                opCtx, collection->ns(), collection->uuid(), infoObj, false);
-        }
-    }
+    uassertStatusOK(
+        indexer.commit(opCtx,
+                       collection,
+                       [&](const BSONObj& spec) {
+                           // If two phase index builds are enabled, the index build will be
+                           // coordinated using startIndexBuild and commitIndexBuild oplog entries.
+                           if (opCtx->writesAreReplicated() &&
+                               !IndexBuildsCoordinator::get(opCtx)->supportsTwoPhaseIndexBuild()) {
+                               opObserver->onCreateIndex(
+                                   opCtx, collection->ns(), collection->uuid(), spec, fromMigrate);
+                           }
+                       },
+                       [&] {
+                           if (opCtx->writesAreReplicated() && buildUUID) {
+                               opObserver->onCommitIndexBuild(opCtx,
+                                                              collection->ns(),
+                                                              collection->uuid(),
+                                                              *buildUUID,
+                                                              indexInfoObjs,
+                                                              fromMigrate);
+                           }
+                       }));
     wunit.commit();
 }
 
diff --git a/src/mongo/db/index_builder.cpp b/src/mongo/db/index_builder.cpp
index f7430beb84b..202679a1a28 100644
--- a/src/mongo/db/index_builder.cpp
+++ b/src/mongo/db/index_builder.cpp
@@ -41,6 +41,7 @@
 #include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/curop.h"
 #include "mongo/db/db_raii.h"
+#include "mongo/db/index_builds_coordinator.h"
 #include "mongo/db/op_observer.h"
 #include "mongo/db/repl/timestamp_block.h"
 #include "mongo/db/server_options.h"
@@ -161,23 +162,53 @@ Status IndexBuilder::_build(OperationContext* opCtx,
         }
     }
 
-    status = writeConflictRetry(opCtx, "Commit index build", ns.ns(), [opCtx, coll, &indexer, &ns] {
-        WriteUnitOfWork wunit(opCtx);
-        auto status = indexer.commit(opCtx,
-                                     coll,
-                                     [opCtx, coll, &ns](const BSONObj& indexSpec) {
-                                         opCtx->getServiceContext()->getOpObserver()->onCreateIndex(
-                                             opCtx, ns, coll->uuid(), indexSpec, false);
-                                     },
-                                     MultiIndexBlock::kNoopOnCommitFn);
-        if (!status.isOK()) {
-            return status;
-        }
+    // Emit startIndexBuild and commitIndexBuild oplog entries if supported by the current
+    // FCV.
+    auto opObserver = opCtx->getServiceContext()->getOpObserver();
+    auto fromMigrate = false;
+    auto buildUUID = serverGlobalParams.featureCompatibility.isVersionInitialized() &&
+            serverGlobalParams.featureCompatibility.getVersion() ==
+                ServerGlobalParams::FeatureCompatibility::Version::kFullyUpgradedTo44
+        ? boost::make_optional(UUID::gen())
+        : boost::none;
+
+    if (buildUUID) {
+        opObserver->onStartIndexBuild(
+            opCtx, coll->ns(), coll->uuid(), *buildUUID, {_index}, fromMigrate);
+    }
 
-        IndexTimestampHelper::setGhostCommitTimestampForCatalogWrite(opCtx, ns);
-        wunit.commit();
-        return Status::OK();
-    });
+    status = writeConflictRetry(
+        opCtx,
+        "Commit index build",
+        ns.ns(),
+        [opCtx, coll, buildUUID, fromMigrate, &spec = _index, &indexer, &ns] {
+            WriteUnitOfWork wunit(opCtx);
+
+            auto status = indexer.commit(
+                opCtx,
+                coll,
+                [opCtx, coll, buildUUID, fromMigrate, &ns](const BSONObj& indexSpec) {
+                    // If two phase index builds are enabled, the index build will be coordinated
+                    // using startIndexBuild and commitIndexBuild oplog entries.
+                    if (!IndexBuildsCoordinator::get(opCtx)->supportsTwoPhaseIndexBuild()) {
+                        opCtx->getServiceContext()->getOpObserver()->onCreateIndex(
+                            opCtx, ns, coll->uuid(), indexSpec, fromMigrate);
+                    }
+                },
+                [opCtx, coll, buildUUID, fromMigrate, &spec, &ns] {
+                    if (buildUUID) {
+                        opCtx->getServiceContext()->getOpObserver()->onCommitIndexBuild(
+                            opCtx, coll->ns(), coll->uuid(), *buildUUID, {spec}, fromMigrate);
+                    }
+                });
+            if (!status.isOK()) {
+                return status;
+            }
+
+            IndexTimestampHelper::setGhostCommitTimestampForCatalogWrite(opCtx, ns);
+            wunit.commit();
+            return Status::OK();
+        });
     if (!status.isOK()) {
         return status;
     }
diff --git a/src/mongo/db/system_index.cpp b/src/mongo/db/system_index.cpp
index 3cde99b3a97..06c1fecc91d 100644
--- a/src/mongo/db/system_index.cpp
+++ b/src/mongo/db/system_index.cpp
@@ -245,12 +245,35 @@ void createSystemIndexes(OperationContext* opCtx, Collection* collection) {
                 opCtx, v3SystemRolesIndexSpec.toBSON(), serverGlobalParams.featureCompatibility));
     }
     if (!indexSpec.isEmpty()) {
-        opCtx->getServiceContext()->getOpObserver()->onCreateIndex(
-            opCtx, ns, collection->uuid(), indexSpec, false /* fromMigrate */);
+        // Emit startIndexBuild and commitIndexBuild oplog entries if supported by the current FCV.
+        auto opObserver = opCtx->getServiceContext()->getOpObserver();
+        auto fromMigrate = false;
+        auto buildUUID = serverGlobalParams.featureCompatibility.isVersionInitialized() &&
+                serverGlobalParams.featureCompatibility.getVersion() ==
+                    ServerGlobalParams::FeatureCompatibility::Version::kFullyUpgradedTo44
+            ? boost::make_optional(UUID::gen())
+            : boost::none;
+
+        if (buildUUID) {
+            opObserver->onStartIndexBuild(
+                opCtx, ns, collection->uuid(), *buildUUID, {indexSpec}, fromMigrate);
+        }
+
+        // If two phase index builds are enabled, the index build will be coordinated using
+        // startIndexBuild and commitIndexBuild oplog entries.
+        if (!IndexBuildsCoordinator::get(opCtx)->supportsTwoPhaseIndexBuild()) {
+            opObserver->onCreateIndex(opCtx, ns, collection->uuid(), indexSpec, fromMigrate);
+        }
+
         // Note that the opObserver is called prior to creating the index.  This ensures the index
         // write gets the same storage timestamp as the oplog entry.
         fassert(40456,
                 collection->getIndexCatalog()->createIndexOnEmptyCollection(opCtx, indexSpec));
+
+        if (buildUUID) {
+            opObserver->onCommitIndexBuild(
+                opCtx, ns, collection->uuid(), *buildUUID, {indexSpec}, fromMigrate);
+        }
     }
 }
 
diff --git a/src/mongo/dbtests/storage_timestamp_tests.cpp b/src/mongo/dbtests/storage_timestamp_tests.cpp
index 45b72b54edc..2dba77e964d 100644
--- a/src/mongo/dbtests/storage_timestamp_tests.cpp
+++ b/src/mongo/dbtests/storage_timestamp_tests.cpp
@@ -2614,8 +2614,10 @@ public:
             return;
         }
 
-        const LogicalTime indexCreateLt = futureLt.addTicks(1);
-        const Timestamp indexCreateTs = indexCreateLt.asTimestamp();
+        // The index build emits three oplog entries.
+        const Timestamp indexStartTs = futureLt.addTicks(1).asTimestamp();
+        const Timestamp indexCreateTs = futureLt.addTicks(2).asTimestamp();
+        const Timestamp indexCompleteTs = futureLt.addTicks(3).asTimestamp();
 
         NamespaceString nss("admin.system.users");
 
@@ -2638,7 +2640,9 @@ public:
         assertNamespaceInIdents(nss, pastTs, false);
         assertNamespaceInIdents(nss, presentTs, false);
         assertNamespaceInIdents(nss, futureTs, true);
+        assertNamespaceInIdents(nss, indexStartTs, true);
         assertNamespaceInIdents(nss, indexCreateTs, true);
+        assertNamespaceInIdents(nss, indexCompleteTs, true);
         assertNamespaceInIdents(nss, nullTs, true);
 
         result = queryOplog(BSON("op"
@@ -2655,7 +2659,11 @@ public:
         assertIdentsMissingAtTimestamp(durableCatalog, "", indexIdent, pastTs);
         assertIdentsMissingAtTimestamp(durableCatalog, "", indexIdent, presentTs);
         assertIdentsMissingAtTimestamp(durableCatalog, "", indexIdent, futureTs);
+        // This is the timestamp of the startIndexBuild oplog entry, which is timestamped before the
+        // index is created as part of the createIndexes oplog entry.
+        assertIdentsMissingAtTimestamp(durableCatalog, "", indexIdent, indexStartTs);
         assertIdentsExistAtTimestamp(durableCatalog, "", indexIdent, indexCreateTs);
+        assertIdentsExistAtTimestamp(durableCatalog, "", indexIdent, indexCompleteTs);
         assertIdentsExistAtTimestamp(durableCatalog, "", indexIdent, nullTs);
     }
 };