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

size_t total_bytes = 0;
        Block header;
        DistributedHeader distributed_header;
        try
        {
            
            ReadBufferFromFile in{file_path};
            distributed_header = readDistributedHeader(in, log);
            if (distributed_header.rows)
            {
                total_rows += distributed_header.rows;
                total_bytes += distributed_header.bytes;
            }
            if (distributed_header.block_header)
                header = distributed_header.block_header;
            if (!total_rows || !header)
            {
                LOG_DEBUG(log, "Processing batch {} with old format (no header/rows)", in.getFileName());
                CompressedReadBuffer decompressing_in(in);
                NativeBlockInputStream block_in(decompressing_in, DBMS_TCP_PROTOCOL_VERSION);
                block_in.readPrefix();
                while (Block block = block_in.read())
                {
                    total_rows += block.rows();
                    total_bytes += block.bytes();
                    if (!header)
                        header = block.cloneEmpty();
                }
                block_in.readSuffix();
            }