Fixed uninitializing value batch_size_bytes of the transfer (#17065)

Author: nshestakov

ydb/core/kqp/host/kqp_gateway_proxy.cpp

@@ -2614,7 +2614,7 @@ class TKqpGatewayProxy : public IKikimrGateway {
                 target.SetDstPath(AdjustPath(dst, GetDatabase()));
                 target.SetTransformLambda(lambda);
                 if (settings.Settings.Batching && settings.Settings.Batching->BatchSizeBytes) {
-                    config.MutableTransferSpecific()->MutableBatching()->SetBatchSizeBytes(settings.Settings.Batching->BatchSizeBytes);
+                    config.MutableTransferSpecific()->MutableBatching()->SetBatchSizeBytes(settings.Settings.Batching->BatchSizeBytes.value());
                 }
                 if (settings.Settings.Batching && settings.Settings.Batching->FlushInterval) {
                     config.MutableTransferSpecific()->MutableBatching()->SetFlushIntervalMilliSeconds(settings.Settings.Batching->FlushInterval.MilliSeconds());
@@ -2672,7 +2672,7 @@ class TKqpGatewayProxy : public IKikimrGateway {
                     op.MutableAlterTransfer()->SetFlushIntervalMilliSeconds(batching->FlushInterval.MilliSeconds());
                 }
                 if (batching->BatchSizeBytes) {
-                    op.MutableAlterTransfer()->SetBatchSizeBytes(batching->BatchSizeBytes);
+                    op.MutableAlterTransfer()->SetBatchSizeBytes(batching->BatchSizeBytes.value());
                 }
             }
 

ydb/core/kqp/provider/yql_kikimr_gateway.h

@@ -916,7 +916,7 @@ struct TTransferSettings : public TReplicationSettingsBase {
 
     struct TBatching {
         TDuration FlushInterval;
-        ui64 BatchSizeBytes;
+        std::optional<ui64> BatchSizeBytes;
     };
 
     TMaybe<TString> ConsumerName;

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

@@ -8174,6 +8174,21 @@ Y_UNIT_TEST_SUITE(KqpScheme) {
             const 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 TRANSFER `/Root/transfer_fi`
+                  FROM `/Root/topic` TO `/Root/table`
+                WITH (
+                    CONNECTION_STRING = "%s",
+                    FLUSH_INTERVAL = Interval('PT1S')
+                );
+            )", kikimr.GetEndpoint().c_str());
+
+            const auto result = session.ExecuteSchemeQuery(query).GetValueSync();
+            UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
+        }
     }
 
     Y_UNIT_TEST(CreateTransfer_QueryService) {
@@ -8417,6 +8432,21 @@ Y_UNIT_TEST_SUITE(KqpScheme) {
             const auto result = session.ExecuteQuery(query, NYdb::NQuery::TTxControl::NoTx()).ExtractValueSync();
             UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
         }
+
+        {
+            auto query = Sprintf(R"(
+                --!syntax_v1
+                CREATE TRANSFER `/Root/transfer_fi`
+                  FROM `/Root/topic` TO `/Root/table`
+                WITH (
+                    CONNECTION_STRING = "%s",
+                    FLUSH_INTERVAL = Interval('PT1S')
+                );
+            )", kikimr.GetEndpoint().c_str());
+
+            const auto result = session.ExecuteQuery(query, NYdb::NQuery::TTxControl::NoTx()).ExtractValueSync();
+            UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
+        }
     }
 
     Y_UNIT_TEST(AlterTransfer) {