Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
2dot5
ClickHouse
提交
09dd2536
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,发现更多精彩内容 >>
提交
09dd2536
编写于
11月 19, 2015
作者:
A
Alexey Milovidov
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
dbms: added support for SAMPLE ... OFFSET [#METR-18847].
上级
8255d767
变更
9
隐藏空白更改
内联
并排
Showing
9 changed file
with
255 addition
and
110 deletion
+255
-110
dbms/include/DB/Core/Types.h
dbms/include/DB/Core/Types.h
+3
-0
dbms/include/DB/Parsers/ASTSelectQuery.h
dbms/include/DB/Parsers/ASTSelectQuery.h
+1
-0
dbms/include/DB/Storages/MergeTree/MergeTreeData.h
dbms/include/DB/Storages/MergeTree/MergeTreeData.h
+6
-6
dbms/include/DB/Storages/MergeTree/MergeTreeDataSelectExecutor.h
...clude/DB/Storages/MergeTree/MergeTreeDataSelectExecutor.h
+17
-5
dbms/src/Parsers/ASTSelectQuery.cpp
dbms/src/Parsers/ASTSelectQuery.cpp
+7
-0
dbms/src/Parsers/ParserSelectQuery.cpp
dbms/src/Parsers/ParserSelectQuery.cpp
+16
-0
dbms/src/Storages/MergeTree/MergeTreeData.cpp
dbms/src/Storages/MergeTree/MergeTreeData.cpp
+5
-5
dbms/src/Storages/MergeTree/MergeTreeDataSelectExecutor.cpp
dbms/src/Storages/MergeTree/MergeTreeDataSelectExecutor.cpp
+195
-94
dbms/src/Storages/StorageView.cpp
dbms/src/Storages/StorageView.cpp
+5
-0
未找到文件。
dbms/include/DB/Core/Types.h
浏览文件 @
09dd2536
...
...
@@ -60,4 +60,7 @@ template <> struct TypeName<Float32> { static std::string get() { return "Float
template
<
>
struct
TypeName
<
Float64
>
{
static
std
::
string
get
()
{
return
"Float64"
;
}
};
template
<
>
struct
TypeName
<
String
>
{
static
std
::
string
get
()
{
return
"String"
;
}
};
/// Эти типы не поддерживаются СУБД. Но используются в других местах.
template
<
>
struct
TypeName
<
long
double
>
{
static
std
::
string
get
()
{
return
"long double"
;
}
};
}
dbms/include/DB/Parsers/ASTSelectQuery.h
浏览文件 @
09dd2536
...
...
@@ -55,6 +55,7 @@ public:
ASTPtr
join
;
/// Обычный (не ARRAY) JOIN.
bool
final
=
false
;
ASTPtr
sample_size
;
ASTPtr
sample_offset
;
ASTPtr
prewhere_expression
;
ASTPtr
where_expression
;
ASTPtr
group_expression_list
;
...
...
dbms/include/DB/Storages/MergeTree/MergeTreeData.h
浏览文件 @
09dd2536
...
...
@@ -710,17 +710,17 @@ public:
/** Возвращает копию списка, чтобы снаружи можно было не заботиться о блокировках.
*/
DataParts
getDataParts
();
DataPartsVector
getDataPartsVector
();
DataParts
getAllDataParts
();
DataParts
getDataParts
()
const
;
DataPartsVector
getDataPartsVector
()
const
;
DataParts
getAllDataParts
()
const
;
/** Размер активной части в количестве байт.
*/
size_t
getTotalActiveSizeInBytes
();
size_t
getTotalActiveSizeInBytes
()
const
;
/** Максимальное количество кусков в одном месяце.
*/
size_t
getMaxPartsCountForMonth
();
size_t
getMaxPartsCountForMonth
()
const
;
/** Если в таблице слишком много активных кусков, спит некоторое время, чтобы дать им возможность смерджиться.
* Если передано until - проснуться раньше, если наступило событие.
...
...
@@ -885,7 +885,7 @@ private:
* То есть, если количество ссылок равно 1 - то кусок не актуален и не используется прямо сейчас, и его можно удалить.
*/
DataParts
all_data_parts
;
Poco
::
FastMutex
all_data_parts_mutex
;
mutable
Poco
::
FastMutex
all_data_parts_mutex
;
/** Выражение, преобразующее типы столбцов.
* Если преобразований типов нет, out_expression=nullptr.
...
...
dbms/include/DB/Storages/MergeTree/MergeTreeDataSelectExecutor.h
浏览文件 @
09dd2536
...
...
@@ -29,7 +29,7 @@ public:
size_t
max_block_size
,
unsigned
threads
,
size_t
*
inout_part_index
,
Int64
max_block_number_to_read
);
Int64
max_block_number_to_read
)
const
;
private:
MergeTreeData
&
data
;
...
...
@@ -45,7 +45,7 @@ private:
ExpressionActionsPtr
prewhere_actions
,
const
String
&
prewhere_column
,
const
Names
&
virt_columns
,
const
Settings
&
settings
);
const
Settings
&
settings
)
const
;
BlockInputStreams
spreadMarkRangesAmongThreadsFinal
(
RangesInDataParts
parts
,
...
...
@@ -57,12 +57,24 @@ private:
const
String
&
prewhere_column
,
const
Names
&
virt_columns
,
const
Settings
&
settings
,
const
Context
&
context
);
const
Context
&
context
)
const
;
/// Получить приблизительное значение (оценку снизу - только по полным засечкам) количества строк, попадающего под индекс.
size_t
getApproximateTotalRowsToRead
(
const
MergeTreeData
::
DataPartsVector
&
parts
,
const
PKCondition
&
key_condition
,
const
Settings
&
settings
)
const
;
/// Создать выражение "Sign == 1".
void
createPositiveSignCondition
(
ExpressionActionsPtr
&
out_expression
,
String
&
out_column
,
const
Context
&
context
);
void
createPositiveSignCondition
(
ExpressionActionsPtr
&
out_expression
,
String
&
out_column
,
const
Context
&
context
)
const
;
MarkRanges
markRangesFromPkRange
(
const
MergeTreeData
::
DataPart
::
Index
&
index
,
PKCondition
&
key_condition
,
const
Settings
&
settings
);
MarkRanges
markRangesFromPkRange
(
const
MergeTreeData
::
DataPart
::
Index
&
index
,
const
PKCondition
&
key_condition
,
const
Settings
&
settings
)
const
;
};
}
dbms/src/Parsers/ASTSelectQuery.cpp
浏览文件 @
09dd2536
...
...
@@ -190,6 +190,7 @@ ASTPtr ASTSelectQuery::cloneImpl(bool traverse_union_all) const
CLONE
(
array_join_expression_list
)
CLONE
(
join
)
CLONE
(
sample_size
)
CLONE
(
sample_offset
)
CLONE
(
prewhere_expression
)
CLONE
(
where_expression
)
CLONE
(
group_expression_list
)
...
...
@@ -275,6 +276,12 @@ void ASTSelectQuery::formatImpl(const FormatSettings & s, FormatState & state, F
{
s
.
ostr
<<
(
s
.
hilite
?
hilite_keyword
:
""
)
<<
s
.
nl_or_ws
<<
indent_str
<<
"SAMPLE "
<<
(
s
.
hilite
?
hilite_none
:
""
);
sample_size
->
formatImpl
(
s
,
state
,
frame
);
if
(
sample_offset
)
{
s
.
ostr
<<
(
s
.
hilite
?
hilite_keyword
:
""
)
<<
' '
<<
indent_str
<<
"OFFSET "
<<
(
s
.
hilite
?
hilite_none
:
""
);
sample_offset
->
formatImpl
(
s
,
state
,
frame
);
}
}
if
(
array_join_expression_list
)
...
...
dbms/src/Parsers/ParserSelectQuery.cpp
浏览文件 @
09dd2536
...
...
@@ -30,6 +30,7 @@ bool ParserSelectQuery::parseImpl(Pos & pos, Pos end, ASTPtr & node, Pos & max_p
ParserString
s_where
(
"WHERE"
,
true
,
true
);
ParserString
s_final
(
"FINAL"
,
true
,
true
);
ParserString
s_sample
(
"SAMPLE"
,
true
,
true
);
ParserString
s_offset
(
"OFFSET"
,
true
,
true
);
ParserString
s_group
(
"GROUP"
,
true
,
true
);
ParserString
s_by
(
"BY"
,
true
,
true
);
ParserString
s_with
(
"WITH"
,
true
,
true
);
...
...
@@ -161,6 +162,19 @@ bool ParserSelectQuery::parseImpl(Pos & pos, Pos end, ASTPtr & node, Pos & max_p
return
false
;
ws
.
ignore
(
pos
,
end
);
/// OFFSET number
if
(
s_offset
.
ignore
(
pos
,
end
,
max_parsed_pos
,
expected
))
{
ws
.
ignore
(
pos
,
end
);
ParserNumber
num
;
if
(
!
num
.
parse
(
pos
,
end
,
select_query
->
sample_offset
,
max_parsed_pos
,
expected
))
return
false
;
ws
.
ignore
(
pos
,
end
);
}
}
return
true
;
...
...
@@ -355,6 +369,8 @@ bool ParserSelectQuery::parseImpl(Pos & pos, Pos end, ASTPtr & node, Pos & max_p
select_query
->
children
.
push_back
(
select_query
->
join
);
if
(
select_query
->
sample_size
)
select_query
->
children
.
push_back
(
select_query
->
sample_size
);
if
(
select_query
->
sample_offset
)
select_query
->
children
.
push_back
(
select_query
->
sample_offset
);
if
(
select_query
->
prewhere_expression
)
select_query
->
children
.
push_back
(
select_query
->
prewhere_expression
);
if
(
select_query
->
where_expression
)
...
...
dbms/src/Storages/MergeTree/MergeTreeData.cpp
浏览文件 @
09dd2536
...
...
@@ -905,21 +905,21 @@ void MergeTreeData::detachPartInPlace(const DataPartPtr & part)
renameAndDetachPart
(
part
,
""
,
false
,
false
);
}
MergeTreeData
::
DataParts
MergeTreeData
::
getDataParts
()
MergeTreeData
::
DataParts
MergeTreeData
::
getDataParts
()
const
{
Poco
::
ScopedLock
<
Poco
::
FastMutex
>
lock
(
data_parts_mutex
);
return
data_parts
;
}
MergeTreeData
::
DataPartsVector
MergeTreeData
::
getDataPartsVector
()
MergeTreeData
::
DataPartsVector
MergeTreeData
::
getDataPartsVector
()
const
{
Poco
::
ScopedLock
<
Poco
::
FastMutex
>
lock
(
data_parts_mutex
);
return
DataPartsVector
(
std
::
begin
(
data_parts
),
std
::
end
(
data_parts
));
}
size_t
MergeTreeData
::
getTotalActiveSizeInBytes
()
size_t
MergeTreeData
::
getTotalActiveSizeInBytes
()
const
{
Poco
::
ScopedLock
<
Poco
::
FastMutex
>
lock
(
data_parts_mutex
);
...
...
@@ -930,14 +930,14 @@ size_t MergeTreeData::getTotalActiveSizeInBytes()
return
res
;
}
MergeTreeData
::
DataParts
MergeTreeData
::
getAllDataParts
()
MergeTreeData
::
DataParts
MergeTreeData
::
getAllDataParts
()
const
{
Poco
::
ScopedLock
<
Poco
::
FastMutex
>
lock
(
all_data_parts_mutex
);
return
all_data_parts
;
}
size_t
MergeTreeData
::
getMaxPartsCountForMonth
()
size_t
MergeTreeData
::
getMaxPartsCountForMonth
()
const
{
Poco
::
ScopedLock
<
Poco
::
FastMutex
>
lock
(
data_parts_mutex
);
...
...
dbms/src/Storages/MergeTree/MergeTreeDataSelectExecutor.cpp
浏览文件 @
09dd2536
...
...
@@ -24,6 +24,7 @@ MergeTreeDataSelectExecutor::MergeTreeDataSelectExecutor(MergeTreeData & data_)
{
}
/// Построить блок состоящий только из возможных значений виртуальных столбцов
static
Block
getBlockWithVirtualColumns
(
const
MergeTreeData
::
DataPartsVector
&
parts
)
{
...
...
@@ -37,6 +38,50 @@ static Block getBlockWithVirtualColumns(const MergeTreeData::DataPartsVector & p
return
res
;
}
size_t
MergeTreeDataSelectExecutor
::
getApproximateTotalRowsToRead
(
const
MergeTreeData
::
DataPartsVector
&
parts
,
const
PKCondition
&
key_condition
,
const
Settings
&
settings
)
const
{
size_t
full_marks_count
=
0
;
/// Узнаем, сколько строк мы бы прочли без семплирования.
LOG_DEBUG
(
log
,
"Preliminary index scan with condition: "
<<
key_condition
.
toString
());
for
(
size_t
i
=
0
;
i
<
parts
.
size
();
++
i
)
{
const
MergeTreeData
::
DataPartPtr
&
part
=
parts
[
i
];
MarkRanges
ranges
=
markRangesFromPkRange
(
part
->
index
,
key_condition
,
settings
);
/** Для того, чтобы получить оценку снизу количества строк, подходящих под условие на PK,
* учитываем только гарантированно полные засечки.
* То есть, не учитываем первую и последнюю засечку, которые могут быть неполными.
*/
for
(
size_t
j
=
0
;
j
<
ranges
.
size
();
++
j
)
if
(
ranges
[
j
].
end
-
ranges
[
j
].
begin
>
2
)
full_marks_count
+=
ranges
[
j
].
end
-
ranges
[
j
].
begin
-
2
;
}
return
full_marks_count
*
data
.
index_granularity
;
}
/** Пожалуй, наиболее удобный способ разбить диапазон 64-битных целых чисел на интервалы по их относительной величине
* - использовать для этого long double. Это некроссплатформенно. Надо, чтобы long double содержал хотя бы 64 бита мантиссы.
*/
using
RelativeSize
=
long
double
;
/// Переводит размер сэмпла в приблизительном количестве строк (вида SAMPLE 1000000) в относительную величину (вида SAMPLE 0.1).
static
RelativeSize
convertAbsoluteSampleSizeToRelative
(
const
ASTPtr
&
node
,
size_t
approx_total_rows
)
{
if
(
approx_total_rows
==
0
)
return
1
;
size_t
absolute_sample_size
=
apply_visitor
(
FieldVisitorConvertToNumber
<
UInt64
>
(),
typeid_cast
<
const
ASTLiteral
&>
(
*
node
).
value
);
return
std
::
min
(
RelativeSize
(
1.0
),
RelativeSize
(
absolute_sample_size
)
/
approx_total_rows
);
}
BlockInputStreams
MergeTreeDataSelectExecutor
::
read
(
const
Names
&
column_names_to_return
,
ASTPtr
query
,
...
...
@@ -46,7 +91,7 @@ BlockInputStreams MergeTreeDataSelectExecutor::read(
const
size_t
max_block_size
,
const
unsigned
threads
,
size_t
*
part_index
,
Int64
max_block_number_to_read
)
Int64
max_block_number_to_read
)
const
{
size_t
part_index_var
=
0
;
if
(
!
part_index
)
...
...
@@ -113,60 +158,89 @@ BlockInputStreams MergeTreeDataSelectExecutor::read(
typedef
Poco
::
SharedPtr
<
ASTFunction
>
ASTFunctionPtr
;
ASTFunctionPtr
filter_function
;
ExpressionActionsPtr
filter_expression
;
double
relative_sample_size
=
0
;
RelativeSize
relative_sample_size
=
0
;
RelativeSize
relative_sample_offset
=
0
;
ASTSelectQuery
&
select
=
*
typeid_cast
<
ASTSelectQuery
*>
(
&*
query
);
if
(
select
.
sample_size
)
{
relative_sample_size
=
apply_visitor
(
FieldVisitorConvertToNumber
<
doubl
e
>
(),
relative_sample_size
=
apply_visitor
(
FieldVisitorConvertToNumber
<
RelativeSiz
e
>
(),
typeid_cast
<
ASTLiteral
&>
(
*
select
.
sample_size
).
value
);
if
(
relative_sample_size
<
0
)
throw
Exception
(
"Negative sample size"
,
ErrorCodes
::
ARGUMENT_OUT_OF_BOUND
);
/// Переводим абсолютную величину сэмплирования (вида SAMPLE 1000000 - сколько строк прочитать) в относительную (какую долю данных читать).
if
(
relative_sample_size
>
1
)
{
size_t
requested_count
=
apply_visitor
(
FieldVisitorConvertToNumber
<
UInt64
>
(),
typeid_cast
<
ASTLiteral
&>
(
*
select
.
sample_size
).
value
);
relative_sample_offset
=
0
;
if
(
select
.
sample_offset
)
relative_sample_offset
=
apply_visitor
(
FieldVisitorConvertToNumber
<
RelativeSize
>
(),
typeid_cast
<
ASTLiteral
&>
(
*
select
.
sample_offset
).
value
);
/// Узнаем, сколько строк мы бы прочли без семплирования.
LOG_DEBUG
(
log
,
"Preliminary index scan with condition: "
<<
key_condition
.
toString
());
size_t
total_count
=
0
;
for
(
size_t
i
=
0
;
i
<
parts
.
size
();
++
i
)
{
MergeTreeData
::
DataPartPtr
&
part
=
parts
[
i
];
MarkRanges
ranges
=
markRangesFromPkRange
(
part
->
index
,
key_condition
,
settings
);
/** Для того, чтобы получить оценку снизу количества строк, подходящих под условие на PK,
* учитываем только гарантированно полные засечки.
* То есть, не учитываем первую и последнюю засечку, которые могут быть неполными.
*/
for
(
size_t
j
=
0
;
j
<
ranges
.
size
();
++
j
)
if
(
ranges
[
j
].
end
-
ranges
[
j
].
begin
>
2
)
total_count
+=
ranges
[
j
].
end
-
ranges
[
j
].
begin
-
2
;
}
total_count
*=
data
.
index_granularity
;
if
(
relative_sample_offset
<
0
)
throw
Exception
(
"Negative sample offset"
,
ErrorCodes
::
ARGUMENT_OUT_OF_BOUND
);
if
(
total_count
==
0
)
relative_sample_size
=
1
;
else
relative_sample_size
=
std
::
min
(
1.
,
static_cast
<
double
>
(
requested_count
)
/
total_count
);
/// Переводим абсолютную величину сэмплирования (вида SAMPLE 1000000 - сколько строк прочитать) в относительную (какую долю данных читать).
size_t
approx_total_rows
=
0
;
if
(
relative_sample_size
>
1
||
relative_sample_offset
>
1
)
approx_total_rows
=
getApproximateTotalRowsToRead
(
parts
,
key_condition
,
settings
);
if
(
relative_sample_size
>
1
)
{
relative_sample_size
=
convertAbsoluteSampleSizeToRelative
(
select
.
sample_size
,
approx_total_rows
);
LOG_DEBUG
(
log
,
"Selected relative sample size: "
<<
relative_sample_size
);
}
/// SAMPLE 1 - то же, что и отсутствие SAMPLE.
if
(
relative_sample_size
==
1
)
relative_sample_size
=
0
;
}
if
((
settings
.
parallel_replicas_count
>
1
)
&&
!
data
.
sampling_expression
.
isNull
()
&&
(
relative_sample_size
==
0
)
)
relative_sample_size
=
1
;
if
(
relative_sample_offset
>
0
&&
0
==
relative_sample_size
)
throw
Exception
(
"Sampling offset is incorrect because no sampling"
,
ErrorCodes
::
ARGUMENT_OUT_OF_BOUND
)
;
if
(
relative_sample_size
!=
0
)
if
(
relative_sample_offset
>
1
)
{
relative_sample_offset
=
convertAbsoluteSampleSizeToRelative
(
select
.
sample_offset
,
approx_total_rows
);
LOG_DEBUG
(
log
,
"Selected relative sample offset: "
<<
relative_sample_offset
);
}
}
/** Какой диапазон значений ключа сэмплирования нужно читать?
* Сначала во всём диапазоне ("юнивёрсум") выбераем интервал
* относительного размера relative_sample_size, смещённый от начала на relative_sample_offset.
*
* Пример: SAMPLE 0.4 OFFSET 0.3:
*
* [------********------]
* ^ - offset
* <------> - size
*
* Если интервал переходит через конец юнивёрсума, то срезаем его правую часть.
*
* Пример: SAMPLE 0.4 OFFSET 0.8:
*
* [----------------****]
* ^ - offset
* <------> - size
*
* Далее, если выставлены настройки parallel_replicas_count, parallel_replica_offset,
* то необходимо разбить полученный интервал ещё на кусочки в количестве parallel_replicas_count,
* и выбрать из них кусочек с номером parallel_replica_offset (от нуля).
*
* Пример: SAMPLE 0.4 OFFSET 0.3, parallel_replicas_count = 2, parallel_replica_offset = 1:
*
* [----------****------]
* ^ - offset
* <------> - size
* <--><--> - кусочки для разных parallel_replica_offset, выбираем второй.
*/
bool
use_sampling
=
relative_sample_size
>
0
||
settings
.
parallel_replicas_count
>
1
;
bool
no_data
=
false
;
/// После сэмплирования ничего не остаётся.
if
(
use_sampling
)
{
UInt64
sampling_column_max
=
0
;
RelativeSize
sampling_column_max
=
0
;
DataTypePtr
type
=
data
.
getPrimaryExpression
()
->
getSampleBlock
().
getByName
(
data
.
sampling_expression
->
getColumnName
()).
type
;
if
(
type
->
getName
()
==
"UInt64"
)
...
...
@@ -180,78 +254,105 @@ BlockInputStreams MergeTreeDataSelectExecutor::read(
else
throw
Exception
(
"Invalid sampling column type in storage parameters: "
+
type
->
getName
()
+
". Must be unsigned integer type."
,
ErrorCodes
::
ILLEGAL_TYPE_OF_COLUMN_FOR_FILTER
);
UInt64
sampling_column_value_lower_limit
;
UInt64
sampling_column_value_upper_limit
;
UInt64
upper_limit
=
static_cast
<
long
double
>
(
relative_sample_size
)
*
sampling_column_max
;
if
(
settings
.
parallel_replicas_count
>
1
)
{
sampling_column_value_lower_limit
=
(
static_cast
<
long
double
>
(
settings
.
parallel_replica_offset
)
/
settings
.
parallel_replicas_count
)
*
upper_limit
;
if
((
settings
.
parallel_replica_offset
+
1
)
<
settings
.
parallel_replicas_count
)
sampling_column_value_upper_limit
=
(
static_cast
<
long
double
>
(
settings
.
parallel_replica_offset
+
1
)
/
settings
.
parallel_replicas_count
)
*
upper_limit
;
else
sampling_column_value_upper_limit
=
(
upper_limit
<
sampling_column_max
)
?
(
upper_limit
+
1
)
:
upper_limit
;
relative_sample_size
/=
settings
.
parallel_replicas_count
;
relative_sample_offset
+=
relative_sample_size
*
settings
.
parallel_replica_offset
;
}
else
no_data
=
relative_sample_size
*
sampling_column_max
<
1
||
relative_sample_offset
>=
1
;
/// Вычисляем полуинтервал [lower, upper) значений столбца.
bool
has_lower_limit
=
false
;
bool
has_upper_limit
=
false
;
UInt64
lower
=
0
;
UInt64
upper
=
0
;
lower
=
relative_sample_offset
*
sampling_column_max
;
if
(
lower
>
0
)
has_lower_limit
=
true
;
RelativeSize
upper_limit_float
=
(
relative_sample_offset
+
relative_sample_size
)
*
sampling_column_max
;
if
(
upper_limit_float
<
sampling_column_max
)
{
sampling_column_value_lower_limit
=
0
;
sampling_column_value_upper_limit
=
(
upper_limit
<
sampling_column_max
)
?
(
upper_limit
+
1
)
:
upper_limit
;
upper
=
upper_limit_float
;
has_upper_limit
=
true
;
}
/// Добавим условие, чтобы отсечь еще что-нибудь при повторном просмотре индекса.
if
(
sampling_column_value_lower_limit
>
0
)
if
(
!
key_condition
.
addCondition
(
data
.
sampling_expression
->
getColumnName
(),
Range
::
createLeftBounded
(
sampling_column_value_lower_limit
,
true
)))
throw
Exception
(
"Sampling column not in primary key"
,
ErrorCodes
::
ILLEGAL_COLUMN
);
if
(
no_data
||
(
!
has_lower_limit
&&
!
has_upper_limit
))
{
use_sampling
=
false
;
}
else
{
/// Добавим условия, чтобы отсечь еще что-нибудь при повторном просмотре индекса и при обработке запроса.
if
(
!
key_condition
.
addCondition
(
data
.
sampling_expression
->
getColumnName
(),
Range
::
createRightBounded
(
sampling_column_value_upper_limit
,
false
)))
throw
Exception
(
"Sampling column not in primary key"
,
ErrorCodes
::
ILLEGAL_COLUMN
);
ASTFunctionPtr
lower_function
;
ASTFunctionPtr
upper_function
;
ASTPtr
upper_filter_args
=
new
ASTExpressionList
;
upper_filter_args
->
children
.
push_back
(
data
.
sampling_expression
);
upper_filter_args
->
children
.
push_back
(
new
ASTLiteral
(
StringRange
(),
sampling_column_value_upper_limit
));
if
(
has_lower_limit
)
{
if
(
!
key_condition
.
addCondition
(
data
.
sampling_expression
->
getColumnName
(),
Range
::
createLeftBounded
(
lower
,
true
)))
throw
Exception
(
"Sampling column not in primary key"
,
ErrorCodes
::
ILLEGAL_COLUMN
);
ASTFunctionPtr
upper_filter_function
=
new
ASTFunction
;
upper_filter_function
->
name
=
"less"
;
upper_filter_function
->
arguments
=
upper_filter_args
;
upper_filter_function
->
children
.
push_back
(
upper_filter_function
->
arguments
);
ASTPtr
args
=
new
ASTExpressionList
;
args
->
children
.
push_back
(
data
.
sampling_expression
);
args
->
children
.
push_back
(
new
ASTLiteral
(
StringRange
(),
lower
));
if
(
sampling_column_value_lower_limit
>
0
)
{
/// Выражение для фильтрации: sampling_expression in [sampling_column_value_lower_limit, sampling_column_value_upper_limit)
lower_function
=
new
ASTFunction
;
lower_function
->
name
=
"greaterOrEquals"
;
lower_function
->
arguments
=
args
;
lower_function
->
children
.
push_back
(
lower_function
->
arguments
);
filter_function
=
lower_function
;
}
ASTPtr
lower_filter_args
=
new
ASTExpressionList
;
lower_filter_args
->
children
.
push_back
(
data
.
sampling_expression
);
lower_filter_args
->
children
.
push_back
(
new
ASTLiteral
(
StringRange
(),
sampling_column_value_lower_limit
));
if
(
has_upper_limit
)
{
if
(
!
key_condition
.
addCondition
(
data
.
sampling_expression
->
getColumnName
(),
Range
::
createRightBounded
(
upper
,
false
)))
throw
Exception
(
"Sampling column not in primary key"
,
ErrorCodes
::
ILLEGAL_COLUMN
);
ASTFunctionPtr
lower_filter_function
=
new
ASTFunction
;
lower_filter_function
->
name
=
"greaterOrEquals"
;
lower_filter_function
->
arguments
=
lower_filter_args
;
lower_filter_function
->
children
.
push_back
(
lower_filter_function
->
arguments
);
ASTPtr
args
=
new
ASTExpressionList
;
args
->
children
.
push_back
(
data
.
sampling_expression
);
args
->
children
.
push_back
(
new
ASTLiteral
(
StringRange
(),
upper
));
ASTPtr
filter_function_args
=
new
ASTExpressionList
;
filter_function_args
->
children
.
push_back
(
lower_filter_function
);
filter_function_args
->
children
.
push_back
(
upper_filter_function
);
upper_function
=
new
ASTFunction
;
upper_function
->
name
=
"less"
;
upper_function
->
arguments
=
args
;
upper_function
->
children
.
push_back
(
upper_function
->
arguments
);
filter_function
=
new
ASTFunction
;
filter_function
->
name
=
"and"
;
filter_function
->
arguments
=
filter_function_args
;
filter_function
->
children
.
push_back
(
filter_function
->
arguments
);
}
else
{
/// Выражение для фильтрации: sampling_expression < sampling_column_value_upper_limit
filter_function
=
upper_filter_function
;
}
filter_function
=
upper_function
;
}
filter_expression
=
ExpressionAnalyzer
(
filter_function
,
context
,
nullptr
,
data
.
getColumnsList
()).
getActions
(
false
);
if
(
has_lower_limit
&&
has_upper_limit
)
{
ASTPtr
args
=
new
ASTExpressionList
;
args
->
children
.
push_back
(
lower_function
);
args
->
children
.
push_back
(
upper_function
);
filter_function
=
new
ASTFunction
;
filter_function
->
name
=
"and"
;
filter_function
->
arguments
=
args
;
filter_function
->
children
.
push_back
(
filter_function
->
arguments
);
}
/// Добавим столбцы, нужные для sampling_expression.
std
::
vector
<
String
>
add_columns
=
filter_expression
->
getRequiredColumns
();
column_names_to_read
.
insert
(
column_names_to_read
.
end
(),
add_columns
.
begin
(),
add_columns
.
end
());
std
::
sort
(
column_names_to_read
.
begin
(),
column_names_to_read
.
end
());
column_names_to_read
.
erase
(
std
::
unique
(
column_names_to_read
.
begin
(),
column_names_to_read
.
end
()),
column_names_to_read
.
end
());
filter_expression
=
ExpressionAnalyzer
(
filter_function
,
context
,
nullptr
,
data
.
getColumnsList
()).
getActions
(
false
);
/// Добавим столбцы, нужные для sampling_expression.
std
::
vector
<
String
>
add_columns
=
filter_expression
->
getRequiredColumns
();
column_names_to_read
.
insert
(
column_names_to_read
.
end
(),
add_columns
.
begin
(),
add_columns
.
end
());
std
::
sort
(
column_names_to_read
.
begin
(),
column_names_to_read
.
end
());
column_names_to_read
.
erase
(
std
::
unique
(
column_names_to_read
.
begin
(),
column_names_to_read
.
end
()),
column_names_to_read
.
end
());
}
}
if
(
no_data
)
{
LOG_DEBUG
(
log
,
"Sampling yields no data."
);
return
{};
}
LOG_DEBUG
(
log
,
"Key condition: "
<<
key_condition
.
toString
());
...
...
@@ -345,7 +446,7 @@ BlockInputStreams MergeTreeDataSelectExecutor::read(
settings
);
}
if
(
relative_sample_size
!=
0
)
if
(
use_sampling
)
for
(
auto
&
stream
:
res
)
stream
=
new
FilterBlockInputStream
(
new
ExpressionBlockInputStream
(
stream
,
filter_expression
),
filter_function
->
getColumnName
());
...
...
@@ -362,7 +463,7 @@ BlockInputStreams MergeTreeDataSelectExecutor::spreadMarkRangesAmongThreads(
ExpressionActionsPtr
prewhere_actions
,
const
String
&
prewhere_column
,
const
Names
&
virt_columns
,
const
Settings
&
settings
)
const
Settings
&
settings
)
const
{
const
std
::
size_t
min_marks_for_concurrent_read
=
(
settings
.
merge_tree_min_rows_for_concurrent_read
+
data
.
index_granularity
-
1
)
/
data
.
index_granularity
;
...
...
@@ -512,7 +613,7 @@ BlockInputStreams MergeTreeDataSelectExecutor::spreadMarkRangesAmongThreadsFinal
const
String
&
prewhere_column
,
const
Names
&
virt_columns
,
const
Settings
&
settings
,
const
Context
&
context
)
const
Context
&
context
)
const
{
const
size_t
max_marks_to_use_cache
=
(
settings
.
merge_tree_max_rows_to_use_cache
+
data
.
index_granularity
-
1
)
/
data
.
index_granularity
;
...
...
@@ -631,7 +732,7 @@ BlockInputStreams MergeTreeDataSelectExecutor::spreadMarkRangesAmongThreadsFinal
return
res
;
}
void
MergeTreeDataSelectExecutor
::
createPositiveSignCondition
(
ExpressionActionsPtr
&
out_expression
,
String
&
out_column
,
const
Context
&
context
)
void
MergeTreeDataSelectExecutor
::
createPositiveSignCondition
(
ExpressionActionsPtr
&
out_expression
,
String
&
out_column
,
const
Context
&
context
)
const
{
ASTFunction
*
function
=
new
ASTFunction
;
ASTPtr
function_ptr
=
function
;
...
...
@@ -664,7 +765,7 @@ void MergeTreeDataSelectExecutor::createPositiveSignCondition(ExpressionActionsP
/// Получает набор диапазонов засечек, вне которых не могут находиться ключи из заданного диапазона.
MarkRanges
MergeTreeDataSelectExecutor
::
markRangesFromPkRange
(
const
MergeTreeData
::
DataPart
::
Index
&
index
,
PKCondition
&
key_condition
,
const
Settings
&
settings
)
const
MergeTreeData
::
DataPart
::
Index
&
index
,
const
PKCondition
&
key_condition
,
const
Settings
&
settings
)
const
{
size_t
min_marks_for_seek
=
(
settings
.
merge_tree_min_rows_for_seek
+
data
.
index_granularity
-
1
)
/
data
.
index_granularity
;
...
...
dbms/src/Storages/StorageView.cpp
浏览文件 @
09dd2536
...
...
@@ -89,8 +89,13 @@ BlockInputStreams StorageView::read(
/// Пробрасываем внутрь SAMPLE и FINAL, если они есть во внешнем запросе и их нет во внутреннем.
if
(
outer_select
.
sample_size
&&
!
inner_select
.
sample_size
)
{
inner_select
.
sample_size
=
outer_select
.
sample_size
;
if
(
outer_select
.
sample_offset
&&
!
inner_select
.
sample_offset
)
inner_select
.
sample_offset
=
outer_select
.
sample_offset
;
}
if
(
outer_select
.
final
&&
!
inner_select
.
final
)
inner_select
.
final
=
outer_select
.
final
;
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录