Also i think it will be better to expand max_threads here to max_distributed_connections similar to

and

The simplest possible workaround for that - to enable prefer_localhost_replica

@filimonov , you mean "to disable prefer_localhost_replica", right?

The test (manual, since that can't be stable in CI/CD because it would measure the time, and i can't just check the EXPLAIN, because of #72188 )

drop table if exists src;
drop table if exists dst;
drop table if exists dst_local;
create table dst_local engine=MergeTree ORDER BY tuple() AS system.numbers;
create table dst engine=Distributed('test_cluster_two_shards', currentDatabase(), dst_local, rand());
create table src engine=Distributed('test_cluster_two_shards', system, numbers_mt, rand());

set parallel_distributed_insert_select=2, prefer_localhost_replica=0

/* select from number_mt used as a trivial illustration of the query which can be much faster when multithreaded  */
insert into dst select number from src where not(sleepEachRow(0.001)) limit 10000 SETTINGS max_block_size = 1, max_threads=50;

Time before the change (local pipeline executed single threaded, linearly with remote pipeline).

0 rows in set. Elapsed: 11.795 sec. Processed 20.09 thousand rows, 160.68 KB (1.70 thousand rows/s., 13.62 KB/s.)
Peak memory usage: 3.25 MiB.

Time after the change (both local & remote are executed as subqueries which can use enough threads).

0 rows in set. Elapsed: 0.234 sec. Processed 20.10 thousand rows, 160.78 KB (85.81 thousand rows/s., 686.49 KB/s.)
Peak memory usage: 2.09 MiB.

This is an automated comment for commit f245a49 with description of existing statuses. It's updated for the latest CI running

❌ Click here to open a full report in a separate page

yeah, i guess the problem is exactly that we execute local query with a single thread. for unclear reason. pipeline looks fine. i mean we could make workaround working, but it won't solve anything in practice, because prefer_locahost_replica enabled by default.
would you like to dig a little deeper and fix the problem? otherwise pls create an issue.

but it won't solve anything in practice, because prefer_locahost_replica enabled by default

Actually i don't like that feature (for many reasons), and prefer to keep it disabled.

Some problems of that are listed here #6349

But in short - it CHANGES the way HOW the query is executed, instead of changing WHERE the query will be executed (as you can think from it's name), it make a lot of things more complex and behavior non-transparent.

I would even opt for changing the default if you ask me, or change it to behave like the name says. Just prefer the localhost insead the other replica from same shard (but still run one more subquery there).

would you like to dig a little deeper and fix the problem?

I did some simple experiments, failed to find why it's single threaded (since it's insert select probably max_insert_threads should help, but it doesn't, or maybe something else).

But anyway i have some doubts that even if we will allow more thread to execute the load pipeline it will be a robust fix. (because if the pipeline mixes ReadFromRemote & some compute-heavy step - there is always a risk that the ReadFromRemote will not get its' thread and will not start the remote jobs while compute-heavy steps will not finish their work).

otherwise pls create an issue.

OK. #72792