GH-50126: [C++] Make S3Filesystem consume key-value options

Author: raulcd

cpp/src/arrow/filesystem/s3fs.cc

@@ -419,6 +419,52 @@ Result<S3Options> S3Options::FromUri(const std::string& uri_string,
   return FromUri(uri, out_path);
 }
 
+namespace {
+
+Result<std::string> GetStringOption(const std::string& key, const std::any& value) {
+  // TODO: Validate this is necessary with tests.
+  if (const auto* s = std::any_cast<std::string>(&value)) {
+    return *s;
+  }
+  return Status::Invalid("S3 filesystem option '", key, "' must be a std::string");
+}
+}  // namespace
+
+Result<S3Options> S3Options::FromUriAndOptions(
+    const ::arrow::util::Uri& uri,
+    const FileSystemFactoryOptions& options, std::string* out_path) {
+  std::optional<std::string> access_key, secret_key, session_token;
+  for (const auto& [key, value] : options) {
+    if (key == "access_key") {
+      ARROW_ASSIGN_OR_RAISE(access_key, GetStringOption(key, value));
+    } else if (key == "secret_key") {
+      ARROW_ASSIGN_OR_RAISE(secret_key, GetStringOption(key, value));
+    } else if (key == "session_token") {
+      ARROW_ASSIGN_OR_RAISE(session_token, GetStringOption(key, value));
+    } else {
+      return Status::Invalid("Unexpected option for S3 filesystem: '", key, "'");
+    }
+  }
+
+  if (access_key.has_value() != secret_key.has_value()) {
+    return Status::Invalid(
+        "Both 'access_key' and 'secret_key' must be provided together");
+  }
+  ARROW_ASSIGN_OR_RAISE(auto s3_options, FromUri(uri, out_path));
+  if (access_key.has_value()) {
+    s3_options.ConfigureAccessKey(*access_key, *secret_key, session_token.value_or(""));
+  }
+  return s3_options;
+}
+
+Result<S3Options> S3Options::FromUriAndOptions(
+    const std::string& uri_string,
+    const FileSystemFactoryOptions& options, std::string* out_path) {
+  Uri uri;
+  RETURN_NOT_OK(uri.Parse(uri_string));
+  return FromUriAndOptions(uri, options, out_path);
+}
+
 bool S3Options::Equals(const S3Options& other) const {
   const int64_t default_metadata_size = default_metadata ? default_metadata->size() : 0;
   const bool default_metadata_equals =
@@ -3606,13 +3652,14 @@ auto kS3FileSystemModule = ARROW_REGISTER_FILESYSTEM(
     [](const arrow::util::Uri& uri, const FileSystemFactoryOptions& options,
        const io::IOContext& io_context,
        std::string* out_path) -> Result<std::shared_ptr<fs::FileSystem>> {
-      if (!options.empty()) {
+      /*if (!options.empty()) {
         return Status::NotImplemented(
             "S3 filesystem factory options are not supported yet, got: ", options.size(),
             " option(s)");
-      }
+      }*/
       RETURN_NOT_OK(EnsureS3Initialized());
-      ARROW_ASSIGN_OR_RAISE(auto s3_options, S3Options::FromUri(uri, out_path));
+      ARROW_ASSIGN_OR_RAISE(auto s3_options,
+                            S3Options::FromUriAndOptions(uri, options, out_path));
       return S3FileSystem::Make(s3_options, io_context);
     },
     [] { DCHECK_OK(EnsureS3Finalized()); });

cpp/src/arrow/filesystem/s3fs.h

@@ -17,6 +17,7 @@
 
 #pragma once
 
+#include <any>
 #include <memory>
 #include <string>
 #include <vector>
@@ -290,6 +291,22 @@ struct ARROW_EXPORT S3Options {
                                    std::string* out_path = NULLPTR);
   static Result<S3Options> FromUri(const std::string& uri,
                                    std::string* out_path = NULLPTR);
+
+  /// Equivalent to FromUri() with specific backend options that can't be represented
+  /// on the URI overlaid on top. Recognized keys:
+  /// - "access_key" (std::string)
+  /// - "secret_key" (std::string)
+  /// - "session_token" (std::string).
+  /// Options take precedence over the URI; unknown keys return
+  /// Status::Invalid.
+  static Result<S3Options> FromUriAndOptions(
+      const ::arrow::util::Uri& uri,
+      const FileSystemFactoryOptions& options,
+      std::string* out_path = NULLPTR);
+  static Result<S3Options> FromUriAndOptions(
+      const std::string& uri,
+      const FileSystemFactoryOptions& options,
+      std::string* out_path = NULLPTR);
 };
 
 /// S3-backed FileSystem implementation.

cpp/src/arrow/filesystem/s3fs_module_test.cc

@@ -83,11 +83,10 @@ TEST(S3Test, FromUri) {
             "&allow_bucket_creation=0&allow_bucket_deletion=0");
 }
 
-TEST(S3Test, FromUriRejectsOptions) {
+TEST(S3Test, FromUriRejectsUnknownOptions) {
   FileSystemFactoryOptions options{{"some_option", 1}};
-  EXPECT_RAISES_WITH_MESSAGE_THAT(
-      NotImplemented, ::testing::HasSubstr("options are not supported"),
-      FileSystemFromUriAndOptions("s3://bucket/key", options));
+  EXPECT_RAISES_WITH_MESSAGE_THAT(Invalid, ::testing::HasSubstr("Unexpected option"),
+                                  FileSystemFromUriAndOptions("s3://bucket/key", options));
 }
 
 }  // namespace arrow::fs