This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new af495a9bf0c branch-4.1: [fix](be) Report stream load size limits in
MiB #66224 (#66351)
af495a9bf0c is described below
commit af495a9bf0cdaee76a4c0f722f0b98668864154a
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Wed Aug 5 16:29:40 2026 +0800
branch-4.1: [fix](be) Report stream load size limits in MiB #66224 (#66351)
Cherry-picked from #66224
Co-authored-by: Refrain <[email protected]>
---
be/src/service/http/action/http_stream.cpp | 13 ++++--
be/src/service/http/action/stream_load.cpp | 23 ++++++----
be/test/service/http/stream_load_test.cpp | 74 +++++++++++++++++++++++++++++-
3 files changed, 97 insertions(+), 13 deletions(-)
diff --git a/be/src/service/http/action/http_stream.cpp
b/be/src/service/http/action/http_stream.cpp
index e519a62779f..8509a187014 100644
--- a/be/src/service/http/action/http_stream.cpp
+++ b/be/src/service/http/action/http_stream.cpp
@@ -67,6 +67,8 @@ using namespace ErrorCode;
namespace {
+constexpr size_t MEBIBYTE = 1024 * 1024;
+
bool is_compressed_file_scan(const TPipelineFragmentParams& params) {
if (!params.__isset.file_scan_params) {
return false;
@@ -231,7 +233,8 @@ Status HttpStreamAction::_on_header(HttpRequest* http_req,
std::shared_ptr<Strea
// TODO(zs) : need Need to request an FE to obtain information such as
format
// check content length
ctx->body_bytes = 0;
- size_t csv_max_body_bytes = config::streaming_load_max_mb * 1024 * 1024;
+ const auto csv_max_body_mb = config::streaming_load_max_mb;
+ size_t csv_max_body_bytes = csv_max_body_mb * MEBIBYTE;
if (!http_req->header(HttpHeaders::CONTENT_LENGTH).empty()) {
try {
ctx->body_bytes =
std::stol(http_req->header(HttpHeaders::CONTENT_LENGTH));
@@ -243,9 +246,11 @@ Status HttpStreamAction::_on_header(HttpRequest* http_req,
std::shared_ptr<Strea
if (ctx->body_bytes > csv_max_body_bytes) {
LOG(WARNING) << "body exceed max size." << ctx->brief();
return Status::Error<ErrorCode::EXCEEDED_LIMIT>(
- "body size {} exceed BE's conf `streaming_load_max_mb` {}.
increase it if you "
- "are sure this load is reasonable",
- ctx->body_bytes, csv_max_body_bytes);
+ "body size {} bytes ({:.2f} MiB) exceeds the limit of {}
bytes ({} MiB) set "
+ "by BE config `streaming_load_max_mb`. Increase it if you
are sure this load "
+ "is reasonable",
+ ctx->body_bytes, static_cast<double>(ctx->body_bytes) /
MEBIBYTE,
+ csv_max_body_bytes, csv_max_body_mb);
}
}
diff --git a/be/src/service/http/action/stream_load.cpp
b/be/src/service/http/action/stream_load.cpp
index 553f2664685..c391895cc5c 100644
--- a/be/src/service/http/action/stream_load.cpp
+++ b/be/src/service/http/action/stream_load.cpp
@@ -81,6 +81,7 @@ bvar::LatencyRecorder
g_stream_load_commit_and_publish_latency_ms("stream_load",
"commit_and_publish_ms");
static constexpr size_t MIN_CHUNK_SIZE = 64 * 1024;
+static constexpr size_t MEBIBYTE = 1024 * 1024;
static const std::string CHUNK = "chunked";
static const std::string OFF_MODE = "off_mode";
static const std::string SYNC_MODE = "sync_mode";
@@ -290,8 +291,10 @@ Status StreamLoadAction::_on_header(HttpRequest* http_req,
std::shared_ptr<Strea
// check content length
ctx->body_bytes = 0;
- size_t csv_max_body_bytes = config::streaming_load_max_mb * 1024 * 1024;
- size_t json_max_body_bytes = config::streaming_load_json_max_mb * 1024 *
1024;
+ const auto csv_max_body_mb = config::streaming_load_max_mb;
+ size_t csv_max_body_bytes = csv_max_body_mb * MEBIBYTE;
+ const auto json_max_body_mb = config::streaming_load_json_max_mb;
+ size_t json_max_body_bytes = json_max_body_mb * MEBIBYTE;
bool read_json_by_line = false;
if (!http_req->header(HTTP_READ_JSON_BY_LINE).empty()) {
if (iequal(http_req->header(HTTP_READ_JSON_BY_LINE), "true")) {
@@ -309,17 +312,21 @@ Status StreamLoadAction::_on_header(HttpRequest*
http_req, std::shared_ptr<Strea
if ((ctx->format == TFileFormatType::FORMAT_JSON) &&
(ctx->body_bytes > json_max_body_bytes) && !read_json_by_line) {
return Status::Error<ErrorCode::EXCEEDED_LIMIT>(
- "json body size {} exceed BE's conf
`streaming_load_json_max_mb` {}. increase "
- "it if you are sure this load is reasonable",
- ctx->body_bytes, json_max_body_bytes);
+ "json body size {} bytes ({:.2f} MiB) exceeds the limit of
{} bytes ({} MiB) "
+ "set by BE's conf streaming_load_json_max_mb. Increase it
if you are sure "
+ "this load is reasonable",
+ ctx->body_bytes, static_cast<double>(ctx->body_bytes) /
MEBIBYTE,
+ json_max_body_bytes, json_max_body_mb);
}
// csv max body size
else if (ctx->body_bytes > csv_max_body_bytes) {
LOG(WARNING) << "body exceed max size." << ctx->brief();
return Status::Error<ErrorCode::EXCEEDED_LIMIT>(
- "body size {} exceed BE's conf `streaming_load_max_mb` {}.
increase it if you "
- "are sure this load is reasonable",
- ctx->body_bytes, csv_max_body_bytes);
+ "body size {} bytes ({:.2f} MiB) exceeds the limit of {}
bytes ({} MiB) set "
+ "by BE's conf streaming_load_max_mb. Increase it if you
are sure this load is "
+ "reasonable",
+ ctx->body_bytes, static_cast<double>(ctx->body_bytes) /
MEBIBYTE,
+ csv_max_body_bytes, csv_max_body_mb);
}
} else {
#ifndef BE_TEST
diff --git a/be/test/service/http/stream_load_test.cpp
b/be/test/service/http/stream_load_test.cpp
index 1ed2a1307bb..93dd9d044c7 100644
--- a/be/test/service/http/stream_load_test.cpp
+++ b/be/test/service/http/stream_load_test.cpp
@@ -28,7 +28,9 @@
#include "event2/http_struct.h"
#include "evhttp.h"
#include "load/group_commit/wal/wal_manager.h"
+#include "load/stream_load/stream_load_context.h"
#include "runtime/exec_env.h"
+#include "service/http/action/http_stream.h"
#include "service/http/ev_http_server.h"
#include "service/http/http_channel.h"
#include "service/http/http_common.h"
@@ -37,6 +39,7 @@
#include "service/http/http_headers.h"
#include "service/http/http_request.h"
#include "service/http/utils.h"
+#include "util/defer_op.h"
namespace doris {
@@ -123,4 +126,73 @@ TEST_F(StreamLoadTest, TestHeader) {
evhttp_request_free(evhttp_req);
}
}
-} // namespace doris
\ No newline at end of file
+
+TEST_F(StreamLoadTest, JsonBodySizeLimitPlusOneErrorIncludesExactBytes) {
+ const auto original_json_max_mb = config::streaming_load_json_max_mb;
+ Defer restore_json_max_mb {
+ [original_json_max_mb] { config::streaming_load_json_max_mb =
original_json_max_mb; }};
+ config::streaming_load_json_max_mb = 100;
+
+ auto* evhttp_req = evhttp_request_new(nullptr, nullptr);
+ HttpRequest req(evhttp_req);
+ req.set_header(HttpHeaders::AUTHORIZATION, "Basic cm9vdDo=");
+ req.set_header(HTTP_FORMAT_KEY, "json");
+ req.set_header(HttpHeaders::CONTENT_LENGTH, "104857601");
+
+ StreamLoadAction action(nullptr);
+ auto ctx = std::make_shared<StreamLoadContext>(nullptr);
+ auto status = action._on_header(&req, ctx);
+
+ EXPECT_TRUE(status.is<ErrorCode::EXCEEDED_LIMIT>());
+ EXPECT_EQ(status.to_string_no_stack(),
+ "[E-217]json body size 104857601 bytes (100.00 MiB) exceeds the
limit of 104857600 "
+ "bytes (100 MiB) set by BE's conf streaming_load_json_max_mb.
Increase it if you "
+ "are sure this load is reasonable");
+ evhttp_request_free(evhttp_req);
+}
+
+TEST_F(StreamLoadTest, CsvBodySizeLimitPlusOneErrorIncludesExactBytes) {
+ const auto original_max_mb = config::streaming_load_max_mb;
+ Defer restore_max_mb {[original_max_mb] { config::streaming_load_max_mb =
original_max_mb; }};
+ config::streaming_load_max_mb = 100;
+
+ auto* evhttp_req = evhttp_request_new(nullptr, nullptr);
+ HttpRequest req(evhttp_req);
+ req.set_header(HttpHeaders::AUTHORIZATION, "Basic cm9vdDo=");
+ req.set_header(HTTP_FORMAT_KEY, "csv");
+ req.set_header(HttpHeaders::CONTENT_LENGTH, "104857601");
+
+ StreamLoadAction action(nullptr);
+ auto ctx = std::make_shared<StreamLoadContext>(nullptr);
+ auto status = action._on_header(&req, ctx);
+
+ EXPECT_TRUE(status.is<ErrorCode::EXCEEDED_LIMIT>());
+ EXPECT_EQ(status.to_string_no_stack(),
+ "[E-217]body size 104857601 bytes (100.00 MiB) exceeds the limit
of 104857600 bytes "
+ "(100 MiB) set by BE's conf streaming_load_max_mb. Increase it
if you are sure this "
+ "load is reasonable");
+ evhttp_request_free(evhttp_req);
+}
+
+TEST_F(StreamLoadTest, HttpStreamBodySizeLimitPlusOneErrorIncludesExactBytes) {
+ const auto original_max_mb = config::streaming_load_max_mb;
+ Defer restore_max_mb {[original_max_mb] { config::streaming_load_max_mb =
original_max_mb; }};
+ config::streaming_load_max_mb = 100;
+
+ auto* evhttp_req = evhttp_request_new(nullptr, nullptr);
+ HttpRequest req(evhttp_req);
+ req.set_header(HttpHeaders::AUTHORIZATION, "Basic cm9vdDo=");
+ req.set_header(HttpHeaders::CONTENT_LENGTH, "104857601");
+
+ HttpStreamAction action(nullptr);
+ auto ctx = std::make_shared<StreamLoadContext>(nullptr);
+ auto status = action._on_header(&req, ctx);
+
+ EXPECT_TRUE(status.is<ErrorCode::EXCEEDED_LIMIT>());
+ EXPECT_EQ(status.to_string_no_stack(),
+ "[E-217]body size 104857601 bytes (100.00 MiB) exceeds the limit
of 104857600 bytes "
+ "(100 MiB) set by BE config `streaming_load_max_mb`. Increase it
if you are sure "
+ "this load is reasonable");
+ evhttp_request_free(evhttp_req);
+}
+} // namespace doris
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
