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).
else if (query.select || query.watch)
    {
        const auto & header = out_chains.at(0).getInputHeader();
        auto actions_dag = ActionsDAG::makeConvertingActions(
                pipeline.getHeader().getColumnsWithTypeAndName(),
                header.getColumnsWithTypeAndName(),
                ActionsDAG::MatchColumnsMode::Position);
        auto actions = std::make_shared<ExpressionActions>(actions_dag, ExpressionActionsSettings::fromContext(getContext(), CompileExpressions::yes));
        pipeline.addSimpleTransform([&](const Block & in_header) -> ProcessorPtr
        {
            return std::make_shared<ExpressionTransform>(in_header, actions);
        });
        
        
        pipeline.addSimpleTransform([&](const Block & in_header) -> ProcessorPtr
        {
            return std::make_shared<MaterializingTransform>(in_header);
        });
        size_t num_select_threads = pipeline.getNumThreads();
        size_t num_insert_threads = std::max_element(out_chains.begin(), out_chains.end(), [&](const auto &a, const auto &b)
        {
            return a.getNumThreads() < b.getNumThreads();
        })->getNumThreads();
        for (auto & chain : out_chains)
            resources = chain.detachResources();
        pipeline.addChains(std::move(out_chains));
        pipeline.setMaxThreads(num_insert_threads);
        
        if (!settings.parallel_view_processing && pipeline.getNumThreads() > num_select_threads)
            pipeline.setMaxThreads(num_select_threads);