Stable 24.8 Backport of #83096 - A bunch of micro-optimizations to speed up small queries

src/Processors/Executors/ExecutingGraph.cpp

@@ -96,7 +96,7 @@ bool ExecutingGraph::addEdges(uint64_t node)
     return was_edge_added;
 }
 
-bool ExecutingGraph::expandPipeline(std::stack<uint64_t> & stack, uint64_t pid)
+bool ExecutingGraph::expandPipeline(boost::container::devector<uint64_t> & stack, uint64_t pid)
 {
     auto & cur_node = *nodes[pid];
     Processors new_processors;
@@ -174,7 +174,7 @@ bool ExecutingGraph::expandPipeline(std::stack<uint64_t> & 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<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);
 
@@ -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<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();
                     }
 
@@ -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);
             }
         }
     }

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>;
@@ -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<uint64_t> & stack, uint64_t pid);
+    bool 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

@@ -49,7 +49,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/ConcurrencyControl.h>
 #include <Common/AllocatorWithMemoryTracking.h>
 
-#include <deque>
 #include <queue>
 #include <memory>
 
+#include <boost/container/devector.hpp>
+
 
 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<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

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

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;