**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 (!remote.getHeader())
        {
            CheckingCompressedReadBuffer checking_in(in);
            remote.writePrepared(checking_in);
            return;
        }
        
        
        
        
        if (!distributed_header.block_header)
        {
            LOG_TRACE(log, "Processing batch {} with old format (no header)", in.getFileName());
            writeAndConvert(remote, in);
            return;
        }
        if (!blocksHaveEqualStructure(distributed_header.block_header, remote.getHeader()))
        {
            LOG_WARNING(log,
                "Structure does not match (remote: {}, local: {}), implicit conversion will be done",
                remote.getHeader().dumpStructure(), distributed_header.block_header.dumpStructure());
            writeAndConvert(remote, in);
            return;
        }
        CheckingCompressedReadBuffer checking_in(in);
        remote.writePrepared(checking_in);
    }
}
StorageDistributedDirectoryMonitor::StorageDistributedDirectoryMonitor(