From 4096f12a4f4d7344abf7707a4967514a0d950159 Mon Sep 17 00:00:00 2001 From: Zee Yin Date: Tue, 12 May 2026 17:15:25 +0800 Subject: [PATCH 1/3] feat: add support for CRC32C checksum in S3 uploads and disable Content-MD5 for S3 Express --- be/src/common/config.cpp | 2 + be/src/common/config.h | 6 +++ be/src/io/fs/s3_obj_storage_client.cpp | 51 ++++++++++++++++++++++++-- be/src/io/fs/s3_obj_storage_client.h | 7 +++- be/src/util/s3_util.cpp | 3 +- 5 files changed, 63 insertions(+), 6 deletions(-) diff --git a/be/src/common/config.cpp b/be/src/common/config.cpp index 9ec587bd100ec5..5fa1fa6091973d 100644 --- a/be/src/common/config.cpp +++ b/be/src/common/config.cpp @@ -1512,6 +1512,8 @@ DEFINE_mInt32(s3_read_max_wait_time_ms, "800"); DEFINE_mBool(enable_s3_object_check_after_upload, "true"); DEFINE_mInt32(aws_client_request_timeout_ms, "30000"); +DEFINE_mBool(s3_disable_content_md5, "false"); + DEFINE_mBool(enable_s3_rate_limiter, "false"); DEFINE_mInt64(s3_get_bucket_tokens, "1000000000000000000"); DEFINE_Validator(s3_get_bucket_tokens, [](int64_t config) -> bool { return config > 0; }); diff --git a/be/src/common/config.h b/be/src/common/config.h index 012f09a735c0cd..c9a6241b45882f 100644 --- a/be/src/common/config.h +++ b/be/src/common/config.h @@ -1621,6 +1621,12 @@ DECLARE_mInt32(s3_read_max_wait_time_ms); DECLARE_mBool(enable_s3_object_check_after_upload); DECLARE_mInt32(aws_client_request_timeout_ms); +// When true, omit the Content-MD5 header on S3 PutObject / UploadPart and send a +// CRC32C checksum instead. Required for S3 Express One Zone, which returns 501 +// NotImplemented for Content-MD5. Endpoints containing "s3express" auto-enable +// this regardless of the flag. +DECLARE_mBool(s3_disable_content_md5); + // write as inverted index tmp directory DECLARE_String(tmp_file_dir); diff --git a/be/src/io/fs/s3_obj_storage_client.cpp b/be/src/io/fs/s3_obj_storage_client.cpp index 1b8bf7b473c076..3950f23bb034a9 100644 --- a/be/src/io/fs/s3_obj_storage_client.cpp +++ b/be/src/io/fs/s3_obj_storage_client.cpp @@ -34,6 +34,7 @@ #include #include #include +#include #include #include #include @@ -61,9 +62,12 @@ #include #include +#include + #include #include +#include "common/config.h" #include "common/logging.h" #include "common/status.h" #include "cpp/sync_point.h" @@ -116,6 +120,31 @@ using namespace Aws::S3::Model; static constexpr int S3_REQUEST_THRESHOLD_MS = 5000; +namespace { +// S3 Express One Zone endpoints follow the pattern +// "*.s3express-..amazonaws.com" — substring match is sufficient +// for both the gateway and the s3express subdomain forms. +bool is_s3_express_endpoint(const std::string& endpoint) { + return endpoint.find("s3express") != std::string::npos; +} + +// AWS expects the CRC32C value as the big-endian 4-byte representation, base64-encoded. +Aws::String compute_crc32c_b64(std::string_view data) { + uint32_t crc = crc32c::Crc32c(reinterpret_cast(data.data()), data.size()); + uint8_t bytes[4] = {static_cast((crc >> 24) & 0xff), + static_cast((crc >> 16) & 0xff), + static_cast((crc >> 8) & 0xff), + static_cast(crc & 0xff)}; + return Aws::Utils::HashingUtils::Base64Encode(Aws::Utils::ByteBuffer(bytes, sizeof(bytes))); +} +} // namespace + +S3ObjStorageClient::S3ObjStorageClient(std::shared_ptr client, + const std::string& endpoint) + : _client(std::move(client)), + _disable_content_md5(config::s3_disable_content_md5 || + is_s3_express_endpoint(endpoint)) {} + ObjectStorageUploadResponse S3ObjStorageClient::create_multipart_upload( const ObjectStoragePathOptions& opts) { CreateMultipartUploadRequest request; @@ -158,8 +187,15 @@ ObjectStorageResponse S3ObjStorageClient::put_object(const ObjectStoragePathOpti Aws::S3::Model::PutObjectRequest request; request.WithBucket(opts.bucket).WithKey(opts.key); auto string_view_stream = std::make_shared(stream.data(), stream.size()); - Aws::Utils::ByteBuffer part_md5(Aws::Utils::HashingUtils::CalculateMD5(*string_view_stream)); - request.SetContentMD5(Aws::Utils::HashingUtils::Base64Encode(part_md5)); + if (_disable_content_md5) { + // S3 Express One Zone rejects Content-MD5; use CRC32C instead. + request.SetChecksumAlgorithm(ChecksumAlgorithm::CRC32C); + request.SetChecksumCRC32C(compute_crc32c_b64(stream)); + } else { + Aws::Utils::ByteBuffer part_md5( + Aws::Utils::HashingUtils::CalculateMD5(*string_view_stream)); + request.SetContentMD5(Aws::Utils::HashingUtils::Base64Encode(part_md5)); + } request.SetBody(string_view_stream); request.SetContentLength(stream.size()); request.SetContentType("application/octet-stream"); @@ -202,8 +238,15 @@ ObjectStorageUploadResponse S3ObjStorageClient::upload_part(const ObjectStorageP request.SetBody(string_view_stream); - Aws::Utils::ByteBuffer part_md5(Aws::Utils::HashingUtils::CalculateMD5(*string_view_stream)); - request.SetContentMD5(Aws::Utils::HashingUtils::Base64Encode(part_md5)); + if (_disable_content_md5) { + // S3 Express One Zone rejects Content-MD5; use CRC32C instead. + request.SetChecksumAlgorithm(ChecksumAlgorithm::CRC32C); + request.SetChecksumCRC32C(compute_crc32c_b64(stream)); + } else { + Aws::Utils::ByteBuffer part_md5( + Aws::Utils::HashingUtils::CalculateMD5(*string_view_stream)); + request.SetContentMD5(Aws::Utils::HashingUtils::Base64Encode(part_md5)); + } request.SetContentLength(stream.size()); request.SetContentType("application/octet-stream"); diff --git a/be/src/io/fs/s3_obj_storage_client.h b/be/src/io/fs/s3_obj_storage_client.h index 45294226594d81..ac048a65a98238 100644 --- a/be/src/io/fs/s3_obj_storage_client.h +++ b/be/src/io/fs/s3_obj_storage_client.h @@ -32,7 +32,8 @@ class ObjClientHolder; class S3ObjStorageClient final : public ObjStorageClient { public: - S3ObjStorageClient(std::shared_ptr client) : _client(std::move(client)) {} + explicit S3ObjStorageClient(std::shared_ptr client, + const std::string& endpoint = {}); ~S3ObjStorageClient() override = default; ObjectStorageUploadResponse create_multipart_upload( const ObjectStoragePathOptions& opts) override; @@ -58,6 +59,10 @@ class S3ObjStorageClient final : public ObjStorageClient { private: std::shared_ptr _client; + // True for S3 Express One Zone endpoints (or when config::s3_disable_content_md5 + // is on). When set, uploads send a CRC32C checksum instead of Content-MD5, + // since S3 Express returns 501 NotImplemented for the latter. + bool _disable_content_md5 = false; }; } // namespace doris::io diff --git a/be/src/util/s3_util.cpp b/be/src/util/s3_util.cpp index fc4c81d4bd96f2..6218c524362fe8 100644 --- a/be/src/util/s3_util.cpp +++ b/be/src/util/s3_util.cpp @@ -532,7 +532,8 @@ std::shared_ptr S3ClientFactory::_create_s3_client( Aws::Client::AWSAuthV4Signer::PayloadSigningPolicy::Never, s3_conf.use_virtual_addressing); - auto obj_client = std::make_shared(std::move(new_client)); + auto obj_client = + std::make_shared(std::move(new_client), s3_conf.endpoint); LOG_INFO("create one s3 client with {}", s3_conf.to_string()); return obj_client; } From 98b7adffee8639c42fc448a30d3ae49aa46f249a Mon Sep 17 00:00:00 2001 From: Zee Yin Date: Thu, 14 May 2026 14:21:04 +0800 Subject: [PATCH 2/3] Upgrade AWS SDK and use new create client to support S3 express --- be/src/util/s3_util.cpp | 28 ++++++++++++++++++++-------- common/cpp/aws_logger.h | 9 +++++++++ thirdparty/vars.sh | 8 ++++---- 3 files changed, 33 insertions(+), 12 deletions(-) diff --git a/be/src/util/s3_util.cpp b/be/src/util/s3_util.cpp index 6218c524362fe8..3fd802f5b8fe9d 100644 --- a/be/src/util/s3_util.cpp +++ b/be/src/util/s3_util.cpp @@ -491,11 +491,13 @@ std::shared_ptr S3ClientFactory::_create_s3_client( TEST_SYNC_POINT_RETURN_WITH_VALUE( "s3_client_factory::create", std::make_shared(std::make_shared())); - Aws::Client::ClientConfiguration aws_config = S3ClientFactory::getClientConfiguration(); - if (s3_conf.need_override_endpoint) { - aws_config.endpointOverride = s3_conf.endpoint; - } + // Use S3ClientConfiguration (not the legacy ClientConfiguration) so the SDK's + // endpoint-rules resolver is active. This is required for S3 Express One Zone: + // the resolver detects the --x-s3 bucket suffix, calls CreateSession, and signs + // requests with service name "s3express" instead of "s3". + Aws::S3::S3ClientConfiguration aws_config; aws_config.region = s3_conf.region; + aws_config.useVirtualAddressing = s3_conf.use_virtual_addressing; if (_ca_cert_file_path.empty()) { _ca_cert_file_path = get_valid_ca_cert_path(doris::split(config::ca_cert_file_paths, ";")); @@ -527,10 +529,20 @@ std::shared_ptr S3ClientFactory::_create_s3_client( aws_config.retryStrategy = std::make_shared( config::max_s3_client_retry /*scaleFactor = 25*/, /*retry_slow_down=*/false); - std::shared_ptr new_client = std::make_shared( - get_aws_credentials_provider(s3_conf), std::move(aws_config), - Aws::Client::AWSAuthV4Signer::PayloadSigningPolicy::Never, - s3_conf.use_virtual_addressing); + // S3 Express buckets are identified by the --x-s3 suffix or an s3express endpoint. + // Skip endpointOverride for these so the SDK resolves the bucket-specific endpoint + // and manages CreateSession automatically. For all other buckets, keep the override. + const bool is_s3_express = s3_conf.endpoint.find("s3express") != std::string::npos || + s3_conf.bucket.find("--x-s3") != std::string::npos; + if (s3_conf.need_override_endpoint && !is_s3_express) { + aws_config.endpointOverride = s3_conf.endpoint; + } + aws_config.disableS3ExpressAuth = !is_s3_express; + + auto new_client = std::make_shared( + get_aws_credentials_provider(s3_conf), + Aws::MakeShared("S3Client"), + aws_config); auto obj_client = std::make_shared(std::move(new_client), s3_conf.endpoint); diff --git a/common/cpp/aws_logger.h b/common/cpp/aws_logger.h index 85734d13e1411c..9f06cec80fac0f 100644 --- a/common/cpp/aws_logger.h +++ b/common/cpp/aws_logger.h @@ -20,6 +20,8 @@ #include #include #include +#include +#include class DorisAWSLogger final : public Aws::Utils::Logging::LogSystemInterface { public: @@ -36,6 +38,13 @@ class DorisAWSLogger final : public Aws::Utils::Logging::LogSystemInterface { _log_impl(log_level, tag, message_stream.str().c_str()); } + void vaLog(Aws::Utils::Logging::LogLevel log_level, const char* tag, const char* format_str, + va_list args) final { + char buf[4096]; + vsnprintf(buf, sizeof(buf), format_str, args); + _log_impl(log_level, tag, buf); + } + void Flush() final {} private: diff --git a/thirdparty/vars.sh b/thirdparty/vars.sh index af46e566b8a30f..1b4ca87919f81b 100644 --- a/thirdparty/vars.sh +++ b/thirdparty/vars.sh @@ -356,10 +356,10 @@ BOOTSTRAP_TABLE_CSS_FILE="bootstrap-table.min.css" BOOTSTRAP_TABLE_CSS_MD5SUM="23389d4456da412e36bae30c469a766a" # aws sdk -AWS_SDK_DOWNLOAD="https://github.com/aws/aws-sdk-cpp/archive/refs/tags/1.11.219.tar.gz" -AWS_SDK_NAME="aws-sdk-cpp-1.11.219.tar.gz" -AWS_SDK_SOURCE="aws-sdk-cpp-1.11.219" -AWS_SDK_MD5SUM="80aa616efe1a3e7a9bf0dfbc44a97864" +AWS_SDK_DOWNLOAD="https://github.com/aws/aws-sdk-cpp/archive/refs/tags/1.11.400.tar.gz" +AWS_SDK_NAME="aws-sdk-cpp-1.11.400.tar.gz" +AWS_SDK_SOURCE="aws-sdk-cpp-1.11.400" +AWS_SDK_MD5SUM="329c46612df55ba6715d9d2ccd61dc03" # tsan_header TSAN_HEADER_DOWNLOAD="https://gcc.gnu.org/git/?p=gcc.git;a=blob_plain;f=libsanitizer/include/sanitizer/tsan_interface_atomic.h;hb=refs/heads/releases/gcc-7" From 0a193178c855d220f0817e26de0efd5003be6a4d Mon Sep 17 00:00:00 2001 From: Refrain Date: Mon, 13 Jul 2026 09:26:42 +0800 Subject: [PATCH 3/3] [fix](be) Detect S3 Express context accurately ### What problem does this PR solve?\n\nIssue Number: close #xxx\n\nRelated PR: #63409\n\nProblem Summary: The initial S3 Express implementation detected directory buckets and endpoints with broad substring matches in two separate locations. This could classify ordinary bucket names or endpoint paths as S3 Express and allowed the client configuration and upload checksum behavior to diverge. Centralize the decision in one helper, require the documented --x-s3 bucket suffix or an s3express endpoint host label, and pass the resulting context into the object storage client. Add unit coverage for valid and invalid names.\n\n### Release note\n\nNone\n\n### Check List (For Author)\n\n- Test: Unit Test (S3UTILTest.is_s3_express_context; ASAN UT rebuild started)\n- Behavior changed: No, except avoiding false-positive S3 Express detection\n- Does this need documentation: No --- be/src/io/fs/s3_obj_storage_client.cpp | 18 ++++------------- be/src/io/fs/s3_obj_storage_client.h | 2 +- be/src/util/s3_util.cpp | 28 ++++++++++++++++++-------- be/src/util/s3_util.h | 1 + be/test/util/s3_util_test.cpp | 8 ++++++++ 5 files changed, 34 insertions(+), 23 deletions(-) diff --git a/be/src/io/fs/s3_obj_storage_client.cpp b/be/src/io/fs/s3_obj_storage_client.cpp index 3950f23bb034a9..1d1f146cd6244f 100644 --- a/be/src/io/fs/s3_obj_storage_client.cpp +++ b/be/src/io/fs/s3_obj_storage_client.cpp @@ -61,7 +61,6 @@ #include #include #include - #include #include @@ -121,29 +120,20 @@ using namespace Aws::S3::Model; static constexpr int S3_REQUEST_THRESHOLD_MS = 5000; namespace { -// S3 Express One Zone endpoints follow the pattern -// "*.s3express-..amazonaws.com" — substring match is sufficient -// for both the gateway and the s3express subdomain forms. -bool is_s3_express_endpoint(const std::string& endpoint) { - return endpoint.find("s3express") != std::string::npos; -} - // AWS expects the CRC32C value as the big-endian 4-byte representation, base64-encoded. Aws::String compute_crc32c_b64(std::string_view data) { uint32_t crc = crc32c::Crc32c(reinterpret_cast(data.data()), data.size()); uint8_t bytes[4] = {static_cast((crc >> 24) & 0xff), static_cast((crc >> 16) & 0xff), - static_cast((crc >> 8) & 0xff), - static_cast(crc & 0xff)}; + static_cast((crc >> 8) & 0xff), static_cast(crc & 0xff)}; return Aws::Utils::HashingUtils::Base64Encode(Aws::Utils::ByteBuffer(bytes, sizeof(bytes))); } } // namespace S3ObjStorageClient::S3ObjStorageClient(std::shared_ptr client, - const std::string& endpoint) + bool is_s3_express) : _client(std::move(client)), - _disable_content_md5(config::s3_disable_content_md5 || - is_s3_express_endpoint(endpoint)) {} + _disable_content_md5(config::s3_disable_content_md5 || is_s3_express) {} ObjectStorageUploadResponse S3ObjStorageClient::create_multipart_upload( const ObjectStoragePathOptions& opts) { @@ -545,4 +535,4 @@ std::string S3ObjStorageClient::generate_presigned_url(const ObjectStoragePathOp expiration_secs); } -} // namespace doris::io \ No newline at end of file +} // namespace doris::io diff --git a/be/src/io/fs/s3_obj_storage_client.h b/be/src/io/fs/s3_obj_storage_client.h index ac048a65a98238..0216318936e4f4 100644 --- a/be/src/io/fs/s3_obj_storage_client.h +++ b/be/src/io/fs/s3_obj_storage_client.h @@ -33,7 +33,7 @@ class ObjClientHolder; class S3ObjStorageClient final : public ObjStorageClient { public: explicit S3ObjStorageClient(std::shared_ptr client, - const std::string& endpoint = {}); + bool is_s3_express = false); ~S3ObjStorageClient() override = default; ObjectStorageUploadResponse create_multipart_upload( const ObjectStoragePathOptions& opts) override; diff --git a/be/src/util/s3_util.cpp b/be/src/util/s3_util.cpp index 3fd802f5b8fe9d..028613f5533dc4 100644 --- a/be/src/util/s3_util.cpp +++ b/be/src/util/s3_util.cpp @@ -532,20 +532,17 @@ std::shared_ptr S3ClientFactory::_create_s3_client( // S3 Express buckets are identified by the --x-s3 suffix or an s3express endpoint. // Skip endpointOverride for these so the SDK resolves the bucket-specific endpoint // and manages CreateSession automatically. For all other buckets, keep the override. - const bool is_s3_express = s3_conf.endpoint.find("s3express") != std::string::npos || - s3_conf.bucket.find("--x-s3") != std::string::npos; - if (s3_conf.need_override_endpoint && !is_s3_express) { + const bool express = is_s3_express(s3_conf.bucket, s3_conf.endpoint); + if (s3_conf.need_override_endpoint && !express) { aws_config.endpointOverride = s3_conf.endpoint; } - aws_config.disableS3ExpressAuth = !is_s3_express; + aws_config.disableS3ExpressAuth = !express; auto new_client = std::make_shared( get_aws_credentials_provider(s3_conf), - Aws::MakeShared("S3Client"), - aws_config); + Aws::MakeShared("S3Client"), aws_config); - auto obj_client = - std::make_shared(std::move(new_client), s3_conf.endpoint); + auto obj_client = std::make_shared(std::move(new_client), express); LOG_INFO("create one s3 client with {}", s3_conf.to_string()); return obj_client; } @@ -807,4 +804,19 @@ std::string hide_access_key(const std::string& ak) { return key; } +bool is_s3_express(std::string_view bucket, std::string_view endpoint) { + constexpr std::string_view bucket_suffix = "--x-s3"; + if (bucket.ends_with(bucket_suffix)) { + return true; + } + + const auto scheme_pos = endpoint.find("://"); + if (scheme_pos != std::string_view::npos) { + endpoint.remove_prefix(scheme_pos + 3); + } + endpoint = endpoint.substr(0, endpoint.find('/')); + return endpoint.starts_with("s3express-") || + endpoint.find(".s3express-") != std::string_view::npos; +} + } // end namespace doris diff --git a/be/src/util/s3_util.h b/be/src/util/s3_util.h index 5546db857c91b1..950600fe8a534f 100644 --- a/be/src/util/s3_util.h +++ b/be/src/util/s3_util.h @@ -63,6 +63,7 @@ extern bvar::LatencyRecorder s3_copy_object_latency; }; // namespace s3_bvar std::string hide_access_key(const std::string& ak); +bool is_s3_express(std::string_view bucket, std::string_view endpoint); int reset_s3_rate_limiter(S3RateLimitType type, size_t max_speed, size_t max_burst, size_t limit); // Rebuild the S3 GET/PUT rate limiters if the related configs have changed. // Safe to call periodically; it is a no-op when nothing changed. diff --git a/be/test/util/s3_util_test.cpp b/be/test/util/s3_util_test.cpp index a87d0e9a2d2980..cba03bec576773 100644 --- a/be/test/util/s3_util_test.cpp +++ b/be/test/util/s3_util_test.cpp @@ -77,6 +77,14 @@ TEST_F(S3UTILTest, hide_access_key_typical_aws_key) { EXPECT_EQ("xxxxxxxFODNN7xxxxxxx", result); } +TEST_F(S3UTILTest, is_s3_express_context) { + EXPECT_TRUE(is_s3_express("bucket--use1-az4--x-s3", "")); + EXPECT_TRUE(is_s3_express("", "bucket.s3express-use1-az4.us-east-1.amazonaws.com")); + EXPECT_FALSE(is_s3_express("bucket--x-s3-suffix", "")); + EXPECT_FALSE(is_s3_express("bucket", "https://example.com/s3express/path")); + EXPECT_FALSE(is_s3_express("bucket", "s3.us-east-1.amazonaws.com")); +} + // Verifies that check_s3_rate_limiter_config_changed() rebuilds the global GET rate // limiter when the related configs change. This is the behavior the cloud vault refresh // thread relies on to apply dynamically modified s3_get_* rate limiter configs without