Merge d7a0c4b into 93cb95f

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,20 @@ struct TSchemeShard::TTxInit : public TTransactionBase<TSchemeShard> {
  item.Metadata = NBackup::TMetadata::Deserialize(rowset.GetValue<Schema::ImportItems::Metadata>());
  }
 
+ if (rowset.HaveValue<Schema::ImportItems::Changefeeds>() && rowset.HaveValue<Schema::ImportItems::Topics>()) {
+ const ui64 count = rowset.GetValue<Schema::ImportItems::Changefeeds>().size();
+ TVector<TImportInfo::TChangefeedImportDescriptions> changefeeds;
+ changefeeds.reserve(count);
+ for (ui64 i = 0; i < count; ++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 = std::move(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,23 @@ 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;
+ TVector<TString> jsonTopics;
+ jsonChangefeeds.reserve(count);
+ jsonTopics.reserve(count);
+
+ 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

@@ -8,6 +8,23 @@
 namespace NKikimr {
 namespace NSchemeShard {
 
+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, topic]: changefeeds) {
+ auto& modifyScheme = *record.AddTransaction();
+ auto& cdcStream = *modifyScheme.MutableCreateCdcStream();
+ Ydb::StatusIds::StatusCode status;
+ auto& cdcStreamDescription = *cdcStream.MutableStreamDescription();
+ if (!FillChangefeedDescription(cdcStreamDescription, changefeed, status, error)) {
+ return false;
+ }
+ cdcStream.SetRetentionPeriodSeconds(topic.Getretention_period().seconds());
+ }
+ return true;
+}
+
 THolder<TEvSchemeShard::TEvModifySchemeTransaction> CreateTablePropose(
  TSchemeShard* ss,
  TTxId txId,
@@ -73,6 +90,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,50 @@ class TSchemeGetter: public TActorBootstrapped<TSchemeGetter> {
  return errorType == S3Errors::RESOURCE_NOT_FOUND || errorType == S3Errors::NO_SUCH_KEY;
  }
 
+ static TString ChangefeedDescriptionKey(const TString& changefeedPrefix) {
+ return TStringBuilder() << changefeedPrefix << "/changefeed_description.pb";
+ }
+
+ static TString TopicDescriptionKey(const TString& changefeedPrefix) {
+ return TStringBuilder() << changefeedPrefix << "/topic_description.pb";
+ }
+
+ void ListObjects(const TString& prefix) {
+ auto request = Model::ListObjectsRequest()
+ .WithPrefix(prefix);
+
+ Send(Client, new TEvExternalStorage::TEvListObjectsRequest(request));
+ }
+
+ void HandleChangefeeds(TEvExternalStorage::TEvListObjectsResponse::TPtr& ev) {
+ const auto& result = ev.Get()->Get()->Result;
+
+ LOG_D("HandleChangefeeds TEvExternalStorage::TEvListObjectResponse"
+ << ": self# " << SelfId()
+ << ", result# " << result);
+
+ if (!CheckResult(result, "ListObject")) {
+ return;
+ }
+
+ const auto& objects = result.GetResult().GetContents();
+ ChangefeedsKeys.reserve(objects.size());
+
+ for (const auto& obj : objects) {
+ const TFsPath& path = obj.GetKey();
+ if (path.GetName() == "changefeed_description.pb") {
+ ChangefeedsKeys.push_back(path.Dirname());
+ }
+ }
+
+ if (!ChangefeedsKeys.empty()) {
+ HeadObject(ChangefeedDescriptionKey(ChangefeedsKeys[0]));
+ } else {
+ Reply();
+ }
+
+ }
+
  void HeadObject(const TString& key) {
  auto request = Model::HeadObjectRequest()
  .WithKey(key);
@@ -128,6 +172,36 @@ class TSchemeGetter: public TActorBootstrapped<TSchemeGetter> {
  GetObject(ChecksumKey, std::make_pair(0, contentLength - 1));
  }
 
+ void HandleChangefeed(TEvExternalStorage::TEvHeadObjectResponse::TPtr& ev) {
+ const auto& result = ev->Get()->Result;
+
+ LOG_D("HandleChangefeed TEvExternalStorage::TEvHeadObjectResponse"
+ << ": self# " << SelfId()
+ << ", result# " << result);
+
+ if (!CheckResult(result, "HeadObject")) {
+ return;
+ }
+
+ const auto contentLength = result.GetResult().GetContentLength();
+ GetObject(ChangefeedDescriptionKey(ChangefeedsKeys[IndexDownloadedChangefeed]), std::make_pair(0, contentLength - 1));
+ }
+
+ void HandleTopic(TEvExternalStorage::TEvHeadObjectResponse::TPtr& ev) {
+ const auto& result = ev->Get()->Result;
+
+ LOG_D("HandleTopic TEvExternalStorage::TEvHeadObjectResponse"
+ << ": self# " << SelfId()
+ << ", result# " << result);
+
+ if (!CheckResult(result, "HeadObject")) {
+ return;
+ }
+
+ const auto contentLength = result.GetResult().GetContentLength();
+ GetObject(TopicDescriptionKey(ChangefeedsKeys[IndexDownloadedChangefeed]), std::make_pair(0, contentLength - 1));
+ }
+
  void GetObject(const TString& key, const std::pair<ui64, ui64>& range) {
  auto request = Model::GetObjectRequest()
  .WithKey(key)
@@ -205,7 +279,7 @@ class TSchemeGetter: public TActorBootstrapped<TSchemeGetter> {
  if (NeedDownloadPermissions) {
  StartDownloadingPermissions();
  } else {
- Reply();
+ StartDownloadingChangefeeds();
  }
  };
 
@@ -242,7 +316,7 @@ class TSchemeGetter: public TActorBootstrapped<TSchemeGetter> {
  item.Permissions = std::move(permissions);
 
  auto nextStep = [this]() {
- Reply();
+ StartDownloadingChangefeeds();
  };
 
  if (NeedValidateChecksums) {
@@ -274,6 +348,82 @@ class TSchemeGetter: public TActorBootstrapped<TSchemeGetter> {
  ChecksumValidatedCallback();
  }
 
+ void HandleChangefeed(TEvExternalStorage::TEvGetObjectResponse::TPtr& ev) {
+ const auto& msg = *ev->Get();
+ const auto& result = msg.Result;
+
+ LOG_D("HandleChangefeed TEvExternalStorage::TEvGetObjectResponse"
+ << ": self# " << SelfId()
+ << ", result# " << result);
+
+ if (!CheckResult(result, "GetObject")) {
+ return;
+ }
+
+ Y_ABORT_UNLESS(ItemIdx < ImportInfo->Items.size());
+ auto& item = ImportInfo->Items.at(ItemIdx);
+
+ LOG_T("Trying to parse changefeed"
+ << ": self# " << SelfId()
+ << ", body# " << SubstGlobalCopy(msg.Body, "\n", "\\n"));
+
+ Ydb::Table::ChangefeedDescription changefeed;
+ if (!google::protobuf::TextFormat::ParseFromString(msg.Body, &changefeed)) {
+ return Reply(false, "Cannot parse сhangefeed");
+ }
+ item.Changefeeds[IndexDownloadedChangefeed].ChangefeedDescription = std::move(changefeed);
+
+ auto nextStep = [this]() {
+ HeadObject(TopicDescriptionKey(ChangefeedsKeys[IndexDownloadedChangefeed]));
+ };
+
+ if (NeedValidateChecksums) {
+ StartValidatingChecksum(ChangefeedDescriptionKey(ChangefeedsKeys[IndexDownloadedChangefeed]), msg.Body, nextStep);
+ } else {
+ nextStep();
+ } 
+ }
+
+ void HandleTopic(TEvExternalStorage::TEvGetObjectResponse::TPtr& ev) {
+ const auto& msg = *ev->Get();
+ const auto& result = msg.Result;
+
+ LOG_D("HandleTopic TEvExternalStorage::TEvGetObjectResponse"
+ << ": self# " << SelfId()
+ << ", result# " << result);
+
+ if (!CheckResult(result, "GetObject")) {
+ return;
+ }
+
+ Y_ABORT_UNLESS(ItemIdx < ImportInfo->Items.size());
+ auto& item = ImportInfo->Items.at(ItemIdx);
+
+ LOG_T("Trying to parse topic"
+ << ": self# " << SelfId()
+ << ", body# " << SubstGlobalCopy(msg.Body, "\n", "\\n"));
+
+ Ydb::Topic::DescribeTopicResult topic;
+ if (!google::protobuf::TextFormat::ParseFromString(msg.Body, &topic)) {
+ return Reply(false, "Cannot parse topic");
+ }
+ item.Changefeeds[IndexDownloadedChangefeed].Topic = std::move(topic);
+
+ auto nextStep = [this]() {
+ if (++IndexDownloadedChangefeed == ChangefeedsKeys.size()) {
+ Reply();
+ } else {
+ HeadObject(ChangefeedDescriptionKey(ChangefeedsKeys[IndexDownloadedChangefeed]));
+ }
+ };
+
+ if (NeedValidateChecksums) {
+ StartValidatingChecksum(TopicDescriptionKey(ChangefeedsKeys[IndexDownloadedChangefeed]), msg.Body, nextStep);
+ } else {
+ nextStep();
+ } 
+ }
+
  template <typename TResult>
  bool CheckResult(const TResult& result, const TStringBuf marker) {
  if (result.IsSuccess()) {
@@ -312,12 +462,20 @@ class TSchemeGetter: public TActorBootstrapped<TSchemeGetter> {
  TActor::PassAway();
  }
 
- void Download(const TString& key) {
+ void DownloadCommon() {
  if (Client) {
  Send(Client, new TEvents::TEvPoisonPill());
  }
  Client = RegisterWithSameMailbox(CreateS3Wrapper(ExternalStorageConfig->ConstructStorageOperator()));
+ }
+
+ void DownloadWithoutKey() {
+ DownloadCommon();
+ ListObjects(ImportInfo->Settings.items(ItemIdx).source_prefix());
+ }
 
+ void Download(const TString& key) {
+ DownloadCommon();
  HeadObject(key);
  }
 
@@ -337,6 +495,10 @@ class TSchemeGetter: public TActorBootstrapped<TSchemeGetter> {
  Download(ChecksumKey);
  }
 
+ void DownloadChangefeeds() {
+ DownloadWithoutKey();
+ }
+
  void ResetRetries() {
  Attempt = 0;
  }
@@ -353,6 +515,12 @@ class TSchemeGetter: public TActorBootstrapped<TSchemeGetter> {
  Become(&TThis::StateDownloadPermissions);
  }
 
+ void StartDownloadingChangefeeds() {
+ ResetRetries();
+ DownloadChangefeeds();
+ Become(&TThis::StateDownloadChangefeeds);
+ }
+
  void StartValidatingChecksum(const TString& key, const TString& object, std::function<void()> checksumValidatedCallback) {
  ChecksumKey = NBackup::ChecksumKey(key);
  Checksum = NBackup::ComputeChecksum(object);
@@ -413,6 +581,17 @@ class TSchemeGetter: public TActorBootstrapped<TSchemeGetter> {
  }
  }
 
+ STATEFN(StateDownloadChangefeeds) {
+ switch (ev->GetTypeRewrite()) {
+ hFunc(TEvExternalStorage::TEvListObjectsResponse, HandleChangefeeds);
+ hFunc(TEvExternalStorage::TEvHeadObjectResponse, HandleChangefeed);
+ hFunc(TEvExternalStorage::TEvGetObjectResponse, HandleChangefeed);
+
+ sFunc(TEvents::TEvWakeup, DownloadChangefeeds);
+ sFunc(TEvents::TEvPoisonPill, PassAway);
+ }
+ }
+
  STATEFN(StateDownloadChecksum) {
  switch (ev->GetTypeRewrite()) {
  hFunc(TEvExternalStorage::TEvHeadObjectResponse, HandleChecksum);
@@ -432,6 +611,8 @@ class TSchemeGetter: public TActorBootstrapped<TSchemeGetter> {
  const TString MetadataKey;
  TString SchemeKey;
  const TString PermissionsKey;
+ TVector<TString> ChangefeedsKeys;
+ ui64 IndexDownloadedChangefeed = 0;
 
  const ui32 Retries;
  ui32 Attempt = 0;

ydb/core/tx/schemeshard/schemeshard_info_types.h

@@ -2837,6 +2837,11 @@ struct TImportInfo: public TSimpleRefCount<TImportInfo> {
  S3 = 0,
  };
 
+ struct TChangefeedImportDescriptions {
+ Ydb::Table::ChangefeedDescription ChangefeedDescription;
+ Ydb::Topic::DescribeTopicResult Topic;
+ };
+
  struct TItem {
  enum class ESubState: ui8 {
  AllocateTxId = 0,
@@ -2851,6 +2856,7 @@ struct TImportInfo: public TSimpleRefCount<TImportInfo> {
  TMaybe<NKikimrSchemeOp::TModifyScheme> PreparedCreationQuery;
  TMaybeFail<Ydb::Scheme::ModifyPermissionsRequest> Permissions;
  NBackup::TMetadata Metadata;
+ TVector<TChangefeedImportDescriptions> Changefeeds;
 
  EState State = EState::GetScheme;
  ESubState SubState = ESubState::AllocateTxId;

ydb/core/tx/schemeshard/schemeshard_schema.h

@@ -1562,6 +1562,8 @@ 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 Topics : Column<14, NScheme::NTypeIds::Json> {};
 
  struct State : Column<7, NScheme::NTypeIds::Byte> {};
  struct WaitTxId : Column<8, NScheme::NTypeIds::Uint64> { using Type = TTxId; };