diff --git a/cmake/cmake.define b/cmake/cmake.define index 542b4b4489..d32200bb91 100644 --- a/cmake/cmake.define +++ b/cmake/cmake.define @@ -141,13 +141,13 @@ ELSE () SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -msse4.2") SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -msse4.2") ENDIF() - IF (COMPILER_SUPPORT_FMA) - SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -mfma") - SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -mfma") - ENDIF() IF ("${SIMD_SUPPORT}" MATCHES "true") - IF (COMPILER_SUPPORT_AVX) + IF (COMPILER_SUPPORT_FMA) + SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -mfma") + SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -mfma") + ENDIF() + IF (COMPILER_SUPPORT_AVX) SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -mavx") SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -mavx") ENDIF() diff --git a/cmake/taostools_CMakeLists.txt.in b/cmake/taostools_CMakeLists.txt.in index 0cc57d1246..e23ebb104b 100644 --- a/cmake/taostools_CMakeLists.txt.in +++ b/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 261fcca + GIT_TAG 11b60a4 SOURCE_DIR "${TD_SOURCE_DIR}/tools/taos-tools" BINARY_DIR "" #BUILD_IN_SOURCE TRUE diff --git a/docs/en/14-reference/12-config/index.md b/docs/en/14-reference/12-config/index.md index bb5516ae70..274c995098 100644 --- a/docs/en/14-reference/12-config/index.md +++ b/docs/en/14-reference/12-config/index.md @@ -733,7 +733,7 @@ To prevent system resource from being exhausted by multiple concurrent streams, | 42 | numOfCommitThreads | Yes | Yes | | 43 | numOfMnodeReadThreads | No | Yes | | 44 | numOfVnodeQueryThreads | No | Yes | -| 45 | numOfVnodeStreamThreads | No | Yes | +| 45 | ratioOfVnodeStreamThreads | No | Yes | | 46 | numOfVnodeFetchThreads | No | Yes | | 47 | numOfVnodeRsmaThreads | No | Yes | | 48 | numOfQnodeQueryThreads | No | Yes | diff --git a/docs/zh/14-reference/12-config/index.md b/docs/zh/14-reference/12-config/index.md index 145c5eed93..959438f48c 100644 --- a/docs/zh/14-reference/12-config/index.md +++ b/docs/zh/14-reference/12-config/index.md @@ -709,7 +709,7 @@ charset 的有效值是 UTF-8。 | 42 | numOfCommitThreads | 是 | 是 | | | 43 | numOfMnodeReadThreads | 否 | 是 | | | 44 | numOfVnodeQueryThreads | 否 | 是 | | -| 45 | numOfVnodeStreamThreads | 否 | 是 | | +| 45 | ratioOfVnodeStreamThreads | 否 | 是 | | | 46 | numOfVnodeFetchThreads | 否 | 是 | | | 47 | numOfVnodeRsmaThreads | 否 | 是 | | | 48 | numOfQnodeQueryThreads | 否 | 是 | | diff --git a/include/common/tglobal.h b/include/common/tglobal.h index 9e8a139b31..d445fc26e8 100644 --- a/include/common/tglobal.h +++ b/include/common/tglobal.h @@ -55,7 +55,7 @@ extern int32_t tsNumOfMnodeQueryThreads; extern int32_t tsNumOfMnodeFetchThreads; extern int32_t tsNumOfMnodeReadThreads; extern int32_t tsNumOfVnodeQueryThreads; -extern int32_t tsNumOfVnodeStreamThreads; +extern float tsRatioOfVnodeStreamThreads; extern int32_t tsNumOfVnodeFetchThreads; extern int32_t tsNumOfVnodeRsmaThreads; extern int32_t tsNumOfQnodeQueryThreads; diff --git a/include/libs/executor/executor.h b/include/libs/executor/executor.h index 412b4b4cf6..cfd5bd1ed7 100644 --- a/include/libs/executor/executor.h +++ b/include/libs/executor/executor.h @@ -213,6 +213,7 @@ int32_t qStreamSourceRecoverStep1(qTaskInfo_t tinfo, int64_t ver); int32_t qStreamSourceRecoverStep2(qTaskInfo_t tinfo, int64_t ver); int32_t qStreamRecoverFinish(qTaskInfo_t tinfo); int32_t qStreamRestoreParam(qTaskInfo_t tinfo); +bool qStreamRecoverScanFinished(qTaskInfo_t tinfo); #ifdef __cplusplus } diff --git a/include/util/tworker.h b/include/util/tworker.h index 8766f87a08..0636f16dbb 100644 --- a/include/util/tworker.h +++ b/include/util/tworker.h @@ -17,6 +17,7 @@ #define _TD_UTIL_WORKER_H_ #include "tqueue.h" +#include "tarray.h" #ifdef __cplusplus extern "C" { @@ -26,10 +27,10 @@ typedef struct SQWorkerPool SQWorkerPool; typedef struct SWWorkerPool SWWorkerPool; typedef struct SQWorker { - int32_t id; // worker id - int64_t pid; // thread pid - TdThread thread; // thread id - SQWorkerPool *pool; + int32_t id; // worker id + int64_t pid; // thread pid + TdThread thread; // thread id + void *pool; } SQWorker; typedef struct SQWorkerPool { @@ -42,6 +43,14 @@ typedef struct SQWorkerPool { TdThreadMutex mutex; } SQWorkerPool; +typedef struct SAutoQWorkerPool { + float ratio; + STaosQset *qset; + const char *name; + SArray *workers; + TdThreadMutex mutex; +} SAutoQWorkerPool; + typedef struct SWWorker { int32_t id; // worker id int64_t pid; // thread pid @@ -65,6 +74,11 @@ void tQWorkerCleanup(SQWorkerPool *pool); STaosQueue *tQWorkerAllocQueue(SQWorkerPool *pool, void *ahandle, FItem fp); void tQWorkerFreeQueue(SQWorkerPool *pool, STaosQueue *queue); +int32_t tAutoQWorkerInit(SAutoQWorkerPool *pool); +void tAutoQWorkerCleanup(SAutoQWorkerPool *pool); +STaosQueue *tAutoQWorkerAllocQueue(SAutoQWorkerPool *pool, void *ahandle, FItem fp); +void tAutoQWorkerFreeQueue(SAutoQWorkerPool *pool, STaosQueue *queue); + int32_t tWWorkerInit(SWWorkerPool *pool); void tWWorkerCleanup(SWWorkerPool *pool); STaosQueue *tWWorkerAllocQueue(SWWorkerPool *pool, void *ahandle, FItems fp); diff --git a/source/common/src/tglobal.c b/source/common/src/tglobal.c index 9caf0cc33e..98b9b566ec 100644 --- a/source/common/src/tglobal.c +++ b/source/common/src/tglobal.c @@ -47,7 +47,7 @@ int32_t tsNumOfMnodeQueryThreads = 4; int32_t tsNumOfMnodeFetchThreads = 1; int32_t tsNumOfMnodeReadThreads = 1; int32_t tsNumOfVnodeQueryThreads = 4; -int32_t tsNumOfVnodeStreamThreads = 2; +float tsRatioOfVnodeStreamThreads = 1.0; int32_t tsNumOfVnodeFetchThreads = 4; int32_t tsNumOfVnodeRsmaThreads = 2; int32_t tsNumOfQnodeQueryThreads = 4; @@ -392,9 +392,7 @@ static int32_t taosAddServerCfg(SConfig *pCfg) { tsNumOfVnodeQueryThreads = TMAX(tsNumOfVnodeQueryThreads, 4); if (cfgAddInt32(pCfg, "numOfVnodeQueryThreads", tsNumOfVnodeQueryThreads, 4, 1024, 0) != 0) return -1; - tsNumOfVnodeStreamThreads = tsNumOfCores / 4; - tsNumOfVnodeStreamThreads = TMAX(tsNumOfVnodeStreamThreads, 4); - if (cfgAddInt32(pCfg, "numOfVnodeStreamThreads", tsNumOfVnodeStreamThreads, 4, 1024, 0) != 0) return -1; + if (cfgAddFloat(pCfg, "ratioOfVnodeStreamThreads", tsRatioOfVnodeStreamThreads, 0.01, 100, 0) != 0) return -1; tsNumOfVnodeFetchThreads = tsNumOfCores / 4; tsNumOfVnodeFetchThreads = TMAX(tsNumOfVnodeFetchThreads, 4); @@ -513,11 +511,9 @@ static int32_t taosUpdateServerCfg(SConfig *pCfg) { pItem->stype = stype; } - pItem = cfgGetItem(tsCfg, "numOfVnodeStreamThreads"); + pItem = cfgGetItem(tsCfg, "ratioOfVnodeStreamThreads"); if (pItem != NULL && pItem->stype == CFG_STYPE_DEFAULT) { - tsNumOfVnodeStreamThreads = numOfCores / 4; - tsNumOfVnodeStreamThreads = TMAX(tsNumOfVnodeStreamThreads, 4); - pItem->i32 = tsNumOfVnodeStreamThreads; + pItem->fval = tsRatioOfVnodeStreamThreads; pItem->stype = stype; } @@ -710,7 +706,7 @@ static int32_t taosSetServerCfg(SConfig *pCfg) { tsNumOfCommitThreads = cfgGetItem(pCfg, "numOfCommitThreads")->i32; tsNumOfMnodeReadThreads = cfgGetItem(pCfg, "numOfMnodeReadThreads")->i32; tsNumOfVnodeQueryThreads = cfgGetItem(pCfg, "numOfVnodeQueryThreads")->i32; - tsNumOfVnodeStreamThreads = cfgGetItem(pCfg, "numOfVnodeStreamThreads")->i32; + tsRatioOfVnodeStreamThreads = cfgGetItem(pCfg, "ratioOfVnodeStreamThreads")->fval; tsNumOfVnodeFetchThreads = cfgGetItem(pCfg, "numOfVnodeFetchThreads")->i32; tsNumOfVnodeRsmaThreads = cfgGetItem(pCfg, "numOfVnodeRsmaThreads")->i32; tsNumOfQnodeQueryThreads = cfgGetItem(pCfg, "numOfQnodeQueryThreads")->i32; @@ -1316,12 +1312,17 @@ void taosSetDebugFlag(int32_t *pFlagPtr, const char *flagName, int32_t flagVal, if (pItem != NULL && (rewrite || pItem->i32 == 0)) { pItem->i32 = flagVal; } - *pFlagPtr = flagVal; + if (pFlagPtr != NULL) { + *pFlagPtr = flagVal; + } } void taosSetAllDebugFlag(int32_t flag, bool rewrite) { if (flag <= 0) return; + taosSetDebugFlag(NULL, "debugFlag", flag, rewrite); + taosSetDebugFlag(NULL, "simDebugFlag", flag, rewrite); + taosSetDebugFlag(NULL, "tmrDebugFlag", flag, rewrite); taosSetDebugFlag(&uDebugFlag, "uDebugFlag", flag, rewrite); taosSetDebugFlag(&rpcDebugFlag, "rpcDebugFlag", flag, rewrite); taosSetDebugFlag(&jniDebugFlag, "jniDebugFlag", flag, rewrite); @@ -1340,6 +1341,5 @@ void taosSetAllDebugFlag(int32_t flag, bool rewrite) { taosSetDebugFlag(&idxDebugFlag, "idxDebugFlag", flag, rewrite); taosSetDebugFlag(&tdbDebugFlag, "tdbDebugFlag", flag, rewrite); taosSetDebugFlag(&metaDebugFlag, "metaDebugFlag", flag, rewrite); - taosSetDebugFlag(&metaDebugFlag, "tmrDebugFlag", flag, rewrite); uInfo("all debug flag are set to %d", flag); } diff --git a/source/dnode/mgmt/exe/dmMain.c b/source/dnode/mgmt/exe/dmMain.c index a8103351b4..a432201413 100644 --- a/source/dnode/mgmt/exe/dmMain.c +++ b/source/dnode/mgmt/exe/dmMain.c @@ -268,6 +268,11 @@ int mainWindows(int argc, char **argv) { if (dmInit() != 0) { dError("failed to init dnode since %s", terrstr()); + + taosCleanupCfg(); + taosCloseLog(); + taosCleanupArgs(); + taosConvDestroy(); return -1; } diff --git a/source/dnode/mgmt/mgmt_vnode/inc/vmInt.h b/source/dnode/mgmt/mgmt_vnode/inc/vmInt.h index b5c554e0ca..6e724f4d43 100644 --- a/source/dnode/mgmt/mgmt_vnode/inc/vmInt.h +++ b/source/dnode/mgmt/mgmt_vnode/inc/vmInt.h @@ -26,20 +26,20 @@ extern "C" { #endif typedef struct SVnodeMgmt { - SDnodeData *pData; - SMsgCb msgCb; - const char *path; - const char *name; - SQWorkerPool queryPool; - SQWorkerPool streamPool; - SWWorkerPool fetchPool; - SSingleWorker mgmtWorker; - SHashObj *hash; - TdThreadRwlock lock; - SVnodesStat state; - STfs *pTfs; - TdThread thread; - bool stop; + SDnodeData *pData; + SMsgCb msgCb; + const char *path; + const char *name; + SQWorkerPool queryPool; + SAutoQWorkerPool streamPool; + SWWorkerPool fetchPool; + SSingleWorker mgmtWorker; + SHashObj *hash; + TdThreadRwlock lock; + SVnodesStat state; + STfs *pTfs; + TdThread thread; + bool stop; } SVnodeMgmt; typedef struct { diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c b/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c index faa94a335d..202dc50ac6 100644 --- a/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c +++ b/source/dnode/mgmt/mgmt_vnode/src/vmWorker.c @@ -140,6 +140,12 @@ static void vmProcessSyncQueue(SQueueInfo *pInfo, STaosQall *qall, int32_t numOf static int32_t vmPutMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg, EQueueType qtype) { const STraceId *trace = &pMsg->info.traceId; + if (pMsg->contLen < sizeof(SMsgHead)) { + dGError("invalid rpc msg with no msg head at pCont. pMsg:%p, type:%s, contLen:%d", pMsg, TMSG_INFO(pMsg->msgType), + pMsg->contLen); + return -1; + } + SMsgHead *pHead = pMsg->pCont; int32_t code = 0; @@ -234,7 +240,7 @@ int32_t vmPutMsgToMgmtQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) { int32_t vmPutRpcMsgToQueue(SVnodeMgmt *pMgmt, EQueueType qtype, SRpcMsg *pRpc) { if (pRpc->contLen < sizeof(SMsgHead)) { - dError("invalid rpc msg since no msg head at pCont. pRpc:%p, type:%s, len:%d", pRpc, TMSG_INFO(pRpc->msgType), + dError("invalid rpc msg with no msg head at pCont. pRpc:%p, type:%s, len:%d", pRpc, TMSG_INFO(pRpc->msgType), pRpc->contLen); rpcFreeCont(pRpc->pCont); pRpc->pCont = NULL; @@ -312,7 +318,7 @@ int32_t vmAllocQueue(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) { (void)tMultiWorkerInit(&pVnode->pApplyW, &acfg); pVnode->pQueryQ = tQWorkerAllocQueue(&pMgmt->queryPool, pVnode, (FItem)vmProcessQueryQueue); - pVnode->pStreamQ = tQWorkerAllocQueue(&pMgmt->streamPool, pVnode, (FItem)vmProcessStreamQueue); + pVnode->pStreamQ = tAutoQWorkerAllocQueue(&pMgmt->streamPool, pVnode, (FItem)vmProcessStreamQueue); pVnode->pFetchQ = tWWorkerAllocQueue(&pMgmt->fetchPool, pVnode, (FItems)vmProcessFetchQueue); if (pVnode->pWriteW.queue == NULL || pVnode->pSyncW.queue == NULL || pVnode->pSyncCtrlW.queue == NULL || @@ -338,7 +344,7 @@ int32_t vmAllocQueue(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) { void vmFreeQueue(SVnodeMgmt *pMgmt, SVnodeObj *pVnode) { tQWorkerFreeQueue(&pMgmt->queryPool, pVnode->pQueryQ); - tQWorkerFreeQueue(&pMgmt->streamPool, pVnode->pStreamQ); + tAutoQWorkerFreeQueue(&pMgmt->streamPool, pVnode->pStreamQ); tWWorkerFreeQueue(&pMgmt->fetchPool, pVnode->pFetchQ); pVnode->pQueryQ = NULL; pVnode->pStreamQ = NULL; @@ -353,11 +359,10 @@ int32_t vmStartWorker(SVnodeMgmt *pMgmt) { pQPool->max = tsNumOfVnodeQueryThreads; if (tQWorkerInit(pQPool) != 0) return -1; - SQWorkerPool *pStreamPool = &pMgmt->streamPool; + SAutoQWorkerPool *pStreamPool = &pMgmt->streamPool; pStreamPool->name = "vnode-stream"; - pStreamPool->min = tsNumOfVnodeStreamThreads; - pStreamPool->max = tsNumOfVnodeStreamThreads; - if (tQWorkerInit(pStreamPool) != 0) return -1; + pStreamPool->ratio = tsRatioOfVnodeStreamThreads; + if (tAutoQWorkerInit(pStreamPool) != 0) return -1; SWWorkerPool *pFPool = &pMgmt->fetchPool; pFPool->name = "vnode-fetch"; @@ -379,7 +384,7 @@ int32_t vmStartWorker(SVnodeMgmt *pMgmt) { void vmStopWorker(SVnodeMgmt *pMgmt) { tQWorkerCleanup(&pMgmt->queryPool); - tQWorkerCleanup(&pMgmt->streamPool); + tAutoQWorkerCleanup(&pMgmt->streamPool); tWWorkerCleanup(&pMgmt->fetchPool); dDebug("vnode workers are closed"); } diff --git a/source/dnode/mgmt/node_mgmt/src/dmTransport.c b/source/dnode/mgmt/node_mgmt/src/dmTransport.c index 756f97d1d0..0ff41d429e 100644 --- a/source/dnode/mgmt/node_mgmt/src/dmTransport.c +++ b/source/dnode/mgmt/node_mgmt/src/dmTransport.c @@ -111,12 +111,12 @@ static void dmProcessRpcMsg(SDnode *pDnode, SRpcMsg *pRpc, SEpSet *pEpSet) { dGError("msg:%p, type:%s pCont is NULL", pRpc, TMSG_INFO(pRpc->msgType)); terrno = TSDB_CODE_INVALID_MSG_LEN; goto _OVER; - } /* else if ((pRpc->code == TSDB_CODE_RPC_NETWORK_UNAVAIL || pRpc->code == TSDB_CODE_RPC_BROKEN_LINK) && - (!IsReq(pRpc)) && (pRpc->pCont == NULL)) { - dGError("msg:%p, type:%s pCont is NULL, err: %s", pRpc, TMSG_INFO(pRpc->msgType), tstrerror(pRpc->code)); - terrno = pRpc->code; - goto _OVER; - }*/ + } else if ((pRpc->code == TSDB_CODE_RPC_NETWORK_UNAVAIL || pRpc->code == TSDB_CODE_RPC_BROKEN_LINK) && + (!IsReq(pRpc)) && (pRpc->pCont == NULL)) { + dGError("msg:%p, type:%s pCont is NULL, err: %s", pRpc, TMSG_INFO(pRpc->msgType), tstrerror(pRpc->code)); + terrno = pRpc->code; + goto _OVER; + } if (pHandle->defaultNtype == NODE_END) { dGError("msg:%p, type:%s not processed since no handle", pRpc, TMSG_INFO(pRpc->msgType)); @@ -248,9 +248,9 @@ static inline void dmReleaseHandle(SRpcHandleInfo *pHandle, int8_t type) { rpcRe static bool rpcRfp(int32_t code, tmsg_t msgType) { if (code == TSDB_CODE_RPC_NETWORK_UNAVAIL || code == TSDB_CODE_RPC_BROKEN_LINK || code == TSDB_CODE_MNODE_NOT_FOUND || - code == TSDB_CODE_RPC_SOMENODE_NOT_CONNECTED || - code == TSDB_CODE_SYN_NOT_LEADER || code == TSDB_CODE_SYN_RESTORING || code == TSDB_CODE_VND_STOPPED || - code == TSDB_CODE_APP_IS_STARTING || code == TSDB_CODE_APP_IS_STOPPING) { + code == TSDB_CODE_RPC_SOMENODE_NOT_CONNECTED || code == TSDB_CODE_SYN_NOT_LEADER || + code == TSDB_CODE_SYN_RESTORING || code == TSDB_CODE_VND_STOPPED || code == TSDB_CODE_APP_IS_STARTING || + code == TSDB_CODE_APP_IS_STOPPING) { if (msgType == TDMT_SCH_QUERY || msgType == TDMT_SCH_MERGE_QUERY || msgType == TDMT_SCH_FETCH || msgType == TDMT_SCH_MERGE_FETCH) { return false; diff --git a/source/dnode/mnode/impl/inc/mndDef.h b/source/dnode/mnode/impl/inc/mndDef.h index 4e93a1d96e..2f824b48b4 100644 --- a/source/dnode/mnode/impl/inc/mndDef.h +++ b/source/dnode/mnode/impl/inc/mndDef.h @@ -193,6 +193,7 @@ typedef struct { int64_t lastAccessTime; int32_t accessTimes; int32_t numOfVnodes; + int32_t numOfOtherNodes; int32_t numOfSupportVnodes; float numOfCores; int64_t memTotal; diff --git a/source/dnode/mnode/impl/src/mndDnode.c b/source/dnode/mnode/impl/src/mndDnode.c index 58ae85a628..d7b16c2c8e 100644 --- a/source/dnode/mnode/impl/src/mndDnode.c +++ b/source/dnode/mnode/impl/src/mndDnode.c @@ -397,8 +397,6 @@ static int32_t mndProcessStatusReq(SRpcMsg *pReq) { bool reboot = (pDnode->rebootTime != statusReq.rebootTime); bool needCheck = !online || dnodeChanged || reboot; - pDnode->accessTimes++; - pDnode->lastAccessTime = curMs; const STraceId *trace = &pReq->info.traceId; mGTrace("dnode:%d, status received, accessTimes:%d check:%d online:%d reboot:%d changed:%d statusSeq:%d", pDnode->id, pDnode->accessTimes, needCheck, online, reboot, dnodeChanged, statusReq.statusSeq); @@ -534,6 +532,8 @@ static int32_t mndProcessStatusReq(SRpcMsg *pReq) { pReq->info.rsp = pHead; } + pDnode->accessTimes++; + pDnode->lastAccessTime = curMs; code = 0; _OVER: diff --git a/source/dnode/mnode/impl/src/mndVgroup.c b/source/dnode/mnode/impl/src/mndVgroup.c index 31ab1f3259..2550c68cfb 100644 --- a/source/dnode/mnode/impl/src/mndVgroup.c +++ b/source/dnode/mnode/impl/src/mndVgroup.c @@ -425,6 +425,7 @@ void *mndBuildDropVnodeReq(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVgOb static bool mndResetDnodesArrayFp(SMnode *pMnode, void *pObj, void *p1, void *p2, void *p3) { SDnodeObj *pDnode = pObj; pDnode->numOfVnodes = 0; + pDnode->numOfOtherNodes = 0; return true; } @@ -447,7 +448,7 @@ static bool mndBuildDnodesArrayFp(SMnode *pMnode, void *pObj, void *p1, void *p2 pDnode->numOfVnodes, pDnode->numOfSupportVnodes, isMnode, online, pDnode->memAvail, pDnode->memUsed); if (isMnode) { - pDnode->numOfVnodes++; + pDnode->numOfOtherNodes++; } if (online && pDnode->numOfSupportVnodes > 0) { @@ -468,14 +469,25 @@ SArray *mndBuildDnodesArray(SMnode *pMnode, int32_t exceptDnodeId) { sdbTraverse(pSdb, SDB_DNODE, mndResetDnodesArrayFp, NULL, NULL, NULL); sdbTraverse(pSdb, SDB_DNODE, mndBuildDnodesArrayFp, pArray, &exceptDnodeId, NULL); + + mDebug("build %d dnodes array", (int32_t)taosArrayGetSize(pArray)); + for (int32_t i = 0; i < (int32_t)taosArrayGetSize(pArray); ++i) { + SDnodeObj *pDnode = taosArrayGet(pArray, i); + mDebug("dnode:%d, vnodes:%d others:%d", pDnode->id, pDnode->numOfVnodes, pDnode->numOfOtherNodes); + } return pArray; } static int32_t mndCompareDnodeId(int32_t *dnode1Id, int32_t *dnode2Id) { return *dnode1Id >= *dnode2Id ? 1 : 0; } +static float mndGetDnodeScore(SDnodeObj *pDnode, int32_t additionDnodes, float ratio) { + float totalDnodes = pDnode->numOfVnodes + (float)pDnode->numOfOtherNodes * ratio + additionDnodes; + return totalDnodes / pDnode->numOfSupportVnodes; +} + static int32_t mndCompareDnodeVnodes(SDnodeObj *pDnode1, SDnodeObj *pDnode2) { - float d1Score = (float)pDnode1->numOfVnodes / pDnode1->numOfSupportVnodes; - float d2Score = (float)pDnode2->numOfVnodes / pDnode2->numOfSupportVnodes; + float d1Score = mndGetDnodeScore(pDnode1, 0, 0.9); + float d2Score = mndGetDnodeScore(pDnode2, 0, 0.9); return d1Score >= d2Score ? 1 : 0; } @@ -494,7 +506,12 @@ static int32_t mndGetAvailableDnode(SMnode *pMnode, SDbObj *pDb, SVgObj *pVgroup int32_t allocedVnodes = 0; void *pIter = NULL; + mDebug("start to sort %d dnodes", (int32_t)taosArrayGetSize(pArray)); taosArraySort(pArray, (__compar_fn_t)mndCompareDnodeVnodes); + for (int32_t i = 0; i < (int32_t)taosArrayGetSize(pArray); ++i) { + SDnodeObj *pDnode = taosArrayGet(pArray, i); + mDebug("dnode:%d, score:%f", pDnode->id, mndGetDnodeScore(pDnode, 0, 0.9)); + } int32_t size = taosArrayGetSize(pArray); if (size < pVgroup->replica) { @@ -875,7 +892,7 @@ static int32_t mndAddVnodeToVgroup(SMnode *pMnode, STrans *pTrans, SVgObj *pVgro taosArraySort(pArray, (__compar_fn_t)mndCompareDnodeVnodes); for (int32_t i = 0; i < taosArrayGetSize(pArray); ++i) { SDnodeObj *pDnode = taosArrayGet(pArray, i); - mInfo("dnode:%d, equivalent vnodes:%d", pDnode->id, pDnode->numOfVnodes); + mInfo("dnode:%d, equivalent vnodes:%d others:%d", pDnode->id, pDnode->numOfVnodes, pDnode->numOfOtherNodes); } SVnodeGid *pVgid = &pVgroup->vnodeGid[pVgroup->replica]; @@ -935,7 +952,7 @@ static int32_t mndRemoveVnodeFromVgroup(SMnode *pMnode, STrans *pTrans, SVgObj * taosArraySort(pArray, (__compar_fn_t)mndCompareDnodeVnodes); for (int32_t i = 0; i < taosArrayGetSize(pArray); ++i) { SDnodeObj *pDnode = taosArrayGet(pArray, i); - mInfo("dnode:%d, equivalent vnodes:%d", pDnode->id, pDnode->numOfVnodes); + mInfo("dnode:%d, equivalent vnodes:%d others:%d", pDnode->id, pDnode->numOfVnodes, pDnode->numOfOtherNodes); } int32_t code = -1; @@ -1970,16 +1987,16 @@ static int32_t mndBalanceVgroup(SMnode *pMnode, SRpcMsg *pReq, SArray *pArray) { taosArraySort(pArray, (__compar_fn_t)mndCompareDnodeVnodes); for (int32_t i = 0; i < taosArrayGetSize(pArray); ++i) { SDnodeObj *pDnode = taosArrayGet(pArray, i); - mInfo("dnode:%d, equivalent vnodes:%d support:%d, score:%f", pDnode->id, pDnode->numOfVnodes, - pDnode->numOfSupportVnodes, (float)pDnode->numOfVnodes / pDnode->numOfSupportVnodes); + mInfo("dnode:%d, equivalent vnodes:%d others:%d support:%d, score:%f", pDnode->id, pDnode->numOfVnodes, + pDnode->numOfSupportVnodes, pDnode->numOfOtherNodes, mndGetDnodeScore(pDnode, 0, 1)); } SDnodeObj *pSrc = taosArrayGet(pArray, taosArrayGetSize(pArray) - 1); SDnodeObj *pDst = taosArrayGet(pArray, 0); - float srcScore = (float)(pSrc->numOfVnodes - 1) / pSrc->numOfSupportVnodes; - float dstScore = (float)(pDst->numOfVnodes + 1) / pDst->numOfSupportVnodes; - mInfo("trans:%d, after balance, src dnode:%d score:%f, dst dnode:%d score:%f", pTrans->id, pSrc->id, srcScore, + float srcScore = mndGetDnodeScore(pSrc, -1, 1); + float dstScore = mndGetDnodeScore(pDst, 1, 1); + mInfo("trans:%d, after balance, src dnode:%d score:%f, dst dnode:%d score:%f", pTrans->id, pSrc->id, dstScore, pDst->id, dstScore); if (srcScore > dstScore - 0.000001) { diff --git a/source/dnode/vnode/src/meta/metaCache.c b/source/dnode/vnode/src/meta/metaCache.c index 513ee5a1c2..d68658b0d9 100644 --- a/source/dnode/vnode/src/meta/metaCache.c +++ b/source/dnode/vnode/src/meta/metaCache.c @@ -57,7 +57,6 @@ struct SMetaCache { TdThreadMutex lock; SHashObj* pTableEntry; SLRUCache* pUidResCache; - uint64_t keyBuf[3]; } sTagFilterResCache; }; @@ -429,20 +428,20 @@ int32_t metaGetCachedTableUidList(SMeta* pMeta, tb_uid_t suid, const uint8_t* pK bool* acquireRes) { // generate the composed key for LRU cache SLRUCache* pCache = pMeta->pCache->sTagFilterResCache.pUidResCache; - uint64_t* pBuf = pMeta->pCache->sTagFilterResCache.keyBuf; SHashObj* pTableMap = pMeta->pCache->sTagFilterResCache.pTableEntry; TdThreadMutex* pLock = &pMeta->pCache->sTagFilterResCache.lock; + uint64_t buf[3] = {0}; uint32_t times = 0; *acquireRes = 0; - pBuf[0] = suid; - memcpy(&pBuf[1], pKey, keyLen); + buf[0] = suid; + memcpy(&buf[1], pKey, keyLen); taosThreadMutexLock(pLock); int32_t len = keyLen + sizeof(uint64_t); - LRUHandle* pHandle = taosLRUCacheLookup(pCache, pBuf, len); + LRUHandle* pHandle = taosLRUCacheLookup(pCache, buf, len); if (pHandle == NULL) { taosThreadMutexUnlock(pLock); return TSDB_CODE_SUCCESS; @@ -476,10 +475,10 @@ int32_t metaGetCachedTableUidList(SMeta* pMeta, tb_uid_t suid, const uint8_t* pK SListNode* pNode = NULL; while ((pNode = tdListNext(&iter)) != NULL) { - memcpy(&pBuf[1], pNode->data, keyLen); + memcpy(&buf[1], pNode->data, keyLen); // check whether it is existed in LRU cache, and remove it from linked list if not. - LRUHandle* pRes = taosLRUCacheLookup(pCache, pBuf, len); + LRUHandle* pRes = taosLRUCacheLookup(pCache, buf, len); if (pRes == NULL) { // remove the item in the linked list taosArrayPush(pInvalidRes, &pNode); } else { @@ -547,14 +546,14 @@ int32_t metaUidFilterCachePut(SMeta* pMeta, uint64_t suid, const void* pKey, int tdListAppend(&(*pEntry)->list, pKey); } - uint64_t* pBuf = pMeta->pCache->sTagFilterResCache.keyBuf; - pBuf[0] = suid; + uint64_t buf[3] = {0}; + buf[0] = suid; - memcpy(&pBuf[1], pKey, keyLen); + memcpy(&buf[1], pKey, keyLen); ASSERT(sizeof(uint64_t) + keyLen == 24); // add to cache. - taosLRUCacheInsert(pCache, pBuf, sizeof(uint64_t) + keyLen, pPayload, payloadLen, freePayload, NULL, + taosLRUCacheInsert(pCache, buf, sizeof(uint64_t) + keyLen, pPayload, payloadLen, freePayload, NULL, TAOS_LRU_PRIORITY_LOW); taosThreadMutexUnlock(pLock); diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c index 045b497371..e366795fd3 100644 --- a/source/dnode/vnode/src/tq/tq.c +++ b/source/dnode/vnode/src/tq/tq.c @@ -1004,8 +1004,10 @@ int32_t tqProcessStreamTaskCheckReq(STQ* pTq, SRpcMsg* pMsg) { int32_t len; tEncodeSize(tEncodeSStreamTaskCheckRsp, &rsp, len, code); if (code < 0) { - ASSERT(0); + tqDebug("tq encode stream check rsp error"); + return -1; } + void* buf = rpcMallocCont(sizeof(SMsgHead) + len); ((SMsgHead*)buf)->vgId = htonl(req.upstreamNodeId); diff --git a/source/dnode/vnode/src/tsdb/tsdbMergeTree.c b/source/dnode/vnode/src/tsdb/tsdbMergeTree.c index a5ad18a22d..af1a42d018 100644 --- a/source/dnode/vnode/src/tsdb/tsdbMergeTree.c +++ b/source/dnode/vnode/src/tsdb/tsdbMergeTree.c @@ -567,7 +567,6 @@ int32_t tMergeTreeOpen(SMergeTree *pMTree, int8_t backward, SDataFReader *pFRead pMTree->pLoadInfo = pBlockLoadInfo; pMTree->destroyLoadInfo = destroyLoadInfo; - ASSERT(pMTree->pLoadInfo != NULL); for (int32_t i = 0; i < pFReader->pSet->nSttF; ++i) { // open all last file struct SLDataIter *pIter = NULL; diff --git a/source/dnode/vnode/src/tsdb/tsdbRead.c b/source/dnode/vnode/src/tsdb/tsdbRead.c index 6634340d75..cb31890573 100644 --- a/source/dnode/vnode/src/tsdb/tsdbRead.c +++ b/source/dnode/vnode/src/tsdb/tsdbRead.c @@ -243,7 +243,7 @@ static int32_t setColumnIdSlotList(SBlockLoadSuppInfo* pSupInfo, SColumnInfo* pC return TSDB_CODE_SUCCESS; } -static void updateBlockSMAInfo(STSchema* pSchema, SBlockLoadSuppInfo* pSupInfo) { +static int32_t updateBlockSMAInfo(STSchema* pSchema, SBlockLoadSuppInfo* pSupInfo) { int32_t i = 0, j = 0; while(i < pSchema->numOfCols && j < pSupInfo->numOfCols) { @@ -251,7 +251,7 @@ static void updateBlockSMAInfo(STSchema* pSchema, SBlockLoadSuppInfo* pSupInfo) if (pTCol->colId == pSupInfo->colId[j]) { if (!IS_BSMA_ON(pTCol)) { pSupInfo->smaValid = false; - return; + return TSDB_CODE_SUCCESS; } i += 1; @@ -260,9 +260,11 @@ static void updateBlockSMAInfo(STSchema* pSchema, SBlockLoadSuppInfo* pSupInfo) // do nothing i += 1; } else { - ASSERT(0); + return TSDB_CODE_INVALID_PARA; } } + + return TSDB_CODE_SUCCESS; } static int32_t initBlockScanInfoBuf(SBlockInfoBuf* pBuf, int32_t numOfTables) { @@ -579,7 +581,7 @@ static int32_t tsdbReaderCreate(SVnode* pVnode, SQueryTableDataCond* pCond, STsd } if (VND_IS_TSMA(pVnode)) { - tsdbDebug("vgId:%d, tsma is selected to query", TD_VID(pVnode)); + tsdbDebug("vgId:%d, tsma is selected to query, %s", TD_VID(pVnode), idstr); } initReaderStatus(&pReader->status); @@ -594,7 +596,6 @@ static int32_t tsdbReaderCreate(SVnode* pVnode, SQueryTableDataCond* pCond, STsd pReader->type = pCond->type; pReader->window = updateQueryTimeWindow(pReader->pTsdb, &pCond->twindows); pReader->blockInfoBuf.numPerBucket = 1000; // 1000 tables per bucket - ASSERT(pCond->numOfCols > 0); if (pReader->pResBlock == NULL) { pReader->freeBlock = true; @@ -605,6 +606,12 @@ static int32_t tsdbReaderCreate(SVnode* pVnode, SQueryTableDataCond* pCond, STsd } } + if (pCond->numOfCols <= 0) { + tsdbError("vgId:%d, invalid column number %d in query cond, %s", TD_VID(pVnode), pCond->numOfCols, idstr); + code = TSDB_CODE_INVALID_PARA; + goto _end; + } + // todo refactor. limitOutputBufferSize(pCond, &pReader->capacity); @@ -794,8 +801,9 @@ static void doCopyColVal(SColumnInfoData* pColInfoData, int32_t rowIndex, int32_ } static SFileDataBlockInfo* getCurrentBlockInfo(SDataBlockIter* pBlockIter) { - if (taosArrayGetSize(pBlockIter->blockList) == 0) { - ASSERT(pBlockIter->numOfBlocks == taosArrayGetSize(pBlockIter->blockList)); + size_t num = taosArrayGetSize(pBlockIter->blockList); + if (num == 0) { + ASSERT(pBlockIter->numOfBlocks == num); return NULL; } @@ -805,73 +813,6 @@ static SFileDataBlockInfo* getCurrentBlockInfo(SDataBlockIter* pBlockIter) { static SDataBlk* getCurrentBlock(SDataBlockIter* pBlockIter) { return &pBlockIter->block; } -int32_t binarySearchForTs(char* pValue, int num, TSKEY key, int order) { - int32_t midPos = -1; - int32_t numOfRows; - - ASSERT(order == TSDB_ORDER_ASC || order == TSDB_ORDER_DESC); - - TSKEY* keyList = (TSKEY*)pValue; - int32_t firstPos = 0; - int32_t lastPos = num - 1; - - if (order == TSDB_ORDER_DESC) { - // find the first position which is smaller than the key - while (1) { - if (key >= keyList[firstPos]) return firstPos; - if (key == keyList[lastPos]) return lastPos; - - if (key < keyList[lastPos]) { - lastPos += 1; - if (lastPos >= num) { - return -1; - } else { - return lastPos; - } - } - - numOfRows = lastPos - firstPos + 1; - midPos = (numOfRows >> 1) + firstPos; - - if (key < keyList[midPos]) { - firstPos = midPos + 1; - } else if (key > keyList[midPos]) { - lastPos = midPos - 1; - } else { - break; - } - } - - } else { - // find the first position which is bigger than the key - while (1) { - if (key <= keyList[firstPos]) return firstPos; - if (key == keyList[lastPos]) return lastPos; - - if (key > keyList[lastPos]) { - lastPos = lastPos + 1; - if (lastPos >= num) - return -1; - else - return lastPos; - } - - numOfRows = lastPos - firstPos + 1; - midPos = (numOfRows >> 1u) + firstPos; - - if (key < keyList[midPos]) { - lastPos = midPos - 1; - } else if (key > keyList[midPos]) { - firstPos = midPos + 1; - } else { - break; - } - } - } - - return midPos; -} - static int doBinarySearchKey(TSKEY* keyList, int num, int pos, TSKEY key, int order) { // start end position int s, e; @@ -972,8 +913,8 @@ static void copyNumericCols(const SColData* pData, SFileBlockDumpInfo* pDumpInfo int32_t step = asc? 1:-1; - // make sure it is aligned to 8bit - ASSERT((((uint64_t)pColData->pData) & (0x8 - 1)) == 0); + // make sure it is aligned to 8bit, the allocated memory address is aligned to 256bit +// ASSERT((((uint64_t)pColData->pData) & (0x8 - 1)) == 0); // 1. copy data in a batch model memcpy(pColData->pData, p, dumpedRows * tDataTypes[pData->type].bytes); @@ -1072,11 +1013,20 @@ static int32_t copyBlockDataToSDataBlock(STsdbReader* pReader, STableBlockScanIn // pDumpInfo->rowIndex = 0; } else if (!asc && pReader->window.ekey >= pBlock->maxKey.ts) { // pDumpInfo->rowIndex = pBlock->nRow - 1; - } else { + } else { // find the appropriate the start position in current block, and set it to be the current rowIndex int32_t pos = asc ? pBlock->nRow - 1 : 0; int32_t order = asc ? TSDB_ORDER_DESC : TSDB_ORDER_ASC; int64_t key = asc ? pReader->window.skey : pReader->window.ekey; pDumpInfo->rowIndex = doBinarySearchKey(pBlockData->aTSKEY, pBlock->nRow, pos, key, order); + + if (pDumpInfo->rowIndex < 0) { + tsdbError( + "%p failed to locate the start position in current block, global index:%d, table index:%d, brange:%" PRId64 + "-%" PRId64 ", minVer:%" PRId64 ", maxVer:%" PRId64 " %s", + pReader, pBlockIter->index, pBlockInfo->tbBlockIdx, pBlock->minKey.ts, pBlock->maxKey.ts, pBlock->minVer, + pBlock->maxVer, pReader->idStr); + return TSDB_CODE_INVALID_PARA; + } } } @@ -1183,7 +1133,6 @@ static int32_t doLoadFileBlockData(STsdbReader* pReader, SDataBlockIter* pBlockI SFileDataBlockInfo* pBlockInfo = getCurrentBlockInfo(pBlockIter); SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo; - ASSERT(pBlockInfo != NULL); SDataBlk* pBlock = getCurrentBlock(pBlockIter); code = tsdbReadDataBlock(pReader->pFileReader, pBlock, pBlockData); @@ -1221,8 +1170,6 @@ static void cleanupBlockOrderSupporter(SBlockOrderSupporter* pSup) { } static int32_t initBlockOrderSupporter(SBlockOrderSupporter* pSup, int32_t numOfTables) { - ASSERT(numOfTables >= 1); - pSup->numOfBlocksPerTable = taosMemoryCalloc(1, sizeof(int32_t) * numOfTables); pSup->indexPerTable = taosMemoryCalloc(1, sizeof(int32_t) * numOfTables); pSup->pDataBlockInfo = taosMemoryCalloc(1, POINTER_BYTES * numOfTables); @@ -1329,7 +1276,10 @@ static int32_t initBlockIterator(STsdbReader* pReader, SDataBlockIter* pBlockIte sup.numOfTables += 1; } - ASSERT(numOfBlocks == cnt); + if (numOfBlocks != cnt && sup.numOfTables != numOfTables) { + cleanupBlockOrderSupporter(&sup); + return TSDB_CODE_INVALID_PARA; + } // since there is only one table qualified, blocks are not sorted if (sup.numOfTables == 1) { @@ -1351,10 +1301,9 @@ static int32_t initBlockIterator(STsdbReader* pReader, SDataBlockIter* pBlockIte tsdbDebug("%p create data blocks info struct completed, %d blocks in %d tables %s", pReader, cnt, sup.numOfTables, pReader->idStr); - ASSERT(cnt <= numOfBlocks && sup.numOfTables <= numOfTables); - SMultiwayMergeTreeInfo* pTree = NULL; - uint8_t ret = tMergeTreeCreate(&pTree, sup.numOfTables, &sup, fileDataBlockOrderCompar); + + uint8_t ret = tMergeTreeCreate(&pTree, sup.numOfTables, &sup, fileDataBlockOrderCompar); if (ret != TSDB_CODE_SUCCESS) { cleanupBlockOrderSupporter(&sup); return TSDB_CODE_OUT_OF_MEMORY; @@ -1432,8 +1381,6 @@ static bool getNeighborBlockOfSameTable(SFileDataBlockInfo* pBlockInfo, STableBl } static int32_t findFileBlockInfoIndex(SDataBlockIter* pBlockIter, SFileDataBlockInfo* pFBlockInfo) { - ASSERT(pBlockIter != NULL && pFBlockInfo != NULL); - int32_t step = ASCENDING_TRAVERSE(pBlockIter->order) ? 1 : -1; int32_t index = pBlockIter->index; @@ -1924,7 +1871,6 @@ static int32_t doMergeFileBlockAndLastBlock(SLastBlockReader* pLastBlockReader, } doMergeRowsInLastBlock(pLastBlockReader, pBlockScanInfo, tsLastBlock, &merge, &pReader->verRange); - ASSERT(mergeBlockData); // merge with block data if ts == key if (tsLastBlock == pBlockData->aTSKEY[pDumpInfo->rowIndex]) { @@ -1990,7 +1936,6 @@ static int32_t mergeFileBlockAndLastBlock(STsdbReader* pReader, SLastBlockReader tRowMergerClear(&merge); return code; } else { - ASSERT(0); return TSDB_CODE_SUCCESS; } } else { // desc order @@ -2011,7 +1956,6 @@ static int32_t doMergeMultiLevelRows(STsdbReader* pReader, STableBlockScanInfo* TSDBROW* pRow = getValidMemRow(&pBlockScanInfo->iter, pDelList, pReader); TSDBROW* piRow = getValidMemRow(&pBlockScanInfo->iiter, pDelList, pReader); - ASSERT(pRow != NULL && piRow != NULL); int64_t tsLast = INT64_MIN; if (hasDataInLastBlock(pLastBlockReader)) { @@ -2235,7 +2179,6 @@ static int32_t initMemDataIterator(STableBlockScanInfo* pBlockScanInfo, STsdbRea if (pReader->pReadSnap->pMem != NULL) { d = tsdbGetTbDataFromMemTable(pReader->pReadSnap->pMem, pReader->suid, pBlockScanInfo->uid); if (d != NULL) { - ASSERT(pBlockScanInfo->iter.iter == NULL); code = tsdbTbDataIterCreate(d, &startKey, backward, &pBlockScanInfo->iter.iter); if (code == TSDB_CODE_SUCCESS) { pBlockScanInfo->iter.hasVal = (tsdbTbDataIterGet(pBlockScanInfo->iter.iter) != NULL); @@ -2349,10 +2292,9 @@ static int64_t getCurrentKeyInLastBlock(SLastBlockReader* pLastBlockReader) { static bool hasDataInLastBlock(SLastBlockReader* pLastBlockReader) { return pLastBlockReader->mergeTree.pIter != NULL; } bool hasDataInFileBlock(const SBlockData* pBlockData, const SFileBlockDumpInfo* pDumpInfo) { - if (pBlockData->nRow > 0) { - ASSERT(pBlockData->nRow == pDumpInfo->totalRows); + if ((pBlockData->nRow > 0) && (pBlockData->nRow != pDumpInfo->totalRows)) { + return false; // this is an invalid result. } - return pBlockData->nRow > 0 && (!pDumpInfo->allDumped); } @@ -2583,7 +2525,6 @@ int32_t initDelSkylineIterator(STableBlockScanInfo* pBlockScanInfo, STsdbReader* int32_t code = 0; SArray* pDelData = taosArrayInit(4, sizeof(SDelData)); - ASSERT(pReader->pReadSnap != NULL); SDelFile* pDelFile = pReader->pReadSnap->fs.pDelFile; if (pDelFile && taosArrayGetSize(pReader->pDelIdx) > 0) { @@ -2868,7 +2809,6 @@ static int32_t doBuildDataBlock(STsdbReader* pReader) { TSDBKEY keyInBuf = getCurrentKeyInBuf(pScanInfo, pReader); if (pBlockInfo == NULL) { // build data block from last data file - ASSERT(pBlockIter->numOfBlocks == 0); code = buildComposedDataBlock(pReader); } else if (fileBlockShouldLoad(pReader, pBlockInfo, pBlock, pScanInfo, keyInBuf, pLastBlockReader)) { code = doLoadFileBlockData(pReader, pBlockIter, &pStatus->fileBlockData, pScanInfo->uid); @@ -3837,15 +3777,16 @@ int32_t tsdbReaderOpen(SVnode* pVnode, SQueryTableDataCond* pCond, void* pTableL } if (pReader->pSchema != NULL) { - updateBlockSMAInfo(pReader->pSchema, &pReader->suppInfo); + code = updateBlockSMAInfo(pReader->pSchema, &pReader->suppInfo); + if (code != TSDB_CODE_SUCCESS) { + goto _err; + } } STsdbReader* p = (pReader->innerReader[0] != NULL) ? pReader->innerReader[0] : pReader; pReader->status.pTableMap = createDataBlockScanInfo(p, &pReader->blockInfoBuf, pTableList, numOfTables); if (pReader->status.pTableMap == NULL) { - tsdbReaderClose(p); *ppReader = NULL; - code = TSDB_CODE_OUT_OF_MEMORY; goto _err; } @@ -4113,25 +4054,27 @@ static void doFillNullColSMA(SBlockLoadSuppInfo* pSup, int32_t numOfRows, int32_ } int32_t tsdbRetrieveDatablockSMA(STsdbReader* pReader, SSDataBlock* pDataBlock, bool* allHave) { + SColumnDataAgg*** pBlockSMA = &pDataBlock->pBlockAgg; + int32_t code = 0; - SColumnDataAgg ***pBlockSMA = &pDataBlock->pBlockAgg; *allHave = false; + *pBlockSMA = NULL; if (pReader->type == TIMEWINDOW_RANGE_EXTERNAL) { - *pBlockSMA = NULL; return TSDB_CODE_SUCCESS; } // there is no statistics data for composed block if (pReader->status.composedDataBlock || (!pReader->suppInfo.smaValid)) { - *pBlockSMA = NULL; return TSDB_CODE_SUCCESS; } SFileDataBlockInfo* pFBlock = getCurrentBlockInfo(&pReader->status.blockIter); SBlockLoadSuppInfo* pSup = &pReader->suppInfo; - ASSERT(pReader->pResBlock->info.id.uid == pFBlock->uid); + if (pReader->pResBlock->info.id.uid != pFBlock->uid) { + return TSDB_CODE_SUCCESS; + } SDataBlk* pBlock = getCurrentBlock(&pReader->status.blockIter); if (tDataBlkHasSma(pBlock)) { @@ -4159,13 +4102,10 @@ int32_t tsdbRetrieveDatablockSMA(STsdbReader* pReader, SSDataBlock* pDataBlock, // update the number of NULL data rows size_t numOfCols = pSup->numOfCols; - int32_t i = 0, j = 0; - size_t size = taosArrayGetSize(pSup->pColAgg); - // ensure capacity - if(pDataBlock->pDataBlock) { - size_t colsNum = taosArrayGetSize(pDataBlock->pDataBlock); - taosArrayEnsureCap(pSup->pColAgg, colsNum); + if (pDataBlock->pDataBlock) { + size_t colsNum = taosArrayGetSize(pDataBlock->pDataBlock); + taosArrayEnsureCap(pSup->pColAgg, colsNum); } SSDataBlock* pResBlock = pReader->pResBlock; @@ -4176,8 +4116,9 @@ int32_t tsdbRetrieveDatablockSMA(STsdbReader* pReader, SSDataBlock* pDataBlock, // do fill all null column value SMA info doFillNullColSMA(pSup, pBlock->nRow, numOfCols, pTsAgg); + size_t size = taosArrayGetSize(pSup->pColAgg); - i = 0, j = 0; + int32_t i = 0, j = 0; while (j < numOfCols && i < size) { SColumnDataAgg* pAgg = taosArrayGet(pSup->pColAgg, i); if (pAgg->colId == pSup->colId[j]) { @@ -4187,7 +4128,7 @@ int32_t tsdbRetrieveDatablockSMA(STsdbReader* pReader, SSDataBlock* pDataBlock, } else if (pAgg->colId < pSup->colId[j]) { i += 1; } else if (pSup->colId[j] < pAgg->colId) { - ASSERT(pSup->colId[j] == PRIMARYKEY_TIMESTAMP_COL_ID); + // ASSERT(pSup->colId[j] == PRIMARYKEY_TIMESTAMP_COL_ID); pResBlock->pBlockAgg[pSup->slotId[j]] = &pSup->tsColAgg; j += 1; } @@ -4420,9 +4361,12 @@ int32_t tsdbGetTableSchema(SVnode* pVnode, int64_t uid, STSchema** pSchema, int6 return terrno; } sversion = mr.me.stbEntry.schemaRow.version; - } else { - ASSERT(mr.me.type == TSDB_NORMAL_TABLE); + } else if (mr.me.type == TSDB_NORMAL_TABLE) { sversion = mr.me.ntbEntry.schemaRow.version; + } else { + terrno = TSDB_CODE_INVALID_PARA; + metaReaderClear(&mr); + return terrno; } metaReaderClear(&mr); diff --git a/source/dnode/vnode/src/vnd/vnodeSvr.c b/source/dnode/vnode/src/vnd/vnodeSvr.c index 048677bfbb..5dda7c9936 100644 --- a/source/dnode/vnode/src/vnd/vnodeSvr.c +++ b/source/dnode/vnode/src/vnd/vnodeSvr.c @@ -197,7 +197,7 @@ int32_t vnodeProcessWriteMsg(SVnode *pVnode, SRpcMsg *pMsg, int64_t version, SRp if (!syncUtilUserCommit(pMsg->msgType)) goto _exit; - if (pMsg->msgType == TDMT_VND_STREAM_RECOVER_BLOCKING_STAGE) { + if (pMsg->msgType == TDMT_VND_STREAM_RECOVER_BLOCKING_STAGE || pMsg->msgType == TDMT_STREAM_TASK_CHECK_RSP) { if (tqCheckLogInWal(pVnode->pTq, version)) return 0; } diff --git a/source/libs/executor/inc/executorimpl.h b/source/libs/executor/inc/executorimpl.h index 647da78a78..e3b4b57052 100644 --- a/source/libs/executor/inc/executorimpl.h +++ b/source/libs/executor/inc/executorimpl.h @@ -136,6 +136,7 @@ typedef struct { SSchemaWrapper* schema; char tbName[TSDB_TABLE_NAME_LEN]; int8_t recoverStep; + int8_t recoverScanFinished; SQueryTableDataCond tableCond; int64_t fillHistoryVer1; int64_t fillHistoryVer2; @@ -182,7 +183,7 @@ struct SExecTaskInfo { SSubplan* pSubplan; struct SOperatorInfo* pRoot; SLocalFetch localFetch; - SArray* pResultBlockList;// result block list + SArray* pResultBlockList; // result block list STaskStopInfo stopInfo; }; @@ -199,7 +200,7 @@ typedef struct SOperatorFpSet { __optr_fn_t getNextFn; __optr_fn_t cleanupFn; // call this function to release the allocated resources ASAP __optr_close_fn_t closeFn; - __optr_reqBuf_fn_t reqBufFn; // total used buffer for blocking operator + __optr_reqBuf_fn_t reqBufFn; // total used buffer for blocking operator __optr_encode_fn_t encodeResultRow; __optr_decode_fn_t decodeResultRow; __optr_explain_fn_t getExplainFn; @@ -255,22 +256,22 @@ typedef struct SLimitInfo { } SLimitInfo; typedef struct SExchangeInfo { - SArray* pSources; - SArray* pSourceDataInfo; - tsem_t ready; - void* pTransporter; + SArray* pSources; + SArray* pSourceDataInfo; + tsem_t ready; + void* pTransporter; // SArray, result block list, used to keep the multi-block that // passed by downstream operator - SArray* pResultBlockList; - SArray* pRecycledBlocks;// build a pool for small data block to avoid to repeatly create and then destroy. - SSDataBlock* pDummyBlock; // dummy block, not keep data - bool seqLoadData; // sequential load data or not, false by default - int32_t current; + SArray* pResultBlockList; + SArray* pRecycledBlocks; // build a pool for small data block to avoid to repeatly create and then destroy. + SSDataBlock* pDummyBlock; // dummy block, not keep data + bool seqLoadData; // sequential load data or not, false by default + int32_t current; SLoadRemoteDataInfo loadInfo; uint64_t self; SLimitInfo limitInfo; - int64_t openedTs; // start exec time stamp, todo: move to SLoadRemoteDataInfo + int64_t openedTs; // start exec time stamp, todo: move to SLoadRemoteDataInfo } SExchangeInfo; typedef struct SScanInfo { @@ -305,9 +306,9 @@ typedef struct { } SAggOptrPushDownInfo; typedef struct STableMetaCacheInfo { - SLRUCache* pTableMetaEntryCache; // 100 by default - uint64_t metaFetch; - uint64_t cacheHit; + SLRUCache* pTableMetaEntryCache; // 100 by default + uint64_t metaFetch; + uint64_t cacheHit; } STableMetaCacheInfo; typedef struct STableScanBase { @@ -325,46 +326,46 @@ typedef struct STableScanBase { } STableScanBase; typedef struct STableScanInfo { - STableScanBase base; - SScanInfo scanInfo; - int32_t scanTimes; - SSDataBlock* pResBlock; - SSampleExecInfo sample; // sample execution info - int32_t currentGroupId; - int32_t currentTable; - int8_t scanMode; - int8_t assignBlockUid; - bool hasGroupByTag; + STableScanBase base; + SScanInfo scanInfo; + int32_t scanTimes; + SSDataBlock* pResBlock; + SSampleExecInfo sample; // sample execution info + int32_t currentGroupId; + int32_t currentTable; + int8_t scanMode; + int8_t assignBlockUid; + bool hasGroupByTag; } STableScanInfo; typedef struct STableMergeScanInfo { - int32_t tableStartIndex; - int32_t tableEndIndex; - bool hasGroupId; - uint64_t groupId; - SArray* queryConds; // array of queryTableDataCond - STableScanBase base; - int32_t bufPageSize; - uint32_t sortBufSize; // max buffer size for in-memory sort - SArray* pSortInfo; - SSortHandle* pSortHandle; - SSDataBlock* pSortInputBlock; - int64_t startTs; // sort start time - SArray* sortSourceParams; - SLimitInfo limitInfo; - int64_t numOfRows; - SScanInfo scanInfo; - SSDataBlock* pResBlock; - SSampleExecInfo sample; // sample execution info - SSortExecInfo sortExecInfo; + int32_t tableStartIndex; + int32_t tableEndIndex; + bool hasGroupId; + uint64_t groupId; + SArray* queryConds; // array of queryTableDataCond + STableScanBase base; + int32_t bufPageSize; + uint32_t sortBufSize; // max buffer size for in-memory sort + SArray* pSortInfo; + SSortHandle* pSortHandle; + SSDataBlock* pSortInputBlock; + int64_t startTs; // sort start time + SArray* sortSourceParams; + SLimitInfo limitInfo; + int64_t numOfRows; + SScanInfo scanInfo; + SSDataBlock* pResBlock; + SSampleExecInfo sample; // sample execution info + SSortExecInfo sortExecInfo; } STableMergeScanInfo; typedef struct STagScanInfo { - SColumnInfo* pCols; - SSDataBlock* pRes; - SColMatchInfo matchInfo; - int32_t curPos; - SReadHandle readHandle; + SColumnInfo* pCols; + SSDataBlock* pRes; + SColMatchInfo matchInfo; + int32_t curPos; + SReadHandle readHandle; } STagScanInfo; typedef enum EStreamScanMode { @@ -468,6 +469,11 @@ typedef struct SStreamScanInfo { SNodeList* pGroupTags; SNode* pTagCond; SNode* pTagIndexCond; + + // recover + int32_t blockRecoverContiCnt; + int32_t blockRecoverTotCnt; + } SStreamScanInfo; typedef struct { @@ -499,8 +505,8 @@ typedef struct STableCountScanOperatorInfo { STableCountScanSupp supp; - int32_t currGrpIdx; - SArray* stbUidList; // when group by db_name and/or stable_name + int32_t currGrpIdx; + SArray* stbUidList; // when group by db_name and/or stable_name } STableCountScanOperatorInfo; typedef struct SOptrBasicInfo { @@ -678,19 +684,19 @@ void setOperatorInfo(SOperatorInfo* pOperator, const char* name, int32 void destroyOperatorInfo(SOperatorInfo* pOperator); int32_t optrDefaultBufFn(SOperatorInfo* pOperator); -void initBasicInfo(SOptrBasicInfo* pInfo, SSDataBlock* pBlock); -void cleanupBasicInfo(SOptrBasicInfo* pInfo); +void initBasicInfo(SOptrBasicInfo* pInfo, SSDataBlock* pBlock); +void cleanupBasicInfo(SOptrBasicInfo* pInfo); int32_t initExprSupp(SExprSupp* pSup, SExprInfo* pExprInfo, int32_t numOfExpr); void cleanupExprSupp(SExprSupp* pSup); -void destroyExprInfo(SExprInfo* pExpr, int32_t numOfExprs); +void destroyExprInfo(SExprInfo* pExpr, int32_t numOfExprs); int32_t initAggSup(SExprSupp* pSup, SAggSupporter* pAggSup, SExprInfo* pExprInfo, int32_t numOfCols, size_t keyBufSize, const char* pkey); void cleanupAggSup(SAggSupporter* pAggSup); -void initResultSizeInfo(SResultInfo* pResultInfo, int32_t numOfRows); +void initResultSizeInfo(SResultInfo* pResultInfo, int32_t numOfRows); void doBuildStreamResBlock(SOperatorInfo* pOperator, SOptrBasicInfo* pbInfo, SGroupResInfo* pGroupResInfo, SDiskbasedBuf* pBuf); @@ -803,10 +809,10 @@ void setInputDataBlock(SExprSupp* pExprSupp, SSDataBlock* pBlock, int32_t order, int32_t checkForQueryBuf(size_t numOfTables); -bool isTaskKilled(SExecTaskInfo* pTaskInfo); -void setTaskKilled(SExecTaskInfo* pTaskInfo, int32_t rspCode); -void doDestroyTask(SExecTaskInfo* pTaskInfo); -void setTaskStatus(SExecTaskInfo* pTaskInfo, int8_t status); +bool isTaskKilled(SExecTaskInfo* pTaskInfo); +void setTaskKilled(SExecTaskInfo* pTaskInfo, int32_t rspCode); +void doDestroyTask(SExecTaskInfo* pTaskInfo); +void setTaskStatus(SExecTaskInfo* pTaskInfo, int8_t status); int32_t createExecTaskInfoImpl(SSubplan* pPlan, SExecTaskInfo** pTaskInfo, SReadHandle* pHandle, uint64_t taskId, char* sql, EOPTR_EXEC_MODEL model); @@ -828,8 +834,8 @@ bool isDeletedWindow(STimeWindow* pWin, uint64_t groupId, SAggSupporter* pSup); bool isDeletedStreamWindow(STimeWindow* pWin, uint64_t groupId, SStreamState* pState, STimeWindowAggSupp* pTwSup); void appendOneRowToStreamSpecialBlock(SSDataBlock* pBlock, TSKEY* pStartTs, TSKEY* pEndTs, uint64_t* pUid, uint64_t* pGp, void* pTbName); -uint64_t calGroupIdByData(SPartitionBySupporter* pParSup, SExprSupp* pExprSup, SSDataBlock* pBlock, int32_t rowId); -void calBlockTbName(SStreamScanInfo* pInfo, SSDataBlock* pBlock); +uint64_t calGroupIdByData(SPartitionBySupporter* pParSup, SExprSupp* pExprSup, SSDataBlock* pBlock, int32_t rowId); +void calBlockTbName(SStreamScanInfo* pInfo, SSDataBlock* pBlock); int32_t finalizeResultRows(SDiskbasedBuf* pBuf, SResultRowPosition* resultRowPosition, SExprSupp* pSup, SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo); diff --git a/source/libs/executor/src/dataDeleter.c b/source/libs/executor/src/dataDeleter.c index eff7a5ef93..a8051ea7c3 100644 --- a/source/libs/executor/src/dataDeleter.c +++ b/source/libs/executor/src/dataDeleter.c @@ -62,8 +62,8 @@ static void toDataCacheEntry(SDataDeleterHandle* pHandle, const SInputData* pInp pEntry->numOfCols = taosArrayGetSize(pInput->pData->pDataBlock); pEntry->dataLen = sizeof(SDeleterRes); - ASSERT(1 == pEntry->numOfRows); - ASSERT(3 == pEntry->numOfCols); +// ASSERT(1 == pEntry->numOfRows); +// ASSERT(3 == pEntry->numOfCols); pBuf->useSize = sizeof(SDataCacheEntry); @@ -167,7 +167,6 @@ static void getDataLength(SDataSinkHandle* pHandle, int64_t* pLen, bool* pQueryE SDataDeleterBuf* pBuf = NULL; taosReadQitem(pDeleter->pDataBlocks, (void**)&pBuf); - ASSERT(NULL != pBuf); memcpy(&pDeleter->nextOutput, pBuf, sizeof(SDataDeleterBuf)); taosFreeQitem(pBuf); diff --git a/source/libs/executor/src/dataDispatcher.c b/source/libs/executor/src/dataDispatcher.c index c2fa438c80..a603bffba5 100644 --- a/source/libs/executor/src/dataDispatcher.c +++ b/source/libs/executor/src/dataDispatcher.c @@ -77,8 +77,8 @@ static void toDataCacheEntry(SDataDispatchHandle* pHandle, const SInputData* pIn pBuf->useSize = sizeof(SDataCacheEntry); pEntry->dataLen = blockEncode(pInput->pData, pEntry->data, numOfCols); - ASSERT(pEntry->numOfRows == *(int32_t*)(pEntry->data + 8)); - ASSERT(pEntry->numOfCols == *(int32_t*)(pEntry->data + 8 + 4)); +// ASSERT(pEntry->numOfRows == *(int32_t*)(pEntry->data + 8)); +// ASSERT(pEntry->numOfCols == *(int32_t*)(pEntry->data + 8 + 4)); pBuf->useSize += pEntry->dataLen; @@ -162,15 +162,14 @@ static void getDataLength(SDataSinkHandle* pHandle, int64_t* pLen, bool* pQueryE SDataDispatchBuf* pBuf = NULL; taosReadQitem(pDispatcher->pDataBlocks, (void**)&pBuf); - ASSERT(NULL != pBuf); memcpy(&pDispatcher->nextOutput, pBuf, sizeof(SDataDispatchBuf)); taosFreeQitem(pBuf); SDataCacheEntry* pEntry = (SDataCacheEntry*)pDispatcher->nextOutput.pData; *pLen = pEntry->dataLen; - ASSERT(pEntry->numOfRows == *(int32_t*)(pEntry->data + 8)); - ASSERT(pEntry->numOfCols == *(int32_t*)(pEntry->data + 8 + 4)); +// ASSERT(pEntry->numOfRows == *(int32_t*)(pEntry->data + 8)); +// ASSERT(pEntry->numOfCols == *(int32_t*)(pEntry->data + 8 + 4)); *pQueryEnd = pDispatcher->queryEnd; qDebug("got data len %" PRId64 ", row num %d in sink", *pLen, @@ -193,8 +192,8 @@ static int32_t getDataBlock(SDataSinkHandle* pHandle, SOutputData* pOutput) { pOutput->numOfCols = pEntry->numOfCols; pOutput->compressed = pEntry->compressed; - ASSERT(pEntry->numOfRows == *(int32_t*)(pEntry->data + 8)); - ASSERT(pEntry->numOfCols == *(int32_t*)(pEntry->data + 8 + 4)); +// ASSERT(pEntry->numOfRows == *(int32_t*)(pEntry->data + 8)); +// ASSERT(pEntry->numOfCols == *(int32_t*)(pEntry->data + 8 + 4)); atomic_sub_fetch_64(&pDispatcher->cachedSize, pEntry->dataLen); atomic_sub_fetch_64(&gDataSinkStat.cachedSize, pEntry->dataLen); diff --git a/source/libs/executor/src/exchangeoperator.c b/source/libs/executor/src/exchangeoperator.c index 4103ca82dc..9873c52006 100644 --- a/source/libs/executor/src/exchangeoperator.c +++ b/source/libs/executor/src/exchangeoperator.c @@ -373,7 +373,6 @@ int32_t loadRemoteDataCallback(void* param, SDataBuf* pMsg, int32_t code) { pRsp->useconds = htobe64(pRsp->useconds); pRsp->numOfBlocks = htonl(pRsp->numOfBlocks); - ASSERT(pRsp != NULL); qDebug("%s fetch rsp received, index:%d, blocks:%d, rows:%" PRId64 ", %p", pSourceDataInfo->taskId, index, pRsp->numOfBlocks, pRsp->numOfRows, pExchangeInfo); } else { diff --git a/source/libs/executor/src/executor.c b/source/libs/executor/src/executor.c index 75de012947..e5ff104d5c 100644 --- a/source/libs/executor/src/executor.c +++ b/source/libs/executor/src/executor.c @@ -936,6 +936,10 @@ int32_t qStreamRestoreParam(qTaskInfo_t tinfo) { } return 0; } +bool qStreamRecoverScanFinished(qTaskInfo_t tinfo) { + SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo; + return pTaskInfo->streamInfo.recoverScanFinished; +} void* qExtractReaderFromStreamScanner(void* scanner) { SStreamScanInfo* pInfo = scanner; diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c index fecdcd8fa3..149efef884 100644 --- a/source/libs/executor/src/executorimpl.c +++ b/source/libs/executor/src/executorimpl.c @@ -104,8 +104,6 @@ static int32_t doCopyToSDataBlock(SExecTaskInfo* pTaskInfo, SSDataBlock* pBlock, void setOperatorCompleted(SOperatorInfo* pOperator) { pOperator->status = OP_EXEC_DONE; - ASSERT(pOperator->pTaskInfo != NULL); - pOperator->cost.totalCost = (taosGetTimestampUs() - pOperator->pTaskInfo->cost.start) / 1000.0; setTaskStatus(pOperator->pTaskInfo, TASK_COMPLETED); } @@ -524,7 +522,7 @@ bool functionNeedToExecute(SqlFunctionCtx* pCtx) { return true; } -static int32_t doCreateConstantValColumnAggInfo(SInputColumnInfoData* pInput, SFunctParam* pFuncParam, int32_t type, +static int32_t doCreateConstantValColumnSMAInfo(SInputColumnInfoData* pInput, SFunctParam* pFuncParam, int32_t type, int32_t paramIndex, int32_t numOfRows) { if (pInput->pData[paramIndex] == NULL) { pInput->pData[paramIndex] = taosMemoryCalloc(1, sizeof(SColumnInfoData)); @@ -548,8 +546,6 @@ static int32_t doCreateConstantValColumnAggInfo(SInputColumnInfoData* pInput, SF da = pInput->pColumnDataAgg[paramIndex]; } - ASSERT(!IS_VAR_DATA_TYPE(type)); - if (type == TSDB_DATA_TYPE_BIGINT) { int64_t v = pFuncParam->param.i; *da = (SColumnDataAgg){.numOfNull = 0, .min = v, .max = v, .sum = v * numOfRows}; @@ -570,7 +566,7 @@ static int32_t doCreateConstantValColumnAggInfo(SInputColumnInfoData* pInput, SF } else if (type == TSDB_DATA_TYPE_TIMESTAMP) { // do nothing } else { - ASSERT(0); + qError("invalid constant type for sma info"); } return TSDB_CODE_SUCCESS; @@ -600,7 +596,7 @@ void setBlockSMAInfo(SqlFunctionCtx* pCtx, SExprInfo* pExprInfo, SSDataBlock* pB // the data in the corresponding SColumnInfoData will not be used. pInput->pData[j] = taosArrayGet(pBlock->pDataBlock, slotId); } else if (pFuncParam->type == FUNC_PARAM_TYPE_VALUE) { - doCreateConstantValColumnAggInfo(pInput, pFuncParam, pFuncParam->param.nType, j, pBlock->info.rows); + doCreateConstantValColumnSMAInfo(pInput, pFuncParam, pFuncParam->param.nType, j, pBlock->info.rows); } } } else { @@ -1577,8 +1573,7 @@ void destroyOperatorInfo(SOperatorInfo* pOperator) { // each operator should be set their own function to return total cost buffer int32_t optrDefaultBufFn(SOperatorInfo* pOperator) { if (pOperator->blocking) { - ASSERT(0); - return 0; + return -1; } else { return 0; } @@ -2205,7 +2200,6 @@ static int32_t extractTbscanInStreamOpTree(SOperatorInfo* pOperator, STableScanI return extractTbscanInStreamOpTree(pOperator->pDownstream[0], ppInfo); } else { SStreamScanInfo* pInfo = pOperator->info; - ASSERT(pInfo->pTableScanOp->operatorType == QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN); *ppInfo = pInfo->pTableScanOp->info; return 0; } @@ -2217,13 +2211,11 @@ int32_t extractTableScanNode(SPhysiNode* pNode, STableScanPhysiNode** ppNode) { *ppNode = (STableScanPhysiNode*)pNode; return 0; } else { - ASSERT(0); terrno = TSDB_CODE_APP_ERROR; return -1; } } else { if (LIST_LENGTH(pNode->pChildren) != 1) { - ASSERT(0); terrno = TSDB_CODE_APP_ERROR; return -1; } @@ -2233,32 +2225,6 @@ int32_t extractTableScanNode(SPhysiNode* pNode, STableScanPhysiNode** ppNode) { return -1; } -#if 0 -int32_t rebuildReader(SOperatorInfo* pOperator, SSubplan* plan, SReadHandle* pHandle, int64_t uid, int64_t ts) { - STableScanInfo* pTableScanInfo = NULL; - if (extractTbscanInStreamOpTree(pOperator, &pTableScanInfo) < 0) { - return -1; - } - - STableScanPhysiNode* pNode = NULL; - if (extractTableScanNode(plan->pNode, &pNode) < 0) { - ASSERT(0); - } - - tsdbReaderClose(pTableScanInfo->dataReader); - - STableListInfo info = {0}; - pTableScanInfo->dataReader = doCreateDataReader(pNode, pHandle, &info, NULL); - if (pTableScanInfo->dataReader == NULL) { - ASSERT(0); - qError("failed to create data reader"); - return TSDB_CODE_APP_ERROR; - } - // TODO: set uid and ts to data reader - return 0; -} -#endif - int32_t createDataSinkParam(SDataSinkNode* pNode, void** pParam, qTaskInfo_t* pTaskInfo, SReadHandle* readHandle) { SExecTaskInfo* pTask = *(SExecTaskInfo**)pTaskInfo; diff --git a/source/libs/executor/src/joinoperator.c b/source/libs/executor/src/joinoperator.c index 8a097a23ce..88ed9eccb3 100644 --- a/source/libs/executor/src/joinoperator.c +++ b/source/libs/executor/src/joinoperator.c @@ -42,38 +42,40 @@ typedef struct SJoinOperatorInfo { static void setJoinColumnInfo(SColumnInfo* pColumn, const SColumnNode* pColumnNode); static SSDataBlock* doMergeJoin(struct SOperatorInfo* pOperator); static void destroyMergeJoinOperator(void* param); -static void extractTimeCondition(SJoinOperatorInfo* pInfo, SOperatorInfo** pDownstream, int32_t numOfDownstream, - SSortMergeJoinPhysiNode* pJoinNode); +static void extractTimeCondition(SJoinOperatorInfo* pInfo, SOperatorInfo** pDownstream, int32_t num, + SSortMergeJoinPhysiNode* pJoinNode, const char* idStr); -static void extractTimeCondition(SJoinOperatorInfo* pInfo, SOperatorInfo** pDownstream, int32_t numOfDownstream, - SSortMergeJoinPhysiNode* pJoinNode) { +static void extractTimeCondition(SJoinOperatorInfo* pInfo, SOperatorInfo** pDownstream, int32_t num, + SSortMergeJoinPhysiNode* pJoinNode, const char* idStr) { SNode* pMergeCondition = pJoinNode->pMergeCondition; - if (nodeType(pMergeCondition) == QUERY_NODE_OPERATOR) { - SOperatorNode* pNode = (SOperatorNode*)pMergeCondition; - SColumnNode* col1 = (SColumnNode*)pNode->pLeft; - SColumnNode* col2 = (SColumnNode*)pNode->pRight; - SColumnNode* leftTsCol = NULL; - SColumnNode* rightTsCol = NULL; - if (col1->dataBlockId == col2->dataBlockId ) { + if (nodeType(pMergeCondition) != QUERY_NODE_OPERATOR) { + qError("not support this in join operator, %s", idStr); + return; // do not handle this + } + + SOperatorNode* pNode = (SOperatorNode*)pMergeCondition; + SColumnNode* col1 = (SColumnNode*)pNode->pLeft; + SColumnNode* col2 = (SColumnNode*)pNode->pRight; + SColumnNode* leftTsCol = NULL; + SColumnNode* rightTsCol = NULL; + if (col1->dataBlockId == col2->dataBlockId) { + leftTsCol = col1; + rightTsCol = col2; + } else { + if (col1->dataBlockId == pDownstream[0]->resultDataBlockId) { + ASSERT(col2->dataBlockId == pDownstream[1]->resultDataBlockId); leftTsCol = col1; rightTsCol = col2; } else { - if (col1->dataBlockId == pDownstream[0]->resultDataBlockId) { - ASSERT(col2->dataBlockId == pDownstream[1]->resultDataBlockId); - leftTsCol = col1; - rightTsCol = col2; - } else { - ASSERT(col1->dataBlockId == pDownstream[1]->resultDataBlockId); - ASSERT(col2->dataBlockId == pDownstream[0]->resultDataBlockId); - leftTsCol = col2; - rightTsCol = col1; - } + ASSERT(col1->dataBlockId == pDownstream[1]->resultDataBlockId); + ASSERT(col2->dataBlockId == pDownstream[0]->resultDataBlockId); + leftTsCol = col2; + rightTsCol = col1; } - setJoinColumnInfo(&pInfo->leftCol, leftTsCol); - setJoinColumnInfo(&pInfo->rightCol, rightTsCol); - } else { - ASSERT(false); - }} + } + setJoinColumnInfo(&pInfo->leftCol, leftTsCol); + setJoinColumnInfo(&pInfo->rightCol, rightTsCol); +} SOperatorInfo* createMergeJoinOperatorInfo(SOperatorInfo** pDownstream, int32_t numOfDownstream, SSortMergeJoinPhysiNode* pJoinNode, SExecTaskInfo* pTaskInfo) { @@ -97,7 +99,7 @@ SOperatorInfo* createMergeJoinOperatorInfo(SOperatorInfo** pDownstream, int32_t pOperator->exprSupp.pExprInfo = pExprInfo; pOperator->exprSupp.numOfExprs = numOfCols; - extractTimeCondition(pInfo, pDownstream, numOfDownstream, pJoinNode); + extractTimeCondition(pInfo, pDownstream, numOfDownstream, pJoinNode, GET_TASKID(pTaskInfo)); if (pJoinNode->pOnConditions != NULL && pJoinNode->node.pConditions != NULL) { pInfo->pCondAfterMerge = nodesMakeNode(QUERY_NODE_LOGIC_CONDITION); @@ -364,8 +366,6 @@ static bool mergeJoinGetNextTimestamp(SOperatorInfo* pOperator, int64_t* pLeftTs char* pRightVal = colDataGetData(pRightCol, pJoinInfo->rightPos); *pRightTs = *(int64_t*)pRightVal; - ASSERT(pLeftCol->info.type == TSDB_DATA_TYPE_TIMESTAMP); - ASSERT(pRightCol->info.type == TSDB_DATA_TYPE_TIMESTAMP); return true; } diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c index d074ceede8..0e22195afa 100644 --- a/source/libs/executor/src/scanoperator.c +++ b/source/libs/executor/src/scanoperator.c @@ -232,30 +232,6 @@ static bool doLoadBlockSMA(STableScanBase* pTableScanInfo, SSDataBlock* pBlock, if (!allColumnsHaveAgg) { return false; } - -#if 0 - // if (allColumnsHaveAgg == true) { - int32_t numOfCols = taosArrayGetSize(pBlock->pDataBlock); - - // todo create this buffer during creating operator - if (pBlock->pBlockAgg == NULL) { - pBlock->pBlockAgg = taosMemoryCalloc(numOfCols, POINTER_BYTES); - if (pBlock->pBlockAgg == NULL) { - T_LONG_JMP(pTaskInfo->env, TSDB_CODE_OUT_OF_MEMORY); - } - } - - size_t num = taosArrayGetSize(pTableScanInfo->matchInfo.pList); - for (int32_t i = 0; i < num; ++i) { - SColMatchItem* pColMatchInfo = taosArrayGet(pTableScanInfo->matchInfo.pList, i); - if (!pColMatchInfo->needOutput) { - continue; - } - - pBlock->pBlockAgg[pColMatchInfo->dstSlotId] = pColAgg[i]; - } -#endif - return true; } @@ -768,8 +744,8 @@ static SSDataBlock* doTableScan(SOperatorInfo* pOperator) { tableListGetGroupList(pTaskInfo->pTableInfoList, pInfo->currentGroupId, &pList, &num); ASSERT(pInfo->base.dataReader == NULL); - int32_t code = tsdbReaderOpen(pInfo->base.readHandle.vnode, &pInfo->base.cond, pList, num, - pInfo->pResBlock, (STsdbReader**)&pInfo->base.dataReader, GET_TASKID(pTaskInfo)); + int32_t code = tsdbReaderOpen(pInfo->base.readHandle.vnode, &pInfo->base.cond, pList, num, pInfo->pResBlock, + (STsdbReader**)&pInfo->base.dataReader, GET_TASKID(pTaskInfo)); if (code != TSDB_CODE_SUCCESS) { T_LONG_JMP(pTaskInfo->env, code); } @@ -986,8 +962,8 @@ static SSDataBlock* readPreVersionData(SOperatorInfo* pTableScanOp, uint64_t tbU SSDataBlock* pBlock = pTableScanInfo->pResBlock; STsdbReader* pReader = NULL; - int32_t code = tsdbReaderOpen(pTableScanInfo->base.readHandle.vnode, &cond, &tblInfo, 1, pBlock, (STsdbReader**)&pReader, - GET_TASKID(pTaskInfo)); + int32_t code = tsdbReaderOpen(pTableScanInfo->base.readHandle.vnode, &cond, &tblInfo, 1, pBlock, + (STsdbReader**)&pReader, GET_TASKID(pTaskInfo)); if (code != TSDB_CODE_SUCCESS) { terrno = code; T_LONG_JMP(pTaskInfo->env, code); @@ -995,7 +971,7 @@ static SSDataBlock* readPreVersionData(SOperatorInfo* pTableScanOp, uint64_t tbU } if (tsdbNextDataBlock(pReader)) { - /*SSDataBlock* p = */tsdbRetrieveDataBlock(pReader, NULL); + /*SSDataBlock* p = */ tsdbRetrieveDataBlock(pReader, NULL); doSetTagColumnData(&pTableScanInfo->base, pBlock, pTaskInfo, pBlock->info.rows); pBlock->info.id.groupId = getTableGroupId(pTaskInfo->pTableInfoList, pBlock->info.id.uid); } @@ -1224,7 +1200,7 @@ static int32_t generateIntervalScanRange(SStreamScanInfo* pInfo, SSDataBlock* pS SColumnInfoData* pSrcUidCol = taosArrayGet(pSrcBlock->pDataBlock, UID_COLUMN_INDEX); SColumnInfoData* pSrcGpCol = taosArrayGet(pSrcBlock->pDataBlock, GROUPID_COLUMN_INDEX); - uint64_t* srcUidData = (uint64_t*)pSrcUidCol->pData; + uint64_t* srcUidData = (uint64_t*)pSrcUidCol->pData; ASSERT(pSrcStartTsCol->info.type == TSDB_DATA_TYPE_TIMESTAMP); TSKEY* srcStartTsCol = (TSKEY*)pSrcStartTsCol->pData; TSKEY* srcEndTsCol = (TSKEY*)pSrcEndTsCol->pData; @@ -1753,11 +1729,18 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) { pTSInfo->scanTimes = 0; pTSInfo->currentGroupId = -1; pTaskInfo->streamInfo.recoverStep = STREAM_RECOVER_STEP__SCAN; + pTaskInfo->streamInfo.recoverScanFinished = false; } if (pTaskInfo->streamInfo.recoverStep == STREAM_RECOVER_STEP__SCAN) { + if (pInfo->blockRecoverContiCnt > 100) { + pInfo->blockRecoverTotCnt += pInfo->blockRecoverContiCnt; + pInfo->blockRecoverContiCnt = 0; + return NULL; + } SSDataBlock* pBlock = doTableScan(pInfo->pTableScanOp); if (pBlock != NULL) { + pInfo->blockRecoverContiCnt++; calBlockTbName(pInfo, pBlock); if (pInfo->pUpdateInfo) { TSKEY maxTs = updateInfoFillBlockData(pInfo->pUpdateInfo, pBlock, pInfo->primaryTsIndex); @@ -1775,6 +1758,7 @@ static SSDataBlock* doStreamScan(SOperatorInfo* pOperator) { pTSInfo->base.cond.startVersion = -1; pTSInfo->base.cond.endVersion = -1; + pTaskInfo->streamInfo.recoverScanFinished = true; return NULL; } @@ -2206,6 +2190,7 @@ static void destroyStreamScanOperatorInfo(void* param) { SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhysiNode* pTableScanNode, SNode* pTagCond, SExecTaskInfo* pTaskInfo) { + SArray* pColIds = NULL; SStreamScanInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamScanInfo)); SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo)); @@ -2228,7 +2213,7 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys } int32_t numOfOutput = taosArrayGetSize(pInfo->matchInfo.pList); - SArray* pColIds = taosArrayInit(numOfOutput, sizeof(int16_t)); + pColIds = taosArrayInit(numOfOutput, sizeof(int16_t)); for (int32_t i = 0; i < numOfOutput; ++i) { SColMatchItem* id = taosArrayGet(pInfo->matchInfo.pList, i); @@ -2285,7 +2270,8 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys if (pHandle->initTableReader) { pTSInfo->scanMode = TABLE_SCAN__TABLE_ORDER; pTSInfo->base.dataReader = NULL; - code = tsdbReaderOpen(pHandle->vnode, &pTSInfo->base.cond, pList, num, pTSInfo->pResBlock, &pTSInfo->base.dataReader, NULL); + code = tsdbReaderOpen(pHandle->vnode, &pTSInfo->base.cond, pList, num, pTSInfo->pResBlock, + &pTSInfo->base.dataReader, NULL); if (code != 0) { terrno = code; destroyTableScanOperatorInfo(pTableScanOp); @@ -2324,6 +2310,7 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys memcpy(&pTaskInfo->streamInfo.tableCond, &pTSInfo->base.cond, sizeof(SQueryTableDataCond)); } else { taosArrayDestroy(pColIds); + pColIds = NULL; } // create the pseduo columns info @@ -2355,7 +2342,8 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys pOperator->exprSupp.numOfExprs = taosArrayGetSize(pInfo->pRes->pDataBlock); __optr_fn_t nextFn = pTaskInfo->execModel == OPTR_EXEC_MODEL_STREAM ? doStreamScan : doQueueScan; - pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, nextFn, NULL, destroyStreamScanOperatorInfo, optrDefaultBufFn, NULL); + pOperator->fpSet = + createOperatorFpSet(optrDummyOpenFn, nextFn, NULL, destroyStreamScanOperatorInfo, optrDefaultBufFn, NULL); return pOperator; @@ -2492,7 +2480,8 @@ SOperatorInfo* createTagScanOperatorInfo(SReadHandle* pReadHandle, STagScanPhysi initResultSizeInfo(&pOperator->resultInfo, 4096); blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity); - pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doTagScan, NULL, destroyTagScanOperatorInfo, optrDefaultBufFn, NULL); + pOperator->fpSet = + createOperatorFpSet(optrDummyOpenFn, doTagScan, NULL, destroyTagScanOperatorInfo, optrDefaultBufFn, NULL); return pOperator; @@ -2513,11 +2502,12 @@ static SSDataBlock* getTableDataBlockImpl(void* param) { SQueryTableDataCond* pQueryCond = taosArrayGet(pInfo->queryConds, readIdx); - int64_t st = taosGetTimestampUs(); - void* p = tableListGetInfo(pTaskInfo->pTableInfoList, readIdx + pInfo->tableStartIndex); + int64_t st = taosGetTimestampUs(); + void* p = tableListGetInfo(pTaskInfo->pTableInfoList, readIdx + pInfo->tableStartIndex); SReadHandle* pHandle = &pInfo->base.readHandle; - int32_t code = tsdbReaderOpen(pHandle->vnode, pQueryCond, p, 1, pBlock, &pInfo->base.dataReader, GET_TASKID(pTaskInfo)); + int32_t code = + tsdbReaderOpen(pHandle->vnode, pQueryCond, p, 1, pBlock, &pInfo->base.dataReader, GET_TASKID(pTaskInfo)); if (code != 0) { T_LONG_JMP(pTaskInfo->env, code); } @@ -2915,8 +2905,8 @@ static void buildVnodeGroupedNtbTableCount(STableCountScanOperatorInfo* SSDataBlock* pRes, char* dbName); static void buildVnodeFilteredTbCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo, STableCountScanSupp* pSupp, SSDataBlock* pRes, char* dbName); -static void buildVnodeGroupedTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo, - STableCountScanSupp* pSupp, SSDataBlock* pRes, int32_t vgId, char* dbName); +static void buildVnodeGroupedTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo, + STableCountScanSupp* pSupp, SSDataBlock* pRes, int32_t vgId, char* dbName); static SSDataBlock* buildVnodeDbTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo, STableCountScanSupp* pSupp, SSDataBlock* pRes); static void buildSysDbGroupedTableCount(SOperatorInfo* pOperator, STableCountScanOperatorInfo* pInfo, @@ -3041,8 +3031,8 @@ SOperatorInfo* createTableCountScanOperatorInfo(SReadHandle* readHandle, STableC setOperatorInfo(pOperator, "TableCountScanOperator", QUERY_NODE_PHYSICAL_PLAN_TABLE_COUNT_SCAN, false, OP_NOT_OPENED, pInfo, pTaskInfo); - pOperator->fpSet = - createOperatorFpSet(optrDummyOpenFn, doTableCountScan, NULL, destoryTableCountScanOperator, optrDefaultBufFn, NULL); + pOperator->fpSet = createOperatorFpSet(optrDummyOpenFn, doTableCountScan, NULL, destoryTableCountScanOperator, + optrDefaultBufFn, NULL); return pOperator; _error: diff --git a/source/libs/executor/src/sortoperator.c b/source/libs/executor/src/sortoperator.c index 7ac007b7cb..f5dc6cc623 100644 --- a/source/libs/executor/src/sortoperator.c +++ b/source/libs/executor/src/sortoperator.c @@ -46,13 +46,15 @@ SOperatorInfo* createSortOperatorInfo(SOperatorInfo* downstream, SSortPhysiNode* pOperator->pTaskInfo = pTaskInfo; SDataBlockDescNode* pDescNode = pSortNode->node.pOutputDataBlockDesc; - int32_t numOfCols = 0; - SSDataBlock* pResBlock = createDataBlockFromDescNode(pDescNode); - SExprInfo* pExprInfo = createExprInfo(pSortNode->pExprs, NULL, &numOfCols); + int32_t numOfCols = 0; + SExprInfo* pExprInfo = createExprInfo(pSortNode->pExprs, NULL, &numOfCols); int32_t numOfOutputCols = 0; int32_t code = extractColMatchInfo(pSortNode->pTargets, pDescNode, &numOfOutputCols, COL_MATCH_FROM_SLOT_ID, &pInfo->matchInfo); + if (code != TSDB_CODE_SUCCESS) { + goto _error; + } pOperator->exprSupp.pCtx = createSqlFunctionCtx(pExprInfo, numOfCols, &pOperator->exprSupp.rowEntryInfoOffset); initResultSizeInfo(&pOperator->resultInfo, 1024); @@ -61,7 +63,7 @@ SOperatorInfo* createSortOperatorInfo(SOperatorInfo* downstream, SSortPhysiNode* goto _error; } - pInfo->binfo.pRes = pResBlock; + pInfo->binfo.pRes = createDataBlockFromDescNode(pDescNode); pInfo->pSortInfo = createSortInfo(pSortNode->pSortKeys); initLimitInfo(pSortNode->node.pLimit, pSortNode->node.pSlimit, &pInfo->limitInfo); @@ -86,7 +88,10 @@ SOperatorInfo* createSortOperatorInfo(SOperatorInfo* downstream, SSortPhysiNode* _error: pTaskInfo->code = TSDB_CODE_OUT_OF_MEMORY; - taosMemoryFree(pInfo); + if (pInfo != NULL) { + destroySortOperatorInfo(pInfo); + } + taosMemoryFree(pOperator); return NULL; } @@ -139,7 +144,6 @@ SSDataBlock* getSortedBlockData(SSortHandle* pHandle, SSDataBlock* pDataBlock, i int32_t numOfCols = taosArrayGetSize(pColMatchInfo); for (int32_t i = 0; i < numOfCols; ++i) { SColMatchItem* pmInfo = taosArrayGet(pColMatchInfo, i); - // ASSERT(pmInfo->matchType == COL_MATCH_FROM_SLOT_ID); SColumnInfoData* pSrc = taosArrayGet(p->pDataBlock, pmInfo->srcSlotId); SColumnInfoData* pDst = taosArrayGet(pDataBlock->pDataBlock, pmInfo->dstSlotId); @@ -272,7 +276,6 @@ void destroySortOperatorInfo(void* param) { } int32_t getExplainExecInfo(SOperatorInfo* pOptr, void** pOptrExplain, uint32_t* len) { - ASSERT(pOptr != NULL); SSortExecInfo* pInfo = taosMemoryCalloc(1, sizeof(SSortExecInfo)); SSortOperatorInfo* pOperatorInfo = (SSortOperatorInfo*)pOptr->info; @@ -329,7 +332,6 @@ SSDataBlock* getGroupSortedBlockData(SSortHandle* pHandle, SSDataBlock* pDataBlo int32_t numOfCols = taosArrayGetSize(pColMatchInfo); for (int32_t i = 0; i < numOfCols; ++i) { SColMatchItem* pmInfo = taosArrayGet(pColMatchInfo, i); - // ASSERT(pmInfo->matchType == COL_MATCH_FROM_SLOT_ID); SColumnInfoData* pSrc = taosArrayGet(p->pDataBlock, pmInfo->srcSlotId); SColumnInfoData* pDst = taosArrayGet(pDataBlock->pDataBlock, pmInfo->dstSlotId); @@ -746,7 +748,6 @@ void destroyMultiwayMergeOperatorInfo(void* param) { } int32_t getMultiwayMergeExplainExecInfo(SOperatorInfo* pOptr, void** pOptrExplain, uint32_t* len) { - ASSERT(pOptr != NULL); SSortExecInfo* pSortExecInfo = taosMemoryCalloc(1, sizeof(SSortExecInfo)); SMultiwayMergeOperatorInfo* pInfo = (SMultiwayMergeOperatorInfo*)pOptr->info; diff --git a/source/libs/executor/src/timewindowoperator.c b/source/libs/executor/src/timewindowoperator.c index e5dcc43797..0de38fe500 100644 --- a/source/libs/executor/src/timewindowoperator.c +++ b/source/libs/executor/src/timewindowoperator.c @@ -2342,6 +2342,17 @@ void doBuildResult(SOperatorInfo* pOperator, SStreamState* pState, SSDataBlock* buildDataBlockFromGroupRes(pOperator, pState, pBlock, &pOperator->exprSupp, pGroupResInfo); } +static int32_t getNextQualifiedFinalWindow(SInterval* pInterval, STimeWindow* pNext, SDataBlockInfo* pDataBlockInfo, + TSKEY* primaryKeys, int32_t prevPosition) { + int32_t startPos = prevPosition + 1; + if (startPos == pDataBlockInfo->rows) { + startPos = -1; + } else { + *pNext = getFinalTimeWindow(primaryKeys[startPos], pInterval); + } + return startPos; +} + static void doStreamIntervalAggImpl(SOperatorInfo* pOperatorInfo, SSDataBlock* pSDataBlock, uint64_t groupId, SHashObj* pUpdatedMap) { SStreamIntervalOperatorInfo* pInfo = (SStreamIntervalOperatorInfo*)pOperatorInfo->info; @@ -2451,8 +2462,12 @@ static void doStreamIntervalAggImpl(SOperatorInfo* pOperatorInfo, SSDataBlock* p } int32_t prevEndPos = (forwardRows - 1) * step + startPos; ASSERT(pSDataBlock->info.window.skey > 0 && pSDataBlock->info.window.ekey > 0); - startPos = - getNextQualifiedWindow(&pInfo->interval, &nextWin, &pSDataBlock->info, tsCols, prevEndPos, TSDB_ORDER_ASC); + if (IS_FINAL_OP(pInfo)) { + startPos = getNextQualifiedFinalWindow(&pInfo->interval, &nextWin, &pSDataBlock->info, tsCols, prevEndPos); + } else { + startPos = + getNextQualifiedWindow(&pInfo->interval, &nextWin, &pSDataBlock->info, tsCols, prevEndPos, TSDB_ORDER_ASC); + } if (startPos < 0) { break; } diff --git a/source/libs/executor/src/tsimplehash.c b/source/libs/executor/src/tsimplehash.c index 484d917069..fd6215e3a1 100644 --- a/source/libs/executor/src/tsimplehash.c +++ b/source/libs/executor/src/tsimplehash.c @@ -49,7 +49,9 @@ static FORCE_INLINE int32_t taosHashCapacity(int32_t length) { } SSHashObj *tSimpleHashInit(size_t capacity, _hash_fn_t fn) { - ASSERT(fn != NULL); + if (fn == NULL) { + return NULL; + } if (capacity == 0) { capacity = 4; @@ -66,7 +68,6 @@ SSHashObj *tSimpleHashInit(size_t capacity, _hash_fn_t fn) { pHashObj->equalFp = memcmp; pHashObj->hashFp = fn; - ASSERT((pHashObj->capacity & (pHashObj->capacity - 1)) == 0); pHashObj->hashList = (SHNode **)taosMemoryCalloc(pHashObj->capacity, sizeof(void *)); if (!pHashObj->hashList) { diff --git a/source/libs/executor/src/tsort.c b/source/libs/executor/src/tsort.c index 30911887bb..fa0cdb3943 100644 --- a/source/libs/executor/src/tsort.c +++ b/source/libs/executor/src/tsort.c @@ -800,6 +800,7 @@ STupleHandle* tsortNextTuple(SSortHandle* pHandle) { } } + // all sources are completed. if (pHandle->cmpParam.numOfSources == pHandle->numOfCompletedSources) { return NULL; } diff --git a/source/libs/function/src/tudf.c b/source/libs/function/src/tudf.c index 0b309bc8f5..acfd5c07af 100644 --- a/source/libs/function/src/tudf.c +++ b/source/libs/function/src/tudf.c @@ -841,36 +841,42 @@ int32_t convertScalarParamToDataBlock(SScalarParam *input, int32_t numOfCols, SS for (int32_t i = 0; i < numOfCols; ++i) { numOfRows = (input[i].numOfRows > numOfRows) ? input[i].numOfRows : numOfRows; } - output->info.rows = numOfRows; - output->pDataBlock = taosArrayInit(numOfCols, sizeof(SColumnInfoData)); - for (int32_t i = 0; i < numOfCols; ++i) { - if ((input+i)->numOfRows < numOfRows) { - SColumnInfoData* pColInfoData = (input+i)->columnData; - int32_t startRow = (input+i)->numOfRows; - int32_t expandRows = numOfRows - startRow; - colInfoDataEnsureCapacity(pColInfoData, numOfRows, false); + + // create the basic block info structure + for(int32_t i = 0; i < numOfCols; ++i) { + SColumnInfoData* pInfo = input[i].columnData; + SColumnInfoData d = {0}; + d.info = pInfo->info; + + blockDataAppendColInfo(output, &d); + } + + blockDataEnsureCapacity(output, numOfRows); + + for(int32_t i = 0; i < numOfCols; ++i) { + SColumnInfoData* pDest = taosArrayGet(output->pDataBlock, i); + + SColumnInfoData* pColInfoData = input[i].columnData; + colDataAssign(pDest, pColInfoData, input[i].numOfRows, &output->info); + + if (input[i].numOfRows < numOfRows) { + int32_t startRow = input[i].numOfRows; + int expandRows = numOfRows - startRow; bool isNull = colDataIsNull_s(pColInfoData, (input+i)->numOfRows - 1); if (isNull) { - colDataAppendNNULL(pColInfoData, startRow, expandRows); + colDataAppendNNULL(pDest, startRow, expandRows); } else { char* src = colDataGetData(pColInfoData, (input + i)->numOfRows - 1); - int32_t bytes = pColInfoData->info.bytes; - char* data = taosMemoryMalloc(bytes); - memcpy(data, src, bytes); for (int j = 0; j < expandRows; ++j) { - colDataAppend(pColInfoData, startRow+j, data, false); + colDataAppend(pDest, startRow+j, src, false); } //colDataAppendNItems(pColInfoData, startRow, data, expandRows); - taosMemoryFree(data); } } - - taosArrayPush(output->pDataBlock, (input + i)->columnData); - - if (IS_VAR_DATA_TYPE((input + i)->columnData->info.type)) { - output->info.hasVarCol = true; - } } + + output->info.rows = numOfRows; + return 0; } @@ -1824,8 +1830,8 @@ int32_t doCallUdfScalarFunc(UdfcFuncHandle handle, SScalarParam *input, int32_t convertDataBlockToScalarParm(&resultBlock, output); taosArrayDestroy(resultBlock.pDataBlock); } - - taosArrayDestroy(inputBlock.pDataBlock); + + blockDataFreeRes(&inputBlock); return err; } diff --git a/source/libs/parser/inc/parAst.h b/source/libs/parser/inc/parAst.h index ef67c7536f..20a2520a79 100644 --- a/source/libs/parser/inc/parAst.h +++ b/source/libs/parser/inc/parAst.h @@ -196,7 +196,7 @@ SNode* createDropComponentNodeStmt(SAstCreateContext* pCxt, ENodeType type, cons SNode* createCreateTopicStmtUseQuery(SAstCreateContext* pCxt, bool ignoreExists, const SToken* pTopicName, SNode* pQuery); SNode* createCreateTopicStmtUseDb(SAstCreateContext* pCxt, bool ignoreExists, const SToken* pTopicName, - const SToken* pSubDbName, bool withMeta); + SToken* pSubDbName, bool withMeta); SNode* createCreateTopicStmtUseTable(SAstCreateContext* pCxt, bool ignoreExists, const SToken* pTopicName, SNode* pRealTable, bool withMeta); SNode* createDropTopicStmt(SAstCreateContext* pCxt, bool ignoreNotExists, const SToken* pTopicName); diff --git a/source/libs/parser/src/parAstCreater.c b/source/libs/parser/src/parAstCreater.c index 4077c27840..17d8297cf0 100644 --- a/source/libs/parser/src/parAstCreater.c +++ b/source/libs/parser/src/parAstCreater.c @@ -1579,8 +1579,11 @@ SNode* createCreateTopicStmtUseQuery(SAstCreateContext* pCxt, bool ignoreExists, } SNode* createCreateTopicStmtUseDb(SAstCreateContext* pCxt, bool ignoreExists, const SToken* pTopicName, - const SToken* pSubDbName, bool withMeta) { + SToken* pSubDbName, bool withMeta) { CHECK_PARSER_STATUS(pCxt); + if (!checkDbName(pCxt, pSubDbName, true)) { + return NULL; + } SCreateTopicStmt* pStmt = (SCreateTopicStmt*)nodesMakeNode(QUERY_NODE_CREATE_TOPIC_STMT); CHECK_OUT_OF_MEM(pStmt); COPY_STRING_FORM_ID_TOKEN(pStmt->topicName, pTopicName); diff --git a/source/libs/parser/src/parTranslater.c b/source/libs/parser/src/parTranslater.c index 37703534d6..0a20484179 100644 --- a/source/libs/parser/src/parTranslater.c +++ b/source/libs/parser/src/parTranslater.c @@ -5640,7 +5640,8 @@ static int32_t addSubtableInfoToCreateStreamQuery(STranslateContext* pCxt, SCrea return code; } -static int32_t checkStreamQuery(STranslateContext* pCxt, SSelectStmt* pSelect) { +static int32_t checkStreamQuery(STranslateContext* pCxt, SCreateStreamStmt* pStmt) { + SSelectStmt* pSelect = (SSelectStmt*)pStmt->pQuery; if (TSDB_DATA_TYPE_TIMESTAMP != ((SExprNode*)nodesListGetNode(pSelect->pProjectionList, 0))->resType.type || !pSelect->isTimeLineResult || crossTableWithoutAggOper(pSelect) || NULL != pSelect->pOrderByList || crossTableWithUdaf(pSelect)) { @@ -5650,6 +5651,10 @@ static int32_t checkStreamQuery(STranslateContext* pCxt, SSelectStmt* pSelect) { return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_STREAM_QUERY, "SUBTABLE expression must be of VARCHAR type"); } + if (NULL == pSelect->pWindow && STREAM_TRIGGER_AT_ONCE != pStmt->pOptions->triggerType) { + return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_STREAM_QUERY, + "The trigger mode of non window query can only be AT_ONCE"); + } return TSDB_CODE_SUCCESS; } @@ -5663,7 +5668,7 @@ static int32_t buildCreateStreamQuery(STranslateContext* pCxt, SCreateStreamStmt code = translateQuery(pCxt, pStmt->pQuery); } if (TSDB_CODE_SUCCESS == code) { - code = checkStreamQuery(pCxt, (SSelectStmt*)pStmt->pQuery); + code = checkStreamQuery(pCxt, pStmt); } if (TSDB_CODE_SUCCESS == code) { getSourceDatabase(pStmt->pQuery, pCxt->pParseCxt->acctId, pReq->sourceDB); diff --git a/source/libs/scheduler/src/schJob.c b/source/libs/scheduler/src/schJob.c index d422f0e88f..6a8f81f8c7 100644 --- a/source/libs/scheduler/src/schJob.c +++ b/source/libs/scheduler/src/schJob.c @@ -668,6 +668,7 @@ void schFreeJobImpl(void *job) { taosMemoryFreeClear(pJob->userRes.execRes); taosMemoryFreeClear(pJob->fetchRes); taosMemoryFreeClear(pJob->sql); + tsem_destroy(&pJob->rspSem); taosMemoryFree(pJob); int32_t jobNum = atomic_sub_fetch_32(&schMgmt.jobNum, 1); @@ -748,7 +749,10 @@ int32_t schInitJob(int64_t *pJobId, SSchedulerReq *pReq) { SCH_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY); } - tsem_init(&pJob->rspSem, 0, 0); + if (tsem_init(&pJob->rspSem, 0, 0)) { + SCH_JOB_ELOG("tsem_init failed, errno:%d", errno); + SCH_ERR_JRET(TSDB_CODE_OUT_OF_MEMORY); + } pJob->refId = taosAddRef(schMgmt.jobRef, pJob); if (pJob->refId < 0) { diff --git a/source/libs/stream/src/streamDispatch.c b/source/libs/stream/src/streamDispatch.c index 4e0b0630bc..f2b1db19e8 100644 --- a/source/libs/stream/src/streamDispatch.c +++ b/source/libs/stream/src/streamDispatch.c @@ -487,8 +487,6 @@ int32_t streamDispatchAllBlocks(SStreamTask* pTask, const SStreamDataBlock* pDat taosMemoryFree(pReqs); } return code; - } else { - ASSERT(0); } return 0; } @@ -514,7 +512,6 @@ int32_t streamDispatch(SStreamTask* pTask) { int32_t code = 0; if (streamDispatchAllBlocks(pTask, pBlock) < 0) { - ASSERT(0); code = -1; streamQueueProcessFail(pTask->outputQueue); atomic_store_8(&pTask->outputStatus, TASK_OUTPUT_STATUS__NORMAL); diff --git a/source/libs/stream/src/streamExec.c b/source/libs/stream/src/streamExec.c index 8e746c6738..3d42f759cb 100644 --- a/source/libs/stream/src/streamExec.c +++ b/source/libs/stream/src/streamExec.c @@ -116,7 +116,11 @@ int32_t streamScanExec(SStreamTask* pTask, int32_t batchSz) { ASSERT(0); } if (output == NULL) { - finished = true; + if (qStreamRecoverScanFinished(exec)) { + finished = true; + } else { + qSetStreamOpOpen(exec); + } break; } diff --git a/source/libs/stream/src/streamRecover.c b/source/libs/stream/src/streamRecover.c index 6889a870d1..52777fd834 100644 --- a/source/libs/stream/src/streamRecover.c +++ b/source/libs/stream/src/streamRecover.c @@ -143,6 +143,7 @@ int32_t streamProcessTaskCheckRsp(SStreamTask* pTask, const SStreamTaskCheckRsp* ASSERT(left >= 0); if (left == 0) { taosArrayDestroy(pTask->checkReqIds); + pTask->checkReqIds = NULL; streamTaskLaunchRecover(pTask, version); } } else if (pTask->outputType == TASK_OUTPUT__FIXED_DISPATCH) { diff --git a/source/libs/stream/src/streamTask.c b/source/libs/stream/src/streamTask.c index f7252ed8a0..e9aba0bc39 100644 --- a/source/libs/stream/src/streamTask.c +++ b/source/libs/stream/src/streamTask.c @@ -175,6 +175,8 @@ void tFreeSStreamTask(SStreamTask* pTask) { } if (pTask->outputType == TASK_OUTPUT__SHUFFLE_DISPATCH) { taosArrayDestroy(pTask->shuffleDispatcher.dbInfo.pVgroupInfos); + taosArrayDestroy(pTask->checkReqIds); + pTask->checkReqIds = NULL; } if (pTask->pState) streamStateClose(pTask->pState); diff --git a/source/libs/sync/src/syncPipeline.c b/source/libs/sync/src/syncPipeline.c index 0443be3f5e..a458882da5 100644 --- a/source/libs/sync/src/syncPipeline.c +++ b/source/libs/sync/src/syncPipeline.c @@ -45,7 +45,7 @@ int32_t syncLogBufferAppend(SSyncLogBuffer* pBuf, SSyncNode* pNode, SSyncRaftEnt if (index - pBuf->startIndex >= pBuf->size) { sError("vgId:%d, failed to append due to sync log buffer full. index:%" PRId64 "", pNode->vgId, index); - goto _out; + goto _err; } ASSERT(index == pBuf->endIndex); @@ -66,9 +66,8 @@ int32_t syncLogBufferAppend(SSyncLogBuffer* pBuf, SSyncNode* pNode, SSyncRaftEnt taosThreadMutexUnlock(&pBuf->mutex); return 0; -_out: +_err: syncLogBufferValidate(pBuf); - syncEntryDestroy(pEntry); taosThreadMutexUnlock(&pBuf->mutex); return -1; } diff --git a/source/libs/transport/src/transSvr.c b/source/libs/transport/src/transSvr.c index 2b1f68d5f6..7384877313 100644 --- a/source/libs/transport/src/transSvr.c +++ b/source/libs/transport/src/transSvr.c @@ -1001,6 +1001,13 @@ void* transInitServer(uint32_t ip, uint32_t port, char* label, int numOfThreads, uv_loop_init(srv->loop); char pipeName[PATH_MAX]; + + if (false == taosValidIpAndPort(srv->ip, srv->port)) { + terrno = TAOS_SYSTEM_ERROR(errno); + tError("invalid ip/port, %d:%d, reason:%s", srv->ip, srv->port, terrstr()); + goto End; + } + #if defined(WINDOWS) || defined(DARWIN) int ret = uv_pipe_init(srv->loop, &srv->pipeListen, 0); if (ret != 0) { @@ -1087,12 +1094,6 @@ void* transInitServer(uint32_t ip, uint32_t port, char* label, int numOfThreads, } #endif - if (false == taosValidIpAndPort(srv->ip, srv->port)) { - terrno = TAOS_SYSTEM_ERROR(errno); - tError("invalid ip/port, %d:%d, reason:%s", srv->ip, srv->port, terrstr()); - goto End; - } - if (false == addHandleToAcceptloop(srv)) { goto End; } @@ -1185,8 +1186,8 @@ void transCloseServer(void* arg) { // impl later SServerObj* srv = arg; - tDebug("send quit msg to accept thread"); if (srv->inited) { + tDebug("send quit msg to accept thread"); uv_async_send(srv->pAcceptAsync); taosThreadJoin(srv->thread, NULL); SRV_RELEASE_UV(srv->loop); diff --git a/source/util/src/terror.c b/source/util/src/terror.c index ff61c7cdc4..6f8b0d8e04 100644 --- a/source/util/src/terror.c +++ b/source/util/src/terror.c @@ -643,13 +643,10 @@ const char* tstrerror(int32_t err) { // this is a system errno if ((err & 0x00ff0000) == 0x00ff0000) { int32_t code = err & 0x0000ffff; - if (code >= 0 && code < 36) { - return strerror(code); - } else { - return "unknown err"; - } + // strerror can handle any invalid code + // invalid code return Unknown error + return strerror(code); } - int32_t s = 0; int32_t e = sizeof(errors) / sizeof(errors[0]); diff --git a/source/util/src/tworker.c b/source/util/src/tworker.c index 863cee9b08..a9a84c1860 100644 --- a/source/util/src/tworker.c +++ b/source/util/src/tworker.c @@ -36,7 +36,7 @@ int32_t tQWorkerInit(SQWorkerPool *pool) { worker->pool = pool; } - uDebug("worker:%s is initialized, min:%d max:%d", pool->name, pool->min, pool->max); + uInfo("worker:%s is initialized, min:%d max:%d", pool->name, pool->min, pool->max); return 0; } @@ -51,8 +51,10 @@ void tQWorkerCleanup(SQWorkerPool *pool) { for (int32_t i = 0; i < pool->max; ++i) { SQWorker *worker = pool->workers + i; if (taosCheckPthreadValid(worker->thread)) { + uInfo("worker:%s:%d is stopping", pool->name, worker->id); taosThreadJoin(worker->thread, NULL); taosThreadClear(&worker->thread); + uInfo("worker:%s:%d is stopped", pool->name, worker->id); } } @@ -60,7 +62,7 @@ void tQWorkerCleanup(SQWorkerPool *pool) { taosCloseQset(pool->qset); taosThreadMutexDestroy(&pool->mutex); - uDebug("worker:%s is closed", pool->name); + uInfo("worker:%s is closed", pool->name); } static void *tQWorkerThreadFp(SQWorker *worker) { @@ -119,7 +121,7 @@ STaosQueue *tQWorkerAllocQueue(SQWorkerPool *pool, void *ahandle, FItem fp) { taosThreadAttrDestroy(&thAttr); pool->num++; - uDebug("worker:%s:%d is launched, total:%d", pool->name, worker->id, pool->num); + uInfo("worker:%s:%d is launched, total:%d", pool->name, worker->id, pool->num); } while (pool->num < pool->min); } @@ -130,7 +132,134 @@ STaosQueue *tQWorkerAllocQueue(SQWorkerPool *pool, void *ahandle, FItem fp) { } void tQWorkerFreeQueue(SQWorkerPool *pool, STaosQueue *queue) { - uDebug("worker:%s, queue:%p is freed", pool->name, queue); + uInfo("worker:%s, queue:%p is freed", pool->name, queue); + taosCloseQueue(queue); +} + +int32_t tAutoQWorkerInit(SAutoQWorkerPool *pool) { + pool->qset = taosOpenQset(); + pool->workers = taosArrayInit(2, sizeof(SQWorker *)); + if (pool->workers == NULL) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + return -1; + } + + (void)taosThreadMutexInit(&pool->mutex, NULL); + + uInfo("worker:%s is initialized as auto", pool->name); + return 0; +} + +void tAutoQWorkerCleanup(SAutoQWorkerPool *pool) { + int32_t size = taosArrayGetSize(pool->workers); + for (int32_t i = 0; i < size; ++i) { + SQWorker *worker = taosArrayGetP(pool->workers, i); + if (taosCheckPthreadValid(worker->thread)) { + taosQsetThreadResume(pool->qset); + } + } + + for (int32_t i = 0; i < size; ++i) { + SQWorker *worker = taosArrayGetP(pool->workers, i); + if (taosCheckPthreadValid(worker->thread)) { + uInfo("worker:%s:%d is stopping", pool->name, worker->id); + taosThreadJoin(worker->thread, NULL); + taosThreadClear(&worker->thread); + uInfo("worker:%s:%d is stopped", pool->name, worker->id); + } + taosMemoryFree(worker); + } + + taosArrayDestroy(pool->workers); + taosCloseQset(pool->qset); + taosThreadMutexDestroy(&pool->mutex); + + uInfo("worker:%s is closed", pool->name); +} + +static void *tAutoQWorkerThreadFp(SQWorker *worker) { + SAutoQWorkerPool *pool = worker->pool; + SQueueInfo qinfo = {0}; + void *msg = NULL; + int32_t code = 0; + + taosBlockSIGPIPE(); + setThreadName(pool->name); + worker->pid = taosGetSelfPthreadId(); + uInfo("worker:%s:%d is running, thread:%08" PRId64, pool->name, worker->id, worker->pid); + + while (1) { + if (taosReadQitemFromQset(pool->qset, (void **)&msg, &qinfo) == 0) { + uInfo("worker:%s:%d qset:%p, got no message and exiting, thread:%08" PRId64, pool->name, worker->id, pool->qset, + worker->pid); + break; + } + + if (qinfo.fp != NULL) { + qinfo.workerId = worker->id; + qinfo.threadNum = taosArrayGetSize(pool->workers); + (*((FItem)qinfo.fp))(&qinfo, msg); + } + + taosUpdateItemSize(qinfo.queue, 1); + } + + return NULL; +} + +STaosQueue *tAutoQWorkerAllocQueue(SAutoQWorkerPool *pool, void *ahandle, FItem fp) { + STaosQueue *queue = taosOpenQueue(); + if (queue == NULL) return NULL; + + taosThreadMutexLock(&pool->mutex); + taosSetQueueFp(queue, fp, NULL); + taosAddIntoQset(pool->qset, queue, ahandle); + + int32_t queueNum = taosGetQueueNumber(pool->qset); + int32_t curWorkerNum = taosArrayGetSize(pool->workers); + int32_t dstWorkerNum = ceil(queueNum * pool->ratio); + if (dstWorkerNum < 1) dstWorkerNum = 1; + + // spawn a thread to process queue + while (curWorkerNum < dstWorkerNum) { + SQWorker *worker = taosMemoryCalloc(1, sizeof(SQWorker)); + if (worker == NULL || taosArrayPush(pool->workers, &worker) == NULL) { + uError("worker:%s:%d failed to create", pool->name, curWorkerNum); + taosMemoryFree(worker); + taosCloseQueue(queue); + terrno = TSDB_CODE_OUT_OF_MEMORY; + return NULL; + } + worker->id = curWorkerNum; + worker->pool = pool; + + TdThreadAttr thAttr; + taosThreadAttrInit(&thAttr); + taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE); + + if (taosThreadCreate(&worker->thread, &thAttr, (ThreadFp)tAutoQWorkerThreadFp, worker) != 0) { + uError("worker:%s:%d failed to create thread, total:%d", pool->name, worker->id, curWorkerNum); + (void)taosArrayPop(pool->workers); + taosMemoryFree(worker); + taosCloseQueue(queue); + terrno = TSDB_CODE_OUT_OF_MEMORY; + return NULL; + } + + taosThreadAttrDestroy(&thAttr); + uInfo("worker:%s:%d is launched, total:%d", pool->name, worker->id, (int32_t)taosArrayGetSize(pool->workers)); + + curWorkerNum++; + } + + taosThreadMutexUnlock(&pool->mutex); + uInfo("worker:%s, queue:%p is allocated, ahandle:%p", pool->name, queue, ahandle); + + return queue; +} + +void tAutoQWorkerFreeQueue(SAutoQWorkerPool *pool, STaosQueue *queue) { + uInfo("worker:%s, queue:%p is freed", pool->name, queue); taosCloseQueue(queue); } @@ -152,7 +281,7 @@ int32_t tWWorkerInit(SWWorkerPool *pool) { worker->pool = pool; } - uDebug("worker:%s is initialized, max:%d", pool->name, pool->max); + uInfo("worker:%s is initialized, max:%d", pool->name, pool->max); return 0; } @@ -169,17 +298,19 @@ void tWWorkerCleanup(SWWorkerPool *pool) { for (int32_t i = 0; i < pool->max; ++i) { SWWorker *worker = pool->workers + i; if (taosCheckPthreadValid(worker->thread)) { + uInfo("worker:%s:%d is stopping", pool->name, worker->id); taosThreadJoin(worker->thread, NULL); taosThreadClear(&worker->thread); taosFreeQall(worker->qall); taosCloseQset(worker->qset); + uInfo("worker:%s:%d is stopped", pool->name, worker->id); } } taosMemoryFreeClear(pool->workers); taosThreadMutexDestroy(&pool->mutex); - uDebug("worker:%s is closed", pool->name); + uInfo("worker:%s is closed", pool->name); } static void *tWWorkerThreadFp(SWWorker *worker) { @@ -235,7 +366,7 @@ STaosQueue *tWWorkerAllocQueue(SWWorkerPool *pool, void *ahandle, FItems fp) { taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE); if (taosThreadCreate(&worker->thread, &thAttr, (ThreadFp)tWWorkerThreadFp, worker) != 0) goto _OVER; - uDebug("worker:%s:%d is launched, max:%d", pool->name, worker->id, pool->max); + uInfo("worker:%s:%d is launched, max:%d", pool->name, worker->id, pool->max); pool->nextId = (pool->nextId + 1) % pool->max; taosThreadAttrDestroy(&thAttr); @@ -259,13 +390,14 @@ _OVER: } else { while (worker->pid <= 0) taosMsleep(10); queue->threadId = worker->pid; - uInfo("worker:%s, queue:%p is allocated, ahandle:%p thread:%08" PRId64, pool->name, queue, ahandle, queue->threadId); + uInfo("worker:%s, queue:%p is allocated, ahandle:%p thread:%08" PRId64, pool->name, queue, ahandle, + queue->threadId); return queue; } } void tWWorkerFreeQueue(SWWorkerPool *pool, STaosQueue *queue) { - uDebug("worker:%s, queue:%p is freed", pool->name, queue); + uInfo("worker:%s, queue:%p is freed", pool->name, queue); taosCloseQueue(queue); } diff --git a/tests/parallel_test/cases.task b/tests/parallel_test/cases.task index 57fff095e8..f5f04ff1e9 100644 --- a/tests/parallel_test/cases.task +++ b/tests/parallel_test/cases.task @@ -677,8 +677,8 @@ ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/create_wrong_topic.py ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/dropDbR3ConflictTransaction.py -N 3 ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/basic5.py -,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/subscribeDb.py -,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/subscribeDb0.py +,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/subscribeDb.py -N 3 -n 3 +,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/subscribeDb0.py -N 3 -n 3 ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/subscribeDb1.py ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/subscribeDb2.py ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/subscribeDb3.py diff --git a/tests/script/tsim/dnode/balance1.sim b/tests/script/tsim/dnode/balance1.sim index 2b0154c8e5..3bb9c4f3eb 100644 --- a/tests/script/tsim/dnode/balance1.sim +++ b/tests/script/tsim/dnode/balance1.sim @@ -74,10 +74,10 @@ sql insert into d2.t2 values(now+5s, 21) sql select * from information_schema.ins_dnodes print dnode1 openVnodes $data(1)[2] print dnode2 openVnodes $data(2)[2] -if $data(1)[2] != 0 then +if $data(1)[2] != 1 then return -1 endi -if $data(2)[2] != 2 then +if $data(2)[2] != 1 then return -1 endi diff --git a/tests/script/tsim/dnode/balance2.sim b/tests/script/tsim/dnode/balance2.sim index 3f5a42d4d3..9eeb7e1251 100644 --- a/tests/script/tsim/dnode/balance2.sim +++ b/tests/script/tsim/dnode/balance2.sim @@ -161,13 +161,13 @@ print dnode1 openVnodes $data(1)[2] print dnode3 openVnodes $data(3)[2] print dnode4 openVnodes $data(4)[2] print dnode5 openVnodes $data(5)[2] -if $data(1)[2] != 2 then +if $data(1)[2] != 3 then return -1 endi if $data(3)[2] != 3 then return -1 endi -if $data(4)[2] != 4 then +if $data(4)[2] != 3 then return -1 endi if $data(5)[2] != 3 then diff --git a/tests/script/tsim/dnode/balance3.sim b/tests/script/tsim/dnode/balance3.sim index ce328f10bd..2fb284b466 100644 --- a/tests/script/tsim/dnode/balance3.sim +++ b/tests/script/tsim/dnode/balance3.sim @@ -127,10 +127,10 @@ print dnode1 openVnodes $data(1)[2] print dnode2 openVnodes $data(2)[2] print dnode3 openVnodes $data(3)[2] print dnode4 openVnodes $data(4)[2] -if $data(1)[2] != 0 then +if $data(1)[2] != 1 then return -1 endi -if $data(2)[2] != 2 then +if $data(2)[2] != 1 then return -1 endi if $data(3)[2] != 2 then @@ -228,10 +228,10 @@ print dnode1 openVnodes $data(1)[2] print dnode3 openVnodes $data(3)[2] print dnode4 openVnodes $data(4)[2] print dnode5 openVnodes $data(5)[2] -if $data(1)[2] != 1 then +if $data(1)[2] != 2 then return -1 endi -if $data(3)[2] != 3 then +if $data(3)[2] != 2 then return -1 endi if $data(4)[2] != 3 then diff --git a/tests/script/tsim/dnode/balancex.sim b/tests/script/tsim/dnode/balancex.sim index 6b16e8b569..0cfc64a954 100644 --- a/tests/script/tsim/dnode/balancex.sim +++ b/tests/script/tsim/dnode/balancex.sim @@ -142,10 +142,10 @@ print dnode1 openVnodes $data(1)[2] print dnode2 openVnodes $data(2)[2] print dnode2 openVnodes $data(3)[2] print dnode2 openVnodes $data(4)[2] -if $data(1)[2] != 0 then +if $data(1)[2] != 1 then return -1 endi -if $data(2)[2] != 2 then +if $data(2)[2] != 1 then return -1 endi if $data(3)[2] != 2 then diff --git a/tests/script/tsim/dnode/vnode_clean.sim b/tests/script/tsim/dnode/vnode_clean.sim index 112e5f28a4..ba1d083c68 100644 --- a/tests/script/tsim/dnode/vnode_clean.sim +++ b/tests/script/tsim/dnode/vnode_clean.sim @@ -71,10 +71,10 @@ sql insert into d2.t2 values(now+5s, 21) sql select * from information_schema.ins_dnodes print dnode1 openVnodes $data(1)[2] print dnode2 openVnodes $data(2)[2] -if $data(1)[2] != 0 then +if $data(1)[2] != 1 then return -1 endi -if $data(2)[2] != 2 then +if $data(2)[2] != 1 then return -1 endi @@ -181,10 +181,10 @@ sql select * from information_schema.ins_dnodes print dnode1 openVnodes $data(1)[2] print dnode2 openVnodes $data(3)[2] print dnode2 openVnodes $data(4)[2] -if $data(1)[2] != 0 then +if $data(1)[2] != 1 then return -1 endi -if $data(3)[2] != 2 then +if $data(3)[2] != 1 then return -1 endi if $data(4)[2] != 1 then @@ -204,10 +204,10 @@ sql select * from information_schema.ins_dnodes print dnode1 openVnodes $data(1)[2] print dnode2 openVnodes $data(3)[2] print dnode2 openVnodes $data(4)[2] -if $data(1)[2] != 0 then +if $data(1)[2] != 1 then return -1 endi -if $data(3)[2] != 2 then +if $data(3)[2] != 1 then return -1 endi if $data(4)[2] != 2 then @@ -220,13 +220,13 @@ sql select * from information_schema.ins_dnodes print dnode1 openVnodes $data(1)[2] print dnode2 openVnodes $data(3)[2] print dnode2 openVnodes $data(4)[2] -if $data(1)[2] != 1 then +if $data(1)[2] != 2 then return -1 endi if $data(3)[2] != null then return -1 endi -if $data(4)[2] != 3 then +if $data(4)[2] != 2 then return -1 endi diff --git a/tests/script/tsim/stream/basic3.sim b/tests/script/tsim/stream/basic3.sim index 48fb860a72..41e19b19af 100644 --- a/tests/script/tsim/stream/basic3.sim +++ b/tests/script/tsim/stream/basic3.sim @@ -1,7 +1,7 @@ system sh/stop_dnodes.sh system sh/deploy.sh -n dnode1 -i 1 -system sh/cfg.sh -n dnode1 -c debugflag -v 131 -system sh/exec.sh -n dnode1 -s start -v +system sh/cfg.sh -n dnode1 -c debugflag 131 +system sh/exec.sh -n dnode1 -s start sleep 5000 diff --git a/tests/script/tsim/stream/sliding.sim b/tests/script/tsim/stream/sliding.sim index c9a1ddd922..8287274cd2 100644 --- a/tests/script/tsim/stream/sliding.sim +++ b/tests/script/tsim/stream/sliding.sim @@ -672,6 +672,123 @@ if $data61 != 1 then goto loop5 endi +print step 8 + +sql drop stream IF EXISTS streams4; +sql drop database IF EXISTS test4; + +sql create database test4 vgroups 6; +sql use test4; +sql create stable st(ts timestamp, a int, b int, c int, d double) tags(ta int,tb int,tc int); +sql create table t1 using st tags(1,1,1); +sql create table t2 using st tags(2,2,2); + +sql create stream streams4 trigger at_once into streamt4 as select _wstart as ts, count(*),min(a) c1 from st interval(10s) sliding(5s); + +sql insert into t1 values(1648791213000,1,1,1,1.0); +sql insert into t1 values(1648791243000,2,1,1,1.0); + +sql insert into t2 values(1648791273000,3,1,1,1.0); +sql insert into t2 values(1648791313000,4,1,1,1.0); + +$loop_count = 0 + +loop6: +sleep 200 + +$loop_count = $loop_count + 1 +if $loop_count == 10 then + return -1 +endi + +sql select * from streamt4 order by 1; + +# row 0 +if $rows != 8 then + print ====loop6=rows=$rows + goto loop6 +endi + +if $data01 != 1 then + print ====loop6=data01=$data01 + goto loop6 +endi + +if $data02 != 1 then + print ====loop6=data02=$data02 + return -1 +endi + +if $data11 != 1 then + print ====loop6=data11=$data11 + goto loop6 +endi + +if $data12 != 1 then + print ====loop6=data12=$data12 + return -1 +endi + +if $data21 != 1 then + print ====loop6=data21=$data21 + goto loop6 +endi + +if $data22 != 2 then + print ====loop6=data22=$data22 + return -1 +endi + +if $data31 != 1 then + print ====loop6=data31=$data31 + goto loop6 +endi + +if $data32 != 2 then + print ====loop6=data32=$data32 + return -1 +endi + +if $data41 != 1 then + print ====loop6=data41=$data41 + goto loop6 +endi + +if $data42 != 3 then + print ====loop6=data42=$data42 + return -1 +endi + +if $data51 != 1 then + print ====loop6=data51=$data51 + goto loop6 +endi + +if $data52 != 3 then + print ====loop6=data52=$data52 + return -1 +endi + +if $data61 != 1 then + print ====loop6=data61=$data61 + return -1 +endi + +if $data62 != 4 then + print ====loop6=data62=$data62 + return -1 +endi + +if $data71 != 1 then + print ====loop6=data71=$data71 + return -1 +endi + +if $data72 != 4 then + print ====loop6=data72=$data72 + return -1 +endi + $loop_all = $loop_all + 1 print ============loop_all=$loop_all diff --git a/tests/system-test/7-tmq/subscribeDb.py b/tests/system-test/7-tmq/subscribeDb.py index fd06eedefd..0fa9bcfbd4 100644 --- a/tests/system-test/7-tmq/subscribeDb.py +++ b/tests/system-test/7-tmq/subscribeDb.py @@ -61,7 +61,7 @@ class TDTestCase: def insertConsumerInfo(self,consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifmanualcommit,cdbName='cdb'): sql = "insert into %s.consumeinfo values "%cdbName - sql += "(now, %d, '%s', '%s', %d, %d, %d)"%(consumerId, topicList, keyList, expectrowcnt, ifcheckdata, ifmanualcommit) + sql += "(now + %ds, %d, '%s', '%s', %d, %d, %d)"%(consumerId, consumerId, topicList, keyList, expectrowcnt, ifcheckdata, ifmanualcommit) tdLog.info("consume info sql: %s"%sql) tdSql.query(sql) @@ -174,12 +174,13 @@ class TDTestCase: 'ctbNum': 10, \ 'rowsPerTbl': 5000, \ 'batchNum': 100, \ + 'replica': self.replicaVar, \ 'startTs': 1640966400000} # 2022-01-01 00:00:00.000 parameterDict['cfg'] = cfgPath self.initConsumerTable() - tdSql.execute("create database if not exists %s vgroups %d" %(parameterDict['dbName'], parameterDict['vgroups'])) + tdSql.execute("create database if not exists %s vgroups %d replica %d" %(parameterDict['dbName'], parameterDict['vgroups'], parameterDict['replica'])) prepareEnvThread = threading.Thread(target=self.prepareEnv, kwargs=parameterDict) prepareEnvThread.start() @@ -271,12 +272,13 @@ class TDTestCase: 'ctbNum': 10, \ 'rowsPerTbl': 5000, \ 'batchNum': 100, \ + 'replica': self.replicaVar, \ 'startTs': 1640966400000} # 2022-01-01 00:00:00.000 parameterDict['cfg'] = cfgPath self.initConsumerTable() - tdSql.execute("create database if not exists %s vgroups %d" %(parameterDict['dbName'], parameterDict['vgroups'])) + tdSql.execute("create database if not exists %s vgroups %d replica %d" %(parameterDict['dbName'], parameterDict['vgroups'], parameterDict['replica'])) prepareEnvThread = threading.Thread(target=self.prepareEnv, kwargs=parameterDict) prepareEnvThread.start() @@ -337,6 +339,7 @@ class TDTestCase: 'ctbNum': 10, \ 'rowsPerTbl': 5000, \ 'batchNum': 100, \ + 'replica': self.replicaVar, \ 'startTs': 1640966400000} # 2022-01-01 00:00:00.000 parameterDict['cfg'] = cfgPath @@ -406,12 +409,13 @@ class TDTestCase: 'ctbNum': 10, \ 'rowsPerTbl': 5000, \ 'batchNum': 100, \ + 'replica': self.replicaVar, \ 'startTs': 1640966400000} # 2022-01-01 00:00:00.000 parameterDict['cfg'] = cfgPath self.initConsumerTable() - tdSql.execute("create database if not exists %s vgroups %d" %(parameterDict['dbName'], parameterDict['vgroups'])) + tdSql.execute("create database if not exists %s vgroups %d replica %d" %(parameterDict['dbName'], parameterDict['vgroups'], parameterDict['replica'])) prepareEnvThread = threading.Thread(target=self.prepareEnv, kwargs=parameterDict) prepareEnvThread.start() diff --git a/tests/system-test/7-tmq/subscribeDb0.py b/tests/system-test/7-tmq/subscribeDb0.py index d4c5e2f87f..50ef52cb15 100644 --- a/tests/system-test/7-tmq/subscribeDb0.py +++ b/tests/system-test/7-tmq/subscribeDb0.py @@ -174,12 +174,13 @@ class TDTestCase: 'ctbNum': 10, \ 'rowsPerTbl': 5000, \ 'batchNum': 100, \ + 'replica': self.replicaVar, \ 'startTs': 1640966400000} # 2022-01-01 00:00:00.000 parameterDict['cfg'] = cfgPath self.initConsumerTable() - tdSql.execute("create database if not exists %s vgroups %d" %(parameterDict['dbName'], parameterDict['vgroups'])) + tdSql.execute("create database if not exists %s vgroups %d replica %d" %(parameterDict['dbName'], parameterDict['vgroups'], parameterDict['replica'])) prepareEnvThread = threading.Thread(target=self.prepareEnv, kwargs=parameterDict) prepareEnvThread.start() @@ -191,6 +192,7 @@ class TDTestCase: 'ctbNum': 10, \ 'rowsPerTbl': 5000, \ 'batchNum': 100, \ + 'replica': self.replicaVar, \ 'startTs': 1640966400000} # 2022-01-01 00:00:00.000 parameterDict['cfg'] = cfgPath @@ -254,12 +256,13 @@ class TDTestCase: 'ctbNum': 10, \ 'rowsPerTbl': 5000, \ 'batchNum': 100, \ + 'replica': self.replicaVar, \ 'startTs': 1640966400000} # 2022-01-01 00:00:00.000 parameterDict['cfg'] = cfgPath self.initConsumerTable() - tdSql.execute("create database if not exists %s vgroups %d" %(parameterDict['dbName'], parameterDict['vgroups'])) + tdSql.execute("create database if not exists %s vgroups %d replica %d" %(parameterDict['dbName'], parameterDict['vgroups'], parameterDict['replica'])) prepareEnvThread = threading.Thread(target=self.prepareEnv, kwargs=parameterDict) prepareEnvThread.start() diff --git a/tests/system-test/7-tmq/tmqCommon.py b/tests/system-test/7-tmq/tmqCommon.py index 141d013270..4cda062401 100644 --- a/tests/system-test/7-tmq/tmqCommon.py +++ b/tests/system-test/7-tmq/tmqCommon.py @@ -60,7 +60,7 @@ class TMQCom: def insertConsumerInfo(self,consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifmanualcommit,cdbName='cdb'): sql = "insert into %s.consumeinfo values "%cdbName - sql += "(now, %d, '%s', '%s', %d, %d, %d)"%(consumerId, topicList, keyList, expectrowcnt, ifcheckdata, ifmanualcommit) + sql += "(now + %ds, %d, '%s', '%s', %d, %d, %d)"%(consumerId, consumerId, topicList, keyList, expectrowcnt, ifcheckdata, ifmanualcommit) tdLog.info("consume info sql: %s"%sql) tdSql.query(sql) diff --git a/tests/system-test/compatibilityAllTest.sh b/tests/system-test/compatibilityAllTest.sh new file mode 100755 index 0000000000..8b599afd86 --- /dev/null +++ b/tests/system-test/compatibilityAllTest.sh @@ -0,0 +1,46 @@ +#!/bin/bash +ulimit -c unlimited +#======================p1-insert=============== + +python3 ./test.py -f 0-others/taosShell.py +python3 ./test.py -f 0-others/taosShellError.py +python3 ./test.py -f 0-others/taosShellNetChk.py +python3 ./test.py -f 1-insert/alter_database.py +python3 ./test.py -f 1-insert/influxdb_line_taosc_insert.py +python3 ./test.py -f 1-insert/opentsdb_telnet_line_taosc_insert.py +python3 ./test.py -f 1-insert/opentsdb_json_taosc_insert.py +python3 ./test.py -f 1-insert/test_stmt_muti_insert_query.py +python3 ./test.py -f 1-insert/test_stmt_set_tbname_tag.py +python3 ./test.py -f 1-insert/alter_stable.py +python3 ./test.py -f 1-insert/alter_table.py +python3 ./test.py -f 1-insert/boundary.py +python3 ./test.py -f 2-query/top.py +python3 ./test.py -f 2-query/top.py -R +python3 ./test.py -f 2-query/tsbsQuery.py +python3 ./test.py -f 2-query/tsbsQuery.py -R +python3 ./test.py -f 2-query/ttl_comment.py +python3 ./test.py -f 2-query/ttl_comment.py -R +python3 ./test.py -f 2-query/twa.py +python3 ./test.py -f 2-query/twa.py -R +python3 ./test.py -f 2-query/union.py +python3 ./test.py -f 2-query/union.py -R +python3 ./test.py -f 6-cluster/5dnode1mnode.py +python3 ./test.py -f 6-cluster/5dnode2mnode.py -N 5 +python3 ./test.py -f 6-cluster/5dnode3mnodeStop.py -N 5 -M 3 +python3 ./test.py -f 6-cluster/5dnode3mnodeStop.py -N 5 -M 3 -i False +python3 ./test.py -f 6-cluster/5dnode3mnodeStop2Follower.py -N 5 -M 3 +python3 ./test.py -f 6-cluster/5dnode3mnodeStop2Follower.py -N 5 -M 3 -i False +python3 ./test.py -f 6-cluster/5dnode3mnodeStopLoop.py -N 5 -M 3 +python3 ./test.py -f 6-cluster/5dnode3mnodeSep1VnodeStopDnodeCreateDb.py -N 6 -M 3 +python3 ./test.py -f 6-cluster/5dnode3mnodeSep1VnodeStopDnodeCreateDb.py -N 6 -M 3 -n 3 +python3 ./test.py -f 7-tmq/subscribeStb4.py +python3 ./test.py -f 7-tmq/db.py +python3 ./test.py -f 7-tmq/tmqError.py +python3 ./test.py -f 7-tmq/schema.py +python3 ./test.py -f 7-tmq/stbFilter.py +python3 ./test.py -f 7-tmq/tmqCheckData.py +python3 ./test.py -f 7-tmq/tmqCheckData1.py +python3 ./test.py -f 7-tmq/tmqConsumerGroup.py +python3 ./test.py -f 7-tmq/tmqShow.py +python3 ./test.py -f 7-tmq/tmqAlterSchema.py +python3 ./test.py -f 99-TDcase/TD-20582.py