Added ability to add a transfer and a async replication to subdirecto… (#22503)

Author: nshestakov

ydb/core/transfer/ut/common/utils.h

@@ -426,6 +426,10 @@ struct MainTestCase {
     };
 
     void CreateTransfer(const std::string& lambda, const CreateTransferSettings& settings = CreateTransferSettings()) {
+        CreateTransfer(TransferName, lambda, settings);
+    }
+
+    void CreateTransfer(const std::string& name, const std::string& lambda, const CreateTransferSettings& settings = CreateTransferSettings()) {
         std::vector<std::string> options;
         if (!settings.LocalTopic) {
             options.push_back(TStringBuilder() << "CONNECTION_STRING = 'grpc://" << ConnectionString << "'");
@@ -457,7 +461,7 @@ struct MainTestCase {
             WITH (
                 %s
             );
-        )", lambda.data(), TransferName.data(), topicName.data(), TableName.data(), optionsStr.data());
+        )", lambda.data(), name.data(), topicName.data(), TableName.data(), optionsStr.data());
 
         ExecuteDDL(ddl, true, settings.ExpectedError);
     }
@@ -497,6 +501,10 @@ struct MainTestCase {
     }
 
     void AlterTransfer(const AlterTransferSettings& settings, bool success = true) {
+        AlterTransfer(TransferName, settings, success);
+    }
+
+    void AlterTransfer(const std::string& name, const AlterTransferSettings& settings, bool success = true) {
         std::string lambda = settings.TransformLambda ? *settings.TransformLambda : "";
         std::string setLambda = settings.TransformLambda ? "SET USING $l" : "";
 
@@ -523,7 +531,7 @@ struct MainTestCase {
             ALTER TRANSFER `%s`
             %s
             %s;
-        )", lambda.data(), TransferName.data(), setLambda.data(), setOptions.data()), success);
+        )", lambda.data(), name.data(), setLambda.data(), setOptions.data()), success);
     }
 
     void DropTransfer() {

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

@@ -1311,5 +1311,36 @@ Y_UNIT_TEST_SUITE(Transfer)
         testCase.DropTopic();
         testCase.DropTable();
     }
+
+    Y_UNIT_TEST(CreateAndAlterTransferInDirectory)
+    {
+        MainTestCase testCase;
+        testCase.CreateTable(R"(
+                CREATE TABLE `%s` (
+                    Key Uint64 NOT NULL,
+                    Message Utf8,
+                    PRIMARY KEY (Key)
+                )  WITH (
+                    STORE = %s
+                );
+            )");
+        testCase.CreateTopic(1);
+
+        testCase.CreateDirectory("/local/subdir");
+        testCase.CreateTransfer(TStringBuilder() << "subdir/" << testCase.TransferName, Sprintf(R"(
+                $l = ($x) -> {
+                    return [
+                        <|
+                            Key:CAST($x._offset AS Uint64),
+                            Message:CAST($x._data AS Utf8)
+                        |>
+                    ];
+                };
+            )", testCase.TableName.data()));
+
+        testCase.AlterTransfer(TStringBuilder() << "subdir/" << testCase.TransferName,
+            MainTestCase::AlterTransferSettings::WithBatching(TDuration::Seconds(1), 1));
+    }
+
 }
 

ydb/core/tx/schemeshard/schemeshard__operation_create_replication.cpp

@@ -382,7 +382,7 @@ class TCreateReplication: public TSubOperation {
             return result;
         }
 
-        auto path = parentPath.Child(name);
+        auto path = parentPath.Child(name, TPath::TSplitChildTag{});
         {
             const auto checks = path.Check();
             checks