Merge branch '3.0' of https://github.com/taosdata/TDengine into fix/ly_res

include/common/tglobal.h

@@ -158,6 +158,7 @@ extern int32_t tsCacheLazyLoadThreshold;  // cost threshold for last/last_row lo
 
 // query client
 extern int32_t tsQueryPolicy;
+extern bool    tsQueryTbNotExistAsEmpty;
 extern int32_t tsQueryRspPolicy;
 extern int64_t tsQueryMaxConcurrentTables;
 extern int32_t tsQuerySmaOptimize;

include/libs/parser/parser.h

@@ -65,6 +65,8 @@ typedef struct SParseCsvCxt {
   const char* pLastSqlPos;  // the location of the last parsed sql
 } SParseCsvCxt;
 
+typedef void(*setQueryFn)(int64_t);
+
 typedef struct SParseContext {
   uint64_t         requestId;
   int64_t          requestRid;
@@ -98,6 +100,7 @@ typedef struct SParseContext {
   void*            parseSqlParam;
   int8_t           biMode;
   SArray*          pSubMetaList;
+  setQueryFn       setQueryFp;
 } SParseContext;
 
 int32_t qParseSql(SParseContext* pCxt, SQuery** pQuery);

include/util/tdef.h

@@ -499,7 +499,7 @@ typedef enum ELogicConditionType {
 #ifdef WINDOWS
 #define TSDB_MAX_RPC_THREADS 4  // windows pipe only support 4 connections.
 #else
-#define TSDB_MAX_RPC_THREADS 10
+#define TSDB_MAX_RPC_THREADS 50
 #endif
 
 #define TSDB_QUERY_TYPE_NON_TYPE 0x00u  // none type

source/client/inc/clientInt.h

@@ -52,11 +52,11 @@ enum {
 #define SHOW_VARIABLES_RESULT_FIELD2_LEN (TSDB_CONFIG_VALUE_LEN + VARSTR_HEADER_SIZE)
 #define SHOW_VARIABLES_RESULT_FIELD3_LEN (TSDB_CONFIG_SCOPE_LEN + VARSTR_HEADER_SIZE)
 
-#define TD_RES_QUERY(res)          (*(int8_t*)res == RES_TYPE__QUERY)
+#define TD_RES_QUERY(res)          (*(int8_t*)(res) == RES_TYPE__QUERY)
-#define TD_RES_TMQ(res)            (*(int8_t*)res == RES_TYPE__TMQ)
+#define TD_RES_TMQ(res)            (*(int8_t*)(res) == RES_TYPE__TMQ)
-#define TD_RES_TMQ_META(res)       (*(int8_t*)res == RES_TYPE__TMQ_META)
+#define TD_RES_TMQ_META(res)       (*(int8_t*)(res) == RES_TYPE__TMQ_META)
-#define TD_RES_TMQ_METADATA(res)   (*(int8_t*)res == RES_TYPE__TMQ_METADATA)
+#define TD_RES_TMQ_METADATA(res)   (*(int8_t*)(res) == RES_TYPE__TMQ_METADATA)
-#define TD_RES_TMQ_BATCH_META(res) (*(int8_t*)res == RES_TYPE__TMQ_BATCH_META)
+#define TD_RES_TMQ_BATCH_META(res) (*(int8_t*)(res) == RES_TYPE__TMQ_BATCH_META)
 
 typedef struct SAppInstInfo SAppInstInfo;
 
@@ -284,6 +284,7 @@ typedef struct SRequestObj {
   bool                 isSubReq;
   bool                 inCallback;
   bool                 isStmtBind;  // is statement bind parameter
+  bool                 isQuery;
   uint32_t             prevCode;  // previous error code: todo refactor, add update flag for catalog
   uint32_t             retry;
   int64_t              allocatorRefId;
@@ -420,6 +421,7 @@ typedef struct SSqlCallbackWrapper {
   void*          pPlanInfo;
 } SSqlCallbackWrapper;
 
+void setQueryRequest(int64_t rId);
 SRequestObj* launchQueryImpl(SRequestObj* pRequest, SQuery* pQuery, bool keepQuery, void** res);
 int32_t      scheduleQuery(SRequestObj* pRequest, SQueryPlan* pDag, SArray* pNodeList);
 void    launchAsyncQuery(SRequestObj* pRequest, SQuery* pQuery, SMetaData* pResultMeta, SSqlCallbackWrapper* pWrapper);

source/client/src/clientImpl.c

@@ -31,6 +31,15 @@
 static int32_t initEpSetFromCfg(const char* firstEp, const char* secondEp, SCorEpSet* pEpSet);
 static int32_t buildConnectMsg(SRequestObj* pRequest, SMsgSendInfo** pMsgSendInfo);
 
+void setQueryRequest(int64_t rId) {
+  SRequestObj* pReq = acquireRequest(rId);
+  if (pReq != NULL) {
+    pReq->isQuery = true;
+    (void)releaseRequest(rId);
+  }
+
+}
+
 static bool stringLengthCheck(const char* str, size_t maxsize) {
   if (str == NULL) {
     return false;
@@ -286,7 +295,8 @@ int32_t parseSql(SRequestObj* pRequest, bool topicQuery, SQuery** pQuery, SStmtC
                        .enableSysInfo = pTscObj->sysInfo,
                        .svrVer = pTscObj->sVer,
                        .nodeOffline = (pTscObj->pAppInfo->onlineDnodes < pTscObj->pAppInfo->totalDnodes),
-                       .isStmtBind = pRequest->isStmtBind};
+                       .isStmtBind = pRequest->isStmtBind,
+                       .setQueryFp = setQueryRequest};
 
   cxt.mgmtEpSet = getEpSet_s(&pTscObj->pAppInfo->mgmtEp);
   int32_t code = catalogGetHandle(pTscObj->pAppInfo->clusterId, &cxt.pCatalog);
@@ -2953,6 +2963,10 @@ void taosAsyncFetchImpl(SRequestObj* pRequest, __taos_async_fn_t fp, void* param
 void doRequestCallback(SRequestObj* pRequest, int32_t code) {
   pRequest->inCallback = true;
   int64_t this = pRequest->self;
+  if (tsQueryTbNotExistAsEmpty && TD_RES_QUERY(&pRequest->resType) && pRequest->isQuery && (code == TSDB_CODE_PAR_TABLE_NOT_EXIST || code == TSDB_CODE_TDB_TABLE_NOT_EXIST)) {
+    code = TSDB_CODE_SUCCESS;
+    pRequest->type = TSDB_SQL_RETRIEVE_EMPTY_RESULT;
+  }
   pRequest->body.queryFp(((SSyncQueryParam*)pRequest->body.interParam)->userParam, pRequest, code);
   SRequestObj* pReq = acquireRequest(this);
   if (pReq != NULL) {

source/client/src/clientMain.c

@@ -1235,7 +1235,8 @@ int32_t createParseContext(const SRequestObj *pRequest, SParseContext **pCxt, SS
                            .nodeOffline = (pTscObj->pAppInfo->onlineDnodes < pTscObj->pAppInfo->totalDnodes),
                            .allocatorId = pRequest->allocatorRefId,
                            .parseSqlFp = clientParseSql,
-                           .parseSqlParam = pWrapper};
+                           .parseSqlParam = pWrapper,
+                           .setQueryFp = setQueryRequest};
   int8_t biMode = atomic_load_8(&((STscObj *)pTscObj)->biMode);
   (*pCxt)->biMode = biMode;
   return TSDB_CODE_SUCCESS;

source/client/src/clientStmt.c

@@ -1241,7 +1241,9 @@ int stmtBindBatch(TAOS_STMT* stmt, TAOS_MULTI_BIND* bind, int32_t colIdx) {
                          .msgLen = ERROR_MSG_BUF_DEFAULT_SIZE,
                          .pTransporter = pStmt->taos->pAppInfo->pTransporter,
                          .pStmtCb = NULL,
-                         .pUser = pStmt->taos->user};
+                         .pUser = pStmt->taos->user,
+                         .setQueryFp = setQueryRequest};
+
     ctx.mgmtEpSet = getEpSet_s(&pStmt->taos->pAppInfo->mgmtEp);
     STMT_ERR_RET(catalogGetHandle(pStmt->taos->pAppInfo->clusterId, &ctx.pCatalog));
 

source/common/src/systable.c

@@ -427,6 +427,7 @@ static const SSysDbTableSchema userMachinesSchema[] = {
     {.name = "id", .bytes = TSDB_CLUSTER_ID_LEN + 1 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = true},
     {.name = "dnode_num", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = true},
     {.name = "machine", .bytes = 7552 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = true},
+    {.name = "version", .bytes = 32 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = true},
 };
 
 static const SSysDbTableSchema encryptionsSchema[] = {

source/common/src/tglobal.c

@@ -162,6 +162,7 @@ int32_t tmqMaxTopicNum = 20;
 int32_t tmqRowSize = 4096;
 // query
 int32_t tsQueryPolicy = 1;
+bool    tsQueryTbNotExistAsEmpty = false;
 int32_t tsQueryRspPolicy = 0;
 int64_t tsQueryMaxConcurrentTables = 200;  // unit is TSDB_TABLE_NUM_UNIT
 bool    tsEnableQueryHb = true;
@@ -569,6 +570,7 @@ static int32_t taosAddClientCfg(SConfig *pCfg) {
   TAOS_CHECK_RETURN(
       cfgAddInt32(pCfg, "compressMsgSize", tsCompressMsgSize, -1, 100000000, CFG_SCOPE_BOTH, CFG_DYN_CLIENT));
   TAOS_CHECK_RETURN(cfgAddInt32(pCfg, "queryPolicy", tsQueryPolicy, 1, 4, CFG_SCOPE_CLIENT, CFG_DYN_ENT_CLIENT));
+  TAOS_CHECK_RETURN(cfgAddBool(pCfg, "queryTableNotExistAsEmpty", tsQueryTbNotExistAsEmpty, CFG_SCOPE_CLIENT, CFG_DYN_CLIENT));
   TAOS_CHECK_RETURN(cfgAddBool(pCfg, "enableQueryHb", tsEnableQueryHb, CFG_SCOPE_CLIENT, CFG_DYN_CLIENT));
   TAOS_CHECK_RETURN(cfgAddBool(pCfg, "enableScience", tsEnableScience, CFG_SCOPE_CLIENT, CFG_DYN_NONE));
   TAOS_CHECK_RETURN(cfgAddInt32(pCfg, "querySmaOptimize", tsQuerySmaOptimize, 0, 1, CFG_SCOPE_CLIENT, CFG_DYN_CLIENT));
@@ -1181,6 +1183,9 @@ static int32_t taosSetClientCfg(SConfig *pCfg) {
   TAOS_CHECK_GET_CFG_ITEM(pCfg, pItem, "queryPolicy");
   tsQueryPolicy = pItem->i32;
 
+  TAOS_CHECK_GET_CFG_ITEM(pCfg, pItem, "queryTableNotExistAsEmpty");
+  tsQueryTbNotExistAsEmpty = pItem->bval;
+
   TAOS_CHECK_GET_CFG_ITEM(pCfg, pItem, "enableQueryHb");
   tsEnableQueryHb = pItem->bval;
 
@@ -2218,6 +2223,7 @@ static int32_t taosCfgDynamicOptionsForClient(SConfig *pCfg, const char *name) {
                                          {"numOfLogLines", &tsNumOfLogLines},
                                          {"querySmaOptimize", &tsQuerySmaOptimize},
                                          {"queryPolicy", &tsQueryPolicy},
+                                         {"queryTableNotExistAsEmpty", &tsQueryTbNotExistAsEmpty},
                                          {"queryPlannerTrace", &tsQueryPlannerTrace},
                                          {"queryNodeChunkSize", &tsQueryNodeChunkSize},
                                          {"queryUseNodeAllocator", &tsQueryUseNodeAllocator},

source/dnode/vnode/src/tsdb/tsdbRead2.c

@@ -211,7 +211,7 @@ static int32_t updateBlockSMAInfo(STSchema* pSchema, SBlockLoadSuppInfo* pSupInf
   while (i < pSchema->numOfCols && j < pSupInfo->numOfCols) {
     STColumn* pTCol = &pSchema->columns[i];
     if (pTCol->colId == pSupInfo->colId[j]) {
-      if (!IS_BSMA_ON(pTCol)) {
+      if (!IS_BSMA_ON(pTCol) && (PRIMARYKEY_TIMESTAMP_COL_ID != pTCol->colId)) {
         pSupInfo->smaValid = false;
         return TSDB_CODE_SUCCESS;
       }

source/dnode/vnode/src/vnd/vnodeSvr.c

@@ -253,11 +253,6 @@ static int32_t vnodePreProcessSubmitTbData(SVnode *pVnode, SDecoder *pCoder, int
   version = (submitTbData.flags >> 8) & 0xff;
   submitTbData.flags = submitTbData.flags & 0xff;
 
-  if (submitTbData.flags & SUBMIT_REQ_FROM_FILE) {
-    code = grantCheck(TSDB_GRANT_CSV);
-    TSDB_CHECK_CODE(code, lino, _exit);
-  }
-
   int64_t uid;
   if (submitTbData.flags & SUBMIT_REQ_AUTO_CREATE_TABLE) {
     code = vnodePreprocessCreateTableReq(pVnode, pCoder, btimeMs, &uid);

source/libs/parser/src/parTranslater.c

@@ -4681,7 +4681,8 @@ int32_t translateTable(STranslateContext* pCxt, SNode** pTable, SNode* pJoinPare
             pCxt, toName(pCxt->pParseCxt->acctId, pRealTable->table.dbName, pRealTable->table.tableName, &name),
             &(pRealTable->pMeta), true);
         if (TSDB_CODE_SUCCESS != code) {
-          return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_GET_META_ERROR, tstrerror(code));
+          (void)generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_GET_META_ERROR, tstrerror(code));
+          return code;
         }
 #ifdef TD_ENTERPRISE
         if (TSDB_VIEW_TABLE == pRealTable->pMeta->tableType && (!pCurrSmt->tagScan || pCxt->pParseCxt->biMode)) {
@@ -6783,6 +6784,10 @@ static int32_t translateSelectFrom(STranslateContext* pCxt, SSelectStmt* pSelect
 }
 
 static int32_t translateSelect(STranslateContext* pCxt, SSelectStmt* pSelect) {
+  if (pCxt->pParseCxt && pCxt->pParseCxt->setQueryFp) {
+    (*pCxt->pParseCxt->setQueryFp)(pCxt->pParseCxt->requestRid);
+  }
+  
   if (NULL == pSelect->pFromTable) {
     return translateSelectWithoutFrom(pCxt, pSelect);
   } else {
@@ -6907,6 +6912,10 @@ static int32_t checkSetOperLimit(STranslateContext* pCxt, SLimitNode* pLimit) {
 }
 
 static int32_t translateSetOperator(STranslateContext* pCxt, SSetOperator* pSetOperator) {
+  if (pCxt->pParseCxt && pCxt->pParseCxt->setQueryFp) {
+    (*pCxt->pParseCxt->setQueryFp)(pCxt->pParseCxt->requestRid);
+  }
+
   int32_t code = translateQuery(pCxt, pSetOperator->pLeft);
   if (TSDB_CODE_SUCCESS == code) {
     code = resetHighLevelTranslateNamespace(pCxt);

source/libs/parser/test/parSelectTest.cpp

@@ -366,9 +366,9 @@ TEST_F(ParserSelectTest, semanticCheck) {
   run("SELECT t1.c1, t1.cc1 FROM t1", TSDB_CODE_PAR_INVALID_COLUMN);
 
   // TSDB_CODE_PAR_GET_META_ERROR
-  run("SELECT * FROM t10", TSDB_CODE_PAR_GET_META_ERROR);
+  run("SELECT * FROM t10", TSDB_CODE_PAR_TABLE_NOT_EXIST);
 
-  run("SELECT * FROM test.t10", TSDB_CODE_PAR_GET_META_ERROR);
+  run("SELECT * FROM test.t10", TSDB_CODE_PAR_TABLE_NOT_EXIST);
 
   // TSDB_CODE_PAR_TABLE_NOT_EXIST
   run("SELECT t2.c1 FROM t1", TSDB_CODE_PAR_TABLE_NOT_EXIST);

source/libs/transport/inc/transComm.h

@@ -272,19 +272,19 @@ bool    transAsyncPoolIsEmpty(SAsyncPool* pool);
     }                                                                \
   } while (0)
 
-#define ASYNC_CHECK_HANDLE(exh1, id)                                \
+#define ASYNC_CHECK_HANDLE(exh1, id)                                   \
-  do {                                                              \
+  do {                                                                 \
-    if (id > 0) {                                                   \
+    if (id > 0) {                                                      \
-      SExHandle* exh2 = transAcquireExHandle(transGetRefMgt(), id); \
+      SExHandle* exh2 = transAcquireExHandle(transGetSvrRefMgt(), id); \
-      if (exh2 == NULL || id != exh2->refId) {                      \
+      if (exh2 == NULL || id != exh2->refId) {                         \
-        tDebug("ref:%" PRId64 " already released" PRIu64, id);      \
+        tDebug("ref:%" PRId64 " already released", id);                \
-        code = terrno;                                              \
+        code = terrno;                                                 \
-        goto _return1;                                              \
+        goto _return1;                                                 \
-      }                                                             \
+      }                                                                \
-    } else {                                                        \
+    } else {                                                           \
-      tWarn("invalid handle to release");                           \
+      tDebug("invalid handle to release");                             \
-      goto _return2;                                                \
+      goto _return2;                                                   \
-    }                                                               \
+    }                                                                  \
   } while (0)
 
 int32_t transInitBuffer(SConnBuffer* buf);
@@ -443,6 +443,7 @@ int32_t transReleaseExHandle(int32_t refMgt, int64_t refId);
 void    transDestroyExHandle(void* handle);
 
 int32_t transGetRefMgt();
+int32_t transGetSvrRefMgt();
 int32_t transGetInstMgt();
 int32_t transGetSyncMsgMgt();
 

source/libs/transport/src/transComm.c

@@ -20,6 +20,7 @@
 static TdThreadOnce transModuleInit = PTHREAD_ONCE_INIT;
 
 static int32_t refMgt;
+static int32_t svrRefMgt;
 static int32_t instMgt;
 static int32_t transSyncMsgMgt;
 
@@ -704,12 +705,14 @@ bool transEpSetIsEqual2(SEpSet* a, SEpSet* b) {
 
 static void transInitEnv() {
   refMgt = transOpenRefMgt(50000, transDestroyExHandle);
+  svrRefMgt = transOpenRefMgt(50000, transDestroyExHandle);
   instMgt = taosOpenRef(50, rpcCloseImpl);
   transSyncMsgMgt = taosOpenRef(50, transDestroySyncMsg);
   (void)uv_os_setenv("UV_TCP_SINGLE_ACCEPT", "1");
 }
 static void transDestroyEnv() {
   transCloseRefMgt(refMgt);
+  transCloseRefMgt(svrRefMgt);
   transCloseRefMgt(instMgt);
   transCloseRefMgt(transSyncMsgMgt);
 }
@@ -724,6 +727,7 @@ int32_t transInit() {
 }
 
 int32_t transGetRefMgt() { return refMgt; }
+int32_t transGetSvrRefMgt() { return svrRefMgt; }
 int32_t transGetInstMgt() { return instMgt; }
 int32_t transGetSyncMsgMgt() { return transSyncMsgMgt; }
 

source/libs/transport/src/transSvr.c

@@ -373,6 +373,7 @@ static bool uvHandleReq(SSvrConn* pConn) {
   STrans*    pTransInst = pConn->pTransInst;
   SWorkThrd* pThrd = pConn->hostThrd;
 
+  int8_t         acquire = 0;
   STransMsgHead* pHead = NULL;
 
   int8_t resetBuf = pConn->status == ConnAcquire ? 0 : 1;
@@ -459,7 +460,13 @@ static bool uvHandleReq(SSvrConn* pConn) {
   // 2. once send out data, cli conn released to conn pool immediately
   // 3. not mixed with persist
   transMsg.info.ahandle = (void*)pHead->ahandle;
-  transMsg.info.handle = (void*)transAcquireExHandle(transGetRefMgt(), pConn->refId);
+
+  if (pHead->noResp == 1) {
+    transMsg.info.handle = NULL;
+  } else {
+    transMsg.info.handle = (void*)transAcquireExHandle(transGetSvrRefMgt(), pConn->refId);
+    acquire = 1;
+  }
   transMsg.info.refId = pConn->refId;
   transMsg.info.traceId = pHead->traceId;
   transMsg.info.cliVer = htonl(pHead->compatibilityVer);
@@ -468,10 +475,10 @@ static bool uvHandleReq(SSvrConn* pConn) {
 
   tGTrace("%s handle %p conn:%p translated to app, refId:%" PRIu64, transLabel(pTransInst), transMsg.info.handle, pConn,
           pConn->refId);
-  if (transMsg.info.handle == NULL) {
+  // if (transMsg.info.handle == NULL) {
-    tError("%s handle %p conn:%p handle failed to init" PRIu64, transLabel(pTransInst), transMsg.info.handle, pConn);
+  //   tError("%s handle %p conn:%p handle failed to init" PRIu64, transLabel(pTransInst), transMsg.info.handle, pConn);
-    return false;
+  //   return false;
-  }
+  // }
 
   if (pHead->noResp == 1) {
     transMsg.info.refId = -1;
@@ -483,7 +490,7 @@ static bool uvHandleReq(SSvrConn* pConn) {
   pConnInfo->clientPort = pConn->port;
   tstrncpy(pConnInfo->user, pConn->user, sizeof(pConnInfo->user));
 
-  (void)transReleaseExHandle(transGetRefMgt(), pConn->refId);
+  if (acquire) transReleaseExHandle(transGetSvrRefMgt(), pConn->refId);
 
   (*pTransInst->cfp)(pTransInst->parent, &transMsg, NULL);
   return true;
@@ -770,15 +777,15 @@ void uvWorkerAsyncCb(uv_async_t* handle) {
 
       SExHandle* exh1 = transMsg.info.handle;
       int64_t    refId = transMsg.info.refId;
-      SExHandle* exh2 = transAcquireExHandle(transGetRefMgt(), refId);
+      SExHandle* exh2 = transAcquireExHandle(transGetSvrRefMgt(), refId);
       if (exh2 == NULL || exh1 != exh2) {
         tTrace("handle except msg %p, ignore it", exh1);
-        (void)transReleaseExHandle(transGetRefMgt(), refId);
+        (void)transReleaseExHandle(transGetSvrRefMgt(), refId);
         destroySmsg(msg);
         continue;
       }
       msg->pConn = exh1->handle;
-      (void)transReleaseExHandle(transGetRefMgt(), refId);
+      (void)transReleaseExHandle(transGetSvrRefMgt(), refId);
       (*transAsyncHandle[msg->type])(msg, pThrd);
     }
   }
@@ -874,15 +881,15 @@ static void uvPrepareCb(uv_prepare_t* handle) {
 
         SExHandle* exh1 = transMsg.info.handle;
         int64_t    refId = transMsg.info.refId;
-        SExHandle* exh2 = transAcquireExHandle(transGetRefMgt(), refId);
+        SExHandle* exh2 = transAcquireExHandle(transGetSvrRefMgt(), refId);
         if (exh2 == NULL || exh1 != exh2) {
           tTrace("handle except msg %p, ignore it", exh1);
-          (void)transReleaseExHandle(transGetRefMgt(), refId);
+          (void)transReleaseExHandle(transGetSvrRefMgt(), refId);
           destroySmsg(msg);
           continue;
         }
         msg->pConn = exh1->handle;
-        (void)transReleaseExHandle(transGetRefMgt(), refId);
+        (void)transReleaseExHandle(transGetSvrRefMgt(), refId);
         (*transAsyncHandle[msg->type])(msg, pThrd);
       }
     }
@@ -1215,14 +1222,14 @@ static FORCE_INLINE SSvrConn* createConn(void* hThrd) {
 
   exh->handle = pConn;
   exh->pThrd = pThrd;
-  exh->refId = transAddExHandle(transGetRefMgt(), exh);
+  exh->refId = transAddExHandle(transGetSvrRefMgt(), exh);
   if (exh->refId < 0) {
     TAOS_CHECK_GOTO(TSDB_CODE_REF_INVALID_ID, NULL, _end);
   }
 
   QUEUE_INIT(&exh->q);
 
-  SExHandle* pSelf = transAcquireExHandle(transGetRefMgt(), exh->refId);
+  SExHandle* pSelf = transAcquireExHandle(transGetSvrRefMgt(), exh->refId);
   if (pSelf != exh) {
     TAOS_CHECK_GOTO(TSDB_CODE_REF_INVALID_ID, NULL, _end);
   }
@@ -1284,8 +1291,8 @@ static FORCE_INLINE void destroyConnRegArg(SSvrConn* conn) {
 }
 static int32_t reallocConnRef(SSvrConn* conn) {
   if (conn->refId > 0) {
-    (void)transReleaseExHandle(transGetRefMgt(), conn->refId);
+    (void)transReleaseExHandle(transGetSvrRefMgt(), conn->refId);
-    (void)transRemoveExHandle(transGetRefMgt(), conn->refId);
+    (void)transRemoveExHandle(transGetSvrRefMgt(), conn->refId);
   }
   // avoid app continue to send msg on invalid handle
   SExHandle* exh = taosMemoryMalloc(sizeof(SExHandle));
@@ -1295,14 +1302,14 @@ static int32_t reallocConnRef(SSvrConn* conn) {
 
   exh->handle = conn;
   exh->pThrd = conn->hostThrd;
-  exh->refId = transAddExHandle(transGetRefMgt(), exh);
+  exh->refId = transAddExHandle(transGetSvrRefMgt(), exh);
   if (exh->refId < 0) {
     taosMemoryFree(exh);
     return TSDB_CODE_REF_INVALID_ID;
   }
 
   QUEUE_INIT(&exh->q);
-  SExHandle* pSelf = transAcquireExHandle(transGetRefMgt(), exh->refId);
+  SExHandle* pSelf = transAcquireExHandle(transGetSvrRefMgt(), exh->refId);
   if (pSelf != exh) {
     tError("conn %p failed to acquire handle", conn);
     taosMemoryFree(exh);
@@ -1321,8 +1328,8 @@ static void uvDestroyConn(uv_handle_t* handle) {
   }
   SWorkThrd* thrd = conn->hostThrd;
 
-  (void)transReleaseExHandle(transGetRefMgt(), conn->refId);
+  (void)transReleaseExHandle(transGetSvrRefMgt(), conn->refId);
-  (void)transRemoveExHandle(transGetRefMgt(), conn->refId);
+  (void)transRemoveExHandle(transGetSvrRefMgt(), conn->refId);
 
   STrans* pTransInst = thrd->pTransInst;
   tDebug("%s conn %p destroy", transLabel(pTransInst), conn);
@@ -1752,15 +1759,15 @@ int32_t transReleaseSrvHandle(void* handle) {
   tDebug("%s conn %p start to release", transLabel(pThrd->pTransInst), exh->handle);
   if ((code = transAsyncSend(pThrd->asyncPool, &m->q)) != 0) {
     destroySmsg(m);
-    (void)transReleaseExHandle(transGetRefMgt(), refId);
+    (void)transReleaseExHandle(transGetSvrRefMgt(), refId);
     return code;
   }
 
-  (void)transReleaseExHandle(transGetRefMgt(), refId);
+  (void)transReleaseExHandle(transGetSvrRefMgt(), refId);
   return 0;
 _return1:
   tDebug("handle %p failed to send to release handle", exh);
-  (void)transReleaseExHandle(transGetRefMgt(), refId);
+  (void)transReleaseExHandle(transGetSvrRefMgt(), refId);
   return code;
 _return2:
   tDebug("handle %p failed to send to release handle", exh);
@@ -1803,17 +1810,17 @@ int32_t transSendResponse(const STransMsg* msg) {
   tGDebug("conn %p start to send resp (1/2)", exh->handle);
   if ((code = transAsyncSend(pThrd->asyncPool, &m->q)) != 0) {
     destroySmsg(m);
-    (void)transReleaseExHandle(transGetRefMgt(), refId);
+    (void)transReleaseExHandle(transGetSvrRefMgt(), refId);
     return code;
   }
 
-  (void)transReleaseExHandle(transGetRefMgt(), refId);
+  (void)transReleaseExHandle(transGetSvrRefMgt(), refId);
   return 0;
 
 _return1:
   tDebug("handle %p failed to send resp", exh);
   rpcFreeCont(msg->pCont);
-  (void)transReleaseExHandle(transGetRefMgt(), refId);
+  (void)transReleaseExHandle(transGetSvrRefMgt(), refId);
   return code;
 _return2:
   tDebug("handle %p failed to send resp", exh);
@@ -1848,17 +1855,17 @@ int32_t transRegisterMsg(const STransMsg* msg) {
   tDebug("%s conn %p start to register brokenlink callback", transLabel(pTransInst), exh->handle);
   if ((code = transAsyncSend(pThrd->asyncPool, &m->q)) != 0) {
     destroySmsg(m);
-    (void)transReleaseExHandle(transGetRefMgt(), refId);
+    (void)transReleaseExHandle(transGetSvrRefMgt(), refId);
     return code;
   }
 
-  (void)transReleaseExHandle(transGetRefMgt(), refId);
+  (void)transReleaseExHandle(transGetSvrRefMgt(), refId);
   return 0;
 
 _return1:
   tDebug("handle %p failed to register brokenlink", exh);
   rpcFreeCont(msg->pCont);
-  (void)transReleaseExHandle(transGetRefMgt(), refId);
+  (void)transReleaseExHandle(transGetSvrRefMgt(), refId);
   return code;
 _return2:
   tDebug("handle %p failed to register brokenlink", exh);

tests/system-test/0-others/information_schema.py

@@ -222,7 +222,7 @@ class TDTestCase:
 
         tdSql.query("select * from information_schema.ins_columns where db_name ='information_schema'")
         tdLog.info(len(tdSql.queryResult))
-        tdSql.checkEqual(True, len(tdSql.queryResult) in range(261, 271))
+        tdSql.checkEqual(True, len(tdSql.queryResult) in range(271, 272))
 
         tdSql.query("select * from information_schema.ins_columns where db_name ='performance_schema'")
         tdSql.checkEqual(56, len(tdSql.queryResult))

tests/system-test/7-tmq/tmq_offset.py

@@ -15,6 +15,10 @@ sys.path.append("./7-tmq")
 from tmqCommon import *
 
 class TDTestCase:
+    clientCfgDict = {'debugFlag': 135}
+    updatecfgDict = {'debugFlag': 135, 'asynclog': 0}
+    updatecfgDict["clientCfg"] = clientCfgDict
+
     def init(self, conn, logSql, replicaVar=1):
         self.replicaVar = int(replicaVar)
         tdLog.debug(f"start to excute {__file__}")