Tests with reboots when exporting changefeeds to S3

ydb/core/protos/flat_scheme_op.proto

@@ -2102,3 +2102,7 @@ message TImportTableChangefeeds {
     }
     repeated TImportChangefeedTopic Changefeeds = 1;
 }
+
+message TChangefeedUnderlyingTopics {
+    repeated TPathDescription ChangefeedUnderlyingTopics = 1;
+}

ydb/core/tx/datashard/export_s3_uploader.cpp

@@ -241,6 +241,7 @@ class TS3Uploader: public TActorBootstrapped<TS3Uploader> {
     }
 
     void UploadChangefeed() {
+        Y_ABORT_UNLESS(!ChangefeedsUploaded);
         if (IndexExportedChangefeed == Changefeeds.size()) {
             ChangefeedsUploaded = true;
             if (Scanner) {

ydb/core/tx/schemeshard/schemeshard__init.cpp

@@ -805,7 +805,7 @@ struct TSchemeShard::TTxInit : public TTransactionBase<TSchemeShard> {
         return true;
     }
 
-    typedef std::tuple<TPathId, TString, TString, TString, TString, bool, TString, ui32, bool, bool> TBackupSettingsRec;
+    typedef std::tuple<TPathId, TString, TString, TString, TString, bool, TString, ui32, bool, bool, TString> TBackupSettingsRec;
     typedef TDeque<TBackupSettingsRec> TBackupSettingsRows;
 
     template <typename SchemaTable, typename TRowSet>
@@ -819,7 +819,8 @@ struct TSchemeShard::TTxInit : public TTransactionBase<TSchemeShard> {
             rowSet.template GetValueOrDefault<typename SchemaTable::TableDescription>(""),
             rowSet.template GetValueOrDefault<typename SchemaTable::NumberOfRetries>(0),
             rowSet.template GetValueOrDefault<typename SchemaTable::EnableChecksums>(false),
-            rowSet.template GetValueOrDefault<typename SchemaTable::EnablePermissions>(false)
+            rowSet.template GetValueOrDefault<typename SchemaTable::EnablePermissions>(false),
+            rowSet.template GetValueOrDefault<typename SchemaTable::ChangefeedUnderlyingTopics>("")
         );
     }
 
@@ -3804,6 +3805,7 @@ struct TSchemeShard::TTxInit : public TTransactionBase<TSchemeShard> {
                 ui32 nRetries = std::get<7>(rec);
                 bool enableChecksums = std::get<8>(rec);
                 bool enablePermissions = std::get<9>(rec);
+                TString changefeedUnderlyingTopics = std::get<10>(rec);
 
                 Y_ABORT_UNLESS(tableName.size() > 0);
 
@@ -3835,6 +3837,14 @@ struct TSchemeShard::TTxInit : public TTransactionBase<TSchemeShard> {
                     auto desc = tableInfo->BackupSettings.MutableTable();
                     Y_ABORT_UNLESS(ParseFromStringNoSizeLimit(*desc, tableDesc));
                 }
+
+                if (changefeedUnderlyingTopics) {
+                    NKikimrSchemeOp::TChangefeedUnderlyingTopics wrapperOverTopics;
+                    Y_ABORT_UNLESS(ParseFromStringNoSizeLimit(wrapperOverTopics, changefeedUnderlyingTopics));
+                    for (const auto& topic : wrapperOverTopics.GetChangefeedUnderlyingTopics()) {
+                        *tableInfo->BackupSettings.AddChangefeedUnderlyingTopics() = topic;
+                    }
+                }
 
                 LOG_DEBUG_S(ctx, NKikimrServices::FLAT_TX_SCHEMESHARD, "Loaded backup settings"
                                 << ", pathId: " << pathId

ydb/core/tx/schemeshard/schemeshard_impl.cpp

@@ -3362,6 +3362,16 @@ void TSchemeShard::PersistTxShardStatus(NIceDb::TNiceDb& db, TOperationId opId,
         );
 }
 
+NKikimrSchemeOp::TChangefeedUnderlyingTopics ConvertChangefeedUnderlyingTopics(
+    const google::protobuf::RepeatedPtrField<NKikimrSchemeOp::TPathDescription>& changefeedUnderlyingTopics
+) {
+    NKikimrSchemeOp::TChangefeedUnderlyingTopics result;
+    for (const auto& x : changefeedUnderlyingTopics) {
+        *result.AddChangefeedUnderlyingTopics() = x;
+    }
+    return result;
+}
+
 void TSchemeShard::PersistBackupSettings(
         NIceDb::TNiceDb& db,
         TPathId pathId,
@@ -3376,6 +3386,7 @@ void TSchemeShard::PersistBackupSettings(
                 NIceDb::TUpdate<Schema::BackupSettings::ScanSettings>(settings.GetScanSettings().SerializeAsString()), \
                 NIceDb::TUpdate<Schema::BackupSettings::NeedToBill>(settings.GetNeedToBill()), \
                 NIceDb::TUpdate<Schema::BackupSettings::TableDescription>(settings.GetTable().SerializeAsString()), \
+                NIceDb::TUpdate<Schema::BackupSettings::ChangefeedUnderlyingTopics>(ConvertChangefeedUnderlyingTopics(settings.GetChangefeedUnderlyingTopics()).SerializeAsString()), \
                 NIceDb::TUpdate<Schema::BackupSettings::NumberOfRetries>(settings.GetNumberOfRetries()), \
                 NIceDb::TUpdate<Schema::BackupSettings::EnableChecksums>(settings.GetEnableChecksums()), \
                 NIceDb::TUpdate<Schema::BackupSettings::EnablePermissions>(settings.GetEnablePermissions())); \
@@ -3386,6 +3397,7 @@ void TSchemeShard::PersistBackupSettings(
                 NIceDb::TUpdate<Schema::MigratedBackupSettings::ScanSettings>(settings.GetScanSettings().SerializeAsString()), \
                 NIceDb::TUpdate<Schema::MigratedBackupSettings::NeedToBill>(settings.GetNeedToBill()), \
                 NIceDb::TUpdate<Schema::MigratedBackupSettings::TableDescription>(settings.GetTable().SerializeAsString()), \
+                NIceDb::TUpdate<Schema::MigratedBackupSettings::ChangefeedUnderlyingTopics>(ConvertChangefeedUnderlyingTopics(settings.GetChangefeedUnderlyingTopics()).SerializeAsString()), \
                 NIceDb::TUpdate<Schema::MigratedBackupSettings::NumberOfRetries>(settings.GetNumberOfRetries()), \
                 NIceDb::TUpdate<Schema::MigratedBackupSettings::EnableChecksums>(settings.GetEnableChecksums()), \
                 NIceDb::TUpdate<Schema::MigratedBackupSettings::EnablePermissions>(settings.GetEnablePermissions())); \

ydb/core/tx/schemeshard/schemeshard_schema.h

@@ -583,6 +583,7 @@ struct Schema : NIceDb::Schema {
         struct YTSettings : Column<3, NScheme::NTypeIds::String> {};
         struct S3Settings : Column<6, NScheme::NTypeIds::String> {};
         struct TableDescription : Column<7, NScheme::NTypeIds::String> {};
+        struct ChangefeedUnderlyingTopics : Column<13, NScheme::NTypeIds::String> {};
         struct NumberOfRetries : Column<8, NScheme::NTypeIds::Uint32> {};
         struct ScanSettings : Column<9, NScheme::NTypeIds::String> {};
         struct NeedToBill : Column<10, NScheme::NTypeIds::Bool> {};
@@ -605,7 +606,8 @@ struct Schema : NIceDb::Schema {
             ScanSettings,
             NeedToBill,
             EnableChecksums,
-            EnablePermissions
+            EnablePermissions,
+            ChangefeedUnderlyingTopics
         >;
     };
 
@@ -617,6 +619,7 @@ struct Schema : NIceDb::Schema {
         struct YTSettings : Column<4, NScheme::NTypeIds::String> {};
         struct S3Settings : Column<7, NScheme::NTypeIds::String> {};
         struct TableDescription : Column<8, NScheme::NTypeIds::String> {};
+        struct ChangefeedUnderlyingTopics : Column<14, NScheme::NTypeIds::String> {};
         struct NumberOfRetries : Column<9, NScheme::NTypeIds::Uint32> {};
         struct ScanSettings : Column<10, NScheme::NTypeIds::String> {};
         struct NeedToBill : Column<11, NScheme::NTypeIds::Bool> {};
@@ -640,7 +643,8 @@ struct Schema : NIceDb::Schema {
             ScanSettings,
             NeedToBill,
             EnableChecksums,
-            EnablePermissions
+            EnablePermissions,
+            ChangefeedUnderlyingTopics
         >;
     };
 

ydb/core/tx/schemeshard/ut_export_reboots_s3/ut_export_reboots_s3.cpp

@@ -509,4 +509,82 @@ Y_UNIT_TEST_SUITE(TExportToS3WithRebootsTests) {
             }
         )");
     }
+
+    class TestData {
+    public:
+        static const TTypedScheme& Table() {
+            return TableScheme;
+        } 
+
+        static const TTypedScheme& Changefeed() {
+            return ChangefeedScheme;
+        }
+
+        static const TString& Request() {
+            return RequestString;
+        }
+
+    private:
+        static const char* TableName;
+        static const TTypedScheme TableScheme;
+        static const TTypedScheme ChangefeedScheme;
+        static const TString RequestString;
+    };
+
+    const char* TestData::TableName = "Table";
+
+    const TTypedScheme TestData::TableScheme = TTypedScheme {
+        EPathTypeTable,
+        Sprintf(R"(
+            Name: "%s"
+            Columns { Name: "key" Type: "Utf8" }
+            Columns { Name: "value" Type: "Utf8" }
+            KeyColumnNames: ["key"]
+        )", TableName)
+    };
+
+    const TTypedScheme TestData::ChangefeedScheme = TTypedScheme {
+        EPathTypeCdcStream,
+        Sprintf(R"(
+            TableName: "%s"
+            StreamDescription {
+                Name: "update_feed"
+                Mode: ECdcStreamModeUpdate
+                Format: ECdcStreamFormatJson
+                State: ECdcStreamStateReady
+            }
+        )", TableName)
+    };
+
+    const TString TestData::RequestString = R"(
+        ExportToS3Settings {
+            endpoint: "localhost:%d"
+            scheme: HTTP
+            items {
+                source_path: "/MyRoot/Table"
+                destination_prefix: ""
+            }
+        }
+    )";
+
+    Y_UNIT_TEST(ShouldSucceedOnSingleShardTableWithChangefeed) {
+        RunS3({
+            TestData::Table(),
+            TestData::Changefeed()
+        }, TestData::Request());
+    }
+
+    Y_UNIT_TEST(CancelOnSingleShardTableWithChangefeed) {
+        CancelS3({
+            TestData::Table(),
+            TestData::Changefeed()
+        }, TestData::Request());
+    }
+
+    Y_UNIT_TEST(ForgetShouldSucceedOnSingleShardTableWithChangefeed) {
+        ForgetS3({
+            TestData::Table(),
+            TestData::Changefeed()
+        }, TestData::Request());
+    }
 }

ydb/core/tx/schemeshard/ut_helpers/export_reboots_common.cpp

@@ -11,20 +11,28 @@ using namespace NKikimrSchemeOp;
 namespace NSchemeShardUT_Private {
 namespace NExportReboots {
 
+void TestCreate(TTestActorRuntime& runtime, ui64 txId, const TString& scheme, NKikimrSchemeOp::EPathType pathType) {
+    using TTestCreateFunc = ui64(*)(TTestActorRuntime&, ui64, const TString&, const TString&, 
+        const TVector<TExpectedResult>&, const TApplyIf&);
+
+    static const THashMap<NKikimrSchemeOp::EPathType, TTestCreateFunc> functions = {
+        {EPathTypeTable, &TestSimpleCreateTable},
+        {EPathTypeView, &TestCreateView},
+        {EPathTypeCdcStream, &TestCreateCdcStream},
+    };
+
+    auto it = functions.find(pathType);
+    if (it != functions.end()) {
+        it->second(runtime, txId, "/MyRoot", scheme, {NKikimrScheme::StatusAccepted}, {});
+    } else {
+        UNIT_FAIL("export is not implemented for the scheme object type: " << pathType);
+    }
+}
+
 void CreateSchemeObjects(TTestWithReboots& t, TTestActorRuntime& runtime, const TVector<TTypedScheme>& schemeObjects) {
     TSet<ui64> toWait;
     for (const auto& [type, scheme, _] : schemeObjects) {
-        switch (type) {
-            case EPathTypeTable:
-                TestCreateTable(runtime, ++t.TxId, "/MyRoot", scheme);
-                break;
-            case EPathTypeView:
-                TestCreateView(runtime, ++t.TxId, "/MyRoot", scheme);
-                break;
-            default:
-                UNIT_FAIL("export is not implemented for the scheme object type: " << type);
-                return;
-        }
+        TestCreate(runtime, ++t.TxId, scheme, type);
         toWait.insert(t.TxId);
     }
     t.TestEnv->TestWaitNotification(runtime, toWait);

ydb/core/tx/schemeshard/ut_helpers/helpers.cpp

@@ -821,6 +821,7 @@ namespace NSchemeShardUT_Private {
 
     // table
     GENERIC_WITH_ATTRS_HELPERS(CreateTable, NKikimrSchemeOp::EOperationType::ESchemeOpCreateTable, &NKikimrSchemeOp::TModifyScheme::MutableCreateTable)
+    GENERIC_HELPERS(SimpleCreateTable, NKikimrSchemeOp::EOperationType::ESchemeOpCreateTable, &NKikimrSchemeOp::TModifyScheme::MutableCreateTable)
     GENERIC_HELPERS(CreateIndexedTable, NKikimrSchemeOp::EOperationType::ESchemeOpCreateIndexedTable, &NKikimrSchemeOp::TModifyScheme::MutableCreateIndexedTable)
     GENERIC_HELPERS(ConsistentCopyTables, NKikimrSchemeOp::EOperationType::ESchemeOpCreateConsistentCopyTables, &NKikimrSchemeOp::TModifyScheme::MutableCreateConsistentCopyTables)
     GENERIC_HELPERS(AlterTable, NKikimrSchemeOp::EOperationType::ESchemeOpAlterTable, &NKikimrSchemeOp::TModifyScheme::MutableAlterTable)

ydb/core/tx/schemeshard/ut_helpers/helpers.h

@@ -188,6 +188,7 @@ namespace NSchemeShardUT_Private {
 
     // table
     GENERIC_WITH_ATTRS_HELPERS(CreateTable);
+    GENERIC_HELPERS(SimpleCreateTable);
     GENERIC_HELPERS(CreateIndexedTable);
     GENERIC_HELPERS(ConsistentCopyTables);
     GENERIC_HELPERS(AlterTable);

ydb/tests/functional/scheme_tests/canondata/tablet_scheme_tests.TestTabletSchemes.test_tablet_schemes_flat_schemeshard_/flat_schemeshard.schema

@@ -1179,6 +1179,11 @@
                 "ColumnId": 12,
                 "ColumnName": "EnablePermissions",
                 "ColumnType": "Bool"
+            },
+            {
+                "ColumnId": 13,
+                "ColumnName": "ChangefeedUnderlyingTopics",
+                "ColumnType": "String"
             }
         ],
         "ColumnsDropped": [],
@@ -1196,7 +1201,8 @@
                     9,
                     10,
                     11,
-                    12
+                    12,
+                    13
                 ],
                 "RoomID": 0,
                 "Codec": 0,
@@ -4577,6 +4583,11 @@
                 "ColumnId": 13,
                 "ColumnName": "EnablePermissions",
                 "ColumnType": "Bool"
+            },
+            {
+                "ColumnId": 14,
+                "ColumnName": "ChangefeedUnderlyingTopics",
+                "ColumnType": "String"
             }
         ],
         "ColumnsDropped": [],
@@ -4595,7 +4606,8 @@
                     10,
                     11,
                     12,
-                    13
+                    13,
+                    14
                 ],
                 "RoomID": 0,
                 "Codec": 0,