1 files changed, 43 insertions, 30 deletions

diff --git a/src/mongo/db/stats/storage_stats.cpp b/src/mongo/db/stats/storage_stats.cpp
index 203c762856d..870dc908021 100644
--- a/src/mongo/db/stats/storage_stats.cpp
+++ b/src/mongo/db/stats/storage_stats.cpp
@@ -36,9 +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/s/balancer_stats_registry.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"
@@ -48,6 +49,36 @@
 
 namespace mongo {
 
+namespace {
+long long 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.exactNumberLong();
+}
+}  // namespace
+
 Status appendCollectionStorageStats(OperationContext* opCtx,
                                     const NamespaceString& nss,
                                     const StorageStatsSpec& storageStatsSpec,
@@ -59,14 +90,11 @@ Status appendCollectionStorageStats(OperationContext* opCtx,
     bool waitForLock = storageStatsSpec.getWaitForLock();
     bool numericOnly = storageStatsSpec.getNumericOnly();
 
-    const auto bucketNss =
-        nss.isTimeseriesBucketsCollection() ? nss : nss.makeTimeseriesBucketsNamespace();
-    // Hold reference to the catalog for collection lookup without locks to be safe.
-    auto catalog = CollectionCatalog::get(opCtx);
-    auto bucketsColl = catalog->lookupCollectionByNamespace(opCtx, bucketNss);
-    const bool mayBeTimeseries = bucketsColl && bucketsColl->getTimeseriesOptions();
+    const auto bucketNss = nss.makeTimeseriesBucketsNamespace();
+    const auto isTimeseries = nss.isTimeseriesBucketsCollection() ||
+        CollectionCatalog::get(opCtx)->lookupCollectionByNamespaceForRead(opCtx, bucketNss);
     const auto collNss =
-        (mayBeTimeseries && !nss.isTimeseriesBucketsCollection()) ? std::move(bucketNss) : nss;
+        (isTimeseries && !nss.isTimeseriesBucketsCollection()) ? std::move(bucketNss) : nss;
 
     boost::optional<AutoGetCollectionForReadCommandMaybeLockFree> autoColl;
     try {
@@ -80,15 +108,7 @@ Status appendCollectionStorageStats(OperationContext* opCtx,
     }
 
     const auto& collection = autoColl->getCollection();  // Will be set if present
-    const bool isTimeseries = collection && collection->getTimeseriesOptions().has_value();
-
-    // We decided the requested namespace was a time series view, so we redirected to the underlying
-    // buckets collection. However, when we tried to acquire that collection, it did not exist or it
-    // did not have time series options, which means it was dropped and potentially recreated in
-    // between the two calls. Logically, the collection that we were looking for does not exist.
-    bool logicallyNotFound = collNss != nss && !isTimeseries;
-
-    if (!collection || logicallyNotFound) {
+    if (!collection) {
         result->appendNumber("size", 0);
         result->appendNumber("count", 0);
         result->appendNumber(kOrphanCountField, 0);
@@ -123,17 +143,10 @@ Status appendCollectionStorageStats(OperationContext* opCtx,
         }
     }
 
-    if (serverGlobalParams.clusterRole == ClusterRole::ShardServer &&
-        !nss.isNamespaceAlwaysUnsharded()) {
-        if (serverGlobalParams.featureCompatibility.isVersionInitialized() &&
-            feature_flags::gOrphanTracking.isEnabled(serverGlobalParams.featureCompatibility)) {
-            result->appendNumber(
-                kOrphanCountField,
-                BalancerStatsRegistry::get(opCtx)->getCollNumOrphanDocsFromDiskIfNeeded(
-                    opCtx, collection->uuid()));
-        }
-    } else {
-        result->appendNumber(kOrphanCountField, 0);
+    if (serverGlobalParams.featureCompatibility.isVersionInitialized() &&
+        feature_flags::gOrphanTracking.isEnabled(serverGlobalParams.featureCompatibility)) {
+        result->appendNumber(kOrphanCountField,
+                             countOrphanDocsForCollection(opCtx, collection->uuid()));
     }
 
     const RecordStore* recordStore = collection->getRecordStore();
@@ -162,8 +175,8 @@ Status appendCollectionStorageStats(OperationContext* opCtx,
     BSONObjBuilder indexDetails;
     std::vector<std::string> indexBuilds;
 
-    auto it = indexCatalog->getIndexIterator(
-        opCtx, IndexCatalog::InclusionPolicy::kReady | IndexCatalog::InclusionPolicy::kUnfinished);
+    std::unique_ptr<IndexCatalog::IndexIterator> it =
+        indexCatalog->getIndexIterator(opCtx, /*includeUnfinishedIndexes=*/true);
     while (it->more()) {
         const IndexCatalogEntry* entry = it->next();
         const IndexDescriptor* descriptor = entry->descriptor();