Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
5fb7723b
T
TDengine
项目概览
taosdata
/
TDengine
12 个月 前同步成功
通知
1180
Star
22014
Fork
4786
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
1
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
T
TDengine
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
1
Issue
1
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
未验证
提交
5fb7723b
编写于
6月 28, 2022
作者:
H
Hui Li
提交者:
GitHub
6月 28, 2022
浏览文件
操作
浏览文件
下载
差异文件
Merge pull request #14302 from taosdata/test3.0/lihui
test: add test case for tmq
上级
64cc8bd8
2f69ba5f
变更
2
隐藏空白更改
内联
并排
Showing
2 changed file
with
124 addition
and
6 deletion
+124
-6
tests/system-test/7-tmq/stbTagFilter.py
tests/system-test/7-tmq/stbTagFilter.py
+113
-0
tests/system-test/7-tmq/tmqCommon.py
tests/system-test/7-tmq/tmqCommon.py
+11
-6
未找到文件。
tests/system-test/7-tmq/stbTagFilter.py
0 → 100644
浏览文件 @
5fb7723b
import
taos
import
sys
import
time
import
socket
import
os
import
threading
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
,
conn
,
logSql
):
tdLog
.
debug
(
f
"start to excute
{
__file__
}
"
)
tdSql
.
init
(
conn
.
cursor
())
#tdSql.init(conn.cursor(), logSql) # output sql.txt file
def
tmqCase1
(
self
):
tdLog
.
printNoPrefix
(
"======== test case 1: "
)
paraDict
=
{
'dbName'
:
'db2'
,
'dropFlag'
:
1
,
'event'
:
''
,
'vgroups'
:
4
,
'stbName'
:
'stb'
,
'colPrefix'
:
'c'
,
'tagPrefix'
:
't'
,
'colSchema'
:
[{
'type'
:
'INT'
,
'count'
:
2
},
{
'type'
:
'binary'
,
'len'
:
20
,
'count'
:
1
},{
'type'
:
'TIMESTAMP'
,
'count'
:
1
}],
'tagSchema'
:
[{
'type'
:
'INT'
,
'count'
:
1
},
{
'type'
:
'binary'
,
'len'
:
20
,
'count'
:
1
}],
'ctbPrefix'
:
'ctb'
,
'ctbStartIdx'
:
0
,
'ctbNum'
:
10
,
'rowsPerTbl'
:
1000
,
'batchNum'
:
10
,
'startTs'
:
1640966400000
,
# 2022-01-01 00:00:00.000
'pollDelay'
:
10
,
'showMsg'
:
1
,
'showRow'
:
1
}
topicNameList
=
[
'topic1'
]
expectRowsList
=
[]
tmqCom
.
initConsumerTable
()
tdCom
.
create_database
(
tdSql
,
paraDict
[
"dbName"
],
paraDict
[
"dropFlag"
],
vgroups
=
4
,
replica
=
1
)
tdLog
.
info
(
"create stb"
)
tdCom
.
create_stable
(
tdSql
,
dbname
=
paraDict
[
"dbName"
],
stbname
=
paraDict
[
"stbName"
],
column_elm_list
=
paraDict
[
'colSchema'
],
tag_elm_list
=
paraDict
[
'tagSchema'
])
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
.
asyncInsertData
(
paraDict
)
tdLog
.
info
(
"create topics from stb with filter"
)
# queryString = "select ts, sin(c1), pow(c2,3) from %s.%s where t2 == 'beijing' or t2 == 'changsha'" %(paraDict['dbName'], paraDict['stbName'])
queryString
=
"select * from %s.%s where t2 == 'beijing' or t2 == 'changsha'"
%
(
paraDict
[
'dbName'
],
paraDict
[
'stbName'
])
sqlString
=
"create topic %s as %s"
%
(
topicNameList
[
0
],
queryString
)
tdLog
.
info
(
"create topic sql: %s"
%
sqlString
)
tdSql
.
execute
(
sqlString
)
# start tmq consume processor
tdLog
.
info
(
"insert consume info to consume processor"
)
consumerId
=
0
expectrowcnt
=
paraDict
[
"rowsPerTbl"
]
*
paraDict
[
"ctbNum"
]
*
2
topicList
=
topicNameList
[
0
]
ifcheckdata
=
0
ifManualCommit
=
1
keyList
=
'group.id:cgrp1, enable.auto.commit:false, auto.commit.interval.ms:2000, auto.offset.reset:earliest'
tmqCom
.
insertConsumerInfo
(
consumerId
,
expectrowcnt
,
topicList
,
keyList
,
ifcheckdata
,
ifManualCommit
)
tdLog
.
info
(
"start consume processor"
)
tmqCom
.
startTmqSimProcess
(
paraDict
[
'pollDelay'
],
paraDict
[
"dbName"
],
paraDict
[
'showMsg'
],
paraDict
[
'showRow'
])
# tmqCom.getStartCommitNotifyFromTmqsim()
tmqCom
.
getStartConsumeNotifyFromTmqsim
()
tdLog
.
info
(
"create some new ctb"
)
paraDict
[
'ctbStartIdx'
]
=
paraDict
[
'ctbStartIdx'
]
+
paraDict
[
'ctbNum'
]
tmqCom
.
create_ctable
(
tdSql
,
dbName
=
paraDict
[
"dbName"
],
stbName
=
paraDict
[
"stbName"
],
ctbPrefix
=
paraDict
[
'ctbPrefix'
],
ctbNum
=
paraDict
[
'ctbNum'
],
ctbStartIdx
=
paraDict
[
'ctbStartIdx'
])
tdLog
.
info
(
"insert data into new ctb"
)
pThread
=
tmqCom
.
asyncInsertData
(
paraDict
)
pThread
.
join
()
tdLog
.
info
(
"wait insert end"
)
tdSql
.
query
(
queryString
)
expectRowsList
.
append
(
tdSql
.
getRows
())
tdLog
.
info
(
"wait the consume result"
)
expectRows
=
1
resultList
=
tmqCom
.
selectConsumeResult
(
expectRows
)
if
expectRowsList
[
0
]
!=
resultList
[
0
]:
tdLog
.
info
(
"expect consume rows: %d, act consume rows: %d"
%
(
expectRowsList
[
0
],
resultList
[
0
]))
tdLog
.
exit
(
"0 tmq consume rows error!"
)
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
.
tmqCase1
()
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/tmqCommon.py
浏览文件 @
5fb7723b
...
...
@@ -159,7 +159,7 @@ class TMQCom:
tdLog
.
debug
(
"complete to create %s.%s"
%
(
dbName
,
stbName
))
return
def
create_ctable
(
self
,
tsql
=
None
,
dbName
=
'dbx'
,
stbName
=
'stb'
,
ctbPrefix
=
'ctb'
,
ctbNum
=
1
):
def
create_ctable
(
self
,
tsql
=
None
,
dbName
=
'dbx'
,
stbName
=
'stb'
,
ctbPrefix
=
'ctb'
,
ctbNum
=
1
,
ctbStartIdx
=
0
):
tsql
.
execute
(
"use %s"
%
dbName
)
pre_create
=
"create table"
sql
=
pre_create
...
...
@@ -168,8 +168,10 @@ class TMQCom:
tagValue
=
'beijing'
if
(
i
%
2
==
0
):
tagValue
=
'shanghai'
elif
(
i
%
3
==
0
):
tagValue
=
'changsha'
sql
+=
" %s%d using %s tags(%d, '%s')"
%
(
ctbPrefix
,
i
,
stbName
,
i
+
1
,
tagValue
)
sql
+=
" %s%d using %s tags(%d, '%s')"
%
(
ctbPrefix
,
i
+
ctbStartIdx
,
stbName
,
i
+
ctbStartIdx
+
1
,
tagValue
)
if
(
i
>
0
)
and
(
i
%
100
==
0
):
tsql
.
execute
(
sql
)
sql
=
pre_create
...
...
@@ -235,7 +237,7 @@ class TMQCom:
tdLog
.
debug
(
"insert data ............ [OK]"
)
return
def
insert_data_2
(
self
,
tsql
,
dbName
,
ctbPrefix
,
ctbNum
,
rowsPerTbl
,
batchNum
,
startTs
):
def
insert_data_2
(
self
,
tsql
,
dbName
,
ctbPrefix
,
ctbNum
,
rowsPerTbl
,
batchNum
,
startTs
,
ctbStartIdx
=
0
):
tdLog
.
debug
(
"start to insert data ............"
)
tsql
.
execute
(
"use %s"
%
dbName
)
pre_insert
=
"insert into "
...
...
@@ -245,7 +247,7 @@ class TMQCom:
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 "
%
(
ctbPrefix
,
i
)
sql
+=
" %s%d values "
%
(
ctbPrefix
,
i
+
ctbStartIdx
)
for
j
in
range
(
rowsPerTbl
):
if
(
j
%
2
==
0
):
sql
+=
"(%d, %d, %d, 'tmqrow_%d', now) "
%
(
startTs
+
j
,
j
,
j
,
j
)
...
...
@@ -254,7 +256,7 @@ class TMQCom:
if
(
j
>
0
)
and
((
j
%
batchNum
==
0
)
or
(
j
==
rowsPerTbl
-
1
)):
tsql
.
execute
(
sql
)
if
j
<
rowsPerTbl
-
1
:
sql
=
"insert into %s%d values "
%
(
ctbPrefix
,
i
)
sql
=
"insert into %s%d values "
%
(
ctbPrefix
,
i
+
ctbStartIdx
)
else
:
sql
=
"insert into "
#end sql
...
...
@@ -354,7 +356,10 @@ class TMQCom:
def
threadFunctionForInsert
(
self
,
**
paraDict
):
# create new connector for new tdSql instance in my thread
newTdSql
=
tdCom
.
newTdSql
()
self
.
insert_data_2
(
newTdSql
,
paraDict
[
"dbName"
],
paraDict
[
"ctbPrefix"
],
paraDict
[
"ctbNum"
],
paraDict
[
"rowsPerTbl"
],
paraDict
[
"batchNum"
],
paraDict
[
"startTs"
])
if
'ctbStartIdx'
in
paraDict
.
keys
():
self
.
insert_data_2
(
newTdSql
,
paraDict
[
"dbName"
],
paraDict
[
"ctbPrefix"
],
paraDict
[
"ctbNum"
],
paraDict
[
"rowsPerTbl"
],
paraDict
[
"batchNum"
],
paraDict
[
"startTs"
],
paraDict
[
"ctbStartIdx"
])
else
:
self
.
insert_data_2
(
newTdSql
,
paraDict
[
"dbName"
],
paraDict
[
"ctbPrefix"
],
paraDict
[
"ctbNum"
],
paraDict
[
"rowsPerTbl"
],
paraDict
[
"batchNum"
],
paraDict
[
"startTs"
])
return
def
asyncInsertData
(
self
,
paraDict
):
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录