thevar1able/ClickHouse
1
0
Fork 0
mirror of https://github.com/ClickHouse/ClickHouse.git synced 2024-11-22 15:42:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge

Browse Source
This commit is contained in:
Andrey Mironov 2015-09-16 20:49:08 +03:00
parent fd60ee37d3
commit 7b149be2c6
2 changed files with 26 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
dbms/include/DB/Storages/MergeTree/MergeTreeReader.h
View File

@ -24,33 +24,19 @@ namespace DB
*/ */
class MergeTreeReader class MergeTreeReader
{ {
typedef std::map<std::string, ColumnPtr> OffsetColumns; using OffsetColumns = std::map<std::string, ColumnPtr>;
using ValueSizeMap = std::map<std::string, double>;
public: public:
MergeTreeReader(const String & path_, /// Путь к куску MergeTreeReader(const String & path, /// Путь к куску
const MergeTreeData::DataPartPtr & data_part, const NamesAndTypesList & columns_, const MergeTreeData::DataPartPtr & data_part, const NamesAndTypesList & columns,
UncompressedCache * uncompressed_cache_, MarkCache * mark_cache_, UncompressedCache * uncompressed_cache, MarkCache * mark_cache,
MergeTreeData & storage_, const MarkRanges & all_mark_ranges, MergeTreeData & storage, const MarkRanges & all_mark_ranges,
size_t aio_threshold_, size_t max_read_buffer_size_) size_t aio_threshold, size_t max_read_buffer_size, const ValueSizeMap & avg_value_size_hints = ValueSizeMap{})
: uncompressed_cache(uncompressed_cache_), mark_cache(mark_cache_), storage(storage_), : avg_value_size_hints(avg_value_size_hints), path(path), data_part(data_part), columns(columns),
aio_threshold(aio_threshold_), max_read_buffer_size(max_read_buffer_size_) uncompressed_cache(uncompressed_cache), mark_cache(mark_cache), storage(storage),
all_mark_ranges(all_mark_ranges), aio_threshold(aio_threshold), max_read_buffer_size(max_read_buffer_size)
{ {
reconfigure(path_, data_part, columns_, all_mark_ranges);
}
/** Allows to use the same MergeTreeReader across multiple data parts and/or columns and/or ranges,
* all while preserving avg_value_size_hints (may and does significantly improve read times). */
void reconfigure(
const String & path, const MergeTreeData::DataPartPtr & data_part, const NamesAndTypesList & columns,
const MarkRanges & all_mark_ranges)
{
this->path = path;
this->data_part = data_part;
this->part_name = data_part->name;
this->columns = columns;
this->all_mark_ranges = all_mark_ranges;
this->streams.clear();
try try
{ {
if (!Poco::File(path).exists()) if (!Poco::File(path).exists())
@ -61,11 +47,13 @@ public:
} }
catch (...) catch (...)
{ {
storage.reportBrokenPart(part_name); storage.reportBrokenPart(data_part->name);
throw; throw;
} }
} }
const ValueSizeMap & getAvgValueSizeHints() const { return avg_value_size_hints; }
/** Если столбцов нет в блоке, добавляет их, если есть - добавляет прочитанные значения к ним в конец. /** Если столбцов нет в блоке, добавляет их, если есть - добавляет прочитанные значения к ним в конец.
* Не добавляет столбцы, для которых нет файлов. Чтобы их добавить, нужно вызвать fillMissingColumns. * Не добавляет столбцы, для которых нет файлов. Чтобы их добавить, нужно вызвать fillMissingColumns.
* В блоке должно быть либо ни одного столбца из columns, либо все, для которых есть файлы. * В блоке должно быть либо ни одного столбца из columns, либо все, для которых есть файлы.
@ -121,7 +109,7 @@ public:
catch (const Exception & e) catch (const Exception & e)
{ {
if (e.code() != ErrorCodes::MEMORY_LIMIT_EXCEEDED) if (e.code() != ErrorCodes::MEMORY_LIMIT_EXCEEDED)
storage.reportBrokenPart(part_name); storage.reportBrokenPart(data_part->name);
/// Более хорошая диагностика. /// Более хорошая диагностика.
throw Exception(e.message() + "\n(while reading from part " + path + " from mark " + toString(from_mark) + " to " throw Exception(e.message() + "\n(while reading from part " + path + " from mark " + toString(from_mark) + " to "
@ -129,7 +117,7 @@ public:
} }
catch (...) catch (...)
{ {
storage.reportBrokenPart(part_name); storage.reportBrokenPart(data_part->name);
throw; throw;
} }
@ -286,10 +274,9 @@ private:
typedef std::map<std::string, std::unique_ptr<Stream> > FileStreams; typedef std::map<std::string, std::unique_ptr<Stream> > FileStreams;
/// Используется в качестве подсказки, чтобы уменьшить количество реаллокаций при создании столбца переменной длины. /// Используется в качестве подсказки, чтобы уменьшить количество реаллокаций при создании столбца переменной длины.
std::map<std::string, double> avg_value_size_hints; ValueSizeMap avg_value_size_hints;
String path; String path;
MergeTreeData::DataPartPtr data_part; MergeTreeData::DataPartPtr data_part;
String part_name;
FileStreams streams; FileStreams streams;
/// Запрашиваемые столбцы. /// Запрашиваемые столбцы.

14
dbms/include/DB/Storages/MergeTree/MergeTreeThreadBlockInputStream.h
View File

@ -121,11 +121,17 @@ private:
} }
else else
{ {
/** reader and possible pre_reader were already created, just configure them to a new data part, ranges and /// retain avg_value_size_hints
* columns to preserve internal state. */ reader = std::make_unique<MergeTreeReader>(
reader->reconfigure(path, task->data_part, task->columns, task->mark_ranges); path, task->data_part, task->columns, owned_uncompressed_cache.get(), owned_mark_cache.get(),
storage, task->mark_ranges, min_bytes_to_use_direct_io, max_read_buffer_size,
reader->getAvgValueSizeHints());
if (prewhere_actions) if (prewhere_actions)
pre_reader->reconfigure(path, task->data_part, task->pre_columns, task->mark_ranges); pre_reader = std::make_unique<MergeTreeReader>(
path, task->data_part, task->pre_columns, owned_uncompressed_cache.get(),
owned_mark_cache.get(), storage, task->mark_ranges, min_bytes_to_use_direct_io,
max_read_buffer_size, pre_reader->getAvgValueSizeHints());
} }
return true; return true;