Added user message for incorrect connection_string (without database) (#26452)

Author: nshestakov

ydb/core/kqp/host/kqp_gateway_proxy.cpp

@@ -2590,6 +2590,10 @@ class TKqpGatewayProxy : public IKikimrGateway {
                 consistency->Serialize(*config.MutableConsistencySettings()->MutableGlobal());
             }
 
+            if (params.GetDatabase().empty()) {
+                return MakeFuture(ResultFromError<TGenericResult>("Database is not specified"));
+            }
+
             auto& targets = *config.MutableSpecific();
             for (const auto& [src, dst] : settings.Targets) {
                 auto& target = *targets.AddTargets();
@@ -2665,6 +2669,10 @@ class TKqpGatewayProxy : public IKikimrGateway {
                     params.SetEndpoint(TString{parseResult.Endpoint});
                     params.SetDatabase(TString{parseResult.Database});
                     params.SetEnableSsl(parseResult.EnableSsl);
+
+                    if (params.GetDatabase().empty()) {
+                        return MakeFuture(ResultFromError<TGenericResult>("Database is not specified"));
+                    }
                 }
                 if (const auto& endpoint = settings.Settings.Endpoint) {
                     params.SetEndpoint(*endpoint);
@@ -2797,6 +2805,10 @@ class TKqpGatewayProxy : public IKikimrGateway {
                 params.SetCaCert(*caCert);
             }
 
+            if (!params.GetEndpoint().empty() && params.GetDatabase().empty()) {
+                return MakeFuture(ResultFromError<TGenericResult>("Database is not specified"));
+            }
+
             {
                 const auto& [src, dst, lambda] = settings.Target;
                 auto& target = *config.MutableTransferSpecific()->MutableTarget();
@@ -2900,6 +2912,10 @@ class TKqpGatewayProxy : public IKikimrGateway {
                     params.SetEndpoint(TString{parseResult.Endpoint});
                     params.SetDatabase(TString{parseResult.Database});
                     params.SetEnableSsl(parseResult.EnableSsl);
+
+                    if (params.GetDatabase().empty()) {
+                        return MakeFuture(ResultFromError<TGenericResult>("Database is not specified"));
+                    }
                 }
                 if (const auto& endpoint = settings.Settings.Endpoint) {
                     params.SetEndpoint(*endpoint);

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

@@ -10272,7 +10272,7 @@ Y_UNIT_TEST_SUITE(KqpScheme) {
                 CREATE TRANSFER `/Root/transfer_fi`
                   FROM `/Root/topic` TO `/Root/table` USING ($x) -> { RETURN <| id:$x._offset |> }
                 WITH (
-                    CONNECTION_STRING = "%s",
+                    CONNECTION_STRING = "%s/?database=/Root",
                     FLUSH_INTERVAL = Interval('PT1S')
                 );
             )", kikimr.GetEndpoint().c_str());
@@ -10461,6 +10461,21 @@ Y_UNIT_TEST_SUITE(KqpScheme) {
             UNIT_ASSERT_STRING_CONTAINS(result.GetIssues().ToOneLineString(), "flush_interval must be Interval");
         }
 
+        {
+            auto query = R"(
+                --!syntax_v1
+                CREATE TRANSFER `/Root/transfer`
+                  FROM `/Root/topic` TO `/Root/table` USING ($x) -> { RETURN <| id:$x._offset |> }
+                WITH (
+                    CONNECTION_STRING = "grpc://localhost:2135?database=/Root"
+                );
+            )";
+
+            const auto result = session.ExecuteQuery(query, NYdb::NQuery::TTxControl::NoTx()).ExtractValueSync();
+            UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::GENERIC_ERROR, result.GetIssues().ToString());
+            UNIT_ASSERT_STRING_CONTAINS(result.GetIssues().ToOneLineString(), "Database is not specified");
+        }
+
         {
             auto query = R"(
                 --!syntax_v1
@@ -10560,7 +10575,7 @@ Y_UNIT_TEST_SUITE(KqpScheme) {
                 CREATE TRANSFER `/Root/transfer_fi`
                   FROM `/Root/topic` TO `/Root/table` USING ($x) -> { RETURN <| id:$x._offset |> }
                 WITH (
-                    CONNECTION_STRING = "%s",
+                    CONNECTION_STRING = "%s/?database=/Root",
                     FLUSH_INTERVAL = Interval('PT1S')
                 );
             )", kikimr.GetEndpoint().c_str());
@@ -10747,7 +10762,8 @@ Y_UNIT_TEST_SUITE(KqpScheme) {
                 --!syntax_v1
                 ALTER TRANSFER `/Root/transfer`
                 SET (
-                    ENDPOINT = "localhost:2136"
+                    ENDPOINT = "localhost:2136",
+                    DATABASE = "/Root"
                 );
             )";
 
@@ -11133,7 +11149,8 @@ Y_UNIT_TEST_SUITE(KqpScheme) {
                 --!syntax_v1
                 ALTER TRANSFER `/Root/transfer`
                 SET (
-                    ENDPOINT = "localhost:2136"
+                    ENDPOINT = "localhost:2136",
+                    DATABASE = "/Root"
                 );
             )";
 

ydb/core/transfer/ut/functional/transfer_ut.cpp

@@ -92,7 +92,7 @@ Y_UNIT_TEST_SUITE(Transfer)
                 CREATE TRANSFER %s
                 FROM %s TO %s USING $l
                 WITH (
-                    CONNECTION_STRING = "grp§c://localhost:2135"
+                    CONNECTION_STRING = "grp§c://localhost:2135/?database=/Root"
                 )
             )", testCase.TransferName.data(), testCase.TopicName.data(), testCase.TableName.data()));
 
@@ -130,7 +130,7 @@ Y_UNIT_TEST_SUITE(Transfer)
                 CREATE TRANSFER %s
                 FROM %s TO %s USING $l
                 WITH (
-                    CONNECTION_STRING = "grpc://domain-not-exists-localhost.com.moc:2135"
+                    CONNECTION_STRING = "grpc://domain-not-exists-localhost.com.moc:2135/?database=/Root"
                 )
             )", testCase.TransferName.data(), testCase.TopicName.data(), testCase.TableName.data()));
 
@@ -992,6 +992,46 @@ Y_UNIT_TEST_SUITE(Transfer)
         CreateTransferSourceNotExists(true);
     }
 
+    void CreateTransferSourceDirNotExists(bool localTopic)
+    {
+        MainTestCase testCase;
+        testCase.CreateTable(R"(
+                CREATE TABLE `%s` (
+                    Key Uint64 NOT NULL,
+                    Message Utf8 NOT NULL,
+                    PRIMARY KEY (Key)
+                )  WITH (
+                    STORE = ROW
+                );
+            )");
+
+        auto settings = MainTestCase::CreateTransferSettings::WithLocalTopic(localTopic);
+        settings.TopicName = "dir_not_exists/topic_name";
+        testCase.CreateTransfer(R"(
+                $l = ($x) -> {
+                    return [
+                        <|
+                            Key:CAST($x._offset AS Uint64),
+                            Message:CAST($x._data AS Utf8)
+                        |>
+                    ];
+                };
+            )", settings);
+
+        testCase.CheckTransferStateError("Discovery error:");
+
+        testCase.DropTransfer();
+        testCase.DropTable();
+    }
+
+    Y_UNIT_TEST(CreateTransferSourceDirNotExists) {
+        CreateTransferSourceDirNotExists(false);
+    }
+
+    Y_UNIT_TEST(CreateTransferSourceDirNotExists_LocalTopic) {
+        CreateTransferSourceDirNotExists(true);
+    }
+
     void TransferSourceDropped(bool localTopic)
     {
         MainTestCase testCase;

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

@@ -159,6 +159,8 @@ class TReplication::TImpl: public TLagProvider {
 
             if (endpoint.empty()) {
                 ydbProxy.Reset(CreateLocalYdbProxy(Database));
+            } else if (database.empty()) {
+                ErrorState("Database is not specified.");
             } else {
                 switch (params.GetCredentialsCase()) {
                 case NKikimrReplication::TConnectionParams::kStaticCredentials: