SERVER-62554 Add orphans info in `collStats` aggregation stage output

3 files changed, 134 insertions, 0 deletions

diff --git a/jstests/sharding/collstats_returns_orphan_count.js b/jstests/sharding/collstats_returns_orphan_count.js
new file mode 100644
index 00000000000..4f7ae294e7a
--- /dev/null
+++ b/jstests/sharding/collstats_returns_orphan_count.js
@@ -0,0 +1,94 @@
+/**
+ * Test that collstats returns the correct number of orphaned documents.
+ *
+ * @tags: [
+ *  requires_fcv_60,
+ * ]
+ */
+
+(function() {
+'use strict';
+
+load("jstests/libs/fail_point_util.js");
+
+const rangeDeleterBatchSize = 128;
+
+const st = new ShardingTest({
+    shards: 2,
+    other: {
+        shardOptions: {setParameter: {rangeDeleterBatchSize: rangeDeleterBatchSize}},
+    }
+});
+
+function assertCollStatsHasCorrectOrphanCount(coll, shardName, numOrphans) {
+    const pipeline = [
+        {'$collStats': {'storageStats': {}}},
+        {'$project': {'shard': true, 'storageStats': {'orphanCount': true}}}
+    ];
+    const storageStats = coll.aggregate(pipeline).toArray();
+    storageStats.forEach((stat) => {
+        if (stat['shard'] === shardName) {
+            assert.eq(stat.storageStats.orphanCount, numOrphans);
+        }
+    });
+}
+
+// Setup database
+const dbName = 'db';
+const db = st.getDB(dbName);
+assert.commandWorked(
+    st.s.adminCommand({enableSharding: dbName, primaryShard: st.shard0.shardName}));
+
+// Test non-existing collection
+const noColl = db['unusedColl'];
+let res = db.runCommand({'collStats': noColl.getFullName()});
+assert.eq(res.shards[st.shard0.shardName].orphanCount, 0);
+
+// Setup collection for test with orphans
+const coll = db['test'];
+const nss = coll.getFullName();
+assert.commandWorked(st.s.adminCommand({shardCollection: nss, key: {_id: 1}}));
+
+// Create two chunks
+assert.commandWorked(st.s.adminCommand({split: nss, middle: {_id: 0}}));
+
+// Insert 1000 docs into the chunk we will move.
+const numDocs = 1000;
+let bulk = coll.initializeUnorderedBulkOp();
+for (let i = 0; i < numDocs; i++) {
+    bulk.insert({_id: i});
+}
+
+// Insert 10 docs into the chunk we will not move.
+const numDocsUnmoved = 10;
+for (let i = -numDocsUnmoved; i < 0; i++) {
+    bulk.insert({_id: i});
+}
+assert.commandWorked(bulk.execute());
+
+// Check there are no orphans before the chunk is moved
+assertCollStatsHasCorrectOrphanCount(coll, st.shard1.shardName, 0);
+
+// Pause before first range deletion task
+let beforeDeletionFailpoint = configureFailPoint(st.shard0, "hangBeforeDoingDeletion");
+let afterDeletionFailpoint = configureFailPoint(st.shard0, "hangAfterDoingDeletion");
+assert.commandWorked(db.adminCommand({moveChunk: nss, find: {_id: 0}, to: st.shard1.shardName}));
+
+// Check the batches are deleted correctly
+const numBatches = numDocs / rangeDeleterBatchSize;
+for (let i = 0; i < numBatches; i++) {
+    // Wait for failpoint and check num orphans
+    beforeDeletionFailpoint.wait();
+    assertCollStatsHasCorrectOrphanCount(
+        coll, st.shard0.shardName, numDocs - rangeDeleterBatchSize * i);
+    // Unset and reset failpoint without allowing any batches deleted in the meantime
+    afterDeletionFailpoint = configureFailPoint(st.shard0, "hangAfterDoingDeletion");
+    beforeDeletionFailpoint.off();
+    afterDeletionFailpoint.wait();
+    beforeDeletionFailpoint = configureFailPoint(st.shard0, "hangBeforeDoingDeletion");
+    afterDeletionFailpoint.off();
+}
+beforeDeletionFailpoint.off();
+
+st.stop();
+})();
diff --git a/src/mongo/db/stats/SConscript b/src/mongo/db/stats/SConscript
index b27a81181ac..17bbe0cb144 100644
--- a/src/mongo/db/stats/SConscript
+++ b/src/mongo/db/stats/SConscript
@@ -126,7 +126,9 @@ env.Library(
         '$BUILD_DIR/mongo/db/catalog/index_catalog',
         '$BUILD_DIR/mongo/db/commands/server_status',
         '$BUILD_DIR/mongo/db/db_raii',
+        '$BUILD_DIR/mongo/db/dbdirectclient', # TODO (SERVER-64162) remove
         '$BUILD_DIR/mongo/db/index/index_access_method',
+        '$BUILD_DIR/mongo/db/pipeline/aggregation_request_helper', # TODO (SERVER-64162) remove
         '$BUILD_DIR/mongo/db/pipeline/document_sources_idl',
         '$BUILD_DIR/mongo/db/timeseries/bucket_catalog',
         '$BUILD_DIR/mongo/db/timeseries/timeseries_stats',
diff --git a/src/mongo/db/stats/storage_stats.cpp b/src/mongo/db/stats/storage_stats.cpp
index c67830fca04..3acf204b1a8 100644
--- a/src/mongo/db/stats/storage_stats.cpp
+++ b/src/mongo/db/stats/storage_stats.cpp
@@ -36,8 +36,10 @@
 #include "mongo/db/catalog/database_holder.h"
 #include "mongo/db/catalog/index_catalog.h"
 #include "mongo/db/db_raii.h"
+#include "mongo/db/dbdirectclient.h"  // TODO (SERVER-64162) remove
 #include "mongo/db/index/index_access_method.h"
 #include "mongo/db/index/index_descriptor.h"
+#include "mongo/db/pipeline/aggregate_command_gen.h"  // TODO (SERVER-64162) remove
 #include "mongo/db/timeseries/bucket_catalog.h"
 #include "mongo/db/timeseries/timeseries_stats.h"
 #include "mongo/logv2/log.h"
@@ -46,10 +48,42 @@
 
 namespace mongo {
 
+namespace {
+int countOrphanDocsForCollection(OperationContext* opCtx, const UUID& uuid) {
+    // TODO (SERVER-64162): move this function to range_deletion_util.cpp and replace
+    // "collectionUuid" and "numOrphanDocs" with RangeDeletionTask field names.
+    DBDirectClient client(opCtx);
+    std::vector<BSONObj> pipeline;
+    pipeline.push_back(BSON("$match" << BSON("collectionUuid" << uuid)));
+    pipeline.push_back(BSON("$group" << BSON("_id"
+                                             << "numOrphans"
+                                             << "count"
+                                             << BSON("$sum"
+                                                     << "$numOrphanDocs"))));
+    AggregateCommandRequest aggRequest(NamespaceString::kRangeDeletionNamespace, pipeline);
+    auto swCursor = DBClientCursor::fromAggregationRequest(
+        &client, aggRequest, false /* secondaryOk */, true /* useExhaust */);
+    if (!swCursor.isOK()) {
+        return 0;
+    }
+    auto cursor = std::move(swCursor.getValue());
+    if (!cursor->more()) {
+        return 0;
+    }
+    auto res = cursor->nextSafe();
+    invariant(!cursor->more());
+    auto numOrphans = res.getField("count");
+    invariant(numOrphans);
+    return numOrphans.numberInt();
+}
+}  // namespace
+
 Status appendCollectionStorageStats(OperationContext* opCtx,
                                     const NamespaceString& nss,
                                     const StorageStatsSpec& storageStatsSpec,
                                     BSONObjBuilder* result) {
+    const std::string kOrphanCountField = "orphanCount";
+
     auto scale = storageStatsSpec.getScale().value_or(1);
     bool verbose = storageStatsSpec.getVerbose();
     bool waitForLock = storageStatsSpec.getWaitForLock();
@@ -76,6 +110,7 @@ Status appendCollectionStorageStats(OperationContext* opCtx,
     if (!collection) {
         result->appendNumber("size", 0);
         result->appendNumber("count", 0);
+        result->appendNumber(kOrphanCountField, 0);
         result->appendNumber("storageSize", 0);
         result->append("totalSize", 0);
         result->append("nindexes", 0);
@@ -107,6 +142,9 @@ Status appendCollectionStorageStats(OperationContext* opCtx,
         }
     }
 
+    result->appendNumber(kOrphanCountField,
+                         countOrphanDocsForCollection(opCtx, collection->uuid()));
+
     const RecordStore* recordStore = collection->getRecordStore();
     auto storageSize =
         static_cast<long long>(recordStore->storageSize(opCtx, result, verbose ? 1 : 0));