[feature](s3) Support S3 Express One Zone

be/src/util/s3_util.cpp

@@ -28,11 +28,14 @@
 #include <aws/core/utils/memory/stl/AWSStringStream.h>
 #include <aws/identity-management/auth/STSAssumeRoleCredentialsProvider.h>
 #include <aws/s3/S3Client.h>
+#include <aws/s3/S3EndpointProvider.h>
 #include <aws/sts/STSClient.h>
 #include <bvar/reducer.h>
 #include <cpp/token_bucket_rate_limiter.h>
 
 #include <atomic>
+#include <optional>
+#include <string_view>
 
 #include "util/string_util.h"
 
@@ -82,11 +85,90 @@ bvar::LatencyRecorder s3_copy_object_latency("s3_copy_object");
 
 namespace {
 
+std::optional<std::string_view> get_s3_express_zone_id(std::string_view bucket) {
+    constexpr std::string_view suffix = "--x-s3";
+    if (!bucket.ends_with(suffix)) {
+        return std::nullopt;
+    }
+    bucket.remove_suffix(suffix.size());
+    const auto zone_separator = bucket.rfind("--");
+    if (zone_separator == std::string_view::npos || zone_separator == 0 ||
+        zone_separator + 2 == bucket.size()) {
+        return std::nullopt;
+    }
+    const auto zone_id = bucket.substr(zone_separator + 2);
+    const auto az_separator = zone_id.rfind("-az");
+    if (az_separator == std::string_view::npos || az_separator == 0 ||
+        az_separator + 3 == zone_id.size()) {
+        return std::nullopt;
+    }
+    for (const char c : zone_id.substr(0, az_separator)) {
+        if (!((c >= 'a' && c <= 'z') || (c >= '0' && c <= '9') || c == '-')) {
+            return std::nullopt;
+        }
+    }
+    for (const char c : zone_id.substr(az_separator + 3)) {
+        if (c < '0' || c > '9') {
+            return std::nullopt;
+        }
+    }
+    return zone_id;
+}
+
+std::optional<std::string_view> infer_s3_express_region(std::string_view zone_id) {
+    // Unknown zone prefixes deliberately fall back to the user-provided region below, so newly
+    // introduced AWS regions do not require Doris-side validation before the SDK can use them.
+    static constexpr std::pair<std::string_view, std::string_view> region_mappings[] = {
+            {"use1", "us-east-1"},  {"use2", "us-east-2"},       {"usw2", "us-west-2"},
+            {"aps1", "ap-south-1"}, {"apne1", "ap-northeast-1"}, {"euw1", "eu-west-1"},
+            {"eun1", "eu-north-1"},
+    };
+    const auto separator = zone_id.find('-');
+    const auto zone_prefix = zone_id.substr(0, separator);
+    for (const auto& [prefix, region] : region_mappings) {
+        if (zone_prefix == prefix) {
+            return region;
+        }
+    }
+    return std::nullopt;
+}
+
+bool use_s3_express_client(const S3ClientConf& conf) {
+    return conf.enable_s3_express_read && conf.provider == io::ObjStorageType::AWS &&
+           get_s3_express_zone_id(conf.bucket).has_value();
+}
+
+std::string_view effective_s3_region(const S3ClientConf& conf) {
+    if (use_s3_express_client(conf)) {
+        if (const auto zone_id = get_s3_express_zone_id(conf.bucket); zone_id.has_value()) {
+            if (const auto region = infer_s3_express_region(*zone_id); region.has_value()) {
+                return *region;
+            }
+        }
+    }
+    return conf.region;
+}
+
+void configure_s3_express_import_read(const StringCaseMap<std::string>& properties,
+                                      S3ClientConf* conf) {
+    const auto express_import = properties.find(S3_EXPRESS_IMPORT_READ);
+    const auto express_zone_id = get_s3_express_zone_id(conf->bucket);
+    conf->enable_s3_express_read =
+            express_import != properties.end() && express_import->second == "true" &&
+            conf->provider == io::ObjStorageType::AWS && express_zone_id.has_value();
+    if (conf->enable_s3_express_read) {
+        if (const auto region = infer_s3_express_region(*express_zone_id); region.has_value()) {
+            conf->region = std::string(*region);
+        }
+    }
+}
+
 doris::Status is_s3_conf_valid(const S3ClientConf& conf) {
-    if (conf.endpoint.empty()) {
+    const bool s3_express = use_s3_express_client(conf);
+    if (conf.endpoint.empty() && !s3_express) {
         return Status::InvalidArgument<false>("Invalid s3 conf, empty endpoint");
     }
-    if (conf.region.empty()) {
+    if (effective_s3_region(conf).empty()) {
         return Status::InvalidArgument<false>("Invalid s3 conf, empty region");
     }
 
@@ -487,14 +569,18 @@ std::shared_ptr<Aws::Auth::AWSCredentialsProvider> S3ClientFactory::get_aws_cred
 
 std::shared_ptr<io::ObjStorageClient> S3ClientFactory::_create_s3_client(
         const S3ClientConf& s3_conf) {
+    const bool s3_express = use_s3_express_client(s3_conf);
     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) {
+    // Let the SDK derive the zonal endpoint from the complete directory bucket name. A user
+    // endpoint is intentionally ignored for this scoped Express import client.
+    if (!s3_express && s3_conf.need_override_endpoint) {
         aws_config.endpointOverride = s3_conf.endpoint;
     }
-    aws_config.region = s3_conf.region;
+    const auto region = effective_s3_region(s3_conf);
+    aws_config.region = Aws::String(region.data(), region.size());
 
     if (_ca_cert_file_path.empty()) {
         _ca_cert_file_path = get_valid_ca_cert_path(doris::split(config::ca_cert_file_paths, ";"));
@@ -519,17 +605,32 @@ std::shared_ptr<io::ObjStorageClient> S3ClientFactory::_create_s3_client(
         aws_config.connectTimeoutMs = s3_conf.connect_timeout_ms;
     }
 
-    if (config::s3_client_http_scheme == "http") {
+    if (s3_express) {
+        aws_config.scheme = Aws::Http::Scheme::HTTPS;
+    } else if (config::s3_client_http_scheme == "http") {
         aws_config.scheme = Aws::Http::Scheme::HTTP;
     }
 
     aws_config.retryStrategy = std::make_shared<S3CustomRetryStrategy>(
             config::max_s3_client_retry /*scaleFactor = 25*/, /*retry_slow_down=*/true);
 
-    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);
+    std::shared_ptr<Aws::S3::S3Client> new_client;
+    if (s3_express) {
+        // Directory buckets use virtual-hosted endpoints. Do not propagate a user path-style
+        // preference into the SDK client selected by the directory bucket name.
+        Aws::S3::S3ClientConfiguration express_config(
+                aws_config, Aws::Client::AWSAuthV4Signer::PayloadSigningPolicy::Never,
+                true /* use virtual addressing */);
+        express_config.disableS3ExpressAuth = false;
+        new_client = std::make_shared<Aws::S3::S3Client>(
+                get_aws_credentials_provider(s3_conf),
+                Aws::MakeShared<Aws::S3::Endpoint::S3EndpointProvider>("S3Client"), express_config);
+    } else {
+        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);
+    }
 
     auto obj_client = std::make_shared<io::S3ObjStorageClient>(std::move(new_client));
     LOG_INFO("create one s3 client with {}", s3_conf.to_string());
@@ -588,6 +689,7 @@ Status S3ClientFactory::convert_properties_to_s3_conf(
     s3_conf->bucket = s3_uri.get_bucket();
     // For azure's compatibility
     s3_conf->client_conf.bucket = s3_uri.get_bucket();
+    configure_s3_express_import_read(properties, &s3_conf->client_conf);
     s3_conf->prefix = "";
 
     // See https://sdk.amazonaws.com/cpp/api/LATEST/class_aws_1_1_s3_1_1_s3_client.html

be/src/util/s3_util.h

@@ -49,6 +49,8 @@ class Adder;
 
 namespace doris {
 
+inline constexpr char S3_EXPRESS_IMPORT_READ[] = "__DORIS_S3_EXPRESS_IMPORT_READ__";
+
 namespace s3_bvar {
 extern bvar::LatencyRecorder s3_get_latency;
 extern bvar::LatencyRecorder s3_put_latency;
@@ -85,6 +87,8 @@ struct S3ClientConf {
     bool use_virtual_addressing = true;
     // For aws s3, no need to override endpoint
     bool need_override_endpoint = true;
+    // Enable S3 Express client selection for a scoped import read.
+    bool enable_s3_express_read = false;
 
     CredProviderType cred_provider_type = CredProviderType::Default;
     std::string role_arn;
@@ -102,6 +106,7 @@ struct S3ClientConf {
         hash_code ^= request_timeout_ms;
         hash_code ^= connect_timeout_ms;
         hash_code ^= use_virtual_addressing;
+        hash_code ^= enable_s3_express_read ? 0x9e3779b97f4a7c15ULL : 0;
         hash_code ^= static_cast<int>(provider);
 
         hash_code ^= static_cast<int>(cred_provider_type);
@@ -114,10 +119,10 @@ struct S3ClientConf {
         return fmt::format(
                 "(ak={}, token={}, endpoint={}, region={}, bucket={}, max_connections={}, "
                 "request_timeout_ms={}, connect_timeout_ms={}, use_virtual_addressing={}, "
-                "cred_provider_type={},role_arn={}, external_id={}",
+                "enable_s3_express_read={}, cred_provider_type={},role_arn={}, external_id={}",
                 hide_access_key(ak), token, endpoint, region, bucket, max_connections,
-                request_timeout_ms, connect_timeout_ms, use_virtual_addressing, cred_provider_type,
-                role_arn, external_id);
+                request_timeout_ms, connect_timeout_ms, use_virtual_addressing,
+                enable_s3_express_read, cred_provider_type, role_arn, external_id);
     }
 };
 

be/test/io/s3_client_factory_test.cpp

@@ -18,6 +18,9 @@
 #include <aws/core/auth/AWSCredentialsProviderChain.h>
 #include <aws/core/auth/STSCredentialsProvider.h>
 #include <aws/identity-management/auth/STSAssumeRoleCredentialsProvider.h>
+#include <aws/s3/S3Client.h>
+#include <aws/s3/S3EndpointProvider.h>
+#include <aws/s3/model/GetObjectRequest.h>
 #include <gtest/gtest.h>
 
 #include <cstdlib>
@@ -30,7 +33,10 @@
 namespace doris {
 
 class S3ClientFactoryTest : public testing::Test {
+protected:
     FRIEND_TEST(S3ClientFactoryTest, S3ClientFactory);
+
+    static void SetUpTestSuite() { static_cast<void>(S3ClientFactory::instance()); }
 };
 
 TEST_F(S3ClientFactoryTest, AwsCredentialsProvider) {
@@ -234,6 +240,163 @@ TEST_F(S3ClientFactoryTest, ConvertPropertiesToS3ConfCredentialValidation) {
     }
 }
 
+TEST_F(S3ClientFactoryTest, ConvertPropertiesToS3ExpressReadConf) {
+    S3URI s3_uri("s3://analytics--usw2-az1--x-s3/path/to/data.parquet");
+    ASSERT_TRUE(s3_uri.parse().ok());
+
+    std::map<std::string, std::string> properties {
+            {"AWS_ENDPOINT", "https://s3.us-west-2.amazonaws.com"},
+            {"AWS_REGION", "us-west-2"},
+    };
+    S3Conf non_read_conf;
+    ASSERT_TRUE(S3ClientFactory::convert_properties_to_s3_conf(properties, s3_uri, &non_read_conf)
+                        .ok());
+    ASSERT_FALSE(non_read_conf.client_conf.enable_s3_express_read);
+
+    properties[S3_EXPRESS_IMPORT_READ] = "true";
+    S3Conf s3_conf;
+    ASSERT_TRUE(S3ClientFactory::convert_properties_to_s3_conf(properties, s3_uri, &s3_conf).ok());
+    ASSERT_TRUE(s3_conf.client_conf.enable_s3_express_read);
+    ASSERT_EQ(s3_conf.client_conf.region, "us-west-2");
+
+    properties["AWS_ENDPOINT"] = "http://endpoint-is-ignored.example.com";
+    properties["AWS_REGION"] = "us-east-1";
+    properties["use_path_style"] = "true";
+    S3Conf ignored_properties_conf;
+    ASSERT_TRUE(S3ClientFactory::convert_properties_to_s3_conf(properties, s3_uri,
+                                                               &ignored_properties_conf)
+                        .ok());
+    EXPECT_EQ(ignored_properties_conf.client_conf.endpoint,
+              "http://endpoint-is-ignored.example.com");
+    EXPECT_EQ(ignored_properties_conf.client_conf.region, "us-west-2");
+    EXPECT_FALSE(ignored_properties_conf.client_conf.use_virtual_addressing);
+
+    properties.clear();
+    properties[S3_EXPRESS_IMPORT_READ] = "true";
+    S3Conf bucket_only_conf;
+    ASSERT_TRUE(
+            S3ClientFactory::convert_properties_to_s3_conf(properties, s3_uri, &bucket_only_conf)
+                    .ok());
+    EXPECT_TRUE(bucket_only_conf.client_conf.endpoint.empty());
+    EXPECT_EQ(bucket_only_conf.client_conf.region, "us-west-2");
+
+    S3URI future_region_uri("s3://analytics--future1-az1--x-s3/path/to/data.parquet");
+    ASSERT_TRUE(future_region_uri.parse().ok());
+    properties["AWS_REGION"] = "future-region-1";
+    S3Conf future_region_conf;
+    ASSERT_TRUE(S3ClientFactory::convert_properties_to_s3_conf(properties, future_region_uri,
+                                                               &future_region_conf)
+                        .ok());
+    EXPECT_TRUE(future_region_conf.client_conf.enable_s3_express_read);
+    EXPECT_EQ(future_region_conf.client_conf.region, "future-region-1");
+
+    properties.erase("AWS_REGION");
+    S3Conf missing_region_conf;
+    ASSERT_FALSE(S3ClientFactory::convert_properties_to_s3_conf(properties, future_region_uri,
+                                                                &missing_region_conf)
+                         .ok());
+
+    S3URI ordinary_uri("s3://ordinary-bucket/path/to/data.parquet");
+    ASSERT_TRUE(ordinary_uri.parse().ok());
+    properties["AWS_ENDPOINT"] = "https://s3.us-west-2.amazonaws.com";
+    properties["AWS_REGION"] = "us-west-2";
+    S3Conf ordinary_conf;
+    ASSERT_TRUE(
+            S3ClientFactory::convert_properties_to_s3_conf(properties, ordinary_uri, &ordinary_conf)
+                    .ok());
+    ASSERT_FALSE(ordinary_conf.client_conf.enable_s3_express_read);
+
+    S3URI malformed_uri("s3://analytics--x-s3/path/to/data.parquet");
+    ASSERT_TRUE(malformed_uri.parse().ok());
+    S3Conf malformed_conf;
+    ASSERT_TRUE(S3ClientFactory::convert_properties_to_s3_conf(properties, malformed_uri,
+                                                               &malformed_conf)
+                        .ok());
+    ASSERT_FALSE(malformed_conf.client_conf.enable_s3_express_read);
+
+    S3URI malformed_zone_uri("s3://analytics--usw2-azx--x-s3/path/to/data.parquet");
+    ASSERT_TRUE(malformed_zone_uri.parse().ok());
+    S3Conf malformed_zone_conf;
+    ASSERT_TRUE(S3ClientFactory::convert_properties_to_s3_conf(properties, malformed_zone_uri,
+                                                               &malformed_zone_conf)
+                        .ok());
+    ASSERT_FALSE(malformed_zone_conf.client_conf.enable_s3_express_read);
+
+    properties["provider"] = "AZURE";
+    S3Conf non_aws_conf;
+    ASSERT_TRUE(
+            S3ClientFactory::convert_properties_to_s3_conf(properties, s3_uri, &non_aws_conf).ok());
+    ASSERT_FALSE(non_aws_conf.client_conf.enable_s3_express_read);
+}
+
+TEST_F(S3ClientFactoryTest, S3ExpressRegionInference) {
+    struct TestCase {
+        const char* zone_id;
+        const char* region;
+    };
+    const TestCase test_cases[] = {
+            {.zone_id = "use1-az4", .region = "us-east-1"},
+            {.zone_id = "use2-az1", .region = "us-east-2"},
+            {.zone_id = "usw2-az1", .region = "us-west-2"},
+            {.zone_id = "aps1-az1", .region = "ap-south-1"},
+            {.zone_id = "apne1-az1", .region = "ap-northeast-1"},
+            {.zone_id = "euw1-az1", .region = "eu-west-1"},
+            {.zone_id = "eun1-az1", .region = "eu-north-1"},
+    };
+    const std::map<std::string, std::string> properties {{S3_EXPRESS_IMPORT_READ, "true"}};
+    for (const auto& test_case : test_cases) {
+        S3URI uri(fmt::format("s3://analytics--{}--x-s3/data.parquet", test_case.zone_id));
+        ASSERT_TRUE(uri.parse().ok());
+        S3Conf conf;
+        ASSERT_TRUE(S3ClientFactory::convert_properties_to_s3_conf(properties, uri, &conf).ok());
+        EXPECT_EQ(conf.client_conf.region, test_case.region);
+    }
+}
+
+TEST_F(S3ClientFactoryTest, S3ExpressReadIsPartOfClientCacheKey) {
+    S3ClientConf regular_conf;
+    regular_conf.bucket = "analytics--usw2-az1--x-s3";
+    regular_conf.endpoint = "s3.us-west-2.amazonaws.com";
+    regular_conf.region = "us-west-2";
+
+    S3ClientConf express_read_conf = regular_conf;
+    express_read_conf.enable_s3_express_read = true;
+
+    ASSERT_NE(regular_conf.get_hash(), express_read_conf.get_hash());
+}
+
+TEST_F(S3ClientFactoryTest, S3ExpressUsesSdkEndpointAndSessionAuth) {
+    Aws::S3::S3ClientConfiguration config(S3ClientFactory::getClientConfiguration(),
+                                          Aws::Client::AWSAuthV4Signer::PayloadSigningPolicy::Never,
+                                          true);
+    config.region = "us-west-2";
+    config.disableS3ExpressAuth = false;
+
+    auto endpoint_provider =
+            Aws::MakeShared<Aws::S3::Endpoint::S3EndpointProvider>("S3ClientFactoryTest");
+    auto client = std::make_shared<Aws::S3::S3Client>(
+            std::make_shared<Aws::Auth::AnonymousAWSCredentialsProvider>(), endpoint_provider,
+            config);
+    ASSERT_NE(client, nullptr);
+
+    Aws::S3::Model::GetObjectRequest request;
+    request.SetBucket("analytics--usw2-az1--x-s3");
+    request.SetKey("data.parquet");
+    auto outcome = endpoint_provider->ResolveEndpoint(request.GetEndpointContextParams());
+    ASSERT_TRUE(outcome.IsSuccess());
+    const auto& endpoint = outcome.GetResult();
+    EXPECT_EQ(endpoint.GetURI().GetAuthority(),
+              "analytics--usw2-az1--x-s3.s3express-usw2-az1.us-west-2.amazonaws.com");
+    ASSERT_TRUE(endpoint.GetAttributes().has_value());
+    const auto& attributes = *endpoint.GetAttributes();
+    EXPECT_EQ(attributes.backend, "S3Express");
+    EXPECT_EQ(attributes.authScheme.GetName(), "S3ExpressSigner");
+    ASSERT_TRUE(attributes.authScheme.GetSigningName().has_value());
+    EXPECT_EQ(*attributes.authScheme.GetSigningName(), "s3express");
+    ASSERT_TRUE(attributes.authScheme.GetSigningRegion().has_value());
+    EXPECT_EQ(*attributes.authScheme.GetSigningRegion(), "us-west-2");
+}
+
 TEST_F(S3ClientFactoryTest, AwsCredentialsProviderV2ProviderTypeWithoutRoleArn) {
     S3ClientFactory& factory = S3ClientFactory::instance();
     config::aws_credentials_provider_version = "v2";