Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
ca2183ac
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看板
提交
ca2183ac
编写于
7月 31, 2023
作者:
H
Haojun Liao
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
refactor: do some internal refactor.
上级
143f39b6
变更
7
显示空白变更内容
内联
并排
Showing
7 changed file
with
46 addition
and
148 deletion
+46
-148
include/libs/executor/executor.h
include/libs/executor/executor.h
+1
-5
include/libs/stream/tstream.h
include/libs/stream/tstream.h
+0
-2
source/dnode/vnode/src/tq/tq.c
source/dnode/vnode/src/tq/tq.c
+1
-1
source/libs/executor/inc/querytask.h
source/libs/executor/inc/querytask.h
+2
-2
source/libs/executor/src/executor.c
source/libs/executor/src/executor.c
+7
-29
source/libs/stream/src/streamExec.c
source/libs/stream/src/streamExec.c
+34
-98
source/libs/stream/src/streamRecover.c
source/libs/stream/src/streamRecover.c
+1
-11
未找到文件。
include/libs/executor/executor.h
浏览文件 @
ca2183ac
...
...
@@ -221,13 +221,9 @@ int32_t qStreamSourceScanParamForHistoryScanStep2(qTaskInfo_t tinfo, SVersionRan
int32_t
qStreamRecoverFinish
(
qTaskInfo_t
tinfo
);
int32_t
qRestoreStreamOperatorOption
(
qTaskInfo_t
tinfo
);
bool
qStreamRecoverScanFinished
(
qTaskInfo_t
tinfo
);
bool
qStreamRecoverScanStep1Finished
(
qTaskInfo_t
tinfo
);
bool
qStreamRecoverScanStep2Finished
(
qTaskInfo_t
tinfo
);
int32_t
qStreamRecoverSetAllStepFinished
(
qTaskInfo_t
tinfo
);
int32_t
qStreamInfoResetTimewindowFilter
(
qTaskInfo_t
tinfo
);
void
resetTaskInfo
(
qTaskInfo_t
tinfo
);
void
qResetStreamInfoTimeWindow
(
qTaskInfo_t
tinfo
);
int32_t
qStreamOperatorReleaseState
(
qTaskInfo_t
tInfo
);
int32_t
qStreamOperatorReloadState
(
qTaskInfo_t
tInfo
);
...
...
include/libs/stream/tstream.h
浏览文件 @
ca2183ac
...
...
@@ -607,8 +607,6 @@ int32_t streamTaskScanHistoryDataComplete(SStreamTask* pTask);
int32_t
streamStartRecoverTask
(
SStreamTask
*
pTask
,
int8_t
igUntreated
);
bool
streamHistoryTaskSetVerRangeStep2
(
SStreamTask
*
pTask
,
int64_t
latestVer
);
bool
streamTaskRecoverScanStep1Finished
(
SStreamTask
*
pTask
);
bool
streamTaskRecoverScanStep2Finished
(
SStreamTask
*
pTask
);
int32_t
streamTaskRecoverSetAllStepFinished
(
SStreamTask
*
pTask
);
// common
...
...
source/dnode/vnode/src/tq/tq.c
浏览文件 @
ca2183ac
...
...
@@ -1296,7 +1296,7 @@ int32_t tqProcessTaskScanHistory(STQ* pTq, SRpcMsg* pMsg) {
"s-task:%s scan-history in stream time window completed, no related fill-history task, reset the time "
"window:%"
PRId64
" - %"
PRId64
,
id
,
pWindow
->
skey
,
pWindow
->
ekey
);
q
ResetStreamInfoTimeWindow
(
pTask
->
exec
.
pExecutor
);
q
StreamInfoResetTimewindowFilter
(
pTask
->
exec
.
pExecutor
);
}
else
{
// when related fill-history task exists, update the fill-history time window only when the
// state transfer is completed.
...
...
source/libs/executor/inc/querytask.h
浏览文件 @
ca2183ac
...
...
@@ -62,8 +62,8 @@ typedef struct {
SSchemaWrapper
*
schema
;
char
tbName
[
TSDB_TABLE_NAME_LEN
];
// this is the current scan table: todo refactor
int8_t
recoverStep
;
bool
recoverStep1Finished
;
bool
recoverStep2Finished
;
//
bool recoverStep1Finished;
//
bool recoverStep2Finished;
int8_t
recoverScanFinished
;
SQueryTableDataCond
tableCond
;
SVersionRange
fillHistoryVer
;
...
...
source/libs/executor/src/executor.c
浏览文件 @
ca2183ac
...
...
@@ -116,17 +116,6 @@ void resetTaskInfo(qTaskInfo_t tinfo) {
clearStreamBlock
(
pTaskInfo
->
pRoot
);
}
void
qResetStreamInfoTimeWindow
(
qTaskInfo_t
tinfo
)
{
SExecTaskInfo
*
pTaskInfo
=
(
SExecTaskInfo
*
)
tinfo
;
if
(
pTaskInfo
==
NULL
)
{
return
;
}
qDebug
(
"%s set stream fill-history window:%"
PRId64
"-%"
PRId64
,
GET_TASKID
(
pTaskInfo
),
INT64_MIN
,
INT64_MAX
);
pTaskInfo
->
streamInfo
.
fillHistoryWindow
.
skey
=
INT64_MIN
;
pTaskInfo
->
streamInfo
.
fillHistoryWindow
.
ekey
=
INT64_MAX
;
}
static
int32_t
doSetStreamBlock
(
SOperatorInfo
*
pOperator
,
void
*
input
,
size_t
numOfBlocks
,
int32_t
type
,
const
char
*
id
)
{
if
(
pOperator
->
operatorType
!=
QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN
)
{
if
(
pOperator
->
numOfDownstream
==
0
)
{
...
...
@@ -341,6 +330,7 @@ qTaskInfo_t qCreateStreamExecTaskInfo(void* msg, SReadHandle* readers, int32_t v
return
NULL
;
}
qStreamInfoResetTimewindowFilter
(
pTaskInfo
);
return
pTaskInfo
;
}
...
...
@@ -890,8 +880,6 @@ int32_t qStreamSourceScanParamForHistoryScanStep1(qTaskInfo_t tinfo, SVersionRan
pStreamInfo
->
fillHistoryVer
=
*
pVerRange
;
pStreamInfo
->
fillHistoryWindow
=
*
pWindow
;
pStreamInfo
->
recoverStep
=
STREAM_RECOVER_STEP__PREPARE1
;
pStreamInfo
->
recoverStep1Finished
=
false
;
pStreamInfo
->
recoverStep2Finished
=
false
;
qDebug
(
"%s step 1. set param for stream scanner for scan-history data, verRange:%"
PRId64
" - %"
PRId64
", window:%"
PRId64
" - %"
PRId64
,
...
...
@@ -909,8 +897,6 @@ int32_t qStreamSourceScanParamForHistoryScanStep2(qTaskInfo_t tinfo, SVersionRan
pStreamInfo
->
fillHistoryVer
=
*
pVerRange
;
pStreamInfo
->
fillHistoryWindow
=
*
pWindow
;
pStreamInfo
->
recoverStep
=
STREAM_RECOVER_STEP__PREPARE2
;
pStreamInfo
->
recoverStep1Finished
=
true
;
pStreamInfo
->
recoverStep2Finished
=
false
;
qDebug
(
"%s step 2. set param for stream scanner for scan-history data, verRange:%"
PRId64
" - %"
PRId64
", window:%"
PRId64
" - %"
PRId64
,
...
...
@@ -1049,23 +1035,15 @@ bool qStreamRecoverScanFinished(qTaskInfo_t tinfo) {
return
pTaskInfo
->
streamInfo
.
recoverScanFinished
;
}
bool
qStreamRecoverScanStep1Finished
(
qTaskInfo_t
tinfo
)
{
int32_t
qStreamInfoResetTimewindowFilter
(
qTaskInfo_t
tinfo
)
{
SExecTaskInfo
*
pTaskInfo
=
(
SExecTaskInfo
*
)
tinfo
;
return
pTaskInfo
->
streamInfo
.
recoverStep1Finished
;
}
STimeWindow
*
pWindow
=
&
pTaskInfo
->
streamInfo
.
fillHistoryWindow
;
bool
qStreamRecoverScanStep2Finished
(
qTaskInfo_t
tinfo
)
{
SExecTaskInfo
*
pTaskInfo
=
(
SExecTaskInfo
*
)
tinfo
;
return
pTaskInfo
->
streamInfo
.
recoverStep2Finished
;
}
int32_t
qStreamRecoverSetAllStepFinished
(
qTaskInfo_t
tinfo
)
{
SExecTaskInfo
*
pTaskInfo
=
(
SExecTaskInfo
*
)
tinfo
;
pTaskInfo
->
streamInfo
.
recoverStep1Finished
=
true
;
pTaskInfo
->
streamInfo
.
recoverStep2Finished
=
true
;
qDebug
(
"%s set remove scan-history filter window:%"
PRId64
"-%"
PRId64
", new window:%"
PRId64
"-%"
PRId64
,
GET_TASKID
(
pTaskInfo
),
pWindow
->
skey
,
pWindow
->
ekey
,
INT64_MIN
,
INT64_MAX
);
// reset the time window
p
TaskInfo
->
streamInfo
.
fillHistoryWindow
.
skey
=
INT64_MIN
;
pWindow
->
skey
=
INT64_MIN
;
p
Window
->
ekey
=
INT64_MAX
;
return
0
;
}
...
...
source/libs/stream/src/streamExec.c
浏览文件 @
ca2183ac
...
...
@@ -164,15 +164,14 @@ static int32_t streamTaskExecImpl(SStreamTask* pTask, SStreamQueueItem* pItem, i
}
int32_t
streamScanExec
(
SStreamTask
*
pTask
,
int32_t
batchSz
)
{
int32_t
code
=
0
;
ASSERT
(
pTask
->
info
.
taskLevel
==
TASK_LEVEL__SOURCE
);
int32_t
code
=
TSDB_CODE_SUCCESS
;
void
*
exec
=
pTask
->
exec
.
pExecutor
;
bool
finished
=
false
;
qSetStreamOpOpen
(
exec
);
bool
finished
=
false
;
while
(
1
)
{
while
(
!
finished
)
{
if
(
streamTaskShouldPause
(
&
pTask
->
status
))
{
double
el
=
(
taosGetTimestampMs
()
-
pTask
->
tsInfo
.
step1Start
)
/
1000
.
0
;
qDebug
(
"s-task:%s paused from the scan-history task, elapsed time:%.2fsec"
,
pTask
->
id
.
idStr
,
el
);
...
...
@@ -185,44 +184,30 @@ int32_t streamScanExec(SStreamTask* pTask, int32_t batchSz) {
return
-
1
;
}
int32_t
batchCnt
=
0
;
int32_t
numOfBlocks
=
0
;
while
(
1
)
{
if
(
streamTaskShouldStop
(
&
pTask
->
status
))
{
taosArrayDestroyEx
(
pRes
,
(
FDelete
)
blockDataFreeRes
);
return
0
;
}
if
(
streamTaskShouldPause
(
&
pTask
->
status
))
{
break
;
}
SSDataBlock
*
output
=
NULL
;
uint64_t
ts
=
0
;
if
(
qExecTask
(
exec
,
&
output
,
&
ts
)
<
0
)
{
continue
;
}
if
(
output
==
NULL
)
{
if
(
qStreamRecoverScanFinished
(
exec
))
{
if
(
output
==
NULL
&&
qStreamRecoverScanFinished
(
exec
))
{
finished
=
true
;
break
;
}
else
{
qSetStreamOpOpen
(
exec
);
if
(
streamTaskShouldPause
(
&
pTask
->
status
))
{
SStreamDataBlock
*
qRes
=
taosAllocateQitem
(
sizeof
(
SStreamDataBlock
),
DEF_QITEM
,
0
);
if
(
qRes
==
NULL
)
{
taosArrayDestroyEx
(
pRes
,
(
FDelete
)
blockDataFreeRes
);
terrno
=
TSDB_CODE_OUT_OF_MEMORY
;
return
-
1
;
}
qRes
->
type
=
STREAM_INPUT__DATA_BLOCK
;
qRes
->
blocks
=
pRes
;
code
=
streamTaskOutputResultBlock
(
pTask
,
qRes
);
if
(
code
==
TSDB_CODE_UTIL_QUEUE_OUT_OF_MEMORY
)
{
taosArrayDestroyEx
(
pRes
,
(
FDelete
)
blockDataFreeRes
);
taosFreeQitem
(
qRes
);
return
code
;
}
return
0
;
}
if
(
output
==
NULL
)
{
ASSERT
(
0
);
}
break
;
}
SSDataBlock
block
=
{
0
};
...
...
@@ -230,26 +215,14 @@ int32_t streamScanExec(SStreamTask* pTask, int32_t batchSz) {
block
.
info
.
childId
=
pTask
->
info
.
selfChildId
;
taosArrayPush
(
pRes
,
&
block
);
batchCnt
++
;
qDebug
(
"s-task:%s scan exec numOfBlocks:%d, limit:%d"
,
pTask
->
id
.
idStr
,
batchCnt
,
batchSz
);
if
(
batchCnt
>=
batchSz
)
{
break
;
}
}
if
(
taosArrayGetSize
(
pRes
)
==
0
)
{
taosArrayDestroy
(
pRes
);
if
(
finished
)
{
qDebug
(
"s-task:%s finish recover exec task "
,
pTask
->
id
.
idStr
);
numOfBlocks
++
;
qDebug
(
"s-task:%s scan exec numOfBlocks:%d, limit:%d"
,
pTask
->
id
.
idStr
,
numOfBlocks
,
batchSz
);
if
(
numOfBlocks
>=
batchSz
)
{
break
;
}
else
{
qDebug
(
"s-task:%s continue recover exec task "
,
pTask
->
id
.
idStr
);
continue
;
}
}
if
(
taosArrayGetSize
(
pRes
)
>
0
)
{
SStreamDataBlock
*
qRes
=
taosAllocateQitem
(
sizeof
(
SStreamDataBlock
),
DEF_QITEM
,
0
);
if
(
qRes
==
NULL
)
{
taosArrayDestroyEx
(
pRes
,
(
FDelete
)
blockDataFreeRes
);
...
...
@@ -259,57 +232,20 @@ int32_t streamScanExec(SStreamTask* pTask, int32_t batchSz) {
qRes
->
type
=
STREAM_INPUT__DATA_BLOCK
;
qRes
->
blocks
=
pRes
;
code
=
streamTaskOutputResultBlock
(
pTask
,
qRes
);
if
(
code
==
TSDB_CODE_UTIL_QUEUE_OUT_OF_MEMORY
)
{
taosArrayDestroyEx
(
pRes
,
(
FDelete
)
blockDataFreeRes
);
taosFreeQitem
(
qRes
);
return
code
;
}
if
(
finished
)
{
break
;
}
}
return
0
;
}
#if 0
int32_t streamBatchExec(SStreamTask* pTask, int32_t batchLimit) {
// fetch all queue item, merge according to batchLimit
int32_t numOfItems = taosReadAllQitems(pTask->inputQueue1, pTask->inputQall);
if (numOfItems == 0) {
qDebug("task: %d, stream task exec over, queue empty", pTask->id.taskId);
return 0;
}
SStreamQueueItem* pMerged = NULL;
SStreamQueueItem* pItem = NULL;
taosGetQitem(pTask->inputQall, (void**)&pItem);
if (pItem == NULL) {
if (pMerged != NULL) {
// process merged item
}
else
{
return 0;
}
taosArrayDestroy
(
pRes
);
}
// if drop
if (pItem->type == STREAM_INPUT__DESTROY) {
// set status drop
return -1;
}
if (pTask->info.taskLevel == TASK_LEVEL__SINK) {
ASSERT(((SStreamQueueItem*)pItem)->type == STREAM_INPUT__DATA_BLOCK);
streamTaskOutputResultBlock(pTask, (SStreamDataBlock*)pItem);
}
// exec impl
// output
// try dispatch
return
0
;
}
#endif
int32_t
updateCheckPointInfo
(
SStreamTask
*
pTask
)
{
int64_t
ckId
=
0
;
...
...
@@ -425,7 +361,7 @@ static int32_t streamDoTransferStateToStreamTask(SStreamTask* pTask) {
// expand the query time window for stream scanner
pTimeWindow
->
skey
=
INT64_MIN
;
q
ResetStreamInfoTimeWindow
(
pStreamTask
->
exec
.
pExecutor
);
q
StreamInfoResetTimewindowFilter
(
pStreamTask
->
exec
.
pExecutor
);
// transfer the ownership of executor state
streamTaskReleaseState
(
pTask
);
...
...
source/libs/stream/src/streamRecover.c
浏览文件 @
ca2183ac
...
...
@@ -655,19 +655,9 @@ int32_t streamTaskScanHistoryDataComplete(SStreamTask* pTask) {
return
0
;
}
bool
streamTaskRecoverScanStep1Finished
(
SStreamTask
*
pTask
)
{
void
*
exec
=
pTask
->
exec
.
pExecutor
;
return
qStreamRecoverScanStep1Finished
(
exec
);
}
bool
streamTaskRecoverScanStep2Finished
(
SStreamTask
*
pTask
)
{
void
*
exec
=
pTask
->
exec
.
pExecutor
;
return
qStreamRecoverScanStep2Finished
(
exec
);
}
int32_t
streamTaskRecoverSetAllStepFinished
(
SStreamTask
*
pTask
)
{
void
*
exec
=
pTask
->
exec
.
pExecutor
;
return
qStream
RecoverSetAllStepFinished
(
exec
);
return
qStream
InfoResetTimewindowFilter
(
exec
);
}
bool
streamHistoryTaskSetVerRangeStep2
(
SStreamTask
*
pTask
,
int64_t
latestVer
)
{
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录