[feature](s3) Support S3 Express One Zone

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; });

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);
 

be/src/io/fs/s3_obj_storage_client.cpp

@@ -34,6 +34,7 @@
 #include <aws/s3/S3Errors.h>
 #include <aws/s3/model/AbortMultipartUploadRequest.h>
 #include <aws/s3/model/AbortMultipartUploadResult.h>
+#include <aws/s3/model/ChecksumAlgorithm.h>
 #include <aws/s3/model/CompleteMultipartUploadRequest.h>
 #include <aws/s3/model/CompleteMultipartUploadResult.h>
 #include <aws/s3/model/CompletedMultipartUpload.h>
@@ -60,10 +61,12 @@
 #include <aws/s3/model/PutObjectResult.h>
 #include <aws/s3/model/UploadPartRequest.h>
 #include <aws/s3/model/UploadPartResult.h>
+#include <crc32c/crc32c.h>
 
 #include <algorithm>
 #include <ranges>
 
+#include "common/config.h"
 #include "common/logging.h"
 #include "common/status.h"
 #include "cpp/sync_point.h"
@@ -116,6 +119,22 @@ using namespace Aws::S3::Model;
 
 static constexpr int S3_REQUEST_THRESHOLD_MS = 5000;
 
+namespace {
+// 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<const uint8_t*>(data.data()), data.size());
+    uint8_t bytes[4] = {static_cast<uint8_t>((crc >> 24) & 0xff),
+                        static_cast<uint8_t>((crc >> 16) & 0xff),
+                        static_cast<uint8_t>((crc >> 8) & 0xff), static_cast<uint8_t>(crc & 0xff)};
+    return Aws::Utils::HashingUtils::Base64Encode(Aws::Utils::ByteBuffer(bytes, sizeof(bytes)));
+}
+} // namespace
+
+S3ObjStorageClient::S3ObjStorageClient(std::shared_ptr<Aws::S3::S3Client> client,
+                                       bool is_s3_express)
+        : _client(std::move(client)),
+          _disable_content_md5(config::s3_disable_content_md5 || is_s3_express) {}
+
 ObjectStorageUploadResponse S3ObjStorageClient::create_multipart_upload(
         const ObjectStoragePathOptions& opts) {
     CreateMultipartUploadRequest request;
@@ -158,8 +177,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<StringViewStream>(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 +228,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");
@@ -502,4 +535,4 @@ std::string S3ObjStorageClient::generate_presigned_url(const ObjectStoragePathOp
                                          expiration_secs);
 }
 
-} // namespace doris::io
+} // namespace doris::io

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<Aws::S3::S3Client> client) : _client(std::move(client)) {}
+    explicit S3ObjStorageClient(std::shared_ptr<Aws::S3::S3Client> client,
+                                bool is_s3_express = false);
     ~S3ObjStorageClient() override = default;
     ObjectStorageUploadResponse create_multipart_upload(
             const ObjectStoragePathOptions& opts) override;
@@ -58,6 +59,10 @@ class S3ObjStorageClient final : public ObjStorageClient {
 
 private:
     std::shared_ptr<Aws::S3::S3Client> _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

be/src/util/s3_util.cpp

@@ -491,11 +491,13 @@ std::shared_ptr<io::ObjStorageClient> S3ClientFactory::_create_s3_client(
     TEST_SYNC_POINT_RETURN_WITH_VALUE(
             "s3_client_factory::create",
             std::make_shared<io::S3ObjStorageClient>(std::make_shared<Aws::S3::S3Client>()));
-    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,12 +529,20 @@ std::shared_ptr<io::ObjStorageClient> S3ClientFactory::_create_s3_client(
     aws_config.retryStrategy = std::make_shared<S3CustomRetryStrategy>(
             config::max_s3_client_retry /*scaleFactor = 25*/, /*retry_slow_down=*/false);
 
-    std::shared_ptr<Aws::S3::S3Client> new_client = std::make_shared<Aws::S3::S3Client>(
-            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 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 = !express;
+
+    auto new_client = std::make_shared<Aws::S3::S3Client>(
+            get_aws_credentials_provider(s3_conf),
+            Aws::MakeShared<Aws::S3::S3EndpointProvider>("S3Client"), aws_config);
 
-    auto obj_client = std::make_shared<io::S3ObjStorageClient>(std::move(new_client));
+    auto obj_client = std::make_shared<io::S3ObjStorageClient>(std::move(new_client), express);
     LOG_INFO("create one s3 client with {}", s3_conf.to_string());
     return obj_client;
 }
@@ -794,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

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.

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

common/cpp/aws_logger.h

@@ -20,6 +20,8 @@
 #include <aws/core/utils/logging/LogLevel.h>
 #include <aws/core/utils/logging/LogSystemInterface.h>
 #include <glog/logging.h>
+#include <cstdarg>
+#include <cstdio>
 
 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:

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"