Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
70 changes: 33 additions & 37 deletions ydb/core/kqp/ut/join/kqp_join_order_ut.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -75,7 +75,11 @@ static TKikimrRunner GetKikimrWithJoinSettings(bool useStreamLookupJoin = false,
return TKikimrRunner(serverSettings);
}

class TChainConstructor {
void PrintPlan(const TString& plan) {
Cout << plan << Endl;
}

class TChainTester {
public:
TChainTester(size_t chainSize)
: Kikimr(GetKikimrWithJoinSettings(false, GetStats(chainSize)))
Expand Down Expand Up @@ -107,22 +111,17 @@ class TChainConstructor {

private:
void CreateTables() {
for (size_t i = 0; i < ChainSize_; ++i) {
TString tableName;

tableName
.append("/Root/table_").append(ToString(i));;

TString createTable;
createTable
+= "CREATE TABLE `" + tableName + "` (id"
+ ToString(i) + " Int32, "
+ "PRIMARY KEY (id" + ToString(i) + "));";

std::cout << createTable << std::endl;
auto res = Session_.ExecuteSchemeQuery(createTable).GetValueSync();
std::cout << res.GetIssues().ToString() << std::endl;
UNIT_ASSERT(res.IsSuccess());
for (size_t i = 0; i < ChainSize; ++i) {
TString tableName = Sprintf("/Root/table_%ld", i);

TString createTable = Sprintf(
"CREATE TABLE `%s` (id%ld Int32, PRIMARY KEY (id%ld));",
tableName.c_str(), i, i
);

auto result = Session.ExecuteSchemeQuery(createTable).GetValueSync();
result.GetIssues().PrintTo(Cerr);
UNIT_ASSERT_VALUES_EQUAL(result.GetStatus(), EStatus::SUCCESS);
}
}

Expand All @@ -131,30 +130,29 @@ class TChainConstructor {

joinRequest.append("SELECT * FROM `/Root/table_0` as t0 ");

for (size_t i = 1; i < ChainSize_; ++i) {
TString table = "/Root/table_" + ToString(i);
for (size_t i = 1; i < ChainSize; ++i) {
TString table = Sprintf("/Root/table_%ld", i);

TString prevAliasTable = "t" + ToString(i - 1);
TString aliasTable = "t" + ToString(i);
TString prevAliasTable = Sprintf("t%ld", i - 1);
TString aliasTable = Sprintf("t%ld", i);

joinRequest
+= "INNER JOIN `" + table + "`" + " AS " + aliasTable + " ON "
+ aliasTable + ".id" + ToString(i) + "=" + prevAliasTable + ".id"
+ ToString(i-1) + " ";
joinRequest +=
Sprintf(
"INNER JOIN `%s` AS %s ON %s.id%ld = %s.id%ld ",
table.c_str(), aliasTable.c_str(), aliasTable.c_str(), i, prevAliasTable.c_str(), i - 1
);
}

auto result = Session_.ExecuteDataQuery(joinRequest, TTxControl::BeginTx().CommitTx()).ExtractValueSync();

std::cout << result.GetIssues().ToString() << std::endl;
std::cout << joinRequest << std::endl;
auto result = Session.ExplainDataQuery(joinRequest).ExtractValueSync();
result.GetIssues().PrintTo(Cerr);
UNIT_ASSERT_VALUES_EQUAL(result.GetStatus(), EStatus::SUCCESS);
PrintPlan(result.GetPlan());
}

private:
TKikimrRunner Kikimr_;
NYdb::NTable::TTableClient TableClient_;
TSession Session_;
size_t ChainSize_;
TKikimrRunner Kikimr;
NYdb::NTable::TTableClient TableClient;
TSession Session;
size_t ChainSize;
};

void ExplainJoinOrderTestDataQuery(const TString& queryPath, bool useStreamLookupJoin) {
Expand Down Expand Up @@ -197,9 +195,7 @@ void ExecuteJoinOrderTestDataQuery(const TString& queryPath, bool useStreamLooku

Y_UNIT_TEST_SUITE(KqpJoinOrder) {
Y_UNIT_TEST(Chain65Nodes) {
TChainConstructor chain(65);
chain.CreateTables();
chain.JoinTables();
TChainTester(65).Test();
}

Y_UNIT_TEST_TWIN(FiveWayJoin, StreamLookupJoin) {
Expand Down