25-3-1: schemeshard: stats processing opt 1 (#28226)

Cherry-pick from `main`:
- 626410b, #27165

Streamline and optimize datashard statistics processing.

Profile guided optimizations of PersistSingleStats() (in synthetic test). Total gain is around 30%.

- remove unreasonable iteration over entire ShardInfos
- single Now() timestamp for entire stats batch
- optimize number of lookups
- stop building now unnecessary storage pool kind mappings
- remove table/store aggregated stats copying
- collect ExternalBlobsEnabled only on PartitionConfig change
- replace ETxType->CounterId map with absl::flat_hash_map
- remove extra OpType->TxType lookup
- remove call to GetMainTableForIndex for not-index-table shards

Author: ijon

ydb/core/tx/schemeshard/olap/store/store.h

@@ -147,11 +147,11 @@ struct TOlapStoreInfo {
 
     ILayoutPolicy::TPtr GetTablesLayoutPolicy() const;
 
-    void UpdateShardStats(TShardIdx shardIdx, const TPartitionStats& newStats) {
+    void UpdateShardStats(TDiskSpaceUsageDelta* diskSpaceUsageDelta, TShardIdx shardIdx, const TPartitionStats& newStats, TInstant now) {
         Stats.Aggregated.PartCount = ColumnShards.size();
         Stats.PartitionStats[shardIdx]; // insert if none
-        Stats.UpdateShardStats(shardIdx, newStats);
+        Stats.UpdateShardStats(diskSpaceUsageDelta, shardIdx, newStats, now);
     }
 };
 
-}
+}

ydb/core/tx/schemeshard/olap/table/table.h

@@ -103,15 +103,15 @@ struct TColumnTableInfo {
         return Stats;
     }
 
