From 0b8ac318b8c8b3d3f86e1aaddc087678fe5cfb34 Mon Sep 17 00:00:00 2001 From: Refrain Date: Sat, 1 Aug 2026 22:12:06 +0800 Subject: [PATCH] [fix](be) Report stream load size limits in MiB (#66224) Stream Load body-size limit errors display raw byte counts next to BE configuration items whose names and values are expressed in MB, making the message misleading. Report both the actual body size and configured limit in MiB for JSON Stream Load, CSV Stream Load, and HTTP Stream. The limit calculation and rejection behavior are unchanged. --- 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 e519a62779f405..8509a187014798 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_ptrbody_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_ptrbody_bytes > csv_max_body_bytes) { LOG(WARNING) << "body exceed max size." << ctx->brief(); return Status::Error( - "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(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 553f2664685a38..c391895cc5c226 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_ptrbody_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_ptrformat == TFileFormatType::FORMAT_JSON) && (ctx->body_bytes > json_max_body_bytes) && !read_json_by_line) { return Status::Error( - "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(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( - "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(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 1ed2a1307bbc8f..93dd9d044c7c8b 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(nullptr); + auto status = action._on_header(&req, ctx); + + EXPECT_TRUE(status.is()); + 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(nullptr); + auto status = action._on_header(&req, ctx); + + EXPECT_TRUE(status.is()); + 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(nullptr); + auto status = action._on_header(&req, ctx); + + EXPECT_TRUE(status.is()); + 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