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

Merge remote-tracking branch 'origin/3.0' into fix/mnode

This commit is contained in:
Shengliang Guan 2022-06-06 17:51:02 +08:00
parent 362f37b5ed 75af9c0089
commit 033ebf49df
33 changed files with 739 additions and 370 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

43
include/common/tmsg.h
View File

@ -1126,9 +1126,9 @@ typedef struct {
STableMetaRsp* pMeta; STableMetaRsp* pMeta;
} SMAlterStbRsp; } SMAlterStbRsp;
int32_t tEncodeSMAlterStbRsp(SEncoder *pEncoder, const SMAlterStbRsp *pRsp); int32_t tEncodeSMAlterStbRsp(SEncoder* pEncoder, const SMAlterStbRsp* pRsp);
int32_t tDecodeSMAlterStbRsp(SDecoder *pDecoder, SMAlterStbRsp *pRsp); int32_t tDecodeSMAlterStbRsp(SDecoder* pDecoder, SMAlterStbRsp* pRsp);
void tFreeSMAlterStbRsp(SMAlterStbRsp* pRsp); void tFreeSMAlterStbRsp(SMAlterStbRsp* pRsp);
int32_t tSerializeSTableMetaRsp(void* buf, int32_t bufLen, STableMetaRsp* pRsp); int32_t tSerializeSTableMetaRsp(void* buf, int32_t bufLen, STableMetaRsp* pRsp);
int32_t tDeserializeSTableMetaRsp(void* buf, int32_t bufLen, STableMetaRsp* pRsp); int32_t tDeserializeSTableMetaRsp(void* buf, int32_t bufLen, STableMetaRsp* pRsp);
@ -1982,6 +1982,7 @@ typedef struct {
typedef struct { typedef struct {
SClientHbKey connKey; SClientHbKey connKey;
int64_t clusterId;
SQueryHbReqBasic* query; SQueryHbReqBasic* query;
SHashObj* info; // hash<Skv.key, Skv> SHashObj* info; // hash<Skv.key, Skv>
} SClientHbReq; } SClientHbReq;
@ -2302,23 +2303,23 @@ typedef struct {
} SVgEpSet; } SVgEpSet;
typedef struct { typedef struct {
int8_t version; // for compatibility(default 0) int8_t version; // for compatibility(default 0)
int8_t intervalUnit; // MACRO: TIME_UNIT_XXX int8_t intervalUnit; // MACRO: TIME_UNIT_XXX
int8_t slidingUnit; // MACRO: TIME_UNIT_XXX int8_t slidingUnit; // MACRO: TIME_UNIT_XXX
int8_t timezoneInt; // sma data expired if timezone changes. int8_t timezoneInt; // sma data expired if timezone changes.
int32_t dstVgId; int32_t dstVgId;
char indexName[TSDB_INDEX_NAME_LEN]; char indexName[TSDB_INDEX_NAME_LEN];
int32_t exprLen; int32_t exprLen;
int32_t tagsFilterLen; int32_t tagsFilterLen;
int32_t numOfVgroups; int32_t numOfVgroups;
int64_t indexUid; int64_t indexUid;
tb_uid_t tableUid; // super/child/common table uid tb_uid_t tableUid; // super/child/common table uid
int64_t interval; int64_t interval;
int64_t offset; // use unit by precision of DB int64_t offset; // use unit by precision of DB
int64_t sliding; int64_t sliding;
char* expr; // sma expression char* expr; // sma expression
char* tagsFilter; char* tagsFilter;
SVgEpSet vgEpSet[]; SVgEpSet* pVgEpSet;
} STSma; // Time-range-wise SMA } STSma; // Time-range-wise SMA
typedef STSma SVCreateTSmaReq; typedef STSma SVCreateTSmaReq;
@ -2404,7 +2405,7 @@ static int32_t tDecodeTSmaWrapper(SDecoder* pDecoder, STSmaWrapper* pReq) {
} }
typedef struct { typedef struct {
int64_t indexUid; int64_t indexUid;
STimeWindow queryWindow; STimeWindow queryWindow;
} SVGetTsmaExpWndsReq; } SVGetTsmaExpWndsReq;

3
include/libs/catalog/catalog.h
View File

