SERVER-27681 Use CatalogCache to load metadata on shards

This change gets rid of the MetadataLoader in place of using the
CatalogCache for loading chunk metadata on shards. This ensures that the
same concurrency control and rate limiting applies on mongos and mongod
and obviates the need for having different kinds of ChunkDiffer.

Also removes the duplicated ShardKeyPattern functionality from
CollectionMetadata in place of using the one from ShardKeyPattern.

(cherry picked from commit 3e53f0ce9cdff926473276a34a351972ea963a57)
(cherry picked from commit f35cb0e4fcc26e5c8950fe75a8ff46aa811f2e56)
(cherry picked from commit d8b0d028e3895b32b86f3e6878d4bc34dd8c7fc8)

23 files changed, 718 insertions, 2032 deletions

diff --git a/jstests/sharding/array_shard_key.js b/jstests/sharding/array_shard_key.js
index 4fd60c3f21d..cdbe4bda885 100644
--- a/jstests/sharding/array_shard_key.js
+++ b/jstests/sharding/array_shard_key.js
@@ -1,111 +1,113 @@
 // Ensure you can't shard on an array key
+(function() {
+    'use strict';
 
-var st = new ShardingTest({name: jsTestName(), shards: 3});
+    var st = new ShardingTest({shards: 3});
 
-var mongos = st.s0;
+    var mongos = st.s0;
 
-var coll = mongos.getCollection(jsTestName() + ".foo");
+    var coll = mongos.getCollection("TestDB.foo");
 
-st.shardColl(coll, {_id: 1, i: 1}, {_id: ObjectId(), i: 1});
-
-printjson(mongos.getDB("config").chunks.find().toArray());
+    st.shardColl(coll, {_id: 1, i: 1}, {_id: ObjectId(), i: 1});
 
-st.printShardingStatus();
+    printjson(mongos.getDB("config").chunks.find().toArray());
 
-print("1: insert some invalid data");
+    print("1: insert some invalid data");
 
-var value = null;
+    var value = null;
 
-// Insert an object with invalid array key
-assert.writeError(coll.insert({i: [1, 2]}));
+    // Insert an object with invalid array key
+    assert.writeError(coll.insert({i: [1, 2]}));
 
-// Insert an object with all the right fields, but an invalid array val for _id
-assert.writeError(coll.insert({_id: [1, 2], i: 3}));
+    // Insert an object with all the right fields, but an invalid array val for _id
+    assert.writeError(coll.insert({_id: [1, 2], i: 3}));
 
-// Insert an object with valid array key
-assert.writeOK(coll.insert({i: 1}));
+    // Insert an object with valid array key
+    assert.writeOK(coll.insert({i: 1}));
 
-// Update the value with valid other field
-value = coll.findOne({i: 1});
-assert.writeOK(coll.update(value, {$set: {j: 2}}));
+    // Update the value with valid other field
+    value = coll.findOne({i: 1});
+    assert.writeOK(coll.update(value, {$set: {j: 2}}));
 
-// Update the value with invalid other fields
-value = coll.findOne({i: 1});
-assert.writeError(coll.update(value, Object.merge(value, {i: [3]})));
+    // Update the value with invalid other fields
+    value = coll.findOne({i: 1});
+    assert.writeError(coll.update(value, Object.merge(value, {i: [3]})));
 
-// Multi-update the value with invalid other fields
-value = coll.findOne({i: 1});
-assert.writeError(coll.update(value, Object.merge(value, {i: [3, 4]}), false, true));
+    // Multi-update the value with invalid other fields
+    value = coll.findOne({i: 1});
+    assert.writeError(coll.update(value, Object.merge(value, {i: [3, 4]}), false, true));
 
-// Multi-update the value with other fields (won't work, but no error)
-value = coll.findOne({i: 1});
-assert.writeOK(coll.update(Object.merge(value, {i: [1, 1]}), {$set: {k: 4}}, false, true));
+    // Multi-update the value with other fields (won't work, but no error)
+    value = coll.findOne({i: 1});
+    assert.writeOK(coll.update(Object.merge(value, {i: [1, 1]}), {$set: {k: 4}}, false, true));
 
-// Query the value with other fields (won't work, but no error)
-value = coll.findOne({i: 1});
-coll.find(Object.merge(value, {i: [1, 1]})).toArray();
+    // Query the value with other fields (won't work, but no error)
+    value = coll.findOne({i: 1});
+    coll.find(Object.merge(value, {i: [1, 1]})).toArray();
 
-// Can't remove using multikey, but shouldn't error
-value = coll.findOne({i: 1});
-coll.remove(Object.extend(value, {i: [1, 2, 3, 4]}));
+    // Can't remove using multikey, but shouldn't error
+    value = coll.findOne({i: 1});
+    coll.remove(Object.extend(value, {i: [1, 2, 3, 4]}));
 
-// Can't remove using multikey, but shouldn't error
-value = coll.findOne({i: 1});
-assert.writeOK(coll.remove(Object.extend(value, {i: [1, 2, 3, 4, 5]})));
-assert.eq(coll.find().itcount(), 1);
+    // Can't remove using multikey, but shouldn't error
+    value = coll.findOne({i: 1});
+    assert.writeOK(coll.remove(Object.extend(value, {i: [1, 2, 3, 4, 5]})));
+    assert.eq(coll.find().itcount(), 1);
 
-value = coll.findOne({i: 1});
-assert.writeOK(coll.remove(Object.extend(value, {i: 1})));
-assert.eq(coll.find().itcount(), 0);
+    value = coll.findOne({i: 1});
+    assert.writeOK(coll.remove(Object.extend(value, {i: 1})));
+    assert.eq(coll.find().itcount(), 0);
 
-coll.ensureIndex({_id: 1, i: 1, j: 1});
-// Can insert document that will make index into a multi-key as long as it's not part of shard key.
-coll.remove({});
-assert.writeOK(coll.insert({i: 1, j: [1, 2]}));
-assert.eq(coll.find().itcount(), 1);
+    coll.ensureIndex({_id: 1, i: 1, j: 1});
+    // Can insert document that will make index into a multi-key as long as it's not part of shard
+    // key.
+    coll.remove({});
+    assert.writeOK(coll.insert({i: 1, j: [1, 2]}));
+    assert.eq(coll.find().itcount(), 1);
 
-// Same is true for updates.
-coll.remove({});
-coll.insert({_id: 1, i: 1});
-assert.writeOK(coll.update({_id: 1, i: 1}, {_id: 1, i: 1, j: [1, 2]}));
-assert.eq(coll.find().itcount(), 1);
+    // Same is true for updates.
+    coll.remove({});
+    coll.insert({_id: 1, i: 1});
+    assert.writeOK(coll.update({_id: 1, i: 1}, {_id: 1, i: 1, j: [1, 2]}));
+    assert.eq(coll.find().itcount(), 1);
 
-// Same for upserts.
-coll.remove({});
-assert.writeOK(coll.update({_id: 1, i: 1}, {_id: 1, i: 1, j: [1, 2]}, true));
-assert.eq(coll.find().itcount(), 1);
+    // Same for upserts.
+    coll.remove({});
+    assert.writeOK(coll.update({_id: 1, i: 1}, {_id: 1, i: 1, j: [1, 2]}, true));
+    assert.eq(coll.find().itcount(), 1);
 
-printjson("Sharding-then-inserting-multikey tested, now trying inserting-then-sharding-multikey");
+    printjson(
+        "Sharding-then-inserting-multikey tested, now trying inserting-then-sharding-multikey");
 
-// Insert a bunch of data then shard over key which is an array
-var coll = mongos.getCollection("" + coll + "2");
-for (var i = 0; i < 10; i++) {
-    // TODO : does not check weird cases like [ i, i ]
-    assert.writeOK(coll.insert({i: [i, i + 1]}));
-}
+    // Insert a bunch of data then shard over key which is an array
+    var coll = mongos.getCollection("" + coll + "2");
+    for (var i = 0; i < 10; i++) {
+        // TODO : does not check weird cases like [ i, i ]
+        assert.writeOK(coll.insert({i: [i, i + 1]}));
+    }
 
-coll.ensureIndex({_id: 1, i: 1});
+    coll.ensureIndex({_id: 1, i: 1});
 
-try {
-    st.shardColl(coll, {_id: 1, i: 1}, {_id: ObjectId(), i: 1});
-} catch (e) {
-    print("Correctly threw error on sharding with multikey index.");
-}
+    try {
+        st.shardColl(coll, {_id: 1, i: 1}, {_id: ObjectId(), i: 1});
+    } catch (e) {
+        print("Correctly threw error on sharding with multikey index.");
+    }
 
-st.printShardingStatus();
+    st.printShardingStatus();
 
-// Insert a bunch of data then shard over key which is not an array
-var coll = mongos.getCollection("" + coll + "3");
-for (var i = 0; i < 10; i++) {
-    // TODO : does not check weird cases like [ i, i ]
-    assert.writeOK(coll.insert({i: i}));
-}
+    // Insert a bunch of data then shard over key which is not an array
+    var coll = mongos.getCollection("" + coll + "3");
+    for (var i = 0; i < 10; i++) {
+        // TODO : does not check weird cases like [ i, i ]
+        assert.writeOK(coll.insert({i: i}));
+    }
 
-coll.ensureIndex({_id: 1, i: 1});
+    coll.ensureIndex({_id: 1, i: 1});
 
-st.shardColl(coll, {_id: 1, i: 1}, {_id: ObjectId(), i: 1});
+    st.shardColl(coll, {_id: 1, i: 1}, {_id: ObjectId(), i: 1});
 
-st.printShardingStatus();
+    st.printShardingStatus();
 
-// Finish
-st.stop();
+    st.stop();
+})();
diff --git a/src/mongo/db/field_ref.h b/src/mongo/db/field_ref.h
index 82116faf6fe..b6b218630b8 100644
--- a/src/mongo/db/field_ref.h
+++ b/src/mongo/db/field_ref.h
@@ -138,11 +138,6 @@ private:
     // with allocations.
     static const size_t kReserveAhead = 4;
 
