Merge pull request #13980 from taosdata/test3.0/lihui

test: add test case for tmq

tests/pytest/util/common.py

@@ -11,13 +11,20 @@
 
 # -*- coding: utf-8 -*-
 
+from collections import defaultdict
 import random
 import string
-from util.sql import tdSql
-from util.dnodes import tdDnodes
 import requests
 import time
 import socket
+
+import taos
+from util.log import *
+from util.sql import *
+from util.cases import *
+from util.dnodes import *
+from util.common import *
+
 class TDCom:
     def init(self, conn, logSql):
         tdSql.init(conn.cursor(), logSql)
@@ -170,4 +177,122 @@ class TDCom:
     def close(self):
         self.cursor.close()
 
+    def create_database(self,tsql, dbName='test',dropFlag=1,precision="ms", **kwargs):
+        if dropFlag == 1:
+            tsql.execute("drop database if exists %s"%(dbName))
+        '''
+        vgroups replica precision strict wal fsync comp cachelast single_stable buffer pagesize pages minrows maxrows duration keep retentions
+        '''
+        sqlString = f'create database if not exists {dbName} precision "{precision}" vgroups 4'
+        if len(kwargs) > 0:
+            dbParams = ""
+            for param, value in kwargs.items():
+                dbParams += f'{param} {value} '
+            sqlString += f'{dbParams}'
+
+        tsql.execute(sqlString)
+        tdLog.debug("complete to create database %s"%(dbName))
+        return
+
+    def create_stable(self,tsql, dbName,stbName,columnDict,tagDict):
+        colSchema = ''
+        for i in range(columnDict['int']):
+            colSchema += ', c%d int'%i
+        tagSchema = ''
+        for i in range(tagDict['int']):
+            if i > 0:
+                tagSchema += ','
+            tagSchema += 't%d int'%i
+
+        tsql.execute("create table if not exists %s.%s (ts timestamp %s) tags(%s)"%(dbName, stbName, colSchema, tagSchema))
+        tdLog.debug("complete to create %s.%s" %(dbName, stbName))
+        return
+
+    def create_ctables(self,tsql, dbName,stbName,ctbNum,tagDict):
+        tsql.execute("use %s" %dbName)        
+        tagsValues = ''
+        for i in range(tagDict['int']):
+            if i > 0:
+                tagsValues += ','
+            tagsValues += '%d'%i
+
+        pre_create = "create table"
+        sql = pre_create
+        #tdLog.debug("doing create one  stable %s and %d  child table in %s  ..." %(stbname, count ,dbname))
+        for i in range(ctbNum):
+            sql += " %s_%d using %s tags(%s)"%(stbName,i,stbName,tagsValues)
+            if (i > 0) and (i%100 == 0):
+                tsql.execute(sql)
+                sql = pre_create
+        if sql != pre_create:
+            tsql.execute(sql)
+        
+        tdLog.debug("complete to create %d child tables in %s.%s" %(ctbNum, dbName, stbName))
+        return
+
+    def insert_data(self,tsql,dbName,stbName,ctbNum,rowsPerTbl,batchNum,startTs=0):
+        tdLog.debug("start to insert data ............")
+        tsql.execute("use %s" %dbName)
+        pre_insert = "insert into "
+        sql = pre_insert
+        if startTs == 0:
+            t = time.time()
+            startTs = int(round(t * 1000))
+        #tdLog.debug("doing insert data into stable:%s rows:%d ..."%(stbName, allRows))
+        for i in range(ctbNum):
+            sql += " %s_%d values "%(stbName,i)
+            for j in range(rowsPerTbl):
+                sql += "(%d, %d, %d)"%(startTs + j, j, j)
+                if (j > 0) and ((j%batchNum == 0) or (j == rowsPerTbl - 1)):
+                    tsql.execute(sql)
+                    if j < rowsPerTbl - 1:
+                        sql = "insert into %s_%d values " %(stbName,i)
+                    else:
+                        sql = "insert into "
+        #end sql
+        if sql != pre_insert:
+            #print("insert sql:%s"%sql)
+            tsql.execute(sql)
+        tdLog.debug("insert data ............ [OK]")
+        return
+        
+    def getBuildPath(self):
+        selfPath = os.path.dirname(os.path.realpath(__file__))
+
+        if ("community" in selfPath):
+            projPath = selfPath[:selfPath.find("community")]
+        else:
+            projPath = selfPath[:selfPath.find("tests")]
+
+        for root, dirs, files in os.walk(projPath):
+            if ("taosd" in files or "taosd.exe" in files):
+                rootRealPath = os.path.dirname(os.path.realpath(root))
+                if ("packaging" not in rootRealPath):
+                    buildPath = root[:len(root) - len("/build/bin")]
+                    break
+        return buildPath        
+
+    def getClientCfgPath(self):
+        buildPath = self.getBuildPath()
+        if (buildPath == ""):
+            tdLog.exit("taosd not found!")
+        else:
+            tdLog.info("taosd found in %s" % buildPath)
+        cfgPath = buildPath + "/../sim/psim/cfg"
+        tdLog.info("cfgPath: %s" % cfgPath)
+        return cfgPath
+
+    def newcur(self,host='localhost',port=6030,user='root',password='taosdata'):
+        cfgPath = self.getClientCfgPath()
+        con=taos.connect(host=host, user=user, password=password, config=cfgPath, port=port)
+        cur=con.cursor()
+        print(cur)
+        return cur
+
+    def newTdSql(self, host='localhost',port=6030,user='root',password='taosdata'):
+        newTdSql = TDSql()
+        cur = self.newcur(host=host,port=port,user=user,password=password)
+        newTdSql.init(cur, False)
+        return newTdSql
+
 tdCom = TDCom()

