SERVER-48882 Remove newly empty database directories when using directoryperdb

20 files changed, 123 insertions, 86 deletions

diff --git a/src/mongo/db/storage/devnull/devnull_kv_engine.h b/src/mongo/db/storage/devnull/devnull_kv_engine.h
index 17e72a3c108..5a7748edb2c 100644
--- a/src/mongo/db/storage/devnull/devnull_kv_engine.h
+++ b/src/mongo/db/storage/devnull/devnull_kv_engine.h
@@ -78,7 +78,7 @@ public:
     virtual std::unique_ptr<SortedDataInterface> getSortedDataInterface(
         OperationContext* opCtx, StringData ident, const IndexDescriptor* desc);
 
-    virtual Status dropIdent(OperationContext* opCtx, RecoveryUnit* ru, StringData ident) {
+    virtual Status dropIdent(RecoveryUnit* ru, StringData ident) {
         return Status::OK();
     }
 
diff --git a/src/mongo/db/storage/durable_catalog_impl.cpp b/src/mongo/db/storage/durable_catalog_impl.cpp
index 84e626b6cbb..7d28e888dd1 100644
--- a/src/mongo/db/storage/durable_catalog_impl.cpp
+++ b/src/mongo/db/storage/durable_catalog_impl.cpp
@@ -230,8 +230,7 @@ public:
     virtual void rollback() {
         // Intentionally ignoring failure.
         auto kvEngine = _engine->getEngine();
-        MONGO_COMPILER_VARIABLE_UNUSED auto status =
-            kvEngine->dropIdent(_opCtx, _recoveryUnit, _ident);
+        MONGO_COMPILER_VARIABLE_UNUSED auto status = kvEngine->dropIdent(_recoveryUnit, _ident);
     }
 
     OperationContext* const _opCtx;
@@ -887,11 +886,10 @@ StatusWith<std::pair<RecordId, std::unique_ptr<RecordStore>>> DurableCatalogImpl
 
     auto ru = opCtx->recoveryUnit();
     CollectionUUID uuid = options.uuid.get();
-    opCtx->recoveryUnit()->onRollback(
-        [opCtx, ru, catalog = this, nss, ident = entry.ident, uuid]() {
-            // Intentionally ignoring failure
-            catalog->_engine->getEngine()->dropIdent(opCtx, ru, ident).ignore();
-        });
+    opCtx->recoveryUnit()->onRollback([ru, catalog = this, nss, ident = entry.ident, uuid]() {
+        // Intentionally ignoring failure
+        catalog->_engine->getEngine()->dropIdent(ru, ident).ignore();
+    });
 
     auto rs =
         _engine->getEngine()->getGroupedRecordStore(opCtx, nss.ns(), entry.ident, options, prefix);
@@ -976,12 +974,12 @@ StatusWith<DurableCatalog::ImportResult> DurableCatalogImpl::importCollection(
 
     auto ru = opCtx->recoveryUnit();
     opCtx->recoveryUnit()->onRollback(
-        [opCtx, ru, catalog = this, nss, ident = entry.ident, indexIdents = indexIdents]() {
+        [ru, catalog = this, nss, ident = entry.ident, indexIdents = indexIdents]() {
             // TODO SERVER-51146: dropIdent without removing the files.
             // Intentionally ignoring failure
-            catalog->_engine->getEngine()->dropIdent(opCtx, ru, ident).ignore();
+            catalog->_engine->getEngine()->dropIdent(ru, ident).ignore();
             for (const auto& indexIdent : indexIdents) {
-                catalog->_engine->getEngine()->dropIdent(opCtx, ru, indexIdent).ignore();
+                catalog->_engine->getEngine()->dropIdent(ru, indexIdent).ignore();
             }
         });
 
diff --git a/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_kv_engine.cpp b/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_kv_engine.cpp
index 23fe763b6f5..b5ecae6c0a3 100644
--- a/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_kv_engine.cpp
+++ b/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_kv_engine.cpp
@@ -162,9 +162,7 @@ std::unique_ptr<mongo::SortedDataInterface> KVEngine::getSortedDataInterface(
         return std::make_unique<SortedDataInterfaceStandard>(opCtx, ident, desc);
 }
 
-Status KVEngine::dropIdent(OperationContext* unusedOpCtx,
-                           mongo::RecoveryUnit* ru,
-                           StringData ident) {
+Status KVEngine::dropIdent(mongo::RecoveryUnit* ru, StringData ident) {
     Status dropStatus = Status::OK();
     stdx::unique_lock lock(_identsLock);
     if (_idents.count(ident.toString()) > 0) {
@@ -174,7 +172,7 @@ Status KVEngine::dropIdent(OperationContext* unusedOpCtx,
         lock.unlock();
         if (isRecordStore) {  // ident is RecordStore.
             CollectionOptions s;
-            auto rs = getRecordStore(/*unused*/ unusedOpCtx, ""_sd, ident, s);
+            auto rs = getRecordStore(nullptr, ""_sd, ident, s);
             dropStatus =
                 checked_cast<RecordStore*>(rs.get())->truncateWithoutUpdatingCount(ru).getStatus();
         } else {  // ident is SortedDataInterface.
diff --git a/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_kv_engine.h b/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_kv_engine.h
index 36b4d435e2e..59a59b96479 100644
--- a/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_kv_engine.h
+++ b/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_kv_engine.h
@@ -96,7 +96,7 @@ public:
 
     virtual void endBackup(OperationContext* opCtx) {}
 
-    virtual Status dropIdent(OperationContext* opCtx, mongo::RecoveryUnit* ru, StringData ident);
+    virtual Status dropIdent(mongo::RecoveryUnit* ru, StringData ident);
 
     virtual bool supportsDirectoryPerDB() const {
         return false;  // Not persistant so no Directories
diff --git a/src/mongo/db/storage/kv/kv_drop_pending_ident_reaper.cpp b/src/mongo/db/storage/kv/kv_drop_pending_ident_reaper.cpp
index 915bb7e0812..3d4437a07bf 100644
--- a/src/mongo/db/storage/kv/kv_drop_pending_ident_reaper.cpp
+++ b/src/mongo/db/storage/kv/kv_drop_pending_ident_reaper.cpp
@@ -46,7 +46,8 @@ KVDropPendingIdentReaper::KVDropPendingIdentReaper(KVEngine* engine) : _engine(e
 
 void KVDropPendingIdentReaper::addDropPendingIdent(const Timestamp& dropTimestamp,
                                                    const NamespaceString& nss,
-                                                   std::shared_ptr<Ident> ident) {
+                                                   std::shared_ptr<Ident> ident,
+                                                   const StorageEngine::DropIdentCallback& onDrop) {
     stdx::lock_guard<Latch> lock(_mutex);
     const auto equalRange = _dropPendingIdents.equal_range(dropTimestamp);
     const auto& lowerBound = equalRange.first;
@@ -57,16 +58,14 @@ void KVDropPendingIdentReaper::addDropPendingIdent(const Timestamp& dropTimestam
         info.nss = nss;
         info.identName = ident->getIdent();
         info.dropToken = ident;
+        info.onDrop = std::move(onDrop);
         _dropPendingIdents.insert(std::make_pair(dropTimestamp, info));
     } else {
-        LOGV2_FATAL_NOTRACE(
-            51023,
-            "Failed to add drop-pending ident {ident} ({namespace}) with drop timestamp "
-            "{dropTimestamp}: duplicate timestamp and ident pair.",
-            "Failed to add drop-pending ident, duplicate timestamp and ident pair",
-            "ident"_attr = ident->getIdent(),
-            "namespace"_attr = nss,
-            "dropTimestamp"_attr = dropTimestamp);
+        LOGV2_FATAL_NOTRACE(51023,
+                            "Failed to add drop-pending ident, duplicate timestamp and ident pair",
+                            "ident"_attr = ident->getIdent(),
+                            logAttrs(nss),
+                            "dropTimestamp"_attr = dropTimestamp);
     }
 }
 
@@ -120,26 +119,25 @@ void KVDropPendingIdentReaper::dropIdentsOlderThan(OperationContext* opCtx, cons
         const auto& nss = identInfo.nss;
         const auto& identName = identInfo.identName;
         LOGV2(22237,
-              "Completing drop for ident {ident} (ns: {namespace}) with drop timestamp "
-              "{dropTimestamp}",
               "Completing drop for ident",
               "ident"_attr = identName,
-              "namespace"_attr = nss,
+              logAttrs(nss),
               "dropTimestamp"_attr = dropTimestamp);
         WriteUnitOfWork wuow(opCtx);
-        auto status = _engine->dropIdent(opCtx, opCtx->recoveryUnit(), identName);
+        auto status = _engine->dropIdent(opCtx->recoveryUnit(), identName);
         if (!status.isOK()) {
-            LOGV2_FATAL_NOTRACE(
-                51022,
-                "Failed to remove drop-pending ident {ident}(ns: {namespace}) with drop "
-                "timestamp {dropTimestamp}: {error}",
-                "Failed to remove drop-pending ident",
-                "ident"_attr = identName,
-                "namespace"_attr = nss,
-                "dropTimestamp"_attr = dropTimestamp,
-                "error"_attr = status);
+            LOGV2_FATAL_NOTRACE(51022,
+                                "Failed to remove drop-pending ident",
+                                "ident"_attr = identName,
+                                logAttrs(nss),
+                                "dropTimestamp"_attr = dropTimestamp,
+                                "error"_attr = status);
         }
         wuow.commit();
+
+        if (identInfo.onDrop) {
+            identInfo.onDrop(nss);
+        }
     }
 
     {
diff --git a/src/mongo/db/storage/kv/kv_drop_pending_ident_reaper.h b/src/mongo/db/storage/kv/kv_drop_pending_ident_reaper.h
index 0a582501c99..ed42159e6b2 100644
--- a/src/mongo/db/storage/kv/kv_drop_pending_ident_reaper.h
+++ b/src/mongo/db/storage/kv/kv_drop_pending_ident_reaper.h
@@ -81,7 +81,8 @@ public:
      */
     void addDropPendingIdent(const Timestamp& dropTimestamp,
                              const NamespaceString& nss,
-                             std::shared_ptr<Ident> ident);
+                             std::shared_ptr<Ident> ident,
+                             const StorageEngine::DropIdentCallback& onDrop = nullptr);
 
     /**
      * Returns earliest drop timestamp in '_dropPendingIdents'.
@@ -120,6 +121,9 @@ private:
         // The collection or index data can be safely dropped when no references to this token
         // remain.
         std::weak_ptr<void> dropToken;
+
+        // Callback to run once the ident has been dropped.
+        StorageEngine::DropIdentCallback onDrop;
     };
 
     // Container type for drop-pending namespaces. We use a multimap so that we can order the
diff --git a/src/mongo/db/storage/kv/kv_drop_pending_ident_reaper_test.cpp b/src/mongo/db/storage/kv/kv_drop_pending_ident_reaper_test.cpp
index e475d7103c1..b2613b66515 100644
--- a/src/mongo/db/storage/kv/kv_drop_pending_ident_reaper_test.cpp
+++ b/src/mongo/db/storage/kv/kv_drop_pending_ident_reaper_test.cpp
@@ -46,7 +46,7 @@ namespace {
  */
 class KVEngineMock : public KVEngine {
 public:
-    Status dropIdent(OperationContext* opCtx, RecoveryUnit* ru, StringData ident) override;
+    Status dropIdent(RecoveryUnit* ru, StringData ident) override;
 
     // Unused KVEngine functions below.
     RecoveryUnit* newRecoveryUnit() override {
@@ -119,14 +119,12 @@ public:
     std::vector<std::string> droppedIdents;
 
     // Override to modify dropIdent() behavior.
-    using DropIdentFn = std::function<Status(OperationContext*, RecoveryUnit*, StringData)>;
-    DropIdentFn dropIdentFn = [](OperationContext*, RecoveryUnit*, StringData) {
-        return Status::OK();
-    };
+    using DropIdentFn = std::function<Status(RecoveryUnit*, StringData)>;
+    DropIdentFn dropIdentFn = [](RecoveryUnit*, StringData) { return Status::OK(); };
 };
 
-Status KVEngineMock::dropIdent(OperationContext* opCtx, RecoveryUnit* ru, StringData ident) {
-    auto status = dropIdentFn(opCtx, ru, ident);
+Status KVEngineMock::dropIdent(RecoveryUnit* ru, StringData ident) {
+    auto status = dropIdentFn(ru, ident);
     if (status.isOK()) {
         droppedIdents.push_back(ident.toString());
     }
@@ -380,13 +378,11 @@ DEATH_TEST_F(KVDropPendingIdentReaperTest,
     ASSERT_EQUALS(dropTimestamp, *reaper.getEarliestDropTimestamp());
 
     // Make KVEngineMock::dropIndent() fail.
-    engine->dropIdentFn =
-        [&identName](OperationContext* opCtx, RecoveryUnit* ru, StringData identToDropName) {
-            ASSERT(opCtx);
-            ASSERT(ru);
-            ASSERT_EQUALS(identName, identToDropName);
-            return Status(ErrorCodes::OperationFailed, "Mock KV engine dropIndent() failed.");
-        };
+    engine->dropIdentFn = [&identName](RecoveryUnit* ru, StringData identToDropName) {
+        ASSERT(ru);
+        ASSERT_EQUALS(identName, identToDropName);
+        return Status(ErrorCodes::OperationFailed, "Mock KV engine dropIndent() failed.");
+    };
 
     auto opCtx = makeOpCtx();
     reaper.dropIdentsOlderThan(opCtx.get(), makeTimestampWithNextInc(dropTimestamp));
diff --git a/src/mongo/db/storage/kv/kv_engine.h b/src/mongo/db/storage/kv/kv_engine.h
index baf705d15e5..5a4757ae847 100644
--- a/src/mongo/db/storage/kv/kv_engine.h
+++ b/src/mongo/db/storage/kv/kv_engine.h
@@ -210,14 +210,12 @@ public:
     virtual Status repairIdent(OperationContext* opCtx, StringData ident) = 0;
 
     /**
-     * Takes both an OperationContext and a RecoveryUnit, since most storage engines except for
-     * mobile only need the RecoveryUnit. Obtaining the RecoveryUnit from the OperationContext is
-     * not necessarily possible, since as of SERVER-44139 dropIdent can be called as part of
+     * Takes a RecoveryUnit. Obtaining the RecoveryUnit from the OperationContext is not
+     * necessarily possible, since as of SERVER-44139 dropIdent can be called as part of
      * multi-document transactions, which use the same RecoveryUnit throughout but not the same
      * OperationContext.
-     * TODO(SERVER-45371) Remove OperationContext argument.
      */
-    virtual Status dropIdent(OperationContext* opCtx, RecoveryUnit* ru, StringData ident) = 0;
+    virtual Status dropIdent(RecoveryUnit* ru, StringData ident) = 0;
 
     /**
      * Attempts to locate and recover a file that is "orphaned" from the storage engine's metadata,
diff --git a/src/mongo/db/storage/kv/kv_engine_test_harness.cpp b/src/mongo/db/storage/kv/kv_engine_test_harness.cpp
index fdc945a0eaf..1c916c31207 100644
--- a/src/mongo/db/storage/kv/kv_engine_test_harness.cpp
+++ b/src/mongo/db/storage/kv/kv_engine_test_harness.cpp
@@ -298,7 +298,7 @@ TEST(KVEngineTestHarness, TemporaryRecordStoreSimple) {
         ASSERT_EQUALS(ident, all[0]);
 
         WriteUnitOfWork wuow(&opCtx);
-        ASSERT_OK(engine->dropIdent(&opCtx, opCtx.recoveryUnit(), ident));
+        ASSERT_OK(engine->dropIdent(opCtx.recoveryUnit(), ident));
         wuow.commit();
     }
 }
diff --git a/src/mongo/db/storage/kv/storage_engine_test.cpp b/src/mongo/db/storage/kv/storage_engine_test.cpp
index 4a24337c281..9e68420730d 100644
--- a/src/mongo/db/storage/kv/storage_engine_test.cpp
+++ b/src/mongo/db/storage/kv/storage_engine_test.cpp
@@ -99,7 +99,7 @@ TEST_F(StorageEngineTest, ReconcileIdentsTest) {
     ASSERT_EQUALS("_id", toRebuild.indexName);
 
     // Now drop the `db.coll1` table, while leaving the DurableCatalog entry.
-    ASSERT_OK(dropIdent(opCtx.get(), opCtx.get()->recoveryUnit(), swCollInfo.getValue().ident));
+    ASSERT_OK(dropIdent(opCtx.get()->recoveryUnit(), swCollInfo.getValue().ident));
     ASSERT_EQUALS(static_cast<const unsigned long>(1), getAllKVEngineIdents(opCtx.get()).size());
 
     // Reconciling this should result in an error.
@@ -115,7 +115,7 @@ TEST_F(StorageEngineTest, LoadCatalogDropsOrphansAfterUncleanShutdown) {
     auto swCollInfo = createCollection(opCtx.get(), collNs);
     ASSERT_OK(swCollInfo.getStatus());
 
-    ASSERT_OK(dropIdent(opCtx.get(), opCtx.get()->recoveryUnit(), swCollInfo.getValue().ident));
+    ASSERT_OK(dropIdent(opCtx.get()->recoveryUnit(), swCollInfo.getValue().ident));
     ASSERT(collectionExists(opCtx.get(), collNs));
 
     // After the catalog is reloaded, we expect that the collection has been dropped because the
@@ -349,7 +349,7 @@ TEST_F(StorageEngineRepairTest, LoadCatalogRecoversOrphans) {
     auto swCollInfo = createCollection(opCtx.get(), collNs);
     ASSERT_OK(swCollInfo.getStatus());
 
-    ASSERT_OK(dropIdent(opCtx.get(), opCtx.get()->recoveryUnit(), swCollInfo.getValue().ident));
+    ASSERT_OK(dropIdent(opCtx.get()->recoveryUnit(), swCollInfo.getValue().ident));
     ASSERT(collectionExists(opCtx.get(), collNs));
 
     // After the catalog is reloaded, we expect that the ident has been recovered because the
@@ -374,7 +374,7 @@ TEST_F(StorageEngineRepairTest, ReconcileSucceeds) {
     auto swCollInfo = createCollection(opCtx.get(), collNs);
     ASSERT_OK(swCollInfo.getStatus());
 
-    ASSERT_OK(dropIdent(opCtx.get(), opCtx.get()->recoveryUnit(), swCollInfo.getValue().ident));
+    ASSERT_OK(dropIdent(opCtx.get()->recoveryUnit(), swCollInfo.getValue().ident));
     ASSERT(collectionExists(opCtx.get(), collNs));
 
     // Reconcile would normally return an error if a collection existed with a missing ident in the
diff --git a/src/mongo/db/storage/kv/temporary_kv_record_store.cpp b/src/mongo/db/storage/kv/temporary_kv_record_store.cpp
index d53a31a4b11..5e11f0e9f28 100644
--- a/src/mongo/db/storage/kv/temporary_kv_record_store.cpp
+++ b/src/mongo/db/storage/kv/temporary_kv_record_store.cpp
@@ -65,7 +65,7 @@ void TemporaryKVRecordStore::finalizeTemporaryTable(OperationContext* opCtx,
     // destructed while we're using it.
     invariant(opCtx->lockState()->isReadLocked());
 
-    auto status = _kvEngine->dropIdent(opCtx, opCtx->recoveryUnit(), _rs->getIdent());
+    auto status = _kvEngine->dropIdent(opCtx->recoveryUnit(), _rs->getIdent());
 
     if (!status.isOK()) {
         LOGV2_ERROR(4841503, "Failed to drop temporary table", "ident"_attr = _rs->getIdent());
diff --git a/src/mongo/db/storage/storage_engine.h b/src/mongo/db/storage/storage_engine.h
index d2bbca28a0d..6d5f88550cd 100644
--- a/src/mongo/db/storage/storage_engine.h
+++ b/src/mongo/db/storage/storage_engine.h
@@ -78,6 +78,8 @@ public:
     using OldestActiveTransactionTimestampCallback =
         std::function<OldestActiveTransactionTimestampResult(Timestamp stableTimestamp)>;
 
+    using DropIdentCallback = std::function<void(const NamespaceString& ns)>;
+
     /**
      * The interface for creating new instances of storage engines.
      *
@@ -472,7 +474,8 @@ public:
      */
     virtual void addDropPendingIdent(const Timestamp& dropTimestamp,
                                      const NamespaceString& nss,
-                                     std::shared_ptr<Ident> ident) = 0;
+                                     std::shared_ptr<Ident> ident,
+                                     const DropIdentCallback& onDrop = nullptr) = 0;
 
     /**
      * Called when the checkpoint thread instructs the storage engine to take a checkpoint. The
@@ -662,6 +665,8 @@ public:
 
     virtual int64_t sizeOnDiskForDb(OperationContext* opCtx, StringData dbName) = 0;
 
+    virtual bool isUsingDirectoryPerDb() const = 0;
+
     virtual KVEngine* getEngine() = 0;
     virtual const KVEngine* getEngine() const = 0;
     virtual DurableCatalog* getCatalog() = 0;
diff --git a/src/mongo/db/storage/storage_engine_impl.cpp b/src/mongo/db/storage/storage_engine_impl.cpp
index f90d2ea6415..4f3d0ab9b66 100644
--- a/src/mongo/db/storage/storage_engine_impl.cpp
+++ b/src/mongo/db/storage/storage_engine_impl.cpp
@@ -489,7 +489,7 @@ StatusWith<StorageEngine::ReconcileResult> StorageEngineImpl::reconcileCatalogAn
         const auto& toRemove = it;
         LOGV2(22251, "Dropping unknown ident", "ident"_attr = toRemove);
         WriteUnitOfWork wuow(opCtx);
-        fassert(40591, _engine->dropIdent(opCtx, opCtx->recoveryUnit(), toRemove));
+        fassert(40591, _engine->dropIdent(opCtx->recoveryUnit(), toRemove));
         wuow.commit();
     }
 
@@ -621,7 +621,7 @@ StatusWith<StorageEngine::ReconcileResult> StorageEngineImpl::reconcileCatalogAn
                       "namespace"_attr = coll,
                       "index"_attr = indexName);
                 // Ensure the `ident` is dropped while we have the `indexIdent` value.
-                fassert(50713, _engine->dropIdent(opCtx, opCtx->recoveryUnit(), indexIdent));
+                fassert(50713, _engine->dropIdent(opCtx->recoveryUnit(), indexIdent));
                 indexesToDrop.push_back(indexName);
                 continue;
             }
@@ -647,7 +647,7 @@ StatusWith<StorageEngine::ReconcileResult> StorageEngineImpl::reconcileCatalogAn
     for (auto&& temp : internalIdentsToDrop) {
         LOGV2(22257, "Dropping internal ident", "ident"_attr = temp);
         WriteUnitOfWork wuow(opCtx);
-        fassert(51067, _engine->dropIdent(opCtx, opCtx->recoveryUnit(), temp));
+        fassert(51067, _engine->dropIdent(opCtx->recoveryUnit(), temp));
         wuow.commit();
     }
 
@@ -1038,8 +1038,9 @@ void StorageEngineImpl::_dumpCatalog(OperationContext* opCtx) {
 
 void StorageEngineImpl::addDropPendingIdent(const Timestamp& dropTimestamp,
                                             const NamespaceString& nss,
-                                            std::shared_ptr<Ident> ident) {
-    _dropPendingIdentReaper.addDropPendingIdent(dropTimestamp, nss, ident);
+                                            std::shared_ptr<Ident> ident,
+                                            const DropIdentCallback& onDrop) {
+    _dropPendingIdentReaper.addDropPendingIdent(dropTimestamp, nss, ident, onDrop);
 }
 
 void StorageEngineImpl::checkpoint() {
diff --git a/src/mongo/db/storage/storage_engine_impl.h b/src/mongo/db/storage/storage_engine_impl.h
index 0efea5dc551..dbaf70a788f 100644
--- a/src/mongo/db/storage/storage_engine_impl.h
+++ b/src/mongo/db/storage/storage_engine_impl.h
@@ -316,7 +316,8 @@ public:
 
     void addDropPendingIdent(const Timestamp& dropTimestamp,
                              const NamespaceString& nss,
-                             std::shared_ptr<Ident> ident) override;
+                             std::shared_ptr<Ident> ident,
+                             const DropIdentCallback& onDrop) override;
 
     void checkpoint() override;
 
@@ -356,6 +357,10 @@ public:
 
     int64_t sizeOnDiskForDb(OperationContext* opCtx, StringData dbName) override;
 
+    bool isUsingDirectoryPerDb() const override {
+        return _options.directoryPerDB;
+    }
+
 private:
     using CollIter = std::list<std::string>::iterator;
 
diff --git a/src/mongo/db/storage/storage_engine_mock.h b/src/mongo/db/storage/storage_engine_mock.h
index 96eb8020b1d..d65863b2479 100644
--- a/src/mongo/db/storage/storage_engine_mock.h
+++ b/src/mongo/db/storage/storage_engine_mock.h
@@ -174,7 +174,8 @@ public:
     }
     void addDropPendingIdent(const Timestamp& dropTimestamp,
                              const NamespaceString& nss,
-                             std::shared_ptr<Ident> ident) final {}
+                             std::shared_ptr<Ident> ident,
+                             const DropIdentCallback& onDrop) final {}
     void checkpoint() final {}
     Status currentFilesCompatible(OperationContext* opCtx) const final {
         return Status::OK();
@@ -182,6 +183,9 @@ public:
     int64_t sizeOnDiskForDb(OperationContext* opCtx, StringData dbName) final {
         return 0;
     }
+    bool isUsingDirectoryPerDb() const final {
+        return false;
+    }
     KVEngine* getEngine() final {
         return nullptr;
     }
diff --git a/src/mongo/db/storage/storage_engine_test_fixture.h b/src/mongo/db/storage/storage_engine_test_fixture.h
index c6ad8880e77..cbbbac1053d 100644
--- a/src/mongo/db/storage/storage_engine_test_fixture.h
+++ b/src/mongo/db/storage/storage_engine_test_fixture.h
@@ -93,11 +93,11 @@ public:
             CollectionCatalog::get(opCtx).lookupCollectionByNamespace(opCtx, nss)->getCatalogId();
         std::string indexIdent =
             _storageEngine->getCatalog()->getIndexIdent(opCtx, catalogId, indexName);
-        return dropIdent(opCtx, opCtx->recoveryUnit(), indexIdent);
+        return dropIdent(opCtx->recoveryUnit(), indexIdent);
     }
 
-    Status dropIdent(OperationContext* opCtx, RecoveryUnit* ru, StringData ident) {
-        return _storageEngine->getEngine()->dropIdent(opCtx, ru, ident);
+    Status dropIdent(RecoveryUnit* ru, StringData ident) {
+        return _storageEngine->getEngine()->dropIdent(ru, ident);
     }
 
     StatusWith<StorageEngine::ReconcileResult> reconcile(OperationContext* opCtx) {
diff --git a/src/mongo/db/storage/storage_util.cpp b/src/mongo/db/storage/storage_util.cpp
index 50ec12b9094..5ba3c562542 100644
--- a/src/mongo/db/storage/storage_util.cpp
+++ b/src/mongo/db/storage/storage_util.cpp
@@ -33,8 +33,11 @@
 
 #include <string>
 
+#include <boost/filesystem/operations.hpp>
+
 #include "mongo/db/storage/storage_util.h"
 
+#include "mongo/db/catalog/collection_catalog.h"
 #include "mongo/db/storage/durable_catalog.h"
 #include "mongo/db/storage/ident.h"
 #include "mongo/db/storage/kv/kv_engine.h"
@@ -94,7 +97,7 @@ void removeIndex(OperationContext* opCtx,
             storageEngine->addDropPendingIdent(*commitTimestamp, nss, ident);
         } else {
             auto kvEngine = storageEngine->getEngine();
-            kvEngine->dropIdent(opCtx, recoveryUnit, ident->getIdent()).ignore();
+            kvEngine->dropIdent(recoveryUnit, ident->getIdent()).ignore();
         }
     });
 }
@@ -119,25 +122,52 @@ Status dropCollection(OperationContext* opCtx,
     // RecoveryUnit throughout but not the same OperationContext.
     auto recoveryUnit = opCtx->recoveryUnit();
     auto storageEngine = opCtx->getServiceContext()->getStorageEngine();
+    const auto& collectionCatalog = CollectionCatalog::get(opCtx);
 
     // Schedule the second phase of drop to delete the data when it is no longer in use, if the
     // first phase is successuflly committed.
-    opCtx->recoveryUnit()->onCommit([opCtx, recoveryUnit, storageEngine, nss, ident](
+    opCtx->recoveryUnit()->onCommit([recoveryUnit, storageEngine, &collectionCatalog, nss, ident](
                                         boost::optional<Timestamp> commitTimestamp) {
+        StorageEngine::DropIdentCallback onDrop = [storageEngine,
+                                                   &collectionCatalog](const NamespaceString& ns) {
+            // Nothing to do if not using directoryperdb or there are still collections in the
+            // database.
+            if (!storageEngine->isUsingDirectoryPerDb() ||
+                collectionCatalog.begin(nullptr, ns.db()) != collectionCatalog.end(nullptr)) {
+                return;
+            }
+
+            boost::system::error_code ec;
+            boost::filesystem::remove(storageEngine->getFilesystemPathForDb(ns.db().toString()),
+                                      ec);
+
+            if (!ec) {
+                LOGV2(4888200, "Removed empty database directory", "db"_attr = ns.db());
+            } else if (collectionCatalog.begin(nullptr, ns.db()) ==
+                       collectionCatalog.end(nullptr)) {
+                // It is possible for a new collection to be created in the database between
+                // when we check whether the database is empty and actually attempting to
+                // remove the directory. In this case, don't log that the removal failed
+                // because it is expected.
+                LOGV2(4888201,
+                      "Failed to remove database directory",
+                      "db"_attr = ns.db(),
+                      "error"_attr = ec.message());
+            }
+        };
+
         if (storageEngine->supportsPendingDrops() && commitTimestamp) {
             LOGV2(22214,
-                  "Deferring table drop for collection '{namespace}'. Ident: {ident}, "
-                  "commit timestamp: {commitTimestamp}",
                   "Deferring table drop for collection",
                   logAttrs(nss),
                   "ident"_attr = ident->getIdent(),
                   "commitTimestamp"_attr = commitTimestamp);
-            storageEngine->addDropPendingIdent(*commitTimestamp, nss, ident);
+            storageEngine->addDropPendingIdent(*commitTimestamp, nss, ident, onDrop);
         } else {
             // Intentionally ignoring failure here. Since we've removed the metadata pointing to
             // the collection, we should never see it again anyway.
-            auto kvEngine = storageEngine->getEngine();
-            kvEngine->dropIdent(opCtx, recoveryUnit, ident->getIdent()).ignore();
+            storageEngine->getEngine()->dropIdent(recoveryUnit, ident->getIdent()).ignore();
+            onDrop(nss);
         }
     });
 
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.cpp
index f56ce964c89..d9285dd2604 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.cpp
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.cpp
@@ -1608,7 +1608,7 @@ std::unique_ptr<RecordStore> WiredTigerKVEngine::makeTemporaryRecordStore(Operat
     return std::move(rs);
 }
 
-Status WiredTigerKVEngine::dropIdent(OperationContext* opCtx, RecoveryUnit* ru, StringData ident) {
+Status WiredTigerKVEngine::dropIdent(RecoveryUnit* ru, StringData ident) {
     string uri = _uri(ident);
 
     WiredTigerRecoveryUnit* wtRu = checked_cast<WiredTigerRecoveryUnit*>(ru);
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.h b/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.h
index edf4930bcda..88dea33f614 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.h
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine.h
@@ -201,7 +201,7 @@ public:
                                                                        const IndexDescriptor* desc,
                                                                        KVPrefix prefix) override;
 
-    Status dropIdent(OperationContext* opCtx, RecoveryUnit* ru, StringData ident) override;
+    Status dropIdent(RecoveryUnit* ru, StringData ident) override;
 
     Status okToRename(OperationContext* opCtx,
                       StringData fromNS,
diff --git a/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine_test.cpp b/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine_test.cpp
index 2580960a76c..0710f443649 100644
--- a/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine_test.cpp
+++ b/src/mongo/db/storage/wiredtiger/wiredtiger_kv_engine_test.cpp
@@ -164,7 +164,7 @@ TEST_F(WiredTigerKVEngineRepairTest, OrphanedDataFilesCanBeRecovered) {
     boost::filesystem::rename(*dataFilePath, tmpFile, err);
     ASSERT(!err) << err.message();
 
-    ASSERT_OK(_engine->dropIdent(opCtxPtr.get(), opCtxPtr.get()->recoveryUnit(), ident));
+    ASSERT_OK(_engine->dropIdent(opCtxPtr.get()->recoveryUnit(), ident));
 
     // The data file is moved back in place so that it becomes an "orphan" of the storage
     // engine and the restoration process can be tested.
@@ -207,7 +207,7 @@ TEST_F(WiredTigerKVEngineRepairTest, UnrecoverableOrphanedDataFilesAreRebuilt) {
 
     ASSERT(boost::filesystem::exists(*dataFilePath));
 
-    ASSERT_OK(_engine->dropIdent(opCtxPtr.get(), opCtxPtr.get()->recoveryUnit(), ident));
+    ASSERT_OK(_engine->dropIdent(opCtxPtr.get()->recoveryUnit(), ident));
 
 #ifdef _WIN32
     auto status =