From 0ead238adbf98fa5e96ee7269f4ab0620a229aa2 Mon Sep 17 00:00:00 2001 From: Liu Jicong Date: Thu, 2 Jun 2022 15:52:59 +0800 Subject: [PATCH] feat(tmq): support check col and tag modifiable --- source/dnode/mnode/impl/inc/mndTopic.h | 2 + source/dnode/mnode/impl/src/mndTopic.c | 50 ++++++++++++++++++++++++ source/dnode/vnode/src/tq/tqPush.c | 54 +++++++++++++------------- source/libs/wal/src/walWrite.c | 2 +- 4 files changed, 79 insertions(+), 29 deletions(-) diff --git a/source/dnode/mnode/impl/inc/mndTopic.h b/source/dnode/mnode/impl/inc/mndTopic.h index c5c4800e02..4aa18ea591 100644 --- a/source/dnode/mnode/impl/inc/mndTopic.h +++ b/source/dnode/mnode/impl/inc/mndTopic.h @@ -37,6 +37,8 @@ const char *mndTopicGetShowName(const char topic[TSDB_TOPIC_FNAME_LEN]); int32_t mndSetTopicCommitLogs(SMnode *pMnode, STrans *pTrans, SMqTopicObj *pTopic); +bool mndCheckColAndTagModifiable(SMnode *pMnode, int64_t suid, const SArray *colIds); + #ifdef __cplusplus } #endif diff --git a/source/dnode/mnode/impl/src/mndTopic.c b/source/dnode/mnode/impl/src/mndTopic.c index 21b5e37e1e..e0d565c9af 100644 --- a/source/dnode/mnode/impl/src/mndTopic.c +++ b/source/dnode/mnode/impl/src/mndTopic.c @@ -70,6 +70,56 @@ const char *mndTopicGetShowName(const char topic[TSDB_TOPIC_FNAME_LEN]) { return strchr(topic, '.') + 1; } +bool mndCheckColAndTagModifiable(SMnode *pMnode, int64_t suid, const SArray *colAndTagIds) { + SSdb *pSdb = pMnode->pSdb; + void *pIter = NULL; + bool found = false; + while (1) { + SMqTopicObj *pTopic = NULL; + pIter = sdbFetch(pSdb, SDB_TOPIC, pIter, (void **)&pTopic); + if (pIter == NULL) break; + if (pTopic->subType != TOPIC_SUB_TYPE__COLUMN) { + sdbRelease(pSdb, pTopic); + continue; + } + + SNode *pAst = NULL; + if (nodesStringToNode(pTopic->ast, &pAst) != 0) { + ASSERT(0); + return false; + } + + SHashObj *pColHash = NULL; + SNodeList *pNodeList; + nodesCollectColumns((SSelectStmt *)pAst, SQL_CLAUSE_FROM, NULL, COLLECT_COL_TYPE_ALL, &pNodeList); + SNode *pNode = NULL; + FOREACH(pNode, pNodeList) { + SColumnNode *pCol = (SColumnNode *)pNode; + if (pCol->tableId != suid) goto NEXT; + if (pColHash == NULL) { + pColHash = taosHashInit(0, taosGetDefaultHashFunction(TSDB_DATA_TYPE_SMALLINT), false, HASH_NO_LOCK); + } + if (pCol->colId > 0) { + taosHashPut(pColHash, &pCol->colId, sizeof(int16_t), NULL, 0); + } + } + + for (int32_t i = 0; i < taosArrayGetSize(colAndTagIds); i++) { + int16_t *pColId = taosArrayGet(colAndTagIds, i); + if (taosHashGet(pColHash, pColId, sizeof(int16_t)) != NULL) { + found = true; + goto NEXT; + } + } + + NEXT: + sdbRelease(pSdb, pTopic); + nodesDestroyNode(pAst); + if (found) return false; + } + return true; +} + SSdbRaw *mndTopicActionEncode(SMqTopicObj *pTopic) { terrno = TSDB_CODE_OUT_OF_MEMORY; diff --git a/source/dnode/vnode/src/tq/tqPush.c b/source/dnode/vnode/src/tq/tqPush.c index 26e9dfe2e2..d94c3e387a 100644 --- a/source/dnode/vnode/src/tq/tqPush.c +++ b/source/dnode/vnode/src/tq/tqPush.c @@ -20,6 +20,28 @@ void tqTmrRspFunc(void* param, void* tmrId) { atomic_store_8(&pHandle->pushHandle.tmrStopped, 1); } +static int32_t tqLoopExecFromQueue(STQ* pTq, STqHandle* pHandle, SStreamDataSubmit** ppSubmit, SMqDataBlkRsp* pRsp) { + SStreamDataSubmit* pSubmit = *ppSubmit; + while (pSubmit != NULL) { + ASSERT(pSubmit->ver == pHandle->pushHandle.processedVer + 1); + if (tqDataExec(pTq, &pHandle->execHandle, pSubmit->data, pRsp, 0) < 0) { + /*ASSERT(0);*/ + } + // update processed + atomic_store_64(&pHandle->pushHandle.processedVer, pSubmit->ver); + streamQSetSuccess(&pHandle->pushHandle.inputQ); + streamDataSubmitRefDec(pSubmit); + if (pRsp->blockNum > 0) { + *ppSubmit = pSubmit; + return 0; + } else { + pSubmit = streamQNextItem(&pHandle->pushHandle.inputQ); + } + } + *ppSubmit = pSubmit; + return -1; +} + int32_t tqExecFromInputQ(STQ* pTq, STqHandle* pHandle) { SMqDataBlkRsp rsp = {0}; // 1. guard and set status executing @@ -42,38 +64,14 @@ int32_t tqExecFromInputQ(STQ* pTq, STqHandle* pHandle) { } // 3. exec, after each success, update processed ver // first run - while (pSubmit != NULL) { - ASSERT(pSubmit->ver == pHandle->pushHandle.processedVer + 1); - if (tqDataExec(pTq, &pHandle->execHandle, pSubmit->data, &rsp, 0) < 0) { - /*ASSERT(0);*/ - } - // update processed - atomic_store_64(&pHandle->pushHandle.processedVer, pSubmit->ver); - streamQSetSuccess(&pHandle->pushHandle.inputQ); - streamDataSubmitRefDec(pSubmit); - if (rsp.blockNum > 0) { - goto SEND_RSP; - } else { - pSubmit = streamQNextItem(&pHandle->pushHandle.inputQ); - } + if (tqLoopExecFromQueue(pTq, pHandle, &pSubmit, &rsp) == 0) { + goto SEND_RSP; } // set exec status closing atomic_store_8(&pHandle->pushHandle.execStatus, TASK_STATUS__CLOSING); // second run - while (pSubmit != NULL) { - ASSERT(pSubmit->ver == pHandle->pushHandle.processedVer + 1); - if (tqDataExec(pTq, &pHandle->execHandle, pSubmit->data, &rsp, 0) < 0) { - /*ASSERT(0);*/ - } - // update processed - atomic_store_64(&pHandle->pushHandle.processedVer, pSubmit->ver); - streamQSetSuccess(&pHandle->pushHandle.inputQ); - streamDataSubmitRefDec(pSubmit); - if (rsp.blockNum > 0) { - goto SEND_RSP; - } else { - pSubmit = streamQNextItem(&pHandle->pushHandle.inputQ); - } + if (tqLoopExecFromQueue(pTq, pHandle, &pSubmit, &rsp) == 0) { + goto SEND_RSP; } // set exec status idle atomic_store_8(&pHandle->pushHandle.execStatus, TASK_STATUS__IDLE); diff --git a/source/libs/wal/src/walWrite.c b/source/libs/wal/src/walWrite.c index 4e2270c37b..a9a8f8a1f4 100644 --- a/source/libs/wal/src/walWrite.c +++ b/source/libs/wal/src/walWrite.c @@ -19,7 +19,7 @@ #include "walInt.h" void walRestoreFromSnapshot(SWal *pWal, int64_t ver) { - pWal->vers.firstVer = -1; + /*pWal->vers.firstVer = -1;*/ pWal->vers.lastVer = ver; pWal->vers.commitVer = ver - 1; pWal->vers.snapshotVer = ver - 1; -- GitLab