Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
f18cfc25
T
TDengine
项目概览
taosdata
/
TDengine
1 年多 前同步成功
通知
1185
Star
22016
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看板
提交
f18cfc25
编写于
3月 24, 2023
作者:
5
54liuyao
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
fix:reset parameter ignoreExpried for fill history
上级
31152651
变更
3
显示空白变更内容
内联
并排
Showing
3 changed file
with
16 addition
and
16 deletion
+16
-16
source/libs/executor/inc/executorimpl.h
source/libs/executor/inc/executorimpl.h
+3
-0
source/libs/executor/src/executor.c
source/libs/executor/src/executor.c
+9
-16
source/libs/executor/src/timewindowoperator.c
source/libs/executor/src/timewindowoperator.c
+4
-0
未找到文件。
source/libs/executor/inc/executorimpl.h
浏览文件 @
f18cfc25
...
@@ -559,6 +559,7 @@ typedef struct SStreamIntervalOperatorInfo {
...
@@ -559,6 +559,7 @@ typedef struct SStreamIntervalOperatorInfo {
STimeWindowAggSupp
twAggSup
;
STimeWindowAggSupp
twAggSup
;
bool
invertible
;
bool
invertible
;
bool
ignoreExpiredData
;
bool
ignoreExpiredData
;
bool
ignoreExpiredDataSaved
;
SArray
*
pDelWins
;
// SWinRes
SArray
*
pDelWins
;
// SWinRes
int32_t
delIndex
;
int32_t
delIndex
;
SSDataBlock
*
pDelRes
;
SSDataBlock
*
pDelRes
;
...
@@ -620,6 +621,7 @@ typedef struct SStreamSessionAggOperatorInfo {
...
@@ -620,6 +621,7 @@ typedef struct SStreamSessionAggOperatorInfo {
SPhysiNode
*
pPhyNode
;
// create new child
SPhysiNode
*
pPhyNode
;
// create new child
bool
isFinal
;
bool
isFinal
;
bool
ignoreExpiredData
;
bool
ignoreExpiredData
;
bool
ignoreExpiredDataSaved
;
SArray
*
pUpdated
;
SArray
*
pUpdated
;
SSHashObj
*
pStUpdated
;
SSHashObj
*
pStUpdated
;
}
SStreamSessionAggOperatorInfo
;
}
SStreamSessionAggOperatorInfo
;
...
@@ -637,6 +639,7 @@ typedef struct SStreamStateAggOperatorInfo {
...
@@ -637,6 +639,7 @@ typedef struct SStreamStateAggOperatorInfo {
void
*
pDelIterator
;
void
*
pDelIterator
;
SArray
*
pChildren
;
// cache for children's result;
SArray
*
pChildren
;
// cache for children's result;
bool
ignoreExpiredData
;
bool
ignoreExpiredData
;
bool
ignoreExpiredDataSaved
;
SArray
*
pUpdated
;
SArray
*
pUpdated
;
SSHashObj
*
pSeUpdated
;
SSHashObj
*
pSeUpdated
;
}
SStreamStateAggOperatorInfo
;
}
SStreamStateAggOperatorInfo
;
...
...
source/libs/executor/src/executor.c
浏览文件 @
f18cfc25
...
@@ -888,7 +888,8 @@ int32_t qStreamSetParamForRecover(qTaskInfo_t tinfo) {
...
@@ -888,7 +888,8 @@ int32_t qStreamSetParamForRecover(qTaskInfo_t tinfo) {
pInfo
->
twAggSup
.
deleteMarkSaved
=
pInfo
->
twAggSup
.
deleteMark
;
pInfo
->
twAggSup
.
deleteMarkSaved
=
pInfo
->
twAggSup
.
deleteMark
;
pInfo
->
twAggSup
.
calTrigger
=
STREAM_TRIGGER_AT_ONCE
;
pInfo
->
twAggSup
.
calTrigger
=
STREAM_TRIGGER_AT_ONCE
;
pInfo
->
twAggSup
.
deleteMark
=
INT64_MAX
;
pInfo
->
twAggSup
.
deleteMark
=
INT64_MAX
;
pInfo
->
ignoreExpiredDataSaved
=
pInfo
->
ignoreExpiredData
;
pInfo
->
ignoreExpiredData
=
false
;
}
else
if
(
pOperator
->
operatorType
==
QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION
||
}
else
if
(
pOperator
->
operatorType
==
QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION
||
pOperator
->
operatorType
==
QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_SESSION
||
pOperator
->
operatorType
==
QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_SESSION
||
pOperator
->
operatorType
==
QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_SESSION
)
{
pOperator
->
operatorType
==
QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_SESSION
)
{
...
@@ -904,6 +905,8 @@ int32_t qStreamSetParamForRecover(qTaskInfo_t tinfo) {
...
@@ -904,6 +905,8 @@ int32_t qStreamSetParamForRecover(qTaskInfo_t tinfo) {
pInfo
->
twAggSup
.
deleteMarkSaved
=
pInfo
->
twAggSup
.
deleteMark
;
pInfo
->
twAggSup
.
deleteMarkSaved
=
pInfo
->
twAggSup
.
deleteMark
;
pInfo
->
twAggSup
.
calTrigger
=
STREAM_TRIGGER_AT_ONCE
;
pInfo
->
twAggSup
.
calTrigger
=
STREAM_TRIGGER_AT_ONCE
;
pInfo
->
twAggSup
.
deleteMark
=
INT64_MAX
;
pInfo
->
twAggSup
.
deleteMark
=
INT64_MAX
;
pInfo
->
ignoreExpiredDataSaved
=
pInfo
->
ignoreExpiredData
;
pInfo
->
ignoreExpiredData
=
false
;
}
else
if
(
pOperator
->
operatorType
==
QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE
)
{
}
else
if
(
pOperator
->
operatorType
==
QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE
)
{
SStreamStateAggOperatorInfo
*
pInfo
=
pOperator
->
info
;
SStreamStateAggOperatorInfo
*
pInfo
=
pOperator
->
info
;
ASSERT
(
pInfo
->
twAggSup
.
calTrigger
==
STREAM_TRIGGER_AT_ONCE
||
ASSERT
(
pInfo
->
twAggSup
.
calTrigger
==
STREAM_TRIGGER_AT_ONCE
||
...
@@ -917,6 +920,8 @@ int32_t qStreamSetParamForRecover(qTaskInfo_t tinfo) {
...
@@ -917,6 +920,8 @@ int32_t qStreamSetParamForRecover(qTaskInfo_t tinfo) {
pInfo
->
twAggSup
.
deleteMarkSaved
=
pInfo
->
twAggSup
.
deleteMark
;
pInfo
->
twAggSup
.
deleteMarkSaved
=
pInfo
->
twAggSup
.
deleteMark
;
pInfo
->
twAggSup
.
calTrigger
=
STREAM_TRIGGER_AT_ONCE
;
pInfo
->
twAggSup
.
calTrigger
=
STREAM_TRIGGER_AT_ONCE
;
pInfo
->
twAggSup
.
deleteMark
=
INT64_MAX
;
pInfo
->
twAggSup
.
deleteMark
=
INT64_MAX
;
pInfo
->
ignoreExpiredDataSaved
=
pInfo
->
ignoreExpiredData
;
pInfo
->
ignoreExpiredData
=
false
;
}
}
// iterate operator tree
// iterate operator tree
...
@@ -944,35 +949,23 @@ int32_t qStreamRestoreParam(qTaskInfo_t tinfo) {
...
@@ -944,35 +949,23 @@ int32_t qStreamRestoreParam(qTaskInfo_t tinfo) {
pOperator
->
operatorType
==
QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_INTERVAL
||
pOperator
->
operatorType
==
QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_INTERVAL
||
pOperator
->
operatorType
==
QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_INTERVAL
)
{
pOperator
->
operatorType
==
QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_INTERVAL
)
{
SStreamIntervalOperatorInfo
*
pInfo
=
pOperator
->
info
;
SStreamIntervalOperatorInfo
*
pInfo
=
pOperator
->
info
;
/*ASSERT(pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE);*/
/*ASSERT(pInfo->twAggSup.deleteMark == INT64_MAX);*/
pInfo
->
twAggSup
.
calTrigger
=
pInfo
->
twAggSup
.
calTriggerSaved
;
pInfo
->
twAggSup
.
calTrigger
=
pInfo
->
twAggSup
.
calTriggerSaved
;
pInfo
->
twAggSup
.
deleteMark
=
pInfo
->
twAggSup
.
deleteMarkSaved
;
pInfo
->
twAggSup
.
deleteMark
=
pInfo
->
twAggSup
.
deleteMarkSaved
;
/*ASSERT(pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE ||*/
pInfo
->
ignoreExpiredData
=
pInfo
->
ignoreExpiredDataSaved
;
/*pInfo->twAggSup.calTrigger == STREAM_TRIGGER_WINDOW_CLOSE);*/
qInfo
(
"restore stream param for interval: %d, %"
PRId64
,
pInfo
->
twAggSup
.
calTrigger
,
pInfo
->
twAggSup
.
deleteMark
);
qInfo
(
"restore stream param for interval: %d, %"
PRId64
,
pInfo
->
twAggSup
.
calTrigger
,
pInfo
->
twAggSup
.
deleteMark
);
}
else
if
(
pOperator
->
operatorType
==
QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION
||
}
else
if
(
pOperator
->
operatorType
==
QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION
||
pOperator
->
operatorType
==
QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_SESSION
||
pOperator
->
operatorType
==
QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_SESSION
||
pOperator
->
operatorType
==
QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_SESSION
)
{
pOperator
->
operatorType
==
QUERY_NODE_PHYSICAL_PLAN_STREAM_FINAL_SESSION
)
{
SStreamSessionAggOperatorInfo
*
pInfo
=
pOperator
->
info
;
SStreamSessionAggOperatorInfo
*
pInfo
=
pOperator
->
info
;
/*ASSERT(pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE);*/
/*ASSERT(pInfo->twAggSup.deleteMark == INT64_MAX);*/
pInfo
->
twAggSup
.
calTrigger
=
pInfo
->
twAggSup
.
calTriggerSaved
;
pInfo
->
twAggSup
.
calTrigger
=
pInfo
->
twAggSup
.
calTriggerSaved
;
pInfo
->
twAggSup
.
deleteMark
=
pInfo
->
twAggSup
.
deleteMarkSaved
;
pInfo
->
twAggSup
.
deleteMark
=
pInfo
->
twAggSup
.
deleteMarkSaved
;
/*ASSERT(pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE ||*/
pInfo
->
ignoreExpiredData
=
pInfo
->
ignoreExpiredDataSaved
;
/*pInfo->twAggSup.calTrigger == STREAM_TRIGGER_WINDOW_CLOSE);*/
qInfo
(
"restore stream param for session: %d, %"
PRId64
,
pInfo
->
twAggSup
.
calTrigger
,
pInfo
->
twAggSup
.
deleteMark
);
qInfo
(
"restore stream param for session: %d, %"
PRId64
,
pInfo
->
twAggSup
.
calTrigger
,
pInfo
->
twAggSup
.
deleteMark
);
}
else
if
(
pOperator
->
operatorType
==
QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE
)
{
}
else
if
(
pOperator
->
operatorType
==
QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE
)
{
SStreamStateAggOperatorInfo
*
pInfo
=
pOperator
->
info
;
SStreamStateAggOperatorInfo
*
pInfo
=
pOperator
->
info
;
/*ASSERT(pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE);*/
/*ASSERT(pInfo->twAggSup.deleteMark == INT64_MAX);*/
pInfo
->
twAggSup
.
calTrigger
=
pInfo
->
twAggSup
.
calTriggerSaved
;
pInfo
->
twAggSup
.
calTrigger
=
pInfo
->
twAggSup
.
calTriggerSaved
;
pInfo
->
twAggSup
.
deleteMark
=
pInfo
->
twAggSup
.
deleteMarkSaved
;
pInfo
->
twAggSup
.
deleteMark
=
pInfo
->
twAggSup
.
deleteMarkSaved
;
/*ASSERT(pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE ||*/
pInfo
->
ignoreExpiredData
=
pInfo
->
ignoreExpiredDataSaved
;
/*pInfo->twAggSup.calTrigger == STREAM_TRIGGER_WINDOW_CLOSE);*/
qInfo
(
"restore stream param for state: %d, %"
PRId64
,
pInfo
->
twAggSup
.
calTrigger
,
pInfo
->
twAggSup
.
deleteMark
);
qInfo
(
"restore stream param for state: %d, %"
PRId64
,
pInfo
->
twAggSup
.
calTrigger
,
pInfo
->
twAggSup
.
deleteMark
);
}
}
...
...
source/libs/executor/src/timewindowoperator.c
浏览文件 @
f18cfc25
...
@@ -2769,6 +2769,7 @@ SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream,
...
@@ -2769,6 +2769,7 @@ SOperatorInfo* createStreamFinalIntervalOperatorInfo(SOperatorInfo* downstream,
pInfo
->
pPullDataMap
=
taosHashInit
(
64
,
hashFn
,
false
,
HASH_NO_LOCK
);
pInfo
->
pPullDataMap
=
taosHashInit
(
64
,
hashFn
,
false
,
HASH_NO_LOCK
);
pInfo
->
pPullDataRes
=
createSpecialDataBlock
(
STREAM_RETRIEVE
);
pInfo
->
pPullDataRes
=
createSpecialDataBlock
(
STREAM_RETRIEVE
);
pInfo
->
ignoreExpiredData
=
pIntervalPhyNode
->
window
.
igExpired
;
pInfo
->
ignoreExpiredData
=
pIntervalPhyNode
->
window
.
igExpired
;
pInfo
->
ignoreExpiredDataSaved
=
false
;
pInfo
->
pDelRes
=
createSpecialDataBlock
(
STREAM_DELETE_RESULT
);
pInfo
->
pDelRes
=
createSpecialDataBlock
(
STREAM_DELETE_RESULT
);
pInfo
->
delIndex
=
0
;
pInfo
->
delIndex
=
0
;
pInfo
->
pDelWins
=
taosArrayInit
(
4
,
sizeof
(
SWinKey
));
pInfo
->
pDelWins
=
taosArrayInit
(
4
,
sizeof
(
SWinKey
));
...
@@ -3587,6 +3588,7 @@ SOperatorInfo* createStreamSessionAggOperatorInfo(SOperatorInfo* downstream, SPh
...
@@ -3587,6 +3588,7 @@ SOperatorInfo* createStreamSessionAggOperatorInfo(SOperatorInfo* downstream, SPh
pInfo
->
isFinal
=
false
;
pInfo
->
isFinal
=
false
;
pInfo
->
pPhyNode
=
pPhyNode
;
pInfo
->
pPhyNode
=
pPhyNode
;
pInfo
->
ignoreExpiredData
=
pSessionNode
->
window
.
igExpired
;
pInfo
->
ignoreExpiredData
=
pSessionNode
->
window
.
igExpired
;
pInfo
->
ignoreExpiredDataSaved
=
false
;
pInfo
->
pUpdated
=
NULL
;
pInfo
->
pUpdated
=
NULL
;
pInfo
->
pStUpdated
=
NULL
;
pInfo
->
pStUpdated
=
NULL
;
...
@@ -4112,6 +4114,7 @@ SOperatorInfo* createStreamStateAggOperatorInfo(SOperatorInfo* downstream, SPhys
...
@@ -4112,6 +4114,7 @@ SOperatorInfo* createStreamStateAggOperatorInfo(SOperatorInfo* downstream, SPhys
pInfo
->
pDelRes
=
createSpecialDataBlock
(
STREAM_DELETE_RESULT
);
pInfo
->
pDelRes
=
createSpecialDataBlock
(
STREAM_DELETE_RESULT
);
pInfo
->
pChildren
=
NULL
;
pInfo
->
pChildren
=
NULL
;
pInfo
->
ignoreExpiredData
=
pStateNode
->
window
.
igExpired
;
pInfo
->
ignoreExpiredData
=
pStateNode
->
window
.
igExpired
;
pInfo
->
ignoreExpiredDataSaved
=
false
;
pInfo
->
pUpdated
=
NULL
;
pInfo
->
pUpdated
=
NULL
;
pInfo
->
pSeUpdated
=
NULL
;
pInfo
->
pSeUpdated
=
NULL
;
...
@@ -4885,6 +4888,7 @@ SOperatorInfo* createStreamIntervalOperatorInfo(SOperatorInfo* downstream, SPhys
...
@@ -4885,6 +4888,7 @@ SOperatorInfo* createStreamIntervalOperatorInfo(SOperatorInfo* downstream, SPhys
pInfo
->
interval
=
interval
;
pInfo
->
interval
=
interval
;
pInfo
->
twAggSup
=
twAggSupp
;
pInfo
->
twAggSup
=
twAggSupp
;
pInfo
->
ignoreExpiredData
=
pIntervalPhyNode
->
window
.
igExpired
;
pInfo
->
ignoreExpiredData
=
pIntervalPhyNode
->
window
.
igExpired
;
pInfo
->
ignoreExpiredDataSaved
=
false
;
pInfo
->
isFinal
=
false
;
pInfo
->
isFinal
=
false
;
SExprSupp
*
pSup
=
&
pOperator
->
exprSupp
;
SExprSupp
*
pSup
=
&
pOperator
->
exprSupp
;
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录