Skip to content

Commit 8e7ec08

Browse files
GrigoriyPAGrigoriyPA
authored
Refactor lazy table creation for WM service (#4900)
1 parent d36c6ff commit 8e7ec08

File tree

8 files changed

+318
-174
lines changed

8 files changed

+318
-174
lines changed

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

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -355,4 +355,14 @@ struct TEvDescribeSecretsResponse : public NActors::TEventLocal<TEvDescribeSecre
355355
TDescription Description;
356356
};
357357

358+
struct TEvScriptExecutionsTablesCreationFinished : public NActors::TEventLocal<TEvScriptExecutionsTablesCreationFinished, TKqpScriptExecutionEvents::EvScriptExecutionsTableCreationFinished> {
359+
TEvScriptExecutionsTablesCreationFinished(bool success, NYql::TIssues issues)
360+
: Success(success)
361+
, Issues(std::move(issues))
362+
{}
363+
364+
const bool Success;
365+
const NYql::TIssues Issues;
366+
};
367+
358368
} // namespace NKikimr::NKqp

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -157,6 +157,7 @@ struct TKqpScriptExecutionEvents {
157157
EvGetScriptExecutionOperationQueryResponse,
158158
EvDescribeSecretsResponse,
159159
EvSaveScriptResultPartFinished,
160+
EvScriptExecutionsTableCreationFinished,
160161
};
161162
};
162163

‎ydb/core/kqp/proxy_service/kqp_proxy_service.cpp‎

Lines changed: 30 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -136,7 +136,6 @@ class TKqpProxyService : public TActorBootstrapped<TKqpProxyService> {
136136
EvOnRequestTimeout,
137137
EvCloseIdleSessions,
138138
EvResourcesSnapshot,
139-
EvScriptExecutionsTableCreationFinished,
140139
};
141140

142141
struct TEvReadyToPublishResources : public TEventLocal<TEvReadyToPublishResources, EEv::EvReadyToPublishResources> {};
@@ -169,10 +168,6 @@ class TKqpProxyService : public TActorBootstrapped<TKqpProxyService> {
169168
TEvResourcesSnapshot(TVector<NKikimrKqp::TKqpNodeResources>&& snapshot)
170169
: Snapshot(std::move(snapshot)) {}
171170
};
172-
173-
struct TEvScriptExecutionsTablesCreationFinished : public NActors::TEventLocal<TEvScriptExecutionsTablesCreationFinished, EvScriptExecutionsTableCreationFinished> {
174-
TEvScriptExecutionsTablesCreationFinished() = default;
175-
};
176171
};
177172

