Skip to content

Commit e327f9d

Browse files
ssmikessmike
authored
Merge 6522bab into c3a716e 
2 parents c3a716e + 6522bab commit e327f9d

16 files changed

+752
-40
lines changed

‎ydb/core/kqp/common/simple/kqp_event_ids.h‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -44,7 +44,8 @@ struct TKqpEvents {
4444
EvListSessionsRequest,
4545
EvListSessionsResponse,
4646
EvListProxyNodesRequest,
47-
EvListProxyNodesResponse
47+
EvListProxyNodesResponse,
48+
EvFinishKqpTasks
4849
};
4950

5051
static_assert (EvCompileInvalidateRequest + 1 == EvAbortExecution);

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

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -134,8 +134,8 @@ using namespace NYql::NDqProto;
134134
IActor* CreateKqpScanComputeActor(const TActorId& executerId, ui64 txId,
135135
TDqTask* task, IDqAsyncIoFactory::TPtr asyncIoFactory,
136136
const NYql::NDq::TComputeRuntimeSettings& settings, const TComputeMemoryLimits& memoryLimits, NWilson::TTraceId traceId,
137-
TIntrusivePtr<NActors::TProtoArenaHolder> arena) {
138-
return new NScanPrivate::TKqpScanComputeActor(executerId, txId, task, std::move(asyncIoFactory),
137+
TIntrusivePtr<NActors::TProtoArenaHolder> arena, TComputeActorSchedulingOptions schedulingOptions) {
138+
return new NScanPrivate::TKqpScanComputeActor(schedulingOptions, executerId, txId, task, std::move(asyncIoFactory),
139139
settings, memoryLimits, std::move(traceId), std::move(arena));
140140
}
141141

‎ydb/core/kqp/compute_actor/kqp_compute_actor.h‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
#include <ydb/core/kqp/compute_actor/kqp_compute_events.h>
44
#include <ydb/core/kqp/counters/kqp_counters.h>
55
#include <ydb/core/kqp/federated_query/kqp_federated_query_helpers.h>
6+
#include <ydb/core/kqp/runtime/kqp_compute_scheduler.h>
67
#include <ydb/core/scheme/scheme_tabledefs.h>
78
#include <ydb/library/yql/dq/actors/compute/dq_compute_actor.h>
89
#include <ydb/library/yql/dq/actors/compute/dq_compute_actor_async_io_factory.h>
@@ -47,12 +48,12 @@ IActor* CreateKqpComputeActor(const TActorId& executerId, ui64 txId, NYql::NDqPr
4748
const NYql::NDq::TComputeRuntimeSettings& settings, const NYql::NDq::TComputeMemoryLimits& memoryLimits,
4849
NWilson::TTraceId traceId,
4950
TIntrusivePtr<NActors::TProtoArenaHolder> arena,
50-
const std::optional<TKqpFederatedQuerySetup>& federatedQuerySetup, const TGUCSettings::TPtr& GUCSettings);
51+
const std::optional<TKqpFederatedQuerySetup>& federatedQuerySetup, const TGUCSettings::TPtr& GUCSettings, TComputeActorSchedulingOptions);
5152

5253
IActor* CreateKqpScanComputeActor(const TActorId& executerId, ui64 txId,
5354
NYql::NDqProto::TDqTask* task, NYql::NDq::IDqAsyncIoFactory::TPtr asyncIoFactory,
5455
const NYql::NDq::TComputeRuntimeSettings& settings, const NYql::NDq::TComputeMemoryLimits& memoryLimits, NWilson::TTraceId traceId,
55-
TIntrusivePtr<NActors::TProtoArenaHolder> arena);
56+
TIntrusivePtr<NActors::TProtoArenaHolder> arena, TComputeActorSchedulingOptions);
5657

