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

{
        if (dir_fsync)
            return disk->getDirectorySyncGuard(path);
        return nullptr;
    }
    void writeAndConvert(RemoteBlockOutputStream & remote, ReadBufferFromFile & in)
    {
        CompressedReadBuffer decompressing_in(in);
        NativeBlockInputStream block_in(decompressing_in, DBMS_TCP_PROTOCOL_VERSION);
        block_in.readPrefix();
        while (Block block = block_in.read())
        {
            ConvertingBlockInputStream convert(
                std::make_shared<OneBlockInputStream>(block),
                remote.getHeader(),
                ConvertingBlockInputStream::MatchColumnsMode::Name);
            auto adopted_block = convert.read();
            remote.write(adopted_block);
        }
        block_in.readSuffix();
    }
    void writeRemoteConvert(const DistributedHeader & distributed_header, RemoteBlockOutputStream & remote, ReadBufferFromFile & in, Poco::Logger * log)