Fix overflow case in stream index lookup join (#11818) (#11823)

Author: gridnevvvit

ydb/core/kqp/runtime/kqp_stream_lookup_worker.cpp

@@ -243,7 +243,7 @@ class TKqpLookupRows : public TKqpStreamLookupWorker {
         }
     }
 
-    std::vector<THolder<TEvDataShard::TEvRead>> RebuildRequest(const ui64& prevReadId, ui32 firstUnprocessedQuery, 
+    std::vector<THolder<TEvDataShard::TEvRead>> RebuildRequest(const ui64& prevReadId, ui32 firstUnprocessedQuery,
         TMaybe<TOwnedCellVec> lastProcessedKey, ui64& newReadId) final {
 
         auto it = PendingKeysByReadId.find(prevReadId);
@@ -292,7 +292,7 @@ class TKqpLookupRows : public TKqpStreamLookupWorker {
             requests.emplace_back(std::move(request));
             PendingKeysByReadId.insert({newReadId, std::move(unprocessedRanges)});
         }
-        
+
         return requests;
     }
 
@@ -507,7 +507,7 @@ class TKqpJoinRows : public TKqpStreamLookupWorker {
         UnprocessedRows.emplace_back(std::make_pair(TOwnedCellVec(joinKeyCells), std::move(inputRow.GetElement(1))));
     }
 
-    std::vector<THolder<TEvDataShard::TEvRead>> RebuildRequest(const ui64& prevReadId, ui32 firstUnprocessedQuery, 
+    std::vector<THolder<TEvDataShard::TEvRead>> RebuildRequest(const ui64& prevReadId, ui32 firstUnprocessedQuery,
         TMaybe<TOwnedCellVec> lastProcessedKey, ui64& newReadId) final {
 
         auto readIt = PendingKeysByReadId.find(prevReadId);
@@ -946,6 +946,7 @@ class TKqpJoinRows : public TKqpStreamLookupWorker {
         if (rowStats.ResultBytesCount > (ui64)freeSpace) {
             resultRow.DeleteUnreferenced();
             rowStats.Clear();
+            return NUdf::TUnboxedValuePod();
         }
 
         return resultRow;