xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge branch 'main' into FIX/TD-21043-main

This commit is contained in:
Benguang Zhao 2022-12-29 10:03:17 +08:00
parent 044e58b712 6e9b5d1be6
commit d307413c36
57 changed files with 760 additions and 471 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
cmake/cmake.define
View File

@ -141,12 +141,12 @@ ELSE ()
SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -msse4.2")
SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -msse4.2")
ENDIF()
IF ("${SIMD_SUPPORT}" MATCHES "true")
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)
SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -mavx")
SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -mavx")

2
cmake/taostools_CMakeLists.txt.in
View File

@ -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

2
docs/en/14-reference/12-config/index.md
View File

@ -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 |

2
docs/zh/14-reference/12-config/index.md
View File

@ -709,7 +709,7 @@ charset 的有效值是 UTF-8。
| 42 | numOfCommitThreads | 是 | 是 | |
| 43 | numOfMnodeReadThreads | 否 | 是 | |
| 44 | numOfVnodeQueryThreads | 否 | 是 | |
| 45 | numOfVnodeStreamThreads | 否 | 是 | |
| 45 | ratioOfVnodeStreamThreads | 否 | 是 | |
| 46 | numOfVnodeFetchThreads | 否 | 是 | |
| 47 | numOfVnodeRsmaThreads | 否 | 是 | |
| 48 | numOfQnodeQueryThreads | 否 | 是 | |

2
include/common/tglobal.h
View File

@ -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;

1
include/libs/executor/executor.h
View File

@ -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
}

16
include/util/tworker.h
View File

@ -17,6 +17,7 @@
#define _TD_UTIL_WORKER_H_
#include "tqueue.h"
#include "tarray.h"
#ifdef __cplusplus
extern "C" {
@ -29,7 +30,7 @@ typedef struct SQWorker {
int32_t id; // worker id
int64_t pid; // thread pid
TdThread thread; // thread id
SQWorkerPool *pool;
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);

20
source/common/src/tglobal.c
View File

@ -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;
}
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);
}

5
source/dnode/mgmt/exe/dmMain.c
View File

@ -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;
}

2
source/dnode/mgmt/mgmt_vnode/inc/vmInt.h
View File

@ -31,7 +31,7 @@ typedef struct SVnodeMgmt {
const char *path;
const char *name;
SQWorkerPool queryPool;
SQWorkerPool streamPool;
SAutoQWorkerPool streamPool;
SWWorkerPool fetchPool;
SSingleWorker mgmtWorker;
SHashObj *hash;

21
source/dnode/mgmt/mgmt_vnode/src/vmWorker.c
View File

@ -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");
}

10
source/dnode/mgmt/node_mgmt/src/dmTransport.c
View File

@ -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) &&
} 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;

1
source/dnode/mnode/impl/inc/mndDef.h
View File

@ -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;

4
source/dnode/mnode/impl/src/mndDnode.c
View File

@ -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:

37
source/dnode/mnode/impl/src/mndVgroup.c
View File

@ -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) {

21
source/dnode/vnode/src/meta/metaCache.c
View File

@ -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);

4
source/dnode/vnode/src/tq/tq.c
View File

@ -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);

1
source/dnode/vnode/src/tsdb/tsdbMergeTree.c
View File

@ -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;

162
source/dnode/vnode/src/tsdb/tsdbRead.c
View File

@ -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,9 +1301,8 @@ 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);
if (ret != TSDB_CODE_SUCCESS) {
cleanupBlockOrderSupporter(&sup);
@ -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,11 +4102,8 @@ 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) {
if (pDataBlock->pDataBlock) {
size_t colsNum = taosArrayGetSize(pDataBlock->pDataBlock);
taosArrayEnsureCap(pSup->pColAgg, colsNum);
}
@ -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);

2
source/dnode/vnode/src/vnd/vnodeSvr.c
View File

@ -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;
}

10
source/libs/executor/inc/executorimpl.h
View File

@ -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;
};
@ -263,7 +264,7 @@ typedef struct SExchangeInfo {
// SArray<SSDataBlock*>, 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.
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;
@ -468,6 +469,11 @@ typedef struct SStreamScanInfo {
SNodeList* pGroupTags;
SNode* pTagCond;
SNode* pTagIndexCond;
// recover
int32_t blockRecoverContiCnt;
int32_t blockRecoverTotCnt;
} SStreamScanInfo;
typedef struct {

5
source/libs/executor/src/dataDeleter.c
View File

@ -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);

13
source/libs/executor/src/dataDispatcher.c
View File

@ -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);

1
source/libs/executor/src/exchangeoperator.c
View File

