SERVER-68268 Add function to insert known catalogId mapping after durable catalog scan

Added a helper function to insert accurate catalogId mappings after scanning the durable catalog after a previous call has returned unknown. This will allow other reads to avoid scanning the durable catalog again in some cases.

3 files changed, 266 insertions, 4 deletions

diff --git a/src/mongo/db/catalog/collection_catalog.cpp b/src/mongo/db/catalog/collection_catalog.cpp
index 6e45e015369..735a4b897b0 100644
--- a/src/mongo/db/catalog/collection_catalog.cpp
+++ b/src/mongo/db/catalog/collection_catalog.cpp
@@ -52,6 +52,13 @@
 
 namespace mongo {
 namespace {
+// Sentinel id for marking a catalogId mapping range as unknown. Must use an invalid RecordId.
+static RecordId kUnknownRangeMarkerId = RecordId::minLong();
+// Maximum number of entries in catalogId mapping when inserting catalogId missing at timestamp.
+// Used to avoid quadratic behavior when inserting entries at the beginning. When threshold is
+// reached we will fall back to more durable catalog scans.
+static constexpr int kMaxCatalogIdMappingLengthForMissingInsert = 1000;
+
 struct LatestCollectionCatalog {
     std::shared_ptr<CollectionCatalog> catalog = std::make_shared<CollectionCatalog>();
 };
@@ -1288,7 +1295,11 @@ CollectionCatalog::CatalogIdLookup CollectionCatalog::lookupCatalogIdByNSS(
         // iterator to get the last entry where the time is less or equal.
         auto catalogId = (--rangeIt)->id;
         if (catalogId) {
-            return {*catalogId, CatalogIdLookup::NamespaceExistence::kExists};
+            if (*catalogId != kUnknownRangeMarkerId) {
+                return {*catalogId, CatalogIdLookup::NamespaceExistence::kExists};
+            } else {
+                return {RecordId{}, CatalogIdLookup::NamespaceExistence::kUnknown};
+            }
         }
         return {RecordId{}, CatalogIdLookup::NamespaceExistence::kNotExists};
     }
@@ -1738,7 +1749,8 @@ void CollectionCatalog::_pushCatalogIdForNSS(const NamespaceString& nss,
         return;
     }
 
-    // Re-write latest entry if timestamp match (multiple changes occured in this transaction)
+    // An entry could exist already if concurrent writes are performed, keep the latest change in
+    // that case.
     if (!ids.empty() && ids.back().ts == *ts) {
         ids.back().id = catalogId;
         return;
@@ -1774,8 +1786,8 @@ void CollectionCatalog::_pushCatalogIdForRename(const NamespaceString& from,
     auto& fromIds = _catalogIds.at(from);
     invariant(!fromIds.empty());
 
-    // Re-write latest entry if timestamp match (multiple changes occured in this transaction),
-    // otherwise push at end
+    // An entry could exist already if concurrent writes are performed, keep the latest change in
+    // that case.
     if (!toIds.empty() && toIds.back().ts == *ts) {
         toIds.back().id = fromIds.back().id;
     } else {
@@ -1795,6 +1807,83 @@ void CollectionCatalog::_pushCatalogIdForRename(const NamespaceString& from,
     }
 }
 
+void CollectionCatalog::_insertCatalogIdForNSSAfterScan(const NamespaceString& nss,
+                                                        boost::optional<RecordId> catalogId,
+                                                        Timestamp ts) {
+    // TODO SERVER-68674: Remove feature flag check.
+    if (!feature_flags::gPointInTimeCatalogLookups.isEnabledAndIgnoreFCV()) {
+        // No-op.
+        return;
+    }
+
+    auto& ids = _catalogIds[nss];
+
+    // Binary search for to the entry with same or larger timestamp
+    auto it =
+        std::lower_bound(ids.begin(), ids.end(), ts, [](const auto& entry, const Timestamp& ts) {
+            return entry.ts < ts;
+        });
+
+    // The logic of what we need to do differs whether we are inserting a valid catalogId or not.
+    if (catalogId) {
+        if (it != ids.end()) {
+            // An entry could exist already if concurrent writes are performed, keep the latest
+            // change in that case.
+            if (it->ts == ts) {
+                it->id = catalogId;
+                return;
+            }
+
+            // If next element has same catalogId, we can adjust its timestamp to cover a longer
+            // range
+            if (it->id == catalogId) {
+                it->ts = ts;
+                _markNamespaceForCatalogIdCleanupIfNeeded(nss, ids);
+                return;
+            }
+        }
+
+        // Otherwise insert new entry at timestamp
+        ids.insert(it, TimestampedCatalogId{catalogId, ts});
+        _markNamespaceForCatalogIdCleanupIfNeeded(nss, ids);
+        return;
+    }
+
+    // Avoid inserting missing mapping when the list has grown past the threshold. Will cause the
+    // system to fall back to scanning the durable catalog.
+    if (ids.size() >= kMaxCatalogIdMappingLengthForMissingInsert) {
+        return;
+    }
+
+    if (it != ids.end() && it->ts == ts) {
+        // An entry could exist already if concurrent writes are performed, keep the latest change
+        // in that case.
+        it->id = boost::none;
+    } else {
+        // Otherwise insert new entry
+        it = ids.insert(it, TimestampedCatalogId{boost::none, ts});
+    }
+
+    // The iterator is positioned on the added/modified element above, reposition it to the next
+    // entry
+    ++it;
+
+    // We don't want to assume that the namespace remains not existing until the next entry, as
+    // there can be times where the namespace actually does exist. To make sure we trigger the
+    // scanning of the durable catalog in this range we will insert a bogus entry using an invalid
+    // RecordId at the next timestamp. This will treat the range forward as unknown.
+    auto nextTs = ts + 1;
+
+    // If the next entry is on the next timestamp already, we can skip adding the bogus entry. If
+    // this function is called for a previously unknown namespace, we may not have any future valid
+    // entries and the iterator would be positioned at and at this point.
+    if (it == ids.end() || it->ts != nextTs) {
+        ids.insert(it, TimestampedCatalogId{kUnknownRangeMarkerId, nextTs});
+    }
+
+    _markNamespaceForCatalogIdCleanupIfNeeded(nss, ids);
+}
+
 void CollectionCatalog::_markNamespaceForCatalogIdCleanupIfNeeded(
     const NamespaceString& nss, const std::vector<TimestampedCatalogId>& ids) {
 
diff --git a/src/mongo/db/catalog/collection_catalog.h b/src/mongo/db/catalog/collection_catalog.h
index d3d3446f1be..0b2561f0996 100644
--- a/src/mongo/db/catalog/collection_catalog.h
+++ b/src/mongo/db/catalog/collection_catalog.h
@@ -736,6 +736,15 @@ private:
                                  const NamespaceString& to,
                                  boost::optional<Timestamp> ts);
 
+    // TODO SERVER-70150: Make private again
+public:
+    // Inserts a catalogId for namespace at given Timestamp. Used after scanning the durable catalog
+    // for a correct mapping at the given timestamp.
+    void _insertCatalogIdForNSSAfterScan(const NamespaceString& nss,
+                                         boost::optional<RecordId> catalogId,
+                                         Timestamp ts);
+
+private:
     // Helper to calculate if a namespace needs to be marked for cleanup for a set of timestamped
     // catalogIds
     void _markNamespaceForCatalogIdCleanupIfNeeded(const NamespaceString& nss,
diff --git a/src/mongo/db/catalog/collection_catalog_test.cpp b/src/mongo/db/catalog/collection_catalog_test.cpp
index 6a554f8fe59..8710d807f73 100644
--- a/src/mongo/db/catalog/collection_catalog_test.cpp
+++ b/src/mongo/db/catalog/collection_catalog_test.cpp
@@ -2114,6 +2114,170 @@ TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingRollback) {
               CollectionCatalog::CatalogIdLookup::NamespaceExistence::kNotExists);
 }
 
+TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingInsert) {
+    RAIIServerParameterControllerForTest featureFlagController(
+        "featureFlagPointInTimeCatalogLookups", true);
+
+    NamespaceString nss("a.b");
+
+    // Create a collection on the namespace
+    createCollection(opCtx.get(), nss, Timestamp(1, 10));
+    dropCollection(opCtx.get(), nss, Timestamp(1, 20));
+    createCollection(opCtx.get(), nss, Timestamp(1, 30));
+
+    auto rid1 = catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 10)).id;
+    auto rid2 = catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 30)).id;
+
+    // Simulate startup where we have a range [oldest, stable] by creating and dropping collections
+    // and then advancing the oldest timestamp and then reading behind it.
+    CollectionCatalog::write(opCtx.get(), [](CollectionCatalog& catalog) {
+        catalog.cleanupForOldestTimestampAdvanced(Timestamp(1, 40));
+    });
+
+    // Confirm that the mappings have been cleaned up
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 15)).result,
+              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kUnknown);
+
+    // TODO SERVER-70150: Use openCollection
+    CollectionCatalog::write(opCtx.get(), [&](CollectionCatalog& catalog) {
+        catalog._insertCatalogIdForNSSAfterScan(nss, rid1, Timestamp(1, 17));
+    });
+
+    // Lookups before the inserted timestamp is still unknown
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 11)).result,
+              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kUnknown);
+
+    // Lookups at or after the inserted timestamp is found, even if they don't match with WT
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 17)).result,
+              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 17)).id, rid1);
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 19)).result,
+              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 19)).id, rid1);
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 25)).result,
+              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 25)).id, rid1);
+    // The entry at Timestamp(1, 30) is unaffected
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 30)).result,
+              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 30)).id, rid2);
+
+    // TODO SERVER-70150: Use openCollection
+    CollectionCatalog::write(opCtx.get(), [&](CollectionCatalog& catalog) {
+        catalog._insertCatalogIdForNSSAfterScan(nss, rid1, Timestamp(1, 12));
+    });
+
+    // We should now have extended the range from Timestamp(1, 17) to Timestamp(1, 12)
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 12)).result,
+              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 12)).id, rid1);
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 16)).result,
+              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 16)).id, rid1);
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 17)).result,
+              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 17)).id, rid1);
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 19)).result,
+              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 19)).id, rid1);
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 25)).result,
+              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 25)).id, rid1);
+    // The entry at Timestamp(1, 30) is unaffected
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 30)).result,
+              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 30)).id, rid2);
+
+    // TODO SERVER-70150: Use openCollection
+    CollectionCatalog::write(opCtx.get(), [&](CollectionCatalog& catalog) {
+        catalog._insertCatalogIdForNSSAfterScan(nss, boost::none, Timestamp(1, 25));
+    });
+
+    // Check the entries, most didn't change
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 17)).result,
+              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 17)).id, rid1);
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 19)).result,
+              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 19)).id, rid1);
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 22)).result,
+              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 22)).id, rid1);
+    // At Timestamp(1, 25) we now return kNotExists
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 25)).result,
+              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kNotExists);
+    // But next timestamp returns unknown
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 26)).result,
+              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kUnknown);
+    // The entry at Timestamp(1, 30) is unaffected
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 30)).result,
+              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 30)).id, rid2);
+
+    // TODO SERVER-70150: Use openCollection
+    CollectionCatalog::write(opCtx.get(), [&](CollectionCatalog& catalog) {
+        catalog._insertCatalogIdForNSSAfterScan(nss, boost::none, Timestamp(1, 26));
+    });
+
+    // We should not have re-written the existing entry at Timestamp(1, 26)
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 17)).result,
+              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 17)).id, rid1);
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 19)).result,
+              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 19)).id, rid1);
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 22)).result,
+              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 22)).id, rid1);
+    // At Timestamp(1, 25) we now return kNotExists
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 25)).result,
+              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kNotExists);
+    // But next timestamp returns unknown
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 26)).result,
+              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kNotExists);
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 27)).result,
+              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kUnknown);
+    // The entry at Timestamp(1, 30) is unaffected
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 30)).result,
+              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kExists);
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 30)).id, rid2);
+
+    // Clean up, check so we are back to the original state
+    CollectionCatalog::write(opCtx.get(), [](CollectionCatalog& catalog) {
+        catalog.cleanupForOldestTimestampAdvanced(Timestamp(1, 41));
+    });
+
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 15)).result,
+              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kUnknown);
+}
+
+TEST_F(CollectionCatalogTimestampTest, CatalogIdMappingInsertUnknown) {
+    RAIIServerParameterControllerForTest featureFlagController(
+        "featureFlagPointInTimeCatalogLookups", true);
+
+    NamespaceString nss("a.b");
+
+    // Simulate startup where we have a range [oldest, stable] by advancing the oldest timestamp and
+    // then reading behind it.
+    CollectionCatalog::write(opCtx.get(), [](CollectionCatalog& catalog) {
+        catalog.cleanupForOldestTimestampAdvanced(Timestamp(1, 40));
+    });
+
+    // Reading before the oldest is unknown
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 15)).result,
+              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kUnknown);
+
+    // Try to instantiate a non existing collection at this timestamp.
+    // TODO SERVER-70150: Use openCollection
+    CollectionCatalog::write(opCtx.get(), [&](CollectionCatalog& catalog) {
+        catalog._insertCatalogIdForNSSAfterScan(nss, boost::none, Timestamp(1, 15));
+    });
+
+    // Lookup should now be not existing
+    ASSERT_EQ(catalog()->lookupCatalogIdByNSS(nss, Timestamp(1, 15)).result,
+              CollectionCatalog::CatalogIdLookup::NamespaceExistence::kNotExists);
+}
+
 TEST_F(CollectionCatalogTimestampTest, CollectionLifetimeTiedToStorageTransactionLifetime) {
     RAIIServerParameterControllerForTest featureFlagController(
         "featureFlagPointInTimeCatalogLookups", true);