Merge 66a5ab1 into f2c77f7

Author: CyberROFL

ydb/core/base/path.h

@@ -37,4 +37,12 @@ inline TVector<TString> ChildPath(const TVector<TString>& parentPath, const TStr
     return path;
 }
 
+inline TVector<TString> ChildPath(const TVector<TString>& parentPath, const TVector<TString>& childPath) {
+    auto path = parentPath;
+    for (const auto& childName : childPath) {
+        path.push_back(childName);
+    }
+    return path;
+}
+
 }

ydb/core/grpc_services/rpc_replication.cpp

@@ -11,6 +11,8 @@
 
 #include <google/protobuf/util/time_util.h>
 
+#include <util/string/builder.h>
+
 namespace NKikimr::NGRpcService {
 
 using namespace Ydb;
@@ -138,9 +140,18 @@ class TDescribeReplicationRPC: public TRpcSchemeRequestActor<TDescribeReplicatio
         return ReplyWithResult(Ydb::StatusIds::SUCCESS, Result, ctx);
     }
 
+    static TString BuildConnectionString(const NKikimrReplication::TConnectionParams& params) {
+        return TStringBuilder()
+            << (params.GetEnableSsl() ? "grpcs://" : "grpc://")
+            << params.GetEndpoint()
+            << "/?database=" << params.GetDatabase();
+    }
+
     static void ConvertConnectionParams(const NKikimrReplication::TConnectionParams& from, Ydb::Replication::ConnectionParams& to) {
         to.set_endpoint(from.GetEndpoint());
         to.set_database(from.GetDatabase());
+        to.set_enable_ssl(from.GetEnableSsl());
+        to.set_connection_string(BuildConnectionString(from));
 
         switch (from.GetCredentialsCase()) {
         case NKikimrReplication::TConnectionParams::kStaticCredentials:

ydb/core/kqp/host/kqp_gateway_proxy.cpp

@@ -1650,6 +1650,7 @@ class TKqpGatewayProxy : public IKikimrGateway {
                 const auto parseResult = NYdb::ParseConnectionString(*connectionString);
                 params.SetEndpoint(parseResult.Endpoint);
                 params.SetDatabase(parseResult.Database);
+                params.SetEnableSsl(parseResult.EnableSsl);
             }
             if (const auto& endpoint = settings.Settings.Endpoint) {
                 params.SetEndpoint(*endpoint);

ydb/core/kqp/ut/scheme/kqp_scheme_ut.cpp

@@ -5605,6 +5605,61 @@ Y_UNIT_TEST_SUITE(KqpScheme) {
             UNIT_ASSERT_EQUAL_C(describe.GetStatus(), EStatus::SCHEME_ERROR, result.GetIssues().ToString());
         }
     }
+
+    void AsyncReplicationConnectionParams(TKikimrRunner& kikimr, const TString& connectionParam, bool ssl = false) {
+        using namespace NReplication;
+
+        auto repl = TReplicationClient(kikimr.GetDriver(), TCommonClientSettings().Database("/Root"));
+        auto db = kikimr.GetTableClient();
+        auto session = db.CreateSession().GetValueSync().GetSession();
+
+        {
+            auto query = R"(
+                --!syntax_v1
+                CREATE TABLE `/Root/table` (Key Uint64, Value String, PRIMARY KEY (Key));
+            )";
+
+            auto result = session.ExecuteSchemeQuery(query).GetValueSync();
+            UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
+        }
+        {
+            auto query = Sprintf(R"(
+                --!syntax_v1
+                CREATE ASYNC REPLICATION `/Root/replication` FOR
+                    `/Root/table` AS `/Root/replica`
+                WITH (
+                    %s, TOKEN = "root@builtin"
+                );
+            )", connectionParam.c_str());
+
+            const auto result = session.ExecuteSchemeQuery(query).GetValueSync();
+            UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
+        }
+        {
+            const auto result = repl.DescribeReplication("/Root/replication").ExtractValueSync();
+            UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
+
+            const auto& params = result.GetReplicationDescription().GetConnectionParams();
+            UNIT_ASSERT_VALUES_EQUAL(params.GetDiscoveryEndpoint(), kikimr.GetEndpoint());
+            UNIT_ASSERT_VALUES_EQUAL(params.GetDatabase(), "/Root");
+            UNIT_ASSERT_VALUES_EQUAL(params.GetEnableSsl(), ssl);
+        }
+    }
+
+    Y_UNIT_TEST(AsyncReplicationConnectionString) {
+        TKikimrRunner kikimr;
+        AsyncReplicationConnectionParams(kikimr, Sprintf(R"(CONNECTION_STRING = "grpc://%s/?database=/Root")", kikimr.GetEndpoint().c_str()));
+    }
+
+    Y_UNIT_TEST(AsyncReplicationConnectionStringWithSsl) {
+        TKikimrRunner kikimr;
+        AsyncReplicationConnectionParams(kikimr, Sprintf(R"(CONNECTION_STRING = "grpcs://%s/?database=/Root")", kikimr.GetEndpoint().c_str()), true);
+    }
+
+    Y_UNIT_TEST(AsyncReplicationEndpointAndDatabase) {
+        TKikimrRunner kikimr;
+        AsyncReplicationConnectionParams(kikimr, Sprintf(R"(ENDPOINT = "%s", DATABASE = "/Root")", kikimr.GetEndpoint().c_str()));
+    }
 }
 
 Y_UNIT_TEST_SUITE(KqpOlapScheme) {

ydb/core/protos/replication.proto

@@ -19,6 +19,7 @@ message TOAuthToken {
 message TConnectionParams {
     optional string Endpoint = 1;
     optional string Database = 2;
+    optional bool EnableSsl = 5;
     // credentials
     oneof Credentials {
         TStaticCredentials StaticCredentials = 3;

ydb/core/tx/replication/controller/dst_alterer.cpp

@@ -41,6 +41,7 @@ class TDstAlterer: public TActorBootstrapped<TDstAlterer> {
 
         switch (Kind) {
         case TReplication::ETargetKind::Table:
+        case TReplication::ETargetKind::IndexTable:
             tx.SetOperationType(NKikimrSchemeOp::ESchemeOpAlterTable);
             PathIdFromPathId(DstPathId, tx.MutableAlterTable()->MutablePathId());
             tx.MutableAlterTable()->MutableReplicationConfig()->SetMode(