Set & check replication config (#2649)

5 files changed, 149 insertions, 3 deletions

diff --git a/ydb/core/protos/out/out.cpp b/ydb/core/protos/out/out.cpp
index a0a225e7e05..eee0c8d61b2 100644
--- a/ydb/core/protos/out/out.cpp
+++ b/ydb/core/protos/out/out.cpp
@@ -211,6 +211,14 @@ Y_DECLARE_OUT_SPEC(, NKikimrSchemeOp::ECdcStreamState, stream, value) {
     stream << NKikimrSchemeOp::ECdcStreamState_Name(value);
 }
 
+Y_DECLARE_OUT_SPEC(, NKikimrSchemeOp::TTableReplicationConfig::EReplicationMode, stream, value) {
+    stream << NKikimrSchemeOp::TTableReplicationConfig::EReplicationMode_Name(value);
+}
+
+Y_DECLARE_OUT_SPEC(, NKikimrSchemeOp::TTableReplicationConfig::EConsistency, stream, value) {
+    stream << NKikimrSchemeOp::TTableReplicationConfig::EConsistency_Name(value);
+}
+
 Y_DECLARE_OUT_SPEC(, NKikimrSubDomains::EServerlessComputeResourcesMode, stream, value) {
     stream << NKikimrSubDomains::EServerlessComputeResourcesMode_Name(value);
 }
diff --git a/ydb/core/tx/replication/controller/dst_creator.cpp b/ydb/core/tx/replication/controller/dst_creator.cpp
index 4aafc544c76..4e78067f37d 100644
--- a/ydb/core/tx/replication/controller/dst_creator.cpp
+++ b/ydb/core/tx/replication/controller/dst_creator.cpp
@@ -114,7 +114,12 @@ class TDstCreator: public TActorBootstrapped<TDstCreator> {
 
         // TODO: support indexed tables
         TxBody.SetOperationType(NKikimrSchemeOp::ESchemeOpCreateTable);
-        TxBody.MutableCreateTable()->SetName(ToString(ExtractBase(DstPath)));
+        auto& desc = *TxBody.MutableCreateTable();
+        desc.SetName(ToString(ExtractBase(DstPath)));
+        // TODO: support other modes
+        auto& replicationConfig = *desc.MutableReplicationConfig();
+        replicationConfig.SetMode(NKikimrSchemeOp::TTableReplicationConfig::REPLICATION_MODE_READ_ONLY);
+        replicationConfig.SetConsistency(NKikimrSchemeOp::TTableReplicationConfig::CONSISTENCY_WEAK);
 
         AllocateTxId();
     }
@@ -249,6 +254,30 @@ class TDstCreator: public TActorBootstrapped<TDstCreator> {
     }
 
     bool CheckTableScheme(const NKikimrSchemeOp::TTableDescription& got, TString& error) const {
+        if (!got.HasReplicationConfig()) {
+            error = "Empty replication config";
+            return false;
+        }
+
+        const auto& replicationConfig = got.GetReplicationConfig();
+
+        switch (replicationConfig.GetMode()) {
+        case NKikimrSchemeOp::TTableReplicationConfig::REPLICATION_MODE_READ_ONLY:
+            break;
+        default:
+            error = "Unsupported replication mode";
+            return false;
+        }
+
+        switch (replicationConfig.GetConsistency()) {
+        case NKikimrSchemeOp::TTableReplicationConfig::CONSISTENCY_WEAK:
+            break;
+        default:
+            error = TStringBuilder() << "Unsupported replication consistency"
+                << ": " << static_cast<int>(replicationConfig.GetConsistency());
+            return false;
+        }
+
         const auto& expected = TxBody.GetCreateTable();
 
         // check key
diff --git a/ydb/core/tx/replication/controller/dst_creator_ut.cpp b/ydb/core/tx/replication/controller/dst_creator_ut.cpp
index 244b8964914..fb96df8e17b 100644
--- a/ydb/core/tx/replication/controller/dst_creator_ut.cpp
+++ b/ydb/core/tx/replication/controller/dst_creator_ut.cpp
@@ -51,6 +51,10 @@ Y_UNIT_TEST_SUITE(DstCreator) {
 
             UNIT_ASSERT(FindIfPtr(tableDesc.Columns, pred));
         }
+
+        const auto& replCfg = replicatedDesc.GetReplicationConfig();
+        UNIT_ASSERT_VALUES_EQUAL(replCfg.GetMode(), NKikimrSchemeOp::TTableReplicationConfig::REPLICATION_MODE_READ_ONLY);
+        UNIT_ASSERT_VALUES_EQUAL(replCfg.GetConsistency(), NKikimrSchemeOp::TTableReplicationConfig::CONSISTENCY_WEAK);
     }
 
     Y_UNIT_TEST(NonExistentSrc) {
@@ -191,6 +195,57 @@ Y_UNIT_TEST_SUITE(DstCreator) {
             },
         });
     }
