fix partition stats on schemeshard loading KIKIMR-15271

ref:828b71a69550cf6e9827e639b49ff7c5845d0688

7 files changed, 52 insertions, 39 deletions

diff --git a/ydb/core/sys_view/common/events.h b/ydb/core/sys_view/common/events.h
index 314511d184..3ee51bd254 100644
--- a/ydb/core/sys_view/common/events.h
+++ b/ydb/core/sys_view/common/events.h
@@ -70,6 +70,8 @@ struct TEvSysView {
         EvGetTopPartitionsRequest,
         EvGetTopPartitionsResponse,
 
+        EvInitPartitionStatsCollector,
+
         EvEnd,
     };
 
@@ -383,6 +385,20 @@ struct TEvSysView {
         NKikimrSysView::TEvGetTopPartitionsResponse,
         EvGetTopPartitionsResponse>
     {};
+
+
+    struct TEvInitPartitionStatsCollector : public TEventLocal<
+        TEvInitPartitionStatsCollector,
+        EvInitPartitionStatsCollector>
+    {
+        TPathId DomainKey;
+        ui64 SysViewProcessorId = 0;
+
+        TEvInitPartitionStatsCollector(TPathId domainKey, ui64 sysViewProcessorId)
+            : DomainKey(domainKey)
+            , SysViewProcessorId(sysViewProcessorId)
+        {}
+    };
 };
 
 } // NSysView
diff --git a/ydb/core/sys_view/partition_stats/partition_stats.cpp b/ydb/core/sys_view/partition_stats/partition_stats.cpp
index ae18049e97..d751ff36a4 100644
--- a/ydb/core/sys_view/partition_stats/partition_stats.cpp
+++ b/ydb/core/sys_view/partition_stats/partition_stats.cpp
@@ -23,27 +23,20 @@ public:
         return NKikimrServices::TActivity::SYSTEM_VIEW_PART_STATS_COLLECTOR;
     }
 
-    explicit TPartitionStatsCollector(TPathId domainKey, ui64 sysViewProcessorId,
-        size_t batchSize, size_t pendingRequestsLimit)
-        : DomainKey(domainKey)
-        , SysViewProcessorId(sysViewProcessorId)
-        , BatchSize(batchSize)
+    explicit TPartitionStatsCollector(size_t batchSize, size_t pendingRequestsLimit)
+        : BatchSize(batchSize)
         , PendingRequestsLimit(pendingRequestsLimit)
     {}
 
     void Bootstrap() {
-        SVLOG_D("NSysView::TPartitionStatsCollector bootstrapped: "
-            << "domain key# " << DomainKey
-            << ", sysview processor id# " << SysViewProcessorId);
+        SVLOG_D("NSysView::TPartitionStatsCollector bootstrapped");
 
         if (AppData()->UsePartitionStatsCollectorForTests) {
             OverloadedPartitionBound = 0.0;
             ProcessOverloadedInterval = TDuration::Seconds(1);
         }
 
-        if (SysViewProcessorId) {
-            Schedule(ProcessOverloadedInterval, new TEvPrivate::TEvProcessOverloaded);
-        }
+        Schedule(ProcessOverloadedInterval, new TEvPrivate::TEvProcessOverloaded);
 
         Become(&TThis::StateWork);
     }
@@ -57,6 +50,7 @@ public:
             hFunc(TEvSysView::TEvGetPartitionStats, Handle);
             hFunc(TEvPrivate::TEvProcess, Handle);
             hFunc(TEvPrivate::TEvProcessOverloaded, Handle);
+            hFunc(TEvSysView::TEvInitPartitionStatsCollector, Handle);
             IgnoreFunc(TEvPipeCache::TEvDeliveryProblem);
             cFunc(TEvents::TEvPoison::EventType, PassAway);
             default:
@@ -343,12 +337,12 @@ private:
     }
 
     void Handle(TEvPrivate::TEvProcessOverloaded::TPtr&) {
+        Schedule(ProcessOverloadedInterval, new TEvPrivate::TEvProcessOverloaded);
+
         if (!SysViewProcessorId) {
             return;
         }
 
-        Schedule(ProcessOverloadedInterval, new TEvPrivate::TEvProcessOverloaded);
-
         auto domainFound = DomainTables.find(DomainKey);
         if (domainFound == DomainTables.end()) {
             SVLOG_D("NSysView::TPartitionStatsCollector: TEvProcessOverloaded: no tables");
@@ -409,6 +403,15 @@ private:
             new TEvPipeCache::TEvForward(sendEvent.Release(), SysViewProcessorId, true));
     }
 
+    void Handle(TEvSysView::TEvInitPartitionStatsCollector::TPtr& ev) {
+        DomainKey = ev->Get()->DomainKey;
+        SysViewProcessorId = ev->Get()->SysViewProcessorId;
+
+        SVLOG_I("NSysView::TPartitionStatsCollector initialized: "
+            << "domain key# " << DomainKey
+            << ", sysview processor id# " << SysViewProcessorId);
+    }
+
     void PassAway() override {
         Send(MakePipePeNodeCacheID(false), new TEvPipeCache::TEvUnlink(0));
         TBase::PassAway();
@@ -419,11 +422,12 @@ private:
     }
 
 private:
