Method DoWrite

cpp/src/arrow/filesystem/s3fs.cc:1889–1937

  }

  Status DoWrite(const void* data, int64_t nbytes,
                 std::shared_ptr<Buffer> owned_buffer = nullptr) {
    if (closed_) {
      return Status::Invalid("Operation on closed stream");
    }

    const int8_t* data_ptr = reinterpret_cast<const int8_t*>(data);
    auto advance_ptr = [&data_ptr, &nbytes](const int64_t offset) {
      data_ptr += offset;
      nbytes -= offset;
    };

    // Handle case where we have some bytes buffered from prior calls.
    if (current_part_size_ > 0) {
      // Try to fill current buffer
      const int64_t to_copy = std::min(nbytes, kPartUploadSize - current_part_size_);
      RETURN_NOT_OK(current_part_->Write(data_ptr, to_copy));
      current_part_size_ += to_copy;
      advance_ptr(to_copy);
      pos_ += to_copy;

      // If buffer isn't full, break
      if (current_part_size_ < kPartUploadSize) {
        return Status::OK();
      }

      // Upload current buffer. We're only reaching this point if we have accumulated
      // enough data to upload.
      RETURN_NOT_OK(CommitCurrentPart());
    }

    // We can upload chunks without copying them into a buffer
    while (nbytes >= kPartUploadSize) {
      RETURN_NOT_OK(UploadPart(data_ptr, kPartUploadSize));
      advance_ptr(kPartUploadSize);
      pos_ += kPartUploadSize;
    }

    // Buffer remaining bytes
    if (nbytes > 0) {
      current_part_size_ = nbytes;
      ARROW_ASSIGN_OR_RAISE(current_part_, io::BufferOutputStream::Create(
                                               kPartUploadSize, io_context_.pool()));
      RETURN_NOT_OK(current_part_->Write(data_ptr, current_part_size_));
      pos_ += current_part_size_;
    }

    return Status::OK();
  }

  Status Flush() override {
    auto fut = FlushAsync();