Clickhouse: Distributed table is slower than a query containing union all for each remote servers.

[achimbab]

I compared the speed of two queries and I got a question.

Two queries do the same thing and also the results are the same.

But the elapsed time of the two queries are different.

Is there a reason that the elapsed times are different?

First Query: Using Distributed Table

select count() from data_dist SETTINGS load_balancing='in_order'

Result: 50546864822
Elapsed Time:
real    0m4.316s
user    0m0.017s
sys     0m0.019s

Second Query: Using union all for each remote server

select sum(cnt) from (
  select count() as cnt from remote('data1.test.com', default, data_local)
  union all
  select count() as cnt from remote('data2.test.com', default, data_local)
  union all
  select count() as cnt from remote('data3.test.com', default, data_local)
  union all
  select count() as cnt from remote('data4.test.com', default, data_local)
  union all
  select count() as cnt from remote('data5.test.com', default, data_local)
  union all
  select count() as cnt from remote('data6.test.com', default, data_local)
  union all
  select count() as cnt from remote('data7.test.com', default, data_local)
  union all
  select count() as cnt from remote('data8.test.com', default, data_local)
  union all
  select count() as cnt from remote('data9.test.com', default, data_local)
  union all
  select count() as cnt from remote('data10.test.com', default, data_local)
)

Result: 50546864822
Elapsed Time:
real    0m3.099s
user    0m0.019s
sys     0m0.017s

Thank you for the reading.

[achimbab]

My cluster consists of 10 shards.
I tried several times to eliminate the bottleneck that occurs at the disk-read.

The distributed query references the 10 shards.

data_dist --+--> data1.test.com/data_local
            |     
            +--> data2.test.com/data_local
            |     
            +--> data3.test.com/data_local
            |     
            +--> data4.test.com/data_local
            |     
            +--> data5.test.com/data_local
            |     
            +--> data6.test.com/data_local
            |     
            +--> data7.test.com/data_local
            |     
            +--> data8.test.com/data_local
            |     
            +--> data9.test.com/data_local
            |     
            +--> data10.test.com/data_local

[sfriquet]

Hi, I might have a similar question, so adding this here in case it's related.

I also see a case where a query is significantly slower on a Distributed table than on the union of all its shards.

e.g:

SELECT count() FROM test.foo GROUP BY toDate(timestamp)

┌───count()─┐
│ 260533159 │
└───────────┘

SELECT count(*) FROM test.foo_local 
UNION ALL 
SELECT count(*) FROM remote('...', 'test.foo_local') 

┌───count()─┐
│ 130266413 │
└───────────┘
┌───count()─┐
│ 130266746 │
└───────────┘

Here's what I see in terms of timings:

Non distributed table:

SELECT id, count(*) FROM test.foo GROUP BY id ORDER BY count(*) DESC;

[...]

9 rows in set. Elapsed: 0.060 sec. Processed 260.53 million rows, 1.04 GB (4.34 billion rows/s., 17.36 GB/s.) 

Distributed table:

SELECT id, count(*) FROM test.foo_all GROUP BY id ORDER BY count(*) DESC;

[...]

9 rows in set. Elapsed: 0.374 sec. Processed 260.53 million rows, 1.04 GB (697.13 million rows/s., 2.79 GB/s.) 

If I query the local tables one by one:

SELECT id, count(*) FROM test.foo_local GROUP BY id ORDER BY count(*) DESC 
UNION ALL
SELECT id, count(*) FROM remote('...', 'test.foo_local') GROUP BY id ORDER BY count(*) DESC


┌─────────id─┬──count()─┐
[...]
└────────────┴──────────┘
┌─────────id─┬──count()─┐
[...]
└────────────┴──────────┘

18 rows in set. Elapsed: 0.069 sec. Processed 260.53 million rows, 1.04 GB (3.78 billion rows/s., 15.13 GB/s.) 

Or with distributed_group_by_no_merge = 1

:) set distributed_group_by_no_merge = 1

SET distributed_group_by_no_merge = 1

Ok.

0 rows in set. Elapsed: 0.001 sec. 

:) SELECT id, count(*) FROM test.foo_all GROUP BY id ORDER BY count(*) DESC;


[ same as above]

18 rows in set. Elapsed: 0.069 sec. Processed 260.53 million rows, 1.04 GB (3.78 billion rows/s., 15.13 GB/s.) 

So at this point I can see:

