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

{
    return std::make_shared<DirectoryMonitorBlockInputStream>(file_name);
}
bool StorageDistributedDirectoryMonitor::addAndSchedule(size_t file_size, size_t ms)
{
    if (quit)
        return false;
    {
        std::lock_guard metrics_lock(metrics_mutex);
        metric_pending_files.add();
        bytes_count += file_size;
        ++files_count;
    }
    return task_handle->scheduleAfter(ms, false);
}
StorageDistributedDirectoryMonitor::Status StorageDistributedDirectoryMonitor::getStatus()
{
    std::lock_guard metrics_lock(metrics_mutex);
    return Status{
        path,
        last_exception,
        error_count,
        files_count,
        bytes_count,
        monitor_blocker.isCancelled(),
    };
}
void StorageDistributedDirectoryMonitor::processFilesWithBatching(const std::map<UInt64, std::string> & files)