Add id/parent_ids into system.processors_profile_log

Signed-off-by: Azat Khuzhin <a.khuzhin@semrush.com>

docs/en/operations/system-tables/processors_profile_log.md

@@ -6,6 +6,8 @@ Columns:
 
 -   `event_date` ([Date](../../sql-reference/data-types/date.md)) — The date when the event happened.
 -   `event_time` ([DateTime64](../../sql-reference/data-types/datetime64.md)) — The date and time when the event happened.
+-   `id` ([UInt64](../../sql-reference/data-types/int-uint.md)) — ID of processor
+-   `parent_ids` ([Array(UInt64)](../../sql-reference/data-types/array.md)) — Parent processors IDs
 -   `query_id` ([String](../../sql-reference/data-types/string.md)) — ID of the query
 -   `name` ([LowCardinality(String)](../../sql-reference/data-types/lowcardinality.md)) — Name of the processor.
 -   `elapsed_us` ([UInt64](../../sql-reference/data-types/int-uint.md)) — Number of microseconds this processor was executed.

src/Interpreters/ProcessorsProfileLog.cpp

@@ -8,6 +8,7 @@
 #include <DataTypes/DataTypeString.h>
 #include <DataTypes/DataTypesNumber.h>
 #include <DataTypes/DataTypeNullable.h>
+#include <DataTypes/DataTypeArray.h>
 #include <base/logger_useful.h>
 
 #include <array>
@@ -23,6 +24,9 @@ NamesAndTypesList ProcessorProfileLogElement::getNamesAndTypes()
         {"event_time", std::make_shared<DataTypeDateTime>()},
         {"event_time_microseconds", std::make_shared<DataTypeDateTime64>(6)},
 
+        {"id", std::make_shared<DataTypeUInt64>()},
+        {"parent_ids", std::make_shared<DataTypeArray>(std::make_shared<DataTypeUInt64>())},
+
         {"query_id", std::make_shared<DataTypeString>()},
         {"name", std::make_shared<DataTypeLowCardinality>(std::make_shared<DataTypeString>())},
         {"elapsed_us", std::make_shared<DataTypeUInt64>()},
@@ -39,6 +43,15 @@ void ProcessorProfileLogElement::appendToBlock(MutableColumns & columns) const
     columns[i++]->insert(event_time);
     columns[i++]->insert(event_time_microseconds);
 
+    columns[i++]->insert(id);
+    {
+        Array parent_ids_array;
+        parent_ids_array.reserve(parent_ids.size());
+        for (const UInt64 parent : parent_ids)
+            parent_ids_array.emplace_back(parent);
+        columns[i++]->insert(parent_ids_array);
+    }
+
     columns[i++]->insertData(query_id.data(), query_id.size());
     columns[i++]->insertData(processor_name.data(), processor_name.size());
     columns[i++]->insert(elapsed_us);

src/Interpreters/ProcessorsProfileLog.h

@@ -13,8 +13,12 @@ struct ProcessorProfileLogElement
     time_t event_time{};
     Decimal64 event_time_microseconds{};
 
+    UInt64 id;
+    std::vector<UInt64> parent_ids;
+
     String query_id;
     String processor_name;
+
     /// Milliseconds spend in IProcessor::work()
     UInt32 elapsed_us{};
     /// IProcessor::NeedData

src/Interpreters/executeQuery.cpp

@@ -877,12 +877,31 @@ static std::tuple<ASTPtr, BlockIO> executeQueryImpl(
                         processor_elem.event_time_microseconds = time_in_microseconds(finish_time);
                         processor_elem.query_id = elem.client_info.current_query_id;
 
+                        auto get_proc_id = [](const IProcessor & proc) -> UInt64
+                        {
+                            return reinterpret_cast<std::uintptr_t>(&proc);
+                        };
+
                         for (const auto & processor : query_pipeline.getProcessors())
                         {
+                            std::vector<UInt64> parents;
+                            for (const auto & port : processor->getOutputs())
+                            {
+                                if (!port.isConnected())
+                                    continue;
+                                const IProcessor & next = port.getInputPort().getProcessor();
+                                parents.push_back(get_proc_id(next));
+                            }
+
+                            processor_elem.id = get_proc_id(*processor);
+                            processor_elem.parent_ids = std::move(parents);
+
                             processor_elem.processor_name = processor->getName();
+
                             processor_elem.elapsed_us = processor->getElapsedUs();
                             processor_elem.input_wait_elapsed_us = processor->getInputWaitElapsedUs();
                             processor_elem.output_wait_elapsed_us = processor->getOutputWaitElapsedUs();
+
                             processors_profile_log->add(processor_elem);
                         }
                     }