-    const TPathId DomainKey;
-    const ui64 SysViewProcessorId;
     const size_t BatchSize;
     const size_t PendingRequestsLimit;
 
+    TPathId DomainKey;
+    ui64 SysViewProcessorId = 0;
+
     double OverloadedPartitionBound = 0.7;
     TDuration ProcessOverloadedInterval = TDuration::Seconds(15);
 
@@ -443,11 +447,9 @@ private:
     bool ProcessInFly = false;
 };
 
-THolder<IActor> CreatePartitionStatsCollector(
-    TPathId domainKey, ui64 sysViewProcessorId, size_t batchSize, size_t pendingRequestsLimit)
+THolder<IActor> CreatePartitionStatsCollector(size_t batchSize, size_t pendingRequestsLimit)
 {
-    return MakeHolder<TPartitionStatsCollector>(
-        domainKey, sysViewProcessorId, batchSize, pendingRequestsLimit);
+    return MakeHolder<TPartitionStatsCollector>(batchSize, pendingRequestsLimit);
 }
 
 
diff --git a/ydb/core/sys_view/partition_stats/partition_stats.h b/ydb/core/sys_view/partition_stats/partition_stats.h
index f1f192b9db..dedf62906f 100644
--- a/ydb/core/sys_view/partition_stats/partition_stats.h
+++ b/ydb/core/sys_view/partition_stats/partition_stats.h
@@ -9,8 +9,6 @@ constexpr size_t STATS_COLLECTOR_BATCH_SIZE = 5000;
 constexpr size_t STATS_COLLECTOR_QUEUE_SIZE_LIMIT = 10;
 
 THolder<IActor> CreatePartitionStatsCollector(
-    TPathId domainKey,
-    ui64 sysViewProcessorId,
     size_t batchSize = STATS_COLLECTOR_BATCH_SIZE,
     size_t pendingRequestsCount = STATS_COLLECTOR_QUEUE_SIZE_LIMIT);
 
