You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by "OliLay (via GitHub)" <gi...@apache.org> on 2024/05/07 07:34:31 UTC
[PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
OliLay opened a new pull request, #41564:
URL: https://github.com/apache/arrow/pull/41564
### Rationale for this change
See #40557. The previous implementation would always issue multi part uploads which come with 3x RTT to S3 instead of just 1x RTT with a `PutObject` request.
### What changes are included in this PR?
Implement logic in the S3 `OutputStream` to use a `PutObject` request if data is below a certain threshold (5 MB) and the output stream is closed. If more data is written, a multi part upload is triggered. Note: Previously, opening the output stream was already expensive because the `CreateMultipartUpload` request was triggered then. With this change opening the output stream becomes cheap, as we rather wait until some data is written to decide which upload method to use. This required some more state-keeping in the output stream class.
### Are these changes tested?
No new tests were added, as there are already tests for very small writes and very large writes, which will trigger both ways of uploading. Everything should therefore be covered by existing tests.
### Are there any user-facing changes?
- Previously, we would fail when opening the output stream if the object already exists. We inferred that by sending the `CreateMultipartUpload` request, which we now do not send anymore upon opening the stream. We now rather fail at closing, or at writing (when >5MB have accumulated). Replicating the old behavior is not possible without sending another request which defeats the purpose of this performance optimization. I hope this is fine.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "OliLay (via GitHub)" <gi...@apache.org>.
OliLay commented on code in PR #41564:
URL: https://github.com/apache/arrow/pull/41564#discussion_r1604485403
##########
cpp/src/arrow/filesystem/s3fs.h:
##########
@@ -166,6 +166,13 @@ struct ARROW_EXPORT S3Options {
/// Whether to allow deletion of buckets
bool allow_bucket_deletion = false;
+ /// Whether to sanitize that a bucket can be written to when calling OpenOutputStream.
+ /// Disabling this will lead to less S3 requests issued and is therefore relevant
+ /// for latency-sensitive applications, at the cost of the OutputStream may throwing an
+ /// exception at a later stage (i.e. at writing or closing) if e.g. the bucket does not
+ /// exist.
+ bool sanitize_bucket_on_open = true;
Review Comment:
Yes, I like this much more :+1:
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "OliLay (via GitHub)" <gi...@apache.org>.
OliLay commented on PR #41564:
URL: https://github.com/apache/arrow/pull/41564#issuecomment-2110043693
Rebased to current main, now waiting for the CI approval again :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on PR #41564:
URL: https://github.com/apache/arrow/pull/41564#issuecomment-2224347360
Gental ping @pitrou @felipecrv . This could be in 18.0.0
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "orf (via GitHub)" <gi...@apache.org>.
orf commented on PR #41564:
URL: https://github.com/apache/arrow/pull/41564#issuecomment-2110143195
> > Previously, we would fail when opening the output stream if the bucket doesn't exist. We inferred that by sending the `CreateMultipartUpload` request, which we now do not send anymore upon opening the stream. We now rather fail at closing, or at writing (when >5MB have accumulated).
>
> Hmm, I'm not sure that is ok. Usually, when opening a file for writing, you expect the initial open to fail if the path cannot be written to. I have no idea how much code relies on that, but that's a common expectation due to how filesystems usually work (e.g. when accessing local storage).
This isn’t guaranteed with the current implementation though? Putting a part, or completing a multipart upload, can fail in various ways? An obvious one would be a checksum failure.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #41564:
URL: https://github.com/apache/arrow/pull/41564#discussion_r1600112929
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1545,6 +1559,13 @@ class ObjectInputFile final : public io::RandomAccessFile {
// (for rational, see: https://github.com/apache/arrow/issues/34363)
static constexpr int64_t kPartUploadSize = 10 * 1024 * 1024;
+// Above this threshold, use a multi-part upload instead of a single request upload.
+static constexpr int64_t kMultiPartUploadThresholdSize = 5 * 1024 * 1024;
Review Comment:
Both is ok for me, here S3OutputStream might manage these part of memory in memory buffer, so I think it shouldn't be too large? changing it default 10MB or making it tunable later is both ok for me
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "pitrou (via GitHub)" <gi...@apache.org>.
pitrou commented on code in PR #41564:
URL: https://github.com/apache/arrow/pull/41564#discussion_r1603507590
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1734,7 +1812,7 @@ class ObjectOutputStream final : public io::OutputStream {
return Status::OK();
}
- // Upload current buffer
+ // Upload current buffer if we are above threshold for multi-part upload
Review Comment:
The comment is misleading: this always uploads the current buffer, right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "pitrou (via GitHub)" <gi...@apache.org>.
pitrou commented on PR #41564:
URL: https://github.com/apache/arrow/pull/41564#issuecomment-2110323934
> We can do that. I would propose that if the optimization is disabled, we directly use multi-part uploads (basically replicating the old behavior).
That sounds reasonable to me.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "OliLay (via GitHub)" <gi...@apache.org>.
OliLay commented on code in PR #41564:
URL: https://github.com/apache/arrow/pull/41564#discussion_r1601475036
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1838,9 +1952,114 @@ class ObjectOutputStream final : public io::OutputStream {
return Status::OK();
}
- static void HandleUploadOutcome(const std::shared_ptr<UploadState>& state,
- int part_number, const S3Model::UploadPartRequest& req,
- const Result<S3Model::UploadPartOutcome>& result) {
+ static Status UploadError(const Aws::S3::Model::PutObjectRequest& request,
+ const Aws::S3::Model::PutObjectOutcome& outcome) {
+ return ErrorToStatus(
+ std::forward_as_tuple("When uploading object with key '", request.GetKey(),
+ "' in bucket '", request.GetBucket(), "': "),
+ "PutObject", outcome.GetError());
+ }
+
+ Status UploadUsingSingleRequest(std::shared_ptr<Buffer> buffer) {
+ return UploadUsingSingleRequest(buffer->data(), buffer->size(), buffer);
+ }
+
+ Status UploadUsingSingleRequest(const void* data, int64_t nbytes,
+ std::shared_ptr<Buffer> owned_buffer = nullptr) {
+ auto sync_result_callback = [](const Aws::S3::Model::PutObjectRequest& request,
+ std::shared_ptr<UploadState> state,
+ int32_t part_number,
+ Aws::S3::Model::PutObjectOutcome outcome) {
+ if (!outcome.IsSuccess()) {
+ return UploadError(request, outcome);
+ }
+ return Status::OK();
+ };
+
+ auto async_result_callback = [](const Aws::S3::Model::PutObjectRequest& request,
Review Comment:
Yes, previously in the sync case we also directly called `AddCompletedPart` which just used the `state` to fiddle around with its `completed_parts` member. For a single upload request, I think we can omit this and hence do not need to modify the state at all.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "pitrou (via GitHub)" <gi...@apache.org>.
pitrou commented on code in PR #41564:
URL: https://github.com/apache/arrow/pull/41564#discussion_r1603527121
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1838,9 +1952,114 @@ class ObjectOutputStream final : public io::OutputStream {
return Status::OK();
}
- static void HandleUploadOutcome(const std::shared_ptr<UploadState>& state,
- int part_number, const S3Model::UploadPartRequest& req,
- const Result<S3Model::UploadPartOutcome>& result) {
+ static Status UploadError(const Aws::S3::Model::PutObjectRequest& request,
+ const Aws::S3::Model::PutObjectOutcome& outcome) {
+ return ErrorToStatus(
+ std::forward_as_tuple("When uploading object with key '", request.GetKey(),
+ "' in bucket '", request.GetBucket(), "': "),
+ "PutObject", outcome.GetError());
+ }
+
+ Status UploadUsingSingleRequest(std::shared_ptr<Buffer> buffer) {
+ return UploadUsingSingleRequest(buffer->data(), buffer->size(), buffer);
+ }
+
+ Status UploadUsingSingleRequest(const void* data, int64_t nbytes,
+ std::shared_ptr<Buffer> owned_buffer = nullptr) {
+ auto sync_result_callback = [](const Aws::S3::Model::PutObjectRequest& request,
+ std::shared_ptr<UploadState> state,
+ int32_t part_number,
+ Aws::S3::Model::PutObjectOutcome outcome) {
+ if (!outcome.IsSuccess()) {
+ return UploadError(request, outcome);
+ }
+ return Status::OK();
+ };
+
+ auto async_result_callback = [](const Aws::S3::Model::PutObjectRequest& request,
+ std::shared_ptr<UploadState> state,
+ int32_t part_number,
+ Aws::S3::Model::PutObjectOutcome outcome) {
+ HandleUploadUsingSingleRequestOutcome(state, request, outcome.GetResult());
+ return Status::OK();
+ };
+
+ Aws::S3::Model::PutObjectRequest req{};
+ RETURN_NOT_OK(SetMetadataInRequest(&req));
+
+ return Upload<Aws::S3::Model::PutObjectRequest, Aws::S3::Model::PutObjectOutcome>(
+ std::move(req), sync_result_callback, async_result_callback, data, nbytes,
+ owned_buffer);
+ }
+
+ Status UploadPart(std::shared_ptr<Buffer> buffer) {
+ return UploadPart(buffer->data(), buffer->size(), buffer);
+ }
+
+ static Status UploadPartError(const Aws::S3::Model::UploadPartRequest& request,
+ const Aws::S3::Model::UploadPartOutcome& outcome) {
+ return ErrorToStatus(
+ std::forward_as_tuple("When uploading part for key '", request.GetKey(),
+ "' in bucket '", request.GetBucket(), "': "),
+ "UploadPart", outcome.GetError());
+ }
+
+ Status UploadPart(const void* data, int64_t nbytes,
+ std::shared_ptr<Buffer> owned_buffer = nullptr) {
+ if (!is_multipart_created_) {
+ RETURN_NOT_OK(CreateMultipartUpload());
+ }
+
+ Aws::S3::Model::UploadPartRequest req{};
+ req.SetPartNumber(part_number_);
+ req.SetUploadId(upload_id_);
+
+ auto sync_result_callback = [](const Aws::S3::Model::UploadPartRequest& request,
+ std::shared_ptr<UploadState> state,
+ int32_t part_number,
+ Aws::S3::Model::UploadPartOutcome outcome) {
+ if (!outcome.IsSuccess()) {
+ return UploadPartError(request, outcome);
+ } else {
+ AddCompletedPart(state, part_number, outcome.GetResult());
+ }
+
+ return Status::OK();
+ };
+
+ auto async_result_callback = [](const Aws::S3::Model::UploadPartRequest& request,
+ std::shared_ptr<UploadState> state,
+ int32_t part_number,
+ Aws::S3::Model::UploadPartOutcome outcome) {
+ HandleUploadPartOutcome(state, part_number, request, outcome.GetResult());
+ return Status::OK();
+ };
+
+ return Upload<Aws::S3::Model::UploadPartRequest, Aws::S3::Model::UploadPartOutcome>(
+ std::move(req), sync_result_callback, async_result_callback, data, nbytes,
+ owned_buffer);
Review Comment:
Same here: more arguments can be moved.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "orf (via GitHub)" <gi...@apache.org>.
orf commented on PR #41564:
URL: https://github.com/apache/arrow/pull/41564#issuecomment-2242743024
@OliLay thank you for your work on this ❤️
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "pitrou (via GitHub)" <gi...@apache.org>.
pitrou commented on PR #41564:
URL: https://github.com/apache/arrow/pull/41564#issuecomment-2242575613
@github-actions crossbow submit -g cpp -g python
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "pitrou (via GitHub)" <gi...@apache.org>.
pitrou commented on code in PR #41564:
URL: https://github.com/apache/arrow/pull/41564#discussion_r1624556269
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1293,12 +1295,14 @@ std::shared_ptr<const KeyValueMetadata> GetObjectMetadata(const ObjectResult& re
template <typename ObjectRequest>
struct ObjectMetadataSetter {
+ static constexpr std::string_view kContentTypeKey = "Content-Type";
Review Comment:
"Content-Type" is still used as a literal above. Should we move this at the top-level and use it everywhere?
Or, conversely, just undo this, since the "Content-Type" literal is unlikely to change value...
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1606,42 +1656,62 @@ class ObjectOutputStream final : public io::OutputStream {
return Status::OK();
}
- ARROW_ASSIGN_OR_RAISE(auto client_lock, holder_->Lock());
+ if (is_multipart_created_) {
+ ARROW_ASSIGN_OR_RAISE(auto client_lock, holder_->Lock());
- S3Model::AbortMultipartUploadRequest req;
- req.SetBucket(ToAwsString(path_.bucket));
- req.SetKey(ToAwsString(path_.key));
- req.SetUploadId(upload_id_);
+ S3Model::AbortMultipartUploadRequest req;
+ req.SetBucket(ToAwsString(path_.bucket));
+ req.SetKey(ToAwsString(path_.key));
+ req.SetUploadId(upload_id_);
- auto outcome = client_lock.Move()->AbortMultipartUpload(req);
- if (!outcome.IsSuccess()) {
- return ErrorToStatus(
- std::forward_as_tuple("When aborting multiple part upload for key '", path_.key,
- "' in bucket '", path_.bucket, "': "),
- "AbortMultipartUpload", outcome.GetError());
+ auto outcome = client_lock.Move()->AbortMultipartUpload(req);
+ if (!outcome.IsSuccess()) {
+ return ErrorToStatus(
+ std::forward_as_tuple("When aborting multiple part upload for key '",
+ path_.key, "' in bucket '", path_.bucket, "': "),
+ "AbortMultipartUpload", outcome.GetError());
+ }
}
+
current_part_.reset();
holder_ = nullptr;
closed_ = true;
+
return Status::OK();
}
// OutputStream interface
+ bool ShouldBeMultipartUpload() const { return pos_ > kMultiPartUploadThresholdSize; }
Review Comment:
Why not instead
```suggestion
bool ShouldBeMultipartUpload() const {
return pos_ > kMultiPartUploadThresholdSize || !allow_delayed_open_;
}
```
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1838,9 +1949,114 @@ class ObjectOutputStream final : public io::OutputStream {
return Status::OK();
}
- static void HandleUploadOutcome(const std::shared_ptr<UploadState>& state,
- int part_number, const S3Model::UploadPartRequest& req,
- const Result<S3Model::UploadPartOutcome>& result) {
+ static Status UploadError(const Aws::S3::Model::PutObjectRequest& request,
Review Comment:
Perhaps call this `UploadUsingSingleRequestError`?
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1606,42 +1656,62 @@ class ObjectOutputStream final : public io::OutputStream {
return Status::OK();
}
- ARROW_ASSIGN_OR_RAISE(auto client_lock, holder_->Lock());
+ if (is_multipart_created_) {
+ ARROW_ASSIGN_OR_RAISE(auto client_lock, holder_->Lock());
- S3Model::AbortMultipartUploadRequest req;
- req.SetBucket(ToAwsString(path_.bucket));
- req.SetKey(ToAwsString(path_.key));
- req.SetUploadId(upload_id_);
+ S3Model::AbortMultipartUploadRequest req;
+ req.SetBucket(ToAwsString(path_.bucket));
+ req.SetKey(ToAwsString(path_.key));
+ req.SetUploadId(upload_id_);
- auto outcome = client_lock.Move()->AbortMultipartUpload(req);
- if (!outcome.IsSuccess()) {
- return ErrorToStatus(
- std::forward_as_tuple("When aborting multiple part upload for key '", path_.key,
- "' in bucket '", path_.bucket, "': "),
- "AbortMultipartUpload", outcome.GetError());
+ auto outcome = client_lock.Move()->AbortMultipartUpload(req);
+ if (!outcome.IsSuccess()) {
+ return ErrorToStatus(
+ std::forward_as_tuple("When aborting multiple part upload for key '",
+ path_.key, "' in bucket '", path_.bucket, "': "),
+ "AbortMultipartUpload", outcome.GetError());
+ }
}
+
current_part_.reset();
holder_ = nullptr;
closed_ = true;
+
return Status::OK();
}
// OutputStream interface
+ bool ShouldBeMultipartUpload() const { return pos_ > kMultiPartUploadThresholdSize; }
+
+ bool IsMultipartUpload() const {
+ return ShouldBeMultipartUpload() || is_multipart_created_;
Review Comment:
Why add `is_multipart_created_` here? Is there any situation where `is_multipart_created_` would be true but `ShouldBeMultipartUpload()` would be false?
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1560,34 +1582,49 @@ class ObjectOutputStream final : public io::OutputStream {
path_(path),
metadata_(metadata),
default_metadata_(options.default_metadata),
- background_writes_(options.background_writes) {}
+ background_writes_(options.background_writes),
+ allow_delayed_open_(options.allow_delayed_open) {}
~ObjectOutputStream() override {
// For compliance with the rest of the IO stack, Close rather than Abort,
// even though it may be more expensive.
io::internal::CloseFromDestructor(this);
}
- Status Init() {
- ARROW_ASSIGN_OR_RAISE(auto client_lock, holder_->Lock());
+ template <typename ObjectRequest>
+ Status SetMetadataInRequest(ObjectRequest* request) {
+ std::shared_ptr<const KeyValueMetadata> metadata;
- // Initiate the multi-part upload
- S3Model::CreateMultipartUploadRequest req;
- req.SetBucket(ToAwsString(path_.bucket));
- req.SetKey(ToAwsString(path_.key));
if (metadata_ && metadata_->size() != 0) {
- RETURN_NOT_OK(SetObjectMetadata(metadata_, &req));
+ metadata = metadata_;
} else if (default_metadata_ && default_metadata_->size() != 0) {
- RETURN_NOT_OK(SetObjectMetadata(default_metadata_, &req));
+ metadata = default_metadata_;
}
- // If we do not set anything then the SDK will default to application/xml
- // which confuses some tools (https://github.com/apache/arrow/issues/11934)
- // So we instead default to application/octet-stream which is less misleading
- if (!req.ContentTypeHasBeenSet()) {
- req.SetContentType("application/octet-stream");
+ if (metadata &&
+ metadata->Contains(ObjectMetadataSetter<ObjectRequest>::kContentTypeKey)) {
+ RETURN_NOT_OK(SetObjectMetadata(metadata, request));
+ } else {
+ // If we do not set anything then the SDK will default to application/xml
+ // which confuses some tools (https://github.com/apache/arrow/issues/11934)
+ // So we instead default to application/octet-stream which is less misleading
+ request->SetContentType("application/octet-stream");
Review Comment:
So `metadata` is ignored if it doesn't contain a "Content-Type" key? Or am I missing something here?
##########
cpp/src/arrow/filesystem/s3fs_test.cc:
##########
@@ -1189,40 +1191,55 @@ TEST_F(TestS3FS, OpenInputFile) {
ASSERT_RAISES(IOError, file->Seek(10));
}
-TEST_F(TestS3FS, OpenOutputStreamBackgroundWrites) { TestOpenOutputStream(); }
+struct S3OptionsTestParameters {
+ bool background_writes{false};
+ bool allow_delayed_open{false};
-TEST_F(TestS3FS, OpenOutputStreamSyncWrites) {
- options_.background_writes = false;
- MakeFileSystem();
- TestOpenOutputStream();
-}
-
-TEST_F(TestS3FS, OpenOutputStreamAbortBackgroundWrites) { TestOpenOutputStreamAbort(); }
+ void apply_to_s3_options(S3Options& options) const {
+ options.background_writes = background_writes;
+ options.allow_delayed_open = allow_delayed_open;
+ }
Review Comment:
Coding style nit: 1) use `CamelCase` for function names, 2) avoid mutable refs; therefore:
```suggestion
void ApplyToS3Options(S3Options* options) const {
options->background_writes = background_writes;
options->allow_delayed_open = allow_delayed_open;
}
```
##########
cpp/src/arrow/filesystem/s3fs_test.cc:
##########
@@ -1189,40 +1191,55 @@ TEST_F(TestS3FS, OpenInputFile) {
ASSERT_RAISES(IOError, file->Seek(10));
}
-TEST_F(TestS3FS, OpenOutputStreamBackgroundWrites) { TestOpenOutputStream(); }
+struct S3OptionsTestParameters {
+ bool background_writes{false};
+ bool allow_delayed_open{false};
-TEST_F(TestS3FS, OpenOutputStreamSyncWrites) {
- options_.background_writes = false;
- MakeFileSystem();
- TestOpenOutputStream();
-}
-
-TEST_F(TestS3FS, OpenOutputStreamAbortBackgroundWrites) { TestOpenOutputStreamAbort(); }
+ void apply_to_s3_options(S3Options& options) const {
+ options.background_writes = background_writes;
+ options.allow_delayed_open = allow_delayed_open;
+ }
-TEST_F(TestS3FS, OpenOutputStreamAbortSyncWrites) {
- options_.background_writes = false;
- MakeFileSystem();
- TestOpenOutputStreamAbort();
-}
+ static std::vector<S3OptionsTestParameters> GetCartesianProduct() {
+ return {
+ S3OptionsTestParameters{true, false},
+ S3OptionsTestParameters{false, false},
+ S3OptionsTestParameters{true, true},
+ S3OptionsTestParameters{false, true},
+ };
+ }
+};
-TEST_F(TestS3FS, OpenOutputStreamDestructorBackgroundWrites) {
- TestOpenOutputStreamDestructor();
+TEST_F(TestS3FS, OpenOutputStream) {
+ for (const auto& combination : S3OptionsTestParameters::GetCartesianProduct()) {
+ combination.apply_to_s3_options(options_);
+ MakeFileSystem();
Review Comment:
I don't think this is deleting the files currently on the filesystem, which means the tests might succeed even if the write path doesn't work for some options.
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1853,8 +2069,8 @@ class ObjectOutputStream final : public io::OutputStream {
}
}
// Notify completion
- if (--state->parts_in_progress == 0) {
- state->pending_parts_completed.MarkFinished(state->status);
+ if (--state->uploads_in_progress == 0) {
+ state->pending_uploads_completed.MarkFinished(state->status);
Review Comment:
Same here.
##########
cpp/src/arrow/filesystem/s3fs_test.cc:
##########
@@ -1189,40 +1191,55 @@ TEST_F(TestS3FS, OpenInputFile) {
ASSERT_RAISES(IOError, file->Seek(10));
}
-TEST_F(TestS3FS, OpenOutputStreamBackgroundWrites) { TestOpenOutputStream(); }
+struct S3OptionsTestParameters {
+ bool background_writes{false};
+ bool allow_delayed_open{false};
-TEST_F(TestS3FS, OpenOutputStreamSyncWrites) {
- options_.background_writes = false;
- MakeFileSystem();
- TestOpenOutputStream();
-}
-
-TEST_F(TestS3FS, OpenOutputStreamAbortBackgroundWrites) { TestOpenOutputStreamAbort(); }
+ void apply_to_s3_options(S3Options& options) const {
+ options.background_writes = background_writes;
+ options.allow_delayed_open = allow_delayed_open;
+ }
-TEST_F(TestS3FS, OpenOutputStreamAbortSyncWrites) {
- options_.background_writes = false;
- MakeFileSystem();
- TestOpenOutputStreamAbort();
-}
+ static std::vector<S3OptionsTestParameters> GetCartesianProduct() {
+ return {
+ S3OptionsTestParameters{true, false},
+ S3OptionsTestParameters{false, false},
+ S3OptionsTestParameters{true, true},
+ S3OptionsTestParameters{false, true},
+ };
+ }
+};
-TEST_F(TestS3FS, OpenOutputStreamDestructorBackgroundWrites) {
- TestOpenOutputStreamDestructor();
+TEST_F(TestS3FS, OpenOutputStream) {
+ for (const auto& combination : S3OptionsTestParameters::GetCartesianProduct()) {
+ combination.apply_to_s3_options(options_);
Review Comment:
Ideally we would also leave a trace in the test log to make diagnosing failures easier (see the `ARROW_SCOPED_TRACE` macro somewhere).
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1873,24 +2089,18 @@ class ObjectOutputStream final : public io::OutputStream {
state->completed_parts[slot] = std::move(part);
}
- static Status UploadPartError(const S3Model::UploadPartRequest& req,
- const S3Model::UploadPartOutcome& outcome) {
- return ErrorToStatus(
- std::forward_as_tuple("When uploading part for key '", req.GetKey(),
- "' in bucket '", req.GetBucket(), "': "),
- "UploadPart", outcome.GetError());
- }
-
protected:
std::shared_ptr<S3ClientHolder> holder_;
const io::IOContext io_context_;
const S3Path path_;
const std::shared_ptr<const KeyValueMetadata> metadata_;
const std::shared_ptr<const KeyValueMetadata> default_metadata_;
const bool background_writes_;
+ const bool allow_delayed_open_;
Aws::String upload_id_;
bool closed_ = true;
+ bool is_multipart_created_ = false;
Review Comment:
For the record, `is_multipart_created_ == true` iff `!upload_id_.empty()`, right? Perhaps this can be consolidated and we can rename `upload_id_` to something more explicit, such as `multipart_upload_id`.
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1838,9 +1949,114 @@ class ObjectOutputStream final : public io::OutputStream {
return Status::OK();
}
- static void HandleUploadOutcome(const std::shared_ptr<UploadState>& state,
- int part_number, const S3Model::UploadPartRequest& req,
- const Result<S3Model::UploadPartOutcome>& result) {
+ static Status UploadError(const Aws::S3::Model::PutObjectRequest& request,
+ const Aws::S3::Model::PutObjectOutcome& outcome) {
+ return ErrorToStatus(
+ std::forward_as_tuple("When uploading object with key '", request.GetKey(),
+ "' in bucket '", request.GetBucket(), "': "),
+ "PutObject", outcome.GetError());
+ }
+
+ Status UploadUsingSingleRequest(std::shared_ptr<Buffer> buffer) {
+ return UploadUsingSingleRequest(buffer->data(), buffer->size(), buffer);
+ }
+
+ Status UploadUsingSingleRequest(const void* data, int64_t nbytes,
+ std::shared_ptr<Buffer> owned_buffer = nullptr) {
+ auto sync_result_callback = [](const Aws::S3::Model::PutObjectRequest& request,
+ std::shared_ptr<UploadState> state,
+ int32_t part_number,
+ Aws::S3::Model::PutObjectOutcome outcome) {
+ if (!outcome.IsSuccess()) {
+ return UploadError(request, outcome);
+ }
+ return Status::OK();
+ };
+
+ auto async_result_callback = [](const Aws::S3::Model::PutObjectRequest& request,
+ std::shared_ptr<UploadState> state,
+ int32_t part_number,
+ Aws::S3::Model::PutObjectOutcome outcome) {
+ HandleUploadUsingSingleRequestOutcome(state, request, outcome.GetResult());
+ return Status::OK();
+ };
+
+ Aws::S3::Model::PutObjectRequest req{};
+ RETURN_NOT_OK(SetMetadataInRequest(&req));
+
+ return Upload<Aws::S3::Model::PutObjectRequest, Aws::S3::Model::PutObjectOutcome>(
+ std::move(req), std::move(sync_result_callback), std::move(async_result_callback),
+ data, nbytes, std::move(owned_buffer));
+ }
+
+ Status UploadPart(std::shared_ptr<Buffer> buffer) {
+ return UploadPart(buffer->data(), buffer->size(), buffer);
+ }
+
+ static Status UploadPartError(const Aws::S3::Model::UploadPartRequest& request,
+ const Aws::S3::Model::UploadPartOutcome& outcome) {
+ return ErrorToStatus(
+ std::forward_as_tuple("When uploading part for key '", request.GetKey(),
+ "' in bucket '", request.GetBucket(), "': "),
+ "UploadPart", outcome.GetError());
+ }
+
+ Status UploadPart(const void* data, int64_t nbytes,
+ std::shared_ptr<Buffer> owned_buffer = nullptr) {
+ if (!is_multipart_created_) {
+ RETURN_NOT_OK(CreateMultipartUpload());
+ }
+
+ Aws::S3::Model::UploadPartRequest req{};
+ req.SetPartNumber(part_number_);
+ req.SetUploadId(upload_id_);
+
+ auto sync_result_callback = [](const Aws::S3::Model::UploadPartRequest& request,
+ std::shared_ptr<UploadState> state,
+ int32_t part_number,
+ Aws::S3::Model::UploadPartOutcome outcome) {
+ if (!outcome.IsSuccess()) {
+ return UploadPartError(request, outcome);
+ } else {
+ AddCompletedPart(state, part_number, outcome.GetResult());
+ }
+
+ return Status::OK();
+ };
+
+ auto async_result_callback = [](const Aws::S3::Model::UploadPartRequest& request,
+ std::shared_ptr<UploadState> state,
+ int32_t part_number,
+ Aws::S3::Model::UploadPartOutcome outcome) {
+ HandleUploadPartOutcome(state, part_number, request, outcome.GetResult());
+ return Status::OK();
+ };
+
+ return Upload<Aws::S3::Model::UploadPartRequest, Aws::S3::Model::UploadPartOutcome>(
+ std::move(req), std::move(sync_result_callback), std::move(async_result_callback),
+ data, nbytes, std::move(owned_buffer));
+ }
+
+ static void HandleUploadUsingSingleRequestOutcome(
+ const std::shared_ptr<UploadState>& state, const S3Model::PutObjectRequest& req,
+ const Result<S3Model::PutObjectOutcome>& result) {
+ std::unique_lock<std::mutex> lock(state->mutex);
+ if (!result.ok()) {
+ state->status &= result.status();
+ } else {
+ const auto& outcome = *result;
+ if (!outcome.IsSuccess()) {
+ state->status &= UploadError(req, outcome);
+ }
+ }
+
+ state->pending_uploads_completed.MarkFinished(state->status);
Review Comment:
Based on https://github.com/apache/arrow/pull/41876/files, we probably want to unlock the mutex before marking the future finished.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "kou (via GitHub)" <gi...@apache.org>.
kou commented on code in PR #41564:
URL: https://github.com/apache/arrow/pull/41564#discussion_r1599248880
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1568,26 +1589,38 @@ class ObjectOutputStream final : public io::OutputStream {
io::internal::CloseFromDestructor(this);
}
- Status Init() {
- ARROW_ASSIGN_OR_RAISE(auto client_lock, holder_->Lock());
+ template <typename ObjectRequest>
+ Status SetMetadataInRequest(ObjectRequest* request) {
+ std::shared_ptr<const KeyValueMetadata> metadata;
- // Initiate the multi-part upload
- S3Model::CreateMultipartUploadRequest req;
- req.SetBucket(ToAwsString(path_.bucket));
- req.SetKey(ToAwsString(path_.key));
if (metadata_ && metadata_->size() != 0) {
- RETURN_NOT_OK(SetObjectMetadata(metadata_, &req));
+ metadata = metadata_;
} else if (default_metadata_ && default_metadata_->size() != 0) {
- RETURN_NOT_OK(SetObjectMetadata(default_metadata_, &req));
+ metadata = default_metadata_;
}
- // If we do not set anything then the SDK will default to application/xml
- // which confuses some tools (https://github.com/apache/arrow/issues/11934)
- // So we instead default to application/octet-stream which is less misleading
- if (!req.ContentTypeHasBeenSet()) {
- req.SetContentType("application/octet-stream");
+ if (metadata == nullptr ||
+ !metadata->Contains(ObjectMetadataSetter<ObjectRequest>::CONTENT_TYPE_KEY)) {
+ // If we do not set anything then the SDK will default to application/xml
+ // which confuses some tools (https://github.com/apache/arrow/issues/11934)
+ // So we instead default to application/octet-stream which is less misleading
+ request->SetContentType("application/octet-stream");
+ } else {
+ RETURN_NOT_OK(SetObjectMetadata(metadata, request));
}
Review Comment:
How about swapping these clauses for easy to read?
```cpp
if (metadata && metadata->Contains(ObjectMetadataSetter<ObjectRequest>::CONTENT_TYPE_KEY)) {
RETURN_NOT_OK(SetObjectMetadata(metadata, request));
} else {
request->SetContentType("application/octet-stream");
}
```
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1568,26 +1589,38 @@ class ObjectOutputStream final : public io::OutputStream {
io::internal::CloseFromDestructor(this);
}
- Status Init() {
- ARROW_ASSIGN_OR_RAISE(auto client_lock, holder_->Lock());
+ template <typename ObjectRequest>
+ Status SetMetadataInRequest(ObjectRequest* request) {
+ std::shared_ptr<const KeyValueMetadata> metadata;
- // Initiate the multi-part upload
- S3Model::CreateMultipartUploadRequest req;
- req.SetBucket(ToAwsString(path_.bucket));
- req.SetKey(ToAwsString(path_.key));
if (metadata_ && metadata_->size() != 0) {
- RETURN_NOT_OK(SetObjectMetadata(metadata_, &req));
+ metadata = metadata_;
} else if (default_metadata_ && default_metadata_->size() != 0) {
- RETURN_NOT_OK(SetObjectMetadata(default_metadata_, &req));
+ metadata = default_metadata_;
}
- // If we do not set anything then the SDK will default to application/xml
- // which confuses some tools (https://github.com/apache/arrow/issues/11934)
- // So we instead default to application/octet-stream which is less misleading
- if (!req.ContentTypeHasBeenSet()) {
- req.SetContentType("application/octet-stream");
+ if (metadata == nullptr ||
Review Comment:
```suggestion
if (!metadata ||
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #41564:
URL: https://github.com/apache/arrow/pull/41564#discussion_r1601154419
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1597,6 +1632,21 @@ class ObjectOutputStream final : public io::OutputStream {
}
upload_id_ = outcome.GetResult().GetUploadId();
upload_state_ = std::make_shared<UploadState>();
Review Comment:
Would this create duplicate `upload_state_` if shifting from single-part req?
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1606,42 +1656,60 @@ class ObjectOutputStream final : public io::OutputStream {
return Status::OK();
}
- ARROW_ASSIGN_OR_RAISE(auto client_lock, holder_->Lock());
+ if (is_multipart_created_) {
+ ARROW_ASSIGN_OR_RAISE(auto client_lock, holder_->Lock());
- S3Model::AbortMultipartUploadRequest req;
- req.SetBucket(ToAwsString(path_.bucket));
- req.SetKey(ToAwsString(path_.key));
- req.SetUploadId(upload_id_);
+ S3Model::AbortMultipartUploadRequest req;
+ req.SetBucket(ToAwsString(path_.bucket));
+ req.SetKey(ToAwsString(path_.key));
+ req.SetUploadId(upload_id_);
- auto outcome = client_lock.Move()->AbortMultipartUpload(req);
- if (!outcome.IsSuccess()) {
- return ErrorToStatus(
- std::forward_as_tuple("When aborting multiple part upload for key '", path_.key,
- "' in bucket '", path_.bucket, "': "),
- "AbortMultipartUpload", outcome.GetError());
+ auto outcome = client_lock.Move()->AbortMultipartUpload(req);
+ if (!outcome.IsSuccess()) {
+ return ErrorToStatus(
+ std::forward_as_tuple("When aborting multiple part upload for key '",
+ path_.key, "' in bucket '", path_.bucket, "': "),
+ "AbortMultipartUpload", outcome.GetError());
+ }
}
+
current_part_.reset();
holder_ = nullptr;
closed_ = true;
+
return Status::OK();
}
// OutputStream interface
+ bool ShouldBeMultipartUpload() { return pos_ > kMultiPartUploadThresholdSize; }
+
+ bool IsMultipartUpload() { return ShouldBeMultipartUpload() || is_multipart_created_; }
Review Comment:
1. Why two functions are added here?
2. Should they be const fn here?
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1606,42 +1656,60 @@ class ObjectOutputStream final : public io::OutputStream {
return Status::OK();
}
- ARROW_ASSIGN_OR_RAISE(auto client_lock, holder_->Lock());
+ if (is_multipart_created_) {
+ ARROW_ASSIGN_OR_RAISE(auto client_lock, holder_->Lock());
- S3Model::AbortMultipartUploadRequest req;
- req.SetBucket(ToAwsString(path_.bucket));
- req.SetKey(ToAwsString(path_.key));
- req.SetUploadId(upload_id_);
+ S3Model::AbortMultipartUploadRequest req;
+ req.SetBucket(ToAwsString(path_.bucket));
+ req.SetKey(ToAwsString(path_.key));
+ req.SetUploadId(upload_id_);
- auto outcome = client_lock.Move()->AbortMultipartUpload(req);
- if (!outcome.IsSuccess()) {
- return ErrorToStatus(
- std::forward_as_tuple("When aborting multiple part upload for key '", path_.key,
- "' in bucket '", path_.bucket, "': "),
- "AbortMultipartUpload", outcome.GetError());
+ auto outcome = client_lock.Move()->AbortMultipartUpload(req);
+ if (!outcome.IsSuccess()) {
+ return ErrorToStatus(
+ std::forward_as_tuple("When aborting multiple part upload for key '",
+ path_.key, "' in bucket '", path_.bucket, "': "),
+ "AbortMultipartUpload", outcome.GetError());
+ }
}
+
current_part_.reset();
holder_ = nullptr;
closed_ = true;
+
return Status::OK();
}
// OutputStream interface
+ bool ShouldBeMultipartUpload() { return pos_ > kMultiPartUploadThresholdSize; }
+
+ bool IsMultipartUpload() { return ShouldBeMultipartUpload() || is_multipart_created_; }
+
Status EnsureReadyToFlushFromClose() {
- if (current_part_) {
- // Upload last part
- RETURN_NOT_OK(CommitCurrentPart());
- }
+ if (IsMultipartUpload()) {
Review Comment:
Can this actually happens when `part_size_` is large enough but not being uploaded?
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1560,34 +1582,47 @@ class ObjectOutputStream final : public io::OutputStream {
path_(path),
metadata_(metadata),
default_metadata_(options.default_metadata),
- background_writes_(options.background_writes) {}
+ background_writes_(options.background_writes),
+ sanitize_bucket_on_open_(options.sanitize_bucket_on_open) {}
~ObjectOutputStream() override {
// For compliance with the rest of the IO stack, Close rather than Abort,
// even though it may be more expensive.
io::internal::CloseFromDestructor(this);
}
- Status Init() {
- ARROW_ASSIGN_OR_RAISE(auto client_lock, holder_->Lock());
+ template <typename ObjectRequest>
+ Status SetMetadataInRequest(ObjectRequest* request) {
+ std::shared_ptr<const KeyValueMetadata> metadata;
- // Initiate the multi-part upload
- S3Model::CreateMultipartUploadRequest req;
- req.SetBucket(ToAwsString(path_.bucket));
- req.SetKey(ToAwsString(path_.key));
if (metadata_ && metadata_->size() != 0) {
- RETURN_NOT_OK(SetObjectMetadata(metadata_, &req));
+ metadata = metadata_;
} else if (default_metadata_ && default_metadata_->size() != 0) {
- RETURN_NOT_OK(SetObjectMetadata(default_metadata_, &req));
+ metadata = default_metadata_;
}
- // If we do not set anything then the SDK will default to application/xml
- // which confuses some tools (https://github.com/apache/arrow/issues/11934)
- // So we instead default to application/octet-stream which is less misleading
- if (!req.ContentTypeHasBeenSet()) {
- req.SetContentType("application/octet-stream");
+ if (metadata == nullptr ||
+ !metadata->Contains(ObjectMetadataSetter<ObjectRequest>::CONTENT_TYPE_KEY)) {
+ // If we do not set anything then the SDK will default to application/xml
+ // which confuses some tools (https://github.com/apache/arrow/issues/11934)
+ // So we instead default to application/octet-stream which is less misleading
+ request->SetContentType("application/octet-stream");
+ } else {
+ RETURN_NOT_OK(SetObjectMetadata(metadata, request));
}
+ return Status::OK();
+ }
+
+ Status CreateMultipartUpload() {
+ ARROW_ASSIGN_OR_RAISE(auto client_lock, holder_->Lock());
Review Comment:
Can we assure that this is not being created?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "pitrou (via GitHub)" <gi...@apache.org>.
pitrou commented on code in PR #41564:
URL: https://github.com/apache/arrow/pull/41564#discussion_r1603510809
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1768,40 +1846,77 @@ class ObjectOutputStream final : public io::OutputStream {
}
// Wait for background writes to finish
std::unique_lock<std::mutex> lock(upload_state_->mutex);
- return upload_state_->pending_parts_completed;
+ return upload_state_->pending_uploads_completed;
}
// Upload-related helpers
Status CommitCurrentPart() {
+ if (!is_multipart_created_) {
+ RETURN_NOT_OK(CreateMultipartUpload());
+ }
+
ARROW_ASSIGN_OR_RAISE(auto buf, current_part_->Finish());
current_part_.reset();
current_part_size_ = 0;
return UploadPart(buf);
}
- Status UploadPart(std::shared_ptr<Buffer> buffer) {
- return UploadPart(buffer->data(), buffer->size(), buffer);
+ Status UploadUsingSingleRequest() {
+ std::shared_ptr<Buffer> buf;
+ if (current_part_ == nullptr) {
+ // In case the stream is closed directly after it has been opened without writing
+ // anything, we'll have to create an empty buffer.
+ buf = Buffer::FromVector<uint8_t>({});
Review Comment:
Nit, but `std::make_shared<Buffer>("", 0)` looks simpler to me.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "orf (via GitHub)" <gi...@apache.org>.
orf commented on PR #41564:
URL: https://github.com/apache/arrow/pull/41564#issuecomment-2108073297
I think it’s worth rebasing to see?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "pitrou (via GitHub)" <gi...@apache.org>.
pitrou commented on PR #41564:
URL: https://github.com/apache/arrow/pull/41564#issuecomment-2110240610
> That is true. I guess the question is if `arrow`'s OutputStream API makes an explicit _guarantee_ that `Open` should throw if the target does not exist. My guess would be that you shouldn't built code upon this assumption if it isn't explicitly stated in `arrow`'s API/docs (which it is not), but of course real-world usage deviates from that (Hyrum's Law).
The API docs generally do not go into that level of detail. However, it is a general assumption that a filesystem "looks like" a local filesystem API-wise.
It is also much more convenient to get an error early, than after you have already "written" multiple megabytes of data to the file.
A compromise would be to add a dedicated option in `S3Options`, but of course the optimization would only benefit those users that enable the option.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "orf (via GitHub)" <gi...@apache.org>.
orf commented on PR #41564:
URL: https://github.com/apache/arrow/pull/41564#issuecomment-2110822421
Just to note, issuing `HeadBucket` doesn't guarantee that a write will succeed - there isn't really a good way to check without actually writing.
I think an optimization flag is appropriate as the behaviour is technically changing. Does it make sense to make the flag a somewhat generic one, rather than specific to this case?
There are a [few other optimizations](https://github.com/apache/arrow/issues/40589) that might also fall into the "more performant, slightly different semantics" category. If I was to contribute a PR to improve one of the linked areas, would we want to add a new specific flag for this case?
The upside would be that it becomes more configurable, whereas the downside is that the testing and support matrix explodes. Perhaps it's better to just have a single `optimized=True` flag, vs reciving issues when specifically `optimize_put_object=True, optimize_delete_dir=False, optimize_move=True, optimize_delete=False, optimize_foobar=True` are set?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #41564:
URL: https://github.com/apache/arrow/pull/41564#discussion_r1601405451
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1606,42 +1656,60 @@ class ObjectOutputStream final : public io::OutputStream {
return Status::OK();
}
- ARROW_ASSIGN_OR_RAISE(auto client_lock, holder_->Lock());
+ if (is_multipart_created_) {
+ ARROW_ASSIGN_OR_RAISE(auto client_lock, holder_->Lock());
- S3Model::AbortMultipartUploadRequest req;
- req.SetBucket(ToAwsString(path_.bucket));
- req.SetKey(ToAwsString(path_.key));
- req.SetUploadId(upload_id_);
+ S3Model::AbortMultipartUploadRequest req;
+ req.SetBucket(ToAwsString(path_.bucket));
+ req.SetKey(ToAwsString(path_.key));
+ req.SetUploadId(upload_id_);
- auto outcome = client_lock.Move()->AbortMultipartUpload(req);
- if (!outcome.IsSuccess()) {
- return ErrorToStatus(
- std::forward_as_tuple("When aborting multiple part upload for key '", path_.key,
- "' in bucket '", path_.bucket, "': "),
- "AbortMultipartUpload", outcome.GetError());
+ auto outcome = client_lock.Move()->AbortMultipartUpload(req);
+ if (!outcome.IsSuccess()) {
+ return ErrorToStatus(
+ std::forward_as_tuple("When aborting multiple part upload for key '",
+ path_.key, "' in bucket '", path_.bucket, "': "),
+ "AbortMultipartUpload", outcome.GetError());
+ }
}
+
current_part_.reset();
holder_ = nullptr;
closed_ = true;
+
return Status::OK();
}
// OutputStream interface
+ bool ShouldBeMultipartUpload() { return pos_ > kMultiPartUploadThresholdSize; }
+
+ bool IsMultipartUpload() { return ShouldBeMultipartUpload() || is_multipart_created_; }
+
Status EnsureReadyToFlushFromClose() {
- if (current_part_) {
- // Upload last part
- RETURN_NOT_OK(CommitCurrentPart());
- }
+ if (IsMultipartUpload()) {
Review Comment:
Just to make it clear, since if `part_size_` is larger than size-limit, it should be switch to multipart during "doWrite"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "OliLay (via GitHub)" <gi...@apache.org>.
OliLay commented on PR #41564:
URL: https://github.com/apache/arrow/pull/41564#issuecomment-2112177646
> A minor question: would single-part only send during closing?
Yes, we only issue the `PutObject` request if we close the stream.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "OliLay (via GitHub)" <gi...@apache.org>.
OliLay commented on PR #41564:
URL: https://github.com/apache/arrow/pull/41564#issuecomment-2111793028
I added a `sanitize_bucket_on_open_` flag to the `S3Options` and also instantiated tests with this flag enabled.
I guess the Python bindings can be tackled in a separate PR, right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "conbench-apache-arrow[bot] (via GitHub)" <gi...@apache.org>.
conbench-apache-arrow[bot] commented on PR #41564:
URL: https://github.com/apache/arrow/pull/41564#issuecomment-2242820711
After merging your PR, Conbench analyzed the 0 benchmarking runs that have been run so far on merge-commit 9e6acbe08a0ff5b569763d377aecc824362dd593.
None of the specified runs were found on the Conbench server.
The [full Conbench report](https://github.com/apache/arrow/runs/27747153355) has more details.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "pitrou (via GitHub)" <gi...@apache.org>.
pitrou commented on code in PR #41564:
URL: https://github.com/apache/arrow/pull/41564#discussion_r1679534888
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1583,6 +1595,14 @@ class ObjectInputFile final : public io::RandomAccessFile {
// (for rational, see: https://github.com/apache/arrow/issues/34363)
static constexpr int64_t kPartUploadSize = 10 * 1024 * 1024;
+// Above this threshold, use a multi-part upload instead of a single request upload. Only
+// relevant if early sanitization of writing to the bucket is activated.
Review Comment:
Isn't it the reverse? For example:
```suggestion
// Above this threshold, use a multi-part upload instead of a single request upload.
// Only relevant if early sanitization of writing to the bucket is disabled
// (see `allow_delayed_open`).
```
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1880,9 +1991,117 @@ class ObjectOutputStream final : public io::OutputStream {
return Status::OK();
}
- static void HandleUploadOutcome(const std::shared_ptr<UploadState>& state,
- int part_number, const S3Model::UploadPartRequest& req,
- const Result<S3Model::UploadPartOutcome>& result) {
+ static Status UploadUsingSingleRequestError(
+ const Aws::S3::Model::PutObjectRequest& request,
+ const Aws::S3::Model::PutObjectOutcome& outcome) {
+ return ErrorToStatus(
+ std::forward_as_tuple("When uploading object with key '", request.GetKey(),
+ "' in bucket '", request.GetBucket(), "': "),
+ "PutObject", outcome.GetError());
+ }
+
+ Status UploadUsingSingleRequest(std::shared_ptr<Buffer> buffer) {
+ return UploadUsingSingleRequest(buffer->data(), buffer->size(), buffer);
+ }
+
+ Status UploadUsingSingleRequest(const void* data, int64_t nbytes,
+ std::shared_ptr<Buffer> owned_buffer = nullptr) {
+ auto sync_result_callback = [](const Aws::S3::Model::PutObjectRequest& request,
+ std::shared_ptr<UploadState> state,
+ int32_t part_number,
+ Aws::S3::Model::PutObjectOutcome outcome) {
+ if (!outcome.IsSuccess()) {
+ return UploadUsingSingleRequestError(request, outcome);
+ }
+ return Status::OK();
+ };
+
+ auto async_result_callback = [](const Aws::S3::Model::PutObjectRequest& request,
+ std::shared_ptr<UploadState> state,
+ int32_t part_number,
+ Aws::S3::Model::PutObjectOutcome outcome) {
+ HandleUploadUsingSingleRequestOutcome(state, request, outcome.GetResult());
+ return Status::OK();
+ };
+
+ Aws::S3::Model::PutObjectRequest req{};
+ RETURN_NOT_OK(SetMetadataInRequest(&req));
+
+ return Upload<Aws::S3::Model::PutObjectRequest, Aws::S3::Model::PutObjectOutcome>(
+ std::move(req), std::move(sync_result_callback), std::move(async_result_callback),
+ data, nbytes, std::move(owned_buffer));
+ }
+
+ Status UploadPart(std::shared_ptr<Buffer> buffer) {
+ return UploadPart(buffer->data(), buffer->size(), buffer);
+ }
+
+ static Status UploadPartError(const Aws::S3::Model::UploadPartRequest& request,
+ const Aws::S3::Model::UploadPartOutcome& outcome) {
+ return ErrorToStatus(
+ std::forward_as_tuple("When uploading part for key '", request.GetKey(),
+ "' in bucket '", request.GetBucket(), "': "),
+ "UploadPart", outcome.GetError());
+ }
+
+ Status UploadPart(const void* data, int64_t nbytes,
+ std::shared_ptr<Buffer> owned_buffer = nullptr) {
+ if (!IsMultipartCreated()) {
+ RETURN_NOT_OK(CreateMultipartUpload());
+ }
+
+ Aws::S3::Model::UploadPartRequest req{};
+ req.SetPartNumber(part_number_);
+ req.SetUploadId(multipart_upload_id_);
+
+ auto sync_result_callback = [](const Aws::S3::Model::UploadPartRequest& request,
+ std::shared_ptr<UploadState> state,
+ int32_t part_number,
+ Aws::S3::Model::UploadPartOutcome outcome) {
+ if (!outcome.IsSuccess()) {
+ return UploadPartError(request, outcome);
+ } else {
+ AddCompletedPart(state, part_number, outcome.GetResult());
+ }
+
+ return Status::OK();
+ };
+
+ auto async_result_callback = [](const Aws::S3::Model::UploadPartRequest& request,
+ std::shared_ptr<UploadState> state,
+ int32_t part_number,
+ Aws::S3::Model::UploadPartOutcome outcome) {
+ HandleUploadPartOutcome(state, part_number, request, outcome.GetResult());
+ return Status::OK();
+ };
+
+ return Upload<Aws::S3::Model::UploadPartRequest, Aws::S3::Model::UploadPartOutcome>(
+ std::move(req), std::move(sync_result_callback), std::move(async_result_callback),
+ data, nbytes, std::move(owned_buffer));
+ }
+
+ static void HandleUploadUsingSingleRequestOutcome(
+ const std::shared_ptr<UploadState>& state, const S3Model::PutObjectRequest& req,
+ const Result<S3Model::PutObjectOutcome>& result) {
+ std::unique_lock<std::mutex> lock(state->mutex);
+ if (!result.ok()) {
+ state->status &= result.status();
+ } else {
+ const auto& outcome = *result;
+ if (!outcome.IsSuccess()) {
+ state->status &= UploadUsingSingleRequestError(req, outcome);
+ }
+ }
+ // GH-41862: avoid potential deadlock if the Future's callback is called
+ // with the mutex taken.
+ lock.unlock();
+ state->pending_uploads_completed.MarkFinished(state->status);
Review Comment:
I suggest to use the same idiom as below, to avoid a race condition accessing the future:
```suggestion
auto fut = state->pending_uploads_completed;
lock.unlock();
fut.MarkFinished(state->status);
```
##########
cpp/src/arrow/filesystem/s3fs_test.cc:
##########
@@ -478,6 +463,74 @@ class TestS3FS : public S3TestMixin {
S3TestMixin::TearDown();
}
+ Status PopulateTestBucket() {
+ Aws::S3::Model::PutObjectRequest req;
+ req.SetBucket(ToAwsString("bucket"));
+ req.SetKey(ToAwsString("emptydir/"));
+ req.SetBody(std::make_shared<std::stringstream>(""));
+ RETURN_NOT_OK(OutcomeToStatus("PutObject", client_->PutObject(req)));
+ // NOTE: no need to create intermediate "directories" somedir/ and
+ // somedir/subdir/
+ req.SetKey(ToAwsString("somedir/subdir/subfile"));
+ req.SetBody(std::make_shared<std::stringstream>("sub data"));
+ RETURN_NOT_OK(OutcomeToStatus("PutObject", client_->PutObject(req)));
+ req.SetKey(ToAwsString("somefile"));
+ req.SetBody(std::make_shared<std::stringstream>("some data"));
+ req.SetContentType("x-arrow/test");
+ RETURN_NOT_OK(OutcomeToStatus("PutObject", client_->PutObject(req)));
+ req.SetKey(ToAwsString("otherdir/1/2/3/otherfile"));
+ req.SetBody(std::make_shared<std::stringstream>("other data"));
+ RETURN_NOT_OK(OutcomeToStatus("PutObject", client_->PutObject(req)));
+
+ return Status::OK();
+ }
+
+ Status RestoreTestBucket() {
+ // First empty the test bucket, and then re-upload initial test files.
+
+ Aws::Vector<Aws::S3::Model::Object> all_objects;
+ {
+ // Mostly taken from
+ // https://github.com/awsdocs/aws-doc-sdk-examples/blob/main/cpp/example_code/s3/list_objects.cpp
+ Aws::S3::Model::ListObjectsV2Request req;
+ req.WithBucket(Aws::String{"bucket"});
+
+ Aws::String continuation_token;
+ do {
+ if (!continuation_token.empty()) {
+ req.SetContinuationToken(continuation_token);
+ }
+
+ auto outcome = client_->ListObjectsV2(req);
+
+ if (!outcome.IsSuccess()) {
+ return OutcomeToStatus("ListObjectsV2", outcome);
+ } else {
+ Aws::Vector<Aws::S3::Model::Object> objects = outcome.GetResult().GetContents();
+ all_objects.insert(all_objects.end(), objects.begin(), objects.end());
+ continuation_token = outcome.GetResult().GetNextContinuationToken();
+ }
+ } while (!continuation_token.empty());
+ }
+
+ {
+ Aws::S3::Model::DeleteObjectsRequest req;
+
+ Aws::S3::Model::Delete delete_object;
+ for (const auto& object : all_objects) {
+ delete_object.AddObjects(
+ Aws::S3::Model::ObjectIdentifier().WithKey(object.GetKey()));
+ }
+
+ req.SetDelete(delete_object);
Review Comment:
Probably unimportant, but we can avoid a copy of the objects vector here.
```suggestion
req.SetDelete(std::move(delete_object));
```
##########
cpp/src/arrow/filesystem/s3fs_test.cc:
##########
@@ -23,6 +23,8 @@
#include <utility>
#include <vector>
+#include <aws/s3/model/DeleteObjectsRequest.h>
+#include <aws/s3/model/ListObjectsV2Request.h>
Review Comment:
Can you move those below with other AWS includes?
##########
cpp/src/arrow/filesystem/s3fs_test.cc:
##########
@@ -478,6 +463,74 @@ class TestS3FS : public S3TestMixin {
S3TestMixin::TearDown();
}
+ Status PopulateTestBucket() {
+ Aws::S3::Model::PutObjectRequest req;
+ req.SetBucket(ToAwsString("bucket"));
+ req.SetKey(ToAwsString("emptydir/"));
+ req.SetBody(std::make_shared<std::stringstream>(""));
+ RETURN_NOT_OK(OutcomeToStatus("PutObject", client_->PutObject(req)));
+ // NOTE: no need to create intermediate "directories" somedir/ and
+ // somedir/subdir/
+ req.SetKey(ToAwsString("somedir/subdir/subfile"));
+ req.SetBody(std::make_shared<std::stringstream>("sub data"));
+ RETURN_NOT_OK(OutcomeToStatus("PutObject", client_->PutObject(req)));
+ req.SetKey(ToAwsString("somefile"));
+ req.SetBody(std::make_shared<std::stringstream>("some data"));
+ req.SetContentType("x-arrow/test");
+ RETURN_NOT_OK(OutcomeToStatus("PutObject", client_->PutObject(req)));
+ req.SetKey(ToAwsString("otherdir/1/2/3/otherfile"));
+ req.SetBody(std::make_shared<std::stringstream>("other data"));
+ RETURN_NOT_OK(OutcomeToStatus("PutObject", client_->PutObject(req)));
+
+ return Status::OK();
+ }
+
+ Status RestoreTestBucket() {
+ // First empty the test bucket, and then re-upload initial test files.
+
+ Aws::Vector<Aws::S3::Model::Object> all_objects;
+ {
+ // Mostly taken from
+ // https://github.com/awsdocs/aws-doc-sdk-examples/blob/main/cpp/example_code/s3/list_objects.cpp
+ Aws::S3::Model::ListObjectsV2Request req;
+ req.WithBucket(Aws::String{"bucket"});
+
+ Aws::String continuation_token;
+ do {
+ if (!continuation_token.empty()) {
+ req.SetContinuationToken(continuation_token);
+ }
+
+ auto outcome = client_->ListObjectsV2(req);
+
+ if (!outcome.IsSuccess()) {
+ return OutcomeToStatus("ListObjectsV2", outcome);
+ } else {
+ Aws::Vector<Aws::S3::Model::Object> objects = outcome.GetResult().GetContents();
+ all_objects.insert(all_objects.end(), objects.begin(), objects.end());
Review Comment:
Note that you could incrementally populate the `delete_object` here instead of going through an intermediate vector `all_objects`.
##########
cpp/src/arrow/filesystem/s3fs_test.cc:
##########
@@ -478,6 +463,74 @@ class TestS3FS : public S3TestMixin {
S3TestMixin::TearDown();
}
+ Status PopulateTestBucket() {
+ Aws::S3::Model::PutObjectRequest req;
+ req.SetBucket(ToAwsString("bucket"));
+ req.SetKey(ToAwsString("emptydir/"));
+ req.SetBody(std::make_shared<std::stringstream>(""));
+ RETURN_NOT_OK(OutcomeToStatus("PutObject", client_->PutObject(req)));
+ // NOTE: no need to create intermediate "directories" somedir/ and
+ // somedir/subdir/
+ req.SetKey(ToAwsString("somedir/subdir/subfile"));
+ req.SetBody(std::make_shared<std::stringstream>("sub data"));
+ RETURN_NOT_OK(OutcomeToStatus("PutObject", client_->PutObject(req)));
+ req.SetKey(ToAwsString("somefile"));
+ req.SetBody(std::make_shared<std::stringstream>("some data"));
+ req.SetContentType("x-arrow/test");
+ RETURN_NOT_OK(OutcomeToStatus("PutObject", client_->PutObject(req)));
+ req.SetKey(ToAwsString("otherdir/1/2/3/otherfile"));
+ req.SetBody(std::make_shared<std::stringstream>("other data"));
+ RETURN_NOT_OK(OutcomeToStatus("PutObject", client_->PutObject(req)));
+
+ return Status::OK();
+ }
+
+ Status RestoreTestBucket() {
+ // First empty the test bucket, and then re-upload initial test files.
+
+ Aws::Vector<Aws::S3::Model::Object> all_objects;
+ {
+ // Mostly taken from
+ // https://github.com/awsdocs/aws-doc-sdk-examples/blob/main/cpp/example_code/s3/list_objects.cpp
+ Aws::S3::Model::ListObjectsV2Request req;
+ req.WithBucket(Aws::String{"bucket"});
Review Comment:
Nit: no need to use the reference-returning methods here
```suggestion
req.SetBucket(Aws::String{"bucket"});
```
##########
cpp/src/arrow/filesystem/s3fs_test.cc:
##########
@@ -584,6 +639,8 @@ class TestS3FS : public S3TestMixin {
ASSERT_OK(stream->Close());
ASSERT_TRUE(weak_fs.expired());
AssertObjectContents(client_.get(), "bucket", "newfile99", "some other data");
+
+ ASSERT_OK(RestoreTestBucket());
Review Comment:
Can we move the `RestoreTestBucket` calls into the actual test loops?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "pitrou (via GitHub)" <gi...@apache.org>.
pitrou commented on PR #41564:
URL: https://github.com/apache/arrow/pull/41564#issuecomment-2110114654
> Previously, we would fail when opening the output stream if the bucket doesn't exist. We inferred that by sending the `CreateMultipartUpload` request, which we now do not send anymore upon opening the stream. We now rather fail at closing, or at writing (when >5MB have accumulated).
Hmm, I'm not sure that is ok. Usually, when opening a file for writing, you expect the initial open to fail if the path cannot be written to. I have no idea how much code relies on that, but that's a common expectation due to how filesystems usually work (e.g. when accessing local storage).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "OliLay (via GitHub)" <gi...@apache.org>.
OliLay commented on code in PR #41564:
URL: https://github.com/apache/arrow/pull/41564#discussion_r1604483833
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1734,7 +1812,7 @@ class ObjectOutputStream final : public io::OutputStream {
return Status::OK();
}
- // Upload current buffer
+ // Upload current buffer if we are above threshold for multi-part upload
Review Comment:
You're right, I just wanted to make the point that we won't call this when we haven't accumulated enough data. I've clarified the comment. [7748824](https://github.com/apache/arrow/pull/41564/commits/77488243b71be6688e9131efac03e9b98d14b1bd)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "pitrou (via GitHub)" <gi...@apache.org>.
pitrou commented on code in PR #41564:
URL: https://github.com/apache/arrow/pull/41564#discussion_r1603499276
##########
cpp/src/arrow/filesystem/s3fs_test.cc:
##########
@@ -1197,6 +1199,19 @@ TEST_F(TestS3FS, OpenOutputStreamSyncWrites) {
TestOpenOutputStream();
}
+TEST_F(TestS3FS, OpenOutputStreamNoBucketSanitizationSyncWrites) {
Review Comment:
Instead of adding more test methods for every combinatorial expansion, perhaps we should instead use a `for` loop on the various tested parameter values?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "pitrou (via GitHub)" <gi...@apache.org>.
pitrou commented on code in PR #41564:
URL: https://github.com/apache/arrow/pull/41564#discussion_r1603514407
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1768,40 +1846,77 @@ class ObjectOutputStream final : public io::OutputStream {
}
// Wait for background writes to finish
std::unique_lock<std::mutex> lock(upload_state_->mutex);
- return upload_state_->pending_parts_completed;
+ return upload_state_->pending_uploads_completed;
}
// Upload-related helpers
Status CommitCurrentPart() {
+ if (!is_multipart_created_) {
+ RETURN_NOT_OK(CreateMultipartUpload());
+ }
+
ARROW_ASSIGN_OR_RAISE(auto buf, current_part_->Finish());
current_part_.reset();
current_part_size_ = 0;
return UploadPart(buf);
}
- Status UploadPart(std::shared_ptr<Buffer> buffer) {
- return UploadPart(buffer->data(), buffer->size(), buffer);
+ Status UploadUsingSingleRequest() {
+ std::shared_ptr<Buffer> buf;
+ if (current_part_ == nullptr) {
+ // In case the stream is closed directly after it has been opened without writing
+ // anything, we'll have to create an empty buffer.
+ buf = Buffer::FromVector<uint8_t>({});
+ } else {
+ ARROW_ASSIGN_OR_RAISE(buf, current_part_->Finish());
+ }
+
+ current_part_.reset();
+ current_part_size_ = 0;
+ return UploadUsingSingleRequest(buf);
}
- Status UploadPart(const void* data, int64_t nbytes,
- std::shared_ptr<Buffer> owned_buffer = nullptr) {
- S3Model::UploadPartRequest req;
+ template <typename RequestType, typename OutcomeType>
+ using UploadResultCallbackFunction =
+ std::function<Status(const RequestType& request, std::shared_ptr<UploadState>,
+ int32_t, OutcomeType outcome)>;
+
+ template <typename RequestType, typename OutcomeType>
+ static Result<OutcomeType> TriggerUploadRequest(
+ const RequestType& request, const std::shared_ptr<S3ClientHolder>& holder);
Review Comment:
I don't think this declaration is actually necessary? i.e. `TriggerUploadRequest` doesn't need to be a template method, it can be a regular overloaded method.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "OliLay (via GitHub)" <gi...@apache.org>.
OliLay commented on PR #41564:
URL: https://github.com/apache/arrow/pull/41564#issuecomment-2107526185
From the log output, it seems like the failing CI jobs are not related to this change. Correct me if I am wrong though. Should I rebase (in case the flaky tests are already fixed on main)?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "pitrou (via GitHub)" <gi...@apache.org>.
pitrou commented on PR #41564:
URL: https://github.com/apache/arrow/pull/41564#issuecomment-2110168942
My point is that _if the path cannot be written to_, the error happens when opening the file, not later on.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #41564:
URL: https://github.com/apache/arrow/pull/41564#discussion_r1599921602
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1545,6 +1559,13 @@ class ObjectInputFile final : public io::RandomAccessFile {
// (for rational, see: https://github.com/apache/arrow/issues/34363)
static constexpr int64_t kPartUploadSize = 10 * 1024 * 1024;
+// Above this threshold, use a multi-part upload instead of a single request upload.
+static constexpr int64_t kMultiPartUploadThresholdSize = 5 * 1024 * 1024;
Review Comment:
Should we config hard limit like this, or Switch to multipart when there're multi parts? (size > 10MiB during append)
##########
cpp/src/arrow/filesystem/s3fs_test.cc:
##########
@@ -521,9 +521,6 @@ class TestS3FS : public S3TestMixin {
void TestOpenOutputStream() {
std::shared_ptr<io::OutputStream> stream;
- // Nonexistent
- ASSERT_RAISES(IOError, fs_->OpenOutputStream("nonexistent-bucket/somefile"));
Review Comment:
Why this is being removed?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "pitrou (via GitHub)" <gi...@apache.org>.
pitrou commented on code in PR #41564:
URL: https://github.com/apache/arrow/pull/41564#discussion_r1603508195
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1734,7 +1812,7 @@ class ObjectOutputStream final : public io::OutputStream {
return Status::OK();
}
- // Upload current buffer
+ // Upload current buffer if we are above threshold for multi-part upload
Review Comment:
I mean, `CommitCurrentPart` starts by calling `CreateMultipartUpload`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "OliLay (via GitHub)" <gi...@apache.org>.
OliLay commented on code in PR #41564:
URL: https://github.com/apache/arrow/pull/41564#discussion_r1601474966
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1768,40 +1846,77 @@ class ObjectOutputStream final : public io::OutputStream {
}
// Wait for background writes to finish
std::unique_lock<std::mutex> lock(upload_state_->mutex);
- return upload_state_->pending_parts_completed;
+ return upload_state_->pending_uploads_completed;
}
// Upload-related helpers
Status CommitCurrentPart() {
+ if (!is_multipart_created_) {
+ RETURN_NOT_OK(CreateMultipartUpload());
+ }
+
ARROW_ASSIGN_OR_RAISE(auto buf, current_part_->Finish());
current_part_.reset();
current_part_size_ = 0;
return UploadPart(buf);
}
- Status UploadPart(std::shared_ptr<Buffer> buffer) {
- return UploadPart(buffer->data(), buffer->size(), buffer);
+ Status UploadUsingSingleRequest() {
+ std::shared_ptr<Buffer> buf;
Review Comment:
We would need another variable for that to save that we issued the single part upload. Considering this, not sure if is worth adding that, considering that we only issue a single part upload in `Close()` anyways. (+ tests would complain that the object already exists if we would trigger both requests, independent in which order I would assume)
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1838,9 +1952,114 @@ class ObjectOutputStream final : public io::OutputStream {
return Status::OK();
}
- static void HandleUploadOutcome(const std::shared_ptr<UploadState>& state,
- int part_number, const S3Model::UploadPartRequest& req,
- const Result<S3Model::UploadPartOutcome>& result) {
+ static Status UploadError(const Aws::S3::Model::PutObjectRequest& request,
+ const Aws::S3::Model::PutObjectOutcome& outcome) {
+ return ErrorToStatus(
+ std::forward_as_tuple("When uploading object with key '", request.GetKey(),
+ "' in bucket '", request.GetBucket(), "': "),
+ "PutObject", outcome.GetError());
+ }
+
+ Status UploadUsingSingleRequest(std::shared_ptr<Buffer> buffer) {
+ return UploadUsingSingleRequest(buffer->data(), buffer->size(), buffer);
+ }
+
+ Status UploadUsingSingleRequest(const void* data, int64_t nbytes,
+ std::shared_ptr<Buffer> owned_buffer = nullptr) {
+ auto sync_result_callback = [](const Aws::S3::Model::PutObjectRequest& request,
+ std::shared_ptr<UploadState> state,
+ int32_t part_number,
+ Aws::S3::Model::PutObjectOutcome outcome) {
+ if (!outcome.IsSuccess()) {
+ return UploadError(request, outcome);
+ }
+ return Status::OK();
+ };
+
+ auto async_result_callback = [](const Aws::S3::Model::PutObjectRequest& request,
Review Comment:
Yes, previously in the sync case we also directly called `AddCompletedPart` which just used the `state` to fiddle around with its `completed_parts` member. For a single upload request, I think we can omit this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "OliLay (via GitHub)" <gi...@apache.org>.
OliLay commented on code in PR #41564:
URL: https://github.com/apache/arrow/pull/41564#discussion_r1601415537
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1606,42 +1656,60 @@ class ObjectOutputStream final : public io::OutputStream {
return Status::OK();
}
- ARROW_ASSIGN_OR_RAISE(auto client_lock, holder_->Lock());
+ if (is_multipart_created_) {
+ ARROW_ASSIGN_OR_RAISE(auto client_lock, holder_->Lock());
- S3Model::AbortMultipartUploadRequest req;
- req.SetBucket(ToAwsString(path_.bucket));
- req.SetKey(ToAwsString(path_.key));
- req.SetUploadId(upload_id_);
+ S3Model::AbortMultipartUploadRequest req;
+ req.SetBucket(ToAwsString(path_.bucket));
+ req.SetKey(ToAwsString(path_.key));
+ req.SetUploadId(upload_id_);
- auto outcome = client_lock.Move()->AbortMultipartUpload(req);
- if (!outcome.IsSuccess()) {
- return ErrorToStatus(
- std::forward_as_tuple("When aborting multiple part upload for key '", path_.key,
- "' in bucket '", path_.bucket, "': "),
- "AbortMultipartUpload", outcome.GetError());
+ auto outcome = client_lock.Move()->AbortMultipartUpload(req);
+ if (!outcome.IsSuccess()) {
+ return ErrorToStatus(
+ std::forward_as_tuple("When aborting multiple part upload for key '",
+ path_.key, "' in bucket '", path_.bucket, "': "),
+ "AbortMultipartUpload", outcome.GetError());
+ }
}
+
current_part_.reset();
holder_ = nullptr;
closed_ = true;
+
return Status::OK();
}
// OutputStream interface
+ bool ShouldBeMultipartUpload() { return pos_ > kMultiPartUploadThresholdSize; }
+
+ bool IsMultipartUpload() { return ShouldBeMultipartUpload() || is_multipart_created_; }
+
Status EnsureReadyToFlushFromClose() {
- if (current_part_) {
- // Upload last part
- RETURN_NOT_OK(CommitCurrentPart());
- }
+ if (IsMultipartUpload()) {
Review Comment:
You mean if `current_part_size_` > `kPartUploadSize` can be true at the time we close the stream?
I think that is not possible (and also wasn't possible with the old impl). The invariant when `Close()` is called is that `current_part_size_` < `kPartUploadSize` holds, because if we write to the stream and `current_part_size_` >= `kPartUploadSize` holds, we always directly upload a part in `DoWrite`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #41564:
URL: https://github.com/apache/arrow/pull/41564#issuecomment-2242599198
Revision: 670232b77c711b0e124e8d2ad02c2c7f62d16a80
Submitted crossbow builds: [ursacomputing/crossbow @ actions-86d9d85a0e](https://github.com/ursacomputing/crossbow/branches/all?query=actions-86d9d85a0e)
|Task|Status|
|----|------|
|example-python-minimal-build-fedora-conda|[](https://github.com/ursacomputing/crossbow/actions/runs/10038909780/job/27741807276)|
|example-python-minimal-build-ubuntu-venv|[](https://github.com/ursacomputing/crossbow/actions/runs/10038909302/job/27741805389)|
|test-alpine-linux-cpp|[](https://github.com/ursacomputing/crossbow/actions/runs/10038909223/job/27741804660)|
|test-build-cpp-fuzz|[](https://github.com/ursacomputing/crossbow/actions/runs/10038909631/job/27741806790)|
|test-conda-cpp|[](https://github.com/ursacomputing/crossbow/actions/runs/10038909326/job/27741805428)|
|test-conda-cpp-valgrind|[](https://github.com/ursacomputing/crossbow/actions/runs/10038908985/job/27741803741)|
|test-conda-python-3.10|[](https://github.com/ursacomputing/crossbow/actions/runs/10038909489/job/27741806077)|
|test-conda-python-3.10-cython2|[](https://github.com/ursacomputing/crossbow/actions/runs/10038909293/job/27741805150)|
|test-conda-python-3.10-hdfs-2.9.2|[](https://github.com/ursacomputing/crossbow/actions/runs/10038909091/job/27741804162)|
|test-conda-python-3.10-hdfs-3.2.1|[](https://github.com/ursacomputing/crossbow/actions/runs/10038908720/job/27741802537)|
|test-conda-python-3.10-pandas-latest-numpy-1.26|[](https://github.com/ursacomputing/crossbow/actions/runs/10038909969/job/27741807690)|
|test-conda-python-3.10-pandas-latest-numpy-latest|[](https://github.com/ursacomputing/crossbow/actions/runs/10038908756/job/27741802579)|
|test-conda-python-3.10-pandas-nightly-numpy-nightly|[](https://github.com/ursacomputing/crossbow/actions/runs/10038909773/job/27741807168)|
|test-conda-python-3.10-substrait|[](https://github.com/ursacomputing/crossbow/actions/runs/10038909528/job/27741806273)|
|test-conda-python-3.11|[](https://github.com/ursacomputing/crossbow/actions/runs/10038908821/job/27741802884)|
|test-conda-python-3.11-dask-latest|[](https://github.com/ursacomputing/crossbow/actions/runs/10038909178/job/27741804281)|
|test-conda-python-3.11-dask-upstream_devel|[](https://github.com/ursacomputing/crossbow/actions/runs/10038908590/job/27741802271)|
|test-conda-python-3.11-hypothesis|[](https://github.com/ursacomputing/crossbow/actions/runs/10038910175/job/27741808279)|
|test-conda-python-3.11-pandas-upstream_devel-numpy-nightly|[](https://github.com/ursacomputing/crossbow/actions/runs/10038909130/job/27741804156)|
|test-conda-python-3.11-spark-master|[](https://github.com/ursacomputing/crossbow/actions/runs/10038909497/job/27741806041)|
|test-conda-python-3.12|[](https://github.com/ursacomputing/crossbow/actions/runs/10038908722/job/27741802571)|
|test-conda-python-3.8|[](https://github.com/ursacomputing/crossbow/actions/runs/10038909696/job/27741807020)|
|test-conda-python-3.8-pandas-1.0-numpy-1.19|[](https://github.com/ursacomputing/crossbow/actions/runs/10038909378/job/27741805500)|
|test-conda-python-3.9|[](https://github.com/ursacomputing/crossbow/actions/runs/10038908829/job/27741802985)|
|test-conda-python-3.9-pandas-latest-numpy-latest|[](https://github.com/ursacomputing/crossbow/actions/runs/10038909243/job/27741804790)|
|test-conda-python-emscripten|[](https://github.com/ursacomputing/crossbow/actions/runs/10038909440/job/27741805797)|
|test-cuda-cpp|[](https://github.com/ursacomputing/crossbow/actions/runs/10038909028/job/27741803574)|
|test-cuda-python|[](https://github.com/ursacomputing/crossbow/actions/runs/10038909247/job/27741804948)|
|test-debian-12-cpp-amd64|[](https://github.com/ursacomputing/crossbow/actions/runs/10038909327/job/27741805165)|
|test-debian-12-cpp-i386|[](https://github.com/ursacomputing/crossbow/actions/runs/10038908834/job/27741803063)|
|test-debian-12-python-3-amd64|[](https://github.com/ursacomputing/crossbow/actions/runs/10038909401/job/27741805490)|
|test-debian-12-python-3-i386|[](https://github.com/ursacomputing/crossbow/actions/runs/10038908521/job/27741801896)|
|test-fedora-39-cpp|[](https://github.com/ursacomputing/crossbow/actions/runs/10038909541/job/27741806128)|
|test-fedora-39-python-3|[](https://github.com/ursacomputing/crossbow/actions/runs/10038909233/job/27741804728)|
|test-ubuntu-20.04-cpp|[](https://github.com/ursacomputing/crossbow/actions/runs/10038908965/job/27741803458)|
|test-ubuntu-20.04-cpp-bundled|[](https://github.com/ursacomputing/crossbow/actions/runs/10038909190/job/27741804298)|
|test-ubuntu-20.04-cpp-minimal-with-formats|[](https://github.com/ursacomputing/crossbow/actions/runs/10038909231/job/27741804655)|
|test-ubuntu-20.04-cpp-thread-sanitizer|[](https://github.com/ursacomputing/crossbow/actions/runs/10038909077/job/27741804052)|
|test-ubuntu-20.04-python-3|[](https://github.com/ursacomputing/crossbow/actions/runs/10038908984/job/27741803728)|
|test-ubuntu-22.04-cpp|[](https://github.com/ursacomputing/crossbow/actions/runs/10038908492/job/27741801871)|
|test-ubuntu-22.04-cpp-20|[](https://github.com/ursacomputing/crossbow/actions/runs/10038908990/job/27741803660)|
|test-ubuntu-22.04-cpp-emscripten|[](https://github.com/ursacomputing/crossbow/actions/runs/10038908583/job/27741802279)|
|test-ubuntu-22.04-cpp-no-threading|[](https://github.com/ursacomputing/crossbow/actions/runs/10038908518/job/27741801872)|
|test-ubuntu-22.04-python-3|[](https://github.com/ursacomputing/crossbow/actions/runs/10038908504/job/27741801834)|
|test-ubuntu-24.04-cpp|[](https://github.com/ursacomputing/crossbow/actions/runs/10038908839/job/27741802963)|
|test-ubuntu-24.04-cpp-gcc-14|[](https://github.com/ursacomputing/crossbow/actions/runs/10038909748/job/27741807050)|
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on PR #41564:
URL: https://github.com/apache/arrow/pull/41564#issuecomment-2130097997
Sorry for delaying review! Would merge after other committers approve
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "pitrou (via GitHub)" <gi...@apache.org>.
pitrou commented on code in PR #41564:
URL: https://github.com/apache/arrow/pull/41564#discussion_r1603496580
##########
cpp/src/arrow/filesystem/s3fs.h:
##########
@@ -166,6 +166,13 @@ struct ARROW_EXPORT S3Options {
/// Whether to allow deletion of buckets
bool allow_bucket_deletion = false;
+ /// Whether to sanitize that a bucket can be written to when calling OpenOutputStream.
+ /// Disabling this will lead to less S3 requests issued and is therefore relevant
+ /// for latency-sensitive applications, at the cost of the OutputStream may throwing an
+ /// exception at a later stage (i.e. at writing or closing) if e.g. the bucket does not
+ /// exist.
+ bool sanitize_bucket_on_open = true;
Review Comment:
Perhaps we should make this more general, to open up other potential optimizations?
```c++
/// Whether to allow file-open methods to return before the actual open
///
/// Enabling this true may reduce the latency of `OpenInputStream`, `OpenOutpuStream`,
/// and similar methods, by reducing the number of roundtrips necessary. It may also
/// allow usage of more efficient S3 APIs for small files.
/// The downside is that failure conditions such as attempting to open a file in a
/// non-existing bucket will only be reported when actual I/O is done (at worse,
/// when attempting to close the file).
bool allow_delayed_open = false;
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #41564:
URL: https://github.com/apache/arrow/pull/41564#discussion_r1601412688
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1768,40 +1846,77 @@ class ObjectOutputStream final : public io::OutputStream {
}
// Wait for background writes to finish
std::unique_lock<std::mutex> lock(upload_state_->mutex);
- return upload_state_->pending_parts_completed;
+ return upload_state_->pending_uploads_completed;
}
// Upload-related helpers
Status CommitCurrentPart() {
+ if (!is_multipart_created_) {
+ RETURN_NOT_OK(CreateMultipartUpload());
+ }
+
ARROW_ASSIGN_OR_RAISE(auto buf, current_part_->Finish());
current_part_.reset();
current_part_size_ = 0;
return UploadPart(buf);
}
- Status UploadPart(std::shared_ptr<Buffer> buffer) {
- return UploadPart(buffer->data(), buffer->size(), buffer);
+ Status UploadUsingSingleRequest() {
+ std::shared_ptr<Buffer> buf;
Review Comment:
(Besides, could we debug checks that multi-part will not called after single part is updated?)
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1838,9 +1952,114 @@ class ObjectOutputStream final : public io::OutputStream {
return Status::OK();
}
- static void HandleUploadOutcome(const std::shared_ptr<UploadState>& state,
- int part_number, const S3Model::UploadPartRequest& req,
- const Result<S3Model::UploadPartOutcome>& result) {
+ static Status UploadError(const Aws::S3::Model::PutObjectRequest& request,
+ const Aws::S3::Model::PutObjectOutcome& outcome) {
+ return ErrorToStatus(
+ std::forward_as_tuple("When uploading object with key '", request.GetKey(),
+ "' in bucket '", request.GetBucket(), "': "),
+ "PutObject", outcome.GetError());
+ }
+
+ Status UploadUsingSingleRequest(std::shared_ptr<Buffer> buffer) {
+ return UploadUsingSingleRequest(buffer->data(), buffer->size(), buffer);
+ }
+
+ Status UploadUsingSingleRequest(const void* data, int64_t nbytes,
+ std::shared_ptr<Buffer> owned_buffer = nullptr) {
+ auto sync_result_callback = [](const Aws::S3::Model::PutObjectRequest& request,
+ std::shared_ptr<UploadState> state,
+ int32_t part_number,
+ Aws::S3::Model::PutObjectOutcome outcome) {
+ if (!outcome.IsSuccess()) {
+ return UploadError(request, outcome);
+ }
+ return Status::OK();
+ };
+
+ auto async_result_callback = [](const Aws::S3::Model::PutObjectRequest& request,
Review Comment:
So just to make it clear, the `async_result_callback` always get a `UploadState` and uses it. And the `sync_result_callback` never uses that?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "pitrou (via GitHub)" <gi...@apache.org>.
pitrou commented on code in PR #41564:
URL: https://github.com/apache/arrow/pull/41564#discussion_r1603516333
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1768,40 +1846,77 @@ class ObjectOutputStream final : public io::OutputStream {
}
// Wait for background writes to finish
std::unique_lock<std::mutex> lock(upload_state_->mutex);
- return upload_state_->pending_parts_completed;
+ return upload_state_->pending_uploads_completed;
}
// Upload-related helpers
Status CommitCurrentPart() {
+ if (!is_multipart_created_) {
+ RETURN_NOT_OK(CreateMultipartUpload());
+ }
+
ARROW_ASSIGN_OR_RAISE(auto buf, current_part_->Finish());
current_part_.reset();
current_part_size_ = 0;
return UploadPart(buf);
}
- Status UploadPart(std::shared_ptr<Buffer> buffer) {
- return UploadPart(buffer->data(), buffer->size(), buffer);
+ Status UploadUsingSingleRequest() {
+ std::shared_ptr<Buffer> buf;
+ if (current_part_ == nullptr) {
+ // In case the stream is closed directly after it has been opened without writing
+ // anything, we'll have to create an empty buffer.
+ buf = Buffer::FromVector<uint8_t>({});
+ } else {
+ ARROW_ASSIGN_OR_RAISE(buf, current_part_->Finish());
+ }
+
+ current_part_.reset();
+ current_part_size_ = 0;
+ return UploadUsingSingleRequest(buf);
}
- Status UploadPart(const void* data, int64_t nbytes,
- std::shared_ptr<Buffer> owned_buffer = nullptr) {
- S3Model::UploadPartRequest req;
+ template <typename RequestType, typename OutcomeType>
+ using UploadResultCallbackFunction =
+ std::function<Status(const RequestType& request, std::shared_ptr<UploadState>,
+ int32_t, OutcomeType outcome)>;
Review Comment:
Let's make this signature more informative
```suggestion
template <typename RequestType, typename OutcomeType>
using UploadResultCallbackFunction =
std::function<Status(const RequestType& request, std::shared_ptr<UploadState>,
int32_t part_number, OutcomeType outcome)>;
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "OliLay (via GitHub)" <gi...@apache.org>.
OliLay commented on code in PR #41564:
URL: https://github.com/apache/arrow/pull/41564#discussion_r1604483071
##########
cpp/src/arrow/filesystem/s3fs_test.cc:
##########
@@ -1197,6 +1199,19 @@ TEST_F(TestS3FS, OpenOutputStreamSyncWrites) {
TestOpenOutputStream();
}
+TEST_F(TestS3FS, OpenOutputStreamNoBucketSanitizationSyncWrites) {
Review Comment:
I added a struct which abstracts the parameter combinations. [7748824](https://github.com/apache/arrow/pull/41564/commits/77488243b71be6688e9131efac03e9b98d14b1bd)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "OliLay (via GitHub)" <gi...@apache.org>.
OliLay commented on PR #41564:
URL: https://github.com/apache/arrow/pull/41564#issuecomment-2175471866
Hi @pitrou , I merged main again due to a conflict. Maybe you could give it another review if you have time.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "pitrou (via GitHub)" <gi...@apache.org>.
pitrou commented on PR #41564:
URL: https://github.com/apache/arrow/pull/41564#issuecomment-2242593334
@github-actions crossbow submit -g cpp -g python
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "OliLay (via GitHub)" <gi...@apache.org>.
OliLay commented on code in PR #41564:
URL: https://github.com/apache/arrow/pull/41564#discussion_r1599986390
##########
cpp/src/arrow/filesystem/s3fs_test.cc:
##########
@@ -521,9 +521,6 @@ class TestS3FS : public S3TestMixin {
void TestOpenOutputStream() {
std::shared_ptr<io::OutputStream> stream;
- // Nonexistent
- ASSERT_RAISES(IOError, fs_->OpenOutputStream("nonexistent-bucket/somefile"));
Review Comment:
See the PR description ("User facing changes") and the discussion below. With this optimization we may throw later if the bucket does not exist because the `Open` of the output stream does not trigger an S3 request any more.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #41564:
URL: https://github.com/apache/arrow/pull/41564#issuecomment-2242581272
Revision: 57ff9921779451b3b92762e35711c9ca82b9f24d
Submitted crossbow builds: [ursacomputing/crossbow @ actions-95140b88e7](https://github.com/ursacomputing/crossbow/branches/all?query=actions-95140b88e7)
|Task|Status|
|----|------|
|example-python-minimal-build-fedora-conda|[](https://github.com/ursacomputing/crossbow/actions/runs/10038771088/job/27741392027)|
|example-python-minimal-build-ubuntu-venv|[](https://github.com/ursacomputing/crossbow/actions/runs/10038771106/job/27741392201)|
|test-alpine-linux-cpp|[](https://github.com/ursacomputing/crossbow/actions/runs/10038770426/job/27741388529)|
|test-build-cpp-fuzz|[](https://github.com/ursacomputing/crossbow/actions/runs/10038770911/job/27741390944)|
|test-conda-cpp|[](https://github.com/ursacomputing/crossbow/actions/runs/10038771334/job/27741393228)|
|test-conda-cpp-valgrind|[](https://github.com/ursacomputing/crossbow/actions/runs/10038771231/job/27741392692)|
|test-conda-python-3.10|[](https://github.com/ursacomputing/crossbow/actions/runs/10038770535/job/27741389311)|
|test-conda-python-3.10-cython2|[](https://github.com/ursacomputing/crossbow/actions/runs/10038770793/job/27741390310)|
|test-conda-python-3.10-hdfs-2.9.2|[](https://github.com/ursacomputing/crossbow/actions/runs/10038770378/job/27741388352)|
|test-conda-python-3.10-hdfs-3.2.1|[](https://github.com/ursacomputing/crossbow/actions/runs/10038770891/job/27741390783)|
|test-conda-python-3.10-pandas-latest|[](https://github.com/ursacomputing/crossbow/actions/runs/10038770568/job/27741389625)|
|test-conda-python-3.10-pandas-nightly|[](https://github.com/ursacomputing/crossbow/actions/runs/10038770505/job/27741389133)|
|test-conda-python-3.10-spark-v3.5.0|[](https://github.com/ursacomputing/crossbow/actions/runs/10038770638/job/27741389764)|
|test-conda-python-3.10-substrait|[](https://github.com/ursacomputing/crossbow/actions/runs/10038770912/job/27741391276)|
|test-conda-python-3.11|[](https://github.com/ursacomputing/crossbow/actions/runs/10038771289/job/27741393203)|
|test-conda-python-3.11-dask-latest|[](https://github.com/ursacomputing/crossbow/actions/runs/10038770659/job/27741389963)|
|test-conda-python-3.11-dask-upstream_devel|[](https://github.com/ursacomputing/crossbow/actions/runs/10038771716/job/27741395089)|
|test-conda-python-3.11-hypothesis|[](https://github.com/ursacomputing/crossbow/actions/runs/10038771470/job/27741394005)|
|test-conda-python-3.11-pandas-upstream_devel|[](https://github.com/ursacomputing/crossbow/actions/runs/10038771501/job/27741394037)|
|test-conda-python-3.11-spark-master|[](https://github.com/ursacomputing/crossbow/actions/runs/10038771270/job/27741393000)|
|test-conda-python-3.12|[](https://github.com/ursacomputing/crossbow/actions/runs/10038770941/job/27741391436)|
|test-conda-python-3.8|[](https://github.com/ursacomputing/crossbow/actions/runs/10038770473/job/27741388910)|
|test-conda-python-3.8-pandas-1.0|[](https://github.com/ursacomputing/crossbow/actions/runs/10038770928/job/27741391098)|
|test-conda-python-3.8-spark-v3.5.0|[](https://github.com/ursacomputing/crossbow/actions/runs/10038770989/job/27741391675)|
|test-conda-python-3.9|[](https://github.com/ursacomputing/crossbow/actions/runs/10038771380/job/27741393428)|
|test-conda-python-3.9-pandas-latest|[](https://github.com/ursacomputing/crossbow/actions/runs/10038771029/job/27741391839)|
|test-cuda-cpp|[](https://github.com/ursacomputing/crossbow/actions/runs/10038770796/job/27741390560)|
|test-cuda-python|[](https://github.com/ursacomputing/crossbow/actions/runs/10038771634/job/27741394365)|
|test-debian-12-cpp-amd64|[](https://github.com/ursacomputing/crossbow/actions/runs/10038770507/job/27741388932)|
|test-debian-12-cpp-i386|[](https://github.com/ursacomputing/crossbow/actions/runs/10038771730/job/27741394815)|
|test-debian-12-python-3-amd64|[](https://github.com/ursacomputing/crossbow/actions/runs/10038771234/job/27741392766)|
|test-debian-12-python-3-i386|[](https://github.com/ursacomputing/crossbow/actions/runs/10038770823/job/27741390668)|
|test-fedora-39-cpp|[](https://github.com/ursacomputing/crossbow/actions/runs/10038771847/job/27741395441)|
|test-fedora-39-python-3|[](https://github.com/ursacomputing/crossbow/actions/runs/10038771261/job/27741392859)|
|test-ubuntu-20.04-cpp|[](https://github.com/ursacomputing/crossbow/actions/runs/10038770763/job/27741390172)|
|test-ubuntu-20.04-cpp-bundled|[](https://github.com/ursacomputing/crossbow/actions/runs/10038770528/job/27741389452)|
|test-ubuntu-20.04-cpp-minimal-with-formats|[](https://github.com/ursacomputing/crossbow/actions/runs/10038770752/job/27741390153)|
|test-ubuntu-20.04-cpp-thread-sanitizer|[](https://github.com/ursacomputing/crossbow/actions/runs/10038771555/job/27741394027)|
|test-ubuntu-20.04-python-3|[](https://github.com/ursacomputing/crossbow/actions/runs/10038770456/job/27741388773)|
|test-ubuntu-22.04-cpp|[](https://github.com/ursacomputing/crossbow/actions/runs/10038771042/job/27741391822)|
|test-ubuntu-22.04-cpp-20|[](https://github.com/ursacomputing/crossbow/actions/runs/10038771410/job/27741393707)|
|test-ubuntu-22.04-cpp-emscripten|[](https://github.com/ursacomputing/crossbow/actions/runs/10038770290/job/27741388265)|
|test-ubuntu-22.04-cpp-no-threading|[](https://github.com/ursacomputing/crossbow/actions/runs/10038771582/job/27741394404)|
|test-ubuntu-22.04-python-3|[](https://github.com/ursacomputing/crossbow/actions/runs/10038770927/job/27741390993)|
|test-ubuntu-24.04-cpp|[](https://github.com/ursacomputing/crossbow/actions/runs/10038770440/job/27741388739)|
|test-ubuntu-24.04-cpp-gcc-14|[](https://github.com/ursacomputing/crossbow/actions/runs/10038770950/job/27741391332)|
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "OliLay (via GitHub)" <gi...@apache.org>.
OliLay commented on PR #41564:
URL: https://github.com/apache/arrow/pull/41564#issuecomment-2149002133
I also merged main into this branch due to a conflict. Should be free of conflicts now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "OliLay (via GitHub)" <gi...@apache.org>.
OliLay commented on PR #41564:
URL: https://github.com/apache/arrow/pull/41564#issuecomment-2110302759
> A compromise would be to add a dedicated option in `S3Options`, but of course the optimization would only benefit those users that enable the option.
We can do that. I would propose that if the optimization is disabled, we directly use multi-part uploads (basically replicating the old behavior). I don't think it makes sense to explicitly issue a [HeadBucket](https://docs.aws.amazon.com/AmazonS3/latest/API/API_HeadBucket.html) request because that will lead to minimum 4 requests with multi-part uploads then. (although we would only have 2 requests for small writes without the optimization compared to current `main`)
What do you think?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "OliLay (via GitHub)" <gi...@apache.org>.
OliLay commented on code in PR #41564:
URL: https://github.com/apache/arrow/pull/41564#discussion_r1601402884
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1606,42 +1656,60 @@ class ObjectOutputStream final : public io::OutputStream {
return Status::OK();
}
- ARROW_ASSIGN_OR_RAISE(auto client_lock, holder_->Lock());
+ if (is_multipart_created_) {
+ ARROW_ASSIGN_OR_RAISE(auto client_lock, holder_->Lock());
- S3Model::AbortMultipartUploadRequest req;
- req.SetBucket(ToAwsString(path_.bucket));
- req.SetKey(ToAwsString(path_.key));
- req.SetUploadId(upload_id_);
+ S3Model::AbortMultipartUploadRequest req;
+ req.SetBucket(ToAwsString(path_.bucket));
+ req.SetKey(ToAwsString(path_.key));
+ req.SetUploadId(upload_id_);
- auto outcome = client_lock.Move()->AbortMultipartUpload(req);
- if (!outcome.IsSuccess()) {
- return ErrorToStatus(
- std::forward_as_tuple("When aborting multiple part upload for key '", path_.key,
- "' in bucket '", path_.bucket, "': "),
- "AbortMultipartUpload", outcome.GetError());
+ auto outcome = client_lock.Move()->AbortMultipartUpload(req);
+ if (!outcome.IsSuccess()) {
+ return ErrorToStatus(
+ std::forward_as_tuple("When aborting multiple part upload for key '",
+ path_.key, "' in bucket '", path_.bucket, "': "),
+ "AbortMultipartUpload", outcome.GetError());
+ }
}
+
current_part_.reset();
holder_ = nullptr;
closed_ = true;
+
return Status::OK();
}
// OutputStream interface
+ bool ShouldBeMultipartUpload() { return pos_ > kMultiPartUploadThresholdSize; }
+
+ bool IsMultipartUpload() { return ShouldBeMultipartUpload() || is_multipart_created_; }
+
Status EnsureReadyToFlushFromClose() {
- if (current_part_) {
- // Upload last part
- RETURN_NOT_OK(CommitCurrentPart());
- }
+ if (IsMultipartUpload()) {
Review Comment:
Can you elaborate more? I think we're calling `EnsureReadyToFlushFromClose` just to flush the data if we close the stream, hence sometimes we still have data in-flight that has to be uploaded (the "last part" which we commit in this branch, or a dummy part if no data was written at all).
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1606,42 +1656,60 @@ class ObjectOutputStream final : public io::OutputStream {
return Status::OK();
}
- ARROW_ASSIGN_OR_RAISE(auto client_lock, holder_->Lock());
+ if (is_multipart_created_) {
+ ARROW_ASSIGN_OR_RAISE(auto client_lock, holder_->Lock());
- S3Model::AbortMultipartUploadRequest req;
- req.SetBucket(ToAwsString(path_.bucket));
- req.SetKey(ToAwsString(path_.key));
- req.SetUploadId(upload_id_);
+ S3Model::AbortMultipartUploadRequest req;
+ req.SetBucket(ToAwsString(path_.bucket));
+ req.SetKey(ToAwsString(path_.key));
+ req.SetUploadId(upload_id_);
- auto outcome = client_lock.Move()->AbortMultipartUpload(req);
- if (!outcome.IsSuccess()) {
- return ErrorToStatus(
- std::forward_as_tuple("When aborting multiple part upload for key '", path_.key,
- "' in bucket '", path_.bucket, "': "),
- "AbortMultipartUpload", outcome.GetError());
+ auto outcome = client_lock.Move()->AbortMultipartUpload(req);
+ if (!outcome.IsSuccess()) {
+ return ErrorToStatus(
+ std::forward_as_tuple("When aborting multiple part upload for key '",
+ path_.key, "' in bucket '", path_.bucket, "': "),
+ "AbortMultipartUpload", outcome.GetError());
+ }
}
+
current_part_.reset();
holder_ = nullptr;
closed_ = true;
+
return Status::OK();
}
// OutputStream interface
+ bool ShouldBeMultipartUpload() { return pos_ > kMultiPartUploadThresholdSize; }
+
+ bool IsMultipartUpload() { return ShouldBeMultipartUpload() || is_multipart_created_; }
Review Comment:
1. Just my personal style. (also, with the added debug assert we now also have to call `ShouldBeMultipartUpload` separately.
2. Made them const in [1e9e3a4](https://github.com/apache/arrow/pull/41564/commits/1e9e3a476874b9d464517f5f8357b370e315fbed).
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1597,6 +1632,21 @@ class ObjectOutputStream final : public io::OutputStream {
}
upload_id_ = outcome.GetResult().GetUploadId();
upload_state_ = std::make_shared<UploadState>();
Review Comment:
Yes, we would create the upload state twice. (basically resetting it)
I changed it in [1e9e3a4](https://github.com/apache/arrow/pull/41564/commits/1e9e3a476874b9d464517f5f8357b370e315fbed)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "OliLay (via GitHub)" <gi...@apache.org>.
OliLay commented on code in PR #41564:
URL: https://github.com/apache/arrow/pull/41564#discussion_r1601401047
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1560,34 +1582,47 @@ class ObjectOutputStream final : public io::OutputStream {
path_(path),
metadata_(metadata),
default_metadata_(options.default_metadata),
- background_writes_(options.background_writes) {}
+ background_writes_(options.background_writes),
+ sanitize_bucket_on_open_(options.sanitize_bucket_on_open) {}
~ObjectOutputStream() override {
// For compliance with the rest of the IO stack, Close rather than Abort,
// even though it may be more expensive.
io::internal::CloseFromDestructor(this);
}
- Status Init() {
- ARROW_ASSIGN_OR_RAISE(auto client_lock, holder_->Lock());
+ template <typename ObjectRequest>
+ Status SetMetadataInRequest(ObjectRequest* request) {
+ std::shared_ptr<const KeyValueMetadata> metadata;
- // Initiate the multi-part upload
- S3Model::CreateMultipartUploadRequest req;
- req.SetBucket(ToAwsString(path_.bucket));
- req.SetKey(ToAwsString(path_.key));
if (metadata_ && metadata_->size() != 0) {
- RETURN_NOT_OK(SetObjectMetadata(metadata_, &req));
+ metadata = metadata_;
} else if (default_metadata_ && default_metadata_->size() != 0) {
- RETURN_NOT_OK(SetObjectMetadata(default_metadata_, &req));
+ metadata = default_metadata_;
}
- // If we do not set anything then the SDK will default to application/xml
- // which confuses some tools (https://github.com/apache/arrow/issues/11934)
- // So we instead default to application/octet-stream which is less misleading
- if (!req.ContentTypeHasBeenSet()) {
- req.SetContentType("application/octet-stream");
+ if (metadata == nullptr ||
+ !metadata->Contains(ObjectMetadataSetter<ObjectRequest>::CONTENT_TYPE_KEY)) {
+ // If we do not set anything then the SDK will default to application/xml
+ // which confuses some tools (https://github.com/apache/arrow/issues/11934)
+ // So we instead default to application/octet-stream which is less misleading
+ request->SetContentType("application/octet-stream");
+ } else {
+ RETURN_NOT_OK(SetObjectMetadata(metadata, request));
}
+ return Status::OK();
+ }
+
+ Status CreateMultipartUpload() {
+ ARROW_ASSIGN_OR_RAISE(auto client_lock, holder_->Lock());
Review Comment:
I added a debug assert asserting that we do not create a multi-part upload if the conditions are not met (not enough data or optimization disabled). [1e9e3a4](https://github.com/apache/arrow/pull/41564/commits/1e9e3a476874b9d464517f5f8357b370e315fbed)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "OliLay (via GitHub)" <gi...@apache.org>.
OliLay commented on PR #41564:
URL: https://github.com/apache/arrow/pull/41564#issuecomment-2110203513
> My point is that _if the path cannot be written to_, the error happens when opening the file, not later on.
That is true. I guess the question is if `arrow`'s OutputStream API makes an explicit _guarantee_ that `Open` should throw if the target does not exist. My guess would be that you shouldn't built code upon this assumption if it isn't explicitly stated in `arrow`'s API/docs (which it is not), but of course real-world usage deviates from that (Hyrum's Law).
But checking if the bucket exists would at least come with another 1x RTT to S3 and the goal of the PR was to reduce the amount of blocking calls to S3 to reduce overall latency. If we add another check here, we'll have a total 2x RTT to S3 for small uploads, which is better than the initial 3x RTT without this change, but still not optimal from a performance-view.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #41564:
URL: https://github.com/apache/arrow/pull/41564#issuecomment-2097642865
:warning: GitHub issue #40557 **has been automatically assigned in GitHub** to PR creator.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "pitrou (via GitHub)" <gi...@apache.org>.
pitrou commented on code in PR #41564:
URL: https://github.com/apache/arrow/pull/41564#discussion_r1603500147
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1293,12 +1295,14 @@ std::shared_ptr<const KeyValueMetadata> GetObjectMetadata(const ObjectResult& re
template <typename ObjectRequest>
struct ObjectMetadataSetter {
+ static constexpr std::string_view CONTENT_TYPE_KEY = "Content-Type";
Review Comment:
Nit: naming conventions for constants
```suggestion
static constexpr std::string_view kContentTypeKey = "Content-Type";
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "pitrou (via GitHub)" <gi...@apache.org>.
pitrou merged PR #41564:
URL: https://github.com/apache/arrow/pull/41564
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "OliLay (via GitHub)" <gi...@apache.org>.
OliLay commented on code in PR #41564:
URL: https://github.com/apache/arrow/pull/41564#discussion_r1601112427
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1545,6 +1559,13 @@ class ObjectInputFile final : public io::RandomAccessFile {
// (for rational, see: https://github.com/apache/arrow/issues/34363)
static constexpr int64_t kPartUploadSize = 10 * 1024 * 1024;
+// Above this threshold, use a multi-part upload instead of a single request upload.
+static constexpr int64_t kMultiPartUploadThresholdSize = 5 * 1024 * 1024;
Review Comment:
[78efe6e](https://github.com/apache/arrow/pull/41564/commits/78efe6e48e11c198073a5912714c9be94421b283)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "OliLay (via GitHub)" <gi...@apache.org>.
OliLay commented on code in PR #41564:
URL: https://github.com/apache/arrow/pull/41564#discussion_r1602735185
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1568,26 +1589,38 @@ class ObjectOutputStream final : public io::OutputStream {
io::internal::CloseFromDestructor(this);
}
- Status Init() {
- ARROW_ASSIGN_OR_RAISE(auto client_lock, holder_->Lock());
+ template <typename ObjectRequest>
+ Status SetMetadataInRequest(ObjectRequest* request) {
+ std::shared_ptr<const KeyValueMetadata> metadata;
- // Initiate the multi-part upload
- S3Model::CreateMultipartUploadRequest req;
- req.SetBucket(ToAwsString(path_.bucket));
- req.SetKey(ToAwsString(path_.key));
if (metadata_ && metadata_->size() != 0) {
- RETURN_NOT_OK(SetObjectMetadata(metadata_, &req));
+ metadata = metadata_;
} else if (default_metadata_ && default_metadata_->size() != 0) {
- RETURN_NOT_OK(SetObjectMetadata(default_metadata_, &req));
+ metadata = default_metadata_;
}
- // If we do not set anything then the SDK will default to application/xml
- // which confuses some tools (https://github.com/apache/arrow/issues/11934)
- // So we instead default to application/octet-stream which is less misleading
- if (!req.ContentTypeHasBeenSet()) {
- req.SetContentType("application/octet-stream");
+ if (metadata == nullptr ||
+ !metadata->Contains(ObjectMetadataSetter<ObjectRequest>::CONTENT_TYPE_KEY)) {
+ // If we do not set anything then the SDK will default to application/xml
+ // which confuses some tools (https://github.com/apache/arrow/issues/11934)
+ // So we instead default to application/octet-stream which is less misleading
+ request->SetContentType("application/octet-stream");
+ } else {
+ RETURN_NOT_OK(SetObjectMetadata(metadata, request));
}
Review Comment:
[92aa434](https://github.com/apache/arrow/pull/41564/commits/92aa4348fa295589e53a8e9541138647e8f00f41)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "OliLay (via GitHub)" <gi...@apache.org>.
OliLay commented on code in PR #41564:
URL: https://github.com/apache/arrow/pull/41564#discussion_r1625851115
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1560,34 +1582,49 @@ class ObjectOutputStream final : public io::OutputStream {
path_(path),
metadata_(metadata),
default_metadata_(options.default_metadata),
- background_writes_(options.background_writes) {}
+ background_writes_(options.background_writes),
+ allow_delayed_open_(options.allow_delayed_open) {}
~ObjectOutputStream() override {
// For compliance with the rest of the IO stack, Close rather than Abort,
// even though it may be more expensive.
io::internal::CloseFromDestructor(this);
}
- Status Init() {
- ARROW_ASSIGN_OR_RAISE(auto client_lock, holder_->Lock());
+ template <typename ObjectRequest>
+ Status SetMetadataInRequest(ObjectRequest* request) {
+ std::shared_ptr<const KeyValueMetadata> metadata;
- // Initiate the multi-part upload
- S3Model::CreateMultipartUploadRequest req;
- req.SetBucket(ToAwsString(path_.bucket));
- req.SetKey(ToAwsString(path_.key));
if (metadata_ && metadata_->size() != 0) {
- RETURN_NOT_OK(SetObjectMetadata(metadata_, &req));
+ metadata = metadata_;
} else if (default_metadata_ && default_metadata_->size() != 0) {
- RETURN_NOT_OK(SetObjectMetadata(default_metadata_, &req));
+ metadata = default_metadata_;
}
- // If we do not set anything then the SDK will default to application/xml
- // which confuses some tools (https://github.com/apache/arrow/issues/11934)
- // So we instead default to application/octet-stream which is less misleading
- if (!req.ContentTypeHasBeenSet()) {
- req.SetContentType("application/octet-stream");
+ if (metadata &&
+ metadata->Contains(ObjectMetadataSetter<ObjectRequest>::kContentTypeKey)) {
+ RETURN_NOT_OK(SetObjectMetadata(metadata, request));
+ } else {
+ // If we do not set anything then the SDK will default to application/xml
+ // which confuses some tools (https://github.com/apache/arrow/issues/11934)
+ // So we instead default to application/octet-stream which is less misleading
+ request->SetContentType("application/octet-stream");
Review Comment:
Good catch, I think this was an issue, I fixed it and made the code also a bit clearer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "OliLay (via GitHub)" <gi...@apache.org>.
OliLay commented on code in PR #41564:
URL: https://github.com/apache/arrow/pull/41564#discussion_r1625852412
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1606,42 +1656,62 @@ class ObjectOutputStream final : public io::OutputStream {
return Status::OK();
}
- ARROW_ASSIGN_OR_RAISE(auto client_lock, holder_->Lock());
+ if (is_multipart_created_) {
+ ARROW_ASSIGN_OR_RAISE(auto client_lock, holder_->Lock());
- S3Model::AbortMultipartUploadRequest req;
- req.SetBucket(ToAwsString(path_.bucket));
- req.SetKey(ToAwsString(path_.key));
- req.SetUploadId(upload_id_);
+ S3Model::AbortMultipartUploadRequest req;
+ req.SetBucket(ToAwsString(path_.bucket));
+ req.SetKey(ToAwsString(path_.key));
+ req.SetUploadId(upload_id_);
- auto outcome = client_lock.Move()->AbortMultipartUpload(req);
- if (!outcome.IsSuccess()) {
- return ErrorToStatus(
- std::forward_as_tuple("When aborting multiple part upload for key '", path_.key,
- "' in bucket '", path_.bucket, "': "),
- "AbortMultipartUpload", outcome.GetError());
+ auto outcome = client_lock.Move()->AbortMultipartUpload(req);
+ if (!outcome.IsSuccess()) {
+ return ErrorToStatus(
+ std::forward_as_tuple("When aborting multiple part upload for key '",
+ path_.key, "' in bucket '", path_.bucket, "': "),
+ "AbortMultipartUpload", outcome.GetError());
+ }
}
+
current_part_.reset();
holder_ = nullptr;
closed_ = true;
+
return Status::OK();
}
// OutputStream interface
+ bool ShouldBeMultipartUpload() const { return pos_ > kMultiPartUploadThresholdSize; }
+
+ bool IsMultipartUpload() const {
+ return ShouldBeMultipartUpload() || is_multipart_created_;
Review Comment:
There is probably no reason, I think it evolved to be like this during the PR review and also before we had the feature flag. I streamlined the handling now to use the upload id instead of this additional boolean.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "OliLay (via GitHub)" <gi...@apache.org>.
OliLay commented on code in PR #41564:
URL: https://github.com/apache/arrow/pull/41564#discussion_r1625853368
##########
cpp/src/arrow/filesystem/s3fs_test.cc:
##########
@@ -1189,40 +1191,55 @@ TEST_F(TestS3FS, OpenInputFile) {
ASSERT_RAISES(IOError, file->Seek(10));
}
-TEST_F(TestS3FS, OpenOutputStreamBackgroundWrites) { TestOpenOutputStream(); }
+struct S3OptionsTestParameters {
+ bool background_writes{false};
+ bool allow_delayed_open{false};
-TEST_F(TestS3FS, OpenOutputStreamSyncWrites) {
- options_.background_writes = false;
- MakeFileSystem();
- TestOpenOutputStream();
-}
-
-TEST_F(TestS3FS, OpenOutputStreamAbortBackgroundWrites) { TestOpenOutputStreamAbort(); }
+ void apply_to_s3_options(S3Options& options) const {
+ options.background_writes = background_writes;
+ options.allow_delayed_open = allow_delayed_open;
+ }
-TEST_F(TestS3FS, OpenOutputStreamAbortSyncWrites) {
- options_.background_writes = false;
- MakeFileSystem();
- TestOpenOutputStreamAbort();
-}
+ static std::vector<S3OptionsTestParameters> GetCartesianProduct() {
+ return {
+ S3OptionsTestParameters{true, false},
+ S3OptionsTestParameters{false, false},
+ S3OptionsTestParameters{true, true},
+ S3OptionsTestParameters{false, true},
+ };
+ }
+};
-TEST_F(TestS3FS, OpenOutputStreamDestructorBackgroundWrites) {
- TestOpenOutputStreamDestructor();
+TEST_F(TestS3FS, OpenOutputStream) {
+ for (const auto& combination : S3OptionsTestParameters::GetCartesianProduct()) {
+ combination.apply_to_s3_options(options_);
+ MakeFileSystem();
Review Comment:
Good catch, I implemented cleanup (i.e. emptying the test bucket and restoring the test files again) after each test run.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "OliLay (via GitHub)" <gi...@apache.org>.
OliLay commented on code in PR #41564:
URL: https://github.com/apache/arrow/pull/41564#discussion_r1625850790
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1293,12 +1295,14 @@ std::shared_ptr<const KeyValueMetadata> GetObjectMetadata(const ObjectResult& re
template <typename ObjectRequest>
struct ObjectMetadataSetter {
+ static constexpr std::string_view kContentTypeKey = "Content-Type";
Review Comment:
I reverted the change to move it to a constant and rather have it inline now everywhere again.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #41564:
URL: https://github.com/apache/arrow/pull/41564#discussion_r1601405451
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1606,42 +1656,60 @@ class ObjectOutputStream final : public io::OutputStream {
return Status::OK();
}
- ARROW_ASSIGN_OR_RAISE(auto client_lock, holder_->Lock());
+ if (is_multipart_created_) {
+ ARROW_ASSIGN_OR_RAISE(auto client_lock, holder_->Lock());
- S3Model::AbortMultipartUploadRequest req;
- req.SetBucket(ToAwsString(path_.bucket));
- req.SetKey(ToAwsString(path_.key));
- req.SetUploadId(upload_id_);
+ S3Model::AbortMultipartUploadRequest req;
+ req.SetBucket(ToAwsString(path_.bucket));
+ req.SetKey(ToAwsString(path_.key));
+ req.SetUploadId(upload_id_);
- auto outcome = client_lock.Move()->AbortMultipartUpload(req);
- if (!outcome.IsSuccess()) {
- return ErrorToStatus(
- std::forward_as_tuple("When aborting multiple part upload for key '", path_.key,
- "' in bucket '", path_.bucket, "': "),
- "AbortMultipartUpload", outcome.GetError());
+ auto outcome = client_lock.Move()->AbortMultipartUpload(req);
+ if (!outcome.IsSuccess()) {
+ return ErrorToStatus(
+ std::forward_as_tuple("When aborting multiple part upload for key '",
+ path_.key, "' in bucket '", path_.bucket, "': "),
+ "AbortMultipartUpload", outcome.GetError());
+ }
}
+
current_part_.reset();
holder_ = nullptr;
closed_ = true;
+
return Status::OK();
}
// OutputStream interface
+ bool ShouldBeMultipartUpload() { return pos_ > kMultiPartUploadThresholdSize; }
+
+ bool IsMultipartUpload() { return ShouldBeMultipartUpload() || is_multipart_created_; }
+
Status EnsureReadyToFlushFromClose() {
- if (current_part_) {
- // Upload last part
- RETURN_NOT_OK(CommitCurrentPart());
- }
+ if (IsMultipartUpload()) {
Review Comment:
Just to make it clear, since if `part_size_` is larger than size-limit, it should be switch to multipart during "doWrite". During closing maybe it cannot limit the size-limit?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "pitrou (via GitHub)" <gi...@apache.org>.
pitrou commented on code in PR #41564:
URL: https://github.com/apache/arrow/pull/41564#discussion_r1603506499
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1606,42 +1656,62 @@ class ObjectOutputStream final : public io::OutputStream {
return Status::OK();
}
- ARROW_ASSIGN_OR_RAISE(auto client_lock, holder_->Lock());
+ if (is_multipart_created_) {
+ ARROW_ASSIGN_OR_RAISE(auto client_lock, holder_->Lock());
- S3Model::AbortMultipartUploadRequest req;
- req.SetBucket(ToAwsString(path_.bucket));
- req.SetKey(ToAwsString(path_.key));
- req.SetUploadId(upload_id_);
+ S3Model::AbortMultipartUploadRequest req;
+ req.SetBucket(ToAwsString(path_.bucket));
+ req.SetKey(ToAwsString(path_.key));
+ req.SetUploadId(upload_id_);
- auto outcome = client_lock.Move()->AbortMultipartUpload(req);
- if (!outcome.IsSuccess()) {
- return ErrorToStatus(
- std::forward_as_tuple("When aborting multiple part upload for key '", path_.key,
- "' in bucket '", path_.bucket, "': "),
- "AbortMultipartUpload", outcome.GetError());
+ auto outcome = client_lock.Move()->AbortMultipartUpload(req);
+ if (!outcome.IsSuccess()) {
+ return ErrorToStatus(
+ std::forward_as_tuple("When aborting multiple part upload for key '",
+ path_.key, "' in bucket '", path_.bucket, "': "),
+ "AbortMultipartUpload", outcome.GetError());
+ }
}
+
current_part_.reset();
holder_ = nullptr;
closed_ = true;
+
return Status::OK();
}
// OutputStream interface
+ bool ShouldBeMultipartUpload() const { return pos_ > kMultiPartUploadThresholdSize; }
+
+ bool IsMultipartUpload() const {
+ return ShouldBeMultipartUpload() || is_multipart_created_;
+ }
+
Status EnsureReadyToFlushFromClose() {
- if (current_part_) {
- // Upload last part
- RETURN_NOT_OK(CommitCurrentPart());
- }
+ if (IsMultipartUpload()) {
+ if (current_part_) {
+ // Upload last part
+ RETURN_NOT_OK(CommitCurrentPart());
+ }
- // S3 mandates at least one part, upload an empty one if necessary
- if (part_number_ == 1) {
- RETURN_NOT_OK(UploadPart("", 0));
+ // S3 mandates at least one part, upload an empty one if necessary
+ if (part_number_ == 1) {
+ RETURN_NOT_OK(UploadPart("", 0));
+ }
+ } else {
+ RETURN_NOT_OK(UploadUsingSingleRequest());
}
return Status::OK();
}
+ Status CleanupAfterFlush() {
Review Comment:
Should this be named `CleanupAfterClose`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "pitrou (via GitHub)" <gi...@apache.org>.
pitrou commented on code in PR #41564:
URL: https://github.com/apache/arrow/pull/41564#discussion_r1603526713
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1838,9 +1952,114 @@ class ObjectOutputStream final : public io::OutputStream {
return Status::OK();
}
- static void HandleUploadOutcome(const std::shared_ptr<UploadState>& state,
- int part_number, const S3Model::UploadPartRequest& req,
- const Result<S3Model::UploadPartOutcome>& result) {
+ static Status UploadError(const Aws::S3::Model::PutObjectRequest& request,
+ const Aws::S3::Model::PutObjectOutcome& outcome) {
+ return ErrorToStatus(
+ std::forward_as_tuple("When uploading object with key '", request.GetKey(),
+ "' in bucket '", request.GetBucket(), "': "),
+ "PutObject", outcome.GetError());
+ }
+
+ Status UploadUsingSingleRequest(std::shared_ptr<Buffer> buffer) {
+ return UploadUsingSingleRequest(buffer->data(), buffer->size(), buffer);
+ }
+
+ Status UploadUsingSingleRequest(const void* data, int64_t nbytes,
+ std::shared_ptr<Buffer> owned_buffer = nullptr) {
+ auto sync_result_callback = [](const Aws::S3::Model::PutObjectRequest& request,
+ std::shared_ptr<UploadState> state,
+ int32_t part_number,
+ Aws::S3::Model::PutObjectOutcome outcome) {
+ if (!outcome.IsSuccess()) {
+ return UploadError(request, outcome);
+ }
+ return Status::OK();
+ };
+
+ auto async_result_callback = [](const Aws::S3::Model::PutObjectRequest& request,
+ std::shared_ptr<UploadState> state,
+ int32_t part_number,
+ Aws::S3::Model::PutObjectOutcome outcome) {
+ HandleUploadUsingSingleRequestOutcome(state, request, outcome.GetResult());
+ return Status::OK();
+ };
+
+ Aws::S3::Model::PutObjectRequest req{};
+ RETURN_NOT_OK(SetMetadataInRequest(&req));
+
+ return Upload<Aws::S3::Model::PutObjectRequest, Aws::S3::Model::PutObjectOutcome>(
+ std::move(req), sync_result_callback, async_result_callback, data, nbytes,
+ owned_buffer);
Review Comment:
Nits: move more arguments
```suggestion
return Upload<Aws::S3::Model::PutObjectRequest, Aws::S3::Model::PutObjectOutcome>(
std::move(req), std::move(sync_result_callback), std::move(async_result_callback),
data, nbytes, std::move(owned_buffer));
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "OliLay (via GitHub)" <gi...@apache.org>.
OliLay commented on code in PR #41564:
URL: https://github.com/apache/arrow/pull/41564#discussion_r1599998938
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1545,6 +1559,13 @@ class ObjectInputFile final : public io::RandomAccessFile {
// (for rational, see: https://github.com/apache/arrow/issues/34363)
static constexpr int64_t kPartUploadSize = 10 * 1024 * 1024;
+// Above this threshold, use a multi-part upload instead of a single request upload.
+static constexpr int64_t kMultiPartUploadThresholdSize = 5 * 1024 * 1024;
Review Comment:
Would also be a valid approach and would simplify the code a bit. AWS S3 docs themselves state that you could/should use `PutObject` request for data up to 100MB (see [here](https://docs.aws.amazon.com/AmazonS3/latest/userguide/mpuoverview.html)), so orienting at the part size which is 10MB would also be fine. What do other think?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-40557: [C++] Use `PutObject` request for S3 in OutputStream when only uploading small data [arrow]
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #41564:
URL: https://github.com/apache/arrow/pull/41564#discussion_r1601428929
##########
cpp/src/arrow/filesystem/s3fs.cc:
##########
@@ -1606,42 +1656,60 @@ class ObjectOutputStream final : public io::OutputStream {
return Status::OK();
}
- ARROW_ASSIGN_OR_RAISE(auto client_lock, holder_->Lock());
+ if (is_multipart_created_) {
+ ARROW_ASSIGN_OR_RAISE(auto client_lock, holder_->Lock());
- S3Model::AbortMultipartUploadRequest req;
- req.SetBucket(ToAwsString(path_.bucket));
- req.SetKey(ToAwsString(path_.key));
- req.SetUploadId(upload_id_);
+ S3Model::AbortMultipartUploadRequest req;
+ req.SetBucket(ToAwsString(path_.bucket));
+ req.SetKey(ToAwsString(path_.key));
+ req.SetUploadId(upload_id_);
- auto outcome = client_lock.Move()->AbortMultipartUpload(req);
- if (!outcome.IsSuccess()) {
- return ErrorToStatus(
- std::forward_as_tuple("When aborting multiple part upload for key '", path_.key,
- "' in bucket '", path_.bucket, "': "),
- "AbortMultipartUpload", outcome.GetError());
+ auto outcome = client_lock.Move()->AbortMultipartUpload(req);
+ if (!outcome.IsSuccess()) {
+ return ErrorToStatus(
+ std::forward_as_tuple("When aborting multiple part upload for key '",
+ path_.key, "' in bucket '", path_.bucket, "': "),
+ "AbortMultipartUpload", outcome.GetError());
+ }
}
+
current_part_.reset();
holder_ = nullptr;
closed_ = true;
+
return Status::OK();
}
// OutputStream interface
+ bool ShouldBeMultipartUpload() { return pos_ > kMultiPartUploadThresholdSize; }
+
+ bool IsMultipartUpload() { return ShouldBeMultipartUpload() || is_multipart_created_; }
+
Status EnsureReadyToFlushFromClose() {
- if (current_part_) {
- // Upload last part
- RETURN_NOT_OK(CommitCurrentPart());
- }
+ if (IsMultipartUpload()) {
Review Comment:
Yes, maybe simplify to `has_multi_part_upload_` or other is ok, but this also looks ok to me
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org