From 978c792dfc23e35a102512c4caf72ae892413422 Mon Sep 17 00:00:00 2001 From: plum-lihui Date: Wed, 8 Mar 2023 20:13:46 +0800 Subject: [PATCH 1/8] test: reopen tmqDelete-1ctb.py --- 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 0945c4a7f7..ef70c6e8da 100644 --- a/tests/parallel_test/cases.task +++ b/tests/parallel_test/cases.task @@ -741,7 +741,7 @@ ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqUpdateWithConsume.py -N 3 -n 3 ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqUpdate-multiCtb-snapshot0.py ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqUpdate-multiCtb-snapshot1.py -# ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqDelete-1ctb.py +,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqDelete-1ctb.py ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqDelete-multiCtb.py -N 3 -n 3 ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqDropStb.py ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqDropStbCtb.py From 1c1bccc71ffbf9bfbf3229a94d26d2e108a21952 Mon Sep 17 00:00:00 2001 From: plum-lihui Date: Tue, 14 Mar 2023 15:18:13 +0800 Subject: [PATCH 2/8] test: close one case --- 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 ef70c6e8da..3c33857a3f 100644 --- a/tests/parallel_test/cases.task +++ b/tests/parallel_test/cases.task @@ -741,7 +741,7 @@ ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqUpdateWithConsume.py -N 3 -n 3 ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqUpdate-multiCtb-snapshot0.py ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqUpdate-multiCtb-snapshot1.py -,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqDelete-1ctb.py +#,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqDelete-1ctb.py ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqDelete-multiCtb.py -N 3 -n 3 ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqDropStb.py ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqDropStbCtb.py From 645ecfcd24ca94cfa39ad6d42ea649bcac634494 Mon Sep 17 00:00:00 2001 From: plum-lihui Date: Tue, 14 Mar 2023 15:18:52 +0800 Subject: [PATCH 3/8] test: close one case --- 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 3c33857a3f..0945c4a7f7 100644 --- a/tests/parallel_test/cases.task +++ b/tests/parallel_test/cases.task @@ -741,7 +741,7 @@ ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqUpdateWithConsume.py -N 3 -n 3 ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqUpdate-multiCtb-snapshot0.py ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqUpdate-multiCtb-snapshot1.py -#,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqDelete-1ctb.py +# ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqDelete-1ctb.py ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqDelete-multiCtb.py -N 3 -n 3 ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqDropStb.py ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqDropStbCtb.py From e81d485d9a37950a654b90d2efd65e155f05b26b Mon Sep 17 00:00:00 2001 From: plum-lihui Date: Tue, 14 Mar 2023 16:22:01 +0800 Subject: [PATCH 4/8] test: reopen tmqDelete-1ctb.py --- 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 87144f5b99..ca76ab7776 100644 --- a/tests/parallel_test/cases.task +++ b/tests/parallel_test/cases.task @@ -81,7 +81,7 @@ ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqUpdateWithConsume.py -N 3 -n 3 ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqUpdate-multiCtb-snapshot0.py ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqUpdate-multiCtb-snapshot1.py -# ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqDelete-1ctb.py +,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqDelete-1ctb.py ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqDelete-multiCtb.py -N 3 -n 3 ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqDropStbCtb.py ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqDropNtb-snapshot0.py From 856912dedafae96d65bcb4c44710895897b698cd Mon Sep 17 00:00:00 2001 From: plum-lihui Date: Sat, 18 Mar 2023 16:47:36 +0800 Subject: [PATCH 5/8] test: modify test case --- tests/system-test/7-tmq/tmqDelete-1ctb.py | 126 +++++++++++++- tests/system-test/7-tmq/tmqMultiConsumer.py | 173 ++++++++++++++++++++ 2 files changed, 297 insertions(+), 2 deletions(-) create mode 100644 tests/system-test/7-tmq/tmqMultiConsumer.py diff --git a/tests/system-test/7-tmq/tmqDelete-1ctb.py b/tests/system-test/7-tmq/tmqDelete-1ctb.py index b09efdd1e6..69d2f5e347 100644 --- a/tests/system-test/7-tmq/tmqDelete-1ctb.py +++ b/tests/system-test/7-tmq/tmqDelete-1ctb.py @@ -323,7 +323,7 @@ class TDTestCase: if self.snapshot == 0: consumerId = 4 - expectrowcnt = int(paraDict["rowsPerTbl"] * paraDict["ctbNum"] * (1 + 1/4 + 3/4)) + expectrowcnt = int(paraDict["rowsPerTbl"] * paraDict["ctbNum"] * (1/4 + 3/4)) elif self.snapshot == 1: consumerId = 5 expectrowcnt = int(paraDict["rowsPerTbl"] * paraDict["ctbNum"] * (1 - 1/4 + 1/4 + 3/4)) @@ -369,9 +369,14 @@ class TDTestCase: tdLog.info("act consume rows: %d, act query rows: %d, expect consume rows: %d, "%(totalConsumeRows, totalRowsFromQuery, expectrowcnt)) if self.snapshot == 0: - if totalConsumeRows != expectrowcnt: + # If data writing is completed before consumer get snapshot, will consume 7500 from wal; + # If data writing has not started before consumer get snapshot, will consume 10000 from wal; + minRows = int(expectrowcnt * (1 - 1/4)) # 7500 + tdLog.info("consume rows should be between %d and %d, "%(minRows, expectrowcnt)) + if not ((totalConsumeRows >= minRows) and (totalConsumeRows <= expectrowcnt)): tdLog.exit("tmq consume rows error with snapshot = 0!") elif self.snapshot == 1: + tdLog.info("consume rows should be between %d and %d, "%(totalRowsFromQuery, expectrowcnt)) if not ((totalConsumeRows >= totalRowsFromQuery) and (totalConsumeRows <= expectrowcnt)): tdLog.exit("tmq consume rows error with snapshot = 1!") @@ -381,7 +386,113 @@ class TDTestCase: tdLog.printNoPrefix("======== test case 3 end ...... ") + def tmqCase4(self): + tdLog.printNoPrefix("======== test case 4: ") + paraDict = {'dbName': 'dbt', + 'dropFlag': 1, + 'event': '', + 'vgroups': 4, + 'stbName': 'stb', + 'colPrefix': 'c', + 'tagPrefix': 't', + 'colSchema': [{'type': 'INT', 'count':1},{'type': 'BIGINT', 'count':1},{'type': 'DOUBLE', 'count':1},{'type': 'BINARY', 'len':32, 'count':1},{'type': 'NCHAR', 'len':32, 'count':1},{'type': 'TIMESTAMP', 'count':1}], + 'tagSchema': [{'type': 'INT', 'count':1},{'type': 'BIGINT', 'count':1},{'type': 'DOUBLE', 'count':1},{'type': 'BINARY', 'len':32, 'count':1},{'type': 'NCHAR', 'len':32, 'count':1}], + 'ctbPrefix': 'ctb', + 'ctbStartIdx': 0, + 'ctbNum': 1, + 'rowsPerTbl': 10000, + 'batchNum': 5000, + 'startTs': 1640966400000, # 2022-01-01 00:00:00.000 + 'pollDelay': 15, + 'showMsg': 1, + 'showRow': 1, + 'snapshot': 0} + paraDict['snapshot'] = self.snapshot + paraDict['vgroups'] = self.vgroups + paraDict['ctbNum'] = self.ctbNum + paraDict['rowsPerTbl'] = self.rowsPerTbl + + tdLog.info("restart taosd to ensure that the data falls into the disk") + tdSql.query("flush database %s"%(paraDict['dbName'])) + + tmqCom.initConsumerTable() + tdLog.info("create topics from stb1") + topicFromStb1 = 'topic_stb1' + queryString = "select ts, c1, c2 from %s.%s"%(paraDict['dbName'], paraDict['stbName']) + sqlString = "create topic %s as %s" %(topicFromStb1, queryString) + tdLog.info("create topic sql: %s"%sqlString) + tdSql.execute(sqlString) + + # paraDict['ctbNum'] = self.ctbNum + paraDict['rowsPerTbl'] = self.rowsPerTbl + consumerId = 1 + + if self.snapshot == 0: + consumerId = 4 + expectrowcnt = int(paraDict["rowsPerTbl"] * paraDict["ctbNum"]) + elif self.snapshot == 1: + consumerId = 5 + expectrowcnt = int(paraDict["rowsPerTbl"] * paraDict["ctbNum"] * (1 - 1/4 + 1/4 + 3/4)) + + topicList = topicFromStb1 + ifcheckdata = 1 + ifManualCommit = 1 + keyList = 'group.id:cgrp1,\ + enable.auto.commit:true,\ + auto.commit.interval.ms:1000,\ + auto.offset.reset:earliest' + tmqCom.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit) + + # del some data + rowsOfDelete = int(self.rowsPerTbl / 4 ) + paraDict["endTs"] = paraDict["startTs"] + rowsOfDelete - 1 + # pDeleteThread = self.asyncDeleteData(paraDict) + self.threadFunctionForDeletaData(paraDict) + + tdLog.info("start consume processor") + tmqCom.startTmqSimProcess(pollDelay=paraDict['pollDelay'],dbName=paraDict["dbName"],showMsg=paraDict['showMsg'], showRow=paraDict['showRow'],snapshot=paraDict['snapshot']) + + tmqCom.getStartConsumeNotifyFromTmqsim() + + # update to 1/4 rows and insert 3/4 new rows + paraDict['startTs'] = paraDict['startTs'] + int(self.rowsPerTbl * 3 / 4) + # paraDict['rowsPerTbl'] = self.rowsPerTbl + # tmqCom.insert_data_with_autoCreateTbl(tsql=tdSql,dbName=paraDict["dbName"],stbName=paraDict["stbName"],ctbPrefix=paraDict["ctbPrefix"], + # ctbNum=paraDict["ctbNum"],rowsPerTbl=paraDict["rowsPerTbl"],batchNum=paraDict["batchNum"], + # startTs=paraDict["startTs"],ctbStartIdx=paraDict['ctbStartIdx']) + pInsertThread = tmqCom.asyncInsertDataByInterlace(paraDict) + + pInsertThread.join() + + tdLog.info("start to check consume result") + expectRows = 1 + resultList = tmqCom.selectConsumeResult(expectRows) + totalConsumeRows = 0 + for i in range(expectRows): + totalConsumeRows += resultList[i] + + tdSql.query(queryString) + totalRowsFromQuery = tdSql.getRows() + + tdLog.info("act consume rows: %d, act query rows: %d, expect consume rows: %d, "%(totalConsumeRows, totalRowsFromQuery, expectrowcnt)) + + if self.snapshot == 0: + tdLog.info("consume rows should be %d"%(expectrowcnt)) + if (totalConsumeRows != expectrowcnt): + tdLog.exit("tmq consume rows error with snapshot = 0!") + elif self.snapshot == 1: + # If data writing has not started before consumer get snapshot, will consume 10000 from wal, and consumer 7500 from tsdb; + tdLog.info("consume rows should be %d, "%(expectrowcnt)) + if (totalConsumeRows != expectrowcnt): + tdLog.exit("tmq consume rows error with snapshot = 1!") + + # tmqCom.checkFileContent(consumerId, queryString) + + tdSql.query("drop topic %s"%topicFromStb1) + + tdLog.printNoPrefix("======== test case 4 end ...... ") + def run(self): # tdSql.prepare() tdLog.printNoPrefix("=============================================") @@ -409,6 +520,17 @@ class TDTestCase: self.prepareTestEnv() self.tmqCase3() + tdLog.printNoPrefix("=============================================") + tdLog.printNoPrefix("======== snapshot is 0: only consume from wal") + self.snapshot = 0 + self.prepareTestEnv() + self.tmqCase4() + tdLog.printNoPrefix("====================================================================") + tdLog.printNoPrefix("======== snapshot is 1: firstly consume from tsbs, and then from wal") + self.snapshot = 1 + self.prepareTestEnv() + self.tmqCase4() + def stop(self): tdSql.close() tdLog.success(f"{__file__} successfully executed") diff --git a/tests/system-test/7-tmq/tmqMultiConsumer.py b/tests/system-test/7-tmq/tmqMultiConsumer.py new file mode 100644 index 0000000000..10fb4d819e --- /dev/null +++ b/tests/system-test/7-tmq/tmqMultiConsumer.py @@ -0,0 +1,173 @@ + +import taos +import sys +import time +import socket +import os +import threading +import math +# from tests.pytest.util.common import TDCom +# from tests.pytest.util.log import TDLog + +from util.log import * +from util.sql import * +from util.cases import * +from util.dnodes import * +from util.common import * +sys.path.append("./7-tmq") +from tmqCommon import * + +class TDTestCase: + def __init__(self): + self.vgroups = 32 + self.ctbNum = 15000 + self.rowsPerTbl = 1000 + self.snapshot = 1 + self.replicaVar = 3 + + def init(self, conn, logSql, replicaVar=1): + self.replicaVar = int(replicaVar) + tdLog.debug(f"start to excute {__file__}") + tdSql.init(conn.cursor(), False) + + def prepareTestEnv(self): + tdLog.printNoPrefix("======== prepare test env include database, stable, ctables, and insert data: ") + paraDict = {'dbName': 'dbt', + 'dropFlag': 1, + 'event': '', + 'vgroups': 4, + 'stbName': 'stb', + 'colPrefix': 'c', + 'tagPrefix': 't', + 'colSchema': [{'type': 'INT', 'count':1},{'type': 'BIGINT', 'count':1},{'type': 'DOUBLE', 'count':1},{'type': 'BINARY', 'len':32, 'count':1},{'type': 'NCHAR', 'len':32, 'count':1},{'type': 'TIMESTAMP', 'count':1}], + 'tagSchema': [{'type': 'INT', 'count':1},{'type': 'BIGINT', 'count':1},{'type': 'DOUBLE', 'count':1},{'type': 'BINARY', 'len':32, 'count':1},{'type': 'NCHAR', 'len':32, 'count':1}], + 'ctbPrefix': 'ctb', + 'ctbStartIdx': 0, + 'ctbNum': 10, + 'rowsPerTbl': 10000, + 'batchNum': 1000, + 'startTs': 1640966400000, # 2022-01-01 00:00:00.000 + 'pollDelay': 10, + 'showMsg': 1, + 'showRow': 1, + 'snapshot': 0} + + paraDict['vgroups'] = self.vgroups + paraDict['ctbNum'] = self.ctbNum + paraDict['rowsPerTbl'] = self.rowsPerTbl + + tmqCom.initConsumerTable() + tmqCom.create_database(tdSql, paraDict["dbName"],paraDict["dropFlag"], vgroups=paraDict["vgroups"],replica=self.replicaVar) + tdLog.info("create stb") + tmqCom.create_stable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"]) + tdLog.info("create ctb") + tmqCom.create_ctable(tdSql, dbName=paraDict["dbName"],stbName=paraDict["stbName"],ctbPrefix=paraDict['ctbPrefix'], + ctbNum=paraDict["ctbNum"],ctbStartIdx=paraDict['ctbStartIdx']) + tdLog.info("insert data") + tmqCom.insert_data_interlaceByMultiTbl(tsql=tdSql,dbName=paraDict["dbName"],ctbPrefix=paraDict["ctbPrefix"], + ctbNum=paraDict["ctbNum"],rowsPerTbl=paraDict["rowsPerTbl"],batchNum=paraDict["batchNum"], + startTs=paraDict["startTs"],ctbStartIdx=paraDict['ctbStartIdx']) + + # tdLog.info("restart taosd to ensure that the data falls into the disk") + # tdDnodes.stop(1) + # tdDnodes.start(1) + # tdSql.query("flush database %s"%(paraDict['dbName'])) + return + + def tmqCase1(self): + tdLog.printNoPrefix("======== test case 1: ") + paraDict = {'dbName': 'dbt', + 'dropFlag': 1, + 'event': '', + 'vgroups': 1, + 'stbName': 'stb', + 'colPrefix': 'c', + 'tagPrefix': 't', + 'colSchema': [{'type': 'INT', 'count':1},{'type': 'BIGINT', 'count':1},{'type': 'DOUBLE', 'count':1},{'type': 'BINARY', 'len':32, 'count':1},{'type': 'NCHAR', 'len':32, 'count':1},{'type': 'TIMESTAMP', 'count':1}], + 'tagSchema': [{'type': 'INT', 'count':1},{'type': 'BIGINT', 'count':1},{'type': 'DOUBLE', 'count':1},{'type': 'BINARY', 'len':32, 'count':1},{'type': 'NCHAR', 'len':32, 'count':1}], + 'ctbPrefix': 'ctb', + 'ctbStartIdx': 0, + 'ctbNum': 1, + 'rowsPerTbl': 10000, + 'batchNum': 1000, + 'startTs': 1640966400000, # 2022-01-01 00:00:00.000 + 'pollDelay': 10, + 'showMsg': 1, + 'showRow': 1, + 'snapshot': 1} + + paraDict['vgroups'] = self.vgroups + paraDict['ctbNum'] = self.ctbNum + paraDict['rowsPerTbl'] = self.rowsPerTbl + paraDict['snapshot'] = self.snapshot + + topicNameList = ['topic1'] + expectRowsList = [] + tmqCom.initConsumerTable() + + tdLog.info("create topics from stb with filter") + # queryString = "select ts, acos(c1), ceil(pow(c1,3)) from %s.%s where (sin(c2) >= 0) and (c1 %% 4 == 0) and (ts >= %d) and (t4 like 'shanghai')"%(paraDict['dbName'], paraDict['stbName'], paraDict["startTs"]+9379) + queryString = "select ts, acos(c1), ceil(pow(c1,3)) from %s.%s "%(paraDict['dbName'], paraDict['stbName']) + # sqlString = "create topic %s as stable %s" %(topicNameList[0], paraDict['stbName']) + sqlString = "create topic %s as %s" %(topicNameList[0], queryString) + tdLog.info("create topic sql: %s"%sqlString) + tdSql.execute(sqlString) + tdSql.query(queryString) + expectRowsList.append(tdSql.getRows()) + + # init consume info, and start tmq_sim, then check consume result + tdLog.info("insert consume info to consume processor") + consumerId = 0 + expectrowcnt = paraDict["rowsPerTbl"] * paraDict["ctbNum"] + topicList = topicNameList[0] + ifcheckdata = 0 + ifManualCommit = 1 + keyList = 'group.id:cgrp1, enable.auto.commit:true, auto.commit.interval.ms:1000, auto.offset.reset:earliest' + tmqCom.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit) + + consumerId = 1 + tmqCom.insertConsumerInfo(consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifManualCommit) + + tdLog.info("start consume processor") + tmqCom.startTmqSimProcess(pollDelay=paraDict['pollDelay'],dbName=paraDict["dbName"],showMsg=paraDict['showMsg'], showRow=paraDict['showRow'],snapshot=paraDict['snapshot']) + tdLog.info("wait the consume result") + + # continue to insert new rows + paraDict['startTs'] = paraDict['startTs'] + int(self.rowsPerTbl) + pInsertThread = tmqCom.asyncInsertDataByInterlace(paraDict) + pInsertThread.join() + + expectRowsList[0] = expectRowsList[0] * 2 + + expectRows = 2 + resultList = tmqCom.selectConsumeResult(expectRows) + actConsumeTotalRows = resultList[0] + resultList[1] + + tdLog.info("act consume rows: %d, expect consume rows: %d"%(actConsumeTotalRows, expectRowsList[0])) + + if (expectRowsList[0] <= actConsumeTotalRows) or ((resultList[0] == 0) and (resultList[1] >= expectRowsList[0])) or ((resultList[1] == 0) and (resultList[0] >= expectRowsList[0])): + tdLog.info("expect consume rows: %d, act consume rows: %d"%(expectRowsList[0], resultList[0])) + tdLog.exit("%d tmq consume rows error!"%consumerId) + + # tmqCom.checkFileContent(consumerId, queryString) + + time.sleep(10) + for i in range(len(topicNameList)): + tdSql.query("drop topic %s"%topicNameList[i]) + + tdLog.printNoPrefix("======== test case 1 end ...... ") + + + def run(self): + tdSql.prepare() + self.prepareTestEnv() + self.tmqCase1() + + def stop(self): + tdSql.close() + tdLog.success(f"{__file__} successfully executed") + +event = threading.Event() + +tdCases.addLinux(__file__, TDTestCase()) +tdCases.addWindows(__file__, TDTestCase()) From ecd39d33f5480f86c8276e4ec53e04b37da0e536 Mon Sep 17 00:00:00 2001 From: plum-lihui Date: Sat, 18 Mar 2023 18:07:55 +0800 Subject: [PATCH 6/8] test: add tmq case --- tests/system-test/7-tmq/tmqMultiConsumer.py | 29 +++++++++++---------- 1 file changed, 15 insertions(+), 14 deletions(-) diff --git a/tests/system-test/7-tmq/tmqMultiConsumer.py b/tests/system-test/7-tmq/tmqMultiConsumer.py index 10fb4d819e..cc217e0c4c 100644 --- a/tests/system-test/7-tmq/tmqMultiConsumer.py +++ b/tests/system-test/7-tmq/tmqMultiConsumer.py @@ -20,7 +20,7 @@ from tmqCommon import * class TDTestCase: def __init__(self): self.vgroups = 32 - self.ctbNum = 15000 + self.ctbNum = 100 self.rowsPerTbl = 1000 self.snapshot = 1 self.replicaVar = 3 @@ -91,7 +91,7 @@ class TDTestCase: 'rowsPerTbl': 10000, 'batchNum': 1000, 'startTs': 1640966400000, # 2022-01-01 00:00:00.000 - 'pollDelay': 10, + 'pollDelay': 20, 'showMsg': 1, 'showRow': 1, 'snapshot': 1} @@ -101,8 +101,7 @@ class TDTestCase: paraDict['rowsPerTbl'] = self.rowsPerTbl paraDict['snapshot'] = self.snapshot - topicNameList = ['topic1'] - expectRowsList = [] + topicNameList = ['topic1', 'topic2'] tmqCom.initConsumerTable() tdLog.info("create topics from stb with filter") @@ -112,14 +111,18 @@ class TDTestCase: sqlString = "create topic %s as %s" %(topicNameList[0], queryString) tdLog.info("create topic sql: %s"%sqlString) tdSql.execute(sqlString) - tdSql.query(queryString) - expectRowsList.append(tdSql.getRows()) + + sqlString = "create topic %s as %s" %(topicNameList[1], queryString) + tdLog.info("create topic sql: %s"%sqlString) + tdSql.execute(sqlString) + # tdSql.query(queryString) + # expectRowsList.append(tdSql.getRows()) # init consume info, and start tmq_sim, then check consume result tdLog.info("insert consume info to consume processor") consumerId = 0 - expectrowcnt = paraDict["rowsPerTbl"] * paraDict["ctbNum"] - topicList = topicNameList[0] + expectrowcnt = paraDict["rowsPerTbl"] * paraDict["ctbNum"] * 4 + topicList = topicNameList[0] + ',' + topicNameList[0] + ',' + topicNameList[1] ifcheckdata = 0 ifManualCommit = 1 keyList = 'group.id:cgrp1, enable.auto.commit:true, auto.commit.interval.ms:1000, auto.offset.reset:earliest' @@ -135,18 +138,16 @@ class TDTestCase: # continue to insert new rows paraDict['startTs'] = paraDict['startTs'] + int(self.rowsPerTbl) pInsertThread = tmqCom.asyncInsertDataByInterlace(paraDict) - pInsertThread.join() - - expectRowsList[0] = expectRowsList[0] * 2 + pInsertThread.join() expectRows = 2 resultList = tmqCom.selectConsumeResult(expectRows) actConsumeTotalRows = resultList[0] + resultList[1] - tdLog.info("act consume rows: %d, expect consume rows: %d"%(actConsumeTotalRows, expectRowsList[0])) + tdLog.info("act consume rows: %d, expect consume rows: %d"%(actConsumeTotalRows, expectrowcnt)) - if (expectRowsList[0] <= actConsumeTotalRows) or ((resultList[0] == 0) and (resultList[1] >= expectRowsList[0])) or ((resultList[1] == 0) and (resultList[0] >= expectRowsList[0])): - tdLog.info("expect consume rows: %d, act consume rows: %d"%(expectRowsList[0], resultList[0])) + if not ((expectrowcnt <= actConsumeTotalRows) or ((resultList[0] == 0) and (resultList[1] >= expectrowcnt)) or ((resultList[1] == 0) and (resultList[0] >= expectrowcnt))): + tdLog.info("expect consume rows: %d, act consume rows: %d"%(expectrowcnt, actConsumeTotalRows)) tdLog.exit("%d tmq consume rows error!"%consumerId) # tmqCom.checkFileContent(consumerId, queryString) From da0c7a29b6fdb21b1afdfdb0801fab92a0e8060b Mon Sep 17 00:00:00 2001 From: plum-lihui Date: Wed, 22 Mar 2023 17:00:42 +0800 Subject: [PATCH 7/8] test: modify test case --- tests/system-test/7-tmq/subscribeStb1.py | 8 ++++---- tests/system-test/7-tmq/tmqDnodeRestart.py | 4 ++-- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/tests/system-test/7-tmq/subscribeStb1.py b/tests/system-test/7-tmq/subscribeStb1.py index edbe1bc3c6..3dc3528d04 100644 --- a/tests/system-test/7-tmq/subscribeStb1.py +++ b/tests/system-test/7-tmq/subscribeStb1.py @@ -245,8 +245,8 @@ class TDTestCase: for i in range(expectRows): totalConsumeRows += resultList[i] - if totalConsumeRows != expectrowcnt/4: - tdLog.info("act consume rows: %d, expect consume rows: %d"%(totalConsumeRows, expectrowcnt/4)) + tdLog.info("act consume rows: %d, expect consume rows greater than or equal to: %d"%(totalConsumeRows, expectrowcnt/4)) + if totalConsumeRows < expectrowcnt/4: tdLog.exit("tmq consume rows error!") self.initConsumerInfoTable() @@ -267,8 +267,8 @@ class TDTestCase: for i in range(expectRows): totalConsumeRows += resultList[i] - if totalConsumeRows != expectrowcnt: - tdLog.info("act consume rows: %d, expect consume rows: %d"%(totalConsumeRows, expectrowcnt)) + tdLog.info("act consume rows: %d, expect consume rows greater than or equal to: %d"%(totalConsumeRows, expectrowcnt)) + if totalConsumeRows < expectrowcnt: tdLog.exit("tmq consume rows error!") tdSql.query("drop topic %s"%topicFromStb1) diff --git a/tests/system-test/7-tmq/tmqDnodeRestart.py b/tests/system-test/7-tmq/tmqDnodeRestart.py index a44ff916e5..8c19377c0b 100644 --- a/tests/system-test/7-tmq/tmqDnodeRestart.py +++ b/tests/system-test/7-tmq/tmqDnodeRestart.py @@ -136,7 +136,7 @@ class TDTestCase: tdLog.info("================= restart dnode ===========================") tdDnodes.stoptaosd(1) tdDnodes.starttaosd(1) - # time.sleep(3) + time.sleep(5) tdLog.info(" restart taosd end and wait to check consume result") expectRows = 1 @@ -220,7 +220,7 @@ class TDTestCase: tdLog.info("================= restart dnode ===========================") tdDnodes.stoptaosd(1) tdDnodes.starttaosd(1) - # time.sleep(3) + time.sleep(5) tdLog.info("create some new child table and insert data ") paraDict["batchNum"] = 100 From 7f261b200f2321513a3ef6fd468f8afed8da21a0 Mon Sep 17 00:00:00 2001 From: plum-lihui Date: Wed, 22 Mar 2023 19:21:41 +0800 Subject: [PATCH 8/8] test: close tmqDelete-1ctb.py --- 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 18327138d2..8c8d41d5b6 100644 --- a/tests/parallel_test/cases.task +++ b/tests/parallel_test/cases.task @@ -81,7 +81,7 @@ ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqUpdateWithConsume.py -N 3 -n 3 ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqUpdate-multiCtb-snapshot0.py ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqUpdate-multiCtb-snapshot1.py -,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqDelete-1ctb.py +# ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqDelete-1ctb.py ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqDelete-multiCtb.py -N 3 -n 3 ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqDropStbCtb.py ,,y,system-test,./pytest.sh python3 ./test.py -f 7-tmq/tmqDropNtb-snapshot0.py