From 32e3f83de3924130723fdf15fb59048c878fbb74 Mon Sep 17 00:00:00 2001 From: Benjamin Kietzman Date: Mon, 6 May 2024 17:36:10 -0400 Subject: [PATCH 1/3] GH-40343: [C++] Move S3FileSystem to the registry --- cpp/cmake_modules/DefineOptions.cmake | 6 ++ cpp/src/arrow/CMakeLists.txt | 12 +++ cpp/src/arrow/filesystem/CMakeLists.txt | 18 ++++ cpp/src/arrow/filesystem/filesystem.cc | 44 ++++++---- cpp/src/arrow/filesystem/filesystem.h | 2 +- cpp/src/arrow/filesystem/localfs.cc | 9 +- cpp/src/arrow/filesystem/localfs_test.cc | 7 ++ cpp/src/arrow/filesystem/s3fs.cc | 90 ++++++++++++++------ cpp/src/arrow/filesystem/s3fs.h | 18 ++-- cpp/src/arrow/filesystem/s3fs_module.cc | 18 ++++ cpp/src/arrow/filesystem/s3fs_module_test.cc | 85 ++++++++++++++++++ cpp/src/arrow/filesystem/util_internal.cc | 9 +- cpp/src/arrow/filesystem/util_internal.h | 18 ++-- cpp/src/arrow/ipc/message.cc | 3 +- cpp/src/arrow/util/io_util.cc | 2 +- docs/source/cpp/env_vars.rst | 6 ++ 16 files changed, 277 insertions(+), 70 deletions(-) create mode 100644 cpp/src/arrow/filesystem/s3fs_module.cc create mode 100644 cpp/src/arrow/filesystem/s3fs_module_test.cc diff --git a/cpp/cmake_modules/DefineOptions.cmake b/cpp/cmake_modules/DefineOptions.cmake index 43e4e7603cfb..b446a6b6848d 100644 --- a/cpp/cmake_modules/DefineOptions.cmake +++ b/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 diff --git a/cpp/src/arrow/CMakeLists.txt b/cpp/src/arrow/CMakeLists.txt index 6e2294371e7a..ef0b05500234 100644 --- a/cpp/src/arrow/CMakeLists.txt +++ b/cpp/src/arrow/CMakeLists.txt @@ -876,6 +876,18 @@ 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 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}) diff --git a/cpp/src/arrow/filesystem/CMakeLists.txt b/cpp/src/arrow/filesystem/CMakeLists.txt index 7afdf566f2fb..5250ed2a8879 100644 --- a/cpp/src/arrow/filesystem/CMakeLists.txt +++ b/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="$" + EXTRA_LINK_LIBS + Boost::filesystem + Boost::system) + target_compile_definitions(arrow-filesystem-test + PUBLIC ARROW_S3_LIBPATH="$") + 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) diff --git a/cpp/src/arrow/filesystem/filesystem.cc b/cpp/src/arrow/filesystem/filesystem.cc index 37619df90fc3..3cf706db867c 100644 --- a/cpp/src/arrow/filesystem/filesystem.cc +++ b/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 ®istry; } + 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 finalizer; + if (it->second.ok()) { + finalizer = it->second.ValueOrDie().finalizer; + } + scheme_to_factory_.erase(it); + lock.unlock(); + + if (finalizer) { + finalizer(); + } + return Status::OK(); + } + Result 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> 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 diff --git a/cpp/src/arrow/filesystem/filesystem.h b/cpp/src/arrow/filesystem/filesystem.h index d4f62f86a748..3a47eb62f524 100644 --- a/cpp/src/arrow/filesystem/filesystem.h +++ b/cpp/src/arrow/filesystem/filesystem.h @@ -198,7 +198,7 @@ class ARROW_EXPORT FileSystem virtual Result 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 MakeUri(std::string path) const; diff --git a/cpp/src/arrow/filesystem/localfs.cc b/cpp/src/arrow/filesystem/localfs.cc index 9fe19cbf2505..0b19cc74b14a 100644 --- a/cpp/src/arrow/filesystem/localfs.cc +++ b/cpp/src/arrow/filesystem/localfs.cc @@ -288,7 +288,14 @@ Result LocalFileSystem::PathFromUri(const std::string& uri_string) Result 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 { diff --git a/cpp/src/arrow/filesystem/localfs_test.cc b/cpp/src/arrow/filesystem/localfs_test.cc index 6dd7a8c75586..2e91783c92dc 100644 --- a/cpp/src/arrow/filesystem/localfs_test.cc +++ b/cpp/src/arrow/filesystem/localfs_test.cc @@ -21,6 +21,7 @@ #include #include +#include #include #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 diff --git a/cpp/src/arrow/filesystem/s3fs.cc b/cpp/src/arrow/filesystem/s3fs.cc index b6a928ecdd34..6e47b092813e 100644 --- a/cpp/src/arrow/filesystem/s3fs.cc +++ b/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 S3FileSystem::PathFromUri(const std::string& uri_string) con internal::AuthorityHandlingBehavior::kPrepend); } +Result 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 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> { + 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 diff --git a/cpp/src/arrow/filesystem/s3fs.h b/cpp/src/arrow/filesystem/s3fs.h index f8dacd520f1b..05451b2312df 100644 --- a/cpp/src/arrow/filesystem/s3fs.h +++ b/cpp/src/arrow/filesystem/s3fs.h @@ -25,20 +25,16 @@ #include "arrow/util/macros.h" #include "arrow/util/uri.h" -namespace Aws { -namespace Auth { - +namespace Aws::Auth { class AWSCredentialsProvider; class STSAssumeRoleCredentialsProvider; +} // namespace Aws::Auth -} // namespace Auth -namespace STS { +namespace Aws::STS { class STSClient; -} -} // namespace Aws +} // namespace Aws::STS -namespace arrow { -namespace fs { +namespace arrow::fs { /// Options for using a proxy for S3 struct ARROW_EXPORT S3ProxyOptions { @@ -308,6 +304,7 @@ class ARROW_EXPORT S3FileSystem : public FileSystem { bool Equals(const FileSystem& other) const override; Result PathFromUri(const std::string& uri_string) const override; + Result MakeUri(std::string path) const override; /// \cond FALSE using FileSystem::CreateDir; @@ -461,5 +458,4 @@ Status EnsureS3Finalized(); ARROW_EXPORT Result ResolveS3BucketRegion(const std::string& bucket); -} // namespace fs -} // namespace arrow +} // namespace arrow::fs diff --git a/cpp/src/arrow/filesystem/s3fs_module.cc b/cpp/src/arrow/filesystem/s3fs_module.cc new file mode 100644 index 000000000000..d91413138b45 --- /dev/null +++ b/cpp/src/arrow/filesystem/s3fs_module.cc @@ -0,0 +1,18 @@ +// 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. + +#include "arrow/filesystem/filesystem_library.h" diff --git a/cpp/src/arrow/filesystem/s3fs_module_test.cc b/cpp/src/arrow/filesystem/s3fs_module_test.cc new file mode 100644 index 000000000000..987a8979b271 --- /dev/null +++ b/cpp/src/arrow/filesystem/s3fs_module_test.cc @@ -0,0 +1,85 @@ +// 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. + +#include +#include +#include +#include +#include +#include +#include + +#include +#include + +#include "arrow/filesystem/filesystem.h" +#include "arrow/filesystem/s3_test_util.h" +#include "arrow/filesystem/s3fs.h" +#include "arrow/filesystem/test_util.h" +#include "arrow/filesystem/util_internal.h" +#include "arrow/result.h" +#include "arrow/status.h" +#include "arrow/testing/future_util.h" +#include "arrow/testing/gtest_util.h" +#include "arrow/testing/matchers.h" +#include "arrow/util/async_generator.h" +#include "arrow/util/checked_cast.h" +#include "arrow/util/future.h" +#include "arrow/util/io_util.h" +#include "arrow/util/key_value_metadata.h" +#include "arrow/util/logging.h" +#include "arrow/util/macros.h" +#include "arrow/util/range.h" +#include "arrow/util/string.h" + +namespace arrow::fs { + +auto* minio_env = ::testing::AddGlobalTestEnvironment(new MinioTestEnvironment); + +MinioTestEnvironment* GetMinioEnv() { + return ::arrow::internal::checked_cast(minio_env); +} + +class RegistrationTestEnvironment : public ::testing::Environment { + public: + void SetUp() override { + // Unregister the s3 filesystem factory so that we can be sure the module loading and + // the factories from the module are actually working + ASSERT_OK(internal::UnregisterFileSystemFactory("s3")); + ASSERT_OK(LoadFileSystemFactories(ARROW_S3_LIBPATH)); + } + void TearDown() override { EnsureFinalized(); } +}; + +auto* lib_env = ::testing::AddGlobalTestEnvironment(new RegistrationTestEnvironment); + +TEST(S3Test, FromUri) { + ASSERT_OK_AND_ASSIGN(auto minio, GetMinioEnv()->GetOneServer()); + + std::string path; + ASSERT_OK_AND_ASSIGN(auto fs, FileSystemFromUri("s3://" + minio->access_key() + ":" + + minio->secret_key() + + "@bucket/somedir/subdir/subfile", + &path)); + + EXPECT_EQ(fs->MakeUri("/" + path), + "s3://minio:miniopass@bucket/somedir/subdir/subfile" + "?region=us-east-1&scheme=https&endpoint_override=" + "&allow_bucket_creation=0&allow_bucket_deletion=0"); +} + +} // namespace arrow::fs diff --git a/cpp/src/arrow/filesystem/util_internal.cc b/cpp/src/arrow/filesystem/util_internal.cc index be43e14e8433..589886840842 100644 --- a/cpp/src/arrow/filesystem/util_internal.cc +++ b/cpp/src/arrow/filesystem/util_internal.cc @@ -26,14 +26,13 @@ #include "arrow/status.h" #include "arrow/util/io_util.h" #include "arrow/util/string.h" +#include "arrow/util/uri.h" namespace arrow { using internal::StatusDetailFromErrno; -using util::Uri; -namespace fs { -namespace internal { +namespace fs::internal { TimePoint CurrentTimePoint() { auto now = std::chrono::system_clock::now(); @@ -262,6 +261,6 @@ Result GlobFiles(const std::shared_ptr& filesystem, FileSystemGlobalOptions global_options; -} // namespace internal -} // namespace fs +} // namespace fs::internal + } // namespace arrow diff --git a/cpp/src/arrow/filesystem/util_internal.h b/cpp/src/arrow/filesystem/util_internal.h index 74ddf015432d..62e253ebc1a1 100644 --- a/cpp/src/arrow/filesystem/util_internal.h +++ b/cpp/src/arrow/filesystem/util_internal.h @@ -24,13 +24,9 @@ #include "arrow/filesystem/filesystem.h" #include "arrow/io/interfaces.h" #include "arrow/status.h" -#include "arrow/util/uri.h" #include "arrow/util/visibility.h" -namespace arrow { -using util::Uri; -namespace fs { -namespace internal { +namespace arrow::fs::internal { ARROW_EXPORT TimePoint CurrentTimePoint(); @@ -101,6 +97,12 @@ Result GlobFiles(const std::shared_ptr& filesystem, extern FileSystemGlobalOptions global_options; -} // namespace internal -} // namespace fs -} // namespace arrow +/// \brief Unregister filesystem factories +/// +/// For testing purposes, it can be useful to remove filesystem factories from +/// the registry. This allows a test to emulate loading an unknown filesystem +/// module even if the library has built-in support for the schemes in the module. +ARROW_EXPORT +Status UnregisterFileSystemFactory(const std::string& scheme); + +} // namespace arrow::fs::internal diff --git a/cpp/src/arrow/ipc/message.cc b/cpp/src/arrow/ipc/message.cc index 27ded52861ea..327e4600825b 100644 --- a/cpp/src/arrow/ipc/message.cc +++ b/cpp/src/arrow/ipc/message.cc @@ -370,7 +370,8 @@ Result> ReadMessage(int64_t offset, int32_t metadata_le ARROW_ASSIGN_OR_RAISE(auto metadata, file->ReadAt(offset, metadata_length)); if (metadata->size() < metadata_length) { return Status::Invalid("Expected to read ", metadata_length, - " metadata bytes but got ", metadata->size()); + " metadata bytes at offset ", offset, " but got ", + metadata->size()); } ARROW_RETURN_NOT_OK(decoder.Consume(metadata)); diff --git a/cpp/src/arrow/util/io_util.cc b/cpp/src/arrow/util/io_util.cc index 18d7de64a0ef..cb03d5ccba83 100644 --- a/cpp/src/arrow/util/io_util.cc +++ b/cpp/src/arrow/util/io_util.cc @@ -2232,7 +2232,7 @@ Result LoadDynamicLibrary(const char* path) { constexpr int kFlags = // All undefined symbols in the shared object are resolved before dlopen() returns. RTLD_NOW - // Symbols defined in this shared object are not made available to + // Symbols defined in this shared object are not made available to // resolve references in subsequently loaded shared objects. | RTLD_LOCAL; if (void* handle = dlopen(path, kFlags)) return handle; diff --git a/docs/source/cpp/env_vars.rst b/docs/source/cpp/env_vars.rst index 0a082b0a5d85..6a8a02254847 100644 --- a/docs/source/cpp/env_vars.rst +++ b/docs/source/cpp/env_vars.rst @@ -108,6 +108,12 @@ that changing their value later will have an effect. `Logging - AWS SDK For C++ `__ +.. envvar:: ARROW_S3_THREADS + + The number of threads to configure when creating AWS' I/O event loop. + + Defaults to 1 as recommended by AWS' doc when the # of connections is + expected to be, at most, in the hundreds. .. envvar:: ARROW_TRACING_BACKEND From 6806bcc46824310b226ba38477d04bed6e0c4e89 Mon Sep 17 00:00:00 2001 From: Benjamin Kietzman Date: Wed, 25 Sep 2024 19:38:33 -0500 Subject: [PATCH 2/3] include warning about which option should be specified --- cpp/src/arrow/CMakeLists.txt | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/cpp/src/arrow/CMakeLists.txt b/cpp/src/arrow/CMakeLists.txt index ef0b05500234..5712a210702d 100644 --- a/cpp/src/arrow/CMakeLists.txt +++ b/cpp/src/arrow/CMakeLists.txt @@ -879,7 +879,8 @@ if(ARROW_FILESYSTEM) if(ARROW_S3_MODULE) if(NOT ARROW_BUILD_SHARED) - message(FATAL_ERROR "ARROW_S3_MODULE without shared libarrow is not supported") + 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) From 1f5a17eb620c25b8a793a923548ebc0ebcfc6a9e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ra=C3=BAl=20Cumplido?= Date: Fri, 10 Jan 2025 16:29:54 +0100 Subject: [PATCH 3/3] Just tests on top of Ben's PR --- cpp/CMakePresets.json | 44 ++++++++++++------------- cpp/src/arrow/filesystem/s3fs.h | 2 +- cpp/src/arrow/filesystem/s3fs_module.cc | 1 + python/pyarrow/_fs.pyx | 10 ++++++ python/pyarrow/includes/libarrow_fs.pxd | 3 ++ python/pyarrow/tests/test_fs.py | 26 ++++++++++++--- 6 files changed, 58 insertions(+), 28 deletions(-) diff --git a/cpp/CMakePresets.json b/cpp/CMakePresets.json index 2fb7b2044f54..ef9c3ab8812c 100644 --- a/cpp/CMakePresets.json +++ b/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" } }, { diff --git a/cpp/src/arrow/filesystem/s3fs.h b/cpp/src/arrow/filesystem/s3fs.h index 05451b2312df..d5f64d78b6e5 100644 --- a/cpp/src/arrow/filesystem/s3fs.h +++ b/cpp/src/arrow/filesystem/s3fs.h @@ -254,7 +254,7 @@ struct ARROW_EXPORT S3Options { /// /// This is recommended if you use the standard AWS environment variables /// and/or configuration file. - static S3Options Defaults(); + ARROW_EXPORT static S3Options Defaults(); /// \brief Initialize with anonymous credentials. /// diff --git a/cpp/src/arrow/filesystem/s3fs_module.cc b/cpp/src/arrow/filesystem/s3fs_module.cc index d91413138b45..31dc78f79167 100644 --- a/cpp/src/arrow/filesystem/s3fs_module.cc +++ b/cpp/src/arrow/filesystem/s3fs_module.cc @@ -16,3 +16,4 @@ // under the License. #include "arrow/filesystem/filesystem_library.h" +#include "arrow/filesystem/s3fs.h" diff --git a/python/pyarrow/_fs.pyx b/python/pyarrow/_fs.pyx index e315dd6381f4..1a01042d0901 100644 --- a/python/pyarrow/_fs.pyx +++ b/python/pyarrow/_fs.pyx @@ -475,6 +475,16 @@ cdef class FileSystem(_Weakrefable): with nogil: result = CFileSystemFromUriOrPath(c_uri, &c_path) return FileSystem.wrap(GetResultValue(result)), frombytes(c_path) + + def load_file_system(str lib_path): + cdef: + const char* c_path + bytes encoded_path + + encoded_path = lib_path.encode('utf-8') + c_path = encoded_path + with nogil: + check_status(CLoadFileSystemFactories(c_path)) cdef init(self, const shared_ptr[CFileSystem]& wrapped): self.wrapped = wrapped diff --git a/python/pyarrow/includes/libarrow_fs.pxd b/python/pyarrow/includes/libarrow_fs.pxd index cc260b80c777..7ca52c6ba93a 100644 --- a/python/pyarrow/includes/libarrow_fs.pxd +++ b/python/pyarrow/includes/libarrow_fs.pxd @@ -93,6 +93,9 @@ cdef extern from "arrow/filesystem/api.h" namespace "arrow::fs" nogil: "arrow::fs::FileSystemFromUriOrPath"(const c_string& uri, c_string* out_path) + CStatus CLoadFileSystemFactories \ + "arrow::fs::LoadFileSystemFactories"(const char* libpath) + cdef cppclass CFileSystemGlobalOptions \ "arrow::fs::FileSystemGlobalOptions": c_string tls_ca_file_path diff --git a/python/pyarrow/tests/test_fs.py b/python/pyarrow/tests/test_fs.py index 99d82c3cbf02..5d68c7027eb2 100644 --- a/python/pyarrow/tests/test_fs.py +++ b/python/pyarrow/tests/test_fs.py @@ -1574,18 +1574,36 @@ def test_filesystem_from_path_object(path): assert path == p.resolve().absolute().as_posix() -@pytest.mark.s3 +def load_shared_library(lib_path, global_scope=False): + import ctypes + if global_scope: + lib = ctypes.CDLL(lib_path, mode=ctypes.RTLD_GLOBAL | os.RTLD_DEEPBIND) + else: + lib = ctypes.CDLL(lib_path) + return lib + + def test_filesystem_from_uri_s3(s3_server): - from pyarrow.fs import S3FileSystem + # Load libarrow_s3fs.so + # s3fs_lib_path = ( + # '/home/raulcd/code/arrow/python/pyarrow/_s3fs.cpython-312-x86_64-linux-gnu.so' + # ) + + libarrow_s3fs_path = '/home/raulcd/code/libarrow_s3fs.so' + FileSystem.load_file_system(libarrow_s3fs_path) + import ctypes + lib = ctypes.CDLL(libarrow_s3fs_path, mode=ctypes.RTLD_GLOBAL) + + assert lib is not None host, port, access_key, secret_key = s3_server['connection'] uri = "s3://{}:{}@mybucket/foo/bar?scheme=http&endpoint_override={}:{}"\ "&allow_bucket_creation=True" \ .format(access_key, secret_key, host, port) - fs, path = FileSystem.from_uri(uri) - assert isinstance(fs, S3FileSystem) + # from pyarrow.fs import S3FileSystem + # assert isinstance(fs, S3FileSystem) assert path == "mybucket/foo/bar" fs.create_dir(path)