diff --git a/src/Processors/Executors/ExecutingGraph.cpp b/src/Processors/Executors/ExecutingGraph.cpp index 6d5b60d815911..174108fc9776a 100644 --- a/src/Processors/Executors/ExecutingGraph.cpp +++ b/src/Processors/Executors/ExecutingGraph.cpp @@ -96,7 +96,7 @@ bool ExecutingGraph::addEdges(uint64_t node) return was_edge_added; } -bool ExecutingGraph::expandPipeline(std::stack & stack, uint64_t pid) +bool ExecutingGraph::expandPipeline(boost::container::devector & stack, uint64_t pid) { auto & cur_node = *nodes[pid]; Processors new_processors; @@ -174,7 +174,7 @@ bool ExecutingGraph::expandPipeline(std::stack & stack, uint64_t pid) if (node.status == ExecutingGraph::ExecStatus::Idle) { node.status = ExecutingGraph::ExecStatus::Preparing; - stack.push(updated_node); + stack.push_front(updated_node); } } @@ -215,9 +215,9 @@ void ExecutingGraph::initializeExecution(Queue & queue) bool ExecutingGraph::updateNode(uint64_t pid, Queue & queue, Queue & async_queue) { - std::stack updated_edges; - std::stack updated_processors; - updated_processors.push(pid); + boost::container::devector updated_edges; + boost::container::devector updated_processors; + updated_processors.push_back(pid); std::shared_lock read_lock(nodes_mutex); @@ -227,8 +227,8 @@ bool ExecutingGraph::updateNode(uint64_t pid, Queue & queue, Queue & async_queue if (updated_processors.empty()) { - auto * edge = updated_edges.top(); - updated_edges.pop(); + auto * edge = updated_edges.front(); + updated_edges.pop_front(); /// Here we have ownership on edge, but node can be concurrently accessed. @@ -248,7 +248,7 @@ bool ExecutingGraph::updateNode(uint64_t pid, Queue & queue, Queue & async_queue if (status == ExecutingGraph::ExecStatus::Idle) { node.status = ExecutingGraph::ExecStatus::Preparing; - updated_processors.push(edge->to); + updated_processors.push_front(edge->to); stack_top_lock = std::move(lock); } else @@ -258,8 +258,8 @@ bool ExecutingGraph::updateNode(uint64_t pid, Queue & queue, Queue & async_queue if (!updated_processors.empty()) { - pid = updated_processors.top(); - updated_processors.pop(); + pid = updated_processors.front(); + updated_processors.pop_front(); /// In this method we have ownership on node. auto & node = *nodes[pid]; @@ -364,14 +364,14 @@ bool ExecutingGraph::updateNode(uint64_t pid, Queue & queue, Queue & async_queue for (auto it = node.post_updated_output_ports.rbegin(); it != node.post_updated_output_ports.rend(); ++it) { auto * edge = static_cast(*it); - updated_edges.push(edge); + updated_edges.push_front(edge); edge->update_info.trigger(); } for (auto it = node.post_updated_input_ports.rbegin(); it != node.post_updated_input_ports.rend(); ++it) { auto * edge = static_cast(*it); - updated_edges.push(edge); + updated_edges.push_front(edge); edge->update_info.trigger(); } @@ -392,7 +392,7 @@ bool ExecutingGraph::updateNode(uint64_t pid, Queue & queue, Queue & async_queue read_lock.lock(); /// Add itself back to be prepared again. - updated_processors.push(pid); + updated_processors.push_front(pid); } } } diff --git a/src/Processors/Executors/ExecutingGraph.h b/src/Processors/Executors/ExecutingGraph.h index 71dcd360a2c03..e30e28549be45 100644 --- a/src/Processors/Executors/ExecutingGraph.h +++ b/src/Processors/Executors/ExecutingGraph.h @@ -6,9 +6,9 @@ #include #include #include -#include #include +#include namespace DB { @@ -120,7 +120,7 @@ class ExecutingGraph /// This queue can grow a lot and lead to OOM. That is why we use non-default /// allocator for container which throws exceptions in operator new - using DequeWithMemoryTracker = std::deque>; + using DequeWithMemoryTracker = boost::container::devector>; using Queue = std::queue; using NodePtr = std::unique_ptr; @@ -155,7 +155,7 @@ class ExecutingGraph /// Update graph after processor (pid) returned ExpandPipeline status. /// All new nodes and nodes with updated ports are pushed into stack. - bool expandPipeline(std::stack & stack, uint64_t pid); + bool expandPipeline(boost::container::devector & stack, uint64_t pid); std::shared_ptr processors; std::vector source_processors; diff --git a/src/Processors/Executors/ExecutorTasks.h b/src/Processors/Executors/ExecutorTasks.h index b2201873edf6a..c0426c8c97e38 100644 --- a/src/Processors/Executors/ExecutorTasks.h +++ b/src/Processors/Executors/ExecutorTasks.h @@ -49,7 +49,7 @@ class ExecutorTasks using Stack = std::stack; /// This queue can grow a lot and lead to OOM. That is why we use non-default /// allocator for container which throws exceptions in operator new - using DequeWithMemoryTracker = std::deque>; + using DequeWithMemoryTracker = boost::container::devector>; using Queue = std::queue; void finish(); diff --git a/src/Processors/Executors/PipelineExecutor.h b/src/Processors/Executors/PipelineExecutor.h index ae119355cb5a8..f54315e6ac5d2 100644 --- a/src/Processors/Executors/PipelineExecutor.h +++ b/src/Processors/Executors/PipelineExecutor.h @@ -7,10 +7,11 @@ #include #include -#include #include #include +#include + namespace DB { @@ -93,7 +94,7 @@ class PipelineExecutor /// This queue can grow a lot and lead to OOM. That is why we use non-default /// allocator for container which throws exceptions in operator new - using DequeWithMemoryTracker = std::deque>; + using DequeWithMemoryTracker = boost::container::devector>; using Queue = std::queue; void initializeExecution(size_t num_threads, bool concurrency_control); /// Initialize executor contexts and task_queue. diff --git a/src/Storages/MergeTree/IMergeTreeReader.cpp b/src/Storages/MergeTree/IMergeTreeReader.cpp index 5916f50005d52..4b2b9d25eeb51 100644 --- a/src/Storages/MergeTree/IMergeTreeReader.cpp +++ b/src/Storages/MergeTree/IMergeTreeReader.cpp @@ -45,18 +45,14 @@ IMergeTreeReader::IMergeTreeReader( , storage_snapshot(storage_snapshot_) , all_mark_ranges(all_mark_ranges_) , alter_conversions(data_part_info_for_read->getAlterConversions()) - /// For wide parts convert plain arrays of Nested to subcolumns - /// to allow to use shared offset column from cache. , original_requested_columns(columns_) - , requested_columns(data_part_info_for_read->isWidePart() - ? Nested::convertToSubcolumns(columns_) - : columns_) + , converted_requested_columns(Nested::convertToSubcolumns(columns_)) , virtual_fields(virtual_fields_) { - columns_to_read.reserve(requested_columns.size()); - serializations.reserve(requested_columns.size()); + columns_to_read.reserve(getColumns().size()); + serializations.reserve(getColumns().size()); - for (const auto & column : requested_columns) + for (const auto & column : getColumns()) { columns_to_read.emplace_back(getColumnInPart(column)); serializations.emplace_back(getSerializationInPart(column)); @@ -77,7 +73,7 @@ const IMergeTreeReader::ValueSizeMap & IMergeTreeReader::getAvgValueSizeHints() void IMergeTreeReader::fillVirtualColumns(Columns & columns, size_t rows) const { - chassert(columns.size() == requested_columns.size()); + chassert(columns.size() == getColumns().size()); const auto * loaded_part_info = typeid_cast(data_part_info_for_read.get()); if (!loaded_part_info) @@ -87,7 +83,7 @@ void IMergeTreeReader::fillVirtualColumns(Columns & columns, size_t rows) const const auto & storage_columns = storage_snapshot->metadata->getColumns(); const auto & virtual_columns = storage_snapshot->virtual_columns; - auto it = requested_columns.begin(); + auto it = getColumns().begin(); for (size_t pos = 0; pos < columns.size(); ++pos, ++it) { if (columns[pos] || storage_columns.has(it->name)) @@ -121,15 +117,38 @@ void IMergeTreeReader::fillMissingColumns(Columns & res_columns, bool & should_e { try { - NamesAndTypesList available_columns(columns_to_read.begin(), columns_to_read.end()); - DB::fillMissingColumns( - res_columns, num_rows, - Nested::convertToSubcolumns(requested_columns), - Nested::convertToSubcolumns(available_columns), - partially_read_columns, storage_snapshot->metadata); - - should_evaluate_missing_defaults = std::any_of( - res_columns.begin(), res_columns.end(), [](const auto & column) { return column == nullptr; }); + /// Do a quick check to see if doing any work is necessary to begin with + size_t num_columns_to_read = res_columns.size(); + bool missing = false; + auto requested_column = converted_requested_columns.begin(); + for (size_t i = 0; i < num_columns_to_read; i++, requested_column++) + { + if (!res_columns[i] || partially_read_columns.contains(requested_column->name)) + { + missing = true; + break; + } + } + + if (!missing) + { + should_evaluate_missing_defaults = false; + } + else + { + NamesAndTypesList available_columns(columns_to_read.begin(), columns_to_read.end()); + + DB::fillMissingColumns( + res_columns, + num_rows, + converted_requested_columns, + Nested::convertToSubcolumns(available_columns), + partially_read_columns, + storage_snapshot->metadata); + + should_evaluate_missing_defaults + = std::any_of(res_columns.begin(), res_columns.end(), [](const auto & column) { return column == nullptr; }); + } } catch (Exception & e) { @@ -284,7 +303,8 @@ void IMergeTreeReader::performRequiredConversions(Columns & res_columns) const { try { - size_t num_columns = requested_columns.size(); + const NamesAndTypesList & required_columns = getColumns(); + size_t num_columns = required_columns.size(); if (res_columns.size() != num_columns) { @@ -293,23 +313,29 @@ void IMergeTreeReader::performRequiredConversions(Columns & res_columns) const "Expected {}, got {}", num_columns, res_columns.size()); } - Block copy_block; - auto name_and_type = requested_columns.begin(); + const auto & column_defaults = storage_snapshot->metadata->getColumns().getDefaults(); + /// Keep columns used by DEFAULT expressions when converting Nullable to non-Nullable. + /// Otherwise, check types manually to avoid unnecessary copies. + Block copy_block; + auto name_and_type = getColumns().begin(); for (size_t pos = 0; pos < num_columns; ++pos, ++name_and_type) { if (res_columns[pos] == nullptr) continue; - - copy_block.insert({res_columns[pos], getColumnInPart(*name_and_type).type, name_and_type->name}); + auto column_in_part = getColumnInPart(*name_and_type); + if (column_defaults.empty() && column_in_part.type->equals(*name_and_type->type)) + continue; + copy_block.insert({res_columns[pos], column_in_part.type, name_and_type->name}); } - DB::performRequiredConversions(copy_block, requested_columns, - data_part_info_for_read->getContext(), - storage_snapshot->metadata->getColumns().getDefaults()); + if (!copy_block.columns()) + return; + + DB::performRequiredConversions(copy_block, getColumns(), data_part_info_for_read->getContext(), column_defaults); /// Move columns from block. - name_and_type = requested_columns.begin(); + name_and_type = getColumns().begin(); for (size_t pos = 0; pos < num_columns; ++pos, ++name_and_type) if (copy_block.has(name_and_type->name)) res_columns[pos] = std::move(copy_block.getByName(name_and_type->name).column); @@ -385,9 +411,13 @@ IMergeTreeReader::findColumnForOffsets(const NameAndTypePair & required_column) void IMergeTreeReader::checkNumberOfColumns(size_t num_columns_to_read) const { - if (num_columns_to_read != requested_columns.size()) - throw Exception(ErrorCodes::LOGICAL_ERROR, "invalid number of columns passed to MergeTreeReader::readRows. " - "Expected {}, got {}", requested_columns.size(), num_columns_to_read); + if (num_columns_to_read != getColumns().size()) + throw Exception( + ErrorCodes::LOGICAL_ERROR, + "invalid number of columns passed to MergeTreeReader::readRows. " + "Expected {}, got {}", + converted_requested_columns.size(), + num_columns_to_read); } String IMergeTreeReader::getMessageForDiagnosticOfBrokenPart(size_t from_mark, size_t max_rows_to_read) const diff --git a/src/Storages/MergeTree/IMergeTreeReader.h b/src/Storages/MergeTree/IMergeTreeReader.h index 4be4c109a1afa..c5b039e7b0afd 100644 --- a/src/Storages/MergeTree/IMergeTreeReader.h +++ b/src/Storages/MergeTree/IMergeTreeReader.h @@ -56,8 +56,8 @@ class IMergeTreeReader : private boost::noncopyable /// then try to perform conversions of columns. void performRequiredConversions(Columns & res_columns) const; - const NamesAndTypesList & getColumns() const { return requested_columns; } - size_t numColumnsInResult() const { return requested_columns.size(); } + ALWAYS_INLINE const NamesAndTypesList & getColumns() const { return data_part_info_for_read->isWidePart() ? converted_requested_columns : original_requested_columns; } + size_t numColumnsInResult() const { return getColumns().size(); } size_t getFirstMarkToRead() const { return all_mark_ranges.front().begin; } @@ -126,7 +126,7 @@ class IMergeTreeReader : private boost::noncopyable NamesAndTypesList original_requested_columns; /// The same as above but with converted Arrays to subcolumns of Nested. - NamesAndTypesList requested_columns; + NamesAndTypesList converted_requested_columns; /// Fields of virtual columns that were filled in previous stages. VirtualFields virtual_fields;