Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
849cb8d1
T
TDengine
项目概览
taosdata
/
TDengine
12 个月 前同步成功
通知
1180
Star
22014
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,发现更多精彩内容 >>
提交
849cb8d1
编写于
6月 23, 2022
作者:
S
slzhou
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
feat: split session and state window on stable
上级
bacd7b60
变更
5
隐藏空白更改
内联
并排
Showing
5 changed file
with
91 addition
and
9 deletion
+91
-9
source/libs/executor/src/timewindowoperator.c
source/libs/executor/src/timewindowoperator.c
+1
-2
source/libs/parser/src/parTranslater.c
source/libs/parser/src/parTranslater.c
+0
-3
source/libs/planner/src/planSpliter.c
source/libs/planner/src/planSpliter.c
+71
-4
source/libs/planner/test/planSessionTest.cpp
source/libs/planner/test/planSessionTest.cpp
+10
-0
source/libs/planner/test/planStateTest.cpp
source/libs/planner/test/planStateTest.cpp
+9
-0
未找到文件。
source/libs/executor/src/timewindowoperator.c
浏览文件 @
849cb8d1
...
@@ -1656,10 +1656,9 @@ static SSDataBlock* doSessionWindowAgg(SOperatorInfo* pOperator) {
...
@@ -1656,10 +1656,9 @@ static SSDataBlock* doSessionWindowAgg(SOperatorInfo* pOperator) {
doBuildResultDatablock
(
pOperator
,
pBInfo
,
&
pInfo
->
groupResInfo
,
pInfo
->
aggSup
.
pResultBuf
);
doBuildResultDatablock
(
pOperator
,
pBInfo
,
&
pInfo
->
groupResInfo
,
pInfo
->
aggSup
.
pResultBuf
);
if
(
pBInfo
->
pRes
->
info
.
rows
==
0
||
!
hasDataInGroupInfo
(
&
pInfo
->
groupResInfo
))
{
if
(
pBInfo
->
pRes
->
info
.
rows
==
0
||
!
hasDataInGroupInfo
(
&
pInfo
->
groupResInfo
))
{
doSetOperatorCompleted
(
pOperator
);
doSetOperatorCompleted
(
pOperator
);
return
NULL
;
}
}
return
pBInfo
->
pRes
;
return
pBInfo
->
pRes
->
info
.
rows
>
0
?
pBInfo
->
pRes
:
NULL
;
}
}
int64_t
st
=
taosGetTimestampUs
();
int64_t
st
=
taosGetTimestampUs
();
...
...
source/libs/parser/src/parTranslater.c
浏览文件 @
849cb8d1
...
@@ -2161,9 +2161,6 @@ static EDealRes checkStateExpr(SNode* pNode, void* pContext) {
...
@@ -2161,9 +2161,6 @@ static EDealRes checkStateExpr(SNode* pNode, void* pContext) {
if
(
COLUMN_TYPE_TAG
==
pCol
->
colType
)
{
if
(
COLUMN_TYPE_TAG
==
pCol
->
colType
)
{
return
generateDealNodeErrMsg
(
pCxt
,
TSDB_CODE_PAR_INVALID_STATE_WIN_COL
);
return
generateDealNodeErrMsg
(
pCxt
,
TSDB_CODE_PAR_INVALID_STATE_WIN_COL
);
}
}
if
(
TSDB_SUPER_TABLE
==
pCol
->
tableType
)
{
return
generateDealNodeErrMsg
(
pCxt
,
TSDB_CODE_PAR_INVALID_STATE_WIN_TABLE
);
}
}
}
return
DEAL_RES_CONTINUE
;
return
DEAL_RES_CONTINUE
;
}
}
...
...
source/libs/planner/src/planSpliter.c
浏览文件 @
849cb8d1
...
@@ -176,10 +176,22 @@ static bool stbSplNeedSplit(bool streamQuery, SLogicNode* pNode) {
...
@@ -176,10 +176,22 @@ static bool stbSplNeedSplit(bool streamQuery, SLogicNode* pNode) {
return
!
stbSplHasGatherExecFunc
(((
SAggLogicNode
*
)
pNode
)
->
pAggFuncs
)
&&
stbSplHasMultiTbScan
(
streamQuery
,
pNode
);
return
!
stbSplHasGatherExecFunc
(((
SAggLogicNode
*
)
pNode
)
->
pAggFuncs
)
&&
stbSplHasMultiTbScan
(
streamQuery
,
pNode
);
case
QUERY_NODE_LOGIC_PLAN_WINDOW
:
{
case
QUERY_NODE_LOGIC_PLAN_WINDOW
:
{
SWindowLogicNode
*
pWindow
=
(
SWindowLogicNode
*
)
pNode
;
SWindowLogicNode
*
pWindow
=
(
SWindowLogicNode
*
)
pNode
;
if
(
WINDOW_TYPE_STATE
==
pWindow
->
winType
||
(
!
streamQuery
&&
WINDOW_TYPE_SESSION
==
pWindow
->
winType
))
{
if
(
WINDOW_TYPE_INTERVAL
==
pWindow
->
winType
)
{
return
false
;
return
!
stbSplHasGatherExecFunc
(
pWindow
->
pFuncs
)
&&
stbSplHasMultiTbScan
(
streamQuery
,
pNode
);
if
(
WINDOW_TYPE_STATE
==
pWindow
->
winType
)
{
if
(
!
streamQuery
)
{
return
stbSplHasMultiTbScan
(
streamQuery
,
pNode
);
}
else
{
return
false
;
}
}
if
(
WINDOW_TYPE_SESSION
==
pWindow
->
winType
)
{
if
(
!
streamQuery
)
{
return
stbSplHasMultiTbScan
(
streamQuery
,
pNode
);
}
else
{
return
!
stbSplHasGatherExecFunc
(
pWindow
->
pFuncs
)
&&
stbSplHasMultiTbScan
(
streamQuery
,
pNode
);
}
}
}
return
!
stbSplHasGatherExecFunc
(
pWindow
->
pFuncs
)
&&
stbSplHasMultiTbScan
(
streamQuery
,
pNode
);
}
}
case
QUERY_NODE_LOGIC_PLAN_SORT
:
case
QUERY_NODE_LOGIC_PLAN_SORT
:
return
stbSplHasMultiTbScan
(
streamQuery
,
pNode
);
return
stbSplHasMultiTbScan
(
streamQuery
,
pNode
);
...
@@ -477,11 +489,64 @@ static int32_t stbSplSplitSessionForStream(SSplitContext* pCxt, SStableSplitInfo
...
@@ -477,11 +489,64 @@ static int32_t stbSplSplitSessionForStream(SSplitContext* pCxt, SStableSplitInfo
return
code
;
return
code
;
}
}
static
void
splSetTableScanType
(
SLogicNode
*
pNode
,
EScanType
scanType
)
{
if
(
QUERY_NODE_LOGIC_PLAN_SCAN
==
nodeType
(
pNode
))
{
((
SScanLogicNode
*
)
pNode
)
->
scanType
=
scanType
;
}
else
{
if
(
1
==
LIST_LENGTH
(
pNode
->
pChildren
))
{
splSetTableScanType
((
SLogicNode
*
)
nodesListGetNode
(
pNode
->
pChildren
,
0
),
scanType
);
}
}
}
static
int32_t
stbSplSplitSessionOrStateForBatch
(
SSplitContext
*
pCxt
,
SStableSplitInfo
*
pInfo
)
{
SLogicNode
*
pWindow
=
pInfo
->
pSplitNode
;
SLogicNode
*
pChild
=
(
SLogicNode
*
)
nodesListGetNode
(
pWindow
->
pChildren
,
0
);
SNodeList
*
pMergeKeys
=
NULL
;
int32_t
code
=
stbSplCreateMergeKeysByPrimaryKey
(((
SWindowLogicNode
*
)
pWindow
)
->
pTspk
,
&
pMergeKeys
);
if
(
TSDB_CODE_SUCCESS
==
code
)
{
code
=
stbSplCreateMergeNode
(
pCxt
,
pInfo
->
pSubplan
,
pChild
,
pMergeKeys
,
(
SLogicNode
*
)
pChild
);
}
if
(
TSDB_CODE_SUCCESS
==
code
)
{
code
=
nodesListMakeStrictAppend
(
&
pInfo
->
pSubplan
->
pChildren
,
(
SNode
*
)
splCreateScanSubplan
(
pCxt
,
pChild
,
SPLIT_FLAG_STABLE_SPLIT
));
}
if
(
TSDB_CODE_SUCCESS
==
code
)
{
splSetTableScanType
(
pChild
,
SCAN_TYPE_TABLE_MERGE
);
++
(
pCxt
->
groupId
);
}
if
(
TSDB_CODE_SUCCESS
==
code
)
{
pInfo
->
pSubplan
->
subplanType
=
SUBPLAN_TYPE_MERGE
;
SPLIT_FLAG_SET_MASK
(
pInfo
->
pSubplan
->
splitFlag
,
SPLIT_FLAG_STABLE_SPLIT
);
}
else
{
nodesDestroyList
(
pMergeKeys
);
}
return
code
;
}
static
int32_t
stbSplSplitSession
(
SSplitContext
*
pCxt
,
SStableSplitInfo
*
pInfo
)
{
static
int32_t
stbSplSplitSession
(
SSplitContext
*
pCxt
,
SStableSplitInfo
*
pInfo
)
{
if
(
pCxt
->
pPlanCxt
->
streamQuery
)
{
if
(
pCxt
->
pPlanCxt
->
streamQuery
)
{
return
stbSplSplitSessionForStream
(
pCxt
,
pInfo
);
return
stbSplSplitSessionForStream
(
pCxt
,
pInfo
);
}
else
{
}
else
{
return
TSDB_CODE_PLAN_INTERNAL_ERROR
;
return
stbSplSplitSessionOrStateForBatch
(
pCxt
,
pInfo
);
}
}
static
int32_t
stbSplSplitStateForStream
(
SSplitContext
*
pCxt
,
SStableSplitInfo
*
pInfo
)
{
return
TSDB_CODE_PLAN_INTERNAL_ERROR
;
}
static
int32_t
stbSplSplitState
(
SSplitContext
*
pCxt
,
SStableSplitInfo
*
pInfo
)
{
if
(
pCxt
->
pPlanCxt
->
streamQuery
)
{
return
stbSplSplitStateForStream
(
pCxt
,
pInfo
);
}
else
{
return
stbSplSplitSessionOrStateForBatch
(
pCxt
,
pInfo
);
}
}
}
}
...
@@ -511,6 +576,8 @@ static int32_t stbSplSplitWindowForMergeTable(SSplitContext* pCxt, SStableSplitI
...
@@ -511,6 +576,8 @@ static int32_t stbSplSplitWindowForMergeTable(SSplitContext* pCxt, SStableSplitI
return
stbSplSplitInterval
(
pCxt
,
pInfo
);
return
stbSplSplitInterval
(
pCxt
,
pInfo
);
case
WINDOW_TYPE_SESSION
:
case
WINDOW_TYPE_SESSION
:
return
stbSplSplitSession
(
pCxt
,
pInfo
);
return
stbSplSplitSession
(
pCxt
,
pInfo
);
case
WINDOW_TYPE_STATE
:
return
stbSplSplitState
(
pCxt
,
pInfo
);
default:
default:
break
;
break
;
}
}
...
...
source/libs/planner/test/planSessionTest.cpp
浏览文件 @
849cb8d1
...
@@ -34,3 +34,13 @@ TEST_F(PlanSessionTest, selectFunc) {
...
@@ -34,3 +34,13 @@ TEST_F(PlanSessionTest, selectFunc) {
// select function along with the columns of select row, and with SESSION clause
// select function along with the columns of select row, and with SESSION clause
run
(
"SELECT MAX(c1), c2 FROM t1 SESSION(ts, 10s)"
);
run
(
"SELECT MAX(c1), c2 FROM t1 SESSION(ts, 10s)"
);
}
}
TEST_F
(
PlanSessionTest
,
stable
)
{
useDb
(
"root"
,
"test"
);
// select function for SESSION clause
run
(
"SELECT MAX(c1), MIN(c1) FROM st1 SESSION(ts, 10s)"
);
// select function along with the columns of select row, and with SESSION clause
run
(
"SELECT MAX(c1), c2 FROM st1 SESSION(ts, 10s)"
);
run
(
"SELECT count(ts) FROM st1 PARTITION BY c1 SESSION(ts, 10s)"
);
}
source/libs/planner/test/planStateTest.cpp
浏览文件 @
849cb8d1
...
@@ -40,3 +40,12 @@ TEST_F(PlanStateTest, selectFunc) {
...
@@ -40,3 +40,12 @@ TEST_F(PlanStateTest, selectFunc) {
// select function along with the columns of select row, and with STATE_WINDOW clause
// select function along with the columns of select row, and with STATE_WINDOW clause
run
(
"SELECT MAX(c1), c2 FROM t1 STATE_WINDOW(c3)"
);
run
(
"SELECT MAX(c1), c2 FROM t1 STATE_WINDOW(c3)"
);
}
}
TEST_F
(
PlanStateTest
,
stable
)
{
useDb
(
"root"
,
"test"
);
// select function for STATE_WINDOW clause
run
(
"SELECT MAX(c1), MIN(c1) FROM st1 STATE_WINDOW(c2)"
);
// select function along with the columns of select row, and with STATE_WINDOW clause
run
(
"SELECT MAX(c1), c2 FROM st1 STATE_WINDOW(c2)"
);
}
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录