SERVER-42152 Delete existing chunks on new epoch in persisted routing table cache

Author: JamesHeppenstall

buildscripts/resmokeconfig/suites/sharding_continuous_config_stepdown.yml

@@ -206,6 +206,7 @@ selector:
   # TODO (SERVER-42143): The tests below need transactional support for refineCollectionShardKey.
   - jstests/sharding/refine_collection_shard_key_basic.js
   - jstests/sharding/refine_collection_shard_key_jumbo.js
+  - jstests/sharding/refine_collection_shard_key_drops_chunks.js
 
 executor:
   config:

buildscripts/resmokeconfig/suites/sharding_last_stable_mongos_and_mixed_shards.yml

@@ -142,6 +142,7 @@ selector:
   - jstests/sharding/explain_exec_stats_on_shards.js
   - jstests/sharding/refine_collection_shard_key_basic.js
   - jstests/sharding/refine_collection_shard_key_jumbo.js
+  - jstests/sharding/refine_collection_shard_key_drops_chunks.js
   - jstests/sharding/move_primary_clone_test.js
   - jstests/sharding/database_versioning_safe_secondary_reads.js
   - jstests/sharding/clone_catalog_data.js

jstests/sharding/refine_collection_shard_key_drops_chunks.js

@@ -0,0 +1,86 @@
+//
+// Tests that refineCollectionShardKey deletes all existing chunks in the persisted routing table
+// cache.
+//
+
+(function() {
+'use strict';
+load('jstests/sharding/libs/sharded_transactions_helpers.js');
+
+const st = new ShardingTest({shards: 1});
+const mongos = st.s0;
+const shard = st.shard0;
+const kDbName = 'db';
+const kCollName = 'foo';
+const kNsName = kDbName + '.' + kCollName;
+const kConfigCacheChunks = 'config.cache.chunks.' + kNsName;
+const oldKeyDoc = {
+    a: 1,
+    b: 1
+};
+const newKeyDoc = {
+    a: 1,
+    b: 1,
+    c: 1,
+    d: 1
+};
+
+assert.commandWorked(mongos.adminCommand({enableSharding: kDbName}));
+assert.commandWorked(mongos.adminCommand({shardCollection: kNsName, key: oldKeyDoc}));
+assert.commandWorked(mongos.getCollection(kNsName).createIndex(newKeyDoc));
+
+// Ensure that there exist three chunks belonging to 'db.foo' covering the entire key range.
+//
+// Chunk 1: {a: MinKey, b: MinKey} -->> {a: 0, b: 0}
+// Chunk 2: {a: 0, b: 0} -->> {a: 5, b: 5}
+// Chunk 3: {a: 5, b: 5} -->> {a: MaxKey, b: MaxKey}
+assert.commandWorked(mongos.adminCommand({split: kNsName, middle: {a: 0, b: 0}}));
+assert.commandWorked(mongos.adminCommand({split: kNsName, middle: {a: 5, b: 5}}));
+
+// Flush the routing table cache and verify that 'config.cache.chunks.db.foo' is as expected
+// before refineCollectionShardKey.
+assert.commandWorked(shard.adminCommand({_flushRoutingTableCacheUpdates: kNsName}));
+let chunkArr = shard.getCollection(kConfigCacheChunks).find({}).sort({min: 1}).toArray();
+assert.eq(3, chunkArr.length);
+assert.eq({a: MinKey, b: MinKey}, chunkArr[0]._id);
+assert.eq({a: 0, b: 0}, chunkArr[0].max);
+assert.eq({a: 0, b: 0}, chunkArr[1]._id);
+assert.eq({a: 5, b: 5}, chunkArr[1].max);
+assert.eq({a: 5, b: 5}, chunkArr[2]._id);
+assert.eq({a: MaxKey, b: MaxKey}, chunkArr[2].max);
+
+assert.commandWorked(mongos.adminCommand({refineCollectionShardKey: kNsName, key: newKeyDoc}));
+
+// Enable failpoint 'hangPersistCollectionAndChangedChunksAfterDropChunks' and flush the routing
+// table cache.
+assert.commandWorked(shard.adminCommand({
+    configureFailPoint: 'hangPersistCollectionAndChangedChunksAfterDropChunks',
+    mode: 'alwaysOn'
+}));
+const awaitShellToFlushRoutingTableCacheUpdates = startParallelShell(() => {
+    assert.commandWorked(db.adminCommand({_flushRoutingTableCacheUpdates: 'db.foo'}));
+}, st.rs0.getPrimary().port);
+
+// Verify that all chunks belonging to 'db.foo' have been deleted.
+waitForFailpoint('Hit hangPersistCollectionAndChangedChunksAfterDropChunks', 1);
+chunkArr = shard.getCollection(kConfigCacheChunks).find({}).sort({min: 1}).toArray();
+assert.eq(0, chunkArr.length);
+
+// Disable failpoint 'hangPersistCollectionAndChangedChunksAfterDropChunks' and continue
+// flushing the routing table cache.
+assert.commandWorked(shard.adminCommand(
+    {configureFailPoint: 'hangPersistCollectionAndChangedChunksAfterDropChunks', mode: 'off'}));
+awaitShellToFlushRoutingTableCacheUpdates();
+
+// Verify that 'config.cache.chunks.db.foo' is as expected after refineCollectionShardKey.
+chunkArr = shard.getCollection(kConfigCacheChunks).find({}).sort({min: 1}).toArray();
+assert.eq(3, chunkArr.length);
+assert.eq({a: MinKey, b: MinKey, c: MinKey, d: MinKey}, chunkArr[0]._id);
+assert.eq({a: 0, b: 0, c: MinKey, d: MinKey}, chunkArr[0].max);
+assert.eq({a: 0, b: 0, c: MinKey, d: MinKey}, chunkArr[1]._id);
+assert.eq({a: 5, b: 5, c: MinKey, d: MinKey}, chunkArr[1].max);
+assert.eq({a: 5, b: 5, c: MinKey, d: MinKey}, chunkArr[2]._id);
+assert.eq({a: MaxKey, b: MaxKey, c: MaxKey, d: MaxKey}, chunkArr[2].max);
+
+st.stop();
+})();

