github-actions[bot] commented on code in PR #66557:
URL: https://github.com/apache/doris/pull/66557#discussion_r3765284431
##########
be/src/io/fs/s3_common.h:
##########
@@ -34,12 +39,146 @@ class StringViewStream :
Aws::Utils::Stream::PreallocatedStreamBuf, public std::
std::iostream(this) {}
};
+// The AWS SDK writes the body of every response into the stream built by the
response
+// stream factory of the request, whatever the status of that response is.
Reading an
+// object range straight into the buffer of the caller therefore breaks as
soon as the
+// server answers with an error: the XML body of a `429 SlowDown` is a few
hundred bytes
+// and does not fit into the buffer of a small range read.
`PreallocatedStreamBuf` does not
+// implement `overflow()`, so the stream turns bad, curl aborts the transfer
with
+// `CURLE_WRITE_ERROR`, and the SDK reports an `INTERNAL_FAILURE` named
"Failed to flush
+// response stream" while never recording the status code of the response.
Both the retry
+// strategy of the SDK and the retry of `S3FileReader` key on that status
code, so an error
+// the server asked us to retry ends up cancelling the query instead.
+//
+// This stream buffer writes into the buffer of the caller as long as the body
fits, which
+// is the case for every successful ranged read, and spills the rest into a
buffer of its
+// own. The stream never turns bad, so the SDK reports the real status code
and can parse
+// the error out of the body.
+class S3ResponseStreamBuf final : public std::streambuf {
+public:
+ // Bodies beyond this size are truncated. Only error documents are
expected to overflow
+ // and their leading bytes already carry the error code and the message.
This bounds the
+ // memory a single failing request can hold, whatever the server answers
with.
+ static constexpr size_t MAX_SPILL_SIZE = 1024 * 1024;
+
+ S3ResponseStreamBuf(void* buf, size_t nbytes) :
_buf(static_cast<char*>(buf)) {
+ setp(_buf, _buf + nbytes);
+ setg(_buf, _buf, _buf);
+ }
+
+protected:
+ std::streamsize xsputn(const char* s, std::streamsize n) override {
+ if (!_spilled) {
+ if (n <= epptr() - pptr()) {
+ std::memcpy(pptr(), s, n);
+ pbump(static_cast<int>(n));
+ return n;
+ }
+ _spill_over();
+ }
+ // Saturating on its own: the spill is clamped when it is filled from
the buffer of
+ // the caller, and this must not underflow into an unbounded write if
it ever is not.
+ auto room = _spill.size() < MAX_SPILL_SIZE ? MAX_SPILL_SIZE -
_spill.size() : 0;
+ auto writable = std::min(static_cast<size_t>(n), room);
Review Comment:
[P1] Do not present truncated XML as a complete error body
Once this response crosses `MAX_SPILL_SIZE`, the stream drops the tail but
still reports every write as consumed. A valid S3-compatible error whose
closing tags fall after the cap is therefore handed to `XmlErrorMarshaller` as
malformed XML. In aws-sdk-cpp 1.11.219, parse failure falls back to the HTTP
code alone: for example, an HTTP 400 `<Code>RequestTimeout</Code>` changes from
a retryable `REQUEST_TIMEOUT` error to non-retryable `UNKNOWN`, and its
message/request id are lost. The comment that the leading bytes are sufficient
is not true for an XML parser; preserve a complete document or surface overflow
without changing the provider error/retry semantics.
##########
be/src/io/fs/s3_common.h:
##########
@@ -34,12 +39,146 @@ class StringViewStream :
Aws::Utils::Stream::PreallocatedStreamBuf, public std::
std::iostream(this) {}
};
+// The AWS SDK writes the body of every response into the stream built by the
response
+// stream factory of the request, whatever the status of that response is.
Reading an
+// object range straight into the buffer of the caller therefore breaks as
soon as the
+// server answers with an error: the XML body of a `429 SlowDown` is a few
hundred bytes
+// and does not fit into the buffer of a small range read.
`PreallocatedStreamBuf` does not
+// implement `overflow()`, so the stream turns bad, curl aborts the transfer
with
+// `CURLE_WRITE_ERROR`, and the SDK reports an `INTERNAL_FAILURE` named
"Failed to flush
+// response stream" while never recording the status code of the response.
Both the retry
+// strategy of the SDK and the retry of `S3FileReader` key on that status
code, so an error
+// the server asked us to retry ends up cancelling the query instead.
+//
+// This stream buffer writes into the buffer of the caller as long as the body
fits, which
+// is the case for every successful ranged read, and spills the rest into a
buffer of its
+// own. The stream never turns bad, so the SDK reports the real status code
and can parse
+// the error out of the body.
+class S3ResponseStreamBuf final : public std::streambuf {
+public:
+ // Bodies beyond this size are truncated. Only error documents are
expected to overflow
+ // and their leading bytes already carry the error code and the message.
This bounds the
+ // memory a single failing request can hold, whatever the server answers
with.
+ static constexpr size_t MAX_SPILL_SIZE = 1024 * 1024;
+
+ S3ResponseStreamBuf(void* buf, size_t nbytes) :
_buf(static_cast<char*>(buf)) {
+ setp(_buf, _buf + nbytes);
+ setg(_buf, _buf, _buf);
+ }
+
+protected:
+ std::streamsize xsputn(const char* s, std::streamsize n) override {
+ if (!_spilled) {
+ if (n <= epptr() - pptr()) {
+ std::memcpy(pptr(), s, n);
+ pbump(static_cast<int>(n));
+ return n;
+ }
+ _spill_over();
+ }
+ // Saturating on its own: the spill is clamped when it is filled from
the buffer of
+ // the caller, and this must not underflow into an unbounded write if
it ever is not.
+ auto room = _spill.size() < MAX_SPILL_SIZE ? MAX_SPILL_SIZE -
_spill.size() : 0;
+ auto writable = std::min(static_cast<size_t>(n), room);
+ _spill.insert(_spill.end(), s, s + writable);
+ // Always report the whole write as consumed. A short write is what
makes curl
+ // abort the transfer and lose the status code of the response.
+ return n;
Review Comment:
[P1] Stop consuming an ignored-range response at the cap
When `_spill` is full, `writable` is zero but this still returns the entire
`n`, so curl keeps downloading until EOF. If a proxy ignores a 4 KiB Range and
returns a multi-GiB object, Doris now drains and discards the whole object
before the length check rejects it; the GET limiter was charged only 4 KiB and
the read metrics never account the actual transfer. Previously the fixed stream
stopped near the requested size. Bound the bytes actually consumed and
propagate an explicit oversized/ignored-range failure (without losing retry
metadata), and cover early abort plus actual-byte accounting in a test.
##########
be/src/io/fs/s3_file_reader.cpp:
##########
@@ -183,9 +183,10 @@ Status S3FileReader::read_at_impl(size_t offset, Slice
result, size_t* bytes_rea
total_sleep_time += wait_time;
continue;
Review Comment:
[P1] Keep one retry budget for recovered 429 responses
After this stream fix preserves the 429, the AWS client already exhausts
`S3CustomRetryStrategy(max_s3_client_retry)` before returning it here. This
branch then calls `get_object()` under the same limit again: with the default
10, one persistent throttle can issue 11 x 11 = 121 GETs and run two backoff
schedules, worsening the overload. The final outer iteration also sleeps even
though no next request exists, then drops the last parsed request id/message
for a generic error. Give the request one retry owner/shared budget and return
the final provider status intact; add an exhaustion test that asserts both the
attempt count and final diagnostics.
##########
be/src/io/fs/s3_common.h:
##########
@@ -34,12 +39,146 @@ class StringViewStream :
Aws::Utils::Stream::PreallocatedStreamBuf, public std::
std::iostream(this) {}
};
+// The AWS SDK writes the body of every response into the stream built by the
response
+// stream factory of the request, whatever the status of that response is.
Reading an
+// object range straight into the buffer of the caller therefore breaks as
soon as the
+// server answers with an error: the XML body of a `429 SlowDown` is a few
hundred bytes
+// and does not fit into the buffer of a small range read.
`PreallocatedStreamBuf` does not
+// implement `overflow()`, so the stream turns bad, curl aborts the transfer
with
+// `CURLE_WRITE_ERROR`, and the SDK reports an `INTERNAL_FAILURE` named
"Failed to flush
+// response stream" while never recording the status code of the response.
Both the retry
+// strategy of the SDK and the retry of `S3FileReader` key on that status
code, so an error
+// the server asked us to retry ends up cancelling the query instead.
+//
+// This stream buffer writes into the buffer of the caller as long as the body
fits, which
+// is the case for every successful ranged read, and spills the rest into a
buffer of its
+// own. The stream never turns bad, so the SDK reports the real status code
and can parse
+// the error out of the body.
+class S3ResponseStreamBuf final : public std::streambuf {
+public:
+ // Bodies beyond this size are truncated. Only error documents are
expected to overflow
+ // and their leading bytes already carry the error code and the message.
This bounds the
+ // memory a single failing request can hold, whatever the server answers
with.
+ static constexpr size_t MAX_SPILL_SIZE = 1024 * 1024;
+
+ S3ResponseStreamBuf(void* buf, size_t nbytes) :
_buf(static_cast<char*>(buf)) {
+ setp(_buf, _buf + nbytes);
+ setg(_buf, _buf, _buf);
+ }
+
+protected:
+ std::streamsize xsputn(const char* s, std::streamsize n) override {
+ if (!_spilled) {
+ if (n <= epptr() - pptr()) {
+ std::memcpy(pptr(), s, n);
+ pbump(static_cast<int>(n));
+ return n;
+ }
+ _spill_over();
+ }
+ // Saturating on its own: the spill is clamped when it is filled from
the buffer of
+ // the caller, and this must not underflow into an unbounded write if
it ever is not.
+ auto room = _spill.size() < MAX_SPILL_SIZE ? MAX_SPILL_SIZE -
_spill.size() : 0;
+ auto writable = std::min(static_cast<size_t>(n), room);
+ _spill.insert(_spill.end(), s, s + writable);
+ // Always report the whole write as consumed. A short write is what
makes curl
+ // abort the transfer and lose the status code of the response.
+ return n;
+ }
+
+ int_type overflow(int_type ch) override {
+ if (traits_type::eq_int_type(ch, traits_type::eof())) {
+ return traits_type::not_eof(ch);
+ }
+ auto c = traits_type::to_char_type(ch);
+ xsputn(&c, 1);
+ return ch;
+ }
+
+ int_type underflow() override {
+ _reset_get_area(_read_pos());
+ if (gptr() == egptr()) {
+ return traits_type::eof();
+ }
+ return traits_type::to_int_type(*gptr());
+ }
+
+ pos_type seekoff(off_type off, std::ios_base::seekdir dir,
+ std::ios_base::openmode which) override {
+ auto size = static_cast<off_type>(_written());
+ if ((which & std::ios_base::out) && !(which & std::ios_base::in)) {
+ // The SDK only asks for the write position, to tell an empty body
apart from a
+ // body it has to parse. Moving the write pointer is not supported.
+ return dir == std::ios_base::cur && off == 0 ? pos_type(size) :
pos_type(off_type(-1));
+ }
+ // A seek asking for both areas at once, which is what the default
argument of
+ // `pubseekoff()` and `pubseekpos()` does, is served as a seek of the
read area. The
+ // write area is append only, so there is nothing to move there.
+ off_type pos = off;
+ if (dir == std::ios_base::cur) {
+ pos += static_cast<off_type>(_read_pos());
+ } else if (dir == std::ios_base::end) {
+ pos += size;
+ }
+ if (pos < 0 || pos > size) {
+ return pos_type(off_type(-1));
+ }
+ _reset_get_area(static_cast<size_t>(pos));
+ return pos_type(pos);
+ }
+
+ pos_type seekpos(pos_type pos, std::ios_base::openmode which) override {
+ return seekoff(pos, std::ios_base::beg, which);
+ }
+
+private:
+ // Moves what has been written so far into the spill buffer, so that the
body stays
+ // contiguous and the SDK can parse the error out of it. Truncated right
here: the buffer
+ // of the caller is the size of the range that was asked for,
`remote_storage_read_buffer_mb`
+ // of it for a prefetched read and the whole file for a download, so it
can be far larger
+ // than the bound of the spill. Starting the spill beyond its own bound
would leave no room
+ // for the truncation to ever apply and let a server answering a ranged
read with the whole
+ // object be buffered in full.
+ void _spill_over() {
+ auto kept = std::min(static_cast<size_t>(pptr() - _buf),
MAX_SPILL_SIZE);
+ _spill.assign(_buf, _buf + kept);
+ setp(nullptr, nullptr);
+ _spilled = true;
+ }
+
+ // Bytes of the body held by this buffer, truncation excluded.
+ size_t _written() const { return _spilled ? _spill.size() : pptr() - _buf;
}
+
+ // Both areas start at the same logical offset, so the read position
survives a spill.
+ size_t _read_pos() const { return gptr() - eback(); }
+
+ void _reset_get_area(size_t pos) {
+ char* begin = _spilled ? _spill.data() : _buf;
+ auto size = _written();
+ pos = std::min(pos, size);
+ setg(begin, begin + pos, begin + size);
+ }
+
+ char* _buf;
+ std::vector<char> _spill;
Review Comment:
[P1] Charge the spill buffer to the task MemTracker
This is request-owned BE memory, but `std::vector<char>` bypasses Doris
allocator accounting. Each concurrent overflowing read can retain up to 1 MiB
(plus vector growth capacity) until AWS finishes transport/error handling, so
an error storm can consume hundreds of MiB outside the query/workload-group
limit. Use an allocator-aware container such as `DorisVector<char>` and, for
this large bounded allocation, the normal reservation/release pattern; add a
tracker-scoped test that proves the charge is released.
##########
be/test/io/fs/s3_response_stream_test.cpp:
##########
@@ -0,0 +1,189 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <gtest/gtest.h>
+
+#include <sstream>
+#include <string>
+#include <vector>
+
+#include "io/fs/s3_common.h"
+
+namespace doris {
+
+namespace {
+
+// What the SDK does with the body of a response it has to build an error from.
+std::string drain(std::iostream& stream) {
+ std::stringstream out;
+ out << stream.rdbuf();
+ return out.str();
+}
+
+// The XML body a MinIO answers a throttled ranged read with, shortened.
+constexpr char SLOW_DOWN_BODY[] =
+ R"(<?xml version="1.0"
encoding="UTF-8"?><Error><Code>SlowDown</Code><Message>Please )"
+ R"(reduce your request
rate.</Message><Key>data/packed_file/2666/x.bin</Key></Error>)";
+
+} // namespace
+
+// A body of the requested size lands in the buffer of the caller, without a
copy.
+TEST(S3ResponseStreamTest, BodyFits) {
+ std::string body(64, 'a');
+ std::vector<char> buffer(body.size());
+
+ S3ResponseStream stream(buffer.data(), buffer.size());
+ stream.write(body.data(), body.size());
+ stream.flush();
+
+ EXPECT_FALSE(stream.fail());
+ EXPECT_EQ(body, std::string(buffer.data(), buffer.size()));
+ EXPECT_EQ(static_cast<std::streampos>(body.size()), stream.tellp());
+ EXPECT_EQ(body, drain(stream));
+}
+
+// An error body larger than the range of the read leaves the stream usable,
which is what
+// keeps curl from aborting the transfer and the SDK from losing the status
code.
+TEST(S3ResponseStreamTest, ErrorBodyOverflowsInOneWrite) {
Review Comment:
[P1] Exercise the AWS retry path this change is meant to fix
These tests drive the stream directly, so they never verify the contract
that motivated the PR: curl's `tellp`/write/flush sequence, XML error
marshalling, creation of a fresh factory stream on retry, and the eventual
caller buffer. They all pass even when an overflowing response changes retry
classification, drains an ignored-range object, or triggers the nested outer
retry budget. Add a deterministic SDK-level test whose first ranged response is
an oversized 429 XML error and whose next response succeeds, plus
terminal/ignored-range cases that assert classification, attempt count, final
diagnostics, exact buffer contents, and actual bytes consumed.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
