1 files changed, 30 insertions, 43 deletions

diff --git a/src/mongo/db/stats/storage_stats.cpp b/src/mongo/db/stats/storage_stats.cpp
index 870dc908021..203c762856d 100644
--- a/src/mongo/db/stats/storage_stats.cpp
+++ b/src/mongo/db/stats/storage_stats.cpp
@@ -36,10 +36,9 @@
 #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/s/balancer_stats_registry.h"
 #include "mongo/db/timeseries/bucket_catalog.h"
 #include "mongo/db/timeseries/timeseries_stats.h"
 #include "mongo/logv2/log.h"
@@ -49,36 +48,6 @@
 
 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,
@@ -90,11 +59,14 @@ Status appendCollectionStorageStats(OperationContext* opCtx,
     bool waitForLock = storageStatsSpec.getWaitForLock();
     bool numericOnly = storageStatsSpec.getNumericOnly();
 
-    const auto bucketNss = nss.makeTimeseriesBucketsNamespace();
-    const auto isTimeseries = nss.isTimeseriesBucketsCollection() ||
-        CollectionCatalog::get(opCtx)->lookupCollectionByNamespaceForRead(opCtx, bucketNss);
+    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 collNss =
-        (isTimeseries && !nss.isTimeseriesBucketsCollection()) ? std::move(bucketNss) : nss;
+        (mayBeTimeseries && !nss.isTimeseriesBucketsCollection()) ? std::move(bucketNss) : nss;
 
     boost::optional<AutoGetCollectionForReadCommandMaybeLockFree> autoColl;
     try {
@@ -108,7 +80,15 @@ Status appendCollectionStorageStats(OperationContext* opCtx,
     }
 
     const auto& collection = autoColl->getCollection();  // Will be set if present
-    if (!collection) {
+    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) {
         result->appendNumber("size", 0);
         result->appendNumber("count", 0);
         result->appendNumber(kOrphanCountField, 0);
@@ -143,10 +123,17 @@ Status appendCollectionStorageStats(OperationContext* opCtx,
         }
     }
 
-    if (serverGlobalParams.featureCompatibility.isVersionInitialized() &&
-        feature_flags::gOrphanTracking.isEnabled(serverGlobalParams.featureCompatibility)) {
-        result->appendNumber(kOrphanCountField,
-                             countOrphanDocsForCollection(opCtx, collection->uuid()));
+    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);
     }
 
     const RecordStore* recordStore = collection->getRecordStore();
@@ -175,8 +162,8 @@ Status appendCollectionStorageStats(OperationContext* opCtx,
     BSONObjBuilder indexDetails;
     std::vector<std::string> indexBuilds;
 
-    std::unique_ptr<IndexCatalog::IndexIterator> it =
-        indexCatalog->getIndexIterator(opCtx, /*includeUnfinishedIndexes=*/true);
+    auto it = indexCatalog->getIndexIterator(
+        opCtx, IndexCatalog::InclusionPolicy::kReady | IndexCatalog::InclusionPolicy::kUnfinished);
     while (it->more()) {
         const IndexCatalogEntry* entry = it->next();
         const IndexDescriptor* descriptor = entry->descriptor();