Merge pull request ClickHouse#78412 from ClickHouse/transposed_metric_log

Allow to use transposed metric log

Author: alesapin

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

@@ -54,6 +54,13 @@ CurrentMetric_LocalThreadActive:                                 0
 CurrentMetric_DistributedFilesToInsert:                          0
 ```
 
+**Schema**
+This table can be configured with different schema types using the XML tag `<schema_type>`. The default schema type is `wide`, where each metric or profile event is stored as a separate column. This schema is the most performant and efficient for single-column reads.
+
+The `transposed` schema stores data in a format similar to `system.asynchronous_metric_log`, where metrics and events are stored as rows. This schema is useful for low-resource setups because it reduces resource consumption during merges.
+
+There is also a compatibility schema, `transposed_with_wide_view`, which stores actual data in a table with the transposed schema (`system.transposed_metric_log`) and creates a view on top of it using the wide schema. This view queries the transposed table, making it useful for migrating from the `wide` schema to the `transposed` schema.
+
 **See also**
 
 - [metric_log setting](../../operations/server-configuration-parameters/settings.md#metric_log) — Enabling and disabling the setting.

src/Common/SystemLogBase.cpp

@@ -2,6 +2,7 @@
 #include <Interpreters/CrashLog.h>
 #include <Interpreters/ErrorLog.h>
 #include <Interpreters/MetricLog.h>
+#include <Interpreters/TransposedMetricLog.h>
 #include <Interpreters/LatencyLog.h>
 #include <Interpreters/OpenTelemetrySpanLog.h>
 #include <Interpreters/PartLog.h>

src/Common/SystemLogBase.h

@@ -83,6 +83,8 @@ class ISystemLog
 
     virtual void savingThreadFunction() = 0;
 
+    virtual bool mustBePreparedAtStartup() const { return false; }
+
 protected:
     std::mutex thread_mutex;
     std::unique_ptr<ThreadFromGlobalPool> saving_thread;

src/Interpreters/Context.cpp

@@ -4761,6 +4761,16 @@ std::shared_ptr<MetricLog> Context::getMetricLog() const
     return shared->system_logs->metric_log;
 }
 
+std::shared_ptr<TransposedMetricLog> Context::getTransposedMetricLog() const
+{
+     SharedLockGuard lock(shared->mutex);
+
+    if (!shared->system_logs)
+        return {};
+
+    return shared->system_logs->transposed_metric_log;
+}
+
 
 std::shared_ptr<LatencyLog> Context::getLatencyLog() const
 {

src/Interpreters/Context.h

@@ -116,6 +116,7 @@ class PartLog;
 class TextLog;
 class TraceLog;
 class MetricLog;
+class TransposedMetricLog;
 class LatencyLog;
 class AsynchronousMetricLog;
 class OpenTelemetrySpanLog;
@@ -1256,6 +1257,7 @@ class Context: public ContextData, public std::enable_shared_from_this<Context>
     std::shared_ptr<TraceLog> getTraceLog() const;
     std::shared_ptr<TextLog> getTextLog() const;
     std::shared_ptr<MetricLog> getMetricLog() const;
+    std::shared_ptr<TransposedMetricLog> getTransposedMetricLog() const;
     std::shared_ptr<LatencyLog> getLatencyLog() const;
     std::shared_ptr<AsynchronousMetricLog> getAsynchronousMetricLog() const;
     std::shared_ptr<OpenTelemetrySpanLog> getOpenTelemetrySpanLog() const;

src/Interpreters/PeriodicLog.cpp

@@ -2,6 +2,7 @@
 #include <Common/SystemLogBase.h>
 #include <Interpreters/ErrorLog.h>
 #include <Interpreters/MetricLog.h>
+#include <Interpreters/TransposedMetricLog.h>
 #include <Interpreters/PeriodicLog.h>
 #include <Interpreters/QueryMetricLog.h>
 #include <Interpreters/LatencyLog.h>

src/Interpreters/PeriodicLog.h

@@ -9,6 +9,7 @@
 #define SYSTEM_PERIODIC_LOG_ELEMENTS(M) \
     M(ErrorLogElement) \
     M(MetricLogElement) \
+    M(TransposedMetricLogElement) \
     M(LatencyLogElement) \
 
 

src/Interpreters/SystemLog.cpp

@@ -21,6 +21,7 @@
 #include <Interpreters/InterpreterInsertQuery.h>
 #include <Interpreters/InterpreterRenameQuery.h>
 #include <Interpreters/MetricLog.h>
+#include <Interpreters/TransposedMetricLog.h>
 #include <Interpreters/LatencyLog.h>
 #include <Interpreters/OpenTelemetrySpanLog.h>
 #include <Interpreters/PartLog.h>
@@ -158,7 +159,11 @@ std::shared_ptr<TSystemLog> createSystemLog(
     SystemLogSettings log_settings;
 
     log_settings.queue_settings.database = config.getString(config_prefix + ".database", default_database_name);
-    log_settings.queue_settings.table = config.getString(config_prefix + ".table", default_table_name);
+
+    if (default_table_name != "transposed_metric_log")
+        log_settings.queue_settings.table = config.getString(config_prefix + ".table", default_table_name);
+    else
+        log_settings.queue_settings.table = default_table_name;
 
     if (log_settings.queue_settings.database != default_database_name)
     {
@@ -213,9 +218,26 @@ std::shared_ptr<TSystemLog> createSystemLog(
         if (!ttl.empty())
             log_settings.engine += " TTL " + ttl;
 
-        /// ORDER BY expr is necessary.
-        String order_by = config.getString(config_prefix + ".order_by", TSystemLog::getDefaultOrderBy());
-        log_settings.engine += " ORDER BY (" + order_by + ")";
+        if (std::is_same_v<TSystemLog, TransposedMetricLog>)
+        {
+            auto schema = config.getString(config_prefix + ".schema_type", "wide");
+            /// NOTE, fixed schema, it's not allowed to change order by for efficiency reasons
+            if (schema == "transposed_with_wide_view")
+            {
+                log_settings.engine += std::string{" ORDER BY ("} + TSystemLog::getDefaultOrderBy() + ")";
+            }
+            else
+            {
+                String order_by = config.getString(config_prefix + ".order_by", TSystemLog::getDefaultOrderBy());
+                log_settings.engine += " ORDER BY (" + order_by + ")";
+            }
+        }
+        else
+        {
+            /// ORDER BY expr is necessary.
+            String order_by = config.getString(config_prefix + ".order_by", TSystemLog::getDefaultOrderBy());
+            log_settings.engine += " ORDER BY (" + order_by + ")";
+        }
 
         /// SETTINGS expr is not necessary.
         ///   https://clickhouse.com/docs/engines/table-engines/mergetree-family/mergetree#settings
@@ -279,7 +301,34 @@ std::shared_ptr<TSystemLog> createSystemLog(
 
     log_settings.queue_settings.turn_off_logger = TSystemLog::shouldTurnOffLogger();
 
+    if constexpr (std::is_same_v<TSystemLog, MetricLog>)
+    {
+        auto schema = config.getString(config_prefix + ".schema_type", "wide");
+        if (schema == "wide")
+            return std::make_shared<TSystemLog>(context, log_settings);
+
+        return {};
+    }
+    else if (std::is_same_v<TSystemLog, TransposedMetricLog>)
+    {
+        auto schema = config.getString(config_prefix + ".schema_type", "wide");
+        if (schema == "transposed_with_wide_view")
+        {
+            log_settings.view_name_for_transposed_metric_log = config.getString(config_prefix + ".table", "metric_log");
+            return std::make_shared<TSystemLog>(context, log_settings);
+        }
+        else if (schema == "transposed")
+        {
+            return std::make_shared<TSystemLog>(context, log_settings);
+        }
+        else if (schema != "wide")
+        {
+            throw Exception(ErrorCodes::BAD_ARGUMENTS, "Unknown schema type {} for metric_log table, only 'wide', 'transposed' and 'transposed_with_wide_view' are allowed", schema);
+        }
+    }
+
     return std::make_shared<TSystemLog>(context, log_settings);
+
 }
 
 
@@ -306,15 +355,28 @@ SystemLogs::SystemLogs(ContextPtr global_context, const Poco::Util::AbstractConf
 
     LIST_OF_ALL_SYSTEM_LOGS(CREATE_PUBLIC_MEMBERS)
 #undef CREATE_PUBLIC_MEMBERS
+
 /// NOLINTEND(bugprone-macro-parentheses)
 
+    if (metric_log == nullptr && config.has("metric_log") && config.getString("metric_log.schema_type", "wide") == "transposed")
+    {
+        transposed_metric_log = createSystemLog<TransposedMetricLog>(
+            global_context, "system", "metric_log", config, "metric_log", TransposedMetricLog::DESCRIPTION);
+    }
+    else if (metric_log == nullptr && config.has("metric_log") && config.getString("metric_log.schema_type", "wide") == "transposed_with_wide_view")
+    {
+        transposed_metric_log = createSystemLog<TransposedMetricLog>(
+            global_context, "system", "transposed_metric_log", config, "metric_log", TransposedMetricLog::DESCRIPTION);
+    }
+
+
     bool should_prepare = global_context->getServerSettings()[ServerSetting::prepare_system_log_tables_on_startup];
     try
     {
         for (auto & log : getAllLogs())
         {
             log->startup();
-            if (should_prepare)
+            if (should_prepare || log->mustBePreparedAtStartup())
                 log->prepareTable();
         }
     }
@@ -332,6 +394,13 @@ SystemLogs::SystemLogs(ContextPtr global_context, const Poco::Util::AbstractConf
         metric_log->startCollect("MetricLog", collect_interval_milliseconds);
     }
 
+    if (transposed_metric_log)
+    {
+        size_t collect_interval_milliseconds = config.getUInt64("metric_log.collect_interval_milliseconds",
+                                                                DEFAULT_METRIC_LOG_COLLECT_INTERVAL_MILLISECONDS);
+        transposed_metric_log->startCollect("TMetricLog", collect_interval_milliseconds);
+    }
+
     if (latency_log)
     {
         size_t collect_interval_milliseconds = config.getUInt64("latency_log.collect_interval_milliseconds",

src/Interpreters/SystemLog.h

@@ -14,6 +14,7 @@
     M(CrashLog,              crash_log,            "Contains information about stack traces for fatal errors. The table does not exist in the database by default, it is created only when fatal errors occur.") \
     M(TextLog,               text_log,             "Contains logging entries which are normally written to a log file or to stdout.") \
     M(MetricLog,             metric_log,           "Contains history of metrics values from tables system.metrics and system.events, periodically flushed to disk.") \
+    M(TransposedMetricLog,   transposed_metric_log,"Contains history of metrics values from tables system.metrics and system.events. Periodically flushed to disk. Transposed form of system.metric_log.") \
     M(LatencyLog,            latency_log,          "Contains history of all latency buckets, periodically flushed to disk.") \
     M(ErrorLog,              error_log,            "Contains history of error values from table system.errors, periodically flushed to disk.") \
     M(FilesystemCacheLog,    filesystem_cache_log, "Contains a history of all events occurred with filesystem cache for objects on a remote filesystem.") \
@@ -99,6 +100,7 @@ struct SystemLogSettings
 
     String engine;
     bool symbolize_traces = false;
+    std::string view_name_for_transposed_metric_log;
 };
 
 template <typename LogElement>