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

: in(file_name)
        , decompressing_in(in)
        , block_in(decompressing_in, DBMS_TCP_PROTOCOL_VERSION)
        , log{&Poco::Logger::get("DirectoryMonitorBlockInputStream")}
    {
        readDistributedHeader(in, log);
        block_in.readPrefix();
        first_block = block_in.read();
        header = first_block.cloneEmpty();
    }
    String getName() const override { return "DirectoryMonitor"; }
protected:
    Block getHeader() const override { return header; }
    Block readImpl() override
    {
        if (first_block)
            return std::move(first_block);
        return block_in.read();
    }
    void readSuffix() override { block_in.readSuffix(); }
private:
    ReadBufferFromFile in;
    CompressedReadBuffer decompressing_in;
    NativeBlockInputStream block_in;
    Block first_block;
    Block header;
    Poco::Logger * log;
};
BlockInputStreamPtr StorageDistributedDirectoryMonitor::createStreamFromFile(const String & file_name)