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

Disable delayed streams for Vertical merges for non-remote disks.

Browse Source
This commit is contained in:
Nikolai Kochetov 2023-02-09 19:39:34 +00:00
parent 859f528fe1
commit 98765cc20b
2 changed files with 19 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
src/Storages/MergeTree/MergeTask.cpp
View File

@ -458,6 +458,17 @@ bool MergeTask::VerticalMergeStage::prepareVerticalMergeForAllColumns() const
ctx->column_num_for_vertical_merge = 0;
ctx->it_name_and_type = global_ctx->gathering_columns.cbegin();
const auto & settings = global_ctx->context->getSettingsRef();
size_t max_delayed_streams = 0;
if (global_ctx->new_data_part->getDataPartStorage().supportParallelWrite())
{
if (settings.max_insert_delayed_streams_for_parallel_write.changed)
max_delayed_streams = settings.max_insert_delayed_streams_for_parallel_write;
else
max_delayed_streams = DEFAULT_DELAYED_STREAMS_FOR_PARALLEL_WRITE;
}
ctx->max_delayed_streams = max_delayed_streams;
return false;
}
@ -549,6 +560,12 @@ void MergeTask::VerticalMergeStage::finalizeVerticalMergeForOneColumn() const
ctx->delayed_streams.emplace_back(std::move(ctx->column_to));
while (ctx->delayed_streams.size() > ctx->max_delayed_streams)
{
ctx->delayed_streams.front()->finish(ctx->need_sync);
ctx->delayed_streams.pop_front();
}
if (global_ctx->rows_written != ctx->column_elems_written)
{
throw Exception(ErrorCodes::LOGICAL_ERROR, "Written {} elements of column {}, but {} rows of PK columns",

3
src/Storages/MergeTree/MergeTask.h
View File

@ -281,7 +281,8 @@ private:
Float64 progress_before = 0;
std::unique_ptr<MergedColumnOnlyOutputStream> column_to{nullptr};
std::vector<std::unique_ptr<MergedColumnOnlyOutputStream>> delayed_streams;
size_t max_delayed_streams = 0;
std::list<std::unique_ptr<MergedColumnOnlyOutputStream>> delayed_streams;
size_t column_elems_written{0};
QueryPipeline column_parts_pipeline;
std::unique_ptr<PullingPipelineExecutor> executor;