tests/system-test/7-tmq/subscribeDb4.py

+import sys
+import time
+import socket
+import os
+import threading
+
+import taos
+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:
+    paraDict = {'dbName': 'db12',
+                'dropFlag':1,
+                'vgroups': 4,
+                'precision': 'ms',
+                'stbName': 'stb0',
+                'ctbNum': 10,
+                'rowsPerTbl': 10000,
+                'batchNum': 10,
+                'startTs': 0, # 1640966400000 ----> 2022-01-01 00:00:00.000
+                'event':'',
+                'columnDict': {'int':2},
+                'tagDict': {'int':1}
+                }
+
+    cdbName = 'cdb'
+    # some parameter to consumer processor
+    consumerId = 0
+    expectrowcnt = 0
+    topicList = ''
+    ifcheckdata = 0
+    ifManualCommit = 1
+    groupId = 'group.id:cgrp1'
+    autoCommit = 'enable.auto.commit:false'
+    autoCommitInterval = 'auto.commit.interval.ms:1000'
+    autoOffset = 'auto.offset.reset:earliest'
+
+    pollDelay = 20
+    showMsg   = 1
+    showRow   = 1  
+
+    hostname = socket.gethostname()
+
+    def init(self, conn, logSql):
+        tdLog.debug(f"start to excute {__file__}")
+        logSql = False
+        tdSql.init(conn.cursor(), logSql)
+
+    def tmqCase12(self):
+        tdLog.printNoPrefix("======== test case 12: ")
+        tdLog.info("step 1: create database, stb, ctb and insert data")
+        
+        tmqCom.initConsumerTable(self.cdbName)
+
+        tdCom.create_database(tdSql,self.paraDict["dbName"],self.paraDict["dropFlag"], self.paraDict['precision'])
+
+        self.paraDict["stbName"] = 'stb1'
+        tdCom.create_stable(tdSql,self.paraDict["dbName"],self.paraDict["stbName"],self.paraDict["columnDict"],self.paraDict["tagDict"])
+        tdCom.create_ctables(tdSql,self.paraDict["dbName"],self.paraDict["stbName"],self.paraDict["ctbNum"],self.paraDict["tagDict"])
+        tdCom.insert_data(tdSql,self.paraDict["dbName"],self.paraDict["stbName"],self.paraDict["ctbNum"],self.paraDict["rowsPerTbl"],self.paraDict["batchNum"])
+
+        self.paraDict["stbName"] = 'stb2'
+        tdCom.create_stable(tdSql,self.paraDict["dbName"],self.paraDict["stbName"],self.paraDict["columnDict"],self.paraDict["tagDict"])
+        tdCom.create_ctables(tdSql,self.paraDict["dbName"],self.paraDict["stbName"],self.paraDict["ctbNum"],self.paraDict["tagDict"])
+        tdCom.insert_data(tdSql,self.paraDict["dbName"],self.paraDict["stbName"],self.paraDict["ctbNum"],self.paraDict["rowsPerTbl"],self.paraDict["batchNum"])
+
+        tdLog.info("create topics from db")
+        topicName1 = 'topic_%s'%(self.paraDict['dbName'])        
+        tdSql.execute("create topic %s as database %s" %(topicName1, self.paraDict['dbName']))
+        
+        topicList = topicName1
+        keyList = '%s,%s,%s,%s'%(self.groupId,self.autoCommit,self.autoCommitInterval,self.autoOffset)
+        self.expectrowcnt = self.paraDict["rowsPerTbl"] * self.paraDict["ctbNum"] * 2
+        tmqCom.insertConsumerInfo(self.consumerId, self.expectrowcnt,topicList,keyList,self.ifcheckdata,self.ifManualCommit)
+        
+        tdLog.info("start consume processor")  
+        tmqCom.startTmqSimProcess(self.pollDelay,self.paraDict["dbName"],self.showMsg, self.showRow,self.cdbName)
+
+        tdLog.info("After waiting for a period of time, drop one stable")
+        time.sleep(10)              
+        tdSql.execute("drop table %s.%s" %(self.paraDict['dbName'], self.paraDict['stbName']))        
+
+        tdLog.info("wait result from consumer, then check it")
+        expectRows = 1
+        resultList = tmqCom.selectConsumeResult(expectRows)
+
+        totalConsumeRows = 0
+        for i in range(expectRows):
+            totalConsumeRows += resultList[i]
+        
+        if not (totalConsumeRows >= self.expectrowcnt/2 and totalConsumeRows <= self.expectrowcnt):
+            tdLog.info("act consume rows: %d, expect consume rows: between %d and %d"%(totalConsumeRows, self.expectrowcnt/2, self.expectrowcnt))
+            tdLog.exit("tmq consume rows error!")
+
+        time.sleep(15)
+        tdSql.query("drop topic %s"%topicName1)
+
+        tdLog.printNoPrefix("======== test case 12 end ...... ")
+
+    def run(self):
+        tdSql.prepare()
+        self.tmqCase12()
+
+    def stop(self):
+        tdSql.close()
+        tdLog.success(f"{__file__} successfully executed")
+
+event = threading.Event()
+
+tdCases.addLinux(__file__, TDTestCase())
+tdCases.addWindows(__file__, TDTestCase())

