Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
小九先生
TDengine
提交
55e3352d
TDengine
项目概览
小九先生
/
TDengine
与 Fork 源项目一致
Fork自
taosdata / TDengine
通知
2
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,发现更多精彩内容 >>
提交
55e3352d
编写于
7月 19, 2020
作者:
H
Haojun Liao
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
[td-225] refactor codes. opt resbuf perf. rename sources file
上级
1593712a
变更
14
隐藏空白更改
内联
并排
Showing
14 changed file
with
368 addition
and
123 deletion
+368
-123
src/client/inc/tsclient.h
src/client/inc/tsclient.h
+1
-1
src/client/src/tscFunctionImpl.c
src/client/src/tscFunctionImpl.c
+1
-7
src/client/src/tscServer.c
src/client/src/tscServer.c
+24
-45
src/query/inc/qexecutor.h
src/query/inc/qexecutor.h
+3
-3
src/query/inc/qresultbuf.h
src/query/inc/qresultbuf.h
+18
-6
src/query/inc/qutil.h
src/query/inc/qutil.h
+1
-3
src/query/inc/tsqlfunction.h
src/query/inc/tsqlfunction.h
+9
-1
src/query/src/qexecutor.c
src/query/src/qexecutor.c
+86
-46
src/query/src/qfilterfunc.c
src/query/src/qfilterfunc.c
+1
-3
src/query/src/qresultbuf.c
src/query/src/qresultbuf.c
+218
-0
src/query/src/qutil.c
src/query/src/qutil.c
+3
-5
src/query/tests/resultBufferTest.cpp
src/query/tests/resultBufferTest.cpp
+1
-1
src/tsdb/src/tsdbRead.c
src/tsdb/src/tsdbRead.c
+1
-1
src/util/src/tutil.c
src/util/src/tutil.c
+1
-1
未找到文件。
src/client/inc/tsclient.h
浏览文件 @
55e3352d
...
...
@@ -30,7 +30,7 @@ extern "C" {
#include "tsqlfunction.h"
#include "tutil.h"
#include "q
E
xecutor.h"
#include "q
e
xecutor.h"
#include "qsqlparser.h"
#include "qtsbuf.h"
#include "tcmdtype.h"
...
...
src/client/src/tscFunctionImpl.c
浏览文件 @
55e3352d
...
...
@@ -74,7 +74,7 @@ for (int32_t i = 0; i < (ctx)->tagInfo.numOfTagCols; ++i) { \
void
noop1
(
SQLFunctionCtx
*
UNUSED_PARAM
(
pCtx
))
{}
void
noop2
(
SQLFunctionCtx
*
UNUSED_PARAM
(
pCtx
),
int32_t
UNUSED_PARAM
(
index
))
{}
void
doFinalizer
(
SQLFunctionCtx
*
pCtx
)
{
resetResultInfo
(
GET_RES_INFO
(
pCtx
));
}
void
doFinalizer
(
SQLFunctionCtx
*
pCtx
)
{
RESET_RESULT_INFO
(
GET_RES_INFO
(
pCtx
));
}
typedef
struct
tValuePair
{
tVariant
v
;
...
...
@@ -330,12 +330,6 @@ int32_t getResultDataInfo(int32_t dataType, int32_t dataBytes, int32_t functionI
return
TSDB_CODE_SUCCESS
;
}
/**
* the numOfRes should be kept, since it may be used later
* and allow the ResultInfo to be re initialized
*/
void
resetResultInfo
(
SResultInfo
*
pResInfo
)
{
pResInfo
->
initialized
=
false
;
}
void
setResultInfoBuf
(
SResultInfo
*
pResInfo
,
int32_t
size
,
bool
superTable
,
char
*
buf
)
{
assert
(
pResInfo
->
interResultBuf
==
NULL
);
...
...
src/client/src/tscServer.c
浏览文件 @
55e3352d
...
...
@@ -242,52 +242,31 @@ void tscProcessMsgFromServer(SRpcMsg *rpcMsg, SRpcEpSet *pEpSet) {
if
(
pEpSet
)
tscMgmtEpSet
=
*
pEpSet
;
}
if
(
rpcMsg
->
pCont
==
NULL
)
{
rpcMsg
->
code
=
TSDB_CODE_RPC_NETWORK_UNAVAIL
;
}
else
{
STableMetaInfo
*
pTableMetaInfo
=
tscGetTableMetaInfoFromCmd
(
pCmd
,
pCmd
->
clauseIndex
,
0
);
// if (rpcMsg->code != TSDB_CODE_RPC_NETWORK_UNAVAIL) {
// if (pCmd->command == TSDB_SQL_CONNECT) {
// rpcMsg->code = TSDB_CODE_RPC_NETWORK_UNAVAIL;
// rpcFreeCont(rpcMsg->pCont);
// return;
// }
// if (pCmd->command == TSDB_SQL_HB) {
// rpcMsg->code = TSDB_CODE_RPC_NOT_READY;
// rpcFreeCont(rpcMsg->pCont);
// return;
// }
// if (pCmd->command == TSDB_SQL_META || pCmd->command == TSDB_SQL_DESCRIBE_TABLE ||
// pCmd->command == TSDB_SQL_STABLEVGROUP || pCmd->command == TSDB_SQL_SHOW ||
// pCmd->command == TSDB_SQL_RETRIEVE) {
// // get table meta/vgroup query will not retry, do nothing
// }
// }
if
((
pCmd
->
command
==
TSDB_SQL_SELECT
||
pCmd
->
command
==
TSDB_SQL_FETCH
||
pCmd
->
command
==
TSDB_SQL_INSERT
||
pCmd
->
command
==
TSDB_SQL_UPDATE_TAGS_VAL
)
&&
(
rpcMsg
->
code
==
TSDB_CODE_TDB_INVALID_TABLE_ID
||
rpcMsg
->
code
==
TSDB_CODE_VND_INVALID_VGROUP_ID
||
rpcMsg
->
code
==
TSDB_CODE_RPC_NETWORK_UNAVAIL
||
rpcMsg
->
code
==
TSDB_CODE_TDB_TABLE_RECONFIGURE
))
{
tscWarn
(
"%p it shall renew table meta, code:%s, retry:%d"
,
pSql
,
tstrerror
(
rpcMsg
->
code
),
++
pSql
->
retry
);
// set the flag to denote that sql string needs to be re-parsed and build submit block with table schema
if
(
rpcMsg
->
code
==
TSDB_CODE_TDB_TABLE_RECONFIGURE
)
{
pSql
->
cmd
.
submitSchema
=
1
;
}
STableMetaInfo
*
pTableMetaInfo
=
tscGetTableMetaInfoFromCmd
(
pCmd
,
pCmd
->
clauseIndex
,
0
);
pSql
->
res
.
code
=
rpcMsg
->
code
;
// keep the previous error code
if
(
pSql
->
retry
>
pSql
->
maxRetry
)
{
tscError
(
"%p max retry %d reached, give up"
,
pSql
,
pSql
->
maxRetry
);
}
else
{
rpcMsg
->
code
=
tscRenewTableMeta
(
pSql
,
pTableMetaInfo
->
name
);
// if there is an error occurring, proceed to the following error handling procedure.
// todo add test cases
if
(
rpcMsg
->
code
==
TSDB_CODE_TSC_ACTION_IN_PROGRESS
)
{
rpcFreeCont
(
rpcMsg
->
pCont
);
return
;
}
int32_t
cmd
=
pCmd
->
command
;
if
((
cmd
==
TSDB_SQL_SELECT
||
cmd
==
TSDB_SQL_FETCH
||
cmd
==
TSDB_SQL_INSERT
||
cmd
==
TSDB_SQL_UPDATE_TAGS_VAL
)
&&
(
rpcMsg
->
code
==
TSDB_CODE_TDB_INVALID_TABLE_ID
||
rpcMsg
->
code
==
TSDB_CODE_VND_INVALID_VGROUP_ID
||
rpcMsg
->
code
==
TSDB_CODE_RPC_NETWORK_UNAVAIL
||
rpcMsg
->
code
==
TSDB_CODE_TDB_TABLE_RECONFIGURE
))
{
tscWarn
(
"%p it shall renew table meta, code:%s, retry:%d"
,
pSql
,
tstrerror
(
rpcMsg
->
code
),
++
pSql
->
retry
);
// set the flag to denote that sql string needs to be re-parsed and build submit block with table schema
if
(
rpcMsg
->
code
==
TSDB_CODE_TDB_TABLE_RECONFIGURE
)
{
pSql
->
cmd
.
submitSchema
=
1
;
}
pSql
->
res
.
code
=
rpcMsg
->
code
;
// keep the previous error code
if
(
pSql
->
retry
>
pSql
->
maxRetry
)
{
tscError
(
"%p max retry %d reached, give up"
,
pSql
,
pSql
->
maxRetry
);
}
else
{
rpcMsg
->
code
=
tscRenewTableMeta
(
pSql
,
pTableMetaInfo
->
name
);
// if there is an error occurring, proceed to the following error handling procedure.
if
(
rpcMsg
->
code
==
TSDB_CODE_TSC_ACTION_IN_PROGRESS
)
{
rpcFreeCont
(
rpcMsg
->
pCont
);
return
;
}
}
}
...
...
src/query/inc/q
E
xecutor.h
→
src/query/inc/q
e
xecutor.h
浏览文件 @
55e3352d
...
...
@@ -19,15 +19,15 @@
#include "hash.h"
#include "qfill.h"
#include "qresult
B
uf.h"
#include "qresult
b
uf.h"
#include "qsqlparser.h"
#include "qtsbuf.h"
#include "query.h"
#include "taosdef.h"
#include "tarray.h"
#include "tlockfree.h"
#include "tsdb.h"
#include "tsqlfunction.h"
#include "query.h"
struct
SColumnFilterElem
;
typedef
bool
(
*
__filter_func_t
)(
struct
SColumnFilterElem
*
pFilter
,
char
*
val1
,
char
*
val2
);
...
...
@@ -158,7 +158,7 @@ typedef struct SQueryRuntimeEnv {
SResultInfo
*
resultInfo
;
// todo refactor to merge with SWindowResInfo
SQuery
*
pQuery
;
SQLFunctionCtx
*
pCtx
;
int
16
_t
numOfRowsPerPage
;
int
32
_t
numOfRowsPerPage
;
int16_t
offset
[
TSDB_MAX_COLUMNS
];
uint16_t
scanFlag
;
// denotes reversed scan of data or not
SFillInfo
*
pFillInfo
;
...
...
src/query/inc/qresult
B
uf.h
→
src/query/inc/qresult
b
uf.h
浏览文件 @
55e3352d
...
...
@@ -33,14 +33,19 @@ typedef struct SDiskbasedResultBuf {
int32_t
fd
;
// data file fd
int32_t
allocateId
;
// allocated page id
int32_t
incStep
;
// minimum allocated pages
char
*
pBuf
;
// mmap buffer pointer
void
*
pBuf
;
// mmap buffer pointer
char
*
path
;
// file path
int32_t
pageSize
;
// current used page size
int32_t
inMemPages
;
// numOfPages that are allocated in memory
SHashObj
*
idsTable
;
// id hash table
SIDList
list
;
// for each id, there is a page id list
void
*
iBuf
;
// inmemory buf
void
*
handle
;
// for debug purpose
}
SDiskbasedResultBuf
;
#define DEFAULT_INTERN_BUF_PAGE_SIZE (8192L*5)
#define DEFAULT_INTERN_BUF_PAGE_SIZE (1024L)
#define DEFAULT_INMEM_BUF_PAGES 10
/**
* create disk-based result buffer
...
...
@@ -49,7 +54,8 @@ typedef struct SDiskbasedResultBuf {
* @param rowSize
* @return
*/
int32_t
createDiskbasedResultBuffer
(
SDiskbasedResultBuf
**
pResultBuf
,
int32_t
size
,
int32_t
rowSize
,
void
*
handle
);
int32_t
createDiskbasedResultBuffer
(
SDiskbasedResultBuf
**
pResultBuf
,
int32_t
numOfPages
,
int32_t
rowSize
,
int32_t
pagesize
,
int32_t
inMemPages
,
void
*
handle
);
/**
*
...
...
@@ -81,8 +87,14 @@ SIDList getDataBufPagesIdList(SDiskbasedResultBuf* pResultBuf, int32_t groupId);
* @param id
* @return
*/
#define GET_RES_BUF_PAGE_BY_ID(buf, id) ((tFilePage*)((buf)->pBuf + DEFAULT_INTERN_BUF_PAGE_SIZE*(id)))
//#define getResBufPage(buf, id) ((tFilePage*)((buf)->pBuf + (buf)->pageSize * (id)))
static
FORCE_INLINE
tFilePage
*
getResBufPage
(
SDiskbasedResultBuf
*
pResultBuf
,
int32_t
id
)
{
if
(
id
<
pResultBuf
->
inMemPages
)
{
return
pResultBuf
->
iBuf
+
id
*
pResultBuf
->
pageSize
;
}
else
{
return
pResultBuf
->
pBuf
+
(
id
-
pResultBuf
->
inMemPages
)
*
pResultBuf
->
pageSize
;
}
}
/**
* get the total buffer size in the format of disk file
* @param pResultBuf
...
...
src/query/inc/q
U
til.h
→
src/query/inc/q
u
til.h
浏览文件 @
55e3352d
...
...
@@ -49,7 +49,7 @@ static FORCE_INLINE char *getPosInResultPage(SQueryRuntimeEnv *pRuntimeEnv, int3
assert
(
pResult
!=
NULL
&&
pRuntimeEnv
!=
NULL
);
SQuery
*
pQuery
=
pRuntimeEnv
->
pQuery
;
tFilePage
*
page
=
GET_RES_BUF_PAGE_BY_ID
(
pRuntimeEnv
->
pResultBuf
,
pResult
->
pos
.
pageId
);
tFilePage
*
page
=
getResBufPage
(
pRuntimeEnv
->
pResultBuf
,
pResult
->
pos
.
pageId
);
int32_t
realRowId
=
pResult
->
pos
.
rowId
*
GET_ROW_PARAM_FOR_MULTIOUTPUT
(
pQuery
,
pRuntimeEnv
->
topBotQuery
,
pRuntimeEnv
->
stableQuery
);
return
((
char
*
)
page
->
data
)
+
pRuntimeEnv
->
offset
[
columnIndex
]
*
pRuntimeEnv
->
numOfRowsPerPage
+
...
...
@@ -59,6 +59,4 @@ static FORCE_INLINE char *getPosInResultPage(SQueryRuntimeEnv *pRuntimeEnv, int3
__filter_func_t
*
getRangeFilterFuncArray
(
int32_t
type
);
__filter_func_t
*
getValueFilterFuncArray
(
int32_t
type
);
bool
supportPrefilter
(
int32_t
type
);
#endif // TDENGINE_QUERYUTIL_H
src/query/inc/tsqlfunction.h
浏览文件 @
55e3352d
...
...
@@ -255,7 +255,15 @@ extern int32_t functionCompatList[]; // compatible check array list
bool
topbot_datablock_filter
(
SQLFunctionCtx
*
pCtx
,
int32_t
functionId
,
const
char
*
minval
,
const
char
*
maxval
);
void
resetResultInfo
(
SResultInfo
*
pResInfo
);
/**
* the numOfRes should be kept, since it may be used later
* and allow the ResultInfo to be re initialized
*/
#define RESET_RESULT_INFO(_r) \
do { \
(_r)->initialized = false; \
} while (0)
void
setResultInfoBuf
(
SResultInfo
*
pResInfo
,
int32_t
size
,
bool
superTable
,
char
*
buf
);
static
FORCE_INLINE
void
initResultInfo
(
SResultInfo
*
pResInfo
)
{
...
...
src/query/src/q
E
xecutor.c
→
src/query/src/q
e
xecutor.c
浏览文件 @
55e3352d
...
...
@@ -20,12 +20,12 @@
#include "exception.h"
#include "hash.h"
#include "q
Executor
.h"
#include "q
Util
.h"
#include "qresult
B
uf.h"
#include "q
ast
.h"
#include "q
executor
.h"
#include "qresult
b
uf.h"
#include "query.h"
#include "queryLog.h"
#include "q
ast
.h"
#include "q
util
.h"
#include "tlosertree.h"
#include "tscompression.h"
#include "ttime.h"
...
...
@@ -233,9 +233,7 @@ bool isGroupbyNormalCol(SSqlGroupbyExpr *pGroupbyExpr) {
for
(
int32_t
i
=
0
;
i
<
pGroupbyExpr
->
numOfGroupCols
;
++
i
)
{
SColIndex
*
pColIndex
=
taosArrayGet
(
pGroupbyExpr
->
columnInfo
,
i
);
if
(
pColIndex
->
flag
==
TSDB_COL_NORMAL
)
{
/*
* make sure the normal column locates at the second position if tbname exists in group by clause
*/
//make sure the normal column locates at the second position if tbname exists in group by clause
if
(
pGroupbyExpr
->
numOfGroupCols
>
1
)
{
assert
(
pColIndex
->
colIndex
>
0
);
}
...
...
@@ -294,6 +292,17 @@ bool isSelectivityWithTagsQuery(SQuery *pQuery) {
return
false
;
}
bool
isProjQuery
(
SQuery
*
pQuery
)
{
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
int32_t
functId
=
pQuery
->
pSelectExpr
[
i
].
base
.
functionId
;
if
(
functId
!=
TSDB_FUNC_PRJ
&&
functId
!=
TSDB_FUNC_TAGPRJ
)
{
return
false
;
}
}
return
true
;
}
bool
isTSCompQuery
(
SQuery
*
pQuery
)
{
return
pQuery
->
pSelectExpr
[
0
].
base
.
functionId
==
TSDB_FUNC_TS_COMP
;
}
static
bool
limitResults
(
SQueryRuntimeEnv
*
pRuntimeEnv
)
{
...
...
@@ -470,7 +479,7 @@ static int32_t addNewWindowResultBuf(SWindowResult *pWindowRes, SDiskbasedResult
pData
=
getNewDataBuf
(
pResultBuf
,
sid
,
&
pageId
);
}
else
{
pageId
=
getLastPageId
(
list
);
pData
=
GET_RES_BUF_PAGE_BY_ID
(
pResultBuf
,
pageId
);
pData
=
getResBufPage
(
pResultBuf
,
pageId
);
if
(
pData
->
num
>=
numOfRowsPerPage
)
{
pData
=
getNewDataBuf
(
pResultBuf
,
sid
,
&
pageId
);
...
...
@@ -1003,7 +1012,6 @@ static int32_t setGroupResultOutputBuf(SQueryRuntimeEnv *pRuntimeEnv, char *pDat
case
TSDB_DATA_TYPE_BIGINT
:
v
=
GET_INT64_VAL
(
pData
);
break
;
}
// assert(pRuntimeEnv->windowResInfo.hashList->size <= 2);
SWindowResult
*
pWindowRes
=
doSetTimeWindowFromKey
(
pRuntimeEnv
,
&
pRuntimeEnv
->
windowResInfo
,
pData
,
bytes
,
true
);
if
(
pWindowRes
==
NULL
)
{
return
-
1
;
...
...
@@ -1201,9 +1209,8 @@ static void rowwiseApplyFunctions(SQueryRuntimeEnv *pRuntimeEnv, SDataStatis *pS
continue
;
}
// interval window query
// interval window query
, decide the time window according to the primary timestamp
if
(
QUERY_IS_INTERVAL_QUERY
(
pQuery
))
{
// decide the time window according to the primary timestamp
int64_t
ts
=
tsCols
[
offset
];
STimeWindow
win
=
getActiveTimeWindow
(
pWindowResInfo
,
ts
,
pQuery
);
...
...
@@ -1225,8 +1232,7 @@ static void rowwiseApplyFunctions(SQueryRuntimeEnv *pRuntimeEnv, SDataStatis *pS
while
(
1
)
{
GET_NEXT_TIMEWINDOW
(
pQuery
,
&
nextWin
);
if
(
/*pWindowResInfo->startTime > nextWin.skey ||*/
(
nextWin
.
skey
>
pQuery
->
window
.
ekey
&&
QUERY_IS_ASC_QUERY
(
pQuery
))
||
if
((
nextWin
.
skey
>
pQuery
->
window
.
ekey
&&
QUERY_IS_ASC_QUERY
(
pQuery
))
||
(
nextWin
.
skey
<
pQuery
->
window
.
ekey
&&
!
QUERY_IS_ASC_QUERY
(
pQuery
)))
{
break
;
}
...
...
@@ -1484,6 +1490,8 @@ static int32_t setupQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv, int16_t order
goto
_clean
;
}
qDebug
(
"QInfo:%p setup runtime env1"
,
GET_QINFO_ADDR
(
pRuntimeEnv
));
pRuntimeEnv
->
offset
[
0
]
=
0
;
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
SSqlFuncMsg
*
pSqlFuncMsg
=
&
pQuery
->
pSelectExpr
[
i
].
base
;
...
...
@@ -1528,6 +1536,8 @@ static int32_t setupQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv, int16_t order
}
}
qDebug
(
"QInfo:%p setup runtime env2"
,
GET_QINFO_ADDR
(
pRuntimeEnv
));
// set the order information for top/bottom query
int32_t
functionId
=
pCtx
->
functionId
;
...
...
@@ -1548,17 +1558,25 @@ static int32_t setupQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv, int16_t order
}
}
qDebug
(
"QInfo:%p setup runtime env3"
,
GET_QINFO_ADDR
(
pRuntimeEnv
));
char
*
buf
=
(
char
*
)
pRuntimeEnv
->
resultInfo
+
sizeof
(
SResultInfo
)
*
pQuery
->
numOfOutput
;
// set the intermediate result output buffer
setWindowResultInfo
(
pRuntimeEnv
->
resultInfo
,
pQuery
,
pRuntimeEnv
->
stableQuery
,
buf
);
qDebug
(
"QInfo:%p setup runtime env4"
,
GET_QINFO_ADDR
(
pRuntimeEnv
));
// if it is group by normal column, do not set output buffer, the output buffer is pResult
if
(
!
isGroupbyNormalCol
(
pQuery
->
pGroupbyExpr
)
&&
!
pRuntimeEnv
->
stableQuery
)
{
if
(
!
pRuntimeEnv
->
groupbyNormalCol
&&
!
pRuntimeEnv
->
stableQuery
)
{
resetCtxOutputBuf
(
pRuntimeEnv
);
}
qDebug
(
"QInfo:%p setup runtime env5"
,
GET_QINFO_ADDR
(
pRuntimeEnv
));
setCtxTagColumnInfo
(
pRuntimeEnv
,
pRuntimeEnv
->
pCtx
);
qDebug
(
"QInfo:%p init completed"
,
GET_QINFO_ADDR
(
pRuntimeEnv
));
return
TSDB_CODE_SUCCESS
;
_clean:
...
...
@@ -1910,9 +1928,20 @@ static int32_t getInitialPageNum(SQInfo *pQInfo) {
return
num
;
}
static
FORCE_INLINE
int32_t
getNumOfRowsInResultPage
(
SQuery
*
pQuery
,
bool
topBotQuery
,
bool
isSTableQuery
)
{
int32_t
rowSize
=
pQuery
->
rowSize
*
GET_ROW_PARAM_FOR_MULTIOUTPUT
(
pQuery
,
topBotQuery
,
isSTableQuery
);
return
(
DEFAULT_INTERN_BUF_PAGE_SIZE
-
sizeof
(
tFilePage
))
/
rowSize
;
static
void
getIntermediateBufInfo
(
SQueryRuntimeEnv
*
pRuntimeEnv
,
int32_t
*
ps
,
int32_t
*
rowsize
)
{
SQuery
*
pQuery
=
pRuntimeEnv
->
pQuery
;
*
rowsize
=
pQuery
->
rowSize
*
GET_ROW_PARAM_FOR_MULTIOUTPUT
(
pQuery
,
pRuntimeEnv
->
topBotQuery
,
pRuntimeEnv
->
stableQuery
);
int32_t
overhead
=
sizeof
(
tFilePage
);
// one page contains at least two rows
*
ps
=
DEFAULT_INTERN_BUF_PAGE_SIZE
;
while
(((
*
rowsize
)
*
2
)
>
(
*
ps
)
-
overhead
)
{
*
ps
=
(
*
ps
<<
1u
);
}
pRuntimeEnv
->
numOfRowsPerPage
=
((
*
ps
)
-
sizeof
(
tFilePage
))
/
(
*
rowsize
);
}
#define IS_PREFILTER_TYPE(_t) ((_t) != TSDB_DATA_TYPE_BINARY && (_t) != TSDB_DATA_TYPE_NCHAR)
...
...
@@ -2038,8 +2067,7 @@ int32_t loadDataBlockOnDemand(SQueryRuntimeEnv *pRuntimeEnv, void* pQueryHandle,
}
else
{
// check if this data block is required to load
// Calculate all time windows that are overlapping or contain current data block.
// If current data block is contained by all possible time window, loading current
// data block is not needed.
// If current data block is contained by all possible time window, do not load current data block.
if
(
QUERY_IS_INTERVAL_QUERY
(
pQuery
)
&&
overlapWithTimeWindow
(
pQuery
,
pBlockInfo
))
{
status
=
BLK_DATA_ALL_NEEDED
;
}
...
...
@@ -2424,7 +2452,7 @@ static void doMerge(SQueryRuntimeEnv *pRuntimeEnv, int64_t timestamp, SWindowRes
pCtx
[
i
].
aOutputBuf
=
pCtx
[
i
].
aOutputBuf
+
pCtx
[
i
].
outputBytes
;
pCtx
[
i
].
currentStage
=
FIRST_STAGE_MERGE
;
resetResultInfo
(
pCtx
[
i
].
resultInfo
);
RESET_RESULT_INFO
(
pCtx
[
i
].
resultInfo
);
aAggs
[
functionId
].
init
(
&
pCtx
[
i
]);
}
...
...
@@ -2661,7 +2689,7 @@ void copyResToQueryResultBuf(SQInfo *pQInfo, SQuery *pQuery) {
int32_t
size
=
taosArrayGetSize
(
list
);
for
(
int32_t
i
=
0
;
i
<
size
;
++
i
)
{
int32_t
*
pgId
=
taosArrayGet
(
list
,
i
);
tFilePage
*
pData
=
GET_RES_BUF_PAGE_BY_ID
(
pResultBuf
,
*
pgId
);
tFilePage
*
pData
=
getResBufPage
(
pResultBuf
,
*
pgId
);
total
+=
pData
->
num
;
}
...
...
@@ -2670,7 +2698,7 @@ void copyResToQueryResultBuf(SQInfo *pQInfo, SQuery *pQuery) {
int32_t
offset
=
0
;
for
(
int32_t
j
=
0
;
j
<
size
;
++
j
)
{
int32_t
*
pgId
=
taosArrayGet
(
list
,
j
);
tFilePage
*
pData
=
GET_RES_BUF_PAGE_BY_ID
(
pResultBuf
,
*
pgId
);
tFilePage
*
pData
=
getResBufPage
(
pResultBuf
,
*
pgId
);
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
int32_t
bytes
=
pRuntimeEnv
->
pCtx
[
i
].
outputBytes
;
...
...
@@ -2860,10 +2888,10 @@ int32_t flushFromResultBuf(SQInfo *pQInfo) {
SQuery
*
pQuery
=
pRuntimeEnv
->
pQuery
;
SDiskbasedResultBuf
*
pResultBuf
=
pRuntimeEnv
->
pResultBuf
;
int32_t
capacity
=
(
DEFAULT_INTERN_BUF_PAGE_SIZE
-
sizeof
(
tFilePage
))
/
pQuery
->
rowSize
;
// the base value for group result, since the maximum number of table for each vnode will not exceed 100,000.
int32_t
pageId
=
-
1
;
int32_t
capacity
=
pResultBuf
->
numOfRowsPerPage
;
int32_t
remain
=
pQuery
->
sdata
[
0
]
->
num
;
int32_t
offset
=
0
;
...
...
@@ -3033,7 +3061,7 @@ void resetCtxOutputBuf(SQueryRuntimeEnv *pRuntimeEnv) {
* set the output buffer information and intermediate buffer
* not all queries require the interResultBuf, such as COUNT/TAGPRJ/PRJ/TAG etc.
*/
resetResultInfo
(
&
pRuntimeEnv
->
resultInfo
[
i
]);
RESET_RESULT_INFO
(
&
pRuntimeEnv
->
resultInfo
[
i
]);
pCtx
->
resultInfo
=
&
pRuntimeEnv
->
resultInfo
[
i
];
// set the timestamp output buffer for top/bottom/diff query
...
...
@@ -3072,7 +3100,7 @@ void forwardCtxOutputBuf(SQueryRuntimeEnv *pRuntimeEnv, int64_t output) {
pRuntimeEnv
->
pCtx
[
j
].
ptsOutputBuf
+=
TSDB_KEYSIZE
*
output
;
}
resetResultInfo
(
pRuntimeEnv
->
pCtx
[
j
].
resultInfo
);
RESET_RESULT_INFO
(
pRuntimeEnv
->
pCtx
[
j
].
resultInfo
);
}
}
...
...
@@ -3326,8 +3354,8 @@ void scanOneTableDataBlocks(SQueryRuntimeEnv *pRuntimeEnv, TSKEY start) {
if
(
pRuntimeEnv
->
pSecQueryHandle
==
NULL
)
{
longjmp
(
pRuntimeEnv
->
env
,
terrno
);
}
pRuntimeEnv
->
windowResInfo
.
curIndex
=
qstatus
.
windowIndex
;
pRuntimeEnv
->
windowResInfo
.
curIndex
=
qstatus
.
windowIndex
;
setQueryStatus
(
pQuery
,
QUERY_NOT_COMPLETED
);
pRuntimeEnv
->
scanFlag
=
REPEAT_SCAN
;
...
...
@@ -3462,7 +3490,6 @@ void setExecutionContext(SQInfo *pQInfo, int32_t groupIndex, TSKEY nextKey) {
return
;
}
int32_t
GROUPRESULTID
=
1
;
SWindowResult
*
pWindowRes
=
doSetTimeWindowFromKey
(
pRuntimeEnv
,
pWindowResInfo
,
(
char
*
)
&
groupIndex
,
sizeof
(
groupIndex
),
true
);
if
(
pWindowRes
==
NULL
)
{
...
...
@@ -3474,7 +3501,7 @@ void setExecutionContext(SQInfo *pQInfo, int32_t groupIndex, TSKEY nextKey) {
* all group belong to one result set, and each group result has different group id so set the id to be one
*/
if
(
pWindowRes
->
pos
.
pageId
==
-
1
)
{
if
(
addNewWindowResultBuf
(
pWindowRes
,
pRuntimeEnv
->
pResultBuf
,
GROUPRESULTID
,
pRuntimeEnv
->
numOfRowsPerPage
)
!=
if
(
addNewWindowResultBuf
(
pWindowRes
,
pRuntimeEnv
->
pResultBuf
,
groupIndex
,
pRuntimeEnv
->
numOfRowsPerPage
)
!=
TSDB_CODE_SUCCESS
)
{
return
;
}
...
...
@@ -4170,7 +4197,6 @@ static SFillColInfo* taosCreateFillColInfo(SQuery* pQuery) {
}
int32_t
doInitQInfo
(
SQInfo
*
pQInfo
,
STSBuf
*
pTsBuf
,
void
*
tsdb
,
int32_t
vgId
,
bool
isSTableQuery
)
{
qDebug
(
"QInfo:%p start to init qhandle"
,
pQInfo
);
SQueryRuntimeEnv
*
pRuntimeEnv
=
&
pQInfo
->
runtimeEnv
;
int32_t
code
=
TSDB_CODE_SUCCESS
;
...
...
@@ -4209,33 +4235,42 @@ int32_t doInitQInfo(SQInfo *pQInfo, STSBuf *pTsBuf, void *tsdb, int32_t vgId, bo
return
code
;
}
pRuntimeEnv
->
numOfRowsPerPage
=
getNumOfRowsInResultPage
(
pQuery
,
pRuntimeEnv
->
topBotQuery
,
isSTableQuery
);
int32_t
ps
=
DEFAULT_PAGE_SIZE
;
int32_t
rowsize
=
0
;
getIntermediateBufInfo
(
pRuntimeEnv
,
&
ps
,
&
rowsize
);
if
(
isSTableQuery
&&
!
onlyQueryTags
(
pRuntimeEnv
->
pQuery
))
{
int32_t
row
s
=
getInitialPageNum
(
pQInfo
);
code
=
createDiskbasedResultBuffer
(
&
pRuntimeEnv
->
pResultBuf
,
rows
,
pQuery
->
rowSize
,
pQInfo
);
int32_t
numOfPage
s
=
getInitialPageNum
(
pQInfo
);
code
=
createDiskbasedResultBuffer
(
&
pRuntimeEnv
->
pResultBuf
,
numOfPages
,
rowsize
,
ps
,
numOfPages
,
pQInfo
);
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
return
code
;
}
if
(
!
QUERY_IS_INTERVAL_QUERY
(
pQuery
))
{
int16_t
type
=
TSDB_DATA_TYPE_NULL
;
int32_t
threshold
=
0
;
if
(
pRuntimeEnv
->
groupbyNormalCol
)
{
// group by columns not tags;
type
=
getGroupbyColumnType
(
pQuery
,
pQuery
->
pGroupbyExpr
);
threshold
=
4000
;
}
else
{
type
=
TSDB_DATA_TYPE_INT
;
// group id
threshold
=
GET_NUM_OF_TABLEGROUP
(
pQInfo
);
if
(
threshold
<
8
)
{
threshold
=
8
;
}
}
code
=
initWindowResInfo
(
&
pRuntimeEnv
->
windowResInfo
,
pRuntimeEnv
,
32
,
4096
,
type
);
code
=
initWindowResInfo
(
&
pRuntimeEnv
->
windowResInfo
,
pRuntimeEnv
,
8
,
threshold
,
type
);
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
return
code
;
}
}
}
else
if
(
pRuntimeEnv
->
groupbyNormalCol
||
QUERY_IS_INTERVAL_QUERY
(
pQuery
))
{
int32_t
rows
=
getInitialPageNum
(
pQInfo
);
code
=
createDiskbasedResultBuffer
(
&
pRuntimeEnv
->
pResultBuf
,
rows
,
pQuery
->
rowSize
,
pQInfo
);
int32_t
numOfResultRows
=
getInitialPageNum
(
pQInfo
);
getIntermediateBufInfo
(
pRuntimeEnv
,
&
ps
,
&
rowsize
);
code
=
createDiskbasedResultBuffer
(
&
pRuntimeEnv
->
pResultBuf
,
numOfResultRows
,
rowsize
,
ps
,
numOfResultRows
,
pQInfo
);
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
return
code
;
}
...
...
@@ -4247,7 +4282,7 @@ int32_t doInitQInfo(SQInfo *pQInfo, STSBuf *pTsBuf, void *tsdb, int32_t vgId, bo
type
=
TSDB_DATA_TYPE_TIMESTAMP
;
}
code
=
initWindowResInfo
(
&
pRuntimeEnv
->
windowResInfo
,
pRuntimeEnv
,
r
ows
,
4096
,
type
);
code
=
initWindowResInfo
(
&
pRuntimeEnv
->
windowResInfo
,
pRuntimeEnv
,
numOfResultR
ows
,
4096
,
type
);
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
return
code
;
}
...
...
@@ -5705,13 +5740,18 @@ static void calResultBufSize(SQuery* pQuery) {
const
int32_t
RESULT_MSG_MIN_ROWS
=
8192
;
const
float
RESULT_THRESHOLD_RATIO
=
0
.
85
;
int32_t
numOfRes
=
RESULT_MSG_MIN_SIZE
/
pQuery
->
rowSize
;
if
(
numOfRes
<
RESULT_MSG_MIN_ROWS
)
{
numOfRes
=
RESULT_MSG_MIN_ROWS
;
}
if
(
isProjQuery
(
pQuery
))
{
int32_t
numOfRes
=
RESULT_MSG_MIN_SIZE
/
pQuery
->
rowSize
;
if
(
numOfRes
<
RESULT_MSG_MIN_ROWS
)
{
numOfRes
=
RESULT_MSG_MIN_ROWS
;
}
pQuery
->
rec
.
capacity
=
numOfRes
;
pQuery
->
rec
.
threshold
=
numOfRes
*
RESULT_THRESHOLD_RATIO
;
pQuery
->
rec
.
capacity
=
numOfRes
;
pQuery
->
rec
.
threshold
=
numOfRes
*
RESULT_THRESHOLD_RATIO
;
}
else
{
// in case of non-prj query, a smaller output buffer will be used.
pQuery
->
rec
.
capacity
=
4096
;
pQuery
->
rec
.
threshold
=
pQuery
->
rec
.
capacity
*
RESULT_THRESHOLD_RATIO
;
}
}
static
SQInfo
*
createQInfoImpl
(
SQueryTableMsg
*
pQueryMsg
,
SArray
*
pTableIdList
,
SSqlGroupbyExpr
*
pGroupbyExpr
,
SExprInfo
*
pExprs
,
...
...
@@ -5723,6 +5763,7 @@ static SQInfo *createQInfoImpl(SQueryTableMsg *pQueryMsg, SArray* pTableIdList,
if
(
pQInfo
==
NULL
)
{
goto
_cleanup_qinfo
;
}
// to make sure third party won't overwrite this structure
pQInfo
->
signature
=
pQInfo
;
pQInfo
->
tableGroupInfo
=
*
pTableGroupInfo
;
...
...
@@ -6041,11 +6082,10 @@ static void freeQInfo(SQInfo *pQInfo) {
tfree
(
pQuery
->
sdata
);
tfree
(
pQuery
);
pQInfo
->
signature
=
0
;
qDebug
(
"QInfo:%p QInfo is freed"
,
pQInfo
);
// destroy signature, in order to avoid the query process pass the object safety check
memset
(
pQInfo
,
0
,
sizeof
(
SQInfo
));
tfree
(
pQInfo
);
}
...
...
@@ -6209,7 +6249,7 @@ int32_t qCreateQueryInfo(void* tsdb, int32_t vgId, SQueryTableMsg* pQueryMsg, vo
}
int64_t
el
=
taosGetTimestampUs
()
-
st
;
qDebug
(
"qmsg:%p tag filter completed,
elapsed time:%"
PRId64
"us"
,
pQueryMsg
,
el
);
qDebug
(
"qmsg:%p tag filter completed,
numOfTables:%zu, elapsed time:%"
PRId64
"us"
,
pQueryMsg
,
tableGroupInfo
.
numOfTables
,
el
);
}
else
{
assert
(
0
);
}
...
...
src/query/src/q
FilterF
unc.c
→
src/query/src/q
filterf
unc.c
浏览文件 @
55e3352d
...
...
@@ -16,7 +16,7 @@
#define _DEFAULT_SOURCE
#include "os.h"
#include "q
E
xecutor.h"
#include "q
e
xecutor.h"
#include "taosmsg.h"
#include "tcompare.h"
#include "tsqlfunction.h"
...
...
@@ -554,5 +554,3 @@ __filter_func_t* getValueFilterFuncArray(int32_t type) {
default:
return
NULL
;
}
}
bool
supportPrefilter
(
int32_t
type
)
{
return
type
!=
TSDB_DATA_TYPE_BINARY
&&
type
!=
TSDB_DATA_TYPE_NCHAR
;
}
src/query/src/qresult
B
uf.c
→
src/query/src/qresult
b
uf.c
浏览文件 @
55e3352d
#include "qresult
B
uf.h"
#include "qresult
b
uf.h"
#include "hash.h"
#include "qextbuffer.h"
#include "taoserror.h"
#include "queryLog.h"
#include "taoserror.h"
int32_t
createDiskbasedResultBuffer
(
SDiskbasedResultBuf
**
pResultBuf
,
int32_t
numOfPages
,
int32_t
rowSize
,
int32_t
pagesize
,
int32_t
inMemPages
,
void
*
handle
)
{
int32_t
createDiskbasedResultBuffer
(
SDiskbasedResultBuf
**
pResultBuf
,
int32_t
size
,
int32_t
rowSize
,
void
*
handle
)
{
*
pResultBuf
=
calloc
(
1
,
sizeof
(
SDiskbasedResultBuf
));
SDiskbasedResultBuf
*
pResBuf
=
*
pResultBuf
;
if
(
pResBuf
==
NULL
)
{
return
TSDB_CODE_COM_OUT_OF_MEMORY
;
}
pResBuf
->
numOfRowsPerPage
=
(
DEFAULT_INTERN_BUF_PAGE_SIZE
-
sizeof
(
tFilePage
))
/
rowSize
;
pResBuf
->
numOfPages
=
size
;
pResBuf
->
totalBufSize
=
pResBuf
->
numOfPages
*
DEFAULT_INTERN_BUF_PAGE_SIZE
;
pResBuf
->
pageSize
=
pagesize
;
pResBuf
->
numOfPages
=
inMemPages
;
// all pages are in buffer in the first place
pResBuf
->
inMemPages
=
inMemPages
;
assert
(
inMemPages
<=
numOfPages
);
pResBuf
->
numOfRowsPerPage
=
(
pagesize
-
sizeof
(
tFilePage
))
/
rowSize
;
pResBuf
->
totalBufSize
=
pResBuf
->
numOfPages
*
pagesize
;
pResBuf
->
incStep
=
4
;
pResBuf
->
allocateId
=
-
1
;
pResBuf
->
iBuf
=
calloc
(
pResBuf
->
inMemPages
,
pResBuf
->
pageSize
);
// init id hash table
pResBuf
->
idsTable
=
taosHashInit
(
size
,
taosGetDefaultHashFunction
(
TSDB_DATA_TYPE_INT
),
false
);
pResBuf
->
list
=
taosArrayInit
(
size
,
POINTER_BYTES
);
pResBuf
->
idsTable
=
taosHashInit
(
numOfPages
,
taosGetDefaultHashFunction
(
TSDB_DATA_TYPE_INT
),
false
);
pResBuf
->
list
=
taosArrayInit
(
numOfPages
,
POINTER_BYTES
);
char
path
[
4096
]
=
{
0
};
char
path
[
PATH_MAX
]
=
{
0
};
getTmpfilePath
(
"tsdb_qbuf"
,
path
);
pResBuf
->
path
=
strdup
(
path
);
pResBuf
->
fd
=
open
(
pResBuf
->
path
,
O_CREAT
|
O_RDWR
,
0666
);
if
(
!
FD_VALID
(
pResBuf
->
fd
))
{
qError
(
"failed to create tmp file: %s on disk. %s"
,
pResBuf
->
path
,
strerror
(
errno
));
pResBuf
->
fd
=
FD_INITIALIZER
;
pResBuf
->
pBuf
=
NULL
;
qDebug
(
"QInfo:%p create resBuf for output, page size:%d, initial pages:%d, %"
PRId64
"bytes"
,
handle
,
pResBuf
->
pageSize
,
pResBuf
->
numOfPages
,
pResBuf
->
totalBufSize
);
return
TSDB_CODE_SUCCESS
;
}
int32_t
getNumOfResultBufGroupId
(
SDiskbasedResultBuf
*
pResultBuf
)
{
return
taosHashGetSize
(
pResultBuf
->
idsTable
);
}
int32_t
getResBufSize
(
SDiskbasedResultBuf
*
pResultBuf
)
{
return
pResultBuf
->
totalBufSize
;
}
#define NUM_OF_PAGES_ON_DISK(_r) ((_r)->numOfPages - (_r)->inMemPages)
#define FILE_SIZE_ON_DISK(_r) (NUM_OF_PAGES_ON_DISK(_r) * (_r)->pageSize)
static
int32_t
createDiskResidesBuf
(
SDiskbasedResultBuf
*
pResultBuf
)
{
pResultBuf
->
fd
=
open
(
pResultBuf
->
path
,
O_CREAT
|
O_RDWR
,
0666
);
if
(
!
FD_VALID
(
pResultBuf
->
fd
))
{
qError
(
"failed to create tmp file: %s on disk. %s"
,
pResultBuf
->
path
,
strerror
(
errno
));
return
TAOS_SYSTEM_ERROR
(
errno
);
}
int32_t
ret
=
ftruncate
(
pResBuf
->
fd
,
pResBuf
->
numOfPages
*
DEFAULT_INTERN_BUF_PAGE_SIZE
);
assert
(
pResultBuf
->
numOfPages
==
pResultBuf
->
inMemPages
);
pResultBuf
->
numOfPages
+=
pResultBuf
->
incStep
;
int32_t
ret
=
ftruncate
(
pResultBuf
->
fd
,
NUM_OF_PAGES_ON_DISK
(
pResultBuf
)
*
pResultBuf
->
pageSize
);
if
(
ret
!=
TSDB_CODE_SUCCESS
)
{
qError
(
"failed to create tmp file: %s on disk. %s"
,
pResBuf
->
path
,
strerror
(
errno
));
qError
(
"failed to create tmp file: %s on disk. %s"
,
pRes
ult
Buf
->
path
,
strerror
(
errno
));
return
TAOS_SYSTEM_ERROR
(
errno
);
}
pRes
Buf
->
pBuf
=
mmap
(
NULL
,
pResBuf
->
totalBufSize
,
PROT_READ
|
PROT_WRITE
,
MAP_SHARED
,
pRes
Buf
->
fd
,
0
);
if
(
pResBuf
->
pBuf
==
MAP_FAILED
)
{
qError
(
"QInfo:%p failed to map temp file: %s. %s"
,
handle
,
pRes
Buf
->
path
,
strerror
(
errno
));
pRes
ultBuf
->
pBuf
=
mmap
(
NULL
,
FILE_SIZE_ON_DISK
(
pResultBuf
),
PROT_READ
|
PROT_WRITE
,
MAP_SHARED
,
pResult
Buf
->
fd
,
0
);
if
(
pRes
ult
Buf
->
pBuf
==
MAP_FAILED
)
{
qError
(
"QInfo:%p failed to map temp file: %s. %s"
,
pResultBuf
->
handle
,
pResult
Buf
->
path
,
strerror
(
errno
));
return
TAOS_SYSTEM_ERROR
(
errno
);
}
qDebug
(
"QInfo:%p create tmp file for output result:%s, %"
PRId64
"bytes"
,
handle
,
pResBuf
->
path
,
pResBuf
->
totalBufSize
);
pResultBuf
->
totalBufSize
=
pResultBuf
->
numOfPages
*
pResultBuf
->
pageSize
;
return
TSDB_CODE_SUCCESS
;
}
int32_t
getNumOfResultBufGroupId
(
SDiskbasedResultBuf
*
pResultBuf
)
{
return
taosHashGetSize
(
pResultBuf
->
idsTable
);
}
static
int32_t
extendDiskFileSize
(
SDiskbasedResultBuf
*
pResultBuf
,
int32_t
incNumOfPages
)
{
assert
(
pResultBuf
->
numOfPages
*
pResultBuf
->
pageSize
==
pResultBuf
->
totalBufSize
);
int32_t
ret
=
TSDB_CODE_SUCCESS
;
int32_t
getResBufSize
(
SDiskbasedResultBuf
*
pResultBuf
)
{
return
pResultBuf
->
totalBufSize
;
}
if
(
pResultBuf
->
pBuf
==
NULL
)
{
assert
(
pResultBuf
->
fd
==
FD_INITIALIZER
);
static
int32_t
extendDiskFileSize
(
SDiskbasedResultBuf
*
pResultBuf
,
int32_t
numOfPages
)
{
assert
(
pResultBuf
->
numOfPages
*
DEFAULT_INTERN_BUF_PAGE_SIZE
==
pResultBuf
->
totalBufSize
);
int32_t
ret
=
munmap
(
pResultBuf
->
pBuf
,
pResultBuf
->
totalBufSize
);
pResultBuf
->
numOfPages
+=
numOfPages
;
/*
* disk-based output buffer is exhausted, try to extend the disk-based buffer, the available disk space may
* be insufficient
*/
ret
=
ftruncate
(
pResultBuf
->
fd
,
pResultBuf
->
numOfPages
*
DEFAULT_INTERN_BUF_PAGE_SIZE
);
if
(
ret
!=
0
)
{
// dError("QInfo:%p failed to create intermediate result output file:%s. %s", pQInfo, pSupporter->extBufFile,
// strerror(errno));
return
TSDB_CODE_QRY_NO_DISKSPACE
;
}
if
((
ret
=
createDiskResidesBuf
(
pResultBuf
))
!=
TSDB_CODE_SUCCESS
)
{
return
ret
;
}
}
else
{
ret
=
munmap
(
pResultBuf
->
pBuf
,
FILE_SIZE_ON_DISK
(
pResultBuf
));
pResultBuf
->
numOfPages
+=
incNumOfPages
;
/*
* disk-based output buffer is exhausted, try to extend the disk-based buffer, the available disk space may
* be insufficient
*/
ret
=
ftruncate
(
pResultBuf
->
fd
,
NUM_OF_PAGES_ON_DISK
(
pResultBuf
)
*
pResultBuf
->
pageSize
);
if
(
ret
!=
TSDB_CODE_SUCCESS
)
{
// dError("QInfo:%p failed to create intermediate result output file:%s. %s", pQInfo, pSupporter->extBufFile,
// strerror(errno));
return
TSDB_CODE_QRY_NO_DISKSPACE
;
}
pResultBuf
->
totalBufSize
=
pResultBuf
->
numOfPages
*
DEFAULT_INTERN_BUF_PAGE_SIZE
;
pResultBuf
->
pBuf
=
mmap
(
NULL
,
pResultBuf
->
totalBufSize
,
PROT_READ
|
PROT_WRITE
,
MAP_SHARED
,
pResultBuf
->
fd
,
0
);
pResultBuf
->
totalBufSize
=
pResultBuf
->
numOfPages
*
pResultBuf
->
pageSize
;
pResultBuf
->
pBuf
=
mmap
(
NULL
,
FILE_SIZE_ON_DISK
(
pResultBuf
)
,
PROT_READ
|
PROT_WRITE
,
MAP_SHARED
,
pResultBuf
->
fd
,
0
);
if
(
pResultBuf
->
pBuf
==
MAP_FAILED
)
{
// dError("QInfo:%p failed to map temp file: %s. %s", pQInfo, pSupporter->extBufFile, strerror(errno));
return
TSDB_CODE_QRY_OUT_OF_MEMORY
;
if
(
pResultBuf
->
pBuf
==
MAP_FAILED
)
{
// dError("QInfo:%p failed to map temp file: %s. %s", pQInfo, pSupporter->extBufFile, strerror(errno));
return
TSDB_CODE_QRY_OUT_OF_MEMORY
;
}
}
return
TSDB_CODE_SUCCESS
;
}
static
FORCE_INLINE
bool
noMoreAvailablePages
(
SDiskbasedResultBuf
*
pResultBuf
)
{
return
(
pResultBuf
->
allocateId
==
pResultBuf
->
numOfPages
-
1
);
}
#define NO_AVAILABLE_PAGES(_b) ((_b)->allocateId == (_b)->numOfPages - 1)
static
FORCE_INLINE
int32_t
getGroupIndex
(
SDiskbasedResultBuf
*
pResultBuf
,
int32_t
groupId
)
{
assert
(
pResultBuf
!=
NULL
);
...
...
@@ -121,20 +151,19 @@ static void registerPageId(SDiskbasedResultBuf* pResultBuf, int32_t groupId, int
}
tFilePage
*
getNewDataBuf
(
SDiskbasedResultBuf
*
pResultBuf
,
int32_t
groupId
,
int32_t
*
pageId
)
{
if
(
noMoreAvailablePages
(
pResultBuf
))
{
if
(
NO_AVAILABLE_PAGES
(
pResultBuf
))
{
if
(
extendDiskFileSize
(
pResultBuf
,
pResultBuf
->
incStep
)
!=
TSDB_CODE_SUCCESS
)
{
return
NULL
;
}
}
// register new id in this group
*
pageId
=
(
pResultBuf
->
allocateId
++
);
*
pageId
=
(
++
pResultBuf
->
allocateId
);
registerPageId
(
pResultBuf
,
groupId
,
*
pageId
);
tFilePage
*
page
=
GET_RES_BUF_PAGE_BY_ID
(
pResultBuf
,
*
pageId
);
// clear memory for the new page
memset
(
page
,
0
,
DEFAULT_INTERN_BUF_PAGE_SIZE
);
tFilePage
*
page
=
getResBufPage
(
pResultBuf
,
*
pageId
);
memset
(
page
,
0
,
pResultBuf
->
pageSize
);
return
page
;
}
...
...
@@ -156,13 +185,18 @@ void destroyResultBuf(SDiskbasedResultBuf* pResultBuf, void* handle) {
}
if
(
FD_VALID
(
pResultBuf
->
fd
))
{
qDebug
(
"QInfo:%p disk-based output buffer closed, total:%"
PRId64
" bytes, file created:%s, file size:%d"
,
handle
,
pResultBuf
->
totalBufSize
,
pResultBuf
->
path
,
FILE_SIZE_ON_DISK
(
pResultBuf
));
close
(
pResultBuf
->
fd
);
munmap
(
pResultBuf
->
pBuf
,
FILE_SIZE_ON_DISK
(
pResultBuf
));
pResultBuf
->
pBuf
=
NULL
;
}
else
{
qDebug
(
"QInfo:%p disk-based output buffer closed, total:%"
PRId64
" bytes, no file created"
,
handle
,
pResultBuf
->
totalBufSize
);
}
qDebug
(
"QInfo:%p disk-based output buffer closed, %"
PRId64
" bytes, file:%s"
,
handle
,
pResultBuf
->
totalBufSize
,
pResultBuf
->
path
);
munmap
(
pResultBuf
->
pBuf
,
pResultBuf
->
totalBufSize
);
unlink
(
pResultBuf
->
path
);
tfree
(
pResultBuf
->
path
);
size_t
size
=
taosArrayGetSize
(
pResultBuf
->
list
);
...
...
src/query/src/q
U
til.c
→
src/query/src/q
u
til.c
浏览文件 @
55e3352d
...
...
@@ -19,19 +19,17 @@
#include "taosmsg.h"
#include "ttime.h"
#include "q
E
xecutor.h"
#include "q
U
til.h"
#include "q
e
xecutor.h"
#include "q
u
til.h"
int32_t
getOutputInterResultBufSize
(
SQuery
*
pQuery
)
{
int32_t
size
=
0
;
for
(
int32_t
i
=
0
;
i
<
pQuery
->
numOfOutput
;
++
i
)
{
assert
(
pQuery
->
pSelectExpr
[
i
].
interBytes
<=
DEFAULT_INTERN_BUF_PAGE_SIZE
);
size
+=
pQuery
->
pSelectExpr
[
i
].
interBytes
;
}
assert
(
size
>
0
);
return
size
;
}
...
...
@@ -243,7 +241,7 @@ void clearTimeWindowResBuf(SQueryRuntimeEnv *pRuntimeEnv, SWindowResult *pWindow
size_t
size
=
pRuntimeEnv
->
pQuery
->
pSelectExpr
[
i
].
bytes
;
memset
(
s
,
0
,
size
);
resetResultInfo
(
pResultInfo
);
RESET_RESULT_INFO
(
pResultInfo
);
}
pWindowRes
->
numOfRows
=
0
;
...
...
src/query/tests/resultBufferTest.cpp
浏览文件 @
55e3352d
...
...
@@ -2,8 +2,8 @@
#include <cassert>
#include <iostream>
#include "qresultbuf.h"
#include "taos.h"
#include "qresultBuf.h"
#include "tsdb.h"
namespace
{
...
...
src/tsdb/src/tsdbRead.c
浏览文件 @
55e3352d
...
...
@@ -1376,7 +1376,7 @@ static int32_t createDataBlocksInfo(STsdbQueryHandle* pQueryHandle, int32_t numO
* }
*/
tsdbDebug
(
"%p %d data blocks sort completed
"
,
pQueryHandle
,
cnt
);
tsdbDebug
(
"%p %d data blocks sort completed
, %p"
,
pQueryHandle
,
cnt
,
pQueryHandle
->
qinfo
);
cleanBlockOrderSupporter
(
&
sup
,
numOfTables
);
free
(
pTree
);
...
...
src/util/src/tutil.c
浏览文件 @
55e3352d
...
...
@@ -522,7 +522,7 @@ int32_t taosFileRename(char *fullPath, char *suffix, char delimiter, char **dstP
void
getTmpfilePath
(
const
char
*
fileNamePrefix
,
char
*
dstPath
)
{
const
char
*
tdengineTmpFileNamePrefix
=
"tdengine-"
;
char
tmpPath
[
PATH_MAX
]
=
{
0
}
;
char
tmpPath
[
PATH_MAX
];
#ifdef WINDOWS
char
*
tmpDir
=
getenv
(
"tmp"
);
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录