Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
19fe9618
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,发现更多精彩内容 >>
未验证
提交
19fe9618
编写于
7月 23, 2022
作者:
L
liuyao
提交者:
GitHub
7月 23, 2022
浏览文件
操作
浏览文件
下载
差异文件
Merge pull request #15341 from taosdata/feature/TD-17357
fix(stream): push retrive datablock
上级
05efe6f9
f86ce16e
变更
1
隐藏空白更改
内联
并排
Showing
1 changed file
with
60 addition
and
28 deletion
+60
-28
source/libs/executor/src/timewindowoperator.c
source/libs/executor/src/timewindowoperator.c
+60
-28
未找到文件。
source/libs/executor/src/timewindowoperator.c
浏览文件 @
19fe9618
...
...
@@ -772,6 +772,41 @@ int32_t binarySearch(void* keyList, int num, TSKEY key, int order, __get_value_f
return
midPos
;
}
int32_t
comparePullWinKey
(
void
*
pKey
,
void
*
data
,
int32_t
index
)
{
SArray
*
res
=
(
SArray
*
)
data
;
SPullWindowInfo
*
pos
=
taosArrayGet
(
res
,
index
);
SPullWindowInfo
*
pData
=
(
SPullWindowInfo
*
)
pKey
;
if
(
pData
->
window
.
skey
==
pos
->
window
.
skey
)
{
if
(
pData
->
groupId
>
pos
->
groupId
)
{
return
1
;
}
else
if
(
pData
->
groupId
<
pos
->
groupId
)
{
return
-
1
;
}
return
0
;
}
else
if
(
pData
->
window
.
skey
>
pos
->
window
.
skey
)
{
return
1
;
}
return
-
1
;
}
static
int32_t
savePullWindow
(
SPullWindowInfo
*
pPullInfo
,
SArray
*
pPullWins
)
{
int32_t
size
=
taosArrayGetSize
(
pPullWins
);
int32_t
index
=
binarySearchCom
(
pPullWins
,
size
,
pPullInfo
,
TSDB_ORDER_DESC
,
comparePullWinKey
);
if
(
index
==
-
1
)
{
index
=
0
;
}
else
{
if
(
comparePullWinKey
(
pPullInfo
,
pPullWins
,
index
)
>
0
)
{
index
++
;
}
else
{
return
TSDB_CODE_SUCCESS
;
}
}
if
(
taosArrayInsert
(
pPullWins
,
index
,
pPullInfo
)
==
NULL
)
{
return
TSDB_CODE_OUT_OF_MEMORY
;
}
return
TSDB_CODE_SUCCESS
;
}
int32_t
compareResKey
(
void
*
pKey
,
void
*
data
,
int32_t
index
)
{
SArray
*
res
=
(
SArray
*
)
data
;
SResKeyPos
*
pos
=
taosArrayGetP
(
res
,
index
);
...
...
@@ -2586,7 +2621,7 @@ static void doHashInterval(SOperatorInfo* pOperatorInfo, SSDataBlock* pSDataBloc
if
(
isDeletedWindow
(
&
nextWin
,
tableGroupId
,
&
pInfo
->
aggSup
)
&&
!
chIds
)
{
SPullWindowInfo
pull
=
{.
window
=
nextWin
,
.
groupId
=
tableGroupId
};
// add pull data request
taosArrayPush
(
pInfo
->
pPullWins
,
&
pull
);
savePullWindow
(
&
pull
,
pInfo
->
pPullWins
);
int32_t
size
=
taosArrayGetSize
(
pInfo
->
pChildren
);
addPullWindow
(
pInfo
->
pPullDataMap
,
&
winRes
,
size
);
qDebug
(
"===stream===prepare retrive %"
PRId64
", size:%d"
,
winRes
.
ts
,
size
);
...
...
@@ -2674,14 +2709,6 @@ void copyUpdateDataBlock(SSDataBlock* pDest, SSDataBlock* pSource, int32_t tsCol
blockDataUpdateTsWindow
(
pDest
,
0
);
}
static
bool
needBreak
(
SStreamFinalIntervalOperatorInfo
*
pInfo
)
{
int32_t
size
=
taosArrayGetSize
(
pInfo
->
pPullWins
);
if
(
pInfo
->
pullIndex
<
size
)
{
return
true
;
}
return
false
;
}
static
void
doBuildPullDataBlock
(
SArray
*
array
,
int32_t
*
pIndex
,
SSDataBlock
*
pBlock
)
{
clearSpecialDataBlock
(
pBlock
);
int32_t
size
=
taosArrayGetSize
(
array
);
...
...
@@ -2748,6 +2775,14 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) {
if
(
pOperator
->
status
==
OP_EXEC_DONE
)
{
return
NULL
;
}
else
if
(
pOperator
->
status
==
OP_RES_TO_RETURN
)
{
doBuildPullDataBlock
(
pInfo
->
pPullWins
,
&
pInfo
->
pullIndex
,
pInfo
->
pPullDataRes
);
if
(
pInfo
->
pPullDataRes
->
info
.
rows
!=
0
)
{
// process the rest of the data
ASSERT
(
IS_FINAL_OP
(
pInfo
));
printDataBlock
(
pInfo
->
pPullDataRes
,
IS_FINAL_OP
(
pInfo
)
?
"interval Final"
:
"interval Semi"
);
return
pInfo
->
pPullDataRes
;
}
doBuildResultDatablock
(
pOperator
,
&
pInfo
->
binfo
,
&
pInfo
->
groupResInfo
,
pInfo
->
aggSup
.
pResultBuf
);
if
(
pInfo
->
binfo
.
pRes
->
info
.
rows
==
0
)
{
pOperator
->
status
=
OP_EXEC_DONE
;
...
...
@@ -2776,13 +2811,13 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) {
// process the rest of the data
return
pInfo
->
pUpdateRes
;
}
doBuildPullDataBlock
(
pInfo
->
pPullWins
,
&
pInfo
->
pullIndex
,
pInfo
->
pPullDataRes
);
if
(
pInfo
->
pPullDataRes
->
info
.
rows
!=
0
)
{
// process the rest of the data
ASSERT
(
IS_FINAL_OP
(
pInfo
));
printDataBlock
(
pInfo
->
pPullDataRes
,
IS_FINAL_OP
(
pInfo
)
?
"interval Final"
:
"interval Semi"
);
return
pInfo
->
pPullDataRes
;
}
//
doBuildPullDataBlock(pInfo->pPullWins, &pInfo->pullIndex, pInfo->pPullDataRes);
//
if (pInfo->pPullDataRes->info.rows != 0) {
//
// process the rest of the data
//
ASSERT(IS_FINAL_OP(pInfo));
//
printDataBlock(pInfo->pPullDataRes, IS_FINAL_OP(pInfo) ? "interval Final" : "interval Semi");
//
return pInfo->pPullDataRes;
//
}
doBuildDeleteResult
(
pInfo
->
pDelWins
,
&
pInfo
->
delIndex
,
pInfo
->
pDelRes
);
if
(
pInfo
->
pDelRes
->
info
.
rows
!=
0
)
{
// process the rest of the data
...
...
@@ -2882,10 +2917,6 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) {
SStreamFinalIntervalOperatorInfo
*
pChInfo
=
pChildOp
->
info
;
setInputDataBlock
(
pChildOp
,
pChildOp
->
exprSupp
.
pCtx
,
pBlock
,
pChInfo
->
order
,
MAIN_SCAN
,
true
);
doHashInterval
(
pChildOp
,
pBlock
,
pBlock
->
info
.
groupId
,
NULL
);
if
(
needBreak
(
pInfo
))
{
break
;
}
}
}
...
...
@@ -2899,6 +2930,15 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) {
finalizeUpdatedResult
(
pOperator
->
exprSupp
.
numOfExprs
,
pInfo
->
aggSup
.
pResultBuf
,
pUpdated
,
pSup
->
rowEntryInfoOffset
);
initMultiResInfoFromArrayList
(
&
pInfo
->
groupResInfo
,
pUpdated
);
blockDataEnsureCapacity
(
pInfo
->
binfo
.
pRes
,
pOperator
->
resultInfo
.
capacity
);
doBuildPullDataBlock
(
pInfo
->
pPullWins
,
&
pInfo
->
pullIndex
,
pInfo
->
pPullDataRes
);
if
(
pInfo
->
pPullDataRes
->
info
.
rows
!=
0
)
{
// process the rest of the data
ASSERT
(
IS_FINAL_OP
(
pInfo
));
printDataBlock
(
pInfo
->
pPullDataRes
,
IS_FINAL_OP
(
pInfo
)
?
"interval Final"
:
"interval Semi"
);
return
pInfo
->
pPullDataRes
;
}
doBuildResultDatablock
(
pOperator
,
&
pInfo
->
binfo
,
&
pInfo
->
groupResInfo
,
pInfo
->
aggSup
.
pResultBuf
);
if
(
pInfo
->
binfo
.
pRes
->
info
.
rows
!=
0
)
{
printDataBlock
(
pInfo
->
binfo
.
pRes
,
IS_FINAL_OP
(
pInfo
)
?
"interval Final"
:
"interval Semi"
);
...
...
@@ -2913,14 +2953,6 @@ static SSDataBlock* doStreamFinalIntervalAgg(SOperatorInfo* pOperator) {
return
pInfo
->
pUpdateRes
;
}
doBuildPullDataBlock
(
pInfo
->
pPullWins
,
&
pInfo
->
pullIndex
,
pInfo
->
pPullDataRes
);
if
(
pInfo
->
pPullDataRes
->
info
.
rows
!=
0
)
{
// process the rest of the data
ASSERT
(
IS_FINAL_OP
(
pInfo
));
printDataBlock
(
pInfo
->
pPullDataRes
,
IS_FINAL_OP
(
pInfo
)
?
"interval Final"
:
"interval Semi"
);
return
pInfo
->
pPullDataRes
;
}
doBuildDeleteResult
(
pInfo
->
pDelWins
,
&
pInfo
->
delIndex
,
pInfo
->
pDelRes
);
if
(
pInfo
->
pDelRes
->
info
.
rows
!=
0
)
{
// process the rest of the data
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录