Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
0468b12d
T
TDengine
项目概览
taosdata
/
TDengine
12 个月 前同步成功
通知
1180
Star
22014
Fork
4786
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
1
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
T
TDengine
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
1
Issue
1
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
未验证
提交
0468b12d
编写于
4月 25, 2022
作者:
H
Haojun Liao
提交者:
GitHub
4月 25, 2022
浏览文件
操作
浏览文件
下载
差异文件
Merge pull request #11839 from taosdata/feature/3.0_liaohj
refactor(query): do some internal refactor.
上级
2c0d6ef0
57b49262
变更
2
显示空白变更内容
内联
并排
Showing
2 changed file
with
50 addition
and
77 deletion
+50
-77
source/libs/executor/inc/executorimpl.h
source/libs/executor/inc/executorimpl.h
+2
-2
source/libs/executor/src/executorimpl.c
source/libs/executor/src/executorimpl.c
+48
-75
未找到文件。
source/libs/executor/inc/executorimpl.h
浏览文件 @
0468b12d
...
...
@@ -399,8 +399,8 @@ typedef struct SOptrBasicInfo {
// TODO move the resultrowsiz together with SOptrBasicInfo:rowCellInfoOffset
typedef
struct
SAggSupporter
{
SHashObj
*
pResultRowHashTable
;
// quick locate the window object for each result
SHashObj
*
pResultRowListSet
;
// used to check if current ResultRowInfo has ResultRow object or not
SArray
*
pResultRowArrayList
;
// The array list that contains the Result rows
//
SHashObj* pResultRowListSet; // used to check if current ResultRowInfo has ResultRow object or not
//
SArray* pResultRowArrayList; // The array list that contains the Result rows
char
*
keyBuf
;
// window key buffer
SDiskbasedBuf
*
pResultBuf
;
// query result buffer based on blocked-wised disk file
int32_t
resultRowSize
;
// the result buffer size for each result row, with the meta data size for each row
...
...
source/libs/executor/src/executorimpl.c
浏览文件 @
0468b12d
...
...
@@ -324,11 +324,16 @@ static void prepareResultListBuffer(SResultRowInfo* pResultRowInfo, jmp_buf env)
newCapacity
=
(
int64_t
)(
pResultRowInfo
->
capacity
*
1
.
5
);
}
if
(
newCapacity
=
=
pResultRowInfo
->
capacity
)
{
if
(
newCapacity
<
=
pResultRowInfo
->
capacity
)
{
newCapacity
+=
4
;
}
pResultRowInfo
->
pPosition
=
taosMemoryRealloc
(
pResultRowInfo
->
pPosition
,
newCapacity
*
sizeof
(
SResultRowPosition
));
char
*
p
=
taosMemoryRealloc
(
pResultRowInfo
->
pPosition
,
newCapacity
*
sizeof
(
SResultRowPosition
));
if
(
p
==
NULL
)
{
longjmp
(
env
,
TSDB_CODE_OUT_OF_MEMORY
);
}
pResultRowInfo
->
pPosition
=
(
SResultRowPosition
*
)
p
;
int32_t
inc
=
(
int32_t
)
newCapacity
-
pResultRowInfo
->
capacity
;
memset
(
&
pResultRowInfo
->
pPosition
[
pResultRowInfo
->
capacity
],
0
,
sizeof
(
SResultRowPosition
)
*
inc
);
...
...
@@ -419,55 +424,33 @@ SResultRow* getNewResultRow_rv(SDiskbasedBuf* pResultBuf, int64_t tableGroupId,
* | 8 bytes | actual length |
* +----------+---------------+
*/
static
SResultRow
*
doSetResultOutBufByKey
_rv
(
SDiskbasedBuf
*
pResultBuf
,
SResultRowInfo
*
pResultRowInfo
,
int64_t
uid
,
static
SResultRow
*
doSetResultOutBufByKey
(
SDiskbasedBuf
*
pResultBuf
,
SResultRowInfo
*
pResultRowInfo
,
int64_t
uid
,
char
*
pData
,
int16_t
bytes
,
bool
masterscan
,
uint64_t
groupId
,
SExecTaskInfo
*
pTaskInfo
,
bool
isIntervalQuery
,
SAggSupporter
*
pSup
)
{
bool
existInCurrentResusltRowInfo
=
false
;
SET_RES_WINDOW_KEY
(
pSup
->
keyBuf
,
pData
,
bytes
,
groupId
);
SResultRowPosition
*
p1
=
(
SResultRowPosition
*
)
taosHashGet
(
pSup
->
pResultRowHashTable
,
pSup
->
keyBuf
,
GET_RES_WINDOW_KEY_LEN
(
bytes
));
SResultRow
*
pResult
=
NULL
;
// in case of repeat scan/reverse scan, no new time window added.
if
(
isIntervalQuery
)
{
if
(
!
masterscan
)
{
// the *p1 may be NULL in case of sliding+offset exists.
if
(
p1
!=
NULL
)
{
return
getResultRowByPos
(
pResultBuf
,
p1
);
}
else
{
return
NULL
;
}
}
if
(
p1
!=
NULL
)
{
if
(
pResultRowInfo
->
size
==
0
)
{
existInCurrentResusltRowInfo
=
false
;
// this time window created by other timestamp that does not belongs to current table.
}
else
if
(
pResultRowInfo
->
size
==
1
)
{
SResultRowPosition
*
p
=
&
pResultRowInfo
->
pPosition
[
0
];
existInCurrentResusltRowInfo
=
(
p
->
pageId
==
p1
->
pageId
&&
p
->
offset
==
p1
->
offset
);
}
else
{
// check if current pResultRowInfo contains the existInCurrentResusltRowInfo pResultRow
SET_RES_EXT_WINDOW_KEY
(
pSup
->
keyBuf
,
pData
,
bytes
,
uid
,
pResultRowInfo
);
int64_t
*
index
=
taosHashGet
(
pSup
->
pResultRowListSet
,
pSup
->
keyBuf
,
GET_RES_EXT_WINDOW_KEY_LEN
(
bytes
));
if
(
index
!=
NULL
)
{
// TODO check the scan order for current opened time window
existInCurrentResusltRowInfo
=
true
;
}
else
{
existInCurrentResusltRowInfo
=
false
;
}
}
if
(
masterscan
&&
p1
!=
NULL
)
{
// the *p1 may be NULL in case of sliding+offset exists.
pResult
=
getResultRowByPos
(
pResultBuf
,
p1
);
}
}
else
{
// In case of group by column query, the required SResultRow object must be existInCurrentResusltRowInfo in the
// pResultRowInfo object.
if
(
p1
!=
NULL
)
{
return
getResultRowByPos
(
pResultBuf
,
p1
);
pResult
=
getResultRowByPos
(
pResultBuf
,
p1
);
}
}
SResultRow
*
pResult
=
NULL
;
if
(
!
existInCurrentResusltRowInfo
)
{
// 1. close current opened time window
if
(
pResultRowInfo
->
cur
.
pageId
!=
-
1
)
{
// todo extract function
if
(
pResultRowInfo
->
cur
.
pageId
!=
-
1
&&
((
pResult
==
NULL
)
||
(
pResult
->
pageId
!=
pResultRowInfo
->
cur
.
pageId
&&
pResult
->
offset
!=
pResultRowInfo
->
cur
.
offset
)))
{
// todo extract function
SResultRowPosition
pos
=
pResultRowInfo
->
cur
;
SFilePage
*
pPage
=
getBufPage
(
pResultBuf
,
pos
.
pageId
);
SResultRow
*
pRow
=
(
SResultRow
*
)((
char
*
)
pPage
+
pos
.
offset
);
...
...
@@ -475,31 +458,21 @@ static SResultRow* doSetResultOutBufByKey_rv(SDiskbasedBuf* pResultBuf, SResultR
releaseBufPage
(
pResultBuf
,
pPage
);
}
// allocate a new buffer page
prepareResultListBuffer
(
pResultRowInfo
,
pTaskInfo
->
env
);
if
(
p1
==
NULL
)
{
if
(
pResult
==
NULL
)
{
pResult
=
getNewResultRow_rv
(
pResultBuf
,
groupId
,
pSup
->
resultRowSize
);
initResultRow
(
pResult
);
// add a new result set for a new group
SResultRowPosition
pos
=
{.
pageId
=
pResult
->
pageId
,
.
offset
=
pResult
->
offset
};
taosHashPut
(
pSup
->
pResultRowHashTable
,
pSup
->
keyBuf
,
GET_RES_WINDOW_KEY_LEN
(
bytes
),
&
pos
,
sizeof
(
SResultRowPosition
));
SResultRowCell
cell
=
{.
groupId
=
groupId
,
.
pos
=
pos
};
taosArrayPush
(
pSup
->
pResultRowArrayList
,
&
cell
);
}
else
{
pResult
=
getResultRowByPos
(
pResultBuf
,
p1
);
taosHashPut
(
pSup
->
pResultRowHashTable
,
pSup
->
keyBuf
,
GET_RES_WINDOW_KEY_LEN
(
bytes
),
&
pos
,
sizeof
(
SResultRowPosition
));
}
// 2. set the new time window to be the new active time window
pResultRowInfo
->
pPosition
[
pResultRowInfo
->
size
++
]
=
(
SResultRowPosition
){.
pageId
=
pResult
->
pageId
,
.
offset
=
pResult
->
offset
};
pResultRowInfo
->
cur
=
(
SResultRowPosition
){.
pageId
=
pResult
->
pageId
,
.
offset
=
pResult
->
offset
};
SET_RES_EXT_WINDOW_KEY
(
pSup
->
keyBuf
,
pData
,
bytes
,
uid
,
pResultRowInfo
);
taosHashPut
(
pSup
->
pResultRowListSet
,
pSup
->
keyBuf
,
GET_RES_EXT_WINDOW_KEY_LEN
(
bytes
),
&
pResultRowInfo
->
cur
,
POINTER_BYTES
);
}
else
{
pResult
=
getResultRowByPos
(
pResultBuf
,
p1
);
}
// too many time window in query
if
(
pResultRowInfo
->
size
>
MAX_INTERVAL_TIME_WINDOW
)
{
...
...
@@ -656,7 +629,7 @@ static int32_t setResultOutputBufByKey_rv(SResultRowInfo* pResultRowInfo, int64_
int32_t
numOfOutput
,
int32_t
*
rowCellInfoOffset
,
SAggSupporter
*
pAggSup
,
SExecTaskInfo
*
pTaskInfo
)
{
assert
(
win
->
skey
<=
win
->
ekey
);
SResultRow
*
pResultRow
=
doSetResultOutBufByKey
_rv
(
pAggSup
->
pResultBuf
,
pResultRowInfo
,
id
,
(
char
*
)
&
win
->
skey
,
SResultRow
*
pResultRow
=
doSetResultOutBufByKey
(
pAggSup
->
pResultBuf
,
pResultRowInfo
,
id
,
(
char
*
)
&
win
->
skey
,
TSDB_KEYSIZE
,
masterscan
,
tableGroupId
,
pTaskInfo
,
true
,
pAggSup
);
if
(
pResultRow
==
NULL
)
{
...
...
@@ -1708,7 +1681,7 @@ int32_t setGroupResultOutputBuf(SOptrBasicInfo* binfo, int32_t numOfCols, char*
SResultRowInfo
*
pResultRowInfo
=
&
binfo
->
resultRowInfo
;
SqlFunctionCtx
*
pCtx
=
binfo
->
pCtx
;
SResultRow
*
pResultRow
=
doSetResultOutBufByKey
_rv
(
pBuf
,
pResultRowInfo
,
groupId
,
(
char
*
)
pData
,
bytes
,
true
,
groupId
,
SResultRow
*
pResultRow
=
doSetResultOutBufByKey
(
pBuf
,
pResultRowInfo
,
groupId
,
(
char
*
)
pData
,
bytes
,
true
,
groupId
,
pTaskInfo
,
false
,
pAggSup
);
assert
(
pResultRow
!=
NULL
);
...
...
@@ -2714,7 +2687,7 @@ void setFunctionResultOutput(SOptrBasicInfo* pInfo, SAggSupporter* pSup, int32_t
int64_t
tid
=
0
;
int64_t
groupId
=
0
;
SResultRow
*
pRow
=
doSetResultOutBufByKey
_rv
(
pSup
->
pResultBuf
,
pResultRowInfo
,
tid
,
(
char
*
)
&
tid
,
sizeof
(
tid
),
true
,
SResultRow
*
pRow
=
doSetResultOutBufByKey
(
pSup
->
pResultBuf
,
pResultRowInfo
,
tid
,
(
char
*
)
&
tid
,
sizeof
(
tid
),
true
,
groupId
,
pTaskInfo
,
false
,
pSup
);
for
(
int32_t
i
=
0
;
i
<
pDataBlock
->
info
.
numOfCols
;
++
i
)
{
...
...
@@ -3002,7 +2975,7 @@ void doSetTableGroupOutputBuf(SAggOperatorInfo* pAggInfo, int32_t numOfOutput, u
int32_t
*
rowCellInfoOffset
=
pAggInfo
->
binfo
.
rowCellInfoOffset
;
SResultRow
*
pResultRow
=
doSetResultOutBufByKey
_rv
(
pAggInfo
->
aggSup
.
pResultBuf
,
pResultRowInfo
,
uid
,
(
char
*
)
&
groupId
,
sizeof
(
groupId
),
doSetResultOutBufByKey
(
pAggInfo
->
aggSup
.
pResultBuf
,
pResultRowInfo
,
uid
,
(
char
*
)
&
groupId
,
sizeof
(
groupId
),
true
,
groupId
,
pTaskInfo
,
false
,
&
pAggInfo
->
aggSup
);
assert
(
pResultRow
!=
NULL
);
...
...
@@ -4899,8 +4872,8 @@ bool aggDecodeResultRow(SOperatorInfo* pOperator, SAggSupporter* pSup, SOptrBasi
initResultRow
(
resultRow
);
prepareResultListBuffer
(
&
pInfo
->
resultRowInfo
,
pOperator
->
pTaskInfo
->
env
);
// pInfo->resultRowInfo.cur = pInfo->resultRowInfo.size;
pInfo
->
resultRowInfo
.
pPosition
[
pInfo
->
resultRowInfo
.
size
++
]
=
(
SResultRowPosition
){.
pageId
=
resultRow
->
pageId
,
.
offset
=
resultRow
->
offset
};
//
pInfo->resultRowInfo.pPosition[pInfo->resultRowInfo.size++] =
//
(SResultRowPosition){.pageId = resultRow->pageId, .offset = resultRow->offset};
pInfo
->
resultRowInfo
.
cur
=
(
SResultRowPosition
){.
pageId
=
resultRow
->
pageId
,
.
offset
=
resultRow
->
offset
};
}
...
...
@@ -5613,10 +5586,10 @@ int32_t doInitAggInfoSup(SAggSupporter* pAggSup, SqlFunctionCtx* pCtx, int32_t n
pAggSup
->
resultRowSize
=
getResultRowSize
(
pCtx
,
numOfOutput
);
pAggSup
->
keyBuf
=
taosMemoryCalloc
(
1
,
keyBufSize
+
POINTER_BYTES
+
sizeof
(
int64_t
));
pAggSup
->
pResultRowHashTable
=
taosHashInit
(
10
,
hashFn
,
true
,
HASH_NO_LOCK
);
pAggSup
->
pResultRowListSet
=
taosHashInit
(
100
,
hashFn
,
false
,
HASH_NO_LOCK
);
pAggSup
->
pResultRowArrayList
=
taosArrayInit
(
10
,
sizeof
(
SResultRowCell
));
//
pAggSup->pResultRowListSet = taosHashInit(100, hashFn, false, HASH_NO_LOCK);
//
pAggSup->pResultRowArrayList = taosArrayInit(10, sizeof(SResultRowCell));
if
(
pAggSup
->
keyBuf
==
NULL
||
pAggSup
->
pResultRowArrayList
==
NULL
||
pAggSup
->
pResultRowListSet
==
NULL
||
if
(
pAggSup
->
keyBuf
==
NULL
/*|| pAggSup->pResultRowArrayList == NULL || pAggSup->pResultRowListSet == NULL*/
||
pAggSup
->
pResultRowHashTable
==
NULL
)
{
return
TSDB_CODE_OUT_OF_MEMORY
;
}
...
...
@@ -5632,8 +5605,8 @@ int32_t doInitAggInfoSup(SAggSupporter* pAggSup, SqlFunctionCtx* pCtx, int32_t n
static
void
cleanupAggSup
(
SAggSupporter
*
pAggSup
)
{
taosMemoryFreeClear
(
pAggSup
->
keyBuf
);
taosHashCleanup
(
pAggSup
->
pResultRowHashTable
);
taosHashCleanup
(
pAggSup
->
pResultRowListSet
);
taosArrayDestroy
(
pAggSup
->
pResultRowArrayList
);
//
taosHashCleanup(pAggSup->pResultRowListSet);
//
taosArrayDestroy(pAggSup->pResultRowArrayList);
destroyDiskbasedBuf
(
pAggSup
->
pResultBuf
);
}
...
...
@@ -6356,7 +6329,7 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
int32_t
numOfCols
=
0
;
tsdbReaderT
pDataReader
=
doCreateDataReader
(
pTableScanNode
,
pHandle
,
pTableGroupInfo
,
(
uint64_t
)
queryId
,
taskId
);
if
(
pDataReader
==
NULL
)
{
if
(
pDataReader
==
NULL
&&
terrno
!=
0
)
{
return
NULL
;
}
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录