Skip to content

Commit 96a2811

Browse files
stanislav-shchetininstanislav-shchetinin
committed
fix after review
1 parent 68f0d4e commit 96a2811

File tree

4 files changed

+80
-63
lines changed

4 files changed

+80
-63
lines changed

‎include/ydb-cpp-sdk/util/datetime/base.h

Lines changed: 30 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -167,8 +167,6 @@ class TTimeBase {
167167
return Value_;
168168
}
169169

170-
auto operator<=>(const TTimeBase&) const = default;
171-
172170
protected:
173171
TValue Value_; // microseconds count
174172
};
@@ -583,6 +581,36 @@ ::NPrivate::TPrintableLocalTime<true, true> FormatIsoLocalUpToSeconds(TInstant i
583581
::NPrivate::TPrintableLocalTime<true, false> FormatLocalUpToSeconds(TInstant instant);
584582
///@}
585583

584+
template <class S>
585+
static constexpr bool operator<(const TTimeBase<S>& l, const TTimeBase<S>& r) noexcept {
586+
return l.GetValue() < r.GetValue();
587+
}
588+
589+
template <class S>
590+
static constexpr bool operator<=(const TTimeBase<S>& l, const TTimeBase<S>& r) noexcept {
591+
return l.GetValue() <= r.GetValue();
592+
}
593+
594+
template <class S>
595+
static constexpr bool operator==(const TTimeBase<S>& l, const TTimeBase<S>& r) noexcept {
596+
return l.GetValue() == r.GetValue();
597+
}
598+
599+
template <class S>
600+
static constexpr bool operator!=(const TTimeBase<S>& l, const TTimeBase<S>& r) noexcept {
601+
return l.GetValue() != r.GetValue();
602+
}
603+
604+
template <class S>
605+
static constexpr bool operator>(const TTimeBase<S>& l, const TTimeBase<S>& r) noexcept {
606+
return l.GetValue() > r.GetValue();
607+
}
608+
609+
template <class S>
610+
static constexpr bool operator>=(const TTimeBase<S>& l, const TTimeBase<S>& r) noexcept {
611+
return l.GetValue() >= r.GetValue();
612+
}
613+
586614
namespace NDateTimeHelpers {
587615
template <typename T>
588616
static constexpr T SumWithSaturation(T a, T b) {

‎tests/integration/bulk_upsert_simple_it/bulk_upsert.cpp

Lines changed: 37 additions & 49 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,10 @@ static std::string JoinPath(const std::string& basePath, const std::string& path
2323
return prefixPathSplit;
2424
}
2525

26+
bool TLogMessage::TPrimaryKeyLogMessage::operator<(const TLogMessage::TPrimaryKeyLogMessage& o) const {
27+
return App < o.App || App == o.App && Host < o.Host || App == o.App && Host == o.Host && Timestamp < o.Timestamp;
28+
}
29+
2630
TRunArgs GetRunArgs() {
2731

2832
std::string database = std::getenv("YDB_DATABASE");
@@ -57,26 +61,26 @@ TStatus CreateLogTable(TTableClient& client, const std::string& table) {
5761
return status;
5862
}
5963

60-
TStatistic GetLogBatch(uint64_t logOffset, std::vector<TLogMessage>& logBatch, std::set<TLogMessage>& setMessage) {
64+
TStatistic GetLogBatch(uint64_t logOffset, std::vector<TLogMessage>& logBatch, std::set<TLogMessage::TPrimaryKeyLogMessage>& setMessage) {
6165
logBatch.clear();
6266
uint32_t correctSumApp = 0;
6367
uint32_t correctSumHost = 0;
6468
uint32_t correctRowCount = 0;
6569

6670
for (size_t i = 0; i < BATCH_SIZE; ++i) {
6771
TLogMessage message;
68-
message.pk.App = "App_" + ToString(logOffset % 10);
69-
message.pk.Host = "192.168.0." + ToString(logOffset % 11);
72+
message.pk.App = "App_" + std::to_string(logOffset % 10);
73+
message.pk.Host = "192.168.0." + std::to_string(logOffset % 11);
7074
message.pk.Timestamp = TInstant::Now() + TDuration::MilliSeconds(i % 1000);
7175
message.HttpCode = 200;
7276
message.Message = i % 2 ? "GET / HTTP/1.1" : "GET /images/logo.png HTTP/1.1";
7377
logBatch.emplace_back(message);
7478

75-
if (!setMessage.contains(message)) {
79+
if (!setMessage.contains(message.pk)) {
7680
correctSumApp += logOffset % 10;
7781
correctSumHost += logOffset % 11;
7882
++correctRowCount;
79-
setMessage.insert(message);
83+
setMessage.insert(message.pk);
8084
}
8185

8286
}
@@ -108,65 +112,49 @@ TStatus WriteLogBatch(TTableClient& tableClient, const std::string& table, const
108112
return status;
109113
}
110114

111-
static TStatus ScanQuerySelect(TTableClient& client, const std::string& path, std::vector <TResultSet>& vectorResultSet) {
115+
static TStatus SelectTransaction(TSession session, const std::string& path,
116+
std::optional<TResultSet>& resultSet) {
112117
std::filesystem::path filesystemPath(path);
113118
auto query = std::format(R"(
114-
--!syntax_v1
115119
PRAGMA TablePathPrefix("{}");
116120
117-
SELECT *
121+
SELECT
122+
SUM(CAST(SUBSTRING(CAST(App as string), 4) as Int32)),
123+
SUM(CAST(SUBSTRING(CAST(Host as string), 10) as Int32)),
124+
COUNT(*)
118125
FROM {}
119-
)", filesystemPath.parent_path().c_str(), filesystemPath.filename().c_str());
126+
)", filesystemPath.parent_path().string(), filesystemPath.filename().string());
120127

121-
auto resultScanQuery = client.StreamExecuteScanQuery(query).GetValueSync();
122-
123-
if (!resultScanQuery.IsSuccess()) {
124-
return resultScanQuery;
125-
}
128+
auto txControl =
129+
TTxControl::BeginTx(TTxSettings::SerializableRW())
130+
.CommitTx();
126131

127-
bool eos = false;
128-
129-
while (!eos) {
130-
auto streamPart = resultScanQuery.ReadNext().ExtractValueSync();
131-
132-
if (!streamPart.IsSuccess()) {
133-
eos = true;
134-
if (!streamPart.EOS()) {
135-
return streamPart;
136-
}
137-
continue;
138-
}
132+
auto result = session.ExecuteDataQuery(query, txControl).GetValueSync();
139133

140-
if (streamPart.HasResultSet()) {
141-
auto rs = streamPart.ExtractResultSet();
142-
vectorResultSet.push_back(rs);
143-
}
134+
if (result.IsSuccess()) {
135+
resultSet = result.GetResultSet(0);
144136
}
145-
return TStatus(EStatus::SUCCESS, NYql::TIssues());
137+
138+
return result;
146139
}
147140

148-
TStatistic ScanQuerySelect(TTableClient& client, const std::string& path) {
149-
std::vector <TResultSet> vectorResultSet;
150-
ThrowOnError(client.RetryOperationSync([path, &vectorResultSet](TTableClient& client) {
151-
return ScanQuerySelect(client, path, vectorResultSet);
141+
TStatistic Select(TTableClient& client, const std::string& path) {
142+
std::optional<TResultSet> resultSet;
143+
ThrowOnError(client.RetryOperationSync([path, &resultSet](TSession session) {
144+
return SelectTransaction(session, path, resultSet);
152145
}));
153146

154-
uint32_t sumApp = 0;
155-
uint32_t sumHost = 0;
156-
uint32_t rowCount = 0;
147+
TResultSetParser parser(*resultSet);
157148

158-
for (TResultSet& resultSet : vectorResultSet) {
159-
TResultSetParser parser(resultSet);
160-
161-
while (parser.TryNextRow()) {
149+
uint64_t sumApp = 0;
150+
uint64_t sumHost = 0;
151+
uint64_t rowCount = 0;
162152

163-
++rowCount;
164-
sumApp += ToString(parser.ColumnParser("App").GetOptionalUtf8()).back() - '0';
165-
std::string strHost = ToString(parser.ColumnParser("Host").GetOptionalUtf8());
166-
char penCharStrHost = strHost[strHost.size() - 2];
167-
sumHost += strHost.back() - '0' + (penCharStrHost == '.' ? 0 : (penCharStrHost - '0') * 10);
168-
}
169-
153+
if (parser.TryNextRow()) {
154+
155+
sumApp = *parser.ColumnParser("column0").GetOptionalInt64();
156+
sumHost = *parser.ColumnParser("column1").GetOptionalInt64();
157+
rowCount = parser.ColumnParser("column2").GetUint64();
170158
}
171159

172160
return {sumApp, sumHost, rowCount};

‎tests/integration/bulk_upsert_simple_it/bulk_upsert.h

Lines changed: 8 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -11,19 +11,18 @@ struct TRunArgs {
1111
std::string path;
1212
};
1313

14-
class TLogMessage {
14+
struct TLogMessage {
1515
struct TPrimaryKeyLogMessage {
1616
std::string App;
1717
std::string Host;
1818
TInstant Timestamp;
19-
auto operator<=>(const TPrimaryKeyLogMessage&) const = default;
19+
bool operator<(const TPrimaryKeyLogMessage& o) const;
2020
};
2121

22-
public:
2322
TPrimaryKeyLogMessage pk;
2423
uint32_t HttpCode;
2524
std::string Message;
26-
auto operator<=>(const TLogMessage& o) const {return pk <=> o.pk;};
25+
bool operator<(const TLogMessage& o) const {return pk < o.pk;};
2726
};
2827

2928
class TYdbErrorException : public yexception {
@@ -35,15 +34,15 @@ class TYdbErrorException : public yexception {
3534
};
3635

3736
struct TStatistic {
38-
uint32_t sumApp;
39-
uint32_t sumHost;
40-
uint32_t rowCount;
37+
uint64_t sumApp;
38+
uint64_t sumHost;
39+
uint64_t rowCount;
4140
};
4241

4342
TRunArgs GetRunArgs();
4443
TStatus CreateLogTable(TTableClient& client, const std::string& table);
45-
TStatistic GetLogBatch(uint64_t logOffset, std::vector<TLogMessage>& logBatch, std::set<TLogMessage>& setMessage);
44+
TStatistic GetLogBatch(uint64_t logOffset, std::vector<TLogMessage>& logBatch, std::set<TLogMessage::TPrimaryKeyLogMessage>& setMessage);
4645
TStatus WriteLogBatch(TTableClient& tableClient, const std::string& table, const std::vector<TLogMessage>& logBatch,
4746
const TRetryOperationSettings& retrySettings);
48-
TStatistic ScanQuerySelect(TTableClient& client, const std::string& path);
47+
TStatistic Select(TTableClient& client, const std::string& path);
4948
void DropTable(TTableClient& client, const std::string& path);

‎tests/integration/bulk_upsert_simple_it/main.cpp

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,8 @@ TEST(Integration, BulkUpsert) {
88

99
uint32_t correctSumApp = 0;
1010
uint32_t correctSumHost = 0;
11-
std::set <TLogMessage> setMessage;
11+
uint32_t correctRowCount = 0;
12+
std::set <TLogMessage::TPrimaryKeyLogMessage> setMessage;
1213

1314
auto [driver, path] = GetRunArgs();
1415

@@ -30,16 +31,17 @@ TEST(Integration, BulkUpsert) {
3031
auto [batchSumApp, batchSumHost, batchRowCount] = GetLogBatch(offset, logBatch, setMessage);
3132
correctSumApp += batchSumApp;
3233
correctSumHost += batchSumHost;
34+
correctRowCount += batchRowCount;
3335

3436
TStatus statusWrite = WriteLogBatch(client, path, logBatch, writeRetrySettings);
3537
if (!statusWrite.IsSuccess()) {
3638
FAIL() << "Write failed with status: " << statusWrite << std::endl;
3739
}
3840
}
3941

40-
auto [sumApp, sumHost, rowCount] = ScanQuerySelect(client, path);
42+
auto [sumApp, sumHost, rowCount] = Select(client, path);
4143

42-
EXPECT_EQ(rowCount, setMessage.size());
44+
EXPECT_EQ(rowCount, correctRowCount);
4345
EXPECT_EQ(sumApp, correctSumApp);
4446
EXPECT_EQ(sumHost, correctSumHost);
4547

0 commit comments

Comments
 (0)