GH-46369: [C++] Add key-value pairs to the FileSystemFactory interface and to S3Options

ci/scripts/cpp_build.sh

@@ -70,6 +70,7 @@ if [ "${ARROW_ENABLE_THREADING:-ON}" = "OFF" ]; then
   ARROW_JEMALLOC=OFF
   ARROW_MIMALLOC=OFF
   ARROW_S3=OFF
+  ARROW_S3_MODULE=OFF
   ARROW_WITH_OPENTELEMETRY=OFF
 fi
 
@@ -229,6 +230,7 @@ else
     -DARROW_PARQUET=${ARROW_PARQUET:-OFF} \
     -DARROW_RUNTIME_SIMD_LEVEL=${ARROW_RUNTIME_SIMD_LEVEL:-MAX} \
     -DARROW_S3=${ARROW_S3:-OFF} \
+    -DARROW_S3_MODULE=${ARROW_S3_MODULE:-OFF} \
     -DARROW_SIMD_LEVEL=${ARROW_SIMD_LEVEL:-DEFAULT} \
     -DARROW_SUBSTRAIT=${ARROW_SUBSTRAIT:-OFF} \
     -DARROW_TEST_LINKAGE=${ARROW_TEST_LINKAGE:-shared} \

cpp/src/arrow/filesystem/filesystem.cc

