Merge e8d8896 into a4b20bc

Author: stanislav-shchetinin

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/ut_export_reboots_s3/ut_export_reboots_s3.cpp

@@ -512,4 +512,83 @@ 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,37 @@ using namespace NKikimrSchemeOp;
 namespace NSchemeShardUT_Private {
 namespace NExportReboots {
 
+using TCreateHandler = std::function<void(TTestActorRuntime&, ui64&, const TString&, const TString&)>;
+
+#define GEN_CREATE_HANDLER(func) \
+    TCreateHandler([](TTestActorRuntime& runtime, ui64& txId, const TString& parentPath, const TString& scheme) { \
+        func(runtime, txId, parentPath, scheme); \
+    })
+
+THashMap<NKikimrSchemeOp::EPathType, TCreateHandler> CreateHandlers = {
+    {EPathTypeTable, GEN_CREATE_HANDLER(TestCreateTable)},
+    {EPathTypeView, GEN_CREATE_HANDLER(TestCreateView)},
+    {EPathTypeCdcStream, GEN_CREATE_HANDLER(TestCreateCdcStream)},
+};
+
+void TestCreate(
+    TTestActorRuntime& runtime,
+    ui64& txId,
+    const TString& scheme,
+    NKikimrSchemeOp::EPathType pathType
+) {
+    if (CreateHandlers.contains(pathType)) {
+        CreateHandlers[pathType](runtime, txId, "/MyRoot", scheme);
+    } 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);