-    void UpdateShardStats(const TShardIdx shardIdx, const TPartitionStats& newStats) {
+    void UpdateShardStats(TDiskSpaceUsageDelta* diskSpaceUsageDelta, const TShardIdx shardIdx, const TPartitionStats& newStats, TInstant now) {
         Stats.Aggregated.PartCount = GetColumnShards().size();
         Stats.PartitionStats[shardIdx]; // insert if none
-        Stats.UpdateShardStats(shardIdx, newStats);
+        Stats.UpdateShardStats(diskSpaceUsageDelta, shardIdx, newStats, now);
     }
 
-    void UpdateTableStats(const TShardIdx shardIdx, const TPathId& pathId, const TPartitionStats& newStats) {
+    void UpdateTableStats(TDiskSpaceUsageDelta* diskSpaceUsageDelta, const TShardIdx shardIdx, const TPathId& pathId, const TPartitionStats& newStats, TInstant now) {
         Stats.TableStats[pathId].Aggregated.PartCount = GetColumnShards().size();
-        Stats.UpdateTableStats(shardIdx, pathId, newStats);
+        Stats.UpdateTableStats(diskSpaceUsageDelta, shardIdx, pathId, newStats, now);
     }
 
     TConclusion<std::shared_ptr<NOlap::NAlter::ISSEntity>> BuildEntity(const TPathId& pathId, const NOlap::NAlter::TEntityInitializationContext& iContext) const;

ydb/core/tx/schemeshard/schemeshard__init.cpp

@@ -1845,6 +1845,8 @@ struct TSchemeShard::TTxInit : public TTransactionBase<TSchemeShard> {
                     }
                 }
 
+                tableInfo->IsExternalBlobsEnabled = PartitionConfigHasExternalBlobsEnabled(tableInfo->PartitionConfig());
+
                 TString alterTabletFull = std::get<4>(rec);
                 TString alterTabletDiff = std::get<5>(rec);
                 if (alterTabletFull) {
@@ -2384,6 +2386,7 @@ struct TSchemeShard::TTxInit : public TTransactionBase<TSchemeShard> {
 
             TPathId prevTableId;
 
+            TInstant now = AppData()->TimeProvider->Now();
             while (!rowSet.EndOfSet()) {
                 const TPathId tableId = TPathId(
                     rowSet.GetValue<Schema::TablePartitionStats::TableOwnerId>(),
@@ -2460,7 +2463,6 @@ struct TSchemeShard::TTxInit : public TTransactionBase<TSchemeShard> {
                 stats.RangeReads = rowSet.GetValue<Schema::TablePartitionStats::RangeReads>();
                 stats.RangeReadRows = rowSet.GetValue<Schema::TablePartitionStats::RangeReadRows>();
 
-                TInstant now = AppData(ctx)->TimeProvider->Now();
                 stats.SetCurrentRawCpuUsage(rowSet.GetValue<Schema::TablePartitionStats::CPU>(), now);
                 stats.Memory = rowSet.GetValue<Schema::TablePartitionStats::Memory>();
                 stats.Network = rowSet.GetValue<Schema::TablePartitionStats::Network>();
@@ -2478,7 +2480,8 @@ struct TSchemeShard::TTxInit : public TTransactionBase<TSchemeShard> {
                 stats.LocksWholeShard = rowSet.GetValueOrDefault<Schema::TablePartitionStats::LocksWholeShard>();
                 stats.LocksBroken = rowSet.GetValueOrDefault<Schema::TablePartitionStats::LocksBroken>();
 
-                tableInfo->UpdateShardStats(shardIdx, stats);
+                TDiskSpaceUsageDelta unusedDelta;
+                tableInfo->UpdateShardStats(&unusedDelta, shardIdx, stats, now);
 
                 // note that we don't update shard metrics here, because we will always update
                 // the shard metrics in TSchemeShard::SetPartitioning

ydb/core/tx/schemeshard/schemeshard__pq_stats.cpp

@@ -22,12 +22,12 @@ class TTxStoreTopicStats: public TTxStoreStats<TEvPersQueue::TEvPeriodicTopicSta
     void Complete(const TActorContext& ) override;
 
     // returns true to continue batching
-    bool PersistSingleStats(const TPathId& pathId, const TStatsQueue<TEvPersQueue::TEvPeriodicTopicStats>::TItem& item, TTransactionContext& txc, const TActorContext& ctx) override;
+    bool PersistSingleStats(const TPathId& pathId, const TStatsQueue<TEvPersQueue::TEvPeriodicTopicStats>::TItem& item, TInstant now, TTransactionContext& txc, const TActorContext& ctx) override;
     void ScheduleNextBatch(const TActorContext& ctx) override;
 };
 
 
-bool TTxStoreTopicStats::PersistSingleStats(const TPathId& pathId, const TStatsQueueItem<TEvPersQueue::TEvPeriodicTopicStats>& item, TTransactionContext& txc, const TActorContext& ctx) {
+bool TTxStoreTopicStats::PersistSingleStats(const TPathId& pathId, const TStatsQueueItem<TEvPersQueue::TEvPeriodicTopicStats>& item, TInstant /*now*/, TTransactionContext& txc, const TActorContext& ctx) {
     const auto& rec = item.Ev->Get()->Record;
 
     TTopicStats newStats;

ydb/core/tx/schemeshard/schemeshard__stats.h

@@ -124,7 +124,7 @@ class TTxStoreStats: public NTabletFlatExecutor::TTransactionBase<TSchemeShard>
     bool Execute(TTransactionContext& txc, const TActorContext& ctx) override;
 
     // returns true to continue batching
-    virtual bool PersistSingleStats(const TPathId& pathId, const TItem& item, TTransactionContext& txc, const TActorContext& ctx) = 0;
+    virtual bool PersistSingleStats(const TPathId& pathId, const TItem& item, TInstant now, TTransactionContext& txc, const TActorContext& ctx) = 0;
 
     virtual void ScheduleNextBatch(const TActorContext& ctx) = 0;
 };

ydb/core/tx/schemeshard/schemeshard__stats_impl.h

@@ -118,13 +118,14 @@ bool TTxStoreStats<TEvent>::Execute(NTabletFlatExecutor::TTransactionContext& tx
         return true;
     }
 
+    TInstant now = AppData(ctx)->TimeProvider->Now();
     TMonotonic start = TMonotonic::Now();
     const ui32 maxBatchSize = Queue.MaxBatchSize();
     ui32 batchSize = 0;
     for (; batchSize < maxBatchSize && !Queue.Empty(); ++batchSize) {
         auto item = Queue.Next();
         Queue.WriteLatencyMetric(item);
-        if (!PersistSingleStats(item.PathId(), item, txc, ctx)) {
+        if (!PersistSingleStats(item.PathId(), item, now, txc, ctx)) {
             break;
         }