SERVER-43914 Permit sharding commands to accept compound hashed shard key specs

12 files changed, 691 insertions, 43 deletions

diff --git a/jstests/sharding/compound_hashed_shard_key_presplitting.js b/jstests/sharding/compound_hashed_shard_key_presplitting.js
new file mode 100644
index 00000000000..ef1b249bd1a
--- /dev/null
+++ b/jstests/sharding/compound_hashed_shard_key_presplitting.js
@@ -0,0 +1,106 @@
+/**
+ * Tests the pre-splitting behaviour of compound hashed shard key, for both the case where the
+ * prefix field is hashed, and where the hashed field is not the prefix.
+ *
+ * @tags: [requires_fcv_44]
+ */
+(function() {
+'use strict';
+const st = new ShardingTest({name: jsTestName(), shards: 2});
+const dbname = "test";
+const mongos = st.s0;
+const db = st.getDB(dbname);
+db.adminCommand({enablesharding: dbname});
+st.ensurePrimaryShard('test', st.shard1.shardName);
+
+/**
+ * Test that 'shardCollection' command works when there is existing data in collection and does not
+ * do pre-splitting.
+ */
+[{a: "hashed", rangeField1: 1, rangeField2: 1}, {rangeField1: 1, a: "hashed", rangeField2: 1}]
+    .forEach(function(shardKey) {
+        db.collWithData.drop();
+        db.collWithData.insert({a: 1});
+        db.collWithData.createIndex(shardKey);
+
+        // Verify that command fails when 'numInitialChunks' is specified.
+        assert.commandFailedWithCode(db.adminCommand({
+            shardcollection: db.collWithData.getFullName(),
+            key: shardKey,
+            numInitialChunks: 500
+        }),
+                                     ErrorCodes.InvalidOptions);
+
+        assert.commandWorked(
+            db.adminCommand({shardcollection: db.collWithData.getFullName(), key: shardKey}));
+        assert.eq(st.config.chunks.count({ns: db.collWithData.getFullName()}),
+                  1,
+                  "sharding non-empty collection should not pre-split");
+    });
+
+/**
+ * Validates that the chunks ranges have all the shard key fields and each shard has expected number
+ * of chunks.
+ */
+function checkValidChunks(coll, shardKey, expectedChunksOnShard0, expectedChunksOnShard1) {
+    const chunks = st.config.chunks.find({"ns": coll.getFullName()}).toArray();
+    let shardCountsMap = {[st.shard0.shardName]: 0, [st.shard1.shardName]: 0};
+    for (let chunk of chunks) {
+        shardCountsMap[chunk.shard]++;
+
+        const assertHasAllShardKeyFields = function(obj) {
+            assert.eq(Object.keys(shardKey).length, Object.keys(obj).length, tojson(obj));
+            for (let key in obj) {
+                assert(key in shardKey, tojson(obj));
+            }
+        };
+        assertHasAllShardKeyFields(chunk.min);
+        assertHasAllShardKeyFields(chunk.max);
+    }
+    assert.eq(expectedChunksOnShard0,
+              shardCountsMap[st.shard0.shardName],
+              'Count mismatch on shard0: ' + tojson(chunks));
+    assert.eq(expectedChunksOnShard1,
+              shardCountsMap[st.shard1.shardName],
+              'Count mismatch on shard1: ' + tojson(chunks));
+}
+
+function runTest(shardKey) {
+    // Supported: Hashed sharding + numInitialChunks + empty collection.
+    // Expected: Even chunk distribution.
+    assert.commandWorked(db.hashedCollEmpty.createIndex(shardKey));
+    let coll = db.hashedCollEmpty;
+    assert.commandWorked(mongos.adminCommand(
+        {shardCollection: coll.getFullName(), key: shardKey, numInitialChunks: 6}));
+    checkValidChunks(coll, shardKey, 3, 3);
+
+    // Supported: Hashed sharding + numInitialChunks + non-existent collection.
+    // Expected: Even chunk distribution.
+    coll = db.hashedCollNonExistent;
+    assert.commandWorked(mongos.adminCommand(
+        {shardCollection: coll.getFullName(), key: shardKey, numInitialChunks: 6}));
+    checkValidChunks(coll, shardKey, 3, 3);
+
+    // Default pre-splitting.
+    coll = db.hashedDefaultPreSplit;
+    assert.commandWorked(mongos.adminCommand({shardCollection: coll.getFullName(), key: shardKey}));
+    checkValidChunks(coll, shardKey, 2, 2);
+}
+
+/**
+ * Hashed field is a prefix.
+ */
+runTest({aKey: "hashed", rangeField1: 1, rangeField2: 1});
+
+/**
+ * When hashed field is not prefix.
+ * TODO SERVER-43917: Add tests when pre-splitting is enabled for non-prefixes.
+ */
+db.coll.drop();
+let shardKey = {rangeField1: 1, a: "hashed", rangeField2: 1};
+assert.commandFailedWithCode(
+    db.adminCommand({shardcollection: db.coll.getFullName(), key: shardKey, numInitialChunks: 500}),
+    ErrorCodes.InvalidOptions);
+
+st.stop();
+})();
\ No newline at end of file
diff --git a/jstests/sharding/compound_hashed_shard_key_sharding_cmds.js b/jstests/sharding/compound_hashed_shard_key_sharding_cmds.js
new file mode 100644
index 00000000000..f379f882d6c
--- /dev/null
+++ b/jstests/sharding/compound_hashed_shard_key_sharding_cmds.js
@@ -0,0 +1,165 @@
+/**
+ * Perform tests for moveChunk and splitChunk commands when the shard key is compound hashed.
+ *
+ * @tags: [requires_fcv_44]
+ */
+(function() {
+'use strict';
+
+const st = new ShardingTest({shards: 2, other: {chunkSize: 1}});
+const configDB = st.s0.getDB('config');
+assert.commandWorked(configDB.adminCommand({enableSharding: 'test'}));
+const shard0 = st.shard0.shardName;
+const shard1 = st.shard1.shardName;
+st.ensurePrimaryShard('test', shard0);
+const testDBOnPrimary = st.rs0.getPrimary().getDB('test');
+
+function verifyChunkSplitIntoTwo(namespace, chunk) {
+    assert.eq(0, configDB.chunks.count({ns: namespace, min: chunk.min, max: chunk.max}));
+    assert.eq(1, configDB.chunks.count({ns: namespace, min: chunk.min}));
+    assert.eq(1, configDB.chunks.count({ns: namespace, max: chunk.max}));
+}
+
+const nonHashedFieldValue = 111;
+const hashedFieldValue = convertShardKeyToHashed(nonHashedFieldValue);
+
+/**
+ * Returns an object which has all the shard key fields. The hashed field will have the value
+ * provided for 'valueForHashedField'. We are doing this because, the 'bounds' and 'middle'
+ * parameters of splitChunk/moveChunk commands expect a hashed value for the hashed field, where as
+ * 'find' expect a non-hashed value.
+ */
+function buildObjWithAllShardKeyFields(shardKey, valueForHashedField) {
+    let splitObj = {};
+    for (let key in shardKey) {
+        if (shardKey[key] === "hashed") {
+            splitObj[key] = valueForHashedField;
+        } else {
+            splitObj[key] = 1;
+        }
+    }
+    return splitObj;
+}
+
+function testSplit(shardKey, collName) {
+    const namespace = testDBOnPrimary[collName].getFullName();
+    assert.commandWorked(configDB.adminCommand({shardCollection: namespace, key: shardKey}));
+
+    // Insert data since both 'find' and 'bounds' based split requires the chunk to contain some
+    // documents.
+    const bulk = st.s.getDB("test")[collName].initializeUnorderedBulkOp();
+    for (let x = -1200; x < 1200; x++) {
+        bulk.insert({x: x, y: x, z: x});
+    }
+    assert.commandWorked(bulk.execute());
+
+    // Attempt to split on a value that is not the shard key.
+    assert.commandFailed(configDB.adminCommand({split: namespace, middle: {someField: 100}}));
+    assert.commandFailed(configDB.adminCommand({split: namespace, find: {someField: 100}}));
+    assert.commandFailed(configDB.adminCommand(
+        {split: namespace, bounds: [{someField: MinKey}, {someField: MaxKey}]}));
+
+    let totalChunksBefore = configDB.chunks.count({ns: namespace});
+    const lowestChunk = configDB.chunks.find({ns: namespace}).sort({min: 1}).limit(1).next();
+    assert(lowestChunk);
+    // Split the chunk based on 'bounds' and verify total chunks increased by one.
+    assert.commandWorked(
+        configDB.adminCommand({split: namespace, bounds: [lowestChunk.min, lowestChunk.max]}));
+    assert.eq(++totalChunksBefore, configDB.chunks.count({ns: namespace}));
+
+    // Verify that a single chunk with the previous bounds no longer exists but split into two.
+    verifyChunkSplitIntoTwo(namespace, lowestChunk);
+
+    // Cannot split if 'min' and 'max' doesn't correspond to the same chunk.
+    assert.commandFailed(
+        configDB.adminCommand({split: namespace, bounds: [lowestChunk.min, lowestChunk.max]}));
+
+    const splitObjWithHashedValue = buildObjWithAllShardKeyFields(shardKey, hashedFieldValue);
+
+    // Find the chunk to which 'splitObjWithHashedValue' belongs to.
+    let chunkToBeSplit = configDB.chunks.findOne(
+        {ns: namespace, min: {$lte: splitObjWithHashedValue}, max: {$gt: splitObjWithHashedValue}});
+    assert(chunkToBeSplit);
+
+    // Split the 'chunkToBeSplit' using 'find'. Note that the object specified for 'find' is not a
+    // split point.
+    const splitObj = buildObjWithAllShardKeyFields(shardKey, nonHashedFieldValue);
+    assert.commandWorked(configDB.adminCommand({split: namespace, find: splitObj}));
+    assert.eq(++totalChunksBefore, configDB.chunks.count({ns: namespace}));
+
+    // Verify that a single chunk with the previous bounds no longer exists but split into two.
+    verifyChunkSplitIntoTwo(namespace, chunkToBeSplit);
+    assert.eq(0, configDB.chunks.count({ns: namespace, min: splitObjWithHashedValue}));
+
+    // Get the new chunk in which 'splitObj' belongs.
+    chunkToBeSplit = configDB.chunks.findOne(
+        {ns: namespace, min: {$lte: splitObjWithHashedValue}, max: {$gt: splitObjWithHashedValue}});
+
+    // Use 'splitObj' as the middle point.
+    assert.commandWorked(
+        configDB.adminCommand({split: namespace, middle: splitObjWithHashedValue}));
+    assert.eq(++totalChunksBefore, configDB.chunks.count({ns: namespace}));
+    verifyChunkSplitIntoTwo(namespace, chunkToBeSplit);
+
+    // Cannot split on existing chunk boundary with 'middle'.
+    assert.commandFailed(configDB.adminCommand({split: namespace, middle: chunkToBeSplit.min}));
+
+    st.s.getDB("test")[collName].drop();
+}
+
+testSplit({x: "hashed", y: 1, z: 1}, "compound_hashed_prefix");
+testSplit({_id: "hashed", y: 1, z: 1}, "compound_hashed_prefix_id");
+testSplit({x: 1, y: "hashed", z: 1}, "compound_nonhashed_prefix");
+testSplit({x: 1, _id: "hashed", z: 1}, "compound_nonhashed_prefix_id");
+
+function testMoveChunk(shardKey) {
+    const ns = 'test.fooHashed';
+    assert.commandWorked(st.s0.adminCommand({shardCollection: ns, key: shardKey}));
+
+    // Fetch a chunk from 'shard0'.
+    const aChunk = configDB.chunks.findOne({ns: ns, shard: shard0});
+    assert(aChunk);
+
+    // Error if either of the bounds is not a valid shard key.
+    assert.commandFailedWithCode(
+        st.s0.adminCommand({moveChunk: ns, bounds: [NaN, aChunk.max], to: shard1}), 10065);
+    assert.commandFailedWithCode(
+        st.s0.adminCommand({moveChunk: ns, bounds: [aChunk.min, NaN], to: shard1}), 10065);
+
+    assert.commandWorked(
+        st.s0.adminCommand({moveChunk: ns, bounds: [aChunk.min, aChunk.max], to: shard1}));
+
+    assert.eq(0, configDB.chunks.count({_id: aChunk._id, shard: shard0}));
+    assert.eq(1, configDB.chunks.count({_id: aChunk._id, shard: shard1}));
+
+    // Fail if 'find' doesn't have full shard key.
+    assert.commandFailed(st.s0.adminCommand({moveChunk: ns, find: {someField: 0}, to: shard1}));
+
+    // Find the chunk to which 'moveObjWithHashedValue' belongs to.
+    const moveObjWithHashedValue = buildObjWithAllShardKeyFields(shardKey, hashedFieldValue);
+    const chunk = st.config.chunks.findOne(
+        {ns: ns, min: {$lte: moveObjWithHashedValue}, max: {$gt: moveObjWithHashedValue}});
+    assert(chunk);
+
+    // Verify that 'moveChunk' with 'find' works with pre-hashed value.
+    const otherShard = (chunk.shard === shard1) ? shard0 : shard1;
+    const moveObj = buildObjWithAllShardKeyFields(shardKey, nonHashedFieldValue);
+    assert.commandWorked(st.s.adminCommand({moveChunk: ns, find: moveObj, to: otherShard}));
+    assert.eq(st.config.chunks.count({ns: ns, min: chunk.min, shard: otherShard}), 1);
+
+    // Fail if 'find' and 'bounds' are both set.
+    assert.commandFailed(st.s0.adminCommand({
+        moveChunk: ns,
+        find: moveObjWithHashedValue,
+        bounds: [aChunk.min, aChunk.max],
+        to: shard1
+    }));
+
+    st.s.getDB("test").fooHashed.drop();
+}
+
+testMoveChunk({_id: "hashed", b: 1, c: 1});
+testMoveChunk({_id: 1, "b.c.d": "hashed", c: 1});
+
+st.stop();
+})();
diff --git a/jstests/sharding/compound_hashed_shard_key_zoning.js b/jstests/sharding/compound_hashed_shard_key_zoning.js
new file mode 100644
index 00000000000..c2efbb77d56
--- /dev/null
+++ b/jstests/sharding/compound_hashed_shard_key_zoning.js
@@ -0,0 +1,276 @@
+/**
+ * Test that updateZoneKeyRange command works correctly in combination with shardCollection command.
+ * In this test we verify the behaviour of:
+ *  - Creating zones after sharding the collection.
+ *  - Creating zones before sharding the collection.
+ *  - Creating zones in collection which has data and then sharding the collection.
+ *
+ * @tags: [requires_fcv_44]
+ */
+(function() {
+'use strict';
+
+const st = new ShardingTest({shards: 3});
+const kDbName = 'test';
+const kCollName = 'foo';
+const ns = kDbName + '.' + kCollName;
+const zoneName = 'zoneName';
+const mongos = st.s0;
+const testDB = mongos.getDB(kDbName);
+const configDB = mongos.getDB('config');
+const shardName = st.shard0.shardName;
+assert.commandWorked(mongos.adminCommand({enableSharding: kDbName}));
+assert.commandWorked(st.s.adminCommand({addShardToZone: shardName, zone: 'zoneName'}));
+
+function fillMissingShardKeyFields(shardKey, doc, value) {
+    for (let key in shardKey) {
+        if (!(key in doc)) {
+            doc[key] = value ? value : {"$minKey": 1};
+        }
+    }
+    return doc;
+}
+/**
+ * Test that 'updateZoneKeyRange' works correctly by verifying 'tags' collection, after sharding the
+ * collection.
+ */
+function testZoningAfterSharding(namespace, shardKey) {
+    assert.commandWorked(st.s.adminCommand({shardCollection: namespace, key: shardKey}));
+
+    // Testing basic assign.
+    assert.commandWorked(st.s.adminCommand(
+        {updateZoneKeyRange: namespace, min: {x: 0}, max: {x: 10}, zone: 'zoneName'}));
+
+    let tagDoc = configDB.tags.findOne();
+
+    assert.eq(namespace, tagDoc.ns);
+    assert.eq(fillMissingShardKeyFields(shardKey, {x: 0}), tagDoc.min);
+    assert.eq(fillMissingShardKeyFields(shardKey, {x: 10}), tagDoc.max);
+    assert.eq('zoneName', tagDoc.tag);
+
+    // Cannot assign overlapping ranges
+    assert.commandFailedWithCode(
+        st.s.adminCommand(
+            {updateZoneKeyRange: namespace, min: {x: -10}, max: {x: 20}, zone: 'zoneName'}),
+        ErrorCodes.RangeOverlapConflict);
+
+    // Cannot have non-shard key fields in tag range.
+    assert.commandFailedWithCode(st.s.adminCommand({
+        updateZoneKeyRange: namespace,
+        min: {newField: -10},
+        max: {newField: 20},
+        zone: 'zoneName'
+    }),
+                                 ErrorCodes.ShardKeyNotFound);
+
+    tagDoc = configDB.tags.findOne();
+    assert.eq(namespace, tagDoc.ns);
+    assert.eq(fillMissingShardKeyFields(shardKey, {x: 0}), tagDoc.min);
+    assert.eq(fillMissingShardKeyFields(shardKey, {x: 10}), tagDoc.max);
+    assert.eq('zoneName', tagDoc.tag);
+
+    // Testing basic remove.
+    const res = assert.commandWorked(st.s.adminCommand({
+        updateZoneKeyRange: namespace,
+        min: fillMissingShardKeyFields(shardKey, {x: 0}, MinKey),
+        max: fillMissingShardKeyFields(shardKey, {x: 10}, MinKey),
+        zone: null
+    }));
+    assert.eq(null, configDB.tags.findOne());
+}
+
+testZoningAfterSharding("test.compound_hashed", {x: 1, y: "hashed", z: 1});
+testZoningAfterSharding("test.compound_hashed_prefix", {x: "hashed", y: 1, z: 1});
+
+/**
+ * Test that shardCollection correctly validates shard key against existing zones.
+ */
+function testZoningBeforeSharding({shardKey, zoneRange, failCode}) {
+    assert.commandWorked(testDB.foo.createIndex(shardKey));
+    assert.commandWorked(st.s.adminCommand({addShardToZone: shardName, zone: zoneName}));
+
+    // Update zone range and verify that the 'tags' collection is updated appropriately.
+    assert.commandWorked(st.s.adminCommand(
+        {updateZoneKeyRange: ns, min: zoneRange[0], max: zoneRange[1], zone: zoneName}));
+    assert.eq(1, configDB.tags.count({ns: ns, min: zoneRange[0], max: zoneRange[1]}));
+
+    if (failCode) {
+        assert.commandFailedWithCode(mongos.adminCommand({shardCollection: ns, key: shardKey}),
+                                     failCode);
+    } else {
+        assert.commandWorked(mongos.adminCommand({shardCollection: ns, key: shardKey}));
+    }
+    assert.commandWorked(testDB.runCommand({drop: kCollName}));
+}
+
+// Fails when hashed field is not number long in 'zoneRange'.
+testZoningBeforeSharding(
+    {shardKey: {x: "hashed"}, zoneRange: [{x: -5}, {x: 5}], failCode: ErrorCodes.InvalidOptions});
+testZoningBeforeSharding({
+    shardKey: {x: "hashed"},
+    zoneRange: [{x: NumberLong(-5)}, {x: 5}],
+    failCode: ErrorCodes.InvalidOptions
+});
+testZoningBeforeSharding({
+    shardKey: {x: "hashed"},
+    zoneRange: [{x: -5}, {x: NumberLong(5)}],
+    failCode: ErrorCodes.InvalidOptions
+});
+testZoningBeforeSharding(
+    {shardKey: {x: "hashed"}, zoneRange: [{x: NumberLong(-5)}, {x: NumberLong(5)}]});
+testZoningBeforeSharding({
+    shardKey: {x: "hashed", y: 1},
+    zoneRange: [{x: NumberLong(-5), y: MinKey}, {x: NumberLong(5), y: MinKey}]
+});
+testZoningBeforeSharding({
+    shardKey: {x: 1, y: "hashed"},
+    zoneRange: [{x: 1, y: NumberLong(-5)}, {x: 2, y: NumberLong(5)}]
+});
+testZoningBeforeSharding({
+    shardKey: {x: 1, y: "hashed"},
+    zoneRange: [{x: 1, y: NumberLong(-5)}, {x: 2, y: 5}],
+    failCode: ErrorCodes.InvalidOptions
+});
+
+// Fails when 'zoneRange' doesn't have a shard key field.
+testZoningBeforeSharding({
+    shardKey: {x: 1, y: "hashed", z: 1},
+    zoneRange: [{x: 1, y: NumberLong(-5)}, {x: 2, y: NumberLong(5)}],
+    failCode: ErrorCodes.InvalidOptions
+});
+
+// Works when shard key field is defined as 'MinKey'.
+testZoningBeforeSharding({
+    shardKey: {x: 1, y: "hashed", z: 1},
+    zoneRange: [{x: 1, y: NumberLong(-5), z: MinKey}, {x: 2, y: NumberLong(5), z: MinKey}],
+});
+testZoningBeforeSharding(
+    {shardKey: {x: 1, y: "hashed"}, zoneRange: [{x: "DUB", y: MinKey}, {x: "NYC", y: MinKey}]});
+
+assert.commandWorked(st.s.adminCommand({removeShardFromZone: shardName, zone: zoneName}));
+
+/**
+ * Test that shardCollection uses existing zone ranges to split chunks.
+ */
+function testChunkSplits({collectionExists, shardKey, zoneRanges, expectedNumChunks}) {
+    const shards = configDB.shards.find().toArray();
+    if (collectionExists) {
+        assert.commandWorked(testDB.foo.createIndex(shardKey));
+    }
+
+    // Create a new zone and assign each zone to the shards using round-robin. Then update each of
+    // the zone's range to the range specified in 'zoneRanges'.
+    for (let i = 0; i < zoneRanges.length; i++) {
+        assert.commandWorked(
+            st.s.adminCommand({addShardToZone: shards[i % shards.length]._id, zone: zoneName + i}));
+        assert.commandWorked(st.s.adminCommand({
+            updateZoneKeyRange: ns,
+            min: zoneRanges[i][0],
+            max: zoneRanges[i][1],
+            zone: zoneName + i
+        }));
+    }
+    assert.eq(configDB.tags.count({ns: ns}), zoneRanges.length);
+    assert.eq(0,
+              configDB.chunks.count({ns: ns}),
+              "expect to see no chunk documents for the collection before shardCollection is run");
+
+    // Shard the collection and validate the resulting chunks.
+    assert.commandWorked(mongos.adminCommand({shardCollection: ns, key: shardKey}));
+    const chunkDocs = configDB.chunks.find({ns: ns}).toArray();
+    assert.eq(chunkDocs.length, expectedNumChunks, chunkDocs);
+
+    // Verify that each of the chunks corresponding to zones are in the right shard.
+    for (let i = 0; i < zoneRanges.length; i++) {
+        assert.eq(1,
+                  configDB.chunks.count({
+                      ns: ns,
+                      min: zoneRanges[i][0],
+                      max: zoneRanges[i][1],
+                      shard: shards[i % shards.length]._id
+                  }),
+                  chunkDocs);
+    }
+    assert.commandWorked(testDB.runCommand({drop: kCollName}));
+}
+
+// When shard key is compound hashed with range prefix.
+testChunkSplits({
+    shardKey: {x: 1, y: "hashed"},
+    zoneRanges: [
+        [{x: 0, y: MinKey}, {x: 5, y: MinKey}],
+        [{x: 10, y: MinKey}, {x: 15, y: MinKey}],
+        [{x: 20, y: MinKey}, {x: 25, y: MinKey}],
+        [{x: 30, y: MinKey}, {x: 35, y: MinKey}],
+    ],
+    expectedNumChunks: 9  // 4 zones + 2 boundaries + 3 gap chunks.
+});
+testChunkSplits({
+    shardKey: {x: 1, y: "hashed", z: 1},
+    zoneRanges: [
+        [{x: 0, y: NumberLong(0), z: MinKey}, {x: 5, y: NumberLong(0), z: MinKey}],
+        [{x: 10, y: NumberLong(0), z: MinKey}, {x: 15, y: NumberLong(0), z: MinKey}],
+        [{x: 20, y: NumberLong(0), z: MinKey}, {x: 25, y: NumberLong(0), z: MinKey}],
+        [{x: 30, y: NumberLong(0), z: MinKey}, {x: 35, y: NumberLong(0), z: MinKey}],
+    ],
+    expectedNumChunks: 9  // 4 zones + 2 boundaries + 3 gap chunks.
+});
+
+// When shard key is compound hashed with hashed prefix.
+testChunkSplits({
+    collectionExists: true,
+    shardKey: {x: "hashed", y: 1},
+    zoneRanges: [
+        [{x: NumberLong(0), y: MinKey}, {x: NumberLong(10), y: MinKey}],
+        [{x: NumberLong(10), y: MinKey}, {x: NumberLong(20), y: MinKey}],
+        [{x: NumberLong(20), y: MinKey}, {x: NumberLong(30), y: MinKey}],
+        [{x: NumberLong(30), y: MinKey}, {x: NumberLong(40), y: MinKey}],
+        [{x: NumberLong(40), y: MinKey}, {x: NumberLong(50), y: MinKey}],
+    ],
+    expectedNumChunks: 7  // 5 zones + 2 boundaries.
+});
+
+/**
+ * Tests that a non-empty collection associated with zones can be sharded.
+ */
+function testNonemptyZonedCollection() {
+    const shardKey = {x: 1, y: "hashed"};
+    const shards = configDB.shards.find().toArray();
+    const testColl = testDB.getCollection(kCollName);
+    const ranges = [
+        {min: {x: 0, y: MinKey}, max: {x: 10, y: MaxKey}},
+        {min: {x: 10, y: MaxKey}, max: {x: 20, y: MinKey}},
+        {min: {x: 20, y: MinKey}, max: {x: 40, y: MaxKey}}
+    ];
+
+    for (let i = 0; i < 40; i++) {
+        assert.commandWorked(testColl.insert({x: 1, y: Math.random()}));
+    }
+
+    assert.commandWorked(testColl.createIndex(shardKey));
+
+    for (let i = 0; i < shards.length; i++) {
+        assert.commandWorked(
+            mongos.adminCommand({addShardToZone: shards[i]._id, zone: zoneName + i}));
+        assert.commandWorked(mongos.adminCommand(
+            {updateZoneKeyRange: ns, min: ranges[i].min, max: ranges[i].max, zone: zoneName + i}));
+    }
+
+    assert.commandWorked(mongos.adminCommand({shardCollection: ns, key: shardKey}));
+
+    // Check that there is initially 1 chunk.
+    assert.eq(1, configDB.chunks.count({ns: ns}));
+
+    st.startBalancer();
+
+    // Check that the chunks were moved properly.
+    assert.soon(
+        () => configDB.chunks.count({ns: ns}) === 5, 'balancer never ran', 5 * 60 * 1000, 1000);
+
+    assert.commandWorked(testDB.runCommand({drop: kCollName}));
+}
+
+testNonemptyZonedCollection();
+
+st.stop();
+})();
diff --git a/jstests/sharding/refine_collection_shard_key_basic.js b/jstests/sharding/refine_collection_shard_key_basic.js
index 368c6cbf0e5..f7f0a7b15ff 100644
--- a/jstests/sharding/refine_collection_shard_key_basic.js
+++ b/jstests/sharding/refine_collection_shard_key_basic.js
@@ -285,7 +285,7 @@ assert.commandFailedWithCode(
 assert.commandFailedWithCode(
     mongos.adminCommand({refineCollectionShardKey: kNsName, key: {_id: -1}}), ErrorCodes.BadValue);
 assert.commandFailedWithCode(
-    mongos.adminCommand({refineCollectionShardKey: kNsName, key: {_id: 1, aKey: 'hashed'}}),
+    mongos.adminCommand({refineCollectionShardKey: kNsName, key: {_id: 'hashed', aKey: 'hashed'}}),
     ErrorCodes.BadValue);
 assert.commandFailedWithCode(
     mongos.adminCommand({refineCollectionShardKey: kNsName, key: {aKey: 'hahashed'}}),
@@ -303,6 +303,9 @@ assert.commandWorked(mongos.adminCommand({refineCollectionShardKey: kNsName, key
 dropAndReshardColl({aKey: 'hashed'});
 assert.commandWorked(
     mongos.adminCommand({refineCollectionShardKey: kNsName, key: {aKey: 'hashed'}}));
+dropAndReshardColl({_id: 1, aKey: 'hashed'});
+assert.commandWorked(
+    mongos.adminCommand({refineCollectionShardKey: kNsName, key: {_id: 1, aKey: 'hashed'}}));
 
 assert.commandWorked(mongos.getDB(kDbName).dropDatabase());
 
@@ -421,6 +424,24 @@ assert.commandWorked(mongos.getCollection(kNsName).insert({_id: 12345}));
 assert.commandWorked(
     mongos.adminCommand({refineCollectionShardKey: kNsName, key: {_id: 1, aKey: 1}}));
 
+// Should work because an index with missing or incomplete shard key entries exists for new shard
+// key {_id: "hashed", aKey: 1} and these entries are treated as null values.
+dropAndReshardColl({_id: "hashed"});
+assert.commandWorked(mongos.getCollection(kNsName).createIndex({_id: "hashed", aKey: 1}));
+assert.commandWorked(mongos.getCollection(kNsName).insert({_id: 12345}));
+
+assert.commandWorked(
+    mongos.adminCommand({refineCollectionShardKey: kNsName, key: {_id: "hashed", aKey: 1}}));
+
+// Should work because an index with missing or incomplete shard key entries exists for new shard
+// key {_id: 1, aKey: "hashed"} and these entries are treated as null values.
+dropAndReshardColl({_id: 1});
+assert.commandWorked(mongos.getCollection(kNsName).createIndex({_id: 1, aKey: "hashed"}));
+assert.commandWorked(mongos.getCollection(kNsName).insert({_id: 12345}));
+
+assert.commandWorked(
+    mongos.adminCommand({refineCollectionShardKey: kNsName, key: {_id: 1, aKey: "hashed"}}));
+
 // Should fail because new shard key {aKey: 1} is not a prefix of current shard key {_id: 1,
 // aKey: 1}.
 dropAndReshardColl({_id: 1, aKey: 1});
@@ -465,7 +486,7 @@ oldEpoch = mongos.getCollection(kConfigCollections).findOne({_id: kNsName}).last
 assert.commandWorked(
     mongos.adminCommand({refineCollectionShardKey: kNsName, key: {_id: 1, aKey: 1}}));
 validateConfigCollections({_id: 1, aKey: 1}, oldEpoch);
-validateConfigChangelog(3);
+validateConfigChangelog(5);
 
 // Should work because a 'useful' index exists for new shard key {a: 1, b.c: 1}. NOTE: We are
 // explicitly verifying that refineCollectionShardKey works with a dotted field.
@@ -476,7 +497,7 @@ oldEpoch = mongos.getCollection(kConfigCollections).findOne({_id: kNsName}).last
 assert.commandWorked(
     mongos.adminCommand({refineCollectionShardKey: kNsName, key: {a: 1, 'b.c': 1}}));
 validateConfigCollections({a: 1, 'b.c': 1}, oldEpoch);
-validateConfigChangelog(4);
+validateConfigChangelog(6);
 
 assert.commandWorked(mongos.getDB(kDbName).dropDatabase());
 
diff --git a/jstests/sharding/shard_collection_basic.js b/jstests/sharding/shard_collection_basic.js
index aa9be219496..445dab8a2e7 100644
--- a/jstests/sharding/shard_collection_basic.js
+++ b/jstests/sharding/shard_collection_basic.js
@@ -143,9 +143,13 @@ testAndClenaupWithKeyNoIndexFailed({x: 1, y: 1});
 assert.commandWorked(mongos.getDB(kDbName).foo.insert({x: 1, y: 1}));
 testAndClenaupWithKeyOK({x: 1, y: 1});
 
-testAndClenaupWithKeyNoIndexFailed({x: 'hashed', y: 1});
+// Multiple hashed fields are not allowed.
+testAndClenaupWithKeyNoIndexFailed({x: 'hashed', a: 1, y: 'hashed'});
 testAndClenaupWithKeyNoIndexFailed({x: 'hashed', y: 'hashed'});
 
+// Negative numbers are not allowed.
+testAndClenaupWithKeyNoIndexFailed({x: 'hashed', a: -1});
+
 // Shard by a key component.
 testAndClenaupWithKeyOK({'z.x': 1});
 testAndClenaupWithKeyOK({'z.x': 'hashed'});
diff --git a/src/mongo/db/s/config/configsvr_update_zone_key_range_command.cpp b/src/mongo/db/s/config/configsvr_update_zone_key_range_command.cpp
index 02d82cafa52..ebab9d7a4d6 100644
--- a/src/mongo/db/s/config/configsvr_update_zone_key_range_command.cpp
+++ b/src/mongo/db/s/config/configsvr_update_zone_key_range_command.cpp
@@ -106,17 +106,15 @@ public:
         auto parsedRequest =
             uassertStatusOK(UpdateZoneKeyRangeRequest::parseFromConfigCommand(cmdObj));
 
-        std::string zoneName;
-        if (!parsedRequest.isRemove()) {
-            zoneName = parsedRequest.getZoneName();
-        }
-
         if (parsedRequest.isRemove()) {
             uassertStatusOK(ShardingCatalogManager::get(opCtx)->removeKeyRangeFromZone(
                 opCtx, parsedRequest.getNS(), parsedRequest.getRange()));
         } else {
             uassertStatusOK(ShardingCatalogManager::get(opCtx)->assignKeyRangeToZone(
-                opCtx, parsedRequest.getNS(), parsedRequest.getRange(), zoneName));
+                opCtx,
+                parsedRequest.getNS(),
+                parsedRequest.getRange(),
+                parsedRequest.getZoneName()));
         }
 
         return true;
diff --git a/src/mongo/db/s/config/initial_split_policy.cpp b/src/mongo/db/s/config/initial_split_policy.cpp
index fe10ee8c6bf..cab4a2ca212 100644
--- a/src/mongo/db/s/config/initial_split_policy.cpp
+++ b/src/mongo/db/s/config/initial_split_policy.cpp
@@ -108,10 +108,12 @@ void InitialSplitPolicy::calculateHashedSplitPointsForEmptyCollection(
     int numInitialChunks,
     std::vector<BSONObj>* initialSplitPoints,
     std::vector<BSONObj>* finalSplitPoints) {
-    if (!shardKeyPattern.isHashedPattern() || !isEmpty) {
+    if (!shardKeyPattern.isHashedPattern() || !shardKeyPattern.hasHashedPrefix() || !isEmpty) {
+        // TODO SERVER-43917: Fix the error message when pre-splitting is enabled for non-hashed
+        // prefixes.
         uassert(ErrorCodes::InvalidOptions,
-                str::stream() << "numInitialChunks is not supported when the collection is not "
-                              << (!shardKeyPattern.isHashedPattern() ? "hashed" : "empty"),
+                str::stream() << "numInitialChunks is only supported when the collection is empty "
+                                 "and has a hashed field as shard key prefix",
                 !numInitialChunks);
         return;
     }
@@ -137,16 +139,21 @@ void InitialSplitPolicy::calculateHashedSplitPointsForEmptyCollection(
 
     const auto proposedKey(shardKeyPattern.getKeyPattern().toBSON());
 
+    auto buildSplitPoint = [&](long long value) {
+        return shardKeyPattern.getKeyPattern().extendRangeBound(
+            BSON(proposedKey.firstElementFieldName() << value), false);
+    };
+
     if (numInitialChunks % 2 == 0) {
-        finalSplitPoints->push_back(BSON(proposedKey.firstElementFieldName() << current));
+        finalSplitPoints->push_back(buildSplitPoint(current));
         current += intervalSize;
     } else {
         current += intervalSize / 2;
     }
 
     for (int i = 0; i < (numInitialChunks - 1) / 2; i++) {
-        finalSplitPoints->push_back(BSON(proposedKey.firstElementFieldName() << current));
-        finalSplitPoints->push_back(BSON(proposedKey.firstElementFieldName() << -current));
+        finalSplitPoints->push_back(buildSplitPoint(current));
+        finalSplitPoints->push_back(buildSplitPoint(-current));
         current += intervalSize;
     }
 
diff --git a/src/mongo/db/s/shardsvr_shard_collection.cpp b/src/mongo/db/s/shardsvr_shard_collection.cpp
index 448f3d1b451..fe1258f5a5a 100644
--- a/src/mongo/db/s/shardsvr_shard_collection.cpp
+++ b/src/mongo/db/s/shardsvr_shard_collection.cpp
@@ -348,15 +348,16 @@ void validateShardKeyAgainstExistingZones(OperationContext* opCtx,
                                   << tag.getMinKey() << " -->> " << tag.getMaxKey(),
                     match);
 
-            if (ShardKeyPattern::isHashedPatternEl(proposedKeyElement) &&
-                (tagMinKeyElement.type() != NumberLong || tagMaxKeyElement.type() != NumberLong)) {
-                uasserted(ErrorCodes::InvalidOptions,
-                          str::stream() << "cannot do hash sharding with the proposed key "
-                                        << proposedKey.toString() << " because there exists a zone "
-                                        << tag.getMinKey() << " -->> " << tag.getMaxKey()
-                                        << " whose boundaries are not "
-                                           "of type NumberLong");
-            }
+            // If the field is hashed, make sure that the min and max values are of supported type.
+            uassert(
+                ErrorCodes::InvalidOptions,
+                str::stream() << "cannot do hash sharding with the proposed key "
+                              << proposedKey.toString() << " because there exists a zone "
+                              << tag.getMinKey() << " -->> " << tag.getMaxKey()
+                              << " whose boundaries are not of type NumberLong, MinKey or MaxKey",
+                !ShardKeyPattern::isHashedPatternEl(proposedKeyElement) ||
+                    (ShardKeyPattern::isValidHashedValue(tagMinKeyElement) &&
+                     ShardKeyPattern::isValidHashedValue(tagMaxKeyElement)));
         }
     }
 }
diff --git a/src/mongo/db/s/split_chunk.cpp b/src/mongo/db/s/split_chunk.cpp
index ec8d3e9b530..5230be82496 100644
--- a/src/mongo/db/s/split_chunk.cpp
+++ b/src/mongo/db/s/split_chunk.cpp
@@ -147,21 +147,18 @@ StatusWith<boost::optional<ChunkRange>> splitChunk(OperationContext* opCtx,
                           << " to split chunk " << chunkRange.toString());
     }
 
-    // If the shard key is hashed, then we must make sure that the split points are of type
-    // NumberLong.
-    if (KeyPattern::isHashedKeyPattern(keyPatternObj)) {
+    // If the shard key is hashed, then we must make sure that the split points are of supported
+    // data types.
+    const auto hashedField = ShardKeyPattern::extractHashedField(keyPatternObj);
+    if (hashedField) {
         for (BSONObj splitKey : splitKeys) {
-            BSONObjIterator it(splitKey);
-            while (it.more()) {
-                BSONElement splitKeyElement = it.next();
-                if (splitKeyElement.type() != NumberLong) {
-                    return {ErrorCodes::CannotSplit,
-                            str::stream()
-                                << "splitChunk cannot split chunk " << chunkRange.toString()
-                                << ", split point " << splitKeyElement.toString()
-                                << " must be of type "
-                                   "NumberLong for hashed shard key patterns"};
-                }
+            auto hashedSplitElement = splitKey[hashedField.fieldName()];
+            if (!ShardKeyPattern::isValidHashedValue(hashedSplitElement)) {
+                return {ErrorCodes::CannotSplit,
+                        str::stream() << "splitChunk cannot split chunk " << chunkRange.toString()
+                                      << ", split point " << hashedSplitElement.toString()
+                                      << "Value of type '" << hashedSplitElement.type()
+                                      << "' is not allowed for hashed fields"};
             }
         }
     }
diff --git a/src/mongo/s/shard_key_pattern.cpp b/src/mongo/s/shard_key_pattern.cpp
index 5e1773ce391..1a6093825ca 100644
--- a/src/mongo/s/shard_key_pattern.cpp
+++ b/src/mongo/s/shard_key_pattern.cpp
@@ -66,6 +66,7 @@ std::vector<std::unique_ptr<FieldRef>> parseShardKeyPattern(const BSONObj& keyPa
 
     std::vector<std::unique_ptr<FieldRef>> parsedPaths;
 
+    auto numHashedFields = 0;
     for (const auto& patternEl : keyPattern) {
         auto newFieldRef(std::make_unique<FieldRef>(patternEl.fieldNameStringData()));
 
@@ -88,7 +89,9 @@ std::vector<std::unique_ptr<FieldRef>> parseShardKeyPattern(const BSONObj& keyPa
                     !newFieldRef->getPart(i).empty());
         }
 
-        // Numeric and ascending (1.0), or "hashed" and single field
+        // Numeric and ascending (1.0), or "hashed" with exactly hashed field.
+        auto isHashedPattern = ShardKeyPattern::isHashedPatternEl(patternEl);
+        numHashedFields += isHashedPattern ? 1 : 0;
         uassert(ErrorCodes::BadValue,
                 str::stream()
                     << "Shard key " << keyPattern.toString()
@@ -96,7 +99,7 @@ std::vector<std::unique_ptr<FieldRef>> parseShardKeyPattern(const BSONObj& keyPa
                     << " or multiple numerical fields set to a value of 1. Failed to parse field "
                     << patternEl.fieldNameStringData(),
                 (patternEl.isNumber() && patternEl.numberInt() == 1) ||
-                    (keyPattern.nFields() == 1 && ShardKeyPattern::isHashedPatternEl(patternEl)));
+                    (isHashedPattern && numHashedFields == 1));
 
         parsedPaths.emplace_back(std::move(newFieldRef));
     }
@@ -173,10 +176,19 @@ Status ShardKeyPattern::checkShardKeyIsValidForMetadataStorage(const BSONObj& sh
     return Status::OK();
 }
 
+BSONElement ShardKeyPattern::extractHashedField(BSONObj keyPattern) {
+    for (auto&& element : keyPattern) {
+        if (isHashedPatternEl(element)) {
+            return element;
+        }
+    }
+    return BSONElement();
+}
 ShardKeyPattern::ShardKeyPattern(const BSONObj& keyPattern)
     : _keyPattern(keyPattern),
       _keyPatternPaths(parseShardKeyPattern(keyPattern)),
-      _hasId(keyPattern.hasField("_id"_sd)) {}
+      _hasId(keyPattern.hasField("_id"_sd)),
+      _hashedField(extractHashedField(keyPattern)) {}
 
 ShardKeyPattern::ShardKeyPattern(const KeyPattern& keyPattern)
     : ShardKeyPattern(keyPattern.toBSON()) {}
@@ -186,9 +198,30 @@ bool ShardKeyPattern::isHashedPatternEl(const BSONElement& el) {
 }
 
 bool ShardKeyPattern::isHashedPattern() const {
+    return !_hashedField.eoo();
+}
+
+bool ShardKeyPattern::isValidHashedValue(const BSONElement& el) {
+    switch (el.type()) {
+        case MinKey:
+        case MaxKey:
+        case NumberLong:
+            return true;
+        default:
+            return false;
+    }
+    MONGO_UNREACHABLE;
+}
+
+
+bool ShardKeyPattern::hasHashedPrefix() const {
     return isHashedPatternEl(_keyPattern.toBSON().firstElement());
 }
 
+BSONElement ShardKeyPattern::getHashedField() const {
+    return _hashedField;
+}
+
 const KeyPattern& ShardKeyPattern::getKeyPattern() const {
     return _keyPattern;
 }
@@ -345,7 +378,7 @@ BSONObj ShardKeyPattern::extractShardKeyFromQuery(const CanonicalQuery& query) c
         if (!isValidShardKeyElementForStorage(equalEl))
             return BSONObj();
 
-        if (isHashedPattern()) {
+        if (_hashedField && _hashedField.fieldNameStringData() == patternPath.dottedField()) {
             keyBuilder.append(
                 patternPath.dottedField(),
                 BSONElementHasher::hash64(equalEl, BSONElementHasher::DEFAULT_HASH_SEED));
diff --git a/src/mongo/s/shard_key_pattern.h b/src/mongo/s/shard_key_pattern.h
index 91eee10080a..c5baaabc407 100644
--- a/src/mongo/s/shard_key_pattern.h
+++ b/src/mongo/s/shard_key_pattern.h
@@ -88,8 +88,23 @@ public:
      */
     static bool isHashedPatternEl(const BSONElement& el);
 
+    /**
+     * Returns the BSONElement pointing to the hashed field. Returns empty BSONElement if not found.
+     */
+    static BSONElement extractHashedField(BSONObj keyPattern);
+
+    /**
+     * Check if the given BSONElement is of type 'MinKey', 'MaxKey' or 'NumberLong', which are the
+     * only acceptable values for hashed fields.
+     */
+    static bool isValidHashedValue(const BSONElement& el);
+
     bool isHashedPattern() const;
 
+    bool hasHashedPrefix() const;
+
+    BSONElement getHashedField() const;
+
     const KeyPattern& getKeyPattern() const;
 
     const std::vector<std::unique_ptr<FieldRef>>& getKeyPatternFields() const;
@@ -255,6 +270,7 @@ private:
     std::vector<std::unique_ptr<FieldRef>> _keyPatternPaths;
 
     bool _hasId;
+    BSONElement _hashedField;
 };
 
 }  // namespace mongo
diff --git a/src/mongo/s/shard_key_pattern_test.cpp b/src/mongo/s/shard_key_pattern_test.cpp
index fd51f371f22..a1196e36e17 100644
--- a/src/mongo/s/shard_key_pattern_test.cpp
+++ b/src/mongo/s/shard_key_pattern_test.cpp
@@ -433,6 +433,19 @@ TEST(ShardKeyPattern, ExtractQueryShardKeyHashed) {
     ASSERT_BSONOBJ_EQ(queryKey(pattern, BSON("a" << BSON("c" << value))), BSONObj());
     ASSERT_BSONOBJ_EQ(queryKey(pattern, BSON("a" << BSON("b" << BSON_ARRAY(value)))), BSONObj());
     ASSERT_BSONOBJ_EQ(queryKey(pattern, BSON("a" << BSON_ARRAY(BSON("b" << value)))), BSONObj());
+
+    pattern = ShardKeyPattern(BSON("a.b"
+                                   << "hashed"
+                                   << "c.d" << 1));
+
+    ASSERT_BSONOBJ_EQ(queryKey(pattern, BSON("a.b" << value << "c.d" << value)),
+                      BSON("a.b" << hashValue << "c.d" << value));
+    ASSERT_BSONOBJ_EQ(
+        queryKey(pattern, fromjson("{a : {b: '12345', p : 1}, c : {d : '12345', q: 2}}")),
+        BSON("a.b" << hashValue << "c.d" << value));
+
+    ASSERT_BSONOBJ_EQ(queryKey(pattern, BSON("a.b" << value)), BSONObj());
+    ASSERT_BSONOBJ_EQ(queryKey(pattern, fromjson("{'a.b': [10], 'c.d': 1}")), BSONObj());
 }
 
 static bool indexComp(const ShardKeyPattern& pattern, const BSONObj& indexPattern) {
@@ -513,5 +526,16 @@ TEST(ShardKeyPattern, UniqueIndexCompatibleHashed) {
     ASSERT(!indexComp(pattern, BSON("c" << -1 << "a.b" << 1)));
 }
 
+TEST(ShardKeyPattern, IsHashedPattern) {
+    ASSERT(ShardKeyPattern(BSON("a.b"
+                                << "hashed"))
+               .isHashedPattern());
+    ASSERT(ShardKeyPattern(BSON("a.b" << 1 << "c"
+                                      << "hashed"
+                                      << "d" << 1))
+               .isHashedPattern());
+    ASSERT(!ShardKeyPattern(BSON("a.b" << 1 << "d" << 1)).isHashedPattern());
+}
+
 }  // namespace
 }  // namespace mongo