Skip to content

Commit ec12505

Browse files
gridnevvvitgridnevvvit
authored
implement syntax v0 to v1 fallback (#28899)
1 parent c2dac27 commit ec12505

File tree

8 files changed

+293
-15
lines changed

8 files changed

+293
-15
lines changed

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

Lines changed: 62 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -24,8 +24,7 @@
2424

2525
#include <ydb/core/base/cputime.h>
2626

27-
namespace NKikimr {
28-
namespace NKqp {
27+
namespace NKikimr::NKqp {
2928

3029
static const TString YqlName = "CompileActor";
3130

@@ -69,7 +68,8 @@ class TKqpCompileActor : public TActorBootstrapped<TKqpCompileActor> {
6968
, UserToken(userToken)
7069
, ClientAddress(clientAddress)
7170
, DbCounters(dbCounters)
72-
, Config(MakeIntrusive<TKikimrConfiguration>())
71+
, KqpSettings(kqpSettings)
72+
, TableServiceConfig(tableServiceConfig)
7373
, QueryServiceConfig(queryServiceConfig)
7474
, CompilationTimeout(TDuration::MilliSeconds(tableServiceConfig.GetCompileTimeoutMs()))
7575
, SplitCtx(std::move(splitCtx))
@@ -80,26 +80,49 @@ class TKqpCompileActor : public TActorBootstrapped<TKqpCompileActor> {
8080
, CollectFullDiagnostics(collectFullDiagnostics)
8181
, CompileAction(compileAction)
8282
, QueryAst(std::move(queryAst))
83+
, EnforcedSqlVersion(tableServiceConfig.GetEnforceSqlVersionV1())
8384
{
84-
Config->Init(kqpSettings->DefaultSettings.GetDefaultSettings(), QueryId.Cluster, kqpSettings->Settings, false);
85+
Config = BuildConfiguration(tableServiceConfig);
86+
PerStatementResult = perStatementResult && Config->EnablePerStatementQueryExecution;
87+
}
88+
89+
TKikimrConfiguration::TPtr BuildConfiguration(const TTableServiceConfig& tableServiceConfig) {
90+
NYql::TKikimrConfiguration::TPtr config = MakeIntrusive<TKikimrConfiguration>();
91+
92+
config->Init(KqpSettings->DefaultSettings.GetDefaultSettings(), QueryId.Cluster, KqpSettings->Settings, false);
8593

8694
if (!QueryId.Database.empty()) {
87-
Config->_KqpTablePathPrefix = QueryId.Database;
95+
config->_KqpTablePathPrefix = QueryId.Database;
8896
}
8997

90-
ApplyServiceConfig(*Config, tableServiceConfig);
98+
ApplyServiceConfig(*config, tableServiceConfig);
99+
100+
if (!tableServiceConfig.HasSqlVersion() || tableServiceConfig.GetSqlVersion() != 0) {
101+
EnforcedSqlVersion = false;
102+
} else if (EnforcedSqlVersion) {
103+
LOG_DEBUG_S(*TlsActivationContext, NKikimrServices::KQP_COMPILE_ACTOR,
104+
"Enforced SQL version 1, "
105+
<< "current sql version: " << tableServiceConfig.GetSqlVersion()
106+
<< " queryText: " << EscapeC(QueryId.Text)
107+
);
108+
109+
config->_KqpYqlSyntaxVersion = 1;
110+
} else {
111+
EnforcedSqlVersion = false;
112+
}
91113

92114
if (QueryId.Settings.QueryType == NKikimrKqp::QUERY_TYPE_SQL_GENERIC_SCRIPT || QueryId.Settings.QueryType == NKikimrKqp::QUERY_TYPE_SQL_GENERIC_QUERY) {
93115
ui32 scriptResultRowsLimit = QueryServiceConfig.GetScriptResultRowsLimit();
94116
if (scriptResultRowsLimit > 0) {
95-
Config->_ResultRowsLimit = scriptResultRowsLimit;
117+
config->_ResultRowsLimit = scriptResultRowsLimit;
96118
} else {
97-
Config->_ResultRowsLimit.Clear();
119+
config->_ResultRowsLimit.Clear();
98120
}
99121
}
100-
PerStatementResult = perStatementResult && Config->EnablePerStatementQueryExecution;
101122

102-
Config->FreezeDefaults();
123+
config->FreezeDefaults();
124+
125+
return config;
103126
}
104127

105128
void Bootstrap(const TActorContext& ctx) {
@@ -253,7 +276,12 @@ class TKqpCompileActor : public TActorBootstrapped<TKqpCompileActor> {
253276
TYqlLogScope logScope(ctx, NKikimrServices::KQP_YQL, YqlName, UserRequestContext->TraceId);
254277

255278
auto prepareSettings = PrepareCompilationSettings(ctx);
279+
StartCompilationWithSettings(prepareSettings);
280+
Continue(ctx);
281+
Become(&TKqpCompileActor::CompileState);
282+
}
256283

284+
void StartCompilationWithSettings(IKqpHost::TPrepareSettings& prepareSettings) {
257285
NCpuTime::TCpuTimer timer(CompileCpuTime);
258286

259287
switch (QueryId.Settings.QueryType) {
@@ -286,9 +314,6 @@ class TKqpCompileActor : public TActorBootstrapped<TKqpCompileActor> {
286314
default:
287315
YQL_ENSURE(false, "Unexpected query type: " << QueryId.Settings.QueryType);
288316
}
289-
290-
Continue(ctx);
291-
Become(&TKqpCompileActor::CompileState);
292317
}
293318

294319
void Continue(const TActorContext &ctx) {
@@ -305,6 +330,7 @@ class TKqpCompileActor : public TActorBootstrapped<TKqpCompileActor> {
305330
}
306331

307332
IKqpHost::TPrepareSettings PrepareCompilationSettings(const TActorContext &ctx) {
333+
// If CurrentSqlVersion differs from the frozen Config, create a new Config with updated SqlVersion
308334
TKqpRequestCounters::TPtr counters = new TKqpRequestCounters;
309335
counters->Counters = Counters;
310336
counters->DbCounters = DbCounters;
@@ -532,6 +558,25 @@ class TKqpCompileActor : public TActorBootstrapped<TKqpCompileActor> {
532558
return;
533559
}
534560

561+
// If compilation failed and we tried SqlVersion = 1, retry with SqlVersion = 0
562+
if (EnforcedSqlVersion && status != Ydb::StatusIds::SUCCESS) {
563+
Counters->ReportCompileEnforceConfigFailed(DbCounters);
564+
LOG_ERROR_S(ctx, NKikimrServices::KQP_COMPILE_ACTOR, "Compilation with SqlVersion = 1 failed, retrying with SqlVersion = 0"
565+
<< ", self: " << ctx.SelfID
566+
<< ", database: " << QueryId.Database
567+
<< ", text: \"" << EscapeC(QueryId.Text) << "\"");
568+
569+
EnforcedSqlVersion = false;
570+
Config = BuildConfiguration(TableServiceConfig);
571+
auto prepareSettings = PrepareCompilationSettings(ctx);
572+
573+
StartCompilationWithSettings(prepareSettings);
574+
Continue(ctx);
575+
return;
576+
} else if (EnforcedSqlVersion && status == Ydb::StatusIds::SUCCESS) {
577+
Counters->ReportCompileEnforceConfigSuccess(DbCounters);
578+
}
579+
535580
auto database = QueryId.Database;
536581
if (kqpResult.SqlVersion) {
537582
Counters->ReportSqlVersion(DbCounters, *kqpResult.SqlVersion);
@@ -613,6 +658,8 @@ class TKqpCompileActor : public TActorBootstrapped<TKqpCompileActor> {
613658
TString ClientAddress;
614659
TKqpDbCountersPtr DbCounters;
615660
TKikimrConfiguration::TPtr Config;
661+
TKqpSettings::TConstPtr KqpSettings;
662+
TTableServiceConfig TableServiceConfig;
616663
TQueryServiceConfig QueryServiceConfig;
617664
TDuration CompilationTimeout;
618665
TInstant StartTime;
@@ -638,6 +685,7 @@ class TKqpCompileActor : public TActorBootstrapped<TKqpCompileActor> {
638685
bool PerStatementResult;
639686
ECompileActorAction CompileAction;
640687
TMaybe<TQueryAst> QueryAst;
688+
bool EnforcedSqlVersion;
641689
};
642690

643691
void ApplyServiceConfig(TKikimrConfiguration& kqpConfig, const TTableServiceConfig& serviceConfig) {
@@ -731,5 +779,4 @@ IActor* CreateKqpCompileActor(const TActorId& owner, const TKqpSettings::TConstP
731779
std::move(splitCtx), std::move(splitExpr));
732780
}
733781

734-
} // namespace NKqp
735-
} // namespace NKikimr
782+
} // namespace NKikimr::NKqp

‎ydb/core/kqp/compile_service/ut/kqp_compile_fallback_ut.cpp‎

Lines changed: 164 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,164 @@
1+
#include <ydb/core/kqp/ut/common/kqp_ut_common.h>
2+
#include <ydb/core/kqp/counters/kqp_counters.h>
3+
4+
#include <library/cpp/testing/unittest/registar.h>
5+
6+
namespace NKikimr {
7+
namespace NKqp {
8+
9+
using namespace NYdb;
10+
using namespace NYdb::NTable;
11+
12+
namespace {
13+
14+
// Helper function to enable debug logging for compile-related services
15+
void EnableCompileDebugLogging(TKikimrRunner& kikimr) {
16+
auto runtime = kikimr.GetTestServer().GetRuntime();
17+
runtime->SetLogPriority(NKikimrServices::KQP_COMPILE_ACTOR, NLog::PRI_DEBUG);
18+
runtime->SetLogPriority(NKikimrServices::KQP_COMPILE_SERVICE, NLog::PRI_DEBUG);
19+
}
20+
21+
std::pair<ui32, ui32> GetEnforceConfigCounters(TKikimrRunner& kikimr) {
22+
auto counters = TKqpCounters(kikimr.GetTestServer().GetRuntime()->GetAppData().Counters);
23+
return {
24+
counters.GetKqpCounters()->GetCounter("Compilation/EnforceConfig/Success")->Val(),
25+
counters.GetKqpCounters()->GetCounter("Compilation/EnforceConfig/Failed")->Val()
26+
};
27+
}
28+
29+
// Helper function to test data query execution with different SqlVersion configurations
30+
// Returns counters for verification
31+
std::pair<ui32, ui32> TestDataQueryWithSqlVersion(TMaybe<ui32> sqlVersion, const TString& query, bool enforceSqlVersionV1 = true) {
32+
NKikimrConfig::TAppConfig appConfig;
33+
if (sqlVersion) {
34+
appConfig.MutableTableServiceConfig()->SetSqlVersion(*sqlVersion);
35+
}
36+
37+
appConfig.MutableTableServiceConfig()->SetEnforceSqlVersionV1(enforceSqlVersionV1);
38+
// If sqlVersion is Nothing(), SqlVersion is not set (defaults to 1)
39+
40+
TKikimrRunner kikimr{ TKikimrSettings(appConfig) };
41+
EnableCompileDebugLogging(kikimr);
42+
43+
auto db = kikimr.GetTableClient();
44+
auto session = db.CreateSession().GetValueSync().GetSession();
45+
46+
auto result = session.ExecuteDataQuery(query, TTxControl::BeginTx().CommitTx()).ExtractValueSync();
47+
UNIT_ASSERT_C(result.IsSuccess(), result.GetIssues().ToString());
48+
49+
return GetEnforceConfigCounters(kikimr);
50+
}
51+
52+
// Helper function to test prepared query with SqlVersion = 0 (triggers fallback)
53+
std::pair<ui32, ui32> TestPreparedQueryWithFallback(const TString& query) {
54+
NKikimrConfig::TAppConfig appConfig;
55+
appConfig.MutableTableServiceConfig()->SetSqlVersion(0);
56+
57+
TKikimrRunner kikimr{ TKikimrSettings(appConfig) };
58+
EnableCompileDebugLogging(kikimr);
59+
60+
auto db = kikimr.GetTableClient();
61+
auto session = db.CreateSession().GetValueSync().GetSession();
62+
63+
auto prepareResult = session.PrepareDataQuery(query).GetValueSync();
64+
UNIT_ASSERT_C(prepareResult.IsSuccess(), prepareResult.GetIssues().ToString());
65+
66+
auto queryObj = prepareResult.GetQuery();
67+
auto result = queryObj.Execute(TTxControl::BeginTx().CommitTx()).ExtractValueSync();
68+
UNIT_ASSERT_C(result.IsSuccess(), result.GetIssues().ToString());
69+
return GetEnforceConfigCounters(kikimr);
70+
}
71+
72+
// Helper function to test scan query with SqlVersion = 0 (triggers fallback)
73+
std::pair<ui32, ui32> TestScanQueryWithFallback(const TString& query) {
74+
NKikimrConfig::TAppConfig appConfig;
75+
appConfig.MutableTableServiceConfig()->SetSqlVersion(0);
76+
77+
TKikimrRunner kikimr{ TKikimrSettings(appConfig) };
78+
EnableCompileDebugLogging(kikimr);
79+
80+
auto queryClient = kikimr.GetQueryClient();
81+
auto session = queryClient.GetSession().ExtractValueSync().GetSession();
82+
83+
auto result = session.ExecuteQuery(query, NQuery::TTxControl::BeginTx().CommitTx()).ExtractValueSync();
84+
UNIT_ASSERT_C(result.IsSuccess(), result.GetIssues().ToString());
85+
86+
return GetEnforceConfigCounters(kikimr);
87+
}
88+
89+
} // anonymous namespace
90+
91+
Y_UNIT_TEST_SUITE(KqpCompileFallback) {
92+
93+
// Test that when SqlVersion = 0, it first tries SqlVersion = 1 and succeeds (no fallback)
94+
Y_UNIT_TEST(FallbackToVersion1Success) {
95+
auto [success, failed] = TestDataQueryWithSqlVersion(0, R"(
96+
SELECT * FROM `/Root/KeyValue` WHERE Key = 1;
97+
)");
98+
// Should succeed with SqlVersion = 1 on first try, so no fallback
99+
UNIT_ASSERT(success > 0);
100+
UNIT_ASSERT_VALUES_EQUAL(failed, 0);
101+
}
102+
103+
// Test that when SqlVersion = 0, compilation works (either succeeds with v1 or falls back to v0)
104+
// This test verifies the fallback mechanism doesn't break normal operation
105+
Y_UNIT_TEST(FallbackMechanismWorks) {
106+
auto [success, failed] = TestDataQueryWithSqlVersion(0, R"(
107+
SELECT * FROM [/Root/KeyValue] LIMIT 1;
108+
)");
109+
// Query should compile successfully (either with or without fallback)
110+
UNIT_ASSERT_VALUES_EQUAL(failed, 1);
111+
}
112+
113+
// Test that when SqlVersion = 0, compilation works (either succeeds with v1 or falls back to v0) when EnforceSqlVersionV1 is false
114+
// This test verifies the fallback mechanism doesn't work when EnforceSqlVersionV1 is false
115+
Y_UNIT_TEST(FallbackMechanismWorksEnforceSqlVersionV1False) {
116+
auto [success, failed] = TestDataQueryWithSqlVersion(0, R"(
117+
SELECT * FROM [/Root/KeyValue] LIMIT 1;
118+
)", false);
119+
// Query should compile successfully (either with or without fallback)
120+
UNIT_ASSERT_VALUES_EQUAL(failed, 0);
121+
UNIT_ASSERT_VALUES_EQUAL(success, 0);
122+
}
123+
124+
// Test that when SqlVersion = 1, no fallback is attempted
125+
Y_UNIT_TEST(NoFallbackWhenSqlVersion1) {
126+
auto [success, failed] = TestDataQueryWithSqlVersion(1, R"(
127+
SELECT * FROM `/Root/KeyValue` WHERE Key = 1;
128+
)");
129+
// Should not use fallback when SqlVersion = 1
130+
UNIT_ASSERT(success == 0);
131+
UNIT_ASSERT(failed == 0);
132+
}
133+
134+
// Test that when SqlVersion is not set (defaults to 1), no fallback is attempted
135+
Y_UNIT_TEST(NoFallbackWhenSqlVersionNotSet) {
136+
auto [success, failed] = TestDataQueryWithSqlVersion(Nothing(), R"(
137+
SELECT * FROM `/Root/KeyValue` WHERE Key = 1;
138+
)");
139+
// Should not use fallback when SqlVersion defaults to 1
140+
UNIT_ASSERT(success == 0);
141+
UNIT_ASSERT(failed == 0);
142+
}
143+
144+
// Test fallback with a prepared query
145+
Y_UNIT_TEST(FallbackWithPreparedQuery) {
146+
auto [success, failed] = TestPreparedQueryWithFallback(R"(
147+
SELECT * FROM [/Root/KeyValue] WHERE Key = 1;
148+
)");
149+
150+
UNIT_ASSERT(failed > 0);
151+
}
152+
153+
// Test that fallback works with scan queries
154+
Y_UNIT_TEST(FallbackWithScanQuery) {
155+
auto [success, failed] = TestScanQueryWithFallback(R"(
156+
SELECT * FROM `/Root/KeyValue` WHERE Key > 0;
157+
)");
158+
159+
UNIT_ASSERT(success > 0);
160+
}
161+
}
162+
163+
} // namespace NKqp
164+
} // namespace NKikimr

‎ydb/core/kqp/compile_service/ut/ya.make‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,29 @@
1+
UNITTEST_FOR(ydb/core/kqp/compile_service)
2+
3+
FORK_SUBTESTS()
4+
SPLIT_FACTOR(50)
5+
6+
IF (WITH_VALGRIND)
7+
SIZE(LARGE)
8+
TAG(ya:fat)
9+
ELSE()
10+
SIZE(MEDIUM)
11+
ENDIF()
12+
13+
SRCS(
14+
kqp_compile_fallback_ut.cpp
15+
)
16+
17+
PEERDIR(
18+
ydb/core/kqp
19+
ydb/core/kqp/ut/common
20+
ydb/core/kqp/compile_service
21+
ydb/public/sdk/cpp/src/client/proto
22+
library/cpp/testing/unittest
23+
yql/essentials/sql/pg_dummy
24+
)
25+
26+
YQL_LAST_ABI_VERSION()
27+
28+
END()
29+

‎ydb/core/kqp/compile_service/ya.make‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,3 +23,7 @@ END()
2323
RECURSE(
2424
helpers
2525
)
26+
27+
RECURSE_FOR_TESTS(
28+
ut
29+
)

0 commit comments

Comments
 (0)