SERVER-25023 Allow multiple indexes on the same fields with different partial index filters

29 files changed, 970 insertions, 220 deletions

diff --git a/jstests/core/index_partial_create_drop.js b/jstests/core/index_partial_create_drop.js
index f3021fccac5..56cfca8e8f8 100644
--- a/jstests/core/index_partial_create_drop.js
+++ b/jstests/core/index_partial_create_drop.js
@@ -29,52 +29,74 @@ var getNumKeys = function(idxName) {
 coll.drop();
 
 // Check bad filter spec on create.
-assert.commandFailed(coll.ensureIndex({x: 1}, {partialFilterExpression: 5}));
-assert.commandFailed(coll.ensureIndex({x: 1}, {partialFilterExpression: {x: {$asdasd: 3}}}));
-assert.commandFailed(coll.ensureIndex({x: 1}, {partialFilterExpression: {$and: 5}}));
-assert.commandFailed(coll.ensureIndex({x: 1}, {partialFilterExpression: {x: /abc/}}));
-assert.commandFailed(coll.ensureIndex({x: 1}, {
-    partialFilterExpression: {$and: [{$and: [{x: {$lt: 2}}, {x: {$gt: 0}}]}, {x: {$exists: true}}]}
-}));
+assert.commandFailed(coll.createIndex({x: 1}, {partialFilterExpression: 5}));
+assert.commandFailed(coll.createIndex({x: 1}, {partialFilterExpression: {x: {$asdasd: 3}}}));
+assert.commandFailed(coll.createIndex({x: 1}, {partialFilterExpression: {$and: 5}}));
+assert.commandFailed(coll.createIndex({x: 1}, {partialFilterExpression: {x: /abc/}}));
+
 // Use of $expr is banned in a partial index filter.
 assert.commandFailed(
     coll.createIndex({x: 1}, {partialFilterExpression: {$expr: {$eq: ["$x", 5]}}}));
 assert.commandFailed(coll.createIndex(
     {x: 1}, {partialFilterExpression: {$expr: {$eq: [{$trim: {input: "$x"}}, "hi"]}}}));
 
+// Only top-level $and is permitted, but by normalizing the input filter we absorb the child $and.
+assert.commandWorked(coll.createIndex({x: 1}, {
+    partialFilterExpression: {$and: [{$and: [{x: {$lt: 2}}, {x: {$gt: 0}}]}, {x: {$exists: true}}]}
+}));
+assert.commandWorked(coll.dropIndexes());
+
 for (var i = 0; i < 10; i++) {
     assert.commandWorked(coll.insert({x: i, a: i}));
 }
 
 // Create partial index.
-assert.commandWorked(coll.ensureIndex({x: 1}, {partialFilterExpression: {a: {$lt: 5}}}));
+assert.commandWorked(coll.createIndex({x: 1}, {partialFilterExpression: {a: {$lt: 5}}}));
 assert.eq(5, getNumKeys("x_1"));
 assert.commandWorked(coll.dropIndex({x: 1}));
 assert.eq(1, coll.getIndexes().length);
 
 // Create partial index in background.
 assert.commandWorked(
-    coll.ensureIndex({x: 1}, {background: true, partialFilterExpression: {a: {$lt: 5}}}));
+    coll.createIndex({x: 1}, {background: true, partialFilterExpression: {a: {$lt: 5}}}));
 assert.eq(5, getNumKeys("x_1"));
 assert.commandWorked(coll.dropIndex({x: 1}));
 assert.eq(1, coll.getIndexes().length);
 
 // Create complete index, same key as previous indexes.
-assert.commandWorked(coll.ensureIndex({x: 1}));
+assert.commandWorked(coll.createIndex({x: 1}));
 assert.eq(10, getNumKeys("x_1"));
 assert.commandWorked(coll.dropIndex({x: 1}));
 assert.eq(1, coll.getIndexes().length);
 
 // Partial indexes can't also be sparse indexes.
-assert.commandFailed(coll.ensureIndex({x: 1}, {partialFilterExpression: {a: 1}, sparse: true}));
-assert.commandFailed(coll.ensureIndex({x: 1}, {partialFilterExpression: {a: 1}, sparse: 1}));
-assert.commandWorked(coll.ensureIndex({x: 1}, {partialFilterExpression: {a: 1}, sparse: false}));
+assert.commandFailed(coll.createIndex({x: 1}, {partialFilterExpression: {a: 1}, sparse: true}));
+assert.commandFailed(coll.createIndex({x: 1}, {partialFilterExpression: {a: 1}, sparse: 1}));
+assert.commandWorked(coll.createIndex({x: 1}, {partialFilterExpression: {a: 1}, sparse: false}));
 assert.eq(2, coll.getIndexes().length);
 assert.commandWorked(coll.dropIndex({x: 1}));
 assert.eq(1, coll.getIndexes().length);
 
 // SERVER-18858: Verify that query compatible w/ partial index succeeds after index drop.
-assert.commandWorked(coll.ensureIndex({x: 1}, {partialFilterExpression: {a: {$lt: 5}}}));
+assert.commandWorked(coll.createIndex({x: 1}, {partialFilterExpression: {a: {$lt: 5}}}));
 assert.commandWorked(coll.dropIndex({x: 1}));
 assert.eq(1, coll.find({x: 0, a: 0}).itcount());
+
+// Can create multiple partial indexes on the same key pattern as long as the filter is different.
+assert.commandWorked(coll.dropIndexes());
+
+let numIndexesBefore = coll.getIndexes().length;
+assert.commandWorked(
+    coll.createIndex({x: 1}, {name: "partialIndex1", partialFilterExpression: {a: {$lt: 5}}}));
+assert.eq(coll.getIndexes().length, numIndexesBefore + 1);
+
+numIndexesBefore = coll.getIndexes().length;
+assert.commandWorked(
+    coll.createIndex({x: 1}, {name: "partialIndex2", partialFilterExpression: {a: {$gte: 5}}}));
+assert.eq(coll.getIndexes().length, numIndexesBefore + 1);
+
+numIndexesBefore = coll.getIndexes().length;
+assert.commandFailedWithCode(coll.dropIndex({x: 1}), ErrorCodes.AmbiguousIndexKeyPattern);
+assert.commandWorked(coll.dropIndex("partialIndex2"));
+assert.eq(coll.getIndexes().length, numIndexesBefore - 1);
 })();
diff --git a/jstests/core/index_partial_read_ops.js b/jstests/core/index_partial_read_ops.js
index 62ba040b3fb..daf32f2342d 100644
--- a/jstests/core/index_partial_read_ops.js
+++ b/jstests/core/index_partial_read_ops.js
@@ -13,7 +13,7 @@ var explain;
 var coll = db.index_partial_read_ops;
 coll.drop();
 
-assert.commandWorked(coll.ensureIndex({x: 1}, {partialFilterExpression: {a: {$lte: 1.5}}}));
+assert.commandWorked(coll.createIndex({x: 1}, {partialFilterExpression: {a: {$lte: 1.5}}}));
 assert.commandWorked(coll.insert({x: 5, a: 2}));  // Not in index.
 assert.commandWorked(coll.insert({x: 6, a: 1}));  // In index.
 
@@ -79,4 +79,46 @@ explain = coll.explain('executionStats')
               .findAndModify({query: {x: {$gt: 1}, a: 2}, update: {$inc: {x: 1}}});
 assert.eq(1, explain.executionStats.nReturned);
 assert(isCollscan(db, explain.queryPlanner.winningPlan));
+
+//
+// Verify functionality with multiple overlapping partial indexes on the same key pattern.
+//
+
+// Remove existing indexes and documents.
+assert.commandWorked(coll.dropIndexes());
+assert.commandWorked(coll.remove({}));
+
+assert.commandWorked(
+    coll.createIndex({a: 1}, {name: "index1", partialFilterExpression: {a: {$gte: 0}}}));
+assert.commandWorked(
+    coll.createIndex({a: 1}, {name: "index2", partialFilterExpression: {a: {$gte: 10}}}));
+assert.commandWorked(
+    coll.createIndex({a: 1}, {name: "index3", partialFilterExpression: {a: {$gte: 100}}}));
+
+assert.commandWorked(coll.insert([{a: 1}, {a: 2}, {a: 3}]));
+assert.commandWorked(coll.insert([{a: 11}, {a: 12}, {a: 13}]));
+assert.commandWorked(coll.insert([{a: 101}, {a: 102}, {a: 103}]));
+
+// Function which verifies that the given query is indexed, that it produces the same output as a
+// COLLSCAN and the given 'expectedResults' array, and that 'numAlternativePlans' were generated.
+function assertIndexedQueryAndResults(query, numAlternativePlans, expectedResults) {
+    const explainOut = coll.explain().find(query).finish();
+    const results = coll.find(query).toArray();
+    assert(isIxscan(db, explainOut), tojson(explainOut));
+    assert.eq(getRejectedPlans(explainOut).length, numAlternativePlans, tojson(explainOut));
+    assert.sameMembers(results, coll.find(query).hint({$natural: 1}).toArray());
+    assert.sameMembers(results.map(doc => (delete doc._id && doc)), expectedResults);
+}
+
+// Queries which fall within the covered ranges generate plans for all applicable partial indexes.
+assertIndexedQueryAndResults({a: {$gt: 0, $lt: 10}}, 0, [{a: 1}, {a: 2}, {a: 3}]);
+assertIndexedQueryAndResults({a: {$gt: 10, $lt: 100}}, 1, [{a: 11}, {a: 12}, {a: 13}]);
+assertIndexedQueryAndResults({a: {$gt: 100, $lt: 1000}}, 2, [{a: 101}, {a: 102}, {a: 103}]);
+assertIndexedQueryAndResults(
+    {a: {$gt: 0}},
+    0,
+    [{a: 1}, {a: 2}, {a: 3}, {a: 11}, {a: 12}, {a: 13}, {a: 101}, {a: 102}, {a: 103}]);
+
+// Queries which fall outside the range of any partial indexes produce a COLLSCAN.
+assert(isCollscan(db, coll.explain().find({a: {$lt: 0}}).finish()));
 })();
