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

Merge branch '3.0' of https://github.com/taosdata/TDengine into fix/TD-30837

This commit is contained in:
54liuyao 2024-09-30 17:12:25 +08:00
parent d13819bbf4 e138cfc1f8
commit c9308e7482
30 changed files with 551 additions and 298 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

101
docs/examples/c/CMakeLists.txt Normal file
View File

@ -0,0 +1,101 @@
PROJECT(TDengine)
IF (TD_LINUX)
INCLUDE_DIRECTORIES(. ${TD_SOURCE_DIR}/src/inc ${TD_SOURCE_DIR}/src/client/inc ${TD_SOURCE_DIR}/inc)
AUX_SOURCE_DIRECTORY(. SRC)
add_executable(docs_connect_example "")
add_executable(docs_create_db_demo "")
add_executable(docs_insert_data_demo "")
add_executable(docs_query_data_demo "")
add_executable(docs_with_reqid_demo "")
add_executable(docs_sml_insert_demo "")
add_executable(docs_stmt_insert_demo "")
add_executable(docs_tmq_demo "")
target_sources(docs_connect_example
PRIVATE
"connect_example.c"
)
target_sources(docs_create_db_demo
PRIVATE
"create_db_demo.c"
)
target_sources(docs_insert_data_demo
PRIVATE
"insert_data_demo.c"
)
target_sources(docs_query_data_demo
PRIVATE
"query_data_demo.c"
)
target_sources(docs_with_reqid_demo
PRIVATE
"with_reqid_demo.c"
)
target_sources(docs_sml_insert_demo
PRIVATE
"sml_insert_demo.c"
)
target_sources(docs_stmt_insert_demo
PRIVATE
"stmt_insert_demo.c"
)
target_sources(docs_tmq_demo
PRIVATE
"tmq_demo.c"
)
target_link_libraries(docs_connect_example
taos
)
target_link_libraries(docs_create_db_demo
taos
)
target_link_libraries(docs_insert_data_demo
taos
)
target_link_libraries(docs_query_data_demo
taos
)
target_link_libraries(docs_with_reqid_demo
taos
)
target_link_libraries(docs_sml_insert_demo
taos
)
target_link_libraries(docs_stmt_insert_demo
taos
)
target_link_libraries(docs_tmq_demo
taos
pthread
)
SET_TARGET_PROPERTIES(docs_connect_example PROPERTIES OUTPUT_NAME docs_connect_example)
SET_TARGET_PROPERTIES(docs_create_db_demo PROPERTIES OUTPUT_NAME docs_create_db_demo)
SET_TARGET_PROPERTIES(docs_insert_data_demo PROPERTIES OUTPUT_NAME docs_insert_data_demo)
SET_TARGET_PROPERTIES(docs_query_data_demo PROPERTIES OUTPUT_NAME docs_query_data_demo)
SET_TARGET_PROPERTIES(docs_with_reqid_demo PROPERTIES OUTPUT_NAME docs_with_reqid_demo)
SET_TARGET_PROPERTIES(docs_sml_insert_demo PROPERTIES OUTPUT_NAME docs_sml_insert_demo)
SET_TARGET_PROPERTIES(docs_stmt_insert_demo PROPERTIES OUTPUT_NAME docs_stmt_insert_demo)
SET_TARGET_PROPERTIES(docs_tmq_demo PROPERTIES OUTPUT_NAME docs_tmq_demo)
ENDIF ()
IF (TD_DARWIN)
INCLUDE_DIRECTORIES(. ${TD_SOURCE_DIR}/src/inc ${TD_SOURCE_DIR}/src/client/inc ${TD_SOURCE_DIR}/inc)
AUX_SOURCE_DIRECTORY(. SRC)
ENDIF ()

34
docs/examples/c/Makefile Normal file
View File

@ -0,0 +1,34 @@
# Makefile for building TDengine examples on TD Linux platform
INCLUDE_DIRS =
TARGETS = connect_example \
create_db_demo \
insert_data_demo \
query_data_demo \
with_reqid_demo \
sml_insert_demo \
stmt_insert_demo \
tmq_demo
SOURCES = connect_example.c \
create_db_demo.c \
insert_data_demo.c \
query_data_demo.c \
with_reqid_demo.c \
sml_insert_demo.c \
stmt_insert_demo.c \
tmq_demo.c
LIBS = -ltaos -lpthread
CFLAGS = -g
all: $(TARGETS)
$(TARGETS):
$(CC) $(CFLAGS) -o $@ $(wildcard $(@F).c) $(LIBS)
clean:
rm -f $(TARGETS)

1
include/common/tmsg.h
View File

