test
```sql
DROP TABLE IF EXISTS t;
DROP TABLE IF EXISTS t_mv;
create table t (a UInt64) Engine = Null;
create materialized view t_mv Engine = Null AS select now() as ts, max(a) from t group by ts;
insert into t select * from numbers_mt(3000000000) settings max_threads = 16, max_insert_threads=16;
```

results:
```
21.8:  
0 rows in set. Elapsed: 3.841 sec. Processed 3.00 billion rows, 24.00 GB (781.01 million rows/s., 6.25 GB/s.)

21.10: 
0 rows in set. Elapsed: 3.844 sec. Processed 6.00 billion rows, 48.00 GB (1.56 billion rows/s., 12.49 GB/s.)

21.11:
0 rows in set. Elapsed: 17.717 sec. Processed 5.98 billion rows, 47.86 GB (337.70 million rows/s., 2.70 GB/s.)

21.12
0 rows in set. Elapsed: 18.569 sec. Processed 6.00 billion rows, 48.00 GB (323.12 million rows/s., 2.58 GB/s.)

22.1
0 rows in set. Elapsed: 19.124 sec. Processed 6.00 billion rows, 48.00 GB (313.74 million rows/s., 2.51 GB/s.)

22.2
0 rows in set. Elapsed: 19.939 sec. Processed 6.00 billion rows, 48.00 GB (300.92 million rows/s., 2.41 GB/s.)

22.3
0 rows in set. Elapsed: 18.727 sec. Processed 6.00 billion rows, 48.00 GB (320.39 million rows/s., 2.56 GB/s.)

22.4
0 rows in set. Elapsed: 19.698 sec. Processed 6.00 billion rows, 48.00 GB (304.60 million rows/s., 2.44 GB/s.)

22.5
0 rows in set. Elapsed: 22.723 sec. Processed 6.00 billion rows, 48.00 GB (264.05 million rows/s., 2.11 GB/s.)
```

SIde note: pay attention to processed rows and progress bar behavior (minor cosmetic issue).
if (query.select && table->isRemote() && settings.parallel_distributed_insert_select)
        
        distributed_pipeline = table->distributedWrite(query, getContext());
    std::vector<Chain> out_chains;
    if (!distributed_pipeline || query.watch)
    {
        size_t out_streams_size = 1;
        if (query.select)
        {
            bool is_trivial_insert_select = false;
            if (settings.optimize_trivial_insert_select)
            {
                const auto & select_query = query.select->as<ASTSelectWithUnionQuery &>();
                const auto & selects = select_query.list_of_selects->children;
                const auto & union_modes = select_query.list_of_modes;
                
                const auto mode_is_all = [](const auto & mode) { return mode == SelectUnionMode::ALL; };
                is_trivial_insert_select =
                    std::all_of(union_modes.begin(), union_modes.end(), std::move(mode_is_all))
                    && std::all_of(selects.begin(), selects.end(), isTrivialSelect);
            }
            if (is_trivial_insert_select)
            {
                /** When doing trivial INSERT INTO ... SELECT ... FROM table,
                  * don't need to process SELECT with more than max_insert_threads
                  * and it's reasonable to set block size for SELECT to the desired block size for INSERT
                  * to avoid unnecessary squashing.
                  */
                Settings new_settings = getContext()->getSettings();
                new_settings.max_threads = std::max<UInt64>(1, settings.max_insert_threads);