Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
e89362fb
T
TDengine
项目概览
taosdata
/
TDengine
1 年多 前同步成功
通知
1185
Star
22017
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看板
提交
e89362fb
编写于
2月 07, 2023
作者:
5
54liuyao
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
feat:stream ignore check update data
上级
1e0e4172
变更
6
隐藏空白更改
内联
并排
Showing
6 changed file
with
72 addition
and
3 deletion
+72
-3
include/libs/nodes/plannodes.h
include/libs/nodes/plannodes.h
+1
-0
source/libs/executor/inc/executorimpl.h
source/libs/executor/inc/executorimpl.h
+2
-0
source/libs/executor/src/groupoperator.c
source/libs/executor/src/groupoperator.c
+1
-1
source/libs/executor/src/scanoperator.c
source/libs/executor/src/scanoperator.c
+4
-0
source/libs/executor/src/timewindowoperator.c
source/libs/executor/src/timewindowoperator.c
+2
-2
tests/script/tsim/stream/ignoreCheckUpdate.sim
tests/script/tsim/stream/ignoreCheckUpdate.sim
+62
-0
未找到文件。
include/libs/nodes/plannodes.h
浏览文件 @
e89362fb
...
...
@@ -357,6 +357,7 @@ typedef struct STableScanPhysiNode {
int64_t
watermark
;
int8_t
igExpired
;
bool
assignBlockUid
;
int8_t
igCheckUpdate
;
}
STableScanPhysiNode
;
typedef
STableScanPhysiNode
STableSeqScanPhysiNode
;
...
...
source/libs/executor/inc/executorimpl.h
浏览文件 @
e89362fb
...
...
@@ -474,6 +474,8 @@ typedef struct SStreamScanInfo {
int32_t
blockRecoverContiCnt
;
int32_t
blockRecoverTotCnt
;
int8_t
igCheckUpdate
;
int8_t
igExpired
;
}
SStreamScanInfo
;
typedef
struct
{
...
...
source/libs/executor/src/groupoperator.c
浏览文件 @
e89362fb
...
...
@@ -1109,7 +1109,7 @@ void initParDownStream(SOperatorInfo* downstream, SPartitionBySupporter* pParSup
SStreamScanInfo
*
pScanInfo
=
downstream
->
info
;
pScanInfo
->
partitionSup
=
*
pParSup
;
pScanInfo
->
pPartScalarSup
=
pExpr
;
if
(
!
pScanInfo
->
pUpdateInfo
)
{
if
(
!
pScanInfo
->
igCheckUpdate
&&
!
pScanInfo
->
pUpdateInfo
)
{
pScanInfo
->
pUpdateInfo
=
updateInfoInit
(
60000
,
TSDB_TIME_PRECISION_MILLI
,
0
);
}
}
...
...
source/libs/executor/src/scanoperator.c
浏览文件 @
e89362fb
...
...
@@ -2368,6 +2368,10 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys
pInfo
->
pUpdateDataRes
=
createSpecialDataBlock
(
STREAM_CLEAR
);
pInfo
->
assignBlockUid
=
pTableScanNode
->
assignBlockUid
;
pInfo
->
partitionSup
.
needCalc
=
false
;
//todo(liuyao) for test
pTableScanNode
->
igCheckUpdate
=
true
;
pInfo
->
igCheckUpdate
=
pTableScanNode
->
igCheckUpdate
;
pInfo
->
igExpired
=
pTableScanNode
->
igExpired
;
setOperatorInfo
(
pOperator
,
"StreamScanOperator"
,
QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN
,
false
,
OP_NOT_OPENED
,
pInfo
,
pTaskInfo
);
...
...
source/libs/executor/src/timewindowoperator.c
浏览文件 @
e89362fb
...
...
@@ -1726,7 +1726,7 @@ void initIntervalDownStream(SOperatorInfo* downstream, uint16_t type, SAggSuppor
SStreamScanInfo
*
pScanInfo
=
downstream
->
info
;
pScanInfo
->
windowSup
.
parentType
=
type
;
pScanInfo
->
windowSup
.
pIntervalAggSup
=
pSup
;
if
(
!
pScanInfo
->
pUpdateInfo
)
{
if
(
!
pScanInfo
->
igCheckUpdate
&&
!
pScanInfo
->
pUpdateInfo
)
{
pScanInfo
->
pUpdateInfo
=
updateInfoInitP
(
pInterval
,
pTwSup
->
waterMark
);
}
pScanInfo
->
interval
=
*
pInterval
;
...
...
@@ -2893,7 +2893,7 @@ void initDownStream(SOperatorInfo* downstream, SStreamAggSupporter* pAggSup, uin
}
SStreamScanInfo
*
pScanInfo
=
downstream
->
info
;
pScanInfo
->
windowSup
=
(
SWindowSupporter
){.
pStreamAggSup
=
pAggSup
,
.
gap
=
pAggSup
->
gap
,
.
parentType
=
type
};
if
(
!
pScanInfo
->
pUpdateInfo
)
{
if
(
!
pScanInfo
->
igCheckUpdate
&&
!
pScanInfo
->
pUpdateInfo
)
{
pScanInfo
->
pUpdateInfo
=
updateInfoInit
(
60000
,
TSDB_TIME_PRECISION_MILLI
,
pTwSup
->
waterMark
);
}
pScanInfo
->
twAggSup
=
*
pTwSup
;
...
...
tests/script/tsim/stream/ignoreCheckUpdate.sim
0 → 100644
浏览文件 @
e89362fb
system sh/stop_dnodes.sh
system sh/deploy.sh -n dnode1 -i 1
system sh/exec.sh -n dnode1 -s start
sleep 50
sql connect
sql drop stream if exists streams0;
sql drop database if exists test;
sql create database test vgroups 1;
sql use test;
sql create table t1(ts timestamp, a int, b int , c int);
sql create stream streams0 trigger at_once ignore update 1 into streamt as select _wstart c1, count(*) c2, max(b) c3 from t1 interval(10s);
sql insert into t1 values(1648791213000,1,1,1);
sql insert into t1 values(1648791213000,2,2,2);
$loop_count = 0
loop0:
sleep 300
sql select * from streamt order by 1,2,3;
$loop_count = $loop_count + 1
if $loop_count == 10 then
return -1
endi
if $data01 != 2 then
print =====data01=$data01
goto loop0
endi
if $data02 != 1 then
print =====data02=$data02
goto loop0
endi
sql insert into t1 values(1648791213000,3,3,3);
$loop_count = 0
loop1:
sleep 300
sql select * from streamt order by 1,2,3;
$loop_count = $loop_count + 1
if $loop_count == 10 then
return -1
endi
if $data01 != 3 then
print =====data01=$data01
goto loop1
endi
if $data02 != 1 then
print =====data02=$data02
goto loop1
endi
system sh/stop_dnodes.sh
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录