Parallel read performance degradation

[BohuTANG]

Summary

This is found from databend ontime dataset tests on AWS EC2 and S3: How to

parallel_read_threads is 1, if we setting it to 4:

set parallel_read_threads=4;

All the query cost almost same as the no setting.

How to check:
if we checkout to f9971bdf335333ffc2253b60b0842b2a3c8ca6cc commit:

commit f9971bdf335333ffc2253b60b0842b2a3c8ca6cc (HEAD -> main, origin/main, origin/HEAD)
Merge: 3c54a8416 7373fc595
Author: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>
Date:   Thu Feb 24 02:07:08 2022 +0000

    Merge pull request #4230 from youngsofun/parquet2

    remove buffer in block_reader of fuse store

Then set parallel_read_threads=4 will have a lot performance improve.

[BohuTANG]

Re-imported a more organized data(import CSV one by one, it organized by Year).

DatabendQuery v-0.1.0-eb581cd-simd(1.60.0-nightly-2022-02-27T09:20:55.414288628+00:00)

fuse_history snapshot

mysql> select * from fuse_history('default', 'ontime');
+----------------------------------+----------------------------------+---------------+-------------+-----------+--------------------+------------------+
| snapshot_id                      | prev_snapshot_id                 | segment_count | block_count | row_count | bytes_uncompressed | bytes_compressed |
+----------------------------------+----------------------------------+---------------+-------------+-----------+--------------------+------------------+
| 970d05036f4c4c428c1eed838b4a4a4b | 23b01d9faa1741aa906855586a019b44 |             1 |         203 | 202687654 |       146705865856 |      11655306085 |
| 23b01d9faa1741aa906855586a019b44 | e7662ebf483b4da9be04deb491c21868 |             0 |           0 |         0 |                  0 |                0 |
| e7662ebf483b4da9be04deb491c21868 | NULL                             |             1 |           1 |       100 |              72153 |            38216 |
+----------------------------------+----------------------------------+---------------+-------------+-----------+--------------------+------------------+

Explain a query(partitions_scanned: 61, partitions_total: 203)

mysql> explain SELECT DayOfWeek, count(*) AS c FROM ontime WHERE Year >= 2000 AND Year <= 2008 GROUP BY DayOfWeek ORDER BY c DESC;
+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
------------------------+
| explain                                                                                                                                                                                                                                         
                        |
+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
------------------------+
| Projection: DayOfWeek:Nullable(UInt8), count(0) as c:UInt64                                                                                                                                                                                     
                        |
|   Sort: count(0):UInt64                                                                                                                                                                                                                         
                        |
|     AggregatorFinal: groupBy=[[DayOfWeek]], aggr=[[count(0)]]                                                                                                                                                                                   
                        |
|       AggregatorPartial: groupBy=[[DayOfWeek]], aggr=[[count(0)]]                                                                                                                                                                               
                        |
|         Expression: DayOfWeek:Nullable(UInt8), 0:Int8 (Before GroupBy)                                                                                                                                                                          
                        |
|           Filter: ((Year >= 2000) and (Year <= 2008))                                                                                                                                                                                           
                        |
|             ReadDataSource: scan schema: [Year:UInt16;N, DayOfWeek:UInt8;N], statistics: [read_rows: 61000000, read_bytes: 183000000, partitions_scanned: 61, partitions_total: 203], push_downs: [projections: [0, 4], filters: [((Year >= 2000
) AND (Year <= 2008))]] |
+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
------------------------+

T0: set parallel_read_threads=1 query cost 1.311 sec

mysql> SELECT DayOfWeek, count(*) AS c FROM ontime WHERE Year >= 2000 AND Year <= 2008 GROUP BY DayOfWeek ORDER BY c DESC;
+-----------+---------+
| DayOfWeek | c       |
+-----------+---------+
|         5 | 8732422 |
|         1 | 8730614 |
|         4 | 8710843 |
|         3 | 8685626 |
|         2 | 8639632 |
|         7 | 8274367 |
|         6 | 7514194 |
+-----------+---------+
7 rows in set (1.32 sec)
Read 61000000 rows, 183 MB in 1.311 sec., 46.52 million rows/sec., 139.55 MB/sec.

T1: set parallel_read_threads=4 query cost 5.12 sec

mysql> set parallel_read_threads=4;
Query OK, 0 rows affected (0.00 sec)
Read 0 rows, 0 B in 0.000 sec., 0 rows/sec., 0 B/sec.

mysql> SELECT DayOfWeek, count(*) AS c FROM ontime WHERE Year >= 2000 AND Year <= 2008 GROUP BY DayOfWeek ORDER BY c DESC;
+-----------+---------+
| DayOfWeek | c       |
+-----------+---------+
|         5 | 8732422 |
|         1 | 8730614 |
|         4 | 8710843 |
|         3 | 8685626 |
|         2 | 8639632 |
|         7 | 8274367 |
|         6 | 7514194 |
+-----------+---------+
7 rows in set (5.12 sec)
Read 61000000 rows, 183 MB in 5.115 sec., 11.93 million rows/sec., 35.78 MB/sec.

