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?
}
namespace DB
{
namespace ErrorCodes
{
    extern const int DIRECTORY_ALREADY_EXISTS;
    extern const int NO_SUCH_DATA_PART;
    extern const int ABORTED;
    extern const int BAD_SIZE_OF_FILE_IN_DATA_PART;
    extern const int CANNOT_WRITE_TO_OSTREAM;
    extern const int CHECKSUM_DOESNT_MATCH;
    extern const int INSECURE_PATH;
    extern const int CORRUPTED_DATA;
    extern const int LOGICAL_ERROR;
    extern const int S3_ERROR;
    extern const int INCORRECT_PART_TYPE;
    extern const int ZERO_COPY_REPLICATION_ERROR;
}
namespace DataPartsExchange
{
namespace
{
constexpr auto REPLICATION_PROTOCOL_VERSION_WITH_PARTS_SIZE = 1;
constexpr auto REPLICATION_PROTOCOL_VERSION_WITH_PARTS_SIZE_AND_TTL_INFOS = 2;
constexpr auto REPLICATION_PROTOCOL_VERSION_WITH_PARTS_TYPE = 3;
constexpr auto REPLICATION_PROTOCOL_VERSION_WITH_PARTS_DEFAULT_COMPRESSION = 4;
constexpr auto REPLICATION_PROTOCOL_VERSION_WITH_PARTS_UUID = 5;
constexpr auto REPLICATION_PROTOCOL_VERSION_WITH_PARTS_ZERO_COPY = 6;
constexpr auto REPLICATION_PROTOCOL_VERSION_WITH_PARTS_PROJECTION = 7;
std::string getEndpointId(const std::string & node_id)