25-1: Fix for a bug with backward compatibility in the export of changefeeds (#15544)

Author: stanislav-shchetinin

ydb/core/protos/feature_flags.proto

@@ -196,4 +196,5 @@ message TFeatureFlags {
     optional bool EnableChangefeedsImport = 170 [default = false];
     optional bool EnablePermissionsExport = 171 [default = false];
     optional bool EnableDataErasure = 172 [default = false];
+    optional bool EnableChangefeedsExport = 174 [default = false];
 }

ydb/core/tx/datashard/export_s3_uploader.cpp

@@ -918,29 +918,31 @@ IActor* TS3Export::CreateUploader(const TActorId& dataShard, ui64 txId) const {
         ? GenYdbScheme(Columns, Task.GetTable())
         : Nothing();
 
-    const auto& persQueues = Task.GetChangefeedUnderlyingTopics();
-    const auto& cdcStreams = Task.GetTable().GetTable().GetCdcStreams();
-    Y_ASSERT(persQueues.size() == cdcStreams.size());
-
-    const int changefeedsCount = cdcStreams.size();
     TVector <TChangefeedExportDescriptions> changefeeds;
-    changefeeds.reserve(changefeedsCount);
-
-    for (int i = 0; i < changefeedsCount; ++i) {
-        Ydb::Table::ChangefeedDescription changefeed;
-        const auto& cdcStream = cdcStreams.at(i);
-        FillChangefeedDescription(changefeed, cdcStream);
-
-        Ydb::Topic::DescribeTopicResult topic;
-        const auto& pq = persQueues.at(i);
-        Ydb::StatusIds::StatusCode status;
-        TString error;
-        FillTopicDescription(topic, pq.GetPersQueueGroup(), pq.GetSelf(), cdcStream.GetName(), status, error);
-        // Unnecessary fields
-        topic.clear_self();
-        topic.clear_topic_stats();
-        
-        changefeeds.emplace_back(changefeed, topic);
+    if (AppData()->FeatureFlags.GetEnableChangefeedsExport()) {
+        const auto& persQueues = Task.GetChangefeedUnderlyingTopics();
+        const auto& cdcStreams = Task.GetTable().GetTable().GetCdcStreams();
+        Y_ASSERT(persQueues.size() == cdcStreams.size());
+
+        const int changefeedsCount = cdcStreams.size();
+        changefeeds.reserve(changefeedsCount);
+
+        for (int i = 0; i < changefeedsCount; ++i) {
+            Ydb::Table::ChangefeedDescription changefeed;
+            const auto& cdcStream = cdcStreams.at(i);
+            FillChangefeedDescription(changefeed, cdcStream);
+
+            Ydb::Topic::DescribeTopicResult topic;
+            const auto& pq = persQueues.at(i);
+            Ydb::StatusIds::StatusCode status;
+            TString error;
+            FillTopicDescription(topic, pq.GetPersQueueGroup(), pq.GetSelf(), cdcStream.GetName(), status, error);
+            // Unnecessary fields
+            topic.clear_self();
+            topic.clear_topic_stats();
+            
+            changefeeds.emplace_back(changefeed, topic);
+        }
     }
 
     auto permissions = (Task.GetEnablePermissions() && Task.GetShardNum() == 0)

ydb/core/tx/schemeshard/ut_export/ut_export.cpp

@@ -2632,6 +2632,8 @@ attributes {
         )", port);
 
         TTestEnv env(runtime, TTestEnvOptions().EnableChecksumsExport(true));
+        runtime.GetAppData().FeatureFlags.SetEnableChangefeedsExport(true);
+        
         Run(runtime, env, TVector<TString>{
             R"(
                 Name: "Table"

ydb/services/ydb/backup_ut/ydb_backup_ut.cpp

@@ -2261,6 +2261,7 @@ Y_UNIT_TEST_SUITE(BackupRestoreS3) {
         NTopic::TTopicClient topicClient(testEnv.GetDriver());
 
         constexpr const char* table = "/Root/table";
+        testEnv.GetServer().GetRuntime()->GetAppData().FeatureFlags.SetEnableChangefeedsExport(true);
         testEnv.GetServer().GetRuntime()->GetAppData().FeatureFlags.SetEnableChangefeedsImport(true);
 
         TestChangefeedAndTopicDescriptionsIsPreserved(