bkietz commented on code in PR #39067:
URL: https://github.com/apache/arrow/pull/39067#discussion_r1511474468
##########
cpp/src/arrow/filesystem/filesystem.cc:
##########
@@ -674,6 +684,134 @@ Status CopyFiles(const std::shared_ptr<FileSystem>&
source_fs,
return CopyFiles(sources, destinations, io_context, chunk_size, use_threads);
}
+class FileSystemFactoryRegistry {
+ public:
+ static FileSystemFactoryRegistry* GetInstance() {
+ static FileSystemFactoryRegistry registry;
+ return ®istry;
+ }
+
+ Result<FileSystemFactory*> FactoryForScheme(const std::string& scheme) {
+ std::shared_lock lock{mutex_};
+ if (finalized_) return AlreadyFinalized();
+
+ auto it = scheme_to_factory_.find(scheme);
+ if (it == scheme_to_factory_.end()) return nullptr;
+
+ return it->second.Map([](const auto& r) { return r.factory; });
+ }
+
+ Status MergeInto(FileSystemFactoryRegistry* main_registry) {
+ std::shared_lock lock{mutex_};
+ if (finalized_) return AlreadyFinalized();
+
+ auto& [main_mutex, main_scheme_to_factory, _] = *main_registry;
+ std::unique_lock main_lock{main_mutex};
+
+ std::vector<std::string_view> duplicated_schemes;
+ for (auto& [scheme, registered] : scheme_to_factory_) {
+ if (!registered.ok()) {
+ duplicated_schemes.emplace_back(scheme);
+ continue;
+ }
+
+ auto [it, success] = main_scheme_to_factory.emplace(std::move(scheme),
registered);
+ if (success) continue;
+
+ duplicated_schemes.emplace_back(it->first);
+ }
+ scheme_to_factory_.clear();
+
+ if (duplicated_schemes.empty()) return Status::OK();
+ return Status::KeyError("Attempted to register ",
duplicated_schemes.size(),
+ " factories for schemes ['",
+ arrow::internal::JoinStrings(duplicated_schemes,
"', '"),
+ "'] but those schemes were already registered.");
+ }
+
+ void EnsureFinalized() {
+ std::unique_lock lock{mutex_};
+ if (finalized_) return;
+
+ for (const auto& [_, registered_or_error] : scheme_to_factory_) {
+ if (!registered_or_error.ok()) continue;
+ registered_or_error->finalizer();
+ }
+ finalized_ = true;
+ }
+
+ Status RegisterFactory(std::string scheme, FileSystemFactory factory, void
finalizer(),
+ bool defer_error) {
+ std::unique_lock lock{mutex_};
+ if (finalized_) return AlreadyFinalized();
+
+ auto [it, success] =
+ scheme_to_factory_.emplace(std::move(scheme), Registered{factory,
finalizer});
+ if (success) {
+ return Status::OK();
+ }
+
+ auto st = Status::KeyError("Attempted to register factory for scheme '",
it->first,
+ "' but that scheme is already registered.");
+ if (!defer_error) return st;
+
+ it->second = std::move(st);
+ return Status::OK();
+ }
+
+ private:
+ struct Registered {
+ FileSystemFactory* factory;
+ void (*finalizer)();
+ };
+
+ static Status AlreadyFinalized() {
+ return Status::Invalid("FileSystem factories were already finalized!");
+ }
+
+ std::shared_mutex mutex_;
+ std::unordered_map<std::string, Result<Registered>> scheme_to_factory_;
+ bool finalized_ = false;
+};
+
+Status RegisterFileSystemFactory(std::string scheme, FileSystemFactory factory,
+ void finalizer()) {
+ return FileSystemFactoryRegistry::GetInstance()->RegisterFactory(
+ std::move(scheme), factory, finalizer, /*defer_error=*/false);
+}
+
+void EnsureFinalized() {
FileSystemFactoryRegistry::GetInstance()->EnsureFinalized(); }
+
+FileSystemRegistrar::FileSystemRegistrar(std::string scheme, FileSystemFactory
factory,
+ void finalizer()) {
+ DCHECK_OK(FileSystemFactoryRegistry::GetInstance()->RegisterFactory(
+ std::move(scheme), std::move(factory), finalizer, /*defer_error=*/true));
+}
+
+extern "C" {
+ARROW_EXPORT void* Arrow_FileSystem_GetRegistry() {
+ return FileSystemFactoryRegistry::GetInstance();
+}
+}
+constexpr auto kGetRegistryName = "Arrow_FileSystem_GetRegistry";
+
+Status LoadFileSystemFactories(const char* libpath) {
+ using ::arrow::internal::GetSymbolAs;
+ using ::arrow::internal::LoadDynamicLibrary;
+
+ ARROW_ASSIGN_OR_RAISE(void* lib, LoadDynamicLibrary(libpath));
+
+ if (auto* get_instance = GetSymbolAs<void*()>(lib,
kGetRegistryName).ValueOr(nullptr)) {
Review Comment:
Failing to find the symbol should probably be an error. I'll fix this
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