diff --git a/ydb/core/sys_view/partition_stats/partition_stats_ut.cpp b/ydb/core/sys_view/partition_stats/partition_stats_ut.cpp
index 17a8e03aa6..83189b339d 100644
--- a/ydb/core/sys_view/partition_stats/partition_stats_ut.cpp
+++ b/ydb/core/sys_view/partition_stats/partition_stats_ut.cpp
@@ -26,7 +26,7 @@ Y_UNIT_TEST_SUITE(PartitionStats) {
         TTestActorRuntime runtime;
         runtime.Initialize(MakeEgg());
 
-        auto collector = CreatePartitionStatsCollector(TPathId(), 0, batchSize);
+        auto collector = CreatePartitionStatsCollector(batchSize);
         auto collectorId = runtime.Register(collector.Release());
         WaitForBootstrap(runtime);
 
@@ -183,7 +183,7 @@ Y_UNIT_TEST_SUITE(PartitionStats) {
         TTestActorRuntime runtime;
         runtime.Initialize(MakeEgg());
 
-        auto collector = CreatePartitionStatsCollector(TPathId(), 0, 1, 0);
+        auto collector = CreatePartitionStatsCollector(1, 0);
         auto collectorId = runtime.Register(collector.Release());
         WaitForBootstrap(runtime);
 
diff --git a/ydb/core/tx/schemeshard/schemeshard__conditional_erase.cpp b/ydb/core/tx/schemeshard/schemeshard__conditional_erase.cpp
index 4d7f18456b..f4f5db4ba7 100644
--- a/ydb/core/tx/schemeshard/schemeshard__conditional_erase.cpp
+++ b/ydb/core/tx/schemeshard/schemeshard__conditional_erase.cpp
@@ -451,7 +451,7 @@ struct TSchemeShard::TTxScheduleConditionalErase : public TTransactionBase<TSche
         LOG_INFO_S(ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, "TTxScheduleConditionalErase Complete"
             << ": at schemeshard: " << Self->TabletID());
 
-        if (StatsCollectorEv && Self->SysPartitionStatsCollector) {
+        if (StatsCollectorEv) {
             ctx.Send(Self->SysPartitionStatsCollector, StatsCollectorEv.Release());
         }
 
diff --git a/ydb/core/tx/schemeshard/schemeshard__table_stats.cpp b/ydb/core/tx/schemeshard/schemeshard__table_stats.cpp
index 5b1d330347..147c4cd820 100644
--- a/ydb/core/tx/schemeshard/schemeshard__table_stats.cpp
+++ b/ydb/core/tx/schemeshard/schemeshard__table_stats.cpp
@@ -23,9 +23,7 @@ static ui64 GetIops(const T& c) {
 }
 
 void TSchemeShard::Handle(NSysView::TEvSysView::TEvGetPartitionStats::TPtr& ev, const TActorContext& ctx) {
-    if (SysPartitionStatsCollector) {
-        ctx.Send(ev->Forward(SysPartitionStatsCollector));
-    }
+    ctx.Send(ev->Forward(SysPartitionStatsCollector));
 }
 
 auto TSchemeShard::BuildStatsForCollector(TPathId pathId, TShardIdx shardIdx, TTabletId datashardId,
@@ -301,11 +299,9 @@ bool TTxStorePartitionStats::PersistSingleStats(TTransactionContext& txc, const
             startTime = rec.GetStartTime();
         }
 
-        if (Self->SysPartitionStatsCollector) {
-            PendingMessages.emplace_back(
-                Self->SysPartitionStatsCollector,
-                Self->BuildStatsForCollector(pathId, shardIdx, datashardId, nodeId, startTime, newStats).Release());
-        }
+        PendingMessages.emplace_back(
+            Self->SysPartitionStatsCollector,
+            Self->BuildStatsForCollector(pathId, shardIdx, datashardId, nodeId, startTime, newStats).Release());
     }
 
     if (isOlapStore) {
diff --git a/ydb/core/tx/schemeshard/schemeshard_impl.cpp b/ydb/core/tx/schemeshard/schemeshard_impl.cpp
index 2891485f1a..cfbcf79398 100644
--- a/ydb/core/tx/schemeshard/schemeshard_impl.cpp
+++ b/ydb/core/tx/schemeshard/schemeshard_impl.cpp
@@ -68,8 +68,9 @@ void TSchemeShard::ActivateAfterInitialization(const TActorContext& ctx,
     domainPtr->UpdateSecurityState(LoginProvider.GetSecurityState());
 
     TTabletId sysViewProcessorId = domainPtr->GetTenantSysViewProcessorID();
-    SysPartitionStatsCollector = Register(NSysView::CreatePartitionStatsCollector(
-        GetDomainKey(subDomainPathId), sysViewProcessorId ? sysViewProcessorId.GetValue() : 0).Release());
+    auto evInit = MakeHolder<NSysView::TEvSysView::TEvInitPartitionStatsCollector>(
+        GetDomainKey(subDomainPathId), sysViewProcessorId ? sysViewProcessorId.GetValue() : 0);
+    Send(SysPartitionStatsCollector, evInit.Release());
 
     Execute(CreateTxInitPopulator(std::move(delayPublications)), ctx);
 
@@ -3313,7 +3314,7 @@ void TSchemeShard::PersistRemoveTable(NIceDb::TNiceDb& db, TPathId pathId, const
     Tables.erase(pathId);
     DecrementPathDbRefCount(pathId, "remove table");
 
-    if (AppData()->FeatureFlags.GetEnableSystemViews() && SysPartitionStatsCollector) {
+    if (AppData()->FeatureFlags.GetEnableSystemViews()) {
         auto ev = MakeHolder<NSysView::TEvSysView::TEvRemoveTable>(GetDomainKey(pathId), pathId);
         Send(SysPartitionStatsCollector, ev.Release());
     }
@@ -3860,6 +3861,8 @@ void TSchemeShard::OnActivateExecutor(const TActorContext &ctx) {
 
     TxAllocatorClient = RegisterWithSameMailbox(CreateTxAllocatorClient(CollectTxAllocators(appData)));
 
+    SysPartitionStatsCollector = Register(NSysView::CreatePartitionStatsCollector().Release());
+
     SplitSettings.Register(appData->Icb);
 
     Executor()->RegisterExternalTabletCounters(TabletCountersPtr);
@@ -5882,7 +5885,7 @@ bool TSchemeShard::FillUniformPartitioning(TVector<TString>& rangeEnds, ui32 key
 void TSchemeShard::SetPartitioning(TPathId pathId, TOlapStoreInfo::TPtr storeInfo) {
     const TVector<TShardIdx>& partitioning = storeInfo->ColumnShards;
 
-    if (AppData()->FeatureFlags.GetEnableSystemViews() && SysPartitionStatsCollector) {
+    if (AppData()->FeatureFlags.GetEnableSystemViews()) {
         TVector<std::pair<ui64, ui64>> shardIndices;
         shardIndices.reserve(partitioning.size());
         for (auto& shardIdx : partitioning) {
@@ -5908,9 +5911,7 @@ void TSchemeShard::SetPartitioning(TPathId pathId, TTableInfo::TPtr tableInfo, T
         auto path = TPath::Init(pathId, this);
         auto ev = MakeHolder<NSysView::TEvSysView::TEvSetPartitioning>(GetDomainKey(pathId), pathId, path.PathString());
         ev->ShardIndices.swap(shardIndices);
-        if (SysPartitionStatsCollector) {
-            Send(SysPartitionStatsCollector, ev.Release());
-        }
+        Send(SysPartitionStatsCollector, ev.Release());
     }
 
     if (!tableInfo->IsBackup) {