Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
d5aa32bf
T
TDengine
项目概览
taosdata
/
TDengine
大约 2 年 前同步成功
通知
1192
Star
22018
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看板
提交
d5aa32bf
编写于
11月 16, 2020
作者:
H
Haojun Liao
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
[TD-2060]
上级
875149dd
变更
4
显示空白变更内容
内联
并排
Showing
4 changed file
with
98 addition
and
94 deletion
+98
-94
src/query/inc/qExecutor.h
src/query/inc/qExecutor.h
+1
-1
src/query/inc/qUtil.h
src/query/inc/qUtil.h
+2
-2
src/query/src/qExecutor.c
src/query/src/qExecutor.c
+92
-88
src/query/src/qUtil.c
src/query/src/qUtil.c
+3
-3
未找到文件。
src/query/inc/qExecutor.h
浏览文件 @
d5aa32bf
...
@@ -152,7 +152,7 @@ typedef struct SQuery {
...
@@ -152,7 +152,7 @@ typedef struct SQuery {
SLimitVal
limit
;
SLimitVal
limit
;
int32_t
rowSize
;
int32_t
rowSize
;
SSqlGroupbyExpr
*
pGroupbyExpr
;
SSqlGroupbyExpr
*
pGroupbyExpr
;
SExprInfo
*
p
SelectExpr
;
SExprInfo
*
p
Expr1
;
SExprInfo
*
pExpr2
;
SExprInfo
*
pExpr2
;
int32_t
numOfExpr2
;
int32_t
numOfExpr2
;
...
...
src/query/inc/qUtil.h
浏览文件 @
d5aa32bf
...
@@ -48,7 +48,7 @@ static FORCE_INLINE SResultRow *getResultRow(SWindowResInfo *pWindowResInfo, int
...
@@ -48,7 +48,7 @@ static FORCE_INLINE SResultRow *getResultRow(SWindowResInfo *pWindowResInfo, int
}
}
#define curTimeWindowIndex(_winres) ((_winres)->curIndex)
#define curTimeWindowIndex(_winres) ((_winres)->curIndex)
#define GET_ROW_PARAM_FOR_MULTIOUTPUT(_q, tbq, sq) (((tbq) && (!sq))? (_q)->p
SelectExpr
[1].base.arg->argValue.i64:1)
#define GET_ROW_PARAM_FOR_MULTIOUTPUT(_q, tbq, sq) (((tbq) && (!sq))? (_q)->p
Expr1
[1].base.arg->argValue.i64:1)
bool
isWindowResClosed
(
SWindowResInfo
*
pWindowResInfo
,
int32_t
slot
);
bool
isWindowResClosed
(
SWindowResInfo
*
pWindowResInfo
,
int32_t
slot
);
...
@@ -62,7 +62,7 @@ static FORCE_INLINE char *getPosInResultPage(SQueryRuntimeEnv *pRuntimeEnv, int3
...
@@ -62,7 +62,7 @@ static FORCE_INLINE char *getPosInResultPage(SQueryRuntimeEnv *pRuntimeEnv, int3
int32_t
realRowId
=
(
int32_t
)(
pResult
->
rowId
*
GET_ROW_PARAM_FOR_MULTIOUTPUT
(
pQuery
,
pRuntimeEnv
->
topBotQuery
,
pRuntimeEnv
->
stableQuery
));
int32_t
realRowId
=
(
int32_t
)(
pResult
->
rowId
*
GET_ROW_PARAM_FOR_MULTIOUTPUT
(
pQuery
,
pRuntimeEnv
->
topBotQuery
,
pRuntimeEnv
->
stableQuery
));
return
((
char
*
)
page
->
data
)
+
pRuntimeEnv
->
offset
[
columnIndex
]
*
pRuntimeEnv
->
numOfRowsPerPage
+
return
((
char
*
)
page
->
data
)
+
pRuntimeEnv
->
offset
[
columnIndex
]
*
pRuntimeEnv
->
numOfRowsPerPage
+
pQuery
->
p
SelectExpr
[
columnIndex
].
bytes
*
realRowId
;
pQuery
->
p
Expr1
[
columnIndex
].
bytes
*
realRowId
;
}
}
bool
isNull_filter
(
SColumnFilterElem
*
pFilter
,
char
*
minval
,
char
*
maxval
);
bool
isNull_filter
(
SColumnFilterElem
*
pFilter
,
char
*
minval
,
char
*
maxval
);
...
...
src/query/src/qExecutor.c
浏览文件 @
d5aa32bf
...
@@ -242,7 +242,7 @@ int64_t getNumOfResult(SQueryRuntimeEnv *pRuntimeEnv) {
...
@@ -242,7 +242,7 @@ int64_t getNumOfResult(SQueryRuntimeEnv *pRuntimeEnv) {
int64_t
maxOutput
=
0
;
int64_t
maxOutput
=
0
;
for
(
int32_t
j
=
0
;
j
<
pQuery
->
numOfOutput
;
++
j
)
{
for
(
int32_t
j
=
0
;
j
<
pQuery
->
numOfOutput
;
++
j
)
{
int32_t
functionId
=
pQuery
->
p
SelectExpr
[
j
].
base
.
functionId
;
int32_t
functionId
=
pQuery
->
p
Expr1
[
j
].
base
.
functionId
;
/*
/*
* ts, tag, tagprj function can not decide the output number of current query
* ts, tag, tagprj function can not decide the output number of current query
...
@@ -337,7 +337,7 @@ bool isSelectivityWithTagsQuery(SQuery *pQuery) {
...
@@ -337,7 +337,7 @@ bool isSelectivityWithTagsQuery(SQuery *pQuery) {
int32_t
numOfSelectivity
=
0
;
int32_t
numOfSelectivity
=
0
;
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
int32_t
functId
=
pQuery
->
p
SelectExpr
[
i
].
base
.
functionId
;
int32_t
functId
=
pQuery
->
p
Expr1
[
i
].
base
.
functionId
;
if
(
functId
==
TSDB_FUNC_TAG_DUMMY
||
functId
==
TSDB_FUNC_TS_DUMMY
)
{
if
(
functId
==
TSDB_FUNC_TAG_DUMMY
||
functId
==
TSDB_FUNC_TS_DUMMY
)
{
hasTags
=
true
;
hasTags
=
true
;
continue
;
continue
;
...
@@ -357,7 +357,7 @@ bool isSelectivityWithTagsQuery(SQuery *pQuery) {
...
@@ -357,7 +357,7 @@ bool isSelectivityWithTagsQuery(SQuery *pQuery) {
bool
isProjQuery
(
SQuery
*
pQuery
)
{
bool
isProjQuery
(
SQuery
*
pQuery
)
{
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
int32_t
functId
=
pQuery
->
p
SelectExpr
[
i
].
base
.
functionId
;
int32_t
functId
=
pQuery
->
p
Expr1
[
i
].
base
.
functionId
;
if
(
functId
!=
TSDB_FUNC_PRJ
&&
functId
!=
TSDB_FUNC_TAGPRJ
)
{
if
(
functId
!=
TSDB_FUNC_PRJ
&&
functId
!=
TSDB_FUNC_TAGPRJ
)
{
return
false
;
return
false
;
}
}
...
@@ -366,7 +366,7 @@ bool isProjQuery(SQuery *pQuery) {
...
@@ -366,7 +366,7 @@ bool isProjQuery(SQuery *pQuery) {
return
true
;
return
true
;
}
}
bool
isTSCompQuery
(
SQuery
*
pQuery
)
{
return
pQuery
->
p
SelectExpr
[
0
].
base
.
functionId
==
TSDB_FUNC_TS_COMP
;
}
bool
isTSCompQuery
(
SQuery
*
pQuery
)
{
return
pQuery
->
p
Expr1
[
0
].
base
.
functionId
==
TSDB_FUNC_TS_COMP
;
}
static
bool
limitResults
(
SQueryRuntimeEnv
*
pRuntimeEnv
)
{
static
bool
limitResults
(
SQueryRuntimeEnv
*
pRuntimeEnv
)
{
SQInfo
*
pQInfo
=
GET_QINFO_ADDR
(
pRuntimeEnv
);
SQInfo
*
pQInfo
=
GET_QINFO_ADDR
(
pRuntimeEnv
);
...
@@ -387,7 +387,7 @@ static bool limitResults(SQueryRuntimeEnv* pRuntimeEnv) {
...
@@ -387,7 +387,7 @@ static bool limitResults(SQueryRuntimeEnv* pRuntimeEnv) {
static
bool
isTopBottomQuery
(
SQuery
*
pQuery
)
{
static
bool
isTopBottomQuery
(
SQuery
*
pQuery
)
{
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
int32_t
functionId
=
pQuery
->
p
SelectExpr
[
i
].
base
.
functionId
;
int32_t
functionId
=
pQuery
->
p
Expr1
[
i
].
base
.
functionId
;
if
(
functionId
==
TSDB_FUNC_TS
)
{
if
(
functionId
==
TSDB_FUNC_TS
)
{
continue
;
continue
;
}
}
...
@@ -401,12 +401,12 @@ static bool isTopBottomQuery(SQuery *pQuery) {
...
@@ -401,12 +401,12 @@ static bool isTopBottomQuery(SQuery *pQuery) {
}
}
static
bool
hasTagValOutput
(
SQuery
*
pQuery
)
{
static
bool
hasTagValOutput
(
SQuery
*
pQuery
)
{
SExprInfo
*
pExprInfo
=
&
pQuery
->
p
SelectExpr
[
0
];
SExprInfo
*
pExprInfo
=
&
pQuery
->
p
Expr1
[
0
];
if
(
pQuery
->
numOfOutput
==
1
&&
pExprInfo
->
base
.
functionId
==
TSDB_FUNC_TS_COMP
)
{
if
(
pQuery
->
numOfOutput
==
1
&&
pExprInfo
->
base
.
functionId
==
TSDB_FUNC_TS_COMP
)
{
return
true
;
return
true
;
}
else
{
// set tag value, by which the results are aggregated.
}
else
{
// set tag value, by which the results are aggregated.
for
(
int32_t
idx
=
0
;
idx
<
pQuery
->
numOfOutput
;
++
idx
)
{
for
(
int32_t
idx
=
0
;
idx
<
pQuery
->
numOfOutput
;
++
idx
)
{
SExprInfo
*
pLocalExprInfo
=
&
pQuery
->
p
SelectExpr
[
idx
];
SExprInfo
*
pLocalExprInfo
=
&
pQuery
->
p
Expr1
[
idx
];
// ts_comp column required the tag value for join filter
// ts_comp column required the tag value for join filter
if
(
TSDB_COL_IS_TAG
(
pLocalExprInfo
->
base
.
colInfo
.
flag
))
{
if
(
TSDB_COL_IS_TAG
(
pLocalExprInfo
->
base
.
colInfo
.
flag
))
{
...
@@ -784,7 +784,7 @@ static void doBlockwiseApplyFunctions(SQueryRuntimeEnv *pRuntimeEnv, bool closed
...
@@ -784,7 +784,7 @@ static void doBlockwiseApplyFunctions(SQueryRuntimeEnv *pRuntimeEnv, bool closed
pCtx
[
k
].
size
=
forwardStep
;
pCtx
[
k
].
size
=
forwardStep
;
pCtx
[
k
].
startOffset
=
(
QUERY_IS_ASC_QUERY
(
pQuery
))
?
offset
:
offset
-
(
forwardStep
-
1
);
pCtx
[
k
].
startOffset
=
(
QUERY_IS_ASC_QUERY
(
pQuery
))
?
offset
:
offset
-
(
forwardStep
-
1
);
int32_t
functionId
=
pQuery
->
p
SelectExpr
[
k
].
base
.
functionId
;
int32_t
functionId
=
pQuery
->
p
Expr1
[
k
].
base
.
functionId
;
if
((
aAggs
[
functionId
].
nStatus
&
TSDB_FUNCSTATE_SELECTIVITY
)
!=
0
)
{
if
((
aAggs
[
functionId
].
nStatus
&
TSDB_FUNCSTATE_SELECTIVITY
)
!=
0
)
{
pCtx
[
k
].
ptsList
=
&
tsCol
[
pCtx
[
k
].
startOffset
];
pCtx
[
k
].
ptsList
=
&
tsCol
[
pCtx
[
k
].
startOffset
];
}
}
...
@@ -813,7 +813,7 @@ static void doRowwiseApplyFunctions(SQueryRuntimeEnv *pRuntimeEnv, bool closed,
...
@@ -813,7 +813,7 @@ static void doRowwiseApplyFunctions(SQueryRuntimeEnv *pRuntimeEnv, bool closed,
for
(
int32_t
k
=
0
;
k
<
pQuery
->
numOfOutput
;
++
k
)
{
for
(
int32_t
k
=
0
;
k
<
pQuery
->
numOfOutput
;
++
k
)
{
pCtx
[
k
].
nStartQueryTimestamp
=
pWin
->
skey
;
pCtx
[
k
].
nStartQueryTimestamp
=
pWin
->
skey
;
int32_t
functionId
=
pQuery
->
p
SelectExpr
[
k
].
base
.
functionId
;
int32_t
functionId
=
pQuery
->
p
Expr1
[
k
].
base
.
functionId
;
if
(
functionNeedToExecute
(
pRuntimeEnv
,
&
pCtx
[
k
],
functionId
))
{
if
(
functionNeedToExecute
(
pRuntimeEnv
,
&
pCtx
[
k
],
functionId
))
{
aAggs
[
functionId
].
xFunctionF
(
&
pCtx
[
k
],
offset
);
aAggs
[
functionId
].
xFunctionF
(
&
pCtx
[
k
],
offset
);
}
}
...
@@ -922,9 +922,9 @@ static char *getDataBlock(SQueryRuntimeEnv *pRuntimeEnv, SArithmeticSupport *sas
...
@@ -922,9 +922,9 @@ static char *getDataBlock(SQueryRuntimeEnv *pRuntimeEnv, SArithmeticSupport *sas
char
*
dataBlock
=
NULL
;
char
*
dataBlock
=
NULL
;
SQuery
*
pQuery
=
pRuntimeEnv
->
pQuery
;
SQuery
*
pQuery
=
pRuntimeEnv
->
pQuery
;
int32_t
functionId
=
pQuery
->
p
SelectExpr
[
col
].
base
.
functionId
;
int32_t
functionId
=
pQuery
->
p
Expr1
[
col
].
base
.
functionId
;
if
(
functionId
==
TSDB_FUNC_ARITHM
)
{
if
(
functionId
==
TSDB_FUNC_ARITHM
)
{
sas
->
pArithExpr
=
&
pQuery
->
p
SelectExpr
[
col
];
sas
->
pArithExpr
=
&
pQuery
->
p
Expr1
[
col
];
sas
->
offset
=
0
;
sas
->
offset
=
0
;
sas
->
colList
=
pQuery
->
colList
;
sas
->
colList
=
pQuery
->
colList
;
...
@@ -954,9 +954,9 @@ static char *getDataBlock(SQueryRuntimeEnv *pRuntimeEnv, SArithmeticSupport *sas
...
@@ -954,9 +954,9 @@ static char *getDataBlock(SQueryRuntimeEnv *pRuntimeEnv, SArithmeticSupport *sas
}
}
}
else
{
// other type of query function
}
else
{
// other type of query function
SColIndex
*
pCol
=
&
pQuery
->
p
SelectExpr
[
col
].
base
.
colInfo
;
SColIndex
*
pCol
=
&
pQuery
->
p
Expr1
[
col
].
base
.
colInfo
;
if
(
TSDB_COL_IS_NORMAL_COL
(
pCol
->
flag
))
{
if
(
TSDB_COL_IS_NORMAL_COL
(
pCol
->
flag
))
{
SColIndex
*
pColIndex
=
&
pQuery
->
p
SelectExpr
[
col
].
base
.
colInfo
;
SColIndex
*
pColIndex
=
&
pQuery
->
p
Expr1
[
col
].
base
.
colInfo
;
SColumnInfoData
*
p
=
taosArrayGet
(
pDataBlock
,
pColIndex
->
colIndex
);
SColumnInfoData
*
p
=
taosArrayGet
(
pDataBlock
,
pColIndex
->
colIndex
);
assert
(
p
->
info
.
colId
==
pColIndex
->
colId
);
assert
(
p
->
info
.
colId
==
pColIndex
->
colId
);
...
@@ -1067,7 +1067,7 @@ static void blockwiseApplyFunctions(SQueryRuntimeEnv *pRuntimeEnv, SDataStatis *
...
@@ -1067,7 +1067,7 @@ static void blockwiseApplyFunctions(SQueryRuntimeEnv *pRuntimeEnv, SDataStatis *
* tag_prj function are changed to be TSDB_FUNC_TAG_DUMMY
* tag_prj function are changed to be TSDB_FUNC_TAG_DUMMY
*/
*/
for
(
int32_t
k
=
0
;
k
<
pQuery
->
numOfOutput
;
++
k
)
{
for
(
int32_t
k
=
0
;
k
<
pQuery
->
numOfOutput
;
++
k
)
{
int32_t
functionId
=
pQuery
->
p
SelectExpr
[
k
].
base
.
functionId
;
int32_t
functionId
=
pQuery
->
p
Expr1
[
k
].
base
.
functionId
;
if
(
functionNeedToExecute
(
pRuntimeEnv
,
&
pCtx
[
k
],
functionId
))
{
if
(
functionNeedToExecute
(
pRuntimeEnv
,
&
pCtx
[
k
],
functionId
))
{
aAggs
[
functionId
].
xFunction
(
&
pCtx
[
k
]);
aAggs
[
functionId
].
xFunction
(
&
pCtx
[
k
]);
}
}
...
@@ -1075,7 +1075,7 @@ static void blockwiseApplyFunctions(SQueryRuntimeEnv *pRuntimeEnv, SDataStatis *
...
@@ -1075,7 +1075,7 @@ static void blockwiseApplyFunctions(SQueryRuntimeEnv *pRuntimeEnv, SDataStatis *
}
}
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
if
(
pQuery
->
p
SelectExpr
[
i
].
base
.
functionId
!=
TSDB_FUNC_ARITHM
)
{
if
(
pQuery
->
p
Expr1
[
i
].
base
.
functionId
!=
TSDB_FUNC_ARITHM
)
{
continue
;
continue
;
}
}
...
@@ -1375,7 +1375,7 @@ static void rowwiseApplyFunctions(SQueryRuntimeEnv *pRuntimeEnv, SDataStatis *pS
...
@@ -1375,7 +1375,7 @@ static void rowwiseApplyFunctions(SQueryRuntimeEnv *pRuntimeEnv, SDataStatis *pS
}
}
for
(
int32_t
k
=
0
;
k
<
pQuery
->
numOfOutput
;
++
k
)
{
for
(
int32_t
k
=
0
;
k
<
pQuery
->
numOfOutput
;
++
k
)
{
int32_t
functionId
=
pQuery
->
p
SelectExpr
[
k
].
base
.
functionId
;
int32_t
functionId
=
pQuery
->
p
Expr1
[
k
].
base
.
functionId
;
if
(
functionNeedToExecute
(
pRuntimeEnv
,
&
pCtx
[
k
],
functionId
))
{
if
(
functionNeedToExecute
(
pRuntimeEnv
,
&
pCtx
[
k
],
functionId
))
{
aAggs
[
functionId
].
xFunctionF
(
&
pCtx
[
k
],
offset
);
aAggs
[
functionId
].
xFunctionF
(
&
pCtx
[
k
],
offset
);
}
}
...
@@ -1404,7 +1404,7 @@ static void rowwiseApplyFunctions(SQueryRuntimeEnv *pRuntimeEnv, SDataStatis *pS
...
@@ -1404,7 +1404,7 @@ static void rowwiseApplyFunctions(SQueryRuntimeEnv *pRuntimeEnv, SDataStatis *pS
// todo refactor: extract method
// todo refactor: extract method
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
if
(
pQuery
->
p
SelectExpr
[
i
].
base
.
functionId
!=
TSDB_FUNC_ARITHM
)
{
if
(
pQuery
->
p
Expr1
[
i
].
base
.
functionId
!=
TSDB_FUNC_ARITHM
)
{
continue
;
continue
;
}
}
...
@@ -1464,11 +1464,11 @@ static int32_t tableApplyFunctionsOnBlock(SQueryRuntimeEnv *pRuntimeEnv, SDataBl
...
@@ -1464,11 +1464,11 @@ static int32_t tableApplyFunctionsOnBlock(SQueryRuntimeEnv *pRuntimeEnv, SDataBl
void
setExecParams
(
SQuery
*
pQuery
,
SQLFunctionCtx
*
pCtx
,
void
*
inputData
,
TSKEY
*
tsCol
,
SDataBlockInfo
*
pBlockInfo
,
void
setExecParams
(
SQuery
*
pQuery
,
SQLFunctionCtx
*
pCtx
,
void
*
inputData
,
TSKEY
*
tsCol
,
SDataBlockInfo
*
pBlockInfo
,
SDataStatis
*
pStatis
,
void
*
param
,
int32_t
colIndex
,
int32_t
vgId
)
{
SDataStatis
*
pStatis
,
void
*
param
,
int32_t
colIndex
,
int32_t
vgId
)
{
int32_t
functionId
=
pQuery
->
p
SelectExpr
[
colIndex
].
base
.
functionId
;
int32_t
functionId
=
pQuery
->
p
Expr1
[
colIndex
].
base
.
functionId
;
int32_t
colId
=
pQuery
->
p
SelectExpr
[
colIndex
].
base
.
colInfo
.
colId
;
int32_t
colId
=
pQuery
->
p
Expr1
[
colIndex
].
base
.
colInfo
.
colId
;
SDataStatis
*
tpField
=
NULL
;
SDataStatis
*
tpField
=
NULL
;
pCtx
->
hasNull
=
hasNullValue
(
&
pQuery
->
p
SelectExpr
[
colIndex
].
base
.
colInfo
,
pStatis
,
&
tpField
);
pCtx
->
hasNull
=
hasNullValue
(
&
pQuery
->
p
Expr1
[
colIndex
].
base
.
colInfo
,
pStatis
,
&
tpField
);
pCtx
->
aInputElemBuf
=
inputData
;
pCtx
->
aInputElemBuf
=
inputData
;
if
(
tpField
!=
NULL
)
{
if
(
tpField
!=
NULL
)
{
...
@@ -1501,7 +1501,7 @@ void setExecParams(SQuery *pQuery, SQLFunctionCtx *pCtx, void* inputData, TSKEY
...
@@ -1501,7 +1501,7 @@ void setExecParams(SQuery *pQuery, SQLFunctionCtx *pCtx, void* inputData, TSKEY
functionId
==
TSDB_FUNC_DIFF
||
(
functionId
>=
TSDB_FUNC_RATE
&&
functionId
<=
TSDB_FUNC_AVG_IRATE
))
{
functionId
==
TSDB_FUNC_DIFF
||
(
functionId
>=
TSDB_FUNC_RATE
&&
functionId
<=
TSDB_FUNC_AVG_IRATE
))
{
/*
/*
* least squares function needs two columns of input, currently, the x value of linear equation is set to
* least squares function needs two columns of input, currently, the x value of linear equation is set to
* timestamp column, and the y-value is the column specified in pQuery->p
SelectExpr
[i].colIdxInBuffer
* timestamp column, and the y-value is the column specified in pQuery->p
Expr1
[i].colIdxInBuffer
*
*
* top/bottom function needs timestamp to indicate when the
* top/bottom function needs timestamp to indicate when the
* top/bottom values emerge, so does diff function
* top/bottom values emerge, so does diff function
...
@@ -1574,7 +1574,7 @@ static int32_t setCtxTagColumnInfo(SQueryRuntimeEnv *pRuntimeEnv, SQLFunctionCtx
...
@@ -1574,7 +1574,7 @@ static int32_t setCtxTagColumnInfo(SQueryRuntimeEnv *pRuntimeEnv, SQLFunctionCtx
}
}
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
SSqlFuncMsg
*
pSqlFuncMsg
=
&
pQuery
->
p
SelectExpr
[
i
].
base
;
SSqlFuncMsg
*
pSqlFuncMsg
=
&
pQuery
->
p
Expr1
[
i
].
base
;
if
(
pSqlFuncMsg
->
functionId
==
TSDB_FUNC_TAG_DUMMY
||
pSqlFuncMsg
->
functionId
==
TSDB_FUNC_TS_DUMMY
)
{
if
(
pSqlFuncMsg
->
functionId
==
TSDB_FUNC_TAG_DUMMY
||
pSqlFuncMsg
->
functionId
==
TSDB_FUNC_TS_DUMMY
)
{
tagLen
+=
pCtx
[
i
].
outputBytes
;
tagLen
+=
pCtx
[
i
].
outputBytes
;
...
@@ -1615,7 +1615,7 @@ static int32_t setupQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv, int16_t order
...
@@ -1615,7 +1615,7 @@ static int32_t setupQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv, int16_t order
pRuntimeEnv
->
offset
[
0
]
=
0
;
pRuntimeEnv
->
offset
[
0
]
=
0
;
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
SSqlFuncMsg
*
pSqlFuncMsg
=
&
pQuery
->
p
SelectExpr
[
i
].
base
;
SSqlFuncMsg
*
pSqlFuncMsg
=
&
pQuery
->
p
Expr1
[
i
].
base
;
SQLFunctionCtx
*
pCtx
=
&
pRuntimeEnv
->
pCtx
[
i
];
SQLFunctionCtx
*
pCtx
=
&
pRuntimeEnv
->
pCtx
[
i
];
SColIndex
*
pIndex
=
&
pSqlFuncMsg
->
colInfo
;
SColIndex
*
pIndex
=
&
pSqlFuncMsg
->
colInfo
;
...
@@ -1649,13 +1649,13 @@ static int32_t setupQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv, int16_t order
...
@@ -1649,13 +1649,13 @@ static int32_t setupQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv, int16_t order
assert
(
isValidDataType
(
pCtx
->
inputType
));
assert
(
isValidDataType
(
pCtx
->
inputType
));
pCtx
->
ptsOutputBuf
=
NULL
;
pCtx
->
ptsOutputBuf
=
NULL
;
pCtx
->
outputBytes
=
pQuery
->
p
SelectExpr
[
i
].
bytes
;
pCtx
->
outputBytes
=
pQuery
->
p
Expr1
[
i
].
bytes
;
pCtx
->
outputType
=
pQuery
->
p
SelectExpr
[
i
].
type
;
pCtx
->
outputType
=
pQuery
->
p
Expr1
[
i
].
type
;
pCtx
->
order
=
pQuery
->
order
.
order
;
pCtx
->
order
=
pQuery
->
order
.
order
;
pCtx
->
functionId
=
pSqlFuncMsg
->
functionId
;
pCtx
->
functionId
=
pSqlFuncMsg
->
functionId
;
pCtx
->
stableQuery
=
pRuntimeEnv
->
stableQuery
;
pCtx
->
stableQuery
=
pRuntimeEnv
->
stableQuery
;
pCtx
->
interBufBytes
=
pQuery
->
p
SelectExpr
[
i
].
interBytes
;
pCtx
->
interBufBytes
=
pQuery
->
p
Expr1
[
i
].
interBytes
;
pCtx
->
numOfParams
=
pSqlFuncMsg
->
numOfParams
;
pCtx
->
numOfParams
=
pSqlFuncMsg
->
numOfParams
;
for
(
int32_t
j
=
0
;
j
<
pCtx
->
numOfParams
;
++
j
)
{
for
(
int32_t
j
=
0
;
j
<
pCtx
->
numOfParams
;
++
j
)
{
...
@@ -1672,7 +1672,7 @@ static int32_t setupQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv, int16_t order
...
@@ -1672,7 +1672,7 @@ static int32_t setupQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv, int16_t order
int32_t
functionId
=
pCtx
->
functionId
;
int32_t
functionId
=
pCtx
->
functionId
;
if
(
functionId
==
TSDB_FUNC_TOP
||
functionId
==
TSDB_FUNC_BOTTOM
||
functionId
==
TSDB_FUNC_DIFF
)
{
if
(
functionId
==
TSDB_FUNC_TOP
||
functionId
==
TSDB_FUNC_BOTTOM
||
functionId
==
TSDB_FUNC_DIFF
)
{
int32_t
f
=
pQuery
->
p
SelectExpr
[
0
].
base
.
functionId
;
int32_t
f
=
pQuery
->
p
Expr1
[
0
].
base
.
functionId
;
assert
(
f
==
TSDB_FUNC_TS
||
f
==
TSDB_FUNC_TS_DUMMY
);
assert
(
f
==
TSDB_FUNC_TS
||
f
==
TSDB_FUNC_TS_DUMMY
);
pCtx
->
param
[
2
].
i64Key
=
order
;
pCtx
->
param
[
2
].
i64Key
=
order
;
...
@@ -1685,7 +1685,7 @@ static int32_t setupQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv, int16_t order
...
@@ -1685,7 +1685,7 @@ static int32_t setupQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv, int16_t order
if
(
i
>
0
)
{
if
(
i
>
0
)
{
pRuntimeEnv
->
offset
[
i
]
=
pRuntimeEnv
->
offset
[
i
-
1
]
+
pRuntimeEnv
->
pCtx
[
i
-
1
].
outputBytes
;
pRuntimeEnv
->
offset
[
i
]
=
pRuntimeEnv
->
offset
[
i
-
1
]
+
pRuntimeEnv
->
pCtx
[
i
-
1
].
outputBytes
;
pRuntimeEnv
->
rowCellInfoOffset
[
i
]
=
pRuntimeEnv
->
rowCellInfoOffset
[
i
-
1
]
+
sizeof
(
SResultRowCellInfo
)
+
pQuery
->
p
SelectExpr
[
i
-
1
].
interBytes
;
pRuntimeEnv
->
rowCellInfoOffset
[
i
]
=
pRuntimeEnv
->
rowCellInfoOffset
[
i
-
1
]
+
sizeof
(
SResultRowCellInfo
)
+
pQuery
->
p
Expr1
[
i
-
1
].
interBytes
;
}
}
}
}
...
@@ -1779,7 +1779,7 @@ static bool isFixedOutputQuery(SQueryRuntimeEnv* pRuntimeEnv) {
...
@@ -1779,7 +1779,7 @@ static bool isFixedOutputQuery(SQueryRuntimeEnv* pRuntimeEnv) {
}
}
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
SSqlFuncMsg
*
pExprMsg
=
&
pQuery
->
p
SelectExpr
[
i
].
base
;
SSqlFuncMsg
*
pExprMsg
=
&
pQuery
->
p
Expr1
[
i
].
base
;
// ignore the ts_comp function
// ignore the ts_comp function
if
(
i
==
0
&&
pExprMsg
->
functionId
==
TSDB_FUNC_PRJ
&&
pExprMsg
->
numOfParams
==
1
&&
if
(
i
==
0
&&
pExprMsg
->
functionId
==
TSDB_FUNC_PRJ
&&
pExprMsg
->
numOfParams
==
1
&&
...
@@ -1802,7 +1802,7 @@ static bool isFixedOutputQuery(SQueryRuntimeEnv* pRuntimeEnv) {
...
@@ -1802,7 +1802,7 @@ static bool isFixedOutputQuery(SQueryRuntimeEnv* pRuntimeEnv) {
// todo refactor with isLastRowQuery
// todo refactor with isLastRowQuery
static
bool
isPointInterpoQuery
(
SQuery
*
pQuery
)
{
static
bool
isPointInterpoQuery
(
SQuery
*
pQuery
)
{
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
int32_t
functionID
=
pQuery
->
p
SelectExpr
[
i
].
base
.
functionId
;
int32_t
functionID
=
pQuery
->
p
Expr1
[
i
].
base
.
functionId
;
if
(
functionID
==
TSDB_FUNC_INTERP
)
{
if
(
functionID
==
TSDB_FUNC_INTERP
)
{
return
true
;
return
true
;
}
}
...
@@ -1814,7 +1814,7 @@ static bool isPointInterpoQuery(SQuery *pQuery) {
...
@@ -1814,7 +1814,7 @@ static bool isPointInterpoQuery(SQuery *pQuery) {
// TODO REFACTOR:MERGE WITH CLIENT-SIDE FUNCTION
// TODO REFACTOR:MERGE WITH CLIENT-SIDE FUNCTION
static
bool
isSumAvgRateQuery
(
SQuery
*
pQuery
)
{
static
bool
isSumAvgRateQuery
(
SQuery
*
pQuery
)
{
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
int32_t
functionId
=
pQuery
->
p
SelectExpr
[
i
].
base
.
functionId
;
int32_t
functionId
=
pQuery
->
p
Expr1
[
i
].
base
.
functionId
;
if
(
functionId
==
TSDB_FUNC_TS
)
{
if
(
functionId
==
TSDB_FUNC_TS
)
{
continue
;
continue
;
}
}
...
@@ -1830,7 +1830,7 @@ static bool isSumAvgRateQuery(SQuery *pQuery) {
...
@@ -1830,7 +1830,7 @@ static bool isSumAvgRateQuery(SQuery *pQuery) {
static
bool
isFirstLastRowQuery
(
SQuery
*
pQuery
)
{
static
bool
isFirstLastRowQuery
(
SQuery
*
pQuery
)
{
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
int32_t
functionID
=
pQuery
->
p
SelectExpr
[
i
].
base
.
functionId
;
int32_t
functionID
=
pQuery
->
p
Expr1
[
i
].
base
.
functionId
;
if
(
functionID
==
TSDB_FUNC_LAST_ROW
)
{
if
(
functionID
==
TSDB_FUNC_LAST_ROW
)
{
return
true
;
return
true
;
}
}
...
@@ -1841,7 +1841,7 @@ static bool isFirstLastRowQuery(SQuery *pQuery) {
...
@@ -1841,7 +1841,7 @@ static bool isFirstLastRowQuery(SQuery *pQuery) {
static
bool
needReverseScan
(
SQuery
*
pQuery
)
{
static
bool
needReverseScan
(
SQuery
*
pQuery
)
{
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
int32_t
functionId
=
pQuery
->
p
SelectExpr
[
i
].
base
.
functionId
;
int32_t
functionId
=
pQuery
->
p
Expr1
[
i
].
base
.
functionId
;
if
(
functionId
==
TSDB_FUNC_TS
||
functionId
==
TSDB_FUNC_TS_DUMMY
||
functionId
==
TSDB_FUNC_TAG
)
{
if
(
functionId
==
TSDB_FUNC_TS
||
functionId
==
TSDB_FUNC_TS_DUMMY
||
functionId
==
TSDB_FUNC_TAG
)
{
continue
;
continue
;
}
}
...
@@ -1852,7 +1852,7 @@ static bool needReverseScan(SQuery *pQuery) {
...
@@ -1852,7 +1852,7 @@ static bool needReverseScan(SQuery *pQuery) {
if
(
functionId
==
TSDB_FUNC_LAST
||
functionId
==
TSDB_FUNC_LAST_DST
)
{
if
(
functionId
==
TSDB_FUNC_LAST
||
functionId
==
TSDB_FUNC_LAST_DST
)
{
// the scan order to acquire the last result of the specified column
// the scan order to acquire the last result of the specified column
int32_t
order
=
(
int32_t
)
pQuery
->
p
SelectExpr
[
i
].
base
.
arg
->
argValue
.
i64
;
int32_t
order
=
(
int32_t
)
pQuery
->
p
Expr1
[
i
].
base
.
arg
->
argValue
.
i64
;
if
(
order
!=
pQuery
->
order
.
order
)
{
if
(
order
!=
pQuery
->
order
.
order
)
{
return
true
;
return
true
;
}
}
...
@@ -1868,7 +1868,7 @@ static bool needReverseScan(SQuery *pQuery) {
...
@@ -1868,7 +1868,7 @@ static bool needReverseScan(SQuery *pQuery) {
*/
*/
static
bool
onlyQueryTags
(
SQuery
*
pQuery
)
{
static
bool
onlyQueryTags
(
SQuery
*
pQuery
)
{
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
SExprInfo
*
pExprInfo
=
&
pQuery
->
p
SelectExpr
[
i
];
SExprInfo
*
pExprInfo
=
&
pQuery
->
p
Expr1
[
i
];
int32_t
functionId
=
pExprInfo
->
base
.
functionId
;
int32_t
functionId
=
pExprInfo
->
base
.
functionId
;
...
@@ -1911,7 +1911,7 @@ static void setScanLimitationByResultBuffer(SQuery *pQuery) {
...
@@ -1911,7 +1911,7 @@ static void setScanLimitationByResultBuffer(SQuery *pQuery) {
}
else
{
}
else
{
bool
hasMultioutput
=
false
;
bool
hasMultioutput
=
false
;
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
SSqlFuncMsg
*
pExprMsg
=
&
pQuery
->
p
SelectExpr
[
i
].
base
;
SSqlFuncMsg
*
pExprMsg
=
&
pQuery
->
p
Expr1
[
i
].
base
;
if
(
pExprMsg
->
functionId
==
TSDB_FUNC_TS
||
pExprMsg
->
functionId
==
TSDB_FUNC_TS_DUMMY
)
{
if
(
pExprMsg
->
functionId
==
TSDB_FUNC_TS
||
pExprMsg
->
functionId
==
TSDB_FUNC_TS_DUMMY
)
{
continue
;
continue
;
}
}
...
@@ -1945,7 +1945,7 @@ bool colIdCheck(SQuery *pQuery) {
...
@@ -1945,7 +1945,7 @@ bool colIdCheck(SQuery *pQuery) {
// the scan order is not matter
// the scan order is not matter
static
bool
onlyOneQueryType
(
SQuery
*
pQuery
,
int32_t
functId
,
int32_t
functIdDst
)
{
static
bool
onlyOneQueryType
(
SQuery
*
pQuery
,
int32_t
functId
,
int32_t
functIdDst
)
{
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
int32_t
functionId
=
pQuery
->
p
SelectExpr
[
i
].
base
.
functionId
;
int32_t
functionId
=
pQuery
->
p
Expr1
[
i
].
base
.
functionId
;
if
(
functionId
==
TSDB_FUNC_TS
||
functionId
==
TSDB_FUNC_TS_DUMMY
||
functionId
==
TSDB_FUNC_TAG
||
if
(
functionId
==
TSDB_FUNC_TS
||
functionId
==
TSDB_FUNC_TS_DUMMY
||
functionId
==
TSDB_FUNC_TAG
||
functionId
==
TSDB_FUNC_TAG_DUMMY
)
{
functionId
==
TSDB_FUNC_TAG_DUMMY
)
{
...
@@ -2175,7 +2175,7 @@ static bool needToLoadDataBlock(SQueryRuntimeEnv* pRuntimeEnv, SDataStatis *pDat
...
@@ -2175,7 +2175,7 @@ static bool needToLoadDataBlock(SQueryRuntimeEnv* pRuntimeEnv, SDataStatis *pDat
if
(
pRuntimeEnv
->
topBotQuery
)
{
if
(
pRuntimeEnv
->
topBotQuery
)
{
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
int32_t
functionId
=
pQuery
->
p
SelectExpr
[
i
].
base
.
functionId
;
int32_t
functionId
=
pQuery
->
p
Expr1
[
i
].
base
.
functionId
;
if
(
functionId
==
TSDB_FUNC_TOP
||
functionId
==
TSDB_FUNC_BOTTOM
)
{
if
(
functionId
==
TSDB_FUNC_TOP
||
functionId
==
TSDB_FUNC_BOTTOM
)
{
return
topbot_datablock_filter
(
&
pCtx
[
i
],
functionId
,
(
char
*
)
&
pDataStatis
[
i
].
min
,
(
char
*
)
&
pDataStatis
[
i
].
max
);
return
topbot_datablock_filter
(
&
pCtx
[
i
],
functionId
,
(
char
*
)
&
pDataStatis
[
i
].
min
,
(
char
*
)
&
pDataStatis
[
i
].
max
);
}
}
...
@@ -2266,7 +2266,7 @@ int32_t loadDataBlockOnDemand(SQueryRuntimeEnv *pRuntimeEnv, SWindowResInfo * pW
...
@@ -2266,7 +2266,7 @@ int32_t loadDataBlockOnDemand(SQueryRuntimeEnv *pRuntimeEnv, SWindowResInfo * pW
}
}
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
SSqlFuncMsg
*
pSqlFunc
=
&
pQuery
->
p
SelectExpr
[
i
].
base
;
SSqlFuncMsg
*
pSqlFunc
=
&
pQuery
->
p
Expr1
[
i
].
base
;
int32_t
functionId
=
pSqlFunc
->
functionId
;
int32_t
functionId
=
pSqlFunc
->
functionId
;
int32_t
colId
=
pSqlFunc
->
colInfo
.
colId
;
int32_t
colId
=
pSqlFunc
->
colInfo
.
colId
;
...
@@ -2390,7 +2390,7 @@ static void ensureOutputBufferSimple(SQueryRuntimeEnv* pRuntimeEnv, int32_t capa
...
@@ -2390,7 +2390,7 @@ static void ensureOutputBufferSimple(SQueryRuntimeEnv* pRuntimeEnv, int32_t capa
}
}
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
int32_t
bytes
=
pQuery
->
p
SelectExpr
[
i
].
bytes
;
int32_t
bytes
=
pQuery
->
p
Expr1
[
i
].
bytes
;
assert
(
bytes
>
0
&&
capacity
>
0
);
assert
(
bytes
>
0
&&
capacity
>
0
);
char
*
tmp
=
realloc
(
pQuery
->
sdata
[
i
],
bytes
*
capacity
+
sizeof
(
tFilePage
));
char
*
tmp
=
realloc
(
pQuery
->
sdata
[
i
],
bytes
*
capacity
+
sizeof
(
tFilePage
));
...
@@ -2421,7 +2421,7 @@ static void ensureOutputBuffer(SQueryRuntimeEnv* pRuntimeEnv, SDataBlockInfo* pB
...
@@ -2421,7 +2421,7 @@ static void ensureOutputBuffer(SQueryRuntimeEnv* pRuntimeEnv, SDataBlockInfo* pB
int32_t
newSize
=
(
int32_t
)(
pRec
->
capacity
+
(
pBlockInfo
->
rows
-
remain
));
int32_t
newSize
=
(
int32_t
)(
pRec
->
capacity
+
(
pBlockInfo
->
rows
-
remain
));
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
int32_t
bytes
=
pQuery
->
p
SelectExpr
[
i
].
bytes
;
int32_t
bytes
=
pQuery
->
p
Expr1
[
i
].
bytes
;
assert
(
bytes
>
0
&&
newSize
>
0
);
assert
(
bytes
>
0
&&
newSize
>
0
);
char
*
tmp
=
realloc
(
pQuery
->
sdata
[
i
],
bytes
*
newSize
+
sizeof
(
tFilePage
));
char
*
tmp
=
realloc
(
pQuery
->
sdata
[
i
],
bytes
*
newSize
+
sizeof
(
tFilePage
));
...
@@ -2435,7 +2435,7 @@ static void ensureOutputBuffer(SQueryRuntimeEnv* pRuntimeEnv, SDataBlockInfo* pB
...
@@ -2435,7 +2435,7 @@ static void ensureOutputBuffer(SQueryRuntimeEnv* pRuntimeEnv, SDataBlockInfo* pB
// set the pCtx output buffer position
// set the pCtx output buffer position
pRuntimeEnv
->
pCtx
[
i
].
aOutputBuf
=
pQuery
->
sdata
[
i
]
->
data
+
pRec
->
rows
*
bytes
;
pRuntimeEnv
->
pCtx
[
i
].
aOutputBuf
=
pQuery
->
sdata
[
i
]
->
data
+
pRec
->
rows
*
bytes
;
int32_t
functionId
=
pQuery
->
p
SelectExpr
[
i
].
base
.
functionId
;
int32_t
functionId
=
pQuery
->
p
Expr1
[
i
].
base
.
functionId
;
if
(
functionId
==
TSDB_FUNC_TOP
||
functionId
==
TSDB_FUNC_BOTTOM
||
functionId
==
TSDB_FUNC_DIFF
)
{
if
(
functionId
==
TSDB_FUNC_TOP
||
functionId
==
TSDB_FUNC_BOTTOM
||
functionId
==
TSDB_FUNC_DIFF
)
{
pRuntimeEnv
->
pCtx
[
i
].
ptsOutputBuf
=
pRuntimeEnv
->
pCtx
[
0
].
aOutputBuf
;
pRuntimeEnv
->
pCtx
[
i
].
ptsOutputBuf
=
pRuntimeEnv
->
pCtx
[
0
].
aOutputBuf
;
}
}
...
@@ -2599,7 +2599,7 @@ void setTagVal(SQueryRuntimeEnv *pRuntimeEnv, void *pTable, void *tsdb) {
...
@@ -2599,7 +2599,7 @@ void setTagVal(SQueryRuntimeEnv *pRuntimeEnv, void *pTable, void *tsdb) {
SQuery
*
pQuery
=
pRuntimeEnv
->
pQuery
;
SQuery
*
pQuery
=
pRuntimeEnv
->
pQuery
;
SQInfo
*
pQInfo
=
GET_QINFO_ADDR
(
pRuntimeEnv
);
SQInfo
*
pQInfo
=
GET_QINFO_ADDR
(
pRuntimeEnv
);
SExprInfo
*
pExprInfo
=
&
pQuery
->
p
SelectExpr
[
0
];
SExprInfo
*
pExprInfo
=
&
pQuery
->
p
Expr1
[
0
];
if
(
pQuery
->
numOfOutput
==
1
&&
pExprInfo
->
base
.
functionId
==
TSDB_FUNC_TS_COMP
)
{
if
(
pQuery
->
numOfOutput
==
1
&&
pExprInfo
->
base
.
functionId
==
TSDB_FUNC_TS_COMP
)
{
assert
(
pExprInfo
->
base
.
numOfParams
==
1
);
assert
(
pExprInfo
->
base
.
numOfParams
==
1
);
...
@@ -2610,7 +2610,7 @@ void setTagVal(SQueryRuntimeEnv *pRuntimeEnv, void *pTable, void *tsdb) {
...
@@ -2610,7 +2610,7 @@ void setTagVal(SQueryRuntimeEnv *pRuntimeEnv, void *pTable, void *tsdb) {
}
else
{
}
else
{
// set tag value, by which the results are aggregated.
// set tag value, by which the results are aggregated.
for
(
int32_t
idx
=
0
;
idx
<
pQuery
->
numOfOutput
;
++
idx
)
{
for
(
int32_t
idx
=
0
;
idx
<
pQuery
->
numOfOutput
;
++
idx
)
{
SExprInfo
*
pLocalExprInfo
=
&
pQuery
->
p
SelectExpr
[
idx
];
SExprInfo
*
pLocalExprInfo
=
&
pQuery
->
p
Expr1
[
idx
];
// ts_comp column required the tag value for join filter
// ts_comp column required the tag value for join filter
if
(
!
TSDB_COL_IS_TAG
(
pLocalExprInfo
->
base
.
colInfo
.
flag
))
{
if
(
!
TSDB_COL_IS_TAG
(
pLocalExprInfo
->
base
.
colInfo
.
flag
))
{
...
@@ -2652,7 +2652,7 @@ static void doMerge(SQueryRuntimeEnv *pRuntimeEnv, int64_t timestamp, SResultRow
...
@@ -2652,7 +2652,7 @@ static void doMerge(SQueryRuntimeEnv *pRuntimeEnv, int64_t timestamp, SResultRow
tFilePage
*
page
=
getResBufPage
(
pRuntimeEnv
->
pResultBuf
,
pWindowRes
->
pageId
);
tFilePage
*
page
=
getResBufPage
(
pRuntimeEnv
->
pResultBuf
,
pWindowRes
->
pageId
);
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
int32_t
functionId
=
pQuery
->
p
SelectExpr
[
i
].
base
.
functionId
;
int32_t
functionId
=
pQuery
->
p
Expr1
[
i
].
base
.
functionId
;
if
(
!
mergeFlag
)
{
if
(
!
mergeFlag
)
{
pCtx
[
i
].
aOutputBuf
=
pCtx
[
i
].
aOutputBuf
+
pCtx
[
i
].
outputBytes
;
pCtx
[
i
].
aOutputBuf
=
pCtx
[
i
].
aOutputBuf
+
pCtx
[
i
].
outputBytes
;
pCtx
[
i
].
currentStage
=
FIRST_STAGE_MERGE
;
pCtx
[
i
].
currentStage
=
FIRST_STAGE_MERGE
;
...
@@ -2680,7 +2680,7 @@ static void doMerge(SQueryRuntimeEnv *pRuntimeEnv, int64_t timestamp, SResultRow
...
@@ -2680,7 +2680,7 @@ static void doMerge(SQueryRuntimeEnv *pRuntimeEnv, int64_t timestamp, SResultRow
}
}
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
int32_t
functionId
=
pQuery
->
p
SelectExpr
[
i
].
base
.
functionId
;
int32_t
functionId
=
pQuery
->
p
Expr1
[
i
].
base
.
functionId
;
if
(
functionId
==
TSDB_FUNC_TAG_DUMMY
)
{
if
(
functionId
==
TSDB_FUNC_TAG_DUMMY
)
{
continue
;
continue
;
}
}
...
@@ -2766,25 +2766,25 @@ void UNUSED_FUNC displayInterResult(tFilePage **pdata, SQueryRuntimeEnv* pRuntim
...
@@ -2766,25 +2766,25 @@ void UNUSED_FUNC displayInterResult(tFilePage **pdata, SQueryRuntimeEnv* pRuntim
for
(
int32_t
j
=
0
;
j
<
numOfRows
;
++
j
)
{
for
(
int32_t
j
=
0
;
j
<
numOfRows
;
++
j
)
{
for
(
int32_t
i
=
0
;
i
<
numOfCols
;
++
i
)
{
for
(
int32_t
i
=
0
;
i
<
numOfCols
;
++
i
)
{
switch
(
pQuery
->
p
SelectExpr
[
i
].
type
)
{
switch
(
pQuery
->
p
Expr1
[
i
].
type
)
{
case
TSDB_DATA_TYPE_BINARY
:
{
case
TSDB_DATA_TYPE_BINARY
:
{
int32_t
type
=
pQuery
->
p
SelectExpr
[
i
].
type
;
int32_t
type
=
pQuery
->
p
Expr1
[
i
].
type
;
printBinaryData
(
pQuery
->
p
SelectExpr
[
i
].
base
.
functionId
,
pdata
[
i
]
->
data
+
pQuery
->
pSelectExpr
[
i
].
bytes
*
j
,
printBinaryData
(
pQuery
->
p
Expr1
[
i
].
base
.
functionId
,
pdata
[
i
]
->
data
+
pQuery
->
pExpr1
[
i
].
bytes
*
j
,
type
);
type
);
break
;
break
;
}
}
case
TSDB_DATA_TYPE_TIMESTAMP
:
case
TSDB_DATA_TYPE_TIMESTAMP
:
case
TSDB_DATA_TYPE_BIGINT
:
case
TSDB_DATA_TYPE_BIGINT
:
printf
(
"%"
PRId64
"
\t
"
,
*
(
int64_t
*
)(
pdata
[
i
]
->
data
+
pQuery
->
p
SelectExpr
[
i
].
bytes
*
j
));
printf
(
"%"
PRId64
"
\t
"
,
*
(
int64_t
*
)(
pdata
[
i
]
->
data
+
pQuery
->
p
Expr1
[
i
].
bytes
*
j
));
break
;
break
;
case
TSDB_DATA_TYPE_INT
:
case
TSDB_DATA_TYPE_INT
:
printf
(
"%d
\t
"
,
*
(
int32_t
*
)(
pdata
[
i
]
->
data
+
pQuery
->
p
SelectExpr
[
i
].
bytes
*
j
));
printf
(
"%d
\t
"
,
*
(
int32_t
*
)(
pdata
[
i
]
->
data
+
pQuery
->
p
Expr1
[
i
].
bytes
*
j
));
break
;
break
;
case
TSDB_DATA_TYPE_FLOAT
:
case
TSDB_DATA_TYPE_FLOAT
:
printf
(
"%f
\t
"
,
*
(
float
*
)(
pdata
[
i
]
->
data
+
pQuery
->
p
SelectExpr
[
i
].
bytes
*
j
));
printf
(
"%f
\t
"
,
*
(
float
*
)(
pdata
[
i
]
->
data
+
pQuery
->
p
Expr1
[
i
].
bytes
*
j
));
break
;
break
;
case
TSDB_DATA_TYPE_DOUBLE
:
case
TSDB_DATA_TYPE_DOUBLE
:
printf
(
"%lf
\t
"
,
*
(
double
*
)(
pdata
[
i
]
->
data
+
pQuery
->
p
SelectExpr
[
i
].
bytes
*
j
));
printf
(
"%lf
\t
"
,
*
(
double
*
)(
pdata
[
i
]
->
data
+
pQuery
->
p
Expr1
[
i
].
bytes
*
j
));
break
;
break
;
}
}
}
}
...
@@ -2951,7 +2951,7 @@ int64_t getNumOfResultWindowRes(SQueryRuntimeEnv* pRuntimeEnv, SResultRow *pResu
...
@@ -2951,7 +2951,7 @@ int64_t getNumOfResultWindowRes(SQueryRuntimeEnv* pRuntimeEnv, SResultRow *pResu
SQuery
*
pQuery
=
pRuntimeEnv
->
pQuery
;
SQuery
*
pQuery
=
pRuntimeEnv
->
pQuery
;
for
(
int32_t
j
=
0
;
j
<
pQuery
->
numOfOutput
;
++
j
)
{
for
(
int32_t
j
=
0
;
j
<
pQuery
->
numOfOutput
;
++
j
)
{
int32_t
functionId
=
pQuery
->
p
SelectExpr
[
j
].
base
.
functionId
;
int32_t
functionId
=
pQuery
->
p
Expr1
[
j
].
base
.
functionId
;
/*
/*
* ts, tag, tagprj function can not decide the output number of current query
* ts, tag, tagprj function can not decide the output number of current query
...
@@ -3236,7 +3236,7 @@ static void disableFuncInReverseScanImpl(SQueryRuntimeEnv* pRuntimeEnv, SWindowR
...
@@ -3236,7 +3236,7 @@ static void disableFuncInReverseScanImpl(SQueryRuntimeEnv* pRuntimeEnv, SWindowR
// open/close the specified query for each group result
// open/close the specified query for each group result
for
(
int32_t
j
=
0
;
j
<
pQuery
->
numOfOutput
;
++
j
)
{
for
(
int32_t
j
=
0
;
j
<
pQuery
->
numOfOutput
;
++
j
)
{
int32_t
functId
=
pQuery
->
p
SelectExpr
[
j
].
base
.
functionId
;
int32_t
functId
=
pQuery
->
p
Expr1
[
j
].
base
.
functionId
;
SResultRowCellInfo
*
pInfo
=
getResultCell
(
pRuntimeEnv
,
pRow
,
j
);
SResultRowCellInfo
*
pInfo
=
getResultCell
(
pRuntimeEnv
,
pRow
,
j
);
if
(((
functId
==
TSDB_FUNC_FIRST
||
functId
==
TSDB_FUNC_FIRST_DST
)
&&
order
==
TSDB_ORDER_ASC
)
||
if
(((
functId
==
TSDB_FUNC_FIRST
||
functId
==
TSDB_FUNC_FIRST_DST
)
&&
order
==
TSDB_ORDER_ASC
)
||
...
@@ -3260,7 +3260,7 @@ void disableFuncInReverseScan(SQInfo *pQInfo) {
...
@@ -3260,7 +3260,7 @@ void disableFuncInReverseScan(SQInfo *pQInfo) {
disableFuncInReverseScanImpl
(
pRuntimeEnv
,
pWindowResInfo
,
order
);
disableFuncInReverseScanImpl
(
pRuntimeEnv
,
pWindowResInfo
,
order
);
}
else
{
// for simple result of table query,
}
else
{
// for simple result of table query,
for
(
int32_t
j
=
0
;
j
<
pQuery
->
numOfOutput
;
++
j
)
{
// todo refactor
for
(
int32_t
j
=
0
;
j
<
pQuery
->
numOfOutput
;
++
j
)
{
// todo refactor
int32_t
functId
=
pQuery
->
p
SelectExpr
[
j
].
base
.
functionId
;
int32_t
functId
=
pQuery
->
p
Expr1
[
j
].
base
.
functionId
;
SQLFunctionCtx
*
pCtx
=
&
pRuntimeEnv
->
pCtx
[
j
];
SQLFunctionCtx
*
pCtx
=
&
pRuntimeEnv
->
pCtx
[
j
];
if
(
pCtx
->
resultInfo
==
NULL
)
{
if
(
pCtx
->
resultInfo
==
NULL
)
{
...
@@ -3331,12 +3331,12 @@ void resetCtxOutputBuf(SQueryRuntimeEnv *pRuntimeEnv) {
...
@@ -3331,12 +3331,12 @@ void resetCtxOutputBuf(SQueryRuntimeEnv *pRuntimeEnv) {
pCtx
->
resultInfo
=
pCellInfo
;
pCtx
->
resultInfo
=
pCellInfo
;
// set the timestamp output buffer for top/bottom/diff query
// set the timestamp output buffer for top/bottom/diff query
int32_t
functionId
=
pQuery
->
p
SelectExpr
[
i
].
base
.
functionId
;
int32_t
functionId
=
pQuery
->
p
Expr1
[
i
].
base
.
functionId
;
if
(
functionId
==
TSDB_FUNC_TOP
||
functionId
==
TSDB_FUNC_BOTTOM
||
functionId
==
TSDB_FUNC_DIFF
)
{
if
(
functionId
==
TSDB_FUNC_TOP
||
functionId
==
TSDB_FUNC_BOTTOM
||
functionId
==
TSDB_FUNC_DIFF
)
{
pCtx
->
ptsOutputBuf
=
pRuntimeEnv
->
pCtx
[
0
].
aOutputBuf
;
pCtx
->
ptsOutputBuf
=
pRuntimeEnv
->
pCtx
[
0
].
aOutputBuf
;
}
}
memset
(
pQuery
->
sdata
[
i
]
->
data
,
0
,
(
size_t
)(
pQuery
->
p
SelectExpr
[
i
].
bytes
*
pQuery
->
rec
.
capacity
));
memset
(
pQuery
->
sdata
[
i
]
->
data
,
0
,
(
size_t
)(
pQuery
->
p
Expr1
[
i
].
bytes
*
pQuery
->
rec
.
capacity
));
}
}
initCtxOutputBuf
(
pRuntimeEnv
);
initCtxOutputBuf
(
pRuntimeEnv
);
...
@@ -3347,7 +3347,7 @@ void forwardCtxOutputBuf(SQueryRuntimeEnv *pRuntimeEnv, int64_t output) {
...
@@ -3347,7 +3347,7 @@ void forwardCtxOutputBuf(SQueryRuntimeEnv *pRuntimeEnv, int64_t output) {
// reset the execution contexts
// reset the execution contexts
for
(
int32_t
j
=
0
;
j
<
pQuery
->
numOfOutput
;
++
j
)
{
for
(
int32_t
j
=
0
;
j
<
pQuery
->
numOfOutput
;
++
j
)
{
int32_t
functionId
=
pQuery
->
p
SelectExpr
[
j
].
base
.
functionId
;
int32_t
functionId
=
pQuery
->
p
Expr1
[
j
].
base
.
functionId
;
assert
(
functionId
!=
TSDB_FUNC_DIFF
);
assert
(
functionId
!=
TSDB_FUNC_DIFF
);
// set next output position
// set next output position
...
@@ -3374,7 +3374,7 @@ void initCtxOutputBuf(SQueryRuntimeEnv *pRuntimeEnv) {
...
@@ -3374,7 +3374,7 @@ void initCtxOutputBuf(SQueryRuntimeEnv *pRuntimeEnv) {
SQuery
*
pQuery
=
pRuntimeEnv
->
pQuery
;
SQuery
*
pQuery
=
pRuntimeEnv
->
pQuery
;
for
(
int32_t
j
=
0
;
j
<
pQuery
->
numOfOutput
;
++
j
)
{
for
(
int32_t
j
=
0
;
j
<
pQuery
->
numOfOutput
;
++
j
)
{
int32_t
functionId
=
pQuery
->
p
SelectExpr
[
j
].
base
.
functionId
;
int32_t
functionId
=
pQuery
->
p
Expr1
[
j
].
base
.
functionId
;
pRuntimeEnv
->
pCtx
[
j
].
currentStage
=
0
;
pRuntimeEnv
->
pCtx
[
j
].
currentStage
=
0
;
SResultRowCellInfo
*
pResInfo
=
GET_RES_INFO
(
&
pRuntimeEnv
->
pCtx
[
j
]);
SResultRowCellInfo
*
pResInfo
=
GET_RES_INFO
(
&
pRuntimeEnv
->
pCtx
[
j
]);
...
@@ -3412,7 +3412,7 @@ void skipResults(SQueryRuntimeEnv *pRuntimeEnv) {
...
@@ -3412,7 +3412,7 @@ void skipResults(SQueryRuntimeEnv *pRuntimeEnv) {
0
,
pQuery
->
rec
.
rows
);
0
,
pQuery
->
rec
.
rows
);
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
int32_t
functionId
=
pQuery
->
p
SelectExpr
[
i
].
base
.
functionId
;
int32_t
functionId
=
pQuery
->
p
Expr1
[
i
].
base
.
functionId
;
int32_t
bytes
=
pRuntimeEnv
->
pCtx
[
i
].
outputBytes
;
int32_t
bytes
=
pRuntimeEnv
->
pCtx
[
i
].
outputBytes
;
memmove
(
pQuery
->
sdata
[
i
]
->
data
,
(
char
*
)
pQuery
->
sdata
[
i
]
->
data
+
bytes
*
numOfSkip
,
(
size_t
)(
pQuery
->
rec
.
rows
*
bytes
));
memmove
(
pQuery
->
sdata
[
i
]
->
data
,
(
char
*
)
pQuery
->
sdata
[
i
]
->
data
+
bytes
*
numOfSkip
,
(
size_t
)(
pQuery
->
rec
.
rows
*
bytes
));
...
@@ -3454,7 +3454,7 @@ bool needScanDataBlocksAgain(SQueryRuntimeEnv *pRuntimeEnv) {
...
@@ -3454,7 +3454,7 @@ bool needScanDataBlocksAgain(SQueryRuntimeEnv *pRuntimeEnv) {
setResultOutputBuf
(
pRuntimeEnv
,
pResult
);
setResultOutputBuf
(
pRuntimeEnv
,
pResult
);
for
(
int32_t
j
=
0
;
j
<
pQuery
->
numOfOutput
;
++
j
)
{
for
(
int32_t
j
=
0
;
j
<
pQuery
->
numOfOutput
;
++
j
)
{
int16_t
functId
=
pQuery
->
p
SelectExpr
[
j
].
base
.
functionId
;
int16_t
functId
=
pQuery
->
p
Expr1
[
j
].
base
.
functionId
;
if
(
functId
==
TSDB_FUNC_TS
)
{
if
(
functId
==
TSDB_FUNC_TS
)
{
continue
;
continue
;
}
}
...
@@ -3467,7 +3467,7 @@ bool needScanDataBlocksAgain(SQueryRuntimeEnv *pRuntimeEnv) {
...
@@ -3467,7 +3467,7 @@ bool needScanDataBlocksAgain(SQueryRuntimeEnv *pRuntimeEnv) {
}
}
}
else
{
}
else
{
for
(
int32_t
j
=
0
;
j
<
pQuery
->
numOfOutput
;
++
j
)
{
for
(
int32_t
j
=
0
;
j
<
pQuery
->
numOfOutput
;
++
j
)
{
int16_t
functId
=
pQuery
->
p
SelectExpr
[
j
].
base
.
functionId
;
int16_t
functId
=
pQuery
->
p
Expr1
[
j
].
base
.
functionId
;
if
(
functId
==
TSDB_FUNC_TS
)
{
if
(
functId
==
TSDB_FUNC_TS
)
{
continue
;
continue
;
}
}
...
@@ -3680,7 +3680,7 @@ void finalizeQueryResult(SQueryRuntimeEnv *pRuntimeEnv) {
...
@@ -3680,7 +3680,7 @@ void finalizeQueryResult(SQueryRuntimeEnv *pRuntimeEnv) {
setResultOutputBuf
(
pRuntimeEnv
,
buf
);
setResultOutputBuf
(
pRuntimeEnv
,
buf
);
for
(
int32_t
j
=
0
;
j
<
pQuery
->
numOfOutput
;
++
j
)
{
for
(
int32_t
j
=
0
;
j
<
pQuery
->
numOfOutput
;
++
j
)
{
aAggs
[
pQuery
->
p
SelectExpr
[
j
].
base
.
functionId
].
xFinalize
(
&
pRuntimeEnv
->
pCtx
[
j
]);
aAggs
[
pQuery
->
p
Expr1
[
j
].
base
.
functionId
].
xFinalize
(
&
pRuntimeEnv
->
pCtx
[
j
]);
}
}
/*
/*
...
@@ -3692,14 +3692,14 @@ void finalizeQueryResult(SQueryRuntimeEnv *pRuntimeEnv) {
...
@@ -3692,14 +3692,14 @@ void finalizeQueryResult(SQueryRuntimeEnv *pRuntimeEnv) {
}
else
{
}
else
{
for
(
int32_t
j
=
0
;
j
<
pQuery
->
numOfOutput
;
++
j
)
{
for
(
int32_t
j
=
0
;
j
<
pQuery
->
numOfOutput
;
++
j
)
{
aAggs
[
pQuery
->
p
SelectExpr
[
j
].
base
.
functionId
].
xFinalize
(
&
pRuntimeEnv
->
pCtx
[
j
]);
aAggs
[
pQuery
->
p
Expr1
[
j
].
base
.
functionId
].
xFinalize
(
&
pRuntimeEnv
->
pCtx
[
j
]);
}
}
}
}
}
}
static
bool
hasMainOutput
(
SQuery
*
pQuery
)
{
static
bool
hasMainOutput
(
SQuery
*
pQuery
)
{
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
int32_t
functionId
=
pQuery
->
p
SelectExpr
[
i
].
base
.
functionId
;
int32_t
functionId
=
pQuery
->
p
Expr1
[
i
].
base
.
functionId
;
if
(
functionId
!=
TSDB_FUNC_TS
&&
functionId
!=
TSDB_FUNC_TAG
&&
functionId
!=
TSDB_FUNC_TAGPRJ
)
{
if
(
functionId
!=
TSDB_FUNC_TS
&&
functionId
!=
TSDB_FUNC_TAG
&&
functionId
!=
TSDB_FUNC_TAGPRJ
)
{
return
true
;
return
true
;
...
@@ -3798,7 +3798,7 @@ void setResultOutputBuf(SQueryRuntimeEnv *pRuntimeEnv, SResultRow *pResult) {
...
@@ -3798,7 +3798,7 @@ void setResultOutputBuf(SQueryRuntimeEnv *pRuntimeEnv, SResultRow *pResult) {
SQLFunctionCtx
*
pCtx
=
&
pRuntimeEnv
->
pCtx
[
i
];
SQLFunctionCtx
*
pCtx
=
&
pRuntimeEnv
->
pCtx
[
i
];
pCtx
->
aOutputBuf
=
getPosInResultPage
(
pRuntimeEnv
,
i
,
pResult
,
page
);
pCtx
->
aOutputBuf
=
getPosInResultPage
(
pRuntimeEnv
,
i
,
pResult
,
page
);
int32_t
functionId
=
pQuery
->
p
SelectExpr
[
i
].
base
.
functionId
;
int32_t
functionId
=
pQuery
->
p
Expr1
[
i
].
base
.
functionId
;
if
(
functionId
==
TSDB_FUNC_TOP
||
functionId
==
TSDB_FUNC_BOTTOM
||
functionId
==
TSDB_FUNC_DIFF
)
{
if
(
functionId
==
TSDB_FUNC_TOP
||
functionId
==
TSDB_FUNC_BOTTOM
||
functionId
==
TSDB_FUNC_DIFF
)
{
pCtx
->
ptsOutputBuf
=
pRuntimeEnv
->
pCtx
[
0
].
aOutputBuf
;
pCtx
->
ptsOutputBuf
=
pRuntimeEnv
->
pCtx
[
0
].
aOutputBuf
;
}
}
...
@@ -3941,7 +3941,7 @@ void setIntervalQueryRange(SQInfo *pQInfo, TSKEY key) {
...
@@ -3941,7 +3941,7 @@ void setIntervalQueryRange(SQInfo *pQInfo, TSKEY key) {
bool
requireTimestamp
(
SQuery
*
pQuery
)
{
bool
requireTimestamp
(
SQuery
*
pQuery
)
{
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
i
++
)
{
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
i
++
)
{
int32_t
functionId
=
pQuery
->
p
SelectExpr
[
i
].
base
.
functionId
;
int32_t
functionId
=
pQuery
->
p
Expr1
[
i
].
base
.
functionId
;
if
((
aAggs
[
functionId
].
nStatus
&
TSDB_FUNCSTATE_NEED_TS
)
!=
0
)
{
if
((
aAggs
[
functionId
].
nStatus
&
TSDB_FUNCSTATE_NEED_TS
)
!=
0
)
{
return
true
;
return
true
;
}
}
...
@@ -4130,12 +4130,16 @@ bool queryHasRemainResForTableQuery(SQueryRuntimeEnv* pRuntimeEnv) {
...
@@ -4130,12 +4130,16 @@ bool queryHasRemainResForTableQuery(SQueryRuntimeEnv* pRuntimeEnv) {
return
false
;
return
false
;
}
}
static
int16_t
getNumOfFinalResCol
(
SQuery
*
pQuery
)
{
return
pQuery
->
pExpr2
==
NULL
?
pQuery
->
numOfOutput
:
pQuery
->
numOfExpr2
;
}
static
void
doCopyQueryResultToMsg
(
SQInfo
*
pQInfo
,
int32_t
numOfRows
,
char
*
data
)
{
static
void
doCopyQueryResultToMsg
(
SQInfo
*
pQInfo
,
int32_t
numOfRows
,
char
*
data
)
{
SQuery
*
pQuery
=
pQInfo
->
runtimeEnv
.
pQuery
;
SQuery
*
pQuery
=
pQInfo
->
runtimeEnv
.
pQuery
;
if
(
pQuery
->
pExpr2
==
NULL
)
{
if
(
pQuery
->
pExpr2
==
NULL
)
{
for
(
int32_t
col
=
0
;
col
<
pQuery
->
numOfOutput
;
++
col
)
{
for
(
int32_t
col
=
0
;
col
<
pQuery
->
numOfOutput
;
++
col
)
{
int32_t
bytes
=
pQuery
->
p
SelectExpr
[
col
].
bytes
;
int32_t
bytes
=
pQuery
->
p
Expr1
[
col
].
bytes
;
memmove
(
data
,
pQuery
->
sdata
[
col
]
->
data
,
bytes
*
numOfRows
);
memmove
(
data
,
pQuery
->
sdata
[
col
]
->
data
,
bytes
*
numOfRows
);
data
+=
bytes
*
numOfRows
;
data
+=
bytes
*
numOfRows
;
...
@@ -4196,10 +4200,9 @@ int32_t doFillGapsInResults(SQueryRuntimeEnv* pRuntimeEnv, tFilePage **pDst, int
...
@@ -4196,10 +4200,9 @@ int32_t doFillGapsInResults(SQueryRuntimeEnv* pRuntimeEnv, tFilePage **pDst, int
ret
-=
(
int32_t
)
pQuery
->
limit
.
offset
;
ret
-=
(
int32_t
)
pQuery
->
limit
.
offset
;
// todo !!!!there exactly number of interpo is not valid.
// todo !!!!there exactly number of interpo is not valid.
// todo refactor move to the beginning of buffer
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
memmove
(
pDst
[
i
]
->
data
,
pDst
[
i
]
->
data
+
pQuery
->
p
SelectExpr
[
i
].
bytes
*
pQuery
->
limit
.
offset
,
memmove
(
pDst
[
i
]
->
data
,
pDst
[
i
]
->
data
+
pQuery
->
p
Expr1
[
i
].
bytes
*
pQuery
->
limit
.
offset
,
ret
*
pQuery
->
p
SelectExpr
[
i
].
bytes
);
ret
*
pQuery
->
p
Expr1
[
i
].
bytes
);
}
}
pQuery
->
limit
.
offset
=
0
;
pQuery
->
limit
.
offset
=
0
;
...
@@ -4559,7 +4562,7 @@ static int32_t setupQueryHandle(void* tsdb, SQInfo* pQInfo, bool isSTableQuery)
...
@@ -4559,7 +4562,7 @@ static int32_t setupQueryHandle(void* tsdb, SQInfo* pQInfo, bool isSTableQuery)
}
}
static
SFillColInfo
*
createFillColInfo
(
SQuery
*
pQuery
)
{
static
SFillColInfo
*
createFillColInfo
(
SQuery
*
pQuery
)
{
int32_t
numOfCols
=
pQuery
->
pExpr2
==
NULL
?
pQuery
->
numOfOutput
:
pQuery
->
numOfExpr2
;
int32_t
numOfCols
=
getNumOfFinalResCol
(
pQuery
)
;
int32_t
offset
=
0
;
int32_t
offset
=
0
;
SFillColInfo
*
pFillCol
=
calloc
(
numOfCols
,
sizeof
(
SFillColInfo
));
SFillColInfo
*
pFillCol
=
calloc
(
numOfCols
,
sizeof
(
SFillColInfo
));
...
@@ -4569,7 +4572,7 @@ static SFillColInfo* createFillColInfo(SQuery* pQuery) {
...
@@ -4569,7 +4572,7 @@ static SFillColInfo* createFillColInfo(SQuery* pQuery) {
// TODO refactor
// TODO refactor
for
(
int32_t
i
=
0
;
i
<
numOfCols
;
++
i
)
{
for
(
int32_t
i
=
0
;
i
<
numOfCols
;
++
i
)
{
SExprInfo
*
pExprInfo
=
(
pQuery
->
pExpr2
==
NULL
)
?
&
pQuery
->
p
SelectExpr
[
i
]
:&
pQuery
->
pExpr2
[
i
];
SExprInfo
*
pExprInfo
=
(
pQuery
->
pExpr2
==
NULL
)
?
&
pQuery
->
p
Expr1
[
i
]
:&
pQuery
->
pExpr2
[
i
];
pFillCol
[
i
].
col
.
bytes
=
pExprInfo
->
bytes
;
pFillCol
[
i
].
col
.
bytes
=
pExprInfo
->
bytes
;
pFillCol
[
i
].
col
.
type
=
(
int8_t
)
pExprInfo
->
type
;
pFillCol
[
i
].
col
.
type
=
(
int8_t
)
pExprInfo
->
type
;
...
@@ -4681,7 +4684,7 @@ int32_t doInitQInfo(SQInfo *pQInfo, STSBuf *pTsBuf, void *tsdb, int32_t vgId, bo
...
@@ -4681,7 +4684,7 @@ int32_t doInitQInfo(SQInfo *pQInfo, STSBuf *pTsBuf, void *tsdb, int32_t vgId, bo
TSKEY
ek
=
MAX
(
pQuery
->
window
.
skey
,
pQuery
->
window
.
ekey
);
TSKEY
ek
=
MAX
(
pQuery
->
window
.
skey
,
pQuery
->
window
.
ekey
);
getAlignQueryTimeWindow
(
pQuery
,
pQuery
->
window
.
skey
,
sk
,
ek
,
&
w
);
getAlignQueryTimeWindow
(
pQuery
,
pQuery
->
window
.
skey
,
sk
,
ek
,
&
w
);
int32_t
numOfCols
=
pQuery
->
pExpr2
==
NULL
?
pQuery
->
numOfOutput
:
pQuery
->
numOfExpr2
;
int32_t
numOfCols
=
getNumOfFinalResCol
(
pQuery
)
;
pRuntimeEnv
->
pFillInfo
=
taosInitFillInfo
(
pQuery
->
order
.
order
,
w
.
skey
,
0
,
(
int32_t
)
pQuery
->
rec
.
capacity
,
numOfCols
,
pRuntimeEnv
->
pFillInfo
=
taosInitFillInfo
(
pQuery
->
order
.
order
,
w
.
skey
,
0
,
(
int32_t
)
pQuery
->
rec
.
capacity
,
numOfCols
,
pQuery
->
interval
.
sliding
,
pQuery
->
interval
.
slidingUnit
,
(
int8_t
)
pQuery
->
precision
,
pQuery
->
interval
.
sliding
,
pQuery
->
interval
.
slidingUnit
,
(
int8_t
)
pQuery
->
precision
,
pQuery
->
fillType
,
pColInfo
,
pQInfo
);
pQuery
->
fillType
,
pColInfo
,
pQInfo
);
...
@@ -5363,7 +5366,7 @@ static void doSecondaryArithmeticProcess(SQuery* pQuery) {
...
@@ -5363,7 +5366,7 @@ static void doSecondaryArithmeticProcess(SQuery* pQuery) {
pArithSup
->
offset
=
0
;
pArithSup
->
offset
=
0
;
pArithSup
->
numOfCols
=
(
int32_t
)
pQuery
->
numOfOutput
;
pArithSup
->
numOfCols
=
(
int32_t
)
pQuery
->
numOfOutput
;
pArithSup
->
exprList
=
pQuery
->
p
SelectExpr
;
pArithSup
->
exprList
=
pQuery
->
p
Expr1
;
pArithSup
->
data
=
calloc
(
pArithSup
->
numOfCols
,
POINTER_BYTES
);
pArithSup
->
data
=
calloc
(
pArithSup
->
numOfCols
,
POINTER_BYTES
);
for
(
int32_t
k
=
0
;
k
<
pArithSup
->
numOfCols
;
++
k
)
{
for
(
int32_t
k
=
0
;
k
<
pArithSup
->
numOfCols
;
++
k
)
{
...
@@ -5378,9 +5381,9 @@ static void doSecondaryArithmeticProcess(SQuery* pQuery) {
...
@@ -5378,9 +5381,9 @@ static void doSecondaryArithmeticProcess(SQuery* pQuery) {
if
(
pSqlFunc
->
functionId
!=
TSDB_FUNC_ARITHM
)
{
if
(
pSqlFunc
->
functionId
!=
TSDB_FUNC_ARITHM
)
{
for
(
int32_t
j
=
0
;
j
<
pQuery
->
numOfOutput
;
++
j
)
{
for
(
int32_t
j
=
0
;
j
<
pQuery
->
numOfOutput
;
++
j
)
{
if
(
pSqlFunc
->
functionId
==
pQuery
->
p
SelectExpr
[
j
].
base
.
functionId
&&
if
(
pSqlFunc
->
functionId
==
pQuery
->
p
Expr1
[
j
].
base
.
functionId
&&
pSqlFunc
->
colInfo
.
colId
==
pQuery
->
p
SelectExpr
[
j
].
base
.
colInfo
.
colId
)
{
pSqlFunc
->
colInfo
.
colId
==
pQuery
->
p
Expr1
[
j
].
base
.
colInfo
.
colId
)
{
memcpy
(
data
[
i
]
->
data
,
pQuery
->
sdata
[
j
]
->
data
,
pQuery
->
p
SelectExpr
[
j
].
bytes
*
pQuery
->
rec
.
rows
);
memcpy
(
data
[
i
]
->
data
,
pQuery
->
sdata
[
j
]
->
data
,
pQuery
->
p
Expr1
[
j
].
bytes
*
pQuery
->
rec
.
rows
);
break
;
break
;
}
}
}
}
...
@@ -6299,10 +6302,10 @@ static int32_t createFilterInfo(void *pQInfo, SQuery *pQuery) {
...
@@ -6299,10 +6302,10 @@ static int32_t createFilterInfo(void *pQInfo, SQuery *pQuery) {
}
}
static
void
doUpdateExprColumnIndex
(
SQuery
*
pQuery
)
{
static
void
doUpdateExprColumnIndex
(
SQuery
*
pQuery
)
{
assert
(
pQuery
->
p
SelectExpr
!=
NULL
&&
pQuery
!=
NULL
);
assert
(
pQuery
->
p
Expr1
!=
NULL
&&
pQuery
!=
NULL
);
for
(
int32_t
k
=
0
;
k
<
pQuery
->
numOfOutput
;
++
k
)
{
for
(
int32_t
k
=
0
;
k
<
pQuery
->
numOfOutput
;
++
k
)
{
SSqlFuncMsg
*
pSqlExprMsg
=
&
pQuery
->
p
SelectExpr
[
k
].
base
;
SSqlFuncMsg
*
pSqlExprMsg
=
&
pQuery
->
p
Expr1
[
k
].
base
;
if
(
pSqlExprMsg
->
functionId
==
TSDB_FUNC_ARITHM
)
{
if
(
pSqlExprMsg
->
functionId
==
TSDB_FUNC_ARITHM
)
{
continue
;
continue
;
}
}
...
@@ -6383,7 +6386,7 @@ static SQInfo *createQInfoImpl(SQueryTableMsg *pQueryMsg, SSqlGroupbyExpr *pGrou
...
@@ -6383,7 +6386,7 @@ static SQInfo *createQInfoImpl(SQueryTableMsg *pQueryMsg, SSqlGroupbyExpr *pGrou
pQuery
->
limit
.
offset
=
pQueryMsg
->
offset
;
pQuery
->
limit
.
offset
=
pQueryMsg
->
offset
;
pQuery
->
order
.
order
=
pQueryMsg
->
order
;
pQuery
->
order
.
order
=
pQueryMsg
->
order
;
pQuery
->
order
.
orderColId
=
pQueryMsg
->
orderColId
;
pQuery
->
order
.
orderColId
=
pQueryMsg
->
orderColId
;
pQuery
->
p
SelectExpr
=
pExprs
;
pQuery
->
p
Expr1
=
pExprs
;
pQuery
->
pExpr2
=
pSecExprs
;
pQuery
->
pExpr2
=
pSecExprs
;
pQuery
->
numOfExpr2
=
pQueryMsg
->
secondStageOutput
;
pQuery
->
numOfExpr2
=
pQueryMsg
->
secondStageOutput
;
pQuery
->
pGroupbyExpr
=
pGroupbyExpr
;
pQuery
->
pGroupbyExpr
=
pGroupbyExpr
;
...
@@ -6425,6 +6428,7 @@ static SQInfo *createQInfoImpl(SQueryTableMsg *pQueryMsg, SSqlGroupbyExpr *pGrou
...
@@ -6425,6 +6428,7 @@ static SQInfo *createQInfoImpl(SQueryTableMsg *pQueryMsg, SSqlGroupbyExpr *pGrou
for
(
int32_t
col
=
0
;
col
<
pQuery
->
numOfOutput
;
++
col
)
{
for
(
int32_t
col
=
0
;
col
<
pQuery
->
numOfOutput
;
++
col
)
{
// allocate additional memory for interResults that are usually larger then final results
// allocate additional memory for interResults that are usually larger then final results
// TODO refactor
int16_t
bytes
=
0
;
int16_t
bytes
=
0
;
if
(
pQuery
->
pExpr2
==
NULL
||
col
>
pQuery
->
numOfExpr2
)
{
if
(
pQuery
->
pExpr2
==
NULL
||
col
>
pQuery
->
numOfExpr2
)
{
bytes
=
pExprs
[
col
].
bytes
;
bytes
=
pExprs
[
col
].
bytes
;
...
@@ -6676,16 +6680,16 @@ static void freeQInfo(SQInfo *pQInfo) {
...
@@ -6676,16 +6680,16 @@ static void freeQInfo(SQInfo *pQInfo) {
}
}
}
}
if
(
pQuery
->
p
SelectExpr
!=
NULL
)
{
if
(
pQuery
->
p
Expr1
!=
NULL
)
{
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
SExprInfo
*
pExprInfo
=
&
pQuery
->
p
SelectExpr
[
i
];
SExprInfo
*
pExprInfo
=
&
pQuery
->
p
Expr1
[
i
];
if
(
pExprInfo
->
pExpr
!=
NULL
)
{
if
(
pExprInfo
->
pExpr
!=
NULL
)
{
tExprTreeDestroy
(
&
pExprInfo
->
pExpr
,
NULL
);
tExprTreeDestroy
(
&
pExprInfo
->
pExpr
,
NULL
);
}
}
}
}
tfree
(
pQuery
->
p
SelectExpr
);
tfree
(
pQuery
->
p
Expr1
);
}
}
if
(
pQuery
->
pGroupbyExpr
!=
NULL
)
{
if
(
pQuery
->
pGroupbyExpr
!=
NULL
)
{
...
@@ -7188,11 +7192,11 @@ static void buildTagQueryResult(SQInfo* pQInfo) {
...
@@ -7188,11 +7192,11 @@ static void buildTagQueryResult(SQInfo* pQInfo) {
assert
(
num
==
pQInfo
->
tableqinfoGroupInfo
.
numOfTables
);
assert
(
num
==
pQInfo
->
tableqinfoGroupInfo
.
numOfTables
);
int32_t
count
=
0
;
int32_t
count
=
0
;
int32_t
functionId
=
pQuery
->
p
SelectExpr
[
0
].
base
.
functionId
;
int32_t
functionId
=
pQuery
->
p
Expr1
[
0
].
base
.
functionId
;
if
(
functionId
==
TSDB_FUNC_TID_TAG
)
{
// return the tags & table Id
if
(
functionId
==
TSDB_FUNC_TID_TAG
)
{
// return the tags & table Id
assert
(
pQuery
->
numOfOutput
==
1
);
assert
(
pQuery
->
numOfOutput
==
1
);
SExprInfo
*
pExprInfo
=
&
pQuery
->
p
SelectExpr
[
0
];
SExprInfo
*
pExprInfo
=
&
pQuery
->
p
Expr1
[
0
];
int32_t
rsize
=
pExprInfo
->
bytes
;
int32_t
rsize
=
pExprInfo
->
bytes
;
count
=
0
;
count
=
0
;
...
@@ -7266,7 +7270,7 @@ static void buildTagQueryResult(SQInfo* pQInfo) {
...
@@ -7266,7 +7270,7 @@ static void buildTagQueryResult(SQInfo* pQInfo) {
continue
;
continue
;
}
}
SExprInfo
*
pExprInfo
=
pQuery
->
p
SelectExpr
;
SExprInfo
*
pExprInfo
=
pQuery
->
p
Expr1
;
STableQueryInfo
*
item
=
taosArrayGetP
(
pa
,
i
);
STableQueryInfo
*
item
=
taosArrayGetP
(
pa
,
i
);
char
*
data
=
NULL
,
*
dst
=
NULL
;
char
*
data
=
NULL
,
*
dst
=
NULL
;
...
...
src/query/src/qUtil.c
浏览文件 @
d5aa32bf
...
@@ -24,7 +24,7 @@ int32_t getOutputInterResultBufSize(SQuery* pQuery) {
...
@@ -24,7 +24,7 @@ int32_t getOutputInterResultBufSize(SQuery* pQuery) {
int32_t
size
=
0
;
int32_t
size
=
0
;
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
size
+=
pQuery
->
p
SelectExpr
[
i
].
interBytes
;
size
+=
pQuery
->
p
Expr1
[
i
].
interBytes
;
}
}
assert
(
size
>=
0
);
assert
(
size
>=
0
);
...
@@ -237,7 +237,7 @@ void clearResultRow(SQueryRuntimeEnv *pRuntimeEnv, SResultRow *pWindowRes) {
...
@@ -237,7 +237,7 @@ void clearResultRow(SQueryRuntimeEnv *pRuntimeEnv, SResultRow *pWindowRes) {
SResultRowCellInfo
*
pResultInfo
=
&
pWindowRes
->
pCellInfo
[
i
];
SResultRowCellInfo
*
pResultInfo
=
&
pWindowRes
->
pCellInfo
[
i
];
char
*
s
=
getPosInResultPage
(
pRuntimeEnv
,
i
,
pWindowRes
,
page
);
char
*
s
=
getPosInResultPage
(
pRuntimeEnv
,
i
,
pWindowRes
,
page
);
size_t
size
=
pRuntimeEnv
->
pQuery
->
p
SelectExpr
[
i
].
bytes
;
size_t
size
=
pRuntimeEnv
->
pQuery
->
p
Expr1
[
i
].
bytes
;
memset
(
s
,
0
,
size
);
memset
(
s
,
0
,
size
);
RESET_RESULT_INFO
(
pResultInfo
);
RESET_RESULT_INFO
(
pResultInfo
);
...
@@ -280,7 +280,7 @@ void copyResultRow(SQueryRuntimeEnv *pRuntimeEnv, SResultRow *dst, const SResult
...
@@ -280,7 +280,7 @@ void copyResultRow(SQueryRuntimeEnv *pRuntimeEnv, SResultRow *dst, const SResult
tFilePage
*
srcpage
=
getResBufPage
(
pRuntimeEnv
->
pResultBuf
,
src
->
pageId
);
tFilePage
*
srcpage
=
getResBufPage
(
pRuntimeEnv
->
pResultBuf
,
src
->
pageId
);
char
*
srcBuf
=
getPosInResultPage
(
pRuntimeEnv
,
i
,
(
SResultRow
*
)
src
,
srcpage
);
char
*
srcBuf
=
getPosInResultPage
(
pRuntimeEnv
,
i
,
(
SResultRow
*
)
src
,
srcpage
);
size_t
s
=
pRuntimeEnv
->
pQuery
->
p
SelectExpr
[
i
].
bytes
;
size_t
s
=
pRuntimeEnv
->
pQuery
->
p
Expr1
[
i
].
bytes
;
memcpy
(
dstBuf
,
srcBuf
,
s
);
memcpy
(
dstBuf
,
srcBuf
,
s
);
}
}
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录