Merge 06e5119 into 4a1f83b

ydb/core/tx/columnshard/columnshard.cpp

@@ -327,14 +327,9 @@ void TColumnShard::FillOlapStats(
         resourceMetrics->Fill(*ev->Record.MutableTabletMetrics());
     }
 
-    TTableStatsBuilder statsBuilder(*ev->Record.MutableTableStats());
-    statsBuilder.FillColumnTableStats(*Counters.GetColumnTablesCounters());
-    statsBuilder.FillTabletStats(*Counters.GetTabletCounters());
-    statsBuilder.FillBackgroundControllerStats(*Counters.GetBackgroundControllerCounters());
-    statsBuilder.FillScanCountersStats(Counters.GetScanCounters());
-    statsBuilder.FillExecutorStats(*Executor());
     if (TablesManager.HasPrimaryIndex()) {
-        statsBuilder.FillColumnEngineStats(TablesManager.MutablePrimaryIndex().GetTotalStats());
+        TTableStatsBuilder statsBuilder(Counters, Executor(), TablesManager.MutablePrimaryIndex());
+        statsBuilder.FillTotalTableStats(*ev->Record.MutableTableStats());
     }
 }
 
@@ -343,6 +338,7 @@ void TColumnShard::FillColumnTableStats(
     std::unique_ptr<TEvDataShard::TEvPeriodicTableStats>& ev
 ) {
     auto tables = TablesManager.GetTables();
+    TTableStatsBuilder tableStatsBuilder(Counters, Executor(), TablesManager.MutablePrimaryIndex());
 
     LOG_S_DEBUG("There are stats for " << tables.size() << " tables");
     for (const auto& [pathId, _] : tables) {
@@ -360,18 +356,7 @@ void TColumnShard::FillColumnTableStats(
             resourceMetrics->Fill(*periodicTableStats->MutableTabletMetrics());
         }
 
-        TTableStatsBuilder statsBuilder(*periodicTableStats->MutableTableStats());
-        statsBuilder.FillColumnTableStats(*Counters.GetColumnTablesCounters()->GetPathIdCounter(pathId));
-        statsBuilder.FillTabletStats(*Counters.GetTabletCounters());
-        statsBuilder.FillBackgroundControllerStats(*Counters.GetBackgroundControllerCounters(), pathId);
-        statsBuilder.FillScanCountersStats(Counters.GetScanCounters());
-        statsBuilder.FillExecutorStats(*Executor());
-        if (TablesManager.HasPrimaryIndex()) {
-            auto columnEngineStats = TablesManager.GetPrimaryIndexSafe().GetStats().FindPtr(pathId);
-            if (columnEngineStats && *columnEngineStats) {
-                statsBuilder.FillColumnEngineStats(**columnEngineStats);
-            }
-        }
+        tableStatsBuilder.FillTableStats(pathId, *(periodicTableStats->MutableTableStats()));
 
         LOG_S_TRACE("Add stats for table, tableLocalID=" << pathId);
     }

ydb/core/tx/columnshard/columnshard__scan.cpp

@@ -30,7 +30,7 @@ void TColumnShard::Handle(TEvColumnShard::TEvScan::TPtr& ev, const TActorContext
         return;
     }
 
-    Counters.GetColumnTablesCounters()->GetPathIdCounter(record.GetLocalPathId())->OnAccess();
+    Counters.GetColumnTablesCounters()->GetPathIdCounter(record.GetLocalPathId())->OnReadEvent();
     ScanTxInFlight.insert({txId, TAppData::TimeProvider->Now()});
     Counters.GetTabletCounters()->SetCounter(COUNTER_SCAN_IN_FLY, ScanTxInFlight.size());
     Execute(new NOlap::NReader::TTxScan(this, ev), ctx);

ydb/core/tx/columnshard/columnshard__write.cpp

@@ -84,7 +84,7 @@ void TColumnShard::Handle(TEvPrivate::TEvWriteBlobsResult::TPtr& ev, const TActo
     auto baseAggregations = wBuffer.GetAggregations();
     wBuffer.InitReplyReceived(TMonotonic::Now());
 
-    auto wg = Counters.GetWritesMonitor()->OnFinishWrite(wBuffer.GetSumSize(), wBuffer.GetAggregations().size());
+    Counters.GetWritesMonitor()->OnFinishWrite(wBuffer.GetSumSize(), wBuffer.GetAggregations().size());
 
     for (auto&& aggr : baseAggregations) {
         const auto& writeMeta = aggr->GetWriteMeta();
@@ -156,7 +156,7 @@ void TColumnShard::Handle(TEvColumnShard::TEvWrite::TPtr& ev, const TActorContex
     const TString dedupId = record.GetDedupId();
     const auto source = ev->Sender;
 
-    Counters.GetColumnTablesCounters()->GetPathIdCounter(tableId)->OnUpdate();
+    Counters.GetColumnTablesCounters()->GetPathIdCounter(tableId)->OnWriteEvent();
 
     std::optional<ui32> granuleShardingVersion;
     if (record.HasGranuleShardingVersion()) {
@@ -375,7 +375,7 @@ void TColumnShard::Handle(NEvents::TDataEvents::TEvWrite::TPtr& ev, const TActor
         return;
     }
 
-    auto wg = Counters.GetWritesMonitor()->OnStartWrite(arrowData->GetSize());
+    Counters.GetWritesMonitor()->OnStartWrite(arrowData->GetSize());
 
     std::optional<ui32> granuleShardingVersionId;
     if (record.HasGranuleShardingVersionId()) {

ydb/core/tx/columnshard/counters/aggregation/table_stats.h

@@ -1,37 +1,49 @@
 #pragma once
 
-#include <ydb/core/tx/columnshard/counters/scan.h>
+#include <ydb/core/protos/table_stats.pb.h>
 #include <ydb/core/tablet_flat/tablet_flat_executor.h>
-#include <ydb/core/tx/columnshard/counters/column_tables.h>
-#include <ydb/core/tx/columnshard/counters/tablet_counters.h>
-#include <ydb/core/tx/columnshard/counters/background_controller.h>
+#include <ydb/core/tx/columnshard/counters/counters_manager.h>
 #include <ydb/core/tx/columnshard/engines/column_engine.h>
-#include <ydb/core/protos/table_stats.pb.h>
 
 namespace NKikimr::NColumnShard {
 
 class TTableStatsBuilder {
 private:
-    ::NKikimrTableStats::TTableStats& TableStats;
+    TCountersManager& Counters;
+    const NTabletFlatExecutor::NFlatExecutorSetup::IExecutor& Executor;
+    NOlap::IColumnEngine& ColumnEngine;
 
 public:
-    TTableStatsBuilder(::NKikimrTableStats::TTableStats& tableStats)
-        : TableStats(tableStats) {
+    TTableStatsBuilder(
+        TCountersManager& counters, const NTabletFlatExecutor::NFlatExecutorSetup::IExecutor* executor, NOlap::IColumnEngine& columnEngine)
+        : Counters(counters)
+        , Executor(*executor)
+        , ColumnEngine(columnEngine) {
     }
 
-    void FillColumnTableStats(const TSingleColumnTableCounters& stats);
-    void FillColumnTableStats(const TColumnTablesCounters& stats);
-
-    void FillTabletStats(const TTabletCountersHandle& stats);
+    void FillTableStats(ui64 pathId, ::NKikimrTableStats::TTableStats& tableStats) {
+        Counters.FillTableStats(pathId, tableStats);
 
-    void FillBackgroundControllerStats(const TBackgroundControllerCounters& stats, ui64 pathId);
-    void FillBackgroundControllerStats(const TBackgroundControllerCounters& stats);
+        auto columnEngineStats = ColumnEngine.GetStats().FindPtr(pathId);
+        if (columnEngineStats && *columnEngineStats) {
+            auto activeStats = (*columnEngineStats)->Active();
+            tableStats.SetRowCount(activeStats.Rows);
+            tableStats.SetDataSize(activeStats.Bytes);
+            tableStats.SetPartCount(activeStats.Portions);
+        }
+    }
 
-    void FillScanCountersStats(const TScanCounters& stats);
+    void FillTotalTableStats(::NKikimrTableStats::TTableStats& tableStats) {
+        Counters.FillTotalTableStats(tableStats);
 
-    void FillExecutorStats(const NTabletFlatExecutor::NFlatExecutorSetup::IExecutor& executor);
+        tableStats.SetInFlightTxCount(Executor.GetStats().TxInFly);
+        tableStats.SetHasLoanedParts(Executor.HasLoanedParts());
 
-    void FillColumnEngineStats(const NOlap::TColumnEngineStats& stats);
+        auto activeStats = ColumnEngine.GetTotalStats().Active();
+        tableStats.SetRowCount(activeStats.Rows);
+        tableStats.SetDataSize(activeStats.Bytes);
+        tableStats.SetPartCount(activeStats.Portions);
+    }
 };
 
 } // namespace NKikimr::NColumnShard

ydb/core/tx/columnshard/counters/aggregation/ya.make

@@ -1,8 +1,6 @@
 LIBRARY()
 
-SRCS(
-    table_stats.cpp
-)
+SRCS()
 
 PEERDIR(
     ydb/core/protos

ydb/core/tx/columnshard/counters/column_tables.h

@@ -47,11 +47,11 @@ class TSingleColumnTableCounters {
         , TotalLastUpdateTime(owner.LastUpdateTime) {
     }
 
-    void OnAccess() {
+    void OnReadEvent() {
         UpdateLastAccessTime(TAppData::TimeProvider->Now());
     }
 
-    void OnUpdate() {
+    void OnWriteEvent() {
         TInstant now = TAppData::TimeProvider->Now();
         UpdateLastUpdateTime(now);
         UpdateLastAccessTime(now);

ydb/core/tx/columnshard/counters/columnshard.cpp

@@ -9,7 +9,8 @@ namespace NKikimr::NColumnShard {
 TCSCounters::TCSCounters(std::shared_ptr<const TTabletCountersHandle> tabletCounters)
     : TBase("CS")
     , TabletCounters(std::move(tabletCounters))
-    , Initialization(*this) {
+    , Initialization(*this)
+    , TxProgress(*this) {
     Y_ABORT_UNLESS(TabletCounters);
 
     StartBackgroundCount = TBase::GetDeriviative("StartBackground/Count");

ydb/core/tx/columnshard/counters/columnshard.h

@@ -1,10 +1,10 @@
 #pragma once
 #include "common/owner.h"
 
-#include <library/cpp/monlib/dynamic_counters/counters.h>
+#include <ydb/core/tx/columnshard/counters/tablet_counters.h>
 
+#include <library/cpp/monlib/dynamic_counters/counters.h>
 #include <util/generic/hash_set.h>
-#include <ydb/core/tx/columnshard/counters/tablet_counters.h>
 
 namespace NKikimr::NColumnShard {
 
@@ -30,7 +30,6 @@ class TCSInitialization: public TCommonCountersOwner {
     const NMonitoring::THistogramPtr HistogramSwitchToWorkFromCreateDurationMs;
 
 public:
-
     void OnTxInitFinished(const TDuration d) const {
         HistogramTxInitDurationMs->Collect(d.MilliSeconds());
     }
@@ -66,6 +65,83 @@ class TCSInitialization: public TCommonCountersOwner {
     }
 };
 
+class TTxProgressCounters: public TCommonCountersOwner {
+private:
+    using TBase = TCommonCountersOwner;
+    using TOpType = TString;
+
+    class TProgressCounters: public TCommonCountersOwner {
+    private:
+        using TBase = TCommonCountersOwner;
+
+    public:
+        NMonitoring::TDynamicCounters::TCounterPtr RegisterTx;
+        NMonitoring::TDynamicCounters::TCounterPtr RegisterTxWithDeadline;
+        NMonitoring::TDynamicCounters::TCounterPtr StartProposeOnExecute;
+        NMonitoring::TDynamicCounters::TCounterPtr StartProposeOnComplete;
+        NMonitoring::TDynamicCounters::TCounterPtr FinishProposeOnExecute;
+        NMonitoring::TDynamicCounters::TCounterPtr FinishProposeOnComplete;
+        NMonitoring::TDynamicCounters::TCounterPtr FinishPlannedTx;
+        NMonitoring::TDynamicCounters::TCounterPtr AbortTx;
+
+        TProgressCounters(TBase& owner)
+            : TBase(owner)
+            , RegisterTx(owner.GetDeriviative("RegisterTx"))
+            , RegisterTxWithDeadline(owner.GetDeriviative("RegisterTxWithDeadline"))
+            , StartProposeOnExecute(owner.GetDeriviative("StartProposeOnExecute"))
+            , StartProposeOnComplete(owner.GetDeriviative("StartProposeOnComplete"))
+            , FinishProposeOnExecute(owner.GetDeriviative("FinishProposeOnExecute"))
+            , FinishProposeOnComplete(owner.GetDeriviative("FinishProposeOnComplete"))
+            , FinishPlannedTx(owner.GetDeriviative("FinishPlannedTx"))
+            , AbortTx(owner.GetDeriviative("AbortTx")) {
+        }
+    };
+
+    THashMap<TOpType, TProgressCounters> SubGroups;
+
+public:
+    void OnRegisterTx(const TOpType& opType) {
+        GetSubGroup(opType).RegisterTx->Add(1);
+    }
+
+    void OnRegisterTxWithDeadline(const TOpType& opType) {
+        GetSubGroup(opType).RegisterTxWithDeadline->Add(1);
+    }
+
+    void OnStartProposeOnExecute(const TOpType& opType) {
+        GetSubGroup(opType).StartProposeOnExecute->Add(1);
+    }
+
+    void OnStartProposeOnComplete(const TOpType& opType) {
+        GetSubGroup(opType).StartProposeOnComplete->Add(1);
+    }
+
+    void OnFinishProposeOnExecute(const TOpType& opType) {
+        GetSubGroup(opType).FinishProposeOnExecute->Add(1);
+    }
+
+    void OnFinishProposeOnComplete(const TOpType& opType) {
+        GetSubGroup(opType).FinishProposeOnComplete->Add(1);
+    }
+
+    void OnFinishPlannedTx(const TOpType& opType) {
+        GetSubGroup(opType).FinishPlannedTx->Add(1);
+    }
+
+    void OnAbortTx(const TOpType& opType) {
+        GetSubGroup(opType).AbortTx->Add(1);
+    }
+
+    TTxProgressCounters(TCommonCountersOwner& owner)
+        : TBase(owner, "txProgress") {  // TODO: fix parameter name?
+    }
+
+private:
+    TProgressCounters& GetSubGroup(const TOpType& opType) {
+        return SubGroups.try_emplace(opType, *this).first->second;
+    }
+};
+
 class TCSCounters: public TCommonCountersOwner {
 private:
     using TBase = TCommonCountersOwner;
@@ -122,6 +198,7 @@ class TCSCounters: public TCommonCountersOwner {
 
 public:
     const TCSInitialization Initialization;
+    TTxProgressCounters TxProgress;
 
     void OnStartWriteRequest() const {
         WriteRequests->Add(1);

ydb/core/tx/columnshard/counters/counters_manager.h

@@ -42,6 +42,18 @@ class TCountersManager {
         , ColumnTablesCounters(std::make_shared<TColumnTablesCounters>())
         , SubscribeCounters(std::make_shared<NOlap::NResourceBroker::NSubscribe::TSubscriberCounters>()) {
     }
+
+    void FillTableStats(ui64 pathId, ::NKikimrTableStats::TTableStats& tableStats) {
+        ColumnTablesCounters->GetPathIdCounter(pathId)->FillStats(tableStats);
+        BackgroundControllerCounters->FillStats(pathId, tableStats);
+    }
+
+    void FillTotalTableStats(::NKikimrTableStats::TTableStats& tableStats) {
+        ColumnTablesCounters->FillStats(tableStats);
+        TabletCounters->FillStats(tableStats);
+        BackgroundControllerCounters->FillTotalStats(tableStats);
+        ScanCounters.FillStats(tableStats);
+    }
 };
 
 } // namespace NKikimr::NColumnShard