Merge 1ba36a7 into 818a8a0

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

@@ -439,4 +439,112 @@ Y_UNIT_TEST_SUITE(TExportToS3WithRebootsTests) {
             }
         )");
     }
+
+    Y_UNIT_TEST(ShouldSucceedOnSingleShardTableWithChangefeed) {
+        const char* tableName = "Table";
+        RunS3({
+            TTypedScheme {
+                Sprintf(R"(
+                    Name: "%s"
+                    Columns { Name: "key" Type: "Utf8" }
+                    Columns { Name: "value" Type: "Utf8" }
+                    KeyColumnNames: ["key"]
+                )", tableName),
+                EPathTypeTable
+            },
+            TTypedScheme {
+                Sprintf(R"(
+                    TableName: "%s"
+                    StreamDescription {
+                        Name: "update_feed"
+                        Mode: ECdcStreamModeUpdate
+                        Format: ECdcStreamFormatJson
+                        State: ECdcStreamStateReady
+                    }
+                )", tableName),
+                EPathTypeCdcStream
+            }
+        }, R"(
+            ExportToS3Settings {
+              endpoint: "localhost:%d"
+              scheme: HTTP
+              items {
+                source_path: "/MyRoot/Table"
+                destination_prefix: ""
+              }
+            }
+        )");
+    }
+
+    Y_UNIT_TEST(CancelOnSingleShardTableWithChangefeed) {
+        const char* tableName = "Table";
+        CancelS3({
+            TTypedScheme {
+                Sprintf(R"(
+                    Name: "%s"
+                    Columns { Name: "key" Type: "Utf8" }
+                    Columns { Name: "value" Type: "Utf8" }
+                    KeyColumnNames: ["key"]
+                )", tableName),
+                EPathTypeTable
+            },
+            TTypedScheme {
+                Sprintf(R"(
+                    TableName: "%s"
+                    StreamDescription {
+                        Name: "update_feed"
+                        Mode: ECdcStreamModeUpdate
+                        Format: ECdcStreamFormatJson
+                        State: ECdcStreamStateReady
+                    }
+                )", tableName),
+                EPathTypeCdcStream
+            }
+        }, R"(
+            ExportToS3Settings {
+              endpoint: "localhost:%d"
+              scheme: HTTP
+              items {
+                source_path: "/MyRoot/Table"
+                destination_prefix: ""
+              }
+            }
+        )");
+    }
+
+    Y_UNIT_TEST(ForgetShouldSucceedOnSingleShardTableWithChangefeed) {
+        const char* tableName = "Table";
+        ForgetS3({
+            TTypedScheme {
+                Sprintf(R"(
+                    Name: "%s"
+                    Columns { Name: "key" Type: "Utf8" }
+                    Columns { Name: "value" Type: "Utf8" }
+                    KeyColumnNames: ["key"]
+                )", tableName),
+                EPathTypeTable
+            },
+            TTypedScheme {
+                Sprintf(R"(
+                    TableName: "%s"
+                    StreamDescription {
+                        Name: "update_feed"
+                        Mode: ECdcStreamModeUpdate
+                        Format: ECdcStreamFormatJson
+                        State: ECdcStreamStateReady
+                    }
+                )", tableName),
+                EPathTypeCdcStream
+            }
+        }, R"(
+            ExportToS3Settings {
+              endpoint: "localhost:%d"
+              scheme: HTTP
+              items {
+                source_path: "/MyRoot/Table"
+                destination_prefix: ""
+              }
+            }
+        )");
+    }
 }

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

@@ -11,16 +11,48 @@ using namespace NKikimrSchemeOp;
 namespace NSchemeShardUT_Private {
 namespace NExportReboots {
 
+// using TCreateHandler = std::function<ui64(TTestActorRuntime&, ui64&, const TString&, const TString&)>;
+
+// template <typename T>
+// TCreateHandler GenCreateHandler(T func) {
+//     return [func](TTestActorRuntime& runtime, ui64& txId, const TString& parentPath, const TString&  scheme) {
+//         return func(runtime, txId, parentPath, scheme);
+//     };
+// }
+
+// THashMap<NKikimrSchemeOp::EPathType, TCreateHandler> CreateHandlers = {
+//     {EPathTypeTable, GenCreateHandler(TestCreateTable)},
+//     {EPathTypeView, GenCreateHandler(TestCreateView)},
+//     {EPathTypeCdcStream, GenCreateHandler(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& [scheme, type] : schemeObjects) {
+        // TestCreate(runtime, ++t.TxId, scheme, type);
         switch (type) {
             case EPathTypeTable:
                 TestCreateTable(runtime, ++t.TxId, "/MyRoot", scheme);
                 break;
             case EPathTypeView:
                 TestCreateView(runtime, ++t.TxId, "/MyRoot", scheme);
                 break;
+            case EPathTypeCdcStream:
+                TestCreateCdcStream(runtime, ++t.TxId, "/MyRoot", scheme);
+                break;
             default:
                 UNIT_FAIL("export is not implemented for the scheme object type: " << type);
                 return;