Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
c96036a7
TDengine
项目概览
taosdata
/
TDengine
1 年多 前同步成功
通知
1185
Star
22016
Fork
4786
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
1
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
TDengine
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
1
Issue
1
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
提交
c96036a7
编写于
5月 21, 2022
作者:
H
Haojun Liao
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
fix(query): add null value check before update the block window info
上级
00821527
变更
1
隐藏空白更改
内联
并排
Showing
1 changed file
with
21 addition
and
18 deletion
+21
-18
source/libs/executor/src/timewindowoperator.c
source/libs/executor/src/timewindowoperator.c
+21
-18
未找到文件。
source/libs/executor/src/timewindowoperator.c
浏览文件 @
c96036a7
...
...
@@ -621,7 +621,7 @@ static void saveDataBlockLastRow(char** pRow, SArray* pDataBlock, int32_t rowInd
}
}
static
SArray
*
hashIntervalAgg
(
SOperatorInfo
*
pOperatorInfo
,
SResultRowInfo
*
pResultRowInfo
,
SSDataBlock
*
p
SData
Block
,
static
SArray
*
hashIntervalAgg
(
SOperatorInfo
*
pOperatorInfo
,
SResultRowInfo
*
pResultRowInfo
,
SSDataBlock
*
pBlock
,
uint64_t
tableGroupId
)
{
SIntervalAggOperatorInfo
*
pInfo
=
(
SIntervalAggOperatorInfo
*
)
pOperatorInfo
->
info
;
...
...
@@ -639,13 +639,17 @@ static SArray* hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pRe
// int32_t prevIndex = pResultRowInfo->curPos;
TSKEY
*
tsCols
=
NULL
;
if
(
p
SData
Block
->
pDataBlock
!=
NULL
)
{
SColumnInfoData
*
pColDataInfo
=
taosArrayGet
(
p
SData
Block
->
pDataBlock
,
pInfo
->
primaryTsIndex
);
if
(
pBlock
->
pDataBlock
!=
NULL
)
{
SColumnInfoData
*
pColDataInfo
=
taosArrayGet
(
pBlock
->
pDataBlock
,
pInfo
->
primaryTsIndex
);
tsCols
=
(
int64_t
*
)
pColDataInfo
->
pData
;
if
(
tsCols
!=
NULL
)
{
blockDataUpdateTsWindow
(
pBlock
,
pInfo
->
primaryTsIndex
);
}
}
int32_t
startPos
=
0
;
TSKEY
ts
=
getStartTsKey
(
&
p
SDataBlock
->
info
.
window
,
tsCols
,
pSData
Block
->
info
.
rows
,
ascScan
);
TSKEY
ts
=
getStartTsKey
(
&
p
Block
->
info
.
window
,
tsCols
,
p
Block
->
info
.
rows
,
ascScan
);
STimeWindow
win
=
getActiveTimeWindow
(
pInfo
->
aggSup
.
pResultBuf
,
pResultRowInfo
,
ts
,
&
pInfo
->
interval
,
pInfo
->
interval
.
precision
,
&
pInfo
->
win
);
...
...
@@ -670,7 +674,7 @@ static SArray* hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pRe
int32_t
forwardStep
=
0
;
TSKEY
ekey
=
ascScan
?
win
.
ekey
:
win
.
skey
;
forwardStep
=
getNumOfRowsInTimeWindow
(
&
p
SData
Block
->
info
,
tsCols
,
startPos
,
ekey
,
binarySearchForKey
,
NULL
,
pInfo
->
order
);
getNumOfRowsInTimeWindow
(
&
pBlock
->
info
,
tsCols
,
startPos
,
ekey
,
binarySearchForKey
,
NULL
,
pInfo
->
order
);
ASSERT
(
forwardStep
>
0
);
// prev time window not interpolation yet.
...
...
@@ -686,7 +690,7 @@ static SArray* hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pRe
}
STimeWindow w = pRes->win;
ret = setTimeWindowOutputBuf(pResultRowInfo, p
SData
Block->info.uid, &w, masterScan, &pResult, tableGroupId,
ret = setTimeWindowOutputBuf(pResultRowInfo, pBlock->info.uid, &w, masterScan, &pResult, tableGroupId,
pInfo->binfo.pCtx, numOfOutput, pInfo->binfo.rowCellInfoOffset, &pInfo->aggSup,
pTaskInfo);
if (ret != TSDB_CODE_SUCCESS) {
...
...
@@ -694,17 +698,17 @@ static SArray* hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pRe
}
assert(!resultRowInterpolated(pResult, RESULT_ROW_END_INTERP));
doTimeWindowInterpolation(pOperatorInfo, &pInfo->binfo, p
SData
Block->pDataBlock, *(TSKEY*)pInfo->pRow[0], -1,
doTimeWindowInterpolation(pOperatorInfo, &pInfo->binfo, pBlock->pDataBlock, *(TSKEY*)pInfo->pRow[0], -1,
tsCols[startPos], startPos, w.ekey, RESULT_ROW_END_INTERP);
setResultRowInterpo(pResult, RESULT_ROW_END_INTERP);
setNotInterpoWindowKey(pInfo->binfo.pCtx, pOperatorInfo->numOfExprs, RESULT_ROW_START_INTERP);
doApplyFunctions(pInfo->binfo.pCtx, &w, &pInfo->timeWindowData, startPos, 0, tsCols, p
SData
Block->info.rows, numOfOutput, TSDB_ORDER_ASC);
doApplyFunctions(pInfo->binfo.pCtx, &w, &pInfo->timeWindowData, startPos, 0, tsCols, pBlock->info.rows, numOfOutput, TSDB_ORDER_ASC);
}
// restore current time window
ret = setTimeWindowOutputBuf(pResultRowInfo, p
SData
Block->info.uid, &win, masterScan, &pResult, tableGroupId,
ret = setTimeWindowOutputBuf(pResultRowInfo, pBlock->info.uid, &win, masterScan, &pResult, tableGroupId,
pInfo->binfo.pCtx, numOfOutput, pInfo->binfo.rowCellInfoOffset, &pInfo->aggSup,
pTaskInfo);
if (ret != TSDB_CODE_SUCCESS) {
...
...
@@ -714,17 +718,17 @@ static SArray* hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pRe
#endif
// window start key interpolation
doWindowBorderInterpolation
(
pOperatorInfo
,
p
SData
Block
,
pInfo
->
binfo
.
pCtx
,
pResult
,
&
win
,
startPos
,
forwardStep
,
doWindowBorderInterpolation
(
pOperatorInfo
,
pBlock
,
pInfo
->
binfo
.
pCtx
,
pResult
,
&
win
,
startPos
,
forwardStep
,
pInfo
->
order
,
false
);
updateTimeWindowInfo
(
&
pInfo
->
twAggSup
.
timeWindowData
,
&
win
,
true
);
doApplyFunctions
(
pTaskInfo
,
pInfo
->
binfo
.
pCtx
,
&
win
,
&
pInfo
->
twAggSup
.
timeWindowData
,
startPos
,
forwardStep
,
tsCols
,
p
SData
Block
->
info
.
rows
,
numOfOutput
,
TSDB_ORDER_ASC
);
pBlock
->
info
.
rows
,
numOfOutput
,
TSDB_ORDER_ASC
);
STimeWindow
nextWin
=
win
;
while
(
1
)
{
int32_t
prevEndPos
=
(
forwardStep
-
1
)
*
step
+
startPos
;
startPos
=
getNextQualifiedWindow
(
&
pInfo
->
interval
,
&
nextWin
,
&
p
SData
Block
->
info
,
tsCols
,
prevEndPos
,
pInfo
->
order
);
startPos
=
getNextQualifiedWindow
(
&
pInfo
->
interval
,
&
nextWin
,
&
pBlock
->
info
,
tsCols
,
prevEndPos
,
pInfo
->
order
);
if
(
startPos
<
0
)
{
break
;
}
...
...
@@ -748,20 +752,20 @@ static SArray* hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pRe
ekey
=
ascScan
?
nextWin
.
ekey
:
nextWin
.
skey
;
forwardStep
=
getNumOfRowsInTimeWindow
(
&
p
SData
Block
->
info
,
tsCols
,
startPos
,
ekey
,
binarySearchForKey
,
NULL
,
pInfo
->
order
);
getNumOfRowsInTimeWindow
(
&
pBlock
->
info
,
tsCols
,
startPos
,
ekey
,
binarySearchForKey
,
NULL
,
pInfo
->
order
);
// window start(end) key interpolation
doWindowBorderInterpolation
(
pOperatorInfo
,
p
SData
Block
,
pInfo
->
binfo
.
pCtx
,
pResult
,
&
nextWin
,
startPos
,
forwardStep
,
doWindowBorderInterpolation
(
pOperatorInfo
,
pBlock
,
pInfo
->
binfo
.
pCtx
,
pResult
,
&
nextWin
,
startPos
,
forwardStep
,
pInfo
->
order
,
false
);
updateTimeWindowInfo
(
&
pInfo
->
twAggSup
.
timeWindowData
,
&
nextWin
,
true
);
doApplyFunctions
(
pTaskInfo
,
pInfo
->
binfo
.
pCtx
,
&
nextWin
,
&
pInfo
->
twAggSup
.
timeWindowData
,
startPos
,
forwardStep
,
tsCols
,
p
SData
Block
->
info
.
rows
,
numOfOutput
,
TSDB_ORDER_ASC
);
pBlock
->
info
.
rows
,
numOfOutput
,
TSDB_ORDER_ASC
);
}
if
(
pInfo
->
timeWindowInterpo
)
{
int32_t
rowIndex
=
ascScan
?
(
p
SData
Block
->
info
.
rows
-
1
)
:
0
;
saveDataBlockLastRow
(
pInfo
->
pRow
,
p
SDataBlock
->
pDataBlock
,
rowIndex
,
pSData
Block
->
info
.
numOfCols
);
int32_t
rowIndex
=
ascScan
?
(
pBlock
->
info
.
rows
-
1
)
:
0
;
saveDataBlockLastRow
(
pInfo
->
pRow
,
p
Block
->
pDataBlock
,
rowIndex
,
p
Block
->
info
.
numOfCols
);
}
return
pUpdated
;
...
...
@@ -790,7 +794,6 @@ static int32_t doOpenIntervalAgg(SOperatorInfo* pOperator) {
}
getTableScanInfo
(
pOperator
,
&
pInfo
->
order
,
&
scanFlag
);
blockDataUpdateTsWindow
(
pBlock
,
pInfo
->
primaryTsIndex
);
// the pDataBlock are always the same one, no need to call this again
setInputDataBlock
(
pOperator
,
pInfo
->
binfo
.
pCtx
,
pBlock
,
pInfo
->
order
,
scanFlag
,
true
);
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录