@ -1685,6 +1685,7 @@ typedef struct {
typedef struct { typedef struct {
int32_t openVnodes; int32_t openVnodes;
int32_t dropVnodes;
int32_t totalVnodes; int32_t totalVnodes;
int32_t masterNum; int32_t masterNum;
int64_t numOfSelectReqs; int64_t numOfSelectReqs;

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

@ -47,7 +47,7 @@ extern "C" {
#define SYNC_HEARTBEAT_SLOW_MS 1500 #define SYNC_HEARTBEAT_SLOW_MS 1500
#define SYNC_HEARTBEAT_REPLY_SLOW_MS 1500 #define SYNC_HEARTBEAT_REPLY_SLOW_MS 1500
#define SYNC_SNAP_RESEND_MS 1000 * 60 #define SYNC_SNAP_RESEND_MS 1000 * 60
#define SYNC_SNAP_TIMEOUT_MS 1000 * 300 #define SYNC_SNAP_TIMEOUT_MS 1000 * 180
#define SYNC_VND_COMMIT_MIN_MS 3000 #define SYNC_VND_COMMIT_MIN_MS 3000

5
include/os/osString.h
View File

@ -51,7 +51,12 @@ typedef enum { M2C = 0, C2M } ConvType;
#define strtod STR_TO_LD_FUNC_TAOS_FORBID #define strtod STR_TO_LD_FUNC_TAOS_FORBID
#define strtold STR_TO_D_FUNC_TAOS_FORBID #define strtold STR_TO_D_FUNC_TAOS_FORBID
#define strtof STR_TO_F_FUNC_TAOS_FORBID #define strtof STR_TO_F_FUNC_TAOS_FORBID
#ifdef strndup
#undef strndup
#endif
#define strndup STR_TO_F_FUNC_TAOS_FORBID #define strndup STR_TO_F_FUNC_TAOS_FORBID
#endif #endif
#define tstrncpy(dst, src, size) \ #define tstrncpy(dst, src, size) \

8
source/client/src/clientRawBlockWrite.c
View File

@ -47,7 +47,7 @@
} \ } \
} while (0) } while (0)
#define LOG_ID_TAG "connId:0x%" PRIx64 ",qid:0x%" PRIx64 #define LOG_ID_TAG "connId:0x%" PRIx64 ",QID:0x%" PRIx64
#define LOG_ID_VALUE *(int64_t*)taos, pRequest->requestId #define LOG_ID_VALUE *(int64_t*)taos, pRequest->requestId
#define TMQ_META_VERSION "1.0" #define TMQ_META_VERSION "1.0"
@ -1211,7 +1211,8 @@ static int32_t taosCreateTable(TAOS* taos, void* meta, int32_t metaLen) {
rebuildTag = true; rebuildTag = true;
} }
} else { } else {
uError("create tb invalid data %s, size:%d index:%d cid:%d", pCreateReq->name, (int)taosArrayGetSize(pTagVals), i, tag->colId); uError("create tb invalid data %s, size:%d index:%d cid:%d", pCreateReq->name,
(int)taosArrayGetSize(pTagVals), i, tag->colId);
} }
} }
} }
@ -1833,7 +1834,8 @@ static int32_t buildCreateTbMap(STaosxRsp* rsp, SHashObj* pHashObj) {
goto end; goto end;
} }
if (taosHashGet(pHashObj, pCreateReq.name, strlen(pCreateReq.name)) == NULL) { if (taosHashGet(pHashObj, pCreateReq.name, strlen(pCreateReq.name)) == NULL) {
RAW_RETURN_CHECK(taosHashPut(pHashObj, pCreateReq.name, strlen(pCreateReq.name), &pCreateReq, sizeof(SVCreateTbReq))); RAW_RETURN_CHECK(
taosHashPut(pHashObj, pCreateReq.name, strlen(pCreateReq.name), &pCreateReq, sizeof(SVCreateTbReq)));
} else { } else {
tDestroySVCreateTbReq(&pCreateReq, TSDB_MSG_FLG_DECODE); tDestroySVCreateTbReq(&pCreateReq, TSDB_MSG_FLG_DECODE);
pCreateReq = (SVCreateTbReq){0}; pCreateReq = (SVCreateTbReq){0};

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

@ -77,6 +77,7 @@ typedef struct {
typedef struct { typedef struct {
int32_t vnodeNum; int32_t vnodeNum;
int32_t opened; int32_t opened;
int32_t dropped;
int32_t failed; int32_t failed;
bool updateVnodesList; bool updateVnodesList;
int32_t threadIndex; int32_t threadIndex;

9
source/dnode/mgmt/mgmt_vnode/src/vmInt.c
View File

@ -311,6 +311,8 @@ static void *vmOpenVnodeInThread(void *param) {
snprintf(path, TSDB_FILENAME_LEN, "vnode%svnode%d", TD_DIRSEP, pCfg->vgId); snprintf(path, TSDB_FILENAME_LEN, "vnode%svnode%d", TD_DIRSEP, pCfg->vgId);
vnodeDestroy(pCfg->vgId, path, pMgmt->pTfs, 0); vnodeDestroy(pCfg->vgId, path, pMgmt->pTfs, 0);
pThread->updateVnodesList = true; pThread->updateVnodesList = true;
pThread->dropped++;
(void)atomic_add_fetch_32(&pMgmt->state.dropVnodes, 1);
continue; continue;
} }
@ -352,8 +354,8 @@ static void *vmOpenVnodeInThread(void *param) {
(void)atomic_add_fetch_32(&pMgmt->state.openVnodes, 1); (void)atomic_add_fetch_32(&pMgmt->state.openVnodes, 1);
} }
dInfo("thread:%d, numOfVnodes:%d, opened:%d failed:%d", pThread->threadIndex, pThread->vnodeNum, pThread->opened, dInfo("thread:%d, numOfVnodes:%d, opened:%d dropped:%d failed:%d", pThread->threadIndex, pThread->vnodeNum,
pThread->failed); pThread->opened, pThread->dropped, pThread->failed);
return NULL; return NULL;
} }
@ -427,7 +429,7 @@ static int32_t vmOpenVnodes(SVnodeMgmt *pMgmt) {
taosMemoryFree(threads); taosMemoryFree(threads);
taosMemoryFree(pCfgs); taosMemoryFree(pCfgs);
if (pMgmt->state.openVnodes != pMgmt->state.totalVnodes) { if ((pMgmt->state.openVnodes + pMgmt->state.dropVnodes) != pMgmt->state.totalVnodes) {
dError("there are total vnodes:%d, opened:%d", pMgmt->state.totalVnodes, pMgmt->state.openVnodes); dError("there are total vnodes:%d, opened:%d", pMgmt->state.totalVnodes, pMgmt->state.openVnodes);
terrno = TSDB_CODE_VND_INIT_FAILED; terrno = TSDB_CODE_VND_INIT_FAILED;
return -1; return -1;
@ -774,6 +776,7 @@ static int32_t vmStartVnodes(SVnodeMgmt *pMgmt) {
} }
pMgmt->state.openVnodes = 0; pMgmt->state.openVnodes = 0;
pMgmt->state.dropVnodes = 0;
dInfo("restore %d vnodes with %d threads", numOfVnodes, threadNum); dInfo("restore %d vnodes with %d threads", numOfVnodes, threadNum);
for (int32_t t = 0; t < threadNum; ++t) { for (int32_t t = 0; t < threadNum; ++t) {

1
source/dnode/mnode/impl/src/mndTrans.c
View File

@ -1637,6 +1637,7 @@ static bool mndTransPerformRedoActionStage(SMnode *pMnode, STrans *pTrans, bool
pTrans->code = code; pTrans->code = code;
bool continueExec = true; bool continueExec = true;
if (code != 0 && code != TSDB_CODE_MND_TRANS_CTX_SWITCH) { if (code != 0 && code != TSDB_CODE_MND_TRANS_CTX_SWITCH) {
taosMsleep(100);
continueExec = true; continueExec = true;
} else { } else {
continueExec = false; continueExec = false;

1
source/dnode/vnode/src/meta/metaQuery.c
View File

@ -1523,6 +1523,7 @@ int32_t metaGetTableTags(void *pVnode, uint64_t suid, SArray *pUidTagInfo) {
} }
memcpy(info.pTagVal, pCur->pVal, pCur->vLen); memcpy(info.pTagVal, pCur->pVal, pCur->vLen);
if (taosArrayPush(pUidTagInfo, &info) == NULL) { if (taosArrayPush(pUidTagInfo, &info) == NULL) {
taosMemoryFreeClear(info.pTagVal);
metaCloseCtbCursor(pCur); metaCloseCtbCursor(pCur);
taosHashCleanup(pSepecifiedUidMap); taosHashCleanup(pSepecifiedUidMap);
return terrno; return terrno;

2
source/dnode/vnode/src/tqCommon/tqCommon.c
View File

@ -516,7 +516,7 @@ int32_t tqStreamTaskProcessCheckRsp(SStreamMeta* pMeta, SRpcMsg* pMsg, bool isLe
} }
tDecoderClear(&decoder); tDecoderClear(&decoder);
tqDebug("tq task:0x%x (vgId:%d) recv check rsp(qid:0x%" PRIx64 ") from 0x%x (vgId:%d) status %d", rsp.upstreamTaskId, tqDebug("tq task:0x%x (vgId:%d) recv check rsp(QID:0x%" PRIx64 ") from 0x%x (vgId:%d) status %d", rsp.upstreamTaskId,
rsp.upstreamNodeId, rsp.reqId, rsp.downstreamTaskId, rsp.downstreamNodeId, rsp.status); rsp.upstreamNodeId, rsp.reqId, rsp.downstreamTaskId, rsp.downstreamNodeId, rsp.status);
if (!isLeader) { if (!isLeader) {

161
source/dnode/vnode/src/tsdb/tsdbCache.c
View File

@ -22,6 +22,12 @@
#define ROCKS_BATCH_SIZE (4096) #define ROCKS_BATCH_SIZE (4096)
void tsdbLRUCacheRelease(SLRUCache *cache, LRUHandle *handle, bool eraseIfLastRef) {
if (!taosLRUCacheRelease(cache, handle, eraseIfLastRef)) {
tsdbTrace(" release lru cache failed");
}
}
static int32_t tsdbOpenBCache(STsdb *pTsdb) { static int32_t tsdbOpenBCache(STsdb *pTsdb) {
int32_t code = 0, lino = 0; int32_t code = 0, lino = 0;
int32_t szPage = pTsdb->pVnode->config.tsdbPageSize; int32_t szPage = pTsdb->pVnode->config.tsdbPageSize;
@ -579,7 +585,7 @@ static void tsdbCacheDeleter(const void *key, size_t klen, void *value, void *ud
if (pLastCol->dirty) { if (pLastCol->dirty) {
if (tsdbCacheFlushDirty(key, klen, pLastCol, ud) != 0) { if (tsdbCacheFlushDirty(key, klen, pLastCol, ud) != 0) {
STsdb *pTsdb = (STsdb *)ud; STsdb *pTsdb = (STsdb *)ud;
tsdbError("tsdb/cache: vgId:%d, flush cache %s failed at line %d.", TD_VID(pTsdb->pVnode), __func__, __LINE__); tsdbTrace("tsdb/cache: vgId:%d, flush cache %s failed at line %d.", TD_VID(pTsdb->pVnode), __func__, __LINE__);
} }
} }
@ -713,9 +719,13 @@ static int32_t tsdbCacheDropTableColumn(STsdb *pTsdb, int64_t uid, int16_t cid,
{ {
SLastCol *pLastCol = NULL; SLastCol *pLastCol = NULL;
code = tsdbCacheDeserialize(values_list[0], values_list_sizes[0], &pLastCol); code = tsdbCacheDeserialize(values_list[0], values_list_sizes[0], &pLastCol);
if (code != TSDB_CODE_SUCCESS) { if (code == TSDB_CODE_INVALID_PARA) {
tsdbWarn("vgId:%d, %s deserialize failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__, tsdbTrace("vgId:%d, %s deserialize failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code)); tstrerror(code));
} else if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s deserialize failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code));
goto _exit;
} }
if (NULL != pLastCol) { if (NULL != pLastCol) {
rocksdb_writebatch_delete(wb, keys_list[0], klen); rocksdb_writebatch_delete(wb, keys_list[0], klen);
@ -724,9 +734,13 @@ static int32_t tsdbCacheDropTableColumn(STsdb *pTsdb, int64_t uid, int16_t cid,
pLastCol = NULL; pLastCol = NULL;
code = tsdbCacheDeserialize(values_list[1], values_list_sizes[1], &pLastCol); code = tsdbCacheDeserialize(values_list[1], values_list_sizes[1], &pLastCol);
if (code != TSDB_CODE_SUCCESS) { if (code == TSDB_CODE_INVALID_PARA) {
tsdbWarn("vgId:%d, %s deserialize failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__, tsdbTrace("vgId:%d, %s deserialize failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code)); tstrerror(code));
} else if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s deserialize failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code));
goto _exit;
} }
if (NULL != pLastCol) { if (NULL != pLastCol) {
rocksdb_writebatch_delete(wb, keys_list[1], klen); rocksdb_writebatch_delete(wb, keys_list[1], klen);
@ -739,9 +753,7 @@ static int32_t tsdbCacheDropTableColumn(STsdb *pTsdb, int64_t uid, int16_t cid,
for (int i = 0; i < 2; i++) { for (int i = 0; i < 2; i++) {
LRUHandle *h = taosLRUCacheLookup(pTsdb->lruCache, keys_list[i], klen); LRUHandle *h = taosLRUCacheLookup(pTsdb->lruCache, keys_list[i], klen);
if (h) { if (h) {
if (taosLRUCacheRelease(pTsdb->lruCache, h, true)) { tsdbLRUCacheRelease(pTsdb->lruCache, h, true);
tsdbInfo("vgId:%d, %s release lru cache failed at line %d.", TD_VID(pTsdb->pVnode), __func__, __LINE__);
}
taosLRUCacheErase(pTsdb->lruCache, keys_list[i], klen); taosLRUCacheErase(pTsdb->lruCache, keys_list[i], klen);
} }
} }
@ -770,17 +782,13 @@ int32_t tsdbCacheNewTable(STsdb *pTsdb, tb_uid_t uid, tb_uid_t suid, SSchemaWrap
code = tsdbCacheNewTableColumn(pTsdb, uid, cid, col_type, LFLAG_LAST_ROW); code = tsdbCacheNewTableColumn(pTsdb, uid, cid, col_type, LFLAG_LAST_ROW);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s new table column failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__, tsdbTrace("vgId:%d, %s new table column failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code)); tstrerror(code));
(void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code);
} }
code = tsdbCacheNewTableColumn(pTsdb, uid, cid, col_type, LFLAG_LAST); code = tsdbCacheNewTableColumn(pTsdb, uid, cid, col_type, LFLAG_LAST);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s new table column failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__, tsdbTrace("vgId:%d, %s new table column failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code)); tstrerror(code));
(void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code);
} }
} }
} else { } else {
@ -798,17 +806,13 @@ int32_t tsdbCacheNewTable(STsdb *pTsdb, tb_uid_t uid, tb_uid_t suid, SSchemaWrap
code = tsdbCacheNewTableColumn(pTsdb, uid, cid, col_type, LFLAG_LAST_ROW); code = tsdbCacheNewTableColumn(pTsdb, uid, cid, col_type, LFLAG_LAST_ROW);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s new table column failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__, tsdbTrace("vgId:%d, %s new table column failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code)); tstrerror(code));
(void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code);
} }
code = tsdbCacheNewTableColumn(pTsdb, uid, cid, col_type, LFLAG_LAST); code = tsdbCacheNewTableColumn(pTsdb, uid, cid, col_type, LFLAG_LAST);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s new table column failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__, tsdbTrace("vgId:%d, %s new table column failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code)); tstrerror(code));
(void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code);
} }
} }
@ -827,10 +831,8 @@ int32_t tsdbCacheDropTable(STsdb *pTsdb, tb_uid_t uid, tb_uid_t suid, SSchemaWra
code = tsdbCacheCommitNoLock(pTsdb); code = tsdbCacheCommitNoLock(pTsdb);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s commit with no lock failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__, tsdbTrace("vgId:%d, %s commit with no lock failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code)); tstrerror(code));
(void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code);
} }
if (pSchemaRow != NULL) { if (pSchemaRow != NULL) {
@ -845,10 +847,8 @@ int32_t tsdbCacheDropTable(STsdb *pTsdb, tb_uid_t uid, tb_uid_t suid, SSchemaWra
code = tsdbCacheDropTableColumn(pTsdb, uid, cid, hasPrimayKey); code = tsdbCacheDropTableColumn(pTsdb, uid, cid, hasPrimayKey);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s drop table column failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__, tsdbTrace("vgId:%d, %s drop table column failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code)); tstrerror(code));
(void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code);
} }
} }
} else { } else {
@ -871,10 +871,8 @@ int32_t tsdbCacheDropTable(STsdb *pTsdb, tb_uid_t uid, tb_uid_t suid, SSchemaWra
code = tsdbCacheDropTableColumn(pTsdb, uid, cid, hasPrimayKey); code = tsdbCacheDropTableColumn(pTsdb, uid, cid, hasPrimayKey);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s drop table column failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__, tsdbTrace("vgId:%d, %s drop table column failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code)); tstrerror(code));
(void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code);
} }
} }
@ -895,10 +893,8 @@ int32_t tsdbCacheDropSubTables(STsdb *pTsdb, SArray *uids, tb_uid_t suid) {
code = tsdbCacheCommitNoLock(pTsdb); code = tsdbCacheCommitNoLock(pTsdb);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s commit with no lock failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__, tsdbTrace("vgId:%d, %s commit with no lock failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code)); tstrerror(code));
(void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code);
} }
STSchema *pTSchema = NULL; STSchema *pTSchema = NULL;
@ -924,11 +920,8 @@ int32_t tsdbCacheDropSubTables(STsdb *pTsdb, SArray *uids, tb_uid_t suid) {
code = tsdbCacheDropTableColumn(pTsdb, uid, cid, hasPrimayKey); code = tsdbCacheDropTableColumn(pTsdb, uid, cid, hasPrimayKey);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s drop table column failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__, tsdbTrace("vgId:%d, %s drop table column failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code)); tstrerror(code));
(void)taosThreadMutexUnlock(&pTsdb->lruMutex);
taosMemoryFree(pTSchema);
TAOS_RETURN(code);
} }
} }
} }
@ -949,17 +942,13 @@ int32_t tsdbCacheNewNTableColumn(STsdb *pTsdb, int64_t uid, int16_t cid, int8_t
code = tsdbCacheNewTableColumn(pTsdb, uid, cid, col_type, 0); code = tsdbCacheNewTableColumn(pTsdb, uid, cid, col_type, 0);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s new table column failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__, tsdbTrace("vgId:%d, %s new table column failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code)); tstrerror(code));
(void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code);
} }
code = tsdbCacheNewTableColumn(pTsdb, uid, cid, col_type, 1); code = tsdbCacheNewTableColumn(pTsdb, uid, cid, col_type, 1);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s new table column failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__, tsdbTrace("vgId:%d, %s new table column failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code)); tstrerror(code));
(void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code);
} }
// rocksMayWrite(pTsdb, true, false, false); // rocksMayWrite(pTsdb, true, false, false);
(void)taosThreadMutexUnlock(&pTsdb->lruMutex); (void)taosThreadMutexUnlock(&pTsdb->lruMutex);
@ -974,18 +963,14 @@ int32_t tsdbCacheDropNTableColumn(STsdb *pTsdb, int64_t uid, int16_t cid, bool h
code = tsdbCacheCommitNoLock(pTsdb); code = tsdbCacheCommitNoLock(pTsdb);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s commit with no lock failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__, tsdbTrace("vgId:%d, %s commit with no lock failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code)); tstrerror(code));
(void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code);
} }
code = tsdbCacheDropTableColumn(pTsdb, uid, cid, hasPrimayKey); code = tsdbCacheDropTableColumn(pTsdb, uid, cid, hasPrimayKey);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s drop table column failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__, tsdbTrace("vgId:%d, %s drop table column failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code)); tstrerror(code));
(void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code);
} }
rocksMayWrite(pTsdb, false); rocksMayWrite(pTsdb, false);
@ -1005,17 +990,13 @@ int32_t tsdbCacheNewSTableColumn(STsdb *pTsdb, SArray *uids, int16_t cid, int8_t
code = tsdbCacheNewTableColumn(pTsdb, uid, cid, col_type, 0); code = tsdbCacheNewTableColumn(pTsdb, uid, cid, col_type, 0);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s new table column failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__, tsdbTrace("vgId:%d, %s new table column failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code)); tstrerror(code));
(void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code);
} }
code = tsdbCacheNewTableColumn(pTsdb, uid, cid, col_type, 1); code = tsdbCacheNewTableColumn(pTsdb, uid, cid, col_type, 1);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s new table column failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__, tsdbTrace("vgId:%d, %s new table column failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code)); tstrerror(code));
(void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code);
} }
} }
@ -1031,10 +1012,8 @@ int32_t tsdbCacheDropSTableColumn(STsdb *pTsdb, SArray *uids, int16_t cid, bool
code = tsdbCacheCommitNoLock(pTsdb); code = tsdbCacheCommitNoLock(pTsdb);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s commit with no lock failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__, tsdbTrace("vgId:%d, %s commit with no lock failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code)); tstrerror(code));
(void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code);
} }
for (int i = 0; i < TARRAY_SIZE(uids); ++i) { for (int i = 0; i < TARRAY_SIZE(uids); ++i) {
@ -1042,10 +1021,8 @@ int32_t tsdbCacheDropSTableColumn(STsdb *pTsdb, SArray *uids, int16_t cid, bool
code = tsdbCacheDropTableColumn(pTsdb, uid, cid, hasPrimayKey); code = tsdbCacheDropTableColumn(pTsdb, uid, cid, hasPrimayKey);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s drop table column failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__, tsdbTrace("vgId:%d, %s drop table column failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code)); tstrerror(code));
(void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code);
} }
} }
@ -1174,9 +1151,7 @@ static int32_t tsdbCacheUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, SArray
} }
} }
if (!taosLRUCacheRelease(pCache, h, false)) { tsdbLRUCacheRelease(pCache, h, false);
tsdbInfo("vgId:%d, %s release lru cache failed at line %d", TD_VID(pTsdb->pVnode), __func__, __LINE__);
}
TAOS_CHECK_EXIT(code); TAOS_CHECK_EXIT(code);
} else { } else {
if (!remainCols) { if (!remainCols) {
@ -1237,9 +1212,13 @@ static int32_t tsdbCacheUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, SArray
SLastCol *pLastCol = NULL; SLastCol *pLastCol = NULL;
code = tsdbCacheDeserialize(values_list[i], values_list_sizes[i], &pLastCol); code = tsdbCacheDeserialize(values_list[i], values_list_sizes[i], &pLastCol);
if (code != TSDB_CODE_SUCCESS) { if (code == TSDB_CODE_INVALID_PARA) {
tsdbWarn("vgId:%d, %s deserialize failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__, tsdbTrace("vgId:%d, %s deserialize failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code)); tstrerror(code));
} else if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s deserialize failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code));
goto _exit;
} }
/* /*
if (code) { if (code) {
@ -1391,9 +1370,8 @@ int32_t tsdbCacheRowFormatUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, int6
} }
code = tSimpleHashIterateRemove(iColHash, &iCol, sizeof(iCol), &pIte, &iter); code = tSimpleHashIterateRemove(iColHash, &iCol, sizeof(iCol), &pIte, &iter);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s tSimpleHashIterateRemove failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, tsdbTrace("vgId:%d, %s tSimpleHashIterateRemove failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__,
__LINE__, tstrerror(code)); __LINE__, tstrerror(code));
TAOS_CHECK_GOTO(code, &lino, _exit);
} }
} }
} }
@ -1402,9 +1380,8 @@ int32_t tsdbCacheRowFormatUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, int6
// 3. do update // 3. do update
code = tsdbCacheUpdate(pTsdb, suid, uid, ctxArray); code = tsdbCacheUpdate(pTsdb, suid, uid, ctxArray);
if (code < TSDB_CODE_SUCCESS) { if (code < TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s tsdbCacheUpdate failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__, tsdbTrace("vgId:%d, %s tsdbCacheUpdate failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code)); tstrerror(code));
TAOS_CHECK_GOTO(code, &lino, _exit);
} }
_exit: _exit:
@ -1491,9 +1468,8 @@ int32_t tsdbCacheColFormatUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, SBlo
// 3. do update // 3. do update
code = tsdbCacheUpdate(pTsdb, suid, uid, ctxArray); code = tsdbCacheUpdate(pTsdb, suid, uid, ctxArray);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s tsdbCacheUpdate failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__, tsdbTrace("vgId:%d, %s tsdbCacheUpdate failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code)); tstrerror(code));
TAOS_CHECK_GOTO(code, &lino, _exit);
} }
_exit: _exit:
@ -1709,9 +1685,13 @@ static int32_t tsdbCacheLoadFromRocks(STsdb *pTsdb, tb_uid_t uid, SArray *pLastA
} }
code = tsdbCacheDeserialize(values_list[i], values_list_sizes[i], &pLastCol); code = tsdbCacheDeserialize(values_list[i], values_list_sizes[i], &pLastCol);
if (code != TSDB_CODE_SUCCESS) { if (code == TSDB_CODE_INVALID_PARA) {
tsdbWarn("vgId:%d, %s deserialize failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__, tsdbTrace("vgId:%d, %s deserialize failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code)); tstrerror(code));
} else if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s deserialize failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code));
goto _exit;
} }
SLastCol *pToFree = pLastCol; SLastCol *pToFree = pLastCol;
SIdxKey *idxKey = &((SIdxKey *)TARRAY_DATA(remainCols))[j]; SIdxKey *idxKey = &((SIdxKey *)TARRAY_DATA(remainCols))[j];
@ -1827,11 +1807,7 @@ int32_t tsdbCacheGetBatch(STsdb *pTsdb, tb_uid_t uid, SArray *pLastArray, SCache
} }
if (h) { if (h) {
code = taosLRUCacheRelease(pCache, h, false); tsdbLRUCacheRelease(pCache, h, false);
if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s release lru cache failed at line %d.", TD_VID(pTsdb->pVnode), __func__, __LINE__);
goto _exit;
}
} }
} }
@ -1858,13 +1834,8 @@ int32_t tsdbCacheGetBatch(STsdb *pTsdb, tb_uid_t uid, SArray *pLastArray, SCache
// no cache or cache is invalid // no cache or cache is invalid
++i; ++i;
} }
if (h) { if (h) {
code = taosLRUCacheRelease(pCache, h, false); tsdbLRUCacheRelease(pCache, h, false);
if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s release lru cache failed at line %d.", TD_VID(pTsdb->pVnode), __func__, __LINE__);
goto _exit;
}
} }
} }
@ -1900,10 +1871,8 @@ int32_t tsdbCacheDel(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, TSKEY sKey, TSKE
code = tsdbCacheCommit(pTsdb); code = tsdbCacheCommit(pTsdb);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s commit failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__, tsdbTrace("vgId:%d, %s commit failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code)); tstrerror(code));
(void)taosThreadMutexUnlock(&pTsdb->lruMutex);
TAOS_RETURN(code);
} }
(void)taosThreadMutexLock(&pTsdb->lruMutex); (void)taosThreadMutexLock(&pTsdb->lruMutex);
@ -1922,9 +1891,7 @@ int32_t tsdbCacheDel(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, TSKEY sKey, TSKE
.cacheStatus = TSDB_LAST_CACHE_NO_CACHE}; .cacheStatus = TSDB_LAST_CACHE_NO_CACHE};
code = tsdbCachePutToLRU(pTsdb, &lastKey, &noneCol, 1); code = tsdbCachePutToLRU(pTsdb, &lastKey, &noneCol, 1);
} }
if (taosLRUCacheRelease(pTsdb->lruCache, h, false) != TSDB_CODE_SUCCESS) { tsdbLRUCacheRelease(pTsdb->lruCache, h, false);
tsdbError("vgId:%d, %s release lru cache failed at line %d.", TD_VID(pTsdb->pVnode), __func__, __LINE__);
}
TAOS_CHECK_EXIT(code); TAOS_CHECK_EXIT(code);
} else { } else {
if (!remainCols) { if (!remainCols) {
@ -1976,9 +1943,13 @@ int32_t tsdbCacheDel(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, TSKEY sKey, TSKE
for (int i = 0; i < numKeys; ++i) { for (int i = 0; i < numKeys; ++i) {
SLastCol *pLastCol = NULL; SLastCol *pLastCol = NULL;
code = tsdbCacheDeserialize(values_list[i], values_list_sizes[i], &pLastCol); code = tsdbCacheDeserialize(values_list[i], values_list_sizes[i], &pLastCol);
if (code != TSDB_CODE_SUCCESS) { if (code == TSDB_CODE_INVALID_PARA) {
tsdbWarn("vgId:%d, %s deserialize failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__, tsdbTrace("vgId:%d, %s deserialize failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code)); tstrerror(code));
} else if (code != TSDB_CODE_SUCCESS) {
tsdbError("vgId:%d, %s deserialize failed at line %d since %s", TD_VID(pTsdb->pVnode), __func__, __LINE__,
tstrerror(code));
goto _exit;
} }
SIdxKey *idxKey = taosArrayGet(remainCols, i); SIdxKey *idxKey = taosArrayGet(remainCols, i);
SLastKey *pLastKey = &idxKey->key; SLastKey *pLastKey = &idxKey->key;
@ -3483,11 +3454,7 @@ _err:
TAOS_RETURN(code); TAOS_RETURN(code);
} }
void tsdbCacheRelease(SLRUCache *pCache, LRUHandle *h) { void tsdbCacheRelease(SLRUCache *pCache, LRUHandle *h) { tsdbLRUCacheRelease(pCache, h, false); }
if (taosLRUCacheRelease(pCache, h, false)) {
tsdbError("%s release lru cache failed at line %d.", __func__, __LINE__);
}
}
void tsdbCacheSetCapacity(SVnode *pVnode, size_t capacity) { void tsdbCacheSetCapacity(SVnode *pVnode, size_t capacity) {
taosLRUCacheSetCapacity(pVnode->pTsdb->lruCache, capacity); taosLRUCacheSetCapacity(pVnode->pTsdb->lruCache, capacity);

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

@ -704,7 +704,7 @@ static int32_t doLoadBlockIndex(STsdbReader* pReader, SDataFileReader* pFileRead
pReader->cost.headFileLoadTime += (et1 - st) / 1000.0; pReader->cost.headFileLoadTime += (et1 - st) / 1000.0;
_end: //_end:
// tsdbBICacheRelease(pFileReader->pTsdb->biCache, handle); // tsdbBICacheRelease(pFileReader->pTsdb->biCache, handle);
return code; return code;
} }
@ -1872,7 +1872,7 @@ static void doPinSttBlock(SSttBlockReader* pSttBlockReader) { tMergeTreePinSttBl
static void doUnpinSttBlock(SSttBlockReader* pSttBlockReader) { tMergeTreeUnpinSttBlock(&pSttBlockReader->mergeTree); } static void doUnpinSttBlock(SSttBlockReader* pSttBlockReader) { tMergeTreeUnpinSttBlock(&pSttBlockReader->mergeTree); }
static bool tryCopyDistinctRowFromSttBlock(TSDBROW* fRow, SSttBlockReader* pSttBlockReader, static int32_t tryCopyDistinctRowFromSttBlock(TSDBROW* fRow, SSttBlockReader* pSttBlockReader,
STableBlockScanInfo* pScanInfo, SRowKey* pSttKey, STsdbReader* pReader, STableBlockScanInfo* pScanInfo, SRowKey* pSttKey, STsdbReader* pReader,
bool* copied) { bool* copied) {
int32_t code = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;

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

@ -72,7 +72,7 @@ int32_t ctgInitGetTbMetaTask(SCtgJob* pJob, int32_t taskIdx, void* param) {
CTG_ERR_RET(terrno); CTG_ERR_RET(terrno);
} }
qDebug("qid:0x%" PRIx64 " the %dth task type %s initialized, tbName:%s", pJob->queryId, taskIdx, qDebug("QID:0x%" PRIx64 " the %dth task type %s initialized, tbName:%s", pJob->queryId, taskIdx,
ctgTaskTypeStr(task.type), name->tname); ctgTaskTypeStr(task.type), name->tname);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -94,7 +94,7 @@ int32_t ctgInitGetTbMetasTask(SCtgJob* pJob, int32_t taskIdx, void* param) {
ctx->pNames = param; ctx->pNames = param;
ctx->pResList = taosArrayInit(pJob->tbMetaNum, sizeof(SMetaRes)); ctx->pResList = taosArrayInit(pJob->tbMetaNum, sizeof(SMetaRes));
if (NULL == ctx->pResList) { if (NULL == ctx->pResList) {
qError("qid:0x%" PRIx64 " taosArrayInit %d SMetaRes %d failed", pJob->queryId, pJob->tbMetaNum, qError("QID:0x%" PRIx64 " taosArrayInit %d SMetaRes %d failed", pJob->queryId, pJob->tbMetaNum,
(int32_t)sizeof(SMetaRes)); (int32_t)sizeof(SMetaRes));
ctgFreeTask(&task, true); ctgFreeTask(&task, true);
CTG_ERR_RET(terrno); CTG_ERR_RET(terrno);
@ -105,7 +105,7 @@ int32_t ctgInitGetTbMetasTask(SCtgJob* pJob, int32_t taskIdx, void* param) {
CTG_ERR_RET(terrno); CTG_ERR_RET(terrno);
} }
qDebug("qid:0x%" PRIx64 " the %dth task type %s initialized, dbNum:%lu, tbNum:%d", pJob->queryId, taskIdx, qDebug("QID:0x%" PRIx64 " the %dth task type %s initialized, dbNum:%lu, tbNum:%d", pJob->queryId, taskIdx,
ctgTaskTypeStr(task.type), taosArrayGetSize(ctx->pNames), pJob->tbMetaNum); ctgTaskTypeStr(task.type), taosArrayGetSize(ctx->pNames), pJob->tbMetaNum);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -133,7 +133,7 @@ int32_t ctgInitGetDbVgTask(SCtgJob* pJob, int32_t taskIdx, void* param) {
CTG_ERR_RET(terrno); CTG_ERR_RET(terrno);
} }
qDebug("qid:0x%" PRIx64 " the %dth task type %s initialized, dbFName:%s", pJob->queryId, taskIdx, qDebug("QID:0x%" PRIx64 " the %dth task type %s initialized, dbFName:%s", pJob->queryId, taskIdx,
ctgTaskTypeStr(task.type), dbFName); ctgTaskTypeStr(task.type), dbFName);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -161,7 +161,7 @@ int32_t ctgInitGetDbCfgTask(SCtgJob* pJob, int32_t taskIdx, void* param) {
CTG_ERR_RET(terrno); CTG_ERR_RET(terrno);
} }
qDebug("qid:0x%" PRIx64 " the %dth task type %s initialized, dbFName:%s", pJob->queryId, taskIdx, qDebug("QID:0x%" PRIx64 " the %dth task type %s initialized, dbFName:%s", pJob->queryId, taskIdx,
ctgTaskTypeStr(task.type), dbFName); ctgTaskTypeStr(task.type), dbFName);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -189,7 +189,7 @@ int32_t ctgInitGetDbInfoTask(SCtgJob* pJob, int32_t taskIdx, void* param) {
CTG_ERR_RET(terrno); CTG_ERR_RET(terrno);
} }
qDebug("qid:0x%" PRIx64 " the %dth task type %s initialized, dbFName:%s", pJob->queryId, taskIdx, qDebug("QID:0x%" PRIx64 " the %dth task type %s initialized, dbFName:%s", pJob->queryId, taskIdx,
ctgTaskTypeStr(task.type), dbFName); ctgTaskTypeStr(task.type), dbFName);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -223,7 +223,7 @@ int32_t ctgInitGetTbHashTask(SCtgJob* pJob, int32_t taskIdx, void* param) {
CTG_ERR_RET(terrno); CTG_ERR_RET(terrno);
} }
qDebug("qid:0x%" PRIx64 " the %dth task type %s initialized, tableName:%s", pJob->queryId, taskIdx, qDebug("QID:0x%" PRIx64 " the %dth task type %s initialized, tableName:%s", pJob->queryId, taskIdx,
ctgTaskTypeStr(task.type), name->tname); ctgTaskTypeStr(task.type), name->tname);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -245,7 +245,7 @@ int32_t ctgInitGetTbHashsTask(SCtgJob* pJob, int32_t taskIdx, void* param) {
ctx->pNames = param; ctx->pNames = param;
ctx->pResList = taosArrayInit(pJob->tbHashNum, sizeof(SMetaRes)); ctx->pResList = taosArrayInit(pJob->tbHashNum, sizeof(SMetaRes));
if (NULL == ctx->pResList) { if (NULL == ctx->pResList) {
qError("qid:0x%" PRIx64 " taosArrayInit %d SMetaRes %d failed", pJob->queryId, pJob->tbHashNum, qError("QID:0x%" PRIx64 " taosArrayInit %d SMetaRes %d failed", pJob->queryId, pJob->tbHashNum,
(int32_t)sizeof(SMetaRes)); (int32_t)sizeof(SMetaRes));
ctgFreeTask(&task, true); ctgFreeTask(&task, true);
CTG_ERR_RET(terrno); CTG_ERR_RET(terrno);
@ -256,7 +256,7 @@ int32_t ctgInitGetTbHashsTask(SCtgJob* pJob, int32_t taskIdx, void* param) {
CTG_ERR_RET(terrno); CTG_ERR_RET(terrno);
} }
qDebug("qid:0x%" PRIx64 " the %dth task type %s initialized, dbNum:%lu, tbNum:%d", pJob->queryId, taskIdx, qDebug("QID:0x%" PRIx64 " the %dth task type %s initialized, dbNum:%lu, tbNum:%d", pJob->queryId, taskIdx,
ctgTaskTypeStr(task.type), taosArrayGetSize(ctx->pNames), pJob->tbHashNum); ctgTaskTypeStr(task.type), taosArrayGetSize(ctx->pNames), pJob->tbHashNum);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -275,7 +275,7 @@ int32_t ctgInitGetQnodeTask(SCtgJob* pJob, int32_t taskIdx, void* param) {
CTG_ERR_RET(terrno); CTG_ERR_RET(terrno);
} }
qDebug("qid:0x%" PRIx64 " the %dth task type %s initialized", pJob->queryId, taskIdx, ctgTaskTypeStr(task.type)); qDebug("QID:0x%" PRIx64 " the %dth task type %s initialized", pJob->queryId, taskIdx, ctgTaskTypeStr(task.type));
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -293,7 +293,7 @@ int32_t ctgInitGetDnodeTask(SCtgJob* pJob, int32_t taskIdx, void* param) {
CTG_ERR_RET(terrno); CTG_ERR_RET(terrno);
} }
qDebug("qid:0x%" PRIx64 " the %dth task type %s initialized", pJob->queryId, taskIdx, ctgTaskTypeStr(task.type)); qDebug("QID:0x%" PRIx64 " the %dth task type %s initialized", pJob->queryId, taskIdx, ctgTaskTypeStr(task.type));
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -320,7 +320,7 @@ int32_t ctgInitGetIndexTask(SCtgJob* pJob, int32_t taskIdx, void* param) {
CTG_ERR_RET(terrno); CTG_ERR_RET(terrno);
} }
qDebug("qid:0x%" PRIx64 " the %dth task type %s initialized, indexFName:%s", pJob->queryId, taskIdx, qDebug("QID:0x%" PRIx64 " the %dth task type %s initialized, indexFName:%s", pJob->queryId, taskIdx,
ctgTaskTypeStr(task.type), name); ctgTaskTypeStr(task.type), name);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -348,7 +348,7 @@ int32_t ctgInitGetUdfTask(SCtgJob* pJob, int32_t taskIdx, void* param) {
CTG_ERR_RET(terrno); CTG_ERR_RET(terrno);
} }
qDebug("qid:0x%" PRIx64 " the %dth task type %s initialized, udfName:%s", pJob->queryId, taskIdx, qDebug("QID:0x%" PRIx64 " the %dth task type %s initialized, udfName:%s", pJob->queryId, taskIdx,
ctgTaskTypeStr(task.type), name); ctgTaskTypeStr(task.type), name);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -376,7 +376,7 @@ int32_t ctgInitGetUserTask(SCtgJob* pJob, int32_t taskIdx, void* param) {
CTG_ERR_RET(terrno); CTG_ERR_RET(terrno);
} }
qDebug("qid:0x%" PRIx64 " the %dth task type %s initialized, user:%s", pJob->queryId, taskIdx, qDebug("QID:0x%" PRIx64 " the %dth task type %s initialized, user:%s", pJob->queryId, taskIdx,
ctgTaskTypeStr(task.type), user->user); ctgTaskTypeStr(task.type), user->user);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -394,7 +394,7 @@ int32_t ctgInitGetSvrVerTask(SCtgJob* pJob, int32_t taskIdx, void* param) {
CTG_ERR_RET(terrno); CTG_ERR_RET(terrno);
} }
qDebug("qid:0x%" PRIx64 " the %dth task type %s initialized", pJob->queryId, taskIdx, ctgTaskTypeStr(task.type)); qDebug("QID:0x%" PRIx64 " the %dth task type %s initialized", pJob->queryId, taskIdx, ctgTaskTypeStr(task.type));
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
@ -426,7 +426,7 @@ int32_t ctgInitGetTbIndexTask(SCtgJob* pJob, int32_t taskIdx, void* param) {
CTG_ERR_RET(terrno); CTG_ERR_RET(terrno);
} }
qDebug("qid:0x%" PRIx64 " the %dth task type %s initialized, tbName:%s", pJob->queryId, taskIdx, qDebug("QID:0x%" PRIx64 " the %dth task type %s initialized, tbName:%s", pJob->queryId, taskIdx,
ctgTaskTypeStr(task.type), name->tname); ctgTaskTypeStr(task.type), name->tname);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -459,7 +459,7 @@ int32_t ctgInitGetTbCfgTask(SCtgJob* pJob, int32_t taskIdx, void* param) {
CTG_ERR_RET(terrno); CTG_ERR_RET(terrno);
} }
qDebug("qid:0x%" PRIx64 " the %dth task type %s initialized, tbName:%s", pJob->queryId, taskIdx, qDebug("QID:0x%" PRIx64 " the %dth task type %s initialized, tbName:%s", pJob->queryId, taskIdx,
ctgTaskTypeStr(task.type), name->tname); ctgTaskTypeStr(task.type), name->tname);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -492,7 +492,7 @@ int32_t ctgInitGetTbTagTask(SCtgJob* pJob, int32_t taskIdx, void* param) {
CTG_ERR_RET(terrno); CTG_ERR_RET(terrno);
} }
qDebug("qid:0x%" PRIx64 " the %dth task type %s initialized, tbName:%s", pJob->queryId, taskIdx, qDebug("QID:0x%" PRIx64 " the %dth task type %s initialized, tbName:%s", pJob->queryId, taskIdx,
ctgTaskTypeStr(task.type), name->tname); ctgTaskTypeStr(task.type), name->tname);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -514,7 +514,7 @@ int32_t ctgInitGetViewsTask(SCtgJob* pJob, int32_t taskIdx, void* param) {
ctx->pNames = param; ctx->pNames = param;
ctx->pResList = taosArrayInit(pJob->viewNum, sizeof(SMetaRes)); ctx->pResList = taosArrayInit(pJob->viewNum, sizeof(SMetaRes));
if (NULL == ctx->pResList) { if (NULL == ctx->pResList) {
qError("qid:0x%" PRIx64 " taosArrayInit %d SMetaRes %d failed", pJob->queryId, pJob->viewNum, qError("QID:0x%" PRIx64 " taosArrayInit %d SMetaRes %d failed", pJob->queryId, pJob->viewNum,
(int32_t)sizeof(SMetaRes)); (int32_t)sizeof(SMetaRes));
ctgFreeTask(&task, true); ctgFreeTask(&task, true);
CTG_ERR_RET(terrno); CTG_ERR_RET(terrno);
@ -525,7 +525,7 @@ int32_t ctgInitGetViewsTask(SCtgJob* pJob, int32_t taskIdx, void* param) {
CTG_ERR_RET(terrno); CTG_ERR_RET(terrno);
} }
qDebug("qid:0x%" PRIx64 " the %dth task type %s initialized, dbNum:%lu, viewNum:%d", pJob->queryId, taskIdx, qDebug("QID:0x%" PRIx64 " the %dth task type %s initialized, dbNum:%lu, viewNum:%d", pJob->queryId, taskIdx,
ctgTaskTypeStr(task.type), taosArrayGetSize(ctx->pNames), pJob->viewNum); ctgTaskTypeStr(task.type), taosArrayGetSize(ctx->pNames), pJob->viewNum);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -546,7 +546,7 @@ int32_t ctgInitGetTbTSMATask(SCtgJob* pJob, int32_t taskId, void* param) {
pTaskCtx->pNames = param; pTaskCtx->pNames = param;
pTaskCtx->pResList = taosArrayInit(pJob->tbTsmaNum, sizeof(SMetaRes)); pTaskCtx->pResList = taosArrayInit(pJob->tbTsmaNum, sizeof(SMetaRes));
if (NULL == pTaskCtx->pResList) { if (NULL == pTaskCtx->pResList) {
qError("qid:0x%" PRIx64 " taosArrayInit %d SMetaRes %d failed", pJob->queryId, pJob->tbTsmaNum, qError("QID:0x%" PRIx64 " taosArrayInit %d SMetaRes %d failed", pJob->queryId, pJob->tbTsmaNum,
(int32_t)sizeof(SMetaRes)); (int32_t)sizeof(SMetaRes));
ctgFreeTask(&task, true); ctgFreeTask(&task, true);
CTG_ERR_RET(terrno); CTG_ERR_RET(terrno);
@ -574,7 +574,7 @@ int32_t ctgInitGetTSMATask(SCtgJob* pJob, int32_t taskId, void* param) {
pTaskCtx->pNames = param; pTaskCtx->pNames = param;
pTaskCtx->pResList = taosArrayInit(pJob->tsmaNum, sizeof(SMetaRes)); pTaskCtx->pResList = taosArrayInit(pJob->tsmaNum, sizeof(SMetaRes));
if (NULL == pTaskCtx->pResList) { if (NULL == pTaskCtx->pResList) {
qError("qid:0x%" PRIx64 " taosArrayInit %d SMetaRes %d failed", pJob->queryId, pJob->tsmaNum, qError("QID:0x%" PRIx64 " taosArrayInit %d SMetaRes %d failed", pJob->queryId, pJob->tsmaNum,
(int32_t)sizeof(SMetaRes)); (int32_t)sizeof(SMetaRes));
ctgFreeTask(&task, true); ctgFreeTask(&task, true);
CTG_ERR_RET(terrno); CTG_ERR_RET(terrno);
@ -603,7 +603,7 @@ static int32_t ctgInitGetTbNamesTask(SCtgJob* pJob, int32_t taskId, void* param)
pTaskCtx->pNames = param; pTaskCtx->pNames = param;
pTaskCtx->pResList = taosArrayInit(pJob->tbNameNum, sizeof(SMetaRes)); pTaskCtx->pResList = taosArrayInit(pJob->tbNameNum, sizeof(SMetaRes));
if (NULL == pTaskCtx->pResList) { if (NULL == pTaskCtx->pResList) {
qError("qid:0x%" PRIx64 " taosArrayInit %d SMetaRes %d failed", pJob->queryId, pJob->tbNameNum, qError("QID:0x%" PRIx64 " taosArrayInit %d SMetaRes %d failed", pJob->queryId, pJob->tbNameNum,
(int32_t)sizeof(SMetaRes)); (int32_t)sizeof(SMetaRes));
ctgFreeTask(&task, true); ctgFreeTask(&task, true);
CTG_ERR_RET(terrno); CTG_ERR_RET(terrno);
@ -1048,7 +1048,7 @@ int32_t ctgInitJob(SCatalog* pCtg, SRequestConnInfo* pConn, SCtgJob** job, const
} }
double el = (taosGetTimestampUs() - st) / 1000.0; double el = (taosGetTimestampUs() - st) / 1000.0;
qDebug("qid:0x%" PRIx64 ", jobId: 0x%" PRIx64 " initialized, task num %d, forceUpdate %d, elapsed time:%.2f ms", qDebug("QID:0x%" PRIx64 ", jobId: 0x%" PRIx64 " initialized, task num %d, forceUpdate %d, elapsed time:%.2f ms",
pJob->queryId, pJob->refId, taskNum, pReq->forceUpdate, el); pJob->queryId, pJob->refId, taskNum, pReq->forceUpdate, el);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -1450,16 +1450,17 @@ _return:
int32_t ctgCallUserCb(void* param) { int32_t ctgCallUserCb(void* param) {
SCtgJob* pJob = (SCtgJob*)param; SCtgJob* pJob = (SCtgJob*)param;
qDebug("qid:0x%" PRIx64 " ctg start to call user cb with rsp %s", pJob->queryId, tstrerror(pJob->jobResCode)); qDebug("QID:0x%" PRIx64 " ctg start to call user cb with rsp %s", pJob->queryId, tstrerror(pJob->jobResCode));
(*pJob->userFp)(&pJob->jobRes, pJob->userParam, pJob->jobResCode); (*pJob->userFp)(&pJob->jobRes, pJob->userParam, pJob->jobResCode);
qDebug("qid:0x%" PRIx64 " ctg end to call user cb", pJob->queryId); qDebug("QID:0x%" PRIx64 " ctg end to call user cb", pJob->queryId);
int64_t refId = pJob->refId; int64_t refId = pJob->refId;
int32_t code = taosRemoveRef(gCtgMgmt.jobPool, refId); int32_t code = taosRemoveRef(gCtgMgmt.jobPool, refId);
if (code) { if (code) {
qError("qid:0x%" PRIx64 " remove ctg job %" PRId64 " from jobPool failed, error:%s", pJob->queryId, refId, tstrerror(code)); qError("QID:0x%" PRIx64 " remove ctg job %" PRId64 " from jobPool failed, error:%s", pJob->queryId, refId,
tstrerror(code));
} }
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -1469,7 +1470,7 @@ void ctgUpdateJobErrCode(SCtgJob* pJob, int32_t errCode) {
if (!NEED_CLIENT_REFRESH_VG_ERROR(errCode) || errCode == TSDB_CODE_SUCCESS) return; if (!NEED_CLIENT_REFRESH_VG_ERROR(errCode) || errCode == TSDB_CODE_SUCCESS) return;
atomic_store_32(&pJob->jobResCode, errCode); atomic_store_32(&pJob->jobResCode, errCode);
qDebug("qid:0x%" PRIx64 " ctg job errCode updated to %s", pJob->queryId, tstrerror(errCode)); qDebug("QID:0x%" PRIx64 " ctg job errCode updated to %s", pJob->queryId, tstrerror(errCode));
return; return;
} }
@ -1481,7 +1482,7 @@ int32_t ctgHandleTaskEnd(SCtgTask* pTask, int32_t rspCode) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
qDebug("qid:0x%" PRIx64 " task %d end with res %s", pJob->queryId, pTask->taskId, tstrerror(rspCode)); qDebug("QID:0x%" PRIx64 " task %d end with res %s", pJob->queryId, pTask->taskId, tstrerror(rspCode));
pTask->code = rspCode; pTask->code = rspCode;
pTask->status = CTG_TASK_DONE; pTask->status = CTG_TASK_DONE;
@ -1490,7 +1491,7 @@ int32_t ctgHandleTaskEnd(SCtgTask* pTask, int32_t rspCode) {
int32_t taskDone = atomic_add_fetch_32(&pJob->taskDone, 1); int32_t taskDone = atomic_add_fetch_32(&pJob->taskDone, 1);
if (taskDone < taosArrayGetSize(pJob->pTasks)) { if (taskDone < taosArrayGetSize(pJob->pTasks)) {
qDebug("qid:0x%" PRIx64 " task done: %d, total: %d", pJob->queryId, taskDone, qDebug("QID:0x%" PRIx64 " task done: %d, total: %d", pJob->queryId, taskDone,
(int32_t)taosArrayGetSize(pJob->pTasks)); (int32_t)taosArrayGetSize(pJob->pTasks));
ctgUpdateJobErrCode(pJob, rspCode); ctgUpdateJobErrCode(pJob, rspCode);
@ -4347,7 +4348,7 @@ int32_t ctgLaunchJob(SCtgJob* pJob) {
CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR); CTG_ERR_RET(TSDB_CODE_CTG_INTERNAL_ERROR);
} }
qDebug("qid:0x%" PRIx64 " ctg launch [%dth] task", pJob->queryId, pTask->taskId); qDebug("QID:0x%" PRIx64 " ctg launch [%dth] task", pJob->queryId, pTask->taskId);
CTG_ERR_RET((*gCtgAsyncFps[pTask->type].launchFp)(pTask)); CTG_ERR_RET((*gCtgAsyncFps[pTask->type].launchFp)(pTask));
pTask = taosArrayGet(pJob->pTasks, i); pTask = taosArrayGet(pJob->pTasks, i);
@ -4360,7 +4361,7 @@ int32_t ctgLaunchJob(SCtgJob* pJob) {
} }
if (taskNum <= 0) { if (taskNum <= 0) {
qDebug("qid:0x%" PRIx64 " ctg call user callback with rsp %s", pJob->queryId, tstrerror(pJob->jobResCode)); qDebug("QID:0x%" PRIx64 " ctg call user callback with rsp %s", pJob->queryId, tstrerror(pJob->jobResCode));
CTG_ERR_RET(taosAsyncExec(ctgCallUserCb, pJob, NULL)); CTG_ERR_RET(taosAsyncExec(ctgCallUserCb, pJob, NULL));
#if CTG_BATCH_FETCH #if CTG_BATCH_FETCH

26
source/libs/catalog/src/ctgRemote.c
View File

@ -47,7 +47,7 @@ int32_t ctgHandleBatchRsp(SCtgJob* pJob, SCtgTaskCallbackParam* cbParam, SDataBu
msgNum = 0; msgNum = 0;
} }
ctgDebug("qid:0x%" PRIx64 " ctg got batch %d rsp %s", pJob->queryId, cbParam->batchId, ctgDebug("QID:0x%" PRIx64 " ctg got batch %d rsp %s", pJob->queryId, cbParam->batchId,
TMSG_INFO(cbParam->reqType + 1)); TMSG_INFO(cbParam->reqType + 1));
SHashObj* pBatchs = taosHashInit(taskNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false, HASH_NO_LOCK); SHashObj* pBatchs = taosHashInit(taskNum, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false, HASH_NO_LOCK);
@ -114,7 +114,7 @@ int32_t ctgHandleBatchRsp(SCtgJob* pJob, SCtgTaskCallbackParam* cbParam, SDataBu
pMsgCtx->pBatchs = pBatchs; pMsgCtx->pBatchs = pBatchs;
ctgDebug("qid:0x%" PRIx64 " ctg task %d idx %d start to handle rsp %s, pBatchs: %p", pJob->queryId, pTask->taskId, ctgDebug("QID:0x%" PRIx64 " ctg task %d idx %d start to handle rsp %s, pBatchs: %p", pJob->queryId, pTask->taskId,
pRsp->msgIdx, TMSG_INFO(taskMsg.msgType + 1), pBatchs); pRsp->msgIdx, TMSG_INFO(taskMsg.msgType + 1), pBatchs);
(void)(*gCtgAsyncFps[pTask->type].handleRspFp)( (void)(*gCtgAsyncFps[pTask->type].handleRspFp)(
@ -454,7 +454,7 @@ int32_t ctgHandleMsgCallback(void* param, SDataBuf* pMsg, int32_t rspCode) {
CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR); CTG_ERR_JRET(TSDB_CODE_CTG_INTERNAL_ERROR);
} }
qDebug("qid:0x%" PRIx64 " ctg task %d start to handle rsp %s", pJob->queryId, pTask->taskId, qDebug("QID:0x%" PRIx64 " ctg task %d start to handle rsp %s", pJob->queryId, pTask->taskId,
TMSG_INFO(cbParam->reqType + 1)); TMSG_INFO(cbParam->reqType + 1));
#if CTG_BATCH_FETCH #if CTG_BATCH_FETCH
@ -808,7 +808,7 @@ int32_t ctgLaunchBatchs(SCatalog* pCtg, SCtgJob* pJob, SHashObj* pBatchs) {
SCtgBatch* pBatch = (SCtgBatch*)p; SCtgBatch* pBatch = (SCtgBatch*)p;
int32_t msgSize = 0; int32_t msgSize = 0;
ctgDebug("qid:0x%" PRIx64 " ctg start to launch batch %d", pJob->queryId, pBatch->batchId); ctgDebug("QID:0x%" PRIx64 " ctg start to launch batch %d", pJob->queryId, pBatch->batchId);
CTG_ERR_JRET(ctgBuildBatchReqMsg(pBatch, *vgId, &msg, &msgSize)); CTG_ERR_JRET(ctgBuildBatchReqMsg(pBatch, *vgId, &msg, &msgSize));
code = ctgAsyncSendMsg(pCtg, &pBatch->conn, pJob, pBatch->pTaskIds, pBatch->batchId, pBatch->pMsgIdxs, code = ctgAsyncSendMsg(pCtg, &pBatch->conn, pJob, pBatch->pTaskIds, pBatch->batchId, pBatch->pMsgIdxs,
@ -1124,7 +1124,8 @@ int32_t ctgGetTbIndexFromMnode(SCatalog* pCtg, SRequestConnInfo* pConn, SName* n
int32_t code = tNameExtractFullName(name, tbFName); int32_t code = tNameExtractFullName(name, tbFName);
if (code) { if (code) {
ctgError("tNameExtractFullName failed, code:%s, type:%d, dbName:%s, tname:%s", tstrerror(code), name->type, name->dbname, name->tname); ctgError("tNameExtractFullName failed, code:%s, type:%d, dbName:%s, tname:%s", tstrerror(code), name->type,
name->dbname, name->tname);
CTG_ERR_RET(code); CTG_ERR_RET(code);
} }
@ -1450,7 +1451,8 @@ int32_t ctgGetTableCfgFromVnode(SCatalog* pCtg, SRequestConnInfo* pConn, const S
int32_t code = tNameExtractFullName(pTableName, tbFName); int32_t code = tNameExtractFullName(pTableName, tbFName);
if (code) { if (code) {
ctgError("tNameExtractFullName failed, code:%s, type:%d, dbName:%s, tname:%s", tstrerror(code), pTableName->type, pTableName->dbname, pTableName->tname); ctgError("tNameExtractFullName failed, code:%s, type:%d, dbName:%s, tname:%s", tstrerror(code), pTableName->type,
pTableName->dbname, pTableName->tname);
CTG_ERR_RET(code); CTG_ERR_RET(code);
} }
@ -1523,7 +1525,8 @@ int32_t ctgGetTableCfgFromMnode(SCatalog* pCtg, SRequestConnInfo* pConn, const S
int32_t code = tNameExtractFullName(pTableName, tbFName); int32_t code = tNameExtractFullName(pTableName, tbFName);
if (code) { if (code) {
ctgError("tNameExtractFullName failed, code:%s, type:%d, dbName:%s, tname:%s", tstrerror(code), pTableName->type, pTableName->dbname, pTableName->tname); ctgError("tNameExtractFullName failed, code:%s, type:%d, dbName:%s, tname:%s", tstrerror(code), pTableName->type,
pTableName->dbname, pTableName->tname);
CTG_ERR_RET(code); CTG_ERR_RET(code);
} }
@ -1635,7 +1638,8 @@ int32_t ctgGetViewInfoFromMnode(SCatalog* pCtg, SRequestConnInfo* pConn, SName*
char fullName[TSDB_TABLE_FNAME_LEN]; char fullName[TSDB_TABLE_FNAME_LEN];
int32_t code = tNameExtractFullName(pName, fullName); int32_t code = tNameExtractFullName(pName, fullName);
if (code) { if (code) {
ctgError("tNameExtractFullName failed, code:%s, type:%d, dbName:%s, tname:%s", tstrerror(code), pName->type, pName->dbname, pName->tname); ctgError("tNameExtractFullName failed, code:%s, type:%d, dbName:%s, tname:%s", tstrerror(code), pName->type,
pName->dbname, pName->tname);
CTG_ERR_RET(code); CTG_ERR_RET(code);
} }
@ -1696,7 +1700,8 @@ int32_t ctgGetTbTSMAFromMnode(SCatalog* pCtg, SRequestConnInfo* pConn, const SNa
char tbFName[TSDB_TABLE_FNAME_LEN]; char tbFName[TSDB_TABLE_FNAME_LEN];
int32_t code = tNameExtractFullName(name, tbFName); int32_t code = tNameExtractFullName(name, tbFName);
if (code) { if (code) {
ctgError("tNameExtractFullName failed, code:%s, type:%d, dbName:%s, tname:%s", tstrerror(code), name->type, name->dbname, name->tname); ctgError("tNameExtractFullName failed, code:%s, type:%d, dbName:%s, tname:%s", tstrerror(code), name->type,
name->dbname, name->tname);
CTG_ERR_RET(code); CTG_ERR_RET(code);
} }
@ -1757,7 +1762,8 @@ int32_t ctgGetStreamProgressFromVnode(SCatalog* pCtg, SRequestConnInfo* pConn, c
char tbFName[TSDB_TABLE_FNAME_LEN]; char tbFName[TSDB_TABLE_FNAME_LEN];
int32_t code = tNameExtractFullName(pTbName, tbFName); int32_t code = tNameExtractFullName(pTbName, tbFName);
if (code) { if (code) {
ctgError("tNameExtractFullName failed, code:%s, type:%d, dbName:%s, tname:%s", tstrerror(code), pTbName->type, pTbName->dbname, pTbName->tname); ctgError("tNameExtractFullName failed, code:%s, type:%d, dbName:%s, tname:%s", tstrerror(code), pTbName->type,
pTbName->dbname, pTbName->tname);
CTG_ERR_RET(code); CTG_ERR_RET(code);
} }

17
source/libs/catalog/src/ctgUtil.c
View File

@ -452,7 +452,8 @@ void ctgClearHandleMeta(SCatalog* pCtg, int64_t* pClearedSize, int64_t* pCleardN
code = taosHashRemove(dbCache->tbCache, key, len); code = taosHashRemove(dbCache->tbCache, key, len);
if (code) { if (code) {
qError("taosHashRemove table cache failed, key:%s, len:%d, error:%s", (char*)key, (int32_t)len, tstrerror(code)); qError("taosHashRemove table cache failed, key:%s, len:%d, error:%s", (char*)key, (int32_t)len,
tstrerror(code));
} }
cacheSize = cacheSize =
@ -1096,7 +1097,7 @@ void ctgFreeJob(void* job) {
taosMemoryFree(job); taosMemoryFree(job);
qDebug("qid:0x%" PRIx64 ", ctg job 0x%" PRIx64 " freed", qid, rid); qDebug("QID:0x%" PRIx64 ", ctg job 0x%" PRIx64 " freed", qid, rid);
} }
int32_t ctgUpdateMsgCtx(SCtgMsgCtx* pCtx, int32_t reqType, void* out, char* target) { int32_t ctgUpdateMsgCtx(SCtgMsgCtx* pCtx, int32_t reqType, void* out, char* target) {
@ -1241,7 +1242,8 @@ int32_t ctgGetVgInfoFromHashValue(SCatalog* pCtg, SEpSet* pMgmtEps, SDBVgInfo* d
char tbFullName[TSDB_TABLE_FNAME_LEN]; char tbFullName[TSDB_TABLE_FNAME_LEN];
code = tNameExtractFullName(pTableName, tbFullName); code = tNameExtractFullName(pTableName, tbFullName);
if (code) { if (code) {
ctgError("tNameExtractFullName failed, error:%s, type:%d, dbName:%s, tname:%s", tstrerror(code), pTableName->type, pTableName->dbname, pTableName->tname); ctgError("tNameExtractFullName failed, error:%s, type:%d, dbName:%s, tname:%s", tstrerror(code), pTableName->type,
pTableName->dbname, pTableName->tname);
CTG_ERR_RET(code); CTG_ERR_RET(code);
} }
@ -1704,7 +1706,8 @@ int32_t ctgCloneTableIndex(SArray* pIndex, SArray** pRes) {
} }
int32_t ctgUpdateSendTargetInfo(SMsgSendInfo* pMsgSendInfo, int32_t msgType, char* dbFName, int32_t vgId) { int32_t ctgUpdateSendTargetInfo(SMsgSendInfo* pMsgSendInfo, int32_t msgType, char* dbFName, int32_t vgId) {
if (msgType == TDMT_VND_TABLE_META || msgType == TDMT_VND_TABLE_CFG || msgType == TDMT_VND_BATCH_META || msgType == TDMT_VND_TABLE_NAME) { if (msgType == TDMT_VND_TABLE_META || msgType == TDMT_VND_TABLE_CFG || msgType == TDMT_VND_BATCH_META ||
msgType == TDMT_VND_TABLE_NAME) {
pMsgSendInfo->target.type = TARGET_TYPE_VNODE; pMsgSendInfo->target.type = TARGET_TYPE_VNODE;
pMsgSendInfo->target.vgId = vgId; pMsgSendInfo->target.vgId = vgId;
pMsgSendInfo->target.dbFName = taosStrdup(dbFName); pMsgSendInfo->target.dbFName = taosStrdup(dbFName);
@ -2010,7 +2013,8 @@ int32_t ctgChkSetTbAuthRes(SCatalog* pCtg, SCtgAuthReq* req, SCtgAuthRsp* res) {
char dbFName[TSDB_DB_FNAME_LEN]; char dbFName[TSDB_DB_FNAME_LEN];
code = tNameExtractFullName(&req->pRawReq->tbName, tbFName); code = tNameExtractFullName(&req->pRawReq->tbName, tbFName);
if (code) { if (code) {
ctgError("tNameExtractFullName failed, error:%s, type:%d, dbName:%s, tname:%s", tstrerror(code), req->pRawReq->tbName.type, req->pRawReq->tbName.dbname, req->pRawReq->tbName.tname); ctgError("tNameExtractFullName failed, error:%s, type:%d, dbName:%s, tname:%s", tstrerror(code),
req->pRawReq->tbName.type, req->pRawReq->tbName.dbname, req->pRawReq->tbName.tname);
CTG_ERR_RET(code); CTG_ERR_RET(code);
} }
@ -2201,7 +2205,8 @@ int32_t ctgChkSetViewAuthRes(SCatalog* pCtg, SCtgAuthReq* req, SCtgAuthRsp* res)
} else { } else {
code = tNameExtractFullName(&req->pRawReq->tbName, viewFName); code = tNameExtractFullName(&req->pRawReq->tbName, viewFName);
if (code) { if (code) {
ctgError("tNameExtractFullName failed, error:%s, type:%d, dbName:%s, tname:%s", tstrerror(code), req->pRawReq->tbName.type, req->pRawReq->tbName.dbname, req->pRawReq->tbName.tname); ctgError("tNameExtractFullName failed, error:%s, type:%d, dbName:%s, tname:%s", tstrerror(code),
req->pRawReq->tbName.type, req->pRawReq->tbName.dbname, req->pRawReq->tbName.tname);
CTG_ERR_RET(code); CTG_ERR_RET(code);
} }
} }

2
source/libs/executor/src/executil.c
View File

@ -1143,11 +1143,11 @@ SSDataBlock* createTagValBlockForFilter(SArray* pColList, int32_t numOfTables, S
varDataSetLen(tmp, tagVal.nData); varDataSetLen(tmp, tagVal.nData);
memcpy(tmp + VARSTR_HEADER_SIZE, tagVal.pData, tagVal.nData); memcpy(tmp + VARSTR_HEADER_SIZE, tagVal.pData, tagVal.nData);
code = colDataSetVal(pColInfo, i, tmp, false); code = colDataSetVal(pColInfo, i, tmp, false);
QUERY_CHECK_CODE(code, lino, _end);
#if TAG_FILTER_DEBUG #if TAG_FILTER_DEBUG
qDebug("tagfilter varch:%s", tmp + 2); qDebug("tagfilter varch:%s", tmp + 2);
#endif #endif
taosMemoryFree(tmp); taosMemoryFree(tmp);
QUERY_CHECK_CODE(code, lino, _end);
} else { } else {
code = colDataSetVal(pColInfo, i, (const char*)&tagVal.i64, false); code = colDataSetVal(pColInfo, i, (const char*)&tagVal.i64, false);
QUERY_CHECK_CODE(code, lino, _end); QUERY_CHECK_CODE(code, lino, _end);

25
source/libs/executor/src/streamfilloperator.c
View File

@ -1148,12 +1148,12 @@ _end:
return code; return code;
} }
static int32_t initResultBuf(SStreamFillSupporter* pFillSup) { static int32_t initResultBuf(SSDataBlock* pInputRes, SStreamFillSupporter* pFillSup) {
pFillSup->rowSize = sizeof(SResultCellData) * pFillSup->numOfAllCols; int32_t numOfCols = taosArrayGetSize(pInputRes->pDataBlock);
for (int i = 0; i < pFillSup->numOfAllCols; i++) { pFillSup->rowSize = sizeof(SResultCellData) * numOfCols;
SFillColInfo* pCol = &pFillSup->pAllColInfo[i]; for (int i = 0; i < numOfCols; i++) {
SResSchema* pSchema = &pCol->pExpr->base.resSchema; SColumnInfoData* pCol = taosArrayGet(pInputRes->pDataBlock, i);
pFillSup->rowSize += pSchema->bytes; pFillSup->rowSize += pCol->info.bytes;
} }
pFillSup->next.key = INT64_MIN; pFillSup->next.key = INT64_MIN;
pFillSup->nextNext.key = INT64_MIN; pFillSup->nextNext.key = INT64_MIN;
@ -1168,7 +1168,7 @@ static int32_t initResultBuf(SStreamFillSupporter* pFillSup) {
} }
static SStreamFillSupporter* initStreamFillSup(SStreamFillPhysiNode* pPhyFillNode, SInterval* pInterval, static SStreamFillSupporter* initStreamFillSup(SStreamFillPhysiNode* pPhyFillNode, SInterval* pInterval,
SExprInfo* pFillExprInfo, int32_t numOfFillCols, SStorageAPI* pAPI) { SExprInfo* pFillExprInfo, int32_t numOfFillCols, SStorageAPI* pAPI, SSDataBlock* pInputRes) {
int32_t code = TSDB_CODE_SUCCESS; int32_t code = TSDB_CODE_SUCCESS;
int32_t lino = 0; int32_t lino = 0;
SStreamFillSupporter* pFillSup = taosMemoryCalloc(1, sizeof(SStreamFillSupporter)); SStreamFillSupporter* pFillSup = taosMemoryCalloc(1, sizeof(SStreamFillSupporter));
@ -1197,7 +1197,7 @@ static SStreamFillSupporter* initStreamFillSup(SStreamFillPhysiNode* pPhyFillNod
pFillSup->interval = *pInterval; pFillSup->interval = *pInterval;
pFillSup->pAPI = pAPI; pFillSup->pAPI = pAPI;
code = initResultBuf(pFillSup); code = initResultBuf(pInputRes, pFillSup);
QUERY_CHECK_CODE(code, lino, _end); QUERY_CHECK_CODE(code, lino, _end);
SExprInfo* noFillExpr = NULL; SExprInfo* noFillExpr = NULL;
@ -1389,7 +1389,11 @@ int32_t createStreamFillOperatorInfo(SOperatorInfo* downstream, SStreamFillPhysi
code = initExprSupp(&pOperator->exprSupp, pFillExprInfo, numOfFillCols, &pTaskInfo->storageAPI.functionStore); code = initExprSupp(&pOperator->exprSupp, pFillExprInfo, numOfFillCols, &pTaskInfo->storageAPI.functionStore);
QUERY_CHECK_CODE(code, lino, _error); QUERY_CHECK_CODE(code, lino, _error);
pInfo->pFillSup = initStreamFillSup(pPhyFillNode, pInterval, pFillExprInfo, numOfFillCols, &pTaskInfo->storageAPI); pInfo->pSrcBlock = createDataBlockFromDescNode(pPhyFillNode->node.pOutputDataBlockDesc);
QUERY_CHECK_NULL(pInfo->pSrcBlock, code, lino, _error, terrno);
pInfo->pFillSup = initStreamFillSup(pPhyFillNode, pInterval, pFillExprInfo, numOfFillCols, &pTaskInfo->storageAPI,
pInfo->pSrcBlock);
if (!pInfo->pFillSup) { if (!pInfo->pFillSup) {
code = TSDB_CODE_FAILED; code = TSDB_CODE_FAILED;
QUERY_CHECK_CODE(code, lino, _error); QUERY_CHECK_CODE(code, lino, _error);
@ -1398,8 +1402,7 @@ int32_t createStreamFillOperatorInfo(SOperatorInfo* downstream, SStreamFillPhysi
initResultSizeInfo(&pOperator->resultInfo, 4096); initResultSizeInfo(&pOperator->resultInfo, 4096);
pInfo->pRes = createDataBlockFromDescNode(pPhyFillNode->node.pOutputDataBlockDesc); pInfo->pRes = createDataBlockFromDescNode(pPhyFillNode->node.pOutputDataBlockDesc);
QUERY_CHECK_NULL(pInfo->pRes, code, lino, _error, terrno); QUERY_CHECK_NULL(pInfo->pRes, code, lino, _error, terrno);
pInfo->pSrcBlock = createDataBlockFromDescNode(pPhyFillNode->node.pOutputDataBlockDesc);
QUERY_CHECK_NULL(pInfo->pSrcBlock, code, lino, _error, terrno);
code = blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity); code = blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity);
QUERY_CHECK_CODE(code, lino, _error); QUERY_CHECK_CODE(code, lino, _error);

6
source/libs/planner/src/planner.c
View File

@ -37,7 +37,7 @@ static int32_t dumpQueryPlan(SQueryPlan* pPlan) {
char* pStr = NULL; char* pStr = NULL;
code = nodesNodeToString((SNode*)pPlan, false, &pStr, NULL); code = nodesNodeToString((SNode*)pPlan, false, &pStr, NULL);
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
planDebugL("qid:0x%" PRIx64 " Query Plan, JsonPlan: %s", pPlan->queryId, pStr); planDebugL("QID:0x%" PRIx64 " Query Plan, JsonPlan: %s", pPlan->queryId, pStr);
taosMemoryFree(pStr); taosMemoryFree(pStr);
} }
return code; return code;
@ -123,7 +123,7 @@ int32_t qContinuePlanPostQuery(void* pPostPlan) {
} }
int32_t qSetSubplanExecutionNode(SSubplan* subplan, int32_t groupId, SDownstreamSourceNode* pSource) { int32_t qSetSubplanExecutionNode(SSubplan* subplan, int32_t groupId, SDownstreamSourceNode* pSource) {
planDebug("qid:0x%" PRIx64 " set subplan execution node, groupId:%d", subplan->id.queryId, groupId); planDebug("QID:0x%" PRIx64 " set subplan execution node, groupId:%d", subplan->id.queryId, groupId);
return setSubplanExecutionNode(subplan->pNode, groupId, pSource); return setSubplanExecutionNode(subplan->pNode, groupId, pSource);
} }
@ -143,7 +143,7 @@ static void clearSubplanExecutionNode(SPhysiNode* pNode) {
} }
void qClearSubplanExecutionNode(SSubplan* pSubplan) { void qClearSubplanExecutionNode(SSubplan* pSubplan) {
planDebug("qid:0x%" PRIx64 " clear subplan execution node, groupId:%d", pSubplan->id.queryId, pSubplan->id.groupId); planDebug("QID:0x%" PRIx64 " clear subplan execution node, groupId:%d", pSubplan->id.queryId, pSubplan->id.groupId);
clearSubplanExecutionNode(pSubplan->pNode); clearSubplanExecutionNode(pSubplan->pNode);
} }

12
source/libs/scalar/src/scalar.c
View File

@ -628,8 +628,8 @@ int32_t sclWalkCaseWhenList(SScalarCtx *ctx, SNodeList *pList, struct SListCell
cell = cell->pNext) { cell = cell->pNext) {
pWhenThen = (SWhenThenNode *)node; pWhenThen = (SWhenThenNode *)node;
SCL_ERR_RET(sclGetNodeRes(pWhenThen->pWhen, ctx, &pWhen)); SCL_ERR_JRET(sclGetNodeRes(pWhenThen->pWhen, ctx, &pWhen));
SCL_ERR_RET(sclGetNodeRes(pWhenThen->pThen, ctx, &pThen)); SCL_ERR_JRET(sclGetNodeRes(pWhenThen->pThen, ctx, &pThen));
SCL_ERR_JRET(vectorCompareImpl(pCase, pWhen, pComp, rowIdx, 1, TSDB_ORDER_ASC, OP_TYPE_EQUAL)); SCL_ERR_JRET(vectorCompareImpl(pCase, pWhen, pComp, rowIdx, 1, TSDB_ORDER_ASC, OP_TYPE_EQUAL));
@ -646,6 +646,10 @@ int32_t sclWalkCaseWhenList(SScalarCtx *ctx, SNodeList *pList, struct SListCell
goto _return; goto _return;
} }
sclFreeParam(pWhen);
sclFreeParam(pThen);
taosMemoryFreeClear(pWhen);
taosMemoryFreeClear(pThen);
} }
if (pElse) { if (pElse) {
@ -672,8 +676,8 @@ _return:
sclFreeParam(pWhen); sclFreeParam(pWhen);
sclFreeParam(pThen); sclFreeParam(pThen);
taosMemoryFree(pWhen); taosMemoryFreeClear(pWhen);
taosMemoryFree(pThen); taosMemoryFreeClear(pThen);
SCL_RET(code); SCL_RET(code);
} }

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

@ -728,7 +728,7 @@ void schFreeJobImpl(void *job) {
uint64_t queryId = pJob->queryId; uint64_t queryId = pJob->queryId;
int64_t refId = pJob->refId; int64_t refId = pJob->refId;
qDebug("qid:0x%" PRIx64 " begin to free sch job, refId:0x%" PRIx64 ", pointer:%p", queryId, refId, pJob); qDebug("QID:0x%" PRIx64 " begin to free sch job, refId:0x%" PRIx64 ", pointer:%p", queryId, refId, pJob);
schDropJobAllTasks(pJob); schDropJobAllTasks(pJob);
@ -786,7 +786,7 @@ void schFreeJobImpl(void *job) {
schCloseJobRef(); schCloseJobRef();
} }
qDebug("qid:0x%" PRIx64 " sch job freed, refId:0x%" PRIx64 ", pointer:%p", queryId, refId, pJob); qDebug("QID:0x%" PRIx64 " sch job freed, refId:0x%" PRIx64 ", pointer:%p", queryId, refId, pJob);
} }
int32_t schJobFetchRows(SSchJob *pJob) { int32_t schJobFetchRows(SSchJob *pJob) {
@ -821,7 +821,7 @@ int32_t schInitJob(int64_t *pJobId, SSchedulerReq *pReq) {
int64_t refId = -1; int64_t refId = -1;
SSchJob *pJob = taosMemoryCalloc(1, sizeof(SSchJob)); SSchJob *pJob = taosMemoryCalloc(1, sizeof(SSchJob));
if (NULL == pJob) { if (NULL == pJob) {
qError("qid:0x%" PRIx64 " calloc %d failed", pReq->pDag->queryId, (int32_t)sizeof(SSchJob)); qError("QID:0x%" PRIx64 " calloc %d failed", pReq->pDag->queryId, (int32_t)sizeof(SSchJob));
SCH_ERR_JRET(terrno); SCH_ERR_JRET(terrno);
} }
@ -831,7 +831,7 @@ int32_t schInitJob(int64_t *pJobId, SSchedulerReq *pReq) {
if (pReq->sql) { if (pReq->sql) {
pJob->sql = taosStrdup(pReq->sql); pJob->sql = taosStrdup(pReq->sql);
if (NULL == pJob->sql) { if (NULL == pJob->sql) {
qError("qid:0x%" PRIx64 " strdup sql %s failed", pReq->pDag->queryId, pReq->sql); qError("QID:0x%" PRIx64 " strdup sql %s failed", pReq->pDag->queryId, pReq->sql);
SCH_ERR_JRET(terrno); SCH_ERR_JRET(terrno);
} }
} }
@ -839,7 +839,7 @@ int32_t schInitJob(int64_t *pJobId, SSchedulerReq *pReq) {
if (pReq->allocatorRefId > 0) { if (pReq->allocatorRefId > 0) {
pJob->allocatorRefId = nodesMakeAllocatorWeakRef(pReq->allocatorRefId); pJob->allocatorRefId = nodesMakeAllocatorWeakRef(pReq->allocatorRefId);
if (pJob->allocatorRefId <= 0) { if (pJob->allocatorRefId <= 0) {
qError("qid:0x%" PRIx64 " nodesMakeAllocatorWeakRef failed", pReq->pDag->queryId); qError("QID:0x%" PRIx64 " nodesMakeAllocatorWeakRef failed", pReq->pDag->queryId);
SCH_ERR_JRET(terrno); SCH_ERR_JRET(terrno);
} }
} }
@ -851,11 +851,11 @@ int32_t schInitJob(int64_t *pJobId, SSchedulerReq *pReq) {
pJob->pWorkerCb = pReq->pWorkerCb; pJob->pWorkerCb = pReq->pWorkerCb;
if (pReq->pNodeList == NULL || taosArrayGetSize(pReq->pNodeList) <= 0) { if (pReq->pNodeList == NULL || taosArrayGetSize(pReq->pNodeList) <= 0) {
qDebug("qid:0x%" PRIx64 " input exec nodeList is empty", pReq->pDag->queryId); qDebug("QID:0x%" PRIx64 " input exec nodeList is empty", pReq->pDag->queryId);
} else { } else {
pJob->nodeList = taosArrayDup(pReq->pNodeList, NULL); pJob->nodeList = taosArrayDup(pReq->pNodeList, NULL);
if (NULL == pJob->nodeList) { if (NULL == pJob->nodeList) {
qError("qid:0x%" PRIx64 " taosArrayDup failed, origNum:%d", pReq->pDag->queryId, qError("QID:0x%" PRIx64 " taosArrayDup failed, origNum:%d", pReq->pDag->queryId,
(int32_t)taosArrayGetSize(pReq->pNodeList)); (int32_t)taosArrayGetSize(pReq->pNodeList));
SCH_ERR_JRET(terrno); SCH_ERR_JRET(terrno);
} }
@ -918,7 +918,7 @@ _return:
int32_t schExecJob(SSchJob *pJob, SSchedulerReq *pReq) { int32_t schExecJob(SSchJob *pJob, SSchedulerReq *pReq) {
int32_t code = 0; int32_t code = 0;
qDebug("qid:0x%" PRIx64 " sch job refId 0x%" PRIx64 " started", pReq->pDag->queryId, pJob->refId); qDebug("QID:0x%" PRIx64 " sch job refId 0x%" PRIx64 " started", pReq->pDag->queryId, pJob->refId);
SCH_ERR_RET(schLaunchJob(pJob)); SCH_ERR_RET(schLaunchJob(pJob));
@ -926,7 +926,7 @@ int32_t schExecJob(SSchJob *pJob, SSchedulerReq *pReq) {
SCH_JOB_DLOG("sync wait for rsp now, job status:%s", SCH_GET_JOB_STATUS_STR(pJob)); SCH_JOB_DLOG("sync wait for rsp now, job status:%s", SCH_GET_JOB_STATUS_STR(pJob));
code = tsem_wait(&pJob->rspSem); code = tsem_wait(&pJob->rspSem);
if (code) { if (code) {
qError("qid:0x%" PRIx64 " tsem_wait sync rspSem failed, error:%s", pReq->pDag->queryId, tstrerror(code)); qError("QID:0x%" PRIx64 " tsem_wait sync rspSem failed, error:%s", pReq->pDag->queryId, tstrerror(code));
SCH_ERR_RET(code); SCH_ERR_RET(code);
} }
} }
@ -1191,7 +1191,7 @@ int32_t schProcessOnCbBegin(SSchJob **job, SSchTask **task, uint64_t qId, int64_
(void)schAcquireJob(rId, &pJob); (void)schAcquireJob(rId, &pJob);
if (NULL == pJob) { if (NULL == pJob) {
qWarn("qid:0x%" PRIx64 ",TID:0x%" PRIx64 "job no exist, may be dropped, refId:0x%" PRIx64, qId, tId, rId); qWarn("QID:0x%" PRIx64 ",TID:0x%" PRIx64 "job no exist, may be dropped, refId:0x%" PRIx64, qId, tId, rId);
SCH_ERR_RET(TSDB_CODE_QRY_JOB_NOT_EXIST); SCH_ERR_RET(TSDB_CODE_QRY_JOB_NOT_EXIST);
} }

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

@ -500,7 +500,7 @@ _return:
int32_t schHandleDropCallback(void *param, SDataBuf *pMsg, int32_t code) { int32_t schHandleDropCallback(void *param, SDataBuf *pMsg, int32_t code) {
SSchTaskCallbackParam *pParam = (SSchTaskCallbackParam *)param; SSchTaskCallbackParam *pParam = (SSchTaskCallbackParam *)param;
qDebug("qid:0x%" PRIx64 ",TID:0x%" PRIx64 " drop task rsp received, code:0x%x", pParam->queryId, pParam->taskId, qDebug("QID:0x%" PRIx64 ",TID:0x%" PRIx64 " drop task rsp received, code:0x%x", pParam->queryId, pParam->taskId,
code); code);
// called if drop task rsp received code // called if drop task rsp received code
(void)rpcReleaseHandle(pMsg->handle, TAOS_CONN_CLIENT); // ignore error (void)rpcReleaseHandle(pMsg->handle, TAOS_CONN_CLIENT); // ignore error
@ -513,7 +513,7 @@ int32_t schHandleDropCallback(void *param, SDataBuf *pMsg, int32_t code) {
int32_t schHandleNotifyCallback(void *param, SDataBuf *pMsg, int32_t code) { int32_t schHandleNotifyCallback(void *param, SDataBuf *pMsg, int32_t code) {
SSchTaskCallbackParam *pParam = (SSchTaskCallbackParam *)param; SSchTaskCallbackParam *pParam = (SSchTaskCallbackParam *)param;
qDebug("qid:0x%" PRIx64 ",TID:0x%" PRIx64 " task notify rsp received, code:0x%x", pParam->queryId, pParam->taskId, qDebug("QID:0x%" PRIx64 ",TID:0x%" PRIx64 " task notify rsp received, code:0x%x", pParam->queryId, pParam->taskId,
code); code);
if (pMsg) { if (pMsg) {
taosMemoryFree(pMsg->pData); taosMemoryFree(pMsg->pData);

8
source/libs/scheduler/src/schTask.c
View File

@ -996,7 +996,7 @@ int32_t schProcessOnTaskStatusRsp(SQueryNodeEpId *pEpId, SArray *pStatusList) {
int32_t code = 0; int32_t code = 0;
qDebug("qid:0x%" PRIx64 ",TID:0x%" PRIx64 ",EID:%d task status in server: %s", pStatus->queryId, pStatus->taskId, qDebug("QID:0x%" PRIx64 ",TID:0x%" PRIx64 ",EID:%d task status in server: %s", pStatus->queryId, pStatus->taskId,
pStatus->execId, jobTaskStatusStr(pStatus->status)); pStatus->execId, jobTaskStatusStr(pStatus->status));
if (schProcessOnCbBegin(&pJob, &pTask, pStatus->queryId, pStatus->refId, pStatus->taskId)) { if (schProcessOnCbBegin(&pJob, &pTask, pStatus->queryId, pStatus->refId, pStatus->taskId)) {
@ -1043,12 +1043,12 @@ int32_t schHandleExplainRes(SArray *pExplainRes) {
continue; continue;
} }
qDebug("qid:0x%" PRIx64 ",TID:0x%" PRIx64 ", begin to handle LOCAL explain rsp msg", localRsp->qId, localRsp->tId); qDebug("QID:0x%" PRIx64 ",TID:0x%" PRIx64 ", begin to handle LOCAL explain rsp msg", localRsp->qId, localRsp->tId);
pJob = NULL; pJob = NULL;
(void)schAcquireJob(localRsp->rId, &pJob); (void)schAcquireJob(localRsp->rId, &pJob);
if (NULL == pJob) { if (NULL == pJob) {
qWarn("qid:0x%" PRIx64 ",TID:0x%" PRIx64 "job no exist, may be dropped, refId:0x%" PRIx64, localRsp->qId, qWarn("QID:0x%" PRIx64 ",TID:0x%" PRIx64 "job no exist, may be dropped, refId:0x%" PRIx64, localRsp->qId,
localRsp->tId, localRsp->rId); localRsp->tId, localRsp->rId);
SCH_ERR_JRET(TSDB_CODE_QRY_JOB_NOT_EXIST); SCH_ERR_JRET(TSDB_CODE_QRY_JOB_NOT_EXIST);
} }
@ -1068,7 +1068,7 @@ int32_t schHandleExplainRes(SArray *pExplainRes) {
(void)schReleaseJob(pJob->refId); (void)schReleaseJob(pJob->refId);
qDebug("qid:0x%" PRIx64 ",TID:0x%" PRIx64 ", end to handle LOCAL explain rsp msg, code:%x", localRsp->qId, qDebug("QID:0x%" PRIx64 ",TID:0x%" PRIx64 ", end to handle LOCAL explain rsp msg, code:%x", localRsp->qId,
localRsp->tId, code); localRsp->tId, code);
SCH_ERR_JRET(code); SCH_ERR_JRET(code);

10
source/libs/stream/src/streamCheckStatus.c
View File

@ -171,14 +171,14 @@ void streamTaskProcessCheckMsg(SStreamMeta* pMeta, SStreamTaskCheckReq* pReq, SS
streamTaskCheckStatus(pTask, pReq->upstreamTaskId, pReq->upstreamNodeId, pReq->stage, &pRsp->oldStage); streamTaskCheckStatus(pTask, pReq->upstreamTaskId, pReq->upstreamNodeId, pReq->stage, &pRsp->oldStage);
SStreamTaskState pState = streamTaskGetStatus(pTask); SStreamTaskState pState = streamTaskGetStatus(pTask);
stDebug("s-task:%s status:%s, stage:%" PRId64 " recv task check req(qid:0x%" PRIx64 stDebug("s-task:%s status:%s, stage:%" PRId64 " recv task check req(QID:0x%" PRIx64
") task:0x%x (vgId:%d), check_status:%d", ") task:0x%x (vgId:%d), check_status:%d",
pTask->id.idStr, pState.name, pRsp->oldStage, pRsp->reqId, pRsp->upstreamTaskId, pRsp->upstreamNodeId, pTask->id.idStr, pState.name, pRsp->oldStage, pRsp->reqId, pRsp->upstreamTaskId, pRsp->upstreamNodeId,
pRsp->status); pRsp->status);
streamMetaReleaseTask(pMeta, pTask); streamMetaReleaseTask(pMeta, pTask);
} else { } else {
pRsp->status = TASK_DOWNSTREAM_NOT_READY; pRsp->status = TASK_DOWNSTREAM_NOT_READY;
stDebug("tq recv task check(taskId:0x%" PRIx64 "-0x%x not built yet) req(qid:0x%" PRIx64 stDebug("tq recv task check(taskId:0x%" PRIx64 "-0x%x not built yet) req(QID:0x%" PRIx64
") from task:0x%x (vgId:%d), rsp check_status %d", ") from task:0x%x (vgId:%d), rsp check_status %d",
pReq->streamId, taskId, pRsp->reqId, pRsp->upstreamTaskId, pRsp->upstreamNodeId, pRsp->status); pReq->streamId, taskId, pRsp->reqId, pRsp->upstreamTaskId, pRsp->upstreamNodeId, pRsp->status);
} }
@ -259,7 +259,8 @@ int32_t streamTaskSendCheckRsp(const SStreamMeta* pMeta, int32_t vgId, SStreamTa
void* buf = rpcMallocCont(sizeof(SMsgHead) + len); void* buf = rpcMallocCont(sizeof(SMsgHead) + len);
if (buf == NULL) { if (buf == NULL) {
stError("s-task:0x%x vgId:%d failed prepare msg, %s at line:%d code:%s", taskId, pMeta->vgId, __func__, __LINE__, tstrerror(code)); stError("s-task:0x%x vgId:%d failed prepare msg, %s at line:%d code:%s", taskId, pMeta->vgId, __func__, __LINE__,
tstrerror(code));
return terrno; return terrno;
} }
@ -373,7 +374,8 @@ void processDownstreamReadyRsp(SStreamTask* pTask) {
int32_t addIntoNodeUpdateList(SStreamTask* pTask, int32_t nodeId) { int32_t addIntoNodeUpdateList(SStreamTask* pTask, int32_t nodeId) {
int32_t vgId = pTask->pMeta->vgId; int32_t vgId = pTask->pMeta->vgId;
int32_t code = 0;; int32_t code = 0;
;
bool existed = false; bool existed = false;
streamMutexLock(&pTask->lock); streamMutexLock(&pTask->lock);

131
source/libs/stream/src/streamCheckpoint.c
View File

@ -120,38 +120,39 @@ int32_t streamProcessCheckpointSourceReq(SStreamTask* pTask, SStreamCheckpointSo
} }
int32_t streamTaskProcessCheckpointTriggerRsp(SStreamTask* pTask, SCheckpointTriggerRsp* pRsp) { int32_t streamTaskProcessCheckpointTriggerRsp(SStreamTask* pTask, SCheckpointTriggerRsp* pRsp) {
SActiveCheckpointInfo* pInfo = pTask->chkInfo.pActiveInfo;
bool unQualified = false;
const char* id = pTask->id.idStr;
if (pTask->info.taskLevel == TASK_LEVEL__SOURCE) { if (pTask->info.taskLevel == TASK_LEVEL__SOURCE) {
stError("s-task:%s invalid msg recv, checkpoint-trigger rsp not handled", pTask->id.idStr); stError("s-task:%s invalid msg recv, checkpoint-trigger rsp not handled", id);
return TSDB_CODE_INVALID_MSG; return TSDB_CODE_INVALID_MSG;
} }
if (pRsp->rspCode != TSDB_CODE_SUCCESS) { if (pRsp->rspCode != TSDB_CODE_SUCCESS) {
stDebug("s-task:%s retrieve checkpoint-trgger rsp from upstream:0x%x invalid, code:%s", pTask->id.idStr, stDebug("s-task:%s retrieve checkpoint-trgger rsp from upstream:0x%x invalid, code:%s", id, pRsp->upstreamTaskId,
pRsp->upstreamTaskId, tstrerror(pRsp->rspCode)); tstrerror(pRsp->rspCode));
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
streamMutexLock(&pTask->lock); streamMutexLock(&pTask->lock);
SStreamTaskState status = streamTaskGetStatus(pTask); SStreamTaskState status = streamTaskGetStatus(pTask);
streamMutexUnlock(&pTask->lock);
if (status.state != TASK_STATUS__CK) { if (status.state != TASK_STATUS__CK) {
stError("s-task:%s status:%s not in checkpoint status, discard the checkpoint-trigger msg", pTask->id.idStr, status.name); stError("s-task:%s status:%s not in checkpoint status, discard the checkpoint-trigger msg", id, status.name);
streamMutexUnlock(&pTask->lock);
return TSDB_CODE_STREAM_TASK_IVLD_STATUS; return TSDB_CODE_STREAM_TASK_IVLD_STATUS;
} }
streamMutexUnlock(&pTask->lock);
SActiveCheckpointInfo* pInfo = pTask->chkInfo.pActiveInfo;
streamMutexLock(&pInfo->lock); streamMutexLock(&pInfo->lock);
if (pInfo->activeId != pRsp->checkpointId || pInfo->transId != pRsp->transId) { unQualified = (pInfo->activeId != pRsp->checkpointId || pInfo->transId != pRsp->transId);
stError("s-task:%s status:%s not in checkpoint status, discard the checkpoint-trigger msg", pTask->id.idStr, status.name);
streamMutexUnlock(&pInfo->lock); streamMutexUnlock(&pInfo->lock);
if (unQualified) {
stError("s-task:%s status:%s not in checkpoint status, discard the checkpoint-trigger msg", id, status.name);
return TSDB_CODE_STREAM_TASK_IVLD_STATUS; return TSDB_CODE_STREAM_TASK_IVLD_STATUS;
} }
streamMutexUnlock(&pInfo->lock);
// NOTE: here we do not do the duplicated checkpoint-trigger msg check, since it will be done by following functions. // NOTE: here we do not do the duplicated checkpoint-trigger msg check, since it will be done by following functions.
int32_t code = appendCheckpointIntoInputQ(pTask, STREAM_INPUT__CHECKPOINT_TRIGGER, pRsp->checkpointId, pRsp->transId, int32_t code = appendCheckpointIntoInputQ(pTask, STREAM_INPUT__CHECKPOINT_TRIGGER, pRsp->checkpointId, pRsp->transId,
pRsp->upstreamTaskId); pRsp->upstreamTaskId);
@ -963,11 +964,44 @@ static int32_t doFindNotSendUpstream(SStreamTask* pTask, SArray* pList, SArray**
return 0; return 0;
} }
static int32_t chkptTriggerRecvMonitorHelper(SStreamTask* pTask, SArray* pNotSendList) {
const char* id = pTask->id.idStr;
SArray* pList = pTask->upstreamInfo.pList; // send msg to retrieve checkpoint trigger msg
SActiveCheckpointInfo* pActiveInfo = pTask->chkInfo.pActiveInfo;
SStreamTmrInfo* pTmrInfo = &pActiveInfo->chkptTriggerMsgTmr;
int32_t vgId = pTask->pMeta->vgId;
int32_t code = doChkptStatusCheck(pTask);
if (code) {
return code;
}
code = doFindNotSendUpstream(pTask, pList, &pNotSendList);
if (code) {
int32_t ref = streamCleanBeforeQuitTmr(pTmrInfo, pTask);
stDebug("s-task:%s failed to find not send upstream, code:%s, out of tmr, ref:%d", id, tstrerror(code), ref);
return code;
}
// do send retrieve checkpoint trigger msg to upstream
code = doSendRetrieveTriggerMsg(pTask, pNotSendList);
if (code) {
stError("s-task:%s vgId:%d failed to retrieve trigger msg, code:%s", pTask->id.idStr, vgId, tstrerror(code));
code = 0;
}
return code;
}
void checkpointTriggerMonitorFn(void* param, void* tmrId) { void checkpointTriggerMonitorFn(void* param, void* tmrId) {
SStreamTask* pTask = param; SStreamTask* pTask = param;
int32_t vgId = pTask->pMeta->vgId; int32_t vgId = pTask->pMeta->vgId;
int64_t now = taosGetTimestampMs(); int64_t now = taosGetTimestampMs();
const char* id = pTask->id.idStr; const char* id = pTask->id.idStr;
SArray* pNotSendList = NULL;
SArray* pList = pTask->upstreamInfo.pList; // send msg to retrieve checkpoint trigger msg
int32_t code = 0;
int32_t numOfNotSend = 0;
SActiveCheckpointInfo* pActiveInfo = pTask->chkInfo.pActiveInfo; SActiveCheckpointInfo* pActiveInfo = pTask->chkInfo.pActiveInfo;
SStreamTmrInfo* pTmrInfo = &pActiveInfo->chkptTriggerMsgTmr; SStreamTmrInfo* pTmrInfo = &pActiveInfo->chkptTriggerMsgTmr;
@ -1008,40 +1042,18 @@ void checkpointTriggerMonitorFn(void* param, void* tmrId) {
} }
streamMutexLock(&pActiveInfo->lock); streamMutexLock(&pActiveInfo->lock);
code = chkptTriggerRecvMonitorHelper(pTask, pNotSendList);
int32_t code = doChkptStatusCheck(pTask);
if (code) {
streamMutexUnlock(&pActiveInfo->lock); streamMutexUnlock(&pActiveInfo->lock);
if (code != TSDB_CODE_SUCCESS) {
streamMetaReleaseTask(pTask->pMeta, pTask); streamMetaReleaseTask(pTask->pMeta, pTask);
return;
}
// send msg to retrieve checkpoint trigger msg
SArray* pList = pTask->upstreamInfo.pList;
SArray* pNotSendList = NULL;
code = doFindNotSendUpstream(pTask, pList, &pNotSendList);
if (code) {
int32_t ref = streamCleanBeforeQuitTmr(pTmrInfo, pTask);
stDebug("s-task:%s failed to find not send upstream, code:%s, out of tmr, ref:%d", id, tstrerror(code), ref);
streamMutexUnlock(&pActiveInfo->lock);
streamMetaReleaseTask(pTask->pMeta, pTask);
taosArrayDestroy(pNotSendList); taosArrayDestroy(pNotSendList);
return; return;
} }
// do send retrieve checkpoint trigger msg to upstream
int32_t size = taosArrayGetSize(pNotSendList);
code = doSendRetrieveTriggerMsg(pTask, pNotSendList);
if (code) {
stError("s-task:%s vgId:%d failed to retrieve trigger msg, code:%s", pTask->id.idStr, vgId, tstrerror(code));
}
streamMutexUnlock(&pActiveInfo->lock);
// check every 100ms // check every 100ms
if (size > 0) { numOfNotSend = taosArrayGetSize(pNotSendList);
if (numOfNotSend > 0) {
stDebug("s-task:%s start to monitor checkpoint-trigger in 10s", id); stDebug("s-task:%s start to monitor checkpoint-trigger in 10s", id);
streamTmrStart(checkpointTriggerMonitorFn, 200, pTask, streamTimer, &pTmrInfo->tmrHandle, vgId, "trigger-recv-monitor"); streamTmrStart(checkpointTriggerMonitorFn, 200, pTask, streamTimer, &pTmrInfo->tmrHandle, vgId, "trigger-recv-monitor");
} else { } else {
@ -1106,19 +1118,13 @@ int32_t doSendRetrieveTriggerMsg(SStreamTask* pTask, SArray* pNotSendList) {
return code; return code;
} }
bool streamTaskAlreadySendTrigger(SStreamTask* pTask, int32_t downstreamNodeId) { static int32_t isAlreadySendTriggerNoLock(SStreamTask* pTask, int32_t downstreamNodeId) {
int64_t now = taosGetTimestampMs(); int64_t now = taosGetTimestampMs();
const char* id = pTask->id.idStr; const char* id = pTask->id.idStr;
SActiveCheckpointInfo* pInfo = pTask->chkInfo.pActiveInfo; SActiveCheckpointInfo* pInfo = pTask->chkInfo.pActiveInfo;
SStreamTaskState pStatus = streamTaskGetStatus(pTask); SStreamTaskState pStatus = streamTaskGetStatus(pTask);
if (pStatus.state != TASK_STATUS__CK) {
return false;
}
streamMutexLock(&pInfo->lock);
if (!pInfo->dispatchTrigger) { if (!pInfo->dispatchTrigger) {
streamMutexUnlock(&pInfo->lock);
return false; return false;
} }
@ -1146,14 +1152,29 @@ bool streamTaskAlreadySendTrigger(SStreamTask* pTask, int32_t downstreamNodeId)
id, pSendInfo->sendTs, before, pInfo->activeId, pInfo->transId); id, pSendInfo->sendTs, before, pInfo->activeId, pInfo->transId);
} }
streamMutexUnlock(&pInfo->lock);
return true; return true;
} }
streamMutexUnlock(&pInfo->lock);
return false; return false;
} }
bool streamTaskAlreadySendTrigger(SStreamTask* pTask, int32_t downstreamNodeId) {
int64_t now = taosGetTimestampMs();
const char* id = pTask->id.idStr;
SActiveCheckpointInfo* pInfo = pTask->chkInfo.pActiveInfo;
SStreamTaskState pStatus = streamTaskGetStatus(pTask);
if (pStatus.state != TASK_STATUS__CK) {
return false;
}
streamMutexLock(&pInfo->lock);
bool send = isAlreadySendTriggerNoLock(pTask, downstreamNodeId);
streamMutexUnlock(&pInfo->lock);
return send;
}
void streamTaskGetTriggerRecvStatus(SStreamTask* pTask, int32_t* pRecved, int32_t* pTotal) { void streamTaskGetTriggerRecvStatus(SStreamTask* pTask, int32_t* pRecved, int32_t* pTotal) {
*pRecved = taosArrayGetSize(pTask->chkInfo.pActiveInfo->pReadyMsgList); *pRecved = taosArrayGetSize(pTask->chkInfo.pActiveInfo->pReadyMsgList);
@ -1169,8 +1190,10 @@ void streamTaskGetTriggerRecvStatus(SStreamTask* pTask, int32_t* pRecved, int32_
int32_t streamTaskInitTriggerDispatchInfo(SStreamTask* pTask) { int32_t streamTaskInitTriggerDispatchInfo(SStreamTask* pTask) {
SActiveCheckpointInfo* pInfo = pTask->chkInfo.pActiveInfo; SActiveCheckpointInfo* pInfo = pTask->chkInfo.pActiveInfo;
int64_t now = taosGetTimestampMs(); int64_t now = taosGetTimestampMs();
int32_t code = 0;
streamMutexLock(&pInfo->lock); streamMutexLock(&pInfo->lock);
pInfo->dispatchTrigger = true; pInfo->dispatchTrigger = true;
if (pTask->outputInfo.type == TASK_OUTPUT__FIXED_DISPATCH) { if (pTask->outputInfo.type == TASK_OUTPUT__FIXED_DISPATCH) {
STaskDispatcherFixed* pDispatch = &pTask->outputInfo.fixedDispatcher; STaskDispatcherFixed* pDispatch = &pTask->outputInfo.fixedDispatcher;
@ -1178,8 +1201,7 @@ int32_t streamTaskInitTriggerDispatchInfo(SStreamTask* pTask) {
STaskTriggerSendInfo p = {.sendTs = now, .recved = false, .nodeId = pDispatch->nodeId, .taskId = pDispatch->taskId}; STaskTriggerSendInfo p = {.sendTs = now, .recved = false, .nodeId = pDispatch->nodeId, .taskId = pDispatch->taskId};
void* px = taosArrayPush(pInfo->pDispatchTriggerList, &p); void* px = taosArrayPush(pInfo->pDispatchTriggerList, &p);
if (px == NULL) { // pause the stream task, if memory not enough if (px == NULL) { // pause the stream task, if memory not enough
streamMutexUnlock(&pInfo->lock); code = terrno;
return terrno;
} }
} else { } else {
for (int32_t i = 0; i < streamTaskGetNumOfDownstream(pTask); ++i) { for (int32_t i = 0; i < streamTaskGetNumOfDownstream(pTask); ++i) {
@ -1191,14 +1213,15 @@ int32_t streamTaskInitTriggerDispatchInfo(SStreamTask* pTask) {
STaskTriggerSendInfo p = {.sendTs = now, .recved = false, .nodeId = pVgInfo->vgId, .taskId = pVgInfo->taskId}; STaskTriggerSendInfo p = {.sendTs = now, .recved = false, .nodeId = pVgInfo->vgId, .taskId = pVgInfo->taskId};
void* px = taosArrayPush(pInfo->pDispatchTriggerList, &p); void* px = taosArrayPush(pInfo->pDispatchTriggerList, &p);
if (px == NULL) { // pause the stream task, if memory not enough if (px == NULL) { // pause the stream task, if memory not enough
streamMutexUnlock(&pInfo->lock); code = terrno;
return terrno; break;
} }
} }
} }
streamMutexUnlock(&pInfo->lock); streamMutexUnlock(&pInfo->lock);
return 0;
return code;
} }
int32_t streamTaskGetNumOfConfirmed(SActiveCheckpointInfo* pInfo) { int32_t streamTaskGetNumOfConfirmed(SActiveCheckpointInfo* pInfo) {

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

@ -792,7 +792,7 @@ static int32_t doStreamExecTask(SStreamTask* pTask) {
// dispatch checkpoint msg to all downstream tasks // dispatch checkpoint msg to all downstream tasks
int32_t type = pInput->type; int32_t type = pInput->type;
if (type == STREAM_INPUT__CHECKPOINT_TRIGGER) { if (type == STREAM_INPUT__CHECKPOINT_TRIGGER) {
int32_t code = streamProcessCheckpointTriggerBlock(pTask, (SStreamDataBlock*)pInput); code = streamProcessCheckpointTriggerBlock(pTask, (SStreamDataBlock*)pInput);
if (code != 0) { if (code != 0) {
stError("s-task:%s failed to process checkpoint-trigger block, code:%s", pTask->id.idStr, tstrerror(code)); stError("s-task:%s failed to process checkpoint-trigger block, code:%s", pTask->id.idStr, tstrerror(code));
} }

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

@ -42,7 +42,7 @@ static bool waitForEnoughDuration(SMetaHbInfo* pInfo) {
static bool existInHbMsg(SStreamHbMsg* pMsg, SDownstreamTaskEpset* pTaskEpset) { static bool existInHbMsg(SStreamHbMsg* pMsg, SDownstreamTaskEpset* pTaskEpset) {
int32_t numOfExisted = taosArrayGetSize(pMsg->pUpdateNodes); int32_t numOfExisted = taosArrayGetSize(pMsg->pUpdateNodes);
for (int k = 0; k < numOfExisted; ++k) { for (int32_t k = 0; k < numOfExisted; ++k) {
if (pTaskEpset->nodeId == *(int32_t*)taosArrayGet(pMsg->pUpdateNodes, k)) { if (pTaskEpset->nodeId == *(int32_t*)taosArrayGet(pMsg->pUpdateNodes, k)) {
return true; return true;
} }
@ -56,7 +56,7 @@ static void addUpdateNodeIntoHbMsg(SStreamTask* pTask, SStreamHbMsg* pMsg) {
streamMutexLock(&pTask->lock); streamMutexLock(&pTask->lock);
int32_t num = taosArrayGetSize(pTask->outputInfo.pNodeEpsetUpdateList); int32_t num = taosArrayGetSize(pTask->outputInfo.pNodeEpsetUpdateList);
for (int j = 0; j < num; ++j) { for (int32_t j = 0; j < num; ++j) {
SDownstreamTaskEpset* pTaskEpset = taosArrayGet(pTask->outputInfo.pNodeEpsetUpdateList, j); SDownstreamTaskEpset* pTaskEpset = taosArrayGet(pTask->outputInfo.pNodeEpsetUpdateList, j);
bool exist = existInHbMsg(pMsg, pTaskEpset); bool exist = existInHbMsg(pMsg, pTaskEpset);

6
source/libs/wal/src/walMeta.c
View File

@ -796,7 +796,7 @@ int32_t walMetaSerialize(SWal* pWal, char** serialized) {
TAOS_RETURN(TSDB_CODE_OUT_OF_MEMORY); TAOS_RETURN(TSDB_CODE_OUT_OF_MEMORY);
} }
if (cJSON_AddItemToObject(pRoot, "meta", pMeta) != 0) { if (!cJSON_AddItemToObject(pRoot, "meta", pMeta)) {
wInfo("vgId:%d, failed to add meta to root", pWal->cfg.vgId); wInfo("vgId:%d, failed to add meta to root", pWal->cfg.vgId);
} }
(void)sprintf(buf, "%" PRId64, pWal->vers.firstVer); (void)sprintf(buf, "%" PRId64, pWal->vers.firstVer);
@ -816,13 +816,13 @@ int32_t walMetaSerialize(SWal* pWal, char** serialized) {
wInfo("vgId:%d, failed to add lastVer to meta", pWal->cfg.vgId); wInfo("vgId:%d, failed to add lastVer to meta", pWal->cfg.vgId);
} }
if (cJSON_AddItemToObject(pRoot, "files", pFiles) != 0) { if (!cJSON_AddItemToObject(pRoot, "files", pFiles)) {
wInfo("vgId:%d, failed to add files to root", pWal->cfg.vgId); wInfo("vgId:%d, failed to add files to root", pWal->cfg.vgId);
} }
SWalFileInfo* pData = pWal->fileInfoSet->pData; SWalFileInfo* pData = pWal->fileInfoSet->pData;
for (int i = 0; i < sz; i++) { for (int i = 0; i < sz; i++) {
SWalFileInfo* pInfo = &pData[i]; SWalFileInfo* pInfo = &pData[i];
if (cJSON_AddItemToArray(pFiles, pField = cJSON_CreateObject()) != 0) { if (!cJSON_AddItemToArray(pFiles, pField = cJSON_CreateObject())) {
wInfo("vgId:%d, failed to add field to files", pWal->cfg.vgId); wInfo("vgId:%d, failed to add field to files", pWal->cfg.vgId);
} }
if (pField == NULL) { if (pField == NULL) {

92
tests/docs-examples-test/c.sh Normal file
View File

@ -0,0 +1,92 @@
#!/bin/bash
pgrep taosd || taosd >> /dev/null 2>&1 &
pgrep taosadapter || taosadapter >> /dev/null 2>&1 &
GREEN='\033[0;32m'
RED='\033[0;31m'
NC='\033[0m'
TEST_PATH="../../docs/examples/c"
echo "setting TEST_PATH: $TEST_PATH"
cd "${TEST_PATH}" || { echo -e "${RED}Failed to change directory to ${TEST_PATH}${NC}"; exit 1; }
LOG_FILE="docs-c-test-out.log"
> $LOG_FILE
make > "$LOG_FILE" 2>&1
if [ $? -eq 0 ]; then
echo -e "${GREEN}Make completed successfully.${NC}"
else
echo -e "${RED}Make failed. Check log file: $LOG_FILE${NC}"
cat "$LOG_FILE"
exit 1
fi
declare -a TEST_EXES=(
"connect_example"
"create_db_demo"
"insert_data_demo"
"query_data_demo"
"with_reqid_demo"
"stmt_insert_demo"
"tmq_demo"
"sml_insert_demo"
)
declare -a NEED_CLEAN=(
"true"
"false"
"false"
"false"
"false"
"false"
"false"
"true"
)
totalCases=0
totalFailed=0
totalSuccess=0
for i in "${!TEST_EXES[@]}"; do
TEST_EXE="${TEST_EXES[$i]}"
NEED_CLEAN_FLAG="${NEED_CLEAN[$i]}"
if [ "$NEED_CLEAN_FLAG" = "true" ]; then
echo "Cleaning database before executing $TEST_EXE..."
taos -s "drop database if exists power" >> $LOG_FILE 2>&1
fi
echo "Executing $TEST_EXE..."
./$TEST_EXE >> $LOG_FILE 2>&1
RESULT=$?
if [ "$RESULT" -eq 0 ]; then
totalSuccess=$((totalSuccess + 1))
echo "[$GREEN OK $NC] $TEST_EXE executed successfully."
else
totalFailed=$((totalFailed + 1))
echo "[$RED FAILED $NC] $TEST_EXE exited with code $RESULT."
fi
totalCases=$((totalCases + 1))
done
tail -n 40 $LOG_FILE
echo -e "\nTotal number of cases executed: $totalCases"
if [ "$totalSuccess" -gt "0" ]; then
echo -e "\n${GREEN} ### Total $totalSuccess C case(s) succeed! ### ${NC}"
fi
if [ "$totalFailed" -ne "0" ]; then
echo -e "\n${RED} ### Total $totalFailed C case(s) failed! ### ${NC}"
exit 1
fi
echo "All tests completed."

1
tests/parallel_test/cases.task
View File

@ -1577,6 +1577,7 @@
,,n,develop-test,python3 ./test.py -f 5-taos-tools/taosbenchmark/telnet_tcp.py -R ,,n,develop-test,python3 ./test.py -f 5-taos-tools/taosbenchmark/telnet_tcp.py -R
#docs-examples test #docs-examples test
,,n,docs-examples-test,bash c.sh
,,n,docs-examples-test,bash python.sh ,,n,docs-examples-test,bash python.sh
,,n,docs-examples-test,bash node.sh ,,n,docs-examples-test,bash node.sh
,,n,docs-examples-test,bash csharp.sh ,,n,docs-examples-test,bash csharp.sh