diff --git a/CMakeLists.txt b/CMakeLists.txt index 06668ab5..77ed0c34 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -52,6 +52,7 @@ option(PAIMON_USE_UBSAN "Use Undefined Behavior Sanitizer" OFF) option(PAIMON_USE_CXX11_ABI "Use C++11 ABI" ON) option(PAIMON_ENABLE_AVRO "Whether to enable avro file format" ON) option(PAIMON_ENABLE_ORC "Whether to enable orc file format" ON) +option(PAIMON_ENABLE_JINDO "Whether to enable jindo file system" OFF) option(PAIMON_ENABLE_LUCENE "Whether to enable lucene index" OFF) if(PAIMON_ENABLE_LUCENE) message(FATAL_ERROR "PAIMON_ENABLE_LUCENE is temporarily disabled. Please configure with -DPAIMON_ENABLE_LUCENE=OFF." @@ -64,6 +65,9 @@ endif() if(PAIMON_ENABLE_AVRO) add_definitions(-DPAIMON_ENABLE_AVRO) endif() +if(PAIMON_ENABLE_JINDO) + add_definitions(-DPAIMON_ENABLE_JINDO) +endif() if(PAIMON_USE_CXX11_ABI) add_definitions(-D_GLIBCXX_USE_CXX11_ABI=1) else() @@ -405,6 +409,11 @@ if(PAIMON_BUILD_TESTS) list(APPEND TEST_STATIC_LINK_LIBS paimon_avro_file_format_shared) list(APPEND TEST_STATIC_LINK_LIBS "-Wl,--as-needed") endif() + if(PAIMON_ENABLE_JINDO) + list(APPEND TEST_STATIC_LINK_LIBS "-Wl,--no-as-needed") + list(APPEND TEST_STATIC_LINK_LIBS paimon_jindo_file_system_shared) + list(APPEND TEST_STATIC_LINK_LIBS "-Wl,--as-needed") + endif() if(PAIMON_ENABLE_LUMINA) list(APPEND TEST_STATIC_LINK_LIBS "-Wl,--no-as-needed") list(APPEND TEST_STATIC_LINK_LIBS paimon_lumina_index_shared) @@ -437,6 +446,9 @@ config_summary_message() add_subdirectory(src/paimon) add_subdirectory(src/paimon/fs/local) +if(PAIMON_ENABLE_JINDO) + add_subdirectory(src/paimon/fs/jindo) +endif() add_subdirectory(src/paimon/format/blob) add_subdirectory(src/paimon/format/orc) add_subdirectory(src/paimon/format/parquet) diff --git a/ci/scripts/build_paimon.sh b/ci/scripts/build_paimon.sh index 45f1bd9c..8847caf4 100755 --- a/ci/scripts/build_paimon.sh +++ b/ci/scripts/build_paimon.sh @@ -39,6 +39,7 @@ CMAKE_ARGS=( "-G Ninja" "-DCMAKE_BUILD_TYPE=${build_type}" "-DPAIMON_BUILD_TESTS=ON" + "-DPAIMON_ENABLE_JINDO=ON" "-DPAIMON_ENABLE_LUCENE=OFF" ) diff --git a/cmake_modules/ThirdpartyToolchain.cmake b/cmake_modules/ThirdpartyToolchain.cmake index fc4f31cf..669301bc 100644 --- a/cmake_modules/ThirdpartyToolchain.cmake +++ b/cmake_modules/ThirdpartyToolchain.cmake @@ -281,6 +281,56 @@ else() endif() endif() +if(APPLE) + set(JINDOSDK_C_DYNAMIC_LIB_NAME "jindosdk_c.${PAIMON_JINDOSDK_C_BUILD_VERSION}") + set(JINDOSDK_C_DYNAMIC_LIB_FILE "lib${JINDOSDK_C_DYNAMIC_LIB_NAME}.dylib") + if(CMAKE_SYSTEM_PROCESSOR MATCHES "x86_64|amd64|AMD64") + set(JINDOSDK_C_BUILD_SHA256_CHECKSUM + "${PAIMON_JINDOSDK_C_MACOS_X86_64_BUILD_SHA256_CHECKSUM}") + if(DEFINED ENV{PAIMON_JINDOSDK_C_MACOS_X86_64_URL}) + set(JINDOSDK_C_SOURCE_URL "$ENV{PAIMON_JINDOSDK_C_MACOS_X86_64_URL}") + else() + set_urls(JINDOSDK_C_SOURCE_URL + "https://jindodata-binary.oss-cn-shanghai.aliyuncs.com/release/${PAIMON_JINDOSDK_C_BUILD_VERSION}/jindosdk-${PAIMON_JINDOSDK_C_BUILD_VERSION}-macos-11_0-x86_64.tar.gz" + ) + endif() + elseif(CMAKE_SYSTEM_PROCESSOR MATCHES "aarch64|arm64|ARM64") + set(JINDOSDK_C_BUILD_SHA256_CHECKSUM + "${PAIMON_JINDOSDK_C_MACOS_AARCH64_BUILD_SHA256_CHECKSUM}") + if(DEFINED ENV{PAIMON_JINDOSDK_C_MACOS_AARCH64_URL}) + set(JINDOSDK_C_SOURCE_URL "$ENV{PAIMON_JINDOSDK_C_MACOS_AARCH64_URL}") + else() + set_urls(JINDOSDK_C_SOURCE_URL + "https://jindodata-binary.oss-cn-shanghai.aliyuncs.com/release/${PAIMON_JINDOSDK_C_BUILD_VERSION}/jindosdk-${PAIMON_JINDOSDK_C_BUILD_VERSION}-macos-11_0-aarch64.tar.gz" + ) + endif() + endif() +else() + set(JINDOSDK_C_DYNAMIC_LIB_NAME "jindosdk_c") + set(JINDOSDK_C_DYNAMIC_LIB_FILE "lib${JINDOSDK_C_DYNAMIC_LIB_NAME}.so") + if(CMAKE_SYSTEM_PROCESSOR MATCHES "x86_64|amd64|AMD64") + set(JINDOSDK_C_BUILD_SHA256_CHECKSUM + "${PAIMON_JINDOSDK_C_LINUX_X86_64_BUILD_SHA256_CHECKSUM}") + if(DEFINED ENV{PAIMON_JINDOSDK_C_LINUX_X86_64_URL}) + set(JINDOSDK_C_SOURCE_URL "$ENV{PAIMON_JINDOSDK_C_LINUX_X86_64_URL}") + else() + set_urls(JINDOSDK_C_SOURCE_URL + "https://jindodata-binary.oss-cn-shanghai.aliyuncs.com/release/${PAIMON_JINDOSDK_C_BUILD_VERSION}/jindosdk-${PAIMON_JINDOSDK_C_BUILD_VERSION}-linux.tar.gz" + ) + endif() + elseif(CMAKE_SYSTEM_PROCESSOR MATCHES "aarch64|arm64|ARM64") + set(JINDOSDK_C_BUILD_SHA256_CHECKSUM + "${PAIMON_JINDOSDK_C_LINUX_AARCH64_BUILD_SHA256_CHECKSUM}") + if(DEFINED ENV{PAIMON_JINDOSDK_C_LINUX_AARCH64_URL}) + set(JINDOSDK_C_SOURCE_URL "$ENV{PAIMON_JINDOSDK_C_LINUX_AARCH64_URL}") + else() + set_urls(JINDOSDK_C_SOURCE_URL + "https://jindodata-binary.oss-cn-shanghai.aliyuncs.com/release/${PAIMON_JINDOSDK_C_BUILD_VERSION}/jindosdk-${PAIMON_JINDOSDK_C_BUILD_VERSION}-linux-el7-aarch64.tar.gz" + ) + endif() + endif() +endif() + set(EP_CXX_FLAGS "${CMAKE_CXX_FLAGS}") set(EP_C_FLAGS "${CMAKE_C_FLAGS}") string(REPLACE "-Wglobal-constructors" "" EP_CXX_FLAGS ${EP_CXX_FLAGS}) @@ -1105,6 +1155,97 @@ macro(build_lz4) add_dependencies(lz4 lz4_ep) endmacro() +macro(build_jindosdk_c) + message(STATUS "Building jindosdk-c from precompiled package") + + set(JINDOSDK_C_PREFIX "${CMAKE_CURRENT_BINARY_DIR}/jindosdk_ep-install") + set(JINDOSDK_C_HOME "${JINDOSDK_C_PREFIX}") + set(JINDOSDK_C_INCLUDE_DIR "${JINDOSDK_C_PREFIX}/include") + set(JINDOSDK_C_LIB_DIR "${JINDOSDK_C_PREFIX}/lib/native") + set(JINDOSDK_C_DYNAMIC_LIB "${JINDOSDK_C_LIB_DIR}/${JINDOSDK_C_DYNAMIC_LIB_FILE}") + + # Extract and install jindosdk from precompiled package + externalproject_add(jindosdk_ep + URL ${JINDOSDK_C_SOURCE_URL} + URL_HASH "SHA256=${JINDOSDK_C_BUILD_SHA256_CHECKSUM}" + ${THIRDPARTY_LOG_OPTIONS} + CONFIGURE_COMMAND "" + BUILD_COMMAND "" + INSTALL_COMMAND bash -c + "cp -r /include/* ${JINDOSDK_C_INCLUDE_DIR}" + COMMAND bash -c + "cp -r /lib/native/${JINDOSDK_C_DYNAMIC_LIB_FILE}* ${JINDOSDK_C_LIB_DIR}" + BUILD_BYPRODUCTS "${JINDOSDK_C_DYNAMIC_LIB}") + + # The include directory must exist before it is referenced by a target. + file(MAKE_DIRECTORY "${JINDOSDK_C_INCLUDE_DIR}") + file(MAKE_DIRECTORY "${JINDOSDK_C_LIB_DIR}") + + add_library(jindosdk::c_sdk SHARED IMPORTED) + set_target_properties(jindosdk::c_sdk + PROPERTIES IMPORTED_LOCATION "${JINDOSDK_C_DYNAMIC_LIB}" + INTERFACE_INCLUDE_DIRECTORIES + "${JINDOSDK_C_INCLUDE_DIR}") + list(APPEND JINDOSDK_INCLUDE_DIR ${JINDOSDK_C_INCLUDE_DIR}) + + add_dependencies(jindosdk::c_sdk jindosdk_ep) + install(DIRECTORY "${JINDOSDK_C_LIB_DIR}/" + DESTINATION ${CMAKE_INSTALL_LIBDIR} + FILES_MATCHING + PATTERN "${JINDOSDK_C_DYNAMIC_LIB_FILE}*") + +endmacro() + +macro(build_jindosdk_nextarch) + message(STATUS "Building jindosdk-nextarch from local source") + + set(JINDOSDK_NEXTARCH_PREFIX + "${CMAKE_CURRENT_BINARY_DIR}/jindosdk-nextarch_ep-install") + set(JINDOSDK_NEXTARCH_HOME "${JINDOSDK_NEXTARCH_PREFIX}") + set(JINDOSDK_NEXTARCH_INCLUDE_DIR "${JINDOSDK_NEXTARCH_PREFIX}/include") + set(JINDOSDK_NEXTARCH_LIB_DIR "${JINDOSDK_NEXTARCH_PREFIX}/lib") + set(JINDOSDK_NEXTARCH_SOURCE_DIR "${CMAKE_SOURCE_DIR}/third_party/jindosdk-nextarch") + set(JINDOSDK_NEXTARCH_STATIC_LIB + "${JINDOSDK_NEXTARCH_LIB_DIR}/libjindosdk-nextarch.a") + + # Get jindosdk dependencies (headers and dynamic library) + get_target_property(JINDOSDK_C_INCLUDE_DIR jindosdk::c_sdk + INTERFACE_INCLUDE_DIRECTORIES) + get_target_property(JINDOSDK_C_LIBRARY_LOCATION jindosdk::c_sdk IMPORTED_LOCATION) + get_filename_component(JINDOSDK_C_DIR_ROOT "${JINDOSDK_C_INCLUDE_DIR}" DIRECTORY) + + # Compile flags for jindosdk-nextarch + set(JINDOSDK_NEXTARCH_CMAKE_CXX_FLAGS "${EP_CXX_FLAGS}") + set(JINDOSDK_NEXTARCH_CMAKE_C_FLAGS "${EP_C_FLAGS}") + set(JINDOSDK_NEXTARCH_CMAKE_ARGS + ${EP_COMMON_CMAKE_ARGS} + "-DCMAKE_INSTALL_PREFIX=${JINDOSDK_NEXTARCH_PREFIX}" + "-DCMAKE_CXX_FLAGS=${JINDOSDK_NEXTARCH_CMAKE_CXX_FLAGS}" + "-DCMAKE_C_FLAGS=${JINDOSDK_NEXTARCH_CMAKE_C_FLAGS}" + -DJINDOSDK_ROOT=${JINDOSDK_C_DIR_ROOT} + -DJINDOSDK_LIBRARY_NAME=${JINDOSDK_C_DYNAMIC_LIB_NAME}) + + externalproject_add(jindosdk-nextarch_ep + SOURCE_DIR ${JINDOSDK_NEXTARCH_SOURCE_DIR} + CMAKE_ARGS ${JINDOSDK_NEXTARCH_CMAKE_ARGS} + BUILD_BYPRODUCTS "${JINDOSDK_NEXTARCH_STATIC_LIB}" + DEPENDS jindosdk::c_sdk ${THIRDPARTY_LOG_OPTIONS}) + + # The include directory must exist before it is referenced by a target. + file(MAKE_DIRECTORY "${JINDOSDK_NEXTARCH_INCLUDE_DIR}") + file(MAKE_DIRECTORY "${JINDOSDK_NEXTARCH_LIB_DIR}") + + add_library(jindosdk::nextarch STATIC IMPORTED) + set_target_properties(jindosdk::nextarch + PROPERTIES IMPORTED_LOCATION "${JINDOSDK_NEXTARCH_STATIC_LIB}" + INTERFACE_INCLUDE_DIRECTORIES + "${JINDOSDK_NEXTARCH_INCLUDE_DIR}") + target_link_libraries(jindosdk::nextarch INTERFACE jindosdk::c_sdk pthread dl) + list(APPEND JINDOSDK_INCLUDE_DIR ${JINDOSDK_NEXTARCH_INCLUDE_DIR}) + + add_dependencies(jindosdk::nextarch jindosdk-nextarch_ep) +endmacro() + macro(build_protobuf) message(STATUS "Building protobuf from source") set(PROTOBUF_PREFIX "${CMAKE_CURRENT_BINARY_DIR}/protobuf_ep-install") @@ -1672,6 +1813,10 @@ if(PAIMON_ENABLE_ORC) resolve_dependency(Protobuf) resolve_dependency(ORC) endif() +if(PAIMON_ENABLE_JINDO) + build_jindosdk_c() + build_jindosdk_nextarch() +endif() if(PAIMON_ENABLE_LUCENE) build_boost() build_lucene() diff --git a/docs/source/build_system.rst b/docs/source/build_system.rst index 96ae92b6..61739eb3 100644 --- a/docs/source/build_system.rst +++ b/docs/source/build_system.rst @@ -86,6 +86,7 @@ Paimon provides a set of built-in optional plugins that you can link to as neede - File system plugins: - ``paimon_local_file_system_shared`` / ``paimon_local_file_system_static`` + - ``paimon_jindo_file_system_shared`` / ``paimon_jindo_file_system_static`` - Index plugins: diff --git a/docs/source/building.rst b/docs/source/building.rst index f63df31b..68efa5ca 100644 --- a/docs/source/building.rst +++ b/docs/source/building.rst @@ -123,6 +123,7 @@ boolean flags to ``cmake``. * ``-DPAIMON_ENABLE_ORC=ON``: Paimon integration with Apache ORC * ``-DPAIMON_ENABLE_AVRO=ON``: Apache Avro libraries and Paimon integration +* ``-DPAIMON_ENABLE_JINDO=ON``: Support for Alibaba Jindo filesystems Third-party dependency source ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ diff --git a/src/paimon/fs/jindo/CMakeLists.txt b/src/paimon/fs/jindo/CMakeLists.txt new file mode 100644 index 00000000..9f79ac56 --- /dev/null +++ b/src/paimon/fs/jindo/CMakeLists.txt @@ -0,0 +1,35 @@ +# 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. + +if(PAIMON_ENABLE_JINDO) + set(PAIMON_JINDO_FILE_SYSTEM jindo_file_system.cpp jindo_file_system_factory.cpp) + + add_paimon_lib(paimon_jindo_file_system + SOURCES + ${PAIMON_JINDO_FILE_SYSTEM} + EXTRA_INCLUDES + ${JINDOSDK_INCLUDE_DIR} + DEPENDENCIES + paimon_shared + jindosdk::nextarch + STATIC_LINK_LIBS + jindosdk::nextarch + fmt + SHARED_LINK_LIBS + paimon_shared + SHARED_LINK_FLAGS + ${PAIMON_VERSION_SCRIPT_FLAGS}) +endif() diff --git a/src/paimon/fs/jindo/jindo_file_status.h b/src/paimon/fs/jindo/jindo_file_status.h new file mode 100644 index 00000000..5ac14cfd --- /dev/null +++ b/src/paimon/fs/jindo/jindo_file_status.h @@ -0,0 +1,67 @@ +/* + * 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. + */ + +#pragma once + +#include +#include + +#include "JdoFileInfo.hpp" // NOLINT(build/include_subdir) +#include "paimon/fs/file_system.h" + +namespace paimon::jindo { +class JindoBasicFileStatus : public BasicFileStatus { + public: + explicit JindoBasicFileStatus(JdoFileInfo&& file_info) : file_info_(std::move(file_info)) {} + + std::string GetPath() const override { + return file_info_.getPath(); + } + + bool IsDir() const override { + return file_info_.isDir(); + } + + private: + JdoFileInfo file_info_; +}; + +class JindoFileStatus : public FileStatus { + public: + explicit JindoFileStatus(JdoFileInfo&& file_info) : file_info_(std::move(file_info)) {} + + std::string GetPath() const override { + return file_info_.getPath(); + } + + uint64_t GetLen() const override { + return file_info_.getLength(); + } + + int64_t GetModificationTime() const override { + return file_info_.getMtime(); + } + + bool IsDir() const override { + return file_info_.isDir(); + } + + private: + JdoFileInfo file_info_; +}; +} // namespace paimon::jindo diff --git a/src/paimon/fs/jindo/jindo_file_system.cpp b/src/paimon/fs/jindo/jindo_file_system.cpp new file mode 100644 index 00000000..afd5a383 --- /dev/null +++ b/src/paimon/fs/jindo/jindo_file_system.cpp @@ -0,0 +1,278 @@ +/* + * 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 "paimon/fs/jindo/jindo_file_system.h" + +#include +#include + +#include "JdoFileInfo.hpp" // NOLINT(build/include_subdir) +#include "JdoFileSystem.hpp" // NOLINT(build/include_subdir) +#include "JdoListResult.hpp" // NOLINT(build/include_subdir) +#include "JdoStatus.hpp" // NOLINT(build/include_subdir) +#include "fmt/format.h" +#include "jdo_error.h" // NOLINT(build/include_subdir) +#include "paimon/fs/jindo/jindo_file_status.h" +#include "paimon/fs/jindo/jindo_utils.h" + +namespace paimon::jindo { + +class JindoFileSystemImpl { + public: + explicit JindoFileSystemImpl(std::unique_ptr&& fs) : fs_(std::move(fs)) { + assert(fs_); + } + ~JindoFileSystemImpl() { + if (fs_) { + [[maybe_unused]] auto status = fs_->destroy(); + assert(status.ok()); + fs_.reset(); + } + } + JdoFileSystem* GetFileSystem() { + assert(fs_); + return fs_.get(); + } + + private: + std::unique_ptr fs_; +}; + +JindoFileSystem::JindoFileSystem(std::unique_ptr&& fs) + : impl_(std::make_shared(std::move(fs))) {} + +Result> JindoFileSystem::Open(const std::string& path) const { + std::unique_ptr reader; + PAIMON_RETURN_NOT_OK_FROM_JINDO(impl_->GetFileSystem()->openReader(path, &reader)); + return std::make_unique(impl_, std::move(reader)); +} + +Result> JindoFileSystem::Create(const std::string& path, + bool overwrite) const { + PAIMON_ASSIGN_OR_RAISE(bool exist, Exists(path)); + if (exist && !overwrite) { + return Status::Invalid( + fmt::format("do not allow overwrite, but the file {} already exists", path)); + } + std::unique_ptr writer; + PAIMON_RETURN_NOT_OK_FROM_JINDO(impl_->GetFileSystem()->openWriter(path, &writer)); + return std::make_unique(impl_, std::move(writer)); +} + +Status JindoFileSystem::Mkdirs(const std::string& path) const { + PAIMON_RETURN_NOT_OK_FROM_JINDO(impl_->GetFileSystem()->mkdir(path, /*recursive=*/true)); + return Status::OK(); +} + +Status JindoFileSystem::Rename(const std::string& src, const std::string& dst) const { + PAIMON_ASSIGN_OR_RAISE(bool is_src_exist, Exists(src)); + if (!is_src_exist) { + return Status::NotExist( + fmt::format("rename {} to {} failed, because: src file not exist", src, dst)); + } + PAIMON_ASSIGN_OR_RAISE(bool is_dst_exist, Exists(dst)); + if (is_dst_exist) { + return Status::Invalid( + fmt::format("rename {} to {} failed, because: dst file already exist", src, dst)); + } + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr file_status, GetFileStatus(src)); + if (!file_status->IsDir() && dst.back() == '/') { + return Status::Invalid( + fmt::format("rename {} to {} failed, because: src file is not a dir", src, dst)); + } + PAIMON_RETURN_NOT_OK_FROM_JINDO(impl_->GetFileSystem()->rename(src, dst)); + return Status::OK(); +} + +Result JindoFileSystem::Exists(const std::string& path) const { + auto status = impl_->GetFileSystem()->exists(path); + if (status.ok()) { + return true; + } else if (status.getErrCode() == JDO_FILE_NOT_FOUND_ERROR) { + return false; + } + return Status::IOError(status.errMsg()); +} + +Status JindoFileSystem::Delete(const std::string& path, bool recursive) const { + PAIMON_RETURN_NOT_OK_FROM_JINDO(impl_->GetFileSystem()->remove(path, recursive)); + return Status::OK(); +} + +Result> JindoFileSystem::GetFileStatus(const std::string& path) const { + JdoFileInfo file_info; + PAIMON_RETURN_NOT_OK_FROM_JINDO(impl_->GetFileSystem()->getFileInfo(path, &file_info)); + return std::make_unique(std::move(file_info)); +} + +Status JindoFileSystem::ListDir( + const std::string& directory, + std::vector>* file_status_list) const { + PAIMON_ASSIGN_OR_RAISE(bool exist, Exists(directory)); + if (!exist) { + return Status::OK(); + } + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr file_status, GetFileStatus(directory)); + if (!file_status->IsDir()) { + return Status::Invalid( + fmt::format("file {} already exists and is not a directory", file_status->GetPath())); + } + JdoListResult list_result; + while (true) { + PAIMON_RETURN_NOT_OK_FROM_JINDO( + impl_->GetFileSystem()->listDir(directory, /*recursive=*/false, &list_result)); + auto file_infos = list_result.getFileInfos(); + file_status_list->reserve(file_status_list->size() + file_infos.size()); + for (auto& file_info : file_infos) { + file_status_list->push_back( + std::make_unique(std::move(file_info))); + } + if (!list_result.isTruncated()) { + break; + } + // TODO(xinyu.lxy): test truncated + list_result.setFileInfos({}); + list_result.setTruncated(false); + } + return Status::OK(); +} + +Status JindoFileSystem::ListFileStatus( + const std::string& path, std::vector>* file_status_list) const { + PAIMON_ASSIGN_OR_RAISE(bool exist, Exists(path)); + if (!exist) { + return Status::OK(); + } + + JdoListResult list_result; + while (true) { + PAIMON_RETURN_NOT_OK_FROM_JINDO( + impl_->GetFileSystem()->listDir(path, /*recursive=*/false, &list_result)); + auto file_infos = list_result.getFileInfos(); + file_status_list->reserve(file_status_list->size() + file_infos.size()); + for (auto& file_info : file_infos) { + // oss not support list FileStatus, only return BasicFileStatus + // call GetFileStatus to return FileStatus + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr file_status, + GetFileStatus(file_info.getPath())); + file_status_list->push_back(std::move(file_status)); + } + if (!list_result.isTruncated()) { + break; + } + list_result.setFileInfos({}); + list_result.setTruncated(false); + } + return Status::OK(); +} + +JindoInputStream::JindoInputStream(const std::shared_ptr& fs, + std::unique_ptr&& reader) + : fs_(fs), reader_(std::move(reader)) {} + +Status JindoInputStream::Seek(int64_t offset, SeekOrigin origin) { + if (origin == FS_SEEK_SET) { + PAIMON_RETURN_NOT_OK_FROM_JINDO(reader_->seek(offset)); + } else if (origin == FS_SEEK_CUR) { + PAIMON_ASSIGN_OR_RAISE(int64_t pos, GetPos()); + PAIMON_RETURN_NOT_OK_FROM_JINDO(reader_->seek(offset + pos)); + } else if (origin == FS_SEEK_END) { + PAIMON_ASSIGN_OR_RAISE(uint64_t len, Length()); + PAIMON_RETURN_NOT_OK_FROM_JINDO(reader_->seek(len + offset)); + } else { + return Status::Invalid("unsupported seek origin"); + } + return Status::OK(); +} + +Result JindoInputStream::GetPos() const { + int64_t pos = -1; + PAIMON_RETURN_NOT_OK_FROM_JINDO(reader_->tell(pos)); + return pos; +} + +Result JindoInputStream::Length() const { + int64_t len = -1; + PAIMON_RETURN_NOT_OK_FROM_JINDO(reader_->getFileLength(len)); + return len; +} + +Result JindoInputStream::Read(char* buffer, uint32_t size) { + PAIMON_RETURN_NOT_OK_FROM_JINDO(reader_->read(size, &result_, buffer)); + return result_.length(); +} + +Result JindoInputStream::Read(char* buffer, uint32_t size, uint64_t offset) { + PAIMON_RETURN_NOT_OK_FROM_JINDO(reader_->pread(offset, size, &result_, buffer)); + return result_.length(); +} + +void JindoInputStream::ReadAsync(char* buffer, uint32_t size, uint64_t offset, + std::function&& callback) { + auto outer_callback = [=](JdoStatus status) { + callback(status.ok() ? Status::OK() : Status::IOError(status.errMsg())); + }; + auto task = reader_->preadAsync(offset, size, &result_, buffer, outer_callback); + assert(task); + auto status = task->perform(); +} + +Status JindoInputStream::Close() { + PAIMON_RETURN_NOT_OK_FROM_JINDO(reader_->close()); + return Status::OK(); +} + +Result JindoInputStream::GetUri() const { + std::string_view uri; + PAIMON_RETURN_NOT_OK_FROM_JINDO(reader_->name(&uri)); + return std::string(uri); +} + +JindoOutputStream::JindoOutputStream(const std::shared_ptr& fs, + std::unique_ptr&& writer) + : fs_(fs), writer_(std::move(writer)) {} + +Result JindoOutputStream::GetPos() const { + int64_t pos = -1; + PAIMON_RETURN_NOT_OK_FROM_JINDO(writer_->tell(pos)); + return pos; +} + +Result JindoOutputStream::Write(const char* buffer, uint32_t size) { + std::string_view data(buffer, size); + PAIMON_RETURN_NOT_OK_FROM_JINDO(writer_->write(data)); + return size; +} + +Status JindoOutputStream::Flush() { + PAIMON_RETURN_NOT_OK_FROM_JINDO(writer_->flush()); + return Status::OK(); +} + +Status JindoOutputStream::Close() { + PAIMON_RETURN_NOT_OK_FROM_JINDO(writer_->close()); + return Status::OK(); +} + +Result JindoOutputStream::GetUri() const { + std::string_view name; + PAIMON_RETURN_NOT_OK_FROM_JINDO(writer_->name(&name)); + return std::string(name); +} + +} // namespace paimon::jindo diff --git a/src/paimon/fs/jindo/jindo_file_system.h b/src/paimon/fs/jindo/jindo_file_system.h new file mode 100644 index 00000000..ce5d1edf --- /dev/null +++ b/src/paimon/fs/jindo/jindo_file_system.h @@ -0,0 +1,106 @@ +/* + * 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. + */ + +#pragma once + +#include +#include +#include +#include +#include +#include + +#include "JdoFileSystem.hpp" // NOLINT(build/include_subdir) +#include "paimon/fs/file_system.h" +#include "paimon/result.h" +#include "paimon/status.h" + +namespace paimon::jindo { + +class JindoFileSystemImpl; + +class JindoFileSystem : public FileSystem { + public: + explicit JindoFileSystem(std::unique_ptr&& fs); + ~JindoFileSystem() override = default; + + Result> Open(const std::string& path) const override; + Result> Create(const std::string& path, + bool overwrite) const override; + Status Mkdirs(const std::string& path) const override; + // noted that oss file system does not support atomic rename, rename func will copy src to dst + // and remove src + Status Rename(const std::string& src, const std::string& dst) const override; + Status Delete(const std::string& path, bool recursive = true) const override; + + Result> GetFileStatus(const std::string& path) const override; + + Status ListDir(const std::string& directory, + std::vector>* file_status_list) const override; + + Status ListFileStatus( + const std::string& path, + std::vector>* file_status_list) const override; + + Result Exists(const std::string& path) const override; + + private: + std::shared_ptr impl_; +}; + +class JindoInputStream : public InputStream { + public: + JindoInputStream(const std::shared_ptr& fs, + std::unique_ptr&& reader); + Status Seek(int64_t offset, SeekOrigin origin) override; + Result GetPos() const override; + Result Read(char* buffer, uint32_t size) override; + Result Read(char* buffer, uint32_t size, uint64_t offset) override; + void ReadAsync(char* buffer, uint32_t size, uint64_t offset, + std::function&& callback) override; + Status Close() override; + Result GetUri() const override; + Result Length() const override; + + private: + // The lifecycle of the fs used to create the Jindo Reader must be longer than the lifecycle of + // the Jindo Reader. + std::shared_ptr fs_; + std::unique_ptr reader_; + std::string_view result_; +}; + +class JindoOutputStream : public OutputStream { + public: + JindoOutputStream(const std::shared_ptr& fs, + std::unique_ptr&& writer); + + Result GetPos() const override; + Result Write(const char* buffer, uint32_t size) override; + Status Flush() override; + Status Close() override; + Result GetUri() const override; + + private: + // The lifecycle of the fs used to create the Jindo Writer must be longer than the lifecycle of + // the Jindo Writer. + std::shared_ptr fs_; + std::unique_ptr writer_; +}; + +} // namespace paimon::jindo diff --git a/src/paimon/fs/jindo/jindo_file_system_factory.cpp b/src/paimon/fs/jindo/jindo_file_system_factory.cpp new file mode 100644 index 00000000..6a2d2bee --- /dev/null +++ b/src/paimon/fs/jindo/jindo_file_system_factory.cpp @@ -0,0 +1,59 @@ +/* + * 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 "paimon/fs/jindo/jindo_file_system_factory.h" + +#include + +#include "JdoConfig.hpp" // NOLINT(build/include_subdir) +#include "JdoFileSystem.hpp" // NOLINT(build/include_subdir) +#include "fmt/format.h" +#include "paimon/common/utils/path_util.h" +#include "paimon/factories/factory.h" +#include "paimon/fs/jindo/jindo_file_system.h" +#include "paimon/fs/jindo/jindo_utils.h" +#include "paimon/status.h" + +namespace paimon::jindo { + +const char JindoFileSystemFactory::IDENTIFIER[] = "jindo"; +const char JindoFileSystemFactory::URI_KEY[] = "fs.oss.uri"; +const char JindoFileSystemFactory::USER_KEY[] = "fs.oss.user"; + +Result> JindoFileSystemFactory::Create( + const std::string& path, const std::map& options) const { + auto user_iter = options.find(USER_KEY); + if (user_iter == options.end()) { + return Status::Invalid( + fmt::format("options must have '{}' key in JindoFileSystem", USER_KEY)); + } + std::shared_ptr config = std::make_shared(); + for (const auto& [k, v] : options) { + config->setString(k, v); + } + PAIMON_ASSIGN_OR_RAISE(Path new_path, PathUtil::ToPath(path)); + new_path.path = ""; // only use scheme and authority + config->setString(URI_KEY, new_path.ToString()); + std::unique_ptr fs = std::make_unique(); + PAIMON_RETURN_NOT_OK_FROM_JINDO(fs->init(path, user_iter->second, config)); + return std::make_unique(std::move(fs)); +} + +REGISTER_PAIMON_FACTORY(JindoFileSystemFactory); + +} // namespace paimon::jindo diff --git a/src/paimon/fs/jindo/jindo_file_system_factory.h b/src/paimon/fs/jindo/jindo_file_system_factory.h new file mode 100644 index 00000000..195043db --- /dev/null +++ b/src/paimon/fs/jindo/jindo_file_system_factory.h @@ -0,0 +1,43 @@ +/* + * 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. + */ + +#pragma once + +#include +#include +#include + +#include "paimon/fs/file_system.h" +#include "paimon/fs/file_system_factory.h" +#include "paimon/result.h" + +namespace paimon::jindo { +class JindoFileSystemFactory : public FileSystemFactory { + public: + static const char IDENTIFIER[]; + static const char URI_KEY[]; + static const char USER_KEY[]; + + const char* Identifier() const override { + return IDENTIFIER; + } + + Result> Create( + const std::string& path, const std::map& options) const override; +}; +} // namespace paimon::jindo diff --git a/src/paimon/fs/jindo/jindo_file_system_factory_test.cpp b/src/paimon/fs/jindo/jindo_file_system_factory_test.cpp new file mode 100644 index 00000000..94b3cbfd --- /dev/null +++ b/src/paimon/fs/jindo/jindo_file_system_factory_test.cpp @@ -0,0 +1,41 @@ +/* + * 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 "paimon/fs/jindo/jindo_file_system_factory.h" + +#include "gtest/gtest.h" +#include "paimon/testing/utils/testharness.h" +namespace paimon::jindo::test { +TEST(JindoFileSystemFactoryTest, TestCreate) { + auto fs_factory = std::make_unique(); + ASSERT_EQ(fs_factory->Identifier(), std::string("jindo")); + { + // invalid options + std::map options; + ASSERT_NOK_WITH_MSG(fs_factory->Create(paimon::test::GetJindoTestDir(), options), + "options must have 'fs.oss.user' key in JindoFileSystem"); + } + { + // options + std::map options = paimon::test::GetJindoTestOptions(); + ASSERT_OK_AND_ASSIGN(auto fs, fs_factory->Create(paimon::test::GetJindoTestDir(), options)); + ASSERT_TRUE(fs); + } +} + +} // namespace paimon::jindo::test diff --git a/src/paimon/fs/jindo/jindo_file_system_test.cpp b/src/paimon/fs/jindo/jindo_file_system_test.cpp new file mode 100644 index 00000000..39b8b577 --- /dev/null +++ b/src/paimon/fs/jindo/jindo_file_system_test.cpp @@ -0,0 +1,129 @@ +/* + * 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 "gtest/gtest.h" +#include "paimon/fs/jindo/jindo_file_system_factory.h" +#include "paimon/testing/utils/testharness.h" +namespace paimon::jindo::test { +// This test shows inconsistent behavior with the local file system in some abnormal scenarios. +class JindoFileSystemTest : public ::testing::Test { + public: + void SetUp() override { + dir_ = paimon::test::UniqueTestDirectory::Create("jindo"); + ASSERT_TRUE(dir_); + test_dir_ = dir_->Str() + "/"; + fs_ = dir_->GetFileSystem(); + } + + void TearDown() override { + dir_.reset(); + fs_.reset(); + } + + protected: + std::unique_ptr dir_; + std::string test_dir_; + std::shared_ptr fs_; +}; + +TEST_F(JindoFileSystemTest, TestLifeCycle) { + std::string file_path = test_dir_ + "file.data"; + std::string content = "content"; + std::map options = paimon::test::GetJindoTestOptions(); + auto fs_factory = std::make_shared(); + ASSERT_OK_AND_ASSIGN(auto tmp_fs, fs_factory->Create(file_path, options)); + ASSERT_OK(fs_->WriteFile(file_path, content, /*overwrite=*/false)); + + // read process + ASSERT_OK_AND_ASSIGN(auto in_stream, tmp_fs->Open(file_path)); + std::string read_content(content.size(), '\0'); + ASSERT_OK_AND_ASSIGN(int32_t read_len, + in_stream->Read(read_content.data(), read_content.size())); + ASSERT_EQ(read_len, read_content.size()); + ASSERT_EQ(content, read_content); + // The lifecycle of the fs used to create the Jindo Reader must be longer than the lifecycle of + // the Jindo Reader. + tmp_fs.reset(); + ASSERT_OK(in_stream->Close()); +} + +TEST_F(JindoFileSystemTest, TestRename) { + // test rename file to non-exist dir + std::string file_path = test_dir_ + "file5/file6/file7"; + ASSERT_OK(fs_->WriteFile(file_path, "content", /*overwrite=*/false)); + ASSERT_OK_AND_ASSIGN(bool is_exist, fs_->Exists(file_path)); + ASSERT_TRUE(is_exist); + std::string file_path2 = test_dir_ + "file8/file9"; + ASSERT_NOK_WITH_MSG(fs_->Rename(/*src=*/file_path, /*dst=*/file_path2), "file8 not found"); + ASSERT_OK_AND_ASSIGN(is_exist, fs_->Exists(file_path)); + ASSERT_TRUE(is_exist); +} + +TEST_F(JindoFileSystemTest, TestSeek) { + std::string content = "abcdefghijk"; + std::string file_path = test_dir_ + "file.data"; + // write process + ASSERT_OK_AND_ASSIGN(auto out_stream, fs_->Create(file_path, /*overwrite=*/true)); + ASSERT_OK_AND_ASSIGN(int32_t write_len, out_stream->Write(content.data(), content.size())); + ASSERT_EQ(write_len, content.size()); + ASSERT_OK(out_stream->Flush()); + ASSERT_OK(out_stream->Close()); + + // read process + ASSERT_OK_AND_ASSIGN(auto in_stream, fs_->Open(file_path)); + ASSERT_OK_AND_ASSIGN(auto pos, in_stream->GetPos()); + ASSERT_EQ(pos, 0); + + // invalid seek + ASSERT_NOK_WITH_MSG(in_stream->Seek(/*offset=*/20, SeekOrigin::FS_SEEK_SET), + "seek file failed: seek EOF"); + ASSERT_OK_AND_ASSIGN(pos, in_stream->GetPos()); + ASSERT_EQ(pos, 0); + // valid seek + ASSERT_OK(in_stream->Seek(/*offset=*/2, SeekOrigin::FS_SEEK_SET)); + ASSERT_OK_AND_ASSIGN(pos, in_stream->GetPos()); + ASSERT_EQ(pos, 2); + + ASSERT_OK(in_stream->Seek(/*offset=*/4, SeekOrigin::FS_SEEK_CUR)); + ASSERT_OK_AND_ASSIGN(pos, in_stream->GetPos()); + ASSERT_EQ(pos, 6); + + ASSERT_OK(in_stream->Seek(/*offset=*/-3, SeekOrigin::FS_SEEK_END)); + ASSERT_OK_AND_ASSIGN(pos, in_stream->GetPos()); + ASSERT_EQ(pos, 8); + + // read from cur pos + std::string read_content(3, '\0'); + ASSERT_OK_AND_ASSIGN(int32_t read_len, + in_stream->Read(read_content.data(), read_content.size())); + ASSERT_EQ(read_len, read_content.size()); + ASSERT_EQ("ijk", read_content); + + // read from offset + ASSERT_OK_AND_ASSIGN(read_len, + in_stream->Read(read_content.data(), read_content.size(), /*offset=*/ + 4)); + ASSERT_EQ(read_len, read_content.size()); + ASSERT_EQ("efg", read_content); + + ASSERT_OK_AND_ASSIGN(pos, in_stream->GetPos()); + ASSERT_EQ(pos, 11); + ASSERT_OK(in_stream->Close()); +} + +} // namespace paimon::jindo::test diff --git a/src/paimon/fs/jindo/jindo_utils.h b/src/paimon/fs/jindo/jindo_utils.h new file mode 100644 index 00000000..9dfcc9d4 --- /dev/null +++ b/src/paimon/fs/jindo/jindo_utils.h @@ -0,0 +1,33 @@ +/* + * 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. + */ + +#pragma once + +#include "JdoStatus.hpp" // NOLINT(build/include_subdir) +#include "paimon/status.h" + +namespace paimon::jindo { +#define PAIMON_RETURN_NOT_OK_FROM_JINDO(JINDO_STATUS) \ + do { \ + auto __s = (JINDO_STATUS); \ + if (PAIMON_UNLIKELY(!(__s).ok())) { \ + return Status::IOError(__s.errMsg()); \ + } \ + } while (false) + +} // namespace paimon::jindo diff --git a/third_party/jindosdk-nextarch/CMakeLists.txt b/third_party/jindosdk-nextarch/CMakeLists.txt new file mode 100644 index 00000000..1f921097 --- /dev/null +++ b/third_party/jindosdk-nextarch/CMakeLists.txt @@ -0,0 +1,46 @@ +# Copyright 2024-present Alibaba Cloud. +# +# Licensed 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. + +cmake_minimum_required(VERSION 3.17) + +project(jindosdk-nextarch) + +set(CMAKE_CXX_STANDARD 17) +set(CMAKE_CXX_STANDARD_REQUIRED ON) + +set(CMAKE_MODULE_PATH "${CMAKE_CURRENT_SOURCE_DIR}/cmake_modules") + +include_directories(${CMAKE_CURRENT_LIST_DIR}/include) + +find_package(jindosdk_c REQUIRED) + +file(GLOB_RECURSE JINDOSDK_SOURCES src/*.cpp) + +add_library(jindosdk-nextarch STATIC ${JINDOSDK_SOURCES}) +target_link_libraries(jindosdk-nextarch PUBLIC pthread dl JINDOSDK::JINDOSDK) +target_include_directories(jindosdk-nextarch + PUBLIC $ + $) + +install(TARGETS jindosdk-nextarch + LIBRARY DESTINATION ${CMAKE_INSTALL_LIBDIR} + ARCHIVE DESTINATION ${CMAKE_INSTALL_LIBDIR} + RUNTIME DESTINATION ${CMAKE_INSTALL_BINDIR} + INCLUDES + DESTINATION ${CMAKE_INSTALL_INCLUDEDIR}) + +install(DIRECTORY ${PROJECT_SOURCE_DIR}/include/ + DESTINATION "include" + FILES_MATCHING + PATTERN "*.hpp") diff --git a/third_party/jindosdk-nextarch/LICENSE.txt b/third_party/jindosdk-nextarch/LICENSE.txt new file mode 100644 index 00000000..d6456956 --- /dev/null +++ b/third_party/jindosdk-nextarch/LICENSE.txt @@ -0,0 +1,202 @@ + + Apache License + Version 2.0, January 2004 + http://www.apache.org/licenses/ + + TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION + + 1. Definitions. + + "License" shall mean the terms and conditions for use, reproduction, + and distribution as defined by Sections 1 through 9 of this document. + + "Licensor" shall mean the copyright owner or entity authorized by + the copyright owner that is granting the License. + + "Legal Entity" shall mean the union of the acting entity and all + other entities that control, are controlled by, or are under common + control with that entity. For the purposes of this definition, + "control" means (i) the power, direct or indirect, to cause the + direction or management of such entity, whether by contract or + otherwise, or (ii) ownership of fifty percent (50%) or more of the + outstanding shares, or (iii) beneficial ownership of such entity. + + "You" (or "Your") shall mean an individual or Legal Entity + exercising permissions granted by this License. + + "Source" form shall mean the preferred form for making modifications, + including but not limited to software source code, documentation + source, and configuration files. + + "Object" form shall mean any form resulting from mechanical + transformation or translation of a Source form, including but + not limited to compiled object code, generated documentation, + and conversions to other media types. + + "Work" shall mean the work of authorship, whether in Source or + Object form, made available under the License, as indicated by a + copyright notice that is included in or attached to the work + (an example is provided in the Appendix below). + + "Derivative Works" shall mean any work, whether in Source or Object + form, that is based on (or derived from) the Work and for which the + editorial revisions, annotations, elaborations, or other modifications + represent, as a whole, an original work of authorship. For the purposes + of this License, Derivative Works shall not include works that remain + separable from, or merely link (or bind by name) to the interfaces of, + the Work and Derivative Works thereof. + + "Contribution" shall mean any work of authorship, including + the original version of the Work and any modifications or additions + to that Work or Derivative Works thereof, that is intentionally + submitted to Licensor for inclusion in the Work by the copyright owner + or by an individual or Legal Entity authorized to submit on behalf of + the copyright owner. For the purposes of this definition, "submitted" + means any form of electronic, verbal, or written communication sent + to the Licensor or its representatives, including but not limited to + communication on electronic mailing lists, source code control systems, + and issue tracking systems that are managed by, or on behalf of, the + Licensor for the purpose of discussing and improving the Work, but + excluding communication that is conspicuously marked or otherwise + designated in writing by the copyright owner as "Not a Contribution." + + "Contributor" shall mean Licensor and any individual or Legal Entity + on behalf of whom a Contribution has been received by Licensor and + subsequently incorporated within the Work. + + 2. Grant of Copyright License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + copyright license to reproduce, prepare Derivative Works of, + publicly display, publicly perform, sublicense, and distribute the + Work and such Derivative Works in Source or Object form. + + 3. Grant of Patent License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + (except as stated in this section) patent license to make, have made, + use, offer to sell, sell, import, and otherwise transfer the Work, + where such license applies only to those patent claims licensable + by such Contributor that are necessarily infringed by their + Contribution(s) alone or by combination of their Contribution(s) + with the Work to which such Contribution(s) was submitted. If You + institute patent litigation against any entity (including a + cross-claim or counterclaim in a lawsuit) alleging that the Work + or a Contribution incorporated within the Work constitutes direct + or contributory patent infringement, then any patent licenses + granted to You under this License for that Work shall terminate + as of the date such litigation is filed. + + 4. Redistribution. You may reproduce and distribute copies of the + Work or Derivative Works thereof in any medium, with or without + modifications, and in Source or Object form, provided that You + meet the following conditions: + + (a) You must give any other recipients of the Work or + Derivative Works a copy of this License; and + + (b) You must cause any modified files to carry prominent notices + stating that You changed the files; and + + (c) You must retain, in the Source form of any Derivative Works + that You distribute, all copyright, patent, trademark, and + attribution notices from the Source form of the Work, + excluding those notices that do not pertain to any part of + the Derivative Works; and + + (d) If the Work includes a "NOTICE" text file as part of its + distribution, then any Derivative Works that You distribute must + include a readable copy of the attribution notices contained + within such NOTICE file, excluding those notices that do not + pertain to any part of the Derivative Works, in at least one + of the following places: within a NOTICE text file distributed + as part of the Derivative Works; within the Source form or + documentation, if provided along with the Derivative Works; or, + within a display generated by the Derivative Works, if and + wherever such third-party notices normally appear. The contents + of the NOTICE file are for informational purposes only and + do not modify the License. You may add Your own attribution + notices within Derivative Works that You distribute, alongside + or as an addendum to the NOTICE text from the Work, provided + that such additional attribution notices cannot be construed + as modifying the License. + + You may add Your own copyright statement to Your modifications and + may provide additional or different license terms and conditions + for use, reproduction, or distribution of Your modifications, or + for any such Derivative Works as a whole, provided Your use, + reproduction, and distribution of the Work otherwise complies with + the conditions stated in this License. + + 5. Submission of Contributions. Unless You explicitly state otherwise, + any Contribution intentionally submitted for inclusion in the Work + by You to the Licensor shall be under the terms and conditions of + this License, without any additional terms or conditions. + Notwithstanding the above, nothing herein shall supersede or modify + the terms of any separate license agreement you may have executed + with Licensor regarding such Contributions. + + 6. Trademarks. This License does not grant permission to use the trade + names, trademarks, service marks, or product names of the Licensor, + except as required for reasonable and customary use in describing the + origin of the Work and reproducing the content of the NOTICE file. + + 7. Disclaimer of Warranty. Unless required by applicable law or + agreed to in writing, Licensor provides the Work (and each + Contributor provides its Contributions) on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + implied, including, without limitation, any warranties or conditions + of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A + PARTICULAR PURPOSE. You are solely responsible for determining the + appropriateness of using or redistributing the Work and assume any + risks associated with Your exercise of permissions under this License. + + 8. Limitation of Liability. In no event and under no legal theory, + whether in tort (including negligence), contract, or otherwise, + unless required by applicable law (such as deliberate and grossly + negligent acts) or agreed to in writing, shall any Contributor be + liable to You for damages, including any direct, indirect, special, + incidental, or consequential damages of any character arising as a + result of this License or out of the use or inability to use the + Work (including but not limited to damages for loss of goodwill, + work stoppage, computer failure or malfunction, or any and all + other commercial damages or losses), even if such Contributor + has been advised of the possibility of such damages. + + 9. Accepting Warranty or Additional Liability. While redistributing + the Work or Derivative Works thereof, You may choose to offer, + and charge a fee for, acceptance of support, warranty, indemnity, + or other liability obligations and/or rights consistent with this + License. However, in accepting such obligations, You may act only + on Your own behalf and on Your sole responsibility, not on behalf + of any other Contributor, and only if You agree to indemnify, + defend, and hold each Contributor harmless for any liability + incurred by, or claims asserted against, such Contributor by reason + of your accepting any such warranty or additional liability. + + END OF TERMS AND CONDITIONS + + APPENDIX: How to apply the Apache License to your work. + + To apply the Apache License to your work, attach the following + boilerplate notice, with the fields enclosed by brackets "[]" + replaced with your own identifying information. (Don't include + the brackets!) The text should be enclosed in the appropriate + comment syntax for the file format. We also recommend that a + file or class name and description of purpose be included on the + same "printed page" as the copyright notice for easier + identification within third-party archives. + + Copyright [yyyy] [name of copyright owner] + + Licensed 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. diff --git a/third_party/jindosdk-nextarch/NOTICE.txt b/third_party/jindosdk-nextarch/NOTICE.txt new file mode 100644 index 00000000..a5518690 --- /dev/null +++ b/third_party/jindosdk-nextarch/NOTICE.txt @@ -0,0 +1,2 @@ +JindoSDK NextArch C++ +Copyright 2024-present Alibaba Cloud. diff --git a/third_party/jindosdk-nextarch/README.md b/third_party/jindosdk-nextarch/README.md new file mode 100644 index 00000000..9f605732 --- /dev/null +++ b/third_party/jindosdk-nextarch/README.md @@ -0,0 +1,31 @@ +# JindoSDK NextArch C++ + +The `jindosdk-nextarch` C++ interface is derived from an internal repository maintained by Alibaba Jindo Team, which also develops and contributes to the open-source project [alibabacloud-jindodata](https://github.com/aliyun/alibabacloud-jindodata/). + +## License + +[Apache-2.0](http://www.apache.org/licenses/LICENSE-2.0) + +Copyright 2024-present Alibaba Cloud. + +## Usage + +### requirement + +g++: support c++17 + +cmake: > 3.17.0 + +jindo_c_sdk: You can download from [alibabacloud-jindodata](https://github.com/aliyun/alibabacloud-jindodata/) + +### build + +```bash +# use your compiler +export CC=${CC:-"/usr/local/bin/gcc"} +export CXX=${CXX:-"/usr/local/bin/g++"} +mkdir -p build +cd build +cmake .. -DJINDOSDK_ROOT= +make -j3 +``` diff --git a/third_party/jindosdk-nextarch/cmake_modules/Findjindosdk_c.cmake b/third_party/jindosdk-nextarch/cmake_modules/Findjindosdk_c.cmake new file mode 100644 index 00000000..c9d5cb49 --- /dev/null +++ b/third_party/jindosdk-nextarch/cmake_modules/Findjindosdk_c.cmake @@ -0,0 +1,76 @@ +# Copyright 2024-present Alibaba Inc. +# +# Licensed 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. + +# JINDOSDK_ROOT environmental variable is used to check for JINDOSDK headers and dynamic library + +# JINDOSDK_INCLUDE_DIR: directory containing headers +# JINDOSDK_LIBRARY: path to libjindosdk_c.so +# JINDOSDK_FOUND: whether JINDOSDK has been found + +if(NOT JINDOSDK_ROOT) + if(DEFINED ENV{JINDOSDK_ROOT}) + set(JINDOSDK_ROOT "$ENV{JINDOSDK_ROOT}") + endif() +endif() + +if(NOT "${JINDOSDK_ROOT}" STREQUAL "") + file(TO_CMAKE_PATH "${JINDOSDK_ROOT}" _jindosdk_path) +endif() + +message(STATUS "JINDOSDK_ROOT: ${JINDOSDK_ROOT}") + +find_path(JINDOSDK_INCLUDE_DIR jdo_api.h + HINTS ${_jindosdk_path} + NO_DEFAULT_PATH + PATH_SUFFIXES "include") + +find_library(JINDOSDK_LIBRARY + NAMES ${JINDOSDK_LIBRARY_NAME} + HINTS ${_jindosdk_path} + PATH_SUFFIXES "lib/native") + +if(JINDOSDK_INCLUDE_DIR AND JINDOSDK_LIBRARY) + set(JINDOSDK_FOUND TRUE) + set(JINDOSDK_HEADER_NAME jdo_api.h) + set(JINDOSDK_HEADER ${JINDOSDK_INCLUDE_DIR}/${JINDOSDK_HEADER_NAME}) +else() + set(JINDOSDK_FOUND FALSE) +endif() + +if(JINDOSDK_FOUND) + message(STATUS "Found the JINDOSDK header: ${JINDOSDK_HEADER}") + message(STATUS "Found the JINDOSDK library: ${JINDOSDK_LIBRARY}") +else() + if(_jindosdk_path) + set(JINDOSDK_ERR_MSG "Could not find JINDOSDK. Looked in ${_jindosdk_path}.") + else() + set(JINDOSDK_ERR_MSG "Could not find JINDOSDK in system search paths.") + endif() + + if(JINDOSDK_FIND_REQUIRED) + message(FATAL_ERROR "${JINDOSDK_ERR_MSG}") + else() + message(STATUS "${JINDOSDK_ERR_MSG}") + endif() +endif() + +mark_as_advanced(JINDOSDK_INCLUDE_DIR JINDOSDK_LIBRARY) + +if(JINDOSDK_FOUND AND NOT TARGET JINDOSDK::JINDOSDK) + add_library(JINDOSDK::JINDOSDK UNKNOWN IMPORTED) + set_target_properties(JINDOSDK::JINDOSDK + PROPERTIES IMPORTED_LOCATION "${JINDOSDK_LIBRARY}" + INTERFACE_INCLUDE_DIRECTORIES + "${JINDOSDK_INCLUDE_DIR}") +endif() diff --git a/third_party/jindosdk-nextarch/include/JdoConfig.hpp b/third_party/jindosdk-nextarch/include/JdoConfig.hpp new file mode 100644 index 00000000..61c131eb --- /dev/null +++ b/third_party/jindosdk-nextarch/include/JdoConfig.hpp @@ -0,0 +1,44 @@ +/* + * Copyright 2024-present Alibaba Cloud. + * + * Licensed 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. + */ + +#pragma once + +#include +#include +#include "jdo_common.h" + +class JDO_EXPORT JdoConfig { +protected: + std::map _options; + +public: + JdoConfig(); + JdoConfig(std::map& options); + + std::string getString(const std::string& key, const std::string& def); + void setString(const std::string& key, const std::string& val); + int64_t getInt64(const std::string& key, int64_t def); + void setInt64(const std::string& key, int64_t val); + int32_t getInt32(const std::string& key, int32_t def); + void setInt32(const std::string& key, int32_t val); + double getDouble(const std::string& key, double def); + void setDouble(const std::string& key, double val); + bool getBool(const std::string& key, bool def); + void setBool(const std::string& key, bool val); + + bool contains(const std::string& key); + std::map getAll(); +}; diff --git a/third_party/jindosdk-nextarch/include/JdoContentSummary.hpp b/third_party/jindosdk-nextarch/include/JdoContentSummary.hpp new file mode 100644 index 00000000..dc33ea86 --- /dev/null +++ b/third_party/jindosdk-nextarch/include/JdoContentSummary.hpp @@ -0,0 +1,50 @@ +/* + * Copyright 2024-present Alibaba Cloud. + * + * Licensed 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. + */ + +#pragma once + +#include "jdo_common.h" + +class JDO_EXPORT JdoContentSummary { +public: + int64_t fileSize = 0; + int64_t fileCount = 0; + int64_t dirCount = 0; + + void setFileSize(int64_t fileSize) { + this->fileSize = fileSize; + } + + int64_t getFileSize() const { + return fileSize; + } + + void setFileCount(int64_t fileCount) { + this->fileCount = fileCount; + } + + int64_t getFileCount() const { + return fileCount; + } + + void setDirCount(int64_t dirCount) { + this->dirCount = dirCount; + } + + int64_t getDirCount() const { + return dirCount; + } +}; diff --git a/third_party/jindosdk-nextarch/include/JdoFileInfo.hpp b/third_party/jindosdk-nextarch/include/JdoFileInfo.hpp new file mode 100644 index 00000000..6207efa8 --- /dev/null +++ b/third_party/jindosdk-nextarch/include/JdoFileInfo.hpp @@ -0,0 +1,101 @@ +/* + * Copyright 2024-present Alibaba Cloud. + * + * Licensed 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. + */ + +#pragma once + +#include +#include "jdo_defines.h" +#include "jdo_common.h" + +class JDO_EXPORT JdoFileInfo { +public: + std::string path; + std::string user; + std::string group; + int8_t type = JDO_FILE_TYPE_UNKNOWN; + int16_t perm = 0755; + int64_t length = -1; + int64_t mtime = 0; + int64_t atime = 0; + + void setPath(const std::string& path) { + this->path = path; + } + + std::string getPath() const { + return path; + } + + void setUser(const std::string& user) { + this->user = user; + } + + std::string getUser() const { + return user; + } + + void setGroup(const std::string& group) { + this->group = group; + } + + std::string getGroup() const { + return group; + } + + void setType(int8_t type) { + this->type = type; + } + + int8_t getType() const { + return type; + } + + bool isDir() const { + return type == JDO_FILE_TYPE_DIRECTORY; + } + + void setPerm(int16_t perm) { + this->perm = perm; + } + + int16_t getPerm() const { + return perm; + } + + void setLength(int64_t length) { + this->length = length; + } + + int64_t getLength() const { + return length; + } + + void setMtime(int64_t mtime) { + this->mtime = mtime; + } + + int64_t getMtime() const { + return mtime; + } + + void setAtime(int64_t atime) { + this->atime = atime; + } + + int64_t getAtime() const { + return atime; + } +}; diff --git a/third_party/jindosdk-nextarch/include/JdoFileSystem.hpp b/third_party/jindosdk-nextarch/include/JdoFileSystem.hpp new file mode 100644 index 00000000..f23790c5 --- /dev/null +++ b/third_party/jindosdk-nextarch/include/JdoFileSystem.hpp @@ -0,0 +1,190 @@ +/* + * Copyright 2024-present Alibaba Cloud. + * + * Licensed 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. + */ + +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "jdo_common.h" +#include "JdoConfig.hpp" +#include "JdoStatus.hpp" + +class JdoFileInfo; +class JdoListResult; +class JdoContentSummary; +using JdoFileInfoPtr = std::shared_ptr; +using JdoListResultPtr = std::shared_ptr; +using JdoContentSummaryPtr = std::shared_ptr; + +class JDO_EXPORT JdoTask { +public: + JdoTask() = default; + + virtual ~JdoTask() = default; + + virtual JdoStatus perform() = 0; + + virtual JdoStatus wait() = 0; + + virtual JdoStatus cancel() = 0; +}; + +typedef std::shared_ptr JdoTaskPtr; + +class JDO_EXPORT JdoReader { +public: + JdoReader() = default; + + virtual ~JdoReader() = default; + + virtual bool closed() const = 0; + + virtual JdoStatus name(std::string_view* result) const = 0; + + virtual JdoStatus read(size_t n, std::string_view* result, char* scratch) = 0; + + virtual JdoTaskPtr readAsync(size_t n, std::string_view* result, char* scratch, + std::function callback) = 0; + + virtual JdoStatus pread(int64_t offset, size_t n, std::string_view* result, char* scratch) = 0; + + virtual JdoTaskPtr preadAsync(int64_t offset, size_t n, std::string_view* result, char* scratch, + std::function callback) = 0; + + virtual JdoStatus preadv(std::vector& offsets, std::vector& lengths, std::vector& scratches) = 0; + + virtual JdoTaskPtr preadvAsync(std::vector& offsets, std::vector& lengths, std::vector& scratches, + std::function callback) = 0; + + virtual JdoStatus tell(int64_t &offset) = 0; + + virtual JdoTaskPtr tellAsync(std::function callback) = 0; + + virtual JdoStatus seek(int64_t offset) = 0; + + virtual JdoTaskPtr seekAsync(int64_t offset, std::function callback) = 0; + + virtual JdoStatus getFileLength(int64_t &length) = 0; + + virtual JdoTaskPtr getFileLengthAsync(std::function callback) = 0; + + virtual JdoStatus close() = 0; + + virtual JdoTaskPtr closeAsync(std::function callback) = 0; +}; + +class JDO_EXPORT JdoWriter { +public: + JdoWriter() = default; + + virtual ~JdoWriter() = default; + + virtual bool closed() const = 0; + + virtual JdoStatus name(std::string_view* result) const = 0; + + virtual JdoStatus write(std::string_view data) = 0; + + virtual JdoTaskPtr writeAsync(std::string_view data, std::function callback) = 0; + + virtual JdoStatus flush() = 0; + + virtual JdoTaskPtr flushAsync(std::function callback) = 0; + + virtual JdoStatus tell(int64_t &offset) = 0; + + virtual JdoTaskPtr tellAsync(std::function callback) = 0; + + virtual JdoStatus close() = 0; + + virtual JdoTaskPtr closeAsync(std::function callback) = 0; +}; + + +class JDO_EXPORT JdoFileSystem { +public: + JdoFileSystem(); + + virtual ~JdoFileSystem(); + + JdoStatus init(const std::string &uri, const std::string &user, const std::shared_ptr &config); + + JdoStatus destroy(); + + JdoStatus openReader(const std::string &path, std::unique_ptr* result); + + JdoStatus openWriter(const std::string &path, std::unique_ptr* result); + + JdoStatus mkdir(const std::string &path, bool recursive); + + JdoTaskPtr mkdirAsync(const std::string &path, bool recursive, std::function callback); + + JdoStatus rename(const std::string &oldpath, const std::string &newpath); + + JdoTaskPtr renameAsync(const std::string &oldpath, const std::string &newpath, + std::function callback); + + JdoStatus remove(const std::string &path, bool recursive); + + JdoTaskPtr removeAsync(const std::string &path, bool recursive, std::function callback); + + JdoStatus exists(const std::string &path); + + JdoTaskPtr existsAsync(const std::string &path, std::function callback); + + JdoStatus getFileInfo(const std::string &path, JdoFileInfo* info); + + JdoTaskPtr getFileInfoAsync(const std::string &path, std::function callback); + + JdoStatus listDir(const std::string &path, bool recursive, JdoListResult* result); + + JdoTaskPtr listDirAsync(const std::string &path, bool recursive, + std::function callback); + + JdoStatus getContentSummary(const std::string &path, bool recursive, JdoContentSummary* result); + + JdoTaskPtr getContentSummaryAsync(const std::string &path, bool recursive, + std::function callback); + + JdoStatus setPermission(const std::string &path, int16_t perm); + + JdoTaskPtr setPermissionAsync(const std::string &path, int16_t perm, std::function callback); + + JdoStatus setOwner(const std::string &path, const std::string& user, const std::string& group); + + JdoTaskPtr setOwnerAsync(const std::string &path, const std::string& user, const std::string& group, std::function callback); + +private: + JdoStatus getFileInfoInternal(JdoStore_t client, + const std::string &path, + JdoFileInfo* info); + + // return JdoStore_t, status + std::tuple GetJdoStore(const std::string &uri); + +private: + JdoStore_t _store = nullptr; + JdoOptions_t _options = nullptr; + bool _inited = false; +}; diff --git a/third_party/jindosdk-nextarch/include/JdoListResult.hpp b/third_party/jindosdk-nextarch/include/JdoListResult.hpp new file mode 100644 index 00000000..9671d71e --- /dev/null +++ b/third_party/jindosdk-nextarch/include/JdoListResult.hpp @@ -0,0 +1,51 @@ +/* + * Copyright 2024-present Alibaba Cloud. + * + * Licensed 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. + */ + +#pragma once + +#include "JdoFileInfo.hpp" +#include "jdo_common.h" + +class JDO_EXPORT JdoListResult { +public: + std::vector infos; + bool truncated = false; + std::string nextMarker; + + void setFileInfos(const std::vector& infos) { + this->infos = infos; + } + + std::vector getFileInfos() const { + return infos; + } + + void setTruncated(bool truncated) { + this->truncated = truncated; + } + + bool isTruncated() const { + return truncated; + } + + void setNextMarker(const std::string& nextMarker) { + this->nextMarker = nextMarker; + } + + std::string getNextMarker() const { + return nextMarker; + } +}; diff --git a/third_party/jindosdk-nextarch/include/JdoStatus.hpp b/third_party/jindosdk-nextarch/include/JdoStatus.hpp new file mode 100644 index 00000000..b4d273f1 --- /dev/null +++ b/third_party/jindosdk-nextarch/include/JdoStatus.hpp @@ -0,0 +1,149 @@ +/* + * Copyright 2024-present Alibaba Cloud. + * + * Licensed 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. + */ + +#pragma once + +#include +#include +#include "jdo_error.h" +#include "jdo_common.h" + +#ifdef __has_builtin +#define JDO_HAS_BUILTIN(x) __has_builtin(x) +#else +#define JDO_HAS_BUILTIN(x) 0 +#endif + +#if (!defined(__NVCC__)) && \ + (JDO_HAS_BUILTIN(__builtin_expect) || (defined(__GNUC__) && __GNUC__ >= 3)) +#define JDO_PREDICT_FALSE(x) (__builtin_expect(x, 0)) +#define JDO_PREDICT_TRUE(x) (__builtin_expect(!!(x), 1)) +#else +#define JDO_PREDICT_FALSE(x) (x) +#define JDO_PREDICT_TRUE(x) (x) +#endif + +#define JDO_RETURN_IF_ERROR(...) \ + do { \ + JdoStatus _status = (__VA_ARGS__); \ + if (JDO_PREDICT_FALSE(!_status.ok())) return _status; \ + } while (0) + +class JDO_EXPORT JdoStatus { +public: + uint64_t _tid = 0; + int _errCode = 0; + std::string _errMsg; + + JdoStatus() { + } + + virtual ~JdoStatus() { + } + + JdoStatus(int errCode, std::string errMsg) { + _errCode = errCode; + _errMsg = std::move(errMsg); + } + + JdoStatus(int errCode, const char* errMsg) { + _errCode = errCode; + _errMsg = std::string(errMsg); + } + + inline bool hasErr() { + return _errCode != 0; + } + + inline void clear() { + _errCode = 0; + _errMsg.clear(); + } + + inline void setErrCode(int errCode) { + _errCode = errCode; + } + + inline int code() { + return _errCode; + } + + inline int getErrCode() { + return _errCode; + } + + inline void setErrMsg(std::string errMsg) { + _errMsg = std::move(errMsg); + } + + inline std::string errMsg() { + return _errMsg; + } + + inline std::string getErrMsg() { + return _errMsg; + } + + inline bool ok() { + return _errCode == 0; + } + + inline bool isOk() { + return _errCode == 0; + } + + friend std::ostream& operator<<(std::ostream &os, const JdoStatus &status) { + os << "code: " << status._errCode << " msg: " << status._errMsg; + return os; + } + + static JdoStatus OK() { + return JdoStatus(); + } + + template + static JdoStatus InitError(Args...args) { + return JdoStatus(JDO_CLIENT_ERROR, concat("init failed ", args...)); + } + + template + static JdoStatus InvalidArgument(Args...args) { + return JdoStatus(JDO_CLIENT_ERROR, concat("invalid argument ", args...)); + } + + template + static JdoStatus NotFound(Args...args) { + return JdoStatus(JDO_FILE_NOT_FOUND_ERROR, concat(args...)); + } + + template + static JdoStatus IOError(Args...args) { + return JdoStatus(JDO_IO_ERROR, concat(args...)); + } + + template + static JdoStatus InternalError(int32_t code, Args...args) { + return JdoStatus(code, concat(args...)); + } + +private: + template + static std::string concat(Args...args) { + std::stringstream os; + (os << ... << args); + return os.str(); + } +}; diff --git a/third_party/jindosdk-nextarch/src/JdoConfig.cpp b/third_party/jindosdk-nextarch/src/JdoConfig.cpp new file mode 100644 index 00000000..72100c67 --- /dev/null +++ b/third_party/jindosdk-nextarch/src/JdoConfig.cpp @@ -0,0 +1,209 @@ +/* + * Copyright 2024-present Alibaba Cloud. + * + * Licensed 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 "JdoConfig.hpp" +#include +#include +#include +#include +#include + +static int32_t StrToInt32(const char* str, int32_t def) { + int64_t retval; + char* end = NULL; + errno = 0; + retval = strtol(str, &end, 0); + + if (EINVAL == errno || 0 != *end) { + std::stringstream ss; + std::cerr << "Invalid int32_t type: " << str << std::endl; + return def; + } + + if (ERANGE == errno || retval > std::numeric_limits::max() || + retval < std::numeric_limits::min()) { + std::cerr << "Underflow/Overflow int32_t type: " << str << std::endl; + return def; + } + + return retval; +} + +static int64_t StrToInt64(const char* str, int64_t def) { + int64_t retval; + char* end = NULL; + errno = 0; + retval = strtoll(str, &end, 0); + + if (EINVAL == errno || 0 != *end) { + std::cerr << "Invalid int64_t type: " << str << std::endl; + return def; + } + + if (ERANGE == errno || retval > std::numeric_limits::max() || + retval < std::numeric_limits::min()) { + std::cerr << "Underflow/Overflow int64_t type: " << str << std::endl; + return def; + } + + return retval; +} + +static double StrToDouble(const char* str, int64_t def) { + double retval; + char* end = NULL; + errno = 0; + retval = strtod(str, &end); + + if (EINVAL == errno || 0 != *end) { + std::cerr << "Invalid double type: " << str; + return def; + } + + if (ERANGE == errno || retval > std::numeric_limits::max() || retval < std::numeric_limits::min()) { + std::cerr << "Underflow/Overflow int64_t type: " << str << std::endl; + return def; + } + + return retval; +} + +static bool StrToBool(const char* str, bool def) { + bool retval = false; + + if (!strcasecmp(str, "true") || !strcmp(str, "1")) { + retval = true; + } else if (!strcasecmp(str, "false") || !strcmp(str, "0")) { + retval = false; + } else { + std::cerr << "Invalid bool type: " << str << std::endl; + return def; + } + + return retval; +} + +static double StrToDouble(const char* str, double def) { + double retval; + char* end = NULL; + errno = 0; + retval = strtod(str, &end); + + if (EINVAL == errno || 0 != *end) { + std::cerr << "Invalid double type: " << str << std::endl; + return def; + } + + if (ERANGE == errno || retval > std::numeric_limits::max() || retval < std::numeric_limits::min()) { + std::cerr << "Underflow/Overflow int64_t type: " << str << std::endl; + return def; + } + + return retval; +} + +JdoConfig::JdoConfig() { + +} + +JdoConfig::JdoConfig(std::map& options) { + _options = options; +} + +std::string JdoConfig::getString(const std::string& key, const std::string& def) { + auto it = _options.find(key); + + if (_options.end() == it) { + return def; + } else { + return it->second; + } +} + +void JdoConfig::setString(const std::string& key, const std::string& val) { + _options[key] = val; +} + +int64_t JdoConfig::getInt64(const std::string& key, int64_t def) { + int64_t retval; + auto it = _options.find(key); + + if (_options.end() == it) { + return def; + } + + return StrToInt64(it->second.c_str(), def); +} + +void JdoConfig::setInt64(const std::string& key, int64_t val) { + _options[key] = std::to_string(val); +} + +int32_t JdoConfig::getInt32(const std::string& key, int32_t def) { + int32_t retval; + auto it = _options.find(key); + + if (_options.end() == it) { + return def; + } + + return StrToInt32(it->second.c_str(), def); +} + +void JdoConfig::setInt32(const std::string& key, int32_t val) { + _options[key] = std::to_string(val); +} + +double JdoConfig::getDouble(const std::string& key, double def) { + double retval; + auto it = _options.find(key); + + if (_options.end() == it) { + return def; + } + + return StrToDouble(it->second.c_str(), def); +} + +void JdoConfig::setDouble(const std::string& key, double val) { + _options[key] = std::to_string(val); +} + +bool JdoConfig::getBool(const std::string& key, bool def) { + bool retval; + auto it = _options.find(key); + + if (_options.end() == it) { + return def; + } + + return StrToBool(it->second.c_str(), def); +} + +void JdoConfig::setBool(const std::string& key, bool val) { + _options[key] = std::to_string(val); +} + +bool JdoConfig::contains(const std::string& key) { + if (_options.find(key) == _options.end()) { + return false; + } + return true; +} + +std::map JdoConfig::getAll() { + return _options; +} diff --git a/third_party/jindosdk-nextarch/src/JdoFileSystem.cpp b/third_party/jindosdk-nextarch/src/JdoFileSystem.cpp new file mode 100644 index 00000000..9a4607fd --- /dev/null +++ b/third_party/jindosdk-nextarch/src/JdoFileSystem.cpp @@ -0,0 +1,1838 @@ +/* + * Copyright 2024-present Alibaba Cloud. + * + * Licensed 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 "JdoFileSystem.hpp" + +#include +#include +#include +#include +#include +#include +#include + +#include "JdoStatus.hpp" +#include "JdoFileInfo.hpp" +#include "JdoListResult.hpp" +#include "JdoContentSummary.hpp" + +#include "jdo_common.h" +#include "jdo_api.h" +#include "jdo_defines.h" +#include "jdo_error.h" +#include "jdo_options.h" +#include "jdo_file_status.h" +#include "jdo_list_dir_result.h" +#include "jdo_content_summary.h" +#include "jdo_file_buffers.h" +#include "jdo_longs.h" + +#define START_CALL(fs) \ + JdoHandleCtx_t ctx = jdo_createHandleCtx1(fs); + +#define START_CALL2(fs, stream) \ + JdoHandleCtx_t ctx = jdo_createHandleCtx2(fs, stream); + +#define END_CALL() \ + auto errorCode = jdo_getHandleCtxErrorCode(ctx); \ + std::string errorMsg; \ + const char* msg = jdo_getHandleCtxErrorMsg(ctx); \ + if (msg != nullptr) { \ + errorMsg.assign(msg); \ + } \ + jdo_freeHandleCtx(ctx); + +static void covertToFileInfo(JdoFileStatus_t fileStatus, + JdoFileInfo* info) { + if (!fileStatus) return; + auto path = jdo_getFileStatusPath(fileStatus); + if (path) { + info->path = path; + } + auto user = jdo_getFileStatusUser(fileStatus); + if (user) { + info->user = user; + } + auto group = jdo_getFileStatusGroup(fileStatus); + if (group) { + info->group = group; + } + info->type = jdo_getFileStatusType(fileStatus); + info->perm = jdo_getFileStatusPerm(fileStatus); + info->length = jdo_getFileStatusSize(fileStatus); + info->mtime = jdo_getFileStatusMtime(fileStatus); + info->atime = jdo_getFileStatusAtime(fileStatus); + if (info->isDir() && info->path.back() != '/') { + info->path.push_back('/'); + } +} + +class JindoOperation : public JdoTask { +public: + JindoOperation(JdoStore_t store, JdoOperationCall_t call) + : JdoTask(), _store(store), _stream(nullptr), _call(call) {} + + JindoOperation(JdoStore_t store, JdoIOContext_t stream, JdoOperationCall_t call) + : JdoTask(), _store(store), _stream(stream), _call(call) {} + + ~JindoOperation() override { + if (_call) { + jdo_freeOperationCall(_call); + _call = nullptr; + } + } + + JdoStatus perform() override { + START_CALL2(_store, _stream) + jdo_perform(ctx, _call); + END_CALL(); + if (errorCode != 0) { + return JdoStatus::InternalError(errorCode, "failed to perform, errmsg: ", errorMsg); + } + return JdoStatus::OK(); + } + + JdoStatus wait() override { + START_CALL2(_store, _stream) + jdo_wait(ctx, _call); + END_CALL(); + if (errorCode != 0) { + return JdoStatus::InternalError(errorCode, "failed to wait, errmsg: ", errorMsg); + } + return JdoStatus::OK(); + } + + JdoStatus cancel() override { + START_CALL2(_store, _stream) + jdo_cancel(ctx, _call); + END_CALL(); + if (errorCode != 0) { + return JdoStatus::InternalError(errorCode, "failed to cancel, errmsg: ", errorMsg); + } + return JdoStatus::OK(); + } + +private: + JdoStore_t _store = nullptr; + JdoIOContext_t _stream = nullptr; + JdoOperationCall_t _call = nullptr; +}; + +struct ReadArgs { + JdoStore_t store = nullptr; + JdoIOContext_t reader = nullptr; + std::string name; + size_t n = 0; + std::string_view *result; + char *scratch; + std::function callback; +}; + +static void readCallback(JdoHandleCtx_t ctx, int64_t num_read, void *userdata) { + ReadArgs *args = (ReadArgs *) userdata; + auto store = args->store; + auto reader = args->reader; + auto name = args->name; + auto n = args->n; + auto result = args->result; + auto scratch = args->scratch; + auto total_read = num_read; + std::function callback = args->callback; + delete args; + + END_CALL() + if (errorCode != 0) { + auto status = JdoStatus::InternalError(errorCode, "failed to read ", name, + " size ", n - total_read, " readed ", num_read, + ", errmsg: ", errorMsg); + if (callback) { + callback(status); + } + return; + } + + JdoStatus status; + if (total_read >= 0) { + while (n - total_read > 0) { + START_CALL2(store, reader) + num_read = jdo_read(ctx, scratch + total_read, n - total_read, nullptr); + END_CALL() + if (errorCode != 0) { + status = JdoStatus::InternalError(errorCode, "failed to read ", name, + " size ", n - total_read, " readed ", num_read, + ", errmsg: ", errorMsg); + if (callback) { + callback(status); + } + return; + } + if (num_read < 0) { + // isEOF + break; + } + total_read += num_read; + } + } + + if ((total_read < 0 || total_read < n) && num_read != -1) { + // if there was an error before satisfying the current read, this logic declares it an error + // and does not try to return any of the bytes read. Don't think it matters, so the code + // is just being conservative. + status = JdoStatus::IOError("Failed to read, ", name, + ", expect read ", n, " bytes" + ", but ", total_read, " bytes readed" + ", last read ", num_read, " bytes"); + if (callback) { + callback(status); + } + return; + } + + if (total_read < 0 || total_read < n) { + // This is not an error per se. The RandomAccessFile interface expects + // that Read returns OutOfRange if fewer bytes were read than requested. + status = JdoStatus::IOError("EOF reached, ", result->size(), + " bytes were read out of ", n, + " bytes requested."); + if (callback) { + callback(status); + } + return; + } + + if (callback) { + callback(JdoStatus::OK()); + } +} + +struct PreadArgs { + JdoStore_t store = nullptr; + JdoIOContext_t reader = nullptr; + std::string name; + int64_t offset = 0; + size_t n = 0; + std::string_view *result; + char *scratch; + std::function callback; +}; + +static void preadCallback(JdoHandleCtx_t ctx, int64_t num_read, void *userdata) { + PreadArgs *args = (PreadArgs *) userdata; + auto store = args->store; + auto reader = args->reader; + auto name = args->name; + auto offset = args->offset; + auto n = args->n; + auto result = args->result; + auto scratch = args->scratch; + auto total_read = num_read; + std::function callback = args->callback; + delete args; + + END_CALL() + if (errorCode != 0) { + auto status = JdoStatus::InternalError(errorCode, "failed to read ", name, " offset ", offset + total_read, + " size ", n - total_read, " readed ", num_read, + ", errmsg: ", errorMsg); + if (callback) { + callback(status); + } + return; + } + + JdoStatus status; + if (total_read >= 0) { + while (n - total_read > 0) { + START_CALL2(store, reader) + num_read = jdo_pread(ctx, scratch + total_read, n - total_read, offset + total_read, nullptr); + END_CALL() + if (errorCode != 0) { + status = JdoStatus::InternalError(errorCode, "failed to read ", name, " offset ", offset + total_read, + " size ", n - total_read, " readed ", num_read, + ", errmsg: ", errorMsg); + if (callback) { + callback(status); + } + return; + } + if (num_read < 0) { + // isEOF + break; + } + total_read += num_read; + } + + + // Set the results. + *result = std::string_view(scratch, total_read); + } + + if ((total_read < 0 || total_read < n) && num_read != -1) { + // if there was an error before satisfying the current read, this logic declares it an error + // and does not try to return any of the bytes read. Don't think it matters, so the code + // is just being conservative. + status = JdoStatus::IOError("Failed to read, ", name, + ", expect read ", n, " bytes" + ", but ", total_read, " bytes readed" + ", last read ", num_read, " bytes"); + if (callback) { + callback(status); + } + return; + } + + if (total_read < 0 || total_read < n) { + // This is not an error per se. The RandomAccessFile interface expects + // that Read returns OutOfRange if fewer bytes were read than requested. + status = JdoStatus::IOError("EOF reached, ", result->size(), + " bytes were read out of ", n, + " bytes requested."); + if (callback) { + callback(status); + } + return; + } + + if (callback) { + callback(JdoStatus::OK()); + } +} + +struct PreadvArgs { + JdoStore_t store = nullptr; + JdoIOContext_t reader = nullptr; + std::string name; + JdoLongs_t lengths; + JdoLongs_t offsets; + JdoFileBuffers_t buffers; + std::function callback; +}; + +static void preadvCallback(JdoHandleCtx_t ctx, int64_t ret, void *userdata) { + PreadvArgs *args = (PreadvArgs *) userdata; + auto store = args->store; + auto reader = args->reader; + auto name = args->name; + auto offsets = args->offsets; + auto lengths = args->lengths; + auto buffers = args->buffers; + auto rangeCnt = jdo_getFileBuffersSize(buffers); + std::function callback = args->callback; + delete args; + + END_CALL() + jdo_freeFileBuffers(buffers); + jdo_freeLongs(lengths); + jdo_freeLongs(offsets); + + JdoStatus status; + if (errorCode != 0) { + status = JdoStatus::InternalError(errorCode, "failed to preadv ", name, + " ranges ", std::to_string(rangeCnt), + ", errmsg: ", errorMsg); + } + + if (callback) { + callback(status); + } +} + +struct Int64Args { + JdoStore_t store = nullptr; + JdoIOContext_t reader = nullptr; + JdoIOContext_t writer = nullptr; + std::function callback; +}; + +static void tellCallback(JdoHandleCtx_t ctx, int64_t result, void *userdata) { + Int64Args* args = (Int64Args*) userdata; + auto store = args->store; + auto reader = args->reader; + std::function callback = args->callback; + delete args; + + END_CALL() + + JdoStatus status; + if (errorCode != 0) { + status = JdoStatus::InternalError(errorCode, "file tell failed: ", errorMsg); + } + + if (callback) { + callback(status, result); + } +} + +static void seekCallback(JdoHandleCtx_t ctx, int64_t result, void *userdata) { + Int64Args* args = (Int64Args*) userdata; + auto store = args->store; + auto reader = args->reader; + std::function callback = args->callback; + delete args; + + END_CALL() + + JdoStatus status; + if (errorCode != 0) { + status = JdoStatus::InternalError(errorCode, "file seek failed: ", errorMsg); + } + + if (callback) { + callback(status, result); + } +} + +static void getFileLengthCallback(JdoHandleCtx_t ctx, int64_t result, void *userdata) { + Int64Args* args = (Int64Args*) userdata; + auto store = args->store; + auto reader = args->reader; + std::function callback = args->callback; + delete args; + + END_CALL() + + JdoStatus status; + if (errorCode != 0) { + status = JdoStatus::InternalError(errorCode, "get file length failed: ", errorMsg); + } + + if (callback) { + callback(status, result); + } +} + +struct CloseReaderArgs { + JdoStore_t store = nullptr; + JdoIOContext_t reader = nullptr; + std::string name; + std::atomic* isClosed = nullptr; + std::function callback; +}; + +static void closeReaderCallback(JdoHandleCtx_t ctx, bool result, void* userdata) { + CloseReaderArgs* args = (CloseReaderArgs*) userdata; + auto store = args->store; + auto reader = args->reader; + auto name = args->name; + std::atomic* isClosed = args->isClosed; + std::function callback = args->callback; + delete args; + + END_CALL() + + *isClosed = true; + + JdoStatus status; + if (errorCode != 0) { + status = JdoStatus::InternalError(errorCode, "close file failed: ", errorMsg); + } + + if (callback) { + callback(status); + } +} + +class JindoReader : public JdoReader { +public: + JindoReader(const std::string &fname, + JdoStore_t store, + JdoIOContext_t reader) + : _name(fname), + _store(store), + _reader(reader), + _is_closed(false) { + } + + ~JindoReader() override { + if (_reader) { + if (!_is_closed) { + JdoHandleCtx_t ctx = jdo_createHandleCtx2(_store, _reader); + jdo_close(ctx, nullptr); + jdo_freeHandleCtx(ctx); + } + jdo_freeIOContext(_reader); + _reader = nullptr; + } + } + + bool closed() const override { + return _is_closed; + } + + JdoStatus name(std::string_view *result) const override { + *result = _name; + return JdoStatus::OK(); + } + + JdoStatus read(size_t n, std::string_view *result, char *scratch) override { + std::lock_guard lock(_lock); + size_t total_read = 0; + int64_t num_read = 0; + while (n - total_read > 0) { + START_CALL2(_store, _reader) + num_read = jdo_read(ctx, scratch + total_read, n - total_read, nullptr); + END_CALL() + if (errorCode != 0) { + return JdoStatus::InternalError(errorCode, "failed to read ", _name, " size ", n - total_read, " readed ", num_read, + ", errmsg: ", errorMsg); + } + if (num_read < 0) { + // isEOF + break; + } + total_read += num_read; + } + // Set the results. + *result = std::string_view(scratch, total_read); + + if (total_read < n && num_read != -1) { + // if there was an error before satisfying the current read, this logic declares it an error + // and does not try to return any of the bytes read. Don't think it matters, so the code + // is just being conservative. + return JdoStatus::IOError("Failed to read, ", _name, + ", expect read ", n, " bytes" + ", but ", total_read, " bytes readed" + ", last read ", num_read, " bytes"); + } + + if (total_read < n) { + // This is not an error per se. The RandomAccessFile interface expects + // that Read returns OutOfRange if fewer bytes were read than requested. + return JdoStatus::IOError("EOF reached, ", result->size(), + " bytes were read out of ", n, + " bytes requested."); + } + + return JdoStatus::OK(); + } + + JdoTaskPtr readAsync(size_t n, std::string_view *result, char *scratch, std::function callback) override { + size_t total_read = 0; + int64_t num_read = 0; + ReadArgs *args = new ReadArgs(); + args->store = _store; + args->reader = _reader; + args->name = _name; + args->n = n; + args->result = result; + args->scratch = scratch; + args->callback = callback; + START_CALL2(_store, _reader) + auto options = jdo_createOptions(); + jdo_setInt64Callback(options, readCallback); + jdo_setCallbackContext(options, args); + auto operationCall = jdo_readAsync(ctx, scratch + total_read, n - total_read, options); + END_CALL() + jdo_freeOptions(options); + + return std::make_shared(_store, _reader, operationCall); + } + + JdoStatus pread(int64_t offset, size_t n, std::string_view *result, char *scratch) override { + size_t total_read = 0; + int64_t num_read = 0; + while (n - total_read > 0) { + START_CALL2(_store, _reader) + num_read = jdo_pread(ctx, scratch + total_read, n - total_read, offset + total_read, nullptr); + END_CALL() + if (errorCode != 0) { + return JdoStatus::InternalError(errorCode, "failed to read ", _name, " offset ", offset + total_read, + " size ", n - total_read, " readed ", num_read, + ", errmsg: ", errorMsg); + } + if (num_read < 0) { + // isEOF + break; + } + total_read += num_read; + } + // Set the results. + *result = std::string_view(scratch, total_read); + + if (total_read < n && num_read != -1) { + // if there was an error before satisfying the current read, this logic declares it an error + // and does not try to return any of the bytes read. Don't think it matters, so the code + // is just being conservative. + return JdoStatus::IOError("Failed to read, ", _name, + ", expect read ", n, " bytes" + ", but ", total_read, " bytes readed" + ", last read ", num_read, " bytes"); + } + + if (total_read < n) { + // This is not an error per se. The RandomAccessFile interface expects + // that Read returns OutOfRange if fewer bytes were read than requested. + return JdoStatus::IOError("EOF reached, ", result->size(), + " bytes were read out of ", n, + " bytes requested."); + } + + return JdoStatus::OK(); + } + + JdoTaskPtr preadAsync(int64_t offset, size_t n, std::string_view *result, char *scratch, + std::function callback) override { + size_t total_read = 0; + int64_t num_read = 0; + PreadArgs *args = new PreadArgs(); + args->store = _store; + args->reader = _reader; + args->name = _name; + args->offset = offset; + args->n = n; + args->result = result; + args->scratch = scratch; + args->callback = callback; + START_CALL2(_store, _reader) + auto options = jdo_createOptions(); + jdo_setInt64Callback(options, preadCallback); + jdo_setCallbackContext(options, args); + auto operationCall = jdo_preadAsync(ctx, scratch + total_read, n - total_read, offset + total_read, options); + END_CALL() + jdo_freeOptions(options); + + return std::make_shared(_store, _reader, operationCall); + } + + JdoStatus preadv(std::vector& offsets, std::vector& lengths, std::vector& scratches) override { + START_CALL2(_store, _reader) + JdoLongs_t offs = jdo_createLongs(); + for (auto& offset : offsets) { + jdo_appendLong(offs, offset); + } + JdoLongs_t lens = jdo_createLongs(); + for (auto& len: lengths) { + jdo_appendLong(lens, len); + } + JdoFileBuffers_t fileBuffers = jdo_createFileBuffers(); + for (auto& scratch : scratches) { + jdo_appendFileBuffer(fileBuffers, scratch); + } + jdo_preadv(ctx, fileBuffers, lens, offs, nullptr); + END_CALL() + jdo_freeFileBuffers(fileBuffers); + jdo_freeLongs(lens); + jdo_freeLongs(offs); + if (errorCode != 0) { + return JdoStatus::InternalError(errorCode, "failed to preadv ", _name, + " ranges ", scratches.size(), + ", errmsg: ", errorMsg); + } + return JdoStatus::OK(); + } + + JdoTaskPtr preadvAsync(std::vector& offsets, std::vector& lengths, std::vector& scratches, + std::function callback) override { + JdoLongs_t offs = jdo_createLongs(); + for (auto& offset : offsets) { + jdo_appendLong(offs, offset); + } + JdoLongs_t lens = jdo_createLongs(); + for (auto& len: lengths) { + jdo_appendLong(lens, len); + } + JdoFileBuffers_t fileBuffers = jdo_createFileBuffers(); + for (auto& scratch : scratches) { + jdo_appendFileBuffer(fileBuffers, scratch); + } + PreadvArgs *args = new PreadvArgs(); + args->store = _store; + args->reader = _reader; + args->name = _name; + args->offsets = offs; + args->lengths = lens; + args->buffers = fileBuffers; + args->callback = callback; + START_CALL2(_store, _reader) + auto options = jdo_createOptions(); + jdo_setInt64Callback(options, preadvCallback); + jdo_setCallbackContext(options, args); + auto operationCall = jdo_preadvAsync(ctx, fileBuffers, lens, offs, options); + END_CALL() + jdo_freeOptions(options); + + return std::make_shared(_store, _reader, operationCall); + } + + JdoStatus tell(int64_t& offset) override { + START_CALL2(_store, _reader) + offset = jdo_tell(ctx, nullptr); + END_CALL() + + if (errorCode != 0) { + return JdoStatus::InternalError(errorCode, "tell file failed: ", errorMsg); + } + + return JdoStatus::OK(); + } + + JdoTaskPtr tellAsync(std::function callback) override { + Int64Args *args = new Int64Args(); + args->store = _store; + args->reader = _reader; + args->callback = callback; + START_CALL2(_store, _reader) + auto options = jdo_createOptions(); + jdo_setInt64Callback(options, tellCallback); + jdo_setCallbackContext(options, args); + auto operationCall = jdo_tellAsync(ctx,options); + END_CALL() + jdo_freeOptions(options); + + return std::make_shared(_store, _reader, operationCall); + } + + + JdoStatus seek(int64_t offset) override { + START_CALL2(_store, _reader) + jdo_seek(ctx, offset, nullptr); + END_CALL() + + if (errorCode != 0) { + return JdoStatus::InternalError(errorCode, "seek file failed: ", errorMsg); + } + + return JdoStatus::OK(); + } + + JdoTaskPtr seekAsync(int64_t offset, std::function callback) override { + Int64Args *args = new Int64Args(); + args->store = _store; + args->reader = _reader; + args->callback = callback; + START_CALL2(_store, _reader) + auto options = jdo_createOptions(); + jdo_setInt64Callback(options, seekCallback); + jdo_setCallbackContext(options, args); + auto operationCall = jdo_seekAsync(ctx, offset, options); + END_CALL() + jdo_freeOptions(options); + + return std::make_shared(_store, _reader, operationCall); + } + + JdoStatus getFileLength(int64_t& length) override { + START_CALL2(_store, _reader) + length = jdo_getFileLength(ctx, nullptr); + END_CALL() + + if (errorCode != 0) { + return JdoStatus::InternalError(errorCode, "tell file failed: ", errorMsg); + } + + return JdoStatus::OK(); + } + + JdoTaskPtr getFileLengthAsync(std::function callback) override { + Int64Args *args = new Int64Args(); + args->store = _store; + args->reader = _reader; + args->callback = callback; + START_CALL2(_store, _reader) + auto options = jdo_createOptions(); + jdo_setInt64Callback(options, getFileLengthCallback); + jdo_setCallbackContext(options, args); + auto operationCall = jdo_getFileLengthAsync(ctx,options); + END_CALL() + jdo_freeOptions(options); + + return std::make_shared(_store, _reader, operationCall); + } + + JdoStatus close() override { + START_CALL2(_store, _reader) + jdo_close(ctx, nullptr); + END_CALL() + + if (errorCode != 0) { + return JdoStatus::InternalError(errorCode, "close file failed: ", errorMsg); + } + + _is_closed = true; + return JdoStatus::OK(); + } + + JdoTaskPtr closeAsync(std::function callback) override { + CloseReaderArgs* args = new CloseReaderArgs(); + args->store = _store; + args->reader = _reader; + args->name = _name; + args->isClosed = &_is_closed; + args->callback = callback; + + START_CALL2(_store, _reader) + auto options = jdo_createOptions(); + jdo_setBoolCallback(options, closeReaderCallback); + jdo_setCallbackContext(options, args); + auto operationCall = jdo_closeAsync(ctx, options); + END_CALL() + jdo_freeOptions(options); + + return std::make_shared(_store, _reader, operationCall); + } + +private: + std::mutex _lock; + std::string _name; + JdoStore_t _store; + JdoIOContext_t _reader = nullptr; + std::atomic _is_closed = false; +}; + +struct WriteArgs { + JdoStore_t store = nullptr; + JdoIOContext_t writer = nullptr; + std::string name; + std::function callback; +}; + +static void writeCallback(JdoHandleCtx_t ctx, int64_t result, void* userdata) { + WriteArgs* args = (WriteArgs*) userdata; + auto store = args->store; + auto writer = args->writer; + auto name = args->name; + std::function callback = args->callback; + delete args; + + END_CALL() + + JdoStatus status; + if (errorCode != 0) { + status = JdoStatus::InternalError(errorCode, "write file failed: ", errorMsg); + } + + if (callback) { + callback(status); + } +} + +struct FlushArgs { + JdoStore_t store = nullptr; + JdoIOContext_t writer = nullptr; + std::string name; + std::function callback; +}; + +static void flushCallback(JdoHandleCtx_t ctx, bool result, void* userdata) { + FlushArgs* args = (FlushArgs*) userdata; + auto store = args->store; + auto writer = args->writer; + auto name = args->name; + std::function callback = args->callback; + delete args; + + END_CALL() + + JdoStatus status; + if (errorCode != 0) { + status = JdoStatus::InternalError(errorCode, "flush file failed: ", errorMsg); + } + + if (callback) { + callback(status); + } +} + +struct CloseWriterArgs { + JdoStore_t store = nullptr; + JdoIOContext_t writer = nullptr; + std::string name; + std::atomic* isClosed = nullptr; + std::function callback; +}; + +static void closeWriterCallback(JdoHandleCtx_t ctx, bool result, void* userdata) { + CloseWriterArgs* args = (CloseWriterArgs*) userdata; + auto store = args->store; + auto writer = args->writer; + auto name = args->name; + std::atomic* isClosed = args->isClosed; + std::function callback = args->callback; + delete args; + + END_CALL() + + JdoStatus status; + if (errorCode != 0) { + status = JdoStatus::InternalError(errorCode, "close file failed: ", errorMsg); + } + + *isClosed = true; + + if (callback) { + callback(status); + } +} + +class JindoWriter : public JdoWriter { +public: + JindoWriter(const std::string& fname, + JdoStore_t store, + JdoIOContext_t writer) + : _name(fname), + _store(store), + _writer(writer), + _is_closed(false) { + } + + ~JindoWriter() override { + if (_writer) { + if (!_is_closed) { + JdoHandleCtx_t ctx = jdo_createHandleCtx2(_store, _writer); + jdo_close(ctx, nullptr); + jdo_freeHandleCtx(ctx); + } + jdo_freeIOContext(_writer); + _writer = nullptr; + } + } + + bool closed() const override { + return _is_closed; + } + + JdoStatus name(std::string_view* result) const override { + *result = _name; + return JdoStatus::OK(); + } + + JdoStatus write(std::string_view data) override { + JDO_RETURN_IF_ERROR(_CheckClosed()); + + START_CALL2(_store, _writer) + jdo_write(ctx, data.data(), data.size(), nullptr); + END_CALL() + + if (errorCode != 0) { + return JdoStatus::InternalError(errorCode, "write file failed: ", errorMsg); + } + + return JdoStatus::OK(); + } + + JdoTaskPtr writeAsync(std::string_view data, std::function callback) override { + WriteArgs* args = new WriteArgs(); + args->store = _store; + args->writer = _writer; + args->name = _name; + args->callback = callback; + + START_CALL2(_store, _writer) + auto options = jdo_createOptions(); + jdo_setInt64Callback(options, writeCallback); + jdo_setCallbackContext(options, args); + auto operationCall = jdo_writeAsync(ctx, data.data(), data.size(), options); + END_CALL() + jdo_freeOptions(options); + + return std::make_shared(_store, _writer, operationCall); + } + + JdoStatus flush() override { + JDO_RETURN_IF_ERROR(_CheckClosed()); + + START_CALL2(_store, _writer) + jdo_flush(ctx, nullptr); + END_CALL() + + if (errorCode != 0) { + return JdoStatus::InternalError(errorCode, "flush file failed: ", errorMsg); + } + + return JdoStatus::OK(); + } + + JdoStatus tell(int64_t& offset) override { + JDO_RETURN_IF_ERROR(_CheckClosed()); + + START_CALL2(_store, _writer) + offset = jdo_tell(ctx, nullptr); + END_CALL() + + if (errorCode != 0) { + return JdoStatus::InternalError(errorCode, "tell file failed: ", errorMsg); + } + + return JdoStatus::OK(); + } + + + JdoTaskPtr tellAsync(std::function callback) override { + Int64Args *args = new Int64Args(); + args->store = _store; + args->writer = _writer; + args->callback = callback; + START_CALL2(_store, _writer) + auto options = jdo_createOptions(); + jdo_setInt64Callback(options, tellCallback); + jdo_setCallbackContext(options, args); + auto operationCall = jdo_tellAsync(ctx,options); + END_CALL() + jdo_freeOptions(options); + + return std::make_shared(_store, _writer, operationCall); + } + + JdoTaskPtr flushAsync(std::function callback) override { + FlushArgs* args = new FlushArgs(); + args->store = _store; + args->writer = _writer; + args->name = _name; + args->callback = callback; + + START_CALL2(_store, _writer) + auto options = jdo_createOptions(); + jdo_setBoolCallback(options, flushCallback); + jdo_setCallbackContext(options, args); + auto operationCall = jdo_flushAsync(ctx, options); + END_CALL() + jdo_freeOptions(options); + + return std::make_shared(_store, _writer, operationCall); + } + + JdoStatus close() override { + JDO_RETURN_IF_ERROR(_CheckClosed()); + + START_CALL2(_store, _writer) + jdo_close(ctx, nullptr); + END_CALL() + + if (errorCode != 0) { + return JdoStatus::InternalError(errorCode, "close file failed: ", errorMsg); + } + _is_closed = true; + return JdoStatus::OK(); + } + + JdoTaskPtr closeAsync(std::function callback) override { + CloseWriterArgs* args = new CloseWriterArgs(); + args->store = _store; + args->writer = _writer; + args->name = _name; + args->isClosed = &_is_closed; + args->callback = callback; + + START_CALL2(_store, _writer) + auto options = jdo_createOptions(); + jdo_setBoolCallback(options, closeWriterCallback); + jdo_setCallbackContext(options, args); + auto operationCall = jdo_closeAsync(ctx, options); + END_CALL() + jdo_freeOptions(options); + + return std::make_shared(_store, _writer, operationCall); + } + +private: + JdoStatus _CheckClosed() { + if (_is_closed) { + return JdoStatus::IOError("Already closed."); + } + + return JdoStatus::OK(); + } + + std::string _name; + JdoStore_t _store; + JdoIOContext_t _writer = nullptr; + + std::atomic _is_closed = false; +}; + +JdoFileSystem::JdoFileSystem() { +} + +JdoFileSystem::~JdoFileSystem() { + if (_store != nullptr) { + jdo_freeStore(_store); + _store = nullptr; + } + if (_options != nullptr) { + jdo_freeOptions(_options); + _options = nullptr; + } +} + +JdoStatus JdoFileSystem::init(const std::string &uri, const std::string &user, const std::shared_ptr& config) { + if (_inited) { + return JdoStatus::InitError("already inited"); + } + + _options = jdo_createOptions(); + auto configMap = config->getAll(); + for (auto& kv : configMap) { + jdo_setOption(_options, kv.first.c_str(), kv.second.c_str()); + } + if (getenv("OSS_ENDPOINT")) { + jdo_setOption(_options, "fs.oss.endpoint", getenv("OSS_ENDPOINT")); + } + if (getenv("OSS_ACCESS_ID")) { + jdo_setOption(_options, "fs.oss.accessKeyId", getenv("OSS_ACCESS_ID")); + } + if (getenv("OSS_ACCESS_KEY")) { + jdo_setOption(_options, "fs.oss.accessKeySecret", getenv("OSS_ACCESS_KEY")); + } + + jdo_setOption(_options, "logger.appender", "file"); + + _store = jdo_createStore(_options, uri.c_str()); + + START_CALL(_store) + jdo_init(ctx, user.c_str()); + END_CALL() + + if (errorCode != 0) { + destroy(); + return JdoStatus::InitError("create jdo filesystem failed" + ", uri: ", uri, + ", user: ", user, + ", error: ", errorMsg); + } + + _inited = true; + return JdoStatus::OK(); +} + +JdoStatus JdoFileSystem::destroy() { + if (!_inited) { + return JdoStatus::OK(); + } + + if (_store != nullptr) { + START_CALL(_store) + jdo_destroyStore(_store); + END_CALL() + jdo_freeStore(_store); + _store = nullptr; + } + if (_options != nullptr) { + jdo_freeOptions(_options); + _options = nullptr; + } + _inited = false; + + return JdoStatus::OK(); +} + +JdoStatus JdoFileSystem::openReader(const std::string &path, + std::unique_ptr *result) { + auto [store, err] = GetJdoStore(path); + if (!err.ok()) { + return err; + } + + START_CALL(store) + auto stream = jdo_open(ctx, path.c_str(), JDO_OPEN_FLAG_READ_ONLY, 0777, nullptr); + END_CALL() + + if (errorCode != 0) { + return JdoStatus::InternalError(errorCode, "open file reader failed: ", errorMsg); + } + + result->reset(new JindoReader(path, store, stream)); + + return JdoStatus::OK(); +} + +JdoStatus JdoFileSystem::openWriter(const std::string &path, + std::unique_ptr *result) { + auto [store, err] = GetJdoStore(path); + if (!err.ok()) { + return err; + } + + START_CALL(store) + auto flag = JDO_OPEN_FLAG_CREATE | JDO_OPEN_FLAG_OVERWRITE; + auto handle = jdo_open(ctx, path.c_str(), flag, 0777, nullptr); + END_CALL() + + if (errorCode != 0) { + return JdoStatus::InternalError(errorCode, "open file writer failed: ", errorMsg); + } + + result->reset(new JindoWriter(path, store, handle)); + return JdoStatus::OK(); +} + +JdoStatus JdoFileSystem::mkdir(const std::string& path, bool recursive) { + auto [store, err] = GetJdoStore(path); + if (!err.ok()) { + return err; + } + + START_CALL(store) + bool result = jdo_mkdir(ctx, path.c_str(), recursive, 0777, nullptr); + END_CALL() + + if (errorCode != 0) { + return JdoStatus::InternalError(errorCode, "mkdir failed: ", errorMsg); + } else if (!result) { + return JdoStatus::IOError("mkdir failed: ", path); + } + + return JdoStatus::OK(); +} + +struct MkdirArgs { + std::string path; + bool recursive; + std::function callback; +}; + +static void mkdirCallback(JdoHandleCtx_t ctx, bool result, void* userdata) { + MkdirArgs* args = (MkdirArgs*) userdata; + auto path = args->path; + std::function callback = args->callback; + delete args; + + END_CALL() + JdoStatus status; + if (errorCode != 0) { + status = JdoStatus::InternalError(errorCode, "mkdir failed: ", errorMsg); + } else if (!result) { + status = JdoStatus::IOError("mkdir failed: ", path); + } + if (callback) { + callback(status); + } +} + +JdoTaskPtr JdoFileSystem::mkdirAsync(const std::string& path, bool recursive, std::function callback) { + auto [store, err] = GetJdoStore(path); + if (!err.ok()) { + return nullptr; + } + + MkdirArgs* args = new MkdirArgs(); + args->path = path; + args->recursive = recursive; + args->callback = callback; + + START_CALL(store) + auto options = jdo_createOptions(); + jdo_setBoolCallback(options, mkdirCallback); + jdo_setCallbackContext(options, args); + auto operationCall = jdo_mkdirAsync(ctx, path.c_str(), recursive, 0777, options); + END_CALL() + jdo_freeOptions(options); + + return std::make_shared(_store, operationCall); +} + +JdoStatus JdoFileSystem::rename(const std::string &oldpath, const std::string &newpath) { + auto [store, err] = GetJdoStore(oldpath); + if (!err.ok()) { + return err; + } + + if (exists(newpath).ok()) { + auto status = remove(newpath, false); + if (!status.ok()) { + return status; + } + } + + START_CALL(store) + bool result = jdo_rename(ctx, oldpath.c_str(), newpath.c_str(), nullptr); + END_CALL() + + if (errorCode != 0) { + return JdoStatus::InternalError(errorCode, "rename failed: ", errorMsg); + } else if (!result) { + return JdoStatus::IOError("rename failed," + " from ", oldpath, + " to ", newpath); + } + return JdoStatus::OK(); +} + +struct RenameArgs { + std::string oldpath; + std::string newpath; + std::function callback; +}; + +static void renameCallback(JdoHandleCtx_t ctx, bool result, void* userdata) { + RenameArgs* args = (RenameArgs*) userdata; + auto oldpath = args->oldpath; + auto newpath = args->newpath; + std::function callback = args->callback; + delete args; + + END_CALL() + JdoStatus status; + if (errorCode != 0) { + status = JdoStatus::InternalError(errorCode, "rename failed: ", errorMsg); + } else if (!result) { + status = JdoStatus::IOError("rename failed," + " from ", oldpath, + " to ", newpath); + } + if (callback) { + callback(status); + } +} + +JdoTaskPtr JdoFileSystem::renameAsync(const std::string &oldpath, const std::string &newpath, std::function callback) { + auto [store, err] = GetJdoStore(oldpath); + if (!err.ok()) { + return nullptr; + } + + if (exists(newpath).ok()) { + auto status = remove(newpath, false); + if (!status.ok()) { + return nullptr; + } + } + + RenameArgs* args = new RenameArgs(); + args->oldpath = oldpath; + args->newpath = newpath; + args->callback = callback; + + START_CALL(store) + auto options = jdo_createOptions(); + jdo_setBoolCallback(options, renameCallback); + jdo_setCallbackContext(options, args); + auto operationCall = jdo_renameAsync(ctx, oldpath.c_str(), newpath.c_str(), options); + END_CALL() + jdo_freeOptions(options); + + return std::make_shared(_store, operationCall); +} + +JdoStatus JdoFileSystem::remove(const std::string& path, bool recursive) { + auto [store, err] = GetJdoStore(path); + if (!err.ok()) { + return err; + } + + START_CALL(store) + bool result = jdo_remove(ctx, path.c_str(), recursive, nullptr); + END_CALL() + + if (errorCode != 0) { + return JdoStatus::InternalError(errorCode, "delete failed: ", errorMsg); + } else if (!result) { + return JdoStatus::IOError("delete failed: ", path, " recursive ", recursive); + } + return JdoStatus::OK(); +} + +struct RemoveArgs { + std::string path; + bool recursive; + std::function callback; +}; + +static void removeCallback(JdoHandleCtx_t ctx, bool result, void* userdata) { + RemoveArgs* args = (RemoveArgs*) userdata; + auto path = args->path; + auto recursive = args->recursive; + std::function callback = args->callback; + delete args; + + END_CALL() + JdoStatus status; + if (errorCode != 0) { + status = JdoStatus::InternalError(errorCode, "delete failed: ", errorMsg); + } else if (!result) { + status = JdoStatus::IOError("delete failed: ", path, " recursive ", recursive); + } + if (callback) { + callback(status); + } +} + +JdoTaskPtr JdoFileSystem::removeAsync(const std::string& path, bool recursive, std::function callback) { + auto [store, err] = GetJdoStore(path); + if (!err.ok()) { + return nullptr; + } + + RemoveArgs* args = new RemoveArgs(); + args->path = path; + args->recursive = recursive; + args->callback = callback; + + START_CALL(store) + auto options = jdo_createOptions(); + jdo_setBoolCallback(options, removeCallback); + jdo_setCallbackContext(options, args); + auto operationCall = jdo_removeAsync(ctx, path.c_str(), recursive, options); + END_CALL() + jdo_freeOptions(options); + + return std::make_shared(_store, operationCall); +} + +JdoStatus JdoFileSystem::exists(const std::string& path) { + auto [store, err] = GetJdoStore(path); + if (!err.ok()) { + return JdoStatus::IOError(err); + } + + START_CALL(store) + bool result = jdo_exists(ctx, path.c_str(), nullptr); + END_CALL() + + if (errorCode != 0) { + return JdoStatus::InternalError(errorCode, "exists failed: ", errorMsg); + } else if (!result) { + return JdoStatus::NotFound(path, " doesn't exist."); + } + + return JdoStatus::OK(); +} + +struct ExistsArgs { + std::string path; + std::function callback; +}; + +static void existsCallback(JdoHandleCtx_t ctx, bool result, void* userdata) { + ExistsArgs* args = (ExistsArgs*) userdata; + auto path = args->path; + std::function callback = args->callback; + delete args; + + END_CALL() + JdoStatus status; + if (errorCode != 0) { + status = JdoStatus::InternalError(errorCode, "exists failed: ", errorMsg); + } else if (!result) { + status = JdoStatus::NotFound(path, " doesn't exist."); + } + if (callback) { + callback(status); + } +} + +JdoTaskPtr JdoFileSystem::existsAsync(const std::string &path, std::function callback) { + auto [store, err] = GetJdoStore(path); + if (!err.ok()) { + return nullptr; + } + + ExistsArgs* args = new ExistsArgs(); + args->path = path; + args->callback = callback; + + START_CALL(store) + auto options = jdo_createOptions(); + jdo_setBoolCallback(options, existsCallback); + jdo_setCallbackContext(options, args); + auto operationCall = jdo_existsAsync(ctx, path.c_str(), options); + END_CALL() + jdo_freeOptions(options); + + return std::make_shared(_store, operationCall); +} + +JdoStatus JdoFileSystem::getFileInfo(const std::string& path, JdoFileInfo* stat) { + auto [store, err] = GetJdoStore(path); + if (!err.ok()) { + return err; + } + + return getFileInfoInternal(store, path, stat); +} + +struct GetFileInfoArgs { + std::string path; + std::function callback; +}; + +static void getFileInfoCallback(JdoHandleCtx_t ctx, JdoFileStatus_t fileStatus, void* userdata) { + GetFileInfoArgs* args = (GetFileInfoArgs*) userdata; + auto path = args->path; + std::function callback = args->callback; + delete args; + + auto info = std::make_shared(); + covertToFileInfo(fileStatus, info.get()); + jdo_freeFileStatus(fileStatus); + + END_CALL() + JdoStatus status; + if (errorCode != 0) { + if (errorCode == JDO_FILE_NOT_FOUND_ERROR) { + status = JdoStatus::NotFound("cant not found ", path); + } else { + status = JdoStatus::InternalError(errorCode, "get status failed: ", errorMsg); + } + } + + if (callback) { + callback(status, info); + } +} + +JdoTaskPtr JdoFileSystem::getFileInfoAsync(const std::string &path, std::function callback) { + auto [store, err] = GetJdoStore(path); + if (!err.ok()) { + return nullptr; + } + + GetFileInfoArgs* args = new GetFileInfoArgs(); + args->path = path; + args->callback = callback; + + START_CALL(store) + auto options = jdo_createOptions(); + jdo_setFileStatusCallback(options, getFileInfoCallback); + jdo_setCallbackContext(options, args); + auto operationCall = jdo_getFileStatusAsync(ctx, path.c_str(), options); + END_CALL() + jdo_freeOptions(options); + + return std::make_shared(_store, operationCall); +} + +JdoStatus JdoFileSystem::listDir(const std::string &dir, bool recursive, JdoListResult *result) { + auto [store, err] = GetJdoStore(dir); + if (!err.ok()) { + return err; + } + + START_CALL(store) + JdoListDirResult_t listResult = jdo_listDir(ctx, dir.c_str(), recursive, nullptr); + END_CALL() + + if (errorCode != 0) { + return JdoStatus::InternalError(errorCode, "get children failed: ", errorMsg); + } + + auto numEntries = jdo_getListDirResultSize(listResult); + for (int i = 0; i < numEntries; i++) { + auto fileStatus = jdo_getListDirFileStatus(listResult, i); + if (fileStatus == nullptr) { + continue; + } + JdoFileInfo info; + covertToFileInfo(fileStatus, &info); + result->infos.push_back(info); + } + result->truncated = jdo_isListDirResultTruncated(listResult); + auto nextMarker = jdo_getListDirResultNextMarker(listResult); + if (nextMarker) { + result->nextMarker = nextMarker; + } + + jdo_freeListDirResult(listResult); + return JdoStatus::OK(); +} + +struct listDirArgs { + std::string path; + std::function callback; +}; + +static void listDirCallback(JdoHandleCtx_t ctx, JdoListDirResult_t listResult, void* userdata) { + listDirArgs* args = (listDirArgs*) userdata; + auto path = args->path; + std::function callback = args->callback; + delete args; + + END_CALL() + JdoStatus status; + auto result = std::make_shared(); + if (errorCode != 0) { + status = JdoStatus::InternalError(errorCode, "get children failed: ", errorMsg); + callback(status, result); + return; + } + + auto numEntries = jdo_getListDirResultSize(listResult); + for (int i = 0; i < numEntries; i++) { + auto fileStatus = jdo_getListDirFileStatus(listResult, i); + if (fileStatus == nullptr) { + continue; + } + JdoFileInfo info; + covertToFileInfo(fileStatus, &info); + result->infos.push_back(info); + } + result->truncated = jdo_isListDirResultTruncated(listResult); + auto nextMarker = jdo_getListDirResultNextMarker(listResult); + if (nextMarker) { + result->nextMarker = nextMarker; + } + + jdo_freeListDirResult(listResult); + + if (callback) { + callback(status, result); + } +} + +JdoTaskPtr JdoFileSystem::listDirAsync(const std::string &path, bool recursive, std::function callback) { + auto [store, err] = GetJdoStore(path); + if (!err.ok()) { + return nullptr; + } + + listDirArgs* args = new listDirArgs(); + args->path = path; + args->callback = callback; + + START_CALL(store) + auto options = jdo_createOptions(); + jdo_setListDirResultCallback(options, listDirCallback); + jdo_setCallbackContext(options, args); + auto operationCall = jdo_listDirAsync(ctx, path.c_str(), false, options); + END_CALL() + jdo_freeOptions(options); + + return std::make_shared(_store, operationCall); +} + +JdoStatus JdoFileSystem::getContentSummary(const std::string &dir, bool recursive, JdoContentSummary *result) { + auto [store, err] = GetJdoStore(dir); + if (!err.ok()) { + return err; + } + + START_CALL(store) + JdoContentSummary_t contentSummary = jdo_getContentSummary(ctx, dir.c_str(), recursive, nullptr); + END_CALL() + + if (errorCode != 0) { + return JdoStatus::InternalError(errorCode, "get content summary failed: ", errorMsg); + } + + result->fileSize = jdo_getContentSummaryFileSize(contentSummary); + result->fileCount = jdo_getContentSummaryFileCount(contentSummary); + result->dirCount = jdo_getContentSummaryDirectoryCount(contentSummary); + + jdo_freeContentSummary(contentSummary); + return JdoStatus::OK(); +} + +struct GetContentSummaryArgs { + std::string path; + std::function callback; +}; + +static void getContentSummaryCallback(JdoHandleCtx_t ctx, JdoContentSummary_t contentSummary, void* userdata) { + GetContentSummaryArgs* args = (GetContentSummaryArgs*) userdata; + auto path = args->path; + std::function callback = args->callback; + delete args; + + END_CALL() + JdoStatus status; + auto result = std::make_shared(); + if (errorCode != 0) { + status = JdoStatus::InternalError(errorCode, "get content summary failed: ", errorMsg); + callback(status, result); + return; + } + + result->fileSize = jdo_getContentSummaryFileSize(contentSummary); + result->fileCount = jdo_getContentSummaryFileCount(contentSummary); + result->dirCount = jdo_getContentSummaryDirectoryCount(contentSummary); + + jdo_freeContentSummary(contentSummary); + + if (callback) { + callback(status, result); + } +} + +JdoTaskPtr JdoFileSystem::getContentSummaryAsync(const std::string &path, bool recursive, std::function callback) { + auto [store, err] = GetJdoStore(path); + if (!err.ok()) { + return nullptr; + } + + GetContentSummaryArgs* args = new GetContentSummaryArgs(); + args->path = path; + args->callback = callback; + + START_CALL(store) + auto options = jdo_createOptions(); + jdo_setContentSummaryCallback(options, getContentSummaryCallback); + jdo_setCallbackContext(options, args); + auto operationCall = jdo_getContentSummaryAsync(ctx, path.c_str(), false, options); + END_CALL() + jdo_freeOptions(options); + + return std::make_shared(_store, operationCall); +} + +JdoStatus JdoFileSystem::setPermission(const std::string& path, int16_t perm) { + auto [store, err] = GetJdoStore(path); + if (!err.ok()) { + return err; + } + + START_CALL(store) + bool result = jdo_setPermission(ctx, path.c_str(), perm, nullptr); + END_CALL() + + if (errorCode != 0) { + return JdoStatus::InternalError(errorCode, "setPermission failed: ", errorMsg); + } else if (!result) { + return JdoStatus::IOError("setPermission failed: ", path); + } + + return JdoStatus::OK(); +} + +struct SetPermissionArgs { + std::string path; + int16_t perm; + std::function callback; +}; + +static void setPermissionCallback(JdoHandleCtx_t ctx, bool result, void* userdata) { + SetPermissionArgs* args = (SetPermissionArgs*) userdata; + auto path = args->path; + std::function callback = args->callback; + delete args; + + END_CALL() + JdoStatus status; + if (errorCode != 0) { + status = JdoStatus::InternalError(errorCode, "setPermission failed: ", errorMsg); + } else if (!result) { + status = JdoStatus::IOError("setPermission failed: ", path); + } + if (callback) { + callback(status); + } +} + +JdoTaskPtr JdoFileSystem::setPermissionAsync(const std::string& path, int16_t perm, std::function callback) { + auto [store, err] = GetJdoStore(path); + if (!err.ok()) { + return nullptr; + } + + SetPermissionArgs* args = new SetPermissionArgs(); + args->path = path; + args->perm = perm; + args->callback = callback; + + START_CALL(store) + auto options = jdo_createOptions(); + jdo_setBoolCallback(options, setPermissionCallback); + jdo_setCallbackContext(options, args); + auto operationCall = jdo_setPermissionAsync(ctx, path.c_str(), perm, options); + END_CALL() + jdo_freeOptions(options); + + return std::make_shared(_store, operationCall); +} + + +JdoStatus JdoFileSystem::setOwner(const std::string& path, const std::string& user, const std::string& group) { + auto [store, err] = GetJdoStore(path); + if (!err.ok()) { + return err; + } + + START_CALL(store) + bool result = jdo_setOwner(ctx, path.c_str(), user.c_str(), group.c_str(), nullptr); + END_CALL() + + if (errorCode != 0) { + return JdoStatus::InternalError(errorCode, "setOwner failed: ", errorMsg); + } else if (!result) { + return JdoStatus::IOError("setOwner failed: ", path); + } + + return JdoStatus::OK(); +} + +struct SetOwnerArgs { + std::string path; + std::string user; + std::string group; + std::function callback; +}; + +static void setOwnerCallback(JdoHandleCtx_t ctx, bool result, void* userdata) { + SetOwnerArgs* args = (SetOwnerArgs*) userdata; + auto path = args->path; + std::function callback = args->callback; + delete args; + + END_CALL() + JdoStatus status; + if (errorCode != 0) { + status = JdoStatus::InternalError(errorCode, "setOwner failed: ", errorMsg); + } else if (!result) { + status = JdoStatus::IOError("setOwner failed: ", path); + } + if (callback) { + callback(status); + } +} + +JdoTaskPtr JdoFileSystem::setOwnerAsync(const std::string& path, const std::string& user, const std::string& group, std::function callback) { + auto [store, err] = GetJdoStore(path); + if (!err.ok()) { + return nullptr; + } + + SetOwnerArgs* args = new SetOwnerArgs(); + args->path = path; + args->user = user; + args->group = group; + args->callback = callback; + + START_CALL(store) + auto options = jdo_createOptions(); + jdo_setBoolCallback(options, setOwnerCallback); + jdo_setCallbackContext(options, args); + auto operationCall = jdo_setOwnerAsync(ctx, path.c_str(), user.c_str(), group.c_str(), options); + END_CALL() + jdo_freeOptions(options); + + return std::make_shared(_store, operationCall); +} + +JdoStatus JdoFileSystem::getFileInfoInternal(JdoStore_t store, + const std::string &path, + JdoFileInfo *info) { + START_CALL(store) + JdoFileStatus_t fileStatus = jdo_getFileStatus(ctx, path.c_str(), nullptr); + END_CALL() + + JdoStatus ret; + if (errorCode != 0) { + if (errorCode == JDO_FILE_NOT_FOUND_ERROR) { + return JdoStatus::NotFound("cant not found ", path); + } + return JdoStatus::InternalError(errorCode, "get status failed: ", errorMsg); + } + + covertToFileInfo(fileStatus, info); + + jdo_freeFileStatus(fileStatus); + return JdoStatus::OK(); +} + +std::tuple JdoFileSystem::GetJdoStore(const std::string &uri) { + if (!_inited) { + return {nullptr, JdoStatus::InitError()}; + } + return {_store, JdoStatus::OK()}; +}