Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
2dot5
ClickHouse
提交
9a2760d9
C
ClickHouse
项目概览
2dot5
/
ClickHouse
通知
3
Star
0
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
DevOps
流水线
流水线任务
计划
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
C
ClickHouse
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
DevOps
DevOps
流水线
流水线任务
计划
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
流水线任务
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
未验证
提交
9a2760d9
编写于
3月 24, 2020
作者:
A
alexey-milovidov
提交者:
GitHub
3月 24, 2020
浏览文件
操作
浏览文件
下载
差异文件
Merge pull request #9673 from azat/processors-max_distributed_connections
Fix max_distributed_connections
上级
9a41113b
550cfef2
变更
12
隐藏空白更改
内联
并排
Showing
12 changed file
with
58 addition
and
14 deletion
+58
-14
dbms/programs/server/TCPHandler.cpp
dbms/programs/server/TCPHandler.cpp
+3
-6
dbms/programs/server/TCPHandler.h
dbms/programs/server/TCPHandler.h
+1
-1
dbms/src/Interpreters/InterpreterSelectQuery.cpp
dbms/src/Interpreters/InterpreterSelectQuery.cpp
+2
-1
dbms/src/Interpreters/InterpreterSelectQuery.h
dbms/src/Interpreters/InterpreterSelectQuery.h
+9
-0
dbms/src/Interpreters/InterpreterSelectWithUnionQuery.cpp
dbms/src/Interpreters/InterpreterSelectWithUnionQuery.cpp
+14
-4
dbms/src/Interpreters/InterpreterSelectWithUnionQuery.h
dbms/src/Interpreters/InterpreterSelectWithUnionQuery.h
+2
-0
dbms/src/Interpreters/executeQuery.cpp
dbms/src/Interpreters/executeQuery.cpp
+1
-1
dbms/src/Storages/StorageDistributed.cpp
dbms/src/Storages/StorageDistributed.cpp
+1
-1
dbms/tests/queries/0_stateless/01085_max_distributed_connections.reference
...s/0_stateless/01085_max_distributed_connections.reference
+2
-0
dbms/tests/queries/0_stateless/01085_max_distributed_connections.sh
.../queries/0_stateless/01085_max_distributed_connections.sh
+13
-0
dbms/tests/queries/0_stateless/01085_max_distributed_connections_http.reference
...tateless/01085_max_distributed_connections_http.reference
+2
-0
dbms/tests/queries/0_stateless/01085_max_distributed_connections_http.sh
...ies/0_stateless/01085_max_distributed_connections_http.sh
+8
-0
未找到文件。
dbms/programs/server/TCPHandler.cpp
浏览文件 @
9a2760d9
...
...
@@ -265,7 +265,7 @@ void TCPHandler::runImpl()
state
.
io
.
onFinish
();
}
else
if
(
state
.
io
.
pipeline
.
initialized
())
processOrdinaryQueryWithProcessors
(
query_context
->
getSettingsRef
().
max_threads
);
processOrdinaryQueryWithProcessors
();
else
processOrdinaryQuery
();
...
...
@@ -544,13 +544,10 @@ void TCPHandler::processOrdinaryQuery()
}
void
TCPHandler
::
processOrdinaryQueryWithProcessors
(
size_t
num_threads
)
void
TCPHandler
::
processOrdinaryQueryWithProcessors
()
{
auto
&
pipeline
=
state
.
io
.
pipeline
;
/// Reduce the number of threads to recommended value.
num_threads
=
std
::
min
(
num_threads
,
pipeline
.
getNumThreads
());
/// Send header-block, to allow client to prepare output format for data to send.
{
auto
&
header
=
pipeline
.
getHeader
();
...
...
@@ -585,7 +582,7 @@ void TCPHandler::processOrdinaryQueryWithProcessors(size_t num_threads)
try
{
executor
->
execute
(
num_threads
);
executor
->
execute
(
pipeline
.
getNumThreads
()
);
}
catch
(...)
{
...
...
dbms/programs/server/TCPHandler.h
浏览文件 @
9a2760d9
...
...
@@ -171,7 +171,7 @@ private:
/// Process a request that does not require the receiving of data blocks from the client
void
processOrdinaryQuery
();
void
processOrdinaryQueryWithProcessors
(
size_t
num_threads
);
void
processOrdinaryQueryWithProcessors
();
void
processTablesStatusRequest
();
...
...
dbms/src/Interpreters/InterpreterSelectQuery.cpp
浏览文件 @
9a2760d9
...
...
@@ -467,7 +467,6 @@ QueryPipeline InterpreterSelectQuery::executeWithProcessors()
{
QueryPipeline
query_pipeline
;
executeImpl
(
query_pipeline
,
input
,
std
::
move
(
input_pipe
),
query_pipeline
);
query_pipeline
.
setMaxThreads
(
max_streams
);
query_pipeline
.
addInterpreterContext
(
context
);
query_pipeline
.
addStorageHolder
(
storage
);
return
query_pipeline
;
...
...
@@ -1301,6 +1300,7 @@ void InterpreterSelectQuery::executeFetchColumns(
{
is_remote
=
true
;
max_streams
=
settings
.
max_distributed_connections
;
pipeline
.
setMaxThreads
(
max_streams
);
}
UInt64
max_block_size
=
settings
.
max_block_size
;
...
...
@@ -1325,6 +1325,7 @@ void InterpreterSelectQuery::executeFetchColumns(
{
max_block_size
=
std
::
max
(
UInt64
(
1
),
limit_length
+
limit_offset
);
max_streams
=
1
;
pipeline
.
setMaxThreads
(
max_streams
);
}
if
(
!
max_block_size
)
...
...
dbms/src/Interpreters/InterpreterSelectQuery.h
浏览文件 @
9a2760d9
...
...
@@ -92,6 +92,8 @@ public:
ASTPtr
getQuery
()
const
{
return
query_ptr
;
}
size_t
getMaxStreams
()
const
{
return
max_streams
;
}
private:
InterpreterSelectQuery
(
const
ASTPtr
&
query_ptr_
,
...
...
@@ -122,6 +124,9 @@ private:
BlockInputStreamPtr
stream_with_non_joined_data
;
bool
union_stream
=
false
;
/// Cache value of InterpreterSelectQuery::max_streams
size_t
max_threads
=
1
;
BlockInputStreamPtr
&
firstStream
()
{
return
streams
.
at
(
0
);
}
template
<
typename
Transform
>
...
...
@@ -147,6 +152,10 @@ private:
bool
hasDelayedStream
()
const
{
return
stream_with_non_joined_data
!=
nullptr
;
}
bool
initialized
()
const
{
return
!
streams
.
empty
();
}
/// Compatibility with QueryPipeline (Processors)
void
setMaxThreads
(
size_t
max_threads_
)
{
max_threads
=
max_threads_
;
}
size_t
getNumThreads
()
const
{
return
max_threads
;
}
};
template
<
typename
TPipeline
>
...
...
dbms/src/Interpreters/InterpreterSelectWithUnionQuery.cpp
浏览文件 @
9a2760d9
...
...
@@ -34,7 +34,8 @@ InterpreterSelectWithUnionQuery::InterpreterSelectWithUnionQuery(
const
Names
&
required_result_column_names
)
:
options
(
options_
),
query_ptr
(
query_ptr_
),
context
(
std
::
make_shared
<
Context
>
(
context_
))
context
(
std
::
make_shared
<
Context
>
(
context_
)),
max_streams
(
context
->
getSettingsRef
().
max_threads
)
{
const
auto
&
ast
=
query_ptr
->
as
<
ASTSelectWithUnionQuery
&>
();
...
...
@@ -196,14 +197,23 @@ BlockInputStreams InterpreterSelectWithUnionQuery::executeWithMultipleStreams(Qu
parent_pipeline
.
addInterpreterContext
(
context
);
}
/// Update max_streams due to:
/// - max_distributed_connections for Distributed() engine
/// - max_streams_to_max_threads_ratio
///
/// XXX: res.pipeline.getMaxThreads() cannot be used since it is capped to
/// number of streams, which is empty for non-Processors case.
max_streams
=
(
*
std
::
min_element
(
nested_interpreters
.
begin
(),
nested_interpreters
.
end
(),
[](
const
auto
&
a
,
const
auto
&
b
)
{
return
a
->
getMaxStreams
()
<
b
->
getMaxStreams
();
}))
->
getMaxStreams
();
return
nested_streams
;
}
BlockIO
InterpreterSelectWithUnionQuery
::
execute
()
{
const
Settings
&
settings
=
context
->
getSettingsRef
();
BlockIO
res
;
BlockInputStreams
nested_streams
=
executeWithMultipleStreams
(
res
.
pipeline
);
BlockInputStreamPtr
result_stream
;
...
...
@@ -219,7 +229,7 @@ BlockIO InterpreterSelectWithUnionQuery::execute()
}
else
{
result_stream
=
std
::
make_shared
<
UnionBlockInputStream
>
(
nested_streams
,
nullptr
,
settings
.
max_thread
s
);
result_stream
=
std
::
make_shared
<
UnionBlockInputStream
>
(
nested_streams
,
nullptr
,
max_stream
s
);
nested_streams
.
clear
();
}
...
...
dbms/src/Interpreters/InterpreterSelectWithUnionQuery.h
浏览文件 @
9a2760d9
...
...
@@ -56,6 +56,8 @@ private:
Block
result_header
;
size_t
max_streams
=
1
;
static
Block
getCommonHeaderForUnion
(
const
Blocks
&
headers
);
};
...
...
dbms/src/Interpreters/executeQuery.cpp
浏览文件 @
9a2760d9
...
...
@@ -751,7 +751,7 @@ void executeQuery(
{
auto
executor
=
pipeline
.
execute
();
executor
->
execute
(
context
.
getSettingsRef
().
max_threads
);
executor
->
execute
(
pipeline
.
getNumThreads
()
);
}
}
}
...
...
dbms/src/Storages/StorageDistributed.cpp
浏览文件 @
9a2760d9
...
...
@@ -427,7 +427,7 @@ Pipes StorageDistributed::read(
if
(
!
smaller_cluster
)
{
LOG_DEBUG
(
log
,
"Reading from "
<<
table_id
.
getNameForLogs
()
<<
(
has_sharding_key
?
""
:
"(no sharding key)"
)
<<
": "
(
has_sharding_key
?
""
:
"
(no sharding key)"
)
<<
": "
"Unable to figure out irrelevant shards from WHERE/PREWHERE clauses - "
"the query will be sent to all shards of the cluster"
);
...
...
dbms/tests/queries/0_stateless/01085_max_distributed_connections.reference
0 → 100644
浏览文件 @
9a2760d9
0
0
dbms/tests/queries/0_stateless/01085_max_distributed_connections.sh
0 → 100755
浏览文件 @
9a2760d9
#!/usr/bin/env bash
CURDIR
=
$(
cd
"
$(
dirname
"
${
BASH_SOURCE
[0]
}
"
)
"
&&
pwd
)
.
$CURDIR
/../shell_config.sh
opts
=(
--max_distributed_connections
2
--max_threads
1
--query
"SELECT sleepEachRow(1) FROM remote('127.{2,3}', system.one)"
)
# 1.8 less then 2 seconds, but long enough to cover possible load peaks
# "$@" left to pass manual options (like --experimental_use_processors 0) during manual testing
timeout
1.8s
$CLICKHOUSE_CLIENT
"
${
opts
[@]
}
"
"
$@
"
dbms/tests/queries/0_stateless/01085_max_distributed_connections_http.reference
0 → 100644
浏览文件 @
9a2760d9
0
0
dbms/tests/queries/0_stateless/01085_max_distributed_connections_http.sh
0 → 100755
浏览文件 @
9a2760d9
#!/usr/bin/env bash
CURDIR
=
$(
cd
"
$(
dirname
"
${
BASH_SOURCE
[0]
}
"
)
"
&&
pwd
)
.
$CURDIR
/../shell_config.sh
query
=
"SELECT sleepEachRow(1) FROM remote('127.{2,3}', system.one)"
# 1.8 less then 2 seconds, but long enough to cover possible load peaks
timeout
1.8s
${
CLICKHOUSE_CURL
}
-sS
"
${
CLICKHOUSE_URL
}
&max_distributed_connections=2&max_threads=1"
-d
"
$query
"
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录