Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
yiicenne
TDengine
提交
2105b033
TDengine
项目概览
yiicenne
/
TDengine
与 Fork 源项目一致
Fork自
taosdata / TDengine
通知
1
Star
0
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
TDengine
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
2105b033
编写于
6月 02, 2021
作者:
L
lichuang
浏览文件
操作
浏览文件
下载
差异文件
Merge branch 'develop' into feature/TD-3963
上级
cdce05d1
945674e2
变更
9
隐藏空白更改
内联
并排
Showing
9 changed file
with
117 addition
and
19 deletion
+117
-19
src/client/inc/tsclient.h
src/client/inc/tsclient.h
+1
-0
src/client/src/tscStream.c
src/client/src/tscStream.c
+94
-9
src/common/inc/tglobal.h
src/common/inc/tglobal.h
+2
-1
src/common/src/tglobal.c
src/common/src/tglobal.c
+3
-2
src/cq/src/cqMain.c
src/cq/src/cqMain.c
+5
-2
src/inc/taosdef.h
src/inc/taosdef.h
+2
-0
src/mnode/src/mnodeDnode.c
src/mnode/src/mnodeDnode.c
+1
-1
src/sync/src/syncMain.c
src/sync/src/syncMain.c
+6
-1
tests/pytest/util/dnodes.py
tests/pytest/util/dnodes.py
+3
-3
未找到文件。
src/client/inc/tsclient.h
浏览文件 @
2105b033
...
...
@@ -283,6 +283,7 @@ typedef struct SSqlStream {
int64_t
ctime
;
// stream created 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
ltime
;
// stream last row time in stream table
SInterval
interval
;
void
*
pTimer
;
...
...
src/client/src/tscStream.c
浏览文件 @
2105b033
...
...
@@ -24,6 +24,7 @@
#include "tutil.h"
#include "tscProfile.h"
#include "tscSubquery.h"
static
void
tscProcessStreamQueryCallback
(
void
*
param
,
TAOS_RES
*
tres
,
int
numOfRows
);
static
void
tscProcessStreamRetrieveResult
(
void
*
param
,
TAOS_RES
*
res
,
int
numOfRows
);
...
...
@@ -47,8 +48,8 @@ static bool isProjectStream(SQueryInfo* pQueryInfo) {
static
int64_t
tscGetRetryDelayTime
(
SSqlStream
*
pStream
,
int64_t
slidingTime
,
int16_t
prec
)
{
float
retryRangeFactor
=
0
.
3
f
;
int64_t
retryDelta
=
(
int64_t
)(
ts
StreamCompRetry
Delay
*
retryRangeFactor
);
retryDelta
=
((
rand
()
%
retryDelta
)
+
ts
StreamCompRetry
Delay
)
*
1000L
;
int64_t
retryDelta
=
(
int64_t
)(
ts
RetryStreamComp
Delay
*
retryRangeFactor
);
retryDelta
=
((
rand
()
%
retryDelta
)
+
ts
RetryStreamComp
Delay
)
*
1000L
;
if
(
pStream
->
interval
.
intervalUnit
!=
'n'
&&
pStream
->
interval
.
intervalUnit
!=
'y'
)
{
// change to ms
...
...
@@ -575,6 +576,14 @@ static void tscCreateStream(void *param, TAOS_RES *res, int code) {
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
);
pCmd
->
command
=
TSDB_SQL_SELECT
;
...
...
@@ -590,7 +599,66 @@ void tscSetStreamDestTable(SSqlStream* pStream, const char* 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
*
))
{
STscObj
*
pObj
=
(
STscObj
*
)
taos
;
if
(
pObj
==
NULL
||
pObj
->
signature
!=
pObj
)
return
NULL
;
...
...
@@ -613,11 +681,16 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p
return
NULL
;
}
pStream
->
stime
=
stime
;
pStream
->
fp
=
fp
;
pStream
->
ltime
=
INT64_MIN
;
pStream
->
stime
=
stime
;
pStream
->
fp
=
fp
;
pStream
->
callback
=
callback
;
pStream
->
param
=
param
;
pStream
->
pSql
=
pSql
;
pStream
->
param
=
param
;
pStream
->
pSql
=
pSql
;
pSql
->
pStream
=
pStream
;
pSql
->
param
=
pStream
;
pSql
->
maxRetry
=
TSDB_MAX_REPLICA
;
tscSetStreamDestTable
(
pStream
,
dstTable
);
pSql
->
pStream
=
pStream
;
pSql
->
param
=
pStream
;
...
...
@@ -640,10 +713,17 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p
tscDebugL
(
"0x%"
PRIx64
" SQL: %s"
,
pSql
->
self
,
pSql
->
sqlstr
);
pSql
->
fp
=
cbParseSql
;
pSql
->
fetchFp
=
cbParseSql
;
registerSqlObj
(
pSql
);
int32_t
code
=
tsParseSql
(
pSql
,
true
);
if
(
code
==
TSDB_CODE_SUCCESS
)
{
tscCreateStream
(
pStream
,
pSql
,
code
);
}
else
if
(
code
!=
TSDB_CODE_TSC_ACTION_IN_PROGRESS
)
{
cbParseSql
(
pStream
,
pSql
,
code
);
}
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
));
taosReleaseRef
(
tscObjRef
,
pSql
->
self
);
free
(
pStream
);
...
...
@@ -653,6 +733,11 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p
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
)
{
SSqlStream
*
pStream
=
(
SSqlStream
*
)
handle
;
...
...
src/common/inc/tglobal.h
浏览文件 @
2105b033
...
...
@@ -39,6 +39,7 @@ extern int8_t tsEnableTelemetryReporting;
extern
char
tsEmail
[];
extern
char
tsArbitrator
[];
extern
int8_t
tsArbOnline
;
extern
int64_t
tsArbOnlineTimestamp
;
extern
int32_t
tsDnodeId
;
// common
...
...
@@ -75,7 +76,7 @@ extern int32_t tsMinSlidingTime;
extern
int32_t
tsMinIntervalTime
;
extern
int32_t
tsMaxStreamComputDelay
;
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
int32_t
tsProjectExecInterval
;
extern
int64_t
tsMaxRetentWindow
;
...
...
src/common/src/tglobal.c
浏览文件 @
2105b033
...
...
@@ -42,6 +42,7 @@ int32_t tsNumOfMnodes = 3;
int8_t
tsEnableVnodeBak
=
1
;
int8_t
tsEnableTelemetryReporting
=
1
;
int8_t
tsArbOnline
=
0
;
int64_t
tsArbOnlineTimestamp
=
TSDB_ARB_DUMMY_TIME
;
char
tsEmail
[
TSDB_FQDN_LEN
]
=
{
0
};
int32_t
tsDnodeId
=
0
;
...
...
@@ -93,7 +94,7 @@ int32_t tsMaxStreamComputDelay = 20000;
int32_t
tsStreamCompStartDelay
=
10000
;
// 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.
float
tsStreamComputDelayRatio
=
0
.
1
f
;
...
...
@@ -710,7 +711,7 @@ static void doInitGlobalConfig(void) {
taosInitConfigOption
(
cfg
);
cfg
.
option
=
"retryStreamCompDelay"
;
cfg
.
ptr
=
&
ts
StreamCompRetry
Delay
;
cfg
.
ptr
=
&
ts
RetryStreamComp
Delay
;
cfg
.
valType
=
TAOS_CFG_VTYPE_INT32
;
cfg
.
cfgType
=
TSDB_CFG_CTYPE_B_CONFIG
|
TSDB_CFG_CTYPE_B_SHOW
;
cfg
.
minValue
=
10
;
...
...
src/cq/src/cqMain.c
浏览文件 @
2105b033
...
...
@@ -437,6 +437,10 @@ static void cqProcessCreateTimer(void *param, void *tmrId) {
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
)
{
pObj
->
pContext
=
pContext
;
...
...
@@ -449,11 +453,10 @@ static void cqCreateStream(SCqContext *pContext, SCqObj *pObj) {
pObj
->
tmrId
=
0
;
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
if
(
pObj
->
pStream
)
{
tscSetStreamDestTable
(
pObj
->
pStream
,
pObj
->
dstTable
);
pContext
->
num
++
;
cDebug
(
"vgId:%d, id:%d CQ:%s is opened"
,
pContext
->
vgId
,
pObj
->
tid
,
pObj
->
sqlStr
);
}
else
{
...
...
src/inc/taosdef.h
浏览文件 @
2105b033
...
...
@@ -375,6 +375,8 @@ do { \
#define TSDB_MAX_WAL_SIZE (1024*1024*3)
#define TSDB_ARB_DUMMY_TIME 4765104000000 // 2121-01-01 00:00:00.000, :P
typedef
enum
{
TAOS_QTYPE_RPC
=
0
,
TAOS_QTYPE_FWD
=
1
,
...
...
src/mnode/src/mnodeDnode.c
浏览文件 @
2105b033
...
...
@@ -941,7 +941,7 @@ static int32_t mnodeRetrieveDnodes(SShowObj *pShow, char *data, int32_t rows, vo
cols
++
;
pWrite
=
data
+
pShow
->
offset
[
cols
]
*
rows
+
pShow
->
bytes
[
cols
]
*
numOfRows
;
*
(
int64_t
*
)
pWrite
=
0
;
*
(
int64_t
*
)
pWrite
=
tsArbOnlineTimestamp
;
cols
++
;
pWrite
=
data
+
pShow
->
offset
[
cols
]
*
rows
+
pShow
->
bytes
[
cols
]
*
numOfRows
;
...
...
src/sync/src/syncMain.c
浏览文件 @
2105b033
...
...
@@ -1150,7 +1150,12 @@ static void syncSetupPeerConnection(SSyncPeer *pPeer) {
pPeer
->
peerFd
=
connFd
;
pPeer
->
role
=
TAOS_SYNC_ROLE_UNSYNCED
;
pPeer
->
pConn
=
syncAllocateTcpConn
(
tsTcpPool
,
pPeer
->
rid
,
connFd
);
if
(
pPeer
->
isArb
)
tsArbOnline
=
1
;
if
(
pPeer
->
isArb
)
{
tsArbOnline
=
1
;
if
(
tsArbOnlineTimestamp
==
TSDB_ARB_DUMMY_TIME
)
{
tsArbOnlineTimestamp
=
taosGetTimestampMs
();
}
}
}
else
{
sDebug
(
"%s, failed to setup peer connection to server since %s, try later"
,
pPeer
->
id
,
strerror
(
errno
));
taosCloseSocket
(
connFd
);
...
...
tests/pytest/util/dnodes.py
浏览文件 @
2105b033
...
...
@@ -432,7 +432,7 @@ class TDDnodes:
self
.
simDeployed
=
False
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"
)
while
(
processID
):
killCmd
=
"kill -TERM %s > /dev/null 2>&1"
%
processID
...
...
@@ -545,14 +545,14 @@ class TDDnodes:
for
i
in
range
(
len
(
self
.
dnodes
)):
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"
)
if
processID
:
cmd
=
"sudo systemctl stop taosd"
os
.
system
(
cmd
)
# if os.system(cmd) != 0 :
# 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"
)
while
(
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.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录