diff --git a/jstests/core/index_signature.js b/jstests/core/index_signature.js
new file mode 100644
index 00000000000..914b5a4dcdf
--- /dev/null
+++ b/jstests/core/index_signature.js
@@ -0,0 +1,133 @@
+/**
+ * Tests that indexes are distinguished by their "signature", i.e. the combination of parameters
+ * which uniquely identify an index. Multiple indexes can be created on the same key pattern if
+ * their signature parameters differ.
+ *
+ * TODO SERVER-46592: This test is multiversion-incompatible in 4.6. If we use 'requires_fcv_46'
+ *                    as the tag for that, removing 'requires_fcv_44' is sufficient. Otherwise,
+ *                    please set the appropriate tag when removing 'requires_fcv_44'
+ * @tags: [requires_fcv_44, requires_fcv_46, requires_non_retryable_writes]
+ */
+(function() {
+"use strict";
+
+const testDB = db.getSiblingDB(jsTestName());
+const coll = testDB.test;
+coll.drop();
+
+// The key pattern and spec against which other indexes will be compared during createIndexes.
+const initialIndexSpec = {
+    name: "initial_index",
+    collation: {locale: "en_US", strength: 1},
+    partialFilterExpression: {a: {$gt: 0, $lt: 10}, b: "blah"}
+};
+const keyPattern = {
+    a: 1
+};
+
+// Helper function to build an index spec based on 'initialIndexSpec'.
+function makeSpec(opts) {
+    return Object.assign({}, initialIndexSpec, opts || {});
+}
+
+// Verifies that the number of indexes changed in accordance with the 'expectedChange' argument.
+function assertNumIndexesAfterComparedToBefore(cmdRes, expectedChange) {
+    // In a sharded cluster, the results from all shards are returned in cmdRes.raw.
+    assert(cmdRes.numIndexesBefore != null || Object.values(cmdRes.raw), tojson(cmdRes));
+    const numIndexesBefore =
+        (cmdRes.numIndexesBefore != null ? cmdRes.numIndexesBefore
+                                         : Object.values(cmdRes.raw)[0].numIndexesBefore);
+    const numIndexesAfter =
+        (cmdRes.numIndexesAfter != null ? cmdRes.numIndexesAfter
+                                        : Object.values(cmdRes.raw)[0].numIndexesAfter);
+    assert.eq(numIndexesAfter, numIndexesBefore + expectedChange);
+}
+
+// Create an index on {a: 1} with an explicit collation and a partial filter expression.
+let cmdRes = assert.commandWorked(coll.createIndex(keyPattern, initialIndexSpec));
+assertNumIndexesAfterComparedToBefore(cmdRes, 1);
+
+// Verify that an index can be built on the same fields if the collation is different.
+cmdRes = assert.commandWorked(coll.createIndex(
+    keyPattern, makeSpec({name: "simple_collation_index", collation: {locale: "simple"}})));
+assertNumIndexesAfterComparedToBefore(cmdRes, 1);
+
+// Verify that an index can be built on the same fields if the partialFilterExpression is different.
+cmdRes = assert.commandWorked(
+    coll.createIndex(keyPattern, makeSpec({
+                         name: "partial_filter_index",
+                         partialFilterExpression: {a: {$gt: 5, $lt: 10}, b: "blah"}
+                     })));
+assertNumIndexesAfterComparedToBefore(cmdRes, 1);
+
+// Verify that partialFilterExpressions are normalized before being compared. Below, the expression
+// is written differently than in the previous index, but the two are considered equivalent. If we
+// attempt to build this index with the same name as the initial index, the operation will return
+// success but will not actually do any work, since the requested index already exists.
+const partialFilterDupeSpec =
+    makeSpec({partialFilterExpression: {$and: [{b: "blah"}, {a: {$lt: 10}}, {a: {$gt: 0}}]}});
+cmdRes = assert.commandWorked(coll.createIndex(keyPattern, partialFilterDupeSpec));
+assertNumIndexesAfterComparedToBefore(cmdRes, 0);
+
+// Verify that attempting to build the dupe index with a different name will result in an error.
+partialFilterDupeSpec.name = "partial_filter_dupe_index";
+assert.commandFailedWithCode(coll.createIndex(keyPattern, partialFilterDupeSpec),
+                             ErrorCodes.IndexOptionsConflict);
+
+// We don't currently take collation into account when checking partialFilterExpression equivalence.
+// In this instance we are using a case-insensitive collation, and so the predicate {b: "BLAH"} will
+// match the same set of documents as {b: "blah"} in the initial index's partialFilterExpression.
+// But we do not consider these filters equivalent, and so this is considered a distinct index.
+// TODO SERVER-47664: take collation into account in MatchExpression::equivalent().
+cmdRes = assert.commandWorked(
+    coll.createIndex(keyPattern, makeSpec({
+                         name: "partial_filter_collator_index",
+                         partialFilterExpression: {a: {$gt: 0, $lt: 10}, b: "BLAH"}
+                     })));
+assertNumIndexesAfterComparedToBefore(cmdRes, 1);
+
+// We do not currently sort MatchExpression trees by leaf predicate value in cases where two or more
+// branches are otherwise identical, meaning that we cannot identify certain trivial cases where two
+// partialFilterExpressions are equivalent.
+// TODO SERVER-47661: provide a way for MatchExpression trees to be sorted in a consistent manner in
+// cases where two or more otherwise identical branches differ only by leaf predicate value.
+const partialFilterUnsortedLeaves = makeSpec({
+    name: "partial_filter_single_field_multiple_predicates_same_matchtype",
+    partialFilterExpression: {$and: [{a: {$type: 1}}, {a: {$type: 2}}]}
+});
+cmdRes = assert.commandWorked(coll.createIndex(keyPattern, partialFilterUnsortedLeaves));
+assertNumIndexesAfterComparedToBefore(cmdRes, 1);
+
+// Change the predicate order of the $and and re-run the createIndex. We would expect this index to
+// be considered identical to the existing index, and for the createIndex to return no-op success.
+// Instead, we throw an exception because the catalog believes we are trying to create an index with
+// the same name but a different partialFilterExpression.
+partialFilterUnsortedLeaves.partialFilterExpression.$and.reverse();
+assert.commandFailedWithCode(coll.createIndex(keyPattern, partialFilterUnsortedLeaves),
+                             ErrorCodes.IndexKeySpecsConflict);
+
+// Verify that non-signature options cannot distinguish a new index from an existing index.
+// TODO SERVER-47657: unique and sparse should be part of the signature.
+const nonSignatureOptions =
+    [{unique: true}, {sparse: true}, {expireAfterSeconds: 10}, {background: true}];
+
+// Build a new, basic index on {a: 1}, since some of the options we intend to test are not
+// compatible with the partialFilterExpression on the existing {a: 1} indexes.
+cmdRes = assert.commandWorked(coll.createIndex(keyPattern, {name: "basic_index_default_opts"}));
+assertNumIndexesAfterComparedToBefore(cmdRes, 1);
+
+// Verify that none of the options in the list are sufficient to uniquely identify an index, meaning
+// that we cannot create a new index on 'keyPattern' by changing any of these fields.
+for (let nonSigOpt of nonSignatureOptions) {
+    assert.commandFailedWithCode(
+        coll.createIndex(keyPattern, Object.assign({name: "non_sig_index"}, nonSigOpt)),
+        ErrorCodes.IndexOptionsConflict);
+}
+
+// Build a new index on {$**: 1} and verify that wildcardProjection is a non-signature field.
+// TODO SERVER-47659: wildcardProjection should be part of the signature.
+cmdRes = assert.commandWorked(coll.createIndex({"$**": 1}, {name: "wildcard_index_default_opts"}));
+assertNumIndexesAfterComparedToBefore(cmdRes, 1);
+assert.commandFailedWithCode(coll.createIndex({"$**": 1}, {wildcardProjection: {a: 1}}),
+                             ErrorCodes.IndexOptionsConflict);
+})();
diff --git a/jstests/multiVersion/index_signature_fcv.js b/jstests/multiVersion/index_signature_fcv.js
new file mode 100644
index 00000000000..db0c4fdc329
--- /dev/null
+++ b/jstests/multiVersion/index_signature_fcv.js
@@ -0,0 +1,90 @@
+/**
+ * Tests that the following FCV constraints are observed when building indexes:
+ *
+ *   - Multiple indexes which differ only by partial filter expression can be built in FCV 4.6.
+ *   - The planner can continue to use these indexes after downgrading to FCV 4.4.
+ *   - These indexes can be dropped in FCV 4.4.
+ *   - Indexes which differ only by partialFilterExpression cannot be created in FCV 4.4.
+ *   - We do not fassert if the set is downgraded to binary 4.4 with "duplicate" indexes present.
+ *
+ * TODO SERVER-47766: this test is specific to the 4.4 - 4.6 upgrade process, and can be removed
+ * after we branch for 4.7.
+ */
+(function() {
+"use strict";
+
+load("jstests/libs/analyze_plan.js");           // For isIxscan and hasRejectedPlans.
+load("jstests/multiVersion/libs/multi_rs.js");  // For upgradeSet.
+
+const rst = new ReplSetTest({
+    nodes: 2,
+    nodeOptions: {binVersion: "latest"},
+});
+rst.startSet();
+rst.initiate();
+
+let testDB = rst.getPrimary().getDB(jsTestName());
+let coll = testDB.test;
+
+// Verifies that the given query is indexed, and that 'numAlternativePlans' were generated.
+function assertIndexedQuery(query, numAlternativePlans) {
+    const explainOut = coll.explain().find(query).finish();
+    const results = coll.find(query).toArray();
+    assert(isIxscan(testDB, explainOut), explainOut);
+    assert.eq(getRejectedPlans(explainOut).length, numAlternativePlans, explainOut);
+}
+
+// Test that multiple indexes differing only by partialFilterExpression can be created in FCV 4.6.
+testDB.adminCommand({setFeatureCompatibilityVersion: latestFCV});
+assert.commandWorked(
+    coll.createIndex({a: 1}, {name: "index1", partialFilterExpression: {a: {$gte: 0}}}));
+assert.commandWorked(
+    coll.createIndex({a: 1}, {name: "index2", partialFilterExpression: {a: {$gte: 10}}}));
+assert.commandWorked(
+    coll.createIndex({a: 1}, {name: "index3", partialFilterExpression: {a: {$gte: 100}}}));
+
+// Test that the planner considers all relevant partial indexes when answering a query in FCV 4.6.
+assertIndexedQuery({a: 1}, 0);
+assertIndexedQuery({a: 11}, 1);
+assertIndexedQuery({a: 101}, 2);
+
+// Test that these indexes are retained and can be used by the planner when we downgrade to FCV 4.4.
+testDB.adminCommand({setFeatureCompatibilityVersion: lastStableFCV});
+assertIndexedQuery({a: 1}, 0);
+assertIndexedQuery({a: 11}, 1);
+assertIndexedQuery({a: 101}, 2);
+
+// Test that indexes distinguished only by partial filter can be dropped by name in FCV 4.4.
+assert.commandWorked(coll.dropIndex("index2"));
+assertIndexedQuery({a: 1}, 0);
+assertIndexedQuery({a: 11}, 0);
+assertIndexedQuery({a: 101}, 1);
+
+// Test that an index distinguished only by partialFilterExpression cannot be created in FCV 4.4.
+assert.commandFailedWithCode(
+    coll.createIndex({a: 1}, {name: "index2", partialFilterExpression: {a: {$gte: 10}}}),
+    ErrorCodes.IndexOptionsConflict);
+
+// Test that attempting to build an index with the same name and identical partialFilterExpression
+// as an existing index results in a no-op in FCV 4.4, and the command reports successful execution.
+const cmdRes = assert.commandWorked(
+    coll.createIndex({a: 1}, {name: "index1", partialFilterExpression: {a: {$gte: 0}}}));
+assert.eq(cmdRes.numIndexesBefore, cmdRes.numIndexesAfter);
+
+// Test that downgrading to binary 4.4 with overlapping partial indexes present does not fassert.
+rst.upgradeSet({binVersion: "last-stable"});
+testDB = rst.getPrimary().getDB(jsTestName());
+coll = testDB.test;
+
+// Test that the indexes still exist and can be used to answer queries on the binary 4.4 replset.
+assertIndexedQuery({a: 1}, 0);
+assertIndexedQuery({a: 11}, 0);
+assertIndexedQuery({a: 101}, 1);
+
+// Test that an index which differs only by partialFilterExpression cannot be created on binary 4.4.
+assert.commandFailedWithCode(
+    coll.createIndex({a: 1}, {name: "index2", partialFilterExpression: {a: {$gte: 10}}}),
+    ErrorCodes.IndexOptionsConflict);
+
+rst.stopSet();
+})();
diff --git a/jstests/multiVersion/libs/multi_rs.js b/jstests/multiVersion/libs/multi_rs.js
index 96c1df55982..c344b1d9393 100644
--- a/jstests/multiVersion/libs/multi_rs.js
+++ b/jstests/multiVersion/libs/multi_rs.js
@@ -19,13 +19,18 @@ ReplSetTest.prototype.upgradeSet = function(options, user, pwd) {
     this.upgradePrimary(primary, Object.assign({}, options), user, pwd);
 };
 
