diff --git a/cmake/cmake.options b/cmake/cmake.options index ab3c5ac1ad..62816a80d5 100644 --- a/cmake/cmake.options +++ b/cmake/cmake.options @@ -146,6 +146,6 @@ option( option( BUILD_WITH_INVERTEDINDEX "If use invertedIndex" - OFF + ON ) diff --git a/include/client/taos.h b/include/client/taos.h index c311e017df..7f317b3a9d 100644 --- a/include/client/taos.h +++ b/include/client/taos.h @@ -103,10 +103,10 @@ typedef void (*__taos_async_fn_t)(void *param, TAOS_RES *, int code); typedef struct TAOS_MULTI_BIND { int buffer_type; - void *buffer; + void * buffer; uintptr_t buffer_length; - int32_t *length; - char *is_null; + int32_t * length; + char * is_null; int num; } TAOS_MULTI_BIND; @@ -130,7 +130,7 @@ DLL_EXPORT void taos_cleanup(void); DLL_EXPORT int taos_options(TSDB_OPTION option, const void *arg, ...); DLL_EXPORT setConfRet taos_set_config(const char *config); DLL_EXPORT int taos_init(void); -DLL_EXPORT TAOS *taos_connect(const char *ip, const char *user, const char *pass, const char *db, uint16_t port); +DLL_EXPORT TAOS *taos_connect(const char *ip, const char *user, const char *pass, const char *db, uint16_t port); DLL_EXPORT TAOS *taos_connect_l(const char *ip, int ipLen, const char *user, int userLen, const char *pass, int passLen, const char *db, int dbLen, uint16_t port); DLL_EXPORT TAOS *taos_connect_auth(const char *ip, const char *user, const char *auth, const char *db, uint16_t port); @@ -147,17 +147,17 @@ DLL_EXPORT int taos_stmt_set_sub_tbname(TAOS_STMT *stmt, const char *name DLL_EXPORT int taos_stmt_get_tag_fields(TAOS_STMT *stmt, int *fieldNum, TAOS_FIELD_E **fields); DLL_EXPORT int taos_stmt_get_col_fields(TAOS_STMT *stmt, int *fieldNum, TAOS_FIELD_E **fields); -DLL_EXPORT int taos_stmt_is_insert(TAOS_STMT *stmt, int *insert); -DLL_EXPORT int taos_stmt_num_params(TAOS_STMT *stmt, int *nums); -DLL_EXPORT int taos_stmt_get_param(TAOS_STMT *stmt, int idx, int *type, int *bytes); -DLL_EXPORT int taos_stmt_bind_param(TAOS_STMT *stmt, TAOS_MULTI_BIND *bind); -DLL_EXPORT int taos_stmt_bind_param_batch(TAOS_STMT *stmt, TAOS_MULTI_BIND *bind); -DLL_EXPORT int taos_stmt_bind_single_param_batch(TAOS_STMT *stmt, TAOS_MULTI_BIND *bind, int colIdx); -DLL_EXPORT int taos_stmt_add_batch(TAOS_STMT *stmt); -DLL_EXPORT int taos_stmt_execute(TAOS_STMT *stmt); +DLL_EXPORT int taos_stmt_is_insert(TAOS_STMT *stmt, int *insert); +DLL_EXPORT int taos_stmt_num_params(TAOS_STMT *stmt, int *nums); +DLL_EXPORT int taos_stmt_get_param(TAOS_STMT *stmt, int idx, int *type, int *bytes); +DLL_EXPORT int taos_stmt_bind_param(TAOS_STMT *stmt, TAOS_MULTI_BIND *bind); +DLL_EXPORT int taos_stmt_bind_param_batch(TAOS_STMT *stmt, TAOS_MULTI_BIND *bind); +DLL_EXPORT int taos_stmt_bind_single_param_batch(TAOS_STMT *stmt, TAOS_MULTI_BIND *bind, int colIdx); +DLL_EXPORT int taos_stmt_add_batch(TAOS_STMT *stmt); +DLL_EXPORT int taos_stmt_execute(TAOS_STMT *stmt); DLL_EXPORT TAOS_RES *taos_stmt_use_result(TAOS_STMT *stmt); DLL_EXPORT int taos_stmt_close(TAOS_STMT *stmt); -DLL_EXPORT char *taos_stmt_errstr(TAOS_STMT *stmt); +DLL_EXPORT char * taos_stmt_errstr(TAOS_STMT *stmt); DLL_EXPORT int taos_stmt_affected_rows(TAOS_STMT *stmt); DLL_EXPORT int taos_stmt_affected_rows_once(TAOS_STMT *stmt); @@ -179,11 +179,11 @@ DLL_EXPORT bool taos_is_update_query(TAOS_RES *res); DLL_EXPORT int taos_fetch_block(TAOS_RES *res, TAOS_ROW *rows); DLL_EXPORT int taos_fetch_block_s(TAOS_RES *res, int *numOfRows, TAOS_ROW *rows); DLL_EXPORT int taos_fetch_raw_block(TAOS_RES *res, int *numOfRows, void **pData); -DLL_EXPORT int *taos_get_column_data_offset(TAOS_RES *res, int columnIndex); +DLL_EXPORT int * taos_get_column_data_offset(TAOS_RES *res, int columnIndex); DLL_EXPORT int taos_validate_sql(TAOS *taos, const char *sql); DLL_EXPORT void taos_reset_current_db(TAOS *taos); -DLL_EXPORT int *taos_fetch_lengths(TAOS_RES *res); +DLL_EXPORT int *taos_fetch_lengths(TAOS_RES *res); DLL_EXPORT TAOS_ROW *taos_result_block(TAOS_RES *res); DLL_EXPORT const char *taos_get_server_info(TAOS *taos); @@ -204,7 +204,7 @@ DLL_EXPORT TAOS_RES *taos_consume(TAOS_SUB *tsub); DLL_EXPORT void taos_unsubscribe(TAOS_SUB *tsub, int keepProgress); #endif -DLL_EXPORT int taos_load_table_info(TAOS *taos, const char *tableNameList); +DLL_EXPORT int taos_load_table_info(TAOS *taos, const char *tableNameList); DLL_EXPORT TAOS_RES *taos_schemaless_insert(TAOS *taos, char *lines[], int numLines, int protocol, int precision); /* --------------------------TMQ INTERFACE------------------------------- */ @@ -229,7 +229,7 @@ DLL_EXPORT tmq_list_t *tmq_list_new(); DLL_EXPORT int32_t tmq_list_append(tmq_list_t *, const char *); DLL_EXPORT void tmq_list_destroy(tmq_list_t *); DLL_EXPORT int32_t tmq_list_get_size(const tmq_list_t *); -DLL_EXPORT char **tmq_list_to_c_array(const tmq_list_t *); +DLL_EXPORT char ** tmq_list_to_c_array(const tmq_list_t *); DLL_EXPORT tmq_t *tmq_consumer_new(tmq_conf_t *conf, char *errstr, int32_t errstrLen); @@ -240,7 +240,7 @@ DLL_EXPORT const char *tmq_err2str(tmq_resp_err_t); DLL_EXPORT tmq_resp_err_t tmq_subscribe(tmq_t *tmq, const tmq_list_t *topic_list); DLL_EXPORT tmq_resp_err_t tmq_unsubscribe(tmq_t *tmq); DLL_EXPORT tmq_resp_err_t tmq_subscription(tmq_t *tmq, tmq_list_t **topics); -DLL_EXPORT TAOS_RES *tmq_consumer_poll(tmq_t *tmq, int64_t timeout); +DLL_EXPORT TAOS_RES * tmq_consumer_poll(tmq_t *tmq, int64_t timeout); DLL_EXPORT tmq_resp_err_t tmq_consumer_close(tmq_t *tmq); DLL_EXPORT tmq_resp_err_t tmq_commit_sync(tmq_t *tmq, const tmq_topic_vgroup_list_t *offsets); DLL_EXPORT void tmq_commit_async(tmq_t *tmq, const tmq_topic_vgroup_list_t *offsets, tmq_commit_cb *cb, void *param); @@ -260,7 +260,7 @@ enum tmq_conf_res_t { typedef enum tmq_conf_res_t tmq_conf_res_t; -DLL_EXPORT tmq_conf_t *tmq_conf_new(); +DLL_EXPORT tmq_conf_t * tmq_conf_new(); DLL_EXPORT tmq_conf_res_t tmq_conf_set(tmq_conf_t *conf, const char *key, const char *value); DLL_EXPORT void tmq_conf_destroy(tmq_conf_t *conf); DLL_EXPORT void tmq_conf_set_auto_commit_cb(tmq_conf_t *conf, tmq_commit_cb *cb, void *param); diff --git a/include/common/tmsg.h b/include/common/tmsg.h index 202aeaf69e..39dc5361e6 100644 --- a/include/common/tmsg.h +++ b/include/common/tmsg.h @@ -1982,6 +1982,7 @@ typedef struct { typedef struct { SClientHbKey connKey; + int64_t clusterId; SQueryHbReqBasic* query; SHashObj* info; // hash } SClientHbReq; diff --git a/include/libs/catalog/catalog.h b/include/libs/catalog/catalog.h index f0e642bc9a..ee237741c3 100644 --- a/include/libs/catalog/catalog.h +++ b/include/libs/catalog/catalog.h @@ -68,6 +68,7 @@ typedef struct SCatalogReq { SArray *pIndex; // element is index name SArray *pUser; // element is SUserAuthInfo bool qNodeRequired; // valid qnode + bool forceUpdate; } SCatalogReq; 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 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); /** diff --git a/include/libs/index/index.h b/include/libs/index/index.h index 180c7e7216..bd601f1d9f 100644 --- a/include/libs/index/index.h +++ b/include/libs/index/index.h @@ -193,8 +193,9 @@ void indexInit(); /* index filter */ typedef struct SIndexMetaArg { - void* metaHandle; void* metaEx; + void* idx; + void* ivtIdx; uint64_t suid; } SIndexMetaArg; diff --git a/include/libs/qcom/query.h b/include/libs/qcom/query.h index 45a7e9a29f..a17be44846 100644 --- a/include/libs/qcom/query.h +++ b/include/libs/qcom/query.h @@ -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) #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 diff --git a/include/libs/scheduler/scheduler.h b/include/libs/scheduler/scheduler.h index 5cb6dbd3bc..22706f0953 100644 --- a/include/libs/scheduler/scheduler.h +++ b/include/libs/scheduler/scheduler.h @@ -104,6 +104,8 @@ void schedulerAsyncFetchRows(int64_t job, schedulerFetchCallback fp, void* param int32_t schedulerGetTasksStatus(int64_t job, SArray *pSub); +void schedulerStopQueryHb(void *pTrans); + /** * Cancel query job diff --git a/source/client/inc/clientInt.h b/source/client/inc/clientInt.h index 09116a6bd1..eaf18884c6 100644 --- a/source/client/inc/clientInt.h +++ b/source/client/inc/clientInt.h @@ -57,11 +57,6 @@ enum { typedef struct SAppInstInfo SAppInstInfo; -typedef struct { - void* param; - SClientHbReq* req; -} SHbConnInfo; - typedef struct { char* key; // statistics @@ -71,11 +66,8 @@ typedef struct { int64_t startTime; // ctl SRWLatch lock; // lock is used in serialization - // connection SAppInstInfo* pAppInstInfo; - // info SHashObj* activeInfo; // hash - SHashObj* connInfo; // hash } SAppHbMgr; 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); void hbDeregisterConn(SAppHbMgr* pAppHbMgr, SClientHbKey connKey); -int hbAddConnInfo(SAppHbMgr* pAppHbMgr, SClientHbKey connKey, void* key, void* value, int32_t keyLen, int32_t valueLen); - // --- mq void hbMgrInitMqHbRspHandle(); diff --git a/source/client/src/clientEnv.c b/source/client/src/clientEnv.c index 4b39a51584..be0a41ba40 100644 --- a/source/client/src/clientEnv.c +++ b/source/client/src/clientEnv.c @@ -130,8 +130,13 @@ void destroyTscObj(void *pObj) { SClientHbKey connKey = {.tscRid = pTscObj->id, .connType = pTscObj->connType}; 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); + schedulerStopQueryHb(pTscObj->pAppInfo->pTransporter); + if (0 == connNum) { + // TODO + //closeTransporter(pTscObj); + } tscDebug("connObj 0x%" PRIx64 " destroyed, totalConn:%" PRId64, pTscObj->id, pTscObj->pAppInfo->numOfConns); taosThreadMutexDestroy(&pTscObj->mutex); taosMemoryFreeClear(pTscObj); @@ -223,6 +228,10 @@ static void doDestroyRequest(void *p) { taosHashRemove(pRequest->pTscObj->pRequests, &pRequest->self, sizeof(pRequest->self)); + if (pRequest->body.queryJob != 0) { + schedulerFreeJob(pRequest->body.queryJob); + } + taosMemoryFreeClear(pRequest->msgBuf); taosMemoryFreeClear(pRequest->sqlstr); taosMemoryFreeClear(pRequest->pDb); @@ -230,10 +239,6 @@ static void doDestroyRequest(void *p) { doFreeReqResultInfo(&pRequest->body.resInfo); qDestroyQueryPlan(pRequest->body.pDag); - if (pRequest->body.queryJob != 0) { - schedulerFreeJob(pRequest->body.queryJob); - } - taosArrayDestroy(pRequest->tableList); taosArrayDestroy(pRequest->dbList); diff --git a/source/client/src/clientHb.c b/source/client/src/clientHb.c index 09c3d269c7..a4c109ee17 100644 --- a/source/client/src/clientHb.c +++ b/source/client/src/clientHb.c @@ -129,9 +129,9 @@ static int32_t hbProcessStbInfoRsp(void *value, int32_t valueLen, struct SCatalo } static int32_t hbQueryHbRspHandle(SAppHbMgr *pAppHbMgr, SClientHbRsp *pRsp) { - SHbConnInfo *info = taosHashGet(pAppHbMgr->connInfo, &pRsp->connKey, sizeof(SClientHbKey)); - if (NULL == info) { - tscWarn("fail to get connInfo, may be dropped, refId:%" PRIx64 ", type:%d", pRsp->connKey.tscRid, + SClientHbReq *pReq = taosHashGet(pAppHbMgr->activeInfo, &pRsp->connKey, sizeof(SClientHbKey)); + if (NULL == pReq) { + tscWarn("pReq to get activeInfo, may be dropped, refId:%" PRIx64 ", type:%d", pRsp->connKey.tscRid, pRsp->connKey.connType); return TSDB_CODE_SUCCESS; } @@ -181,12 +181,11 @@ static int32_t hbQueryHbRspHandle(SAppHbMgr *pAppHbMgr, SClientHbRsp *pRsp) { break; } - int64_t *clusterId = (int64_t *)info->param; struct SCatalog *pCatalog = NULL; - int32_t code = catalogGetHandle(*clusterId, &pCatalog); + int32_t code = catalogGetHandle(pReq->clusterId, &pCatalog); 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; } @@ -199,12 +198,11 @@ static int32_t hbQueryHbRspHandle(SAppHbMgr *pAppHbMgr, SClientHbRsp *pRsp) { break; } - int64_t *clusterId = (int64_t *)info->param; struct SCatalog *pCatalog = NULL; - int32_t code = catalogGetHandle(*clusterId, &pCatalog); + int32_t code = catalogGetHandle(pReq->clusterId, &pCatalog); 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; } @@ -217,12 +215,11 @@ static int32_t hbQueryHbRspHandle(SAppHbMgr *pAppHbMgr, SClientHbRsp *pRsp) { break; } - int64_t *clusterId = (int64_t *)info->param; struct SCatalog *pCatalog = NULL; - int32_t code = catalogGetHandle(*clusterId, &pCatalog); + int32_t code = catalogGetHandle(pReq->clusterId, &pCatalog); 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; } @@ -547,13 +544,10 @@ SClientHbBatchReq *hbGatherAllInfo(SAppHbMgr *pAppHbMgr) { pOneReq = taosArrayPush(pBatchReq->reqs, pOneReq); - SHbConnInfo *info = taosHashGet(pAppHbMgr->connInfo, &pOneReq->connKey, sizeof(SClientHbKey)); - if (info) { - code = (*clientHbMgr.reqHandle[pOneReq->connKey.connType])(&pOneReq->connKey, info->param, pOneReq); - if (code) { - pIter = taosHashIterate(pAppHbMgr->activeInfo, pIter); - continue; - } + code = (*clientHbMgr.reqHandle[pOneReq->connKey.connType])(&pOneReq->connKey, &pOneReq->clusterId, pOneReq); + if (code) { + pIter = taosHashIterate(pAppHbMgr->activeInfo, pIter); + continue; } //hbClearClientHbReq(pOneReq); @@ -569,23 +563,6 @@ SClientHbBatchReq *hbGatherAllInfo(SAppHbMgr *pAppHbMgr) { 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) { atomic_store_8(&clientHbMgr.threadStop, 2); } @@ -715,14 +692,6 @@ SAppHbMgr *appHbMgrInit(SAppInstInfo *pAppInstInfo, char *key) { } 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); taosArrayPush(clientHbMgr.appHbMgrs, &pAppHbMgr); @@ -745,15 +714,6 @@ void appHbMgrCleanup(void) { taosHashCleanup(pTarget->activeInfo); 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); } @@ -791,7 +751,7 @@ void hbMgrCleanUp() { clientHbMgr.appHbMgrs = NULL; } -int hbRegisterConnImpl(SAppHbMgr *pAppHbMgr, SClientHbKey connKey, SHbConnInfo *info) { +int hbRegisterConnImpl(SAppHbMgr *pAppHbMgr, SClientHbKey connKey, int64_t clusterId) { // init hash in activeinfo void *data = taosHashGet(pAppHbMgr->activeInfo, &connKey, sizeof(SClientHbKey)); if (data != NULL) { @@ -799,17 +759,11 @@ int hbRegisterConnImpl(SAppHbMgr *pAppHbMgr, SClientHbKey connKey, SHbConnInfo * } SClientHbReq hbReq = {0}; hbReq.connKey = connKey; + hbReq.clusterId = clusterId; //hbReq.info = taosHashInit(64, hbKeyHashFunc, 1, HASH_ENTRY_LOCK); 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); return 0; } @@ -819,15 +773,10 @@ int hbRegisterConn(SAppHbMgr *pAppHbMgr, int64_t tscRefId, int64_t clusterId, in .tscRid = tscRefId, .connType = connType, }; - SHbConnInfo info = {0}; switch (connType) { case CONN_TYPE__QUERY: { - int64_t *pClusterId = taosMemoryMalloc(sizeof(int64_t)); - *pClusterId = clusterId; - - info.param = pClusterId; - return hbRegisterConnImpl(pAppHbMgr, connKey, &info); + return hbRegisterConnImpl(pAppHbMgr, connKey, clusterId); } case CONN_TYPE__TMQ: { return 0; @@ -844,26 +793,10 @@ void hbDeregisterConn(SAppHbMgr *pAppHbMgr, SClientHbKey connKey) { taosHashRemove(pAppHbMgr->activeInfo, &connKey, sizeof(SClientHbKey)); } - SHbConnInfo *info = taosHashGet(pAppHbMgr->connInfo, &connKey, sizeof(SClientHbKey)); - if (info) { - taosMemoryFree(info->param); - taosHashRemove(pAppHbMgr->connInfo, &connKey, sizeof(SClientHbKey)); - } - - if (NULL == pReq || NULL == info) { + if (NULL == pReq) { return; } 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; -} diff --git a/source/client/src/clientMsgHandler.c b/source/client/src/clientMsgHandler.c index 1039d36362..2d5199f181 100644 --- a/source/client/src/clientMsgHandler.c +++ b/source/client/src/clientMsgHandler.c @@ -180,7 +180,7 @@ int32_t processUseDbRsp(void* param, const SDataBuf* pMsg, int32_t code) { taosMemoryFreeClear(output.dbVgroup); 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; int32_t code1 = catalogGetHandle(pRequest->pTscObj->pAppInfo->clusterId, &pCatalog); diff --git a/source/client/src/clientSml.c b/source/client/src/clientSml.c index c559ac58f8..5643af746d 100644 --- a/source/client/src/clientSml.c +++ b/source/client/src/clientSml.c @@ -2332,6 +2332,8 @@ static int32_t isSchemalessDb(SSmlHandle* info){ smlBuildInvalidDataMsg(&info->msgBuf, "catalogGetDBCfg error, code:", tstrerror(code)); return code; } + taosArrayDestroy(pInfo.pRetensions); + if (!pInfo.schemaless){ info->pRequest->code = TSDB_CODE_SML_INVALID_DB_CONF; smlBuildInvalidDataMsg(&info->msgBuf, "can not insert into schemaless db:", dbFname); diff --git a/source/dnode/vnode/inc/vnode.h b/source/dnode/vnode/inc/vnode.h index 7bd2eadd4b..8c7a78a5af 100644 --- a/source/dnode/vnode/inc/vnode.h +++ b/source/dnode/vnode/inc/vnode.h @@ -56,7 +56,7 @@ int32_t vnodePreprocessReq(SVnode *pVnode, SRpcMsg *pMsg); int32_t vnodeProcessWriteReq(SVnode *pVnode, SRpcMsg *pMsg, int64_t version, SRpcMsg *pRsp); int32_t vnodeProcessCMsg(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp); int32_t vnodeProcessSyncReq(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp); -int32_t vnodePreprocessQueryMsg(SVnode * pVnode, SRpcMsg * pMsg); +int32_t vnodePreprocessQueryMsg(SVnode *pVnode, SRpcMsg *pMsg); int32_t vnodeProcessQueryMsg(SVnode *pVnode, SRpcMsg *pMsg); int32_t vnodeProcessFetchMsg(SVnode *pVnode, SRpcMsg *pMsg, SQueueInfo *pInfo); int32_t vnodeGetLoad(SVnode *pVnode, SVnodeLoad *pLoad); @@ -86,7 +86,7 @@ typedef struct SMetaFltParam { tb_uid_t suid; int16_t cid; int16_t type; - char *val; + char * val; bool reverse; int (*filterFunc)(void *a, void *b, int16_t type); @@ -121,7 +121,8 @@ int32_t tsdbGetFileBlocksDistInfo(tsdbReaderT *pReader, STableBlockDistInfo bool isTsdbCacheLastRow(tsdbReaderT *pReader); int32_t tsdbGetAllTableList(SMeta *pMeta, uint64_t uid, SArray *list); int32_t tsdbGetCtbIdList(SMeta *pMeta, int64_t suid, SArray *list); -void *tsdbGetIdx(SMeta *pMeta); +void * tsdbGetIdx(SMeta *pMeta); +void * tsdbGetIvtIdx(SMeta *pMeta); int64_t tsdbGetNumOfRowsInMemTable(tsdbReaderT *pHandle); bool tsdbNextDataBlock(tsdbReaderT pTsdbReadHandle); @@ -197,7 +198,7 @@ struct SMetaEntry { int64_t version; int8_t type; tb_uid_t uid; - char *name; + char * name; union { struct { SSchemaWrapper schemaRow; @@ -225,17 +226,17 @@ struct SMetaEntry { struct SMetaReader { int32_t flags; - SMeta *pMeta; + SMeta * pMeta; SDecoder coder; SMetaEntry me; - void *pBuf; + void * pBuf; int32_t szBuf; }; struct SMTbCursor { - TBC *pDbc; - void *pKey; - void *pVal; + TBC * pDbc; + void * pKey; + void * pVal; int32_t kLen; int32_t vLen; SMetaReader mr; diff --git a/source/dnode/vnode/src/inc/meta.h b/source/dnode/vnode/src/inc/meta.h index b610676c19..63a8e7ec68 100644 --- a/source/dnode/vnode/src/inc/meta.h +++ b/source/dnode/vnode/src/inc/meta.h @@ -69,12 +69,11 @@ struct SMeta { TTB* pUidIdx; TTB* pNameIdx; TTB* pCtbIdx; -#ifdef USE_INVERTED_INDEX + // ivt idx and idx void* pTagIvtIdx; -#else - TTB* pTagIdx; -#endif - TTB* pTtlIdx; + TTB* pTagIdx; + TTB* pTtlIdx; + TTB* pSmaIdx; SMetaIdx* pIdx; }; @@ -117,7 +116,7 @@ typedef struct { } SSmaIdxKey; // metaTable ================== -int metaCreateTagIdxKey(tb_uid_t suid, int32_t cid, const void* pTagData, int32_t nTagData, int8_t type, tb_uid_t uid, +int metaCreateTagIdxKey(tb_uid_t suid, int32_t cid, const void* pTagData, int32_t nTagData, int8_t type, tb_uid_t uid, STagIdxKey** ppTagIdxKey, int32_t* nTagIdxKey); #ifndef META_REFACT diff --git a/source/dnode/vnode/src/inc/vnodeInt.h b/source/dnode/vnode/src/inc/vnodeInt.h index c4da909c21..15469786e0 100644 --- a/source/dnode/vnode/src/inc/vnodeInt.h +++ b/source/dnode/vnode/src/inc/vnodeInt.h @@ -104,6 +104,7 @@ int32_t metaSnapshotReaderOpen(SMeta* pMeta, SMetaSnapshotReader** ppRea int32_t metaSnapshotReaderClose(SMetaSnapshotReader* pReader); int32_t metaSnapshotRead(SMetaSnapshotReader* pReader, void** ppData, uint32_t* nData); void* metaGetIdx(SMeta* pMeta); +void* metaGetIvtIdx(SMeta* pMeta); int32_t metaCreateTSma(SMeta* pMeta, int64_t version, SSmaCfg* pCfg); int32_t metaDropTSma(SMeta* pMeta, int64_t indexUid); diff --git a/source/dnode/vnode/src/meta/metaOpen.c b/source/dnode/vnode/src/meta/metaOpen.c index 86637d2850..95510c25e5 100644 --- a/source/dnode/vnode/src/meta/metaOpen.c +++ b/source/dnode/vnode/src/meta/metaOpen.c @@ -93,7 +93,6 @@ int metaOpen(SVnode *pVnode, SMeta **ppMeta) { } // open pTagIdx -#ifdef USE_INVERTED_INDEX // TODO(yihaoDeng), refactor later char indexFullPath[128] = {0}; sprintf(indexFullPath, "%s/%s", pMeta->path, "invert"); @@ -104,13 +103,11 @@ int metaOpen(SVnode *pVnode, SMeta **ppMeta) { goto _err; } -#else ret = tdbTbOpen("tag.idx", -1, 0, tagIdxKeyCmpr, pMeta->pEnv, &pMeta->pTagIdx); if (ret < 0) { metaError("vgId:%d, failed to open meta tag index since %s", TD_VID(pVnode), tstrerror(terrno)); goto _err; } -#endif // open pTtlIdx ret = tdbTbOpen("ttl.idx", sizeof(STtlIdxKey), 0, ttlIdxKeyCmpr, pMeta->pEnv, &pMeta->pTtlIdx); @@ -141,11 +138,8 @@ _err: if (pMeta->pIdx) metaCloseIdx(pMeta); if (pMeta->pSmaIdx) tdbTbClose(pMeta->pSmaIdx); if (pMeta->pTtlIdx) tdbTbClose(pMeta->pTtlIdx); -#ifdef USE_INVERTED_INDEX if (pMeta->pTagIvtIdx) indexClose(pMeta->pTagIvtIdx); -#else if (pMeta->pTagIdx) tdbTbClose(pMeta->pTagIdx); -#endif if (pMeta->pCtbIdx) tdbTbClose(pMeta->pCtbIdx); if (pMeta->pNameIdx) tdbTbClose(pMeta->pNameIdx); if (pMeta->pUidIdx) tdbTbClose(pMeta->pUidIdx); diff --git a/source/dnode/vnode/src/meta/metaSma.c b/source/dnode/vnode/src/meta/metaSma.c index fde9d30346..689cd511c4 100644 --- a/source/dnode/vnode/src/meta/metaSma.c +++ b/source/dnode/vnode/src/meta/metaSma.c @@ -28,9 +28,9 @@ int32_t metaCreateTSma(SMeta *pMeta, int64_t version, SSmaCfg *pCfg) { int vLen = 0; const void *pKey = NULL; const void *pVal = NULL; - void *pBuf = NULL; + void * pBuf = NULL; int32_t szBuf = 0; - void *p = NULL; + void * p = NULL; SMetaReader mr = {0}; // validate req @@ -83,8 +83,8 @@ int32_t metaDropTSma(SMeta *pMeta, int64_t indexUid) { static int metaSaveSmaToDB(SMeta *pMeta, const SMetaEntry *pME) { STbDbKey tbDbKey; - void *pKey = NULL; - void *pVal = NULL; + void * pKey = NULL; + void * pVal = NULL; int kLen = 0; int vLen = 0; SEncoder coder = {0}; diff --git a/source/dnode/vnode/src/meta/metaTable.c b/source/dnode/vnode/src/meta/metaTable.c index b45fccaf29..4f4a940bc7 100644 --- a/source/dnode/vnode/src/meta/metaTable.c +++ b/source/dnode/vnode/src/meta/metaTable.c @@ -15,6 +15,7 @@ #include "meta.h" +static int metaSaveJsonVarToIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry, const SSchema *pSchema); static int metaHandleEntry(SMeta *pMeta, const SMetaEntry *pME); static int metaSaveToTbDb(SMeta *pMeta, const SMetaEntry *pME); static int metaUpdateUidIdx(SMeta *pMeta, const SMetaEntry *pME); @@ -25,7 +26,7 @@ static int metaUpdateCtbIdx(SMeta *pMeta, const SMetaEntry *pME); static int metaUpdateTagIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry); static int metaDropTableByUid(SMeta *pMeta, tb_uid_t uid, int *type); -static int metaUpdateMetaRsp(tb_uid_t uid, char* tbName, SSchemaWrapper *pSchema, STableMetaRsp *pMetaRsp) { +static int metaUpdateMetaRsp(tb_uid_t uid, char *tbName, SSchemaWrapper *pSchema, STableMetaRsp *pMetaRsp) { pMetaRsp->pSchemas = taosMemoryMalloc(pSchema->nCols * sizeof(SSchema)); if (NULL == pMetaRsp->pSchemas) { terrno = TSDB_CODE_VND_OUT_OF_MEMORY; @@ -43,6 +44,68 @@ static int metaUpdateMetaRsp(tb_uid_t uid, char* tbName, SSchemaWrapper *pSchema return 0; } +static int metaSaveJsonVarToIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry, const SSchema *pSchema) { +#ifdef USE_INVERTED_INDEX + if (pMeta->pTagIvtIdx == NULL || pCtbEntry == NULL) { + return -1; + } + void * data = pCtbEntry->ctbEntry.pTags; + const char *tagName = pSchema->name; + + tb_uid_t suid = pCtbEntry->ctbEntry.suid; + tb_uid_t tuid = pCtbEntry->uid; + const void *pTagData = pCtbEntry->ctbEntry.pTags; + int32_t nTagData = 0; + + SArray *pTagVals = NULL; + if (tTagToValArray((const STag *)data, &pTagVals) != 0) { + return -1; + } + char key[512] = {0}; + + SIndexMultiTerm *terms = indexMultiTermCreate(); + int16_t nCols = taosArrayGetSize(pTagVals); + for (int i = 0; i < nCols; i++) { + STagVal *pTagVal = (STagVal *)taosArrayGet(pTagVals, i); + char type = pTagVal->type; + sprintf(key, "%s_%s", tagName, pTagVal->pKey); + int32_t nKey = strlen(key); + + SIndexTerm *term = NULL; + if (type == TSDB_DATA_TYPE_NULL) { + // handle null value + } else if (type == TSDB_DATA_TYPE_NCHAR) { + if (pTagVal->nData > 0) { + char * val = taosMemoryCalloc(1, pTagVal->nData + VARSTR_HEADER_SIZE); + int32_t len = taosUcs4ToMbs((TdUcs4 *)pTagVal->pData, pTagVal->nData, val + VARSTR_HEADER_SIZE); + memcpy(val, (uint16_t *)&len, VARSTR_HEADER_SIZE); + type = TSDB_DATA_TYPE_VARCHAR; + term = indexTermCreate(suid, ADD_VALUE, type, key, nKey, val, len); + } else if (pTagVal->nData == 0) { + char * val = NULL; + int32_t len = 0; + // handle NULL key + } + } else if (type == TSDB_DATA_TYPE_DOUBLE) { + double val = *(double *)(&pTagVal->i64); + int len = 0; + term = indexTermCreate(suid, ADD_VALUE, type, key, nKey, (const char *)&val, len); + } else if (type == TSDB_DATA_TYPE_BOOL) { + int val = *(int *)(&pTagVal->i64); + int len = 0; + term = indexTermCreate(suid, ADD_VALUE, type, key, nKey, (const char *)&val, len); + } + if (term != NULL) { + indexMultiTermAdd(terms, term); + } + memset(key, 0, sizeof(key)); + } + tIndexJsonPut(pMeta->pTagIvtIdx, terms, tuid); + indexMultiTermDestroy(terms); +#endif + return -1; +} + int metaCreateSTable(SMeta *pMeta, int64_t version, SVCreateStbReq *pReq) { SMetaEntry me = {0}; int kLen = 0; @@ -341,7 +404,6 @@ static int metaDropTableByUid(SMeta *pMeta, tb_uid_t uid, int *type) { return 0; } - static int metaAlterTableColumn(SMeta *pMeta, int64_t version, SVAlterTbReq *pAlterTbReq, STableMetaRsp *pMetaRsp) { void * pVal = NULL; int nVal = 0; @@ -824,28 +886,16 @@ static int metaUpdateTagIdx(SMeta *pMeta, const SMetaEntry *pCtbEntry) { } else { // pTagData = pCtbEntry->ctbEntry.pTags; // nTagData = ((const STag *)pCtbEntry->ctbEntry.pTags)->len; + pTagData = pCtbEntry->ctbEntry.pTags; + nTagData = ((const STag *)pCtbEntry->ctbEntry.pTags)->len; + return metaSaveJsonVarToIdx(pMeta, pCtbEntry, pTagColumn); } - - // update tag index -#ifdef USE_INVERTED_INDEX - tb_uid_t suid = pCtbEntry->ctbEntry.suid; - tb_uid_t tuid = pCtbEntry->uid; - - SIndexMultiTerm *tmGroup = indexMultiTermCreate(); - - SIndexTerm *tm = indexTermCreate(suid, ADD_VALUE, pTagColumn->type, pTagColumn->name, sizeof(pTagColumn->name), - pTagData, pTagData == NULL ? 0 : strlen(pTagData)); - indexMultiTermAdd(tmGroup, tm); - int ret = indexPut((SIndex *)pMeta->pTagIvtIdx, tmGroup, tuid); - indexMultiTermDestroy(tmGroup); -#else if (metaCreateTagIdxKey(pCtbEntry->ctbEntry.suid, pTagColumn->colId, pTagData, nTagData, pTagColumn->type, pCtbEntry->uid, &pTagIdxKey, &nTagIdxKey) < 0) { return -1; } tdbTbInsert(pMeta->pTagIdx, pTagIdxKey, nTagIdxKey, NULL, 0, &pMeta->txn); metaDestroyTagIdxKey(pTagIdxKey); -#endif tDecoderClear(&dc); tdbFree(pData); return 0; @@ -930,10 +980,5 @@ _err: return -1; } // refactor later -void *metaGetIdx(SMeta *pMeta) { -#ifdef USE_INVERTED_INDEX - return pMeta->pTagIvtIdx; -#else - return pMeta->pTagIdx; -#endif -} +void *metaGetIdx(SMeta *pMeta) { return pMeta->pTagIdx; } +void *metaGetIvtIdx(SMeta *pMeta) { return pMeta->pTagIvtIdx; } diff --git a/source/dnode/vnode/src/tsdb/tsdbRead.c b/source/dnode/vnode/src/tsdb/tsdbRead.c index 56220beea4..5fdb052ee5 100644 --- a/source/dnode/vnode/src/tsdb/tsdbRead.c +++ b/source/dnode/vnode/src/tsdb/tsdbRead.c @@ -2834,6 +2834,12 @@ void* tsdbGetIdx(SMeta* pMeta) { } return metaGetIdx(pMeta); } +void* tsdbGetIvtIdx(SMeta* pMeta) { + if (pMeta == NULL) { + return NULL; + } + return metaGetIvtIdx(pMeta); +} int32_t tsdbGetAllTableList(SMeta* pMeta, uint64_t uid, SArray* list) { SMCtbCursor* pCur = metaOpenCtbCursor(pMeta, uid); diff --git a/source/libs/catalog/inc/catalogInt.h b/source/libs/catalog/inc/catalogInt.h index cebe696390..9219a382e4 100644 --- a/source/libs/catalog/inc/catalogInt.h +++ b/source/libs/catalog/inc/catalogInt.h @@ -52,6 +52,7 @@ enum { CTG_OP_UPDATE_VGROUP = 0, CTG_OP_UPDATE_TB_META, CTG_OP_DROP_DB_CACHE, + CTG_OP_DROP_DB_VGROUP, CTG_OP_DROP_STB_META, CTG_OP_DROP_TB_META, CTG_OP_UPDATE_USER, @@ -266,26 +267,32 @@ typedef struct SCtgUpdateTblMsg { STableMetaOutput* output; } SCtgUpdateTblMsg; -typedef struct SCtgRemoveDBMsg { +typedef struct SCtgDropDBMsg { SCatalog* pCtg; char dbFName[TSDB_DB_FNAME_LEN]; 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; char dbFName[TSDB_DB_FNAME_LEN]; char stbName[TSDB_TABLE_NAME_LEN]; uint64_t dbId; uint64_t suid; -} SCtgRemoveStbMsg; +} SCtgDropStbMetaMsg; -typedef struct SCtgRemoveTblMsg { +typedef struct SCtgDropTblMetaMsg { SCatalog* pCtg; char dbFName[TSDB_DB_FNAME_LEN]; char tbName[TSDB_TABLE_NAME_LEN]; uint64_t dbId; -} SCtgRemoveTblMsg; +} SCtgDropTblMetaMsg; typedef struct SCtgUpdateUserMsg { SCatalog* pCtg; @@ -451,6 +458,7 @@ int32_t ctgGetTbMetaFromCache(CTG_PARAMS, SCtgTbMetaCtx* ctx, STableMeta** pTabl int32_t ctgOpUpdateVgroup(SCtgCacheOperation *action); int32_t ctgOpUpdateTbMeta(SCtgCacheOperation *action); int32_t ctgOpDropDbCache(SCtgCacheOperation *action); +int32_t ctgOpDropDbVgroup(SCtgCacheOperation *action); int32_t ctgOpDropStbMeta(SCtgCacheOperation *action); int32_t ctgOpDropTbMeta(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 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 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 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); diff --git a/source/libs/catalog/src/ctgAsync.c b/source/libs/catalog/src/ctgAsync.c index ce94eace66..6998d8c778 100644 --- a/source/libs/catalog/src/ctgAsync.c +++ b/source/libs/catalog/src/ctgAsync.c @@ -286,6 +286,9 @@ int32_t ctgInitJob(CTG_PARAMS, SCtgJob** job, uint64_t reqId, const SCatalogReq* int32_t taskIdx = 0; for (int32_t i = 0; i < dbVgNum; ++i) { char* dbFName = taosArrayGet(pReq->pDbVgroup, i); + if (pReq->forceUpdate) { + ctgDropDbVgroupEnqueue(pCtg, dbFName, true); + } 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) { SName* name = taosArrayGet(pReq->pTableMeta, i); + if (pReq->forceUpdate) { + catalogRemoveTableMeta(pCtg, name); + } CTG_ERR_JRET(ctgInitGetTbMetaTask(pJob, taskIdx++, name)); } diff --git a/source/libs/catalog/src/ctgCache.c b/source/libs/catalog/src/ctgCache.c index 0f1344c343..8332c7b068 100644 --- a/source/libs/catalog/src/ctgCache.c +++ b/source/libs/catalog/src/ctgCache.c @@ -35,6 +35,11 @@ SCtgOperation gCtgCacheOperation[CTG_OP_MAX] = { "drop DB", ctgOpDropDbCache }, + { + CTG_OP_DROP_DB_VGROUP, + "drop DBVgroup", + ctgOpDropDbVgroup + }, { CTG_OP_DROP_STB_META, "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 code = 0; SCtgCacheOperation action= {.opId = CTG_OP_DROP_DB_CACHE}; - SCtgRemoveDBMsg *msg = taosMemoryMalloc(sizeof(SCtgRemoveDBMsg)); + SCtgDropDBMsg *msg = taosMemoryMalloc(sizeof(SCtgDropDBMsg)); 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); } @@ -590,13 +595,43 @@ _return: 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 code = 0; SCtgCacheOperation action= {.opId = CTG_OP_DROP_STB_META, .syncOp = syncOp}; - SCtgRemoveStbMsg *msg = taosMemoryMalloc(sizeof(SCtgRemoveStbMsg)); + SCtgDropStbMetaMsg *msg = taosMemoryMalloc(sizeof(SCtgDropStbMetaMsg)); 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); } @@ -623,9 +658,9 @@ _return: int32_t ctgDropTbMetaEnqueue(SCatalog* pCtg, const char *dbFName, int64_t dbId, const char *tbName, bool syncOp) { int32_t code = 0; SCtgCacheOperation action= {.opId = CTG_OP_DROP_TB_META, .syncOp = syncOp}; - SCtgRemoveTblMsg *msg = taosMemoryMalloc(sizeof(SCtgRemoveTblMsg)); + SCtgDropTblMetaMsg *msg = taosMemoryMalloc(sizeof(SCtgDropTblMetaMsg)); 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); } @@ -1281,7 +1316,7 @@ _return: int32_t ctgOpDropDbCache(SCtgCacheOperation *operation) { int32_t code = 0; - SCtgRemoveDBMsg *msg = operation->data; + SCtgDropDBMsg *msg = operation->data; SCatalog* pCtg = msg->pCtg; SCtgDBCache *dbCache = NULL; @@ -1304,6 +1339,33 @@ _return: 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 code = 0; @@ -1353,7 +1415,7 @@ _return: int32_t ctgOpDropStbMeta(SCtgCacheOperation *operation) { int32_t code = 0; - SCtgRemoveStbMsg *msg = operation->data; + SCtgDropStbMetaMsg *msg = operation->data; SCatalog* pCtg = msg->pCtg; SCtgDBCache *dbCache = NULL; @@ -1399,7 +1461,7 @@ _return: int32_t ctgOpDropTbMeta(SCtgCacheOperation *operation) { int32_t code = 0; - SCtgRemoveTblMsg *msg = operation->data; + SCtgDropTblMetaMsg *msg = operation->data; SCatalog* pCtg = msg->pCtg; SCtgDBCache *dbCache = NULL; diff --git a/source/libs/catalog/src/ctgDbg.c b/source/libs/catalog/src/ctgDbg.c index fdab50db0f..42baa530ce 100644 --- a/source/libs/catalog/src/ctgDbg.c +++ b/source/libs/catalog/src/ctgDbg.c @@ -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; SCatalogReq req = {0}; 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.pUser = taosArrayInit(2, sizeof(SUserAuthInfo)); req.qNodeRequired = true; + req.forceUpdate = forceUpdate; SName name = {0}; char dbFName[TSDB_DB_FNAME_LEN] = {0}; diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c index b4e366cfb7..92b3195f36 100644 --- a/source/libs/executor/src/executorimpl.c +++ b/source/libs/executor/src/executorimpl.c @@ -4389,7 +4389,7 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo SScanPhysiNode* pScanPhyNode = (SScanPhysiNode*)pPhyNode; // simple child table. STableScanPhysiNode* pTableScanNode = (STableScanPhysiNode*)pPhyNode; STimeWindowAggSupp twSup = { - .waterMark = pTableScanNode->watermark, .calTrigger = pTableScanNode->triggerType, .maxTs = INT64_MIN}; + .waterMark = pTableScanNode->watermark, .calTrigger = pTableScanNode->triggerType, .maxTs = INT64_MIN}; tsdbReaderT pDataReader = NULL; if (pHandle->vnode) { pDataReader = doCreateDataReader(pTableScanNode, pHandle, pTableListInfo, (uint64_t)queryId, taskId, pTagCond); @@ -4797,7 +4797,8 @@ int32_t getTableList(void* metaHandle, int32_t tableType, uint64_t tableUid, STa if (tableType == TSDB_SUPER_TABLE) { if (pTagCond) { - SIndexMetaArg metaArg = {.metaEx = metaHandle, .metaHandle = tsdbGetIdx(metaHandle), .suid = tableUid}; + SIndexMetaArg metaArg = { + .metaEx = metaHandle, .idx = tsdbGetIdx(metaHandle), .ivtIdx = tsdbGetIvtIdx(metaHandle), .suid = tableUid}; SArray* res = taosArrayInit(8, sizeof(uint64_t)); code = doFilterTag(pTagCond, &metaArg, res); diff --git a/source/libs/index/src/index.c b/source/libs/index/src/index.c index ba3aea969f..058f3c1915 100644 --- a/source/libs/index/src/index.c +++ b/source/libs/index/src/index.c @@ -202,7 +202,7 @@ int indexPut(SIndex* index, SIndexMultiTerm* fVals, uint64_t uid) { char buf[128] = {0}; ICacheKey key = {.suid = p->suid, .colName = p->colName, .nColName = strlen(p->colName), .colType = p->colType}; int32_t sz = indexSerialCacheKey(&key, buf); - indexDebug("suid: %" PRIu64 ", colName: %s, colType: %d", key.suid, key.colName, key.colType); + indexDebug("w suid: %" PRIu64 ", colName: %s, colType: %d", key.suid, key.colName, key.colType); IndexCache** cache = taosHashGet(index->colObj, buf, sz); assert(*cache != NULL); @@ -330,7 +330,7 @@ static int indexTermSearch(SIndex* sIdx, SIndexTermQuery* query, SArray** result char buf[128] = {0}; ICacheKey key = { .suid = term->suid, .colName = term->colName, .nColName = strlen(term->colName), .colType = term->colType}; - indexDebug("suid: %" PRIu64 ", colName: %s, colType: %d", key.suid, key.colName, key.colType); + indexDebug("r suid: %" PRIu64 ", colName: %s, colType: %d", key.suid, key.colName, key.colType); int32_t sz = indexSerialCacheKey(&key, buf); taosThreadMutexLock(&sIdx->mtx); diff --git a/source/libs/index/src/indexComm.c b/source/libs/index/src/indexComm.c index 5310e1c345..74b1773e87 100644 --- a/source/libs/index/src/indexComm.c +++ b/source/libs/index/src/indexComm.c @@ -402,16 +402,16 @@ int32_t indexConvertDataToStr(void* src, int8_t type, void** dst) { break; } case TSDB_DATA_TYPE_VARCHAR: { // TSDB_DATA_TYPE_BINARY - tlen = taosEncodeBinary(NULL, src, strlen(src)); + tlen = taosEncodeBinary(NULL, varDataVal(src), varDataLen(src)); *dst = taosMemoryCalloc(1, tlen + 1); - tlen = taosEncodeBinary(dst, src, strlen(src)); + tlen = taosEncodeBinary(dst, varDataVal(src), varDataLen(src)); *dst = (char*)*dst - tlen; break; } case TSDB_DATA_TYPE_VARBINARY: - tlen = taosEncodeBinary(NULL, src, strlen(src)); + tlen = taosEncodeBinary(NULL, varDataVal(src), varDataLen(src)); *dst = taosMemoryCalloc(1, tlen + 1); - tlen = taosEncodeBinary(dst, src, strlen(src)); + tlen = taosEncodeBinary(dst, varDataVal(src), varDataLen(src)); *dst = (char*)*dst - tlen; break; default: diff --git a/source/libs/index/src/indexFilter.c b/source/libs/index/src/indexFilter.c index 306490b21f..34251ab918 100644 --- a/source/libs/index/src/indexFilter.c +++ b/source/libs/index/src/indexFilter.c @@ -31,7 +31,7 @@ typedef struct SIFParam { SHashObj *pFilter; SArray *result; - char *condValue; + char * condValue; SIdxFltStatus status; uint8_t colValType; @@ -45,7 +45,7 @@ typedef struct SIFParam { typedef struct SIFCtx { int32_t code; - SHashObj *pRes; /* element is SIFParam */ + SHashObj * pRes; /* element is SIFParam */ bool noExec; // true: just iterate condition tree, and add hint to executor plan SIndexMetaArg arg; // SIdxFltStatus st; @@ -128,7 +128,7 @@ static int32_t sifGetValueFromNode(SNode *node, char **value) { // covert data From snode; SValueNode *vn = (SValueNode *)node; - char *pData = nodesGetValueFromNode(vn); + char * pData = nodesGetValueFromNode(vn); SDataType *pType = &vn->node.resType; int32_t type = pType->type; int32_t valLen = 0; @@ -162,12 +162,27 @@ static int32_t sifGetValueFromNode(SNode *node, char **value) { return TSDB_CODE_SUCCESS; } +static int32_t sifInitJsonParam(SNode *node, SIFParam *param, SIFCtx *ctx) { + SOperatorNode *nd = (SOperatorNode *)node; + assert(nodeType(node) == QUERY_NODE_OPERATOR); + SColumnNode *l = (SColumnNode *)nd->pLeft; + SValueNode * r = (SValueNode *)nd->pRight; + + param->colId = l->colId; + param->colValType = l->node.resType.type; + memcpy(param->dbName, l->dbName, sizeof(l->dbName)); + sprintf(param->colName, "%s_%s", l->colName, r->literal); + param->colValType = r->typeData; + return 0; + // memcpy(param->colName, l->colName, sizeof(l->colName)); +} static int32_t sifInitParam(SNode *node, SIFParam *param, SIFCtx *ctx) { switch (nodeType(node)) { case QUERY_NODE_VALUE: { SValueNode *vn = (SValueNode *)node; SIF_ERR_RET(sifGetValueFromNode(node, ¶m->condValue)); param->colId = -1; + param->colValType = (uint8_t)(vn->node.resType.type); break; } case QUERY_NODE_COLUMN: { @@ -219,17 +234,31 @@ static int32_t sifInitOperParams(SIFParam **params, SOperatorNode *node, SIFCtx indexError("invalid operation node, left: %p, rigth: %p", node->pLeft, node->pRight); SIF_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT); } + if (node->opType == OP_TYPE_JSON_GET_VALUE || node->opType == OP_TYPE_JSON_CONTAINS) { + return code; + } SIFParam *paramList = taosMemoryCalloc(nParam, sizeof(SIFParam)); if (NULL == paramList) { SIF_ERR_RET(TSDB_CODE_QRY_OUT_OF_MEMORY); } - SIF_ERR_JRET(sifInitParam(node->pLeft, ¶mList[0], ctx)); - if (nParam > 1) { - SIF_ERR_JRET(sifInitParam(node->pRight, ¶mList[1], ctx)); + if (nodeType(node->pLeft) == QUERY_NODE_OPERATOR) { + SNode *interNode = (node->pLeft); + SIF_ERR_JRET(sifInitJsonParam(interNode, ¶mList[0], ctx)); + if (nParam > 1) { + SIF_ERR_JRET(sifInitParam(node->pRight, ¶mList[1], ctx)); + } + paramList[0].colValType = TSDB_DATA_TYPE_JSON; + *params = paramList; + return TSDB_CODE_SUCCESS; + } else { + SIF_ERR_JRET(sifInitParam(node->pLeft, ¶mList[0], ctx)); + if (nParam > 1) { + SIF_ERR_JRET(sifInitParam(node->pRight, ¶mList[1], ctx)); + } + *params = paramList; + return TSDB_CODE_SUCCESS; } - *params = paramList; - return TSDB_CODE_SUCCESS; _return: taosMemoryFree(paramList); SIF_RET(code); @@ -306,39 +335,34 @@ static Filter sifGetFilterFunc(EIndexQueryType type, bool *reverse) { } static int32_t sifDoIndex(SIFParam *left, SIFParam *right, int8_t operType, SIFParam *output) { SIndexMetaArg *arg = &output->arg; -#ifdef USE_INVERTED_INDEX - SIndexTerm *tm = indexTermCreate(arg->suid, DEFAULT, left->colValType, left->colName, strlen(left->colName), - right->condValue, strlen(right->condValue)); - if (tm == NULL) { - return TSDB_CODE_QRY_OUT_OF_MEMORY; + int ret = 0; + + EIndexQueryType qtype = 0; + SIF_ERR_RET(sifGetFuncFromSql(operType, &qtype)); + if (left->colValType == TSDB_DATA_TYPE_JSON) { + SIndexTerm *tm = indexTermCreate(arg->suid, DEFAULT, right->colValType, left->colName, strlen(left->colName), + right->condValue, strlen(right->condValue)); + if (tm == NULL) { + return TSDB_CODE_QRY_OUT_OF_MEMORY; + } + + SIndexMultiTermQuery *mtm = indexMultiTermQueryCreate(MUST); + indexMultiTermQueryAdd(mtm, tm, qtype); + ret = tIndexJsonSearch(arg->ivtIdx, mtm, output->result); + } else { + bool reverse; + Filter filterFunc = sifGetFilterFunc(qtype, &reverse); + + SMetaFltParam param = {.suid = arg->suid, + .cid = left->colId, + .type = left->colValType, + .val = right->condValue, + .reverse = reverse, + .filterFunc = filterFunc}; + + ret = metaFilteTableIds(arg->metaEx, ¶m, output->result); } - - EIndexQueryType qtype = 0; - SIF_ERR_RET(sifGetFuncFromSql(operType, &qtype)); - - SIndexMultiTermQuery *mtm = indexMultiTermQueryCreate(MUST); - indexMultiTermQueryAdd(mtm, tm, qtype); - int ret = indexSearch(arg->metaHandle, mtm, output->result); - indexDebug("index filter data size: %d", (int)taosArrayGetSize(output->result)); - indexMultiTermQueryDestroy(mtm); return ret; -#else - EIndexQueryType qtype = 0; - SIF_ERR_RET(sifGetFuncFromSql(operType, &qtype)); - bool reverse; - Filter filterFunc = sifGetFilterFunc(qtype, &reverse); - - SMetaFltParam param = {.suid = arg->suid, - .cid = left->colId, - .type = left->colValType, - .val = right->condValue, - .reverse = reverse, - .filterFunc = filterFunc}; - - int ret = metaFilteTableIds(arg->metaEx, ¶m, output->result); - return ret; -#endif - return 0; } static int32_t sifLessThanFunc(SIFParam *left, SIFParam *right, SIFParam *output) { @@ -392,6 +416,14 @@ static int32_t sifNotMatchFunc(SIFParam *left, SIFParam *right, SIFParam *output int id = OP_TYPE_NMATCH; return sifDoIndex(left, right, id, output); } +static int32_t sifJsonContains(SIFParam *left, SIFParam *right, SIFParam *output) { + // return 0 + return 0; +} +static int32_t sifJsonGetValue(SIFParam *left, SIFParam *rigth, SIFParam *output) { + // return 0 + return 0; +} static int32_t sifDefaultFunc(SIFParam *left, SIFParam *right, SIFParam *output) { // add more except @@ -445,6 +477,14 @@ static int32_t sifGetOperFn(int32_t funcId, sif_func_t *func, SIdxFltStatus *sta *status = SFLT_NOT_INDEX; *func = sifNotMatchFunc; return 0; + case OP_TYPE_JSON_CONTAINS: + *status = SFLT_ACCURATE_INDEX; + *func = sifJsonContains; + return 0; + case OP_TYPE_JSON_GET_VALUE: + *status = SFLT_ACCURATE_INDEX; + *func = sifJsonGetValue; + return 0; default: *status = SFLT_NOT_INDEX; *func = sifNullFunc; @@ -457,12 +497,17 @@ static int32_t sifGetOperFn(int32_t funcId, sif_func_t *func, SIdxFltStatus *sta static int32_t sifExecOper(SOperatorNode *node, SIFCtx *ctx, SIFParam *output) { int32_t code = 0; int32_t nParam = sifGetOperParamNum(node->opType); + if (nParam <= 1) { + SIF_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT); + } + if (node->opType == OP_TYPE_JSON_GET_VALUE || node->opType == OP_TYPE_JSON_CONTAINS) { + return code; + } SIFParam *params = NULL; - SIF_ERR_RET(sifInitOperParams(¶ms, node, ctx)); + SIF_ERR_RET(sifInitOperParams(¶ms, node, ctx)); // ugly code, refactor later output->arg = ctx->arg; - sif_func_t operFn = sifNullFunc; code = sifGetOperFn(node->opType, &operFn, &output->status); if (!ctx->noExec) { @@ -567,7 +612,9 @@ EDealRes sifCalcWalker(SNode *node, void *context) { if (QUERY_NODE_LOGIC_CONDITION == nodeType(node)) { return sifWalkLogic(node, ctx); } + if (QUERY_NODE_OPERATOR == nodeType(node)) { + indexInfo("node type for index filter, type: %d", nodeType(node)); return sifWalkOper(node, ctx); } @@ -658,7 +705,7 @@ int32_t doFilterTag(const SNode *pFilterNode, SIndexMetaArg *metaArg, SArray *re // todo move to the initialization function // SIF_ERR_RET(filterInitFromNode((SNode *)pFilterNode, &filter, 0)); - SArray *output = taosArrayInit(8, sizeof(uint64_t)); + SArray * output = taosArrayInit(8, sizeof(uint64_t)); SIFParam param = {.arg = *metaArg, .result = output}; SIF_ERR_RET(sifCalculate((SNode *)pFilterNode, ¶m)); diff --git a/source/libs/index/test/indexTests.cc b/source/libs/index/test/indexTests.cc index 74a30c3387..327a4ae219 100644 --- a/source/libs/index/test/indexTests.cc +++ b/source/libs/index/test/indexTests.cc @@ -498,10 +498,15 @@ TEST_F(IndexTFileEnv, test_tfile_write) { } taosArrayDestroy(data); - std::string colName("voltage"); - std::string colVal("ab"); - SIndexTerm* term = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), - colVal.c_str(), colVal.size()); + std::string colName("voltage"); + std::string colVal("ab"); + + char buf[256] = {0}; + int16_t sz = colVal.size(); + memcpy(buf, (uint16_t*)&sz, 2); + memcpy(buf + 2, colVal.c_str(), colVal.size()); + SIndexTerm* term = + indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), buf, sizeof(buf)); SIndexTermQuery query = {term, QUERY_TERM}; SArray* result = (SArray*)taosArrayInit(1, sizeof(uint64_t)); @@ -564,6 +569,18 @@ class IndexCacheEnv : public ::testing::Test { CacheObj* coj; }; +SIndexTerm* indexTermCreateT(int64_t suid, SIndexOperOnColumn oper, uint8_t colType, const char* colName, + int32_t nColName, const char* colVal, int32_t nColVal) { + char buf[256] = {0}; + int16_t sz = nColVal; + memcpy(buf, (uint16_t*)&sz, 2); + memcpy(buf + 2, colVal, nColVal); + if (colType == TSDB_DATA_TYPE_BINARY) { + return indexTermCreate(suid, oper, colType, colName, nColName, buf, sizeof(buf)); + } else { + return indexTermCreate(suid, oper, colType, colName, nColName, colVal, nColVal); + } +} #define MAX_TERM_KEY_LEN 128 TEST_F(IndexCacheEnv, cache_test) { int version = 0; @@ -574,37 +591,37 @@ TEST_F(IndexCacheEnv, cache_test) { std::string colName("voltage"); { std::string colVal("v1"); - SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), - colVal.c_str(), colVal.size()); + SIndexTerm* term = indexTermCreateT(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), + colVal.c_str(), colVal.size()); coj->Put(term, colId, version++, suid++); indexTermDestroy(term); // indexTermDestry(term); } { std::string colVal("v3"); - SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), - colVal.c_str(), colVal.size()); + SIndexTerm* term = indexTermCreateT(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), + colVal.c_str(), colVal.size()); coj->Put(term, colId, version++, suid++); indexTermDestroy(term); } { std::string colVal("v2"); - SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), - colVal.c_str(), colVal.size()); + SIndexTerm* term = indexTermCreateT(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), + colVal.c_str(), colVal.size()); coj->Put(term, colId, version++, suid++); indexTermDestroy(term); } { std::string colVal("v3"); - SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), - colVal.c_str(), colVal.size()); + SIndexTerm* term = indexTermCreateT(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), + colVal.c_str(), colVal.size()); coj->Put(term, colId, version++, suid++); indexTermDestroy(term); } { std::string colVal("v3"); - SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), - colVal.c_str(), colVal.size()); + SIndexTerm* term = indexTermCreateT(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), + colVal.c_str(), colVal.size()); coj->Put(term, colId, version++, suid++); indexTermDestroy(term); } @@ -612,15 +629,15 @@ TEST_F(IndexCacheEnv, cache_test) { std::cout << "--------first----------" << std::endl; { std::string colVal("v3"); - SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), - colVal.c_str(), colVal.size()); + SIndexTerm* term = indexTermCreateT(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), + colVal.c_str(), colVal.size()); coj->Put(term, othColId, version++, suid++); indexTermDestroy(term); } { std::string colVal("v4"); - SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), - colVal.c_str(), colVal.size()); + SIndexTerm* term = indexTermCreateT(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), + colVal.c_str(), colVal.size()); coj->Put(term, othColId, version++, suid++); indexTermDestroy(term); } @@ -630,8 +647,8 @@ TEST_F(IndexCacheEnv, cache_test) { std::string colVal("v4"); for (size_t i = 0; i < 10; i++) { colVal[colVal.size() - 1] = 'a' + i; - SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), - colVal.c_str(), colVal.size()); + SIndexTerm* term = indexTermCreateT(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), + colVal.c_str(), colVal.size()); coj->Put(term, colId, version++, suid++); indexTermDestroy(term); } @@ -640,8 +657,8 @@ TEST_F(IndexCacheEnv, cache_test) { // begin query { std::string colVal("v3"); - SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), - colVal.c_str(), colVal.size()); + SIndexTerm* term = indexTermCreateT(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), + colVal.c_str(), colVal.size()); SIndexTermQuery query = {term, QUERY_TERM}; SArray* ret = (SArray*)taosArrayInit(4, sizeof(suid)); STermValueType valType; @@ -655,8 +672,8 @@ TEST_F(IndexCacheEnv, cache_test) { } { std::string colVal("v2"); - SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), - colVal.c_str(), colVal.size()); + SIndexTerm* term = indexTermCreateT(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), + colVal.c_str(), colVal.size()); SIndexTermQuery query = {term, QUERY_TERM}; SArray* ret = (SArray*)taosArrayInit(4, sizeof(suid)); STermValueType valType; @@ -690,8 +707,8 @@ class IndexObj { return ret; } void Del(const std::string& colName, const std::string& colVal, uint64_t uid) { - SIndexTerm* term = indexTermCreate(0, DEL_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), - colVal.c_str(), colVal.size()); + SIndexTerm* term = indexTermCreateT(0, DEL_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), + colVal.c_str(), colVal.size()); SIndexMultiTerm* terms = indexMultiTermCreate(); indexMultiTermAdd(terms, term); Put(terms, uid); @@ -699,8 +716,8 @@ class IndexObj { } int WriteMillonData(const std::string& colName, const std::string& colVal = "Hello world", size_t numOfTable = 100 * 10000) { - SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), - colVal.c_str(), colVal.size()); + SIndexTerm* term = indexTermCreateT(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), + colVal.c_str(), colVal.size()); SIndexMultiTerm* terms = indexMultiTermCreate(); indexMultiTermAdd(terms, term); for (size_t i = 0; i < numOfTable; i++) { @@ -721,8 +738,8 @@ class IndexObj { // opt tColVal[taosRand() % colValSize] = 'a' + k % 26; } - SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), - tColVal.c_str(), tColVal.size()); + SIndexTerm* term = indexTermCreateT(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), + tColVal.c_str(), tColVal.size()); SIndexMultiTerm* terms = indexMultiTermCreate(); indexMultiTermAdd(terms, term); for (size_t j = 0; j < skip; j++) { @@ -757,8 +774,8 @@ class IndexObj { int SearchOne(const std::string& colName, const std::string& colVal) { SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST); - SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), - colVal.c_str(), colVal.size()); + SIndexTerm* term = indexTermCreateT(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), + colVal.c_str(), colVal.size()); indexMultiTermQueryAdd(mq, term, QUERY_TERM); SArray* result = (SArray*)taosArrayInit(1, sizeof(uint64_t)); @@ -779,8 +796,8 @@ class IndexObj { } int SearchOneTarget(const std::string& colName, const std::string& colVal, uint64_t val) { SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST); - SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), - colVal.c_str(), colVal.size()); + SIndexTerm* term = indexTermCreateT(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), + colVal.c_str(), colVal.size()); indexMultiTermQueryAdd(mq, term, QUERY_TERM); SArray* result = (SArray*)taosArrayInit(1, sizeof(uint64_t)); @@ -804,16 +821,16 @@ class IndexObj { void PutOne(const std::string& colName, const std::string& colVal) { SIndexMultiTerm* terms = indexMultiTermCreate(); - SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), - colVal.c_str(), colVal.size()); + SIndexTerm* term = indexTermCreateT(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), + colVal.c_str(), colVal.size()); indexMultiTermAdd(terms, term); Put(terms, 10); indexMultiTermDestroy(terms); } void PutOneTarge(const std::string& colName, const std::string& colVal, uint64_t val) { SIndexMultiTerm* terms = indexMultiTermCreate(); - SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), - colVal.c_str(), colVal.size()); + SIndexTerm* term = indexTermCreateT(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), + colVal.c_str(), colVal.size()); indexMultiTermAdd(terms, term); Put(terms, val); indexMultiTermDestroy(terms); @@ -858,8 +875,8 @@ TEST_F(IndexEnv2, testIndexOpen) { { std::string colName("tag1"), colVal("Hello"); - SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), - colVal.c_str(), colVal.size()); + SIndexTerm* term = indexTermCreateT(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), + colVal.c_str(), colVal.size()); SIndexMultiTerm* terms = indexMultiTermCreate(); indexMultiTermAdd(terms, term); for (size_t i = 0; i < targetSize; i++) { @@ -873,8 +890,8 @@ TEST_F(IndexEnv2, testIndexOpen) { size_t size = 200; std::string colName("tag1"), colVal("hello"); - SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), - colVal.c_str(), colVal.size()); + SIndexTerm* term = indexTermCreateT(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), + colVal.c_str(), colVal.size()); SIndexMultiTerm* terms = indexMultiTermCreate(); indexMultiTermAdd(terms, term); for (size_t i = 0; i < size; i++) { @@ -888,8 +905,8 @@ TEST_F(IndexEnv2, testIndexOpen) { size_t size = 200; std::string colName("tag1"), colVal("Hello"); - SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), - colVal.c_str(), colVal.size()); + SIndexTerm* term = indexTermCreateT(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), + colVal.c_str(), colVal.size()); SIndexMultiTerm* terms = indexMultiTermCreate(); indexMultiTermAdd(terms, term); for (size_t i = size * 3; i < size * 4; i++) { @@ -903,8 +920,8 @@ TEST_F(IndexEnv2, testIndexOpen) { { std::string colName("tag1"), colVal("Hello"); SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST); - SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), - colVal.c_str(), colVal.size()); + SIndexTerm* term = indexTermCreateT(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), + colVal.c_str(), colVal.size()); indexMultiTermQueryAdd(mq, term, QUERY_TERM); SArray* result = (SArray*)taosArrayInit(1, sizeof(uint64_t)); @@ -926,8 +943,8 @@ TEST_F(IndexEnv2, testEmptyIndexOpen) { { std::string colName("tag1"), colVal("Hello"); - SIndexTerm* term = indexTermCreate(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), - colVal.c_str(), colVal.size()); + SIndexTerm* term = indexTermCreateT(0, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), + colVal.c_str(), colVal.size()); SIndexMultiTerm* terms = indexMultiTermCreate(); indexMultiTermAdd(terms, term); for (size_t i = 0; i < targetSize; i++) { diff --git a/source/libs/index/test/jsonUT.cc b/source/libs/index/test/jsonUT.cc index 48ce8839c4..f795fbad01 100644 --- a/source/libs/index/test/jsonUT.cc +++ b/source/libs/index/test/jsonUT.cc @@ -19,6 +19,18 @@ static std::string dir = TD_TMP_DIR_PATH "json"; static std::string logDir = TD_TMP_DIR_PATH "log"; +SIndexTerm* indexTermCreateT(int64_t suid, SIndexOperOnColumn oper, uint8_t colType, const char* colName, + int32_t nColName, const char* colVal, int32_t nColVal) { + char buf[256] = {0}; + int16_t sz = nColVal; + memcpy(buf, (uint16_t*)&sz, 2); + memcpy(buf + 2, colVal, nColVal); + if (colType == TSDB_DATA_TYPE_BINARY) { + return indexTermCreate(suid, oper, colType, colName, nColName, buf, sizeof(buf)); + } else { + return indexTermCreate(suid, oper, colType, colName, nColName, colVal, nColVal); + } +} static void initLog() { const char* defaultLogFileNamePrefix = "taoslog"; const int32_t maxLogFileNum = 10; @@ -59,8 +71,8 @@ class JsonEnv : public ::testing::Test { static void WriteData(SIndexJson* index, const std::string& colName, int8_t dtype, void* data, int dlen, int tableId, int8_t operType = ADD_VALUE) { - SIndexTerm* term = - indexTermCreate(1, (SIndexOperOnColumn)operType, dtype, colName.c_str(), colName.size(), (const char*)data, dlen); + SIndexTerm* term = indexTermCreateT(1, (SIndexOperOnColumn)operType, dtype, colName.c_str(), colName.size(), + (const char*)data, dlen); SIndexMultiTerm* terms = indexMultiTermCreate(); indexMultiTermAdd(terms, term); tIndexJsonPut(index, terms, (int64_t)tableId); @@ -70,8 +82,8 @@ static void WriteData(SIndexJson* index, const std::string& colName, int8_t dtyp static void delData(SIndexJson* index, const std::string& colName, int8_t dtype, void* data, int dlen, int tableId, int8_t operType = DEL_VALUE) { - SIndexTerm* term = - indexTermCreate(1, (SIndexOperOnColumn)operType, dtype, colName.c_str(), colName.size(), (const char*)data, dlen); + SIndexTerm* term = indexTermCreateT(1, (SIndexOperOnColumn)operType, dtype, colName.c_str(), colName.size(), + (const char*)data, dlen); SIndexMultiTerm* terms = indexMultiTermCreate(); indexMultiTermAdd(terms, term); tIndexJsonPut(index, terms, (int64_t)tableId); @@ -83,7 +95,7 @@ static void Search(SIndexJson* index, const std::string& colNam, int8_t dtype, v std::string colName(colNam); SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST); - SIndexTerm* q = indexTermCreate(1, ADD_VALUE, dtype, colName.c_str(), colName.size(), (const char*)data, dlen); + SIndexTerm* q = indexTermCreateT(1, ADD_VALUE, dtype, colName.c_str(), colName.size(), (const char*)data, dlen); SArray* res = taosArrayInit(1, sizeof(uint64_t)); indexMultiTermQueryAdd(mq, q, (EIndexQueryType)filterType); @@ -95,8 +107,8 @@ TEST_F(JsonEnv, testWrite) { { std::string colName("test"); std::string colVal("ab"); - SIndexTerm* term = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), - colVal.c_str(), colVal.size()); + SIndexTerm* term = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), + colVal.c_str(), colVal.size()); SIndexMultiTerm* terms = indexMultiTermCreate(); indexMultiTermAdd(terms, term); @@ -108,8 +120,8 @@ TEST_F(JsonEnv, testWrite) { { std::string colName("voltage"); std::string colVal("ab1"); - SIndexTerm* term = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), - colVal.c_str(), colVal.size()); + SIndexTerm* term = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), + colVal.c_str(), colVal.size()); SIndexMultiTerm* terms = indexMultiTermCreate(); indexMultiTermAdd(terms, term); @@ -121,8 +133,8 @@ TEST_F(JsonEnv, testWrite) { { std::string colName("voltage"); std::string colVal("123"); - SIndexTerm* term = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), - colVal.c_str(), colVal.size()); + SIndexTerm* term = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), + colVal.c_str(), colVal.size()); SIndexMultiTerm* terms = indexMultiTermCreate(); indexMultiTermAdd(terms, term); @@ -136,8 +148,8 @@ TEST_F(JsonEnv, testWrite) { std::string colVal("ab"); SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST); - SIndexTerm* q = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), - colVal.c_str(), colVal.size()); + SIndexTerm* q = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), + colVal.c_str(), colVal.size()); SArray* result = taosArrayInit(1, sizeof(uint64_t)); indexMultiTermQueryAdd(mq, q, QUERY_TERM); @@ -150,8 +162,8 @@ TEST_F(JsonEnv, testWriteMillonData) { { std::string colName("test"); std::string colVal("ab"); - SIndexTerm* term = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), - colVal.c_str(), colVal.size()); + SIndexTerm* term = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), + colVal.c_str(), colVal.size()); SIndexMultiTerm* terms = indexMultiTermCreate(); indexMultiTermAdd(terms, term); @@ -165,8 +177,8 @@ TEST_F(JsonEnv, testWriteMillonData) { std::string colVal("abxxxxxxxxxxxx"); for (int i = 0; i < 10; i++) { colVal[i % colVal.size()] = '0' + i % 128; - SIndexTerm* term = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), - colVal.c_str(), colVal.size()); + SIndexTerm* term = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), + colVal.c_str(), colVal.size()); SIndexMultiTerm* terms = indexMultiTermCreate(); indexMultiTermAdd(terms, term); @@ -179,8 +191,8 @@ TEST_F(JsonEnv, testWriteMillonData) { { std::string colName("voltagefdadfa"); std::string colVal("abxxxxxxxxxxxx"); - SIndexTerm* term = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), - colVal.c_str(), colVal.size()); + SIndexTerm* term = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), + colVal.c_str(), colVal.size()); SIndexMultiTerm* terms = indexMultiTermCreate(); indexMultiTermAdd(terms, term); @@ -194,8 +206,8 @@ TEST_F(JsonEnv, testWriteMillonData) { std::string colVal("ab"); SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST); - SIndexTerm* q = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), - colVal.c_str(), colVal.size()); + SIndexTerm* q = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), + colVal.c_str(), colVal.size()); SArray* result = taosArrayInit(1, sizeof(uint64_t)); indexMultiTermQueryAdd(mq, q, QUERY_TERM); @@ -209,8 +221,8 @@ TEST_F(JsonEnv, testWriteMillonData) { std::string colVal("ab"); SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST); - SIndexTerm* q = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), - colVal.c_str(), colVal.size()); + SIndexTerm* q = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), + colVal.c_str(), colVal.size()); SArray* result = taosArrayInit(1, sizeof(uint64_t)); indexMultiTermQueryAdd(mq, q, QUERY_GREATER_THAN); @@ -224,8 +236,8 @@ TEST_F(JsonEnv, testWriteMillonData) { std::string colVal("ab"); SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST); - SIndexTerm* q = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), - colVal.c_str(), colVal.size()); + SIndexTerm* q = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), + colVal.c_str(), colVal.size()); SArray* result = taosArrayInit(1, sizeof(uint64_t)); indexMultiTermQueryAdd(mq, q, QUERY_GREATER_EQUAL); @@ -241,8 +253,8 @@ TEST_F(JsonEnv, testWriteJsonNumberData) { std::string colName("test"); // std::string colVal("10"); int val = 10; - SIndexTerm* term = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(), - (const char*)&val, sizeof(val)); + SIndexTerm* term = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(), + (const char*)&val, sizeof(val)); SIndexMultiTerm* terms = indexMultiTermCreate(); indexMultiTermAdd(terms, term); @@ -254,8 +266,8 @@ TEST_F(JsonEnv, testWriteJsonNumberData) { { std::string colName("test2"); int val = 20; - SIndexTerm* term = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(), - (const char*)&val, sizeof(val)); + SIndexTerm* term = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(), + (const char*)&val, sizeof(val)); SIndexMultiTerm* terms = indexMultiTermCreate(); indexMultiTermAdd(terms, term); @@ -267,8 +279,8 @@ TEST_F(JsonEnv, testWriteJsonNumberData) { { std::string colName("test"); int val = 15; - SIndexTerm* term = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(), - (const char*)&val, sizeof(val)); + SIndexTerm* term = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(), + (const char*)&val, sizeof(val)); SIndexMultiTerm* terms = indexMultiTermCreate(); indexMultiTermAdd(terms, term); @@ -280,8 +292,8 @@ TEST_F(JsonEnv, testWriteJsonNumberData) { { std::string colName("test2"); const char* val = "test"; - SIndexTerm* term = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), - (const char*)val, strlen(val)); + SIndexTerm* term = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), + (const char*)val, strlen(val)); SIndexMultiTerm* terms = indexMultiTermCreate(); indexMultiTermAdd(terms, term); @@ -294,8 +306,8 @@ TEST_F(JsonEnv, testWriteJsonNumberData) { std::string colName("test"); int val = 15; SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST); - SIndexTerm* q = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(), - (const char*)&val, sizeof(val)); + SIndexTerm* q = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(), + (const char*)&val, sizeof(val)); SArray* result = taosArrayInit(1, sizeof(uint64_t)); indexMultiTermQueryAdd(mq, q, QUERY_TERM); @@ -308,8 +320,8 @@ TEST_F(JsonEnv, testWriteJsonNumberData) { int val = 15; SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST); - SIndexTerm* q = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(), - (const char*)&val, sizeof(val)); + SIndexTerm* q = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(), + (const char*)&val, sizeof(val)); SArray* result = taosArrayInit(1, sizeof(uint64_t)); indexMultiTermQueryAdd(mq, q, QUERY_GREATER_THAN); @@ -323,8 +335,8 @@ TEST_F(JsonEnv, testWriteJsonNumberData) { ; SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST); - SIndexTerm* q = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(), - (const char*)&val, sizeof(int)); + SIndexTerm* q = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(), + (const char*)&val, sizeof(int)); SArray* result = taosArrayInit(1, sizeof(uint64_t)); indexMultiTermQueryAdd(mq, q, QUERY_GREATER_EQUAL); @@ -338,8 +350,8 @@ TEST_F(JsonEnv, testWriteJsonNumberData) { // std::string colVal("10"); SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST); - SIndexTerm* q = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(), - (const char*)&val, sizeof(val)); + SIndexTerm* q = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(), + (const char*)&val, sizeof(val)); SArray* result = taosArrayInit(1, sizeof(uint64_t)); indexMultiTermQueryAdd(mq, q, QUERY_LESS_THAN); @@ -353,8 +365,8 @@ TEST_F(JsonEnv, testWriteJsonNumberData) { // std::string colVal("10"); SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST); - SIndexTerm* q = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(), - (const char*)&val, sizeof(val)); + SIndexTerm* q = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(), + (const char*)&val, sizeof(val)); SArray* result = taosArrayInit(1, sizeof(uint64_t)); indexMultiTermQueryAdd(mq, q, QUERY_LESS_EQUAL); @@ -368,8 +380,8 @@ TEST_F(JsonEnv, testWriteJsonTfileAndCache_INT) { { std::string colName("test1"); int val = 10; - SIndexTerm* term = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(), - (const char*)&val, sizeof(val)); + SIndexTerm* term = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(), + (const char*)&val, sizeof(val)); SIndexMultiTerm* terms = indexMultiTermCreate(); indexMultiTermAdd(terms, term); @@ -381,8 +393,8 @@ TEST_F(JsonEnv, testWriteJsonTfileAndCache_INT) { { std::string colName("test"); std::string colVal("xxxxxxxxxxxxxxxxxxx"); - SIndexTerm* term = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), - colVal.c_str(), colVal.size()); + SIndexTerm* term = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_BINARY, colName.c_str(), colName.size(), + colVal.c_str(), colVal.size()); SIndexMultiTerm* terms = indexMultiTermCreate(); indexMultiTermAdd(terms, term); @@ -396,8 +408,8 @@ TEST_F(JsonEnv, testWriteJsonTfileAndCache_INT) { int val = 10; SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST); - SIndexTerm* q = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(), - (const char*)&val, sizeof(val)); + SIndexTerm* q = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(), + (const char*)&val, sizeof(val)); SArray* result = taosArrayInit(1, sizeof(uint64_t)); indexMultiTermQueryAdd(mq, q, QUERY_TERM); @@ -410,8 +422,8 @@ TEST_F(JsonEnv, testWriteJsonTfileAndCache_INT) { int val = 10; SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST); - SIndexTerm* q = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(), - (const char*)&val, sizeof(int)); + SIndexTerm* q = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(), + (const char*)&val, sizeof(int)); SArray* result = taosArrayInit(1, sizeof(uint64_t)); indexMultiTermQueryAdd(mq, q, QUERY_GREATER_THAN); @@ -425,8 +437,8 @@ TEST_F(JsonEnv, testWriteJsonTfileAndCache_INT) { int val = 10; SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST); - SIndexTerm* q = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(), - (const char*)&val, sizeof(val)); + SIndexTerm* q = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(), + (const char*)&val, sizeof(val)); SArray* result = taosArrayInit(1, sizeof(uint64_t)); indexMultiTermQueryAdd(mq, q, QUERY_GREATER_EQUAL); @@ -439,8 +451,8 @@ TEST_F(JsonEnv, testWriteJsonTfileAndCache_INT) { int val = 10; SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST); - SIndexTerm* q = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(), - (const char*)&val, sizeof(val)); + SIndexTerm* q = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(), + (const char*)&val, sizeof(val)); SArray* result = taosArrayInit(1, sizeof(uint64_t)); indexMultiTermQueryAdd(mq, q, QUERY_GREATER_THAN); @@ -453,8 +465,8 @@ TEST_F(JsonEnv, testWriteJsonTfileAndCache_INT) { int val = 10; SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST); - SIndexTerm* q = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(), - (const char*)&val, sizeof(val)); + SIndexTerm* q = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(), + (const char*)&val, sizeof(val)); SArray* result = taosArrayInit(1, sizeof(uint64_t)); indexMultiTermQueryAdd(mq, q, QUERY_LESS_EQUAL); @@ -466,8 +478,8 @@ TEST_F(JsonEnv, testWriteJsonTfileAndCache_INT) { std::string colName("other_column"); int val = 100; - SIndexTerm* term = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(), - (const char*)&val, sizeof(val)); + SIndexTerm* term = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(), + (const char*)&val, sizeof(val)); SIndexMultiTerm* terms = indexMultiTermCreate(); indexMultiTermAdd(terms, term); @@ -482,8 +494,8 @@ TEST_F(JsonEnv, testWriteJsonTfileAndCache_INT) { // std::string colVal("10"); SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST); - SIndexTerm* q = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(), - (const char*)&val, sizeof(val)); + SIndexTerm* q = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(), + (const char*)&val, sizeof(val)); SArray* result = taosArrayInit(1, sizeof(uint64_t)); indexMultiTermQueryAdd(mq, q, QUERY_LESS_THAN); @@ -494,8 +506,8 @@ TEST_F(JsonEnv, testWriteJsonTfileAndCache_INT) { { std::string colName("test1"); int val = 15; - SIndexTerm* term = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(), - (const char*)&val, sizeof(val)); + SIndexTerm* term = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(), + (const char*)&val, sizeof(val)); SIndexMultiTerm* terms = indexMultiTermCreate(); indexMultiTermAdd(terms, term); @@ -510,8 +522,8 @@ TEST_F(JsonEnv, testWriteJsonTfileAndCache_INT) { // std::string colVal("10"); SIndexMultiTermQuery* mq = indexMultiTermQueryCreate(MUST); - SIndexTerm* q = indexTermCreate(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(), - (const char*)&val, sizeof(val)); + SIndexTerm* q = indexTermCreateT(1, ADD_VALUE, TSDB_DATA_TYPE_INT, colName.c_str(), colName.size(), + (const char*)&val, sizeof(val)); SArray* result = taosArrayInit(1, sizeof(uint64_t)); indexMultiTermQueryAdd(mq, q, QUERY_GREATER_EQUAL); diff --git a/source/libs/qworker/inc/qwInt.h b/source/libs/qworker/inc/qwInt.h index 3804c114cb..082db6428f 100644 --- a/source/libs/qworker/inc/qwInt.h +++ b/source/libs/qworker/inc/qwInt.h @@ -34,6 +34,7 @@ extern "C" { #define QW_DEFAULT_SCH_TASK_NUMBER 10000 #define QW_DEFAULT_SHORT_RUN_TIMES 2 #define QW_DEFAULT_HEARTBEAT_MSEC 5000 +#define QW_SCH_TIMEOUT_MSEC 180000 enum { QW_PHASE_PRE_QUERY = 1, @@ -137,7 +138,7 @@ typedef struct SQWTaskCtx { } SQWTaskCtx; typedef struct SQWSchStatus { - int32_t lastAccessTs; // timestamp in second + int64_t hbBrokenTs; // timestamp in msecond SRWLatch hbConnLock; SRpcHandleInfo hbConnInfo; SQueryNodeEpId hbEpId; @@ -354,6 +355,8 @@ int32_t qwOpenRef(void); void qwSetHbParam(int64_t refId, SQWHbParam **pParam); int32_t qwUpdateTimeInQueue(SQWorker *mgmt, int64_t ts, 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); int32_t qwDbgValidateStatus(QW_FPARAMS_DEF, int8_t oriStatus, int8_t newStatus, bool *ignore); diff --git a/source/libs/qworker/src/qwUtil.c b/source/libs/qworker/src/qwUtil.c index 3d0204e355..8bfb80f061 100644 --- a/source/libs/qworker/src/qwUtil.c +++ b/source/libs/qworker/src/qwUtil.c @@ -535,3 +535,9 @@ int64_t qwGetTimeInQueue(SQWorker *mgmt, EQueueType type) { return -1; } + +void qwClearExpiredSch(SArray* pExpiredSch) { + +} + + diff --git a/source/libs/qworker/src/qworker.c b/source/libs/qworker/src/qworker.c index 8c0366fa0b..44a8fdf7f4 100644 --- a/source/libs/qworker/src/qworker.c +++ b/source/libs/qworker/src/qworker.c @@ -21,10 +21,12 @@ int32_t qwProcessHbLinkBroken(SQWorker *mgmt, SQWMsg *qwMsg, SSchedulerHbReq *re SSchedulerHbRsp rsp = {0}; 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); + sch->hbBrokenTs = taosGetTimestampMs(); + if (qwMsg->connInfo.handle == sch->hbConnInfo.handle) { tmsgReleaseHandle(&sch->hbConnInfo, TAOS_CONN_SERVER); sch->hbConnInfo.handle = NULL; @@ -794,6 +796,7 @@ void qwProcessHbTimerEvent(void *param, void *tmrId) { SQWSchStatus *sch = NULL; int32_t taskNum = 0; SQWHbInfo *rspList = NULL; + SArray *pExpiredSch = NULL; int32_t code = 0; qwDbgDumpMgmtInfo(mgmt); @@ -809,8 +812,11 @@ void qwProcessHbTimerEvent(void *param, void *tmrId) { } 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); + taosMemoryFree(rspList); + taosArrayDestroy(pExpiredSch); QW_ELOG("calloc %d SQWHbInfo failed", schNum); taosTmrReset(qwProcessHbTimerEvent, QW_DEFAULT_HEARTBEAT_MSEC, param, mgmt->timer, &mgmt->hbTimer); qwRelease(refId); @@ -820,6 +826,7 @@ void qwProcessHbTimerEvent(void *param, void *tmrId) { void *key = NULL; size_t keyLen = 0; int32_t i = 0; + int64_t currentMs = taosGetTimestampMs(); void *pIter = taosHashIterate(mgmt->schHash, NULL); while (pIter) { @@ -827,6 +834,11 @@ void qwProcessHbTimerEvent(void *param, void *tmrId) { if (NULL == sch->hbConnInfo.handle) { uint64_t *sId = taosHashGetKey(pIter, NULL); 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); continue; } @@ -852,7 +864,12 @@ _return: tFreeSSchedulerHbRsp(&rspList[j].rsp); } + if (taosArrayGetSize(pExpiredSch) > 0) { + qwClearExpiredSch(pExpiredSch); + } + taosMemoryFreeClear(rspList); + taosArrayDestroy(pExpiredSch); taosTmrReset(qwProcessHbTimerEvent, QW_DEFAULT_HEARTBEAT_MSEC, param, mgmt->timer, &mgmt->hbTimer); qwRelease(refId); diff --git a/source/libs/scheduler/inc/schedulerInt.h b/source/libs/scheduler/inc/schedulerInt.h index c8fee1e532..d961780ae5 100644 --- a/source/libs/scheduler/inc/schedulerInt.h +++ b/source/libs/scheduler/inc/schedulerInt.h @@ -102,11 +102,11 @@ typedef struct SSchedulerMgmt { uint64_t taskId; // sequential taksId uint64_t sId; // schedulerId SSchedulerCfg cfg; - SRWLatch lock; bool exit; int32_t jobRef; int32_t jobNum; SSchStat stat; + SRWLatch hbLock; SHashObj *hbConnections; } SSchedulerMgmt; @@ -320,6 +320,8 @@ extern SSchedulerMgmt schMgmt; #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 schBuildAndSendMsg(SSchJob *job, SSchTask *task, SQueryNodeAddr *addr, int32_t msgType); SSchJob *schAcquireJob(int64_t refId); diff --git a/source/libs/scheduler/src/schJob.c b/source/libs/scheduler/src/schJob.c index 3e23c395c9..ca90d2fe34 100644 --- a/source/libs/scheduler/src/schJob.c +++ b/source/libs/scheduler/src/schJob.c @@ -126,30 +126,6 @@ _return: 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) { schDeregisterTaskHb(pJob, pTask); @@ -377,15 +353,21 @@ int32_t schDropTaskExecNode(SSchJob *pJob, SSchTask *pTask, void *handle, int32_ 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 + SCH_TASK_DLOG("execIdx %d is not current execIdx %d", execIdx, pTask->execIdx); SCH_RET(TSDB_CODE_SCH_IGNORE_ERROR); } 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) { 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)); nodeInfo->handle = handle; + SCH_TASK_DLOG("handle updated to %p for execIdx %d", handle, execIdx); + return TSDB_CODE_SUCCESS; } @@ -403,7 +387,7 @@ int32_t schUpdateTaskHandle(SSchJob *pJob, SSchTask *pTask, bool dropExecNode, v SCH_SET_TASK_HANDLE(pTask, handle); - schUpdateTaskExecNode(pTask, handle, execIdx); + schUpdateTaskExecNode(pJob, pTask, handle, execIdx); 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_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; } } @@ -1110,6 +1096,7 @@ int32_t schProcessOnTaskSuccess(SSchJob *pJob, SSchTask *pTask) { SCH_UNLOCK(SCH_WRITE, &parent->lock); 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)); } } @@ -1186,7 +1173,7 @@ void schDropTaskOnExecNode(SSchJob *pJob, SSchTask *pTask) { 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); - 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); taosHashClear(pTask->execNodes); schProcessOnTaskFailure(pJob, pTask, TSDB_CODE_SCH_TIMEOUT_ERROR); @@ -1306,9 +1294,10 @@ int32_t schLaunchTaskImpl(SSchJob *pJob, SSchTask *pTask) { int32_t code = 0; atomic_add_fetch_32(&pTask->level->taskLaunchedNum, 1); - pTask->execIdx++; + SCH_TASK_DLOG("start to launch task's %dth exec", pTask->execIdx); + SCH_LOG_TASK_START_TS(pTask); 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); - atomic_sub_fetch_32(&schMgmt.jobNum, 1); - - schCloseJobRef(); + int32_t jobNum = atomic_sub_fetch_32(&schMgmt.jobNum, 1); + if (jobNum == 0) { + schCloseJobRef(); + } } int32_t schExecJobImpl(void *pTrans, SArray *pNodeList, SQueryPlan *pDag, int64_t *job, const char *sql, diff --git a/source/libs/scheduler/src/schRemote.c b/source/libs/scheduler/src/schRemote.c index b336ce8c76..bf51d8d631 100644 --- a/source/libs/scheduler/src/schRemote.c +++ b/source/libs/scheduler/src/schRemote.c @@ -648,31 +648,6 @@ _return: 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) { SSchedulerHbReq req = {0}; int32_t code = 0; @@ -684,17 +659,20 @@ int32_t schBuildAndSendHbMsg(SQueryNodeEpId *nodeEpId, SArray* taskAction) { req.sId = schMgmt.sId; memcpy(&req.epId, nodeEpId, sizeof(SQueryNodeEpId)); + SCH_LOCK(SCH_READ, &schMgmt.hbLock); SSchHbTrans *hb = taosHashGet(schMgmt.hbConnections, nodeEpId, sizeof(SQueryNodeEpId)); 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); - SCH_ERR_RET(code); + return TSDB_CODE_SUCCESS; } SCH_LOCK(SCH_WRITE, &hb->lock); code = schCloneHbRpcCtx(&hb->rpcCtx, &rpcCtx); memcpy(&trans, &hb->trans, sizeof(trans)); SCH_UNLOCK(SCH_WRITE, &hb->lock); + SCH_UNLOCK(SCH_READ, &schMgmt.hbLock); SCH_ERR_RET(code); @@ -764,60 +742,6 @@ _return: 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) { SSchedulerHbRsp rsp = {0}; SSchTaskCallbackParam *pParam = (SSchTaskCallbackParam *)param; @@ -1037,6 +961,7 @@ int32_t schBuildAndSendMsg(SSchJob *pJob, SSchTask *pTask, SQueryNodeAddr *addr, if (NULL == addr) { addr = taosArrayGet(pTask->candidateAddrs, pTask->candidateIdx); isCandidateAddr = true; + SCH_TASK_DLOG("target candidateIdx %d", pTask->candidateIdx); } SEpSet epSet = addr->epSet; diff --git a/source/libs/scheduler/src/schUtil.c b/source/libs/scheduler/src/schUtil.c index 81c95ea976..18398802db 100644 --- a/source/libs/scheduler/src/schUtil.c +++ b/source/libs/scheduler/src/schUtil.c @@ -21,17 +21,189 @@ #include "tref.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) { if (!atomic_load_8((int8_t *)&schMgmt.exit)) { return; } - SCH_LOCK(SCH_WRITE, &schMgmt.lock); - if (atomic_load_32(&schMgmt.jobNum) <= 0 && schMgmt.jobRef >= 0) { + if (schMgmt.jobRef >= 0) { taosCloseRef(schMgmt.jobRef); schMgmt.jobRef = -1; } - SCH_UNLOCK(SCH_WRITE, &schMgmt.lock); } 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); } - diff --git a/source/libs/scheduler/src/scheduler.c b/source/libs/scheduler/src/scheduler.c index 32e00c1b70..b4dc067f4c 100644 --- a/source/libs/scheduler/src/scheduler.c +++ b/source/libs/scheduler/src/scheduler.c @@ -182,6 +182,14 @@ int32_t scheduleCancelJob(int64_t job) { SCH_RET(code); } +void schedulerStopQueryHb(void *pTrans) { + if (NULL == pTrans) { + return; + } + + schCleanClusterHb(pTrans); +} + void schedulerFreeJob(int64_t job) { SSchJob *pJob = schAcquireJob(job); if (NULL == pJob) { @@ -220,6 +228,7 @@ void schedulerDestroy(void) { } } + SCH_LOCK(SCH_WRITE, &schMgmt.hbLock); if (schMgmt.hbConnections) { void *pIter = taosHashIterate(schMgmt.hbConnections, NULL); while (pIter != NULL) { @@ -230,4 +239,5 @@ void schedulerDestroy(void) { taosHashCleanup(schMgmt.hbConnections); schMgmt.hbConnections = NULL; } + SCH_UNLOCK(SCH_WRITE, &schMgmt.hbLock); } diff --git a/source/libs/transport/src/transSvr.c b/source/libs/transport/src/transSvr.c index 608fd00b2c..50f99128b2 100644 --- a/source/libs/transport/src/transSvr.c +++ b/source/libs/transport/src/transSvr.c @@ -180,6 +180,12 @@ static bool addHandleToAcceptloop(void* arg); if (!transQueuePush(&conn->srvMsgs, srvMsg)) { \ 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); \ return; \ } \ @@ -1154,6 +1160,10 @@ int transGetConnInfo(void* thandle, STransHandleInfo* pInfo) { } SExHandle* ex = thandle; 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; pInfo->clientIp = (uint32_t)(addr.sin_addr.s_addr);