@ -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 {

4
source/libs/executor/src/executor.c
View File

@ -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;

42
source/libs/executor/src/executorimpl.c
View File

@ -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;

24
source/libs/executor/src/joinoperator.c
View File

@ -42,19 +42,23 @@ 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) {
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 ) {
if (col1->dataBlockId == col2->dataBlockId) {
leftTsCol = col1;
rightTsCol = col2;
} else {
@ -71,9 +75,7 @@ static void extractTimeCondition(SJoinOperatorInfo* pInfo, SOperatorInfo** pDown
}
setJoinColumnInfo(&pInfo->leftCol, leftTsCol);
setJoinColumnInfo(&pInfo->rightCol, rightTsCol);
} else {
ASSERT(false);
}}
}
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;
}

62
source/libs/executor/src/scanoperator.c
View File

@ -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);
}
@ -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;
@ -2517,7 +2506,8 @@ static SSDataBlock* getTableDataBlockImpl(void* param) {
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);
}
@ -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:

15
source/libs/executor/src/sortoperator.c
View File

@ -47,12 +47,14 @@ SOperatorInfo* createSortOperatorInfo(SOperatorInfo* downstream, SSortPhysiNode*
SDataBlockDescNode* pDescNode = pSortNode->node.pOutputDataBlockDesc;
int32_t numOfCols = 0;
SSDataBlock* pResBlock = createDataBlockFromDescNode(pDescNode);
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;

15
source/libs/executor/src/timewindowoperator.c
View File

@ -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);
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;
}

5
source/libs/executor/src/tsimplehash.c
View File

@ -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) {

1
source/libs/executor/src/tsort.c
View File

@ -800,6 +800,7 @@ STupleHandle* tsortNextTuple(SSortHandle* pHandle) {
}
}
// all sources are completed.
if (pHandle->cmpParam.numOfSources == pHandle->numOfCompletedSources) {
return NULL;
}

46
source/libs/function/src/tudf.c
View File

@ -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);
output->info.rows = numOfRows;
if (IS_VAR_DATA_TYPE((input + i)->columnData->info.type)) {
output->info.hasVarCol = true;
}
}
return 0;
}
@ -1825,7 +1831,7 @@ int32_t doCallUdfScalarFunc(UdfcFuncHandle handle, SScalarParam *input, int32_t
taosArrayDestroy(resultBlock.pDataBlock);
}
taosArrayDestroy(inputBlock.pDataBlock);
blockDataFreeRes(&inputBlock);
return err;
}

2
source/libs/parser/inc/parAst.h
View File

@ -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);

5
source/libs/parser/src/parAstCreater.c
View File

@ -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);

9
source/libs/parser/src/parTranslater.c
View File

@ -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);

6
source/libs/scheduler/src/schJob.c
View File

@ -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) {

3
source/libs/stream/src/streamDispatch.c
View File

@ -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);

4
source/libs/stream/src/streamExec.c
View File

@ -116,7 +116,11 @@ int32_t streamScanExec(SStreamTask* pTask, int32_t batchSz) {
ASSERT(0);
}
if (output == NULL) {
if (qStreamRecoverScanFinished(exec)) {
finished = true;
} else {
qSetStreamOpOpen(exec);
}
break;
}

1
source/libs/stream/src/streamRecover.c
View File

@ -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) {

2
source/libs/stream/src/streamTask.c
View File

@ -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);

5
source/libs/sync/src/syncPipeline.c
View File

@ -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;
}

15
source/libs/transport/src/transSvr.c
View File

@ -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);

7
source/util/src/terror.c
View File

@ -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) {
// strerror can handle any invalid code
// invalid code return Unknown error
return strerror(code);
} else {
return "unknown err";
}
}
int32_t s = 0;
int32_t e = sizeof(errors) / sizeof(errors[0]);

150
source/util/src/tworker.c
View File

@ -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);
}

4
tests/parallel_test/cases.task
View File

@ -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

4
tests/script/tsim/dnode/balance1.sim
View File

@ -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

4
tests/script/tsim/dnode/balance2.sim
View File

@ -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

8
tests/script/tsim/dnode/balance3.sim
View File

@ -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

4
tests/script/tsim/dnode/balancex.sim
View File

@ -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

16
tests/script/tsim/dnode/vnode_clean.sim
View File

@ -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

4
tests/script/tsim/stream/basic3.sim
View File

@ -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

117
tests/script/tsim/stream/sliding.sim
View File

@ -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

12
tests/system-test/7-tmq/subscribeDb.py
View File

@ -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()

7
tests/system-test/7-tmq/subscribeDb0.py
View File

@ -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()

2
tests/system-test/7-tmq/tmqCommon.py
View File

@ -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)

46
tests/system-test/compatibilityAllTest.sh Executable file
View File

@ -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