Skip to content

Commit 85cfe4e

Browse files
raulcdraulcd
committed
GH-50126: [C++] Make S3Filesystem consume key-value options
Fix linting after rebase Add s3 fs module test for FromUriAndOptionsCredentials and refactor S3Options::FromUri and S3Options::FromUriAndOptions Add tests for FromUriAndOptions credentials Add S3RetryStrategy to options Add option for default_metadata Minor fix
1 parent c9c887c commit 85cfe4e

4 files changed

Lines changed: 248 additions & 31 deletions

File tree

‎cpp/src/arrow/filesystem/s3fs.cc‎

Lines changed: 99 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -327,7 +327,68 @@ S3Options S3Options::FromAssumeRoleWithWebIdentity() {
327327
}
328328

329329
Result<S3Options> S3Options::FromUri(const Uri& uri, std::string* out_path) {
330-
S3Options options;
330+
return FromUriAndOptions(uri, FileSystemFactoryOptions{}, out_path);
331+
}
332+
333+
Result<S3Options> S3Options::FromUri(const std::string& uri_string,
334+
std::string* out_path) {
335+
Uri uri;
336+
RETURN_NOT_OK(uri.Parse(uri_string));
337+
return FromUri(uri, out_path);
338+
}
339+
340+
namespace {
341+
342+
template <typename T>
343+
Result<T> GetOption(const std::string& key, const std::any& value) {
344+
if (const auto* v = std::any_cast<T>(&value)) {
345+
return *v;
346+
}
347+
return Status::Invalid("S3 filesystem option '", key, "' has the wrong type");
348+
}
349+
350+
template <typename T>
351+
Result<std::shared_ptr<const T>> GetConstSharedPtrOption(const std::string& key,
352+
const std::any& value) {
353+
if (const auto* v = std::any_cast<std::shared_ptr<const T>>(&value)) return *v;
354+
if (const auto* v = std::any_cast<std::shared_ptr<T>>(&value)) return *v;
355+
return Status::Invalid("S3 filesystem option '", key, "' has the wrong type");
356+
}
357+
358+
} // namespace
359+
360+
Result<S3Options> S3Options::FromUriAndOptions(const ::arrow::util::Uri& uri,
361+
const FileSystemFactoryOptions& options,
362+
std::string* out_path) {
363+
std::optional<std::string> access_key, secret_key, session_token;
364+
std::shared_ptr<S3RetryStrategy> retry_strategy;
365+
std::shared_ptr<const KeyValueMetadata> default_metadata;
366+
for (const auto& [key, value] : options) {
367+
if (key == "access_key") {
368+
ARROW_ASSIGN_OR_RAISE(access_key, GetOption<std::string>(key, value));
369+
} else if (key == "secret_key") {
370+
ARROW_ASSIGN_OR_RAISE(secret_key, GetOption<std::string>(key, value));
371+
} else if (key == "session_token") {
372+
ARROW_ASSIGN_OR_RAISE(session_token, GetOption<std::string>(key, value));
373+
} else if (key == "retry_strategy") {
374+
ARROW_ASSIGN_OR_RAISE(retry_strategy,
375+
GetOption<std::shared_ptr<S3RetryStrategy>>(key, value));
376+
} else if (key == "default_metadata") {
377+
ARROW_ASSIGN_OR_RAISE(default_metadata,
378+
GetConstSharedPtrOption<KeyValueMetadata>(key, value));
379+
} else {
380+
return Status::Invalid("Unexpected option for S3 filesystem: '", key, "'");
381+
}
382+
}
383+
384+
if (access_key.has_value() != secret_key.has_value()) {
385+
return Status::Invalid(
386+
"Both 'access_key' and 'secret_key' must be provided together");
387+
}
388+
if (session_token.has_value() && !access_key.has_value()) {
389+
return Status::Invalid("'session_token' requires 'access_key' and 'secret_key'");
390+
}
391+
S3Options s3_options;
331392

332393
const auto bucket = uri.host();
333394
auto path = uri.path();
@@ -355,68 +416,81 @@ Result<S3Options> S3Options::FromUri(const Uri& uri, std::string* out_path) {
355416
options_map.emplace(kv.first, kv.second);
356417
}
357418

358-
const auto username = uri.username();
359-
if (!username.empty()) {
360-
options.ConfigureAccessKey(username, uri.password());
419+
if (access_key.has_value()) {
420+
s3_options.ConfigureAccessKey(*access_key, *secret_key, session_token.value_or(""));
361421
} else {
362-
options.ConfigureDefaultCredentials();
422+
const auto username = uri.username();
423+
if (!username.empty()) {
424+
s3_options.ConfigureAccessKey(username, uri.password());
425+
} else {
426+
s3_options.ConfigureDefaultCredentials();
427+
}
363428
}
429+
364430
// Prefer AWS service-specific endpoint url
365431
auto s3_endpoint_env = arrow::internal::GetEnvVar(kAwsEndpointUrlS3EnvVar);
366432
if (s3_endpoint_env.ok()) {
367-
options.endpoint_override = *s3_endpoint_env;
433+
s3_options.endpoint_override = *s3_endpoint_env;
368434
} else {
369435
auto endpoint_env = arrow::internal::GetEnvVar(kAwsEndpointUrlEnvVar);
370436
if (endpoint_env.ok()) {
371-
options.endpoint_override = *endpoint_env;
437+
s3_options.endpoint_override = *endpoint_env;
372438
}
373439
}
374440

375441
bool region_set = false;
376442
for (const auto& kv : options_map) {
377443
if (kv.first == "region") {
378-
options.region = kv.second;
444+
s3_options.region = kv.second;
379445
region_set = true;
380446
} else if (kv.first == "scheme") {
381-
options.scheme = kv.second;
447+
s3_options.scheme = kv.second;
382448
} else if (kv.first == "endpoint_override") {
383-
options.endpoint_override = kv.second;
449+
s3_options.endpoint_override = kv.second;
384450
} else if (kv.first == "allow_delayed_open") {
385-
ARROW_ASSIGN_OR_RAISE(options.allow_delayed_open,
451+
ARROW_ASSIGN_OR_RAISE(s3_options.allow_delayed_open,
386452
::arrow::internal::ParseBoolean(kv.second));
387453
} else if (kv.first == "allow_bucket_creation") {
388-
ARROW_ASSIGN_OR_RAISE(options.allow_bucket_creation,
454+
ARROW_ASSIGN_OR_RAISE(s3_options.allow_bucket_creation,
389455
::arrow::internal::ParseBoolean(kv.second));
390456
} else if (kv.first == "allow_bucket_deletion") {
391-
ARROW_ASSIGN_OR_RAISE(options.allow_bucket_deletion,
457+
ARROW_ASSIGN_OR_RAISE(s3_options.allow_bucket_deletion,
392458
::arrow::internal::ParseBoolean(kv.second));
393459
} else if (kv.first == "tls_ca_file_path") {
394-
options.tls_ca_file_path = kv.second;
460+
s3_options.tls_ca_file_path = kv.second;
395461
} else if (kv.first == "tls_ca_dir_path") {
396-
options.tls_ca_dir_path = kv.second;
462+
s3_options.tls_ca_dir_path = kv.second;
397463
} else if (kv.first == "tls_verify_certificates") {
398-
ARROW_ASSIGN_OR_RAISE(options.tls_verify_certificates,
464+
ARROW_ASSIGN_OR_RAISE(s3_options.tls_verify_certificates,
399465
::arrow::internal::ParseBoolean(kv.second));
400466
} else if (kv.first == "smart_defaults") {
401-
options.smart_defaults = kv.second;
467+
s3_options.smart_defaults = kv.second;
402468
} else {
403469
return Status::Invalid("Unexpected query parameter in S3 URI: '", kv.first, "'");
404470
}
405471
}
406472

407-
if (!region_set && !bucket.empty() && options.endpoint_override.empty()) {
473+
if (retry_strategy) {
474+
s3_options.retry_strategy = std::move(retry_strategy);
475+
}
476+
if (default_metadata) {
477+
s3_options.default_metadata = std::move(default_metadata);
478+
}
479+
480+
if (!region_set && !bucket.empty() && s3_options.endpoint_override.empty()) {
408481
// XXX Should we use a dedicated resolver with the given credentials?
409-
ARROW_ASSIGN_OR_RAISE(options.region, ResolveS3BucketRegion(bucket));
482+
ARROW_ASSIGN_OR_RAISE(s3_options.region, ResolveS3BucketRegion(bucket));
410483
}
411484

412-
return options;
485+
return s3_options;
413486
}
414487

415-
Result<S3Options> S3Options::FromUri(const std::string& uri_string,
416-
std::string* out_path) {
488+
Result<S3Options> S3Options::FromUriAndOptions(const std::string& uri_string,
489+
const FileSystemFactoryOptions& options,
490+
std::string* out_path) {
417491
Uri uri;
418492
RETURN_NOT_OK(uri.Parse(uri_string));
419-
return FromUri(uri, out_path);
493+
return FromUriAndOptions(uri, options, out_path);
420494
}
421495

422496
bool S3Options::Equals(const S3Options& other) const {
@@ -3606,13 +3680,9 @@ auto kS3FileSystemModule = ARROW_REGISTER_FILESYSTEM(
36063680
[](const arrow::util::Uri& uri, const FileSystemFactoryOptions& options,
36073681
const io::IOContext& io_context,
36083682
std::string* out_path) -> Result<std::shared_ptr<fs::FileSystem>> {
3609-
if (!options.empty()) {
3610-
return Status::NotImplemented(
3611-
"S3 filesystem factory options are not supported yet, got: ", options.size(),
3612-
" option(s)");
3613-
}
36143683
RETURN_NOT_OK(EnsureS3Initialized());
3615-
ARROW_ASSIGN_OR_RAISE(auto s3_options, S3Options::FromUri(uri, out_path));
3684+
ARROW_ASSIGN_OR_RAISE(auto s3_options,
3685+
S3Options::FromUriAndOptions(uri, options, out_path));
36163686
return S3FileSystem::Make(s3_options, io_context);
36173687
},
36183688
[] { DCHECK_OK(EnsureS3Finalized()); });

‎cpp/src/arrow/filesystem/s3fs.h‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@
1717

1818
#pragma once
1919

20+
#include <any>
2021
#include <memory>
2122
#include <string>
2223
#include <vector>
@@ -290,6 +291,23 @@ struct ARROW_EXPORT S3Options {
290291
std::string* out_path = NULLPTR);
291292
static Result<S3Options> FromUri(const std::string& uri,
292293
std::string* out_path = NULLPTR);
294+
295+
/// Equivalent to FromUri() with specific backend options that can't be represented
296+
/// on the URI or are better kept out of it (such as credentials).
297+
/// Each option is a (name, value) pair. Recognized keys:
298+
/// - "access_key" (std::string)
299+
/// - "secret_key" (std::string)
300+
/// - "session_token" (std::string)
301+
/// - "retry_strategy" (std::shared_ptr<S3RetryStrategy>)
302+
/// - "default_metadata" (std::shared_ptr<const KeyValueMetadata>)
303+
/// Options take precedence over the URI; unknown keys or invalid values return
304+
/// Status::Invalid.
305+
static Result<S3Options> FromUriAndOptions(const ::arrow::util::Uri& uri,
306+
const FileSystemFactoryOptions& options,
307+
std::string* out_path = NULLPTR);
308+
static Result<S3Options> FromUriAndOptions(const std::string& uri,
309+
const FileSystemFactoryOptions& options,
310+
std::string* out_path = NULLPTR);
293311
};
294312

295313
/// S3-backed FileSystem implementation.

‎cpp/src/arrow/filesystem/s3fs_module_test.cc‎

Lines changed: 45 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -83,10 +83,53 @@ TEST(S3Test, FromUri) {
8383
"&allow_bucket_creation=0&allow_bucket_deletion=0");
8484
}
8585

86-
TEST(S3Test, FromUriRejectsOptions) {
86+
TEST(S3Test, FromUriAndOptionsCredentials) {
87+
ASSERT_OK_AND_ASSIGN(auto minio, GetMinioEnv()->GetOneServer());
88+
std::string path;
89+
FileSystemFactoryOptions options{
90+
{"access_key", std::string(minio->access_key())},
91+
{"secret_key", std::string(minio->secret_key())},
92+
};
93+
// Credentials supplied via options, NOT in the URI.
94+
ASSERT_OK_AND_ASSIGN(
95+
auto fs,
96+
FileSystemFromUriAndOptions("s3://bucket/somedir/subdir/subfile", options, &path));
97+
// They crossed the module boundary and were applied -> reflected in MakeUri.
98+
EXPECT_EQ(fs->MakeUri("/" + path),
99+
"s3://minio:miniopass@bucket/somedir/subdir/subfile"
100+
"?region=us-east-1&scheme=https&endpoint_override="
101+
"&allow_bucket_creation=0&allow_bucket_deletion=0");
102+
}
103+
104+
namespace {
105+
class NoopRetryStrategy : public S3RetryStrategy {
106+
public:
107+
bool ShouldRetry(const AWSErrorDetail&, int64_t) override { return false; }
108+
int64_t CalculateDelayBeforeNextRetry(const AWSErrorDetail&, int64_t) override {
109+
return 0;
110+
}
111+
};
112+
} // namespace
113+
114+
TEST(S3Test, FromUriAndOptionsRetryStrategy) {
115+
ASSERT_OK_AND_ASSIGN(auto minio, GetMinioEnv()->GetOneServer());
116+
FileSystemFactoryOptions options{
117+
{"access_key", std::string(minio->access_key())},
118+
{"secret_key", std::string(minio->secret_key())},
119+
{"retry_strategy",
120+
std::shared_ptr<S3RetryStrategy>(std::make_shared<NoopRetryStrategy>())},
121+
};
122+
std::string path;
123+
ASSERT_OK_AND_ASSIGN(
124+
auto fs,
125+
FileSystemFromUriAndOptions("s3://bucket/somedir/subdir/subfile", options, &path));
126+
ASSERT_NE(fs, nullptr);
127+
}
128+
129+
TEST(S3Test, FromUriRejectsUnknownOptions) {
87130
FileSystemFactoryOptions options{{"some_option", 1}};
88131
EXPECT_RAISES_WITH_MESSAGE_THAT(
89-
NotImplemented, ::testing::HasSubstr("options are not supported"),
132+
Invalid, ::testing::HasSubstr("Unexpected option"),
90133
FileSystemFromUriAndOptions("s3://bucket/key", options));
91134
}
92135

‎cpp/src/arrow/filesystem/s3fs_test.cc‎

Lines changed: 86 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -398,6 +398,92 @@ TEST_F(S3OptionsTest, FromAccessKey) {
398398
ASSERT_EQ(options.GetSessionToken(), "token");
399399
}
400400

401+
TEST_F(S3OptionsTest, FromUriAndOptionsCredentials) {
402+
std::string path;
403+
S3Options options;
404+
FileSystemFactoryOptions kv{
405+
{"access_key", std::string("ak")},
406+
{"secret_key", std::string("sk")},
407+
};
408+
ASSERT_OK_AND_ASSIGN(options, S3Options::FromUriAndOptions("s3://", kv, &path));
409+
ASSERT_EQ(options.GetAccessKey(), "ak");
410+
ASSERT_EQ(options.GetSecretKey(), "sk");
411+
ASSERT_EQ(options.GetSessionToken(), "");
412+
413+
kv.push_back({"session_token", std::string("tok")});
414+
ASSERT_OK_AND_ASSIGN(options, S3Options::FromUriAndOptions("s3://", kv, &path));
415+
ASSERT_EQ(options.GetAccessKey(), "ak");
416+
ASSERT_EQ(options.GetSecretKey(), "sk");
417+
ASSERT_EQ(options.GetSessionToken(), "tok");
418+
419+
// Failure scenarios
420+
// Pairing is correct
421+
ASSERT_THAT(
422+
S3Options::FromUriAndOptions("s3://", {{"access_key", std::string("ak")}}, &path),
423+
Raises(StatusCode::Invalid, ::testing::HasSubstr("must be provided together")));
424+
ASSERT_THAT(
425+
S3Options::FromUriAndOptions("s3://", {{"session_token", std::string("tok")}},
426+
&path),
427+
Raises(StatusCode::Invalid, ::testing::HasSubstr("session_token' requires")));
428+
// unknown option key
429+
ASSERT_THAT(S3Options::FromUriAndOptions("s3://", {{"bogus", std::string("x")}}, &path),
430+
Raises(StatusCode::Invalid, ::testing::HasSubstr("Unexpected option")));
431+
// const char* is rejected. Options must be an explicit std::string
432+
ASSERT_THAT(S3Options::FromUriAndOptions("s3://", {{"access_key", "42"}}, &path),
433+
Raises(StatusCode::Invalid, ::testing::HasSubstr("wrong type")));
434+
}
435+
436+
TEST_F(S3OptionsTest, FromUriAndOptionsCredentialPrecedence) {
437+
std::string path;
438+
S3Options options;
439+
// options override URI userinfo when not empty
440+
FileSystemFactoryOptions kv{
441+
{"access_key", std::string("opt_access_key")},
442+
{"secret_key", std::string("opt_secret_key")},
443+
};
444+
ASSERT_OK_AND_ASSIGN(
445+
options,
446+
S3Options::FromUriAndOptions(
447+
"s3://uri_access_key:uri_secret_key@mybucket?region=us-east-1", kv, &path));
448+
ASSERT_EQ(options.GetAccessKey(), "opt_access_key");
449+
ASSERT_EQ(options.GetSecretKey(), "opt_secret_key");
450+
451+
ASSERT_OK_AND_ASSIGN(
452+
options,
453+
S3Options::FromUriAndOptions(
454+
"s3://uri_access_key:uri_secret_key@mybucket?region=us-east-1", {}, &path));
455+
ASSERT_EQ(options.GetAccessKey(), "uri_access_key");
456+
ASSERT_EQ(options.GetSecretKey(), "uri_secret_key");
457+
}
458+
459+
TEST_F(S3OptionsTest, FromUriAndOptionsRetryStrategy) {
460+
std::string path;
461+
S3Options options;
462+
auto retry = S3RetryStrategy::GetAwsDefaultRetryStrategy(/*max_attempts=*/3);
463+
FileSystemFactoryOptions kv{{"retry_strategy", retry}};
464+
ASSERT_OK_AND_ASSIGN(options, S3Options::FromUriAndOptions("s3://", kv, &path));
465+
// The exact typed object crossed through std::any unchanged.
466+
ASSERT_EQ(options.retry_strategy, retry);
467+
468+
// wrong type is rejected, same as the credential keys
469+
ASSERT_THAT(S3Options::FromUriAndOptions("s3://",
470+
{{"retry_strategy", std::string("x")}}, &path),
471+
Raises(StatusCode::Invalid, ::testing::HasSubstr("wrong type")));
472+
}
473+
474+
TEST_F(S3OptionsTest, FromUriAndOptionsDefaultMetadata) {
475+
std::string path;
476+
S3Options options;
477+
auto metadata = KeyValueMetadata::Make({"Content-Type"}, {"x-arrow/test"});
478+
FileSystemFactoryOptions kv{{"default_metadata", metadata}};
479+
ASSERT_OK_AND_ASSIGN(options, S3Options::FromUriAndOptions("s3://", kv, &path));
480+
ASSERT_EQ(options.default_metadata, metadata);
481+
482+
ASSERT_THAT(S3Options::FromUriAndOptions(
483+
"s3://", {{"default_metadata", std::string("x")}}, &path),
484+
Raises(StatusCode::Invalid, ::testing::HasSubstr("wrong type")));
485+
}
486+
401487
TEST_F(S3OptionsTest, FromAssumeRole) {
402488
S3Options options;
403489

0 commit comments

Comments
 (0)