Common TChangeRecord in ydb/core/change_exchange

Author: CyberROFL

ydb/core/change_exchange/change_record.cpp

@@ -2,15 +2,13 @@
 
 #include <ydb/core/protos/change_exchange.pb.h>
 
-namespace NKikimr::NDataShard {
+namespace NKikimr::NChangeExchange {
 
 void TChangeRecord::Serialize(NKikimrChangeExchange::TChangeRecord& record) const {
     record.SetOrder(Order);
     record.SetGroup(Group);
     record.SetStep(Step);
     record.SetTxId(TxId);
-    record.SetPathOwnerId(PathId.OwnerId);
-    record.SetLocalPathId(PathId.LocalPathId);
 
     switch (Kind) {
         case EKind::AsyncIndex: {
@@ -70,15 +68,6 @@ TInstant TChangeRecord::GetApproximateCreationDateTime() const {
         : TInstant::MilliSeconds(GetStep());
 }
 
-bool TChangeRecord::IsBroadcast() const {
-    switch (Kind) {
-        case EKind::CdcHeartbeat:
-            return true;
-        default:
-            return false;
-    }
-}
-
 TString TChangeRecord::ToString() const {
     TString result;
     TStringOutput out(result);
@@ -92,14 +81,9 @@ void TChangeRecord::Out(IOutputStream& out) const {
         << " Group: " << Group
         << " Step: " << Step
         << " TxId: " << TxId
-        << " PathId: " << PathId
         << " Kind: " << Kind
         << " Source: " << Source
         << " Body: " << Body.size() << "b"
-        << " TableId: " << TableId
-        << " SchemaVersion: " << SchemaVersion
-        << " LockId: " << LockId
-        << " LockOffset: " << LockOffset
     << " }";
 }
 

ydb/core/change_exchange/change_record.h

@@ -1,8 +1,5 @@
 #pragma once
 
-#include "datashard_user_table.h"
-
-#include <ydb/core/scheme/scheme_pathid.h>
 #include <ydb/core/scheme/scheme_tablecell.h>
 
 #include <util/generic/maybe.h>
@@ -12,12 +9,12 @@ namespace NKikimrChangeExchange {
     class TChangeRecord;
 }
 
-namespace NKikimr::NDataShard {
+namespace NKikimr::NChangeExchange {
 
-class TChangeRecordBuilder;
+template <typename T, typename TDerived> class TChangeRecordBuilder;
 
 class TChangeRecord {
-    friend class TChangeRecordBuilder;
+    template <typename T, typename TDerived> friend class TChangeRecordBuilder;
 
 public:
     enum class ESource: ui8 {
@@ -36,50 +33,37 @@ class TChangeRecord {
     ui64 GetGroup() const { return Group; }
     ui64 GetStep() const { return Step; }
     ui64 GetTxId() const { return TxId; }
-    ui64 GetLockId() const { return LockId; }
-    ui64 GetLockOffset() const { return LockOffset; }
-    const TPathId& GetPathId() const { return PathId; }
     EKind GetKind() const { return Kind; }
     const TString& GetBody() const { return Body; }
     ESource GetSource() const { return Source; }
 
-    const TPathId& GetTableId() const { return TableId; }
-    ui64 GetSchemaVersion() const { return SchemaVersion; }
-    TUserTable::TCPtr GetSchema() const { return Schema; }
-
     void Serialize(NKikimrChangeExchange::TChangeRecord& record) const;
 
     TConstArrayRef<TCell> GetKey() const;
     i64 GetSeqNo() const;
-    TString GetPartitionKey() const;
     TInstant GetApproximateCreationDateTime() const;
-    bool IsBroadcast() const;
 
     TString ToString() const;
     void Out(IOutputStream& out) const;
 
-private:
+protected:
     ui64 Order = Max<ui64>();
     ui64 Group = 0;
     ui64 Step = 0;
     ui64 TxId = 0;
-    ui64 LockId = 0;
-    ui64 LockOffset = 0;
-    TPathId PathId;
     EKind Kind;
     TString Body;
     ESource Source = ESource::Unspecified;
 
-    ui64 SchemaVersion;
-    TPathId TableId;
-    TUserTable::TCPtr Schema;
-
     mutable TMaybe<TOwnedCellVec> Key;
     mutable TMaybe<TString> PartitionKey;
 
 }; // TChangeRecord
 
+template <typename T, typename TDerived>
 class TChangeRecordBuilder {
+protected:
+    using TSelf = TDerived;
     using EKind = TChangeRecord::EKind;
     using ESource = TChangeRecord::ESource;
 
@@ -88,87 +72,56 @@ class TChangeRecordBuilder {
         Record.Kind = kind;
     }
 
-    explicit TChangeRecordBuilder(TChangeRecord&& record)
-        : Record(std::move(record))
-    {
-    }
-
-    TChangeRecordBuilder& WithLockId(ui64 lockId) {
-        Record.LockId = lockId;
-        return *this;
-    }
-
-    TChangeRecordBuilder& WithLockOffset(ui64 lockOffset) {
-        Record.LockOffset = lockOffset;
-        return *this;
+    explicit TChangeRecordBuilder(T&& record) {
+        Record = std::move(record);
     }
 
-    TChangeRecordBuilder& WithOrder(ui64 order) {
+    TSelf& WithOrder(ui64 order) {
         Record.Order = order;
-        return *this;
+        return static_cast<TSelf&>(*this);
     }
 
-    TChangeRecordBuilder& WithGroup(ui64 group) {
+    TSelf& WithGroup(ui64 group) {
         Record.Group = group;
-        return *this;
+        return static_cast<TSelf&>(*this);
     }
 
-    TChangeRecordBuilder& WithStep(ui64 step) {
+    TSelf& WithStep(ui64 step) {
         Record.Step = step;
-        return *this;
+        return static_cast<TSelf&>(*this);
     }
 
-    TChangeRecordBuilder& WithTxId(ui64 txId) {
+    TSelf& WithTxId(ui64 txId) {
         Record.TxId = txId;
-        return *this;
-    }
-
-    TChangeRecordBuilder& WithPathId(const TPathId& pathId) {
-        Record.PathId = pathId;
-        return *this;
-    }
-
-    TChangeRecordBuilder& WithTableId(const TPathId& tableId) {
-        Record.TableId = tableId;
-        return *this;
-    }
-
-    TChangeRecordBuilder& WithSchemaVersion(ui64 version) {
-        Record.SchemaVersion = version;
-        return *this;
-    }
-
-    TChangeRecordBuilder& WithSchema(TUserTable::TCPtr schema) {
-        Record.Schema = schema;
-        return *this;
+        return static_cast<TSelf&>(*this);
     }
 
-    TChangeRecordBuilder& WithBody(const TString& body) {
+    TSelf& WithBody(const TString& body) {
         Record.Body = body;
-        return *this;
+        return static_cast<TSelf&>(*this);
     }
 
-    TChangeRecordBuilder& WithBody(TString&& body) {
+    TSelf& WithBody(TString&& body) {
         Record.Body = std::move(body);
-        return *this;
+        return static_cast<TSelf&>(*this);
     }
 
-    TChangeRecordBuilder& WithSource(ESource source) {
+    TSelf& WithSource(ESource source) {
         Record.Source = source;
-        return *this;
+        return static_cast<TSelf&>(*this);
     }
 
-    TChangeRecord&& Build() {
+    T&& Build() {
         return std::move(Record);
     }
 
-private:
-    TChangeRecord Record;
+protected:
+    T Record;
 
 }; // TChangeRecordBuilder
 
 }
 
-Y_DECLARE_OUT_SPEC(inline, NKikimr::NDataShard::TChangeRecord, out, value) {
+Y_DECLARE_OUT_SPEC(inline, NKikimr::NChangeExchange::TChangeRecord, out, value) {
     return value.Out(out);
 }

ydb/core/change_exchange/ya.make

@@ -0,0 +1,16 @@
+LIBRARY()
+
+SRCS(
+    change_record.cpp
+)
+
+GENERATE_ENUM_SERIALIZATION(change_record.h)
+
+PEERDIR(
+    ydb/core/protos
+    ydb/core/scheme
+)
+
+YQL_LAST_ABI_VERSION()
+
+END()

ydb/core/tx/datashard/change_record.cpp

@@ -5,69 +5,9 @@
 namespace NKikimr::NDataShard {
 
 void TChangeRecord::Serialize(NKikimrChangeExchange::TChangeRecord& record) const {
-    record.SetOrder(Order);
-    record.SetGroup(Group);
-    record.SetStep(Step);
-    record.SetTxId(TxId);
+    NChangeExchange::TChangeRecord::Serialize(record);
     record.SetPathOwnerId(PathId.OwnerId);
     record.SetLocalPathId(PathId.LocalPathId);
-
-    switch (Kind) {
-        case EKind::AsyncIndex: {
-            Y_ABORT_UNLESS(record.MutableAsyncIndex()->ParseFromArray(Body.data(), Body.size()));
-            break;
-        }
-        case EKind::CdcDataChange: {
-            Y_ABORT_UNLESS(record.MutableCdcDataChange()->ParseFromArray(Body.data(), Body.size()));
-            break;
-        }
-        case EKind::CdcHeartbeat: {
-            break;
-        }
-    }
-}
-
-static auto ParseBody(const TString& protoBody) {
-    NKikimrChangeExchange::TDataChange body;
-    Y_ABORT_UNLESS(body.ParseFromArray(protoBody.data(), protoBody.size()));
-    return body;
-}
-
-TConstArrayRef<TCell> TChangeRecord::GetKey() const {
-    if (Key) {
-        return *Key;
-    }
-
-    switch (Kind) {
-        case EKind::AsyncIndex:
-        case EKind::CdcDataChange: {
-            const auto parsed = ParseBody(Body);
-
-            TSerializedCellVec key;
-            Y_ABORT_UNLESS(TSerializedCellVec::TryParse(parsed.GetKey().GetData(), key));
-
-            Key.ConstructInPlace(key.GetCells());
-            break;
-        }
-
-        case EKind::CdcHeartbeat: {
-            Y_ABORT("Not supported");
-        }
-    }
-
-    Y_ABORT_UNLESS(Key);
-    return *Key;
-}
-
-i64 TChangeRecord::GetSeqNo() const {
-    Y_ABORT_UNLESS(Order <= Max<i64>());
-    return static_cast<i64>(Order);
-}
-
-TInstant TChangeRecord::GetApproximateCreationDateTime() const {
-    return GetGroup()
-        ? TInstant::MicroSeconds(GetGroup())
-        : TInstant::MilliSeconds(GetStep());
 }
 
 bool TChangeRecord::IsBroadcast() const {
@@ -79,13 +19,6 @@ bool TChangeRecord::IsBroadcast() const {
     }
 }
 
-TString TChangeRecord::ToString() const {
-    TString result;
-    TStringOutput out(result);
-    Out(out);
-    return result;
-}
-
 void TChangeRecord::Out(IOutputStream& out) const {
     out << "{"
         << " Order: " << Order