From e8d53c2a96cae26483243ffdae27f646684e545d Mon Sep 17 00:00:00 2001 From: wangmm0220 Date: Tue, 27 Jun 2023 10:50:57 +0800 Subject: [PATCH 01/11] fix:add log --- source/client/src/clientTmq.c | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/source/client/src/clientTmq.c b/source/client/src/clientTmq.c index 8758cec2ec..88bd1e87e9 100644 --- a/source/client/src/clientTmq.c +++ b/source/client/src/clientTmq.c @@ -813,7 +813,7 @@ void tmqSendHbReq(void* param, void* tmrId) { offRows->offset = pVg->offsetInfo.currentOffset; char buf[TSDB_OFFSET_LEN] = {0}; tFormatOffset(buf, TSDB_OFFSET_LEN, &offRows->offset); - tscInfo("report offset: vgId:%d, offset:%s, rows:%"PRId64, offRows->vgId, buf, offRows->rows); + tscInfo("consumer:0x%" PRIx64 ",report offset: vgId:%d, offset:%s, rows:%"PRId64, tmq->consumerId, offRows->vgId, buf, offRows->rows); } } // tmq->needReportOffsetRows = false; @@ -1489,7 +1489,8 @@ static void initClientTopicFromRsp(SMqClientTopic* pTopic, SMqSubTopicEp* pTopic makeTopicVgroupKey(vgKey, pTopic->topicName, pVgEp->vgId); SVgroupSaveInfo* pInfo = taosHashGet(pVgOffsetHashMap, vgKey, strlen(vgKey)); - STqOffsetVal offsetNew = {.type = tmq->resetOffsetCfg}; + STqOffsetVal offsetNew = {0}; + offsetNew.type = tmq->resetOffsetCfg; SMqClientVg clientVg = { .pollCnt = 0, From 1fc664693e9f956565e5872d77c8074fd4d09d1b Mon Sep 17 00:00:00 2001 From: wangmm0220 Date: Wed, 28 Jun 2023 11:54:48 +0800 Subject: [PATCH 02/11] fix:continue consume if wal is dropped --- source/dnode/vnode/src/tq/tqRead.c | 10 +++++----- source/libs/executor/src/executor.c | 10 ++++++++++ 2 files changed, 15 insertions(+), 5 deletions(-) diff --git a/source/dnode/vnode/src/tq/tqRead.c b/source/dnode/vnode/src/tq/tqRead.c index 77a966715e..5f53f1c50c 100644 --- a/source/dnode/vnode/src/tq/tqRead.c +++ b/source/dnode/vnode/src/tq/tqRead.c @@ -388,7 +388,7 @@ bool tqNextBlockInWal(STqReader* pReader, const char* id) { int32_t numOfBlocks = taosArrayGetSize(pReader->submit.aSubmitTbData); while (pReader->nextBlk < numOfBlocks) { - tqDebug("tq reader next data block %d/%d, len:%d %" PRId64 " %d", pReader->nextBlk, + tqTrace("tq reader next data block %d/%d, len:%d %" PRId64 " %d", pReader->nextBlk, numOfBlocks, pReader->msg.msgLen, pReader->msg.ver, pReader->nextBlk); SSubmitTbData* pSubmitTbData = taosArrayGet(pReader->submit.aSubmitTbData, pReader->nextBlk); @@ -403,7 +403,7 @@ bool tqNextBlockInWal(STqReader* pReader, const char* id) { void* ret = taosHashGet(pReader->tbIdHash, &pSubmitTbData->uid, sizeof(int64_t)); if (ret != NULL) { - tqDebug("tq reader return submit block, uid:%" PRId64 ", ver:%" PRId64, pSubmitTbData->uid, pReader->msg.ver); + tqTrace("tq reader return submit block, uid:%" PRId64 ", ver:%" PRId64, pSubmitTbData->uid, pReader->msg.ver); SSDataBlock* pRes = NULL; int32_t code = tqRetrieveDataBlock(pReader, &pRes, NULL); @@ -412,11 +412,11 @@ bool tqNextBlockInWal(STqReader* pReader, const char* id) { } } else { pReader->nextBlk += 1; - tqDebug("tq reader discard submit block, uid:%" PRId64 ", continue", pSubmitTbData->uid); + tqTrace("tq reader discard submit block, uid:%" PRId64 ", continue", pSubmitTbData->uid); } } - qDebug("stream scan return empty, all %d submit blocks consumed, %s", numOfBlocks, id); + qTrace("stream scan return empty, all %d submit blocks consumed, %s", numOfBlocks, id); tDestroySubmitReq(&pReader->submit, TSDB_MSG_FLG_DECODE); pReader->msg.msgStr = NULL; @@ -604,7 +604,7 @@ static int32_t doSetVal(SColumnInfoData* pColumnInfoData, int32_t rowIndex, SCol } int32_t tqRetrieveDataBlock(STqReader* pReader, SSDataBlock** pRes, const char* id) { - tqDebug("tq reader retrieve data block %p, index:%d", pReader->msg.msgStr, pReader->nextBlk); + tqTrace("tq reader retrieve data block %p, index:%d", pReader->msg.msgStr, pReader->nextBlk); SSubmitTbData* pSubmitTbData = taosArrayGet(pReader->submit.aSubmitTbData, pReader->nextBlk++); SSDataBlock* pBlock = pReader->pResBlock; diff --git a/source/libs/executor/src/executor.c b/source/libs/executor/src/executor.c index c8b66836d5..88e2165a12 100644 --- a/source/libs/executor/src/executor.c +++ b/source/libs/executor/src/executor.c @@ -1078,6 +1078,16 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, STqOffsetVal* pOffset, int8_t subT SOperatorInfo* pOperator = pTaskInfo->pRoot; const char* id = GET_TASKID(pTaskInfo); + if(subType == TOPIC_SUB_TYPE__COLUMN && pOffset->type == TMQ_OFFSET__LOG){ + pOperator = extractOperatorInTree(pOperator, QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN, id); + if (pOperator == NULL) { + return -1; + } + SStreamScanInfo* pInfo = pOperator->info; + SStoreTqReader* pReaderAPI = &pTaskInfo->storageAPI.tqReaderFn; + SWalReader* pWalReader = pReaderAPI->tqReaderGetWalReader(pInfo->tqReader); + walReaderVerifyOffset(pWalReader, pOffset); + } // if pOffset equal to current offset, means continue consume if (tOffsetEqual(pOffset, &pTaskInfo->streamInfo.currentOffset)) { return 0; From 4a17f4b9f54a161b6452fec1164f164d422e2085 Mon Sep 17 00:00:00 2001 From: wangmm0220 Date: Wed, 28 Jun 2023 14:57:31 +0800 Subject: [PATCH 03/11] fix:add rows to pSub if rebalance --- source/dnode/mnode/impl/src/mndConsumer.c | 2 +- source/dnode/mnode/impl/src/mndSubscribe.c | 40 ++++++++++++---------- 2 files changed, 23 insertions(+), 19 deletions(-) diff --git a/source/dnode/mnode/impl/src/mndConsumer.c b/source/dnode/mnode/impl/src/mndConsumer.c index 4dded61ce3..a13f41d03c 100644 --- a/source/dnode/mnode/impl/src/mndConsumer.c +++ b/source/dnode/mnode/impl/src/mndConsumer.c @@ -1190,7 +1190,7 @@ static int32_t mndRetrieveConsumer(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock * tFormatOffset(buf, TSDB_OFFSET_LEN, &pVal); char parasStr[64 + TSDB_OFFSET_LEN + VARSTR_HEADER_SIZE] = {0}; - sprintf(varDataVal(parasStr), "tbname:%d,commit:%d,interval:%d,reset:%s", pConsumer->withTbName, pConsumer->autoCommit, pConsumer->autoCommitInterval, buf); + sprintf(varDataVal(parasStr), "tbname:%d,commit:%d,interval:%dms,reset:%s", pConsumer->withTbName, pConsumer->autoCommit, pConsumer->autoCommitInterval, buf); varDataSetLen(parasStr, strlen(varDataVal(parasStr))); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); diff --git a/source/dnode/mnode/impl/src/mndSubscribe.c b/source/dnode/mnode/impl/src/mndSubscribe.c index 61691a30d5..f1b870cbf3 100644 --- a/source/dnode/mnode/impl/src/mndSubscribe.c +++ b/source/dnode/mnode/impl/src/mndSubscribe.c @@ -468,7 +468,7 @@ static int32_t mndDoRebalance(SMnode *pMnode, const SMqRebInputObj *pInput, SMqR } } - if(taosHashGetSize(pOutput->pSub->consumerHash) == 0) { // if all consumer is removed +// if(taosHashGetSize(pOutput->pSub->consumerHash) == 0) { // if all consumer is removed SMqSubscribeObj *pSub = mndAcquireSubscribeByKey(pMnode, pInput->pRebInfo->key); // put all offset rows if (pSub) { taosRLockLatch(&pSub->lock); @@ -501,7 +501,7 @@ static int32_t mndDoRebalance(SMnode *pMnode, const SMqRebInputObj *pInput, SMqR } taosRUnLockLatch(&pSub->lock); mndReleaseSubscribe(pMnode, pSub); - } +// } } // 8. generate logs @@ -771,8 +771,10 @@ static int32_t mndProcessRebalanceReq(SRpcMsg *pMsg) { } static int32_t mndProcessDropCgroupReq(SRpcMsg *pMsg) { - SMnode *pMnode = pMsg->info.node; - SMDropCgroupReq dropReq = {0}; + SMnode *pMnode = pMsg->info.node; + SMDropCgroupReq dropReq = {0}; + STrans *pTrans = NULL; + int32_t code = TSDB_CODE_ACTION_IN_PROGRESS; if (tDeserializeSMDropCgroupReq(pMsg->pCont, pMsg->contLen, &dropReq) != 0) { terrno = TSDB_CODE_INVALID_MSG; @@ -791,38 +793,40 @@ static int32_t mndProcessDropCgroupReq(SRpcMsg *pMsg) { } } + taosWLockLatch(&pSub->lock); if (taosHashGetSize(pSub->consumerHash) != 0) { terrno = TSDB_CODE_MND_CGROUP_USED; mError("cgroup:%s on topic:%s, failed to drop since %s", dropReq.cgroup, dropReq.topic, terrstr()); - mndReleaseSubscribe(pMnode, pSub); - return -1; + code = -1; + goto end; } - STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_NOTHING, pMsg, "drop-cgroup"); + pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_NOTHING, pMsg, "drop-cgroup"); if (pTrans == NULL) { mError("cgroup: %s on topic:%s, failed to drop since %s", dropReq.cgroup, dropReq.topic, terrstr()); - mndReleaseSubscribe(pMnode, pSub); - mndTransDrop(pTrans); - return -1; + code = -1; + goto end; } mInfo("trans:%d, used to drop cgroup:%s on topic %s", pTrans->id, dropReq.cgroup, dropReq.topic); if (mndSetDropSubCommitLogs(pMnode, pTrans, pSub) < 0) { mError("cgroup %s on topic:%s, failed to drop since %s", dropReq.cgroup, dropReq.topic, terrstr()); - mndReleaseSubscribe(pMnode, pSub); - mndTransDrop(pTrans); - return -1; + code = -1; + goto end; } if (mndTransPrepare(pMnode, pTrans) < 0) { - mndReleaseSubscribe(pMnode, pSub); - mndTransDrop(pTrans); - return -1; + code = -1; + goto end; } - mndReleaseSubscribe(pMnode, pSub); - return TSDB_CODE_ACTION_IN_PROGRESS; +end: + taosWUnLockLatch(&pSub->lock); + mndReleaseSubscribe(pMnode, pSub); + mndTransDrop(pTrans); + + return code; } void mndCleanupSubscribe(SMnode *pMnode) {} From ce0b71c6349fabcc88bf6fdc6d0a93bd1c484762 Mon Sep 17 00:00:00 2001 From: wangmm0220 Date: Wed, 28 Jun 2023 15:57:30 +0800 Subject: [PATCH 04/11] fix:add rows to pSub if rebalance --- source/dnode/mnode/impl/src/mndSubscribe.c | 16 ++++++++++++++-- 1 file changed, 14 insertions(+), 2 deletions(-) diff --git a/source/dnode/mnode/impl/src/mndSubscribe.c b/source/dnode/mnode/impl/src/mndSubscribe.c index f1b870cbf3..cfc164c26c 100644 --- a/source/dnode/mnode/impl/src/mndSubscribe.c +++ b/source/dnode/mnode/impl/src/mndSubscribe.c @@ -484,16 +484,28 @@ static int32_t mndDoRebalance(SMnode *pMnode, const SMqRebInputObj *pInput, SMqR SMqConsumerEp *pConsumerEp = (SMqConsumerEp *)pIter; if (init) { taosArrayAddAll(pOutput->pSub->offsetRows, pConsumerEp->offsetRows); -// mDebug("pSub->offsetRows is init"); + mInfo("pSub->offsetRows is init"); } else { + SMqConsumerEp *pConsumerEpNew = taosHashGet(pOutput->pSub->consumerHash, &pConsumerEp->consumerId, sizeof(int64_t)); + for (int j = 0; j < taosArrayGetSize(pConsumerEp->offsetRows); j++) { OffsetRows *d1 = taosArrayGet(pConsumerEp->offsetRows, j); + bool jump = false; + for (int i = 0; pConsumerEpNew && i < taosArrayGetSize(pConsumerEpNew->vgs); i++){ + SMqVgEp *pVgEp = taosArrayGetP(pConsumerEpNew->vgs, j); + if(pVgEp->vgId == d1->vgId){ + jump = true; + mInfo("pSub->offsetRows jump, because consumer id:%"PRIx64 " and vgId:%d not change", pConsumerEp->consumerId, pVgEp->vgId); + break; + } + } + if(jump) continue; for (int i = 0; i < taosArrayGetSize(pOutput->pSub->offsetRows); i++) { OffsetRows *d2 = taosArrayGet(pOutput->pSub->offsetRows, i); if (d1->vgId == d2->vgId) { d2->rows += d1->rows; d2->offset = d1->offset; -// mDebug("pSub->offsetRows add vgId:%d, after:%"PRId64", before:%"PRId64, d2->vgId, d2->rows, d1->rows); + mInfo("pSub->offsetRows add vgId:%d, after:%"PRId64", before:%"PRId64, d2->vgId, d2->rows, d1->rows); } } } From 865a3b2509b8f60204d325356934946075b54f2b Mon Sep 17 00:00:00 2001 From: wangmm0220 Date: Wed, 28 Jun 2023 16:37:57 +0800 Subject: [PATCH 05/11] fix:add rows to pSub if rebalance --- source/dnode/mnode/impl/src/mndSubscribe.c | 49 +++++++++++----------- 1 file changed, 24 insertions(+), 25 deletions(-) diff --git a/source/dnode/mnode/impl/src/mndSubscribe.c b/source/dnode/mnode/impl/src/mndSubscribe.c index cfc164c26c..1106ef2c82 100644 --- a/source/dnode/mnode/impl/src/mndSubscribe.c +++ b/source/dnode/mnode/impl/src/mndSubscribe.c @@ -472,43 +472,42 @@ static int32_t mndDoRebalance(SMnode *pMnode, const SMqRebInputObj *pInput, SMqR SMqSubscribeObj *pSub = mndAcquireSubscribeByKey(pMnode, pInput->pRebInfo->key); // put all offset rows if (pSub) { taosRLockLatch(&pSub->lock); - bool init = false; if (pOutput->pSub->offsetRows == NULL) { pOutput->pSub->offsetRows = taosArrayInit(4, sizeof(OffsetRows)); - init = true; } pIter = NULL; while (1) { pIter = taosHashIterate(pSub->consumerHash, pIter); if (pIter == NULL) break; SMqConsumerEp *pConsumerEp = (SMqConsumerEp *)pIter; - if (init) { - taosArrayAddAll(pOutput->pSub->offsetRows, pConsumerEp->offsetRows); - mInfo("pSub->offsetRows is init"); - } else { - SMqConsumerEp *pConsumerEpNew = taosHashGet(pOutput->pSub->consumerHash, &pConsumerEp->consumerId, sizeof(int64_t)); + SMqConsumerEp *pConsumerEpNew = taosHashGet(pOutput->pSub->consumerHash, &pConsumerEp->consumerId, sizeof(int64_t)); - for (int j = 0; j < taosArrayGetSize(pConsumerEp->offsetRows); j++) { - OffsetRows *d1 = taosArrayGet(pConsumerEp->offsetRows, j); - bool jump = false; - for (int i = 0; pConsumerEpNew && i < taosArrayGetSize(pConsumerEpNew->vgs); i++){ - SMqVgEp *pVgEp = taosArrayGetP(pConsumerEpNew->vgs, j); - if(pVgEp->vgId == d1->vgId){ - jump = true; - mInfo("pSub->offsetRows jump, because consumer id:%"PRIx64 " and vgId:%d not change", pConsumerEp->consumerId, pVgEp->vgId); - break; - } + for (int j = 0; j < taosArrayGetSize(pConsumerEp->offsetRows); j++) { + OffsetRows *d1 = taosArrayGet(pConsumerEp->offsetRows, j); + bool jump = false; + for (int i = 0; pConsumerEpNew && i < taosArrayGetSize(pConsumerEpNew->vgs); i++){ + SMqVgEp *pVgEp = taosArrayGetP(pConsumerEpNew->vgs, j); + if(pVgEp->vgId == d1->vgId){ + jump = true; + mInfo("pSub->offsetRows jump, because consumer id:%"PRIx64 " and vgId:%d not change", pConsumerEp->consumerId, pVgEp->vgId); + break; } - if(jump) continue; - for (int i = 0; i < taosArrayGetSize(pOutput->pSub->offsetRows); i++) { - OffsetRows *d2 = taosArrayGet(pOutput->pSub->offsetRows, i); - if (d1->vgId == d2->vgId) { - d2->rows += d1->rows; - d2->offset = d1->offset; - mInfo("pSub->offsetRows add vgId:%d, after:%"PRId64", before:%"PRId64, d2->vgId, d2->rows, d1->rows); - } + } + if(jump) continue; + bool find = false; + for (int i = 0; i < taosArrayGetSize(pOutput->pSub->offsetRows); i++) { + OffsetRows *d2 = taosArrayGet(pOutput->pSub->offsetRows, i); + if (d1->vgId == d2->vgId) { + d2->rows += d1->rows; + d2->offset = d1->offset; + find = true; + mInfo("pSub->offsetRows add vgId:%d, after:%"PRId64", before:%"PRId64, d2->vgId, d2->rows, d1->rows); + break; } } + if(!find){ + taosArrayPush(pOutput->pSub->offsetRows, d1); + } } } taosRUnLockLatch(&pSub->lock); From 4ead47bcbdffecdab9aa7fd14cece385a0060e93 Mon Sep 17 00:00:00 2001 From: wangmm0220 Date: Wed, 28 Jun 2023 16:43:30 +0800 Subject: [PATCH 06/11] fix:add rows to pSub if rebalance --- source/dnode/mnode/impl/src/mndSubscribe.c | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/source/dnode/mnode/impl/src/mndSubscribe.c b/source/dnode/mnode/impl/src/mndSubscribe.c index 1106ef2c82..2fe169657c 100644 --- a/source/dnode/mnode/impl/src/mndSubscribe.c +++ b/source/dnode/mnode/impl/src/mndSubscribe.c @@ -486,7 +486,7 @@ static int32_t mndDoRebalance(SMnode *pMnode, const SMqRebInputObj *pInput, SMqR OffsetRows *d1 = taosArrayGet(pConsumerEp->offsetRows, j); bool jump = false; for (int i = 0; pConsumerEpNew && i < taosArrayGetSize(pConsumerEpNew->vgs); i++){ - SMqVgEp *pVgEp = taosArrayGetP(pConsumerEpNew->vgs, j); + SMqVgEp *pVgEp = taosArrayGetP(pConsumerEpNew->vgs, i); if(pVgEp->vgId == d1->vgId){ jump = true; mInfo("pSub->offsetRows jump, because consumer id:%"PRIx64 " and vgId:%d not change", pConsumerEp->consumerId, pVgEp->vgId); From c98a298b624fcdc7a85804b45af2df54915810e3 Mon Sep 17 00:00:00 2001 From: wangmm0220 Date: Wed, 28 Jun 2023 17:13:14 +0800 Subject: [PATCH 07/11] fix:test case --- tests/system-test/7-tmq/checkOffsetRowParams.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/system-test/7-tmq/checkOffsetRowParams.py b/tests/system-test/7-tmq/checkOffsetRowParams.py index 8a24148064..f7e4c61c9c 100644 --- a/tests/system-test/7-tmq/checkOffsetRowParams.py +++ b/tests/system-test/7-tmq/checkOffsetRowParams.py @@ -245,7 +245,7 @@ class TDTestCase: tdSql.query("show consumers") tdSql.checkRows(1) - tdSql.checkData(0, 8, "tbname:1,commit:1,interval:2000,reset:earliest") + tdSql.checkData(0, 8, "tbname:1,commit:1,interval:2000ms,reset:earliest") time.sleep(2) tdLog.info("start insert data") From fdcfa879948b93181e0ae82aafdd0c50f9de5a5e Mon Sep 17 00:00:00 2001 From: wangmm0220 Date: Wed, 28 Jun 2023 18:34:22 +0800 Subject: [PATCH 08/11] fix:case error --- tests/system-test/7-tmq/tmqParamsTest.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/system-test/7-tmq/tmqParamsTest.py b/tests/system-test/7-tmq/tmqParamsTest.py index 82000ba02a..f48eaa84d4 100644 --- a/tests/system-test/7-tmq/tmqParamsTest.py +++ b/tests/system-test/7-tmq/tmqParamsTest.py @@ -94,7 +94,7 @@ class TDTestCase: consumer_commit = 1 if consumer_dict["enable.auto.commit"] == "true" else 0 consumer_tbname = 1 if consumer_dict["msg.with.table.name"] == "true" else 0 consumer_ret = "earliest" if offset_value == "" else offset_value - expected_parameters=f'tbname:{consumer_tbname},commit:{consumer_commit},interval:{paraDict["auto_commit_interval"]},reset:{consumer_ret}' + expected_parameters=f'tbname:{consumer_tbname},commit:{consumer_commit},interval:{paraDict["auto_commit_interval"]}ms,reset:{consumer_ret}' if len(offset_value) == 0: del consumer_dict["auto.offset.reset"] consumer = Consumer(consumer_dict) From 98ce30f55cb4c52a2fa8fbf47123b15c4090e133 Mon Sep 17 00:00:00 2001 From: wangmm0220 Date: Wed, 28 Jun 2023 19:26:34 +0800 Subject: [PATCH 09/11] docs:modify docs for tmq --- docs/en/07-develop/07-tmq.mdx | 15 ++++++++++----- docs/en/12-taos-sql/24-show.md | 2 +- docs/zh/07-develop/07-tmq.mdx | 12 ++++++++---- docs/zh/12-taos-sql/24-show.md | 2 +- 4 files changed, 20 insertions(+), 11 deletions(-) diff --git a/docs/en/07-develop/07-tmq.mdx b/docs/en/07-develop/07-tmq.mdx index 578f38e73d..f5e0378a00 100644 --- a/docs/en/07-develop/07-tmq.mdx +++ b/docs/en/07-develop/07-tmq.mdx @@ -244,6 +244,8 @@ The following SQL statement creates a topic in TDengine: CREATE TOPIC topic_name AS SELECT ts, c1, c2, c3 FROM tmqdb.stb WHERE c1 > 1; ``` +- There is an upper limit to the number of topics created, controlled by the parameter tmqMaxTopicNum, with a default of 20 + Multiple subscription types are supported. #### Subscribe to a Column @@ -265,14 +267,15 @@ You can subscribe to a topic through a SELECT statement. Statements that specify Syntax: ```sql -CREATE TOPIC topic_name AS STABLE stb_name +CREATE TOPIC topic_name [with meta] AS STABLE stb_name [where_condition] ``` Creating a topic in this manner differs from a `SELECT * from stbName` statement as follows: - The table schema can be modified. - Unstructured data is returned. The format of the data returned changes based on the supertable schema. -- A different table schema may exist for every data block to be processed. +- The 'with meta' parameter is optional. When selected, statements such as creating super tables and sub tables will be returned, mainly used for Taosx to perform super table migration +- The 'where_condition' parameter is optional and will be used to filter and subscribe to sub tables that meet the criteria. Where conditions cannot have ordinary columns, only tags or tbnames. Functions can be used in where conditions to filter tags, but cannot be aggregate functions because sub table tag values cannot be aggregated. It can also be a constant expression, such as 2>1 (subscribing to all child tables), Or false (subscribe to 0 sub tables) - The data returned does not include tags. ### Subscribe to a Database @@ -280,10 +283,12 @@ Creating a topic in this manner differs from a `SELECT * from stbName` statement Syntax: ```sql -CREATE TOPIC topic_name [WITH META] AS DATABASE db_name; +CREATE TOPIC topic_name [with meta] AS DATABASE db_name; ``` -This SQL statement creates a subscription to all tables in the database. You can add the `WITH META` parameter to include schema changes in the subscription, including creating and deleting supertables; adding, deleting, and modifying columns; and creating, deleting, and modifying the tags of subtables. Consumers can determine the message type from the API. Note that this differs from Kafka. +This SQL statement creates a subscription to all tables in the database. + +- The 'with meta' parameter is optional. When selected, it will return statements for creating all super tables and sub tables in the database, mainly used for Taosx database migration ## Create a Consumer @@ -295,7 +300,7 @@ You configure the following parameters when creating a consumer: | `td.connect.user` | string | User Name | | | `td.connect.pass` | string | Password | | | `td.connect.port` | string | Port of the server side | | -| `group.id` | string | Consumer group ID; consumers with the same ID are in the same group | **Required**. Maximum length: 192. | +| `group.id` | string | Consumer group ID; consumers with the same ID are in the same group | **Required**. Maximum length: 192. Each topic can create up to 100 consumer groups. | | `client.id` | string | Client ID | Maximum length: 192. | | `auto.offset.reset` | enum | Initial offset for the consumer group | Specify `earliest`, `latest`, or `none`(default) | | `enable.auto.commit` | boolean | Commit automatically; true: user application doesn't need to explicitly commit; false: user application need to handle commit by itself | Default value is true | diff --git a/docs/en/12-taos-sql/24-show.md b/docs/en/12-taos-sql/24-show.md index eb70a7664b..bd4a60b20e 100644 --- a/docs/en/12-taos-sql/24-show.md +++ b/docs/en/12-taos-sql/24-show.md @@ -36,7 +36,7 @@ Shows information about connections to the system. SHOW CONSUMERS; ``` -Shows information about all active consumers in the system. +Shows information about all consumers in the system. ## SHOW CREATE DATABASE diff --git a/docs/zh/07-develop/07-tmq.mdx b/docs/zh/07-develop/07-tmq.mdx index a87a1f64f8..54a8af2287 100644 --- a/docs/zh/07-develop/07-tmq.mdx +++ b/docs/zh/07-develop/07-tmq.mdx @@ -243,6 +243,7 @@ TDengine 使用 SQL 创建一个 topic： ```sql CREATE TOPIC topic_name AS SELECT ts, c1, c2, c3 FROM tmqdb.stb WHERE c1 > 1; ``` +- topic创建个数有上限，通过参数 tmqMaxTopicNum 控制，默认 20 个 TMQ 支持多种订阅类型： @@ -265,14 +266,15 @@ CREATE TOPIC topic_name as subquery 语法： ```sql -CREATE TOPIC topic_name AS STABLE stb_name +CREATE TOPIC topic_name [with meta] AS STABLE stb_name [where_condition] ``` 与 `SELECT * from stbName` 订阅的区别是： - 不会限制用户的表结构变更。 - 返回的是非结构化的数据：返回数据的结构会随之超级表的表结构变化而变化。 -- 用户对于要处理的每一个数据块都可能有不同的表结构。 +- with meta 参数可选，选择时将返回创建超级表，子表等语句，主要用于taosx做超级表迁移 +- where_condition 参数可选，选择时将用来过滤符合条件的子表，订阅这些子表。where 条件里不能有普通列，只能是tag或tbname，where条件里可以用函数，用来过滤tag，但是不能是聚合函数，因为子表tag值无法做聚合。也可以是常量表达式，比如 2 > 1（订阅全部子表），或者 false（订阅0个子表） - 返回数据不包含标签。 ### 数据库订阅 @@ -280,11 +282,13 @@ CREATE TOPIC topic_name AS STABLE stb_name 语法： ```sql -CREATE TOPIC topic_name AS DATABASE db_name; +CREATE TOPIC topic_name [with meta] AS DATABASE db_name; ``` 通过该语句可创建一个包含数据库所有表数据的订阅 +- with meta 参数可选，选择时将返回创建数据库里所有超级表，子表的语句，主要用于taosx做数据库迁移 + ## 创建消费者 *consumer* 消费者需要通过一系列配置选项创建，基础配置项如下表所示： @@ -295,7 +299,7 @@ CREATE TOPIC topic_name AS DATABASE db_name; | `td.connect.user` | string | 用户名 | | | `td.connect.pass` | string | 密码 | | | `td.connect.port` | integer | 服务端的端口号 | | -| `group.id` | string | 消费组 ID，同一消费组共享消费进度 | **必填项**。最大长度：192。 | +| `group.id` | string | 消费组 ID，同一消费组共享消费进度 |
**必填项**。最大长度：192。
每个topic最多可建立100个 consumer group | | `client.id` | string | 客户端 ID | 最大长度：192。 | | `auto.offset.reset` | enum | 消费组订阅的初始位置 |
`earliest`: default;从头开始订阅;
`latest`: 仅从最新数据开始订阅;
`none`: 没有提交的 offset 无法订阅 | | `enable.auto.commit` | boolean | 是否启用消费位点自动提交，true: 自动提交，客户端应用无需commit；false：客户端应用需要自行commit | 默认值为 true | diff --git a/docs/zh/12-taos-sql/24-show.md b/docs/zh/12-taos-sql/24-show.md index 12ad665e42..f3397ae82d 100644 --- a/docs/zh/12-taos-sql/24-show.md +++ b/docs/zh/12-taos-sql/24-show.md @@ -36,7 +36,7 @@ SHOW CONNECTIONS; SHOW CONSUMERS; ``` -显示当前数据库下所有活跃的消费者的信息。 +显示当前数据库下所有消费者的信息。 ## SHOW CREATE DATABASE From d73d3d0ead719d41e4d8c0be3f95306cb437d1a4 Mon Sep 17 00:00:00 2001 From: Haojun Liao Date: Thu, 29 Jun 2023 10:59:59 +0800 Subject: [PATCH 10/11] fix(stream): fix error in type bytes. --- source/common/src/systable.c | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/source/common/src/systable.c b/source/common/src/systable.c index 7cd33955c1..f2250d33e2 100644 --- a/source/common/src/systable.c +++ b/source/common/src/systable.c @@ -160,9 +160,9 @@ static const SSysDbTableSchema streamSchema[] = { static const SSysDbTableSchema streamTaskSchema[] = { {.name = "stream_name", .bytes = SYSTABLE_SCH_DB_NAME_LEN, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = false}, - {.name = "task_id", .bytes = 8, .type = TSDB_DATA_TYPE_INT, .sysInfo = false}, + {.name = "task_id", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = false}, {.name = "node_type", .bytes = SYSTABLE_SCH_DB_NAME_LEN, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = false}, - {.name = "node_id", .bytes = 8, .type = TSDB_DATA_TYPE_INT, .sysInfo = false}, + {.name = "node_id", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = false}, {.name = "level", .bytes = 20 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = false}, {.name = "status", .bytes = 20 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = false}, }; From 2035bc6f7d9abc5dac19b4b7b1bba19b3e609237 Mon Sep 17 00:00:00 2001 From: wangmm0220 Date: Thu, 29 Jun 2023 11:46:10 +0800 Subject: [PATCH 11/11] fix:diable test case temporary --- tests/parallel_test/cases.task | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/parallel_test/cases.task b/tests/parallel_test/cases.task index 1dff2a90d0..a9dc68ac1c 100644 --- a/tests/parallel_test/cases.task +++ b/tests/parallel_test/cases.task @@ -780,7 +780,7 @@ ,,y,script,./test.sh -f tsim/user/basic.sim ,,y,script,./test.sh -f tsim/user/password.sim ,,y,script,./test.sh -f tsim/user/privilege_db.sim -,,y,script,./test.sh -f tsim/user/privilege_sysinfo.sim +#,,y,script,./test.sh -f tsim/user/privilege_sysinfo.sim ,,y,script,./test.sh -f tsim/user/privilege_topic.sim ,,y,script,./test.sh -f tsim/user/privilege_table.sim ,,y,script,./test.sh -f tsim/db/alter_option.sim