diff --git a/include/common/tdatablock.h b/include/common/tdatablock.h index b2aff6fd7e..bcb40f4175 100644 --- a/include/common/tdatablock.h +++ b/include/common/tdatablock.h @@ -262,6 +262,9 @@ char* dumpBlockData(SSDataBlock* pDataBlock, const char* flag, char** dumpBuf, c int32_t buildSubmitReqFromDataBlock(SSubmitReq2** pReq, const SSDataBlock* pDataBlocks, const STSchema* pTSchema, int64_t uid, int32_t vgId, tb_uid_t suid); +bool alreadyAddGroupId(char* ctbName); +bool isAutoTableName(char* ctbName); +void buildCtbNameAddGruopId(char* ctbName, uint64_t groupId); char* buildCtbNameByGroupId(const char* stbName, uint64_t groupId); int32_t buildCtbNameByGroupIdImpl(const char* stbName, uint64_t groupId, char* pBuf); diff --git a/include/libs/qcom/query.h b/include/libs/qcom/query.h index 8b5bf1abb1..f79a0a0718 100644 --- a/include/libs/qcom/query.h +++ b/include/libs/qcom/query.h @@ -21,12 +21,12 @@ extern "C" { #endif +#include "systable.h" #include "tarray.h" #include "thash.h" #include "tlog.h" #include "tmsg.h" #include "tmsgcb.h" -#include "systable.h" typedef enum { JOB_TASK_STATUS_NULL = 0, @@ -90,8 +90,7 @@ typedef struct SExecResult { void* res; } SExecResult; - -#pragma pack(push, 1) +#pragma pack(push, 1) typedef struct SCTableMeta { uint64_t uid; uint64_t suid; @@ -100,8 +99,7 @@ typedef struct SCTableMeta { } SCTableMeta; #pragma pack(pop) - -#pragma pack(push, 1) +#pragma pack(push, 1) typedef struct STableMeta { // BEGIN: KEEP THIS PART SAME WITH SCTableMeta uint64_t uid; @@ -137,8 +135,8 @@ typedef struct SDBVgInfo { int8_t hashMethod; int32_t numOfTable; // DB's table num, unit is TSDB_TABLE_NUM_UNIT int64_t stateTs; - SHashObj* vgHash; // key:vgId, value:SVgroupInfo - SArray* vgArray; // SVgroupInfo + SHashObj* vgHash; // key:vgId, value:SVgroupInfo + SArray* vgArray; // SVgroupInfo } SDBVgInfo; typedef struct SUseDbOutput { @@ -173,6 +171,7 @@ typedef struct SDataBuf { void* pData; uint32_t len; void* handle; + int64_t handleRefId; SEpSet* pEpSet; } SDataBuf; @@ -283,7 +282,7 @@ void getColumnTypeFromMeta(STableMeta* pMeta, char* pName, ETableColumnType* int32_t cloneDbVgInfo(SDBVgInfo* pSrc, SDBVgInfo** pDst); int32_t cloneSVreateTbReq(SVCreateTbReq* pSrc, SVCreateTbReq** pDst); void freeVgInfo(SDBVgInfo* vgInfo); -void freeDbCfgInfo(SDbCfgInfo *pInfo); +void freeDbCfgInfo(SDbCfgInfo* pInfo); extern int32_t (*queryBuildMsg[TDMT_MAX])(void* input, char** msg, int32_t msgSize, int32_t* msgLen, void* (*mallocFp)(int64_t)); @@ -314,7 +313,9 @@ extern int32_t (*queryProcessMsgRsp[TDMT_MAX])(void* output, char* msg, int32_t ((_code) == TSDB_CODE_SYN_NOT_LEADER || (_code) == TSDB_CODE_SYN_RESTORING || (_code) == TSDB_CODE_SYN_INTERNAL_ERROR) #define SYNC_OTHER_LEADER_REDIRECT_ERROR(_code) ((_code) == TSDB_CODE_MNODE_NOT_FOUND) -#define NO_RET_REDIRECT_ERROR(_code) ((_code) == TSDB_CODE_RPC_BROKEN_LINK || (_code) == TSDB_CODE_RPC_NETWORK_UNAVAIL || (_code) == TSDB_CODE_RPC_SOMENODE_NOT_CONNECTED) +#define NO_RET_REDIRECT_ERROR(_code) \ + ((_code) == TSDB_CODE_RPC_BROKEN_LINK || (_code) == TSDB_CODE_RPC_NETWORK_UNAVAIL || \ + (_code) == TSDB_CODE_RPC_SOMENODE_NOT_CONNECTED) #define NEED_REDIRECT_ERROR(_code) \ (NO_RET_REDIRECT_ERROR(_code) || SYNC_UNKNOWN_LEADER_REDIRECT_ERROR(_code) || \ diff --git a/include/libs/stream/tstream.h b/include/libs/stream/tstream.h index 05462f1fd8..c6923a2233 100644 --- a/include/libs/stream/tstream.h +++ b/include/libs/stream/tstream.h @@ -62,9 +62,10 @@ typedef struct SStreamTask SStreamTask; typedef struct SStreamQueue SStreamQueue; typedef struct SStreamTaskSM SStreamTaskSM; -#define SSTREAM_TASK_VER 2 +#define SSTREAM_TASK_VER 3 #define SSTREAM_TASK_INCOMPATIBLE_VER 1 #define SSTREAM_TASK_NEED_CONVERT_VER 2 +#define SSTREAM_TASK_SUBTABLE_CHANGED_VER 3 enum { STREAM_STATUS__NORMAL = 0, diff --git a/source/client/src/clientImpl.c b/source/client/src/clientImpl.c index 28a29c2138..d4f89d6b54 100644 --- a/source/client/src/clientImpl.c +++ b/source/client/src/clientImpl.c @@ -285,8 +285,8 @@ int32_t parseSql(SRequestObj* pRequest, bool topicQuery, SQuery** pQuery, SStmtC int32_t execLocalCmd(SRequestObj* pRequest, SQuery* pQuery) { SRetrieveTableRsp* pRsp = NULL; - int8_t biMode = atomic_load_8(&pRequest->pTscObj->biMode); - int32_t code = qExecCommand(&pRequest->pTscObj->id, pRequest->pTscObj->sysInfo, pQuery->pRoot, &pRsp, biMode); + int8_t biMode = atomic_load_8(&pRequest->pTscObj->biMode); + int32_t code = qExecCommand(&pRequest->pTscObj->id, pRequest->pTscObj->sysInfo, pQuery->pRoot, &pRsp, biMode); if (TSDB_CODE_SUCCESS == code && NULL != pRsp) { code = setQueryResultFromRsp(&pRequest->body.resInfo, pRsp, false, true); } @@ -324,7 +324,8 @@ void asyncExecLocalCmd(SRequestObj* pRequest, SQuery* pQuery) { return; } - int32_t code = qExecCommand(&pRequest->pTscObj->id, pRequest->pTscObj->sysInfo, pQuery->pRoot, &pRsp, atomic_load_8(&pRequest->pTscObj->biMode)); + int32_t code = qExecCommand(&pRequest->pTscObj->id, pRequest->pTscObj->sysInfo, pQuery->pRoot, &pRsp, + atomic_load_8(&pRequest->pTscObj->biMode)); if (TSDB_CODE_SUCCESS == code && NULL != pRsp) { code = setQueryResultFromRsp(&pRequest->body.resInfo, pRsp, false, true); } @@ -492,7 +493,8 @@ void setResSchemaInfo(SReqResultInfo* pResInfo, const SSchema* pSchema, int32_t pResInfo->userFields[i].bytes = pSchema[i].bytes; pResInfo->userFields[i].type = pSchema[i].type; - if (pSchema[i].type == TSDB_DATA_TYPE_VARCHAR || pSchema[i].type == TSDB_DATA_TYPE_VARBINARY || pSchema[i].type == TSDB_DATA_TYPE_GEOMETRY) { + if (pSchema[i].type == TSDB_DATA_TYPE_VARCHAR || pSchema[i].type == TSDB_DATA_TYPE_VARBINARY || + pSchema[i].type == TSDB_DATA_TYPE_GEOMETRY) { pResInfo->userFields[i].bytes -= VARSTR_HEADER_SIZE; } else if (pSchema[i].type == TSDB_DATA_TYPE_NCHAR || pSchema[i].type == TSDB_DATA_TYPE_JSON) { pResInfo->userFields[i].bytes = (pResInfo->userFields[i].bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE; @@ -1103,9 +1105,9 @@ static int32_t asyncExecSchQuery(SRequestObj* pRequest, SQuery* pQuery, SMetaDat SSqlCallbackWrapper* pWrapper) { int32_t code = TSDB_CODE_SUCCESS; pRequest->type = pQuery->msgType; - SArray* pMnodeList = NULL; - SQueryPlan* pDag = NULL; - int64_t st = taosGetTimestampUs(); + SArray* pMnodeList = NULL; + SQueryPlan* pDag = NULL; + int64_t st = taosGetTimestampUs(); if (!pRequest->parseOnly) { pMnodeList = taosArrayInit(4, sizeof(SQueryNodeLoad)); @@ -1278,7 +1280,7 @@ int32_t removeMeta(STscObj* pTscObj, SArray* tbList, bool isView) { if (isView) { for (int32_t i = 0; i < tbNum; ++i) { SName* pViewName = taosArrayGet(tbList, i); - char dbFName[TSDB_DB_FNAME_LEN]; + char dbFName[TSDB_DB_FNAME_LEN]; tNameGetFullDbName(pViewName, dbFName); catalogRemoveViewMeta(pCatalog, dbFName, 0, pViewName->tname, 0); } @@ -1510,8 +1512,12 @@ int32_t doProcessMsgFromServer(void* param) { updateTargetEpSet(pSendInfo, pTscObj, pMsg, pEpSet); - SDataBuf buf = { - .msgType = pMsg->msgType, .len = pMsg->contLen, .pData = NULL, .handle = pMsg->info.handle, .pEpSet = pEpSet}; + SDataBuf buf = {.msgType = pMsg->msgType, + .len = pMsg->contLen, + .pData = NULL, + .handle = pMsg->info.handle, + .handleRefId = pMsg->info.refId, + .pEpSet = pEpSet}; if (pMsg->contLen > 0) { buf.pData = taosMemoryCalloc(1, pMsg->contLen); @@ -2538,13 +2544,13 @@ static void fetchCallback(void* pResult, void* param, int32_t code) { if (code != TSDB_CODE_SUCCESS) { pRequest->code = code; taosMemoryFreeClear(pResultInfo->pData); - pRequest->body.fetchFp(((SSyncQueryParam *)pRequest->body.interParam)->userParam, pRequest, 0); + pRequest->body.fetchFp(((SSyncQueryParam*)pRequest->body.interParam)->userParam, pRequest, 0); return; } if (pRequest->code != TSDB_CODE_SUCCESS) { taosMemoryFreeClear(pResultInfo->pData); - pRequest->body.fetchFp(((SSyncQueryParam *)pRequest->body.interParam)->userParam, pRequest, 0); + pRequest->body.fetchFp(((SSyncQueryParam*)pRequest->body.interParam)->userParam, pRequest, 0); return; } @@ -2565,12 +2571,12 @@ static void fetchCallback(void* pResult, void* param, int32_t code) { atomic_add_fetch_64((int64_t*)&pActivity->fetchBytes, pRequest->body.resInfo.payloadLen); } - pRequest->body.fetchFp(((SSyncQueryParam *)pRequest->body.interParam)->userParam, pRequest, pResultInfo->numOfRows); + pRequest->body.fetchFp(((SSyncQueryParam*)pRequest->body.interParam)->userParam, pRequest, pResultInfo->numOfRows); } void taosAsyncFetchImpl(SRequestObj* pRequest, __taos_async_fn_t fp, void* param) { pRequest->body.fetchFp = fp; - ((SSyncQueryParam *)pRequest->body.interParam)->userParam = param; + ((SSyncQueryParam*)pRequest->body.interParam)->userParam = param; SReqResultInfo* pResultInfo = &pRequest->body.resInfo; @@ -2611,7 +2617,7 @@ void taosAsyncFetchImpl(SRequestObj* pRequest, __taos_async_fn_t fp, void* param void doRequestCallback(SRequestObj* pRequest, int32_t code) { pRequest->inCallback = true; int64_t this = pRequest->self; - pRequest->body.queryFp(((SSyncQueryParam *)pRequest->body.interParam)->userParam, pRequest, code); + pRequest->body.queryFp(((SSyncQueryParam*)pRequest->body.interParam)->userParam, pRequest, code); SRequestObj* pReq = acquireRequest(this); if (pReq != NULL) { pReq->inCallback = false; @@ -2619,11 +2625,11 @@ void doRequestCallback(SRequestObj* pRequest, int32_t code) { } } -int32_t clientParseSql(void* param, const char* dbName, const char* sql, bool parseOnly, const char* effectiveUser, SParseSqlRes* pRes) { +int32_t clientParseSql(void* param, const char* dbName, const char* sql, bool parseOnly, const char* effectiveUser, + SParseSqlRes* pRes) { #ifndef TD_ENTERPRISE return TSDB_CODE_SUCCESS; #else return clientParseSqlImpl(param, dbName, sql, parseOnly, effectiveUser, pRes); #endif } - diff --git a/source/common/src/tdatablock.c b/source/common/src/tdatablock.c index 054cff560f..f0d1630480 100644 --- a/source/common/src/tdatablock.c +++ b/source/common/src/tdatablock.c @@ -2118,6 +2118,30 @@ _end: return TSDB_CODE_SUCCESS; } +void buildCtbNameAddGruopId(char* ctbName, uint64_t groupId){ + char tmp[TSDB_TABLE_NAME_LEN] = {0}; + snprintf(tmp, TSDB_TABLE_NAME_LEN, "_%"PRIu64, groupId); + ctbName[TSDB_TABLE_NAME_LEN - strlen(tmp) - 1] = 0; // put groupId to the end + strcat(ctbName, tmp); +} + +bool isAutoTableName(char* ctbName){ + return (strlen(ctbName) == 34 && ctbName[0] == 't' && ctbName[1] == '_'); +} + +bool alreadyAddGroupId(char* ctbName){ + size_t len = strlen(ctbName); + size_t _location = len - 1; + while(_location > 0){ + if(ctbName[_location] < '0' || ctbName[_location] > '9'){ + break; + } + _location--; + } + + return ctbName[_location] == '_' && len - 1 - _location > 15; //15 means the min length of groupid +} + char* buildCtbNameByGroupId(const char* stbFullName, uint64_t groupId) { char* pBuf = taosMemoryCalloc(1, TSDB_TABLE_NAME_LEN + 1); if (!pBuf) { diff --git a/source/dnode/vnode/src/inc/tq.h b/source/dnode/vnode/src/inc/tq.h index 0ef29fcb3a..cded4ddd7c 100644 --- a/source/dnode/vnode/src/inc/tq.h +++ b/source/dnode/vnode/src/inc/tq.h @@ -145,7 +145,7 @@ int32_t tqOffsetCommitFile(STqOffsetStore* pStore); // tqSink int32_t tqBuildDeleteReq(STQ* pTq, const char* stbFullName, const SSDataBlock* pDataBlock, SBatchDeleteReq* deleteReq, - const char* pIdStr); + const char* pIdStr, bool newSubTableRule); void tqSinkDataIntoDstTable(SStreamTask* pTask, void* vnode, void* data); // tqOffset @@ -164,7 +164,7 @@ int32_t setDstTableDataPayload(uint64_t suid, const STSchema* pTSchema, int32_t int32_t doMergeExistedRows(SSubmitTbData* pExisted, const SSubmitTbData* pNew, const char* id); SVCreateTbReq* buildAutoCreateTableReq(const char* stbFullName, int64_t suid, int32_t numOfCols, - SSDataBlock* pDataBlock, SArray* pTagArray); + SSDataBlock* pDataBlock, SArray* pTagArray, bool newSubTableRule); #ifdef __cplusplus } diff --git a/source/dnode/vnode/src/sma/smaRollup.c b/source/dnode/vnode/src/sma/smaRollup.c index fd4cc7eaa5..138bcbb133 100644 --- a/source/dnode/vnode/src/sma/smaRollup.c +++ b/source/dnode/vnode/src/sma/smaRollup.c @@ -703,7 +703,7 @@ static int32_t tdRSmaExecAndSubmitResult(SSma *pSma, qTaskInfo_t taskInfo, SRSma code = terrno; TSDB_CHECK_CODE(code, lino, _exit); } - code = tqBuildDeleteReq(pSma->pVnode->pTq, NULL, output, &deleteReq, ""); + code = tqBuildDeleteReq(pSma->pVnode->pTq, NULL, output, &deleteReq, "", true); TSDB_CHECK_CODE(code, lino, _exit); code = tdRSmaProcessDelReq(pSma, suid, pItem->level, &deleteReq); TSDB_CHECK_CODE(code, lino, _exit); diff --git a/source/dnode/vnode/src/sma/smaTimeRange.c b/source/dnode/vnode/src/sma/smaTimeRange.c index 289986e01f..f537ede8c1 100644 --- a/source/dnode/vnode/src/sma/smaTimeRange.c +++ b/source/dnode/vnode/src/sma/smaTimeRange.c @@ -188,7 +188,7 @@ int32_t smaBlockToSubmit(SVnode *pVnode, const SArray *pBlocks, const STSchema * if (pDataBlock->info.type == STREAM_DELETE_RESULT) { pDeleteReq->suid = suid; pDeleteReq->deleteReqs = taosArrayInit(0, sizeof(SSingleDeleteReq)); - code = tqBuildDeleteReq(pVnode->pTq, stbFullName, pDataBlock, pDeleteReq, ""); + code = tqBuildDeleteReq(pVnode->pTq, stbFullName, pDataBlock, pDeleteReq, "", true); TSDB_CHECK_CODE(code, lino, _exit); continue; } @@ -196,7 +196,7 @@ int32_t smaBlockToSubmit(SVnode *pVnode, const SArray *pBlocks, const STSchema * SSubmitTbData tbData = {.suid = suid, .uid = 0, .sver = pTSchema->version, .flags = SUBMIT_REQ_AUTO_CREATE_TABLE,}; int32_t cid = taosArrayGetSize(pDataBlock->pDataBlock) + 1; - tbData.pCreateTbReq = buildAutoCreateTableReq(stbFullName, suid, cid, pDataBlock, tagArray); + tbData.pCreateTbReq = buildAutoCreateTableReq(stbFullName, suid, cid, pDataBlock, tagArray, true); { uint64_t groupId = pDataBlock->info.id.groupId; diff --git a/source/dnode/vnode/src/tq/tqRead.c b/source/dnode/vnode/src/tq/tqRead.c index bfa8cfdb53..0b05573aae 100644 --- a/source/dnode/vnode/src/tq/tqRead.c +++ b/source/dnode/vnode/src/tq/tqRead.c @@ -596,10 +596,12 @@ static int32_t doSetVal(SColumnInfoData* pColumnInfoData, int32_t rowIndex, SCol if (IS_STR_DATA_TYPE(pColVal->type)) { char val[65535 + 2] = {0}; - if (pColVal->value.pData != NULL) { - memcpy(varDataVal(val), pColVal->value.pData, pColVal->value.nData); + if(COL_VAL_IS_VALUE(pColVal)){ + if (pColVal->value.pData != NULL) { + memcpy(varDataVal(val), pColVal->value.pData, pColVal->value.nData); + } varDataSetLen(val, pColVal->value.nData); - code = colDataSetVal(pColumnInfoData, rowIndex, val, !COL_VAL_IS_VALUE(pColVal)); + code = colDataSetVal(pColumnInfoData, rowIndex, val, false); } else { colDataSetNULL(pColumnInfoData, rowIndex); } @@ -869,22 +871,8 @@ int32_t tqRetrieveTaosxBlock(STqReader* pReader, SArray* blocks, SArray* schemas sourceIdx++; } else if (pCol->cid == pColData->info.colId) { tColDataGetValue(pCol, i, &colVal); - - if (IS_STR_DATA_TYPE(colVal.type)) { - if (colVal.value.pData != NULL) { - char val[65535 + 2]; - memcpy(varDataVal(val), colVal.value.pData, colVal.value.nData); - varDataSetLen(val, colVal.value.nData); - if (colDataSetVal(pColData, curRow - lastRow, val, !COL_VAL_IS_VALUE(&colVal)) < 0) { - goto FAIL; - } - } else { - colDataSetNULL(pColData, curRow - lastRow); - } - } else { - if (colDataSetVal(pColData, curRow - lastRow, (void*)&colVal.value.val, !COL_VAL_IS_VALUE(&colVal)) < 0) { - goto FAIL; - } + if(doSetVal(pColData, curRow - lastRow, &colVal) != TDB_CODE_SUCCESS){ + goto FAIL; } sourceIdx++; targetIdx++; @@ -967,23 +955,10 @@ int32_t tqRetrieveTaosxBlock(STqReader* pReader, SArray* blocks, SArray* schemas if (colVal.cid < pColData->info.colId) { sourceIdx++; } else if (colVal.cid == pColData->info.colId) { - if (IS_STR_DATA_TYPE(colVal.type)) { - if (colVal.value.pData != NULL) { - char val[65535 + 2]; - memcpy(varDataVal(val), colVal.value.pData, colVal.value.nData); - varDataSetLen(val, colVal.value.nData); - if (colDataSetVal(pColData, curRow - lastRow, val, !COL_VAL_IS_VALUE(&colVal)) < 0) { - goto FAIL; - } - } else { - colDataSetNULL(pColData, curRow - lastRow); - } - } else { - if (colDataSetVal(pColData, curRow - lastRow, (void*)&colVal.value.val, !COL_VAL_IS_VALUE(&colVal)) < 0) { - goto FAIL; - } + if(doSetVal(pColData, curRow - lastRow, &colVal) != TDB_CODE_SUCCESS){ + goto FAIL; } - sourceIdx++; + sourceIdx++; targetIdx++; } } diff --git a/source/dnode/vnode/src/tq/tqSink.c b/source/dnode/vnode/src/tq/tqSink.c index c2e48d5d92..7fcb86d84a 100644 --- a/source/dnode/vnode/src/tq/tqSink.c +++ b/source/dnode/vnode/src/tq/tqSink.c @@ -41,10 +41,10 @@ static bool isValidDstChildTable(SMetaReader* pReader, int32_t vgId, const ch static int32_t initCreateTableMsg(SVCreateTbReq* pCreateTableReq, uint64_t suid, const char* stbFullName, int32_t numOfTags); static SArray* createDefaultTagColName(); static void setCreateTableMsgTableName(SVCreateTbReq* pCreateTableReq, SSDataBlock* pDataBlock, const char* stbFullName, - int64_t gid); + int64_t gid, bool newSubTableRule); int32_t tqBuildDeleteReq(STQ* pTq, const char* stbFullName, const SSDataBlock* pDataBlock, SBatchDeleteReq* deleteReq, - const char* pIdStr) { + const char* pIdStr, bool newSubTableRule) { int32_t totalRows = pDataBlock->info.rows; SColumnInfoData* pStartTsCol = taosArrayGet(pDataBlock->pDataBlock, START_TS_COLUMN_INDEX); SColumnInfoData* pEndTsCol = taosArrayGet(pDataBlock->pDataBlock, END_TS_COLUMN_INDEX); @@ -68,6 +68,12 @@ int32_t tqBuildDeleteReq(STQ* pTq, const char* stbFullName, const SSDataBlock* p if (varTbName != NULL && varTbName != (void*)-1) { name = taosMemoryCalloc(1, TSDB_TABLE_NAME_LEN); memcpy(name, varDataVal(varTbName), varDataLen(varTbName)); + if(newSubTableRule && + !isAutoTableName(name) && + !alreadyAddGroupId(name) && + groupId != 0) { + buildCtbNameAddGruopId(name, groupId); + } } else if (stbFullName) { name = buildCtbNameByGroupId(stbFullName, groupId); } else { @@ -173,9 +179,18 @@ SArray* createDefaultTagColName() { } void setCreateTableMsgTableName(SVCreateTbReq* pCreateTableReq, SSDataBlock* pDataBlock, const char* stbFullName, - int64_t gid) { + int64_t gid, bool newSubTableRule) { if (pDataBlock->info.parTbName[0]) { - pCreateTableReq->name = taosStrdup(pDataBlock->info.parTbName); + if(newSubTableRule && + !isAutoTableName(pDataBlock->info.parTbName) && + !alreadyAddGroupId(pDataBlock->info.parTbName) && + gid != 0) { + pCreateTableReq->name = taosMemoryCalloc(1, TSDB_TABLE_NAME_LEN); + strcpy(pCreateTableReq->name, pDataBlock->info.parTbName); + buildCtbNameAddGruopId(pCreateTableReq->name, gid); + }else{ + pCreateTableReq->name = taosStrdup(pDataBlock->info.parTbName); + } } else { pCreateTableReq->name = buildCtbNameByGroupId(stbFullName, gid); } @@ -247,7 +262,7 @@ static int32_t doBuildAndSendCreateTableMsg(SVnode* pVnode, char* stbFullName, S ASSERT(gid == *(int64_t*)pGpIdData); } - setCreateTableMsgTableName(pCreateTbReq, pDataBlock, stbFullName, gid); + setCreateTableMsgTableName(pCreateTbReq, pDataBlock, stbFullName, gid, pTask->ver >= SSTREAM_TASK_SUBTABLE_CHANGED_VER); taosArrayPush(reqs.pArray, pCreateTbReq); tqDebug("s-task:%s build create table:%s msg complete", pTask->id.idStr, pCreateTbReq->name); @@ -357,7 +372,8 @@ int32_t doBuildAndSendDeleteMsg(SVnode* pVnode, char* stbFullName, SSDataBlock* int64_t suid) { SBatchDeleteReq deleteReq = {.suid = suid, .deleteReqs = taosArrayInit(0, sizeof(SSingleDeleteReq))}; - int32_t code = tqBuildDeleteReq(pVnode->pTq, stbFullName, pDataBlock, &deleteReq, pTask->id.idStr); + int32_t code = tqBuildDeleteReq(pVnode->pTq, stbFullName, pDataBlock, &deleteReq, pTask->id.idStr, + pTask->ver >= SSTREAM_TASK_SUBTABLE_CHANGED_VER); if (code != TSDB_CODE_SUCCESS) { return code; } @@ -411,7 +427,7 @@ bool isValidDstChildTable(SMetaReader* pReader, int32_t vgId, const char* ctbNam } SVCreateTbReq* buildAutoCreateTableReq(const char* stbFullName, int64_t suid, int32_t numOfCols, - SSDataBlock* pDataBlock, SArray* pTagArray) { + SSDataBlock* pDataBlock, SArray* pTagArray, bool newSubTableRule) { SVCreateTbReq* pCreateTbReq = taosMemoryCalloc(1, sizeof(SVCreateTbReq)); if (pCreateTbReq == NULL) { terrno = TSDB_CODE_OUT_OF_MEMORY; @@ -436,7 +452,7 @@ SVCreateTbReq* buildAutoCreateTableReq(const char* stbFullName, int64_t suid, in pCreateTbReq->ctb.tagName = createDefaultTagColName(); // set table name - setCreateTableMsgTableName(pCreateTbReq, pDataBlock, stbFullName, pDataBlock->info.id.groupId); + setCreateTableMsgTableName(pCreateTbReq, pDataBlock, stbFullName, pDataBlock->info.id.groupId, newSubTableRule); return pCreateTbReq; } @@ -637,6 +653,7 @@ int32_t setDstTableDataUid(SVnode* pVnode, SStreamTask* pTask, SSDataBlock* pDat tqDebug("s-task:%s vgId:%d, gropuId:%" PRIu64 " datablock table name is null, set name:%s", id, vgId, groupId, dstTableName); } else { + tstrncpy(dstTableName, pTableSinkInfo->name.data, pTableSinkInfo->name.len + 1); if (pTableSinkInfo->uid != 0) { tqDebug("s-task:%s write %d rows into groupId:%" PRIu64 " dstTable:%s(uid:%" PRIu64 ")", id, numOfRows, groupId, dstTableName, pTableSinkInfo->uid); @@ -649,10 +666,17 @@ int32_t setDstTableDataUid(SVnode* pVnode, SStreamTask* pTask, SSDataBlock* pDat if (dstTableName[0] == 0) { memset(dstTableName, 0, TSDB_TABLE_NAME_LEN); buildCtbNameByGroupIdImpl(stbFullName, groupId, dstTableName); + }else{ + if(pTask->ver >= SSTREAM_TASK_SUBTABLE_CHANGED_VER && + !isAutoTableName(dstTableName) && + !alreadyAddGroupId(dstTableName) && + groupId != 0) { + buildCtbNameAddGruopId(dstTableName, groupId); + } } int32_t nameLen = strlen(dstTableName); - pTableSinkInfo = taosMemoryCalloc(1, sizeof(STableSinkInfo) + nameLen); + pTableSinkInfo = taosMemoryCalloc(1, sizeof(STableSinkInfo) + nameLen + 1); if (pTableSinkInfo == NULL) { return TSDB_CODE_OUT_OF_MEMORY; } @@ -690,7 +714,7 @@ int32_t setDstTableDataUid(SVnode* pVnode, SStreamTask* pTask, SSDataBlock* pDat pTableData->flags = SUBMIT_REQ_AUTO_CREATE_TABLE; pTableData->pCreateTbReq = - buildAutoCreateTableReq(stbFullName, suid, pTSchema->numOfCols + 1, pDataBlock, pTagArray); + buildAutoCreateTableReq(stbFullName, suid, pTSchema->numOfCols + 1, pDataBlock, pTagArray, pTask->ver >= SSTREAM_TASK_SUBTABLE_CHANGED_VER); taosArrayDestroy(pTagArray); if (pTableData->pCreateTbReq == NULL) { @@ -719,7 +743,7 @@ int32_t setDstTableDataUid(SVnode* pVnode, SStreamTask* pTask, SSDataBlock* pDat } } - return TSDB_CODE_SUCCESS; + return TDB_CODE_SUCCESS; } int32_t setDstTableDataPayload(uint64_t suid, const STSchema *pTSchema, int32_t blockIndex, SSDataBlock* pDataBlock, diff --git a/source/dnode/vnode/src/tsdb/tsdbRead2.c b/source/dnode/vnode/src/tsdb/tsdbRead2.c index 8a7ec981e8..e958600093 100644 --- a/source/dnode/vnode/src/tsdb/tsdbRead2.c +++ b/source/dnode/vnode/src/tsdb/tsdbRead2.c @@ -4100,9 +4100,9 @@ void tsdbReaderClose2(STsdbReader* pReader) { size_t numOfTables = tSimpleHashGetSize(pReader->status.pTableMap); if (pReader->status.pTableMap != NULL) { destroyAllBlockScanInfo(pReader->status.pTableMap); - clearBlockScanInfoBuf(&pReader->blockInfoBuf); pReader->status.pTableMap = NULL; } + clearBlockScanInfoBuf(&pReader->blockInfoBuf); if (pReader->pFileReader != NULL) { tsdbDataFileReaderClose(&pReader->pFileReader); diff --git a/source/libs/parser/src/parTranslater.c b/source/libs/parser/src/parTranslater.c index 4b051378bd..4d7df17545 100644 --- a/source/libs/parser/src/parTranslater.c +++ b/source/libs/parser/src/parTranslater.c @@ -2743,6 +2743,14 @@ static EDealRes doCheckAggColCoexist(SNode** pNode, void* pContext) { return DEAL_RES_CONTINUE; } +static int32_t checkIsEmptyResult(STranslateContext* pCxt, SSelectStmt* pSelect) { + if (pSelect->timeRange.skey > pSelect->timeRange.ekey + && !pSelect->hasCountFunc) { + pSelect->isEmptyResult = true; + } + return TSDB_CODE_SUCCESS; +} + static int32_t resetSelectFuncNumWithoutDup(SSelectStmt* pSelect) { if (pSelect->selectFuncNum <= 1) return TSDB_CODE_SUCCESS; pSelect->selectFuncNum = 0; @@ -4416,9 +4424,6 @@ static int32_t translateWhere(STranslateContext* pCxt, SSelectStmt* pSelect) { if (TSDB_CODE_SUCCESS == code) { code = getQueryTimeRange(pCxt, pSelect->pWhere, &pSelect->timeRange); } - if (TSDB_CODE_SUCCESS == code && pSelect->timeRange.skey > pSelect->timeRange.ekey) { - pSelect->isEmptyResult = true; - } if (pSelect->pWhere != NULL) { setTableVgroupsFromEqualTbnameCond(pCxt, pSelect); } @@ -4606,6 +4611,9 @@ static int32_t translateSelectFrom(STranslateContext* pCxt, SSelectStmt* pSelect if (TSDB_CODE_SUCCESS == code) { code = translateOrderBy(pCxt, pSelect); } + if (TSDB_CODE_SUCCESS == code) { + code = checkIsEmptyResult(pCxt, pSelect); + } if (TSDB_CODE_SUCCESS == code) { resetSelectFuncNumWithoutDup(pSelect); code = checkAggColCoexist(pCxt, pSelect); diff --git a/source/libs/scheduler/inc/schInt.h b/source/libs/scheduler/inc/schInt.h index 3b7a76bfc7..1c90e61ea3 100644 --- a/source/libs/scheduler/inc/schInt.h +++ b/source/libs/scheduler/inc/schInt.h @@ -52,10 +52,10 @@ typedef enum { SCH_ALL, } SCH_POLICY; -#define SCHEDULE_DEFAULT_MAX_JOB_NUM 1000 -#define SCHEDULE_DEFAULT_MAX_TASK_NUM 1000 -#define SCHEDULE_DEFAULT_POLICY SCH_LOAD_SEQ -#define SCHEDULE_DEFAULT_MAX_NODE_NUM 20 +#define SCHEDULE_DEFAULT_MAX_JOB_NUM 1000 +#define SCHEDULE_DEFAULT_MAX_TASK_NUM 1000 +#define SCHEDULE_DEFAULT_POLICY SCH_LOAD_SEQ +#define SCHEDULE_DEFAULT_MAX_NODE_NUM 20 #define SCH_DEFAULT_TASK_TIMEOUT_USEC 5000000 #define SCH_MAX_TASK_TIMEOUT_USEC 300000000 @@ -68,8 +68,9 @@ typedef struct SSchDebug { } SSchDebug; typedef struct SSchTrans { - void *pTrans; - void *pHandle; + void *pTrans; + void *pHandle; + int64_t pHandleId; } SSchTrans; typedef struct SSchHbTrans { @@ -299,7 +300,7 @@ typedef struct SSchJob { void *fetchRes; // TODO free it or not bool fetched; bool noMoreRetry; - int64_t resNumOfRows; // from int32_t to int64_t + int64_t resNumOfRows; // from int32_t to int64_t SSchResInfo userRes; char *sql; SQueryProfileSummary summary; @@ -334,14 +335,15 @@ extern SSchedulerMgmt schMgmt; (!SCH_IS_DATA_BIND_QRY_TASK(_task))) #define SCH_UPDATE_REDIRECT_CODE(job, _code) atomic_val_compare_exchange_32(&((job)->redirectCode), 0, _code) -#define SCH_GET_REDIRECT_CODE(job, _code) (((!NO_RET_REDIRECT_ERROR(_code)) || (job)->redirectCode == 0) ? (_code) : (job)->redirectCode) +#define SCH_GET_REDIRECT_CODE(job, _code) \ + (((!NO_RET_REDIRECT_ERROR(_code)) || (job)->redirectCode == 0) ? (_code) : (job)->redirectCode) #define SCH_SET_TASK_STATUS(task, st) atomic_store_8(&(task)->status, st) #define SCH_GET_TASK_STATUS(task) atomic_load_8(&(task)->status) #define SCH_GET_TASK_STATUS_STR(task) jobTaskStatusStr(SCH_GET_TASK_STATUS(task)) #define SCH_TASK_ALREADY_LAUNCHED(task) (SCH_GET_TASK_STATUS(task) >= JOB_TASK_STATUS_EXEC) -#define SCH_TASK_EXEC_DONE(task) (SCH_GET_TASK_STATUS(task) >= JOB_TASK_STATUS_PART_SUCC) +#define SCH_TASK_EXEC_DONE(task) (SCH_GET_TASK_STATUS(task) >= JOB_TASK_STATUS_PART_SUCC) #define SCH_GET_TASK_HANDLE(_task) ((_task) ? (_task)->handle : NULL) #define SCH_SET_TASK_HANDLE(_task, _handle) ((_task)->handle = (_handle)) @@ -362,8 +364,8 @@ extern SSchedulerMgmt schMgmt; #define SCH_JOB_NEED_FLOW_CTRL(_job) ((_job)->attr.needFlowCtrl) #define SCH_TASK_NEED_FLOW_CTRL(_job, _task) \ (SCH_IS_DATA_BIND_QRY_TASK(_task) && SCH_JOB_NEED_FLOW_CTRL(_job) && SCH_IS_LEVEL_UNFINISHED((_task)->level)) -#define SCH_FETCH_TYPE(_pSrcTask) (SCH_IS_DATA_BIND_QRY_TASK(_pSrcTask) ? TDMT_SCH_FETCH : TDMT_SCH_MERGE_FETCH) -#define SCH_TASK_NEED_FETCH(_task) ((_task)->plan->subplanType != SUBPLAN_TYPE_MODIFY) +#define SCH_FETCH_TYPE(_pSrcTask) (SCH_IS_DATA_BIND_QRY_TASK(_pSrcTask) ? TDMT_SCH_FETCH : TDMT_SCH_MERGE_FETCH) +#define SCH_TASK_NEED_FETCH(_task) ((_task)->plan->subplanType != SUBPLAN_TYPE_MODIFY) #define SCH_MULTI_LEVEL_LAUNCHED(_job) ((_job)->levelIdx != ((_job)->levelNum - 1)) #define SCH_SET_JOB_TYPE(_job, type) \ @@ -380,25 +382,26 @@ extern SSchedulerMgmt schMgmt; #define SCH_JOB_NEED_WAIT(_job) (!SCH_IS_QUERY_JOB(_job)) #define SCH_JOB_NEED_DROP(_job) (SCH_IS_QUERY_JOB(_job)) #define SCH_IS_EXPLAIN_JOB(_job) (EXPLAIN_MODE_ANALYZE == (_job)->attr.explainMode) -#define SCH_NETWORK_ERR(_code) ((_code) == TSDB_CODE_RPC_BROKEN_LINK || (_code) == TSDB_CODE_RPC_NETWORK_UNAVAIL || (_code) == TSDB_CODE_RPC_SOMENODE_NOT_CONNECTED) +#define SCH_NETWORK_ERR(_code) \ + ((_code) == TSDB_CODE_RPC_BROKEN_LINK || (_code) == TSDB_CODE_RPC_NETWORK_UNAVAIL || \ + (_code) == TSDB_CODE_RPC_SOMENODE_NOT_CONNECTED) #define SCH_REDIRECT_MSGTYPE(_msgType) \ ((_msgType) == TDMT_SCH_LINK_BROKEN || (_msgType) == TDMT_SCH_QUERY || (_msgType) == TDMT_SCH_MERGE_QUERY || \ (_msgType) == TDMT_SCH_FETCH || (_msgType) == TDMT_SCH_MERGE_FETCH) #define SCH_LOW_LEVEL_NETWORK_ERR(_job, _task, _code) \ - (SCH_NETWORK_ERR(_code) && ((_task)->level->level == (_job)->levelIdx)) + (SCH_NETWORK_ERR(_code) && ((_task)->level->level == (_job)->levelIdx)) #define SCH_TOP_LEVEL_NETWORK_ERR(_job, _task, _code) \ - (SCH_NETWORK_ERR(_code) && ((_task)->level->level > (_job)->levelIdx)) -#define SCH_TASK_RETRY_NETWORK_ERR(_task, _code) \ - (SCH_NETWORK_ERR(_code) && (_task)->redirectCtx.inRedirect) + (SCH_NETWORK_ERR(_code) && ((_task)->level->level > (_job)->levelIdx)) +#define SCH_TASK_RETRY_NETWORK_ERR(_task, _code) (SCH_NETWORK_ERR(_code) && (_task)->redirectCtx.inRedirect) -#define SCH_JOB_NEED_RETRY(_job, _task, _msgType, _code) \ - (SCH_REDIRECT_MSGTYPE(_msgType) && SCH_TOP_LEVEL_NETWORK_ERR(_job, _task, _code)) -#define SCH_TASKSET_NEED_RETRY(_job, _task, _msgType, _code) \ - (SCH_REDIRECT_MSGTYPE(_msgType) && \ - (NEED_SCHEDULER_REDIRECT_ERROR(_code) || SCH_LOW_LEVEL_NETWORK_ERR((_job), (_task), (_code)) || SCH_TASK_RETRY_NETWORK_ERR((_task), (_code)))) +#define SCH_JOB_NEED_RETRY(_job, _task, _msgType, _code) \ + (SCH_REDIRECT_MSGTYPE(_msgType) && SCH_TOP_LEVEL_NETWORK_ERR(_job, _task, _code)) +#define SCH_TASKSET_NEED_RETRY(_job, _task, _msgType, _code) \ + (SCH_REDIRECT_MSGTYPE(_msgType) && \ + (NEED_SCHEDULER_REDIRECT_ERROR(_code) || SCH_LOW_LEVEL_NETWORK_ERR((_job), (_task), (_code)) || \ + SCH_TASK_RETRY_NETWORK_ERR((_task), (_code)))) #define SCH_TASK_NEED_RETRY(_msgType, _code) \ - ((SCH_REDIRECT_MSGTYPE(_msgType) && SCH_NETWORK_ERR(_code)) || (_code) == TSDB_CODE_SCH_TIMEOUT_ERROR) - + ((SCH_REDIRECT_MSGTYPE(_msgType) && SCH_NETWORK_ERR(_code)) || (_code) == TSDB_CODE_SCH_TIMEOUT_ERROR) #define SCH_IS_LEVEL_UNFINISHED(_level) ((_level)->taskLaunchedNum < (_level)->taskNum) #define SCH_GET_CUR_EP(_addr) (&(_addr)->epSet.eps[(_addr)->epSet.inUse]) @@ -488,50 +491,51 @@ extern SSchedulerMgmt schMgmt; #define TD_RWLATCH_WRITE_FLAG_COPY 0x40000000 -#define SCH_LOCK(type, _lock) \ - do { \ - if (SCH_READ == (type)) { \ - ASSERTS(atomic_load_32(_lock) >= 0, "invalid lock value before read lock"); \ - SCH_LOCK_DEBUG("SCH RLOCK%p:%d, %s:%d B", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \ - taosRLockLatch(_lock); \ - SCH_LOCK_DEBUG("SCH RLOCK%p:%d, %s:%d E", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \ - ASSERTS(atomic_load_32(_lock) > 0, "invalid lock value after read lock"); \ - } else { \ - ASSERTS(atomic_load_32(_lock) >= 0, "invalid lock value before write lock"); \ - SCH_LOCK_DEBUG("SCH WLOCK%p:%d, %s:%d B", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \ - taosWLockLatch(_lock); \ - SCH_LOCK_DEBUG("SCH WLOCK%p:%d, %s:%d E", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \ - ASSERTS(atomic_load_32(_lock) == TD_RWLATCH_WRITE_FLAG_COPY, "invalid lock value after write lock"); \ - } \ +#define SCH_LOCK(type, _lock) \ + do { \ + if (SCH_READ == (type)) { \ + ASSERTS(atomic_load_32(_lock) >= 0, "invalid lock value before read lock"); \ + SCH_LOCK_DEBUG("SCH RLOCK%p:%d, %s:%d B", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \ + taosRLockLatch(_lock); \ + SCH_LOCK_DEBUG("SCH RLOCK%p:%d, %s:%d E", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \ + ASSERTS(atomic_load_32(_lock) > 0, "invalid lock value after read lock"); \ + } else { \ + ASSERTS(atomic_load_32(_lock) >= 0, "invalid lock value before write lock"); \ + SCH_LOCK_DEBUG("SCH WLOCK%p:%d, %s:%d B", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \ + taosWLockLatch(_lock); \ + SCH_LOCK_DEBUG("SCH WLOCK%p:%d, %s:%d E", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \ + ASSERTS(atomic_load_32(_lock) == TD_RWLATCH_WRITE_FLAG_COPY, "invalid lock value after write lock"); \ + } \ } while (0) -#define SCH_UNLOCK(type, _lock) \ - do { \ - if (SCH_READ == (type)) { \ - ASSERTS(atomic_load_32((_lock)) > 0, "invalid lock value before read unlock"); \ - SCH_LOCK_DEBUG("SCH RULOCK%p:%d, %s:%d B", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \ - taosRUnLockLatch(_lock); \ - SCH_LOCK_DEBUG("SCH RULOCK%p:%d, %s:%d E", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \ - ASSERTS(atomic_load_32((_lock)) >= 0, "invalid lock value after read unlock"); \ - } else { \ - ASSERTS(atomic_load_32((_lock)) & TD_RWLATCH_WRITE_FLAG_COPY, "invalid lock value before write unlock"); \ - SCH_LOCK_DEBUG("SCH WULOCK%p:%d, %s:%d B", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \ - taosWUnLockLatch(_lock); \ - SCH_LOCK_DEBUG("SCH WULOCK%p:%d, %s:%d E", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \ - ASSERTS(atomic_load_32((_lock)) >= 0, "invalid lock value after write unlock"); \ - } \ +#define SCH_UNLOCK(type, _lock) \ + do { \ + if (SCH_READ == (type)) { \ + ASSERTS(atomic_load_32((_lock)) > 0, "invalid lock value before read unlock"); \ + SCH_LOCK_DEBUG("SCH RULOCK%p:%d, %s:%d B", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \ + taosRUnLockLatch(_lock); \ + SCH_LOCK_DEBUG("SCH RULOCK%p:%d, %s:%d E", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \ + ASSERTS(atomic_load_32((_lock)) >= 0, "invalid lock value after read unlock"); \ + } else { \ + ASSERTS(atomic_load_32((_lock)) & TD_RWLATCH_WRITE_FLAG_COPY, "invalid lock value before write unlock"); \ + SCH_LOCK_DEBUG("SCH WULOCK%p:%d, %s:%d B", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \ + taosWUnLockLatch(_lock); \ + SCH_LOCK_DEBUG("SCH WULOCK%p:%d, %s:%d E", (_lock), atomic_load_32(_lock), __FILE__, __LINE__); \ + ASSERTS(atomic_load_32((_lock)) >= 0, "invalid lock value after write unlock"); \ + } \ } while (0) -#define SCH_RESET_JOB_LEVEL_IDX(_job) do { \ - (_job)->levelIdx = (_job)->levelNum - 1; \ - SCH_JOB_DLOG("set job levelIdx to %d", (_job)->levelIdx); \ -} while (0) +#define SCH_RESET_JOB_LEVEL_IDX(_job) \ + do { \ + (_job)->levelIdx = (_job)->levelNum - 1; \ + SCH_JOB_DLOG("set job levelIdx to %d", (_job)->levelIdx); \ + } while (0) void schDeregisterTaskHb(SSchJob *pJob, SSchTask *pTask); void schCleanClusterHb(void *pTrans); int32_t schLaunchTask(SSchJob *job, SSchTask *task); int32_t schDelayLaunchTask(SSchJob *pJob, SSchTask *pTask); -int32_t schBuildAndSendMsg(SSchJob *job, SSchTask *task, SQueryNodeAddr *addr, int32_t msgType, void* param); +int32_t schBuildAndSendMsg(SSchJob *job, SSchTask *task, SQueryNodeAddr *addr, int32_t msgType, void *param); SSchJob *schAcquireJob(int64_t refId); int32_t schReleaseJob(int64_t refId); void schFreeFlowCtrl(SSchJob *pJob); diff --git a/source/libs/scheduler/src/schRemote.c b/source/libs/scheduler/src/schRemote.c index f987420ece..5c67c7974f 100644 --- a/source/libs/scheduler/src/schRemote.c +++ b/source/libs/scheduler/src/schRemote.c @@ -505,7 +505,7 @@ int32_t schHandleLinkBrokenCallback(void *param, SDataBuf *pMsg, int32_t code) { taosMemoryFree(pMsg->pEpSet); SSchHbCallbackParam *hbParam = (SSchHbCallbackParam *)param; - SSchTrans trans = {.pTrans = hbParam->pTrans, .pHandle = NULL}; + SSchTrans trans = {.pTrans = hbParam->pTrans, .pHandle = NULL, .pHandleId = 0}; SCH_ERR_RET(schUpdateHbConnection(&hbParam->nodeEpId, &trans)); SCH_ERR_RET(schBuildAndSendHbMsg(&hbParam->nodeEpId, NULL)); @@ -537,6 +537,7 @@ int32_t schHandleHbCallback(void *param, SDataBuf *pMsg, int32_t code) { SSchTrans trans = {0}; trans.pTrans = pParam->pTrans; trans.pHandle = pMsg->handle; + trans.pHandleId = pMsg->handleRefId; SCH_ERR_JRET(schUpdateHbConnection(&rsp.epId, &trans)); SCH_ERR_JRET(schProcessOnTaskStatusRsp(&rsp.epId, rsp.taskStatus)); diff --git a/source/libs/scheduler/src/schUtil.c b/source/libs/scheduler/src/schUtil.c index df9cde6b7a..39c54ea731 100644 --- a/source/libs/scheduler/src/schUtil.c +++ b/source/libs/scheduler/src/schUtil.c @@ -42,7 +42,7 @@ char *schDumpEpSet(SEpSet *pEpSet) { } int32_t maxSize = 1024; - char *str = taosMemoryMalloc(maxSize); + char *str = taosMemoryMalloc(maxSize); if (NULL == str) { return NULL; } @@ -73,7 +73,7 @@ char *schGetOpStr(SCH_OP_TYPE type) { } void schFreeHbTrans(SSchHbTrans *pTrans) { - rpcReleaseHandle(pTrans->trans.pHandle, TAOS_CONN_CLIENT); + rpcReleaseHandle((void *)pTrans->trans.pHandleId, TAOS_CONN_CLIENT); schFreeRpcCtx(&pTrans->rpcCtx); } @@ -209,7 +209,7 @@ int32_t schEnsureHbConnection(SSchJob *pJob, SSchTask *pTask) { if (!tsEnableQueryHb) { return TSDB_CODE_SUCCESS; } - + SQueryNodeAddr *addr = taosArrayGet(pTask->candidateAddrs, pTask->candidateIdx); SQueryNodeEpId epId = {0}; @@ -234,7 +234,8 @@ int32_t schUpdateHbConnection(SQueryNodeEpId *epId, SSchTrans *trans) { hb = taosHashGet(schMgmt.hbConnections, epId, sizeof(SQueryNodeEpId)); if (NULL == hb) { SCH_UNLOCK(SCH_READ, &schMgmt.hbLock); - qInfo("taosHashGet hb connection not exists, nodeId:%d, fqdn:%s, port:%d", epId->nodeId, epId->ep.fqdn, epId->ep.port); + qInfo("taosHashGet hb connection not exists, nodeId:%d, fqdn:%s, port:%d", epId->nodeId, epId->ep.fqdn, + epId->ep.port); SCH_ERR_RET(TSDB_CODE_APP_ERROR); } diff --git a/source/libs/stream/src/streamDispatch.c b/source/libs/stream/src/streamDispatch.c index e2c8704b1c..b51845d152 100644 --- a/source/libs/stream/src/streamDispatch.c +++ b/source/libs/stream/src/streamDispatch.c @@ -547,25 +547,21 @@ int32_t streamSearchAndAddBlock(SStreamTask* pTask, SStreamDispatchReq* pReqs, S memcpy(pDataBlock->info.parTbName, pBln->parTbName, strlen(pBln->parTbName)); } } else { - char* ctbName = taosMemoryCalloc(1, TSDB_TABLE_FNAME_LEN); - if (ctbName == NULL) { - return -1; - } - + char ctbName[TSDB_TABLE_FNAME_LEN] = {0}; if (pDataBlock->info.parTbName[0]) { - snprintf(ctbName, TSDB_TABLE_NAME_LEN, "%s.%s", pTask->outputInfo.shuffleDispatcher.dbInfo.db, - pDataBlock->info.parTbName); + if(pTask->ver >= SSTREAM_TASK_SUBTABLE_CHANGED_VER && + !isAutoTableName(pDataBlock->info.parTbName) && + !alreadyAddGroupId(pDataBlock->info.parTbName) && + groupId != 0){ + buildCtbNameAddGruopId(pDataBlock->info.parTbName, groupId); + } } else { buildCtbNameByGroupIdImpl(pTask->outputInfo.shuffleDispatcher.stbFullName, groupId, pDataBlock->info.parTbName); - snprintf(ctbName, TSDB_TABLE_NAME_LEN, "%s.%s", pTask->outputInfo.shuffleDispatcher.dbInfo.db, - pDataBlock->info.parTbName); } - + snprintf(ctbName, TSDB_TABLE_NAME_LEN, "%s.%s", pTask->outputInfo.shuffleDispatcher.dbInfo.db, pDataBlock->info.parTbName); /*uint32_t hashValue = MurmurHash3_32(ctbName, strlen(ctbName));*/ SUseDbRsp* pDbInfo = &pTask->outputInfo.shuffleDispatcher.dbInfo; - hashValue = - taosGetTbHashVal(ctbName, strlen(ctbName), pDbInfo->hashMethod, pDbInfo->hashPrefix, pDbInfo->hashSuffix); - taosMemoryFree(ctbName); + hashValue = taosGetTbHashVal(ctbName, strlen(ctbName), pDbInfo->hashMethod, pDbInfo->hashPrefix, pDbInfo->hashSuffix); SBlockName bln = {0}; bln.hashValue = hashValue; memcpy(bln.parTbName, pDataBlock->info.parTbName, strlen(pDataBlock->info.parTbName)); diff --git a/source/libs/stream/src/streamMeta.c b/source/libs/stream/src/streamMeta.c index 14ad8d0557..6e35e39a0a 100644 --- a/source/libs/stream/src/streamMeta.c +++ b/source/libs/stream/src/streamMeta.c @@ -180,10 +180,8 @@ int32_t streamMetaCheckBackendCompatible(SStreamMeta* pMeta) { } if (info.msgVer <= SSTREAM_TASK_INCOMPATIBLE_VER) { ret = STREAM_STATA_NO_COMPATIBLE; - } else if (info.msgVer == SSTREAM_TASK_NEED_CONVERT_VER) { + } else if (info.msgVer >= SSTREAM_TASK_NEED_CONVERT_VER) { ret = STREAM_STATA_NEED_CONVERT; - } else if (info.msgVer == SSTREAM_TASK_VER) { - ret = STREAM_STATA_COMPATIBLE; } tDecoderClear(&decoder); break; diff --git a/source/libs/stream/src/streamTask.c b/source/libs/stream/src/streamTask.c index c07955eb9d..094068a06e 100644 --- a/source/libs/stream/src/streamTask.c +++ b/source/libs/stream/src/streamTask.c @@ -213,7 +213,7 @@ int32_t tDecodeStreamTask(SDecoder* pDecoder, SStreamTask* pTask) { if (tStartDecode(pDecoder) < 0) return -1; if (tDecodeI64(pDecoder, &pTask->ver) < 0) return -1; - if (pTask->ver != SSTREAM_TASK_VER) return -1; + if (pTask->ver <= SSTREAM_TASK_INCOMPATIBLE_VER) return -1; if (tDecodeI64(pDecoder, &pTask->id.streamId) < 0) return -1; if (tDecodeI32(pDecoder, &pTask->id.taskId) < 0) return -1; @@ -300,7 +300,7 @@ int32_t tDecodeStreamTaskChkInfo(SDecoder* pDecoder, SCheckpointInfo* pChkpInfo) if (tStartDecode(pDecoder) < 0) return -1; if (tDecodeI64(pDecoder, &pChkpInfo->msgVer) < 0) return -1; - // if (ver != SSTREAM_TASK_VER) return -1; + // if (ver <= SSTREAM_TASK_INCOMPATIBLE_VER) return -1; if (tDecodeI64(pDecoder, &skip64) < 0) return -1; if (tDecodeI32(pDecoder, &skip32) < 0) return -1; @@ -328,7 +328,7 @@ int32_t tDecodeStreamTaskId(SDecoder* pDecoder, STaskId* pTaskId) { int64_t ver; if (tStartDecode(pDecoder) < 0) return -1; if (tDecodeI64(pDecoder, &ver) < 0) return -1; - if (ver != SSTREAM_TASK_VER) return -1; + if (ver <= SSTREAM_TASK_INCOMPATIBLE_VER) return -1; if (tDecodeI64(pDecoder, &pTaskId->streamId) < 0) return -1; diff --git a/source/libs/transport/src/transCli.c b/source/libs/transport/src/transCli.c index 246605694b..15b35030d3 100644 --- a/source/libs/transport/src/transCli.c +++ b/source/libs/transport/src/transCli.c @@ -414,6 +414,7 @@ void cliHandleResp(SCliConn* conn) { // buf's mem alread translated to transMsg.pCont if (!CONN_NO_PERSIST_BY_APP(conn)) { transMsg.info.handle = (void*)conn->refId; + transMsg.info.refId = (int64_t)(void*)conn->refId; tDebug("%s conn %p ref by app", CONN_GET_INST_LABEL(conn), conn); } diff --git a/tests/army/community/query/query_basic.json b/tests/army/community/query/query_basic.json index 4190fbdd72..c6a3482cd4 100644 --- a/tests/army/community/query/query_basic.json +++ b/tests/army/community/query/query_basic.json @@ -7,7 +7,7 @@ "password": "taosdata", "connection_pool_size": 8, "num_of_records_per_req": 4000, - "prepared_rand": 1000, + "prepared_rand": 10000, "thread_count": 3, "create_table_thread_count": 1, "confirm_parameter_prompt": "no", diff --git a/tests/army/community/query/query_basic.py b/tests/army/community/query/query_basic.py index 07bf6093c0..e3fdeb0f34 100644 --- a/tests/army/community/query/query_basic.py +++ b/tests/army/community/query/query_basic.py @@ -41,6 +41,7 @@ class TDTestCase(TBase): etool.benchMark(json=jfile) tdSql.execute(f"use {self.db}") + tdSql.execute("select database();") # set insert data information self.childtable_count = 6 self.insert_rows = 100000 @@ -50,10 +51,29 @@ class TDTestCase(TBase): def doQuery(self): tdLog.info(f"do query.") - # top bottom - sql = f"select top(uti, 5) from {self.stb} " + # __group_key + sql = f"select count(*),_group_key(uti),uti from {self.stb} partition by uti;" tdSql.execute(sql) + tdSql.checkRows(251) + sql = f"select count(*),_group_key(usi) from {self.stb} group by usi;" + tdSql.execute(sql) + tdSql.checkRows(997) + + # tail + sql1 = "select ts,ui from d0 order by ts desc limit 5 offset 2;" + sql2 = "select ts,tail(ui,5,2) from d0;" + self.checkSameResult(sql1, sql2) + + # uninqe + sql1 = "select distinct uti from d0 order by uti;" + sql2 = "select UNIQUE(uti) from d0 order by uti asc;" + self.checkSameResult(sql1, sql2) + + # top + sql1 = "select top(bi,10) from stb;" + sql2 = "select bi from stb where bi is not null order by bi desc limit 10;" + self.checkSameResult(sql1, sql2) # run def run(self): diff --git a/tests/army/frame/caseBase.py b/tests/army/frame/caseBase.py index 82295d8fdb..fcfcb7d066 100644 --- a/tests/army/frame/caseBase.py +++ b/tests/army/frame/caseBase.py @@ -16,6 +16,7 @@ import os import time import datetime import random +import copy from frame.log import * from frame.sql import * @@ -183,6 +184,46 @@ class TBase: sql = f"select {col} from {self.stb} order by _c0 asc limit 1" tdSql.checkFirstValue(sql, expect) + + # check sql1 is same result with sql2 + def checkSameResult(self, sql1, sql2): + tdLog.info(f"sql1={sql1}") + tdLog.info(f"sql2={sql2}") + tdLog.info("compare sql1 same with sql2 ...") + + # sql + rows1 = tdSql.query(sql1,queryTimes=2) + res1 = copy.deepcopy(tdSql.queryResult) + + tdSql.query(sql2,queryTimes=2) + res2 = tdSql.queryResult + + rowlen1 = len(res1) + rowlen2 = len(res2) + errCnt = 0 + + if rowlen1 != rowlen2: + tdLog.exit(f"both row count not equal. rowlen1={rowlen1} rowlen2={rowlen2} ") + return False + + for i in range(rowlen1): + row1 = res1[i] + row2 = res2[i] + collen1 = len(row1) + collen2 = len(row2) + if collen1 != collen2: + tdLog.exit(f"both col count not equal. collen1={collen1} collen2={collen2}") + return False + for j in range(collen1): + if row1[j] != row2[j]: + tdLog.info(f"error both column value not equal. row={i} col={j} col1={row1[j]} col2={row2[j]} .") + errCnt += 1 + + if errCnt > 0: + tdLog.exit(f"sql2 column value different with sql1. different count ={errCnt} ") + + tdLog.info("sql1 same result with sql2.") + # # get db information # diff --git a/tests/parallel_test/cases.task b/tests/parallel_test/cases.task index a3940108cb..1c2f5ec23c 100644 --- a/tests/parallel_test/cases.task +++ b/tests/parallel_test/cases.task @@ -146,6 +146,10 @@ ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/like.py -Q 2 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/like.py -Q 3 ,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/like.py -Q 4 +,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/td-28068.py +,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/td-28068.py -Q 2 +,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/td-28068.py -Q 3 +,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/td-28068.py -Q 4 ,,y,system-test,./pytest.sh python3 ./test.py -f 3-enterprise/restore/restoreDnode.py -N 5 -M 3 -i False ,,y,system-test,./pytest.sh python3 ./test.py -f 3-enterprise/restore/restoreVnode.py -N 5 -M 3 -i False diff --git a/tests/pytest/util/common.py b/tests/pytest/util/common.py index 010f45a573..c4885747d1 100644 --- a/tests/pytest/util/common.py +++ b/tests/pytest/util/common.py @@ -1845,6 +1845,23 @@ class TDCom: if i == 1: self.record_history_ts = ts_value + def get_subtable(self, tbname_pre): + tdSql.query(f'show tables') + tbname_list = list(map(lambda x:x[0], tdSql.queryResult)) + for tbname in tbname_list: + if tbname_pre in tbname: + return tbname + + def get_subtable_wait(self, tbname_pre): + tbname = self.get_subtable(tbname_pre) + latency = 0 + while tbname is None: + tbname = self.get_subtable(tbname_pre) + if latency < self.stream_timeout: + latency += 1 + time.sleep(1) + return tbname + def is_json(msg): if isinstance(msg, str): try: diff --git a/tests/pytest/util/sql.py b/tests/pytest/util/sql.py index 000040d958..62af9a1af9 100644 --- a/tests/pytest/util/sql.py +++ b/tests/pytest/util/sql.py @@ -88,7 +88,7 @@ class TDSql: self.affectedRows = self.cursor.execute(sql) return self.affectedRows except Exception as e: - tdLog.notice("Try to execute sql again, query times: %d "%i) + tdLog.notice("Try to execute sql again, execute times: %d "%i) if i == queryTimes: caller = inspect.getframeinfo(inspect.stack()[1][0]) args = (caller.filename, caller.lineno, sql, repr(e)) diff --git a/tests/script/tsim/stream/udTableAndTag0.sim b/tests/script/tsim/stream/udTableAndTag0.sim index c81927abcb..e8b51a3846 100644 --- a/tests/script/tsim/stream/udTableAndTag0.sim +++ b/tests/script/tsim/stream/udTableAndTag0.sim @@ -48,16 +48,6 @@ if $rows != 2 then goto loop0 endi -if $data00 != aaa-t1 then - print =====data00=$data00 - goto loop0 -endi - -if $data10 != aaa-t2 then - print =====data10=$data10 - goto loop0 -endi - $loop_count = 0 loop1: @@ -264,17 +254,6 @@ if $rows != 2 then goto loop6 endi -if $data00 != tbn-t1 then - print =====data00=$data00 - goto loop6 -endi - -if $data10 != tbn-t2 then - print =====data10=$data10 - goto loop6 -endi - - print ===== step5 print ===== tag name + table name @@ -312,21 +291,6 @@ if $rows != 3 then goto loop7 endi -if $data00 != tbn-t1 then - print =====data00=$data00 - goto loop7 -endi - -if $data10 != tbn-t2 then - print =====data10=$data10 - goto loop7 -endi - -if $data20 != tbn-t3 then - print =====data20=$data20 - goto loop7 -endi - $loop_count = 0 loop8: diff --git a/tests/script/tsim/stream/udTableAndTag1.sim b/tests/script/tsim/stream/udTableAndTag1.sim index e9dfbaabcf..e6427aab10 100644 --- a/tests/script/tsim/stream/udTableAndTag1.sim +++ b/tests/script/tsim/stream/udTableAndTag1.sim @@ -47,16 +47,6 @@ if $rows != 2 then goto loop0 endi -if $data00 != aaa-1 then - print =====data00=$data00 - goto loop0 -endi - -if $data10 != aaa-2 then - print =====data10=$data10 - goto loop0 -endi - $loop_count = 0 loop1: @@ -264,16 +254,6 @@ if $rows != 2 then goto loop6 endi -if $data00 != tbn-1 then - print =====data00=$data00 - goto loop6 -endi - -if $data10 != tbn-2 then - print =====data10=$data10 - goto loop6 -endi - print ===== step5 print ===== tag name + table name @@ -311,21 +291,6 @@ if $rows != 3 then goto loop7 endi -if $data00 != tbn-t1 then - print =====data00=$data00 - goto loop7 -endi - -if $data10 != tbn-t2 then - print =====data10=$data10 - goto loop7 -endi - -if $data20 != tbn-t3 then - print =====data20=$data20 - goto loop7 -endi - $loop_count = 0 loop8: diff --git a/tests/script/tsim/stream/udTableAndTag2.sim b/tests/script/tsim/stream/udTableAndTag2.sim index 973c55b9ef..8977dc219b 100644 --- a/tests/script/tsim/stream/udTableAndTag2.sim +++ b/tests/script/tsim/stream/udTableAndTag2.sim @@ -415,11 +415,6 @@ if $data00 != aaa then goto loop9 endi -if $data10 != aaa-1 then - print =====data00=$data00 - goto loop9 -endi - $loop_count = 0 loop10: diff --git a/tests/system-test/2-query/csum.py b/tests/system-test/2-query/csum.py index 1e6b26ada0..b16f511491 100644 --- a/tests/system-test/2-query/csum.py +++ b/tests/system-test/2-query/csum.py @@ -162,6 +162,16 @@ class TDTestCase: case6 = {"col": "c9"} self.checkcsum(**case6) + # unsigned check + case61 = {"col": "c11"} + self.checkcsum(**case61) + case62 = {"col": "c12"} + self.checkcsum(**case62) + case63 = {"col": "c13"} + self.checkcsum(**case63) + case64 = {"col": "c14"} + self.checkcsum(**case64) + # case7~8: nested query case7 = {"table_expr": "(select ts,c1 from db.stb1 order by ts, tbname )"} self.checkcsum(**case7) @@ -317,14 +327,14 @@ class TDTestCase: f"insert into t{i} values (" f"{basetime + (j+1)*10 + i * msec_per_min}, {random.randint(-200, -1)}, {random.uniform(200, -1)}, {basetime + random.randint(-200, -1)}, " f"'binary_{j}', {random.uniform(-200, -1)}, {random.choice([0,1])}, {random.randint(-200,-1)}, " - f"{random.randint(-200, -1)}, {random.randint(-127, -1)}, 'nchar_{j}' )" + f"{random.randint(-200, -1)}, {random.randint(-127, -1)}, 'nchar_{j}', {j},{j},{j},{j} )" ) tdSql.execute( f"insert into t{i} values (" f"{basetime - (j+1) * 10 + i * msec_per_min}, {random.randint(1, 200)}, {random.uniform(1, 200)}, {basetime - random.randint(1, 200)}, " f"'binary_{j}_1', {random.uniform(1, 200)}, {random.choice([0, 1])}, {random.randint(1,200)}, " - f"{random.randint(1,200)}, {random.randint(1,127)}, 'nchar_{j}_1' )" + f"{random.randint(1,200)}, {random.randint(1,127)}, 'nchar_{j}_1', {j*2},{j*2},{j*2},{j*2} )" ) tdSql.execute( f"insert into tt{i} values ( {basetime-(j+1) * 10 + i * msec_per_min}, {random.randint(1, 200)} )" @@ -340,8 +350,8 @@ class TDTestCase: tdSql.execute( "create stable db.stb1 (\ ts timestamp, c1 int, c2 float, c3 timestamp, c4 binary(16), c5 double, c6 bool, \ - c7 bigint, c8 smallint, c9 tinyint, c10 nchar(16)\ - ) \ + c7 bigint, c8 smallint, c9 tinyint, c10 nchar(16),\ + c11 int unsigned, c12 smallint unsigned, c13 tinyint unsigned, c14 bigint unsigned) \ tags(st1 int)" ) tdSql.execute( @@ -373,10 +383,10 @@ class TDTestCase: tdLog.printNoPrefix("######## insert data in the range near the max(bigint/double):") self.csum_test_table(tbnum) - tdSql.execute(f"insert into db.t1(ts, c1,c2,c5,c7) values " - f"({nowtime - (per_table_rows + 1) * 10 + i * msec_per_min}, {2**31-1}, {3.4*10**38}, {1.7*10**308}, {2**63-1})") - tdSql.execute(f"insert into db.t1(ts, c1,c2,c5,c7) values " - f"({nowtime - (per_table_rows + 2) * 10 + i * msec_per_min}, {2**31-1}, {3.4*10**38}, {1.7*10**308}, {2**63-1})") + tdSql.execute(f"insert into db.t1(ts, c1,c2,c5,c7,c11) values " + f"({nowtime - (per_table_rows + 1) * 10 + i * msec_per_min}, {2**31-1}, {3.4*10**38}, {1.7*10**308}, {2**63-1}, 128)") + tdSql.execute(f"insert into db.t1(ts, c1,c2,c5,c7,c11) values " + f"({nowtime - (per_table_rows + 2) * 10 + i * msec_per_min}, {2**31-1}, {3.4*10**38}, {1.7*10**308}, {2**63-1}, 129)") self.csum_current_query() self.csum_error_query() @@ -460,7 +470,7 @@ class TDTestCase: tdSql.checkRows(40) # bug need fix - tdSql.query("select tbname , csum(c1) from db.stb1 partition by tbname") + tdSql.query("select tbname , csum(c1), csum(c12) from db.stb1 partition by tbname") tdSql.checkRows(40) tdSql.query("select tbname , csum(st1) from db.stb1 partition by tbname") tdSql.checkRows(70) @@ -468,7 +478,7 @@ class TDTestCase: tdSql.checkRows(7) # partition by tags - tdSql.query("select st1 , csum(c1) from db.stb1 partition by st1") + tdSql.query("select st1 , csum(c1), csum(c13) from db.stb1 partition by st1") tdSql.checkRows(40) tdSql.query("select csum(c1) from db.stb1 partition by st1") tdSql.checkRows(40) diff --git a/tests/system-test/2-query/nestedQueryInterval.py b/tests/system-test/2-query/nestedQueryInterval.py index f0cd561363..3f1abe666b 100644 --- a/tests/system-test/2-query/nestedQueryInterval.py +++ b/tests/system-test/2-query/nestedQueryInterval.py @@ -1310,7 +1310,7 @@ class TDTestCase: ts = ts + 20 tdSql.query(f"select tbname,count(*) from nested.stable_null_childtable group by tbname order by tbname;") tdSql.checkRows(1) - tdSql.checkData(0, 1, 52); + tdSql.checkData(0, 1, 52) #stables @@ -1347,16 +1347,16 @@ class TDTestCase: tdSql.query(f"select tbname,count(*) from nested.stable_1 group by tbname order by tbname;") tdSql.checkRows(6) - tdSql.checkData(0, 1, 162); - tdSql.checkData(1, 1, 200); + tdSql.checkData(0, 1, 162) + tdSql.checkData(1, 1, 200) tdSql.query(f"select tbname,count(*) from nested.stable_null_data group by tbname order by tbname;") tdSql.checkRows(1) - tdSql.checkData(0, 1, 62); + tdSql.checkData(0, 1, 62) tdSql.query(f"select tbname,count(*) from nested.stable_null_childtable group by tbname order by tbname;") tdSql.checkRows(1) - tdSql.checkData(0, 1, 62); + tdSql.checkData(0, 1, 62) #test special character @@ -1366,25 +1366,25 @@ class TDTestCase: tdSql.query(f"select tbname,count(*) from nested.stable_1 group by tbname order by tbname;") tdSql.checkRows(7) - tdSql.checkData(0, 1, 1); - tdSql.checkData(1, 1, 162); - tdSql.checkData(2, 1, 200); + tdSql.checkData(0, 1, 1) + tdSql.checkData(1, 1, 162) + tdSql.checkData(2, 1, 200) tdSql.query(f"select count(*) from nested.stable_1 where tbname ='!@!@$$^$' ;") - tdSql.checkData(0, 0, 1); + tdSql.checkData(0, 0, 1) tdSql.query(f"select tbname,count(*) from nested.stable_null_data group by tbname order by tbname;") tdSql.checkRows(2) - tdSql.checkData(0, 1, 1); - tdSql.checkData(1, 1, 62); + tdSql.checkData(0, 1, 1) + tdSql.checkData(1, 1, 62) tdSql.query(f"select count(*) from nested.stable_null_data where tbname ='%^$^&^&' ;") - tdSql.checkData(0, 0, 1); + tdSql.checkData(0, 0, 1) tdSql.query(f"select tbname,count(*) from nested.stable_null_childtable group by tbname order by tbname;") tdSql.checkRows(2) - tdSql.checkData(0, 1, 1); - tdSql.checkData(1, 1, 62); + tdSql.checkData(0, 1, 1) + tdSql.checkData(1, 1, 62) tdSql.query(f"select count(*) from nested.stable_null_childtable where tbname ='$^%$%^&' ;") - tdSql.checkData(0, 0, 1); + tdSql.checkData(0, 0, 1) #test csv sql1 = "select tbname,ts,q_int,q_binary from nested.stable_1 >>'%s/stable_1.csv';" %self.testcasePath @@ -1404,31 +1404,31 @@ class TDTestCase: tdSql.query(f"select tbname,count(*) from nested.stable_1 group by tbname order by tbname;") tdSql.checkRows(7) - tdSql.checkData(0, 1, 1); - tdSql.checkData(1, 1, 162); - tdSql.checkData(2, 1, 200); + tdSql.checkData(0, 1, 1) + tdSql.checkData(1, 1, 162) + tdSql.checkData(2, 1, 200) tdSql.query(f"select count(*) from nested.stable_1 where tbname ='!@!@$$^$' ;") - tdSql.checkData(0, 0, 1); + tdSql.checkData(0, 0, 1) tdSql.query(f"select count(q_bool) from nested.stable_1;") - tdSql.checkData(0, 0, 0); + tdSql.checkData(0, 0, 0) tdSql.query(f"select tbname,count(*) from nested.stable_null_data group by tbname order by tbname;") tdSql.checkRows(2) - tdSql.checkData(0, 1, 1); - tdSql.checkData(1, 1, 62); + tdSql.checkData(0, 1, 1) + tdSql.checkData(1, 1, 62) tdSql.query(f"select count(*) from nested.stable_null_data where tbname ='%^$^&^&' ;") - tdSql.checkData(0, 0, 1); + tdSql.checkData(0, 0, 1) tdSql.query(f"select count(q_bool) from nested.stable_null_data;") - tdSql.checkData(0, 0, 0); + tdSql.checkData(0, 0, 0) tdSql.query(f"select tbname,count(*) from nested.stable_null_childtable group by tbname order by tbname;") tdSql.checkRows(2) - tdSql.checkData(0, 1, 1); - tdSql.checkData(1, 1, 62); + tdSql.checkData(0, 1, 1) + tdSql.checkData(1, 1, 62) tdSql.query(f"select count(*) from nested.stable_null_childtable where tbname ='$^%$%^&' ;") - tdSql.checkData(0, 0, 1); + tdSql.checkData(0, 0, 1) tdSql.query(f"select count(q_bool) from nested.stable_null_childtable;") - tdSql.checkData(0, 0, 0); + tdSql.checkData(0, 0, 0) tdSql.query(f"delete from nested.stable_1;") @@ -1440,31 +1440,40 @@ class TDTestCase: tdSql.query(f"select tbname,count(*) from nested.stable_1 group by tbname order by tbname;") tdSql.checkRows(7) - tdSql.checkData(0, 1, 1); - tdSql.checkData(1, 1, 162); - tdSql.checkData(2, 1, 200); + tdSql.checkData(0, 1, 1) + tdSql.checkData(1, 1, 162) + tdSql.checkData(2, 1, 200) tdSql.query(f"select count(*) from nested.stable_1 where tbname ='!@!@$$^$' ;") - tdSql.checkData(0, 0, 1); + tdSql.checkData(0, 0, 1) tdSql.query(f"select count(q_bool) from nested.stable_1;") - tdSql.checkData(0, 0, 0); + tdSql.checkData(0, 0, 0) tdSql.query(f"select tbname,count(*) from nested.stable_null_data group by tbname order by tbname;") tdSql.checkRows(2) - tdSql.checkData(0, 1, 1); - tdSql.checkData(1, 1, 62); + tdSql.checkData(0, 1, 1) + tdSql.checkData(1, 1, 62) tdSql.query(f"select count(*) from nested.stable_null_data where tbname ='%^$^&^&' ;") - tdSql.checkData(0, 0, 1); + tdSql.checkData(0, 0, 1) tdSql.query(f"select count(q_bool) from nested.stable_null_data;") - tdSql.checkData(0, 0, 0); + tdSql.checkData(0, 0, 0) tdSql.query(f"select tbname,count(*) from nested.stable_null_childtable group by tbname order by tbname;") tdSql.checkRows(2) - tdSql.checkData(0, 1, 1); - tdSql.checkData(1, 1, 62); + tdSql.checkData(0, 1, 1) + tdSql.checkData(1, 1, 62) tdSql.query(f"select count(*) from nested.stable_null_childtable where tbname ='$^%$%^&' ;") - tdSql.checkData(0, 0, 1); + tdSql.checkData(0, 0, 1) tdSql.query(f"select count(q_bool) from nested.stable_null_childtable;") - tdSql.checkData(0, 0, 0); + tdSql.checkData(0, 0, 0) + + tdSql.query(f"select tbname,count(*) from nested.stable_1 where ts is not null group by tbname order by tbname;") + tdSql.checkRows(7) + tdSql.query(f"select tbname,count(*) from nested.stable_1 where ts is null group by tbname order by tbname;") + tdSql.checkRows(7) + tdSql.query(f"select tbname,last(*) from nested.stable_1 where ts is not null group by tbname order by tbname;") + tdSql.checkRows(3) + tdSql.query(f"select tbname,last(*) from nested.stable_1 where ts is null group by tbname order by tbname;") + tdSql.checkRows(0) #test stable diff --git a/tests/system-test/2-query/sum.py b/tests/system-test/2-query/sum.py index 5e372bca20..f449597841 100644 --- a/tests/system-test/2-query/sum.py +++ b/tests/system-test/2-query/sum.py @@ -245,6 +245,10 @@ class TDTestCase: # stddev tdSql.checkData(0, 5, 0, True) + sql = f"select twa({col}) from {dbname}.d0" + tdSql.query(sql) + tdSql.checkData(0, 0, 1, True) + i += 1 diff --git a/tests/system-test/2-query/td-28068.py b/tests/system-test/2-query/td-28068.py new file mode 100644 index 0000000000..d77e012d9a --- /dev/null +++ b/tests/system-test/2-query/td-28068.py @@ -0,0 +1,68 @@ +from util.cases import * +from util.sql import * + +class TDTestCase: + def init(self, conn, logSql, replicaVar=1): + tdLog.debug("start to execute %s" % __file__) + tdSql.init(conn.cursor(), True) + + tdSql.execute("drop database if exists td_28068;") + tdSql.execute("create database td_28068;") + tdSql.execute("create database if not exists td_28068;") + tdSql.execute("create stable td_28068.st (ts timestamp, test_case nchar(10), time_cost float, num float) tags (branch nchar(10), scenario nchar(10));") + tdSql.execute("insert into td_28068.ct1 using td_28068.st (branch, scenario) tags ('3.0', 'scenario1') values (now(), 'query1', 1,2);") + tdSql.execute("insert into td_28068.ct1 using td_28068.st (branch, scenario) tags ('3.0', 'scenario1') values (now(), 'query1', 2,3);") + tdSql.execute("insert into td_28068.ct2 using td_28068.st (branch, scenario) tags ('3.0', 'scenario2') values (now(), 'query1', 10,1);") + tdSql.execute("insert into td_28068.ct2 using td_28068.st (branch, scenario) tags ('3.0', 'scenario2') values (now(), 'query1', 11,5);") + tdSql.execute("insert into td_28068.ct3 using td_28068.st (branch, scenario) tags ('3.1', 'scenario1') values (now(), 'query1', 20,4);") + tdSql.execute("insert into td_28068.ct3 using td_28068.st (branch, scenario) tags ('3.1', 'scenario1') values (now(), 'query1', 30,1);") + tdSql.execute("insert into td_28068.ct4 using td_28068.st (branch, scenario) tags ('3.1', 'scenario2') values (now(), 'query1', 8,8);") + tdSql.execute("insert into td_28068.ct4 using td_28068.st (branch, scenario) tags ('3.1', 'scenario2') values (now(), 'query1', 9,10);") + + def run(self): + tdSql.error('select last(ts) as ts, last(branch) as branch, last(scenario) as scenario, last(test_case) as test_case from td_28068.st group by branch, scenario order by last(branch);') + tdSql.error('select last(ts) as ts, last(branch) as branch1, last(scenario) as scenario, last(test_case) as test_case from td_28068.st group by branch, scenario order by last(branch), last(scenario); ') + + tdSql.query('select last(ts) as ts, last(branch) as branch1, last(scenario) as scenario, last(test_case) as test_case from td_28068.st group by branch, scenario order by last(branch); ') + tdSql.checkRows(4) + + tdSql.query('select last(ts) as ts, last(branch) as branch1, last(scenario) as scenario, last(test_case) from td_28068.st group by branch, scenario order by last(branch), last(test_case);') + tdSql.checkRows(4) + + tdSql.query('select last(ts) as ts, last(branch) as branch1, last(scenario) as scenario1, last(test_case) as test_case from td_28068.st group by branch, scenario order by last(branch), last(scenario);') + tdSql.checkRows(4) + + tdSql.query('select last(ts) as ts, last(branch) as branch1, last(scenario) as scenario1, last(test_case) as test_case from td_28068.st group by branch, scenario order by branch1, scenario1;') + tdSql.checkRows(4) + + tdSql.query('select last(ts) as ts, last(branch) as branch1, last(scenario) as scenario1, last(test_case) as test_case from td_28068.st group by tbname; ') + tdSql.checkRows(4) + + tdSql.query('select last(ts) as ts, last(branch) as branch1, last(scenario) as scenario1, last(test_case) as test_case from td_28068.st group by branch, scenario order by test_case;') + tdSql.checkRows(4) + + tdSql.query('select last(ts) as ts, last(branch) as branch1, last(scenario) as scenario1, last(test_case) as test_case1 from td_28068.st group by branch, scenario order by last(test_case);') + tdSql.checkRows(4) + + tdSql.query('select time_cost, num, time_cost + num as final_cost from td_28068.st partition by branch; ') + tdSql.checkRows(8) + + tdSql.query('select count(*) from td_28068.st partition by branch order by branch; ') + tdSql.checkRows(2) + + tdSql.query('select time_cost, num, time_cost + num as final_cost from td_28068.st order by time_cost;') + tdSql.checkRows(8) + + tdSql.query('select time_cost, num, time_cost + num as final_cost from td_28068.st order by final_cost;') + tdSql.checkRows(8) + + tdSql.execute("drop database if exists td_28068;") + + def stop(self): + tdSql.close() + tdLog.success(f"{__file__} successfully executed") + + + +tdCases.addLinux(__file__, TDTestCase()) +tdCases.addWindows(__file__, TDTestCase()) diff --git a/tests/system-test/7-tmq/tmqVnodeSplit-stb-false.py b/tests/system-test/7-tmq/tmqVnodeSplit-stb-false.py new file mode 100644 index 0000000000..384959c52b --- /dev/null +++ b/tests/system-test/7-tmq/tmqVnodeSplit-stb-false.py @@ -0,0 +1,218 @@ + +import taos +import sys +import time +import socket +import os +import threading +import math + +from util.log import * +from util.sql import * +from util.cases import * +from util.dnodes import * +from util.common import * +from util.cluster import * +sys.path.append("./7-tmq") +from tmqCommon import * + +from util.cluster import * +sys.path.append("./6-cluster") +from clusterCommonCreate import * +from clusterCommonCheck import clusterComCheck + +class TDTestCase: + def __init__(self): + self.vgroups = 1 + self.ctbNum = 10 + self.rowsPerTbl = 1000 + + def init(self, conn, logSql, replicaVar=1): + self.replicaVar = int(replicaVar) + tdLog.debug(f"start to excute {__file__}") + tdSql.init(conn.cursor(), True) + + def getDataPath(self): + selfPath = tdCom.getBuildPath() + + return selfPath + '/../sim/dnode%d/data/vnode/vnode%d/wal/*'; + + def prepareTestEnv(self): + tdLog.printNoPrefix("======== prepare test env include database, stable, ctables, and insert data: ") + paraDict = {'dbName': 'dbt', + 'dropFlag': 1, + 'event': '', + 'vgroups': 1, + 'stbName': 'stb', + 'colPrefix': 'c', + 'tagPrefix': 't', + 'colSchema': [{'type': 'INT', 'count':1},{'type': 'BIGINT', 'count':1},{'type': 'DOUBLE', 'count':1},{'type': 'BINARY', 'len':32, 'count':1},{'type': 'NCHAR', 'len':32, 'count':1},{'type': 'TIMESTAMP', 'count':1}], + 'tagSchema': [{'type': 'INT', 'count':1},{'type': 'BIGINT', 'count':1},{'type': 'DOUBLE', 'count':1},{'type': 'BINARY', 'len':32, 'count':1},{'type': 'NCHAR', 'len':32, 'count':1}], + 'ctbPrefix': 'ctb', + 'ctbStartIdx': 0, + 'ctbNum': 10, + 'rowsPerTbl': 1000, + 'batchNum': 10, + 'startTs': 1640966400000, # 2022-01-01 00:00:00.000 + 'pollDelay': 60, + 'showMsg': 1, + 'showRow': 1, + 'snapshot': 0} + + paraDict['vgroups'] = self.vgroups + paraDict['ctbNum'] = self.ctbNum + paraDict['rowsPerTbl'] = self.rowsPerTbl + + tdCom.drop_all_db() + tmqCom.initConsumerTable() + tdCom.create_database(tdSql, paraDict["dbName"],paraDict["dropFlag"], wal_retention_period=36000,vgroups=paraDict["vgroups"],replica=self.replicaVar) + tdLog.info("create stb") + tmqCom.create_stable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"]) + return + + def restartAndRemoveWal(self, deleteWal): + tdDnodes = cluster.dnodes + tdSql.query("select * from information_schema.ins_vnodes") + for result in tdSql.queryResult: + if result[2] == 'dbt': + tdLog.debug("dnode is %d"%(result[0])) + dnodeId = result[0] + vnodeId = result[1] + + tdDnodes[dnodeId - 1].stoptaosd() + time.sleep(1) + dataPath = self.getDataPath() + dataPath = dataPath%(dnodeId,vnodeId) + tdLog.debug("dataPath:%s"%dataPath) + if deleteWal: + if os.system('rm -rf ' + dataPath) != 0: + tdLog.exit("rm error") + + tdDnodes[dnodeId - 1].starttaosd() + time.sleep(1) + break + tdLog.debug("restart dnode ok") + + def splitVgroups(self): + tdSql.query("select * from information_schema.ins_vnodes") + vnodeId = 0 + for result in tdSql.queryResult: + if result[2] == 'dbt': + vnodeId = result[1] + tdLog.debug("vnode is %d"%(vnodeId)) + break + splitSql = "split vgroup %d" %(vnodeId) + tdLog.debug("splitSql:%s"%(splitSql)) + tdSql.query(splitSql) + tdLog.debug("splitSql ok") + + def tmqCase1(self, deleteWal=False): + tdLog.printNoPrefix("======== test case 1: ") + paraDict = {'dbName': 'dbt', + 'dropFlag': 1, + 'event': '', + 'vgroups': 1, + 'stbName': 'stb', + 'colPrefix': 'c', + 'tagPrefix': 't', + 'colSchema': [{'type': 'INT', 'count':1},{'type': 'BIGINT', 'count':1},{'type': 'DOUBLE', 'count':1},{'type': 'BINARY', 'len':32, 'count':1},{'type': 'NCHAR', 'len':32, 'count':1},{'type': 'TIMESTAMP', 'count':1}], + 'tagSchema': [{'type': 'INT', 'count':1},{'type': 'BIGINT', 'count':1},{'type': 'DOUBLE', 'count':1},{'type': 'BINARY', 'len':32, 'count':1},{'type': 'NCHAR', 'len':32, 'count':1}], + 'ctbPrefix': 'ctb1', + 'ctbStartIdx': 0, + 'ctbNum': 10, + 'rowsPerTbl': 1000, + 'batchNum': 10, + 'startTs': 1640966400000, # 2022-01-01 00:00:00.000 + 'pollDelay': 120, + 'showMsg': 1, + 'showRow': 1, + 'snapshot': 0} + + paraDict['vgroups'] = self.vgroups + paraDict['ctbNum'] = self.ctbNum + paraDict['rowsPerTbl'] = self.rowsPerTbl + + topicNameList = ['topic1'] + # expectRowsList = [] + tmqCom.initConsumerTable() + + tdLog.info("create topics from stb ") + queryString = "stable %s.%s"%(paraDict['dbName'], paraDict['stbName']) + # sqlString = "create topic %s as stable %s" %(topicNameList[0], paraDict['stbName']) + sqlString = "create topic %s as %s" %(topicNameList[0], queryString) + tdLog.info("create topic sql: %s"%sqlString) + tdSql.execute(sqlString) + # tdSql.query(queryString) + # expectRowsList.append(tdSql.getRows()) + + # init consume info, and start tmq_sim, then check consume result + tdLog.info("insert consume info to consume processor") + consumerId = 0 + expectrowcnt = paraDict["rowsPerTbl"] * paraDict["ctbNum"] * 2 + topicList = topicNameList[0] + ifcheckdata = 1 + ifManualCommit = 1 + keyList = 'group.id:cgrp1, enable.auto.commit:true, auto.commit.interval.ms:200, auto.offset.reset:earliest' + tmqCom.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit) + + tdLog.info("start consume processor") + tmqCom.startTmqSimProcess(pollDelay=paraDict['pollDelay'],dbName=paraDict["dbName"],showMsg=paraDict['showMsg'], showRow=paraDict['showRow'],snapshot=paraDict['snapshot']) + tdLog.info("wait the consume result") + + tdLog.info("create ctb1") + tmqCom.create_ctable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"],ctbPrefix=paraDict['ctbPrefix'], + ctbNum=paraDict["ctbNum"],ctbStartIdx=paraDict['ctbStartIdx']) + + tdLog.info("create ctb2") + paraDict2 = paraDict.copy() + paraDict2['ctbPrefix'] = "ctb2" + tmqCom.create_ctable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"],ctbPrefix=paraDict2['ctbPrefix'], + ctbNum=paraDict["ctbNum"],ctbStartIdx=paraDict['ctbStartIdx']) + + tdLog.info("insert ctb1 data") + pInsertThread = tmqCom.asyncInsertDataByInterlace(paraDict) + + tmqCom.getStartConsumeNotifyFromTmqsim() + tmqCom.getStartCommitNotifyFromTmqsim() + + #restart dnode & remove wal + self.restartAndRemoveWal(deleteWal) + + # split vgroup + self.splitVgroups() + + + tdLog.info("insert ctb2 data") + pInsertThread1 = tmqCom.asyncInsertDataByInterlace(paraDict2) + pInsertThread.join() + pInsertThread1.join() + + expectRows = 1 + resultList = tmqCom.selectConsumeResult(expectRows) + + if expectrowcnt / 2 > resultList[0]: + tdLog.info("expect consume rows: %d, act consume rows: %d"%(expectrowcnt / 2, resultList[0])) + tdLog.exit("%d tmq consume rows error!"%consumerId) + + # tmqCom.checkFileContent(consumerId, queryString) + + time.sleep(2) + for i in range(len(topicNameList)): + tdSql.query("drop topic %s"%topicNameList[i]) + + if deleteWal == True: + clusterComCheck.check_vgroups_status(vgroup_numbers=2,db_replica=self.replicaVar,db_name="dbt",count_number=240) + tdLog.printNoPrefix("======== test case 1 end ...... ") + + def run(self): + self.prepareTestEnv() + self.tmqCase1(False) + + def stop(self): + tdSql.close() + tdLog.success(f"{__file__} successfully executed") + +event = threading.Event() + +tdCases.addLinux(__file__, TDTestCase()) +tdCases.addWindows(__file__, TDTestCase()) diff --git a/tests/system-test/7-tmq/tmqVnodeSplit-stb-select-false.py b/tests/system-test/7-tmq/tmqVnodeSplit-stb-select-false.py new file mode 100644 index 0000000000..f01bf2558c --- /dev/null +++ b/tests/system-test/7-tmq/tmqVnodeSplit-stb-select-false.py @@ -0,0 +1,222 @@ + +import taos +import sys +import time +import socket +import os +import threading +import math + +from util.log import * +from util.sql import * +from util.cases import * +from util.dnodes import * +from util.common import * +from util.cluster import * +sys.path.append("./7-tmq") +from tmqCommon import * + +from util.cluster import * +sys.path.append("./6-cluster") +from clusterCommonCreate import * +from clusterCommonCheck import clusterComCheck + + + +class TDTestCase: + def __init__(self): + self.vgroups = 1 + self.ctbNum = 10 + self.rowsPerTbl = 1000 + + def init(self, conn, logSql, replicaVar=1): + self.replicaVar = int(replicaVar) + tdLog.debug(f"start to excute {__file__}") + tdSql.init(conn.cursor(), True) + + def getDataPath(self): + selfPath = tdCom.getBuildPath() + + return selfPath + '/../sim/dnode%d/data/vnode/vnode%d/wal/*'; + + def prepareTestEnv(self): + tdLog.printNoPrefix("======== prepare test env include database, stable, ctables, and insert data: ") + paraDict = {'dbName': 'dbt', + 'dropFlag': 1, + 'event': '', + 'vgroups': 1, + 'stbName': 'stb', + 'colPrefix': 'c', + 'tagPrefix': 't', + 'colSchema': [{'type': 'INT', 'count':1},{'type': 'BIGINT', 'count':1},{'type': 'DOUBLE', 'count':1},{'type': 'BINARY', 'len':32, 'count':1},{'type': 'NCHAR', 'len':32, 'count':1},{'type': 'TIMESTAMP', 'count':1}], + 'tagSchema': [{'type': 'INT', 'count':1},{'type': 'BIGINT', 'count':1},{'type': 'DOUBLE', 'count':1},{'type': 'BINARY', 'len':32, 'count':1},{'type': 'NCHAR', 'len':32, 'count':1}], + 'ctbPrefix': 'ctb', + 'ctbStartIdx': 0, + 'ctbNum': 10, + 'rowsPerTbl': 1000, + 'batchNum': 10, + 'startTs': 1640966400000, # 2022-01-01 00:00:00.000 + 'pollDelay': 60, + 'showMsg': 1, + 'showRow': 1, + 'snapshot': 0} + + paraDict['vgroups'] = self.vgroups + paraDict['ctbNum'] = self.ctbNum + paraDict['rowsPerTbl'] = self.rowsPerTbl + + tdCom.drop_all_db() + tmqCom.initConsumerTable() + tdCom.create_database(tdSql, paraDict["dbName"],paraDict["dropFlag"], wal_retention_period=36000,vgroups=paraDict["vgroups"],replica=self.replicaVar) + tdLog.info("create stb") + tmqCom.create_stable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"]) + return + + def restartAndRemoveWal(self, deleteWal): + tdDnodes = cluster.dnodes + tdSql.query("select * from information_schema.ins_vnodes") + for result in tdSql.queryResult: + if result[2] == 'dbt': + tdLog.debug("dnode is %d"%(result[0])) + dnodeId = result[0] + vnodeId = result[1] + + tdDnodes[dnodeId - 1].stoptaosd() + time.sleep(1) + dataPath = self.getDataPath() + dataPath = dataPath%(dnodeId,vnodeId) + tdLog.debug("dataPath:%s"%dataPath) + if deleteWal: + if os.system('rm -rf ' + dataPath) != 0: + tdLog.exit("rm error") + + tdDnodes[dnodeId - 1].starttaosd() + time.sleep(1) + break + tdLog.debug("restart dnode ok") + + def splitVgroups(self): + tdSql.query("select * from information_schema.ins_vnodes") + vnodeId = 0 + for result in tdSql.queryResult: + if result[2] == 'dbt': + vnodeId = result[1] + tdLog.debug("vnode is %d"%(vnodeId)) + break + splitSql = "split vgroup %d" %(vnodeId) + tdLog.debug("splitSql:%s"%(splitSql)) + tdSql.query(splitSql) + tdLog.debug("splitSql ok") + + def tmqCase1(self, deleteWal=False): + tdLog.printNoPrefix("======== test case 1: ") + paraDict = {'dbName': 'dbt', + 'dropFlag': 1, + 'event': '', + 'vgroups': 1, + 'stbName': 'stb', + 'colPrefix': 'c', + 'tagPrefix': 't', + 'colSchema': [{'type': 'INT', 'count':1},{'type': 'BIGINT', 'count':1},{'type': 'DOUBLE', 'count':1},{'type': 'BINARY', 'len':32, 'count':1},{'type': 'NCHAR', 'len':32, 'count':1},{'type': 'TIMESTAMP', 'count':1}], + 'tagSchema': [{'type': 'INT', 'count':1},{'type': 'BIGINT', 'count':1},{'type': 'DOUBLE', 'count':1},{'type': 'BINARY', 'len':32, 'count':1},{'type': 'NCHAR', 'len':32, 'count':1}], + 'ctbPrefix': 'ctb1', + 'ctbStartIdx': 0, + 'ctbNum': 10, + 'rowsPerTbl': 1000, + 'batchNum': 10, + 'startTs': 1640966400000, # 2022-01-01 00:00:00.000 + 'pollDelay': 180, + 'showMsg': 1, + 'showRow': 1, + 'snapshot': 0} + + paraDict['vgroups'] = self.vgroups + paraDict['ctbNum'] = self.ctbNum + paraDict['rowsPerTbl'] = self.rowsPerTbl + + topicNameList = ['topic1'] + # expectRowsList = [] + tmqCom.initConsumerTable() + + tdLog.info("create topics from stb with filter") + queryString = "select * from %s.%s"%(paraDict['dbName'], paraDict['stbName']) + # sqlString = "create topic %s as stable %s" %(topicNameList[0], paraDict['stbName']) + sqlString = "create topic %s as %s" %(topicNameList[0], queryString) + tdLog.info("create topic sql: %s"%sqlString) + tdSql.execute(sqlString) + # tdSql.query(queryString) + # expectRowsList.append(tdSql.getRows()) + + # init consume info, and start tmq_sim, then check consume result + tdLog.info("insert consume info to consume processor") + consumerId = 0 + expectrowcnt = paraDict["rowsPerTbl"] * paraDict["ctbNum"] * 2 + topicList = topicNameList[0] + ifcheckdata = 1 + ifManualCommit = 1 + keyList = 'group.id:cgrp1, enable.auto.commit:true, auto.commit.interval.ms:200, auto.offset.reset:earliest' + tmqCom.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit) + + tdLog.info("start consume processor") + tmqCom.startTmqSimProcess(pollDelay=paraDict['pollDelay'],dbName=paraDict["dbName"],showMsg=paraDict['showMsg'], showRow=paraDict['showRow'],snapshot=paraDict['snapshot']) + tdLog.info("wait the consume result") + + tdLog.info("create ctb1") + tmqCom.create_ctable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"],ctbPrefix=paraDict['ctbPrefix'], + ctbNum=paraDict["ctbNum"],ctbStartIdx=paraDict['ctbStartIdx']) + + tdLog.info("create ctb2") + paraDict2 = paraDict.copy() + paraDict2['ctbPrefix'] = "ctb2" + tmqCom.create_ctable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"],ctbPrefix=paraDict2['ctbPrefix'], + ctbNum=paraDict["ctbNum"],ctbStartIdx=paraDict['ctbStartIdx']) + + tdLog.info("insert ctb1 data") + pInsertThread = tmqCom.asyncInsertDataByInterlace(paraDict) + + tmqCom.getStartConsumeNotifyFromTmqsim() + tmqCom.getStartCommitNotifyFromTmqsim() + + #restart dnode & remove wal + self.restartAndRemoveWal(deleteWal) + + # split vgroup + self.splitVgroups() + + + tdLog.info("insert ctb2 data") + pInsertThread1 = tmqCom.asyncInsertDataByInterlace(paraDict2) + pInsertThread.join() + pInsertThread1.join() + + expectRows = 1 + + + resultList = tmqCom.selectConsumeResult(expectRows) + + if expectrowcnt / 2 > resultList[0]: + tdLog.info("expect consume rows: %d, act consume rows: %d"%(expectrowcnt / 2, resultList[0])) + tdLog.exit("%d tmq consume rows error!"%consumerId) + + # tmqCom.checkFileContent(consumerId, queryString) + + time.sleep(2) + for i in range(len(topicNameList)): + tdSql.query("drop topic %s"%topicNameList[i]) + + if deleteWal == True: + clusterComCheck.check_vgroups_status(vgroup_numbers=2,db_replica=self.replicaVar,db_name="dbt",count_number=240) + tdLog.printNoPrefix("======== test case 1 end ...... ") + + def run(self): + self.prepareTestEnv() + self.tmqCase1(False) + + def stop(self): + tdSql.close() + tdLog.success(f"{__file__} successfully executed") + +event = threading.Event() + +tdCases.addLinux(__file__, TDTestCase()) +tdCases.addWindows(__file__, TDTestCase()) diff --git a/tests/system-test/7-tmq/tmqVnodeSplit-stb-select.py b/tests/system-test/7-tmq/tmqVnodeSplit-stb-select.py index e79a0ca77d..5e11de04cb 100644 --- a/tests/system-test/7-tmq/tmqVnodeSplit-stb-select.py +++ b/tests/system-test/7-tmq/tmqVnodeSplit-stb-select.py @@ -190,6 +190,8 @@ class TDTestCase: pInsertThread1.join() expectRows = 1 + + resultList = tmqCom.selectConsumeResult(expectRows) if expectrowcnt / 2 > resultList[0]: @@ -209,8 +211,6 @@ class TDTestCase: def run(self): self.prepareTestEnv() self.tmqCase1(True) - self.prepareTestEnv() - self.tmqCase1(False) def stop(self): tdSql.close() diff --git a/tests/system-test/7-tmq/tmqVnodeSplit-stb.py b/tests/system-test/7-tmq/tmqVnodeSplit-stb.py index a96d52ee2b..5aa2054e96 100644 --- a/tests/system-test/7-tmq/tmqVnodeSplit-stb.py +++ b/tests/system-test/7-tmq/tmqVnodeSplit-stb.py @@ -207,8 +207,6 @@ class TDTestCase: def run(self): self.prepareTestEnv() self.tmqCase1(True) - self.prepareTestEnv() - self.tmqCase1(False) def stop(self): tdSql.close() diff --git a/tests/system-test/8-stream/at_once_interval.py b/tests/system-test/8-stream/at_once_interval.py index 8f5438be37..763b88bc69 100644 --- a/tests/system-test/8-stream/at_once_interval.py +++ b/tests/system-test/8-stream/at_once_interval.py @@ -106,14 +106,18 @@ class TDTestCase: ptn_counter = 0 for c1_value in tdSql.queryResult: if partition == "c1": - tdSql.query(f'select count(*) from `{tname}_{self.tdCom.subtable_prefix}{abs(c1_value[1])}{self.tdCom.subtable_suffix}`;') + tbname = self.tdCom.get_subtable_wait(f'{tname}_{self.tdCom.subtable_prefix}{abs(c1_value[1])}{self.tdCom.subtable_suffix}') + tdSql.query(f'select count(*) from `{tbname}`') elif partition is None: - tdSql.query(f'select count(*) from `{tname}_{self.tdCom.subtable_prefix}no_partition{self.tdCom.subtable_suffix}`;') + tbname = self.tdCom.get_subtable_wait(f'{tname}_{self.tdCom.subtable_prefix}no_partition{self.tdCom.subtable_suffix}') + tdSql.query(f'select count(*) from `{tbname}`') elif partition == "abs(c1)": abs_c1_value = abs(c1_value[1]) - tdSql.query(f'select count(*) from `{tname}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}`;') + tbname = self.tdCom.get_subtable_wait(f'{tname}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}') + tdSql.query(f'select count(*) from `{tbname}`') elif partition == "tbname" and ptn_counter == 0: - tdSql.query(f'select count(*) from `{tname}_{self.tdCom.subtable_prefix}{self.ctb_name}{self.tdCom.subtable_suffix}`;') + tbname = self.tdCom.get_subtable_wait(f'{tname}_{self.tdCom.subtable_prefix}{self.ctb_name}{self.tdCom.subtable_suffix}') + tdSql.query(f'select count(*) from `{tbname}`') ptn_counter += 1 tdSql.checkEqual(tdSql.queryResult[0][0] > 0, True) @@ -121,14 +125,18 @@ class TDTestCase: ptn_counter = 0 for c1_value in tdSql.queryResult: if partition == "c1": - tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}{abs(c1_value[1])}{self.tdCom.subtable_suffix}`;') + tbname = self.tdCom.get_subtable_wait(f'{self.tb_name}_{self.tdCom.subtable_prefix}{abs(c1_value[1])}{self.tdCom.subtable_suffix}') + tdSql.query(f'select count(*) from `{tbname}`') elif partition is None: - tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}no_partition{self.tdCom.subtable_suffix}`;') + tbname = self.tdCom.get_subtable_wait(f'{self.tb_name}_{self.tdCom.subtable_prefix}no_partition{self.tdCom.subtable_suffix}') + tdSql.query(f'select count(*) from `{tbname}`') elif partition == "abs(c1)": abs_c1_value = abs(c1_value[1]) - tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}`;') + tbname = self.tdCom.get_subtable_wait(f'{self.tb_name}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}') + tdSql.query(f'select count(*) from `{tbname}`') elif partition == "tbname" and ptn_counter == 0: - tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}{self.tb_name}{self.tdCom.subtable_suffix}`;') + tbname = self.tdCom.get_subtable_wait(f'{self.tb_name}_{self.tdCom.subtable_prefix}{self.tb_name}{self.tdCom.subtable_suffix}') + tdSql.query(f'select count(*) from `{tbname}`') ptn_counter += 1 tdSql.checkEqual(tdSql.queryResult[0][0] > 0, True) @@ -208,8 +216,8 @@ class TDTestCase: self.at_once_interval(interval=random.randint(10, 15), partition=None, delete=True) self.at_once_interval(interval=random.randint(10, 15), partition=self.tdCom.stream_case_when_tbname, case_when=f'case when {self.tdCom.stream_case_when_tbname} = tbname then {self.tdCom.partition_tbname_alias} else tbname end') self.at_once_interval(interval=random.randint(10, 15), partition="tbname", fill_history_value=1, fill_value="NULL") - # for fill_value in ["NULL", "PREV", "NEXT", "LINEAR", "VALUE,1,2,3,4,5,6,7,8,9,10,11,1,2,3,4,5,6,7,8,9,10,11"]: - for fill_value in ["PREV", "NEXT", "LINEAR", "VALUE,1,2,3,4,5,6,7,8,9,10,11,1,2,3,4,5,6,7,8,9,10,11"]: + for fill_value in ["NULL", "PREV", "NEXT", "LINEAR", "VALUE,1,2,3,4,5,6,7,8,9,10,11,1,2,3,4,5,6,7,8,9,10,11"]: + # for fill_value in ["PREV", "NEXT", "LINEAR", "VALUE,1,2,3,4,5,6,7,8,9,10,11,1,2,3,4,5,6,7,8,9,10,11"]: self.at_once_interval(interval=random.randint(10, 15), partition="tbname", fill_value=fill_value) self.at_once_interval(interval=random.randint(10, 15), partition="tbname", fill_value=fill_value, delete=True) diff --git a/tests/system-test/8-stream/at_once_interval_ext.py b/tests/system-test/8-stream/at_once_interval_ext.py index e1dc057448..cbb4d7ca5a 100644 --- a/tests/system-test/8-stream/at_once_interval_ext.py +++ b/tests/system-test/8-stream/at_once_interval_ext.py @@ -153,12 +153,15 @@ class TDTestCase: ptn_counter = 0 for c1_value in tdSql.queryResult: if partition == "c1": - tdSql.query(f'select count(*) from `{tname}_{self.tdCom.subtable_prefix}{abs(c1_value[1])}{self.tdCom.subtable_suffix}`;') + tbname = self.tdCom.get_subtable_wait(f'{tname}_{self.tdCom.subtable_prefix}{abs(c1_value[1])}{self.tdCom.subtable_suffix}') + tdSql.query(f'select count(*) from `{tbname}`') elif partition == "abs(c1)": abs_c1_value = abs(c1_value[1]) - tdSql.query(f'select count(*) from `{tname}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}`;') + tbname = self.tdCom.get_subtable_wait(f'{tname}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}') + tdSql.query(f'select count(*) from `{tbname}`') elif partition == "tbname" and ptn_counter == 0: - tdSql.query(f'select count(*) from `{tname}_{self.tdCom.subtable_prefix}{self.ctb_name}{self.tdCom.subtable_suffix}`;') + tbname = self.tdCom.get_subtable_wait(f'{tname}_{self.tdCom.subtable_prefix}{self.ctb_name}{self.tdCom.subtable_suffix}') + tdSql.query(f'select count(*) from `{tbname}`') ptn_counter += 1 else: tdSql.query(f'select cast(cast(cast({c1_value[1]} as int unsigned) as bigint) as varchar(100))') @@ -166,7 +169,8 @@ class TDTestCase: if subtable == "constant": return else: - tdSql.query(f'select count(*) from `{tname}_{self.tdCom.subtable_prefix}{subtable_value}{self.tdCom.subtable_suffix}`;') + tbname = self.tdCom.get_subtable_wait(f'{tname}_{self.tdCom.subtable_prefix}{subtable_value}{self.tdCom.subtable_suffix}') + tdSql.query(f'select count(*) from `{tbname}`') tdSql.checkEqual(tdSql.queryResult[0][0] > 0, True) def run(self): diff --git a/tests/system-test/8-stream/at_once_session.py b/tests/system-test/8-stream/at_once_session.py index 3462561c4e..6f25e5ad97 100644 --- a/tests/system-test/8-stream/at_once_session.py +++ b/tests/system-test/8-stream/at_once_session.py @@ -171,35 +171,43 @@ class TDTestCase: for c1_value in tdSql.queryResult: if c1_value[1] is not None: if partition == "c1": - tdSql.query(f'select count(*) from `{self.ctb_name}_{self.tdCom.subtable_prefix}{c1_value[1]}{self.tdCom.subtable_suffix}`;') + tbname = self.tdCom.get_subtable_wait(f'{self.ctb_name}_{self.tdCom.subtable_prefix}{c1_value[1]}{self.tdCom.subtable_suffix}') + tdSql.query(f'select count(*) from `{tbname}`') elif partition == "abs(c1)": if subtable: abs_c1_value = abs(c1_value[1]) - tdSql.query(f'select count(*) from `{self.ctb_name}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}`;') + tbname = self.tdCom.get_subtable_wait(f'{self.ctb_name}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}') + tdSql.query(f'select count(*) from `{tbname}`') else: - tdSql.query(f'select count(*) from `{self.ctb_name}_{self.tdCom.subtable_prefix}{partition_elm_alias}{self.tdCom.subtable_suffix}`;') + tbname = self.tdCom.get_subtable_wait(f'{self.ctb_name}_{self.tdCom.subtable_prefix}{partition_elm_alias}{self.tdCom.subtable_suffix}') + tdSql.query(f'select count(*) from `{tbname}`') elif partition == "tbname" and ptn_counter == 0: - tdSql.query(f'select count(*) from `{self.ctb_name}_{self.tdCom.subtable_prefix}{self.ctb_name}{self.tdCom.subtable_suffix}`;') + tbname = self.tdCom.get_subtable_wait(f'{self.ctb_name}_{self.tdCom.subtable_prefix}{self.ctb_name}{self.tdCom.subtable_suffix}') + tdSql.query(f'select count(*) from `{tbname}`') ptn_counter += 1 - tdSql.checkEqual(tdSql.queryResult[0][0] > 0, True) + tdSql.checkEqual(tdSql.queryResult[0][0] > 0, True) if subtable is not None else tdSql.checkEqual(tdSql.queryResult[0][0] >= 0, True) tdSql.query(f'select * from {self.tb_name}') ptn_counter = 0 for c1_value in tdSql.queryResult: if c1_value[1] is not None: if partition == "c1": - tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}{c1_value[1]}{self.tdCom.subtable_suffix}`;') + tbname = self.tdCom.get_subtable_wait(f'{self.tb_name}_{self.tdCom.subtable_prefix}{c1_value[1]}{self.tdCom.subtable_suffix}') + tdSql.query(f'select count(*) from `{tbname}`') elif partition == "abs(c1)": if subtable: abs_c1_value = abs(c1_value[1]) - tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}`;') + tbname = self.tdCom.get_subtable_wait(f'{self.tb_name}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}') + tdSql.query(f'select count(*) from `{tbname}`') else: - tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}{partition_elm_alias}{self.tdCom.subtable_suffix}`;') + tbname = self.tdCom.get_subtable_wait(f'{self.tb_name}_{self.tdCom.subtable_prefix}{partition_elm_alias}{self.tdCom.subtable_suffix}') + tdSql.query(f'select count(*) from `{tbname}`') elif partition == "tbname" and ptn_counter == 0: - tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}{self.tb_name}{self.tdCom.subtable_suffix}`;') + tbname = self.tdCom.get_subtable_wait(f'{self.tb_name}_{self.tdCom.subtable_prefix}{self.tb_name}{self.tdCom.subtable_suffix}') + tdSql.query(f'select count(*) from `{tbname}`') ptn_counter += 1 - tdSql.checkEqual(tdSql.queryResult[0][0] > 0, True) + tdSql.checkEqual(tdSql.queryResult[0][0] > 0, True) if subtable is not None else tdSql.checkEqual(tdSql.queryResult[0][0] >= 0, True) @@ -224,4 +232,4 @@ class TDTestCase: event = threading.Event() tdCases.addLinux(__file__, TDTestCase()) -tdCases.addWindows(__file__, TDTestCase()) \ No newline at end of file +tdCases.addWindows(__file__, TDTestCase()) diff --git a/tests/system-test/8-stream/at_once_state_window.py b/tests/system-test/8-stream/at_once_state_window.py index 60a4f4e890..933a41553e 100644 --- a/tests/system-test/8-stream/at_once_state_window.py +++ b/tests/system-test/8-stream/at_once_state_window.py @@ -94,15 +94,19 @@ class TDTestCase: for c1_value in tdSql.queryResult: if partition == "c1": if subtable: - tdSql.query(f'select count(*) from `{self.ctb_name}_{self.tdCom.subtable_prefix}{c1_value[1]}{self.tdCom.subtable_suffix}`;') + tbname = self.tdCom.get_subtable_wait(f'{self.ctb_name}_{self.tdCom.subtable_prefix}{c1_value[1]}{self.tdCom.subtable_suffix}') + tdSql.query(f'select count(*) from `{tbname}`') else: - tdSql.query(f'select count(*) from `{self.ctb_name}_{self.tdCom.subtable_prefix}{partition_elm_alias}{self.tdCom.subtable_suffix}`;') + tbname = self.tdCom.get_subtable_wait(f'{self.ctb_name}_{self.tdCom.subtable_prefix}{partition_elm_alias}{self.tdCom.subtable_suffix}') + tdSql.query(f'select count(*) from `{tbname}`') return elif partition == "abs(c1)": abs_c1_value = abs(c1_value[1]) - tdSql.query(f'select count(*) from `{self.ctb_name}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}`;') + tbname = self.tdCom.get_subtable_wait(f'{self.ctb_name}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}') + tdSql.query(f'select count(*) from `{tbname}`') elif partition == "tbname" and ptn_counter == 0: - tdSql.query(f'select count(*) from `{self.ctb_name}_{self.tdCom.subtable_prefix}{self.ctb_name}{self.tdCom.subtable_suffix}`;') + tbname = self.tdCom.get_subtable_wait(f'{self.ctb_name}_{self.tdCom.subtable_prefix}{self.ctb_name}{self.tdCom.subtable_suffix}') + tdSql.query(f'select count(*) from `{tbname}`') ptn_counter += 1 tdSql.checkEqual(tdSql.queryResult[0][0] > 0, True) @@ -111,17 +115,20 @@ class TDTestCase: for c1_value in tdSql.queryResult: if partition == "c1": if subtable: - tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}{c1_value[1]}{self.tdCom.subtable_suffix}`;') + tbname = self.tdCom.get_subtable_wait(f'{self.tb_name}_{self.tdCom.subtable_prefix}{c1_value[1]}{self.tdCom.subtable_suffix}') + tdSql.query(f'select count(*) from `{tbname}`') else: - tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}{partition_elm_alias}{self.tdCom.subtable_suffix}`;') + tbname = self.tdCom.get_subtable_wait(f'{self.tb_name}_{self.tdCom.subtable_prefix}{partition_elm_alias}{self.tdCom.subtable_suffix}') + tdSql.query(f'select count(*) from `{tbname}`') return elif partition == "abs(c1)": abs_c1_value = abs(c1_value[1]) - tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}`;') + tbname = self.tdCom.get_subtable_wait(f'{self.tb_name}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}') + tdSql.query(f'select count(*) from `{tbname}`') elif partition == "tbname" and ptn_counter == 0: - tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}{self.tb_name}{self.tdCom.subtable_suffix}`;') + tbname = self.tdCom.get_subtable_wait(f'{self.tb_name}_{self.tdCom.subtable_prefix}{self.tb_name}{self.tdCom.subtable_suffix}') + tdSql.query(f'select count(*) from `{tbname}`') ptn_counter += 1 - tdSql.checkEqual(tdSql.queryResult[0][0] > 0, True) diff --git a/tests/system-test/8-stream/pause_resume_test.py b/tests/system-test/8-stream/pause_resume_test.py index 484383f1ce..8b71466f01 100644 --- a/tests/system-test/8-stream/pause_resume_test.py +++ b/tests/system-test/8-stream/pause_resume_test.py @@ -117,14 +117,18 @@ class TDTestCase: ptn_counter = 0 for c1_value in tdSql.queryResult: if partition == "c1": - tdSql.query(f'select count(*) from `{tname}_{self.tdCom.subtable_prefix}{abs(c1_value[1])}{self.tdCom.subtable_suffix}`;') + tbname = self.tdCom.get_subtable_wait(f'{tname}_{self.tdCom.subtable_prefix}{abs(c1_value[1])}{self.tdCom.subtable_suffix}') + tdSql.query(f'select count(*) from `{tbname}`') elif partition is None: - tdSql.query(f'select count(*) from `{tname}_{self.tdCom.subtable_prefix}no_partition{self.tdCom.subtable_suffix}`;') + tbname = self.tdCom.get_subtable_wait(f'{tname}_{self.tdCom.subtable_prefix}no_partition{self.tdCom.subtable_suffix}') + tdSql.query(f'select count(*) from `{tbname}`') elif partition == "abs(c1)": abs_c1_value = abs(c1_value[1]) - tdSql.query(f'select count(*) from `{tname}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}`;') + tbname = self.tdCom.get_subtable_wait(f'{tname}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}') + tdSql.query(f'select count(*) from `{tbname}`') elif partition == "tbname" and ptn_counter == 0: - tdSql.query(f'select count(*) from `{tname}_{self.tdCom.subtable_prefix}{self.ctb_name}{self.tdCom.subtable_suffix}`;') + tbname = self.tdCom.get_subtable_wait(f'{tname}_{self.tdCom.subtable_prefix}{self.ctb_name}{self.tdCom.subtable_suffix}') + tdSql.query(f'select count(*) from `{tbname}`') ptn_counter += 1 tdSql.checkEqual(tdSql.queryResult[0][0] > 0, True) @@ -132,14 +136,18 @@ class TDTestCase: ptn_counter = 0 for c1_value in tdSql.queryResult: if partition == "c1": - tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}{abs(c1_value[1])}{self.tdCom.subtable_suffix}`;') + tbname = self.tdCom.get_subtable_wait(f'{self.tb_name}_{self.tdCom.subtable_prefix}{abs(c1_value[1])}{self.tdCom.subtable_suffix}') + tdSql.query(f'select count(*) from `{tbname}`') elif partition is None: - tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}no_partition{self.tdCom.subtable_suffix}`;') + tbname = self.tdCom.get_subtable_wait(f'{self.tb_name}_{self.tdCom.subtable_prefix}no_partition{self.tdCom.subtable_suffix}') + tdSql.query(f'select count(*) from `{tbname}`') elif partition == "abs(c1)": abs_c1_value = abs(c1_value[1]) - tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}`;') + tbname = self.tdCom.get_subtable_wait(f'{self.tb_name}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}') + tdSql.query(f'select count(*) from `{tbname}`') elif partition == "tbname" and ptn_counter == 0: - tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}{self.tb_name}{self.tdCom.subtable_suffix}`;') + tbname = self.tdCom.get_subtable_wait(f'{self.tb_name}_{self.tdCom.subtable_prefix}{self.tb_name}{self.tdCom.subtable_suffix}') + tdSql.query(f'select count(*) from `{tbname}`') ptn_counter += 1 tdSql.checkEqual(tdSql.queryResult[0][0] > 0, True) diff --git a/tests/system-test/8-stream/window_close_interval.py b/tests/system-test/8-stream/window_close_interval.py index 31a566a0f8..1ac7a61d2f 100644 --- a/tests/system-test/8-stream/window_close_interval.py +++ b/tests/system-test/8-stream/window_close_interval.py @@ -146,12 +146,15 @@ class TDTestCase: ptn_counter = 0 for c1_value in tdSql.queryResult: if partition == "c1": - tdSql.query(f'select count(*) from `{tname}_{self.tdCom.subtable_prefix}{c1_value[1]}{self.tdCom.subtable_suffix}`;', count_expected_res=self.tdCom.range_count) + tbname = self.tdCom.get_subtable_wait(f'{tname}_{self.tdCom.subtable_prefix}{c1_value[1]}{self.tdCom.subtable_suffix}') + tdSql.query(f'select count(*) from `{tbname}`', count_expected_res=self.tdCom.range_count) elif partition == "abs(c1)": abs_c1_value = abs(c1_value[1]) - tdSql.query(f'select count(*) from `{tname}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}`;', count_expected_res=self.tdCom.range_count) + tbname = self.tdCom.get_subtable_wait(f'{tname}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}') + tdSql.query(f'select count(*) from `{tbname}`', count_expected_res=self.tdCom.range_count) elif partition == "tbname" and ptn_counter == 0: - tdSql.query(f'select count(*) from `{tname}_{self.tdCom.subtable_prefix}{self.ctb_name}{self.tdCom.subtable_suffix}`;', count_expected_res=self.tdCom.range_count) + tbname = self.tdCom.get_subtable_wait(f'{tname}_{self.tdCom.subtable_prefix}{self.ctb_name}{self.tdCom.subtable_suffix}') + tdSql.query(f'select count(*) from `{tbname}`', count_expected_res=self.tdCom.range_count) ptn_counter += 1 tdSql.checkEqual(tdSql.queryResult[0][0] , self.tdCom.range_count) @@ -161,12 +164,15 @@ class TDTestCase: ptn_counter = 0 for c1_value in tdSql.queryResult: if partition == "c1": - tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}{c1_value[1]}{self.tdCom.subtable_suffix}`;') + tbname = self.tdCom.get_subtable_wait(f'{self.tb_name}_{self.tdCom.subtable_prefix}{c1_value[1]}{self.tdCom.subtable_suffix}') + tdSql.query(f'select count(*) from `{tbname}`') elif partition == "abs(c1)": abs_c1_value = abs(c1_value[1]) - tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}`;') + tbname = self.tdCom.get_subtable_wait(f'{self.tb_name}_{self.tdCom.subtable_prefix}{abs_c1_value}{self.tdCom.subtable_suffix}') + tdSql.query(f'select count(*) from `{tbname}`') elif partition == "tbname" and ptn_counter == 0: - tdSql.query(f'select count(*) from `{self.tb_name}_{self.tdCom.subtable_prefix}{self.tb_name}{self.tdCom.subtable_suffix}`;') + tbname = self.tdCom.get_subtable_wait(f'{self.tb_name}_{self.tdCom.subtable_prefix}{self.tb_name}{self.tdCom.subtable_suffix}') + tdSql.query(f'select count(*) from `{tbname}`') ptn_counter += 1 tdSql.checkEqual(tdSql.queryResult[0][0] > 0, True)