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

remote->writePrefix();
                }
                writeRemoteConvert(distributed_header, *remote, in, parent.log);
            }
            if (remote)
                remote->writeSuffix();
        }
        catch (const Exception & e)
        {
            if (isFileBrokenErrorCode(e.code(), e.isRemoteException()))
            {
                tryLogCurrentException(parent.log, "Failed to send batch due to");
                batch_broken = true;
            }
            else
                throw;
        }
        if (!batch_broken)
        {
            LOG_TRACE(parent.log, "Sent a batch of {} files (took {} ms).", file_indices.size(), watch.elapsedMilliseconds());
            auto dir_sync_guard = getDirectorySyncGuard(dir_fsync, parent.disk, parent.relative_path);
            for (UInt64 file_index : file_indices)
                parent.markAsSend(file_index_to_path.at(file_index));
        }
        else
        {
            LOG_ERROR(parent.log, "Marking a batch of {} files as broken.", file_indices.size());
            for (UInt64 file_idx : file_indices)
            {
                auto file_path = file_index_to_path.find(file_idx);