25.8 stable backport of #83096: A bunch of micro-optimizations to speed up small queries

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<uint64_t> & stack, uint64_t pid)
+ExecutingGraph::UpdateNodeStatus ExecutingGraph::expandPipeline(boost::container::devector<uint64_t> & stack, uint64_t pid)
 {
     auto & cur_node = *nodes[pid];
     Processors new_processors;
@@ -178,7 +178,7 @@ ExecutingGraph::UpdateNodeStatus ExecutingGraph::expandPipeline(std::stack<uint6
         if (node.status == ExecutingGraph::ExecStatus::Idle)
         {
             node.status = ExecutingGraph::ExecStatus::Preparing;
-            stack.push(updated_node);
+            stack.push_front(updated_node);
         }
     }
 
@@ -213,9 +213,9 @@ void ExecutingGraph::initializeExecution(Queue & queue, Queue & async_queue)
 
 ExecutingGraph::UpdateNodeStatus ExecutingGraph::updateNode(uint64_t pid, Queue & queue, Queue & async_queue)
 {
-    std::stack<Edge *> updated_edges;
-    std::stack<uint64_t> updated_processors;
-    updated_processors.push(pid);
+    boost::container::devector<Edge *> updated_edges;
+    boost::container::devector<uint64_t> 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<ExecutingGraph::Edge *>(*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<ExecutingGraph::Edge *>(*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);
             }
         }
     }

src/Processors/Executors/ExecutingGraph.h

@@ -6,9 +6,9 @@
 #include <Common/AllocatorWithMemoryTracking.h>
 #include <mutex>
 #include <queue>
-#include <stack>
 #include <vector>
 
+#include <boost/container/devector.hpp>
 
 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<ExecutingGraph::Node *, AllocatorWithMemoryTracking<ExecutingGraph::Node *>>;
+    using DequeWithMemoryTracker = boost::container::devector<ExecutingGraph::Node *, AllocatorWithMemoryTracking<ExecutingGraph::Node *>>;
     using Queue = std::queue<ExecutingGraph::Node *, DequeWithMemoryTracker>;
 
     using NodePtr = std::unique_ptr<Node>;
@@ -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<uint64_t> & stack, uint64_t pid);
+    UpdateNodeStatus expandPipeline(boost::container::devector<uint64_t> & stack, uint64_t pid);
 
     std::shared_ptr<Processors> processors;
     std::vector<bool> source_processors;

src/Processors/Executors/ExecutorTasks.h

@@ -69,7 +69,7 @@ class ExecutorTasks
     using Stack = std::stack<UInt64>;
     /// 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<ExecutingGraph::Node *, AllocatorWithMemoryTracking<ExecutingGraph::Node *>>;
+    using DequeWithMemoryTracker = boost::container::devector<ExecutingGraph::Node *, AllocatorWithMemoryTracking<ExecutingGraph::Node *>>;
     using Queue = std::queue<ExecutingGraph::Node *, DequeWithMemoryTracker>;
 
     void finish();

src/Processors/Executors/PipelineExecutor.h

@@ -7,10 +7,11 @@
 #include <Common/ISlotControl.h>
 #include <Common/AllocatorWithMemoryTracking.h>
 
-#include <deque>
 #include <queue>
 #include <memory>
 
+#include <boost/container/devector.hpp>
+
 
 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<ExecutingGraph::Node *, AllocatorWithMemoryTracking<ExecutingGraph::Node *>>;
+    using DequeWithMemoryTracker = boost::container::devector<ExecutingGraph::Node *, AllocatorWithMemoryTracking<ExecutingGraph::Node *>>;
     using Queue = std::queue<ExecutingGraph::Node *, DequeWithMemoryTracker>;
 
     void initializeExecution(size_t num_threads, bool concurrency_control); /// Initialize executor contexts and task_queue.

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<const LoadedMergeTreeDataPartInfoForReader *>(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

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;

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

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;