Skip to content

Commit 8773c1c

Browse files
gridnevvvitgridnevvvit
authored
merge performance improvement changes to stable-24-1 (#1576)
* support trailing generic query responses (#1441) * improving potential bottleneck in grpc layer of the query service (#1278)
1 parent 6149d92 commit 8773c1c

File tree

16 files changed

+330
-154
lines changed

16 files changed

+330
-154
lines changed

‎ydb/core/driver_lib/run/run.cpp‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -832,7 +832,7 @@ void TKikimrRunner::InitializeGRpc(const TKikimrRunConfig& runConfig) {
832832

833833
if (hasQueryService) {
834834
server.AddService(new NGRpcService::TGRpcYdbQueryService(ActorSystem.Get(), Counters,
835-
grpcRequestProxies[0], hasDataStreams.IsRlAllowed()));
835+
grpcRequestProxies, hasDataStreams.IsRlAllowed(), grpcConfig.GetHandlersPerCompletionQueue()));
836836
}
837837

838838
if (hasLogStore) {

‎ydb/core/grpc_services/query/rpc_execute_query.cpp‎

Lines changed: 40 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@ using TEvExecuteQueryRequest = TGrpcRequestNoOperationCall<Ydb::Query::ExecuteQu
2424
struct TProducerState {
2525
TMaybe<ui64> LastSeqNo;
2626
ui64 AckedFreeSpaceBytes = 0;
27+
TActorId ActorId;
2728
};
2829

2930
class TRpcFlowControlState {
@@ -244,8 +245,7 @@ class TExecuteQueryRPC : public TActorBootstrapped<TExecuteQueryRPC> {
244245
const auto traceId = Request_->GetTraceId();
245246

246247
NYql::TIssues issues;
247-
NKikimrKqp::EQueryAction queryAction;
248-
if (!ParseQueryAction(*req, queryAction, issues)) {
248+
if (!ParseQueryAction(*req, QueryAction, issues)) {
249249
return ReplyFinishStream(Ydb::StatusIds::BAD_REQUEST, std::move(issues));
250250
}
251251

@@ -274,7 +274,7 @@ class TExecuteQueryRPC : public TActorBootstrapped<TExecuteQueryRPC> {
274274
cachePolicy->set_keep_in_cache(true);
275275

276276
auto ev = MakeHolder<NKqp::TEvKqp::TEvQueryRequest>(
277-
queryAction,
277+
QueryAction,
278278
queryType,
279279
SelfId(),
280280
Request_,
@@ -288,7 +288,8 @@ class TExecuteQueryRPC : public TActorBootstrapped<TExecuteQueryRPC> {
288288
nullptr, // operationParams
289289
false, // keepSession
290290
false, // useCancelAfter
291-
syntax);
291+
syntax,
292+
true);
292293

293294
if (!ctx.Send(NKqp::MakeKqpProxyID(ctx.SelfID.NodeId()), ev.Release())) {
294295
NYql::TIssues issues;
@@ -322,23 +323,24 @@ class TExecuteQueryRPC : public TActorBootstrapped<TExecuteQueryRPC> {
322323

323324
ui64 freeSpaceBytes = FlowControl_.FreeSpaceBytes();
324325

325-
for (auto& pair : StreamProducers_) {
326-
const auto& producerId = pair.first;
327-
auto& producer = pair.second;
326+
for (auto& pair : StreamChannels_) {
327+
const auto& channelId = pair.first;
328+
auto& channel = pair.second;
328329

329-
if (freeSpaceBytes > 0 && producer.LastSeqNo && producer.AckedFreeSpaceBytes == 0) {
330+
if (freeSpaceBytes > 0 && channel.LastSeqNo && channel.AckedFreeSpaceBytes == 0) {
330331
LOG_DEBUG_S(ctx, NKikimrServices::RPC_REQUEST, this->SelfId() << "Resume execution, "
331-
<< ", producer: " << producerId
332-
<< ", seqNo: " << producer.LastSeqNo
332+
<< ", channel: " << channelId
333+
<< ", seqNo: " << channel.LastSeqNo
333334
<< ", freeSpace: " << freeSpaceBytes);
334335

335336
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>();
336-
resp->Record.SetSeqNo(*producer.LastSeqNo);
337+
resp->Record.SetSeqNo(*channel.LastSeqNo);
337338
resp->Record.SetFreeSpace(freeSpaceBytes);
339+
resp->Record.SetChannelId(channelId);
338340

339-
ctx.Send(producerId, resp.Release());
341+
ctx.Send(channel.ActorId, resp.Release());
340342

341-
producer.AckedFreeSpaceBytes = freeSpaceBytes;
343+
channel.AckedFreeSpaceBytes = freeSpaceBytes;
342344
}
343345
}
344346

@@ -358,9 +360,10 @@ class TExecuteQueryRPC : public TActorBootstrapped<TExecuteQueryRPC> {
358360

359361
Request_->SendSerializedResult(std::move(out), Ydb::StatusIds::SUCCESS);
360362

361-
auto& producer = StreamProducers_[ev->Sender];
362-
producer.LastSeqNo = ev->Get()->Record.GetSeqNo();
363-
producer.AckedFreeSpaceBytes = freeSpaceBytes;
363+
auto& channel = StreamChannels_[ev->Get()->Record.GetChannelId()];
364+
channel.ActorId = ev->Sender;
365+
channel.LastSeqNo = ev->Get()->Record.GetSeqNo();
366+
channel.AckedFreeSpaceBytes = freeSpaceBytes;
364367

365368
LOG_DEBUG_S(ctx, NKikimrServices::RPC_REQUEST, this->SelfId() << "Send stream data ack"
366369
<< ", seqNo: " << ev->Get()->Record.GetSeqNo()
@@ -371,8 +374,9 @@ class TExecuteQueryRPC : public TActorBootstrapped<TExecuteQueryRPC> {
371374
auto resp = MakeHolder<NKqp::TEvKqpExecuter::TEvStreamDataAck>();
372375
resp->Record.SetSeqNo(ev->Get()->Record.GetSeqNo());
373376
resp->Record.SetFreeSpace(freeSpaceBytes);
377+
resp->Record.SetChannelId(ev->Get()->Record.GetChannelId());
374378

375-
ctx.Send(ev->Sender, resp.Release());
379+
ctx.Send(channel.ActorId, resp.Release());
376380
}
377381

378382
void Handle(NKqp::TEvKqp::TEvQueryResponse::TPtr& ev, const TActorContext&) {
@@ -381,14 +385,30 @@ class TExecuteQueryRPC : public TActorBootstrapped<TExecuteQueryRPC> {
381385
const auto& issueMessage = record.GetResponse().GetQueryIssues();
382386

383387
bool hasTrailingMessage = false;
384-
388+
389+
auto& kqpResponse = record.GetResponse();
390+
if (kqpResponse.GetYdbResults().size() > 1) {
391+
auto issue = MakeIssue(NKikimrIssues::TIssuesIds::DEFAULT_ERROR,
392+
"Unexpected trailing message with multiple result sets.");
393+
ReplyFinishStream(Ydb::StatusIds::INTERNAL_ERROR, issue);
394+
return;
395+
}
396+
385397
if (record.GetYdbStatus() == Ydb::StatusIds::SUCCESS) {
386398
Request_->SetRuHeader(record.GetConsumedRu());
387399

388400
auto& kqpResponse = record.GetResponse();
389401

390402
Ydb::Query::ExecuteQueryResponsePart response;
391403

404+
if (QueryAction == NKikimrKqp::QUERY_ACTION_EXECUTE) {
405+
for(int i = 0; i < kqpResponse.GetYdbResults().size(); i++) {
406+
hasTrailingMessage = true;
407+
response.set_result_set_index(i);
408+
response.mutable_result_set()->Swap(record.MutableResponse()->MutableYdbResults(i));
409+
}
410+
}
411+
392412
AuditContextAppend(Request_.get(), *Request_->GetProtoRequest(), response);
393413

394414
if (kqpResponse.HasTxMeta()) {
@@ -492,8 +512,9 @@ class TExecuteQueryRPC : public TActorBootstrapped<TExecuteQueryRPC> {
492512
private:
493513
std::shared_ptr<TEvExecuteQueryRequest> Request_;
494514

515+
NKikimrKqp::EQueryAction QueryAction;
495516
TRpcFlowControlState FlowControl_;
496-
TMap<TActorId, TProducerState> StreamProducers_;
517+
TMap<ui64, TProducerState> StreamChannels_;
497518
};
498519

499520
} // namespace

‎ydb/core/kqp/common/events/query.h‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,8 @@ struct TEvQueryRequest: public NActors::TEventLocal<TEvQueryRequest, TKqpEvents:
3434
const ::Ydb::Operations::OperationParams* operationParams,
3535
bool keepSession = false,
3636
bool useCancelAfter = true,
37-
const ::Ydb::Query::Syntax syntax = Ydb::Query::Syntax::SYNTAX_UNSPECIFIED);
37+
const ::Ydb::Query::Syntax syntax = Ydb::Query::Syntax::SYNTAX_UNSPECIFIED,
38+
bool supportsStreamTrailingResult = false);
3839

3940
TEvQueryRequest() = default;
4041

@@ -285,6 +286,10 @@ struct TEvQueryRequest: public NActors::TEventLocal<TEvQueryRequest, TKqpEvents:
285286
ProgressStatsPeriod = progressStatsPeriod;
286287
}
287288

289+
bool GetSupportsStreamTrailingResult() const {
290+
return SupportsStreamTrailingResult;
291+
}
292+
288293
TDuration GetProgressStatsPeriod() const {
289294
return ProgressStatsPeriod;
290295
}
@@ -317,6 +322,7 @@ struct TEvQueryRequest: public NActors::TEventLocal<TEvQueryRequest, TKqpEvents:
317322
const ::Ydb::Query::Syntax Syntax = Ydb::Query::Syntax::SYNTAX_UNSPECIFIED;
318323
TIntrusivePtr<TUserRequestContext> UserRequestContext;
319324
TDuration ProgressStatsPeriod;
325+
bool SupportsStreamTrailingResult = false;
320326
};
321327

322328
struct TEvDataQueryStreamPart: public TEventPB<TEvDataQueryStreamPart,

‎ydb/core/kqp/common/kqp_event_impl.cpp‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,8 @@ TEvKqp::TEvQueryRequest::TEvQueryRequest(
2020
const ::Ydb::Operations::OperationParams* operationParams,
2121
bool keepSession,
2222
bool useCancelAfter,
23-
const ::Ydb::Query::Syntax syntax)
23+
const ::Ydb::Query::Syntax syntax,
24+
bool supportsStreamTrailingResult)
2425
: RequestCtx(ctx)
2526
, RequestActorId(requestActorId)
2627
, Database(CanonizePath(ctx->GetDatabaseName().GetOrElse("")))
@@ -36,6 +37,7 @@ TEvKqp::TEvQueryRequest::TEvQueryRequest(
3637
, HasOperationParams(operationParams)
3738
, KeepSession(keepSession)
3839
, Syntax(syntax)
40+
, SupportsStreamTrailingResult(supportsStreamTrailingResult)
3941
{
4042
if (HasOperationParams) {
4143
OperationTimeout = GetDuration(operationParams->operation_timeout());

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

Lines changed: 5 additions & 40 deletions
Original file line numberDiff line numberDiff line change
@@ -130,10 +130,9 @@ class TKqpDataExecuter : public TKqpExecuterBase<TKqpDataExecuter, EExecType::Da
130130
const TActorId& creator, TDuration maximalSecretsSnapshotWaitTime, const TIntrusivePtr<TUserRequestContext>& userRequestContext,
131131
const bool enableOlapSink)
132132
: TBase(std::move(request), database, userToken, counters, executerRetriesConfig, chanTransportVersion, aggregation,
133-
maximalSecretsSnapshotWaitTime, userRequestContext, TWilsonKqp::DataExecuter, "DataExecuter"
133+
maximalSecretsSnapshotWaitTime, userRequestContext, TWilsonKqp::DataExecuter, "DataExecuter", streamResult
134134
)
135135
, AsyncIoFactory(std::move(asyncIoFactory))
136-
, StreamResult(streamResult)
137136
, EnableOlapSink(enableOlapSink)
138137
{
139138
Target = creator;
@@ -347,7 +346,8 @@ class TKqpDataExecuter : public TKqpExecuterBase<TKqpDataExecuter, EExecType::Da
347346
hFunc(TEvPersQueue::TEvProposeTransactionResult, HandlePrepare);
348347
hFunc(TEvPrivate::TEvReattachToShard, HandleExecute);
349348
hFunc(TEvDqCompute::TEvState, HandlePrepare); // from CA
350-
hFunc(TEvDqCompute::TEvChannelData, HandleExecute); // from CA
349+
hFunc(TEvDqCompute::TEvChannelData, HandleChannelData); // from CA
350+
hFunc(TEvKqpExecuter::TEvStreamDataAck, HandleStreamAck);
351351
hFunc(TEvPipeCache::TEvDeliveryProblem, HandlePrepare);
352352
hFunc(TEvKqp::TEvAbortExecution, HandlePrepare);
353353
hFunc(TEvents::TEvUndelivered, HandleUndelivered);
@@ -935,7 +935,8 @@ class TKqpDataExecuter : public TKqpExecuterBase<TKqpDataExecuter, EExecType::Da
935935
hFunc(TEvKqpNode::TEvStartKqpTasksResponse, HandleStartKqpTasksResponse);
936936
hFunc(TEvTxProxy::TEvProposeTransactionStatus, HandleExecute);
937937
hFunc(TEvDqCompute::TEvState, HandleComputeStats);
938-
hFunc(TEvDqCompute::TEvChannelData, HandleExecute);
938+
hFunc(NYql::NDq::TEvDqCompute::TEvChannelData, HandleChannelData);
939+
hFunc(TEvKqpExecuter::TEvStreamDataAck, HandleStreamAck);
939940
hFunc(TEvKqp::TEvAbortExecution, HandleExecute);
940941
IgnoreFunc(TEvInterconnect::TEvNodeConnected);
941942
default:
@@ -1286,41 +1287,6 @@ class TKqpDataExecuter : public TKqpExecuterBase<TKqpDataExecuter, EExecType::Da
12861287
}
12871288
}
12881289

1289-
void HandleExecute(TEvDqCompute::TEvChannelData::TPtr& ev) {
1290-
auto& record = ev->Get()->Record;
1291-
auto& channelData = record.GetChannelData();
1292-
1293-
TDqSerializedBatch batch;
1294-
batch.Proto = std::move(*record.MutableChannelData()->MutableData());
1295-
if (batch.Proto.HasPayloadId()) {
1296-
batch.Payload = ev->Get()->GetPayload(batch.Proto.GetPayloadId());
1297-
}
1298-
1299-
auto& channel = TasksGraph.GetChannel(channelData.GetChannelId());
1300-
YQL_ENSURE(channel.DstTask == 0);
1301-
auto shardId = TasksGraph.GetTask(channel.SrcTask).Meta.ShardId;
1302-
1303-
if (Stats) {
1304-
Stats->ResultBytes += batch.Size();
1305-
Stats->ResultRows += batch.RowCount();
1306-
}
1307-
1308-
LOG_T("Got result, channelId: " << channel.Id << ", shardId: " << shardId
1309-
<< ", inputIndex: " << channel.DstInputIndex << ", from: " << ev->Sender
1310-
<< ", finished: " << channelData.GetFinished());
1311-
1312-
ResponseEv->TakeResult(channel.DstInputIndex, std::move(batch));
1313-
{
1314-
LOG_T("Send ack to channelId: " << channel.Id << ", seqNo: " << record.GetSeqNo() << ", to: " << ev->Sender);
1315-
1316-
auto ackEv = MakeHolder<TEvDqCompute::TEvChannelDataAck>();
1317-
ackEv->Record.SetSeqNo(record.GetSeqNo());
1318-
ackEv->Record.SetChannelId(channel.Id);
1319-
ackEv->Record.SetFreeSpace(50_MB);
1320-
Send(ev->Sender, ackEv.Release(), /* TODO: undelivery */ 0, /* cookie */ channel.Id);
1321-
}
1322-
}
1323-
13241290
private:
13251291
bool IsReadOnlyTx() const {
13261292
if (Request.TopicOperations.HasOperations()) {
@@ -2417,7 +2383,6 @@ class TKqpDataExecuter : public TKqpExecuterBase<TKqpDataExecuter, EExecType::Da
24172383

24182384
private:
24192385
NYql::NDq::IDqAsyncIoFactory::TPtr AsyncIoFactory;
2420-
bool StreamResult = false;
24212386
bool EnableOlapSink = false;
24222387

24232388
bool HasExternalSources = false;

0 commit comments

Comments
 (0)