diff --git a/be/src/util/s3_util.cpp b/be/src/util/s3_util.cpp index c6e86a7f4b290f..046fb3d4bfe0cf 100644 --- a/be/src/util/s3_util.cpp +++ b/be/src/util/s3_util.cpp @@ -28,11 +28,14 @@ #include #include #include +#include #include #include #include #include +#include +#include #include "util/string_util.h" @@ -82,11 +85,90 @@ bvar::LatencyRecorder s3_copy_object_latency("s3_copy_object"); namespace { +std::optional 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 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 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& 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("Invalid s3 conf, empty endpoint"); } - if (conf.region.empty()) { + if (effective_s3_region(conf).empty()) { return Status::InvalidArgument("Invalid s3 conf, empty region"); } @@ -487,14 +569,18 @@ std::shared_ptr S3ClientFactory::get_aws_cred std::shared_ptr 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(std::make_shared())); 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 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( config::max_s3_client_retry /*scaleFactor = 25*/, /*retry_slow_down=*/true); - 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); + std::shared_ptr 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( + get_aws_credentials_provider(s3_conf), + Aws::MakeShared("S3Client"), express_config); + } else { + 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); + } auto obj_client = std::make_shared(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 diff --git a/be/src/util/s3_util.h b/be/src/util/s3_util.h index 064f88accd8539..a4c1b8498fa0f8 100644 --- a/be/src/util/s3_util.h +++ b/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(provider); hash_code ^= static_cast(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); } }; diff --git a/be/test/io/s3_client_factory_test.cpp b/be/test/io/s3_client_factory_test.cpp index 792d815cbca5f8..f99c4a37841da8 100644 --- a/be/test/io/s3_client_factory_test.cpp +++ b/be/test/io/s3_client_factory_test.cpp @@ -18,6 +18,9 @@ #include #include #include +#include +#include +#include #include #include @@ -30,7 +33,10 @@ namespace doris { class S3ClientFactoryTest : public testing::Test { +protected: FRIEND_TEST(S3ClientFactoryTest, S3ClientFactory); + + static void SetUpTestSuite() { static_cast(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 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 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("S3ClientFactoryTest"); + auto client = std::make_shared( + std::make_shared(), 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"; diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/BrokerDesc.java b/fe/fe-core/src/main/java/org/apache/doris/analysis/BrokerDesc.java index 02f8fe532cd567..b2514b44d0f0e6 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/analysis/BrokerDesc.java +++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/BrokerDesc.java @@ -22,6 +22,7 @@ import org.apache.doris.common.io.Text; import org.apache.doris.common.io.Writable; import org.apache.doris.datasource.property.storage.BrokerProperties; +import org.apache.doris.datasource.property.storage.S3Properties; import org.apache.doris.datasource.property.storage.StorageProperties; import org.apache.doris.foundation.property.StoragePropertiesException; import org.apache.doris.persist.gson.GsonUtils; @@ -118,6 +119,19 @@ public BrokerDesc(String name, StorageBackend.StorageType storageType, Map properties) { + BrokerDesc brokerDesc = new BrokerDesc(); + brokerDesc.name = name; + brokerDesc.storageType = StorageType.S3; + brokerDesc.properties = Maps.newHashMap(); + if (properties != null) { + brokerDesc.properties.putAll(properties); + } + brokerDesc.storageProperties = S3Properties.createForS3ExpressImport(brokerDesc.properties); + return brokerDesc; + } + public String getFileLocation(String location) throws UserException { return (null != storageProperties) ? storageProperties.validateAndNormalizeUri(location) : location; } diff --git a/fe/fe-core/src/main/java/org/apache/doris/cloud/load/CloudBrokerLoadJob.java b/fe/fe-core/src/main/java/org/apache/doris/cloud/load/CloudBrokerLoadJob.java index cd5c8dd1b0c52f..634b8a0b98f71f 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/cloud/load/CloudBrokerLoadJob.java +++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/load/CloudBrokerLoadJob.java @@ -155,7 +155,8 @@ protected LoadLoadingTask createTask(Database db, OlapTable table, List 3; + String zoneId = nameAndZone.substring(zoneSeparator + 2); + return zoneId.matches("[a-z0-9-]+-az[0-9]+"); } /** diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/property/storage/AbstractS3CompatibleProperties.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/property/storage/AbstractS3CompatibleProperties.java index 3b589daab34775..887bcf479f7d75 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/property/storage/AbstractS3CompatibleProperties.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/property/storage/AbstractS3CompatibleProperties.java @@ -54,6 +54,9 @@ */ public abstract class AbstractS3CompatibleProperties extends StorageProperties implements ObjectStorageProperties { private static final Logger LOG = LogManager.getLogger(AbstractS3CompatibleProperties.class); + // Internal FE-to-BE scope marker; raw user properties never bind this flag. + public static final String S3_EXPRESS_IMPORT_READ = "__DORIS_S3_EXPRESS_IMPORT_READ__"; + private boolean s3ExpressImportRead; /** * Constructor to initialize the object storage properties with the provided type and original properties map. @@ -118,9 +121,22 @@ private Map doBuildS3Configuration(String maxConnections, if (StringUtils.isNotBlank(credentialsProviderType)) { s3Props.put("AWS_CREDENTIALS_PROVIDER_TYPE", credentialsProviderType); } + if (s3ExpressImportRead) { + s3Props.put(S3_EXPRESS_IMPORT_READ, Boolean.TRUE.toString()); + } return s3Props; } + void enableS3ExpressImportRead() { + s3ExpressImportRead = true; + } + + /** Returns whether the typed properties carry the trusted S3 Express import marker. */ + public static boolean isS3ExpressImportRead(StorageProperties properties) { + return properties instanceof AbstractS3CompatibleProperties + && ((AbstractS3CompatibleProperties) properties).s3ExpressImportRead; + } + protected String getAwsCredentialsProviderTypeForBackend() { if (StringUtils.isBlank(getAccessKey()) && StringUtils.isBlank(getSecretKey())) { return AwsCredentialsProviderMode.ANONYMOUS.name(); @@ -148,19 +164,24 @@ public AwsCredentialsProvider getAwsCredentialsProvider() { @Override public void initNormalizeAndCheckProps() { super.initNormalizeAndCheckProps(); - setEndpointIfPossible(); - setRegionIfPossible(); + // S3 Express imports use the directory bucket name and SDK endpoint rules. Keep the + // user endpoint and region untouched: the object client ignores endpoint, and uses + // region only as a fallback when the bucket Zone ID is unknown to Doris. + if (!s3ExpressImportRead) { + setEndpointIfPossible(); + setRegionIfPossible(); + } //Allow anonymous access if both access_key and secret_key are empty //But not recommended for production use. if (StringUtils.isBlank(getAccessKey()) != StringUtils.isBlank(getSecretKey())) { throw new IllegalArgumentException("Both the access key and the secret key must be set."); } - if (StringUtils.isBlank(getRegion())) { + if (!s3ExpressImportRead && StringUtils.isBlank(getRegion())) { throw new IllegalArgumentException("Region is not set. If you are using a standard endpoint, the region " + "will be detected automatically. Otherwise, please specify it explicitly." ); } - if (StringUtils.isBlank(getEndpoint())) { + if (!s3ExpressImportRead && StringUtils.isBlank(getEndpoint())) { throw new IllegalArgumentException("Endpoint is not set. Please specify it explicitly." ); } diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/property/storage/S3Properties.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/property/storage/S3Properties.java index 01c5cc6033c3af..b5b3613ff263f0 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/datasource/property/storage/S3Properties.java +++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/property/storage/S3Properties.java @@ -22,6 +22,8 @@ import org.apache.doris.cloud.proto.Cloud.ObjectStoreInfoPB.Provider; import org.apache.doris.common.Config; import org.apache.doris.common.DdlException; +import org.apache.doris.common.UserException; +import org.apache.doris.common.util.S3URI; import org.apache.doris.datasource.property.common.AwsCredentialsProviderFactory; import org.apache.doris.datasource.property.common.AwsCredentialsProviderMode; import org.apache.doris.foundation.property.ConnectorPropertiesUtils; @@ -56,6 +58,9 @@ public class S3Properties extends AbstractS3CompatibleProperties { private static final Logger LOG = LogManager.getLogger(S3Properties.class); + private static final String S3_DIRECTORY_BUCKET_SUFFIX = "--x-s3"; + private static final Pattern S3_EXPRESS_URI_AUTHORITY_PATTERN = + Pattern.compile("^[^:/?#]+://([^/?#]+)"); public static final String USE_PATH_STYLE = "use_path_style"; public static final String ENDPOINT = "s3.endpoint"; @@ -63,6 +68,7 @@ public class S3Properties extends AbstractS3CompatibleProperties { public static final String ROLE_ARN = "s3.role_arn"; public static final String EXTERNAL_ID = "s3.external_id"; public static final String CREDENTIALS_PROVIDER_TYPE = "s3.credentials_provider_type"; + public static final String PROVIDER = "s3.provider"; private static final String[] ENDPOINT_NAMES_FOR_GUESSING = { "s3.endpoint", "AWS_ENDPOINT", "endpoint", "ENDPOINT", "aws.endpoint", "glue.endpoint", @@ -190,6 +196,11 @@ public class S3Properties extends AbstractS3CompatibleProperties { @Getter private AwsCredentialsProviderMode awsCredentialsProviderMode; + @Getter + @ConnectorProperty(names = {PROVIDER, "provider"}, required = false, + description = "The S3 service provider.") + private String s3Provider = ""; + public static S3Properties of(Map properties) { S3Properties propertiesObj = new S3Properties(properties); ConnectorPropertiesUtils.bindConnectorProperties(propertiesObj, properties); @@ -197,6 +208,112 @@ public static S3Properties of(Map properties) { return propertiesObj; } + /** Creates the trusted S3 properties used only by the scoped S3 import paths. */ + public static S3Properties createForS3ExpressImport(Map properties) { + S3Properties propertiesObj = new S3Properties(properties); + propertiesObj.enableS3ExpressImportRead(); + propertiesObj.initNormalizeAndCheckProps(); + return propertiesObj; + } + + /** Returns whether the user explicitly selected AWS as the S3 provider. */ + public static boolean isAwsProvider(Map properties) { + Optional provider = getPropertyIgnoreCase(properties, PROVIDER); + if (provider.isEmpty()) { + provider = getPropertyIgnoreCase(properties, "provider"); + } + return provider.map(String::trim).filter(value -> "AWS".equalsIgnoreCase(value)).isPresent(); + } + + /** Returns whether a scoped S3 TVF request can use SDK-managed S3 Express access. */ + public static boolean isS3ExpressImport(Map properties) { + if (!isAwsProvider(properties)) { + return false; + } + Optional uri = getPropertyIgnoreCase(properties, "uri"); + if (uri.isEmpty()) { + return false; + } + return isS3ExpressUri(uri.get()); + } + + /** + * Validates an S3 Express request after first recognizing the user's intent from the bucket suffix. + * Ordinary S3 URIs return false without changing their existing provider detection path. + */ + public static boolean validateS3ExpressImport(Map properties, boolean scopeAllowed) { + return getPropertyIgnoreCase(properties, "uri") + .map(uri -> validateS3ExpressImport(uri, properties, scopeAllowed)) + .orElse(false); + } + + /** + * Validates an S3 Express request whose URI is supplied separately, as in Broker Load. + */ + public static boolean validateS3ExpressImport( + String uri, Map properties, boolean scopeAllowed) { + if (!hasS3ExpressIntent(uri)) { + return false; + } + if (!scopeAllowed) { + throw new IllegalArgumentException("S3 Express reads are supported only by " + + "SELECT ... FROM S3(...), INSERT INTO ... SELECT ... FROM S3(...), " + + "and Broker Load WITH S3."); + } + if (!isAwsProvider(properties)) { + throw new IllegalArgumentException( + "S3 Express directory buckets require \"s3.provider\" = \"AWS\"."); + } + + S3URI parsedUri; + try { + parsedUri = S3URI.create(uri); + } catch (UserException e) { + throw new IllegalArgumentException("Invalid S3 Express URI \"" + uri + "\"; expected " + + "\"s3://----x-s3/\".", e); + } + if (!parsedUri.useS3DirectoryBucket()) { + throw new IllegalArgumentException("Invalid S3 Express directory bucket name \"" + + parsedUri.getBucket() + "\"; expected " + + "\"----x-s3\", for example " + + "\"bucket-name--usw2-az1--x-s3\"."); + } + return true; + } + + /** Returns whether the URI bucket clearly carries the S3 Express suffix. */ + public static boolean hasS3ExpressIntent(String uri) { + if (StringUtils.isBlank(uri)) { + return false; + } + try { + return hasS3ExpressBucketSuffix(S3URI.create(uri).getBucket()); + } catch (UserException e) { + java.util.regex.Matcher matcher = S3_EXPRESS_URI_AUTHORITY_PATTERN.matcher(uri); + return matcher.find() && hasS3ExpressBucketSuffix(matcher.group(1)); + } + } + + private static boolean hasS3ExpressBucketSuffix(String bucket) { + return StringUtils.endsWithIgnoreCase(bucket, S3_DIRECTORY_BUCKET_SUFFIX); + } + + private static Optional getPropertyIgnoreCase(Map properties, String propertyName) { + return properties.entrySet().stream() + .filter(entry -> entry.getKey().equalsIgnoreCase(propertyName)) + .map(Map.Entry::getValue) + .findFirst(); + } + + /** Returns whether the URI contains a complete S3 Express directory bucket name. */ + public static boolean isS3ExpressUri(String uri) { + try { + return S3URI.create(uri).useS3DirectoryBucket(); + } catch (UserException e) { + return false; + } + } + /** * Pattern to match various AWS S3 endpoint formats and extract the region part. *