@ -68,6 +68,7 @@ typedef struct SCatalogReq {
SArray *pIndex; // element is index name SArray *pIndex; // element is index name
SArray *pUser; // element is SUserAuthInfo SArray *pUser; // element is SUserAuthInfo
bool qNodeRequired; // valid qnode bool qNodeRequired; // valid qnode
bool forceUpdate;
} SCatalogReq; } SCatalogReq;
typedef struct SMetaData { typedef struct SMetaData {
@ -280,7 +281,7 @@ int32_t catalogUpdateUserAuthInfo(SCatalog* pCtg, SGetUserAuthRsp* pAuth);
int32_t catalogUpdateVgEpSet(SCatalog* pCtg, const char* dbFName, int32_t vgId, SEpSet *epSet); int32_t catalogUpdateVgEpSet(SCatalog* pCtg, const char* dbFName, int32_t vgId, SEpSet *epSet);
int32_t ctgdLaunchAsyncCall(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps, uint64_t reqId); int32_t ctgdLaunchAsyncCall(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps, uint64_t reqId, bool forceUpdate);
/** /**

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

@ -222,7 +222,7 @@ extern int32_t (*queryProcessMsgRsp[TDMT_MAX])(void* output, char* msg, int32_t
|| (_type) == TDMT_VND_DROP_TABLE || (_type) == TDMT_VND_DROP_STB) || (_type) == TDMT_VND_DROP_TABLE || (_type) == TDMT_VND_DROP_STB)
#define NEED_SCHEDULER_RETRY_ERROR(_code) \ #define NEED_SCHEDULER_RETRY_ERROR(_code) \
((_code) == TSDB_CODE_RPC_REDIRECT || (_code) == TSDB_CODE_RPC_NETWORK_UNAVAIL) ((_code) == TSDB_CODE_RPC_REDIRECT || (_code) == TSDB_CODE_RPC_NETWORK_UNAVAIL || (_code) == TSDB_CODE_SCH_TIMEOUT_ERROR)
#define REQUEST_MAX_TRY_TIMES 1 #define REQUEST_MAX_TRY_TIMES 1

2
include/libs/scheduler/scheduler.h
View File

@ -104,6 +104,8 @@ void schedulerAsyncFetchRows(int64_t job, schedulerFetchCallback fp, void* param
int32_t schedulerGetTasksStatus(int64_t job, SArray *pSub); int32_t schedulerGetTasksStatus(int64_t job, SArray *pSub);
void schedulerStopQueryHb(void *pTrans);
/** /**
* Cancel query job * Cancel query job

10
source/client/inc/clientInt.h
View File

@ -57,11 +57,6 @@ enum {
typedef struct SAppInstInfo SAppInstInfo; typedef struct SAppInstInfo SAppInstInfo;
typedef struct {
void* param;
SClientHbReq* req;
} SHbConnInfo;
typedef struct { typedef struct {
char* key; char* key;
// statistics // statistics
@ -71,11 +66,8 @@ typedef struct {
int64_t startTime; int64_t startTime;
// ctl // ctl
SRWLatch lock; // lock is used in serialization SRWLatch lock; // lock is used in serialization
// connection
SAppInstInfo* pAppInstInfo; SAppInstInfo* pAppInstInfo;
// info
SHashObj* activeInfo; // hash<SClientHbKey, SClientHbReq> SHashObj* activeInfo; // hash<SClientHbKey, SClientHbReq>
SHashObj* connInfo; // hash<SClientHbKey, SHbConnInfo>
} SAppHbMgr; } SAppHbMgr;
typedef int32_t (*FHbRspHandle)(SAppHbMgr* pAppHbMgr, SClientHbRsp* pRsp); typedef int32_t (*FHbRspHandle)(SAppHbMgr* pAppHbMgr, SClientHbRsp* pRsp);
@ -325,8 +317,6 @@ void appHbMgrCleanup(void);
int hbRegisterConn(SAppHbMgr* pAppHbMgr, int64_t tscRefId, int64_t clusterId, int8_t connType); int hbRegisterConn(SAppHbMgr* pAppHbMgr, int64_t tscRefId, int64_t clusterId, int8_t connType);
void hbDeregisterConn(SAppHbMgr* pAppHbMgr, SClientHbKey connKey); void hbDeregisterConn(SAppHbMgr* pAppHbMgr, SClientHbKey connKey);
int hbAddConnInfo(SAppHbMgr* pAppHbMgr, SClientHbKey connKey, void* key, void* value, int32_t keyLen, int32_t valueLen);
// --- mq // --- mq
void hbMgrInitMqHbRspHandle(); void hbMgrInitMqHbRspHandle();

15
source/client/src/clientEnv.c
View File

@ -130,8 +130,13 @@ void destroyTscObj(void *pObj) {
SClientHbKey connKey = {.tscRid = pTscObj->id, .connType = pTscObj->connType}; SClientHbKey connKey = {.tscRid = pTscObj->id, .connType = pTscObj->connType};
hbDeregisterConn(pTscObj->pAppInfo->pAppHbMgr, connKey); hbDeregisterConn(pTscObj->pAppInfo->pAppHbMgr, connKey);
atomic_sub_fetch_64(&pTscObj->pAppInfo->numOfConns, 1); int64_t connNum = atomic_sub_fetch_64(&pTscObj->pAppInfo->numOfConns, 1);
closeAllRequests(pTscObj->pRequests); closeAllRequests(pTscObj->pRequests);
schedulerStopQueryHb(pTscObj->pAppInfo->pTransporter);
if (0 == connNum) {
// TODO
//closeTransporter(pTscObj);
}
tscDebug("connObj 0x%" PRIx64 " destroyed, totalConn:%" PRId64, pTscObj->id, pTscObj->pAppInfo->numOfConns); tscDebug("connObj 0x%" PRIx64 " destroyed, totalConn:%" PRId64, pTscObj->id, pTscObj->pAppInfo->numOfConns);
taosThreadMutexDestroy(&pTscObj->mutex); taosThreadMutexDestroy(&pTscObj->mutex);
taosMemoryFreeClear(pTscObj); taosMemoryFreeClear(pTscObj);
@ -223,6 +228,10 @@ static void doDestroyRequest(void *p) {
taosHashRemove(pRequest->pTscObj->pRequests, &pRequest->self, sizeof(pRequest->self)); taosHashRemove(pRequest->pTscObj->pRequests, &pRequest->self, sizeof(pRequest->self));
if (pRequest->body.queryJob != 0) {
schedulerFreeJob(pRequest->body.queryJob);
}
taosMemoryFreeClear(pRequest->msgBuf); taosMemoryFreeClear(pRequest->msgBuf);
taosMemoryFreeClear(pRequest->sqlstr); taosMemoryFreeClear(pRequest->sqlstr);
taosMemoryFreeClear(pRequest->pDb); taosMemoryFreeClear(pRequest->pDb);
@ -230,10 +239,6 @@ static void doDestroyRequest(void *p) {
doFreeReqResultInfo(&pRequest->body.resInfo); doFreeReqResultInfo(&pRequest->body.resInfo);
qDestroyQueryPlan(pRequest->body.pDag); qDestroyQueryPlan(pRequest->body.pDag);
if (pRequest->body.queryJob != 0) {
schedulerFreeJob(pRequest->body.queryJob);
}
taosArrayDestroy(pRequest->tableList); taosArrayDestroy(pRequest->tableList);
taosArrayDestroy(pRequest->dbList); taosArrayDestroy(pRequest->dbList);

101
source/client/src/clientHb.c
View File

@ -129,9 +129,9 @@ static int32_t hbProcessStbInfoRsp(void *value, int32_t valueLen, struct SCatalo
} }
static int32_t hbQueryHbRspHandle(SAppHbMgr *pAppHbMgr, SClientHbRsp *pRsp) { static int32_t hbQueryHbRspHandle(SAppHbMgr *pAppHbMgr, SClientHbRsp *pRsp) {
SHbConnInfo *info = taosHashGet(pAppHbMgr->connInfo, &pRsp->connKey, sizeof(SClientHbKey)); SClientHbReq *pReq = taosHashGet(pAppHbMgr->activeInfo, &pRsp->connKey, sizeof(SClientHbKey));
if (NULL == info) { if (NULL == pReq) {
tscWarn("fail to get connInfo, may be dropped, refId:%" PRIx64 ", type:%d", pRsp->connKey.tscRid, tscWarn("pReq to get activeInfo, may be dropped, refId:%" PRIx64 ", type:%d", pRsp->connKey.tscRid,
pRsp->connKey.connType); pRsp->connKey.connType);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -181,12 +181,11 @@ static int32_t hbQueryHbRspHandle(SAppHbMgr *pAppHbMgr, SClientHbRsp *pRsp) {
break; break;
} }
int64_t *clusterId = (int64_t *)info->param;
struct SCatalog *pCatalog = NULL; struct SCatalog *pCatalog = NULL;
int32_t code = catalogGetHandle(*clusterId, &pCatalog); int32_t code = catalogGetHandle(pReq->clusterId, &pCatalog);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
tscWarn("catalogGetHandle failed, clusterId:%" PRIx64 ", error:%s", *clusterId, tstrerror(code)); tscWarn("catalogGetHandle failed, clusterId:%" PRIx64 ", error:%s", pReq->clusterId, tstrerror(code));
break; break;
} }
@ -199,12 +198,11 @@ static int32_t hbQueryHbRspHandle(SAppHbMgr *pAppHbMgr, SClientHbRsp *pRsp) {
break; break;
} }
int64_t *clusterId = (int64_t *)info->param;
struct SCatalog *pCatalog = NULL; struct SCatalog *pCatalog = NULL;
int32_t code = catalogGetHandle(*clusterId, &pCatalog); int32_t code = catalogGetHandle(pReq->clusterId, &pCatalog);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
tscWarn("catalogGetHandle failed, clusterId:%" PRIx64 ", error:%s", *clusterId, tstrerror(code)); tscWarn("catalogGetHandle failed, clusterId:%" PRIx64 ", error:%s", pReq->clusterId, tstrerror(code));
break; break;
} }
@ -217,12 +215,11 @@ static int32_t hbQueryHbRspHandle(SAppHbMgr *pAppHbMgr, SClientHbRsp *pRsp) {
break; break;
} }
int64_t *clusterId = (int64_t *)info->param;
struct SCatalog *pCatalog = NULL; struct SCatalog *pCatalog = NULL;
int32_t code = catalogGetHandle(*clusterId, &pCatalog); int32_t code = catalogGetHandle(pReq->clusterId, &pCatalog);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
tscWarn("catalogGetHandle failed, clusterId:%" PRIx64 ", error:%s", *clusterId, tstrerror(code)); tscWarn("catalogGetHandle failed, clusterId:%" PRIx64 ", error:%s", pReq->clusterId, tstrerror(code));
break; break;
} }
@ -547,13 +544,10 @@ SClientHbBatchReq *hbGatherAllInfo(SAppHbMgr *pAppHbMgr) {
pOneReq = taosArrayPush(pBatchReq->reqs, pOneReq); pOneReq = taosArrayPush(pBatchReq->reqs, pOneReq);
SHbConnInfo *info = taosHashGet(pAppHbMgr->connInfo, &pOneReq->connKey, sizeof(SClientHbKey)); code = (*clientHbMgr.reqHandle[pOneReq->connKey.connType])(&pOneReq->connKey, &pOneReq->clusterId, pOneReq);
if (info) { if (code) {
code = (*clientHbMgr.reqHandle[pOneReq->connKey.connType])(&pOneReq->connKey, info->param, pOneReq); pIter = taosHashIterate(pAppHbMgr->activeInfo, pIter);
if (code) { continue;
pIter = taosHashIterate(pAppHbMgr->activeInfo, pIter);
continue;
}
} }
//hbClearClientHbReq(pOneReq); //hbClearClientHbReq(pOneReq);
@ -569,23 +563,6 @@ SClientHbBatchReq *hbGatherAllInfo(SAppHbMgr *pAppHbMgr) {
return pBatchReq; return pBatchReq;
} }
void hbClearReqInfo(SAppHbMgr *pAppHbMgr) {
void *pIter = taosHashIterate(pAppHbMgr->activeInfo, NULL);
while (pIter != NULL) {
SClientHbReq *pOneReq = pIter;
tFreeReqKvHash(pOneReq->info);
taosHashClear(pOneReq->info);
if (pOneReq->query) {
taosArrayDestroy(pOneReq->query->queryDesc);
taosMemoryFreeClear(pOneReq->query);
}
pIter = taosHashIterate(pAppHbMgr->activeInfo, pIter);
}
}
void hbThreadFuncUnexpectedStopped(void) { void hbThreadFuncUnexpectedStopped(void) {
atomic_store_8(&clientHbMgr.threadStop, 2); atomic_store_8(&clientHbMgr.threadStop, 2);
} }
@ -715,14 +692,6 @@ SAppHbMgr *appHbMgrInit(SAppInstInfo *pAppInstInfo, char *key) {
} }
taosHashSetFreeFp(pAppHbMgr->activeInfo, tFreeClientHbReq); taosHashSetFreeFp(pAppHbMgr->activeInfo, tFreeClientHbReq);
// init getInfoFunc
pAppHbMgr->connInfo = taosHashInit(64, hbKeyHashFunc, 1, HASH_ENTRY_LOCK);
if (pAppHbMgr->connInfo == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
taosMemoryFree(pAppHbMgr);
return NULL;
}
taosThreadMutexLock(&clientHbMgr.lock); taosThreadMutexLock(&clientHbMgr.lock);
taosArrayPush(clientHbMgr.appHbMgrs, &pAppHbMgr); taosArrayPush(clientHbMgr.appHbMgrs, &pAppHbMgr);
@ -745,15 +714,6 @@ void appHbMgrCleanup(void) {
taosHashCleanup(pTarget->activeInfo); taosHashCleanup(pTarget->activeInfo);
pTarget->activeInfo = NULL; pTarget->activeInfo = NULL;
pIter = taosHashIterate(pTarget->connInfo, NULL);
while (pIter != NULL) {
SHbConnInfo *info = pIter;
taosMemoryFree(info->param);
pIter = taosHashIterate(pTarget->connInfo, pIter);
}
taosHashCleanup(pTarget->connInfo);
pTarget->connInfo = NULL;
taosMemoryFree(pTarget->key); taosMemoryFree(pTarget->key);
taosMemoryFree(pTarget); taosMemoryFree(pTarget);
} }
@ -791,7 +751,7 @@ void hbMgrCleanUp() {
clientHbMgr.appHbMgrs = NULL; clientHbMgr.appHbMgrs = NULL;
} }
int hbRegisterConnImpl(SAppHbMgr *pAppHbMgr, SClientHbKey connKey, SHbConnInfo *info) { int hbRegisterConnImpl(SAppHbMgr *pAppHbMgr, SClientHbKey connKey, int64_t clusterId) {
// init hash in activeinfo // init hash in activeinfo
void *data = taosHashGet(pAppHbMgr->activeInfo, &connKey, sizeof(SClientHbKey)); void *data = taosHashGet(pAppHbMgr->activeInfo, &connKey, sizeof(SClientHbKey));
if (data != NULL) { if (data != NULL) {
@ -799,17 +759,11 @@ int hbRegisterConnImpl(SAppHbMgr *pAppHbMgr, SClientHbKey connKey, SHbConnInfo *
} }
SClientHbReq hbReq = {0}; SClientHbReq hbReq = {0};
hbReq.connKey = connKey; hbReq.connKey = connKey;
hbReq.clusterId = clusterId;
//hbReq.info = taosHashInit(64, hbKeyHashFunc, 1, HASH_ENTRY_LOCK); //hbReq.info = taosHashInit(64, hbKeyHashFunc, 1, HASH_ENTRY_LOCK);
taosHashPut(pAppHbMgr->activeInfo, &connKey, sizeof(SClientHbKey), &hbReq, sizeof(SClientHbReq)); taosHashPut(pAppHbMgr->activeInfo, &connKey, sizeof(SClientHbKey), &hbReq, sizeof(SClientHbReq));
// init hash
if (info != NULL) {
SClientHbReq *pReq = taosHashGet(pAppHbMgr->activeInfo, &connKey, sizeof(SClientHbKey));
info->req = pReq;
taosHashPut(pAppHbMgr->connInfo, &connKey, sizeof(SClientHbKey), info, sizeof(SHbConnInfo));
}
atomic_add_fetch_32(&pAppHbMgr->connKeyCnt, 1); atomic_add_fetch_32(&pAppHbMgr->connKeyCnt, 1);
return 0; return 0;
} }
@ -819,15 +773,10 @@ int hbRegisterConn(SAppHbMgr *pAppHbMgr, int64_t tscRefId, int64_t clusterId, in
.tscRid = tscRefId, .tscRid = tscRefId,
.connType = connType, .connType = connType,
}; };
SHbConnInfo info = {0};
switch (connType) { switch (connType) {
case CONN_TYPE__QUERY: { case CONN_TYPE__QUERY: {
int64_t *pClusterId = taosMemoryMalloc(sizeof(int64_t)); return hbRegisterConnImpl(pAppHbMgr, connKey, clusterId);
*pClusterId = clusterId;
info.param = pClusterId;
return hbRegisterConnImpl(pAppHbMgr, connKey, &info);
} }
case CONN_TYPE__TMQ: { case CONN_TYPE__TMQ: {
return 0; return 0;
@ -844,26 +793,10 @@ void hbDeregisterConn(SAppHbMgr *pAppHbMgr, SClientHbKey connKey) {
taosHashRemove(pAppHbMgr->activeInfo, &connKey, sizeof(SClientHbKey)); taosHashRemove(pAppHbMgr->activeInfo, &connKey, sizeof(SClientHbKey));
} }
SHbConnInfo *info = taosHashGet(pAppHbMgr->connInfo, &connKey, sizeof(SClientHbKey)); if (NULL == pReq) {
if (info) {
taosMemoryFree(info->param);
taosHashRemove(pAppHbMgr->connInfo, &connKey, sizeof(SClientHbKey));
}
if (NULL == pReq || NULL == info) {
return; return;
} }
atomic_sub_fetch_32(&pAppHbMgr->connKeyCnt, 1); atomic_sub_fetch_32(&pAppHbMgr->connKeyCnt, 1);
} }
int hbAddConnInfo(SAppHbMgr *pAppHbMgr, SClientHbKey connKey, void *key, void *value, int32_t keyLen,
int32_t valueLen) {
// find req by connection id
SClientHbReq *pReq = taosHashGet(pAppHbMgr->activeInfo, &connKey, sizeof(SClientHbKey));
ASSERT(pReq != NULL);
taosHashPut(pReq->info, key, keyLen, value, valueLen);
return 0;
}

2
source/client/src/clientMsgHandler.c
View File

@ -180,7 +180,7 @@ int32_t processUseDbRsp(void* param, const SDataBuf* pMsg, int32_t code) {
taosMemoryFreeClear(output.dbVgroup); taosMemoryFreeClear(output.dbVgroup);
tscError("0x%" PRIx64 " failed to build use db output since %s", pRequest->requestId, terrstr()); tscError("0x%" PRIx64 " failed to build use db output since %s", pRequest->requestId, terrstr());
} else if (output.dbVgroup) { } else if (output.dbVgroup && output.dbVgroup->vgHash) {
struct SCatalog* pCatalog = NULL; struct SCatalog* pCatalog = NULL;
int32_t code1 = catalogGetHandle(pRequest->pTscObj->pAppInfo->clusterId, &pCatalog); int32_t code1 = catalogGetHandle(pRequest->pTscObj->pAppInfo->clusterId, &pCatalog);

2
source/client/src/clientSml.c
View File

@ -2332,6 +2332,8 @@ static int32_t isSchemalessDb(SSmlHandle* info){
smlBuildInvalidDataMsg(&info->msgBuf, "catalogGetDBCfg error, code:", tstrerror(code)); smlBuildInvalidDataMsg(&info->msgBuf, "catalogGetDBCfg error, code:", tstrerror(code));
return code; return code;
} }
taosArrayDestroy(pInfo.pRetensions);
if (!pInfo.schemaless){ if (!pInfo.schemaless){
info->pRequest->code = TSDB_CODE_SML_INVALID_DB_CONF; info->pRequest->code = TSDB_CODE_SML_INVALID_DB_CONF;
smlBuildInvalidDataMsg(&info->msgBuf, "can not insert into schemaless db:", dbFname); smlBuildInvalidDataMsg(&info->msgBuf, "can not insert into schemaless db:", dbFname);

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

@ -1590,7 +1590,6 @@ int32_t buildSubmitReqFromDataBlock(SSubmitReq** pReq, const SArray* pDataBlocks
for (int32_t k = 0; k < colNum; ++k) { // iterate by column for (int32_t k = 0; k < colNum; ++k) { // iterate by column
SColumnInfoData* pColInfoData = taosArrayGet(pDataBlock->pDataBlock, k); SColumnInfoData* pColInfoData = taosArrayGet(pDataBlock->pDataBlock, k);
STColumn* pCol = &pTSchema->columns[k]; STColumn* pCol = &pTSchema->columns[k];
ASSERT(pCol->type == pColInfoData->info.type);
void* var = POINTER_SHIFT(pColInfoData->pData, j * pColInfoData->info.bytes); void* var = POINTER_SHIFT(pColInfoData->pData, j * pColInfoData->info.bytes);
switch (pColInfoData->info.type) { switch (pColInfoData->info.type) {
case TSDB_DATA_TYPE_TIMESTAMP: case TSDB_DATA_TYPE_TIMESTAMP:
@ -1614,20 +1613,39 @@ int32_t buildSubmitReqFromDataBlock(SSubmitReq** pReq, const SArray* pDataBlocks
case TSDB_DATA_TYPE_VARBINARY: case TSDB_DATA_TYPE_VARBINARY:
case TSDB_DATA_TYPE_DECIMAL: case TSDB_DATA_TYPE_DECIMAL:
case TSDB_DATA_TYPE_BLOB: case TSDB_DATA_TYPE_BLOB:
case TSDB_DATA_TYPE_JSON:
case TSDB_DATA_TYPE_MEDIUMBLOB: case TSDB_DATA_TYPE_MEDIUMBLOB:
uError("the column type %" PRIi16 " is defined but not implemented yet", pColInfoData->info.type); uError("the column type %" PRIi16 " is defined but not implemented yet", pColInfoData->info.type);
TASSERT(0); TASSERT(0);
break; break;
default: default:
if (pColInfoData->info.type < TSDB_DATA_TYPE_MAX && pColInfoData->info.type > TSDB_DATA_TYPE_NULL) { if (pColInfoData->info.type < TSDB_DATA_TYPE_MAX && pColInfoData->info.type > TSDB_DATA_TYPE_NULL) {
tdAppendColValToRow(&rb, PRIMARYKEY_TIMESTAMP_COL_ID + k, pColInfoData->info.type, TD_VTYPE_NORM, var, true, offset, k); char tv[8] = {0};
if (pColInfoData->info.type == TSDB_DATA_TYPE_FLOAT) {
float v = 0;
GET_TYPED_DATA(v, float, pColInfoData->info.type, var);
SET_TYPED_DATA(&tv, pCol->type, v);
} else if (pColInfoData->info.type == TSDB_DATA_TYPE_DOUBLE) {
double v = 0;
GET_TYPED_DATA(v, double, pColInfoData->info.type, var);
SET_TYPED_DATA(&tv, pCol->type, v);
} else if (IS_SIGNED_NUMERIC_TYPE(pColInfoData->info.type)) {
int64_t v = 0;
GET_TYPED_DATA(v, int64_t, pColInfoData->info.type, var);
SET_TYPED_DATA(&tv, pCol->type, v);
} else {
uint64_t v = 0;
GET_TYPED_DATA(v, uint64_t, pColInfoData->info.type, var);
SET_TYPED_DATA(&tv, pCol->type, v);
}
tdAppendColValToRow(&rb, PRIMARYKEY_TIMESTAMP_COL_ID + k, pCol->type, TD_VTYPE_NORM, tv, true, offset, k);
} else { } else {
uError("the column type %" PRIi16 " is undefined\n", pColInfoData->info.type); uError("the column type %" PRIi16 " is undefined\n", pColInfoData->info.type);
TASSERT(0); TASSERT(0);
} }
break; break;
} }
offset += TYPE_BYTES[pColInfoData->info.type]; offset += TYPE_BYTES[pCol->type]; // sum/avg would convert to int64_t/uint64_t/double during aggregation
} }
dataLen += TD_ROW_LEN(rb.pBuf); dataLen += TD_ROW_LEN(rb.pBuf);
#ifdef TD_DEBUG_PRINT_ROW #ifdef TD_DEBUG_PRINT_ROW

60
source/common/src/tmsg.c
View File

@ -694,7 +694,6 @@ void tFreeSMAltertbReq(SMAlterStbReq *pReq) {
pReq->pFields = NULL; pReq->pFields = NULL;
} }
int32_t tSerializeSEpSet(void *buf, int32_t bufLen, const SEpSet *pEpset) { int32_t tSerializeSEpSet(void *buf, int32_t bufLen, const SEpSet *pEpset) {
SEncoder encoder = {0}; SEncoder encoder = {0};
tEncoderInit(&encoder, buf, bufLen); tEncoderInit(&encoder, buf, bufLen);
@ -3674,12 +3673,12 @@ int32_t tEncodeTSma(SEncoder *pCoder, const STSma *pSma) {
if (tEncodeCStr(pCoder, pSma->tagsFilter) < 0) return -1; if (tEncodeCStr(pCoder, pSma->tagsFilter) < 0) return -1;
} }
for (int32_t v = 0; v < pSma->numOfVgroups; ++v) { for (int32_t v = 0; v < pSma->numOfVgroups; ++v) {
if (tEncodeI32(pCoder, pSma->vgEpSet[v].vgId) < 0) return -1; if (tEncodeI32(pCoder, pSma->pVgEpSet[v].vgId) < 0) return -1;
if (tEncodeI8(pCoder, pSma->vgEpSet[v].epSet.inUse) < 0) return -1; if (tEncodeI8(pCoder, pSma->pVgEpSet[v].epSet.inUse) < 0) return -1;
int8_t numOfEps = pSma->vgEpSet[v].epSet.numOfEps; int8_t numOfEps = pSma->pVgEpSet[v].epSet.numOfEps;
if (tEncodeI8(pCoder, numOfEps) < 0) return -1; if (tEncodeI8(pCoder, numOfEps) < 0) return -1;
for (int32_t n = 0; n < numOfEps; ++n) { for (int32_t n = 0; n < numOfEps; ++n) {
const SEp *pEp = &pSma->vgEpSet[v].epSet.eps[n]; const SEp *pEp = &pSma->pVgEpSet[v].epSet.eps[n];
if (tEncodeCStr(pCoder, pEp->fqdn) < 0) return -1; if (tEncodeCStr(pCoder, pEp->fqdn) < 0) return -1;
if (tEncodeU16(pCoder, pEp->port) < 0) return -1; if (tEncodeU16(pCoder, pEp->port) < 0) return -1;
} }
@ -3712,15 +3711,25 @@ int32_t tDecodeTSma(SDecoder *pCoder, STSma *pSma) {
} else { } else {
pSma->tagsFilter = NULL; pSma->tagsFilter = NULL;
} }
for (int32_t v = 0; v < pSma->numOfVgroups; ++v) { if (pSma->numOfVgroups > 0) {
if (tDecodeI32(pCoder, &pSma->vgEpSet[v].vgId) < 0) return -1; pSma->pVgEpSet = (SVgEpSet *)tDecoderMalloc(pCoder, pSma->numOfVgroups * sizeof(SVgEpSet));
if (tDecodeI8(pCoder, &pSma->vgEpSet[v].epSet.inUse) < 0) return -1; if (!pSma->pVgEpSet) {
if (tDecodeI8(pCoder, &pSma->vgEpSet[v].epSet.numOfEps) < 0) return -1; terrno = TSDB_CODE_OUT_OF_MEMORY;
int8_t numOfEps = pSma->vgEpSet[v].epSet.numOfEps; return -1;
for (int32_t n = 0; n < numOfEps; ++n) { }
SEp *pEp = &pSma->vgEpSet[v].epSet.eps[n];
if (tDecodeCStrTo(pCoder, pEp->fqdn) < 0) return -1; memset(pSma->pVgEpSet, 0, pSma->numOfVgroups * sizeof(SVgEpSet));
if (tDecodeU16(pCoder, &pEp->port) < 0) return -1;
for (int32_t v = 0; v < pSma->numOfVgroups; ++v) {
if (tDecodeI32(pCoder, &pSma->pVgEpSet[v].vgId) < 0) return -1;
if (tDecodeI8(pCoder, &pSma->pVgEpSet[v].epSet.inUse) < 0) return -1;
if (tDecodeI8(pCoder, &pSma->pVgEpSet[v].epSet.numOfEps) < 0) return -1;
int8_t numOfEps = pSma->pVgEpSet[v].epSet.numOfEps;
for (int32_t n = 0; n < numOfEps; ++n) {
SEp *pEp = &pSma->pVgEpSet[v].epSet.eps[n];
if (tDecodeCStrTo(pCoder, pEp->fqdn) < 0) return -1;
if (tDecodeU16(pCoder, &pEp->port) < 0) return -1;
}
} }
} }
@ -3765,7 +3774,7 @@ int32_t tDecodeSVDropTSmaReq(SDecoder *pCoder, SVDropTSmaReq *pReq) {
return 0; return 0;
} }
int32_t tEncodeSVGetTSmaExpWndsReq(SEncoder* pCoder, const SVGetTsmaExpWndsReq* pReq) { int32_t tEncodeSVGetTSmaExpWndsReq(SEncoder *pCoder, const SVGetTsmaExpWndsReq *pReq) {
if (tStartEncode(pCoder) < 0) return -1; if (tStartEncode(pCoder) < 0) return -1;
if (tEncodeI64(pCoder, pReq->indexUid) < 0) return -1; if (tEncodeI64(pCoder, pReq->indexUid) < 0) return -1;
@ -3776,7 +3785,7 @@ int32_t tEncodeSVGetTSmaExpWndsReq(SEncoder* pCoder, const SVGetTsmaExpWndsReq*
return 0; return 0;
} }
int32_t tDecodeSVGetTsmaExpWndsReq(SDecoder* pCoder, SVGetTsmaExpWndsReq* pReq) { int32_t tDecodeSVGetTsmaExpWndsReq(SDecoder *pCoder, SVGetTsmaExpWndsReq *pReq) {
if (tStartDecode(pCoder) < 0) return -1; if (tStartDecode(pCoder) < 0) return -1;
if (tDecodeI64(pCoder, &pReq->indexUid) < 0) return -1; if (tDecodeI64(pCoder, &pReq->indexUid) < 0) return -1;
@ -3787,7 +3796,7 @@ int32_t tDecodeSVGetTsmaExpWndsReq(SDecoder* pCoder, SVGetTsmaExpWndsReq* pReq)
return 0; return 0;
} }
int32_t tEncodeSVGetTSmaExpWndsRsp(SEncoder* pCoder, const SVGetTsmaExpWndsRsp* pReq) { int32_t tEncodeSVGetTSmaExpWndsRsp(SEncoder *pCoder, const SVGetTsmaExpWndsRsp *pReq) {
if (tStartEncode(pCoder) < 0) return -1; if (tStartEncode(pCoder) < 0) return -1;
if (tEncodeI64(pCoder, pReq->indexUid) < 0) return -1; if (tEncodeI64(pCoder, pReq->indexUid) < 0) return -1;
@ -3814,7 +3823,7 @@ int32_t tDecodeSVGetTsmaExpWndsRsp(SDecoder *pCoder, SVGetTsmaExpWndsRsp *pReq)
return 0; return 0;
} }
int32_t tEncodeSVDeleteReq(SEncoder* pCoder, const SVDeleteReq* pReq) { int32_t tEncodeSVDeleteReq(SEncoder *pCoder, const SVDeleteReq *pReq) {
if (tStartEncode(pCoder) < 0) return -1; if (tStartEncode(pCoder) < 0) return -1;
if (tEncodeI64(pCoder, pReq->delUid) < 0) return -1; if (tEncodeI64(pCoder, pReq->delUid) < 0) return -1;
@ -3832,7 +3841,7 @@ int32_t tEncodeSVDeleteReq(SEncoder* pCoder, const SVDeleteReq* pReq) {
return 0; return 0;
} }
int32_t tDecodeSVDeleteReq(SDecoder* pCoder, SVDeleteReq* pReq) { int32_t tDecodeSVDeleteReq(SDecoder *pCoder, SVDeleteReq *pReq) {
if (tStartDecode(pCoder) < 0) return -1; if (tStartDecode(pCoder) < 0) return -1;
if (tDecodeI64(pCoder, &pReq->delUid) < 0) return -1; if (tDecodeI64(pCoder, &pReq->delUid) < 0) return -1;
@ -3850,7 +3859,7 @@ int32_t tDecodeSVDeleteReq(SDecoder* pCoder, SVDeleteReq* pReq) {
return 0; return 0;
} }
int32_t tEncodeSVDeleteRsp(SEncoder* pCoder, const SVDeleteRsp* pReq) { int32_t tEncodeSVDeleteRsp(SEncoder *pCoder, const SVDeleteRsp *pReq) {
if (tStartEncode(pCoder) < 0) return -1; if (tStartEncode(pCoder) < 0) return -1;
if (tEncodeI32(pCoder, pReq->code) < 0) return -1; if (tEncodeI32(pCoder, pReq->code) < 0) return -1;
@ -3860,7 +3869,7 @@ int32_t tEncodeSVDeleteRsp(SEncoder* pCoder, const SVDeleteRsp* pReq) {
return 0; return 0;
} }
int32_t tDecodeSVDeleteRsp(SDecoder* pCoder, SVDeleteRsp* pReq) { int32_t tDecodeSVDeleteRsp(SDecoder *pCoder, SVDeleteRsp *pReq) {
if (tStartDecode(pCoder) < 0) return -1; if (tStartDecode(pCoder) < 0) return -1;
if (tDecodeI32(pCoder, &pReq->code) < 0) return -1; if (tDecodeI32(pCoder, &pReq->code) < 0) return -1;
@ -4502,7 +4511,7 @@ int32_t tDecodeSVAlterTbRsp(SDecoder *pDecoder, SVAlterTbRsp *pRsp) {
} }
int32_t tDeserializeSVAlterTbRsp(void *buf, int32_t bufLen, SVAlterTbRsp *pRsp) { int32_t tDeserializeSVAlterTbRsp(void *buf, int32_t bufLen, SVAlterTbRsp *pRsp) {
int32_t meta = 0; int32_t meta = 0;
SDecoder decoder = {0}; SDecoder decoder = {0};
tDecoderInit(&decoder, buf, bufLen); tDecoderInit(&decoder, buf, bufLen);
@ -4543,7 +4552,7 @@ int32_t tDecodeSMAlterStbRsp(SDecoder *pDecoder, SMAlterStbRsp *pRsp) {
} }
int32_t tDeserializeSMAlterStbRsp(void *buf, int32_t bufLen, SMAlterStbRsp *pRsp) { int32_t tDeserializeSMAlterStbRsp(void *buf, int32_t bufLen, SMAlterStbRsp *pRsp) {
int32_t meta = 0; int32_t meta = 0;
SDecoder decoder = {0}; SDecoder decoder = {0};
tDecoderInit(&decoder, buf, bufLen); tDecoderInit(&decoder, buf, bufLen);
@ -4559,7 +4568,7 @@ int32_t tDeserializeSMAlterStbRsp(void *buf, int32_t bufLen, SMAlterStbRsp *pRsp
return 0; return 0;
} }
void tFreeSMAlterStbRsp(SMAlterStbRsp* pRsp) { void tFreeSMAlterStbRsp(SMAlterStbRsp *pRsp) {
if (NULL == pRsp) { if (NULL == pRsp) {
return; return;
} }
@ -4569,6 +4578,3 @@ void tFreeSMAlterStbRsp(SMAlterStbRsp* pRsp) {
taosMemoryFree(pRsp->pMeta); taosMemoryFree(pRsp->pMeta);
} }
} }

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

@ -298,28 +298,30 @@ typedef struct {
} SVgObj; } SVgObj;
typedef struct { typedef struct {
char name[TSDB_TABLE_FNAME_LEN]; char name[TSDB_TABLE_FNAME_LEN];
char stb[TSDB_TABLE_FNAME_LEN]; char stb[TSDB_TABLE_FNAME_LEN];
char db[TSDB_DB_FNAME_LEN]; char db[TSDB_DB_FNAME_LEN];
int64_t createdTime; int64_t createdTime;
int64_t uid; int64_t uid;
int64_t stbUid; int64_t stbUid;
int64_t dbUid; int64_t dbUid;
int8_t intervalUnit; int8_t intervalUnit;
int8_t slidingUnit; int8_t slidingUnit;
int8_t timezone; int8_t timezone;
int32_t dstVgId; // for stream int32_t dstVgId; // for stream
int64_t interval; int64_t interval;
int64_t offset; int64_t offset;
int64_t sliding; int64_t sliding;
int32_t exprLen; // strlen + 1 int32_t exprLen; // strlen + 1
int32_t tagsFilterLen; int32_t tagsFilterLen;
int32_t sqlLen; int32_t sqlLen;
int32_t astLen; int32_t astLen;
char* expr; int32_t numOfVgroups;
char* tagsFilter; char* expr;
char* sql; char* tagsFilter;
char* ast; char* sql;
char* ast;
SVgEpSet* pVgEpSet;
} SSmaObj; } SSmaObj;
typedef struct { typedef struct {

61
source/dnode/mnode/impl/src/mndSma.c
View File

@ -36,6 +36,7 @@ static SSdbRow *mndSmaActionDecode(SSdbRaw *pRaw);
static int32_t mndSmaActionInsert(SSdb *pSdb, SSmaObj *pSma); static int32_t mndSmaActionInsert(SSdb *pSdb, SSmaObj *pSma);
static int32_t mndSmaActionDelete(SSdb *pSdb, SSmaObj *pSpSmatb); static int32_t mndSmaActionDelete(SSdb *pSdb, SSmaObj *pSpSmatb);
static int32_t mndSmaActionUpdate(SSdb *pSdb, SSmaObj *pOld, SSmaObj *pNew); static int32_t mndSmaActionUpdate(SSdb *pSdb, SSmaObj *pOld, SSmaObj *pNew);
static int32_t mndSmaGetVgEpSet(SMnode *pMnode, SDbObj *pDb, SVgEpSet **ppVgEpSet, int32_t *numOfVgroups);
static int32_t mndProcessMCreateSmaReq(SRpcMsg *pReq); static int32_t mndProcessMCreateSmaReq(SRpcMsg *pReq);
static int32_t mndProcessMDropSmaReq(SRpcMsg *pReq); static int32_t mndProcessMDropSmaReq(SRpcMsg *pReq);
static int32_t mndProcessGetSmaReq(SRpcMsg *pReq); static int32_t mndProcessGetSmaReq(SRpcMsg *pReq);
@ -262,6 +263,8 @@ static void *mndBuildVCreateSmaReq(SMnode *pMnode, SVgObj *pVgroup, SSmaObj *pSm
req.sliding = pSma->sliding; req.sliding = pSma->sliding;
req.expr = pSma->expr; req.expr = pSma->expr;
req.tagsFilter = pSma->tagsFilter; req.tagsFilter = pSma->tagsFilter;
req.numOfVgroups = pSma->numOfVgroups;
req.pVgEpSet = pSma->pVgEpSet;
// get length // get length
int32_t ret = 0; int32_t ret = 0;
@ -420,6 +423,15 @@ static int32_t mndSetCreateSmaVgroupRedoActions(SMnode *pMnode, STrans *pTrans,
mndReleaseDnode(pMnode, pDnode); mndReleaseDnode(pMnode, pDnode);
// todo add sma info here // todo add sma info here
SVgEpSet *pVgEpSet = NULL;
int32_t numOfVgroups = 0;
if (mndSmaGetVgEpSet(pMnode, pDb, &pVgEpSet, &numOfVgroups) != 0) {
return -1;
}
pSma->pVgEpSet = pVgEpSet;
pSma->numOfVgroups = numOfVgroups;
int32_t smaContLen = 0; int32_t smaContLen = 0;
void *pSmaReq = mndBuildVCreateSmaReq(pMnode, pVgroup, pSma, &smaContLen); void *pSmaReq = mndBuildVCreateSmaReq(pMnode, pVgroup, pSma, &smaContLen);
if (pSmaReq == NULL) return -1; if (pSmaReq == NULL) return -1;
@ -963,3 +975,52 @@ static void mndCancelGetNextSma(SMnode *pMnode, void *pIter) {
SSdb *pSdb = pMnode->pSdb; SSdb *pSdb = pMnode->pSdb;
sdbCancelFetch(pSdb, pIter); sdbCancelFetch(pSdb, pIter);
} }
static int32_t mndSmaGetVgEpSet(SMnode *pMnode, SDbObj *pDb, SVgEpSet **ppVgEpSet, int32_t *numOfVgroups) {
SSdb *pSdb = pMnode->pSdb;
SVgObj *pVgroup = NULL;
void *pIter = NULL;
SVgEpSet *pVgEpSet = NULL;
int32_t nAllocVgs = 16;
int32_t nVgs = 0;
pVgEpSet = taosMemoryCalloc(nAllocVgs, sizeof(SVgEpSet));
if (!pVgEpSet) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
while (1) {
pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
if (pIter == NULL) break;
if (pVgroup->dbUid != pDb->uid) {
sdbRelease(pSdb, pVgroup);
continue;
}
if (nVgs >= nAllocVgs) {
void *p = taosMemoryRealloc(pVgEpSet, nAllocVgs * 2 * sizeof(SVgEpSet));
if (!p) {
taosMemoryFree(pVgEpSet);
sdbCancelFetch(pSdb, pIter);
sdbRelease(pSdb, pVgroup);
terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1;
}
pVgEpSet = (SVgEpSet *)p;
nAllocVgs *= 2;
}
(pVgEpSet + nVgs)->vgId = pVgroup->vgId;
(pVgEpSet + nVgs)->epSet = mndGetVgroupEpset(pMnode, pVgroup);
++nVgs;
sdbRelease(pSdb, pVgroup);
}
*ppVgEpSet = pVgEpSet;
*numOfVgroups = nVgs;
return 0;
}

2
source/dnode/vnode/src/meta/metaTable.c
View File

@ -103,7 +103,7 @@ static int metaSaveJsonVarToIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry, const
tIndexJsonPut(pMeta->pTagIvtIdx, terms, tuid); tIndexJsonPut(pMeta->pTagIvtIdx, terms, tuid);
indexMultiTermDestroy(terms); indexMultiTermDestroy(terms);
#endif #endif
return -1; return 0;
} }
int metaCreateSTable(SMeta *pMeta, int64_t version, SVCreateStbReq *pReq) { int metaCreateSTable(SMeta *pMeta, int64_t version, SVCreateStbReq *pReq) {

21
source/libs/catalog/inc/catalogInt.h
View File

@ -52,6 +52,7 @@ enum {
CTG_OP_UPDATE_VGROUP = 0, CTG_OP_UPDATE_VGROUP = 0,
CTG_OP_UPDATE_TB_META, CTG_OP_UPDATE_TB_META,
CTG_OP_DROP_DB_CACHE, CTG_OP_DROP_DB_CACHE,
CTG_OP_DROP_DB_VGROUP,
CTG_OP_DROP_STB_META, CTG_OP_DROP_STB_META,
CTG_OP_DROP_TB_META, CTG_OP_DROP_TB_META,
CTG_OP_UPDATE_USER, CTG_OP_UPDATE_USER,
@ -266,26 +267,32 @@ typedef struct SCtgUpdateTblMsg {
STableMetaOutput* output; STableMetaOutput* output;
} SCtgUpdateTblMsg; } SCtgUpdateTblMsg;
typedef struct SCtgRemoveDBMsg { typedef struct SCtgDropDBMsg {
SCatalog* pCtg; SCatalog* pCtg;
char dbFName[TSDB_DB_FNAME_LEN]; char dbFName[TSDB_DB_FNAME_LEN];
uint64_t dbId; uint64_t dbId;
} SCtgRemoveDBMsg; } SCtgDropDBMsg;
typedef struct SCtgRemoveStbMsg { typedef struct SCtgDropDbVgroupMsg {
SCatalog* pCtg;
char dbFName[TSDB_DB_FNAME_LEN];
} SCtgDropDbVgroupMsg;
typedef struct SCtgDropStbMetaMsg {
SCatalog* pCtg; SCatalog* pCtg;
char dbFName[TSDB_DB_FNAME_LEN]; char dbFName[TSDB_DB_FNAME_LEN];
char stbName[TSDB_TABLE_NAME_LEN]; char stbName[TSDB_TABLE_NAME_LEN];
uint64_t dbId; uint64_t dbId;
uint64_t suid; uint64_t suid;
} SCtgRemoveStbMsg; } SCtgDropStbMetaMsg;
typedef struct SCtgRemoveTblMsg { typedef struct SCtgDropTblMetaMsg {
SCatalog* pCtg; SCatalog* pCtg;
char dbFName[TSDB_DB_FNAME_LEN]; char dbFName[TSDB_DB_FNAME_LEN];
char tbName[TSDB_TABLE_NAME_LEN]; char tbName[TSDB_TABLE_NAME_LEN];
uint64_t dbId; uint64_t dbId;
} SCtgRemoveTblMsg; } SCtgDropTblMetaMsg;
typedef struct SCtgUpdateUserMsg { typedef struct SCtgUpdateUserMsg {
SCatalog* pCtg; SCatalog* pCtg;
@ -451,6 +458,7 @@ int32_t ctgGetTbMetaFromCache(CTG_PARAMS, SCtgTbMetaCtx* ctx, STableMeta** pTabl
int32_t ctgOpUpdateVgroup(SCtgCacheOperation *action); int32_t ctgOpUpdateVgroup(SCtgCacheOperation *action);
int32_t ctgOpUpdateTbMeta(SCtgCacheOperation *action); int32_t ctgOpUpdateTbMeta(SCtgCacheOperation *action);
int32_t ctgOpDropDbCache(SCtgCacheOperation *action); int32_t ctgOpDropDbCache(SCtgCacheOperation *action);
int32_t ctgOpDropDbVgroup(SCtgCacheOperation *action);
int32_t ctgOpDropStbMeta(SCtgCacheOperation *action); int32_t ctgOpDropStbMeta(SCtgCacheOperation *action);
int32_t ctgOpDropTbMeta(SCtgCacheOperation *action); int32_t ctgOpDropTbMeta(SCtgCacheOperation *action);
int32_t ctgOpUpdateUser(SCtgCacheOperation *action); int32_t ctgOpUpdateUser(SCtgCacheOperation *action);
@ -464,6 +472,7 @@ int32_t ctgReadTbMetaFromCache(SCatalog* pCtg, SCtgTbMetaCtx* ctx, STableMeta**
int32_t ctgReadTbVerFromCache(SCatalog *pCtg, const SName *pTableName, int32_t *sver, int32_t *tver, int32_t *tbType, uint64_t *suid, char *stbName); int32_t ctgReadTbVerFromCache(SCatalog *pCtg, const SName *pTableName, int32_t *sver, int32_t *tver, int32_t *tbType, uint64_t *suid, char *stbName);
int32_t ctgChkAuthFromCache(SCatalog* pCtg, const char* user, const char* dbFName, AUTH_TYPE type, bool *inCache, bool *pass); int32_t ctgChkAuthFromCache(SCatalog* pCtg, const char* user, const char* dbFName, AUTH_TYPE type, bool *inCache, bool *pass);
int32_t ctgDropDbCacheEnqueue(SCatalog* pCtg, const char *dbFName, int64_t dbId); int32_t ctgDropDbCacheEnqueue(SCatalog* pCtg, const char *dbFName, int64_t dbId);
int32_t ctgDropDbVgroupEnqueue(SCatalog* pCtg, const char *dbFName, bool syncReq);
int32_t ctgDropStbMetaEnqueue(SCatalog* pCtg, const char *dbFName, int64_t dbId, const char *stbName, uint64_t suid, bool syncReq); int32_t ctgDropStbMetaEnqueue(SCatalog* pCtg, const char *dbFName, int64_t dbId, const char *stbName, uint64_t suid, bool syncReq);
int32_t ctgDropTbMetaEnqueue(SCatalog* pCtg, const char *dbFName, int64_t dbId, const char *tbName, bool syncReq); int32_t ctgDropTbMetaEnqueue(SCatalog* pCtg, const char *dbFName, int64_t dbId, const char *tbName, bool syncReq);
int32_t ctgUpdateVgroupEnqueue(SCatalog* pCtg, const char *dbFName, int64_t dbId, SDBVgInfo* dbInfo, bool syncReq); int32_t ctgUpdateVgroupEnqueue(SCatalog* pCtg, const char *dbFName, int64_t dbId, SDBVgInfo* dbInfo, bool syncReq);

6
source/libs/catalog/src/ctgAsync.c
View File

@ -286,6 +286,9 @@ int32_t ctgInitJob(CTG_PARAMS, SCtgJob** job, uint64_t reqId, const SCatalogReq*
int32_t taskIdx = 0; int32_t taskIdx = 0;
for (int32_t i = 0; i < dbVgNum; ++i) { for (int32_t i = 0; i < dbVgNum; ++i) {
char* dbFName = taosArrayGet(pReq->pDbVgroup, i); char* dbFName = taosArrayGet(pReq->pDbVgroup, i);
if (pReq->forceUpdate) {
ctgDropDbVgroupEnqueue(pCtg, dbFName, true);
}
CTG_ERR_JRET(ctgInitGetDbVgTask(pJob, taskIdx++, dbFName)); CTG_ERR_JRET(ctgInitGetDbVgTask(pJob, taskIdx++, dbFName));
} }
@ -301,6 +304,9 @@ int32_t ctgInitJob(CTG_PARAMS, SCtgJob** job, uint64_t reqId, const SCatalogReq*
for (int32_t i = 0; i < tbMetaNum; ++i) { for (int32_t i = 0; i < tbMetaNum; ++i) {
SName* name = taosArrayGet(pReq->pTableMeta, i); SName* name = taosArrayGet(pReq->pTableMeta, i);
if (pReq->forceUpdate) {
catalogRemoveTableMeta(pCtg, name);
}
CTG_ERR_JRET(ctgInitGetTbMetaTask(pJob, taskIdx++, name)); CTG_ERR_JRET(ctgInitGetTbMetaTask(pJob, taskIdx++, name));
} }

80
source/libs/catalog/src/ctgCache.c
View File

@ -35,6 +35,11 @@ SCtgOperation gCtgCacheOperation[CTG_OP_MAX] = {
"drop DB", "drop DB",
ctgOpDropDbCache ctgOpDropDbCache
}, },
{
CTG_OP_DROP_DB_VGROUP,
"drop DBVgroup",
ctgOpDropDbVgroup
},
{ {
CTG_OP_DROP_STB_META, CTG_OP_DROP_STB_META,
"drop stbMeta", "drop stbMeta",
@ -563,9 +568,9 @@ int32_t ctgEnqueue(SCatalog* pCtg, SCtgCacheOperation *operation) {
int32_t ctgDropDbCacheEnqueue(SCatalog* pCtg, const char *dbFName, int64_t dbId) { int32_t ctgDropDbCacheEnqueue(SCatalog* pCtg, const char *dbFName, int64_t dbId) {
int32_t code = 0; int32_t code = 0;
SCtgCacheOperation action= {.opId = CTG_OP_DROP_DB_CACHE}; SCtgCacheOperation action= {.opId = CTG_OP_DROP_DB_CACHE};
SCtgRemoveDBMsg *msg = taosMemoryMalloc(sizeof(SCtgRemoveDBMsg)); SCtgDropDBMsg *msg = taosMemoryMalloc(sizeof(SCtgDropDBMsg));
if (NULL == msg) { if (NULL == msg) {
ctgError("malloc %d failed", (int32_t)sizeof(SCtgRemoveDBMsg)); ctgError("malloc %d failed", (int32_t)sizeof(SCtgDropDBMsg));
CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR); CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
} }
@ -590,13 +595,43 @@ _return:
CTG_RET(code); CTG_RET(code);
} }
int32_t ctgDropDbVgroupEnqueue(SCatalog* pCtg, const char *dbFName, bool syncOp) {
int32_t code = 0;
SCtgCacheOperation action= {.opId = CTG_OP_DROP_DB_VGROUP, .syncOp = syncOp};
SCtgDropDbVgroupMsg *msg = taosMemoryMalloc(sizeof(SCtgDropDbVgroupMsg));
if (NULL == msg) {
ctgError("malloc %d failed", (int32_t)sizeof(SCtgDropDbVgroupMsg));
CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
}
char *p = strchr(dbFName, '.');
if (p && CTG_IS_SYS_DBNAME(p + 1)) {
dbFName = p + 1;
}
msg->pCtg = pCtg;
strncpy(msg->dbFName, dbFName, sizeof(msg->dbFName));
action.data = msg;
CTG_ERR_JRET(ctgEnqueue(pCtg, &action));
return TSDB_CODE_SUCCESS;
_return:
taosMemoryFreeClear(action.data);
CTG_RET(code);
}
int32_t ctgDropStbMetaEnqueue(SCatalog* pCtg, const char *dbFName, int64_t dbId, const char *stbName, uint64_t suid, bool syncOp) { int32_t ctgDropStbMetaEnqueue(SCatalog* pCtg, const char *dbFName, int64_t dbId, const char *stbName, uint64_t suid, bool syncOp) {
int32_t code = 0; int32_t code = 0;
SCtgCacheOperation action= {.opId = CTG_OP_DROP_STB_META, .syncOp = syncOp}; SCtgCacheOperation action= {.opId = CTG_OP_DROP_STB_META, .syncOp = syncOp};
SCtgRemoveStbMsg *msg = taosMemoryMalloc(sizeof(SCtgRemoveStbMsg)); SCtgDropStbMetaMsg *msg = taosMemoryMalloc(sizeof(SCtgDropStbMetaMsg));
if (NULL == msg) { if (NULL == msg) {
ctgError("malloc %d failed", (int32_t)sizeof(SCtgRemoveStbMsg)); ctgError("malloc %d failed", (int32_t)sizeof(SCtgDropStbMetaMsg));
CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR); CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
} }
@ -623,9 +658,9 @@ _return:
int32_t ctgDropTbMetaEnqueue(SCatalog* pCtg, const char *dbFName, int64_t dbId, const char *tbName, bool syncOp) { int32_t ctgDropTbMetaEnqueue(SCatalog* pCtg, const char *dbFName, int64_t dbId, const char *tbName, bool syncOp) {
int32_t code = 0; int32_t code = 0;
SCtgCacheOperation action= {.opId = CTG_OP_DROP_TB_META, .syncOp = syncOp}; SCtgCacheOperation action= {.opId = CTG_OP_DROP_TB_META, .syncOp = syncOp};
SCtgRemoveTblMsg *msg = taosMemoryMalloc(sizeof(SCtgRemoveTblMsg)); SCtgDropTblMetaMsg *msg = taosMemoryMalloc(sizeof(SCtgDropTblMetaMsg));
if (NULL == msg) { if (NULL == msg) {
ctgError("malloc %d failed", (int32_t)sizeof(SCtgRemoveTblMsg)); ctgError("malloc %d failed", (int32_t)sizeof(SCtgDropTblMetaMsg));
CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR); CTG_ERR_RET(TSDB_CODE_CTG_MEM_ERROR);
} }
@ -1281,7 +1316,7 @@ _return:
int32_t ctgOpDropDbCache(SCtgCacheOperation *operation) { int32_t ctgOpDropDbCache(SCtgCacheOperation *operation) {
int32_t code = 0; int32_t code = 0;
SCtgRemoveDBMsg *msg = operation->data; SCtgDropDBMsg *msg = operation->data;
SCatalog* pCtg = msg->pCtg; SCatalog* pCtg = msg->pCtg;
SCtgDBCache *dbCache = NULL; SCtgDBCache *dbCache = NULL;
@ -1304,6 +1339,33 @@ _return:
CTG_RET(code); CTG_RET(code);
} }
int32_t ctgOpDropDbVgroup(SCtgCacheOperation *operation) {
int32_t code = 0;
SCtgDropDbVgroupMsg *msg = operation->data;
SCatalog* pCtg = msg->pCtg;
SCtgDBCache *dbCache = NULL;
ctgGetDBCache(msg->pCtg, msg->dbFName, &dbCache);
if (NULL == dbCache) {
goto _return;
}
CTG_ERR_RET(ctgWAcquireVgInfo(pCtg, dbCache));
ctgFreeVgInfo(dbCache->vgInfo);
dbCache->vgInfo = NULL;
ctgDebug("db vgInfo removed, dbFName:%s", msg->dbFName);
ctgWReleaseVgInfo(dbCache);
_return:
taosMemoryFreeClear(msg);
CTG_RET(code);
}
int32_t ctgOpUpdateTbMeta(SCtgCacheOperation *operation) { int32_t ctgOpUpdateTbMeta(SCtgCacheOperation *operation) {
int32_t code = 0; int32_t code = 0;
@ -1353,7 +1415,7 @@ _return:
int32_t ctgOpDropStbMeta(SCtgCacheOperation *operation) { int32_t ctgOpDropStbMeta(SCtgCacheOperation *operation) {
int32_t code = 0; int32_t code = 0;
SCtgRemoveStbMsg *msg = operation->data; SCtgDropStbMetaMsg *msg = operation->data;
SCatalog* pCtg = msg->pCtg; SCatalog* pCtg = msg->pCtg;
SCtgDBCache *dbCache = NULL; SCtgDBCache *dbCache = NULL;
@ -1399,7 +1461,7 @@ _return:
int32_t ctgOpDropTbMeta(SCtgCacheOperation *operation) { int32_t ctgOpDropTbMeta(SCtgCacheOperation *operation) {
int32_t code = 0; int32_t code = 0;
SCtgRemoveTblMsg *msg = operation->data; SCtgDropTblMetaMsg *msg = operation->data;
SCatalog* pCtg = msg->pCtg; SCatalog* pCtg = msg->pCtg;
SCtgDBCache *dbCache = NULL; SCtgDBCache *dbCache = NULL;

18
source/libs/catalog/src/ctgDbg.c
View File

@ -132,7 +132,22 @@ void ctgdUserCallback(SMetaData* pResult, void* param, int32_t code) {
} }
} }
int32_t ctgdLaunchAsyncCall(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps, uint64_t reqId) {
/*
prepare SQL:
create database db1;
use db1;
create stable st1 (ts timestamp, f1 int) tags(t1 int);
create table tb1 using st1 tags(1);
insert into tb1 values (now, 1);
create qnode on dnode 1;
create user user1 pass "abc";
create database db2;
grant write on db2.* to user1;
create function udf1 as '/tmp/libudf1.so' outputtype int;
create aggregate function udf2 as '/tmp/libudf2.so' outputtype int;
*/
int32_t ctgdLaunchAsyncCall(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps, uint64_t reqId, bool forceUpdate) {
int32_t code = 0; int32_t code = 0;
SCatalogReq req = {0}; SCatalogReq req = {0};
req.pTableMeta = taosArrayInit(2, sizeof(SName)); req.pTableMeta = taosArrayInit(2, sizeof(SName));
@ -144,6 +159,7 @@ int32_t ctgdLaunchAsyncCall(SCatalog* pCtg, void *pTrans, const SEpSet* pMgmtEps
req.pIndex = NULL;//taosArrayInit(2, TSDB_INDEX_FNAME_LEN); req.pIndex = NULL;//taosArrayInit(2, TSDB_INDEX_FNAME_LEN);
req.pUser = taosArrayInit(2, sizeof(SUserAuthInfo)); req.pUser = taosArrayInit(2, sizeof(SUserAuthInfo));
req.qNodeRequired = true; req.qNodeRequired = true;
req.forceUpdate = forceUpdate;
SName name = {0}; SName name = {0};
char dbFName[TSDB_DB_FNAME_LEN] = {0}; char dbFName[TSDB_DB_FNAME_LEN] = {0};

3
source/libs/index/src/indexFilter.c
View File

@ -38,7 +38,7 @@ typedef struct SIFParam {
col_id_t colId; col_id_t colId;
int64_t suid; // add later int64_t suid; // add later
char dbName[TSDB_DB_NAME_LEN]; char dbName[TSDB_DB_NAME_LEN];
char colName[TSDB_COL_NAME_LEN]; char colName[TSDB_COL_NAME_LEN * 2 + 4];
SIndexMetaArg arg; SIndexMetaArg arg;
} SIFParam; } SIFParam;
@ -171,6 +171,7 @@ static int32_t sifInitJsonParam(SNode *node, SIFParam *param, SIFCtx *ctx) {
param->colId = l->colId; param->colId = l->colId;
param->colValType = l->node.resType.type; param->colValType = l->node.resType.type;
memcpy(param->dbName, l->dbName, sizeof(l->dbName)); memcpy(param->dbName, l->dbName, sizeof(l->dbName));
sprintf(param->colName, "%s_%s", l->colName, r->literal); sprintf(param->colName, "%s_%s", l->colName, r->literal);
param->colValType = r->typeData; param->colValType = r->typeData;
return 0; return 0;

8
source/libs/parser/src/parInsert.c
View File

@ -1006,7 +1006,7 @@ static int32_t parseTagsClause(SInsertParseContext* pCxt, SSchema* pSchema, uint
} }
SSchema* pTagSchema = &pSchema[pCxt->tags.boundColumns[i]]; SSchema* pTagSchema = &pSchema[pCxt->tags.boundColumns[i]];
char* tmpTokenBuf = taosMemoryCalloc(1, sToken.n); // this can be optimize with parse column char* tmpTokenBuf = taosMemoryCalloc(1, sToken.n); //todo this can be optimize with parse column
code = checkAndTrimValue(&sToken, tmpTokenBuf, &pCxt->msg); code = checkAndTrimValue(&sToken, tmpTokenBuf, &pCxt->msg);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
taosMemoryFree(tmpTokenBuf); taosMemoryFree(tmpTokenBuf);
@ -1018,7 +1018,11 @@ static int32_t parseTagsClause(SInsertParseContext* pCxt, SSchema* pSchema, uint
taosMemoryFree(tmpTokenBuf); taosMemoryFree(tmpTokenBuf);
goto end; goto end;
} }
code = parseJsontoTagData(sToken.z, pTagVals, &pTag, &pCxt->msg); if(isNullStr(&sToken)) {
code = tTagNew(pTagVals, 1, true, &pTag);
} else {
code = parseJsontoTagData(sToken.z, pTagVals, &pTag, &pCxt->msg);
}
taosMemoryFree(tmpTokenBuf); taosMemoryFree(tmpTokenBuf);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
goto end; goto end;

5
source/libs/qworker/inc/qwInt.h
View File

@ -34,6 +34,7 @@ extern "C" {
#define QW_DEFAULT_SCH_TASK_NUMBER 10000 #define QW_DEFAULT_SCH_TASK_NUMBER 10000
#define QW_DEFAULT_SHORT_RUN_TIMES 2 #define QW_DEFAULT_SHORT_RUN_TIMES 2
#define QW_DEFAULT_HEARTBEAT_MSEC 5000 #define QW_DEFAULT_HEARTBEAT_MSEC 5000
#define QW_SCH_TIMEOUT_MSEC 180000
enum { enum {
QW_PHASE_PRE_QUERY = 1, QW_PHASE_PRE_QUERY = 1,
@ -137,7 +138,7 @@ typedef struct SQWTaskCtx {
} SQWTaskCtx; } SQWTaskCtx;
typedef struct SQWSchStatus { typedef struct SQWSchStatus {
int32_t lastAccessTs; // timestamp in second int64_t hbBrokenTs; // timestamp in msecond
SRWLatch hbConnLock; SRWLatch hbConnLock;
SRpcHandleInfo hbConnInfo; SRpcHandleInfo hbConnInfo;
SQueryNodeEpId hbEpId; SQueryNodeEpId hbEpId;
@ -354,6 +355,8 @@ int32_t qwOpenRef(void);
void qwSetHbParam(int64_t refId, SQWHbParam **pParam); void qwSetHbParam(int64_t refId, SQWHbParam **pParam);
int32_t qwUpdateTimeInQueue(SQWorker *mgmt, int64_t ts, EQueueType type); int32_t qwUpdateTimeInQueue(SQWorker *mgmt, int64_t ts, EQueueType type);
int64_t qwGetTimeInQueue(SQWorker *mgmt, EQueueType type); int64_t qwGetTimeInQueue(SQWorker *mgmt, EQueueType type);
void qwClearExpiredSch(SArray* pExpiredSch);
int32_t qwAcquireScheduler(SQWorker *mgmt, uint64_t sId, int32_t rwType, SQWSchStatus **sch);
void qwDbgDumpMgmtInfo(SQWorker *mgmt); void qwDbgDumpMgmtInfo(SQWorker *mgmt);
int32_t qwDbgValidateStatus(QW_FPARAMS_DEF, int8_t oriStatus, int8_t newStatus, bool *ignore); int32_t qwDbgValidateStatus(QW_FPARAMS_DEF, int8_t oriStatus, int8_t newStatus, bool *ignore);

6
source/libs/qworker/src/qwUtil.c
View File

@ -535,3 +535,9 @@ int64_t qwGetTimeInQueue(SQWorker *mgmt, EQueueType type) {
return -1; return -1;
} }
void qwClearExpiredSch(SArray* pExpiredSch) {
}

21
source/libs/qworker/src/qworker.c
View File

@ -21,10 +21,12 @@ int32_t qwProcessHbLinkBroken(SQWorker *mgmt, SQWMsg *qwMsg, SSchedulerHbReq *re
SSchedulerHbRsp rsp = {0}; SSchedulerHbRsp rsp = {0};
SQWSchStatus *sch = NULL; SQWSchStatus *sch = NULL;
QW_ERR_RET(qwAcquireAddScheduler(mgmt, req->sId, QW_READ, &sch)); QW_ERR_RET(qwAcquireScheduler(mgmt, req->sId, QW_READ, &sch));
QW_LOCK(QW_WRITE, &sch->hbConnLock); QW_LOCK(QW_WRITE, &sch->hbConnLock);
sch->hbBrokenTs = taosGetTimestampMs();
if (qwMsg->connInfo.handle == sch->hbConnInfo.handle) { if (qwMsg->connInfo.handle == sch->hbConnInfo.handle) {
tmsgReleaseHandle(&sch->hbConnInfo, TAOS_CONN_SERVER); tmsgReleaseHandle(&sch->hbConnInfo, TAOS_CONN_SERVER);
sch->hbConnInfo.handle = NULL; sch->hbConnInfo.handle = NULL;
@ -794,6 +796,7 @@ void qwProcessHbTimerEvent(void *param, void *tmrId) {
SQWSchStatus *sch = NULL; SQWSchStatus *sch = NULL;
int32_t taskNum = 0; int32_t taskNum = 0;
SQWHbInfo *rspList = NULL; SQWHbInfo *rspList = NULL;
SArray *pExpiredSch = NULL;
int32_t code = 0; int32_t code = 0;
qwDbgDumpMgmtInfo(mgmt); qwDbgDumpMgmtInfo(mgmt);
@ -809,8 +812,11 @@ void qwProcessHbTimerEvent(void *param, void *tmrId) {
} }
rspList = taosMemoryCalloc(schNum, sizeof(SQWHbInfo)); rspList = taosMemoryCalloc(schNum, sizeof(SQWHbInfo));
if (NULL == rspList) { pExpiredSch = taosArrayInit(schNum, sizeof(uint64_t));
if (NULL == rspList || NULL == pExpiredSch) {
QW_UNLOCK(QW_READ, &mgmt->schLock); QW_UNLOCK(QW_READ, &mgmt->schLock);
taosMemoryFree(rspList);
taosArrayDestroy(pExpiredSch);
QW_ELOG("calloc %d SQWHbInfo failed", schNum); QW_ELOG("calloc %d SQWHbInfo failed", schNum);
taosTmrReset(qwProcessHbTimerEvent, QW_DEFAULT_HEARTBEAT_MSEC, param, mgmt->timer, &mgmt->hbTimer); taosTmrReset(qwProcessHbTimerEvent, QW_DEFAULT_HEARTBEAT_MSEC, param, mgmt->timer, &mgmt->hbTimer);
qwRelease(refId); qwRelease(refId);
@ -820,6 +826,7 @@ void qwProcessHbTimerEvent(void *param, void *tmrId) {
void *key = NULL; void *key = NULL;
size_t keyLen = 0; size_t keyLen = 0;
int32_t i = 0; int32_t i = 0;
int64_t currentMs = taosGetTimestampMs();
void *pIter = taosHashIterate(mgmt->schHash, NULL); void *pIter = taosHashIterate(mgmt->schHash, NULL);
while (pIter) { while (pIter) {
@ -827,6 +834,11 @@ void qwProcessHbTimerEvent(void *param, void *tmrId) {
if (NULL == sch->hbConnInfo.handle) { if (NULL == sch->hbConnInfo.handle) {
uint64_t *sId = taosHashGetKey(pIter, NULL); uint64_t *sId = taosHashGetKey(pIter, NULL);
QW_TLOG("cancel send hb to sch %" PRIx64 " cause of no connection handle", *sId); QW_TLOG("cancel send hb to sch %" PRIx64 " cause of no connection handle", *sId);
if (sch->hbBrokenTs > 0 && ((currentMs - sch->hbBrokenTs) > QW_SCH_TIMEOUT_MSEC) && taosHashGetSize(sch->tasksHash) <= 0) {
taosArrayPush(pExpiredSch, sId);
}
pIter = taosHashIterate(mgmt->schHash, pIter); pIter = taosHashIterate(mgmt->schHash, pIter);
continue; continue;
} }
@ -852,7 +864,12 @@ _return:
tFreeSSchedulerHbRsp(&rspList[j].rsp); tFreeSSchedulerHbRsp(&rspList[j].rsp);
} }
if (taosArrayGetSize(pExpiredSch) > 0) {
qwClearExpiredSch(pExpiredSch);
}
taosMemoryFreeClear(rspList); taosMemoryFreeClear(rspList);
taosArrayDestroy(pExpiredSch);
taosTmrReset(qwProcessHbTimerEvent, QW_DEFAULT_HEARTBEAT_MSEC, param, mgmt->timer, &mgmt->hbTimer); taosTmrReset(qwProcessHbTimerEvent, QW_DEFAULT_HEARTBEAT_MSEC, param, mgmt->timer, &mgmt->hbTimer);
qwRelease(refId); qwRelease(refId);

4
source/libs/scheduler/inc/schedulerInt.h
View File

@ -102,11 +102,11 @@ typedef struct SSchedulerMgmt {
uint64_t taskId; // sequential taksId uint64_t taskId; // sequential taksId
uint64_t sId; // schedulerId uint64_t sId; // schedulerId
SSchedulerCfg cfg; SSchedulerCfg cfg;
SRWLatch lock;
bool exit; bool exit;
int32_t jobRef; int32_t jobRef;
int32_t jobNum; int32_t jobNum;
SSchStat stat; SSchStat stat;
SRWLatch hbLock;
SHashObj *hbConnections; SHashObj *hbConnections;
} SSchedulerMgmt; } SSchedulerMgmt;
@ -320,6 +320,8 @@ extern SSchedulerMgmt schMgmt;
#define SCH_UNLOCK(type, _lock) (SCH_READ == (type) ? taosRUnLockLatch(_lock) : taosWUnLockLatch(_lock)) #define SCH_UNLOCK(type, _lock) (SCH_READ == (type) ? taosRUnLockLatch(_lock) : taosWUnLockLatch(_lock))
void schDeregisterTaskHb(SSchJob *pJob, SSchTask *pTask);
void schCleanClusterHb(void* pTrans);
int32_t schLaunchTask(SSchJob *job, SSchTask *task); int32_t schLaunchTask(SSchJob *job, SSchTask *task);
int32_t schBuildAndSendMsg(SSchJob *job, SSchTask *task, SQueryNodeAddr *addr, int32_t msgType); int32_t schBuildAndSendMsg(SSchJob *job, SSchTask *task, SQueryNodeAddr *addr, int32_t msgType);
SSchJob *schAcquireJob(int64_t refId); SSchJob *schAcquireJob(int64_t refId);

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

@ -126,30 +126,6 @@ _return:
SCH_RET(code); SCH_RET(code);
} }
void schDeregisterTaskHb(SSchJob *pJob, SSchTask *pTask) {
if (!pTask->registerdHb) {
return;
}
SQueryNodeAddr *addr = taosArrayGet(pTask->candidateAddrs, pTask->candidateIdx);
SQueryNodeEpId epId = {0};
epId.nodeId = addr->nodeId;
SEp* pEp = SCH_GET_CUR_EP(addr);
strcpy(epId.ep.fqdn, pEp->fqdn);
epId.ep.port = pEp->port;
SSchHbTrans *hb = taosHashGet(schMgmt.hbConnections, &epId, sizeof(SQueryNodeEpId));
if (NULL == hb) {
SCH_TASK_ELOG("nodeId %d fqdn %s port %d not in hb connections", epId.nodeId, epId.ep.fqdn, epId.ep.port);
return;
}
atomic_sub_fetch_64(&hb->taskNum, 1);
pTask->registerdHb = false;
}
void schFreeTask(SSchJob *pJob, SSchTask *pTask) { void schFreeTask(SSchJob *pJob, SSchTask *pTask) {
schDeregisterTaskHb(pJob, pTask); schDeregisterTaskHb(pJob, pTask);
@ -377,15 +353,21 @@ int32_t schDropTaskExecNode(SSchJob *pJob, SSchTask *pTask, void *handle, int32_
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
taosHashRemove(pTask->execNodes, &execIdx, sizeof(execIdx)); if (taosHashRemove(pTask->execNodes, &execIdx, sizeof(execIdx))) {
SCH_TASK_ELOG("fail to remove execIdx %d from execNodeList", execIdx);
} else {
SCH_TASK_DLOG("execIdx %d removed from execNodeList", execIdx);
}
if (execIdx != pTask->execIdx) { // ignore it if (execIdx != pTask->execIdx) { // ignore it
SCH_TASK_DLOG("execIdx %d is not current execIdx %d", execIdx, pTask->execIdx);
SCH_RET(TSDB_CODE_SCH_IGNORE_ERROR); SCH_RET(TSDB_CODE_SCH_IGNORE_ERROR);
} }
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int32_t schUpdateTaskExecNode(SSchTask *pTask, void *handle, int32_t execIdx) { int32_t schUpdateTaskExecNode(SSchJob *pJob, SSchTask *pTask, void *handle, int32_t execIdx) {
if (taosHashGetSize(pTask->execNodes) <= 0) { if (taosHashGetSize(pTask->execNodes) <= 0) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -393,6 +375,8 @@ int32_t schUpdateTaskExecNode(SSchTask *pTask, void *handle, int32_t execIdx) {
SSchNodeInfo *nodeInfo = taosHashGet(pTask->execNodes, &execIdx, sizeof(execIdx)); SSchNodeInfo *nodeInfo = taosHashGet(pTask->execNodes, &execIdx, sizeof(execIdx));
nodeInfo->handle = handle; nodeInfo->handle = handle;
SCH_TASK_DLOG("handle updated to %p for execIdx %d", handle, execIdx);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -403,7 +387,7 @@ int32_t schUpdateTaskHandle(SSchJob *pJob, SSchTask *pTask, bool dropExecNode, v
SCH_SET_TASK_HANDLE(pTask, handle); SCH_SET_TASK_HANDLE(pTask, handle);
schUpdateTaskExecNode(pTask, handle, execIdx); schUpdateTaskExecNode(pJob, pTask, handle, execIdx);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -551,6 +535,8 @@ int32_t schSetAddrsFromNodeList(SSchJob *pJob, SSchTask *pTask) {
SCH_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY); SCH_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY);
} }
SCH_TASK_DLOG("set %dth condidate addr, id %d, fqdn:%s, port:%d", i, naddr->nodeId, SCH_GET_CUR_EP(naddr)->fqdn, SCH_GET_CUR_EP(naddr)->port);
++addNum; ++addNum;
} }
} }
@ -1110,6 +1096,7 @@ int32_t schProcessOnTaskSuccess(SSchJob *pJob, SSchTask *pTask) {
SCH_UNLOCK(SCH_WRITE, &parent->lock); SCH_UNLOCK(SCH_WRITE, &parent->lock);
if (SCH_TASK_READY_FOR_LAUNCH(readyNum, parent)) { if (SCH_TASK_READY_FOR_LAUNCH(readyNum, parent)) {
SCH_TASK_DLOG("all %d children task done, start to launch parent task %" PRIx64, readyNum, parent->taskId);
SCH_ERR_RET(schLaunchTask(pJob, parent)); SCH_ERR_RET(schLaunchTask(pJob, parent));
} }
} }
@ -1186,7 +1173,7 @@ void schDropTaskOnExecNode(SSchJob *pJob, SSchTask *pTask) {
nodeInfo = taosHashIterate(pTask->execNodes, nodeInfo); nodeInfo = taosHashIterate(pTask->execNodes, nodeInfo);
} }
SCH_TASK_DLOG("task has %d exec address", size); SCH_TASK_DLOG("task has been dropped on %d exec nodes", size);
} }
@ -1196,7 +1183,8 @@ int32_t schRescheduleTask(SSchJob *pJob, SSchTask *pTask) {
} }
SCH_LOCK_TASK(pTask); SCH_LOCK_TASK(pTask);
if (JOB_TASK_STATUS_EXECUTING == pTask->status && pJob->fetchTask != pTask) { if (JOB_TASK_STATUS_EXECUTING == pTask->status && pJob->fetchTask != pTask && taosArrayGetSize(pTask->candidateAddrs) > 1) {
SCH_TASK_DLOG("task execIdx %d will be rescheduled now", pTask->execIdx);
schDropTaskOnExecNode(pJob, pTask); schDropTaskOnExecNode(pJob, pTask);
taosHashClear(pTask->execNodes); taosHashClear(pTask->execNodes);
schProcessOnTaskFailure(pJob, pTask, TSDB_CODE_SCH_TIMEOUT_ERROR); schProcessOnTaskFailure(pJob, pTask, TSDB_CODE_SCH_TIMEOUT_ERROR);
@ -1306,9 +1294,10 @@ int32_t schLaunchTaskImpl(SSchJob *pJob, SSchTask *pTask) {
int32_t code = 0; int32_t code = 0;
atomic_add_fetch_32(&pTask->level->taskLaunchedNum, 1); atomic_add_fetch_32(&pTask->level->taskLaunchedNum, 1);
pTask->execIdx++; pTask->execIdx++;
SCH_TASK_DLOG("start to launch task's %dth exec", pTask->execIdx);
SCH_LOG_TASK_START_TS(pTask); SCH_LOG_TASK_START_TS(pTask);
if (schJobNeedToStop(pJob, &status)) { if (schJobNeedToStop(pJob, &status)) {
@ -1471,9 +1460,10 @@ void schFreeJobImpl(void *job) {
qDebug("QID:0x%" PRIx64 " job freed, refId:%" PRIx64 ", pointer:%p", queryId, refId, pJob); qDebug("QID:0x%" PRIx64 " job freed, refId:%" PRIx64 ", pointer:%p", queryId, refId, pJob);
atomic_sub_fetch_32(&schMgmt.jobNum, 1); int32_t jobNum = atomic_sub_fetch_32(&schMgmt.jobNum, 1);
if (jobNum == 0) {
schCloseJobRef(); schCloseJobRef();
}
} }
int32_t schExecJobImpl(void *pTrans, SArray *pNodeList, SQueryPlan *pDag, int64_t *job, const char *sql, int32_t schExecJobImpl(void *pTrans, SArray *pNodeList, SQueryPlan *pDag, int64_t *job, const char *sql,

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

@ -648,31 +648,6 @@ _return:
SCH_RET(code); SCH_RET(code);
} }
int32_t schRegisterHbConnection(SSchJob *pJob, SSchTask *pTask, SQueryNodeEpId *epId, bool *exist) {
int32_t code = 0;
SSchHbTrans hb = {0};
hb.trans.pTrans = pJob->pTrans;
SCH_ERR_RET(schMakeHbRpcCtx(pJob, pTask, &hb.rpcCtx));
code = taosHashPut(schMgmt.hbConnections, epId, sizeof(SQueryNodeEpId), &hb, sizeof(SSchHbTrans));
if (code) {
schFreeRpcCtx(&hb.rpcCtx);
if (HASH_NODE_EXIST(code)) {
*exist = true;
return TSDB_CODE_SUCCESS;
}
qError("taosHashPut hb trans failed, nodeId:%d, fqdn:%s, port:%d", epId->nodeId, epId->ep.fqdn, epId->ep.port);
SCH_ERR_RET(code);
}
return TSDB_CODE_SUCCESS;
}
int32_t schBuildAndSendHbMsg(SQueryNodeEpId *nodeEpId, SArray* taskAction) { int32_t schBuildAndSendHbMsg(SQueryNodeEpId *nodeEpId, SArray* taskAction) {
SSchedulerHbReq req = {0}; SSchedulerHbReq req = {0};
int32_t code = 0; int32_t code = 0;
@ -684,17 +659,20 @@ int32_t schBuildAndSendHbMsg(SQueryNodeEpId *nodeEpId, SArray* taskAction) {
req.sId = schMgmt.sId; req.sId = schMgmt.sId;
memcpy(&req.epId, nodeEpId, sizeof(SQueryNodeEpId)); memcpy(&req.epId, nodeEpId, sizeof(SQueryNodeEpId));
SCH_LOCK(SCH_READ, &schMgmt.hbLock);
SSchHbTrans *hb = taosHashGet(schMgmt.hbConnections, nodeEpId, sizeof(SQueryNodeEpId)); SSchHbTrans *hb = taosHashGet(schMgmt.hbConnections, nodeEpId, sizeof(SQueryNodeEpId));
if (NULL == hb) { if (NULL == hb) {
qError("taosHashGet hb connection failed, nodeId:%d, fqdn:%s, port:%d", nodeEpId->nodeId, nodeEpId->ep.fqdn, SCH_UNLOCK(SCH_READ, &schMgmt.hbLock);
qError("hb connection no longer exist, nodeId:%d, fqdn:%s, port:%d", nodeEpId->nodeId, nodeEpId->ep.fqdn,
nodeEpId->ep.port); nodeEpId->ep.port);
SCH_ERR_RET(code); return TSDB_CODE_SUCCESS;
} }
SCH_LOCK(SCH_WRITE, &hb->lock); SCH_LOCK(SCH_WRITE, &hb->lock);
code = schCloneHbRpcCtx(&hb->rpcCtx, &rpcCtx); code = schCloneHbRpcCtx(&hb->rpcCtx, &rpcCtx);
memcpy(&trans, &hb->trans, sizeof(trans)); memcpy(&trans, &hb->trans, sizeof(trans));
SCH_UNLOCK(SCH_WRITE, &hb->lock); SCH_UNLOCK(SCH_WRITE, &hb->lock);
SCH_UNLOCK(SCH_READ, &schMgmt.hbLock);
SCH_ERR_RET(code); SCH_ERR_RET(code);
@ -764,60 +742,6 @@ _return:
SCH_RET(code); SCH_RET(code);
} }
int32_t schEnsureHbConnection(SSchJob *pJob, SSchTask *pTask) {
SQueryNodeAddr *addr = taosArrayGet(pTask->candidateAddrs, pTask->candidateIdx);
SQueryNodeEpId epId = {0};
epId.nodeId = addr->nodeId;
SEp* pEp = SCH_GET_CUR_EP(addr);
strcpy(epId.ep.fqdn, pEp->fqdn);
epId.ep.port = pEp->port;
SSchHbTrans *hb = NULL;
while (true) {
hb = taosHashGet(schMgmt.hbConnections, &epId, sizeof(SQueryNodeEpId));
if (NULL == hb) {
bool exist = false;
SCH_ERR_RET(schRegisterHbConnection(pJob, pTask, &epId, &exist));
if (!exist) {
SCH_ERR_RET(schBuildAndSendHbMsg(&epId, NULL));
}
continue;
}
break;
}
atomic_add_fetch_64(&hb->taskNum, 1);
pTask->registerdHb = true;
return TSDB_CODE_SUCCESS;
}
int32_t schUpdateHbConnection(SQueryNodeEpId *epId, SSchTrans *trans) {
int32_t code = 0;
SSchHbTrans *hb = NULL;
hb = taosHashGet(schMgmt.hbConnections, epId, sizeof(SQueryNodeEpId));
if (NULL == hb) {
qError("taosHashGet hb connection failed, nodeId:%d, fqdn:%s, port:%d", epId->nodeId, epId->ep.fqdn, epId->ep.port);
SCH_ERR_RET(TSDB_CODE_QRY_APP_ERROR);
}
SCH_LOCK(SCH_WRITE, &hb->lock);
memcpy(&hb->trans, trans, sizeof(*trans));
SCH_UNLOCK(SCH_WRITE, &hb->lock);
qDebug("hb connection updated, sId:%" PRIx64 ", nodeId:%d, fqdn:%s, port:%d, pTrans:%p, pHandle:%p", schMgmt.sId,
epId->nodeId, epId->ep.fqdn, epId->ep.port, trans->pTrans, trans->pHandle);
return TSDB_CODE_SUCCESS;
}
int32_t schHandleHbCallback(void *param, const SDataBuf *pMsg, int32_t code) { int32_t schHandleHbCallback(void *param, const SDataBuf *pMsg, int32_t code) {
SSchedulerHbRsp rsp = {0}; SSchedulerHbRsp rsp = {0};
SSchTaskCallbackParam *pParam = (SSchTaskCallbackParam *)param; SSchTaskCallbackParam *pParam = (SSchTaskCallbackParam *)param;
@ -1037,6 +961,7 @@ int32_t schBuildAndSendMsg(SSchJob *pJob, SSchTask *pTask, SQueryNodeAddr *addr,
if (NULL == addr) { if (NULL == addr) {
addr = taosArrayGet(pTask->candidateAddrs, pTask->candidateIdx); addr = taosArrayGet(pTask->candidateAddrs, pTask->candidateIdx);
isCandidateAddr = true; isCandidateAddr = true;
SCH_TASK_DLOG("target candidateIdx %d", pTask->candidateIdx);
} }
SEpSet epSet = addr->epSet; SEpSet epSet = addr->epSet;

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

@ -21,17 +21,189 @@
#include "tref.h" #include "tref.h"
#include "trpc.h" #include "trpc.h"
void schCleanClusterHb(void* pTrans) {
SCH_LOCK(SCH_WRITE, &schMgmt.hbLock);
SSchHbTrans *hb = taosHashIterate(schMgmt.hbConnections, NULL);
while (hb) {
if (hb->trans.pTrans == pTrans) {
SQueryNodeEpId* pEpId = taosHashGetKey(hb, NULL);
rpcReleaseHandle(hb->trans.pHandle, TAOS_CONN_CLIENT);
taosHashRemove(schMgmt.hbConnections, pEpId, sizeof(SQueryNodeEpId));
}
hb = taosHashIterate(schMgmt.hbConnections, hb);
}
SCH_UNLOCK(SCH_WRITE, &schMgmt.hbLock);
}
int32_t schRemoveHbConnection(SSchJob *pJob, SSchTask *pTask, SQueryNodeEpId *epId) {
return TSDB_CODE_SUCCESS; // TODO ENABLE IT WHEN RPC IS READY
int32_t code = 0;
SCH_LOCK(SCH_WRITE, &schMgmt.hbLock);
SSchHbTrans *hb = taosHashGet(schMgmt.hbConnections, epId, sizeof(SQueryNodeEpId));
if (NULL == hb) {
SCH_UNLOCK(SCH_WRITE, &schMgmt.hbLock);
SCH_TASK_ELOG("nodeId %d fqdn %s port %d not in hb connections", epId->nodeId, epId->ep.fqdn, epId->ep.port);
return TSDB_CODE_SUCCESS;
}
int64_t taskNum = atomic_load_64(&hb->taskNum);
if (taskNum <= 0) {
rpcReleaseHandle(hb->trans.pHandle, TAOS_CONN_CLIENT);
taosHashRemove(schMgmt.hbConnections, epId, sizeof(SQueryNodeEpId));
}
SCH_UNLOCK(SCH_WRITE, &schMgmt.hbLock);
return TSDB_CODE_SUCCESS;
}
int32_t schAddHbConnection(SSchJob *pJob, SSchTask *pTask, SQueryNodeEpId *epId, bool *exist) {
int32_t code = 0;
SSchHbTrans hb = {0};
hb.trans.pTrans = pJob->pTrans;
hb.taskNum = 1;
SCH_ERR_RET(schMakeHbRpcCtx(pJob, pTask, &hb.rpcCtx));
SCH_LOCK(SCH_WRITE, &schMgmt.hbLock);
code = taosHashPut(schMgmt.hbConnections, epId, sizeof(SQueryNodeEpId), &hb, sizeof(SSchHbTrans));
if (code) {
SCH_UNLOCK(SCH_WRITE, &schMgmt.hbLock);
schFreeRpcCtx(&hb.rpcCtx);
if (HASH_NODE_EXIST(code)) {
*exist = true;
return TSDB_CODE_SUCCESS;
}
qError("taosHashPut hb trans failed, nodeId:%d, fqdn:%s, port:%d", epId->nodeId, epId->ep.fqdn, epId->ep.port);
SCH_ERR_RET(code);
}
SCH_UNLOCK(SCH_WRITE, &schMgmt.hbLock);
return TSDB_CODE_SUCCESS;
}
int32_t schRegisterHbConnection(SSchJob *pJob, SSchTask *pTask, SQueryNodeEpId *pEpId) {
SSchHbTrans *hb = NULL;
while (true) {
SCH_LOCK(SCH_READ, &schMgmt.hbLock);
hb = taosHashGet(schMgmt.hbConnections, pEpId, sizeof(SQueryNodeEpId));
if (NULL == hb) {
bool exist = false;
SCH_UNLOCK(SCH_READ, &schMgmt.hbLock);
SCH_ERR_RET(schAddHbConnection(pJob, pTask, pEpId, &exist));
if (!exist) {
SCH_RET(schBuildAndSendHbMsg(pEpId, NULL));
}
continue;
}
break;
}
atomic_add_fetch_64(&hb->taskNum, 1);
SCH_UNLOCK(SCH_READ, &schMgmt.hbLock);
return TSDB_CODE_SUCCESS;
}
void schDeregisterTaskHb(SSchJob *pJob, SSchTask *pTask) {
if (!pTask->registerdHb) {
return;
}
SQueryNodeAddr *addr = taosArrayGet(pTask->candidateAddrs, pTask->candidateIdx);
SQueryNodeEpId epId = {0};
epId.nodeId = addr->nodeId;
SEp* pEp = SCH_GET_CUR_EP(addr);
strcpy(epId.ep.fqdn, pEp->fqdn);
epId.ep.port = pEp->port;
SCH_LOCK(SCH_READ, &schMgmt.hbLock);
SSchHbTrans *hb = taosHashGet(schMgmt.hbConnections, &epId, sizeof(SQueryNodeEpId));
if (NULL == hb) {
SCH_UNLOCK(SCH_READ, &schMgmt.hbLock);
SCH_TASK_WLOG("nodeId %d fqdn %s port %d not in hb connections", epId.nodeId, epId.ep.fqdn, epId.ep.port);
return;
}
int64_t taskNum = atomic_sub_fetch_64(&hb->taskNum, 1);
if (0 == taskNum) {
SCH_UNLOCK(SCH_READ, &schMgmt.hbLock);
schRemoveHbConnection(pJob, pTask, &epId);
} else {
SCH_UNLOCK(SCH_READ, &schMgmt.hbLock);
}
pTask->registerdHb = false;
}
int32_t schEnsureHbConnection(SSchJob *pJob, SSchTask *pTask) {
SQueryNodeAddr *addr = taosArrayGet(pTask->candidateAddrs, pTask->candidateIdx);
SQueryNodeEpId epId = {0};
epId.nodeId = addr->nodeId;
SEp* pEp = SCH_GET_CUR_EP(addr);
strcpy(epId.ep.fqdn, pEp->fqdn);
epId.ep.port = pEp->port;
SCH_ERR_RET(schRegisterHbConnection(pJob, pTask, &epId));
pTask->registerdHb = true;
return TSDB_CODE_SUCCESS;
}
int32_t schUpdateHbConnection(SQueryNodeEpId *epId, SSchTrans *trans) {
int32_t code = 0;
SSchHbTrans *hb = NULL;
SCH_LOCK(SCH_READ, &schMgmt.hbLock);
hb = taosHashGet(schMgmt.hbConnections, epId, sizeof(SQueryNodeEpId));
if (NULL == hb) {
SCH_UNLOCK(SCH_READ, &schMgmt.hbLock);
qError("taosHashGet hb connection failed, nodeId:%d, fqdn:%s, port:%d", epId->nodeId, epId->ep.fqdn, epId->ep.port);
SCH_ERR_RET(TSDB_CODE_QRY_APP_ERROR);
}
SCH_LOCK(SCH_WRITE, &hb->lock);
memcpy(&hb->trans, trans, sizeof(*trans));
SCH_UNLOCK(SCH_WRITE, &hb->lock);
SCH_UNLOCK(SCH_READ, &schMgmt.hbLock);
qDebug("hb connection updated, sId:%" PRIx64 ", nodeId:%d, fqdn:%s, port:%d, pTrans:%p, pHandle:%p", schMgmt.sId,
epId->nodeId, epId->ep.fqdn, epId->ep.port, trans->pTrans, trans->pHandle);
return TSDB_CODE_SUCCESS;
}
void schCloseJobRef(void) { void schCloseJobRef(void) {
if (!atomic_load_8((int8_t *)&schMgmt.exit)) { if (!atomic_load_8((int8_t *)&schMgmt.exit)) {
return; return;
} }
SCH_LOCK(SCH_WRITE, &schMgmt.lock); if (schMgmt.jobRef >= 0) {
if (atomic_load_32(&schMgmt.jobNum) <= 0 && schMgmt.jobRef >= 0) {
taosCloseRef(schMgmt.jobRef); taosCloseRef(schMgmt.jobRef);
schMgmt.jobRef = -1; schMgmt.jobRef = -1;
} }
SCH_UNLOCK(SCH_WRITE, &schMgmt.lock);
} }
uint64_t schGenTaskId(void) { return atomic_add_fetch_64(&schMgmt.taskId, 1); } uint64_t schGenTaskId(void) { return atomic_add_fetch_64(&schMgmt.taskId, 1); }
@ -88,4 +260,3 @@ void schFreeRpcCtx(SRpcCtx *pCtx) {
(*pCtx->freeFunc)(pCtx->brokenVal.val); (*pCtx->freeFunc)(pCtx->brokenVal.val);
} }

10
source/libs/scheduler/src/scheduler.c
View File

@ -182,6 +182,14 @@ int32_t scheduleCancelJob(int64_t job) {
SCH_RET(code); SCH_RET(code);
} }
void schedulerStopQueryHb(void *pTrans) {
if (NULL == pTrans) {
return;
}
schCleanClusterHb(pTrans);
}
void schedulerFreeJob(int64_t job) { void schedulerFreeJob(int64_t job) {
SSchJob *pJob = schAcquireJob(job); SSchJob *pJob = schAcquireJob(job);
if (NULL == pJob) { if (NULL == pJob) {
@ -220,6 +228,7 @@ void schedulerDestroy(void) {
} }
} }
SCH_LOCK(SCH_WRITE, &schMgmt.hbLock);
if (schMgmt.hbConnections) { if (schMgmt.hbConnections) {
void *pIter = taosHashIterate(schMgmt.hbConnections, NULL); void *pIter = taosHashIterate(schMgmt.hbConnections, NULL);
while (pIter != NULL) { while (pIter != NULL) {
@ -230,4 +239,5 @@ void schedulerDestroy(void) {
taosHashCleanup(schMgmt.hbConnections); taosHashCleanup(schMgmt.hbConnections);
schMgmt.hbConnections = NULL; schMgmt.hbConnections = NULL;
} }
SCH_UNLOCK(SCH_WRITE, &schMgmt.hbLock);
} }

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

@ -180,6 +180,12 @@ static bool addHandleToAcceptloop(void* arg);
if (!transQueuePush(&conn->srvMsgs, srvMsg)) { \ if (!transQueuePush(&conn->srvMsgs, srvMsg)) { \
return; \ return; \
} \ } \
if (conn->regArg.init) { \
tTrace("server conn %p release, notify server app", conn); \
STrans* pTransInst = conn->pTransInst; \
(*pTransInst->cfp)(pTransInst->parent, &(conn->regArg.msg), NULL); \
memset(&conn->regArg, 0, sizeof(conn->regArg)); \
} \
uvStartSendRespInternal(srvMsg); \ uvStartSendRespInternal(srvMsg); \
return; \ return; \
} \ } \
@ -1154,6 +1160,10 @@ int transGetConnInfo(void* thandle, STransHandleInfo* pInfo) {
} }
SExHandle* ex = thandle; SExHandle* ex = thandle;
SSvrConn* pConn = ex->handle; SSvrConn* pConn = ex->handle;
if (pConn == NULL) {
tTrace("invalid handle %p, failed to Get Conn info", thandle);
return -1;
}
struct sockaddr_in addr = pConn->addr; struct sockaddr_in addr = pConn->addr;
pInfo->clientIp = (uint32_t)(addr.sin_addr.s_addr); pInfo->clientIp = (uint32_t)(addr.sin_addr.s_addr);

113
tests/system-test/1-insert/create_table_comment.py Normal file
View File

@ -0,0 +1,113 @@
###################################################################
# Copyright (c) 2016 by TAOS Technologies, Inc.
# All rights reserved.
#
# This file is proprietary and confidential to TAOS Technologies.
# No part of this file may be reproduced, stored, transmitted,
# disclosed or used in any form or by any means other than as
# expressly provided by the written permission from Jianhui Tao
#
###################################################################
# -*- coding: utf-8 -*-
import random
import string
from util.log import *
from util.cases import *
from util.sql import *
class TDTestCase:
def init(self, conn, logSql):
tdLog.debug("start to execute %s" % __file__)
tdSql.init(conn.cursor())
def get_long_name(self, length, mode="mixed"):
"""
generate long name
mode could be numbers/letters/letters_mixed/mixed
"""
if mode == "numbers":
population = string.digits
elif mode == "letters":
population = string.ascii_letters.lower()
elif mode == "letters_mixed":
population = string.ascii_letters.upper() + string.ascii_letters.lower()
else:
population = string.ascii_letters.lower() + string.digits
return "".join(random.choices(population, k=length))
def __create_tb(self,dbname,stbname,tbname,comment):
tdSql.execute(f'create database if not exists {dbname}')
tdSql.execute(f'use {dbname}')
tdSql.execute(
f'create table {stbname} (ts timestamp,c0 int) tags(t0 int) ')
tdSql.execute(
f'create table {tbname} using {stbname} tags(1) comment "{comment}"')
def __create_normaltb(self,dbname,tbname,comment):
tdSql.execute(f'create database if not exists {dbname}')
tdSql.execute(f'use {dbname}')
tdSql.execute(
f'create table {tbname} (ts timestamp,c0 int) comment "{comment}"')
def check_comment(self):
dbname = self.get_long_name(length=10, mode="letters")
ntbname = self.get_long_name(length=5, mode="letters")
# create normal table with comment
comment = self.get_long_name(length=10, mode="letters")
self.__create_normaltb(dbname,ntbname,comment)
ntb_kv_list = tdSql.getResult("show tables")
print(ntb_kv_list)
tdSql.checkEqual(ntb_kv_list[0][8], comment)
tdSql.error('alter table {ntbname} comment "test1"')
tdSql.execute(f'drop database {dbname}')
# max length(1024)
comment = self.get_long_name(length=1024, mode="letters")
self.__create_normaltb(dbname,ntbname,comment)
ntb_kv_list = tdSql.getResult("show tables")
tdSql.checkEqual(ntb_kv_list[0][8], comment)
tdSql.execute(f'drop database {dbname}')
# error overlength
comment = self.get_long_name(length=1025, mode="letters")
tdSql.execute(f'create database if not exists {dbname}')
tdSql.execute(f'use {dbname}')
tdSql.error(f"create table ntb (ts timestamp,c0 int) comment '{comment}'")
tdSql.execute(f'drop database {dbname}')
# create child table with comment
comment = self.get_long_name(length=10, mode="letters")
stbname = self.get_long_name(length=5, mode="letters")
tbname = self.get_long_name(length=3, mode="letters")
self.__create_tb(dbname,stbname,tbname,comment)
ntb_kv_list = tdSql.getResult("show tables")
tdSql.checkEqual(ntb_kv_list[0][8], comment)
tdSql.error(f'alter table {tbname} comment "test1"')
tdSql.execute(f'drop database {dbname}')
# max length 1024
comment = self.get_long_name(length=1024, mode="letters")
self.__create_tb(dbname,ntbname,comment)
ntb_kv_list = tdSql.getResult("show tables")
tdSql.checkEqual(ntb_kv_list[0][8], comment)
tdSql.execute(f'drop database {dbname}')
# error overlength
comment = self.get_long_name(length=1025, mode="letters")
tdSql.execute(f'create database if not exists {dbname}')
tdSql.execute(f'use {dbname}')
tdSql.execute(f"create table stb (ts timestamp,c0 int) tags(t0 int)")
tdSql.error(f'create table stb_1 us stb tags(1) comment "{comment}"')
tdSql.execute(f'drop database {dbname}')
def run(self):
self.check_comment()
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())

85
tests/system-test/2-query/json_tag.py
View File

@ -61,6 +61,8 @@ class TDTestCase:
# test invalidate json # test invalidate json
tdSql.error("CREATE TABLE if not exists jsons1_14 using jsons1 tags('\"efwewf\"')") tdSql.error("CREATE TABLE if not exists jsons1_14 using jsons1 tags('\"efwewf\"')")
tdSql.error("CREATE TABLE if not exists jsons1_14 using jsons1 tags('3333')") tdSql.error("CREATE TABLE if not exists jsons1_14 using jsons1 tags('3333')")
tdSql.error("CREATE TABLE if not exists jsons1_14 using jsons1 tags(76)")
tdSql.error("CREATE TABLE if not exists jsons1_14 using jsons1 tags(hell)")
tdSql.error("CREATE TABLE if not exists jsons1_14 using jsons1 tags('33.33')") tdSql.error("CREATE TABLE if not exists jsons1_14 using jsons1 tags('33.33')")
tdSql.error("CREATE TABLE if not exists jsons1_14 using jsons1 tags('false')") tdSql.error("CREATE TABLE if not exists jsons1_14 using jsons1 tags('false')")
tdSql.error("CREATE TABLE if not exists jsons1_14 using jsons1 tags('[1,true]')") tdSql.error("CREATE TABLE if not exists jsons1_14 using jsons1 tags('[1,true]')")
@ -128,7 +130,7 @@ class TDTestCase:
tdSql.error("select ceil(jtag) from jsons1") tdSql.error("select ceil(jtag) from jsons1")
# #
# # test select normal column # # test select normal column
tdSql.query("select dataint from jsons1") tdSql.query("select dataint from jsons1 order by dataint")
tdSql.checkRows(9) tdSql.checkRows(9)
tdSql.checkData(1, 0, 1) tdSql.checkData(1, 0, 1)
@ -137,9 +139,9 @@ class TDTestCase:
tdSql.checkRows(9) tdSql.checkRows(9)
tdSql.query("select jtag from jsons1") tdSql.query("select jtag from jsons1")
tdSql.checkRows(13) tdSql.checkRows(13)
tdSql.query("select jtag from jsons1 where jtag is null") # tdSql.query("select jtag from jsons1 where jtag is null")
# tdSql.checkRows(5) # tdSql.checkRows(5)
tdSql.query("select jtag from jsons1 where jtag is not null") # tdSql.query("select jtag from jsons1 where jtag is not null")
# tdSql.checkRows(8) # tdSql.checkRows(8)
# test jtag is NULL # test jtag is NULL
@ -259,12 +261,6 @@ class TDTestCase:
# tdSql.query("select * from jsons1 where jtag->'tag1'=null") # only json suport =null. This synatx will change later. # tdSql.query("select * from jsons1 where jtag->'tag1'=null") # only json suport =null. This synatx will change later.
# tdSql.checkRows(1) # tdSql.checkRows(1)
# #
# # where json is null
tdSql.query("select * from jsons1 where jtag is null")
# tdSql.checkRows(1)
tdSql.query("select * from jsons1 where jtag is not null")
# tdSql.checkRows(8)
#
# # where json key is null # # where json key is null
# tdSql.query("select * from jsons1 where jtag->'tag_no_exist'=3") # tdSql.query("select * from jsons1 where jtag->'tag_no_exist'=3")
# tdSql.checkRows(0) # tdSql.checkRows(0)
@ -358,25 +354,28 @@ class TDTestCase:
# tdSql.checkRows(0) # tdSql.checkRows(0)
# #
# # test join # # test join
# tdSql.execute("create table if not exists jsons2(ts timestamp, dataInt int, dataBool bool, dataStr nchar(50), dataStrBin binary(150)) tags(jtag json)") tdSql.execute("create table if not exists jsons2(ts timestamp, dataInt int, dataBool bool, dataStr nchar(50), dataStrBin binary(150)) tags(jtag json)")
# tdSql.execute("insert into jsons2_1 using jsons2 tags('{\"tag1\":\"fff\",\"tag2\":5, \"tag3\":true}') values(1591060618000, 2, false, 'json2', '你是2')") tdSql.execute("insert into jsons2_1 using jsons2 tags('{\"tag1\":\"fff\",\"tag2\":5, \"tag3\":true}') values(1591060618000, 2, false, 'json2', '你是2')")
# tdSql.execute("insert into jsons2_2 using jsons2 tags('{\"tag1\":5,\"tag2\":null}') values (1591060628000, 2, true, 'json2', 'sss')") tdSql.execute("insert into jsons2_2 using jsons2 tags('{\"tag1\":5,\"tag2\":null}') values (1591060628000, 2, true, 'json2', 'sss')")
#
# tdSql.execute("create table if not exists jsons3(ts timestamp, dataInt int, dataBool bool, dataStr nchar(50), dataStrBin binary(150)) tags(jtag json)") tdSql.execute("create table if not exists jsons3(ts timestamp, dataInt int, dataBool bool, dataStr nchar(50), dataStrBin binary(150)) tags(jtag json)")
# tdSql.execute("insert into jsons3_1 using jsons3 tags('{\"tag1\":\"fff\",\"tag2\":5, \"tag3\":true}') values(1591060618000, 3, false, 'json3', '你是3')") tdSql.execute("insert into jsons3_1 using jsons3 tags('{\"tag1\":\"fff\",\"tag2\":5, \"tag3\":true}') values(1591060618000, 3, false, 'json3', '你是3')")
# tdSql.execute("insert into jsons3_2 using jsons3 tags('{\"tag1\":5,\"tag2\":\"beijing\"}') values (1591060638000, 2, true, 'json3', 'sss')") tdSql.execute("insert into jsons3_2 using jsons3 tags('{\"tag1\":5,\"tag2\":\"beijing\"}') values (1591060638000, 2, true, 'json3', 'sss')")
# tdSql.query("select 'sss',33,a.jtag->'tag3' from jsons2 a,jsons3 b where a.ts=b.ts and a.jtag->'tag1'=b.jtag->'tag1'") tdSql.query("select 'sss',33,a.jtag->'tag3' from jsons2 a,jsons3 b where a.ts=b.ts and a.jtag->'tag1'=b.jtag->'tag1'")
# tdSql.checkData(0, 0, "sss") tdSql.checkData(0, 0, "sss")
# tdSql.checkData(0, 2, "true") tdSql.checkData(0, 2, "true")
#
# res = tdSql.getColNameList("select 'sss',33,a.jtag->'tag3' from jsons2 a,jsons3 b where a.ts=b.ts and a.jtag->'tag1'=b.jtag->'tag1'") res = tdSql.getColNameList("select 'sss',33,a.jtag->'tag3' from jsons2 a,jsons3 b where a.ts=b.ts and a.jtag->'tag1'=b.jtag->'tag1'")
# cname_list = [] cname_list = []
# cname_list.append("sss") cname_list.append("'sss'")
# cname_list.append("33") cname_list.append("33")
# cname_list.append("a.jtag->'tag3'") cname_list.append("a.jtag->'tag3'")
# tdSql.checkColNameList(res, cname_list) tdSql.checkColNameList(res, cname_list)
# #
# test group by & order by json tag # test group by & order by json tag
# tdSql.error("select count(*) from jsons1 group by jtag")
# tdSql.error("select count(*) from jsons1 partition by jtag")
# tdSql.error("select count(*) from jsons1 group by jtag order by jtag")
tdSql.error("select count(*) from jsons1 group by jtag->'tag1' order by jtag->'tag2'") tdSql.error("select count(*) from jsons1 group by jtag->'tag1' order by jtag->'tag2'")
tdSql.error("select count(*) from jsons1 group by jtag->'tag1' order by jtag") tdSql.error("select count(*) from jsons1 group by jtag->'tag1' order by jtag")
tdSql.query("select count(*),jtag->'tag1' from jsons1 group by jtag->'tag1' order by jtag->'tag1' desc") tdSql.query("select count(*),jtag->'tag1' from jsons1 group by jtag->'tag1' order by jtag->'tag1' desc")
@ -432,14 +431,14 @@ class TDTestCase:
tdSql.checkData(10, 1, '"femail"') tdSql.checkData(10, 1, '"femail"')
# test having # test having
# tdSql.query("select stddev(dataint) from jsons1 group by jtag->'tag1' having stddev(dataint) > 0") tdSql.query("select count(*),jtag->'tag1' from jsons1 group by jtag->'tag1' having count(*) > 1")
# tdSql.checkRows(2) tdSql.checkRows(3)
# subquery with json tag # subquery with json tag
tdSql.query("select * from (select jtag, dataint from jsons1) order by dataint") tdSql.query("select * from (select jtag, dataint from jsons1) order by dataint")
tdSql.checkRows(11) tdSql.checkRows(11)
tdSql.checkData(1, 1, 1) tdSql.checkData(1, 1, 1)
tdSql.checkData(2, 0, '{"tag1":5,"tag2":"beijing"}') tdSql.checkData(5, 0, '{"tag1":false,"tag2":"beijing"}')
# tdSql.query("select jtag->'tag1' from (select jtag->'tag1', dataint from jsons1)") # tdSql.query("select jtag->'tag1' from (select jtag->'tag1', dataint from jsons1)")
# tdSql.checkRows(11) # tdSql.checkRows(11)
@ -457,16 +456,18 @@ class TDTestCase:
# tdSql.checkData(1, 2, '"femail"') # tdSql.checkData(1, 2, '"femail"')
# #
# # union all # # union all
# tdSql.error("select jtag->'tag1' from jsons1 union all select jtag->'tag2' from jsons2") tdSql.query("select jtag->'tag1' from jsons1 union all select jtag->'tag2' from jsons2")
# tdSql.error("select jtag->'tag1' from jsons1_1 union all select jtag->'tag2' from jsons2_1") tdSql.checkRows(17)
# tdSql.query("select jtag->'tag1' from jsons1_1 union all select jtag->'tag2' from jsons2_1")
# tdSql.query("select jtag->'tag1' from jsons1_1 union all select jtag->'tag1' from jsons2_1") tdSql.checkRows(2)
# tdSql.checkRows(2)
# tdSql.query("select dataint,jtag->'tag1',tbname from jsons1 union all select dataint,jtag->'tag1',tbname from jsons2") tdSql.query("select jtag->'tag1' from jsons1_1 union all select jtag->'tag1' from jsons2_1")
# tdSql.checkRows(13) tdSql.checkRows(2)
# tdSql.query("select dataint,jtag,tbname from jsons1 union all select dataint,jtag,tbname from jsons2") tdSql.query("select dataint,jtag->'tag1',tbname from jsons1 union all select dataint,jtag->'tag1',tbname from jsons2")
# tdSql.checkRows(13) tdSql.checkRows(13)
# tdSql.query("select dataint,jtag,tbname from jsons1 union all select dataint,jtag,tbname from jsons2")
tdSql.checkRows(13)
# #show create table # #show create table
# tdSql.query("show create table jsons1") # tdSql.query("show create table jsons1")
# tdSql.checkData(0, 1, 'CREATE TABLE `jsons1` (`ts` TIMESTAMP,`dataint` INT,`databool` BOOL,`datastr` NCHAR(50),`datastrbin` BINARY(150)) TAGS (`jtag` JSON)') # tdSql.checkData(0, 1, 'CREATE TABLE `jsons1` (`ts` TIMESTAMP,`dataint` INT,`databool` BOOL,`datastr` NCHAR(50),`datastrbin` BINARY(150)) TAGS (`jtag` JSON)')
@ -528,9 +529,9 @@ class TDTestCase:
tdSql.execute("CREATE TABLE if not exists jsons1_20 using jsons1 tags(NULL)") tdSql.execute("CREATE TABLE if not exists jsons1_20 using jsons1 tags(NULL)")
tdSql.query("select jtag from jsons1_20") tdSql.query("select jtag from jsons1_20")
tdSql.checkData(0, 0, None) tdSql.checkData(0, 0, None)
# tdSql.execute("insert into jsons1_21 using jsons1 tags(NULL) values(1591061628000, 11, false, '你就会','')") tdSql.execute("insert into jsons1_21 using jsons1 tags(NULL) values(1591061628000, 11, false, '你就会','')")
# tdSql.query("select jtag from jsons1_21") tdSql.query("select jtag from jsons1_21")
# tdSql.checkData(0, 0, None) tdSql.checkData(0, 0, None)
# #
# #test TD-12389 # #test TD-12389
tdSql.query("describe jsons1") tdSql.query("describe jsons1")

16
tests/system-test/7-tmq/schema.py
View File

@ -282,7 +282,7 @@ class TDTestCase:
tdSql.execute("create topic %s as select ts, c1, c2, t1, t2 from %s.%s" %(columnTopicFromStb, parameterDict['dbName'], parameterDict['stbName'])) tdSql.execute("create topic %s as select ts, c1, c2, t1, t2 from %s.%s" %(columnTopicFromStb, parameterDict['dbName'], parameterDict['stbName']))
tdSql.execute("create topic %s as select ts, c1, c2 from %s.%s" %(columnTopicFromNtb, parameterDict['dbName'], ntbName)) tdSql.execute("create topic %s as select ts, c1, c2 from %s.%s" %(columnTopicFromNtb, parameterDict['dbName'], ntbName))
tsLog.info("======== super table test:") tdLog.info("======== super table test:")
# alter actions prohibited: drop column/tag, modify column/tag type, rename column/tag included in topic # alter actions prohibited: drop column/tag, modify column/tag type, rename column/tag included in topic
tdSql.error("alter table %s.%s drop column c1"%(parameterDict['dbName'], parameterDict['stbName'])) tdSql.error("alter table %s.%s drop column c1"%(parameterDict['dbName'], parameterDict['stbName']))
tdSql.error("alter table %s.%s drop column c2"%(parameterDict['dbName'], parameterDict['stbName'])) tdSql.error("alter table %s.%s drop column c2"%(parameterDict['dbName'], parameterDict['stbName']))
@ -316,7 +316,7 @@ class TDTestCase:
tdSql.query("alter table %s.%s add tag t3 int"%(parameterDict['dbName'], parameterDict['stbName'])) tdSql.query("alter table %s.%s add tag t3 int"%(parameterDict['dbName'], parameterDict['stbName']))
tdSql.query("alter table %s.%s add tag t4 float"%(parameterDict['dbName'], parameterDict['stbName'])) tdSql.query("alter table %s.%s add tag t4 float"%(parameterDict['dbName'], parameterDict['stbName']))
tsLog.info("======== normal table test:") tdLog.info("======== normal table test:")
# alter actions prohibited: drop column/tag, modify column/tag type, rename column/tag included in topic # alter actions prohibited: drop column/tag, modify column/tag type, rename column/tag included in topic
tdSql.error("alter table %s.%s drop column c1"%(parameterDict['dbName'], ntbName)) tdSql.error("alter table %s.%s drop column c1"%(parameterDict['dbName'], ntbName))
tdSql.error("alter table %s.%s drop column c2"%(parameterDict['dbName'], ntbName)) tdSql.error("alter table %s.%s drop column c2"%(parameterDict['dbName'], ntbName))
@ -419,7 +419,7 @@ class TDTestCase:
tdSql.execute("create topic %s as select ts, c1, c2, t1, t2 from %s.%s where c3 > 3 and c4 like 'abc' and t3 = 5 and t4 = 'beijing'" %(columnTopicFromStb, parameterDict['dbName'], parameterDict['stbName'])) tdSql.execute("create topic %s as select ts, c1, c2, t1, t2 from %s.%s where c3 > 3 and c4 like 'abc' and t3 = 5 and t4 = 'beijing'" %(columnTopicFromStb, parameterDict['dbName'], parameterDict['stbName']))
tdSql.execute("create topic %s as select ts, c1, c2 from %s.%s where c3 > 3 and c4 like 'abc'" %(columnTopicFromNtb, parameterDict['dbName'], ntbName)) tdSql.execute("create topic %s as select ts, c1, c2 from %s.%s where c3 > 3 and c4 like 'abc'" %(columnTopicFromNtb, parameterDict['dbName'], ntbName))
tsLog.info("======== super table test:") tdLog.info("======== super table test:")
# alter actions prohibited: drop column/tag, modify column/tag type, rename column/tag included in topic # alter actions prohibited: drop column/tag, modify column/tag type, rename column/tag included in topic
tdSql.error("alter table %s.%s drop column c1"%(parameterDict['dbName'], parameterDict['stbName'])) tdSql.error("alter table %s.%s drop column c1"%(parameterDict['dbName'], parameterDict['stbName']))
tdSql.error("alter table %s.%s drop column c2"%(parameterDict['dbName'], parameterDict['stbName'])) tdSql.error("alter table %s.%s drop column c2"%(parameterDict['dbName'], parameterDict['stbName']))
@ -457,7 +457,7 @@ class TDTestCase:
tdSql.query("alter table %s.%s add column c5 int"%(parameterDict['dbName'], parameterDict['stbName'])) tdSql.query("alter table %s.%s add column c5 int"%(parameterDict['dbName'], parameterDict['stbName']))
tdSql.query("alter table %s.%s add tag t5 float"%(parameterDict['dbName'], parameterDict['stbName'])) tdSql.query("alter table %s.%s add tag t5 float"%(parameterDict['dbName'], parameterDict['stbName']))
tsLog.info("======== normal table test:") tdLog.info("======== normal table test:")
# alter actions prohibited: drop column/tag, modify column/tag type, rename column/tag included in topic # alter actions prohibited: drop column/tag, modify column/tag type, rename column/tag included in topic
tdSql.error("alter table %s.%s drop column c1"%(parameterDict['dbName'], ntbName)) tdSql.error("alter table %s.%s drop column c1"%(parameterDict['dbName'], ntbName))
tdSql.error("alter table %s.%s drop column c2"%(parameterDict['dbName'], ntbName)) tdSql.error("alter table %s.%s drop column c2"%(parameterDict['dbName'], ntbName))
@ -566,7 +566,7 @@ class TDTestCase:
tdSql.execute("create topic %s as select * from %s.%s" %(columnTopicFromStb, parameterDict['dbName'], parameterDict['stbName'])) tdSql.execute("create topic %s as select * from %s.%s" %(columnTopicFromStb, parameterDict['dbName'], parameterDict['stbName']))
tdSql.execute("create topic %s as select * from %s.%s " %(columnTopicFromNtb, parameterDict['dbName'], ntbName)) tdSql.execute("create topic %s as select * from %s.%s " %(columnTopicFromNtb, parameterDict['dbName'], ntbName))
tsLog.info("======== super table test:") tdLog.info("======== super table test:")
# alter actions prohibited: drop column/tag, modify column/tag type, rename column/tag included in topic # alter actions prohibited: drop column/tag, modify column/tag type, rename column/tag included in topic
tdSql.error("alter table %s.%s drop column c1"%(parameterDict['dbName'], parameterDict['stbName'])) tdSql.error("alter table %s.%s drop column c1"%(parameterDict['dbName'], parameterDict['stbName']))
tdSql.error("alter table %s.%s drop column c2"%(parameterDict['dbName'], parameterDict['stbName'])) tdSql.error("alter table %s.%s drop column c2"%(parameterDict['dbName'], parameterDict['stbName']))
@ -601,7 +601,7 @@ class TDTestCase:
tdSql.query("alter table %s.%s add column c6 int"%(parameterDict['dbName'], parameterDict['stbName'])) tdSql.query("alter table %s.%s add column c6 int"%(parameterDict['dbName'], parameterDict['stbName']))
tdSql.query("alter table %s.%s add tag t6 float"%(parameterDict['dbName'], parameterDict['stbName'])) tdSql.query("alter table %s.%s add tag t6 float"%(parameterDict['dbName'], parameterDict['stbName']))
tsLog.info("======== normal table test:") tdLog.info("======== normal table test:")
# alter actions prohibited: drop column/tag, modify column/tag type, rename column/tag included in topic # alter actions prohibited: drop column/tag, modify column/tag type, rename column/tag included in topic
tdSql.error("alter table %s.%s drop column c1"%(parameterDict['dbName'], ntbName)) tdSql.error("alter table %s.%s drop column c1"%(parameterDict['dbName'], ntbName))
tdSql.error("alter table %s.%s drop column c2"%(parameterDict['dbName'], ntbName)) tdSql.error("alter table %s.%s drop column c2"%(parameterDict['dbName'], ntbName))
@ -687,8 +687,8 @@ class TDTestCase:
tdLog.info("cfgPath: %s" % cfgPath) tdLog.info("cfgPath: %s" % cfgPath)
self.tmqCase1(cfgPath, buildPath) self.tmqCase1(cfgPath, buildPath)
self.tmqCase2(cfgPath, buildPath) # self.tmqCase2(cfgPath, buildPath)
self.tmqCase3(cfgPath, buildPath) # self.tmqCase3(cfgPath, buildPath)
def stop(self): def stop(self):
tdSql.close() tdSql.close()

2
tests/system-test/fulltest.sh
View File

@ -21,6 +21,7 @@ python3 ./test.py -f 1-insert/opentsdb_json_taosc_insert.py
#python3 ./test.py -f 1-insert/test_stmt_muti_insert_query.py #python3 ./test.py -f 1-insert/test_stmt_muti_insert_query.py
python3 ./test.py -f 1-insert/alter_stable.py python3 ./test.py -f 1-insert/alter_stable.py
python3 ./test.py -f 1-insert/alter_table.py python3 ./test.py -f 1-insert/alter_table.py
# python3 ./test.py -f 1-inerst/create_table_comment.py
python3 ./test.py -f 2-query/between.py python3 ./test.py -f 2-query/between.py
python3 ./test.py -f 2-query/distinct.py python3 ./test.py -f 2-query/distinct.py
python3 ./test.py -f 2-query/varchar.py python3 ./test.py -f 2-query/varchar.py
@ -57,6 +58,7 @@ python3 ./test.py -f 2-query/To_unixtimestamp.py
python3 ./test.py -f 2-query/timetruncate.py python3 ./test.py -f 2-query/timetruncate.py
python3 ./test.py -f 2-query/diff.py python3 ./test.py -f 2-query/diff.py
python3 ./test.py -f 2-query/Timediff.py python3 ./test.py -f 2-query/Timediff.py
python3 ./test.py -f 2-query/json_tag.py
python3 ./test.py -f 2-query/top.py python3 ./test.py -f 2-query/top.py
python3 ./test.py -f 2-query/bottom.py python3 ./test.py -f 2-query/bottom.py