S3 filesystem raul tests

cpp/CMakePresets.json

@@ -101,18 +101,18 @@
       "inherits": "features-basic",
       "hidden": true,
       "cacheVariables": {
-        "ARROW_SUBSTRAIT": "ON",
-        "ARROW_ACERO": "ON",
+        "ARROW_SUBSTRAIT": "OFF",
+        "ARROW_ACERO": "OFF",
         "ARROW_MIMALLOC": "ON",
-        "ARROW_PARQUET": "ON",
-        "ARROW_WITH_BROTLI": "ON",
-        "ARROW_WITH_BZ2": "ON",
-        "ARROW_WITH_LZ4": "ON",
-        "ARROW_WITH_RE2": "ON",
-        "ARROW_WITH_SNAPPY": "ON",
-        "ARROW_WITH_UTF8PROC": "ON",
-        "ARROW_WITH_ZLIB": "ON",
-        "ARROW_WITH_ZSTD": "ON"
+        "ARROW_PARQUET": "OFF",
+        "ARROW_WITH_BROTLI": "OFF",
+        "ARROW_WITH_BZ2": "OFF",
+        "ARROW_WITH_LZ4": "OFF",
+        "ARROW_WITH_RE2": "OFF",
+        "ARROW_WITH_SNAPPY": "OFF",
+        "ARROW_WITH_UTF8PROC": "OFF",
+        "ARROW_WITH_ZLIB": "OFF",
+        "ARROW_WITH_ZSTD": "OFF"
       }
     },
     {
@@ -128,10 +128,11 @@
       "inherits": "features-basic",
       "hidden": true,
       "cacheVariables": {
-        "ARROW_AZURE": "ON",
-        "ARROW_GCS": "ON",
-        "ARROW_HDFS": "ON",
-        "ARROW_S3": "ON"
+        "ARROW_AZURE": "OFF",
+        "ARROW_GCS": "OFF",
+        "ARROW_HDFS": "OFF",
+        "ARROW_S3": "OFF",
+        "ARROW_S3_MODULE": "OFF"
       }
     },
     {
@@ -165,10 +166,10 @@
       ],
       "hidden": true,
       "cacheVariables": {
-        "ARROW_COMPUTE": "ON",
-        "ARROW_CSV": "ON",
+        "ARROW_COMPUTE": "OFF",
+        "ARROW_CSV": "OFF",
         "ARROW_FILESYSTEM": "ON",
-        "ARROW_JSON": "ON"
+        "ARROW_JSON": "OFF"
       }
     },
     {
@@ -189,17 +190,14 @@
     {
       "name": "features-python-maximal",
       "inherits": [
-        "features-cuda",
         "features-filesystems",
-        "features-flight-sql",
-        "features-gandiva",
         "features-main",
         "features-python-minimal"
       ],
       "hidden": true,
       "cacheVariables": {
-        "ARROW_ORC": "ON",
-        "PARQUET_REQUIRE_ENCRYPTION": "ON"
+        "ARROW_ORC": "OFF",
+        "PARQUET_REQUIRE_ENCRYPTION": "OFF"
       }
     },
     {

cpp/cmake_modules/DefineOptions.cmake

@@ -400,6 +400,12 @@ takes precedence over ccache if a storage backend is configured" ON)
                 DEPENDS
                 ARROW_FILESYSTEM)
 
+  define_option(ARROW_S3_MODULE
+                "Build the Arrow S3 filesystem as a dynamic module"
+                OFF
+                DEPENDS
+                ARROW_S3)
+
   define_option(ARROW_SKYHOOK
                 "Build the Skyhook libraries"
                 OFF

cpp/src/arrow/CMakeLists.txt

@@ -876,6 +876,19 @@ if(ARROW_FILESYSTEM)
     foreach(ARROW_FILESYSTEM_TARGET ${ARROW_FILESYSTEM_TARGETS})
       target_link_libraries(${ARROW_FILESYSTEM_TARGET} PRIVATE ${AWSSDK_LINK_LIBRARIES})
     endforeach()
+
+    if(ARROW_S3_MODULE)
+      if(NOT ARROW_BUILD_SHARED)
+        message(FATAL_ERROR "ARROW_S3_MODULE without shared libarrow (-DARROW_BUILD_SHARED=ON) is not supported"
+        )
+      endif()
+
+      add_library(arrow_s3fs MODULE filesystem/s3fs_module.cc filesystem/s3fs.cc)
+      target_link_libraries(arrow_s3fs PRIVATE ${AWSSDK_LINK_LIBRARIES} arrow_shared)
+      set_source_files_properties(filesystem/s3fs.cc filesystem/s3fs_module.cc
+                                  PROPERTIES SKIP_PRECOMPILE_HEADERS ON
+                                             SKIP_UNITY_BUILD_INCLUSION ON)
+    endif()
   endif()
 
   list(APPEND ARROW_TESTING_SHARED_LINK_LIBS ${ARROW_GTEST_GMOCK})

cpp/src/arrow/filesystem/CMakeLists.txt

@@ -121,6 +121,24 @@ if(ARROW_S3)
       target_link_libraries(arrow-filesystem-s3fs-benchmark PRIVATE parquet_shared)
     endif()
   endif()
+
+  if(ARROW_S3_MODULE)
+    add_arrow_test(s3fs_module_test
+                   SOURCES
+                   s3fs_module_test.cc
+                   s3_test_util.cc
+                   EXTRA_LABELS
+                   filesystem
+                   DEFINITIONS
+                   ARROW_S3_LIBPATH="$<TARGET_FILE:arrow_s3fs>"
+                   EXTRA_LINK_LIBS
+                   Boost::filesystem
+                   Boost::system)
+    target_compile_definitions(arrow-filesystem-test
+                               PUBLIC ARROW_S3_LIBPATH="$<TARGET_FILE:arrow_s3fs>")
+    target_sources(arrow-filesystem-test PUBLIC s3fs_module_test.cc s3_test_util.cc)
+    target_link_libraries(arrow-filesystem-test PUBLIC Boost::filesystem Boost::system)
+  endif()
 endif()
 
 if(ARROW_HDFS)

cpp/src/arrow/filesystem/filesystem.cc

@@ -34,9 +34,6 @@
 #ifdef ARROW_HDFS
 #  include "arrow/filesystem/hdfs.h"
 #endif
-#ifdef ARROW_S3
-#  include "arrow/filesystem/s3fs.h"
-#endif
 #include "arrow/filesystem/localfs.h"
 #include "arrow/filesystem/mockfs.h"
 #include "arrow/filesystem/path_util.h"
@@ -722,6 +719,29 @@ class FileSystemFactoryRegistry {
     return &registry;
   }
 
+  Status Unregister(const std::string& scheme) {
+    std::shared_lock lock{mutex_};
+    RETURN_NOT_OK(CheckValid());
+
+    auto it = scheme_to_factory_.find(scheme);
+    if (it == scheme_to_factory_.end()) {
+      return Status::KeyError("No factories found for scheme ", scheme,
+                              ", can't unregister");
+    }
+
+    std::function<void()> finalizer;
+    if (it->second.ok()) {
+      finalizer = it->second.ValueOrDie().finalizer;
+    }
+    scheme_to_factory_.erase(it);
+    lock.unlock();
+
+    if (finalizer) {
+      finalizer();
+    }
+    return Status::OK();
+  }
+
   Result<const FileSystemFactory*> FactoryForScheme(const std::string& scheme) {
     std::shared_lock lock{mutex_};
     RETURN_NOT_OK(CheckValid());
@@ -771,7 +791,7 @@ class FileSystemFactoryRegistry {
     if (finalized_) return;
 
     for (const auto& [_, registered_or_error] : scheme_to_factory_) {
-      if (!registered_or_error.ok()) continue;
+      if (!registered_or_error.ok() || !registered_or_error->finalizer) continue;
       registered_or_error->finalizer();
     }
     finalized_ = true;
@@ -841,6 +861,10 @@ FileSystemRegistrar::FileSystemRegistrar(std::string scheme, FileSystemFactory f
 
 namespace internal {
 void* GetFileSystemRegistry() { return FileSystemFactoryRegistry::GetInstance(); }
+
+Status UnregisterFileSystemFactory(const std::string& scheme) {
+  return FileSystemFactoryRegistry::GetInstance()->Unregister(scheme);
+}
 }  // namespace internal
 
 Status LoadFileSystemFactories(const char* libpath) {
@@ -918,18 +942,6 @@ Result<std::shared_ptr<FileSystem>> FileSystemFromUriReal(const Uri& uri,
         "without HDFS support");
 #endif
   }
-  if (scheme == "s3") {
-#ifdef ARROW_S3
-    RETURN_NOT_OK(EnsureS3Initialized());
-    ARROW_ASSIGN_OR_RAISE(auto options, S3Options::FromUri(uri, out_path));
-    ARROW_ASSIGN_OR_RAISE(auto s3fs, S3FileSystem::Make(options, io_context));
-    return s3fs;
-#else
-    return Status::NotImplemented(
-        "Got S3 URI but Arrow compiled "
-        "without S3 support");
-#endif
-  }
 
   if (scheme == "mock") {
     // MockFileSystem does not have an

cpp/src/arrow/filesystem/filesystem.h

@@ -198,7 +198,7 @@ class ARROW_EXPORT FileSystem
   virtual Result<std::string> PathFromUri(const std::string& uri_string) const;
 
   /// \brief Make a URI from which FileSystemFromUri produces an equivalent filesystem
-  /// \param path The path component to use in the resulting URI
+  /// \param path The path component to use in the resulting URI. Must be absolute.
   /// \return A URI string, or an error if an equivalent URI cannot be produced
   virtual Result<std::string> MakeUri(std::string path) const;
 

cpp/src/arrow/filesystem/localfs.cc

@@ -288,7 +288,14 @@ Result<std::string> LocalFileSystem::PathFromUri(const std::string& uri_string)
 
 Result<std::string> LocalFileSystem::MakeUri(std::string path) const {
   ARROW_ASSIGN_OR_RAISE(path, DoNormalizePath(std::move(path)));
-  return "file://" + path + (options_.use_mmap ? "?use_mmap" : "");
+  if (!internal::DetectAbsolutePath(path)) {
+    return Status::Invalid("MakeUri requires an absolute path, got ", path);
+  }
+  ARROW_ASSIGN_OR_RAISE(auto uri, util::UriFromAbsolutePath(path));
+  if (uri[0] == '/') {
+    uri = "file://" + uri;
+  }
+  return uri + (options_.use_mmap ? "?use_mmap" : "");
 }
 
 bool LocalFileSystem::Equals(const FileSystem& other) const {

cpp/src/arrow/filesystem/localfs_test.cc

@@ -21,6 +21,7 @@
 #include <utility>
 #include <vector>
 
+#include <gmock/gmock-matchers.h>
 #include <gtest/gtest.h>
 
 #include "arrow/filesystem/filesystem.h"
@@ -428,9 +429,15 @@ TYPED_TEST(TestLocalFS, FileSystemFromUriFile) {
 
   this->TestLocalUri("file:///_?use_mmap", "/_");
   if (this->path_formatter_.supports_uri()) {
+    EXPECT_THAT(this->fs_->MakeUri(""), Raises(StatusCode::Invalid));
+    EXPECT_THAT(this->fs_->MakeUri("a/b"), Raises(StatusCode::Invalid));
+
     ASSERT_TRUE(this->local_fs_->options().use_mmap);
     ASSERT_OK_AND_ASSIGN(auto uri, this->fs_->MakeUri("/_"));
     EXPECT_EQ(uri, "file:///_?use_mmap");
+
+    ASSERT_OK_AND_ASSIGN(uri, this->fs_->MakeUri("/hello world/b/c"));
+    EXPECT_EQ(uri, "file:///hello%20world/b/c?use_mmap");
   }
 
 #ifdef _WIN32

cpp/src/arrow/filesystem/s3fs.cc

@@ -138,6 +138,7 @@
 #include "arrow/util/string.h"
 #include "arrow/util/task_group.h"
 #include "arrow/util/thread_pool.h"
+#include "arrow/util/value_parsing.h"
 
 namespace arrow::fs {
 
@@ -169,6 +170,8 @@ static constexpr const char kAwsEndpointUrlEnvVar[] = "AWS_ENDPOINT_URL";
 static constexpr const char kAwsEndpointUrlS3EnvVar[] = "AWS_ENDPOINT_URL_S3";
 static constexpr const char kAwsDirectoryContentType[] = "application/x-directory";
 
+using namespace std::string_literals;  // NOLINT(build/namespaces)
+
 // -----------------------------------------------------------------------
 // S3ProxyOptions implementation
 
@@ -3082,6 +3085,30 @@ Result<std::string> S3FileSystem::PathFromUri(const std::string& uri_string) con
                                      internal::AuthorityHandlingBehavior::kPrepend);
 }
 
+Result<std::string> S3FileSystem::MakeUri(std::string path) const {
+  if (path.length() <= 1 || path[0] != '/') {
+    return Status::Invalid("MakeUri requires an absolute, non-root path, got ", path);
+  }
+  ARROW_ASSIGN_OR_RAISE(auto uri, util::UriFromAbsolutePath(path));
+  if (!options().GetAccessKey().empty()) {
+    uri = "s3://" + options().GetAccessKey() + ":" + options().GetSecretKey() + "@" +
+          uri.substr("file:///"s.size());
+  } else {
+    uri = "s3" + uri.substr("file"s.size());
+  }
+  uri += "?";
+  uri += "region=" + util::UriEscape(options().region);
+  uri += "&";
+  uri += "scheme=" + options().scheme;
+  uri += "&";
+  uri += "endpoint_override=" + util::UriEscape(options().endpoint_override);
+  uri += "&";
+  uri += "allow_bucket_creation="s + (options().allow_bucket_creation ? "1" : "0");
+  uri += "&";
+  uri += "allow_bucket_deletion="s + (options().allow_bucket_deletion ? "1" : "0");
+  return uri;
+}
+
 S3Options S3FileSystem::options() const { return impl_->options(); }
 
 std::string S3FileSystem::region() const { return impl_->region(); }
@@ -3573,32 +3600,33 @@ bool IsS3Finalized() { return GetAwsInstance()->IsFinalized(); }
 
 S3GlobalOptions S3GlobalOptions::Defaults() {
   auto log_level = S3LogLevel::Fatal;
-
-  auto result = arrow::internal::GetEnvVar("ARROW_S3_LOG_LEVEL");
-
-  if (result.ok()) {
-    // Extract, trim, and downcase the value of the environment variable
-    auto value =
-        arrow::internal::AsciiToLower(arrow::internal::TrimString(result.ValueUnsafe()));
-
-    if (value == "fatal") {
-      log_level = S3LogLevel::Fatal;
-    } else if (value == "error") {
-      log_level = S3LogLevel::Error;
-    } else if (value == "warn") {
-      log_level = S3LogLevel::Warn;
-    } else if (value == "info") {
-      log_level = S3LogLevel::Info;
-    } else if (value == "debug") {
-      log_level = S3LogLevel::Debug;
-    } else if (value == "trace") {
-      log_level = S3LogLevel::Trace;
-    } else if (value == "off") {
-      log_level = S3LogLevel::Off;
-    }
-  }
-
-  return S3GlobalOptions{log_level};
+  int num_event_loop_threads = 1;
+  // Extract, trim, and downcase the value of the environment variable
+  auto value = arrow::internal::GetEnvVar("ARROW_S3_LOG_LEVEL")
+                   .Map(arrow::internal::AsciiToLower)
+                   .Map(arrow::internal::TrimString)
+                   .ValueOr("fatal");
+  if (value == "fatal") {
+    log_level = S3LogLevel::Fatal;
+  } else if (value == "error") {
+    log_level = S3LogLevel::Error;
+  } else if (value == "warn") {
+    log_level = S3LogLevel::Warn;
+  } else if (value == "info") {
+    log_level = S3LogLevel::Info;
+  } else if (value == "debug") {
+    log_level = S3LogLevel::Debug;
+  } else if (value == "trace") {
+    log_level = S3LogLevel::Trace;
+  } else if (value == "off") {
+    log_level = S3LogLevel::Off;
+  }
+
+  value = arrow::internal::GetEnvVar("ARROW_S3_THREADS").ValueOr("1");
+  if (uint32_t u; ::arrow::internal::ParseUnsigned(value.data(), value.size(), &u)) {
+    num_event_loop_threads = u;
+  }
+  return S3GlobalOptions{log_level, num_event_loop_threads};
 }
 
 // -----------------------------------------------------------------------
@@ -3616,4 +3644,14 @@ Result<std::string> ResolveS3BucketRegion(const std::string& bucket) {
   return resolver->ResolveRegion(bucket);
 }
 
+auto kS3FileSystemModule = ARROW_REGISTER_FILESYSTEM(
+    "s3",
+    [](const arrow::util::Uri& uri, const io::IOContext& io_context,
+       std::string* out_path) -> Result<std::shared_ptr<fs::FileSystem>> {
+      RETURN_NOT_OK(EnsureS3Initialized());
+      ARROW_ASSIGN_OR_RAISE(auto options, S3Options::FromUri(uri, out_path));
+      return S3FileSystem::Make(options, io_context);
+    },
+    [] { DCHECK_OK(EnsureS3Finalized()); });
+
 }  // namespace arrow::fs