T2:set parallel_read_threads=8 query cost 5.130 sec

mysql> set parallel_read_threads=8;
Query OK, 0 rows affected (0.00 sec)
Read 0 rows, 0 B in 0.000 sec., 0 rows/sec., 0 B/sec.

mysql> SELECT DayOfWeek, count(*) AS c FROM ontime WHERE Year >= 2000 AND Year <= 2008 GROUP BY DayOfWeek ORDER BY c DESC;
+-----------+---------+
| DayOfWeek | c       |
+-----------+---------+
|         5 | 8732422 |
|         1 | 8730614 |
|         4 | 8710843 |
|         3 | 8685626 |
|         2 | 8639632 |
|         7 | 8274367 |
|         6 | 7514194 |
+-----------+---------+
7 rows in set (5.14 sec)
Read 61000000 rows, 183 MB in 5.130 sec., 11.89 million rows/sec., 35.67 MB/sec.

T3:set parallel_read_threads=16 query cost 1.700 sec

mysql> set parallel_read_threads=16;
Query OK, 0 rows affected (0.01 sec)
Read 0 rows, 0 B in 0.000 sec., 0 rows/sec., 0 B/sec.

mysql> SELECT DayOfWeek, count(*) AS c FROM ontime WHERE Year >= 2000 AND Year <= 2008 GROUP BY DayOfWeek ORDER BY c DESC;
+-----------+---------+
| DayOfWeek | c       |
+-----------+---------+
|         5 | 8732422 |
|         1 | 8730614 |
|         4 | 8710843 |
|         3 | 8685626 |
|         2 | 8639632 |
|         7 | 8274367 |
|         6 | 7514194 |
+-----------+---------+
7 rows in set (1.70 sec)
Read 61000000 rows, 183 MB in 1.700 sec., 35.89 million rows/sec., 107.68 MB/sec.

[BohuTANG]

These may be useful to you, or may wish me to provide other information, @dantengsky @Xuanwo

[Xuanwo]

I'm working on addressing this issue now.

[BohuTANG]

After testing, the new version(v0.1.3) does not address this issue, we still need to locate it.

[Xuanwo]

We have a parallel_range_read benchmark based on the example that @sundy-li contributed now.

On my local benchmark with s3 over minio, it seems that opendal works well with parallel:

s3_parallel/parallel_range_read_1
                        time:   [2.6362 ms 2.8162 ms 3.0113 ms]
                        thrpt:  [2.5944 GiB/s 2.7741 GiB/s 2.9635 GiB/s]
s3_parallel/parallel_range_read_2
                        time:   [3.3120 ms 3.3710 ms 3.4387 ms]
                        thrpt:  [4.5439 GiB/s 4.6351 GiB/s 4.7177 GiB/s]
s3_parallel/parallel_range_read_4
                        time:   [4.5608 ms 4.6776 ms 4.7988 ms]
                        thrpt:  [6.5120 GiB/s 6.6808 GiB/s 6.8518 GiB/s]
s3_parallel/parallel_range_read_8
                        time:   [8.9568 ms 9.3281 ms 9.6966 ms]
                        thrpt:  [6.4455 GiB/s 6.7002 GiB/s 6.9779 GiB/s]
s3_parallel/parallel_range_read_16
                        time:   [19.478 ms 20.004 ms 20.549 ms]
                        thrpt:  [6.0831 GiB/s 6.2487 GiB/s 6.4173 GiB/s]

We set runtime threads to 4 during this test and start parallel tasks to read files.

Benchmark on tmpfs has the similar results:

fs_parallel/parallel_range_read_1
                        time:   [1.4370 ms 1.4634 ms 1.5090 ms]
                        thrpt:  [5.1771 GiB/s 5.3386 GiB/s 5.4365 GiB/s]
fs_parallel/parallel_range_read_2
                        time:   [1.3994 ms 1.4775 ms 1.5524 ms]
                        thrpt:  [10.065 GiB/s 10.575 GiB/s 11.166 GiB/s]
fs_parallel/parallel_range_read_4
                        time:   [3.4201 ms 3.4835 ms 3.5609 ms]
                        thrpt:  [8.7758 GiB/s 8.9708 GiB/s 9.1372 GiB/s]
fs_parallel/parallel_range_read_8
                        time:   [9.9594 ms 10.172 ms 10.378 ms]
                        thrpt:  [6.0225 GiB/s 6.1443 GiB/s 6.2755 GiB/s]
fs_parallel/parallel_range_read_16
                        time:   [22.447 ms 22.953 ms 23.471 ms]
                        thrpt:  [5.3257 GiB/s 5.4460 GiB/s 5.5688 GiB/s]

[BohuTANG]

Good, let's close.