Merge pull request ClickHouse#79945 from Algunenano/micro_clickbench

Algunenano · web-flow · commit 48a32336e99d · 2025-05-08T16:37:21.000Z
Speedup queries with trivial count optimization
diff --git a/src/Common/OpenTelemetryTraceContext.h b/src/Common/OpenTelemetryTraceContext.h
@@ -153,7 +153,7 @@ struct TracingContextHolder
 using TracingContextHolderPtr = std::unique_ptr<TracingContextHolder>;
 
 /// A span holder that creates span automatically in a (function) scope if tracing is enabled.
-/// Once it's created or destructed, it automatically maitains the tracing context on the thread that it lives.
+/// Once it's created or destructed, it automatically maintains the tracing context on the thread that it lives.
 struct SpanHolder : public Span
 {
     explicit SpanHolder(std::string_view, SpanKind _kind = SpanKind::INTERNAL);
diff --git a/src/Core/SettingsQuirks.cpp b/src/Core/SettingsQuirks.cpp
@@ -52,6 +52,14 @@ namespace Setting
 {
     extern const SettingsBool async_query_sending_for_remote;
     extern const SettingsBool async_socket_for_remote;
+    extern const SettingsUInt64 input_format_parquet_max_block_size;
+    extern const SettingsUInt64 max_block_size;
+    extern const SettingsUInt64 max_insert_block_size;
+    extern const SettingsUInt64 min_insert_block_size_rows;
+    extern const SettingsUInt64 min_insert_block_size_bytes_for_materialized_views;
+    extern const SettingsUInt64 min_external_table_block_size_rows;
+    extern const SettingsUInt64 max_joined_block_size_rows;
+    extern const SettingsMaxThreads max_threads;
     extern const SettingsUInt64 query_profiler_cpu_time_period_ns;
     extern const SettingsUInt64 query_profiler_real_time_period_ns;
     extern const SettingsBool use_hedged_requests;
@@ -101,51 +109,43 @@ void applySettingsQuirks(Settings & settings, LoggerPtr log)
 
 void doSettingsSanityCheckClamp(Settings & current_settings, LoggerPtr log)
 {
-    auto get_current_value = [&current_settings](const std::string_view name) -> Field
-    {
-        Field current_value;
-        bool has_current_value = current_settings.tryGet(name, current_value);
-        chassert(has_current_value);
-        return current_value;
-    };
-
-    UInt64 max_threads = get_current_value("max_threads").safeGet<UInt64>();
+    UInt64 max_threads = current_settings[Setting::max_threads];
     UInt64 max_threads_max_value = 256 * getNumberOfCPUCoresToUse();
     if (max_threads > max_threads_max_value)
     {
         if (log)
             LOG_WARNING(log, "Sanity check: Too many threads requested ({}). Reduced to {}", max_threads, max_threads_max_value);
-        current_settings.set("max_threads", max_threads_max_value);
+        current_settings[Setting::max_threads] = max_threads_max_value;
     }
 
     static constexpr UInt64 max_sane_block_rows_size = 4294967296; // 2^32
 
     using namespace std::literals;
-    static constexpr std::array block_rows_settings{
-        "max_block_size"sv,
-        "max_insert_block_size"sv,
-        "min_insert_block_size_rows"sv,
-        "min_insert_block_size_bytes_for_materialized_views"sv,
-        "min_external_table_block_size_rows"sv,
-        "max_joined_block_size_rows"sv,
-        "input_format_parquet_max_block_size"sv};
-
-    for (auto const setting : block_rows_settings)
-    {
-        if (auto block_size = get_current_value(setting).safeGet<UInt64>();
-            block_size > max_sane_block_rows_size)
-        {
-            if (log)
-                LOG_WARNING(log, "Sanity check: '{}' value is too high ({}). Reduced to {}", setting, block_size, max_sane_block_rows_size);
-            current_settings.set(setting, max_sane_block_rows_size);
-        }
+#define CHECK_MAX_VALUE(SETTING_VALUE) \
+    if (UInt64 block_size = current_settings[Setting::SETTING_VALUE]; block_size > max_sane_block_rows_size) \
+    { \
+        if (log) \
+            LOG_WARNING( \
+                log, "Sanity check: '{}' value is too high ({}). Reduced to {}", #SETTING_VALUE, block_size, max_sane_block_rows_size); \
+        current_settings[Setting::SETTING_VALUE] = max_sane_block_rows_size; \
     }
 
-    if (auto max_block_size = get_current_value("max_block_size").safeGet<UInt64>(); max_block_size == 0)
+    CHECK_MAX_VALUE(max_block_size)
+    CHECK_MAX_VALUE(max_insert_block_size)
+    CHECK_MAX_VALUE(min_insert_block_size_rows)
+    CHECK_MAX_VALUE(min_insert_block_size_bytes_for_materialized_views)
+    CHECK_MAX_VALUE(min_external_table_block_size_rows)
+    CHECK_MAX_VALUE(max_joined_block_size_rows)
+    CHECK_MAX_VALUE(input_format_parquet_max_block_size)
+
+#undef CHECK_MAX_VALUE
+
+
+    if (auto max_block_size = current_settings[Setting::max_block_size]; max_block_size == 0)
     {
         if (log)
             LOG_WARNING(log, "Sanity check: 'max_block_size' cannot be 0. Set to default value {}", DEFAULT_BLOCK_SIZE);
-        current_settings.set("max_block_size", DEFAULT_BLOCK_SIZE);
+        current_settings[Setting::max_block_size] = DEFAULT_BLOCK_SIZE;
     }
 }
 
diff --git a/src/Interpreters/executeQuery.cpp b/src/Interpreters/executeQuery.cpp
@@ -632,7 +632,7 @@ void logQueryFinishImpl(
 
     }
 
-    if (query_span)
+    if (query_span && query_span->isTraceEnabled())
     {
         query_span->addAttribute("db.statement", elem.query);
         query_span->addAttribute("clickhouse.query_id", elem.client_info.current_query_id);
diff --git a/src/Planner/Planner.cpp b/src/Planner/Planner.cpp
@@ -51,6 +51,8 @@
 #include <Storages/StorageDummy.h>
 #include <Storages/StorageMerge.h>
 
+#include <AggregateFunctions/IAggregateFunction.h>
+
 #include <Analyzer/Utils.h>
 #include <Analyzer/ColumnNode.h>
 #include <Analyzer/ConstantNode.h>
@@ -564,13 +566,21 @@ void addMergingAggregatedStep(QueryPlan & query_plan,
       * but it can work more slowly.
       */
 
-    auto keys = aggregation_analysis_result.aggregation_keys;
+    const auto & keys = aggregation_analysis_result.aggregation_keys;
+
+    /// For count() without parameters try to use just one thread
+    /// Typically this will either be a trivial count or a really small number of states
+    size_t max_threads = settings[Setting::max_threads];
+    if (keys.empty() && aggregation_analysis_result.aggregate_descriptions.size() == 1
+        && aggregation_analysis_result.aggregate_descriptions[0].function->getName() == String{"count"}
+        && aggregation_analysis_result.grouping_sets_parameters_list.empty())
+        max_threads = 1;
 
     Aggregator::Params params(
         keys,
         aggregation_analysis_result.aggregate_descriptions,
         query_analysis_result.aggregate_overflow_row,
-        settings[Setting::max_threads],
+        max_threads,
         settings[Setting::max_block_size],
         settings[Setting::min_hit_rate_to_use_consecutive_keys_optimization]);
 
@@ -1317,7 +1327,7 @@ void Planner::buildQueryPlanIfNeeded()
         return;
 
     LOG_TRACE(
-        getLogger("Planner"),
+        log,
         "Query to stage {}{}",
         QueryProcessingStage::toString(select_query_options.to_stage),
         select_query_options.only_analyze ? " only analyze" : "");
@@ -1483,7 +1493,7 @@ void Planner::buildPlanForQueryNode()
 
             auto & mutable_context = planner_context->getMutableQueryContext();
             mutable_context->setSetting("allow_experimental_parallel_reading_from_replicas", Field(0));
-            LOG_DEBUG(getLogger("Planner"), "Disabling parallel replicas to execute a query with IN with subquery");
+            LOG_DEBUG(log, "Disabling parallel replicas to execute a query with IN with subquery");
         }
     }
 
@@ -1520,9 +1530,7 @@ void Planner::buildPlanForQueryNode()
                 if (settings[Setting::allow_experimental_parallel_reading_from_replicas] >= 2)
                     throw Exception(ErrorCodes::SUPPORT_IS_DISABLED, "FINAL modifier is not supported with parallel replicas");
 
-                LOG_DEBUG(
-                    getLogger("Planner"),
-                    "FINAL modifier is not supported with parallel replicas. Query will be executed without using them.");
+                LOG_DEBUG(log, "FINAL modifier is not supported with parallel replicas. Query will be executed without using them.");
                 auto & mutable_context = planner_context->getMutableQueryContext();
                 mutable_context->setSetting("allow_experimental_parallel_reading_from_replicas", Field(0));
             }
@@ -1537,7 +1545,7 @@ void Planner::buildPlanForQueryNode()
             if (settings[Setting::allow_experimental_parallel_reading_from_replicas] >= 2)
                 throw Exception(ErrorCodes::SUPPORT_IS_DISABLED, "JOINs are not supported with parallel replicas");
 
-            LOG_DEBUG(getLogger("Planner"), "JOINs are not supported with parallel replicas. Query will be executed without using them.");
+            LOG_DEBUG(log, "JOINs are not supported with parallel replicas. Query will be executed without using them.");
 
             auto & mutable_context = planner_context->getMutableQueryContext();
             mutable_context->setSetting("allow_experimental_parallel_reading_from_replicas", Field(0));
@@ -1568,7 +1576,7 @@ void Planner::buildPlanForQueryNode()
     query_node_to_plan_step_mapping.insert(mapping.begin(), mapping.end());
 
     LOG_TRACE(
-        getLogger("Planner"),
+        log,
         "Query from stage {} to stage {}{}",
         QueryProcessingStage::toString(from_stage),
         QueryProcessingStage::toString(select_query_options.to_stage),
diff --git a/src/Planner/Planner.h b/src/Planner/Planner.h
@@ -75,6 +75,7 @@ class Planner
 
     void buildPlanForQueryNode();
 
+    LoggerPtr log = getLogger("Planner");
     QueryTreeNodePtr query_tree;
     SelectQueryOptions & select_query_options;
     PlannerContextPtr planner_context;
diff --git a/tests/queries/0_stateless/00172_early_constant_folding.reference b/tests/queries/0_stateless/00172_early_constant_folding.reference
@@ -8,11 +8,10 @@ ExpressionTransform × 10
         (ReadFromPreparedSource)
         SourceFromSingleChunk 0 → 1
 (Expression)
-ExpressionTransform × 10
+ExpressionTransform
   (MergingAggregated)
-  Resize 1 → 10
-    MergingAggregatedTransform
-      (Expression)
-      ExpressionTransform
-        (ReadFromPreparedSource)
-        SourceFromSingleChunk 0 → 1
+  MergingAggregatedTransform
+    (Expression)
+    ExpressionTransform
+      (ReadFromPreparedSource)
+      SourceFromSingleChunk 0 → 1
diff --git a/tests/queries/0_stateless/03448_trivial_count_single_threaded_merge.reference b/tests/queries/0_stateless/03448_trivial_count_single_threaded_merge.reference
@@ -0,0 +1,56 @@
+-- { echo On }
+-- We should use just a single thread to merge the state of trivial count
+EXPLAIN PIPELINE SELECT count() FROM trivial_count;
+(Expression)
+ExpressionTransform
+  (MergingAggregated)
+  MergingAggregatedTransform
+    (Expression)
+    ExpressionTransform
+      (ReadFromPreparedSource)
+      SourceFromSingleChunk 0 → 1
+-- But not if we are filtering or doing other operations (no trivial count)
+EXPLAIN PIPELINE SELECT count() FROM trivial_count WHERE number % 3 = 2;
+(Expression)
+ExpressionTransform × 4
+  (Aggregating)
+  Resize 1 → 4
+    AggregatingTransform
+      (Expression)
+      ExpressionTransform
+        (Filter)
+        FilterTransform
+          (ReadFromMergeTree)
+          MergeTreeSelect(pool: ReadPoolInOrder, algorithm: InOrder) 0 → 1
+EXPLAIN PIPELINE SELECT count() FROM trivial_count GROUP BY number % 10;
+(Expression)
+ExpressionTransform × 4
+  (Aggregating)
+  Resize 1 → 4
+    AggregatingTransform
+      (Expression)
+      ExpressionTransform
+        (ReadFromMergeTree)
+        MergeTreeSelect(pool: ReadPoolInOrder, algorithm: InOrder) 0 → 1
+-- Other aggregations should still use as many threads as necessary
+EXPLAIN PIPELINE SELECT sum(number) FROM trivial_count;
+(Expression)
+ExpressionTransform × 4
+  (Aggregating)
+  Resize 1 → 4
+    AggregatingTransform
+      (Expression)
+      ExpressionTransform
+        (ReadFromMergeTree)
+        MergeTreeSelect(pool: ReadPoolInOrder, algorithm: InOrder) 0 → 1
+EXPLAIN PIPELINE SELECT count(), sum(number) FROM trivial_count;
+(Expression)
+ExpressionTransform × 4
+  (Aggregating)
+  Resize 1 → 4
+    AggregatingTransform
+      (Expression)
+      ExpressionTransform
+        (ReadFromMergeTree)
+        MergeTreeSelect(pool: ReadPoolInOrder, algorithm: InOrder) 0 → 1
+DROP TABLE IF EXISTS trivial_count;
diff --git a/tests/queries/0_stateless/03448_trivial_count_single_threaded_merge.sql b/tests/queries/0_stateless/03448_trivial_count_single_threaded_merge.sql
@@ -0,0 +1,22 @@
+-- Tags: no-object-storage
+-- no-object-storage since the output of the pipeline depends on the read method
+
+SET enable_analyzer = 1;
+SET max_threads=4;
+
+DROP TABLE IF EXISTS trivial_count;
+CREATE TABLE trivial_count ENGINE = MergeTree() ORDER BY number AS Select * from numbers(10) ;
+
+-- { echo On }
+-- We should use just a single thread to merge the state of trivial count
+EXPLAIN PIPELINE SELECT count() FROM trivial_count;
+
+-- But not if we are filtering or doing other operations (no trivial count)
+EXPLAIN PIPELINE SELECT count() FROM trivial_count WHERE number % 3 = 2;
+EXPLAIN PIPELINE SELECT count() FROM trivial_count GROUP BY number % 10;
+
+-- Other aggregations should still use as many threads as necessary
+EXPLAIN PIPELINE SELECT sum(number) FROM trivial_count;
+EXPLAIN PIPELINE SELECT count(), sum(number) FROM trivial_count;
+
+DROP TABLE IF EXISTS trivial_count;

Original file line number	Diff line number	Diff line change
`@@ -153,7 +153,7 @@ struct TracingContextHolder`
`153`	`153`	`using TracingContextHolderPtr = std::unique_ptr<TracingContextHolder>;`
`154`	`154`
`155`	`155`	`/// A span holder that creates span automatically in a (function) scope if tracing is enabled.`
`156`		`-/// Once it's created or destructed, it automatically maitains the tracing context on the thread that it lives.`
	`156`	`+/// Once it's created or destructed, it automatically maintains the tracing context on the thread that it lives.`
`157`	`157`	`struct SpanHolder : public Span`
`158`	`158`	`{`
`159`	`159`	`explicit SpanHolder(std::string_view, SpanKind _kind = SpanKind::INTERNAL);`
Original file line number	Diff line number	Diff line change
`@@ -632,7 +632,7 @@ void logQueryFinishImpl(`
`632`	`632`
`633`	`633`	`}`
`634`	`634`
`635`		`- if (query_span)`
	`635`	`+ if (query_span && query_span->isTraceEnabled())`
`636`	`636`	`{`
`637`	`637`	`query_span->addAttribute("db.statement", elem.query);`
`638`	`638`	`query_span->addAttribute("clickhouse.query_id", elem.client_info.current_query_id);`