@@ -893,20 +893,24 @@ Status LoadFileSystemFactories(const char* libpath) {
 
 namespace {
 
-Result<std::shared_ptr<FileSystem>> FileSystemFromUriReal(const Uri& uri,
-                                                          const std::string& uri_string,
-                                                          const io::IOContext& io_context,
-                                                          std::string* out_path) {
+Result<std::shared_ptr<FileSystem>> FileSystemFromUriReal(
+    const Uri& uri, const std::string& uri_string,
+    const std::vector<std::pair<std::string, std::any>>& options,
+    const io::IOContext& io_context, std::string* out_path) {
   const auto scheme = uri.scheme();
 
   {
     ARROW_ASSIGN_OR_RAISE(
         auto* factory,
         FileSystemFactoryRegistry::GetInstance()->FactoryForScheme(scheme));
     if (factory != nullptr) {
-      return factory->function(uri, io_context, out_path);
+      return factory->function(uri, options, io_context, out_path);
     }
   }
+  if (!options.empty()) {
+    return Status::NotImplemented("Filesystem options are not supported yet for scheme '",
+                                  scheme, "', got ", options.size(), " option(s)");
+  }
 
   if (scheme == "abfs" || scheme == "abfss") {
 #ifdef ARROW_AZURE
@@ -962,14 +966,28 @@ Result<std::shared_ptr<FileSystem>> FileSystemFromUriReal(const Uri& uri,
 
 Result<std::shared_ptr<FileSystem>> FileSystemFromUri(const std::string& uri_string,
                                                       std::string* out_path) {
-  return FileSystemFromUri(uri_string, io::default_io_context(), out_path);
+  return FileSystemFromUri(uri_string, /*options=*/{}, io::default_io_context(),
+                           out_path);
+}
+
+Result<std::shared_ptr<FileSystem>> FileSystemFromUri(
+    const std::string& uri_string,
+    const std::vector<std::pair<std::string, std::any>>& options, std::string* out_path) {
+  return FileSystemFromUri(uri_string, options, io::default_io_context(), out_path);
 }
 
 Result<std::shared_ptr<FileSystem>> FileSystemFromUri(const std::string& uri_string,
                                                       const io::IOContext& io_context,
                                                       std::string* out_path) {
+  return FileSystemFromUri(uri_string, /*options=*/{}, io_context, out_path);
+}
+
+Result<std::shared_ptr<FileSystem>> FileSystemFromUri(
+    const std::string& uri_string,
+    const std::vector<std::pair<std::string, std::any>>& options,
+    const io::IOContext& io_context, std::string* out_path) {
   ARROW_ASSIGN_OR_RAISE(auto fsuri, ParseFileSystemUri(uri_string));
-  return FileSystemFromUriReal(fsuri, uri_string, io_context, out_path);
+  return FileSystemFromUriReal(fsuri, uri_string, options, io_context, out_path);
 }
 
 Result<std::shared_ptr<FileSystem>> FileSystemFromUriOrPath(const std::string& uri_string,

cpp/src/arrow/filesystem/filesystem.h

@@ -17,6 +17,7 @@
 
 #pragma once
 
+#include <any>
 #include <chrono>
 #include <cstdint>
 #include <functional>
@@ -28,6 +29,7 @@
 
 #include "arrow/filesystem/type_fwd.h"
 #include "arrow/io/interfaces.h"
+#include "arrow/result.h"
 #include "arrow/type_fwd.h"
 #include "arrow/util/compare.h"
 #include "arrow/util/macros.h"
@@ -359,11 +361,34 @@ class ARROW_EXPORT FileSystem
 
 struct FileSystemFactory {
   std::function<Result<std::shared_ptr<FileSystem>>(
-      const Uri& uri, const io::IOContext& io_context, std::string* out_path)>
+      const Uri& uri, const std::vector<std::pair<std::string, std::any>>& options,
+      const io::IOContext& io_context, std::string* out_path)>
       function;
   std::string_view file;
   int line;
 
+  /// Construct from an options-aware factory function.
+  FileSystemFactory(std::function<Result<std::shared_ptr<FileSystem>>(
+                        const Uri&, const std::vector<std::pair<std::string, std::any>>&,
+                        const io::IOContext&, std::string*)>
+                        fn,
+                    std::string_view file, int line)
+      : function(std::move(fn)), file(file), line(line) {}
+
+  /// Construct from a non-options aware factory function maintaining source compatibility
+  /// with existing factories.
+  FileSystemFactory(std::function<Result<std::shared_ptr<FileSystem>>(
+                        const Uri&, const io::IOContext&, std::string*)>
+                        fn,
+                    std::string_view file, int line)
+      : function([fn = std::move(fn)](
+                     const Uri& uri,
+                     const std::vector<std::pair<std::string, std::any>>& /*ignored*/,
+                     const io::IOContext& ctx,
+                     std::string* out_path) { return fn(uri, ctx, out_path); }),
+        file(file),
+        line(line) {}
+
   bool operator==(const FileSystemFactory& other) const {
     // In the case where libarrow is linked statically both to the executable and to a
     // dynamically loaded filesystem implementation library, the library contains a
@@ -547,6 +572,30 @@ ARROW_EXPORT
 Result<std::shared_ptr<FileSystem>> FileSystemFromUri(const std::string& uri,
                                                       std::string* out_path = NULLPTR);
 
+/// \brief Create a new FileSystem by URI with extended backend-specific filesystem
+/// options
+///
+/// Recognized schemes are "file", "mock", "hdfs", "viewfs", "s3",
+/// "gs" and "gcs".
+///
+/// Support for other schemes can be added using RegisterFileSystemFactory.
+///
+/// \param[in] uri the URI to give access to
+/// \param[in] options a list of backend-specific filesystem options
+///            Each option is a (name, value) pair.
+///            The expected type is specific to the backend and
+///            option name.
+///            Options are only forwarded to schemes dispatched through a
+///            registered FileSystemFactory (currently "s3" and any scheme
+///            added via RegisterFileSystemFactory). They are ignored by the
+///            built-in schemes.
+/// \param[out] out_path (optional) Path inside the filesystem.
+/// \return out_fs FileSystem instance.
+ARROW_EXPORT
+Result<std::shared_ptr<FileSystem>> FileSystemFromUri(
+    const std::string& uri, const std::vector<std::pair<std::string, std::any>>& options,
+    std::string* out_path = NULLPTR);
+
 /// \brief Create a new FileSystem by URI with a custom IO context
 ///
 /// Recognized schemes are "file", "mock", "hdfs", "viewfs", "s3",
@@ -563,6 +612,31 @@ Result<std::shared_ptr<FileSystem>> FileSystemFromUri(const std::string& uri,
                                                       const io::IOContext& io_context,
                                                       std::string* out_path = NULLPTR);
 
+/// \brief Create a new FileSystem by URI with a custom IO context with backend-specific
+/// filesystem options
+///
+/// Recognized schemes are "file", "mock", "hdfs", "viewfs", "s3",
+/// "gs" and "gcs".
+///
+/// Support for other schemes can be added using RegisterFileSystemFactory.
+///
+/// \param[in] uri a URI-based path, ex: file:///some/local/path
+/// \param[in] options a list of backend-specific filesystem options
+///            Each option is a (name, value) pair.
+///            The expected type is specific to the backend and
+///            option name.
+///            Options are only forwarded to schemes dispatched through a
+///            registered FileSystemFactory (currently "s3" and any scheme
+///            added via RegisterFileSystemFactory). They are ignored by the
+///            built-in schemes.
+/// \param[in] io_context an IOContext which will be associated with the filesystem
+/// \param[out] out_path (optional) Path inside the filesystem.
+/// \return out_fs FileSystem instance.
+ARROW_EXPORT
+Result<std::shared_ptr<FileSystem>> FileSystemFromUri(
+    const std::string& uri, const std::vector<std::pair<std::string, std::any>>& options,
+    const io::IOContext& io_context, std::string* out_path = NULLPTR);
+
 /// \brief Create a new FileSystem by URI
 ///
 /// Support for other schemes can be added using RegisterFileSystemFactory.

cpp/src/arrow/filesystem/filesystem_test.cc

@@ -15,6 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
+#include <any>
 #include <memory>
 #include <string>
 #include <utility>
@@ -640,6 +641,11 @@ TEST_F(TestMockFS, FileSystemFromUri) {
       Invalid, ::testing::HasSubstr("syntax error at character ' ' (position 12)"),
       FileSystemFromUri("mock:/folder name/bar", &path));
   CheckDirs({});
+  std::vector<std::pair<std::string, std::any>> options{{"some_option", 1}};
+  EXPECT_RAISES_WITH_MESSAGE_THAT(NotImplemented,
+                                  ::testing::HasSubstr("options are not supported"),
+                                  FileSystemFromUri("mock:///foo/bar", options, &path));
+  CheckDirs({});
 }
 
 ////////////////////////////////////////////////////////////////////////////

cpp/src/arrow/filesystem/localfs_test.cc

@@ -15,6 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
+#include <any>
 #include <memory>
 #include <sstream>
 #include <string>
@@ -144,10 +145,18 @@ TEST(FileSystemFromUri, LoadedRegisteredFactory) {
   EXPECT_THAT(FileSystemFromUri("example:///hey/yo", &path), Raises(StatusCode::Invalid));
 
   EXPECT_THAT(LoadFileSystemFactories(ARROW_FILESYSTEM_EXAMPLE_LIBPATH), Ok());
-
   ASSERT_OK_AND_ASSIGN(auto fs, FileSystemFromUri("example:///hey/yo", &path));
   EXPECT_EQ(path, "/hey/yo");
   EXPECT_EQ(fs->type_name(), "local");
+
+  // Validate extra options are forwarded to the factory.
+  std::vector<std::pair<std::string, std::any>> options{
+      {"example_option_string", std::string("example_value")},
+      {"example_option_int", 42},
+  };
+  ASSERT_OK_AND_ASSIGN(fs, FileSystemFromUri("example:///hey/yo", options, &path));
+  EXPECT_EQ(path, "/hey/yo/example_value/42");
+  EXPECT_EQ(fs->type_name(), "local");
 }
 
 TEST(FileSystemFromUri, RuntimeRegisteredFactory) {
@@ -171,10 +180,21 @@ TEST(FileSystemFromUri, RuntimeRegisteredFactory) {
 }
 
 FileSystemRegistrar kSegfaultFileSystemModule[]{
-    ARROW_REGISTER_FILESYSTEM("segfault", nullptr, {}),
-    ARROW_REGISTER_FILESYSTEM("segfault", nullptr, {}),
-    ARROW_REGISTER_FILESYSTEM("segfault", nullptr, {}),
-};
+    ARROW_REGISTER_FILESYSTEM(
+        "segfault",
+        std::function<Result<std::shared_ptr<FileSystem>>(
+            const Uri&, const io::IOContext&, std::string*)>(nullptr),
+        {}),
+    ARROW_REGISTER_FILESYSTEM(
+        "segfault",
+        std::function<Result<std::shared_ptr<FileSystem>>(
+            const Uri&, const io::IOContext&, std::string*)>(nullptr),
+        {}),
+    ARROW_REGISTER_FILESYSTEM(
+        "segfault",
+        std::function<Result<std::shared_ptr<FileSystem>>(
+            const Uri&, const io::IOContext&, std::string*)>(nullptr),
+        {})};
 TEST(FileSystemFromUri, LinkedRegisteredFactoryNameCollision) {
   // Since multiple registrars are defined in this translation unit which all
   // register factories for the 'segfault' scheme, using that scheme in FileSystemFromUri

cpp/src/arrow/filesystem/s3fs.cc

@@ -3603,11 +3603,18 @@ Result<std::string> ResolveS3BucketRegion(const std::string& bucket) {
 
 auto kS3FileSystemModule = ARROW_REGISTER_FILESYSTEM(
     "s3",
-    [](const arrow::util::Uri& uri, const io::IOContext& io_context,
+    [](const arrow::util::Uri& uri,
+       const std::vector<std::pair<std::string, std::any>>& options,
+       const io::IOContext& io_context,
        std::string* out_path) -> Result<std::shared_ptr<fs::FileSystem>> {
+      if (!options.empty()) {
+        return Status::NotImplemented(
+            "S3 filesystem factory options are not supported yet, got: ", options.size(),
+            " options");
+      }
       RETURN_NOT_OK(EnsureS3Initialized());
-      ARROW_ASSIGN_OR_RAISE(auto options, S3Options::FromUri(uri, out_path));
-      return S3FileSystem::Make(options, io_context);
+      ARROW_ASSIGN_OR_RAISE(auto s3_options, S3Options::FromUri(uri, out_path));
+      return S3FileSystem::Make(s3_options, io_context);
     },
     [] { DCHECK_OK(EnsureS3Finalized()); });
 

cpp/src/arrow/filesystem/s3fs_module_test.cc

@@ -16,6 +16,7 @@
 // under the License.
 
 #include <algorithm>
+#include <any>
 #include <exception>
 #include <memory>
 #include <sstream>
@@ -82,4 +83,11 @@ TEST(S3Test, FromUri) {
             "&allow_bucket_creation=0&allow_bucket_deletion=0");
 }
 
+TEST(S3Test, FromUriRejectsOptions) {
+  std::vector<std::pair<std::string, std::any>> options{{"some_option", 1}};
+  EXPECT_RAISES_WITH_MESSAGE_THAT(NotImplemented,
+                                  ::testing::HasSubstr("options are not supported"),
+                                  FileSystemFromUri("s3://bucket/key", options));
+}
+
 }  // namespace arrow::fs

cpp/src/arrow/testing/examplefs.cc

@@ -15,6 +15,9 @@
 // specific language governing permissions and limitations
 // under the License.
 
+#include <any>
+#include <typeinfo>
+
 #include "arrow/filesystem/filesystem.h"
 #include "arrow/filesystem/filesystem_library.h"
 #include "arrow/result.h"
@@ -26,12 +29,31 @@ namespace arrow::fs {
 
 auto kExampleFileSystemModule = ARROW_REGISTER_FILESYSTEM(
     "example",
-    [](const Uri& uri, const io::IOContext& io_context,
+    [](const Uri& uri, const std::vector<std::pair<std::string, std::any>>& options,
+       const io::IOContext& io_context,
        std::string* out_path) -> Result<std::shared_ptr<FileSystem>> {
       constexpr std::string_view kScheme = "example";
       EXPECT_EQ(uri.scheme(), kScheme);
       auto local_uri = "file" + uri.ToString().substr(kScheme.size());
-      return FileSystemFromUri(local_uri, io_context, out_path);
+      ARROW_ASSIGN_OR_RAISE(auto fs,
+                            FileSystemFromUri(local_uri, options, io_context, out_path));
+      for (const auto& [key, value] : options) {
+        EXPECT_TRUE(value.has_value());
+        if (key == "example_option_string") {
+          EXPECT_TRUE(value.type() == typeid(std::string));
+          if (out_path != nullptr) {
+            *out_path += "/" + std::any_cast<std::string>(value);
+          }
+        } else if (key == "example_option_int") {
+          EXPECT_TRUE(value.type() == typeid(int));
+          if (out_path != nullptr) {
+            *out_path += "/" + std::to_string(std::any_cast<int>(value));
+          }
+        } else {
+          ADD_FAILURE() << "Unexpected option: " << key;
+        }
+      }
+      return fs;
     },
     {});