**Describe the situation**
```
$ ./clickhouse-21.3.3 client --query "SELECT * FROM system.numbers_mt FORMAT Native" | pv > /dev/null
^C,1GiB 0:00:45 [ 710MiB/s] [                                                                                                                                                                                               <=>                      ]

$ ./clickhouse-21.3.3 client --compress 0 --query "SELECT * FROM system.numbers_mt FORMAT Native" | pv > /dev/null
^C,2GiB 0:00:12 [1,48GiB/s] [                                                   <=>                                                                                                                                                                  ]
```

{
    Stopwatch watch;
    auto timeouts = ConnectionTimeouts::getTCPTimeoutsWithFailover(storage.getContext()->getSettingsRef());
    try
    {
        CurrentMetrics::Increment metric_increment{CurrentMetrics::DistributedSend};
        ReadBufferFromFile in(file_path);
        const auto & distributed_header = readDistributedHeader(in, log);
        LOG_DEBUG(log, "Started processing `{}` ({} rows, {} bytes)", file_path,
            formatReadableQuantity(distributed_header.rows),
            formatReadableSizeWithBinarySuffix(distributed_header.bytes));
        auto connection = pool->get(timeouts, &distributed_header.insert_settings);
        RemoteBlockOutputStream remote{*connection, timeouts,
            distributed_header.insert_query,
            distributed_header.insert_settings,
            distributed_header.client_info};
        remote.writePrefix();
        writeRemoteConvert(distributed_header, remote, in, log);
        remote.writeSuffix();
    }
    catch (const Exception & e)