-    /**
-     * Parses 'path' into parts.
-     */
-    void _parse(StringData path);
-
     /** Converts the field part index to the variable part equivalent */
     size_t getIndex(size_t i) const {
         return i - kReserveAhead;
diff --git a/src/mongo/db/s/SConscript b/src/mongo/db/s/SConscript
index dcb179b846c..b2abb86c056 100644
--- a/src/mongo/db/s/SConscript
+++ b/src/mongo/db/s/SConscript
@@ -6,7 +6,6 @@ env.Library(
     target='metadata',
     source=[
         'collection_metadata.cpp',
-        'metadata_loader.cpp',
     ],
     LIBDEPS=[
         '$BUILD_DIR/mongo/base',
@@ -187,7 +186,6 @@ env.CppUnitTest(
 env.CppUnitTest(
     target='sharding_metadata_test',
     source=[
-        'metadata_loader_test.cpp',
         'collection_metadata_test.cpp',
     ],
     LIBDEPS=[
diff --git a/src/mongo/db/s/cleanup_orphaned_cmd.cpp b/src/mongo/db/s/cleanup_orphaned_cmd.cpp
index a9b361027bb..5775ff52e08 100644
--- a/src/mongo/db/s/cleanup_orphaned_cmd.cpp
+++ b/src/mongo/db/s/cleanup_orphaned_cmd.cpp
@@ -86,7 +86,7 @@ CleanupResult cleanupOrphanedData(OperationContext* txn,
         metadata = CollectionShardingState::get(txn, ns.toString())->getMetadata();
     }
 
-    if (!metadata || metadata->getKeyPattern().isEmpty()) {
+    if (!metadata) {
         warning() << "skipping orphaned data cleanup for " << ns.toString()
                   << ", collection is not sharded";
 
@@ -214,15 +214,10 @@ public:
             return false;
         }
 
-        if (ns == "") {
-            errmsg = "no collection name specified";
-            return false;
-        }
-
-        if (!NamespaceString(ns).isValid()) {
-            errmsg = "invalid namespace";
-            return false;
-        }
+        const NamespaceString nss(ns);
+        uassert(ErrorCodes::InvalidNamespace,
+                str::stream() << "Invalid namespace: " << nss.ns(),
+                nss.isValid());
 
         BSONObj startingFromKey;
         if (!FieldParser::extract(cmdObj, startingFromKeyField, &startingFromKey, &errmsg)) {
@@ -242,21 +237,12 @@ public:
             return false;
         }
 
-        ChunkVersion shardVersion;
-        Status status = shardingState->refreshMetadataNow(txn, NamespaceString(ns), &shardVersion);
-        if (!status.isOK()) {
-            if (status.code() == ErrorCodes::RemoteChangeDetected) {
-                warning() << "Shard version in transition detected while refreshing "
-                          << "metadata for " << ns << " at version " << shardVersion;
-            } else {
-                errmsg = str::stream() << "failed to refresh shard metadata: " << redact(status);
-                return false;
-            }
-        }
+        ChunkVersion unusedShardVersion;
+        uassertStatusOK(shardingState->refreshMetadataNow(txn, nss, &unusedShardVersion));
 
         BSONObj stoppedAtKey;
-        CleanupResult cleanupResult = cleanupOrphanedData(
-            txn, NamespaceString(ns), startingFromKey, writeConcern, &stoppedAtKey, &errmsg);
+        CleanupResult cleanupResult =
+            cleanupOrphanedData(txn, nss, startingFromKey, writeConcern, &stoppedAtKey, &errmsg);
 
         if (cleanupResult == CleanupResult_Error) {
             return false;
diff --git a/src/mongo/db/s/collection_metadata.cpp b/src/mongo/db/s/collection_metadata.cpp
index b2c19420643..b18e325adb0 100644
--- a/src/mongo/db/s/collection_metadata.cpp
+++ b/src/mongo/db/s/collection_metadata.cpp
@@ -41,66 +41,73 @@
 
 namespace mongo {
 
-using std::unique_ptr;
-using std::make_pair;
-using std::string;
-using std::vector;
-using str::stream;
-
-CollectionMetadata::CollectionMetadata()
-    : _pendingMap(SimpleBSONObjComparator::kInstance.makeBSONObjIndexedMap<CachedChunkInfo>()),
-      _chunksMap(SimpleBSONObjComparator::kInstance.makeBSONObjIndexedMap<CachedChunkInfo>()),
-      _rangesMap(SimpleBSONObjComparator::kInstance.makeBSONObjIndexedMap<CachedChunkInfo>()) {}
-
-CollectionMetadata::CollectionMetadata(const BSONObj& keyPattern, ChunkVersion collectionVersion)
-    : _collVersion(collectionVersion),
-      _shardVersion(ChunkVersion(0, 0, collectionVersion.epoch())),
-      _keyPattern(keyPattern.getOwned()),
+CollectionMetadata::CollectionMetadata(const BSONObj& keyPattern,
+                                       ChunkVersion collectionVersion,
+                                       ChunkVersion shardVersion,
+                                       RangeMap shardChunksMap)
+    : _shardKeyPattern(keyPattern),
+      _collVersion(collectionVersion),
+      _shardVersion(shardVersion),
+      _chunksMap(std::move(shardChunksMap)),
       _pendingMap(SimpleBSONObjComparator::kInstance.makeBSONObjIndexedMap<CachedChunkInfo>()),
-      _chunksMap(SimpleBSONObjComparator::kInstance.makeBSONObjIndexedMap<CachedChunkInfo>()),
-      _rangesMap(SimpleBSONObjComparator::kInstance.makeBSONObjIndexedMap<CachedChunkInfo>()) {}
+      _rangesMap(SimpleBSONObjComparator::kInstance.makeBSONObjIndexedMap<CachedChunkInfo>()) {
 
-CollectionMetadata::~CollectionMetadata() = default;
+    invariant(_shardKeyPattern.isValid());
+    invariant(_collVersion.epoch() == _shardVersion.epoch());
+    invariant(_collVersion.isSet());
+    invariant(_collVersion >= _shardVersion);
 
-unique_ptr<CollectionMetadata> CollectionMetadata::clonePlusChunk(
-    const BSONObj& minKey, const BSONObj& maxKey, const ChunkVersion& chunkVersion) const {
-    invariant(chunkVersion.epoch() == _shardVersion.epoch());
-    invariant(chunkVersion.isSet());
-    invariant(chunkVersion > _shardVersion);
-    invariant(minKey.woCompare(maxKey) < 0);
-    invariant(!rangeMapOverlaps(_chunksMap, minKey, maxKey));
-
-    unique_ptr<CollectionMetadata> metadata(stdx::make_unique<CollectionMetadata>());
-    metadata->_keyPattern = _keyPattern.getOwned();
-    metadata->fillKeyPatternFields();
-    metadata->_pendingMap = _pendingMap;
-    metadata->_chunksMap = _chunksMap;
-    metadata->_chunksMap.insert(
-        make_pair(minKey.getOwned(), CachedChunkInfo(maxKey.getOwned(), chunkVersion)));
-    metadata->_shardVersion = chunkVersion;
-    metadata->_collVersion = chunkVersion > _collVersion ? chunkVersion : _collVersion;
-    metadata->fillRanges();
-
-    invariant(metadata->isValid());
-    return metadata;
+    if (_chunksMap.empty()) {
+        invariant(!_shardVersion.isSet());
+        return;
+    }
+
+    invariant(_shardVersion.isSet());
+
+    // Load the chunk information, coallesceing their ranges. The version for this shard would be
+    // the highest version for any of the chunks.
+    BSONObj min, max;
+
+    for (const auto& entry : _chunksMap) {
+        BSONObj currMin = entry.first;
+        BSONObj currMax = entry.second.getMaxKey();
+
+        // Coalesce the chunk's bounds in ranges if they are adjacent chunks
+        if (min.isEmpty()) {
+            min = currMin;
+            max = currMax;
+            continue;
+        }
+
+        if (SimpleBSONObjComparator::kInstance.evaluate(max == currMin)) {
+            max = currMax;
+            continue;
+        }
+
+        _rangesMap.emplace(min, CachedChunkInfo(max, ChunkVersion::IGNORED()));
+
+        min = currMin;
+        max = currMax;
+    }
+
+    invariant(!min.isEmpty());
+    invariant(!max.isEmpty());
+
+    _rangesMap.emplace(min, CachedChunkInfo(max, ChunkVersion::IGNORED()));
 }
 
+CollectionMetadata::~CollectionMetadata() = default;
+
 std::unique_ptr<CollectionMetadata> CollectionMetadata::cloneMinusPending(
     const ChunkType& chunk) const {
     invariant(rangeMapContains(_pendingMap, chunk.getMin(), chunk.getMax()));
 
-    unique_ptr<CollectionMetadata> metadata(stdx::make_unique<CollectionMetadata>());
-    metadata->_keyPattern = _keyPattern.getOwned();
-    metadata->fillKeyPatternFields();
+    auto metadata(stdx::make_unique<CollectionMetadata>(
+        _shardKeyPattern.toBSON(), getCollVersion(), getShardVersion(), getChunks()));
     metadata->_pendingMap = _pendingMap;
-    metadata->_pendingMap.erase(chunk.getMin());
 
-    metadata->_chunksMap = _chunksMap;
-    metadata->_rangesMap = _rangesMap;
-    metadata->_shardVersion = _shardVersion;
-    metadata->_collVersion = _collVersion;
+    metadata->_pendingMap.erase(chunk.getMin());
 
-    invariant(metadata->isValid());
     return metadata;
 }
 
@@ -108,14 +115,9 @@ std::unique_ptr<CollectionMetadata> CollectionMetadata::clonePlusPending(
     const ChunkType& chunk) const {
     invariant(!rangeMapOverlaps(_chunksMap, chunk.getMin(), chunk.getMax()));
 
-    unique_ptr<CollectionMetadata> metadata(stdx::make_unique<CollectionMetadata>());
-    metadata->_keyPattern = _keyPattern.getOwned();
-    metadata->fillKeyPatternFields();
+    auto metadata(stdx::make_unique<CollectionMetadata>(
+        _shardKeyPattern.toBSON(), getCollVersion(), getShardVersion(), getChunks()));
     metadata->_pendingMap = _pendingMap;
-    metadata->_chunksMap = _chunksMap;
-    metadata->_rangesMap = _rangesMap;
-    metadata->_shardVersion = _shardVersion;
-    metadata->_collVersion = _collVersion;
 
     // If there are any pending chunks on the interval to be added this is ok, since pending chunks
     // aren't officially tracked yet and something may have changed on servers we do not see yet.
@@ -138,25 +140,18 @@ std::unique_ptr<CollectionMetadata> CollectionMetadata::clonePlusPending(
 
     // The pending map entry cannot contain a specific chunk version because we don't know what
     // version would be generated for it at commit time. That's why we insert an IGNORED value.
-    metadata->_pendingMap.insert(
-        make_pair(chunk.getMin(), CachedChunkInfo(chunk.getMax(), ChunkVersion::IGNORED())));
+    metadata->_pendingMap.emplace(chunk.getMin(),
+                                  CachedChunkInfo(chunk.getMax(), ChunkVersion::IGNORED()));
 
-    invariant(metadata->isValid());
     return metadata;
 }
 
 bool CollectionMetadata::keyBelongsToMe(const BSONObj& key) const {
-    // For now, collections don't move. So if the collection is not sharded, assume
-    // the document with the given key can be accessed.
-    if (_keyPattern.isEmpty()) {
-        return true;
-    }
-
-    if (_rangesMap.size() <= 0) {
+    if (_rangesMap.empty()) {
         return false;
     }
 
-    RangeMap::const_iterator it = _rangesMap.upper_bound(key);
+    auto it = _rangesMap.upper_bound(key);
     if (it != _rangesMap.begin())
         it--;
 
@@ -164,21 +159,15 @@ bool CollectionMetadata::keyBelongsToMe(const BSONObj& key) const {
 }
 
 bool CollectionMetadata::keyIsPending(const BSONObj& key) const {
-    // If we aren't sharded, then the key is never pending (though it belongs-to-me)
-    if (_keyPattern.isEmpty()) {
+    if (_pendingMap.empty()) {
         return false;
     }
 
-    if (_pendingMap.size() <= 0) {
-        return false;
-    }
-
-    RangeMap::const_iterator it = _pendingMap.upper_bound(key);
+    auto it = _pendingMap.upper_bound(key);
     if (it != _pendingMap.begin())
         it--;
 
-    bool isPending = rangeContains(it->first, it->second.getMaxKey(), key);
-    return isPending;
+    return rangeContains(it->first, it->second.getMaxKey(), key);
 }
 
 bool CollectionMetadata::getNextChunk(const BSONObj& lookupKey, ChunkType* chunk) const {
@@ -252,7 +241,7 @@ Status CollectionMetadata::checkChunkIsValid(const ChunkType& chunk) {
 void CollectionMetadata::toBSONBasic(BSONObjBuilder& bb) const {
     _collVersion.addToBSON(bb, "collVersion");
     _shardVersion.addToBSON(bb, "shardVersion");
-    bb.append("keyPattern", _keyPattern);
+    bb.append("keyPattern", _shardKeyPattern.toBSON());
 }
 
 void CollectionMetadata::toBSONChunks(BSONArrayBuilder& bb) const {
@@ -279,15 +268,12 @@ void CollectionMetadata::toBSONPending(BSONArrayBuilder& bb) const {
     }
 }
 
-string CollectionMetadata::toStringBasic() const {
-    return stream() << "collection version: " << _collVersion.toString()
-                    << ", shard version: " << _shardVersion.toString();
+std::string CollectionMetadata::toStringBasic() const {
+    return str::stream() << "collection version: " << _collVersion.toString()
+                         << ", shard version: " << _shardVersion.toString();
 }
 
 bool CollectionMetadata::getNextOrphanRange(const BSONObj& origLookupKey, KeyRange* range) const {
-    if (_keyPattern.isEmpty())
-        return false;
-
     BSONObj lookupKey = origLookupKey;
     BSONObj maxKey = getMaxKey();  // so we don't keep rebuilding
     while (lookupKey.woCompare(maxKey) < 0) {
@@ -337,7 +323,7 @@ bool CollectionMetadata::getNextOrphanRange(const BSONObj& origLookupKey, KeyRan
         // bounds of the surrounding ranges in both maps.
         //
 
-        range->keyPattern = _keyPattern;
+        range->keyPattern = _shardKeyPattern.toBSON();
         range->minKey = getMinKey();
         range->maxKey = maxKey;
 
@@ -367,98 +353,15 @@ bool CollectionMetadata::getNextOrphanRange(const BSONObj& origLookupKey, KeyRan
 }
 
 BSONObj CollectionMetadata::getMinKey() const {
-    BSONObjIterator it(_keyPattern);
-    BSONObjBuilder minKeyB;
-    while (it.more())
-        minKeyB << it.next().fieldName() << MINKEY;
-    return minKeyB.obj();
+    return _shardKeyPattern.getKeyPattern().globalMin();
 }
 
 BSONObj CollectionMetadata::getMaxKey() const {
-    BSONObjIterator it(_keyPattern);
-    BSONObjBuilder maxKeyB;
-    while (it.more())
-        maxKeyB << it.next().fieldName() << MAXKEY;
-    return maxKeyB.obj();
-}
-
-bool CollectionMetadata::isValid() const {
-    if (_shardVersion > _collVersion)
-        return false;
-    if (_collVersion.majorVersion() == 0)
-        return false;
-    if (_collVersion.epoch() != _shardVersion.epoch())
-        return false;
-
-    if (_shardVersion.majorVersion() > 0) {
-        // Must be chunks
-        if (_rangesMap.size() == 0 || _chunksMap.size() == 0)
-            return false;
-    } else {
-        // No chunks
-        if (_shardVersion.minorVersion() > 0)
-            return false;
-        if (_rangesMap.size() > 0 || _chunksMap.size() > 0)
-            return false;
-    }
-
-    return true;
+    return _shardKeyPattern.getKeyPattern().globalMax();
 }
 
 bool CollectionMetadata::isValidKey(const BSONObj& key) const {
-    BSONObjIterator it(_keyPattern);
-    while (it.more()) {
-        BSONElement next = it.next();
-        if (!key.hasField(next.fieldName()))
-            return false;
-    }
-    return key.nFields() == _keyPattern.nFields();
-}
-
-void CollectionMetadata::fillRanges() {
-    if (_chunksMap.empty())
-        return;
-
-    // Load the chunk information, coallesceing their ranges. The version for this shard would be
-    // the highest version for any of the chunks.
-    BSONObj min, max;
-    for (const auto& entry : _chunksMap) {
-        BSONObj currMin = entry.first;
-        BSONObj currMax = entry.second.getMaxKey();
-
-        // coalesce the chunk's bounds in ranges if they are adjacent chunks
-        if (min.isEmpty()) {
-            min = currMin;
-            max = currMax;
-            continue;
-        }
-        if (SimpleBSONObjComparator::kInstance.evaluate(max == currMin)) {
-            max = currMax;
-            continue;
-        }
-
-        _rangesMap.insert(make_pair(min, CachedChunkInfo(max, ChunkVersion::IGNORED())));
-
-        min = currMin;
-        max = currMax;
-    }
-
-    invariant(!min.isEmpty());
-    invariant(!max.isEmpty());
-
-    _rangesMap.insert(make_pair(min, CachedChunkInfo(max, ChunkVersion::IGNORED())));
-}
-
-void CollectionMetadata::fillKeyPatternFields() {
-    // Parse the shard keys into the states 'keys' and 'keySet' members.
-    BSONObjIterator patternIter = _keyPattern.begin();
-    while (patternIter.more()) {
-        BSONElement current = patternIter.next();
-
-        _keyFields.mutableVector().push_back(new FieldRef);
-        FieldRef* const newFieldRef = _keyFields.mutableVector().back();
-        newFieldRef->parse(current.fieldNameStringData());
-    }
+    return _shardKeyPattern.isShardKey(key);
 }
 
 }  // namespace mongo
diff --git a/src/mongo/db/s/collection_metadata.h b/src/mongo/db/s/collection_metadata.h
index a8c8d54de20..29365301a7a 100644
--- a/src/mongo/db/s/collection_metadata.h
+++ b/src/mongo/db/s/collection_metadata.h
@@ -28,12 +28,9 @@
 
 #pragma once
 
-#include "mongo/base/disallow_copying.h"
-#include "mongo/base/owned_pointer_vector.h"
-#include "mongo/db/field_ref_set.h"
-#include "mongo/db/jsobj.h"
 #include "mongo/db/range_arithmetic.h"
 #include "mongo/s/chunk_version.h"
+#include "mongo/s/shard_key_pattern.h"
 
 namespace mongo {
 
@@ -52,8 +49,6 @@ class ChunkType;
  * This class is immutable once constructed.
  */
 class CollectionMetadata {
-    MONGO_DISALLOW_COPYING(CollectionMetadata);
-
 public:
     /**
      * The main way to construct CollectionMetadata is through MetadataLoader or the clone*()
@@ -61,8 +56,11 @@ public:
      *
      * The constructors should not be used directly outside of tests.
      */
-    CollectionMetadata();
-    CollectionMetadata(const BSONObj& keyPattern, ChunkVersion collectionVersion);
+    CollectionMetadata(const BSONObj& keyPattern,
+                       ChunkVersion collectionVersion,
+                       ChunkVersion shardVersion,
+                       RangeMap shardChunksMap);
+
     ~CollectionMetadata();
 
     /**
@@ -148,12 +146,12 @@ public:
         return _chunksMap;
     }
 
-    BSONObj getKeyPattern() const {
-        return _keyPattern;
+    const BSONObj& getKeyPattern() const {
+        return _shardKeyPattern.toBSON();
     }
 
     const std::vector<FieldRef*>& getKeyPatternFields() const {
-        return _keyFields.vector();
+        return _shardKeyPattern.getKeyPatternFields();
     }
 
     BSONObj getMinKey() const;
@@ -164,10 +162,6 @@ public:
         return _chunksMap.size();
     }
 
-    std::size_t getNumPending() const {
-        return _pendingMap.size();
-    }
-
     /**
      * BSON output of the basic metadata information (chunk and shard version).
      */
@@ -188,70 +182,27 @@ public:
      */
     std::string toStringBasic() const;
 
-    /**
-     * This method is used only for unit-tests and it returns a new metadata's instance based on the
-     * current state by adding a chunk with the specified bounds and version. The chunk's version
-     * must be higher than that of all chunks which are in the cache.
-     *
-     * It will fassert if the chunk bounds are incorrect or overlap an existing chunk or if the
-     * chunk version is lower than the maximum one.
-     */
-    std::unique_ptr<CollectionMetadata> clonePlusChunk(const BSONObj& minKey,
-                                                       const BSONObj& maxKey,
-                                                       const ChunkVersion& chunkVersion) const;
-
-    /**
-     * Returns true if this metadata was loaded with all necessary information.
-     */
-    bool isValid() const;
-
 private:
-    // Effectively, the MetadataLoader is this class's builder. So we open an exception and grant it
-    // friendship.
-    friend class MetadataLoader;
+    // Shard key pattern for the collection
+    ShardKeyPattern _shardKeyPattern;
 
     // a version for this collection that identifies the collection incarnation (ie, a
     // dropped and recreated collection with the same name would have a different version)
     ChunkVersion _collVersion;
 
-    //
-    // sharded state below, for when the collection gets sharded
-    //
-
     // highest ChunkVersion for which this metadata's information is accurate
     ChunkVersion _shardVersion;
 
-    // key pattern for chunks under this range
-    BSONObj _keyPattern;
-
-    // A vector owning the FieldRefs parsed from the shard-key pattern of field names.
-    OwnedPointerVector<FieldRef> _keyFields;
-
-    //
-    // RangeMaps represent chunks by mapping the min key to the chunk's max key, allowing
-    // efficient lookup and intersection.
-    //
+    // Map of chunks tracked by this shard
+    RangeMap _chunksMap;
 
     // Map of ranges of chunks that are migrating but have not been confirmed added yet
     RangeMap _pendingMap;
 
-    // Map of chunks tracked by this shard
-    RangeMap _chunksMap;
-
     // A second map from a min key into a range or contiguous chunks. The map is redundant
     // w.r.t. _chunkMap but we expect high chunk contiguity, especially in small
     // installations.
     RangeMap _rangesMap;
-
-    /**
-     * Try to find chunks that are adjacent and record these intervals in the _rangesMap
-     */
-    void fillRanges();
-
-    /**
-     * Creates the _keyField* local data
-     */
-    void fillKeyPatternFields();
 };
 
 }  // namespace mongo
diff --git a/src/mongo/db/s/collection_metadata_test.cpp b/src/mongo/db/s/collection_metadata_test.cpp
index e3c0cc318e8..c765c67ee3a 100644
--- a/src/mongo/db/s/collection_metadata_test.cpp
+++ b/src/mongo/db/s/collection_metadata_test.cpp
@@ -29,111 +29,66 @@
 #include "mongo/platform/basic.h"
 
 #include "mongo/base/status.h"
-#include "mongo/client/remote_command_targeter_mock.h"
-#include "mongo/db/commands.h"
 #include "mongo/db/s/collection_metadata.h"
-#include "mongo/db/s/metadata_loader.h"
-#include "mongo/s/catalog/sharding_catalog_test_fixture.h"
 #include "mongo/s/catalog/type_chunk.h"
-#include "mongo/s/catalog/type_collection.h"
 #include "mongo/s/chunk_version.h"
+#include "mongo/stdx/memory.h"
+#include "mongo/unittest/unittest.h"
 
 namespace mongo {
 namespace {
 
-using std::string;
-using std::unique_ptr;
-using std::vector;
 using unittest::assertGet;
 
-class NoChunkFixture : public ShardingCatalogTestFixture {
+class NoChunkFixture : public unittest::Test {
 protected:
-    void setUp() {
-        ShardingCatalogTestFixture::setUp();
-        setRemote(HostAndPort("FakeRemoteClient:34567"));
-        configTargeter()->setFindHostReturnValue(configHost);
-
-        OID epoch = OID::gen();
-
-        CollectionType collType;
-        collType.setNs(NamespaceString{"test.foo"});
-        collType.setKeyPattern(BSON("a" << 1));
-        collType.setUnique(false);
-        collType.setUpdatedAt(Date_t::fromMillisSinceEpoch(1));
-        collType.setEpoch(epoch);
-        ASSERT_OK(collType.validate());
-
-        // Need a chunk on another shard, otherwise the chunks are invalid in general and we
-        // can't load metadata
-        ChunkType chunkType;
-        chunkType.setNS(NamespaceString{"test.foo"}.ns());
-        chunkType.setShard(ShardId("shard0001"));
-        chunkType.setMin(BSON("a" << MINKEY));
-        chunkType.setMax(BSON("a" << MAXKEY));
-        chunkType.setVersion(ChunkVersion(1, 0, epoch));
-        ASSERT_OK(chunkType.validate());
-        std::vector<BSONObj> chunksToSend{chunkType.toBSON()};
-
-        auto future = launchAsync([this] {
-            auto status = MetadataLoader::makeCollectionMetadata(operationContext(),
-                                                                 catalogClient(),
-                                                                 "test.foo",
-                                                                 "shard0000",
-                                                                 NULL, /* no old metadata */
-                                                                 &_metadata);
-            ASSERT_OK(status);
-            ASSERT_EQUALS(0u, _metadata.getNumChunks());
-        });
-
-        expectFindOnConfigSendBSONObjVector(std::vector<BSONObj>{collType.toBSON()});
-        expectFindOnConfigSendBSONObjVector(chunksToSend);
-
-        future.timed_get(kFutureTimeout);
+    std::unique_ptr<CollectionMetadata> makeCollectionMetadata() const {
+        const OID epoch = OID::gen();
+
+        return stdx::make_unique<CollectionMetadata>(
+            BSON("a" << 1),
+            ChunkVersion(1, 0, epoch),
+            ChunkVersion(0, 0, epoch),
+            SimpleBSONObjComparator::kInstance.makeBSONObjIndexedMap<CachedChunkInfo>());
     }
-
-    const CollectionMetadata& getCollMetadata() const {
-        return _metadata;
-    }
-
-private:
-    CollectionMetadata _metadata;
-    const HostAndPort configHost{HostAndPort(CONFIG_HOST_PORT)};
 };
 
 TEST_F(NoChunkFixture, BasicBelongsToMe) {
-    ASSERT_FALSE(getCollMetadata().keyBelongsToMe(BSON("a" << MINKEY)));
-    ASSERT_FALSE(getCollMetadata().keyBelongsToMe(BSON("a" << 10)));
+    ASSERT_FALSE(makeCollectionMetadata()->keyBelongsToMe(BSON("a" << MINKEY)));
+    ASSERT_FALSE(makeCollectionMetadata()->keyBelongsToMe(BSON("a" << 10)));
 }
 
 TEST_F(NoChunkFixture, CompoundKeyBelongsToMe) {
-    ASSERT_FALSE(getCollMetadata().keyBelongsToMe(BSON("a" << 1 << "b" << 2)));
+    ASSERT_FALSE(makeCollectionMetadata()->keyBelongsToMe(BSON("a" << 1 << "b" << 2)));
 }
 
 TEST_F(NoChunkFixture, IsKeyValid) {
-    ASSERT_TRUE(getCollMetadata().isValidKey(BSON("a"
-                                                  << "abcde")));
-    ASSERT_TRUE(getCollMetadata().isValidKey(BSON("a" << 3)));
-    ASSERT_FALSE(getCollMetadata().isValidKey(BSON("a"
-                                                   << "abcde"
-                                                   << "b"
-                                                   << 1)));
-    ASSERT_FALSE(getCollMetadata().isValidKey(BSON("c"
-                                                   << "abcde")));
+    ASSERT_TRUE(makeCollectionMetadata()->isValidKey(BSON("a"
+                                                          << "abcde")));
+    ASSERT_TRUE(makeCollectionMetadata()->isValidKey(BSON("a" << 3)));
+    ASSERT_FALSE(makeCollectionMetadata()->isValidKey(BSON("a"
+                                                           << "abcde"
+                                                           << "b"
+                                                           << 1)));
+    ASSERT_FALSE(makeCollectionMetadata()->isValidKey(BSON("c"
+                                                           << "abcde")));
 }
 
 TEST_F(NoChunkFixture, getNextFromEmpty) {
     ChunkType nextChunk;
-    ASSERT(!getCollMetadata().getNextChunk(getCollMetadata().getMinKey(), &nextChunk));
+    ASSERT(
+        !makeCollectionMetadata()->getNextChunk(makeCollectionMetadata()->getMinKey(), &nextChunk));
 }
 
 TEST_F(NoChunkFixture, getDifferentFromEmpty) {
     ChunkType differentChunk;
-    ASSERT(!getCollMetadata().getDifferentChunk(getCollMetadata().getMinKey(), &differentChunk));
+    ASSERT(!makeCollectionMetadata()->getDifferentChunk(makeCollectionMetadata()->getMinKey(),
+                                                        &differentChunk));
 }
 
 TEST_F(NoChunkFixture, NoPendingChunks) {
-    ASSERT(!getCollMetadata().keyIsPending(BSON("a" << 15)));
-    ASSERT(!getCollMetadata().keyIsPending(BSON("a" << 25)));
+    ASSERT(!makeCollectionMetadata()->keyIsPending(BSON("a" << 15)));
+    ASSERT(!makeCollectionMetadata()->keyIsPending(BSON("a" << 25)));
 }
 
 TEST_F(NoChunkFixture, FirstPendingChunk) {
@@ -141,7 +96,7 @@ TEST_F(NoChunkFixture, FirstPendingChunk) {
     chunk.setMin(BSON("a" << 10));
     chunk.setMax(BSON("a" << 20));
 
-    unique_ptr<CollectionMetadata> cloned(getCollMetadata().clonePlusPending(chunk));
+    auto cloned(makeCollectionMetadata()->clonePlusPending(chunk));
     ASSERT(cloned->keyIsPending(BSON("a" << 15)));
     ASSERT(!cloned->keyIsPending(BSON("a" << 25)));
     ASSERT(cloned->keyIsPending(BSON("a" << 10)));
@@ -153,7 +108,7 @@ TEST_F(NoChunkFixture, EmptyMultiPendingChunk) {
     chunk.setMin(BSON("a" << 10));
     chunk.setMax(BSON("a" << 20));
 
-    unique_ptr<CollectionMetadata> cloned(getCollMetadata().clonePlusPending(chunk));
+    auto cloned(makeCollectionMetadata()->clonePlusPending(chunk));
 
     chunk.setMin(BSON("a" << 40));
     chunk.setMax(BSON("a" << 50));
@@ -170,7 +125,7 @@ TEST_F(NoChunkFixture, MinusPendingChunk) {
     chunk.setMin(BSON("a" << 10));
     chunk.setMax(BSON("a" << 20));
 
-    unique_ptr<CollectionMetadata> cloned(getCollMetadata().clonePlusPending(chunk));
+    auto cloned(makeCollectionMetadata()->clonePlusPending(chunk));
 
     cloned = cloned->cloneMinusPending(chunk);
     ASSERT(!cloned->keyIsPending(BSON("a" << 15)));
@@ -182,7 +137,7 @@ TEST_F(NoChunkFixture, OverlappingPendingChunk) {
     chunk.setMin(BSON("a" << 10));
     chunk.setMax(BSON("a" << 30));
 
-    unique_ptr<CollectionMetadata> cloned(getCollMetadata().clonePlusPending(chunk));
+    auto cloned(makeCollectionMetadata()->clonePlusPending(chunk));
 
     chunk.setMin(BSON("a" << 20));
     chunk.setMax(BSON("a" << 40));
@@ -199,7 +154,7 @@ TEST_F(NoChunkFixture, OverlappingPendingChunks) {
     chunk.setMin(BSON("a" << 10));
     chunk.setMax(BSON("a" << 30));
 
-    unique_ptr<CollectionMetadata> cloned(getCollMetadata().clonePlusPending(chunk));
+    auto cloned(makeCollectionMetadata()->clonePlusPending(chunk));
 
     chunk.setMin(BSON("a" << 30));
     chunk.setMax(BSON("a" << 50));
@@ -218,39 +173,39 @@ TEST_F(NoChunkFixture, OverlappingPendingChunks) {
 }
 
 TEST_F(NoChunkFixture, OrphanedDataRangeBegin) {
-    const CollectionMetadata& metadata = getCollMetadata();
+    auto metadata(makeCollectionMetadata());
 
     KeyRange keyRange;
-    BSONObj lookupKey = metadata.getMinKey();
-    ASSERT(metadata.getNextOrphanRange(lookupKey, &keyRange));
+    BSONObj lookupKey = metadata->getMinKey();
+    ASSERT(metadata->getNextOrphanRange(lookupKey, &keyRange));
 
-    ASSERT(keyRange.minKey.woCompare(metadata.getMinKey()) == 0);
-    ASSERT(keyRange.maxKey.woCompare(metadata.getMaxKey()) == 0);
+    ASSERT(keyRange.minKey.woCompare(metadata->getMinKey()) == 0);
+    ASSERT(keyRange.maxKey.woCompare(metadata->getMaxKey()) == 0);
 
     // Make sure we don't have any more ranges
-    ASSERT(!metadata.getNextOrphanRange(keyRange.maxKey, &keyRange));
+    ASSERT(!metadata->getNextOrphanRange(keyRange.maxKey, &keyRange));
 }
 
 TEST_F(NoChunkFixture, OrphanedDataRangeMiddle) {
-    const CollectionMetadata& metadata = getCollMetadata();
+    auto metadata(makeCollectionMetadata());
 
     KeyRange keyRange;
     BSONObj lookupKey = BSON("a" << 20);
-    ASSERT(metadata.getNextOrphanRange(lookupKey, &keyRange));
+    ASSERT(metadata->getNextOrphanRange(lookupKey, &keyRange));
 
-    ASSERT(keyRange.minKey.woCompare(metadata.getMinKey()) == 0);
-    ASSERT(keyRange.maxKey.woCompare(metadata.getMaxKey()) == 0);
-    ASSERT(keyRange.keyPattern.woCompare(metadata.getKeyPattern()) == 0);
+    ASSERT(keyRange.minKey.woCompare(metadata->getMinKey()) == 0);
+    ASSERT(keyRange.maxKey.woCompare(metadata->getMaxKey()) == 0);
+    ASSERT(keyRange.keyPattern.woCompare(metadata->getKeyPattern()) == 0);
 
     // Make sure we don't have any more ranges
-    ASSERT(!metadata.getNextOrphanRange(keyRange.maxKey, &keyRange));
+    ASSERT(!metadata->getNextOrphanRange(keyRange.maxKey, &keyRange));
 }
 
 TEST_F(NoChunkFixture, OrphanedDataRangeEnd) {
-    const CollectionMetadata& metadata = getCollMetadata();
+    auto metadata(makeCollectionMetadata());
 
     KeyRange keyRange;
-    ASSERT(!metadata.getNextOrphanRange(metadata.getMaxKey(), &keyRange));
+    ASSERT(!metadata->getNextOrphanRange(metadata->getMaxKey(), &keyRange));
 }
 
 TEST_F(NoChunkFixture, PendingOrphanedDataRanges) {
@@ -258,7 +213,7 @@ TEST_F(NoChunkFixture, PendingOrphanedDataRanges) {
     chunk.setMin(BSON("a" << 10));
     chunk.setMax(BSON("a" << 20));
 
-    unique_ptr<CollectionMetadata> cloned(getCollMetadata().clonePlusPending(chunk));
+    auto cloned(makeCollectionMetadata()->clonePlusPending(chunk));
 
     KeyRange keyRange;
     ASSERT(cloned->getNextOrphanRange(cloned->getMinKey(), &keyRange));
@@ -278,92 +233,59 @@ TEST_F(NoChunkFixture, PendingOrphanedDataRanges) {
  * Fixture with single chunk containing:
  * [10->20)
  */
-class SingleChunkFixture : public ShardingCatalogTestFixture {
+class SingleChunkFixture : public unittest::Test {
 protected:
-    void setUp() {
-        ShardingCatalogTestFixture::setUp();
-        setRemote(HostAndPort("FakeRemoteClient:34567"));
-        configTargeter()->setFindHostReturnValue(configHost);
-
-        CollectionType collType;
-        collType.setNs(NamespaceString{"test.foo"});
-        collType.setKeyPattern(BSON("a" << 1));
-        collType.setUnique(false);
-        collType.setUpdatedAt(Date_t::fromMillisSinceEpoch(1));
-        collType.setEpoch(chunkVersion.epoch());
-
-        BSONObj fooSingle = BSON(
-            ChunkType::name("test.foo-a_10")
-            << ChunkType::ns("test.foo")
-            << ChunkType::min(BSON("a" << 10))
-            << ChunkType::max(BSON("a" << 20))
-            << ChunkType::DEPRECATED_lastmod(Date_t::fromMillisSinceEpoch(chunkVersion.toLong()))
-            << ChunkType::DEPRECATED_epoch(chunkVersion.epoch())
-            << ChunkType::shard("shard0000"));
-        std::vector<BSONObj> chunksToSend{fooSingle};
-
-        auto future = launchAsync([this] {
-            auto status = MetadataLoader::makeCollectionMetadata(operationContext(),
-                                                                 catalogClient(),
-                                                                 "test.foo",
-                                                                 "shard0000",
-                                                                 NULL, /* no old metadata */
-                                                                 &_metadata);
-            ASSERT_OK(status);
-        });
-
-        expectFindOnConfigSendBSONObjVector(std::vector<BSONObj>{collType.toBSON()});
-        expectFindOnConfigSendBSONObjVector(chunksToSend);
-
-        future.timed_get(kFutureTimeout);
-    }
-
-    const CollectionMetadata& getCollMetadata() const {
-        return _metadata;
+    std::unique_ptr<CollectionMetadata> makeCollectionMetadata() const {
+        const OID epoch = OID::gen();
+
+        auto shardChunksMap =
+            SimpleBSONObjComparator::kInstance.makeBSONObjIndexedMap<CachedChunkInfo>();
+        shardChunksMap.emplace(BSON("a" << 10),
+                               CachedChunkInfo(BSON("a" << 20), ChunkVersion(1, 0, epoch)));
+
+        return stdx::make_unique<CollectionMetadata>(BSON("a" << 1),
+                                                     ChunkVersion(1, 0, epoch),
+                                                     ChunkVersion(1, 0, epoch),
+                                                     std::move(shardChunksMap));
     }
-
-    const ChunkVersion chunkVersion{ChunkVersion(1, 0, OID::gen())};
-
-private:
-    CollectionMetadata _metadata;
-    const HostAndPort configHost{HostAndPort(CONFIG_HOST_PORT)};
 };
 
 TEST_F(SingleChunkFixture, BasicBelongsToMe) {
-    ASSERT(getCollMetadata().keyBelongsToMe(BSON("a" << 10)));
-    ASSERT(getCollMetadata().keyBelongsToMe(BSON("a" << 15)));
-    ASSERT(getCollMetadata().keyBelongsToMe(BSON("a" << 19)));
+    ASSERT(makeCollectionMetadata()->keyBelongsToMe(BSON("a" << 10)));
+    ASSERT(makeCollectionMetadata()->keyBelongsToMe(BSON("a" << 15)));
+    ASSERT(makeCollectionMetadata()->keyBelongsToMe(BSON("a" << 19)));
 }
 
 TEST_F(SingleChunkFixture, DoesntBelongsToMe) {
-    ASSERT_FALSE(getCollMetadata().keyBelongsToMe(BSON("a" << 0)));
-    ASSERT_FALSE(getCollMetadata().keyBelongsToMe(BSON("a" << 9)));
-    ASSERT_FALSE(getCollMetadata().keyBelongsToMe(BSON("a" << 20)));
-    ASSERT_FALSE(getCollMetadata().keyBelongsToMe(BSON("a" << 1234)));
-    ASSERT_FALSE(getCollMetadata().keyBelongsToMe(BSON("a" << MINKEY)));
-    ASSERT_FALSE(getCollMetadata().keyBelongsToMe(BSON("a" << MAXKEY)));
+    ASSERT_FALSE(makeCollectionMetadata()->keyBelongsToMe(BSON("a" << 0)));
+    ASSERT_FALSE(makeCollectionMetadata()->keyBelongsToMe(BSON("a" << 9)));
+    ASSERT_FALSE(makeCollectionMetadata()->keyBelongsToMe(BSON("a" << 20)));
+    ASSERT_FALSE(makeCollectionMetadata()->keyBelongsToMe(BSON("a" << 1234)));
+    ASSERT_FALSE(makeCollectionMetadata()->keyBelongsToMe(BSON("a" << MINKEY)));
+    ASSERT_FALSE(makeCollectionMetadata()->keyBelongsToMe(BSON("a" << MAXKEY)));
 }
 
 TEST_F(SingleChunkFixture, CompoudKeyBelongsToMe) {
-    ASSERT(getCollMetadata().keyBelongsToMe(BSON("a" << 15 << "a" << 14)));
+    ASSERT(makeCollectionMetadata()->keyBelongsToMe(BSON("a" << 15 << "a" << 14)));
 }
 
 TEST_F(SingleChunkFixture, getNextFromEmpty) {
     ChunkType nextChunk;
-    ASSERT(getCollMetadata().getNextChunk(getCollMetadata().getMinKey(), &nextChunk));
+    ASSERT(
+        makeCollectionMetadata()->getNextChunk(makeCollectionMetadata()->getMinKey(), &nextChunk));
     ASSERT_EQUALS(0, nextChunk.getMin().woCompare(BSON("a" << 10)));
     ASSERT_EQUALS(0, nextChunk.getMax().woCompare(BSON("a" << 20)));
-    ASSERT_EQUALS(chunkVersion, nextChunk.getVersion());
 }
 
 TEST_F(SingleChunkFixture, GetLastChunkIsFalse) {
     ChunkType nextChunk;
-    ASSERT(!getCollMetadata().getNextChunk(getCollMetadata().getMaxKey(), &nextChunk));
+    ASSERT(
+        !makeCollectionMetadata()->getNextChunk(makeCollectionMetadata()->getMaxKey(), &nextChunk));
 }
 
 TEST_F(SingleChunkFixture, getDifferentFromOneIsFalse) {
     ChunkType differentChunk;
-    ASSERT(!getCollMetadata().getDifferentChunk(BSON("a" << 10), &differentChunk));
+    ASSERT(!makeCollectionMetadata()->getDifferentChunk(BSON("a" << 10), &differentChunk));
 }
 
 TEST_F(SingleChunkFixture, PlusPendingChunk) {
@@ -371,7 +293,7 @@ TEST_F(SingleChunkFixture, PlusPendingChunk) {
     chunk.setMin(BSON("a" << 20));
     chunk.setMax(BSON("a" << 30));
 
-    unique_ptr<CollectionMetadata> cloned(getCollMetadata().clonePlusPending(chunk));
+    auto cloned(makeCollectionMetadata()->clonePlusPending(chunk));
 
     ASSERT(cloned->keyBelongsToMe(BSON("a" << 15)));
     ASSERT(!cloned->keyBelongsToMe(BSON("a" << 25)));
@@ -381,172 +303,96 @@ TEST_F(SingleChunkFixture, PlusPendingChunk) {
 
 TEST_F(SingleChunkFixture, ChunkOrphanedDataRanges) {
     KeyRange keyRange;
-    ASSERT(getCollMetadata().getNextOrphanRange(getCollMetadata().getMinKey(), &keyRange));
-    ASSERT(keyRange.minKey.woCompare(getCollMetadata().getMinKey()) == 0);
+    ASSERT(makeCollectionMetadata()->getNextOrphanRange(makeCollectionMetadata()->getMinKey(),
+                                                        &keyRange));
+    ASSERT(keyRange.minKey.woCompare(makeCollectionMetadata()->getMinKey()) == 0);
     ASSERT(keyRange.maxKey.woCompare(BSON("a" << 10)) == 0);
-    ASSERT(keyRange.keyPattern.woCompare(getCollMetadata().getKeyPattern()) == 0);
+    ASSERT(keyRange.keyPattern.woCompare(makeCollectionMetadata()->getKeyPattern()) == 0);
 
-    ASSERT(getCollMetadata().getNextOrphanRange(keyRange.maxKey, &keyRange));
+    ASSERT(makeCollectionMetadata()->getNextOrphanRange(keyRange.maxKey, &keyRange));
     ASSERT(keyRange.minKey.woCompare(BSON("a" << 20)) == 0);
-    ASSERT(keyRange.maxKey.woCompare(getCollMetadata().getMaxKey()) == 0);
-    ASSERT(keyRange.keyPattern.woCompare(getCollMetadata().getKeyPattern()) == 0);
+    ASSERT(keyRange.maxKey.woCompare(makeCollectionMetadata()->getMaxKey()) == 0);
+    ASSERT(keyRange.keyPattern.woCompare(makeCollectionMetadata()->getKeyPattern()) == 0);
 
-    ASSERT(!getCollMetadata().getNextOrphanRange(keyRange.maxKey, &keyRange));
+    ASSERT(!makeCollectionMetadata()->getNextOrphanRange(keyRange.maxKey, &keyRange));
 }
 
 /**
  * Fixture with single chunk containing:
  * [(min, min)->(max, max))
  */
-class SingleChunkMinMaxCompoundKeyFixture : public ShardingCatalogTestFixture {
+class SingleChunkMinMaxCompoundKeyFixture : public unittest::Test {
 protected:
-    void setUp() {
-        ShardingCatalogTestFixture::setUp();
-        setRemote(HostAndPort("FakeRemoteClient:34567"));
-        configTargeter()->setFindHostReturnValue(configHost);
-
-        OID epoch = OID::gen();
-
-        ChunkVersion chunkVersion = ChunkVersion(1, 0, epoch);
-
-        CollectionType collType;
-        collType.setNs(NamespaceString{"test.foo"});
-        collType.setKeyPattern(BSON("a" << 1));
-        collType.setUnique(false);
-        collType.setUpdatedAt(Date_t::fromMillisSinceEpoch(1));
-        collType.setEpoch(epoch);
-
-        BSONObj fooSingle = BSON(
-            ChunkType::name("test.foo-a_MinKey")
-            << ChunkType::ns("test.foo")
-            << ChunkType::min(BSON("a" << MINKEY << "b" << MINKEY))
-            << ChunkType::max(BSON("a" << MAXKEY << "b" << MAXKEY))
-            << ChunkType::DEPRECATED_lastmod(Date_t::fromMillisSinceEpoch(chunkVersion.toLong()))
-            << ChunkType::DEPRECATED_epoch(epoch)
-            << ChunkType::shard("shard0000"));
-        std::vector<BSONObj> chunksToSend{fooSingle};
-
-        auto future = launchAsync([this] {
-            auto status = MetadataLoader::makeCollectionMetadata(operationContext(),
-                                                                 catalogClient(),
-                                                                 "test.foo",
-                                                                 "shard0000",
-                                                                 NULL, /* no old metadata */
-                                                                 &_metadata);
-            ASSERT_OK(status);
-        });
-
-        expectFindOnConfigSendBSONObjVector(std::vector<BSONObj>{collType.toBSON()});
-        expectFindOnConfigSendBSONObjVector(chunksToSend);
-
-        future.timed_get(kFutureTimeout);
-    }
-
-    const CollectionMetadata& getCollMetadata() const {
-        return _metadata;
+    std::unique_ptr<CollectionMetadata> makeCollectionMetadata() const {
+        const OID epoch = OID::gen();
+
+        auto shardChunksMap =
+            SimpleBSONObjComparator::kInstance.makeBSONObjIndexedMap<CachedChunkInfo>();
+        shardChunksMap.emplace(
+            BSON("a" << MINKEY << "b" << MINKEY),
+            CachedChunkInfo(BSON("a" << MAXKEY << "b" << MAXKEY), ChunkVersion(1, 0, epoch)));
+
+        return stdx::make_unique<CollectionMetadata>(BSON("a" << 1 << "b" << 1),
+                                                     ChunkVersion(1, 0, epoch),
+                                                     ChunkVersion(1, 0, epoch),
+                                                     std::move(shardChunksMap));
     }
-
-    const ChunkVersion chunkVersion{ChunkVersion(1, 0, OID::gen())};
-
-private:
-    CollectionMetadata _metadata;
-    const HostAndPort configHost{HostAndPort(CONFIG_HOST_PORT)};
 };
 
 // Note: no tests for single key belongsToMe because they are not allowed
 // if shard key is compound.
 
 TEST_F(SingleChunkMinMaxCompoundKeyFixture, CompoudKeyBelongsToMe) {
-    ASSERT(getCollMetadata().keyBelongsToMe(BSON("a" << MINKEY << "b" << MINKEY)));
-    ASSERT_FALSE(getCollMetadata().keyBelongsToMe(BSON("a" << MAXKEY << "b" << MAXKEY)));
-    ASSERT(getCollMetadata().keyBelongsToMe(BSON("a" << MINKEY << "b" << 10)));
-    ASSERT(getCollMetadata().keyBelongsToMe(BSON("a" << 10 << "b" << 20)));
+    ASSERT(makeCollectionMetadata()->keyBelongsToMe(BSON("a" << MINKEY << "b" << MINKEY)));
+    ASSERT_FALSE(makeCollectionMetadata()->keyBelongsToMe(BSON("a" << MAXKEY << "b" << MAXKEY)));
+    ASSERT(makeCollectionMetadata()->keyBelongsToMe(BSON("a" << MINKEY << "b" << 10)));
+    ASSERT(makeCollectionMetadata()->keyBelongsToMe(BSON("a" << 10 << "b" << 20)));
 }
 
 /**
  * Fixture with chunks:
  * [(10, 0)->(20, 0)), [(30, 0)->(40, 0))
  */
-class TwoChunksWithGapCompoundKeyFixture : public ShardingCatalogTestFixture {
+class TwoChunksWithGapCompoundKeyFixture : public unittest::Test {
 protected:
-    void setUp() {
-        ShardingCatalogTestFixture::setUp();
-        setRemote(HostAndPort("FakeRemoteClient:34567"));
-        configTargeter()->setFindHostReturnValue(configHost);
-
-        ChunkVersion chunkVersion = ChunkVersion(1, 0, OID::gen());
-
-        CollectionType collType;
-        collType.setNs(NamespaceString{"test.foo"});
-        collType.setKeyPattern(BSON("a" << 1));
-        collType.setUnique(false);
-        collType.setUpdatedAt(Date_t::fromMillisSinceEpoch(1));
-        collType.setEpoch(chunkVersion.epoch());
-
-        std::vector<BSONObj> chunksToSend;
-        chunksToSend.push_back(BSON(
-            ChunkType::name("test.foo-a_10")
-            << ChunkType::ns("test.foo")
-            << ChunkType::min(BSON("a" << 10 << "b" << 0))
-            << ChunkType::max(BSON("a" << 20 << "b" << 0))
-            << ChunkType::DEPRECATED_lastmod(Date_t::fromMillisSinceEpoch(chunkVersion.toLong()))
-            << ChunkType::DEPRECATED_epoch(chunkVersion.epoch())
-            << ChunkType::shard("shard0000")));
-
-        chunkVersion.incMinor();
-
-        chunksToSend.push_back(BSON(
-            ChunkType::name("test.foo-a_10")
-            << ChunkType::ns("test.foo")
-            << ChunkType::min(BSON("a" << 30 << "b" << 0))
-            << ChunkType::max(BSON("a" << 40 << "b" << 0))
-            << ChunkType::DEPRECATED_lastmod(Date_t::fromMillisSinceEpoch(chunkVersion.toLong()))
-            << ChunkType::DEPRECATED_epoch(chunkVersion.epoch())
-            << ChunkType::shard("shard0000")));
-
-        auto future = launchAsync([this] {
-            auto status = MetadataLoader::makeCollectionMetadata(operationContext(),
-                                                                 catalogClient(),
-                                                                 "test.foo",
-                                                                 "shard0000",
-                                                                 NULL, /* no old metadata */
-                                                                 &_metadata);
-            ASSERT_OK(status);
-        });
-
-        expectFindOnConfigSendBSONObjVector(std::vector<BSONObj>{collType.toBSON()});
-        expectFindOnConfigSendBSONObjVector(chunksToSend);
-
-        future.timed_get(kFutureTimeout);
+    std::unique_ptr<CollectionMetadata> makeCollectionMetadata() const {
+        const OID epoch = OID::gen();
+
+        auto shardChunksMap =
+            SimpleBSONObjComparator::kInstance.makeBSONObjIndexedMap<CachedChunkInfo>();
+        shardChunksMap.emplace(
+            BSON("a" << 10 << "b" << 0),
+            CachedChunkInfo(BSON("a" << 20 << "b" << 0), ChunkVersion(1, 0, epoch)));
+        shardChunksMap.emplace(
+            BSON("a" << 30 << "b" << 0),
+            CachedChunkInfo(BSON("a" << 40 << "b" << 0), ChunkVersion(1, 1, epoch)));
+
+        return stdx::make_unique<CollectionMetadata>(BSON("a" << 1 << "b" << 1),
+                                                     ChunkVersion(1, 1, epoch),
+                                                     ChunkVersion(1, 1, epoch),
+                                                     std::move(shardChunksMap));
     }
-
-    const CollectionMetadata& getCollMetadata() const {
-        return _metadata;
-    }
-
-private:
-    CollectionMetadata _metadata;
-    const HostAndPort configHost{HostAndPort(CONFIG_HOST_PORT)};
 };
 
 TEST_F(TwoChunksWithGapCompoundKeyFixture, ChunkGapOrphanedDataRanges) {
     KeyRange keyRange;
-    ASSERT(getCollMetadata().getNextOrphanRange(getCollMetadata().getMinKey(), &keyRange));
-    ASSERT(keyRange.minKey.woCompare(getCollMetadata().getMinKey()) == 0);
+    ASSERT(makeCollectionMetadata()->getNextOrphanRange(makeCollectionMetadata()->getMinKey(),
+                                                        &keyRange));
+    ASSERT(keyRange.minKey.woCompare(makeCollectionMetadata()->getMinKey()) == 0);
     ASSERT(keyRange.maxKey.woCompare(BSON("a" << 10 << "b" << 0)) == 0);
-    ASSERT(keyRange.keyPattern.woCompare(getCollMetadata().getKeyPattern()) == 0);
+    ASSERT(keyRange.keyPattern.woCompare(makeCollectionMetadata()->getKeyPattern()) == 0);
 
-    ASSERT(getCollMetadata().getNextOrphanRange(keyRange.maxKey, &keyRange));
+    ASSERT(makeCollectionMetadata()->getNextOrphanRange(keyRange.maxKey, &keyRange));
     ASSERT(keyRange.minKey.woCompare(BSON("a" << 20 << "b" << 0)) == 0);
     ASSERT(keyRange.maxKey.woCompare(BSON("a" << 30 << "b" << 0)) == 0);
-    ASSERT(keyRange.keyPattern.woCompare(getCollMetadata().getKeyPattern()) == 0);
+    ASSERT(keyRange.keyPattern.woCompare(makeCollectionMetadata()->getKeyPattern()) == 0);
 
-    ASSERT(getCollMetadata().getNextOrphanRange(keyRange.maxKey, &keyRange));
+    ASSERT(makeCollectionMetadata()->getNextOrphanRange(keyRange.maxKey, &keyRange));
     ASSERT(keyRange.minKey.woCompare(BSON("a" << 40 << "b" << 0)) == 0);
-    ASSERT(keyRange.maxKey.woCompare(getCollMetadata().getMaxKey()) == 0);
-    ASSERT(keyRange.keyPattern.woCompare(getCollMetadata().getKeyPattern()) == 0);
+    ASSERT(keyRange.maxKey.woCompare(makeCollectionMetadata()->getMaxKey()) == 0);
+    ASSERT(keyRange.keyPattern.woCompare(makeCollectionMetadata()->getKeyPattern()) == 0);
 
-    ASSERT(!getCollMetadata().getNextOrphanRange(keyRange.maxKey, &keyRange));
+    ASSERT(!makeCollectionMetadata()->getNextOrphanRange(keyRange.maxKey, &keyRange));
 }
 
 TEST_F(TwoChunksWithGapCompoundKeyFixture, ChunkGapAndPendingOrphanedDataRanges) {
@@ -554,7 +400,7 @@ TEST_F(TwoChunksWithGapCompoundKeyFixture, ChunkGapAndPendingOrphanedDataRanges)
     chunk.setMin(BSON("a" << 20 << "b" << 0));
     chunk.setMax(BSON("a" << 30 << "b" << 0));
 
-    unique_ptr<CollectionMetadata> cloned(getCollMetadata().clonePlusPending(chunk));
+    auto cloned(makeCollectionMetadata()->clonePlusPending(chunk));
 
     KeyRange keyRange;
     ASSERT(cloned->getNextOrphanRange(cloned->getMinKey(), &keyRange));
@@ -574,149 +420,97 @@ TEST_F(TwoChunksWithGapCompoundKeyFixture, ChunkGapAndPendingOrphanedDataRanges)
  * Fixture with chunk containing:
  * [min->10) , [10->20) , <gap> , [30->max)
  */
-class ThreeChunkWithRangeGapFixture : public ShardingCatalogTestFixture {
+class ThreeChunkWithRangeGapFixture : public unittest::Test {
 protected:
-    void setUp() {
-        ShardingCatalogTestFixture::setUp();
-        setRemote(HostAndPort("FakeRemoteClient:34567"));
-        configTargeter()->setFindHostReturnValue(configHost);
-
-        OID epoch = OID::gen();
-
-        CollectionType collType;
-        collType.setNs(NamespaceString{"x.y"});
-        collType.setKeyPattern(BSON("a" << 1));
-        collType.setUnique(false);
-        collType.setUpdatedAt(Date_t::fromMillisSinceEpoch(1));
-        collType.setEpoch(epoch);
-
-        std::vector<BSONObj> chunksToSend;
-        {
-            ChunkVersion version(1, 1, epoch);
-            chunksToSend.push_back(BSON(
-                ChunkType::name("x.y-a_MinKey")
-                << ChunkType::ns("x.y")
-                << ChunkType::min(BSON("a" << MINKEY))
-                << ChunkType::max(BSON("a" << 10))
-                << ChunkType::DEPRECATED_lastmod(Date_t::fromMillisSinceEpoch(version.toLong()))
-                << ChunkType::DEPRECATED_epoch(version.epoch())
-                << ChunkType::shard("shard0000")));
-        }
-
-        {
-            ChunkVersion version(1, 3, epoch);
-            chunksToSend.push_back(BSON(
-                ChunkType::name("x.y-a_10")
-                << ChunkType::ns("x.y")
-                << ChunkType::min(BSON("a" << 10))
-                << ChunkType::max(BSON("a" << 20))
-                << ChunkType::DEPRECATED_lastmod(Date_t::fromMillisSinceEpoch(version.toLong()))
-                << ChunkType::DEPRECATED_epoch(version.epoch())
-                << ChunkType::shard("shard0000")));
-        }
-
-        {
-            ChunkVersion version(1, 2, epoch);
-            chunksToSend.push_back(BSON(
-                ChunkType::name("x.y-a_30")
-                << ChunkType::ns("x.y")
-                << ChunkType::min(BSON("a" << 30))
-                << ChunkType::max(BSON("a" << MAXKEY))
-                << ChunkType::DEPRECATED_lastmod(Date_t::fromMillisSinceEpoch(version.toLong()))
-                << ChunkType::DEPRECATED_epoch(version.epoch())
-                << ChunkType::shard("shard0000")));
-        }
-
-        auto future = launchAsync([this] {
-            auto status = MetadataLoader::makeCollectionMetadata(operationContext(),
-                                                                 catalogClient(),
-                                                                 "test.foo",
-                                                                 "shard0000",
-                                                                 NULL, /* no old metadata */
-                                                                 &_metadata);
-            ASSERT_OK(status);
-        });
-
-        expectFindOnConfigSendBSONObjVector(std::vector<BSONObj>{collType.toBSON()});
-        expectFindOnConfigSendBSONObjVector(chunksToSend);
-
-        future.timed_get(kFutureTimeout);
+    std::unique_ptr<CollectionMetadata> makeCollectionMetadata() const {
+        const OID epoch = OID::gen();
+
+        auto shardChunksMap =
+            SimpleBSONObjComparator::kInstance.makeBSONObjIndexedMap<CachedChunkInfo>();
+        shardChunksMap.emplace(BSON("a" << MINKEY),
+                               CachedChunkInfo(BSON("a" << 10), ChunkVersion(1, 1, epoch)));
+        shardChunksMap.emplace(BSON("a" << 10),
+                               CachedChunkInfo(BSON("a" << 20), ChunkVersion(1, 3, epoch)));
+        shardChunksMap.emplace(BSON("a" << 30),
+                               CachedChunkInfo(BSON("a" << MAXKEY), ChunkVersion(1, 2, epoch)));
+
+        return stdx::make_unique<CollectionMetadata>(BSON("a" << 1),
+                                                     ChunkVersion(1, 3, epoch),
+                                                     ChunkVersion(1, 3, epoch),
+                                                     std::move(shardChunksMap));
     }
-
-    const CollectionMetadata& getCollMetadata() const {
-        return _metadata;
-    }
-
-private:
-    CollectionMetadata _metadata;
-    const HostAndPort configHost{HostAndPort(CONFIG_HOST_PORT)};
 };
 
 TEST_F(ThreeChunkWithRangeGapFixture, ChunkVersionsMatch) {
-    const OID epoch = getCollMetadata().getCollVersion().epoch();
+    auto metadata(makeCollectionMetadata());
 
     ChunkType chunk;
 
-    ASSERT(getCollMetadata().getNextChunk(BSON("a" << MINKEY), &chunk));
-    ASSERT_EQ(ChunkVersion(1, 1, epoch), chunk.getVersion());
+    ASSERT(metadata->getNextChunk(BSON("a" << MINKEY), &chunk));
+    ASSERT_EQ(ChunkVersion(1, 1, metadata->getCollVersion().epoch()), chunk.getVersion());
+    ASSERT_BSONOBJ_EQ(metadata->getMinKey(), chunk.getMin());
 
-    ASSERT(getCollMetadata().getNextChunk(BSON("a" << 30), &chunk));
-    ASSERT_EQ(ChunkVersion(1, 2, epoch), chunk.getVersion());
+    ASSERT(metadata->getNextChunk(BSON("a" << 10), &chunk));
+    ASSERT_EQ(ChunkVersion(1, 3, metadata->getCollVersion().epoch()), chunk.getVersion());
 
-    ASSERT(getCollMetadata().getNextChunk(BSON("a" << 10), &chunk));
-    ASSERT_EQ(ChunkVersion(1, 3, epoch), chunk.getVersion());
+    ASSERT(metadata->getNextChunk(BSON("a" << 30), &chunk));
+    ASSERT_EQ(ChunkVersion(1, 2, metadata->getCollVersion().epoch()), chunk.getVersion());
+    ASSERT_BSONOBJ_EQ(metadata->getMaxKey(), chunk.getMax());
 }
 
 TEST_F(ThreeChunkWithRangeGapFixture, ShardOwnsDoc) {
-    ASSERT(getCollMetadata().keyBelongsToMe(BSON("a" << 5)));
-    ASSERT(getCollMetadata().keyBelongsToMe(BSON("a" << 10)));
-    ASSERT(getCollMetadata().keyBelongsToMe(BSON("a" << 30)));
-    ASSERT(getCollMetadata().keyBelongsToMe(BSON("a" << 40)));
+    ASSERT(makeCollectionMetadata()->keyBelongsToMe(BSON("a" << 5)));
+    ASSERT(makeCollectionMetadata()->keyBelongsToMe(BSON("a" << 10)));
+    ASSERT(makeCollectionMetadata()->keyBelongsToMe(BSON("a" << 30)));
+    ASSERT(makeCollectionMetadata()->keyBelongsToMe(BSON("a" << 40)));
 }
 
 TEST_F(ThreeChunkWithRangeGapFixture, ShardDoesntOwnDoc) {
-    ASSERT_FALSE(getCollMetadata().keyBelongsToMe(BSON("a" << 25)));
-    ASSERT_FALSE(getCollMetadata().keyBelongsToMe(BSON("a" << MAXKEY)));
+    ASSERT_FALSE(makeCollectionMetadata()->keyBelongsToMe(BSON("a" << 25)));
+    ASSERT_FALSE(makeCollectionMetadata()->keyBelongsToMe(BSON("a" << MAXKEY)));
 }
 
 TEST_F(ThreeChunkWithRangeGapFixture, GetNextFromEmpty) {
     ChunkType nextChunk;
-    ASSERT(getCollMetadata().getNextChunk(getCollMetadata().getMinKey(), &nextChunk));
+    ASSERT(
+        makeCollectionMetadata()->getNextChunk(makeCollectionMetadata()->getMinKey(), &nextChunk));
     ASSERT_EQUALS(0, nextChunk.getMin().woCompare(BSON("a" << MINKEY)));
     ASSERT_EQUALS(0, nextChunk.getMax().woCompare(BSON("a" << 10)));
 }
 
 TEST_F(ThreeChunkWithRangeGapFixture, GetNextFromMiddle) {
     ChunkType nextChunk;
-    ASSERT(getCollMetadata().getNextChunk(BSON("a" << 20), &nextChunk));
+    ASSERT(makeCollectionMetadata()->getNextChunk(BSON("a" << 20), &nextChunk));
     ASSERT_EQUALS(0, nextChunk.getMin().woCompare(BSON("a" << 30)));
     ASSERT_EQUALS(0, nextChunk.getMax().woCompare(BSON("a" << MAXKEY)));
 }
 
 TEST_F(ThreeChunkWithRangeGapFixture, GetNextFromLast) {
     ChunkType nextChunk;
-    ASSERT(getCollMetadata().getNextChunk(BSON("a" << 30), &nextChunk));
+    ASSERT(makeCollectionMetadata()->getNextChunk(BSON("a" << 30), &nextChunk));
     ASSERT_EQUALS(0, nextChunk.getMin().woCompare(BSON("a" << 30)));
     ASSERT_EQUALS(0, nextChunk.getMax().woCompare(BSON("a" << MAXKEY)));
 }
 
 TEST_F(ThreeChunkWithRangeGapFixture, GetDifferentFromBeginning) {
+    auto metadata(makeCollectionMetadata());
+
     ChunkType differentChunk;
-    ASSERT(getCollMetadata().getDifferentChunk(getCollMetadata().getMinKey(), &differentChunk));
-    ASSERT_EQUALS(0, differentChunk.getMin().woCompare(BSON("a" << 10)));
-    ASSERT_EQUALS(0, differentChunk.getMax().woCompare(BSON("a" << 20)));
+    ASSERT(metadata->getDifferentChunk(metadata->getMinKey(), &differentChunk));
+    ASSERT_BSONOBJ_EQ(BSON("a" << 10), differentChunk.getMin());
+    ASSERT_BSONOBJ_EQ(BSON("a" << 20), differentChunk.getMax());
 }
 
 TEST_F(ThreeChunkWithRangeGapFixture, GetDifferentFromMiddle) {
     ChunkType differentChunk;
-    ASSERT(getCollMetadata().getDifferentChunk(BSON("a" << 10), &differentChunk));
+    ASSERT(makeCollectionMetadata()->getDifferentChunk(BSON("a" << 10), &differentChunk));
     ASSERT_EQUALS(0, differentChunk.getMin().woCompare(BSON("a" << MINKEY)));
     ASSERT_EQUALS(0, differentChunk.getMax().woCompare(BSON("a" << 10)));
 }
 
 TEST_F(ThreeChunkWithRangeGapFixture, GetDifferentFromLast) {
     ChunkType differentChunk;
-    ASSERT(getCollMetadata().getDifferentChunk(BSON("a" << 30), &differentChunk));
+    ASSERT(makeCollectionMetadata()->getDifferentChunk(BSON("a" << 30), &differentChunk));
     ASSERT_EQUALS(0, differentChunk.getMin().woCompare(BSON("a" << MINKEY)));
     ASSERT_EQUALS(0, differentChunk.getMax().woCompare(BSON("a" << 10)));
 }
diff --git a/src/mongo/db/s/merge_chunks_command.cpp b/src/mongo/db/s/merge_chunks_command.cpp
index e7728f27560..f4a8b6f3310 100644
--- a/src/mongo/db/s/merge_chunks_command.cpp
+++ b/src/mongo/db/s/merge_chunks_command.cpp
@@ -90,10 +90,10 @@ Status mergeChunks(OperationContext* txn,
         txn, nss.ns(), whyMessage, DistLockManager::kSingleLockAttemptTimeout);
 
     if (!scopedDistLock.isOK()) {
-        std::string errmsg = stream() << "could not acquire collection lock for " << nss.ns()
-                                      << " to merge chunks in [" << redact(minKey) << ", "
-                                      << redact(maxKey) << ")"
-                                      << causedBy(scopedDistLock.getStatus());
+        const std::string errmsg = stream() << "could not acquire collection lock for " << nss.ns()
+                                            << " to merge chunks in [" << redact(minKey) << ", "
+                                            << redact(maxKey) << ")"
+                                            << causedBy(scopedDistLock.getStatus());
 
         warning() << errmsg;
         return Status(scopedDistLock.getStatus().code(), errmsg);
@@ -109,7 +109,7 @@ Status mergeChunks(OperationContext* txn,
     Status refreshStatus = shardingState->refreshMetadataNow(txn, nss, &shardVersion);
 
     if (!refreshStatus.isOK()) {
-        std::string errmsg = str::stream()
+        const std::string errmsg = str::stream()
             << "could not merge chunks, failed to refresh metadata for " << nss.ns()
             << causedBy(redact(refreshStatus));
 
@@ -118,7 +118,7 @@ Status mergeChunks(OperationContext* txn,
     }
 
     if (epoch.isSet() && shardVersion.epoch() != epoch) {
-        std::string errmsg = stream()
+        const std::string errmsg = stream()
             << "could not merge chunks, collection " << nss.ns() << " has changed"
             << " since merge was sent"
             << "(sent epoch : " << epoch.toString()
@@ -133,9 +133,9 @@ Status mergeChunks(OperationContext* txn,
         AutoGetCollection autoColl(txn, nss, MODE_IS);
 
         metadata = CollectionShardingState::get(txn, nss.ns())->getMetadata();
-        if (!metadata || metadata->getKeyPattern().isEmpty()) {
-            std::string errmsg = stream() << "could not merge chunks, collection " << nss.ns()
-                                          << " is not sharded";
+        if (!metadata) {
+            const std::string errmsg = stream() << "could not merge chunks, collection " << nss.ns()
+                                                << " is not sharded";
 
             warning() << errmsg;
             return Status(ErrorCodes::IllegalOperation, errmsg);
@@ -145,10 +145,11 @@ Status mergeChunks(OperationContext* txn,
     dassert(metadata->getShardVersion().equals(shardVersion));
 
     if (!metadata->isValidKey(minKey) || !metadata->isValidKey(maxKey)) {
-        std::string errmsg = stream() << "could not merge chunks, the range "
-                                      << redact(rangeToString(minKey, maxKey)) << " is not valid"
-                                      << " for collection " << nss.ns() << " with key pattern "
-                                      << metadata->getKeyPattern().toString();
+        const std::string errmsg = stream()
+            << "could not merge chunks, the range " << redact(rangeToString(minKey, maxKey))
+            << " is not valid"
+            << " for collection " << nss.ns() << " with key pattern "
+            << metadata->getKeyPattern().toString();
 
         warning() << errmsg;
         return Status(ErrorCodes::IllegalOperation, errmsg);
diff --git a/src/mongo/db/s/metadata_loader.cpp b/src/mongo/db/s/metadata_loader.cpp
deleted file mode 100644
index a181328222a..00000000000
--- a/src/mongo/db/s/metadata_loader.cpp
+++ /dev/null
@@ -1,240 +0,0 @@
-/**
- *    Copyright (C) 2012 10gen Inc.
- *
- *    This program is free software: you can redistribute it and/or  modify
- *    it under the terms of the GNU Affero General Public License, version 3,
- *    as published by the Free Software Foundation.
- *
- *    This program is distributed in the hope that it will be useful,
- *    but WITHOUT ANY WARRANTY; without even the implied warranty of
- *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- *    GNU Affero General Public License for more details.
- *
- *    You should have received a copy of the GNU Affero General Public License
- *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
- *
- *    As a special exception, the copyright holders give permission to link the
- *    code of portions of this program with the OpenSSL library under certain
- *    conditions as described in each individual source file and distribute
- *    linked combinations including the program with the OpenSSL library. You
- *    must comply with the GNU Affero General Public License in all respects
- *    for all of the code used other than as permitted herein. If you modify
- *    file(s) with this exception, you may extend this exception to your
- *    version of the file(s), but you are not obligated to do so. If you do not
- *    wish to do so, delete this exception statement from your version. If you
- *    delete this exception statement from all source files in the program,
- *    then also delete it in the license file.
- */
-
-#define MONGO_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kSharding
-
-#include "mongo/platform/basic.h"
-
-#include "mongo/db/s/metadata_loader.h"
-
-#include <vector>
-
-#include "mongo/db/s/collection_metadata.h"
-#include "mongo/s/catalog/sharding_catalog_client.h"
-#include "mongo/s/catalog/type_chunk.h"
-#include "mongo/s/catalog/type_collection.h"
-#include "mongo/s/chunk_diff.h"
-#include "mongo/s/chunk_version.h"
-#include "mongo/util/log.h"
-
-namespace mongo {
-
-using std::make_pair;
-using std::map;
-using std::pair;
-using std::string;
-
-namespace {
-
-/**
- * This is an adapter so we can use config diffs - mongos and mongod do them slightly
- * differently.
- *
- * The mongod adapter here tracks only a single shard, and stores ranges by (min, max).
- */
-class SCMConfigDiffTracker : public ConfigDiffTracker<CachedChunkInfo> {
-public:
-    SCMConfigDiffTracker(const std::string& ns,
-                         RangeMap* currMap,
-                         ChunkVersion* maxVersion,
-                         MaxChunkVersionMap* maxShardVersions,
-                         const ShardId& currShard)
-        : ConfigDiffTracker<CachedChunkInfo>(ns, currMap, maxVersion, maxShardVersions),
-          _currShard(currShard) {}
-
-    virtual bool isTracked(const ChunkType& chunk) const {
-        return chunk.getShard() == _currShard;
-    }
-
-    virtual pair<BSONObj, CachedChunkInfo> rangeFor(OperationContext* txn,
-                                                    const ChunkType& chunk) const {
-        return make_pair(chunk.getMin(), CachedChunkInfo(chunk.getMax(), chunk.getVersion()));
-    }
-
-    virtual ShardId shardFor(OperationContext* txn, const ShardId& name) const {
-        return name;
-    }
-
-    virtual string nameFrom(const string& shard) const {
-        return shard;
-    }
-
-private:
-    const ShardId _currShard;
-};
-
-}  // namespace
-
-Status MetadataLoader::makeCollectionMetadata(OperationContext* txn,
-                                              ShardingCatalogClient* catalogClient,
-                                              const string& ns,
-                                              const string& shard,
-                                              const CollectionMetadata* oldMetadata,
-                                              CollectionMetadata* metadata) {
-    Status initCollectionStatus = _initCollection(txn, catalogClient, ns, shard, metadata);
-    if (!initCollectionStatus.isOK()) {
-        return initCollectionStatus;
-    }
-
-    return _initChunks(txn, catalogClient, ns, shard, oldMetadata, metadata);
-}
-
-Status MetadataLoader::_initCollection(OperationContext* txn,
-                                       ShardingCatalogClient* catalogClient,
-                                       const string& ns,
-                                       const string& shard,
-                                       CollectionMetadata* metadata) {
-    auto coll = catalogClient->getCollection(txn, ns);
-    if (!coll.isOK()) {
-        return coll.getStatus();
-    }
-
-    const auto& collInfo = coll.getValue().value;
-    if (collInfo.getDropped()) {
-        return {ErrorCodes::NamespaceNotFound,
-                str::stream() << "Could not load metadata because collection " << ns
-                              << " was dropped"};
-    }
-
-    metadata->_keyPattern = collInfo.getKeyPattern().toBSON();
-    metadata->fillKeyPatternFields();
-    metadata->_shardVersion = ChunkVersion(0, 0, collInfo.getEpoch());
-    metadata->_collVersion = ChunkVersion(0, 0, collInfo.getEpoch());
-
-    return Status::OK();
-}
-
-Status MetadataLoader::_initChunks(OperationContext* txn,
-                                   ShardingCatalogClient* catalogClient,
-                                   const string& ns,
-                                   const string& shard,
-                                   const CollectionMetadata* oldMetadata,
-                                   CollectionMetadata* metadata) {
-    const OID epoch = metadata->getCollVersion().epoch();
-
-    SCMConfigDiffTracker::MaxChunkVersionMap versionMap;
-    versionMap[shard] = metadata->_shardVersion;
-
-    bool fullReload = true;
-
-    // Check to see if we should use the old version or not.
-    if (oldMetadata) {
-        // If our epochs are compatible, it's useful to use the old metadata for diffs: this leads
-        // to a performance gain because not all the chunks must be reloaded, just the ones this
-        // shard has not seen -- they will have higher versions than present in oldMetadata.
-        if (oldMetadata->getCollVersion().hasEqualEpoch(epoch)) {
-            fullReload = false;
-            invariant(oldMetadata->isValid());
-
-            versionMap[shard] = oldMetadata->_shardVersion;
-            metadata->_collVersion = oldMetadata->_collVersion;
-
-            // TODO: This could be made more efficient if copying not required, but
-            // not as frequently reloaded as in mongos.
-            metadata->_chunksMap = oldMetadata->_chunksMap;
-
-            LOG(2) << "loading new chunks for collection " << ns
-                   << " using old metadata w/ version " << oldMetadata->getShardVersion() << " and "
-                   << metadata->_chunksMap.size() << " chunks";
-        } else {
-            warning() << "reloading collection metadata for " << ns << " with new epoch "
-                      << epoch.toString() << ", the current epoch is "
-                      << oldMetadata->getCollVersion().epoch().toString();
-        }
-    }
-
-    // Exposes the new metadata's range map and version to the "differ" which would ultimately be
-    // responsible for filling them up
-    SCMConfigDiffTracker differ(
-        ns, &metadata->_chunksMap, &metadata->_collVersion, &versionMap, shard);
-
-    try {
-        const auto diffQuery = SCMConfigDiffTracker::createConfigDiffQuery(NamespaceString(ns),
-                                                                           metadata->_collVersion);
-        std::vector<ChunkType> chunks;
-        Status status = catalogClient->getChunks(txn,
-                                                 diffQuery.query,
-                                                 diffQuery.sort,
-                                                 boost::none,
-                                                 &chunks,
-                                                 nullptr,
-                                                 repl::ReadConcernLevel::kMajorityReadConcern);
-        if (!status.isOK()) {
-            return status;
-        }
-
-        //
-        // The diff tracker should always find at least one chunk (the highest chunk we saw
-        // last time).  If not, something has changed on the config server (potentially between
-        // when we read the collection data and when we read the chunks data).
-        //
-        int diffsApplied = differ.calculateConfigDiff(txn, chunks);
-        if (diffsApplied > 0) {
-            // Chunks found, return ok
-            LOG(2) << "loaded " << diffsApplied << " chunks into new metadata for " << ns
-                   << " with version " << metadata->_collVersion;
-
-            // If the last chunk was moved off of this shard, the shardVersion should be reset to
-            // zero (if we did not conduct a full reload and oldMetadata was present,
-            // versionMap[shard] was previously set to the oldMetadata's shardVersion for
-            // performance gains).
-            if (!fullReload && metadata->_chunksMap.empty()) {
-                versionMap[shard] = ChunkVersion(0, 0, epoch);
-            }
-
-            metadata->_shardVersion = versionMap[shard];
-            metadata->fillRanges();
-
-            invariant(metadata->isValid());
-            return Status::OK();
-        } else if (diffsApplied == 0) {
-            // No chunks found, the collection is dropping or we're confused
-            // If this is a full reload, assume it is a drop for backwards compatibility
-            // TODO: drop the config.collections entry *before* the chunks and eliminate this
-            // ambiguity
-
-            return {fullReload ? ErrorCodes::NamespaceNotFound : ErrorCodes::RemoteChangeDetected,
-                    str::stream() << "No chunks found when reloading " << ns
-                                  << ", previous version was "
-                                  << metadata->_collVersion.toString()
-                                  << (fullReload ? ", this is a drop" : "")};
-        } else {
-            // Invalid chunks found, our epoch may have changed because we dropped/recreated the
-            // collection
-            return {ErrorCodes::RemoteChangeDetected,
-                    str::stream() << "Invalid chunks found when reloading " << ns
-                                  << ", previous version was "
-                                  << metadata->_collVersion.toString()
-                                  << ", this should be rare"};
-        }
-    } catch (const DBException& ex) {
-        return ex.toStatus();
-    }
-}
-
-}  // namespace mongo
diff --git a/src/mongo/db/s/metadata_loader.h b/src/mongo/db/s/metadata_loader.h
deleted file mode 100644
index e4624a73f8f..00000000000
--- a/src/mongo/db/s/metadata_loader.h
+++ /dev/null
@@ -1,130 +0,0 @@
-/**
- *    Copyright (C) 2012 10gen Inc.
- *
- *    This program is free software: you can redistribute it and/or  modify
- *    it under the terms of the GNU Affero General Public License, version 3,
- *    as published by the Free Software Foundation.
- *
- *    This program is distributed in the hope that it will be useful,
- *    but WITHOUT ANY WARRANTY; without even the implied warranty of
- *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- *    GNU Affero General Public License for more details.
- *
- *    You should have received a copy of the GNU Affero General Public License
- *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
- *
- *    As a special exception, the copyright holders give permission to link the
- *    code of portions of this program with the OpenSSL library under certain
- *    conditions as described in each individual source file and distribute
- *    linked combinations including the program with the OpenSSL library. You
- *    must comply with the GNU Affero General Public License in all respects
- *    for all of the code used other than as permitted herein. If you modify
- *    file(s) with this exception, you may extend this exception to your
- *    version of the file(s), but you are not obligated to do so. If you do not
- *    wish to do so, delete this exception statement from your version. If you
- *    delete this exception statement from all source files in the program,
- *    then also delete it in the license file.
- */
-
-#pragma once
-
-#include <string>
-
-#include "mongo/base/status.h"
-
-namespace mongo {
-
-class ShardingCatalogClient;
-class CollectionMetadata;
-class CollectionType;
-class OperationContext;
-
-/**
- * The MetadataLoader is responsible for interfacing with the config servers and previous
- * metadata to build new instances of CollectionMetadata.  MetadataLoader is the "builder"
- * class for metadata.
- *
- * CollectionMetadata has both persisted and volatile state (for now) - the persisted
- * config server chunk state and the volatile pending state which is only tracked locally
- * while a server is the primary.  This requires a two-step loading process - the persisted
- * chunk state *cannot* be loaded in a DBLock lock while the pending chunk state *must* be.
- *
- * Example usage:
- * beforeMetadata = <get latest local metadata>;
- * remoteMetadata = makeCollectionMetadata( beforeMetadata, remoteMetadata );
- * DBLock lock(txn, dbname, MODE_X);
- * afterMetadata = <get latest local metadata>;
- *
- * The loader will go out of its way to try to fetch the smaller amount possible of data
- * from the config server without sacrificing the freshness and accuracy of the metadata it
- * builds. (See ConfigDiffTracker class.)
- *
- */
-class MetadataLoader {
-public:
-    /**
-     * Fills a new metadata instance representing the chunkset of the collection 'ns'
-     * (or its entirety, if not sharded) that lives on 'shard' with data from the config server.
-     * Optionally, uses an 'oldMetadata' for the same 'ns'/'shard'; the contents of
-     * 'oldMetadata' can help reducing the amount of data read from the config servers.
-     *
-     * Locking note:
-     *    + Must not be called in a DBLock, since this loads over the network
-     *
-     * OK on success.
-     *
-     * Failure return values:
-     * Abnormal:
-     * @return FailedToParse if there was an error parsing the remote config data
-     * Normal:
-     * @return NamespaceNotFound if the collection no longer exists
-     * @return HostUnreachable if there was an error contacting the config servers
-     * @return RemoteChangeDetected if the data loaded was modified by another operation
-     */
-    static Status makeCollectionMetadata(OperationContext* txn,
-                                         ShardingCatalogClient* catalogClient,
-                                         const std::string& ns,
-                                         const std::string& shard,
-                                         const CollectionMetadata* oldMetadata,
-                                         CollectionMetadata* metadata);
-
-private:
-    /**
-     * Returns OK and fills in the internal state of 'metadata' with general collection
-     * information, not including chunks.
-     *
-     * If information about the collection can be accessed or is invalid, returns:
-     * @return NamespaceNotFound if the collection no longer exists
-     * @return FailedToParse if there was an error parsing the remote config data
-     * @return HostUnreachable if there was an error contacting the config servers
-     * @return RemoteChangeDetected if the collection doc loaded is unexpectedly different
-     *
-     */
-    static Status _initCollection(OperationContext* txn,
-                                  ShardingCatalogClient* catalogClient,
-                                  const std::string& ns,
-                                  const std::string& shard,
-                                  CollectionMetadata* metadata);
-
-    /**
-     * Returns OK and fills in the chunk state of 'metadata' to portray the chunks of the
-     * collection 'ns' that sit in 'shard'. If provided, uses the contents of 'oldMetadata'
-     * as a base (see description in initCollection above).
-     *
-     * If information about the chunks can be accessed or is invalid, returns:
-     * @return HostUnreachable if there was an error contacting the config servers
-     * @return RemoteChangeDetected if the chunks loaded are unexpectedly different
-     *
-     * For backwards compatibility,
-     * @return NamespaceNotFound if there are no chunks loaded and an epoch change is detected
-     * TODO: @return FailedToParse
-     */
-    static Status _initChunks(OperationContext* txn,
-                              ShardingCatalogClient* catalogClient,
-                              const std::string& ns,
-                              const std::string& shard,
-                              const CollectionMetadata* oldMetadata,
-                              CollectionMetadata* metadata);
-};
-
-}  // namespace mongo
diff --git a/src/mongo/db/s/metadata_loader_test.cpp b/src/mongo/db/s/metadata_loader_test.cpp
deleted file mode 100644
index 13a16b313ad..00000000000
--- a/src/mongo/db/s/metadata_loader_test.cpp
+++ /dev/null
@@ -1,413 +0,0 @@
-/**
- *    Copyright (C) 2012 10gen Inc.
- *
- *    This program is free software: you can redistribute it and/or  modify
- *    it under the terms of the GNU Affero General Public License, version 3,
- *    as published by the Free Software Foundation.
- *
- *    This program is distributed in the hope that it will be useful,
- *    but WITHOUT ANY WARRANTY; without even the implied warranty of
- *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- *    GNU Affero General Public License for more details.
- *
- *    You should have received a copy of the GNU Affero General Public License
- *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
- *
- *    As a special exception, the copyright holders give permission to link the
- *    code of portions of this program with the OpenSSL library under certain
- *    conditions as described in each individual source file and distribute
- *    linked combinations including the program with the OpenSSL library. You
- *    must comply with the GNU Affero General Public License in all respects
- *    for all of the code used other than as permitted herein. If you modify
- *    file(s) with this exception, you may extend this exception to your
- *    version of the file(s), but you are not obligated to do so. If you do not
- *    wish to do so, delete this exception statement from your version. If you
- *    delete this exception statement from all source files in the program,
- *    then also delete it in the license file.
- */
-
-#include "mongo/platform/basic.h"
-
-#include "mongo/base/status.h"
-#include "mongo/client/remote_command_targeter_mock.h"
-#include "mongo/db/s/collection_metadata.h"
-#include "mongo/db/s/metadata_loader.h"
-#include "mongo/s/catalog/sharding_catalog_test_fixture.h"
-#include "mongo/s/catalog/type_chunk.h"
-#include "mongo/s/catalog/type_collection.h"
-
-namespace mongo {
-namespace {
-
-using std::string;
-using std::unique_ptr;
-using std::vector;
-
-class MetadataLoaderFixture : public ShardingCatalogTestFixture {
-public:
-    MetadataLoaderFixture() = default;
-    ~MetadataLoaderFixture() = default;
-
-protected:
-    void setUp() override {
-        ShardingCatalogTestFixture::setUp();
-        setRemote(HostAndPort("FakeRemoteClient:34567"));
-        configTargeter()->setFindHostReturnValue(configHost);
-        _maxCollVersion = ChunkVersion(1, 0, OID::gen());
-        _loader.reset(new MetadataLoader);
-    }
-
-    void expectFindOnConfigSendCollectionDefault() {
-        CollectionType collType;
-        collType.setNs(NamespaceString{"test.foo"});
-        collType.setKeyPattern(BSON("a" << 1));
-        collType.setUnique(false);
-        collType.setUpdatedAt(Date_t::fromMillisSinceEpoch(1));
-        collType.setEpoch(_maxCollVersion.epoch());
-        ASSERT_OK(collType.validate());
-        expectFindOnConfigSendBSONObjVector(std::vector<BSONObj>{collType.toBSON()});
-    }
-
-    void expectFindOnConfigSendChunksDefault() {
-        BSONObj chunk = BSON(
-            ChunkType::name("test.foo-a_MinKey")
-            << ChunkType::ns("test.foo")
-            << ChunkType::min(BSON("a" << MINKEY))
-            << ChunkType::max(BSON("a" << MAXKEY))
-            << ChunkType::DEPRECATED_lastmod(Date_t::fromMillisSinceEpoch(_maxCollVersion.toLong()))
-            << ChunkType::DEPRECATED_epoch(_maxCollVersion.epoch())
-            << ChunkType::shard("shard0000"));
-        expectFindOnConfigSendBSONObjVector(std::vector<BSONObj>{chunk});
-    }
-
-    MetadataLoader& loader() const {
-        return *_loader;
-    }
-
-    void getMetadataFor(const OwnedPointerVector<ChunkType>& chunks, CollectionMetadata* metadata) {
-        // Infer namespace, shard, epoch, keypattern from first chunk
-        const ChunkType* firstChunk = *(chunks.vector().begin());
-        const string ns = firstChunk->getNS();
-        const string shardName = firstChunk->getShard().toString();
-        const OID epoch = firstChunk->getVersion().epoch();
-
-        CollectionType coll;
-        coll.setNs(NamespaceString{ns});
-        coll.setKeyPattern(BSON("a" << 1));
-        coll.setUpdatedAt(Date_t::fromMillisSinceEpoch(1));
-        coll.setEpoch(epoch);
-        ASSERT_OK(coll.validate());
-        std::vector<BSONObj> collToSend{coll.toBSON()};
-
-        ChunkVersion version(1, 0, epoch);
-        std::vector<BSONObj> chunksToSend;
-        for (const auto chunkVal : chunks.vector()) {
-            ChunkType chunk(*chunkVal);
-
-            if (!chunk.isVersionSet()) {
-                chunk.setVersion(version);
-                version.incMajor();
-            }
-
-            ASSERT(chunk.validate().isOK());
-            chunksToSend.push_back(chunk.toBSON());
-        }
-
-        auto future = launchAsync([this, ns, shardName, metadata] {
-            auto status = loader().makeCollectionMetadata(operationContext(),
-                                                          catalogClient(),
-                                                          ns,
-                                                          shardName,
-                                                          NULL, /* no old metadata */
-                                                          metadata);
-            ASSERT_OK(status);
-        });
-
-        expectFindOnConfigSendBSONObjVector(collToSend);
-        expectFindOnConfigSendBSONObjVector(chunksToSend);
-
-        future.timed_get(kFutureTimeout);
-    }
-
-    ChunkVersion getMaxCollVersion() const {
-        return _maxCollVersion;
-    }
-
-    ChunkVersion getMaxShardVersion() const {
-        return _maxCollVersion;
-    }
-
-private:
-    const HostAndPort configHost{HostAndPort(CONFIG_HOST_PORT)};
-
-    unique_ptr<MetadataLoader> _loader;
-    ChunkVersion _maxCollVersion;
-};
-
-// TODO: Test config server down
-// TODO: Test read of chunks with new epoch
-// TODO: Test that you can properly load config using format with deprecated fields?
-
-TEST_F(MetadataLoaderFixture, DroppedColl) {
-    CollectionType collType;
-    collType.setNs(NamespaceString{"test.foo"});
-    collType.setKeyPattern(BSON("a" << 1));
-    collType.setUpdatedAt(Date_t());
-    collType.setEpoch(OID());
-    collType.setDropped(true);
-    ASSERT_OK(collType.validate());
-    auto future = launchAsync([this] {
-        CollectionMetadata metadata;
-        auto status = MetadataLoader::makeCollectionMetadata(operationContext(),
-                                                             catalogClient(),
-                                                             "test.foo",
-                                                             "shard0000",
-                                                             NULL, /* no old metadata */
-                                                             &metadata);
-        ASSERT_EQUALS(status.code(), ErrorCodes::NamespaceNotFound);
-    });
-    expectFindOnConfigSendBSONObjVector(std::vector<BSONObj>{collType.toBSON()});
-    future.timed_get(kFutureTimeout);
-}
-
-TEST_F(MetadataLoaderFixture, EmptyColl) {
-    auto future = launchAsync([this] {
-        CollectionMetadata metadata;
-        auto status = MetadataLoader::makeCollectionMetadata(operationContext(),
-                                                             catalogClient(),
-                                                             "test.foo",
-                                                             "shard0000",
-                                                             NULL, /* no old metadata */
-                                                             &metadata);
-        ASSERT_EQUALS(status.code(), ErrorCodes::NamespaceNotFound);
-    });
-    expectFindOnConfigSendErrorCode(ErrorCodes::NamespaceNotFound);
-    future.timed_get(kFutureTimeout);
-}
-
-TEST_F(MetadataLoaderFixture, BadColl) {
-    BSONObj badCollToSend = BSON(CollectionType::fullNs("test.foo"));
-    auto future = launchAsync([this] {
-        CollectionMetadata metadata;
-        auto status = MetadataLoader::makeCollectionMetadata(operationContext(),
-                                                             catalogClient(),
-                                                             "test.foo",
-                                                             "shard0000",
-                                                             NULL, /* no old metadata */
-                                                             &metadata);
-        ASSERT_EQUALS(status.code(), ErrorCodes::NoSuchKey);
-    });
-    expectFindOnConfigSendBSONObjVector(std::vector<BSONObj>{badCollToSend});
-    future.timed_get(kFutureTimeout);
-}
-
-TEST_F(MetadataLoaderFixture, BadChunk) {
-    CollectionType collType;
-    collType.setNs(NamespaceString{"test.foo"});
-    collType.setUpdatedAt(Date_t::fromMillisSinceEpoch(1));
-    collType.setKeyPattern(BSON("a" << 1));
-    collType.setEpoch(OID::gen());
-    ASSERT_OK(collType.validate());
-
-    ChunkType chunkInfo;
-    chunkInfo.setNS(NamespaceString{"test.foo"}.ns());
-    chunkInfo.setVersion(ChunkVersion(1, 0, collType.getEpoch()));
-    ASSERT(!chunkInfo.validate().isOK());
-
-    auto future = launchAsync([this] {
-        CollectionMetadata metadata;
-        auto status = MetadataLoader::makeCollectionMetadata(operationContext(),
-                                                             catalogClient(),
-                                                             "test.foo",
-                                                             "shard0000",
-                                                             NULL, /* no old metadata */
-                                                             &metadata);
-        ASSERT_EQUALS(status.code(), ErrorCodes::NoSuchKey);
-    });
-
-    expectFindOnConfigSendBSONObjVector(std::vector<BSONObj>{collType.toBSON()});
-    expectFindOnConfigSendBSONObjVector(std::vector<BSONObj>{chunkInfo.toBSON()});
-    future.timed_get(kFutureTimeout);
-}
-
-TEST_F(MetadataLoaderFixture, NoChunksIsDropped) {
-    OID epoch = OID::gen();
-
-    CollectionType collType;
-    collType.setNs(NamespaceString{"test.foo"});
-    collType.setKeyPattern(BSON("a" << 1));
-    collType.setUnique(false);
-    collType.setUpdatedAt(Date_t::fromMillisSinceEpoch(1));
-    collType.setEpoch(epoch);
-
-    auto future = launchAsync([this] {
-        CollectionMetadata metadata;
-        auto status = MetadataLoader::makeCollectionMetadata(operationContext(),
-                                                             catalogClient(),
-                                                             "test.foo",
-                                                             "shard0000",
-                                                             NULL, /* no old metadata */
-                                                             &metadata);
-        // This is interpreted as a dropped ns, since we drop the chunks first
-        ASSERT_EQUALS(status.code(), ErrorCodes::NamespaceNotFound);
-    });
-    expectFindOnConfigSendBSONObjVector(std::vector<BSONObj>{collType.toBSON()});
-    expectFindOnConfigSendErrorCode(ErrorCodes::NamespaceNotFound);
-    future.timed_get(kFutureTimeout);
-}
-
-TEST_F(MetadataLoaderFixture, CheckNumChunk) {
-    OID epoch = OID::gen();
-
-    CollectionType collType;
-    collType.setNs(NamespaceString{"test.foo"});
-    collType.setKeyPattern(BSON("a" << 1));
-    collType.setUnique(false);
-    collType.setUpdatedAt(Date_t::fromMillisSinceEpoch(1));
-    collType.setEpoch(epoch);
-    ASSERT_OK(collType.validate());
-
-    // Need a chunk on another shard, otherwise the chunks are invalid in general and we
-    // can't load metadata
-    ChunkType chunkType;
-    chunkType.setNS("test.foo");
-    chunkType.setShard(ShardId("shard0001"));
-    chunkType.setMin(BSON("a" << MINKEY));
-    chunkType.setMax(BSON("a" << MAXKEY));
-    chunkType.setVersion(ChunkVersion(1, 0, epoch));
-    ASSERT(chunkType.validate().isOK());
-
-    auto future = launchAsync([this] {
-        CollectionMetadata metadata;
-        auto status = MetadataLoader::makeCollectionMetadata(operationContext(),
-                                                             catalogClient(),
-                                                             "test.foo",
-                                                             "shard0000",
-                                                             NULL, /* no old metadata */
-                                                             &metadata);
-        std::cout << "status: " << status << std::endl;
-        ASSERT_OK(status);
-        ASSERT_EQUALS(0U, metadata.getNumChunks());
-        ASSERT_EQUALS(1, metadata.getCollVersion().majorVersion());
-        ASSERT_EQUALS(0, metadata.getShardVersion().majorVersion());
-        ASSERT_NOT_EQUALS(OID(), metadata.getCollVersion().epoch());
-        ASSERT_NOT_EQUALS(OID(), metadata.getShardVersion().epoch());
-    });
-
-    expectFindOnConfigSendBSONObjVector(std::vector<BSONObj>{collType.toBSON()});
-    expectFindOnConfigSendBSONObjVector(std::vector<BSONObj>{chunkType.toBSON()});
-
-    future.timed_get(kFutureTimeout);
-}
-
-TEST_F(MetadataLoaderFixture, SingleChunkCheckNumChunk) {
-    auto future = launchAsync([this] {
-        CollectionMetadata metadata;
-        auto status = MetadataLoader::makeCollectionMetadata(operationContext(),
-                                                             catalogClient(),
-                                                             "test.foo",
-                                                             "shard0000",
-                                                             NULL, /* no old metadata */
-                                                             &metadata);
-        ASSERT_OK(status);
-        ASSERT_EQUALS(1U, metadata.getNumChunks());
-    });
-
-    expectFindOnConfigSendCollectionDefault();
-    expectFindOnConfigSendChunksDefault();
-
-    future.timed_get(kFutureTimeout);
-}
-
-TEST_F(MetadataLoaderFixture, SingleChunkGetNext) {
-    auto future = launchAsync([this] {
-        CollectionMetadata metadata;
-        auto status = MetadataLoader::makeCollectionMetadata(operationContext(),
-                                                             catalogClient(),
-                                                             "test.foo",
-                                                             "shard0000",
-                                                             NULL, /* no old metadata */
-                                                             &metadata);
-        ChunkType chunkInfo;
-        ASSERT_TRUE(metadata.getNextChunk(metadata.getMinKey(), &chunkInfo));
-    });
-
-    expectFindOnConfigSendCollectionDefault();
-    expectFindOnConfigSendChunksDefault();
-
-    future.timed_get(kFutureTimeout);
-}
-
-TEST_F(MetadataLoaderFixture, SingleChunkGetShardKey) {
-    auto future = launchAsync([this] {
-        CollectionMetadata metadata;
-        auto status = MetadataLoader::makeCollectionMetadata(operationContext(),
-                                                             catalogClient(),
-                                                             "test.foo",
-                                                             "shard0000",
-                                                             NULL, /* no old metadata */
-                                                             &metadata);
-        ChunkType chunkInfo;
-        ASSERT_BSONOBJ_EQ(metadata.getKeyPattern(), BSON("a" << 1));
-    });
-
-    expectFindOnConfigSendCollectionDefault();
-    expectFindOnConfigSendChunksDefault();
-
-    future.timed_get(kFutureTimeout);
-}
-
-TEST_F(MetadataLoaderFixture, SingleChunkGetMaxCollVersion) {
-    auto future = launchAsync([this] {
-        CollectionMetadata metadata;
-        auto status = MetadataLoader::makeCollectionMetadata(operationContext(),
-                                                             catalogClient(),
-                                                             "test.foo",
-                                                             "shard0000",
-                                                             NULL, /* no old metadata */
-                                                             &metadata);
-        ASSERT_TRUE(getMaxCollVersion().equals(metadata.getCollVersion()));
-    });
-
-    expectFindOnConfigSendCollectionDefault();
-    expectFindOnConfigSendChunksDefault();
-
-    future.timed_get(kFutureTimeout);
-}
-TEST_F(MetadataLoaderFixture, SingleChunkGetMaxShardVersion) {
-    auto future = launchAsync([this] {
-        CollectionMetadata metadata;
-        auto status = MetadataLoader::makeCollectionMetadata(operationContext(),
-                                                             catalogClient(),
-                                                             "test.foo",
-                                                             "shard0000",
-                                                             NULL, /* no old metadata */
-                                                             &metadata);
-        ASSERT_TRUE(getMaxShardVersion().equals(metadata.getShardVersion()));
-    });
-    expectFindOnConfigSendCollectionDefault();
-    expectFindOnConfigSendChunksDefault();
-
-    future.timed_get(kFutureTimeout);
-}
-
-TEST_F(MetadataLoaderFixture, NoChunks) {
-    auto future = launchAsync([this] {
-        CollectionMetadata metadata;
-        auto status = MetadataLoader::makeCollectionMetadata(operationContext(),
-                                                             catalogClient(),
-                                                             "test.foo",
-                                                             "shard0000",
-                                                             NULL, /* no old metadata */
-                                                             &metadata);
-        // NSNotFound because we're reloading with no old metadata
-        ASSERT_EQUALS(status.code(), ErrorCodes::NamespaceNotFound);
-    });
-    expectFindOnConfigSendCollectionDefault();
-    expectFindOnConfigSendErrorCode(ErrorCodes::NamespaceNotFound);
-
-    future.timed_get(kFutureTimeout);
-}
-
-}  // namespace
-}  // namespace mongo
diff --git a/src/mongo/db/s/metadata_manager.cpp b/src/mongo/db/s/metadata_manager.cpp
index 77c83bb3626..faa062e9476 100644
--- a/src/mongo/db/s/metadata_manager.cpp
+++ b/src/mongo/db/s/metadata_manager.cpp
@@ -257,8 +257,6 @@ RangeMap MetadataManager::getCopyOfReceivingChunks() {
 }
 
 void MetadataManager::_setActiveMetadata_inlock(std::unique_ptr<CollectionMetadata> newMetadata) {
-    invariant(!newMetadata || newMetadata->isValid());
-
     if (_activeMetadataTracker->usageCounter > 0) {
         _metadataInUse.push_front(std::move(_activeMetadataTracker));
     }
@@ -300,11 +298,11 @@ ScopedCollectionMetadata::~ScopedCollectionMetadata() {
     _decrementUsageCounter();
 }
 
-CollectionMetadata* ScopedCollectionMetadata::operator->() {
+CollectionMetadata* ScopedCollectionMetadata::operator->() const {
     return _tracker->metadata.get();
 }
 
-CollectionMetadata* ScopedCollectionMetadata::getMetadata() {
+CollectionMetadata* ScopedCollectionMetadata::getMetadata() const {
     return _tracker->metadata.get();
 }
 
diff --git a/src/mongo/db/s/metadata_manager.h b/src/mongo/db/s/metadata_manager.h
index 44ab0dbbf71..5c2e7f2e64a 100644
--- a/src/mongo/db/s/metadata_manager.h
+++ b/src/mongo/db/s/metadata_manager.h
@@ -243,8 +243,8 @@ public:
     /**
      * Dereferencing the ScopedCollectionMetadata will dereference the internal CollectionMetadata.
      */
-    CollectionMetadata* operator->();
-    CollectionMetadata* getMetadata();
+    CollectionMetadata* operator->() const;
+    CollectionMetadata* getMetadata() const;
 
     /**
      * True if the ScopedCollectionMetadata stores a metadata (is not empty)
diff --git a/src/mongo/db/s/metadata_manager_test.cpp b/src/mongo/db/s/metadata_manager_test.cpp
index 08f26ac3298..242ef274be1 100644
--- a/src/mongo/db/s/metadata_manager_test.cpp
+++ b/src/mongo/db/s/metadata_manager_test.cpp
@@ -42,14 +42,12 @@
 #include "mongo/stdx/memory.h"
 #include "mongo/unittest/unittest.h"
 #include "mongo/util/assert_util.h"
-#include "mongo/util/scopeguard.h"
 
 namespace mongo {
+namespace {
 
 using unittest::assertGet;
 
-namespace {
-
 class MetadataManagerTest : public ServiceContextMongoDTest {
 protected:
     void setUp() override {
@@ -59,11 +57,42 @@ protected:
     }
 
     static std::unique_ptr<CollectionMetadata> makeEmptyMetadata() {
-        return stdx::make_unique<CollectionMetadata>(BSON("key" << 1),
-                                                     ChunkVersion(1, 0, OID::gen()));
+        const OID epoch = OID::gen();
+
+        return stdx::make_unique<CollectionMetadata>(
+            BSON("key" << 1),
+            ChunkVersion(1, 0, epoch),
+            ChunkVersion(0, 0, epoch),
+            SimpleBSONObjComparator::kInstance.makeBSONObjIndexedMap<CachedChunkInfo>());
     }
-};
 
+    /**
+     * Returns a new metadata's instance based on the current state by adding a chunk with the
+     * specified bounds and version. The chunk's version must be higher than that of all chunks
+     * which are in the input metadata.
+     *
+     * It will fassert if the chunk bounds are incorrect or overlap an existing chunk or if the
+     * chunk version is lower than the maximum one.
+     */
+    static std::unique_ptr<CollectionMetadata> cloneMetadataPlusChunk(
+        const CollectionMetadata& metadata,
+        const BSONObj& minKey,
+        const BSONObj& maxKey,
+        const ChunkVersion& chunkVersion) {
+        invariant(chunkVersion.epoch() == metadata.getShardVersion().epoch());
+        invariant(chunkVersion.isSet());
+        invariant(chunkVersion > metadata.getCollVersion());
+        invariant(minKey.woCompare(maxKey) < 0);
+        invariant(!rangeMapOverlaps(metadata.getChunks(), minKey, maxKey));
+
+        auto chunksMap = metadata.getChunks();
+        chunksMap.insert(
+            std::make_pair(minKey.getOwned(), CachedChunkInfo(maxKey.getOwned(), chunkVersion)));
+
+        return stdx::make_unique<CollectionMetadata>(
+            metadata.getKeyPattern(), chunkVersion, chunkVersion, std::move(chunksMap));
+    }
+};
 
 TEST_F(MetadataManagerTest, SetAndGetActiveMetadata) {
     MetadataManager manager(getServiceContext(), NamespaceString("TestDb", "CollDB"));
@@ -85,8 +114,8 @@ TEST_F(MetadataManagerTest, ResetActiveMetadata) {
 
     ChunkVersion newVersion = scopedMetadata1->getCollVersion();
     newVersion.incMajor();
-    std::unique_ptr<CollectionMetadata> cm2 =
-        scopedMetadata1->clonePlusChunk(BSON("key" << 0), BSON("key" << 10), newVersion);
+    std::unique_ptr<CollectionMetadata> cm2 = cloneMetadataPlusChunk(
+        *scopedMetadata1.getMetadata(), BSON("key" << 0), BSON("key" << 10), newVersion);
     auto cm2Ptr = cm2.get();
 
     manager.refreshActiveMetadata(std::move(cm2));
@@ -274,8 +303,8 @@ TEST_F(MetadataManagerTest, RefreshAfterSuccessfulMigrationSinglePending) {
     ChunkVersion version = manager.getActiveMetadata()->getCollVersion();
     version.incMajor();
 
-    manager.refreshActiveMetadata(
-        manager.getActiveMetadata()->clonePlusChunk(cr1.getMin(), cr1.getMax(), version));
+    manager.refreshActiveMetadata(cloneMetadataPlusChunk(
+        *manager.getActiveMetadata().getMetadata(), cr1.getMin(), cr1.getMax(), version));
     ASSERT_EQ(manager.getCopyOfReceivingChunks().size(), 0UL);
     ASSERT_EQ(manager.getActiveMetadata()->getChunks().size(), 1UL);
 }
@@ -297,8 +326,8 @@ TEST_F(MetadataManagerTest, RefreshAfterSuccessfulMigrationMultiplePending) {
         ChunkVersion version = manager.getActiveMetadata()->getCollVersion();
         version.incMajor();
 
-        manager.refreshActiveMetadata(
-            manager.getActiveMetadata()->clonePlusChunk(cr1.getMin(), cr1.getMax(), version));
+        manager.refreshActiveMetadata(cloneMetadataPlusChunk(
+            *manager.getActiveMetadata().getMetadata(), cr1.getMin(), cr1.getMax(), version));
         ASSERT_EQ(manager.getCopyOfReceivingChunks().size(), 1UL);
         ASSERT_EQ(manager.getActiveMetadata()->getChunks().size(), 1UL);
     }
@@ -307,8 +336,8 @@ TEST_F(MetadataManagerTest, RefreshAfterSuccessfulMigrationMultiplePending) {
         ChunkVersion version = manager.getActiveMetadata()->getCollVersion();
         version.incMajor();
 
-        manager.refreshActiveMetadata(
-            manager.getActiveMetadata()->clonePlusChunk(cr2.getMin(), cr2.getMax(), version));
+        manager.refreshActiveMetadata(cloneMetadataPlusChunk(
+            *manager.getActiveMetadata().getMetadata(), cr2.getMin(), cr2.getMax(), version));
         ASSERT_EQ(manager.getCopyOfReceivingChunks().size(), 0UL);
         ASSERT_EQ(manager.getActiveMetadata()->getChunks().size(), 2UL);
     }
@@ -330,8 +359,8 @@ TEST_F(MetadataManagerTest, RefreshAfterNotYetCompletedMigrationMultiplePending)
     ChunkVersion version = manager.getActiveMetadata()->getCollVersion();
     version.incMajor();
 
-    manager.refreshActiveMetadata(
-        manager.getActiveMetadata()->clonePlusChunk(BSON("key" << 50), BSON("key" << 60), version));
+    manager.refreshActiveMetadata(cloneMetadataPlusChunk(
+        *manager.getActiveMetadata().getMetadata(), BSON("key" << 50), BSON("key" << 60), version));
     ASSERT_EQ(manager.getCopyOfReceivingChunks().size(), 2UL);
     ASSERT_EQ(manager.getActiveMetadata()->getChunks().size(), 1UL);
 }
@@ -368,8 +397,8 @@ TEST_F(MetadataManagerTest, RefreshMetadataAfterDropAndRecreate) {
         ChunkVersion newVersion = metadata->getCollVersion();
         newVersion.incMajor();
 
-        manager.refreshActiveMetadata(
-            metadata->clonePlusChunk(BSON("key" << 0), BSON("key" << 10), newVersion));
+        manager.refreshActiveMetadata(cloneMetadataPlusChunk(
+            *metadata.getMetadata(), BSON("key" << 0), BSON("key" << 10), newVersion));
     }
 
     // Now, pretend that the collection was dropped and recreated
@@ -377,8 +406,8 @@ TEST_F(MetadataManagerTest, RefreshMetadataAfterDropAndRecreate) {
     ChunkVersion newVersion = recreateMetadata->getCollVersion();
     newVersion.incMajor();
 
-    manager.refreshActiveMetadata(
-        recreateMetadata->clonePlusChunk(BSON("key" << 20), BSON("key" << 30), newVersion));
+    manager.refreshActiveMetadata(cloneMetadataPlusChunk(
+        *recreateMetadata, BSON("key" << 20), BSON("key" << 30), newVersion));
     ASSERT_EQ(manager.getActiveMetadata()->getChunks().size(), 1UL);
 
     const auto chunkEntry = manager.getActiveMetadata()->getChunks().begin();
diff --git a/src/mongo/db/s/migration_source_manager.cpp b/src/mongo/db/s/migration_source_manager.cpp
index f5ab20c0e37..0ad841ea56f 100644
--- a/src/mongo/db/s/migration_source_manager.cpp
+++ b/src/mongo/db/s/migration_source_manager.cpp
@@ -145,9 +145,6 @@ MigrationSourceManager::MigrationSourceManager(OperationContext* txn,
     // With nonzero shard version, we must have a coll version >= our shard version
     invariant(collectionVersion >= shardVersion);
 
-    // With nonzero shard version, we must have a shard key
-    invariant(!_collectionMetadata->getKeyPattern().isEmpty());
-
     ChunkType chunkToMove;
     chunkToMove.setMin(_args.getMinKey());
     chunkToMove.setMax(_args.getMaxKey());
diff --git a/src/mongo/db/s/sharding_state.cpp b/src/mongo/db/s/sharding_state.cpp
index 998b6c368ce..a0f0c5b1a4f 100644
--- a/src/mongo/db/s/sharding_state.cpp
+++ b/src/mongo/db/s/sharding_state.cpp
@@ -46,7 +46,6 @@
 #include "mongo/db/repl/replication_coordinator_global.h"
 #include "mongo/db/s/collection_metadata.h"
 #include "mongo/db/s/collection_sharding_state.h"
-#include "mongo/db/s/metadata_loader.h"
 #include "mongo/db/s/operation_sharding_state.h"
 #include "mongo/db/s/sharded_connection_info.h"
 #include "mongo/db/s/sharding_initialization_mongod.h"
@@ -58,6 +57,7 @@
 #include "mongo/rpc/metadata/metadata_hook.h"
 #include "mongo/s/catalog/sharding_catalog_client.h"
 #include "mongo/s/catalog/type_chunk.h"
+#include "mongo/s/catalog_cache.h"
 #include "mongo/s/chunk_version.h"
 #include "mongo/s/client/shard_registry.h"
 #include "mongo/s/client/sharding_network_connection_hook.h"
@@ -83,12 +83,6 @@ namespace {
 
 const auto getShardingState = ServiceContext::declareDecoration<ShardingState>();
 
-// Max number of concurrent config server refresh threads
-const int kMaxConfigServerRefreshThreads = 3;
-
-// Maximum number of times to try to refresh the collection metadata if conflicts are occurring
-const int kMaxNumMetadataRefreshAttempts = 3;
-
 /**
  * Updates the config server field of the shardIdentity document with the given connection string
  * if setName is equal to the config server replica set name.
@@ -123,7 +117,6 @@ const std::set<std::string> ShardingState::_commandsThatInitializeShardingAwaren
 ShardingState::ShardingState()
     : _initializationState(static_cast<uint32_t>(InitializationState::kNew)),
       _initializationStatus(Status(ErrorCodes::InternalError, "Uninitialized value")),
-      _configServerTickets(kMaxConfigServerRefreshThreads),
       _globalInit(&initializeGlobalShardingStateForMongod),
       _scheduleWorkFn([](NamespaceString nss) {}) {}
 
@@ -272,30 +265,24 @@ Status ShardingState::onStaleShardVersion(OperationContext* txn,
         }
     }
 
-    auto refreshStatusAndVersion =
-        _refreshMetadata(txn, nss, (currentMetadata ? currentMetadata.getMetadata() : nullptr));
-    return refreshStatusAndVersion.getStatus();
+    try {
+        _refreshMetadata(txn, nss);
+        return Status::OK();
+    } catch (const DBException& ex) {
+        log() << "Failed to refresh metadata for collection" << nss << causedBy(redact(ex));
+        return ex.toStatus();
+    }
 }
 
 Status ShardingState::refreshMetadataNow(OperationContext* txn,
                                          const NamespaceString& nss,
                                          ChunkVersion* latestShardVersion) {
-    ScopedCollectionMetadata currentMetadata;
-
-    {
-        AutoGetCollection autoColl(txn, nss, MODE_IS);
-
-        currentMetadata = CollectionShardingState::get(txn, nss)->getMetadata();
-    }
-
-    auto refreshLatestShardVersionStatus =
-        _refreshMetadata(txn, nss, currentMetadata.getMetadata());
-    if (!refreshLatestShardVersionStatus.isOK()) {
-        return refreshLatestShardVersionStatus.getStatus();
+    try {
+        *latestShardVersion = _refreshMetadata(txn, nss);
+        return Status::OK();
+    } catch (const DBException& ex) {
+        return ex.toStatus();
     }
-
-    *latestShardVersion = refreshLatestShardVersionStatus.getValue();
-    return Status::OK();
 }
 
 void ShardingState::initializeFromConfigConnString(OperationContext* txn,
@@ -628,87 +615,58 @@ StatusWith<bool> ShardingState::initializeShardingAwarenessIfNeeded(OperationCon
     }
 }
 
-StatusWith<ChunkVersion> ShardingState::_refreshMetadata(
-    OperationContext* txn, const NamespaceString& nss, const CollectionMetadata* metadataForDiff) {
+ChunkVersion ShardingState::_refreshMetadata(OperationContext* txn, const NamespaceString& nss) {
     invariant(!txn->lockState()->isLocked());
+    invariant(enabled());
 
-    {
-        Status status = _waitForInitialization(txn->getDeadline());
-        if (!status.isOK())
-            return status;
-    }
-
-    // We can't reload if a shard name has not yet been set
-    {
-        stdx::lock_guard<stdx::mutex> lk(_mutex);
-
-        if (_shardName.empty()) {
-            string errMsg = str::stream() << "cannot refresh metadata for " << nss.ns()
-                                          << " before shard name has been set";
-            warning() << errMsg;
-            return {ErrorCodes::NotYetInitialized, errMsg};
-        }
-    }
+    const ShardId shardId = getShardName();
 
-    Status status = {ErrorCodes::InternalError, "metadata refresh not performed"};
-    Timer t;
-    int numAttempts = 0;
-    std::unique_ptr<CollectionMetadata> remoteMetadata;
+    uassert(ErrorCodes::NotYetInitialized,
+            str::stream() << "Cannot refresh metadata for " << nss.ns()
+                          << " before shard name has been set",
+            shardId.isValid());
 
-    do {
-        // The _configServerTickets serializes this process such that only a small number of threads
-        // can try to refresh at the same time in order to avoid overloading the config server.
-        _configServerTickets.waitForTicket();
-        TicketHolderReleaser needTicketFrom(&_configServerTickets);
+    auto newCollectionMetadata = [&]() -> std::unique_ptr<CollectionMetadata> {
+        auto const catalogCache = Grid::get(txn)->catalogCache();
+        catalogCache->invalidateShardedCollection(nss);
 
-        if (status == ErrorCodes::RemoteChangeDetected) {
-            metadataForDiff = nullptr;
-            log() << "Refresh failed and will be retried as full reload " << status;
+        const auto routingInfo = uassertStatusOK(catalogCache->getCollectionRoutingInfo(txn, nss));
+        const auto cm = routingInfo.cm();
+        if (!cm) {
+            return nullptr;
         }
 
-        log() << "MetadataLoader loading chunks for " << nss.ns() << " based on: "
-              << (metadataForDiff ? metadataForDiff->getCollVersion().toString() : "(empty)");
+        RangeMap shardChunksMap =
+            SimpleBSONObjComparator::kInstance.makeBSONObjIndexedMap<CachedChunkInfo>();
 
-        remoteMetadata = stdx::make_unique<CollectionMetadata>();
-        status = MetadataLoader::makeCollectionMetadata(txn,
-                                                        grid.catalogClient(txn),
-                                                        nss.ns(),
-                                                        getShardName(),
-                                                        metadataForDiff,
-                                                        remoteMetadata.get());
-    } while (status == ErrorCodes::RemoteChangeDetected &&
-             ++numAttempts < kMaxNumMetadataRefreshAttempts);
+        for (const auto& chunkMapEntry : cm->chunkMap()) {
+            const auto& chunk = chunkMapEntry.second;
 
-    if (!status.isOK() && status != ErrorCodes::NamespaceNotFound) {
-        warning() << "MetadataLoader failed after " << t.millis() << " ms"
-                  << causedBy(redact(status));
+            if (chunk->getShardId() != shardId)
+                continue;
 
-        return status;
-    }
+            shardChunksMap.emplace(chunk->getMin(),
+                                   CachedChunkInfo(chunk->getMax(), chunk->getLastmod()));
+        }
+
+        return stdx::make_unique<CollectionMetadata>(cm->getShardKeyPattern().toBSON(),
+                                                     cm->getVersion(),
+                                                     cm->getVersion(shardId),
+                                                     std::move(shardChunksMap));
+    }();
 
     // Exclusive collection lock needed since we're now changing the metadata
     ScopedTransaction transaction(txn, MODE_IX);
     AutoGetCollection autoColl(txn, nss, MODE_IX, MODE_X);
 
     auto css = CollectionShardingState::get(txn, nss);
+    css->refreshMetadata(txn, std::move(newCollectionMetadata));
 
-    if (!status.isOK()) {
-        invariant(status == ErrorCodes::NamespaceNotFound);
-        css->refreshMetadata(txn, nullptr);
-
-        log() << "MetadataLoader took " << t.millis() << " ms and did not find the namespace";
-
+    if (!css->getMetadata()) {
         return ChunkVersion::UNSHARDED();
     }
 
-    css->refreshMetadata(txn, std::move(remoteMetadata));
-
-    auto metadata = css->getMetadata();
-
-    log() << "MetadataLoader took " << t.millis() << " ms and found version "
-          << metadata->getCollVersion();
-
-    return metadata->getShardVersion();
+    return css->getMetadata()->getShardVersion();
 }
 
 StatusWith<ScopedRegisterDonateChunk> ShardingState::registerDonateChunk(
diff --git a/src/mongo/db/s/sharding_state.h b/src/mongo/db/s/sharding_state.h
index 8481c282bed..e53fee63515 100644
--- a/src/mongo/db/s/sharding_state.h
+++ b/src/mongo/db/s/sharding_state.h
@@ -43,8 +43,6 @@
 #include "mongo/stdx/memory.h"
 #include "mongo/stdx/mutex.h"
 #include "mongo/stdx/unordered_map.h"
-#include "mongo/util/concurrency/ticketholder.h"
-#include "mongo/util/time_support.h"
 
 namespace mongo {
 
@@ -347,13 +345,8 @@ private:
      * Refreshes collection metadata by asking the config server for the latest information and
      * returns the latest version at the time the reload was done. This call does network I/O and
      * should never be called with a lock.
-     *
-     * The metadataForDiff argument indicates that the specified metadata should be used as a base
-     * from which to only load the differences. If nullptr is passed, a full reload will be done.
      */
-    StatusWith<ChunkVersion> _refreshMetadata(OperationContext* txn,
-                                              const NamespaceString& nss,
-                                              const CollectionMetadata* metadataForDiff);
+    ChunkVersion _refreshMetadata(OperationContext* opCtx, const NamespaceString& nss);
 
     // Initializes a TaskExecutor for cleaning up orphaned ranges
     void _initializeRangeDeleterTaskExecutor();
@@ -379,9 +372,6 @@ private:
     // Sets the shard name for this host (comes through setShardVersion)
     std::string _shardName;
 
-    // Protects from hitting the config server from too many threads at once
-    TicketHolder _configServerTickets;
-
     // Cache of collection metadata on this shard. It is not safe to look-up values from this map
     // without holding some form of collection lock. It is only safe to add/remove values when
     // holding X lock on the respective namespace.
diff --git a/src/mongo/db/s/sharding_state_test.cpp b/src/mongo/db/s/sharding_state_test.cpp
index 173964edaea..7d454a4818f 100644
--- a/src/mongo/db/s/sharding_state_test.cpp
+++ b/src/mongo/db/s/sharding_state_test.cpp
@@ -34,7 +34,6 @@
 #include "mongo/db/dbdirectclient.h"
 #include "mongo/db/jsobj.h"
 #include "mongo/db/namespace_string.h"
-#include "mongo/db/query/query_request.h"
 #include "mongo/db/repl/replication_coordinator_mock.h"
 #include "mongo/db/s/sharding_state.h"
 #include "mongo/db/s/type_shard_identity.h"
@@ -43,8 +42,6 @@
 #include "mongo/db/storage/storage_options.h"
 #include "mongo/s/catalog/dist_lock_manager_mock.h"
 #include "mongo/s/catalog/sharding_catalog_client_impl.h"
-#include "mongo/s/catalog/type_chunk.h"
-#include "mongo/s/catalog/type_collection.h"
 #include "mongo/s/client/shard_registry.h"
 #include "mongo/s/sharding_mongod_test_fixture.h"
 
@@ -64,31 +61,6 @@ public:
         return _shardName.toString();
     }
 
-    void setupCollectionMetadata(const NamespaceString& nss,
-                                 const OID& epoch,
-                                 const std::vector<BSONObj>& initChunks) {
-        auto future = launchAsync([this, &nss] {
-            ChunkVersion latestShardVersion;
-            Client::initThreadIfNotAlready();
-            ASSERT_OK(
-                shardingState()->refreshMetadataNow(operationContext(), nss, &latestShardVersion));
-        });
-
-        ChunkVersion initVersion(1, 0, epoch);
-        onFindCommand([&nss, &initVersion](const RemoteCommandRequest&) {
-            CollectionType coll;
-            coll.setNs(nss);
-            coll.setUpdatedAt(Date_t());
-            coll.setEpoch(initVersion.epoch());
-            coll.setKeyPattern(BSON("x" << 1));
-            return std::vector<BSONObj>{coll.toBSON()};
-        });
-
-        onFindCommand([&initChunks](const RemoteCommandRequest&) { return initChunks; });
-
-        future.timed_get(kFutureTimeout);
-    }
-
 protected:
     // Used to write to set up local collections before exercising server logic.
     std::unique_ptr<DBDirectClient> _dbDirectClient;
@@ -610,293 +582,5 @@ TEST_F(ShardingStateTest,
     ASSERT_FALSE(swShardingInitialized.getValue());
 }
 
-TEST_F(ShardingStateTest, MetadataRefreshShouldUseDiffQuery) {
-    ShardIdentityType shardIdentity;
-    shardIdentity.setConfigsvrConnString(
-        ConnectionString(ConnectionString::SET, "a:1,b:2", "config"));
-    shardIdentity.setShardName(shardName());
-    shardIdentity.setClusterId(OID::gen());
-
-    ASSERT_OK(shardingState()->initializeFromShardIdentity(operationContext(), shardIdentity));
-
-    const NamespaceString nss("test.user");
-    const OID initEpoch(OID::gen());
-
-    {
-        ChunkType chunk;
-        chunk.setNS(nss.ns());
-        chunk.setMin(BSON("x" << 0));
-        chunk.setMax(BSON("x" << 10));
-        chunk.setShard(ShardId(shardName()));
-        chunk.setVersion(ChunkVersion(2, 0, initEpoch));
-        setupCollectionMetadata(nss, initEpoch, std::vector<BSONObj>{chunk.toBSON()});
-    }
-
-    const ChunkVersion newVersion(3, 0, initEpoch);
-    auto future = launchAsync([&] {
-        Client::initThreadIfNotAlready();
-        ASSERT_OK(shardingState()->onStaleShardVersion(operationContext(), nss, newVersion));
-    });
-
-    onFindCommand([&nss, &initEpoch](const RemoteCommandRequest&) {
-        CollectionType coll;
-        coll.setNs(nss);
-        coll.setUpdatedAt(Date_t());
-        coll.setEpoch(initEpoch);
-        coll.setKeyPattern(BSON("x" << 1));
-        return std::vector<BSONObj>{coll.toBSON()};
-    });
-
-    onFindCommand([this, &nss, &initEpoch](const RemoteCommandRequest& request) {
-        auto diffQueryStatus = QueryRequest::makeFromFindCommand(nss, request.cmdObj, false);
-        ASSERT_OK(diffQueryStatus.getStatus());
-
-        auto diffQuery = std::move(diffQueryStatus.getValue());
-        ASSERT_BSONOBJ_EQ(BSON("ns" << nss.ns() << "lastmod" << BSON("$gte" << Timestamp(2, 0))),
-                          diffQuery->getFilter());
-
-        ChunkType chunk;
-        chunk.setNS(nss.ns());
-        chunk.setMin(BSON("x" << 10));
-        chunk.setMax(BSON("x" << 20));
-        chunk.setShard(ShardId(shardName()));
-        chunk.setVersion(ChunkVersion(3, 10, initEpoch));
-        return std::vector<BSONObj>{chunk.toBSON()};
-    });
-
-    future.timed_get(kFutureTimeout);
-}
-
-/**
- * Test where the epoch changed right before the chunk diff query.
- */
-TEST_F(ShardingStateTest, MetadataRefreshShouldUseFullQueryOnEpochMismatch) {
-    ShardIdentityType shardIdentity;
-    shardIdentity.setConfigsvrConnString(
-        ConnectionString(ConnectionString::SET, "a:1,b:2", "config"));
-    shardIdentity.setShardName(shardName());
-    shardIdentity.setClusterId(OID::gen());
-
-    ASSERT_OK(shardingState()->initializeFromShardIdentity(operationContext(), shardIdentity));
-
-    const NamespaceString nss("test.user");
-    const OID initEpoch(OID::gen());
-
-    {
-        ChunkType chunk;
-        chunk.setNS(nss.ns());
-        chunk.setMin(BSON("x" << 0));
-        chunk.setMax(BSON("x" << 10));
-        chunk.setShard(ShardId(shardName()));
-        chunk.setVersion(ChunkVersion(2, 0, initEpoch));
-        setupCollectionMetadata(nss, initEpoch, std::vector<BSONObj>{chunk.toBSON()});
-    }
-
-
-    auto future = launchAsync([&] {
-        Client::initThreadIfNotAlready();
-        ASSERT_OK(shardingState()->onStaleShardVersion(
-            operationContext(), nss, ChunkVersion(3, 0, initEpoch)));
-    });
-
-    onFindCommand([&nss, &initEpoch](const RemoteCommandRequest&) {
-        CollectionType coll;
-        coll.setNs(nss);
-        coll.setUpdatedAt(Date_t());
-        coll.setEpoch(initEpoch);
-        coll.setKeyPattern(BSON("x" << 1));
-        return std::vector<BSONObj>{coll.toBSON()};
-    });
-
-    // Now when the diff query is performed, it will get chunks with a different epoch.
-    const ChunkVersion newVersion(3, 0, OID::gen());
-    onFindCommand([this, &nss, &newVersion](const RemoteCommandRequest& request) {
-        auto diffQueryStatus = QueryRequest::makeFromFindCommand(nss, request.cmdObj, false);
-        ASSERT_OK(diffQueryStatus.getStatus());
-
-        auto diffQuery = std::move(diffQueryStatus.getValue());
-        ASSERT_BSONOBJ_EQ(BSON("ns" << nss.ns() << "lastmod" << BSON("$gte" << Timestamp(2, 0))),
-                          diffQuery->getFilter());
-
-        ChunkType chunk;
-        chunk.setNS(nss.ns());
-        chunk.setMin(BSON("x" << 10));
-        chunk.setMax(BSON("x" << 20));
-        chunk.setShard(ShardId(shardName()));
-        chunk.setVersion(ChunkVersion(3, 10, newVersion.epoch()));
-        return std::vector<BSONObj>{chunk.toBSON()};
-    });
-
-    // Retry the refresh again. Now doing a full reload.
-
-    onFindCommand([&nss, &newVersion](const RemoteCommandRequest&) {
-        CollectionType coll;
-        coll.setNs(nss);
-        coll.setUpdatedAt(Date_t());
-        coll.setEpoch(newVersion.epoch());
-        coll.setKeyPattern(BSON("x" << 1));
-        return std::vector<BSONObj>{coll.toBSON()};
-    });
-
-    onFindCommand([this, &nss, &newVersion](const RemoteCommandRequest& request) {
-        auto diffQueryStatus = QueryRequest::makeFromFindCommand(nss, request.cmdObj, false);
-        ASSERT_OK(diffQueryStatus.getStatus());
-
-        auto diffQuery = std::move(diffQueryStatus.getValue());
-        ASSERT_BSONOBJ_EQ(BSON("ns" << nss.ns() << "lastmod" << BSON("$gte" << Timestamp(0, 0))),
-                          diffQuery->getFilter());
-
-        ChunkType chunk;
-        chunk.setNS(nss.ns());
-        chunk.setMin(BSON("x" << 10));
-        chunk.setMax(BSON("x" << 20));
-        chunk.setShard(ShardId(shardName()));
-        chunk.setVersion(ChunkVersion(3, 10, newVersion.epoch()));
-        return std::vector<BSONObj>{chunk.toBSON()};
-    });
-
-    future.timed_get(kFutureTimeout);
-}
-
-TEST_F(ShardingStateTest, FullMetadataOnEpochMismatchShouldStopAfterMaxRetries) {
-    ShardIdentityType shardIdentity;
-    shardIdentity.setConfigsvrConnString(
-        ConnectionString(ConnectionString::SET, "a:1,b:2", "config"));
-    shardIdentity.setShardName(shardName());
-    shardIdentity.setClusterId(OID::gen());
-
-    ASSERT_OK(shardingState()->initializeFromShardIdentity(operationContext(), shardIdentity));
-
-    const NamespaceString nss("test.user");
-    const OID initEpoch(OID::gen());
-
-    {
-        ChunkType chunk;
-        chunk.setNS(nss.ns());
-        chunk.setMin(BSON("x" << 0));
-        chunk.setMax(BSON("x" << 10));
-        chunk.setShard(ShardId(shardName()));
-        chunk.setVersion(ChunkVersion(2, 0, initEpoch));
-        setupCollectionMetadata(nss, initEpoch, std::vector<BSONObj>{chunk.toBSON()});
-    }
-
-
-    auto future = launchAsync([&] {
-        Client::initThreadIfNotAlready();
-        auto status = shardingState()->onStaleShardVersion(
-            operationContext(), nss, ChunkVersion(3, 0, initEpoch));
-        ASSERT_EQ(ErrorCodes::RemoteChangeDetected, status);
-    });
-
-    OID lastEpoch(initEpoch);
-    OID nextEpoch(OID::gen());
-    for (int tries = 0; tries < 3; tries++) {
-        onFindCommand([&nss, &lastEpoch](const RemoteCommandRequest&) {
-            CollectionType coll;
-            coll.setNs(nss);
-            coll.setUpdatedAt(Date_t());
-            coll.setEpoch(lastEpoch);
-            coll.setKeyPattern(BSON("x" << 1));
-            return std::vector<BSONObj>{coll.toBSON()};
-        });
-
-        onFindCommand([this, &nss, &nextEpoch, tries](const RemoteCommandRequest& request) {
-            auto diffQueryStatus = QueryRequest::makeFromFindCommand(nss, request.cmdObj, false);
-            ASSERT_OK(diffQueryStatus.getStatus());
-
-            auto diffQuery = std::move(diffQueryStatus.getValue());
-            Timestamp expectedLastMod = (tries == 0) ? Timestamp(2, 0) : Timestamp(0, 0);
-            ASSERT_BSONOBJ_EQ(
-                BSON("ns" << nss.ns() << "lastmod" << BSON("$gte" << expectedLastMod)),
-                diffQuery->getFilter());
-
-            ChunkType chunk;
-            chunk.setNS(nss.ns());
-            chunk.setMin(BSON("x" << 10));
-            chunk.setMax(BSON("x" << 20));
-            chunk.setShard(ShardId(shardName()));
-            chunk.setVersion(ChunkVersion(3, 10, nextEpoch));
-            return std::vector<BSONObj>{chunk.toBSON()};
-        });
-
-        lastEpoch = nextEpoch;
-        nextEpoch = OID::gen();
-    }
-
-    future.timed_get(kFutureTimeout);
-}
-
-TEST_F(ShardingStateTest, MetadataRefreshShouldBeOkWhenCollectionWasDropped) {
-    ShardIdentityType shardIdentity;
-    shardIdentity.setConfigsvrConnString(
-        ConnectionString(ConnectionString::SET, "a:1,b:2", "config"));
-    shardIdentity.setShardName(shardName());
-    shardIdentity.setClusterId(OID::gen());
-
-    ASSERT_OK(shardingState()->initializeFromShardIdentity(operationContext(), shardIdentity));
-
-    const NamespaceString nss("test.user");
-    const OID initEpoch(OID::gen());
-
-    {
-        ChunkType chunk;
-        chunk.setNS(nss.ns());
-        chunk.setMin(BSON("x" << 0));
-        chunk.setMax(BSON("x" << 10));
-        chunk.setShard(ShardId(shardName()));
-        chunk.setVersion(ChunkVersion(2, 0, initEpoch));
-        setupCollectionMetadata(nss, initEpoch, std::vector<BSONObj>{chunk.toBSON()});
-    }
-
-    const ChunkVersion newVersion(3, 0, initEpoch);
-    auto future = launchAsync([&] {
-        Client::initThreadIfNotAlready();
-        ASSERT_OK(shardingState()->onStaleShardVersion(operationContext(), nss, newVersion));
-    });
-
-    onFindCommand([&nss, &initEpoch](const RemoteCommandRequest&) {
-        CollectionType coll;
-        coll.setNs(nss);
-        coll.setUpdatedAt(Date_t());
-        coll.setEpoch(initEpoch);
-        coll.setKeyPattern(BSON("x" << 1));
-        coll.setDropped(true);
-        return std::vector<BSONObj>{coll.toBSON()};
-    });
-
-    future.timed_get(kFutureTimeout);
-}
-
-TEST_F(ShardingStateTest, MetadataRefreshShouldNotRetryOtherTypesOfError) {
-    ShardIdentityType shardIdentity;
-    shardIdentity.setConfigsvrConnString(
-        ConnectionString(ConnectionString::SET, "a:1,b:2", "config"));
-    shardIdentity.setShardName(shardName());
-    shardIdentity.setClusterId(OID::gen());
-
-    ASSERT_OK(shardingState()->initializeFromShardIdentity(operationContext(), shardIdentity));
-
-    const NamespaceString nss("test.user");
-    const OID initEpoch(OID::gen());
-
-    {
-        ChunkType chunk;
-        chunk.setNS(nss.ns());
-        chunk.setMin(BSON("x" << 0));
-        chunk.setMax(BSON("x" << 10));
-        chunk.setShard(ShardId(shardName()));
-        chunk.setVersion(ChunkVersion(2, 0, initEpoch));
-        setupCollectionMetadata(nss, initEpoch, std::vector<BSONObj>{chunk.toBSON()});
-    }
-
-    auto configTargeter =
-        RemoteCommandTargeterMock::get(shardRegistry()->getConfigShard()->getTargeter());
-    configTargeter->setFindHostReturnValue({ErrorCodes::HostNotFound, "host erased by test"});
-
-    auto status = shardingState()->onStaleShardVersion(
-        operationContext(), nss, ChunkVersion(3, 0, initEpoch));
-    ASSERT_EQ(ErrorCodes::HostNotFound, status);
-}
-
 }  // namespace
 }  // namespace mongo
diff --git a/src/mongo/s/catalog_cache.cpp b/src/mongo/s/catalog_cache.cpp
index 7482c8d27c7..f9ee6cde0a0 100644
--- a/src/mongo/s/catalog_cache.cpp
+++ b/src/mongo/s/catalog_cache.cpp
@@ -161,7 +161,7 @@ StatusWith<CachedCollectionRoutingInfo> CatalogCache::getCollectionRoutingInfo(
                     numRefreshAttempts < kMaxInconsistentRoutingInfoRefreshAttempts) {
                     ul.unlock();
 
-                    log() << "Metadata refresh for " << nss.ns() << " failed and will be retried"
+                    log() << "Metadata refresh for " << nss << " failed and will be retried"
                           << causedBy(redact(ex));
 
                     // Do the sleep outside of the mutex
@@ -303,7 +303,8 @@ std::shared_ptr<ChunkManager> CatalogCache::refreshCollectionRoutingInfo(
         chunkMap = existingRoutingInfo->chunkMap();
     }
 
-    log() << "Refreshing chunks based on version " << startingCollectionVersion;
+    log() << "Refreshing chunks for collection " << nss << " based on version "
+          << startingCollectionVersion;
 
     // Diff tracker should *always* find at least one chunk if collection exists
     const auto diffQuery =
@@ -329,9 +330,10 @@ std::shared_ptr<ChunkManager> CatalogCache::refreshCollectionRoutingInfo(
     const int diffsApplied = differ.calculateConfigDiff(opCtx, newChunks);
 
     if (diffsApplied < 1) {
-        log() << "Refresh took " << t.millis() << " ms and failed because the collection's "
-                                                  "sharding metadata either changed in between or "
-                                                  "became corrupted";
+        log() << "Refresh for collection " << nss << " took " << t.millis()
+              << " ms and failed because the collection's "
+                 "sharding metadata either changed in between or "
+                 "became corrupted";
 
         uasserted(ErrorCodes::ConflictingOperationInProgress,
                   "Collection sharding status changed during refresh or became corrupted");
@@ -345,7 +347,8 @@ std::shared_ptr<ChunkManager> CatalogCache::refreshCollectionRoutingInfo(
     // sequence number to detect batch writes not making progress because of chunks moving across
     // shards too frequently.
     if (collectionVersion == startingCollectionVersion) {
-        log() << "Refresh took " << t.millis() << " ms and didn't find any metadata changes";
+        log() << "Refresh for collection " << nss << " took " << t.millis()
+              << " ms and didn't find any metadata changes";
 
         return existingRoutingInfo;
     }
@@ -357,7 +360,8 @@ std::shared_ptr<ChunkManager> CatalogCache::refreshCollectionRoutingInfo(
                                               ->makeFromBSON(coll.getDefaultCollation()));
     }
 
-    log() << "Refresh took " << t.millis() << " ms and found version " << collectionVersion;
+    log() << "Refresh for collection " << nss << " took " << t.millis() << " ms and found version "
+          << collectionVersion;
 
     return stdx::make_unique<ChunkManager>(nss,
                                            coll.getKeyPattern(),
diff --git a/src/mongo/s/chunk_manager.cpp b/src/mongo/s/chunk_manager.cpp
index de37ac85147..ecc743fd63f 100644
--- a/src/mongo/s/chunk_manager.cpp
+++ b/src/mongo/s/chunk_manager.cpp
@@ -34,6 +34,7 @@
 
 #include <vector>
 
+#include "mongo/base/owned_pointer_vector.h"
 #include "mongo/bson/simple_bsonobj_comparator.h"
 #include "mongo/db/matcher/extensions_callback_noop.h"
 #include "mongo/db/query/collation/collation_index_key.h"
diff --git a/src/mongo/s/chunk_manager_refresh_test.cpp b/src/mongo/s/chunk_manager_refresh_test.cpp
index 511a4ed15d9..432405056e6 100644
--- a/src/mongo/s/chunk_manager_refresh_test.cpp
+++ b/src/mongo/s/chunk_manager_refresh_test.cpp
@@ -33,18 +33,174 @@
 #include <set>
 
 #include "mongo/db/client.h"
+#include "mongo/db/query/query_request.h"
 #include "mongo/s/catalog/type_chunk.h"
 #include "mongo/s/catalog/type_collection.h"
 #include "mongo/s/catalog/type_shard.h"
 #include "mongo/s/catalog_cache.h"
 #include "mongo/s/chunk_manager_test_fixture.h"
-#include "mongo/util/scopeguard.h"
 
 namespace mongo {
 namespace {
 
+using executor::RemoteCommandRequest;
+using unittest::assertGet;
+
 using ChunkManagerLoadTest = ChunkManagerTestFixture;
 
+TEST_F(ChunkManagerLoadTest, FullLoad) {
+    const OID epoch = OID::gen();
+    const ShardKeyPattern shardKeyPattern(BSON("_id" << 1));
+
+    auto future = launchAsync([&] {
+        auto client = serviceContext()->makeClient("Test");
+        auto opCtx = client->makeOperationContext();
+        return CatalogCache::refreshCollectionRoutingInfo(opCtx.get(), kNss, nullptr);
+    });
+
+    expectFindOnConfigSendBSONObjVector([&]() {
+        CollectionType collType;
+        collType.setNs(kNss);
+        collType.setEpoch(epoch);
+        collType.setKeyPattern(shardKeyPattern.toBSON());
+        collType.setUnique(false);
+
+        return std::vector<BSONObj>{collType.toBSON()};
+    }());
+
+    expectFindOnConfigSendBSONObjVector([&]() {
+        ChunkVersion version(1, 0, epoch);
+
+        ChunkType chunk1(kNss,
+                         {shardKeyPattern.getKeyPattern().globalMin(), BSON("_id" << -100)},
+                         version,
+                         {"0"});
+        version.incMinor();
+
+        ChunkType chunk2(kNss, {BSON("_id" << -100), BSON("_id" << 0)}, version, {"1"});
+        version.incMinor();
+
+        ChunkType chunk3(kNss, {BSON("_id" << 0), BSON("_id" << 100)}, version, {"0"});
+        version.incMinor();
+
+        ChunkType chunk4(kNss,
+                         {BSON("_id" << 100), shardKeyPattern.getKeyPattern().globalMax()},
+                         version,
+                         {"1"});
+        version.incMinor();
+
+        return std::vector<BSONObj>{
+            chunk1.toBSON(), chunk2.toBSON(), chunk3.toBSON(), chunk4.toBSON()};
+    }());
+
+    expectFindOnConfigSendBSONObjVector([&]() {
+        ShardType shard1;
+        shard1.setName("0");
+        shard1.setHost(str::stream() << "Host0:12345");
+
+        ShardType shard2;
+        shard2.setName("1");
+        shard2.setHost(str::stream() << "Host1:12345");
+
+        return std::vector<BSONObj>{shard1.toBSON(), shard2.toBSON()};
+    }());
+
+    auto routingInfo = future.timed_get(kFutureTimeout);
+    ASSERT_EQ(4, routingInfo->numChunks());
+}
+
+TEST_F(ChunkManagerLoadTest, CollectionNotFound) {
+    auto future = launchAsync([&] {
+        auto client = serviceContext()->makeClient("Test");
+        auto opCtx = client->makeOperationContext();
+        return CatalogCache::refreshCollectionRoutingInfo(opCtx.get(), kNss, nullptr);
+    });
+
+    // Return an empty collection
+    expectFindOnConfigSendBSONObjVector({});
+
+    ASSERT(nullptr == future.timed_get(kFutureTimeout));
+}
+
+TEST_F(ChunkManagerLoadTest, NoChunksFoundForCollection) {
+    const OID epoch = OID::gen();
+    const ShardKeyPattern shardKeyPattern(BSON("_id" << 1));
+
+    auto future = launchAsync([&] {
+        auto client = serviceContext()->makeClient("Test");
+        auto opCtx = client->makeOperationContext();
+        return CatalogCache::refreshCollectionRoutingInfo(opCtx.get(), kNss, nullptr);
+    });
+
+    expectFindOnConfigSendBSONObjVector([&]() {
+        CollectionType collType;
+        collType.setNs(kNss);
+        collType.setEpoch(epoch);
+        collType.setKeyPattern(shardKeyPattern.toBSON());
+        collType.setUnique(false);
+
+        return std::vector<BSONObj>{collType.toBSON()};
+    }());
+
+    // Return no chunks
+    expectFindOnConfigSendBSONObjVector({});
+
+    try {
+        auto routingInfo = future.timed_get(kFutureTimeout);
+        FAIL(str::stream() << "Returning no chunks for collection did not fail and returned "
+                           << (routingInfo ? routingInfo->toString() : "nullptr"));
+    } catch (const DBException& ex) {
+        ASSERT_EQ(ErrorCodes::ConflictingOperationInProgress, ex.getCode());
+    }
+}
+
+TEST_F(ChunkManagerLoadTest, ChunkEpochChangeDuringIncrementalLoad) {
+    const ShardKeyPattern shardKeyPattern(BSON("_id" << 1));
+
+    auto initialRoutingInfo(makeChunkManager(shardKeyPattern, nullptr, true, {}));
+    ASSERT_EQ(1, initialRoutingInfo->numChunks());
+
+    auto future = launchAsync([&] {
+        auto client = serviceContext()->makeClient("Test");
+        auto opCtx = client->makeOperationContext();
+        return CatalogCache::refreshCollectionRoutingInfo(opCtx.get(), kNss, initialRoutingInfo);
+    });
+
+    expectFindOnConfigSendBSONObjVector([&]() {
+        CollectionType collType;
+        collType.setNs(kNss);
+        collType.setEpoch(initialRoutingInfo->getVersion().epoch());
+        collType.setKeyPattern(shardKeyPattern.toBSON());
+        collType.setUnique(false);
+
+        return std::vector<BSONObj>{collType.toBSON()};
+    }());
+
+    ChunkVersion version = initialRoutingInfo->getVersion();
+
+    // Return set of chunks, one of which has different epoch
+    expectFindOnConfigSendBSONObjVector([&]() {
+        version.incMajor();
+        ChunkType chunk1(
+            kNss, {shardKeyPattern.getKeyPattern().globalMin(), BSON("_id" << 0)}, version, {"0"});
+
+        ChunkType chunk2(kNss,
+                         {BSON("_id" << 0), shardKeyPattern.getKeyPattern().globalMax()},
+                         ChunkVersion(1, 0, OID::gen()),
+                         {"0"});
+
+        return std::vector<BSONObj>{chunk1.toBSON(), chunk2.toBSON()};
+    }());
+
+    try {
+        auto routingInfo = future.timed_get(kFutureTimeout);
+        FAIL(str::stream() << "Returning chunks with different epoch did not fail and returned "
+                           << (routingInfo ? routingInfo->toString() : "nullptr"));
+    } catch (const DBException& ex) {
+        ASSERT_EQ(ErrorCodes::ConflictingOperationInProgress, ex.getCode());
+    }
+}
+
 TEST_F(ChunkManagerLoadTest, IncrementalLoadAfterSplit) {
     const ShardKeyPattern shardKeyPattern(BSON("_id" << 1));
 
@@ -70,7 +226,15 @@ TEST_F(ChunkManagerLoadTest, IncrementalLoadAfterSplit) {
     }());
 
     // Return set of chunks, which represent a split
-    expectFindOnConfigSendBSONObjVector([&]() {
+    onFindCommand([&](const RemoteCommandRequest& request) {
+        // Ensure it is a differential query
+        const auto diffQuery =
+            assertGet(QueryRequest::makeFromFindCommand(kNss, request.cmdObj, false));
+        ASSERT_BSONOBJ_EQ(
+            BSON("ns" << kNss.ns() << "lastmod"
+                      << BSON("$gte" << Timestamp(version.majorVersion(), version.minorVersion()))),
+            diffQuery->getFilter());
+
         version.incMajor();
         ChunkType chunk1(
             kNss, {shardKeyPattern.getKeyPattern().globalMin(), BSON("_id" << 0)}, version, {"0"});
@@ -80,7 +244,7 @@ TEST_F(ChunkManagerLoadTest, IncrementalLoadAfterSplit) {
             kNss, {BSON("_id" << 0), shardKeyPattern.getKeyPattern().globalMax()}, version, {"0"});
 
         return std::vector<BSONObj>{chunk1.toBSON(), chunk2.toBSON()};
-    }());
+    });
 
     auto newRoutingInfo(future.timed_get(kFutureTimeout));
     ASSERT_EQ(2, newRoutingInfo->numChunks());
diff --git a/src/mongo/s/shard_key_pattern.cpp b/src/mongo/s/shard_key_pattern.cpp
index 1a8e77658e1..f3454c31a62 100644
--- a/src/mongo/s/shard_key_pattern.cpp
+++ b/src/mongo/s/shard_key_pattern.cpp
@@ -43,33 +43,19 @@
 
 namespace mongo {
 
-using std::make_pair;
-using std::pair;
 using std::shared_ptr;
 using std::string;
 using std::unique_ptr;
 using std::vector;
 
 using pathsupport::EqualityMatches;
-using mongoutils::str::stream;
 
 const int ShardKeyPattern::kMaxShardKeySizeBytes = 512;
 const unsigned int ShardKeyPattern::kMaxFlattenedInCombinations = 4000000;
 
-Status ShardKeyPattern::checkShardKeySize(const BSONObj& shardKey) {
-    if (shardKey.objsize() <= kMaxShardKeySizeBytes)
-        return Status::OK();
-
-    return Status(ErrorCodes::ShardKeyTooBig,
-                  stream() << "shard keys must be less than " << kMaxShardKeySizeBytes
-                           << " bytes, but key "
-                           << shardKey
-                           << " is "
-                           << shardKey.objsize()
-                           << " bytes");
-}
+namespace {
 
-static bool isHashedPatternEl(const BSONElement& el) {
+bool isHashedPatternEl(const BSONElement& el) {
     return el.type() == String && el.String() == IndexNames::HASHED;
 }
 
@@ -78,40 +64,67 @@ static bool isHashedPatternEl(const BSONElement& el) {
  * i) a hashed single field, e.g. { a : "hashed" }, or
  * ii) a compound list of ascending, potentially-nested field paths, e.g. { a : 1 , b.c : 1 }
  */
-static vector<FieldRef*> parseShardKeyPattern(const BSONObj& keyPattern) {
-    OwnedPointerVector<FieldRef> parsedPaths;
-    static const vector<FieldRef*> empty;
+std::vector<FieldRef*> parseShardKeyPattern(const BSONObj& keyPattern) {
+    std::vector<FieldRef*> parsedPaths;
 
-    BSONObjIterator patternIt(keyPattern);
-    while (patternIt.more()) {
-        BSONElement patternEl = patternIt.next();
-        parsedPaths.push_back(new FieldRef(patternEl.fieldNameStringData()));
-        const FieldRef& patternPath = *parsedPaths.back();
+    for (const auto& patternEl : keyPattern) {
+        auto newFieldRef(stdx::make_unique<FieldRef>(patternEl.fieldNameStringData()));
 
         // Empty path
-        if (patternPath.numParts() == 0)
-            return empty;
+        if (newFieldRef->numParts() == 0)
+            return {};
 
         // Extra "." in path?
-        if (patternPath.dottedField() != patternEl.fieldNameStringData())
-            return empty;
+        if (newFieldRef->dottedField() != patternEl.fieldNameStringData())
+            return {};
 
         // Empty parts of the path, ".."?
-        for (size_t i = 0; i < patternPath.numParts(); ++i) {
-            if (patternPath.getPart(i).size() == 0)
-                return empty;
+        for (size_t i = 0; i < newFieldRef->numParts(); ++i) {
+            if (newFieldRef->getPart(i).empty())
+                return {};
         }
 
         // Numeric and ascending (1.0), or "hashed" and single field
         if (!patternEl.isNumber()) {
             if (keyPattern.nFields() != 1 || !isHashedPatternEl(patternEl))
-                return empty;
+                return {};
         } else if (patternEl.numberInt() != 1) {
-            return empty;
+            return {};
         }
+
+        parsedPaths.emplace_back(newFieldRef.release());
     }
 
-    return parsedPaths.release();
+    return parsedPaths;
+}
+
+bool isShardKeyElement(const BSONElement& element, bool allowRegex) {
+    if (element.eoo() || element.type() == Array)
+        return false;
+
+    // TODO: Disallow regex all the time
+    if (!allowRegex && element.type() == RegEx)
+        return false;
+
+    if (element.type() == Object && !element.embeddedObject().okForStorage())
+        return false;
+
+    return true;
+}
+
+}  // namespace
+
+Status ShardKeyPattern::checkShardKeySize(const BSONObj& shardKey) {
+    if (shardKey.objsize() <= kMaxShardKeySizeBytes)
+        return Status::OK();
+
+    return {ErrorCodes::ShardKeyTooBig,
+            str::stream() << "shard keys must be less than " << kMaxShardKeySizeBytes
+                          << " bytes, but key "
+                          << shardKey
+                          << " is "
+                          << shardKey.objsize()
+                          << " bytes"};
 }
 
 ShardKeyPattern::ShardKeyPattern(const BSONObj& keyPattern)
@@ -119,8 +132,7 @@ ShardKeyPattern::ShardKeyPattern(const BSONObj& keyPattern)
       _keyPattern(_keyPatternPaths.empty() ? BSONObj() : keyPattern) {}
 
 ShardKeyPattern::ShardKeyPattern(const KeyPattern& keyPattern)
-    : _keyPatternPaths(parseShardKeyPattern(keyPattern.toBSON())),
-      _keyPattern(_keyPatternPaths.empty() ? KeyPattern(BSONObj()) : keyPattern) {}
+    : ShardKeyPattern(keyPattern.toBSON()) {}
 
 bool ShardKeyPattern::isValid() const {
     return !_keyPattern.toBSON().isEmpty();
@@ -134,6 +146,10 @@ const KeyPattern& ShardKeyPattern::getKeyPattern() const {
     return _keyPattern;
 }
 
+const std::vector<FieldRef*>& ShardKeyPattern::getKeyPatternFields() const {
+    return _keyPatternPaths.vector();
+}
+
 const BSONObj& ShardKeyPattern::toBSON() const {
     return _keyPattern.toBSON();
 }
@@ -142,30 +158,22 @@ string ShardKeyPattern::toString() const {
     return toBSON().toString();
 }
 
-static bool isShardKeyElement(const BSONElement& element, bool allowRegex) {
-    // TODO: Disallow regex all the time
-    if (element.eoo() || element.type() == Array || (!allowRegex && element.type() == RegEx) ||
-        (element.type() == Object && !element.embeddedObject().okForStorage()))
-        return false;
-    return true;
-}
-
 bool ShardKeyPattern::isShardKey(const BSONObj& shardKey) const {
     // Shard keys are always of the form: { 'nested.path' : value, 'nested.path2' : value }
 
     if (!isValid())
         return false;
 
-    BSONObjIterator patternIt(_keyPattern.toBSON());
-    while (patternIt.more()) {
-        BSONElement patternEl = patternIt.next();
+    const auto& keyPatternBSON = _keyPattern.toBSON();
 
+    for (const auto& patternEl : keyPatternBSON) {
         BSONElement keyEl = shardKey[patternEl.fieldNameStringData()];
+
         if (!isShardKeyElement(keyEl, true))
             return false;
     }
 
-    return true;
+    return shardKey.nFields() == keyPatternBSON.nFields();
 }
 
 BSONObj ShardKeyPattern::normalizeShardKey(const BSONObj& shardKey) const {
@@ -351,18 +359,21 @@ BoundList ShardKeyPattern::flattenBounds(const IndexBounds& indexBounds) const {
             return BoundList();
         }
     }
-    // To construct our bounds we will generate intervals based on bounds for
-    // the first field, then compound intervals based on constraints for the first
-    // 2 fields, then compound intervals for the first 3 fields, etc.
-    // As we loop through the fields, we start generating new intervals that will later
-    // get extended in another iteration of the loop.  We define these partially constructed
-    // intervals using pairs of BSONObjBuilders (shared_ptrs, since after one iteration of the
-    // loop they still must exist outside their scope).
-    typedef vector<pair<shared_ptr<BSONObjBuilder>, shared_ptr<BSONObjBuilder>>> BoundBuilders;
+
+    // To construct our bounds we will generate intervals based on bounds for the first field, then
+    // compound intervals based on constraints for the first 2 fields, then compound intervals for
+    // the first 3 fields, etc.
+    //
+    // As we loop through the fields, we start generating new intervals that will later get extended
+    // in another iteration of the loop. We define these partially constructed intervals using pairs
+    // of BSONObjBuilders (shared_ptrs, since after one iteration of the loop they still must exist
+    // outside their scope).
+    typedef vector<std::pair<std::shared_ptr<BSONObjBuilder>, std::shared_ptr<BSONObjBuilder>>>
+        BoundBuilders;
 
     BoundBuilders builders;
-    builders.push_back(make_pair(shared_ptr<BSONObjBuilder>(new BSONObjBuilder()),
-                                 shared_ptr<BSONObjBuilder>(new BSONObjBuilder())));
+    builders.emplace_back(shared_ptr<BSONObjBuilder>(new BSONObjBuilder()),
+                          shared_ptr<BSONObjBuilder>(new BSONObjBuilder()));
     BSONObjIterator keyIter(_keyPattern.toBSON());
     // until equalityOnly is false, we are just dealing with equality (no range or $in queries).
     bool equalityOnly = true;
@@ -404,9 +415,8 @@ BoundList ShardKeyPattern::flattenBounds(const IndexBounds& indexBounds) const {
                         uassert(17439,
                                 "combinatorial limit of $in partitioning of results exceeded",
                                 newBuilders.size() < kMaxFlattenedInCombinations);
-                        newBuilders.push_back(  //
-                            make_pair(shared_ptr<BSONObjBuilder>(new BSONObjBuilder()),
-                                      shared_ptr<BSONObjBuilder>(new BSONObjBuilder())));
+                        newBuilders.emplace_back(shared_ptr<BSONObjBuilder>(new BSONObjBuilder()),
+                                                 shared_ptr<BSONObjBuilder>(new BSONObjBuilder()));
                         newBuilders.back().first->appendElements(first);
                         newBuilders.back().second->appendElements(second);
                         newBuilders.back().first->appendAs(interval->start, fieldName);
@@ -425,9 +435,11 @@ BoundList ShardKeyPattern::flattenBounds(const IndexBounds& indexBounds) const {
             }
         }
     }
+
     BoundList ret;
     for (BoundBuilders::const_iterator i = builders.begin(); i != builders.end(); ++i)
-        ret.push_back(make_pair(i->first->obj(), i->second->obj()));
+        ret.emplace_back(i->first->obj(), i->second->obj());
+
     return ret;
 }
 
diff --git a/src/mongo/s/shard_key_pattern.h b/src/mongo/s/shard_key_pattern.h
index 67c8eeb4a2f..73038fcf9ed 100644
--- a/src/mongo/s/shard_key_pattern.h
+++ b/src/mongo/s/shard_key_pattern.h
@@ -91,6 +91,8 @@ public:
 
     const KeyPattern& getKeyPattern() const;
 
+    const std::vector<FieldRef*>& getKeyPatternFields() const;
+
     const BSONObj& toBSON() const;
 
     std::string toString() const;