Support batch/scan mode for TEvRead (#11201)

Author: snaury

ydb/core/tablet_flat/flat_scan_actor.h

@@ -314,7 +314,7 @@ namespace NOps {
         }
 
         bool MayProgress() noexcept {
-            return Cache->MayProgress() && ColdPartLoaders.empty();
+            return !IsPaused() && Cache->MayProgress() && ColdPartLoaders.empty();
         }
 
         void Touch(EScan scan) noexcept override
@@ -336,7 +336,9 @@ namespace NOps {
                     return Terminate(EAbort::None);
 
                 case EScan::Sleep:
-                    Y_ABORT("Scan actor got an unexpected EScan::Sleep");
+                    Pause();
+
+                    return Spent->Alter(/* resources not available */ false);
             }
 
             Y_ABORT("Scan actor got an unexpected EScan value");
@@ -521,7 +523,7 @@ namespace NOps {
 
             ContinueInFly = false;
 
-            if (!IsPaused() && MayProgress()) {
+            if (MayProgress()) {
                 React();
             }
         }

ydb/core/tablet_flat/flat_scan_feed.h

@@ -20,6 +20,10 @@ namespace NTable {
 
         }
 
+        void Pause() noexcept {
+            OnPause = true;
+        }
+
         void Resume(EScan op) noexcept
         {
             Y_DEBUG_ABORT_UNLESS(op == EScan::Feed || op == EScan::Reset);

ydb/core/tx/datashard/datashard.cpp

@@ -4393,7 +4393,7 @@ void TDataShard::ScanComplete(NTable::EAbort,
 
         noTxScan->OnFinished(this);
         prod.Destroy();
-    } else {
+    } else if (cookie != 0 && cookie != Max<ui64>()) {
         LOG_DEBUG_S(ctx, NKikimrServices::TX_DATASHARD,
                     "FullScan complete at " << TabletID());
 

ydb/core/tx/datashard/datashard.h

@@ -338,6 +338,10 @@ namespace TEvDataShard {
         EvReshuffleKMeansRequest,
         EvReshuffleKMeansResponse,
 
+        // Sent by the scan actor when EvRead uses a scan
+        EvReadScanStarted,
+        EvReadScanFinished,
+
         EvEnd
     };
 
@@ -924,6 +928,9 @@ namespace TEvDataShard {
                                NKikimrTxDataShard::TEvRead,
                                TEvDataShard::EvRead>;
 
+        static constexpr ui32 HINT_BATCH = NKikimrTxDataShard::TEvRead::HINT_BATCH;
+        static constexpr ui32 HINT_LOW_PRIORITY = NKikimrTxDataShard::TEvRead::HINT_LOW_PRIORITY;
+
         TEvRead() = default;
 
         TString ToString() const override;
@@ -1042,12 +1049,10 @@ namespace TEvDataShard {
     };
 
     struct TEvReadContinue : public TEventLocal<TEvReadContinue, TEvDataShard::EvReadContinue> {
-        TActorId Reader;
-        ui64 ReadId;
+        const ui64 LocalReadId;
 
-        TEvReadContinue(TActorId reader, ui64 readId)
-            : Reader(reader)
-            , ReadId(readId)
+        explicit TEvReadContinue(ui64 localReadId)
+            : LocalReadId(localReadId)
         {}
     };
 
@@ -1063,6 +1068,24 @@ namespace TEvDataShard {
         TEvReadCancel() = default;
     };
 
+    struct TEvReadScanStarted : public TEventLocal<TEvReadScanStarted, EvReadScanStarted> {
+        const ui64 LocalReadId;
+
+        // Event sender is the scan actor
+        explicit TEvReadScanStarted(ui64 localReadId)
+            : LocalReadId(localReadId)
+        {}
+    };
+
+    struct TEvReadScanFinished : public TEventLocal<TEvReadScanFinished, EvReadScanFinished> {
+        const ui64 LocalReadId;
+
+        // Event sender is the scan actor
+        explicit TEvReadScanFinished(ui64 localReadId)
+            : LocalReadId(localReadId)
+        {}
+    };
+
     struct TEvReadColumnsRequest : public TEventPB<TEvReadColumnsRequest,
                                                    NKikimrTxDataShard::TEvReadColumnsRequest,
                                                    TEvDataShard::EvReadColumnsRequest> {