SERVER-26514 Create command should take idIndex option

19 files changed, 997 insertions, 249 deletions

diff --git a/jstests/core/batch_write_command_insert.js b/jstests/core/batch_write_command_insert.js
index 581639002d8..274f35513e7 100644
--- a/jstests/core/batch_write_command_insert.js
+++ b/jstests/core/batch_write_command_insert.js
@@ -343,7 +343,7 @@ print(tojson(result));
 assert(result.ok, tojson(result));
 assert.eq(0, result.n);
 assert.eq(0, result.writeErrors[0].index);
-assert.eq(coll.getIndexes().length, 1);
+assert.eq(coll.getIndexes().length, 0);
 
 //
 // Invalid index desc
diff --git a/jstests/core/create_collection.js b/jstests/core/create_collection.js
index 483219cad19..c7858f5f902 100644
--- a/jstests/core/create_collection.js
+++ b/jstests/core/create_collection.js
@@ -1,8 +1,131 @@
 // Tests for the "create" command.
 (function() {
-    'use strict';
+    "use strict";
 
-    var res = db.createCollection('create_collection', {unknown: 1});
-    assert.commandFailed(res);
-    assert.eq(res.codeName, 'InvalidOptions');
+    load("jstests/libs/get_index_helpers.js");
+
+    // "create" command rejects invalid options.
+    db.create_collection.drop();
+    assert.commandFailedWithCode(db.createCollection("create_collection", {unknown: 1}),
+                                 ErrorCodes.InvalidOptions);
+
+    //
+    // Tests for "idIndex" field.
+    //
+
+    // "idIndex" field not allowed with "viewOn".
+    db.create_collection.drop();
+    assert.commandWorked(db.createCollection("create_collection"));
+    assert.commandFailedWithCode(db.runCommand({
+        create: "create_view",
+        viewOn: "create_collection",
+        idIndex: {key: {_id: 1}, name: "_id_"}
+    }),
+                                 ErrorCodes.InvalidOptions);
+
+    // "idIndex" field not allowed with "autoIndexId".
+    db.create_collection.drop();
+    assert.commandFailedWithCode(
+        db.createCollection("create_collection",
+                            {autoIndexId: false, idIndex: {key: {_id: 1}, name: "_id_"}}),
+        ErrorCodes.InvalidOptions);
+
+    // "idIndex" field must be an object.
+    db.create_collection.drop();
+    assert.commandFailedWithCode(db.createCollection("create_collection", {idIndex: 1}),
+                                 ErrorCodes.TypeMismatch);
+
+    // "idIndex" field cannot be empty.
+    db.create_collection.drop();
+    assert.commandFailedWithCode(db.createCollection("create_collection", {idIndex: {}}),
+                                 ErrorCodes.FailedToParse);
+
+    // "idIndex" field must be a specification for an _id index.
+    db.create_collection.drop();
+    assert.commandFailedWithCode(
+        db.createCollection("create_collection", {idIndex: {key: {a: 1}, name: "a_1"}}),
+        ErrorCodes.BadValue);
+
+    // "idIndex" field must have "key" equal to {_id: 1}.
+    db.create_collection.drop();
+    assert.commandFailedWithCode(
+        db.createCollection("create_collection", {idIndex: {key: {a: 1}, name: "_id_"}}),
+        ErrorCodes.BadValue);
+
+    // "idIndex" field must have name equal to "_id_".
+    db.create_collection.drop();
+    assert.commandFailedWithCode(
+        db.createCollection("create_collection", {idIndex: {key: {_id: 1}, name: "a_1"}}),
+        ErrorCodes.BadValue);
+
+    // "idIndex" field must only contain fields that are allowed for an _id index.
+    db.create_collection.drop();
+    assert.commandFailedWithCode(
+        db.createCollection("create_collection",
+                            {idIndex: {key: {_id: 1}, name: "_id_", sparse: true}}),
+        ErrorCodes.InvalidIndexSpecificationOption);
+
+    // "create" creates v=2 _id index when "v" is not specified in "idIndex".
+    db.create_collection.drop();
+    assert.commandWorked(
+        db.createCollection("create_collection", {idIndex: {key: {_id: 1}, name: "_id_"}}));
+    var indexSpec = GetIndexHelpers.findByName(db.create_collection.getIndexes(), "_id_");
+    assert.neq(indexSpec, null);
+    assert.eq(indexSpec.v, 2, tojson(indexSpec));
+
+    // "create" creates v=1 _id index when "idIndex" has "v" equal to 1.
+    db.create_collection.drop();
+    assert.commandWorked(
+        db.createCollection("create_collection", {idIndex: {key: {_id: 1}, name: "_id_", v: 1}}));
+    indexSpec = GetIndexHelpers.findByName(db.create_collection.getIndexes(), "_id_");
+    assert.neq(indexSpec, null);
+    assert.eq(indexSpec.v, 1, tojson(indexSpec));
+
+    // "create" creates v=2 _id index when "idIndex" has "v" equal to 2.
+    db.create_collection.drop();
+    assert.commandWorked(
+        db.createCollection("create_collection", {idIndex: {key: {_id: 1}, name: "_id_", v: 2}}));
+    indexSpec = GetIndexHelpers.findByName(db.create_collection.getIndexes(), "_id_");
+    assert.neq(indexSpec, null);
+    assert.eq(indexSpec.v, 2, tojson(indexSpec));
+
+    // "collation" field of "idIndex" must match collection default collation.
+    db.create_collection.drop();
+    assert.commandFailedWithCode(
+        db.createCollection("create_collection",
+                            {idIndex: {key: {_id: 1}, name: "_id_", collation: {locale: "en_US"}}}),
+        ErrorCodes.BadValue);
+
+    db.create_collection.drop();
+    assert.commandFailedWithCode(db.createCollection("create_collection", {
+        collation: {locale: "fr_CA"},
+        idIndex: {key: {_id: 1}, name: "_id_", collation: {locale: "en_US"}}
+    }),
+                                 ErrorCodes.BadValue);
+
+    db.create_collection.drop();
+    assert.commandFailedWithCode(db.createCollection("create_collection", {
+        collation: {locale: "fr_CA"},
+        idIndex: {key: {_id: 1}, name: "_id_", collation: {locale: "simple"}}
+    }),
+                                 ErrorCodes.BadValue);
+
+    db.create_collection.drop();
+    assert.commandWorked(db.createCollection("create_collection", {
+        collation: {locale: "en_US", strength: 3},
+        idIndex: {key: {_id: 1}, name: "_id_", collation: {locale: "en_US"}}
+    }));
+    indexSpec = GetIndexHelpers.findByName(db.create_collection.getIndexes(), "_id_");
+    assert.neq(indexSpec, null);
+    assert.eq(indexSpec.collation.locale, "en_US", tojson(indexSpec));
+
+    // If "collation" field is not present in "idIndex", _id index inherits collection default
+    // collation.
+    db.create_collection.drop();
+    assert.commandWorked(db.createCollection(
+        "create_collection",
+        {collation: {locale: "en_US"}, idIndex: {key: {_id: 1}, name: "_id_"}}));
+    indexSpec = GetIndexHelpers.findByName(db.create_collection.getIndexes(), "_id_");
+    assert.neq(indexSpec, null);
+    assert.eq(indexSpec.collation.locale, "en_US", tojson(indexSpec));
 })();
diff --git a/jstests/noPassthrough/index_version_v2.js b/jstests/noPassthrough/index_version_v2.js
index 0c8681f348b..8aebb35fdfb 100644
--- a/jstests/noPassthrough/index_version_v2.js
+++ b/jstests/noPassthrough/index_version_v2.js
@@ -149,6 +149,14 @@
     indexSpec = getIndexSpecByName(testDB.index_version, "_id_");
     assert.eq(1, indexSpec.v, tojson(indexSpec));
 
+    // Test that the _id index of a collection is created with v=1 when the
+    // featureCompatibilityVersion is 3.2 and an "idIndex" spec is provided without a version.
+    testDB.index_version.drop();
+    assert.commandWorked(
+        testDB.runCommand({create: "index_version", idIndex: {key: {_id: 1}, name: "_id_"}}));
+    indexSpec = getIndexSpecByName(testDB.index_version, "_id_");
+    assert.eq(1, indexSpec.v, tojson(indexSpec));
+
     // Test that an index created on an existing collection is created with v=1 when the
     // featureCompatibilityVersion is 3.2.
     assert.commandWorked(testDB.index_version.createIndex({defaultToV1: 1}, {name: "defaultToV1"}));
diff --git a/src/mongo/db/catalog/SConscript b/src/mongo/db/catalog/SConscript
index a7c769b461e..a5f7bb1f68b 100644
--- a/src/mongo/db/catalog/SConscript
+++ b/src/mongo/db/catalog/SConscript
@@ -28,6 +28,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/common',
         '$BUILD_DIR/mongo/db/index/index_descriptor',
         '$BUILD_DIR/mongo/db/index_names',
+        '$BUILD_DIR/mongo/db/query/collation/collator_factory_interface',
         '$BUILD_DIR/mongo/util/fail_point',
     ],
 )
@@ -39,6 +40,7 @@ env.CppUnitTest(
         'index_spec_validate_test.cpp',
     ],
     LIBDEPS=[
+        '$BUILD_DIR/mongo/db/query/query_test_service_context',
         'index_key_validate',
     ]
 )
