Method Upload

cpp/src/arrow/filesystem/s3fs.cc:1942–1998

  template <typename RequestType, typename OutcomeType>
  Status Upload(
      RequestType&& req,
      UploadResultCallbackFunction<RequestType, OutcomeType> sync_result_callback,
      UploadResultCallbackFunction<RequestType, OutcomeType> async_result_callback,
      const void* data, int64_t nbytes, std::shared_ptr<Buffer> owned_buffer = nullptr) {
    req.SetBucket(ToAwsString(path_.bucket));
    req.SetKey(ToAwsString(path_.key));
    req.SetContentLength(nbytes);
    RETURN_NOT_OK(SetSSECustomerKey(&req, sse_customer_key_));

    if (!background_writes_) {
      // GH-45304: avoid setting a body stream if length is 0.
      // This workaround can be removed once we require AWS SDK 1.11.489 or later.
      if (nbytes != 0) {
        req.SetBody(std::make_shared<StringViewStream>(data, nbytes));
      }

      ARROW_ASSIGN_OR_RAISE(auto outcome, TriggerUploadRequest(req, holder_));

      RETURN_NOT_OK(sync_result_callback(req, upload_state_, part_number_, outcome));
    } else {
      // (GH-45304: avoid setting a body stream if length is 0, see above)
      if (nbytes != 0) {
        // If the data isn't owned, make an immutable copy for the lifetime of the closure
        if (owned_buffer == nullptr) {
          ARROW_ASSIGN_OR_RAISE(owned_buffer, AllocateBuffer(nbytes, io_context_.pool()));
          memcpy(owned_buffer->mutable_data(), data, nbytes);
        } else {
          DCHECK_EQ(data, owned_buffer->data());
          DCHECK_EQ(nbytes, owned_buffer->size());
        }
        req.SetBody(std::make_shared<StringViewStream>(owned_buffer->data(),
                                                       owned_buffer->size()));
      }

      {
        std::unique_lock<std::mutex> lock(upload_state_->mutex);
        if (upload_state_->uploads_in_progress++ == 0) {
          upload_state_->pending_uploads_completed = Future<>::Make();
        }
      }

      // The closure keeps the buffer and the upload state alive
      auto deferred = [owned_buffer, holder = holder_, req = std::move(req),
                       state = upload_state_, async_result_callback,
                       part_number = part_number_]() mutable -> Status {
        ARROW_ASSIGN_OR_RAISE(auto outcome, TriggerUploadRequest(req, holder));

        return async_result_callback(req, state, part_number, outcome);
      };
      RETURN_NOT_OK(SubmitIO(io_context_, std::move(deferred)));
    }

    ++part_number_;

    return Status::OK();
  }