• query on the non distributed dataset takes 0.06s
• queries on the shards, each take ~0.03s (which is expected, since a shard has half the rows)
• query on the distributed table takes 0.3s, i.e. 5x more, than the query on the non distributed set, and this is what I don't understand

Using --send_logs_level=trace, I can see that the query on the non distributed table uses 32 threads, which match the settings
e.g:

[sfriquet-1] 2020.04.10 11:51:16.326815 [ 2279 ] {c062dec2-1e0f-4647-91e6-7b114ecf224e} <Trace> AggregatingTransform: Aggregated. 8781824 to 8 rows (from 33.500 MiB) in 0.053 sec. (165746749.368 rows/sec., 632.274 MiB/sec.)
[sfriquet-1] 2020.04.10 11:51:16.326815 [ 2466 ] {c062dec2-1e0f-4647-91e6-7b114ecf224e} <Trace> AggregatingTransform: Aggregated. 8798208 to 8 rows (from 33.562 MiB) in 0.053 sec. (166146616.258 rows/sec., 633.799 MiB/sec.)
[sfriquet-1] 2020.04.10 11:51:16.326863 [ 2461 ] {c062dec2-1e0f-4647-91e6-7b114ecf224e} <Trace> AggregatingTransform: Aggregated. 5849088 to 8 rows (from 22.312 MiB) in 0.053 sec. (110377146.941 rows/sec., 421.055 MiB/sec.)
[sfriquet-1] 2020.04.10 11:51:16.326871 [ 2283 ] {c062dec2-1e0f-4647-91e6-7b114ecf224e} <Trace> AggregatingTransform: Aggregated. 8732672 to 8 rows (from 33.312 MiB) in 0.053 sec. (164692147.452 rows/sec., 628.251 MiB/sec.)
[sfriquet-1] 2020.04.10 11:51:16.326831 [ 2301 ] {c062dec2-1e0f-4647-91e6-7b114ecf224e} <Trace> AggregatingTransform: Aggregated. 9256960 to 8 rows (from 35.312 MiB) in 0.053 sec. (174654343.467 rows/sec., 666.253 MiB/sec.)
[sfriquet-1] 2020.04.10 11:51:16.326894 [ 2298 ] {c062dec2-1e0f-4647-91e6-7b114ecf224e} <Trace> AggregatingTransform: Aggregated. 6144000 to 8 rows (from 23.438 MiB) in 0.053 sec. (115801407.372 rows/sec., 441.747 MiB/sec.)
[sfriquet-1] 2020.04.10 11:51:16.326871 [ 2278 ] {c062dec2-1e0f-4647-91e6-7b114ecf224e} <Trace> AggregatingTransform: Aggregated. 5668864 to 8 rows (from 21.625 MiB) in 0.053 sec. (106949094.010 rows/sec., 407.978 MiB/sec.)
[sfriquet-1] 2020.04.10 11:51:16.326821 [ 2464 ] {c062dec2-1e0f-4647-91e6-7b114ecf224e} <Trace> AggregatingTransform: Aggregated. 8732672 to 8 rows (from 33.312 MiB) in 0.053 sec. (164764725.794 rows/sec., 628.528 MiB/sec.)
[sfriquet-1] 2020.04.10 11:51:16.326895 [ 2303 ] {c062dec2-1e0f-4647-91e6-7b114ecf224e} <Trace> AggregatingTransform: Aggregated. 9043968 to 8 rows (from 34.500 MiB) in 0.053 sec. (170553733.395 rows/sec., 650.611 MiB/sec.)
[sfriquet-1] 2020.04.10 11:51:16.326892 [ 2454 ] {c062dec2-1e0f-4647-91e6-7b114ecf224e} <Trace> AggregatingTransform: Aggregated. 8863744 to 8 rows (from 33.812 MiB) in 0.053 sec. (167009500.749 rows/sec., 637.091 MiB/sec.)
[sfriquet-1] 2020.04.10 11:51:16.326822 [ 2468 ] {c062dec2-1e0f-4647-91e6-7b114ecf224e} <Trace> AggregatingTransform: Aggregated. 8978432 to 8 rows (from 34.250 MiB) in 0.053 sec. (169448210.094 rows/sec., 646.394 MiB/sec.)
[sfriquet-1] 2020.04.10 11:51:16.326814 [ 2300 ] {c062dec2-1e0f-4647-91e6-7b114ecf224e} <Trace> AggregatingTransform: Aggregated. 9043968 to 8 rows (from 34.500 MiB) in 0.053 sec. (170749473.330 rows/sec., 651.358 MiB/sec.)
[sfriquet-1] 2020.04.10 11:51:16.326991 [ 2462 ] {c062dec2-1e0f-4647-91e6-7b114ecf224e} <Trace> AggregatingTransform: Aggregated. 5701632 to 8 rows (from 21.750 MiB) in 0.053 sec. (107282021.771 rows/sec., 409.248 MiB/sec.)
[sfriquet-1] 2020.04.10 11:51:16.326830 [ 2505 ] {c062dec2-1e0f-4647-91e6-7b114ecf224e} <Trace> AggregatingTransform: Aggregated. 9453568 to 8 rows (from 36.062 MiB) in 0.053 sec. (178581976.179 rows/sec., 681.236 MiB/sec.)
[sfriquet-1] 2020.04.10 11:51:16.326838 [ 2506 ] {c062dec2-1e0f-4647-91e6-7b114ecf224e} <Trace> AggregatingTransform: Aggregated. 9011200 to 9 rows (from 34.375 MiB) in 0.053 sec. (170038958.833 rows/sec., 648.647 MiB/sec.)
[sfriquet-1] 2020.04.10 11:51:16.326815 [ 2305 ] {c062dec2-1e0f-4647-91e6-7b114ecf224e} <Trace> AggregatingTransform: Aggregated. 8667136 to 8 rows (from 33.062 MiB) in 0.053 sec. (163742800.540 rows/sec., 624.629 MiB/sec.)
[sfriquet-1] 2020.04.10 11:51:16.326866 [ 2453 ] {c062dec2-1e0f-4647-91e6-7b114ecf224e} <Trace> AggregatingTransform: Aggregated. 6111232 to 8 rows (from 23.312 MiB) in 0.053 sec. (115291937.496 rows/sec., 439.804 MiB/sec.)
[sfriquet-1] 2020.04.10 11:51:16.326879 [ 2304 ] {c062dec2-1e0f-4647-91e6-7b114ecf224e} <Trace> AggregatingTransform: Aggregated. 8749056 to 8 rows (from 33.375 MiB) in 0.053 sec. (164933723.705 rows/sec., 629.172 MiB/sec.)
[sfriquet-1] 2020.04.10 11:51:16.326900 [ 2467 ] {c062dec2-1e0f-4647-91e6-7b114ecf224e} <Trace> AggregatingTransform: Aggregated. 8781824 to 8 rows (from 33.500 MiB) in 0.053 sec. (165645317.634 rows/sec., 631.887 MiB/sec.)
[sfriquet-1] 2020.04.10 11:51:16.326910 [ 2277 ] {c062dec2-1e0f-4647-91e6-7b114ecf224e} <Trace> AggregatingTransform: Aggregated. 8667136 to 8 rows (from 33.062 MiB) in 0.053 sec. (163370302.659 rows/sec., 623.208 MiB/sec.)
[sfriquet-1] 2020.04.10 11:51:16.326823 [ 2455 ] {c062dec2-1e0f-4647-91e6-7b114ecf224e} <Trace> AggregatingTransform: Aggregated. 8842151 to 8 rows (from 33.730 MiB) in 0.053 sec. (166807170.035 rows/sec., 636.319 MiB/sec.)
[sfriquet-1] 2020.04.10 11:51:16.326914 [ 2460 ] {c062dec2-1e0f-4647-91e6-7b114ecf224e} <Trace> AggregatingTransform: Aggregated. 5570560 to 8 rows (from 21.250 MiB) in 0.053 sec. (104943170.414 rows/sec., 400.326 MiB/sec.)
[sfriquet-1] 2020.04.10 11:51:16.326920 [ 2308 ] {c062dec2-1e0f-4647-91e6-7b114ecf224e} <Trace> AggregatingTransform: Aggregated. 8929280 to 8 rows (from 34.062 MiB) in 0.053 sec. (168294061.364 rows/sec., 641.991 MiB/sec.)
[sfriquet-1] 2020.04.10 11:51:16.326925 [ 2457 ] {c062dec2-1e0f-4647-91e6-7b114ecf224e} <Trace> AggregatingTransform: Aggregated. 8339456 to 8 rows (from 31.812 MiB) in 0.053 sec. (157239007.607 rows/sec., 599.819 MiB/sec.)
[sfriquet-1] 2020.04.10 11:51:16.326936 [ 2306 ] {c062dec2-1e0f-4647-91e6-7b114ecf224e} <Trace> AggregatingTransform: Aggregated. 9420800 to 8 rows (from 35.938 MiB) in 0.053 sec. (177579595.839 rows/sec., 677.412 MiB/sec.)
[sfriquet-1] 2020.04.10 11:51:16.326824 [ 2459 ] {c062dec2-1e0f-4647-91e6-7b114ecf224e} <Trace> AggregatingTransform: Aggregated. 8978432 to 8 rows (from 34.250 MiB) in 0.053 sec. (169374295.844 rows/sec., 646.112 MiB/sec.)
[sfriquet-1] 2020.04.10 11:51:16.327072 [ 2465 ] {c062dec2-1e0f-4647-91e6-7b114ecf224e} <Trace> AggregatingTransform: Aggregated. 9502720 to 8 rows (from 36.250 MiB) in 0.053 sec. (178458951.128 rows/sec., 680.767 MiB/sec.)
[sfriquet-1] 2020.04.10 11:51:16.326818 [ 2302 ] {c062dec2-1e0f-4647-91e6-7b114ecf224e} <Trace> AggregatingTransform: Aggregated. 8798208 to 8 rows (from 33.562 MiB) in 0.053 sec. (166181914.766 rows/sec., 633.934 MiB/sec.)
[sfriquet-1] 2020.04.10 11:51:16.326818 [ 2456 ] {c062dec2-1e0f-4647-91e6-7b114ecf224e} <Trace> AggregatingTransform: Aggregated. 9043968 to 8 rows (from 34.500 MiB) in 0.053 sec. (170619491.305 rows/sec., 650.862 MiB/sec.)
[sfriquet-1] 2020.04.10 11:51:16.326871 [ 2469 ] {c062dec2-1e0f-4647-91e6-7b114ecf224e} <Trace> AggregatingTransform: Aggregated. 5586944 to 8 rows (from 21.312 MiB) in 0.053 sec. (105409099.419 rows/sec., 402.104 MiB/sec.)
[sfriquet-1] 2020.04.10 11:51:16.327113 [ 2307 ] {c062dec2-1e0f-4647-91e6-7b114ecf224e} <Trace> AggregatingTransform: Aggregated. 8781824 to 8 rows (from 33.500 MiB) in 0.053 sec. (164836720.907 rows/sec., 628.802 MiB/sec.)
[sfriquet-1] 2020.04.10 11:51:16.327128 [ 2309 ] {c062dec2-1e0f-4647-91e6-7b114ecf224e} <Trace> AggregatingTransform: Aggregated. 5701632 to 8 rows (from 21.750 MiB) in 0.053 sec. (106914024.157 rows/sec., 407.845 MiB/sec.)
[sfriquet-1] 2020.04.10 11:51:16.328655 [ 2309 ] {c062dec2-1e0f-4647-91e6-7b114ecf224e} <Trace> Aggregator: Merging aggregated data
[sfriquet-1] 2020.04.10 11:51:16.330106 [ 2337 ] {c062dec2-1e0f-4647-91e6-7b114ecf224e} <Information> executeQuery: Read 260533159 rows, 993.86 MiB in 0.057 sec., 4568029420 rows/sec., 17.02 GiB/sec.