src/mongo/db/s/shard_metadata_util.cpp

@@ -432,6 +432,21 @@ Status dropChunksAndDeleteCollectionsEntry(OperationContext* opCtx, const Namesp
     }
 }
 
+void dropChunks(OperationContext* opCtx, const NamespaceString& nss) {
+    DBDirectClient client(opCtx);
+
+    // Drop the config.chunks collection associated with namespace 'nss'.
+    BSONObj result;
+    if (!client.dropCollection(ChunkType::ShardNSPrefix + nss.ns(), kLocalWriteConcern, &result)) {
+        auto status = getStatusFromCommandResult(result);
+        if (status != ErrorCodes::NamespaceNotFound) {
+            uassertStatusOK(status);
+        }
+    }
+
+    LOG(1) << "Successfully cleared persisted chunk metadata for collection '" << nss << "'.";
+}
+
 Status deleteDatabasesEntry(OperationContext* opCtx, StringData dbName) {
     try {
         DBDirectClient client(opCtx);

src/mongo/db/s/shard_metadata_util.h

@@ -210,6 +210,11 @@ Status updateShardChunks(OperationContext* opCtx,
  */
 Status dropChunksAndDeleteCollectionsEntry(OperationContext* opCtx, const NamespaceString& nss);
 
+/**
+ * Drops locally persisted chunk metadata associated with 'nss': only drops the chunks collection.
+ */
+void dropChunks(OperationContext* opCtx, const NamespaceString& nss);
+
 /**
  * Deletes locally persisted database metadata associated with 'dbName': removes the databases
  * collection entry.

src/mongo/db/s/shard_server_catalog_cache_loader.cpp

@@ -46,6 +46,7 @@
 #include "mongo/s/catalog/type_shard_database.h"
 #include "mongo/s/client/shard_registry.h"
 #include "mongo/s/grid.h"
+#include "mongo/util/fail_point_service.h"
 #include "mongo/util/log.h"
 
 namespace mongo {
@@ -56,6 +57,8 @@ using CollectionAndChangedChunks = CatalogCacheLoader::CollectionAndChangedChunk
 
 namespace {
 
+MONGO_FAIL_POINT_DEFINE(hangPersistCollectionAndChangedChunksAfterDropChunks);
+
 AtomicWord<unsigned long long> taskIdGenerator{0};
 
 /**
@@ -75,6 +78,24 @@ ThreadPool::Options makeDefaultThreadPoolOptions() {
     return options;
 }
 
+void dropChunksIfEpochChanged(OperationContext* opCtx,
+                              const NamespaceString& nss,
+                              const CollectionAndChangedChunks& collAndChunks,
+                              const ChunkVersion& maxLoaderVersion) {
+    if (collAndChunks.epoch != maxLoaderVersion.epoch() &&
+        maxLoaderVersion != ChunkVersion::UNSHARDED()) {
+        // If the collection has a new epoch, delete all existing chunks in the persisted routing
+        // table cache.
+        dropChunks(opCtx, nss);
+
+        if (MONGO_FAIL_POINT(hangPersistCollectionAndChangedChunksAfterDropChunks)) {
+            log() << "Hit hangPersistCollectionAndChangedChunksAfterDropChunks failpoint";
+            MONGO_FAIL_POINT_PAUSE_WHILE_SET_OR_INTERRUPTED(
+                opCtx, hangPersistCollectionAndChangedChunksAfterDropChunks);
+        }
+    }
+}
+
 /**
  * Takes a CollectionAndChangedChunks object and persists the changes to the shard's metadata
  * collections.
@@ -83,7 +104,8 @@ ThreadPool::Options makeDefaultThreadPoolOptions() {
  */
 Status persistCollectionAndChangedChunks(OperationContext* opCtx,
                                          const NamespaceString& nss,
-                                         const CollectionAndChangedChunks& collAndChunks) {
+                                         const CollectionAndChangedChunks& collAndChunks,
+                                         const ChunkVersion& maxLoaderVersion) {
     // Update the collections collection entry for 'nss' in case there are any new updates.
     ShardCollectionType update = ShardCollectionType(
         nss, collAndChunks.epoch, collAndChunks.shardKeyPattern, collAndChunks.shardKeyIsUnique);
@@ -107,6 +129,12 @@ Status persistCollectionAndChangedChunks(OperationContext* opCtx,
     }
 
     // Update the chunks.
+    try {
+        dropChunksIfEpochChanged(opCtx, nss, collAndChunks, maxLoaderVersion);
+    } catch (const DBException& ex) {
+        return ex.toStatus();
+    }
+
     status = updateShardChunks(opCtx, nss, collAndChunks.changedChunks, collAndChunks.epoch);
     if (!status.isOK()) {
         return status;
@@ -1036,7 +1064,8 @@ void ShardServerCatalogCacheLoader::_updatePersistedCollAndChunksMetadata(
     }
 
     uassertStatusOKWithContext(
-        persistCollectionAndChangedChunks(opCtx, nss, *task.collectionAndChangedChunks),
+        persistCollectionAndChangedChunks(
+            opCtx, nss, *task.collectionAndChangedChunks, task.minQueryVersion),
         str::stream() << "Failed to update the persisted chunk metadata for collection '"
                       << nss.ns() << "' from '" << task.minQueryVersion.toString() << "' to '"
                       << task.maxQueryVersion.toString() << "'. Will be retried.");