[#29343] YSQL: Make sure each batch has enough of PgsqlOps

Summary:
We could not reproduce the issue reported by #29343 in a controlled
environment, however after core dump analysis we figured out that while
making batches to fetch data from the base table by ybctids the
PgDocReadOp instance did not have sufficient number PgsqlOps/ReadReqs.
There was one PgsqlOp/ReadReq while the table had 2 tablets.

In 8fa242d where the issue was
allegedly introduced we changed the logic of PgsqlOps/ReadReqs
repopulation after cleanup between the batches. We create as many
PgsqlOps/ReadReqs as there were in the previous batch, assuming that the
number of the PgsqlOps/ReadReqs needed does not change between the
batches. Apparently that assumption was incorrect, so here we
recalculate the number for each batch and make sure we have that number
of the PgsqlOps/ReadReqs created.

Test Plan:
Make sure there is no regression in Jenkins
Since the issue was only reproduced in perf environment, rerun perf tests on the diff

Reviewers: jason

Reviewed By: jason

Subscribers: yql

Tags: #jenkins-ready

Differential Revision: https://phorge.dev.yugabyte.com/D48246

Author: andrei-mart

src/yb/yql/pggate/pg_doc_op.cc

@@ -1235,11 +1235,9 @@ void PgDocReadOp::InitializeYbctidOperators() {
     DCHECK(pgsql_ops_.empty());
     pgsql_op_arena_ = SharedThreadSafeArena();
   }
-  if (pgsql_ops_.empty()) {
-    ClonePgsqlOps(table_->GetPartitionListSize());
-  } else {
-    ResetInactivePgsqlOps();
-  }
+  ResetInactivePgsqlOps();
+  // Make sure we have one operation per partition
+  ClonePgsqlOps(table_->GetPartitionListSize());
 }
 
 Result<bool> PgDocReadOp::BindExprsRegular(
@@ -1346,27 +1344,27 @@ Result<bool> PgDocReadOp::PopulateNextHashPermutationOps() {
   // Permutations and operations are reset between parallel ranges, in this case we need to clone
   // a new set of operations.
   if (!hash_permutations_) {
+    DCHECK(pgsql_ops_.empty());
     hash_permutations_.emplace(InitializeHashPermutationStates());
+    // Setup arena to control memory footprint
+    if (!pgsql_op_arena_) {
+      pgsql_op_arena_ = SharedThreadSafeArena();
+    }
   } else {
     ResetInactivePgsqlOps();
   }
+
   // Continue with operations if any are still in flight
   if (active_op_count_ > 0) {
     return false;
   }
 
-  // Setup arena to control memory footprint
-  if (!pgsql_op_arena_) {
-    pgsql_op_arena_ = SharedThreadSafeArena();
-  }
   // Setup operations
-  if (pgsql_ops_.empty()) {
-    auto max_op_count = std::min(hash_permutations_->Size(),
-                                 IsHashBatchingEnabled() ?
-                                     table_->GetPartitionListSize() :
-                                     implicit_cast<size_t>(FLAGS_ysql_request_limit));
-    ClonePgsqlOps(max_op_count);
-  }
+  auto max_op_count = std::min(hash_permutations_->Size(),
+                               IsHashBatchingEnabled() ?
+                                   table_->GetPartitionListSize() :
+                                   implicit_cast<size_t>(FLAGS_ysql_request_limit));
+  ClonePgsqlOps(max_op_count);
   if (IsHashBatchingEnabled()) {
     const auto batch_count = table_->GetPartitionList();
     std::vector<std::pair<bool, LWPgsqlExpressionPB*>> partition_batches(
@@ -1650,15 +1648,12 @@ void PgDocReadOp::SetReadTimeForBackfill() {
 
 void PgDocReadOp::ResetInactivePgsqlOps() {
   auto op_count = pgsql_ops_.size();
-  if (pgsql_op_arena_ && active_op_count_ == 0) {
+  if (pgsql_op_arena_ && active_op_count_ == 0 && op_count > 0) {
     // All past operations are done, can perform full reset to release memory
-    if (op_count > 0) {
-      VLOG_WITH_FUNC(3) << "do full reset";
-      ResetResultStream();
-      pgsql_ops_.clear();
-      pgsql_op_arena_->Reset(ResetMode::kKeepLast);
-      ClonePgsqlOps(op_count);
-    }
+    VLOG_WITH_FUNC(3) << "do full reset";
+    ResetResultStream();
+    pgsql_ops_.clear();
+    pgsql_op_arena_->Reset(ResetMode::kKeepLast);
     return;
   }
   // Clear the existing requests.