Merge a62c95e into 55e9497

Author: stanislav-shchetinin

ydb/core/protos/flat_scheme_op.proto

@@ -1247,6 +1247,8 @@ message TBackupTask {
     optional uint64 SnapshotStep = 14;
     optional uint64 SnapshotTxId = 15;
     optional bool EnableChecksums = 16; // currently available for s3
+
+    repeated TPathDescription PersQueue = 17; // for further restore
 }
 
 message TRestoreTask {

ydb/core/tx/datashard/backup_restore_traits.cpp

@@ -76,6 +76,14 @@ TString PermissionsKeySuffix() {
     return "permissions.pb";
 }
 
+TString TopicKeySuffix() {
+    return "topic_description.pb";
+}
+
+TString ChangefeedKeySuffix() {
+    return "changefeed_description.pb";
+}
+
 TString SchemeKeySuffix() {
     return "scheme.pb";
 }

ydb/core/tx/datashard/backup_restore_traits.h

@@ -31,6 +31,8 @@ ECompressionCodec NextCompressionCodec(ECompressionCodec cur);
 TString DataFileExtension(EDataFormat format, ECompressionCodec codec);
 
 TString PermissionsKeySuffix();
+TString TopicKeySuffix();
+TString ChangefeedKeySuffix();
 TString SchemeKeySuffix();
 TString MetadataKeySuffix();
 TString DataKeySuffix(ui32 n, EDataFormat format, ECompressionCodec codec);

ydb/core/tx/datashard/export_common.cpp

@@ -3,6 +3,8 @@
 #include <ydb/core/engine/mkql_proto.h>
 #include <ydb/core/ydb_convert/table_description.h>
 #include <ydb/core/ydb_convert/ydb_convert.h>
+#include <ydb/services/persqueue_v1/actors/schema_actors.h>
+#include <ydb/library/persqueue/obfuscate/obfuscate.h>
 
 #include <util/generic/algorithm.h>
 

ydb/core/tx/datashard/export_s3.h

@@ -7,6 +7,9 @@
 #include "export_iface.h"
 #include "export_s3_buffer.h"
 
+#include <ydb/public/api/protos/ydb_table.pb.h>
+#include <ydb/public/api/protos/ydb_topic.pb.h>
+
 namespace NKikimr {
 namespace NDataShard {
 
@@ -47,6 +50,11 @@ class TS3Export: public IExport {
     const TTableColumns Columns;
 };
 
+struct TChangefeedExportDescriptions {
+    Ydb::Table::ChangefeedDescription ChangefeedDescription;
+    Ydb::Topic::DescribeTopicResult Topic;
+}
+
 } // NDataShard
 } // NKikimr
 

ydb/core/tx/datashard/export_s3_uploader.cpp

@@ -12,6 +12,8 @@
 #include <ydb/core/wrappers/s3_storage_config.h>
 #include <ydb/core/wrappers/s3_wrapper.h>
 #include <ydb/core/wrappers/events/common.h>
+#include <ydb/core/ydb_convert/table_description.h>
+#include <ydb/core/ydb_convert/topic_description.h>
 #include <ydb/library/actors/core/actor_bootstrapped.h>
 #include <ydb/library/actors/core/hfunc.h>
 #include <ydb/library/actors/http/http_proxy.h>
@@ -163,6 +165,8 @@ class TS3Uploader: public TActorBootstrapped<TS3Uploader> {
             UploadPermissions();
         } else if (!SchemeUploaded) {
             UploadScheme();
+        } else if (!ChangefeedsUploaded) {
+            UploadChangefeeds();
         } else {
             this->Become(&TThis::StateUploadData);
 
@@ -212,6 +216,60 @@ class TS3Uploader: public TActorBootstrapped<TS3Uploader> {
         this->Become(&TThis::StateUploadPermissions);
     }
 
+    void PutChangefeedDescription(const Ydb::Table::ChangefeedDescription& changefeed, const TString& changefeedKeyPattern, const ui64 index) {
+        google::protobuf::TextFormat::PrintToString(changefeed, &Buffer);
+        auto request = Aws::S3::Model::PutObjectRequest()
+            .WithKey(Settings.GetChangefeedKey(changefeedKeyPattern));
+        this->Send(Client, new TEvExternalStorage::TEvPutObjectRequest(request, std::move(Buffer)));
+        this->Become(TThis::StateUploadOneChangefeed(index));
+    }
+
+    void PutTopicDescription(const Ydb::Topic::DescribeTopicResult& topic, const TString& changefeedKeyPattern, const ui64 index) {
+        google::protobuf::TextFormat::PrintToString(topic, &Buffer);
+        auto request = Aws::S3::Model::PutObjectRequest()
+            .WithKey(Settings.GetTopicKey(changefeedKeyPattern));
+        this->Send(Client, new TEvExternalStorage::TEvPutObjectRequest(request, std::move(Buffer)));
+        this->Become(TThis::StateUploadOneTopic(index));
+    }
+
+    void UploadOneChangefeed(ui64 index) {
+        if (index >= ChangefeedsExportDescs.size()) {
+            //Error!
+            return;
+        }
+        if (index == ChangefeedsExportDescs.size()) {
+            ChangefeedsUploaded = true;
+            this->Become(&TThis::StateUploadData);
+            return;
+        }
+        const auto& changefeed = ChangefeedsExportDescs[index].ChangefeedDescription;
+        const auto changefeedKeyPattern = TStringBuilder() << Settings.ObjectKeyPattern << "/" << changefeed.Getname();
+        PutChangefeedDescription(changefeed, changefeedKeyPattern, index);
+    }
+
+    void UploadOneTopic(ui64 index) {
+        if (index >= ChangefeedsExportDescs.size()) {
+            //Error!
+            return;
+        }
+        auto& descs = ChangefeedsExportDescs[index];
+        const auto changefeedKeyPattern = TStringBuilder() << Settings.ObjectKeyPattern << "/" 
+            << descs.ChangefeedDescription.Getname();
+        PutTopicDescription(descs.Topic, changefeedKeyPattern, index);
+    }
+
+    void UploadChangefeeds() {
+        Y_ABORT_UNLESS(!ChangefeedsUploaded);
+
+        for (const auto &[changefeed, topic] : ChangefeedsExportDescs) {
+            google::protobuf::TextFormat::PrintToString(changefeed, &Buffer);
+            const auto changefeedKeyPattern = TStringBuilder() << Settings.ObjectKeyPattern << "/" << changefeed.Getname();
+            PutChangefeedDescription(changefeed, changefeedKeyPattern);
+            PutTopicDescription(topic, changefeedKeyPattern);
+        }
+        this->Become(&TThis::StateUploadScheme);
+    }
+
     void UploadMetadata() {
         Y_ABORT_UNLESS(!MetadataUploaded);
 
@@ -258,7 +316,7 @@ class TS3Uploader: public TActorBootstrapped<TS3Uploader> {
             if (Scanner) {
                 this->Send(Scanner, new TEvExportScan::TEvFeed());
             }
-            this->Become(&TThis::StateUploadData);
+            UploadOneChangefeed(0);
         };
 
         if (EnableChecksums) {
@@ -293,6 +351,44 @@ class TS3Uploader: public TActorBootstrapped<TS3Uploader> {
         }
     }
 
+    auto HandleOneChangefeed(ui64 index) {
+        return [index, this](TEvExternalStorage::TEvPutObjectResponse::TPtr& ev) {
+            const auto& result = ev->Get()->Result;
+
+            EXPORT_LOG_D("HandleMetadata TEvExternalStorage::TEvPutObjectResponse"
+                << ": self# " << this->SelfId()
+                << ", result# " << result);
+
+            if (!CheckResult(result, TStringBuf("PutObject (changefeed)"))) {
+                return;
+            }
+
+            auto nextStep = [index, this]() {
+                UploadOneTopic(index);
+            };
+            nextStep();
+        };
+    }
+
+    auto HandleOneTopic(ui64 index) {
+        return [index, this](TEvExternalStorage::TEvPutObjectResponse::TPtr& ev) {
+            const auto& result = ev->Get()->Result;
+
+            EXPORT_LOG_D("HandleMetadata TEvExternalStorage::TEvPutObjectResponse"
+                << ": self# " << this->SelfId()
+                << ", result# " << result);
+
+            if (!CheckResult(result, TStringBuf("PutObject (topic)"))) {
+                return;
+            }
+
+            auto nextStep = [index, this]() {
+                UploadOneChangefeed(index + 1);
+            };
+            nextStep();
+        };
+    }
+
     void HandleMetadata(TEvExternalStorage::TEvPutObjectResponse::TPtr& ev) {
         const auto& result = ev->Get()->Result;
 
@@ -659,6 +755,7 @@ class TS3Uploader: public TActorBootstrapped<TS3Uploader> {
             const TActorId& dataShard, ui64 txId,
             const NKikimrSchemeOp::TBackupTask& task,
             TMaybe<Ydb::Table::CreateTableRequest>&& scheme,
+            TVector<TChangefeedExportDescriptions> changefeedsExportDescs,
             TMaybe<Ydb::Scheme::ModifyPermissionsRequest>&& permissions,
             TString&& metadata)
         : ExternalStorageConfig(new TS3ExternalStorageConfig(task.GetS3Settings()))
@@ -670,6 +767,7 @@ class TS3Uploader: public TActorBootstrapped<TS3Uploader> {
         , DataShard(dataShard)
         , TxId(txId)
         , Scheme(std::move(scheme))
+        , ChangefeedsExportDescs(std::move(changefeedsExportDescs))
         , Metadata(std::move(metadata))
         , Permissions(std::move(permissions))
         , Retries(task.GetNumberOfRetries())
@@ -728,6 +826,26 @@ class TS3Uploader: public TActorBootstrapped<TS3Uploader> {
         }
     }
 
+    std::function<void(TAutoPtr<::NActors::IEventHandle>&)> StateUploadOneChangefeed(ui64 index) {
+        return [index, this](TAutoPtr<::NActors ::IEventHandle> &ev) {
+            switch (ev->GetTypeRewrite()) {
+                hFunc(TEvExternalStorage::TEvPutObjectResponse, HandleOneChangefeed(index));
+            default:
+                return StateBase(ev);
+            }
+        };
+    }
+
+    std::function<void(TAutoPtr<::NActors::IEventHandle>&)> StateUploadOneTopic(ui64 index) {
+        return [index, this](TAutoPtr<::NActors ::IEventHandle> &ev) {
+            switch (ev->GetTypeRewrite()) {
+                hFunc(TEvExternalStorage::TEvPutObjectResponse, HandleOneTopic(index));
+            default:
+                return StateBase(ev);
+            }
+        };
+    }
+
     STATEFN(StateUploadMetadata) {
         switch (ev->GetTypeRewrite()) {
             hFunc(TEvExternalStorage::TEvPutObjectResponse, HandleMetadata);
@@ -773,6 +891,7 @@ class TS3Uploader: public TActorBootstrapped<TS3Uploader> {
     const TActorId DataShard;
     const ui64 TxId;
     const TMaybe<Ydb::Table::CreateTableRequest> Scheme;
+    TVector<TChangefeedExportDescriptions> ChangefeedsExportDescs;
     const TString Metadata;
     const TMaybe<Ydb::Scheme::ModifyPermissionsRequest> Permissions;
 
@@ -784,6 +903,7 @@ class TS3Uploader: public TActorBootstrapped<TS3Uploader> {
 
     TActorId Client;
     bool SchemeUploaded;
+    bool ChangefeedsUploaded;
     bool MetadataUploaded;
     bool PermissionsUploaded;
     bool MultiPart;
@@ -810,6 +930,17 @@ IActor* TS3Export::CreateUploader(const TActorId& dataShard, ui64 txId) const {
         ? GenYdbScheme(Columns, Task.GetTable())
         : Nothing();
 
+    const auto& persQueuesTPathDesc = Task.GetPersQueue();
+
+    const auto& cdcStreams = Task.GetTable().GetTable().GetCdcStreams();
+    const int changefeedsCount = cdcStreams.size();
+    TVector <TChangefeedExportDescriptions> changefeedsExportDescs(changefeedsCount);
+
+    for (int i = 0; i < changefeedsCount; ++i) {
+        FillChangefeedDescription(changefeedsExportDescs[i].ChangefeedDescription, cdcStreams[i]);
+        FillTopicDescription(changefeedsExportDescs[i].Topic, persQueuesTPathDesc[i].GetPersQueueGroup());
+    }
+
     auto permissions = (Task.GetShardNum() == 0)
         ? GenYdbPermissions(Task.GetTable())
         : Nothing();
@@ -825,7 +956,7 @@ IActor* TS3Export::CreateUploader(const TActorId& dataShard, ui64 txId) const {
     metadata.AddFullBackup(backup);
 
     return new TS3Uploader(
-        dataShard, txId, Task, std::move(scheme), std::move(permissions), metadata.Serialize());
+        dataShard, txId, Task, std::move(scheme), std::move(changefeedsExportDescs), std::move(permissions), metadata.Serialize());
 }
 
 } // NDataShard

ydb/core/tx/datashard/extstorage_usage_config.h

@@ -46,6 +46,14 @@ class TS3Settings {
         return ObjectKeyPattern + '/' + NBackupRestoreTraits::PermissionsKeySuffix();
     }
 
+    inline TString GetTopicKey(const TString& changefeedName) const {
+        return TStringBuilder() << ObjectKeyPattern << '/'<< changefeedName << "/" << NBackupRestoreTraits::TopicKeySuffix();
+    }
+
+     inline TString GetChangefeedKey(const TString& changefeedName) const {
+        return TStringBuilder() << ObjectKeyPattern << '/'<< changefeedName << "/" << NBackupRestoreTraits::ChangefeedKeySuffix();
+    }
+
     inline TString GetMetadataKey() const {
         return ObjectKeyPattern + '/' + NBackupRestoreTraits::MetadataKeySuffix();
     }

ydb/core/tx/schemeshard/schemeshard_export_flow_proposals.cpp

@@ -71,7 +71,7 @@ THolder<TEvSchemeShard::TEvModifySchemeTransaction> CopyTablesPropose(
     return propose;
 }
 
-static NKikimrSchemeOp::TPathDescription GetTableDescription(TSchemeShard* ss, const TPathId& pathId) {
+static NKikimrSchemeOp::TPathDescription GetDescription(TSchemeShard* ss, const TPathId& pathId) {
     NKikimrSchemeOp::TDescribeOptions opts;
     opts.SetReturnPartitioningInfo(false);
     opts.SetReturnPartitionConfig(true);
@@ -145,12 +145,27 @@ THolder<TEvSchemeShard::TEvModifySchemeTransaction> BackupPropose(
     const TPath sourcePath = TPath::Init(exportInfo->Items[itemIdx].SourcePathId, ss);
     const TPath exportItemPath = exportPath.Child(ToString(itemIdx));
     if (sourcePath.IsResolved() && exportItemPath.IsResolved()) {
-        auto sourceDescription = GetTableDescription(ss, sourcePath.Base()->PathId);
+        auto sourceDescription = GetDescription(ss, sourcePath.Base()->PathId);
+        Cerr << "srcD: " << sourceDescription.DebugString() << Endl;
         if (sourceDescription.HasTable()) {
             FillSetValForSequences(
                 ss, *sourceDescription.MutableTable(), exportItemPath.Base()->PathId);
             FillPartitioning(ss, *sourceDescription.MutableTable(), exportItemPath.Base()->PathId);
         }
+        auto cdcDesc = sourceDescription.GetTable().GetCdcStreams();
+        for (const auto& x : cdcDesc) {
+            TPathId pathId = TPathId::FromProto(x.GetPathId());
+            auto cdcPathDesc =  GetDescription(ss, pathId);
+            for (const auto& child : cdcPathDesc.GetChildren()) {
+                if (child.GetPathType() == NKikimrSchemeOp::EPathTypePersQueueGroup) {
+                    TPathId pathId = {child.GetSchemeshardId(), child.GetPathId()};
+                    ::NKikimrSchemeOp::TPathDescription* newPersQueue = task.AddPersQueue();
+                    *newPersQueue = GetDescription(ss, pathId);
+                }
+            }
+        }
+        
+        Cerr << "tsz: " << task.GetPersQueue().size() << Endl;
         task.MutableTable()->CopyFrom(sourceDescription);
     }