Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
电商架构修炼
TDengine
提交
d31c7033
TDengine
项目概览
电商架构修炼
/
TDengine
与 Fork 源项目一致
Fork自
taosdata / TDengine
通知
1
Star
1
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
1
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
TDengine
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
1
Issue
1
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
d31c7033
编写于
9月 06, 2021
作者:
H
Haojun Liao
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
[td-255] code refactor.
上级
261bfd00
变更
3
隐藏空白更改
内联
并排
Showing
3 changed file
with
69 addition
and
81 deletion
+69
-81
src/client/inc/tscUtil.h
src/client/inc/tscUtil.h
+1
-0
src/client/src/tscSubquery.c
src/client/src/tscSubquery.c
+51
-71
src/client/src/tscUtil.c
src/client/src/tscUtil.c
+17
-10
未找到文件。
src/client/inc/tscUtil.h
浏览文件 @
d31c7033
...
...
@@ -318,6 +318,7 @@ void doAddGroupColumnForSubquery(SQueryInfo* pQueryInfo, int32_t tagIndex, SSqlC
int16_t
tscGetJoinTagColIdByUid
(
STagCond
*
pTagCond
,
uint64_t
uid
);
int16_t
tscGetTagColIndexById
(
STableMeta
*
pTableMeta
,
int16_t
colId
);
int32_t
doInitSubState
(
SSqlObj
*
pSql
,
int32_t
numOfSubqueries
);
void
tscPrintSelNodeList
(
SSqlObj
*
pSql
,
int32_t
subClauseIndex
);
...
...
src/client/src/tscSubquery.c
浏览文件 @
d31c7033
...
...
@@ -2459,11 +2459,48 @@ static void doSendQueryReqs(SSchedMsg* pSchedMsg) {
tfree
(
p
);
}
static
void
doConcurrentlySendSubQueries
(
SSqlObj
*
pSql
)
{
SSubqueryState
*
pState
=
&
pSql
->
subState
;
// concurrently sent the query requests.
const
int32_t
MAX_REQUEST_PER_TASK
=
8
;
int32_t
numOfTasks
=
(
pState
->
numOfSub
+
MAX_REQUEST_PER_TASK
-
1
)
/
MAX_REQUEST_PER_TASK
;
assert
(
numOfTasks
>=
1
);
int32_t
num
;
if
(
pState
->
numOfSub
/
numOfTasks
==
MAX_REQUEST_PER_TASK
)
{
num
=
MAX_REQUEST_PER_TASK
;
}
else
{
num
=
pState
->
numOfSub
/
numOfTasks
+
1
;
}
tscDebug
(
"0x%"
PRIx64
" query will be sent by %d threads"
,
pSql
->
self
,
numOfTasks
);
for
(
int32_t
j
=
0
;
j
<
numOfTasks
;
++
j
)
{
SSchedMsg
schedMsg
=
{
0
};
schedMsg
.
fp
=
doSendQueryReqs
;
schedMsg
.
ahandle
=
(
void
*
)
pSql
;
schedMsg
.
thandle
=
NULL
;
SPair
*
p
=
calloc
(
1
,
sizeof
(
SPair
));
p
->
first
=
j
*
num
;
if
(
j
==
numOfTasks
-
1
)
{
p
->
second
=
pState
->
numOfSub
;
}
else
{
p
->
second
=
(
j
+
1
)
*
num
;
}
schedMsg
.
msg
=
p
;
taosScheduleTask
(
tscQhandle
,
&
schedMsg
);
}
}
int32_t
tscHandleMasterSTableQuery
(
SSqlObj
*
pSql
)
{
SSqlRes
*
pRes
=
&
pSql
->
res
;
SSqlCmd
*
pCmd
=
&
pSql
->
cmd
;
// pRes->code check only serves in launching
metric
sub-queries
// pRes->code check only serves in launching
super table
sub-queries
if
(
pRes
->
code
==
TSDB_CODE_TSC_QUERY_CANCELLED
)
{
pCmd
->
command
=
TSDB_SQL_RETRIEVE_GLOBALMERGE
;
// enable the abort of kill super table function.
return
pRes
->
code
;
...
...
@@ -2474,22 +2511,23 @@ int32_t tscHandleMasterSTableQuery(SSqlObj *pSql) {
pRes
->
qId
=
0x1
;
// hack the qhandle check
const
uint32_t
nBufferSize
=
(
1u
<<
18u
);
// 256KB
const
uint32_t
nBufferSize
=
(
1u
<<
18u
);
// 256KB
, default buffer size
SQueryInfo
*
pQueryInfo
=
tscGetQueryInfo
(
pCmd
);
STableMetaInfo
*
pTableMetaInfo
=
tscGetMetaInfo
(
pQueryInfo
,
0
);
SSubqueryState
*
pState
=
&
pSql
->
subState
;
pState
->
numOfSub
=
0
;
if
(
pTableMetaInfo
->
pVgroupTables
==
NULL
)
{
pState
->
numOfSub
=
pTableMetaInfo
->
vgroupList
->
numOfVgroups
;
}
else
{
pState
->
numOfSub
=
(
int32_t
)
taosArrayGetSize
(
pTableMetaInfo
->
pVgroupTables
);
int32_t
numOfSub
=
(
pTableMetaInfo
->
pVgroupTables
==
NULL
)
?
pTableMetaInfo
->
vgroupList
->
numOfVgroups
:
(
int32_t
)
taosArrayGetSize
(
pTableMetaInfo
->
pVgroupTables
);
int32_t
ret
=
doInitSubState
(
pSql
,
numOfSub
);
if
(
ret
!=
0
)
{
tscAsyncResultOnError
(
pSql
);
return
ret
;
}
assert
(
pState
->
numOfSub
>
0
);
int32_t
ret
=
tscCreateGlobalMergerEnv
(
pQueryInfo
,
&
pMemoryBuf
,
pSql
->
subState
.
numOfSub
,
&
pDesc
,
nBufferSize
,
pSql
->
self
);
ret
=
tscCreateGlobalMergerEnv
(
pQueryInfo
,
&
pMemoryBuf
,
pSql
->
subState
.
numOfSub
,
&
pDesc
,
nBufferSize
,
pSql
->
self
);
if
(
ret
!=
0
)
{
pRes
->
code
=
ret
;
tscAsyncResultOnError
(
pSql
);
...
...
@@ -2499,32 +2537,6 @@ int32_t tscHandleMasterSTableQuery(SSqlObj *pSql) {
}
tscDebug
(
"0x%"
PRIx64
" retrieved query data from %d vnode(s)"
,
pSql
->
self
,
pState
->
numOfSub
);
pSql
->
pSubs
=
calloc
(
pState
->
numOfSub
,
POINTER_BYTES
);
if
(
pSql
->
pSubs
==
NULL
)
{
tfree
(
pSql
->
pSubs
);
pRes
->
code
=
TSDB_CODE_TSC_OUT_OF_MEMORY
;
tscDestroyGlobalMergerEnv
(
pMemoryBuf
,
pDesc
,
pState
->
numOfSub
);
tscAsyncResultOnError
(
pSql
);
return
ret
;
}
if
(
pState
->
states
==
NULL
)
{
pState
->
states
=
calloc
(
pState
->
numOfSub
,
sizeof
(
*
pState
->
states
));
if
(
pState
->
states
==
NULL
)
{
pRes
->
code
=
TSDB_CODE_TSC_OUT_OF_MEMORY
;
tscDestroyGlobalMergerEnv
(
pMemoryBuf
,
pDesc
,
pState
->
numOfSub
);
tscAsyncResultOnError
(
pSql
);
return
ret
;
}
pthread_mutex_init
(
&
pState
->
mutex
,
NULL
);
}
memset
(
pState
->
states
,
0
,
sizeof
(
*
pState
->
states
)
*
pState
->
numOfSub
);
tscDebug
(
"0x%"
PRIx64
" reset all sub states to 0"
,
pSql
->
self
);
pRes
->
code
=
TSDB_CODE_SUCCESS
;
int32_t
i
=
0
;
...
...
@@ -2545,8 +2557,8 @@ int32_t tscHandleMasterSTableQuery(SSqlObj *pSql) {
break
;
}
trs
->
subqueryIndex
=
i
;
trs
->
pParentSql
=
pSql
;
trs
->
subqueryIndex
=
i
;
trs
->
pParentSql
=
pSql
;
SSqlObj
*
pNew
=
tscCreateSTableSubquery
(
pSql
,
trs
,
NULL
);
if
(
pNew
==
NULL
)
{
...
...
@@ -2582,39 +2594,7 @@ int32_t tscHandleMasterSTableQuery(SSqlObj *pSql) {
return
pRes
->
code
;
}
// concurrently sent the query requests.
const
int32_t
MAX_REQUEST_PER_TASK
=
8
;
int32_t
numOfTasks
=
(
pState
->
numOfSub
+
MAX_REQUEST_PER_TASK
-
1
)
/
MAX_REQUEST_PER_TASK
;
assert
(
numOfTasks
>=
1
);
int32_t
num
;
if
(
pState
->
numOfSub
/
numOfTasks
==
MAX_REQUEST_PER_TASK
)
{
num
=
MAX_REQUEST_PER_TASK
;
}
else
{
num
=
pState
->
numOfSub
/
numOfTasks
+
1
;
}
tscDebug
(
"0x%"
PRIx64
" query will be sent by %d threads"
,
pSql
->
self
,
numOfTasks
);
for
(
int32_t
j
=
0
;
j
<
numOfTasks
;
++
j
)
{
SSchedMsg
schedMsg
=
{
0
};
schedMsg
.
fp
=
doSendQueryReqs
;
schedMsg
.
ahandle
=
(
void
*
)
pSql
;
schedMsg
.
thandle
=
NULL
;
SPair
*
p
=
calloc
(
1
,
sizeof
(
SPair
));
p
->
first
=
j
*
num
;
if
(
j
==
numOfTasks
-
1
)
{
p
->
second
=
pState
->
numOfSub
;
}
else
{
p
->
second
=
(
j
+
1
)
*
num
;
}
schedMsg
.
msg
=
p
;
taosScheduleTask
(
tscQhandle
,
&
schedMsg
);
}
doConcurrentlySendSubQueries
(
pSql
);
return
TSDB_CODE_SUCCESS
;
}
...
...
src/client/src/tscUtil.c
浏览文件 @
d31c7033
...
...
@@ -3944,6 +3944,21 @@ static void tscSubqueryCompleteCallback(void* param, TAOS_RES* tres, int code) {
taos_fetch_rows_a
(
tres
,
tscSubqueryRetrieveCallback
,
param
);
}
int32_t
doInitSubState
(
SSqlObj
*
pSql
,
int32_t
numOfSubqueries
)
{
assert
(
pSql
->
subState
.
numOfSub
==
0
&&
pSql
->
pSubs
==
NULL
&&
pSql
->
subState
.
states
==
NULL
);
pSql
->
subState
.
numOfSub
=
numOfSubqueries
;
pSql
->
pSubs
=
calloc
(
pSql
->
subState
.
numOfSub
,
POINTER_BYTES
);
pSql
->
subState
.
states
=
calloc
(
pSql
->
subState
.
numOfSub
,
sizeof
(
int8_t
));
int32_t
code
=
pthread_mutex_init
(
&
pSql
->
subState
.
mutex
,
NULL
);
if
(
pSql
->
pSubs
==
NULL
||
pSql
->
subState
.
states
==
NULL
||
code
!=
0
)
{
return
TSDB_CODE_TSC_OUT_OF_MEMORY
;
}
return
TSDB_CODE_SUCCESS
;
}
// do execute the query according to the query execution plan
void
executeQuery
(
SSqlObj
*
pSql
,
SQueryInfo
*
pQueryInfo
)
{
int32_t
code
=
TSDB_CODE_SUCCESS
;
...
...
@@ -3959,16 +3974,8 @@ void executeQuery(SSqlObj* pSql, SQueryInfo* pQueryInfo) {
}
if
(
taosArrayGetSize
(
pQueryInfo
->
pUpstream
)
>
0
)
{
// nest query. do execute it firstly
assert
(
pSql
->
subState
.
numOfSub
==
0
);
pSql
->
subState
.
numOfSub
=
(
int32_t
)
taosArrayGetSize
(
pQueryInfo
->
pUpstream
);
assert
(
pSql
->
pSubs
==
NULL
);
pSql
->
pSubs
=
calloc
(
pSql
->
subState
.
numOfSub
,
POINTER_BYTES
);
assert
(
pSql
->
subState
.
states
==
NULL
);
pSql
->
subState
.
states
=
calloc
(
pSql
->
subState
.
numOfSub
,
sizeof
(
int8_t
));
code
=
pthread_mutex_init
(
&
pSql
->
subState
.
mutex
,
NULL
);
if
(
pSql
->
pSubs
==
NULL
||
pSql
->
subState
.
states
==
NULL
||
code
!=
TSDB_CODE_SUCCESS
)
{
code
=
TSDB_CODE_TSC_OUT_OF_MEMORY
;
code
=
doInitSubState
(
pSql
,
(
int32_t
)
taosArrayGetSize
(
pQueryInfo
->
pUpstream
));
if
(
code
!=
TSDB_CODE_SUCCESS
)
{
goto
_error
;
}
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录