[25.2] Sequential reads for BATCH queries (#28906) (#28924)

Author: dahbka-lis

ydb/core/kqp/runtime/kqp_read_actor.cpp

@@ -455,7 +455,7 @@ class TKqpReadActor : public TActorBootstrapped<TKqpReadActor>, public NYql::NDq
     }
 
     bool StartShards() {
-        const ui32 maxAllowedInFlight = Settings->GetSorted() ? 1 : MaxInFlight;
+        const ui32 maxAllowedInFlight = Settings->GetSorted() || Settings->GetIsBatch() ? 1 : MaxInFlight;
         CA_LOG_D("effective maxinflight " << maxAllowedInFlight << " sorted " << Settings->GetSorted());
         bool isFirst = true;
         while (!PendingShards.Empty() && RunningReads() + 1 <= maxAllowedInFlight) {
@@ -1052,7 +1052,7 @@ class TKqpReadActor : public TActorBootstrapped<TKqpReadActor>, public NYql::NDq
         ui64 seqNo = record.GetSeqNo();
         Reads[id].RegisterMessage(msg);
 
-        if (Settings->GetIsBatch() && msg.GetRowsCount() > 0) {
+        if (Settings->GetIsBatch() && msg.GetRowsCount() > 0 && BatchOperationMaxRow.GetCells().empty()) {
             auto cells = msg.GetCells(msg.GetRowsCount() - 1);
             BatchOperationMaxRow = TSerializedCellVec{cells};
         }