SERVER-62437 Disallow createIndex implicit clustered collection creation

9 files changed, 66 insertions, 164 deletions

diff --git a/jstests/core/clustered_collection_create_index_clustered.js b/jstests/core/clustered_collection_create_index_clustered.js
index 3aefaa8821b..571f6ce9f5c 100644
--- a/jstests/core/clustered_collection_create_index_clustered.js
+++ b/jstests/core/clustered_collection_create_index_clustered.js
@@ -1,8 +1,5 @@
 /**
- * Tests createIndexes with the 'clustered' option on a replicated collection. Note: there are
- * different restrictions for non-replicated versus replicated clustered collections - eg replicated
- * collections can only be created with cluster key _id whereas non-replicated collections can be
- * created with arbitrary single field cluster keys.
+ * Tests createIndexes with the 'clustered' option on a replicated collection.
  *
  * @tags: [
  *   requires_fcv_52,
@@ -26,11 +23,4 @@ const replicatedDB = db.getSiblingDB("create_index_clustered");
 const collName = "coll";
 
 CreateIndexesClusteredTest.runBaseTests(replicatedDB, collName);
-
-// Only cluster key _id is valid for creating replicated clustered collections.
-CreateIndexesClusteredTest.assertCreateIndexesImplicitCreateFails(
-    replicatedDB,
-    collName,
-    {createIndexes: collName, indexes: [{key: {a: 1}, name: "a_1", clustered: true, unique: true}]},
-    5979701);
 })();
diff --git a/jstests/core/clustered_collection_creation.js b/jstests/core/clustered_collection_creation.js
index 7b3e02db552..6ee8c526eea 100644
--- a/jstests/core/clustered_collection_creation.js
+++ b/jstests/core/clustered_collection_creation.js
@@ -60,7 +60,7 @@ const validateCreateIndexOnClusterKey = function(db, collName, fullCreateOptions
 
     // 'clustered' is not a valid option for an index not on the cluster key.
     assert.commandFailedWithCode(
-        db[collName].createIndex({notMyIndex: 1}, {clustered: true, unique: true}), 6100904);
+        db[collName].createIndex({notMyIndex: 1}, {clustered: true, unique: true}), 6243700);
 
     assert.commandFailedWithCode(db[collName].runCommand({
         createIndexes: collName,
@@ -69,7 +69,7 @@ const validateCreateIndexOnClusterKey = function(db, collName, fullCreateOptions
             {key: {b: 1}, name: "b_1_clustered", clustered: true, unique: true}
         ]
     }),
-                                 6100904);
+                                 6243700);
 
     // The listIndexes output should be unchanged.
     const listIndexes1 = assert.commandWorked(db[collName].runCommand("listIndexes"));
diff --git a/jstests/libs/clustered_collections/clustered_collection_create_index_clustered_common.js b/jstests/libs/clustered_collections/clustered_collection_create_index_clustered_common.js
index 13e7f7dc666..f69d04fb18c 100644
--- a/jstests/libs/clustered_collections/clustered_collection_create_index_clustered_common.js
+++ b/jstests/libs/clustered_collections/clustered_collection_create_index_clustered_common.js
@@ -1,47 +1,11 @@
 /**
  * Encapsulates testing that verifies the behavior of createIndexes with the 'clustered' option. The
- * 'clustered' option may be used to implicitly create a clustered collection via createIndexes.
+ * 'clustered' option may not be used to create a clustered collection impicitly via createIndexes.
  */
 const CreateIndexesClusteredTest = (function() {
     "use strict";
 
     /**
-     * From createIndex with the 'clustered' option 'true', generates the corresponding
-     * createCollection options for the implicitly created 'coll'.
-     */
-    const getImplicitCreateOptsFromCreateIndex = function(cmd) {
-        const fullCreateOptions =
-            ClusteredCollectionUtil.constructFullCreateOptions({clusteredIndex: cmd.indexes[0]});
-        delete fullCreateOptions.clusteredIndex.clustered;
-        return fullCreateOptions;
-    };
-
-    /**
-     * Asserts that running the createIndexes 'cmd' implicitly creates a clustered collection.
-     */
-    const assertCreateIndexesImplicitCreateSucceeds = function(testDB, collName, cmd) {
-        assertDropCollection(testDB, collName);
-        assert.commandWorked(testDB.runCommand(cmd));
-
-        // From the createIndex command, generate the corresponding createCollection options.
-        const fullCreateOptions = getImplicitCreateOptsFromCreateIndex(cmd);
-
-        ClusteredCollectionUtil.validateListCollections(testDB, collName, fullCreateOptions);
-        ClusteredCollectionUtil.validateListIndexes(testDB, collName, fullCreateOptions);
-    };
-
-    /**
-     * Asserts that running createIndexes 'cmd' on a non-existant collection fails with 'errorCode'.
-     */
-    const assertCreateIndexesImplicitCreateFails = function(testDB, collName, cmd, errorCode) {
-        assertDropCollection(testDB, collName);
-        assert.commandFailedWithCode(
-            testDB.runCommand(cmd),
-            errorCode,
-            `Expected indexOpts ${tojson(cmd)} to fail with error code ${errorCode}`);
-    };
-
-    /**
      * Tests that createIndex with the 'clustered' option fails when a collection exists and is not
      * clustered.
      */
@@ -71,11 +35,19 @@ const CreateIndexesClusteredTest = (function() {
     };
 
     /**
-     * Tests running createIndex on a clustered collection
+     * Tests running createIndex on a clustered collection.
      */
     const runClusteredCollectionTest = function(testDB, collName) {
         assertDropCollection(testDB, collName);
 
+        // The createIndex 'clustered' option is disallowed for implicit collection creation.
+        assert.commandFailedWithCode(testDB.runCommand({
+            createIndexes: collName,
+            indexes: [{key: {_id: 1}, name: "_id_", clustered: true, unique: true}],
+        }),
+                                     6100900);
+
+        // Create the clustered collection.
         const createOptions = {
             clusteredIndex: {key: {_id: 1}, name: "theClusterKeyName", unique: true}
         };
@@ -97,10 +69,10 @@ const CreateIndexesClusteredTest = (function() {
 
         // 'clustered' is not a valid option for an index not on the cluster key.
         assert.commandFailedWithCode(
-            testColl.createIndex({notMyIndex: 1}, {clustered: true, unique: true}), 6100904);
+            testColl.createIndex({notMyIndex: 1}, {clustered: true, unique: true}), 6243700);
 
-        // Insert some docs. Sometimes empty collections are treated as special when it comes to
-        // index builds.
+        // Insert some docs. Empty collections are treated as special (single phase) when
+        // it comes to index builds.
         const batchSize = 100;
         const bulk = testColl.initializeUnorderedBulkOp();
         for (let i = 0; i < batchSize; i++) {
@@ -111,6 +83,19 @@ const CreateIndexesClusteredTest = (function() {
         assert.commandWorked(testColl.createIndex({_id: 1}));
         assert.commandWorked(testColl.createIndex({_id: 1}, {clustered: true, unique: true}));
 
+        // 'clustered' is still not a valid option for an index not on the cluster key.
+        assert.commandFailedWithCode(testColl.createIndex({a: 1}, {clustered: true, unique: true}),
+                                     6243700);
+
+        assert.commandFailedWithCode(testDB.runCommand({
+            createIndexes: collName,
+            indexes: [
+                {key: {_id: 1}, name: "_id_", clustered: true, unique: true},
+                {key: {a: 1}, name: "a_1", clustered: true, unique: true}
+            ],
+        }),
+                                     6243700);
+
         // Note: this a quirk of how we handle the 'name' field for indexes of {_id: 1}. The
         // createIndex is still a no-op, and the specified name is discarded.
         //
@@ -122,72 +107,14 @@ const CreateIndexesClusteredTest = (function() {
     };
 
     /**
-     * Runs test cases where createIndexes with 'clustered' should succeed in implicit collecton
-     * creation, regardless of whether the database is replicated.
-     */
-    const runBaseSuccessTests = function(testDB, collName) {
-        assertCreateIndexesImplicitCreateSucceeds(testDB, collName, {
-            createIndexes: collName,
-            indexes: [{key: {_id: 1}, name: "_id_", clustered: true, unique: true}]
-        });
-        assertCreateIndexesImplicitCreateSucceeds(testDB, collName, {
-            createIndexes: collName,
-            indexes: [{key: {_id: 1}, name: "_id_", clustered: true, unique: true, v: 2}]
-        });
-        assertCreateIndexesImplicitCreateSucceeds(testDB, collName, {
-            createIndexes: collName,
-            indexes: [{key: {_id: 1}, name: "uniqueIdName", clustered: true, unique: true, v: 2}]
-        });
-    };
-
-    /**
-     * Runs test cases where createIndexes with 'clustered' fails, regardless of whether the
-     * database is replciated.
-     */
-    const runBaseFailureTests = function(testDB, collName) {
-        // Missing 'unique' option.
-        assertCreateIndexesImplicitCreateFails(
-            testDB,
-            collName,
-            {createIndexes: collName, indexes: [{key: {_id: 1}, name: "_id_", clustered: true}]},
-            ErrorCodes.CannotCreateIndex);
-        // Two 'clustered' indexes.
-        assertCreateIndexesImplicitCreateFails(
-            testDB,
-            collName,
-            {
-                createIndexes: collName,
-                indexes: [
-                    {key: {_id: 1}, name: "_id_", clustered: true, unique: true},
-                    {key: {a: 1}, name: "a_1", clustered: true, unique: true}
-                ]
-            },
-            6100901);
-        assertCreateIndexesImplicitCreateFails(
-            testDB,
-            collName,
-            {
-                createIndexes: collName,
-                indexes: [
-                    {key: {_id: 1}, name: "_id_", clustered: true, unique: true, hidden: true},
-                ]
-            },
-            ErrorCodes.InvalidIndexSpecificationOption);
-    };
-
-    /**
      * Runs test cases that are agnostic to whether the database is replicated or not.
      */
     const runBaseTests = function(testDB, collName) {
         runNonClusteredCollectionTest(testDB, collName);
         runClusteredCollectionTest(testDB, collName);
-        runBaseSuccessTests(testDB, collName);
-        runBaseFailureTests(testDB, collName);
     };
 
     return {
-        assertCreateIndexesImplicitCreateSucceeds: assertCreateIndexesImplicitCreateSucceeds,
-        assertCreateIndexesImplicitCreateFails: assertCreateIndexesImplicitCreateFails,
         runBaseTests: runBaseTests,
     };
 })();
diff --git a/jstests/noPassthrough/clustered_collection_create_index_clustered_nonreplicated.js b/jstests/noPassthrough/clustered_collection_create_index_clustered_nonreplicated.js
index cbd1186b4d6..5f509b149c3 100644
--- a/jstests/noPassthrough/clustered_collection_create_index_clustered_nonreplicated.js
+++ b/jstests/noPassthrough/clustered_collection_create_index_clustered_nonreplicated.js
@@ -1,8 +1,5 @@
 /**
- * Tests createIndexes with the 'clustered' option on a replicated collection. Note: there are
- * different restrictions for non-replicated versus replicated clustered collections - eg replicated
- * collections can only be created with cluster key _id whereas non-replicated collections can be
- * created with arbitrary single field cluster keys.
+ * Tests createIndexes with the 'clustered' option on a replicated collection.
  *
  * @tags: [
  *   requires_fcv_52,
@@ -31,10 +28,5 @@ const nonReplicatedColl = nonReplicatedDB[collName];
 
 CreateIndexesClusteredTest.runBaseTests(nonReplicatedDB, collName);
 
-CreateIndexesClusteredTest.assertCreateIndexesImplicitCreateSucceeds(nonReplicatedDB, collName, {
-    createIndexes: collName,
-    indexes: [{key: {a: 1}, name: "clusterKeyYay", clustered: true, unique: true}]
-});
-
 MongoRunner.stopMongod(conn);
 })();
diff --git a/src/mongo/db/catalog/clustered_collection_util.cpp b/src/mongo/db/catalog/clustered_collection_util.cpp
index 88c1ed92f42..d71c6360050 100644
--- a/src/mongo/db/catalog/clustered_collection_util.cpp
+++ b/src/mongo/db/catalog/clustered_collection_util.cpp
@@ -121,19 +121,30 @@ bool isClusteredOnId(const boost::optional<ClusteredCollectionInfo>& collInfo) {
     return clustered_util::matchesClusterKey(BSON("_id" << 1), collInfo);
 }
 
-bool matchesClusterKey(const BSONObj& obj,
+bool matchesClusterKey(const BSONObj& keyPatternObj,
                        const boost::optional<ClusteredCollectionInfo>& collInfo) {
     if (!collInfo) {
         return false;
     }
 
-    const auto nFields = obj.nFields();
+    if (kDebugBuild) {
+        if (keyPatternObj.hasField("key")) {
+            // Double check the caller didn't accidentally provide the full index spec instead of
+            // key pattern.
+            tassert(6243701,
+                    "matchesClusterKey should be provided the raw key pattern object, not the full "
+                    "index spec",
+                    collInfo->getIndexSpec().getKey().hasField("key"));
+        }
+    }
+
+    const auto nFields = keyPatternObj.nFields();
     invariant(nFields > 0);
     if (nFields > 1) {
         // Clustered key cannot be compound.
         return false;
     }
-    return obj.firstElement().fieldNameStringData() ==
+    return keyPatternObj.firstElement().fieldNameStringData() ==
         collInfo->getIndexSpec().getKey().firstElement().fieldNameStringData();
 }
 
diff --git a/src/mongo/db/catalog/clustered_collection_util.h b/src/mongo/db/catalog/clustered_collection_util.h
index cc5fed15a6e..2dd603430de 100644
--- a/src/mongo/db/catalog/clustered_collection_util.h
+++ b/src/mongo/db/catalog/clustered_collection_util.h
@@ -74,9 +74,10 @@ bool requiresLegacyFormat(const NamespaceString& nss);
 BSONObj formatClusterKeyForListIndexes(const ClusteredCollectionInfo& collInfo);
 
 /**
- * Returns true if the BSON object matches the collection's cluster key.
+ * Returns true if the BSON object matches the collection's cluster key. Caller's should ensure
+ * keyPatternObj is the 'key' of the index spec of interest, not the entire index spec BSON.
  */
-bool matchesClusterKey(const BSONObj& obj,
+bool matchesClusterKey(const BSONObj& keyPatternObj,
                        const boost::optional<ClusteredCollectionInfo>& collInfo);
 
 /**
diff --git a/src/mongo/db/catalog/index_catalog_impl.cpp b/src/mongo/db/catalog/index_catalog_impl.cpp
index cf99ee1b45e..976e79f350f 100644
--- a/src/mongo/db/catalog/index_catalog_impl.cpp
+++ b/src/mongo/db/catalog/index_catalog_impl.cpp
@@ -898,6 +898,16 @@ Status IndexCatalogImpl::_doesSpecConflictWithExisting(OperationContext* opCtx,
 
     const BSONObj key = spec.getObjectField(IndexDescriptor::kKeyPatternFieldName);
 
+    // Check if the spec conflicts with the clusteredIndex.
+    if (spec["clustered"]) {
+        uassert(6243700,
+                "Cannot create index with option 'clustered' that does not match an existing "
+                "clustered index",
+                clustered_util::matchesClusterKey(
+                    spec.getObjectField(IndexDescriptor::kKeyPatternFieldName),
+                    collection->getClusteredInfo()));
+    }
+
     {
         // Check whether an index with the specified candidate name already exists in the catalog.
         const IndexDescriptor* desc = findIndexByName(opCtx, name, includeUnfinishedIndexes);
diff --git a/src/mongo/db/commands/create_indexes.cpp b/src/mongo/db/commands/create_indexes.cpp
index ddab8ae556b..5132b08ec3e 100644
--- a/src/mongo/db/commands/create_indexes.cpp
+++ b/src/mongo/db/commands/create_indexes.cpp
@@ -106,17 +106,9 @@ std::vector<BSONObj> parseAndValidateIndexSpecs(OperationContext* opCtx,
     const auto ns = cmd.getNamespace();
     const bool ignoreUnknownIndexOptions = cmd.getIgnoreUnknownIndexOptions();
 
-    bool containsClusteredIndex = false;
     std::vector<BSONObj> indexSpecs;
     for (const auto& index : cmd.getIndexes()) {
         BSONObj parsedIndexSpec = index;
-        if (parsedIndexSpec.hasField("clustered")) {
-            uassert(6100901,
-                    "A collection may only be clustered by a single index",
-                    !containsClusteredIndex);
-            containsClusteredIndex = true;
-        }
-
         if (ignoreUnknownIndexOptions) {
             parsedIndexSpec = index_key_validate::removeUnknownFields(parsedIndexSpec);
         }
@@ -324,21 +316,16 @@ CreateIndexesReply runCreateIndexesOnNewCollection(
             !db || !ViewCatalog::get(db)->lookup(opCtx, ns));
 
     if (createCollImplicitly) {
+        for (const auto& spec : specs) {
+            uassert(6100900,
+                    "Cannot implicitly create a new collection with createIndex 'clustered' option",
+                    !spec["clustered"]);
+        }
+
         // We need to create the collection.
         BSONObjBuilder builder;
         builder.append("create", ns.coll());
         CollectionOptions options;
-        for (const auto& spec : specs) {
-            if (spec["clustered"] && !ns.isTimeseriesBucketsCollection()) {
-                // Timeseries buckets collections are created differently than standard clustered
-                // indexes.
-                tassert(6100900,
-                        "CollectionOptions are not expected to have a clusteredIndex yet",
-                        !options.clusteredIndex.is_initialized());
-                options.clusteredIndex = clustered_util::createClusteredInfoForNewCollection(spec);
-            }
-        }
-
         builder.appendElements(options.toBSON());
         BSONObj idIndexSpec;
 
diff --git a/src/mongo/db/index_builds_coordinator.cpp b/src/mongo/db/index_builds_coordinator.cpp
index c243a40f998..a8b60a314d6 100644
--- a/src/mongo/db/index_builds_coordinator.cpp
+++ b/src/mongo/db/index_builds_coordinator.cpp
@@ -1639,25 +1639,9 @@ void IndexBuildsCoordinator::createIndexesOnEmptyCollection(OperationContext* op
     // Always run single phase index build for empty collection. And, will be coordinated using
     // createIndexes oplog entry.
     for (const auto& spec : specs) {
-        uassert(
-            6100903,
-            "An index may not have the field 'clustered' unless it is on a clustered collection",
-            !spec.hasField("clustered") || collection->isClustered());
-
-        if (collection->isClustered()) {
-            bool matchesClusterKey = clustered_util::matchesClusterKey(
-                spec.getObjectField(IndexDescriptor::kKeyPatternFieldName),
-                collection->getClusteredInfo());
-
-            uassert(6100904,
-                    "'clustered' is not a valid index option for an index that doesn't match the "
-                    "cluster key",
-                    !spec.hasField("clustered") || matchesClusterKey);
-
-            if (matchesClusterKey) {
-                // The index is already built implicitly.
-                continue;
-            }
+        if (spec.hasField("clustered") && spec.getBoolField("clustered")) {
+            // The index is already built implicitly.
+            continue;
         }
 
         // Each index will be added to the mdb catalog using the preceding createIndexes