From 6c2a0c8737798c5c178cccaa06edd77e9f302d2e Mon Sep 17 00:00:00 2001 From: dapan1121 Date: Thu, 27 Oct 2022 15:15:50 +0800 Subject: [PATCH] fix: fix catalog ut issues and add query hb configuration --- include/common/tglobal.h | 1 + source/common/src/tglobal.c | 5 ++ source/libs/catalog/src/ctgCache.c | 63 ++++++++++++++++++++-- source/libs/catalog/test/CMakeLists.txt | 8 +-- source/libs/catalog/test/catalogTests.cpp | 66 +++++++++++++++-------- source/libs/scheduler/src/schTask.c | 2 +- source/libs/scheduler/src/schUtil.c | 5 ++ 7 files changed, 119 insertions(+), 31 deletions(-) diff --git a/include/common/tglobal.h b/include/common/tglobal.h index 4a2e14eb78..40f73eb06c 100644 --- a/include/common/tglobal.h +++ b/include/common/tglobal.h @@ -91,6 +91,7 @@ extern bool tsQueryPlannerTrace; extern int32_t tsQueryNodeChunkSize; extern bool tsQueryUseNodeAllocator; extern bool tsKeepColumnName; +extern bool tsEnableQueryHb; // client extern int32_t tsMinSlidingTime; diff --git a/source/common/src/tglobal.c b/source/common/src/tglobal.c index 5025cbcac5..f98f81f8c5 100644 --- a/source/common/src/tglobal.c +++ b/source/common/src/tglobal.c @@ -82,6 +82,7 @@ bool tsSmlDataFormat = false; // query int32_t tsQueryPolicy = 1; int32_t tsQueryRspPolicy = 0; +bool tsEnableQueryHb = false; int32_t tsQuerySmaOptimize = 0; int32_t tsQueryRsmaTolerance = 1000; // the tolerance time (ms) to judge from which level to query rsma data. bool tsQueryPlannerTrace = false; @@ -284,6 +285,7 @@ static int32_t taosAddClientCfg(SConfig *pCfg) { if (cfgAddInt32(pCfg, "compressMsgSize", tsCompressMsgSize, -1, 100000000, 1) != 0) return -1; if (cfgAddInt32(pCfg, "compressColData", tsCompressColData, -1, 100000000, 1) != 0) return -1; if (cfgAddInt32(pCfg, "queryPolicy", tsQueryPolicy, 1, 4, 1) != 0) return -1; + if (cfgAddBool(pCfg, "enableQueryHb", tsEnableQueryHb, false) != 0) return -1; if (cfgAddInt32(pCfg, "querySmaOptimize", tsQuerySmaOptimize, 0, 1, 1) != 0) return -1; if (cfgAddBool(pCfg, "queryPlannerTrace", tsQueryPlannerTrace, true) != 0) return -1; if (cfgAddInt32(pCfg, "queryNodeChunkSize", tsQueryNodeChunkSize, 1024, 128 * 1024, true) != 0) return -1; @@ -644,6 +646,7 @@ static int32_t taosSetClientCfg(SConfig *pCfg) { tsCompressColData = cfgGetItem(pCfg, "compressColData")->i32; tsNumOfTaskQueueThreads = cfgGetItem(pCfg, "numOfTaskQueueThreads")->i32; tsQueryPolicy = cfgGetItem(pCfg, "queryPolicy")->i32; + tsEnableQueryHb = cfgGetItem(pCfg, "enableQueryHb")->bval; tsQuerySmaOptimize = cfgGetItem(pCfg, "querySmaOptimize")->i32; tsQueryPlannerTrace = cfgGetItem(pCfg, "queryPlannerTrace")->bval; tsQueryNodeChunkSize = cfgGetItem(pCfg, "queryNodeChunkSize")->i32; @@ -780,6 +783,8 @@ int32_t taosSetCfg(SConfig *pCfg, char *name) { if (strcasecmp("enableCoreFile", name) == 0) { bool enableCore = cfgGetItem(pCfg, "enableCoreFile")->bval; taosSetCoreDump(enableCore); + } else if (strcasecmp("enableQueryHb", name) == 0) { + tsEnableQueryHb = cfgGetItem(pCfg, "enableQueryHb")->bval; } break; } diff --git a/source/libs/catalog/src/ctgCache.c b/source/libs/catalog/src/ctgCache.c index ca562aee3e..61da0724c5 100644 --- a/source/libs/catalog/src/ctgCache.c +++ b/source/libs/catalog/src/ctgCache.c @@ -621,14 +621,18 @@ int32_t ctgEnqueue(SCatalog *pCtg, SCtgCacheOperation *operation) { node->op = operation; CTG_LOCK(CTG_WRITE, &gCtgMgmt.queue.qlock); + if (gCtgMgmt.queue.stopQueue) { ctgFreeQNode(node); CTG_UNLOCK(CTG_WRITE, &gCtgMgmt.queue.qlock); CTG_RET(TSDB_CODE_CTG_EXIT); } - gCtgMgmt.queue.stopQueue = operation->stopQueue; + gCtgMgmt.queue.tail->next = node; gCtgMgmt.queue.tail = node; + + gCtgMgmt.queue.stopQueue = operation->stopQueue; + CTG_UNLOCK(CTG_WRITE, &gCtgMgmt.queue.qlock); ctgDebug("action [%s] added into queue", opName); @@ -1997,6 +2001,59 @@ _return: CTG_RET(code); } +void ctgFreeCacheOperationData(SCtgCacheOperation *op) { + if (NULL == op || NULL == op->data) { + return; + } + + switch (op->opId) { + case CTG_OP_UPDATE_VGROUP: { + SCtgUpdateVgMsg *msg = op->data; + ctgFreeVgInfo(msg->dbInfo); + taosMemoryFreeClear(op->data); + break; + } + case CTG_OP_UPDATE_TB_META: { + SCtgUpdateTbMetaMsg *msg = op->data; + taosMemoryFreeClear(msg->pMeta->tbMeta); + taosMemoryFreeClear(msg->pMeta); + taosMemoryFreeClear(op->data); + break; + } + case CTG_OP_DROP_DB_CACHE: + case CTG_OP_DROP_DB_VGROUP: + case CTG_OP_DROP_STB_META: + case CTG_OP_DROP_TB_META: + case CTG_OP_UPDATE_VG_EPSET: + case CTG_OP_DROP_TB_INDEX: + case CTG_OP_CLEAR_CACHE: { + taosMemoryFreeClear(op->data); + break; + } + case CTG_OP_UPDATE_USER: { + SCtgUpdateUserMsg *msg = op->data; + taosHashCleanup(msg->userAuth.createdDbs); + taosHashCleanup(msg->userAuth.readDbs); + taosHashCleanup(msg->userAuth.writeDbs); + taosMemoryFreeClear(op->data); + break; + } + case CTG_OP_UPDATE_TB_INDEX: { + SCtgUpdateTbIndexMsg *msg = op->data; + if (msg->pIndex) { + taosArrayDestroyEx(msg->pIndex->pIndex, tFreeSTableIndexInfo); + taosMemoryFreeClear(msg->pIndex); + } + taosMemoryFreeClear(op->data); + break; + } + default: { + qError("invalid cache op id:%d", op->opId); + break; + } + } +} + void ctgCleanupCacheQueue(void) { SCtgQNode *node = NULL; SCtgQNode *nodeNext = NULL; @@ -2015,7 +2072,7 @@ void ctgCleanupCacheQueue(void) { stopQueue = true; CTG_RT_STAT_INC(numOfOpDequeue, 1); } else { - taosMemoryFree(op->data); + ctgFreeCacheOperationData(op); CTG_RT_STAT_INC(numOfOpAbort, 1); } @@ -2053,7 +2110,7 @@ void *ctgUpdateThreadFunc(void *param) { qError("ctg tsem_wait failed, error:%s", tstrerror(TAOS_SYSTEM_ERROR(errno))); } - if (atomic_load_8((int8_t *)&gCtgMgmt.exit)) { + if (atomic_load_8((int8_t *)&gCtgMgmt.queue.stopQueue)) { ctgCleanupCacheQueue(); break; } diff --git a/source/libs/catalog/test/CMakeLists.txt b/source/libs/catalog/test/CMakeLists.txt index b8d50b9c63..dbbb24ce0b 100644 --- a/source/libs/catalog/test/CMakeLists.txt +++ b/source/libs/catalog/test/CMakeLists.txt @@ -18,8 +18,8 @@ IF(NOT TD_DARWIN) PRIVATE "${TD_SOURCE_DIR}/source/libs/catalog/inc" ) - #add_test( - # NAME catalogTest - # COMMAND catalogTest - #) + add_test( + NAME catalogTest + COMMAND catalogTest + ) ENDIF() diff --git a/source/libs/catalog/test/catalogTests.cpp b/source/libs/catalog/test/catalogTests.cpp index 4a5f090d18..864d1bbf04 100644 --- a/source/libs/catalog/test/catalogTests.cpp +++ b/source/libs/catalog/test/catalogTests.cpp @@ -27,8 +27,8 @@ #ifdef WINDOWS #define TD_USE_WINSOCK #endif -#include "catalog.h" #include "catalogInt.h" +#include "catalog.h" #include "os.h" #include "stub.h" #include "taos.h" @@ -49,8 +49,6 @@ void ctgTestSetRspCTableMeta(); void ctgTestSetRspSTableMeta(); void ctgTestSetRspMultiSTableMeta(); -// extern "C" SCatalogMgmt gCtgMgmt; - enum { CTGT_RSP_VGINFO = 1, CTGT_RSP_TBMETA, @@ -285,6 +283,8 @@ void ctgTestBuildSTableMetaRsp(STableMetaRsp *rspMsg) { } void ctgTestRspDbVgroups(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp) { + rpcFreeCont(pMsg->pCont); + SUseDbRsp usedbRsp = {0}; strcpy(usedbRsp.db, ctgTestDbname); usedbRsp.vgVersion = ctgTestVgVersion; @@ -322,9 +322,13 @@ void ctgTestRspDbVgroups(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg * pRsp->code = 0; pRsp->contLen = contLen; pRsp->pCont = pReq; + + taosArrayDestroy(usedbRsp.pVgroupInfos); } void ctgTestRspTableMeta(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp) { + rpcFreeCont(pMsg->pCont); + STableMetaRsp metaRsp = {0}; strcpy(metaRsp.dbFName, ctgTestDbname); strcpy(metaRsp.tbName, ctgTestTablename); @@ -364,10 +368,14 @@ void ctgTestRspTableMeta(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg * } void ctgTestRspTableMetaNotExist(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp) { + rpcFreeCont(pMsg->pCont); + pRsp->code = CTG_ERR_CODE_TABLE_NOT_EXIST; } void ctgTestRspCTableMeta(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp) { + rpcFreeCont(pMsg->pCont); + STableMetaRsp metaRsp = {0}; strcpy(metaRsp.dbFName, ctgTestDbname); strcpy(metaRsp.tbName, ctgTestCurrentCTableName ? ctgTestCurrentCTableName : ctgTestCTablename); @@ -414,6 +422,8 @@ void ctgTestRspCTableMeta(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg } void ctgTestRspSTableMeta(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp) { + rpcFreeCont(pMsg->pCont); + STableMetaRsp metaRsp = {0}; strcpy(metaRsp.dbFName, ctgTestDbname); strcpy(metaRsp.tbName, ctgTestCurrentSTableName ? ctgTestCurrentSTableName : ctgTestSTablename); @@ -460,6 +470,8 @@ void ctgTestRspSTableMeta(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg } void ctgTestRspMultiSTableMeta(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp) { + rpcFreeCont(pMsg->pCont); + static int32_t idx = 1; STableMetaRsp metaRsp = {0}; @@ -511,6 +523,8 @@ void ctgTestRspMultiSTableMeta(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRp void ctgTestRspErrIndexInfo(void *shandle, SEpSet *pEpSet, SRpcMsg *pMsg, SRpcMsg *pRsp) { + rpcFreeCont(pMsg->pCont); + pRsp->code = TSDB_CODE_MND_DB_INDEX_NOT_EXIST; pRsp->contLen = 0; pRsp->pCont = NULL; @@ -981,6 +995,8 @@ TEST(tableMeta, normalTable) { ASSERT_EQ(tableMeta->tableInfo.precision, 1); ASSERT_EQ(tableMeta->tableInfo.rowSize, 12); + taosMemoryFree(tableMeta); + while (true) { uint32_t n = ctgdGetClusterCacheNum(pCtg, CTG_DBG_META_NUM); if (0 == n) { @@ -1002,6 +1018,8 @@ TEST(tableMeta, normalTable) { ASSERT_EQ(tableMeta->tableInfo.precision, 1); ASSERT_EQ(tableMeta->tableInfo.rowSize, 12); + taosMemoryFree(tableMeta); + SDbVgVersion *dbs = NULL; SSTableVersion *stb = NULL; uint32_t dbNum = 0, stbNum = 0, allDbNum = 0, allStbNum = 0; @@ -1038,7 +1056,6 @@ TEST(tableMeta, normalTable) { ASSERT_EQ(allStbNum, 0); catalogDestroy(); - memset(&gCtgMgmt, 0, sizeof(gCtgMgmt)); } TEST(tableMeta, childTableCase) { @@ -1076,6 +1093,8 @@ TEST(tableMeta, childTableCase) { ASSERT_EQ(tableMeta->tableInfo.precision, 1); ASSERT_EQ(tableMeta->tableInfo.rowSize, 12); + taosMemoryFree(tableMeta); + while (true) { uint32_t n = ctgdGetClusterCacheNum(pCtg, CTG_DBG_META_NUM); if (0 == n) { @@ -1111,6 +1130,8 @@ TEST(tableMeta, childTableCase) { ASSERT_EQ(tableMeta->tableInfo.precision, 1); ASSERT_EQ(tableMeta->tableInfo.rowSize, 12); + taosMemoryFree(tableMeta); + SDbVgVersion *dbs = NULL; SSTableVersion *stb = NULL; uint32_t dbNum = 0, stbNum = 0, allDbNum = 0, allStbNum = 0; @@ -1147,7 +1168,6 @@ TEST(tableMeta, childTableCase) { ASSERT_EQ(allStbNum, 1); catalogDestroy(); - memset(&gCtgMgmt, 0, sizeof(gCtgMgmt)); } TEST(tableMeta, superTableCase) { @@ -1185,6 +1205,8 @@ TEST(tableMeta, superTableCase) { ASSERT_EQ(tableMeta->tableInfo.precision, 1); ASSERT_EQ(tableMeta->tableInfo.rowSize, 12); + taosMemoryFree(tableMeta); + while (true) { uint32_t n = ctgdGetClusterCacheNum(pCtg, CTG_DBG_META_NUM); if (0 == n) { @@ -1211,6 +1233,8 @@ TEST(tableMeta, superTableCase) { ASSERT_EQ(tableMeta->tableInfo.precision, 1); ASSERT_EQ(tableMeta->tableInfo.rowSize, 12); + taosMemoryFree(tableMeta); + while (true) { uint32_t n = ctgdGetClusterCacheNum(pCtg, CTG_DBG_META_NUM); if (2 != n) { @@ -1232,6 +1256,8 @@ TEST(tableMeta, superTableCase) { ASSERT_EQ(tableMeta->tableInfo.precision, 1); ASSERT_EQ(tableMeta->tableInfo.rowSize, 12); + taosMemoryFree(tableMeta); + SDbVgVersion *dbs = NULL; SSTableVersion *stb = NULL; uint32_t dbNum = 0, stbNum = 0, allDbNum = 0, allStbNum = 0; @@ -1269,7 +1295,6 @@ TEST(tableMeta, superTableCase) { ASSERT_EQ(allStbNum, 1); catalogDestroy(); - memset(&gCtgMgmt, 0, sizeof(gCtgMgmt)); } TEST(tableMeta, rmStbMeta) { @@ -1309,6 +1334,8 @@ TEST(tableMeta, rmStbMeta) { ASSERT_EQ(tableMeta->tableInfo.precision, 1); ASSERT_EQ(tableMeta->tableInfo.rowSize, 12); + taosMemoryFree(tableMeta); + while (true) { uint32_t n = ctgdGetClusterCacheNum(pCtg, CTG_DBG_META_NUM); if (0 == n) { @@ -1338,7 +1365,6 @@ TEST(tableMeta, rmStbMeta) { ASSERT_EQ(ctgdGetClusterCacheNum(pCtg, CTG_DBG_STB_RENT_NUM), 0); catalogDestroy(); - memset(&gCtgMgmt, 0, sizeof(gCtgMgmt)); } TEST(tableMeta, updateStbMeta) { @@ -1428,12 +1454,12 @@ TEST(tableMeta, updateStbMeta) { taosMemoryFreeClear(tableMeta); catalogDestroy(); - memset(&gCtgMgmt.stat, 0, sizeof(gCtgMgmt.stat)); } TEST(getIndexInfo, notExists) { struct SCatalog *pCtg = NULL; - SRequestConnInfo *mockPointer = (SRequestConnInfo *)0x1; + SRequestConnInfo connInfo = {0}; + SRequestConnInfo *mockPointer = (SRequestConnInfo *)&connInfo; SVgroupInfo vgInfo = {0}; SArray *vgList = NULL; @@ -1456,6 +1482,8 @@ TEST(getIndexInfo, notExists) { SIndexInfo info; code = catalogGetIndexMeta(pCtg, mockPointer, "index1", &info); ASSERT_TRUE(code != 0); + + catalogDestroy(); } TEST(refreshGetMeta, normal2normal) { @@ -1535,7 +1563,6 @@ TEST(refreshGetMeta, normal2normal) { taosMemoryFreeClear(tableMeta); catalogDestroy(); - memset(&gCtgMgmt, 0, sizeof(gCtgMgmt)); } TEST(refreshGetMeta, normal2notexist) { @@ -1606,7 +1633,6 @@ TEST(refreshGetMeta, normal2notexist) { ASSERT_TRUE(tableMeta == NULL); catalogDestroy(); - memset(&gCtgMgmt, 0, sizeof(gCtgMgmt)); } TEST(refreshGetMeta, normal2child) { @@ -1688,7 +1714,6 @@ TEST(refreshGetMeta, normal2child) { taosMemoryFreeClear(tableMeta); catalogDestroy(); - memset(&gCtgMgmt, 0, sizeof(gCtgMgmt)); ctgTestCurrentCTableName = NULL; ctgTestCurrentSTableName = NULL; } @@ -1776,7 +1801,6 @@ TEST(refreshGetMeta, stable2child) { taosMemoryFreeClear(tableMeta); catalogDestroy(); - memset(&gCtgMgmt, 0, sizeof(gCtgMgmt)); ctgTestCurrentCTableName = NULL; ctgTestCurrentSTableName = NULL; } @@ -1863,7 +1887,6 @@ TEST(refreshGetMeta, stable2stable) { taosMemoryFreeClear(tableMeta); catalogDestroy(); - memset(&gCtgMgmt, 0, sizeof(gCtgMgmt)); ctgTestCurrentCTableName = NULL; ctgTestCurrentSTableName = NULL; } @@ -1950,7 +1973,6 @@ TEST(refreshGetMeta, child2stable) { taosMemoryFreeClear(tableMeta); catalogDestroy(); - memset(&gCtgMgmt, 0, sizeof(gCtgMgmt)); ctgTestCurrentCTableName = NULL; ctgTestCurrentSTableName = NULL; } @@ -1990,7 +2012,6 @@ TEST(tableDistVgroup, normalTable) { ASSERT_TRUE(code != 0); catalogDestroy(); - memset(&gCtgMgmt, 0, sizeof(gCtgMgmt)); } TEST(tableDistVgroup, childTableCase) { @@ -2029,7 +2050,6 @@ TEST(tableDistVgroup, childTableCase) { ASSERT_TRUE(code != 0); catalogDestroy(); - memset(&gCtgMgmt, 0, sizeof(gCtgMgmt)); } TEST(tableDistVgroup, superTableCase) { @@ -2076,8 +2096,9 @@ TEST(tableDistVgroup, superTableCase) { ASSERT_EQ(vgInfo->vgId, 3); ASSERT_EQ(vgInfo->epSet.numOfEps, 3); + taosArrayDestroy(vgList); + catalogDestroy(); - memset(&gCtgMgmt, 0, sizeof(gCtgMgmt)); } TEST(dbVgroup, getSetDbVgroupCase) { @@ -2116,6 +2137,8 @@ TEST(dbVgroup, getSetDbVgroupCase) { ASSERT_EQ(code, 0); ASSERT_EQ(taosArrayGetSize((const SArray *)vgList), ctgTestVgNum); + taosArrayDestroy(vgList); + while (true) { uint64_t n = 0; ctgdGetStatNum("runtime.numOfOpDequeue", (void *)&n); @@ -2156,7 +2179,6 @@ TEST(dbVgroup, getSetDbVgroupCase) { ASSERT_TRUE(code != 0); catalogDestroy(); - memset(&gCtgMgmt, 0, sizeof(gCtgMgmt)); } TEST(multiThread, getSetRmSameDbVgroup) { @@ -2209,7 +2231,6 @@ TEST(multiThread, getSetRmSameDbVgroup) { taosSsleep(1); catalogDestroy(); - memset(&gCtgMgmt, 0, sizeof(gCtgMgmt)); } TEST(multiThread, getSetRmDiffDbVgroup) { @@ -2262,7 +2283,6 @@ TEST(multiThread, getSetRmDiffDbVgroup) { taosSsleep(1); catalogDestroy(); - memset(&gCtgMgmt, 0, sizeof(gCtgMgmt)); } TEST(multiThread, ctableMeta) { @@ -2314,7 +2334,6 @@ TEST(multiThread, ctableMeta) { taosSsleep(2); catalogDestroy(); - memset(&gCtgMgmt, 0, sizeof(gCtgMgmt)); } TEST(rentTest, allRent) { @@ -2362,6 +2381,8 @@ TEST(rentTest, allRent) { ASSERT_EQ(tableMeta->tableInfo.precision, 1); ASSERT_EQ(tableMeta->tableInfo.rowSize, 12); + taosMemoryFree(tableMeta); + while (ctgdGetClusterCacheNum(pCtg, CTG_DBG_META_NUM) < i) { taosMsleep(50); } @@ -2392,7 +2413,6 @@ TEST(rentTest, allRent) { } catalogDestroy(); - memset(&gCtgMgmt, 0, sizeof(gCtgMgmt)); } diff --git a/source/libs/scheduler/src/schTask.c b/source/libs/scheduler/src/schTask.c index de6063fbbc..dbdccff302 100644 --- a/source/libs/scheduler/src/schTask.c +++ b/source/libs/scheduler/src/schTask.c @@ -915,7 +915,7 @@ int32_t schLaunchRemoteTask(SSchJob *pJob, SSchTask *pTask) { SCH_ERR_RET(schSetTaskCandidateAddrs(pJob, pTask)); if (SCH_IS_QUERY_JOB(pJob)) { -// SCH_ERR_RET(schEnsureHbConnection(pJob, pTask)); + SCH_ERR_RET(schEnsureHbConnection(pJob, pTask)); } SCH_RET(schBuildAndSendMsg(pJob, pTask, NULL, plan->msgType)); diff --git a/source/libs/scheduler/src/schUtil.c b/source/libs/scheduler/src/schUtil.c index e20ed9d58e..1f1288fcfd 100644 --- a/source/libs/scheduler/src/schUtil.c +++ b/source/libs/scheduler/src/schUtil.c @@ -17,6 +17,7 @@ #include "command.h" #include "query.h" #include "schInt.h" +#include "tglobal.h" #include "tmsg.h" #include "tref.h" #include "trpc.h" @@ -184,6 +185,10 @@ void schDeregisterTaskHb(SSchJob *pJob, SSchTask *pTask) { } int32_t schEnsureHbConnection(SSchJob *pJob, SSchTask *pTask) { + if (!tsEnableQueryHb) { + return TSDB_CODE_SUCCESS; + } + SQueryNodeAddr *addr = taosArrayGet(pTask->candidateAddrs, pTask->candidateIdx); SQueryNodeEpId epId = {0}; -- GitLab