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

Merge branch '3.0' into fix/3_liaohj

This commit is contained in:
Haojun Liao 2024-01-17 15:14:52 +08:00
parent 1ea671e373 fc16d26da2
commit 56eac88349
42 changed files with 966 additions and 347 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
include/common/tdatablock.h
View File

@ -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, int32_t buildSubmitReqFromDataBlock(SSubmitReq2** pReq, const SSDataBlock* pDataBlocks, const STSchema* pTSchema, int64_t uid, int32_t vgId,
tb_uid_t suid); 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); char* buildCtbNameByGroupId(const char* stbName, uint64_t groupId);
int32_t buildCtbNameByGroupIdImpl(const char* stbName, uint64_t groupId, char* pBuf); int32_t buildCtbNameByGroupIdImpl(const char* stbName, uint64_t groupId, char* pBuf);

9
include/libs/qcom/query.h
View File

@ -21,12 +21,12 @@
extern "C" { extern "C" {
#endif #endif
#include "systable.h"
#include "tarray.h" #include "tarray.h"
#include "thash.h" #include "thash.h"
#include "tlog.h" #include "tlog.h"
#include "tmsg.h" #include "tmsg.h"
#include "tmsgcb.h" #include "tmsgcb.h"
#include "systable.h"
typedef enum { typedef enum {
JOB_TASK_STATUS_NULL = 0, JOB_TASK_STATUS_NULL = 0,
@ -90,7 +90,6 @@ typedef struct SExecResult {
void* res; void* res;
} SExecResult; } SExecResult;
#pragma pack(push, 1) #pragma pack(push, 1)
typedef struct SCTableMeta { typedef struct SCTableMeta {
uint64_t uid; uint64_t uid;
@ -100,7 +99,6 @@ typedef struct SCTableMeta {
} SCTableMeta; } SCTableMeta;
#pragma pack(pop) #pragma pack(pop)
#pragma pack(push, 1) #pragma pack(push, 1)
typedef struct STableMeta { typedef struct STableMeta {
// BEGIN: KEEP THIS PART SAME WITH SCTableMeta // BEGIN: KEEP THIS PART SAME WITH SCTableMeta
@ -173,6 +171,7 @@ typedef struct SDataBuf {
void* pData; void* pData;
uint32_t len; uint32_t len;
void* handle; void* handle;
int64_t handleRefId;
SEpSet* pEpSet; SEpSet* pEpSet;
} SDataBuf; } SDataBuf;
@ -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) ((_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 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) \ #define NEED_REDIRECT_ERROR(_code) \
(NO_RET_REDIRECT_ERROR(_code) || SYNC_UNKNOWN_LEADER_REDIRECT_ERROR(_code) || \ (NO_RET_REDIRECT_ERROR(_code) || SYNC_UNKNOWN_LEADER_REDIRECT_ERROR(_code) || \

3
include/libs/stream/tstream.h
View File

@ -62,9 +62,10 @@ typedef struct SStreamTask SStreamTask;
typedef struct SStreamQueue SStreamQueue; typedef struct SStreamQueue SStreamQueue;
typedef struct SStreamTaskSM SStreamTaskSM; typedef struct SStreamTaskSM SStreamTaskSM;
#define SSTREAM_TASK_VER 2 #define SSTREAM_TASK_VER 3
#define SSTREAM_TASK_INCOMPATIBLE_VER 1 #define SSTREAM_TASK_INCOMPATIBLE_VER 1
#define SSTREAM_TASK_NEED_CONVERT_VER 2 #define SSTREAM_TASK_NEED_CONVERT_VER 2
#define SSTREAM_TASK_SUBTABLE_CHANGED_VER 3
enum { enum {
STREAM_STATUS__NORMAL = 0, STREAM_STATUS__NORMAL = 0,

18
source/client/src/clientImpl.c
View File

@ -324,7 +324,8 @@ void asyncExecLocalCmd(SRequestObj* pRequest, SQuery* pQuery) {
return; 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) { if (TSDB_CODE_SUCCESS == code && NULL != pRsp) {
code = setQueryResultFromRsp(&pRequest->body.resInfo, pRsp, false, true); 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].bytes = pSchema[i].bytes;
pResInfo->userFields[i].type = pSchema[i].type; 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; pResInfo->userFields[i].bytes -= VARSTR_HEADER_SIZE;
} else if (pSchema[i].type == TSDB_DATA_TYPE_NCHAR || pSchema[i].type == TSDB_DATA_TYPE_JSON) { } 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; pResInfo->userFields[i].bytes = (pResInfo->userFields[i].bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE;
@ -1510,8 +1512,12 @@ int32_t doProcessMsgFromServer(void* param) {
updateTargetEpSet(pSendInfo, pTscObj, pMsg, pEpSet); updateTargetEpSet(pSendInfo, pTscObj, pMsg, pEpSet);
SDataBuf buf = { SDataBuf buf = {.msgType = pMsg->msgType,
.msgType = pMsg->msgType, .len = pMsg->contLen, .pData = NULL, .handle = pMsg->info.handle, .pEpSet = pEpSet}; .len = pMsg->contLen,
.pData = NULL,
.handle = pMsg->info.handle,
.handleRefId = pMsg->info.refId,
.pEpSet = pEpSet};
if (pMsg->contLen > 0) { if (pMsg->contLen > 0) {
buf.pData = taosMemoryCalloc(1, pMsg->contLen); buf.pData = taosMemoryCalloc(1, pMsg->contLen);
@ -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 #ifndef TD_ENTERPRISE
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
#else #else
return clientParseSqlImpl(param, dbName, sql, parseOnly, effectiveUser, pRes); return clientParseSqlImpl(param, dbName, sql, parseOnly, effectiveUser, pRes);
#endif #endif
} }

24
source/common/src/tdatablock.c
View File

@ -2118,6 +2118,30 @@ _end:
return TSDB_CODE_SUCCESS; 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* buildCtbNameByGroupId(const char* stbFullName, uint64_t groupId) {
char* pBuf = taosMemoryCalloc(1, TSDB_TABLE_NAME_LEN + 1); char* pBuf = taosMemoryCalloc(1, TSDB_TABLE_NAME_LEN + 1);
if (!pBuf) { if (!pBuf) {

4
source/dnode/vnode/src/inc/tq.h
View File

@ -145,7 +145,7 @@ int32_t tqOffsetCommitFile(STqOffsetStore* pStore);
// tqSink // tqSink
int32_t tqBuildDeleteReq(STQ* pTq, const char* stbFullName, const SSDataBlock* pDataBlock, SBatchDeleteReq* deleteReq, 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); void tqSinkDataIntoDstTable(SStreamTask* pTask, void* vnode, void* data);
// tqOffset // 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); int32_t doMergeExistedRows(SSubmitTbData* pExisted, const SSubmitTbData* pNew, const char* id);
SVCreateTbReq* buildAutoCreateTableReq(const char* stbFullName, int64_t suid, int32_t numOfCols, SVCreateTbReq* buildAutoCreateTableReq(const char* stbFullName, int64_t suid, int32_t numOfCols,
SSDataBlock* pDataBlock, SArray* pTagArray); SSDataBlock* pDataBlock, SArray* pTagArray, bool newSubTableRule);
#ifdef __cplusplus #ifdef __cplusplus
} }

2
source/dnode/vnode/src/sma/smaRollup.c
View File

@ -703,7 +703,7 @@ static int32_t tdRSmaExecAndSubmitResult(SSma *pSma, qTaskInfo_t taskInfo, SRSma
code = terrno; code = terrno;
TSDB_CHECK_CODE(code, lino, _exit); 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); TSDB_CHECK_CODE(code, lino, _exit);
code = tdRSmaProcessDelReq(pSma, suid, pItem->level, &deleteReq); code = tdRSmaProcessDelReq(pSma, suid, pItem->level, &deleteReq);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);

4
source/dnode/vnode/src/sma/smaTimeRange.c
View File

@ -188,7 +188,7 @@ int32_t smaBlockToSubmit(SVnode *pVnode, const SArray *pBlocks, const STSchema *
if (pDataBlock->info.type == STREAM_DELETE_RESULT) { if (pDataBlock->info.type == STREAM_DELETE_RESULT) {
pDeleteReq->suid = suid; pDeleteReq->suid = suid;
pDeleteReq->deleteReqs = taosArrayInit(0, sizeof(SSingleDeleteReq)); 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); TSDB_CHECK_CODE(code, lino, _exit);
continue; 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,}; SSubmitTbData tbData = {.suid = suid, .uid = 0, .sver = pTSchema->version, .flags = SUBMIT_REQ_AUTO_CREATE_TABLE,};
int32_t cid = taosArrayGetSize(pDataBlock->pDataBlock) + 1; 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; uint64_t groupId = pDataBlock->info.id.groupId;

35
source/dnode/vnode/src/tq/tqRead.c
View File

@ -596,10 +596,12 @@ static int32_t doSetVal(SColumnInfoData* pColumnInfoData, int32_t rowIndex, SCol
if (IS_STR_DATA_TYPE(pColVal->type)) { if (IS_STR_DATA_TYPE(pColVal->type)) {
char val[65535 + 2] = {0}; char val[65535 + 2] = {0};
if(COL_VAL_IS_VALUE(pColVal)){
if (pColVal->value.pData != NULL) { if (pColVal->value.pData != NULL) {
memcpy(varDataVal(val), pColVal->value.pData, pColVal->value.nData); memcpy(varDataVal(val), pColVal->value.pData, pColVal->value.nData);
}
varDataSetLen(val, 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 { } else {
colDataSetNULL(pColumnInfoData, rowIndex); colDataSetNULL(pColumnInfoData, rowIndex);
} }
@ -869,23 +871,9 @@ int32_t tqRetrieveTaosxBlock(STqReader* pReader, SArray* blocks, SArray* schemas
sourceIdx++; sourceIdx++;
} else if (pCol->cid == pColData->info.colId) { } else if (pCol->cid == pColData->info.colId) {
tColDataGetValue(pCol, i, &colVal); tColDataGetValue(pCol, i, &colVal);
if(doSetVal(pColData, curRow - lastRow, &colVal) != TDB_CODE_SUCCESS){
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; 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;
}
}
sourceIdx++; sourceIdx++;
targetIdx++; targetIdx++;
} }
@ -967,22 +955,9 @@ int32_t tqRetrieveTaosxBlock(STqReader* pReader, SArray* blocks, SArray* schemas
if (colVal.cid < pColData->info.colId) { if (colVal.cid < pColData->info.colId) {
sourceIdx++; sourceIdx++;
} else if (colVal.cid == pColData->info.colId) { } else if (colVal.cid == pColData->info.colId) {
if (IS_STR_DATA_TYPE(colVal.type)) { if(doSetVal(pColData, curRow - lastRow, &colVal) != TDB_CODE_SUCCESS){
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; 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;
}
}
sourceIdx++; sourceIdx++;
targetIdx++; targetIdx++;
} }

44
source/dnode/vnode/src/tq/tqSink.c
View File

@ -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 int32_t initCreateTableMsg(SVCreateTbReq* pCreateTableReq, uint64_t suid, const char* stbFullName, int32_t numOfTags);
static SArray* createDefaultTagColName(); static SArray* createDefaultTagColName();
static void setCreateTableMsgTableName(SVCreateTbReq* pCreateTableReq, SSDataBlock* pDataBlock, const char* stbFullName, 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, 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; int32_t totalRows = pDataBlock->info.rows;
SColumnInfoData* pStartTsCol = taosArrayGet(pDataBlock->pDataBlock, START_TS_COLUMN_INDEX); SColumnInfoData* pStartTsCol = taosArrayGet(pDataBlock->pDataBlock, START_TS_COLUMN_INDEX);
SColumnInfoData* pEndTsCol = taosArrayGet(pDataBlock->pDataBlock, END_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) { if (varTbName != NULL && varTbName != (void*)-1) {
name = taosMemoryCalloc(1, TSDB_TABLE_NAME_LEN); name = taosMemoryCalloc(1, TSDB_TABLE_NAME_LEN);
memcpy(name, varDataVal(varTbName), varDataLen(varTbName)); memcpy(name, varDataVal(varTbName), varDataLen(varTbName));
if(newSubTableRule &&
!isAutoTableName(name) &&
!alreadyAddGroupId(name) &&
groupId != 0) {
buildCtbNameAddGruopId(name, groupId);
}
} else if (stbFullName) { } else if (stbFullName) {
name = buildCtbNameByGroupId(stbFullName, groupId); name = buildCtbNameByGroupId(stbFullName, groupId);
} else { } else {
@ -173,9 +179,18 @@ SArray* createDefaultTagColName() {
} }
void setCreateTableMsgTableName(SVCreateTbReq* pCreateTableReq, SSDataBlock* pDataBlock, const char* stbFullName, void setCreateTableMsgTableName(SVCreateTbReq* pCreateTableReq, SSDataBlock* pDataBlock, const char* stbFullName,
int64_t gid) { int64_t gid, bool newSubTableRule) {
if (pDataBlock->info.parTbName[0]) { if (pDataBlock->info.parTbName[0]) {
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); pCreateTableReq->name = taosStrdup(pDataBlock->info.parTbName);
}
} else { } else {
pCreateTableReq->name = buildCtbNameByGroupId(stbFullName, gid); pCreateTableReq->name = buildCtbNameByGroupId(stbFullName, gid);
} }
@ -247,7 +262,7 @@ static int32_t doBuildAndSendCreateTableMsg(SVnode* pVnode, char* stbFullName, S
ASSERT(gid == *(int64_t*)pGpIdData); 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); taosArrayPush(reqs.pArray, pCreateTbReq);
tqDebug("s-task:%s build create table:%s msg complete", pTask->id.idStr, pCreateTbReq->name); 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) { int64_t suid) {
SBatchDeleteReq deleteReq = {.suid = suid, .deleteReqs = taosArrayInit(0, sizeof(SSingleDeleteReq))}; 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) { if (code != TSDB_CODE_SUCCESS) {
return code; 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, 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)); SVCreateTbReq* pCreateTbReq = taosMemoryCalloc(1, sizeof(SVCreateTbReq));
if (pCreateTbReq == NULL) { if (pCreateTbReq == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
@ -436,7 +452,7 @@ SVCreateTbReq* buildAutoCreateTableReq(const char* stbFullName, int64_t suid, in
pCreateTbReq->ctb.tagName = createDefaultTagColName(); pCreateTbReq->ctb.tagName = createDefaultTagColName();
// set table name // set table name
setCreateTableMsgTableName(pCreateTbReq, pDataBlock, stbFullName, pDataBlock->info.id.groupId); setCreateTableMsgTableName(pCreateTbReq, pDataBlock, stbFullName, pDataBlock->info.id.groupId, newSubTableRule);
return pCreateTbReq; 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, tqDebug("s-task:%s vgId:%d, gropuId:%" PRIu64 " datablock table name is null, set name:%s", id, vgId, groupId,
dstTableName); dstTableName);
} else { } else {
tstrncpy(dstTableName, pTableSinkInfo->name.data, pTableSinkInfo->name.len + 1);
if (pTableSinkInfo->uid != 0) { if (pTableSinkInfo->uid != 0) {
tqDebug("s-task:%s write %d rows into groupId:%" PRIu64 " dstTable:%s(uid:%" PRIu64 ")", id, numOfRows, groupId, tqDebug("s-task:%s write %d rows into groupId:%" PRIu64 " dstTable:%s(uid:%" PRIu64 ")", id, numOfRows, groupId,
dstTableName, pTableSinkInfo->uid); dstTableName, pTableSinkInfo->uid);
@ -649,10 +666,17 @@ int32_t setDstTableDataUid(SVnode* pVnode, SStreamTask* pTask, SSDataBlock* pDat
if (dstTableName[0] == 0) { if (dstTableName[0] == 0) {
memset(dstTableName, 0, TSDB_TABLE_NAME_LEN); memset(dstTableName, 0, TSDB_TABLE_NAME_LEN);
buildCtbNameByGroupIdImpl(stbFullName, groupId, dstTableName); 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); int32_t nameLen = strlen(dstTableName);
pTableSinkInfo = taosMemoryCalloc(1, sizeof(STableSinkInfo) + nameLen); pTableSinkInfo = taosMemoryCalloc(1, sizeof(STableSinkInfo) + nameLen + 1);
if (pTableSinkInfo == NULL) { if (pTableSinkInfo == NULL) {
return TSDB_CODE_OUT_OF_MEMORY; 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->flags = SUBMIT_REQ_AUTO_CREATE_TABLE;
pTableData->pCreateTbReq = 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); taosArrayDestroy(pTagArray);
if (pTableData->pCreateTbReq == NULL) { 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, int32_t setDstTableDataPayload(uint64_t suid, const STSchema *pTSchema, int32_t blockIndex, SSDataBlock* pDataBlock,

2
source/dnode/vnode/src/tsdb/tsdbRead2.c
View File

@ -4100,9 +4100,9 @@ void tsdbReaderClose2(STsdbReader* pReader) {
size_t numOfTables = tSimpleHashGetSize(pReader->status.pTableMap); size_t numOfTables = tSimpleHashGetSize(pReader->status.pTableMap);
if (pReader->status.pTableMap != NULL) { if (pReader->status.pTableMap != NULL) {
destroyAllBlockScanInfo(pReader->status.pTableMap); destroyAllBlockScanInfo(pReader->status.pTableMap);
clearBlockScanInfoBuf(&pReader->blockInfoBuf);
pReader->status.pTableMap = NULL; pReader->status.pTableMap = NULL;
} }
clearBlockScanInfoBuf(&pReader->blockInfoBuf);
if (pReader->pFileReader != NULL) { if (pReader->pFileReader != NULL) {
tsdbDataFileReaderClose(&pReader->pFileReader); tsdbDataFileReaderClose(&pReader->pFileReader);

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

@ -2743,6 +2743,14 @@ static EDealRes doCheckAggColCoexist(SNode** pNode, void* pContext) {
return DEAL_RES_CONTINUE; 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) { static int32_t resetSelectFuncNumWithoutDup(SSelectStmt* pSelect) {
if (pSelect->selectFuncNum <= 1) return TSDB_CODE_SUCCESS; if (pSelect->selectFuncNum <= 1) return TSDB_CODE_SUCCESS;
pSelect->selectFuncNum = 0; pSelect->selectFuncNum = 0;
@ -4416,9 +4424,6 @@ static int32_t translateWhere(STranslateContext* pCxt, SSelectStmt* pSelect) {
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = getQueryTimeRange(pCxt, pSelect->pWhere, &pSelect->timeRange); 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) { if (pSelect->pWhere != NULL) {
setTableVgroupsFromEqualTbnameCond(pCxt, pSelect); setTableVgroupsFromEqualTbnameCond(pCxt, pSelect);
} }
@ -4606,6 +4611,9 @@ static int32_t translateSelectFrom(STranslateContext* pCxt, SSelectStmt* pSelect
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = translateOrderBy(pCxt, pSelect); code = translateOrderBy(pCxt, pSelect);
} }
if (TSDB_CODE_SUCCESS == code) {
code = checkIsEmptyResult(pCxt, pSelect);
}
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
resetSelectFuncNumWithoutDup(pSelect); resetSelectFuncNumWithoutDup(pSelect);
code = checkAggColCoexist(pCxt, pSelect); code = checkAggColCoexist(pCxt, pSelect);

18
source/libs/scheduler/inc/schInt.h
View File

@ -70,6 +70,7 @@ typedef struct SSchDebug {
typedef struct SSchTrans { typedef struct SSchTrans {
void *pTrans; void *pTrans;
void *pHandle; void *pHandle;
int64_t pHandleId;
} SSchTrans; } SSchTrans;
typedef struct SSchHbTrans { typedef struct SSchHbTrans {
@ -334,7 +335,8 @@ extern SSchedulerMgmt schMgmt;
(!SCH_IS_DATA_BIND_QRY_TASK(_task))) (!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_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_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(task) atomic_load_8(&(task)->status)
@ -380,7 +382,9 @@ extern SSchedulerMgmt schMgmt;
#define SCH_JOB_NEED_WAIT(_job) (!SCH_IS_QUERY_JOB(_job)) #define SCH_JOB_NEED_WAIT(_job) (!SCH_IS_QUERY_JOB(_job))
#define SCH_JOB_NEED_DROP(_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_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) \ #define SCH_REDIRECT_MSGTYPE(_msgType) \
((_msgType) == TDMT_SCH_LINK_BROKEN || (_msgType) == TDMT_SCH_QUERY || (_msgType) == TDMT_SCH_MERGE_QUERY || \ ((_msgType) == TDMT_SCH_LINK_BROKEN || (_msgType) == TDMT_SCH_QUERY || (_msgType) == TDMT_SCH_MERGE_QUERY || \
(_msgType) == TDMT_SCH_FETCH || (_msgType) == TDMT_SCH_MERGE_FETCH) (_msgType) == TDMT_SCH_FETCH || (_msgType) == TDMT_SCH_MERGE_FETCH)
@ -388,18 +392,17 @@ extern SSchedulerMgmt schMgmt;
(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) \ #define SCH_TOP_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_TASK_RETRY_NETWORK_ERR(_task, _code) \ #define SCH_TASK_RETRY_NETWORK_ERR(_task, _code) (SCH_NETWORK_ERR(_code) && (_task)->redirectCtx.inRedirect)
(SCH_NETWORK_ERR(_code) && (_task)->redirectCtx.inRedirect)
#define SCH_JOB_NEED_RETRY(_job, _task, _msgType, _code) \ #define SCH_JOB_NEED_RETRY(_job, _task, _msgType, _code) \
(SCH_REDIRECT_MSGTYPE(_msgType) && SCH_TOP_LEVEL_NETWORK_ERR(_job, _task, _code)) (SCH_REDIRECT_MSGTYPE(_msgType) && SCH_TOP_LEVEL_NETWORK_ERR(_job, _task, _code))
#define SCH_TASKSET_NEED_RETRY(_job, _task, _msgType, _code) \ #define SCH_TASKSET_NEED_RETRY(_job, _task, _msgType, _code) \
(SCH_REDIRECT_MSGTYPE(_msgType) && \ (SCH_REDIRECT_MSGTYPE(_msgType) && \
(NEED_SCHEDULER_REDIRECT_ERROR(_code) || SCH_LOW_LEVEL_NETWORK_ERR((_job), (_task), (_code)) || SCH_TASK_RETRY_NETWORK_ERR((_task), (_code)))) (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) \ #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_IS_LEVEL_UNFINISHED(_level) ((_level)->taskLaunchedNum < (_level)->taskNum)
#define SCH_GET_CUR_EP(_addr) (&(_addr)->epSet.eps[(_addr)->epSet.inUse]) #define SCH_GET_CUR_EP(_addr) (&(_addr)->epSet.eps[(_addr)->epSet.inUse])
#define SCH_SWITCH_EPSET(_addr) ((_addr)->epSet.inUse = ((_addr)->epSet.inUse + 1) % (_addr)->epSet.numOfEps) #define SCH_SWITCH_EPSET(_addr) ((_addr)->epSet.inUse = ((_addr)->epSet.inUse + 1) % (_addr)->epSet.numOfEps)
@ -522,7 +525,8 @@ extern SSchedulerMgmt schMgmt;
} \ } \
} while (0) } while (0)
#define SCH_RESET_JOB_LEVEL_IDX(_job) do { \ #define SCH_RESET_JOB_LEVEL_IDX(_job) \
do { \
(_job)->levelIdx = (_job)->levelNum - 1; \ (_job)->levelIdx = (_job)->levelNum - 1; \
SCH_JOB_DLOG("set job levelIdx to %d", (_job)->levelIdx); \ SCH_JOB_DLOG("set job levelIdx to %d", (_job)->levelIdx); \
} while (0) } while (0)

3
source/libs/scheduler/src/schRemote.c
View File

@ -505,7 +505,7 @@ int32_t schHandleLinkBrokenCallback(void *param, SDataBuf *pMsg, int32_t code) {
taosMemoryFree(pMsg->pEpSet); taosMemoryFree(pMsg->pEpSet);
SSchHbCallbackParam *hbParam = (SSchHbCallbackParam *)param; 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(schUpdateHbConnection(&hbParam->nodeEpId, &trans));
SCH_ERR_RET(schBuildAndSendHbMsg(&hbParam->nodeEpId, NULL)); SCH_ERR_RET(schBuildAndSendHbMsg(&hbParam->nodeEpId, NULL));
@ -537,6 +537,7 @@ int32_t schHandleHbCallback(void *param, SDataBuf *pMsg, int32_t code) {
SSchTrans trans = {0}; SSchTrans trans = {0};
trans.pTrans = pParam->pTrans; trans.pTrans = pParam->pTrans;
trans.pHandle = pMsg->handle; trans.pHandle = pMsg->handle;
trans.pHandleId = pMsg->handleRefId;
SCH_ERR_JRET(schUpdateHbConnection(&rsp.epId, &trans)); SCH_ERR_JRET(schUpdateHbConnection(&rsp.epId, &trans));
SCH_ERR_JRET(schProcessOnTaskStatusRsp(&rsp.epId, rsp.taskStatus)); SCH_ERR_JRET(schProcessOnTaskStatusRsp(&rsp.epId, rsp.taskStatus));

5
source/libs/scheduler/src/schUtil.c
View File

@ -73,7 +73,7 @@ char *schGetOpStr(SCH_OP_TYPE type) {
} }
void schFreeHbTrans(SSchHbTrans *pTrans) { void schFreeHbTrans(SSchHbTrans *pTrans) {
rpcReleaseHandle(pTrans->trans.pHandle, TAOS_CONN_CLIENT); rpcReleaseHandle((void *)pTrans->trans.pHandleId, TAOS_CONN_CLIENT);
schFreeRpcCtx(&pTrans->rpcCtx); schFreeRpcCtx(&pTrans->rpcCtx);
} }
@ -234,7 +234,8 @@ int32_t schUpdateHbConnection(SQueryNodeEpId *epId, SSchTrans *trans) {
hb = taosHashGet(schMgmt.hbConnections, epId, sizeof(SQueryNodeEpId)); hb = taosHashGet(schMgmt.hbConnections, epId, sizeof(SQueryNodeEpId));
if (NULL == hb) { if (NULL == hb) {
SCH_UNLOCK(SCH_READ, &schMgmt.hbLock); 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); SCH_ERR_RET(TSDB_CODE_APP_ERROR);
} }

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

@ -547,25 +547,21 @@ int32_t streamSearchAndAddBlock(SStreamTask* pTask, SStreamDispatchReq* pReqs, S
memcpy(pDataBlock->info.parTbName, pBln->parTbName, strlen(pBln->parTbName)); memcpy(pDataBlock->info.parTbName, pBln->parTbName, strlen(pBln->parTbName));
} }
} else { } else {
char* ctbName = taosMemoryCalloc(1, TSDB_TABLE_FNAME_LEN); char ctbName[TSDB_TABLE_FNAME_LEN] = {0};
if (ctbName == NULL) {
return -1;
}
if (pDataBlock->info.parTbName[0]) { if (pDataBlock->info.parTbName[0]) {
snprintf(ctbName, TSDB_TABLE_NAME_LEN, "%s.%s", pTask->outputInfo.shuffleDispatcher.dbInfo.db, if(pTask->ver >= SSTREAM_TASK_SUBTABLE_CHANGED_VER &&
pDataBlock->info.parTbName); !isAutoTableName(pDataBlock->info.parTbName) &&
!alreadyAddGroupId(pDataBlock->info.parTbName) &&
groupId != 0){
buildCtbNameAddGruopId(pDataBlock->info.parTbName, groupId);
}
} else { } else {
buildCtbNameByGroupIdImpl(pTask->outputInfo.shuffleDispatcher.stbFullName, groupId, pDataBlock->info.parTbName); 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));*/ /*uint32_t hashValue = MurmurHash3_32(ctbName, strlen(ctbName));*/
SUseDbRsp* pDbInfo = &pTask->outputInfo.shuffleDispatcher.dbInfo; SUseDbRsp* pDbInfo = &pTask->outputInfo.shuffleDispatcher.dbInfo;
hashValue = hashValue = taosGetTbHashVal(ctbName, strlen(ctbName), pDbInfo->hashMethod, pDbInfo->hashPrefix, pDbInfo->hashSuffix);
taosGetTbHashVal(ctbName, strlen(ctbName), pDbInfo->hashMethod, pDbInfo->hashPrefix, pDbInfo->hashSuffix);
taosMemoryFree(ctbName);
SBlockName bln = {0}; SBlockName bln = {0};
bln.hashValue = hashValue; bln.hashValue = hashValue;
memcpy(bln.parTbName, pDataBlock->info.parTbName, strlen(pDataBlock->info.parTbName)); memcpy(bln.parTbName, pDataBlock->info.parTbName, strlen(pDataBlock->info.parTbName));

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

@ -180,10 +180,8 @@ int32_t streamMetaCheckBackendCompatible(SStreamMeta* pMeta) {
} }
if (info.msgVer <= SSTREAM_TASK_INCOMPATIBLE_VER) { if (info.msgVer <= SSTREAM_TASK_INCOMPATIBLE_VER) {
ret = STREAM_STATA_NO_COMPATIBLE; 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; ret = STREAM_STATA_NEED_CONVERT;
} else if (info.msgVer == SSTREAM_TASK_VER) {
ret = STREAM_STATA_COMPATIBLE;
} }
tDecoderClear(&decoder); tDecoderClear(&decoder);
break; break;

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

@ -213,7 +213,7 @@ int32_t tDecodeStreamTask(SDecoder* pDecoder, SStreamTask* pTask) {
if (tStartDecode(pDecoder) < 0) return -1; if (tStartDecode(pDecoder) < 0) return -1;
if (tDecodeI64(pDecoder, &pTask->ver) < 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 (tDecodeI64(pDecoder, &pTask->id.streamId) < 0) return -1;
if (tDecodeI32(pDecoder, &pTask->id.taskId) < 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 (tStartDecode(pDecoder) < 0) return -1;
if (tDecodeI64(pDecoder, &pChkpInfo->msgVer) < 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 (tDecodeI64(pDecoder, &skip64) < 0) return -1;
if (tDecodeI32(pDecoder, &skip32) < 0) return -1; if (tDecodeI32(pDecoder, &skip32) < 0) return -1;
@ -328,7 +328,7 @@ int32_t tDecodeStreamTaskId(SDecoder* pDecoder, STaskId* pTaskId) {
int64_t ver; int64_t ver;
if (tStartDecode(pDecoder) < 0) return -1; if (tStartDecode(pDecoder) < 0) return -1;
if (tDecodeI64(pDecoder, &ver) < 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; if (tDecodeI64(pDecoder, &pTaskId->streamId) < 0) return -1;

1
source/libs/transport/src/transCli.c
View File

@ -414,6 +414,7 @@ void cliHandleResp(SCliConn* conn) {
// buf's mem alread translated to transMsg.pCont // buf's mem alread translated to transMsg.pCont
if (!CONN_NO_PERSIST_BY_APP(conn)) { if (!CONN_NO_PERSIST_BY_APP(conn)) {
transMsg.info.handle = (void*)conn->refId; 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); tDebug("%s conn %p ref by app", CONN_GET_INST_LABEL(conn), conn);
} }

2
tests/army/community/query/query_basic.json
View File

@ -7,7 +7,7 @@
"password": "taosdata", "password": "taosdata",
"connection_pool_size": 8, "connection_pool_size": 8,
"num_of_records_per_req": 4000, "num_of_records_per_req": 4000,
"prepared_rand": 1000, "prepared_rand": 10000,
"thread_count": 3, "thread_count": 3,
"create_table_thread_count": 1, "create_table_thread_count": 1,
"confirm_parameter_prompt": "no", "confirm_parameter_prompt": "no",

24
tests/army/community/query/query_basic.py
View File

@ -41,6 +41,7 @@ class TDTestCase(TBase):
etool.benchMark(json=jfile) etool.benchMark(json=jfile)
tdSql.execute(f"use {self.db}") tdSql.execute(f"use {self.db}")
tdSql.execute("select database();")
# set insert data information # set insert data information
self.childtable_count = 6 self.childtable_count = 6
self.insert_rows = 100000 self.insert_rows = 100000
@ -50,10 +51,29 @@ class TDTestCase(TBase):
def doQuery(self): def doQuery(self):
tdLog.info(f"do query.") tdLog.info(f"do query.")
# top bottom # __group_key
sql = f"select top(uti, 5) from {self.stb} " sql = f"select count(*),_group_key(uti),uti from {self.stb} partition by uti;"
tdSql.execute(sql) 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 # run
def run(self): def run(self):

41
tests/army/frame/caseBase.py
View File

@ -16,6 +16,7 @@ import os
import time import time
import datetime import datetime
import random import random
import copy
from frame.log import * from frame.log import *
from frame.sql import * from frame.sql import *
@ -183,6 +184,46 @@ class TBase:
sql = f"select {col} from {self.stb} order by _c0 asc limit 1" sql = f"select {col} from {self.stb} order by _c0 asc limit 1"
tdSql.checkFirstValue(sql, expect) 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 # get db information
# #

4
tests/parallel_test/cases.task
View File

@ -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 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 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/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/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 ,,y,system-test,./pytest.sh python3 ./test.py -f 3-enterprise/restore/restoreVnode.py -N 5 -M 3 -i False

17
tests/pytest/util/common.py
View File

@ -1845,6 +1845,23 @@ class TDCom:
if i == 1: if i == 1:
self.record_history_ts = ts_value 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): def is_json(msg):
if isinstance(msg, str): if isinstance(msg, str):
try: try:

2
tests/pytest/util/sql.py
View File

@ -88,7 +88,7 @@ class TDSql:
self.affectedRows = self.cursor.execute(sql) self.affectedRows = self.cursor.execute(sql)
return self.affectedRows return self.affectedRows
except Exception as e: 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: if i == queryTimes:
caller = inspect.getframeinfo(inspect.stack()[1][0]) caller = inspect.getframeinfo(inspect.stack()[1][0])
args = (caller.filename, caller.lineno, sql, repr(e)) args = (caller.filename, caller.lineno, sql, repr(e))

36
tests/script/tsim/stream/udTableAndTag0.sim
View File

@ -48,16 +48,6 @@ if $rows != 2 then
goto loop0 goto loop0
endi 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 $loop_count = 0
loop1: loop1:
@ -264,17 +254,6 @@ if $rows != 2 then
goto loop6 goto loop6
endi 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 ===== step5
print ===== tag name + table name print ===== tag name + table name
@ -312,21 +291,6 @@ if $rows != 3 then
goto loop7 goto loop7
endi 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 $loop_count = 0
loop8: loop8:

35
tests/script/tsim/stream/udTableAndTag1.sim
View File

@ -47,16 +47,6 @@ if $rows != 2 then
goto loop0 goto loop0
endi 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 $loop_count = 0
loop1: loop1:
@ -264,16 +254,6 @@ if $rows != 2 then
goto loop6 goto loop6
endi 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 ===== step5
print ===== tag name + table name print ===== tag name + table name
@ -311,21 +291,6 @@ if $rows != 3 then
goto loop7 goto loop7
endi 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 $loop_count = 0
loop8: loop8:

5
tests/script/tsim/stream/udTableAndTag2.sim
View File

@ -415,11 +415,6 @@ if $data00 != aaa then
goto loop9 goto loop9
endi endi
if $data10 != aaa-1 then
print =====data00=$data00
goto loop9
endi
$loop_count = 0 $loop_count = 0
loop10: loop10:

30
tests/system-test/2-query/csum.py
View File

@ -162,6 +162,16 @@ class TDTestCase:
case6 = {"col": "c9"} case6 = {"col": "c9"}
self.checkcsum(**case6) 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~8: nested query
case7 = {"table_expr": "(select ts,c1 from db.stb1 order by ts, tbname )"} case7 = {"table_expr": "(select ts,c1 from db.stb1 order by ts, tbname )"}
self.checkcsum(**case7) self.checkcsum(**case7)
@ -317,14 +327,14 @@ class TDTestCase:
f"insert into t{i} values (" 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"{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"'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( tdSql.execute(
f"insert into t{i} values (" 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"{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"'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( tdSql.execute(
f"insert into tt{i} values ( {basetime-(j+1) * 10 + i * msec_per_min}, {random.randint(1, 200)} )" 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( tdSql.execute(
"create stable db.stb1 (\ "create stable db.stb1 (\
ts timestamp, c1 int, c2 float, c3 timestamp, c4 binary(16), c5 double, c6 bool, \ 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)" tags(st1 int)"
) )
tdSql.execute( tdSql.execute(
@ -373,10 +383,10 @@ class TDTestCase:
tdLog.printNoPrefix("######## insert data in the range near the max(bigint/double):") tdLog.printNoPrefix("######## insert data in the range near the max(bigint/double):")
self.csum_test_table(tbnum) self.csum_test_table(tbnum)
tdSql.execute(f"insert into db.t1(ts, c1,c2,c5,c7) values " 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})") 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) values " 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})") 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_current_query()
self.csum_error_query() self.csum_error_query()
@ -460,7 +470,7 @@ class TDTestCase:
tdSql.checkRows(40) tdSql.checkRows(40)
# bug need fix # 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.checkRows(40)
tdSql.query("select tbname , csum(st1) from db.stb1 partition by tbname") tdSql.query("select tbname , csum(st1) from db.stb1 partition by tbname")
tdSql.checkRows(70) tdSql.checkRows(70)
@ -468,7 +478,7 @@ class TDTestCase:
tdSql.checkRows(7) tdSql.checkRows(7)
# partition by tags # 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.checkRows(40)
tdSql.query("select csum(c1) from db.stb1 partition by st1") tdSql.query("select csum(c1) from db.stb1 partition by st1")
tdSql.checkRows(40) tdSql.checkRows(40)

91
tests/system-test/2-query/nestedQueryInterval.py
View File

@ -1310,7 +1310,7 @@ class TDTestCase:
ts = ts + 20 ts = ts + 20
tdSql.query(f"select tbname,count(*) from nested.stable_null_childtable group by tbname order by tbname;") tdSql.query(f"select tbname,count(*) from nested.stable_null_childtable group by tbname order by tbname;")
tdSql.checkRows(1) tdSql.checkRows(1)
tdSql.checkData(0, 1, 52); tdSql.checkData(0, 1, 52)
#stables #stables
@ -1347,16 +1347,16 @@ class TDTestCase:
tdSql.query(f"select tbname,count(*) from nested.stable_1 group by tbname order by tbname;") tdSql.query(f"select tbname,count(*) from nested.stable_1 group by tbname order by tbname;")
tdSql.checkRows(6) tdSql.checkRows(6)
tdSql.checkData(0, 1, 162); tdSql.checkData(0, 1, 162)
tdSql.checkData(1, 1, 200); tdSql.checkData(1, 1, 200)
tdSql.query(f"select tbname,count(*) from nested.stable_null_data group by tbname order by tbname;") tdSql.query(f"select tbname,count(*) from nested.stable_null_data group by tbname order by tbname;")
tdSql.checkRows(1) 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.query(f"select tbname,count(*) from nested.stable_null_childtable group by tbname order by tbname;")
tdSql.checkRows(1) tdSql.checkRows(1)
tdSql.checkData(0, 1, 62); tdSql.checkData(0, 1, 62)
#test special character #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.query(f"select tbname,count(*) from nested.stable_1 group by tbname order by tbname;")
tdSql.checkRows(7) tdSql.checkRows(7)
tdSql.checkData(0, 1, 1); tdSql.checkData(0, 1, 1)
tdSql.checkData(1, 1, 162); tdSql.checkData(1, 1, 162)
tdSql.checkData(2, 1, 200); tdSql.checkData(2, 1, 200)
tdSql.query(f"select count(*) from nested.stable_1 where tbname ='!@!@$$^$' ;") 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.query(f"select tbname,count(*) from nested.stable_null_data group by tbname order by tbname;")
tdSql.checkRows(2) tdSql.checkRows(2)
tdSql.checkData(0, 1, 1); tdSql.checkData(0, 1, 1)
tdSql.checkData(1, 1, 62); tdSql.checkData(1, 1, 62)
tdSql.query(f"select count(*) from nested.stable_null_data where tbname ='%^$^&^&' ;") 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.query(f"select tbname,count(*) from nested.stable_null_childtable group by tbname order by tbname;")
tdSql.checkRows(2) tdSql.checkRows(2)
tdSql.checkData(0, 1, 1); tdSql.checkData(0, 1, 1)
tdSql.checkData(1, 1, 62); tdSql.checkData(1, 1, 62)
tdSql.query(f"select count(*) from nested.stable_null_childtable where tbname ='$^%$%^&' ;") tdSql.query(f"select count(*) from nested.stable_null_childtable where tbname ='$^%$%^&' ;")
tdSql.checkData(0, 0, 1); tdSql.checkData(0, 0, 1)
#test csv #test csv
sql1 = "select tbname,ts,q_int,q_binary from nested.stable_1 >>'%s/stable_1.csv';" %self.testcasePath 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.query(f"select tbname,count(*) from nested.stable_1 group by tbname order by tbname;")
tdSql.checkRows(7) tdSql.checkRows(7)
tdSql.checkData(0, 1, 1); tdSql.checkData(0, 1, 1)
tdSql.checkData(1, 1, 162); tdSql.checkData(1, 1, 162)
tdSql.checkData(2, 1, 200); tdSql.checkData(2, 1, 200)
tdSql.query(f"select count(*) from nested.stable_1 where tbname ='!@!@$$^$' ;") 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.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.query(f"select tbname,count(*) from nested.stable_null_data group by tbname order by tbname;")
tdSql.checkRows(2) tdSql.checkRows(2)
tdSql.checkData(0, 1, 1); tdSql.checkData(0, 1, 1)
tdSql.checkData(1, 1, 62); tdSql.checkData(1, 1, 62)
tdSql.query(f"select count(*) from nested.stable_null_data where tbname ='%^$^&^&' ;") 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.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.query(f"select tbname,count(*) from nested.stable_null_childtable group by tbname order by tbname;")
tdSql.checkRows(2) tdSql.checkRows(2)
tdSql.checkData(0, 1, 1); tdSql.checkData(0, 1, 1)
tdSql.checkData(1, 1, 62); tdSql.checkData(1, 1, 62)
tdSql.query(f"select count(*) from nested.stable_null_childtable where tbname ='$^%$%^&' ;") 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.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;") 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.query(f"select tbname,count(*) from nested.stable_1 group by tbname order by tbname;")
tdSql.checkRows(7) tdSql.checkRows(7)
tdSql.checkData(0, 1, 1); tdSql.checkData(0, 1, 1)
tdSql.checkData(1, 1, 162); tdSql.checkData(1, 1, 162)
tdSql.checkData(2, 1, 200); tdSql.checkData(2, 1, 200)
tdSql.query(f"select count(*) from nested.stable_1 where tbname ='!@!@$$^$' ;") 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.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.query(f"select tbname,count(*) from nested.stable_null_data group by tbname order by tbname;")
tdSql.checkRows(2) tdSql.checkRows(2)
tdSql.checkData(0, 1, 1); tdSql.checkData(0, 1, 1)
tdSql.checkData(1, 1, 62); tdSql.checkData(1, 1, 62)
tdSql.query(f"select count(*) from nested.stable_null_data where tbname ='%^$^&^&' ;") 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.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.query(f"select tbname,count(*) from nested.stable_null_childtable group by tbname order by tbname;")
tdSql.checkRows(2) tdSql.checkRows(2)
tdSql.checkData(0, 1, 1); tdSql.checkData(0, 1, 1)
tdSql.checkData(1, 1, 62); tdSql.checkData(1, 1, 62)
tdSql.query(f"select count(*) from nested.stable_null_childtable where tbname ='$^%$%^&' ;") 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.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 #test stable

4
tests/system-test/2-query/sum.py
View File

@ -245,6 +245,10 @@ class TDTestCase:
# stddev # stddev
tdSql.checkData(0, 5, 0, True) 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 i += 1

68
tests/system-test/2-query/td-28068.py Normal file
View File

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

218
tests/system-test/7-tmq/tmqVnodeSplit-stb-false.py Normal file
View File

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

222
tests/system-test/7-tmq/tmqVnodeSplit-stb-select-false.py Normal file
View File

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

4
tests/system-test/7-tmq/tmqVnodeSplit-stb-select.py
View File

@ -190,6 +190,8 @@ class TDTestCase:
pInsertThread1.join() pInsertThread1.join()
expectRows = 1 expectRows = 1
resultList = tmqCom.selectConsumeResult(expectRows) resultList = tmqCom.selectConsumeResult(expectRows)
if expectrowcnt / 2 > resultList[0]: if expectrowcnt / 2 > resultList[0]:
@ -209,8 +211,6 @@ class TDTestCase:
def run(self): def run(self):
self.prepareTestEnv() self.prepareTestEnv()
self.tmqCase1(True) self.tmqCase1(True)
self.prepareTestEnv()
self.tmqCase1(False)
def stop(self): def stop(self):
tdSql.close() tdSql.close()

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

@ -207,8 +207,6 @@ class TDTestCase:
def run(self): def run(self):
self.prepareTestEnv() self.prepareTestEnv()
self.tmqCase1(True) self.tmqCase1(True)
self.prepareTestEnv()
self.tmqCase1(False)
def stop(self): def stop(self):
tdSql.close() tdSql.close()

28
tests/system-test/8-stream/at_once_interval.py
View File

@ -106,14 +106,18 @@ class TDTestCase:
ptn_counter = 0 ptn_counter = 0
for c1_value in tdSql.queryResult: for c1_value in tdSql.queryResult:
if partition == "c1": 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: 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)": elif partition == "abs(c1)":
abs_c1_value = abs(c1_value[1]) 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: 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 ptn_counter += 1
tdSql.checkEqual(tdSql.queryResult[0][0] > 0, True) tdSql.checkEqual(tdSql.queryResult[0][0] > 0, True)
@ -121,14 +125,18 @@ class TDTestCase:
ptn_counter = 0 ptn_counter = 0
for c1_value in tdSql.queryResult: for c1_value in tdSql.queryResult:
if partition == "c1": 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: 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)": elif partition == "abs(c1)":
abs_c1_value = abs(c1_value[1]) 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: 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 ptn_counter += 1
tdSql.checkEqual(tdSql.queryResult[0][0] > 0, True) 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=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=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") 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 ["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 ["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)
self.at_once_interval(interval=random.randint(10, 15), partition="tbname", fill_value=fill_value, delete=True) self.at_once_interval(interval=random.randint(10, 15), partition="tbname", fill_value=fill_value, delete=True)

12
tests/system-test/8-stream/at_once_interval_ext.py
View File

@ -153,12 +153,15 @@ class TDTestCase:
ptn_counter = 0 ptn_counter = 0
for c1_value in tdSql.queryResult: for c1_value in tdSql.queryResult:
if partition == "c1": 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)": elif partition == "abs(c1)":
abs_c1_value = abs(c1_value[1]) 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: 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 ptn_counter += 1
else: else:
tdSql.query(f'select cast(cast(cast({c1_value[1]} as int unsigned) as bigint) as varchar(100))') 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": if subtable == "constant":
return return
else: 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) tdSql.checkEqual(tdSql.queryResult[0][0] > 0, True)
def run(self): def run(self):

28
tests/system-test/8-stream/at_once_session.py
View File

@ -171,35 +171,43 @@ class TDTestCase:
for c1_value in tdSql.queryResult: for c1_value in tdSql.queryResult:
if c1_value[1] is not None: if c1_value[1] is not None:
if partition == "c1": 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)": elif partition == "abs(c1)":
if subtable: if subtable:
abs_c1_value = abs(c1_value[1]) 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: 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: 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 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}') tdSql.query(f'select * from {self.tb_name}')
ptn_counter = 0 ptn_counter = 0
for c1_value in tdSql.queryResult: for c1_value in tdSql.queryResult:
if c1_value[1] is not None: if c1_value[1] is not None:
if partition == "c1": 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)": elif partition == "abs(c1)":
if subtable: if subtable:
abs_c1_value = abs(c1_value[1]) 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: 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: 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 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)

25
tests/system-test/8-stream/at_once_state_window.py
View File

@ -94,15 +94,19 @@ class TDTestCase:
for c1_value in tdSql.queryResult: for c1_value in tdSql.queryResult:
if partition == "c1": if partition == "c1":
if subtable: 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: 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 return
elif partition == "abs(c1)": elif partition == "abs(c1)":
abs_c1_value = abs(c1_value[1]) 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: 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 ptn_counter += 1
tdSql.checkEqual(tdSql.queryResult[0][0] > 0, True) tdSql.checkEqual(tdSql.queryResult[0][0] > 0, True)
@ -111,17 +115,20 @@ class TDTestCase:
for c1_value in tdSql.queryResult: for c1_value in tdSql.queryResult:
if partition == "c1": if partition == "c1":
if subtable: 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: 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 return
elif partition == "abs(c1)": elif partition == "abs(c1)":
abs_c1_value = abs(c1_value[1]) 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: 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 ptn_counter += 1
tdSql.checkEqual(tdSql.queryResult[0][0] > 0, True) tdSql.checkEqual(tdSql.queryResult[0][0] > 0, True)

24
tests/system-test/8-stream/pause_resume_test.py
View File

@ -117,14 +117,18 @@ class TDTestCase:
ptn_counter = 0 ptn_counter = 0
for c1_value in tdSql.queryResult: for c1_value in tdSql.queryResult:
if partition == "c1": 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: 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)": elif partition == "abs(c1)":
abs_c1_value = abs(c1_value[1]) 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: 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 ptn_counter += 1
tdSql.checkEqual(tdSql.queryResult[0][0] > 0, True) tdSql.checkEqual(tdSql.queryResult[0][0] > 0, True)
@ -132,14 +136,18 @@ class TDTestCase:
ptn_counter = 0 ptn_counter = 0
for c1_value in tdSql.queryResult: for c1_value in tdSql.queryResult:
if partition == "c1": 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: 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)": elif partition == "abs(c1)":
abs_c1_value = abs(c1_value[1]) 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: 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 ptn_counter += 1
tdSql.checkEqual(tdSql.queryResult[0][0] > 0, True) tdSql.checkEqual(tdSql.queryResult[0][0] > 0, True)

18
tests/system-test/8-stream/window_close_interval.py
View File

@ -146,12 +146,15 @@ class TDTestCase:
ptn_counter = 0 ptn_counter = 0
for c1_value in tdSql.queryResult: for c1_value in tdSql.queryResult:
if partition == "c1": 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)": elif partition == "abs(c1)":
abs_c1_value = abs(c1_value[1]) 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: 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 ptn_counter += 1
tdSql.checkEqual(tdSql.queryResult[0][0] , self.tdCom.range_count) tdSql.checkEqual(tdSql.queryResult[0][0] , self.tdCom.range_count)
@ -161,12 +164,15 @@ class TDTestCase:
ptn_counter = 0 ptn_counter = 0
for c1_value in tdSql.queryResult: for c1_value in tdSql.queryResult:
if partition == "c1": 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)": elif partition == "abs(c1)":
abs_c1_value = abs(c1_value[1]) 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: 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 ptn_counter += 1
tdSql.checkEqual(tdSql.queryResult[0][0] > 0, True) tdSql.checkEqual(tdSql.queryResult[0][0] > 0, True)