Merge pull request #1120 from ByConity/cherry-pick3

Daily cherry-pick from internal
ByConity · Jan 30, 2024 · 333a3d9 · 333a3d9
2 parents 6c24d54 + 46fee9a
commit 333a3d9
Show file tree

Hide file tree

Showing 4 changed files with 26 additions and 15 deletions.
diff --git a/src/CloudServices/CnchServerServiceImpl.cpp b/src/CloudServices/CnchServerServiceImpl.cpp
@@ -38,13 +38,15 @@
 #include <CloudServices/CnchDataWriter.h>
 #include <CloudServices/DedupWorkerManager.h>
 #include <CloudServices/DedupWorkerStatus.h>
+#include <CloudServices/CnchBGThreadsMap.h>
 #include <Catalog/CatalogUtils.h>
 #include <WorkerTasks/ManipulationType.h>
 #include <Storages/Kafka/CnchKafkaConsumeManager.h>
 #include <Storages/PartCacheManager.h>
 #include <Access/AccessControlManager.h>
 #include <Access/KVAccessStorage.h>
 
+
 #if USE_MYSQL
 #include <Storages/StorageMaterializeMySQL.h>
 #include <Databases/MySQL/MaterializedMySQLSyncThreadManager.h>
@@ -659,7 +661,7 @@ void CnchServerServiceImpl::getBackgroundThreadStatus(
     RPCHelpers::serviceHandler(
         done,
         response,
-        [request = request, response = response, done = done, log = log] {
+        [request = request, response = response, done = done, global_context = getContext(), log = log] {
             brpc::ClosureGuard done_guard(done);
 
             try
@@ -669,10 +671,8 @@ void CnchServerServiceImpl::getBackgroundThreadStatus(
                 auto type = CnchBGThreadType(request->type());
                 if (type >= CnchBGThreadType::ServerMinType && type <= CnchBGThreadType::ServerMaxType)
                 {
-#if 0
-                    auto threads = global_context.getCnchBGThreads(type);
+                    auto threads = global_context->getCnchBGThreadsMap(type);
                     res = threads->getStatusMap();
-#endif
                 }
                 else
                 {
@@ -703,7 +703,7 @@ void CnchServerServiceImpl::getNumBackgroundThreads(
 {
 }
 void CnchServerServiceImpl::controlCnchBGThread(
-    google::protobuf::RpcController * /*cntl*/,
+    google::protobuf::RpcController * cntl,
     const Protos::ControlCnchBGThreadReq * request,
     Protos::ControlCnchBGThreadResp * response,
     google::protobuf::Closure * done)
@@ -712,7 +712,7 @@ void CnchServerServiceImpl::controlCnchBGThread(
     RPCHelpers::serviceHandler(
         done,
         response,
-        [request = request, response = response, done = done, & global_context = *context_ptr, log = log] {
+        [cntl = cntl, request = request, response = response, done = done, & global_context = *context_ptr, log = log] {
             brpc::ClosureGuard done_guard(done);
 
             try
@@ -722,6 +722,10 @@ void CnchServerServiceImpl::controlCnchBGThread(
                     storage_id = RPCHelpers::createStorageID(request->storage_id());
                 auto type = CnchBGThreadType(request->type());
                 auto action = CnchBGThreadAction(request->action());
+                auto & controller = static_cast<brpc::Controller &>(*cntl);
+                LOG_DEBUG(log, "Received controlBGThread for {} type {} action {} from {}",
+                    storage_id.empty() ? "empty storage" : storage_id.getNameForLogs(),
+                    toString(type), toString(action), butil::endpoint2str(controller.remote_side()).c_str());
                 global_context.controlCnchBGThread(storage_id, type, action);
             }
             catch (...)

diff --git a/src/DaemonManager/DaemonJobServerBGThread.cpp b/src/DaemonManager/DaemonJobServerBGThread.cpp
@@ -209,7 +209,10 @@ std::map<String, UInt64> fetchServerStartTimes(Context & context, CnchTopologyMa
         String rpc_address = host_port.getRPCAddress();
         CnchServerClientPtr client_ptr = context.getCnchServerClientPool().get(host_port);
         if (!client_ptr)
+        {
+            LOG_WARNING(log, "Not able to connect to server with rpc address {}", rpc_address);
             continue;
+        }
 
         try
         {
@@ -218,7 +221,7 @@ std::map<String, UInt64> fetchServerStartTimes(Context & context, CnchTopologyMa
         }
         catch (...)
         {
-            LOG_INFO(log, "Failed to reach server: {}", rpc_address);
+            LOG_INFO(log, "Failed to reach server with rpc address: {}", rpc_address);
         }
     }
 
@@ -553,8 +556,12 @@ std::optional<std::unordered_multimap<UUID, BGJobInfoFromServer>> fetchBGThreadF
         {
             CnchServerClientPtr client_ptr = context.getCnchServerClientPool().get(server);
             if (!client_ptr)
+            {
+                LOG_ERROR(log, "Not able to connect to server with address {}", server);
                 return {};
+            }
             auto tasks = client_ptr->getBackGroundStatus(type);
+            LOG_DEBUG(log, "Get {} jobs from server with address {}", tasks.size(), server);
             for (const auto & task : tasks)
             {
                 StorageID storage_id = RPCHelpers::createStorageID(task.storage_id());
@@ -1039,10 +1046,14 @@ BackgroundJobs DaemonJobServerBGThread::fetchCnchBGThreadStatus()
     for (const auto & cnch_server : cnch_servers)
     {
         if (!cnch_server)
+        {
+            LOG_WARNING(log, "Not able to connect to server with address {}", cnch_server->getRPCAddress());
             continue;
+        }
 
         server_start_times[cnch_server->getRPCAddress()] = cnch_server->getServerStartTime();
         auto tasks = cnch_server->getBackGroundStatus(type);
+        LOG_DEBUG(log, "Get {} jobs from server with address {}", tasks.size(), cnch_server->getRPCAddress());
         for (const auto & task : tasks)
         {
             StorageID storage_id = RPCHelpers::createStorageID(task.storage_id());

diff --git a/src/Databases/DatabaseCnch.cpp b/src/Databases/DatabaseCnch.cpp
@@ -279,17 +279,11 @@ void DatabaseCnch::drop(ContextPtr local_context)
         throw Exception("Cnch transaction is not initialized", ErrorCodes::CNCH_TRANSACTION_NOT_INITIALIZED);
 
     // get the lock of tables in current database
-    std::vector<StoragePtr> tables_to_drop;
     std::vector<IntentLockPtr> locks;
 
-    for (auto iterator = getTablesIterator(getContext(), [](const String &) { return true; }); iterator->isValid(); iterator->next())
+    for (auto iterator = getTablesIteratorLightweight(getContext(), [](const String &) { return true; }); iterator->isValid(); iterator->next())
     {
-        StoragePtr table = iterator->table();
-        if (!table)
-            continue;
-        const auto & storage_id = table->getStorageID();
-        locks.emplace_back(txn->createIntentLock(IntentLock::TB_LOCK_PREFIX, storage_id.database_name, storage_id.table_name));
-        tables_to_drop.emplace_back(table);
+        locks.emplace_back(txn->createIntentLock(IntentLock::TB_LOCK_PREFIX, database_name, iterator->name()));
     }
 
     for (const auto & lock : locks)

diff --git a/src/Interpreters/Context.cpp b/src/Interpreters/Context.cpp
@@ -4870,6 +4870,8 @@ UInt64 Context::tryGetTimestamp(const String & pretty_func_name) const
     }
     catch (...)
     {
+        if (!getConfigRef().getBool("tso_service.use_fallback", true))
+            throw;
         tryLogCurrentException(
             pretty_func_name.c_str(), fmt::format("Unable to reach TSO from {} during call to tryGetTimestamp", tryGetTSOLeaderHostPort()));
         return TxnTimestamp::fallbackTS();