However when targeting the distributed table, I can see this:

[sfriquet-1] 2020.04.10 11:52:22.738885 [ 2455 ] {68044646-b3bc-4754-a163-34abb49b37d8} <Trace> AggregatingTransform: Aggregated. 65344920 to 8 rows (from 249.271 MiB) in 0.332 sec. (197027458.220 rows/sec., 751.600 MiB/sec.)
[sfriquet-1] 2020.04.10 11:52:22.738939 [ 2455 ] {68044646-b3bc-4754-a163-34abb49b37d8} <Trace> AggregatingTransform: Aggregated. 0 to 0 rows (from 0.000 MiB) in 0.332 sec. (0.000 rows/sec., 0.000 MiB/sec.)
[sfriquet-1] 2020.04.10 11:52:22.738959 [ 2455 ] {68044646-b3bc-4754-a163-34abb49b37d8} <Trace> AggregatingTransform: Aggregated. 0 to 0 rows (from 0.000 MiB) in 0.332 sec. (0.000 rows/sec., 0.000 MiB/sec.)
[sfriquet-1] 2020.04.10 11:52:22.739011 [ 2455 ] {68044646-b3bc-4754-a163-34abb49b37d8} <Trace> AggregatingTransform: Aggregated. 0 to 0 rows (from 0.000 MiB) in 0.332 sec. (0.000 rows/sec., 0.000 MiB/sec.)
[sfriquet-1] 2020.04.10 11:52:22.738960 [ 2456 ] {68044646-b3bc-4754-a163-34abb49b37d8} <Trace> AggregatingTransform: Aggregated. 64921493 to 9 rows (from 247.656 MiB) in 0.332 sec. (195705962.063 rows/sec., 746.559 MiB/sec.)
[sfriquet-1] 2020.04.10 11:52:22.739032 [ 2455 ] {68044646-b3bc-4754-a163-34abb49b37d8} <Trace> AggregatingTransform: Aggregated. 0 to 0 rows (from 0.000 MiB) in 0.332 sec. (0.000 rows/sec., 0.000 MiB/sec.)
[...]

