Skip to content

Commit 841c78c

Browse files
ulya-sidorinaulya-sidorina
committed
feature(kqp): enable stream lookup for data query
1 parent 06f9376 commit 841c78c

16 files changed

+67
-19
lines changed

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

+10-3
Original file line numberDiff line numberDiff line change
@@ -1650,11 +1650,18 @@ class TKqpDataExecuter : public TKqpExecuterBase<TKqpDataExecuter, EExecType::Da
16501650
}
16511651
}
16521652

1653-
bool HassDmlOperationOnOlap(NKqpProto::TKqpPhyTx_EType queryType, const NKqpProto::TKqpPhyStage& stage) {
1653+
bool HasDmlOperationOnOlap(NKqpProto::TKqpPhyTx_EType queryType, const NKqpProto::TKqpPhyStage& stage) {
16541654
if (queryType == NKqpProto::TKqpPhyTx::TYPE_DATA) {
16551655
return true;
16561656
}
1657-
for (const auto &tableOp : stage.GetTableOps()) {
1657+
1658+
for (const auto& input : stage.GetInputs()) {
1659+
if (input.GetTypeCase() == NKqpProto::TKqpPhyConnection::kStreamLookup) {
1660+
return true;
1661+
}
1662+
}
1663+
1664+
for (const auto& tableOp : stage.GetTableOps()) {
16581665
if (tableOp.GetTypeCase() != NKqpProto::TKqpPhyTableOperation::kReadOlapRange) {
16591666
return true;
16601667
}
@@ -1691,7 +1698,7 @@ class TKqpDataExecuter : public TKqpExecuterBase<TKqpDataExecuter, EExecType::Da
16911698
}
16921699
}
16931700

1694-
if (stageInfo.Meta.IsOlap() && HassDmlOperationOnOlap(tx.Body->GetType(), stage)) {
1701+
if (stageInfo.Meta.IsOlap() && HasDmlOperationOnOlap(tx.Body->GetType(), stage)) {
16951702
auto error = TStringBuilder() << "Data manipulation queries do not support column shard tables.";
16961703
LOG_E(error);
16971704
ReplyErrorAndDie(Ydb::StatusIds::PRECONDITION_FAILED,

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

+1
Original file line numberDiff line numberDiff line change
@@ -112,6 +112,7 @@ void FillKqpTasksGraphStages(TKqpTasksGraph& tasksGraph, const TVector<IKqpGatew
112112
meta.TableId = MakeTableId(input.GetStreamLookup().GetTable());
113113
meta.TablePath = input.GetStreamLookup().GetTable().GetPath();
114114
meta.TableConstInfo = tx.Body->GetTableConstInfoById()->Map.at(meta.TableId);
115+
meta.ShardOperations.insert(TKeyDesc::ERowOperation::Read);
115116
}
116117

117118
if (input.GetTypeCase() == NKqpProto::TKqpPhyConnection::kSequencer) {

‎ydb/core/kqp/opt/logical/kqp_opt_log_ranges_predext.cpp

-1
Original file line numberDiff line numberDiff line change
@@ -387,7 +387,6 @@ TExprBase KqpPushExtractedPredicateToReadTable(TExprBase node, TExprContext& ctx
387387
.Columns(read.Columns())
388388
.LookupKeys(keys)
389389
.Index(indexName.Cast())
390-
.LookupKeys(keys)
391390
.Done();
392391
}
393392
} else if (kqpCtx.IsDataQuery()) {

‎ydb/core/kqp/ut/cost/kqp_cost_ut.cpp

+3-2
Original file line numberDiff line numberDiff line change
@@ -11,10 +11,11 @@ namespace NKqp {
1111
using namespace NYdb;
1212
using namespace NYdb::NTable;
1313

14-
static NKikimrConfig::TAppConfig GetAppConfig(bool sourceRead) {
14+
static NKikimrConfig::TAppConfig GetAppConfig(bool sourceRead, bool streamLookup = true) {
1515
auto app = NKikimrConfig::TAppConfig();
1616
app.MutableTableServiceConfig()->SetEnableKqpDataQuerySourceRead(sourceRead);
1717
app.MutableTableServiceConfig()->SetEnableKqpScanQuerySourceRead(sourceRead);
18+
app.MutableTableServiceConfig()->SetEnableKqpDataQueryStreamLookup(streamLookup);
1819
return app;
1920
}
2021

@@ -43,7 +44,7 @@ Y_UNIT_TEST_SUITE(KqpCost) {
4344
//runtime->SetLogPriority(NKikimrServices::GRPC_SERVER, NActors::NLog::PRI_DEBUG);
4445
}
4546
Y_UNIT_TEST_TWIN(PointLookup, SourceRead) {
46-
TKikimrRunner kikimr(GetAppConfig(SourceRead));
47+
TKikimrRunner kikimr(GetAppConfig(SourceRead, false));
4748
auto db = kikimr.GetTableClient();
4849
auto session = db.CreateSession().GetValueSync().GetSession();
4950

‎ydb/core/kqp/ut/effects/kqp_inplace_update_ut.cpp

+1
Original file line numberDiff line numberDiff line change
@@ -401,6 +401,7 @@ Y_UNIT_TEST_TWIN(BigRow, EnableInplaceUpdate) {
401401
// source read use iterator interface, that doesn't use datashard transactions
402402
NKikimrConfig::TAppConfig appConfig;
403403
appConfig.MutableTableServiceConfig()->SetEnableKqpDataQuerySourceRead(false);
404+
appConfig.MutableTableServiceConfig()->SetEnableKqpDataQueryStreamLookup(false);
404405

405406
auto settings = TKikimrSettings()
406407
.SetAppConfig(appConfig)

‎ydb/core/kqp/ut/indexes/kqp_indexes_ut.cpp

+4-2
Original file line numberDiff line numberDiff line change
@@ -3962,9 +3962,9 @@ R"([[#;#;["Primary1"];[41u]];[["Secondary2"];[2u];["Primary2"];[42u]];[["Seconda
39623962

39633963
auto& stats = NYdb::TProtoAccessor::GetProto(*result2.GetStats());
39643964

3965-
int readPhase = 1;
3965+
int readPhase = 0;
39663966
if (serverSettings.AppConfig.GetTableServiceConfig().GetEnableKqpDataQueryStreamLookup()) {
3967-
UNIT_ASSERT_VALUES_EQUAL(stats.query_phases().size(), 2);
3967+
UNIT_ASSERT_VALUES_EQUAL(stats.query_phases().size(), 1);
39683968

39693969
UNIT_ASSERT_VALUES_EQUAL(stats.query_phases(readPhase).table_access().size(), 2);
39703970
UNIT_ASSERT_VALUES_EQUAL(stats.query_phases(readPhase).table_access(0).name(), "/Root/SecondaryComplexKeys");
@@ -3974,6 +3974,8 @@ R"([[#;#;["Primary1"];[41u]];[["Secondary2"];[2u];["Primary2"];[42u]];[["Seconda
39743974
} else {
39753975
UNIT_ASSERT_VALUES_EQUAL(stats.query_phases().size(), 3);
39763976

3977+
readPhase++;
3978+
39773979
UNIT_ASSERT_VALUES_EQUAL(stats.query_phases(readPhase).table_access().size(), 1);
39783980
UNIT_ASSERT_VALUES_EQUAL(stats.query_phases(readPhase).table_access(0).name(), "/Root/SecondaryComplexKeys/Index/indexImplTable");
39793981
UNIT_ASSERT_VALUES_EQUAL(stats.query_phases(readPhase).table_access(0).reads().rows(), 1);

‎ydb/core/kqp/ut/join/kqp_index_lookup_join_ut.cpp

+1
Original file line numberDiff line numberDiff line change
@@ -90,6 +90,7 @@ Y_UNIT_TEST_SUITE(KqpIndexLookupJoin) {
9090
void Test(const TString& query, const TString& answer, size_t rightTableReads, bool useStreamLookup = false) {
9191
NKikimrConfig::TAppConfig appConfig;
9292
appConfig.MutableTableServiceConfig()->SetEnableKqpDataQueryStreamIdxLookupJoin(useStreamLookup);
93+
appConfig.MutableTableServiceConfig()->SetEnableKqpDataQueryStreamLookup(false);
9394

9495
auto settings = TKikimrSettings().SetAppConfig(appConfig);
9596
TKikimrRunner kikimr(settings);

‎ydb/core/kqp/ut/opt/kqp_extract_predicate_unpack_ut.cpp

+10-3
Original file line numberDiff line numberDiff line change
@@ -143,8 +143,14 @@ void Test(const TString& query, const TString& answer, THashSet<TString> allowSc
143143
}
144144
}
145145

146-
void TestRange(const TString& query, const TString& answer, ui64 rowsRead, int stagesCount = 1) {
147-
TKikimrRunner kikimr;
146+
void TestRange(const TString& query, const TString& answer, ui64 rowsRead, int stagesCount = 1, bool streamLookup = true) {
147+
NKikimrConfig::TAppConfig appConfig;
148+
appConfig.MutableTableServiceConfig()->SetEnableKqpDataQueryStreamLookup(streamLookup);
149+
150+
auto settings = TKikimrSettings()
151+
.SetAppConfig(appConfig);
152+
153+
TKikimrRunner kikimr(settings);
148154
auto db = kikimr.GetTableClient();
149155
auto session = db.CreateSession().GetValueSync().GetSession();
150156

@@ -189,7 +195,8 @@ Y_UNIT_TEST(OverflowLookup) {
189195
)",
190196
R"([])",
191197
0,
192-
2);
198+
2,
199+
false);
193200

194201
TestRange(
195202
R"(

‎ydb/core/kqp/ut/opt/kqp_ne_ut.cpp

+7-2
Original file line numberDiff line numberDiff line change
@@ -205,7 +205,12 @@ Y_UNIT_TEST_SUITE(KqpNewEngine) {
205205

206206
auto explainResult = session.ExplainDataQuery(query).GetValueSync();
207207
UNIT_ASSERT_VALUES_EQUAL_C(explainResult.GetStatus(), EStatus::SUCCESS, explainResult.GetIssues().ToString());
208-
UNIT_ASSERT_C(explainResult.GetAst().Contains("KqpLookupTable"), explainResult.GetAst());
208+
209+
if (settings.AppConfig.GetTableServiceConfig().GetEnableKqpDataQueryStreamLookup()) {
210+
UNIT_ASSERT_C(explainResult.GetAst().Contains("KqpCnStreamLookup"), explainResult.GetAst());
211+
} else {
212+
UNIT_ASSERT_C(explainResult.GetAst().Contains("KqpLookupTable"), explainResult.GetAst());
213+
}
209214

210215
auto params = kikimr.GetTableClient().GetParamsBuilder()
211216
.AddParam("$group").OptionalUint32(1).Build()
@@ -1224,7 +1229,7 @@ Y_UNIT_TEST_SUITE(KqpNewEngine) {
12241229
size_t phase = 0;
12251230
if (stats.query_phases().size() == 2) {
12261231
phase = 1;
1227-
} else if (stats.query_phases().size() == 0) {
1232+
} else if (stats.query_phases().size() == 1) {
12281233
phase = 0;
12291234
} else {
12301235
UNIT_ASSERT(false);

‎ydb/core/kqp/ut/perf/kqp_query_perf_ut.cpp

+1
Original file line numberDiff line numberDiff line change
@@ -280,6 +280,7 @@ Y_UNIT_TEST_SUITE(KqpQueryPerf) {
280280

281281
Y_UNIT_TEST_QUAD(IndexLookupJoin, EnableStreamLookup, QueryService) {
282282
NKikimrConfig::TAppConfig appConfig;
283+
appConfig.MutableTableServiceConfig()->SetEnableKqpDataQueryStreamLookup(false);
283284
appConfig.MutableTableServiceConfig()->SetEnableKqpDataQueryStreamIdxLookupJoin(EnableStreamLookup);
284285
auto settings = TKikimrSettings()
285286
.SetAppConfig(appConfig);

‎ydb/core/kqp/ut/query/kqp_explain_ut.cpp

+19-4
Original file line numberDiff line numberDiff line change
@@ -497,7 +497,13 @@ Y_UNIT_TEST_SUITE(KqpExplain) {
497497
UNIT_ASSERT_EQUAL(node.GetMapSafe().at("Table").GetStringSafe(), "KeyValue");
498498
node = FindPlanNodeByKv(plan, "Name", "TableFullScan");
499499
UNIT_ASSERT_EQUAL(node.GetMapSafe().at("Table").GetStringSafe(), "KeyValue");
500-
node = FindPlanNodeByKv(plan, "Name", "TablePointLookup");
500+
501+
if (settings.AppConfig.GetTableServiceConfig().GetEnableKqpDataQueryStreamLookup()) {
502+
node = FindPlanNodeByKv(plan, "Node Type", "TableLookup");
503+
} else {
504+
node = FindPlanNodeByKv(plan, "Name", "TablePointLookup");
505+
}
506+
501507
UNIT_ASSERT_EQUAL(node.GetMapSafe().at("Table").GetStringSafe(), "KeyValue");
502508
}
503509

@@ -535,8 +541,13 @@ Y_UNIT_TEST_SUITE(KqpExplain) {
535541
UNIT_ASSERT_VALUES_EQUAL(rangeScansCount, 1);
536542

537543
ui32 lookupsCount = 0;
538-
lookupsCount = CountPlanNodesByKv(plan, "Node Type", "Stage-TablePointLookup");
539-
lookupsCount += CountPlanNodesByKv(plan, "Node Type", "TablePointLookup-ConstantExpr");
544+
if (settings.AppConfig.GetTableServiceConfig().GetEnableKqpDataQueryStreamLookup()) {
545+
lookupsCount = CountPlanNodesByKv(plan, "Node Type", "TableLookup");
546+
} else {
547+
lookupsCount = CountPlanNodesByKv(plan, "Node Type", "Stage-TablePointLookup");
548+
lookupsCount += CountPlanNodesByKv(plan, "Node Type", "TablePointLookup-ConstantExpr");
549+
}
550+
540551
UNIT_ASSERT_VALUES_EQUAL(lookupsCount, 1);
541552

542553
/* check tables section */
@@ -902,7 +913,11 @@ Y_UNIT_TEST_SUITE(KqpExplain) {
902913
}
903914

904915
Y_UNIT_TEST(MultiJoinCteLinks) {
905-
TKikimrRunner kikimr;
916+
NKikimrConfig::TAppConfig appConfig;
917+
appConfig.MutableTableServiceConfig()->SetEnableKqpDataQueryStreamLookup(false);
918+
auto settings = TKikimrSettings()
919+
.SetAppConfig(appConfig);
920+
TKikimrRunner kikimr{settings};
906921
auto db = kikimr.GetTableClient();
907922
auto session = db.CreateSession().GetValueSync().GetSession();
908923

‎ydb/core/kqp/ut/query/kqp_query_ut.cpp

+2
Original file line numberDiff line numberDiff line change
@@ -351,6 +351,7 @@ Y_UNIT_TEST_SUITE(KqpQuery) {
351351
Y_UNIT_TEST(QueryTimeoutImmediate) {
352352
NKikimrConfig::TAppConfig appConfig;
353353
appConfig.MutableTableServiceConfig()->SetEnableKqpDataQuerySourceRead(false);
354+
appConfig.MutableTableServiceConfig()->SetEnableKqpDataQueryStreamLookup(false);
354355
auto settings = TKikimrSettings()
355356
.SetAppConfig(appConfig);
356357
TKikimrRunner kikimr{settings};
@@ -490,6 +491,7 @@ Y_UNIT_TEST_SUITE(KqpQuery) {
490491

491492
Y_UNIT_TEST(QueryCancelImmediate) {
492493
NKikimrConfig::TAppConfig appConfig;
494+
appConfig.MutableTableServiceConfig()->SetEnableKqpDataQueryStreamLookup(false);
493495
appConfig.MutableTableServiceConfig()->SetEnableKqpDataQuerySourceRead(false);
494496
auto settings = TKikimrSettings()
495497
.SetAppConfig(appConfig);

‎ydb/core/kqp/ut/query/kqp_stats_ut.cpp

+1
Original file line numberDiff line numberDiff line change
@@ -102,6 +102,7 @@ TCollectedStreamResult JoinStatsBasic(
102102
std::function<Iterator(TKikimrRunner&, ECollectQueryStatsMode, const TString&)> getIter) {
103103
NKikimrConfig::TAppConfig appConfig;
104104
appConfig.MutableTableServiceConfig()->SetEnableKqpScanQueryStreamLookup(false);
105+
appConfig.MutableTableServiceConfig()->SetEnableKqpDataQueryStreamLookup(false);
105106
appConfig.MutableTableServiceConfig()->SetEnableKqpScanQuerySourceRead(true);
106107
auto settings = TKikimrSettings()
107108
.SetAppConfig(appConfig);

‎ydb/core/kqp/ut/tx/kqp_mvcc_ut.cpp

+1-1
Original file line numberDiff line numberDiff line change
@@ -50,7 +50,7 @@ Y_UNIT_TEST_SUITE(KqpSnapshotRead) {
5050
if (result.GetStatus() == EStatus::SUCCESS)
5151
continue;
5252

53-
if (settings.AppConfig.GetTableServiceConfig().GetEnableKqpDataQuerySourceRead() && false) {
53+
if (settings.AppConfig.GetTableServiceConfig().GetEnableKqpDataQueryStreamLookup()) {
5454
UNIT_ASSERT_C(HasIssue(result.GetIssues(), NYql::TIssuesIds::DEFAULT_ERROR,
5555
[](const NYql::TIssue& issue){
5656
return issue.GetMessage().Contains("has no snapshot at");

‎ydb/core/protos/table_service_config.proto

+1-1
Original file line numberDiff line numberDiff line change
@@ -224,7 +224,7 @@ message TTableServiceConfig {
224224
optional uint64 SessionIdleDurationSeconds = 28 [default = 600];
225225
optional TAggregationConfig AggregationConfig = 29;
226226
optional bool EnableKqpScanQueryStreamLookup = 30 [default = true];
227-
optional bool EnableKqpDataQueryStreamLookup = 31 [default = false];
227+
optional bool EnableKqpDataQueryStreamLookup = 31 [default = true];
228228
optional TExecuterRetriesConfig ExecuterRetriesConfig = 32;
229229
reserved 33; // optional bool EnableKqpDataQueryStreamPointLookup = 33 [default = false];
230230
optional bool EnablePublishKqpProxyByRM = 34 [default = true];

‎ydb/core/tx/datashard/datashard_ut_order.cpp

+5
Original file line numberDiff line numberDiff line change
@@ -1526,6 +1526,7 @@ Y_UNIT_TEST(TestMvccReadDoesntBlockWrites) {
15261526
TPortManager pm;
15271527
NKikimrConfig::TAppConfig app;
15281528
app.MutableTableServiceConfig()->SetEnableKqpDataQuerySourceRead(false);
1529+
app.MutableTableServiceConfig()->SetEnableKqpDataQueryStreamLookup(false);
15291530
TServerSettings serverSettings(pm.GetPort(2134));
15301531
serverSettings.SetEnableMvccSnapshotReads(false);
15311532
serverSettings.SetDomainName("Root")
@@ -1863,9 +1864,12 @@ Y_UNIT_TEST_TWIN(TestOutOfOrderNonConflictingWrites, StreamLookup) {
18631864

18641865
Y_UNIT_TEST(MvccTestOutOfOrderRestartLocksSingleWithoutBarrier) {
18651866
TPortManager pm;
1867+
NKikimrConfig::TAppConfig app;
1868+
app.MutableTableServiceConfig()->SetEnableKqpDataQueryStreamLookup(false);
18661869
TServerSettings serverSettings(pm.GetPort(2134));
18671870
serverSettings.SetEnableMvccSnapshotReads(false);
18681871
serverSettings.SetDomainName("Root")
1872+
.SetAppConfig(app)
18691873
.SetUseRealThreads(false);
18701874

18711875
Tests::TServer::TPtr server = new TServer(serverSettings);
@@ -3507,6 +3511,7 @@ Y_UNIT_TEST(MvccTestSnapshotRead) {
35073511
TPortManager pm;
35083512
NKikimrConfig::TAppConfig app;
35093513
app.MutableTableServiceConfig()->SetEnableKqpDataQuerySourceRead(false);
3514+
app.MutableTableServiceConfig()->SetEnableKqpDataQueryStreamLookup(false);
35103515
TServerSettings serverSettings(pm.GetPort(2134));
35113516
serverSettings.SetDomainName("Root")
35123517
.SetAppConfig(app)

0 commit comments

Comments
 (0)