Feature/backups/direct restore

ydb/core/base/appdata_fwd.h

@@ -116,6 +116,7 @@ namespace NMiniKQL {
 
 namespace NDataShard {
     class IExportFactory;
+    class IRestoreIncrementalBackupFactory;
 }
 
 namespace NSQS {
@@ -153,6 +154,7 @@ struct TAppData {
     const NScheme::TTypeRegistry* TypeRegistry = nullptr;
     const NMiniKQL::IFunctionRegistry* FunctionRegistry = nullptr;
     const NDataShard::IExportFactory *DataShardExportFactory = nullptr;
+    const NDataShard::IRestoreIncrementalBackupFactory *DataShardRestoreIncrementalBackupFactory = nullptr;
     const TFormatFactory* FormatFactory = nullptr;
     const NSQS::IEventsWriterFactory* SqsEventsWriterFactory = nullptr;
 

ydb/core/change_exchange/change_sender_common_ops.h

@@ -137,6 +137,9 @@ class TBaseChangeSender {
             auto it = Senders.find(partitionId);
             if (it != Senders.end()) {
                 senders.emplace(partitionId, std::move(it->second));
+                if (it->second.Ready) {
+                    --ReadySenders;
+                }
                 Senders.erase(it);
             } else {
                 LazyCreateSender(senders, partitionId);
@@ -208,6 +211,8 @@ class TBaseChangeSender {
         THashSet<ui64> registrations;
         bool needToResolve = false;
 
+        // int tmp = PendingSent.size();
+
         while (it != PendingSent.end()) {
             if (Enqueued && Enqueued.begin()->Order <= it->first) {
                 break;
@@ -258,6 +263,8 @@ class TBaseChangeSender {
             it = PendingSent.erase(it);
         }
 
+        // Y_ABORT_S("something strange: " << sendTo.size() << " " << tmp);
+
         for (const auto partitionId : registrations) {
             RegisterSender(partitionId);
         }
@@ -279,6 +286,7 @@ class TBaseChangeSender {
 
         Y_ABORT_UNLESS(sender.Ready);
         sender.Ready = false;
+        ReadySenders--;
 
         sender.Pending.reserve(sender.Prepared.size());
         for (const auto& record : sender.Prepared) {
@@ -527,6 +535,7 @@ class TBaseChangeSender {
 
         auto& sender = it->second;
         sender.Ready = true;
+        ReadySenders++;
 
         if (sender.Pending) {
             RemoveRecords(std::exchange(sender.Pending, {}));
@@ -551,6 +560,9 @@ class TBaseChangeSender {
         }
 
         ReEnqueueRecords(it->second);
+        if (it->second.Ready) {
+            --ReadySenders;
+        }
         Senders.erase(it);
         GonePartitions.push_back(partitionId);
 
@@ -576,6 +588,10 @@ class TBaseChangeSender {
             , MemUsage(0)
     {}
 
+    bool AllReady() {
+        return ReadySenders == Senders.size();
+    }
+
     void RenderHtmlPage(ui64 tabletId, NMon::TEvRemoteHttpInfo::TPtr& ev, const TActorContext& ctx) {
         const auto& cgi = ev->Get()->Cgi();
         if (const auto& str = cgi.Get("partitionId")) {
@@ -776,6 +792,7 @@ class TBaseChangeSender {
     ui64 MemUsage;
 
     THashMap<ui64, TSender> Senders; // ui64 is partition id
+    ui64 ReadySenders = 0;
     TSet<TEnqueuedRecord> Enqueued;
     TSet<TIncompleteRecord> PendingBody;
     TMap<ui64, typename TChangeRecord::TPtr> PendingSent; // ui64 is order

ydb/core/protos/flat_scheme_op.proto

@@ -1015,7 +1015,10 @@ message TDropContinuousBackup {
 
 message TRestoreIncrementalBackup {
     optional string SrcTableName = 1;
+    optional NKikimrProto.TPathID SrcPathId = 3;
+
     optional string DstTableName = 2;
+    optional NKikimrProto.TPathID DstPathId = 4;
 }
 
 enum EIndexType {

ydb/core/protos/tx_datashard.proto

@@ -480,6 +480,8 @@ message TFlatSchemeTransaction {
     optional TAlterCdcStreamNotice AlterCdcStreamNotice = 19;
     optional TDropCdcStreamNotice DropCdcStreamNotice = 20;
     optional TMoveIndex MoveIndex = 21;
+
+    optional NKikimrSchemeOp.TRestoreIncrementalBackup RestoreIncrementalBackupSrc = 22;
 }
 
 message TDistributedEraseTransaction {