Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
6a98d55c
T
TDengine
项目概览
taosdata
/
TDengine
大约 1 年 前同步成功
通知
1184
Star
22015
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,发现更多精彩内容 >>
提交
6a98d55c
编写于
4月 12, 2023
作者:
L
liuyao
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
add ci test
上级
ed4aaade
变更
4
隐藏空白更改
内联
并排
Showing
4 changed file
with
95 addition
and
106 deletion
+95
-106
include/common/tcommon.h
include/common/tcommon.h
+4
-4
source/libs/executor/src/timewindowoperator.c
source/libs/executor/src/timewindowoperator.c
+4
-40
tests/parallel_test/cases.task
tests/parallel_test/cases.task
+58
-42
tests/script/win-test-file
tests/script/win-test-file
+29
-20
未找到文件。
include/common/tcommon.h
浏览文件 @
6a98d55c
...
...
@@ -51,15 +51,15 @@ static inline int winKeyCmprImpl(const void* pKey1, const void* pKey2) {
SWinKey
*
pWin1
=
(
SWinKey
*
)
pKey1
;
SWinKey
*
pWin2
=
(
SWinKey
*
)
pKey2
;
if
(
pWin1
->
ts
>
pWin2
->
ts
)
{
if
(
pWin1
->
groupId
>
pWin2
->
groupId
)
{
return
1
;
}
else
if
(
pWin1
->
ts
<
pWin2
->
ts
)
{
}
else
if
(
pWin1
->
groupId
<
pWin2
->
groupId
)
{
return
-
1
;
}
if
(
pWin1
->
groupId
>
pWin2
->
groupId
)
{
if
(
pWin1
->
ts
>
pWin2
->
ts
)
{
return
1
;
}
else
if
(
pWin1
->
groupId
<
pWin2
->
groupId
)
{
}
else
if
(
pWin1
->
ts
<
pWin2
->
ts
)
{
return
-
1
;
}
...
...
source/libs/executor/src/timewindowoperator.c
浏览文件 @
6a98d55c
...
...
@@ -1454,42 +1454,8 @@ STimeWindow getFinalTimeWindow(int64_t ts, SInterval* pInterval) {
return
w
;
}
static
void
deleteIntervalDiscBuf
(
SStreamState
*
pState
,
SHashObj
*
pPullDataMap
,
TSKEY
mark
,
SInterval
*
pInterval
,
SWinKey
*
key
)
{
STimeWindow
tw
=
getFinalTimeWindow
(
key
->
ts
,
pInterval
);
SWinKey
next
=
{
0
};
while
(
tw
.
ekey
<
mark
)
{
SStreamStateCur
*
pCur
=
streamStateSeekKeyNext
(
pState
,
key
);
int32_t
code
=
streamStateGetKVByCur
(
pCur
,
&
next
,
NULL
,
0
);
streamStateFreeCur
(
pCur
);
void
*
chIds
=
taosHashGet
(
pPullDataMap
,
key
,
sizeof
(
SWinKey
));
if
(
chIds
&&
pPullDataMap
)
{
SArray
*
chAy
=
*
(
SArray
**
)
chIds
;
int32_t
size
=
taosArrayGetSize
(
chAy
);
qDebug
(
"===stream===window %"
PRId64
" wait child size:%d"
,
key
->
ts
,
size
);
for
(
int32_t
i
=
0
;
i
<
size
;
i
++
)
{
qDebug
(
"===stream===window %"
PRId64
" wait child id:%d"
,
key
->
ts
,
*
(
int32_t
*
)
taosArrayGet
(
chAy
,
i
));
}
break
;
}
qDebug
(
"===stream===delete window %"
PRId64
,
key
->
ts
);
int32_t
codeDel
=
streamStateDel
(
pState
,
key
);
if
(
codeDel
!=
TSDB_CODE_SUCCESS
)
{
code
=
streamStateGetFirst
(
pState
,
key
);
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
qDebug
(
"===stream===stream state first key: empty-empty"
);
return
;
}
continue
;
}
if
(
code
==
TSDB_CODE_SUCCESS
)
{
*
key
=
next
;
tw
=
getFinalTimeWindow
(
key
->
ts
,
pInterval
);
}
else
{
break
;
}
}
static
void
deleteIntervalDiscBuf
(
SStreamState
*
pState
,
TSKEY
mark
)
{
//todo(liuyao) delete expired check point
}
static
void
closeChildIntervalWindow
(
SOperatorInfo
*
pOperator
,
SArray
*
pChildren
,
TSKEY
maxTs
)
{
...
...
@@ -2609,8 +2575,7 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) {
setStreamDataVersion
(
pTaskInfo
,
pInfo
->
dataVersion
,
pInfo
->
pState
->
checkPointId
);
qDebug
(
"===stream===clear semi operator"
);
}
else
{
deleteIntervalDiscBuf
(
pInfo
->
pState
,
pInfo
->
pPullDataMap
,
pInfo
->
twAggSup
.
maxTs
-
pInfo
->
twAggSup
.
deleteMark
,
&
pInfo
->
interval
,
&
pInfo
->
delKey
);
deleteIntervalDiscBuf
(
pInfo
->
pState
,
pInfo
->
twAggSup
.
maxTs
-
pInfo
->
twAggSup
.
deleteMark
);
if
(
pInfo
->
twAggSup
.
maxTs
-
pInfo
->
twAggSup
.
checkPointInterval
>
pInfo
->
twAggSup
.
checkPointTs
)
{
streamStateCommit
(
pInfo
->
pState
);
pInfo
->
twAggSup
.
checkPointTs
=
pInfo
->
twAggSup
.
maxTs
;
...
...
@@ -4848,8 +4813,7 @@ static SSDataBlock* doStreamIntervalAgg(SOperatorInfo* pOperator) {
printDataBlock
(
pInfo
->
binfo
.
pRes
,
"single interval"
);
return
pInfo
->
binfo
.
pRes
;
}
deleteIntervalDiscBuf
(
pInfo
->
pState
,
NULL
,
pInfo
->
twAggSup
.
maxTs
-
pInfo
->
twAggSup
.
deleteMark
,
&
pInfo
->
interval
,
&
pInfo
->
delKey
);
deleteIntervalDiscBuf
(
pInfo
->
pState
,
pInfo
->
twAggSup
.
maxTs
-
pInfo
->
twAggSup
.
deleteMark
);
setOperatorCompleted
(
pOperator
);
if
(
pInfo
->
twAggSup
.
maxTs
-
pInfo
->
twAggSup
.
checkPointInterval
>
pInfo
->
twAggSup
.
checkPointTs
)
{
streamStateCommit
(
pInfo
->
pState
);
...
...
tests/parallel_test/cases.task
浏览文件 @
6a98d55c
...
...
@@ -223,40 +223,47 @@
,,n,script,./test.sh -f tsim/stream/basic0.sim -g
,,y,script,./test.sh -f tsim/stream/basic1.sim
,,y,script,./test.sh -f tsim/stream/basic2.sim
,,y,script,./test.sh -f tsim/stream/basic3.sim
,,y,script,./test.sh -f tsim/stream/basic4.sim
,,y,script,./test.sh -f tsim/stream/checkStreamSTable1.sim
,,y,script,./test.sh -f tsim/stream/checkStreamSTable.sim
,,y,script,./test.sh -f tsim/stream/deleteInterval.sim
,,y,script,./test.sh -f tsim/stream/deleteSession.sim
,,y,script,./test.sh -f tsim/stream/deleteState.sim
,,y,script,./test.sh -f tsim/stream/distributeInterval0.sim
,,y,script,./test.sh -f tsim/stream/distributeIntervalRetrive0.sim
,,y,script,./test.sh -f tsim/stream/distributeSession0.sim
,,y,script,./test.sh -f tsim/stream/drop_stream.sim
,,y,script,./test.sh -f tsim/stream/fillHistoryBasic1.sim
,,y,script,./test.sh -f tsim/stream/fillHistoryBasic2.sim
,,y,script,./test.sh -f tsim/stream/fillHistoryBasic3.sim
,,y,script,./test.sh -f tsim/stream/distributeInterval0.sim
,,y,script,./test.sh -f tsim/stream/distributeIntervalRetrive0.sim
,,y,script,./test.sh -f tsim/stream/distributeSession0.sim
,,y,script,./test.sh -f tsim/stream/session0.sim
,,y,script,./test.sh -f tsim/stream/session1.sim
,,y,script,./test.sh -f tsim/stream/state0.sim
,,y,script,./test.sh -f tsim/stream/triggerInterval0.sim
,,y,script,./test.sh -f tsim/stream/triggerSession0.sim
,,y,script,./test.sh -f tsim/stream/partitionby.sim
,,y,script,./test.sh -f tsim/stream/partitionby1.sim
,,y,script,./test.sh -f tsim/stream/schedSnode.sim
,,y,script,./test.sh -f tsim/stream/windowClose.sim
,,y,script,./test.sh -f tsim/stream/ignoreExpiredData.sim
,,y,script,./test.sh -f tsim/stream/sliding.sim
,,y,script,./test.sh -f tsim/stream/partitionbyColumnInterval.sim
,,y,script,./test.sh -f tsim/stream/partitionbyColumnSession.sim
,,y,script,./test.sh -f tsim/stream/partitionbyColumnState.sim
,,y,script,./test.sh -f tsim/stream/deleteInterval.sim
,,y,script,./test.sh -f tsim/stream/deleteSession.sim
,,y,script,./test.sh -f tsim/stream/deleteState.sim
,,y,script,./test.sh -f tsim/stream/fillIntervalDelete0.sim
,,y,script,./test.sh -f tsim/stream/fillIntervalDelete1.sim
,,y,script,./test.sh -f tsim/stream/fillIntervalLinear.sim
,,y,script,./test.sh -f tsim/stream/fillIntervalPartitionBy.sim
,,y,script,./test.sh -f tsim/stream/fillIntervalPrevNext1.sim
,,y,script,./test.sh -f tsim/stream/fillIntervalPrevNext.sim
,,y,script,./test.sh -f tsim/stream/fillIntervalRange.sim
,,y,script,./test.sh -f tsim/stream/fillIntervalValue.sim
,,y,script,./test.sh -f tsim/stream/ignoreCheckUpdate.sim
,,y,script,./test.sh -f tsim/stream/ignoreExpiredData.sim
,,y,script,./test.sh -f tsim/stream/partitionby1.sim
,,y,script,./test.sh -f tsim/stream/partitionbyColumnInterval.sim
,,y,script,./test.sh -f tsim/stream/partitionbyColumnSession.sim
,,y,script,./test.sh -f tsim/stream/partitionbyColumnState.sim
,,y,script,./test.sh -f tsim/stream/partitionby.sim
,,y,script,./test.sh -f tsim/stream/schedSnode.sim
,,y,script,./test.sh -f tsim/stream/session0.sim
,,y,script,./test.sh -f tsim/stream/session1.sim
,,y,script,./test.sh -f tsim/stream/sliding.sim
,,y,script,./test.sh -f tsim/stream/state0.sim
,,y,script,./test.sh -f tsim/stream/state1.sim
,,y,script,./test.sh -f tsim/stream/triggerInterval0.sim
,,y,script,./test.sh -f tsim/stream/triggerSession0.sim
,,y,script,./test.sh -f tsim/stream/udTableAndTag0.sim
,,y,script,./test.sh -f tsim/stream/udTableAndTag1.sim
,,y,script,./test.sh -f tsim/stream/
checkStreamSTable
.sim
,,y,script,./test.sh -f tsim/stream/
checkStreamSTable1
.sim
,,y,script,./test.sh -f tsim/stream/
udTableAndTag2
.sim
,,y,script,./test.sh -f tsim/stream/
windowClose
.sim
,,y,script,./test.sh -f tsim/trans/lossdata1.sim
,,y,script,./test.sh -f tsim/trans/create_db.sim
,,y,script,./test.sh -f tsim/tmq/basic1.sim
...
...
@@ -1334,38 +1341,47 @@
,,n,script,./test.sh -f tsim/stream/basic0.sim -g
,,y,script,./test.sh -f tsim/stream/basic1.sim
,,y,script,./test.sh -f tsim/stream/basic2.sim
,,y,script,./test.sh -f tsim/stream/basic3.sim
,,y,script,./test.sh -f tsim/stream/basic4.sim
,,y,script,./test.sh -f tsim/stream/checkStreamSTable1.sim
,,y,script,./test.sh -f tsim/stream/checkStreamSTable.sim
,,y,script,./test.sh -f tsim/stream/deleteInterval.sim
,,y,script,./test.sh -f tsim/stream/deleteSession.sim
,,y,script,./test.sh -f tsim/stream/deleteState.sim
,,y,script,./test.sh -f tsim/stream/distributeInterval0.sim
,,y,script,./test.sh -f tsim/stream/distributeIntervalRetrive0.sim
,,y,script,./test.sh -f tsim/stream/distributeSession0.sim
,,y,script,./test.sh -f tsim/stream/drop_stream.sim
,,y,script,./test.sh -f tsim/stream/fillHistoryBasic1.sim
,,y,script,./test.sh -f tsim/stream/fillHistoryBasic2.sim
,,y,script,./test.sh -f tsim/stream/fillHistoryBasic3.sim
,,y,script,./test.sh -f tsim/stream/distributeInterval0.sim
,,y,script,./test.sh -f tsim/stream/distributeIntervalRetrive0.sim
,,y,script,./test.sh -f tsim/stream/distributeSession0.sim
,,y,script,./test.sh -f tsim/stream/session0.sim
,,y,script,./test.sh -f tsim/stream/session1.sim
,,y,script,./test.sh -f tsim/stream/state0.sim
,,y,script,./test.sh -f tsim/stream/triggerInterval0.sim
,,y,script,./test.sh -f tsim/stream/triggerSession0.sim
,,y,script,./test.sh -f tsim/stream/partitionby.sim
,,y,script,./test.sh -f tsim/stream/partitionby1.sim
,,y,script,./test.sh -f tsim/stream/schedSnode.sim
,,y,script,./test.sh -f tsim/stream/windowClose.sim
,,y,script,./test.sh -f tsim/stream/ignoreExpiredData.sim
,,y,script,./test.sh -f tsim/stream/sliding.sim
,,y,script,./test.sh -f tsim/stream/partitionbyColumnInterval.sim
,,y,script,./test.sh -f tsim/stream/partitionbyColumnSession.sim
,,y,script,./test.sh -f tsim/stream/partitionbyColumnState.sim
,,y,script,./test.sh -f tsim/stream/deleteInterval.sim
,,y,script,./test.sh -f tsim/stream/deleteSession.sim
,,y,script,./test.sh -f tsim/stream/deleteState.sim
,,y,script,./test.sh -f tsim/stream/fillIntervalDelete0.sim
,,y,script,./test.sh -f tsim/stream/fillIntervalDelete1.sim
,,y,script,./test.sh -f tsim/stream/fillIntervalLinear.sim
,,y,script,./test.sh -f tsim/stream/fillIntervalPartitionBy.sim
,,y,script,./test.sh -f tsim/stream/fillIntervalPrevNext1.sim
,,y,script,./test.sh -f tsim/stream/fillIntervalPrevNext.sim
,,y,script,./test.sh -f tsim/stream/fillIntervalRange.sim
,,y,script,./test.sh -f tsim/stream/fillIntervalValue.sim
,,y,script,./test.sh -f tsim/stream/ignoreCheckUpdate.sim
,,y,script,./test.sh -f tsim/stream/ignoreExpiredData.sim
,,y,script,./test.sh -f tsim/stream/partitionby1.sim
,,y,script,./test.sh -f tsim/stream/partitionbyColumnInterval.sim
,,y,script,./test.sh -f tsim/stream/partitionbyColumnSession.sim
,,y,script,./test.sh -f tsim/stream/partitionbyColumnState.sim
,,y,script,./test.sh -f tsim/stream/partitionby.sim
,,y,script,./test.sh -f tsim/stream/schedSnode.sim
,,y,script,./test.sh -f tsim/stream/session0.sim
,,y,script,./test.sh -f tsim/stream/session1.sim
,,y,script,./test.sh -f tsim/stream/sliding.sim
,,y,script,./test.sh -f tsim/stream/state0.sim
,,y,script,./test.sh -f tsim/stream/state1.sim
,,y,script,./test.sh -f tsim/stream/triggerInterval0.sim
,,y,script,./test.sh -f tsim/stream/triggerSession0.sim
,,y,script,./test.sh -f tsim/stream/udTableAndTag0.sim
,,y,script,./test.sh -f tsim/stream/udTableAndTag1.sim
,,y,script,./test.sh -f tsim/stream/udTableAndTag2.sim
,,y,script,./test.sh -f tsim/stream/windowClose.sim
,,y,script,./test.sh -f tsim/trans/lossdata1.sim
,,y,script,./test.sh -f tsim/tmq/basic1.sim
,,y,script,./test.sh -f tsim/tmq/basic2.sim
...
...
tests/script/win-test-file
浏览文件 @
6a98d55c
...
...
@@ -213,38 +213,47 @@ rem ./test.sh -f tsim/db/alter_replica_13.sim
./test.sh -f tsim/stream/basic0.sim -g
./test.sh -f tsim/stream/basic1.sim
./test.sh -f tsim/stream/basic2.sim
./test.sh -f tsim/stream/basic3.sim
./test.sh -f tsim/stream/basic4.sim
./test.sh -f tsim/stream/checkStreamSTable1.sim
./test.sh -f tsim/stream/checkStreamSTable.sim
./test.sh -f tsim/stream/deleteInterval.sim
./test.sh -f tsim/stream/deleteSession.sim
./test.sh -f tsim/stream/deleteState.sim
./test.sh -f tsim/stream/distributeInterval0.sim
./test.sh -f tsim/stream/distributeIntervalRetrive0.sim
./test.sh -f tsim/stream/distributeSession0.sim
./test.sh -f tsim/stream/drop_stream.sim
./test.sh -f tsim/stream/fillHistoryBasic1.sim
./test.sh -f tsim/stream/fillHistoryBasic2.sim
./test.sh -f tsim/stream/fillHistoryBasic3.sim
./test.sh -f tsim/stream/distributeInterval0.sim
./test.sh -f tsim/stream/distributeIntervalRetrive0.sim
./test.sh -f tsim/stream/distributeSession0.sim
./test.sh -f tsim/stream/session0.sim
./test.sh -f tsim/stream/session1.sim
./test.sh -f tsim/stream/state0.sim
./test.sh -f tsim/stream/triggerInterval0.sim
./test.sh -f tsim/stream/triggerSession0.sim
./test.sh -f tsim/stream/partitionby.sim
./test.sh -f tsim/stream/partitionby1.sim
./test.sh -f tsim/stream/schedSnode.sim
./test.sh -f tsim/stream/windowClose.sim
./test.sh -f tsim/stream/ignoreExpiredData.sim
./test.sh -f tsim/stream/sliding.sim
./test.sh -f tsim/stream/partitionbyColumnInterval.sim
./test.sh -f tsim/stream/partitionbyColumnSession.sim
./test.sh -f tsim/stream/partitionbyColumnState.sim
./test.sh -f tsim/stream/deleteInterval.sim
./test.sh -f tsim/stream/deleteSession.sim
./test.sh -f tsim/stream/deleteState.sim
./test.sh -f tsim/stream/fillIntervalDelete0.sim
./test.sh -f tsim/stream/fillIntervalDelete1.sim
./test.sh -f tsim/stream/fillIntervalLinear.sim
./test.sh -f tsim/stream/fillIntervalPartitionBy.sim
./test.sh -f tsim/stream/fillIntervalPrevNext1.sim
./test.sh -f tsim/stream/fillIntervalPrevNext.sim
./test.sh -f tsim/stream/fillIntervalRange.sim
./test.sh -f tsim/stream/fillIntervalValue.sim
./test.sh -f tsim/stream/ignoreCheckUpdate.sim
./test.sh -f tsim/stream/ignoreExpiredData.sim
./test.sh -f tsim/stream/partitionby1.sim
./test.sh -f tsim/stream/partitionbyColumnInterval.sim
./test.sh -f tsim/stream/partitionbyColumnSession.sim
./test.sh -f tsim/stream/partitionbyColumnState.sim
./test.sh -f tsim/stream/partitionby.sim
./test.sh -f tsim/stream/schedSnode.sim
./test.sh -f tsim/stream/session0.sim
./test.sh -f tsim/stream/session1.sim
./test.sh -f tsim/stream/sliding.sim
./test.sh -f tsim/stream/state0.sim
./test.sh -f tsim/stream/state1.sim
./test.sh -f tsim/stream/triggerInterval0.sim
./test.sh -f tsim/stream/triggerSession0.sim
./test.sh -f tsim/stream/udTableAndTag0.sim
./test.sh -f tsim/stream/udTableAndTag1.sim
./test.sh -f tsim/stream/udTableAndTag2.sim
./test.sh -f tsim/stream/windowClose.sim
./test.sh -f tsim/trans/lossdata1.sim
./test.sh -f tsim/trans/create_db.sim
./test.sh -f tsim/tmq/basic1.sim
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录