tests/system-test/7-tmq/subscribeStb4.py

@@ -196,16 +196,16 @@ class TDTestCase:
         auotCtbPrefix = 'autoCtb'
 
         # create and start thread
-        parameterDict = {'cfg':        '',       \
-                         'actionType': 0,        \
-                         'dbName':     'db1',    \
-                         'dropFlag':   1,        \
-                         'vgroups':    4,        \
-                         'replica':    1,        \
-                         'stbName':    'stb1',    \
-                         'ctbNum':     10,       \
-                         'rowsPerTbl': 10000,    \
-                         'batchNum':   100,      \
+        parameterDict = {'cfg':        '',
+                         'actionType': 0,
+                         'dbName':     'db1',
+                         'dropFlag':   1,
+                         'vgroups':    4,
+                         'replica':    1,
+                         'stbName':    'stb1',
+                         'ctbNum':     10,
+                         'rowsPerTbl': 10000,
+                         'batchNum':   100,
                          'startTs':    1640966400000}  # 2022-01-01 00:00:00.000
         parameterDict['cfg'] = cfgPath
         
@@ -349,7 +349,5 @@ class TDTestCase:
         tdSql.close()
         tdLog.success(f"{__file__} successfully executed")
 
-event = threading.Event()
-
 tdCases.addLinux(__file__, TDTestCase())
 tdCases.addWindows(__file__, TDTestCase())

tests/system-test/7-tmq/tmqCommon.py

+###################################################################
+#           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 collections import defaultdict
+import random
+import string
+import threading
+import requests
+import time
+# import socketfrom
+
+import taos
+from util.log import *
+from util.sql import *
+from util.cases import *
+from util.dnodes import *
+from util.common import *
+
+# class actionType(Enum):
+#     CREATE_DATABASE = 0
+#     CREATE_STABLE   = 1
+#     CREATE_CTABLE   = 2
+#     INSERT_DATA     = 3
+
+class TMQCom:
+    def init(self, conn, logSql):
+        tdSql.init(conn.cursor())
+        # tdSql.init(conn.cursor(), logSql)  # output sql.txt file
+
+    def initConsumerTable(self,cdbName='cdb'):        
+        tdLog.info("create consume database, and consume info table, and consume result table")
+        tdSql.query("create database if not exists %s vgroups 1"%(cdbName))
+        tdSql.query("drop table if exists %s.consumeinfo "%(cdbName))
+        tdSql.query("drop table if exists %s.consumeresult "%(cdbName))
+
+        tdSql.query("create table %s.consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int, ifmanualcommit int)"%cdbName)
+        tdSql.query("create table %s.consumeresult (ts timestamp, consumerid int, consummsgcnt bigint, consumrowcnt bigint, checkresult int)"%cdbName)
+
+    def initConsumerInfoTable(self,cdbName='cdb'):        
+        tdLog.info("drop consumeinfo table")
+        tdSql.query("drop table if exists %s.consumeinfo "%(cdbName))
+        tdSql.query("create table %s.consumeinfo (ts timestamp, consumerid int, topiclist binary(1024), keylist binary(1024), expectmsgcnt bigint, ifcheckdata int, ifmanualcommit int)"%cdbName)
+
+    def insertConsumerInfo(self,consumerId, expectrowcnt,topicList,keyList,ifcheckdata,ifmanualcommit,cdbName='cdb'):    
+        sql = "insert into %s.consumeinfo values "%cdbName
+        sql += "(now, %d, '%s', '%s', %d, %d, %d)"%(consumerId, topicList, keyList, expectrowcnt, ifcheckdata, ifmanualcommit)
+        tdLog.info("consume info sql: %s"%sql)
+        tdSql.query(sql)
+
+    def selectConsumeResult(self,expectRows,cdbName='cdb'):
+        resultList=[]
+        while 1:
+            tdSql.query("select * from %s.consumeresult"%cdbName)
+            #tdLog.info("row: %d, %l64d, %l64d"%(tdSql.getData(0, 1),tdSql.getData(0, 2),tdSql.getData(0, 3))
+            if tdSql.getRows() == expectRows:
+                break
+            else:
+                time.sleep(5)
+        
+        for i in range(expectRows):
+            tdLog.info ("consume id: %d, consume msgs: %d, consume rows: %d"%(tdSql.getData(i , 1), tdSql.getData(i , 2), tdSql.getData(i , 3)))
+            resultList.append(tdSql.getData(i , 3))
+        
+        return resultList
+
+    def startTmqSimProcess(self,pollDelay,dbName,showMsg=1,showRow=1,cdbName='cdb',valgrind=0):
+        buildPath = tdCom.getBuildPath()
+        cfgPath = tdCom.getClientCfgPath()
+        if valgrind == 1:
+            logFile = cfgPath + '/../log/valgrind-tmq.log'
+            shellCmd = 'nohup valgrind --log-file=' + logFile
+            shellCmd += '--tool=memcheck --leak-check=full --show-reachable=no --track-origins=yes --show-leak-kinds=all --num-callers=20 -v --workaround-gcc296-bugs=yes '
+        
+        if (platform.system().lower() == 'windows'):
+            shellCmd = 'mintty -h never -w hide ' + buildPath + '\\build\\bin\\tmq_sim.exe -c ' + cfgPath
+            shellCmd += " -y %d -d %s -g %d -r %d -w %s "%(pollDelay, dbName, showMsg, showRow, cdbName) 
+            shellCmd += "> nul 2>&1 &"   
+        else:
+            shellCmd = 'nohup ' + buildPath + '/build/bin/tmq_sim -c ' + cfgPath
+            shellCmd += " -y %d -d %s -g %d -r %d -w %s "%(pollDelay, dbName, showMsg, showRow, cdbName) 
+            shellCmd += "> /dev/null 2>&1 &"
+        tdLog.info(shellCmd)
+        os.system(shellCmd)
+
+    def syncCreateDbStbCtbInsertData(self, tsql, paraDict):
+        tdCom.create_database(tsql, paraDict["dbName"],paraDict["dropFlag"], paraDict['precision'])
+        tdCom.create_stable(tsql, paraDict["dbName"],paraDict["stbName"], paraDict["columnDict"], paraDict["tagDict"])
+        tdCom.create_ctables(tsql, paraDict["dbName"],paraDict["stbName"],paraDict["ctbNum"],paraDict["tagDict"])
+        if "event" in paraDict and type(paraDict['event']) == type(threading.Event()):
+            paraDict["event"].set()
+        tdCom.insert_data(tsql,paraDict["dbName"],paraDict["stbName"],paraDict["ctbNum"],paraDict["rowsPerTbl"],paraDict["batchNum"],paraDict["startTs"])
+        return 
+
+    def threadFunction(self, **paraDict):
+        # create new connector for new tdSql instance in my thread
+        newTdSql = tdCom.newTdSql()
+        self.syncCreateDbStbCtbInsertData(self, newTdSql, paraDict)
+        return
+
+    def asyncCreateDbStbCtbInsertData(self, paraDict):
+        pThread = threading.Thread(target=self.threadFunction, kwargs=paraDict)
+        pThread.start()
+        return pThread
+
+    def close(self):
+        self.cursor.close()
+
+tmqCom = TMQCom()