done

Author: Enjection

ydb/core/tx/datashard/datashard_ut_incremental_restore_scan.cpp

@@ -117,26 +117,31 @@ Y_UNIT_TEST_SUITE(IncrementalRestoreScan) {
         runtime.SetLogPriority(NKikimrServices::TX_PROXY, NLog::PRI_TRACE);
 
         TUserTable::TPtr table = new TUserTable;
-
-        table->Columns.emplace(0, TUserTable::TUserColumn{});
-
         NTable::TScheme::TTableSchema tableSchema;
-        tableSchema.Columns[0] = NTable::TColumn("test", 0, {}, "");
+        table->Columns.emplace(0, TUserTable::TUserColumn(NScheme::TTypeInfo(NScheme::NTypeIds::Uint32), "", "Key", true));
+        tableSchema.Columns[0] = NTable::TColumn("key", 0, {}, "");
         tableSchema.Columns[0].KeyOrder = 0;
+
+        table->Columns.emplace(1, TUserTable::TUserColumn(NScheme::TTypeInfo(NScheme::NTypeIds::Bool), "", "__ydb_incrBackupImpl_deleted", false));
+        tableSchema.Columns[1] = NTable::TColumn("__ydb_incrBackupImpl_deleted", 1, {}, "");
+        tableSchema.Columns[1].KeyOrder = 1;
+
         auto scheme = NTable::TRowScheme::Make(tableSchema.Columns, NUtil::TSecond());
 
-        TPathId targetPathId{};
-        ui64 txId = 0;
+        TPathId sourcePathId{1, 2};
+        TPathId targetPathId{3, 4};
+        ui64 txId = 1337;
 
         auto* scan = CreateIncrementalRestoreScan(
             sender,
             [&](const TActorContext&) {
                 return sender2;
             },
-            TPathId{} /*sourcePathId*/,
+            sourcePathId,
             table,
             targetPathId,
-            txId).Release();
+            txId,
+            {}).Release();
 
         TDriverMock driver;
 

ydb/core/tx/datashard/incr_restore_scan.cpp

@@ -2,15 +2,14 @@
 #include "change_exchange_impl.h"
 #include "datashard_impl.h"
 
-#include <ydb/library/actors/core/actor.h>
-#include <ydb/core/tx/datashard/change_record_body_serializer.h>
-#include <ydb/core/tx/datashard/datashard_user_table.h>
-#include <ydb/core/tx/datashard/change_record.h>
 #include <ydb/core/change_exchange/change_exchange.h>
 #include <ydb/core/tx/datashard/change_collector.h>
-#include <ydb/library/services/services.pb.h>
-#include <ydb/core/tx/datashard/stream_scan_common.h>
+#include <ydb/core/tx/datashard/change_record.h>
+#include <ydb/core/tx/datashard/change_record_body_serializer.h>
+#include <ydb/core/tx/datashard/datashard_user_table.h>
 #include <ydb/core/tx/datashard/incr_restore_helpers.h>
+#include <ydb/library/actors/core/actor.h>
+#include <ydb/library/services/services.pb.h>
 
 namespace NKikimr::NDataShard {
 
@@ -45,27 +44,34 @@ class TIncrementalRestoreScan
             ui64 txId,
             const TPathId& sourcePathId,
             TUserTable::TCPtr table,
-            const TPathId& targetPathId)
-        : IActorCallback(static_cast<TReceiveFunc>(&TIncrementalRestoreScan::StateWork), NKikimrServices::TActivity::CDC_STREAM_SCAN_ACTOR)
+            const TPathId& targetPathId,
+            NStreamScan::TLimits limits)
+        : IActorCallback(static_cast<TReceiveFunc>(&TIncrementalRestoreScan::StateWork), NKikimrServices::TActivity::INCREMENTAL_RESTORE_SCAN_ACTOR)
         , Parent(parent)
         , ChangeSenderFactory(changeSenderFactory)
         , TxId(txId)
         , SourcePathId(sourcePathId)
         , TargetPathId(targetPathId)
         , ValueTags(InitValueTags(table))
-        , Limits()
+        , Limits(limits)
     {}
 
     static TVector<TTag> InitValueTags(TUserTable::TCPtr table) {
-        Y_VERIFY(table->Columns.size() >= 1);
+        Y_VERIFY(table->Columns.size() >= 2);
         TVector<TTag> valueTags;
         valueTags.reserve(table->Columns.size() - 1);
+        bool deletedMarkerColumnFound = false;
         for (const auto& [tag, column] : table->Columns) {
             if (!column.IsKey) {
                 valueTags.push_back(tag);
+                if (column.Name == "__ydb_incrBackupImpl_deleted") {
+                    deletedMarkerColumnFound = true;
+                }
             }
         }
 
+        Y_VERIFY(deletedMarkerColumnFound);
+
         return valueTags;
     }
 
@@ -267,15 +273,17 @@ THolder<NTable::IScan> CreateIncrementalRestoreScan(
         TPathId sourcePathId,
         TUserTable::TCPtr table,
         const TPathId& targetPathId,
-        ui64 txId)
+        ui64 txId,
+        NStreamScan::TLimits limits)
 {
     return MakeHolder<TIncrementalRestoreScan>(
         parent,
         changeSenderFactory,
         txId,
         sourcePathId,
         table,
-        targetPathId);
+        targetPathId,
+        limits);
 }
 
 } // namespace NKikimr::NDataShard

ydb/core/tx/datashard/incr_restore_scan.h

@@ -1,10 +1,11 @@
 #pragma once
 
+#include <ydb/core/base/events.h>
 #include <ydb/core/scheme/scheme_pathid.h>
 #include <ydb/core/tablet_flat/flat_scan_iface.h>
-#include <ydb/library/actors/core/actor.h>
-#include <ydb/core/base/events.h>
 #include <ydb/core/tx/datashard/datashard_user_table.h>
+#include <ydb/core/tx/datashard/stream_scan_common.h>
+#include <ydb/library/actors/core/actor.h>
 
 #include <functional>
 
@@ -32,6 +33,7 @@ THolder<NTable::IScan> CreateIncrementalRestoreScan(
         TPathId tablePathId,
         TUserTable::TCPtr table,
         const TPathId& targetPathId,
-        ui64 txId);
+        ui64 txId,
+        NStreamScan::TLimits limits);
 
 } // namespace NKikimr::NDataShard