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

settings.distributed_replica_error_cap);
}
std::map<UInt64, std::string> StorageDistributedDirectoryMonitor::getFiles()
{
    std::map<UInt64, std::string> files;
    size_t new_bytes_count = 0;
    Poco::DirectoryIterator end;
    for (Poco::DirectoryIterator it{path}; it != end; ++it)
    {
        const auto & file_path_str = it->path();
        Poco::Path file_path{file_path_str};
        if (!it->isDirectory() && startsWith(file_path.getExtension(), "bin"))
        {
            files[parse<UInt64>(file_path.getBaseName())] = file_path_str;
            new_bytes_count += Poco::File(file_path).getSize();
        }
    }
    {
        std::lock_guard metrics_lock(metrics_mutex);
        if (files_count != files.size())
            LOG_TRACE(log, "Files set to {} (was {})", files.size(), files_count);
        if (bytes_count != new_bytes_count)
            LOG_TRACE(log, "Bytes set to {} (was {})", new_bytes_count, bytes_count);
        metric_pending_files.changeTo(files.size());
        files_count = files.size();
        bytes_count = new_bytes_count;
    }
    return files;
}
bool StorageDistributedDirectoryMonitor::processFiles(const std::map<UInt64, std::string> & files)