Skip to content

Commit 603c442

Browse files
nikvas0nikvas0
authored
Merge 31e480a into 59fa9fb 
2 parents 59fa9fb + 31e480a commit 603c442

File tree

3 files changed

+74
-35
lines changed

3 files changed

+74
-35
lines changed

‎ydb/library/workload/stock/stock.cpp‎

Lines changed: 64 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -35,25 +35,45 @@ TStockWorkloadGenerator::TStockWorkloadGenerator(const TStockWorkloadParams* par
3535

3636
std::string TStockWorkloadGenerator::GetDDLQueries() const {
3737
std::string stockPartitionsDdl = "";
38-
std::string ordersPartitionsDdl = "WITH (READ_REPLICAS_SETTINGS = \"per_az:1\")";
38+
std::string ordersPartitionsDdl = "";
3939
std::string orderLinesPartitionsDdl = "";
40-
if (Params.PartitionsByLoad) {
41-
stockPartitionsDdl = std::format(R"(WITH (
42-
AUTO_PARTITIONING_BY_LOAD = ENABLED
40+
41+
if (Params.GetStoreType() == TStockWorkloadParams::EStoreType::Row) {
42+
ordersPartitionsDdl = "WITH (READ_REPLICAS_SETTINGS = \"per_az:1\")";
43+
if (Params.PartitionsByLoad) {
44+
stockPartitionsDdl = std::format(R"(WITH (
45+
STORE = ROW
46+
, AUTO_PARTITIONING_BY_LOAD = ENABLED
47+
, AUTO_PARTITIONING_MIN_PARTITIONS_COUNT = {0}
48+
))", Params.MinPartitions);
49+
ordersPartitionsDdl = std::format(R"(WITH (
50+
STORE = ROW
51+
, READ_REPLICAS_SETTINGS = "per_az:1"
52+
, AUTO_PARTITIONING_BY_LOAD = ENABLED
53+
, AUTO_PARTITIONING_MIN_PARTITIONS_COUNT = {0}
54+
, AUTO_PARTITIONING_MAX_PARTITIONS_COUNT = 1000
55+
, UNIFORM_PARTITIONS = {0}
56+
))", Params.MinPartitions);
57+
orderLinesPartitionsDdl = std::format(R"(WITH (
58+
STORE = ROW
59+
, AUTO_PARTITIONING_BY_LOAD = ENABLED
60+
, AUTO_PARTITIONING_MIN_PARTITIONS_COUNT = {0}
61+
, AUTO_PARTITIONING_MAX_PARTITIONS_COUNT = 1000
62+
, UNIFORM_PARTITIONS = {0}
63+
))", Params.MinPartitions);
64+
}
65+
} else {
66+
stockPartitionsDdl = std::format(R"(WITH (
67+
STORE = COLUMN
4368
, AUTO_PARTITIONING_MIN_PARTITIONS_COUNT = {0}
4469
))", Params.MinPartitions);
4570
ordersPartitionsDdl = std::format(R"(WITH (
46-
READ_REPLICAS_SETTINGS = "per_az:1"
47-
, AUTO_PARTITIONING_BY_LOAD = ENABLED
71+
STORE = COLUMN
4872
, AUTO_PARTITIONING_MIN_PARTITIONS_COUNT = {0}
49-
, AUTO_PARTITIONING_MAX_PARTITIONS_COUNT = 1000
50-
, UNIFORM_PARTITIONS = {0}
5173
))", Params.MinPartitions);
5274
orderLinesPartitionsDdl = std::format(R"(WITH (
53-
AUTO_PARTITIONING_BY_LOAD = ENABLED
75+
STORE = COLUMN
5476
, AUTO_PARTITIONING_MIN_PARTITIONS_COUNT = {0}
55-
, AUTO_PARTITIONING_MAX_PARTITIONS_COUNT = 1000
56-
, UNIFORM_PARTITIONS = {0}
5777
))", Params.MinPartitions);
5878
}
5979

@@ -66,11 +86,13 @@ std::string TStockWorkloadGenerator::GetDDLQueries() const {
6686
}
6787

6888
return std::format(R"(--!syntax_v1
69-
CREATE TABLE `{0}/stock`(product Utf8, quantity Int64, PRIMARY KEY(product)) {1};
70-
CREATE TABLE `{0}/orders`(id Uint64, customer Utf8, created Datetime, processed Datetime, PRIMARY KEY(id), INDEX ix_cust GLOBAL ON (customer, created) COVER (processed)) {2};
71-
CREATE TABLE `{0}/orderLines`(id_order Uint64, product Utf8, quantity Int64, PRIMARY KEY(id_order, product)) {3};
89+
CREATE TABLE `{0}/stock`(product Utf8 {5}, quantity Int64, PRIMARY KEY(product)) {1};
90+
CREATE TABLE `{0}/orders`(id Uint64 {5}, customer Utf8, created Datetime, processed Datetime, PRIMARY KEY(id) {6}) {2};
91+
CREATE TABLE `{0}/orderLines`(id_order Uint64 {5}, product Utf8 {5}, quantity Int64 {5}, PRIMARY KEY(id_order, product)) {3};
7292
{4}
73-
)", DbPath, stockPartitionsDdl, ordersPartitionsDdl, orderLinesPartitionsDdl, changefeeds);
93+
)", DbPath, stockPartitionsDdl, ordersPartitionsDdl, orderLinesPartitionsDdl, changefeeds,
94+
Params.GetStoreType() == TStockWorkloadParams::EStoreType::Row ? "" : "NOT NULL",
95+
Params.GetStoreType() == TStockWorkloadParams::EStoreType::Row ? ", INDEX ix_cust GLOBAL ON (customer, created) COVER (processed)" : "");
7496
}
7597