diff --git a/src/mongo/db/catalog/collection_compact.cpp b/src/mongo/db/catalog/collection_compact.cpp
index 260d2a1e692..adce9262c28 100644
--- a/src/mongo/db/catalog/collection_compact.cpp
+++ b/src/mongo/db/catalog/collection_compact.cpp
@@ -155,7 +155,8 @@ StatusWith<CompactStats> Collection::compact(OperationContext* txn,
 
             const BSONObj spec = _compactAdjustIndexSpec(descriptor->infoObj());
             const BSONObj key = spec.getObjectField("key");
-            const Status keyStatus = validateKeyPattern(key, descriptor->version());
+            const Status keyStatus =
+                index_key_validate::validateKeyPattern(key, descriptor->version());
             if (!keyStatus.isOK()) {
                 return StatusWith<CompactStats>(
                     ErrorCodes::CannotCreateIndex,
diff --git a/src/mongo/db/catalog/create_collection.cpp b/src/mongo/db/catalog/create_collection.cpp
index afa5204fcbf..743c1481213 100644
--- a/src/mongo/db/catalog/create_collection.cpp
+++ b/src/mongo/db/catalog/create_collection.cpp
@@ -40,7 +40,10 @@
 #include "mongo/db/repl/replication_coordinator_global.h"
 
 namespace mongo {
-Status createCollection(OperationContext* txn, const std::string& dbName, const BSONObj& cmdObj) {
+Status createCollection(OperationContext* txn,
+                        const std::string& dbName,
+                        const BSONObj& cmdObj,
+                        const BSONObj& idIndex) {
     BSONObjIterator it(cmdObj);
 
     // Extract ns from first cmdObj element.
@@ -79,7 +82,8 @@ Status createCollection(OperationContext* txn, const std::string& dbName, const
         WriteUnitOfWork wunit(txn);
 
         // Create collection.
-        status = userCreateNS(txn, ctx.db(), nss.ns(), options);
+        const bool createDefaultIndexes = true;
+        status = userCreateNS(txn, ctx.db(), nss.ns(), options, createDefaultIndexes, idIndex);
         if (!status.isOK()) {
             return status;
         }
diff --git a/src/mongo/db/catalog/create_collection.h b/src/mongo/db/catalog/create_collection.h
index 84a042dae12..5f503a692a2 100644
--- a/src/mongo/db/catalog/create_collection.h
+++ b/src/mongo/db/catalog/create_collection.h
@@ -29,13 +29,19 @@
 #include <string>
 
 #include "mongo/base/status.h"
+#include "mongo/bson/bsonobj.h"
 
 namespace mongo {
 class BSONObj;
 class OperationContext;
 
 /**
- * Creates a collection as described in "cmdObj" on the database "dbName".
+ * Creates a collection as described in "cmdObj" on the database "dbName". Creates the collection's
+ * _id index according to 'idIndex', if it is non-empty. When 'idIndex' is empty, creates the
+ * default _id index.
  */
-Status createCollection(OperationContext* txn, const std::string& dbName, const BSONObj& cmdObj);
+Status createCollection(OperationContext* txn,
+                        const std::string& dbName,
+                        const BSONObj& cmdObj,
+                        const BSONObj& idIndex = BSONObj());
 }  // namespace mongo
diff --git a/src/mongo/db/catalog/database.cpp b/src/mongo/db/catalog/database.cpp
index 1a48923a7bb..31d0ca25a3e 100644
--- a/src/mongo/db/catalog/database.cpp
+++ b/src/mongo/db/catalog/database.cpp
@@ -560,7 +560,8 @@ Status Database::createView(OperationContext* txn,
 Collection* Database::createCollection(OperationContext* txn,
                                        StringData ns,
                                        const CollectionOptions& options,
-                                       bool createIdIndex) {
+                                       bool createIdIndex,
+                                       const BSONObj& idIndex) {
     invariant(txn->lockState()->isDbLockedForMode(name(), MODE_X));
     invariant(!options.isView());
 
@@ -594,7 +595,9 @@ Collection* Database::createCollection(OperationContext* txn,
                     serverGlobalParams.featureCompatibility.version.load();
                 IndexCatalog* ic = collection->getIndexCatalog();
                 uassertStatusOK(ic->createIndexOnEmptyCollection(
-                    txn, ic->getDefaultIdIndexSpec(featureCompatibilityVersion)));
+                    txn,
+                    !idIndex.isEmpty() ? idIndex
+                                       : ic->getDefaultIdIndexSpec(featureCompatibilityVersion)));
             }
         }
 
@@ -667,15 +670,12 @@ void Database::dropDatabase(OperationContext* txn, Database* db) {
     getGlobalServiceContext()->getGlobalStorageEngine()->dropDatabase(txn, name);
 }
 
-/** { ..., capped: true, size: ..., max: ... }
- * @param createDefaultIndexes - if false, defers id (and other) index creation.
- * @return true if successful
-*/
 Status userCreateNS(OperationContext* txn,
                     Database* db,
                     StringData ns,
                     BSONObj options,
-                    bool createDefaultIndexes) {
+                    bool createDefaultIndexes,
+                    const BSONObj& idIndex) {
     invariant(db);
 
     LOG(1) << "create collection " << ns << ' ' << options;
@@ -740,7 +740,7 @@ Status userCreateNS(OperationContext* txn,
     if (collectionOptions.isView()) {
         uassertStatusOK(db->createView(txn, ns, collectionOptions));
     } else {
-        invariant(db->createCollection(txn, ns, collectionOptions, createDefaultIndexes));
+        invariant(db->createCollection(txn, ns, collectionOptions, createDefaultIndexes, idIndex));
     }
 
     return Status::OK();
diff --git a/src/mongo/db/catalog/database.h b/src/mongo/db/catalog/database.h
index bb5741cb498..1af75756abc 100644
--- a/src/mongo/db/catalog/database.h
+++ b/src/mongo/db/catalog/database.h
@@ -156,7 +156,8 @@ public:
     Collection* createCollection(OperationContext* txn,
                                  StringData ns,
                                  const CollectionOptions& options = CollectionOptions(),
-                                 bool createDefaultIndexes = true);
+                                 bool createDefaultIndexes = true,
+                                 const BSONObj& idIndex = BSONObj());
 
     Status createView(OperationContext* txn, StringData viewName, const CollectionOptions& options);
 
@@ -258,10 +259,17 @@ private:
 
 void dropAllDatabasesExceptLocal(OperationContext* txn);
 
+/**
+ * Creates the namespace 'ns' in the database 'db' according to 'options'. If 'createDefaultIndexes'
+ * is true, creates the _id index for the collection (and the system indexes, in the case of system
+ * collections). Creates the collection's _id index according to 'idIndex', if it is non-empty. When
+ * 'idIndex' is empty, creates the default _id index.
+ */
 Status userCreateNS(OperationContext* txn,
                     Database* db,
                     StringData ns,
                     BSONObj options,
-                    bool createDefaultIndexes = true);
+                    bool createDefaultIndexes = true,
+                    const BSONObj& idIndex = BSONObj());
 
 }  // namespace mongo
diff --git a/src/mongo/db/catalog/index_catalog.cpp b/src/mongo/db/catalog/index_catalog.cpp
index a4c484c5a53..e680a852360 100644
--- a/src/mongo/db/catalog/index_catalog.cpp
+++ b/src/mongo/db/catalog/index_catalog.cpp
@@ -496,7 +496,7 @@ Status IndexCatalog::_isSpecOk(OperationContext* txn, const BSONObj& spec) const
     auto indexVersion = static_cast<IndexVersion>(*vEltAsInt);
 
     if (indexVersion >= IndexVersion::kV2) {
-        auto status = validateIndexSpecFieldNames(spec);
+        auto status = index_key_validate::validateIndexSpecFieldNames(spec);
         if (!status.isOK()) {
             return status;
         }
@@ -562,7 +562,7 @@ Status IndexCatalog::_isSpecOk(OperationContext* txn, const BSONObj& spec) const
                                     << "\" is too long (127 byte max)");
 
     const BSONObj key = spec.getObjectField("key");
-    const Status keyStatus = validateKeyPattern(key, indexVersion);
+    const Status keyStatus = index_key_validate::validateKeyPattern(key, indexVersion);
     if (!keyStatus.isOK()) {
         return Status(ErrorCodes::CannotCreateIndex,
                       str::stream() << "bad index key pattern " << key << ": "
diff --git a/src/mongo/db/catalog/index_key_validate.cpp b/src/mongo/db/catalog/index_key_validate.cpp
index c6d9a7ecb95..eb7f794136e 100644
--- a/src/mongo/db/catalog/index_key_validate.cpp
+++ b/src/mongo/db/catalog/index_key_validate.cpp
@@ -42,11 +42,14 @@
 #include "mongo/db/index_names.h"
 #include "mongo/db/jsobj.h"
 #include "mongo/db/namespace_string.h"
+#include "mongo/db/query/collation/collator_factory_interface.h"
+#include "mongo/db/service_context.h"
 #include "mongo/util/fail_point_service.h"
 #include "mongo/util/mongoutils/str.h"
 #include "mongo/util/represent_as.h"
 
 namespace mongo {
+namespace index_key_validate {
 
 using std::string;
 
@@ -57,6 +60,41 @@ namespace {
 // names will be disabled. This will allow for creation of indexes with invalid field names in their
 // specification.
 MONGO_FP_DECLARE(skipIndexCreateFieldNameValidation);
+
+static const std::set<StringData> allowedFieldNames = {
+    IndexDescriptor::k2dIndexMaxFieldName,
+    IndexDescriptor::k2dIndexBitsFieldName,
+    IndexDescriptor::k2dIndexMaxFieldName,
+    IndexDescriptor::k2dIndexMinFieldName,
+    IndexDescriptor::k2dsphereCoarsestIndexedLevel,
+    IndexDescriptor::k2dsphereFinestIndexedLevel,
+    IndexDescriptor::k2dsphereVersionFieldName,
+    IndexDescriptor::kBackgroundFieldName,
+    IndexDescriptor::kCollationFieldName,
+    IndexDescriptor::kDefaultLanguageFieldName,
+    IndexDescriptor::kDropDuplicatesFieldName,
+    IndexDescriptor::kExpireAfterSecondsFieldName,
+    IndexDescriptor::kGeoHaystackBucketSize,
+    IndexDescriptor::kIndexNameFieldName,
+    IndexDescriptor::kIndexVersionFieldName,
+    IndexDescriptor::kKeyPatternFieldName,
+    IndexDescriptor::kLanguageOverrideFieldName,
+    IndexDescriptor::kNamespaceFieldName,
+    IndexDescriptor::kPartialFilterExprFieldName,
+    IndexDescriptor::kSparseFieldName,
+    IndexDescriptor::kStorageEngineFieldName,
+    IndexDescriptor::kTextVersionFieldName,
+    IndexDescriptor::kUniqueFieldName,
+    IndexDescriptor::kWeightsFieldName,
+    // Index creation under legacy writeMode can result in an index spec with an _id field.
+    "_id"};
+
+static const std::set<StringData> allowedIdIndexFieldNames = {
+    IndexDescriptor::kCollationFieldName,
+    IndexDescriptor::kIndexNameFieldName,
+    IndexDescriptor::kIndexVersionFieldName,
+    IndexDescriptor::kKeyPatternFieldName,
+    IndexDescriptor::kNamespaceFieldName};
 }
 
 Status validateKeyPattern(const BSONObj& key, IndexDescriptor::IndexVersion indexVersion) {
@@ -173,6 +211,7 @@ StatusWith<BSONObj> validateIndexSpec(
     const NamespaceString& expectedNamespace,
     const ServerGlobalParams::FeatureCompatibility& featureCompatibility) {
     bool hasKeyPatternField = false;
+    bool hasIndexNameField = false;
     bool hasNamespaceField = false;
     bool hasVersionField = false;
     bool hasCollationField = false;
@@ -215,6 +254,15 @@ StatusWith<BSONObj> validateIndexSpec(
             }
 
             hasKeyPatternField = true;
+        } else if (IndexDescriptor::kIndexNameFieldName == indexSpecElemFieldName) {
+            if (indexSpecElem.type() != BSONType::String) {
+                return {ErrorCodes::TypeMismatch,
+                        str::stream() << "The field '" << IndexDescriptor::kIndexNameFieldName
+                                      << "' must be a string, but got "
+                                      << typeName(indexSpecElem.type())};
+            }
+
+            hasIndexNameField = true;
         } else if (IndexDescriptor::kNamespaceFieldName == indexSpecElemFieldName) {
             if (indexSpecElem.type() != BSONType::String) {
                 return {ErrorCodes::TypeMismatch,
@@ -271,11 +319,17 @@ StatusWith<BSONObj> validateIndexSpec(
         } else if (IndexDescriptor::kCollationFieldName == indexSpecElemFieldName) {
             if (indexSpecElem.type() != BSONType::Object) {
                 return {ErrorCodes::TypeMismatch,
-                        str::stream() << "The field '" << IndexDescriptor::kNamespaceFieldName
+                        str::stream() << "The field '" << IndexDescriptor::kCollationFieldName
                                       << "' must be an object, but got "
                                       << typeName(indexSpecElem.type())};
             }
 
+            if (indexSpecElem.Obj().isEmpty()) {
+                return {ErrorCodes::BadValue,
+                        str::stream() << "The field '" << IndexDescriptor::kCollationFieldName
+                                      << "' cannot be an empty object."};
+            }
+
             hasCollationField = true;
         } else {
             // We can assume field name is valid at this point. Validation of fieldname is handled
@@ -295,6 +349,12 @@ StatusWith<BSONObj> validateIndexSpec(
                               << "' field is a required property of an index specification"};
     }
 
+    if (!hasIndexNameField) {
+        return {ErrorCodes::FailedToParse,
+                str::stream() << "The '" << IndexDescriptor::kIndexNameFieldName
+                              << "' field is a required property of an index specification"};
+    }
+
     if (hasCollationField && *resolvedIndexVersion < IndexVersion::kV2) {
         return {ErrorCodes::CannotCreateIndex,
                 str::stream() << "Invalid index specification " << indexSpec
@@ -329,6 +389,41 @@ StatusWith<BSONObj> validateIndexSpec(
     return indexSpec;
 }
 
+Status validateIdIndexSpec(const BSONObj& indexSpec) {
+    for (auto&& indexSpecElem : indexSpec) {
+        auto indexSpecElemFieldName = indexSpecElem.fieldNameStringData();
+        if (!allowedIdIndexFieldNames.count(indexSpecElemFieldName)) {
+            return {
+                ErrorCodes::InvalidIndexSpecificationOption,
+                str::stream() << "The field '" << indexSpecElemFieldName
+                              << "' is not valid for an _id index specification. Specification: "
+                              << indexSpec};
+        }
+    }
+
+    auto keyPatternElem = indexSpec[IndexDescriptor::kKeyPatternFieldName];
+    // validateIndexSpec() should have already verified that 'keyPatternElem' is an object.
+    invariant(keyPatternElem.type() == BSONType::Object);
+    if (SimpleBSONObjComparator::kInstance.evaluate(keyPatternElem.Obj() != BSON("_id" << 1))) {
+        return {ErrorCodes::BadValue,
+                str::stream() << "The field '" << IndexDescriptor::kKeyPatternFieldName
+                              << "' for an _id index must be {_id: 1}, but got "
+                              << keyPatternElem.Obj()};
+    }
+
+    auto nameElem = indexSpec[IndexDescriptor::kIndexNameFieldName];
+    // validateIndexSpec() should have already verified that 'nameElem' is a string.
+    invariant(nameElem.type() == BSONType::String);
+    if (nameElem.String() != "_id_"_sd) {
+        return {ErrorCodes::BadValue,
+                str::stream() << "The field '" << IndexDescriptor::kIndexNameFieldName
+                              << "' for an _id index must be '_id_', but got "
+                              << nameElem.String()};
+    }
+
+    return Status::OK();
+}
+
 /**
  * Top-level index spec field names are validated here. When adding a new field with a document as
  * value, is the the sub-module's responsibility to ensure that the content is valid and that only
@@ -339,34 +434,6 @@ Status validateIndexSpecFieldNames(const BSONObj& indexSpec) {
         return Status::OK();
     }
 
-    const std::set<StringData> allowedFieldNames = {
-        IndexDescriptor::k2dIndexMaxFieldName,
-        IndexDescriptor::k2dIndexBitsFieldName,
-        IndexDescriptor::k2dIndexMaxFieldName,
-        IndexDescriptor::k2dIndexMinFieldName,
-        IndexDescriptor::k2dsphereCoarsestIndexedLevel,
-        IndexDescriptor::k2dsphereFinestIndexedLevel,
-        IndexDescriptor::k2dsphereVersionFieldName,
-        IndexDescriptor::kBackgroundFieldName,
-        IndexDescriptor::kCollationFieldName,
-        IndexDescriptor::kDefaultLanguageFieldName,
-        IndexDescriptor::kDropDuplicatesFieldName,
-        IndexDescriptor::kExpireAfterSecondsFieldName,
-        IndexDescriptor::kGeoHaystackBucketSize,
-        IndexDescriptor::kIndexNameFieldName,
-        IndexDescriptor::kIndexVersionFieldName,
-        IndexDescriptor::kKeyPatternFieldName,
-        IndexDescriptor::kLanguageOverrideFieldName,
-        IndexDescriptor::kNamespaceFieldName,
-        IndexDescriptor::kPartialFilterExprFieldName,
-        IndexDescriptor::kSparseFieldName,
-        IndexDescriptor::kStorageEngineFieldName,
-        IndexDescriptor::kTextVersionFieldName,
-        IndexDescriptor::kUniqueFieldName,
-        IndexDescriptor::kWeightsFieldName,
-        // Index creation under legacy writeMode can result in an index spec with an _id field.
-        "_id"};
-
     for (auto&& indexSpecElem : indexSpec) {
         auto indexSpecElemFieldName = indexSpecElem.fieldNameStringData();
         if (!allowedFieldNames.count(indexSpecElemFieldName)) {
@@ -380,4 +447,62 @@ Status validateIndexSpecFieldNames(const BSONObj& indexSpec) {
     return Status::OK();
 }
 
+StatusWith<BSONObj> validateIndexSpecCollation(OperationContext* txn,
+                                               const BSONObj& indexSpec,
+                                               const CollatorInterface* defaultCollator) {
+    if (auto collationElem = indexSpec[IndexDescriptor::kCollationFieldName]) {
+        // validateIndexSpec() should have already verified that 'collationElem' is an object.
+        invariant(collationElem.type() == BSONType::Object);
+
+        auto collator = CollatorFactoryInterface::get(txn->getServiceContext())
+                            ->makeFromBSON(collationElem.Obj());
+        if (!collator.isOK()) {
+            return collator.getStatus();
+        }
+
+        if (collator.getValue()) {
+            // If the collator factory returned a non-null collator, then inject the entire
+            // collation specification into the index specification. This is necessary to fill
+            // in any options that the user omitted.
+            BSONObjBuilder bob;
+
+            for (auto&& indexSpecElem : indexSpec) {
+                if (IndexDescriptor::kCollationFieldName != indexSpecElem.fieldNameStringData()) {
+                    bob.append(indexSpecElem);
+                }
+            }
+            bob.append(IndexDescriptor::kCollationFieldName,
+                       collator.getValue()->getSpec().toBSON());
+
+            return bob.obj();
+        } else {
+            // If the collator factory returned a null collator (representing the "simple"
+            // collation), then we simply omit the "collation" from the index specification.
+            // This is desirable to make the representation for the "simple" collation
+            // consistent between v=1 and v=2 indexes.
+            return indexSpec.removeField(IndexDescriptor::kCollationFieldName);
+        }
+    } else if (defaultCollator) {
+        // validateIndexSpec() should have added the "v" field if it was not present and
+        // verified that 'versionElem' is a number.
+        auto versionElem = indexSpec[IndexDescriptor::kIndexVersionFieldName];
+        invariant(versionElem.isNumber());
+
+        if (IndexVersion::kV2 <= static_cast<IndexVersion>(versionElem.numberInt())) {
+            // The user did not specify an explicit collation for this index and the collection
+            // has a default collator. If we're building a v=2 index, then we should inherit the
+            // collection default. However, if we're building a v=1 index, then we're implicitly
+            // building an index that's using the "simple" collation.
+            BSONObjBuilder bob;
+
+            bob.appendElements(indexSpec);
+            bob.append(IndexDescriptor::kCollationFieldName, defaultCollator->getSpec().toBSON());
+
+            return bob.obj();
+        }
+    }
+    return indexSpec;
+}
+
+}  // namespace index_key_validate
 }  // namespace mongo
diff --git a/src/mongo/db/catalog/index_key_validate.h b/src/mongo/db/catalog/index_key_validate.h
index 3b2105bdd5c..bb2cc7ff123 100644
--- a/src/mongo/db/catalog/index_key_validate.h
+++ b/src/mongo/db/catalog/index_key_validate.h
@@ -33,11 +33,14 @@
 
 namespace mongo {
 class BSONObj;
+class CollatorInterface;
 class NamespaceString;
 class Status;
 template <typename T>
 class StatusWith;
 
+namespace index_key_validate {
+
 /**
  * Checks if the key is valid for building an index according to the validation rules for the given
  * index version.
@@ -55,9 +58,25 @@ StatusWith<BSONObj> validateIndexSpec(
     const ServerGlobalParams::FeatureCompatibility& featureCompatibility);
 
 /**
+ * Performs additional validation for _id index specifications. This should be called after
+ * validateIndexSpec().
+ */
+Status validateIdIndexSpec(const BSONObj& indexSpec);
+
+/**
  * Confirms that 'indexSpec' contains only valid field names. Returns an error if an unexpected
  * field name is found.
  */
 Status validateIndexSpecFieldNames(const BSONObj& indexSpec);
 
+/**
+ * Validates the 'collation' field in the index specification 'indexSpec' and fills in the full
+ * collation spec. If 'collation' is missing, fills it in with the spec for 'defaultCollator'.
+ * Returns the index specification with 'collation' filled in.
+ */
+StatusWith<BSONObj> validateIndexSpecCollation(OperationContext* txn,
+                                               const BSONObj& indexSpec,
+                                               const CollatorInterface* defaultCollator);
+
+}  // namespace index_key_validate
 }  // namespace mongo
diff --git a/src/mongo/db/catalog/index_key_validate_test.cpp b/src/mongo/db/catalog/index_key_validate_test.cpp
index aa4f5a7a5db..113ab187092 100644
--- a/src/mongo/db/catalog/index_key_validate_test.cpp
+++ b/src/mongo/db/catalog/index_key_validate_test.cpp
@@ -42,6 +42,7 @@ namespace mongo {
 namespace {
 
 using IndexVersion = IndexDescriptor::IndexVersion;
+using index_key_validate::validateKeyPattern;
 
 TEST(IndexKeyValidateTest, KeyElementValueOfSmallPositiveIntSucceeds) {
     for (auto indexVersion : IndexDescriptor::getSupportedIndexVersions()) {
diff --git a/src/mongo/db/catalog/index_spec_validate_test.cpp b/src/mongo/db/catalog/index_spec_validate_test.cpp
index 798ee88a399..e2e77d3cfe2 100644
--- a/src/mongo/db/catalog/index_spec_validate_test.cpp
+++ b/src/mongo/db/catalog/index_spec_validate_test.cpp
@@ -38,11 +38,17 @@
 #include "mongo/bson/bsonmisc.h"
 #include "mongo/bson/bsonobjbuilder.h"
 #include "mongo/db/namespace_string.h"
+#include "mongo/db/query/collation/collator_interface_mock.h"
+#include "mongo/db/query/query_test_service_context.h"
 #include "mongo/unittest/unittest.h"
 
 namespace mongo {
 namespace {
 
+using index_key_validate::validateIndexSpec;
+using index_key_validate::validateIdIndexSpec;
+using index_key_validate::validateIndexSpecCollation;
+
 const NamespaceString kTestNamespace("test", "index_spec_validate");
 
 /**
@@ -63,14 +69,22 @@ TEST(IndexSpecValidateTest, ReturnsAnErrorIfKeyPatternIsNotAnObject) {
     featureCompatibility.validateFeaturesAsMaster.store(true);
 
     ASSERT_EQ(ErrorCodes::TypeMismatch,
-              validateIndexSpec(BSON("key" << 1), kTestNamespace, featureCompatibility));
+              validateIndexSpec(BSON("key" << 1 << "name"
+                                           << "indexName"),
+                                kTestNamespace,
+                                featureCompatibility));
     ASSERT_EQ(ErrorCodes::TypeMismatch,
               validateIndexSpec(BSON("key"
-                                     << "not an object"),
+                                     << "not an object"
+                                     << "name"
+                                     << "indexName"),
                                 kTestNamespace,
                                 featureCompatibility));
     ASSERT_EQ(ErrorCodes::TypeMismatch,
-              validateIndexSpec(BSON("key" << BSONArray()), kTestNamespace, featureCompatibility));
+              validateIndexSpec(BSON("key" << BSONArray() << "name"
+                                           << "indexName"),
+                                kTestNamespace,
+                                featureCompatibility));
 }
 
 TEST(IndexSpecValidateTest, ReturnsAnErrorIfFieldRepeatedInKeyPattern) {
@@ -79,12 +93,15 @@ TEST(IndexSpecValidateTest, ReturnsAnErrorIfFieldRepeatedInKeyPattern) {
     featureCompatibility.validateFeaturesAsMaster.store(true);
 
     ASSERT_EQ(ErrorCodes::BadValue,
-              validateIndexSpec(BSON("key" << BSON("field" << 1 << "field" << 1)),
+              validateIndexSpec(BSON("key" << BSON("field" << 1 << "field" << 1) << "name"
+                                           << "indexName"),
                                 kTestNamespace,
                                 featureCompatibility));
     ASSERT_EQ(ErrorCodes::BadValue,
               validateIndexSpec(BSON("key" << BSON("field" << 1 << "otherField" << -1 << "field"
-                                                           << "2dsphere")),
+                                                           << "2dsphere")
+                                           << "name"
+                                           << "indexName"),
                                 kTestNamespace,
                                 featureCompatibility));
 }
@@ -95,7 +112,31 @@ TEST(IndexSpecValidateTest, ReturnsAnErrorIfKeyPatternIsNotPresent) {
     featureCompatibility.validateFeaturesAsMaster.store(true);
 
     ASSERT_EQ(ErrorCodes::FailedToParse,
-              validateIndexSpec(BSONObj(), kTestNamespace, featureCompatibility));
+              validateIndexSpec(BSON("name"
+                                     << "indexName"),
+                                kTestNamespace,
+                                featureCompatibility));
+}
+
+TEST(IndexSpecValidateTest, ReturnsAnErrorIfNameIsNotAString) {
+    ServerGlobalParams::FeatureCompatibility featureCompatibility;
+    featureCompatibility.version.store(ServerGlobalParams::FeatureCompatibility::Version::k34);
+    featureCompatibility.validateFeaturesAsMaster.store(true);
+
+    ASSERT_EQ(ErrorCodes::TypeMismatch,
+              validateIndexSpec(BSON("key" << BSON("field" << 1) << "name" << 1),
+                                kTestNamespace,
+                                featureCompatibility));
+}
+
+TEST(IndexSpecValidateTest, ReturnsAnErrorIfNameIsNotPresent) {
+    ServerGlobalParams::FeatureCompatibility featureCompatibility;
+    featureCompatibility.version.store(ServerGlobalParams::FeatureCompatibility::Version::k34);
+    featureCompatibility.validateFeaturesAsMaster.store(true);
+
+    ASSERT_EQ(
+        ErrorCodes::FailedToParse,
+        validateIndexSpec(BSON("key" << BSON("field" << 1)), kTestNamespace, featureCompatibility));
 }
 
 TEST(IndexSpecValidateTest, ReturnsAnErrorIfNamespaceIsNotAString) {
@@ -104,11 +145,17 @@ TEST(IndexSpecValidateTest, ReturnsAnErrorIfNamespaceIsNotAString) {
     featureCompatibility.validateFeaturesAsMaster.store(true);
 
     ASSERT_EQ(ErrorCodes::TypeMismatch,
-              validateIndexSpec(BSON("key" << BSON("field" << 1) << "ns" << 1),
+              validateIndexSpec(BSON("key" << BSON("field" << 1) << "name"
+                                           << "indexName"
+                                           << "ns"
+                                           << 1),
                                 kTestNamespace,
                                 featureCompatibility));
     ASSERT_EQ(ErrorCodes::TypeMismatch,
-              validateIndexSpec(BSON("key" << BSON("field" << 1) << "ns" << BSONObj()),
+              validateIndexSpec(BSON("key" << BSON("field" << 1) << "name"
+                                           << "indexName"
+                                           << "ns"
+                                           << BSONObj()),
                                 kTestNamespace,
                                 featureCompatibility));
 }
@@ -119,7 +166,9 @@ TEST(IndexSpecValidateTest, ReturnsAnErrorIfNamespaceIsEmptyString) {
     featureCompatibility.validateFeaturesAsMaster.store(true);
 
     ASSERT_EQ(ErrorCodes::BadValue,
-              validateIndexSpec(BSON("key" << BSON("field" << 1) << "ns"
+              validateIndexSpec(BSON("key" << BSON("field" << 1) << "name"
+                                           << "indexName"
+                                           << "ns"
                                            << ""),
                                 NamespaceString(),
                                 featureCompatibility));
@@ -131,7 +180,9 @@ TEST(IndexSpecValidateTest, ReturnsAnErrorIfNamespaceDoesNotMatch) {
     featureCompatibility.validateFeaturesAsMaster.store(true);
 
     ASSERT_EQ(ErrorCodes::BadValue,
-              validateIndexSpec(BSON("key" << BSON("field" << 1) << "ns"
+              validateIndexSpec(BSON("key" << BSON("field" << 1) << "name"
+                                           << "indexName"
+                                           << "ns"
                                            << "some string"),
                                 kTestNamespace,
                                 featureCompatibility));
@@ -139,7 +190,10 @@ TEST(IndexSpecValidateTest, ReturnsAnErrorIfNamespaceDoesNotMatch) {
     // Verify that we reject the index specification when the "ns" field only contains the
     // collection name.
     ASSERT_EQ(ErrorCodes::BadValue,
-              validateIndexSpec(BSON("key" << BSON("field" << 1) << "ns" << kTestNamespace.coll()),
+              validateIndexSpec(BSON("key" << BSON("field" << 1) << "name"
+                                           << "indexName"
+                                           << "ns"
+                                           << kTestNamespace.coll()),
                                 kTestNamespace,
                                 featureCompatibility));
 }
@@ -149,14 +203,22 @@ TEST(IndexSpecValidateTest, ReturnsIndexSpecWithNamespaceFilledInIfItIsNotPresen
     featureCompatibility.version.store(ServerGlobalParams::FeatureCompatibility::Version::k34);
     featureCompatibility.validateFeaturesAsMaster.store(true);
 
-    auto result = validateIndexSpec(
-        BSON("key" << BSON("field" << 1) << "v" << 1), kTestNamespace, featureCompatibility);
+    auto result = validateIndexSpec(BSON("key" << BSON("field" << 1) << "name"
+                                               << "indexName"
+                                               << "v"
+                                               << 1),
+                                    kTestNamespace,
+                                    featureCompatibility);
     ASSERT_OK(result.getStatus());
 
     // We don't care about the order of the fields in the resulting index specification.
-    ASSERT_BSONOBJ_EQ(
-        sorted(BSON("key" << BSON("field" << 1) << "ns" << kTestNamespace.ns() << "v" << 1)),
-        sorted(result.getValue()));
+    ASSERT_BSONOBJ_EQ(sorted(BSON("key" << BSON("field" << 1) << "name"
+                                        << "indexName"
+                                        << "ns"
+                                        << kTestNamespace.ns()
+                                        << "v"
+                                        << 1)),
+                      sorted(result.getValue()));
 
     // Verify that the index specification we returned is still considered valid.
     ASSERT_OK(validateIndexSpec(result.getValue(), kTestNamespace, featureCompatibility));
@@ -167,14 +229,20 @@ TEST(IndexSpecValidateTest, ReturnsIndexSpecUnchangedIfNamespaceAndVersionArePre
     featureCompatibility.version.store(ServerGlobalParams::FeatureCompatibility::Version::k34);
     featureCompatibility.validateFeaturesAsMaster.store(true);
 
-    auto result = validateIndexSpec(
-        BSON("key" << BSON("field" << 1) << "ns" << kTestNamespace.ns() << "v" << 1),
-        kTestNamespace,
-        featureCompatibility);
+    auto result = validateIndexSpec(BSON("key" << BSON("field" << 1) << "name"
+                                               << "indexName"
+                                               << "ns"
+                                               << kTestNamespace.ns()
+                                               << "v"
+                                               << 1),
+                                    kTestNamespace,
+                                    featureCompatibility);
     ASSERT_OK(result.getStatus());
 
     // We don't care about the order of the fields in the resulting index specification.
-    ASSERT_BSONOBJ_EQ(sorted(BSON("key" << BSON("field" << 1) << "ns"
+    ASSERT_BSONOBJ_EQ(sorted(BSON("key" << BSON("field" << 1) << "name"
+                                        << "indexName"
+                                        << "ns"
                                         << "test.index_spec_validate"
                                         << "v"
                                         << 1)),
@@ -187,12 +255,17 @@ TEST(IndexSpecValidateTest, ReturnsAnErrorIfVersionIsNotANumber) {
     featureCompatibility.validateFeaturesAsMaster.store(true);
 
     ASSERT_EQ(ErrorCodes::TypeMismatch,
-              validateIndexSpec(BSON("key" << BSON("field" << 1) << "v"
+              validateIndexSpec(BSON("key" << BSON("field" << 1) << "name"
+                                           << "indexName"
+                                           << "v"
                                            << "not a number"),
                                 kTestNamespace,
                                 featureCompatibility));
     ASSERT_EQ(ErrorCodes::TypeMismatch,
-              validateIndexSpec(BSON("key" << BSON("field" << 1) << "v" << BSONObj()),
+              validateIndexSpec(BSON("key" << BSON("field" << 1) << "name"
+                                           << "indexName"
+                                           << "v"
+                                           << BSONObj()),
                                 kTestNamespace,
                                 featureCompatibility));
 }
@@ -203,23 +276,33 @@ TEST(IndexSpecValidateTest, ReturnsAnErrorIfVersionIsNotRepresentableAsInt) {
     featureCompatibility.validateFeaturesAsMaster.store(true);
 
     ASSERT_EQ(ErrorCodes::BadValue,
-              validateIndexSpec(BSON("key" << BSON("field" << 1) << "v" << 2.2),
+              validateIndexSpec(BSON("key" << BSON("field" << 1) << "name"
+                                           << "indexName"
+                                           << "v"
+                                           << 2.2),
                                 kTestNamespace,
                                 featureCompatibility));
     ASSERT_EQ(ErrorCodes::BadValue,
-              validateIndexSpec(BSON("key" << BSON("field" << 1) << "v" << std::nan("1")),
+              validateIndexSpec(BSON("key" << BSON("field" << 1) << "name"
+                                           << "indexName"
+                                           << "v"
+                                           << std::nan("1")),
                                 kTestNamespace,
                                 featureCompatibility));
     ASSERT_EQ(ErrorCodes::BadValue,
-              validateIndexSpec(BSON("key" << BSON("field" << 1) << "v"
+              validateIndexSpec(BSON("key" << BSON("field" << 1) << "name"
+                                           << "indexName"
+                                           << "v"
                                            << std::numeric_limits<double>::infinity()),
                                 kTestNamespace,
                                 featureCompatibility));
     ASSERT_EQ(ErrorCodes::BadValue,
-              validateIndexSpec(
-                  BSON("key" << BSON("field" << 1) << "v" << std::numeric_limits<long long>::max()),
-                  kTestNamespace,
-                  featureCompatibility));
+              validateIndexSpec(BSON("key" << BSON("field" << 1) << "name"
+                                           << "indexName"
+                                           << "v"
+                                           << std::numeric_limits<long long>::max()),
+                                kTestNamespace,
+                                featureCompatibility));
 }
 
 TEST(IndexSpecValidateTest, ReturnsAnErrorIfVersionIsV0) {
@@ -228,7 +311,10 @@ TEST(IndexSpecValidateTest, ReturnsAnErrorIfVersionIsV0) {
     featureCompatibility.validateFeaturesAsMaster.store(true);
 
     ASSERT_EQ(ErrorCodes::CannotCreateIndex,
-              validateIndexSpec(BSON("key" << BSON("field" << 1) << "v" << 0),
+              validateIndexSpec(BSON("key" << BSON("field" << 1) << "name"
+                                           << "indexName"
+                                           << "v"
+                                           << 0),
                                 kTestNamespace,
                                 featureCompatibility));
 }
@@ -239,14 +325,21 @@ TEST(IndexSpecValidateTest, ReturnsAnErrorIfVersionIsUnsupported) {
     featureCompatibility.validateFeaturesAsMaster.store(true);
 
     ASSERT_EQ(ErrorCodes::CannotCreateIndex,
-              validateIndexSpec(
-                  BSON("key" << BSON("field" << 1) << "v" << 3 << "collation" << BSON("locale"
-                                                                                      << "en")),
-                  kTestNamespace,
-                  featureCompatibility));
+              validateIndexSpec(BSON("key" << BSON("field" << 1) << "name"
+                                           << "indexName"
+                                           << "v"
+                                           << 3
+                                           << "collation"
+                                           << BSON("locale"
+                                                   << "en")),
+                                kTestNamespace,
+                                featureCompatibility));
 
     ASSERT_EQ(ErrorCodes::CannotCreateIndex,
-              validateIndexSpec(BSON("key" << BSON("field" << 1) << "v" << -3LL),
+              validateIndexSpec(BSON("key" << BSON("field" << 1) << "name"
+                                           << "indexName"
+                                           << "v"
+                                           << -3LL),
                                 kTestNamespace,
                                 featureCompatibility));
 }
@@ -257,7 +350,10 @@ TEST(IndexSpecValidateTest, ReturnsAnErrorIfIndexVersionIsV2AndFeatureCompatibil
     featureCompatibility.validateFeaturesAsMaster.store(true);
 
     ASSERT_EQ(ErrorCodes::CannotCreateIndex,
-              validateIndexSpec(BSON("key" << BSON("field" << 1) << "v" << 2),
+              validateIndexSpec(BSON("key" << BSON("field" << 1) << "name"
+                                           << "indexName"
+                                           << "v"
+                                           << 2),
                                 kTestNamespace,
                                 featureCompatibility));
 }
@@ -267,14 +363,22 @@ TEST(IndexSpecValidateTest, AcceptsIndexVersionV2WhenValidateFeaturesAsMasterFal
     featureCompatibility.version.store(ServerGlobalParams::FeatureCompatibility::Version::k32);
     featureCompatibility.validateFeaturesAsMaster.store(false);
 
-    auto result = validateIndexSpec(
-        BSON("key" << BSON("field" << 1) << "v" << 2), kTestNamespace, featureCompatibility);
+    auto result = validateIndexSpec(BSON("key" << BSON("field" << 1) << "name"
+                                               << "indexName"
+                                               << "v"
+                                               << 2),
+                                    kTestNamespace,
+                                    featureCompatibility);
     ASSERT_OK(result.getStatus());
 
     // We don't care about the order of the fields in the resulting index specification.
-    ASSERT_BSONOBJ_EQ(
-        sorted(BSON("key" << BSON("field" << 1) << "ns" << kTestNamespace.ns() << "v" << 2)),
-        sorted(result.getValue()));
+    ASSERT_BSONOBJ_EQ(sorted(BSON("key" << BSON("field" << 1) << "name"
+                                        << "indexName"
+                                        << "ns"
+                                        << kTestNamespace.ns()
+                                        << "v"
+                                        << 2)),
+                      sorted(result.getValue()));
 }
 
 TEST(IndexSpecValidateTest, AcceptsIndexVersionsThatAreAllowedForCreation) {
@@ -282,23 +386,39 @@ TEST(IndexSpecValidateTest, AcceptsIndexVersionsThatAreAllowedForCreation) {
     featureCompatibility.version.store(ServerGlobalParams::FeatureCompatibility::Version::k34);
     featureCompatibility.validateFeaturesAsMaster.store(true);
 
-    auto result = validateIndexSpec(
-        BSON("key" << BSON("field" << 1) << "v" << 1), kTestNamespace, featureCompatibility);
+    auto result = validateIndexSpec(BSON("key" << BSON("field" << 1) << "name"
+                                               << "indexName"
+                                               << "v"
+                                               << 1),
+                                    kTestNamespace,
+                                    featureCompatibility);
     ASSERT_OK(result.getStatus());
 
     // We don't care about the order of the fields in the resulting index specification.
-    ASSERT_BSONOBJ_EQ(
-        sorted(BSON("key" << BSON("field" << 1) << "ns" << kTestNamespace.ns() << "v" << 1)),
-        sorted(result.getValue()));
+    ASSERT_BSONOBJ_EQ(sorted(BSON("key" << BSON("field" << 1) << "name"
+                                        << "indexName"
+                                        << "ns"
+                                        << kTestNamespace.ns()
+                                        << "v"
+                                        << 1)),
+                      sorted(result.getValue()));
 
-    result = validateIndexSpec(
-        BSON("key" << BSON("field" << 1) << "v" << 2LL), kTestNamespace, featureCompatibility);
+    result = validateIndexSpec(BSON("key" << BSON("field" << 1) << "name"
+                                          << "indexName"
+                                          << "v"
+                                          << 2LL),
+                               kTestNamespace,
+                               featureCompatibility);
     ASSERT_OK(result.getStatus());
 
     // We don't care about the order of the fields in the resulting index specification.
-    ASSERT_BSONOBJ_EQ(
-        sorted(BSON("key" << BSON("field" << 1) << "ns" << kTestNamespace.ns() << "v" << 2LL)),
-        sorted(result.getValue()));
+    ASSERT_BSONOBJ_EQ(sorted(BSON("key" << BSON("field" << 1) << "name"
+                                        << "indexName"
+                                        << "ns"
+                                        << kTestNamespace.ns()
+                                        << "v"
+                                        << 2LL)),
+                      sorted(result.getValue()));
 }
 
 TEST(IndexSpecValidateTest, DefaultIndexVersionIsV1IfFeatureCompatibilityVersionIs32) {
@@ -306,16 +426,22 @@ TEST(IndexSpecValidateTest, DefaultIndexVersionIsV1IfFeatureCompatibilityVersion
     featureCompatibility.version.store(ServerGlobalParams::FeatureCompatibility::Version::k32);
     featureCompatibility.validateFeaturesAsMaster.store(true);
 
-    auto result =
-        validateIndexSpec(BSON("key" << BSON("field" << 1) << "ns" << kTestNamespace.ns()),
-                          kTestNamespace,
-                          featureCompatibility);
+    auto result = validateIndexSpec(BSON("key" << BSON("field" << 1) << "name"
+                                               << "indexName"
+                                               << "ns"
+                                               << kTestNamespace.ns()),
+                                    kTestNamespace,
+                                    featureCompatibility);
     ASSERT_OK(result.getStatus());
 
     // We don't care about the order of the fields in the resulting index specification.
-    ASSERT_BSONOBJ_EQ(
-        sorted(BSON("key" << BSON("field" << 1) << "ns" << kTestNamespace.ns() << "v" << 1)),
-        sorted(result.getValue()));
+    ASSERT_BSONOBJ_EQ(sorted(BSON("key" << BSON("field" << 1) << "name"
+                                        << "indexName"
+                                        << "ns"
+                                        << kTestNamespace.ns()
+                                        << "v"
+                                        << 1)),
+                      sorted(result.getValue()));
 
     // Verify that the index specification we returned is still considered valid.
     ASSERT_OK(validateIndexSpec(result.getValue(), kTestNamespace, featureCompatibility));
@@ -327,16 +453,22 @@ TEST(IndexSpecValidateTest,
     featureCompatibility.version.store(ServerGlobalParams::FeatureCompatibility::Version::k32);
     featureCompatibility.validateFeaturesAsMaster.store(false);
 
-    auto result =
-        validateIndexSpec(BSON("key" << BSON("field" << 1) << "ns" << kTestNamespace.ns()),
-                          kTestNamespace,
-                          featureCompatibility);
+    auto result = validateIndexSpec(BSON("key" << BSON("field" << 1) << "name"
+                                               << "indexName"
+                                               << "ns"
+                                               << kTestNamespace.ns()),
+                                    kTestNamespace,
+                                    featureCompatibility);
     ASSERT_OK(result.getStatus());
 
     // We don't care about the order of the fields in the resulting index specification.
-    ASSERT_BSONOBJ_EQ(
-        sorted(BSON("key" << BSON("field" << 1) << "ns" << kTestNamespace.ns() << "v" << 1)),
-        sorted(result.getValue()));
+    ASSERT_BSONOBJ_EQ(sorted(BSON("key" << BSON("field" << 1) << "name"
+                                        << "indexName"
+                                        << "ns"
+                                        << kTestNamespace.ns()
+                                        << "v"
+                                        << 1)),
+                      sorted(result.getValue()));
 
     // Verify that the index specification we returned is still considered valid.
     ASSERT_OK(validateIndexSpec(result.getValue(), kTestNamespace, featureCompatibility));
@@ -347,16 +479,22 @@ TEST(IndexSpecValidateTest, DefaultIndexVersionIsV2IfFeatureCompatibilityVersion
     featureCompatibility.version.store(ServerGlobalParams::FeatureCompatibility::Version::k34);
     featureCompatibility.validateFeaturesAsMaster.store(true);
 
-    auto result =
-        validateIndexSpec(BSON("key" << BSON("field" << 1) << "ns" << kTestNamespace.ns()),
-                          kTestNamespace,
-                          featureCompatibility);
+    auto result = validateIndexSpec(BSON("key" << BSON("field" << 1) << "name"
+                                               << "indexName"
+                                               << "ns"
+                                               << kTestNamespace.ns()),
+                                    kTestNamespace,
+                                    featureCompatibility);
     ASSERT_OK(result.getStatus());
 
     // We don't care about the order of the fields in the resulting index specification.
-    ASSERT_BSONOBJ_EQ(
-        sorted(BSON("key" << BSON("field" << 1) << "ns" << kTestNamespace.ns() << "v" << 2)),
-        sorted(result.getValue()));
+    ASSERT_BSONOBJ_EQ(sorted(BSON("key" << BSON("field" << 1) << "name"
+                                        << "indexName"
+                                        << "ns"
+                                        << kTestNamespace.ns()
+                                        << "v"
+                                        << 2)),
+                      sorted(result.getValue()));
 
     // Verify that the index specification we returned is still considered valid.
     ASSERT_OK(validateIndexSpec(result.getValue(), kTestNamespace, featureCompatibility));
@@ -367,14 +505,22 @@ TEST(IndexSpecValidateTest, AcceptsIndexVersionV1WhenFeatureCompatibilityVersion
     featureCompatibility.version.store(ServerGlobalParams::FeatureCompatibility::Version::k34);
     featureCompatibility.validateFeaturesAsMaster.store(true);
 
-    auto result = validateIndexSpec(
-        BSON("key" << BSON("field" << 1) << "v" << 1), kTestNamespace, featureCompatibility);
+    auto result = validateIndexSpec(BSON("key" << BSON("field" << 1) << "name"
+                                               << "indexName"
+                                               << "v"
+                                               << 1),
+                                    kTestNamespace,
+                                    featureCompatibility);
     ASSERT_OK(result.getStatus());
 
     // We don't care about the order of the fields in the resulting index specification.
-    ASSERT_BSONOBJ_EQ(
-        sorted(BSON("key" << BSON("field" << 1) << "ns" << kTestNamespace.ns() << "v" << 1)),
-        sorted(result.getValue()));
+    ASSERT_BSONOBJ_EQ(sorted(BSON("key" << BSON("field" << 1) << "name"
+                                        << "indexName"
+                                        << "ns"
+                                        << kTestNamespace.ns()
+                                        << "v"
+                                        << 1)),
+                      sorted(result.getValue()));
 }
 
 TEST(IndexSpecValidateTest, ReturnsAnErrorIfCollationIsNotAnObject) {
@@ -383,77 +529,107 @@ TEST(IndexSpecValidateTest, ReturnsAnErrorIfCollationIsNotAnObject) {
     featureCompatibility.validateFeaturesAsMaster.store(true);
 
     ASSERT_EQ(ErrorCodes::TypeMismatch,
-              validateIndexSpec(BSON("key" << BSON("field" << 1) << "collation" << 1),
+              validateIndexSpec(BSON("key" << BSON("field" << 1) << "name"
+                                           << "indexName"
+                                           << "collation"
+                                           << 1),
                                 kTestNamespace,
                                 featureCompatibility));
     ASSERT_EQ(ErrorCodes::TypeMismatch,
-              validateIndexSpec(BSON("key" << BSON("field" << 1) << "collation"
+              validateIndexSpec(BSON("key" << BSON("field" << 1) << "name"
+                                           << "indexName"
+                                           << "collation"
                                            << "not an object"),
                                 kTestNamespace,
                                 featureCompatibility));
     ASSERT_EQ(ErrorCodes::TypeMismatch,
-              validateIndexSpec(BSON("key" << BSON("field" << 1) << "collation" << BSONArray()),
+              validateIndexSpec(BSON("key" << BSON("field" << 1) << "name"
+                                           << "indexName"
+                                           << "collation"
+                                           << BSONArray()),
                                 kTestNamespace,
                                 featureCompatibility));
 }
 
-TEST(IndexSpecValidateTest, ReturnsAnErrorIfCollationIsPresentAndVersionIsLessThanV2) {
+TEST(IndexSpecValidateTest, ReturnsAnErrorIfCollationIsEmpty) {
     ServerGlobalParams::FeatureCompatibility featureCompatibility;
     featureCompatibility.version.store(ServerGlobalParams::FeatureCompatibility::Version::k34);
     featureCompatibility.validateFeaturesAsMaster.store(true);
 
-    ASSERT_EQ(
-        ErrorCodes::CannotCreateIndex,
-        validateIndexSpec(BSON("key" << BSON("field" << 1) << "collation" << BSONObj() << "v" << 1),
-                          kTestNamespace,
-                          featureCompatibility));
+    ASSERT_EQ(ErrorCodes::BadValue,
+              validateIndexSpec(BSON("key" << BSON("field" << 1) << "name"
+                                           << "indexName"
+                                           << "collation"
+                                           << BSONObj()),
+                                kTestNamespace,
+                                featureCompatibility));
 }
 
-TEST(IndexSpecValidateTest, AcceptsAnyObjectValueForCollation) {
+TEST(IndexSpecValidateTest, ReturnsAnErrorIfCollationIsPresentAndVersionIsLessThanV2) {
     ServerGlobalParams::FeatureCompatibility featureCompatibility;
     featureCompatibility.version.store(ServerGlobalParams::FeatureCompatibility::Version::k34);
     featureCompatibility.validateFeaturesAsMaster.store(true);
 
-    auto result = validateIndexSpec(
-        BSON("key" << BSON("field" << 1) << "v" << 2 << "collation" << BSON("locale"
-                                                                            << "simple")),
-        kTestNamespace,
-        featureCompatibility);
-    ASSERT_OK(result.getStatus());
+    ASSERT_EQ(ErrorCodes::CannotCreateIndex,
+              validateIndexSpec(BSON("key" << BSON("field" << 1) << "name"
+                                           << "indexName"
+                                           << "collation"
+                                           << BSON("locale"
+                                                   << "simple")
+                                           << "v"
+                                           << 1),
+                                kTestNamespace,
+                                featureCompatibility));
+}
 
-    // We don't care about the order of the fields in the resulting index specification.
-    ASSERT_BSONOBJ_EQ(
-        sorted(BSON("key" << BSON("field" << 1) << "ns" << kTestNamespace.ns() << "v" << 2
-                          << "collation"
-                          << BSON("locale"
-                                  << "simple"))),
-        sorted(result.getValue()));
-
-    result =
-        validateIndexSpec(BSON("key" << BSON("field" << 1) << "v" << 2 << "collation" << BSONObj()),
-                          kTestNamespace,
-                          featureCompatibility);
+TEST(IndexSpecValidateTest, AcceptsAnyNonEmptyObjectValueForCollation) {
+    ServerGlobalParams::FeatureCompatibility featureCompatibility;
+    featureCompatibility.version.store(ServerGlobalParams::FeatureCompatibility::Version::k34);
+    featureCompatibility.validateFeaturesAsMaster.store(true);
+
+    auto result = validateIndexSpec(BSON("key" << BSON("field" << 1) << "name"
+                                               << "indexName"
+                                               << "v"
+                                               << 2
+                                               << "collation"
+                                               << BSON("locale"
+                                                       << "simple")),
+                                    kTestNamespace,
+                                    featureCompatibility);
     ASSERT_OK(result.getStatus());
 
     // We don't care about the order of the fields in the resulting index specification.
-    ASSERT_BSONOBJ_EQ(
-        sorted(BSON("key" << BSON("field" << 1) << "ns" << kTestNamespace.ns() << "v" << 2
-                          << "collation"
-                          << BSONObj())),
-        sorted(result.getValue()));
+    ASSERT_BSONOBJ_EQ(sorted(BSON("key" << BSON("field" << 1) << "name"
+                                        << "indexName"
+                                        << "ns"
+                                        << kTestNamespace.ns()
+                                        << "v"
+                                        << 2
+                                        << "collation"
+                                        << BSON("locale"
+                                                << "simple"))),
+                      sorted(result.getValue()));
 
-    result = validateIndexSpec(BSON("key" << BSON("field" << 1) << "v" << 2 << "collation"
+    result = validateIndexSpec(BSON("key" << BSON("field" << 1) << "name"
+                                          << "indexName"
+                                          << "v"
+                                          << 2
+                                          << "collation"
                                           << BSON("unknownCollationOption" << true)),
                                kTestNamespace,
                                featureCompatibility);
     ASSERT_OK(result.getStatus());
 
     // We don't care about the order of the fields in the resulting index specification.
-    ASSERT_BSONOBJ_EQ(
-        sorted(BSON("key" << BSON("field" << 1) << "ns" << kTestNamespace.ns() << "v" << 2
-                          << "collation"
-                          << BSON("unknownCollationOption" << true))),
-        sorted(result.getValue()));
+    ASSERT_BSONOBJ_EQ(sorted(BSON("key" << BSON("field" << 1) << "name"
+                                        << "indexName"
+                                        << "ns"
+                                        << kTestNamespace.ns()
+                                        << "v"
+                                        << 2
+                                        << "collation"
+                                        << BSON("unknownCollationOption" << true))),
+                      sorted(result.getValue()));
 }
 
 TEST(IndexSpecValidateTest, AcceptsIndexSpecIfCollationIsPresentAndVersionIsEqualToV2) {
@@ -461,20 +637,28 @@ TEST(IndexSpecValidateTest, AcceptsIndexSpecIfCollationIsPresentAndVersionIsEqua
     featureCompatibility.version.store(ServerGlobalParams::FeatureCompatibility::Version::k34);
     featureCompatibility.validateFeaturesAsMaster.store(true);
 
-    auto result = validateIndexSpec(
-        BSON("key" << BSON("field" << 1) << "v" << 2 << "collation" << BSON("locale"
-                                                                            << "en")),
-        kTestNamespace,
-        featureCompatibility);
+    auto result = validateIndexSpec(BSON("key" << BSON("field" << 1) << "name"
+                                               << "indexName"
+                                               << "v"
+                                               << 2
+                                               << "collation"
+                                               << BSON("locale"
+                                                       << "en")),
+                                    kTestNamespace,
+                                    featureCompatibility);
     ASSERT_OK(result.getStatus());
 
     // We don't care about the order of the fields in the resulting index specification.
-    ASSERT_BSONOBJ_EQ(
-        sorted(BSON("key" << BSON("field" << 1) << "ns" << kTestNamespace.ns() << "v" << 2
-                          << "collation"
-                          << BSON("locale"
-                                  << "en"))),
-        sorted(result.getValue()));
+    ASSERT_BSONOBJ_EQ(sorted(BSON("key" << BSON("field" << 1) << "name"
+                                        << "indexName"
+                                        << "ns"
+                                        << kTestNamespace.ns()
+                                        << "v"
+                                        << 2
+                                        << "collation"
+                                        << BSON("locale"
+                                                << "en"))),
+                      sorted(result.getValue()));
 }
 
 TEST(IndexSpecValidateTest, ReturnsAnErrorIfUnknownFieldIsPresentInSpecV2) {
@@ -482,10 +666,14 @@ TEST(IndexSpecValidateTest, ReturnsAnErrorIfUnknownFieldIsPresentInSpecV2) {
     featureCompatibility.version.store(ServerGlobalParams::FeatureCompatibility::Version::k34);
     featureCompatibility.validateFeaturesAsMaster.store(true);
 
-    auto result =
-        validateIndexSpec(BSON("key" << BSON("field" << 1) << "v" << 2 << "unknownField" << 1),
-                          kTestNamespace,
-                          featureCompatibility);
+    auto result = validateIndexSpec(BSON("key" << BSON("field" << 1) << "name"
+                                               << "indexName"
+                                               << "v"
+                                               << 2
+                                               << "unknownField"
+                                               << 1),
+                                    kTestNamespace,
+                                    featureCompatibility);
     ASSERT_EQ(ErrorCodes::InvalidIndexSpecificationOption, result);
 }
 
@@ -494,12 +682,247 @@ TEST(IndexSpecValidateTest, ReturnsAnErrorIfUnknownFieldIsPresentInSpecV1) {
     featureCompatibility.version.store(ServerGlobalParams::FeatureCompatibility::Version::k34);
     featureCompatibility.validateFeaturesAsMaster.store(true);
 
-    auto result =
-        validateIndexSpec(BSON("key" << BSON("field" << 1) << "v" << 1 << "unknownField" << 1),
-                          kTestNamespace,
-                          featureCompatibility);
+    auto result = validateIndexSpec(BSON("key" << BSON("field" << 1) << "name"
+                                               << "indexName"
+                                               << "v"
+                                               << 1
+                                               << "unknownField"
+                                               << 1),
+                                    kTestNamespace,
+                                    featureCompatibility);
     ASSERT_EQ(ErrorCodes::InvalidIndexSpecificationOption, result);
 }
 
+TEST(IdIndexSpecValidateTest, ReturnsAnErrorIfKeyPatternIsIncorrectForIdIndex) {
+    ASSERT_EQ(ErrorCodes::BadValue,
+              validateIdIndexSpec(BSON("key" << BSON("_id" << -1) << "name"
+                                             << "_id_"
+                                             << "ns"
+                                             << kTestNamespace.ns()
+                                             << "v"
+                                             << 2)));
+    ASSERT_EQ(ErrorCodes::BadValue,
+              validateIdIndexSpec(BSON("key" << BSON("a" << 1) << "name"
+                                             << "_id_"
+                                             << "ns"
+                                             << kTestNamespace.ns()
+                                             << "v"
+                                             << 2)));
+}
+
+TEST(IdIndexSpecValidateTest, ReturnsAnErrorIfNameIsIncorrectForIdIndex) {
+    ASSERT_EQ(ErrorCodes::BadValue,
+              validateIdIndexSpec(BSON("key" << BSON("_id" << 1) << "name"
+                                             << "indexName"
+                                             << "ns"
+                                             << kTestNamespace.ns()
+                                             << "v"
+                                             << 2)));
+}
+
+TEST(IdIndexSpecValidateTest, ReturnsOKStatusIfKeyAndNameCorrectForIdIndex) {
+    ASSERT_OK(validateIdIndexSpec(BSON("key" << BSON("_id" << 1) << "name"
+                                             << "_id_"
+                                             << "ns"
+                                             << kTestNamespace.ns()
+                                             << "v"
+                                             << 2)));
+}
+
+TEST(IdIndexSpecValidateTest, ReturnsAnErrorIfFieldNotAllowedForIdIndex) {
+    ASSERT_EQ(ErrorCodes::InvalidIndexSpecificationOption,
+              validateIdIndexSpec(BSON("key" << BSON("_id" << 1) << "name"
+                                             << "_id_"
+                                             << "ns"
+                                             << kTestNamespace.ns()
+                                             << "v"
+                                             << 2
+                                             << "background"
+                                             << false)));
+    ASSERT_EQ(ErrorCodes::InvalidIndexSpecificationOption,
+              validateIdIndexSpec(BSON("key" << BSON("_id" << 1) << "name"
+                                             << "_id_"
+                                             << "ns"
+                                             << kTestNamespace.ns()
+                                             << "v"
+                                             << 2
+                                             << "unique"
+                                             << true)));
+    ASSERT_EQ(ErrorCodes::InvalidIndexSpecificationOption,
+              validateIdIndexSpec(BSON("key" << BSON("_id" << 1) << "name"
+                                             << "_id_"
+                                             << "ns"
+                                             << kTestNamespace.ns()
+                                             << "v"
+                                             << 2
+                                             << "partialFilterExpression"
+                                             << BSON("a" << 5))));
+    ASSERT_EQ(ErrorCodes::InvalidIndexSpecificationOption,
+              validateIdIndexSpec(BSON("key" << BSON("_id" << 1) << "name"
+                                             << "_id_"
+                                             << "ns"
+                                             << kTestNamespace.ns()
+                                             << "v"
+                                             << 2
+                                             << "sparse"
+                                             << false)));
+    ASSERT_EQ(ErrorCodes::InvalidIndexSpecificationOption,
+              validateIdIndexSpec(BSON("key" << BSON("_id" << 1) << "name"
+                                             << "_id_"
+                                             << "ns"
+                                             << kTestNamespace.ns()
+                                             << "v"
+                                             << 2
+                                             << "expireAfterSeconds"
+                                             << 3600)));
+    ASSERT_EQ(ErrorCodes::InvalidIndexSpecificationOption,
+              validateIdIndexSpec(BSON("key" << BSON("_id" << 1) << "name"
+                                             << "_id_"
+                                             << "ns"
+                                             << kTestNamespace.ns()
+                                             << "v"
+                                             << 2
+                                             << "storageEngine"
+                                             << BSONObj())));
+}
+
+TEST(IdIndexSpecValidateTest, ReturnsOKStatusIfAllFieldsAllowedForIdIndex) {
+    ASSERT_OK(validateIdIndexSpec(BSON("key" << BSON("_id" << 1) << "name"
+                                             << "_id_"
+                                             << "ns"
+                                             << kTestNamespace.ns()
+                                             << "v"
+                                             << 2
+                                             << "collation"
+                                             << BSON("locale"
+                                                     << "simple"))));
+}
+
+TEST(IndexSpecCollationValidateTest, FillsInFullCollationSpec) {
+    QueryTestServiceContext serviceContext;
+    auto txn = serviceContext.makeOperationContext();
+
+    const CollatorInterface* defaultCollator = nullptr;
+
+    auto result = validateIndexSpecCollation(txn.get(),
+                                             BSON("key" << BSON("field" << 1) << "name"
+                                                        << "indexName"
+                                                        << "ns"
+                                                        << kTestNamespace.ns()
+                                                        << "v"
+                                                        << 2
+                                                        << "collation"
+                                                        << BSON("locale"
+                                                                << "mock_reverse_string")),
+                                             defaultCollator);
+    ASSERT_OK(result.getStatus());
+
+    // We don't care about the order of the fields in the resulting index specification.
+    ASSERT_BSONOBJ_EQ(sorted(BSON("key" << BSON("field" << 1) << "name"
+                                        << "indexName"
+                                        << "ns"
+                                        << kTestNamespace.ns()
+                                        << "v"
+                                        << 2
+                                        << "collation"
+                                        << BSON("locale"
+                                                << "mock_reverse_string"
+                                                << "caseLevel"
+                                                << false
+                                                << "caseFirst"
+                                                << "off"
+                                                << "strength"
+                                                << 3
+                                                << "numericOrdering"
+                                                << false
+                                                << "alternate"
+                                                << "non-ignorable"
+                                                << "maxVariable"
+                                                << "punct"
+                                                << "normalization"
+                                                << false
+                                                << "backwards"
+                                                << false
+                                                << "version"
+                                                << "mock_version"))),
+                      sorted(result.getValue()));
+}
+
+TEST(IndexSpecCollationValidateTest, RemovesCollationFieldIfSimple) {
+    QueryTestServiceContext serviceContext;
+    auto txn = serviceContext.makeOperationContext();
+
+    const CollatorInterface* defaultCollator = nullptr;
+
+    auto result = validateIndexSpecCollation(txn.get(),
+                                             BSON("key" << BSON("field" << 1) << "name"
+                                                        << "indexName"
+                                                        << "ns"
+                                                        << kTestNamespace.ns()
+                                                        << "v"
+                                                        << 2
+                                                        << "collation"
+                                                        << BSON("locale"
+                                                                << "simple")),
+                                             defaultCollator);
+    ASSERT_OK(result.getStatus());
+
+    // We don't care about the order of the fields in the resulting index specification.
+    ASSERT_BSONOBJ_EQ(sorted(BSON("key" << BSON("field" << 1) << "name"
+                                        << "indexName"
+                                        << "ns"
+                                        << kTestNamespace.ns()
+                                        << "v"
+                                        << 2)),
+                      sorted(result.getValue()));
+}
+
+TEST(IndexSpecCollationValidateTest, FillsInCollationFieldWithCollectionDefaultIfNotPresent) {
+    QueryTestServiceContext serviceContext;
+    auto txn = serviceContext.makeOperationContext();
+
+    const CollatorInterfaceMock defaultCollator(CollatorInterfaceMock::MockType::kReverseString);
+
+    auto result = validateIndexSpecCollation(txn.get(),
+                                             BSON("key" << BSON("field" << 1) << "name"
+                                                        << "indexName"
+                                                        << "ns"
+                                                        << kTestNamespace.ns()
+                                                        << "v"
+                                                        << 2),
+                                             &defaultCollator);
+    ASSERT_OK(result.getStatus());
+
+    // We don't care about the order of the fields in the resulting index specification.
+    ASSERT_BSONOBJ_EQ(sorted(BSON("key" << BSON("field" << 1) << "name"
+                                        << "indexName"
+                                        << "ns"
+                                        << kTestNamespace.ns()
+                                        << "v"
+                                        << 2
+                                        << "collation"
+                                        << BSON("locale"
+                                                << "mock_reverse_string"
+                                                << "caseLevel"
+                                                << false
+                                                << "caseFirst"
+                                                << "off"
+                                                << "strength"
+                                                << 3
+                                                << "numericOrdering"
+                                                << false
+                                                << "alternate"
+                                                << "non-ignorable"
+                                                << "maxVariable"
+                                                << "punct"
+                                                << "normalization"
+                                                << false
+                                                << "backwards"
+                                                << false
+                                                << "version"
+                                                << "mock_version"))),
+                      sorted(result.getValue()));
+}
+
 }  // namespace
 }  // namespace mongo
diff --git a/src/mongo/db/commands/create_indexes.cpp b/src/mongo/db/commands/create_indexes.cpp
index c1b5343524b..fbd131211b4 100644
--- a/src/mongo/db/commands/create_indexes.cpp
+++ b/src/mongo/db/commands/create_indexes.cpp
@@ -45,13 +45,11 @@
 #include "mongo/db/index/index_descriptor.h"
 #include "mongo/db/op_observer.h"
 #include "mongo/db/ops/insert.h"
-#include "mongo/db/query/collation/collator_factory_interface.h"
 #include "mongo/db/repl/repl_client_info.h"
 #include "mongo/db/repl/replication_coordinator_global.h"
 #include "mongo/db/s/collection_metadata.h"
 #include "mongo/db/s/collection_sharding_state.h"
 #include "mongo/db/server_options.h"
-#include "mongo/db/service_context.h"
 #include "mongo/db/views/view_catalog.h"
 #include "mongo/s/shard_key_pattern.h"
 #include "mongo/util/scopeguard.h"
@@ -99,7 +97,8 @@ StatusWith<std::vector<BSONObj>> parseAndValidateIndexSpecs(
                                           << typeName(indexesElem.type())};
                 }
 
-                auto indexSpec = validateIndexSpec(indexesElem.Obj(), ns, featureCompatibility);
+                auto indexSpec = index_key_validate::validateIndexSpec(
+                    indexesElem.Obj(), ns, featureCompatibility);
                 if (!indexSpec.isOK()) {
                     return indexSpec.getStatus();
                 }
@@ -145,61 +144,12 @@ StatusWith<std::vector<BSONObj>> resolveCollectionDefaultProperties(
     std::vector<BSONObj> indexSpecsWithDefaults = std::move(indexSpecs);
 
     for (size_t i = 0, numIndexSpecs = indexSpecsWithDefaults.size(); i < numIndexSpecs; ++i) {
-        const BSONObj& indexSpec = indexSpecsWithDefaults[i];
-        if (auto collationElem = indexSpec[IndexDescriptor::kCollationFieldName]) {
-            // validateIndexSpec() should have already verified that 'collationElem' is an object.
-            invariant(collationElem.type() == BSONType::Object);
-
-            auto collator = CollatorFactoryInterface::get(txn->getServiceContext())
-                                ->makeFromBSON(collationElem.Obj());
-            if (!collator.isOK()) {
-                return collator.getStatus();
-            }
-
-            if (collator.getValue()) {
-                // If the collator factory returned a non-null collator, then inject the entire
-                // collation specification into the index specification. This is necessary to fill
-                // in any options that the user omitted.
-                BSONObjBuilder bob;
-
-                for (auto&& indexSpecElem : indexSpec) {
-                    if (IndexDescriptor::kCollationFieldName !=
-                        indexSpecElem.fieldNameStringData()) {
-                        bob.append(indexSpecElem);
-                    }
-                }
-                bob.append(IndexDescriptor::kCollationFieldName,
-                           collator.getValue()->getSpec().toBSON());
-
-                indexSpecsWithDefaults[i] = bob.obj();
-            } else {
-                // If the collator factory returned a null collator (representing the "simple"
-                // collation), then we simply omit the "collation" from the index specification.
-                // This is desirable to make the representation for the "simple" collation
-                // consistent between v=1 and v=2 indexes.
-                indexSpecsWithDefaults[i] =
-                    indexSpec.removeField(IndexDescriptor::kCollationFieldName);
-            }
-        } else if (collection->getDefaultCollator()) {
-            // validateIndexSpec() should have added the "v" field if it was not present and
-            // verified that 'versionElem' is a number.
-            auto versionElem = indexSpec[IndexDescriptor::kIndexVersionFieldName];
-            invariant(versionElem.isNumber());
-
-            if (IndexVersion::kV2 <= static_cast<IndexVersion>(versionElem.numberInt())) {
-                // The user did not specify an explicit collation for this index and the collection
-                // has a default collator. If we're building a v=2 index, then we should inherit the
-                // collection default. However, if we're building a v=1 index, then we're implicitly
-                // building an index that's using the "simple" collation.
-                BSONObjBuilder bob;
-
-                bob.appendElements(indexSpec);
-                bob.append(IndexDescriptor::kCollationFieldName,
-                           collection->getDefaultCollator()->getSpec().toBSON());
-
-                indexSpecsWithDefaults[i] = bob.obj();
-            }
+        auto validatedSpec = index_key_validate::validateIndexSpecCollation(
+            txn, indexSpecsWithDefaults[i], collection->getDefaultCollator());
+        if (!validatedSpec.isOK()) {
+            return validatedSpec.getStatus();
         }
+        indexSpecsWithDefaults[i] = validatedSpec.getValue();
     }
 
     return indexSpecsWithDefaults;
diff --git a/src/mongo/db/commands/dbcommands.cpp b/src/mongo/db/commands/dbcommands.cpp
index 81f03b9d24a..a4aa95e2298 100644
--- a/src/mongo/db/commands/dbcommands.cpp
+++ b/src/mongo/db/commands/dbcommands.cpp
@@ -55,6 +55,7 @@
 #include "mongo/db/catalog/create_collection.h"
 #include "mongo/db/catalog/drop_collection.h"
 #include "mongo/db/catalog/drop_database.h"
+#include "mongo/db/catalog/index_key_validate.h"
 #include "mongo/db/clientcursor.h"
 #include "mongo/db/commands.h"
 #include "mongo/db/commands/server_status.h"
@@ -77,6 +78,7 @@
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/op_observer.h"
 #include "mongo/db/ops/insert.h"
+#include "mongo/db/query/collation/collator_factory_interface.h"
 #include "mongo/db/query/get_executor.h"
 #include "mongo/db/query/internal_plans.h"
 #include "mongo/db/query/query_planner.h"
@@ -533,6 +535,8 @@ public:
                      int,
                      string& errmsg,
                      BSONObjBuilder& result) {
+        const NamespaceString ns(parseNs(dbname, cmdObj));
+
         if (cmdObj.hasField("autoIndexId")) {
             const char* deprecationWarning =
                 "the autoIndexId option is deprecated and will be removed in a future release";
@@ -553,7 +557,79 @@ public:
                  "http://dochub.mongodb.org/core/3.4-feature-compatibility."});
         }
 
-        return appendCommandStatus(result, createCollection(txn, dbname, cmdObj));
+        // Validate _id index spec and fill in missing fields.
+        if (auto idIndexElem = cmdObj["idIndex"]) {
+            if (cmdObj["viewOn"]) {
+                return appendCommandStatus(
+                    result,
+                    {ErrorCodes::InvalidOptions,
+                     str::stream() << "'idIndex' is not allowed with 'viewOn': " << idIndexElem});
+            }
+            if (cmdObj["autoIndexId"]) {
+                return appendCommandStatus(result,
+                                           {ErrorCodes::InvalidOptions,
+                                            str::stream()
+                                                << "'idIndex' is not allowed with 'autoIndexId': "
+                                                << idIndexElem});
+            }
+
+            if (idIndexElem.type() != BSONType::Object) {
+                return appendCommandStatus(
+                    result,
+                    {ErrorCodes::TypeMismatch,
+                     str::stream() << "'idIndex' has to be a document: " << idIndexElem});
+            }
+
+            auto idIndexSpec = idIndexElem.Obj();
+
+            // Perform index spec validation.
+            idIndexSpec = uassertStatusOK(index_key_validate::validateIndexSpec(
+                idIndexSpec, ns, serverGlobalParams.featureCompatibility));
+            uassertStatusOK(index_key_validate::validateIdIndexSpec(idIndexSpec));
+
+            // Validate or fill in _id index collation.
+            std::unique_ptr<CollatorInterface> defaultCollator;
+            if (auto collationElem = cmdObj["collation"]) {
+                if (collationElem.type() != BSONType::Object) {
+                    return appendCommandStatus(
+                        result,
+                        {ErrorCodes::TypeMismatch,
+                         str::stream() << "'collation' has to be a document: " << collationElem});
+                }
+                auto collatorStatus = CollatorFactoryInterface::get(txn->getServiceContext())
+                                          ->makeFromBSON(collationElem.Obj());
+                if (!collatorStatus.isOK()) {
+                    return appendCommandStatus(result, collatorStatus.getStatus());
+                }
+                defaultCollator = std::move(collatorStatus.getValue());
+            }
+            idIndexSpec = uassertStatusOK(index_key_validate::validateIndexSpecCollation(
+                txn, idIndexSpec, defaultCollator.get()));
+            std::unique_ptr<CollatorInterface> idIndexCollator;
+            if (auto collationElem = idIndexSpec["collation"]) {
+                auto collatorStatus = CollatorFactoryInterface::get(txn->getServiceContext())
+                                          ->makeFromBSON(collationElem.Obj());
+                // validateIndexSpecCollation() should have checked that the _id index collation
+                // spec is valid.
+                invariant(collatorStatus.isOK());
+                idIndexCollator = std::move(collatorStatus.getValue());
+            }
+            if (!CollatorInterface::collatorsMatch(defaultCollator.get(), idIndexCollator.get())) {
+                return appendCommandStatus(
+                    result,
+                    {ErrorCodes::BadValue,
+                     "'idIndex' must have the same collation as the collection."});
+            }
+
+            // Remove "idIndex" field from command.
+            auto resolvedCmdObj = cmdObj.removeField("idIndex");
+
+            return appendCommandStatus(result,
+                                       createCollection(txn, dbname, resolvedCmdObj, idIndexSpec));
+        }
+
+        BSONObj idIndexSpec;
+        return appendCommandStatus(result, createCollection(txn, dbname, cmdObj, idIndexSpec));
     }
 } cmdCreate;
 
diff --git a/src/mongo/db/commands/drop_indexes.cpp b/src/mongo/db/commands/drop_indexes.cpp
index 32e402d6782..9c4a4b7c1a2 100644
--- a/src/mongo/db/commands/drop_indexes.cpp
+++ b/src/mongo/db/commands/drop_indexes.cpp
@@ -178,7 +178,8 @@ public:
                 }
 
                 const BSONObj key = spec.getObjectField("key");
-                const Status keyStatus = validateKeyPattern(key, defaultIndexVersion);
+                const Status keyStatus =
+                    index_key_validate::validateKeyPattern(key, defaultIndexVersion);
                 if (!keyStatus.isOK()) {
                     errmsg = str::stream()
                         << "Cannot rebuild index " << spec << ": " << keyStatus.reason()
diff --git a/src/mongo/db/repair_database.cpp b/src/mongo/db/repair_database.cpp
index dde257369eb..d688e7d5f44 100644
--- a/src/mongo/db/repair_database.cpp
+++ b/src/mongo/db/repair_database.cpp
@@ -102,7 +102,7 @@ Status rebuildIndexesOnCollection(OperationContext* txn,
             }
 
             const BSONObj key = spec.getObjectField("key");
-            const Status keyStatus = validateKeyPattern(key, newIndexVersion);
+            const Status keyStatus = index_key_validate::validateKeyPattern(key, newIndexVersion);
             if (!keyStatus.isOK()) {
                 return Status(
                     ErrorCodes::CannotCreateIndex,
diff --git a/src/mongo/shell/shell_utils.cpp b/src/mongo/shell/shell_utils.cpp
index 654333affdc..d165ae3e004 100644
--- a/src/mongo/shell/shell_utils.cpp
+++ b/src/mongo/shell/shell_utils.cpp
@@ -174,7 +174,8 @@ BSONObj validateIndexKey(const BSONObj& a, void* data) {
     BSONObj key = a[0].Obj();
     // This is related to old upgrade-checking code when v:1 indexes were the latest version, hence
     // always validate using v:1 rules here.
-    Status indexValid = validateKeyPattern(key, IndexDescriptor::IndexVersion::kV1);
+    Status indexValid =
+        index_key_validate::validateKeyPattern(key, IndexDescriptor::IndexVersion::kV1);
     if (!indexValid.isOK()) {
         return BSON("" << BSON("ok" << false << "type" << indexValid.codeString() << "errmsg"
                                     << indexValid.reason()));