**Describe the situation**

The author implemented it but enabled only for DiskS3.
CC @excitoon 
{
        std::lock_guard lock(mutex);
        return nextAssumeLocked();
    }
private:
    String nextAssumeLocked()
    {
        if (buffer_iter != buffer.end())
        {
            auto answer = *buffer_iter;
            ++buffer_iter;
            return answer;
        }
        if (is_finished)
            return {};
        fillInternalBufferAssumeLocked();
        return nextAssumeLocked();
    }
    void fillInternalBufferAssumeLocked()
    {
        buffer.clear();
        outcome = client.ListObjectsV2(request);
        if (!outcome.IsSuccess())
            throw Exception(ErrorCodes::S3_ERROR, "Could not list objects in bucket {} with prefix {}, S3 exception: {}, message: {}",
                            quoteString(request.GetBucket()), quoteString(request.GetPrefix()),
                            backQuote(outcome.GetError().GetExceptionName()), quoteString(outcome.GetError().GetMessage()));
        const auto & result_batch = outcome.GetResult().GetContents();