+
+    Y_UNIT_TEST(EmptyReplicationConfig) {
+        auto clearConfig = [](const TTestTableDescription& desc) {
+            auto copy = desc;
+            copy.ReplicationConfig.Clear();
+            return copy;
+        };
+
+        ExistingDst(NKikimrScheme::StatusSchemeError, "Empty replication config", clearConfig, TTestTableDescription{
+            .Name = "Table",
+            .KeyColumns = {"key"},
+            .Columns = {
+                {.Name = "key", .Type = "Uint32"},
+                {.Name = "value", .Type = "Utf8"},
+            },
+        });
+    }
+
+    Y_UNIT_TEST(UnsupportedReplicationMode) {
+        auto clearMode = [](const TTestTableDescription& desc) {
+            auto copy = desc;
+            copy.ReplicationConfig->Mode = TTestTableDescription::TReplicationConfig::MODE_NONE;
+            return copy;
+        };
+
+        ExistingDst(NKikimrScheme::StatusSchemeError, "Unsupported replication mode", clearMode, TTestTableDescription{
+            .Name = "Table",
+            .KeyColumns = {"key"},
+            .Columns = {
+                {.Name = "key", .Type = "Uint32"},
+                {.Name = "value", .Type = "Utf8"},
+            },
+        });
+    }
+
+    Y_UNIT_TEST(UnsupportedReplicationConsistency) {
+        auto changeConsistency = [](const TTestTableDescription& desc) {
+            auto copy = desc;
+            copy.ReplicationConfig->Consistency = TTestTableDescription::TReplicationConfig::CONSISTENCY_STRONG;
+            return copy;
+        };
+
+        ExistingDst(NKikimrScheme::StatusSchemeError, "Unsupported replication consistency", changeConsistency, TTestTableDescription{
+            .Name = "Table",
+            .KeyColumns = {"key"},
+            .Columns = {
+                {.Name = "key", .Type = "Uint32"},
+                {.Name = "value", .Type = "Utf8"},
+            },
+        });
+    }
 }
 
 }
diff --git a/ydb/core/tx/replication/ut_helpers/test_table.cpp b/ydb/core/tx/replication/ut_helpers/test_table.cpp
index 76718cad1b0..0afe60fd749 100644
--- a/ydb/core/tx/replication/ut_helpers/test_table.cpp
+++ b/ydb/core/tx/replication/ut_helpers/test_table.cpp
@@ -9,10 +9,39 @@ void TTestTableDescription::TColumn::SerializeTo(NKikimrSchemeOp::TColumnDescrip
     proto.SetType(Type);
 }
 
