From 2441ec22bd6412b70aeff342258e6b985684e580 Mon Sep 17 00:00:00 2001 From: Liu Jicong Date: Fri, 22 Jul 2022 16:05:28 +0800 Subject: [PATCH] feat(tmq): check alter for ntb --- include/common/taosdef.h | 9 +- include/common/tmsg.h | 9 ++ include/common/tmsgdef.h | 1 + include/libs/executor/executor.h | 3 +- include/util/taoserror.h | 1 + source/common/src/tmsg.c | 27 +++++ source/dnode/mgmt/mgmt_mnode/src/mmHandle.c | 1 + source/dnode/mgmt/mgmt_vnode/src/vmHandle.c | 1 + source/dnode/mnode/impl/inc/mndDef.h | 57 ++++++----- source/dnode/mnode/impl/src/mndTopic.c | 108 +++++++++++++++++++- source/dnode/vnode/src/inc/tq.h | 4 +- source/dnode/vnode/src/inc/vnodeInt.h | 5 +- source/dnode/vnode/src/meta/metaTable.c | 12 +++ source/dnode/vnode/src/tq/tq.c | 40 ++++++-- source/dnode/vnode/src/tq/tqMeta.c | 3 +- source/dnode/vnode/src/tq/tqSink.c | 1 - source/dnode/vnode/src/vnd/vnodeOpen.c | 2 +- source/dnode/vnode/src/vnd/vnodeSvr.c | 6 ++ source/libs/executor/inc/executorimpl.h | 1 + source/libs/executor/src/executor.c | 3 +- source/libs/executor/src/executorMain.c | 16 ++- source/libs/executor/src/executorimpl.c | 3 + source/libs/executor/src/scanoperator.c | 16 ++- source/util/src/terror.c | 1 + tests/pytest/topic/topicQuery.py | 91 ----------------- tests/system-test/7-tmq/schema.py | 6 +- tests/test/c/CMakeLists.txt | 3 +- 27 files changed, 273 insertions(+), 157 deletions(-) delete mode 100644 tests/pytest/topic/topicQuery.py diff --git a/include/common/taosdef.h b/include/common/taosdef.h index 516df71b0b..9bfee56e29 100644 --- a/include/common/taosdef.h +++ b/include/common/taosdef.h @@ -34,11 +34,10 @@ typedef enum { TSDB_SUPER_TABLE = 1, // super table TSDB_CHILD_TABLE = 2, // table created from super table TSDB_NORMAL_TABLE = 3, // ordinary table - TSDB_STREAM_TABLE = 4, // table created from stream computing - TSDB_TEMP_TABLE = 5, // temp table created by nest query - TSDB_SYSTEM_TABLE = 6, - TSDB_TSMA_TABLE = 7, // time-range-wise sma - TSDB_TABLE_MAX = 8 + TSDB_TEMP_TABLE = 4, // temp table created by nest query + TSDB_SYSTEM_TABLE = 5, + TSDB_TSMA_TABLE = 6, // time-range-wise sma + TSDB_TABLE_MAX = 7 } ETableType; typedef enum { diff --git a/include/common/tmsg.h b/include/common/tmsg.h index 77b6ca1833..45be3a8507 100644 --- a/include/common/tmsg.h +++ b/include/common/tmsg.h @@ -2537,6 +2537,15 @@ static FORCE_INLINE void* tDecodeSMqRebVgReq(const void* buf, SMqRebVgReq* pReq) return (void*)buf; } +typedef struct { + char topic[TSDB_TOPIC_FNAME_LEN]; + int64_t ntbUid; + SArray* colIdList; // SArray +} SCheckAlterInfo; + +int32_t tEncodeSCheckAlterInfo(SEncoder* pEncoder, const SCheckAlterInfo* pInfo); +int32_t tDecodeSCheckAlterInfo(SDecoder* pDecoder, SCheckAlterInfo* pInfo); + typedef struct { int32_t vgId; int64_t offset; diff --git a/include/common/tmsgdef.h b/include/common/tmsgdef.h index a9d8871c59..56e0935ce1 100644 --- a/include/common/tmsgdef.h +++ b/include/common/tmsgdef.h @@ -186,6 +186,7 @@ enum { TD_DEF_MSG_TYPE(TDMT_VND_MQ_VG_CHANGE, "vnode-mq-vg-change", SMqRebVgReq, SMqRebVgRsp) TD_DEF_MSG_TYPE(TDMT_VND_MQ_VG_DELETE, "vnode-mq-vg-delete", SMqVDeleteReq, SMqVDeleteRsp) TD_DEF_MSG_TYPE(TDMT_VND_MQ_COMMIT_OFFSET, "vnode-commit-offset", STqOffset, STqOffset) + TD_DEF_MSG_TYPE(TDMT_VND_CHECK_ALTER_INFO, "vnode-alter-check-info", NULL, NULL) TD_DEF_MSG_TYPE(TDMT_VND_CREATE_TOPIC, "vnode-create-topic", NULL, NULL) TD_DEF_MSG_TYPE(TDMT_VND_ALTER_TOPIC, "vnode-alter-topic", NULL, NULL) TD_DEF_MSG_TYPE(TDMT_VND_DROP_TOPIC, "vnode-drop-topic", NULL, NULL) diff --git a/include/libs/executor/executor.h b/include/libs/executor/executor.h index 65244ec11a..25fd1a5e79 100644 --- a/include/libs/executor/executor.h +++ b/include/libs/executor/executor.h @@ -64,7 +64,8 @@ qTaskInfo_t qCreateStreamExecTaskInfo(void* msg, SReadHandle* readers); * @param SReadHandle * @return */ -qTaskInfo_t qCreateQueueExecTaskInfo(void* msg, SReadHandle* readers, int32_t* numOfCols, SSchemaWrapper** pSchemaWrapper); +qTaskInfo_t qCreateQueueExecTaskInfo(void* msg, SReadHandle* readers, int32_t* numOfCols, + SSchemaWrapper** pSchemaWrapper, int64_t* ntbUid); /** * Set the input data block for the stream scan. diff --git a/include/util/taoserror.h b/include/util/taoserror.h index 89b0e6f952..431abe034f 100644 --- a/include/util/taoserror.h +++ b/include/util/taoserror.h @@ -332,6 +332,7 @@ int32_t* taosGetErrno(); #define TSDB_CODE_VND_INVALID_TABLE_ACTION TAOS_DEF_ERROR_CODE(0, 0x0519) #define TSDB_CODE_VND_COL_ALREADY_EXISTS TAOS_DEF_ERROR_CODE(0, 0x051a) #define TSDB_CODE_VND_TABLE_COL_NOT_EXISTS TAOS_DEF_ERROR_CODE(0, 0x051b) +#define TSDB_CODE_VND_COL_SUBSCRIBED TAOS_DEF_ERROR_CODE(0, 0x051c) // tsdb #define TSDB_CODE_TDB_INVALID_TABLE_ID TAOS_DEF_ERROR_CODE(0, 0x0600) diff --git a/source/common/src/tmsg.c b/source/common/src/tmsg.c index 3e733d291b..2a3452c534 100644 --- a/source/common/src/tmsg.c +++ b/source/common/src/tmsg.c @@ -5623,6 +5623,33 @@ int32_t tDecodeSTqOffset(SDecoder *pDecoder, STqOffset *pOffset) { return 0; } +int32_t tEncodeSCheckAlterInfo(SEncoder *pEncoder, const SCheckAlterInfo *pInfo) { + if (tEncodeCStr(pEncoder, pInfo->topic) < 0) return -1; + if (tEncodeI64(pEncoder, pInfo->ntbUid) < 0) return -1; + int32_t sz = taosArrayGetSize(pInfo->colIdList); + if (tEncodeI32(pEncoder, sz) < 0) return -1; + for (int32_t i = 0; i < sz; i++) { + int16_t colId = *(int16_t *)taosArrayGet(pInfo->colIdList, i); + if (tEncodeI16(pEncoder, colId) < 0) return -1; + } + return pEncoder->pos; +} + +int32_t tDecodeSCheckAlterInfo(SDecoder *pDecoder, SCheckAlterInfo *pInfo) { + if (tDecodeCStrTo(pDecoder, pInfo->topic) < 0) return -1; + if (tDecodeI64(pDecoder, &pInfo->ntbUid) < 0) return -1; + int32_t sz; + if (tDecodeI32(pDecoder, &sz) < 0) return -1; + pInfo->colIdList = taosArrayInit(sz, sizeof(int16_t)); + if (pInfo->colIdList == NULL) return -1; + for (int32_t i = 0; i < sz; i++) { + int16_t colId; + if (tDecodeI16(pDecoder, &colId) < 0) return -1; + taosArrayPush(pInfo->colIdList, &colId); + } + return 0; +} + int32_t tEncodeDeleteRes(SEncoder *pCoder, const SDeleteRes *pRes) { int32_t nUid = taosArrayGetSize(pRes->uidList); diff --git a/source/dnode/mgmt/mgmt_mnode/src/mmHandle.c b/source/dnode/mgmt/mgmt_mnode/src/mmHandle.c index 8db43a460e..22a53f07f6 100644 --- a/source/dnode/mgmt/mgmt_mnode/src/mmHandle.c +++ b/source/dnode/mgmt/mgmt_mnode/src/mmHandle.c @@ -224,6 +224,7 @@ SArray *mmGetMsgHandles() { if (dmSetMgmtHandle(pArray, TDMT_VND_DROP_SMA_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_VG_CHANGE_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_VG_DELETE_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_CHECK_ALTER_INFO_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_SCH_DROP_TASK, mmPutMsgToFetchQueue, 1) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_DEPLOY_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_DROP_RSP, mmPutMsgToWriteQueue, 0) == NULL) goto _OVER; diff --git a/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c b/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c index 9571a83116..0471e2b850 100644 --- a/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c +++ b/source/dnode/mgmt/mgmt_vnode/src/vmHandle.c @@ -352,6 +352,7 @@ SArray *vmGetMsgHandles() { if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_VG_CHANGE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_VG_DELETE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_VND_MQ_COMMIT_OFFSET, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER; + if (dmSetMgmtHandle(pArray, TDMT_VND_CHECK_ALTER_INFO, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_VND_CONSUME, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_VND_DELETE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER; if (dmSetMgmtHandle(pArray, TDMT_VND_COMMIT, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER; diff --git a/source/dnode/mnode/impl/inc/mndDef.h b/source/dnode/mnode/impl/inc/mndDef.h index 977f33de49..975d533a44 100644 --- a/source/dnode/mnode/impl/inc/mndDef.h +++ b/source/dnode/mnode/impl/inc/mndDef.h @@ -144,33 +144,33 @@ typedef enum { } ECsmUpdateType; typedef struct { - int32_t id; - ETrnStage stage; - ETrnPolicy policy; - ETrnConflct conflict; - ETrnExec exec; - EOperType oper; - int32_t code; - int32_t failedTimes; - void* rpcRsp; - int32_t rpcRspLen; - int32_t redoActionPos; - SArray* redoActions; - SArray* undoActions; - SArray* commitActions; - int64_t createdTime; - int64_t lastExecTime; - int32_t lastAction; - int32_t lastErrorNo; - tmsg_t lastMsgType; - SEpSet lastEpset; - char dbname1[TSDB_DB_FNAME_LEN]; - char dbname2[TSDB_DB_FNAME_LEN]; - int32_t startFunc; - int32_t stopFunc; - int32_t paramLen; - void* param; - SArray* pRpcArray; + int32_t id; + ETrnStage stage; + ETrnPolicy policy; + ETrnConflct conflict; + ETrnExec exec; + EOperType oper; + int32_t code; + int32_t failedTimes; + void* rpcRsp; + int32_t rpcRspLen; + int32_t redoActionPos; + SArray* redoActions; + SArray* undoActions; + SArray* commitActions; + int64_t createdTime; + int64_t lastExecTime; + int32_t lastAction; + int32_t lastErrorNo; + tmsg_t lastMsgType; + SEpSet lastEpset; + char dbname1[TSDB_DB_FNAME_LEN]; + char dbname2[TSDB_DB_FNAME_LEN]; + int32_t startFunc; + int32_t stopFunc; + int32_t paramLen; + void* param; + SArray* pRpcArray; } STrans; typedef struct { @@ -477,6 +477,9 @@ typedef struct { char* physicalPlan; SSchemaWrapper schema; int64_t stbUid; + // forbid condition + int64_t ntbUid; + SArray* ntbColIds; } SMqTopicObj; typedef struct { diff --git a/source/dnode/mnode/impl/src/mndTopic.c b/source/dnode/mnode/impl/src/mndTopic.c index 268cbaa55c..a9f155ec86 100644 --- a/source/dnode/mnode/impl/src/mndTopic.c +++ b/source/dnode/mnode/impl/src/mndTopic.c @@ -57,6 +57,7 @@ int32_t mndInitTopic(SMnode *pMnode) { mndSetMsgHandle(pMnode, TDMT_MND_CREATE_TOPIC, mndProcessCreateTopicReq); mndSetMsgHandle(pMnode, TDMT_MND_DROP_TOPIC, mndProcessDropTopicReq); mndSetMsgHandle(pMnode, TDMT_VND_DROP_TOPIC_RSP, mndTransProcessRsp); + mndSetMsgHandle(pMnode, TDMT_VND_CHECK_ALTER_INFO_RSP, mndTransProcessRsp); mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_TOPICS, mndRetrieveTopic); mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_TOPICS, mndCancelGetNextTopic); @@ -127,8 +128,10 @@ SSdbRaw *mndTopicActionEncode(SMqTopicObj *pTopic) { if (pTopic->schema.nCols) { schemaLen = taosEncodeSSchemaWrapper(NULL, &pTopic->schema); } - int32_t size = - sizeof(SMqTopicObj) + physicalPlanLen + pTopic->sqlLen + pTopic->astLen + schemaLen + MND_TOPIC_RESERVE_SIZE; + int32_t ntbColLen = taosArrayGetSize(pTopic->ntbColIds) * sizeof(int16_t); + + int32_t size = sizeof(SMqTopicObj) + physicalPlanLen + pTopic->sqlLen + pTopic->astLen + schemaLen + ntbColLen + + MND_TOPIC_RESERVE_SIZE; SSdbRaw *pRaw = sdbAllocRaw(SDB_TOPIC, MND_TOPIC_VER_NUMBER, size); if (pRaw == NULL) goto TOPIC_ENCODE_OVER; @@ -164,6 +167,15 @@ SSdbRaw *mndTopicActionEncode(SMqTopicObj *pTopic) { taosEncodeSSchemaWrapper(&aswBuf, &pTopic->schema); SDB_SET_BINARY(pRaw, dataPos, swBuf, schemaLen, TOPIC_ENCODE_OVER); } + SDB_SET_INT64(pRaw, dataPos, pTopic->ntbUid, TOPIC_ENCODE_OVER); + if (pTopic->ntbUid != 0) { + int32_t sz = taosArrayGetSize(pTopic->ntbColIds); + SDB_SET_INT32(pRaw, dataPos, sz, TOPIC_ENCODE_OVER); + for (int32_t i = 0; i < sz; i++) { + int16_t colId = *(int16_t *)taosArrayGet(pTopic->ntbColIds, i); + SDB_SET_INT16(pRaw, dataPos, colId, TOPIC_ENCODE_OVER); + } + } SDB_SET_RESERVE(pRaw, dataPos, MND_TOPIC_RESERVE_SIZE, TOPIC_ENCODE_OVER); SDB_SET_DATALEN(pRaw, dataPos, TOPIC_ENCODE_OVER); @@ -259,6 +271,19 @@ SSdbRow *mndTopicActionDecode(SSdbRaw *pRaw) { pTopic->schema.version = 0; pTopic->schema.pSchema = NULL; } + SDB_GET_INT64(pRaw, dataPos, &pTopic->ntbUid, TOPIC_DECODE_OVER); + if (pTopic->ntbUid != 0) { + int32_t ntbColNum; + SDB_GET_INT32(pRaw, dataPos, &ntbColNum, TOPIC_DECODE_OVER); + pTopic->ntbColIds = taosArrayInit(ntbColNum, sizeof(int16_t)); + if (pTopic->ntbColIds == NULL) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + goto TOPIC_DECODE_OVER; + } + int16_t colId; + SDB_GET_INT16(pRaw, dataPos, &colId, TOPIC_DECODE_OVER); + taosArrayPush(pTopic->ntbColIds, &colId); + } SDB_GET_RESERVE(pRaw, dataPos, MND_TOPIC_RESERVE_SIZE, TOPIC_DECODE_OVER); @@ -346,6 +371,19 @@ static int32_t mndCheckCreateTopicReq(SCMCreateTopicReq *pCreate) { return 0; } +static int32_t extractTopicTbInfo(SNode *pAst, int64_t *ntbUid, SArray *colIds) { + SNodeList *pNodeList = NULL; + nodesCollectColumns((SSelectStmt *)pAst, SQL_CLAUSE_FROM, NULL, COLLECT_COL_TYPE_ALL, &pNodeList); + SNode *pNode = NULL; + FOREACH(pNode, pNodeList) { + SColumnNode *pCol = (SColumnNode *)pNode; + if (pCol->tableType != TSDB_NORMAL_TABLE) return -1; + *ntbUid = pCol->tableId; + taosArrayPush(colIds, &pCol->colId); + } + return 0; +} + static int32_t mndCreateTopic(SMnode *pMnode, SRpcMsg *pReq, SCMCreateTopicReq *pCreate, SDbObj *pDb) { mDebug("topic:%s to create", pCreate->name); SMqTopicObj topicObj = {0}; @@ -386,6 +424,19 @@ static int32_t mndCreateTopic(SMnode *pMnode, SRpcMsg *pReq, SCMCreateTopicReq * return -1; } + int64_t ntbUid; + SArray *colIds = taosArrayInit(0, sizeof(int16_t)); + if (colIds == NULL) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + return -1; + } + if (extractTopicTbInfo(pAst, &ntbUid, colIds) < 0) { + taosArrayDestroy(colIds); + } else { + topicObj.ntbUid = ntbUid; + topicObj.ntbColIds = colIds; + } + if (qExtractResultSchema(pAst, &topicObj.schema.nCols, &topicObj.schema.pSchema) != 0) { mError("topic:%s, failed to create since %s", pCreate->name, terrstr()); taosMemoryFree(topicObj.ast); @@ -433,6 +484,59 @@ static int32_t mndCreateTopic(SMnode *pMnode, SRpcMsg *pReq, SCMCreateTopicReq * } sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY); + if (topicObj.ntbUid != 0) { + SCheckAlterInfo info; + memcpy(info.topic, topicObj.name, TSDB_TOPIC_FNAME_LEN); + info.ntbUid = topicObj.ntbUid; + info.colIdList = topicObj.ntbColIds; + // broadcast forbid alter info + void *pIter = NULL; + SSdb *pSdb = pMnode->pSdb; + SVgObj *pVgroup = NULL; + while (1) { + // iterate vg + pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup); + if (pIter == NULL) break; + if (!mndVgroupInDb(pVgroup, topicObj.dbUid)) { + sdbRelease(pSdb, pVgroup); + continue; + } + + // encoder check alter info + int32_t len; + int32_t code; + tEncodeSize(tEncodeSCheckAlterInfo, &info, len, code); + if (code < 0) { + sdbRelease(pSdb, pVgroup); + mndTransDrop(pTrans); + return -1; + } + void *buf = taosMemoryCalloc(1, sizeof(SMsgHead) + len); + void *abuf = POINTER_SHIFT(buf, sizeof(SMsgHead)); + SEncoder encoder; + tEncoderInit(&encoder, abuf, len); + if (tEncodeSCheckAlterInfo(&encoder, &info) < 0) { + sdbRelease(pSdb, pVgroup); + mndTransDrop(pTrans); + return -1; + } + tEncoderClear(&encoder); + ((SMsgHead *)buf)->vgId = pVgroup->vgId; + // add redo action + STransAction action = {0}; + action.epSet = mndGetVgroupEpset(pMnode, pVgroup); + action.pCont = buf; + action.contLen = sizeof(SMsgHead) + len; + action.msgType = TDMT_VND_CHECK_ALTER_INFO; + if (mndTransAppendRedoAction(pTrans, &action) != 0) { + taosMemoryFree(buf); + sdbRelease(pSdb, pVgroup); + mndTransDrop(pTrans); + return -1; + } + } + } + if (mndTransPrepare(pMnode, pTrans) != 0) { mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr()); taosMemoryFreeClear(topicObj.physicalPlan); diff --git a/source/dnode/vnode/src/inc/tq.h b/source/dnode/vnode/src/inc/tq.h index 93c70e1109..744a5d7eac 100644 --- a/source/dnode/vnode/src/inc/tq.h +++ b/source/dnode/vnode/src/inc/tq.h @@ -89,7 +89,7 @@ typedef struct { STqExecDb execDb; }; int32_t numOfCols; // number of out pout column, temporarily used - SSchemaWrapper *pSchemaWrapper; // columns that are involved in query + SSchemaWrapper* pSchemaWrapper; // columns that are involved in query } STqExecHandle; typedef struct { @@ -120,9 +120,9 @@ struct STQ { SHashObj* pushMgr; // consumerId -> STqHandle* SHashObj* handles; // subKey -> STqHandle SHashObj* pStreamTasks; // taksId -> SStreamTask + SHashObj* pAlterInfo; // topic -> SAlterCheckInfo STqOffsetStore* pOffsetStore; SVnode* pVnode; - SWal* pWal; TDB* pMetaStore; TTB* pExecStore; }; diff --git a/source/dnode/vnode/src/inc/vnodeInt.h b/source/dnode/vnode/src/inc/vnodeInt.h index 5e87e35d68..12dc4e94db 100644 --- a/source/dnode/vnode/src/inc/vnodeInt.h +++ b/source/dnode/vnode/src/inc/vnodeInt.h @@ -137,12 +137,13 @@ STsdbReader tsdbQueryCacheLastT(STsdb* tsdb, SQueryTableDataCond* pCond, STableL // tq int tqInit(); void tqCleanUp(); -STQ* tqOpen(const char* path, SVnode* pVnode, SWal* pWal); +STQ* tqOpen(const char* path, SVnode* pVnode); void tqClose(STQ*); int tqPushMsg(STQ*, void* msg, int32_t msgLen, tmsg_t msgType, int64_t ver); int tqCommit(STQ*); int32_t tqUpdateTbUidList(STQ* pTq, const SArray* tbUidList, bool isAdd); -int32_t tqCheckColModifiable(STQ* pTq, int32_t colId); +int32_t tqCheckColModifiable(STQ* pTq, int64_t tbUid, int32_t colId); +int32_t tqProcessCheckAlterInfoReq(STQ* pTq, char* msg, int32_t msgLen); int32_t tqProcessVgChangeReq(STQ* pTq, char* msg, int32_t msgLen); int32_t tqProcessVgDeleteReq(STQ* pTq, char* msg, int32_t msgLen); int32_t tqProcessOffsetCommitReq(STQ* pTq, char* msg, int32_t msgLen); diff --git a/source/dnode/vnode/src/meta/metaTable.c b/source/dnode/vnode/src/meta/metaTable.c index 5255724b58..e2453e4f63 100644 --- a/source/dnode/vnode/src/meta/metaTable.c +++ b/source/dnode/vnode/src/meta/metaTable.c @@ -637,6 +637,10 @@ static int metaAlterTableColumn(SMeta *pMeta, int64_t version, SVAlterTbReq *pAl terrno = TSDB_CODE_VND_INVALID_TABLE_ACTION; goto _err; } + if (tqCheckColModifiable(pMeta->pVnode->pTq, uid, pColumn->colId) != 0) { + terrno = TSDB_CODE_VND_COL_SUBSCRIBED; + goto _err; + } pSchema->version++; tlen = (pSchema->nCols - iCol - 1) * sizeof(SSchema); if (tlen) { @@ -653,6 +657,10 @@ static int metaAlterTableColumn(SMeta *pMeta, int64_t version, SVAlterTbReq *pAl terrno = TSDB_CODE_VND_INVALID_TABLE_ACTION; goto _err; } + if (tqCheckColModifiable(pMeta->pVnode->pTq, uid, pColumn->colId) != 0) { + terrno = TSDB_CODE_VND_COL_SUBSCRIBED; + goto _err; + } pSchema->version++; pColumn->bytes = pAlterTbReq->colModBytes; break; @@ -661,6 +669,10 @@ static int metaAlterTableColumn(SMeta *pMeta, int64_t version, SVAlterTbReq *pAl terrno = TSDB_CODE_VND_TABLE_COL_NOT_EXISTS; goto _err; } + if (tqCheckColModifiable(pMeta->pVnode->pTq, uid, pColumn->colId) != 0) { + terrno = TSDB_CODE_VND_COL_SUBSCRIBED; + goto _err; + } pSchema->version++; strcpy(pColumn->name, pAlterTbReq->colNewName); break; diff --git a/source/dnode/vnode/src/tq/tq.c b/source/dnode/vnode/src/tq/tq.c index 3739897ec0..8c1f13d2a8 100644 --- a/source/dnode/vnode/src/tq/tq.c +++ b/source/dnode/vnode/src/tq/tq.c @@ -51,7 +51,7 @@ void tqCleanUp() { } } -STQ* tqOpen(const char* path, SVnode* pVnode, SWal* pWal) { +STQ* tqOpen(const char* path, SVnode* pVnode) { STQ* pTq = taosMemoryCalloc(1, sizeof(STQ)); if (pTq == NULL) { terrno = TSDB_CODE_TQ_OUT_OF_MEMORY; @@ -59,7 +59,6 @@ STQ* tqOpen(const char* path, SVnode* pVnode, SWal* pWal) { } pTq->path = strdup(path); pTq->pVnode = pVnode; - pTq->pWal = pWal; pTq->handles = taosHashInit(64, MurmurHash3_32, true, HASH_ENTRY_LOCK); @@ -67,6 +66,8 @@ STQ* tqOpen(const char* path, SVnode* pVnode, SWal* pWal) { pTq->pushMgr = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_ENTRY_LOCK); + pTq->pAlterInfo = taosHashInit(64, MurmurHash3_32, true, HASH_ENTRY_LOCK); + if (tqMetaOpen(pTq) < 0) { ASSERT(0); } @@ -91,6 +92,7 @@ void tqClose(STQ* pTq) { } taosHashCleanup(pTq->pStreamTasks); taosHashCleanup(pTq->pushMgr); + taosHashCleanup(pTq->pAlterInfo); taosMemoryFree(pTq->path); tqMetaClose(pTq); taosMemoryFree(pTq); @@ -208,18 +210,18 @@ int32_t tqProcessOffsetCommitReq(STQ* pTq, char* msg, int32_t msgLen) { return 0; } -int32_t tqCheckColModifiable(STQ* pTq, int32_t colId) { +int32_t tqCheckColModifiable(STQ* pTq, int64_t tbUid, int32_t colId) { void* pIter = NULL; while (1) { - pIter = taosHashIterate(pTq->handles, pIter); + pIter = taosHashIterate(pTq->pAlterInfo, pIter); if (pIter == NULL) break; - STqHandle* pExec = (STqHandle*)pIter; - if (pExec->execHandle.subType == TOPIC_SUB_TYPE__COLUMN) { - int32_t sz = pExec->execHandle.pSchemaWrapper->nCols; + SCheckAlterInfo* pCheck = (SCheckAlterInfo*)pIter; + if (pCheck->ntbUid == tbUid) { + int32_t sz = taosArrayGetSize(pCheck->colIdList); for (int32_t i = 0; i < sz; i++) { - SSchema* pSchema = &pExec->execHandle.pSchemaWrapper->pSchema[i]; - if (pSchema->colId == colId) { - taosHashCancelIterate(pTq->handles, pIter); + int16_t forbidColId = *(int16_t*)taosArrayGet(pCheck->colIdList, i); + if (forbidColId == colId) { + taosHashCancelIterate(pTq->pAlterInfo, pIter); return -1; } } @@ -488,6 +490,22 @@ int32_t tqProcessVgDeleteReq(STQ* pTq, char* msg, int32_t msgLen) { return 0; } +int32_t tqProcessCheckAlterInfoReq(STQ* pTq, char* msg, int32_t msgLen) { + SCheckAlterInfo info = {0}; + SDecoder decoder; + tDecoderInit(&decoder, msg, msgLen); + if (tDecodeSCheckAlterInfo(&decoder, &info) < 0) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + return -1; + } + tDecoderClear(&decoder); + if (taosHashPut(pTq->pAlterInfo, info.topic, strlen(info.topic), &info, sizeof(SCheckAlterInfo)) < 0) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + return -1; + } + return 0; +} + int32_t tqProcessVgChangeReq(STQ* pTq, char* msg, int32_t msgLen) { SMqRebVgReq req = {0}; tDecodeSMqRebVgReq(msg, &req); @@ -524,7 +542,7 @@ int32_t tqProcessVgChangeReq(STQ* pTq, char* msg, int32_t msgLen) { }; pHandle->execHandle.execCol.task = qCreateQueueExecTaskInfo(pHandle->execHandle.execCol.qmsg, &handle, &pHandle->execHandle.numOfCols, - &pHandle->execHandle.pSchemaWrapper); + &pHandle->execHandle.pSchemaWrapper, &pHandle->ntbUid); ASSERT(pHandle->execHandle.execCol.task); void* scanner = NULL; qExtractStreamScanner(pHandle->execHandle.execCol.task, &scanner); diff --git a/source/dnode/vnode/src/tq/tqMeta.c b/source/dnode/vnode/src/tq/tqMeta.c index 835ffb02fd..4ee87e3203 100644 --- a/source/dnode/vnode/src/tq/tqMeta.c +++ b/source/dnode/vnode/src/tq/tqMeta.c @@ -90,7 +90,8 @@ int32_t tqMetaOpen(STQ* pTq) { }; handle.execHandle.execCol.task = - qCreateQueueExecTaskInfo(handle.execHandle.execCol.qmsg, &reader, &handle.execHandle.numOfCols, &handle.execHandle.pSchemaWrapper); + qCreateQueueExecTaskInfo(handle.execHandle.execCol.qmsg, &reader, &handle.execHandle.numOfCols, + &handle.execHandle.pSchemaWrapper, &handle.ntbUid); ASSERT(handle.execHandle.execCol.task); void* scanner = NULL; qExtractStreamScanner(handle.execHandle.execCol.task, &scanner); diff --git a/source/dnode/vnode/src/tq/tqSink.c b/source/dnode/vnode/src/tq/tqSink.c index 464a3a3ee1..850394b15d 100644 --- a/source/dnode/vnode/src/tq/tqSink.c +++ b/source/dnode/vnode/src/tq/tqSink.c @@ -110,7 +110,6 @@ SSubmitReq* tdBlockToSubmit(const SArray* pBlocks, const STSchema* pTSchema, boo // TODO ret = rpcMallocCont(cap); ret->header.vgId = vgId; - ret->version = htonl(1); ret->length = sizeof(SSubmitReq); ret->numOfBlocks = htonl(sz); diff --git a/source/dnode/vnode/src/vnd/vnodeOpen.c b/source/dnode/vnode/src/vnd/vnodeOpen.c index fb2ac722a6..e273a41f48 100644 --- a/source/dnode/vnode/src/vnd/vnodeOpen.c +++ b/source/dnode/vnode/src/vnd/vnodeOpen.c @@ -135,7 +135,7 @@ SVnode *vnodeOpen(const char *path, STfs *pTfs, SMsgCb msgCb) { // open tq sprintf(tdir, "%s%s%s", dir, TD_DIRSEP, VNODE_TQ_DIR); taosRealPath(tdir, NULL, sizeof(tdir)); - pVnode->pTq = tqOpen(tdir, pVnode, pVnode->pWal); + pVnode->pTq = tqOpen(tdir, pVnode); if (pVnode->pTq == NULL) { vError("vgId:%d, failed to open vnode tq since %s", TD_VID(pVnode), tstrerror(terrno)); goto _err; diff --git a/source/dnode/vnode/src/vnd/vnodeSvr.c b/source/dnode/vnode/src/vnd/vnodeSvr.c index 9929258df0..33e3ec275f 100644 --- a/source/dnode/vnode/src/vnd/vnodeSvr.c +++ b/source/dnode/vnode/src/vnd/vnodeSvr.c @@ -206,6 +206,12 @@ int32_t vnodeProcessWriteMsg(SVnode *pVnode, SRpcMsg *pMsg, int64_t version, SRp goto _err; } break; + case TDMT_VND_CHECK_ALTER_INFO: + if (tqProcessCheckAlterInfoReq(pVnode->pTq, POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)), + pMsg->contLen - sizeof(SMsgHead)) < 0) { + goto _err; + } + break; case TDMT_STREAM_TASK_DEPLOY: { if (tqProcessTaskDeployReq(pVnode->pTq, POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)), pMsg->contLen - sizeof(SMsgHead)) < 0) { diff --git a/source/libs/executor/inc/executorimpl.h b/source/libs/executor/inc/executorimpl.h index b36a5ebdd1..db2f9d6bed 100644 --- a/source/libs/executor/inc/executorimpl.h +++ b/source/libs/executor/inc/executorimpl.h @@ -157,6 +157,7 @@ typedef struct { SQueryTableDataCond tableCond; int64_t recoverStartVer; int64_t recoverEndVer; + int64_t ntbUid; } SStreamTaskInfo; typedef struct { diff --git a/source/libs/executor/src/executor.c b/source/libs/executor/src/executor.c index c5aa90e0eb..e33e115ec9 100644 --- a/source/libs/executor/src/executor.c +++ b/source/libs/executor/src/executor.c @@ -121,7 +121,7 @@ int32_t qSetMultiStreamInput(qTaskInfo_t tinfo, const void* pBlocks, size_t numO } qTaskInfo_t qCreateQueueExecTaskInfo(void* msg, SReadHandle* readers, int32_t* numOfCols, - SSchemaWrapper** pSchemaWrapper) { + SSchemaWrapper** pSchemaWrapper, int64_t* ntbUid) { if (msg == NULL) { // TODO create raw scan return NULL; @@ -156,6 +156,7 @@ qTaskInfo_t qCreateQueueExecTaskInfo(void* msg, SReadHandle* readers, int32_t* n } *pSchemaWrapper = tCloneSSchemaWrapper(((SExecTaskInfo*)pTaskInfo)->schemaInfo.qsw); + *ntbUid = ((SExecTaskInfo*)pTaskInfo)->streamInfo.ntbUid; return pTaskInfo; } diff --git a/source/libs/executor/src/executorMain.c b/source/libs/executor/src/executorMain.c index 27d8b3f1cc..d4b543f5f8 100644 --- a/source/libs/executor/src/executorMain.c +++ b/source/libs/executor/src/executorMain.c @@ -315,6 +315,9 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, const STqOffsetVal* pOffset) { if (type == QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) { SStreamScanInfo* pInfo = pOperator->info; if (pOffset->type == TMQ_OFFSET__LOG) { + STableScanInfo* pTSInfo = pInfo->pTableScanOp->info; + tsdbReaderClose(pTSInfo->dataReader); + pTSInfo->dataReader = NULL; #if 0 if (tOffsetEqual(pOffset, &pTaskInfo->streamInfo.lastStatus) && pInfo->tqReader->pWalReader->curVersion != pOffset->version) { @@ -348,9 +351,8 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, const STqOffsetVal* pOffset) { int32_t tableSz = taosArrayGetSize(pTaskInfo->tableqinfoList.pTableList); #ifndef NDEBUG - - qDebug("switch to next table %ld (cursor %d), %ld rows returned", uid, - pTableScanInfo->currentTable, pInfo->pTableScanOp->resultInfo.totalRows); + qDebug("switch to next table %ld (cursor %d), %ld rows returned", uid, pTableScanInfo->currentTable, + pInfo->pTableScanOp->resultInfo.totalRows); pInfo->pTableScanOp->resultInfo.totalRows = 0; #endif @@ -367,6 +369,14 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, const STqOffsetVal* pOffset) { // TODO after dropping table, table may be not found ASSERT(found); + if (pTableScanInfo->dataReader == NULL) { + if (tsdbReaderOpen(pTableScanInfo->readHandle.vnode, &pTableScanInfo->cond, + pTaskInfo->tableqinfoList.pTableList, &pTableScanInfo->dataReader, NULL) < 0 || + pTableScanInfo->dataReader == NULL) { + ASSERT(0); + } + } + tsdbSetTableId(pTableScanInfo->dataReader, uid); int64_t oldSkey = pTableScanInfo->cond.twindows.skey; pTableScanInfo->cond.twindows.skey = ts + 1; diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c index 07bd368c90..cf0fce8c70 100644 --- a/source/libs/executor/src/executorimpl.c +++ b/source/libs/executor/src/executorimpl.c @@ -4169,6 +4169,9 @@ int32_t extractTableSchemaInfo(SReadHandle* pHandle, SScanPhysiNode* pScanNode, pSchemaInfo->tversion = mr.me.stbEntry.schemaTag.version; } else { pSchemaInfo->sw = tCloneSSchemaWrapper(&mr.me.ntbEntry.schemaRow); + if (pTaskInfo->execModel == OPTR_EXEC_MODEL_QUEUE) { + pTaskInfo->streamInfo.ntbUid = mr.me.uid; + } } metaReaderClear(&mr); diff --git a/source/libs/executor/src/scanoperator.c b/source/libs/executor/src/scanoperator.c index 574aa648e5..6486749303 100644 --- a/source/libs/executor/src/scanoperator.c +++ b/source/libs/executor/src/scanoperator.c @@ -982,6 +982,9 @@ static SSDataBlock* doRangeScan(SStreamScanInfo* pInfo, SSDataBlock* pSDB, int32 if (!pResult) { blockDataCleanup(pSDB); *pRowIndex = 0; + STableScanInfo* pTableScanInfo = pInfo->pTableScanOp->info; + tsdbReaderClose(pTableScanInfo->dataReader); + pTableScanInfo->dataReader = NULL; return NULL; } @@ -1003,6 +1006,9 @@ static SSDataBlock* doDataScan(SStreamScanInfo* pInfo, SSDataBlock* pSDB, int32_ } if (!pResult) { pInfo->updateWin = (STimeWindow){.skey = INT64_MIN, .ekey = INT64_MAX}; + STableScanInfo* pTableScanInfo = pInfo->pTableScanOp->info; + tsdbReaderClose(pTableScanInfo->dataReader); + pTableScanInfo->dataReader = NULL; return NULL; } @@ -2051,8 +2057,8 @@ static SSDataBlock* sysTableScanUserTables(SOperatorInfo* pOperator) { uint64_t suid = pInfo->pCur->mr.me.ctbEntry.suid; int32_t code = metaGetTableEntryByUid(&mr, suid); if (code != TSDB_CODE_SUCCESS) { - qError("failed to get super table meta, cname:%s, suid:0x%" PRIx64 ", code:%s, %s", - pInfo->pCur->mr.me.name, suid, tstrerror(terrno), GET_TASKID(pTaskInfo)); + qError("failed to get super table meta, cname:%s, suid:0x%" PRIx64 ", code:%s, %s", pInfo->pCur->mr.me.name, + suid, tstrerror(terrno), GET_TASKID(pTaskInfo)); metaReaderClear(&mr); metaCloseTbCursor(pInfo->pCur); pInfo->pCur = NULL; @@ -2158,7 +2164,6 @@ static SSDataBlock* sysTableScanUserTables(SOperatorInfo* pOperator) { } } - static SSDataBlock* sysTableScanUserSTables(SOperatorInfo* pOperator) { SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; SSysTableScanInfo* pInfo = pOperator->info; @@ -2184,12 +2189,13 @@ static SSDataBlock* doSysTableScan(SOperatorInfo* pOperator) { getDBNameFromCondition(pInfo->pCondition, dbName); sprintf(pInfo->req.db, "%d.%s", pInfo->accountId, dbName); } - + if (strncasecmp(name, TSDB_INS_TABLE_USER_TABLES, TSDB_TABLE_FNAME_LEN) == 0) { return sysTableScanUserTables(pOperator); } else if (strncasecmp(name, TSDB_INS_TABLE_USER_TAGS, TSDB_TABLE_FNAME_LEN) == 0) { return sysTableScanUserTags(pOperator); - } else if (strncasecmp(name, TSDB_INS_TABLE_USER_STABLES, TSDB_TABLE_FNAME_LEN) == 0 && IS_SYS_DBNAME(pInfo->req.db)) { + } else if (strncasecmp(name, TSDB_INS_TABLE_USER_STABLES, TSDB_TABLE_FNAME_LEN) == 0 && + IS_SYS_DBNAME(pInfo->req.db)) { return sysTableScanUserSTables(pOperator); } else { // load the meta from mnode of the given epset if (pOperator->status == OP_EXEC_DONE) { diff --git a/source/util/src/terror.c b/source/util/src/terror.c index 184b1a9edf..927cb65f2f 100644 --- a/source/util/src/terror.c +++ b/source/util/src/terror.c @@ -336,6 +336,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_VND_TABLE_NOT_EXIST, "Table does not exists TAOS_DEFINE_ERROR(TSDB_CODE_VND_INVALID_TABLE_ACTION, "Invalid table action") TAOS_DEFINE_ERROR(TSDB_CODE_VND_COL_ALREADY_EXISTS, "Table column already exists") TAOS_DEFINE_ERROR(TSDB_CODE_VND_TABLE_COL_NOT_EXISTS, "Table column not exists") +TAOS_DEFINE_ERROR(TSDB_CODE_VND_COL_SUBSCRIBED, "Table column is subscribed") // tsdb TAOS_DEFINE_ERROR(TSDB_CODE_TDB_INVALID_TABLE_ID, "Invalid table ID") diff --git a/tests/pytest/topic/topicQuery.py b/tests/pytest/topic/topicQuery.py deleted file mode 100644 index 1ee3c3a427..0000000000 --- a/tests/pytest/topic/topicQuery.py +++ /dev/null @@ -1,91 +0,0 @@ -################################################################### -# Copyright (c) 2016 by TAOS Technologies, Inc. -# All rights reserved. -# -# This file is proprietary and confidential to TAOS Technologies. -# No part of this file may be reproduced, stored, transmitted, -# disclosed or used in any form or by any means other than as -# expressly provided by the written permission from Jianhui Tao -# -################################################################### - -# -*- coding: utf-8 -*- - -from util.log import tdLog -from util.cases import tdCases -from util.sql import tdSql - -class TDTestCase: - def init(self, conn, logSql): - tdLog.debug("start to execute %s" % __file__) - tdSql.init(conn.cursor(), logSql) - - self.ts = 1538548685000 - - def run(self): - tdSql.prepare() - - # test case for https://jira.taosdata.com:18080/browse/TD-3679 - print("==============step1") - tdSql.execute( - "create topic tq_test partitions 10") - tdSql.execute( - "insert into tq_test.p1(off, ts, content) values(0, %d, 'aaaa')" % self.ts) - tdSql.execute( - "insert into tq_test.p1(off, ts, content) values(1, %d, 'aaaa')" % (self.ts + 1)) - tdSql.execute( - "insert into tq_test.p1(off, ts, content) values(2, %d, 'aaaa')" % (self.ts + 2)) - tdSql.execute( - "insert into tq_test.p1(off, ts, content) values(3, %d, 'aaaa')" % (self.ts + 3)) - - print("==============step2") - - tdSql.query("select * from tq_test.p1") - tdSql.checkRows(4) - - tdSql.query("select * from tq_test.p1 where ts >= %d" % self.ts) - tdSql.checkRows(4) - - tdSql.query("select * from tq_test.p1 where ts > %d" % self.ts) - tdSql.checkRows(3) - - tdSql.query("select * from tq_test.p1 where ts = %d" % self.ts) - tdSql.checkRows(1) - - - tdSql.execute("use db") - tdSql.execute("create table test(ts timestamp, start timestamp, value int)") - tdSql.execute("insert into test values(%d, %d, 1)" % (self.ts, self.ts)) - tdSql.execute("insert into test values(%d, %d, 1)" % (self.ts + 1, self.ts + 1)) - tdSql.execute("insert into test values(%d, %d, 1)" % (self.ts + 2, self.ts + 2)) - tdSql.execute("insert into test values(%d, %d, 1)" % (self.ts + 3, self.ts + 3)) - - tdSql.query("select * from test") - tdSql.checkRows(4) - - tdSql.query("select * from test where ts >= %d" % self.ts) - tdSql.checkRows(4) - - tdSql.query("select * from test where ts > %d" % self.ts) - tdSql.checkRows(3) - - tdSql.query("select * from test where ts = %d" % self.ts) - tdSql.checkRows(1) - - tdSql.query("select * from test where start >= %d" % self.ts) - tdSql.checkRows(4) - - tdSql.query("select * from test where start > %d" % self.ts) - tdSql.checkRows(3) - - tdSql.query("select * from test where start = %d" % self.ts) - tdSql.checkRows(1) - - - def stop(self): - tdSql.close() - tdLog.success("%s successfully executed" % __file__) - - -tdCases.addWindows(__file__, TDTestCase()) -tdCases.addLinux(__file__, TDTestCase()) \ No newline at end of file diff --git a/tests/system-test/7-tmq/schema.py b/tests/system-test/7-tmq/schema.py index a3462feb14..faa9695c02 100644 --- a/tests/system-test/7-tmq/schema.py +++ b/tests/system-test/7-tmq/schema.py @@ -830,9 +830,9 @@ class TDTestCase: cfgPath = buildPath + "/../sim/psim/cfg" tdLog.info("cfgPath: %s" % cfgPath) - # self.tmqCase1(cfgPath, buildPath) - # self.tmqCase2(cfgPath, buildPath) - # self.tmqCase3(cfgPath, buildPath) + self.tmqCase1(cfgPath, buildPath) + self.tmqCase2(cfgPath, buildPath) + self.tmqCase3(cfgPath, buildPath) self.tmqCase4(cfgPath, buildPath) self.tmqCase5(cfgPath, buildPath) diff --git a/tests/test/c/CMakeLists.txt b/tests/test/c/CMakeLists.txt index 505c290f2a..5db97a0f0f 100644 --- a/tests/test/c/CMakeLists.txt +++ b/tests/test/c/CMakeLists.txt @@ -28,6 +28,7 @@ target_link_libraries( sdbDump PUBLIC dnode PUBLIC mnode + PUBLIC stream PUBLIC sdb PUBLIC os ) @@ -37,4 +38,4 @@ target_include_directories( PRIVATE "${TD_SOURCE_DIR}/source/dnode/mnode/impl/inc" PRIVATE "${TD_SOURCE_DIR}/source/dnode/mnode/sdb/inc" PRIVATE "${TD_SOURCE_DIR}/source/dnode/mgmt/node_mgmt/inc" -) \ No newline at end of file +) -- GitLab