CreateChangefeedsPropose

Author: stanislav-shchetinin

ydb/core/tx/schemeshard/schemeshard_import_flow_proposals.cpp

@@ -8,16 +8,20 @@
 namespace NKikimr {
 namespace NSchemeShard {
 
-void CreateChangefeedsPropose(THolder<TEvSchemeShard::TEvModifySchemeTransaction>& propose, const TImportInfo::TItem& item) {
+bool CreateChangefeedsPropose(THolder<TEvSchemeShard::TEvModifySchemeTransaction>& propose, const TImportInfo::TItem& item, TString& error) {
     auto& record = propose->Record;
     const auto& changefeeds = item.Changefeeds;
 
     for (const auto& changefeed : changefeeds) {
         auto& modifyScheme = *record.AddTransaction();
-        const auto& cdcStream = modifyScheme.MutableCreateCdcStream();
-        cdcStream->MutableTableName();
+        auto& cdcStream = *modifyScheme.MutableCreateCdcStream();
+        Ydb::StatusIds::StatusCode status;
+        auto& cdcStreamDescription = *cdcStream.MutableStreamDescription();
+        if (!FillChangefeedDescription(cdcStreamDescription, changefeed, status, error)) {
+            return false;
+        }
     }
-
+    return true;
 }
 
 THolder<TEvSchemeShard::TEvModifySchemeTransaction> CreateTablePropose(
@@ -85,6 +89,8 @@ THolder<TEvSchemeShard::TEvModifySchemeTransaction> CreateTablePropose(
         return nullptr;
     }
 
+    CreateChangefeedsPropose(propose, item, error);
+
     return propose;
 }
 

ydb/core/tx/schemeshard/schemeshard_import_scheme_getter.cpp

@@ -51,6 +51,11 @@ class TSchemeGetter: public TActorBootstrapped<TSchemeGetter> {
         return errorType == S3Errors::RESOURCE_NOT_FOUND || errorType == S3Errors::NO_SUCH_KEY;
     }
 
+    static TString ChangefeedKeyFromSettings(const Ydb::Import::ImportFromS3Settings& settings, ui32 itemIdx, const TString& changefeedName) {
+        Y_ABORT_UNLESS(itemIdx < (ui32)settings.items_size());
+        return TStringBuilder() << settings.items(itemIdx).source_prefix() << "/" << changefeedName << "/changefeed_description.pb";
+    }
+
     void HeadObject(const TString& key) {
         auto request = Model::HeadObjectRequest()
             .WithKey(key);
@@ -432,6 +437,7 @@ class TSchemeGetter: public TActorBootstrapped<TSchemeGetter> {
     const TString MetadataKey;
     TString SchemeKey;
     const TString PermissionsKey;
+    const TString ChangefeedKey;
 
     const ui32 Retries;
     ui32 Attempt = 0;

ydb/core/ydb_convert/table_description.cpp

@@ -1200,8 +1200,9 @@ void FillChangefeedDescription(Ydb::Table::DescribeTableResult& out,
     }
 }
 
-bool FillChangefeedDescription(NKikimrSchemeOp::TCdcStreamDescription& out,
-        const Ydb::Table::Changefeed& in, Ydb::StatusIds::StatusCode& status, TString& error) {
+template <typename T>
+bool FillChangefeedDescriptionCommon(NKikimrSchemeOp::TCdcStreamDescription& out,
+        const T& in, Ydb::StatusIds::StatusCode& status, TString& error) {
 
     out.SetName(in.name());
     out.SetVirtualTimestamps(in.virtual_timestamps());
@@ -1241,6 +1242,17 @@ bool FillChangefeedDescription(NKikimrSchemeOp::TCdcStreamDescription& out,
         return false;
     }
 
+    for (const auto& [key, value] : in.attributes()) {
+        auto& attr = *out.AddUserAttributes();
+        attr.SetKey(key);
+        attr.SetValue(value);
+    }
+
+    return true;
+}
+
+bool FillChangefeedDescription(NKikimrSchemeOp::TCdcStreamDescription& out,
+        const Ydb::Table::Changefeed& in, Ydb::StatusIds::StatusCode& status, TString& error) {
     if (in.initial_scan()) {
         if (!AppData()->FeatureFlags.GetEnableChangefeedInitialScan()) {
             status = Ydb::StatusIds::UNSUPPORTED;
@@ -1249,14 +1261,12 @@ bool FillChangefeedDescription(NKikimrSchemeOp::TCdcStreamDescription& out,
         }
         out.SetState(NKikimrSchemeOp::ECdcStreamState::ECdcStreamStateScan);
     }
+    return FillChangefeedDescriptionCommon(out, in, status, error);
+}
 
-    for (const auto& [key, value] : in.attributes()) {
-        auto& attr = *out.AddUserAttributes();
-        attr.SetKey(key);
-        attr.SetValue(value);
-    }
-
-    return true;
+bool FillChangefeedDescription(NKikimrSchemeOp::TCdcStreamDescription& out,
+        const Ydb::Table::ChangefeedDescription& in, Ydb::StatusIds::StatusCode& status, TString& error) {
+    return FillChangefeedDescriptionCommon(out, in, status, error);
 }
 
 void FillTableStats(Ydb::Table::DescribeTableResult& out,