178173
public:
@@ -1320,7 +1315,7 @@ class TKqpProxyService : public TActorBootstrapped<TKqpProxyService> {
13201315
hFunc(NNodeWhiteboard::TEvWhiteboard::TEvSystemStateResponse, Handle);
13211316
hFunc(TEvKqp::TEvCreateSessionResponse, ForwardEvent);
13221317
hFunc(TEvPrivate::TEvCloseIdleSessions, Handle);
1323-
hFunc(TEvPrivate::TEvScriptExecutionsTablesCreationFinished, Handle);
1318+
hFunc(TEvScriptExecutionsTablesCreationFinished, Handle);
13241319
hFunc(NKqp::TEvForgetScriptExecutionOperation, Handle);
13251320
hFunc(NKqp::TEvGetScriptExecutionOperation, Handle);
13261321
hFunc(NKqp::TEvListScriptExecutionOperations, Handle);
@@ -1582,11 +1577,16 @@ class TKqpProxyService : public TActorBootstrapped<TKqpProxyService> {
15821577
switch (ScriptExecutionsCreationStatus) {
15831578
case EScriptExecutionsCreationStatus::NotStarted:
15841579
ScriptExecutionsCreationStatus = EScriptExecutionsCreationStatus::Pending;
1585-
Register(CreateScriptExecutionsTablesCreator(MakeHolder<TEvPrivate::TEvScriptExecutionsTablesCreationFinished>()), TMailboxType::HTSwap, AppData()->SystemPoolId);
1580+
Register(CreateScriptExecutionsTablesCreator(), TMailboxType::HTSwap, AppData()->SystemPoolId);
15861581
[[fallthrough]];
15871582
case EScriptExecutionsCreationStatus::Pending:
15881583
if (DelayedEventsQueue.size() < 10000) {
1589-
DelayedEventsQueue.emplace_back(std::move(ev));
1584+
DelayedEventsQueue.push_back({
1585+
.Event = std::move(ev),
1586+
.ResponseBuilder = [](Ydb::StatusIds::StatusCode status, NYql::TIssues issues) {
1587+
return new TResponse(status, std::move(issues));
1588+
}
1589+
});
15901590
} else {
15911591
NYql::TIssues issues;
15921592
issues.AddIssue("Too many queued requests");
@@ -1598,10 +1598,25 @@ class TKqpProxyService : public TActorBootstrapped<TKqpProxyService> {
15981598
}
15991599
}
16001600

1601-
void Handle(TEvPrivate::TEvScriptExecutionsTablesCreationFinished::TPtr&) {
1601+
void Handle(TEvScriptExecutionsTablesCreationFinished::TPtr& ev) {
16021602
ScriptExecutionsCreationStatus = EScriptExecutionsCreationStatus::Finished;
1603+
1604+
NYql::TIssue rootIssue;
1605+
if (!ev->Get()->Success) {
1606+
ScriptExecutionsCreationStatus = EScriptExecutionsCreationStatus::NotStarted;
1607+
rootIssue.SetMessage("Failed to create script execution tables");
1608+
for (const NYql::TIssue& issue : ev->Get()->Issues) {
1609+
rootIssue.AddSubIssue(MakeIntrusive<NYql::TIssue>(issue));
1610+
}
1611+
}
1612+
16031613
while (!DelayedEventsQueue.empty()) {
1604-
Send(std::move(DelayedEventsQueue.front()));
1614+
auto delayedEvent = std::move(DelayedEventsQueue.front());
1615+
if (ev->Get()->Success) {
1616+
Send(std::move(delayedEvent.Event));
1617+
} else {
1618+
Send(delayedEvent.Event->Sender, delayedEvent.ResponseBuilder(Ydb::StatusIds::INTERNAL_ERROR, {rootIssue}));
1619+
}
16051620
DelayedEventsQueue.pop_front();
16061621
}
16071622
}
@@ -1765,8 +1780,12 @@ class TKqpProxyService : public TActorBootstrapped<TKqpProxyService> {
17651780
Pending,
17661781
Finished,
17671782
};
1783+
struct TDelayedEvent {
1784+
THolder<IEventHandle> Event;
1785+
std::function<IEventBase*(Ydb::StatusIds::StatusCode, NYql::TIssues)> ResponseBuilder;
1786+
};
17681787
EScriptExecutionsCreationStatus ScriptExecutionsCreationStatus = EScriptExecutionsCreationStatus::NotStarted;
1769-
std::deque<THolder<IEventHandle>> DelayedEventsQueue;
1788+
std::deque<TDelayedEvent> DelayedEventsQueue;
17701789
bool IsLookupByRmScheduled = false;
17711790
TActorId KqpTempTablesAgentActor;
17721791
};

‎ydb/core/kqp/proxy_service/kqp_script_executions.cpp‎

Lines changed: 75 additions & 123 deletions
Original file line numberDiff line numberDiff line change
@@ -79,140 +79,92 @@ class TQueryBase : public NKikimr::TQueryBase {
7979
};
8080

8181

82-
class TScriptExecutionsTablesCreator : public TActorBootstrapped<TScriptExecutionsTablesCreator> {
83-
public:
84-
explicit TScriptExecutionsTablesCreator(THolder<NActors::IEventBase> resultEvent)
85-
: ResultEvent(std::move(resultEvent))
86-
{
87-
}
88-
89-
void Registered(NActors::TActorSystem* sys, const NActors::TActorId& owner) override {
90-
NActors::TActorBootstrapped<TScriptExecutionsTablesCreator>::Registered(sys, owner);
91-
Owner = owner;
92-
}
82+
class TScriptExecutionsTablesCreator : public NTableCreator::TMultiTableCreator {
83+
using TBase = NTableCreator::TMultiTableCreator;
9384

94-
void Bootstrap() {
95-
Become(&TScriptExecutionsTablesCreator::StateFunc);
96-
RunCreateScriptExecutions();
97-
RunCreateScriptExecutionLeases();
98-
RunCreateScriptResultSets();
99-
}
85+
public:
86+
explicit TScriptExecutionsTablesCreator()
87+
: TBase({
88+
GetScriptExecutionsCreator(),
89+
GetScriptExecutionLeasesCreator(),
90+
GetScriptResultSetsCreator()
91+
})
92+
{}
10093

10194
private:
102-
static NKikimrSchemeOp::TColumnDescription Col(const TString& columnName, const char* columnType) {
103-
NKikimrSchemeOp::TColumnDescription desc;
104-
desc.SetName(columnName);
105-
desc.SetType(columnType);
106-
return desc;
107-
}
108-
109-
static NKikimrSchemeOp::TColumnDescription Col(const TString& columnName, NScheme::TTypeId columnType) {
110-
return Col(columnName, NScheme::TypeName(columnType));
111-
}
112-
113-
static NKikimrSchemeOp::TTTLSettings TtlCol(const TString& columnName) {
114-
NKikimrSchemeOp::TTTLSettings settings;
115-
settings.MutableEnabled()->SetExpireAfterSeconds(DEADLINE_OFFSET.Seconds());
116-
settings.MutableEnabled()->SetColumnName(columnName);
117-
settings.MutableEnabled()->MutableSysSettings()->SetRunInterval(BRO_RUN_INTERVAL.MicroSeconds());
118-
return settings;
119-
}
120-
121-
void RunCreateScriptExecutions() {
122-
TablesCreating++;
123-
Register(
124-
CreateTableCreator(
125-
{ ".metadata", "script_executions" },
126-
{
127-
Col("database", NScheme::NTypeIds::Text),
128-
Col("execution_id", NScheme::NTypeIds::Text),
129-
Col("run_script_actor_id", NScheme::NTypeIds::Text),
130-
Col("operation_status", NScheme::NTypeIds::Int32),
131-
Col("execution_status", NScheme::NTypeIds::Int32),
132-
Col("finalization_status", NScheme::NTypeIds::Int32),
133-
Col("execution_mode", NScheme::NTypeIds::Int32),
134-
Col("start_ts", NScheme::NTypeIds::Timestamp),
135-
Col("end_ts", NScheme::NTypeIds::Timestamp),
136-
Col("query_text", NScheme::NTypeIds::Text),
137-
Col("syntax", NScheme::NTypeIds::Int32),
138-
Col("ast", NScheme::NTypeIds::Text),
139-
Col("ast_compressed", NScheme::NTypeIds::String),
140-
Col("ast_compression_method", NScheme::NTypeIds::Text),
141-
Col("issues", NScheme::NTypeIds::JsonDocument),
142-
Col("plan", NScheme::NTypeIds::JsonDocument),
143-
Col("meta", NScheme::NTypeIds::JsonDocument),
144-
Col("parameters", NScheme::NTypeIds::String), // TODO: store aparameters separately to support bigger storage.
145-
Col("result_set_metas", NScheme::NTypeIds::JsonDocument),
146-
Col("stats", NScheme::NTypeIds::JsonDocument),
147-
Col("expire_at", NScheme::NTypeIds::Timestamp), // Will be deleted from database after this deadline.
148-
Col("customer_supplied_id", NScheme::NTypeIds::Text),
149-
Col("user_token", NScheme::NTypeIds::Text),
150-
Col("script_sinks", NScheme::NTypeIds::JsonDocument),
151-
Col("script_secret_names", NScheme::NTypeIds::JsonDocument),
152-
},
153-
{ "database", "execution_id" },
154-
NKikimrServices::KQP_PROXY,
155-
TtlCol("expire_at")
156-
)
95+
static IActor* GetScriptExecutionsCreator() {
96+
return CreateTableCreator(
97+
{ ".metadata", "script_executions" },
98+
{
99+
Col("database", NScheme::NTypeIds::Text),
100+
Col("execution_id", NScheme::NTypeIds::Text),
101+
Col("run_script_actor_id", NScheme::NTypeIds::Text),
102+
Col("operation_status", NScheme::NTypeIds::Int32),
103+
Col("execution_status", NScheme::NTypeIds::Int32),
104+
Col("finalization_status", NScheme::NTypeIds::Int32),
105+
Col("execution_mode", NScheme::NTypeIds::Int32),
106+
Col("start_ts", NScheme::NTypeIds::Timestamp),
107+
Col("end_ts", NScheme::NTypeIds::Timestamp),
108+
Col("query_text", NScheme::NTypeIds::Text),
109+
Col("syntax", NScheme::NTypeIds::Int32),
110+
Col("ast", NScheme::NTypeIds::Text),
111+
Col("ast_compressed", NScheme::NTypeIds::String),
112+
Col("ast_compression_method", NScheme::NTypeIds::Text),
113+
Col("issues", NScheme::NTypeIds::JsonDocument),
114+
Col("plan", NScheme::NTypeIds::JsonDocument),
115+
Col("meta", NScheme::NTypeIds::JsonDocument),
116+
Col("parameters", NScheme::NTypeIds::String), // TODO: store aparameters separately to support bigger storage.
117+
Col("result_set_metas", NScheme::NTypeIds::JsonDocument),
118+
Col("stats", NScheme::NTypeIds::JsonDocument),
119+
Col("expire_at", NScheme::NTypeIds::Timestamp), // Will be deleted from database after this deadline.
120+
Col("customer_supplied_id", NScheme::NTypeIds::Text),
121+
Col("user_token", NScheme::NTypeIds::Text),
122+
Col("script_sinks", NScheme::NTypeIds::JsonDocument),
123+
Col("script_secret_names", NScheme::NTypeIds::JsonDocument),
124+
},
125+
{ "database", "execution_id" },
126+
NKikimrServices::KQP_PROXY,
127+
TtlCol("expire_at", DEADLINE_OFFSET, BRO_RUN_INTERVAL)
157128
);
158129
}
159130

160-
void RunCreateScriptExecutionLeases() {
161-
TablesCreating++;
162-
Register(
163-
CreateTableCreator(
164-
{ ".metadata", "script_execution_leases" },
165-
{
166-
Col("database", NScheme::NTypeIds::Text),
167-
Col("execution_id", NScheme::NTypeIds::Text),
168-
Col("lease_deadline", NScheme::NTypeIds::Timestamp),
169-
Col("lease_generation", NScheme::NTypeIds::Int64),
170-
Col("expire_at", NScheme::NTypeIds::Timestamp), // Will be deleted from database after this deadline.
171-
},
172-
{ "database", "execution_id" },
173-
NKikimrServices::KQP_PROXY,
174-
TtlCol("expire_at")
175-
)
131+
static IActor* GetScriptExecutionLeasesCreator() {
132+
return CreateTableCreator(
133+
{ ".metadata", "script_execution_leases" },
134+
{
135+
Col("database", NScheme::NTypeIds::Text),
136+
Col("execution_id", NScheme::NTypeIds::Text),
137+
Col("lease_deadline", NScheme::NTypeIds::Timestamp),
138+
Col("lease_generation", NScheme::NTypeIds::Int64),
139+
Col("expire_at", NScheme::NTypeIds::Timestamp), // Will be deleted from database after this deadline.
140+
},
141+
{ "database", "execution_id" },
142+
NKikimrServices::KQP_PROXY,
143+
TtlCol("expire_at", DEADLINE_OFFSET, BRO_RUN_INTERVAL)
176144
);
177145
}
178146

179-
void RunCreateScriptResultSets() {
180-
TablesCreating++;
181-
Register(
182-
CreateTableCreator(
183-
{ ".metadata", "result_sets" },
184-
{
185-
Col("database", NScheme::NTypeIds::Text),
186-
Col("execution_id", NScheme::NTypeIds::Text),
187-
Col("result_set_id", NScheme::NTypeIds::Int32),
188-
Col("row_id", NScheme::NTypeIds::Int64),
189-
Col("expire_at", NScheme::NTypeIds::Timestamp),
190-
Col("result_set", NScheme::NTypeIds::String),
191-
Col("accumulated_size", NScheme::NTypeIds::Int64),
192-
},
193-
{ "database", "execution_id", "result_set_id", "row_id" },
194-
NKikimrServices::KQP_PROXY,
195-
TtlCol("expire_at")
196-
)
147+
static IActor* GetScriptResultSetsCreator() {
148+
return CreateTableCreator(
149+
{ ".metadata", "result_sets" },
150+
{
151+
Col("database", NScheme::NTypeIds::Text),
152+
Col("execution_id", NScheme::NTypeIds::Text),
153+
Col("result_set_id", NScheme::NTypeIds::Int32),
154+
Col("row_id", NScheme::NTypeIds::Int64),
155+
Col("expire_at", NScheme::NTypeIds::Timestamp),
156+
Col("result_set", NScheme::NTypeIds::String),
157+
Col("accumulated_size", NScheme::NTypeIds::Int64),
158+
},
159+
{ "database", "execution_id", "result_set_id", "row_id" },
160+
NKikimrServices::KQP_PROXY,
161+
TtlCol("expire_at", DEADLINE_OFFSET, BRO_RUN_INTERVAL)
197162
);
198163
}
199164

200-
void Handle(TEvTableCreator::TEvCreateTableResponse::TPtr&) {
201-
Y_ABORT_UNLESS(TablesCreating > 0);
202-
if (--TablesCreating == 0) {
203-
Send(Owner, std::move(ResultEvent));
204-
PassAway();
205-
}
165+
void OnTablesCreated(bool success, NYql::TIssues issues) override {
166+
Send(Owner, new TEvScriptExecutionsTablesCreationFinished(success, std::move(issues)));
206167
}
207-
208-
STRICT_STFUNC(StateFunc,
209-
hFunc(TEvTableCreator::TEvCreateTableResponse, Handle);
210-
)
211-
212-
private:
213-
THolder<NActors::IEventBase> ResultEvent;
214-
NActors::TActorId Owner;
215-
size_t TablesCreating = 0;
216168
};
217169

218170
Ydb::Query::ExecMode GetExecModeFromAction(NKikimrKqp::EQueryAction action) {
@@ -2867,8 +2819,8 @@ NActors::IActor* CreateScriptExecutionCreatorActor(TEvKqp::TEvScriptRequest::TPt
28672819
return new TCreateScriptExecutionActor(std::move(ev), queryServiceConfig, counters, maxRunTime);
28682820
}
28692821

2870-
NActors::IActor* CreateScriptExecutionsTablesCreator(THolder<NActors::IEventBase> resultEvent) {
2871-
return new TScriptExecutionsTablesCreator(std::move(resultEvent));
2822+
NActors::IActor* CreateScriptExecutionsTablesCreator() {
2823+
return new TScriptExecutionsTablesCreator();
28722824
}
28732825

28742826
NActors::IActor* CreateForgetScriptExecutionOperationActor(TEvForgetScriptExecutionOperation::TPtr ev) {

‎ydb/core/kqp/proxy_service/kqp_script_executions.h‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@ namespace NKikimr::NKqp {
1212

1313
// Creates all needed tables.
1414
// Sends result event back when the work is done.
15-
NActors::IActor* CreateScriptExecutionsTablesCreator(THolder<NActors::IEventBase> resultEvent);
15+
NActors::IActor* CreateScriptExecutionsTablesCreator();
1616

1717
// Create script execution and run it.
1818
NActors::IActor* CreateScriptExecutionCreatorActor(TEvKqp::TEvScriptRequest::TPtr&& ev, const NKikimrConfig::TQueryServiceConfig& queryServiceConfig, TIntrusivePtr<TKqpCounters> counters, TDuration maxRunTime = SCRIPT_TIMEOUT_LIMIT);

0 commit comments

Comments
 (0)