Skip to content

Commit 4800934

Browse files
VPolkaVPolka
authored
Merge 3055f36 into bd5ea97 
2 parents bd5ea97 + 3055f36 commit 4800934

28 files changed

+762
-84
lines changed

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

Lines changed: 13 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -14,21 +14,23 @@ namespace NKikimr::NKqp::NPrivateEvents {
1414

1515
struct TEvCompileRequest: public TEventLocal<TEvCompileRequest, TKqpEvents::EvCompileRequest> {
1616
TEvCompileRequest(const TIntrusiveConstPtr<NACLib::TUserToken>& userToken, const TMaybe<TString>& uid,
17-
TMaybe<TKqpQueryId>&& query, bool keepInCache, TInstant deadline,
17+
TMaybe<TKqpQueryId>&& query, bool keepInCache, bool canDevideIntoStatements, TInstant deadline,
1818
TKqpDbCountersPtr dbCounters, std::shared_ptr<std::atomic<bool>> intrestedInResult,
1919
const TIntrusivePtr<TUserRequestContext>& userRequestContext, NLWTrace::TOrbit orbit = {},
20-
TKqpTempTablesState::TConstPtr tempTablesState = nullptr, bool collectDiagnostics = false)
20+
TKqpTempTablesState::TConstPtr tempTablesState = nullptr, bool collectDiagnostics = false, TMaybe<TQueryAst> queryAst = Nothing())
2121
: UserToken(userToken)
2222
, Uid(uid)
2323
, Query(std::move(query))
2424
, KeepInCache(keepInCache)
25+
, CanDevideIntoStatements(canDevideIntoStatements)
2526
, Deadline(deadline)
2627
, DbCounters(dbCounters)
2728
, UserRequestContext(userRequestContext)
2829
, Orbit(std::move(orbit))
2930
, TempTablesState(std::move(tempTablesState))
3031
, IntrestedInResult(std::move(intrestedInResult))
3132
, CollectDiagnostics(collectDiagnostics)
33+
, QueryAst(queryAst)
3234
{
3335
Y_ENSURE(Uid.Defined() != Query.Defined());
3436
}
@@ -37,6 +39,7 @@ struct TEvCompileRequest: public TEventLocal<TEvCompileRequest, TKqpEvents::EvCo
3739
TMaybe<TString> Uid;
3840
TMaybe<TKqpQueryId> Query;
3941
bool KeepInCache = false;
42+
bool CanDevideIntoStatements = false;
4043
// it is allowed for local event to use absolute time (TInstant) instead of time interval (TDuration)
4144
TInstant Deadline;
4245
TKqpDbCountersPtr DbCounters;
@@ -49,6 +52,8 @@ struct TEvCompileRequest: public TEventLocal<TEvCompileRequest, TKqpEvents::EvCo
4952
std::shared_ptr<std::atomic<bool>> IntrestedInResult;
5053

5154
bool CollectDiagnostics = false;
55+
56+
TMaybe<TQueryAst> QueryAst;
5257
};
5358

5459
struct TEvRecompileRequest: public TEventLocal<TEvRecompileRequest, TKqpEvents::EvRecompileRequest> {
@@ -99,12 +104,14 @@ struct TEvCompileResponse: public TEventLocal<TEvCompileResponse, TKqpEvents::Ev
99104
};
100105

101106
struct TEvParseResponse: public TEventLocal<TEvParseResponse, TKqpEvents::EvParseResponse> {
102-
TEvParseResponse(const TKqpQueryId& query, TMaybe<TQueryAst> astResult)
103-
: AstResult(std::move(astResult))
104-
, Query(query) {}
107+
TEvParseResponse(const TKqpQueryId& query, TVector<TQueryAst> astStatements, NLWTrace::TOrbit orbit = {})
108+
: AstStatements(std::move(astStatements))
109+
, Query(query)
110+
, Orbit(std::move(orbit)) {}
105111

106-
TMaybe<TQueryAst> AstResult;
112+
TVector<TQueryAst> AstStatements;
107113
TKqpQueryId Query;
114+
NLWTrace::TOrbit Orbit;
108115
};
109116

110117
struct TEvCompileInvalidateRequest: public TEventLocal<TEvCompileInvalidateRequest,

