Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
1ceb1edd
TDengine
项目概览
taosdata
/
TDengine
大约 1 年 前同步成功
通知
1184
Star
22015
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看板
体验新版 GitCode,发现更多精彩内容 >>
未验证
提交
1ceb1edd
编写于
7月 07, 2022
作者:
S
shenglian-zhou
提交者:
GitHub
7月 07, 2022
浏览文件
操作
浏览文件
下载
差异文件
Merge pull request #14628 from taosdata/szhou/feature/push-project-condition
fix: add filter to session/state/interval operator
上级
886af82c
f3c3b0f1
变更
5
隐藏空白更改
内联
并排
Showing
5 changed file
with
84 addition
and
39 deletion
+84
-39
source/libs/executor/inc/executorimpl.h
source/libs/executor/inc/executorimpl.h
+5
-2
source/libs/executor/src/executorimpl.c
source/libs/executor/src/executorimpl.c
+2
-2
source/libs/executor/src/groupoperator.c
source/libs/executor/src/groupoperator.c
+13
-6
source/libs/executor/src/timewindowoperator.c
source/libs/executor/src/timewindowoperator.c
+62
-27
source/libs/planner/src/planOptimizer.c
source/libs/planner/src/planOptimizer.c
+2
-2
未找到文件。
source/libs/executor/inc/executorimpl.h
浏览文件 @
1ceb1edd
...
@@ -588,6 +588,7 @@ typedef struct SSessionAggOperatorInfo {
...
@@ -588,6 +588,7 @@ typedef struct SSessionAggOperatorInfo {
int64_t
gap
;
// session window gap
int64_t
gap
;
// session window gap
int32_t
tsSlotId
;
// primary timestamp slot id
int32_t
tsSlotId
;
// primary timestamp slot id
STimeWindowAggSupp
twAggSup
;
STimeWindowAggSupp
twAggSup
;
SNode
*
pCondition
;
}
SSessionAggOperatorInfo
;
}
SSessionAggOperatorInfo
;
typedef
struct
SResultWindowInfo
{
typedef
struct
SResultWindowInfo
{
...
@@ -649,6 +650,7 @@ typedef struct SStateWindowOperatorInfo {
...
@@ -649,6 +650,7 @@ typedef struct SStateWindowOperatorInfo {
int32_t
tsSlotId
;
// primary timestamp column slot id
int32_t
tsSlotId
;
// primary timestamp column slot id
STimeWindowAggSupp
twAggSup
;
STimeWindowAggSupp
twAggSup
;
// bool reptScan;
// bool reptScan;
const
SNode
*
pCondition
;
}
SStateWindowOperatorInfo
;
}
SStateWindowOperatorInfo
;
typedef
struct
SStreamStateAggOperatorInfo
{
typedef
struct
SStreamStateAggOperatorInfo
{
...
@@ -806,7 +808,7 @@ SOperatorInfo* createStreamIntervalOperatorInfo(SOperatorInfo* downstream, SExpr
...
@@ -806,7 +808,7 @@ SOperatorInfo* createStreamIntervalOperatorInfo(SOperatorInfo* downstream, SExpr
STimeWindowAggSupp
*
pTwAggSupp
,
SExecTaskInfo
*
pTaskInfo
);
STimeWindowAggSupp
*
pTwAggSupp
,
SExecTaskInfo
*
pTaskInfo
);
SOperatorInfo
*
createSessionAggOperatorInfo
(
SOperatorInfo
*
downstream
,
SExprInfo
*
pExprInfo
,
int32_t
numOfCols
,
SOperatorInfo
*
createSessionAggOperatorInfo
(
SOperatorInfo
*
downstream
,
SExprInfo
*
pExprInfo
,
int32_t
numOfCols
,
SSDataBlock
*
pResBlock
,
int64_t
gap
,
int32_t
tsSlotId
,
STimeWindowAggSupp
*
pTwAggSupp
,
SSDataBlock
*
pResBlock
,
int64_t
gap
,
int32_t
tsSlotId
,
STimeWindowAggSupp
*
pTwAggSupp
,
SExecTaskInfo
*
pTaskInfo
);
S
Node
*
pCondition
,
S
ExecTaskInfo
*
pTaskInfo
);
SOperatorInfo
*
createGroupOperatorInfo
(
SOperatorInfo
*
downstream
,
SExprInfo
*
pExprInfo
,
int32_t
numOfCols
,
SOperatorInfo
*
createGroupOperatorInfo
(
SOperatorInfo
*
downstream
,
SExprInfo
*
pExprInfo
,
int32_t
numOfCols
,
SSDataBlock
*
pResultBlock
,
SArray
*
pGroupColList
,
SNode
*
pCondition
,
SSDataBlock
*
pResultBlock
,
SArray
*
pGroupColList
,
SNode
*
pCondition
,
SExprInfo
*
pScalarExprInfo
,
int32_t
numOfScalarExpr
,
SExecTaskInfo
*
pTaskInfo
);
SExprInfo
*
pScalarExprInfo
,
int32_t
numOfScalarExpr
,
SExecTaskInfo
*
pTaskInfo
);
...
@@ -820,7 +822,8 @@ SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SFillPhysiNode*
...
@@ -820,7 +822,8 @@ SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SFillPhysiNode*
SExecTaskInfo
*
pTaskInfo
);
SExecTaskInfo
*
pTaskInfo
);
SOperatorInfo
*
createStatewindowOperatorInfo
(
SOperatorInfo
*
downstream
,
SExprInfo
*
pExpr
,
int32_t
numOfCols
,
SOperatorInfo
*
createStatewindowOperatorInfo
(
SOperatorInfo
*
downstream
,
SExprInfo
*
pExpr
,
int32_t
numOfCols
,
SSDataBlock
*
pResBlock
,
STimeWindowAggSupp
*
pTwAggSupp
,
int32_t
tsSlotId
,
SColumn
*
pStateKeyCol
,
SExecTaskInfo
*
pTaskInfo
);
SSDataBlock
*
pResBlock
,
STimeWindowAggSupp
*
pTwAggSupp
,
int32_t
tsSlotId
,
SColumn
*
pStateKeyCol
,
SNode
*
pCondition
,
SExecTaskInfo
*
pTaskInfo
);
SOperatorInfo
*
createPartitionOperatorInfo
(
SOperatorInfo
*
downstream
,
SPartitionPhysiNode
*
pPartNode
,
SExecTaskInfo
*
pTaskInfo
);
SOperatorInfo
*
createPartitionOperatorInfo
(
SOperatorInfo
*
downstream
,
SPartitionPhysiNode
*
pPartNode
,
SExecTaskInfo
*
pTaskInfo
);
...
...
source/libs/executor/src/executorimpl.c
浏览文件 @
1ceb1edd
...
@@ -4455,7 +4455,7 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
...
@@ -4455,7 +4455,7 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
int32_t
tsSlotId
=
((
SColumnNode
*
)
pSessionNode
->
window
.
pTspk
)
->
slotId
;
int32_t
tsSlotId
=
((
SColumnNode
*
)
pSessionNode
->
window
.
pTspk
)
->
slotId
;
pOptr
=
pOptr
=
createSessionAggOperatorInfo
(
ops
[
0
],
pExprInfo
,
num
,
pResBlock
,
pSessionNode
->
gap
,
tsSlotId
,
&
as
,
pTaskInfo
);
createSessionAggOperatorInfo
(
ops
[
0
],
pExprInfo
,
num
,
pResBlock
,
pSessionNode
->
gap
,
tsSlotId
,
&
as
,
p
PhyNode
->
pConditions
,
p
TaskInfo
);
}
else
if
(
QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION
==
type
)
{
}
else
if
(
QUERY_NODE_PHYSICAL_PLAN_STREAM_SESSION
==
type
)
{
pOptr
=
createStreamSessionAggOperatorInfo
(
ops
[
0
],
pPhyNode
,
pTaskInfo
);
pOptr
=
createStreamSessionAggOperatorInfo
(
ops
[
0
],
pPhyNode
,
pTaskInfo
);
}
else
if
(
QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_SESSION
==
type
)
{
}
else
if
(
QUERY_NODE_PHYSICAL_PLAN_STREAM_SEMI_SESSION
==
type
)
{
...
@@ -4477,7 +4477,7 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
...
@@ -4477,7 +4477,7 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
SColumnNode
*
pColNode
=
(
SColumnNode
*
)((
STargetNode
*
)
pStateNode
->
pStateKey
)
->
pExpr
;
SColumnNode
*
pColNode
=
(
SColumnNode
*
)((
STargetNode
*
)
pStateNode
->
pStateKey
)
->
pExpr
;
SColumn
col
=
extractColumnFromColumnNode
(
pColNode
);
SColumn
col
=
extractColumnFromColumnNode
(
pColNode
);
pOptr
=
createStatewindowOperatorInfo
(
ops
[
0
],
pExprInfo
,
num
,
pResBlock
,
&
as
,
tsSlotId
,
&
col
,
pTaskInfo
);
pOptr
=
createStatewindowOperatorInfo
(
ops
[
0
],
pExprInfo
,
num
,
pResBlock
,
&
as
,
tsSlotId
,
&
col
,
p
PhyNode
->
pConditions
,
p
TaskInfo
);
}
else
if
(
QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE
==
type
)
{
}
else
if
(
QUERY_NODE_PHYSICAL_PLAN_STREAM_STATE
==
type
)
{
pOptr
=
createStreamStateAggOperatorInfo
(
ops
[
0
],
pPhyNode
,
pTaskInfo
);
pOptr
=
createStreamStateAggOperatorInfo
(
ops
[
0
],
pPhyNode
,
pTaskInfo
);
}
else
if
(
QUERY_NODE_PHYSICAL_PLAN_MERGE_JOIN
==
type
)
{
}
else
if
(
QUERY_NODE_PHYSICAL_PLAN_MERGE_JOIN
==
type
)
{
...
...
source/libs/executor/src/groupoperator.c
浏览文件 @
1ceb1edd
...
@@ -302,14 +302,21 @@ static SSDataBlock* hashGroupbyAggregate(SOperatorInfo* pOperator) {
...
@@ -302,14 +302,21 @@ static SSDataBlock* hashGroupbyAggregate(SOperatorInfo* pOperator) {
SSDataBlock
*
pRes
=
pInfo
->
binfo
.
pRes
;
SSDataBlock
*
pRes
=
pInfo
->
binfo
.
pRes
;
if
(
pOperator
->
status
==
OP_RES_TO_RETURN
)
{
if
(
pOperator
->
status
==
OP_RES_TO_RETURN
)
{
doBuildResultDatablock
(
pOperator
,
&
pInfo
->
binfo
,
&
pInfo
->
groupResInfo
,
pInfo
->
aggSup
.
pResultBuf
);
while
(
1
)
{
doBuildResultDatablock
(
pOperator
,
&
pInfo
->
binfo
,
&
pInfo
->
groupResInfo
,
pInfo
->
aggSup
.
pResultBuf
);
doFilter
(
pInfo
->
pCondition
,
pRes
);
bool
hasRemain
=
hasDataInGroupInfo
(
&
pInfo
->
groupResInfo
);
if
(
!
hasRemain
)
{
doSetOperatorCompleted
(
pOperator
);
break
;
}
size_t
rows
=
pRes
->
info
.
rows
;
if
(
pRes
->
info
.
rows
>
0
)
{
if
(
rows
==
0
||
!
hasDataInGroupInfo
(
&
pInfo
->
groupResInfo
))
{
break
;
doSetOperatorCompleted
(
pOperator
);
}
}
}
pOperator
->
resultInfo
.
totalRows
+=
pRes
->
info
.
rows
;
pOperator
->
resultInfo
.
totalRows
+=
rows
;
return
(
pRes
->
info
.
rows
==
0
)
?
NULL
:
pRes
;
return
(
pRes
->
info
.
rows
==
0
)
?
NULL
:
pRes
;
}
}
...
...
source/libs/executor/src/timewindowoperator.c
浏览文件 @
1ceb1edd
...
@@ -1146,13 +1146,22 @@ static SSDataBlock* doStateWindowAgg(SOperatorInfo* pOperator) {
...
@@ -1146,13 +1146,22 @@ static SSDataBlock* doStateWindowAgg(SOperatorInfo* pOperator) {
SOptrBasicInfo
*
pBInfo
=
&
pInfo
->
binfo
;
SOptrBasicInfo
*
pBInfo
=
&
pInfo
->
binfo
;
if
(
pOperator
->
status
==
OP_RES_TO_RETURN
)
{
if
(
pOperator
->
status
==
OP_RES_TO_RETURN
)
{
doBuildResultDatablock
(
pOperator
,
pBInfo
,
&
pInfo
->
groupResInfo
,
pInfo
->
aggSup
.
pResultBuf
);
while
(
1
)
{
if
(
pBInfo
->
pRes
->
info
.
rows
==
0
||
!
hasDataInGroupInfo
(
&
pInfo
->
groupResInfo
))
{
doBuildResultDatablock
(
pOperator
,
&
pInfo
->
binfo
,
&
pInfo
->
groupResInfo
,
pInfo
->
aggSup
.
pResultBuf
);
doSetOperatorCompleted
(
pOperator
);
doFilter
(
pInfo
->
pCondition
,
pBInfo
->
pRes
);
return
NULL
;
}
return
pBInfo
->
pRes
;
bool
hasRemain
=
hasDataInGroupInfo
(
&
pInfo
->
groupResInfo
);
if
(
!
hasRemain
)
{
doSetOperatorCompleted
(
pOperator
);
break
;
}
if
(
pBInfo
->
pRes
->
info
.
rows
>
0
)
{
break
;
}
}
pOperator
->
resultInfo
.
totalRows
+=
pBInfo
->
pRes
->
info
.
rows
;
return
(
pBInfo
->
pRes
->
info
.
rows
==
0
)
?
NULL
:
pBInfo
->
pRes
;
}
}
int32_t
order
=
TSDB_ORDER_ASC
;
int32_t
order
=
TSDB_ORDER_ASC
;
...
@@ -1178,15 +1187,22 @@ static SSDataBlock* doStateWindowAgg(SOperatorInfo* pOperator) {
...
@@ -1178,15 +1187,22 @@ static SSDataBlock* doStateWindowAgg(SOperatorInfo* pOperator) {
initGroupedResultInfo
(
&
pInfo
->
groupResInfo
,
pInfo
->
aggSup
.
pResultRowHashTable
,
TSDB_ORDER_ASC
);
initGroupedResultInfo
(
&
pInfo
->
groupResInfo
,
pInfo
->
aggSup
.
pResultRowHashTable
,
TSDB_ORDER_ASC
);
blockDataEnsureCapacity
(
pBInfo
->
pRes
,
pOperator
->
resultInfo
.
capacity
);
blockDataEnsureCapacity
(
pBInfo
->
pRes
,
pOperator
->
resultInfo
.
capacity
);
doBuildResultDatablock
(
pOperator
,
pBInfo
,
&
pInfo
->
groupResInfo
,
pInfo
->
aggSup
.
pResultBuf
);
while
(
1
)
{
if
(
pBInfo
->
pRes
->
info
.
rows
==
0
||
!
hasDataInGroupInfo
(
&
pInfo
->
groupResInfo
))
{
doBuildResultDatablock
(
pOperator
,
&
pInfo
->
binfo
,
&
pInfo
->
groupResInfo
,
pInfo
->
aggSup
.
pResultBuf
);
doSetOperatorCompleted
(
pOperator
);
doFilter
(
pInfo
->
pCondition
,
pBInfo
->
pRes
);
}
size_t
rows
=
pBInfo
->
pRes
->
info
.
rows
;
bool
hasRemain
=
hasDataInGroupInfo
(
&
pInfo
->
groupResInfo
);
pOperator
->
resultInfo
.
totalRows
+=
rows
;
if
(
!
hasRemain
)
{
doSetOperatorCompleted
(
pOperator
);
break
;
}
return
(
rows
==
0
)
?
NULL
:
pBInfo
->
pRes
;
if
(
pBInfo
->
pRes
->
info
.
rows
>
0
)
{
break
;
}
}
pOperator
->
resultInfo
.
totalRows
+=
pBInfo
->
pRes
->
info
.
rows
;
return
(
pBInfo
->
pRes
->
info
.
rows
==
0
)
?
NULL
:
pBInfo
->
pRes
;
}
}
static
SSDataBlock
*
doBuildIntervalResult
(
SOperatorInfo
*
pOperator
)
{
static
SSDataBlock
*
doBuildIntervalResult
(
SOperatorInfo
*
pOperator
)
{
...
@@ -1880,12 +1896,22 @@ static SSDataBlock* doSessionWindowAgg(SOperatorInfo* pOperator) {
...
@@ -1880,12 +1896,22 @@ static SSDataBlock* doSessionWindowAgg(SOperatorInfo* pOperator) {
SExprSupp
*
pSup
=
&
pOperator
->
exprSupp
;
SExprSupp
*
pSup
=
&
pOperator
->
exprSupp
;
if
(
pOperator
->
status
==
OP_RES_TO_RETURN
)
{
if
(
pOperator
->
status
==
OP_RES_TO_RETURN
)
{
doBuildResultDatablock
(
pOperator
,
pBInfo
,
&
pInfo
->
groupResInfo
,
pInfo
->
aggSup
.
pResultBuf
);
while
(
1
)
{
if
(
pBInfo
->
pRes
->
info
.
rows
==
0
||
!
hasDataInGroupInfo
(
&
pInfo
->
groupResInfo
))
{
doBuildResultDatablock
(
pOperator
,
&
pInfo
->
binfo
,
&
pInfo
->
groupResInfo
,
pInfo
->
aggSup
.
pResultBuf
);
doSetOperatorCompleted
(
pOperator
);
doFilter
(
pInfo
->
pCondition
,
pBInfo
->
pRes
);
}
return
pBInfo
->
pRes
->
info
.
rows
>
0
?
pBInfo
->
pRes
:
NULL
;
bool
hasRemain
=
hasDataInGroupInfo
(
&
pInfo
->
groupResInfo
);
if
(
!
hasRemain
)
{
doSetOperatorCompleted
(
pOperator
);
break
;
}
if
(
pBInfo
->
pRes
->
info
.
rows
>
0
)
{
break
;
}
}
pOperator
->
resultInfo
.
totalRows
+=
pBInfo
->
pRes
->
info
.
rows
;
return
(
pBInfo
->
pRes
->
info
.
rows
==
0
)
?
NULL
:
pBInfo
->
pRes
;
}
}
int64_t
st
=
taosGetTimestampUs
();
int64_t
st
=
taosGetTimestampUs
();
...
@@ -1914,15 +1940,22 @@ static SSDataBlock* doSessionWindowAgg(SOperatorInfo* pOperator) {
...
@@ -1914,15 +1940,22 @@ static SSDataBlock* doSessionWindowAgg(SOperatorInfo* pOperator) {
initGroupedResultInfo
(
&
pInfo
->
groupResInfo
,
pInfo
->
aggSup
.
pResultRowHashTable
,
TSDB_ORDER_ASC
);
initGroupedResultInfo
(
&
pInfo
->
groupResInfo
,
pInfo
->
aggSup
.
pResultRowHashTable
,
TSDB_ORDER_ASC
);
blockDataEnsureCapacity
(
pBInfo
->
pRes
,
pOperator
->
resultInfo
.
capacity
);
blockDataEnsureCapacity
(
pBInfo
->
pRes
,
pOperator
->
resultInfo
.
capacity
);
doBuildResultDatablock
(
pOperator
,
pBInfo
,
&
pInfo
->
groupResInfo
,
pInfo
->
aggSup
.
pResultBuf
);
while
(
1
)
{
if
(
pBInfo
->
pRes
->
info
.
rows
==
0
||
!
hasDataInGroupInfo
(
&
pInfo
->
groupResInfo
))
{
doBuildResultDatablock
(
pOperator
,
&
pInfo
->
binfo
,
&
pInfo
->
groupResInfo
,
pInfo
->
aggSup
.
pResultBuf
);
doSetOperatorCompleted
(
pOperator
);
doFilter
(
pInfo
->
pCondition
,
pBInfo
->
pRes
);
}
size_t
rows
=
pBInfo
->
pRes
->
info
.
rows
;
bool
hasRemain
=
hasDataInGroupInfo
(
&
pInfo
->
groupResInfo
);
pOperator
->
resultInfo
.
totalRows
+=
rows
;
if
(
!
hasRemain
)
{
doSetOperatorCompleted
(
pOperator
);
break
;
}
return
(
rows
==
0
)
?
NULL
:
pBInfo
->
pRes
;
if
(
pBInfo
->
pRes
->
info
.
rows
>
0
)
{
break
;
}
}
pOperator
->
resultInfo
.
totalRows
+=
pBInfo
->
pRes
->
info
.
rows
;
return
(
pBInfo
->
pRes
->
info
.
rows
==
0
)
?
NULL
:
pBInfo
->
pRes
;
}
}
static
void
doKeepPrevRows
(
STimeSliceOperatorInfo
*
pSliceInfo
,
const
SSDataBlock
*
pBlock
,
int32_t
rowIndex
)
{
static
void
doKeepPrevRows
(
STimeSliceOperatorInfo
*
pSliceInfo
,
const
SSDataBlock
*
pBlock
,
int32_t
rowIndex
)
{
...
@@ -2235,7 +2268,7 @@ _error:
...
@@ -2235,7 +2268,7 @@ _error:
SOperatorInfo
*
createStatewindowOperatorInfo
(
SOperatorInfo
*
downstream
,
SExprInfo
*
pExpr
,
int32_t
numOfCols
,
SOperatorInfo
*
createStatewindowOperatorInfo
(
SOperatorInfo
*
downstream
,
SExprInfo
*
pExpr
,
int32_t
numOfCols
,
SSDataBlock
*
pResBlock
,
STimeWindowAggSupp
*
pTwAggSup
,
int32_t
tsSlotId
,
SSDataBlock
*
pResBlock
,
STimeWindowAggSupp
*
pTwAggSup
,
int32_t
tsSlotId
,
SColumn
*
pStateKeyCol
,
SExecTaskInfo
*
pTaskInfo
)
{
SColumn
*
pStateKeyCol
,
S
Node
*
pCondition
,
S
ExecTaskInfo
*
pTaskInfo
)
{
SStateWindowOperatorInfo
*
pInfo
=
taosMemoryCalloc
(
1
,
sizeof
(
SStateWindowOperatorInfo
));
SStateWindowOperatorInfo
*
pInfo
=
taosMemoryCalloc
(
1
,
sizeof
(
SStateWindowOperatorInfo
));
SOperatorInfo
*
pOperator
=
taosMemoryCalloc
(
1
,
sizeof
(
SOperatorInfo
));
SOperatorInfo
*
pOperator
=
taosMemoryCalloc
(
1
,
sizeof
(
SOperatorInfo
));
if
(
pInfo
==
NULL
||
pOperator
==
NULL
)
{
if
(
pInfo
==
NULL
||
pOperator
==
NULL
)
{
...
@@ -2246,6 +2279,7 @@ SOperatorInfo* createStatewindowOperatorInfo(SOperatorInfo* downstream, SExprInf
...
@@ -2246,6 +2279,7 @@ SOperatorInfo* createStatewindowOperatorInfo(SOperatorInfo* downstream, SExprInf
pInfo
->
stateKey
.
type
=
pInfo
->
stateCol
.
type
;
pInfo
->
stateKey
.
type
=
pInfo
->
stateCol
.
type
;
pInfo
->
stateKey
.
bytes
=
pInfo
->
stateCol
.
bytes
;
pInfo
->
stateKey
.
bytes
=
pInfo
->
stateCol
.
bytes
;
pInfo
->
stateKey
.
pData
=
taosMemoryCalloc
(
1
,
pInfo
->
stateCol
.
bytes
);
pInfo
->
stateKey
.
pData
=
taosMemoryCalloc
(
1
,
pInfo
->
stateCol
.
bytes
);
pInfo
->
pCondition
=
pCondition
;
if
(
pInfo
->
stateKey
.
pData
==
NULL
)
{
if
(
pInfo
->
stateKey
.
pData
==
NULL
)
{
goto
_error
;
goto
_error
;
}
}
...
@@ -2289,7 +2323,7 @@ void destroySWindowOperatorInfo(void* param, int32_t numOfOutput) {
...
@@ -2289,7 +2323,7 @@ void destroySWindowOperatorInfo(void* param, int32_t numOfOutput) {
SOperatorInfo
*
createSessionAggOperatorInfo
(
SOperatorInfo
*
downstream
,
SExprInfo
*
pExprInfo
,
int32_t
numOfCols
,
SOperatorInfo
*
createSessionAggOperatorInfo
(
SOperatorInfo
*
downstream
,
SExprInfo
*
pExprInfo
,
int32_t
numOfCols
,
SSDataBlock
*
pResBlock
,
int64_t
gap
,
int32_t
tsSlotId
,
SSDataBlock
*
pResBlock
,
int64_t
gap
,
int32_t
tsSlotId
,
STimeWindowAggSupp
*
pTwAggSupp
,
SExecTaskInfo
*
pTaskInfo
)
{
STimeWindowAggSupp
*
pTwAggSupp
,
S
Node
*
pCondition
,
S
ExecTaskInfo
*
pTaskInfo
)
{
SSessionAggOperatorInfo
*
pInfo
=
taosMemoryCalloc
(
1
,
sizeof
(
SSessionAggOperatorInfo
));
SSessionAggOperatorInfo
*
pInfo
=
taosMemoryCalloc
(
1
,
sizeof
(
SSessionAggOperatorInfo
));
SOperatorInfo
*
pOperator
=
taosMemoryCalloc
(
1
,
sizeof
(
SOperatorInfo
));
SOperatorInfo
*
pOperator
=
taosMemoryCalloc
(
1
,
sizeof
(
SOperatorInfo
));
if
(
pInfo
==
NULL
||
pOperator
==
NULL
)
{
if
(
pInfo
==
NULL
||
pOperator
==
NULL
)
{
...
@@ -2315,6 +2349,7 @@ SOperatorInfo* createSessionAggOperatorInfo(SOperatorInfo* downstream, SExprInfo
...
@@ -2315,6 +2349,7 @@ SOperatorInfo* createSessionAggOperatorInfo(SOperatorInfo* downstream, SExprInfo
pInfo
->
binfo
.
pRes
=
pResBlock
;
pInfo
->
binfo
.
pRes
=
pResBlock
;
pInfo
->
winSup
.
prevTs
=
INT64_MIN
;
pInfo
->
winSup
.
prevTs
=
INT64_MIN
;
pInfo
->
reptScan
=
false
;
pInfo
->
reptScan
=
false
;
pInfo
->
pCondition
=
pCondition
;
pOperator
->
name
=
"SessionWindowAggOperator"
;
pOperator
->
name
=
"SessionWindowAggOperator"
;
pOperator
->
operatorType
=
QUERY_NODE_PHYSICAL_PLAN_MERGE_SESSION
;
pOperator
->
operatorType
=
QUERY_NODE_PHYSICAL_PLAN_MERGE_SESSION
;
pOperator
->
blocking
=
true
;
pOperator
->
blocking
=
true
;
...
...
source/libs/planner/src/planOptimizer.c
浏览文件 @
1ceb1edd
...
@@ -890,7 +890,7 @@ static int32_t pushDownCondOptDealProject(SOptimizeContext* pCxt, SProjectLogicN
...
@@ -890,7 +890,7 @@ static int32_t pushDownCondOptDealProject(SOptimizeContext* pCxt, SProjectLogicN
return
code
;
return
code
;
}
}
static
int32_t
pushDownCondOpt
DealLogicNode
(
SOptimizeContext
*
pCxt
,
SLogicNode
*
pLogicNode
)
{
static
int32_t
pushDownCondOpt
TrivialPushDown
(
SOptimizeContext
*
pCxt
,
SLogicNode
*
pLogicNode
)
{
if
(
NULL
==
pLogicNode
->
pConditions
||
if
(
NULL
==
pLogicNode
->
pConditions
||
OPTIMIZE_FLAG_TEST_MASK
(
pLogicNode
->
optimizedFlag
,
OPTIMIZE_FLAG_PUSH_DOWN_CONDE
))
{
OPTIMIZE_FLAG_TEST_MASK
(
pLogicNode
->
optimizedFlag
,
OPTIMIZE_FLAG_PUSH_DOWN_CONDE
))
{
return
TSDB_CODE_SUCCESS
;
return
TSDB_CODE_SUCCESS
;
...
@@ -921,7 +921,7 @@ static int32_t pushDownCondOptimizeImpl(SOptimizeContext* pCxt, SLogicNode* pLog
...
@@ -921,7 +921,7 @@ static int32_t pushDownCondOptimizeImpl(SOptimizeContext* pCxt, SLogicNode* pLog
break
;
break
;
case
QUERY_NODE_LOGIC_PLAN_SORT
:
case
QUERY_NODE_LOGIC_PLAN_SORT
:
case
QUERY_NODE_LOGIC_PLAN_PARTITION
:
case
QUERY_NODE_LOGIC_PLAN_PARTITION
:
code
=
pushDownCondOpt
DealLogicNode
(
pCxt
,
pLogicNode
);
code
=
pushDownCondOpt
TrivialPushDown
(
pCxt
,
pLogicNode
);
break
;
break
;
default:
default:
break
;
break
;
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录