ClickHouse/dbms/src/Storages/MergeTree/MergeTreeThreadSelectBlockInputStream.cpp

118 lines
4.5 KiB
C++
Raw Normal View History

#include <Storages/MergeTree/MergeTreeReader.h>
#include <Storages/MergeTree/MergeTreeReadPool.h>
2018-11-29 09:19:42 +00:00
#include <Storages/MergeTree/MergeTreeThreadSelectBlockInputStream.h>
2016-11-20 12:43:20 +00:00
namespace DB
{
2018-11-29 09:19:42 +00:00
MergeTreeThreadSelectBlockInputStream::MergeTreeThreadSelectBlockInputStream(
const size_t thread,
const MergeTreeReadPoolPtr & pool,
const size_t min_marks_to_read_,
const size_t max_block_size_rows,
size_t preferred_block_size_bytes,
size_t preferred_max_column_in_block_size_bytes,
2018-10-17 03:13:00 +00:00
const MergeTreeData & storage,
const bool use_uncompressed_cache,
const PrewhereInfoPtr & prewhere_info,
const Settings & settings,
const Names & virt_column_names)
:
2018-11-29 09:19:42 +00:00
MergeTreeBaseSelectBlockInputStream{storage, prewhere_info, max_block_size_rows,
preferred_block_size_bytes, preferred_max_column_in_block_size_bytes, settings.min_bytes_to_use_direct_io,
settings.max_read_buffer_size, use_uncompressed_cache, true, virt_column_names},
thread{thread},
pool{pool}
{
/// round min_marks_to_read up to nearest multiple of block_size expressed in marks
if (max_block_size_rows)
{
min_marks_to_read = (min_marks_to_read_ * storage.index_granularity + max_block_size_rows - 1)
/ max_block_size_rows * max_block_size_rows / storage.index_granularity;
}
else
min_marks_to_read = min_marks_to_read_;
2018-04-19 15:18:26 +00:00
ordered_names = getHeader().getNames();
}
2016-11-20 12:43:20 +00:00
2018-11-29 09:19:42 +00:00
Block MergeTreeThreadSelectBlockInputStream::getHeader() const
{
auto res = pool->getHeader();
2018-04-16 12:21:36 +00:00
executePrewhereActions(res, prewhere_info);
2018-09-07 20:23:28 +00:00
injectVirtualColumns(res);
return res;
}
2016-11-20 12:43:20 +00:00
/// Requests read task from MergeTreeReadPool and signals whether it got one
2018-11-29 09:19:42 +00:00
bool MergeTreeThreadSelectBlockInputStream::getNewTask()
2016-11-20 12:43:20 +00:00
{
2018-04-19 15:18:26 +00:00
task = pool->getTask(min_marks_to_read, thread, ordered_names);
if (!task)
{
/** Close the files (before destroying the object).
* When many sources are created, but simultaneously reading only a few of them,
* buffers don't waste memory.
*/
reader.reset();
pre_reader.reset();
return false;
}
const std::string path = task->data_part->getFullPath();
/// Allows pool to reduce number of threads in case of too slow reads.
auto profile_callback = [this](ReadBufferFromFileBase::ProfileInfo info) { pool->profileFeedback(info); };
if (!reader)
{
auto rest_mark_ranges = pool->getRestMarks(path, task->mark_ranges[0]);
if (use_uncompressed_cache)
owned_uncompressed_cache = storage.global_context.getUncompressedCache();
owned_mark_cache = storage.global_context.getMarkCache();
reader = std::make_unique<MergeTreeReader>(
path, task->data_part, task->columns, owned_uncompressed_cache.get(), owned_mark_cache.get(), save_marks_in_cache,
storage, rest_mark_ranges, min_bytes_to_use_direct_io, max_read_buffer_size, MergeTreeReader::ValueSizeMap{}, profile_callback);
if (prewhere_info)
pre_reader = std::make_unique<MergeTreeReader>(
path, task->data_part, task->pre_columns, owned_uncompressed_cache.get(), owned_mark_cache.get(), save_marks_in_cache,
storage, rest_mark_ranges, min_bytes_to_use_direct_io,
max_read_buffer_size, MergeTreeReader::ValueSizeMap{}, profile_callback);
}
else
{
/// in other case we can reuse readers, anyway they will be "seeked" to required mark
if (path != last_readed_part_path)
2018-10-03 17:12:38 +00:00
{
auto rest_mark_ranges = pool->getRestMarks(path, task->mark_ranges[0]);
/// retain avg_value_size_hints
reader = std::make_unique<MergeTreeReader>(
path, task->data_part, task->columns, owned_uncompressed_cache.get(), owned_mark_cache.get(), save_marks_in_cache,
storage, rest_mark_ranges, min_bytes_to_use_direct_io, max_read_buffer_size,
reader->getAvgValueSizeHints(), profile_callback);
if (prewhere_info)
pre_reader = std::make_unique<MergeTreeReader>(
path, task->data_part, task->pre_columns, owned_uncompressed_cache.get(), owned_mark_cache.get(), save_marks_in_cache,
storage, rest_mark_ranges, min_bytes_to_use_direct_io,
max_read_buffer_size, pre_reader->getAvgValueSizeHints(), profile_callback);
}
}
2018-10-04 08:52:56 +00:00
last_readed_part_path = path;
return true;
2016-11-20 12:43:20 +00:00
}
2018-11-29 09:19:42 +00:00
MergeTreeThreadSelectBlockInputStream::~MergeTreeThreadSelectBlockInputStream() = default;
2016-11-20 12:43:20 +00:00
}