SERVER-43917 Support pre-splitting compound hashed shard keys with 'presplitHashedZones' flag

9 files changed, 2324 insertions, 638 deletions

diff --git a/jstests/sharding/compound_hashed_shard_key_presplitting.js b/jstests/sharding/compound_hashed_shard_key_presplitting.js
index 0f994193431..f37b1a6c97c 100644
--- a/jstests/sharding/compound_hashed_shard_key_presplitting.js
+++ b/jstests/sharding/compound_hashed_shard_key_presplitting.js
@@ -6,7 +6,7 @@
  */
 (function() {
 'use strict';
-const st = new ShardingTest({name: jsTestName(), shards: 2});
+const st = new ShardingTest({name: jsTestName(), shards: 3});
 const dbname = "test";
 const mongos = st.s0;
 const db = st.getDB(dbname);
@@ -39,12 +39,13 @@ st.ensurePrimaryShard('test', st.shard1.shardName);
     });
 
 /**
- * Validates that the chunks ranges have all the shard key fields and each shard has expected number
- * of chunks.
+ * Helper function to validate that the chunks ranges have all the shard key fields and each shard
+ * has expected number of chunks.
  */
-function checkValidChunks(coll, shardKey, expectedChunksOnShard0, expectedChunksOnShard1) {
+function checkValidChunks(coll, shardKey, expectedChunks) {
     const chunks = st.config.chunks.find({"ns": coll.getFullName()}).toArray();
-    let shardCountsMap = {[st.shard0.shardName]: 0, [st.shard1.shardName]: 0};
+    let shardCountsMap =
+        {[st.shard0.shardName]: 0, [st.shard1.shardName]: 0, [st.shard2.shardName]: 0};
     for (let chunk of chunks) {
         shardCountsMap[chunk.shard]++;
 
@@ -57,50 +58,291 @@ function checkValidChunks(coll, shardKey, expectedChunksOnShard0, expectedChunks
         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));
+    let index = 0;
+    for (let shardName in shardCountsMap) {
+        assert.eq(expectedChunks[index++],
+                  shardCountsMap[shardName],
+                  "Expected chunks did not match for " + shardName + ". " + 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);
+//
+// Test cases for compound hashed shard keys with hashed prefix.
+//
+let shardKey = {hashedField: "hashed", rangeField1: 1, rangeField2: 1};
+
+// 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, [2, 2, 2]);
+
+// Supported: Hashed sharding + numInitialChunks + non-existent collection.
+// Expected: Even chunk distribution and the remainder chunks on the first shard.
+coll = db.hashedCollNonExistent;
+assert.commandWorked(
+    mongos.adminCommand({shardCollection: coll.getFullName(), key: shardKey, numInitialChunks: 8}));
+checkValidChunks(coll, shardKey, [4, 2, 2]);
+
+// When 'numInitialChunks' is one, primary shard should have the chunk.
+coll = db.hashedNumInitialChunksOne;
+assert.commandWorked(
+    mongos.adminCommand({shardCollection: coll.getFullName(), key: shardKey, numInitialChunks: 1}));
+checkValidChunks(coll, shardKey, [0, 1, 0]);
+
+// Default pre-splitting assigns two chunks per shard.
+coll = db.hashedDefaultPreSplit;
+assert.commandWorked(mongos.adminCommand({shardCollection: coll.getFullName(), key: shardKey}));
+checkValidChunks(coll, shardKey, [2, 2, 2]);
+
+db.hashedPrefixColl.drop();
+
+// 'presplitHashedZones' cannot be passed without setting up zones.
+assert.commandFailedWithCode(db.adminCommand({
+    shardcollection: db.hashedPrefixColl.getFullName(),
+    key: shardKey,
+    numInitialChunks: 500,
+    presplitHashedZones: true
+}),
+                             31387);
+
+// Verify that 'shardCollection' command will fail if the zones are set up incorrectly.
+assert.commandWorked(
+    st.s.adminCommand({addShardToZone: st.shard2.shardName, zone: 'hashedPrefix'}));
+assert.commandWorked(st.s.adminCommand({
+    updateZoneKeyRange: db.hashedPrefixColl.getFullName(),
+    min: {hashedField: MinKey, rangeField1: MinKey, rangeField2: MinKey},
+    max: {
+        hashedField: MaxKey,
+        rangeField1: MaxKey,
+        rangeField2: MinKey
+    },  // All fields should be MaxKey for a valid zone.
+    zone: 'hashedPrefix'
+}));
+assert.commandFailedWithCode(db.adminCommand({
+    shardcollection: db.hashedPrefixColl.getFullName(),
+    key: shardKey,
+    presplitHashedZones: true
+}),
+                             31412);
+
+// 'numInitialChunks' is ignored when zones are present and 'presplitHashedZones' is not set.
+// Creates chunks based on the zones.
+assert.commandWorked(db.adminCommand(
+    {shardcollection: db.hashedPrefixColl.getFullName(), key: shardKey, numInitialChunks: 2}));
+checkValidChunks(
+    db.hashedPrefixColl, shardKey, [1 /* Boundary chunk */, 0, 1 /* zone present in shard2 */]);
+
+// Verify that 'shardCollection' command will pre-split chunks if a single zone is set up ranging
+// from MinKey to MaxKey and 'presplitHashedZones' flag is set.
+db.hashedPrefixColl.drop();
+assert.commandWorked(
+    st.s.adminCommand({addShardToZone: st.shard1.shardName, zone: 'hashedPrefix'}));
+assert.commandWorked(
+    st.s.adminCommand({addShardToZone: st.shard2.shardName, zone: 'hashedPrefix'}));
+assert.commandWorked(st.s.adminCommand({
+    updateZoneKeyRange: db.hashedPrefixColl.getFullName(),
+    min: {hashedField: MinKey, rangeField1: MinKey, rangeField2: MinKey},
+    max: {hashedField: MaxKey, rangeField1: MaxKey, rangeField2: MaxKey},
+    zone: 'hashedPrefix'
+}));
+assert.commandWorked(db.adminCommand({
+    shardcollection: db.hashedPrefixColl.getFullName(),
+    key: shardKey,
+    presplitHashedZones: true
+}));
+
+// By default, we create two chunks per shard for each shard that contains at least one zone.
+checkValidChunks(db.hashedPrefixColl, shardKey, [0, 2, 2]);
+
+// Verify that 'shardCollection' command will pre-split chunks equally among all the eligible
+// shards.
+db.hashedPrefixColl.drop();
+assert.commandWorked(
+    st.s.adminCommand({addShardToZone: st.shard0.shardName, zone: 'hashedPrefix'}));
+assert.commandWorked(
+    st.s.adminCommand({addShardToZone: st.shard1.shardName, zone: 'hashedPrefix'}));
+assert.commandWorked(
+    st.s.adminCommand({addShardToZone: st.shard2.shardName, zone: 'hashedPrefix'}));
+
+assert.commandWorked(st.s.adminCommand({
+    updateZoneKeyRange: db.hashedPrefixColl.getFullName(),
+    min: {hashedField: MinKey, rangeField1: MinKey, rangeField2: MinKey},
+    max: {hashedField: MaxKey, rangeField1: MaxKey, rangeField2: MaxKey},
+    zone: 'hashedPrefix'
+}));
+assert.commandWorked(db.adminCommand({
+    shardcollection: db.hashedPrefixColl.getFullName(),
+    key: shardKey,
+    presplitHashedZones: true,
+    numInitialChunks: 100
+}));
+checkValidChunks(db.hashedPrefixColl, shardKey, [34, 34, 34]);
+
+//
+// Test cases for compound hashed shard keys with non-hashed prefix.
+//
+
+/**
+ * Helper function to create two non-overlapping interweaving tag ranges for each of the specified
+ * zones.
+ */
+function createZoneRanges(coll, zoneNames) {
+    let offsetFirstTag = 0, offsetSecondTag = 0;
+
+    // Create zone ranges such that first zone has tag ranges [a, b) and [A, B). Second zone has tag
+    // ranges [c, d) and [C, D). Third zone has tag ranges [e, f) and [E, f) so on.
+    for (let zoneName of zoneNames) {
+        assert.commandWorked(st.s.adminCommand({
+            updateZoneKeyRange: coll.getFullName(),
+            min: {
+                rangeField1: String.fromCharCode(97 + offsetFirstTag++),
+                hashedField: MinKey,
+                rangeField2: MinKey
+            },
+            max: {
+                rangeField1: String.fromCharCode(97 + offsetFirstTag++),
+                hashedField: MinKey,
+                rangeField2: MinKey
+            },
+            zone: zoneName
+        }));
+
+        assert.commandWorked(st.s.adminCommand({
+            updateZoneKeyRange: coll.getFullName(),
+            min: {
+                rangeField1: String.fromCharCode(65 + offsetSecondTag++),
+                hashedField: MinKey,
+                rangeField2: MinKey
+            },
+            max: {
+                rangeField1: String.fromCharCode(65 + offsetSecondTag++),
+                hashedField: MinKey,
+                rangeField2: MinKey
+            },
+            zone: zoneName
+        }));
+    }
 }
 
 /**
- * Hashed field is a prefix.
+ * Helper function to set up two zones named 'nonHashedPrefix1' and 'nonHashedPrefix2' on 'shard0'.
  */
-runTest({aKey: "hashed", rangeField1: 1, rangeField2: 1});
+function setUpTwoZonesOnShard0(coll) {
+    assert.commandWorked(
+        st.s.adminCommand({addShardToZone: st.shard0.shardName, zone: 'nonHashedPrefix1'}));
+    assert.commandWorked(
+        st.s.adminCommand({addShardToZone: st.shard0.shardName, zone: 'nonHashedPrefix2'}));
+    assert.commandWorked(
+        st.s.adminCommand({removeShardFromZone: st.shard1.shardName, zone: 'nonHashedPrefix1'}));
+    assert.commandWorked(
+        st.s.adminCommand({removeShardFromZone: st.shard1.shardName, zone: 'nonHashedPrefix2'}));
+    assert.commandWorked(
+        st.s.adminCommand({removeShardFromZone: st.shard2.shardName, zone: 'nonHashedPrefix2'}));
+
+    createZoneRanges(coll, ['nonHashedPrefix1', 'nonHashedPrefix2']);
+}
 
 /**
- * When hashed field is not prefix.
- * TODO SERVER-43917: Add tests when pre-splitting is enabled for non-prefixes.
+ * Helper function to set up two zones such that 'nonHashedPrefix1' zone is assigned to shard0 and
+ * shard1. 'nonHashedPrefix2' is assinged to shard1 and shard2.
  */
+function setUpTwoZones(coll) {
+    assert.commandWorked(
+        st.s.adminCommand({removeShardFromZone: st.shard0.shardName, zone: 'nonHashedPrefix2'}));
+    assert.commandWorked(
+        st.s.adminCommand({addShardToZone: st.shard0.shardName, zone: 'nonHashedPrefix1'}));
+    assert.commandWorked(
+        st.s.adminCommand({addShardToZone: st.shard1.shardName, zone: 'nonHashedPrefix1'}));
+    assert.commandWorked(
+        st.s.adminCommand({addShardToZone: st.shard1.shardName, zone: 'nonHashedPrefix2'}));
+    assert.commandWorked(
+        st.s.adminCommand({addShardToZone: st.shard2.shardName, zone: 'nonHashedPrefix2'}));
+    createZoneRanges(coll, ['nonHashedPrefix1', 'nonHashedPrefix2']);
+}
+
+shardKey = {
+    rangeField1: 1,
+    hashedField: "hashed",
+    rangeField2: 1
+};
 db.coll.drop();
-let shardKey = {rangeField1: 1, a: "hashed", rangeField2: 1};
+setUpTwoZonesOnShard0(db.coll);
+
+// 'numInitialChunks' cannot be passed without 'presplitHashedZones'.
 assert.commandFailedWithCode(
     db.adminCommand({shardcollection: db.coll.getFullName(), key: shardKey, numInitialChunks: 500}),
     ErrorCodes.InvalidOptions);
 
+db.coll.drop();
+// 'presplitHashedZones' cannot be passed without setting up zones.
+assert.commandFailedWithCode(db.adminCommand({
+    shardcollection: db.coll.getFullName(),
+    key: shardKey,
+    numInitialChunks: 500,
+    presplitHashedZones: true
+}),
+                             31387);
+
+// Verify that shardCollection command will fail if the zones are set up incorrectly.
+assert.commandWorked(
+    st.s.adminCommand({addShardToZone: st.shard0.shardName, zone: 'nonHashedPrefix1'}));
+assert.commandWorked(st.s.adminCommand({
+    updateZoneKeyRange: db.coll.getFullName(),
+    min: {rangeField1: "A", hashedField: MinKey, rangeField2: MinKey},
+    max: {rangeField1: "A", hashedField: MaxKey, rangeField2: MinKey},
+    zone: 'nonHashedPrefix1'
+}));
+assert.commandFailedWithCode(db.adminCommand({
+    shardcollection: db.coll.getFullName(),
+    key: shardKey,
+    numInitialChunks: 500,
+    presplitHashedZones: true
+}),
+                             31390);
+
+// Verify that 'presplitHashedZones' with 'numInitialChunks' works correctly when zones are set up.
+db.coll.drop();
+setUpTwoZones(db.coll);
+assert.commandWorked(db.adminCommand({
+    shardcollection: db.coll.getFullName(),
+    key: shardKey,
+    numInitialChunks: 500,
+    presplitHashedZones: true
+}));
+
+// We need to create ceil(500/3) = 167 chunks per shard. Shard1 has 4 tags(2 per zone) while others
+// shards have 2 tags. So we create ceil(167/4) = 42 per tag on shard1 = 168, while we create
+// ceil(167/2) = 84 per tag on others. In addition, we create 5 chunks for boundaries which will be
+// distributed among the three shards using round robin.
+checkValidChunks(db.coll, shardKey, [170, 170, 169]);
+
+// When 'numInitialChunks = 1'.
+db.coll.drop();
+setUpTwoZones(db.coll);
+assert.commandWorked(db.adminCommand({
+    shardcollection: db.coll.getFullName(),
+    key: shardKey,
+    numInitialChunks: 1,
+    presplitHashedZones: true
+}));
+
+// The chunk distribution from zones should be [2, 2+2 (two zones), 2]. The 5 gap chunks should be
+// distributed among three shards.
+checkValidChunks(db.coll, shardKey, [4, 6, 3]);
+
+// Verify that 'presplitHashedZones' uses default value of two per shard when 'numInitialChunks' is
+// not passed.
+db.coll.drop();
+setUpTwoZonesOnShard0(db.coll);
+assert.commandWorked(db.adminCommand(
+    {shardcollection: db.coll.getFullName(), key: shardKey, presplitHashedZones: true}));
+
+// Since only Shard0 has chunks, we create on chunk per tag on shard0. The 5 gap chunks should be
+// distributed among three shards.
+checkValidChunks(db.coll, shardKey, [6, 2, 1]);
+
 st.stop();
 })();
\ No newline at end of file
diff --git a/src/mongo/db/s/config/configsvr_shard_collection_command.cpp b/src/mongo/db/s/config/configsvr_shard_collection_command.cpp
index f145d3a85e6..4fca10fda03 100644
--- a/src/mongo/db/s/config/configsvr_shard_collection_command.cpp
+++ b/src/mongo/db/s/config/configsvr_shard_collection_command.cpp
@@ -337,6 +337,7 @@ public:
         shardsvrShardCollectionRequest.setKey(request.getKey());
         shardsvrShardCollectionRequest.setUnique(request.getUnique());
         shardsvrShardCollectionRequest.setNumInitialChunks(request.getNumInitialChunks());
+        shardsvrShardCollectionRequest.setPresplitHashedZones(request.getPresplitHashedZones());
         shardsvrShardCollectionRequest.setInitialSplitPoints(request.getInitialSplitPoints());
         shardsvrShardCollectionRequest.setCollation(request.getCollation());
         shardsvrShardCollectionRequest.setGetUUIDfromPrimaryShard(
diff --git a/src/mongo/db/s/config/initial_split_policy.cpp b/src/mongo/db/s/config/initial_split_policy.cpp
index b3abf21b16c..6826a1c8123 100644
--- a/src/mongo/db/s/config/initial_split_policy.cpp
+++ b/src/mongo/db/s/config/initial_split_policy.cpp
@@ -65,8 +65,8 @@ void appendChunk(const NamespaceString& nss,
 /*
  * Returns a map mapping each tag name to a vector of shard ids with that tag name
  */
-StringMap<std::vector<ShardId>> getTagToShardIds(OperationContext* opCtx,
-                                                 const std::vector<TagsType>& tags) {
+StringMap<std::vector<ShardId>> buildTagsToShardIdsMap(OperationContext* opCtx,
+                                                       const std::vector<TagsType>& tags) {
     StringMap<std::vector<ShardId>> tagToShardIds;
     if (tags.empty()) {
         return tagToShardIds;
@@ -101,33 +101,14 @@ StringMap<std::vector<ShardId>> getTagToShardIds(OperationContext* opCtx,
 
 }  // namespace
 
-void InitialSplitPolicy::calculateHashedSplitPointsForEmptyCollection(
-    const ShardKeyPattern& shardKeyPattern,
-    bool isEmpty,
-    int numShards,
-    int numInitialChunks,
-    std::vector<BSONObj>* initialSplitPoints,
-    std::vector<BSONObj>* finalSplitPoints) {
-    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 only supported when the collection is empty "
-                                 "and has a hashed field as shard key prefix",
-                !numInitialChunks);
-        return;
-    }
+std::vector<BSONObj> InitialSplitPolicy::calculateHashedSplitPoints(
+    const ShardKeyPattern& shardKeyPattern, BSONObj prefix, int numInitialChunks) {
+    invariant(shardKeyPattern.isHashedPattern());
+    invariant(numInitialChunks > 0);
 
-    // no split points are needed
+    std::vector<BSONObj> splitPoints;
     if (numInitialChunks == 1) {
-        return;
-    }
-
-    // If initial split points are not specified, only pre-split when using a hashed shard key and
-    // the collection is empty
-    if (numInitialChunks <= 0) {
-        // Default the number of initial chunks it they are not specified
-        numInitialChunks = 2 * numShards;
+        return splitPoints;
     }
 
     // Hashes are signed, 64-bit integers. So we divide the range (-MIN long, +MAX long) into
@@ -140,35 +121,40 @@ void InitialSplitPolicy::calculateHashedSplitPointsForEmptyCollection(
     const auto proposedKey(shardKeyPattern.getKeyPattern().toBSON());
 
     auto buildSplitPoint = [&](long long value) {
-        return shardKeyPattern.getKeyPattern().extendRangeBound(
-            BSON(proposedKey.firstElementFieldName() << value), false);
+        // Forward the iterator until hashed field is reached.
+        auto shardKeyPatternItr = BSONObjIterator(shardKeyPattern.getKeyPattern().toBSON());
+        while (shardKeyPattern.getHashedField().fieldNameStringData() !=
+               (*shardKeyPatternItr++).fieldNameStringData()) {
+        }
+
+        // Append the prefix fields to the new splitpoint, if any such fields exist.
+        BSONObjBuilder bob(prefix);
+
+        // Append the value of the hashed field for the current splitpoint.
+        bob.append(shardKeyPattern.getHashedField().fieldNameStringData(), value);
+
+        // Set all subsequent shard key fields to MinKey.
+        while (shardKeyPatternItr.more()) {
+            bob.appendMinKey((*shardKeyPatternItr++).fieldNameStringData());
+        }
+        return bob.obj();
     };
 
     if (numInitialChunks % 2 == 0) {
-        finalSplitPoints->push_back(buildSplitPoint(current));
+        splitPoints.push_back(buildSplitPoint(current));
         current += intervalSize;
     } else {
         current += intervalSize / 2;
     }
 
     for (int i = 0; i < (numInitialChunks - 1) / 2; i++) {
-        finalSplitPoints->push_back(buildSplitPoint(current));
-        finalSplitPoints->push_back(buildSplitPoint(-current));
+        splitPoints.push_back(buildSplitPoint(current));
+        splitPoints.push_back(buildSplitPoint(-current));
         current += intervalSize;
     }
 
-    sort(finalSplitPoints->begin(),
-         finalSplitPoints->end(),
-         SimpleBSONObjComparator::kInstance.makeLessThan());
-
-    // The initial splits define the "big chunks" that we will subdivide later.
-    int lastIndex = -1;
-    for (int i = 1; i < numShards; i++) {
-        if (lastIndex < (i * numInitialChunks) / numShards - 1) {
-            lastIndex = (i * numInitialChunks) / numShards - 1;
-            initialSplitPoints->push_back(finalSplitPoints->at(lastIndex));
-        }
-    }
+    sort(splitPoints.begin(), splitPoints.end(), SimpleBSONObjComparator::kInstance.makeLessThan());
+    return splitPoints;
 }
 
 InitialSplitPolicy::ShardCollectionConfig InitialSplitPolicy::generateShardCollectionInitialChunks(
@@ -216,194 +202,57 @@ InitialSplitPolicy::ShardCollectionConfig InitialSplitPolicy::generateShardColle
     return {std::move(chunks)};
 }
 
-InitialSplitPolicy::ShardCollectionConfig
-InitialSplitPolicy::generateShardCollectionInitialZonedChunks(
-    const NamespaceString& nss,
+std::unique_ptr<InitialSplitPolicy> InitialSplitPolicy::calculateOptimizationStrategy(
+    OperationContext* opCtx,
     const ShardKeyPattern& shardKeyPattern,
-    const Timestamp& validAfter,
+    const ShardsvrShardCollection& request,
     const std::vector<TagsType>& tags,
-    const StringMap<std::vector<ShardId>>& tagToShards,
-    const std::vector<ShardId>& shardIdsForGaps) {
-    invariant(!shardIdsForGaps.empty());
-    invariant(!tags.empty());
-
-    const auto& keyPattern = shardKeyPattern.getKeyPattern();
-
-    auto nextShardIdForHole = [&, indx = 0]() mutable {
-        return shardIdsForGaps[indx++ % shardIdsForGaps.size()];
-    };
-
-    auto nextShardIdForTag = [&, tagToIndx = StringMap<size_t>()](const auto& tag) mutable {
-        const auto it = tagToShards.find(tag.getTag());
-        invariant(it != tagToShards.end());
-        const auto& shardIdsForTag = it->second;
-        uassert(50973,
-                str::stream()
-                    << "Cannot shard collection " << nss.ns() << " due to zone " << tag.getTag()
-                    << " which is not assigned to a shard. Please assign this zone to a shard.",
-                !shardIdsForTag.empty());
-        const auto nextShardIndx = tagToIndx[tag.getTag()]++;
-        return shardIdsForTag[nextShardIndx % shardIdsForTag.size()];
-    };
-
-    std::vector<ChunkType> chunks;
-
-    ChunkVersion version(1, 0, OID::gen());
-    auto lastChunkMax = keyPattern.globalMin();
-
-    for (const auto& tag : tags) {
-        // Create a chunk for the hole [lastChunkMax, tag.getMinKey)
-        if (tag.getMinKey().woCompare(lastChunkMax) > 0) {
-            appendChunk(nss,
-                        lastChunkMax,
-                        tag.getMinKey(),
-                        &version,
-                        validAfter,
-                        nextShardIdForHole(),
-                        &chunks);
-        }
-
-        // Create chunk for the actual tag - [tag.getMinKey, tag.getMaxKey)
-        appendChunk(nss,
-                    tag.getMinKey(),
-                    tag.getMaxKey(),
-                    &version,
-                    validAfter,
-                    nextShardIdForTag(tag),
-                    &chunks);
-        lastChunkMax = tag.getMaxKey();
+    size_t numShards,
+    bool collectionIsEmpty) {
+    uassert(ErrorCodes::InvalidOptions,
+            str::stream() << "numInitialChunks is only supported when the collection is empty "
+                             "and has a hashed field in the shard key pattern",
+            !request.getNumInitialChunks() ||
+                (shardKeyPattern.isHashedPattern() && collectionIsEmpty));
+    uassert(ErrorCodes::InvalidOptions,
+            str::stream()
+                << "When the prefix of the hashed shard key is a range field, "
+                   "'numInitialChunks' can only be used when the 'presplitHashedZones' is true",
+            !request.getNumInitialChunks() || shardKeyPattern.hasHashedPrefix() ||
+                request.getPresplitHashedZones());
+    uassert(ErrorCodes::InvalidOptions,
+            str::stream() << "Cannot have both initial split points and tags set",
+            !request.getInitialSplitPoints() || tags.empty());
+
+    // If 'presplitHashedZones' flag is set, we always use 'PresplitHashedZonesSplitPolicy', to make
+    // sure we throw the correct assertion if further validation fails.
+    if (request.getPresplitHashedZones()) {
+        return std::make_unique<PresplitHashedZonesSplitPolicy>(
+            opCtx, shardKeyPattern, tags, request.getNumInitialChunks(), collectionIsEmpty);
     }
 
-    // Create a chunk for the hole [lastChunkMax, MaxKey]
-    if (lastChunkMax.woCompare(keyPattern.globalMax()) < 0) {
-        appendChunk(nss,
-                    lastChunkMax,
-                    keyPattern.globalMax(),
-                    &version,
-                    validAfter,
-                    nextShardIdForHole(),
-                    &chunks);
+    // The next preference is to use split points based strategy. This is only possible if
+    // 'initialSplitPoints' is set, or if the collection is empty with shard key having a hashed
+    // prefix.
+    if (request.getInitialSplitPoints()) {
+        return std::make_unique<SplitPointsBasedSplitPolicy>(*request.getInitialSplitPoints());
     }
-
-    return {std::move(chunks)};
-}
-
-InitialSplitPolicy::ShardingOptimizationType InitialSplitPolicy::calculateOptimizationType(
-    const std::vector<BSONObj>& splitPoints,
-    const std::vector<TagsType>& tags,
-    bool collectionIsEmpty) {
-    if (!splitPoints.empty()) {
-        return SplitPointsProvided;
+    if (tags.empty() && shardKeyPattern.hasHashedPrefix() && collectionIsEmpty) {
+        return std::make_unique<SplitPointsBasedSplitPolicy>(
+            shardKeyPattern, numShards, request.getNumInitialChunks());
     }
 
     if (!tags.empty()) {
         if (collectionIsEmpty) {
-            return TagsProvidedWithEmptyCollection;
+            return std::make_unique<SingleChunkPerTagSplitPolicy>(opCtx, tags);
         }
-        return TagsProvidedWithNonEmptyCollection;
+        return std::make_unique<SingleChunkOnPrimarySplitPolicy>();
     }
 
     if (collectionIsEmpty) {
-        return EmptyCollection;
+        return std::make_unique<SingleChunkOnPrimarySplitPolicy>();
     }
-
-    return None;
-}
-
-InitialSplitPolicy::ShardCollectionConfig InitialSplitPolicy::createFirstChunksOptimized(
-    OperationContext* opCtx,
-    const NamespaceString& nss,
-    const ShardKeyPattern& shardKeyPattern,
-    const ShardId& primaryShardId,
-    const std::vector<BSONObj>& splitPoints,
-    const std::vector<TagsType>& tags,
-    ShardingOptimizationType optimizationType,
-    bool isEmpty,
-    int numContiguousChunksPerShard) {
-    uassert(ErrorCodes::InvalidOptions,
-            "Cannot generate initial chunks based on both split points and zones",
-            tags.empty() || splitPoints.empty());
-
-    const auto shardRegistry = Grid::get(opCtx)->shardRegistry();
-
-    const auto& keyPattern = shardKeyPattern.getKeyPattern();
-
-    const auto validAfter = LogicalClock::get(opCtx)->getClusterTime().asTimestamp();
-
-    // On which shards are the generated chunks allowed to be placed
-    std::vector<ShardId> shardIds;
-    if (isEmpty) {
-        shardRegistry->getAllShardIdsNoReload(&shardIds);
-    } else {
-        shardIds.push_back(primaryShardId);
-    }
-
-    ShardCollectionConfig initialChunks;
-
-    if (optimizationType == ShardingOptimizationType::SplitPointsProvided) {
-        initialChunks = generateShardCollectionInitialChunks(nss,
-                                                             shardKeyPattern,
-                                                             primaryShardId,
-                                                             validAfter,
-                                                             splitPoints,
-                                                             shardIds,
-                                                             numContiguousChunksPerShard);
-    } else if (optimizationType == ShardingOptimizationType::TagsProvidedWithEmptyCollection) {
-        initialChunks = generateShardCollectionInitialZonedChunks(
-            nss, shardKeyPattern, validAfter, tags, getTagToShardIds(opCtx, tags), shardIds);
-    } else if (optimizationType == ShardingOptimizationType::TagsProvidedWithNonEmptyCollection ||
-               optimizationType == ShardingOptimizationType::EmptyCollection) {
-        ChunkVersion version(1, 0, OID::gen());
-        appendChunk(nss,
-                    keyPattern.globalMin(),
-                    keyPattern.globalMax(),
-                    &version,
-                    validAfter,
-                    primaryShardId,
-                    &initialChunks.chunks);
-    } else {
-        MONGO_UNREACHABLE;
-    }
-
-    return initialChunks;
-}
-
-InitialSplitPolicy::ShardCollectionConfig InitialSplitPolicy::createFirstChunksUnoptimized(
-    OperationContext* opCtx,
-    const NamespaceString& nss,
-    const ShardKeyPattern& shardKeyPattern,
-    const ShardId& primaryShardId) {
-    const auto& keyPattern = shardKeyPattern.getKeyPattern();
-    const auto validAfter = LogicalClock::get(opCtx)->getClusterTime().asTimestamp();
-
-    // On which shards are the generated chunks allowed to be placed
-    std::vector<ShardId> shardIds{primaryShardId};
-
-    auto primaryShard =
-        uassertStatusOK(Grid::get(opCtx)->shardRegistry()->getShard(opCtx, primaryShardId));
-
-    // Refresh the balancer settings to ensure the chunk size setting, which is sent as part of
-    // the splitVector command and affects the number of chunks returned, has been loaded.
-    const auto balancerConfig = Grid::get(opCtx)->getBalancerConfiguration();
-    uassertStatusOK(balancerConfig->refreshAndCheck(opCtx));
-
-    const auto shardSelectedSplitPoints = uassertStatusOK(shardutil::selectChunkSplitPoints(
-        opCtx,
-        primaryShardId,
-        nss,
-        shardKeyPattern,
-        ChunkRange(keyPattern.globalMin(), keyPattern.globalMax()),
-        balancerConfig->getMaxChunkSizeBytes(),
-        0));
-
-    return generateShardCollectionInitialChunks(nss,
-                                                shardKeyPattern,
-                                                primaryShardId,
-                                                validAfter,
-                                                shardSelectedSplitPoints,
-                                                shardIds,
-                                                1  // numContiguousChunksPerShard
-    );
+    return std::make_unique<UnoptimizedSplitPolicy>();
 }
 
 boost::optional<CollectionType> InitialSplitPolicy::checkIfCollectionAlreadyShardedWithSameOptions(
@@ -445,4 +294,332 @@ boost::optional<CollectionType> InitialSplitPolicy::checkIfCollectionAlreadyShar
     return existingOptions;
 }
 
+InitialSplitPolicy::ShardCollectionConfig SingleChunkOnPrimarySplitPolicy::createFirstChunks(
+    OperationContext* opCtx, const ShardKeyPattern& shardKeyPattern, SplitPolicyParams params) {
+    ShardCollectionConfig initialChunks;
+    ChunkVersion version(1, 0, OID::gen());
+    const auto& keyPattern = shardKeyPattern.getKeyPattern();
+    appendChunk(params.nss,
+                keyPattern.globalMin(),
+                keyPattern.globalMax(),
+                &version,
+                LogicalClock::get(opCtx)->getClusterTime().asTimestamp(),
+                params.primaryShardId,
+                &initialChunks.chunks);
+    return initialChunks;
+}
+
+InitialSplitPolicy::ShardCollectionConfig UnoptimizedSplitPolicy::createFirstChunks(
+    OperationContext* opCtx, const ShardKeyPattern& shardKeyPattern, SplitPolicyParams params) {
+    // Under this policy, chunks are only placed on the primary shard.
+    std::vector<ShardId> shardIds{params.primaryShardId};
+
+    // Refresh the balancer settings to ensure the chunk size setting, which is sent as part of
+    // the splitVector command and affects the number of chunks returned, has been loaded.
+    const auto balancerConfig = Grid::get(opCtx)->getBalancerConfiguration();
+    uassertStatusOK(balancerConfig->refreshAndCheck(opCtx));
+    const auto shardSelectedSplitPoints = uassertStatusOK(
+        shardutil::selectChunkSplitPoints(opCtx,
+                                          params.primaryShardId,
+                                          params.nss,
+                                          shardKeyPattern,
+                                          ChunkRange(shardKeyPattern.getKeyPattern().globalMin(),
+                                                     shardKeyPattern.getKeyPattern().globalMax()),
+                                          balancerConfig->getMaxChunkSizeBytes(),
+                                          0));
+    return generateShardCollectionInitialChunks(
+        params.nss,
+        shardKeyPattern,
+        params.primaryShardId,
+        LogicalClock::get(opCtx)->getClusterTime().asTimestamp(),
+        shardSelectedSplitPoints,
+        shardIds,
+        1  // numContiguousChunksPerShard
+    );
+}
+
+InitialSplitPolicy::ShardCollectionConfig SplitPointsBasedSplitPolicy::createFirstChunks(
+    OperationContext* opCtx, const ShardKeyPattern& shardKeyPattern, SplitPolicyParams params) {
+
+    const auto shardRegistry = Grid::get(opCtx)->shardRegistry();
+
+    // On which shards are the generated chunks allowed to be placed.
+    std::vector<ShardId> shardIds;
+    shardRegistry->getAllShardIdsNoReload(&shardIds);
+
+    return generateShardCollectionInitialChunks(
+        params.nss,
+        shardKeyPattern,
+        params.primaryShardId,
+        LogicalClock::get(opCtx)->getClusterTime().asTimestamp(),
+        _splitPoints,
+        shardIds,
+        _numContiguousChunksPerShard);
+}
+
+AbstractTagsBasedSplitPolicy::AbstractTagsBasedSplitPolicy(OperationContext* opCtx,
+                                                           std::vector<TagsType> tags)
+    : _tags(tags) {
+    _tagToShardIds = buildTagsToShardIdsMap(opCtx, tags);
+}
+
+AbstractTagsBasedSplitPolicy::SplitInfo SingleChunkPerTagSplitPolicy::buildSplitInfoForTag(
+    TagsType tag, const ShardKeyPattern& shardKeyPattern) {
+    const auto nextShardIndex = _nextShardIndexForZone[tag.getTag()]++;
+    const auto& shardIdsForTag = getTagsToShardIds().find(tag.getTag())->second;
+    auto shardId = shardIdsForTag[nextShardIndex % shardIdsForTag.size()];
+
+    // Do not generate any split points when using this strategy. We create one chunk on a shard
+    // choosen using round-robin.
+    return {{}, {std::make_pair(shardId, 1)}};
+}
+
+InitialSplitPolicy::ShardCollectionConfig AbstractTagsBasedSplitPolicy::createFirstChunks(
+    OperationContext* opCtx, const ShardKeyPattern& shardKeyPattern, SplitPolicyParams params) {
+    invariant(!_tags.empty());
+
+    const auto shardRegistry = Grid::get(opCtx)->shardRegistry();
+
+    std::vector<ShardId> shardIds;
+    shardRegistry->getAllShardIdsNoReload(&shardIds);
+    const auto validAfter = LogicalClock::get(opCtx)->getClusterTime().asTimestamp();
+    const auto& keyPattern = shardKeyPattern.getKeyPattern();
+
+    auto tagToShards = getTagsToShardIds();
+
+    auto nextShardIdForHole = [&, indx = 0L]() mutable {
+        return shardIds[indx++ % shardIds.size()];
+    };
+
+    ChunkVersion version(1, 0, OID::gen());
+    auto lastChunkMax = keyPattern.globalMin();
+    std::vector<ChunkType> chunks;
+    for (const auto& tag : _tags) {
+        // Create a chunk for the hole [lastChunkMax, tag.getMinKey)
+        if (tag.getMinKey().woCompare(lastChunkMax) > 0) {
+            appendChunk(params.nss,
+                        lastChunkMax,
+                        tag.getMinKey(),
+                        &version,
+                        validAfter,
+                        nextShardIdForHole(),
+                        &chunks);
+        }
+        // Create chunk for the actual tag - [tag.getMinKey, tag.getMaxKey)
+        const auto it = tagToShards.find(tag.getTag());
+        invariant(it != tagToShards.end());
+        uassert(50973,
+                str::stream()
+                    << "Cannot shard collection " << params.nss.ns() << " due to zone "
+                    << tag.getTag()
+                    << " which is not assigned to a shard. Please assign this zone to a shard.",
+                !it->second.empty());
+
+        // The buildSplitInfoForTag() should provide split points which are in sorted order. So we
+        // don't need to sort them again while generating chunks.
+        auto splitInfo = buildSplitInfoForTag(tag, shardKeyPattern);
+
+        // Ensure that the number of splitPoints is consistent with the computed chunk distribution.
+        // The resulting number of chunks will be one more than the number of split points to
+        // accommodate boundaries.
+        invariant(splitInfo.splitPoints.size() + 1 ==
+                  std::accumulate(splitInfo.chunkDistribution.begin(),
+                                  splitInfo.chunkDistribution.end(),
+                                  static_cast<size_t>(0),  // initial value for 'runningSum'.
+                                  [](size_t runningSum, const auto& currentElem) {
+                                      return runningSum + currentElem.second;
+                                  }));
+
+        // Generate chunks using 'splitPoints' and distribute them among shards based on
+        // 'chunkDistributionPerShard'.
+        size_t splitPointIdx = 0;
+        for (auto&& chunksOnShard : splitInfo.chunkDistribution) {
+            const auto [targetShard, numChunksForShard] = chunksOnShard;
+            for (size_t i = 0; i < numChunksForShard; ++i, ++splitPointIdx) {
+                const BSONObj min = (splitPointIdx == 0) ? tag.getMinKey()
+                                                         : splitInfo.splitPoints[splitPointIdx - 1];
+                const BSONObj max = (splitPointIdx == splitInfo.splitPoints.size())
+                    ? tag.getMaxKey()
+                    : splitInfo.splitPoints[splitPointIdx];
+                appendChunk(params.nss, min, max, &version, validAfter, targetShard, &chunks);
+            }
+        }
+        lastChunkMax = tag.getMaxKey();
+    }
+
+    // Create a chunk for the hole [lastChunkMax, MaxKey]
+    if (lastChunkMax.woCompare(keyPattern.globalMax()) < 0) {
+        appendChunk(params.nss,
+                    lastChunkMax,
+                    keyPattern.globalMax(),
+                    &version,
+                    validAfter,
+                    nextShardIdForHole(),
+                    &chunks);
+    }
+
+    return {std::move(chunks)};
+}
+
+AbstractTagsBasedSplitPolicy::SplitInfo PresplitHashedZonesSplitPolicy::buildSplitInfoForTag(
+    TagsType tag, const ShardKeyPattern& shardKeyPattern) {
+    // Returns the ceiling number for the decimal value of x/y.
+    auto ceilOfXOverY = [](auto x, auto y) { return (x / y) + (x % y != 0); };
+
+    // This strategy presplits each tag such that at least 1 chunk is placed on every shard to which
+    // the tag is assigned. We distribute the chunks such that at least '_numInitialChunks' are
+    // created across the cluster, and we make a best-effort attempt to ensure that an equal number
+    // of chunks are created on each shard regardless of how the zones are laid out.
+
+    //  We take the ceiling when the number is not divisible so that the final number of chunks
+    //  we generate are at least '_numInitialChunks'.
+    auto numChunksPerShard = ceilOfXOverY(_numInitialChunks, _numTagsPerShard.size());
+
+    const auto& tagsToShardsMap = getTagsToShardIds();
+    invariant(tagsToShardsMap.find(tag.getTag()) != tagsToShardsMap.end());
+    const auto& shardsForCurrentTag = tagsToShardsMap.find(tag.getTag())->second;
+
+    // For each shard in the current zone, find the quota of chunks that can be allocated to that
+    // zone. We distribute chunks equally to all the zones present on a shard.
+    std::vector<std::pair<ShardId, size_t>> chunkDistribution;
+    chunkDistribution.reserve((shardsForCurrentTag.size()));
+    auto numChunksForCurrentTag = 0;
+    for (auto&& shard : shardsForCurrentTag) {
+        auto numChunksForCurrentTagOnShard =
+            ceilOfXOverY(numChunksPerShard, _numTagsPerShard[shard.toString()]);
+        chunkDistribution.push_back({shard, numChunksForCurrentTagOnShard});
+        numChunksForCurrentTag += (numChunksForCurrentTagOnShard);
+    }
+
+    // Extract the fields preceding the hashed field. We use this object as a base for building
+    // split points.
+    BSONObjBuilder bob;
+    for (auto&& elem : tag.getMinKey()) {
+        if (elem.fieldNameStringData() == shardKeyPattern.getHashedField().fieldNameStringData()) {
+            break;
+        }
+        bob.append(elem);
+    }
+    auto prefixBSON = bob.obj();
+
+    return {calculateHashedSplitPoints(shardKeyPattern, prefixBSON, numChunksForCurrentTag),
+            std::move(chunkDistribution)};
+}
+
+PresplitHashedZonesSplitPolicy::PresplitHashedZonesSplitPolicy(
+    OperationContext* opCtx,
+    const ShardKeyPattern& shardKeyPattern,
+    std::vector<TagsType> tags,
+    size_t numInitialChunks,
+    bool isCollectionEmpty)
+    : AbstractTagsBasedSplitPolicy(opCtx, tags) {
+    // Verify that tags have been set up correctly for this split policy.
+    _validate(shardKeyPattern, isCollectionEmpty);
+
+    // Calculate the count of zones on each shard and save it in a map for later.
+    const auto& tagsToShards = getTagsToShardIds();
+    for (auto&& tag : tags) {
+        auto& shardsForCurrentTag = tagsToShards.find(tag.getTag())->second;
+        for (auto&& shard : shardsForCurrentTag) {
+            _numTagsPerShard[shard.toString()]++;
+        }
+    }
+    // If we are here, we have confirmed that at least one tag is already set up. A tag can only be
+    // created if they are associated with a zone and the zone has to be assigned to a shard.
+    invariant(!_numTagsPerShard.empty());
+
+    // If 'numInitialChunks' was not specified, use default value.
+    _numInitialChunks = numInitialChunks ? numInitialChunks : _numTagsPerShard.size() * 2;
+}
+
+/**
+ * If 'presplitHashedZones' flag is set with shard key prefix being a non-hashed field then all
+ * zones must be set up according to the following rules:
+ *  1. All lower-bound prefix fields of the shard key must have a value other than MinKey or
+ * MaxKey.
+ *  2. All lower-bound fields from the hash field onwards must be MinKey.
+ *  3. At least one upper-bound prefix field must be different than the lower bound counterpart.
+ *
+ * Examples for shard key {country : 1, hashedField: "hashed", suffix : 1}:
+ * Zone with range : [{country : "US", hashedField: MinKey, suffix: MinKey}, {country :MaxKey,
+ * hashedField: MaxKey, suffix: MaxKey}) is valid.
+ * Zone with range : [{country : MinKey, hashedField: MinKey, suffix: MinKey}, {country : "US",
+ * hashedField: MinKey, suffix: MinKey}) is invalid since it violates #1 rule.
+ * Zone with range: [{country : "US", hashedField: MinKey, suffix: "someVal"}, {country :MaxKey,
+ * hashedField: MaxKey, suffix: MaxKey}) is invalid since it violates #2 rule.
+ * Zone with range: [{country : "US", hashedField: MinKey, suffix: MinKey}, {country : "US",
+ * hashedField: MaxKey, suffix: MaxKey}) is invalid since it violates #3 rule.
+ *
+ * If the shard key has a hashed prefix, then pre-splitting is only supported if there is a single
+ * zone defined from global MinKey to global MaxKey. i.e, if the shard key is {x: "hashed", y: 1}
+ * then there should be exactly one zone ranging from {x:MinKey, y:MinKey} to {x:MaxKey, y:MaxKey}.
+ */
+void PresplitHashedZonesSplitPolicy::_validate(const ShardKeyPattern& shardKeyPattern,
+                                               bool isCollectionEmpty) {
+    const auto& tags = getTags();
+    uassert(
+        31387,
+        "'presplitHashedZones' is only supported when the collection is empty, zones are set up "
+        "and shard key pattern has a hashed field",
+        isCollectionEmpty && !tags.empty() && shardKeyPattern.isHashedPattern());
+
+    if (shardKeyPattern.hasHashedPrefix()) {
+        uassert(31412,
+                "For hashed prefix shard keys, 'presplitHashedZones' is only supported when there "
+                "is a single zone defined which covers entire shard key range",
+                (tags.size() == 1) &&
+                    !shardKeyPattern.getKeyPattern().globalMin().woCompare(tags[0].getMinKey()) &&
+                    !shardKeyPattern.getKeyPattern().globalMax().woCompare(tags[0].getMaxKey()));
+        return;
+    }
+    for (auto&& tag : tags) {
+        auto startItr = BSONObjIterator(tag.getMinKey());
+        auto endItr = BSONObjIterator(tag.getMaxKey());
+
+        // We cannot pre-split if the lower bound fields preceding the hashed field are same as
+        // the upper bound. We validate that at least one of the preceding field is different.
+        // Additionally we all make sure that none of the lower-bound prefix fields have Minkey
+        // or MaxKey.
+        bool isPrefixDifferent = false;
+        do {
+            uassert(31388,
+                    str::stream()
+                        << "One or more zones are not defined in a manner that supports hashed "
+                           "pre-splitting. Cannot have MinKey or MaxKey in the lower bound for "
+                           "fields preceding the hashed field but found one, for zone "
+                        << tag.getTag(),
+                    (*startItr).type() != BSONType::MinKey &&
+                        (*startItr).type() != BSONType::MaxKey);
+            isPrefixDifferent = isPrefixDifferent || (*startItr).woCompare(*endItr);
+            ++endItr;
+            // Forward the iterator until hashed field is reached.
+        } while ((*++startItr).fieldNameStringData() !=
+                 shardKeyPattern.getHashedField().fieldNameStringData());
+        uassert(31390,
+                str::stream() << "One or more zones are not defined in a manner that supports "
+                                 "hashed pre-splitting. The value preceding hashed field of the "
+                                 "upper bound should be greater than that of lower bound, for zone "
+                              << tag.getTag(),
+                isPrefixDifferent);
+
+        uassert(
+            31389,
+            str::stream() << "One or more zones are not defined in a manner that supports "
+                             "hashed pre-splitting. The hashed field value for lower bound must "
+                             "be MinKey, for zone "
+                          << tag.getTag(),
+            (*startItr).type() == BSONType::MinKey);
+
+        // Each field in the lower bound after the hashed field must be set to MinKey.
+        while (startItr.more()) {
+            uassert(31391,
+                    str::stream() << "One or more zones are not defined in a manner that supports "
+                                     "hashed pre-splitting. The fields after the hashed field must "
+                                     "have MinKey value, for zone "
+                                  << tag.getTag(),
+                    (*startItr++).type() == BSONType::MinKey);
+        }
+    }
+}
+
 }  // namespace mongo
diff --git a/src/mongo/db/s/config/initial_split_policy.h b/src/mongo/db/s/config/initial_split_policy.h
index beb034e5468..3b5a5d9ea12 100644
--- a/src/mongo/db/s/config/initial_split_policy.h
+++ b/src/mongo/db/s/config/initial_split_policy.h
@@ -44,55 +44,37 @@
 
 namespace mongo {
 
+struct SplitPolicyParams {
+    NamespaceString nss;
+    ShardId primaryShardId;
+};
+
 class InitialSplitPolicy {
 public:
     /**
-     * Indicates the optimization allowed for sharding a collection given the collection's initial
-     * properties.
+     * Returns the optimization strategy for building initial chunks based on the input parameters
+     * and the collection state.
      */
-    enum ShardingOptimizationType {
-        // If split points are provided, we directly generate corresponding initial chunks.
-        SplitPointsProvided,
-
-        // If tags are provided and the collection is empty, we directly write corresponding zones
-        // to the config server.
-        TagsProvidedWithEmptyCollection,
-
-        // If tags are provided and the collection is not empty, we create one chunk on the primary
-        // shard and leave the balancer to do zone splitting and placement.
-        TagsProvidedWithNonEmptyCollection,
-
-        // If the chunk is empty, we create one chunk on the primary shard.
-        EmptyCollection,
-
-        // If no optimizations are available, we calculate split points on the primary shard.
-        None,
-    };
-
-    static ShardingOptimizationType calculateOptimizationType(
-        const std::vector<BSONObj>& splitPoints,
+    static std::unique_ptr<InitialSplitPolicy> calculateOptimizationStrategy(
+        OperationContext* opCtx,
+        const ShardKeyPattern& shardKeyPattern,
+        const ShardsvrShardCollection& request,
         const std::vector<TagsType>& tags,
+        size_t numShards,
         bool collectionIsEmpty);
 
-
     /**
-     * For new collections which use hashed shard keys, we can can pre-split the range of possible
-     * hashes into a large number of chunks, and distribute them evenly at creation time.
-     *
-     * Until we design a better initialization scheme, the most performant way to pre-split is to
-     * make one big chunk for each shard and migrate them one at a time. Because of this:
-     * - 'initialSplitPoints' is populated with the split points to use on the primary shard to
-     * produce the initial "big chunks."
-     * - 'finalSplitPoints' is populated with the additional split points to use on the "big chunks"
-     * after the "big chunks" have been spread evenly across shards through migrations.
+     * Returns split points to use for creating chunks in cases where the shard key contains a
+     * hashed field. For new collections which use hashed shard keys, we can can pre-split the range
+     * of possible hashes into a large number of chunks, and distribute them evenly at creation
+     * time. In the case where the shard key is compound hashed, the 'prefix' object specifies the
+     * non-hashed prefix to be prepended to each hashed splitpoint. If no such prefix exists, this
+     * will be an empty BSONObj. It is an error to pass a 'prefix' object which is not consistent
+     * with the given ShardKeyPattern.
      */
-    static void calculateHashedSplitPointsForEmptyCollection(
-        const ShardKeyPattern& shardKeyPattern,
-        bool isEmpty,
-        int numShards,
-        int numInitialChunks,
-        std::vector<BSONObj>* initialSplitPoints,
-        std::vector<BSONObj>* finalSplitPoints);
+    static std::vector<BSONObj> calculateHashedSplitPoints(const ShardKeyPattern& shardKeyPattern,
+                                                           BSONObj prefix,
+                                                           int numInitialChunks);
 
     struct ShardCollectionConfig {
         std::vector<ChunkType> chunks;
@@ -129,51 +111,6 @@ public:
         const int numContiguousChunksPerShard = 1);
 
     /**
-     * Produces the initial chunks that need to be written for an *empty* collection which is being
-     * sharded based on the given 'tags'.
-     *
-     * NOTE: The function performs some basic validation of the input parameters, but there is no
-     * checking whether the collection contains any data or not.
-     *
-     * The contents of 'tags' will be used to create chunks, which correspond to these zones and
-     * chunks will be assigned to shards from 'tagToShards'. If there are any holes in between the
-     * zones (zones are not contiguous), these holes will be assigned to 'shardIdsForGaps' in
-     * round-robin fashion.
-     */
-    static ShardCollectionConfig generateShardCollectionInitialZonedChunks(
-        const NamespaceString& nss,
-        const ShardKeyPattern& shardKeyPattern,
-        const Timestamp& validAfter,
-        const std::vector<TagsType>& tags,
-        const StringMap<std::vector<ShardId>>& tagToShards,
-        const std::vector<ShardId>& shardIdsForGaps);
-
-    /**
-     * Generates a list with what are the most optimal first chunks and placement for a newly
-     * sharded collection.
-     *
-     * If the collection 'isEmpty', chunks will be spread across all available (appropriate based on
-     * zoning rules) shards. Otherwise, they will all end up on the primary shard after which the
-     * balancer will take care of properly distributing them around.
-     */
-    static ShardCollectionConfig createFirstChunksOptimized(
-        OperationContext* opCtx,
-        const NamespaceString& nss,
-        const ShardKeyPattern& shardKeyPattern,
-        const ShardId& primaryShardId,
-        const std::vector<BSONObj>& splitPoints,
-        const std::vector<TagsType>& tags,
-        ShardingOptimizationType optimizationType,
-        bool isEmpty,
-        int numContiguousChunksPerShard = 1);
-
-    static ShardCollectionConfig createFirstChunksUnoptimized(
-        OperationContext* opCtx,
-        const NamespaceString& nss,
-        const ShardKeyPattern& shardKeyPattern,
-        const ShardId& primaryShardId);
-
-    /**
      * Throws an exception if the collection is already sharded with different options.
      *
      * If the collection is already sharded with the same options, returns the existing
@@ -184,5 +121,169 @@ public:
         const NamespaceString& nss,
         const ShardsvrShardCollection& request,
         repl::ReadConcernLevel readConcernLevel);
+
+    /**
+     * Generates a list of initial chunks to be created during a shardCollection operation.
+     */
+    virtual ShardCollectionConfig createFirstChunks(OperationContext* opCtx,
+                                                    const ShardKeyPattern& shardKeyPattern,
+                                                    SplitPolicyParams params) = 0;
+
+    /**
+     * Returns whether the chunk generation strategy being used is optimized or not. Since there is
+     * only a single unoptimized policy, we return true by default here.
+     */
+    virtual bool isOptimized() {
+        return true;
+    }
+
+    virtual ~InitialSplitPolicy() {}
+};
+
+/**
+ * Default optimization strategy where we create a single chunk on the primary shard.
+ */
+class SingleChunkOnPrimarySplitPolicy : public InitialSplitPolicy {
+public:
+    ShardCollectionConfig createFirstChunks(OperationContext* opCtx,
+                                            const ShardKeyPattern& shardKeyPattern,
+                                            SplitPolicyParams params);
 };
+
+/**
+ * Split point building strategy to be used when no optimizations are available. We send a
+ * splitVector command to the primary shard in order to calculate the appropriate split points.
+ */
+class UnoptimizedSplitPolicy : public InitialSplitPolicy {
+public:
+    ShardCollectionConfig createFirstChunks(OperationContext* opCtx,
+                                            const ShardKeyPattern& shardKeyPattern,
+                                            SplitPolicyParams params);
+    bool isOptimized() {
+        return false;
+    }
+};
+
+/**
+ * Split point building strategy to be used when explicit split points are supplied, or where the
+ * appropriate splitpoints can be trivially deduced from the shard key.
+ */
+class SplitPointsBasedSplitPolicy : public InitialSplitPolicy {
+public:
+    /**
+     * Constructor used when split points are provided.
+     */
+    SplitPointsBasedSplitPolicy(std::vector<BSONObj> explicitSplitPoints)
+        : _splitPoints(std::move(explicitSplitPoints)) {
+        _numContiguousChunksPerShard = 1;
+    }
+
+    /**
+     * Constructor used when generating split points for a hashed-prefix shard key.
+     */
+    SplitPointsBasedSplitPolicy(const ShardKeyPattern& shardKeyPattern,
+                                size_t numShards,
+                                size_t numInitialChunks) {
+        // If 'numInitialChunks' was not specified, use default value.
+        numInitialChunks = numInitialChunks ? numInitialChunks : numShards * 2;
+        _splitPoints = calculateHashedSplitPoints(shardKeyPattern, BSONObj(), numInitialChunks);
+        _numContiguousChunksPerShard =
+            std::max(numInitialChunks / numShards, static_cast<size_t>(1));
+    }
+
+    ShardCollectionConfig createFirstChunks(OperationContext* opCtx,
+                                            const ShardKeyPattern& shardKeyPattern,
+                                            SplitPolicyParams params);
+
+    // Helpers for unit testing.
+    const std::vector<BSONObj>& getSplitPoints() const {
+        return _splitPoints;
+    }
+    size_t getNumContiguousChunksPerShard() const {
+        return _numContiguousChunksPerShard;
+    }
+
+private:
+    std::vector<BSONObj> _splitPoints;
+    size_t _numContiguousChunksPerShard;
+};
+
+/**
+ * Abstract base class for all split policies which depend upon zones having already been defined at
+ * the moment the collection is sharded.
+ */
+class AbstractTagsBasedSplitPolicy : public InitialSplitPolicy {
+public:
+    /**
+     * Records the splitpoints and chunk distribution among shards within a particular tag range.
+     */
+    struct SplitInfo {
+        std::vector<BSONObj> splitPoints;
+        std::vector<std::pair<ShardId, size_t>> chunkDistribution;
+    };
+
+    AbstractTagsBasedSplitPolicy(OperationContext* opCtx, std::vector<TagsType> tags);
+
+    ShardCollectionConfig createFirstChunks(OperationContext* opCtx,
+                                            const ShardKeyPattern& shardKeyPattern,
+                                            SplitPolicyParams params);
+
+    /**
+     * Returns the split points to be used for generating chunks within a given tag.
+     */
+    virtual SplitInfo buildSplitInfoForTag(TagsType tag,
+                                           const ShardKeyPattern& shardKeyPattern) = 0;
+
+    const std::vector<TagsType>& getTags() const {
+        return _tags;
+    }
+
+    const StringMap<std::vector<ShardId>>& getTagsToShardIds() const {
+        return _tagToShardIds;
+    }
+
+private:
+    const std::vector<TagsType> _tags;
+    StringMap<std::vector<ShardId>> _tagToShardIds;
+};
+
+/**
+ * In this strategy we directly generate a single chunk for each tag range.
+ */
+class SingleChunkPerTagSplitPolicy : public AbstractTagsBasedSplitPolicy {
+public:
+    SingleChunkPerTagSplitPolicy(OperationContext* opCtx, std::vector<TagsType> tags)
+        : AbstractTagsBasedSplitPolicy(opCtx, tags) {}
+
+    SplitInfo buildSplitInfoForTag(TagsType tag, const ShardKeyPattern& shardKeyPattern);
+
+private:
+    StringMap<size_t> _nextShardIndexForZone;
+};
+
+/**
+ * Split point building strategy to be used when 'presplitHashedZones' flag is set. This policy is
+ * only relevant when the zones are set up before sharding and the shard key is hashed. In this
+ * case, we generate one chunk per tag range and then further split each of these using the hashed
+ * field of the shard key.
+ */
+class PresplitHashedZonesSplitPolicy : public AbstractTagsBasedSplitPolicy {
+public:
+    PresplitHashedZonesSplitPolicy(OperationContext* opCtx,
+                                   const ShardKeyPattern& shardKeyPattern,
+                                   std::vector<TagsType> tags,
+                                   size_t numInitialChunks,
+                                   bool isCollectionEmpty);
+
+    SplitInfo buildSplitInfoForTag(TagsType tag, const ShardKeyPattern& shardKeyPattern);
+
+private:
+    /**
+     * Validates that each of tags are set up correctly so that the tags can be split further.
+     */
+    void _validate(const ShardKeyPattern& shardKeyPattern, bool isCollectionEmpty);
+    size_t _numInitialChunks;
+    StringMap<size_t> _numTagsPerShard;
+};
+
 }  // namespace mongo
diff --git a/src/mongo/db/s/config/initial_split_policy_test.cpp b/src/mongo/db/s/config/initial_split_policy_test.cpp
index 68fdb44abd8..7d4309aa4fb 100644
--- a/src/mongo/db/s/config/initial_split_policy_test.cpp
+++ b/src/mongo/db/s/config/initial_split_policy_test.cpp
@@ -31,9 +31,12 @@
 
 #include "mongo/platform/basic.h"
 
+#include "mongo/bson/json.h"
+#include "mongo/db/logical_clock.h"
 #include "mongo/db/s/config/initial_split_policy.h"
 #include "mongo/s/catalog/type_shard.h"
 #include "mongo/s/catalog/type_tags.h"
+#include "mongo/s/config_server_test_fixture.h"
 #include "mongo/unittest/unittest.h"
 #include "mongo/util/log.h"
 
@@ -70,82 +73,127 @@ void assertChunkVectorsAreEqual(const std::vector<ChunkType>& expected,
 }
 
 /**
- * Returns a test hashed shard key pattern if isHashed is true.
- * Otherwise, returns a regular shard key pattern.
- */
-ShardKeyPattern makeShardKeyPattern(bool isHashed) {
-    if (isHashed)
-        return ShardKeyPattern(BSON("x"
-                                    << "hashed"));
-    return ShardKeyPattern(BSON("x" << 1));
-}
-
-/**
- * Calls calculateHashedSplitPointsForEmptyCollection according to the given arguments
+ * Calls calculateHashedSplitPoints according to the given arguments
  * and asserts that calculated split points match with the expected split points.
  */
-void checkCalculatedHashedSplitPoints(bool isHashed,
-                                      bool isEmpty,
+void checkCalculatedHashedSplitPoints(const ShardKeyPattern& shardKeyPattern,
                                       int numShards,
                                       int numInitialChunks,
-                                      const std::vector<BSONObj>* expectedInitialSplitPoints,
-                                      const std::vector<BSONObj>* expectedFinalSplitPoints) {
-    std::vector<BSONObj> initialSplitPoints;
-    std::vector<BSONObj> finalSplitPoints;
-    InitialSplitPolicy::calculateHashedSplitPointsForEmptyCollection(makeShardKeyPattern(isHashed),
-                                                                     isEmpty,
-                                                                     numShards,
-                                                                     numInitialChunks,
-                                                                     &initialSplitPoints,
-                                                                     &finalSplitPoints);
-    assertBSONObjVectorsAreEqual(*expectedInitialSplitPoints, initialSplitPoints);
-    assertBSONObjVectorsAreEqual(*expectedFinalSplitPoints, finalSplitPoints);
-}
-
-TEST(CalculateHashedSplitPointsTest, EmptyCollectionMoreChunksThanShards) {
-    const std::vector<BSONObj> expectedInitialSplitPoints = {BSON("x" << 0)};
-    const std::vector<BSONObj> expectedFinalSplitPoints = {
+                                      const std::vector<BSONObj>* expectedSplitPoints,
+                                      int expectNumChunkPerShard) {
+    SplitPointsBasedSplitPolicy policy(shardKeyPattern, numShards, numInitialChunks);
+    assertBSONObjVectorsAreEqual(*expectedSplitPoints, policy.getSplitPoints());
+    ASSERT_EQUALS(expectNumChunkPerShard, policy.getNumContiguousChunksPerShard());
+}
+
+TEST(CalculateHashedSplitPointsTest, HashedPrefixMoreChunksThanShardsWithEqualDistribution) {
+    auto shardKeyPattern = ShardKeyPattern(BSON("x"
+                                                << "hashed"
+                                                << "y" << 1));
+    const std::vector<BSONObj> expectedSplitPoints = {
+        BSON("x" << -4611686018427387902LL << "y" << MINKEY),
+        BSON("x" << 0 << "y" << MINKEY),
+        BSON("x" << 4611686018427387902LL << "y" << MINKEY)};
+    int expectNumChunkPerShard = 2;
+    checkCalculatedHashedSplitPoints(
+        shardKeyPattern, 2, 4, &expectedSplitPoints, expectNumChunkPerShard);
+}
+
+TEST(CalculateHashedSplitPointsTest, HashedPrefixMoreChunksThanShardsWithUnequalDistribution) {
+    auto shardKeyPattern = ShardKeyPattern(BSON("x"
+                                                << "hashed"));
+    const std::vector<BSONObj> expectedSplitPoints = {
         BSON("x" << -4611686018427387902LL), BSON("x" << 0), BSON("x" << 4611686018427387902LL)};
+    int expectNumChunkPerShard = 1;
+    checkCalculatedHashedSplitPoints(
+        shardKeyPattern, 3, 4, &expectedSplitPoints, expectNumChunkPerShard);
+}
+
+TEST(CalculateHashedSplitPointsTest, HashedPrefixChunksEqualToShards) {
+    auto shardKeyPattern = ShardKeyPattern(BSON("x"
+                                                << "hashed"
+                                                << "y" << 1));
+    const std::vector<BSONObj> expectedSplitPoints = {
+        BSON("x" << -3074457345618258602LL << "y" << MINKEY),
+        BSON("x" << 3074457345618258602LL << "y" << MINKEY)};
+    int expectNumChunkPerShard = 1;
     checkCalculatedHashedSplitPoints(
-        true, true, 2, 4, &expectedInitialSplitPoints, &expectedFinalSplitPoints);
+        shardKeyPattern, 3, 3, &expectedSplitPoints, expectNumChunkPerShard);
 }
 
-TEST(CalculateHashedSplitPointsTest, EmptyCollectionChunksEqualToShards) {
-    const std::vector<BSONObj> expectedSplitPoints = {BSON("x" << -3074457345618258602LL),
-                                                      BSON("x" << 3074457345618258602LL)};
-    checkCalculatedHashedSplitPoints(true, true, 3, 3, &expectedSplitPoints, &expectedSplitPoints);
+TEST(CalculateHashedSplitPointsTest, HashedPrefixChunksLessThanShards) {
+    const std::vector<BSONObj> expectedSplitPoints = {BSON("x" << 0)};
+    int expectNumChunkPerShard = 1;
+    checkCalculatedHashedSplitPoints(ShardKeyPattern(BSON("x"
+                                                          << "hashed")),
+                                     5,
+                                     2,
+                                     &expectedSplitPoints,
+                                     expectNumChunkPerShard);
 }
 
-TEST(CalculateHashedSplitPointsTest, EmptyCollectionHashedWithNoInitialSplitsReturnsEmptySplits) {
+TEST(CalculateHashedSplitPointsTest, HashedPrefixChunksOneReturnsNoSplitPoints) {
     const std::vector<BSONObj> expectedSplitPoints;
-    checkCalculatedHashedSplitPoints(true, true, 2, 1, &expectedSplitPoints, &expectedSplitPoints);
+    int expectNumChunkPerShard = 1;
+    checkCalculatedHashedSplitPoints(ShardKeyPattern(BSON("x"
+                                                          << "hashed")),
+                                     2,
+                                     1,
+                                     &expectedSplitPoints,
+                                     expectNumChunkPerShard);
 }
 
-TEST(CalculateHashedSplitPointsTest, EmptyCollectionNumInitialChunksZero) {
-    const std::vector<BSONObj> expectedInitialSplitPoints = {BSON("x" << 0)};
-    const std::vector<BSONObj> expectedFinalSplitPoints = {
+TEST(CalculateHashedSplitPointsTest, HashedPrefixChunksZeroUsesDefault) {
+    const std::vector<BSONObj> expectedSplitPoints = {
         BSON("x" << -4611686018427387902LL), BSON("x" << 0), BSON("x" << 4611686018427387902LL)};
-    checkCalculatedHashedSplitPoints(
-        true, true, 2, 0, &expectedInitialSplitPoints, &expectedFinalSplitPoints);
+    int expectNumChunkPerShard = 2;
+    checkCalculatedHashedSplitPoints(ShardKeyPattern(BSON("x"
+                                                          << "hashed")),
+                                     2,
+                                     0,
+                                     &expectedSplitPoints,
+                                     expectNumChunkPerShard);
 }
 
-TEST(CalculateHashedSplitPointsTest, NonEmptyCollectionHashedWithInitialSplitsFails) {
-    std::vector<BSONObj> expectedSplitPoints;
-    ASSERT_THROWS_CODE(checkCalculatedHashedSplitPoints(
-                           true, false, 2, 3, &expectedSplitPoints, &expectedSplitPoints),
-                       AssertionException,
-                       ErrorCodes::InvalidOptions);
+TEST(CalculateHashedSplitPointsTest, HashedSuffix) {
+    auto shardKeyPattern = ShardKeyPattern(BSON("x.a" << 1 << "y.b" << 1 << "z.c"
+                                                      << "hashed"));
+    const auto preDefinedPrefix = fromjson("{'x.a': {p: 1}, 'y.b': 'val'}");
+    const std::vector<BSONObj> expectedSplitPoints = {
+        BSONObjBuilder(preDefinedPrefix).append("z.c", -4611686018427387902LL).obj(),
+        BSONObjBuilder(preDefinedPrefix).append("z.c", 0LL).obj(),
+        BSONObjBuilder(preDefinedPrefix).append("z.c", 4611686018427387902LL).obj()};
+    assertBSONObjVectorsAreEqual(
+        expectedSplitPoints,
+        InitialSplitPolicy::calculateHashedSplitPoints(shardKeyPattern, preDefinedPrefix, 4));
 }
 
-TEST(CalculateHashedSplitPointsTest, NotHashedWithInitialSplitsFails) {
-    std::vector<BSONObj> expectedSplitPoints;
-    ASSERT_THROWS_CODE(checkCalculatedHashedSplitPoints(
-                           false, true, 2, 3, &expectedSplitPoints, &expectedSplitPoints),
-                       AssertionException,
-                       ErrorCodes::InvalidOptions);
+TEST(CalculateHashedSplitPointsTest, HashedInfix) {
+    auto shardKeyPattern = ShardKeyPattern(BSON("x.a" << 1 << "y.b"
+                                                      << "hashed"
+                                                      << "z.c" << 1 << "a" << 1));
+    const auto preDefinedPrefix = fromjson("{'x.a': {p: 1}}");
+    const std::vector<BSONObj> expectedSplitPoints = {BSONObjBuilder(preDefinedPrefix)
+                                                          .append("y.b", -4611686018427387902LL)
+                                                          .appendMinKey("z.c")
+                                                          .appendMinKey("a")
+                                                          .obj(),
+                                                      BSONObjBuilder(preDefinedPrefix)
+                                                          .append("y.b", 0LL)
+                                                          .appendMinKey("z.c")
+                                                          .appendMinKey("a")
+                                                          .obj(),
+                                                      BSONObjBuilder(preDefinedPrefix)
+                                                          .append("y.b", 4611686018427387902LL)
+                                                          .appendMinKey("z.c")
+                                                          .appendMinKey("a")
+                                                          .obj()};
+    assertBSONObjVectorsAreEqual(
+        expectedSplitPoints,
+        InitialSplitPolicy::calculateHashedSplitPoints(shardKeyPattern, preDefinedPrefix, 4));
 }
 
-class GenerateInitialSplitChunksTestBase : public unittest::Test {
+class GenerateInitialSplitChunksTestBase : public ConfigServerTestFixture {
 public:
     /**
      * Returns a vector of ChunkType objects for the given chunk ranges.
@@ -153,14 +201,15 @@ public:
      * Checks that chunkRanges and shardIds have the same length.
      */
     const std::vector<ChunkType> makeChunks(const std::vector<ChunkRange> chunkRanges,
-                                            const std::vector<ShardId> shardIds) {
+                                            const std::vector<ShardId> shardIds,
+                                            Timestamp timeStamp) {
         ASSERT_EQ(chunkRanges.size(), shardIds.size());
         std::vector<ChunkType> chunks;
 
         for (unsigned long i = 0; i < chunkRanges.size(); ++i) {
             ChunkVersion version(1, 0, OID::gen());
             ChunkType chunk(_nss, chunkRanges[i], version, shardIds[i]);
-            chunk.setHistory({ChunkHistory(_timeStamp, shardIds[i])});
+            chunk.setHistory({ChunkHistory(timeStamp, shardIds[i])});
             chunks.push_back(chunk);
         }
         return chunks;
@@ -200,7 +249,8 @@ public:
 
 private:
     const NamespaceString _nss{"test.foo"};
-    const ShardKeyPattern _shardKeyPattern = makeShardKeyPattern(true);
+    const ShardKeyPattern _shardKeyPattern = ShardKeyPattern(BSON("x"
+                                                                  << "hashed"));
     const std::string _shardName = "testShard";
     const Timestamp _timeStamp{Date_t::now()};
 };
@@ -233,8 +283,10 @@ TEST_F(GenerateInitialHashedSplitChunksTest, NoSplitPoints) {
         nss(), shardKeyPattern(), shardIds[0], timeStamp(), splitPoints, shardIds);
 
     // there should only be one chunk
-    const auto expectedChunks = makeChunks(
-        {ChunkRange(keyPattern().globalMin(), keyPattern().globalMax())}, {shardId("0")});
+    const auto expectedChunks =
+        makeChunks({ChunkRange(keyPattern().globalMin(), keyPattern().globalMax())},
+                   {shardId("0")},
+                   timeStamp());
     assertChunkVectorsAreEqual(expectedChunks, shardCollectionConfig.chunks);
 }
 
@@ -244,8 +296,8 @@ TEST_F(GenerateInitialHashedSplitChunksTest, SplitPointsMoreThanAvailableShards)
         nss(), shardKeyPattern(), shardIds[0], timeStamp(), hashedSplitPoints(), shardIds);
 
     // // chunks should be distributed in a round-robin manner
-    const std::vector<ChunkType> expectedChunks =
-        makeChunks(hashedChunkRanges(), {shardId("0"), shardId("1"), shardId("0"), shardId("1")});
+    const std::vector<ChunkType> expectedChunks = makeChunks(
+        hashedChunkRanges(), {shardId("0"), shardId("1"), shardId("0"), shardId("1")}, timeStamp());
     assertChunkVectorsAreEqual(expectedChunks, shardCollectionConfig.chunks);
 }
 
@@ -256,32 +308,34 @@ TEST_F(GenerateInitialHashedSplitChunksTest,
         nss(), shardKeyPattern(), shardIds[0], timeStamp(), hashedSplitPoints(), shardIds, 2);
 
     // chunks should be distributed in a round-robin manner two chunks at a time
-    const std::vector<ChunkType> expectedChunks =
-        makeChunks(hashedChunkRanges(), {shardId("0"), shardId("0"), shardId("1"), shardId("1")});
+    const std::vector<ChunkType> expectedChunks = makeChunks(
+        hashedChunkRanges(), {shardId("0"), shardId("0"), shardId("1"), shardId("1")}, timeStamp());
     assertChunkVectorsAreEqual(expectedChunks, shardCollectionConfig.chunks);
 }
 
-class GenerateShardCollectionInitialZonedChunksTest : public GenerateInitialSplitChunksTestBase {
+class SingleChunkPerTagSplitPolicyTest : public GenerateInitialSplitChunksTestBase {
 public:
     /**
-     * Calls generateShardCollectionInitialZonedChunks according to the given arguments
-     * and asserts that returned chunks match with the chunks created using expectedChunkRanges
-     * and expectedShardIds.
+     * Calls SingleChunkPerTagSplitPolicy::createFirstChunks() according to the given arguments and
+     * asserts that returned chunks match with the chunks created using expectedChunkRanges and
+     * expectedShardIds.
      */
-    void checkGeneratedInitialZoneChunks(const std::vector<TagsType>& tags,
-                                         const int numShards,
+    void checkGeneratedInitialZoneChunks(const std::vector<ShardType> shards,
+                                         const std::vector<TagsType>& tags,
                                          const std::vector<ChunkRange>& expectedChunkRanges,
-                                         const std::vector<ShardId>& expectedShardIds) {
+                                         const std::vector<ShardId>& expectedShardIds,
+                                         const ShardKeyPattern& shardKeyPattern) {
+        auto opCtx = operationContext();
+        setupShards(shards);
+        shardRegistry()->reload(opCtx);
+        SingleChunkPerTagSplitPolicy splitPolicy(opCtx, tags);
         const auto shardCollectionConfig =
-            InitialSplitPolicy::generateShardCollectionInitialZonedChunks(
-                nss(),
-                shardKeyPattern(),
-                timeStamp(),
-                tags,
-                makeTagToShards(numShards),
-                makeShardIds(numShards));
+            splitPolicy.createFirstChunks(opCtx, shardKeyPattern, {});
+
         const std::vector<ChunkType> expectedChunks =
-            makeChunks(expectedChunkRanges, expectedShardIds);
+            makeChunks(expectedChunkRanges,
+                       expectedShardIds,
+                       LogicalClock::get(opCtx)->getClusterTime().asTimestamp());
         assertChunkVectorsAreEqual(expectedChunks, shardCollectionConfig.chunks);
     }
 
@@ -304,33 +358,30 @@ public:
         return assertGet(TagsType::fromBSON(tagDocBuilder.obj()));
     }
 
-    /**
-     * Returns a map of size numTags mapping _zoneName\d to _shardName\d
-     */
-    StringMap<std::vector<ShardId>> makeTagToShards(const int numTags) {
-        StringMap<std::vector<ShardId>> tagToShards;
-        for (int i = 0; i < numTags; i++) {
-            tagToShards[zoneName(std::to_string(i))] = {shardId(std::to_string(i))};
-        }
-        return tagToShards;
-    }
-
 private:
-    const ShardKeyPattern _shardKeyPattern = makeShardKeyPattern(true);
+    const ShardKeyPattern _shardKeyPattern = ShardKeyPattern(BSON("x"
+                                                                  << "hashed"));
     const std::string _zoneName = "zoneName";
     const std::string _shardKey = "x";
 };
 
-TEST_F(GenerateShardCollectionInitialZonedChunksTest, PredefinedZonesSpanFromMinToMax) {
+TEST_F(SingleChunkPerTagSplitPolicyTest, PredefinedZonesSpanFromMinToMax) {
+    const std::vector<ShardType> kShards{
+        ShardType(shardId("0").toString(), "rs0/shard0:123", {zoneName("0")}),
+        ShardType(shardId("1").toString(), "rs1/shard1:123")};
     const std::vector<ChunkRange> expectedChunkRanges = {
         ChunkRange(keyPattern().globalMin(), keyPattern().globalMax()),  // corresponds to a zone
     };
     const std::vector<TagsType> tags = {makeTag(expectedChunkRanges[0], zoneName("0"))};
     const std::vector<ShardId> expectedShardIds = {shardId("0")};
-    checkGeneratedInitialZoneChunks(tags, 1, expectedChunkRanges, expectedShardIds);
+    checkGeneratedInitialZoneChunks(
+        kShards, tags, expectedChunkRanges, expectedShardIds, ShardKeyPattern(BSON("x" << 1)));
 }
 
-TEST_F(GenerateShardCollectionInitialZonedChunksTest, PredefinedZonesSpanDoNotSpanFromMinToMax) {
+TEST_F(SingleChunkPerTagSplitPolicyTest, PredefinedZonesSpanDoNotSpanFromMinToMax) {
+    const std::vector<ShardType> kShards{
+        ShardType(shardId("0").toString(), "rs0/shard0:123", {zoneName("0")}),
+        ShardType(shardId("1").toString(), "rs1/shard1:123")};
     const std::vector<ChunkRange> expectedChunkRanges = {
         ChunkRange(keyPattern().globalMin(), BSON(shardKey() << 0)),
         ChunkRange(BSON(shardKey() << 0), BSON(shardKey() << 10)),  // corresponds to a zone
@@ -338,30 +389,46 @@ TEST_F(GenerateShardCollectionInitialZonedChunksTest, PredefinedZonesSpanDoNotSp
     };
     const std::vector<TagsType> tags = {makeTag(expectedChunkRanges[1], zoneName("0"))};
     const std::vector<ShardId> expectedShardIds = {shardId("0"), shardId("0"), shardId("1")};
-    checkGeneratedInitialZoneChunks(tags, 2, expectedChunkRanges, expectedShardIds);
+    checkGeneratedInitialZoneChunks(kShards,
+                                    tags,
+                                    expectedChunkRanges,
+                                    expectedShardIds,
+                                    ShardKeyPattern(BSON("x"
+                                                         << "hashed")));
 }
 
-TEST_F(GenerateShardCollectionInitialZonedChunksTest, PredefinedZonesContainGlobalMin) {
+TEST_F(SingleChunkPerTagSplitPolicyTest, PredefinedZonesContainGlobalMin) {
+    const std::vector<ShardType> kShards{
+        ShardType(shardId("0").toString(), "rs0/shard0:123", {zoneName("0")}),
+        ShardType(shardId("1").toString(), "rs1/shard1:123")};
     const std::vector<ChunkRange> expectedChunkRanges = {
         ChunkRange(keyPattern().globalMin(), BSON(shardKey() << 0)),  // corresponds to a zone
         ChunkRange(BSON(shardKey() << 0), keyPattern().globalMax()),
     };
     const std::vector<TagsType> tags = {makeTag(expectedChunkRanges[0], zoneName("0"))};
     const std::vector<ShardId> expectedShardIds = {shardId("0"), shardId("0")};
-    checkGeneratedInitialZoneChunks(tags, 2, expectedChunkRanges, expectedShardIds);
+    checkGeneratedInitialZoneChunks(
+        kShards, tags, expectedChunkRanges, expectedShardIds, ShardKeyPattern(BSON("x" << 1)));
 }
 
-TEST_F(GenerateShardCollectionInitialZonedChunksTest, PredefinedZonesContainGlobalMax) {
+TEST_F(SingleChunkPerTagSplitPolicyTest, PredefinedZonesContainGlobalMax) {
+    const std::vector<ShardType> kShards{
+        ShardType(shardId("0").toString(), "rs0/shard0:123", {zoneName("0")}),
+        ShardType(shardId("1").toString(), "rs1/shard1:123")};
     const std::vector<ChunkRange> expectedChunkRanges = {
         ChunkRange(keyPattern().globalMin(), BSON(shardKey() << 0)),
         ChunkRange(BSON(shardKey() << 0), keyPattern().globalMax()),  // corresponds to a zone
     };
     const std::vector<TagsType> tags = {makeTag(expectedChunkRanges[1], zoneName("0"))};
     const std::vector<ShardId> expectedShardIds = {shardId("0"), shardId("0")};
-    checkGeneratedInitialZoneChunks(tags, 2, expectedChunkRanges, expectedShardIds);
+    checkGeneratedInitialZoneChunks(
+        kShards, tags, expectedChunkRanges, expectedShardIds, ShardKeyPattern(BSON("x" << 1)));
 }
 
-TEST_F(GenerateShardCollectionInitialZonedChunksTest, PredefinedZonesContainGlobalMinAndMax) {
+TEST_F(SingleChunkPerTagSplitPolicyTest, PredefinedZonesContainGlobalMinAndMax) {
+    const std::vector<ShardType> kShards{
+        ShardType(shardId("0").toString(), "rs0/shard0:123", {zoneName("0")}),
+        ShardType(shardId("1").toString(), "rs1/shard1:123", {zoneName("1")})};
     const std::vector<ChunkRange> expectedChunkRanges = {
         ChunkRange(keyPattern().globalMin(), BSON(shardKey() << 0)),  // corresponds to a zone
         ChunkRange(BSON(shardKey() << 0), BSON(shardKey() << 10)),
@@ -370,10 +437,14 @@ TEST_F(GenerateShardCollectionInitialZonedChunksTest, PredefinedZonesContainGlob
     const std::vector<TagsType> tags = {makeTag(expectedChunkRanges[0], zoneName("0")),
                                         makeTag(expectedChunkRanges[2], zoneName("1"))};
     const std::vector<ShardId> expectedShardIds = {shardId("0"), shardId("0"), shardId("1")};
-    checkGeneratedInitialZoneChunks(tags, 2, expectedChunkRanges, expectedShardIds);
+    checkGeneratedInitialZoneChunks(
+        kShards, tags, expectedChunkRanges, expectedShardIds, ShardKeyPattern(BSON("x" << 1)));
 }
 
-TEST_F(GenerateShardCollectionInitialZonedChunksTest, PredefinedZonesContiguous) {
+TEST_F(SingleChunkPerTagSplitPolicyTest, PredefinedZonesContiguous) {
+    const std::vector<ShardType> kShards{
+        ShardType(shardId("0").toString(), "rs0/shard0:123", {zoneName("0")}),
+        ShardType(shardId("1").toString(), "rs1/shard1:123", {zoneName("1")})};
     const std::vector<ChunkRange> expectedChunkRanges = {
         ChunkRange(keyPattern().globalMin(), BSON(shardKey() << 0)),
         ChunkRange(BSON(shardKey() << 0), BSON(shardKey() << 10)),   // corresponds to a zone
@@ -384,10 +455,15 @@ TEST_F(GenerateShardCollectionInitialZonedChunksTest, PredefinedZonesContiguous)
                                         makeTag(expectedChunkRanges[2], zoneName("0"))};
     const std::vector<ShardId> expectedShardIds = {
         shardId("0"), shardId("1"), shardId("0"), shardId("1")};
-    checkGeneratedInitialZoneChunks(tags, 2, expectedChunkRanges, expectedShardIds);
+    checkGeneratedInitialZoneChunks(
+        kShards, tags, expectedChunkRanges, expectedShardIds, ShardKeyPattern(BSON("x" << 1)));
 }
 
-TEST_F(GenerateShardCollectionInitialZonedChunksTest, PredefinedZonesNotContiguous) {
+TEST_F(SingleChunkPerTagSplitPolicyTest, PredefinedZonesNotContiguous) {
+    const std::vector<ShardType> kShards{
+        ShardType(shardId("0").toString(), "rs0/shard0:123", {zoneName("0")}),
+        ShardType(shardId("1").toString(), "rs1/shard1:123", {zoneName("1")}),
+        ShardType(shardId("2").toString(), "rs1/shard1:123")};
     const std::vector<ChunkRange> expectedChunkRanges = {
         ChunkRange(keyPattern().globalMin(), BSON(shardKey() << 0)),
         ChunkRange(BSON(shardKey() << 0), BSON(shardKey() << 10)),  // corresponds to a zone
@@ -399,10 +475,14 @@ TEST_F(GenerateShardCollectionInitialZonedChunksTest, PredefinedZonesNotContiguo
                                         makeTag(expectedChunkRanges[3], zoneName("1"))};
     const std::vector<ShardId> expectedShardIds = {
         shardId("0"), shardId("0"), shardId("1"), shardId("1"), shardId("2")};
-    checkGeneratedInitialZoneChunks(tags, 3, expectedChunkRanges, expectedShardIds);
+    checkGeneratedInitialZoneChunks(
+        kShards, tags, expectedChunkRanges, expectedShardIds, ShardKeyPattern(BSON("x" << 1)));
 }
 
-TEST_F(GenerateShardCollectionInitialZonedChunksTest, NumRemainingChunksGreaterThanNumShards) {
+TEST_F(SingleChunkPerTagSplitPolicyTest, NumRemainingChunksGreaterThanNumShards) {
+    const std::vector<ShardType> kShards{
+        ShardType(shardId("0").toString(), "rs0/shard0:123", {zoneName("0")}),
+        ShardType(shardId("1").toString(), "rs1/shard1:123", {zoneName("1")})};
     const std::vector<ChunkRange> expectedChunkRanges = {
         ChunkRange(keyPattern().globalMin(), BSON(shardKey() << 0)),
         ChunkRange(BSON(shardKey() << 0), BSON(shardKey() << 10)),  // corresponds to a zone
@@ -415,12 +495,15 @@ TEST_F(GenerateShardCollectionInitialZonedChunksTest, NumRemainingChunksGreaterT
     // shard assignment should wrap around to the first shard
     const std::vector<ShardId> expectedShardIds = {
         shardId("0"), shardId("0"), shardId("1"), shardId("1"), shardId("0")};
-    checkGeneratedInitialZoneChunks(tags, 2, expectedChunkRanges, expectedShardIds);
+    checkGeneratedInitialZoneChunks(
+        kShards, tags, expectedChunkRanges, expectedShardIds, ShardKeyPattern(BSON("x" << 1)));
 }
 
-TEST_F(GenerateShardCollectionInitialZonedChunksTest, MultipleChunksToOneZoneWithMultipleShards) {
+TEST_F(SingleChunkPerTagSplitPolicyTest, MultipleChunksToOneZoneWithMultipleShards) {
     const auto zone0 = zoneName("Z0");
-
+    const std::vector<ShardType> kShards{
+        ShardType(shardId("0").toString(), "rs0/shard0:123", {zone0}),
+        ShardType(shardId("1").toString(), "rs1/shard1:123", {zone0})};
     const std::vector<ChunkRange> expectedChunkRanges = {
         ChunkRange(keyPattern().globalMin(), BSON(shardKey() << 0)),   // zone0
         ChunkRange(BSON(shardKey() << 0), BSON(shardKey() << 10)),     // gap
@@ -432,22 +515,18 @@ TEST_F(GenerateShardCollectionInitialZonedChunksTest, MultipleChunksToOneZoneWit
         makeTag(expectedChunkRanges.at(2), zone0),
         makeTag(expectedChunkRanges.at(3), zone0),
     };
-    const StringMap<std::vector<ShardId>> tagToShards = {{zone0, {{"S0"}, {"S1"}}}};
-    const std::vector<ShardId> shardIdsForGaps = {{"G0"}};
-    const std::vector<ShardId> expectedShardIds = {{"S0"}, {"G0"}, {"S1"}, {"S0"}};
-    const auto shardCollectionConfig =
-        InitialSplitPolicy::generateShardCollectionInitialZonedChunks(
-            nss(), shardKeyPattern(), timeStamp(), tags, tagToShards, shardIdsForGaps);
-
-    const std::vector<ChunkType> expectedChunks = makeChunks(expectedChunkRanges, expectedShardIds);
-    assertChunkVectorsAreEqual(expectedChunks, shardCollectionConfig.chunks);
+    const std::vector<ShardId> expectedShardIds = {
+        shardId("0"), shardId("0"), shardId("1"), shardId("0")};
+    checkGeneratedInitialZoneChunks(
+        kShards, tags, expectedChunkRanges, expectedShardIds, ShardKeyPattern(BSON("x" << 1)));
 };
 
-TEST_F(GenerateShardCollectionInitialZonedChunksTest,
-       MultipleChunksToInterleavedZonesWithMultipleShards) {
+TEST_F(SingleChunkPerTagSplitPolicyTest, MultipleChunksToInterleavedZonesWithMultipleShards) {
     const auto zone0 = zoneName("Z0");
     const auto zone1 = zoneName("Z1");
-
+    const std::vector<ShardType> kShards{
+        ShardType(shardId("0").toString(), "rs0/shard0:123", {zone0, zone1}),
+        ShardType(shardId("1").toString(), "rs1/shard1:123", {zone0, zone1})};
     const std::vector<ChunkRange> expectedChunkRanges = {
         ChunkRange(keyPattern().globalMin(), BSON(shardKey() << 0)),   // zone0
         ChunkRange(BSON(shardKey() << 0), BSON(shardKey() << 10)),     // zone1
@@ -459,34 +538,1165 @@ TEST_F(GenerateShardCollectionInitialZonedChunksTest,
         makeTag(expectedChunkRanges.at(1), zone1),
         makeTag(expectedChunkRanges.at(3), zone0),
     };
-    const StringMap<std::vector<ShardId>> tagToShards = {
-        {zone0, {{"Z0-S0"}, {"Z0-S1"}}},
-        {zone1, {{"Z1-S0"}, {"Z1-S1"}}},
-    };
-    const std::vector<ShardId> shardIdsForGaps = {{"G0"}};
-    const std::vector<ShardId> expectedShardIds = {{"Z0-S0"}, {"Z1-S0"}, {"G0"}, {"Z0-S1"}};
-    const auto shardCollectionConfig =
-        InitialSplitPolicy::generateShardCollectionInitialZonedChunks(
-            nss(), shardKeyPattern(), timeStamp(), tags, tagToShards, shardIdsForGaps);
 
-    const std::vector<ChunkType> expectedChunks = makeChunks(expectedChunkRanges, expectedShardIds);
-    assertChunkVectorsAreEqual(expectedChunks, shardCollectionConfig.chunks);
+    const std::vector<ShardId> expectedShardIds = {
+        shardId("0"), shardId("0"), shardId("0"), shardId("1")};
+    checkGeneratedInitialZoneChunks(
+        kShards, tags, expectedChunkRanges, expectedShardIds, ShardKeyPattern(BSON("x" << 1)));
 };
 
-TEST_F(GenerateShardCollectionInitialZonedChunksTest, ZoneNotAssociatedWithAnyShardShouldFail) {
+TEST_F(SingleChunkPerTagSplitPolicyTest, ZoneNotAssociatedWithAnyShardShouldFail) {
+    const std::vector<ShardType> kShards{
+        ShardType(shardId("0").toString(), "rs0/shard0:123", {zoneName("0")}),
+        ShardType(shardId("1").toString(), "rs1/shard1:123")};
+    setupShards(kShards);
+    shardRegistry()->reload(operationContext());
     const auto zone1 = zoneName("0");
     const auto zone2 = zoneName("1");
 
     const std::vector<TagsType> tags{
         makeTag(ChunkRange(keyPattern().globalMin(), BSON(shardKey() << 0)), zone1),
         makeTag(ChunkRange(BSON(shardKey() << 0), keyPattern().globalMax()), zone2)};
-    const StringMap<std::vector<ShardId>> tagToShards{{zone1, {ShardId("Shard0")}}, {zone2, {}}};
 
+    SingleChunkPerTagSplitPolicy splitPolicy(operationContext(), tags);
+
+    ASSERT_THROWS_CODE(splitPolicy.createFirstChunks(operationContext(), shardKeyPattern(), {}),
+                       AssertionException,
+                       50973);
+}
+
+class PresplitHashedZonesChunksTest : public SingleChunkPerTagSplitPolicyTest {
+public:
+    /**
+     * Calls PresplitHashedZonesSplitPolicy::createFirstChunks() according to the given arguments
+     * and asserts that returned chunks match with the chunks created using expectedChunkRanges and
+     * expectedShardIds.
+     */
+    void checkGeneratedInitialZoneChunks(const std::vector<TagsType>& tags,
+                                         const std::vector<ChunkRange>& expectedChunkRanges,
+                                         const std::vector<ShardId>& expectedShardIds,
+                                         const ShardKeyPattern& shardKeyPattern,
+                                         int numInitialChunk,
+                                         bool isCollEmpty = true) {
+        PresplitHashedZonesSplitPolicy splitPolicy(
+            operationContext(), shardKeyPattern, tags, numInitialChunk, isCollEmpty);
+        const auto shardCollectionConfig =
+            splitPolicy.createFirstChunks(operationContext(), shardKeyPattern, {});
+
+        const std::vector<ChunkType> expectedChunks =
+            makeChunks(expectedChunkRanges,
+                       expectedShardIds,
+                       LogicalClock::get(operationContext())->getClusterTime().asTimestamp());
+        assertChunkVectorsAreEqual(expectedChunks, shardCollectionConfig.chunks);
+    }
+};
+
+/**
+ * Builds a BSON object using the lower bound of the given tag. The hashed field will be replaced
+ * with the input value.
+ */
+BSONObj buildObj(const ShardKeyPattern& shardKeyPattern, TagsType tag, long long value) {
+    BSONObjBuilder bob;
+    for (auto&& elem : tag.getMinKey()) {
+        if (shardKeyPattern.getHashedField().fieldNameStringData() == elem.fieldNameStringData()) {
+            bob.appendNumber(elem.fieldNameStringData(), value);
+        } else {
+            bob.append(elem);
+        }
+    }
+    return bob.obj();
+};
+
+/**
+ * Generates chunk ranges for each tag using the split points.
+ */
+std::vector<ChunkRange> buildExpectedChunkRanges(const std::vector<TagsType>& tags,
+                                                 const ShardKeyPattern& shardKeyPattern,
+                                                 const std::vector<int> numChunksPerTag) {
+
+    // The hashed values are intentionally hard coded here, so that the behaviour of
+    // 'InitialSplitPolicy::calculateHashedSplitPoints()' is also tested.
+    auto getHashedSplitPoints = [&](int numChunks) -> std::vector<long long> {
+        switch (numChunks) {
+            case 1:
+                return {};
+            case 2:
+                return {0LL};
+            case 3:
+                return {-3074457345618258602LL, 3074457345618258602LL};
+            case 4:
+                return {-4611686018427387902LL, 0LL, 4611686018427387902LL};
+            case 5:
+                return {-5534023222112865483LL,
+                        -1844674407370955161LL,
+                        1844674407370955161LL,
+                        5534023222112865483LL};
+            case 6:
+                return {-6148914691236517204LL,
+                        -3074457345618258602LL,
+                        0LL,
+                        3074457345618258602LL,
+                        6148914691236517204LL};
+            default:
+                auto splitPoints = InitialSplitPolicy::calculateHashedSplitPoints(
+                    shardKeyPattern, BSONObj(), numChunks);
+                auto field = shardKeyPattern.getHashedField();
+                std::vector<long long> output;
+                for (auto&& splitPoint : splitPoints) {
+                    output.push_back(splitPoint[field.fieldName()].numberLong());
+                }
+                return output;
+        };
+        MONGO_UNREACHABLE;
+    };
+    ASSERT(!tags.empty() && tags.size() == numChunksPerTag.size());
+
+    std::vector<ChunkRange> output;
+    output.reserve(numChunksPerTag.size() * tags.size());
+
+    // Global MinKey to first tag's start value. We only add this chunk if the first tag's min bound
+    // isn't already global MinKey.
+    if (tags[0].getMinKey().woCompare(shardKeyPattern.getKeyPattern().globalMin())) {
+        output.push_back(
+            ChunkRange(shardKeyPattern.getKeyPattern().globalMin(), tags[0].getMinKey()));
+    }
+    for (size_t tagIdx = 0; tagIdx < tags.size(); tagIdx++) {
+        auto tag = tags[tagIdx];
+
+        // If there is a gap between consecutive tags (previous tag's MaxKey and current tag's
+        // MinKey), create a chunk to fill the gap.
+        if ((tagIdx != 0 && (tags[tagIdx - 1].getMaxKey().woCompare(tag.getMinKey())))) {
+            output.push_back(ChunkRange(tags[tagIdx - 1].getMaxKey(), tag.getMinKey()));
+        }
+
+        std::vector<long long> hashedSplitValues = getHashedSplitPoints(numChunksPerTag[tagIdx]);
+        // Generated single chunk for tag if there are no split points.
+        if (hashedSplitValues.empty()) {
+            output.push_back(ChunkRange(tag.getMinKey(), tag.getMaxKey()));
+            continue;
+        }
+        output.push_back(
+            ChunkRange(tag.getMinKey(), buildObj(shardKeyPattern, tag, hashedSplitValues[0])));
+
+        // Generate 'n-1' chunks using the split values.
+        for (size_t i = 0; i < hashedSplitValues.size() - 1; i++) {
+            output.push_back(ChunkRange(buildObj(shardKeyPattern, tag, hashedSplitValues[i]),
+                                        buildObj(shardKeyPattern, tag, hashedSplitValues[i + 1])));
+        }
+        output.push_back(ChunkRange(
+            buildObj(shardKeyPattern, tag, hashedSplitValues[hashedSplitValues.size() - 1]),
+            tag.getMaxKey()));
+    }
+    // Last tag's end value to global MaxKey. We only add this chunk if the last tag's max bound
+    // isn't already global MaxKey.
+    if (tags[tags.size() - 1].getMaxKey().woCompare(shardKeyPattern.getKeyPattern().globalMax())) {
+        output.push_back(ChunkRange(tags[tags.size() - 1].getMaxKey(),
+                                    shardKeyPattern.getKeyPattern().globalMax()));
+    }
+    return output;
+}
+
+TEST_F(PresplitHashedZonesChunksTest, WithHashedPrefix) {
+    const std::vector<ShardType> kShards{
+        ShardType(shardId("0").toString(), "rs0/shard0:123", {zoneName("0")}),
+        ShardType(shardId("1").toString(), "rs1/shard1:123", {zoneName("0")}),
+        ShardType(shardId("2").toString(), "rs1/shard1:123", {zoneName("0")})};
+    setupShards(kShards);
+    shardRegistry()->reload(operationContext());
+    auto shardKeyPattern = ShardKeyPattern(BSON("x"
+                                                << "hashed"
+                                                << "y" << 1));
+    const auto zoneRange =
+        ChunkRange(BSON("x" << MINKEY << "y" << MINKEY), BSON("x" << MAXKEY << "y" << MAXKEY));
+    const std::vector<TagsType> tags = {makeTag(zoneRange, zoneName("0"))};
+
+    // numInitialChunks = 0.
+    std::vector<ChunkRange> expectedChunkRanges =
+        buildExpectedChunkRanges(tags, shardKeyPattern, {2 * 3});
+    std::vector<ShardId> expectedShardIds = {
+        shardId("0"), shardId("0"), shardId("1"), shardId("1"), shardId("2"), shardId("2")};
+    checkGeneratedInitialZoneChunks(
+        tags, expectedChunkRanges, expectedShardIds, shardKeyPattern, 0 /* numInitialChunks*/);
+
+    // numInitialChunks = 1.
+    expectedChunkRanges = buildExpectedChunkRanges(tags, shardKeyPattern, {3});
+    expectedShardIds = {shardId("0"), shardId("1"), shardId("2")};
+    checkGeneratedInitialZoneChunks(
+        tags, expectedChunkRanges, expectedShardIds, shardKeyPattern, 1 /* numInitialChunks*/);
+
+    // numInitialChunks = 4.
+    expectedChunkRanges = buildExpectedChunkRanges(tags, shardKeyPattern, {6});
+    expectedShardIds = {
+        shardId("0"), shardId("0"), shardId("1"), shardId("1"), shardId("2"), shardId("2")};
+    checkGeneratedInitialZoneChunks(
+        tags, expectedChunkRanges, expectedShardIds, shardKeyPattern, 4 /* numInitialChunks*/);
+}
+
+
+TEST_F(PresplitHashedZonesChunksTest, SingleZone) {
+    const std::vector<ShardType> kShards{
+        ShardType(shardId("0").toString(), "rs0/shard0:123", {zoneName("0")}),
+        ShardType(shardId("1").toString(), "rs1/shard1:123")};
+    setupShards(kShards);
+    shardRegistry()->reload(operationContext());
+    auto shardKeyPattern = ShardKeyPattern(BSON("x" << 1 << "y"
+                                                    << "hashed"));
+    const auto zoneRange = ChunkRange(BSON("x"
+                                           << "UK"
+                                           << "y" << MINKEY),
+                                      BSON("x"
+                                           << "US"
+                                           << "y" << MINKEY));
+    const std::vector<TagsType> tags = {makeTag(zoneRange, zoneName("0"))};
+
+    // numInitialChunks = 0.
+    std::vector<ChunkRange> expectedChunkRanges =
+        buildExpectedChunkRanges(tags, shardKeyPattern, {2});
+    std::vector<ShardId> expectedShardIds = {
+        shardId("0"), shardId("0"), shardId("0"), shardId("1")};
+    checkGeneratedInitialZoneChunks(
+        tags, expectedChunkRanges, expectedShardIds, shardKeyPattern, 0 /* numInitialChunks*/);
+
+    // numInitialChunks = 1.
+    expectedChunkRanges = buildExpectedChunkRanges(tags, shardKeyPattern, {1});
+    expectedShardIds = {shardId("0"), shardId("0"), shardId("1")};
+    checkGeneratedInitialZoneChunks(
+        tags, expectedChunkRanges, expectedShardIds, shardKeyPattern, 1 /* numInitialChunks*/);
+
+    // numInitialChunks = 3.
+    expectedChunkRanges = buildExpectedChunkRanges(tags, shardKeyPattern, {3});
+    expectedShardIds = {shardId("0"), shardId("0"), shardId("0"), shardId("0"), shardId("1")};
+    checkGeneratedInitialZoneChunks(
+        tags, expectedChunkRanges, expectedShardIds, shardKeyPattern, 3 /* numInitialChunks*/);
+}
+
+TEST_F(PresplitHashedZonesChunksTest, WithMultipleZonesContiguous) {
+    const std::vector<ShardType> kShards{
+        ShardType(shardId("0").toString(), "rs0/shard0:123", {zoneName("0")}),
+        ShardType(shardId("1").toString(), "rs1/shard1:123", {zoneName("1")}),
+        ShardType(shardId("2").toString(), "rs1/shard1:123", {zoneName("2")})};
+    setupShards(kShards);
+    shardRegistry()->reload(operationContext());
+
+    auto shardKeyPattern = ShardKeyPattern(BSON("country" << 1 << "city" << 1 << "hashedField"
+                                                          << "hashed"
+                                                          << "suffix" << 1));
+    const std::vector<TagsType> tags = {
+        makeTag(ChunkRange(BSON("country"
+                                << "IE"
+                                << "city"
+                                << "Dublin"
+                                << "hashedField" << MINKEY << "suffix" << MINKEY),
+                           BSON("country"
+                                << "UK"
+                                << "city"
+                                << "London"
+                                << "hashedField" << MINKEY << "suffix" << MINKEY)),
+                zoneName("0")),
+        makeTag(ChunkRange(BSON("country"
+                                << "UK"
+                                << "city"
+                                << "London"
+                                << "hashedField" << MINKEY << "suffix" << MINKEY),
+                           BSON("country"
+                                << "US"
+                                << "city"
+                                << "NewYork"
+                                << "hashedField" << MINKEY << "suffix" << MINKEY)),
+                zoneName("1")),
+        makeTag(
+            ChunkRange(BSON("country"
+                            << "US"
+                            << "city"
+                            << "NewYork"
+                            << "hashedField" << MINKEY << "suffix" << MINKEY),
+                       BSON("country"
+                            << "US"
+                            << "city" << MAXKEY << "hashedField" << MINKEY << "suffix" << MINKEY)),
+            zoneName("2"))};
+
+    // numInitialChunks = 0.
+    // This should have 8 chunks, 2 for each zone and 2 boundaries.
+    std::vector<ChunkRange> expectedChunkRanges =
+        buildExpectedChunkRanges(tags, shardKeyPattern, {2, 2, 2});
+
+    std::vector<ShardId> expectedShardForEachChunk = {
+        shardId("0"),  // Lower bound.
+        shardId("0"),
+        shardId("0"),
+        shardId("1"),
+        shardId("1"),
+        shardId("2"),
+        shardId("2"),
+        shardId("1")  // Upper bound.
+    };
+    checkGeneratedInitialZoneChunks(tags,
+                                    expectedChunkRanges,
+                                    expectedShardForEachChunk,
+                                    shardKeyPattern,
+                                    0 /* numInitialChunks*/);
+
+    // numInitialChunks = 1.
+    // This should have 5 chunks, 1 for each zone and 2 boundaries.
+    expectedChunkRanges = buildExpectedChunkRanges(tags, shardKeyPattern, {1, 1, 1});
+
+    expectedShardForEachChunk = {
+        shardId("0"),  // Lower bound.
+        shardId("0"),
+        shardId("1"),
+        shardId("2"),
+        shardId("1")  // Upper bound.
+    };
+    checkGeneratedInitialZoneChunks(tags,
+                                    expectedChunkRanges,
+                                    expectedShardForEachChunk,
+                                    shardKeyPattern,
+                                    1 /* numInitialChunks*/);
+
+    // numInitialChunks = 10.
+    // This should have 14 chunks, 4 for each zone and 2 boundaries.
+    expectedChunkRanges = buildExpectedChunkRanges(tags, shardKeyPattern, {4, 4, 4});
+
+    expectedShardForEachChunk = {
+        shardId("0"),  // Lower bound.
+        shardId("0"),
+        shardId("0"),
+        shardId("0"),
+        shardId("0"),
+        shardId("1"),
+        shardId("1"),
+        shardId("1"),
+        shardId("1"),
+        shardId("2"),
+        shardId("2"),
+        shardId("2"),
+        shardId("2"),
+        shardId("1")  // Upper bound.
+    };
+    checkGeneratedInitialZoneChunks(tags,
+                                    expectedChunkRanges,
+                                    expectedShardForEachChunk,
+                                    shardKeyPattern,
+                                    10 /* numInitialChunks*/);
+}
+
+TEST_F(PresplitHashedZonesChunksTest, MultipleContiguousZonesWithEachZoneHavingMultipleShards) {
+    const std::vector<ShardType> kShards{
+        ShardType(shardId("0").toString(), "rs0/shard0:123", {zoneName("0")}),
+        ShardType(shardId("1").toString(), "rs1/shard5:123"),
+        ShardType(shardId("2").toString(), "rs0/shard1:123", {zoneName("1")}),
+        ShardType(shardId("3").toString(), "rs0/shard2:123", {zoneName("0")}),
+        ShardType(shardId("4").toString(), "rs1/shard3:123", {zoneName("1")}),
+        ShardType(shardId("5").toString(), "rs1/shard4:123", {zoneName("0")})};
+    setupShards(kShards);
+    shardRegistry()->reload(operationContext());
+
+    auto shardKeyPattern = ShardKeyPattern(BSON("country" << 1 << "city" << 1 << "hashedField"
+                                                          << "hashed"
+                                                          << "suffix" << 1));
+    const std::vector<TagsType> tags = {
+        makeTag(ChunkRange(BSON("country"
+                                << "IE"
+                                << "city"
+                                << "Dublin"
+                                << "hashedField" << MINKEY << "suffix" << MINKEY),
+                           BSON("country"
+                                << "UK"
+                                << "city"
+                                << "London"
+                                << "hashedField" << MINKEY << "suffix" << MINKEY)),
+                zoneName("0")),
+        makeTag(ChunkRange(BSON("country"
+                                << "UK"
+                                << "city"
+                                << "London"
+                                << "hashedField" << MINKEY << "suffix" << MINKEY),
+                           BSON("country"
+                                << "US"
+                                << "city"
+                                << "Los Angeles"
+                                << "hashedField" << MINKEY << "suffix" << MINKEY)),
+                zoneName("1"))};
+
+    // numInitialChunks = 0.
+    // This should have 12 chunks, 6 for zone0, 4 for zone1 and 2 boundaries.
+    std::vector<ChunkRange> expectedChunkRanges =
+        buildExpectedChunkRanges(tags, shardKeyPattern, {6, 4});
+
+    std::vector<ShardId> expectedShardForEachChunk = {
+        shardId("0"),  // Lower bound.
+        shardId("0"),  // zone 0.
+        shardId("0"),  // zone 0.
+        shardId("3"),  // zone 0.
+        shardId("3"),  // zone 0.
+        shardId("5"),  // zone 0.
+        shardId("5"),  // zone 0.
+        shardId("2"),  // zone 1.
+        shardId("2"),  // zone 1.
+        shardId("4"),  // zone 1.
+        shardId("4"),  // zone 1.
+        shardId("1")   // Upper bound.
+    };
+    checkGeneratedInitialZoneChunks(tags,
+                                    expectedChunkRanges,
+                                    expectedShardForEachChunk,
+                                    shardKeyPattern,
+                                    0 /* numInitialChunks*/);
+
+    // numInitialChunks = 1.
+    // This should have 7 chunks, 5 for all zones and 2 boundaries.
+    expectedChunkRanges = buildExpectedChunkRanges(tags, shardKeyPattern, {3, 2});
+
+    expectedShardForEachChunk = {
+        shardId("0"),  // Lower bound.
+        shardId("0"),  // zone0.
+        shardId("3"),  // zone0.
+        shardId("5"),  // zone0.
+        shardId("2"),  // zone1.
+        shardId("4"),  // zone1.
+        shardId("1")   // Upper bound.
+    };
+    checkGeneratedInitialZoneChunks(tags,
+                                    expectedChunkRanges,
+                                    expectedShardForEachChunk,
+                                    shardKeyPattern,
+                                    1 /* numInitialChunks*/);
+
+    // numInitialChunks = 5.
+    // This should have 7 chunks, 5 for all zones and 2 boundaries.
+    expectedChunkRanges = buildExpectedChunkRanges(tags, shardKeyPattern, {3, 2});
+
+    expectedShardForEachChunk = {
+        shardId("0"),  // Lower bound.
+        shardId("0"),  // zone0.
+        shardId("3"),  // zone0.
+        shardId("5"),  // zone0.
+        shardId("2"),  // zone1.
+        shardId("4"),  // zone1.
+        shardId("1")   // Upper bound.
+    };
+    checkGeneratedInitialZoneChunks(tags,
+                                    expectedChunkRanges,
+                                    expectedShardForEachChunk,
+                                    shardKeyPattern,
+                                    5 /* numInitialChunks*/);
+}
+
+TEST_F(PresplitHashedZonesChunksTest, MultipleZonesWithGaps) {
+    const std::vector<ShardType> kShards{
+        ShardType(shardId("0").toString(), "rs0/shard0:123", {zoneName("0")}),
+        ShardType(shardId("noZone").toString(), "rs1/shard1:123"),
+        ShardType(shardId("1").toString(), "rs1/shard1:123", {zoneName("1")}),
+        ShardType(shardId("2").toString(), "rs1/shard1:123", {zoneName("2")})};
+    setupShards(kShards);
+    shardRegistry()->reload(operationContext());
+
+    auto shardKeyPattern = ShardKeyPattern(BSON("country" << 1 << "city" << 1 << "hashedField"
+                                                          << "hashed"
+                                                          << "suffix" << 1));
+    const std::vector<TagsType> tags = {
+        makeTag(ChunkRange(BSON("country"
+                                << "IE"
+                                << "city"
+                                << "Dublin"
+                                << "hashedField" << MINKEY << "suffix" << MINKEY),
+                           BSON("country"
+                                << "UK"
+                                << "city" << MINKEY << "hashedField" << MINKEY << "suffix"
+                                << "SomeValue")),
+                zoneName("0")),
+        makeTag(ChunkRange(BSON("country"
+                                << "UK"
+                                << "city"
+                                << "London"
+                                << "hashedField" << MINKEY << "suffix" << MINKEY),
+                           BSON("country"
+                                << "US"
+                                << "city" << MINKEY << "hashedField" << 100LL << "suffix"
+                                << "someValue")),
+                zoneName("1")),
+        makeTag(
+            ChunkRange(BSON("country"
+                            << "US"
+                            << "city"
+                            << "NewYork"
+                            << "hashedField" << MINKEY << "suffix" << MINKEY),
+                       BSON("country"
+                            << "US"
+                            << "city" << MAXKEY << "hashedField" << MINKEY << "suffix" << MINKEY)),
+            zoneName("2"))};
+
+    // numInitialChunks = 0.
+    // This should have 10 chunks, 2 for each zone (6), 2 gaps and 2 boundaries.
+    std::vector<ChunkRange> expectedChunkRanges =
+        buildExpectedChunkRanges(tags, shardKeyPattern, {2, 2, 2});
+    // The holes should use round-robin to choose a shard.
+    std::vector<ShardId> expectedShardForEachChunk = {
+        shardId("0"),  // LowerBound.
+        shardId("0"),
+        shardId("0"),
+        shardId("1"),  // Hole.
+        shardId("1"),
+        shardId("1"),
+        shardId("2"),  // Hole.
+        shardId("2"),
+        shardId("2"),
+        shardId("noZone"),  // UpperBound.
+    };
+    checkGeneratedInitialZoneChunks(tags,
+                                    expectedChunkRanges,
+                                    expectedShardForEachChunk,
+                                    shardKeyPattern,
+                                    0 /* numInitialChunks*/);
+
+    // numInitialChunks = 1.
+    // This should have 7 chunks, 1 for each zone (3), 2 gaps and 2 boundaries.
+    expectedChunkRanges = buildExpectedChunkRanges(tags, shardKeyPattern, {1, 1, 1});
+    // The holes should use round-robin to choose a shard.
+    expectedShardForEachChunk = {
+        shardId("0"),  // LowerBound.
+        shardId("0"),
+        shardId("1"),  // Hole.
+        shardId("1"),
+        shardId("2"),  // Hole.
+        shardId("2"),
+        shardId("noZone"),  // UpperBound.
+    };
+    checkGeneratedInitialZoneChunks(tags,
+                                    expectedChunkRanges,
+                                    expectedShardForEachChunk,
+                                    shardKeyPattern,
+                                    1 /* numInitialChunks*/);
+
+    // numInitialChunks = 12.
+    // This should have 16 chunks, 4 for each zone (12), 2 gaps and 2 boundaries.
+    expectedChunkRanges = buildExpectedChunkRanges(tags, shardKeyPattern, {4, 4, 4});
+    // The holes should use round-robin to choose a shard.
+    expectedShardForEachChunk = {
+        shardId("0"),  // LowerBound.
+        shardId("0"),
+        shardId("0"),
+        shardId("0"),
+        shardId("0"),
+        shardId("1"),  // Hole.
+        shardId("1"),
+        shardId("1"),
+        shardId("1"),
+        shardId("1"),
+        shardId("2"),  // Hole.
+        shardId("2"),
+        shardId("2"),
+        shardId("2"),
+        shardId("2"),
+        shardId("noZone"),  // UpperBound.
+    };
+    checkGeneratedInitialZoneChunks(tags,
+                                    expectedChunkRanges,
+                                    expectedShardForEachChunk,
+                                    shardKeyPattern,
+                                    12 /* numInitialChunks*/);
+}
+
+TEST_F(PresplitHashedZonesChunksTest, MultipleContiguousZonesWithEachShardHavingMultipleZones) {
+    const std::vector<ShardType> kShards{
+        ShardType(shardId("0").toString(), "rs0/shard0:123", {zoneName("0"), zoneName("2")}),
+        ShardType(shardId("1").toString(),
+                  "rs1/shard1:123",
+                  {zoneName("1"), zoneName("2"), zoneName("3")}),
+        ShardType(shardId("2").toString(), "rs1/shard1:123")};
+    setupShards(kShards);
+    shardRegistry()->reload(operationContext());
+
+    auto shardKeyPattern = ShardKeyPattern(BSON("country" << 1 << "city" << 1 << "hashedField"
+                                                          << "hashed"
+                                                          << "suffix" << 1));
+
+    // No gap between the zone ranges.
+    const std::vector<TagsType> tags = {
+        makeTag(ChunkRange(BSON("country"
+                                << "IE"
+                                << "city"
+                                << "Dublin"
+                                << "hashedField" << MINKEY << "suffix" << MINKEY),
+                           BSON("country"
+                                << "UK"
+                                << "city"
+                                << "London"
+                                << "hashedField" << MINKEY << "suffix" << MINKEY)),
+                zoneName("0")),
+        makeTag(ChunkRange(BSON("country"
+                                << "UK"
+                                << "city"
+                                << "London"
+                                << "hashedField" << MINKEY << "suffix" << MINKEY),
+                           BSON("country"
+                                << "US"
+                                << "city"
+                                << "Los Angeles"
+                                << "hashedField" << MINKEY << "suffix" << MINKEY)),
+                zoneName("1")),
+        makeTag(ChunkRange(BSON("country"
+                                << "US"
+                                << "city"
+                                << "Los Angeles"
+                                << "hashedField" << MINKEY << "suffix" << MINKEY),
+                           BSON("country"
+                                << "US"
+                                << "city"
+                                << "New York"
+                                << "hashedField" << MINKEY << "suffix" << MINKEY)),
+                zoneName("2")),
+        makeTag(
+            ChunkRange(BSON("country"
+                            << "US"
+                            << "city"
+                            << "New York"
+                            << "hashedField" << MINKEY << "suffix" << MINKEY),
+                       BSON("country"
+                            << "US"
+                            << "city" << MAXKEY << "hashedField" << MINKEY << "suffix" << MINKEY)),
+            zoneName("3"))};
+
+    // numInitialChunks = 0.
+    // This should have 7 chunks, 5 for all zones and 2 boundaries.
+    std::vector<ChunkRange> expectedChunkRanges =
+        buildExpectedChunkRanges(tags, shardKeyPattern, {1, 1, 2, 1} /* numChunksPerTag*/);
+
+    std::vector<ShardId> expectedShardForEachChunk = {
+        shardId("0"),  // Lower bound.
+        shardId("0"),  // zone0.
+        shardId("1"),  // zone1.
+        shardId("0"),  // zone2.
+        shardId("1"),  // zone2.
+        shardId("1"),  // zone3.
+        shardId("1")   // Upper bound.
+    };
+    checkGeneratedInitialZoneChunks(tags,
+                                    expectedChunkRanges,
+                                    expectedShardForEachChunk,
+                                    shardKeyPattern,
+                                    0 /* numInitialChunks*/);
+
+    // numInitialChunks = 1.
+    // This should have 7 chunks, 5 for all zones and 2 boundaries.
+    expectedChunkRanges =
+        buildExpectedChunkRanges(tags, shardKeyPattern, {1, 1, 2, 1} /* numChunksPerTag*/);
+
+    expectedShardForEachChunk = {
+        shardId("0"),  // Lower bound.
+        shardId("0"),  // zone0.
+        shardId("1"),  // zone1.
+        shardId("0"),  // zone2.
+        shardId("1"),  // zone2.
+        shardId("1"),  // zone3.
+        shardId("1")   // Upper bound.
+    };
+    checkGeneratedInitialZoneChunks(tags,
+                                    expectedChunkRanges,
+                                    expectedShardForEachChunk,
+                                    shardKeyPattern,
+                                    1 /* numInitialChunks*/);
+
+    // numInitialChunks = 7.
+    // This should have 10 chunks, 10 for all zones and 2 boundaries.
+    expectedChunkRanges =
+        buildExpectedChunkRanges(tags, shardKeyPattern, {2, 2, 4, 2} /* numChunksPerTag*/);
+
+    expectedShardForEachChunk = {
+        shardId("0"),  // Lower bound.
+        shardId("0"),  // zone0.
+        shardId("0"),  // zone0.
+        shardId("1"),  // zone1.
+        shardId("1"),  // zone1.
+        shardId("0"),  // zone2.
+        shardId("0"),  // zone2.
+        shardId("1"),  // zone2.
+        shardId("1"),  // zone2.
+        shardId("1"),  // zone3.
+        shardId("1"),  // zone3.
+        shardId("1")   // Upper bound.
+    };
+    checkGeneratedInitialZoneChunks(tags,
+                                    expectedChunkRanges,
+                                    expectedShardForEachChunk,
+                                    shardKeyPattern,
+                                    7 /* numInitialChunks*/);
+}
+
+TEST_F(PresplitHashedZonesChunksTest, OneLargeZoneAndOtherSmallZonesSharingASingleShard) {
+    const std::vector<ShardType> kShards{
+        ShardType(shardId("0").toString(),
+                  "rs0/shard0:123",
+                  {zoneName("0"), zoneName("1"), zoneName("2"), zoneName("4")}),
+        ShardType(shardId("1").toString(), "rs1/shard1:123"),
+        ShardType(shardId("2").toString(), "rs1/shard1:123", {zoneName("3")}),
+        ShardType(shardId("3").toString(), "rs1/shard1:123", {zoneName("3")}),
+        ShardType(shardId("4").toString(), "rs1/shard1:123", {zoneName("3")}),
+        ShardType(shardId("5").toString(), "rs1/shard1:123", {zoneName("3")}),
+        ShardType(shardId("6").toString(), "rs1/shard1:123", {zoneName("3")}),
+    };
+    setupShards(kShards);
+    shardRegistry()->reload(operationContext());
+
+    auto shardKeyPattern = ShardKeyPattern(BSON("country" << 1 << "city" << 1 << "hashedField"
+                                                          << "hashed"
+                                                          << "suffix" << 1));
+
+    // With gaps after each zone ranges.
+    const std::vector<TagsType> tags = {
+        makeTag(
+            ChunkRange(BSON("country"
+                            << "country1"
+                            << "city"
+                            << "city1"
+                            << "hashedField" << MINKEY << "suffix" << MINKEY),
+                       BSON("country"
+                            << "country2"
+                            << "city" << MINKEY << "hashedField" << MINKEY << "suffix" << MINKEY)),
+            zoneName("0")),
+        makeTag(
+            ChunkRange(BSON("country"
+                            << "country2"
+                            << "city"
+                            << "city2"
+                            << "hashedField" << MINKEY << "suffix" << MINKEY),
+                       BSON("country"
+                            << "country3"
+                            << "city" << MINKEY << "hashedField" << MINKEY << "suffix" << MINKEY)),
+            zoneName("1")),
+        makeTag(
+            ChunkRange(BSON("country"
+                            << "country3"
+                            << "city"
+                            << "city3"
+                            << "hashedField" << MINKEY << "suffix" << MINKEY),
+                       BSON("country"
+                            << "country4"
+                            << "city" << MINKEY << "hashedField" << MINKEY << "suffix" << MINKEY)),
+            zoneName("2")),
+        makeTag(
+            ChunkRange(BSON("country"
+                            << "country4"
+                            << "city"
+                            << "city4"
+                            << "hashedField" << MINKEY << "suffix" << MINKEY),
+                       BSON("country"
+                            << "country5"
+                            << "city" << MAXKEY << "hashedField" << MINKEY << "suffix" << MINKEY)),
+            zoneName("3")),
+        makeTag(
+            ChunkRange(BSON("country"
+                            << "country6"  // Skip country 5.
+                            << "city"
+                            << "city5"
+                            << "hashedField" << MINKEY << "suffix" << MINKEY),
+                       BSON("country"
+                            << "country7"
+                            << "city" << MAXKEY << "hashedField" << MINKEY << "suffix" << MINKEY)),
+            zoneName("4"))};
+
+    // numInitialChunks = 0.
+    // This should have 20 chunks, 14 for all zones, 4 gaps and 2 boundaries.
+    std::vector<ChunkRange> expectedChunkRanges =
+        buildExpectedChunkRanges(tags, shardKeyPattern, {1, 1, 1, 2 * 5, 1} /* numChunksPerTag*/);
+
+    std::vector<ShardId> expectedShardForEachChunk = {
+        shardId("0"),  // Lower bound.
+        shardId("0"),  // zone0.
+        shardId("1"),  // hole.
+        shardId("0"),  // zone1.
+        shardId("2"),  // hole.
+        shardId("0"),  // zone2.
+        shardId("3"),  // hole.
+        shardId("2"),  // zone3.
+        shardId("2"),  // zone3.
+        shardId("3"),  // zone3.
+        shardId("3"),  // zone3.
+        shardId("4"),  // zone3.
+        shardId("4"),  // zone3.
+        shardId("5"),  // zone3.
+        shardId("5"),  // zone3.
+        shardId("6"),  // zone3.
+        shardId("6"),  // zone3.
+        shardId("4"),  // hole.
+        shardId("0"),  // zone4.
+        shardId("5")   // Upper bound.
+    };
+    checkGeneratedInitialZoneChunks(tags,
+                                    expectedChunkRanges,
+                                    expectedShardForEachChunk,
+                                    shardKeyPattern,
+                                    0 /* numInitialChunks*/);
+
+    // numInitialChunks = 1.
+    // This should have 15 chunks, 9 for all zones, 4 gap and 2 boundaries.
+    expectedChunkRanges =
+        buildExpectedChunkRanges(tags, shardKeyPattern, {1, 1, 1, 5, 1} /* numChunksPerTag*/);
+
+    expectedShardForEachChunk = {
+        shardId("0"),  // Lower bound.
+        shardId("0"),  // zone0.
+        shardId("1"),  // hole.
+        shardId("0"),  // zone1.
+        shardId("2"),  // hole.
+        shardId("0"),  // zone2.
+        shardId("3"),  // hole.
+        shardId("2"),  // zone3.
+        shardId("3"),  // zone3.
+        shardId("4"),  // zone3.
+        shardId("5"),  // zone3.
+        shardId("6"),  // zone3.
+        shardId("4"),  // hole.
+        shardId("0"),  // zone4.
+        shardId("5")   // Upper bound.
+    };
+    checkGeneratedInitialZoneChunks(tags,
+                                    expectedChunkRanges,
+                                    expectedShardForEachChunk,
+                                    shardKeyPattern,
+                                    1 /* numInitialChunks*/);
+
+    // numInitialChunks = 11.
+    // This should have 10 chunks, 10 for all zones and 2 boundaries.
+    expectedChunkRanges =
+        buildExpectedChunkRanges(tags, shardKeyPattern, {1, 1, 1, 10, 1} /* numChunksPerTag*/);
+
+    expectedShardForEachChunk = {
+        shardId("0"),  // Lower bound.
+        shardId("0"),  // zone0.
+        shardId("1"),  // hole.
+        shardId("0"),  // zone1.
+        shardId("2"),  // hole.
+        shardId("0"),  // zone2.
+        shardId("3"),  // hole.
+        shardId("2"),  // zone3.
+        shardId("2"),  // zone3.
+        shardId("3"),  // zone3.
+        shardId("3"),  // zone3.
+        shardId("4"),  // zone3.
+        shardId("4"),  // zone3.
+        shardId("5"),  // zone3.
+        shardId("5"),  // zone3.
+        shardId("6"),  // zone3.
+        shardId("6"),  // zone3.
+        shardId("4"),  // hole.
+        shardId("0"),  // zone4.
+        shardId("5")   // Upper bound.
+    };
+    checkGeneratedInitialZoneChunks(tags,
+                                    expectedChunkRanges,
+                                    expectedShardForEachChunk,
+                                    shardKeyPattern,
+                                    11 /* numInitialChunks*/);
+}
+
+TEST_F(PresplitHashedZonesChunksTest, InterweavingZones) {
+    const std::vector<ShardType> kShards{
+        ShardType(shardId("0").toString(), "rs0/shard0:123", {zoneName("0"), zoneName("2")}),
+        ShardType(shardId("1").toString(), "rs1/shard1:123"),
+        ShardType(shardId("2").toString(), "rs1/shard1:123", {zoneName("1")}),
+        ShardType(shardId("3").toString(), "rs1/shard1:123", {zoneName("1")}),
+    };
+    setupShards(kShards);
+    shardRegistry()->reload(operationContext());
+
+    auto shardKeyPattern = ShardKeyPattern(BSON("country" << 1 << "city" << 1 << "hashedField"
+                                                          << "hashed"
+                                                          << "suffix" << 1));
+
+    // With gaps after each zone ranges.
+    const std::vector<TagsType> tags = {
+        makeTag(
+            ChunkRange(BSON("country"
+                            << "country1"
+                            << "city"
+                            << "city1"
+                            << "hashedField" << MINKEY << "suffix" << MINKEY),
+                       BSON("country"
+                            << "country2"
+                            << "city" << MINKEY << "hashedField" << MINKEY << "suffix" << MINKEY)),
+            zoneName("0")),
+        makeTag(
+            ChunkRange(BSON("country"
+                            << "country2"
+                            << "city"
+                            << "city2"
+                            << "hashedField" << MINKEY << "suffix" << MINKEY),
+                       BSON("country"
+                            << "country3"
+                            << "city" << MINKEY << "hashedField" << MINKEY << "suffix" << MINKEY)),
+            zoneName("1")),
+        makeTag(
+            ChunkRange(BSON("country"
+                            << "country3"
+                            << "city"
+                            << "city3"
+                            << "hashedField" << MINKEY << "suffix" << MINKEY),
+                       BSON("country"
+                            << "country4"
+                            << "city" << MINKEY << "hashedField" << MINKEY << "suffix" << MINKEY)),
+            zoneName("2")),
+        makeTag(
+            ChunkRange(BSON("country"
+                            << "country4"
+                            << "city"
+                            << "city4"
+                            << "hashedField" << MINKEY << "suffix" << MINKEY),
+                       BSON("country"
+                            << "country5"
+                            << "city" << MAXKEY << "hashedField" << MINKEY << "suffix" << MINKEY)),
+            zoneName("0")),
+        makeTag(
+            ChunkRange(BSON("country"
+                            << "country6"
+                            << "city"
+                            << "city5"
+                            << "hashedField" << MINKEY << "suffix" << MINKEY),
+                       BSON("country"
+                            << "country7"
+                            << "city" << MAXKEY << "hashedField" << MINKEY << "suffix" << MINKEY)),
+            zoneName("1"))};
+
+    // numInitialChunks = 0.
+    // This should have 13 chunks, 7 for all zones, 4 gaps and 2 boundaries.
+    std::vector<ChunkRange> expectedChunkRanges = buildExpectedChunkRanges(
+        tags, shardKeyPattern, {1, 1 * 2, 1, 1, 1 * 2} /* numChunksPerTag*/);
+
+    std::vector<ShardId> expectedShardForEachChunk = {
+        shardId("0"),  // Lower bound.
+        shardId("0"),  // tag0.
+        shardId("1"),  // hole.
+        shardId("2"),  // tag1.
+        shardId("3"),  // tag1.
+        shardId("2"),  // hole.
+        shardId("0"),  // tag2.
+        shardId("3"),  // hole.
+        shardId("0"),  // tag3.
+        shardId("0"),  // hole.
+        shardId("2"),  // tag4.
+        shardId("3"),  // tag4.
+        shardId("1")   // Upper bound.
+    };
+    checkGeneratedInitialZoneChunks(tags,
+                                    expectedChunkRanges,
+                                    expectedShardForEachChunk,
+                                    shardKeyPattern,
+                                    0 /* numInitialChunks*/);
+
+    // numInitialChunks = 1.
+    // This should have 13 chunks, 7 for all zones, 4 gaps and 2 boundaries.
+    expectedChunkRanges =
+        buildExpectedChunkRanges(tags, shardKeyPattern, {1, 2, 1, 1, 2} /* numChunksPerTag*/);
+    expectedShardForEachChunk = {
+        shardId("0"),  // Lower bound.
+        shardId("0"),  // tag0.
+        shardId("1"),  // hole.
+        shardId("2"),  // tag1.
+        shardId("3"),  // tag1.
+        shardId("2"),  // hole.
+        shardId("0"),  // tag2.
+        shardId("3"),  // hole.
+        shardId("0"),  // tag3.
+        shardId("0"),  // hole.
+        shardId("2"),  // tag4.
+        shardId("3"),  // tag4.
+        shardId("1")   // Upper bound.
+    };
+    checkGeneratedInitialZoneChunks(tags,
+                                    expectedChunkRanges,
+                                    expectedShardForEachChunk,
+                                    shardKeyPattern,
+                                    1 /* numInitialChunks*/);
+
+    // numInitialChunks = 7.
+    // This should have 17 chunks, 11 for all zones and 6 gaps + boundary.
+    expectedChunkRanges = buildExpectedChunkRanges(
+        tags, shardKeyPattern, {1, 2 * 2, 1, 1, 2 * 2} /* numChunksPerTag*/);
+
+    expectedShardForEachChunk = {
+        shardId("0"),  // Lower bound.
+        shardId("0"),  // tag0.
+        shardId("1"),  // hole.
+        shardId("2"),  // tag1.
+        shardId("2"),  // tag1.
+        shardId("3"),  // tag1.
+        shardId("3"),  // tag1.
+        shardId("2"),  // hole.
+        shardId("0"),  // tag2.
+        shardId("3"),  // hole.
+        shardId("0"),  // tag3.
+        shardId("0"),  // hole.
+        shardId("2"),  // tag4.
+        shardId("2"),  // tag4.
+        shardId("3"),  // tag4.
+        shardId("3"),  // tag4.
+        shardId("1")   // Upper bound.
+    };
+    checkGeneratedInitialZoneChunks(tags,
+                                    expectedChunkRanges,
+                                    expectedShardForEachChunk,
+                                    shardKeyPattern,
+                                    7 /* numInitialChunks*/);
+}
+
+TEST_F(PresplitHashedZonesChunksTest, FailsWhenNoZones) {
+    setupShards({ShardType(shardId("0").toString(), "rs0/shard0:123")});
+    shardRegistry()->reload(operationContext());
+
+    auto shardKeyPattern = ShardKeyPattern(BSON("x" << 1 << "y"
+                                                    << "hashed"));
+    ASSERT_THROWS_CODE(
+        PresplitHashedZonesSplitPolicy(operationContext(), shardKeyPattern, {}, 0, true),
+        DBException,
+        31387);
+}
+
+TEST_F(PresplitHashedZonesChunksTest, FailsWhenCollectionNotEmpty) {
+    setupShards({ShardType(shardId("0").toString(), "rs0/shard0:123")});
+    shardRegistry()->reload(operationContext());
+    auto shardKeyPattern = ShardKeyPattern(BSON("x" << 1 << "y"
+                                                    << "hashed"));
+    const auto zoneRange = ChunkRange(BSON("x"
+                                           << "UK"
+                                           << "y" << MINKEY),
+                                      BSON("x"
+                                           << "US"
+                                           << "y" << MINKEY));
+    ASSERT_THROWS_CODE(
+        PresplitHashedZonesSplitPolicy(
+            operationContext(), shardKeyPattern, {makeTag(zoneRange, zoneName("1"))}, 0, false),
+        DBException,
+        31387);
+}
+
+TEST_F(PresplitHashedZonesChunksTest, FailsWhenShardKeyNotHashed) {
+    setupShards({ShardType(shardId("0").toString(), "rs0/shard0:123")});
+    shardRegistry()->reload(operationContext());
+    auto shardKeyPattern = ShardKeyPattern(BSON("x" << 1 << "y" << 1));
+    const auto zoneRange = ChunkRange(BSON("x"
+                                           << "UK"
+                                           << "y" << MINKEY),
+                                      BSON("x"
+                                           << "US"
+                                           << "y" << MINKEY));
+    ASSERT_THROWS_CODE(
+        PresplitHashedZonesSplitPolicy(
+            operationContext(), shardKeyPattern, {makeTag(zoneRange, zoneName("1"))}, 0, true),
+        DBException,
+        31387);
+}
+
+TEST_F(PresplitHashedZonesChunksTest, FailsWhenLowerBoundPrecedingHashedFieldHasMinKeyOrMaxKey) {
+    setupShards({ShardType(shardId("0").toString(), "rs0/shard0:123")});
+    shardRegistry()->reload(operationContext());
+    auto shardKeyPattern = ShardKeyPattern(BSON("a" << 1 << "b" << 1 << "c"
+                                                    << "hashed"
+                                                    << "d" << 1));
+    auto zoneRange = ChunkRange(BSON("a" << 1 << "b" << MINKEY << "c" << MINKEY << "d" << MINKEY),
+                                BSON("a" << 2 << "b" << 1 << "c" << MINKEY << "d" << MINKEY));
+    ASSERT_THROWS_CODE(
+        PresplitHashedZonesSplitPolicy(
+            operationContext(), shardKeyPattern, {makeTag(zoneRange, zoneName("1"))}, 0, true),
+        DBException,
+        31388);
+    zoneRange = ChunkRange(BSON("a" << 1 << "b" << MAXKEY << "c" << MINKEY << "d" << MINKEY),
+                           BSON("a" << 2 << "b" << 1 << "c" << MINKEY << "d" << MINKEY));
     ASSERT_THROWS_CODE(
-        InitialSplitPolicy::generateShardCollectionInitialZonedChunks(
-            nss(), shardKeyPattern(), timeStamp(), tags, tagToShards, makeShardIds(1)),
-        AssertionException,
-        50973);
+        PresplitHashedZonesSplitPolicy(
+            operationContext(), shardKeyPattern, {makeTag(zoneRange, zoneName("1"))}, 0, true),
+        DBException,
+        31388);
+    zoneRange = ChunkRange(BSON("a" << MINKEY << "b" << MINKEY << "c" << MINKEY << "d" << MINKEY),
+                           BSON("a" << 2 << "b" << 2 << "c" << MINKEY << "d" << MINKEY));
+    ASSERT_THROWS_CODE(
+        PresplitHashedZonesSplitPolicy(
+            operationContext(), shardKeyPattern, {makeTag(zoneRange, zoneName("1"))}, 0, true),
+        DBException,
+        31388);
+}
+
+TEST_F(PresplitHashedZonesChunksTest, FailsWhenLowerBoundOfTheHashedFieldIsNotMinKey) {
+    setupShards({ShardType(shardId("0").toString(), "rs0/shard0:123")});
+    shardRegistry()->reload(operationContext());
+    auto shardKeyPattern = ShardKeyPattern(BSON("x" << 1 << "y"
+                                                    << "hashed"));
+    auto zoneRange = ChunkRange(BSON("x" << 1 << "y" << 1), BSON("x" << 2 << "y" << MINKEY));
+    ASSERT_THROWS_CODE(
+        PresplitHashedZonesSplitPolicy(
+            operationContext(), shardKeyPattern, {makeTag(zoneRange, zoneName("1"))}, 0, true),
+        DBException,
+        31389);
+    zoneRange = ChunkRange(BSON("x" << 1 << "y" << MAXKEY), BSON("x" << 2 << "y" << MINKEY));
+    ASSERT_THROWS_CODE(
+        PresplitHashedZonesSplitPolicy(
+            operationContext(), shardKeyPattern, {makeTag(zoneRange, zoneName("1"))}, 0, true),
+        DBException,
+        31389);
+}
+
+TEST_F(PresplitHashedZonesChunksTest, FailsWhenLowerBoundPrecedingHashedFieldIsSameAsUpperBound) {
+    setupShards({ShardType(shardId("0").toString(), "rs0/shard0:123")});
+    shardRegistry()->reload(operationContext());
+    auto shardKeyPattern = ShardKeyPattern(BSON("a" << 1 << "b" << 1 << "c"
+                                                    << "hashed"
+                                                    << "d" << 1));
+    auto zoneRange = ChunkRange(BSON("a" << 2 << "b" << 2 << "c" << MINKEY << "d" << MINKEY),
+                                BSON("a" << 2 << "b" << 2 << "c" << MAXKEY << "d" << MINKEY));
+    ASSERT_THROWS_CODE(
+        PresplitHashedZonesSplitPolicy(
+            operationContext(), shardKeyPattern, {makeTag(zoneRange, zoneName("1"))}, 0, true),
+        DBException,
+        31390);
+    zoneRange = ChunkRange(BSON("a" << 1 << "b" << 1 << "c" << MINKEY << "d" << MINKEY),
+                           BSON("a" << 1 << "b" << 1 << "c" << MINKEY << "d" << MAXKEY));
+    ASSERT_THROWS_CODE(
+        PresplitHashedZonesSplitPolicy(
+            operationContext(), shardKeyPattern, {makeTag(zoneRange, zoneName("1"))}, 0, true),
+        DBException,
+        31390);
+}
+
+TEST_F(PresplitHashedZonesChunksTest, FailsWhenLowerBoundAfterHashedFieldIsNotMinKey) {
+    setupShards({ShardType(shardId("0").toString(), "rs0/shard0:123")});
+    shardRegistry()->reload(operationContext());
+    auto shardKeyPattern = ShardKeyPattern(BSON("a" << 1 << "b"
+                                                    << "hashed"
+                                                    << "c" << 1 << "d" << 1));
+    auto zoneRange = ChunkRange(BSON("a" << 1 << "b" << MINKEY << "c" << 1 << "d" << MINKEY),
+                                BSON("a" << 2 << "b" << 2 << "c" << MAXKEY << "d" << MINKEY));
+    ASSERT_THROWS_CODE(
+        PresplitHashedZonesSplitPolicy(
+            operationContext(), shardKeyPattern, {makeTag(zoneRange, zoneName("1"))}, 0, true),
+        DBException,
+        31391);
+    zoneRange = ChunkRange(BSON("a" << 1 << "b" << MINKEY << "c" << MINKEY << "d" << 1),
+                           BSON("a" << 2 << "b" << 2 << "c" << MAXKEY << "d" << MINKEY));
+    ASSERT_THROWS_CODE(
+        PresplitHashedZonesSplitPolicy(
+            operationContext(), shardKeyPattern, {makeTag(zoneRange, zoneName("1"))}, 0, true),
+        DBException,
+        31391);
+
+    zoneRange = ChunkRange(BSON("a" << 1 << "b" << MINKEY << "c" << MINKEY << "d" << MAXKEY),
+                           BSON("a" << 2 << "b" << 2 << "c" << MAXKEY << "d" << MINKEY));
+    ASSERT_THROWS_CODE(
+        PresplitHashedZonesSplitPolicy(
+            operationContext(), shardKeyPattern, {makeTag(zoneRange, zoneName("1"))}, 0, true),
+        DBException,
+        31391);
+}
+
+TEST_F(PresplitHashedZonesChunksTest, RestrictionsDoNotApplyToUpperBound) {
+    setupShards({ShardType(shardId("0").toString(), "rs0/shard0:123", {zoneName("1")})});
+    shardRegistry()->reload(operationContext());
+    auto shardKeyPattern = ShardKeyPattern(BSON("a" << 1 << "b"
+                                                    << "hashed"
+                                                    << "c" << 1 << "d" << 1));
+    auto zoneRange =
+        ChunkRange(BSON("a" << 1 << "b" << MINKEY << "c" << MINKEY << "d" << MINKEY),
+                   BSON("a" << MAXKEY << "b" << MAXKEY << "c" << MAXKEY << "d" << MAXKEY));
+    PresplitHashedZonesSplitPolicy(
+        operationContext(), shardKeyPattern, {makeTag(zoneRange, zoneName("1"))}, 0, true);
 }
 
 }  // namespace
diff --git a/src/mongo/db/s/config/sharding_catalog_manager_shard_collection_test.cpp b/src/mongo/db/s/config/sharding_catalog_manager_shard_collection_test.cpp
index 17ac9f5e03d..df5ce39bbeb 100644
--- a/src/mongo/db/s/config/sharding_catalog_manager_shard_collection_test.cpp
+++ b/src/mongo/db/s/config/sharding_catalog_manager_shard_collection_test.cpp
@@ -108,33 +108,31 @@ protected:
 };
 
 TEST_F(CreateFirstChunksTest, Split_Disallowed_With_Both_SplitPoints_And_Zones) {
+    ShardsvrShardCollection request;
+    std::vector<BSONObj> splitPoints = {BSONObj()};
+    request.setInitialSplitPoints(splitPoints);
+    std::vector<TagsType> tags = {
+        TagsType(kNamespace,
+                 "TestZone",
+                 ChunkRange(kShardKeyPattern.getKeyPattern().globalMin(), BSON("x" << 0)))};
+
     ASSERT_THROWS_CODE(
-        InitialSplitPolicy::createFirstChunksOptimized(
-            operationContext(),
-            kNamespace,
-            kShardKeyPattern,
-            ShardId("shard1"),
-            {BSON("x" << 0)},
-            {TagsType(kNamespace,
-                      "TestZone",
-                      ChunkRange(kShardKeyPattern.getKeyPattern().globalMin(), BSON("x" << 0)))},
-            InitialSplitPolicy::ShardingOptimizationType::SplitPointsProvided,
-            true /* isEmpty */),
+        InitialSplitPolicy::calculateOptimizationStrategy(operationContext(),
+                                                          kShardKeyPattern,
+                                                          request,
+                                                          tags,
+                                                          2 /* numShards */,
+                                                          true /* collectionIsEmpty */),
         AssertionException,
         ErrorCodes::InvalidOptions);
 
     ASSERT_THROWS_CODE(
-        InitialSplitPolicy::createFirstChunksOptimized(
-            operationContext(),
-            kNamespace,
-            kShardKeyPattern,
-            ShardId("shard1"),
-            {BSON("x" << 0)}, /* No split points */
-            {TagsType(kNamespace,
-                      "TestZone",
-                      ChunkRange(kShardKeyPattern.getKeyPattern().globalMin(), BSON("x" << 0)))},
-            InitialSplitPolicy::ShardingOptimizationType::TagsProvidedWithEmptyCollection,
-            false /* isEmpty */),
+        InitialSplitPolicy::calculateOptimizationStrategy(operationContext(),
+                                                          kShardKeyPattern,
+                                                          request,
+                                                          tags,
+                                                          2 /* numShards */,
+                                                          false /* collectionIsEmpty */),
         AssertionException,
         ErrorCodes::InvalidOptions);
 }
@@ -159,14 +157,17 @@ TEST_F(CreateFirstChunksTest, NonEmptyCollection_SplitPoints_FromSplitVector_Man
         ThreadClient tc("Test", getServiceContext());
         auto opCtx = cc().makeOperationContext();
 
+        ShardsvrShardCollection request;
         auto optimization =
-            InitialSplitPolicy::calculateOptimizationType({}, /* splitPoints */
-                                                          {}, /* tags */
-                                                          false /* collectionIsEmpty */);
-
-        ASSERT_EQ(optimization, InitialSplitPolicy::ShardingOptimizationType::None);
-        return InitialSplitPolicy::createFirstChunksUnoptimized(
-            opCtx.get(), kNamespace, kShardKeyPattern, ShardId("shard1"));
+            InitialSplitPolicy::calculateOptimizationStrategy(operationContext(),
+                                                              kShardKeyPattern,
+                                                              request,
+                                                              {}, /* tags */
+                                                              3 /* numShards */,
+                                                              false /* collectionIsEmpty */);
+        ASSERT(!optimization->isOptimized());
+        return optimization->createFirstChunks(
+            opCtx.get(), kShardKeyPattern, {kNamespace, ShardId("shard1")});
     });
 
     expectSplitVector(connStr.getServers()[0], kShardKeyPattern, BSON_ARRAY(BSON("x" << 0)));
@@ -201,18 +202,19 @@ TEST_F(CreateFirstChunksTest, NonEmptyCollection_SplitPoints_FromClient_ManyChun
         std::vector<TagsType> zones{};
         bool collectionIsEmpty = false;
 
-        auto optimization =
-            InitialSplitPolicy::calculateOptimizationType(splitPoints, zones, collectionIsEmpty);
-        ASSERT_EQ(optimization, InitialSplitPolicy::ShardingOptimizationType::SplitPointsProvided);
+        ShardsvrShardCollection request;
+        request.setInitialSplitPoints(splitPoints);
+        auto optimization = InitialSplitPolicy::calculateOptimizationStrategy(operationContext(),
+                                                                              kShardKeyPattern,
+                                                                              request,
+                                                                              zones,
+                                                                              3 /* numShards */,
+                                                                              collectionIsEmpty);
 
-        return InitialSplitPolicy::createFirstChunksOptimized(opCtx.get(),
-                                                              kNamespace,
-                                                              kShardKeyPattern,
-                                                              ShardId("shard1"),
-                                                              splitPoints,
-                                                              zones,
-                                                              optimization,
-                                                              collectionIsEmpty);
+
+        ASSERT(optimization->isOptimized());
+        return optimization->createFirstChunks(
+            opCtx.get(), kShardKeyPattern, {kNamespace, ShardId("shard1")});
     });
 
     const auto& firstChunks = future.default_timed_get();
@@ -228,26 +230,21 @@ TEST_F(CreateFirstChunksTest, NonEmptyCollection_WithZones_OneChunkToPrimary) {
     setupShards(kShards);
     shardRegistry()->reload(operationContext());
 
-    std::vector<BSONObj> splitPoints{};
     std::vector<TagsType> zones{
         TagsType(kNamespace,
                  "TestZone",
                  ChunkRange(kShardKeyPattern.getKeyPattern().globalMin(), BSON("x" << 0)))};
     bool collectionIsEmpty = false;
 
-    auto optimization =
-        InitialSplitPolicy::calculateOptimizationType(splitPoints, zones, collectionIsEmpty);
-    ASSERT_EQ(optimization,
-              InitialSplitPolicy::ShardingOptimizationType::TagsProvidedWithNonEmptyCollection);
+    ShardsvrShardCollection request;
+    auto optimization = InitialSplitPolicy::calculateOptimizationStrategy(
+        operationContext(), kShardKeyPattern, request, zones, 3 /* numShards */, collectionIsEmpty);
+
 
-    const auto firstChunks = InitialSplitPolicy::createFirstChunksOptimized(operationContext(),
-                                                                            kNamespace,
-                                                                            kShardKeyPattern,
-                                                                            ShardId("shard1"),
-                                                                            splitPoints,
-                                                                            zones,
-                                                                            optimization,
-                                                                            collectionIsEmpty);
+    ASSERT(optimization->isOptimized());
+
+    const auto firstChunks = optimization->createFirstChunks(
+        operationContext(), kShardKeyPattern, {kNamespace, ShardId("shard1")});
 
     ASSERT_EQ(1U, firstChunks.chunks.size());
     ASSERT_EQ(kShards[1].getName(), firstChunks.chunks[0].getShard());
@@ -277,18 +274,20 @@ TEST_F(CreateFirstChunksTest, EmptyCollection_SplitPoints_FromClient_ManyChunksD
         std::vector<TagsType> zones{};
         bool collectionIsEmpty = true;
 
-        auto optimization =
-            InitialSplitPolicy::calculateOptimizationType(splitPoints, zones, collectionIsEmpty);
-        ASSERT_EQ(optimization, InitialSplitPolicy::ShardingOptimizationType::SplitPointsProvided);
+        ShardsvrShardCollection request;
+        request.setInitialSplitPoints(splitPoints);
+        auto optimization = InitialSplitPolicy::calculateOptimizationStrategy(operationContext(),
+                                                                              kShardKeyPattern,
+                                                                              request,
+                                                                              zones,
+                                                                              3 /* numShards */,
+                                                                              collectionIsEmpty);
 
-        return InitialSplitPolicy::createFirstChunksOptimized(opCtx.get(),
-                                                              kNamespace,
-                                                              kShardKeyPattern,
-                                                              ShardId("shard1"),
-                                                              splitPoints,
-                                                              zones,
-                                                              optimization,
-                                                              collectionIsEmpty);
+
+        ASSERT(optimization->isOptimized());
+
+        return optimization->createFirstChunks(
+            operationContext(), kShardKeyPattern, {kNamespace, ShardId("shard1")});
     });
 
     const auto& firstChunks = future.default_timed_get();
@@ -322,18 +321,20 @@ TEST_F(CreateFirstChunksTest, EmptyCollection_NoSplitPoints_OneChunkToPrimary) {
         std::vector<TagsType> zones{};
         bool collectionIsEmpty = true;
 
-        auto optimization =
-            InitialSplitPolicy::calculateOptimizationType(splitPoints, zones, collectionIsEmpty);
-        ASSERT_EQ(optimization, InitialSplitPolicy::ShardingOptimizationType::EmptyCollection);
+        ShardsvrShardCollection request;
+        request.setInitialSplitPoints(splitPoints);
+        auto optimization = InitialSplitPolicy::calculateOptimizationStrategy(operationContext(),
+                                                                              kShardKeyPattern,
+                                                                              request,
+                                                                              zones,
+                                                                              3 /* numShards */,
+                                                                              collectionIsEmpty);
 
-        return InitialSplitPolicy::createFirstChunksOptimized(opCtx.get(),
-                                                              kNamespace,
-                                                              kShardKeyPattern,
-                                                              ShardId("shard1"),
-                                                              splitPoints,
-                                                              zones,
-                                                              optimization,
-                                                              collectionIsEmpty);
+
+        ASSERT(optimization->isOptimized());
+
+        return optimization->createFirstChunks(
+            operationContext(), kShardKeyPattern, {kNamespace, ShardId("shard1")});
     });
 
     const auto& firstChunks = future.default_timed_get();
@@ -354,20 +355,15 @@ TEST_F(CreateFirstChunksTest, EmptyCollection_WithZones_ManyChunksOnFirstZoneSha
                  "TestZone",
                  ChunkRange(kShardKeyPattern.getKeyPattern().globalMin(), BSON("x" << 0)))};
     bool collectionIsEmpty = true;
+    ShardsvrShardCollection request;
+    auto optimization = InitialSplitPolicy::calculateOptimizationStrategy(
+        operationContext(), kShardKeyPattern, request, zones, 3 /* numShards */, collectionIsEmpty);
+
+
+    ASSERT(optimization->isOptimized());
 
-    auto optimization =
-        InitialSplitPolicy::calculateOptimizationType(splitPoints, zones, collectionIsEmpty);
-    ASSERT_EQ(optimization,
-              InitialSplitPolicy::ShardingOptimizationType::TagsProvidedWithEmptyCollection);
-
-    const auto firstChunks = InitialSplitPolicy::createFirstChunksOptimized(operationContext(),
-                                                                            kNamespace,
-                                                                            kShardKeyPattern,
-                                                                            ShardId("shard1"),
-                                                                            splitPoints,
-                                                                            zones,
-                                                                            optimization,
-                                                                            collectionIsEmpty);
+    const auto firstChunks = optimization->createFirstChunks(
+        operationContext(), kShardKeyPattern, {kNamespace, ShardId("shard1")});
 
     ASSERT_EQ(2U, firstChunks.chunks.size());
     ASSERT_EQ(kShards[0].getName(), firstChunks.chunks[0].getShard());
diff --git a/src/mongo/db/s/shardsvr_shard_collection.cpp b/src/mongo/db/s/shardsvr_shard_collection.cpp
index bf03a452569..ddcb0de6702 100644
--- a/src/mongo/db/s/shardsvr_shard_collection.cpp
+++ b/src/mongo/db/s/shardsvr_shard_collection.cpp
@@ -82,8 +82,6 @@ struct ShardCollectionTargetState {
     ShardKeyPattern shardKeyPattern;
     std::vector<TagsType> tags;
     bool collectionIsEmpty;
-    std::vector<BSONObj> splitPoints;
-    int numContiguousChunksPerShard;
 };
 
 const ReadPreferenceSetting kConfigReadSelector(ReadPreference::Nearest, TagSet{});
@@ -445,34 +443,6 @@ int getNumShards(OperationContext* opCtx) {
     return shardIds.size();
 }
 
-struct SplitPoints {
-    std::vector<BSONObj> initialPoints;
-    std::vector<BSONObj> finalPoints;
-};
-
-SplitPoints calculateInitialAndFinalSplitPoints(const ShardsvrShardCollection& request,
-                                                const ShardKeyPattern& shardKeyPattern,
-                                                std::vector<TagsType>& tags,
-                                                int numShards,
-                                                bool collectionIsEmpty) {
-    std::vector<BSONObj> initialSplitPoints;
-    std::vector<BSONObj> finalSplitPoints;
-
-    if (request.getInitialSplitPoints()) {
-        finalSplitPoints = *request.getInitialSplitPoints();
-    } else if (tags.empty()) {
-        InitialSplitPolicy::calculateHashedSplitPointsForEmptyCollection(
-            shardKeyPattern,
-            collectionIsEmpty,
-            numShards,
-            request.getNumInitialChunks(),
-            &initialSplitPoints,
-            &finalSplitPoints);
-    }
-
-    return {std::move(initialSplitPoints), std::move(finalSplitPoints)};
-}
-
 ShardCollectionTargetState calculateTargetState(OperationContext* opCtx,
                                                 const NamespaceString& nss,
                                                 const ShardsvrShardCollection& request) {
@@ -488,25 +458,7 @@ ShardCollectionTargetState calculateTargetState(OperationContext* opCtx,
     auto uuid = getOrGenerateUUID(opCtx, nss, request);
 
     const bool isEmpty = checkIfCollectionIsEmpty(opCtx, nss);
-
-    int numShards = getNumShards(opCtx);
-
-    auto splitPoints =
-        calculateInitialAndFinalSplitPoints(request, shardKeyPattern, tags, numShards, isEmpty);
-
-    auto initialSplitPoints = splitPoints.initialPoints;
-    auto finalSplitPoints = splitPoints.finalPoints;
-
-    const int numContiguousChunksPerShard = initialSplitPoints.empty()
-        ? 1
-        : (finalSplitPoints.size() + 1) / (initialSplitPoints.size() + 1);
-
-    return {uuid,
-            std::move(shardKeyPattern),
-            tags,
-            isEmpty,
-            finalSplitPoints,
-            numContiguousChunksPerShard};
+    return {uuid, std::move(shardKeyPattern), tags, isEmpty};
 }
 
 void logStartShardCollection(OperationContext* opCtx,
@@ -531,8 +483,6 @@ void logStartShardCollection(OperationContext* opCtx,
         prerequisites.uuid.appendToBuilder(&collectionDetail, "uuid");
         collectionDetail.append("empty", prerequisites.collectionIsEmpty);
         collectionDetail.append("primary", primaryShard->toString());
-        collectionDetail.append("numChunks",
-                                static_cast<int>(prerequisites.splitPoints.size() + 1));
         uassertStatusOK(ShardingLogging::get(opCtx)->logChangeChecked(
             opCtx,
             "shardCollection.start",
@@ -689,7 +639,7 @@ UUID shardCollection(OperationContext* opCtx,
         return *collectionOptional->getUUID();
     }
 
-    InitialSplitPolicy::ShardingOptimizationType optimizationType;
+    std::unique_ptr<InitialSplitPolicy> splitPolicy;
     InitialSplitPolicy::ShardCollectionConfig initialChunks;
     boost::optional<ShardCollectionTargetState> targetState;
 
@@ -733,23 +683,18 @@ UUID shardCollection(OperationContext* opCtx,
         critSec.enterCommitPhase();
 
         pauseShardCollectionCommitPhase.pauseWhileSet();
-
         logStartShardCollection(opCtx, cmdObj, nss, request, *targetState, dbPrimaryShardId);
 
-        optimizationType = InitialSplitPolicy::calculateOptimizationType(
-            targetState->splitPoints, targetState->tags, targetState->collectionIsEmpty);
-        if (optimizationType != InitialSplitPolicy::ShardingOptimizationType::None) {
-            initialChunks = InitialSplitPolicy::createFirstChunksOptimized(
-                opCtx,
-                nss,
-                targetState->shardKeyPattern,
-                dbPrimaryShardId,
-                targetState->splitPoints,
-                targetState->tags,
-                optimizationType,
-                targetState->collectionIsEmpty,
-                targetState->numContiguousChunksPerShard);
-
+        splitPolicy =
+            InitialSplitPolicy::calculateOptimizationStrategy(opCtx,
+                                                              targetState->shardKeyPattern,
+                                                              request,
+                                                              targetState->tags,
+                                                              getNumShards(opCtx),
+                                                              targetState->collectionIsEmpty);
+        if (splitPolicy->isOptimized()) {
+            initialChunks = splitPolicy->createFirstChunks(
+                opCtx, targetState->shardKeyPattern, {nss, dbPrimaryShardId});
             createCollectionOnShardsReceivingChunks(
                 opCtx, nss, initialChunks.chunks, dbPrimaryShardId);
 
@@ -760,11 +705,11 @@ UUID shardCollection(OperationContext* opCtx,
     // We have now left the critical section.
     pauseShardCollectionAfterCriticalSection.pauseWhileSet();
 
-    if (optimizationType == InitialSplitPolicy::ShardingOptimizationType::None) {
+    if (!splitPolicy->isOptimized()) {
         invariant(initialChunks.chunks.empty());
 
-        initialChunks = InitialSplitPolicy::createFirstChunksUnoptimized(
-            opCtx, nss, targetState->shardKeyPattern, dbPrimaryShardId);
+        initialChunks = splitPolicy->createFirstChunks(
+            opCtx, targetState->shardKeyPattern, {nss, dbPrimaryShardId});
 
         writeChunkDocumentsAndRefreshShards(*targetState, initialChunks);
     }
@@ -777,7 +722,8 @@ UUID shardCollection(OperationContext* opCtx,
         opCtx,
         "shardCollection.end",
         nss.ns(),
-        BSON("version" << initialChunks.collVersion().toString()),
+        BSON("version" << initialChunks.collVersion().toString() << "numChunks"
+                       << static_cast<int>(initialChunks.chunks.size())),
         ShardingCatalogClient::kMajorityWriteConcern);
 
     return targetState->uuid;
diff --git a/src/mongo/s/commands/cluster_shard_collection_cmd.cpp b/src/mongo/s/commands/cluster_shard_collection_cmd.cpp
index 502fdefde1c..a2eddf22614 100644
--- a/src/mongo/s/commands/cluster_shard_collection_cmd.cpp
+++ b/src/mongo/s/commands/cluster_shard_collection_cmd.cpp
@@ -116,6 +116,7 @@ public:
         configShardCollRequest.setKey(shardCollRequest.getKey());
         configShardCollRequest.setUnique(shardCollRequest.getUnique());
         configShardCollRequest.setNumInitialChunks(shardCollRequest.getNumInitialChunks());
+        configShardCollRequest.setPresplitHashedZones(shardCollRequest.getPresplitHashedZones());
         configShardCollRequest.setCollation(shardCollRequest.getCollation());
 
         // Invalidate the routing table cache entry for this collection so that we reload the
diff --git a/src/mongo/s/request_types/shard_collection.idl b/src/mongo/s/request_types/shard_collection.idl
index 5a1e9229d38..5df76c033e7 100644
--- a/src/mongo/s/request_types/shard_collection.idl
+++ b/src/mongo/s/request_types/shard_collection.idl
@@ -58,6 +58,10 @@ structs:
                 type: safeInt64
                 description: "The number of chunks to create initially when sharding an empty collection with a hashed shard key."
                 default: 0
+            presplitHashedZones:
+                type: bool
+                description: "True if the chunks should be pre-split based on the existing zones when sharding a collection with hashed shard key"
+                default: false
             collation:
                 type: object
                 description: "The collation to use for the shard key index."
@@ -82,6 +86,10 @@ structs:
                 type: safeInt64
                 description: "The number of chunks to create initially when sharding an empty collection with a hashed shard key."
                 default: 0
+            presplitHashedZones:
+                type: bool
+                description: "True if the chunks should be pre-split based on the existing zones when sharding a collection with hashed shard key"
+                default: false
             initialSplitPoints:
                 type: array<object>
                 description: "A specific set of points to create initial splits at, currently used only by mapReduce"
@@ -126,6 +134,10 @@ structs:
                 type: safeInt64
                 description: "The number of chunks to create initially when sharding an empty collection with a hashed shard key."
                 default: 0
+            presplitHashedZones:
+                type: bool
+                description: "True if the chunks should be pre-split based on the existing zones when sharding a collection with hashed shard key"
+                default: false
             initialSplitPoints:
                 type: array<object>
                 description: "A specific set of points to create initial splits at, currently used only by mapReduce"
@@ -161,4 +173,4 @@ structs:
             collectionUUID:
                 type: uuid
                 description: "The UUID of the collection that just got sharded."
-                optional: true
\ No newline at end of file
+                optional: true