Raft: Increase max retry times to avoid too large remote requests (#10301) (#10307)

close #10300

Raft: Increase max retry times to avoid too large remote requests
  * Increase the max retry number between LearnerRead and acquiring snapshot from the storage layer by the number of query regions

Signed-off-by: ti-chi-bot <[email protected]>
Signed-off-by: JaySon-Huang <[email protected]>

Co-authored-by: JaySon <[email protected]>
Co-authored-by: JaySon-Huang <[email protected]>

Author: ti-chi-bot

dbms/src/Flash/Coprocessor/DAGStorageInterpreter.cpp

@@ -945,15 +945,12 @@ bool DAGStorageInterpreter::checkRetriableForBatchCopOrMPP(
     const TableID & table_id,
     const SelectQueryInfo & query_info,
     const RegionException & e,
-    int num_allow_retry)
+    const Int32 num_allow_retry)
 {
     const DAGContext & dag_context = *context.getDAGContext();
     assert((dag_context.isBatchCop() || dag_context.isMPPTask()));
     const auto & dag_regions = dag_context.getTableRegionsInfoByTableID(table_id).local_regions;
     FmtBuffer buffer;
-    // Normally there is only few regions need to retry when super batch is enabled. Retry to read
-    // from local first. However, too many retry in different places may make the whole process
-    // time out of control. We limit the number of retries to 1 now.
     if (likely(num_allow_retry > 0))
     {
         auto & regions_query_info = query_info.mvcc_query_info->regions_query_info;
@@ -967,16 +964,22 @@ bool DAGStorageInterpreter::checkRetriableForBatchCopOrMPP(
                     region_retry_from_local_region.emplace_back(region_iter->second);
                     buffer.fmtAppend("{},", region_iter->first);
                 }
+                // remove the unavailable region for next local read attempt
                 iter = regions_query_info.erase(iter);
             }
             else
             {
                 ++iter;
             }
         }
+        // `tot_num_remote_region` is the total number of regions that we will retry from other tiflash nodes among all retries
+        // `current_retry_regions` is the number of regions that we will retry from other tiflash nodes in this retry
         LOG_WARNING(
             log,
-            "RegionException after read from storage, regions [{}], message: {}{}",
+            "RegionException after read from storage, tot_num_remote_region={} cur_retry_regions={}"
+            " regions [{}], message: {}{}",
+            region_retry_from_local_region.size(),
+            e.unavailable_region.size(),
             buffer.toString(),
             e.message(),
             (regions_query_info.empty() ? "" : ", retry to read from local"));
@@ -996,15 +999,44 @@ bool DAGStorageInterpreter::checkRetriableForBatchCopOrMPP(
                 buffer.fmtAppend("{},", iter->first);
             }
         }
+        // `tot_num_remote_region` is the total number of regions that we will retry from other tiflash nodes among all retries
+        // `current_retry_regions` is the number of regions that we will retry from other tiflash nodes in this retry
         LOG_WARNING(
             log,
-            "RegionException after read from storage, regions [{}], message: {}",
+            "RegionException after read from storage, tot_num_remote_region={} cur_retry_regions={}"
+            " regions [{}], message: {}",
+            region_retry_from_local_region.size(),
+            e.unavailable_region.size(),
             buffer.toString(),
             e.message());
         return false; // break retry loop
     }
 }
 
+namespace
+{
+Int32 getMaxAllowRetryForLocalRead(const SelectQueryInfo & query_info)
+{
+    size_t region_num = query_info.mvcc_query_info->regions_query_info.size();
+    if (region_num > 1000)
+    {
+        // 1000 regions is about 93GB for 96MB region size / 250GB for 256MB region size.
+        return 10;
+    }
+    else if (region_num > 500)
+    {
+        // 500 regions is about 46.5GB for 96MB region size / 125GB for 256MB region size.
+        return 8;
+    }
+    else if (region_num > 100)
+    {
+        // 100 regions is about 9.3GB for 96MB region size / 25GB for 256MB region size.
+        return 5;
+    }
+    return 1;
+}
+} // namespace
+
 DM::Remote::DisaggPhysicalTableReadSnapshotPtr DAGStorageInterpreter::buildLocalStreamsForPhysicalTable(
     const TableID & table_id,
     const SelectQueryInfo & query_info,
@@ -1021,7 +1053,14 @@ DM::Remote::DisaggPhysicalTableReadSnapshotPtr DAGStorageInterpreter::buildLocal
 
     const DAGContext & dag_context = *context.getDAGContext();
     const auto keyspace_id = dag_context.getKeyspaceID();
-    for (int num_allow_retry = 1; num_allow_retry >= 0; --num_allow_retry)
+    // Normally there is only few regions need to retry when super batch is enabled. Retry to read
+    // from local first.
+    // When the table is large and too hot for writing, the number of regions may be large
+    // and region split is frequent. In this case, we allow more retries for building
+    // inputstream from local in order to avoid large number of RemoteRead requests.
+    // However, too many retry may make the whole execution time out of control.
+    Int32 num_allow_retry = getMaxAllowRetryForLocalRead(query_info);
+    for (; num_allow_retry >= 0; --num_allow_retry)
     {
         try
         {
@@ -1063,7 +1102,7 @@ DM::Remote::DisaggPhysicalTableReadSnapshotPtr DAGStorageInterpreter::buildLocal
                 // clean all streams from local because we are not sure the correctness of those streams
                 pipeline.streams.clear();
                 if (likely(checkRetriableForBatchCopOrMPP(table_id, query_info, e, num_allow_retry)))
-                    continue;
+                    continue; // next retry to read from local storage
                 else
                     break;
             }

dbms/src/Storages/DeltaMerge/Delta/ColumnFilePersistedSet.cpp

@@ -349,7 +349,19 @@ bool ColumnFilePersistedSet::installCompactionResults(const MinorCompactionPtr &
                     || (file->getId() != (*old_persisted_files_iter)->getId())
                     || (file->getRows() != (*old_persisted_files_iter)->getRows())))
             {
-                throw Exception("Compaction algorithm broken", ErrorCodes::LOGICAL_ERROR);
+                throw Exception(
+                    ErrorCodes::LOGICAL_ERROR,
+                    "Compaction algorithm broken, "
+                    "compaction={{{}}} persisted_files={} "
+                    "old_persisted_files_iter.is_end={} "
+                    "file->getId={} old_persist_files->getId={} file->getRows={} old_persist_files->getRows={}",
+                    compaction->info(),
+                    detailInfo(),
+                    old_persisted_files_iter == persisted_files.end(),
+                    file->getId(),
+                    old_persisted_files_iter == persisted_files.end() ? -1 : (*old_persisted_files_iter)->getId(),
+                    file->getRows(),
+                    old_persisted_files_iter == persisted_files.end() ? -1 : (*old_persisted_files_iter)->getRows());
             }
             old_persisted_files_iter++;
         }