YQ-1095: Force cleanup data

ref:4d9f0447a6fd2aedfbc284356ff373e17e26ee8a

1 files changed, 7 insertions, 2 deletions

diff --git a/ydb/library/yql/dq/actors/compute/dq_async_compute_actor.cpp b/ydb/library/yql/dq/actors/compute/dq_async_compute_actor.cpp
index 9982e93ee3c..e72a299a054 100644
--- a/ydb/library/yql/dq/actors/compute/dq_async_compute_actor.cpp
+++ b/ydb/library/yql/dq/actors/compute/dq_async_compute_actor.cpp
@@ -493,8 +493,13 @@ private:
         ProcessOutputsState.Inflight --;
         ProcessOutputsState.HasDataToSend |= !sinkInfo.Finished;
 
-        auto guard = BindAllocator();
-        sinkInfo.AsyncOutput->SendData(std::move(batch), size, std::move(checkpoint), finished);
+        {
+            auto guard = BindAllocator();
+            NKikimr::NMiniKQL::TUnboxedValueVector data = std::move(batch);
+            sinkInfo.AsyncOutput->SendData(std::move(data), size, std::move(checkpoint), finished);
+        }
+
+        Y_VERIFY(batch.empty());
         CA_LOG_D("sink " << outputIndex << ": sent " << dataSize << " bytes of data and " << checkpointSize << " bytes of checkpoint barrier");
 
         CA_LOG_D("Drain sink " << outputIndex