We have a big replicated table (~15TB/server compressed) with two replicas. When I tried to add two more replicas on new servers, they started to restart because of OOM:
![image](https://user-images.githubusercontent.com/979624/183004927-32fbbf77-854b-4bc0-805e-59d8832e7922.png)

I tried to set the `max_replicated_fetches_network_bandwidth` setting on the new replicas but it didn't help; the RAM usage continues to grow, although more slowly, until it eventually blows up again:
![image](https://user-images.githubusercontent.com/979624/183005247-27a1e354-42cc-465e-8b7c-f2d302e65735.png)

I was looking for what's causing it and it appears Clickhouse is fetching the parts into memory first, as the numbers of the below query correlate with the RAM usage reported by the system:
```
SELECT
    database,
    table,
    formatReadableSize(sum(total_size_bytes_compressed)) AS total,
    formatReadableSize(sum(bytes_read_compressed)) AS read
FROM system.replicated_fetches
GROUP BY
    database,
    table

Query id: f7d90b30-04a8-441f-a809-ed8c9f8fdf69

┌─database────────────┬─table────────────┬─total────┬─read─────┐
│ xxxxxxxxxx_20210401 │ xxxxxxxxxxxxxxxx │ 2.29 TiB │ 1.17 TiB │
└─────────────────────┴──────────────────┴──────────┴──────────┘

1 row in set. Elapsed: 0.004 sec. 
```
I can't see any setting that would limit the RAM usage by replicated fetches, and it's also quite unexpected that RAM is used instead of writing the replicated parts straight to the disk. Am I missing something?
{
            
            const auto & our_checksum = part->checksums.files.find(name + ".proj")->second;
            data_checksums.addFile(name + ".proj", our_checksum.file_size, our_checksum.file_hash);
        }
    }
    writeBinary(checksums.files.size(), out);
    for (const auto & it : checksums.files)
    {
        String file_name = it.first;
        UInt64 size = part->data_part_storage->getFileSize(file_name);
        writeStringBinary(it.first, out);
        writeBinary(size, out);
        auto file_in = part->data_part_storage->readFile(file_name, {}, std::nullopt, std::nullopt);
        HashingWriteBuffer hashing_out(out);
        copyDataWithThrottler(*file_in, hashing_out, blocker.getCounter(), data.getSendsThrottler());
        if (blocker.isCancelled())
            throw Exception("Transferring part to replica was cancelled", ErrorCodes::ABORTED);
        if (hashing_out.count() != size)
            throw Exception(
                ErrorCodes::BAD_SIZE_OF_FILE_IN_DATA_PART,
                "Unexpected size of file {}, expected {} got {}",
                std::string(fs::path(part->data_part_storage->getRelativePath()) / file_name),
                hashing_out.count(), size);
        writePODBinary(hashing_out.getHash(), out);
        if (!file_names_without_checksums.contains(file_name))
            data_checksums.addFile(file_name, hashing_out.count(), hashing_out.getHash());
    }
    part->checksums.checkEqual(data_checksums, false);
    return data_checksums;