Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
慢慢CG
TDengine
提交
054ab0d8
T
TDengine
项目概览
慢慢CG
/
TDengine
与 Fork 源项目一致
Fork自
taosdata / TDengine
通知
1
Star
0
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
T
TDengine
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
提交
054ab0d8
编写于
6月 01, 2021
作者:
B
bryanchang0603
浏览文件
操作
浏览文件
下载
差异文件
Merge branch 'master' into test/TD-4422
上级
f26a6a23
506da276
变更
6
显示空白变更内容
内联
并排
Showing
6 changed file
with
97 addition
and
16 deletion
+97
-16
src/client/inc/tsclient.h
src/client/inc/tsclient.h
+1
-0
src/client/src/tscStream.c
src/client/src/tscStream.c
+85
-8
src/common/inc/tglobal.h
src/common/inc/tglobal.h
+1
-1
src/common/src/tglobal.c
src/common/src/tglobal.c
+2
-2
src/cq/src/cqMain.c
src/cq/src/cqMain.c
+5
-2
tests/pytest/util/dnodes.py
tests/pytest/util/dnodes.py
+3
-3
未找到文件。
src/client/inc/tsclient.h
浏览文件 @
054ab0d8
...
@@ -422,6 +422,7 @@ typedef struct SSqlStream {
...
@@ -422,6 +422,7 @@ typedef struct SSqlStream {
int64_t
ctime
;
// stream created time
int64_t
ctime
;
// stream created time
int64_t
stime
;
// stream next executed time
int64_t
stime
;
// stream next executed time
int64_t
etime
;
// stream end query time, when time is larger then etime, the stream will be closed
int64_t
etime
;
// stream end query time, when time is larger then etime, the stream will be closed
int64_t
ltime
;
// stream last row time in stream table
SInterval
interval
;
SInterval
interval
;
void
*
pTimer
;
void
*
pTimer
;
...
...
src/client/src/tscStream.c
浏览文件 @
054ab0d8
...
@@ -25,6 +25,7 @@
...
@@ -25,6 +25,7 @@
#include "tutil.h"
#include "tutil.h"
#include "tscProfile.h"
#include "tscProfile.h"
#include "tscSubquery.h"
static
void
tscProcessStreamQueryCallback
(
void
*
param
,
TAOS_RES
*
tres
,
int
numOfRows
);
static
void
tscProcessStreamQueryCallback
(
void
*
param
,
TAOS_RES
*
tres
,
int
numOfRows
);
static
void
tscProcessStreamRetrieveResult
(
void
*
param
,
TAOS_RES
*
res
,
int
numOfRows
);
static
void
tscProcessStreamRetrieveResult
(
void
*
param
,
TAOS_RES
*
res
,
int
numOfRows
);
...
@@ -48,8 +49,8 @@ static bool isProjectStream(SQueryInfo* pQueryInfo) {
...
@@ -48,8 +49,8 @@ static bool isProjectStream(SQueryInfo* pQueryInfo) {
static
int64_t
tscGetRetryDelayTime
(
SSqlStream
*
pStream
,
int64_t
slidingTime
,
int16_t
prec
)
{
static
int64_t
tscGetRetryDelayTime
(
SSqlStream
*
pStream
,
int64_t
slidingTime
,
int16_t
prec
)
{
float
retryRangeFactor
=
0
.
3
f
;
float
retryRangeFactor
=
0
.
3
f
;
int64_t
retryDelta
=
(
int64_t
)(
ts
StreamCompRetry
Delay
*
retryRangeFactor
);
int64_t
retryDelta
=
(
int64_t
)(
ts
RetryStreamComp
Delay
*
retryRangeFactor
);
retryDelta
=
((
rand
()
%
retryDelta
)
+
ts
StreamCompRetry
Delay
)
*
1000L
;
retryDelta
=
((
rand
()
%
retryDelta
)
+
ts
RetryStreamComp
Delay
)
*
1000L
;
if
(
pStream
->
interval
.
intervalUnit
!=
'n'
&&
pStream
->
interval
.
intervalUnit
!=
'y'
)
{
if
(
pStream
->
interval
.
intervalUnit
!=
'n'
&&
pStream
->
interval
.
intervalUnit
!=
'y'
)
{
// change to ms
// change to ms
...
@@ -572,6 +573,14 @@ static void tscCreateStream(void *param, TAOS_RES *res, int code) {
...
@@ -572,6 +573,14 @@ static void tscCreateStream(void *param, TAOS_RES *res, int code) {
pStream
->
stime
=
tscGetStreamStartTimestamp
(
pSql
,
pStream
,
pStream
->
stime
);
pStream
->
stime
=
tscGetStreamStartTimestamp
(
pSql
,
pStream
,
pStream
->
stime
);
// set stime with ltime if ltime > stime
const
char
*
dstTable
=
pStream
->
dstTable
?
pStream
->
dstTable
:
""
;
tscDebug
(
" CQ table=%s ltime is %"
PRId64
,
dstTable
,
pStream
->
ltime
);
if
(
pStream
->
ltime
!=
INT64_MIN
&&
pStream
->
ltime
>
pStream
->
stime
)
{
tscWarn
(
" CQ set stream %s stime=%"
PRId64
" replace with ltime=%"
PRId64
" if ltime>0 "
,
dstTable
,
pStream
->
stime
,
pStream
->
ltime
);
pStream
->
stime
=
pStream
->
ltime
;
}
int64_t
starttime
=
tscGetLaunchTimestamp
(
pStream
);
int64_t
starttime
=
tscGetLaunchTimestamp
(
pStream
);
pCmd
->
command
=
TSDB_SQL_SELECT
;
pCmd
->
command
=
TSDB_SQL_SELECT
;
...
@@ -587,7 +596,66 @@ void tscSetStreamDestTable(SSqlStream* pStream, const char* dstTable) {
...
@@ -587,7 +596,66 @@ void tscSetStreamDestTable(SSqlStream* pStream, const char* dstTable) {
pStream
->
dstTable
=
dstTable
;
pStream
->
dstTable
=
dstTable
;
}
}
TAOS_STREAM
*
taos_open_stream
(
TAOS
*
taos
,
const
char
*
sqlstr
,
void
(
*
fp
)(
void
*
param
,
TAOS_RES
*
,
TAOS_ROW
row
),
// fetchFp call back
void
fetchFpStreamLastRow
(
void
*
param
,
TAOS_RES
*
res
,
int
num
)
{
SSqlStream
*
pStream
=
(
SSqlStream
*
)
param
;
SSqlObj
*
pSql
=
res
;
// get row data set to ltime
tscSetSqlOwner
(
pSql
);
TAOS_ROW
row
=
doSetResultRowData
(
pSql
);
if
(
row
&&
row
[
0
]
)
{
pStream
->
ltime
=
*
((
int64_t
*
)
row
[
0
]);
const
char
*
dstTable
=
pStream
->
dstTable
?
pStream
->
dstTable
:
""
;
tscDebug
(
" CQ stream table=%s last row time=%"
PRId64
" ."
,
dstTable
,
pStream
->
ltime
);
}
tscClearSqlOwner
(
pSql
);
// no condition call
tscCreateStream
(
param
,
pStream
->
pSql
,
TSDB_CODE_SUCCESS
);
taos_free_result
(
res
);
}
// fp callback
void
fpStreamLastRow
(
void
*
param
,
TAOS_RES
*
res
,
int
code
)
{
// check result successful
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
tscCreateStream
(
param
,
res
,
TSDB_CODE_SUCCESS
);
taos_free_result
(
res
);
return
;
}
// asynchronous fetch last row data
taos_fetch_rows_a
(
res
,
fetchFpStreamLastRow
,
param
);
}
void
cbParseSql
(
void
*
param
,
TAOS_RES
*
res
,
int
code
)
{
// check result successful
SSqlStream
*
pStream
=
(
SSqlStream
*
)
param
;
SSqlObj
*
pSql
=
pStream
->
pSql
;
SSqlCmd
*
pCmd
=
&
pSql
->
cmd
;
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
pSql
->
res
.
code
=
code
;
tscDebug
(
"0x%"
PRIx64
" open stream parse sql failed, sql:%s, reason:%s, code:%s"
,
pSql
->
self
,
pSql
->
sqlstr
,
pCmd
->
payload
,
tstrerror
(
code
));
pStream
->
fp
(
pStream
->
param
,
NULL
,
NULL
);
return
;
}
// check dstTable valid
if
(
pStream
->
dstTable
==
NULL
||
strlen
(
pStream
->
dstTable
)
==
0
)
{
tscDebug
(
" cbParseSql dstTable is empty."
);
tscCreateStream
(
param
,
res
,
code
);
return
;
}
// query stream last row time async
char
sql
[
128
]
=
""
;
sprintf
(
sql
,
"select last_row(*) from %s;"
,
pStream
->
dstTable
);
taos_query_a
(
pSql
->
pTscObj
,
sql
,
fpStreamLastRow
,
param
);
return
;
}
TAOS_STREAM
*
taos_open_stream_withname
(
TAOS
*
taos
,
const
char
*
dstTable
,
const
char
*
sqlstr
,
void
(
*
fp
)(
void
*
param
,
TAOS_RES
*
,
TAOS_ROW
row
),
int64_t
stime
,
void
*
param
,
void
(
*
callback
)(
void
*
))
{
int64_t
stime
,
void
*
param
,
void
(
*
callback
)(
void
*
))
{
STscObj
*
pObj
=
(
STscObj
*
)
taos
;
STscObj
*
pObj
=
(
STscObj
*
)
taos
;
if
(
pObj
==
NULL
||
pObj
->
signature
!=
pObj
)
return
NULL
;
if
(
pObj
==
NULL
||
pObj
->
signature
!=
pObj
)
return
NULL
;
...
@@ -610,6 +678,7 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p
...
@@ -610,6 +678,7 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p
return
NULL
;
return
NULL
;
}
}
pStream
->
ltime
=
INT64_MIN
;
pStream
->
stime
=
stime
;
pStream
->
stime
=
stime
;
pStream
->
fp
=
fp
;
pStream
->
fp
=
fp
;
pStream
->
callback
=
callback
;
pStream
->
callback
=
callback
;
...
@@ -618,6 +687,7 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p
...
@@ -618,6 +687,7 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p
pSql
->
pStream
=
pStream
;
pSql
->
pStream
=
pStream
;
pSql
->
param
=
pStream
;
pSql
->
param
=
pStream
;
pSql
->
maxRetry
=
TSDB_MAX_REPLICA
;
pSql
->
maxRetry
=
TSDB_MAX_REPLICA
;
tscSetStreamDestTable
(
pStream
,
dstTable
);
pSql
->
sqlstr
=
calloc
(
1
,
strlen
(
sqlstr
)
+
1
);
pSql
->
sqlstr
=
calloc
(
1
,
strlen
(
sqlstr
)
+
1
);
if
(
pSql
->
sqlstr
==
NULL
)
{
if
(
pSql
->
sqlstr
==
NULL
)
{
...
@@ -632,15 +702,17 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p
...
@@ -632,15 +702,17 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p
tscDebugL
(
"%p SQL: %s"
,
pSql
,
pSql
->
sqlstr
);
tscDebugL
(
"%p SQL: %s"
,
pSql
,
pSql
->
sqlstr
);
tsem_init
(
&
pSql
->
rspSem
,
0
,
0
);
tsem_init
(
&
pSql
->
rspSem
,
0
,
0
);
pSql
->
fp
=
tscCreateStream
;
pSql
->
fp
=
cbParseSql
;
pSql
->
fetchFp
=
tscCreateStream
;
pSql
->
fetchFp
=
cbParseSql
;
registerSqlObj
(
pSql
);
registerSqlObj
(
pSql
);
int32_t
code
=
tsParseSql
(
pSql
,
true
);
int32_t
code
=
tsParseSql
(
pSql
,
true
);
if
(
code
==
TSDB_CODE_SUCCESS
)
{
if
(
code
==
TSDB_CODE_SUCCESS
)
{
tscCreateStream
(
pStream
,
pSql
,
code
);
cbParseSql
(
pStream
,
pSql
,
code
);
}
else
if
(
code
!=
TSDB_CODE_TSC_ACTION_IN_PROGRESS
)
{
}
else
if
(
code
==
TSDB_CODE_TSC_ACTION_IN_PROGRESS
)
{
tscDebug
(
" CQ taso_open_stream IN Process. sql=%s"
,
sqlstr
);
}
else
{
tscError
(
"0x%"
PRIx64
" open stream failed, sql:%s, code:%s"
,
pSql
->
self
,
sqlstr
,
tstrerror
(
code
));
tscError
(
"0x%"
PRIx64
" open stream failed, sql:%s, code:%s"
,
pSql
->
self
,
sqlstr
,
tstrerror
(
code
));
taosReleaseRef
(
tscObjRef
,
pSql
->
self
);
taosReleaseRef
(
tscObjRef
,
pSql
->
self
);
free
(
pStream
);
free
(
pStream
);
...
@@ -650,6 +722,11 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p
...
@@ -650,6 +722,11 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p
return
pStream
;
return
pStream
;
}
}
TAOS_STREAM
*
taos_open_stream
(
TAOS
*
taos
,
const
char
*
sqlstr
,
void
(
*
fp
)(
void
*
param
,
TAOS_RES
*
,
TAOS_ROW
row
),
int64_t
stime
,
void
*
param
,
void
(
*
callback
)(
void
*
))
{
return
taos_open_stream_withname
(
taos
,
""
,
sqlstr
,
fp
,
stime
,
param
,
callback
);
}
void
taos_close_stream
(
TAOS_STREAM
*
handle
)
{
void
taos_close_stream
(
TAOS_STREAM
*
handle
)
{
SSqlStream
*
pStream
=
(
SSqlStream
*
)
handle
;
SSqlStream
*
pStream
=
(
SSqlStream
*
)
handle
;
...
...
src/common/inc/tglobal.h
浏览文件 @
054ab0d8
...
@@ -74,7 +74,7 @@ extern int32_t tsMinSlidingTime;
...
@@ -74,7 +74,7 @@ extern int32_t tsMinSlidingTime;
extern
int32_t
tsMinIntervalTime
;
extern
int32_t
tsMinIntervalTime
;
extern
int32_t
tsMaxStreamComputDelay
;
extern
int32_t
tsMaxStreamComputDelay
;
extern
int32_t
tsStreamCompStartDelay
;
extern
int32_t
tsStreamCompStartDelay
;
extern
int32_t
ts
StreamCompRetry
Delay
;
extern
int32_t
ts
RetryStreamComp
Delay
;
extern
float
tsStreamComputDelayRatio
;
// the delayed computing ration of the whole time window
extern
float
tsStreamComputDelayRatio
;
// the delayed computing ration of the whole time window
extern
int32_t
tsProjectExecInterval
;
extern
int32_t
tsProjectExecInterval
;
extern
int64_t
tsMaxRetentWindow
;
extern
int64_t
tsMaxRetentWindow
;
...
...
src/common/src/tglobal.c
浏览文件 @
054ab0d8
...
@@ -92,7 +92,7 @@ int32_t tsMaxStreamComputDelay = 20000;
...
@@ -92,7 +92,7 @@ int32_t tsMaxStreamComputDelay = 20000;
int32_t
tsStreamCompStartDelay
=
10000
;
int32_t
tsStreamCompStartDelay
=
10000
;
// the stream computing delay time after executing failed, change accordingly
// the stream computing delay time after executing failed, change accordingly
int32_t
ts
StreamCompRetryDelay
=
1
0
;
int32_t
ts
RetryStreamCompDelay
=
10
*
100
0
;
// The delayed computing ration. 10% of the whole computing time window by default.
// The delayed computing ration. 10% of the whole computing time window by default.
float
tsStreamComputDelayRatio
=
0
.
1
f
;
float
tsStreamComputDelayRatio
=
0
.
1
f
;
...
@@ -696,7 +696,7 @@ static void doInitGlobalConfig(void) {
...
@@ -696,7 +696,7 @@ static void doInitGlobalConfig(void) {
taosInitConfigOption
(
cfg
);
taosInitConfigOption
(
cfg
);
cfg
.
option
=
"retryStreamCompDelay"
;
cfg
.
option
=
"retryStreamCompDelay"
;
cfg
.
ptr
=
&
ts
StreamCompRetry
Delay
;
cfg
.
ptr
=
&
ts
RetryStreamComp
Delay
;
cfg
.
valType
=
TAOS_CFG_VTYPE_INT32
;
cfg
.
valType
=
TAOS_CFG_VTYPE_INT32
;
cfg
.
cfgType
=
TSDB_CFG_CTYPE_B_CONFIG
|
TSDB_CFG_CTYPE_B_SHOW
;
cfg
.
cfgType
=
TSDB_CFG_CTYPE_B_CONFIG
|
TSDB_CFG_CTYPE_B_SHOW
;
cfg
.
minValue
=
10
;
cfg
.
minValue
=
10
;
...
...
src/cq/src/cqMain.c
浏览文件 @
054ab0d8
...
@@ -437,6 +437,10 @@ static void cqProcessCreateTimer(void *param, void *tmrId) {
...
@@ -437,6 +437,10 @@ static void cqProcessCreateTimer(void *param, void *tmrId) {
taosReleaseRef
(
cqObjRef
,
(
int64_t
)
param
);
taosReleaseRef
(
cqObjRef
,
(
int64_t
)
param
);
}
}
// inner implement in tscStream.c
TAOS_STREAM
*
taos_open_stream_withname
(
TAOS
*
taos
,
const
char
*
desName
,
const
char
*
sqlstr
,
void
(
*
fp
)(
void
*
param
,
TAOS_RES
*
,
TAOS_ROW
row
),
int64_t
stime
,
void
*
param
,
void
(
*
callback
)(
void
*
));
static
void
cqCreateStream
(
SCqContext
*
pContext
,
SCqObj
*
pObj
)
{
static
void
cqCreateStream
(
SCqContext
*
pContext
,
SCqObj
*
pObj
)
{
pObj
->
pContext
=
pContext
;
pObj
->
pContext
=
pContext
;
...
@@ -449,11 +453,10 @@ static void cqCreateStream(SCqContext *pContext, SCqObj *pObj) {
...
@@ -449,11 +453,10 @@ static void cqCreateStream(SCqContext *pContext, SCqObj *pObj) {
pObj
->
tmrId
=
0
;
pObj
->
tmrId
=
0
;
if
(
pObj
->
pStream
==
NULL
)
{
if
(
pObj
->
pStream
==
NULL
)
{
pObj
->
pStream
=
taos_open_stream
(
pContext
->
dbConn
,
pObj
->
sqlStr
,
cqProcessStreamRes
,
INT64_MIN
,
(
void
*
)
pObj
->
rid
,
NULL
);
pObj
->
pStream
=
taos_open_stream
_withname
(
pContext
->
dbConn
,
pObj
->
dstTable
,
pObj
->
sqlStr
,
cqProcessStreamRes
,
INT64_MIN
,
(
void
*
)
pObj
->
rid
,
NULL
);
// TODO the pObj->pStream may be released if error happens
// TODO the pObj->pStream may be released if error happens
if
(
pObj
->
pStream
)
{
if
(
pObj
->
pStream
)
{
tscSetStreamDestTable
(
pObj
->
pStream
,
pObj
->
dstTable
);
pContext
->
num
++
;
pContext
->
num
++
;
cDebug
(
"vgId:%d, id:%d CQ:%s is opened"
,
pContext
->
vgId
,
pObj
->
tid
,
pObj
->
sqlStr
);
cDebug
(
"vgId:%d, id:%d CQ:%s is opened"
,
pContext
->
vgId
,
pObj
->
tid
,
pObj
->
sqlStr
);
}
else
{
}
else
{
...
...
tests/pytest/util/dnodes.py
浏览文件 @
054ab0d8
...
@@ -432,7 +432,7 @@ class TDDnodes:
...
@@ -432,7 +432,7 @@ class TDDnodes:
self
.
simDeployed
=
False
self
.
simDeployed
=
False
def
init
(
self
,
path
):
def
init
(
self
,
path
):
psCmd
=
"ps -ef|grep -w taosd| grep -v grep | awk '{print $2}'"
psCmd
=
"ps -ef|grep -w taosd| grep -v grep
| grep -v defunct
| awk '{print $2}'"
processID
=
subprocess
.
check_output
(
psCmd
,
shell
=
True
).
decode
(
"utf-8"
)
processID
=
subprocess
.
check_output
(
psCmd
,
shell
=
True
).
decode
(
"utf-8"
)
while
(
processID
):
while
(
processID
):
killCmd
=
"kill -TERM %s > /dev/null 2>&1"
%
processID
killCmd
=
"kill -TERM %s > /dev/null 2>&1"
%
processID
...
@@ -545,14 +545,14 @@ class TDDnodes:
...
@@ -545,14 +545,14 @@ class TDDnodes:
for
i
in
range
(
len
(
self
.
dnodes
)):
for
i
in
range
(
len
(
self
.
dnodes
)):
self
.
dnodes
[
i
].
stop
()
self
.
dnodes
[
i
].
stop
()
psCmd
=
"ps -ef | grep -w taosd | grep 'root' | grep -v grep | awk '{print $2}'"
psCmd
=
"ps -ef | grep -w taosd | grep 'root' | grep -v grep
| grep -v defunct
| awk '{print $2}'"
processID
=
subprocess
.
check_output
(
psCmd
,
shell
=
True
).
decode
(
"utf-8"
)
processID
=
subprocess
.
check_output
(
psCmd
,
shell
=
True
).
decode
(
"utf-8"
)
if
processID
:
if
processID
:
cmd
=
"sudo systemctl stop taosd"
cmd
=
"sudo systemctl stop taosd"
os
.
system
(
cmd
)
os
.
system
(
cmd
)
# if os.system(cmd) != 0 :
# if os.system(cmd) != 0 :
# tdLog.exit(cmd)
# tdLog.exit(cmd)
psCmd
=
"ps -ef|grep -w taosd| grep -v grep | awk '{print $2}'"
psCmd
=
"ps -ef|grep -w taosd| grep -v grep
| grep -v defunct
| awk '{print $2}'"
processID
=
subprocess
.
check_output
(
psCmd
,
shell
=
True
).
decode
(
"utf-8"
)
processID
=
subprocess
.
check_output
(
psCmd
,
shell
=
True
).
decode
(
"utf-8"
)
while
(
processID
):
while
(
processID
):
killCmd
=
"kill -TERM %s > /dev/null 2>&1"
%
processID
killCmd
=
"kill -TERM %s > /dev/null 2>&1"
%
processID
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录