5758
IActor* CreateKqpScanFetcher(const NKikimrKqp::TKqpSnapshot& snapshot, std::vector<NActors::TActorId>&& computeActors,
5859
const NKikimrTxDataShard::TKqpTransaction::TScanTaskMeta& meta, const NYql::NDq::TComputeRuntimeSettings& settings,

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

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -14,8 +14,8 @@ TKqpComputeActor::TKqpComputeActor(const TActorId& executerId, ui64 txId, NDqPro
1414
IDqAsyncIoFactory::TPtr asyncIoFactory,
1515
const TComputeRuntimeSettings& settings, const TComputeMemoryLimits& memoryLimits,
1616
NWilson::TTraceId traceId, TIntrusivePtr<NActors::TProtoArenaHolder> arena,
17-
const std::optional<TKqpFederatedQuerySetup>& federatedQuerySetup, const TGUCSettings::TPtr& GUCSettings)
18-
: TBase(executerId, txId, task, std::move(asyncIoFactory), settings, memoryLimits, /* ownMemoryQuota = */ true, /* passExceptions = */ true, /*taskCounters = */ nullptr, std::move(traceId), std::move(arena), GUCSettings)
17+
const std::optional<TKqpFederatedQuerySetup>& federatedQuerySetup, const TGUCSettings::TPtr& GUCSettings, TComputeActorSchedulingOptions schedulingOptions)
18+
: TBase(schedulingOptions, executerId, txId, task, std::move(asyncIoFactory), settings, memoryLimits, /* ownMemoryQuota = */ true, /* passExceptions = */ true, /*taskCounters = */ nullptr, std::move(traceId), std::move(arena), GUCSettings)
1919
, ComputeCtx(settings.StatsMode)
2020
, FederatedQuerySetup(federatedQuerySetup)
2121
{
@@ -121,9 +121,11 @@ void TKqpComputeActor::DoBootstrap() {
121121

122122
ContinueExecute();
123123
Become(&TKqpComputeActor::StateFunc);
124+
124125
}
125126

126127
STFUNC(TKqpComputeActor::StateFunc) {
128+
CA_LOG_D("CA StateFunc " << ev->GetTypeRewrite());
127129
try {
128130
switch (ev->GetTypeRewrite()) {
129131
hFunc(TEvKqpCompute::TEvScanInitActor, HandleExecute);
@@ -281,10 +283,10 @@ IActor* CreateKqpComputeActor(const TActorId& executerId, ui64 txId, NDqProto::T
281283
const TComputeRuntimeSettings& settings, const TComputeMemoryLimits& memoryLimits,
282284
NWilson::TTraceId traceId, TIntrusivePtr<NActors::TProtoArenaHolder> arena,
283285
const std::optional<TKqpFederatedQuerySetup>& federatedQuerySetup,
284-
const TGUCSettings::TPtr& GUCSettings)
286+
const TGUCSettings::TPtr& GUCSettings, TComputeActorSchedulingOptions cpuOptions)
285287
{
286288
return new TKqpComputeActor(executerId, txId, task, std::move(asyncIoFactory),
287-
settings, memoryLimits, std::move(traceId), std::move(arena), federatedQuerySetup, GUCSettings);
289+
settings, memoryLimits, std::move(traceId), std::move(arena), federatedQuerySetup, GUCSettings, cpuOptions);
288290
}
289291

290292
} // namespace NKqp

‎ydb/core/kqp/compute_actor/kqp_pure_compute_actor.h‎

Lines changed: 7 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -8,17 +8,16 @@
88
#include <ydb/core/kqp/rm_service/kqp_rm_service.h>
99
#include <ydb/core/kqp/runtime/kqp_compute.h>
1010
#include <ydb/core/kqp/runtime/kqp_scan_data.h>
11+
#include <ydb/core/kqp/runtime/kqp_compute_scheduler.h>
1112
#include <ydb/core/sys_view/scan.h>
1213
#include <ydb/library/yverify_stream/yverify_stream.h>
1314

14-
#include <ydb/library/yql/dq/actors/compute/dq_sync_compute_actor_base.h>
15-
1615

1716
namespace NKikimr {
1817
namespace NKqp {
1918

20-
class TKqpComputeActor : public TDqSyncComputeActorBase<TKqpComputeActor> {
21-
using TBase = TDqSyncComputeActorBase<TKqpComputeActor>;
19+
class TKqpComputeActor : public TSchedulableComputeActorBase<TKqpComputeActor> {
20+
using TBase = TSchedulableComputeActorBase<TKqpComputeActor>;
2221

2322
public:
2423
static constexpr NKikimrServices::TActivity::EType ActorActivityType() {
@@ -29,7 +28,8 @@ class TKqpComputeActor : public TDqSyncComputeActorBase<TKqpComputeActor> {
2928
IDqAsyncIoFactory::TPtr asyncIoFactory,
3029
const TComputeRuntimeSettings& settings, const TComputeMemoryLimits& memoryLimits,
3130
NWilson::TTraceId traceId, TIntrusivePtr<NActors::TProtoArenaHolder> arena,
32-
const std::optional<TKqpFederatedQuerySetup>& federatedQuerySetup, const TGUCSettings::TPtr& GUCSettings);
31+
const std::optional<TKqpFederatedQuerySetup>& federatedQuerySetup, const TGUCSettings::TPtr& GUCSettings,
32+
TComputeActorSchedulingOptions);
3333

3434
void DoBootstrap();
3535

@@ -68,7 +68,8 @@ IActor* CreateKqpComputeActor(const TActorId& executerId, ui64 txId, NDqProto::T
6868
IDqAsyncIoFactory::TPtr asyncIoFactory,
6969
const TComputeRuntimeSettings& settings, const TComputeMemoryLimits& memoryLimits,
7070
NWilson::TTraceId traceId, TIntrusivePtr<NActors::TProtoArenaHolder> arena,
71-
const std::optional<TKqpFederatedQuerySetup>& federatedQuerySetup, const TGUCSettings::TPtr& GUCSettings);
71+
const std::optional<TKqpFederatedQuerySetup>& federatedQuerySetup, const TGUCSettings::TPtr& GUCSettings,
72+
TComputeActorSchedulingOptions);
7273

7374
} // namespace NKqp
7475
} // namespace NKikimr

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

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -23,11 +23,11 @@ static constexpr TDuration RL_MAX_BATCH_DELAY = TDuration::Seconds(50);
2323

2424
} // anonymous namespace
2525

26-
TKqpScanComputeActor::TKqpScanComputeActor(const TActorId& executerId, ui64 txId, NDqProto::TDqTask* task,
26+
TKqpScanComputeActor::TKqpScanComputeActor(TComputeActorSchedulingOptions cpuOptions, const TActorId& executerId, ui64 txId, NDqProto::TDqTask* task,
2727
IDqAsyncIoFactory::TPtr asyncIoFactory,
2828
const TComputeRuntimeSettings& settings, const TComputeMemoryLimits& memoryLimits, NWilson::TTraceId traceId,
2929
TIntrusivePtr<NActors::TProtoArenaHolder> arena)
30-
: TBase(executerId, txId, task, std::move(asyncIoFactory), settings,
30+
: TBase(cpuOptions, executerId, txId, task, std::move(asyncIoFactory), settings,
3131
memoryLimits, /* ownMemoryQuota = */ true, /* passExceptions = */ true, /*taskCounters = */ nullptr, std::move(traceId), std::move(arena))
3232
, ComputeCtx(settings.StatsMode)
3333
{

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

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -2,15 +2,15 @@
22
#include "kqp_scan_events.h"
33

44
#include <ydb/core/kqp/runtime/kqp_scan_data.h>
5-
#include <ydb/library/yql/dq/actors/compute/dq_sync_compute_actor_base.h>
5+
#include <ydb/core/kqp/runtime/kqp_compute_scheduler.h>
66
#include <ydb/library/yql/dq/actors/compute/dq_compute_actor_async_io.h>
77
#include <ydb/library/yql/dq/actors/compute/dq_compute_actor.h>
88

99
namespace NKikimr::NKqp::NScanPrivate {
1010

11-
class TKqpScanComputeActor: public NYql::NDq::TDqSyncComputeActorBase<TKqpScanComputeActor> {
11+
class TKqpScanComputeActor: public TSchedulableComputeActorBase<TKqpScanComputeActor> {
1212
private:
13-
using TBase = NYql::NDq::TDqSyncComputeActorBase<TKqpScanComputeActor>;
13+
using TBase = TSchedulableComputeActorBase<TKqpScanComputeActor>;
1414
NMiniKQL::TKqpScanComputeContext ComputeCtx;
1515
NKikimrTxDataShard::TKqpTransaction::TScanTaskMeta Meta;
1616
using TBase::TaskRunner;
@@ -30,7 +30,7 @@ class TKqpScanComputeActor: public NYql::NDq::TDqSyncComputeActorBase<TKqpScanCo
3030
return NKikimrServices::TActivity::KQP_SCAN_COMPUTE_ACTOR;
3131
}
3232

33-
TKqpScanComputeActor(const TActorId& executerId, ui64 txId,
33+
TKqpScanComputeActor(TComputeActorSchedulingOptions, const TActorId& executerId, ui64 txId,
3434
NYql::NDqProto::TDqTask* task, NYql::NDq::IDqAsyncIoFactory::TPtr asyncIoFactory,
3535
const NYql::NDq::TComputeRuntimeSettings& settings, const NYql::NDq::TComputeMemoryLimits& memoryLimits, NWilson::TTraceId traceId,
3636
TIntrusivePtr<NActors::TProtoArenaHolder> arena);

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -359,7 +359,7 @@ void TKqpPlanner::ExecuteDataComputeTask(ui64 taskId, bool shareMailbox, bool op
359359
limits.MemoryQuotaManager = std::make_shared<NYql::NDq::TGuaranteeQuotaManager>(limit * 2, limit);
360360

361361
auto computeActor = NKikimr::NKqp::CreateKqpComputeActor(ExecuterId, TxId, taskDesc, AsyncIoFactory,
362-
settings, limits, ExecuterSpan.GetTraceId(), TasksGraph.GetMeta().GetArenaIntrusivePtr(), FederatedQuerySetup, GUCSettings);
362+
settings, limits, ExecuterSpan.GetTraceId(), TasksGraph.GetMeta().GetArenaIntrusivePtr(), FederatedQuerySetup, GUCSettings, {});
363363

364364
if (optimizeProtoForLocalExecution) {
365365
TVector<google::protobuf::Message*>& taskSourceSettings = static_cast<TKqpComputeActor*>(computeActor)->MutableTaskSourceSettings();

‎ydb/core/kqp/node_service/kqp_node_service.cpp‎

Lines changed: 61 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@
1414
#include <ydb/core/kqp/rm_service/kqp_rm_service.h>
1515
#include <ydb/core/kqp/runtime/kqp_read_actor.h>
1616
#include <ydb/core/kqp/runtime/kqp_read_iterator_common.h>
17+
#include <ydb/core/kqp/runtime/kqp_compute_scheduler.h>
1718
#include <ydb/core/kqp/common/kqp_resolve.h>
1819

1920
#include <ydb/library/wilson_ids/wilson.h>
@@ -150,6 +151,9 @@ class TKqpNodeService : public TActorBootstrapped<TKqpNodeService> {
150151
if (config.HasIteratorReadQuotaSettings()) {
151152
SetIteratorReadsQuotaSettings(config.GetIteratorReadQuotaSettings());
152153
}
154+
if (config.HasComputePoolsConfiguration()) {
155+
SetPriorities(config.GetComputePoolsConfiguration());
156+
}
153157
}
154158

155159
void Bootstrap() {
@@ -174,9 +178,12 @@ class TKqpNodeService : public TActorBootstrapped<TKqpNodeService> {
174178

175179
private:
176180
STATEFN(WorkState) {
181+
Y_DEFER {
182+
Scheduler.AdvanceTime(TlsActivationContext->Monotonic());
183+
};
177184
switch (ev->GetTypeRewrite()) {
178185
hFunc(TEvKqpNode::TEvStartKqpTasksRequest, HandleWork);
179-
hFunc(TEvKqpNode::TEvFinishKqpTask, HandleWork); // used only for unit tests
186+
hFunc(TEvFinishKqpTask, HandleWork); // used only for unit tests
180187
hFunc(TEvKqpNode::TEvCancelKqpTasksRequest, HandleWork);
181188
hFunc(TEvents::TEvWakeup, HandleWork);
182189
// misc
@@ -385,6 +392,8 @@ class TKqpNodeService : public TActorBootstrapped<TKqpNodeService> {
385392

386393
TComputeStagesWithScan computesByStage;
387394

395+
auto now = TlsActivationContext->Monotonic();
396+
388397
// start compute actors
389398
for (int i = 0; i < msg.GetTasks().size(); ++i) {
390399
auto& dqTask = *msg.MutableTasks(i);
@@ -444,12 +453,27 @@ class TKqpNodeService : public TActorBootstrapped<TKqpNodeService> {
444453
tableKind = tableKindExtract(meta);
445454
}
446455

456+
TComputeActorSchedulingOptions schedulingOptions {
457+
.Now = now,
458+
.NodeService = SelfId(),
459+
.Scheduler = &Scheduler,
460+
.Group = msg.GetRuntimeSettings().GetExecType() == NYql::NDqProto::TComputeRuntimeSettings::SCAN ? "olap" : "",
461+
.Weight = 1,
462+
.NoThrottle = false,
463+
};
464+
465+
if (msg.GetRuntimeSettings().GetExecType() == NYql::NDqProto::TComputeRuntimeSettings::DATA) {
466+
schedulingOptions.NoThrottle = true;
467+
schedulingOptions.Scheduler = nullptr;
468+
}
469+
447470
IActor* computeActor;
448471
if (tableKind == ETableKind::Datashard || tableKind == ETableKind::Olap) {
449472
auto& info = computesByStage.UpsertTaskWithScan(dqTask, meta, !AppData()->FeatureFlags.GetEnableSeparationComputeActorsFromRead());
450473
computeActor = CreateKqpScanComputeActor(request.Executer, txId, &dqTask,
451474
AsyncIoFactory, runtimeSettings, memoryLimits,
452-
NWilson::TTraceId(ev->TraceId), ev->Get()->Arena);
475+
NWilson::TTraceId(ev->TraceId), ev->Get()->Arena,
476+
schedulingOptions);
453477
taskCtx.ComputeActorId = Register(computeActor);
454478
info.MutableActorIds().emplace_back(taskCtx.ComputeActorId);
455479
} else {
@@ -459,7 +483,8 @@ class TKqpNodeService : public TActorBootstrapped<TKqpNodeService> {
459483
}
460484
if (Y_LIKELY(!CaFactory)) {
461485
computeActor = CreateKqpComputeActor(request.Executer, txId, &dqTask, AsyncIoFactory,
462-
runtimeSettings, memoryLimits, NWilson::TTraceId(ev->TraceId), ev->Get()->Arena, FederatedQuerySetup, GUCSettings);
486+
runtimeSettings, memoryLimits, NWilson::TTraceId(ev->TraceId), ev->Get()->Arena, FederatedQuerySetup, GUCSettings,
487+
schedulingOptions);
463488
taskCtx.ComputeActorId = Register(computeActor);
464489
} else {
465490
computeActor = CaFactory->CreateKqpComputeActor(request.Executer, txId, &dqTask,
@@ -490,8 +515,11 @@ class TKqpNodeService : public TActorBootstrapped<TKqpNodeService> {
490515
}
491516

492517
// used only for unit tests
493-
void HandleWork(TEvKqpNode::TEvFinishKqpTask::TPtr& ev) {
518+
void HandleWork(TEvFinishKqpTask::TPtr& ev) {
494519
auto& msg = *ev->Get();
520+
if (msg.SchedulerEntity) {
521+
Scheduler.Deregister(*msg.SchedulerEntity);
522+
}
495523
FinishKqpTask(msg.TxId, msg.TaskId, msg.Success, GetStateBucketByTx(Buckets, msg.TxId), GetKqpResourceManager());
496524
}
497525

@@ -568,6 +596,10 @@ class TKqpNodeService : public TActorBootstrapped<TKqpNodeService> {
568596
SetIteratorReadsQuotaSettings(event.GetConfig().GetTableServiceConfig().GetIteratorReadQuotaSettings());
569597
}
570598

599+
if (event.GetConfig().GetTableServiceConfig().HasComputePoolsConfiguration()) {
600+
SetPriorities(event.GetConfig().GetTableServiceConfig().GetComputePoolsConfiguration());
601+
}
602+
571603
auto responseEv = MakeHolder<NConsole::TEvConsole::TEvConfigNotificationResponse>(event);
572604
Send(ev->Sender, responseEv.Release(), IEventHandle::FlagTrackDelivery, ev->Cookie);
573605
}
@@ -576,6 +608,29 @@ class TKqpNodeService : public TActorBootstrapped<TKqpNodeService> {
576608
SetDefaultIteratorQuotaSettings(settings.GetMaxRows(), settings.GetMaxBytes());
577609
}
578610

611+
void SetPriorities(const NKikimrConfig::TTableServiceConfig::TComputePoolConfiguration& conf) {
612+
std::function<TComputeScheduler::TDistributionRule(const NKikimrConfig::TTableServiceConfig::TComputePoolConfiguration&)> convert
613+
= [&](const NKikimrConfig::TTableServiceConfig::TComputePoolConfiguration& conf)
614+
{
615+
if (conf.HasName()) {
616+
return TComputeScheduler::TDistributionRule{.Share = conf.GetMaxCpuShare(), .Name = conf.GetName()};
617+
} else if (conf.HasSubPoolsConfiguration()) {
618+
auto res = TComputeScheduler::TDistributionRule{.Share = conf.GetMaxCpuShare()};
619+
for (auto& subConf : conf.GetSubPoolsConfiguration().GetSubPools()) {
620+
res.SubRules.push_back(convert(subConf));
621+
}
622+
return res;
623+
} else {
624+
Y_ENSURE(false, "unknown case");
625+
}
626+
};
627+
auto converted = convert(conf);
628+
629+
auto threads = TlsActivationContext->ActorSystem()->GetPoolThreadsCount(SelfId().PoolID());
630+
Y_ENSURE(threads.has_value());
631+
Scheduler.SetPriorities(converted.empty() ? TComputeScheduler::TDistributionRule{.Share = 1, .Name = "olap"} : converted, threads.value(), TlsActivationContext->Monotonic());
632+
}
633+
579634
void SetIteratorReadsRetrySettings(const NKikimrConfig::TTableServiceConfig::TIteratorReadsRetrySettings& settings) {
580635
auto ptr = MakeIntrusive<NKikimr::NKqp::TIteratorReadBackoffSettings>();
581636
ptr->StartRetryDelay = TDuration::MilliSeconds(settings.GetStartDelayMs());
@@ -670,6 +725,8 @@ class TKqpNodeService : public TActorBootstrapped<TKqpNodeService> {
670725
NYql::NDq::IDqAsyncIoFactory::TPtr AsyncIoFactory;
671726
const std::optional<TKqpFederatedQuerySetup> FederatedQuerySetup;
672727

728+
TComputeScheduler Scheduler;
729+
673730
//state sharded by TxId
674731
std::shared_ptr<TBucketArray> Buckets;
675732
};

‎ydb/core/kqp/node_service/kqp_node_service.h‎

Lines changed: 0 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,6 @@ struct TKqpNodeEvents {
2626
enum EKqpNodeEvents {
2727
EvStartKqpTasksRequest = EventSpaceBegin(TKikimrEvents::ES_KQP) + 320,
2828
EvStartKqpTasksResponse,
29-
EvFinishKqpTasks,
3029
EvCancelKqpTasksRequest,
3130
EvCancelKqpTasksResponse,
3231
};
@@ -46,19 +45,6 @@ struct TEvKqpNode {
4645
struct TEvStartKqpTasksResponse : public TEventPB<TEvStartKqpTasksResponse,
4746
NKikimrKqp::TEvStartKqpTasksResponse, TKqpNodeEvents::EvStartKqpTasksResponse> {};
4847

49-
struct TEvFinishKqpTask : public TEventLocal<TEvFinishKqpTask, TKqpNodeEvents::EvFinishKqpTasks> {
50-
const ui64 TxId;
51-
const ui64 TaskId;
52-
const bool Success;
53-
const NYql::TIssues Issues;
54-
55-
TEvFinishKqpTask(ui64 txId, ui64 taskId, bool success, const NYql::TIssues& issues = {})
56-
: TxId(txId)
57-
, TaskId(taskId)
58-
, Success(success)
59-
, Issues(issues) {}
60-
};
61-
6248
struct TEvCancelKqpTasksRequest : public TEventPB<TEvCancelKqpTasksRequest,
6349
NKikimrKqp::TEvCancelKqpTasksRequest, TKqpNodeEvents::EvCancelKqpTasksRequest> {};
6450

0 commit comments

Comments
 (0)