Skip to content

Commit e5fc449

Browse files
CyberROFLCyberROFL
authored
TEvRecords helpers, little style guide changes (#8421)
1 parent dee88c3 commit e5fc449

10 files changed

+63
-71
lines changed

‎ydb/core/change_exchange/change_exchange.cpp

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -86,11 +86,6 @@ TString TEvChangeExchange::TEvRemoveRecords::ToString() const {
8686
}
8787

8888
// TEvRecords
89-
TEvChangeExchange::TEvRecords::TEvRecords(const TChangeRecordVector& records)
90-
: Records(records)
91-
{
92-
}
93-
9489
TEvChangeExchange::TEvRecords::TEvRecords(TChangeRecordVector&& records)
9590
: Records(std::move(records))
9691
{

‎ydb/core/change_exchange/change_exchange.h

Lines changed: 19 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,9 @@
1010
#include <util/string/builder.h>
1111
#include <util/string/join.h>
1212

13+
#include <memory>
14+
#include <variant>
15+
1316
namespace NKikimr {
1417

1518
namespace NDataShard {
@@ -51,12 +54,6 @@ struct TChangeRecordContainer {};
5154

5255
namespace NKikimr::NChangeExchange {
5356

54-
using TChangeRecordVector = std::variant<
55-
std::shared_ptr<TChangeRecordContainer<NDataShard::TChangeRecord>>,
56-
std::shared_ptr<TChangeRecordContainer<NReplication::NService::TChangeRecord>>,
57-
std::shared_ptr<TChangeRecordContainer<NBackup::NImpl::TChangeRecord>>
58-
>;
59-
6057
struct TEvChangeExchange {
6158
enum EEv {
6259
// Enqueue for sending
@@ -120,11 +117,26 @@ struct TEvChangeExchange {
120117
};
121118

122119
struct TEvRecords: public TEventLocal<TEvRecords, EvRecords> {
120+
using TChangeRecordVector = std::variant<
121+
std::shared_ptr<TChangeRecordContainer<NDataShard::TChangeRecord>>,
122+
std::shared_ptr<TChangeRecordContainer<NReplication::NService::TChangeRecord>>,
123+
std::shared_ptr<TChangeRecordContainer<NBackup::NImpl::TChangeRecord>>
124+
>;
125+
123126
TChangeRecordVector Records;
124127

125-
explicit TEvRecords(const TChangeRecordVector& records);
126128
explicit TEvRecords(TChangeRecordVector&& records);
127129
TString ToString() const override;
130+
131+
template <typename T>
132+
static TEvRecords* New(TVector<typename T::TPtr>&& records) {
133+
return new TEvRecords(std::make_shared<TChangeRecordContainer<T>>(std::move(records)));
134+
}
135+
136+
template <typename T>
137+
inline auto& GetRecords() {
138+
return std::get<std::shared_ptr<TChangeRecordContainer<T>>>(Records)->Records;
139+
}
128140
};
129141

130142
struct TEvForgetRecords: public TEventLocal<TEvForgetRecords, EvForgetRecods> {

‎ydb/core/change_exchange/change_sender_common_ops.h

Lines changed: 5 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,8 @@
11
#pragma once
22

33
#include "change_exchange.h"
4-
#include "change_sender_resolver.h"
54
#include "change_sender_partitioner.h"
5+
#include "change_sender_resolver.h"
66

77
#include <ydb/core/change_exchange/change_sender_monitoring.h>
88

@@ -18,8 +18,6 @@
1818
#include <util/generic/set.h>
1919
#include <util/string/builder.h>
2020

21-
#include <concepts>
22-
2321
namespace NKikimr::NChangeExchange {
2422

2523
struct TEvChangeExchangePrivate {
@@ -72,7 +70,7 @@ class ISenderFactory {
7270
virtual IActor* CreateSender(ui64 partitionId) const = 0;
7371
};
7472

75-
template <class TChangeRecord>
73+
template <typename TChangeRecord>
7674
class TBaseChangeSender {
7775
using TIncompleteRecord = TEvChangeExchange::TEvRequestRecords::TRecordInfo;
7876
// we need this to safely cast and call Out on a container
@@ -291,7 +289,7 @@ class TBaseChangeSender {
291289
}
292290

293291
Y_ABORT_UNLESS(sender.ActorId);
294-
ActorOps->Send(sender.ActorId, new TEvChangeExchange::TEvRecords(std::make_shared<TChangeRecordContainer<TChangeRecord>>(std::exchange(sender.Prepared, {}))));
292+
ActorOps->Send(sender.ActorId, TEvChangeExchange::TEvRecords::New<TChangeRecord>(std::exchange(sender.Prepared, {})));
295293
}
296294

297295
void ReEnqueueRecords(const TSender& sender) {
@@ -767,6 +765,8 @@ class TBaseChangeSender {
767765
IActorOps* const ActorOps;
768766
IChangeSenderResolver* const Resolver;
769767
ISenderFactory* const SenderFactory;
768+
THolder<IChangeSenderPartitioner<TChangeRecord>> Partitioner;
769+
770770
protected:
771771
TActorId ChangeServer;
772772
const TPathId PathId;
@@ -783,7 +783,6 @@ class TBaseChangeSender {
783783

784784
TVector<ui64> GonePartitions;
785785

786-
THolder<IChangeSenderPartitioner<TChangeRecord>> Partitioner;
787786
}; // TBaseChangeSender
788787

789788
}

‎ydb/core/change_exchange/change_sender_partitioner.cpp

Lines changed: 23 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -3,30 +3,29 @@
33
namespace NKikimr::NChangeExchange {
44

55
ui64 ResolveSchemaBoundaryPartitionId(const NKikimr::TKeyDesc& keyDesc, TConstArrayRef<TCell> key) {
6-
const auto& partitions = keyDesc.GetPartitions();
7-
Y_ABORT_UNLESS(partitions);
8-
const auto& schema = keyDesc.KeyColumnTypes;
9-
10-
const auto range = TTableRange(key);
11-
Y_ABORT_UNLESS(range.Point);
12-
13-
const auto it = LowerBound(
14-
partitions.cbegin(), partitions.cend(), true,
15-
[&](const auto& partition, bool) {
16-
Y_ABORT_UNLESS(partition.Range);
17-
const int compares = CompareBorders<true, false>(
18-
partition.Range->EndKeyPrefix.GetCells(), range.From,
19-
partition.Range->IsInclusive || partition.Range->IsPoint,
20-
range.InclusiveFrom || range.Point, schema
21-
);
22-
23-
return (compares < 0);
24-
}
25-
);
26-
27-
Y_ABORT_UNLESS(it != partitions.cend());
28-
return it->ShardId;
6+
const auto& partitions = keyDesc.GetPartitions();
7+
Y_ABORT_UNLESS(partitions);
8+
const auto& schema = keyDesc.KeyColumnTypes;
9+
10+
const auto range = TTableRange(key);
11+
Y_ABORT_UNLESS(range.Point);
12+
13+
const auto it = LowerBound(
14+
partitions.cbegin(), partitions.cend(), true,
15+
[&](const auto& partition, bool) {
16+
Y_ABORT_UNLESS(partition.Range);
17+
const int compares = CompareBorders<true, false>(
18+
partition.Range->EndKeyPrefix.GetCells(), range.From,
19+
partition.Range->IsInclusive || partition.Range->IsPoint,
20+
range.InclusiveFrom || range.Point, schema
21+
);
22+
23+
return (compares < 0);
24+
}
25+
);
26+
27+
Y_ABORT_UNLESS(it != partitions.cend());
28+
return it->ShardId;
2929
}
3030

31-
3231
} // NKikimr::NChangeExchange

‎ydb/core/change_exchange/change_sender_partitioner.h

Lines changed: 8 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -4,26 +4,24 @@
44
#include <ydb/core/scheme/scheme_tabledefs.h>
55
#include <ydb/core/scheme_types/scheme_type_info.h>
66

7-
#include <util/generic/vector.h>
8-
97
namespace NKikimr::NChangeExchange {
108

11-
template<typename TChangeRecord>
9+
template <typename TChangeRecord>
1210
class IChangeSenderPartitioner {
1311
public:
1412
virtual ~IChangeSenderPartitioner() = default;
1513

1614
virtual ui64 ResolvePartitionId(const typename TChangeRecord::TPtr& record) const = 0;
1715
};
1816

19-
2017
ui64 ResolveSchemaBoundaryPartitionId(const NKikimr::TKeyDesc& keyDesc, TConstArrayRef<TCell> key);
2118

22-
template<typename TChangeRecord>
23-
class SchemaBoundaryPartitioner final : public IChangeSenderPartitioner<TChangeRecord> {
19+
template <typename TChangeRecord>
20+
class TSchemaBoundaryPartitioner final : public IChangeSenderPartitioner<TChangeRecord> {
2421
public:
25-
SchemaBoundaryPartitioner(const NKikimr::TKeyDesc& keyDesc)
26-
: KeyDesc(keyDesc) {
22+
TSchemaBoundaryPartitioner(const NKikimr::TKeyDesc& keyDesc)
23+
: KeyDesc(keyDesc)
24+
{
2725
}
2826

2927
ui64 ResolvePartitionId(const typename TChangeRecord::TPtr& record) const override {
@@ -34,11 +32,9 @@ class SchemaBoundaryPartitioner final : public IChangeSenderPartitioner<TChangeR
3432
const NKikimr::TKeyDesc& KeyDesc;
3533
};
3634

37-
38-
template<typename TChangeRecord>
35+
template <typename TChangeRecord>
3936
IChangeSenderPartitioner<TChangeRecord>* CreateSchemaBoundaryPartitioner(const NKikimr::TKeyDesc& keyDesc) {
40-
return new SchemaBoundaryPartitioner<TChangeRecord>(keyDesc);
37+
return new TSchemaBoundaryPartitioner<TChangeRecord>(keyDesc);
4138
}
4239

43-
4440
} // NKikimr::NChangeExchange

‎ydb/core/change_exchange/change_sender_resolver.h

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -4,8 +4,6 @@
44
#include <ydb/core/scheme/scheme_tabledefs.h>
55
#include <ydb/core/scheme_types/scheme_type_info.h>
66

7-
#include <util/generic/vector.h>
8-
97
namespace NKikimr::NChangeExchange {
108

119
class IChangeSenderResolver {

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

Lines changed: 2 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -128,9 +128,7 @@ class TAsyncIndexChangeSenderShard: public TActorBootstrapped<TAsyncIndexChangeS
128128
records->Record.SetOrigin(DataShard.TabletId);
129129
records->Record.SetGeneration(DataShard.Generation);
130130

131-
auto& evRecords = std::get<std::shared_ptr<TChangeRecordContainer<NKikimr::NDataShard::TChangeRecord>>>(ev->Get()->Records)->Records;
132-
133-
for (auto& recordPtr : evRecords) {
131+
for (auto recordPtr : ev->Get()->GetRecords<TChangeRecord>()) {
134132
const auto& record = *recordPtr;
135133

136134
if (record.GetOrder() <= LastRecordOrder) {
@@ -728,8 +726,7 @@ class TAsyncIndexChangeSenderMain
728726

729727
void Handle(NChangeExchange::TEvChangeExchange::TEvRecords::TPtr& ev) {
730728
LOG_D("Handle " << ev->Get()->ToString());
731-
auto& records = std::get<std::shared_ptr<TChangeRecordContainer<NKikimr::NDataShard::TChangeRecord>>>(ev->Get()->Records)->Records;
732-
ProcessRecords(std::move(records));
729+
ProcessRecords(std::move(ev->Get()->GetRecords<TChangeRecord>()));
733730
}
734731

735732
void Handle(NChangeExchange::TEvChangeExchange::TEvForgetRecords::TPtr& ev) {

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

Lines changed: 2 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -95,8 +95,7 @@ class TCdcChangeSenderPartition: public TActorBootstrapped<TCdcChangeSenderParti
9595
LOG_D("Handle " << ev->Get()->ToString());
9696
NKikimrClient::TPersQueueRequest request;
9797

98-
auto& records = std::get<std::shared_ptr<TChangeRecordContainer<NKikimr::NDataShard::TChangeRecord>>>(ev->Get()->Records)->Records;
99-
for (auto recordPtr : records) {
98+
for (auto recordPtr : ev->Get()->GetRecords<TChangeRecord>()) {
10099
const auto& record = *recordPtr;
101100

102101
if (record.GetSeqNo() <= MaxSeqNo) {
@@ -699,8 +698,7 @@ class TCdcChangeSenderMain
699698

700699
void Handle(NChangeExchange::TEvChangeExchange::TEvRecords::TPtr& ev) {
701700
LOG_D("Handle " << ev->Get()->ToString());
702-
auto& records = std::get<std::shared_ptr<TChangeRecordContainer<NKikimr::NDataShard::TChangeRecord>>>(ev->Get()->Records)->Records;
703-
ProcessRecords(std::move(records));
701+
ProcessRecords(std::move(ev->Get()->GetRecords<TChangeRecord>()));
704702
}
705703

706704
void Handle(NChangeExchange::TEvChangeExchange::TEvForgetRecords::TPtr& ev) {

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -233,7 +233,7 @@ class TDataShard::TTxRequestChangeRecords: public TTransactionBase<TDataShard> {
233233
LOG_DEBUG_S(ctx, NKikimrServices::TX_DATASHARD, "Send " << records.size() << " change records"
234234
<< ": to# " << to
235235
<< ", at tablet# " << Self->TabletID());
236-
ctx.Send(to, new NChangeExchange::TEvChangeExchange::TEvRecords(std::make_shared<TChangeRecordContainer<NKikimr::NDataShard::TChangeRecord>>(std::move(records))));
236+
ctx.Send(to, NChangeExchange::TEvChangeExchange::TEvRecords::New<TChangeRecord>(std::move(records)));
237237
}
238238

239239
size_t forgotten = 0;

‎ydb/core/tx/replication/service/table_writer_impl.h

Lines changed: 3 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,7 @@
2222

2323
namespace NKikimr::NReplication::NService {
2424

25-
template <class TChangeRecord>
25+
template <typename TChangeRecord>
2626
class TTablePartitionWriter: public TActorBootstrapped<TTablePartitionWriter<TChangeRecord>> {
2727
using TBase = TActorBootstrapped<TTablePartitionWriter<TChangeRecord>>;
2828
using TThis = TTablePartitionWriter;
@@ -83,9 +83,7 @@ class TTablePartitionWriter: public TActorBootstrapped<TTablePartitionWriter<TCh
8383

8484
TString source;
8585

86-
auto& records = std::get<std::shared_ptr<TChangeRecordContainer<TChangeRecord>>>(ev->Get()->Records)->Records;
87-
88-
for (auto recordPtr : records) {
86+
for (auto recordPtr : ev->Get()->GetRecords<TChangeRecord>()) {
8987
const auto& record = *recordPtr;
9088
record.Serialize(*event->Record.AddChanges(), BuilderContext);
9189

@@ -213,7 +211,7 @@ class TTablePartitionWriter: public TActorBootstrapped<TTablePartitionWriter<TCh
213211

214212
}; // TTablePartitionWriter
215213

216-
template <class TChangeRecord>
214+
template <typename TChangeRecord>
217215
class TLocalTableWriter
218216
: public TActor<TLocalTableWriter<TChangeRecord>>
219217
, public NChangeExchange::TBaseChangeSender<TChangeRecord>

0 commit comments

Comments
 (0)