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 enh/TD-30987-22

This commit is contained in:
Hongze Cheng 2024-07-29 14:08:06 +08:00
parent 6a31321b8b 2131d3240d
commit 542345bc4c
10 changed files with 116 additions and 117 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
source/dnode/mnode/impl/src/mndQnode.c
View File

@ -224,7 +224,7 @@ int32_t mndSetCreateQnodeRedoActions(STrans *pTrans, SDnodeObj *pDnode, SQnodeOb
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1; return -1;
} }
tSerializeSCreateDropMQSNodeReq(pReq, contLen, &createReq); (void)tSerializeSCreateDropMQSNodeReq(pReq, contLen, &createReq);
STransAction action = {0}; STransAction action = {0};
action.epSet = mndGetDnodeEpset(pDnode); action.epSet = mndGetDnodeEpset(pDnode);
@ -252,7 +252,7 @@ static int32_t mndSetCreateQnodeUndoActions(STrans *pTrans, SDnodeObj *pDnode, S
code = TSDB_CODE_OUT_OF_MEMORY; code = TSDB_CODE_OUT_OF_MEMORY;
TAOS_RETURN(code); TAOS_RETURN(code);
} }
tSerializeSCreateDropMQSNodeReq(pReq, contLen, &dropReq); (void)tSerializeSCreateDropMQSNodeReq(pReq, contLen, &dropReq);
STransAction action = {0}; STransAction action = {0};
action.epSet = mndGetDnodeEpset(pDnode); action.epSet = mndGetDnodeEpset(pDnode);
@ -330,7 +330,7 @@ static int32_t mndProcessCreateQnodeReq(SRpcMsg *pReq) {
if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS; if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;
char obj[33] = {0}; char obj[33] = {0};
sprintf(obj, "%d", createReq.dnodeId); (void)sprintf(obj, "%d", createReq.dnodeId);
auditRecord(pReq, pMnode->clusterId, "createQnode", "", obj, createReq.sql, createReq.sqlLen); auditRecord(pReq, pMnode->clusterId, "createQnode", "", obj, createReq.sql, createReq.sqlLen);
_OVER: _OVER:
@ -383,7 +383,7 @@ static int32_t mndSetDropQnodeRedoActions(STrans *pTrans, SDnodeObj *pDnode, SQn
code = TSDB_CODE_OUT_OF_MEMORY; code = TSDB_CODE_OUT_OF_MEMORY;
TAOS_RETURN(code); TAOS_RETURN(code);
} }
tSerializeSCreateDropMQSNodeReq(pReq, contLen, &dropReq); (void)tSerializeSCreateDropMQSNodeReq(pReq, contLen, &dropReq);
STransAction action = {0}; STransAction action = {0};
action.epSet = mndGetDnodeEpset(pDnode); action.epSet = mndGetDnodeEpset(pDnode);
@ -459,7 +459,7 @@ static int32_t mndProcessDropQnodeReq(SRpcMsg *pReq) {
if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS; if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;
char obj[33] = {0}; char obj[33] = {0};
sprintf(obj, "%d", dropReq.dnodeId); (void)sprintf(obj, "%d", dropReq.dnodeId);
auditRecord(pReq, pMnode->clusterId, "dropQnode", "", obj, dropReq.sql, dropReq.sqlLen); auditRecord(pReq, pMnode->clusterId, "dropQnode", "", obj, dropReq.sql, dropReq.sqlLen);
@ -531,7 +531,7 @@ static int32_t mndProcessQnodeListReq(SRpcMsg *pReq) {
goto _OVER; goto _OVER;
} }
tSerializeSQnodeListRsp(pRsp, rspLen, &qlistRsp); (void)tSerializeSQnodeListRsp(pRsp, rspLen, &qlistRsp);
pReq->info.rspLen = rspLen; pReq->info.rspLen = rspLen;
pReq->info.rsp = pRsp; pReq->info.rsp = pRsp;
@ -556,15 +556,15 @@ static int32_t mndRetrieveQnodes(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pB
cols = 0; cols = 0;
SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataSetVal(pColInfo, numOfRows, (const char *)&pObj->id, false); (void)colDataSetVal(pColInfo, numOfRows, (const char *)&pObj->id, false);
char ep[TSDB_EP_LEN + VARSTR_HEADER_SIZE] = {0}; char ep[TSDB_EP_LEN + VARSTR_HEADER_SIZE] = {0};
STR_WITH_MAXSIZE_TO_VARSTR(ep, pObj->pDnode->ep, pShow->pMeta->pSchemas[cols].bytes); STR_WITH_MAXSIZE_TO_VARSTR(ep, pObj->pDnode->ep, pShow->pMeta->pSchemas[cols].bytes);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataSetVal(pColInfo, numOfRows, (const char *)ep, false); (void)colDataSetVal(pColInfo, numOfRows, (const char *)ep, false);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataSetVal(pColInfo, numOfRows, (const char *)&pObj->createdTime, false); (void)colDataSetVal(pColInfo, numOfRows, (const char *)&pObj->createdTime, false);
numOfRows++; numOfRows++;
sdbRelease(pSdb, pObj); sdbRelease(pSdb, pObj);

7
source/dnode/mnode/impl/src/mndQuery.c
View File

@ -27,7 +27,7 @@ int32_t mndPreProcessQueryMsg(SRpcMsg *pMsg) {
void mndPostProcessQueryMsg(SRpcMsg *pMsg) { void mndPostProcessQueryMsg(SRpcMsg *pMsg) {
if (TDMT_SCH_QUERY != pMsg->msgType && TDMT_SCH_MERGE_QUERY != pMsg->msgType) return; if (TDMT_SCH_QUERY != pMsg->msgType && TDMT_SCH_MERGE_QUERY != pMsg->msgType) return;
SMnode *pMnode = pMsg->info.node; SMnode *pMnode = pMsg->info.node;
qWorkerAbortPreprocessQueryMsg(pMnode->pQuery, pMsg); (void)qWorkerAbortPreprocessQueryMsg(pMnode->pQuery, pMsg);
} }
int32_t mndProcessQueryMsg(SRpcMsg *pMsg, SQueueInfo* pInfo) { int32_t mndProcessQueryMsg(SRpcMsg *pMsg, SQueueInfo* pInfo) {
@ -134,7 +134,10 @@ int32_t mndProcessBatchMetaMsg(SRpcMsg *pMsg) {
rsp.msgLen = reqMsg.info.rspLen; rsp.msgLen = reqMsg.info.rspLen;
rsp.msg = reqMsg.info.rsp; rsp.msg = reqMsg.info.rsp;
taosArrayPush(batchRsp.pRsps, &rsp); if (taosArrayPush(batchRsp.pRsps, &rsp) == NULL) {
mError("msg:%p, failed to put array since %s, app:%p type:%s", pMsg, terrstr(), pMsg->info.ahandle,
TMSG_INFO(pMsg->msgType));
}
} }
rspSize = tSerializeSBatchRsp(NULL, 0, &batchRsp); rspSize = tSerializeSBatchRsp(NULL, 0, &batchRsp);

31
source/dnode/mnode/impl/src/mndScheduler.c
View File

@ -115,7 +115,7 @@ int32_t mndSetSinkTaskInfo(SStreamObj* pStream, SStreamTask* pTask) {
} else { } else {
pInfo->type = TASK_OUTPUT__TABLE; pInfo->type = TASK_OUTPUT__TABLE;
pInfo->tbSink.stbUid = pStream->targetStbUid; pInfo->tbSink.stbUid = pStream->targetStbUid;
memcpy(pInfo->tbSink.stbFullName, pStream->targetSTbName, TSDB_TABLE_FNAME_LEN); (void)memcpy(pInfo->tbSink.stbFullName, pStream->targetSTbName, TSDB_TABLE_FNAME_LEN);
pInfo->tbSink.pSchemaWrapper = tCloneSSchemaWrapper(&pStream->outputSchema); pInfo->tbSink.pSchemaWrapper = tCloneSSchemaWrapper(&pStream->outputSchema);
if (pInfo->tbSink.pSchemaWrapper == NULL) { if (pInfo->tbSink.pSchemaWrapper == NULL) {
return TSDB_CODE_OUT_OF_MEMORY; return TSDB_CODE_OUT_OF_MEMORY;
@ -145,7 +145,7 @@ int32_t mndAddDispatcherForInternalTask(SMnode* pMnode, SStreamObj* pStream, SAr
int32_t numOfSinkNodes = taosArrayGetSize(pSinkNodeList); int32_t numOfSinkNodes = taosArrayGetSize(pSinkNodeList);
if (isShuffle) { if (isShuffle) {
memcpy(pTask->outputInfo.shuffleDispatcher.stbFullName, pStream->targetSTbName, TSDB_TABLE_FNAME_LEN); (void)memcpy(pTask->outputInfo.shuffleDispatcher.stbFullName, pStream->targetSTbName, TSDB_TABLE_FNAME_LEN);
SArray* pVgs = pTask->outputInfo.shuffleDispatcher.dbInfo.pVgroupInfos; SArray* pVgs = pTask->outputInfo.shuffleDispatcher.dbInfo.pVgroupInfos;
int32_t numOfVgroups = taosArrayGetSize(pVgs); int32_t numOfVgroups = taosArrayGetSize(pVgs);
@ -363,10 +363,14 @@ static int32_t buildSourceTask(SStreamObj* pStream, SEpSet* pEpset, bool isFillh
static void addNewTaskList(SStreamObj* pStream) { static void addNewTaskList(SStreamObj* pStream) {
SArray* pTaskList = taosArrayInit(0, POINTER_BYTES); SArray* pTaskList = taosArrayInit(0, POINTER_BYTES);
taosArrayPush(pStream->tasks, &pTaskList); if (taosArrayPush(pStream->tasks, &pTaskList) == NULL) {
mError("failed to put array");
}
if (pStream->conf.fillHistory) { if (pStream->conf.fillHistory) {
pTaskList = taosArrayInit(0, POINTER_BYTES); pTaskList = taosArrayInit(0, POINTER_BYTES);
taosArrayPush(pStream->pHTasksList, &pTaskList); if (taosArrayPush(pStream->pHTasksList, &pTaskList) == NULL) {
mError("failed to put array");
}
} }
} }
@ -584,10 +588,15 @@ static int32_t addSinkTask(SMnode* pMnode, SStreamObj* pStream, SEpSet* pEpset)
} }
static void bindTaskToSinkTask(SStreamObj* pStream, SMnode* pMnode, SArray* pSinkTaskList, SStreamTask* task) { static void bindTaskToSinkTask(SStreamObj* pStream, SMnode* pMnode, SArray* pSinkTaskList, SStreamTask* task) {
mndAddDispatcherForInternalTask(pMnode, pStream, pSinkTaskList, task); int32_t code = 0;
if ((code = mndAddDispatcherForInternalTask(pMnode, pStream, pSinkTaskList, task)) != 0) {
mError("failed bind task to sink task since %s", tstrerror(code));
}
for (int32_t k = 0; k < taosArrayGetSize(pSinkTaskList); k++) { for (int32_t k = 0; k < taosArrayGetSize(pSinkTaskList); k++) {
SStreamTask* pSinkTask = taosArrayGetP(pSinkTaskList, k); SStreamTask* pSinkTask = taosArrayGetP(pSinkTaskList, k);
streamTaskSetUpstreamInfo(pSinkTask, task); if ((code = streamTaskSetUpstreamInfo(pSinkTask, task)) != 0) {
mError("failed bind task to sink task since %s", tstrerror(code));
}
} }
mDebug("bindTaskToSinkTask taskId:%s to sink task list", task->id.idStr); mDebug("bindTaskToSinkTask taskId:%s to sink task list", task->id.idStr);
} }
@ -604,6 +613,7 @@ static void bindAggSink(SStreamObj* pStream, SMnode* pMnode, SArray* tasks) {
} }
static void bindSourceSink(SStreamObj* pStream, SMnode* pMnode, SArray* tasks, bool hasExtraSink) { static void bindSourceSink(SStreamObj* pStream, SMnode* pMnode, SArray* tasks, bool hasExtraSink) {
int32_t code = 0;
SArray* pSinkTaskList = taosArrayGetP(tasks, SINK_NODE_LEVEL); SArray* pSinkTaskList = taosArrayGetP(tasks, SINK_NODE_LEVEL);
SArray* pSourceTaskList = taosArrayGetP(tasks, hasExtraSink ? SINK_NODE_LEVEL + 1 : SINK_NODE_LEVEL); SArray* pSourceTaskList = taosArrayGetP(tasks, hasExtraSink ? SINK_NODE_LEVEL + 1 : SINK_NODE_LEVEL);
@ -614,12 +624,15 @@ static void bindSourceSink(SStreamObj* pStream, SMnode* pMnode, SArray* tasks, b
if (hasExtraSink) { if (hasExtraSink) {
bindTaskToSinkTask(pStream, pMnode, pSinkTaskList, pSourceTask); bindTaskToSinkTask(pStream, pMnode, pSinkTaskList, pSourceTask);
} else { } else {
mndSetSinkTaskInfo(pStream, pSourceTask); if ((code = mndSetSinkTaskInfo(pStream, pSourceTask)) != 0) {
mError("failed bind task to sink task since %s", tstrerror(code));
}
} }
} }
} }
static void bindTwoLevel(SArray* tasks, int32_t begin, int32_t end) { static void bindTwoLevel(SArray* tasks, int32_t begin, int32_t end) {
int32_t code = 0;
size_t size = taosArrayGetSize(tasks); size_t size = taosArrayGetSize(tasks);
ASSERT(size >= 2); ASSERT(size >= 2);
SArray* pDownTaskList = taosArrayGetP(tasks, size - 1); SArray* pDownTaskList = taosArrayGetP(tasks, size - 1);
@ -631,7 +644,9 @@ static void bindTwoLevel(SArray* tasks, int32_t begin, int32_t end) {
SStreamTask* pUpTask = taosArrayGetP(pUpTaskList, i); SStreamTask* pUpTask = taosArrayGetP(pUpTaskList, i);
pUpTask->info.selfChildId = i - begin; pUpTask->info.selfChildId = i - begin;
streamTaskSetFixedDownstreamInfo(pUpTask, *pDownTask); streamTaskSetFixedDownstreamInfo(pUpTask, *pDownTask);
streamTaskSetUpstreamInfo(*pDownTask, pUpTask); if ((code = streamTaskSetUpstreamInfo(*pDownTask, pUpTask)) != 0) {
mError("failed bind task to sink task since %s", tstrerror(code));
}
} }
mDebug("bindTwoLevel task list(%d-%d) to taskId:%s", begin, end - 1, (*(pDownTask))->id.idStr); mDebug("bindTwoLevel task list(%d-%d) to taskId:%s", begin, end - 1, (*(pDownTask))->id.idStr);
} }

10
source/dnode/mnode/impl/src/mndShow.c
View File

@ -158,7 +158,7 @@ static SShowObj *mndCreateShowObj(SMnode *pMnode, SRetrieveTableReq *pReq) {
showObj.id = showId; showObj.id = showId;
showObj.pMnode = pMnode; showObj.pMnode = pMnode;
showObj.type = convertToRetrieveType(pReq->tb, tListLen(pReq->tb)); showObj.type = convertToRetrieveType(pReq->tb, tListLen(pReq->tb));
memcpy(showObj.db, pReq->db, TSDB_DB_FNAME_LEN); (void)memcpy(showObj.db, pReq->db, TSDB_DB_FNAME_LEN);
tstrncpy(showObj.filterTb, pReq->filterTb, TSDB_TABLE_NAME_LEN); tstrncpy(showObj.filterTb, pReq->filterTb, TSDB_TABLE_NAME_LEN);
int32_t keepTime = tsShellActivityTimer * 6 * 1000; int32_t keepTime = tsShellActivityTimer * 6 * 1000;
@ -270,9 +270,9 @@ static int32_t mndProcessRetrieveSysTableReq(SRpcMsg *pReq) {
mDebug("show:0x%" PRIx64 ", start retrieve data, type:%d", pShow->id, pShow->type); mDebug("show:0x%" PRIx64 ", start retrieve data, type:%d", pShow->id, pShow->type);
if (retrieveReq.user[0] != 0) { if (retrieveReq.user[0] != 0) {
memcpy(pReq->info.conn.user, retrieveReq.user, TSDB_USER_LEN); (void)memcpy(pReq->info.conn.user, retrieveReq.user, TSDB_USER_LEN);
} else { } else {
memcpy(pReq->info.conn.user, TSDB_DEFAULT_USER, strlen(TSDB_DEFAULT_USER) + 1); (void)memcpy(pReq->info.conn.user, TSDB_DEFAULT_USER, strlen(TSDB_DEFAULT_USER) + 1);
} }
code = -1; code = -1;
if (retrieveReq.db[0] && if (retrieveReq.db[0] &&
@ -303,10 +303,10 @@ static int32_t mndProcessRetrieveSysTableReq(SRpcMsg *pReq) {
idata.info.bytes = p->bytes; idata.info.bytes = p->bytes;
idata.info.type = p->type; idata.info.type = p->type;
idata.info.colId = p->colId; idata.info.colId = p->colId;
blockDataAppendColInfo(pBlock, &idata); TAOS_CHECK_RETURN(blockDataAppendColInfo(pBlock, &idata));
} }
blockDataEnsureCapacity(pBlock, rowsToRead); TAOS_CHECK_RETURN(blockDataEnsureCapacity(pBlock, rowsToRead));
if (mndCheckRetrieveFinished(pShow)) { if (mndCheckRetrieveFinished(pShow)) {
mDebug("show:0x%" PRIx64 ", read finished, numOfRows:%d", pShow->id, pShow->numOfRows); mDebug("show:0x%" PRIx64 ", read finished, numOfRows:%d", pShow->id, pShow->numOfRows);

5
source/dnode/vnode/src/inc/tsdb.h
View File

@ -378,7 +378,7 @@ struct STsdb {
struct { struct {
SVHashTable *ht; SVHashTable *ht;
SArray *arr; SArray *arr;
} *commitInfo; } * commitInfo;
}; };
struct TSDBKEY { struct TSDBKEY {
@ -937,7 +937,7 @@ int32_t tsdbCacheDel(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, TSKEY sKey, TSKE
int32_t tsdbCacheInsertLast(SLRUCache *pCache, tb_uid_t uid, TSDBROW *row, STsdb *pTsdb); int32_t tsdbCacheInsertLast(SLRUCache *pCache, tb_uid_t uid, TSDBROW *row, STsdb *pTsdb);
int32_t tsdbCacheInsertLastrow(SLRUCache *pCache, STsdb *pTsdb, tb_uid_t uid, TSDBROW *row, bool dup); int32_t tsdbCacheInsertLastrow(SLRUCache *pCache, STsdb *pTsdb, tb_uid_t uid, TSDBROW *row, bool dup);
int32_t tsdbCacheRelease(SLRUCache *pCache, LRUHandle *h); void tsdbCacheRelease(SLRUCache *pCache, LRUHandle *h);
int32_t tsdbCacheGetBlockIdx(SLRUCache *pCache, SDataFReader *pFileReader, LRUHandle **handle); int32_t tsdbCacheGetBlockIdx(SLRUCache *pCache, SDataFReader *pFileReader, LRUHandle **handle);
int32_t tsdbBICacheRelease(SLRUCache *pCache, LRUHandle *h); int32_t tsdbBICacheRelease(SLRUCache *pCache, LRUHandle *h);
@ -945,7 +945,6 @@ int32_t tsdbBICacheRelease(SLRUCache *pCache, LRUHandle *h);
int32_t tsdbCacheGetBlockS3(SLRUCache *pCache, STsdbFD *pFD, LRUHandle **handle); int32_t tsdbCacheGetBlockS3(SLRUCache *pCache, STsdbFD *pFD, LRUHandle **handle);
int32_t tsdbCacheGetPageS3(SLRUCache *pCache, STsdbFD *pFD, int64_t pgno, LRUHandle **handle); int32_t tsdbCacheGetPageS3(SLRUCache *pCache, STsdbFD *pFD, int64_t pgno, LRUHandle **handle);
int32_t tsdbCacheSetPageS3(SLRUCache *pCache, STsdbFD *pFD, int64_t pgno, uint8_t *pPage); int32_t tsdbCacheSetPageS3(SLRUCache *pCache, STsdbFD *pFD, int64_t pgno, uint8_t *pPage);
int32_t tsdbCacheRelease(SLRUCache *pCache, LRUHandle *h);
int32_t tsdbCacheDeleteLastrow(SLRUCache *pCache, tb_uid_t uid, TSKEY eKey); int32_t tsdbCacheDeleteLastrow(SLRUCache *pCache, tb_uid_t uid, TSKEY eKey);
int32_t tsdbCacheDeleteLast(SLRUCache *pCache, tb_uid_t uid, TSKEY eKey); int32_t tsdbCacheDeleteLast(SLRUCache *pCache, tb_uid_t uid, TSKEY eKey);

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

@ -117,7 +117,7 @@ typedef struct {
static void tsdbGetRocksPath(STsdb *pTsdb, char *path) { static void tsdbGetRocksPath(STsdb *pTsdb, char *path) {
SVnode *pVnode = pTsdb->pVnode; SVnode *pVnode = pTsdb->pVnode;
vnodeGetPrimaryDir(pTsdb->path, pVnode->diskPrimary, pVnode->pTfs, path, TSDB_FILENAME_LEN); (void)vnodeGetPrimaryDir(pTsdb->path, pVnode->diskPrimary, pVnode->pTfs, path, TSDB_FILENAME_LEN);
int32_t offset = strlen(path); int32_t offset = strlen(path);
snprintf(path + offset, TSDB_FILENAME_LEN - offset - 1, "%scache.rdb", TD_DIRSEP); snprintf(path + offset, TSDB_FILENAME_LEN - offset - 1, "%scache.rdb", TD_DIRSEP);
@ -722,20 +722,14 @@ static int32_t tsdbCacheDropTableColumn(STsdb *pTsdb, int64_t uid, int16_t cid,
rocksdb_writebatch_t *wb = pTsdb->rCache.writebatch; rocksdb_writebatch_t *wb = pTsdb->rCache.writebatch;
{ {
SLastCol *pLastCol = NULL; SLastCol *pLastCol = NULL;
code = tsdbCacheDeserialize(values_list[0], values_list_sizes[0], &pLastCol); (void)tsdbCacheDeserialize(values_list[0], values_list_sizes[0], &pLastCol);
if (code) {
tsdbError("tsdb/cache: vgId:%d, deserialize failed since %s.", TD_VID(pTsdb->pVnode), tstrerror(code));
}
if (NULL != pLastCol) { if (NULL != pLastCol) {
rocksdb_writebatch_delete(wb, keys_list[0], klen); rocksdb_writebatch_delete(wb, keys_list[0], klen);
} }
taosMemoryFreeClear(pLastCol); taosMemoryFreeClear(pLastCol);
pLastCol = NULL; pLastCol = NULL;
code = tsdbCacheDeserialize(values_list[1], values_list_sizes[1], &pLastCol); (void)tsdbCacheDeserialize(values_list[1], values_list_sizes[1], &pLastCol);
if (code) {
tsdbError("tsdb/cache: vgId:%d, deserialize failed since %s.", TD_VID(pTsdb->pVnode), tstrerror(code));
}
if (NULL != pLastCol) { if (NULL != pLastCol) {
rocksdb_writebatch_delete(wb, keys_list[1], klen); rocksdb_writebatch_delete(wb, keys_list[1], klen);
} }
@ -748,7 +742,7 @@ static int32_t tsdbCacheDropTableColumn(STsdb *pTsdb, int64_t uid, int16_t cid,
LRUHandle *h = taosLRUCacheLookup(pTsdb->lruCache, keys_list[0], klen); LRUHandle *h = taosLRUCacheLookup(pTsdb->lruCache, keys_list[0], klen);
if (h) { if (h) {
erase = true; erase = true;
taosLRUCacheRelease(pTsdb->lruCache, h, erase); (void)taosLRUCacheRelease(pTsdb->lruCache, h, erase);
} }
if (erase) { if (erase) {
taosLRUCacheErase(pTsdb->lruCache, keys_list[0], klen); taosLRUCacheErase(pTsdb->lruCache, keys_list[0], klen);
@ -758,7 +752,7 @@ static int32_t tsdbCacheDropTableColumn(STsdb *pTsdb, int64_t uid, int16_t cid,
h = taosLRUCacheLookup(pTsdb->lruCache, keys_list[1], klen); h = taosLRUCacheLookup(pTsdb->lruCache, keys_list[1], klen);
if (h) { if (h) {
erase = true; erase = true;
taosLRUCacheRelease(pTsdb->lruCache, h, erase); (void)taosLRUCacheRelease(pTsdb->lruCache, h, erase);
} }
if (erase) { if (erase) {
taosLRUCacheErase(pTsdb->lruCache, keys_list[1], klen); taosLRUCacheErase(pTsdb->lruCache, keys_list[1], klen);
@ -1066,12 +1060,12 @@ static int32_t tsdbCacheUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, SArray
if (cmp_res < 0 || (cmp_res == 0 && !COL_VAL_IS_NONE(pColVal))) { if (cmp_res < 0 || (cmp_res == 0 && !COL_VAL_IS_NONE(pColVal))) {
tsdbCacheUpdateLastCol(pLastCol, pRowKey, pColVal); tsdbCacheUpdateLastCol(pLastCol, pRowKey, pColVal);
} }
taosLRUCacheRelease(pCache, h, false); (void)taosLRUCacheRelease(pCache, h, false);
} else { } else {
if (!remainCols) { if (!remainCols) {
remainCols = taosArrayInit(num_keys * 2, sizeof(SIdxKey)); remainCols = taosArrayInit(num_keys * 2, sizeof(SIdxKey));
} }
taosArrayPush(remainCols, &(SIdxKey){i, *key}); (void)taosArrayPush(remainCols, &(SIdxKey){i, *key});
} }
} }
@ -1123,10 +1117,12 @@ static int32_t tsdbCacheUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, SArray
SColVal *pColVal = &updCtx->colVal; SColVal *pColVal = &updCtx->colVal;
SLastCol *pLastCol = NULL; SLastCol *pLastCol = NULL;
code = tsdbCacheDeserialize(values_list[i], values_list_sizes[i], &pLastCol); (void)tsdbCacheDeserialize(values_list[i], values_list_sizes[i], &pLastCol);
/*
if (code) { if (code) {
tsdbError("tsdb/cache: vgId:%d, deserialize failed since %s.", TD_VID(pTsdb->pVnode), tstrerror(code)); tsdbError("tsdb/cache: vgId:%d, deserialize failed since %s.", TD_VID(pTsdb->pVnode), tstrerror(code));
} }
*/
SLastCol *PToFree = pLastCol; SLastCol *PToFree = pLastCol;
if (IS_LAST_KEY(idxKey->key) && !COL_VAL_IS_VALUE(pColVal)) { if (IS_LAST_KEY(idxKey->key) && !COL_VAL_IS_VALUE(pColVal)) {
@ -1242,18 +1238,18 @@ int32_t tsdbCacheRowFormatUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, int6
tsdbRowGetKey(&lRow, &tsdbRowKey); tsdbRowGetKey(&lRow, &tsdbRowKey);
STSDBRowIter iter = {0}; STSDBRowIter iter = {0};
tsdbRowIterOpen(&iter, &lRow, pTSchema); (void)tsdbRowIterOpen(&iter, &lRow, pTSchema);
int32_t iCol = 0; int32_t iCol = 0;
for (SColVal *pColVal = tsdbRowIterNext(&iter); pColVal && iCol < nCol; pColVal = tsdbRowIterNext(&iter), iCol++) { for (SColVal *pColVal = tsdbRowIterNext(&iter); pColVal && iCol < nCol; pColVal = tsdbRowIterNext(&iter), iCol++) {
SLastUpdateCtx updateCtx = {.lflag = LFLAG_LAST_ROW, .tsdbRowKey = tsdbRowKey, .colVal = *pColVal}; SLastUpdateCtx updateCtx = {.lflag = LFLAG_LAST_ROW, .tsdbRowKey = tsdbRowKey, .colVal = *pColVal};
taosArrayPush(ctxArray, &updateCtx); (void)taosArrayPush(ctxArray, &updateCtx);
if (!COL_VAL_IS_VALUE(pColVal)) { if (!COL_VAL_IS_VALUE(pColVal)) {
tSimpleHashPut(iColHash, &iCol, sizeof(iCol), NULL, 0); (void)tSimpleHashPut(iColHash, &iCol, sizeof(iCol), NULL, 0);
continue; continue;
} }
updateCtx.lflag = LFLAG_LAST; updateCtx.lflag = LFLAG_LAST;
taosArrayPush(ctxArray, &updateCtx); (void)taosArrayPush(ctxArray, &updateCtx);
} }
tsdbRowClose(&iter); tsdbRowClose(&iter);
@ -1277,14 +1273,14 @@ int32_t tsdbCacheRowFormatUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, int6
if (COL_VAL_IS_VALUE(&colVal)) { if (COL_VAL_IS_VALUE(&colVal)) {
SLastUpdateCtx updateCtx = {.lflag = LFLAG_LAST, .tsdbRowKey = tsdbRowKey, .colVal = colVal}; SLastUpdateCtx updateCtx = {.lflag = LFLAG_LAST, .tsdbRowKey = tsdbRowKey, .colVal = colVal};
taosArrayPush(ctxArray, &updateCtx); (void)taosArrayPush(ctxArray, &updateCtx);
tSimpleHashIterateRemove(iColHash, &iCol, sizeof(iCol), &pIte, &iter); (void)tSimpleHashIterateRemove(iColHash, &iCol, sizeof(iCol), &pIte, &iter);
} }
} }
} }
// 3. do update // 3. do update
tsdbCacheUpdate(pTsdb, suid, uid, ctxArray); (void)tsdbCacheUpdate(pTsdb, suid, uid, ctxArray);
_exit: _exit:
taosMemoryFreeClear(pTSchema); taosMemoryFreeClear(pTSchema);
@ -1317,7 +1313,7 @@ int32_t tsdbCacheColFormatUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, SBlo
.tsdbRowKey = tsdbRowKey, .tsdbRowKey = tsdbRowKey,
.colVal = COL_VAL_VALUE(PRIMARYKEY_TIMESTAMP_COL_ID, ((SValue){.type = TSDB_DATA_TYPE_TIMESTAMP, .colVal = COL_VAL_VALUE(PRIMARYKEY_TIMESTAMP_COL_ID, ((SValue){.type = TSDB_DATA_TYPE_TIMESTAMP,
.val = lRow.pBlockData->aTSKEY[lRow.iRow]}))}; .val = lRow.pBlockData->aTSKEY[lRow.iRow]}))};
taosArrayPush(ctxArray, &updateCtx); (void)taosArrayPush(ctxArray, &updateCtx);
} }
TSDBROW tRow = tsdbRowFromBlockData(pBlockData, 0); TSDBROW tRow = tsdbRowFromBlockData(pBlockData, 0);
@ -1338,7 +1334,7 @@ int32_t tsdbCacheColFormatUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, SBlo
tColDataGetValue(pColData, tRow.iRow, &colVal); tColDataGetValue(pColData, tRow.iRow, &colVal);
SLastUpdateCtx updateCtx = {.lflag = LFLAG_LAST, .tsdbRowKey = tsdbRowKey, .colVal = colVal}; SLastUpdateCtx updateCtx = {.lflag = LFLAG_LAST, .tsdbRowKey = tsdbRowKey, .colVal = colVal};
taosArrayPush(ctxArray, &updateCtx); (void)taosArrayPush(ctxArray, &updateCtx);
break; break;
} }
} }
@ -1346,15 +1342,15 @@ int32_t tsdbCacheColFormatUpdate(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, SBlo
// 2. prepare last row // 2. prepare last row
STSDBRowIter iter = {0}; STSDBRowIter iter = {0};
tsdbRowIterOpen(&iter, &lRow, pTSchema); (void)tsdbRowIterOpen(&iter, &lRow, pTSchema);
for (SColVal *pColVal = tsdbRowIterNext(&iter); pColVal; pColVal = tsdbRowIterNext(&iter)) { for (SColVal *pColVal = tsdbRowIterNext(&iter); pColVal; pColVal = tsdbRowIterNext(&iter)) {
SLastUpdateCtx updateCtx = {.lflag = LFLAG_LAST_ROW, .tsdbRowKey = tsdbRowKey, .colVal = *pColVal}; SLastUpdateCtx updateCtx = {.lflag = LFLAG_LAST_ROW, .tsdbRowKey = tsdbRowKey, .colVal = *pColVal};
taosArrayPush(ctxArray, &updateCtx); (void)taosArrayPush(ctxArray, &updateCtx);
} }
tsdbRowClose(&iter); tsdbRowClose(&iter);
// 3. do update // 3. do update
tsdbCacheUpdate(pTsdb, suid, uid, ctxArray); (void)tsdbCacheUpdate(pTsdb, suid, uid, ctxArray);
_exit: _exit:
taosMemoryFreeClear(pTSchema); taosMemoryFreeClear(pTSchema);
@ -1379,7 +1375,7 @@ static int32_t tsdbCacheLoadFromRaw(STsdb *pTsdb, tb_uid_t uid, SArray *pLastArr
if (idxKey->key.cid != PRIMARYKEY_TIMESTAMP_COL_ID) { if (idxKey->key.cid != PRIMARYKEY_TIMESTAMP_COL_ID) {
SLastKey *key = &(SLastKey){.lflag = ltype, .uid = uid, .cid = PRIMARYKEY_TIMESTAMP_COL_ID}; SLastKey *key = &(SLastKey){.lflag = ltype, .uid = uid, .cid = PRIMARYKEY_TIMESTAMP_COL_ID};
taosArrayInsert(remainCols, 0, &(SIdxKey){0, *key}); (void)taosArrayInsert(remainCols, 0, &(SIdxKey){0, *key});
} }
int num_keys = TARRAY_SIZE(remainCols); int num_keys = TARRAY_SIZE(remainCols);
@ -1416,7 +1412,7 @@ static int32_t tsdbCacheLoadFromRaw(STsdb *pTsdb, tb_uid_t uid, SArray *pLastArr
if (NULL == lastTmpIndexArray) { if (NULL == lastTmpIndexArray) {
lastTmpIndexArray = taosArrayInit(num_keys, sizeof(int32_t)); lastTmpIndexArray = taosArrayInit(num_keys, sizeof(int32_t));
} }
taosArrayPush(lastTmpIndexArray, &(i)); (void)taosArrayPush(lastTmpIndexArray, &(i));
lastColIds[lastIndex] = idxKey->key.cid; lastColIds[lastIndex] = idxKey->key.cid;
lastSlotIds[lastIndex] = pr->pSlotIds[idxKey->idx]; lastSlotIds[lastIndex] = pr->pSlotIds[idxKey->idx];
lastIndex++; lastIndex++;
@ -1424,7 +1420,7 @@ static int32_t tsdbCacheLoadFromRaw(STsdb *pTsdb, tb_uid_t uid, SArray *pLastArr
if (NULL == lastrowTmpIndexArray) { if (NULL == lastrowTmpIndexArray) {
lastrowTmpIndexArray = taosArrayInit(num_keys, sizeof(int32_t)); lastrowTmpIndexArray = taosArrayInit(num_keys, sizeof(int32_t));
} }
taosArrayPush(lastrowTmpIndexArray, &(i)); (void)taosArrayPush(lastrowTmpIndexArray, &(i));
lastrowColIds[lastrowIndex] = idxKey->key.cid; lastrowColIds[lastrowIndex] = idxKey->key.cid;
lastrowSlotIds[lastrowIndex] = pr->pSlotIds[idxKey->idx]; lastrowSlotIds[lastrowIndex] = pr->pSlotIds[idxKey->idx];
lastrowIndex++; lastrowIndex++;
@ -1434,16 +1430,17 @@ static int32_t tsdbCacheLoadFromRaw(STsdb *pTsdb, tb_uid_t uid, SArray *pLastArr
pTmpColArray = taosArrayInit(lastIndex + lastrowIndex, sizeof(SLastCol)); pTmpColArray = taosArrayInit(lastIndex + lastrowIndex, sizeof(SLastCol));
if (lastTmpIndexArray != NULL) { if (lastTmpIndexArray != NULL) {
mergeLastCid(uid, pTsdb, &lastTmpColArray, pr, lastColIds, lastIndex, lastSlotIds); (void)mergeLastCid(uid, pTsdb, &lastTmpColArray, pr, lastColIds, lastIndex, lastSlotIds);
for (int i = 0; i < taosArrayGetSize(lastTmpColArray); i++) { for (int i = 0; i < taosArrayGetSize(lastTmpColArray); i++) {
taosArrayInsert(pTmpColArray, *(int32_t *)taosArrayGet(lastTmpIndexArray, i), taosArrayGet(lastTmpColArray, i)); (void)taosArrayInsert(pTmpColArray, *(int32_t *)taosArrayGet(lastTmpIndexArray, i),
taosArrayGet(lastTmpColArray, i));
} }
} }
if (lastrowTmpIndexArray != NULL) { if (lastrowTmpIndexArray != NULL) {
mergeLastRowCid(uid, pTsdb, &lastrowTmpColArray, pr, lastrowColIds, lastrowIndex, lastrowSlotIds); (void)mergeLastRowCid(uid, pTsdb, &lastrowTmpColArray, pr, lastrowColIds, lastrowIndex, lastrowSlotIds);
for (int i = 0; i < taosArrayGetSize(lastrowTmpColArray); i++) { for (int i = 0; i < taosArrayGetSize(lastrowTmpColArray); i++) {
taosArrayInsert(pTmpColArray, *(int32_t *)taosArrayGet(lastrowTmpIndexArray, i), (void)taosArrayInsert(pTmpColArray, *(int32_t *)taosArrayGet(lastrowTmpIndexArray, i),
taosArrayGet(lastrowTmpColArray, i)); taosArrayGet(lastrowTmpColArray, i));
} }
} }
@ -1586,10 +1583,7 @@ static int32_t tsdbCacheLoadFromRocks(STsdb *pTsdb, tb_uid_t uid, SArray *pLastA
SLRUCache *pCache = pTsdb->lruCache; SLRUCache *pCache = pTsdb->lruCache;
for (int i = 0, j = 0; i < num_keys && j < TARRAY_SIZE(remainCols); ++i) { for (int i = 0, j = 0; i < num_keys && j < TARRAY_SIZE(remainCols); ++i) {
SLastCol *pLastCol = NULL; SLastCol *pLastCol = NULL;
code = tsdbCacheDeserialize(values_list[i], values_list_sizes[i], &pLastCol); (void)tsdbCacheDeserialize(values_list[i], values_list_sizes[i], &pLastCol);
if (code) {
tsdbError("tsdb/cache: vgId:%d, deserialize failed since %s.", TD_VID(pTsdb->pVnode), tstrerror(code));
}
SLastCol *PToFree = pLastCol; SLastCol *PToFree = pLastCol;
SIdxKey *idxKey = &((SIdxKey *)TARRAY_DATA(remainCols))[j]; SIdxKey *idxKey = &((SIdxKey *)TARRAY_DATA(remainCols))[j];
if (pLastCol) { if (pLastCol) {
@ -1682,19 +1676,19 @@ int32_t tsdbCacheGetBatch(STsdb *pTsdb, tb_uid_t uid, SArray *pLastArray, SCache
TAOS_CHECK_RETURN(reallocVarDataVal(&lastCol.rowKey.pks[j])); TAOS_CHECK_RETURN(reallocVarDataVal(&lastCol.rowKey.pks[j]));
} }
TAOS_CHECK_RETURN(reallocVarData(&lastCol.colVal)); TAOS_CHECK_RETURN(reallocVarData(&lastCol.colVal));
taosArrayPush(pLastArray, &lastCol); (void)taosArrayPush(pLastArray, &lastCol);
taosLRUCacheRelease(pCache, h, false); (void)taosLRUCacheRelease(pCache, h, false);
} else { } else {
SLastCol noneCol = {.rowKey.ts = TSKEY_MIN, SLastCol noneCol = {.rowKey.ts = TSKEY_MIN,
.colVal = COL_VAL_NONE(cid, pr->pSchema->columns[pr->pSlotIds[i]].type)}; .colVal = COL_VAL_NONE(cid, pr->pSchema->columns[pr->pSlotIds[i]].type)};
taosArrayPush(pLastArray, &noneCol); (void)taosArrayPush(pLastArray, &noneCol);
if (!remainCols) { if (!remainCols) {
remainCols = taosArrayInit(num_keys, sizeof(SIdxKey)); remainCols = taosArrayInit(num_keys, sizeof(SIdxKey));
} }
taosArrayPush(remainCols, &(SIdxKey){i, key}); (void)taosArrayPush(remainCols, &(SIdxKey){i, key});
} }
} }
@ -1721,7 +1715,7 @@ int32_t tsdbCacheGetBatch(STsdb *pTsdb, tb_uid_t uid, SArray *pLastArray, SCache
} }
taosArraySet(pLastArray, idxKey->idx, &lastCol); taosArraySet(pLastArray, idxKey->idx, &lastCol);
taosLRUCacheRelease(pCache, h, false); (void)taosLRUCacheRelease(pCache, h, false);
taosArrayRemove(remainCols, i); taosArrayRemove(remainCols, i);
} else { } else {
@ -1810,10 +1804,7 @@ int32_t tsdbCacheDel(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, TSKEY sKey, TSKE
rocksdb_writebatch_t *wb = pTsdb->rCache.writebatch; rocksdb_writebatch_t *wb = pTsdb->rCache.writebatch;
for (int i = 0; i < num_keys; ++i) { for (int i = 0; i < num_keys; ++i) {
SLastCol *pLastCol = NULL; SLastCol *pLastCol = NULL;
code = tsdbCacheDeserialize(values_list[i], values_list_sizes[i], &pLastCol); (void)tsdbCacheDeserialize(values_list[i], values_list_sizes[i], &pLastCol);
if (code) {
tsdbError("tsdb/cache: vgId:%d, deserialize failed since %s.", TD_VID(pTsdb->pVnode), tstrerror(code));
}
(void)taosThreadMutexLock(&pTsdb->rCache.rMutex); (void)taosThreadMutexLock(&pTsdb->rCache.rMutex);
if (NULL != pLastCol && (pLastCol->rowKey.ts <= eKey && pLastCol->rowKey.ts >= sKey)) { if (NULL != pLastCol && (pLastCol->rowKey.ts <= eKey && pLastCol->rowKey.ts >= sKey)) {
rocksdb_writebatch_delete(wb, keys_list[i], klen); rocksdb_writebatch_delete(wb, keys_list[i], klen);
@ -1821,10 +1812,7 @@ int32_t tsdbCacheDel(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, TSKEY sKey, TSKE
taosMemoryFreeClear(pLastCol); taosMemoryFreeClear(pLastCol);
pLastCol = NULL; pLastCol = NULL;
code = tsdbCacheDeserialize(values_list[i + num_keys], values_list_sizes[i + num_keys], &pLastCol); (void)tsdbCacheDeserialize(values_list[i + num_keys], values_list_sizes[i + num_keys], &pLastCol);
if (code) {
tsdbError("tsdb/cache: vgId:%d, deserialize failed since %s.", TD_VID(pTsdb->pVnode), tstrerror(code));
}
if (NULL != pLastCol && (pLastCol->rowKey.ts <= eKey && pLastCol->rowKey.ts >= sKey)) { if (NULL != pLastCol && (pLastCol->rowKey.ts <= eKey && pLastCol->rowKey.ts >= sKey)) {
rocksdb_writebatch_delete(wb, keys_list[num_keys + i], klen); rocksdb_writebatch_delete(wb, keys_list[num_keys + i], klen);
} }
@ -1846,7 +1834,7 @@ int32_t tsdbCacheDel(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, TSKEY sKey, TSKE
if (pLastCol->rowKey.ts <= eKey && pLastCol->rowKey.ts >= sKey) { if (pLastCol->rowKey.ts <= eKey && pLastCol->rowKey.ts >= sKey) {
erase = true; erase = true;
} }
taosLRUCacheRelease(pTsdb->lruCache, h, erase); (void)taosLRUCacheRelease(pTsdb->lruCache, h, erase);
} }
if (erase) { if (erase) {
taosLRUCacheErase(pTsdb->lruCache, keys_list[i], klen); taosLRUCacheErase(pTsdb->lruCache, keys_list[i], klen);
@ -1862,7 +1850,7 @@ int32_t tsdbCacheDel(STsdb *pTsdb, tb_uid_t suid, tb_uid_t uid, TSKEY sKey, TSKE
if (pLastCol->rowKey.ts <= eKey && pLastCol->rowKey.ts >= sKey) { if (pLastCol->rowKey.ts <= eKey && pLastCol->rowKey.ts >= sKey) {
erase = true; erase = true;
} }
taosLRUCacheRelease(pTsdb->lruCache, h, erase); (void)taosLRUCacheRelease(pTsdb->lruCache, h, erase);
} }
if (erase) { if (erase) {
taosLRUCacheErase(pTsdb->lruCache, keys_list[num_keys + i], klen); taosLRUCacheErase(pTsdb->lruCache, keys_list[num_keys + i], klen);
@ -1982,7 +1970,7 @@ static int32_t getTableDelDataFromTbData(STbData *pTbData, SArray *aDelData) {
SDelData *pDelData = pTbData ? pTbData->pHead : NULL; SDelData *pDelData = pTbData ? pTbData->pHead : NULL;
for (; pDelData; pDelData = pDelData->pNext) { for (; pDelData; pDelData = pDelData->pNext) {
taosArrayPush(aDelData, pDelData); (void)taosArrayPush(aDelData, pDelData);
} }
return code; return code;
@ -2005,7 +1993,7 @@ static STableLoadInfo *getTableLoadInfo(SCacheRowsReader *pReader, uint64_t uid)
if (!ppInfo) { if (!ppInfo) {
pInfo = taosMemoryCalloc(1, sizeof(STableLoadInfo)); pInfo = taosMemoryCalloc(1, sizeof(STableLoadInfo));
if (pInfo) { if (pInfo) {
tSimpleHashPut(pReader->pTableMap, &uid, sizeof(uint64_t), &pInfo, POINTER_BYTES); (void)tSimpleHashPut(pReader->pTableMap, &uid, sizeof(uint64_t), &pInfo, POINTER_BYTES);
} }
return pInfo; return pInfo;
@ -2117,11 +2105,11 @@ static int32_t loadTombFromBlk(const TTombBlkArray *pTombBlkArray, SCacheRowsRea
TD_VID(pReader->pTsdb->pVnode), pReader->pCurFileSet->fid, record.skey, record.ekey, uid);*/ TD_VID(pReader->pTsdb->pVnode), pReader->pCurFileSet->fid, record.skey, record.ekey, uid);*/
SDelData delData = {.version = record.version, .sKey = record.skey, .eKey = record.ekey}; SDelData delData = {.version = record.version, .sKey = record.skey, .eKey = record.ekey};
taosArrayPush(pInfo->pTombData, &delData); (void)taosArrayPush(pInfo->pTombData, &delData);
} }
} }
tTombBlockDestroy(&block); (void)tTombBlockDestroy(&block);
if (finished) { if (finished) {
TAOS_RETURN(code); TAOS_RETURN(code);
@ -2312,7 +2300,7 @@ static int32_t getNextRowFromFS(void *iter, TSDBROW **ppRow, bool *pIgnoreEarlie
state->pr->pCurFileSet = state->pFileSet; state->pr->pCurFileSet = state->pFileSet;
loadDataTomb(state->pr, state->pr->pFileReader); (void)loadDataTomb(state->pr, state->pr->pFileReader);
TAOS_CHECK_GOTO(tsdbDataFileReadBrinBlk(state->pr->pFileReader, &state->pr->pBlkArray), &lino, _err); TAOS_CHECK_GOTO(tsdbDataFileReadBrinBlk(state->pr->pFileReader, &state->pr->pBlkArray), &lino, _err);
} }
@ -2329,7 +2317,7 @@ static int32_t getNextRowFromFS(void *iter, TSDBROW **ppRow, bool *pIgnoreEarlie
SBrinBlk *pBrinBlk = &pBlkArray->data[i]; SBrinBlk *pBrinBlk = &pBlkArray->data[i];
if (state->suid >= pBrinBlk->minTbid.suid && state->suid <= pBrinBlk->maxTbid.suid) { if (state->suid >= pBrinBlk->minTbid.suid && state->suid <= pBrinBlk->maxTbid.suid) {
if (state->uid >= pBrinBlk->minTbid.uid && state->uid <= pBrinBlk->maxTbid.uid) { if (state->uid >= pBrinBlk->minTbid.uid && state->uid <= pBrinBlk->maxTbid.uid) {
taosArrayPush(state->pIndexList, pBrinBlk); (void)taosArrayPush(state->pIndexList, pBrinBlk);
} }
} else if (state->suid > pBrinBlk->maxTbid.suid || } else if (state->suid > pBrinBlk->maxTbid.suid ||
(state->suid == pBrinBlk->maxTbid.suid && state->uid > pBrinBlk->maxTbid.uid)) { (state->suid == pBrinBlk->maxTbid.suid && state->uid > pBrinBlk->maxTbid.uid)) {
@ -2385,7 +2373,7 @@ static int32_t getNextRowFromFS(void *iter, TSDBROW **ppRow, bool *pIgnoreEarlie
if (!state->pLastRow) { if (!state->pLastRow) {
if (state->pLastIter) { if (state->pLastIter) {
lastIterClose(&state->pLastIter); (void)lastIterClose(&state->pLastIter);
} }
clearLastFileSet(state); clearLastFileSet(state);
@ -2493,7 +2481,7 @@ static int32_t getNextRowFromFS(void *iter, TSDBROW **ppRow, bool *pIgnoreEarlie
if (!state->pLastRow) { if (!state->pLastRow) {
if (state->pLastIter) { if (state->pLastIter) {
lastIterClose(&state->pLastIter); (void)lastIterClose(&state->pLastIter);
} }
*ppRow = &state->row; *ppRow = &state->row;
@ -2517,7 +2505,7 @@ static int32_t getNextRowFromFS(void *iter, TSDBROW **ppRow, bool *pIgnoreEarlie
} else { } else {
// TODO: merge rows and *ppRow = mergedRow // TODO: merge rows and *ppRow = mergedRow
SRowMerger *pMerger = &state->rowMerger; SRowMerger *pMerger = &state->rowMerger;
tsdbRowMergerInit(pMerger, state->pTSchema); (void)tsdbRowMergerInit(pMerger, state->pTSchema);
TAOS_CHECK_GOTO(tsdbRowMergerAdd(pMerger, &state->row, state->pTSchema), &lino, _err); TAOS_CHECK_GOTO(tsdbRowMergerAdd(pMerger, &state->row, state->pTSchema), &lino, _err);
TAOS_CHECK_GOTO(tsdbRowMergerAdd(pMerger, state->pLastRow, state->pTSchema), &lino, _err); TAOS_CHECK_GOTO(tsdbRowMergerAdd(pMerger, state->pLastRow, state->pTSchema), &lino, _err);
@ -2682,7 +2670,7 @@ int32_t clearNextRowFromFS(void *iter) {
} }
if (state->pLastIter) { if (state->pLastIter) {
lastIterClose(&state->pLastIter); (void)lastIterClose(&state->pLastIter);
} }
if (state->pBlockData) { if (state->pBlockData) {
@ -2715,7 +2703,7 @@ int32_t clearNextRowFromFS(void *iter) {
static void clearLastFileSet(SFSNextRowIter *state) { static void clearLastFileSet(SFSNextRowIter *state) {
if (state->pLastIter) { if (state->pLastIter) {
lastIterClose(&state->pLastIter); (void)lastIterClose(&state->pLastIter);
} }
if (state->pBlockData) { if (state->pBlockData) {
@ -2724,7 +2712,7 @@ static void clearLastFileSet(SFSNextRowIter *state) {
} }
if (state->pr->pFileReader) { if (state->pr->pFileReader) {
tsdbDataFileReaderClose(&state->pr->pFileReader); (void)tsdbDataFileReaderClose(&state->pr->pFileReader);
state->pr->pFileReader = NULL; state->pr->pFileReader = NULL;
state->pr->pCurFileSet = NULL; state->pr->pCurFileSet = NULL;
@ -2814,7 +2802,7 @@ static int32_t nextRowIterClose(CacheNextRowIter *pIter) {
for (int i = 0; i < 3; ++i) { for (int i = 0; i < 3; ++i) {
if (pIter->input[i].nextRowClearFn) { if (pIter->input[i].nextRowClearFn) {
pIter->input[i].nextRowClearFn(pIter->input[i].iter); (void)pIter->input[i].nextRowClearFn(pIter->input[i].iter);
} }
} }
@ -2898,7 +2886,7 @@ static int32_t nextRowIterGet(CacheNextRowIter *pIter, TSDBROW **ppRow, bool *pI
pInfo->pTombData = taosArrayInit(4, sizeof(SDelData)); pInfo->pTombData = taosArrayInit(4, sizeof(SDelData));
} }
taosArrayAddAll(pInfo->pTombData, pIter->pMemDelData); (void)taosArrayAddAll(pInfo->pTombData, pIter->pMemDelData);
size_t delSize = TARRAY_SIZE(pInfo->pTombData); size_t delSize = TARRAY_SIZE(pInfo->pTombData);
if (delSize > 0) { if (delSize > 0) {
@ -2944,7 +2932,7 @@ static int32_t initLastColArrayPartial(STSchema *pTSchema, SArray **ppColArray,
int16_t slotId = slotIds[i]; int16_t slotId = slotIds[i];
SLastCol col = {.rowKey.ts = 0, SLastCol col = {.rowKey.ts = 0,
.colVal = COL_VAL_NULL(pTSchema->columns[slotId].colId, pTSchema->columns[slotId].type)}; .colVal = COL_VAL_NULL(pTSchema->columns[slotId].colId, pTSchema->columns[slotId].type)};
taosArrayPush(pColArray, &col); (void)taosArrayPush(pColArray, &col);
} }
*ppColArray = pColArray; *ppColArray = pColArray;
@ -2998,18 +2986,18 @@ static int32_t mergeLastCid(tb_uid_t uid, STsdb *pTsdb, SArray **ppLastArray, SC
} }
for (int i = 0; i < nCols; ++i) { for (int i = 0; i < nCols; ++i) {
taosArrayPush(aColArray, &aCols[i]); (void)taosArrayPush(aColArray, &aCols[i]);
} }
STsdbRowKey lastRowKey = {.key.ts = TSKEY_MAX}; STsdbRowKey lastRowKey = {.key.ts = TSKEY_MAX};
// inverse iterator // inverse iterator
CacheNextRowIter iter = {0}; CacheNextRowIter iter = {0};
nextRowIterOpen(&iter, uid, pTsdb, pTSchema, pr->info.suid, pr->pLDataIterArray, pr->pReadSnap, pr->lastTs, pr); (void)nextRowIterOpen(&iter, uid, pTsdb, pTSchema, pr->info.suid, pr->pLDataIterArray, pr->pReadSnap, pr->lastTs, pr);
do { do {
TSDBROW *pRow = NULL; TSDBROW *pRow = NULL;
nextRowIterGet(&iter, &pRow, &ignoreEarlierTs, true, TARRAY_DATA(aColArray), TARRAY_SIZE(aColArray)); (void)nextRowIterGet(&iter, &pRow, &ignoreEarlierTs, true, TARRAY_DATA(aColArray), TARRAY_SIZE(aColArray));
if (!pRow) { if (!pRow) {
break; break;
@ -3142,7 +3130,7 @@ static int32_t mergeLastCid(tb_uid_t uid, STsdb *pTsdb, SArray **ppLastArray, SC
} }
*ppLastArray = pColArray; *ppLastArray = pColArray;
nextRowIterClose(&iter); (void)nextRowIterClose(&iter);
taosArrayDestroy(aColArray); taosArrayDestroy(aColArray);
TAOS_RETURN(code); TAOS_RETURN(code);
@ -3184,16 +3172,16 @@ static int32_t mergeLastRowCid(tb_uid_t uid, STsdb *pTsdb, SArray **ppLastArray,
} }
for (int i = 0; i < nCols; ++i) { for (int i = 0; i < nCols; ++i) {
taosArrayPush(aColArray, &aCols[i]); (void)taosArrayPush(aColArray, &aCols[i]);
} }
// inverse iterator // inverse iterator
CacheNextRowIter iter = {0}; CacheNextRowIter iter = {0};
nextRowIterOpen(&iter, uid, pTsdb, pTSchema, pr->info.suid, pr->pLDataIterArray, pr->pReadSnap, pr->lastTs, pr); (void)nextRowIterOpen(&iter, uid, pTsdb, pTSchema, pr->info.suid, pr->pLDataIterArray, pr->pReadSnap, pr->lastTs, pr);
do { do {
TSDBROW *pRow = NULL; TSDBROW *pRow = NULL;
nextRowIterGet(&iter, &pRow, &ignoreEarlierTs, false, TARRAY_DATA(aColArray), TARRAY_SIZE(aColArray)); (void)nextRowIterGet(&iter, &pRow, &ignoreEarlierTs, false, TARRAY_DATA(aColArray), TARRAY_SIZE(aColArray));
if (!pRow) { if (!pRow) {
break; break;
@ -3266,7 +3254,7 @@ static int32_t mergeLastRowCid(tb_uid_t uid, STsdb *pTsdb, SArray **ppLastArray,
} }
*ppLastArray = pColArray; *ppLastArray = pColArray;
nextRowIterClose(&iter); (void)nextRowIterClose(&iter);
taosArrayDestroy(aColArray); taosArrayDestroy(aColArray);
TAOS_RETURN(code); TAOS_RETURN(code);
@ -3286,13 +3274,7 @@ _err:
TAOS_RETURN(code); TAOS_RETURN(code);
} }
int32_t tsdbCacheRelease(SLRUCache *pCache, LRUHandle *h) { void tsdbCacheRelease(SLRUCache *pCache, LRUHandle *h) { taosLRUCacheRelease(pCache, h, false); }
int32_t code = 0;
taosLRUCacheRelease(pCache, h, false);
return code;
}
void tsdbCacheSetCapacity(SVnode *pVnode, size_t capacity) { void tsdbCacheSetCapacity(SVnode *pVnode, size_t capacity) {
taosLRUCacheSetCapacity(pVnode->pTsdb->lruCache, capacity); taosLRUCacheSetCapacity(pVnode->pTsdb->lruCache, capacity);

2
source/libs/sync/src/syncRaftStore.c
View File

@ -134,7 +134,7 @@ int32_t raftStoreWriteFile(SSyncNode *pNode) {
if (taosFsyncFile(pFile) < 0) TAOS_CHECK_GOTO(TAOS_SYSTEM_ERROR(errno), &lino, _OVER); if (taosFsyncFile(pFile) < 0) TAOS_CHECK_GOTO(TAOS_SYSTEM_ERROR(errno), &lino, _OVER);
taosCloseFile(&pFile); (void)taosCloseFile(&pFile);
if (taosRenameFile(file, realfile) != 0) TAOS_CHECK_GOTO(TAOS_SYSTEM_ERROR(errno), &lino, _OVER); if (taosRenameFile(file, realfile) != 0) TAOS_CHECK_GOTO(TAOS_SYSTEM_ERROR(errno), &lino, _OVER);
code = 0; code = 0;

4
source/libs/sync/src/syncReplication.c
View File

@ -84,7 +84,7 @@ int32_t syncNodeReplicateWithoutLock(SSyncNode* pNode) {
int32_t syncNodeSendAppendEntries(SSyncNode* pSyncNode, const SRaftId* destRaftId, SRpcMsg* pRpcMsg) { int32_t syncNodeSendAppendEntries(SSyncNode* pSyncNode, const SRaftId* destRaftId, SRpcMsg* pRpcMsg) {
SyncAppendEntries* pMsg = pRpcMsg->pCont; SyncAppendEntries* pMsg = pRpcMsg->pCont;
pMsg->destId = *destRaftId; pMsg->destId = *destRaftId;
syncNodeSendMsgById(destRaftId, pSyncNode, pRpcMsg); (void)syncNodeSendMsgById(destRaftId, pSyncNode, pRpcMsg);
TAOS_RETURN(TSDB_CODE_SUCCESS); TAOS_RETURN(TSDB_CODE_SUCCESS);
} }
@ -113,7 +113,7 @@ int32_t syncNodeHeartbeatPeers(SSyncNode* pSyncNode) {
// send msg // send msg
syncLogSendHeartbeat(pSyncNode, pSyncMsg, true, 0, 0); syncLogSendHeartbeat(pSyncNode, pSyncMsg, true, 0, 0);
syncNodeSendHeartbeat(pSyncNode, &pSyncMsg->destId, &rpcMsg); (void)syncNodeSendHeartbeat(pSyncNode, &pSyncMsg->destId, &rpcMsg);
} }
TAOS_RETURN(TSDB_CODE_SUCCESS); TAOS_RETURN(TSDB_CODE_SUCCESS);

2
source/libs/sync/src/syncRequestVote.c
View File

@ -135,7 +135,7 @@ int32_t syncNodeOnRequestVote(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
// trace log // trace log
syncLogRecvRequestVote(ths, pMsg, pReply->voteGranted, ""); syncLogRecvRequestVote(ths, pMsg, pReply->voteGranted, "");
syncLogSendRequestVoteReply(ths, pReply, ""); syncLogSendRequestVoteReply(ths, pReply, "");
syncNodeSendMsgById(&pReply->destId, ths, &rpcMsg); (void)syncNodeSendMsgById(&pReply->destId, ths, &rpcMsg);
if (resetElect) syncNodeResetElectTimer(ths); if (resetElect) syncNodeResetElectTimer(ths);

6
source/util/src/tlrucache.c
View File

@ -320,7 +320,7 @@ static void taosLRUCacheShardEvictLRU(SLRUCacheShard *shard, size_t charge, SArr
ASSERT(TAOS_LRU_ENTRY_IN_CACHE(old) && !TAOS_LRU_ENTRY_HAS_REFS(old)); ASSERT(TAOS_LRU_ENTRY_IN_CACHE(old) && !TAOS_LRU_ENTRY_HAS_REFS(old));
taosLRUCacheShardLRURemove(shard, old); taosLRUCacheShardLRURemove(shard, old);
taosLRUEntryTableRemove(&shard->table, old->keyData, old->keyLength, old->hash); (void)taosLRUEntryTableRemove(&shard->table, old->keyData, old->keyLength, old->hash);
TAOS_LRU_ENTRY_SET_IN_CACHE(old, false); TAOS_LRU_ENTRY_SET_IN_CACHE(old, false);
ASSERT(shard->usage >= old->totalCharge); ASSERT(shard->usage >= old->totalCharge);
@ -531,7 +531,7 @@ static void taosLRUCacheShardEraseUnrefEntries(SLRUCacheShard *shard) {
SLRUEntry *old = shard->lru.next; SLRUEntry *old = shard->lru.next;
ASSERT(TAOS_LRU_ENTRY_IN_CACHE(old) && !TAOS_LRU_ENTRY_HAS_REFS(old)); ASSERT(TAOS_LRU_ENTRY_IN_CACHE(old) && !TAOS_LRU_ENTRY_HAS_REFS(old));
taosLRUCacheShardLRURemove(shard, old); taosLRUCacheShardLRURemove(shard, old);
taosLRUEntryTableRemove(&shard->table, old->keyData, old->keyLength, old->hash); (void)taosLRUEntryTableRemove(&shard->table, old->keyData, old->keyLength, old->hash);
TAOS_LRU_ENTRY_SET_IN_CACHE(old, false); TAOS_LRU_ENTRY_SET_IN_CACHE(old, false);
ASSERT(shard->usage >= old->totalCharge); ASSERT(shard->usage >= old->totalCharge);
shard->usage -= old->totalCharge; shard->usage -= old->totalCharge;
@ -577,7 +577,7 @@ static bool taosLRUCacheShardRelease(SLRUCacheShard *shard, LRUHandle *handle, b
if (shard->usage > shard->capacity || eraseIfLastRef) { if (shard->usage > shard->capacity || eraseIfLastRef) {
ASSERT(shard->lru.next == &shard->lru || eraseIfLastRef); ASSERT(shard->lru.next == &shard->lru || eraseIfLastRef);
taosLRUEntryTableRemove(&shard->table, e->keyData, e->keyLength, e->hash); (void)taosLRUEntryTableRemove(&shard->table, e->keyData, e->keyLength, e->hash);
TAOS_LRU_ENTRY_SET_IN_CACHE(e, false); TAOS_LRU_ENTRY_SET_IN_CACHE(e, false);
} else { } else {
taosLRUCacheShardLRUInsert(shard, e); taosLRUCacheShardLRUInsert(shard, e);