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).
out.getInputHeader(),
            table_prefers_large_blocks ? settings.min_insert_block_size_rows : settings.max_block_size,
            table_prefers_large_blocks ? settings.min_insert_block_size_bytes : 0));
    }
    auto counting = std::make_shared<CountingTransform>(out.getInputHeader(), thread_status, getContext()->getQuota());
    counting->setProcessListElement(context_ptr->getProcessListElement());
    out.addSource(std::move(counting));
    return out;
}
BlockIO InterpreterInsertQuery::execute()
{
    const Settings & settings = getContext()->getSettingsRef();
    auto & query = query_ptr->as<ASTInsertQuery &>();
    QueryPipelineBuilder pipeline;
    std::optional<QueryPipeline> distributed_pipeline;
    QueryPlanResourceHolder resources;
    StoragePtr table = getTable(query);
    checkStorageSupportsTransactionsIfNeeded(table, getContext());
    StoragePtr inner_table;
    if (const auto * mv = dynamic_cast<const StorageMaterializedView *>(table.get()))
        inner_table = mv->getTargetTable();
    if (query.partition_by && !table->supportsPartitionBy())
        throw Exception(ErrorCodes::NOT_IMPLEMENTED, "PARTITION BY clause is not supported by storage");
    auto table_lock = table->lockForShare(getContext()->getInitialQueryId(), settings.lock_acquire_timeout);
    auto metadata_snapshot = table->getInMemoryMetadataPtr();
    auto query_sample_block = getSampleBlock(query, table, metadata_snapshot);
    
    
    if (!query.table_function)
        getContext()->checkAccess(AccessType::INSERT, query.table_id, query_sample_block.getNames());