+function mergeNodeOptions(nodeOptions, options) {
+    for (let nodeName in nodeOptions) {
+        nodeOptions[nodeName] = Object.merge(nodeOptions[nodeName], options);
+    }
+    return nodeOptions;
+}
+
 ReplSetTest.prototype.upgradeMembers = function(primary, members, options, user, pwd) {
     const noDowntimePossible = this.nodes.length > 2;
 
     // Merge new options into node settings.
-    for (let nodeName in this.nodeOptions) {
-        this.nodeOptions[nodeName] = Object.merge(this.nodeOptions[nodeName], options);
-    }
+    this.nodeOptions = mergeNodeOptions(this.nodeOptions, options);
 
     for (let member of members) {
         this.upgradeNode(member, options, user, pwd);
@@ -48,19 +53,20 @@ ReplSetTest.prototype.upgradeSecondaries = function(primary, options, user, pwd)
 
 ReplSetTest.prototype.upgradeArbiters = function(primary, options, user, pwd) {
     // We don't support downgrading data files for arbiters. We need to instead delete the dbpath.
+    const oldStartClean = {startClean: (options && !!options["startClean"])};
     if (options && options.binVersion == "last-stable") {
         options["startClean"] = true;
     }
     this.upgradeMembers(primary, this.getArbiters(), options, user, pwd);
+    // Make sure we don't set {startClean:true} on other nodes unless the user explicitly requested.
+    this.nodeOptions = mergeNodeOptions(this.nodeOptions, oldStartClean);
 };
 
 ReplSetTest.prototype.upgradePrimary = function(primary, options, user, pwd) {
     const noDowntimePossible = this.nodes.length > 2;
 
     // Merge new options into node settings.
-    for (let nodeName in this.nodeOptions) {
-        this.nodeOptions[nodeName] = Object.merge(this.nodeOptions[nodeName], options);
-    }
+    this.nodeOptions = mergeNodeOptions(this.nodeOptions, options);
 
     let oldPrimary = this.stepdown(primary);
     this.waitForState(oldPrimary, ReplSetTest.State.SECONDARY);
diff --git a/src/mongo/db/catalog/SConscript b/src/mongo/db/catalog/SConscript
index 298478faa9b..aedda2ed6af 100644
--- a/src/mongo/db/catalog/SConscript
+++ b/src/mongo/db/catalog/SConscript
@@ -472,6 +472,7 @@ env.CppUnitTest(
         'index_build_entry_test.cpp',
         'index_builds_manager_test.cpp',
         'index_key_validate_test.cpp',
+        'index_signature_test.cpp',
         'index_spec_validate_test.cpp',
         'multi_index_block_test.cpp',
         'rename_collection_test.cpp',
diff --git a/src/mongo/db/catalog/index_catalog.h b/src/mongo/db/catalog/index_catalog.h
index 79add4ec623..9a97a935405 100644
--- a/src/mongo/db/catalog/index_catalog.h
+++ b/src/mongo/db/catalog/index_catalog.h
@@ -215,19 +215,15 @@ public:
         const bool includeUnfinishedIndexes = false) const = 0;
 
     /**
-     * Find index by matching key pattern and collation spec.  The key pattern and collation spec
-     * uniquely identify an index.
+     * Find index by matching key pattern and options. The key pattern, collation spec, and partial
+     * filter expression together uniquely identify an index.
      *
-     * Collation is specified as a normalized collation spec as returned by
-     * CollationInterface::getSpec.  An empty object indicates the simple collation.
-     *
-     * @return null if cannot find index, otherwise the index with a matching key pattern and
-     * collation.
+     * @return null if cannot find index, otherwise the index with a matching signature.
      */
-    virtual const IndexDescriptor* findIndexByKeyPatternAndCollationSpec(
+    virtual const IndexDescriptor* findIndexByKeyPatternAndOptions(
         OperationContext* const opCtx,
         const BSONObj& key,
-        const BSONObj& collationSpec,
+        const BSONObj& indexSpec,
         const bool includeUnfinishedIndexes = false) const = 0;
 
     /**
diff --git a/src/mongo/db/catalog/index_catalog_entry_impl.cpp b/src/mongo/db/catalog/index_catalog_entry_impl.cpp
index cca329b0007..16a69488248 100644
--- a/src/mongo/db/catalog/index_catalog_entry_impl.cpp
+++ b/src/mongo/db/catalog/index_catalog_entry_impl.cpp
@@ -105,13 +105,11 @@ IndexCatalogEntryImpl::IndexCatalogEntryImpl(OperationContext* const opCtx,
 
         // Parsing the partial filter expression is not expected to fail here since the
         // expression would have been successfully parsed upstream during index creation.
-        StatusWithMatchExpression statusWithMatcher =
-            MatchExpressionParser::parse(filter,
-                                         _expCtxForFilter,
-                                         ExtensionsCallbackNoop(),
-                                         MatchExpressionParser::kBanAllSpecialFeatures);
-        invariant(statusWithMatcher.getStatus());
-        _filterExpression = std::move(statusWithMatcher.getValue());
+        _filterExpression =
+            MatchExpressionParser::parseAndNormalize(filter,
+                                                     _expCtxForFilter,
+                                                     ExtensionsCallbackNoop(),
+                                                     MatchExpressionParser::kBanAllSpecialFeatures);
         LOGV2_DEBUG(20350,
                     2,
                     "have filter expression for {ns} {descriptor_indexName} {filter}",
diff --git a/src/mongo/db/catalog/index_catalog_impl.cpp b/src/mongo/db/catalog/index_catalog_impl.cpp
index c1e444128ff..c49b1b9424b 100644
--- a/src/mongo/db/catalog/index_catalog_impl.cpp
+++ b/src/mongo/db/catalog/index_catalog_impl.cpp
@@ -773,86 +773,93 @@ Status IndexCatalogImpl::_isSpecOk(OperationContext* opCtx, const BSONObj& spec)
 Status IndexCatalogImpl::_doesSpecConflictWithExisting(OperationContext* opCtx,
                                                        const BSONObj& spec,
                                                        const bool includeUnfinishedIndexes) const {
-    const char* name = spec.getStringField("name");
+    const char* name = spec.getStringField(IndexDescriptor::kIndexNameFieldName);
     invariant(name[0]);
 
-    const BSONObj key = spec.getObjectField("key");
-    const BSONObj collation = spec.getObjectField("collation");
+    const BSONObj key = spec.getObjectField(IndexDescriptor::kKeyPatternFieldName);
 
     {
+        // Check whether an index with the specified candidate name already exists in the catalog.
         const IndexDescriptor* desc = findIndexByName(opCtx, name, includeUnfinishedIndexes);
-        if (desc) {
-            // index already exists with same name
 
-            if (SimpleBSONObjComparator::kInstance.evaluate(desc->keyPattern() == key) &&
-                SimpleBSONObjComparator::kInstance.evaluate(
-                    desc->infoObj().getObjectField("collation") != collation)) {
-                // key patterns are equal but collations differ.
-                return Status(ErrorCodes::IndexOptionsConflict,
-                              str::stream()
-                                  << "An index with the same key pattern, but a different "
-                                  << "collation already exists with the same name.  Try again with "
-                                  << "a unique name. "
-                                  << "Existing index: " << desc->infoObj()
-                                  << " Requested index: " << spec);
-            }
+        if (desc) {
+            // Index already exists with same name. Check whether the options are the same as well.
+            IndexDescriptor candidate(_collection, _getAccessMethodName(key), spec);
+            auto indexComparison = candidate.compareIndexOptions(opCtx, getEntry(desc));
 
-            if (SimpleBSONObjComparator::kInstance.evaluate(desc->keyPattern() != key) ||
-                SimpleBSONObjComparator::kInstance.evaluate(
-                    desc->infoObj().getObjectField("collation") != collation)) {
+            // Key pattern or another uniquely-identifying option differs. We can build this index,
+            // but not with the specified (duplicate) name. User must specify another index name.
+            if (indexComparison == IndexDescriptor::Comparison::kDifferent) {
                 return Status(ErrorCodes::IndexKeySpecsConflict,
-                              str::stream()
-                                  << "Index must have unique name."
-                                  << "The existing index: " << desc->infoObj()
-                                  << " has the same name as the requested index: " << spec);
+                              str::stream() << "An existing index has the same name as the "
+                                               "requested index. Requested index: "
+                                            << spec << ", existing index: " << desc->infoObj());
             }
 
-            IndexDescriptor temp(_collection, _getAccessMethodName(key), spec);
-            if (!desc->areIndexOptionsEquivalent(&temp))
+            // The candidate's key and uniquely-identifying options are equivalent to an existing
+            // index, but some other options are not identical. Return a message to that effect.
+            if (indexComparison == IndexDescriptor::Comparison::kEquivalent) {
                 return Status(ErrorCodes::IndexOptionsConflict,
-                              str::stream() << "Index with name: " << name
-                                            << " already exists with different options");
+                              str::stream() << "An equivalent index already exists with the same "
+                                               "name but different options. Requested index: "
+                                            << spec << ", existing index: " << desc->infoObj());
+            }
 
+            // If we've reached this point, the requested index is identical to an existing index.
+            invariant(indexComparison == IndexDescriptor::Comparison::kIdentical);
 
             // If an identical index exists, but it is frozen, return an error with a different
-            // code to the user, forcing the user to drop before recreating the index.
+            // error code to the user, forcing the user to drop before recreating the index.
             auto entry = getEntry(desc);
             if (entry->isFrozen()) {
                 return Status(ErrorCodes::CannotCreateIndex,
-                              str::stream() << "An identical, unfinished index already exists. The "
-                                               "index must be dropped first: "
-                                            << name << ", spec: " << desc->infoObj());
+                              str::stream()
+                                  << "An identical, unfinished index '" << name
+                                  << "' already exists. Must drop before recreating. Spec: "
+                                  << desc->infoObj());
             }
 
-            // Index already exists with the same options, so no need to build a new
-            // one (not an error). Most likely requested by a client using ensureIndex.
+            // Index already exists with the same options, so there is no need to build a new one.
+            // This is not an error condition.
             return Status(ErrorCodes::IndexAlreadyExists,
                           str::stream() << "Identical index already exists: " << name);
         }
     }
 
     {
+        // No index with the candidate name exists. Check for an index with conflicting options.
         const IndexDescriptor* desc =
-            findIndexByKeyPatternAndCollationSpec(opCtx, key, collation, includeUnfinishedIndexes);
+            findIndexByKeyPatternAndOptions(opCtx, key, spec, includeUnfinishedIndexes);
+
         if (desc) {
             LOGV2_DEBUG(20353,
                         2,
-                        "Index already exists with a different name: {name} pattern: {key} "
-                        "collation: {collation}",
-                        "name"_attr = name,
-                        "key"_attr = key,
-                        "collation"_attr = collation);
-
-            IndexDescriptor temp(_collection, _getAccessMethodName(key), spec);
-            if (!desc->areIndexOptionsEquivalent(&temp))
+                        "Index already exists with a different name: {name}, spec: {spec}",
+                        "Index already exists with a different name",
+                        "name"_attr = desc->indexName(),
+                        "spec"_attr = desc->infoObj());
+
+            // Index already exists with a different name. Check whether the options are identical.
+            // We will return an error in either case, but this check allows us to generate a more
+            // informative error message.
+            IndexDescriptor candidate(_collection, _getAccessMethodName(key), spec);
+            auto indexComparison = candidate.compareIndexOptions(opCtx, getEntry(desc));
+
+            // The candidate's key and uniquely-identifying options are equivalent to an existing
+            // index, but some other options are not identical. Return a message to that effect.
+            if (indexComparison == IndexDescriptor::Comparison::kEquivalent)
                 return Status(ErrorCodes::IndexOptionsConflict,
-                              str::stream()
-                                  << "Index: " << spec
-                                  << " already exists with different options: " << desc->infoObj());
+                              str::stream() << "An equivalent index already exists with a "
+                                               "different name and options. Requested index: "
+                                            << spec << ", existing index: " << desc->infoObj());
 
+            // If we've reached this point, the requested index is identical to an existing index.
+            invariant(indexComparison == IndexDescriptor::Comparison::kIdentical);
+
+            // An identical index already exists with a different name. We cannot build this index.
             return Status(ErrorCodes::IndexOptionsConflict,
-                          str::stream() << "Index with name: " << name
-                                        << " already exists with a different name");
+                          str::stream() << "Index already exists with a different name: "
+                                        << desc->indexName());
         }
     }
 
@@ -1203,18 +1210,17 @@ const IndexDescriptor* IndexCatalogImpl::findIndexByName(OperationContext* opCtx
     return nullptr;
 }
 
-const IndexDescriptor* IndexCatalogImpl::findIndexByKeyPatternAndCollationSpec(
+const IndexDescriptor* IndexCatalogImpl::findIndexByKeyPatternAndOptions(
     OperationContext* opCtx,
     const BSONObj& key,
-    const BSONObj& collationSpec,
+    const BSONObj& indexSpec,
     bool includeUnfinishedIndexes) const {
     std::unique_ptr<IndexIterator> ii = getIndexIterator(opCtx, includeUnfinishedIndexes);
+    IndexDescriptor needle(_collection, _getAccessMethodName(key), indexSpec);
     while (ii->more()) {
-        const IndexDescriptor* desc = ii->next()->descriptor();
-        if (SimpleBSONObjComparator::kInstance.evaluate(desc->keyPattern() == key) &&
-            SimpleBSONObjComparator::kInstance.evaluate(
-                desc->infoObj().getObjectField("collation") == collationSpec)) {
-            return desc;
+        const auto* entry = ii->next();
+        if (needle.compareIndexOptions(opCtx, entry) != IndexDescriptor::Comparison::kDifferent) {
+            return entry->descriptor();
         }
     }
     return nullptr;
@@ -1242,7 +1248,7 @@ const IndexDescriptor* IndexCatalogImpl::findShardKeyPrefixedIndex(OperationCont
     std::unique_ptr<IndexIterator> ii = getIndexIterator(opCtx, false);
     while (ii->more()) {
         const IndexDescriptor* desc = ii->next()->descriptor();
-        bool hasSimpleCollation = desc->infoObj().getObjectField("collation").isEmpty();
+        bool hasSimpleCollation = desc->collation().isEmpty();
 
         if (desc->isPartial() || desc->isSparse())
             continue;
diff --git a/src/mongo/db/catalog/index_catalog_impl.h b/src/mongo/db/catalog/index_catalog_impl.h
index fb3ec499a0d..fa195004756 100644
--- a/src/mongo/db/catalog/index_catalog_impl.h
+++ b/src/mongo/db/catalog/index_catalog_impl.h
@@ -95,19 +95,15 @@ public:
                                            bool includeUnfinishedIndexes = false) const override;
 
     /**
-     * Find index by matching key pattern and collation spec.  The key pattern and collation spec
-     * uniquely identify an index.
+     * Find index by matching key pattern and options. The key pattern, collation spec, and partial
+     * filter expression together uniquely identify an index.
      *
-     * Collation is specified as a normalized collation spec as returned by
-     * CollationInterface::getSpec.  An empty object indicates the simple collation.
-     *
-     * @return null if cannot find index, otherwise the index with a matching key pattern and
-     * collation.
+     * @return null if cannot find index, otherwise the index with a matching signature.
      */
-    const IndexDescriptor* findIndexByKeyPatternAndCollationSpec(
+    const IndexDescriptor* findIndexByKeyPatternAndOptions(
         OperationContext* opCtx,
         const BSONObj& key,
-        const BSONObj& collationSpec,
+        const BSONObj& indexSpec,
         bool includeUnfinishedIndexes = false) const override;
 
     /**
diff --git a/src/mongo/db/catalog/index_catalog_noop.h b/src/mongo/db/catalog/index_catalog_noop.h
index ebe91ec2b72..3c66fcd90b1 100644
--- a/src/mongo/db/catalog/index_catalog_noop.h
+++ b/src/mongo/db/catalog/index_catalog_noop.h
@@ -81,10 +81,10 @@ public:
         return nullptr;
     }
 
-    IndexDescriptor* findIndexByKeyPatternAndCollationSpec(
+    IndexDescriptor* findIndexByKeyPatternAndOptions(
         OperationContext* const opCtx,
         const BSONObj& key,
-        const BSONObj& collationSpec,
+        const BSONObj& indexSpec,
         const bool includeUnfinishedIndexes = false) const override {
         return nullptr;
     }
diff --git a/src/mongo/db/catalog/index_signature_test.cpp b/src/mongo/db/catalog/index_signature_test.cpp
new file mode 100644
index 00000000000..07f8d70d755
--- /dev/null
+++ b/src/mongo/db/catalog/index_signature_test.cpp
@@ -0,0 +1,242 @@
+/**
+ *    Copyright (C) 2020-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#include "mongo/platform/basic.h"
+
+#include "mongo/db/catalog/catalog_test_fixture.h"
+#include "mongo/db/catalog/collection.h"
+#include "mongo/db/catalog/index_catalog_entry_impl.h"
+#include "mongo/db/catalog_raii.h"
+#include "mongo/db/index/index_descriptor.h"
+#include "mongo/db/query/collection_query_info.h"
+
+namespace mongo {
+namespace {
+
+class IndexSignatureTest : public CatalogTestFixture {
+public:
+    IndexSignatureTest() : CatalogTestFixture() {}
+
+    StatusWith<const IndexCatalogEntry*> createIndex(BSONObj spec) {
+        // Get the index catalog associated with the test collection.
+        auto* indexCatalog = coll()->getIndexCatalog();
+        // Build the specified index on the collection.
+        WriteUnitOfWork wuow(opCtx());
+        auto status = indexCatalog->createIndexOnEmptyCollection(opCtx(), spec);
+        if (!status.isOK()) {
+            return status.getStatus();
+        }
+        wuow.commit();
+        // Find the index entry and return it.
+        return indexCatalog->getEntry(indexCatalog->findIndexByName(
+            opCtx(), spec.getStringField(IndexDescriptor::kIndexNameFieldName)));
+    }
+
+    std::unique_ptr<IndexDescriptor> makeIndexDescriptor(BSONObj spec) {
+        auto keyPattern = spec.getObjectField(IndexDescriptor::kKeyPatternFieldName);
+        return std::make_unique<IndexDescriptor>(
+            coll(), IndexNames::findPluginName(keyPattern), spec);
+    }
+
+    const NamespaceString& nss() const {
+        return _nss;
+    }
+
+    Collection* coll() const {
+        return _coll->getCollection();
+    }
+
+    OperationContext* opCtx() {
+        return operationContext();
+    }
+
+protected:
+    void setUp() override {
+        CatalogTestFixture::setUp();
+        ASSERT_OK(storageInterface()->createCollection(opCtx(), _nss, {}));
+        _coll.emplace(opCtx(), _nss, MODE_X);
+    }
+
+    void tearDown() override {
+        _coll.reset();
+        CatalogTestFixture::tearDown();
+    }
+
+    // Helper function to add all elements in 'fields' into 'input', replacing any existing fields.
+    // Returns a new object containing the added fields.
+    BSONObj _addFields(BSONObj input, BSONObj fields) {
+        for (auto&& field : fields) {
+            input = input.addField(field);
+        }
+        return input;
+    }
+
+private:
+    boost::optional<AutoGetCollection> _coll;
+    NamespaceString _nss{"fooDB.barColl"};
+};
+
+TEST_F(IndexSignatureTest, CanCreateMultipleIndexesOnSameKeyPatternWithDifferentCollations) {
+    // Create an index on {a: 1} with an 'en_US' collation.
+    auto indexSpec = fromjson("{v: 2, name: 'a_1', key: {a: 1}, collation: {locale: 'en_US'}}");
+    auto* basicIndex = unittest::assertGet(createIndex(indexSpec));
+
+    // Create an index descriptor on the same keyPattern whose collation has an explicit strength.
+    // The two indexes compare as kIdentical, because the collation comparison is performed using
+    // the parsed collator rather than the static collation object from the BSON index specs.
+    auto collationDesc = makeIndexDescriptor(
+        _addFields(indexSpec, fromjson("{collation: {locale: 'en_US', strength: 3}}")));
+    ASSERT(collationDesc->compareIndexOptions(opCtx(), basicIndex) ==
+           IndexDescriptor::Comparison::kIdentical);
+
+    // Confirm that attempting to build this index will result in ErrorCodes::IndexAlreadyExists.
+    ASSERT_EQ(createIndex(collationDesc->infoObj()), ErrorCodes::IndexAlreadyExists);
+
+    // Now set the unique field. The indexes now compare kEquivalent, but not kIdentical. This means
+    // that all signature fields which uniquely identify the index match, but other fields differ.
+    auto collationUniqueDesc =
+        makeIndexDescriptor(_addFields(collationDesc->infoObj(), fromjson("{unique: true}")));
+    ASSERT(collationUniqueDesc->compareIndexOptions(opCtx(), basicIndex) ==
+           IndexDescriptor::Comparison::kEquivalent);
+
+    // Attempting to build the index, whether with the same name or a different name, now throws
+    // IndexOptionsConflict. The error message returned with the exception specifies whether the
+    // name matches an existing index.
+    ASSERT_EQ(createIndex(collationUniqueDesc->infoObj()), ErrorCodes::IndexOptionsConflict);
+    ASSERT_EQ(createIndex(_addFields(collationUniqueDesc->infoObj(),
+                                     fromjson("{name: 'collationUnique'}"))),
+              ErrorCodes::IndexOptionsConflict);
+
+    // Now create an index spec with an entirely different collation. The two indexes compare as
+    // being kDifferent; this means that both of these indexes can co-exist together.
+    auto differentCollationDesc = makeIndexDescriptor(
+        _addFields(collationDesc->infoObj(), fromjson("{collation: {locale: 'fr'}}")));
+    ASSERT(differentCollationDesc->compareIndexOptions(opCtx(), basicIndex) ==
+           IndexDescriptor::Comparison::kDifferent);
+
+    // Verify that we can build this index alongside the existing indexes.
+    ASSERT_OK(createIndex(
+        _addFields(differentCollationDesc->infoObj(), fromjson("{name: 'differentCollation'}"))));
+}
+
+TEST_F(IndexSignatureTest,
+       CanCreateMultipleIndexesOnSameKeyPatternWithDifferentPartialFilterExpressions) {
+    // Create a basic index on {a: 1} with no partialFilterExpression.
+    auto indexSpec = fromjson("{v: 2, name: 'a_1', key: {a: 1}}");
+    auto* basicIndex = unittest::assertGet(createIndex(indexSpec));
+
+    // Create an index descriptor on the same keyPattern with a partialFilterExpression. The two
+    // indexes compare as kDifferent, because the partialFilterExpression is one of the fields in
+    // the index's signature.
+    auto partialFilterDesc = makeIndexDescriptor(_addFields(
+        indexSpec, fromjson("{partialFilterExpression: {a: {$gt: 5, $lt: 10}, b: 'blah'}}")));
+    ASSERT(partialFilterDesc->compareIndexOptions(opCtx(), basicIndex) ==
+           IndexDescriptor::Comparison::kDifferent);
+
+    // Verify that we can build an index with this spec alongside the original index.
+    auto* partialFilterIndex = unittest::assertGet(
+        createIndex(_addFields(partialFilterDesc->infoObj(), fromjson("{name: 'partialFilter'}"))));
+
+    // Verify that partialFilterExpressions are normalized before being compared. Here, the filter
+    // is expressed differently than in the existing index, but the two still compare as kIdentical.
+    auto partialFilterDupeDesc = makeIndexDescriptor(
+        _addFields(indexSpec,
+                   fromjson("{name: 'partialFilter', partialFilterExpression: {$and: [{b: 'blah'}, "
+                            "{a: {$lt: 10}}, {a: {$gt: 5}}]}}")));
+    ASSERT(partialFilterDupeDesc->compareIndexOptions(opCtx(), partialFilterIndex) ==
+           IndexDescriptor::Comparison::kIdentical);
+
+    // Confirm that attempting to build this index will result in ErrorCodes::IndexAlreadyExists.
+    ASSERT_EQ(createIndex(partialFilterDupeDesc->infoObj()), ErrorCodes::IndexAlreadyExists);
+
+    // Now set the unique field. The indexes now compare kEquivalent, but not kIdentical. This means
+    // that all signature fields which uniquely identify the index match, but other fields differ.
+    auto partialFilterUniqueDesc =
+        makeIndexDescriptor(_addFields(partialFilterDesc->infoObj(), fromjson("{unique: true}")));
+    ASSERT(partialFilterUniqueDesc->compareIndexOptions(opCtx(), partialFilterIndex) ==
+           IndexDescriptor::Comparison::kEquivalent);
+
+    // Attempting to build the index, whether with the same name or a different name, now throws
+    // IndexOptionsConflict. The error message returned with the exception specifies whether the
+    // name matches an existing index.
+    ASSERT_EQ(createIndex(_addFields(partialFilterUniqueDesc->infoObj(),
+                                     fromjson("{name: 'partialFilterExpression'}"))),
+              ErrorCodes::IndexOptionsConflict);
+    ASSERT_EQ(createIndex(_addFields(partialFilterUniqueDesc->infoObj(),
+                                     fromjson("{name: 'partialFilterUnique'}"))),
+              ErrorCodes::IndexOptionsConflict);
+
+    // Now create an index spec with an entirely different partialFilterExpression. The two indexes
+    // compare as kDifferent; this means that both of these indexes can co-exist together.
+    auto differentPartialFilterDesc = makeIndexDescriptor(
+        _addFields(partialFilterDesc->infoObj(),
+                   fromjson("{partialFilterExpression: {a: {$gt: 0, $lt: 10}, b: 'blah'}}")));
+    ASSERT(differentPartialFilterDesc->compareIndexOptions(opCtx(), partialFilterIndex) ==
+           IndexDescriptor::Comparison::kDifferent);
+
+    // Verify that we can build this index alongside the existing indexes.
+    ASSERT_OK(createIndex(_addFields(differentPartialFilterDesc->infoObj(),
+                                     fromjson("{name: 'differentPartialFilter'}"))));
+}
+
+TEST_F(IndexSignatureTest, CannotCreateMultipleIndexesOnSameKeyPatternIfNonSignatureFieldsDiffer) {
+    // Create a basic index on {a: 1} with all other options set to their default values.
+    auto indexSpec = fromjson("{v: 2, name: 'a_1', key: {a: 1}}");
+    auto* basicIndex = unittest::assertGet(createIndex(indexSpec));
+
+    // TODO SERVER-47657: unique and sparse should be part of the signature.
+    std::vector<BSONObj> nonSigOptions = {
+        BSON(IndexDescriptor::kUniqueFieldName << true),
+        BSON(IndexDescriptor::kSparseFieldName << true),
+        BSON(IndexDescriptor::kExpireAfterSecondsFieldName << 10)};
+
+    // Verify that changing each of the non-signature fields does not distinguish this index from
+    // the existing index. The two are considered equivalent, and we cannot build the new index.
+    for (auto&& nonSigOpt : nonSigOptions) {
+        auto nonSigDesc = makeIndexDescriptor(_addFields(indexSpec, nonSigOpt));
+        ASSERT(nonSigDesc->compareIndexOptions(opCtx(), basicIndex) ==
+               IndexDescriptor::Comparison::kEquivalent);
+        ASSERT_EQ(createIndex(nonSigDesc->infoObj()), ErrorCodes::IndexOptionsConflict);
+    }
+
+    // Build a wildcard index and confirm that 'wildcardProjection' is a non-signature field.
+    // TODO SERVER-47659: wildcardProjection should be part of the signature.
+    auto* wildcardIndex =
+        unittest::assertGet(createIndex(fromjson("{v: 2, name: '$**_1', key: {'$**': 1}}")));
+    auto nonSigWildcardDesc = makeIndexDescriptor(_addFields(
+        wildcardIndex->descriptor()->infoObj(), fromjson("{wildcardProjection: {a: 1}}")));
+    ASSERT(nonSigWildcardDesc->compareIndexOptions(opCtx(), wildcardIndex) ==
+           IndexDescriptor::Comparison::kEquivalent);
+    ASSERT_EQ(createIndex(
+                  _addFields(nonSigWildcardDesc->infoObj(), fromjson("{name: 'nonSigWildcard'}"))),
+              ErrorCodes::IndexOptionsConflict);
+}
+
+}  // namespace
+}  // namespace mongo
diff --git a/src/mongo/db/commands/create_indexes.cpp b/src/mongo/db/commands/create_indexes.cpp
index f4c2297eadc..689f66e2b6c 100644
--- a/src/mongo/db/commands/create_indexes.cpp
+++ b/src/mongo/db/commands/create_indexes.cpp
@@ -319,13 +319,11 @@ boost::optional<CommitQuorumOptions> parseAndGetCommitQuorum(OperationContext* o
 std::vector<BSONObj> resolveDefaultsAndRemoveExistingIndexes(OperationContext* opCtx,
                                                              const Collection* collection,
                                                              std::vector<BSONObj> indexSpecs) {
-    auto swDefaults = collection->addCollationDefaultsToIndexSpecsForCreate(opCtx, indexSpecs);
-    uassertStatusOK(swDefaults.getStatus());
+    // Normalize the specs' collations, wildcard projections, and partial filters as applicable.
+    auto normalSpecs = IndexBuildsCoordinator::normalizeIndexSpecs(opCtx, collection, indexSpecs);
 
-    auto indexCatalog = collection->getIndexCatalog();
-
-    return indexCatalog->removeExistingIndexes(
-        opCtx, swDefaults.getValue(), false /*removeIndexBuildsToo*/);
+    return collection->getIndexCatalog()->removeExistingIndexes(
+        opCtx, normalSpecs, false /*removeIndexBuildsToo*/);
 }
 
 /**
diff --git a/src/mongo/db/index/SConscript b/src/mongo/db/index/SConscript
index fcff88d1e14..eaa3ebaaf40 100644
--- a/src/mongo/db/index/SConscript
+++ b/src/mongo/db/index/SConscript
@@ -17,6 +17,8 @@ env.Library(
         LIBDEPS_PRIVATE=[
             '$BUILD_DIR/mongo/db/index_names',
             '$BUILD_DIR/mongo/db/namespace_string',
+            '$BUILD_DIR/mongo/db/matcher/expressions',
+            '$BUILD_DIR/mongo/db/query/collation/collator_factory_interface',
         ],
 )
 
diff --git a/src/mongo/db/index/index_descriptor.cpp b/src/mongo/db/index/index_descriptor.cpp
index 17f765d87cf..c4da010008c 100644
--- a/src/mongo/db/index/index_descriptor.cpp
+++ b/src/mongo/db/index/index_descriptor.cpp
@@ -34,6 +34,8 @@
 #include "mongo/db/catalog/collection.h"
 #include "mongo/db/catalog/index_catalog.h"
 #include "mongo/db/index/index_descriptor.h"
+#include "mongo/db/matcher/expression_parser.h"
+#include "mongo/db/query/collation/collator_factory_interface.h"
 
 #include <algorithm>
 
@@ -50,9 +52,9 @@ void populateOptionsMap(std::map<StringData, BSONElement>& theMap, const BSONObj
         const BSONElement e = it.next();
 
         StringData fieldName = e.fieldNameStringData();
-        if (fieldName == IndexDescriptor::kKeyPatternFieldName ||
-            fieldName == IndexDescriptor::kNamespaceFieldName ||  // removed in 4.4
-            fieldName == IndexDescriptor::kIndexNameFieldName ||
+        if (fieldName == IndexDescriptor::kKeyPatternFieldName ||  // checked specially
+            fieldName == IndexDescriptor::kNamespaceFieldName ||   // removed in 4.4
+            fieldName == IndexDescriptor::kIndexNameFieldName ||   // checked separately
             fieldName ==
                 IndexDescriptor::kIndexVersionFieldName ||  // not considered for equivalence
             fieldName == IndexDescriptor::kTextVersionFieldName ||      // same as index version
@@ -60,9 +62,9 @@ void populateOptionsMap(std::map<StringData, BSONElement>& theMap, const BSONObj
             fieldName ==
                 IndexDescriptor::kBackgroundFieldName ||  // this is a creation time option only
             fieldName == IndexDescriptor::kDropDuplicatesFieldName ||  // this is now ignored
-            fieldName == IndexDescriptor::kSparseFieldName ||          // checked specially
-            fieldName == IndexDescriptor::kHiddenFieldName ||  // not considered for equivalence
-            fieldName == IndexDescriptor::kUniqueFieldName     // check specially
+            fieldName == IndexDescriptor::kHiddenFieldName ||     // not considered for equivalence
+            fieldName == IndexDescriptor::kCollationFieldName ||  // checked specially
+            fieldName == IndexDescriptor::kPartialFilterExprFieldName  // checked specially
         ) {
             continue;
         }
@@ -176,25 +178,69 @@ const NamespaceString& IndexDescriptor::parentNS() const {
     return _collection->ns();
 }
 
-bool IndexDescriptor::areIndexOptionsEquivalent(const IndexDescriptor* other) const {
-    if (isSparse() != other->isSparse()) {
-        return false;
+IndexDescriptor::Comparison IndexDescriptor::compareIndexOptions(
+    OperationContext* opCtx, const IndexCatalogEntry* other) const {
+    // We first check whether the key pattern is identical for both indexes.
+    if (SimpleBSONObjComparator::kInstance.evaluate(keyPattern() !=
+                                                    other->descriptor()->keyPattern())) {
+        return Comparison::kDifferent;
     }
 
-    if (!isIdIndex() && unique() != other->unique()) {
-        // Note: { _id: 1 } or { _id: -1 } implies unique: true.
-        return false;
+    // Check whether both indexes have the same collation. If not, then they are not equivalent.
+    auto collator = collation().isEmpty()
+        ? nullptr
+        : uassertStatusOK(
+              CollatorFactoryInterface::get(opCtx->getServiceContext())->makeFromBSON(collation()));
+    if (!CollatorInterface::collatorsMatch(collator.get(), other->getCollator())) {
+        return Comparison::kDifferent;
     }
 
-    // Then compare the rest of the options.
+    // The partialFilterExpression is only part of the index signature if FCV has been set to 4.6.
+    // TODO SERVER-47766: remove these FCV checks after we branch for 4.7.
+    auto isFCV46 = serverGlobalParams.featureCompatibility.isVersion(
+        ServerGlobalParams::FeatureCompatibility::Version::kFullyUpgradedTo46);
+
+    // If we have a partial filter expression and the other index doesn't, or vice-versa, then the
+    // two indexes are not equivalent. We therefore return Comparison::kDifferent immediately.
+    if (isFCV46 && isPartial() != other->descriptor()->isPartial()) {
+        return Comparison::kDifferent;
+    }
+    // Compare 'partialFilterExpression' in each descriptor to see if they are equivalent. We use
+    // the collator that we parsed earlier to create the filter's ExpressionContext, although we
+    // don't currently consider collation when comparing string predicates for filter equivalence.
+    // For instance, under a case-sensitive collation, the predicates {a: "blah"} and {a: "BLAH"}
+    // would match the same set of documents, but these are not currently considered equivalent.
+    // TODO SERVER-47664: take collation into account while comparing string predicates.
+    if (isFCV46 && other->getFilterExpression()) {
+        auto expCtx =
+            make_intrusive<ExpressionContext>(opCtx, std::move(collator), _collection->ns());
+        auto filter = MatchExpressionParser::parseAndNormalize(partialFilterExpression(), expCtx);
+        if (!filter->equivalent(other->getFilterExpression())) {
+            return Comparison::kDifferent;
+        }
+    }
+
+    // If we are here, then the two descriptors match on all option fields that uniquely distinguish
+    // an index, and so the return value will be at least Comparison::kEquivalent. We now proceed to
+    // compare the rest of the options to see if we should return Comparison::kIdentical instead.
 
     std::map<StringData, BSONElement> existingOptionsMap;
     populateOptionsMap(existingOptionsMap, infoObj());
 
     std::map<StringData, BSONElement> newOptionsMap;
-    populateOptionsMap(newOptionsMap, other->infoObj());
+    populateOptionsMap(newOptionsMap, other->descriptor()->infoObj());
+
+    // If the FCV has not been upgraded to 4.6, add partialFilterExpression to the options map. It
+    // does not contribute to the index signature, but can determine whether or not the candidate
+    // index is identical to the existing index.
+    if (!isFCV46) {
+        existingOptionsMap[IndexDescriptor::kPartialFilterExprFieldName] =
+            other->descriptor()->infoObj()[IndexDescriptor::kPartialFilterExprFieldName];
+        newOptionsMap[IndexDescriptor::kPartialFilterExprFieldName] =
+            infoObj()[IndexDescriptor::kPartialFilterExprFieldName];
+    }
 
-    return existingOptionsMap.size() == newOptionsMap.size() &&
+    const bool optsIdentical = existingOptionsMap.size() == newOptionsMap.size() &&
         std::equal(existingOptionsMap.begin(),
                    existingOptionsMap.end(),
                    newOptionsMap.begin(),
@@ -204,6 +250,10 @@ bool IndexDescriptor::areIndexOptionsEquivalent(const IndexDescriptor* other) co
                            SimpleBSONElementComparator::kInstance.evaluate(lhs.second ==
                                                                            rhs.second);
                    });
+
+    // If all non-identifying options also match, the descriptors are identical. Otherwise, we
+    // consider them equivalent; two indexes with these options and the same key cannot coexist.
+    return optsIdentical ? Comparison::kIdentical : Comparison::kEquivalent;
 }
 
 }  // namespace mongo
diff --git a/src/mongo/db/index/index_descriptor.h b/src/mongo/db/index/index_descriptor.h
index 4eab168a65a..c3d68d7a65d 100644
--- a/src/mongo/db/index/index_descriptor.h
+++ b/src/mongo/db/index/index_descriptor.h
@@ -59,6 +59,13 @@ public:
     enum class IndexVersion { kV1 = 1, kV2 = 2 };
     static constexpr IndexVersion kLatestIndexVersion = IndexVersion::kV2;
 
+    // Used to report the result of a comparison between two indexes.
+    enum class Comparison {
+        kDifferent,   // Indicates that the indexes do not match.
+        kEquivalent,  // Indicates that the options which uniquely identify an index match.
+        kIdentical    // Indicates that all applicable index options match.
+    };
+
     static constexpr StringData k2dIndexBitsFieldName = "bits"_sd;
     static constexpr StringData k2dIndexMinFieldName = "min"_sd;
     static constexpr StringData k2dIndexMaxFieldName = "max"_sd;
@@ -218,7 +225,12 @@ public:
     }
     const IndexCatalog* getIndexCatalog() const;
 
-    bool areIndexOptionsEquivalent(const IndexDescriptor* other) const;
+    /**
+     * Compares the current IndexDescriptor against the given index entry. Returns kIdentical if all
+     * index options are logically identical, kEquivalent if all options which uniquely identify an
+     * index are logically identical, and kDifferent otherwise.
+     */
+    Comparison compareIndexOptions(OperationContext* opCtx, const IndexCatalogEntry* other) const;
 
     const BSONObj& collation() const {
         return _collation;
diff --git a/src/mongo/db/index_builds_coordinator.cpp b/src/mongo/db/index_builds_coordinator.cpp
index 89502ea72cd..0c3e6e794db 100644
--- a/src/mongo/db/index_builds_coordinator.cpp
+++ b/src/mongo/db/index_builds_coordinator.cpp
@@ -45,6 +45,7 @@
 #include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/curop.h"
 #include "mongo/db/db_raii.h"
+#include "mongo/db/index/wildcard_key_generator.h"
 #include "mongo/db/index_build_entry_helpers.h"
 #include "mongo/db/op_observer.h"
 #include "mongo/db/operation_context.h"
@@ -2453,15 +2454,15 @@ std::vector<BSONObj> IndexBuildsCoordinator::prepareSpecListForCreate(
         return indexSpecs;
     }
 
-    auto specsWithCollationDefaults =
-        uassertStatusOK(collection->addCollationDefaultsToIndexSpecsForCreate(opCtx, indexSpecs));
+    // Normalize the specs' collations, wildcard projections, and partial filters as applicable.
+    auto normalSpecs = normalizeIndexSpecs(opCtx, collection, indexSpecs);
 
+    // Remove any index specifications which already exist in the catalog.
     auto indexCatalog = collection->getIndexCatalog();
-    std::vector<BSONObj> resultSpecs;
-
-    resultSpecs = indexCatalog->removeExistingIndexes(
-        opCtx, specsWithCollationDefaults, true /*removeIndexBuildsToo*/);
+    auto resultSpecs =
+        indexCatalog->removeExistingIndexes(opCtx, normalSpecs, true /*removeIndexBuildsToo*/);
 
+    // Verify that each spec is compatible with the collection's sharding state.
     for (const BSONObj& spec : resultSpecs) {
         if (spec[kUniqueFieldName].trueValue()) {
             checkShardKeyRestrictions(opCtx, nss, spec[kKeyFieldName].Obj());
@@ -2471,4 +2472,50 @@ std::vector<BSONObj> IndexBuildsCoordinator::prepareSpecListForCreate(
     return resultSpecs;
 }
 
+std::vector<BSONObj> IndexBuildsCoordinator::normalizeIndexSpecs(
+    OperationContext* opCtx, const Collection* collection, const std::vector<BSONObj>& indexSpecs) {
+    // This helper function may be called before the collection is created, when we are attempting
+    // to check whether the candidate index collides with any existing indexes. If 'collection' is
+    // nullptr, skip normalization. Since the collection does not exist there cannot be a conflict,
+    // and we will normalize once the candidate spec is submitted to the IndexBuildsCoordinator.
+    if (!collection) {
+        return indexSpecs;
+    }
+
+    // Add collection-default collation where needed and normalize the collation in each index spec.
+    auto normalSpecs =
+        uassertStatusOK(collection->addCollationDefaultsToIndexSpecsForCreate(opCtx, indexSpecs));
+
+    // If the index spec has a partialFilterExpression, we normalize it by parsing to an optimized,
+    // sorted MatchExpression tree, re-serialize it to BSON, and add it back into the index spec.
+    const auto expCtx = make_intrusive<ExpressionContext>(opCtx, nullptr, collection->ns());
+    std::transform(normalSpecs.begin(), normalSpecs.end(), normalSpecs.begin(), [&](auto& spec) {
+        const auto kPartialFilterName = IndexDescriptor::kPartialFilterExprFieldName;
+        auto partialFilterExpr = spec.getObjectField(kPartialFilterName);
+        if (partialFilterExpr.isEmpty()) {
+            return spec;
+        }
+        // Parse, optimize and sort the MatchExpression to reduce it to its normalized form.
+        // Serialize the normalized filter back into the index spec before returning.
+        auto partialFilter = MatchExpressionParser::parseAndNormalize(partialFilterExpr, expCtx);
+        return spec.addField(BSON(kPartialFilterName << partialFilter->serialize()).firstElement());
+    });
+
+    // If any of the specs describe wildcard indexes, normalize the wildcard projections if present.
+    // This will change all specs of the form {"a.b.c": 1} to normalized form {a: {b: {c : 1}}}.
+    std::transform(normalSpecs.begin(), normalSpecs.end(), normalSpecs.begin(), [](auto& spec) {
+        const auto kProjectionName = IndexDescriptor::kPathProjectionFieldName;
+        const auto pathProjectionSpec = spec.getObjectField(kProjectionName);
+        static const auto kWildcardKeyPattern = BSON("$**" << 1);
+        if (pathProjectionSpec.isEmpty()) {
+            return spec;
+        }
+        auto wildcardProjection =
+            WildcardKeyGenerator::createProjectionExecutor(kWildcardKeyPattern, pathProjectionSpec);
+        auto normalizedProjection =
+            wildcardProjection.exec()->serializeTransformation(boost::none).toBson();
+        return spec.addField(BSON(kProjectionName << normalizedProjection).firstElement());
+    });
+    return normalSpecs;
+}
 }  // namespace mongo
diff --git a/src/mongo/db/index_builds_coordinator.h b/src/mongo/db/index_builds_coordinator.h
index aecb7de265b..10464097996 100644
--- a/src/mongo/db/index_builds_coordinator.h
+++ b/src/mongo/db/index_builds_coordinator.h
@@ -424,6 +424,22 @@ public:
                                                          const std::vector<BSONObj>& indexSpecs);
 
     /**
+     * Helper function which normalizes a vector of index specs. This function will populate a
+     * complete collation spec in cases where the index spec specifies a collation, and will add
+     * the collection-default collation, if present, in cases where collation is omitted. If the
+     * index spec omits the collation and the collection does not have a default, the collation
+     * field is omitted from the spec. This function also converts 'wildcardProjection' and
+     * 'partialFilterExpression' to canonical form in any cases where they exist.
+     *
+     * If 'collection' is null, no changes are made to the input specs.
+     *
+     * This function throws on error.
+     */
+    static std::vector<BSONObj> normalizeIndexSpecs(OperationContext* opCtx,
+                                                    const Collection* collection,
+                                                    const std::vector<BSONObj>& indexSpecs);
+
+    /**
      * Returns total number of indexes in collection, including unfinished/in-progress indexes.
      *
      * Used to set statistics on index build results.
diff --git a/src/mongo/db/matcher/expression.cpp b/src/mongo/db/matcher/expression.cpp
index c035e417848..680b75a8001 100644
--- a/src/mongo/db/matcher/expression.cpp
+++ b/src/mongo/db/matcher/expression.cpp
@@ -40,12 +40,77 @@ namespace mongo {
  */
 MONGO_FAIL_POINT_DEFINE(disableMatchExpressionOptimization);
 
+namespace {
+/**
+ * Comparator for MatchExpression nodes.  Returns an integer less than, equal to, or greater
+ * than zero if 'lhs' is less than, equal to, or greater than 'rhs', respectively.
+ *
+ * Sorts by:
+ * 1) operator type (MatchExpression::MatchType)
+ * 2) path name (MatchExpression::path())
+ * 3) sort order of children
+ * 4) number of children (MatchExpression::numChildren())
+ *
+ * The third item is needed to ensure that match expression trees which should have the same
+ * cache key always sort the same way. If you're wondering when the tuple (operator type, path
+ * name) could ever be equal, consider this query:
+ *
+ * {$and:[{$or:[{a:1},{a:2}]},{$or:[{a:1},{b:2}]}]}
+ *
+ * The two OR nodes would compare as equal in this case were it not for tuple item #3 (sort
+ * order of children).
+ */
+int matchExpressionComparator(const MatchExpression* lhs, const MatchExpression* rhs) {
+    MatchExpression::MatchType lhsMatchType = lhs->matchType();
+    MatchExpression::MatchType rhsMatchType = rhs->matchType();
+    if (lhsMatchType != rhsMatchType) {
+        return lhsMatchType < rhsMatchType ? -1 : 1;
+    }
+
+    StringData lhsPath = lhs->path();
+    StringData rhsPath = rhs->path();
+    int pathsCompare = lhsPath.compare(rhsPath);
+    if (pathsCompare != 0) {
+        return pathsCompare;
+    }
+
+    const size_t numChildren = std::min(lhs->numChildren(), rhs->numChildren());
+    for (size_t childIdx = 0; childIdx < numChildren; ++childIdx) {
+        int childCompare =
+            matchExpressionComparator(lhs->getChild(childIdx), rhs->getChild(childIdx));
+        if (childCompare != 0) {
+            return childCompare;
+        }
+    }
+
+    if (lhs->numChildren() != rhs->numChildren()) {
+        return lhs->numChildren() < rhs->numChildren() ? -1 : 1;
+    }
+
+    // They're equal!
+    return 0;
+}
+
+bool matchExpressionLessThan(const MatchExpression* lhs, const MatchExpression* rhs) {
+    return matchExpressionComparator(lhs, rhs) < 0;
+}
+
+}  // namespace
+
 MatchExpression::MatchExpression(MatchType type) : _matchType(type) {}
 
+// static
+void MatchExpression::sortTree(MatchExpression* tree) {
+    for (size_t i = 0; i < tree->numChildren(); ++i) {
+        sortTree(tree->getChild(i));
+    }
+    if (auto&& children = tree->getChildVector()) {
+        std::stable_sort(children->begin(), children->end(), matchExpressionLessThan);
+    }
+}
+
 std::string MatchExpression::toString() const {
-    BSONObjBuilder bob;
-    serialize(&bob, true);
-    return bob.obj().toString();
+    return serialize().toString();
 }
 
 std::string MatchExpression::debugString() const {
diff --git a/src/mongo/db/matcher/expression.h b/src/mongo/db/matcher/expression.h
index f870dbcbcda..052ab19c7ea 100644
--- a/src/mongo/db/matcher/expression.h
+++ b/src/mongo/db/matcher/expression.h
@@ -158,6 +158,21 @@ public:
         }
     }
 
+    /**
+     * Traverses expression tree post-order. Sorts children at each non-leaf node by (MatchType,
+     * path(), children, number of children).
+     */
+    static void sortTree(MatchExpression* tree);
+
+    /**
+     * Convenience method which normalizes a MatchExpression tree by optimizing and then sorting it.
+     */
+    static std::unique_ptr<MatchExpression> normalize(std::unique_ptr<MatchExpression> tree) {
+        tree = optimize(std::move(tree));
+        sortTree(tree.get());
+        return tree;
+    }
+
     MatchExpression(MatchType type);
     virtual ~MatchExpression() {}
 
@@ -299,6 +314,15 @@ public:
     virtual void serialize(BSONObjBuilder* out, bool includePath = true) const = 0;
 
     /**
+     * Convenience method which serializes this MatchExpression to a BSONObj.
+     */
+    BSONObj serialize(bool includePath = true) const {
+        BSONObjBuilder bob;
+        serialize(&bob, includePath);
+        return bob.obj();
+    }
+
+    /**
      * Returns true if this expression will always evaluate to false, such as an $or with no
      * children.
      */
diff --git a/src/mongo/db/matcher/expression_parser.cpp b/src/mongo/db/matcher/expression_parser.cpp
index a06e07b82d7..91caf31241c 100644
--- a/src/mongo/db/matcher/expression_parser.cpp
+++ b/src/mongo/db/matcher/expression_parser.cpp
@@ -1741,6 +1741,15 @@ StatusWithMatchExpression MatchExpressionParser::parse(
     }
 }
 
+std::unique_ptr<MatchExpression> MatchExpressionParser::parseAndNormalize(
+    const BSONObj& obj,
+    const boost::intrusive_ptr<ExpressionContext>& expCtx,
+    const ExtensionsCallback& extensionsCallback,
+    AllowedFeatureSet allowedFeatures) {
+    auto parsedTree = uassertStatusOK(parse(obj, expCtx, extensionsCallback, allowedFeatures));
+    return MatchExpression::normalize(std::move(parsedTree));
+}
+
 namespace {
 // Maps from query operator string name to function.
 std::unique_ptr<StringMap<
diff --git a/src/mongo/db/matcher/expression_parser.h b/src/mongo/db/matcher/expression_parser.h
index b4048949055..c0b7691d108 100644
--- a/src/mongo/db/matcher/expression_parser.h
+++ b/src/mongo/db/matcher/expression_parser.h
@@ -124,5 +124,15 @@ public:
         const boost::intrusive_ptr<ExpressionContext>& expCtx,
         const ExtensionsCallback& extensionsCallback = ExtensionsCallbackNoop(),
         AllowedFeatureSet allowedFeatures = kDefaultSpecialFeatures);
+
+    /**
+     * Parse the given MatchExpression and normalize the resulting tree by optimizing and then
+     * sorting it. Throws if the given BSONObj fails to parse.
+     */
+    static std::unique_ptr<MatchExpression> parseAndNormalize(
+        const BSONObj& obj,
+        const boost::intrusive_ptr<ExpressionContext>& expCtx,
+        const ExtensionsCallback& extensionsCallback = ExtensionsCallbackNoop(),
+        AllowedFeatureSet allowedFeatures = kDefaultSpecialFeatures);
 };
 }  // namespace mongo
