From 21ea4e3a0b7062bef1f6574baf8f8285544f8a50 Mon Sep 17 00:00:00 2001 From: Hongze Cheng Date: Mon, 20 Jun 2022 02:38:08 +0000 Subject: [PATCH] more work --- source/dnode/vnode/inc/vnode.h | 9 ++- source/dnode/vnode/src/tq/tq.c | 2 +- source/dnode/vnode/src/tsdb/tsdbMemTable.c | 2 - source/dnode/vnode/src/tsdb/tsdbRead.c | 45 ------------- source/dnode/vnode/src/vnd/vnodeQuery.c | 49 +++++++++++++- source/libs/executor/src/executil.c | 75 ++++++++++------------ 6 files changed, 87 insertions(+), 95 deletions(-) diff --git a/source/dnode/vnode/inc/vnode.h b/source/dnode/vnode/inc/vnode.h index 6189f97b47..db1b0171ec 100644 --- a/source/dnode/vnode/inc/vnode.h +++ b/source/dnode/vnode/inc/vnode.h @@ -70,6 +70,10 @@ int32_t vnodeSnapshotReaderOpen(SVnode *pVnode, SVSnapshotReader **ppReader, int int32_t vnodeSnapshotReaderClose(SVSnapshotReader *pReader); int32_t vnodeSnapshotRead(SVSnapshotReader *pReader, const void **ppData, uint32_t *nData); int32_t vnodeProcessCreateTSma(SVnode *pVnode, void *pCont, uint32_t contLen); +int32_t vnodeGetAllTableList(SVnode *pVnode, uint64_t uid, SArray *list); +int32_t vnodeGetCtbIdList(SVnode *pVnode, int64_t suid, SArray *list); +void *vnodeGetIdx(SVnode *pVnode); +void *vnodeGetIvtIdx(SVnode *pVnode); void vnodeProposeMsg(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs); void vnodeApplyMsg(SQueueInfo *pInfo, STaosQall *qall, int32_t numOfMsgs); @@ -127,11 +131,6 @@ void tsdbResetReadHandle(STsdbReader *pReader, SQueryTableDataCond *pCond, in int32_t tsdbGetFileBlocksDistInfo(STsdbReader *pReader, STableBlockDistInfo *pTableBlockInfo); int64_t tsdbGetNumOfRowsInMemTable(STsdbReader *pHandle); -int32_t tsdbGetAllTableList(SMeta *pMeta, uint64_t uid, SArray *list); -int32_t tsdbGetCtbIdList(SMeta *pMeta, int64_t suid, SArray *list); -void *tsdbGetIdx(SMeta *pMeta); -void *tsdbGetIvtIdx(SMeta *pMeta); - // tq typedef struct STqReadHandle STqReadHandle; diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c index 70b09ec701..a3d4ef8c16 100644 --- a/source/dnode/vnode/src/tq/tq.c +++ b/source/dnode/vnode/src/tq/tq.c @@ -345,7 +345,7 @@ int32_t tqProcessVgChangeReq(STQ* pTq, char* msg, int32_t msgLen) { } else if (pHandle->execHandle.subType == TOPIC_SUB_TYPE__TABLE) { pHandle->execHandle.execTb.suid = req.suid; SArray* tbUidList = taosArrayInit(0, sizeof(int64_t)); - tsdbGetCtbIdList(pTq->pVnode->pMeta, req.suid, tbUidList); + vnodeGetCtbIdList(pTq->pVnode, req.suid, tbUidList); tqDebug("vg %d, tq try get suid: %ld", pTq->pVnode->config.vgId, req.suid); for (int32_t i = 0; i < taosArrayGetSize(tbUidList); i++) { int64_t tbUid = *(int64_t*)taosArrayGet(tbUidList, i); diff --git a/source/dnode/vnode/src/tsdb/tsdbMemTable.c b/source/dnode/vnode/src/tsdb/tsdbMemTable.c index 86dcdf09ab..a293960545 100644 --- a/source/dnode/vnode/src/tsdb/tsdbMemTable.c +++ b/source/dnode/vnode/src/tsdb/tsdbMemTable.c @@ -25,8 +25,6 @@ #define SL_MOVE_BACKWARD 0x1 #define SL_MOVE_FROM_POS 0x2 -static int32_t tPutTSDBRow(uint8_t *p, TSDBROW *pRow); -static int32_t tGetTSDBRow(uint8_t *p, TSDBROW *pRow); static void tbDataMovePosTo(STbData *pTbData, SMemSkipListNode **pos, TSDBKEY *pKey, int32_t flags); static int32_t tsdbGetOrCreateTbData(SMemTable *pMemTable, tb_uid_t suid, tb_uid_t uid, STbData **ppTbData); static int32_t tsdbInsertTableDataImpl(SMemTable *pMemTable, STbData *pTbData, int64_t version, diff --git a/source/dnode/vnode/src/tsdb/tsdbRead.c b/source/dnode/vnode/src/tsdb/tsdbRead.c index a43e4a8673..119633964c 100644 --- a/source/dnode/vnode/src/tsdb/tsdbRead.c +++ b/source/dnode/vnode/src/tsdb/tsdbRead.c @@ -2375,51 +2375,6 @@ _err: // return numOfRows; // } -// void* tsdbGetIdx(SMeta* pMeta) { -// if (pMeta == NULL) { -// return NULL; -// } -// return metaGetIdx(pMeta); -// } -// void* tsdbGetIvtIdx(SMeta* pMeta) { -// if (pMeta == NULL) { -// return NULL; -// } -// return metaGetIvtIdx(pMeta); -// } -// int32_t tsdbGetAllTableList(SMeta* pMeta, uint64_t uid, SArray* list) { -// SMCtbCursor* pCur = metaOpenCtbCursor(pMeta, uid); - -// while (1) { -// tb_uid_t id = metaCtbCursorNext(pCur); -// if (id == 0) { -// break; -// } - -// STableKeyInfo info = {.lastKey = TSKEY_INITIAL_VAL, uid = id}; -// taosArrayPush(list, &info); -// } - -// metaCloseCtbCursor(pCur); -// return TSDB_CODE_SUCCESS; -// } - -// int32_t tsdbGetCtbIdList(SMeta* pMeta, int64_t suid, SArray* list) { -// SMCtbCursor* pCur = metaOpenCtbCursor(pMeta, suid); - -// while (1) { -// tb_uid_t id = metaCtbCursorNext(pCur); -// if (id == 0) { -// break; -// } - -// taosArrayPush(list, &id); -// } - -// metaCloseCtbCursor(pCur); -// return TSDB_CODE_SUCCESS; -// } - // static void destroyHelper(void* param) { // if (param == NULL) { // return; diff --git a/source/dnode/vnode/src/vnd/vnodeQuery.c b/source/dnode/vnode/src/vnd/vnodeQuery.c index e8d96ce7fb..c044eb739b 100644 --- a/source/dnode/vnode/src/vnd/vnodeQuery.c +++ b/source/dnode/vnode/src/vnd/vnodeQuery.c @@ -148,4 +148,51 @@ void vnodeGetInfo(SVnode *pVnode, const char **dbname, int32_t *vgId) { if (vgId) { *vgId = TD_VID(pVnode); } -} \ No newline at end of file +} + +int32_t vnodeGetAllTableList(SVnode *pVnode, uint64_t uid, SArray *list) { + SMCtbCursor *pCur = metaOpenCtbCursor(pVnode->pMeta, uid); + + while (1) { + tb_uid_t id = metaCtbCursorNext(pCur); + if (id == 0) { + break; + } + + STableKeyInfo info = {.lastKey = TSKEY_INITIAL_VAL, uid = id}; + taosArrayPush(list, &info); + } + + metaCloseCtbCursor(pCur); + return TSDB_CODE_SUCCESS; +} + +int32_t vnodeGetCtbIdList(SVnode *pVnode, int64_t suid, SArray *list) { + SMCtbCursor *pCur = metaOpenCtbCursor(pVnode->pMeta, suid); + + while (1) { + tb_uid_t id = metaCtbCursorNext(pCur); + if (id == 0) { + break; + } + + taosArrayPush(list, &id); + } + + metaCloseCtbCursor(pCur); + return TSDB_CODE_SUCCESS; +} + +void *vnodeGetIdx(SVnode *pVnode) { + if (pVnode == NULL) { + return NULL; + } + return metaGetIdx(pVnode->pMeta); +} + +void *vnodeGetIvtIdx(SVnode *pVnode) { + if (pVnode == NULL) { + return NULL; + } + return metaGetIvtIdx(pVnode->pMeta); +} diff --git a/source/libs/executor/src/executil.c b/source/libs/executor/src/executil.c index 99139be409..ce78fe630b 100644 --- a/source/libs/executor/src/executil.c +++ b/source/libs/executor/src/executil.c @@ -13,10 +13,10 @@ * along with this program. If not, see . */ -#include "os.h" -#include "index.h" #include "function.h" #include "functionMgt.h" +#include "index.h" +#include "os.h" #include "tdatablock.h" #include "thash.h" #include "tmsg.h" @@ -25,45 +25,41 @@ #include "executorimpl.h" #include "tcompression.h" -void initResultRowInfo(SResultRowInfo *pResultRowInfo) { - pResultRowInfo->size = 0; +void initResultRowInfo(SResultRowInfo* pResultRowInfo) { + pResultRowInfo->size = 0; pResultRowInfo->cur.pageId = -1; } -void cleanupResultRowInfo(SResultRowInfo *pResultRowInfo) { +void cleanupResultRowInfo(SResultRowInfo* pResultRowInfo) { if (pResultRowInfo == NULL) { return; } - for(int32_t i = 0; i < pResultRowInfo->size; ++i) { -// if (pResultRowInfo->pResult[i]) { -// taosMemoryFreeClear(pResultRowInfo->pResult[i]->key); -// } + for (int32_t i = 0; i < pResultRowInfo->size; ++i) { + // if (pResultRowInfo->pResult[i]) { + // taosMemoryFreeClear(pResultRowInfo->pResult[i]->key); + // } } } -void closeAllResultRows(SResultRowInfo *pResultRowInfo) { -// do nothing +void closeAllResultRows(SResultRowInfo* pResultRowInfo) { + // do nothing } -bool isResultRowClosed(SResultRow* pRow) { - return (pRow->closed == true); -} +bool isResultRowClosed(SResultRow* pRow) { return (pRow->closed == true); } -void closeResultRow(SResultRow* pResultRow) { - pResultRow->closed = true; -} +void closeResultRow(SResultRow* pResultRow) { pResultRow->closed = true; } // TODO refactor: use macro SResultRowEntryInfo* getResultEntryInfo(const SResultRow* pRow, int32_t index, const int32_t* offset) { assert(index >= 0 && offset != NULL); - return (SResultRowEntryInfo*)((char*) pRow->pEntryInfo + offset[index]); + return (SResultRowEntryInfo*)((char*)pRow->pEntryInfo + offset[index]); } size_t getResultRowSize(SqlFunctionCtx* pCtx, int32_t numOfOutput) { int32_t rowSize = (numOfOutput * sizeof(SResultRowEntryInfo)) + sizeof(SResultRow); - for(int32_t i = 0; i < numOfOutput; ++i) { + for (int32_t i = 0; i < numOfOutput; ++i) { rowSize += pCtx[i].resDataInfo.interBufSize; } @@ -74,31 +70,29 @@ void cleanupGroupResInfo(SGroupResInfo* pGroupResInfo) { assert(pGroupResInfo != NULL); taosArrayDestroy(pGroupResInfo->pRows); - pGroupResInfo->pRows = NULL; - pGroupResInfo->index = 0; + pGroupResInfo->pRows = NULL; + pGroupResInfo->index = 0; } static int32_t resultrowComparAsc(const void* p1, const void* p2) { - SResKeyPos* pp1 = *(SResKeyPos**) p1; - SResKeyPos* pp2 = *(SResKeyPos**) p2; + SResKeyPos* pp1 = *(SResKeyPos**)p1; + SResKeyPos* pp2 = *(SResKeyPos**)p2; if (pp1->groupId == pp2->groupId) { - int64_t pts1 = *(int64_t*) pp1->key; - int64_t pts2 = *(int64_t*) pp2->key; + int64_t pts1 = *(int64_t*)pp1->key; + int64_t pts2 = *(int64_t*)pp2->key; if (pts1 == pts2) { return 0; } else { - return pts1 < pts2? -1:1; + return pts1 < pts2 ? -1 : 1; } } else { - return pp1->groupId < pp2->groupId? -1:1; + return pp1->groupId < pp2->groupId ? -1 : 1; } } -static int32_t resultrowComparDesc(const void* p1, const void* p2) { - return resultrowComparAsc(p2, p1); -} +static int32_t resultrowComparDesc(const void* p1, const void* p2) { return resultrowComparAsc(p2, p1); } void initGroupedResultInfo(SGroupResInfo* pGroupResInfo, SHashObj* pHashmap, int32_t order) { if (pGroupResInfo->pRows != NULL) { @@ -110,20 +104,20 @@ void initGroupedResultInfo(SGroupResInfo* pGroupResInfo, SHashObj* pHashmap, int pGroupResInfo->pRows = taosArrayInit(10, POINTER_BYTES); size_t keyLen = 0; - while((pData = taosHashIterate(pHashmap, pData)) != NULL) { + while ((pData = taosHashIterate(pHashmap, pData)) != NULL) { void* key = taosHashGetKey(pData, &keyLen); SResKeyPos* p = taosMemoryMalloc(keyLen + sizeof(SResultRowPosition)); - p->groupId = *(uint64_t*) key; - p->pos = *(SResultRowPosition*) pData; + p->groupId = *(uint64_t*)key; + p->pos = *(SResultRowPosition*)pData; memcpy(p->key, (char*)key + sizeof(uint64_t), keyLen - sizeof(uint64_t)); taosArrayPush(pGroupResInfo->pRows, &p); } if (order == TSDB_ORDER_ASC || order == TSDB_ORDER_DESC) { - __compar_fn_t fn = (order == TSDB_ORDER_ASC)? resultrowComparAsc:resultrowComparDesc; + __compar_fn_t fn = (order == TSDB_ORDER_ASC) ? resultrowComparAsc : resultrowComparDesc; qsort(pGroupResInfo->pRows->pData, taosArrayGetSize(pGroupResInfo->pRows), POINTER_BYTES, fn); } @@ -155,7 +149,7 @@ int32_t getNumOfTotalRes(SGroupResInfo* pGroupResInfo) { return 0; } - return (int32_t) taosArrayGetSize(pGroupResInfo->pRows); + return (int32_t)taosArrayGetSize(pGroupResInfo->pRows); } SArray* createSortInfo(SNodeList* pNodeList) { @@ -223,7 +217,7 @@ int32_t getTableList(void* metaHandle, SScanPhysiNode* pScanNode, STableListInfo if (pScanNode->tableType == TSDB_SUPER_TABLE) { if (pTagCond) { SIndexMetaArg metaArg = { - .metaEx = metaHandle, .idx = tsdbGetIdx(metaHandle), .ivtIdx = tsdbGetIvtIdx(metaHandle), .suid = tableUid}; + .metaEx = metaHandle, .idx = vnodeGetIdx(metaHandle), .ivtIdx = vnodeGetIvtIdx(metaHandle), .suid = tableUid}; SArray* res = taosArrayInit(8, sizeof(uint64_t)); code = doFilterTag(pTagCond, &metaArg, res); @@ -244,7 +238,7 @@ int32_t getTableList(void* metaHandle, SScanPhysiNode* pScanNode, STableListInfo } taosArrayDestroy(res); } else { - code = tsdbGetAllTableList(metaHandle, tableUid, pListInfo->pTableList); + code = vnodeGetAllTableList(metaHandle, tableUid, pListInfo->pTableList); } } else { // Create one table group. STableKeyInfo info = {.lastKey = 0, .uid = tableUid}; @@ -255,7 +249,7 @@ int32_t getTableList(void* metaHandle, SScanPhysiNode* pScanNode, STableListInfo } SArray* extractPartitionColInfo(SNodeList* pNodeList) { - if(!pNodeList) { + if (!pNodeList) { return NULL; } @@ -284,7 +278,6 @@ SArray* extractPartitionColInfo(SNodeList* pNodeList) { return pList; } - SArray* extractColMatchInfo(SNodeList* pNodeList, SDataBlockDescNode* pOutputNodeList, int32_t* numOfOutputCols, int32_t type) { size_t numOfCols = LIST_LENGTH(pNodeList); @@ -569,8 +562,8 @@ SqlFunctionCtx* createSqlFunctionCtx(SExprInfo* pExprInfo, int32_t numOfOutput, } for (int32_t i = 1; i < numOfOutput; ++i) { - (*rowEntryInfoOffset)[i] = - (int32_t)((*rowEntryInfoOffset)[i - 1] + sizeof(SResultRowEntryInfo) + pFuncCtx[i - 1].resDataInfo.interBufSize); + (*rowEntryInfoOffset)[i] = (int32_t)((*rowEntryInfoOffset)[i - 1] + sizeof(SResultRowEntryInfo) + + pFuncCtx[i - 1].resDataInfo.interBufSize); } setSelectValueColumnInfo(pFuncCtx, numOfOutput); -- GitLab