Skip to content

Commit f97d523

Browse files
Hor911Hor911
authored
CS statistics (Rows + Bytes) (#14549)
1 parent d0a16a7 commit f97d523

File tree

8 files changed

+629
-168
lines changed

8 files changed

+629
-168
lines changed

‎ydb/core/kqp/compute_actor/kqp_scan_compute_actor.cpp

Lines changed: 28 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -86,12 +86,36 @@ void TKqpScanComputeActor::FillExtraStats(NDqProto::TDqComputeActorStats* dst, b
8686
auto* taskStats = dst->MutableTasks(0);
8787
auto* tableStats = taskStats->AddTables();
8888

89+
auto& sourceStats = *taskStats->AddSources();
90+
91+
// sourceStats.SetInputIndex(0); // do not have real input index
92+
sourceStats.SetIngressName("CS");
93+
94+
auto& ingressStats = *sourceStats.MutableIngress();
95+
8996
tableStats->SetTablePath(ScanData->TablePath);
9097

91-
if (auto* x = ScanData->BasicStats.get()) {
92-
tableStats->SetReadRows(x->Rows);
93-
tableStats->SetReadBytes(x->Bytes);
94-
tableStats->SetAffectedPartitions(x->AffectedShards);
98+
if (auto* stats = ScanData->BasicStats.get()) {
99+
ingressStats.SetRows(stats->Rows);
100+
ingressStats.SetBytes(stats->Bytes);
101+
ingressStats.SetFirstMessageMs(stats->FirstMessageMs);
102+
ingressStats.SetLastMessageMs(stats->LastMessageMs);
103+
104+
for (auto& [shardId, stat] : stats->ExternalStats) {
105+
auto& externalStat = *sourceStats.AddExternalPartitions();
106+
externalStat.SetPartitionId(ToString(shardId));
107+
externalStat.SetExternalRows(stat.ExternalRows);
108+
externalStat.SetExternalBytes(stat.ExternalBytes);
109+
externalStat.SetFirstMessageMs(stat.FirstMessageMs);
110+
externalStat.SetLastMessageMs(stat.LastMessageMs);
111+
}
112+
113+
taskStats->SetIngressRows(taskStats->GetIngressRows() + stats->Rows);
114+
taskStats->SetIngressBytes(taskStats->GetIngressBytes() + stats->Bytes);
115+
116+
tableStats->SetReadRows(stats->Rows);
117+
tableStats->SetReadBytes(stats->Bytes);
118+
tableStats->SetAffectedPartitions(stats->AffectedShards);
95119
// TODO: CpuTime
96120
}
97121

‎ydb/core/kqp/executer_actor/kqp_executer_stats.cpp

Lines changed: 114 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -189,8 +189,9 @@ void TStageExecutionStats::Resize(ui32 taskCount) {
189189
FinishTimeMs.resize(taskCount);
190190
StartTimeMs.resize(taskCount);
191191
DurationUs.resize(taskCount);
192-
WaitInputTimeUs.resize(taskCount);
193-
WaitOutputTimeUs.resize(taskCount);
192+
193+
WaitInputTimeUs.Resize(taskCount);
194+
WaitOutputTimeUs.Resize(taskCount);
194195

195196
SpillingComputeBytes.Resize(taskCount);
196197
SpillingChannelBytes.Resize(taskCount);
@@ -215,6 +216,10 @@ void TStageExecutionStats::SetHistorySampleCount(ui32 historySampleCount) {
215216
HistorySampleCount = historySampleCount;
216217
CpuTimeUs.HistorySampleCount = historySampleCount;
217218
MaxMemoryUsage.HistorySampleCount = historySampleCount;
219+
220+
WaitInputTimeUs.HistorySampleCount = historySampleCount;
221+
WaitOutputTimeUs.HistorySampleCount = historySampleCount;
222+
218223
SpillingComputeBytes.HistorySampleCount = historySampleCount;
219224
SpillingChannelBytes.HistorySampleCount = historySampleCount;
220225
SpillingComputeTimeUs.HistorySampleCount = historySampleCount;
@@ -252,6 +257,12 @@ void TStageExecutionStats::ExportHistory(ui64 baseTimeMs, NYql::NDqProto::TDqSta
252257
if (stageStats.HasMaxMemoryUsage()) {
253258
MaxMemoryUsage.ExportHistory(baseTimeMs, *stageStats.MutableMaxMemoryUsage());
254259
}
260+
if (stageStats.HasWaitInputTimeUs()) {
261+
WaitInputTimeUs.ExportHistory(baseTimeMs, *stageStats.MutableWaitInputTimeUs());
262+
}
263+
if (stageStats.HasWaitOutputTimeUs()) {
264+
WaitOutputTimeUs.ExportHistory(baseTimeMs, *stageStats.MutableWaitOutputTimeUs());
265+
}
255266
if (stageStats.HasSpillingComputeBytes()) {
256267
SpillingComputeBytes.ExportHistory(baseTimeMs, *stageStats.MutableSpillingComputeBytes());
257268
}
@@ -346,8 +357,8 @@ ui64 TStageExecutionStats::UpdateStats(const NYql::NDqProto::TDqTaskStats& taskS
346357
baseTimeMs = NonZeroMin(baseTimeMs, finishTimeMs);
347358

348359
SetNonZero(DurationUs[index], durationUs);
349-
SetNonZero(WaitInputTimeUs[index], taskStats.GetWaitInputTimeUs());
350-
SetNonZero(WaitOutputTimeUs[index], taskStats.GetWaitOutputTimeUs());
360+
WaitInputTimeUs.SetNonZero(index, taskStats.GetWaitInputTimeUs());
361+
WaitOutputTimeUs.SetNonZero(index, taskStats.GetWaitOutputTimeUs());
351362

352363
SpillingComputeBytes.SetNonZero(index, taskStats.GetSpillingComputeWriteBytes());
353364
SpillingChannelBytes.SetNonZero(index, taskStats.GetSpillingChannelWriteBytes());
@@ -466,6 +477,13 @@ TProgressStatEntry operator - (const TProgressStatEntry& l, const TProgressStatE
466477
};
467478
}
468479

480+
void MergeAggr(NDqProto::TDqStatsAggr& aggr, const NDqProto::TDqStatsAggr& stat) noexcept {
481+
aggr.SetMin(NonZeroMin(aggr.GetMin(), stat.GetMin()));
482+
aggr.SetMax(std::max(aggr.GetMax(), stat.GetMax()));
483+
aggr.SetSum(aggr.GetSum() + stat.GetSum());
484+
aggr.SetCnt(aggr.GetCnt() + stat.GetCnt());
485+
}
486+
469487
void UpdateAggr(NDqProto::TDqStatsAggr* aggr, ui64 value) noexcept {
470488
if (value) {
471489
if (aggr->GetMin() == 0) {
@@ -479,6 +497,19 @@ void UpdateAggr(NDqProto::TDqStatsAggr* aggr, ui64 value) noexcept {
479497
}
480498
}
481499

500+
void MergeExternal(NDqProto::TDqExternalAggrStats& asyncAggr, const NDqProto::TDqExternalAggrStats& asyncStat) noexcept {
501+
MergeAggr(*asyncAggr.MutableExternalRows(), asyncStat.GetExternalRows());
502+
MergeAggr(*asyncAggr.MutableExternalBytes(), asyncStat.GetExternalBytes());
503+
MergeAggr(*asyncAggr.MutableStorageRows(), asyncStat.GetStorageRows());
504+
MergeAggr(*asyncAggr.MutableStorageBytes(), asyncStat.GetStorageBytes());
505+
MergeAggr(*asyncAggr.MutableCpuTimeUs(), asyncStat.GetCpuTimeUs());
506+
MergeAggr(*asyncAggr.MutableWaitInputTimeUs(), asyncStat.GetWaitInputTimeUs());
507+
MergeAggr(*asyncAggr.MutableWaitOutputTimeUs(), asyncStat.GetWaitOutputTimeUs());
508+
MergeAggr(*asyncAggr.MutableFirstMessageMs(), asyncStat.GetFirstMessageMs());
509+
MergeAggr(*asyncAggr.MutableLastMessageMs(), asyncStat.GetLastMessageMs());
510+
asyncAggr.SetPartitionCount(asyncAggr.GetPartitionCount() + asyncStat.GetExternalRows().GetCnt());
511+
}
512+
482513
ui64 UpdateAsyncAggr(NDqProto::TDqAsyncStatsAggr& asyncAggr, const NDqProto::TDqAsyncBufferStats& asyncStat) noexcept {
483514
ui64 baseTimeMs = 0;
484515

@@ -652,9 +683,27 @@ void TQueryExecutionStats::AddComputeActorFullStatsByTask(
652683
FillStageDurationUs(*stageStats);
653684

654685
for (auto& sourcesStat : task.GetSources()) {
655-
BaseTimeMs = NonZeroMin(BaseTimeMs, UpdateAsyncAggr(*(*stageStats->MutableIngress())[sourcesStat.GetIngressName()].MutableIngress(), sourcesStat.GetIngress()));
656-
BaseTimeMs = NonZeroMin(BaseTimeMs, UpdateAsyncAggr(*(*stageStats->MutableIngress())[sourcesStat.GetIngressName()].MutablePush(), sourcesStat.GetPush()));
657-
BaseTimeMs = NonZeroMin(BaseTimeMs, UpdateAsyncAggr(*(*stageStats->MutableIngress())[sourcesStat.GetIngressName()].MutablePop(), sourcesStat.GetPop()));
686+
auto& ingress = (*stageStats->MutableIngress())[sourcesStat.GetIngressName()];
687+
MergeExternal(*ingress.MutableExternal(), sourcesStat.GetExternal());
688+
689+
const auto& [it, inserted] = ExternalPartitionStats.emplace(stageStats->GetStageId(), sourcesStat.GetIngressName());
690+
auto& externalPartitionStat = it->second;
691+
692+
for (auto& externalPartition : sourcesStat.GetExternalPartitions()) {
693+
const auto& [it, inserted] = externalPartitionStat.Stat.emplace(externalPartition.GetPartitionId(),
694+
TExternalPartitionStat(externalPartition.GetExternalRows(), externalPartition.GetExternalBytes(),
695+
externalPartition.GetFirstMessageMs(), externalPartition.GetLastMessageMs()));
696+
if (!inserted) {
697+
it->second.ExternalRows += externalPartition.GetExternalRows();
698+
it->second.ExternalBytes += externalPartition.GetExternalBytes();
699+
it->second.FirstMessageMs = NonZeroMin(it->second.FirstMessageMs, externalPartition.GetFirstMessageMs());
700+
it->second.LastMessageMs = std::max(it->second.LastMessageMs, externalPartition.GetLastMessageMs());
701+
}
702+
}
703+
704+
BaseTimeMs = NonZeroMin(BaseTimeMs, UpdateAsyncAggr(*ingress.MutableIngress(), sourcesStat.GetIngress()));
705+
BaseTimeMs = NonZeroMin(BaseTimeMs, UpdateAsyncAggr(*ingress.MutablePush(), sourcesStat.GetPush()));
706+
BaseTimeMs = NonZeroMin(BaseTimeMs, UpdateAsyncAggr(*ingress.MutablePop(), sourcesStat.GetPop()));
658707
}
659708
for (auto& inputChannelStat : task.GetInputChannels()) {
660709
BaseTimeMs = NonZeroMin(BaseTimeMs, UpdateAsyncAggr(*(*stageStats->MutableInput())[inputChannelStat.GetSrcStageId()].MutablePush(), inputChannelStat.GetPush()));
@@ -823,9 +872,14 @@ void TQueryExecutionStats::AddDatashardFullStatsByTask(
823872
UpdateAggr(stageStats->MutableFinishTimeMs(), finishTimeMs);
824873
BaseTimeMs = NonZeroMin(BaseTimeMs, finishTimeMs);
825874

875+
FillStageDurationUs(*stageStats);
826876
UpdateAggr(stageStats->MutableWaitInputTimeUs(), task.GetWaitInputTimeUs());
827877
UpdateAggr(stageStats->MutableWaitOutputTimeUs(), task.GetWaitOutputTimeUs());
828-
FillStageDurationUs(*stageStats);
878+
879+
UpdateAggr(stageStats->MutableSpillingComputeBytes(), task.GetSpillingComputeWriteBytes());
880+
UpdateAggr(stageStats->MutableSpillingChannelBytes(), task.GetSpillingChannelWriteBytes());
881+
UpdateAggr(stageStats->MutableSpillingComputeTimeUs(), task.GetSpillingComputeReadTimeUs() + task.GetSpillingComputeWriteTimeUs());
882+
UpdateAggr(stageStats->MutableSpillingChannelTimeUs(), task.GetSpillingChannelReadTimeUs() + task.GetSpillingChannelWriteTimeUs());
829883

830884
for (auto& tableStats: task.GetTables()) {
831885
auto* tableAggrStats = GetOrCreateTableAggrStats(stageStats, tableStats.GetTablePath());
@@ -1094,8 +1148,8 @@ void TQueryExecutionStats::ExportExecStats(NYql::NDqProto::TDqExecutionStats& st
10941148
ExportOffsetAggStats(stageStat.StartTimeMs, *stageStats.MutableStartTimeMs(), BaseTimeMs);
10951149
ExportOffsetAggStats(stageStat.FinishTimeMs, *stageStats.MutableFinishTimeMs(), BaseTimeMs);
10961150
ExportAggStats(stageStat.DurationUs, *stageStats.MutableDurationUs());
1097-
ExportAggStats(stageStat.WaitInputTimeUs, *stageStats.MutableWaitInputTimeUs());
1098-
ExportAggStats(stageStat.WaitOutputTimeUs, *stageStats.MutableWaitOutputTimeUs());
1151+
stageStat.WaitInputTimeUs.ExportAggStats(BaseTimeMs, *stageStats.MutableWaitInputTimeUs());
1152+
stageStat.WaitOutputTimeUs.ExportAggStats(BaseTimeMs, *stageStats.MutableWaitOutputTimeUs());
10991153

11001154
stageStat.SpillingComputeBytes.ExportAggStats(BaseTimeMs, *stageStats.MutableSpillingComputeBytes());
11011155
stageStat.SpillingChannelBytes.ExportAggStats(BaseTimeMs, *stageStats.MutableSpillingChannelBytes());
@@ -1152,6 +1206,15 @@ void TQueryExecutionStats::ExportExecStats(NYql::NDqProto::TDqExecutionStats& st
11521206
}
11531207
}
11541208

1209+
void TQueryExecutionStats::AdjustExternalAggr(NYql::NDqProto::TDqExternalAggrStats& stats) {
1210+
if (stats.HasFirstMessageMs()) {
1211+
AdjustDqStatsAggr(*stats.MutableFirstMessageMs());
1212+
}
1213+
if (stats.HasLastMessageMs()) {
1214+
AdjustDqStatsAggr(*stats.MutableLastMessageMs());
1215+
}
1216+
}
1217+
11551218
void TQueryExecutionStats::AdjustAsyncAggr(NYql::NDqProto::TDqAsyncStatsAggr& stats) {
11561219
if (stats.HasFirstMessageMs()) {
11571220
AdjustDqStatsAggr(*stats.MutableFirstMessageMs());
@@ -1168,6 +1231,9 @@ void TQueryExecutionStats::AdjustAsyncAggr(NYql::NDqProto::TDqAsyncStatsAggr& st
11681231
}
11691232

11701233
void TQueryExecutionStats::AdjustAsyncBufferAggr(NYql::NDqProto::TDqAsyncBufferStatsAggr& stats) {
1234+
if (stats.HasExternal()) {
1235+
AdjustExternalAggr(*stats.MutableExternal());
1236+
}
11711237
if (stats.HasIngress()) {
11721238
AdjustAsyncAggr(*stats.MutableIngress());
11731239
}
@@ -1184,13 +1250,15 @@ void TQueryExecutionStats::AdjustAsyncBufferAggr(NYql::NDqProto::TDqAsyncBufferS
11841250

11851251
void TQueryExecutionStats::AdjustDqStatsAggr(NYql::NDqProto::TDqStatsAggr& stats) {
11861252
if (auto min = stats.GetMin()) {
1187-
stats.SetMin(min - BaseTimeMs);
1253+
stats.SetMin(min > BaseTimeMs ? min - BaseTimeMs : 0);
11881254
}
11891255
if (auto max = stats.GetMax()) {
1190-
stats.SetMax(max - BaseTimeMs);
1256+
stats.SetMax(max > BaseTimeMs ? max - BaseTimeMs : 0);
11911257
}
11921258
if (auto cnt = stats.GetCnt()) {
1193-
stats.SetSum(stats.GetSum() - BaseTimeMs * cnt);
1259+
auto sum = stats.GetSum();
1260+
auto baseSum = BaseTimeMs * cnt;
1261+
stats.SetSum(sum > baseSum ? sum - baseSum : 0);
11941262
}
11951263
}
11961264

@@ -1222,6 +1290,39 @@ void TQueryExecutionStats::Finish() {
12221290
for (auto& [stageId, stagetype] : TasksGraph->GetStagesInfo()) {
12231291
auto stageStats = GetOrCreateStageStats(stageId, *TasksGraph, *Result);
12241292
stageStats->SetBaseTimeMs(BaseTimeMs);
1293+
1294+
if (ExternalPartitionStats.contains(stageStats->GetStageId())) {
1295+
auto& externalPartitionStat = ExternalPartitionStats[stageStats->GetStageId()];
1296+
auto& ingress = (*stageStats->MutableIngress())[externalPartitionStat.Name];
1297+
auto& external = *ingress.MutableExternal();
1298+
for (auto& [partitionId, partitionStat] : externalPartitionStat.Stat) {
1299+
auto& externalRows = *external.MutableExternalRows();
1300+
externalRows.SetMin(NonZeroMin(externalRows.GetMin(), partitionStat.ExternalRows));
1301+
externalRows.SetMax(std::max(externalRows.GetMax(), partitionStat.ExternalRows));
1302+
externalRows.SetSum(externalRows.GetSum() + partitionStat.ExternalRows);
1303+
externalRows.SetCnt(externalRows.GetCnt() + 1);
1304+
1305+
auto& externalBytes = *external.MutableExternalBytes();
1306+
externalBytes.SetMin(NonZeroMin(externalBytes.GetMin(), partitionStat.ExternalBytes));
1307+
externalBytes.SetMax(std::max(externalBytes.GetMax(), partitionStat.ExternalBytes));
1308+
externalBytes.SetSum(externalBytes.GetSum() + partitionStat.ExternalBytes);
1309+
externalBytes.SetCnt(externalBytes.GetCnt() + 1);
1310+
1311+
auto& firstMessageMs = *external.MutableFirstMessageMs();
1312+
firstMessageMs.SetMin(NonZeroMin(firstMessageMs.GetMin(), partitionStat.FirstMessageMs));
1313+
firstMessageMs.SetMax(std::max(firstMessageMs.GetMax(), partitionStat.FirstMessageMs));
1314+
firstMessageMs.SetSum(firstMessageMs.GetSum() + partitionStat.FirstMessageMs);
1315+
firstMessageMs.SetCnt(firstMessageMs.GetCnt() + 1);
1316+
1317+
auto& lastMessageMs = *external.MutableLastMessageMs();
1318+
lastMessageMs.SetMin(NonZeroMin(lastMessageMs.GetMin(), partitionStat.LastMessageMs));
1319+
lastMessageMs.SetMax(std::max(lastMessageMs.GetMax(), partitionStat.LastMessageMs));
1320+
lastMessageMs.SetSum(lastMessageMs.GetSum() + partitionStat.LastMessageMs);
1321+
lastMessageMs.SetCnt(lastMessageMs.GetCnt() + 1);
1322+
}
1323+
external.SetPartitionCount(external.GetPartitionCount() + externalPartitionStat.Stat.size());
1324+
}
1325+
12251326
AdjustBaseTime(stageStats);
12261327
auto it = StageStats.find(stageId.StageId);
12271328
if (it != StageStats.end()) {

‎ydb/core/kqp/executer_actor/kqp_executer_stats.h

Lines changed: 32 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -66,6 +66,16 @@ struct TAsyncBufferStats {
6666
void ExportHistory(ui64 baseTimeMs, NYql::NDqProto::TDqAsyncBufferStatsAggr& stats);
6767
};
6868

69+
struct TIngressStats : public TAsyncBufferStats {
70+
71+
TIngressStats() = default;
72+
TIngressStats(ui32 taskCount) {
73+
Resize(taskCount);
74+
}
75+
76+
void Resize(ui32 taskCount);
77+
};
78+
6979
struct TTableStats {
7080

7181
TTableStats() = default;
@@ -125,8 +135,8 @@ struct TStageExecutionStats {
125135
std::vector<ui64> FinishTimeMs;
126136
std::vector<ui64> StartTimeMs;
127137
std::vector<ui64> DurationUs;
128-
std::vector<ui64> WaitInputTimeUs;
129-
std::vector<ui64> WaitOutputTimeUs;
138+
TTimeSeriesStats WaitInputTimeUs;
139+
TTimeSeriesStats WaitOutputTimeUs;
130140

131141
TTimeSeriesStats SpillingComputeBytes;
132142
TTimeSeriesStats SpillingChannelBytes;
@@ -153,14 +163,34 @@ struct TStageExecutionStats {
153163
ui64 UpdateStats(const NYql::NDqProto::TDqTaskStats& taskStats, ui64 maxMemoryUsage, ui64 durationUs);
154164
};
155165

166+
struct TExternalPartitionStat {
167+
ui64 ExternalRows;
168+
ui64 ExternalBytes;
169+
ui64 FirstMessageMs;
170+
ui64 LastMessageMs;
171+
TExternalPartitionStat() = default;
172+
TExternalPartitionStat(ui64 externalRows, ui64 externalBytes, ui64 firstMessageMs, ui64 lastMessageMs)
173+
: ExternalRows(externalRows), ExternalBytes(externalBytes), FirstMessageMs(firstMessageMs), LastMessageMs(lastMessageMs)
174+
{}
175+
};
176+
177+
struct TIngressExternalPartitionStat {
178+
TString Name;
179+
std::map<TString, TExternalPartitionStat> Stat;
180+
TIngressExternalPartitionStat() = default;
181+
TIngressExternalPartitionStat(const TString& name) : Name(name) {}
182+
};
183+
156184
struct TQueryExecutionStats {
157185
private:
158186
std::map<ui32, std::map<ui32, ui32>> ShardsCountByNode;
159187
std::map<ui32, bool> UseLlvmByStageId;
160188
std::map<ui32, TStageExecutionStats> StageStats;
189+
std::map<ui32, TIngressExternalPartitionStat> ExternalPartitionStats; // FIXME: several ingresses
161190
ui64 BaseTimeMs = 0;
162191
void ExportAggAsyncStats(TAsyncStats& data, NYql::NDqProto::TDqAsyncStatsAggr& stats);
163192
void ExportAggAsyncBufferStats(TAsyncBufferStats& data, NYql::NDqProto::TDqAsyncBufferStatsAggr& stats);
193+
void AdjustExternalAggr(NYql::NDqProto::TDqExternalAggrStats& stats);
164194
void AdjustAsyncAggr(NYql::NDqProto::TDqAsyncStatsAggr& stats);
165195
void AdjustAsyncBufferAggr(NYql::NDqProto::TDqAsyncBufferStatsAggr& stats);
166196
void AdjustDqStatsAggr(NYql::NDqProto::TDqStatsAggr& stats);

‎ydb/core/kqp/opt/kqp_query_plan.cpp

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2921,6 +2921,38 @@ TString AddExecStatsToTxPlan(const TString& txPlanJson, const NYql::NDqProto::TD
29212921
for (auto ingress : (*stat)->GetIngress()) {
29222922
auto& ingressInfo = ingressStats.AppendValue(NJson::JSON_MAP);
29232923
ingressInfo["Name"] = ingress.first;
2924+
if (ingress.second.HasExternal()) {
2925+
auto& node = ingressInfo.InsertValue("External", NJson::JSON_MAP);
2926+
auto& externalInfo = ingress.second.GetExternal();
2927+
if (externalInfo.HasExternalRows()) {
2928+
FillAggrStat(node, externalInfo.GetExternalRows(), "ExternalRows");
2929+
}
2930+
if (externalInfo.HasExternalBytes()) {
2931+
FillAggrStat(node, externalInfo.GetExternalBytes(), "ExternalBytes");
2932+
}
2933+
if (externalInfo.HasStorageRows()) {
2934+
FillAggrStat(node, externalInfo.GetStorageRows(), "StorageRows");
2935+
}
2936+
if (externalInfo.HasStorageBytes()) {
2937+
FillAggrStat(node, externalInfo.GetStorageBytes(), "StorageBytes");
2938+
}
2939+
if (externalInfo.HasCpuTimeUs()) {
2940+
FillAggrStat(node, externalInfo.GetCpuTimeUs(), "CpuTimeUs");
2941+
}
2942+
if (externalInfo.HasWaitInputTimeUs()) {
2943+
FillAggrStat(node, externalInfo.GetWaitInputTimeUs(), "WaitInputTimeUs");
2944+
}
2945+
if (externalInfo.HasWaitOutputTimeUs()) {
2946+
FillAggrStat(node, externalInfo.GetWaitOutputTimeUs(), "WaitOutputTimeUs");
2947+
}
2948+
if (externalInfo.HasFirstMessageMs()) {
2949+
FillAggrStat(node, externalInfo.GetFirstMessageMs(), "FirstMessageMs");
2950+
}
2951+
if (externalInfo.HasLastMessageMs()) {
2952+
FillAggrStat(node, externalInfo.GetLastMessageMs(), "LastMessageMs");
2953+
}
2954+
SetNonZero(node, "PartitionCount", externalInfo.GetPartitionCount());
2955+
}
29242956
if (ingress.second.HasIngress()) {
29252957
FillAsyncAggrStat(ingressInfo.InsertValue("Ingress", NJson::JSON_MAP), ingress.second.GetIngress());
29262958
}

0 commit comments

Comments
 (0)