**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] [                                                   <=>                                                                                                                                                                  ]
```

{
        maybeMarkAsBroken(file_path, e);
        throw;
    }
    auto dir_sync_guard = getDirectorySyncGuard(dir_fsync, disk, relative_path);
    markAsSend(file_path);
    LOG_TRACE(log, "Finished processing `{}` (took {} ms)", file_path, watch.elapsedMilliseconds());
}
struct StorageDistributedDirectoryMonitor::BatchHeader
{
    Settings settings;
    String query;
    ClientInfo client_info;
    Block header;
    BatchHeader(Settings settings_, String query_, ClientInfo client_info_, Block header_)
        : settings(std::move(settings_))
        , query(std::move(query_))
        , client_info(std::move(client_info_))
        , header(std::move(header_))
    {
    }
    bool operator==(const BatchHeader & other) const
    {
        return std::tie(settings, query, client_info.query_kind) ==
               std::tie(other.settings, other.query, other.client_info.query_kind) &&
               blocksHaveEqualStructure(header, other.header);
    }
    struct Hash
    {
        size_t operator()(const BatchHeader & batch_header) const