Skip to content

Commit 59654e9

Browse files
pashandor789pashandor789
authored
SQL command ANALYZE (#6996)
1 parent b2e3ab3 commit 59654e9

28 files changed

+710
-8
lines changed

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

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22
#include "kqp_executer_impl.h"
33

44
#include <ydb/core/kqp/gateway/actors/scheme.h>
5+
#include <ydb/core/kqp/gateway/actors/analyze_actor.h>
56
#include <ydb/core/kqp/gateway/local_rpc/helper.h>
67
#include <ydb/core/tx/tx_proxy/proxy.h>
78
#include <ydb/core/kqp/session_actor/kqp_worker_common.h>
@@ -307,6 +308,29 @@ class TKqpSchemeExecuter : public TActorBootstrapped<TKqpSchemeExecuter> {
307308
break;
308309
}
309310

311+
case NKqpProto::TKqpSchemeOperation::kAnalyzeTable: {
312+
const auto& analyzeOperation = schemeOp.GetAnalyzeTable();
313+
314+
auto analyzePromise = NewPromise<IKqpGateway::TGenericResult>();
315+
316+
TVector<TString> columns{analyzeOperation.columns().begin(), analyzeOperation.columns().end()};
317+
IActor* analyzeActor = new TAnalyzeActor(analyzeOperation.GetTablePath(), columns, analyzePromise);
318+
319+
auto actorSystem = TlsActivationContext->AsActorContext().ExecutorThread.ActorSystem;
320+
RegisterWithSameMailbox(analyzeActor);
321+
322+
auto selfId = SelfId();
323+
analyzePromise.GetFuture().Subscribe([actorSystem, selfId](const TFuture<IKqpGateway::TGenericResult>& future) {
324+
auto ev = MakeHolder<TEvPrivate::TEvResult>();
325+
ev->Result = future.GetValue();
326+
327+
actorSystem->Send(selfId, ev.Release());
328+
});
329+
330+
Become(&TKqpSchemeExecuter::ExecuteState);
331+
return;
332+
}
333+
310334
default:
311335
InternalError(TStringBuilder() << "Unexpected scheme operation: "
312336
<< (ui32) schemeOp.GetOperationCase());

‎ydb/core/kqp/gateway/actors/analyze_actor.cpp

Lines changed: 227 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,227 @@
1+
#include "analyze_actor.h"
2+
3+
#include <ydb/core/base/path.h>
4+
#include <ydb/core/base/tablet_pipecache.h>
5+
#include <ydb/library/actors/core/log.h>
6+
#include <ydb/library/services/services.pb.h>
7+
8+
9+
namespace NKikimr::NKqp {
10+
11+
enum {
12+
FirstRoundCookie = 0,
13+
SecondRoundCookie = 1,
14+
};
15+
16+
using TNavigate = NSchemeCache::TSchemeCacheNavigate;
17+
18+
void TAnalyzeActor::Bootstrap() {
19+
using TNavigate = NSchemeCache::TSchemeCacheNavigate;
20+
auto navigate = std::make_unique<TNavigate>();
21+
auto& entry = navigate->ResultSet.emplace_back();
22+
entry.Path = SplitPath(TablePath);
23+
entry.Operation = TNavigate::EOp::OpTable;
24+
entry.RequestType = TNavigate::TEntry::ERequestType::ByPath;
25+
navigate->Cookie = FirstRoundCookie;
26+
27+
Send(NKikimr::MakeSchemeCacheID(), new TEvTxProxySchemeCache::TEvNavigateKeySet(navigate.release()));
28+
29+
Become(&TAnalyzeActor::StateWork);
30+
}
31+
32+
void TAnalyzeActor::SendAnalyzeStatus() {
33+
Y_ABORT_UNLESS(StatisticsAggregatorId.has_value());
34+
35+
auto getStatus = std::make_unique<NStat::TEvStatistics::TEvAnalyzeStatus>();
36+
auto& record = getStatus->Record;
37+
PathIdFromPathId(PathId, record.MutablePathId());
38+
39+
Send(
40+
MakePipePerNodeCacheID(false),
41+
new TEvPipeCache::TEvForward(getStatus.release(), StatisticsAggregatorId.value(), true)
42+
);
43+
}
44+
45+
void TAnalyzeActor::Handle(NStat::TEvStatistics::TEvAnalyzeResponse::TPtr& ev, const TActorContext& ctx) {
46+
Y_UNUSED(ev);
47+
Y_UNUSED(ctx);
48+
49+
SendAnalyzeStatus();
50+
}
51+
52+
void TAnalyzeActor::Handle(TEvAnalyzePrivate::TEvAnalyzeStatusCheck::TPtr& ev, const TActorContext& ctx) {
53+
Y_UNUSED(ev);
54+
Y_UNUSED(ctx);
55+
56+
SendAnalyzeStatus();
57+
}
58+
59+
void TAnalyzeActor::Handle(NStat::TEvStatistics::TEvAnalyzeStatusResponse::TPtr& ev, const TActorContext& ctx) {
60+
auto& record = ev->Get()->Record;
61+
switch (record.GetStatus()) {
62+
case NKikimrStat::TEvAnalyzeStatusResponse::STATUS_UNSPECIFIED: {
63+
Promise.SetValue(
64+
NYql::NCommon::ResultFromError<NYql::IKikimrGateway::TGenericResult>(
65+
YqlIssue(
66+
{}, NYql::TIssuesIds::UNEXPECTED,
67+
TStringBuilder() << "Statistics Aggregator unspecified error"
68+
)
69+
)
70+
);
71+
this->Die(ctx);
72+
return;
73+
}
74+
case NKikimrStat::TEvAnalyzeStatusResponse::STATUS_NO_OPERATION: {
75+
NYql::IKikimrGateway::TGenericResult result;
76+
result.SetSuccess();
77+
Promise.SetValue(std::move(result));
78+
79+
this->Die(ctx);
80+
return;
81+
}
82+
case NKikimrStat::TEvAnalyzeStatusResponse::STATUS_ENQUEUED: {
83+
Schedule(TDuration::Seconds(10), new TEvAnalyzePrivate::TEvAnalyzeStatusCheck());
84+
return;
85+
}
86+
case NKikimrStat::TEvAnalyzeStatusResponse::STATUS_IN_PROGRESS: {
87+
Schedule(TDuration::Seconds(5), new TEvAnalyzePrivate::TEvAnalyzeStatusCheck());
88+
return;
89+
}
90+
}
91+
}
92+
93+
void TAnalyzeActor::Handle(TEvTxProxySchemeCache::TEvNavigateKeySetResult::TPtr& ev, const TActorContext& ctx) {
94+
std::unique_ptr<TNavigate> navigate(ev->Get()->Request.Release());
95+
Y_ABORT_UNLESS(navigate->ResultSet.size() == 1);
96+
auto& entry = navigate->ResultSet.front();
97+
98+
if (entry.Status != TNavigate::EStatus::Ok) {
99+
NYql::EYqlIssueCode error;
100+
switch (entry.Status) {
101+
case TNavigate::EStatus::PathErrorUnknown:
102+
case TNavigate::EStatus::RootUnknown:
103+
case TNavigate::EStatus::PathNotTable:
104+
case TNavigate::EStatus::TableCreationNotComplete:
105+
error = NYql::TIssuesIds::KIKIMR_SCHEME_ERROR;
106+
case TNavigate::EStatus::LookupError:
107+
case TNavigate::EStatus::RedirectLookupError:
108+
error = NYql::TIssuesIds::KIKIMR_TEMPORARILY_UNAVAILABLE;
109+
default:
110+
error = NYql::TIssuesIds::DEFAULT_ERROR;
111+
}
112+
Promise.SetValue(
113+
NYql::NCommon::ResultFromIssues<NYql::IKikimrGateway::TGenericResult>(
114+
error,
115+
TStringBuilder() << "Can't get statistics aggregator ID. " << entry.Status,
116+
{}
117+
)
118+
);
119+
this->Die(ctx);
120+
return;
121+
}
122+
123+
if (navigate->Cookie == SecondRoundCookie) {
124+
if (entry.DomainInfo->Params.HasStatisticsAggregator()) {
125+
SendStatisticsAggregatorAnalyze(entry, ctx);
126+
} else {
127+
Promise.SetValue(
128+
NYql::NCommon::ResultFromIssues<NYql::IKikimrGateway::TGenericResult>(
129+
NYql::TIssuesIds::DEFAULT_ERROR,
130+
TStringBuilder() << "Can't get statistics aggregator ID.", {}
131+
)
132+
);
133+
}
134+
135+
this->Die(ctx);
136+
return;
137+
}
138+
139+
PathId = entry.TableId.PathId;
140+
141+
auto& domainInfo = entry.DomainInfo;
142+
143+
auto navigateDomainKey = [this] (TPathId domainKey) {
144+
using TNavigate = NSchemeCache::TSchemeCacheNavigate;
145+
auto navigate = std::make_unique<TNavigate>();
146+
auto& entry = navigate->ResultSet.emplace_back();
147+
entry.TableId = TTableId(domainKey.OwnerId, domainKey.LocalPathId);
148+
entry.Operation = TNavigate::EOp::OpPath;
149+
entry.RequestType = TNavigate::TEntry::ERequestType::ByTableId;
150+
entry.RedirectRequired = false;
151+
navigate->Cookie = SecondRoundCookie;
152+
153+
Send(MakeSchemeCacheID(), new TEvTxProxySchemeCache::TEvNavigateKeySet(navigate.release()));
154+
};
155+
156+
if (!domainInfo->IsServerless()) {
157+
if (domainInfo->Params.HasStatisticsAggregator()) {
158+
SendStatisticsAggregatorAnalyze(entry, ctx);
159+
return;
160+
}
161+
162+
navigateDomainKey(domainInfo->DomainKey);
163+
} else {
164+
navigateDomainKey(domainInfo->ResourcesDomainKey);
165+
}
166+
}
167+
168+
void TAnalyzeActor::SendStatisticsAggregatorAnalyze(const NSchemeCache::TSchemeCacheNavigate::TEntry& entry, const TActorContext& ctx) {
169+
Y_ABORT_UNLESS(entry.DomainInfo->Params.HasStatisticsAggregator());
170+
171+
StatisticsAggregatorId = entry.DomainInfo->Params.GetStatisticsAggregator();
172+
173+
auto analyzeRequest = std::make_unique<NStat::TEvStatistics::TEvAnalyze>();
174+
auto& record = analyzeRequest->Record;
175+
auto table = record.AddTables();
176+
177+
PathIdFromPathId(PathId, table->MutablePathId());
178+
179+
180+
THashMap<TString, ui32> tagByColumnName;
181+
for (const auto& [_, tableInfo]: entry.Columns) {
182+
tagByColumnName[TString(tableInfo.Name)] = tableInfo.Id;
183+
}
184+
185+
for (const auto& columnName: Columns) {
186+
if (!tagByColumnName.contains(columnName)){
187+
Promise.SetValue(
188+
NYql::NCommon::ResultFromError<NYql::IKikimrGateway::TGenericResult>(
189+
YqlIssue(
190+
{}, NYql::TIssuesIds::UNEXPECTED,
191+
TStringBuilder() << "No such column: " << columnName << " in the " << TablePath
192+
)
193+
)
194+
);
195+
this->Die(ctx);
196+
return;
197+
}
198+
199+
*table->MutableColumnTags()->Add() = tagByColumnName[columnName];
200+
}
201+
202+
Send(
203+
MakePipePerNodeCacheID(false),
204+
new TEvPipeCache::TEvForward(analyzeRequest.release(), entry.DomainInfo->Params.GetStatisticsAggregator(), true),
205+
IEventHandle::FlagTrackDelivery
206+
);
207+
}
208+
209+
void TAnalyzeActor::HandleUnexpectedEvent(ui32 typeRewrite) {
210+
ALOG_CRIT(
211+
NKikimrServices::KQP_GATEWAY,
212+
"TAnalyzeActor, unexpected event, request type: " << typeRewrite;
213+
);
214+
215+
Promise.SetValue(
216+
NYql::NCommon::ResultFromError<NYql::IKikimrGateway::TGenericResult>(
217+
YqlIssue(
218+
{}, NYql::TIssuesIds::UNEXPECTED,
219+
TStringBuilder() << "Unexpected event: " << typeRewrite
220+
)
221+
)
222+
);
223+
224+
this->PassAway();
225+
}
226+
227+
}// end of NKikimr::NKqp

‎ydb/core/kqp/gateway/actors/analyze_actor.h

Lines changed: 66 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,66 @@
1+
#include <ydb/core/tx/scheme_cache/scheme_cache.h>
2+
#include <ydb/core/statistics/events.h>
3+
#include <ydb/library/actors/core/actor_bootstrapped.h>
4+
#include <ydb/library/actors/core/hfunc.h>
5+
6+
#include <ydb/core/kqp/provider/yql_kikimr_gateway.h>
7+
8+
9+
namespace NKikimr::NKqp {
10+
11+
12+
struct TEvAnalyzePrivate {
13+
enum EEv {
14+
EvAnalyzeStatusCheck = EventSpaceBegin(TEvents::ES_PRIVATE),
15+
EvEnd
16+
};
17+
18+
struct TEvAnalyzeStatusCheck : public TEventLocal<TEvAnalyzeStatusCheck, EvAnalyzeStatusCheck> {};
19+
};
20+
21+
class TAnalyzeActor : public NActors::TActorBootstrapped<TAnalyzeActor> {
22+
public:
23+
TAnalyzeActor(TString tablePath, TVector<TString> columns, NThreading::TPromise<NYql::IKikimrGateway::TGenericResult> promise)
24+
: TablePath(tablePath)
25+
, Columns(columns)
26+
, Promise(promise)
27+
{}
28+
29+
void Bootstrap();
30+
31+
STFUNC(StateWork) {
32+
switch(ev->GetTypeRewrite()) {
33+
HFunc(TEvTxProxySchemeCache::TEvNavigateKeySetResult, Handle);
34+
HFunc(NStat::TEvStatistics::TEvAnalyzeResponse, Handle);
35+
HFunc(NStat::TEvStatistics::TEvAnalyzeStatusResponse, Handle);
36+
HFunc(TEvAnalyzePrivate::TEvAnalyzeStatusCheck, Handle);
37+
default:
38+
HandleUnexpectedEvent(ev->GetTypeRewrite());
39+
}
40+
}
41+
42+
private:
43+
void Handle(NStat::TEvStatistics::TEvAnalyzeResponse::TPtr& ev, const TActorContext& ctx);
44+
45+
void Handle(NStat::TEvStatistics::TEvAnalyzeStatusResponse::TPtr& ev, const TActorContext& ctx);
46+
47+
void Handle(TEvTxProxySchemeCache::TEvNavigateKeySetResult::TPtr& ev, const TActorContext& ctx);
48+
49+
void Handle(TEvAnalyzePrivate::TEvAnalyzeStatusCheck::TPtr& ev, const TActorContext& ctx);
50+
51+
void HandleUnexpectedEvent(ui32 typeRewrite);
52+
53+
void SendStatisticsAggregatorAnalyze(const NSchemeCache::TSchemeCacheNavigate::TEntry&, const TActorContext&);
54+
55+
void SendAnalyzeStatus();
56+
57+
private:
58+
TString TablePath;
59+
TVector<TString> Columns;
60+
NThreading::TPromise<NYql::IKikimrGateway::TGenericResult> Promise;
61+
// For Statistics Aggregator
62+
std::optional<ui64> StatisticsAggregatorId;
63+
TPathId PathId;
64+
};
65+
66+
} // end of NKikimr::NKqp

‎ydb/core/kqp/gateway/actors/ya.make

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@ LIBRARY()
22

33
SRCS(
44
scheme.cpp
5+
analyze_actor.cpp
56
)
67

78
PEERDIR(
@@ -11,6 +12,7 @@ PEERDIR(
1112
ydb/library/yql/providers/common/gateway
1213
ydb/core/tx/schemeshard
1314
ydb/library/actors/core
15+
ydb/library/services
1416
)
1517

1618
YQL_LAST_ABI_VERSION()

‎ydb/core/kqp/gateway/kqp_ic_gateway.cpp

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
#include "kqp_gateway.h"
22
#include "actors/kqp_ic_gateway_actors.h"
3+
#include "actors/analyze_actor.h"
34
#include "actors/scheme.h"
45
#include "kqp_metadata_loader.h"
56
#include "local_rpc/helper.h"
@@ -1386,6 +1387,22 @@ class TKikimrIcGateway : public IKqpGateway {
13861387
}
13871388
}
13881389

1390+
TFuture<TGenericResult> Analyze(const TString& cluster, const NYql::TAnalyzeSettings& settings) override {
1391+
try {
1392+
if (!CheckCluster(cluster)) {
1393+
return InvalidCluster<TGenericResult>(cluster);
1394+
}
1395+
1396+
auto analyzePromise = NewPromise<TGenericResult>();
1397+
IActor* analyzeActor = new TAnalyzeActor(settings.TablePath, settings.Columns, analyzePromise);
1398+
RegisterActor(analyzeActor);
1399+
1400+
return analyzePromise.GetFuture();
1401+
} catch (yexception& e) {
1402+
return MakeFuture(ResultFromException<TGenericResult>(e));
1403+
}
1404+
}
1405+
13891406
template <class TSettings>
13901407
class IObjectModifier {
13911408
public:

0 commit comments

Comments
 (0)