Add system.user_processes table

src/Storages/System/StorageSystemUserProcesses.cpp

@@ -0,0 +1,60 @@
+#include <Columns/ColumnArray.h>
+#include <Columns/ColumnsNumber.h>
+#include <Core/Settings.h>
+#include <DataTypes/DataTypeArray.h>
+#include <DataTypes/DataTypeFactory.h>
+#include <DataTypes/DataTypeMap.h>
+#include <DataTypes/DataTypeString.h>
+#include <DataTypes/DataTypesNumber.h>
+#include <Interpreters/Context.h>
+#include <Interpreters/ProcessList.h>
+#include <Interpreters/ProfileEventsExt.h>
+#include <Storages/System/StorageSystemUserProcesses.h>
+#include <Common/IPv6ToBinary.h>
+#include <Common/typeid_cast.h>
+
+
+namespace DB
+{
+
+NamesAndTypesList StorageSystemUserProcesses::getNamesAndTypes()
+{
+    return {
+        {"user", std::make_shared<DataTypeString>()},
+        {"memory_usage", std::make_shared<DataTypeInt64>()},
+        {"peak_memory_usage", std::make_shared<DataTypeInt64>()},
+        {"ProfileEvents", std::make_shared<DataTypeMap>(std::make_shared<DataTypeString>(), std::make_shared<DataTypeUInt64>())},
+    };
+}
+
+NamesAndAliases StorageSystemUserProcesses::getNamesAndAliases()
+{
+    return {
+        {"ProfileEvents.Names", {std::make_shared<DataTypeArray>(std::make_shared<DataTypeString>())}, "mapKeys(ProfileEvents)"},
+        {"ProfileEvents.Values", {std::make_shared<DataTypeArray>(std::make_shared<DataTypeUInt64>())}, "mapValues(ProfileEvents)"}};
+}
+
+void StorageSystemUserProcesses::fillData(MutableColumns & res_columns, ContextPtr context, const SelectQueryInfo &) const
+{
+    const auto user_info = context->getProcessList().getUserInfo(true);
+
+    for (const auto & [user, info] : user_info)
+    {
+        size_t i = 0;
+
+        res_columns[i++]->insert(user);
+        res_columns[i++]->insert(info.memory_usage);
+        res_columns[i++]->insert(info.peak_memory_usage);
+        {
+            IColumn * column = res_columns[i++].get();
+
+            if (info.profile_counters)
+                ProfileEvents::dumpToMapColumn(*info.profile_counters, column, true);
+            else
+            {
+                column->insertDefault();
+            }
+        }
+    }
+}
+}

src/Storages/System/StorageSystemUserProcesses.h

@@ -0,0 +1,29 @@
+#pragma once
+
+#include <Storages/System/IStorageSystemOneBlock.h>
+
+
+namespace DB
+{
+
+class Context;
+
+
+/** Implements `processes` system table, which allows you to get information about the queries that are currently executing.
+  */
+class StorageSystemUserProcesses final : public IStorageSystemOneBlock<StorageSystemUserProcesses>
+{
+public:
+    std::string getName() const override { return "SystemUserProcesses"; }
+
+    static NamesAndTypesList getNamesAndTypes();
+
+    static NamesAndAliases getNamesAndAliases();
+
+protected:
+    using IStorageSystemOneBlock::IStorageSystemOneBlock;
+
+    void fillData(MutableColumns & res_columns, ContextPtr context, const SelectQueryInfo & query_info) const override;
+};
+
+}

src/Storages/System/attachSystemTables.cpp

@@ -36,6 +36,7 @@
 #include <Storages/System/StorageSystemPartsColumns.h>
 #include <Storages/System/StorageSystemProjectionPartsColumns.h>
 #include <Storages/System/StorageSystemProcesses.h>
+#include <Storages/System/StorageSystemUserProcesses.h>
 #include <Storages/System/StorageSystemReplicas.h>
 #include <Storages/System/StorageSystemReplicationQueue.h>
 #include <Storages/System/StorageSystemDistributionQueue.h>
@@ -185,6 +186,7 @@ void attachSystemTablesServer(ContextPtr context, IDatabase & system_database, b
     attach<StorageSystemRemoteDataPaths>(context, system_database, "remote_data_paths");
     attach<StorageSystemCertificates>(context, system_database, "certificates");
     attach<StorageSystemNamedCollections>(context, system_database, "named_collections");
+    attach<StorageSystemUserProcesses>(context, system_database, "user_processes");
 
     if (has_zookeeper)
     {