do not check explicit sources every cleanup (#3674)

Author: alexnick88

ydb/core/persqueue/sourceid.cpp

@@ -253,7 +253,7 @@ void TSourceIdStorage::DeregisterSourceId(const TString& sourceId) {
         ExplicitSourceIds.erase(sourceId);
     }
 
-    SourceIdsByOffset.erase(std::make_pair(it->second.Offset, sourceId));
+    SourceIdsByOffset[it->second.Explicit].erase(std::make_pair(it->second.Offset, sourceId));
     InMemorySourceIds.erase(it);
 
     auto jt = SourceIdOwners.find(sourceId);
@@ -277,7 +277,7 @@ bool TSourceIdStorage::DropOldSourceIds(TEvKeyValue::TEvRequest* request, TInsta
     const auto ttl = TDuration::Seconds(config.GetSourceIdLifetimeSeconds());
     ui32 size = request->Record.ByteSize();
 
-    for (const auto& [offset, sourceId] : SourceIdsByOffset) {
+    for (const auto& [offset, sourceId] : SourceIdsByOffset[0]) {
         if (offset >= startOffset && toDelOffsets.size() >= maxDeleteSourceIds) {
             break;
         }
@@ -323,7 +323,7 @@ bool TSourceIdStorage::DropOldSourceIds(TEvKeyValue::TEvRequest* request, TInsta
         size_t res = InMemorySourceIds.erase(t.second);
         Y_ABORT_UNLESS(res == 1);
         // delete sourceID from offsets
-        res = SourceIdsByOffset.erase(t);
+        res = SourceIdsByOffset[0].erase(t);
         Y_ABORT_UNLESS(res == 1);
         // save owners to drop and delete records from map
         auto it = SourceIdOwners.find(t.second);
@@ -372,14 +372,14 @@ void TSourceIdStorage::RegisterSourceIdInfo(const TString& sourceId, TSourceIdIn
     auto it = InMemorySourceIds.find(sourceId);
     if (it != InMemorySourceIds.end()) {
         if (!load || it->second.Offset < sourceIdInfo.Offset) {
-            const auto res = SourceIdsByOffset.erase(std::make_pair(it->second.Offset, sourceId));
+            const auto res = SourceIdsByOffset[sourceIdInfo.Explicit].erase(std::make_pair(it->second.Offset, sourceId));
             Y_ABORT_UNLESS(res == 1);
         } else {
             return;
         }
     }
 
-    const bool res = SourceIdsByOffset.emplace(sourceIdInfo.Offset, sourceId).second;
+    const bool res = SourceIdsByOffset[sourceIdInfo.Explicit].emplace(sourceIdInfo.Offset, sourceId).second;
     Y_ABORT_UNLESS(res);
 
     if (sourceIdInfo.Explicit) {
@@ -421,10 +421,12 @@ void TSourceIdStorage::MarkOwnersForDeletedSourceId(THashMap<TString, TOwnerInfo
 
 TInstant TSourceIdStorage::MinAvailableTimestamp(TInstant now) const {
     TInstant ds = now;
-    if (!SourceIdsByOffset.empty()) {
-        auto it = InMemorySourceIds.find(SourceIdsByOffset.begin()->second);
-        Y_ABORT_UNLESS(it != InMemorySourceIds.end());
-        ds = Min(ds, it->second.WriteTimestamp);
+    for (ui32 i = 0 ; i < 2; ++i) {
+        if (!SourceIdsByOffset[i].empty()) {
+            auto it = InMemorySourceIds.find(SourceIdsByOffset[i].begin()->second);
+            Y_ABORT_UNLESS(it != InMemorySourceIds.end());
+            ds = Min(ds, it->second.WriteTimestamp);
+        }
     }
 
     return ds;

ydb/core/persqueue/sourceid.h

@@ -108,7 +108,7 @@ class TSourceIdStorage: private THeartbeatProcessor {
     TSourceIdMap InMemorySourceIds;
     THashMap<TString, TString> SourceIdOwners;
     TVector<TString> OwnersToDrop;
-    TSet<std::pair<ui64, TString>> SourceIdsByOffset;
+    TSet<std::pair<ui64, TString>> SourceIdsByOffset[2];
     // used to track heartbeats
     THashSet<TString> ExplicitSourceIds;
 

ydb/core/persqueue/ut/sourceid_ut.cpp

@@ -460,6 +460,26 @@ Y_UNIT_TEST_SUITE(TSourceIdTests) {
         }
     }
 
+    Y_UNIT_TEST(ExpensiveCleanup) {
+        TSourceIdStorage storage;
+        ui64 offset = 0;
+
+        // initial info w/o heartbeats
+        for (ui32 i = 1; i <= 100000; ++i) {
+            storage.RegisterSourceId(TestSourceId(i), MakeExplicitSourceIdInfo(++offset));
+        }
+
+        NKikimrPQ::TPartitionConfig config;
+        config.SetSourceIdLifetimeSeconds(TDuration::Hours(1).Seconds());
+
+        auto request = MakeHolder<TEvKeyValue::TEvRequest>();
+        for (ui32 i = 0; i < 1000; ++i) {
+            Cerr << "Iteration " << i << "\n";
+            const auto dropped = storage.DropOldSourceIds(request.Get(), TInstant::Hours(2), 1'000'000, TPartitionId(TestPartition), config);
+            UNIT_ASSERT_EQUAL(dropped, false);
+        }
+
+    }
 } // TSourceIdTests
 
 } // namespace NKikimr::NPQ