Require snapshot in internal scans (#14304)

Author: zverevgeny

ydb/core/tx/columnshard/columnshard.h

@@ -98,14 +98,13 @@ namespace TEvColumnShard {
     struct TEvInternalScan: public TEventLocal<TEvInternalScan, EvInternalScan> {
     private:
         YDB_READONLY(ui64, PathId, 0);
+        YDB_READONLY(NOlap::TSnapshot, Snapshot, NOlap::TSnapshot::Zero());
         YDB_READONLY_DEF(std::optional<ui64>, LockId);
         YDB_ACCESSOR(bool, Reverse, false);
         YDB_ACCESSOR(ui32, ItemsLimit, 0);
         YDB_READONLY_DEF(std::vector<ui32>, ColumnIds);
         std::set<ui32> ColumnIdsSet;
     public:
-        std::optional<NOlap::TSnapshot> ReadFromSnapshot;
-        std::optional<NOlap::TSnapshot> ReadToSnapshot;
         TString TaskIdentifier;
         std::shared_ptr<NOlap::TPKRangesFilter> RangesFilter;
     public:
@@ -114,11 +113,12 @@ namespace TEvColumnShard {
             ColumnIds.emplace_back(id);
         }
 
-        TEvInternalScan(const ui64 pathId, const std::optional<ui64> lockId)
+        TEvInternalScan(const ui64 pathId, const NOlap::TSnapshot& snapshot, const std::optional<ui64> lockId)
             : PathId(pathId)
+            , Snapshot(snapshot)
             , LockId(lockId)
         {
-
+            AFL_VERIFY(Snapshot.Valid());
         }
     };
 

ydb/core/tx/columnshard/engines/reader/transaction/tx_internal_scan.cpp

@@ -34,7 +34,7 @@ void TTxInternalScan::Complete(const TActorContext& ctx) {
 
     auto& request = *InternalScanEvent->Get();
     auto scanComputeActor = InternalScanEvent->Sender;
-    const TSnapshot snapshot = request.ReadToSnapshot.value_or(NOlap::TSnapshot(Self->LastPlannedStep, Self->LastPlannedTxId));
+    const TSnapshot snapshot = request.GetSnapshot();
     const NActors::TLogContextGuard gLogging =
         NActors::TLogContextBuilder::Build()("tablet", Self->TabletID())("snapshot", snapshot.DebugString())("task_id", request.TaskIdentifier);
     TReadMetadataPtr readMetadataRange;

ydb/core/tx/columnshard/operations/batch_builder/restore.cpp

@@ -8,9 +8,8 @@
 namespace NKikimr::NOlap {
 
 std::unique_ptr<TEvColumnShard::TEvInternalScan> TModificationRestoreTask::DoBuildRequestInitiator() const {
-    auto request = std::make_unique<TEvColumnShard::TEvInternalScan>(LocalPathId, WriteData.GetWriteMeta().GetLockIdOptional());
+    auto request = std::make_unique<TEvColumnShard::TEvInternalScan>(LocalPathId, Snapshot, WriteData.GetWriteMeta().GetLockIdOptional());
     request->TaskIdentifier = GetTaskId();
-    request->ReadToSnapshot = Snapshot;
     AFL_DEBUG(NKikimrServices::TX_COLUMNSHARD_RESTORE)("event", "restore_start")("count", IncomingData.HasContainer() ? IncomingData->num_rows() : 0)(
         "task_id", WriteData.GetWriteMeta().GetId());
     auto pkData = NArrow::TColumnOperator().VerifyIfAbsent().Extract(IncomingData.GetContainer(), Context.GetActualSchema()->GetPKColumnNames());
@@ -70,6 +69,7 @@ TModificationRestoreTask::TModificationRestoreTask(NEvWrite::TWriteData&& writeD
     , Snapshot(actualSnapshot)
     , IncomingData(incomingData)
     , Context(context) {
+        AFL_VERIFY(actualSnapshot.Valid());
 }
 
 void TModificationRestoreTask::SendErrorMessage(