diff --git a/src/mongo/db/query/canonical_query.cpp b/src/mongo/db/query/canonical_query.cpp
index fdb8cc496ec..b27040895ba 100644
--- a/src/mongo/db/query/canonical_query.cpp
+++ b/src/mongo/db/query/canonical_query.cpp
@@ -46,60 +46,6 @@
 namespace mongo {
 namespace {
 
-/**
- * Comparator for MatchExpression nodes.  Returns an integer less than, equal to, or greater
- * than zero if 'lhs' is less than, equal to, or greater than 'rhs', respectively.
- *
- * Sorts by:
- * 1) operator type (MatchExpression::MatchType)
- * 2) path name (MatchExpression::path())
- * 3) sort order of children
- * 4) number of children (MatchExpression::numChildren())
- *
- * The third item is needed to ensure that match expression trees which should have the same
- * cache key always sort the same way. If you're wondering when the tuple (operator type, path
- * name) could ever be equal, consider this query:
- *
- * {$and:[{$or:[{a:1},{a:2}]},{$or:[{a:1},{b:2}]}]}
- *
- * The two OR nodes would compare as equal in this case were it not for tuple item #3 (sort
- * order of children).
- */
-int matchExpressionComparator(const MatchExpression* lhs, const MatchExpression* rhs) {
-    MatchExpression::MatchType lhsMatchType = lhs->matchType();
-    MatchExpression::MatchType rhsMatchType = rhs->matchType();
-    if (lhsMatchType != rhsMatchType) {
-        return lhsMatchType < rhsMatchType ? -1 : 1;
-    }
-
-    StringData lhsPath = lhs->path();
-    StringData rhsPath = rhs->path();
-    int pathsCompare = lhsPath.compare(rhsPath);
-    if (pathsCompare != 0) {
-        return pathsCompare;
-    }
-
-    const size_t numChildren = std::min(lhs->numChildren(), rhs->numChildren());
-    for (size_t childIdx = 0; childIdx < numChildren; ++childIdx) {
-        int childCompare =
-            matchExpressionComparator(lhs->getChild(childIdx), rhs->getChild(childIdx));
-        if (childCompare != 0) {
-            return childCompare;
-        }
-    }
-
-    if (lhs->numChildren() != rhs->numChildren()) {
-        return lhs->numChildren() < rhs->numChildren() ? -1 : 1;
-    }
-
-    // They're equal!
-    return 0;
-}
-
-bool matchExpressionLessThan(const MatchExpression* lhs, const MatchExpression* rhs) {
-    return matchExpressionComparator(lhs, rhs) < 0;
-}
-
 bool parsingCanProduceNoopMatchNodes(const ExtensionsCallback& extensionsCallback,
                                      MatchExpressionParser::AllowedFeatureSet allowedFeatures) {
     return extensionsCallback.hasNoopExtensions() &&
@@ -225,9 +171,8 @@ Status CanonicalQuery::init(OperationContext* opCtx,
 
     _canHaveNoopMatchNodes = canHaveNoopMatchNodes;
 
-    // Normalize, sort and validate tree.
-    _root = MatchExpression::optimize(std::move(root));
-    sortTree(_root.get());
+    // Normalize and validate tree.
+    _root = MatchExpression::normalize(std::move(root));
     auto validStatus = isValid(_root.get(), *_qr);
     if (!validStatus.isOK()) {
         return validStatus.getStatus();
@@ -335,17 +280,6 @@ bool CanonicalQuery::isSimpleIdQuery(const BSONObj& query) {
     return hasID;
 }
 
-// static
-void CanonicalQuery::sortTree(MatchExpression* tree) {
-    for (size_t i = 0; i < tree->numChildren(); ++i) {
-        sortTree(tree->getChild(i));
-    }
-    if (auto&& children = tree->getChildVector()) {
-        std::stable_sort(children->begin(), children->end(), matchExpressionLessThan);
-    }
-}
-
-// static
 size_t CanonicalQuery::countNodes(const MatchExpression* root, MatchExpression::MatchType type) {
     size_t sum = 0;
     if (type == root->matchType()) {
diff --git a/src/mongo/db/query/canonical_query.h b/src/mongo/db/query/canonical_query.h
index e1261805e20..ca5199e3502 100644
--- a/src/mongo/db/query/canonical_query.h
+++ b/src/mongo/db/query/canonical_query.h
@@ -191,12 +191,6 @@ public:
     std::string toStringShort() const;
 
     /**
-     * Traverses expression tree post-order.
-     * Sorts children at each non-leaf node by (MatchType, path(), children, number of children)
-     */
-    static void sortTree(MatchExpression* tree);
-
-    /**
      * Returns a count of 'type' nodes in expression tree.
      */
     static size_t countNodes(const MatchExpression* root, MatchExpression::MatchType type);
diff --git a/src/mongo/db/query/canonical_query_test.cpp b/src/mongo/db/query/canonical_query_test.cpp
index 1afb06988f7..7d159ae47f2 100644
--- a/src/mongo/db/query/canonical_query_test.cpp
+++ b/src/mongo/db/query/canonical_query_test.cpp
@@ -118,11 +118,11 @@ TEST(CanonicalQueryTest, IsValidSortKeyMetaProjection) {
 }
 
 //
-// Tests for CanonicalQuery::sortTree
+// Tests for MatchExpression::sortTree
 //
 
 /**
- * Helper function for testing CanonicalQuery::sortTree().
+ * Helper function for testing MatchExpression::sortTree().
  *
  * Verifies that sorting the expression 'unsortedQueryStr' yields an expression equivalent to
  * the expression 'sortedQueryStr'.
@@ -140,12 +140,12 @@ void testSortTree(const char* unsortedQueryStr, const char* sortedQueryStr) {
     // Sanity check that sorting the sorted expression is a no-op.
     {
         unique_ptr<MatchExpression> sortedQueryExprClone(parseMatchExpression(sortedQueryObj));
-        CanonicalQuery::sortTree(sortedQueryExprClone.get());
+        MatchExpression::sortTree(sortedQueryExprClone.get());
         assertEquivalent(unsortedQueryStr, sortedQueryExpr.get(), sortedQueryExprClone.get());
     }
 
     // Test that sorting the unsorted expression yields the sorted expression.
-    CanonicalQuery::sortTree(unsortedQueryExpr.get());
+    MatchExpression::sortTree(unsortedQueryExpr.get());
     assertEquivalent(unsortedQueryStr, unsortedQueryExpr.get(), sortedQueryExpr.get());
 }
 
diff --git a/src/mongo/db/query/query_planner_partialidx_test.cpp b/src/mongo/db/query/query_planner_partialidx_test.cpp
index ecb574fa728..46adc5869ab 100644
--- a/src/mongo/db/query/query_planner_partialidx_test.cpp
+++ b/src/mongo/db/query/query_planner_partialidx_test.cpp
@@ -502,5 +502,49 @@ TEST_F(QueryPlannerTest, InternalExprEqCannotUsePartialIndex) {
     assertNoSolutions();
 }
 
+TEST_F(QueryPlannerTest, MultipleOverlappingPartialIndexes) {
+    params.options = QueryPlannerParams::NO_TABLE_SCAN;
+
+    // Create two partial indexes with an overlapping range on the same key pattern.
+    BSONObj minus5To5 = fromjson("{a: {$gte: -5, $lte: 5}}");
+    BSONObj gte0 = fromjson("{a: {$gte: 0}}");
+    std::unique_ptr<MatchExpression> minus5To5Filter = parseMatchExpression(minus5To5);
+    std::unique_ptr<MatchExpression> gte0Filter = parseMatchExpression(gte0);
+    addIndex(fromjson("{a: 1}"), minus5To5Filter.get());
+    params.indices.back().identifier = CoreIndexInfo::Identifier{"minus5To5"};
+    addIndex(fromjson("{a: 1}"), gte0Filter.get());
+    params.indices.back().identifier = CoreIndexInfo::Identifier{"gte0"};
+
+    // Test that both indexes are eligible when the query falls within the intersecting range.
+    runQuery(fromjson("{a: 1}"));
+    assertNumSolutions(2U);
+    assertSolutionExists(
+        "{fetch: {filter: null, node: {ixscan: "
+        "{filter: null, pattern: {a: 1}, name: 'minus5To5', "
+        "bounds: {a: [[1, 1, true, true]]}}}}}");
+    assertSolutionExists(
+        "{fetch: {filter: null, node: {ixscan: "
+        "{filter: null, pattern: {a: 1}, name: 'gte0', "
+        "bounds: {a: [[1, 1, true, true]]}}}}}");
+
+    // Test that only a single index is eligible when the query falls within a single range.
+    runQuery(fromjson("{a: -5}"));
+    assertNumSolutions(1U);
+    assertSolutionExists(
+        "{fetch: {filter: null, node: {ixscan: "
+        "{filter: null, pattern: {a: 1}, name: 'minus5To5', "
+        "bounds: {a: [[-5, -5, true, true]]}}}}}");
+    runQuery(fromjson("{a: 10}"));
+    assertNumSolutions(1U);
+    assertSolutionExists(
+        "{fetch: {filter: null, node: {ixscan: "
+        "{filter: null, pattern: {a: 1}, name: 'gte0', "
+        "bounds: {a: [[10, 10, true, true]]}}}}}");
+
+    // Test that neither index is eligible when the query falls outside both ranges.
+    runInvalidQuery(fromjson("{a: -10}"));
+    assertNoSolutions();
+}
+
 }  // namespace
 }  // namespace mongo
diff --git a/src/mongo/db/query/query_planner_test_lib.cpp b/src/mongo/db/query/query_planner_test_lib.cpp
index 6082fc79e61..7e424c23c2a 100644
--- a/src/mongo/db/query/query_planner_test_lib.cpp
+++ b/src/mongo/db/query/query_planner_test_lib.cpp
@@ -82,9 +82,9 @@ bool filterMatches(const BSONObj& testFilter,
         return false;
     }
     const std::unique_ptr<MatchExpression> root = std::move(statusWithMatcher.getValue());
-    CanonicalQuery::sortTree(root.get());
+    MatchExpression::sortTree(root.get());
     std::unique_ptr<MatchExpression> trueFilter(trueFilterNode->filter->shallowClone());
-    CanonicalQuery::sortTree(trueFilter.get());
+    MatchExpression::sortTree(trueFilter.get());
     return trueFilter->equivalent(root.get());
 }
 
diff --git a/src/mongo/dbtests/plan_executor_invalidation_test.cpp b/src/mongo/dbtests/plan_executor_invalidation_test.cpp
index a7e949465bd..23acefa8019 100644
--- a/src/mongo/dbtests/plan_executor_invalidation_test.cpp
+++ b/src/mongo/dbtests/plan_executor_invalidation_test.cpp
@@ -38,6 +38,7 @@
 #include "mongo/db/dbdirectclient.h"
 #include "mongo/db/exec/collection_scan.h"
 #include "mongo/db/exec/plan_stage.h"
+#include "mongo/db/index/index_descriptor.h"
 #include "mongo/db/json.h"
 #include "mongo/db/matcher/expression_parser.h"
 #include "mongo/db/query/internal_plans.h"
@@ -100,9 +101,8 @@ public:
     std::unique_ptr<PlanExecutor, PlanExecutor::Deleter> makeIxscanPlan(BSONObj keyPattern,
                                                                         BSONObj startKey,
                                                                         BSONObj endKey) {
-        auto indexDescriptor =
-            collection()->getIndexCatalog()->findIndexByKeyPatternAndCollationSpec(
-                &_opCtx, keyPattern, {});
+        auto indexDescriptor = collection()->getIndexCatalog()->findIndexByKeyPatternAndOptions(
+            &_opCtx, keyPattern, _makeMinimalIndexSpec(keyPattern));
         ASSERT(indexDescriptor);
         return InternalPlanner::indexScan(&_opCtx,
                                           collection(),
@@ -134,6 +134,13 @@ public:
     DBDirectClient _client;
 
     boost::intrusive_ptr<ExpressionContext> _expCtx;
+
+private:
+    BSONObj _makeMinimalIndexSpec(BSONObj keyPattern) {
+        return BSON(IndexDescriptor::kKeyPatternFieldName
+                    << keyPattern << IndexDescriptor::kIndexVersionFieldName
+                    << IndexDescriptor::getDefaultIndexVersion());
+    }
 };
 
 TEST_F(PlanExecutorInvalidationTest, ExecutorToleratesDeletedDocumentsDuringYield) {
diff --git a/src/mongo/dbtests/query_stage_near.cpp b/src/mongo/dbtests/query_stage_near.cpp
index 938a98e2f88..2976c4cfeb1 100644
--- a/src/mongo/dbtests/query_stage_near.cpp
+++ b/src/mongo/dbtests/query_stage_near.cpp
@@ -69,12 +69,18 @@ public:
         _autoColl.emplace(_opCtx, NamespaceString{kTestNamespace});
         auto* coll = _autoColl->getCollection();
         ASSERT(coll);
-        _mockGeoIndex = coll->getIndexCatalog()->findIndexByKeyPatternAndCollationSpec(
-            _opCtx, kTestKeyPattern, BSONObj{});
+        _mockGeoIndex = coll->getIndexCatalog()->findIndexByKeyPatternAndOptions(
+            _opCtx, kTestKeyPattern, _makeMinimalIndexSpec(kTestKeyPattern));
         ASSERT(_mockGeoIndex);
     }
 
 protected:
+    BSONObj _makeMinimalIndexSpec(BSONObj keyPattern) {
+        return BSON(IndexDescriptor::kKeyPatternFieldName
+                    << keyPattern << IndexDescriptor::kIndexVersionFieldName
+                    << IndexDescriptor::getDefaultIndexVersion());
+    }
+
     const ServiceContext::UniqueOperationContext _uniqOpCtx = cc().makeOperationContext();
     OperationContext* const _opCtx = _uniqOpCtx.get();
     DBDirectClient directClient{_opCtx};