@@ -205,20 +322,15 @@ public static S3Properties of(Map properties) { * - s3.dualstack.us-east-1.amazonaws.com => region = us-east-1 * - s3-fips.us-east-2.amazonaws.com => region = us-east-2 * - s3-fips.dualstack.us-east-2.amazonaws.com => region = us-east-2 - * - s3express-control.us-west-2.amazonaws.com => region = us-west-2 (S3 Directory Bucket Regional) - * - s3express-usw2-az1.us-west-2.amazonaws.com => region = us-west-2 (S3 Directory Bucket Zonal) *

- * Group(1), Group(2), or Group(3) in the pattern captures the region part if available. + * Group(1) captures the region part if available. *

* For Glue https://docs.aws.amazon.com/general/latest/gr/glue.html */ private static final Set ENDPOINT_PATTERN = ImmutableSet.of( Pattern.compile( - "^(?:https?://)?(?:" - + "s3(?:[-.]fips)?(?:[-.]dualstack)?[-.]([a-z0-9-]+)|" // Standard S3 endpoints - + "s3express-control\\.([a-z0-9-]+)|" // Directory bucket regional - + "s3express-[a-z0-9-]+\\.([a-z0-9-]+)" // Directory bucket zonal - + ")\\.amazonaws\\.com(?:/.*)?$", + "^(?:https?://)?s3(?:[-.]fips)?(?:[-.]dualstack)?[-.]([a-z0-9-]+)" + + "\\.amazonaws\\.com(?:/.*)?$", Pattern.CASE_INSENSITIVE), Pattern.compile( "^(?:https?://)?glue(?:-fips)?\\.([a-z0-9-]+)\\.(amazonaws\\.com(?:\\.cn)?|api\\.aws)$", @@ -295,6 +407,10 @@ protected Set schemas() { public Map getBackendConfigProperties() { Map backendProperties = generateBackendS3Configuration(); + if (isS3ExpressImportRead(this) && StringUtils.isNotBlank(s3Provider)) { + backendProperties.put("provider", s3Provider); + } + if (StringUtils.isNotBlank(s3IAMRole)) { backendProperties.put("AWS_ROLE_ARN", s3IAMRole); } diff --git a/fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemFactory.java b/fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemFactory.java index 955e27f9cf99a2..d5b1882495c80d 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemFactory.java +++ b/fe/fe-core/src/main/java/org/apache/doris/fs/FileSystemFactory.java @@ -24,6 +24,7 @@ import org.apache.doris.common.Config; import org.apache.doris.common.UserException; import org.apache.doris.common.util.NetUtils; +import org.apache.doris.datasource.property.storage.AbstractS3CompatibleProperties; import org.apache.doris.datasource.property.storage.BrokerProperties; import org.apache.doris.datasource.property.storage.StorageProperties; import org.apache.doris.filesystem.spi.FileSystemProvider; @@ -84,6 +85,15 @@ public static void initPluginManager(FileSystemPluginManager manager) { */ public static org.apache.doris.filesystem.FileSystem getFileSystem(Map properties) throws IOException { + // A raw map can come directly from user-managed resources or catalogs. Only the typed + // StorageProperties overload below may carry this import-scope marker into the S3 plugin. + Map untrustedProperties = new HashMap<>(properties); + untrustedProperties.remove(AbstractS3CompatibleProperties.S3_EXPRESS_IMPORT_READ); + return getFileSystemInternal(untrustedProperties); + } + + private static org.apache.doris.filesystem.FileSystem getFileSystemInternal(Map properties) + throws IOException { FileSystemPluginManager mgr = pluginManager; if (mgr != null) { return mgr.createFileSystem(properties); @@ -112,7 +122,7 @@ public static org.apache.doris.filesystem.FileSystem getFileSystem(Map fileGroup.getFilePaths().stream()) + .noneMatch(S3Properties::isS3ExpressUri)) { + return brokerDesc; + } + return BrokerDesc.createForS3ExpressImport(brokerDesc.getName(), brokerDesc.getProperties()); + } + @Override public void beginTxn() throws LabelAlreadyUsedException, BeginTransactionException, AnalysisException, DuplicatedRequestException, @@ -146,7 +159,8 @@ protected void unprotectedExecuteJob() { } protected LoadTask createPendingTask() { - return new BrokerLoadPendingTask(this, fileGroupAggInfo.getAggKeyToFileGroups(), brokerDesc, getPriority()); + return new BrokerLoadPendingTask(this, fileGroupAggInfo.getAggKeyToFileGroups(), + brokerDescForS3ExpressImport(), getPriority()); } /** @@ -259,7 +273,7 @@ public void setComputeGroup() { protected LoadLoadingTask createTask(Database db, OlapTable table, List brokerFileGroups, boolean isEnableMemtableOnSinkNode, int batchSize, FileGroupAggKey aggKey, BrokerPendingTaskAttachment attachment) throws UserException { - LoadLoadingTask task = new LoadLoadingTask(this.userInfo, db, table, brokerDesc, + LoadLoadingTask task = new LoadLoadingTask(this.userInfo, db, table, brokerDescForS3ExpressImport(), brokerFileGroups, getDeadlineMs(), getExecMemLimit(), isStrictMode(), isPartialUpdate(), getPartialUpdateNewKeyPolicy(), transactionId, this, getTimeZone(), getTimeout(), diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java index cd6a985218dd60..b3a9abde76cd56 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java @@ -314,6 +314,8 @@ public enum TableFrom { private final Set> materializationRewrittenSuccessSet = new HashSet<>(); private boolean isInsert = false; + // Trusted statement-scope marker for one-shot SELECT/INSERT reads from the internal S3 TVF. + private boolean s3ExpressImportRead = false; private Optional>> mvRefreshPredicates = Optional.empty(); // For Iceberg rewrite operations: store file scan tasks to be used by @@ -1244,6 +1246,14 @@ public boolean isInsert() { return isInsert; } + public void setS3ExpressImportRead(boolean s3ExpressImportRead) { + this.s3ExpressImportRead = s3ExpressImportRead; + } + + public boolean isS3ExpressImportRead() { + return s3ExpressImportRead; + } + public boolean isQueryStatsRecorded() { return queryStatsRecorded; } diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/parser/LogicalPlanBuilder.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/parser/LogicalPlanBuilder.java index c470bc789d9e78..59a46c5532ede5 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/parser/LogicalPlanBuilder.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/parser/LogicalPlanBuilder.java @@ -64,6 +64,7 @@ import org.apache.doris.common.util.PropertyAnalyzer; import org.apache.doris.datasource.FileCacheAdmissionManager; import org.apache.doris.datasource.InternalCatalog; +import org.apache.doris.datasource.property.storage.S3Properties; import org.apache.doris.dictionary.LayoutType; import org.apache.doris.info.TableRefInfo; import org.apache.doris.info.TableValuedFunctionRefInfo; @@ -2168,12 +2169,19 @@ public Set visitPropertyKeyClause(DorisParser.PropertyKeyClauseContext c @Override public BrokerDesc visitWithRemoteStorageSystem(WithRemoteStorageSystemContext ctx) { + return visitWithRemoteStorageSystem(ctx, false); + } + + private BrokerDesc visitWithRemoteStorageSystem(WithRemoteStorageSystemContext ctx, + boolean s3ExpressImportRead) { BrokerDesc brokerDesc = null; Map brokerPropertiesMap = visitPropertyItemList(ctx.brokerProperties); if (ctx.S3() != null) { - brokerDesc = new BrokerDesc("S3", StorageBackend.StorageType.S3, brokerPropertiesMap); + brokerDesc = s3ExpressImportRead && S3Properties.isAwsProvider(brokerPropertiesMap) + ? BrokerDesc.createForS3ExpressImport("S3", brokerPropertiesMap) + : new BrokerDesc("S3", StorageBackend.StorageType.S3, brokerPropertiesMap); } else if (ctx.HDFS() != null) { brokerDesc = new BrokerDesc("HDFS", StorageBackend.StorageType.HDFS, brokerPropertiesMap); } else if (ctx.LOCAL() != null) { @@ -2223,9 +2231,33 @@ public List> visitMultiStatements(MultiState */ @Override public LogicalPlan visitLoad(DorisParser.LoadContext ctx) { + List allFilePaths = ctx.dataDescs.stream() + .flatMap(dataDesc -> dataDesc.filePaths.stream()) + .map(Token::getText) + .map(path -> path.substring(1, path.length() - 1)) + .collect(ImmutableList.toImmutableList()); + boolean hasS3ExpressIntent = allFilePaths.stream().anyMatch(S3Properties::hasS3ExpressIntent); + if (hasS3ExpressIntent && (ctx.withRemoteStorageSystem() == null + || ctx.withRemoteStorageSystem().S3() == null)) { + allFilePaths.stream() + .filter(S3Properties::hasS3ExpressIntent) + .findFirst() + .ifPresent(path -> S3Properties.validateS3ExpressImport(path, ImmutableMap.of(), false)); + } + BrokerDesc brokerDesc = null; if (ctx.withRemoteStorageSystem() != null) { - brokerDesc = visitWithRemoteStorageSystem(ctx.withRemoteStorageSystem()); + boolean s3ExpressImportRead = false; + if (hasS3ExpressIntent) { + Map brokerProperties = + visitPropertyItemList(ctx.withRemoteStorageSystem().brokerProperties); + for (String filePath : allFilePaths) { + s3ExpressImportRead |= S3Properties.validateS3ExpressImport( + filePath, brokerProperties, true); + } + } + brokerDesc = visitWithRemoteStorageSystem( + ctx.withRemoteStorageSystem(), s3ExpressImportRead); } ResourceDesc resourceDesc = null; diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/table/S3.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/table/S3.java index 3c4c924d7f41e6..f650a6b3743420 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/table/S3.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/table/S3.java @@ -24,6 +24,7 @@ import org.apache.doris.nereids.trees.plans.logical.SupportPruneNestedColumn; import org.apache.doris.nereids.trees.plans.logical.SupportPruneNestedColumnFormats; import org.apache.doris.nereids.types.coercion.AnyDataType; +import org.apache.doris.qe.ConnectContext; import org.apache.doris.tablefunction.S3TableValuedFunction; import org.apache.doris.tablefunction.TableValuedFunctionIf; @@ -44,7 +45,10 @@ public FunctionSignature customSignature() { protected TableValuedFunctionIf toCatalogFunction() { try { Map arguments = getTVFProperties().getMap(); - return new S3TableValuedFunction(arguments); + ConnectContext context = ConnectContext.get(); + boolean s3ExpressImportRead = context != null && context.getStatementContext() != null + && context.getStatementContext().isS3ExpressImportRead(); + return new S3TableValuedFunction(arguments, s3ExpressImportRead); } catch (Throwable t) { throw new AnalysisException("Can not build s3(): " + t.getMessage(), t); } diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/LoadCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/LoadCommand.java index 117f0dd3bfb802..2fdfc4d6e39d47 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/LoadCommand.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/LoadCommand.java @@ -34,6 +34,7 @@ import org.apache.doris.common.UserException; import org.apache.doris.common.util.S3Util; import org.apache.doris.common.util.TimeUtils; +import org.apache.doris.datasource.property.storage.AbstractS3CompatibleProperties; import org.apache.doris.datasource.property.storage.ObjectStorageProperties; import org.apache.doris.load.EtlJobType; import org.apache.doris.load.LoadJobRowResult; @@ -457,7 +458,9 @@ public void run(ConnectContext ctx, StmtExecutor executor) throws Exception { } else if (brokerDesc != null) { etlJobType = EtlJobType.BROKER; if (brokerDesc.getFileType() != null && brokerDesc.getFileType().equals(TFileType.FILE_S3) - && brokerDesc.getStorageProperties() instanceof ObjectStorageProperties) { + && brokerDesc.getStorageProperties() instanceof ObjectStorageProperties + && !AbstractS3CompatibleProperties.isS3ExpressImportRead( + brokerDesc.getStorageProperties())) { //@zykkk todo We should use a unified connectivity check — it doesn’t really belong here. ObjectStorageProperties storageProperties = (ObjectStorageProperties) brokerDesc.getStorageProperties(); String endpoint = storageProperties.getEndpoint(); diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java b/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java index 795a81c8a76ed7..c15a5a99ce7dac 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java @@ -81,6 +81,7 @@ import org.apache.doris.nereids.PlanProcess; import org.apache.doris.nereids.StatementContext; import org.apache.doris.nereids.analyzer.UnboundBaseExternalTableSink; +import org.apache.doris.nereids.analyzer.UnboundResultSink; import org.apache.doris.nereids.analyzer.UnboundTableSink; import org.apache.doris.nereids.exceptions.ParseException; import org.apache.doris.nereids.glue.LogicalPlanAdapter; @@ -98,6 +99,7 @@ import org.apache.doris.nereids.trees.plans.commands.DeleteFromCommand; import org.apache.doris.nereids.trees.plans.commands.DeleteFromUsingCommand; import org.apache.doris.nereids.trees.plans.commands.EmptyCommand; +import org.apache.doris.nereids.trees.plans.commands.ExplainCommand; import org.apache.doris.nereids.trees.plans.commands.Forward; import org.apache.doris.nereids.trees.plans.commands.LoadCommand; import org.apache.doris.nereids.trees.plans.commands.PrepareCommand; @@ -774,6 +776,8 @@ private void executeByNereids(TUniqueId queryId) throws Exception { "Nereids only process LogicalPlanAdapter, but parsedStmt is " + parsedStmt.getClass().getName()); context.getState().setNereids(true); LogicalPlan logicalPlan = ((LogicalPlanAdapter) parsedStmt).getLogicalPlan(); + context.getStatementContext().setS3ExpressImportRead( + shouldEnableS3ExpressImportRead(logicalPlan, context.getState().isInternal())); checkSqlBlocked(logicalPlan.getClass()); if (context.getCommand() == MysqlCommand.COM_STMT_PREPARE) { if (isForwardToMaster()) { @@ -976,6 +980,17 @@ private void executeByNereids(TUniqueId queryId) throws Exception { } } + static boolean shouldEnableS3ExpressImportRead(LogicalPlan logicalPlan, boolean internal) { + if (internal) { + return false; + } + if (logicalPlan instanceof ExplainCommand) { + logicalPlan = ((ExplainCommand) logicalPlan).getLogicalPlan(); + } + return logicalPlan.getClass() == InsertIntoTableCommand.class + || logicalPlan instanceof UnboundResultSink; + } + public static void initBlockSqlAstNames() { blockSqlAstNames.clear(); blockSqlAstNames = Pattern.compile(",") diff --git a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java index 6c7962e7267033..06d2c35f064aa1 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java +++ b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java @@ -46,6 +46,7 @@ import org.apache.doris.datasource.property.fileformat.CsvFileFormatProperties; import org.apache.doris.datasource.property.fileformat.FileFormatProperties; import org.apache.doris.datasource.property.fileformat.TextFileFormatProperties; +import org.apache.doris.datasource.property.storage.AbstractS3CompatibleProperties; import org.apache.doris.datasource.property.storage.ObjectStorageProperties; import org.apache.doris.datasource.property.storage.StorageProperties; import org.apache.doris.datasource.tvf.source.TVFScanNode; @@ -161,7 +162,8 @@ protected void parseFile() throws AnalysisException { BrokerDesc brokerDesc = getBrokerDesc(); try { StorageProperties sp = brokerDesc.getStorageProperties(); - if (sp instanceof ObjectStorageProperties) { + if (sp instanceof ObjectStorageProperties + && !AbstractS3CompatibleProperties.isS3ExpressImportRead(sp)) { S3Util.validateAndTestEndpoint(((ObjectStorageProperties) sp).getEndpoint()); } try (org.apache.doris.filesystem.FileSystem fs = FileSystemFactory.getFileSystem(brokerDesc)) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/FileTableValuedFunction.java b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/FileTableValuedFunction.java index 2cb5bedb4463ce..b14397813a448f 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/FileTableValuedFunction.java +++ b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/FileTableValuedFunction.java @@ -27,6 +27,7 @@ import org.apache.doris.datasource.property.storage.HdfsCompatibleProperties; import org.apache.doris.datasource.property.storage.HttpProperties; import org.apache.doris.datasource.property.storage.LocalProperties; +import org.apache.doris.datasource.property.storage.S3Properties; import org.apache.doris.datasource.property.storage.StorageProperties; import org.apache.doris.planner.PlanNodeId; import org.apache.doris.planner.ScanNode; @@ -50,6 +51,7 @@ public class FileTableValuedFunction extends ExternalFileTableValuedFunction { public FileTableValuedFunction(Map properties) throws AnalysisException { // We don't need to parseCommonProperties because the corresponding Storage will do it // Map props = super.parseCommonProperties(properties); + S3Properties.validateS3ExpressImport(properties, false); try { this.storageProperties = StorageProperties.createPrimary(properties); if (this.storageProperties instanceof AbstractS3CompatibleProperties diff --git a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/S3TableValuedFunction.java b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/S3TableValuedFunction.java index 7d38c50f565ac2..f7d2a48300c5e4 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/S3TableValuedFunction.java +++ b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/S3TableValuedFunction.java @@ -21,6 +21,7 @@ import org.apache.doris.common.AnalysisException; import org.apache.doris.common.FeConstants; import org.apache.doris.common.UserException; +import org.apache.doris.datasource.property.storage.S3Properties; import org.apache.doris.datasource.property.storage.StorageProperties; import org.apache.doris.thrift.TFileType; @@ -39,12 +40,23 @@ */ public class S3TableValuedFunction extends ExternalFileTableValuedFunction { public static final String NAME = "s3"; + private final boolean s3ExpressImportRead; public S3TableValuedFunction(Map properties) throws AnalysisException { + this(properties, false); + } + + public S3TableValuedFunction(Map properties, boolean s3ExpressImportRead) + throws AnalysisException { // 1. analyze common properties Map props = super.parseCommonProperties(properties); + this.s3ExpressImportRead = S3Properties.validateS3ExpressImport(props, s3ExpressImportRead); try { - this.storageProperties = StorageProperties.createPrimary(props); + if (this.s3ExpressImportRead) { + this.storageProperties = S3Properties.createForS3ExpressImport(props); + } else { + this.storageProperties = StorageProperties.createPrimary(props); + } this.backendConnectProperties.putAll(storageProperties.getBackendConfigProperties()); String uri = storageProperties.validateAndGetUri(props); filePath = storageProperties.validateAndNormalizeUri(uri); @@ -76,6 +88,9 @@ public String getFilePath() { @Override public BrokerDesc getBrokerDesc() { + if (s3ExpressImportRead) { + return BrokerDesc.createForS3ExpressImport("S3TvfBroker", processedParams); + } return new BrokerDesc("S3TvfBroker", processedParams); } @@ -85,4 +100,3 @@ public String getTableName() { return "S3TableValuedFunction"; } } - diff --git a/fe/fe-core/src/test/java/org/apache/doris/common/util/S3URITest.java b/fe/fe-core/src/test/java/org/apache/doris/common/util/S3URITest.java index 5cfd889ab97f90..0508beb9627c04 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/common/util/S3URITest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/common/util/S3URITest.java @@ -235,6 +235,7 @@ public void testS3DirectoryBucket() throws UserException { Assert.assertFalse(S3URI.isS3DirectoryBucket("my-bucket--usw2-az1--x-s4")); // wrong suffix Assert.assertFalse(S3URI.isS3DirectoryBucket("my-bucket-usw2-az1--x-s3")); // incorrect format Assert.assertFalse(S3URI.isS3DirectoryBucket("my-bucket--usw2az1--x-s3")); // azid without hyphen + Assert.assertFalse(S3URI.isS3DirectoryBucket("my-bucket--usw2-azx--x-s3")); // invalid az number Assert.assertFalse(S3URI.isS3DirectoryBucket("my-bucket---x-s3")); // empty azid Assert.assertFalse(S3URI.isS3DirectoryBucket(null)); Assert.assertFalse(S3URI.isS3DirectoryBucket("")); diff --git a/fe/fe-core/src/test/java/org/apache/doris/datasource/property/storage/S3PropertiesTest.java b/fe/fe-core/src/test/java/org/apache/doris/datasource/property/storage/S3PropertiesTest.java index f9cfde32b0633d..6398d0eaf148bc 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/datasource/property/storage/S3PropertiesTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/datasource/property/storage/S3PropertiesTest.java @@ -386,36 +386,6 @@ public void testGetAwsCredentialsProviderWithAccessKeyAndSecretKey() throws User Assertions.assertTrue(provider instanceof StaticCredentialsProvider); } - @Test - public void testS3ExpressEndpointPattern() throws UserException { - origProps.put("s3.access_key", "myS3AccessKey"); - origProps.put("s3.secret_key", "myS3SecretKey"); - - // S3 Express Control Endpoint (Regional) - String endpointControl = "s3express-control.us-west-2.amazonaws.com"; - origProps.put("s3.endpoint", endpointControl); - S3Properties s3Properties = (S3Properties) StorageProperties.createPrimary(origProps); - Assertions.assertEquals("us-west-2", s3Properties.getRegion()); - - // S3 Express Zonal Endpoint - String endpointZonal = "s3express-usw2-az1.us-west-2.amazonaws.com"; - origProps.put("s3.endpoint", endpointZonal); - s3Properties = (S3Properties) StorageProperties.createPrimary(origProps); - Assertions.assertEquals("us-west-2", s3Properties.getRegion()); - - // Test with https scheme - String endpointWithScheme = "https://s3express-control.eu-central-1.amazonaws.com"; - origProps.put("s3.endpoint", endpointWithScheme); - s3Properties = (S3Properties) StorageProperties.createPrimary(origProps); - Assertions.assertEquals("eu-central-1", s3Properties.getRegion()); - - // Test with path - String endpointWithPath = "https://s3express-control.eu-central-1.amazonaws.com/path/to/obj"; - origProps.put("s3.endpoint", endpointWithPath); - s3Properties = (S3Properties) StorageProperties.createPrimary(origProps); - Assertions.assertEquals("eu-central-1", s3Properties.getRegion()); - } - @Test public void testInvalidEndpoint() { origProps.put("s3.access_key", "myS3AccessKey"); @@ -554,4 +524,93 @@ public void testS3DisableHadoopCache() throws UserException { Assertions.assertEquals("false", s3Properties.hadoopStorageConfig.get("fs.s3.impl.disable.cache")); Assertions.assertEquals("false", s3Properties.hadoopStorageConfig.get("fs.s3n.impl.disable.cache")); } + + @Test + public void testS3ExpressImportMarkerRequiresInternalOptIn() { + origProps.put("s3.endpoint", "https://s3.us-west-2.amazonaws.com"); + origProps.put("s3.region", "us-west-2"); + origProps.put("s3.provider", "AWS"); + origProps.put(AbstractS3CompatibleProperties.S3_EXPRESS_IMPORT_READ, "true"); + S3Properties s3Properties = (S3Properties) StorageProperties.createPrimary(origProps); + + Assertions.assertFalse(s3Properties.getBackendConfigProperties() + .containsKey(AbstractS3CompatibleProperties.S3_EXPRESS_IMPORT_READ)); + Assertions.assertFalse(s3Properties.getBackendConfigProperties().containsKey("provider")); + + s3Properties.enableS3ExpressImportRead(); + Assertions.assertEquals("true", s3Properties.getBackendConfigProperties() + .get(AbstractS3CompatibleProperties.S3_EXPRESS_IMPORT_READ)); + Assertions.assertEquals("AWS", s3Properties.getBackendConfigProperties().get("provider")); + } + + @Test + public void testS3ExpressImportAllowsSdkEndpointResolution() { + origProps.put("uri", "s3://analytics--usw2-az1--x-s3/data/file.parquet"); + origProps.put("s3.provider", "AWS"); + origProps.put("s3.access_key", "myS3AccessKey"); + origProps.put("s3.secret_key", "myS3SecretKey"); + + Assertions.assertTrue(S3Properties.isS3ExpressImport(origProps)); + S3Properties s3Properties = S3Properties.createForS3ExpressImport(origProps); + Assertions.assertTrue(AbstractS3CompatibleProperties.isS3ExpressImportRead(s3Properties)); + Map backendProperties = s3Properties.getBackendConfigProperties(); + Assertions.assertEquals("true", + backendProperties.get(AbstractS3CompatibleProperties.S3_EXPRESS_IMPORT_READ)); + Assertions.assertEquals("AWS", backendProperties.get("provider")); + Assertions.assertEquals("", backendProperties.get("AWS_ENDPOINT")); + Assertions.assertEquals("", backendProperties.get("AWS_REGION")); + } + + @Test + public void testS3ExpressIntentValidation() { + origProps.put("uri", "s3://analytics--usw2-az1--x-s3/data/file.parquet"); + origProps.put("s3.provider", " AWS "); + + Assertions.assertTrue(S3Properties.hasS3ExpressIntent(origProps.get("uri"))); + Assertions.assertTrue(S3Properties.validateS3ExpressImport(origProps, true)); + Assertions.assertFalse(S3Properties.hasS3ExpressIntent( + "s3://ordinary-bucket/data/part--x-s3/file.parquet")); + Assertions.assertFalse(S3Properties.validateS3ExpressImport( + "s3://ordinary-bucket/data/file.parquet", origProps, true)); + + IllegalArgumentException scopeException = Assertions.assertThrows(IllegalArgumentException.class, + () -> S3Properties.validateS3ExpressImport(origProps, false)); + Assertions.assertTrue(scopeException.getMessage().contains( + "SELECT ... FROM S3(...), INSERT INTO ... SELECT ... FROM S3(...), and Broker Load WITH S3")); + + Map missingProvider = new HashMap<>(origProps); + missingProvider.remove("s3.provider"); + IllegalArgumentException providerException = Assertions.assertThrows(IllegalArgumentException.class, + () -> S3Properties.validateS3ExpressImport(missingProvider, true)); + Assertions.assertEquals( + "S3 Express directory buckets require \"s3.provider\" = \"AWS\".", + providerException.getMessage()); + + Map invalidBucket = new HashMap<>(origProps); + invalidBucket.put("uri", "s3://analytics--usw2-azx--x-s3/data/file.parquet"); + Assertions.assertTrue(S3Properties.hasS3ExpressIntent(invalidBucket.get("uri"))); + IllegalArgumentException bucketException = Assertions.assertThrows(IllegalArgumentException.class, + () -> S3Properties.validateS3ExpressImport(invalidBucket, true)); + Assertions.assertTrue(bucketException.getMessage().contains( + "Invalid S3 Express directory bucket name \"analytics--usw2-azx--x-s3\"")); + + Map missingObjectPath = new HashMap<>(origProps); + missingObjectPath.put("uri", "s3://analytics--usw2-az1--x-s3"); + Assertions.assertTrue(S3Properties.hasS3ExpressIntent(missingObjectPath.get("uri"))); + IllegalArgumentException uriException = Assertions.assertThrows(IllegalArgumentException.class, + () -> S3Properties.validateS3ExpressImport(missingObjectPath, true)); + Assertions.assertTrue(uriException.getMessage().contains("Invalid S3 Express URI")); + Assertions.assertTrue(uriException.getMessage().contains("/")); + } + + @Test + public void testS3ExpressProviderUsesCanonicalAliasPrecedence() { + origProps.put("s3.provider", "S3"); + origProps.put("provider", "AWS"); + Assertions.assertFalse(S3Properties.isAwsProvider(origProps)); + + origProps.put("s3.provider", "AWS"); + origProps.put("provider", "S3"); + Assertions.assertTrue(S3Properties.isAwsProvider(origProps)); + } } diff --git a/fe/fe-core/src/test/java/org/apache/doris/fs/FileSystemFactoryS3ExpressScopeTest.java b/fe/fe-core/src/test/java/org/apache/doris/fs/FileSystemFactoryS3ExpressScopeTest.java new file mode 100644 index 00000000000000..b0a88a8c98878c --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/fs/FileSystemFactoryS3ExpressScopeTest.java @@ -0,0 +1,76 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.fs; + +import org.apache.doris.datasource.property.storage.AbstractS3CompatibleProperties; +import org.apache.doris.datasource.property.storage.S3Properties; +import org.apache.doris.filesystem.FileSystem; + +import org.junit.After; +import org.junit.Assert; +import org.junit.Test; +import org.mockito.ArgumentCaptor; +import org.mockito.Mockito; + +import java.util.HashMap; +import java.util.Map; + +public class FileSystemFactoryS3ExpressScopeTest { + + @After + public void tearDown() { + FileSystemFactory.clearProviderCache(); + } + + @Test + public void testRawMapCannotInjectS3ExpressImportMarker() throws Exception { + FileSystemPluginManager manager = Mockito.mock(FileSystemPluginManager.class); + Mockito.when(manager.createFileSystem(Mockito.anyMap())).thenReturn(Mockito.mock(FileSystem.class)); + FileSystemFactory.initPluginManager(manager); + Map rawProperties = new HashMap<>(); + rawProperties.put(AbstractS3CompatibleProperties.S3_EXPRESS_IMPORT_READ, "true"); + + FileSystemFactory.getFileSystem(rawProperties); + + ArgumentCaptor> captor = ArgumentCaptor.forClass(Map.class); + Mockito.verify(manager).createFileSystem(captor.capture()); + Assert.assertFalse(captor.getValue() + .containsKey(AbstractS3CompatibleProperties.S3_EXPRESS_IMPORT_READ)); + Assert.assertTrue(rawProperties + .containsKey(AbstractS3CompatibleProperties.S3_EXPRESS_IMPORT_READ)); + } + + @Test + public void testTypedS3ExpressImportKeepsTrustedMarker() throws Exception { + FileSystemPluginManager manager = Mockito.mock(FileSystemPluginManager.class); + Mockito.when(manager.createFileSystem(Mockito.anyMap())).thenReturn(Mockito.mock(FileSystem.class)); + FileSystemFactory.initPluginManager(manager); + Map properties = new HashMap<>(); + properties.put("uri", "s3://analytics--usw2-az1--x-s3/data.parquet"); + properties.put("s3.provider", "AWS"); + properties.put("s3.access_key", "ak"); + properties.put("s3.secret_key", "sk"); + + FileSystemFactory.getFileSystem(S3Properties.createForS3ExpressImport(properties)); + + ArgumentCaptor> captor = ArgumentCaptor.forClass(Map.class); + Mockito.verify(manager).createFileSystem(captor.capture()); + Assert.assertEquals("true", captor.getValue() + .get(AbstractS3CompatibleProperties.S3_EXPRESS_IMPORT_READ)); + } +} diff --git a/fe/fe-core/src/test/java/org/apache/doris/load/loadv2/BrokerLoadJobTest.java b/fe/fe-core/src/test/java/org/apache/doris/load/loadv2/BrokerLoadJobTest.java index 3764e0e286da95..825823f57695db 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/load/loadv2/BrokerLoadJobTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/load/loadv2/BrokerLoadJobTest.java @@ -18,6 +18,7 @@ package org.apache.doris.load.loadv2; import org.apache.doris.analysis.BrokerDesc; +import org.apache.doris.analysis.StorageBackend; import org.apache.doris.catalog.Database; import org.apache.doris.catalog.Env; import org.apache.doris.catalog.OlapTable; @@ -27,9 +28,11 @@ import org.apache.doris.common.Status; import org.apache.doris.common.jmockit.Deencapsulation; import org.apache.doris.datasource.InternalCatalog; +import org.apache.doris.datasource.property.storage.AbstractS3CompatibleProperties; import org.apache.doris.load.BrokerFileGroup; import org.apache.doris.load.BrokerFileGroupAggInfo; import org.apache.doris.load.BrokerFileGroupAggInfo.FileGroupAggKey; +import org.apache.doris.load.EtlJobType; import org.apache.doris.load.EtlStatus; import org.apache.doris.metric.MetricRepo; import org.apache.doris.nereids.load.NereidsBrokerFileGroup; @@ -67,6 +70,42 @@ public static void start() { MetricRepo.init(); } + @Test + public void testS3ExpressImportPropertyIsTaskScopedToBrokerLoad() { + Map properties = Maps.newHashMap(); + properties.put("s3.provider", "AWS"); + properties.put("s3.endpoint", "https://endpoint-is-ignored.example.com"); + properties.put("s3.region", "us-east-1"); + BrokerDesc original = new BrokerDesc("S3", StorageBackend.StorageType.S3, properties); + BrokerLoadJob brokerLoadJob = new BrokerLoadJob(); + Deencapsulation.setField(brokerLoadJob, "brokerDesc", original); + BrokerFileGroup expressFileGroup = Mockito.mock(BrokerFileGroup.class); + Mockito.when(expressFileGroup.getFilePaths()).thenReturn( + Collections.singletonList("s3://analytics--usw2-az1--x-s3/data.parquet")); + BrokerFileGroupAggInfo fileGroupAggInfo = new BrokerFileGroupAggInfo(); + Deencapsulation.setField(fileGroupAggInfo, "aggKeyToFileGroups", + Collections.singletonMap(new FileGroupAggKey(1L, null), + Collections.singletonList(expressFileGroup))); + Deencapsulation.setField(brokerLoadJob, "fileGroupAggInfo", fileGroupAggInfo); + + BrokerDesc taskBrokerDesc = brokerLoadJob.brokerDescForS3ExpressImport(); + + Assert.assertNotSame(original, taskBrokerDesc); + Assert.assertFalse(original.getBackendConfigProperties() + .containsKey(AbstractS3CompatibleProperties.S3_EXPRESS_IMPORT_READ)); + Assert.assertEquals("true", taskBrokerDesc.getBackendConfigProperties() + .get(AbstractS3CompatibleProperties.S3_EXPRESS_IMPORT_READ)); + + Mockito.when(expressFileGroup.getFilePaths()).thenReturn( + Collections.singletonList("s3://ordinary-bucket/data.parquet")); + Assert.assertSame(original, brokerLoadJob.brokerDescForS3ExpressImport()); + + Mockito.when(expressFileGroup.getFilePaths()).thenReturn( + Collections.singletonList("s3://analytics--usw2-az1--x-s3/data.parquet")); + Deencapsulation.setField(brokerLoadJob, "jobType", EtlJobType.COPY); + Assert.assertSame(original, brokerLoadJob.brokerDescForS3ExpressImport()); + } + @Test public void testGetTableNames() throws MetaNotFoundException { BrokerFileGroupAggInfo fileGroupAggInfo = Mockito.mock(BrokerFileGroupAggInfo.class); diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/LoadCommandTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/LoadCommandTest.java index 60ee030dcf3624..8caa01abe9ef05 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/LoadCommandTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/LoadCommandTest.java @@ -22,6 +22,7 @@ import org.apache.doris.common.Pair; import org.apache.doris.datasource.property.fileformat.CsvFileFormatProperties; import org.apache.doris.datasource.property.fileformat.DeferredFileFormatProperties; +import org.apache.doris.datasource.property.storage.AbstractS3CompatibleProperties; import org.apache.doris.nereids.StatementContext; import org.apache.doris.nereids.analyzer.UnboundSlot; import org.apache.doris.nereids.load.NereidsDataDescription; @@ -145,6 +146,58 @@ public void testLoadCommandWithMultipleFiles() { Assertions.assertEquals("s3://bucket/customer/part-3", filePaths.get(2)); } + @Test + public void testS3ExpressLoadWithoutEndpointAndRegion() { + String loadSql = "LOAD LABEL customer_s3_express_test( " + + " DATA INFILE(\"s3://analytics--usw2-az1--x-s3/customer/*.parquet\") " + + " INTO TABLE customer" + + " ) " + + " WITH S3( " + + " \"s3.provider\" = \"AWS\", " + + " \"s3.access_key\" = \"AK\", " + + " \"s3.secret_key\" = \"SK\", " + + " \"use_path_style\" = \"false\");"; + + List> statements = new NereidsParser().parseMultiple(loadSql); + Assertions.assertFalse(statements.isEmpty()); + + LoadCommand command = (LoadCommand) statements.get(0).first; + Map backendProperties = command.getBrokerDesc().getBackendConfigProperties(); + Assertions.assertEquals("true", + backendProperties.get(AbstractS3CompatibleProperties.S3_EXPRESS_IMPORT_READ)); + Assertions.assertEquals("AWS", backendProperties.get("provider")); + Assertions.assertEquals("", backendProperties.get("AWS_ENDPOINT")); + Assertions.assertEquals("", backendProperties.get("AWS_REGION")); + + String missingProviderSql = loadSql.replace( + " \"s3.provider\" = \"AWS\", ", ""); + IllegalArgumentException providerException = Assertions.assertThrows( + IllegalArgumentException.class, + () -> new NereidsParser().parseMultiple(missingProviderSql)); + Assertions.assertTrue(providerException.getMessage().contains( + "S3 Express directory buckets require \"s3.provider\" = \"AWS\"")); + + String malformedBucketSql = loadSql.replace( + "analytics--usw2-az1--x-s3", "analytics--usw2-azx--x-s3"); + IllegalArgumentException bucketException = Assertions.assertThrows( + IllegalArgumentException.class, + () -> new NereidsParser().parseMultiple(malformedBucketSql)); + Assertions.assertTrue(bucketException.getMessage().contains( + "Invalid S3 Express directory bucket name")); + + String brokerScopeSql = loadSql.replace("WITH S3(", "WITH BROKER \"broker0\" ("); + IllegalArgumentException scopeException = Assertions.assertThrows( + IllegalArgumentException.class, + () -> new NereidsParser().parseMultiple(brokerScopeSql)); + Assertions.assertTrue(scopeException.getMessage().contains( + "S3 Express reads are supported only by")); + + String ordinaryBucketSql = loadSql.replace( + "analytics--usw2-az1--x-s3", "ordinary-bucket"); + Assertions.assertThrows(IllegalArgumentException.class, + () -> new NereidsParser().parseMultiple(ordinaryBucketSql)); + } + @Test public void testLoadCommand() throws Exception { String loadSql1 = "LOAD LABEL customer_lable_for_test( " diff --git a/fe/fe-core/src/test/java/org/apache/doris/qe/S3ExpressImportScopeTest.java b/fe/fe-core/src/test/java/org/apache/doris/qe/S3ExpressImportScopeTest.java new file mode 100644 index 00000000000000..2b668d772d2f0a --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/qe/S3ExpressImportScopeTest.java @@ -0,0 +1,93 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.qe; + +import org.apache.doris.nereids.parser.NereidsParser; +import org.apache.doris.nereids.trees.plans.PlanType; +import org.apache.doris.nereids.trees.plans.commands.insert.InsertIntoDictionaryCommand; +import org.apache.doris.nereids.trees.plans.commands.insert.InsertIntoTableCommand; +import org.apache.doris.nereids.trees.plans.commands.insert.WarmupSelectCommand; +import org.apache.doris.nereids.trees.plans.logical.LogicalPlan; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; + +class S3ExpressImportScopeTest { + + private static final String S3_SELECT_SQL = "SELECT * FROM S3(" + + "\"uri\" = \"s3://analytics--usw2-az1--x-s3/data/file.csv\", " + + "\"s3.provider\" = \"AWS\", " + + "\"format\" = \"csv\")"; + + @Test + void enableOnlyForTopLevelOneShotSelectAndInsert() { + ConnectContext previousContext = ConnectContext.get(); + ConnectContext context = new ConnectContext(); + context.setDatabase("test"); + context.setThreadLocalInfo(); + try { + NereidsParser parser = new NereidsParser(); + + LogicalPlan select = parser.parseSingle(S3_SELECT_SQL); + Assertions.assertEquals(PlanType.LOGICAL_UNBOUND_RESULT_SINK, select.getType()); + Assertions.assertTrue(StmtExecutor.shouldEnableS3ExpressImportRead(select, false)); + Assertions.assertTrue(StmtExecutor.shouldEnableS3ExpressImportRead( + parser.parseSingle("EXPLAIN " + S3_SELECT_SQL), false)); + Assertions.assertFalse(StmtExecutor.shouldEnableS3ExpressImportRead(select, true)); + + LogicalPlan insert = parser.parseSingle("INSERT INTO target_table " + S3_SELECT_SQL); + Assertions.assertEquals(InsertIntoTableCommand.class, insert.getClass()); + Assertions.assertTrue(StmtExecutor.shouldEnableS3ExpressImportRead(insert, false)); + Assertions.assertTrue(StmtExecutor.shouldEnableS3ExpressImportRead( + parser.parseSingle("EXPLAIN INSERT INTO target_table " + S3_SELECT_SQL), false)); + Assertions.assertFalse(StmtExecutor.shouldEnableS3ExpressImportRead(insert, true)); + + LogicalPlan overwrite = parser.parseSingle("INSERT OVERWRITE TABLE target_table " + S3_SELECT_SQL); + Assertions.assertFalse(StmtExecutor.shouldEnableS3ExpressImportRead( + overwrite, false)); + Assertions.assertFalse(StmtExecutor.shouldEnableS3ExpressImportRead( + parser.parseSingle("EXPLAIN INSERT OVERWRITE TABLE target_table " + S3_SELECT_SQL), false)); + Assertions.assertFalse(StmtExecutor.shouldEnableS3ExpressImportRead( + parser.parseSingle("CREATE TABLE target_table AS " + S3_SELECT_SQL), false)); + LogicalPlan outfile = parser.parseSingle(S3_SELECT_SQL + + " INTO OUTFILE \"file:///tmp/s3-express-select\" FORMAT AS CSV"); + Assertions.assertEquals(PlanType.LOGICAL_FILE_SINK, outfile.getType()); + Assertions.assertFalse(StmtExecutor.shouldEnableS3ExpressImportRead(outfile, false)); + Assertions.assertFalse(StmtExecutor.shouldEnableS3ExpressImportRead( + plan(PlanType.INSERT_INTO_TVF_COMMAND), false)); + Assertions.assertFalse(StmtExecutor.shouldEnableS3ExpressImportRead( + plan(PlanType.COPY_INTO_COMMAND), false)); + Assertions.assertFalse(StmtExecutor.shouldEnableS3ExpressImportRead( + Mockito.mock(WarmupSelectCommand.class), false)); + Assertions.assertFalse(StmtExecutor.shouldEnableS3ExpressImportRead( + Mockito.mock(InsertIntoDictionaryCommand.class), false)); + } finally { + ConnectContext.remove(); + if (previousContext != null) { + previousContext.setThreadLocalInfo(); + } + } + } + + private static LogicalPlan plan(PlanType type) { + LogicalPlan plan = Mockito.mock(LogicalPlan.class); + Mockito.when(plan.getType()).thenReturn(type); + return plan; + } +} diff --git a/fe/fe-core/src/test/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunctionTest.java b/fe/fe-core/src/test/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunctionTest.java index e5b06bd5dd4101..7ed802f20bf68a 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunctionTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunctionTest.java @@ -21,8 +21,16 @@ import org.apache.doris.catalog.PrimitiveType; import org.apache.doris.common.AnalysisException; import org.apache.doris.common.Config; +import org.apache.doris.common.FeConstants; import org.apache.doris.common.util.FileFormatConstants; import org.apache.doris.common.util.FileFormatUtils; +import org.apache.doris.datasource.property.storage.AbstractS3CompatibleProperties; +import org.apache.doris.nereids.StatementContext; +import org.apache.doris.nereids.trees.expressions.Properties; +import org.apache.doris.nereids.trees.expressions.functions.table.File; +import org.apache.doris.nereids.trees.expressions.functions.table.S3; +import org.apache.doris.qe.ConnectContext; +import org.apache.doris.qe.OriginStatement; import com.google.common.collect.Lists; import com.google.common.collect.Maps; @@ -116,4 +124,97 @@ public void testCsvSchemaParse() { Assert.fail(); } } + + @Test + public void testS3ExpressMarkerIsScopedToAllowedOneShotRead() throws AnalysisException { + boolean previousRunningUnitTest = FeConstants.runningUnitTest; + ConnectContext previousContext = ConnectContext.get(); + FeConstants.runningUnitTest = true; + try { + Map properties = Maps.newHashMap(); + properties.put("uri", "s3://analytics--usw2-az1--x-s3/data/file.csv"); + properties.put("s3.provider", "AWS"); + properties.put("s3.endpoint", "https://s3.us-west-2.amazonaws.com"); + properties.put("s3.region", "us-west-2"); + properties.put("format", "csv"); + + IllegalArgumentException directException = Assert.assertThrows( + IllegalArgumentException.class, () -> new S3TableValuedFunction(properties)); + Assert.assertTrue(directException.getMessage().contains( + "S3 Express reads are supported only by")); + + ConnectContext context = new ConnectContext(); + StatementContext statementContext = new StatementContext( + context, new OriginStatement("select * from s3(...)", 0)); + context.setStatementContext(statementContext); + context.setThreadLocalInfo(); + + org.apache.doris.nereids.exceptions.AnalysisException unmarkedException = Assert.assertThrows( + org.apache.doris.nereids.exceptions.AnalysisException.class, + () -> new S3(new Properties(properties)).getCatalogFunction()); + Assert.assertTrue(unmarkedException.getMessage().contains( + "S3 Express reads are supported only by")); + + statementContext.setS3ExpressImportRead(true); + Map regularBucketProperties = Maps.newHashMap(properties); + regularBucketProperties.put("uri", "s3://analytics/data/file.csv"); + S3TableValuedFunction regularBucketTvf = (S3TableValuedFunction) new S3( + new Properties(regularBucketProperties)).getCatalogFunction(); + Assert.assertFalse(regularBucketTvf.getBackendConnectProperties() + .containsKey(AbstractS3CompatibleProperties.S3_EXPRESS_IMPORT_READ)); + Assert.assertFalse(regularBucketTvf.getBrokerDesc().getBackendConfigProperties() + .containsKey(AbstractS3CompatibleProperties.S3_EXPRESS_IMPORT_READ)); + + Map missingProviderProperties = Maps.newHashMap(properties); + missingProviderProperties.remove("s3.provider"); + org.apache.doris.nereids.exceptions.AnalysisException missingProviderException = Assert.assertThrows( + org.apache.doris.nereids.exceptions.AnalysisException.class, + () -> new S3(new Properties(missingProviderProperties)).getCatalogFunction()); + Assert.assertTrue(missingProviderException.getMessage().contains( + "S3 Express directory buckets require \"s3.provider\" = \"AWS\"")); + + Map malformedBucketProperties = Maps.newHashMap(properties); + malformedBucketProperties.put("uri", + "s3://analytics--usw2-azx--x-s3/data/file.csv"); + org.apache.doris.nereids.exceptions.AnalysisException malformedBucketException = Assert.assertThrows( + org.apache.doris.nereids.exceptions.AnalysisException.class, + () -> new S3(new Properties(malformedBucketProperties)).getCatalogFunction()); + Assert.assertTrue(malformedBucketException.getMessage().contains( + "Invalid S3 Express directory bucket name")); + + Map recommendedProperties = Maps.newHashMap(properties); + recommendedProperties.remove("s3.endpoint"); + recommendedProperties.remove("s3.region"); + S3TableValuedFunction selectTvf = (S3TableValuedFunction) new S3( + new Properties(recommendedProperties)) + .getCatalogFunction(); + Assert.assertEquals("true", selectTvf.getBackendConnectProperties() + .get(AbstractS3CompatibleProperties.S3_EXPRESS_IMPORT_READ)); + Assert.assertEquals("true", selectTvf.getBrokerDesc().getBackendConfigProperties() + .get(AbstractS3CompatibleProperties.S3_EXPRESS_IMPORT_READ)); + Assert.assertFalse(selectTvf.processedParams + .containsKey(AbstractS3CompatibleProperties.S3_EXPRESS_IMPORT_READ)); + Assert.assertEquals("", selectTvf.getBackendConnectProperties().get("AWS_ENDPOINT")); + Assert.assertEquals("", selectTvf.getBackendConnectProperties().get("AWS_REGION")); + + org.apache.doris.nereids.exceptions.AnalysisException fileTvfException = Assert.assertThrows( + org.apache.doris.nereids.exceptions.AnalysisException.class, + () -> new File(new Properties(recommendedProperties)).getCatalogFunction()); + Assert.assertTrue(fileTvfException.getMessage().contains( + "S3 Express reads are supported only by")); + + statementContext.setS3ExpressImportRead(false); + org.apache.doris.nereids.exceptions.AnalysisException streamingException = Assert.assertThrows( + org.apache.doris.nereids.exceptions.AnalysisException.class, + () -> new S3(new Properties(properties)).getCatalogFunction()); + Assert.assertTrue(streamingException.getMessage().contains( + "S3 Express reads are supported only by")); + } finally { + FeConstants.runningUnitTest = previousRunningUnitTest; + ConnectContext.remove(); + if (previousContext != null) { + previousContext.setThreadLocalInfo(); + } + } + } } diff --git a/fe/fe-filesystem/fe-filesystem-s3/src/main/java/org/apache/doris/filesystem/s3/S3FileSystem.java b/fe/fe-filesystem/fe-filesystem-s3/src/main/java/org/apache/doris/filesystem/s3/S3FileSystem.java index f738653ebec568..bb510f6de9a09e 100644 --- a/fe/fe-filesystem/fe-filesystem-s3/src/main/java/org/apache/doris/filesystem/s3/S3FileSystem.java +++ b/fe/fe-filesystem/fe-filesystem-s3/src/main/java/org/apache/doris/filesystem/s3/S3FileSystem.java @@ -19,6 +19,7 @@ import org.apache.doris.filesystem.spi.S3CompatibleFileSystem; +import java.io.IOException; import java.util.List; import java.util.Optional; @@ -28,6 +29,7 @@ public class S3FileSystem extends S3CompatibleFileSystem { private final S3FileSystemProperties properties; + private final S3ObjStorage s3ObjStorage; public S3FileSystem(S3FileSystemProperties properties) { this(properties, new S3ObjStorage(properties)); @@ -36,11 +38,13 @@ public S3FileSystem(S3FileSystemProperties properties) { S3FileSystem(S3FileSystemProperties properties, S3ObjStorage objStorage) { super(objStorage, objStorage.isUsePathStyle(), objStorage.getSupportedSchemes()); this.properties = properties; + this.s3ObjStorage = objStorage; } public S3FileSystem(S3ObjStorage objStorage) { super(objStorage, objStorage.isUsePathStyle(), objStorage.getSupportedSchemes()); this.properties = null; + this.s3ObjStorage = objStorage; } public Optional properties() { @@ -48,23 +52,20 @@ public Optional properties() { } @Override - protected String globListPrefix(String globPattern) { - if (isDirectoryBucketEndpoint()) { + protected String globListPrefix(String bucket, String globPattern) throws IOException { + if (s3ObjStorage.usesS3ExpressRead(bucket)) { return slashTerminatedNonGlobPrefix(globPattern); } - return super.globListPrefix(globPattern); + return super.globListPrefix(bucket, globPattern); } @Override - protected List globListPrefixes(String globPattern, String listPrefix) { - if (isDirectoryBucketEndpoint()) { + protected List globListPrefixes(String bucket, String globPattern, String listPrefix) + throws IOException { + if (s3ObjStorage.usesS3ExpressRead(bucket)) { return List.of(listPrefix); } - return super.globListPrefixes(globPattern, listPrefix); - } - - private boolean isDirectoryBucketEndpoint() { - return properties != null && properties.isDirectoryBucketEndpoint(); + return super.globListPrefixes(bucket, globPattern, listPrefix); } private static String slashTerminatedNonGlobPrefix(String globPattern) { diff --git a/fe/fe-filesystem/fe-filesystem-s3/src/main/java/org/apache/doris/filesystem/s3/S3FileSystemProperties.java b/fe/fe-filesystem/fe-filesystem-s3/src/main/java/org/apache/doris/filesystem/s3/S3FileSystemProperties.java index 97cfbb2c4b03d0..8670ce89224834 100644 --- a/fe/fe-filesystem/fe-filesystem-s3/src/main/java/org/apache/doris/filesystem/s3/S3FileSystemProperties.java +++ b/fe/fe-filesystem/fe-filesystem-s3/src/main/java/org/apache/doris/filesystem/s3/S3FileSystemProperties.java @@ -63,6 +63,10 @@ public final class S3FileSystemProperties public static final String CONNECTION_TIMEOUT_MS = "s3.connection.timeout"; public static final String USE_PATH_STYLE = "use_path_style"; public static final String CREDENTIALS_PROVIDER_TYPE = "s3.credentials_provider_type"; + private static final String S3_PROVIDER = "s3.provider"; + private static final String PROVIDER = "provider"; + // Must match AbstractS3CompatibleProperties.S3_EXPRESS_IMPORT_READ and the BE property key. + static final String S3_EXPRESS_IMPORT_READ = "__DORIS_S3_EXPRESS_IMPORT_READ__"; public static final String DEFAULT_MAX_CONNECTIONS = "50"; public static final String DEFAULT_REQUEST_TIMEOUT_MS = "3000"; @@ -72,11 +76,8 @@ public final class S3FileSystemProperties private static final Pattern[] ENDPOINT_PATTERNS = new Pattern[] { Pattern.compile( - "^(?:https?://)?(?:" - + "s3(?:[-.]fips)?(?:[-.]dualstack)?[-.]([a-z0-9-]+)|" - + "s3express-control\\.([a-z0-9-]+)|" - + "s3express-[a-z0-9-]+\\.([a-z0-9-]+)" - + ")\\.amazonaws\\.com(?:/.*)?$", + "^(?:https?://)?s3(?:[-.]fips)?(?:[-.]dualstack)?[-.]([a-z0-9-]+)" + + "\\.amazonaws\\.com(?:/.*)?$", Pattern.CASE_INSENSITIVE), Pattern.compile( "^(?:https?://)?glue(?:-fips)?\\.([a-z0-9-]+)\\.(amazonaws\\.com(?:\\.cn)?|api\\.aws)$", @@ -186,7 +187,9 @@ private S3FileSystemProperties(Map rawProperties) { this.rawProperties = Collections.unmodifiableMap(new HashMap<>(rawProperties)); this.matchedProperties = Collections.unmodifiableMap(collectMatchedProperties(rawProperties)); ConnectorPropertiesUtils.bindConnectorProperties(this, rawProperties); - normalizeForLegacyS3Compatibility(); + if (!isScopedAwsS3ExpressImport()) { + normalizeForLegacyS3Compatibility(); + } } /** Binds and validates raw properties. */ @@ -207,7 +210,8 @@ public void validate() { "s3.external_id must be used together with s3.role_arn") .check(this::hasUnsupportedCredentialsProviderType, "Unsupported s3.credentials_provider_type: " + credentialsProviderType) - .check(() -> StringUtils.isBlank(endpoint) && StringUtils.isBlank(region), + .check(() -> StringUtils.isBlank(endpoint) && StringUtils.isBlank(region) + && !isScopedAwsS3ExpressImport(), "Either s3.endpoint or s3.region must be set") .check(this::hasInvalidUsePathStyle, "use_path_style must be true or false, got: '" + getUsePathStyle() + "'") @@ -333,9 +337,24 @@ public boolean hasAssumeRole() { return StringUtils.isNotBlank(roleArn); } - public boolean isDirectoryBucketEndpoint() { - return StringUtils.containsIgnoreCase(endpoint, "s3express-control.") - || StringUtils.containsIgnoreCase(endpoint, "s3express-"); + boolean isS3ExpressImportReadEnabled() { + return Boolean.parseBoolean(rawProperties.get(S3_EXPRESS_IMPORT_READ)); + } + + boolean isAwsProvider() { + return isAwsProvider(rawProperties); + } + + boolean isScopedAwsS3ExpressImport() { + return isS3ExpressImportReadEnabled() && isAwsProvider(); + } + + static boolean isAwsProvider(Map properties) { + return properties.entrySet().stream() + .filter(entry -> entry.getKey().equalsIgnoreCase(S3_PROVIDER) + || entry.getKey().equalsIgnoreCase(PROVIDER)) + .map(Map.Entry::getValue) + .anyMatch(value -> "AWS".equalsIgnoreCase(value)); } private static void putIfNotBlank(Map map, String key, String value) { diff --git a/fe/fe-filesystem/fe-filesystem-s3/src/main/java/org/apache/doris/filesystem/s3/S3FileSystemProvider.java b/fe/fe-filesystem/fe-filesystem-s3/src/main/java/org/apache/doris/filesystem/s3/S3FileSystemProvider.java index 1022074791909b..0ae5e8c38fa6fd 100644 --- a/fe/fe-filesystem/fe-filesystem-s3/src/main/java/org/apache/doris/filesystem/s3/S3FileSystemProvider.java +++ b/fe/fe-filesystem/fe-filesystem-s3/src/main/java/org/apache/doris/filesystem/s3/S3FileSystemProvider.java @@ -61,6 +61,10 @@ public class S3FileSystemProvider implements FileSystemProvider properties) { + if (Boolean.parseBoolean(properties.get(S3FileSystemProperties.S3_EXPRESS_IMPORT_READ)) + && S3FileSystemProperties.isAwsProvider(properties)) { + return true; + } boolean hasCredential = hasAny(properties, ACCESS_KEY_NAMES) || hasAny(properties, ROLE_ARN_NAMES) || hasAny(properties, CREDENTIALS_PROVIDER_TYPE_NAMES); diff --git a/fe/fe-filesystem/fe-filesystem-s3/src/main/java/org/apache/doris/filesystem/s3/S3ObjStorage.java b/fe/fe-filesystem/fe-filesystem-s3/src/main/java/org/apache/doris/filesystem/s3/S3ObjStorage.java index 6ddf0434966023..ee6cdaae542251 100644 --- a/fe/fe-filesystem/fe-filesystem-s3/src/main/java/org/apache/doris/filesystem/s3/S3ObjStorage.java +++ b/fe/fe-filesystem/fe-filesystem-s3/src/main/java/org/apache/doris/filesystem/s3/S3ObjStorage.java @@ -76,8 +76,10 @@ import java.time.Duration; import java.util.ArrayList; import java.util.Collections; +import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.Set; import java.util.UUID; import java.util.concurrent.atomic.AtomicBoolean; @@ -91,6 +93,15 @@ public class S3ObjStorage implements ObjStorage { private static final Logger LOG = LogManager.getLogger(S3ObjStorage.class); + private static final String DIRECTORY_BUCKET_SUFFIX = "--x-s3"; + private static final Map S3_EXPRESS_REGIONS = Map.of( + "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"); /** Validity period for pre-signed URLs and STS tokens (seconds). */ private static final int SESSION_EXPIRE_SECONDS = 3600; @@ -100,6 +111,7 @@ public class S3ObjStorage implements ObjStorage { /** Bucket name; may be null if not provided (listObjectsWithPrefix and related methods will fail). */ private final String bucket; private final AtomicBoolean closed = new AtomicBoolean(false); + private final Map expressClients = new HashMap<>(); private volatile S3Client client; public S3ObjStorage(S3FileSystemProperties properties) { @@ -145,26 +157,35 @@ protected S3Client buildClient() throws IOException { return buildClient( s3Properties.getEndpoint(), s3Properties.getRegion(), - buildCredentialsProvider()); + buildCredentialsProvider(), false); } - private S3Client buildClient(String endpointStr, String region, AwsCredentialsProvider credentialsProvider) + protected S3Client buildExpressClient(String region) throws IOException { + return buildClient( + "", + region, + buildCredentialsProvider(region), true); + } + + private S3Client buildClient(String endpointStr, String region, + AwsCredentialsProvider clientCredentialsProvider, boolean express) throws IOException { S3ClientBuilder builder = S3Client.builder() .httpClient(UrlConnectionHttpClient.builder() .socketTimeout(Duration.ofSeconds(30)) .connectionTimeout(Duration.ofSeconds(30)) .build()) - .credentialsProvider(credentialsProvider) + .credentialsProvider(clientCredentialsProvider) .region(Region.of(region)) .serviceConfiguration(S3Configuration.builder() .chunkedEncodingEnabled(false) - .pathStyleAccessEnabled(usePathStyle) + .pathStyleAccessEnabled(express ? false : usePathStyle) .build()); + if (express) { + builder.disableS3ExpressSessionAuth(false); + } - // endpointOverride is only set for non-AWS endpoints (MinIO, COS, OSS, etc.). - // Standard AWS S3 access uses region-only routing without an explicit endpoint. - if (StringUtils.isNotBlank(endpointStr)) { + if (!express && StringUtils.isNotBlank(endpointStr)) { if (!endpointStr.contains("://")) { endpointStr = "https://" + endpointStr; } @@ -189,6 +210,35 @@ protected AwsCredentialsProvider buildCredentialsProvider() { return S3CredentialsProviderFactory.createClientProvider(s3Properties, this::buildStsClient); } + protected AwsCredentialsProvider buildCredentialsProvider(String region) { + return S3CredentialsProviderFactory.createClientProvider( + s3Properties, (sourceCredentials, ignoredUserRegion) -> + buildStsClient(sourceCredentials, region)); + } + + boolean usesS3ExpressRead(String requestBucket) { + return s3Properties.isScopedAwsS3ExpressImport() + && getS3ExpressZoneId(requestBucket).isPresent(); + } + + private S3Client getExpressClient(String requestBucket) throws IOException { + if (closed.get()) { + throw new IOException("S3ObjStorage is already closed"); + } + String region = getS3ExpressRegion(requestBucket); + synchronized (expressClients) { + if (closed.get()) { + throw new IOException("S3ObjStorage is already closed"); + } + S3Client expressClient = expressClients.get(region); + if (expressClient == null) { + expressClient = buildExpressClient(region); + expressClients.put(region, expressClient); + } + return expressClient; + } + } + private AwsCredentialsProvider buildStsSourceCredentialsProvider() { return S3CredentialsProviderFactory.createStsSourceProvider(s3Properties); } @@ -210,13 +260,20 @@ public RemoteObjects listObjects(String remotePath, String continuationToken) th @Override public RemoteObjects listObjectsWithOptions(String remotePath, ObjectListOptions options) throws IOException { ObjectStorageUri uri = ObjectStorageUri.parse(remotePath, usePathStyle, getSupportedSchemes()); + boolean expressRead = usesS3ExpressRead(uri.bucket()); + String requestPrefix = expressRead ? slashTerminatedPrefix(uri.key()) : uri.key(); ListObjectsV2Request.Builder builder = ListObjectsV2Request.builder() - .bucket(uri.bucket()) - .prefix(uri.key()); + .bucket(uri.bucket()); + if (!expressRead || !requestPrefix.isEmpty()) { + builder.prefix(requestPrefix); + } if (options != null) { if (StringUtils.isNotBlank(options.continuationToken())) { builder.continuationToken(options.continuationToken()); } else if (StringUtils.isNotBlank(options.startAfter())) { + if (expressRead) { + throw new IOException("StartAfter is not supported for AWS Directory Bucket listings"); + } builder.startAfter(options.startAfter()); } if (options.maxKeys() > 0) { @@ -227,7 +284,8 @@ public RemoteObjects listObjectsWithOptions(String remotePath, ObjectListOptions } } try { - ListObjectsV2Response response = getClient().listObjectsV2(builder.build()); + S3Client listClient = expressRead ? getExpressClient(uri.bucket()) : getClient(); + ListObjectsV2Response response = listClient.listObjectsV2(builder.build()); List objects = response.contents().stream() .map(s3Obj -> new org.apache.doris.filesystem.spi.RemoteObject( s3Obj.key(), @@ -238,11 +296,28 @@ public RemoteObjects listObjectsWithOptions(String remotePath, ObjectListOptions .collect(Collectors.toList()); return new RemoteObjects(objects, response.isTruncated(), response.nextContinuationToken()); + } catch (S3Exception e) { + if (expressRead) { + throw new IOException("Create S3 Express session or list directory bucket failed for " + + remotePath + ": " + formatS3Exception(e), e); + } + throw new IOException("Failed to list objects at " + remotePath + ": " + e.getMessage(), e); } catch (SdkException e) { + if (expressRead) { + throw new IOException("Create S3 Express session or list directory bucket failed for " + + remotePath + ": " + e.getMessage(), e); + } throw new IOException("Failed to list objects at " + remotePath + ": " + e.getMessage(), e); } } + private static String formatS3Exception(S3Exception exception) { + return "HTTP status=" + exception.statusCode() + + ", AWS error code=" + exception.awsErrorDetails().errorCode() + + ", message=" + exception.awsErrorDetails().errorMessage() + + ", request ID=" + exception.requestId(); + } + /** * Strips {@code prefix} (with implicit trailing-slash normalisation) from {@code key} * to produce the per-listing relative path. Identical normalisation to @@ -656,6 +731,60 @@ private static String getRelativePathSafe(String prefix, String key) { return key.substring(normalized.length()); } + private static Optional getS3ExpressZoneId(String bucketName) { + if (bucketName == null || !bucketName.endsWith(DIRECTORY_BUCKET_SUFFIX)) { + return Optional.empty(); + } + String nameAndZone = bucketName.substring( + 0, bucketName.length() - DIRECTORY_BUCKET_SUFFIX.length()); + int zoneSeparator = nameAndZone.lastIndexOf("--"); + if (zoneSeparator <= 0 || zoneSeparator + 2 == nameAndZone.length()) { + return Optional.empty(); + } + String zoneId = nameAndZone.substring(zoneSeparator + 2); + int azSeparator = zoneId.lastIndexOf("-az"); + if (azSeparator <= 0 || azSeparator + 3 == zoneId.length()) { + return Optional.empty(); + } + for (int i = 0; i < azSeparator; i++) { + char c = zoneId.charAt(i); + if (!(c >= 'a' && c <= 'z') && !(c >= '0' && c <= '9') && c != '-') { + return Optional.empty(); + } + } + for (int i = azSeparator + 3; i < zoneId.length(); i++) { + char c = zoneId.charAt(i); + if (c < '0' || c > '9') { + return Optional.empty(); + } + } + return Optional.of(zoneId); + } + + private String getS3ExpressRegion(String bucketName) throws IOException { + String zoneId = getS3ExpressZoneId(bucketName).orElseThrow( + () -> new IOException("Invalid AWS Directory Bucket name " + bucketName + + "; expected ----x-s3")); + String zonePrefix = zoneId.substring(0, zoneId.indexOf('-')); + String inferredRegion = S3_EXPRESS_REGIONS.get(zonePrefix); + if (inferredRegion != null) { + return inferredRegion; + } + if (StringUtils.isBlank(s3Properties.getRegion())) { + throw new IOException("Cannot infer AWS region from Directory Bucket zone " + zoneId + + "; set s3.region for newly introduced AWS regions"); + } + return s3Properties.getRegion(); + } + + private static String slashTerminatedPrefix(String key) { + if (key.isEmpty() || key.endsWith("/")) { + return key; + } + int slash = key.lastIndexOf('/'); + return slash < 0 ? "" : key.substring(0, slash + 1); + } + @Override public void close() throws IOException { if (closed.compareAndSet(false, true)) { @@ -663,6 +792,12 @@ public void close() throws IOException { client.close(); client = null; } + synchronized (expressClients) { + for (S3Client expressClient : expressClients.values()) { + expressClient.close(); + } + expressClients.clear(); + } } } } diff --git a/fe/fe-filesystem/fe-filesystem-s3/src/test/java/org/apache/doris/filesystem/s3/S3FileSystemProviderTest.java b/fe/fe-filesystem/fe-filesystem-s3/src/test/java/org/apache/doris/filesystem/s3/S3FileSystemProviderTest.java index fe5c21d9515460..1a27a7f98d43a1 100644 --- a/fe/fe-filesystem/fe-filesystem-s3/src/test/java/org/apache/doris/filesystem/s3/S3FileSystemProviderTest.java +++ b/fe/fe-filesystem/fe-filesystem-s3/src/test/java/org/apache/doris/filesystem/s3/S3FileSystemProviderTest.java @@ -89,6 +89,28 @@ void supports_acceptsExplicitS3SupportWithoutCredentials() { Assertions.assertTrue(provider.supports(props)); } + @Test + void supports_acceptsScopedAwsExpressWithoutEndpointOrRegion() { + Map props = new HashMap<>(); + props.put("s3.provider", "AWS"); + props.put(S3FileSystemProperties.S3_EXPRESS_IMPORT_READ, "true"); + + Assertions.assertTrue(provider.supports(props)); + S3FileSystemProperties bound = provider.bind(props); + Assertions.assertTrue(bound.isScopedAwsS3ExpressImport()); + Assertions.assertEquals("", bound.getEndpoint()); + Assertions.assertEquals("", bound.getRegion()); + } + + @Test + void supports_rejectsExpressMarkerWithoutExplicitAwsProvider() { + Map props = new HashMap<>(); + props.put("provider", "S3"); + props.put(S3FileSystemProperties.S3_EXPRESS_IMPORT_READ, "true"); + + Assertions.assertFalse(provider.supports(props)); + } + @Test void bind_returnsValidatedS3FileSystemProperties() { Map props = new HashMap<>(); diff --git a/fe/fe-filesystem/fe-filesystem-s3/src/test/java/org/apache/doris/filesystem/s3/S3FileSystemTest.java b/fe/fe-filesystem/fe-filesystem-s3/src/test/java/org/apache/doris/filesystem/s3/S3FileSystemTest.java index fe9b28dd8c90b4..ce2a2a85407f82 100644 --- a/fe/fe-filesystem/fe-filesystem-s3/src/test/java/org/apache/doris/filesystem/s3/S3FileSystemTest.java +++ b/fe/fe-filesystem/fe-filesystem-s3/src/test/java/org/apache/doris/filesystem/s3/S3FileSystemTest.java @@ -45,6 +45,8 @@ */ class S3FileSystemTest { + private static final String DIRECTORY_BUCKET = "analytics--usw2-az1--x-s3"; + private S3ObjStorage mockStorage; private S3FileSystem fs; @@ -676,13 +678,15 @@ void globListWithLimit_paginatesExpandedPrefixesInUtf8BinaryOrder() throws IOExc } @Test - void globListWithLimit_directoryBucketFallsBackToSlashTerminatedStaticPrefix() throws IOException { + void globListWithLimit_regionalDirectoryBucketUsesSlashTerminatedPrefix() throws IOException { S3FileSystemProperties properties = S3FileSystemProperties.of(Map.of( - "s3.endpoint", "https://s3express-usw2-az1.us-west-2.amazonaws.com", + "s3.endpoint", "https://s3.us-west-2.amazonaws.com", "s3.region", "us-west-2")); S3FileSystem directoryBucketFs = new S3FileSystem(properties, mockStorage); + Mockito.when(mockStorage.usesS3ExpressRead(DIRECTORY_BUCKET)).thenReturn(true); Mockito.when(mockStorage.listObjects( - ArgumentMatchers.eq("s3://bucket/data/"), ArgumentMatchers.isNull())) + ArgumentMatchers.eq("s3://" + DIRECTORY_BUCKET + "/data/"), + ArgumentMatchers.isNull())) .thenReturn(new RemoteObjects( List.of( new RemoteObject("data/a.csv", "a.csv", null, 10L, 0L), @@ -690,16 +694,77 @@ void globListWithLimit_directoryBucketFallsBackToSlashTerminatedStaticPrefix() t false, null)); GlobListing listing = directoryBucketFs.globListWithLimit( - Location.of("s3://bucket/data/[ab]*.csv"), null, 0L, 0L); + Location.of("s3://" + DIRECTORY_BUCKET + "/data/[ab]*.csv"), null, 0L, 0L); Assertions.assertEquals(2, listing.getFiles().size()); Assertions.assertEquals("data/", listing.getPrefix()); Mockito.verify(mockStorage).listObjects( - ArgumentMatchers.eq("s3://bucket/data/"), ArgumentMatchers.isNull()); - Mockito.verify(mockStorage, Mockito.never()).listObjects( - ArgumentMatchers.eq("s3://bucket/data/a"), ArgumentMatchers.any()); - Mockito.verify(mockStorage, Mockito.never()).listObjects( - ArgumentMatchers.eq("s3://bucket/data/b"), ArgumentMatchers.any()); + ArgumentMatchers.eq("s3://" + DIRECTORY_BUCKET + "/data/"), + ArgumentMatchers.isNull()); + } + + @Test + void globListWithLimit_regionalDirectoryBucketExactPathFiltersSibling() throws IOException { + S3FileSystemProperties properties = S3FileSystemProperties.of(Map.of( + "s3.endpoint", "https://s3.us-west-2.amazonaws.com", + "s3.region", "us-west-2")); + S3FileSystem directoryBucketFs = new S3FileSystem(properties, mockStorage); + Mockito.when(mockStorage.usesS3ExpressRead(DIRECTORY_BUCKET)).thenReturn(true); + Mockito.when(mockStorage.listObjects( + ArgumentMatchers.eq("s3://" + DIRECTORY_BUCKET + "/data/"), + ArgumentMatchers.isNull())) + .thenReturn(new RemoteObjects( + List.of( + new RemoteObject("data/target.csv", "target.csv", null, 10L, 0L), + new RemoteObject("data/target.csv.bak", "target.csv.bak", null, 20L, 0L)), + false, null)); + + GlobListing listing = directoryBucketFs.globListWithLimit( + Location.of("s3://" + DIRECTORY_BUCKET + "/data/target.csv"), null, 0L, 0L); + + Assertions.assertEquals(1, listing.getFiles().size()); + Assertions.assertEquals( + "s3://" + DIRECTORY_BUCKET + "/data/target.csv", + listing.getFiles().get(0).location().uri()); + } + + @Test + void globListWithLimit_regionalDirectoryBucketPaginatesWithOpaqueToken() throws IOException { + S3FileSystemProperties properties = S3FileSystemProperties.of(Map.of( + "s3.endpoint", "https://s3.us-west-2.amazonaws.com", + "s3.region", "us-west-2")); + S3FileSystem directoryBucketFs = new S3FileSystem(properties, mockStorage); + String listUri = "s3://" + DIRECTORY_BUCKET + "/data/"; + Mockito.when(mockStorage.usesS3ExpressRead(DIRECTORY_BUCKET)).thenReturn(true); + Mockito.doAnswer(invocation -> { + ObjectListOptions options = invocation.getArgument(1); + if (options.continuationToken() == null) { + return new RemoteObjects( + List.of(new RemoteObject("data/b.csv", "b.csv", null, 20L, 0L)), + true, "opaque-token"); + } + return new RemoteObjects( + List.of(new RemoteObject("data/a.csv", "a.csv", null, 10L, 0L)), + false, null); + }).when(mockStorage).listObjectsWithOptions( + ArgumentMatchers.eq(listUri), ArgumentMatchers.any(ObjectListOptions.class)); + + GlobListing listing = directoryBucketFs.globListWithLimit( + Location.of("s3://" + DIRECTORY_BUCKET + "/data/*.csv"), null, 0L, 0L); + + Assertions.assertEquals(2, listing.getFiles().size()); + Assertions.assertEquals( + List.of( + "s3://" + DIRECTORY_BUCKET + "/data/b.csv", + "s3://" + DIRECTORY_BUCKET + "/data/a.csv"), + listing.getFiles().stream().map(file -> file.location().uri()).toList()); + ArgumentCaptor optionsCaptor = ArgumentCaptor.forClass(ObjectListOptions.class); + Mockito.verify(mockStorage, Mockito.times(2)).listObjectsWithOptions( + ArgumentMatchers.eq(listUri), optionsCaptor.capture()); + Assertions.assertNull(optionsCaptor.getAllValues().get(0).startAfter()); + Assertions.assertEquals( + "opaque-token", optionsCaptor.getAllValues().get(1).continuationToken()); + Assertions.assertNull(optionsCaptor.getAllValues().get(1).startAfter()); } @Test diff --git a/fe/fe-filesystem/fe-filesystem-s3/src/test/java/org/apache/doris/filesystem/s3/S3ObjStorageMockTest.java b/fe/fe-filesystem/fe-filesystem-s3/src/test/java/org/apache/doris/filesystem/s3/S3ObjStorageMockTest.java index deefd3925a5855..cc6a74c26a937f 100644 --- a/fe/fe-filesystem/fe-filesystem-s3/src/test/java/org/apache/doris/filesystem/s3/S3ObjStorageMockTest.java +++ b/fe/fe-filesystem/fe-filesystem-s3/src/test/java/org/apache/doris/filesystem/s3/S3ObjStorageMockTest.java @@ -18,6 +18,7 @@ package org.apache.doris.filesystem.s3; import org.apache.doris.filesystem.UploadPartResult; +import org.apache.doris.filesystem.spi.ObjectListOptions; import org.apache.doris.filesystem.spi.RemoteObject; import org.apache.doris.filesystem.spi.RemoteObjects; import org.apache.doris.filesystem.spi.RequestBody; @@ -31,6 +32,8 @@ import software.amazon.awssdk.auth.credentials.AnonymousCredentialsProvider; import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider; import software.amazon.awssdk.auth.credentials.EnvironmentVariableCredentialsProvider; +import software.amazon.awssdk.awscore.exception.AwsErrorDetails; +import software.amazon.awssdk.core.exception.SdkException; import software.amazon.awssdk.services.s3.S3Client; import software.amazon.awssdk.services.s3.model.AbortMultipartUploadRequest; import software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest; @@ -57,16 +60,22 @@ import java.io.FileNotFoundException; import java.io.IOException; import java.time.Instant; +import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Set; +import java.util.function.Function; /** * Full unit tests for {@link S3ObjStorage} using a testable subclass that overrides - * {@link S3ObjStorage#buildClient()} to inject a mock S3Client. + * {@link S3ObjStorage#buildClient()} and {@link S3ObjStorage#buildExpressClient(String)} + * to inject mock S3 clients. */ class S3ObjStorageMockTest { + private static final String DIRECTORY_BUCKET = "analytics--usw2-az1--x-s3"; + private S3Client mockS3; private S3ObjStorage storage; @@ -150,6 +159,307 @@ void listObjects_truncatedResultReturnsContinuationToken() throws IOException { Assertions.assertEquals("next-token", result.getContinuationToken()); } + @Test + void listObjects_directoryBucketUsesExpressClientAndDirectoryPrefix() throws IOException { + S3Client regularClient = Mockito.mock(S3Client.class); + S3Client expressClient = Mockito.mock(S3Client.class); + S3ObjStorage expressStorage = directoryBucketStorage( + "https://s3.us-west-2.amazonaws.com", "us-west-2", false, + regularClient, expressClient); + Mockito.when(expressClient.listObjectsV2(ArgumentMatchers.any(ListObjectsV2Request.class))) + .thenReturn(ListObjectsV2Response.builder().contents(List.of()).isTruncated(false).build()); + + expressStorage.listObjects("s3://" + DIRECTORY_BUCKET + "/data/file-*.csv", null); + + ArgumentCaptor captor = ArgumentCaptor.forClass(ListObjectsV2Request.class); + Mockito.verify(expressClient).listObjectsV2(captor.capture()); + Assertions.assertEquals(DIRECTORY_BUCKET, captor.getValue().bucket()); + Assertions.assertEquals("data/", captor.getValue().prefix()); + Assertions.assertNull(captor.getValue().startAfter()); + Mockito.verify(regularClient, Mockito.never()).listObjectsV2( + ArgumentMatchers.any(ListObjectsV2Request.class)); + } + + @Test + void listObjects_directoryBucketUsesOpaqueContinuationToken() throws IOException { + S3Client regularClient = Mockito.mock(S3Client.class); + S3Client expressClient = Mockito.mock(S3Client.class); + S3ObjStorage expressStorage = directoryBucketStorage( + "s3.us-west-2.amazonaws.com", "us-west-2", false, + regularClient, expressClient); + Mockito.when(expressClient.listObjectsV2(ArgumentMatchers.any(ListObjectsV2Request.class))) + .thenReturn(ListObjectsV2Response.builder().contents(List.of()).isTruncated(false).build()); + + expressStorage.listObjects("s3://" + DIRECTORY_BUCKET + "/data/file.csv", "opaque-token"); + + ArgumentCaptor captor = ArgumentCaptor.forClass(ListObjectsV2Request.class); + Mockito.verify(expressClient).listObjectsV2(captor.capture()); + Assertions.assertEquals("opaque-token", captor.getValue().continuationToken()); + Assertions.assertNull(captor.getValue().startAfter()); + } + + @Test + void listObjects_directoryBucketRootDoesNotSendPrefix() throws IOException { + S3Client regularClient = Mockito.mock(S3Client.class); + S3Client expressClient = Mockito.mock(S3Client.class); + S3ObjStorage expressStorage = directoryBucketStorage( + "https://s3.us-west-2.amazonaws.com", "us-west-2", false, + regularClient, expressClient); + Mockito.when(expressClient.listObjectsV2(ArgumentMatchers.any(ListObjectsV2Request.class))) + .thenReturn(ListObjectsV2Response.builder().contents(List.of()).isTruncated(false).build()); + + expressStorage.listObjects("s3://" + DIRECTORY_BUCKET + "/", null); + + ArgumentCaptor captor = ArgumentCaptor.forClass(ListObjectsV2Request.class); + Mockito.verify(expressClient).listObjectsV2(captor.capture()); + Assertions.assertNull(captor.getValue().prefix()); + Mockito.verifyNoInteractions(regularClient); + } + + @Test + void listObjects_directoryBucketS3ExceptionPreservesServiceErrorDetails() { + S3Client regularClient = Mockito.mock(S3Client.class); + S3Client expressClient = Mockito.mock(S3Client.class); + S3ObjStorage expressStorage = directoryBucketStorage( + "", "us-west-2", false, regularClient, expressClient); + S3Exception failure = (S3Exception) S3Exception.builder() + .statusCode(403) + .requestId("request-123") + .awsErrorDetails(AwsErrorDetails.builder() + .serviceName("S3") + .errorCode("AccessDenied") + .errorMessage("CreateSession is not allowed") + .build()) + .message("CreateSession is not allowed") + .build(); + Mockito.when(expressClient.listObjectsV2(ArgumentMatchers.any(ListObjectsV2Request.class))) + .thenThrow(failure); + + IOException exception = Assertions.assertThrows(IOException.class, + () -> expressStorage.listObjects( + "s3://" + DIRECTORY_BUCKET + "/data/file.parquet", null)); + + Assertions.assertAll( + () -> Assertions.assertTrue(exception.getMessage().contains( + "Create S3 Express session or list directory bucket failed")), + () -> Assertions.assertTrue(exception.getMessage().contains("HTTP status=403")), + () -> Assertions.assertTrue(exception.getMessage().contains("AWS error code=AccessDenied")), + () -> Assertions.assertTrue(exception.getMessage().contains( + "message=CreateSession is not allowed")), + () -> Assertions.assertTrue(exception.getMessage().contains("request ID=request-123")), + () -> Assertions.assertSame(failure, exception.getCause())); + Mockito.verifyNoInteractions(regularClient); + } + + @Test + void listObjects_directoryBucketSdkExceptionUsesCombinedOperationContext() { + S3Client regularClient = Mockito.mock(S3Client.class); + S3Client expressClient = Mockito.mock(S3Client.class); + S3ObjStorage expressStorage = directoryBucketStorage( + "", "us-west-2", false, regularClient, expressClient); + SdkException failure = SdkException.builder().message("network down").build(); + Mockito.when(expressClient.listObjectsV2(ArgumentMatchers.any(ListObjectsV2Request.class))) + .thenThrow(failure); + + IOException exception = Assertions.assertThrows(IOException.class, + () -> expressStorage.listObjects( + "s3://" + DIRECTORY_BUCKET + "/data/file.parquet", null)); + + Assertions.assertTrue(exception.getMessage().contains( + "Create S3 Express session or list directory bucket failed")); + Assertions.assertTrue(exception.getMessage().contains("network down")); + Assertions.assertSame(failure, exception.getCause()); + Mockito.verifyNoInteractions(regularClient); + } + + @Test + void listObjects_regularBucketKeepsExistingS3ExceptionContext() { + S3Exception failure = (S3Exception) S3Exception.builder() + .statusCode(403) + .requestId("request-ordinary") + .awsErrorDetails(AwsErrorDetails.builder() + .serviceName("S3") + .errorCode("AccessDenied") + .errorMessage("Forbidden") + .build()) + .message("Forbidden") + .build(); + Mockito.when(mockS3.listObjectsV2(ArgumentMatchers.any(ListObjectsV2Request.class))) + .thenThrow(failure); + + IOException exception = Assertions.assertThrows(IOException.class, + () -> storage.listObjects("s3://my-bucket/data/file.parquet", null)); + + Assertions.assertTrue(exception.getMessage().startsWith("Failed to list objects at")); + Assertions.assertFalse(exception.getMessage().contains("S3 Express")); + Assertions.assertSame(failure, exception.getCause()); + } + + @Test + void listObjects_directoryBucketWithoutImportMarkerStaysOnRegularClient() throws IOException { + S3Client regularClient = Mockito.mock(S3Client.class); + S3Client expressClient = Mockito.mock(S3Client.class); + S3ObjStorage nonImportStorage = new TestableS3ObjStorage( + directoryBucketProperties("https://s3.us-west-2.amazonaws.com", "us-west-2", false), + regularClient, expressClient); + Mockito.when(regularClient.listObjectsV2(ArgumentMatchers.any(ListObjectsV2Request.class))) + .thenReturn(ListObjectsV2Response.builder().contents(List.of()).isTruncated(false).build()); + + nonImportStorage.listObjectsWithOptions( + "s3://" + DIRECTORY_BUCKET + "/data/file.csv", + ObjectListOptions.builder().startAfter("data/previous.csv").build()); + + ArgumentCaptor captor = ArgumentCaptor.forClass(ListObjectsV2Request.class); + Mockito.verify(regularClient).listObjectsV2(captor.capture()); + Assertions.assertEquals("data/file.csv", captor.getValue().prefix()); + Assertions.assertEquals("data/previous.csv", captor.getValue().startAfter()); + Mockito.verifyNoInteractions(expressClient); + } + + @Test + void listObjects_directoryBucketRejectsStartAfter() { + S3Client regularClient = Mockito.mock(S3Client.class); + S3Client expressClient = Mockito.mock(S3Client.class); + S3ObjStorage expressStorage = directoryBucketStorage( + "https://s3.us-west-2.amazonaws.com", "us-west-2", false, + regularClient, expressClient); + + IOException exception = Assertions.assertThrows(IOException.class, + () -> expressStorage.listObjectsWithOptions( + "s3://" + DIRECTORY_BUCKET + "/data/file.csv", + ObjectListOptions.builder().startAfter("data/previous.csv").build())); + + Assertions.assertTrue(exception.getMessage().contains("StartAfter")); + Mockito.verifyNoInteractions(regularClient, expressClient); + } + + @Test + void listObjects_directoryBucketIgnoresConfiguredEndpoint() throws IOException { + S3Client regularClient = Mockito.mock(S3Client.class); + S3Client expressClient = Mockito.mock(S3Client.class); + S3ObjStorage expressStorage = directoryBucketStorage( + "https://minio.example.com", "us-west-2", false, + regularClient, expressClient); + Mockito.when(expressClient.listObjectsV2(ArgumentMatchers.any(ListObjectsV2Request.class))) + .thenReturn(ListObjectsV2Response.builder().contents(List.of()).isTruncated(false).build()); + + expressStorage.listObjects("s3://" + DIRECTORY_BUCKET + "/data/file", null); + + ArgumentCaptor captor = ArgumentCaptor.forClass(ListObjectsV2Request.class); + Mockito.verify(expressClient).listObjectsV2(captor.capture()); + Assertions.assertEquals("data/", captor.getValue().prefix()); + Mockito.verifyNoInteractions(regularClient); + } + + @Test + void usesS3ExpressRead_requiresScopeProviderAndCompleteBucketName() { + S3Client regularClient = Mockito.mock(S3Client.class); + S3Client expressClient = Mockito.mock(S3Client.class); + + Assertions.assertTrue(directoryBucketStorage( + "http://ignored.example.com", "wrong-region", true, + regularClient, expressClient).usesS3ExpressRead(DIRECTORY_BUCKET)); + Map nonAwsProperties = directoryBucketProperties("", "us-west-2", false); + nonAwsProperties.put(S3FileSystemProperties.S3_EXPRESS_IMPORT_READ, "true"); + nonAwsProperties.put("provider", "S3"); + Assertions.assertFalse(new TestableS3ObjStorage( + nonAwsProperties, regularClient, expressClient).usesS3ExpressRead(DIRECTORY_BUCKET)); + + S3ObjStorage expressStorage = directoryBucketStorage( + "", "us-west-2", false, regularClient, expressClient); + Assertions.assertTrue(expressStorage.usesS3ExpressRead( + "analytics--archive--usw2-az1--x-s3")); + Assertions.assertFalse(expressStorage.usesS3ExpressRead("ordinary-bucket")); + Assertions.assertFalse(expressStorage.usesS3ExpressRead("invalid--x-s3")); + Assertions.assertFalse(expressStorage.usesS3ExpressRead("--usw2-az1--x-s3")); + Assertions.assertFalse(expressStorage.usesS3ExpressRead("analytics----x-s3")); + Assertions.assertFalse(expressStorage.usesS3ExpressRead("analytics--usw2-az--x-s3")); + Assertions.assertFalse(expressStorage.usesS3ExpressRead("analytics--usw2-azx--x-s3")); + Assertions.assertFalse(expressStorage.usesS3ExpressRead("analytics--USW2-az1--x-s3")); + } + + @Test + void listObjects_knownZonePrefixesOverrideConfiguredRegion() throws IOException { + S3Client regularClient = Mockito.mock(S3Client.class); + S3Client expressClient = Mockito.mock(S3Client.class); + Mockito.when(expressClient.listObjectsV2(ArgumentMatchers.any(ListObjectsV2Request.class))) + .thenReturn(ListObjectsV2Response.builder().contents(List.of()).isTruncated(false).build()); + TestableS3ObjStorage expressStorage = directoryBucketStorage( + "https://ignored.example.com", "wrong-region", false, + regularClient, expressClient); + Map expectedRegions = Map.of( + "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"); + + for (String zonePrefix : expectedRegions.keySet()) { + expressStorage.listObjects( + "s3://analytics--" + zonePrefix + "-az1--x-s3/data/file.csv", null); + } + + Assertions.assertEquals( + Set.copyOf(expectedRegions.values()), + Set.copyOf(expressStorage.getBuiltExpressRegions())); + Mockito.verifyNoInteractions(regularClient); + } + + @Test + void listObjects_unknownZonePrefixUsesConfiguredRegionVerbatim() throws IOException { + S3Client regularClient = Mockito.mock(S3Client.class); + S3Client expressClient = Mockito.mock(S3Client.class); + Mockito.when(expressClient.listObjectsV2(ArgumentMatchers.any(ListObjectsV2Request.class))) + .thenReturn(ListObjectsV2Response.builder().contents(List.of()).isTruncated(false).build()); + TestableS3ObjStorage expressStorage = directoryBucketStorage( + "https://ignored.example.com", "me-central-future-1", false, + regularClient, expressClient); + + expressStorage.listObjects( + "s3://analytics--mec9-az1--x-s3/data/file.csv", null); + + Assertions.assertEquals( + List.of("me-central-future-1"), expressStorage.getBuiltExpressRegions()); + } + + @Test + void listObjects_unknownZonePrefixWithoutConfiguredRegionFails() { + S3Client regularClient = Mockito.mock(S3Client.class); + S3Client expressClient = Mockito.mock(S3Client.class); + TestableS3ObjStorage expressStorage = directoryBucketStorage( + "https://ignored.example.com", "", false, + regularClient, expressClient); + + IOException exception = Assertions.assertThrows(IOException.class, + () -> expressStorage.listObjects( + "s3://analytics--mec9-az1--x-s3/data/file.csv", null)); + + Assertions.assertTrue(exception.getMessage().contains("set s3.region")); + Mockito.verifyNoInteractions(regularClient, expressClient); + } + + @Test + void listObjects_knownZoneWithoutEndpointOrRegionCachesClientByEffectiveRegion() throws IOException { + S3Client regularClient = Mockito.mock(S3Client.class); + S3Client expressClient = Mockito.mock(S3Client.class); + Mockito.when(expressClient.listObjectsV2(ArgumentMatchers.any(ListObjectsV2Request.class))) + .thenReturn(ListObjectsV2Response.builder().contents(List.of()).isTruncated(false).build()); + TestableS3ObjStorage expressStorage = directoryBucketStorage( + "", "", true, regularClient, expressClient); + + expressStorage.listObjects( + "s3://analytics--usw2-az1--x-s3/data/file.csv", null); + expressStorage.listObjects( + "s3://logs--archive--usw2-lax1-az1--x-s3/data/file.csv", null); + + Assertions.assertEquals(List.of("us-west-2"), expressStorage.getBuiltExpressRegions()); + Mockito.verify(expressClient, Mockito.times(2)).listObjectsV2( + ArgumentMatchers.any(ListObjectsV2Request.class)); + Mockito.verifyNoInteractions(regularClient); + } + // ------------------------------------------------------------------ // headObject() // ------------------------------------------------------------------ @@ -171,6 +481,22 @@ void headObject_returnsRemoteObjectOnSuccess() throws IOException { Assertions.assertEquals("etag-xyz", result.getEtag()); } + @Test + void headObject_directoryBucketStaysOnRegularClient() throws IOException { + S3Client regularClient = Mockito.mock(S3Client.class); + S3Client expressClient = Mockito.mock(S3Client.class); + S3ObjStorage expressStorage = directoryBucketStorage( + "https://s3.us-west-2.amazonaws.com", "us-west-2", false, + regularClient, expressClient); + Mockito.when(regularClient.headObject(ArgumentMatchers.any(HeadObjectRequest.class))) + .thenReturn(HeadObjectResponse.builder().contentLength(1L).build()); + + expressStorage.headObject("s3://" + DIRECTORY_BUCKET + "/data/file.csv"); + + Mockito.verify(regularClient).headObject(ArgumentMatchers.any(HeadObjectRequest.class)); + Mockito.verifyNoInteractions(expressClient); + } + @Test void headObject_throwsFileNotFoundForNoSuchKeyException() { Mockito.when(mockS3.headObject(ArgumentMatchers.any(HeadObjectRequest.class))) @@ -209,6 +535,27 @@ void putObject_delegatesToS3Client() throws IOException { Assertions.assertEquals("obj", captor.getValue().key()); } + @Test + void putObject_directoryBucketStaysOnRegularClient() throws IOException { + S3Client regularClient = Mockito.mock(S3Client.class); + S3Client expressClient = Mockito.mock(S3Client.class); + S3ObjStorage expressStorage = directoryBucketStorage( + "https://s3.us-west-2.amazonaws.com", "us-west-2", false, + regularClient, expressClient); + Mockito.when(regularClient.putObject( + ArgumentMatchers.any(PutObjectRequest.class), + ArgumentMatchers.any(software.amazon.awssdk.core.sync.RequestBody.class))) + .thenReturn(PutObjectResponse.builder().build()); + + RequestBody body = RequestBody.of(new ByteArrayInputStream(new byte[]{1}), 1); + expressStorage.putObject("s3://" + DIRECTORY_BUCKET + "/data/file.csv", body); + + Mockito.verify(regularClient).putObject( + ArgumentMatchers.any(PutObjectRequest.class), + ArgumentMatchers.any(software.amazon.awssdk.core.sync.RequestBody.class)); + Mockito.verifyNoInteractions(expressClient); + } + // ------------------------------------------------------------------ // deleteObject() // ------------------------------------------------------------------ @@ -427,6 +774,23 @@ void buildCredentialsProvider_returnsAssumeRoleProviderWhenRoleArnConfigured() { credentialsProvider.getClass().getSimpleName()); } + @Test + void buildExpressCredentialsProvider_usesRegionInferredFromBucketForAssumeRole() { + Map props = directoryBucketProperties("", "", false); + props.remove("AWS_ACCESS_KEY"); + props.remove("AWS_SECRET_KEY"); + props.put("AWS_ROLE_ARN", "arn:aws:iam::123456789012:role/MyRole"); + props.put(S3FileSystemProperties.S3_EXPRESS_IMPORT_READ, "true"); + InspectableS3ObjStorage roleArnStorage = new InspectableS3ObjStorage(props, mockS3); + + AwsCredentialsProvider credentialsProvider = + roleArnStorage.inspectBuildCredentialsProvider("us-west-2"); + + Assertions.assertEquals("StsAssumeRoleCredentialsProvider", + credentialsProvider.getClass().getSimpleName()); + Assertions.assertEquals(List.of("us-west-2"), roleArnStorage.getBuiltStsRegions()); + } + @Test void buildCredentialsProvider_usesTypedCredentialsProviderType() { Map props = new HashMap<>(); @@ -478,30 +842,101 @@ void close_closesS3Client() throws IOException { Mockito.verify(mockS3).close(); } + @Test + void close_closesRegularAndAllRegionalExpressClients() throws IOException { + S3Client regularClient = Mockito.mock(S3Client.class); + S3Client westExpressClient = Mockito.mock(S3Client.class); + S3Client eastExpressClient = Mockito.mock(S3Client.class); + Map props = directoryBucketProperties("", "", false); + props.put(S3FileSystemProperties.S3_EXPRESS_IMPORT_READ, "true"); + S3ObjStorage expressStorage = new TestableS3ObjStorage( + props, regularClient, region -> region.equals("us-west-2") + ? westExpressClient : eastExpressClient); + Mockito.when(westExpressClient.listObjectsV2(ArgumentMatchers.any(ListObjectsV2Request.class))) + .thenReturn(ListObjectsV2Response.builder().contents(List.of()).isTruncated(false).build()); + Mockito.when(eastExpressClient.listObjectsV2(ArgumentMatchers.any(ListObjectsV2Request.class))) + .thenReturn(ListObjectsV2Response.builder().contents(List.of()).isTruncated(false).build()); + expressStorage.getClient(); + expressStorage.listObjects("s3://analytics--usw2-az1--x-s3/data/", null); + expressStorage.listObjects("s3://analytics--use1-az1--x-s3/data/", null); + + expressStorage.close(); + + Mockito.verify(regularClient).close(); + Mockito.verify(westExpressClient).close(); + Mockito.verify(eastExpressClient).close(); + } + // ------------------------------------------------------------------ // Test infrastructure // ------------------------------------------------------------------ private static class TestableS3ObjStorage extends S3ObjStorage { private final S3Client mockClient; + private final Function expressClientFactory; + private final List builtExpressRegions = new ArrayList<>(); TestableS3ObjStorage(Map properties, S3Client mockClient) { + this(properties, mockClient, mockClient); + } + + TestableS3ObjStorage(Map properties, S3Client mockClient, + S3Client mockExpressClient) { + this(properties, mockClient, region -> mockExpressClient); + } + + TestableS3ObjStorage(Map properties, S3Client mockClient, + Function expressClientFactory) { super(properties); this.mockClient = mockClient; + this.expressClientFactory = expressClientFactory; } TestableS3ObjStorage(S3FileSystemProperties properties, S3Client mockClient) { super(properties); this.mockClient = mockClient; + this.expressClientFactory = region -> mockClient; } @Override protected S3Client buildClient() { return mockClient; } + + @Override + protected S3Client buildExpressClient(String region) { + builtExpressRegions.add(region); + return expressClientFactory.apply(region); + } + + List getBuiltExpressRegions() { + return builtExpressRegions; + } + } + + private static TestableS3ObjStorage directoryBucketStorage(String endpoint, String region, + boolean usePathStyle, S3Client regularClient, S3Client expressClient) { + Map props = directoryBucketProperties(endpoint, region, usePathStyle); + props.put(S3FileSystemProperties.S3_EXPRESS_IMPORT_READ, "true"); + return new TestableS3ObjStorage(props, regularClient, expressClient); + } + + private static Map directoryBucketProperties(String endpoint, String region, + boolean usePathStyle) { + Map props = new HashMap<>(); + props.put("AWS_ENDPOINT", endpoint); + props.put("AWS_REGION", region); + props.put("provider", "AWS"); + props.put("AWS_ACCESS_KEY", "testAK"); + props.put("AWS_SECRET_KEY", "testSK"); + props.put("AWS_BUCKET", DIRECTORY_BUCKET); + props.put("use_path_style", Boolean.toString(usePathStyle)); + return props; } private static class InspectableS3ObjStorage extends TestableS3ObjStorage { + private final List builtStsRegions = new ArrayList<>(); + InspectableS3ObjStorage(Map properties, S3Client mockClient) { super(properties, mockClient); } @@ -514,8 +949,17 @@ AwsCredentialsProvider inspectBuildCredentialsProvider() { return buildCredentialsProvider(); } + AwsCredentialsProvider inspectBuildCredentialsProvider(String region) { + return buildCredentialsProvider(region); + } + + List getBuiltStsRegions() { + return builtStsRegions; + } + @Override protected StsClient buildStsClient(AwsCredentialsProvider credentialsProvider, String region) { + builtStsRegions.add(region); return Mockito.mock(StsClient.class); } } diff --git a/fe/fe-filesystem/fe-filesystem-s3/src/test/java/org/apache/doris/filesystem/s3/S3ObjStorageTest.java b/fe/fe-filesystem/fe-filesystem-s3/src/test/java/org/apache/doris/filesystem/s3/S3ObjStorageTest.java index c5c92619af314e..3e2ff057f00b3d 100644 --- a/fe/fe-filesystem/fe-filesystem-s3/src/test/java/org/apache/doris/filesystem/s3/S3ObjStorageTest.java +++ b/fe/fe-filesystem/fe-filesystem-s3/src/test/java/org/apache/doris/filesystem/s3/S3ObjStorageTest.java @@ -21,7 +21,9 @@ import org.junit.jupiter.api.Test; import software.amazon.awssdk.regions.Region; import software.amazon.awssdk.services.s3.S3Client; +import software.amazon.awssdk.services.s3.model.GetUrlRequest; +import java.net.URL; import java.util.HashMap; import java.util.Map; @@ -48,6 +50,30 @@ void getClient_endpointOnlyConfigurationUsesRegionBuiltByProperties() throws Exc } } + @Test + void buildExpressClient_usesSdkZonalEndpointAndVirtualHostedAddressing() throws Exception { + Map props = new HashMap<>(); + props.put("AWS_ENDPOINT", "https://endpoint-is-ignored.example.com"); + props.put("AWS_REGION", "us-east-1"); + props.put("AWS_ACCESS_KEY", "ak"); + props.put("AWS_SECRET_KEY", "sk"); + props.put("use_path_style", "true"); + + S3ObjStorage storage = new S3ObjStorage(props); + try (S3Client client = storage.buildExpressClient("us-west-2")) { + URL url = client.utilities().getUrl(GetUrlRequest.builder() + .bucket("analytics--usw2-az1--x-s3") + .key("data.parquet") + .build()); + + Assertions.assertEquals(Region.US_WEST_2, client.serviceClientConfiguration().region()); + Assertions.assertEquals("https", url.getProtocol()); + Assertions.assertEquals( + "analytics--usw2-az1--x-s3.s3express-usw2-az1.us-west-2.amazonaws.com", + url.getHost()); + } + } + // ------------------------------------------------------------------ // close() // ------------------------------------------------------------------ diff --git a/fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/S3CompatibleFileSystem.java b/fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/S3CompatibleFileSystem.java index 5faee2bce6960d..773add5b7445e1 100644 --- a/fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/S3CompatibleFileSystem.java +++ b/fe/fe-filesystem/fe-filesystem-spi/src/main/java/org/apache/doris/filesystem/spi/S3CompatibleFileSystem.java @@ -868,11 +868,12 @@ protected static List expandedGlobListPrefixes(String globPattern) { return prefixes == null ? List.of(longestNonGlobPrefix(globPattern)) : prefixes; } - protected String globListPrefix(String globPattern) { + protected String globListPrefix(String bucket, String globPattern) throws IOException { return longestNonGlobPrefix(globPattern); } - protected List globListPrefixes(String globPattern, String listPrefix) { + protected List globListPrefixes(String bucket, String globPattern, String listPrefix) + throws IOException { return expandedGlobListPrefixes(globPattern); } @@ -1205,8 +1206,8 @@ public GlobListing globListWithLimit(Location path, String startAfter, long maxB // separator is '\' which would corrupt object storage keys, and (b) Paths.get rejects keys // containing characters illegal in the host OS path syntax (':', '\', etc.). Pattern matcher = Pattern.compile(globToRegex(expandedKeyPattern)); - String listPrefix = globListPrefix(expandedKeyPattern); - List listPrefixes = globListPrefixes(expandedKeyPattern, listPrefix); + String listPrefix = globListPrefix(bucket, expandedKeyPattern); + List listPrefixes = globListPrefixes(bucket, expandedKeyPattern, listPrefix); List files = new ArrayList<>(); long totalSize = 0L;