‎ydb/core/kqp/common/kqp_lwtrace_probes.h

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -68,6 +68,9 @@ struct TQueryAction {
6868
PROBE(KqpCompileServiceReplyFromCache, GROUPS("KQP"), \
6969
TYPES(), \
7070
NAMES()) \
71+
PROBE(KqpCompileServiceReplyStatements, GROUPS("KQP"), \
72+
TYPES(), \
73+
NAMES()) \
7174
PROBE(KqpCompileServiceReplyError, GROUPS("KQP"), \
7275
TYPES(), \
7376
NAMES()) \

‎ydb/core/kqp/compile_service/kqp_compile_actor.cpp

Lines changed: 43 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -52,14 +52,14 @@ class TKqpCompileActor : public TActorBootstrapped<TKqpCompileActor> {
5252
TKqpDbCountersPtr dbCounters, std::optional<TKqpFederatedQuerySetup> federatedQuerySetup,
5353
const TIntrusivePtr<TUserRequestContext>& userRequestContext,
5454
NWilson::TTraceId traceId, TKqpTempTablesState::TConstPtr tempTablesState, bool collectFullDiagnostics,
55-
ECompileActorAction compileAction, TMaybe<TQueryAst> astResult)
55+
bool canDevideIntoStatements, ECompileActorAction compileAction, TMaybe<TQueryAst> queryAst)
5656
: Owner(owner)
5757
, ModuleResolverState(moduleResolverState)
5858
, Counters(counters)
5959
, FederatedQuerySetup(federatedQuerySetup)
6060
, Uid(uid)
6161
, QueryId(queryId)
62-
, QueryRef(QueryId.Text, QueryId.QueryParameterTypes, astResult)
62+
, QueryRef(QueryId.Text, QueryId.QueryParameterTypes, queryAst)
6363
, UserToken(userToken)
6464
, DbCounters(dbCounters)
6565
, Config(MakeIntrusive<TKikimrConfiguration>())
@@ -70,8 +70,9 @@ class TKqpCompileActor : public TActorBootstrapped<TKqpCompileActor> {
7070
, CompileActorSpan(TWilsonKqp::CompileActor, std::move(traceId), "CompileActor")
7171
, TempTablesState(std::move(tempTablesState))
7272
, CollectFullDiagnostics(collectFullDiagnostics)
73+
, CanDevideIntoStatements(canDevideIntoStatements)
7374
, CompileAction(compileAction)
74-
, AstResult(std::move(astResult))
75+
, QueryAst(std::move(queryAst))
7576
{
7677
Config->Init(kqpSettings->DefaultSettings.GetDefaultSettings(), QueryId.Cluster, kqpSettings->Settings, false);
7778

@@ -127,26 +128,22 @@ class TKqpCompileActor : public TActorBootstrapped<TKqpCompileActor> {
127128
}
128129

129130
private:
130-
void SetQueryAst(const TActorContext &ctx) {
131+
132+
TVector<TQueryAst> GetAstStatements(const TActorContext &ctx) {
131133
TString cluster = QueryId.Cluster;
132134
TString kqpTablePathPrefix = Config->_KqpTablePathPrefix.Get().GetRef();
133135
ui16 kqpYqlSyntaxVersion = Config->_KqpYqlSyntaxVersion.Get().GetRef();
134136
NSQLTranslation::EBindingsMode bindingsMode = Config->BindingsMode;
135137
bool isEnableExternalDataSources = AppData(ctx)->FeatureFlags.GetEnableExternalDataSources();
136138
bool isEnablePgConstsToParams = Config->EnablePgConstsToParams;
139+
bool perStatement = Config->EnableQueriesPerStatement && CanDevideIntoStatements;
137140

138-
auto astResult = ParseQuery(ConvertType(QueryId.Settings.QueryType), QueryId.Settings.Syntax, QueryId.Text, QueryId.QueryParameterTypes, QueryId.IsSql(), cluster, kqpTablePathPrefix, kqpYqlSyntaxVersion, bindingsMode, isEnableExternalDataSources, isEnablePgConstsToParams);
139-
YQL_ENSURE(astResult.Ast);
140-
if (astResult.Ast->IsOk()) {
141-
AstResult = std::move(astResult);
142-
}
141+
return SqlToAstStatements(ConvertType(QueryId.Settings.QueryType), QueryId.Settings.Syntax, QueryId.Text, QueryId.QueryParameterTypes, cluster, kqpTablePathPrefix, kqpYqlSyntaxVersion, bindingsMode, isEnableExternalDataSources, isEnablePgConstsToParams, QueryId.IsSql(), perStatement);
143142
}
144143

145144
void StartParsing(const TActorContext &ctx) {
146-
SetQueryAst(ctx);
147-
148145
Become(&TKqpCompileActor::CompileState);
149-
ReplyParseResult(ctx);
146+
ReplyParseResult(ctx, GetAstStatements(ctx));
150147
}
151148

152149
void StartCompilation(const TActorContext &ctx) {
@@ -352,15 +349,37 @@ class TKqpCompileActor : public TActorBootstrapped<TKqpCompileActor> {
352349
<< ", at state:" << state);
353350
}
354351

355-
void ReplyParseResult(const TActorContext &ctx) {
352+
void ReplyParseResult(const TActorContext &ctx, TVector<TQueryAst> astStatements) {
356353
Y_UNUSED(ctx);
354+
355+
if (astStatements.empty()) {
356+
NYql::TIssue issue(NYql::TPosition(), "Parsing result of query is empty");
357+
ReplyError(Ydb::StatusIds::GENERIC_ERROR, {issue});
358+
}
359+
360+
for (size_t statementId = 0; statementId < astStatements.size(); ++statementId) {
361+
if (!astStatements[statementId].Ast || !astStatements[statementId].Ast->IsOk() || !astStatements[statementId].Ast->Root) {
362+
ALOG_ERROR(NKikimrServices::KQP_COMPILE_ACTOR, "Get parsing result with error"
363+
<< ", self: " << SelfId()
364+
<< ", owner: " << Owner
365+
<< ", statement id: " << statementId);
366+
367+
NYql::TIssue issue(NYql::TPosition(), "Internal error while parsing query.");
368+
for (const auto& i : astStatements[statementId].Ast->Issues) {
369+
issue.AddSubIssue(MakeIntrusive<NYql::TIssue>(i));
370+
}
371+
372+
ReplyError(Ydb::StatusIds::GENERIC_ERROR, {issue});
373+
return;
374+
}
375+
}
376+
357377
ALOG_DEBUG(NKikimrServices::KQP_COMPILE_ACTOR, "Send parsing result"
358378
<< ", self: " << SelfId()
359379
<< ", owner: " << Owner
360-
<< (AstResult && AstResult->Ast->IsOk() ? ", parsing is successful" : ", parsing is not successful"));
380+
<< ", statements size: " << astStatements.size());
361381

362-
auto responseEv = MakeHolder<TEvKqp::TEvParseResponse>(QueryId, std::move(AstResult));
363-
AstResult = Nothing();
382+
auto responseEv = MakeHolder<TEvKqp::TEvParseResponse>(QueryId, astStatements);
364383
Send(Owner, responseEv.Release());
365384

366385
Counters->ReportCompileFinish(DbCounters);
@@ -410,8 +429,8 @@ class TKqpCompileActor : public TActorBootstrapped<TKqpCompileActor> {
410429
KqpCompileResult->PreparedQuery = preparedQueryHolder;
411430
KqpCompileResult->AllowCache = CanCacheQuery(KqpCompileResult->PreparedQuery->GetPhysicalQuery());
412431

413-
if (AstResult) {
414-
KqpCompileResult->Ast = AstResult->Ast;
432+
if (QueryAst) {
433+
KqpCompileResult->Ast = QueryAst->Ast;
415434
}
416435
}
417436

@@ -476,8 +495,9 @@ class TKqpCompileActor : public TActorBootstrapped<TKqpCompileActor> {
476495
TKqpTempTablesState::TConstPtr TempTablesState;
477496
bool CollectFullDiagnostics;
478497

498+
bool CanDevideIntoStatements;
479499
ECompileActorAction CompileAction;
480-
TMaybe<TQueryAst> AstResult;
500+
TMaybe<TQueryAst> QueryAst;
481501
};
482502

483503
void ApplyServiceConfig(TKikimrConfiguration& kqpConfig, const TTableServiceConfig& serviceConfig) {
@@ -506,6 +526,7 @@ void ApplyServiceConfig(TKikimrConfiguration& kqpConfig, const TTableServiceConf
506526
kqpConfig.IndexAutoChooserMode = serviceConfig.GetIndexAutoChooseMode();
507527
kqpConfig.EnablePgConstsToParams = serviceConfig.GetEnablePgConstsToParams();
508528
kqpConfig.ExtractPredicateRangesLimit = serviceConfig.GetExtractPredicateRangesLimit();
529+
kqpConfig.EnableQueriesPerStatement = serviceConfig.GetEnableQueriesPerStatement();
509530

510531
if (const auto limit = serviceConfig.GetResourceManager().GetMkqlHeavyProgramMemoryLimit()) {
511532
kqpConfig._KqpYqlCombinerMemoryLimit = std::max(1_GB, limit - (limit >> 2U));
@@ -521,14 +542,15 @@ IActor* CreateKqpCompileActor(const TActorId& owner, const TKqpSettings::TConstP
521542
std::optional<TKqpFederatedQuerySetup> federatedQuerySetup,
522543
TKqpDbCountersPtr dbCounters, const TIntrusivePtr<TUserRequestContext>& userRequestContext,
523544
NWilson::TTraceId traceId, TKqpTempTablesState::TConstPtr tempTablesState,
524-
ECompileActorAction compileAction, TMaybe<TQueryAst> astResult, bool collectFullDiagnostics)
545+
ECompileActorAction compileAction, TMaybe<TQueryAst> queryAst, bool collectFullDiagnostics,
546+
bool canDevideIntoStatements)
525547
{
526548
return new TKqpCompileActor(owner, kqpSettings, tableServiceConfig, queryServiceConfig, metadataProviderConfig,
527549
moduleResolverState, counters,
528550
uid, query, userToken, dbCounters,
529551
federatedQuerySetup, userRequestContext,
530552
std::move(traceId), std::move(tempTablesState), collectFullDiagnostics,
531-
compileAction, std::move(astResult));
553+
canDevideIntoStatements, compileAction, std::move(queryAst));
532554
}
533555

534556
} // namespace NKqp

0 commit comments

Comments
 (0)