7698
TQueryInfoList TStockWorkloadGenerator::GetInitialData() {
@@ -205,15 +227,17 @@ TQueryInfo TStockWorkloadGenerator::ExecuteOrder(const uint64_t orderID) {
205227
}
206228

207229
TQueryInfo TStockWorkloadGenerator::SelectCustomerHistory(const std::string& customerId, const unsigned int limit) {
208-
std::string query = R"(--!syntax_v1
209-
DECLARE $cust as Utf8;
210-
DECLARE $limit as UInt32;
211-
select id, customer, created
212-
from orders view ix_cust
213-
where customer = $cust
214-
order by customer desc, created desc
215-
limit $limit;
216-
)";
230+
const std::string query = [this]() {
231+
return std::format(R"(--!syntax_v1
232+
DECLARE $cust as Utf8;
233+
DECLARE $limit as UInt32;
234+
select id, customer, created
235+
from orders {}
236+
where customer = $cust
237+
order by customer desc, created desc
238+
limit $limit;
239+
)", Params.GetStoreType() == TStockWorkloadParams::EStoreType::Row ? "view ix_cust" : "");
240+
}();
217241

218242
NYdb::TParamsBuilder paramsBuilder;
219243
paramsBuilder
@@ -304,6 +328,20 @@ TQueryInfoList TStockWorkloadGenerator::GetCustomerHistory() {
304328
}
305329

306330
void TStockWorkloadParams::ConfigureOpts(NLastGetopt::TOpts& opts, const ECommandType commandType, int workloadType) {
331+
auto addStorageTypeParam = [&]() {
332+
opts.AddLongOption("store", "Storage type."
333+
" Options: row, column\n"
334+
"row - use row-based storage engine;\n"
335+
"column - use column-based storage engine.")
336+
.DefaultValue(StoreType)
337+
.Handler1T<TStringBuf>([this](TStringBuf arg) {
338+
const auto l = to_lower(TString(arg));
339+
if (!TryFromString(arg, StoreType)) {
340+
throw yexception() << "Ivalid store type: " << arg;
341+
}
342+
});
343+
};
344+
307345
switch (commandType) {
308346
case TWorkloadParams::ECommandType::Init:
309347
opts.AddLongOption('p', "products", "Product count. Value in 1..500 000.")
@@ -318,8 +356,10 @@ void TStockWorkloadParams::ConfigureOpts(NLastGetopt::TOpts& opts, const EComman
318356
.DefaultValue(true).StoreResult(&PartitionsByLoad);
319357
opts.AddLongOption("enable-cdc", "Create changefeeds on tables.")
320358
.DefaultValue(false).StoreTrue(&EnableCdc).Hidden();
359+
addStorageTypeParam();
321360
break;
322361
case TWorkloadParams::ECommandType::Run:
362+
addStorageTypeParam();
323363
switch (static_cast<TStockWorkloadGenerator::EType>(workloadType)) {
324364
case TStockWorkloadGenerator::EType::InsertRandomOrder:
325365
case TStockWorkloadGenerator::EType::SubmitRandomOrder:

‎ydb/library/workload/stock/stock.h‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,11 @@ namespace NYdbWorkload {
1010

1111
class TStockWorkloadParams final: public TWorkloadParams {
1212
public:
13+
enum class EStoreType {
14+
Row /* "row" */,
15+
Column /* "column" */,
16+
};
17+
1318
void ConfigureOpts(NLastGetopt::TOpts& opts, const ECommandType commandType, int workloadType) override;
1419
THolder<IWorkloadQueryGenerator> CreateGenerator() const override;
1520
TString GetWorkloadName() const override;
@@ -20,6 +25,7 @@ class TStockWorkloadParams final: public TWorkloadParams {
2025
unsigned int Limit = 0;
2126
bool PartitionsByLoad = true;
2227
bool EnableCdc = false;
28+
YDB_READONLY(EStoreType, StoreType, EStoreType::Row);
2329
};
2430

2531
class TStockWorkloadGenerator final: public TWorkloadQueryGeneratorBase<TStockWorkloadParams> {

‎ydb/public/lib/ydb_cli/commands/ydb_workload.cpp‎

Lines changed: 4 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -482,18 +482,11 @@ int TWorkloadCommandInit::DoRun(NYdbWorkload::IWorkloadQueryGenerator& workloadG
482482
Cout << Endl;
483483
}
484484
} else {
485-
auto session = GetSession();
486485
for (auto queryInfo : queryInfoList) {
487-
auto prepareResult = session.PrepareDataQuery(queryInfo.Query.c_str()).GetValueSync();
488-
if (!prepareResult.IsSuccess()) {
489-
Cerr << "Prepare failed: " << prepareResult.GetIssues().ToString() << Endl
490-
<< "Query:\n" << queryInfo.Query << Endl;
491-
return EXIT_FAILURE;
492-
}
493-
494-
auto dataQuery = prepareResult.GetQuery();
495-
auto result = dataQuery.Execute(NYdb::NTable::TTxControl::BeginTx(NYdb::NTable::TTxSettings::SerializableRW()).CommitTx(),
496-
std::move(queryInfo.Params)).GetValueSync();
486+
auto result = QueryClient->ExecuteQuery(
487+
queryInfo.Query.c_str(),
488+
NYdb::NQuery::TTxControl::BeginTx(NYdb::NQuery::TTxSettings::SerializableRW()).CommitTx(),
489+
std::move(queryInfo.Params)).GetValueSync();
497490
if (!result.IsSuccess()) {
498491
Cerr << "Query execution failed: " << result.GetIssues().ToString() << Endl
499492
<< "Query:\n" << queryInfo.Query << Endl;

0 commit comments

Comments
 (0)