Merge remote-tracking branch 'origin/3.0' into fix/TD-20643

cmake/taostools_CMakeLists.txt.in

@@ -2,7 +2,7 @@
 # taos-tools
 ExternalProject_Add(taos-tools
         GIT_REPOSITORY https://github.com/taosdata/taos-tools.git
-        GIT_TAG 7e9ce09
+        GIT_TAG 2aac500
         SOURCE_DIR "${TD_SOURCE_DIR}/tools/taos-tools"
         BINARY_DIR ""
         #BUILD_IN_SOURCE TRUE

include/libs/parser/parser.h

@@ -29,7 +29,7 @@ struct SMetaData;
 typedef struct SStmtCallback {
   TAOS_STMT* pStmt;
   int32_t (*getTbNameFn)(TAOS_STMT*, char**);
-  int32_t (*setInfoFn)(TAOS_STMT*, STableMeta*, void*, char*, bool, SHashObj*, SHashObj*, const char*);
+  int32_t (*setInfoFn)(TAOS_STMT*, STableMeta*, void*, SName*, bool, SHashObj*, SHashObj*, const char*);
   int32_t (*getExecInfoFn)(TAOS_STMT*, SHashObj**, SHashObj**);
 } SStmtCallback;
 

include/util/tdef.h

@@ -406,7 +406,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 20
 #endif
 
 #define TSDB_QUERY_TYPE_NON_TYPE 0x00u  // none type

source/client/src/clientImpl.c

@@ -131,7 +131,7 @@ STscObj* taos_connect_internal(const char* ip, const char* user, const char* pas
     p = taosMemoryCalloc(1, sizeof(struct SAppInstInfo));
     p->mgmtEp = epSet;
     taosThreadMutexInit(&p->qnodeMutex, NULL);
-    p->pTransporter = openTransporter(user, secretEncrypt, tsNumOfCores);
+    p->pTransporter = openTransporter(user, secretEncrypt, tsNumOfCores / 2);
     p->pAppHbMgr = appHbMgrInit(p, key);
     if (NULL == p->pAppHbMgr) {
       destroyAppInst(p);
@@ -2293,10 +2293,16 @@ TAOS_RES* taosQueryImpl(TAOS* taos, const char* sql, bool validateOnly) {
 
   taosAsyncQueryImpl(*(int64_t*)taos, sql, syncQueryFn, param, validateOnly);
   tsem_wait(&param->sem);
+
+  SRequestObj *pRequest = NULL;
   if (param->pRequest != NULL) {
     param->pRequest->syncQuery = true;
+    pRequest = param->pRequest;
+  } else {
+    taosMemoryFree(param);
   }
-  return param->pRequest;
+  
+  return pRequest;
 }
 
 TAOS_RES* taosQueryImplWithReqid(TAOS* taos, const char* sql, bool validateOnly, int64_t reqid) {
@@ -2310,8 +2316,14 @@ TAOS_RES* taosQueryImplWithReqid(TAOS* taos, const char* sql, bool validateOnly,
 
   taosAsyncQueryImplWithReqid(*(int64_t*)taos, sql, syncQueryFn, param, validateOnly, reqid);
   tsem_wait(&param->sem);
+
+  SRequestObj *pRequest = NULL;
   if (param->pRequest != NULL) {
     param->pRequest->syncQuery = true;
+    pRequest = param->pRequest;
+  } else {
+    taosMemoryFree(param);
   }
-  return param->pRequest;
+  
+  return pRequest;
 }

source/client/src/clientStmt.c

@@ -152,9 +152,12 @@ int32_t stmtRestoreQueryFields(STscStmt* pStmt) {
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t stmtUpdateBindInfo(TAOS_STMT* stmt, STableMeta* pTableMeta, void* tags, char* tbFName, const char* sTableName, bool autoCreateTbl) {
+int32_t stmtUpdateBindInfo(TAOS_STMT* stmt, STableMeta* pTableMeta, void* tags, SName* tbName, const char* sTableName, bool autoCreateTbl) {
   STscStmt* pStmt = (STscStmt*)stmt;
+  char           tbFName[TSDB_TABLE_FNAME_LEN];
+  tNameExtractFullName(tbName, tbFName);
 
+  memcpy(&pStmt->bInfo.sname, tbName, sizeof(*tbName));
   strncpy(pStmt->bInfo.tbFName, tbFName, sizeof(pStmt->bInfo.tbFName) - 1);
   pStmt->bInfo.tbFName[sizeof(pStmt->bInfo.tbFName) - 1] = 0;
 
@@ -178,11 +181,11 @@ int32_t stmtUpdateExecInfo(TAOS_STMT* stmt, SHashObj* pVgHash, SHashObj* pBlockH
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t stmtUpdateInfo(TAOS_STMT* stmt, STableMeta* pTableMeta, void* tags, char* tbFName, bool autoCreateTbl,
+int32_t stmtUpdateInfo(TAOS_STMT* stmt, STableMeta* pTableMeta, void* tags, SName* tbName, bool autoCreateTbl,
                        SHashObj* pVgHash, SHashObj* pBlockHash, const char* sTableName) {
   STscStmt* pStmt = (STscStmt*)stmt;
 
-  STMT_ERR_RET(stmtUpdateBindInfo(stmt, pTableMeta, tags, tbFName, sTableName, autoCreateTbl));
+  STMT_ERR_RET(stmtUpdateBindInfo(stmt, pTableMeta, tags, tbName, sTableName, autoCreateTbl));
   STMT_ERR_RET(stmtUpdateExecInfo(stmt, pVgHash, pBlockHash, autoCreateTbl));
 
   pStmt->sql.autoCreateTbl = autoCreateTbl;
@@ -773,7 +776,9 @@ int stmtAddBatch(TAOS_STMT* stmt) {
 int stmtUpdateTableUid(STscStmt* pStmt, SSubmitRsp* pRsp) {
   tscDebug("stmt start to update tbUid, blockNum: %d", pRsp->nBlocks);
 
-  size_t             keyLen = 0;
+  int32_t code = 0;
+  int32_t finalCode = 0;
+  size_t  keyLen = 0;
   STableDataBlocks** pIter = taosHashIterate(pStmt->exec.pBlockHash, NULL);
   while (pIter) {
     STableDataBlocks* pBlock = *pIter;
@@ -809,10 +814,20 @@ int stmtUpdateTableUid(STscStmt* pStmt, SSubmitRsp* pRsp) {
     } else {
       tscDebug("table %s not found in submit rsp, will update from catalog", pStmt->bInfo.tbFName);
       if (NULL == pStmt->pCatalog) {
-        STMT_ERR_RET(catalogGetHandle(pStmt->taos->pAppInfo->clusterId, &pStmt->pCatalog));
+        code = catalogGetHandle(pStmt->taos->pAppInfo->clusterId, &pStmt->pCatalog);
+        if (code) {
+          pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
+          finalCode = code;
+          continue;
+        }
       }
 
-      STMT_ERR_RET(stmtCreateRequest(pStmt));
+      code = stmtCreateRequest(pStmt);
+      if (code) {
+        pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
+        finalCode = code;
+        continue;
+      }
 
       STableMeta*      pTableMeta = NULL;
       SRequestConnInfo conn = {.pTrans = pStmt->taos->pAppInfo->pTransporter,
@@ -823,20 +838,23 @@ int stmtUpdateTableUid(STscStmt* pStmt, SSubmitRsp* pRsp) {
 
       taos_free_result(pStmt->exec.pRequest);
       pStmt->exec.pRequest = NULL;
-      
-      if (TSDB_CODE_PAR_TABLE_NOT_EXIST == code) {
-        tscDebug("tb %s not exist", pStmt->bInfo.tbFName);
-        return TSDB_CODE_SUCCESS;
+
+      if (code || NULL == pTableMeta) {
+        pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
+        finalCode = code;
+        taosMemoryFree(pTableMeta);
+        continue;
       }
 
       pMeta->uid = pTableMeta->uid;
       pStmt->bInfo.tbUid = pTableMeta->uid;
+      taosMemoryFree(pTableMeta);      
     }
 
     pIter = taosHashIterate(pStmt->exec.pBlockHash, pIter);
   }
 
-  return TSDB_CODE_SUCCESS;
+  return finalCode;
 }
 
 int stmtExec(TAOS_STMT* stmt) {

source/common/src/tglobal.c

@@ -308,6 +308,9 @@ static int32_t taosAddClientCfg(SConfig *pCfg) {
 
   tsNumOfTaskQueueThreads = tsNumOfCores / 2;
   tsNumOfTaskQueueThreads = TMAX(tsNumOfTaskQueueThreads, 4);
+  if (tsNumOfTaskQueueThreads >= 10) {
+    tsNumOfTaskQueueThreads = 10;
+  }
   if (cfgAddInt32(pCfg, "numOfTaskQueueThreads", tsNumOfTaskQueueThreads, 4, 1024, 0) != 0) return -1;
 
   return 0;

source/dnode/mgmt/node_mgmt/src/dmTransport.c

@@ -250,7 +250,7 @@ int32_t dmInitClient(SDnode *pDnode) {
 
   SRpcInit rpcInit = {0};
   rpcInit.label = "DND-C";
-  rpcInit.numOfThreads = 4;
+  rpcInit.numOfThreads = tsNumOfRpcThreads;
   rpcInit.cfp = (RpcCfp)dmProcessRpcMsg;
   rpcInit.sessions = 1024;
   rpcInit.connType = TAOS_CONN_CLIENT;

source/libs/catalog/src/ctgAsync.c

@@ -1205,7 +1205,7 @@ int32_t ctgHandleGetTbMetasRsp(SCtgTaskReq* tReq, int32_t reqType, const SDataBu
           stbCtx.pName = &stbName;
 
           STableMeta* stbMeta = NULL;
-          ctgReadTbMetaFromCache(pCtg, &stbCtx, &stbMeta);
+          (void)ctgReadTbMetaFromCache(pCtg, &stbCtx, &stbMeta);
           if (stbMeta && stbMeta->sversion >= pOut->tbMeta->sversion) {
             ctgDebug("use cached stb meta, tbName:%s", tNameGetTableName(pName));
             exist = 1;

source/libs/command/src/explain.c

@@ -782,13 +782,18 @@ int32_t qExplainResNodeToRowsImpl(SExplainResNode *pResNode, SExplainCtx *ctx, i
     }
     case QUERY_NODE_PHYSICAL_PLAN_EXCHANGE: {
       SExchangePhysiNode *pExchNode = (SExchangePhysiNode *)pNode;
-      SExplainGroup      *group = taosHashGet(ctx->groupHash, &pExchNode->srcStartGroupId, sizeof(pExchNode->srcStartGroupId));
-      if (NULL == group) {
-        qError("exchange src group %d not in groupHash", pExchNode->srcStartGroupId);
-        QRY_ERR_RET(TSDB_CODE_QRY_APP_ERROR);
+      int32_t nodeNum = 0;
+      for (int32_t i = pExchNode->srcStartGroupId; i <= pExchNode->srcEndGroupId; ++i) {
+        SExplainGroup      *group = taosHashGet(ctx->groupHash, &pExchNode->srcStartGroupId, sizeof(pExchNode->srcStartGroupId));
+        if (NULL == group) {
+          qError("exchange src group %d not in groupHash", pExchNode->srcStartGroupId);
+          QRY_ERR_RET(TSDB_CODE_QRY_APP_ERROR);
+        }
+        
+        nodeNum += group->nodeNum;
       }
 
-      EXPLAIN_ROW_NEW(level, EXPLAIN_EXCHANGE_FORMAT, pExchNode->singleChannel ? 1 : group->nodeNum);
+      EXPLAIN_ROW_NEW(level, EXPLAIN_EXCHANGE_FORMAT, pExchNode->singleChannel ? 1 : nodeNum);
       EXPLAIN_ROW_APPEND(EXPLAIN_LEFT_PARENTHESIS_FORMAT);
       if (pResNode->pExecInfo) {
         QRY_ERR_RET(qExplainBufAppendExecInfo(pResNode->pExecInfo, tbuf, &tlen));
@@ -819,7 +824,9 @@ int32_t qExplainResNodeToRowsImpl(SExplainResNode *pResNode, SExplainCtx *ctx, i
         }      
       }
 
-      QRY_ERR_RET(qExplainAppendGroupResRows(ctx, pExchNode->srcStartGroupId, level + 1, pExchNode->singleChannel));
+      for (int32_t i = pExchNode->srcStartGroupId; i <= pExchNode->srcEndGroupId; ++i) {
+        QRY_ERR_RET(qExplainAppendGroupResRows(ctx, i, level + 1, pExchNode->singleChannel));
+      }
       break;
     }
     case QUERY_NODE_PHYSICAL_PLAN_SORT: {

source/libs/parser/src/parAstCreater.c

@@ -1407,7 +1407,7 @@ SNode* createShowTableTagsStmt(SAstCreateContext* pCxt, SNode* pTbName, SNode* p
 
 SNode* createCreateUserStmt(SAstCreateContext* pCxt, SToken* pUserName, const SToken* pPassword, int8_t sysinfo) {
   CHECK_PARSER_STATUS(pCxt);
-  char password[TSDB_USET_PASSWORD_LEN] = {0};
+  char password[TSDB_USET_PASSWORD_LEN + 3] = {0};
   if (!checkUserName(pCxt, pUserName) || !checkPassword(pCxt, pPassword, password)) {
     return NULL;
   }

source/libs/parser/src/parInsertSql.c

@@ -1529,9 +1529,7 @@ static int32_t setStmtInfo(SInsertParseContext* pCxt, SVnodeModifOpStmt* pStmt)
   memcpy(tags, &pCxt->tags, sizeof(pCxt->tags));
 
   SStmtCallback* pStmtCb = pCxt->pComCxt->pStmtCb;
-  char           tbFName[TSDB_TABLE_FNAME_LEN];
-  tNameExtractFullName(&pStmt->targetTableName, tbFName);
-  int32_t code = (*pStmtCb->setInfoFn)(pStmtCb->pStmt, pStmt->pTableMeta, tags, tbFName, pStmt->usingTableProcessing,
+  int32_t code = (*pStmtCb->setInfoFn)(pStmtCb->pStmt, pStmt->pTableMeta, tags, &pStmt->targetTableName, pStmt->usingTableProcessing,
                                        pStmt->pVgroupsHashObj, pStmt->pTableBlockHashObj, pStmt->usingTableName.tname);
 
   memset(&pCxt->tags, 0, sizeof(pCxt->tags));

source/libs/transport/src/trans.c

@@ -58,6 +58,9 @@ void* rpcOpen(const SRpcInit* pInit) {
   pRpc->destroyFp = pInit->dfp;
 
   pRpc->numOfThreads = pInit->numOfThreads > TSDB_MAX_RPC_THREADS ? TSDB_MAX_RPC_THREADS : pInit->numOfThreads;
+  if (pRpc->numOfThreads <= 0) {
+    pRpc->numOfThreads = 1;
+  }
 
   uint32_t ip = 0;
   if (pInit->connType == TAOS_CONN_SERVER) {

source/libs/transport/src/transCli.c

@@ -656,6 +656,7 @@ static SCliConn* cliCreateConn(SCliThrd* pThrd) {
   conn->stream = (uv_stream_t*)taosMemoryMalloc(sizeof(uv_tcp_t));
   uv_tcp_init(pThrd->loop, (uv_tcp_t*)(conn->stream));
   conn->stream->data = conn;
+  transSetConnOption((uv_tcp_t*)conn->stream);
 
   uv_timer_t* timer = taosArrayGetSize(pThrd->timerList) > 0 ? *(uv_timer_t**)taosArrayPop(pThrd->timerList) : NULL;
   if (timer == NULL) {

source/libs/transport/src/transComm.c

@@ -202,9 +202,8 @@ bool transReadComplete(SConnBuffer* connBuf) {
 }
 
 int transSetConnOption(uv_tcp_t* stream) {
-  uv_tcp_nodelay(stream, 0);
-  int ret = uv_tcp_keepalive(stream, 5, 60);
-  return ret;
+  return uv_tcp_nodelay(stream, 1);
+  // int ret = uv_tcp_keepalive(stream, 5, 60);
 }
 
 SAsyncPool* transAsyncPoolCreate(uv_loop_t* loop, int sz, void* arg, AsyncCB cb) {

source/libs/transport/src/transSvr.c

@@ -846,7 +846,7 @@ static bool addHandleToAcceptloop(void* arg) {
   return true;
 }
 void* transWorkerThread(void* arg) {
-  setThreadName("trans-worker");
+  setThreadName("trans-svr-work");
   SWorkThrd* pThrd = (SWorkThrd*)arg;
   uv_run(pThrd->loop, UV_RUN_DEFAULT);
 

source/util/src/tarray.c

@@ -205,6 +205,9 @@ void* taosArrayPop(SArray* pArray) {
 }
 
 void* taosArrayGet(const SArray* pArray, size_t index) {
+  if (NULL == pArray) {
+    return NULL;
+  }
   assert(index < pArray->size);
   return TARRAY_GET_ELEM(pArray, index);
 }

source/util/src/tlog.c

@@ -21,7 +21,7 @@
 
 #define LOG_MAX_LINE_SIZE             (1024)
 #define LOG_MAX_LINE_BUFFER_SIZE      (LOG_MAX_LINE_SIZE + 3)
-#define LOG_MAX_LINE_DUMP_SIZE        (65 * 1024)
+#define LOG_MAX_LINE_DUMP_SIZE        (1024 * 1024)
 #define LOG_MAX_LINE_DUMP_BUFFER_SIZE (LOG_MAX_LINE_DUMP_SIZE + 3)
 
 #define LOG_FILE_NAME_LEN    300
@@ -496,7 +496,7 @@ void taosPrintLongString(const char *flags, ELogLevel level, int32_t dflag, cons
   if (!osLogSpaceAvailable()) return;
   if (!(dflag & DEBUG_FILE) && !(dflag & DEBUG_SCREEN)) return;
 
-  char    buffer[LOG_MAX_LINE_DUMP_BUFFER_SIZE];
+  char *buffer = taosMemoryMalloc(LOG_MAX_LINE_DUMP_BUFFER_SIZE);
   int32_t len = taosBuildLogHead(buffer, flags);
 
   va_list argpointer;
@@ -509,6 +509,7 @@ void taosPrintLongString(const char *flags, ELogLevel level, int32_t dflag, cons
   buffer[len] = 0;
 
   taosPrintLogImp(level, dflag, buffer, len);
+  taosMemoryFree(buffer);
 }
 
 void taosDumpData(unsigned char *msg, int32_t len) {

tests/system-test/simpletest.bat

@@ -7,12 +7,12 @@ python3 .\test.py -f 0-others\taosdMonitor.py
 @REM python3 .\test.py -f 0-others\udfTest.py
 @REM python3 .\test.py -f 0-others\udf_create.py
 @REM python3 .\test.py -f 0-others\udf_restart_taosd.py
-@REM python3 .\test.py -f 0-others\cachelast.py
+python3 .\test.py -f 0-others\cachemodel.py
 
 @REM python3 .\test.py -f 0-others\user_control.py
 @REM python3 .\test.py -f 0-others\fsync.py
 
-@REM python3 .\test.py -f 1-insert\influxdb_line_taosc_insert.py
+python3 .\test.py -f 1-insert\influxdb_line_taosc_insert.py
 @REM python3 .\test.py -f 1-insert\opentsdb_telnet_line_taosc_insert.py
 @REM python3 .\test.py -f 1-insert\opentsdb_json_taosc_insert.py
 @REM #python3 .\test.py -f 1-insert\test_stmt_muti_insert_query.py
@@ -72,7 +72,7 @@ python3 .\test.py -f 0-others\taosdMonitor.py
 @REM python3 .\test.py -f 2-query\arcsin.py
 @REM python3 .\test.py -f 2-query\arccos.py
 @REM python3 .\test.py -f 2-query\arctan.py
-@REM python3 .\test.py -f 2-query\query_cols_tags_and_or.py
+python3 .\test.py -f 2-query\query_cols_tags_and_or.py
 @REM # python3 .\test.py -f 2-query\nestedQuery.py
 @REM # TD-15983 subquery output duplicate name column. 
 @REM # Please Xiangyang Guo modify the following script
@@ -94,7 +94,7 @@ python3 .\test.py -f 0-others\taosdMonitor.py
 @REM python3 .\test.py -f 7-tmq\subscribeDb.py
 @REM python3 .\test.py -f 7-tmq\subscribeDb0.py
 @REM python3 .\test.py -f 7-tmq\subscribeDb1.py
-@REM python3 .\test.py -f 7-tmq\subscribeStb.py
+python3 .\test.py -f 7-tmq\subscribeStb.py
 @REM python3 .\test.py -f 7-tmq\subscribeStb0.py
 @REM python3 .\test.py -f 7-tmq\subscribeStb1.py
 @REM python3 .\test.py -f 7-tmq\subscribeStb2.py