diff --git a/src/Processors/Executors/ExecutingGraph.cpp b/src/Processors/Executors/ExecutingGraph.cpp index 675dc07da9fa..69277bdeca80 100644 --- a/src/Processors/Executors/ExecutingGraph.cpp +++ b/src/Processors/Executors/ExecutingGraph.cpp @@ -100,7 +100,7 @@ bool ExecutingGraph::addEdges(uint64_t node) return was_edge_added; } -ExecutingGraph::UpdateNodeStatus ExecutingGraph::expandPipeline(std::stack & stack, uint64_t pid) +ExecutingGraph::UpdateNodeStatus ExecutingGraph::expandPipeline(boost::container::devector & stack, uint64_t pid) { auto & cur_node = *nodes[pid]; Processors new_processors; @@ -178,7 +178,7 @@ ExecutingGraph::UpdateNodeStatus ExecutingGraph::expandPipeline(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); @@ -225,8 +225,8 @@ ExecutingGraph::UpdateNodeStatus ExecutingGraph::updateNode(uint64_t pid, 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. @@ -246,7 +246,7 @@ ExecutingGraph::UpdateNodeStatus ExecutingGraph::updateNode(uint64_t pid, 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 @@ -256,8 +256,8 @@ ExecutingGraph::UpdateNodeStatus ExecutingGraph::updateNode(uint64_t pid, 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 @@ ExecutingGraph::UpdateNodeStatus ExecutingGraph::updateNode(uint64_t pid, 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(); } @@ -393,7 +393,7 @@ ExecutingGraph::UpdateNodeStatus ExecutingGraph::updateNode(uint64_t pid, 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 676e13fcc96b..8ef0e9527a5d 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; @@ -162,7 +162,7 @@ class ExecutingGraph /// Update graph after processor (pid) returned ExpandPipeline status. /// All new nodes and nodes with updated ports are pushed into stack. - UpdateNodeStatus expandPipeline(std::stack & stack, uint64_t pid); + UpdateNodeStatus 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 c60c70a09b4e..e5bfc83600d6 100644 --- a/src/Processors/Executors/ExecutorTasks.h +++ b/src/Processors/Executors/ExecutorTasks.h @@ -69,7 +69,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 2481f85e349f..75343b4f0cac 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 { @@ -109,7 +110,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 8ff7c6ca50b5..b6a3633e34b0 100644 --- a/src/Storages/MergeTree/IMergeTreeReader.cpp +++ b/src/Storages/MergeTree/IMergeTreeReader.cpp @@ -49,18 +49,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()) { const auto & column_to_read = columns_to_read.emplace_back(getColumnInPart(column)); serializations.emplace_back(getSerializationInPart(column)); @@ -80,7 +76,7 @@ const ValueSizeMap & IMergeTreeReader::getAvgValueSizeHints() const 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) @@ -90,7 +86,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)) @@ -124,15 +120,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); + /// 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; + } + } - should_evaluate_missing_defaults = std::any_of( - res_columns.begin(), res_columns.end(), [](const auto & column) { return column == nullptr; }); + 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) { @@ -313,7 +332,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) { @@ -322,21 +342,26 @@ void IMergeTreeReader::performRequiredConversions(Columns & res_columns) const "Expected {}, got {}", num_columns, res_columns.size()); } + /// We check types manually while iterating to avoid unnecessary copies Block copy_block; - auto name_and_type = requested_columns.begin(); - + 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_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.empty()) + return; + + DB::performRequiredConversions(copy_block, getColumns(), data_part_info_for_read->getContext(), storage_snapshot->metadata->getColumns().getDefaults()); /// 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); @@ -412,9 +437,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, size_t offset) const diff --git a/src/Storages/MergeTree/IMergeTreeReader.h b/src/Storages/MergeTree/IMergeTreeReader.h index c1e10ee35b06..e40c670aaf0e 100644 --- a/src/Storages/MergeTree/IMergeTreeReader.h +++ b/src/Storages/MergeTree/IMergeTreeReader.h @@ -60,8 +60,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; } @@ -135,7 +135,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; diff --git a/tests/queries/0_stateless/03172_format_settings_clauses.reference b/tests/queries/0_stateless/03172_format_settings_clauses.reference index 8a98b137f4b1..c14a4885a949 100644 --- a/tests/queries/0_stateless/03172_format_settings_clauses.reference +++ b/tests/queries/0_stateless/03172_format_settings_clauses.reference @@ -1,14 +1,33 @@ +-- { echoOn } +-- Take the following example: +SELECT 1 UNION ALL SELECT 2 FORMAT TSV; 1 2 +-- Each subquery can be put in parentheses and have its own settings: +(SELECT getSetting('max_block_size') SETTINGS max_block_size = 1) UNION ALL (SELECT getSetting('max_block_size') SETTINGS max_block_size = 2) FORMAT TSV; 1 2 +-- And the whole query can have settings: +(SELECT getSetting('max_block_size') SETTINGS max_block_size = 1) UNION ALL (SELECT getSetting('max_block_size') SETTINGS max_block_size = 2) FORMAT TSV SETTINGS max_block_size = 3; 1 2 +-- A single query with output is parsed in the same way as the UNION ALL chain: +SELECT getSetting('max_block_size') SETTINGS max_block_size = 1 FORMAT TSV SETTINGS max_block_size = 3; 1 +-- So while these forms have a slightly different meaning, they both exist: +SELECT getSetting('max_block_size') SETTINGS max_block_size = 1 FORMAT TSV; 1 +SELECT getSetting('max_block_size') FORMAT TSV SETTINGS max_block_size = 3; 3 +-- And due to this effect, the users expect that the FORMAT and SETTINGS may go in an arbitrary order. +-- But while this work: +(SELECT getSetting('max_block_size')) UNION ALL (SELECT getSetting('max_block_size')) FORMAT TSV SETTINGS max_block_size = 3; 3 3 +-- This does not work automatically, unless we explicitly allow different orders: +(SELECT getSetting('max_block_size')) UNION ALL (SELECT getSetting('max_block_size')) SETTINGS max_block_size = 3 FORMAT TSV; 3 3 +-- Inevitably, we allow this: +SELECT getSetting('max_block_size') SETTINGS max_block_size = 1 SETTINGS max_block_size = 3 FORMAT TSV; 1 diff --git a/tests/queries/0_stateless/03172_format_settings_clauses.sql b/tests/queries/0_stateless/03172_format_settings_clauses.sql index 0d1aa4dcfbb8..450afc5fe7d4 100644 --- a/tests/queries/0_stateless/03172_format_settings_clauses.sql +++ b/tests/queries/0_stateless/03172_format_settings_clauses.sql @@ -1,5 +1,5 @@ SET max_block_size = 10, max_threads = 1; - +-- { echoOn } -- Take the following example: SELECT 1 UNION ALL SELECT 2 FORMAT TSV;