Skip to content

Commit 9e75040

Browse files
azevaykinazevaykin
authored
Statistics test refactoring (#7149)
1 parent 123a335 commit 9e75040

File tree

5 files changed

+232
-169
lines changed

5 files changed

+232
-169
lines changed

‎ydb/core/statistics/aggregator/aggregator_impl.cpp

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -218,7 +218,7 @@ void TStatisticsAggregator::Handle(TEvPrivate::TEvFastPropagateCheck::TPtr&) {
218218
}
219219

220220
void TStatisticsAggregator::Handle(TEvPrivate::TEvPropagate::TPtr&) {
221-
SA_LOG_D("[" << TabletID() << "] EvPropagate");
221+
SA_LOG_T("[" << TabletID() << "] EvPropagate");
222222

223223
if (EnableStatistics) {
224224
PropagateStatistics();
@@ -309,14 +309,15 @@ void TStatisticsAggregator::SendStatisticsToNode(TNodeId nodeId, const std::vect
309309
}
310310

311311
void TStatisticsAggregator::PropagateStatistics() {
312-
SA_LOG_D("[" << TabletID() << "] PropagateStatistics()"
313-
<< ", node count = " << Nodes.size()
314-
<< ", schemeshard count = " << RequestedSchemeShards.size());
315-
316312
if (Nodes.empty() || RequestedSchemeShards.empty()) {
313+
SA_LOG_T("[" << TabletID() << "] PropagateStatistics() No data");
317314
return;
318315
}
319316

317+
SA_LOG_D("[" << TabletID() << "] PropagateStatistics()"
318+
<< ", node count = " << Nodes.size()
319+
<< ", schemeshard count = " << RequestedSchemeShards.size());
320+
320321
std::vector<TNodeId> nodeIds;
321322
nodeIds.reserve(Nodes.size());
322323
for (const auto& [nodeId, _] : Nodes) {

‎ydb/core/statistics/aggregator/tx_schedule_scan.cpp

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@ struct TStatisticsAggregator::TTxScheduleScan : public TTxBase {
1212
TTxType GetTxType() const override { return TXTYPE_SCHEDULE_SCAN; }
1313

1414
bool Execute(TTransactionContext& txc, const TActorContext&) override {
15-
SA_LOG_D("[" << Self->TabletID() << "] TTxScheduleScan::Execute");
15+
SA_LOG_T("[" << Self->TabletID() << "] TTxScheduleScan::Execute");
1616

1717
Self->Schedule(Self->ScheduleScanIntervalTime, new TEvPrivate::TEvScheduleScan());
1818

@@ -30,7 +30,7 @@ struct TStatisticsAggregator::TTxScheduleScan : public TTxBase {
3030
}
3131

3232
void Complete(const TActorContext&) override {
33-
SA_LOG_D("[" << Self->TabletID() << "] TTxScheduleScan::Complete");
33+
SA_LOG_T("[" << Self->TabletID() << "] TTxScheduleScan::Complete");
3434
}
3535
};
3636

‎ydb/core/statistics/aggregator/ut/ut_aggregator.cpp

Lines changed: 10 additions & 156 deletions
Original file line numberDiff line numberDiff line change
@@ -7,157 +7,13 @@
77
#include <ydb/core/statistics/events.h>
88
#include <ydb/core/statistics/service/service.h>
99

10-
#include <ydb/public/sdk/cpp/client/ydb_result/result.h>
11-
#include <ydb/public/sdk/cpp/client/ydb_table/table.h>
12-
#include <ydb/public/sdk/cpp/client/ydb_scheme/scheme.h>
13-
1410
#include <thread>
1511

1612
namespace NKikimr {
1713
namespace NStat {
1814

19-
using namespace NYdb;
20-
using namespace NYdb::NTable;
21-
using namespace NYdb::NScheme;
22-
2315
namespace {
2416

25-
void CreateUniformTable(TTestEnv& env, const TString& databaseName, const TString& tableName) {
26-
TTableClient client(env.GetDriver());
27-
auto session = client.CreateSession().GetValueSync().GetSession();
28-
29-
auto result = session.ExecuteSchemeQuery(Sprintf(R"(
30-
CREATE TABLE `Root/%s/%s` (
31-
Key Uint64,
32-
Value Uint64,
33-
PRIMARY KEY (Key)
34-
)
35-
WITH ( UNIFORM_PARTITIONS = 4 );
36-
)", databaseName.c_str(), tableName.c_str())).GetValueSync();
37-
UNIT_ASSERT_C(result.IsSuccess(), result.GetIssues().ToString());
38-
39-
TStringBuilder replace;
40-
replace << Sprintf("REPLACE INTO `Root/%s/%s` (Key, Value) VALUES ",
41-
databaseName.c_str(), tableName.c_str());
42-
for (ui32 i = 0; i < 4; ++i) {
43-
if (i > 0) {
44-
replace << ", ";
45-
}
46-
ui64 value = 4000000000000000000ull * (i + 1);
47-
replace << Sprintf("(%" PRIu64 "ul, %" PRIu64 "ul)", value, value);
48-
}
49-
replace << ";";
50-
result = session.ExecuteDataQuery(replace, TTxControl::BeginTx().CommitTx()).GetValueSync();
51-
UNIT_ASSERT_C(result.IsSuccess(), result.GetIssues().ToString());
52-
}
53-
54-
void CreateColumnStoreTable(TTestEnv& env, const TString& databaseName, const TString& tableName,
55-
int shardCount)
56-
{
57-
TTableClient client(env.GetDriver());
58-
auto session = client.CreateSession().GetValueSync().GetSession();
59-
60-
auto fullTableName = Sprintf("Root/%s/%s", databaseName.c_str(), tableName.c_str());
61-
auto result = session.ExecuteSchemeQuery(Sprintf(R"(
62-
CREATE TABLE `%s` (
63-
Key Uint64 NOT NULL,
64-
Value Uint64,
65-
PRIMARY KEY (Key)
66-
)
67-
PARTITION BY HASH(Key)
68-
WITH (
69-
STORE = COLUMN,
70-
AUTO_PARTITIONING_MIN_PARTITIONS_COUNT = %d
71-
);
72-
)", fullTableName.c_str(), shardCount)).GetValueSync();
73-
UNIT_ASSERT_C(result.IsSuccess(), result.GetIssues().ToString());
74-
75-
NYdb::TValueBuilder rows;
76-
rows.BeginList();
77-
for (size_t i = 0; i < 100; ++i) {
78-
auto key = TValueBuilder().Uint64(i).Build();
79-
auto value = TValueBuilder().OptionalUint64(i).Build();
80-
rows.AddListItem();
81-
rows.BeginStruct();
82-
rows.AddMember("Key", key);
83-
rows.AddMember("Value", value);
84-
rows.EndStruct();
85-
}
86-
rows.EndList();
87-
88-
result = client.BulkUpsert(fullTableName, rows.Build()).GetValueSync();
89-
UNIT_ASSERT_C(result.IsSuccess(), result.GetIssues().ToString());
90-
}
91-
92-
void DropTable(TTestEnv& env, const TString& databaseName, const TString& tableName) {
93-
TTableClient client(env.GetDriver());
94-
auto session = client.CreateSession().GetValueSync().GetSession();
95-
96-
auto result = session.ExecuteSchemeQuery(Sprintf(R"(
97-
DROP TABLE `Root/%s/%s`;
98-
)", databaseName.c_str(), tableName.c_str())).GetValueSync();
99-
UNIT_ASSERT_C(result.IsSuccess(), result.GetIssues().ToString());
100-
}
101-
102-
std::shared_ptr<TCountMinSketch> ExtractCountMin(TTestActorRuntime& runtime, TPathId pathId) {
103-
auto statServiceId = NStat::MakeStatServiceID(runtime.GetNodeId(1));
104-
105-
NStat::TRequest req;
106-
req.PathId = pathId;
107-
req.ColumnTag = 1;
108-
109-
auto evGet = std::make_unique<TEvStatistics::TEvGetStatistics>();
110-
evGet->StatType = NStat::EStatType::COUNT_MIN_SKETCH;
111-
evGet->StatRequests.push_back(req);
112-
113-
auto sender = runtime.AllocateEdgeActor(1);
114-
runtime.Send(statServiceId, sender, evGet.release(), 1, true);
115-
auto evResult = runtime.GrabEdgeEventRethrow<TEvStatistics::TEvGetStatisticsResult>(sender);
116-
117-
UNIT_ASSERT(evResult);
118-
UNIT_ASSERT(evResult->Get());
119-
UNIT_ASSERT(evResult->Get()->StatResponses.size() == 1);
120-
121-
auto rsp = evResult->Get()->StatResponses[0];
122-
auto stat = rsp.CountMinSketch;
123-
UNIT_ASSERT(rsp.Success);
124-
UNIT_ASSERT(stat.CountMin);
125-
126-
return stat.CountMin;
127-
}
128-
129-
void ValidateCountMin(TTestActorRuntime& runtime, TPathId pathId) {
130-
auto countMin = ExtractCountMin(runtime, pathId);
131-
132-
for (ui32 i = 0; i < 4; ++i) {
133-
ui64 value = 4000000000000000000ull * (i + 1);
134-
auto probe = countMin->Probe((const char *)&value, sizeof(ui64));
135-
UNIT_ASSERT_VALUES_EQUAL(probe, 1);
136-
}
137-
}
138-
139-
void ValidateCountMinAbsense(TTestActorRuntime& runtime, TPathId pathId) {
140-
auto statServiceId = NStat::MakeStatServiceID(runtime.GetNodeId(1));
141-
142-
NStat::TRequest req;
143-
req.PathId = pathId;
144-
req.ColumnTag = 1;
145-
146-
auto evGet = std::make_unique<TEvStatistics::TEvGetStatistics>();
147-
evGet->StatType = NStat::EStatType::COUNT_MIN_SKETCH;
148-
evGet->StatRequests.push_back(req);
149-
150-
auto sender = runtime.AllocateEdgeActor(1);
151-
runtime.Send(statServiceId, sender, evGet.release(), 1, true);
152-
auto evResult = runtime.GrabEdgeEventRethrow<TEvStatistics::TEvGetStatisticsResult>(sender);
153-
154-
UNIT_ASSERT(evResult);
155-
UNIT_ASSERT(evResult->Get());
156-
UNIT_ASSERT(evResult->Get()->StatResponses.size() == 1);
157-
158-
auto rsp = evResult->Get()->StatResponses[0];
159-
UNIT_ASSERT(!rsp.Success);
160-
}
16117

16218
} // namespace
16319

@@ -175,8 +31,8 @@ Y_UNIT_TEST_SUITE(StatisticsAggregator) {
17531
runtime.SimulateSleep(TDuration::Seconds(5));
17632
initThread.join();
17733

178-
ui64 tabletId;
179-
auto pathId = ResolvePathId(runtime, "/Root/Database/Table", nullptr, &tabletId);
34+
ui64 saTabletId;
35+
auto pathId = ResolvePathId(runtime, "/Root/Database/Table", nullptr, &saTabletId);
18036

18137
runtime.SimulateSleep(TDuration::Seconds(30));
18238

@@ -185,7 +41,7 @@ Y_UNIT_TEST_SUITE(StatisticsAggregator) {
18541
PathIdFromPathId(pathId, record.AddTables()->MutablePathId());
18642

18743
auto sender = runtime.AllocateEdgeActor();
188-
runtime.SendToPipe(tabletId, sender, ev.release());
44+
runtime.SendToPipe(saTabletId, sender, ev.release());
18945
runtime.GrabEdgeEventRethrow<TEvStatistics::TEvAnalyzeResponse>(sender);
19046

19147
ValidateCountMin(runtime, pathId);
@@ -208,8 +64,8 @@ Y_UNIT_TEST_SUITE(StatisticsAggregator) {
20864
// TODO remove sleep
20965
runtime.SimulateSleep(TDuration::Seconds(30));
21066

211-
ui64 tabletId1;
212-
auto pathId1 = ResolvePathId(runtime, "/Root/Database/Table1", nullptr, &tabletId1);
67+
ui64 saTabletId1;
68+
auto pathId1 = ResolvePathId(runtime, "/Root/Database/Table1", nullptr, &saTabletId1);
21369
auto pathId2 = ResolvePathId(runtime, "/Root/Database/Table2");
21470

21571
auto ev = std::make_unique<TEvStatistics::TEvAnalyze>();
@@ -218,7 +74,7 @@ Y_UNIT_TEST_SUITE(StatisticsAggregator) {
21874
PathIdFromPathId(pathId2, record.AddTables()->MutablePathId());
21975

22076
auto sender = runtime.AllocateEdgeActor();
221-
runtime.SendToPipe(tabletId1, sender, ev.release());
77+
runtime.SendToPipe(saTabletId1, sender, ev.release());
22278
runtime.GrabEdgeEventRethrow<TEvStatistics::TEvAnalyzeResponse>(sender);
22379
runtime.GrabEdgeEventRethrow<TEvStatistics::TEvAnalyzeResponse>(sender);
22480

@@ -335,8 +191,8 @@ Y_UNIT_TEST_SUITE(StatisticsAggregator) {
335191
runtime.SimulateSleep(TDuration::Seconds(5));
336192
initThread.join();
337193

338-
ui64 tabletId = 0;
339-
auto pathId = ResolvePathId(runtime, "/Root/Database/Table", nullptr, &tabletId);
194+
ui64 saTabletId = 0;
195+
auto pathId = ResolvePathId(runtime, "/Root/Database/Table", nullptr, &saTabletId);
340196

341197
auto init2 = [&] () {
342198
DropTable(env, "Database", "Table");
@@ -351,7 +207,7 @@ Y_UNIT_TEST_SUITE(StatisticsAggregator) {
351207
PathIdFromPathId(pathId, record.AddTables()->MutablePathId());
352208

353209
auto sender = runtime.AllocateEdgeActor();
354-
runtime.SendToPipe(tabletId, sender, ev.release());
210+
runtime.SendToPipe(saTabletId, sender, ev.release());
355211

356212
runtime.SimulateSleep(TDuration::Seconds(60));
357213

@@ -370,9 +226,7 @@ Y_UNIT_TEST_SUITE(StatisticsAggregator) {
370226
runtime.SimulateSleep(TDuration::Seconds(30));
371227
initThread.join();
372228

373-
ui64 tabletId = 0;
374-
auto pathId = ResolvePathId(runtime, "/Root/Database/Table", nullptr, &tabletId);
375-
Y_UNUSED(pathId);
229+
auto pathId = ResolvePathId(runtime, "/Root/Database/Table");
376230

377231
runtime.SimulateSleep(TDuration::Seconds(30));
378232

0 commit comments

Comments
 (0)