Merge 462116f into 0a7a679

Author: stanislav-shchetinin

ydb/library/backup/backup.cpp

@@ -3,9 +3,12 @@
 #include "util.h"
 
 #include <ydb/public/api/protos/ydb_table.pb.h>
+#include <ydb/public/lib/ydb_cli/commands/ydb_common.h>
 #include <ydb/public/lib/ydb_cli/common/recursive_remove.h>
+#include <ydb/public/lib/ydb_cli/common/retry_func.h>
 #include <ydb/public/lib/ydb_cli/dump/util/util.h>
 #include <ydb/public/lib/yson_value/ydb_yson_value.h>
+#include <ydb/public/sdk/cpp/client/ydb_proto/accessor.h>
 #include <ydb/public/sdk/cpp/client/ydb_driver/driver.h>
 #include <ydb/public/sdk/cpp/client/ydb_result/result.h>
 #include <ydb/public/sdk/cpp/client/ydb_table/table.h>
@@ -34,6 +37,8 @@ namespace NYdb::NBackup {
 
 static constexpr const char *SCHEME_FILE_NAME = "scheme.pb";
 static constexpr const char *PERMISSIONS_FILE_NAME = "permissions.pb";
+static constexpr const char *CHANGEFEED_DESCRIPTION_FILE_NAME = "changefeed_description.pb";
+static constexpr const char *TOPIC_DESCRIPTION_FILE_NAME = "topic_description.pb";
 static constexpr const char *INCOMPLETE_DATA_FILE_NAME = "incomplete.csv";
 static constexpr const char *INCOMPLETE_FILE_NAME = "incomplete";
 static constexpr const char *EMPTY_FILE_NAME = "empty_dir";
@@ -471,6 +476,53 @@ void BackupPermissions(TDriver driver, const TString& dbPrefix, const TString& p
  outFile.Write(permissionsStr.data(), permissionsStr.size());
 }
 
+TFsPath CreateDirectory(const TFsPath& folderPath, const TString& name) {
+ TFsPath childFolderPath = folderPath.Child(name);
+ LOG_D("Process " << childFolderPath.GetPath().Quote());
+ childFolderPath.MkDir();
+ return childFolderPath;
+}
+
+Ydb::Table::ChangefeedDescription ProtoFromChangefeedDesc(const NTable::TChangefeedDescription& changefeedDesc) {
+ Ydb::Table::ChangefeedDescription protoChangeFeedDesc;
+ changefeedDesc.SerializeTo(protoChangeFeedDesc);
+ return protoChangeFeedDesc;
+}
+
+NTopic::TDescribeTopicResult GetTopicDescription(TDriver driver, const TString& path) {
+ NYdb::NTopic::TTopicClient client(driver);
+ return NConsoleClient::RetryFunction([&]() {
+ return client.DescribeTopic(path).GetValueSync();
+ });
+}
+
+void WriteProtoToFile(const google::protobuf::Message& proto, const TFsPath& folderPath, const TString& fileName) {
+ TString changefeedStr;
+ google::protobuf::TextFormat::PrintToString(proto, &changefeedStr);
+ LOG_D("Write changefeed into " << folderPath.Child(fileName).GetPath().Quote());
+ TFile outFile(folderPath.Child(fileName), CreateAlways | WrOnly);
+ outFile.Write(changefeedStr.data(), changefeedStr.size());
+}
+
+void BackupChangefeeds(TDriver driver, const TString& dbPrefix, const TString& path, const TFsPath& folderPath) {
+
+ const auto dirPath = JoinDatabasePath(dbPrefix, path);
+ auto desc = DescribeTable(driver, dirPath);
+
+ for (const auto& changefeedDesc : desc.GetChangefeedDescriptions()) {
+ TFsPath changefeedDirPath = CreateDirectory(folderPath, changefeedDesc.GetName());
+ 
+ auto protoChangeFeedDesc = ProtoFromChangefeedDesc(changefeedDesc);
+ const auto descTopicResult = GetTopicDescription(driver, JoinDatabasePath(dirPath, changefeedDesc.GetName()));
+ NConsoleClient::ThrowOnError(descTopicResult);
+ const auto& topicDescription = descTopicResult.GetTopicDescription();
+ const auto protoTopicDescription = NYdb::TProtoAccessor::GetProto(topicDescription);
+
+ WriteProtoToFile(protoChangeFeedDesc, changefeedDirPath, CHANGEFEED_DESCRIPTION_FILE_NAME);
+ WriteProtoToFile(protoTopicDescription, changefeedDirPath, TOPIC_DESCRIPTION_FILE_NAME);
+ }
+}
+
 void BackupTable(TDriver driver, const TString& dbPrefix, const TString& backupPrefix, const TString& path,
  const TFsPath& folderPath, bool schemaOnly, bool preservePoolKinds, bool ordered) {
  Y_ENSURE(!path.empty());
@@ -489,6 +541,7 @@ void BackupTable(TDriver driver, const TString& dbPrefix, const TString& backupP
  TFile outFile(folderPath.Child(SCHEME_FILE_NAME), CreateAlways | WrOnly);
  outFile.Write(schemaStr.data(), schemaStr.size());
 
+ BackupChangefeeds(driver, dbPrefix, path, folderPath);
  BackupPermissions(driver, dbPrefix, path, folderPath);
 
  if (!schemaOnly) {

ydb/public/lib/ydb_cli/dump/dump.h

@@ -100,6 +100,7 @@ struct TRestoreSettings: public TOperationRequestSettings<TRestoreSettings> {
  FLUENT_SETTING_DEFAULT(bool, DryRun, false);
  FLUENT_SETTING_DEFAULT(bool, RestoreData, true);
  FLUENT_SETTING_DEFAULT(bool, RestoreIndexes, true);
+ FLUENT_SETTING_DEFAULT(bool, RestoreChangefeed, true);
  FLUENT_SETTING_DEFAULT(bool, RestoreACL, true);
  FLUENT_SETTING_DEFAULT(bool, SkipDocumentTables, false);
  FLUENT_SETTING_DEFAULT(bool, SavePartialResult, false);

ydb/public/lib/ydb_cli/dump/restore_impl.cpp

@@ -363,6 +363,15 @@ TRestoreResult TRestoreClient::RestoreTable(const TFsPath& fsPath, const TString
  LOG_D("Skip restoring indexes of " << dbPath.Quote());
  }
 
+ if (settings.RestoreChangefeed_) {
+ auto result = RestoreChangefeeds(dbPath, dumpedDesc);
+ if (!result.IsSuccess()) {
+ return result;
+ }
+ } else if (!dumpedDesc.GetChangefeedDescriptions().empty()) {
+ LOG_D("Skip restoring changefeeds of " << dbPath.Quote());
+ }
+
  return RestorePermissions(fsPath, dbPath, settings, oldEntries);
 }
 
@@ -626,6 +635,52 @@ TRestoreResult TRestoreClient::RestoreIndexes(const TString& dbPath, const TTabl
  return Result<TRestoreResult>();
 }
 
+TRestoreResult TRestoreClient::RestoreChangefeeds(const TString& dbPath, const TTableDescription& desc) {
+ TMaybe<TTableDescription> actualDesc;
+ auto descResult = DescribeTable(TableClient, dbPath, actualDesc);
+ if (!descResult.IsSuccess()) {
+ return Result<TRestoreResult>(dbPath, std::move(descResult));
+ }
+
+ for (const auto& changefeed : desc.GetChangefeedDescriptions()) {
+ if (FindPtr(actualDesc->GetChangefeedDescriptions(), changefeed)) {
+ continue;
+ }
+ LOG_D("Restore changefeed " << changefeed.GetName().Quote() << " on " << dbPath.Quote());
+
+ TOperation::TOperationId buildChangefeedId;
+ auto buildChangefeedStatus = TableClient.RetryOperationSync([&, &outId = buildChangefeedId](TSession session) {
+ auto settings = TAlterTableSettings().AppendAddChangefeeds(changefeed);
+ auto result = session.AlterTableLong(dbPath, settings).GetValueSync();
+ if (IsOperationStarted(result.Status())) {
+ outId = result.Id();
+ }
+ return result.Status();
+ });
+
+ if (!IsOperationStarted(buildChangefeedStatus)) {
+ LOG_E("Error building changefeed " << changefeed.GetName().Quote() << " on " << dbPath.Quote());
+ return Result<TRestoreResult>(dbPath, std::move(buildChangefeedStatus));
+ }
+
+ auto waitForIndexBuildStatus = WaitForIndexBuild(OperationClient, buildChangefeedId);
+ if (!waitForIndexBuildStatus.IsSuccess()) {
+ LOG_E("Error building changefeed " << changefeed.GetName().Quote() << " on " << dbPath.Quote());
+ return Result<TRestoreResult>(dbPath, std::move(waitForIndexBuildStatus));
+ }
+
+ auto forgetStatus = NConsoleClient::RetryFunction([&]() {
+ return OperationClient.Forget(buildChangefeedId).GetValueSync();
+ });
+ if (!forgetStatus.IsSuccess()) {
+ LOG_E("Error building changefeed " << changefeed.GetName().Quote() << " on " << dbPath.Quote());
+ return Result<TRestoreResult>(dbPath, std::move(forgetStatus));
+ }
+ }
+
+ return Result<TRestoreResult>();
+}
+
 TRestoreResult TRestoreClient::RestorePermissions(const TFsPath& fsPath, const TString& dbPath,
  const TRestoreSettings& settings, const THashSet<TString>& oldEntries)
 { 

ydb/public/lib/ydb_cli/dump/restore_impl.h

@@ -129,6 +129,7 @@ class TRestoreClient {
  TRestoreResult CheckSchema(const TString& dbPath, const NTable::TTableDescription& desc);
  TRestoreResult RestoreData(const TFsPath& fsPath, const TString& dbPath, const TRestoreSettings& settings, const NTable::TTableDescription& desc);
  TRestoreResult RestoreIndexes(const TString& dbPath, const NTable::TTableDescription& desc);
+ TRestoreResult RestoreChangefeeds(const TString& dbPath, const NTable::TTableDescription& desc);
  TRestoreResult RestorePermissions(const TFsPath& fsPath, const TString& dbPath, const TRestoreSettings& settings, const THashSet<TString>& oldEntries);
 
  THolder<NPrivate::IDataWriter> CreateDataWriter(const TString& dbPath, const TRestoreSettings& settings,

ydb/public/sdk/cpp/client/ydb_table/table.cpp

@@ -2869,6 +2869,68 @@ void TChangefeedDescription::SerializeTo(Ydb::Table::Changefeed& proto) const {
  }
 }
 
+void TChangefeedDescription::SerializeTo(Ydb::Table::ChangefeedDescription& proto) const {
+ proto.set_name(Name_);
+ proto.set_virtual_timestamps(VirtualTimestamps_);
+ proto.set_aws_region(AwsRegion_);
+
+ switch (State_) {
+ case EChangefeedState::Enabled:
+ proto.set_state(Ydb::Table::ChangefeedDescription_State::ChangefeedDescription_State_STATE_ENABLED);
+ break;
+ case EChangefeedState::Disabled:
+ proto.set_state(Ydb::Table::ChangefeedDescription_State::ChangefeedDescription_State_STATE_DISABLED);
+ break;
+ case EChangefeedState::InitialScan:
+ proto.set_state(Ydb::Table::ChangefeedDescription_State::ChangefeedDescription_State_STATE_INITIAL_SCAN);
+ break;
+ case EChangefeedState::Unknown:
+ break; 
+ }
+
+ switch (Mode_) {
+ case EChangefeedMode::KeysOnly:
+ proto.set_mode(Ydb::Table::ChangefeedMode::MODE_KEYS_ONLY);
+ break;
+ case EChangefeedMode::Updates:
+ proto.set_mode(Ydb::Table::ChangefeedMode::MODE_UPDATES);
+ break;
+ case EChangefeedMode::NewImage:
+ proto.set_mode(Ydb::Table::ChangefeedMode::MODE_NEW_IMAGE);
+ break;
+ case EChangefeedMode::OldImage:
+ proto.set_mode(Ydb::Table::ChangefeedMode::MODE_OLD_IMAGE);
+ break;
+ case EChangefeedMode::NewAndOldImages:
+ proto.set_mode(Ydb::Table::ChangefeedMode::MODE_NEW_AND_OLD_IMAGES);
+ break;
+ case EChangefeedMode::Unknown:
+ break;
+ }
+
+ switch (Format_) {
+ case EChangefeedFormat::Json:
+ proto.set_format(Ydb::Table::ChangefeedFormat::FORMAT_JSON);
+ break;
+ case EChangefeedFormat::DynamoDBStreamsJson:
+ proto.set_format(Ydb::Table::ChangefeedFormat::FORMAT_DYNAMODB_STREAMS_JSON);
+ break;
+ case EChangefeedFormat::DebeziumJson:
+ proto.set_format(Ydb::Table::ChangefeedFormat::FORMAT_DEBEZIUM_JSON);
+ break;
+ case EChangefeedFormat::Unknown:
+ break;
+ }
+
+ if (ResolvedTimestamps_) {
+ SetDuration(*ResolvedTimestamps_, *proto.mutable_resolved_timestamps_interval());
+ }
+
+ for (const auto& [key, value] : Attributes_) {
+ (*proto.mutable_attributes())[key] = value;
+ }
+}
+
 TString TChangefeedDescription::ToString() const {
  TString result;
  TStringOutput out(result);

ydb/public/sdk/cpp/client/ydb_table/table.h

@@ -389,6 +389,7 @@ class TChangefeedDescription {
  const std::optional<TInitialScanProgress>& GetInitialScanProgress() const;
 
  void SerializeTo(Ydb::Table::Changefeed& proto) const;
+ void SerializeTo(Ydb::Table::ChangefeedDescription& proto) const;
  TString ToString() const;
  void Out(IOutputStream& o) const;