Skip to content

Commit eaa23fa

Browse files
GrigoriyPAGrigoriyPA
authored
Merge 3496c5e into 1091a58 
2 parents 1091a58 + 3496c5e commit eaa23fa

File tree

19 files changed

+85
-25
lines changed

19 files changed

+85
-25
lines changed

‎ydb/core/grpc_services/rpc_kqp_base.cpp

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -67,6 +67,7 @@ void FillQueryStats(Ydb::TableStats::QueryStats& queryStats, const NKqpProto::TK
6767
queryStats.set_process_cpu_time_us(kqpStats.GetWorkerCpuTimeUs());
6868
queryStats.set_total_cpu_time_us(totalCpuTimeUs);
6969
queryStats.set_total_duration_us(kqpStats.GetDurationUs());
70+
queryStats.set_queued_time_us(kqpStats.GetQueuedTimeUs());
7071
}
7172

7273
void FillQueryStats(Ydb::TableStats::QueryStats& queryStats, const NKikimrKqp::TQueryResponse& kqpResponse) {

‎ydb/core/kqp/session_actor/kqp_query_state.h

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -114,6 +114,7 @@ class TKqpQueryState : public TNonCopyable {
114114
bool IsDocumentApiRestricted_ = false;
115115

116116
TInstant StartTime;
117+
TInstant ContinueTime;
117118
NYql::TKikimrQueryDeadlines QueryDeadlines;
118119
TKqpQueryStats QueryStats;
119120
bool KeepSession = false;

‎ydb/core/kqp/session_actor/kqp_query_stats.cpp

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -210,6 +210,7 @@ ui64 CalcRequestUnit(const TKqpQueryStats& stats) {
210210
NKqpProto::TKqpStatsQuery TKqpQueryStats::ToProto() const {
211211
NKqpProto::TKqpStatsQuery result;
212212
result.SetDurationUs(DurationUs);
213+
result.SetQueuedTimeUs(QueuedTimeUs);
213214

214215
if (Compilation) {
215216
result.MutableCompilation()->SetFromCache(Compilation->FromCache);

‎ydb/core/kqp/session_actor/kqp_query_stats.h

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ namespace NKikimr::NKqp {
88

99
struct TKqpQueryStats {
1010
ui64 DurationUs = 0;
11+
ui64 QueuedTimeUs = 0;
1112
std::optional<TKqpStatsCompile> Compilation;
1213

1314
ui64 WorkerCpuTimeUs = 0;

‎ydb/core/kqp/session_actor/kqp_session_actor.cpp

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -475,6 +475,7 @@ class TKqpSessionActor : public TActorBootstrapped<TKqpSessionActor> {
475475

476476
void Handle(NWorkload::TEvContinueRequest::TPtr& ev) {
477477
YQL_ENSURE(QueryState);
478+
QueryState->ContinueTime = TInstant::Now();
478479

479480
if (ev->Get()->Status == Ydb::StatusIds::UNSUPPORTED) {
480481
LOG_T("Failed to place request in resource pool, feature flag is disabled");
@@ -1552,6 +1553,9 @@ class TKqpSessionActor : public TActorBootstrapped<TKqpSessionActor> {
15521553

15531554
stats->DurationUs = ((TInstant::Now() - QueryState->StartTime).MicroSeconds());
15541555
stats->WorkerCpuTimeUs = (QueryState->GetCpuTime().MicroSeconds());
1556+
if (const auto continueTime = QueryState->ContinueTime) {
1557+
stats->QueuedTimeUs = ((TInstant::Now() - continueTime).MicroSeconds());
1558+
}
15551559
if (QueryState->CompileResult) {
15561560
stats->Compilation.emplace();
15571561
stats->Compilation->FromCache = (QueryState->CompileStats.FromCache);

‎ydb/core/kqp/ut/service/kqp_qs_queries_ut.cpp

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -245,13 +245,15 @@ Y_UNIT_TEST_SUITE(KqpQueryService) {
245245
auto db = kikimr.GetQueryClient();
246246

247247
TExecuteQuerySettings settings;
248+
settings.StatsMode(EStatsMode::Full);
248249

249250
{ // Existing pool
250251
settings.PoolId("default");
251252

252253
const TString query = "SELECT Key, Value2 FROM TwoShard WHERE Value2 > 0 ORDER BY Key";
253254
auto result = db.ExecuteQuery(query, TTxControl::BeginTx().CommitTx(), settings).ExtractValueSync();
254255
CheckQueryResult(result);
256+
UNIT_ASSERT_VALUES_UNEQUAL(result.GetStats()->GetQueuedTime(), TDuration::Zero());
255257
}
256258

257259
{ // Not existing pool (check workload manager enabled)

‎ydb/core/kqp/ut/service/kqp_qs_scripts_ut.cpp

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -109,13 +109,16 @@ Y_UNIT_TEST_SUITE(KqpQueryServiceScripts) {
109109
auto db = kikimr.GetQueryClient();
110110

111111
TExecuteScriptSettings settings;
112+
settings.StatsMode(EStatsMode::Full);
112113

113114
{ // Existing pool
114115
settings.PoolId("default");
115116

116117
auto scripOp = db.ExecuteScript("SELECT 42", settings).ExtractValueSync();
117118
UNIT_ASSERT_VALUES_EQUAL_C(scripOp.Status().GetStatus(), EStatus::SUCCESS, scripOp.Status().GetIssues().ToString());
118-
CheckScriptResults(scripOp, WaitScriptExecutionOperation(scripOp.Id(), kikimr.GetDriver()), db);
119+
auto readyOp = WaitScriptExecutionOperation(scripOp.Id(), kikimr.GetDriver());
120+
CheckScriptResults(scripOp, readyOp, db);
121+
UNIT_ASSERT_VALUES_UNEQUAL(readyOp.Metadata().ExecStats.GetQueuedTime(), TDuration::Zero());
119122
}
120123

121124
{ // Not existing pool (check workload manager enabled)

‎ydb/core/kqp/workload_service/actors/pool_handlers_acors.cpp

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -54,6 +54,12 @@ class TPoolHandlerActorBase : public TActor<TDerived> {
5454
UpdateConfigCounters(poolConfig);
5555
}
5656

57+
void CollectRequestLatency(TInstant continueTime) {
58+
if (continueTime) {
59+
RequestsLatencyMs->Collect((TInstant::Now() - continueTime).MilliSeconds());
60+
}
61+
}
62+
5763
void UpdateConfigCounters(const NResourcePool::TPoolSettings& poolConfig) {
5864
InFlightLimit->Set(std::max(poolConfig.ConcurrentQueryLimit, 0));
5965
QueueSizeLimit->Set(std::max(poolConfig.QueueSize, 0));
@@ -106,6 +112,7 @@ class TPoolHandlerActorBase : public TActor<TDerived> {
106112
const TActorId WorkerActorId;
107113
const TString SessionId;
108114
const TInstant StartTime = TInstant::Now();
115+
TInstant ContinueTime;
109116

110117
EState State = EState::Pending;
111118
bool Started = false; // after TEvContinueRequest success
@@ -267,6 +274,7 @@ class TPoolHandlerActorBase : public TActor<TDerived> {
267274
if (status == Ydb::StatusIds::SUCCESS) {
268275
LocalInFlight++;
269276
request->Started = true;
277+
request->ContinueTime = TInstant::Now();
270278
Counters.LocalInFly->Inc();
271279
Counters.ContinueOk->Inc();
272280
Counters.DelayedTimeMs->Collect((TInstant::Now() - request->StartTime).MilliSeconds());
@@ -387,7 +395,7 @@ class TPoolHandlerActorBase : public TActor<TDerived> {
387395

388396
if (status == Ydb::StatusIds::SUCCESS) {
389397
Counters.CleanupOk->Inc();
390-
Counters.RequestsLatencyMs->Collect((TInstant::Now() - request->StartTime).MilliSeconds());
398+
Counters.CollectRequestLatency(request->ContinueTime);
391399
LOG_D("Reply cleanup success to " << request->WorkerActorId << ", session id: " << request->SessionId << ", local in flight: " << LocalInFlight);
392400
} else {
393401
Counters.CleanupError->Inc();
@@ -401,7 +409,7 @@ class TPoolHandlerActorBase : public TActor<TDerived> {
401409
this->Send(MakeKqpProxyID(this->SelfId().NodeId()), ev.release());
402410

403411
Counters.Cancelled->Inc();
404-
Counters.RequestsLatencyMs->Collect((TInstant::Now() - request->StartTime).MilliSeconds());
412+
Counters.CollectRequestLatency(request->ContinueTime);
405413
LOG_I("Cancel request for worker " << request->WorkerActorId << ", session id: " << request->SessionId << ", local in flight: " << LocalInFlight);
406414
}
407415

‎ydb/core/kqp/workload_service/ut/common/kqp_workload_service_ut_common.cpp

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -487,6 +487,7 @@ class TWorkloadServiceYdbSetup : public IYdbSetup {
487487
request->SetQuery(query);
488488
request->SetType(NKikimrKqp::QUERY_TYPE_SQL_GENERIC_QUERY);
489489
request->SetAction(NKikimrKqp::QUERY_ACTION_EXECUTE);
490+
request->SetCollectStats(Ydb::Table::QueryStatsCollection::STATS_COLLECTION_FULL);
490491
request->SetDatabase(Settings_.DomainName_);
491492
request->SetPoolId(settings.PoolId_);
492493

@@ -567,6 +568,10 @@ const std::vector<Ydb::ResultSet>& TQueryRunnerResult::GetResultSets() const {
567568
return ResultSets;
568569
}
569570

571+
const NKqpProto::TKqpStatsQuery& TQueryRunnerResult::GetQueryStats() const {
572+
return Response.GetResponse().GetQueryStats();
573+
}
574+
570575
//// TQueryRunnerResultAsync
571576

572577
TQueryRunnerResult TQueryRunnerResultAsync::GetResult(TDuration timeout) const {

‎ydb/core/kqp/workload_service/ut/common/kqp_workload_service_ut_common.h

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -44,6 +44,7 @@ struct TQueryRunnerResult {
4444

4545
const Ydb::ResultSet& GetResultSet(size_t resultIndex) const;
4646
const std::vector<Ydb::ResultSet>& GetResultSets() const;
47+
const NKqpProto::TKqpStatsQuery& GetQueryStats() const;
4748
};
4849

4950
struct TQueryRunnerResultAsync {

‎ydb/core/kqp/workload_service/ut/kqp_workload_service_ut.cpp

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,8 @@
22

33
#include <ydb/core/kqp/workload_service/ut/common/kqp_workload_service_ut_common.h>
44

5+
#include <ydb/core/protos/kqp_stats.pb.h>
6+
57

68
namespace NKikimr::NKqp {
79

@@ -44,7 +46,9 @@ Y_UNIT_TEST_SUITE(KqpWorkloadService) {
4446
.EnableResourcePools(false)
4547
.Create();
4648

47-
TSampleQueries::TSelect42::CheckResult(ydb->ExecuteQuery(TSampleQueries::TSelect42::Query, TQueryRunnerSettings().PoolId("another_pool_id")));
49+
auto result = ydb->ExecuteQuery(TSampleQueries::TSelect42::Query, TQueryRunnerSettings().PoolId("another_pool_id"));
50+
TSampleQueries::TSelect42::CheckResult(result);
51+
UNIT_ASSERT_VALUES_EQUAL(result.GetQueryStats().GetQueuedTimeUs(), 0);
4852
}
4953

5054
TQueryRunnerResultAsync StartQueueSizeCheckRequests(TIntrusivePtr<IYdbSetup> ydb, const TQueryRunnerSettings& settings) {

‎ydb/core/protos/kqp_stats.proto

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -73,6 +73,7 @@ message TKqpExecutionExtraStats {
7373
message TKqpStatsQuery {
7474
// Basic stats
7575
uint64 DurationUs = 1;
76+
uint64 QueuedTimeUs = 9;
7677
TKqpStatsCompile Compilation = 2;
7778

7879
reserved 3; // repeated TKqpStatsExecution Executions = 3;

‎ydb/public/api/protos/ydb_query_stats.proto

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -43,4 +43,5 @@ message QueryStats {
4343
string query_ast = 5;
4444
uint64 total_duration_us = 6;
4545
uint64 total_cpu_time_us = 7;
46+
uint64 queued_time_us = 8;
4647
}

‎ydb/public/sdk/cpp/client/ydb_query/stats.cpp

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -60,6 +60,10 @@ TDuration TExecStats::GetTotalDuration() const {
6060
return TDuration::MicroSeconds(Impl_->Proto.total_duration_us());
6161
}
6262

63+
TDuration TExecStats::GetQueuedTime() const {
64+
return TDuration::MicroSeconds(Impl_->Proto.queued_time_us());
65+
}
66+
6367
TDuration TExecStats::GetTotalCpuTime() const {
6468
return TDuration::MicroSeconds(Impl_->Proto.total_cpu_time_us());
6569
}

‎ydb/public/sdk/cpp/client/ydb_query/stats.h

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@ class TExecStats {
3333
TMaybe<TString> GetAst() const;
3434

3535
TDuration GetTotalDuration() const;
36+
TDuration GetQueuedTime() const;
3637
TDuration GetTotalCpuTime() const;
3738

3839
private:

‎ydb/tests/tools/kqprun/kqprun.cpp

Lines changed: 0 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -91,13 +91,11 @@ void RunArgumentQueries(const TExecutionOptions& executionOptions, NKqpRun::TKqp
9191
Cout << "..." << colors.Default() << Endl;
9292
}
9393

94-
TInstant startTime = TInstant::Now();
9594
switch (executionCase) {
9695
case TExecutionOptions::EExecutionCase::GenericScript:
9796
if (!runner.ExecuteScript(executionOptions.ScriptQueries[id], executionOptions.ScriptQueryAction, executionOptions.TraceId)) {
9897
ythrow yexception() << TInstant::Now().ToIsoStringLocal() << " Script execution failed";
9998
}
100-
Cout << colors.Cyan() << "Script request finished. Time: " << TInstant::Now() - startTime << colors.Default() << Endl;
10199
Cout << colors.Yellow() << TInstant::Now().ToIsoStringLocal() << " Fetching script results..." << colors.Default() << Endl;
102100
if (!runner.FetchScriptResults()) {
103101
ythrow yexception() << TInstant::Now().ToIsoStringLocal() << " Fetch script results failed";
@@ -114,14 +112,12 @@ void RunArgumentQueries(const TExecutionOptions& executionOptions, NKqpRun::TKqp
114112
if (!runner.ExecuteQuery(executionOptions.ScriptQueries[id], executionOptions.ScriptQueryAction, executionOptions.TraceId)) {
115113
ythrow yexception() << TInstant::Now().ToIsoStringLocal() << " Query execution failed";
116114
}
117-
Cout << colors.Cyan() << "Generic request finished. Time: " << TInstant::Now() - startTime << colors.Default() << Endl;
118115
break;
119116

120117
case TExecutionOptions::EExecutionCase::YqlScript:
121118
if (!runner.ExecuteYqlScript(executionOptions.ScriptQueries[id], executionOptions.ScriptQueryAction, executionOptions.TraceId)) {
122119
ythrow yexception() << TInstant::Now().ToIsoStringLocal() << " Yql script execution failed";
123120
}
124-
Cout << colors.Cyan() << "Yql script request finished. Time: " << TInstant::Now() - startTime << colors.Default() << Endl;
125121
break;
126122

127123
case TExecutionOptions::EExecutionCase::AsyncQuery:

‎ydb/tests/tools/kqprun/src/kqp_runner.cpp

Lines changed: 22 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -133,15 +133,19 @@ class TKqpRunner::TImpl {
133133

134134
bool ExecuteQuery(const TString& query, NKikimrKqp::EQueryAction action, const TString& traceId, EQueryType queryType) {
135135
StartScriptTraceOpt();
136+
StartTime_ = TInstant::Now();
136137

138+
TString queryTypeStr;
137139
TQueryMeta meta;
138140
TRequestResult status;
139141
switch (queryType) {
140142
case EQueryType::ScriptQuery:
143+
queryTypeStr = "Generic";
141144
status = YdbSetup_.QueryRequest(query, action, traceId, meta, ResultSets_, GetProgressCallback());
142145
break;
143146

144147
case EQueryType::YqlScriptQuery:
148+
queryTypeStr = "Yql script";
145149
status = YdbSetup_.YqlScriptRequest(query, action, traceId, meta, ResultSets_);
146150
break;
147151

@@ -153,8 +157,9 @@ class TKqpRunner::TImpl {
153157
TYdbSetup::StopTraceOpt();
154158

155159
PrintScriptAst(meta.Ast);
156-
160+
PrintScriptProgress(ExecutionMeta_.Plan);
157161
PrintScriptPlan(meta.Plan);
162+
PrintScriptFinish(meta, queryTypeStr);
158163

159164
if (!status.IsSuccess()) {
160165
Cerr << CerrColors_.Red() << "Failed to execute query, reason:" << CerrColors_.Default() << Endl << status.ToString() << Endl;
@@ -220,6 +225,7 @@ class TKqpRunner::TImpl {
220225

221226
private:
222227
bool WaitScriptExecutionOperation() {
228+
StartTime_ = TInstant::Now();
223229
ExecutionMeta_ = TExecutionMeta();
224230

225231
TDuration getOperationPeriod = TDuration::Seconds(1);
@@ -245,8 +251,8 @@ class TKqpRunner::TImpl {
245251
}
246252

247253
PrintScriptAst(ExecutionMeta_.Ast);
248-
249254
PrintScriptPlan(ExecutionMeta_.Plan);
255+
PrintScriptFinish(ExecutionMeta_, "Script");
250256

251257
if (!status.IsSuccess() || ExecutionMeta_.ExecutionStatus != NYdb::NQuery::EExecStatus::Completed) {
252258
Cerr << CerrColors_.Red() << "Failed to execute script, invalid final status, reason:" << CerrColors_.Default() << Endl << status.ToString() << Endl;
@@ -377,6 +383,19 @@ class TKqpRunner::TImpl {
377383
}
378384
}
379385

386+
void PrintScriptFinish(const TQueryMeta& meta, const TString& queryType) const {
387+
Cout << CoutColors_.Cyan() << queryType << " request finished.";
388+
if (meta.TotalDuration) {
389+
Cout << " Total duration: " << meta.TotalDuration;
390+
} else {
391+
Cout << " Estimated duration: " << TInstant::Now() - StartTime_;
392+
}
393+
if (meta.QueuedTime) {
394+
Cout << ", Queued time: " << meta.QueuedTime;
395+
}
396+
Cout << CoutColors_.Default() << Endl;
397+
}
398+
380399
private:
381400
TRunnerOptions Options_;
382401

@@ -388,6 +407,7 @@ class TKqpRunner::TImpl {
388407
TString ExecutionOperation_;
389408
TExecutionMeta ExecutionMeta_;
390409
std::vector<Ydb::ResultSet> ResultSets_;
410+
TInstant StartTime_;
391411
};
392412

393413

‎ydb/tests/tools/kqprun/src/ydb_setup.cpp

Lines changed: 13 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -60,6 +60,15 @@ class TStaticSecuredCredentialsFactory : public NYql::ISecuredServiceAccountCred
6060
TString YqlToken_;
6161
};
6262

63+
void FillQueryMeta(TQueryMeta& meta, const NKikimrKqp::TQueryResponse& response) {
64+
meta.Ast = response.GetQueryAst();
65+
if (const auto& plan = response.GetQueryPlan()) {
66+
meta.Plan = plan;
67+
}
68+
meta.TotalDuration = TDuration::MicroSeconds(response.GetQueryStats().GetDurationUs());
69+
meta.QueuedTime = TDuration::MicroSeconds(response.GetQueryStats().GetQueuedTimeUs());
70+
}
71+
6372
} // anonymous namespace
6473

6574

@@ -424,10 +433,7 @@ TRequestResult TYdbSetup::QueryRequest(const TString& query, NKikimrKqp::EQueryA
424433
const auto& responseRecord = queryOperationResponse.GetResponse();
425434

426435
resultSets = std::move(queryResponse.ResultSets);
427-
meta.Ast = responseRecord.GetQueryAst();
428-
if (const auto& plan = responseRecord.GetQueryPlan()) {
429-
meta.Plan = plan;
430-
}
436+
FillQueryMeta(meta, responseRecord);
431437

432438
return TRequestResult(queryOperationResponse.GetYdbStatus(), responseRecord.GetQueryIssues());
433439
}
@@ -438,8 +444,7 @@ TRequestResult TYdbSetup::YqlScriptRequest(const TString& query, NKikimrKqp::EQu
438444
auto yqlQueryOperationResponse = Impl_->YqlScriptRequest(query, action, traceId)->Get()->Record.GetRef();
439445
const auto& responseRecord = yqlQueryOperationResponse.GetResponse();
440446

441-
meta.Ast = responseRecord.GetQueryAst();
442-
meta.Plan = responseRecord.GetQueryPlan();
447+
FillQueryMeta(meta, responseRecord);
443448

444449
resultSets.reserve(responseRecord.results_size());
445450
for (const auto& result : responseRecord.results()) {
@@ -466,6 +471,8 @@ TRequestResult TYdbSetup::GetScriptExecutionOperationRequest(const TString& oper
466471
if (deserializedMeta.exec_stats().query_plan() != "{}") {
467472
meta.Plan = deserializedMeta.exec_stats().query_plan();
468473
}
474+
meta.TotalDuration = TDuration::MicroSeconds(deserializedMeta.exec_stats().total_duration_us());
475+
meta.QueuedTime = TDuration::MicroSeconds(deserializedMeta.exec_stats().queued_time_us());
469476
}
470477

471478
return TRequestResult(scriptExecutionOperation->Get()->Status, scriptExecutionOperation->Get()->Issues);

‎ydb/tests/tools/kqprun/src/ydb_setup.h

Lines changed: 8 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -14,20 +14,19 @@ struct TSchemeMeta {
1414
};
1515

1616

17-
struct TExecutionMeta {
18-
bool Ready = false;
19-
NYdb::NQuery::EExecStatus ExecutionStatus = NYdb::NQuery::EExecStatus::Unspecified;
20-
21-
i32 ResultSetsCount = 0;
22-
17+
struct TQueryMeta {
2318
TString Ast;
2419
TString Plan;
20+
TDuration TotalDuration;
21+
TDuration QueuedTime;
2522
};
2623

2724

28-
struct TQueryMeta {
29-
TString Ast;
30-
TString Plan;
25+
struct TExecutionMeta : public TQueryMeta {
26+
bool Ready = false;
27+
NYdb::NQuery::EExecStatus ExecutionStatus = NYdb::NQuery::EExecStatus::Unspecified;
28+
29+
i32 ResultSetsCount = 0;
3130
};
3231

3332

0 commit comments

Comments
 (0)