Update upstream source from tag 'upstream/6.0.10'debian/6.0.10-1

Update to upstream version '6.0.10'
with Debian dir 2d176fa254eee97b139f712fec5709641335a8c3

64 files changed, 1835 insertions, 658 deletions

diff --git a/src/mongo/db/catalog/README.md b/src/mongo/db/catalog/README.md
index ddb5d479da2..5fe1a00de68 100644
--- a/src/mongo/db/catalog/README.md
+++ b/src/mongo/db/catalog/README.md
@@ -432,6 +432,30 @@ See
 [wtRcToStatus](https://github.com/mongodb/mongo/blob/c799851554dc01493d35b43701416e9c78b3665c/src/mongo/db/storage/wiredtiger/wiredtiger_util.cpp#L178-L183)
 where we throw the exception in WiredTiger.
 See [TemporarilyUnavailableException](https://github.com/mongodb/mongo/blob/c799851554dc01493d35b43701416e9c78b3665c/src/mongo/db/concurrency/temporarily_unavailable_exception.h#L39-L45).
+
+## TransactionTooLargeForCacheException
+
+A TransactionTooLargeForCacheException may be thrown inside the server to indicate that an operation
+was rolled-back and is unlikely to ever complete because the storage engine cache is insufficient,
+even in the absence of concurrent operations. This is determined by a simple heuristic wherein,
+after a rollback, a threshold on the proportion of total dirty cache bytes the running transaction
+can represent and still be considered fullfillable is checked. The threshold can be tuned with the
+`transactionTooLargeForCacheThreshold` parameter. Setting this threshold to its maximum value (1.0)
+causes the check to be skipped and TransactionTooLargeForCacheException to be disabled.
+
+On replica sets, if an operation succeeds on a primary, it should also succeed on a secondary. It
+would be possible to convert to both TemporarilyUnavailableException and WriteConflictException,
+as if TransactionTooLargeForCacheException was disabled. But on secondaries the only
+difference between the two is the rate at which the operation is retried. Hence,
+TransactionTooLargeForCacheException is always converted to a WriteConflictException, which retries
+faster, to avoid stalling replication longer than necessary.
+
+Prior to 6.3, or when TransactionTooLargeForCacheException is disabled, multi-document
+transactions always return a WriteConflictException, which may result in drivers retrying  an
+operation indefinitely. For non-multi-document operations, there is a limited number of retries on
+TemporarilyUnavailableException, but it might still be beneficial to not retry operations which are
+unlikely to complete and are disruptive for concurrent operations.
+
 ## Collection and Index Writes
 
 Collection write operations (inserts, updates, and deletes) perform storage engine writes to both
diff --git a/src/mongo/db/catalog/SConscript b/src/mongo/db/catalog/SConscript
index b116e2ad284..0a22459bc19 100644
--- a/src/mongo/db/catalog/SConscript
+++ b/src/mongo/db/catalog/SConscript
@@ -121,7 +121,7 @@ env.Library(
     ],
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/catalog_raii',
-        '$BUILD_DIR/mongo/db/concurrency/write_conflict_exception',
+        '$BUILD_DIR/mongo/db/concurrency/exception_util',
         '$BUILD_DIR/mongo/db/curop',
         '$BUILD_DIR/mongo/db/index/index_access_method',
         '$BUILD_DIR/mongo/db/storage/key_string',
@@ -137,7 +137,7 @@ env.Library(
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/base',
         '$BUILD_DIR/mongo/db/catalog_raii',
-        '$BUILD_DIR/mongo/db/concurrency/write_conflict_exception',
+        '$BUILD_DIR/mongo/db/concurrency/exception_util',
         '$BUILD_DIR/mongo/db/curop',
         '$BUILD_DIR/mongo/db/query/query_knobs',
         '$BUILD_DIR/mongo/db/storage/record_store_base',
@@ -169,8 +169,7 @@ env.Library(
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/base',
         '$BUILD_DIR/mongo/db/common',
-        '$BUILD_DIR/mongo/db/index/index_descriptor',
-        '$BUILD_DIR/mongo/db/index/key_generator',
+        '$BUILD_DIR/mongo/db/index/index_access_method',
         '$BUILD_DIR/mongo/db/index_names',
         '$BUILD_DIR/mongo/db/matcher/expressions',
         '$BUILD_DIR/mongo/db/query/collation/collator_factory_interface',
@@ -210,7 +209,6 @@ env.Library(
         '$BUILD_DIR/mongo/db/collection_index_usage_tracker',
         '$BUILD_DIR/mongo/db/common',
         '$BUILD_DIR/mongo/db/index/index_access_method',
-        '$BUILD_DIR/mongo/db/index/index_descriptor',
         '$BUILD_DIR/mongo/db/index_names',
         '$BUILD_DIR/mongo/db/repl/repl_coordinator_interface',
         '$BUILD_DIR/mongo/db/ttl_collection_cache',
@@ -253,7 +251,7 @@ env.Library(
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/base',
         '$BUILD_DIR/mongo/db/catalog_raii',
-        '$BUILD_DIR/mongo/db/concurrency/write_conflict_exception',
+        '$BUILD_DIR/mongo/db/concurrency/exception_util',
         '$BUILD_DIR/mongo/db/curop',
         '$BUILD_DIR/mongo/db/index/index_access_method',
         '$BUILD_DIR/mongo/db/repl/repl_coordinator_interface',
@@ -283,7 +281,6 @@ env.Library(
         'views_for_database.cpp',
     ],
     LIBDEPS_PRIVATE=[
-        '$BUILD_DIR/mongo/db/concurrency/write_conflict_exception',
         '$BUILD_DIR/mongo/db/multitenancy',
         '$BUILD_DIR/mongo/db/namespace_string',
         '$BUILD_DIR/mongo/db/profile_filter',
@@ -320,6 +317,8 @@ env.Library(
         '$BUILD_DIR/mongo/db/multitenancy',
         '$BUILD_DIR/mongo/db/rebuild_indexes',
         '$BUILD_DIR/mongo/db/service_context',
+        '$BUILD_DIR/mongo/db/timeseries/timeseries_extended_range',
+        'catalog_stats',
         'collection',
         'collection_catalog',
         'database_holder',
@@ -363,8 +362,6 @@ env.Library(
         '$BUILD_DIR/mongo/db/curop',
         '$BUILD_DIR/mongo/db/db_raii',
         '$BUILD_DIR/mongo/db/index/index_access_method',
-        '$BUILD_DIR/mongo/db/index/index_access_method_factory',
-        '$BUILD_DIR/mongo/db/index/index_access_methods',
         '$BUILD_DIR/mongo/db/multitenancy',
         '$BUILD_DIR/mongo/db/op_observer',
         '$BUILD_DIR/mongo/db/record_id_helpers',
@@ -383,6 +380,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/storage/storage_util',
         '$BUILD_DIR/mongo/db/system_index',
         '$BUILD_DIR/mongo/db/timeseries/timeseries_conversion_util',
+        '$BUILD_DIR/mongo/db/timeseries/timeseries_extended_range',
         '$BUILD_DIR/mongo/db/transaction',
         '$BUILD_DIR/mongo/db/ttl_collection_cache',
         '$BUILD_DIR/mongo/db/vector_clock',
@@ -432,7 +430,7 @@ env.Library(
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/base',
         '$BUILD_DIR/mongo/bson/util/bson_column',
-        '$BUILD_DIR/mongo/db/concurrency/write_conflict_exception',
+        '$BUILD_DIR/mongo/db/concurrency/exception_util',
         '$BUILD_DIR/mongo/db/curop',
         '$BUILD_DIR/mongo/db/db_raii',
         '$BUILD_DIR/mongo/db/index/index_access_method',
@@ -499,6 +497,7 @@ env.Library(
     ],
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/base',
+        '$BUILD_DIR/mongo/db/concurrency/exception_util',
         '$BUILD_DIR/mongo/db/db_raii',
         '$BUILD_DIR/mongo/db/index/index_access_method',
         '$BUILD_DIR/mongo/db/index_builds_coordinator_interface',
@@ -610,6 +609,7 @@ if wiredtiger:
             'collection_test.cpp',
             'collection_validation_test.cpp',
             'collection_writer_test.cpp',
+            'coll_mod_test.cpp',
             'commit_quorum_options_test.cpp',
             'create_collection_test.cpp',
             'database_test.cpp',
diff --git a/src/mongo/db/catalog/capped_utils.cpp b/src/mongo/db/catalog/capped_utils.cpp
index 945881e58c5..bd841f09345 100644
--- a/src/mongo/db/catalog/capped_utils.cpp
+++ b/src/mongo/db/catalog/capped_utils.cpp
@@ -43,7 +43,7 @@
 #include "mongo/db/catalog/rename_collection.h"
 #include "mongo/db/catalog_raii.h"
 #include "mongo/db/client.h"
-#include "mongo/db/concurrency/write_conflict_exception.h"
+#include "mongo/db/concurrency/exception_util.h"
 #include "mongo/db/curop.h"
 #include "mongo/db/index_builds_coordinator.h"
 #include "mongo/db/namespace_string.h"
@@ -254,7 +254,7 @@ void cloneCollectionAsCapped(OperationContext* opCtx,
         } catch (const WriteConflictException&) {
             CurOp::get(opCtx)->debug().additiveMetrics.incrementWriteConflicts(1);
             retries++;  // logAndBackoff expects this to be 1 on first call.
-            WriteConflictException::logAndBackoff(retries, "cloneCollectionAsCapped", fromNss.ns());
+            logWriteConflictAndBackoff(retries, "cloneCollectionAsCapped", fromNss.ns());
 
             // Can't use writeConflictRetry since we need to save/restore exec around call to
             // abandonSnapshot.
diff --git a/src/mongo/db/catalog/catalog_control.cpp b/src/mongo/db/catalog/catalog_control.cpp
index dadfed598c0..60d59076d2c 100644
--- a/src/mongo/db/catalog/catalog_control.cpp
+++ b/src/mongo/db/catalog/catalog_control.cpp
@@ -35,6 +35,7 @@
 
 #include "mongo/db/catalog/catalog_control.h"
 
+#include "mongo/db/catalog/catalog_stats.h"
 #include "mongo/db/catalog/collection.h"
 #include "mongo/db/catalog/collection_catalog.h"
 #include "mongo/db/catalog/database.h"
@@ -43,17 +44,18 @@
 #include "mongo/db/index_builds_coordinator.h"
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/rebuild_indexes.h"
+#include "mongo/db/repl/oplog.h"
 #include "mongo/db/tenant_database_name.h"
+#include "mongo/db/timeseries/timeseries_extended_range.h"
 #include "mongo/logv2/log.h"
 
 namespace mongo {
 namespace catalog {
 namespace {
-void reopenAllDatabasesAndReloadCollectionCatalog(
-    OperationContext* opCtx,
-    StorageEngine* storageEngine,
-    const MinVisibleTimestampMap& minVisibleTimestampMap,
-    Timestamp stableTimestamp) {
+void reopenAllDatabasesAndReloadCollectionCatalog(OperationContext* opCtx,
+                                                  StorageEngine* storageEngine,
+                                                  const PreviousCatalogState& previousCatalogState,
+                                                  Timestamp stableTimestamp) {
 
     // Open all databases and repopulate the CollectionCatalog.
     LOGV2(20276, "openCatalog: reopening all databases");
@@ -80,7 +82,7 @@ void reopenAllDatabasesAndReloadCollectionCatalog(
                       str::stream()
                           << "failed to get valid collection pointer for namespace " << collNss);
 
-            if (minVisibleTimestampMap.count(collection->uuid()) > 0) {
+            if (previousCatalogState.minVisibleTimestampMap.count(collection->uuid()) > 0) {
                 // After rolling back to a stable timestamp T, the minimum visible timestamp for
                 // each collection must be reset to (at least) its value at T. Additionally, there
                 // cannot exist a minimum visible timestamp greater than lastApplied. This allows us
@@ -90,14 +92,31 @@ void reopenAllDatabasesAndReloadCollectionCatalog(
                 // bound the minimum visible timestamp (where necessary) to the stable timestamp.
                 // The benefit of fine grained tracking is assumed to be low-value compared to the
                 // cost/effort.
-                auto minVisible = std::min(stableTimestamp,
-                                           minVisibleTimestampMap.find(collection->uuid())->second);
+                auto minVisible = std::min(
+                    stableTimestamp,
+                    previousCatalogState.minVisibleTimestampMap.find(collection->uuid())->second);
                 auto writableCollection =
                     catalogWriter.get()->lookupCollectionByUUIDForMetadataWrite(opCtx,
                                                                                 collection->uuid());
                 writableCollection->setMinimumVisibleSnapshot(minVisible);
             }
 
+            if (collection->getTimeseriesOptions()) {
+                bool extendedRangeSetting;
+                if (auto it = previousCatalogState.requiresTimestampExtendedRangeSupportMap.find(
+                        collection->uuid());
+                    it != previousCatalogState.requiresTimestampExtendedRangeSupportMap.end()) {
+                    extendedRangeSetting = it->second;
+                } else {
+                    extendedRangeSetting =
+                        timeseries::collectionMayRequireExtendedRangeSupport(opCtx, collection);
+                }
+
+                if (extendedRangeSetting) {
+                    collection->setRequiresTimeseriesExtendedRangeSupport(opCtx);
+                }
+            }
+
             // If this is the oplog collection, re-establish the replication system's cached pointer
             // to the oplog.
             if (collNss.isOplog()) {
@@ -114,19 +133,18 @@ void reopenAllDatabasesAndReloadCollectionCatalog(
 
     // Opening CollectionCatalog: The collection catalog is now in sync with the storage engine
     // catalog. Clear the pre-closing state.
-    CollectionCatalog::write(opCtx,
-                             [&](CollectionCatalog& catalog) { catalog.onOpenCatalog(opCtx); });
+    CollectionCatalog::write(opCtx, [](CollectionCatalog& catalog) { catalog.onOpenCatalog(); });
     opCtx->getServiceContext()->incrementCatalogGeneration();
     LOGV2(20278, "openCatalog: finished reloading collection catalog");
 }
 }  // namespace
 
-MinVisibleTimestampMap closeCatalog(OperationContext* opCtx) {
+PreviousCatalogState closeCatalog(OperationContext* opCtx) {
     invariant(opCtx->lockState()->isW());
 
     IndexBuildsCoordinator::get(opCtx)->assertNoIndexBuildInProgress();
 
-    MinVisibleTimestampMap minVisibleTimestampMap;
+    PreviousCatalogState previousCatalogState;
     std::vector<TenantDatabaseName> allDbs =
         opCtx->getServiceContext()->getStorageEngine()->listDatabases();
 
@@ -150,7 +168,12 @@ MinVisibleTimestampMap closeCatalog(OperationContext* opCtx) {
                             "coll_ns"_attr = coll->ns(),
                             "uuid"_attr = coll->uuid(),
                             "minVisible"_attr = minVisible);
-                minVisibleTimestampMap[coll->uuid()] = *minVisible;
+                previousCatalogState.minVisibleTimestampMap[coll->uuid()] = *minVisible;
+            }
+
+            if (coll->getTimeseriesOptions()) {
+                previousCatalogState.requiresTimestampExtendedRangeSupportMap[coll->uuid()] =
+                    coll->getRequiresTimeseriesExtendedRangeSupport();
             }
         }
     }
@@ -158,14 +181,13 @@ MinVisibleTimestampMap closeCatalog(OperationContext* opCtx) {
     // Need to mark the CollectionCatalog as open if we our closeAll fails, dismissed if successful.
     ScopeGuard reopenOnFailure([opCtx] {
         CollectionCatalog::write(opCtx,
-                                 [&](CollectionCatalog& catalog) { catalog.onOpenCatalog(opCtx); });
+                                 [](CollectionCatalog& catalog) { catalog.onOpenCatalog(); });
     });
     // Closing CollectionCatalog: only lookupNSSByUUID will fall back to using pre-closing state to
     // allow authorization for currently unknown UUIDs. This is needed because authorization needs
     // to work before acquiring locks, and might otherwise spuriously regard a UUID as unknown
     // while reloading the catalog.
-    CollectionCatalog::write(opCtx,
-                             [&](CollectionCatalog& catalog) { catalog.onCloseCatalog(opCtx); });
+    CollectionCatalog::write(opCtx, [](CollectionCatalog& catalog) { catalog.onCloseCatalog(); });
 
     LOGV2_DEBUG(20270, 1, "closeCatalog: closing collection catalog");
 
@@ -177,12 +199,16 @@ MinVisibleTimestampMap closeCatalog(OperationContext* opCtx) {
     LOGV2(20272, "closeCatalog: closing storage engine catalog");
     opCtx->getServiceContext()->getStorageEngine()->closeCatalog(opCtx);
 
+    // Reset the stats counter for extended range time-series collections. This is maintained
+    // outside the catalog itself.
+    catalog_stats::requiresTimeseriesExtendedRangeSupport.store(0);
+
     reopenOnFailure.dismiss();
-    return minVisibleTimestampMap;
+    return previousCatalogState;
 }
 
 void openCatalog(OperationContext* opCtx,
-                 const MinVisibleTimestampMap& minVisibleTimestampMap,
+                 const PreviousCatalogState& previousCatalogState,
                  Timestamp stableTimestamp) {
     invariant(opCtx->lockState()->isW());
 
@@ -254,7 +280,7 @@ void openCatalog(OperationContext* opCtx,
         opCtx, reconcileResult.indexBuildsToRestart, reconcileResult.indexBuildsToResume);
 
     reopenAllDatabasesAndReloadCollectionCatalog(
-        opCtx, storageEngine, minVisibleTimestampMap, stableTimestamp);
+        opCtx, storageEngine, previousCatalogState, stableTimestamp);
 }
 
 
diff --git a/src/mongo/db/catalog/catalog_control.h b/src/mongo/db/catalog/catalog_control.h
index 182f073c307..f7bc3fbbdba 100644
--- a/src/mongo/db/catalog/catalog_control.h
+++ b/src/mongo/db/catalog/catalog_control.h
@@ -36,6 +36,11 @@ namespace catalog {
 
 using MinVisibleTimestamp = Timestamp;
 using MinVisibleTimestampMap = std::map<UUID, MinVisibleTimestamp>;
+using RequiresTimestampExtendedRangeSupportMap = std::map<UUID, bool>;
+struct PreviousCatalogState {
+    MinVisibleTimestampMap minVisibleTimestampMap;
+    RequiresTimestampExtendedRangeSupportMap requiresTimestampExtendedRangeSupportMap;
+};
 
 /**
  * Closes the catalog, destroying all associated in-memory data structures for all databases. After
@@ -43,7 +48,7 @@ using MinVisibleTimestampMap = std::map<UUID, MinVisibleTimestamp>;
  *
  * Must be called with the global lock acquired in exclusive mode.
  */
-MinVisibleTimestampMap closeCatalog(OperationContext* opCtx);
+PreviousCatalogState closeCatalog(OperationContext* opCtx);
 
 /**
  * Restores the catalog and all in-memory state after a call to closeCatalog().
@@ -51,7 +56,7 @@ MinVisibleTimestampMap closeCatalog(OperationContext* opCtx);
  * Must be called with the global lock acquired in exclusive mode.
  */
 void openCatalog(OperationContext* opCtx,
-                 const MinVisibleTimestampMap& catalogState,
+                 const PreviousCatalogState& catalogState,
                  Timestamp stableTimestamp);
 
 /**
diff --git a/src/mongo/db/catalog/catalog_control_test.cpp b/src/mongo/db/catalog/catalog_control_test.cpp
index 4eca3df6a9a..901c69450ee 100644
--- a/src/mongo/db/catalog/catalog_control_test.cpp
+++ b/src/mongo/db/catalog/catalog_control_test.cpp
@@ -75,8 +75,8 @@ TEST_F(CatalogControlTest, CloseAndOpenCatalog) {
     ServiceContext::UniqueOperationContext opCtx = cc().makeOperationContext();
     Lock::GlobalLock globalLk(opCtx.get(), MODE_X);
 
-    auto map = catalog::closeCatalog(opCtx.get());
-    ASSERT_EQUALS(0U, map.size());
+    auto previousState = catalog::closeCatalog(opCtx.get());
+    ASSERT_EQUALS(0U, previousState.minVisibleTimestampMap.size());
     catalog::openCatalog(opCtx.get(), {}, Timestamp());
 }
 
diff --git a/src/mongo/db/catalog/catalog_stats.cpp b/src/mongo/db/catalog/catalog_stats.cpp
index ea7b4ee2276..d346e67f971 100644
--- a/src/mongo/db/catalog/catalog_stats.cpp
+++ b/src/mongo/db/catalog/catalog_stats.cpp
@@ -29,7 +29,7 @@
 
 #define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kCommand
 
-#include "mongo/platform/basic.h"
+#include "mongo/db/catalog/catalog_stats.h"
 
 #include "mongo/db/catalog/collection_catalog.h"
 #include "mongo/db/catalog/database_holder.h"
@@ -37,7 +37,10 @@
 #include "mongo/db/db_raii.h"
 #include "mongo/logv2/log.h"
 
-namespace mongo {
+namespace mongo::catalog_stats {
+
+// Number of time-series collections requiring extended range support
+AtomicWord<int> requiresTimeseriesExtendedRangeSupport;
 
 namespace {
 class CatalogStatsSSS : public ServerStatusSection {
@@ -58,6 +61,7 @@ public:
         int timeseries = 0;
         int internalCollections = 0;
         int internalViews = 0;
+        int timeseriesExtendedRange = 0;
 
         void toBson(BSONObjBuilder* builder) const {
             builder->append("collections", collections);
@@ -67,6 +71,9 @@ public:
             builder->append("views", views);
             builder->append("internalCollections", internalCollections);
             builder->append("internalViews", internalViews);
+            if (timeseriesExtendedRange > 0) {
+                builder->append("timeseriesExtendedRange", timeseriesExtendedRange);
+            }
         }
     };
 
@@ -80,6 +87,7 @@ public:
         stats.capped = catalogStats.userCapped;
         stats.clustered = catalogStats.userClustered;
         stats.internalCollections = catalogStats.internal;
+        stats.timeseriesExtendedRange = requiresTimeseriesExtendedRangeSupport.load();
 
         const auto viewCatalogDbNames = catalog->getViewCatalogDbNames(opCtx);
         for (const auto& tenantDbName : viewCatalogDbNames) {
@@ -106,4 +114,4 @@ public:
 
 } catalogStatsSSS;
 }  // namespace
-}  // namespace mongo
+}  // namespace mongo::catalog_stats
diff --git a/src/mongo/db/catalog/catalog_stats.h b/src/mongo/db/catalog/catalog_stats.h
new file mode 100644
index 00000000000..54cc04f0bb6
--- /dev/null
+++ b/src/mongo/db/catalog/catalog_stats.h
@@ -0,0 +1,38 @@
+/**
+ *    Copyright (C) 2022-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#pragma once
+
+#include "mongo/platform/atomic_word.h"
+
+namespace mongo::catalog_stats {
+
+extern AtomicWord<int> requiresTimeseriesExtendedRangeSupport;
+
+}  // namespace mongo::catalog_stats
diff --git a/src/mongo/db/catalog/coll_mod.cpp b/src/mongo/db/catalog/coll_mod.cpp
index 8af0ba8efc9..9e92ed1d260 100644
--- a/src/mongo/db/catalog/coll_mod.cpp
+++ b/src/mongo/db/catalog/coll_mod.cpp
@@ -33,6 +33,7 @@
 
 #include "mongo/db/catalog/coll_mod.h"
 
+#include "mongo/db/stats/counters.h"
 #include <boost/optional.hpp>
 #include <memory>
 
@@ -44,7 +45,7 @@
 #include "mongo/db/catalog/index_catalog.h"
 #include "mongo/db/catalog/index_key_validate.h"
 #include "mongo/db/coll_mod_gen.h"
-#include "mongo/db/concurrency/write_conflict_exception.h"
+#include "mongo/db/concurrency/exception_util.h"
 #include "mongo/db/curop_failpoint_helpers.h"
 #include "mongo/db/db_raii.h"
 #include "mongo/db/index/index_descriptor.h"
@@ -53,6 +54,7 @@
 #include "mongo/db/repl/replication_coordinator.h"
 #include "mongo/db/s/collection_sharding_state.h"
 #include "mongo/db/s/database_sharding_state.h"
+#include "mongo/db/s/shard_key_index_util.h"
 #include "mongo/db/server_options.h"
 #include "mongo/db/service_context.h"
 #include "mongo/db/storage/recovery_unit.h"
@@ -62,6 +64,7 @@
 #include "mongo/db/views/view_catalog_helpers.h"
 #include "mongo/idl/command_generic_argument.h"
 #include "mongo/logv2/log.h"
+#include "mongo/s/grid.h"
 #include "mongo/util/fail_point.h"
 #include "mongo/util/version/releases.h"
 #include "mongo/util/visit_helper.h"
@@ -109,7 +112,7 @@ struct ParsedCollModRequest {
     boost::optional<Collection::Validator> collValidator;
     boost::optional<ValidationActionEnum> collValidationAction;
     boost::optional<ValidationLevelEnum> collValidationLevel;
-    bool recordPreImages = false;
+    boost::optional<bool> recordPreImages;
     boost::optional<ChangeStreamPreAndPostImagesOptions> changeStreamPreAndPostImagesOptions;
     int numModifications = 0;
     bool dryRun = false;
@@ -250,7 +253,8 @@ StatusWith<std::pair<ParsedCollModRequest, BSONObj>> parseCollModRequest(Operati
                         "TTL indexes are not supported for capped collections."};
             }
             if (auto status = index_key_validate::validateExpireAfterSeconds(
-                    *cmdIndex.getExpireAfterSeconds());
+                    *cmdIndex.getExpireAfterSeconds(),
+                    index_key_validate::ValidateExpireAfterSecondsMode::kSecondaryTTLIndex);
                 !status.isOK()) {
                 return {ErrorCodes::InvalidOptions, status.reason()};
             }
@@ -284,7 +288,8 @@ StatusWith<std::pair<ParsedCollModRequest, BSONObj>> parseCollModRequest(Operati
             }
         } else {
             std::vector<const IndexDescriptor*> indexes;
-            coll->getIndexCatalog()->findIndexesByKeyPattern(opCtx, keyPattern, false, &indexes);
+            coll->getIndexCatalog()->findIndexesByKeyPattern(
+                opCtx, keyPattern, IndexCatalog::InclusionPolicy::kReady, &indexes);
 
             if (indexes.size() > 1) {
                 return {ErrorCodes::AmbiguousIndexKeyPattern,
@@ -337,7 +342,7 @@ StatusWith<std::pair<ParsedCollModRequest, BSONObj>> parseCollModRequest(Operati
             if (cmrIndex->idx->unique()) {
                 indexForOplog->setUnique(boost::none);
             } else {
-                // Disallow one-step unique convertion. The user has to set
+                // Disallow one-step unique conversion. The user has to set
                 // 'prepareUnique' to true first.
                 if (!cmrIndex->idx->prepareUnique()) {
                     return Status(ErrorCodes::InvalidOptions,
@@ -363,6 +368,28 @@ StatusWith<std::pair<ParsedCollModRequest, BSONObj>> parseCollModRequest(Operati
                 return {ErrorCodes::BadValue, "can't hide _id index"};
             }
 
+            // If the index is not hidden and we are trying to hide it, check if it is possible
+            // to drop the shard key index, so it could be possible to hide it.
+            if (!cmrIndex->idx->hidden() && *cmdIndex.getHidden()) {
+                if (auto catalogClient = Grid::get(opCtx)->catalogClient()) {
+                    try {
+                        auto shardedColl = catalogClient->getCollection(opCtx, nss);
+
+                        if (isLastNonHiddenShardKeyIndex(opCtx,
+                                                         coll,
+                                                         coll->getIndexCatalog(),
+                                                         cmrIndex->idx->indexName(),
+                                                         shardedColl.getKeyPattern().toBSON())) {
+                            return {ErrorCodes::InvalidOptions,
+                                    "Can't hide the only compatible index for this collection's "
+                                    "shard key"};
+                        }
+                    } catch (ExceptionFor<ErrorCodes::NamespaceNotFound>&) {
+                        // The collection is unsharded or doesn't exist.
+                    }
+                }
+            }
+
             // Hiding a hidden index or unhiding a visible index should be treated as a no-op.
             if (cmrIndex->idx->hidden() == *cmdIndex.getHidden()) {
                 indexForOplog->setHidden(boost::none);
@@ -378,6 +405,24 @@ StatusWith<std::pair<ParsedCollModRequest, BSONObj>> parseCollModRequest(Operati
                 cmrIndex->idx->unique()) {
                 indexForOplog->setPrepareUnique(boost::none);
             } else {
+                // Checks if the index key pattern conflicts with the shard key pattern.
+                if (auto catalogClient = Grid::get(opCtx)->catalogClient()) {
+                    try {
+                        auto shardedColl = catalogClient->getCollection(opCtx, nss);
+                        const ShardKeyPattern shardKeyPattern(shardedColl.getKeyPattern());
+                        if (!shardKeyPattern.isIndexUniquenessCompatible(
+                                cmrIndex->idx->keyPattern())) {
+                            return {ErrorCodes::InvalidOptions,
+                                    fmt::format(
+                                        "cannot set 'prepareUnique' for index {} with shard key "
+                                        "pattern {}",
+                                        cmrIndex->idx->keyPattern().toString(),
+                                        shardKeyPattern.toBSON().toString())};
+                        }
+                    } catch (ExceptionFor<ErrorCodes::NamespaceNotFound>&) {
+                        // The collection is unsharded or doesn't exist.
+                    }
+                }
                 cmrIndex->indexPrepareUnique = cmdIndex.getPrepareUnique();
             }
         }
@@ -430,6 +475,11 @@ StatusWith<std::pair<ParsedCollModRequest, BSONObj>> parseCollModRequest(Operati
                                                     validatorObj.getOwned(),
                                                     MatchExpressionParser::kDefaultSpecialFeatures,
                                                     maxFeatureCompatibilityVersion);
+
+        // Increment counters to track the usage of schema validators.
+        validatorCounters.incrementCounters(
+            cmd.kCommandName, parsed.collValidator->validatorDoc, parsed.collValidator->isOK());
+
         if (!parsed.collValidator->isOK()) {
             return parsed.collValidator->getStatus();
         }
@@ -533,7 +583,9 @@ StatusWith<std::pair<ParsedCollModRequest, BSONObj>> parseCollModRequest(Operati
                 },
                 [&oplogEntryBuilder](std::int64_t value) {
                     oplogEntryBuilder.append(CollMod::kExpireAfterSecondsFieldName, value);
-                    return index_key_validate::validateExpireAfterSeconds(value);
+                    return index_key_validate::validateExpireAfterSeconds(
+                        value,
+                        index_key_validate::ValidateExpireAfterSecondsMode::kClusteredTTLIndex);
                 },
             },
             *expireAfterSeconds);
@@ -596,7 +648,8 @@ void _setClusteredExpireAfterSeconds(
                 if (!oldExpireAfterSeconds) {
                     auto ttlCache = &TTLCollectionCache::get(opCtx->getServiceContext());
                     opCtx->recoveryUnit()->onCommit([ttlCache, uuid = coll->uuid()](auto _) {
-                        ttlCache->registerTTLInfo(uuid, TTLCollectionCache::ClusteredId());
+                        ttlCache->registerTTLInfo(
+                            uuid, TTLCollectionCache::Info{TTLCollectionCache::ClusteredId{}});
                     });
                 }
 
@@ -846,7 +899,7 @@ Status _collModInternal(OperationContext* opCtx,
                 cmrNew.recordPreImages = false;
             }
 
-            if (cmrNew.recordPreImages) {
+            if (cmrNew.recordPreImages && *cmrNew.recordPreImages) {
                 cmrNew.changeStreamPreAndPostImagesOptions =
                     ChangeStreamPreAndPostImagesOptions(false);
             }
@@ -893,8 +946,9 @@ Status _collModInternal(OperationContext* opCtx,
                                        "Failed to set validationLevel");
         }
 
-        if (cmrNew.recordPreImages != oldCollOptions.recordPreImages) {
-            coll.getWritableCollection(opCtx)->setRecordPreImages(opCtx, cmrNew.recordPreImages);
+        if (cmrNew.recordPreImages.has_value() &&
+            *cmrNew.recordPreImages != oldCollOptions.recordPreImages) {
+            coll.getWritableCollection(opCtx)->setRecordPreImages(opCtx, *cmrNew.recordPreImages);
         }
 
         if (cmrNew.changeStreamPreAndPostImagesOptions.has_value() &&
@@ -963,6 +1017,39 @@ Status _collModInternal(OperationContext* opCtx,
 
 }  // namespace
 
+bool isCollModIndexUniqueConversion(const CollModRequest& request) {
+    auto index = request.getIndex();
+    if (!index) {
+        return false;
+    }
+    if (auto indexUnique = index->getUnique(); !indexUnique) {
+        return false;
+    }
+    // Checks if the request is an actual unique conversion instead of a dry run.
+    if (auto dryRun = request.getDryRun(); dryRun && *dryRun) {
+        return false;
+    }
+    return true;
+}
+
+CollModRequest makeCollModDryRunRequest(const CollModRequest& request) {
+    CollModRequest dryRunRequest;
+    CollModIndex dryRunIndex;
+    const auto& requestIndex = request.getIndex();
+    dryRunIndex.setUnique(true);
+    if (auto keyPattern = requestIndex->getKeyPattern()) {
+        dryRunIndex.setKeyPattern(keyPattern);
+    } else if (auto name = requestIndex->getName()) {
+        dryRunIndex.setName(name);
+    }
+    if (auto uuid = request.getCollectionUUID()) {
+        dryRunRequest.setCollectionUUID(uuid);
+    }
+    dryRunRequest.setIndex(dryRunIndex);
+    dryRunRequest.setDryRun(true);
+    return dryRunRequest;
+}
+
 Status processCollModCommand(OperationContext* opCtx,
                              const NamespaceStringOrUUID& nsOrUUID,
                              const CollMod& cmd,
diff --git a/src/mongo/db/catalog/coll_mod.h b/src/mongo/db/catalog/coll_mod.h
index f08aca11444..f2b0c1702d5 100644
--- a/src/mongo/db/catalog/coll_mod.h
+++ b/src/mongo/db/catalog/coll_mod.h
@@ -48,6 +48,18 @@ class OperationContext;
 void addCollectionUUIDs(OperationContext* opCtx);
 
 /**
+ * Checks if the collMod request is converting an index to unique.
+ */
+bool isCollModIndexUniqueConversion(const CollModRequest& request);
+
+/**
+ * Constructs a valid collMod dry-run request from the original request.
+ * The 'dryRun' option can only be used with the index 'unique' option, so we assume 'request' must
+ * have the 'unique' option. The function will also remove other options from the original request.
+ */
+CollModRequest makeCollModDryRunRequest(const CollModRequest& request);
+
+/**
  * Performs the collection modification described in "cmd" on the collection "ns".
  */
 Status processCollModCommand(OperationContext* opCtx,
diff --git a/src/mongo/db/catalog/coll_mod_index.cpp b/src/mongo/db/catalog/coll_mod_index.cpp
index d56acc639f4..2ffecf7f1e1 100644
--- a/src/mongo/db/catalog/coll_mod_index.cpp
+++ b/src/mongo/db/catalog/coll_mod_index.cpp
@@ -67,8 +67,10 @@ void _processCollModIndexRequestExpireAfterSeconds(OperationContext* opCtx,
         // Do not refer to 'idx' within this commit handler as it may be be invalidated by
         // IndexCatalog::refreshEntry().
         opCtx->recoveryUnit()->onCommit(
-            [ttlCache, uuid = coll->uuid(), indexName = idx->indexName()](auto _) {
-                ttlCache->registerTTLInfo(uuid, indexName);
+            [ttlCache, uuid = coll->uuid(), indexName = idx->indexName(), indexExpireAfterSeconds](
+                auto _) {
+                ttlCache->registerTTLInfo(
+                    uuid, TTLCollectionCache::Info{indexName, /*isExpireAfterSecondsNaN=*/false});
             });
 
         // Change the value of "expireAfterSeconds" on disk.
@@ -77,6 +79,29 @@ void _processCollModIndexRequestExpireAfterSeconds(OperationContext* opCtx,
         return;
     }
 
+    // If the current `expireAfterSeconds` is NaN, it can never be equal to
+    // 'indexExpireAfterSeconds'.
+    if (oldExpireSecsElement.isNaN()) {
+        // Setting *oldExpireSecs is mostly for informational purposes.
+        // We could also use index_key_validate::kExpireAfterSecondsForInactiveTTLIndex but
+        // 0 is more consistent with the previous safeNumberLong() behavior and avoids potential
+        // showing the same value for the new and old values in the collMod response.
+        *oldExpireSecs = 0;
+
+        // Change the value of "expireAfterSeconds" on disk.
+        autoColl->getWritableCollection(opCtx)->updateTTLSetting(
+            opCtx, idx->indexName(), indexExpireAfterSeconds);
+
+        // Keep the TTL information maintained by the TTLCollectionCache in sync so that we don't
+        // try to fix up the TTL index during the next step-up.
+        auto ttlCache = &TTLCollectionCache::get(opCtx->getServiceContext());
+        const auto& coll = autoColl->getCollection();
+        opCtx->recoveryUnit()->onCommit(
+            [ttlCache, uuid = coll->uuid(), indexName = idx->indexName(), indexExpireAfterSeconds](
+                auto _) { ttlCache->unsetTTLIndexExpireAfterSecondsNaN(uuid, indexName); });
+        return;
+    }
+
     // This collection is already TTL. Compare the requested value against the existing setting
     // before updating the catalog.
     *oldExpireSecs = oldExpireSecsElement.safeNumberLong();
diff --git a/src/mongo/db/catalog/coll_mod_test.cpp b/src/mongo/db/catalog/coll_mod_test.cpp
new file mode 100644
index 00000000000..c6335aa1c67
--- /dev/null
+++ b/src/mongo/db/catalog/coll_mod_test.cpp
@@ -0,0 +1,106 @@
+/**
+ *    Copyright (C) 2022-present MongoDB, Inc.
+ *
+ *    This program is free software: you can redistribute it and/or modify
+ *    it under the terms of the Server Side Public License, version 1,
+ *    as published by MongoDB, Inc.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    Server Side Public License for more details.
+ *
+ *    You should have received a copy of the Server Side Public License
+ *    along with this program. If not, see
+ *    <http://www.mongodb.com/licensing/server-side-public-license>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the Server Side Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#include "mongo/db/catalog/coll_mod.h"
+
+#include <boost/optional.hpp>
+
+#include "mongo/db/coll_mod_gen.h"
+#include "mongo/unittest/unittest.h"
+
+namespace mongo {
+namespace {
+TEST(CollModOptionTest, isConvertingIndexToUnique) {
+    IDLParserErrorContext ctx("collMod");
+    auto requestObj = fromjson("{index: {keyPattern: {a: 1}, unique: true}}");
+    auto request = CollModRequest::parse(ctx, requestObj);
+    ASSERT_TRUE(isCollModIndexUniqueConversion(request));
+
+    requestObj = fromjson("{index: {keyPattern: {a: 1}, unique: true, hidden: true}}");
+    request = CollModRequest::parse(ctx, requestObj);
+    ASSERT_TRUE(isCollModIndexUniqueConversion(request));
+
+    requestObj = fromjson(
+        "{index: {keyPattern: {a: 1}, unique: true, hidden: true}, validationAction: 'warn'}");
+    request = CollModRequest::parse(ctx, requestObj);
+    ASSERT_TRUE(isCollModIndexUniqueConversion(request));
+
+    requestObj = fromjson("{index: {keyPattern: {a: 1}, unique: true}, dryRun: true}");
+    request = CollModRequest::parse(ctx, requestObj);
+    ASSERT_FALSE(isCollModIndexUniqueConversion(request));
+
+    requestObj = fromjson("{index: {keyPattern: {a: 1}, unique: true}, dryRun: false}");
+    request = CollModRequest::parse(ctx, requestObj);
+    ASSERT_TRUE(isCollModIndexUniqueConversion(request));
+
+    requestObj = fromjson("{index: {keyPattern: {a: 1}, prepareUnique: true}}");
+    request = CollModRequest::parse(ctx, requestObj);
+    ASSERT_FALSE(isCollModIndexUniqueConversion(request));
+
+    requestObj = fromjson("{validationAction: 'warn'}");
+    request = CollModRequest::parse(ctx, requestObj);
+    ASSERT_FALSE(isCollModIndexUniqueConversion(request));
+}
+
+TEST(CollModOptionTest, makeDryRunRequest) {
+    IDLParserErrorContext ctx("collMod");
+    auto requestObj = fromjson("{index: {keyPattern: {a: 1}, unique: true}}");
+    auto request = CollModRequest::parse(ctx, requestObj);
+    auto dryRunRequest = makeCollModDryRunRequest(request);
+    ASSERT_TRUE(dryRunRequest.getIndex()->getKeyPattern()->binaryEqual(fromjson("{a: 1}")));
+    ASSERT_TRUE(dryRunRequest.getIndex()->getUnique() && *dryRunRequest.getIndex()->getUnique());
+    ASSERT_TRUE(dryRunRequest.getDryRun() && *dryRunRequest.getDryRun());
+
+    requestObj = fromjson("{index: {keyPattern: {a: 1}, unique: true, hidden: true}}");
+    request = CollModRequest::parse(ctx, requestObj);
+    dryRunRequest = makeCollModDryRunRequest(request);
+    ASSERT_TRUE(dryRunRequest.getIndex()->getKeyPattern()->binaryEqual(fromjson("{a: 1}")));
+    ASSERT_TRUE(dryRunRequest.getIndex()->getUnique() && *dryRunRequest.getIndex()->getUnique());
+    ASSERT_FALSE(dryRunRequest.getIndex()->getHidden());
+    ASSERT_TRUE(dryRunRequest.getDryRun() && *dryRunRequest.getDryRun());
+
+    requestObj = fromjson(
+        "{index: {keyPattern: {a: 1}, unique: true, hidden: true}, validationAction: 'warn'}");
+    request = CollModRequest::parse(ctx, requestObj);
+    dryRunRequest = makeCollModDryRunRequest(request);
+    ASSERT_TRUE(dryRunRequest.getIndex()->getKeyPattern()->binaryEqual(fromjson("{a: 1}")));
+    ASSERT_TRUE(dryRunRequest.getIndex()->getUnique() && *dryRunRequest.getIndex()->getUnique());
+    ASSERT_FALSE(dryRunRequest.getIndex()->getHidden());
+    ASSERT_FALSE(dryRunRequest.getValidationAction());
+    ASSERT_TRUE(dryRunRequest.getDryRun() && *dryRunRequest.getDryRun());
+
+    requestObj = fromjson("{index: {keyPattern: {a: 1}, unique: true}, dryRun: false}");
+    request = CollModRequest::parse(ctx, requestObj);
+    dryRunRequest = makeCollModDryRunRequest(request);
+    ASSERT_TRUE(dryRunRequest.getIndex()->getKeyPattern()->binaryEqual(fromjson("{a: 1}")));
+    ASSERT_TRUE(dryRunRequest.getIndex()->getUnique() && *dryRunRequest.getIndex()->getUnique());
+    ASSERT_TRUE(dryRunRequest.getDryRun() && *dryRunRequest.getDryRun());
+}
+}  // namespace
+}  // namespace mongo
diff --git a/src/mongo/db/catalog/collection.h b/src/mongo/db/catalog/collection.h
index 425b02dc9d5..6c619551f32 100644
--- a/src/mongo/db/catalog/collection.h
+++ b/src/mongo/db/catalog/collection.h
@@ -562,6 +562,22 @@ public:
     virtual bool doesTimeseriesBucketsDocContainMixedSchemaData(
         const BSONObj& bucketsDoc) const = 0;
 
+    /**
+     * Returns true if the time-series collection may have dates outside the standard range (roughly
+     * 1970-2038). The value may be updated in the background by another thread between calls, even
+     * if the caller holds a lock on the collection. The value may only transition from false to
+     * true.
+     */
+    virtual bool getRequiresTimeseriesExtendedRangeSupport() const = 0;
+
+    /**
+     * Sets the in-memory flag for this collection. This value can be retrieved by
+     * 'getRequiresTimeseriesExtendedRangeSupport'.
+     *
+     * Throws if this is not a time-series collection.
+     */
+    virtual void setRequiresTimeseriesExtendedRangeSupport(OperationContext* opCtx) const = 0;
+
     /*
      * Returns true if this collection is clustered. That is, its RecordIds store the value of the
      * cluster key. If the collection is clustered on _id, there is no separate _id index.
diff --git a/src/mongo/db/catalog/collection_catalog.cpp b/src/mongo/db/catalog/collection_catalog.cpp
index 4e5df27b592..bf9e287afce 100644
--- a/src/mongo/db/catalog/collection_catalog.cpp
+++ b/src/mongo/db/catalog/collection_catalog.cpp
@@ -469,20 +469,28 @@ void CollectionCatalog::write(OperationContext* opCtx,
     write(opCtx->getServiceContext(), std::move(job));
 }
 
-Status CollectionCatalog::createView(
-    OperationContext* opCtx,
-    const NamespaceString& viewName,
-    const NamespaceString& viewOn,
-    const BSONArray& pipeline,
-    const BSONObj& collation,
-    const ViewsForDatabase::PipelineValidatorFn& pipelineValidator) const {
-    invariant(opCtx->lockState()->isCollectionLockedForMode(viewName, MODE_IX));
+Status CollectionCatalog::createView(OperationContext* opCtx,
+                                     const NamespaceString& viewName,
+                                     const NamespaceString& viewOn,
+                                     const BSONArray& pipeline,
+                                     const BSONObj& collation,
+                                     const ViewsForDatabase::PipelineValidatorFn& pipelineValidator,
+                                     const ViewUpsertMode insertViewMode) const {
+    // A view document direct write can occur via the oplog application path, which may only hold a
+    // lock on the collection being updated (the database views collection).
+    invariant(insertViewMode == ViewUpsertMode::kAlreadyDurableView ||
+              opCtx->lockState()->isCollectionLockedForMode(viewName, MODE_IX));
     invariant(opCtx->lockState()->isCollectionLockedForMode(
         NamespaceString(viewName.db(), NamespaceString::kSystemDotViewsCollectionName), MODE_X));
 
     invariant(_viewsForDatabase.contains(viewName.db()));
     const ViewsForDatabase& viewsForDb = *_getViewsForDatabase(opCtx, viewName.db());
 
+    auto& uncommittedCatalogUpdates = UncommittedCatalogUpdates::get(opCtx);
+    if (uncommittedCatalogUpdates.shouldIgnoreExternalViewChanges(viewName.db())) {
+        return Status::OK();
+    }
+
     if (viewName.db() != viewOn.db())
         return Status(ErrorCodes::BadValue,
                       "View must be created on a view or collection in the same database");
@@ -508,7 +516,8 @@ Status CollectionCatalog::createView(
                                      pipeline,
                                      pipelineValidator,
                                      std::move(collator.getValue()),
-                                     ViewsForDatabase{viewsForDb});
+                                     ViewsForDatabase{viewsForDb},
+                                     insertViewMode);
     }
 
     return result;
@@ -550,7 +559,8 @@ Status CollectionCatalog::modifyView(
                                      pipeline,
                                      pipelineValidator,
                                      CollatorInterface::cloneCollator(viewPtr->defaultCollator()),
-                                     ViewsForDatabase{viewsForDb});
+                                     ViewsForDatabase{viewsForDb},
+                                     ViewUpsertMode::kUpdateView);
     }
 
     return result;
@@ -689,16 +699,17 @@ void CollectionCatalog::onCloseDatabase(OperationContext* opCtx, TenantDatabaseN
     _viewsForDatabase.erase(tenantDbName.dbName());
 }
 
-void CollectionCatalog::onCloseCatalog(OperationContext* opCtx) {
-    invariant(opCtx->lockState()->isW());
-    invariant(!_shadowCatalog);
+void CollectionCatalog::onCloseCatalog() {
+    if (_shadowCatalog) {
+        return;
+    }
+
     _shadowCatalog.emplace();
     for (auto& entry : _catalog)
         _shadowCatalog->insert({entry.first, entry.second->ns()});
 }
 
-void CollectionCatalog::onOpenCatalog(OperationContext* opCtx) {
-    invariant(opCtx->lockState()->isW());
+void CollectionCatalog::onOpenCatalog() {
     invariant(_shadowCatalog);
     _shadowCatalog.reset();
     ++_epoch;
@@ -1086,6 +1097,12 @@ std::vector<TenantDatabaseName> CollectionCatalog::getAllDbNames() const {
     return ret;
 }
 
+void CollectionCatalog::setAllDatabaseProfileFilters(std::shared_ptr<ProfileFilter> filter) {
+    for (auto& [_, settings] : _databaseProfileSettings) {
+        settings.filter = filter;
+    }
+}
+
 void CollectionCatalog::setDatabaseProfileSettings(
     StringData dbName, CollectionCatalog::ProfileSettings newProfileSettings) {
     _databaseProfileSettings[dbName] = newProfileSettings;
@@ -1403,15 +1420,19 @@ Status CollectionCatalog::_createOrUpdateView(
     const BSONArray& pipeline,
     const ViewsForDatabase::PipelineValidatorFn& pipelineValidator,
     std::unique_ptr<CollatorInterface> collator,
-    ViewsForDatabase&& viewsForDb) const {
-    invariant(opCtx->lockState()->isCollectionLockedForMode(viewName, MODE_IX));
+    ViewsForDatabase&& viewsForDb,
+    ViewUpsertMode insertViewMode) const {
+    // A view document direct write can occur via the oplog application path, which may only hold a
+    // lock on the collection being updated (the database views collection).
+    invariant(insertViewMode == ViewUpsertMode::kAlreadyDurableView ||
+              opCtx->lockState()->isCollectionLockedForMode(viewName, MODE_IX));
     invariant(opCtx->lockState()->isCollectionLockedForMode(
         NamespaceString(viewName.db(), NamespaceString::kSystemDotViewsCollectionName), MODE_X));
 
     viewsForDb.requireValidCatalog();
 
-    // Build the BSON definition for this view to be saved in the durable view catalog. If the
-    // collation is empty, omit it from the definition altogether.
+    // Build the BSON definition for this view to be saved in the durable view catalog and/or to
+    // insert in the viewMap. If the collation is empty, omit it from the definition altogether.
     BSONObjBuilder viewDefBuilder;
     viewDefBuilder.append("_id", viewName.ns());
     viewDefBuilder.append("viewOn", viewOn.coll());
@@ -1420,25 +1441,42 @@ Status CollectionCatalog::_createOrUpdateView(
         viewDefBuilder.append("collation", collator->getSpec().toBSON());
     }
 
+    BSONObj viewDef = viewDefBuilder.obj();
     BSONObj ownedPipeline = pipeline.getOwned();
-    auto view = std::make_shared<ViewDefinition>(
+    ViewDefinition view(
         viewName.db(), viewName.coll(), viewOn.coll(), ownedPipeline, std::move(collator));
 
-    // Check that the resulting dependency graph is acyclic and within the maximum depth.
-    Status graphStatus = viewsForDb.upsertIntoGraph(opCtx, *(view.get()), pipelineValidator);
+    // If the view is already in the durable view catalog, we don't need to validate the graph. If
+    // we need to update the durable view catalog, we need to check that the resulting dependency
+    // graph is acyclic and within the maximum depth.
+    const bool viewGraphNeedsValidation = insertViewMode != ViewUpsertMode::kAlreadyDurableView;
+    Status graphStatus =
+        viewsForDb.upsertIntoGraph(opCtx, view, pipelineValidator, viewGraphNeedsValidation);
     if (!graphStatus.isOK()) {
         return graphStatus;
     }
 
-    viewsForDb.durable->upsert(opCtx, viewName, viewDefBuilder.obj());
+    if (insertViewMode != ViewUpsertMode::kAlreadyDurableView) {
+        viewsForDb.durable->upsert(opCtx, viewName, viewDef);
+    }
 
-    viewsForDb.viewMap.clear();
     viewsForDb.valid = false;
-    viewsForDb.viewGraphNeedsRefresh = true;
-    viewsForDb.stats = {};
+    auto res = [&] {
+        switch (insertViewMode) {
+            case ViewUpsertMode::kCreateView:
+            case ViewUpsertMode::kAlreadyDurableView:
+                return viewsForDb.insert(opCtx, viewDef);
+            case ViewUpsertMode::kUpdateView:
+                viewsForDb.viewMap.clear();
+                viewsForDb.viewGraphNeedsRefresh = true;
+                viewsForDb.stats = {};
+
+                // Reload the view catalog with the changes applied.
+                return viewsForDb.reload(opCtx);
+        }
+        MONGO_UNREACHABLE;
+    }();
 
-    // Reload the view catalog with the changes applied.
-    auto res = viewsForDb.reload(opCtx);
     if (res.isOK()) {
         auto& uncommittedCatalogUpdates = UncommittedCatalogUpdates::get(opCtx);
         uncommittedCatalogUpdates.addView(opCtx, viewName);
@@ -1527,10 +1565,7 @@ const Collection* LookupCollectionForYieldRestore::operator()(OperationContext*
     // state. After a query yields its locks, the replication state may have changed, invalidating
     // our current choice of ReadSource. Using the same preconditions, change our ReadSource if
     // necessary.
-    auto [newReadSource, _] = SnapshotHelper::shouldChangeReadSource(opCtx, collection->ns());
-    if (newReadSource) {
-        opCtx->recoveryUnit()->setTimestampReadSource(*newReadSource);
-    }
+    SnapshotHelper::changeReadSourceIfNeeded(opCtx, collection->ns());
 
     return collection.get();
 }
diff --git a/src/mongo/db/catalog/collection_catalog.h b/src/mongo/db/catalog/collection_catalog.h
index f47f94397f9..027774e9142 100644
--- a/src/mongo/db/catalog/collection_catalog.h
+++ b/src/mongo/db/catalog/collection_catalog.h
@@ -97,7 +97,7 @@ public:
 
     struct ProfileSettings {
         int level;
-        std::shared_ptr<ProfileFilter> filter;  // nullable
+        std::shared_ptr<const ProfileFilter> filter;  // nullable
 
         ProfileSettings(int level, std::shared_ptr<ProfileFilter> filter)
             : level(level), filter(filter) {
@@ -115,6 +115,19 @@ public:
         }
     };
 
+    enum class ViewUpsertMode {
+        // Insert all data for that view into the view map, view graph, and durable view catalog.
+        kCreateView,
+
+        // Insert into the view map and view graph without reinserting the view into the durable
+        // view catalog. Skip view graph validation.
+        kAlreadyDurableView,
+
+        // Reload the view map, insert into the view graph (flagging it as needing refresh), and
+        // update the durable view catalog.
+        kUpdateView,
+    };
+
     static std::shared_ptr<const CollectionCatalog> get(ServiceContext* svcCtx);
     static std::shared_ptr<const CollectionCatalog> get(OperationContext* opCtx);
 
@@ -147,14 +160,16 @@ public:
      *
      * Must be in WriteUnitOfWork. View creation rolls back if the unit of work aborts.
      *
-     * Caller must ensure corresponding database exists.
+     * Caller must ensure corresponding database exists. Expects db.system.views MODE_X lock and
+     * view namespace MODE_IX lock (unless 'insertViewMode' is set to kAlreadyDurableView).
      */
     Status createView(OperationContext* opCtx,
                       const NamespaceString& viewName,
                       const NamespaceString& viewOn,
                       const BSONArray& pipeline,
                       const BSONObj& collation,
-                      const ViewsForDatabase::PipelineValidatorFn& pipelineValidator) const;
+                      const ViewsForDatabase::PipelineValidatorFn& pipelineValidator,
+                      ViewUpsertMode insertViewMode = ViewUpsertMode::kCreateView) const;
 
     /**
      * Drop the view named 'viewName'.
@@ -406,6 +421,11 @@ public:
     std::vector<TenantDatabaseName> getAllDbNames() const;
 
     /**
+     * Updates the profile filter on all databases with non-default settings.
+     */
+    void setAllDatabaseProfileFilters(std::shared_ptr<ProfileFilter> filter);
+
+    /**
      * Sets 'newProfileSettings' as the profiling settings for the database 'dbName'.
      */
     void setDatabaseProfileSettings(StringData dbName, ProfileSettings newProfileSettings);
@@ -474,14 +494,14 @@ public:
      *
      * Must be called with the global lock acquired in exclusive mode.
      */
-    void onCloseCatalog(OperationContext* opCtx);
+    void onCloseCatalog();
 
     /**
      * Puts the catalog back in open state, removing the pre-close state. See onCloseCatalog.
      *
      * Must be called with the global lock acquired in exclusive mode.
      */
-    void onOpenCatalog(OperationContext* opCtx);
+    void onOpenCatalog();
 
     /**
      * The epoch is incremented whenever the catalog is closed and re-opened.
@@ -548,7 +568,8 @@ private:
                                const BSONArray& pipeline,
                                const ViewsForDatabase::PipelineValidatorFn& pipelineValidator,
                                std::unique_ptr<CollatorInterface> collator,
-                               ViewsForDatabase&& viewsForDb) const;
+                               ViewsForDatabase&& viewsForDb,
+                               ViewUpsertMode insertViewMode) const;
 
     /**
      * Returns true if this CollectionCatalog instance is part of an ongoing batched catalog write.
diff --git a/src/mongo/db/catalog/collection_catalog_test.cpp b/src/mongo/db/catalog/collection_catalog_test.cpp
index e28bcd4ad28..f00588380a1 100644
--- a/src/mongo/db/catalog/collection_catalog_test.cpp
+++ b/src/mongo/db/catalog/collection_catalog_test.cpp
@@ -519,7 +519,7 @@ TEST_F(CollectionCatalogTest, RenameCollection) {
 TEST_F(CollectionCatalogTest, LookupNSSByUUIDForClosedCatalogReturnsOldNSSIfDropped) {
     {
         Lock::GlobalLock globalLk(opCtx.get(), MODE_X);
-        catalog.onCloseCatalog(opCtx.get());
+        catalog.onCloseCatalog();
     }
 
     catalog.deregisterCollection(opCtx.get(), colUUID);
@@ -528,7 +528,7 @@ TEST_F(CollectionCatalogTest, LookupNSSByUUIDForClosedCatalogReturnsOldNSSIfDrop
 
     {
         Lock::GlobalLock globalLk(opCtx.get(), MODE_X);
-        catalog.onOpenCatalog(opCtx.get());
+        catalog.onOpenCatalog();
     }
 
     ASSERT_EQUALS(catalog.lookupNSSByUUID(opCtx.get(), colUUID), boost::none);
@@ -543,7 +543,7 @@ TEST_F(CollectionCatalogTest, LookupNSSByUUIDForClosedCatalogReturnsNewlyCreated
     // Ensure that looking up non-existing UUIDs doesn't affect later registration of those UUIDs.
     {
         Lock::GlobalLock globalLk(opCtx.get(), MODE_X);
-        catalog.onCloseCatalog(opCtx.get());
+        catalog.onCloseCatalog();
     }
 
     ASSERT(catalog.lookupCollectionByUUID(opCtx.get(), newUUID) == nullptr);
@@ -555,7 +555,7 @@ TEST_F(CollectionCatalogTest, LookupNSSByUUIDForClosedCatalogReturnsNewlyCreated
     // Ensure that collection still exists after opening the catalog again.
     {
         Lock::GlobalLock globalLk(opCtx.get(), MODE_X);
-        catalog.onOpenCatalog(opCtx.get());
+        catalog.onOpenCatalog();
     }
 
     ASSERT_EQUALS(catalog.lookupCollectionByUUID(opCtx.get(), newUUID), newCol);
@@ -569,7 +569,7 @@ TEST_F(CollectionCatalogTest, LookupNSSByUUIDForClosedCatalogReturnsFreshestNSS)
 
     {
         Lock::GlobalLock globalLk(opCtx.get(), MODE_X);
-        catalog.onCloseCatalog(opCtx.get());
+        catalog.onCloseCatalog();
     }
 
     catalog.deregisterCollection(opCtx.get(), colUUID);
@@ -582,7 +582,7 @@ TEST_F(CollectionCatalogTest, LookupNSSByUUIDForClosedCatalogReturnsFreshestNSS)
     // Ensure that collection still exists after opening the catalog again.
     {
         Lock::GlobalLock globalLk(opCtx.get(), MODE_X);
-        catalog.onOpenCatalog(opCtx.get());
+        catalog.onOpenCatalog();
     }
 
     ASSERT_EQUALS(catalog.lookupCollectionByUUID(opCtx.get(), colUUID), newCol);
@@ -595,8 +595,8 @@ TEST_F(CollectionCatalogTest, CollectionCatalogEpoch) {
 
     {
         Lock::GlobalLock globalLk(opCtx.get(), MODE_X);
-        catalog.onCloseCatalog(opCtx.get());
-        catalog.onOpenCatalog(opCtx.get());
+        catalog.onCloseCatalog();
+        catalog.onOpenCatalog();
     }
 
     auto incrementedEpoch = catalog.getEpoch();
diff --git a/src/mongo/db/catalog/collection_compact.cpp b/src/mongo/db/catalog/collection_compact.cpp
index 4fed779eaae..b6cc7cb444d 100644
--- a/src/mongo/db/catalog/collection_compact.cpp
+++ b/src/mongo/db/catalog/collection_compact.cpp
@@ -74,10 +74,10 @@ StatusWith<int64_t> compactCollection(OperationContext* opCtx,
     Database* database = autoDb.getDb();
     uassert(ErrorCodes::NamespaceNotFound, "database does not exist", database);
 
-    // The collection lock will be downgraded to an intent lock if the record store supports
-    // online compaction.
+    // The collection lock will be upgraded to an exclusive lock if the record store does not
+    // support online compaction.
     boost::optional<Lock::CollectionLock> collLk;
-    collLk.emplace(opCtx, collectionNss, MODE_X);
+    collLk.emplace(opCtx, collectionNss, MODE_IX);
 
     CollectionPtr collection = getCollectionForCompact(opCtx, collectionNss);
     DisableDocumentValidation validationDisabler(opCtx);
@@ -91,10 +91,9 @@ StatusWith<int64_t> compactCollection(OperationContext* opCtx,
                       str::stream() << "cannot compact collection with record store: "
                                     << recordStore->name());
 
-    if (recordStore->supportsOnlineCompaction()) {
-        // Storage engines that allow online compaction should do so using an intent lock on the
-        // collection.
-        collLk.emplace(opCtx, collectionNss, MODE_IX);
+    if (!recordStore->supportsOnlineCompaction()) {
+        // Storage engines that disallow online compaction should compact under an exclusive lock.
+        collLk.emplace(opCtx, collectionNss, MODE_X);
 
         // Ensure the collection was not dropped during the re-lock.
         collection = getCollectionForCompact(opCtx, collectionNss);
diff --git a/src/mongo/db/catalog/collection_impl.cpp b/src/mongo/db/catalog/collection_impl.cpp
index b79c8c78914..18f2d817d7e 100644
--- a/src/mongo/db/catalog/collection_impl.cpp
+++ b/src/mongo/db/catalog/collection_impl.cpp
@@ -40,6 +40,7 @@
 #include "mongo/bson/simple_bsonobj_comparator.h"
 #include "mongo/crypto/fle_crypto.h"
 #include "mongo/db/auth/security_token.h"
+#include "mongo/db/catalog/catalog_stats.h"
 #include "mongo/db/catalog/collection_catalog.h"
 #include "mongo/db/catalog/collection_options.h"
 #include "mongo/db/catalog/document_validation.h"
@@ -80,6 +81,7 @@
 #include "mongo/db/storage/record_store.h"
 #include "mongo/db/storage/storage_parameters_gen.h"
 #include "mongo/db/timeseries/timeseries_constants.h"
+#include "mongo/db/timeseries/timeseries_extended_range.h"
 #include "mongo/db/timeseries/timeseries_index_schema_conversion_functions.h"
 #include "mongo/db/transaction_participant.h"
 #include "mongo/db/ttl_collection_cache.h"
@@ -539,11 +541,11 @@ void CollectionImpl::init(OperationContext* opCtx) {
             if (opCtx->lockState()->inAWriteUnitOfWork()) {
                 opCtx->recoveryUnit()->onCommit([svcCtx, uuid](auto ts) {
                     TTLCollectionCache::get(svcCtx).registerTTLInfo(
-                        uuid, TTLCollectionCache::ClusteredId{});
+                        uuid, TTLCollectionCache::Info{TTLCollectionCache::ClusteredId{}});
                 });
             } else {
-                TTLCollectionCache::get(svcCtx).registerTTLInfo(uuid,
-                                                                TTLCollectionCache::ClusteredId{});
+                TTLCollectionCache::get(svcCtx).registerTTLInfo(
+                    uuid, TTLCollectionCache::Info{TTLCollectionCache::ClusteredId{}});
             }
         }
     }
@@ -725,6 +727,8 @@ Collection::Validator CollectionImpl::parseValidator(
     auto expCtx = make_intrusive<ExpressionContext>(
         opCtx, CollatorInterface::cloneCollator(_shared->_collator.get()), ns());
 
+    expCtx->variables.setDefaultRuntimeConstants(opCtx);
+
     // The MatchExpression and contained ExpressionContext created as part of the validator are
     // owned by the Collection and will outlive the OperationContext they were created under.
     expCtx->opCtx = nullptr;
@@ -811,9 +815,9 @@ Status CollectionImpl::insertDocumentsForOplog(OperationContext* opCtx,
 
     _cappedDeleteAsNeeded(opCtx, records->begin()->id);
 
-    opCtx->recoveryUnit()->onCommit(
-        [this](boost::optional<Timestamp>) { _shared->notifyCappedWaitersIfNeeded(); });
-
+    // We do not need to notify capped waiters, as we have not yet updated oplog visibility, so
+    // these inserts will not be visible.  When visibility updates, it will notify capped
+    // waiters.
     return status;
 }
 
@@ -1617,6 +1621,29 @@ bool CollectionImpl::doesTimeseriesBucketsDocContainMixedSchemaData(
     return doesMinMaxHaveMixedSchemaData(minObj, maxObj);
 }
 
+bool CollectionImpl::getRequiresTimeseriesExtendedRangeSupport() const {
+    return _shared->_requiresTimeseriesExtendedRangeSupport.load();
+}
+
+void CollectionImpl::setRequiresTimeseriesExtendedRangeSupport(OperationContext* opCtx) const {
+    uassert(6679401, "This is not a time-series collection", _metadata->options.timeseries);
+
+    bool expected = false;
+    bool set = _shared->_requiresTimeseriesExtendedRangeSupport.compareAndSwap(&expected, true);
+    if (set) {
+        catalog_stats::requiresTimeseriesExtendedRangeSupport.fetchAndAdd(1);
+        if (!timeseries::collectionHasTimeIndex(opCtx, *this)) {
+            LOGV2_WARNING(
+                6679402,
+                "Time-series collection contains dates outside the standard range. Some query "
+                "optimizations may be disabled. Please consider building an index on timeField to "
+                "re-enable them.",
+                "nss"_attr = ns().getTimeseriesViewNamespace(),
+                "timeField"_attr = _metadata->options.timeseries->getTimeField());
+        }
+    }
+}
+
 bool CollectionImpl::isClustered() const {
     return getClusteredInfo().is_initialized();
 }
@@ -1755,7 +1782,8 @@ uint64_t CollectionImpl::getIndexSize(OperationContext* opCtx,
                                       int scale) const {
     const IndexCatalog* idxCatalog = getIndexCatalog();
 
-    std::unique_ptr<IndexCatalog::IndexIterator> ii = idxCatalog->getIndexIterator(opCtx, true);
+    auto ii = idxCatalog->getIndexIterator(
+        opCtx, IndexCatalog::InclusionPolicy::kReady | IndexCatalog::InclusionPolicy::kUnfinished);
 
     uint64_t totalSize = 0;
 
@@ -1776,9 +1804,18 @@ uint64_t CollectionImpl::getIndexSize(OperationContext* opCtx,
 }
 
 uint64_t CollectionImpl::getIndexFreeStorageBytes(OperationContext* const opCtx) const {
+    // Unfinished index builds are excluded to avoid a potential deadlock when trying to collect
+    // statistics from the index table while the index build is in the bulk load phase. See
+    // SERVER-77018. This should not be too impactful as:
+    // - During the collection scan phase, the index table is unused.
+    // - During the bulk load phase, getFreeStorageBytes will probably return EBUSY, as the ident is
+    //  in use by the index builder. (And worst case results in the deadlock).
+    // - It might be possible to return meaningful data post bulk-load, but reusable bytes should be
+    //  low anyways as the collection has been bulk loaded. Additionally, this would be a inaccurate
+    //  anyways as the build is in progress.
+    // - Once the index build is finished, this will be eventually accounted for.
     const auto idxCatalog = getIndexCatalog();
-    const bool includeUnfinished = true;
-    auto indexIt = idxCatalog->getIndexIterator(opCtx, includeUnfinished);
+    auto indexIt = idxCatalog->getIndexIterator(opCtx, IndexCatalog::InclusionPolicy::kReady);
 
     uint64_t totalSize = 0;
     while (indexIt->more()) {
@@ -1802,8 +1839,7 @@ Status CollectionImpl::truncate(OperationContext* opCtx) {
     // 1) store index specs
     std::vector<BSONObj> indexSpecs;
     {
-        std::unique_ptr<IndexCatalog::IndexIterator> ii =
-            _indexCatalog->getIndexIterator(opCtx, false);
+        auto ii = _indexCatalog->getIndexIterator(opCtx, IndexCatalog::InclusionPolicy::kReady);
         while (ii->more()) {
             const IndexDescriptor* idx = ii->next()->descriptor();
             indexSpecs.push_back(idx->infoObj().getOwned());
@@ -2228,44 +2264,65 @@ bool CollectionImpl::isIndexMultikey(OperationContext* opCtx,
                                      StringData indexName,
                                      MultikeyPaths* multikeyPaths,
                                      int indexOffset) const {
-    auto isMultikey = [this, multikeyPaths, indexName, indexOffset](
-                          const BSONCollectionCatalogEntry::MetaData& metadata) {
-        int offset = indexOffset;
-        if (offset < 0) {
-            offset = metadata.findIndexOffset(indexName);
-            invariant(offset >= 0,
-                      str::stream() << "cannot get multikey for index " << indexName << " @ "
-                                    << getCatalogId() << " : " << metadata.toBSON());
-        } else {
-            invariant(offset < int(metadata.indexes.size()),
-                      str::stream()
-                          << "out of bounds index offset for multikey info " << indexName << " @ "
-                          << getCatalogId() << " : " << metadata.toBSON() << "; offset : " << offset
-                          << " ; actual : " << metadata.findIndexOffset(indexName));
-            invariant(indexName == metadata.indexes[offset].nameStringData(),
-                      str::stream()
-                          << "invalid index offset for multikey info " << indexName << " @ "
-                          << getCatalogId() << " : " << metadata.toBSON() << "; offset : " << offset
-                          << " ; actual : " << metadata.findIndexOffset(indexName));
-        }
-
-        const auto& index = metadata.indexes[offset];
-        stdx::lock_guard lock(index.multikeyMutex);
-        if (multikeyPaths && !index.multikeyPaths.empty()) {
-            *multikeyPaths = index.multikeyPaths;
-        }
-
-        return index.multikey;
-    };
-
+    int offset = indexOffset;
+    if (offset < 0) {
+        offset = _metadata->findIndexOffset(indexName);
+        invariant(offset >= 0,
+                  str::stream() << "cannot get multikey for index " << indexName << " @ "
+                                << getCatalogId() << " : " << _metadata->toBSON());
+    } else {
+        invariant(offset < int(_metadata->indexes.size()),
+                  str::stream() << "out of bounds index offset for multikey info " << indexName
+                                << " @ " << getCatalogId() << " : " << _metadata->toBSON()
+                                << "; offset : " << offset
+                                << " ; actual : " << _metadata->findIndexOffset(indexName));
+        invariant(indexName == _metadata->indexes[offset].nameStringData(),
+                  str::stream() << "invalid index offset for multikey info " << indexName << " @ "
+                                << getCatalogId() << " : " << _metadata->toBSON()
+                                << "; offset : " << offset
+                                << " ; actual : " << _metadata->findIndexOffset(indexName));
+    }
+
+    // If we have uncommitted multikey writes we need to check here to read our own writes
     const auto& uncommittedMultikeys = UncommittedMultikey::get(opCtx).resources();
     if (uncommittedMultikeys) {
         if (auto it = uncommittedMultikeys->find(this); it != uncommittedMultikeys->end()) {
-            return isMultikey(it->second);
+            const auto& index = it->second.indexes[offset];
+            if (multikeyPaths && !index.multikeyPaths.empty()) {
+                *multikeyPaths = index.multikeyPaths;
+            }
+            return index.multikey;
+        }
+    }
+
+    // Otherwise read from the metadata cache if there are no concurrent multikey writers
+    {
+        const auto& index = _metadata->indexes[offset];
+        // Check for concurrent writers, this can race with writers where it can be set immediately
+        // after checking. This is fine we know that the reader in that case opened its snapshot
+        // before the writer and we do not need to observe its result.
+        if (index.concurrentWriters.load() == 0) {
+            stdx::lock_guard lock(index.multikeyMutex);
+            if (multikeyPaths && !index.multikeyPaths.empty()) {
+                *multikeyPaths = index.multikeyPaths;
+            }
+            return index.multikey;
         }
     }
 
-    return isMultikey(*_metadata);
+    // We need to read from the durable catalog if there are concurrent multikey writers to avoid
+    // reading between the multikey write committing in the storage engine but before its onCommit
+    // handler made the write visible for readers.
+    auto snapshotMetadata = DurableCatalog::get(opCtx)->getMetaData(opCtx, getCatalogId());
+    int snapshotOffset = snapshotMetadata->findIndexOffset(indexName);
+    invariant(snapshotOffset >= 0,
+              str::stream() << "cannot get multikey for index " << indexName << " @ "
+                            << getCatalogId() << " : " << _metadata->toBSON());
+    const auto& index = snapshotMetadata->indexes[snapshotOffset];
+    if (multikeyPaths && !index.multikeyPaths.empty()) {
+        *multikeyPaths = index.multikeyPaths;
+    }
+    return index.multikey;
 }
 
 bool CollectionImpl::setIndexIsMultikey(OperationContext* opCtx,
@@ -2273,31 +2330,31 @@ bool CollectionImpl::setIndexIsMultikey(OperationContext* opCtx,
                                         const MultikeyPaths& multikeyPaths,
                                         int indexOffset) const {
 
-    auto setMultikey = [this, indexName, multikeyPaths, indexOffset](
-                           const BSONCollectionCatalogEntry::MetaData& metadata) {
-        int offset = indexOffset;
-        if (offset < 0) {
-            offset = metadata.findIndexOffset(indexName);
-            invariant(offset >= 0,
-                      str::stream() << "cannot set multikey for index " << indexName << " @ "
-                                    << getCatalogId() << " : " << metadata.toBSON());
-        } else {
-            invariant(offset < int(metadata.indexes.size()),
-                      str::stream()
-                          << "out of bounds index offset for multikey update" << indexName << " @ "
-                          << getCatalogId() << " : " << metadata.toBSON() << "; offset : " << offset
-                          << " ; actual : " << metadata.findIndexOffset(indexName));
-            invariant(indexName == metadata.indexes[offset].nameStringData(),
-                      str::stream()
-                          << "invalid index offset for multikey update " << indexName << " @ "
-                          << getCatalogId() << " : " << metadata.toBSON() << "; offset : " << offset
-                          << " ; actual : " << metadata.findIndexOffset(indexName));
-        }
-
+    int offset = indexOffset;
+    if (offset < 0) {
+        offset = _metadata->findIndexOffset(indexName);
+        invariant(offset >= 0,
+                  str::stream() << "cannot set multikey for index " << indexName << " @ "
+                                << getCatalogId() << " : " << _metadata->toBSON());
+    } else {
+        invariant(offset < int(_metadata->indexes.size()),
+                  str::stream() << "out of bounds index offset for multikey update" << indexName
+                                << " @ " << getCatalogId() << " : " << _metadata->toBSON()
+                                << "; offset : " << offset
+                                << " ; actual : " << _metadata->findIndexOffset(indexName));
+        invariant(indexName == _metadata->indexes[offset].nameStringData(),
+                  str::stream() << "invalid index offset for multikey update " << indexName << " @ "
+                                << getCatalogId() << " : " << _metadata->toBSON()
+                                << "; offset : " << offset
+                                << " ; actual : " << _metadata->findIndexOffset(indexName));
+    }
+
+    auto setMultikey = [offset,
+                        multikeyPaths](const BSONCollectionCatalogEntry::MetaData& metadata) {
         auto* index = &metadata.indexes[offset];
         stdx::lock_guard lock(index->multikeyMutex);
 
-        auto tracksPathLevelMultikeyInfo = !metadata.indexes[offset].multikeyPaths.empty();
+        auto tracksPathLevelMultikeyInfo = !index->multikeyPaths.empty();
         if (!tracksPathLevelMultikeyInfo) {
             invariant(multikeyPaths.empty());
 
@@ -2313,7 +2370,7 @@ bool CollectionImpl::setIndexIsMultikey(OperationContext* opCtx,
 
         // We are tracking path-level multikey information for this index.
         invariant(!multikeyPaths.empty());
-        invariant(multikeyPaths.size() == metadata.indexes[offset].multikeyPaths.size());
+        invariant(multikeyPaths.size() == index->multikeyPaths.size());
 
         index->multikey = true;
 
@@ -2352,11 +2409,31 @@ bool CollectionImpl::setIndexIsMultikey(OperationContext* opCtx,
     }
     BSONCollectionCatalogEntry::MetaData* metadata = nullptr;
     bool hasSetMultikey = false;
+
     if (auto it = uncommittedMultikeys->find(this); it != uncommittedMultikeys->end()) {
         metadata = &it->second;
         hasSetMultikey = setMultikey(*metadata);
     } else {
-        BSONCollectionCatalogEntry::MetaData metadataLocal(*_metadata);
+        // First time this OperationContext needs to change multikey information for this
+        // collection. We cannot use the cached metadata in this collection as we may have just
+        // committed a multikey change concurrently to the storage engine without being able to
+        // observe it if its onCommit handlers haven't run yet.
+        auto metadataLocal = *DurableCatalog::get(opCtx)->getMetaData(opCtx, getCatalogId());
+        // When reading from the durable catalog the index offsets are different because when
+        // removing indexes in-memory just zeros out the slot instead of actually removing it. We
+        // must adjust the entries so they match how they are stored in _metadata so we can rely on
+        // the index offsets being stable. The order of valid indexes are the same, so we can
+        // iterate from the end and move them into the right positions.
+        int localIdx = metadataLocal.indexes.size() - 1;
+        metadataLocal.indexes.resize(_metadata->indexes.size());
+        for (int i = _metadata->indexes.size() - 1; i >= 0 && localIdx != i; --i) {
+            if (_metadata->indexes[i].isPresent()) {
+                metadataLocal.indexes[i] = std::move(metadataLocal.indexes[localIdx]);
+                metadataLocal.indexes[localIdx] = {};
+                --localIdx;
+            }
+        }
+
         hasSetMultikey = setMultikey(metadataLocal);
         if (hasSetMultikey) {
             metadata = &uncommittedMultikeys->emplace(this, std::move(metadataLocal)).first->second;
@@ -2371,8 +2448,44 @@ bool CollectionImpl::setIndexIsMultikey(OperationContext* opCtx,
 
     DurableCatalog::get(opCtx)->putMetaData(opCtx, getCatalogId(), *metadata);
 
+    // RAII Helper object to ensure we decrement the concurrent counter if and only if we
+    // incremented it in a preCommit handler.
+    class ConcurrentMultikeyWriteTracker {
+    public:
+        ConcurrentMultikeyWriteTracker(
+            std::shared_ptr<const BSONCollectionCatalogEntry::MetaData> meta, int indexOffset)
+            : metadata(std::move(meta)), offset(indexOffset) {}
+
+        ~ConcurrentMultikeyWriteTracker() {
+            if (hasIncremented) {
+                metadata->indexes[offset].concurrentWriters.fetchAndSubtract(1);
+            }
+        }
+
+        void preCommit() {
+            metadata->indexes[offset].concurrentWriters.fetchAndAdd(1);
+            hasIncremented = true;
+        }
+
+    private:
+        std::shared_ptr<const BSONCollectionCatalogEntry::MetaData> metadata;
+        int offset;
+        bool hasIncremented = false;
+    };
+
+    auto concurrentWriteTracker =
+        std::make_shared<ConcurrentMultikeyWriteTracker>(_metadata, offset);
+
+    // Mark this index that there is an ongoing multikey write. This forces readers to read from the
+    // durable catalog to determine if the index is multikey or not.
+    opCtx->recoveryUnit()->registerPreCommitHook(
+        [concurrentWriteTracker](OperationContext*) { concurrentWriteTracker->preCommit(); });
+
+    // Capture a reference to 'concurrentWriteTracker' to extend the lifetime of this object until
+    // commiting/rolling back the transaction is fully complete.
     opCtx->recoveryUnit()->onCommit(
-        [this, uncommittedMultikeys, setMultikey = std::move(setMultikey)](auto ts) {
+        [this, uncommittedMultikeys, setMultikey = std::move(setMultikey), concurrentWriteTracker](
+            auto ts) {
             // Merge in changes to this index, other indexes may have been updated since we made our
             // copy. Don't check for result as another thread could be setting multikey at the same
             // time
diff --git a/src/mongo/db/catalog/collection_impl.h b/src/mongo/db/catalog/collection_impl.h
index d3379568d22..a6f2043434f 100644
--- a/src/mongo/db/catalog/collection_impl.h
+++ b/src/mongo/db/catalog/collection_impl.h
@@ -331,6 +331,9 @@ public:
 
     bool doesTimeseriesBucketsDocContainMixedSchemaData(const BSONObj& bucketsDoc) const final;
 
+    bool getRequiresTimeseriesExtendedRangeSupport() const final;
+    void setRequiresTimeseriesExtendedRangeSupport(OperationContext* opCtx) const final;
+
     /**
      * isClustered() relies on the object returned from getClusteredInfo(). If
      * ClusteredCollectionInfo exists, the collection is clustered.
@@ -579,6 +582,18 @@ private:
 
         AtomicWord<bool> _committed{true};
 
+        // Time-series collections are allowed to contain measurements with arbitrary dates;
+        // however, many of our query optimizations only work properly with dates that can be stored
+        // as an offset in seconds from the Unix epoch within 31 bits (roughly 1970-2038). When this
+        // flag is set to true, these optimizations will be disabled. It must be set to true if the
+        // collection contains any measurements with dates outside this normal range.
+        //
+        // This is set from the write path where we only hold an IX lock, so we want to be able to
+        // set it from a const method on the Collection. In order to do this, we need to make it
+        // mutable. Given that the value may only transition from false to true, but never back
+        // again, and that we store and retrieve it atomically, this should be safe.
+        mutable AtomicWord<bool> _requiresTimeseriesExtendedRangeSupport{false};
+
         // Capped information.
         const bool _isCapped;
 
diff --git a/src/mongo/db/catalog/collection_mock.h b/src/mongo/db/catalog/collection_mock.h
index 00d21ec21dd..80bb10e95fc 100644
--- a/src/mongo/db/catalog/collection_mock.h
+++ b/src/mongo/db/catalog/collection_mock.h
@@ -257,6 +257,14 @@ public:
         std::abort();
     }
 
+    bool getRequiresTimeseriesExtendedRangeSupport() const {
+        MONGO_UNREACHABLE;
+    }
+
+    void setRequiresTimeseriesExtendedRangeSupport(OperationContext* opCtx) const {
+        MONGO_UNREACHABLE;
+    }
+
     bool isClustered() const {
         return false;
     }
diff --git a/src/mongo/db/catalog/collection_operation_source.cpp b/src/mongo/db/catalog/collection_operation_source.cpp
index 84b291f19ce..198e929c7f4 100644
--- a/src/mongo/db/catalog/collection_operation_source.cpp
+++ b/src/mongo/db/catalog/collection_operation_source.cpp
@@ -38,6 +38,8 @@ StringData toString(OperationSource source) {
     static constexpr StringData kTimeseriesInsertString = "time-series insert"_sd;
     static constexpr StringData kTimeseriesUpdateString = "time-series update"_sd;
     static constexpr StringData kTimeseriesDeleteString = "time-series delete"_sd;
+    static constexpr StringData kTimeseriesBucketCompressionString =
+        "time-series bucket compression"_sd;
 
     switch (source) {
         case OperationSource::kStandard:
@@ -50,6 +52,8 @@ StringData toString(OperationSource source) {
             return kTimeseriesUpdateString;
         case OperationSource::kTimeseriesDelete:
             return kTimeseriesDeleteString;
+        case OperationSource::kTimeseriesBucketCompression:
+            return kTimeseriesBucketCompressionString;
     }
 
     MONGO_UNREACHABLE;
diff --git a/src/mongo/db/catalog/collection_operation_source.h b/src/mongo/db/catalog/collection_operation_source.h
index 6cfff61882f..7546d681850 100644
--- a/src/mongo/db/catalog/collection_operation_source.h
+++ b/src/mongo/db/catalog/collection_operation_source.h
@@ -43,6 +43,7 @@ enum class OperationSource {
     kTimeseriesInsert,
     kTimeseriesUpdate,
     kTimeseriesDelete,
+    kTimeseriesBucketCompression
 };
 
 StringData toString(OperationSource source);
diff --git a/src/mongo/db/catalog/collection_validation.cpp b/src/mongo/db/catalog/collection_validation.cpp
index 42c5ac18dc7..f6e1e4139e7 100644
--- a/src/mongo/db/catalog/collection_validation.cpp
+++ b/src/mongo/db/catalog/collection_validation.cpp
@@ -78,8 +78,7 @@ void _validateIndexesInternalStructure(OperationContext* opCtx,
     // Need to use the IndexCatalog here because the 'validateState->indexes' object hasn't been
     // constructed yet. It must be initialized to ensure we're validating all indexes.
     const IndexCatalog* indexCatalog = validateState->getCollection()->getIndexCatalog();
-    const std::unique_ptr<IndexCatalog::IndexIterator> it =
-        indexCatalog->getIndexIterator(opCtx, false);
+    const auto it = indexCatalog->getIndexIterator(opCtx, IndexCatalog::InclusionPolicy::kReady);
 
     // Validate Indexes Internal Structure, checking if index files have been compromised or
     // corrupted.
@@ -98,14 +97,11 @@ void _validateIndexesInternalStructure(OperationContext* opCtx,
 
         auto& curIndexResults = (results->indexResultsMap)[descriptor->indexName()];
 
-        int64_t numValidated;
-        iam->validate(opCtx, &numValidated, &curIndexResults);
+        iam->validate(opCtx, nullptr, &curIndexResults);
 
         if (!curIndexResults.valid) {
             results->valid = false;
         }
-
-        curIndexResults.keysTraversedFromFullValidate = numValidated;
     }
 }
 
@@ -137,33 +133,6 @@ void _validateIndexes(OperationContext* opCtx,
         auto& curIndexResults = (results->indexResultsMap)[descriptor->indexName()];
         curIndexResults.keysTraversed = numTraversedKeys;
 
-        // If we are performing a full index validation, we have information on the number of index
-        // keys validated in _validateIndexesInternalStructure (when we validated the internal
-        // structure of the index). Check if this is consistent with 'numTraversedKeys' from
-        // traverseIndex above.
-        if (validateState->isFullIndexValidation()) {
-            invariant(opCtx->lockState()->isCollectionLockedForMode(validateState->nss(), MODE_X));
-
-            // The number of keys counted in _validateIndexesInternalStructure, when checking the
-            // internal structure of the index.
-            const int64_t numIndexKeys = curIndexResults.keysTraversedFromFullValidate;
-
-            // Check if currIndexResults is valid to ensure that this index is not corrupted or
-            // comprised (which was set in _validateIndexesInternalStructure). If the index is
-            // corrupted, there is no use in checking if the traversal yielded the same key count.
-            if (curIndexResults.valid) {
-                if (numIndexKeys != numTraversedKeys) {
-                    curIndexResults.valid = false;
-                    string msg = str::stream()
-                        << "number of traversed index entries (" << numTraversedKeys
-                        << ") does not match the number of expected index entries (" << numIndexKeys
-                        << ")";
-                    results->errors.push_back(msg);
-                    results->valid = false;
-                }
-            }
-        }
-
         if (!curIndexResults.valid) {
             results->valid = false;
         }
@@ -230,7 +199,7 @@ void _gatherIndexEntryErrors(OperationContext* opCtx,
 
     LOGV2_OPTIONS(20301, {LogComponent::kIndex}, "Finished traversing through all the indexes");
 
-    indexConsistency->addIndexEntryErrors(result);
+    indexConsistency->addIndexEntryErrors(opCtx, result);
 }
 
 void _validateIndexKeyCount(OperationContext* opCtx,
@@ -247,6 +216,81 @@ void _validateIndexKeyCount(OperationContext* opCtx,
     }
 }
 
+void _printIndexSpec(const ValidateState* validateState, StringData indexName) {
+    auto& indexes = validateState->getIndexes();
+    auto indexEntry =
+        std::find_if(indexes.begin(),
+                     indexes.end(),
+                     [&](const std::shared_ptr<const IndexCatalogEntry> indexEntry) -> bool {
+                         return indexEntry->descriptor()->indexName() == indexName;
+                     });
+    if (indexEntry != indexes.end()) {
+        auto indexSpec = (*indexEntry)->descriptor()->infoObj();
+        LOGV2_ERROR(7463100, "Index failed validation", "spec"_attr = indexSpec);
+    }
+}
+
+/**
+ * Logs oplog entries related to corrupted records/indexes in validation results.
+ */
+void _logOplogEntriesForInvalidResults(OperationContext* opCtx, ValidateResults* results) {
+    if (results->recordTimestamps.empty()) {
+        return;
+    }
+
+    LOGV2(
+        7464200,
+        "Validation failed: oplog timestamps referenced by corrupted collection and index entries",
+        "numTimestamps"_attr = results->recordTimestamps.size());
+
+    // Set up read on oplog collection.
+    try {
+        AutoGetOplog oplogRead(opCtx, OplogAccessMode::kRead);
+        const auto& oplogCollection = oplogRead.getCollection();
+
+        // Log oplog entries in reverse from most recent timestamp to oldest.
+        // Due to oplog truncation, if we fail to find any oplog entry for a particular timestamp,
+        // we can stop searching for oplog entries with earlier timestamps.
+        auto recordStore = oplogCollection->getRecordStore();
+        uassert(ErrorCodes::InternalError,
+                "Validation failed: Unable to get oplog record store for corrupted collection and "
+                "index entries",
+                recordStore);
+
+        auto cursor = recordStore->getCursor(opCtx, /*forward=*/false);
+        uassert(ErrorCodes::CursorNotFound,
+                "Validation failed: Unable to get cursor to oplog collection.",
+                cursor);
+
+        for (auto it = results->recordTimestamps.rbegin(); it != results->recordTimestamps.rend();
+             it++) {
+            const auto& timestamp = *it;
+
+            // A record id in the oplog collection is equivalent to the document's timestamp field.
+            RecordId recordId(timestamp.asULL());
+            auto record = cursor->seekExact(recordId);
+            if (!record) {
+                LOGV2(7464201,
+                      "    Validation failed: Stopping oplog entry search for corrupted collection "
+                      "and index entries.",
+                      "timestamp"_attr = timestamp);
+                break;
+            }
+
+            LOGV2(
+                7464202,
+                "    Validation failed: Oplog entry found for corrupted collection and index entry",
+                "timestamp"_attr = timestamp,
+                "oplogEntryDoc"_attr = redact(record->data.toBson()));
+        }
+    } catch (DBException& ex) {
+        LOGV2_ERROR(7464203,
+                    "Validation failed: Unable to fetch entries from oplog collection for "
+                    "corrupted collection and index entries",
+                    "ex"_attr = ex);
+    }
+}
+
 void _reportValidationResults(OperationContext* opCtx,
                               ValidateState* validateState,
                               ValidateResults* results,
@@ -263,17 +307,19 @@ void _reportValidationResults(OperationContext* opCtx,
 
     // Report detailed index validation results gathered when using {full: true} for validated
     // indexes.
-    for (const auto& index : validateState->getIndexes()) {
-        const std::string indexName = index->descriptor()->indexName();
-        auto& indexResultsMap = results->indexResultsMap;
-        if (indexResultsMap.find(indexName) == indexResultsMap.end()) {
-            continue;
-        }
-
-        auto& vr = indexResultsMap.at(indexName);
-
+    int nIndexes = results->indexResultsMap.size();
+    for (const auto& [indexName, vr] : results->indexResultsMap) {
         if (!vr.valid) {
             results->valid = false;
+            _printIndexSpec(validateState, indexName);
+        }
+
+        if (validateState->getSkippedIndexes().contains(indexName)) {
+            // Index internal state was checked and cleared, so it was reported in indexResultsMap,
+            // but we did not verify the index contents against the collection, so we should exclude
+            // it from this report.
+            --nIndexes;
+            continue;
         }
 
         BSONObjBuilder bob(indexDetails.subobjStart(indexName));
@@ -294,7 +340,7 @@ void _reportValidationResults(OperationContext* opCtx,
         results->errors.insert(results->errors.end(), vr.errors.begin(), vr.errors.end());
     }
 
-    output->append("nIndexes", static_cast<int>(validateState->getIndexes().size()));
+    output->append("nIndexes", nIndexes);
     output->append("keysPerIndex", keysPerIndex.done());
     output->append("indexDetails", indexDetails.done());
 }
@@ -304,6 +350,7 @@ void _reportInvalidResults(OperationContext* opCtx,
                            ValidateResults* results,
                            BSONObjBuilder* output) {
     _reportValidationResults(opCtx, validateState, results, output);
+    _logOplogEntriesForInvalidResults(opCtx, results);
     LOGV2_OPTIONS(20302,
                   {LogComponent::kIndex},
                   "Validation complete -- Corruption found",
@@ -410,7 +457,10 @@ void _validateCatalogEntry(OperationContext* opCtx,
     }
 
     const auto& indexCatalog = collection->getIndexCatalog();
-    auto indexIt = indexCatalog->getIndexIterator(opCtx, /*includeUnfinishedIndexes=*/true);
+    auto indexIt = indexCatalog->getIndexIterator(opCtx,
+                                                  IndexCatalog::InclusionPolicy::kReady |
+                                                      IndexCatalog::InclusionPolicy::kUnfinished |
+                                                      IndexCatalog::InclusionPolicy::kFrozen);
 
     while (indexIt->more()) {
         const IndexCatalogEntry* indexEntry = indexIt->next();
@@ -561,12 +611,12 @@ Status validate(OperationContext* opCtx,
                 RepairMode repairMode,
                 ValidateResults* results,
                 BSONObjBuilder* output,
-                bool turnOnExtraLoggingForTest) {
+                bool logDiagnostics) {
     invariant(!opCtx->lockState()->isLocked() || storageGlobalParams.repair);
 
     // This is deliberately outside of the try-catch block, so that any errors thrown in the
     // constructor fail the cmd, as opposed to returning OK with valid:false.
-    ValidateState validateState(opCtx, nss, mode, repairMode, turnOnExtraLoggingForTest);
+    ValidateState validateState(opCtx, nss, mode, repairMode, logDiagnostics);
 
     const auto replCoord = repl::ReplicationCoordinator::get(opCtx);
     // Check whether we are allowed to read from this node after acquiring our locks. If we are
diff --git a/src/mongo/db/catalog/collection_validation.h b/src/mongo/db/catalog/collection_validation.h
index 378244f39b7..40b12b0c8c5 100644
--- a/src/mongo/db/catalog/collection_validation.h
+++ b/src/mongo/db/catalog/collection_validation.h
@@ -103,7 +103,7 @@ Status validate(OperationContext* opCtx,
                 RepairMode repairMode,
                 ValidateResults* results,
                 BSONObjBuilder* output,
-                bool turnOnExtraLoggingForTest = false);
+                bool logDiagnostics);
 
 /**
  * Checks whether a failpoint has been hit in the above validate() code..
diff --git a/src/mongo/db/catalog/collection_validation_test.cpp b/src/mongo/db/catalog/collection_validation_test.cpp
index 4ffff4f653b..a81ee7bd2fe 100644
--- a/src/mongo/db/catalog/collection_validation_test.cpp
+++ b/src/mongo/db/catalog/collection_validation_test.cpp
@@ -47,6 +47,9 @@ namespace {
 const NamespaceString kNss = NamespaceString("test.t");
 
 class CollectionValidationTest : public CatalogTestFixture {
+protected:
+    CollectionValidationTest(Options options = {}) : CatalogTestFixture(std::move(options)) {}
+
 private:
     void setUp() override {
         CatalogTestFixture::setUp();
@@ -58,6 +61,11 @@ private:
     };
 };
 
+class CollectionValidationDiskTest : public CollectionValidationTest {
+protected:
+    CollectionValidationDiskTest() : CollectionValidationTest(Options{}.ephemeral(false)) {}
+};
+
 /**
  * Calls validate on collection kNss with both kValidateFull and kValidateNormal validation levels
  * and verifies the results.
@@ -79,7 +87,7 @@ std::vector<std::pair<BSONObj, ValidateResults>> foregroundValidate(
         ValidateResults validateResults;
         BSONObjBuilder output;
         ASSERT_OK(CollectionValidation::validate(
-            opCtx, kNss, mode, repairMode, &validateResults, &output));
+            opCtx, kNss, mode, repairMode, &validateResults, &output, /*logDiagnostics=*/false));
         BSONObj obj = output.obj();
         BSONObjBuilder validateResultsBuilder;
         validateResults.appendToResultObj(&validateResultsBuilder, true /* debugging */);
@@ -146,7 +154,8 @@ void backgroundValidate(OperationContext* opCtx,
                                              CollectionValidation::ValidateMode::kBackground,
                                              CollectionValidation::RepairMode::kNone,
                                              &validateResults,
-                                             &output));
+                                             &output,
+                                             /*logDiagnostics=*/false));
     BSONObj obj = output.obj();
 
     ASSERT_EQ(validateResults.valid, valid);
@@ -270,6 +279,19 @@ TEST_F(CollectionValidationTest, ValidateEnforceFastCount) {
                        {CollectionValidation::ValidateMode::kForegroundFullEnforceFastCount});
 }
 
+TEST_F(CollectionValidationDiskTest, ValidateIndexDetailResultsSurfaceVerifyErrors) {
+    FailPointEnableBlock fp{"WTValidateIndexStructuralDamage"};
+    auto opCtx = operationContext();
+    insertDataRange(opCtx, 0, 5);  // initialize collection
+    foregroundValidate(
+        opCtx,
+        /*valid*/ false,
+        /*numRecords*/ std::numeric_limits<int32_t>::min(),           // uninitialized
+        /*numInvalidDocuments*/ std::numeric_limits<int32_t>::min(),  // uninitialized
+        /*numErrors*/ 1,
+        {CollectionValidation::ValidateMode::kForegroundFull});
+}
+
 /**
  * Waits for a parallel running collection validation operation to start and then hang at a
  * failpoint.
diff --git a/src/mongo/db/catalog/create_collection.cpp b/src/mongo/db/catalog/create_collection.cpp
index 74be2bd73e4..c64decf02a3 100644
--- a/src/mongo/db/catalog/create_collection.cpp
+++ b/src/mongo/db/catalog/create_collection.cpp
@@ -44,7 +44,7 @@
 #include "mongo/db/catalog/index_key_validate.h"
 #include "mongo/db/commands.h"
 #include "mongo/db/commands/create_gen.h"
-#include "mongo/db/concurrency/write_conflict_exception.h"
+#include "mongo/db/concurrency/exception_util.h"
 #include "mongo/db/curop.h"
 #include "mongo/db/db_raii.h"
 #include "mongo/db/index/index_descriptor.h"
@@ -117,7 +117,9 @@ Status validateClusteredIndexSpec(OperationContext* opCtx,
 
     if (expireAfterSeconds) {
         // Not included in the indexSpec itself.
-        auto status = index_key_validate::validateExpireAfterSeconds(*expireAfterSeconds);
+        auto status = index_key_validate::validateExpireAfterSeconds(
+            *expireAfterSeconds,
+            index_key_validate::ValidateExpireAfterSecondsMode::kClusteredTTLIndex);
         if (!status.isOK()) {
             return status;
         }
@@ -206,6 +208,117 @@ Status _createView(OperationContext* opCtx,
     });
 }
 
+BSONObj _generateTimeseriesValidator(StringData timeField) {
+    // '$jsonSchema' : {
+    //     bsonType: 'object',
+    //     required: ['_id', 'control', 'data'],
+    //     properties: {
+    //         _id: {bsonType: 'objectId'},
+    //         control: {
+    //             bsonType: 'object',
+    //             required: ['version', 'min', 'max'],
+    //             properties: {
+    //                 version: {bsonType: 'number'},
+    //                 min: {
+    //                     bsonType: 'object',
+    //                     required: ['%s'],
+    //                     properties: {'%s': {bsonType: 'date'}}
+    //                 },
+    //                 max: {
+    //                     bsonType: 'object',
+    //                     required: ['%s'],
+    //                     properties: {'%s': {bsonType: 'date'}}
+    //                 },
+    //                 closed: {bsonType: 'bool'},
+    //                 count: {bsonType: 'number', minimum: 1},
+    //             },
+    //             additionalProperties: false,
+    //         },
+    //         data: {bsonType: 'object'},
+    //         meta: {}
+    //     },
+    //     additionalProperties: false
+    //   }
+    BSONObjBuilder validator;
+    BSONObjBuilder schema(validator.subobjStart("$jsonSchema"));
+    schema.append("bsonType", "object");
+    schema.append("required",
+                  BSON_ARRAY("_id"
+                             << "control"
+                             << "data"));
+    {
+        BSONObjBuilder properties(schema.subobjStart("properties"));
+        {
+            BSONObjBuilder _id(properties.subobjStart("_id"));
+            _id.append("bsonType", "objectId");
+            _id.done();
+        }
+        {
+            BSONObjBuilder control(properties.subobjStart("control"));
+            control.append("bsonType", "object");
+            control.append("required",
+                           BSON_ARRAY("version"
+                                      << "min"
+                                      << "max"));
+            {
+                BSONObjBuilder innerProperties(control.subobjStart("properties"));
+                {
+                    BSONObjBuilder version(innerProperties.subobjStart("version"));
+                    version.append("bsonType", "number");
+                    version.done();
+                }
+                {
+                    BSONObjBuilder min(innerProperties.subobjStart("min"));
+                    min.append("bsonType", "object");
+                    min.append("required", BSON_ARRAY(timeField));
+                    BSONObjBuilder minProperties(min.subobjStart("properties"));
+                    BSONObjBuilder timeFieldObj(minProperties.subobjStart(timeField));
+                    timeFieldObj.append("bsonType", "date");
+                    timeFieldObj.done();
+                    minProperties.done();
+                    min.done();
+                }
+
+                {
+                    BSONObjBuilder max(innerProperties.subobjStart("max"));
+                    max.append("bsonType", "object");
+                    max.append("required", BSON_ARRAY(timeField));
+                    BSONObjBuilder maxProperties(max.subobjStart("properties"));
+                    BSONObjBuilder timeFieldObj(maxProperties.subobjStart(timeField));
+                    timeFieldObj.append("bsonType", "date");
+                    timeFieldObj.done();
+                    maxProperties.done();
+                    max.done();
+                }
+                {
+                    BSONObjBuilder closed(innerProperties.subobjStart("closed"));
+                    closed.append("bsonType", "bool");
+                    closed.done();
+                }
+                {
+                    BSONObjBuilder count(innerProperties.subobjStart("count"));
+                    count.append("bsonType", "number");
+                    count.append("minimum", 1);
+                    count.done();
+                }
+                innerProperties.done();
+            }
+            control.append("additionalProperties", false);
+            control.done();
+        }
+        {
+            BSONObjBuilder data(properties.subobjStart("data"));
+            data.append("bsonType", "object");
+            data.done();
+        }
+        properties.append("meta", BSONObj{});
+        properties.done();
+    }
+    schema.append("additionalProperties", false);
+    schema.done();
+    return validator.obj();
+}
+
 Status _createTimeseries(OperationContext* opCtx,
                          const NamespaceString& ns,
                          const CollectionOptions& optionsArg) {
@@ -230,46 +343,13 @@ Status _createTimeseries(OperationContext* opCtx,
                 maxSpanSeconds == options.timeseries->getBucketMaxSpanSeconds());
     options.timeseries->setBucketMaxSpanSeconds(maxSpanSeconds);
 
+
     // Set the validator option to a JSON schema enforcing constraints on bucket documents.
     // This validation is only structural to prevent accidental corruption by users and
     // cannot cover all constraints. Leave the validationLevel and validationAction to their
     // strict/error defaults.
     auto timeField = options.timeseries->getTimeField();
-    auto validatorObj = fromjson(fmt::sprintf(R"(
-{
-'$jsonSchema' : {
-    bsonType: 'object',
-    required: ['_id', 'control', 'data'],
-    properties: {
-        _id: {bsonType: 'objectId'},
-        control: {
-            bsonType: 'object',
-            required: ['version', 'min', 'max'],
-            properties: {
-                version: {bsonType: 'number'},
-                min: {
-                    bsonType: 'object',
-                    required: ['%s'],
-                    properties: {'%s': {bsonType: 'date'}}
-                },
-                max: {
-                    bsonType: 'object',
-                    required: ['%s'],
-                    properties: {'%s': {bsonType: 'date'}}
-                },
-                closed: {bsonType: 'bool'}
-            }
-        },
-        data: {bsonType: 'object'},
-        meta: {}
-    },
-    additionalProperties: false
-}
-})",
-                                              timeField,
-                                              timeField,
-                                              timeField,
-                                              timeField));
+    auto validatorObj = _generateTimeseriesValidator(timeField);
 
     bool existingBucketCollectionIsCompatible = false;
 
@@ -321,8 +401,9 @@ Status _createTimeseries(OperationContext* opCtx,
             // Cluster time-series buckets collections by _id.
             auto expireAfterSeconds = options.expireAfterSeconds;
             if (expireAfterSeconds) {
-                uassertStatusOK(
-                    index_key_validate::validateExpireAfterSeconds(*expireAfterSeconds));
+                uassertStatusOK(index_key_validate::validateExpireAfterSeconds(
+                    *expireAfterSeconds,
+                    index_key_validate::ValidateExpireAfterSecondsMode::kClusteredTTLIndex));
                 bucketsOptions.expireAfterSeconds = expireAfterSeconds;
             }
 
diff --git a/src/mongo/db/catalog/create_collection_test.cpp b/src/mongo/db/catalog/create_collection_test.cpp
index 3141d18905c..b5cce6e6686 100644
--- a/src/mongo/db/catalog/create_collection_test.cpp
+++ b/src/mongo/db/catalog/create_collection_test.cpp
@@ -34,7 +34,7 @@
 #include "mongo/db/catalog/collection_catalog.h"
 #include "mongo/db/catalog/create_collection.h"
 #include "mongo/db/catalog/database_holder.h"
-#include "mongo/db/concurrency/write_conflict_exception.h"
+#include "mongo/db/concurrency/exception_util.h"
 #include "mongo/db/db_raii.h"
 #include "mongo/db/jsobj.h"
 #include "mongo/db/repl/replication_coordinator.h"
diff --git a/src/mongo/db/catalog/database.h b/src/mongo/db/catalog/database.h
index 2640abefbcb..f3a5a1393be 100644
--- a/src/mongo/db/catalog/database.h
+++ b/src/mongo/db/catalog/database.h
@@ -110,13 +110,18 @@ public:
      * If we are applying a 'drop' oplog entry on a secondary, 'dropOpTime' will contain the optime
      * of the oplog entry.
      *
+     * When fromMigrate is set, the related oplog entry will be marked with a 'fromMigrate' field to
+     * reduce its visibility (e.g. in change streams).
+     *
      * The caller should hold a DB X lock and ensure there are no index builds in progress on the
      * collection.
      * N.B. Namespace argument is passed by value as it may otherwise disappear or change.
      */
     virtual Status dropCollection(OperationContext* opCtx,
                                   NamespaceString nss,
-                                  repl::OpTime dropOpTime = {}) const = 0;
+                                  repl::OpTime dropOpTime = {},
+                                  bool markFromMigrate = false) const = 0;
+
     virtual Status dropCollectionEvenIfSystem(OperationContext* opCtx,
                                               NamespaceString nss,
                                               repl::OpTime dropOpTime = {},
diff --git a/src/mongo/db/catalog/database_holder_impl.cpp b/src/mongo/db/catalog/database_holder_impl.cpp
index 28e44082c2a..ab643d7af2f 100644
--- a/src/mongo/db/catalog/database_holder_impl.cpp
+++ b/src/mongo/db/catalog/database_holder_impl.cpp
@@ -37,7 +37,7 @@
 #include "mongo/db/catalog/collection_catalog.h"
 #include "mongo/db/catalog/collection_impl.h"
 #include "mongo/db/catalog/database_impl.h"
-#include "mongo/db/concurrency/write_conflict_exception.h"
+#include "mongo/db/concurrency/exception_util.h"
 #include "mongo/db/index_builds_coordinator.h"
 #include "mongo/db/op_observer.h"
 #include "mongo/db/operation_context.h"
diff --git a/src/mongo/db/catalog/database_impl.cpp b/src/mongo/db/catalog/database_impl.cpp
index 077fc8d2313..62f789ad003 100644
--- a/src/mongo/db/catalog/database_impl.cpp
+++ b/src/mongo/db/catalog/database_impl.cpp
@@ -299,7 +299,7 @@ void DatabaseImpl::clearTmpCollections(OperationContext* opCtx) const {
     CollectionCatalog::CollectionInfoFn callback = [&](const CollectionPtr& collection) {
         try {
             WriteUnitOfWork wuow(opCtx);
-            Status status = dropCollection(opCtx, collection->ns(), {});
+            Status status = dropCollection(opCtx, collection->ns(), {}, false);
             if (!status.isOK()) {
                 LOGV2_WARNING(20327,
                               "could not drop temp collection '{namespace}': {error}",
@@ -438,7 +438,8 @@ Status DatabaseImpl::dropView(OperationContext* opCtx, NamespaceString viewName)
 
 Status DatabaseImpl::dropCollection(OperationContext* opCtx,
                                     NamespaceString nss,
-                                    repl::OpTime dropOpTime) const {
+                                    repl::OpTime dropOpTime,
+                                    bool markFromMigrate) const {
     // Cannot drop uncommitted collections.
     invariant(!UncommittedCatalogUpdates::isCreatedCollection(opCtx, nss));
 
@@ -474,7 +475,7 @@ Status DatabaseImpl::dropCollection(OperationContext* opCtx,
         }
     }
 
-    return dropCollectionEvenIfSystem(opCtx, nss, dropOpTime);
+    return dropCollectionEvenIfSystem(opCtx, nss, dropOpTime, markFromMigrate);
 }
 
 Status DatabaseImpl::dropCollectionEvenIfSystem(OperationContext* opCtx,
@@ -744,7 +745,7 @@ void DatabaseImpl::_checkCanCreateCollection(OperationContext* opCtx,
                                              const CollectionOptions& options) const {
     if (CollectionCatalog::get(opCtx)->lookupCollectionByNamespace(opCtx, nss)) {
         if (options.isView()) {
-            uasserted(17399,
+            uasserted(ErrorCodes::NamespaceExists,
                       str::stream()
                           << "Cannot create collection " << nss << " - collection already exists.");
         } else {
@@ -1089,6 +1090,10 @@ Status DatabaseImpl::userCreateNS(OperationContext* opCtx,
                                                               ExtensionsCallbackNoop(),
                                                               allowedFeatures);
 
+        // Increment counters to track the usage of schema validators.
+        validatorCounters.incrementCounters(
+            "create", collectionOptions.validator, statusWithMatcher.isOK());
+
         // We check the status of the parse to see if there are any banned features, but we don't
         // actually need the result for now.
         if (!statusWithMatcher.isOK()) {
diff --git a/src/mongo/db/catalog/database_impl.h b/src/mongo/db/catalog/database_impl.h
index fa09177bc1b..c87f811c7d8 100644
--- a/src/mongo/db/catalog/database_impl.h
+++ b/src/mongo/db/catalog/database_impl.h
@@ -62,16 +62,20 @@ public:
      * If we are applying a 'drop' oplog entry on a secondary, 'dropOpTime' will contain the optime
      * of the oplog entry.
      *
+     * When fromMigrate is set, the related oplog entry will be marked with a 'fromMigrate' field to
+     * reduce its visibility (e.g. in change streams).
+     *
      * The caller should hold a DB X lock and ensure there are no index builds in progress on the
      * collection.
      */
     Status dropCollection(OperationContext* opCtx,
                           NamespaceString nss,
-                          repl::OpTime dropOpTime) const final;
+                          repl::OpTime dropOpTime,
+                          bool markFromMigrate) const final;
     Status dropCollectionEvenIfSystem(OperationContext* opCtx,
                                       NamespaceString nss,
                                       repl::OpTime dropOpTime,
-                                      bool markFromMigrate = false) const final;
+                                      bool markFromMigrate) const final;
 
     Status dropView(OperationContext* opCtx, NamespaceString viewName) const final;
 
diff --git a/src/mongo/db/catalog/database_test.cpp b/src/mongo/db/catalog/database_test.cpp
index dd72c38588c..59308be7233 100644
--- a/src/mongo/db/catalog/database_test.cpp
+++ b/src/mongo/db/catalog/database_test.cpp
@@ -39,7 +39,7 @@
 #include "mongo/db/catalog/index_catalog.h"
 #include "mongo/db/client.h"
 #include "mongo/db/concurrency/d_concurrency.h"
-#include "mongo/db/concurrency/write_conflict_exception.h"
+#include "mongo/db/concurrency/exception_util.h"
 #include "mongo/db/db_raii.h"
 #include "mongo/db/index/index_descriptor.h"
 #include "mongo/db/jsobj.h"
diff --git a/src/mongo/db/catalog/drop_collection.cpp b/src/mongo/db/catalog/drop_collection.cpp
index f04a8fa2b79..12e78022b2b 100644
--- a/src/mongo/db/catalog/drop_collection.cpp
+++ b/src/mongo/db/catalog/drop_collection.cpp
@@ -36,9 +36,10 @@
 #include "mongo/db/audit.h"
 #include "mongo/db/catalog/collection_catalog.h"
 #include "mongo/db/catalog/collection_uuid_mismatch.h"
+#include "mongo/db/catalog/collection_uuid_mismatch_info.h"
 #include "mongo/db/catalog/index_catalog.h"
 #include "mongo/db/client.h"
-#include "mongo/db/concurrency/write_conflict_exception.h"
+#include "mongo/db/concurrency/exception_util.h"
 #include "mongo/db/curop.h"
 #include "mongo/db/db_raii.h"
 #include "mongo/db/index_builds_coordinator.h"
@@ -56,7 +57,16 @@ namespace {
 MONGO_FAIL_POINT_DEFINE(hangDropCollectionBeforeLockAcquisition);
 MONGO_FAIL_POINT_DEFINE(hangDuringDropCollection);
 
-Status _checkNssAndReplState(OperationContext* opCtx, const CollectionPtr& coll) {
+Status _checkNssAndReplState(OperationContext* opCtx,
+                             const CollectionPtr& coll,
+                             const NamespaceString& nss,
+                             const boost::optional<UUID>& expectedUUID = boost::none) {
+    try {
+        checkCollectionUUIDMismatch(opCtx, nss, coll, expectedUUID);
+    } catch (const DBException& ex) {
+        return ex.toStatus();
+    }
+
     if (!coll) {
         return Status(ErrorCodes::NamespaceNotFound, "ns not found");
     }
@@ -195,19 +205,13 @@ Status _abortIndexBuildsAndDrop(OperationContext* opCtx,
 
     CollectionPtr coll =
         CollectionCatalog::get(opCtx)->lookupCollectionByNamespace(opCtx, startingNss);
-    Status status = _checkNssAndReplState(opCtx, coll);
+    Status status = _checkNssAndReplState(opCtx, coll, startingNss, expectedUUID);
     if (!status.isOK()) {
         return status;
     }
 
     warnEncryptedCollectionsIfNeeded(opCtx, coll);
 
-    try {
-        checkCollectionUUIDMismatch(opCtx, startingNss, coll, expectedUUID);
-    } catch (const DBException& ex) {
-        return ex.toStatus();
-    }
-
     if (MONGO_unlikely(hangDuringDropCollection.shouldFail())) {
         LOGV2(518090,
               "hangDuringDropCollection fail point enabled. Blocking until fail point is "
@@ -256,7 +260,7 @@ Status _abortIndexBuildsAndDrop(OperationContext* opCtx,
         opCtx->recoveryUnit()->abandonSnapshot();
 
         coll = CollectionCatalog::get(opCtx)->lookupCollectionByUUID(opCtx, collectionUUID);
-        status = _checkNssAndReplState(opCtx, coll);
+        status = _checkNssAndReplState(opCtx, coll, startingNss, expectedUUID);
         if (!status.isOK()) {
             return status;
         }
@@ -304,7 +308,7 @@ Status _dropCollectionForApplyOps(OperationContext* opCtx,
     Lock::CollectionLock collLock(opCtx, collectionName, MODE_X);
     const CollectionPtr& coll =
         CollectionCatalog::get(opCtx)->lookupCollectionByNamespace(opCtx, collectionName);
-    Status status = _checkNssAndReplState(opCtx, coll);
+    Status status = _checkNssAndReplState(opCtx, coll, collectionName);
     if (!status.isOK()) {
         return status;
     }
@@ -348,6 +352,7 @@ Status _dropCollection(OperationContext* opCtx,
                        const boost::optional<UUID>& expectedUUID,
                        DropReply* reply,
                        DropCollectionSystemCollectionMode systemCollectionMode,
+                       bool fromMigrate,
                        boost::optional<UUID> dropIfUUIDNotMatching = boost::none) {
 
     try {
@@ -355,7 +360,13 @@ Status _dropCollection(OperationContext* opCtx,
             AutoGetDb autoDb(opCtx, collectionName.db(), MODE_IX);
             auto db = autoDb.getDb();
             if (!db) {
-                return Status(ErrorCodes::NamespaceNotFound, "ns not found");
+                return expectedUUID
+                    ? Status{CollectionUUIDMismatchInfo(collectionName.db().toString(),
+                                                        *expectedUUID,
+                                                        collectionName.coll().toString(),
+                                                        boost::none),
+                             "Database does not exist"}
+                    : Status(ErrorCodes::NamespaceNotFound, "ns not found");
             }
 
             if (CollectionCatalog::get(opCtx)->lookupCollectionByNamespace(opCtx, collectionName)) {
@@ -364,13 +375,14 @@ Status _dropCollection(OperationContext* opCtx,
                     std::move(autoDb),
                     collectionName,
                     expectedUUID,
-                    [opCtx, systemCollectionMode](Database* db, const NamespaceString& resolvedNs) {
+                    [opCtx, systemCollectionMode, fromMigrate](Database* db,
+                                                               const NamespaceString& resolvedNs) {
                         WriteUnitOfWork wuow(opCtx);
 
                         auto status = systemCollectionMode ==
                                 DropCollectionSystemCollectionMode::kDisallowSystemCollectionDrops
-                            ? db->dropCollection(opCtx, resolvedNs)
-                            : db->dropCollectionEvenIfSystem(opCtx, resolvedNs);
+                            ? db->dropCollection(opCtx, resolvedNs, {}, fromMigrate)
+                            : db->dropCollectionEvenIfSystem(opCtx, resolvedNs, {}, fromMigrate);
                         if (!status.isOK()) {
                             return status;
                         }
@@ -383,14 +395,18 @@ Status _dropCollection(OperationContext* opCtx,
                     dropIfUUIDNotMatching);
             }
 
-            auto dropTimeseries = [opCtx, &expectedUUID, &autoDb, &collectionName, &reply](
-                                      const NamespaceString& bucketNs, bool dropView) {
+            auto dropTimeseries = [opCtx,
+                                   &expectedUUID,
+                                   &autoDb,
+                                   &collectionName,
+                                   &reply,
+                                   fromMigrate](const NamespaceString& bucketNs, bool dropView) {
                 return _abortIndexBuildsAndDrop(
                     opCtx,
                     std::move(autoDb),
                     bucketNs,
                     expectedUUID,
-                    [opCtx, dropView, &expectedUUID, &collectionName, &reply](
+                    [opCtx, dropView, &expectedUUID, &collectionName, &reply, fromMigrate](
                         Database* db, const NamespaceString& bucketsNs) {
                         // Disallow checking the expectedUUID when dropping time-series collections.
                         uassert(ErrorCodes::InvalidOptions,
@@ -414,11 +430,13 @@ Status _dropCollection(OperationContext* opCtx,
 
                         // Drop the buckets collection in its own writeConflictRetry so that if
                         // it throws a WCE, only the buckets collection drop is retried.
-                        writeConflictRetry(opCtx, "drop", bucketsNs.ns(), [opCtx, db, &bucketsNs] {
-                            WriteUnitOfWork wuow(opCtx);
-                            db->dropCollectionEvenIfSystem(opCtx, bucketsNs).ignore();
-                            wuow.commit();
-                        });
+                        writeConflictRetry(
+                            opCtx, "drop", bucketsNs.ns(), [opCtx, db, &bucketsNs, fromMigrate] {
+                                WriteUnitOfWork wuow(opCtx);
+                                db->dropCollectionEvenIfSystem(opCtx, bucketsNs, {}, fromMigrate)
+                                    .ignore();
+                                wuow.commit();
+                            });
 
                         return Status::OK();
                     },
@@ -468,7 +486,8 @@ Status dropCollection(OperationContext* opCtx,
                       const NamespaceString& nss,
                       const boost::optional<UUID>& expectedUUID,
                       DropReply* reply,
-                      DropCollectionSystemCollectionMode systemCollectionMode) {
+                      DropCollectionSystemCollectionMode systemCollectionMode,
+                      bool fromMigrate) {
     if (!serverGlobalParams.quiet.load()) {
         LOGV2(518070, "CMD: drop", logAttrs(nss));
     }
@@ -483,14 +502,16 @@ Status dropCollection(OperationContext* opCtx,
     const auto collectionName =
         nss.isTimeseriesBucketsCollection() ? nss.getTimeseriesViewNamespace() : nss;
 
-    return _dropCollection(opCtx, collectionName, expectedUUID, reply, systemCollectionMode);
+    return _dropCollection(
+        opCtx, collectionName, expectedUUID, reply, systemCollectionMode, fromMigrate);
 }
 
 Status dropCollection(OperationContext* opCtx,
                       const NamespaceString& nss,
                       DropReply* reply,
-                      DropCollectionSystemCollectionMode systemCollectionMode) {
-    return dropCollection(opCtx, nss, boost::none, reply, systemCollectionMode);
+                      DropCollectionSystemCollectionMode systemCollectionMode,
+                      bool fromMigrate) {
+    return dropCollection(opCtx, nss, boost::none, reply, systemCollectionMode, fromMigrate);
 }
 
 Status dropCollectionIfUUIDNotMatching(OperationContext* opCtx,
@@ -512,6 +533,7 @@ Status dropCollectionIfUUIDNotMatching(OperationContext* opCtx,
                                boost::none,
                                &repl,
                                DropCollectionSystemCollectionMode::kDisallowSystemCollectionDrops,
+                               false /*fromMigrate*/,
                                expectedUUID);
     }
 
diff --git a/src/mongo/db/catalog/drop_collection.h b/src/mongo/db/catalog/drop_collection.h
index 7f857d88547..016bc5e3d1a 100644
--- a/src/mongo/db/catalog/drop_collection.h
+++ b/src/mongo/db/catalog/drop_collection.h
@@ -50,17 +50,21 @@ enum class DropCollectionSystemCollectionMode {
  * Drops the collection "collectionName" and populates "reply" with statistics about what
  * was removed. Aborts in-progress index builds on the collection if two phase index builds are
  * supported. Throws if the expectedUUID does not match the UUID of the collection being dropped.
+ * When fromMigrate is set, the related oplog entry will be marked accordingly using the
+ * 'fromMigrate' field to reduce its visibility (e.g. in change streams).
  */
 Status dropCollection(OperationContext* opCtx,
                       const NamespaceString& collectionName,
                       const boost::optional<UUID>& expectedUUID,
                       DropReply* reply,
-                      DropCollectionSystemCollectionMode systemCollectionMode);
+                      DropCollectionSystemCollectionMode systemCollectionMode,
+                      bool fromMigrate = false);
 
 Status dropCollection(OperationContext* opCtx,
                       const NamespaceString& collectionName,
                       DropReply* reply,
-                      DropCollectionSystemCollectionMode systemCollectionMode);
+                      DropCollectionSystemCollectionMode systemCollectionMode,
+                      bool fromMigrate = false);
 
 /**
  * Drops the collection "collectionName" only if its uuid is not matching "expectedUUID".
diff --git a/src/mongo/db/catalog/drop_database.cpp b/src/mongo/db/catalog/drop_database.cpp
index 85e2087a321..631378885e2 100644
--- a/src/mongo/db/catalog/drop_database.cpp
+++ b/src/mongo/db/catalog/drop_database.cpp
@@ -38,7 +38,7 @@
 #include "mongo/db/catalog/database_holder.h"
 #include "mongo/db/catalog_raii.h"
 #include "mongo/db/client.h"
-#include "mongo/db/concurrency/write_conflict_exception.h"
+#include "mongo/db/concurrency/exception_util.h"
 #include "mongo/db/curop.h"
 #include "mongo/db/index_builds_coordinator.h"
 #include "mongo/db/op_observer.h"
diff --git a/src/mongo/db/catalog/drop_database_test.cpp b/src/mongo/db/catalog/drop_database_test.cpp
index 0e46980f33c..e14cf74b809 100644
--- a/src/mongo/db/catalog/drop_database_test.cpp
+++ b/src/mongo/db/catalog/drop_database_test.cpp
@@ -36,7 +36,7 @@
 #include "mongo/db/catalog/database_holder.h"
 #include "mongo/db/catalog/drop_database.h"
 #include "mongo/db/client.h"
-#include "mongo/db/concurrency/write_conflict_exception.h"
+#include "mongo/db/concurrency/exception_util.h"
 #include "mongo/db/db_raii.h"
 #include "mongo/db/jsobj.h"
 #include "mongo/db/namespace_string.h"
diff --git a/src/mongo/db/catalog/drop_indexes.cpp b/src/mongo/db/catalog/drop_indexes.cpp
index 2c10eb005c5..329e08f93ef 100644
--- a/src/mongo/db/catalog/drop_indexes.cpp
+++ b/src/mongo/db/catalog/drop_indexes.cpp
@@ -39,7 +39,7 @@
 #include "mongo/db/catalog/collection_uuid_mismatch.h"
 #include "mongo/db/catalog/index_catalog.h"
 #include "mongo/db/client.h"
-#include "mongo/db/concurrency/write_conflict_exception.h"
+#include "mongo/db/concurrency/exception_util.h"
 #include "mongo/db/curop.h"
 #include "mongo/db/db_raii.h"
 #include "mongo/db/index/index_descriptor.h"
@@ -107,9 +107,13 @@ Status checkReplState(OperationContext* opCtx,
 StatusWith<const IndexDescriptor*> getDescriptorByKeyPattern(OperationContext* opCtx,
                                                              const IndexCatalog* indexCatalog,
                                                              const BSONObj& keyPattern) {
-    const bool includeUnfinished = true;
     std::vector<const IndexDescriptor*> indexes;
-    indexCatalog->findIndexesByKeyPattern(opCtx, keyPattern, includeUnfinished, &indexes);
+    indexCatalog->findIndexesByKeyPattern(opCtx,
+                                          keyPattern,
+                                          IndexCatalog::InclusionPolicy::kReady |
+                                              IndexCatalog::InclusionPolicy::kUnfinished |
+                                              IndexCatalog::InclusionPolicy::kFrozen,
+                                          &indexes);
     if (indexes.empty()) {
         return Status(ErrorCodes::IndexNotFound,
                       str::stream() << "can't find index with key: " << keyPattern);
@@ -338,17 +342,20 @@ void dropReadyIndexes(OperationContext* opCtx,
         return;
     }
 
-    bool includeUnfinished = true;
     for (const auto& indexName : indexNames) {
         if (collDescription.isSharded()) {
             uassert(
                 ErrorCodes::CannotDropShardKeyIndex,
                 "Cannot drop the only compatible index for this collection's shard key",
-                !isLastShardKeyIndex(
+                !isLastNonHiddenShardKeyIndex(
                     opCtx, collection, indexCatalog, indexName, collDescription.getKeyPattern()));
         }
 
-        auto desc = indexCatalog->findIndexByName(opCtx, indexName, includeUnfinished);
+        auto desc = indexCatalog->findIndexByName(opCtx,
+                                                  indexName,
+                                                  IndexCatalog::InclusionPolicy::kReady |
+                                                      IndexCatalog::InclusionPolicy::kUnfinished |
+                                                      IndexCatalog::InclusionPolicy::kFrozen);
         if (!desc) {
             uasserted(ErrorCodes::IndexNotFound,
                       str::stream() << "index not found with name [" << indexName << "]");
@@ -507,7 +514,6 @@ DropIndexesReply dropIndexes(OperationContext* opCtx,
             // the index catalog. This would indicate that while we yielded our locks during the
             // abort phase, a new identical index was created.
             auto indexCatalog = collection->getWritableCollection(opCtx)->getIndexCatalog();
-            const bool includeUnfinished = false;
             for (const auto& indexName : indexNames) {
                 auto collDescription =
                     CollectionShardingState::get(opCtx, nss)->getCollectionDescription(opCtx);
@@ -515,14 +521,19 @@ DropIndexesReply dropIndexes(OperationContext* opCtx,
                 if (collDescription.isSharded()) {
                     uassert(ErrorCodes::CannotDropShardKeyIndex,
                             "Cannot drop the only compatible index for this collection's shard key",
-                            !isLastShardKeyIndex(opCtx,
-                                                 collection->getCollection(),
-                                                 indexCatalog,
-                                                 indexName,
-                                                 collDescription.getKeyPattern()));
+                            !isLastNonHiddenShardKeyIndex(opCtx,
+                                                          collection->getCollection(),
+                                                          indexCatalog,
+                                                          indexName,
+                                                          collDescription.getKeyPattern()));
                 }
 
-                auto desc = indexCatalog->findIndexByName(opCtx, indexName, includeUnfinished);
+                auto desc =
+                    indexCatalog->findIndexByName(opCtx,
+                                                  indexName,
+                                                  IndexCatalog::InclusionPolicy::kReady |
+                                                      IndexCatalog::InclusionPolicy::kUnfinished |
+                                                      IndexCatalog::InclusionPolicy::kFrozen);
                 if (!desc) {
                     // A similar index wasn't created while we yielded the locks during abort.
                     continue;
diff --git a/src/mongo/db/catalog/health_log.cpp b/src/mongo/db/catalog/health_log.cpp
index aa49302f169..46f14a5d662 100644
--- a/src/mongo/db/catalog/health_log.cpp
+++ b/src/mongo/db/catalog/health_log.cpp
@@ -31,7 +31,6 @@
 
 #include "mongo/db/catalog/health_log.h"
 #include "mongo/db/catalog/health_log_gen.h"
-#include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/db_raii.h"
 
 namespace mongo {
diff --git a/src/mongo/db/catalog/index_build_block.cpp b/src/mongo/db/catalog/index_build_block.cpp
index 6174e891b64..a5645e9de72 100644
--- a/src/mongo/db/catalog/index_build_block.cpp
+++ b/src/mongo/db/catalog/index_build_block.cpp
@@ -44,7 +44,6 @@
 #include "mongo/db/query/collection_index_usage_tracker_decoration.h"
 #include "mongo/db/query/collection_query_info.h"
 #include "mongo/db/storage/durable_catalog.h"
-#include "mongo/db/storage/storage_parameters_gen.h"
 #include "mongo/db/ttl_collection_cache.h"
 #include "mongo/db/vector_clock.h"
 #include "mongo/logv2/log.h"
@@ -58,14 +57,7 @@ IndexBuildBlock::IndexBuildBlock(const NamespaceString& nss,
                                  const BSONObj& spec,
                                  IndexBuildMethod method,
                                  boost::optional<UUID> indexBuildUUID)
-    : _nss(nss),
-      _spec(spec.getOwned()),
-      _method(method),
-      _buildUUID(indexBuildUUID),
-      _pooledBuilder(
-          gOperationMemoryPoolBlockInitialSizeKB.loadRelaxed() * static_cast<size_t>(1024),
-          SharedBufferFragmentBuilder::DoubleGrowStrategy(
-              gOperationMemoryPoolBlockMaxSizeKB.loadRelaxed() * static_cast<size_t>(1024))) {}
+    : _nss(nss), _spec(spec.getOwned()), _method(method), _buildUUID(indexBuildUUID) {}
 
 void IndexBuildBlock::keepTemporaryTables() {
     if (_indexBuildInterceptor) {
@@ -92,7 +84,9 @@ Status IndexBuildBlock::initForResume(OperationContext* opCtx,
 
     _indexName = _spec.getStringField("name").toString();
     auto descriptor = collection->getIndexCatalog()->findIndexByName(
-        opCtx, _indexName, true /* includeUnfinishedIndexes */);
+        opCtx,
+        _indexName,
+        IndexCatalog::InclusionPolicy::kReady | IndexCatalog::InclusionPolicy::kUnfinished);
 
     auto indexCatalogEntry = descriptor->getEntry();
 
@@ -274,7 +268,10 @@ void IndexBuildBlock::success(OperationContext* opCtx, Collection* collection) {
             // Note that TTL deletion is supported on capped clustered collections via bounded
             // collection scan, which does not use an index.
             if (spec.hasField(IndexDescriptor::kExpireAfterSecondsFieldName) && !coll->isCapped()) {
-                TTLCollectionCache::get(svcCtx).registerTTLInfo(coll->uuid(), indexName);
+                TTLCollectionCache::get(svcCtx).registerTTLInfo(
+                    coll->uuid(),
+                    TTLCollectionCache::Info{
+                        indexName, spec[IndexDescriptor::kExpireAfterSecondsFieldName].isNaN()});
             }
         });
 }
@@ -282,14 +279,18 @@ void IndexBuildBlock::success(OperationContext* opCtx, Collection* collection) {
 const IndexCatalogEntry* IndexBuildBlock::getEntry(OperationContext* opCtx,
                                                    const CollectionPtr& collection) const {
     auto descriptor = collection->getIndexCatalog()->findIndexByName(
-        opCtx, _indexName, true /* includeUnfinishedIndexes */);
+        opCtx,
+        _indexName,
+        IndexCatalog::InclusionPolicy::kReady | IndexCatalog::InclusionPolicy::kUnfinished);
 
     return descriptor->getEntry();
 }
 
 IndexCatalogEntry* IndexBuildBlock::getEntry(OperationContext* opCtx, Collection* collection) {
     auto descriptor = collection->getIndexCatalog()->findIndexByName(
-        opCtx, _indexName, true /* includeUnfinishedIndexes */);
+        opCtx,
+        _indexName,
+        IndexCatalog::InclusionPolicy::kReady | IndexCatalog::InclusionPolicy::kUnfinished);
 
     return descriptor->getEntry();
 }
diff --git a/src/mongo/db/catalog/index_build_block.h b/src/mongo/db/catalog/index_build_block.h
index 48d0f6e49af..b1086aac5ed 100644
--- a/src/mongo/db/catalog/index_build_block.h
+++ b/src/mongo/db/catalog/index_build_block.h
@@ -111,13 +111,6 @@ public:
         return _spec;
     }
 
-    /**
-     * Returns a memory pool for creating temporary objects for this index build.
-     */
-    SharedBufferFragmentBuilder& getPooledBuilder() {
-        return _pooledBuilder;
-    }
-
 private:
     void _completeInit(OperationContext* opCtx, Collection* collection);
 
@@ -131,7 +124,5 @@ private:
     std::string _indexNamespace;
 
     std::unique_ptr<IndexBuildInterceptor> _indexBuildInterceptor;
-
-    SharedBufferFragmentBuilder _pooledBuilder;
 };
 }  // namespace mongo
diff --git a/src/mongo/db/catalog/index_builds_manager.cpp b/src/mongo/db/catalog/index_builds_manager.cpp
index 82483f9cf21..a5db8f4536f 100644
--- a/src/mongo/db/catalog/index_builds_manager.cpp
+++ b/src/mongo/db/catalog/index_builds_manager.cpp
@@ -38,7 +38,7 @@
 #include "mongo/db/catalog/index_catalog.h"
 #include "mongo/db/catalog/index_repair.h"
 #include "mongo/db/catalog_raii.h"
-#include "mongo/db/concurrency/write_conflict_exception.h"
+#include "mongo/db/concurrency/exception_util.h"
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/operation_context.h"
 #include "mongo/db/storage/storage_repair_observer.h"
diff --git a/src/mongo/db/catalog/index_catalog.h b/src/mongo/db/catalog/index_catalog.h
index 2754ac3e22f..58e48e08be5 100644
--- a/src/mongo/db/catalog/index_catalog.h
+++ b/src/mongo/db/catalog/index_catalog.h
@@ -197,6 +197,12 @@ public:
         std::unique_ptr<std::vector<IndexCatalogEntry*>> _ownedContainer;
     };
 
+    enum class InclusionPolicy {
+        kReady = 1 << 0,
+        kUnfinished = 1 << 1,
+        kFrozen = 1 << 2,
+    };
+
     IndexCatalog() = default;
     virtual ~IndexCatalog() = default;
 
@@ -237,9 +243,10 @@ public:
      *
      * @return null if cannot find
      */
-    virtual const IndexDescriptor* findIndexByName(OperationContext* opCtx,
-                                                   StringData name,
-                                                   bool includeUnfinishedIndexes = false) const = 0;
+    virtual const IndexDescriptor* findIndexByName(
+        OperationContext* opCtx,
+        StringData name,
+        InclusionPolicy inclusionPolicy = InclusionPolicy::kReady) const = 0;
 
     /**
      * Find index by matching key pattern and options. The key pattern, collation spec, and partial
@@ -251,7 +258,7 @@ public:
         OperationContext* opCtx,
         const BSONObj& key,
         const BSONObj& indexSpec,
-        bool includeUnfinishedIndexes = false) const = 0;
+        InclusionPolicy inclusionPolicy = InclusionPolicy::kReady) const = 0;
 
     /**
      * Find indexes with a matching key pattern, putting them into the vector 'matches'.  The key
@@ -261,12 +268,13 @@ public:
      */
     virtual void findIndexesByKeyPattern(OperationContext* opCtx,
                                          const BSONObj& key,
-                                         bool includeUnfinishedIndexes,
+                                         InclusionPolicy inclusionPolicy,
                                          std::vector<const IndexDescriptor*>* matches) const = 0;
-    virtual void findIndexByType(OperationContext* opCtx,
-                                 const std::string& type,
-                                 std::vector<const IndexDescriptor*>& matches,
-                                 bool includeUnfinishedIndexes = false) const = 0;
+    virtual void findIndexByType(
+        OperationContext* opCtx,
+        const std::string& type,
+        std::vector<const IndexDescriptor*>& matches,
+        InclusionPolicy inclusionPolicy = InclusionPolicy::kReady) const = 0;
 
     /**
      * Reload the index definition for 'oldDesc' from the CollectionCatalogEntry.  'oldDesc'
@@ -308,7 +316,7 @@ public:
      * Returns an iterator for the index descriptors in this IndexCatalog.
      */
     virtual std::unique_ptr<IndexIterator> getIndexIterator(
-        OperationContext* opCtx, bool includeUnfinishedIndexes) const = 0;
+        OperationContext* opCtx, InclusionPolicy inclusionPolicy) const = 0;
 
     // ---- index set modifiers ------
 
@@ -521,4 +529,16 @@ public:
                                    Collection* coll,
                                    IndexCatalogEntry* index) = 0;
 };
+
+inline IndexCatalog::InclusionPolicy operator|(IndexCatalog::InclusionPolicy lhs,
+                                               IndexCatalog::InclusionPolicy rhs) {
+    return static_cast<IndexCatalog::InclusionPolicy>(
+        static_cast<std::underlying_type_t<IndexCatalog::InclusionPolicy>>(lhs) |
+        static_cast<std::underlying_type_t<IndexCatalog::InclusionPolicy>>(rhs));
+}
+
+inline bool operator&(IndexCatalog::InclusionPolicy lhs, IndexCatalog::InclusionPolicy rhs) {
+    return static_cast<std::underlying_type_t<IndexCatalog::InclusionPolicy>>(lhs) &
+        static_cast<std::underlying_type_t<IndexCatalog::InclusionPolicy>>(rhs);
+}
 }  // namespace mongo
diff --git a/src/mongo/db/catalog/index_catalog_entry.h b/src/mongo/db/catalog/index_catalog_entry.h
index fa77370bb96..9761f590c9d 100644
--- a/src/mongo/db/catalog/index_catalog_entry.h
+++ b/src/mongo/db/catalog/index_catalog_entry.h
@@ -64,8 +64,6 @@ public:
     inline IndexCatalogEntry(IndexCatalogEntry&&) = delete;
     inline IndexCatalogEntry& operator=(IndexCatalogEntry&&) = delete;
 
-    virtual void init(std::unique_ptr<IndexAccessMethod> accessMethod) = 0;
-
     virtual const std::string& getIdent() const = 0;
     virtual std::shared_ptr<Ident> getSharedIdent() const = 0;
 
@@ -75,6 +73,8 @@ public:
 
     virtual IndexAccessMethod* accessMethod() const = 0;
 
+    virtual void setAccessMethod(std::unique_ptr<IndexAccessMethod> accessMethod) = 0;
+
     virtual bool isHybridBuilding() const = 0;
 
     virtual IndexBuildInterceptor* indexBuildInterceptor() const = 0;
diff --git a/src/mongo/db/catalog/index_catalog_entry_impl.cpp b/src/mongo/db/catalog/index_catalog_entry_impl.cpp
index daf74714c84..498165df637 100644
--- a/src/mongo/db/catalog/index_catalog_entry_impl.cpp
+++ b/src/mongo/db/catalog/index_catalog_entry_impl.cpp
@@ -39,7 +39,7 @@
 #include "mongo/base/init.h"
 #include "mongo/db/catalog/collection.h"
 #include "mongo/db/concurrency/d_concurrency.h"
-#include "mongo/db/concurrency/write_conflict_exception.h"
+#include "mongo/db/concurrency/exception_util.h"
 #include "mongo/db/index/index_access_method.h"
 #include "mongo/db/index/index_descriptor.h"
 #include "mongo/db/matcher/expression.h"
@@ -122,7 +122,7 @@ IndexCatalogEntryImpl::IndexCatalogEntryImpl(OperationContext* const opCtx,
     }
 }
 
-void IndexCatalogEntryImpl::init(std::unique_ptr<IndexAccessMethod> accessMethod) {
+void IndexCatalogEntryImpl::setAccessMethod(std::unique_ptr<IndexAccessMethod> accessMethod) {
     invariant(!_accessMethod);
     _accessMethod = std::move(accessMethod);
 }
@@ -370,7 +370,8 @@ Status IndexCatalogEntryImpl::_setMultikeyInMultiDocumentTransaction(
 }
 
 std::shared_ptr<Ident> IndexCatalogEntryImpl::getSharedIdent() const {
-    return {shared_from_this(), _accessMethod->getIdentPtr()};  // aliasing constructor
+    return _accessMethod ? std::shared_ptr<Ident>{shared_from_this(), _accessMethod->getIdentPtr()}
+                         : nullptr;
 }
 
 // ----
diff --git a/src/mongo/db/catalog/index_catalog_entry_impl.h b/src/mongo/db/catalog/index_catalog_entry_impl.h
index 821176164e1..30ef5a80921 100644
--- a/src/mongo/db/catalog/index_catalog_entry_impl.h
+++ b/src/mongo/db/catalog/index_catalog_entry_impl.h
@@ -61,8 +61,6 @@ public:
                           std::unique_ptr<IndexDescriptor> descriptor,  // ownership passes to me
                           bool isFrozen);
 
-    void init(std::unique_ptr<IndexAccessMethod> accessMethod) final;
-
     const std::string& getIdent() const final {
         return _ident;
     }
@@ -80,6 +78,8 @@ public:
         return _accessMethod.get();
     }
 
+    void setAccessMethod(std::unique_ptr<IndexAccessMethod> accessMethod) final;
+
     bool isHybridBuilding() const final {
         return _indexBuildInterceptor != nullptr;
     }
diff --git a/src/mongo/db/catalog/index_catalog_impl.cpp b/src/mongo/db/catalog/index_catalog_impl.cpp
index 6733dac0ba1..4a4d0b64826 100644
--- a/src/mongo/db/catalog/index_catalog_impl.cpp
+++ b/src/mongo/db/catalog/index_catalog_impl.cpp
@@ -47,7 +47,6 @@
 #include "mongo/db/catalog/uncommitted_catalog_updates.h"
 #include "mongo/db/client.h"
 #include "mongo/db/clientcursor.h"
-#include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/curop.h"
 #include "mongo/db/field_ref.h"
 #include "mongo/db/fts/fts_spec.h"
@@ -205,13 +204,30 @@ Status IndexCatalogImpl::init(OperationContext* opCtx, Collection* collection) {
 
         auto descriptor = std::make_unique<IndexDescriptor>(_getAccessMethodName(keyPattern), spec);
 
+        // TTL indexes with NaN 'expireAfterSeconds' cause problems in multiversion settings.
+        if (spec.hasField(IndexDescriptor::kExpireAfterSecondsFieldName)) {
+            if (spec[IndexDescriptor::kExpireAfterSecondsFieldName].isNaN()) {
+                LOGV2_OPTIONS(6852200,
+                              {logv2::LogTag::kStartupWarnings},
+                              "Found an existing TTL index with NaN 'expireAfterSeconds' in the "
+                              "catalog.",
+                              "ns"_attr = collection->ns(),
+                              "uuid"_attr = collection->uuid(),
+                              "index"_attr = indexName,
+                              "spec"_attr = spec);
+            }
+        }
+
         // TTL indexes are not compatible with capped collections.
         // Note that TTL deletion is supported on capped clustered collections via bounded
         // collection scan, which does not use an index.
         if (spec.hasField(IndexDescriptor::kExpireAfterSecondsFieldName) &&
             !collection->isCapped()) {
             TTLCollectionCache::get(opCtx->getServiceContext())
-                .registerTTLInfo(collection->uuid(), indexName);
+                .registerTTLInfo(
+                    collection->uuid(),
+                    TTLCollectionCache::Info{
+                        indexName, spec[IndexDescriptor::kExpireAfterSecondsFieldName].isNaN()});
         }
 
         bool ready = collection->isIndexReady(indexName);
@@ -257,8 +273,8 @@ Status IndexCatalogImpl::init(OperationContext* opCtx, Collection* collection) {
 }
 
 std::unique_ptr<IndexCatalog::IndexIterator> IndexCatalogImpl::getIndexIterator(
-    OperationContext* const opCtx, const bool includeUnfinishedIndexes) const {
-    if (!includeUnfinishedIndexes) {
+    OperationContext* const opCtx, InclusionPolicy inclusionPolicy) const {
+    if (inclusionPolicy == InclusionPolicy::kReady) {
         // If the caller only wants the ready indexes, we return an iterator over the catalog's
         // ready indexes vector. When the user advances this iterator, it will filter out any
         // indexes that were not ready at the OperationContext's read timestamp.
@@ -266,17 +282,28 @@ std::unique_ptr<IndexCatalog::IndexIterator> IndexCatalogImpl::getIndexIterator(
             opCtx, _readyIndexes.begin(), _readyIndexes.end());
     }
 
-    // If the caller wants all indexes, for simplicity of implementation, we copy the pointers to
-    // a new vector. The vector's ownership is passed to the iterator. The query code path from an
-    // external client is not expected to hit this case so the cost isn't paid by the important
-    // code path.
+    // If the caller doesn't only want the ready indexes, for simplicity of implementation, we copy
+    // the pointers to a new vector. The vector's ownership is passed to the iterator. The query
+    // code path from an external client is not expected to hit this case so the cost isn't paid by
+    // the important code path.
     auto allIndexes = std::make_unique<std::vector<IndexCatalogEntry*>>();
-    for (auto it = _readyIndexes.begin(); it != _readyIndexes.end(); ++it) {
-        allIndexes->push_back(it->get());
+
+    if (inclusionPolicy & InclusionPolicy::kReady) {
+        for (auto it = _readyIndexes.begin(); it != _readyIndexes.end(); ++it) {
+            allIndexes->push_back(it->get());
+        }
+    }
+
+    if (inclusionPolicy & InclusionPolicy::kUnfinished) {
+        for (auto it = _buildingIndexes.begin(); it != _buildingIndexes.end(); ++it) {
+            allIndexes->push_back(it->get());
+        }
     }
 
-    for (auto it = _buildingIndexes.begin(); it != _buildingIndexes.end(); ++it) {
-        allIndexes->push_back(it->get());
+    if (inclusionPolicy & InclusionPolicy::kFrozen) {
+        for (auto it = _frozenIndexes.begin(); it != _frozenIndexes.end(); ++it) {
+            allIndexes->push_back(it->get());
+        }
     }
 
     return std::make_unique<AllIndexesIterator>(opCtx, std::move(allIndexes));
@@ -350,6 +377,7 @@ void IndexCatalogImpl::_logInternalState(OperationContext* opCtx,
                 "numIndexesInCollectionCatalogEntry"_attr = numIndexesInCollectionCatalogEntry,
                 "numReadyIndexes"_attr = _readyIndexes.size(),
                 "numBuildingIndexes"_attr = _buildingIndexes.size(),
+                "numFrozenIndexes"_attr = _frozenIndexes.size(),
                 "indexNamesToDrop"_attr = indexNamesToDrop);
 
     // Report the ready indexes.
@@ -425,7 +453,8 @@ StatusWith<BSONObj> IndexCatalogImpl::prepareSpecForCreate(
     }
 
     // First check against only the ready indexes for conflicts.
-    status = _doesSpecConflictWithExisting(opCtx, collection, validatedSpec, false);
+    status =
+        _doesSpecConflictWithExisting(opCtx, collection, validatedSpec, InclusionPolicy::kReady);
     if (!status.isOK()) {
         return status;
     }
@@ -441,7 +470,12 @@ StatusWith<BSONObj> IndexCatalogImpl::prepareSpecForCreate(
     // The index catalog cannot currently iterate over only in-progress indexes. So by previously
     // checking against only ready indexes without error, we know that any errors encountered
     // checking against all indexes occurred due to an in-progress index.
-    status = _doesSpecConflictWithExisting(opCtx, collection, validatedSpec, true);
+    status = _doesSpecConflictWithExisting(opCtx,
+                                           collection,
+                                           validatedSpec,
+                                           IndexCatalog::InclusionPolicy::kReady |
+                                               IndexCatalog::InclusionPolicy::kUnfinished |
+                                               IndexCatalog::InclusionPolicy::kFrozen);
     if (!status.isOK()) {
         if (ErrorCodes::IndexAlreadyExists == status.code()) {
             // Callers need to be able to distinguish conflicts against ready indexes versus
@@ -470,8 +504,11 @@ std::vector<BSONObj> IndexCatalogImpl::removeExistingIndexesNoChecks(
         // _doesSpecConflictWithExisting currently does more work than we require here: we are only
         // interested in the index already exists error.
         if (ErrorCodes::IndexAlreadyExists ==
-            _doesSpecConflictWithExisting(
-                opCtx, collection, spec, true /*includeUnfinishedIndexes*/)) {
+            _doesSpecConflictWithExisting(opCtx,
+                                          collection,
+                                          spec,
+                                          IndexCatalog::InclusionPolicy::kReady |
+                                              IndexCatalog::InclusionPolicy::kUnfinished)) {
             continue;
         }
 
@@ -534,19 +571,20 @@ IndexCatalogEntry* IndexCatalogImpl::createIndexEntry(OperationContext* opCtx,
         engine->getEngine()->alterIdentMetadata(opCtx, ident, desc, isForceUpdateMetadata);
     }
 
-    const auto& collOptions = collection->getCollectionOptions();
-    std::unique_ptr<SortedDataInterface> sdi = engine->getEngine()->getSortedDataInterface(
-        opCtx, collection->ns(), collOptions, ident, desc);
-
-    std::unique_ptr<IndexAccessMethod> accessMethod =
-        IndexAccessMethodFactory::get(opCtx)->make(entry.get(), std::move(sdi));
-
-    entry->init(std::move(accessMethod));
-
+    if (!frozen) {
+        const auto& collOptions = collection->getCollectionOptions();
+        std::unique_ptr<SortedDataInterface> sdi = engine->getEngine()->getSortedDataInterface(
+            opCtx, collection->ns(), collOptions, ident, desc);
+        std::unique_ptr<IndexAccessMethod> accessMethod =
+            IndexAccessMethod::make(entry.get(), std::move(sdi));
+        entry->setAccessMethod(std::move(accessMethod));
+    }
 
     IndexCatalogEntry* save = entry.get();
     if (isReadyIndex) {
         _readyIndexes.add(std::move(entry));
+    } else if (frozen) {
+        _frozenIndexes.add(std::move(entry));
     } else {
         _buildingIndexes.add(std::move(entry));
     }
@@ -873,10 +911,11 @@ Status IndexCatalogImpl::_isSpecOk(OperationContext* opCtx,
         }
         const std::unique_ptr<MatchExpression> filterExpr = std::move(statusWithMatcher.getValue());
 
-        Status status =
-            _checkValidFilterExpressions(filterExpr.get(),
-                                         feature_flags::gTimeseriesMetricIndexes.isEnabled(
-                                             serverGlobalParams.featureCompatibility));
+        Status status = _checkValidFilterExpressions(
+            filterExpr.get(),
+            !serverGlobalParams.featureCompatibility.isVersionInitialized() ||
+                feature_flags::gTimeseriesMetricIndexes.isEnabled(
+                    serverGlobalParams.featureCompatibility));
         if (!status.isOK()) {
             return status;
         }
@@ -951,7 +990,7 @@ Status IndexCatalogImpl::_isSpecOk(OperationContext* opCtx,
 Status IndexCatalogImpl::_doesSpecConflictWithExisting(OperationContext* opCtx,
                                                        const CollectionPtr& collection,
                                                        const BSONObj& spec,
-                                                       const bool includeUnfinishedIndexes) const {
+                                                       InclusionPolicy inclusionPolicy) const {
     StringData name = spec.getStringField(IndexDescriptor::kIndexNameFieldName);
     invariant(name[0]);
 
@@ -965,7 +1004,7 @@ Status IndexCatalogImpl::_doesSpecConflictWithExisting(OperationContext* opCtx,
 
     {
         // Check whether an index with the specified candidate name already exists in the catalog.
-        const IndexDescriptor* desc = findIndexByName(opCtx, name, includeUnfinishedIndexes);
+        const IndexDescriptor* desc = findIndexByName(opCtx, name, inclusionPolicy);
 
         if (desc) {
             // Index already exists with same name. Check whether the options are the same as well.
@@ -1018,7 +1057,7 @@ Status IndexCatalogImpl::_doesSpecConflictWithExisting(OperationContext* opCtx,
     {
         // No index with the candidate name exists. Check for an index with conflicting options.
         const IndexDescriptor* desc =
-            findIndexByKeyPatternAndOptions(opCtx, key, spec, includeUnfinishedIndexes);
+            findIndexByKeyPatternAndOptions(opCtx, key, spec, inclusionPolicy);
 
         if (desc) {
             LOGV2_DEBUG(20353,
@@ -1069,7 +1108,7 @@ Status IndexCatalogImpl::_doesSpecConflictWithExisting(OperationContext* opCtx,
     string pluginName = IndexNames::findPluginName(key);
     if (pluginName == IndexNames::TEXT) {
         vector<const IndexDescriptor*> textIndexes;
-        findIndexByType(opCtx, IndexNames::TEXT, textIndexes, includeUnfinishedIndexes);
+        findIndexByType(opCtx, IndexNames::TEXT, textIndexes, inclusionPolicy);
         if (textIndexes.size() > 0) {
             return Status(ErrorCodes::CannotCreateIndex,
                           str::stream() << "only one text index per collection allowed, "
@@ -1110,7 +1149,10 @@ void IndexCatalogImpl::dropIndexes(OperationContext* opCtx,
     vector<string> indexNamesToDrop;
     {
         int seen = 0;
-        std::unique_ptr<IndexIterator> ii = getIndexIterator(opCtx, true);
+        auto ii = getIndexIterator(opCtx,
+                                   IndexCatalog::InclusionPolicy::kReady |
+                                       IndexCatalog::InclusionPolicy::kUnfinished |
+                                       IndexCatalog::InclusionPolicy::kFrozen);
         while (ii->more()) {
             seen++;
             const IndexDescriptor* desc = ii->next()->descriptor();
@@ -1125,7 +1167,11 @@ void IndexCatalogImpl::dropIndexes(OperationContext* opCtx,
 
     for (size_t i = 0; i < indexNamesToDrop.size(); i++) {
         string indexName = indexNamesToDrop[i];
-        const IndexDescriptor* desc = findIndexByName(opCtx, indexName, true);
+        const IndexDescriptor* desc = findIndexByName(
+            opCtx,
+            indexName,
+            IndexCatalog::InclusionPolicy::kReady | IndexCatalog::InclusionPolicy::kUnfinished |
+                IndexCatalog::InclusionPolicy::kFrozen);
         invariant(desc);
         LOGV2_DEBUG(20355, 1, "\t dropAllIndexes dropping: {desc}", "desc"_attr = *desc);
         IndexCatalogEntry* entry = desc->getEntry();
@@ -1251,25 +1297,26 @@ Status IndexCatalogImpl::dropIndexEntry(OperationContext* opCtx,
 
     audit::logDropIndex(opCtx->getClient(), indexName, collection->ns());
 
-    auto released = _readyIndexes.release(entry->descriptor());
-    if (released) {
-        invariant(released.get() == entry);
-        opCtx->recoveryUnit()->registerChange(
-            std::make_unique<IndexRemoveChange>(opCtx,
-                                                collection->ns(),
-                                                collection->uuid(),
-                                                std::move(released),
-                                                collection->getSharedDecorations()));
-    } else {
-        released = _buildingIndexes.release(entry->descriptor());
-        invariant(released.get() == entry);
-        opCtx->recoveryUnit()->registerChange(
-            std::make_unique<IndexRemoveChange>(opCtx,
-                                                collection->ns(),
-                                                collection->uuid(),
-                                                std::move(released),
-                                                collection->getSharedDecorations()));
-    }
+    auto released = [&] {
+        if (auto released = _readyIndexes.release(entry->descriptor())) {
+            return released;
+        }
+        if (auto released = _buildingIndexes.release(entry->descriptor())) {
+            return released;
+        }
+        if (auto released = _frozenIndexes.release(entry->descriptor())) {
+            return released;
+        }
+        MONGO_UNREACHABLE;
+    }();
+
+    invariant(released.get() == entry);
+    opCtx->recoveryUnit()->registerChange(
+        std::make_unique<IndexRemoveChange>(opCtx,
+                                            collection->ns(),
+                                            collection->uuid(),
+                                            std::move(released),
+                                            collection->getSharedDecorations()));
 
     CollectionQueryInfo::get(collection).rebuildIndexData(opCtx, collection);
     CollectionIndexUsageTrackerDecoration::get(collection->getSharedDecorations())
@@ -1289,7 +1336,11 @@ void IndexCatalogImpl::_deleteIndexFromDisk(OperationContext* opCtx,
                                             Collection* collection,
                                             const string& indexName,
                                             std::shared_ptr<Ident> ident) {
-    invariant(!findIndexByName(opCtx, indexName, true /* includeUnfinishedIndexes*/));
+    invariant(!findIndexByName(opCtx,
+                               indexName,
+                               IndexCatalog::InclusionPolicy::kReady |
+                                   IndexCatalog::InclusionPolicy::kUnfinished |
+                                   IndexCatalog::InclusionPolicy::kFrozen));
     catalog::removeIndex(opCtx, indexName, collection, std::move(ident));
 }
 
@@ -1319,7 +1370,7 @@ int IndexCatalogImpl::numIndexesTotal(OperationContext* opCtx) const {
 
 int IndexCatalogImpl::numIndexesReady(OperationContext* opCtx) const {
     std::vector<const IndexDescriptor*> itIndexes;
-    std::unique_ptr<IndexIterator> ii = getIndexIterator(opCtx, /*includeUnfinished*/ false);
+    auto ii = getIndexIterator(opCtx, InclusionPolicy::kReady);
     while (ii->more()) {
         itIndexes.push_back(ii->next()->descriptor());
     }
@@ -1331,7 +1382,7 @@ bool IndexCatalogImpl::haveIdIndex(OperationContext* opCtx) const {
 }
 
 const IndexDescriptor* IndexCatalogImpl::findIdIndex(OperationContext* opCtx) const {
-    std::unique_ptr<IndexIterator> ii = getIndexIterator(opCtx, false);
+    auto ii = getIndexIterator(opCtx, InclusionPolicy::kReady);
     while (ii->more()) {
         const IndexDescriptor* desc = ii->next()->descriptor();
         if (desc->isIdIndex())
@@ -1342,8 +1393,8 @@ const IndexDescriptor* IndexCatalogImpl::findIdIndex(OperationContext* opCtx) co
 
 const IndexDescriptor* IndexCatalogImpl::findIndexByName(OperationContext* opCtx,
                                                          StringData name,
-                                                         bool includeUnfinishedIndexes) const {
-    std::unique_ptr<IndexIterator> ii = getIndexIterator(opCtx, includeUnfinishedIndexes);
+                                                         InclusionPolicy inclusionPolicy) const {
+    auto ii = getIndexIterator(opCtx, inclusionPolicy);
     while (ii->more()) {
         const IndexDescriptor* desc = ii->next()->descriptor();
         if (desc->indexName() == name)
@@ -1356,8 +1407,8 @@ const IndexDescriptor* IndexCatalogImpl::findIndexByKeyPatternAndOptions(
     OperationContext* opCtx,
     const BSONObj& key,
     const BSONObj& indexSpec,
-    bool includeUnfinishedIndexes) const {
-    std::unique_ptr<IndexIterator> ii = getIndexIterator(opCtx, includeUnfinishedIndexes);
+    InclusionPolicy inclusionPolicy) const {
+    auto ii = getIndexIterator(opCtx, inclusionPolicy);
     IndexDescriptor needle(_getAccessMethodName(key), indexSpec);
     while (ii->more()) {
         const auto* entry = ii->next();
@@ -1371,10 +1422,10 @@ const IndexDescriptor* IndexCatalogImpl::findIndexByKeyPatternAndOptions(
 
 void IndexCatalogImpl::findIndexesByKeyPattern(OperationContext* opCtx,
                                                const BSONObj& key,
-                                               bool includeUnfinishedIndexes,
+                                               InclusionPolicy inclusionPolicy,
                                                std::vector<const IndexDescriptor*>* matches) const {
     invariant(matches);
-    std::unique_ptr<IndexIterator> ii = getIndexIterator(opCtx, includeUnfinishedIndexes);
+    auto ii = getIndexIterator(opCtx, inclusionPolicy);
     while (ii->more()) {
         const IndexDescriptor* desc = ii->next()->descriptor();
         if (SimpleBSONObjComparator::kInstance.evaluate(desc->keyPattern() == key)) {
@@ -1386,8 +1437,8 @@ void IndexCatalogImpl::findIndexesByKeyPattern(OperationContext* opCtx,
 void IndexCatalogImpl::findIndexByType(OperationContext* opCtx,
                                        const string& type,
                                        vector<const IndexDescriptor*>& matches,
-                                       bool includeUnfinishedIndexes) const {
-    std::unique_ptr<IndexIterator> ii = getIndexIterator(opCtx, includeUnfinishedIndexes);
+                                       InclusionPolicy inclusionPolicy) const {
+    auto ii = getIndexIterator(opCtx, inclusionPolicy);
     while (ii->more()) {
         const IndexDescriptor* desc = ii->next()->descriptor();
         if (IndexNames::findPluginName(desc->keyPattern()) == type) {
@@ -1628,7 +1679,10 @@ Status IndexCatalogImpl::indexRecords(OperationContext* opCtx,
 
     for (const MultikeyPathInfo& newPath : newPaths) {
         invariant(newPath.nss == coll->ns());
-        auto idx = findIndexByName(opCtx, newPath.indexName, /*includeUnfinishedIndexes=*/true);
+        auto idx = findIndexByName(opCtx,
+                                   newPath.indexName,
+                                   IndexCatalog::InclusionPolicy::kReady |
+                                       IndexCatalog::InclusionPolicy::kUnfinished);
         if (!idx) {
             return Status(ErrorCodes::IndexNotFound,
                           str::stream()
@@ -1731,7 +1785,10 @@ Status IndexCatalogImpl::compactIndexes(OperationContext* opCtx) const {
 }
 
 std::string::size_type IndexCatalogImpl::getLongestIndexNameLength(OperationContext* opCtx) const {
-    std::unique_ptr<IndexIterator> it = getIndexIterator(opCtx, true);
+    auto it = getIndexIterator(opCtx,
+                               IndexCatalog::InclusionPolicy::kReady |
+                                   IndexCatalog::InclusionPolicy::kUnfinished |
+                                   IndexCatalog::InclusionPolicy::kFrozen);
     std::string::size_type longestIndexNameLength = 0;
     while (it->more()) {
         auto thisLength = it->next()->descriptor()->indexName().length();
@@ -1779,7 +1836,11 @@ void IndexCatalogImpl::indexBuildSuccess(OperationContext* opCtx,
     invariant(releasedEntry.get() == index);
     _readyIndexes.add(std::move(releasedEntry));
 
-    index->setIndexBuildInterceptor(nullptr);
+    // Wait to unset the interceptor until the index actually commits. If a write conflict is
+    // encountered and the index commit process is restated, the multikey information from the
+    // interceptor may still be needed.
+    opCtx->recoveryUnit()->onCommit(
+        [index](boost::optional<Timestamp>) { index->setIndexBuildInterceptor(nullptr); });
     index->setIsReady(true);
 }
 
diff --git a/src/mongo/db/catalog/index_catalog_impl.h b/src/mongo/db/catalog/index_catalog_impl.h
index 259732da61c..f9baf272f11 100644
--- a/src/mongo/db/catalog/index_catalog_impl.h
+++ b/src/mongo/db/catalog/index_catalog_impl.h
@@ -96,9 +96,10 @@ public:
      *
      * @return null if cannot find
      */
-    const IndexDescriptor* findIndexByName(OperationContext* opCtx,
-                                           StringData name,
-                                           bool includeUnfinishedIndexes = false) const override;
+    const IndexDescriptor* findIndexByName(
+        OperationContext* opCtx,
+        StringData name,
+        InclusionPolicy inclusionPolicy = InclusionPolicy::kReady) const override;
 
     /**
      * Find index by matching key pattern and options. The key pattern, collation spec, and partial
@@ -110,7 +111,7 @@ public:
         OperationContext* opCtx,
         const BSONObj& key,
         const BSONObj& indexSpec,
-        bool includeUnfinishedIndexes = false) const override;
+        InclusionPolicy inclusionPolicy = InclusionPolicy::kReady) const override;
 
     /**
      * Find indexes with a matching key pattern, putting them into the vector 'matches'.  The key
@@ -120,12 +121,12 @@ public:
      */
     void findIndexesByKeyPattern(OperationContext* opCtx,
                                  const BSONObj& key,
-                                 bool includeUnfinishedIndexes,
+                                 InclusionPolicy inclusionPolicy,
                                  std::vector<const IndexDescriptor*>* matches) const override;
     void findIndexByType(OperationContext* opCtx,
                          const std::string& type,
                          std::vector<const IndexDescriptor*>& matches,
-                         bool includeUnfinishedIndexes = false) const override;
+                         InclusionPolicy inclusionPolicy = InclusionPolicy::kReady) const override;
 
 
     /**
@@ -153,7 +154,7 @@ public:
 
     using IndexIterator = IndexCatalog::IndexIterator;
     std::unique_ptr<IndexIterator> getIndexIterator(OperationContext* opCtx,
-                                                    bool includeUnfinishedIndexes) const override;
+                                                    InclusionPolicy inclusionPolicy) const override;
 
     // ---- index set modifiers ------
 
@@ -377,7 +378,7 @@ private:
     Status _doesSpecConflictWithExisting(OperationContext* opCtx,
                                          const CollectionPtr& collection,
                                          const BSONObj& spec,
-                                         bool includeUnfinishedIndexes) const;
+                                         InclusionPolicy inclusionPolicy) const;
 
     /**
      * Returns true if the replica set member's config has {buildIndexes:false} set, which means
@@ -393,5 +394,6 @@ private:
 
     IndexCatalogEntryContainer _readyIndexes;
     IndexCatalogEntryContainer _buildingIndexes;
+    IndexCatalogEntryContainer _frozenIndexes;
 };
 }  // namespace mongo
diff --git a/src/mongo/db/catalog/index_consistency.cpp b/src/mongo/db/catalog/index_consistency.cpp
index 55e35b744f8..17f678375b6 100644
--- a/src/mongo/db/catalog/index_consistency.cpp
+++ b/src/mongo/db/catalog/index_consistency.cpp
@@ -38,7 +38,7 @@
 #include "mongo/db/catalog/index_catalog.h"
 #include "mongo/db/catalog/index_repair.h"
 #include "mongo/db/catalog/validate_gen.h"
-#include "mongo/db/concurrency/write_conflict_exception.h"
+#include "mongo/db/concurrency/exception_util.h"
 #include "mongo/db/db_raii.h"
 #include "mongo/db/index/index_access_method.h"
 #include "mongo/db/index/index_descriptor.h"
@@ -106,7 +106,7 @@ IndexConsistency::IndexConsistency(OperationContext* opCtx,
 
 void IndexConsistency::addMultikeyMetadataPath(const KeyString::Value& ks, IndexInfo* indexInfo) {
     auto hash = _hashKeyString(ks, indexInfo->indexNameHash);
-    if (MONGO_unlikely(_validateState->extraLoggingForTest())) {
+    if (MONGO_unlikely(_validateState->logDiagnostics())) {
         LOGV2(6208500,
               "[validate](multikeyMetadataPath) Adding with the hash",
               "hash"_attr = hash,
@@ -118,7 +118,7 @@ void IndexConsistency::addMultikeyMetadataPath(const KeyString::Value& ks, Index
 void IndexConsistency::removeMultikeyMetadataPath(const KeyString::Value& ks,
                                                   IndexInfo* indexInfo) {
     auto hash = _hashKeyString(ks, indexInfo->indexNameHash);
-    if (MONGO_unlikely(_validateState->extraLoggingForTest())) {
+    if (MONGO_unlikely(_validateState->logDiagnostics())) {
         LOGV2(6208501,
               "[validate](multikeyMetadataPath) Removing with the hash",
               "hash"_attr = hash,
@@ -132,9 +132,26 @@ size_t IndexConsistency::getMultikeyMetadataPathCount(IndexInfo* indexInfo) {
 }
 
 bool IndexConsistency::haveEntryMismatch() const {
-    return std::any_of(_indexKeyBuckets.begin(),
-                       _indexKeyBuckets.end(),
-                       [](const IndexKeyBucket& bucket) -> bool { return bucket.indexKeyCount; });
+    bool haveMismatch =
+        std::any_of(_indexKeyBuckets.begin(),
+                    _indexKeyBuckets.end(),
+                    [](const IndexKeyBucket& bucket) -> bool { return bucket.indexKeyCount; });
+
+    if (haveMismatch && _validateState->logDiagnostics()) {
+        for (size_t i = 0; i < _indexKeyBuckets.size(); i++) {
+            if (_indexKeyBuckets[i].indexKeyCount == 0) {
+                continue;
+            }
+
+            LOGV2(7404500,
+                  "[validate](bucket entry mismatch)",
+                  "hash"_attr = i,
+                  "indexKeyCount"_attr = _indexKeyBuckets[i].indexKeyCount,
+                  "bucketBytesSize"_attr = _indexKeyBuckets[i].bucketSizeBytes);
+        }
+    }
+
+    return haveMismatch;
 }
 
 void IndexConsistency::setSecondPhase() {
@@ -192,7 +209,7 @@ void IndexConsistency::repairMissingIndexEntries(OperationContext* opCtx,
     }
 }
 
-void IndexConsistency::addIndexEntryErrors(ValidateResults* results) {
+void IndexConsistency::addIndexEntryErrors(OperationContext* opCtx, ValidateResults* results) {
     invariant(!_firstPhase);
 
     // We'll report up to 1MB for extra index entry errors and missing index entry errors.
@@ -206,11 +223,26 @@ void IndexConsistency::addIndexEntryErrors(ValidateResults* results) {
         numExtraIndexEntryErrors += item.second.size();
     }
 
+    // Sort missing index entries by size so we can process in order of increasing size and return
+    // as many as possible within memory limits.
+    using MissingIt = decltype(_missingIndexEntries)::const_iterator;
+    std::vector<MissingIt> missingIndexEntriesBySize;
+    missingIndexEntriesBySize.reserve(_missingIndexEntries.size());
+    for (auto it = _missingIndexEntries.begin(); it != _missingIndexEntries.end(); ++it) {
+        missingIndexEntriesBySize.push_back(it);
+    }
+    std::sort(missingIndexEntriesBySize.begin(),
+              missingIndexEntriesBySize.end(),
+              [](const MissingIt& a, const MissingIt& b) {
+                  return a->second.keyString.getSize() < b->second.keyString.getSize();
+              });
+
     // Inform which indexes have inconsistencies and add the BSON objects of the inconsistent index
     // entries to the results vector.
     bool missingIndexEntrySizeLimitWarning = false;
-    for (const auto& missingIndexEntry : _missingIndexEntries) {
-        const IndexEntryInfo& entryInfo = missingIndexEntry.second;
+    bool first = true;
+    for (const auto& missingIndexEntry : missingIndexEntriesBySize) {
+        const IndexEntryInfo& entryInfo = missingIndexEntry->second;
         KeyString::Value ks = entryInfo.keyString;
         auto indexKey =
             KeyString::toBsonSafe(ks.getBuffer(), ks.getSize(), entryInfo.ord, ks.getTypeBits());
@@ -221,8 +253,9 @@ void IndexConsistency::addIndexEntryErrors(ValidateResults* results) {
                                             entryInfo.idKey);
 
         numMissingIndexEntriesSizeBytes += entry.objsize();
-        if (numMissingIndexEntriesSizeBytes <= kErrorSizeBytes) {
+        if (first || numMissingIndexEntriesSizeBytes <= kErrorSizeBytes) {
             results->missingIndexEntries.push_back(entry);
+            first = false;
         } else if (!missingIndexEntrySizeLimitWarning) {
             StringBuilder ss;
             ss << "Not all missing index entry inconsistencies are listed due to size limitations.";
@@ -231,6 +264,8 @@ void IndexConsistency::addIndexEntryErrors(ValidateResults* results) {
             missingIndexEntrySizeLimitWarning = true;
         }
 
+        _printMetadata(opCtx, results, entryInfo);
+
         std::string indexName = entry["indexName"].String();
         if (!results->indexResultsMap.at(indexName).valid) {
             continue;
@@ -243,33 +278,58 @@ void IndexConsistency::addIndexEntryErrors(ValidateResults* results) {
         results->indexResultsMap.at(indexName).valid = false;
     }
 
-    bool extraIndexEntrySizeLimitWarning = false;
+    // Sort extra index entries by size so we can process in order of increasing size and return as
+    // many as possible within memory limits.
+    using ExtraIt = SimpleBSONObjSet::const_iterator;
+    std::vector<ExtraIt> extraIndexEntriesBySize;
+    // Since the extra entries are stored in a map of sets, we have to iterate the entries in the
+    // map and sum the size of the sets in order to get the total number. Given that we can have at
+    // most 64 indexes per collection, and the total number of entries could potentially be in the
+    // millions, we expect that iterating the map will be much less costly than the additional
+    // allocations and copies that could result from not calling 'reserve' on the vector.
+    size_t totalExtraIndexEntriesCount =
+        std::accumulate(_extraIndexEntries.begin(),
+                        _extraIndexEntries.end(),
+                        0,
+                        [](size_t total, const std::pair<IndexKey, SimpleBSONObjSet>& set) {
+                            return total + set.second.size();
+                        });
+    extraIndexEntriesBySize.reserve(totalExtraIndexEntriesCount);
     for (const auto& extraIndexEntry : _extraIndexEntries) {
         const SimpleBSONObjSet& entries = extraIndexEntry.second;
-        for (const auto& entry : entries) {
-            numExtraIndexEntriesSizeBytes += entry.objsize();
-            if (numExtraIndexEntriesSizeBytes <= kErrorSizeBytes) {
-                results->extraIndexEntries.push_back(entry);
-            } else if (!extraIndexEntrySizeLimitWarning) {
-                StringBuilder ss;
-                ss << "Not all extra index entry inconsistencies are listed due to size "
-                      "limitations.";
-                results->errors.push_back(ss.str());
-
-                extraIndexEntrySizeLimitWarning = true;
-            }
-
-            std::string indexName = entry["indexName"].String();
-            if (!results->indexResultsMap.at(indexName).valid) {
-                continue;
-            }
+        for (auto it = entries.begin(); it != entries.end(); ++it) {
+            extraIndexEntriesBySize.push_back(it);
+        }
+    }
+    std::sort(extraIndexEntriesBySize.begin(),
+              extraIndexEntriesBySize.end(),
+              [](const ExtraIt& a, const ExtraIt& b) { return a->objsize() < b->objsize(); });
 
+    bool extraIndexEntrySizeLimitWarning = false;
+    for (const auto& entry : extraIndexEntriesBySize) {
+        numExtraIndexEntriesSizeBytes += entry->objsize();
+        if (first || numExtraIndexEntriesSizeBytes <= kErrorSizeBytes) {
+            results->extraIndexEntries.push_back(*entry);
+            first = false;
+        } else if (!extraIndexEntrySizeLimitWarning) {
             StringBuilder ss;
-            ss << "Index with name '" << indexName << "' has inconsistencies.";
+            ss << "Not all extra index entry inconsistencies are listed due to size "
+                  "limitations.";
             results->errors.push_back(ss.str());
 
-            results->indexResultsMap.at(indexName).valid = false;
+            extraIndexEntrySizeLimitWarning = true;
         }
+
+        std::string indexName = (*entry)["indexName"].String();
+        if (!results->indexResultsMap.at(indexName).valid) {
+            continue;
+        }
+
+        StringBuilder ss;
+        ss << "Index with name '" << indexName << "' has inconsistencies.";
+        results->errors.push_back(ss.str());
+
+        results->indexResultsMap.at(indexName).valid = false;
     }
 
     // Inform how many inconsistencies were detected.
@@ -302,7 +362,8 @@ void IndexConsistency::addDocumentMultikeyPaths(IndexInfo* indexInfo,
 void IndexConsistency::addDocKey(OperationContext* opCtx,
                                  const KeyString::Value& ks,
                                  IndexInfo* indexInfo,
-                                 RecordId recordId) {
+                                 RecordId recordId,
+                                 ValidateResults* results) {
     auto rawHash = ks.hash(indexInfo->indexNameHash);
     auto hashLower = rawHash % kNumHashBuckets;
     auto hashUpper = (rawHash / kNumHashBuckets) % kNumHashBuckets;
@@ -318,7 +379,7 @@ void IndexConsistency::addDocKey(OperationContext* opCtx,
         upper.bucketSizeBytes += ks.getSize();
         indexInfo->numRecords++;
 
-        if (MONGO_unlikely(_validateState->extraLoggingForTest())) {
+        if (MONGO_unlikely(_validateState->logDiagnostics())) {
             LOGV2(4666602,
                   "[validate](record) Adding with hashes",
                   "hashUpper"_attr = hashUpper,
@@ -348,9 +409,6 @@ void IndexConsistency::addDocKey(OperationContext* opCtx,
         invariant(_missingIndexEntries.count(key) == 0);
         _missingIndexEntries.insert(
             std::make_pair(key, IndexEntryInfo(*indexInfo, recordId, idKeyBuilder.obj(), ks)));
-
-        // Prints the collection document's metadata.
-        _validateState->getCollection()->getRecordStore()->printRecordMetadata(opCtx, recordId);
     }
 }
 
@@ -374,7 +432,7 @@ void IndexConsistency::addIndexKey(OperationContext* opCtx,
         upper.bucketSizeBytes += ks.getSize();
         indexInfo->numKeys++;
 
-        if (MONGO_unlikely(_validateState->extraLoggingForTest())) {
+        if (MONGO_unlikely(_validateState->logDiagnostics())) {
             LOGV2(4666603,
                   "[validate](index) Adding with hashes",
                   "hashUpper"_attr = hashUpper,
@@ -423,9 +481,12 @@ void IndexConsistency::addIndexKey(OperationContext* opCtx,
                 SimpleBSONObjSet infoSet = {info};
                 _extraIndexEntries.insert(std::make_pair(key, infoSet));
 
-                // Prints the collection document's metadata.
-                _validateState->getCollection()->getRecordStore()->printRecordMetadata(opCtx,
-                                                                                       recordId);
+                // Prints the collection document's and index entry's metadata.
+                _validateState->getCollection()->getRecordStore()->printRecordMetadata(
+                    opCtx, recordId, &(results->recordTimestamps));
+                indexInfo->accessMethod->asSortedData()
+                    ->getSortedDataInterface()
+                    ->printIndexEntryMetadata(opCtx, ks);
                 return;
             }
             search->second.insert(info);
@@ -453,48 +514,57 @@ bool IndexConsistency::limitMemoryUsageForSecondPhase(ValidateResults* result) {
         return true;
     }
 
-    bool hasNonZeroBucket = false;
-    uint64_t memoryUsedSoFarBytes = 0;
-    uint32_t smallestBucketBytes = std::numeric_limits<uint32_t>::max();
-    // Zero out any nonzero buckets that would put us over maxMemoryUsageBytes.
-    std::for_each(_indexKeyBuckets.begin(), _indexKeyBuckets.end(), [&](IndexKeyBucket& bucket) {
-        if (bucket.indexKeyCount == 0) {
-            return;
-        }
+    // At this point we know we'll exceed the memory limit, and will pare back some of the buckets.
+    // First we'll see what the smallest bucket is, and if that's over the limit by itself, then
+    // we can zero out all the other buckets. Otherwise we'll keep as many buckets as we can.
 
-        smallestBucketBytes = std::min(smallestBucketBytes, bucket.bucketSizeBytes);
-        if (bucket.bucketSizeBytes + memoryUsedSoFarBytes > maxMemoryUsageBytes) {
-            // Including this bucket would put us over the memory limit, so zero this bucket. We
-            // don't want to keep any entry that will exceed the memory limit in the second phase so
-            // we don't double the 'maxMemoryUsageBytes' here.
-            bucket.indexKeyCount = 0;
-            return;
-        }
-        memoryUsedSoFarBytes += bucket.bucketSizeBytes;
-        hasNonZeroBucket = true;
-    });
+    auto smallestBucketWithAnInconsistency = std::min_element(
+        _indexKeyBuckets.begin(),
+        _indexKeyBuckets.end(),
+        [](const IndexKeyBucket& lhs, const IndexKeyBucket& rhs) {
+            if (lhs.indexKeyCount != 0) {
+                return rhs.indexKeyCount == 0 || lhs.bucketSizeBytes < rhs.bucketSizeBytes;
+            }
+            return false;
+        });
+    invariant(smallestBucketWithAnInconsistency->indexKeyCount != 0);
+
+    if (smallestBucketWithAnInconsistency->bucketSizeBytes > maxMemoryUsageBytes) {
+        // We're going to just keep the smallest bucket, and zero everything else.
+        std::for_each(
+            _indexKeyBuckets.begin(), _indexKeyBuckets.end(), [&](IndexKeyBucket& bucket) {
+                if (&bucket == &(*smallestBucketWithAnInconsistency)) {
+                    // We keep the smallest bucket.
+                    return;
+                }
 
-    StringBuilder memoryLimitMessage;
-    memoryLimitMessage << "Memory limit for validation is currently set to "
-                       << maxValidateMemoryUsageMB.load()
-                       << "MB and can be configured via the 'maxValidateMemoryUsageMB' parameter.";
+                bucket.indexKeyCount = 0;
+            });
+    } else {
+        // We're going to scan through the buckets and keep as many as we can.
+        std::uint32_t memoryUsedSoFarBytes = 0;
+        std::for_each(
+            _indexKeyBuckets.begin(), _indexKeyBuckets.end(), [&](IndexKeyBucket& bucket) {
+                if (bucket.indexKeyCount == 0) {
+                    return;
+                }
 
-    if (!hasNonZeroBucket) {
-        const uint32_t minMemoryNeededMB = (smallestBucketBytes / (1024 * 1024)) + 1;
-        StringBuilder ss;
-        ss << "Unable to report index entry inconsistencies due to memory limitations. Need at "
-              "least "
-           << minMemoryNeededMB << "MB to report at least one index entry inconsistency. "
-           << memoryLimitMessage.str();
-        result->errors.push_back(ss.str());
-        result->valid = false;
-
-        return false;
+                if (bucket.bucketSizeBytes + memoryUsedSoFarBytes > maxMemoryUsageBytes) {
+                    // Including this bucket would put us over the memory limit, so zero this
+                    // bucket. We don't want to keep any entry that will exceed the memory limit in
+                    // the second phase so we don't double the 'maxMemoryUsageBytes' here.
+                    bucket.indexKeyCount = 0;
+                    return;
+                }
+                memoryUsedSoFarBytes += bucket.bucketSizeBytes;
+            });
     }
 
     StringBuilder ss;
-    ss << "Not all index entry inconsistencies are reported due to memory limitations. "
-       << memoryLimitMessage.str();
+    ss << "Not all index entry inconsistencies are reported due to memory limitations. Memory "
+          "limit for validation is currently set to "
+       << maxValidateMemoryUsageMB.load()
+       << "MB and can be configured via the 'maxValidateMemoryUsageMB' parameter.";
     result->errors.push_back(ss.str());
     result->valid = false;
 
@@ -540,4 +610,16 @@ uint32_t IndexConsistency::_hashKeyString(const KeyString::Value& ks,
                                           uint32_t indexNameHash) const {
     return ks.hash(indexNameHash);
 }
+
+void IndexConsistency::_printMetadata(OperationContext* opCtx,
+                                      ValidateResults* results,
+                                      const IndexEntryInfo& entryInfo) {
+    _validateState->getCollection()->getRecordStore()->printRecordMetadata(
+        opCtx, entryInfo.recordId, &(results->recordTimestamps));
+    getIndexInfo(entryInfo.indexName)
+        .accessMethod->asSortedData()
+        ->getSortedDataInterface()
+        ->printIndexEntryMetadata(opCtx, entryInfo.keyString);
+}
+
 }  // namespace mongo
diff --git a/src/mongo/db/catalog/index_consistency.h b/src/mongo/db/catalog/index_consistency.h
index dab1d2d8d97..f4eab27660b 100644
--- a/src/mongo/db/catalog/index_consistency.h
+++ b/src/mongo/db/catalog/index_consistency.h
@@ -108,7 +108,8 @@ public:
     void addDocKey(OperationContext* opCtx,
                    const KeyString::Value& ks,
                    IndexInfo* indexInfo,
-                   RecordId recordId);
+                   RecordId recordId,
+                   ValidateResults* results);
 
     /**
      * During the first phase of validation, given the index entry's KeyString, decrement the
@@ -168,7 +169,7 @@ public:
      * Records the errors gathered from the second phase of index validation into the provided
      * ValidateResultsMap and ValidateResults.
      */
-    void addIndexEntryErrors(ValidateResults* results);
+    void addIndexEntryErrors(OperationContext* opCtx, ValidateResults* results);
 
     /**
      * Sets up this IndexConsistency object to limit memory usage in the second phase of index
@@ -179,8 +180,8 @@ public:
 
 private:
     struct IndexKeyBucket {
-        uint32_t indexKeyCount;
-        uint32_t bucketSizeBytes;
+        uint32_t indexKeyCount = 0;
+        uint32_t bucketSizeBytes = 0;
     };
 
     IndexConsistency() = delete;
@@ -241,5 +242,11 @@ private:
      */
     uint32_t _hashKeyString(const KeyString::Value& ks, uint32_t indexNameHash) const;
 
+    /**
+     * Prints the collection document's and index entry's metadata.
+     */
+    void _printMetadata(OperationContext* opCtx,
+                        ValidateResults* results,
+                        const IndexEntryInfo& info);
 };  // IndexConsistency
 }  // namespace mongo
diff --git a/src/mongo/db/catalog/index_key_validate.cpp b/src/mongo/db/catalog/index_key_validate.cpp
index d6f70219594..45f85b244f7 100644
--- a/src/mongo/db/catalog/index_key_validate.cpp
+++ b/src/mongo/db/catalog/index_key_validate.cpp
@@ -58,7 +58,7 @@
 namespace mongo {
 namespace index_key_validate {
 
-std::function<void(std::set<StringData>&)> filterAllowedIndexFieldNames;
+std::function<void(std::map<StringData, std::set<IndexType>>&)> filterAllowedIndexFieldNames;
 
 using IndexVersion = IndexDescriptor::IndexVersion;
 
@@ -68,6 +68,10 @@ namespace {
 // specification.
 MONGO_FAIL_POINT_DEFINE(skipIndexCreateFieldNameValidation);
 
+// When the skipTTLIndexNaNExpireAfterSecondsValidation failpoint is enabled, validation for
+// TTL index 'expireAfterSeconds' will be disabled.
+MONGO_FAIL_POINT_DEFINE(skipTTLIndexNaNExpireAfterSecondsValidation);
+
 static const std::set<StringData> allowedIdIndexFieldNames = {
     IndexDescriptor::kCollationFieldName,
     IndexDescriptor::kIndexNameFieldName,
@@ -104,12 +108,16 @@ Status isIndexVersionAllowedForCreation(IndexVersion indexVersion, const BSONObj
 BSONObj buildRepairedIndexSpec(
     const NamespaceString& ns,
     const BSONObj& indexSpec,
-    const std::set<StringData>& allowedFieldNames,
+    const std::map<StringData, std::set<IndexType>>& allowedFieldNames,
     std::function<void(const BSONElement&, BSONObjBuilder*)> indexSpecHandleFn) {
+    const auto key = indexSpec.getObjectField(IndexDescriptor::kKeyPatternFieldName);
+    const auto indexName = IndexNames::nameToType(IndexNames::findPluginName(key));
     BSONObjBuilder builder;
     for (const auto& indexSpecElem : indexSpec) {
         StringData fieldName = indexSpecElem.fieldNameStringData();
-        if (allowedFieldNames.count(fieldName)) {
+        auto it = allowedFieldNames.find(fieldName);
+        if (it != allowedFieldNames.end() &&
+            (it->second.empty() || it->second.count(indexName) != 0)) {
             indexSpecHandleFn(indexSpecElem, &builder);
         } else {
             LOGV2_WARNING(23878,
@@ -263,9 +271,9 @@ BSONObj removeUnknownFields(const NamespaceString& ns, const BSONObj& indexSpec)
 
 BSONObj repairIndexSpec(const NamespaceString& ns,
                         const BSONObj& indexSpec,
-                        const std::set<StringData>& allowedFieldNames) {
-    auto fixBoolIndexSpecFn = [&indexSpec, &ns](const BSONElement& indexSpecElem,
-                                                BSONObjBuilder* builder) {
+                        const std::map<StringData, std::set<IndexType>>& allowedFieldNames) {
+    auto fixIndexSpecFn = [&indexSpec, &ns](const BSONElement& indexSpecElem,
+                                            BSONObjBuilder* builder) {
         StringData fieldName = indexSpecElem.fieldNameStringData();
         if ((IndexDescriptor::kBackgroundFieldName == fieldName ||
              IndexDescriptor::kUniqueFieldName == fieldName ||
@@ -279,17 +287,28 @@ BSONObj repairIndexSpec(const NamespaceString& ns,
                           "fieldName"_attr = redact(fieldName),
                           "indexSpec"_attr = redact(indexSpec));
             builder->appendBool(fieldName, true);
+        } else if (IndexDescriptor::kExpireAfterSecondsFieldName == fieldName &&
+                   !(indexSpecElem.isNumber() && !indexSpecElem.isNaN())) {
+            LOGV2_WARNING(6835900,
+                          "Fixing expire field from TTL index spec",
+                          "namespace"_attr = redact(ns.toString()),
+                          "fieldName"_attr = redact(fieldName),
+                          "indexSpec"_attr = redact(indexSpec));
+            builder->appendNumber(fieldName,
+                                  durationCount<Seconds>(kExpireAfterSecondsForInactiveTTLIndex));
         } else {
             builder->append(indexSpecElem);
         }
     };
-    return buildRepairedIndexSpec(ns, indexSpec, allowedFieldNames, fixBoolIndexSpecFn);
+
+    return buildRepairedIndexSpec(ns, indexSpec, allowedFieldNames, fixIndexSpecFn);
 }
 
 StatusWith<BSONObj> validateIndexSpec(OperationContext* opCtx, const BSONObj& indexSpec) {
     bool hasKeyPatternField = false;
     bool hasIndexNameField = false;
     bool hasNamespaceField = false;
+    bool isTTLIndexWithNaNExpireAfterSeconds = false;
     bool hasVersionField = false;
     bool hasCollationField = false;
     bool hasWeightsField = false;
@@ -537,6 +556,8 @@ StatusWith<BSONObj> validateIndexSpec(OperationContext* opCtx, const BSONObj& in
                     str::stream() << "The field '" << indexSpecElemFieldName
                                   << "' must be a number, but got "
                                   << typeName(indexSpecElem.type())};
+        } else if (IndexDescriptor::kExpireAfterSecondsFieldName == indexSpecElemFieldName) {
+            isTTLIndexWithNaNExpireAfterSeconds = indexSpecElem.isNaN();
         } else {
             // We can assume field name is valid at this point. Validation of fieldname is handled
             // prior to this in validateIndexSpecFieldNames().
@@ -608,6 +629,19 @@ StatusWith<BSONObj> validateIndexSpec(OperationContext* opCtx, const BSONObj& in
         modifiedSpec = modifiedSpec.removeField(IndexDescriptor::kNamespaceFieldName);
     }
 
+    if (isTTLIndexWithNaNExpireAfterSeconds &&
+        !skipTTLIndexNaNExpireAfterSecondsValidation.shouldFail()) {
+        // We create a new index specification with the 'expireAfterSeconds' field set as
+        // kExpireAfterSecondsForInactiveTTLIndex if the current value is NaN. A similar
+        // treatment is done in repairIndexSpec(). This rewrites the 'expireAfterSeconds'
+        // value to be compliant with the 'safeInt' IDL type for the listIndexes response.
+        BSONObjBuilder builder;
+        builder.appendNumber(IndexDescriptor::kExpireAfterSecondsFieldName,
+                             durationCount<Seconds>(kExpireAfterSecondsForInactiveTTLIndex));
+        auto obj = builder.obj();
+        modifiedSpec = modifiedSpec.addField(obj.firstElement());
+    }
+
     if (!hasVersionField) {
         // We create a new index specification with the 'v' field set as 'defaultIndexVersion' if
         // the field was omitted.
@@ -768,7 +802,8 @@ StatusWith<BSONObj> validateIndexSpecCollation(OperationContext* opCtx,
     return indexSpec;
 }
 
-Status validateExpireAfterSeconds(std::int64_t expireAfterSeconds) {
+Status validateExpireAfterSeconds(std::int64_t expireAfterSeconds,
+                                  ValidateExpireAfterSecondsMode mode) {
     if (expireAfterSeconds < 0) {
         return {ErrorCodes::InvalidOptions,
                 str::stream() << "TTL index '" << IndexDescriptor::kExpireAfterSecondsFieldName
@@ -779,16 +814,31 @@ Status validateExpireAfterSeconds(std::int64_t expireAfterSeconds) {
         << "TTL index '" << IndexDescriptor::kExpireAfterSecondsFieldName
         << "' option must be within an acceptable range, try a lower number";
 
-    // There are two cases where we can encounter an issue here.
-    // The first case is when we try to cast to millseconds from seconds, which could cause an
-    // overflow. The second case is where 'expireAfterSeconds' is larger than the current epoch
-    // time.
-    if (expireAfterSeconds > std::numeric_limits<std::int64_t>::max() / 1000) {
-        return {ErrorCodes::InvalidOptions, tooLargeErr};
-    }
-    auto expireAfterMillis = duration_cast<Milliseconds>(Seconds(expireAfterSeconds));
-    if (expireAfterMillis > Date_t::now().toDurationSinceEpoch()) {
-        return {ErrorCodes::InvalidOptions, tooLargeErr};
+    if (mode == ValidateExpireAfterSecondsMode::kSecondaryTTLIndex) {
+        // Relax epoch restriction on TTL indexes. This allows us to export and import existing
+        // TTL indexes with large values or NaN for the 'expireAfterSeconds' field.
+        // Additionally, the 'expireAfterSeconds' for TTL indexes is defined as safeInt (int32_t)
+        // in the IDL for listIndexes and collMod. See list_indexes.idl and coll_mod.idl.
+        if (expireAfterSeconds > std::numeric_limits<std::int32_t>::max()) {
+            return {ErrorCodes::InvalidOptions, tooLargeErr};
+        }
+    } else {
+        // Clustered collections with TTL.
+        // Note that 'expireAfterSeconds' is defined as safeInt64 in the IDL for the create and
+        // collMod commands. See create.idl and coll_mod.idl.
+        // There are two cases where we can encounter an issue here.
+        // The first case is when we try to cast to millseconds from seconds, which could cause an
+        // overflow. The second case is where 'expireAfterSeconds' is larger than the current epoch
+        // time. This isn't necessarily problematic for the general case, but for the specific case
+        // of time series collections, we cluster the collection by an OID value, where the
+        // timestamp portion is only a 32-bit unsigned integer offset of seconds since the epoch.
+        if (expireAfterSeconds > std::numeric_limits<std::int64_t>::max() / 1000) {
+            return {ErrorCodes::InvalidOptions, tooLargeErr};
+        }
+        auto expireAfterMillis = duration_cast<Milliseconds>(Seconds(expireAfterSeconds));
+        if (expireAfterMillis > Date_t::now().toDurationSinceEpoch()) {
+            return {ErrorCodes::InvalidOptions, tooLargeErr};
+        }
     }
     return Status::OK();
 }
@@ -812,7 +862,9 @@ Status validateIndexSpecTTL(const BSONObj& indexSpec) {
                               << "'. Index spec: " << indexSpec};
     }
 
-    if (auto status = validateExpireAfterSeconds(expireAfterSecondsElt.safeNumberLong());
+    if (auto status =
+            validateExpireAfterSeconds(expireAfterSecondsElt.safeNumberLong(),
+                                       ValidateExpireAfterSecondsMode::kSecondaryTTLIndex);
         !status.isOK()) {
         return {ErrorCodes::CannotCreateIndex,
                 str::stream() << status.reason() << ". Index spec: " << indexSpec};
diff --git a/src/mongo/db/catalog/index_key_validate.h b/src/mongo/db/catalog/index_key_validate.h
index 8f226749c21..45b383fee1c 100644
--- a/src/mongo/db/catalog/index_key_validate.h
+++ b/src/mongo/db/catalog/index_key_validate.h
@@ -43,35 +43,51 @@ class StatusWith;
 
 namespace index_key_validate {
 
-static std::set<StringData> allowedFieldNames = {
-    IndexDescriptor::k2dIndexBitsFieldName,
-    IndexDescriptor::k2dIndexMaxFieldName,
-    IndexDescriptor::k2dIndexMinFieldName,
-    IndexDescriptor::k2dsphereCoarsestIndexedLevel,
-    IndexDescriptor::k2dsphereFinestIndexedLevel,
-    IndexDescriptor::k2dsphereVersionFieldName,
-    IndexDescriptor::kBackgroundFieldName,
-    IndexDescriptor::kCollationFieldName,
-    IndexDescriptor::kDefaultLanguageFieldName,
-    IndexDescriptor::kDropDuplicatesFieldName,
-    IndexDescriptor::kExpireAfterSecondsFieldName,
-    IndexDescriptor::kHiddenFieldName,
-    IndexDescriptor::kIndexNameFieldName,
-    IndexDescriptor::kIndexVersionFieldName,
-    IndexDescriptor::kKeyPatternFieldName,
-    IndexDescriptor::kLanguageOverrideFieldName,
-    IndexDescriptor::kNamespaceFieldName,
-    IndexDescriptor::kPartialFilterExprFieldName,
-    IndexDescriptor::kPathProjectionFieldName,
-    IndexDescriptor::kSparseFieldName,
-    IndexDescriptor::kStorageEngineFieldName,
-    IndexDescriptor::kTextVersionFieldName,
-    IndexDescriptor::kUniqueFieldName,
-    IndexDescriptor::kWeightsFieldName,
-    IndexDescriptor::kOriginalSpecFieldName,
-    IndexDescriptor::kPrepareUniqueFieldName,
+// TTL indexes with 'expireAfterSeconds' are repaired with this duration, which is chosen to be
+// the largest possible value for the 'safeInt' type that can be returned in the listIndexes
+// response.
+constexpr auto kExpireAfterSecondsForInactiveTTLIndex =
+    Seconds(std::numeric_limits<int32_t>::max());
+
+/**
+ * Describe which field names are considered valid options when creating an index. If the set
+ * associated with the field name is empty, the option is always valid, otherwise it will be allowed
+ * only when creating the set of index types listed in the set.
+ */
+static std::map<StringData, std::set<IndexType>> allowedFieldNames = {
+    {IndexDescriptor::k2dIndexBitsFieldName, {IndexType::INDEX_2D}},
+    {IndexDescriptor::k2dIndexMaxFieldName, {IndexType::INDEX_2D}},
+    {IndexDescriptor::k2dIndexMinFieldName, {IndexType::INDEX_2D}},
+    {IndexDescriptor::k2dsphereCoarsestIndexedLevel, {IndexType::INDEX_2DSPHERE}},
+    {IndexDescriptor::k2dsphereFinestIndexedLevel, {IndexType::INDEX_2DSPHERE}},
+    {IndexDescriptor::k2dsphereVersionFieldName,
+     {IndexType::INDEX_2DSPHERE, IndexType::INDEX_2DSPHERE_BUCKET}},
+    {IndexDescriptor::kBackgroundFieldName, {}},
+    {IndexDescriptor::kCollationFieldName, {}},
+    {IndexDescriptor::kDefaultLanguageFieldName, {}},
+    {IndexDescriptor::kDropDuplicatesFieldName, {}},
+    {IndexDescriptor::kExpireAfterSecondsFieldName, {}},
+    {IndexDescriptor::kHiddenFieldName, {}},
+    {IndexDescriptor::kIndexNameFieldName, {}},
+    {IndexDescriptor::kIndexVersionFieldName, {}},
+    {IndexDescriptor::kKeyPatternFieldName, {}},
+    {IndexDescriptor::kLanguageOverrideFieldName, {}},
+    {IndexDescriptor::kNamespaceFieldName, {}},
+    {IndexDescriptor::kPartialFilterExprFieldName, {}},
+    {IndexDescriptor::kPathProjectionFieldName, {IndexType::INDEX_WILDCARD}},
+    {IndexDescriptor::kSparseFieldName, {}},
+    {IndexDescriptor::kStorageEngineFieldName, {}},
+    {IndexDescriptor::kTextVersionFieldName, {IndexType::INDEX_TEXT}},
+    {IndexDescriptor::kUniqueFieldName, {}},
+    {IndexDescriptor::kWeightsFieldName, {IndexType::INDEX_TEXT}},
+    {IndexDescriptor::kOriginalSpecFieldName, {}},
+    {IndexDescriptor::kPrepareUniqueFieldName, {}},
     // Index creation under legacy writeMode can result in an index spec with an _id field.
-    "_id"};
+    {"_id", {}},
+    // TODO SERVER-76108: Field names are not validated to match index type. This was used for the
+    // removed 'geoHaystack' index type, but users could have set it for other index types as well.
+    // We need to keep allowing it until FCV upgrade is implemented to clean this up.
+    {"bucketSize"_sd, {}}};
 
 /**
  * Checks if the key is valid for building an index according to the validation rules for the given
@@ -92,12 +108,12 @@ StatusWith<BSONObj> validateIndexSpec(OperationContext* opCtx, const BSONObj& in
 BSONObj removeUnknownFields(const NamespaceString& ns, const BSONObj& indexSpec);
 
 /**
- * Returns a new index spec with boolean values in correct types and unkown field names removed.
+ * Returns a new index spec with boolean values in correct types and unknown field names removed.
  */
-BSONObj repairIndexSpec(
-    const NamespaceString& ns,
-    const BSONObj& indexSpec,
-    const std::set<StringData>& allowedFieldNames = index_key_validate::allowedFieldNames);
+BSONObj repairIndexSpec(const NamespaceString& ns,
+                        const BSONObj& indexSpec,
+                        const std::map<StringData, std::set<IndexType>>& allowedFieldNames =
+                            index_key_validate::allowedFieldNames);
 
 /**
  * Performs additional validation for _id index specifications. This should be called after
@@ -121,9 +137,14 @@ StatusWith<BSONObj> validateIndexSpecCollation(OperationContext* opCtx,
                                                const CollatorInterface* defaultCollator);
 
 /**
- * Validates the the 'expireAfterSeconds' value for a TTL index..
+ * Validates the the 'expireAfterSeconds' value for a TTL index or clustered collection.
  */
-Status validateExpireAfterSeconds(std::int64_t expireAfterSeconds);
+enum class ValidateExpireAfterSecondsMode {
+    kSecondaryTTLIndex,
+    kClusteredTTLIndex,
+};
+Status validateExpireAfterSeconds(std::int64_t expireAfterSeconds,
+                                  ValidateExpireAfterSecondsMode mode);
 
 /**
  * Returns true if 'indexSpec' refers to a TTL index.
@@ -145,7 +166,7 @@ bool isIndexAllowedInAPIVersion1(const IndexDescriptor& indexDesc);
  * Optional filtering function to adjust allowed index field names at startup.
  * Set it in a MONGO_INITIALIZER with 'FilterAllowedIndexFieldNames' as a dependant.
  */
-extern std::function<void(std::set<StringData>& allowedIndexFieldNames)>
+extern std::function<void(std::map<StringData, std::set<IndexType>>& allowedIndexFieldNames)>
     filterAllowedIndexFieldNames;
 
 }  // namespace index_key_validate
diff --git a/src/mongo/db/catalog/index_key_validate_test.cpp b/src/mongo/db/catalog/index_key_validate_test.cpp
index 2c25d1b8791..659a0bfb7b2 100644
--- a/src/mongo/db/catalog/index_key_validate_test.cpp
+++ b/src/mongo/db/catalog/index_key_validate_test.cpp
@@ -318,5 +318,100 @@ TEST(IndexKeyValidateTest, Background) {
         nullptr, fromjson("{key: {a: 1}, name: 'index', background: []}")));
 }
 
+TEST(IndexKeyValidateTest, RemoveUnkownFieldsFromIndexSpecs) {
+    ASSERT(fromjson("{key: {a: 1}, name: 'index'}")
+               .binaryEqual(index_key_validate::removeUnknownFields(
+                   NamespaceString("coll"),
+                   fromjson("{key: {a: 1}, name: 'index', safe: true, force: true}"))));
+}
+
+TEST(IndexKeyValidateTest, UpdateTTLIndexNaNExpireAfterSeconds) {
+    ASSERT(BSON("key" << BSON("a" << 1) << "name"
+                      << "index"
+                      << "expireAfterSeconds" << std::numeric_limits<int32_t>::max()
+                      << IndexDescriptor::kIndexVersionFieldName << IndexVersion::kV2)
+               .binaryEqual(unittest::assertGet(index_key_validate::validateIndexSpec(
+                   nullptr, fromjson("{key: {a: 1}, name: 'index', expireAfterSeconds: NaN}")))));
+}
+
+TEST(IndexKeyValidateTest, ValidateAfterSecondsAcceptsFloatingPointNumber) {
+    auto spec = unittest::assertGet(index_key_validate::validateIndexSpec(
+        nullptr, fromjson("{key: {a: 1}, name: 'index', expireAfterSeconds: 123.456}")));
+
+    // TTLMonitor extracts 'expireAfterSeconds' using BSONElement::safeNumberLong().
+    ASSERT_EQUALS(spec["expireAfterSeconds"].safeNumberLong(), 123LL);
+}
+
+TEST(IndexKeyValidateTest, RepairIndexSpecs) {
+    ASSERT(fromjson("{key: {a: 1}, name: 'index'}")
+               .binaryEqual(index_key_validate::repairIndexSpec(
+                   NamespaceString("coll"),
+                   fromjson("{key: {a: 1}, name: 'index', safe: true, force: true}"))));
+
+    ASSERT(fromjson("{key: {a: 1}, name: 'index', sparse: true}")
+               .binaryEqual(index_key_validate::repairIndexSpec(
+                   NamespaceString("coll"),
+                   fromjson("{key: {a: 1}, name: 'index', sparse: 'true'}"))));
+
+    ASSERT(fromjson("{key: {a: 1}, name: 'index', background: true}")
+               .binaryEqual(index_key_validate::repairIndexSpec(
+                   NamespaceString("coll"),
+                   fromjson("{key: {a: 1}, name: 'index', background: '1'}"))));
+
+    ASSERT(fromjson("{key: {a: 1}, name: 'index', sparse: true, background: true}")
+               .binaryEqual(index_key_validate::repairIndexSpec(
+                   NamespaceString("coll"),
+                   fromjson("{key: {a: 1}, name: 'index', sparse: 'true', background: '1'}"))));
+
+    ASSERT(fromjson("{key: {a: 1}, name: 'index', sparse: true, background: true}")
+               .binaryEqual(index_key_validate::repairIndexSpec(
+                   NamespaceString("coll"),
+                   fromjson("{key: {a: 1}, name: 'index', sparse: 'true', background: '1', safe: "
+                            "true, force: true}"))));
+
+    ASSERT(fromjson("{key: {a: 1}, name: 'index'}")
+               .binaryEqual(index_key_validate::repairIndexSpec(
+                   NamespaceString("coll"),
+                   fromjson("{key: {a: 1}, name: 'index', weights: {key: 1, name: 1}}"))));
+
+    ASSERT(fromjson("{key: {'a': 'text'}, name: 'index', weights: {key: 1, name: 1}}")
+               .binaryEqual(index_key_validate::repairIndexSpec(
+                   NamespaceString("coll"),
+                   fromjson("{key: {'a': 'text'}, name: 'index', weights: {key: 1, name: 1}}"))));
+
+    ASSERT(fromjson("{key: {'$**': 'text'}, name: 'index', weights: {key: 1, name: 1}}")
+               .binaryEqual(index_key_validate::repairIndexSpec(
+                   NamespaceString("coll"),
+                   fromjson("{key: {'$**': 'text'}, name: 'index', weights: {key: 1, name: 1}}"))));
+
+    ASSERT(fromjson("{key: {'data.loc' : '2dsphere_bucket'}, 'name': 'loc_2dsphere', "
+                    "'2dsphereIndexVersion' : 3}")
+               .binaryEqual(index_key_validate::repairIndexSpec(
+                   NamespaceString("coll"),
+                   fromjson("{key: {'data.loc' : '2dsphere_bucket'}, 'name': 'loc_2dsphere', "
+                            "'2dsphereIndexVersion' : 3}"))));
+
+    ASSERT(
+        fromjson("{key: {a: 1, 'name': 'text'}, name: 'index', weights: {key: 1, name: 1}}")
+            .binaryEqual(index_key_validate::repairIndexSpec(
+                NamespaceString("coll"),
+                fromjson(
+                    "{key: {a: 1, 'name': 'text'}, name: 'index', weights: {key: 1, name: 1}}"))));
+
+    ASSERT(BSON("key" << BSON("a" << 1) << "name"
+                      << "index"
+                      << "expireAfterSeconds" << std::numeric_limits<int32_t>::max())
+               .binaryEqual(index_key_validate::repairIndexSpec(
+                   NamespaceString("coll"),
+                   fromjson("{key: {a: 1}, name: 'index', expireAfterSeconds: NaN}"))));
+
+    ASSERT(BSON("key" << BSON("a" << 1) << "name"
+                      << "index"
+                      << "expireAfterSeconds" << std::numeric_limits<int32_t>::max())
+               .binaryEqual(index_key_validate::repairIndexSpec(
+                   NamespaceString("coll"),
+                   fromjson("{key: {a: 1}, name: 'index', expireAfterSeconds: '123'}"))));
+}
+
 }  // namespace
 }  // namespace mongo
diff --git a/src/mongo/db/catalog/index_repair.cpp b/src/mongo/db/catalog/index_repair.cpp
index 6effefcd909..4e863cec74c 100644
--- a/src/mongo/db/catalog/index_repair.cpp
+++ b/src/mongo/db/catalog/index_repair.cpp
@@ -31,7 +31,7 @@
 #include "mongo/base/status_with.h"
 #include "mongo/db/catalog/validate_state.h"
 #include "mongo/db/catalog_raii.h"
-#include "mongo/db/concurrency/write_conflict_exception.h"
+#include "mongo/db/concurrency/exception_util.h"
 #include "mongo/db/index/index_access_method.h"
 #include "mongo/logv2/log_debug.h"
 
diff --git a/src/mongo/db/catalog/list_indexes.cpp b/src/mongo/db/catalog/list_indexes.cpp
index ce1515a4c13..660c4068f83 100644
--- a/src/mongo/db/catalog/list_indexes.cpp
+++ b/src/mongo/db/catalog/list_indexes.cpp
@@ -38,7 +38,7 @@
 #include "mongo/bson/bsonobj.h"
 #include "mongo/bson/bsonobjbuilder.h"
 #include "mongo/db/catalog/clustered_collection_util.h"
-#include "mongo/db/concurrency/write_conflict_exception.h"
+#include "mongo/db/concurrency/exception_util.h"
 #include "mongo/db/curop_failpoint_helpers.h"
 #include "mongo/db/db_raii.h"
 #include "mongo/db/namespace_string.h"
diff --git a/src/mongo/db/catalog/multi_index_block.cpp b/src/mongo/db/catalog/multi_index_block.cpp
index 324f6e489ae..e39dc2ed754 100644
--- a/src/mongo/db/catalog/multi_index_block.cpp
+++ b/src/mongo/db/catalog/multi_index_block.cpp
@@ -41,7 +41,7 @@
 #include "mongo/db/catalog/collection_catalog.h"
 #include "mongo/db/catalog/multi_index_block_gen.h"
 #include "mongo/db/client.h"
-#include "mongo/db/concurrency/write_conflict_exception.h"
+#include "mongo/db/concurrency/exception_util.h"
 #include "mongo/db/index/multikey_paths.h"
 #include "mongo/db/multi_key_path_tracker.h"
 #include "mongo/db/op_observer.h"
@@ -721,7 +721,6 @@ Status MultiIndexBlock::_insert(OperationContext* opCtx,
         try {
             idxStatus = _indexes[i].bulk->insert(opCtx,
                                                  collection,
-                                                 _indexes[i].block->getPooledBuilder(),
                                                  doc,
                                                  loc,
                                                  _indexes[i].options,
diff --git a/src/mongo/db/catalog/rename_collection.cpp b/src/mongo/db/catalog/rename_collection.cpp
index 186e4a38330..d828afac1f3 100644
--- a/src/mongo/db/catalog/rename_collection.cpp
+++ b/src/mongo/db/catalog/rename_collection.cpp
@@ -43,8 +43,8 @@
 #include "mongo/db/catalog/list_indexes.h"
 #include "mongo/db/catalog/local_oplog_info.h"
 #include "mongo/db/client.h"
+#include "mongo/db/concurrency/exception_util.h"
 #include "mongo/db/concurrency/lock_state.h"
-#include "mongo/db/concurrency/write_conflict_exception.h"
 #include "mongo/db/curop.h"
 #include "mongo/db/db_raii.h"
 #include "mongo/db/index/index_descriptor.h"
@@ -116,7 +116,10 @@ Status checkSourceAndTargetNamespaces(OperationContext* opCtx,
                       str::stream() << "Source collection " << source.ns() << " does not exist");
     }
 
-    if (sourceColl->getCollectionOptions().encryptedFieldConfig) {
+    if (sourceColl->getCollectionOptions().encryptedFieldConfig &&
+        !AuthorizationSession::get(opCtx->getClient())
+             ->isAuthorizedForActionsOnResource(ResourcePattern::forClusterResource(),
+                                                ActionType::setUserWriteBlockMode)) {
         return Status(ErrorCodes::IllegalOperation, "Cannot rename an encrypted collection");
     }
 
@@ -129,7 +132,10 @@ Status checkSourceAndTargetNamespaces(OperationContext* opCtx,
             return Status(ErrorCodes::NamespaceExists,
                           str::stream() << "a view already exists with that name: " << target);
     } else {
-        if (targetColl->getCollectionOptions().encryptedFieldConfig) {
+        if (targetColl->getCollectionOptions().encryptedFieldConfig &&
+            !AuthorizationSession::get(opCtx->getClient())
+                 ->isAuthorizedForActionsOnResource(ResourcePattern::forClusterResource(),
+                                                    ActionType::setUserWriteBlockMode)) {
             return Status(ErrorCodes::IllegalOperation,
                           "Cannot rename to an existing encrypted collection");
         }
@@ -622,7 +628,10 @@ Status renameBetweenDBs(OperationContext* opCtx,
 
     // Copy the index descriptions from the source collection.
     std::vector<BSONObj> indexesToCopy;
-    for (auto sourceIndIt = sourceColl->getIndexCatalog()->getIndexIterator(opCtx, true);
+    for (auto sourceIndIt = sourceColl->getIndexCatalog()->getIndexIterator(
+             opCtx,
+             IndexCatalog::InclusionPolicy::kReady | IndexCatalog::InclusionPolicy::kUnfinished |
+                 IndexCatalog::InclusionPolicy::kFrozen);
          sourceIndIt->more();) {
         auto descriptor = sourceIndIt->next()->descriptor();
         if (descriptor->isIdIndex()) {
@@ -854,9 +863,18 @@ void validateNamespacesForRenameCollection(OperationContext* opCtx,
             "renaming system.views collection or renaming to system.views is not allowed",
             !source.isSystemDotViews() && !target.isSystemDotViews());
 
-    uassert(ErrorCodes::IllegalOperation,
-            "Renaming system.buckets collections is not allowed",
-            !source.isTimeseriesBucketsCollection());
+    if (source.isTimeseriesBucketsCollection()) {
+        uassert(ErrorCodes::IllegalOperation,
+                "Renaming system.buckets collections is not allowed",
+                AuthorizationSession::get(opCtx->getClient())
+                    ->isAuthorizedForActionsOnResource(ResourcePattern::forClusterResource(),
+                                                       ActionType::setUserWriteBlockMode));
+
+        uassert(ErrorCodes::IllegalOperation,
+                str::stream() << "Cannot rename time-series buckets collection {" << source.ns()
+                              << "} to a non-time-series buckets namespace {" << target.ns() << "}",
+                target.isTimeseriesBucketsCollection());
+    }
 }
 
 void validateAndRunRenameCollection(OperationContext* opCtx,
diff --git a/src/mongo/db/catalog/rename_collection_test.cpp b/src/mongo/db/catalog/rename_collection_test.cpp
index 215069fd9b7..b092cf4fffe 100644
--- a/src/mongo/db/catalog/rename_collection_test.cpp
+++ b/src/mongo/db/catalog/rename_collection_test.cpp
@@ -40,7 +40,7 @@
 #include "mongo/db/catalog/index_catalog.h"
 #include "mongo/db/catalog/rename_collection.h"
 #include "mongo/db/client.h"
-#include "mongo/db/concurrency/write_conflict_exception.h"
+#include "mongo/db/concurrency/exception_util.h"
 #include "mongo/db/db_raii.h"
 #include "mongo/db/index/index_descriptor.h"
 #include "mongo/db/index_builds_coordinator.h"
diff --git a/src/mongo/db/catalog/validate_adaptor.cpp b/src/mongo/db/catalog/validate_adaptor.cpp
index acbef39ba9c..34d2cb042fd 100644
--- a/src/mongo/db/catalog/validate_adaptor.cpp
+++ b/src/mongo/db/catalog/validate_adaptor.cpp
@@ -41,7 +41,7 @@
 #include "mongo/db/catalog/index_catalog.h"
 #include "mongo/db/catalog/index_consistency.h"
 #include "mongo/db/catalog/throttle_cursor.h"
-#include "mongo/db/concurrency/write_conflict_exception.h"
+#include "mongo/db/concurrency/exception_util.h"
 #include "mongo/db/curop.h"
 #include "mongo/db/index/index_access_method.h"
 #include "mongo/db/index/index_descriptor.h"
@@ -63,6 +63,7 @@ namespace mongo {
 namespace {
 
 MONGO_FAIL_POINT_DEFINE(crashOnMultikeyValidateFailure);
+MONGO_FAIL_POINT_DEFINE(failIndexKeyOrdering);
 
 // Set limit for size of corrupted records that will be reported.
 const long long kMaxErrorSizeBytes = 1 * 1024 * 1024;
@@ -129,6 +130,24 @@ void schemaValidationFailed(CollectionValidation::ValidateState* state,
     }
 }
 
+
+BSONObj rehydrateKey(const BSONObj& keyPattern, const BSONObj& indexKey) {
+    // We need to rehydrate the indexKey for improved readability.
+    // {"": ObjectId(...)} -> {"_id": ObjectId(...)}
+    auto keysIt = keyPattern.begin();
+    auto valuesIt = indexKey.begin();
+
+    BSONObjBuilder b;
+    while (keysIt != keyPattern.end()) {
+        // keysIt and valuesIt must have the same number of elements.
+        invariant(valuesIt != indexKey.end());
+        b.appendAs(*valuesIt, keysIt->fieldName());
+        keysIt++;
+        valuesIt++;
+    }
+
+    return b.obj();
+}
 }  // namespace
 
 Status ValidateAdaptor::validateRecord(OperationContext* opCtx,
@@ -143,7 +162,7 @@ Status ValidateAdaptor::validateRecord(OperationContext* opCtx,
     BSONObj recordBson = record.toBson();
     *dataSize = recordBson.objsize();
 
-    if (MONGO_unlikely(_validateState->extraLoggingForTest())) {
+    if (MONGO_unlikely(_validateState->logDiagnostics())) {
         LOGV2(4666601, "[validate]", "recordId"_attr = recordId, "recordData"_attr = recordBson);
     }
 
@@ -187,6 +206,26 @@ Status ValidateAdaptor::validateRecord(OperationContext* opCtx,
             {multikeyMetadataKeys->begin(), multikeyMetadataKeys->end()},
             *documentMultikeyPaths);
 
+        auto printMultikeyMetadata = [&]() {
+            LOGV2(7556100,
+                  "Index is not multikey but document has multikey data",
+                  "indexName"_attr = descriptor->indexName(),
+                  "recordId"_attr = recordId,
+                  "record"_attr = redact(recordBson));
+            for (auto& key : *documentKeySet) {
+                auto indexKey = KeyString::toBsonSafe(key.getBuffer(),
+                                                      key.getSize(),
+                                                      iam->getSortedDataInterface()->getOrdering(),
+                                                      key.getTypeBits());
+                const BSONObj rehydratedKey = rehydrateKey(descriptor->keyPattern(), indexKey);
+                LOGV2(7556101,
+                      "Index key for document with multikey inconsistency",
+                      "indexName"_attr = descriptor->indexName(),
+                      "recordId"_attr = recordId,
+                      "indexKey"_attr = redact(rehydratedKey));
+            }
+        };
+
         if (!index->isMultikey(opCtx, coll) && shouldBeMultikey) {
             if (_validateState->fixErrors()) {
                 writeConflictRetry(opCtx, "setIndexAsMultikey", coll->ns().ns(), [&] {
@@ -204,10 +243,17 @@ Status ValidateAdaptor::validateRecord(OperationContext* opCtx,
                                                           << " set to multikey.");
                 results->repaired = true;
             } else {
+                printMultikeyMetadata();
+
                 auto& curRecordResults = (results->indexResultsMap)[descriptor->indexName()];
-                std::string msg = str::stream() << "Index " << descriptor->indexName()
-                                                << " is not multikey but has more than one"
-                                                << " key in document " << recordId;
+                const std::string msg = fmt::format(
+                    "Index {} is not multikey but document with RecordId({}) and {} has multikey "
+                    "data, "
+                    "{} key(s)",
+                    descriptor->indexName(),
+                    recordId.toString(),
+                    recordBson.getField("_id").toString(),
+                    documentKeySet->size());
                 curRecordResults.errors.push_back(msg);
                 curRecordResults.valid = false;
                 if (crashOnMultikeyValidateFailure.shouldFail()) {
@@ -235,6 +281,8 @@ Status ValidateAdaptor::validateRecord(OperationContext* opCtx,
                                                               << " multikey paths updated.");
                     results->repaired = true;
                 } else {
+                    printMultikeyMetadata();
+
                     std::string msg = str::stream()
                         << "Index " << descriptor->indexName()
                         << " multikey paths do not cover a document. RecordId: " << recordId;
@@ -267,7 +315,7 @@ Status ValidateAdaptor::validateRecord(OperationContext* opCtx,
         for (const auto& keyString : *documentKeySet) {
             try {
                 _totalIndexKeys++;
-                _indexConsistency->addDocKey(opCtx, keyString, &indexInfo, recordId);
+                _indexConsistency->addDocKey(opCtx, keyString, &indexInfo, recordId, results);
             } catch (...) {
                 return exceptionToStatus();
             }
@@ -288,7 +336,7 @@ void _validateKeyOrder(OperationContext* opCtx,
 
     // KeyStrings will be in strictly increasing order because all keys are sorted and they are in
     // the format (Key, RID), and all RecordIDs are unique.
-    if (currKey.compare(prevKey) <= 0) {
+    if (currKey.compare(prevKey) <= 0 || MONGO_unlikely(failIndexKeyOrdering.shouldFail())) {
         if (results && results->valid) {
             results->errors.push_back(str::stream()
                                       << "index '" << descriptor->indexName()
@@ -570,11 +618,6 @@ void ValidateAdaptor::traverseRecordStore(OperationContext* opCtx,
         size_t validatedSize = 0;
         Status status = validateRecord(opCtx, record->id, record->data, &validatedSize, results);
 
-        // RecordStores are required to return records in RecordId order.
-        if (prevRecordId.isValid()) {
-            invariant(prevRecordId < record->id);
-        }
-
         // validatedSize = dataSize is not a general requirement as some storage engines may use
         // padding, but we still require that they return the unpadded record data.
         if (!status.isOK() || validatedSize != static_cast<size_t>(dataSize)) {
diff --git a/src/mongo/db/catalog/validate_results.h b/src/mongo/db/catalog/validate_results.h
index baa6d78ad42..c02235a68bc 100644
--- a/src/mongo/db/catalog/validate_results.h
+++ b/src/mongo/db/catalog/validate_results.h
@@ -30,6 +30,7 @@
 #pragma once
 
 #include <map>
+#include <set>
 #include <string>
 #include <vector>
 
@@ -44,7 +45,6 @@ struct IndexValidateResults {
     std::vector<std::string> errors;
     std::vector<std::string> warnings;
     int64_t keysTraversed = 0;
-    int64_t keysTraversedFromFullValidate = 0;
     int64_t keysRemovedFromRecordStore = 0;
 };
 
@@ -60,6 +60,9 @@ struct ValidateResults {
     std::vector<BSONObj> extraIndexEntries;
     std::vector<BSONObj> missingIndexEntries;
     std::vector<RecordId> corruptRecords;
+    // Timestamps (startTs, startDurable, stopTs, stopDurableTs) related to records
+    // with validation errors. See WiredTigerRecordStore::printRecordMetadata().
+    std::set<Timestamp> recordTimestamps;
     long long numRemovedCorruptRecords = 0;
     long long numRemovedExtraIndexEntries = 0;
     long long numInsertedMissingIndexEntries = 0;
diff --git a/src/mongo/db/catalog/validate_state.cpp b/src/mongo/db/catalog/validate_state.cpp
index 434950b9540..f7a3df678b6 100644
--- a/src/mongo/db/catalog/validate_state.cpp
+++ b/src/mongo/db/catalog/validate_state.cpp
@@ -54,12 +54,12 @@ ValidateState::ValidateState(OperationContext* opCtx,
                              const NamespaceString& nss,
                              ValidateMode mode,
                              RepairMode repairMode,
-                             bool turnOnExtraLoggingForTest)
+                             bool logDiagnostics)
     : _nss(nss),
       _mode(mode),
       _repairMode(repairMode),
       _dataThrottle(opCtx),
-      _extraLoggingForTest(turnOnExtraLoggingForTest) {
+      _logDiagnostics(logDiagnostics) {
 
     // Subsequent re-locks will use the UUID when 'background' is true.
     if (isBackground()) {
@@ -238,15 +238,16 @@ void ValidateState::initializeCursors(OperationContext* opCtx) {
     const IndexCatalog* indexCatalog = _collection->getIndexCatalog();
     // The index iterator for ready indexes is timestamp-aware and will only return indexes that
     // are visible at our read time.
-    const std::unique_ptr<IndexCatalog::IndexIterator> it =
-        indexCatalog->getIndexIterator(opCtx, /*includeUnfinished*/ false);
+    const auto it = indexCatalog->getIndexIterator(opCtx, IndexCatalog::InclusionPolicy::kReady);
     while (it->more()) {
         const IndexCatalogEntry* entry = it->next();
         const IndexDescriptor* desc = entry->descriptor();
 
         auto iam = entry->accessMethod()->asSortedData();
-        if (!iam)
+        if (!iam) {
+            _skippedIndexes.emplace(desc->indexName());
             continue;
+        }
 
         _indexCursors.emplace(
             desc->indexName(),
diff --git a/src/mongo/db/catalog/validate_state.h b/src/mongo/db/catalog/validate_state.h
index df796c686ce..8144ddb5cc3 100644
--- a/src/mongo/db/catalog/validate_state.h
+++ b/src/mongo/db/catalog/validate_state.h
@@ -55,15 +55,11 @@ class ValidateState {
     ValidateState& operator=(const ValidateState&) = delete;
 
 public:
-    /**
-     * 'turnOnExtraLoggingForTest' turns on extra logging for test debugging. This parameter is for
-     * unit testing only.
-     */
     ValidateState(OperationContext* opCtx,
                   const NamespaceString& nss,
                   ValidateMode mode,
                   RepairMode repairMode,
-                  bool turnOnExtraLoggingForTest = false);
+                  bool logDiagnostics);
 
     const NamespaceString& nss() const {
         return _nss;
@@ -123,6 +119,10 @@ public:
         return _indexes;
     }
 
+    const StringSet& getSkippedIndexes() const {
+        return _skippedIndexes;
+    }
+
     /**
      * Map of index names to index cursors.
      */
@@ -162,11 +162,9 @@ public:
 
     /**
      * Indicates whether extra logging should occur during validation.
-     *
-     * This is for unit testing only. Intended to improve diagnosibility.
      */
-    bool extraLoggingForTest() {
-        return _extraLoggingForTest;
+    bool logDiagnostics() {
+        return _logDiagnostics;
     }
 
     boost::optional<Timestamp> getValidateTimestamp() {
@@ -235,6 +233,10 @@ private:
     std::unique_ptr<SeekableRecordThrottleCursor> _traverseRecordStoreCursor;
     std::unique_ptr<SeekableRecordThrottleCursor> _seekRecordStoreCursor;
 
+    // Stores the set of indexes that will not be validated for some reason, e.g. they are not
+    // ready.
+    StringSet _skippedIndexes;
+
     RecordId _firstRecordId;
 
     DataThrottle _dataThrottle;
@@ -242,8 +244,8 @@ private:
     // Used to detect when the catalog is re-opened while yielding locks.
     uint64_t _catalogGeneration;
 
-    // Can be set by unit tests to obtain better insight into what validate sees/does.
-    bool _extraLoggingForTest;
+    // Can be set to obtain better insight into what validate sees/does.
+    bool _logDiagnostics;
 
     boost::optional<Timestamp> _validateTs = boost::none;
 };
diff --git a/src/mongo/db/catalog/validate_state_test.cpp b/src/mongo/db/catalog/validate_state_test.cpp
index e8e4bfbbbd9..f6b6fe43cd4 100644
--- a/src/mongo/db/catalog/validate_state_test.cpp
+++ b/src/mongo/db/catalog/validate_state_test.cpp
@@ -157,7 +157,8 @@ TEST_F(ValidateStateTest, NonExistentCollectionShouldThrowNamespaceNotFoundError
         CollectionValidation::ValidateState(opCtx,
                                             kNss,
                                             CollectionValidation::ValidateMode::kForeground,
-                                            CollectionValidation::RepairMode::kNone),
+                                            CollectionValidation::RepairMode::kNone,
+                                            /*logDiagnostics=*/false),
         AssertionException,
         ErrorCodes::NamespaceNotFound);
 
@@ -165,7 +166,8 @@ TEST_F(ValidateStateTest, NonExistentCollectionShouldThrowNamespaceNotFoundError
         CollectionValidation::ValidateState(opCtx,
                                             kNss,
                                             CollectionValidation::ValidateMode::kBackground,
-                                            CollectionValidation::RepairMode::kNone),
+                                            CollectionValidation::RepairMode::kNone,
+                                            /*logDiagnostics=*/false),
         AssertionException,
         ErrorCodes::NamespaceNotFound);
 }
@@ -184,7 +186,8 @@ TEST_F(ValidateStateTest, UncheckpointedCollectionShouldBeAbleToInitializeCursor
         opCtx,
         kNss,
         CollectionValidation::ValidateMode::kBackground,
-        CollectionValidation::RepairMode::kNone);
+        CollectionValidation::RepairMode::kNone,
+        /*logDiagnostics=*/false);
     // Assert that cursors are able to created on the new collection.
     validateState.initializeCursors(opCtx);
     // There should only be a first record id if cursors were initialized successfully.
@@ -211,7 +214,8 @@ TEST_F(ValidateStateTest, OpenCursorsOnAllIndexes) {
             opCtx,
             kNss,
             CollectionValidation::ValidateMode::kForeground,
-            CollectionValidation::RepairMode::kNone);
+            CollectionValidation::RepairMode::kNone,
+            /*logDiagnostics=*/false);
         validateState.initializeCursors(opCtx);
 
         // Make sure all of the indexes were found and cursors opened against them. Including the
@@ -229,7 +233,8 @@ TEST_F(ValidateStateTest, OpenCursorsOnAllIndexes) {
         opCtx,
         kNss,
         CollectionValidation::ValidateMode::kForeground,
-        CollectionValidation::RepairMode::kNone);
+        CollectionValidation::RepairMode::kNone,
+        /*logDiagnostics=*/false);
     validateState.initializeCursors(opCtx);
     ASSERT_EQ(validateState.getIndexes().size(), 5);
 }
@@ -256,7 +261,8 @@ TEST_F(ValidateStateTest, OpenCursorsOnAllIndexesWithBackground) {
         opCtx,
         kNss,
         CollectionValidation::ValidateMode::kBackground,
-        CollectionValidation::RepairMode::kNone);
+        CollectionValidation::RepairMode::kNone,
+        /*logDiagnostics=*/false);
     validateState.initializeCursors(opCtx);
 
     // We should be able to open a cursor on each index.
@@ -293,7 +299,8 @@ TEST_F(ValidateStateTest, CursorsAreNotOpenedAgainstCheckpointedIndexesThatWereL
             opCtx,
             kNss,
             CollectionValidation::ValidateMode::kBackground,
-            CollectionValidation::RepairMode::kNone);
+            CollectionValidation::RepairMode::kNone,
+            /*logDiagnostics=*/false);
         validateState.initializeCursors(opCtx);
         ASSERT_EQ(validateState.getIndexes().size(), 3);
     }
@@ -306,7 +313,8 @@ TEST_F(ValidateStateTest, CursorsAreNotOpenedAgainstCheckpointedIndexesThatWereL
         opCtx,
         kNss,
         CollectionValidation::ValidateMode::kBackground,
-        CollectionValidation::RepairMode::kNone);
+        CollectionValidation::RepairMode::kNone,
+        /*logDiagnostics=*/false);
     validateState.initializeCursors(opCtx);
     ASSERT_EQ(validateState.getIndexes().size(), 3);
 }
diff --git a/src/mongo/db/catalog/views_for_database.cpp b/src/mongo/db/catalog/views_for_database.cpp
index 776cf5e3266..404297c11a3 100644
--- a/src/mongo/db/catalog/views_for_database.cpp
+++ b/src/mongo/db/catalog/views_for_database.cpp
@@ -61,47 +61,8 @@ std::shared_ptr<const ViewDefinition> ViewsForDatabase::lookup(const NamespaceSt
 }
 
 Status ViewsForDatabase::reload(OperationContext* opCtx) {
-    auto reloadCallback = [&](const BSONObj& view) -> Status {
-        BSONObj collationSpec = view.hasField("collation") ? view["collation"].Obj() : BSONObj();
-        auto collator = parseCollator(opCtx, collationSpec);
-        if (!collator.isOK()) {
-            return collator.getStatus();
-        }
-
-        NamespaceString viewName(view["_id"].str());
-
-        auto pipeline = view["pipeline"].Obj();
-        for (auto&& stage : pipeline) {
-            if (BSONType::Object != stage.type()) {
-                return Status(ErrorCodes::InvalidViewDefinition,
-                              str::stream() << "View 'pipeline' entries must be objects, but "
-                                            << viewName.toString()
-                                            << " has a pipeline element of type " << stage.type());
-            }
-        }
-
-        auto viewDef = std::make_shared<ViewDefinition>(viewName.db(),
-                                                        viewName.coll(),
-                                                        view["viewOn"].str(),
-                                                        pipeline,
-                                                        std::move(collator.getValue()));
-
-        if (!viewName.isOnInternalDb() && !viewName.isSystem()) {
-            if (viewDef->timeseries()) {
-                stats.userTimeseries += 1;
-            } else {
-                stats.userViews += 1;
-            }
-        } else {
-            stats.internal += 1;
-        }
-
-        viewMap[viewName.ns()] = std::move(viewDef);
-        return Status::OK();
-    };
-
     try {
-        durable->iterate(opCtx, reloadCallback);
+        durable->iterate(opCtx, [&](const BSONObj& view) { return _insert(opCtx, view); });
     } catch (const DBException& ex) {
         auto status = ex.toStatus();
         LOGV2(22547,
@@ -110,9 +71,60 @@ Status ViewsForDatabase::reload(OperationContext* opCtx) {
               "error"_attr = status);
         return status;
     }
+    valid = true;
+    return Status::OK();
+}
+
 
+Status ViewsForDatabase::insert(OperationContext* opCtx, const BSONObj& view) {
+    auto status = _insert(opCtx, view);
+    if (!status.isOK()) {
+        LOGV2(5387000,
+              "Could not insert view",
+              "db"_attr = durable->getName(),
+              "error"_attr = status);
+        return status;
+    }
     valid = true;
+    return Status::OK();
+};
+
+Status ViewsForDatabase::_insert(OperationContext* opCtx, const BSONObj& view) {
+    BSONObj collationSpec = view.hasField("collation") ? view["collation"].Obj() : BSONObj();
+    auto collator = parseCollator(opCtx, collationSpec);
+    if (!collator.isOK()) {
+        return collator.getStatus();
+    }
+
+    NamespaceString viewName(view["_id"].str());
+
+    auto pipeline = view["pipeline"].Obj();
+    for (auto&& stage : pipeline) {
+        if (BSONType::Object != stage.type()) {
+            return Status(ErrorCodes::InvalidViewDefinition,
+                          str::stream() << "View 'pipeline' entries must be objects, but "
+                                        << viewName.toString() << " has a pipeline element of type "
+                                        << stage.type());
+        }
+    }
+
+    auto viewDef = std::make_shared<ViewDefinition>(viewName.db(),
+                                                    viewName.coll(),
+                                                    view["viewOn"].str(),
+                                                    pipeline,
+                                                    std::move(collator.getValue()));
+
+    if (!viewName.isOnInternalDb() && !viewName.isSystem()) {
+        if (viewDef->timeseries()) {
+            stats.userTimeseries += 1;
+        } else {
+            stats.userViews += 1;
+        }
+    } else {
+        stats.internal += 1;
+    }
 
+    viewMap[viewName.ns()] = std::move(viewDef);
     return Status::OK();
 }
 
@@ -135,7 +147,8 @@ Status ViewsForDatabase::validateCollation(OperationContext* opCtx,
 
 Status ViewsForDatabase::upsertIntoGraph(OperationContext* opCtx,
                                          const ViewDefinition& viewDef,
-                                         const PipelineValidatorFn& validatePipeline) {
+                                         const PipelineValidatorFn& validatePipeline,
+                                         const bool needsValidation) {
     // Performs the insert into the graph.
     auto doInsert = [this, opCtx, &validatePipeline](const ViewDefinition& viewDef,
                                                      bool needsValidation) -> Status {
@@ -190,7 +203,7 @@ Status ViewsForDatabase::upsertIntoGraph(OperationContext* opCtx,
     // is simply a no-op.
     viewGraph.remove(viewDef.name());
 
-    return doInsert(viewDef, true);
+    return doInsert(viewDef, needsValidation);
 }
 
 }  // namespace mongo
diff --git a/src/mongo/db/catalog/views_for_database.h b/src/mongo/db/catalog/views_for_database.h
index 914adf60df7..ab4329bf9d9 100644
--- a/src/mongo/db/catalog/views_for_database.h
+++ b/src/mongo/db/catalog/views_for_database.h
@@ -93,6 +93,11 @@ public:
     Status reload(OperationContext* opCtx);
 
     /**
+     * Inserts the view into the view map.
+     */
+    Status insert(OperationContext* opCtx, const BSONObj& view);
+
+    /**
      * Returns Status::OK if each view namespace in 'refs' has the same default collation as
      * 'view'. Otherwise, returns ErrorCodes::OptionNotSupportedOnView.
      */
@@ -103,11 +108,17 @@ public:
     /**
      * Parses the view definition pipeline, attempts to upsert into the view graph, and
      * refreshes the graph if necessary. Returns an error status if the resulting graph
-     * would be invalid.
+     * would be invalid. needsValidation can be set to false if the view already exists in the
+     * durable view catalog and skips checking that the resulting dependency graph is acyclic and
+     * within the maximum depth.
      */
     Status upsertIntoGraph(OperationContext* opCtx,
                            const ViewDefinition& viewDef,
-                           const PipelineValidatorFn&);
+                           const PipelineValidatorFn&,
+                           bool needsValidation);
+
+private:
+    Status _insert(OperationContext* opCtx, const BSONObj& view);
 };
 
 }  // namespace mongo