Rename TChangeRecordBuilderContextTrait to TSerializationContext, move it inside TChangeRecord (#9178)

Author: CyberROFL

ydb/core/backup/impl/table_writer_impl.h

@@ -7,28 +7,6 @@
 #include <ydb/core/change_exchange/change_record.h>
 #include <ydb/core/protos/change_exchange.pb.h>
 
-namespace NKikimr {
-
-namespace NBackup::NImpl {
-
-class TChangeRecord;
-
-} // namespace NBackup::NImpl
-
-template <>
-struct TChangeRecordBuilderContextTrait<NBackup::NImpl::TChangeRecord> {
-    NBackup::NImpl::EWriterType Type;
-
-    TChangeRecordBuilderContextTrait(NBackup::NImpl::EWriterType type)
-        : Type(type)
-    {}
-
-    // just copy type
-    TChangeRecordBuilderContextTrait(const TChangeRecordBuilderContextTrait<NBackup::NImpl::TChangeRecord>& other) = default;
-};
-
-} // namespace NKikimr
-
 namespace NKikimr::NBackup::NImpl {
 
 class TChangeRecordBuilder;
@@ -40,6 +18,17 @@ class TChangeRecord: public NChangeExchange::TChangeRecordBase {
     using TPtr = TIntrusivePtr<TChangeRecord>;
     using TBuilder = TChangeRecordBuilder;
 
+    struct TSerializationContext {
+        const NBackup::NImpl::EWriterType Type;
+
+        TSerializationContext(NBackup::NImpl::EWriterType type)
+            : Type(type)
+        {}
+
+        // just copy type
+        TSerializationContext(const TSerializationContext& other) = default;
+    };
+
     const static NKikimrSchemeOp::ECdcStreamFormat StreamType = NKikimrSchemeOp::ECdcStreamFormatProto;
 
     ui64 GetGroup() const override {
@@ -58,10 +47,7 @@ class TChangeRecord: public NChangeExchange::TChangeRecordBase {
         return SourceId;
     }
 
-    void Serialize(
-        NKikimrTxDataShard::TEvApplyReplicationChanges::TChange& record,
-        TChangeRecordBuilderContextTrait<TChangeRecord>& ctx) const
-    {
+    void Serialize(NKikimrTxDataShard::TEvApplyReplicationChanges::TChange& record, TSerializationContext& ctx) const {
         switch (ctx.Type) {
             case EWriterType::Backup:
                 return SerializeBackup(record);

ydb/core/backup/impl/table_writer_ut.cpp

@@ -48,7 +48,7 @@ Y_UNIT_TEST_SUITE(TableWriter) {
                     .Build();
 
             NKikimrTxDataShard::TEvApplyReplicationChanges_TChange result;
-            TChangeRecordBuilderContextTrait<NBackup::NImpl::TChangeRecord> ctx(EWriterType::Backup);
+            NBackup::NImpl::TChangeRecord::TSerializationContext ctx(EWriterType::Backup);
             record->Serialize(result, ctx);
 
             TVector<TCell> outCells{
@@ -89,7 +89,7 @@ Y_UNIT_TEST_SUITE(TableWriter) {
                     .Build();
 
             NKikimrTxDataShard::TEvApplyReplicationChanges_TChange result;
-            TChangeRecordBuilderContextTrait<NBackup::NImpl::TChangeRecord> ctx(EWriterType::Backup);
+            NBackup::NImpl::TChangeRecord::TSerializationContext ctx(EWriterType::Backup);
             record->Serialize(result, ctx);
 
             TVector<TCell> outCells{
@@ -144,7 +144,7 @@ Y_UNIT_TEST_SUITE(TableWriter) {
                     .Build();
 
             NKikimrTxDataShard::TEvApplyReplicationChanges_TChange result;
-            TChangeRecordBuilderContextTrait<NBackup::NImpl::TChangeRecord> ctx(EWriterType::Restore);
+            NBackup::NImpl::TChangeRecord::TSerializationContext ctx(EWriterType::Restore);
             record->Serialize(result, ctx);
 
             UNIT_ASSERT_VALUES_EQUAL(TSerializedCellVec::Serialize(keyCells), result.GetKey());

ydb/core/change_exchange/change_record.h

@@ -4,13 +4,6 @@
 #include <util/generic/string.h>
 #include <util/stream/output.h>
 
-namespace NKikimr {
-
-template <typename TChangeRecord>
-struct TChangeRecordBuilderContextTrait {};
-
-} // namespace NKikimr
-
 namespace NKikimr::NChangeExchange {
 
 class IChangeSenderResolver;

ydb/core/tx/replication/service/json_change_record.h

@@ -27,12 +27,23 @@ class TChangeRecordBuilder;
 
 class TChangeRecord: public NChangeExchange::TChangeRecordBase {
     friend class TChangeRecordBuilder;
-    using TSerializationContext = TChangeRecordBuilderContextTrait<TChangeRecord>;
 
 public:
     using TPtr = TIntrusivePtr<TChangeRecord>;
     using TBuilder = TChangeRecordBuilder;
 
+    struct TSerializationContext {
+        TMemoryPool MemoryPool;
+
+        TSerializationContext()
+            : MemoryPool(256)
+        {}
+
+        TSerializationContext(const TSerializationContext&)
+            : MemoryPool(256) // do not preserve any state between writers, just construct new one.
+        {}
+    };
+
     const static NKikimrSchemeOp::ECdcStreamFormat StreamType = NKikimrSchemeOp::ECdcStreamFormatJson;
 
     ui64 GetGroup() const override;
@@ -89,20 +100,6 @@ struct TChangeRecordContainer<NReplication::NService::TChangeRecord>
     using TBaseChangeRecordContainer<NReplication::NService::TChangeRecord>::TBaseChangeRecordContainer;
 };
 
-template <>
-struct TChangeRecordBuilderContextTrait<NReplication::NService::TChangeRecord> {
-    TMemoryPool MemoryPool;
-
-    TChangeRecordBuilderContextTrait()
-        : MemoryPool(256)
-    {}
-
-    // do not preserve any state between writers, just construct new one.
-    TChangeRecordBuilderContextTrait(const TChangeRecordBuilderContextTrait<NReplication::NService::TChangeRecord>&)
-        : MemoryPool(256)
-    {}
-};
-
 }
 
 Y_DECLARE_OUT_SPEC(inline, NKikimr::NReplication::NService::TChangeRecord, out, value) {

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

@@ -22,6 +22,12 @@
 
 namespace NKikimr::NReplication::NService {
 
+template <typename TChangeRecord>
+using TSerializationContext = typename TChangeRecord::TSerializationContext;
+
+template <typename TChangeRecord>
+using TBuilder = typename TChangeRecord::TBuilder;
+
 template <typename TChangeRecord>
 class TTablePartitionWriter: public TActorBootstrapped<TTablePartitionWriter<TChangeRecord>> {
     using TBase = TActorBootstrapped<TTablePartitionWriter<TChangeRecord>>;
@@ -85,7 +91,7 @@ class TTablePartitionWriter: public TActorBootstrapped<TTablePartitionWriter<TCh
 
         for (auto recordPtr : ev->Get()->GetRecords<TChangeRecord>()) {
             const auto& record = *recordPtr;
-            record.Serialize(*event->Record.AddChanges(), BuilderContext);
+            record.Serialize(*event->Record.AddChanges(), SerializationContext);
 
             if (!source) {
                 source = record.GetSourceId();
@@ -180,11 +186,11 @@ class TTablePartitionWriter: public TActorBootstrapped<TTablePartitionWriter<TCh
             const TActorId& parent,
             ui64 tabletId,
             const TTableId& tableId,
-            TChangeRecordBuilderContextTrait<TChangeRecord> builderContext)
+            const TSerializationContext<TChangeRecord>& ctx)
         : Parent(parent)
         , TabletId(tabletId)
         , TableId(tableId)
-        , BuilderContext(builderContext)
+        , SerializationContext(ctx)
     {}
 
     void Bootstrap() {
@@ -207,7 +213,7 @@ class TTablePartitionWriter: public TActorBootstrapped<TTablePartitionWriter<TCh
 
     TActorId LeaderPipeCache;
     ui64 SubscribeCookie = 0;
-    TChangeRecordBuilderContextTrait<TChangeRecord> BuilderContext;
+    TSerializationContext<TChangeRecord> SerializationContext;
 
 }; // TTablePartitionWriter
 
@@ -435,11 +441,8 @@ class TLocalTableWriter
     }
 
     IActor* CreateSender(ui64 partitionId) const override {
-        return new TTablePartitionWriter<TChangeRecord>(
-            this->SelfId(),
-            partitionId,
-            TTableId(TablePathId, Schema->Version),
-            BuilderContext);
+        const auto tableId = TTableId(TablePathId, Schema->Version);
+        return new TTablePartitionWriter<TChangeRecord>(this->SelfId(), partitionId, tableId, SerializationContext);
     }
 
     void Handle(TEvWorker::TEvData::TPtr& ev) {
@@ -450,7 +453,7 @@ class TLocalTableWriter
 
         for (auto& record : ev->Get()->Records) {
             records.emplace_back(record.Offset, TablePathId, record.Data.size());
-            auto res = PendingRecords.emplace(record.Offset, typename TChangeRecord::TBuilder()
+            auto res = PendingRecords.emplace(record.Offset, TBuilder<TChangeRecord>()
                 .WithSourceId(ev->Get()->Source)
                 .WithOrder(record.Offset)
                 .WithBody(std::move(record.Data))
@@ -531,7 +534,7 @@ class TLocalTableWriter
         : TBase(&TThis::StateWork)
         , TBaseSender(this, this, this, this, TActorId())
         , TablePathId(tablePathId)
-        , BuilderContext(std::forward<TArgs>(args)...)
+        , SerializationContext(std::forward<TArgs>(args)...)
     {
     }
 
@@ -557,7 +560,7 @@ class TLocalTableWriter
 private:
     mutable TMaybe<TString> LogPrefix;
     const TPathId TablePathId;
-    TChangeRecordBuilderContextTrait<TChangeRecord> BuilderContext;
+    TSerializationContext<TChangeRecord> SerializationContext;
 
     TActorId Worker;
     ui64 TableVersion = 0;