Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
lzh_me
TDengine
提交
a5b4ae69
TDengine
项目概览
lzh_me
/
TDengine
与 Fork 源项目一致
Fork自
taosdata / TDengine
通知
1
Star
0
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
TDengine
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
a5b4ae69
编写于
3月 30, 2021
作者:
D
dapan1121
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
support child table having query
上级
22882196
变更
4
展开全部
隐藏空白更改
内联
并排
Showing
4 changed file
with
2048 addition
and
5 deletion
+2048
-5
src/client/src/tscLocalMerge.c
src/client/src/tscLocalMerge.c
+2
-2
src/query/inc/qExecutor.h
src/query/inc/qExecutor.h
+8
-0
src/query/src/qExecutor.c
src/query/src/qExecutor.c
+199
-3
tests/script/general/parser/having_child.sim
tests/script/general/parser/having_child.sim
+1839
-0
未找到文件。
src/client/src/tscLocalMerge.c
浏览文件 @
a5b4ae69
...
...
@@ -1235,7 +1235,7 @@ static bool saveGroupResultInfo(SSqlObj *pSql) {
}
bool
doFilterFieldData
(
SQueryInfo
*
pQueryInfo
,
char
*
input
,
tFilePage
*
pOut
put
,
SExprFilter
*
pFieldFilters
,
int16_t
type
,
bool
*
notSkipped
)
{
bool
doFilterFieldData
(
char
*
in
put
,
SExprFilter
*
pFieldFilters
,
int16_t
type
,
bool
*
notSkipped
)
{
bool
qualified
=
false
;
for
(
int32_t
k
=
0
;
k
<
pFieldFilters
->
pFilters
->
numOfFilters
;
++
k
)
{
...
...
@@ -1293,7 +1293,7 @@ int32_t doHavingFilter(SQueryInfo* pQueryInfo, tFilePage* pOutput, bool* notSkip
char
*
pInput
=
pOutput
->
data
+
pOutput
->
num
*
pFieldFilters
->
pSqlExpr
->
offset
;
doFilterFieldData
(
p
QueryInfo
,
pInput
,
pOut
put
,
pFieldFilters
,
type
,
notSkipped
);
doFilterFieldData
(
p
In
put
,
pFieldFilters
,
type
,
notSkipped
);
if
(
*
notSkipped
==
false
)
{
return
TSDB_CODE_SUCCESS
;
}
...
...
src/query/inc/qExecutor.h
浏览文件 @
a5b4ae69
...
...
@@ -189,6 +189,8 @@ typedef struct SQuery {
bool
stabledev
;
// super table stddev query
int32_t
interBufSize
;
// intermediate buffer sizse
int32_t
havingNum
;
// having expr number
SOrderVal
order
;
int16_t
numOfCols
;
int16_t
numOfTags
;
...
...
@@ -284,6 +286,7 @@ enum OPERATOR_TYPE_E {
OP_Fill
=
13
,
OP_MultiTableAggregate
=
14
,
OP_MultiTableTimeInterval
=
15
,
OP_Having
=
16
,
};
typedef
struct
SOperatorInfo
{
...
...
@@ -401,6 +404,11 @@ typedef struct SOffsetOperatorInfo {
int64_t
offset
;
}
SOffsetOperatorInfo
;
typedef
struct
SHavingOperatorInfo
{
SArray
*
fp
;
}
SHavingOperatorInfo
;
typedef
struct
SFillOperatorInfo
{
SFillInfo
*
pFillInfo
;
SSDataBlock
*
pRes
;
...
...
src/query/src/qExecutor.c
浏览文件 @
a5b4ae69
...
...
@@ -181,6 +181,7 @@ static SOperatorInfo* createMultiTableAggOperatorInfo(SQueryRuntimeEnv* pRuntime
static
SOperatorInfo
*
createMultiTableTimeIntervalOperatorInfo
(
SQueryRuntimeEnv
*
pRuntimeEnv
,
SOperatorInfo
*
upstream
,
SExprInfo
*
pExpr
,
int32_t
numOfOutput
);
static
SOperatorInfo
*
createTagScanOperatorInfo
(
SQueryRuntimeEnv
*
pRuntimeEnv
,
SExprInfo
*
pExpr
,
int32_t
numOfOutput
);
static
SOperatorInfo
*
createTableBlockInfoScanOperator
(
void
*
pTsdbQueryHandle
,
SQueryRuntimeEnv
*
pRuntimeEnv
);
static
SOperatorInfo
*
createHavingOperatorInfo
(
SQueryRuntimeEnv
*
pRuntimeEnv
,
SOperatorInfo
*
upstream
,
SExprInfo
*
pExpr
,
int32_t
numOfOutput
);
static
void
destroyBasicOperatorInfo
(
void
*
param
,
int32_t
numOfOutput
);
static
void
destroySFillOperatorInfo
(
void
*
param
,
int32_t
numOfOutput
);
...
...
@@ -1819,6 +1820,10 @@ static int32_t setupQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv, int32_t numOf
}
}
if
(
pQuery
->
havingNum
>
0
)
{
pRuntimeEnv
->
proot
=
createHavingOperatorInfo
(
pRuntimeEnv
,
pRuntimeEnv
->
proot
,
pQuery
->
pExpr1
,
pQuery
->
numOfOutput
);
}
if
(
pQuery
->
limit
.
offset
>
0
)
{
pRuntimeEnv
->
proot
=
createOffsetOperatorInfo
(
pRuntimeEnv
,
pRuntimeEnv
->
proot
);
}
...
...
@@ -4653,6 +4658,111 @@ static SSDataBlock* doOffset(void* param) {
}
}
bool
doFilterData
(
SColumnInfoData
*
p
,
int32_t
rid
,
SColumnFilterElem
*
filterElem
,
__filter_func_t
fp
)
{
char
*
input
=
p
->
pData
+
p
->
info
.
bytes
*
rid
;
bool
isnull
=
isNull
(
input
,
p
->
info
.
type
);
if
(
isnull
)
{
return
(
fp
==
isNullOperator
)
?
true
:
false
;
}
else
{
if
(
fp
==
notNullOperator
)
{
return
true
;
}
else
if
(
fp
==
isNullOperator
)
{
return
false
;
}
}
if
(
fp
(
filterElem
,
input
,
input
,
p
->
info
.
type
))
{
return
true
;
}
return
false
;
}
void
doHavingImpl
(
SOperatorInfo
*
pOperator
,
SSDataBlock
*
pBlock
)
{
SHavingOperatorInfo
*
pInfo
=
pOperator
->
info
;
int32_t
f
=
0
;
int32_t
allQualified
=
1
;
int32_t
exprQualified
=
0
;
for
(
int32_t
r
=
0
;
r
<
pBlock
->
info
.
rows
;
++
r
)
{
allQualified
=
1
;
for
(
int32_t
i
=
0
;
i
<
pOperator
->
numOfOutput
;
++
i
)
{
SExprInfo
*
pExprInfo
=
&
(
pOperator
->
pExpr
[
i
]);
if
(
pExprInfo
->
pFilter
==
NULL
)
{
continue
;
}
SArray
*
es
=
taosArrayGetP
(
pInfo
->
fp
,
i
);
assert
(
es
);
size_t
fpNum
=
taosArrayGetSize
(
es
);
exprQualified
=
0
;
for
(
int32_t
m
=
0
;
m
<
fpNum
;
++
m
)
{
__filter_func_t
fp
=
taosArrayGetP
(
es
,
m
);
assert
(
fp
);
//SColIndex* colIdx = &pExprInfo->base.colInfo;
SColumnInfoData
*
p
=
taosArrayGet
(
pBlock
->
pDataBlock
,
i
);
SColumnFilterElem
filterElem
=
{.
filterInfo
=
*
pExprInfo
->
pFilter
};
if
(
doFilterData
(
p
,
r
,
&
filterElem
,
fp
))
{
exprQualified
=
1
;
break
;
}
}
if
(
exprQualified
==
0
)
{
allQualified
=
0
;
break
;
}
}
if
(
allQualified
==
0
)
{
continue
;
}
for
(
int32_t
i
=
0
;
i
<
pBlock
->
info
.
numOfCols
;
++
i
)
{
SColumnInfoData
*
pColInfoData
=
taosArrayGet
(
pBlock
->
pDataBlock
,
i
);
int16_t
bytes
=
pColInfoData
->
info
.
bytes
;
memmove
(
pColInfoData
->
pData
+
f
*
bytes
,
pColInfoData
->
pData
+
bytes
*
r
,
bytes
);
}
++
f
;
}
pBlock
->
info
.
rows
=
f
;
}
static
SSDataBlock
*
doHaving
(
void
*
param
)
{
SOperatorInfo
*
pOperator
=
(
SOperatorInfo
*
)
param
;
if
(
pOperator
->
status
==
OP_EXEC_DONE
)
{
return
NULL
;
}
SQueryRuntimeEnv
*
pRuntimeEnv
=
pOperator
->
pRuntimeEnv
;
while
(
1
)
{
SSDataBlock
*
pBlock
=
pOperator
->
upstream
->
exec
(
pOperator
->
upstream
);
if
(
pBlock
==
NULL
)
{
setQueryStatus
(
pRuntimeEnv
,
QUERY_COMPLETED
);
pOperator
->
status
=
OP_EXEC_DONE
;
return
NULL
;
}
doHavingImpl
(
pOperator
,
pBlock
);
return
pBlock
;
}
}
static
SSDataBlock
*
doIntervalAgg
(
void
*
param
)
{
SOperatorInfo
*
pOperator
=
(
SOperatorInfo
*
)
param
;
if
(
pOperator
->
status
==
OP_EXEC_DONE
)
{
...
...
@@ -5003,6 +5113,13 @@ static void destroyTagScanOperatorInfo(void* param, int32_t numOfOutput) {
pInfo
->
pRes
=
destroyOutputBuf
(
pInfo
->
pRes
);
}
static
void
destroyHavingOperatorInfo
(
void
*
param
,
int32_t
numOfOutput
)
{
SHavingOperatorInfo
*
pInfo
=
(
SHavingOperatorInfo
*
)
param
;
if
(
pInfo
->
fp
)
{
taosArrayDestroy
(
pInfo
->
fp
);
}
}
SOperatorInfo
*
createMultiTableAggOperatorInfo
(
SQueryRuntimeEnv
*
pRuntimeEnv
,
SOperatorInfo
*
upstream
,
SExprInfo
*
pExpr
,
int32_t
numOfOutput
)
{
SAggOperatorInfo
*
pInfo
=
calloc
(
1
,
sizeof
(
SAggOperatorInfo
));
...
...
@@ -5059,6 +5176,81 @@ SOperatorInfo* createArithOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorI
return
pOperator
;
}
int32_t
initFilterFp
(
SExprInfo
*
pExpr
,
int32_t
numOfOutput
,
SArray
**
fps
)
{
__filter_func_t
fp
=
NULL
;
*
fps
=
taosArrayInit
(
numOfOutput
,
sizeof
(
SArray
*
));
if
(
*
fps
==
NULL
)
{
return
TSDB_CODE_TSC_OUT_OF_MEMORY
;
}
for
(
int32_t
i
=
0
;
i
<
numOfOutput
;
++
i
)
{
SExprInfo
*
pExprInfo
=
&
(
pExpr
[
i
]);
SColIndex
*
colIdx
=
&
pExprInfo
->
base
.
colInfo
;
if
(
pExprInfo
->
pFilter
==
NULL
||
!
TSDB_COL_IS_NORMAL_COL
(
colIdx
->
flag
))
{
taosArrayPush
(
*
fps
,
&
fp
);
continue
;
}
int32_t
filterNum
=
pExprInfo
->
base
.
filterNum
;
SColumnFilterInfo
*
filterInfo
=
pExprInfo
->
pFilter
;
SArray
*
es
=
taosArrayInit
(
filterNum
,
sizeof
(
__filter_func_t
));
for
(
int32_t
j
=
0
;
j
<
filterNum
;
++
j
)
{
int32_t
lower
=
filterInfo
->
lowerRelOptr
;
int32_t
upper
=
filterInfo
->
upperRelOptr
;
if
(
lower
==
TSDB_RELATION_INVALID
&&
upper
==
TSDB_RELATION_INVALID
)
{
qError
(
"invalid rel optr"
);
return
TSDB_CODE_QRY_APP_ERROR
;
}
__filter_func_t
ffp
=
getFilterOperator
(
lower
,
upper
);
if
(
ffp
==
NULL
)
{
qError
(
"invalid filter info"
);
return
TSDB_CODE_QRY_APP_ERROR
;
}
taosArrayPush
(
es
,
&
ffp
);
filterInfo
+=
1
;
}
taosArrayPush
(
*
fps
,
&
es
);
}
return
TSDB_CODE_SUCCESS
;
}
SOperatorInfo
*
createHavingOperatorInfo
(
SQueryRuntimeEnv
*
pRuntimeEnv
,
SOperatorInfo
*
upstream
,
SExprInfo
*
pExpr
,
int32_t
numOfOutput
)
{
SHavingOperatorInfo
*
pInfo
=
calloc
(
1
,
sizeof
(
SHavingOperatorInfo
));
initFilterFp
(
pExpr
,
numOfOutput
,
&
pInfo
->
fp
);
assert
(
pInfo
->
fp
);
SOperatorInfo
*
pOperator
=
calloc
(
1
,
sizeof
(
SOperatorInfo
));
pOperator
->
name
=
"HavingOperator"
;
pOperator
->
operatorType
=
OP_Having
;
pOperator
->
blockingOptr
=
false
;
pOperator
->
status
=
OP_IN_EXECUTING
;
pOperator
->
numOfOutput
=
numOfOutput
;
pOperator
->
pExpr
=
pExpr
;
pOperator
->
upstream
=
upstream
;
pOperator
->
exec
=
doHaving
;
pOperator
->
info
=
pInfo
;
pOperator
->
pRuntimeEnv
=
pRuntimeEnv
;
pOperator
->
cleanup
=
destroyHavingOperatorInfo
;
return
pOperator
;
}
SOperatorInfo
*
createLimitOperatorInfo
(
SQueryRuntimeEnv
*
pRuntimeEnv
,
SOperatorInfo
*
upstream
)
{
SLimitOperatorInfo
*
pInfo
=
calloc
(
1
,
sizeof
(
SLimitOperatorInfo
));
pInfo
->
limit
=
pRuntimeEnv
->
pQuery
->
limit
.
limit
;
...
...
@@ -5856,8 +6048,8 @@ int32_t cloneExprFilterInfo(SColumnFilterInfo **dst, SColumnFilterInfo* src, int
memcpy
(
*
dst
,
src
,
sizeof
(
*
src
)
*
filterNum
);
for
(
int32_t
i
=
0
;
i
<
filterNum
;
i
++
)
{
if
(
dst
[
i
]
->
filterstr
&&
dst
[
i
]
->
len
>
0
)
{
void
*
pz
=
calloc
(
1
,
dst
[
i
]
->
len
+
1
);
if
(
(
*
dst
)[
i
].
filterstr
&&
dst
[
i
]
->
len
>
0
)
{
void
*
pz
=
calloc
(
1
,
(
*
dst
)[
i
].
len
+
1
);
if
(
pz
==
NULL
)
{
if
(
i
==
0
)
{
...
...
@@ -5871,7 +6063,7 @@ int32_t cloneExprFilterInfo(SColumnFilterInfo **dst, SColumnFilterInfo* src, int
memcpy
(
pz
,
(
void
*
)
src
->
pz
,
src
->
len
+
1
);
dst
[
i
]
->
pz
=
(
int64_t
)
pz
;
(
*
dst
)[
i
].
pz
=
(
int64_t
)
pz
;
}
}
...
...
@@ -6288,6 +6480,10 @@ SQInfo* createQInfoImpl(SQueryTableMsg* pQueryMsg, SSqlGroupbyExpr* pGroupbyExpr
if
(
TSDB_COL_IS_TAG
(
pExprs
[
col
].
base
.
colInfo
.
flag
))
{
pQuery
->
tagLen
+=
pExprs
[
col
].
bytes
;
}
if
(
pExprs
[
col
].
pFilter
)
{
++
pQuery
->
havingNum
;
}
}
doUpdateExprColumnIndex
(
pQuery
);
...
...
tests/script/general/parser/having_child.sim
0 → 100644
浏览文件 @
a5b4ae69
此差异已折叠。
点击以展开。
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录