Does it mean the query on the first shard / host used only 2 threads instead of 32 ?
If so what's the reason and is there a way to improve this ?

Thanks

PS: This is on ClickHouse version 20.3.5 revision 54433.

[alexey-milovidov]

@sfriquet Distributed table has to merge the intermediate results from remote servers.
It is needed because data for the same GROUP BY key (in your case - id) may reside on multiple shards.

[alexey-milovidov]

@achimbab But your case is different.

[sfriquet]

@sfriquet Distributed table has to merge the intermediate results from remote servers.
It is needed because data for the same GROUP BY key (in your case - id) may reside on multiple shards.

@alexey-milovidov Thanks for your reply.

In cases where there are few distinct keys to aggregate (here only 9), what intermediate results are we speaking of ?

In the case I presented, it appears it would be an order of magnitude faster to collect the results from each shards in parallel and then merge the result at the end. Is this a proper workaround in such case?

I'm also curious why there seem to be only 2 threads used on the first server (where the query is issued) vs 32 on other servers or 32 when not querying a distributed table.

I can perhaps open a new task if this looks different than the initial issue.

Thanks.

[sfriquet]

@alexey-milovidov I think it might still be related to this issue:

Here's another example without group by:

:) select sum(id), count(*) from test.foo;


┌────────────sum(id)─┬───count()─┐
│ 385975829488090336 │ 260533159 │
└────────────────────┴───────────┘