+void TTestTableDescription::TReplicationConfig::SerializeTo(NKikimrSchemeOp::TTableReplicationConfig& proto) const {
+    switch (Mode) {
+    case MODE_NONE:
+        proto.SetMode(NKikimrSchemeOp::TTableReplicationConfig::REPLICATION_MODE_NONE);
+        break;
+    case MODE_READ_ONLY:
+        proto.SetMode(NKikimrSchemeOp::TTableReplicationConfig::REPLICATION_MODE_READ_ONLY);
+        break;
+    default:
+        Y_ABORT("Unexpected mode");
+    }
+
+    switch (Consistency) {
+    case CONSISTENCY_STRONG:
+        proto.SetConsistency(NKikimrSchemeOp::TTableReplicationConfig::CONSISTENCY_STRONG);
+        break;
+    case CONSISTENCY_WEAK:
+        proto.SetConsistency(NKikimrSchemeOp::TTableReplicationConfig::CONSISTENCY_WEAK);
+        break;
+    default:
+        Y_ABORT("Unexpected consistency");
+    }
+}
+
+TTestTableDescription::TReplicationConfig TTestTableDescription::TReplicationConfig::Default() {
+    return TReplicationConfig{
+        .Mode = MODE_READ_ONLY,
+        .Consistency = CONSISTENCY_WEAK,
+    };
+}
+
 void TTestTableDescription::SerializeTo(NKikimrSchemeOp::TTableDescription& proto) const {
     proto.SetName(Name);
-    proto.MutableReplicationConfig()->SetMode(NKikimrSchemeOp::TTableReplicationConfig::REPLICATION_MODE_READ_ONLY);
-    proto.MutableReplicationConfig()->SetConsistency(NKikimrSchemeOp::TTableReplicationConfig::CONSISTENCY_WEAK);
 
     for (const auto& keyColumn : KeyColumns) {
         proto.AddKeyColumnNames(keyColumn);
@@ -21,6 +50,10 @@ void TTestTableDescription::SerializeTo(NKikimrSchemeOp::TTableDescription& prot
     for (const auto& column : Columns) {
         column.SerializeTo(*proto.AddColumns());
     }
+
+    if (ReplicationConfig) {
+        ReplicationConfig->SerializeTo(*proto.MutableReplicationConfig());
+    }
 }
 
 THolder<NKikimrSchemeOp::TTableDescription> MakeTableDescription(const TTestTableDescription& desc) {
diff --git a/ydb/core/tx/replication/ut_helpers/test_table.h b/ydb/core/tx/replication/ut_helpers/test_table.h
index 0314bc1a2be..06122b4e489 100644
--- a/ydb/core/tx/replication/ut_helpers/test_table.h
+++ b/ydb/core/tx/replication/ut_helpers/test_table.h
@@ -1,5 +1,6 @@
 #pragma once
 
+#include <util/generic/maybe.h>
 #include <util/generic/ptr.h>
 #include <util/generic/string.h>
 #include <util/generic/vector.h>
@@ -7,6 +8,7 @@
 namespace NKikimrSchemeOp {
     class TColumnDescription;
     class TTableDescription;
+    class TTableReplicationConfig;
 }
 
 namespace NKikimr::NReplication::NTestHelpers {
@@ -19,9 +21,28 @@ struct TTestTableDescription {
         void SerializeTo(NKikimrSchemeOp::TColumnDescription& proto) const;
     };
 
+    struct TReplicationConfig {
+        enum EMode {
+            MODE_NONE = 0,
+            MODE_READ_ONLY = 1,
+        };
+
+        enum EConsistency {
+            CONSISTENCY_STRONG = 1,
+            CONSISTENCY_WEAK = 2,
+        };
+
+        EMode Mode;
+        EConsistency Consistency;
+
+        void SerializeTo(NKikimrSchemeOp::TTableReplicationConfig& proto) const;
+        static TReplicationConfig Default();
+    };
+
     TString Name;
     TVector<TString> KeyColumns;
     TVector<TColumn> Columns;
+    TMaybe<TReplicationConfig> ReplicationConfig = TReplicationConfig::Default();
 
     void SerializeTo(NKikimrSchemeOp::TTableDescription& proto) const;
 };