Merge pull request ClickHouse#88451 from ClickHouse/backport/25.8/88339

Backport ClickHouse#88339 to 25.8: fix threads count for inserts

Author: CheSema

src/Interpreters/InterpreterInsertQuery.cpp

@@ -726,7 +726,7 @@ QueryPipeline InterpreterInsertQuery::buildInsertPipeline(ASTInsertQuery & query
 
     QueryPipeline pipeline = QueryPipeline(std::move(chain));
 
-    pipeline.setNumThreads(max_insert_threads);
+    pipeline.setNumThreads(max_threads);
     pipeline.setConcurrencyControl(settings[Setting::use_concurrency_control]);
 
     if (query.hasInlinedData() && !async_insert)

tests/queries/0_stateless/03652_threads_count_insert.reference

@@ -0,0 +1,8 @@
+max_threads: 1 max_insert_threads: 1
+1
+max_threads: 1 max_insert_threads: 5
+1
+max_threads: 10 max_insert_threads: 1
+10
+max_threads: 10 max_insert_threads: 5
+10

tests/queries/0_stateless/03652_threads_count_insert.sh

@@ -0,0 +1,56 @@
+#!/usr/bin/env bash
+# Tags: no-object-storage, no-parallel, no-fasttest
+
+# no-object-storage: s3 has 20 more threads
+# no-parallel: it checks the number of threads, which can be lowered in presence of other queries
+
+CUR_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+# shellcheck source=../shell_config.sh
+. "$CUR_DIR"/../shell_config.sh
+
+
+cat <<EOF | $CLICKHOUSE_CLIENT -n $SETTINGS
+drop table if exists testX;
+drop table if exists testXA;
+drop table if exists testXB;
+drop table if exists testXC;
+
+create table testX (A Int64) engine=MergeTree order by tuple();
+
+create materialized view testXA engine=MergeTree order by tuple() as select sleep(0.1) from testX;
+create materialized view testXB engine=MergeTree order by tuple() as select sleep(0.2), throwIf(A=1) from testX;
+create materialized view testXC engine=MergeTree order by tuple() as select sleep(0.1) from testX;
+EOF
+
+for max_threads in 1 10; do
+    for max_insert_threads in 1 5; do
+        echo "max_threads: $max_threads max_insert_threads: $max_insert_threads"
+
+        QUERY_ID="03652_query_id_$RANDOM"
+        SETTINGS="--query_id=$QUERY_ID "
+        SETTINGS="$SETTINGS --max_threads=$max_threads "
+        SETTINGS="$SETTINGS --max_insert_threads=$max_insert_threads "
+
+        SETTINGS="$SETTINGS --max_block_size=10 "
+        SETTINGS="$SETTINGS --min_insert_block_size_rows=10 "
+        SETTINGS="$SETTINGS --materialized_views_ignore_errors=1 "
+        SETTINGS="$SETTINGS --parallel_view_processing=1 "
+        SETTINGS="$SETTINGS --log_queries=1 "
+        SETTINGS="$SETTINGS --send_logs_level=error "
+
+        $CLICKHOUSE_CLIENT -q 'select * from numbers(50) format TSV' | $CLICKHOUSE_CLIENT $SETTINGS -q 'insert into testX FORMAT TSV'
+
+        $CLICKHOUSE_CLIENT -q 'system flush logs system.query_log;'
+
+        cat <<EOF | $CLICKHOUSE_CLIENT
+select
+    if(peak_threads_usage >= 10, 10, peak_threads_usage),
+from system.query_log where
+    current_database = currentDatabase() and
+    type != 'QueryStart' and
+    query_id = '$QUERY_ID'
+order by ALL;
+EOF
+
+    done
+done