1 rows in set. Elapsed: 0.023 sec. Processed 260.53 million rows, 1.04 GB (11.42 billion rows/s., 45.67 GB/s.) 

:) SELECT sum(id), count(*) FROM test.foo_local UNION ALL SELECT sum(id), count(*) FROM remote('...', 'test.foo_local')

┌────────────sum(id)─┬───count()─┐
│ 192998165815731942 │ 130266413 │
└────────────────────┴───────────┘
┌────────────sum(id)─┬───count()─┐
│ 192977663672358394 │ 130266746 │
└────────────────────┴───────────┘

2 rows in set. Elapsed: 0.032 sec. Processed 260.53 million rows, 1.04 GB (8.21 billion rows/s., 32.82 GB/s.) 

 :) select sum(id), count(*) from test.foo_all;


┌────────────sum(id)─┬───count()─┐
│ 385975829488090336 │ 260533159 │
└────────────────────┴───────────┘

1 rows in set. Elapsed: 0.089 sec. Processed 260.53 million rows, 1.04 GB (2.93 billion rows/s., 11.71 GB/s.) 

the distributed sum is 4x slower than the non distributed one and 3x slower than the union of all shards.

[azat]

Here's another example without group by:

without GROUP BY but with aggregation (since sum()/count() requires final aggregate)

But does not looks like this is how it should work anyway.
Can you provide select name, value from system.settings where changed ? (and full output with send_logs_level='trace' may be useful too)

PS: This is on ClickHouse version 20.3.5 revision 54433.

Are you sure? I just remember one bug that you may hit that has been fixed in #9673 (but it has been backported to the 20.3.5.21-stable - cb49e8bdb91f86a842a934adec5e2183942f0c45)

In the case I presented, it appears it would be an order of magnitude faster to collect the results from each shards in parallel and then merge the result at the end

Indeed, this is how it works

Does it mean the query on the first shard / host used only 2 threads instead of 32 ?

It don't have two use more then 2 threads, since it has only two underlying streams

[achimbab]

@alexey-milovidov @azat @sfriquet
Thank you.

[alexey-milovidov]

I'm not sure that this issue is solved. I did not have time to investigate it.

[azat]

@achimbab does the initial issue solved for you? how, after upgrade?

If not please provide:

• log of the query, (set send_logs_level='trace')
• version that you had before and now (if you upgraded it)
• select name, value from system.settings where changed
• full output of the following query select sleep(1) s from remote('127.{1,1}', system.one) group by s