added topics

Author: stanislav-shchetinin

ydb/core/tx/schemeshard/schemeshard__init.cpp

@@ -4464,13 +4464,18 @@ struct TSchemeShard::TTxInit : public TTransactionBase<TSchemeShard> {
                         item.Metadata = NBackup::TMetadata::Deserialize(rowset.GetValue<Schema::ImportItems::Metadata>());
                     }
 
-                    if (rowset.HaveValue<Schema::ImportItems::Changefeeds>()) {
+                    if (rowset.HaveValue<Schema::ImportItems::Changefeeds>() && rowset.HaveValue<Schema::ImportItems::Topics>()) {
                         const ui64 count = rowset.GetValue<Schema::ImportItems::Changefeeds>().size();
-                        TVector<Ydb::Table::ChangefeedDescription> changefeeds(count);
+                        TVector<TChangefeedImportDescriptions> changefeeds;
+                        changefeeds.reserve(count);
                         for (ui64 i = 0; i < count; ++i) {
-                            Y_ABORT_UNLESS(ParseFromStringNoSizeLimit(changefeeds[i], rowset.GetValue<Schema::ImportItems::Changefeeds>())[i]);
+                            Ydb::Table::ChangefeedDescription changefeed;
+                            Ydb::Topic::DescribeTopicResult topic;
+                            Y_ABORT_UNLESS(ParseFromStringNoSizeLimit(changefeed, rowset.GetValue<Schema::ImportItems::Changefeeds>())[i]);
+                            Y_ABORT_UNLESS(ParseFromStringNoSizeLimit(topic, rowset.GetValue<Schema::ImportItems::Topics>())[i]);
+                            changefeeds.emplace_back(changefeed, topic);
                         }
-                        item.Changefeeds = changefeeds;
+                        item.Changefeeds = std::move(changefeeds);
                     }
 
                     item.State = static_cast<TImportInfo::EState>(rowset.GetValue<Schema::ImportItems::State>());

ydb/core/tx/schemeshard/schemeshard_import.cpp

@@ -191,15 +191,21 @@ void TSchemeShard::PersistImportItemScheme(NIceDb::TNiceDb& db, const TImportInf
     );
     const ui64 count = item.Changefeeds.size();
     TVector<TString> jsonChangefeeds;
+    TVector<TString> jsonTopics;
     jsonChangefeeds.reserve(count);
+    jsonTopics.reserve(count);
 
-    for (const auto& changefeed : item.Changefeeds) {
+    for (const auto& [changefeed, topic] : item.Changefeeds) {
         jsonChangefeeds.push_back(changefeed.SerializeAsString());
+        jsonTopics.push_back(topic.SerializeAsString());
     }
 
     db.Table<Schema::ImportItems>().Key(importInfo->Id, itemIdx).Update(
         NIceDb::TUpdate<Schema::ImportItems::Changefeeds>(jsonChangefeeds)
     );
+    db.Table<Schema::ImportItems>().Key(importInfo->Id, itemIdx).Update(
+        NIceDb::TUpdate<Schema::ImportItems::Topics>(jsonTopics)
+    );
 }
 
 void TSchemeShard::PersistImportItemPreparedCreationQuery(NIceDb::TNiceDb& db, const TImportInfo::TPtr importInfo, ui32 itemIdx) {

ydb/core/tx/schemeshard/schemeshard_import_flow_proposals.cpp

@@ -20,7 +20,7 @@ bool CreateChangefeedsPropose(THolder<TEvSchemeShard::TEvModifySchemeTransaction
         if (!FillChangefeedDescription(cdcStreamDescription, changefeed, status, error)) {
             return false;
         }
-        cdcStreamDescription.
+        cdcStream.SetRetentionPeriodSeconds(topic.Getretention_period().seconds());
     }
     return true;
 }

ydb/core/tx/schemeshard/schemeshard_schema.h

@@ -1563,6 +1563,7 @@ struct Schema : NIceDb::Schema {
         struct Permissions : Column<11, NScheme::NTypeIds::String> {};
         struct Metadata : Column<12, NScheme::NTypeIds::String> {};
         struct Changefeeds : Column<13, NScheme::NTypeIds::Json> {};
+        struct Topics : Column<14, NScheme::NTypeIds::Json> {};
 
         struct State : Column<7, NScheme::NTypeIds::Byte> {};
         struct WaitTxId : Column<8, NScheme::NTypeIds::Uint64> { using Type = TTxId; };