start

Author: stanislav-shchetinin

ydb/core/tablet_flat/flat_cxx_database.h

@@ -238,6 +238,7 @@ template <> struct NSchemeTypeMapper<NScheme::NTypeIds::Date32> { typedef i32 Ty
 template <> struct NSchemeTypeMapper<NScheme::NTypeIds::Datetime64> { typedef i64 Type; };
 template <> struct NSchemeTypeMapper<NScheme::NTypeIds::Timestamp64> { typedef i64 Type; };
 template <> struct NSchemeTypeMapper<NScheme::NTypeIds::Interval64> { typedef i64 Type; };
+template <> struct NSchemeTypeMapper<NScheme::NTypeIds::Json> { typedef TVector<TString> Type; };
 
 /// only for compatibility with old code
 template <NScheme::TTypeId ValType>

ydb/core/tx/schemeshard/schemeshard__init.cpp

@@ -4464,6 +4464,15 @@ struct TSchemeShard::TTxInit : public TTransactionBase<TSchemeShard> {
                         item.Metadata = NBackup::TMetadata::Deserialize(rowset.GetValue<Schema::ImportItems::Metadata>());
                     }
 
+                    if (rowset.HaveValue<Schema::ImportItems::Changefeeds>()) {
+                        const ui64 count = rowset.GetValue<Schema::ImportItems::Changefeeds>().size();
+                        TVector<Ydb::Table::ChangefeedDescription> changefeeds(count);
+                        for (ui64 i = 0; i < count; ++i) {
+                            Y_ABORT_UNLESS(ParseFromStringNoSizeLimit(changefeeds[i], rowset.GetValue<Schema::ImportItems::Changefeeds>())[i]);
+                        }
+                        item.Changefeeds = changefeeds;
+                    }
+
                     item.State = static_cast<TImportInfo::EState>(rowset.GetValue<Schema::ImportItems::State>());
                     item.WaitTxId = rowset.GetValueOrDefault<Schema::ImportItems::WaitTxId>(InvalidTxId);
                     item.NextIndexIdx = rowset.GetValueOrDefault<Schema::ImportItems::NextIndexIdx>(0);

ydb/core/tx/schemeshard/schemeshard_import.cpp

@@ -189,6 +189,17 @@ void TSchemeShard::PersistImportItemScheme(NIceDb::TNiceDb& db, const TImportInf
     db.Table<Schema::ImportItems>().Key(importInfo->Id, itemIdx).Update(
         NIceDb::TUpdate<Schema::ImportItems::Metadata>(item.Metadata.Serialize())
     );
+    const ui64 count = item.Changefeeds.size();
+    TVector<TString> jsonChangefeeds;
+    jsonChangefeeds.reserve(count);
+
+    for (const auto& changefeed : item.Changefeeds) {
+        jsonChangefeeds.push_back(changefeed.SerializeAsString());
+    }
+
+    db.Table<Schema::ImportItems>().Key(importInfo->Id, itemIdx).Update(
+        NIceDb::TUpdate<Schema::ImportItems::Changefeeds>(jsonChangefeeds)
+    );
 }
 
 void TSchemeShard::PersistImportItemPreparedCreationQuery(NIceDb::TNiceDb& db, const TImportInfo::TPtr importInfo, ui32 itemIdx) {

ydb/core/tx/schemeshard/schemeshard_info_types.h

@@ -2851,6 +2851,7 @@ struct TImportInfo: public TSimpleRefCount<TImportInfo> {
         TMaybe<NKikimrSchemeOp::TModifyScheme> PreparedCreationQuery;
         TMaybeFail<Ydb::Scheme::ModifyPermissionsRequest> Permissions;
         NBackup::TMetadata Metadata;
+        TVector<Ydb::Table::ChangefeedDescription> Changefeeds;
 
         EState State = EState::GetScheme;
         ESubState SubState = ESubState::AllocateTxId;

ydb/core/tx/schemeshard/schemeshard_schema.h

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