SERVER-37726 Make dropIndexes abort in-progress index builds

11 files changed, 744 insertions, 123 deletions

diff --git a/jstests/noPassthrough/drop_indexes_aborts_in_progress_index_builds_complex_name.js b/jstests/noPassthrough/drop_indexes_aborts_in_progress_index_builds_complex_name.js
new file mode 100644
index 00000000000..1b661eeee23
--- /dev/null
+++ b/jstests/noPassthrough/drop_indexes_aborts_in_progress_index_builds_complex_name.js
@@ -0,0 +1,52 @@
+/**
+ * Tests that the "dropIndexes" command can abort in-progress index builds. The "dropIndexes"
+ * command will only abort in-progress index builds if the user specifies all of the indexes that a
+ * single builder is building together, as we can only abort at the index builder granularity level.
+ *
+ * In this file, we test calling "dropIndexes" with a complex index name whose index build is
+ * in-progress.
+ */
+(function() {
+"use strict";
+
+load("jstests/noPassthrough/libs/index_build.js");
+
+const mongodOptions = {};
+const conn = MongoRunner.runMongod(mongodOptions);
+
+const dbName = "drop_indexes_aborts_in_progress_index_builds_complex_name";
+const collName = "test";
+
+TestData.dbName = dbName;
+TestData.collName = collName;
+
+const testDB = conn.getDB(dbName);
+testDB.getCollection(collName).drop();
+
+assert.commandWorked(testDB.createCollection(collName));
+const coll = testDB.getCollection(collName);
+
+jsTest.log("Aborting index builder with one index build and complex index spec");
+assert.commandWorked(testDB.getCollection(collName).insert({a: 1}));
+assert.commandWorked(testDB.getCollection(collName).insert({b: 1}));
+
+IndexBuildTest.pauseIndexBuilds(testDB.getMongo());
+
+const awaitIndexBuild = IndexBuildTest.startIndexBuild(
+    testDB.getMongo(), coll.getFullName(), {a: 1, b: 1}, {}, [ErrorCodes.IndexBuildAborted]);
+IndexBuildTest.waitForIndexBuildToScanCollection(testDB, collName, "a_1_b_1");
+
+const awaitDropIndex = startParallelShell(() => {
+    const testDB = db.getSiblingDB(TestData.dbName);
+    assert.commandWorked(testDB.runCommand({dropIndexes: TestData.collName, index: "a_1_b_1"}));
+}, conn.port);
+
+checkLog.contains(testDB.getMongo(), "About to abort index builder");
+IndexBuildTest.resumeIndexBuilds(testDB.getMongo());
+awaitIndexBuild();
+awaitDropIndex();
+
+assert.eq(1, testDB.getCollection(collName).getIndexes().length);
+
+MongoRunner.stopMongod(conn);
+}());
diff --git a/jstests/noPassthrough/drop_indexes_aborts_in_progress_index_builds_key_pattern.js b/jstests/noPassthrough/drop_indexes_aborts_in_progress_index_builds_key_pattern.js
new file mode 100644
index 00000000000..274610d2bbc
--- /dev/null
+++ b/jstests/noPassthrough/drop_indexes_aborts_in_progress_index_builds_key_pattern.js
@@ -0,0 +1,51 @@
+/**
+ * Tests that the "dropIndexes" command can abort in-progress index builds. The "dropIndexes"
+ * command will only abort in-progress index builds if the user specifies all of the indexes that a
+ * single builder is building together, as we can only abort at the index builder granularity level.
+ *
+ * In this file, we test calling "dropIndexes" with a key pattern whose index build is in-progress.
+ */
+(function() {
+"use strict";
+
+load("jstests/noPassthrough/libs/index_build.js");
+
+const mongodOptions = {};
+const conn = MongoRunner.runMongod(mongodOptions);
+
+const dbName = "drop_indexes_aborts_in_progress_index_builds_key_pattern";
+const collName = "test";
+
+TestData.dbName = dbName;
+TestData.collName = collName;
+
+const testDB = conn.getDB(dbName);
+testDB.getCollection(collName).drop();
+
+assert.commandWorked(testDB.createCollection(collName));
+const coll = testDB.getCollection(collName);
+
+jsTest.log("Aborting index builder by key pattern");
+assert.commandWorked(testDB.getCollection(collName).insert({a: 1}));
+assert.commandWorked(testDB.getCollection(collName).insert({b: 1}));
+
+IndexBuildTest.pauseIndexBuilds(testDB.getMongo());
+
+const awaitIndexBuild = IndexBuildTest.startIndexBuild(
+    testDB.getMongo(), coll.getFullName(), {a: 1, b: 1}, {}, [ErrorCodes.IndexBuildAborted]);
+IndexBuildTest.waitForIndexBuildToScanCollection(testDB, collName, "a_1_b_1");
+
+const awaitDropIndex = startParallelShell(() => {
+    const testDB = db.getSiblingDB(TestData.dbName);
+    assert.commandWorked(testDB.runCommand({dropIndexes: TestData.collName, index: {a: 1, b: 1}}));
+}, conn.port);
+
+checkLog.contains(testDB.getMongo(), "About to abort index builder");
+IndexBuildTest.resumeIndexBuilds(testDB.getMongo());
+awaitIndexBuild();
+awaitDropIndex();
+
+assert.eq(1, testDB.getCollection(collName).getIndexes().length);
+
+MongoRunner.stopMongod(conn);
+}());
diff --git a/jstests/noPassthrough/drop_indexes_aborts_in_progress_index_builds_multiple.js b/jstests/noPassthrough/drop_indexes_aborts_in_progress_index_builds_multiple.js
new file mode 100644
index 00000000000..4e4e72e0a05
--- /dev/null
+++ b/jstests/noPassthrough/drop_indexes_aborts_in_progress_index_builds_multiple.js
@@ -0,0 +1,58 @@
+/**
+ * Tests that the "dropIndexes" command can abort in-progress index builds. The "dropIndexes"
+ * command will only abort in-progress index builds if the user specifies all of the indexes that a
+ * single builder is building together, as we can only abort at the index builder granularity level.
+ *
+ * In this file, we test calling "dropIndexes" with a list of index names, which will be used to
+ * abort a single index builder only, which was building all the given indexes.
+ */
+(function() {
+"use strict";
+
+load("jstests/noPassthrough/libs/index_build.js");
+
+const mongodOptions = {};
+const conn = MongoRunner.runMongod(mongodOptions);
+
+const dbName = "drop_indexes_aborts_in_progress_index_builds_multiple";
+const collName = "test";
+
+TestData.dbName = dbName;
+TestData.collName = collName;
+
+const testDB = conn.getDB(dbName);
+testDB.getCollection(collName).drop();
+
+assert.commandWorked(testDB.createCollection(collName));
+const coll = testDB.getCollection(collName);
+
+jsTest.log("Aborting index builder with multiple index builds");
+assert.commandWorked(testDB.getCollection(collName).insert({a: 1}));
+assert.commandWorked(testDB.getCollection(collName).insert({b: 1}));
+
+IndexBuildTest.pauseIndexBuilds(testDB.getMongo());
+
+const awaitIndexBuild = IndexBuildTest.startIndexBuild(testDB.getMongo(),
+                                                       coll.getFullName(),
+                                                       [{a: 1}, {b: 1}, {a: 1, b: 1}],
+                                                       {},
+                                                       [ErrorCodes.IndexBuildAborted]);
+IndexBuildTest.waitForIndexBuildToScanCollection(testDB, collName, "a_1");
+IndexBuildTest.waitForIndexBuildToScanCollection(testDB, collName, "b_1");
+IndexBuildTest.waitForIndexBuildToScanCollection(testDB, collName, "a_1_b_1");
+
+const awaitDropIndex = startParallelShell(() => {
+    const testDB = db.getSiblingDB(TestData.dbName);
+    assert.commandWorked(
+        testDB.runCommand({dropIndexes: TestData.collName, index: ["b_1", "a_1_b_1", "a_1"]}));
+}, conn.port);
+
+checkLog.contains(testDB.getMongo(), "About to abort index builder");
+IndexBuildTest.resumeIndexBuilds(testDB.getMongo());
+awaitIndexBuild();
+awaitDropIndex();
+
+assert.eq(1, testDB.getCollection(collName).getIndexes().length);
+
+MongoRunner.stopMongod(conn);
+}());
diff --git a/jstests/noPassthrough/drop_indexes_aborts_in_progress_index_builds_simple_name.js b/jstests/noPassthrough/drop_indexes_aborts_in_progress_index_builds_simple_name.js
new file mode 100644
index 00000000000..7af24355021
--- /dev/null
+++ b/jstests/noPassthrough/drop_indexes_aborts_in_progress_index_builds_simple_name.js
@@ -0,0 +1,51 @@
+/**
+ * Tests that the "dropIndexes" command can abort in-progress index builds. The "dropIndexes"
+ * command will only abort in-progress index builds if the user specifies all of the indexes that a
+ * single builder is building together, as we can only abort at the index builder granularity level.
+ *
+ * In this file, we test calling "dropIndexes" with a simple index name whose index build is
+ * in-progress.
+ */
+(function() {
+"use strict";
+
+load("jstests/noPassthrough/libs/index_build.js");
+
+const mongodOptions = {};
+const conn = MongoRunner.runMongod(mongodOptions);
+
+const dbName = "drop_indexes_aborts_in_progress_index_builds_simple_name";
+const collName = "test";
+
+TestData.dbName = dbName;
+TestData.collName = collName;
+
+const testDB = conn.getDB(dbName);
+testDB.getCollection(collName).drop();
+
+assert.commandWorked(testDB.createCollection(collName));
+const coll = testDB.getCollection(collName);
+
+jsTest.log("Aborting index builder with one index build and simple index spec");
+assert.commandWorked(testDB.getCollection(collName).insert({a: 1}));
+
+IndexBuildTest.pauseIndexBuilds(testDB.getMongo());
+
+const awaitIndexBuild = IndexBuildTest.startIndexBuild(
+    testDB.getMongo(), coll.getFullName(), {a: 1}, {}, [ErrorCodes.IndexBuildAborted]);
+IndexBuildTest.waitForIndexBuildToScanCollection(testDB, collName, "a_1");
+
+const awaitDropIndex = startParallelShell(() => {
+    const testDB = db.getSiblingDB(TestData.dbName);
+    assert.commandWorked(testDB.runCommand({dropIndexes: TestData.collName, index: "a_1"}));
+}, conn.port);
+
+checkLog.contains(testDB.getMongo(), "About to abort index builder");
+IndexBuildTest.resumeIndexBuilds(testDB.getMongo());
+awaitIndexBuild();
+awaitDropIndex();
+
+assert.eq(1, testDB.getCollection(collName).getIndexes().length);
+
+MongoRunner.stopMongod(conn);
+}());
diff --git a/jstests/noPassthrough/drop_indexes_aborts_in_progress_index_builds_wildcard.js b/jstests/noPassthrough/drop_indexes_aborts_in_progress_index_builds_wildcard.js
new file mode 100644
index 00000000000..a69d499b4b2
--- /dev/null
+++ b/jstests/noPassthrough/drop_indexes_aborts_in_progress_index_builds_wildcard.js
@@ -0,0 +1,59 @@
+/**
+ * Tests that the "dropIndexes" command can abort in-progress index builds. The "dropIndexes"
+ * command will only abort in-progress index builds if the user specifies all of the indexes that a
+ * single builder is building together, as we can only abort at the index builder granularity level.
+ *
+ * In this file, we test calling "dropIndexes" with the "*" wildcard which will abort all
+ * in-progress index builds and remove ready indexes.
+ */
+(function() {
+"use strict";
+
+load("jstests/noPassthrough/libs/index_build.js");
+
+const mongodOptions = {};
+const conn = MongoRunner.runMongod(mongodOptions);
+
+const dbName = "drop_indexes_aborts_in_progress_index_builds_wildcard";
+const collName = "test";
+
+TestData.dbName = dbName;
+TestData.collName = collName;
+
+const testDB = conn.getDB(dbName);
+testDB.getCollection(collName).drop();
+
+assert.commandWorked(testDB.createCollection(collName));
+const coll = testDB.getCollection(collName);
+
+jsTest.log("Aborting all index builders and ready indexes with '*' wildcard");
+assert.commandWorked(testDB.getCollection(collName).insert({a: 1}));
+assert.commandWorked(testDB.getCollection(collName).insert({b: 1}));
+
+assert.commandWorked(testDB.getCollection(collName).createIndex({a: 1}));
+
+IndexBuildTest.pauseIndexBuilds(testDB.getMongo());
+
+const awaitFirstIndexBuild = IndexBuildTest.startIndexBuild(
+    testDB.getMongo(), coll.getFullName(), {a: 1, b: 1}, {}, [ErrorCodes.IndexBuildAborted]);
+IndexBuildTest.waitForIndexBuildToScanCollection(testDB, collName, "a_1_b_1");
+
+const awaitSecondIndexBuild = IndexBuildTest.startIndexBuild(
+    testDB.getMongo(), coll.getFullName(), {b: 1}, {}, [ErrorCodes.IndexBuildAborted]);
+IndexBuildTest.waitForIndexBuildToScanCollection(testDB, collName, "b_1");
+
+const awaitDropIndex = startParallelShell(() => {
+    const testDB = db.getSiblingDB(TestData.dbName);
+    assert.commandWorked(testDB.runCommand({dropIndexes: TestData.collName, index: "*"}));
+}, conn.port);
+
+checkLog.contains(testDB.getMongo(), "About to abort all index builders on collection with UUID");
+IndexBuildTest.resumeIndexBuilds(testDB.getMongo());
+awaitFirstIndexBuild();
+awaitSecondIndexBuild();
+awaitDropIndex();
+
+assert.eq(1, testDB.getCollection(collName).getIndexes().length);
+
+MongoRunner.stopMongod(conn);
+}());
diff --git a/jstests/noPassthrough/drop_indexes_aborts_index_builder_matching_input.js b/jstests/noPassthrough/drop_indexes_aborts_index_builder_matching_input.js
new file mode 100644
index 00000000000..2575e752581
--- /dev/null
+++ b/jstests/noPassthrough/drop_indexes_aborts_index_builder_matching_input.js
@@ -0,0 +1,73 @@
+/**
+ * Tests that the "dropIndexes" command can abort in-progress index builds. The "dropIndexes"
+ * command will only abort in-progress index builds if the user specifies all of the indexes that a
+ * single builder is building together, as we can only abort at the index builder granularity level.
+ */
+(function() {
+"use strict";
+
+load("jstests/noPassthrough/libs/index_build.js");
+
+const mongodOptions = {};
+const conn = MongoRunner.runMongod(mongodOptions);
+
+const dbName = "drop_indexes_aborts_in_progress_index_builds_wildcard";
+const collName = "test";
+
+TestData.dbName = dbName;
+TestData.collName = collName;
+
+const testDB = conn.getDB(dbName);
+testDB.getCollection(collName).drop();
+
+assert.commandWorked(testDB.createCollection(collName));
+const coll = testDB.getCollection(collName);
+
+assert.commandWorked(testDB.getCollection(collName).insert({a: 1}));
+assert.commandWorked(testDB.getCollection(collName).insert({b: 1}));
+assert.commandWorked(testDB.getCollection(collName).insert({c: 1}));
+
+assert.commandWorked(testDB.getCollection(collName).createIndex({a: 1}));
+
+IndexBuildTest.pauseIndexBuilds(testDB.getMongo());
+
+const awaitFirstIndexBuild = IndexBuildTest.startIndexBuild(
+    testDB.getMongo(), coll.getFullName(), {b: 1}, {}, [ErrorCodes.IndexBuildAborted]);
+IndexBuildTest.waitForIndexBuildToScanCollection(testDB, collName, "b_1");
+
+const awaitSecondIndexBuild = IndexBuildTest.startIndexBuild(
+    testDB.getMongo(), coll.getFullName(), {c: 1}, {}, [ErrorCodes.IndexBuildAborted]);
+IndexBuildTest.waitForIndexBuildToScanCollection(testDB, collName, "c_1");
+
+let awaitDropIndex = startParallelShell(() => {
+    const testDB = db.getSiblingDB(TestData.dbName);
+    assert.commandFailedWithCode(
+        testDB.runCommand({dropIndexes: TestData.collName, index: ["a_1", "b_1"]}),
+        [ErrorCodes.BackgroundOperationInProgressForNamespace]);
+}, conn.port);
+awaitDropIndex();
+
+awaitDropIndex = startParallelShell(() => {
+    const testDB = db.getSiblingDB(TestData.dbName);
+    assert.commandFailedWithCode(
+        testDB.runCommand({dropIndexes: TestData.collName, index: ["a_1", "c_1"]}),
+        [ErrorCodes.BackgroundOperationInProgressForNamespace]);
+}, conn.port);
+awaitDropIndex();
+
+awaitDropIndex = startParallelShell(() => {
+    const testDB = db.getSiblingDB(TestData.dbName);
+    assert.commandFailedWithCode(
+        testDB.runCommand({dropIndexes: TestData.collName, index: ["b_1", "c_1"]}),
+        [ErrorCodes.BackgroundOperationInProgressForNamespace]);
+}, conn.port);
+awaitDropIndex();
+
+IndexBuildTest.resumeIndexBuilds(testDB.getMongo());
+awaitFirstIndexBuild();
+awaitSecondIndexBuild();
+
+assert.eq(4, testDB.getCollection(collName).getIndexes().length);
+
+MongoRunner.stopMongod(conn);
+}());
diff --git a/jstests/noPassthrough/libs/index_build.js b/jstests/noPassthrough/libs/index_build.js
index fa8fc380242..d2789f22a5c 100644
--- a/jstests/noPassthrough/libs/index_build.js
+++ b/jstests/noPassthrough/libs/index_build.js
@@ -3,13 +3,26 @@
 class IndexBuildTest {
     /**
      * Starts an index build in a separate mongo shell process with given options.
+     * Ensures the index build worked or failed with one of the expected failures.
      */
-    static startIndexBuild(conn, ns, keyPattern, options) {
+    static startIndexBuild(conn, ns, keyPattern, options, expectedFailures) {
         options = options || {};
-        return startParallelShell('const coll = db.getMongo().getCollection("' + ns + '");' +
-                                      'assert.commandWorked(coll.createIndex(' +
-                                      tojson(keyPattern) + ', ' + tojson(options) + '));',
-                                  conn.port);
+        expectedFailures = expectedFailures || [];
+
+        if (Array.isArray(keyPattern)) {
+            return startParallelShell(
+                'const coll = db.getMongo().getCollection("' + ns + '");' +
+                    'assert.commandWorkedOrFailedWithCode(coll.createIndexes(' +
+                    JSON.stringify(keyPattern) + ', ' + tojson(options) + '), ' +
+                    JSON.stringify(expectedFailures) + ');',
+                conn.port);
+        } else {
+            return startParallelShell('const coll = db.getMongo().getCollection("' + ns + '");' +
+                                          'assert.commandWorkedOrFailedWithCode(coll.createIndex(' +
+                                          tojson(keyPattern) + ', ' + tojson(options) + '), ' +
+                                          JSON.stringify(expectedFailures) + ');',
+                                      conn.port);
+        }
     }
 
     /**
diff --git a/src/mongo/db/catalog/drop_indexes.cpp b/src/mongo/db/catalog/drop_indexes.cpp
index 1d55cd77d59..7eaf3d6bd4a 100644
--- a/src/mongo/db/catalog/drop_indexes.cpp
+++ b/src/mongo/db/catalog/drop_indexes.cpp
@@ -52,13 +52,109 @@
 namespace mongo {
 namespace {
 
-// Field name in dropIndexes command for indexes to drop. This field can contain one of:
-// 1) '*' - drop all indexes.
-// 2) <index name> - name of single index to drop.
-// 3) <index key pattern> - BSON document representing key pattern of index to drop.
-// 4) [<index name 1>, <index name 2>, ...] - array containing names of indexes to drop.
+// Field name in dropIndexes command for indexes to drop.
 constexpr auto kIndexFieldName = "index"_sd;
 
+Status checkForViewOrMissingNS(OperationContext* opCtx,
+                               const NamespaceString& nss,
+                               Database* db,
+                               Collection* collection) {
+    if (!collection) {
+        if (db && ViewCatalog::get(db)->lookup(opCtx, nss.ns())) {
+            return Status(ErrorCodes::CommandNotSupportedOnView,
+                          str::stream() << "Cannot drop indexes on view " << nss);
+        }
+        return Status(ErrorCodes::NamespaceNotFound, str::stream() << "ns not found " << nss);
+    }
+    return Status::OK();
+}
+
+/**
+ * Validates the key pattern passed through the command.
+ */
+StatusWith<const IndexDescriptor*> getDescriptorByKeyPattern(OperationContext* opCtx,
+                                                             IndexCatalog* indexCatalog,
+                                                             const BSONElement& keyPattern) {
+    const bool includeUnfinished = true;
+    std::vector<const IndexDescriptor*> indexes;
+    indexCatalog->findIndexesByKeyPattern(
+        opCtx, keyPattern.embeddedObject(), includeUnfinished, &indexes);
+    if (indexes.empty()) {
+        return Status(ErrorCodes::IndexNotFound,
+                      str::stream()
+                          << "can't find index with key: " << keyPattern.embeddedObject());
+    } else if (indexes.size() > 1) {
+        return Status(ErrorCodes::AmbiguousIndexKeyPattern,
+                      str::stream() << indexes.size() << " indexes found for key: "
+                                    << keyPattern.embeddedObject() << ", identify by name instead."
+                                    << " Conflicting indexes: " << indexes[0]->infoObj() << ", "
+                                    << indexes[1]->infoObj());
+    }
+
+    const IndexDescriptor* desc = indexes[0];
+    if (desc->isIdIndex()) {
+        return Status(ErrorCodes::InvalidOptions, "cannot drop _id index");
+    }
+
+    if (desc->indexName() == "*") {
+        // Dropping an index named '*' results in an drop-index oplog entry with a name of '*',
+        // which in 3.6 and later is interpreted by replication as meaning "drop all indexes on
+        // this collection".
+        return Status(ErrorCodes::InvalidOptions,
+                      "cannot drop an index named '*' by key pattern.  You must drop the "
+                      "entire collection, drop all indexes on the collection by using an index "
+                      "name of '*', or downgrade to 3.4 to drop only this index.");
+    }
+
+    return desc;
+}
+
+/**
+ * Returns a list of index names that the caller requested to abort/drop. Requires a collection lock
+ * to be held to look up the index name from the key pattern.
+ */
+StatusWith<std::vector<std::string>> getIndexNames(OperationContext* opCtx,
+                                                   Collection* collection,
+                                                   const BSONElement& indexElem) {
+    invariant(opCtx->lockState()->isCollectionLockedForMode(collection->ns(), MODE_IX));
+
+    std::vector<std::string> indexNames;
+    if (indexElem.type() == String) {
+        std::string indexToAbort = indexElem.valuestr();
+        indexNames.push_back(indexToAbort);
+    } else if (indexElem.type() == Object) {
+        auto swDescriptor =
+            getDescriptorByKeyPattern(opCtx, collection->getIndexCatalog(), indexElem);
+        if (!swDescriptor.isOK()) {
+            return swDescriptor.getStatus();
+        }
+        indexNames.push_back(swDescriptor.getValue()->indexName());
+    } else if (indexElem.type() == Array) {
+        for (auto indexNameElem : indexElem.Array()) {
+            invariant(indexNameElem.type() == String);
+            indexNames.push_back(indexNameElem.valuestr());
+        }
+    }
+
+    return indexNames;
+}
+
+/**
+ * Attempts to abort a single index builder that is responsible for all the index names passed in.
+ */
+std::vector<UUID> abortIndexBuildByIndexNamesNoWait(OperationContext* opCtx,
+                                                    Collection* collection,
+                                                    std::vector<std::string> indexNames) {
+
+    boost::optional<UUID> buildUUID =
+        IndexBuildsCoordinator::get(opCtx)->abortIndexBuildByIndexNamesNoWait(
+            opCtx, collection->uuid(), indexNames, Timestamp(), "dropIndexes command");
+    if (buildUUID) {
+        return {*buildUUID};
+    }
+    return {};
+}
+
 /**
  * Drops single index given a descriptor.
  */
@@ -98,128 +194,288 @@ Status dropIndexByDescriptor(OperationContext* opCtx,
     return Status::OK();
 }
 
-Status wrappedRun(OperationContext* opCtx,
-                  Collection* collection,
-                  const BSONObj& jsobj,
-                  BSONObjBuilder* anObjBuilder) {
+/**
+ * Aborts all the index builders on the collection if the first element in 'indexesToAbort' is "*",
+ * otherwise this attempts to abort a single index builder building the given index names.
+ */
+std::vector<UUID> abortActiveIndexBuilders(OperationContext* opCtx,
+                                           Collection* collection,
+                                           const std::vector<std::string>& indexNames) {
+    invariant(opCtx->lockState()->isCollectionLockedForMode(collection->ns(), MODE_IX));
 
-    IndexCatalog* indexCatalog = collection->getIndexCatalog();
-    anObjBuilder->appendNumber("nIndexesWas", indexCatalog->numIndexesTotal(opCtx));
+    if (indexNames.empty()) {
+        return {};
+    }
 
-    BSONElement indexElem = jsobj.getField(kIndexFieldName);
-    if (indexElem.type() == String) {
-        std::string indexToDelete = indexElem.valuestr();
-
-        if (indexToDelete == "*") {
-            indexCatalog->dropAllIndexes(
-                opCtx, false, [opCtx, collection](const IndexDescriptor* desc) {
-                    opCtx->getServiceContext()->getOpObserver()->onDropIndex(opCtx,
-                                                                             collection->ns(),
-                                                                             collection->uuid(),
-                                                                             desc->indexName(),
-                                                                             desc->infoObj());
-                });
-
-            anObjBuilder->append("msg", "non-_id indexes dropped for collection");
-            return Status::OK();
-        }
+    if (indexNames.front() == "*") {
+        return IndexBuildsCoordinator::get(opCtx)->abortCollectionIndexBuildsNoWait(
+            collection->uuid(), "dropIndexes command");
+    }
+
+    return abortIndexBuildByIndexNamesNoWait(opCtx, collection, indexNames);
+}
+
+Status dropReadyIndexes(OperationContext* opCtx,
+                        Collection* collection,
+                        const std::vector<std::string>& indexNames,
+                        BSONObjBuilder* anObjBuilder) {
+    invariant(opCtx->lockState()->isCollectionLockedForMode(collection->ns(), MODE_X));
+
+    if (indexNames.empty()) {
+        return Status::OK();
+    }
 
-        bool includeUnfinished = true;
-        auto desc = indexCatalog->findIndexByName(opCtx, indexToDelete, includeUnfinished);
+    IndexCatalog* indexCatalog = collection->getIndexCatalog();
+    if (indexNames.front() == "*") {
+        indexCatalog->dropAllIndexes(
+            opCtx, false, [opCtx, collection](const IndexDescriptor* desc) {
+                opCtx->getServiceContext()->getOpObserver()->onDropIndex(opCtx,
+                                                                         collection->ns(),
+                                                                         collection->uuid(),
+                                                                         desc->indexName(),
+                                                                         desc->infoObj());
+            });
+
+        anObjBuilder->append("msg", "non-_id indexes dropped for collection");
+        return Status::OK();
+    }
+
+    bool includeUnfinished = true;
+    for (const auto& indexName : indexNames) {
+        auto desc = indexCatalog->findIndexByName(opCtx, indexName, includeUnfinished);
         if (!desc) {
             return Status(ErrorCodes::IndexNotFound,
-                          str::stream() << "index not found with name [" << indexToDelete << "]");
+                          str::stream() << "index not found with name [" << indexName << "]");
+        }
+        Status status = dropIndexByDescriptor(opCtx, collection, indexCatalog, desc);
+        if (!status.isOK()) {
+            return status;
         }
-        return dropIndexByDescriptor(opCtx, collection, indexCatalog, desc);
     }
+    return Status::OK();
+}
 
-    if (indexElem.type() == Object) {
-        const bool includeUnfinished = true;
-        std::vector<const IndexDescriptor*> indexes;
-        collection->getIndexCatalog()->findIndexesByKeyPattern(
-            opCtx, indexElem.embeddedObject(), includeUnfinished, &indexes);
-        if (indexes.empty()) {
-            return Status(ErrorCodes::IndexNotFound,
-                          str::stream()
-                              << "can't find index with key: " << indexElem.embeddedObject());
-        } else if (indexes.size() > 1) {
-            return Status(ErrorCodes::AmbiguousIndexKeyPattern,
-                          str::stream() << indexes.size()
-                                        << " indexes found for key: " << indexElem.embeddedObject()
-                                        << ", identify by name instead."
-                                        << " Conflicting indexes: " << indexes[0]->infoObj() << ", "
-                                        << indexes[1]->infoObj());
-        }
+}  // namespace
 
-        const IndexDescriptor* desc = indexes[0];
-        if (desc->isIdIndex()) {
-            return Status(ErrorCodes::InvalidOptions, "cannot drop _id index");
-        }
+Status dropIndexes(OperationContext* opCtx,
+                   const NamespaceString& nss,
+                   const BSONObj& cmdObj,
+                   BSONObjBuilder* result) {
+    // Protects the command from replica set state changes during its execution.
+    Lock::GlobalLock globalLk(opCtx, MODE_IX);
 
-        if (desc->indexName() == "*") {
-            // Dropping an index named '*' results in an drop-index oplog entry with a name of '*',
-            // which in 3.6 and later is interpreted by replication as meaning "drop all indexes on
-            // this collection".
-            return Status(ErrorCodes::InvalidOptions,
-                          "cannot drop an index named '*' by key pattern.  You must drop the "
-                          "entire collection, drop all indexes on the collection by using an index "
-                          "name of '*', or downgrade to 3.4 to drop only this index.");
-        }
+    // We only need to hold an intent lock to send abort signals to the active index builder(s) we
+    // intend to abort.
+    boost::optional<AutoGetCollection> autoColl;
+    autoColl.emplace(opCtx, nss, MODE_IX);
+
+    bool writesAreReplicatedAndNotPrimary = opCtx->writesAreReplicated() &&
+        !repl::ReplicationCoordinator::get(opCtx)->canAcceptWritesFor(opCtx, nss);
+
+    if (writesAreReplicatedAndNotPrimary) {
+        return Status(ErrorCodes::NotMaster,
+                      str::stream() << "Not primary while dropping indexes in " << nss);
+    }
 
-        return dropIndexByDescriptor(opCtx, collection, indexCatalog, desc);
+    Database* db = autoColl->getDb();
+    Collection* collection = autoColl->getCollection();
+    Status status = checkForViewOrMissingNS(opCtx, nss, db, collection);
+    if (!status.isOK()) {
+        return status;
     }
 
-    // The 'index' field contains a list of names of indexes to drop.
-    // Drops all or none of the indexes due to the enclosing WriteUnitOfWork.
+    const UUID collectionUUID = collection->uuid();
+    const NamespaceStringOrUUID nssOrUUID = {nss.db().toString(), collectionUUID};
+
+    if (!serverGlobalParams.quiet.load()) {
+        LOGV2(51806,
+              "CMD: dropIndexes {nss}: {cmdObj_kIndexFieldName_false}",
+              "nss"_attr = nss,
+              "cmdObj_kIndexFieldName_false"_attr = cmdObj[kIndexFieldName].toString(false));
+    }
+
+    result->appendNumber("nIndexesWas", collection->getIndexCatalog()->numIndexesTotal(opCtx));
+
+    // Validate basic user input.
+    BSONElement indexElem = cmdObj.getField(kIndexFieldName);
+    const bool isWildcard = indexElem.type() == String && indexElem.String() == "*";
+
+    // If an Array was passed in, verify that all the elements are of type String.
     if (indexElem.type() == Array) {
         for (auto indexNameElem : indexElem.Array()) {
             if (indexNameElem.type() != String) {
                 return Status(ErrorCodes::TypeMismatch,
                               str::stream()
-                                  << "dropIndexes " << collection->ns() << " ("
-                                  << collection->uuid() << ") failed to drop multiple indexes "
+                                  << "dropIndexes " << collection->ns() << " (" << collectionUUID
+                                  << ") failed to drop multiple indexes "
                                   << indexElem.toString(false) << ": index name must be a string");
             }
+        }
+    }
 
-            auto indexToDelete = indexNameElem.String();
-            bool includeUnfinished = true;
-            auto desc = indexCatalog->findIndexByName(opCtx, indexToDelete, includeUnfinished);
-            if (!desc) {
-                return Status(ErrorCodes::IndexNotFound,
-                              str::stream()
-                                  << "index not found with name [" << indexToDelete << "]");
-            }
-            auto status = dropIndexByDescriptor(opCtx, collection, indexCatalog, desc);
-            if (!status.isOK()) {
-                return status.withContext(
-                    str::stream() << "dropIndexes " << collection->ns() << " ("
-                                  << collection->uuid() << ") failed to drop multiple indexes "
-                                  << indexElem.toString(false) << ": " << indexToDelete);
-            }
+    IndexBuildsCoordinator* indexBuildsCoord = IndexBuildsCoordinator::get(opCtx);
+
+    // When releasing the collection lock to send the abort signal to the index builders, it's
+    // possible for new index builds to start. Keep aborting in-progress index builds if they
+    // satisfy the caller's input.
+    std::vector<UUID> abortedIndexBuilders;
+    std::vector<std::string> indexNames;
+    while (true) {
+        auto swIndexNames = getIndexNames(opCtx, collection, indexElem);
+        if (!swIndexNames.isOK()) {
+            return swIndexNames.getStatus();
         }
 
-        return Status::OK();
+        indexNames = swIndexNames.getValue();
+
+        // Send the abort signal to any index builders that match the users request.
+        abortedIndexBuilders = abortActiveIndexBuilders(opCtx, collection, indexNames);
+
+        // Now that the abort signals were sent to the intended index builders, release our lock
+        // temporarily to allow the index builders to process the abort signal. Holding a lock here
+        // will cause the index builders to block indefinitely.
+        autoColl = boost::none;
+        if (abortedIndexBuilders.size() == 1) {
+            indexBuildsCoord->awaitIndexBuildFinished(collectionUUID, abortedIndexBuilders.front());
+        } else if (abortedIndexBuilders.size() > 1) {
+            // Only the "*" wildcard can abort multiple index builders.
+            invariant(isWildcard);
+            indexBuildsCoord->awaitNoIndexBuildInProgressForCollection(collectionUUID);
+        }
+
+        // Take an exclusive lock on the collection now to be able to perform index catalog writes
+        // when removing ready indexes from disk.
+        autoColl.emplace(opCtx, nssOrUUID, MODE_X);
+
+        // Abandon the snapshot as the index catalog will compare the in-memory state to the disk
+        // state, which may have changed when we released the lock temporarily.
+        opCtx->recoveryUnit()->abandonSnapshot();
+
+        db = autoColl->getDb();
+        collection = autoColl->getCollection();
+        if (!collection) {
+            return {ErrorCodes::NamespaceNotFound,
+                    str::stream() << "Collection not found on database " << nss.db()
+                                  << " with UUID " << collectionUUID};
+        }
+
+        // Check to see if a new index build was started that the caller requested to be aborted.
+        bool abortAgain = false;
+        if (isWildcard) {
+            abortAgain = indexBuildsCoord->inProgForCollection(collectionUUID);
+        } else {
+            abortAgain = indexBuildsCoord->hasIndexBuilder(opCtx, collectionUUID, indexNames);
+        }
+
+        if (!abortAgain) {
+            break;
+        }
+
+        // We only need to hold an intent lock to send abort signals to the active index
+        // builder(s) we intend to abort.
+        autoColl = boost::none;
+        autoColl.emplace(opCtx, nssOrUUID, MODE_IX);
+
+        // Abandon the snapshot as the index catalog will compare the in-memory state to the
+        // disk state, which may have changed when we released the lock temporarily.
+        opCtx->recoveryUnit()->abandonSnapshot();
+
+        db = autoColl->getDb();
+        collection = autoColl->getCollection();
+        if (!collection) {
+            return {ErrorCodes::NamespaceNotFound,
+                    str::stream() << "Collection not found on database " << nss.db()
+                                  << " with UUID " << collectionUUID};
+        }
     }
 
-    return Status(ErrorCodes::IndexNotFound,
-                  str::stream() << "invalid index name spec: " << indexElem.toString(false));
-}
+    // If the "*" wildcard was not specified, verify that all the index names belonging to the
+    // index builder were aborted. If not, they must be ready, so we drop them.
+    if (!isWildcard && !abortedIndexBuilders.empty()) {
+        invariant(abortedIndexBuilders.size() == 1);
+
+        return writeConflictRetry(
+            opCtx, "dropIndexes", nss.db(), [opCtx, &collection, &indexNames, result] {
+                WriteUnitOfWork wunit(opCtx);
+
+                // This is necessary to check shard version.
+                OldClientContext ctx(opCtx, collection->ns().ns());
+
+                size_t numReady = 0;
+                const bool includeUnfinished = false;
+                IndexCatalog* indexCatalog = collection->getIndexCatalog();
+                for (const auto& indexName : indexNames) {
+                    const IndexDescriptor* desc =
+                        indexCatalog->findIndexByName(opCtx, indexName, includeUnfinished);
+                    if (!desc) {
+                        // The given index name was successfully aborted.
+                        continue;
+                    }
+
+                    Status status = dropIndexByDescriptor(opCtx, collection, indexCatalog, desc);
+                    if (!status.isOK()) {
+                        return status;
+                    }
+
+                    numReady++;
+                }
+
+                invariant(numReady == 0 || numReady == indexNames.size());
+
+                wunit.commit();
+                return Status::OK();
+            });
+    }
 
-}  // namespace
+    if (!abortedIndexBuilders.empty()) {
+        // All the index builders were sent the abort signal, remove all the remaining indexes in
+        // the index catalog.
+        invariant(isWildcard);
+        invariant(indexNames.size() == 1);
+        invariant(indexNames.front() == "*");
+        invariant(collection->getIndexCatalog()->numIndexesInProgress(opCtx) == 0);
+    } else {
+        // The index catalog requires that no active index builders are running when dropping
+        // indexes.
+        BackgroundOperation::assertNoBgOpInProgForNs(collection->ns());
+        IndexBuildsCoordinator::get(opCtx)->assertNoIndexBuildInProgForCollection(collectionUUID);
+    }
 
-Status dropIndexes(OperationContext* opCtx,
-                   const NamespaceString& nss,
-                   const BSONObj& cmdObj,
-                   BSONObjBuilder* result) {
+    return writeConflictRetry(
+        opCtx, "dropIndexes", nss.db(), [opCtx, &collection, &indexNames, result] {
+            WriteUnitOfWork wunit(opCtx);
+
+            // This is necessary to check shard version.
+            OldClientContext ctx(opCtx, collection->ns().ns());
+
+            // Use an empty BSONObjBuilder to avoid duplicate appends to result on retry loops.
+            BSONObjBuilder tempObjBuilder;
+            Status status = dropReadyIndexes(opCtx, collection, indexNames, &tempObjBuilder);
+            if (!status.isOK()) {
+                return status;
+            }
+
+            wunit.commit();
+
+            result->appendElementsUnique(
+                tempObjBuilder.done());  // This append will only happen once.
+            return Status::OK();
+        });
+}
+
+Status dropIndexesForApplyOps(OperationContext* opCtx,
+                              const NamespaceString& nss,
+                              const BSONObj& cmdObj,
+                              BSONObjBuilder* result) {
     return writeConflictRetry(opCtx, "dropIndexes", nss.db(), [opCtx, &nss, &cmdObj, result] {
         AutoGetCollection autoColl(opCtx, nss, MODE_X);
 
-        bool userInitiatedWritesAndNotPrimary = opCtx->writesAreReplicated() &&
-            !repl::ReplicationCoordinator::get(opCtx)->canAcceptWritesFor(opCtx, nss);
-
-        if (userInitiatedWritesAndNotPrimary) {
-            return Status(ErrorCodes::NotMaster,
-                          str::stream() << "Not primary while dropping indexes in " << nss);
+        // If db/collection does not exist, short circuit and return.
+        Database* db = autoColl.getDb();
+        Collection* collection = autoColl.getCollection();
+        Status status = checkForViewOrMissingNS(opCtx, nss, db, collection);
+        if (!status.isOK()) {
+            return status;
         }
 
         if (!serverGlobalParams.quiet.load()) {
@@ -229,22 +485,16 @@ Status dropIndexes(OperationContext* opCtx,
                   "cmdObj_kIndexFieldName_false"_attr = cmdObj[kIndexFieldName].toString(false));
         }
 
-        // If db/collection does not exist, short circuit and return.
-        Database* db = autoColl.getDb();
-        Collection* collection = autoColl.getCollection();
-        if (!collection) {
-            if (db && ViewCatalog::get(db)->lookup(opCtx, nss.ns())) {
-                return Status(ErrorCodes::CommandNotSupportedOnView,
-                              str::stream() << "Cannot drop indexes on view " << nss);
-            }
-
-            return Status(ErrorCodes::NamespaceNotFound, "ns not found");
-        }
-
         BackgroundOperation::assertNoBgOpInProgForNs(nss);
         IndexBuildsCoordinator::get(opCtx)->assertNoIndexBuildInProgForCollection(
             collection->uuid());
 
+        BSONElement indexElem = cmdObj.getField(kIndexFieldName);
+        auto swIndexNames = getIndexNames(opCtx, collection, indexElem);
+        if (!swIndexNames.isOK()) {
+            return swIndexNames.getStatus();
+        }
+
         WriteUnitOfWork wunit(opCtx);
 
         // This is necessary to check shard version.
@@ -252,7 +502,7 @@ Status dropIndexes(OperationContext* opCtx,
 
         // Use an empty BSONObjBuilder to avoid duplicate appends to result on retry loops.
         BSONObjBuilder tempObjBuilder;
-        Status status = wrappedRun(opCtx, collection, cmdObj, &tempObjBuilder);
+        status = dropReadyIndexes(opCtx, collection, swIndexNames.getValue(), &tempObjBuilder);
         if (!status.isOK()) {
             return status;
         }
diff --git a/src/mongo/db/catalog/drop_indexes.h b/src/mongo/db/catalog/drop_indexes.h
index 65f74e4d860..c465f6dab52 100644
--- a/src/mongo/db/catalog/drop_indexes.h
+++ b/src/mongo/db/catalog/drop_indexes.h
@@ -36,12 +36,27 @@ class NamespaceString;
 class OperationContext;
 
 /**
- * Drops the index from collection "nss" that matches the "idxDescriptor" and populates
- * "result" with some statistics about the dropped index.
+ * Drops one or more ready indexes, or aborts a single index builder from the "nss" collection that
+ * matches the caller's "cmdObj" input. Populates "result" with some statistics about the operation.
+ *
+ * "cmdObj" must have a field named "index" that has one of the following as its value:
+ * 1) "*" <-- Aborts all index builders and drops all ready indexes except the _id index.
+ * 2) "indexName" <-- Aborts an index builder or drops a ready index with the given name.
+ * 3) { keyPattern } <-- Aborts an index builder or drops a ready index with a matching key pattern.
+ * 4) ["indexName1", ..., "indexNameN"] <-- Aborts an index builder or drops ready indexes that
+ *                                          match the given names.
  */
 Status dropIndexes(OperationContext* opCtx,
                    const NamespaceString& nss,
-                   const BSONObj& idxDescriptor,
+                   const BSONObj& cmdObj,
                    BSONObjBuilder* result);
 
+/**
+ * Same behaviour as "dropIndexes" but only drops ready indexes.
+ */
+Status dropIndexesForApplyOps(OperationContext* opCtx,
+                              const NamespaceString& nss,
+                              const BSONObj& cmdObj,
+                              BSONObjBuilder* result);
+
 }  // namespace mongo
diff --git a/src/mongo/db/catalog/index_catalog_impl.cpp b/src/mongo/db/catalog/index_catalog_impl.cpp
index 35fc295c73e..27f873b532f 100644
--- a/src/mongo/db/catalog/index_catalog_impl.cpp
+++ b/src/mongo/db/catalog/index_catalog_impl.cpp
@@ -947,7 +947,6 @@ void IndexCatalogImpl::dropAllIndexes(OperationContext* opCtx, bool includingIdI
 
 Status IndexCatalogImpl::dropIndex(OperationContext* opCtx, const IndexDescriptor* desc) {
     invariant(opCtx->lockState()->isCollectionLockedForMode(_collection->ns(), MODE_X));
-    invariant(!haveAnyIndexesInProgress());
 
     IndexCatalogEntry* entry = _readyIndexes.find(desc);
 
diff --git a/src/mongo/db/repl/oplog.cpp b/src/mongo/db/repl/oplog.cpp
index c792ba15917..14d57c18b74 100644
--- a/src/mongo/db/repl/oplog.cpp
+++ b/src/mongo/db/repl/oplog.cpp
@@ -824,7 +824,7 @@ const StringMap<ApplyOpMetadata> kOpsMap = {
      {[](OperationContext* opCtx, const OplogEntry& entry, OplogApplication::Mode mode) -> Status {
           BSONObjBuilder resultWeDontCareAbout;
           const auto& cmd = entry.getObject();
-          return dropIndexes(
+          return dropIndexesForApplyOps(
               opCtx, extractNsFromUUID(opCtx, entry.getUuid().get()), cmd, &resultWeDontCareAbout);
       },
       {ErrorCodes::NamespaceNotFound, ErrorCodes::IndexNotFound}}},
@@ -832,7 +832,7 @@ const StringMap<ApplyOpMetadata> kOpsMap = {
      {[](OperationContext* opCtx, const OplogEntry& entry, OplogApplication::Mode mode) -> Status {
           BSONObjBuilder resultWeDontCareAbout;
           const auto& cmd = entry.getObject();
-          return dropIndexes(
+          return dropIndexesForApplyOps(
               opCtx, extractNsFromUUID(opCtx, entry.getUuid().get()), cmd, &resultWeDontCareAbout);
       },
       {ErrorCodes::NamespaceNotFound, ErrorCodes::IndexNotFound}}},
@@ -840,7 +840,7 @@ const StringMap<ApplyOpMetadata> kOpsMap = {
      {[](OperationContext* opCtx, const OplogEntry& entry, OplogApplication::Mode mode) -> Status {
           BSONObjBuilder resultWeDontCareAbout;
           const auto& cmd = entry.getObject();
-          return dropIndexes(
+          return dropIndexesForApplyOps(
               opCtx, extractNsFromUUID(opCtx, entry.getUuid().get()), cmd, &resultWeDontCareAbout);
       },
       {ErrorCodes::NamespaceNotFound, ErrorCodes::IndexNotFound}}},
@@ -848,7 +848,7 @@ const StringMap<ApplyOpMetadata> kOpsMap = {
      {[](OperationContext* opCtx, const OplogEntry& entry, OplogApplication::Mode mode) -> Status {
           BSONObjBuilder resultWeDontCareAbout;
           const auto& cmd = entry.getObject();
-          return dropIndexes(
+          return dropIndexesForApplyOps(
               opCtx, extractNsFromUUID(opCtx, entry.getUuid().get()), cmd, &resultWeDontCareAbout);
       },
       {ErrorCodes::NamespaceNotFound, ErrorCodes::IndexNotFound}}},