2016-06-07 08:23:15 +00:00
# pragma once
# include <DB/DataStreams/IProfilingBlockInputStream.h>
# include <DB/Storages/MergeTree/MergeTreeData.h>
# include <DB/Storages/MergeTree/PKCondition.h>
# include <DB/Storages/MergeTree/MergeTreeReader.h>
namespace DB
{
namespace ErrorCodes
{
extern const int ILLEGAL_TYPE_OF_COLUMN_FOR_FILTER ;
}
/// Для чтения из одного куска. Для чтения сразу из многих, Storage использует сразу много таких объектов.
class MergeTreeBlockInputStream : public IProfilingBlockInputStream
{
public :
MergeTreeBlockInputStream ( const String & path_ , /// Путь к куску
size_t block_size_ , Names column_names ,
MergeTreeData & storage_ , const MergeTreeData : : DataPartPtr & owned_data_part_ ,
const MarkRanges & mark_ranges_ , bool use_uncompressed_cache_ ,
ExpressionActionsPtr prewhere_actions_ , String prewhere_column_ , bool check_columns ,
size_t min_bytes_to_use_direct_io_ , size_t max_read_buffer_size_ ,
bool save_marks_in_cache_ )
:
path ( path_ ) , block_size ( block_size_ ) ,
storage ( storage_ ) , owned_data_part ( owned_data_part_ ) ,
part_columns_lock ( new Poco : : ScopedReadRWLock ( owned_data_part - > columns_lock ) ) ,
all_mark_ranges ( mark_ranges_ ) , remaining_mark_ranges ( mark_ranges_ ) ,
use_uncompressed_cache ( use_uncompressed_cache_ ) ,
prewhere_actions ( prewhere_actions_ ) , prewhere_column ( prewhere_column_ ) ,
log ( & Logger : : get ( " MergeTreeBlockInputStream " ) ) ,
ordered_names { column_names } ,
min_bytes_to_use_direct_io ( min_bytes_to_use_direct_io_ ) , max_read_buffer_size ( max_read_buffer_size_ ) ,
save_marks_in_cache ( save_marks_in_cache_ )
{
try
{
/** @note можно было бы просто поменять местами reverse в if и else ветках MergeTreeDataSelectExecutor,
* а э т о т reverse у б р а т ь . */
std : : reverse ( remaining_mark_ranges . begin ( ) , remaining_mark_ranges . end ( ) ) ;
/// inject columns required for defaults evaluation
const auto injected_columns = injectRequiredColumns ( column_names ) ;
should_reorder = ! injected_columns . empty ( ) ;
Names pre_column_names ;
if ( prewhere_actions )
{
pre_column_names = prewhere_actions - > getRequiredColumns ( ) ;
if ( pre_column_names . empty ( ) )
pre_column_names . push_back ( column_names [ 0 ] ) ;
const auto injected_pre_columns = injectRequiredColumns ( pre_column_names ) ;
if ( ! injected_pre_columns . empty ( ) )
should_reorder = true ;
const NameSet pre_name_set ( pre_column_names . begin ( ) , pre_column_names . end ( ) ) ;
/// Если выражение в PREWHERE - не столбец таблицы, не нужно отдавать наружу столбец с ним
/// (от storage ожидают получить только столбцы таблицы).
remove_prewhere_column = ! pre_name_set . count ( prewhere_column ) ;
Names post_column_names ;
for ( const auto & name : column_names )
if ( ! pre_name_set . count ( name ) )
post_column_names . push_back ( name ) ;
column_names = post_column_names ;
}
/// will be used to distinguish between PREWHERE and WHERE columns when applying filter
column_name_set = NameSet { column_names . begin ( ) , column_names . end ( ) } ;
if ( check_columns )
{
/// Под owned_data_part->columns_lock проверим, что все запрошенные столбцы в куске того же типа, что в таблице.
/// Это может быть не так во время ALTER MODIFY.
if ( ! pre_column_names . empty ( ) )
storage . check ( owned_data_part - > columns , pre_column_names ) ;
if ( ! column_names . empty ( ) )
storage . check ( owned_data_part - > columns , column_names ) ;
pre_columns = storage . getColumnsList ( ) . addTypes ( pre_column_names ) ;
columns = storage . getColumnsList ( ) . addTypes ( column_names ) ;
}
else
{
pre_columns = owned_data_part - > columns . addTypes ( pre_column_names ) ;
columns = owned_data_part - > columns . addTypes ( column_names ) ;
}
/// Оценим общее количество строк - для прогресс-б а р а .
size_t total_rows = 0 ;
for ( const auto & range : all_mark_ranges )
total_rows + = range . end - range . begin ;
total_rows * = storage . index_granularity ;
LOG_TRACE ( log , " Reading " < < all_mark_ranges . size ( ) < < " ranges from part " < < owned_data_part - > name
< < " , approx. " < < total_rows
< < ( all_mark_ranges . size ( ) > 1
? " , up to " + toString ( ( all_mark_ranges . back ( ) . end - all_mark_ranges . front ( ) . begin ) * storage . index_granularity )
: " " )
< < " rows starting from " < < all_mark_ranges . front ( ) . begin * storage . index_granularity ) ;
setTotalRowsApprox ( total_rows ) ;
}
catch ( const Exception & e )
{
/// Подозрение на битый кусок. Кусок добавляется в очередь для проверки.
if ( e . code ( ) ! = ErrorCodes : : MEMORY_LIMIT_EXCEEDED )
storage . reportBrokenPart ( owned_data_part - > name ) ;
throw ;
}
catch ( . . . )
{
storage . reportBrokenPart ( owned_data_part - > name ) ;
throw ;
}
}
String getName ( ) const override { return " MergeTree " ; }
String getID ( ) const override
{
std : : stringstream res ;
res < < " MergeTree( " < < path < < " , columns " ;
for ( const NameAndTypePair & column : columns )
res < < " , " < < column . name ;
if ( prewhere_actions )
res < < " , prewhere, " < < prewhere_actions - > getID ( ) ;
res < < " , marks " ;
for ( size_t i = 0 ; i < all_mark_ranges . size ( ) ; + + i )
res < < " , " < < all_mark_ranges [ i ] . begin < < " , " < < all_mark_ranges [ i ] . end ;
res < < " ) " ;
return res . str ( ) ;
}
protected :
/// Будем вызывать progressImpl самостоятельно.
void progress ( const Progress & value ) override { }
/** Если некоторых запрошенных столбцов нет в куске,
* т о в ы я с н я е м , к а к и е с т о л б ц ы м о ж е т б ы т ь н е о б х о д и м о д о п о л н и т е л ь н о п р о ч и т а т ь ,
* ч т о б ы м о ж н о б ы л о в ы ч и с л и т ь DEFAULT в ы р а ж е н и е д л я э т и х с т о л б ц о в .
* Д о б а в л я е т и х в columns .
*/
NameSet injectRequiredColumns ( Names & columns ) const
{
NameSet required_columns { std : : begin ( columns ) , std : : end ( columns ) } ;
NameSet injected_columns ;
auto all_column_files_missing = true ;
for ( size_t i = 0 ; i < columns . size ( ) ; + + i )
{
const auto & column_name = columns [ i ] ;
/// column has files and hence does not require evaluation
if ( owned_data_part - > hasColumnFiles ( column_name ) )
{
all_column_files_missing = false ;
continue ;
}
const auto default_it = storage . column_defaults . find ( column_name ) ;
/// columns has no explicit default expression
if ( default_it = = std : : end ( storage . column_defaults ) )
continue ;
/// collect identifiers required for evaluation
IdentifierNameSet identifiers ;
default_it - > second . expression - > collectIdentifierNames ( identifiers ) ;
for ( const auto & identifier : identifiers )
{
if ( storage . hasColumn ( identifier ) )
{
/// ensure each column is added only once
if ( required_columns . count ( identifier ) = = 0 )
{
columns . emplace_back ( identifier ) ;
required_columns . emplace ( identifier ) ;
injected_columns . emplace ( identifier ) ;
}
}
}
}
/// If all files are missing read at least one column to determine correct column sizes
if ( all_column_files_missing )
{
const auto minimum_size_column_name = owned_data_part - > getMinimumSizeColumnName ( ) ;
columns . push_back ( minimum_size_column_name ) ;
injected_columns . insert ( minimum_size_column_name ) ;
}
return injected_columns ;
}
Block readImpl ( ) override
{
Block res ;
if ( remaining_mark_ranges . empty ( ) )
return res ;
if ( ! reader )
{
if ( use_uncompressed_cache )
owned_uncompressed_cache = storage . context . getUncompressedCache ( ) ;
owned_mark_cache = storage . context . getMarkCache ( ) ;
reader . reset ( new MergeTreeReader (
path , owned_data_part , columns , owned_uncompressed_cache . get ( ) ,
owned_mark_cache . get ( ) , save_marks_in_cache , storage ,
all_mark_ranges , min_bytes_to_use_direct_io , max_read_buffer_size ) ) ;
if ( prewhere_actions )
pre_reader . reset ( new MergeTreeReader (
path , owned_data_part , pre_columns , owned_uncompressed_cache . get ( ) ,
owned_mark_cache . get ( ) , save_marks_in_cache , storage ,
all_mark_ranges , min_bytes_to_use_direct_io , max_read_buffer_size ) ) ;
}
if ( prewhere_actions )
{
do
{
/// Прочитаем полный блок столбцов, нужных для вычисления выражения в PREWHERE.
size_t space_left = std : : max ( 1LU , block_size / storage . index_granularity ) ;
MarkRanges ranges_to_read ;
while ( ! remaining_mark_ranges . empty ( ) & & space_left & & ! isCancelled ( ) )
{
MarkRange & range = remaining_mark_ranges . back ( ) ;
size_t marks_to_read = std : : min ( range . end - range . begin , space_left ) ;
pre_reader - > readRange ( range . begin , range . begin + marks_to_read , res ) ;
ranges_to_read . push_back ( MarkRange ( range . begin , range . begin + marks_to_read ) ) ;
space_left - = marks_to_read ;
range . begin + = marks_to_read ;
if ( range . begin = = range . end )
remaining_mark_ranges . pop_back ( ) ;
}
/// В случае isCancelled.
if ( ! res )
return res ;
progressImpl ( Progress ( res . rowsInFirstColumn ( ) , res . bytes ( ) ) ) ;
pre_reader - > fillMissingColumns ( res , ordered_names , should_reorder ) ;
/// Вычислим выражение в PREWHERE.
prewhere_actions - > execute ( res ) ;
ColumnPtr column = res . getByName ( prewhere_column ) . column ;
if ( remove_prewhere_column )
res . erase ( prewhere_column ) ;
size_t pre_bytes = res . bytes ( ) ;
/** Если фильтр - константа (например, написано PREWHERE 1),
* т о л и б о в е р н ё м п у с т о й б л о к , л и б о в е р н ё м б л о к б е з и з м е н е н и й .
*/
if ( ColumnConstUInt8 * column_const = typeid_cast < ColumnConstUInt8 * > ( & * column ) )
{
if ( ! column_const - > getData ( ) )
{
res . clear ( ) ;
return res ;
}
for ( size_t i = 0 ; i < ranges_to_read . size ( ) ; + + i )
{
const MarkRange & range = ranges_to_read [ i ] ;
reader - > readRange ( range . begin , range . end , res ) ;
}
progressImpl ( Progress ( 0 , res . bytes ( ) - pre_bytes ) ) ;
}
else if ( ColumnUInt8 * column_vec = typeid_cast < ColumnUInt8 * > ( & * column ) )
{
size_t index_granularity = storage . index_granularity ;
const IColumn : : Filter & pre_filter = column_vec - > getData ( ) ;
IColumn : : Filter post_filter ( pre_filter . size ( ) ) ;
/// Прочитаем в нужных отрезках остальные столбцы и составим для них свой фильтр.
size_t pre_filter_pos = 0 ;
size_t post_filter_pos = 0 ;
for ( size_t i = 0 ; i < ranges_to_read . size ( ) ; + + i )
{
const MarkRange & range = ranges_to_read [ i ] ;
size_t begin = range . begin ;
size_t pre_filter_begin_pos = pre_filter_pos ;
for ( size_t mark = range . begin ; mark < = range . end ; + + mark )
{
UInt8 nonzero = 0 ;
if ( mark ! = range . end )
{
size_t limit = std : : min ( pre_filter . size ( ) , pre_filter_pos + index_granularity ) ;
for ( size_t row = pre_filter_pos ; row < limit ; + + row )
nonzero | = pre_filter [ row ] ;
}
if ( ! nonzero )
{
if ( mark > begin )
{
memcpy (
& post_filter [ post_filter_pos ] ,
& pre_filter [ pre_filter_begin_pos ] ,
pre_filter_pos - pre_filter_begin_pos ) ;
post_filter_pos + = pre_filter_pos - pre_filter_begin_pos ;
reader - > readRange ( begin , mark , res ) ;
}
begin = mark + 1 ;
pre_filter_begin_pos = std : : min ( pre_filter_pos + index_granularity , pre_filter . size ( ) ) ;
}
if ( mark < range . end )
pre_filter_pos = std : : min ( pre_filter_pos + index_granularity , pre_filter . size ( ) ) ;
}
}
if ( ! post_filter_pos )
{
res . clear ( ) ;
continue ;
}
progressImpl ( Progress ( 0 , res . bytes ( ) - pre_bytes ) ) ;
post_filter . resize ( post_filter_pos ) ;
/// Отфильтруем столбцы, относящиеся к PREWHERE, используя pre_filter,
/// остальные столбцы - используя post_filter.
size_t rows = 0 ;
for ( size_t i = 0 ; i < res . columns ( ) ; + + i )
{
ColumnWithTypeAndName & column = res . getByPosition ( i ) ;
if ( column . name = = prewhere_column & & res . columns ( ) > 1 )
continue ;
column . column = column . column - > filter ( column_name_set . count ( column . name ) ? post_filter : pre_filter , - 1 ) ;
rows = column . column - > size ( ) ;
}
/// Заменим столбец с о значением условия из PREWHERE на константу.
if ( ! remove_prewhere_column )
2016-06-08 13:08:20 +00:00
res . getByName ( prewhere_column ) . column = std : : make_shared < ColumnConstUInt8 > ( rows , 1 ) ;
2016-06-07 08:23:15 +00:00
}
else
throw Exception ( " Illegal type " + column - > getName ( ) + " of column for filter. Must be ColumnUInt8 or ColumnConstUInt8. " , ErrorCodes : : ILLEGAL_TYPE_OF_COLUMN_FOR_FILTER ) ;
if ( res )
reader - > fillMissingColumnsAndReorder ( res , ordered_names ) ;
}
while ( ! remaining_mark_ranges . empty ( ) & & ! res & & ! isCancelled ( ) ) ;
}
else
{
size_t space_left = std : : max ( 1LU , block_size / storage . index_granularity ) ;
while ( ! remaining_mark_ranges . empty ( ) & & space_left & & ! isCancelled ( ) )
{
MarkRange & range = remaining_mark_ranges . back ( ) ;
size_t marks_to_read = std : : min ( range . end - range . begin , space_left ) ;
reader - > readRange ( range . begin , range . begin + marks_to_read , res ) ;
space_left - = marks_to_read ;
range . begin + = marks_to_read ;
if ( range . begin = = range . end )
remaining_mark_ranges . pop_back ( ) ;
}
/// В случае isCancelled.
if ( ! res )
return res ;
progressImpl ( Progress ( res . rowsInFirstColumn ( ) , res . bytes ( ) ) ) ;
reader - > fillMissingColumns ( res , ordered_names ) ;
}
if ( remaining_mark_ranges . empty ( ) )
{
/** Закрываем файлы (ещё до уничтожения объекта).
* Ч т о б ы п р и с о з д а н и и м н о г и х и с т о ч н и к о в , н о о д н о в р е м е н н о м ч т е н и и т о л ь к о и з н е с к о л ь к и х ,
* б у ф е р ы н е в и с е л и в п а м я т и .
*/
reader . reset ( ) ;
pre_reader . reset ( ) ;
part_columns_lock . reset ( ) ;
owned_data_part . reset ( ) ;
}
return res ;
}
private :
const String path ;
size_t block_size ;
NamesAndTypesList columns ;
NameSet column_name_set ;
NamesAndTypesList pre_columns ;
MergeTreeData & storage ;
MergeTreeData : : DataPartPtr owned_data_part ; /// Кусок не будет удалён, пока им владеет этот объект.
std : : unique_ptr < Poco : : ScopedReadRWLock > part_columns_lock ; /// Н е дадим изменить список столбцов куска, пока мы из него читаем.
MarkRanges all_mark_ranges ; /// В каких диапазонах засечек читать. В порядке возрастания номеров.
MarkRanges remaining_mark_ranges ; /// В каких диапазонах засечек еще не прочли.
/// В порядке убывания номеров, чтобы можно было выбрасывать из конца.
bool use_uncompressed_cache ;
std : : unique_ptr < MergeTreeReader > reader ;
std : : unique_ptr < MergeTreeReader > pre_reader ;
ExpressionActionsPtr prewhere_actions ;
String prewhere_column ;
bool remove_prewhere_column ;
Logger * log ;
/// column names in specific order as expected by other stages
Names ordered_names ;
bool should_reorder { false } ;
size_t min_bytes_to_use_direct_io ;
size_t max_read_buffer_size ;
UncompressedCachePtr owned_uncompressed_cache ;
MarkCachePtr owned_mark_cache ;
/// Если выставлено в false - при отсутствии засечек в кэше, считавать засечки, но не сохранять их в кэш, чтобы не вымывать оттуда другие данные.
bool save_marks_in_cache ;
} ;
}