diff --git a/CMakeLists.txt b/CMakeLists.txt index e99034dc..bb210347 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -61,6 +61,7 @@ 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_OSS "Whether to enable OSS SDK v2 file system" OFF) option(PAIMON_ENABLE_S3 "Whether to enable S3 file system" OFF) option(PAIMON_ENABLE_NETWORK_TESTS "Whether to enable tests that access real remote services over the network" OFF) @@ -75,8 +76,10 @@ endif() if(PAIMON_ENABLE_REST) add_definitions(-DPAIMON_ENABLE_REST) endif() -# libcurl backs the HTTP client shared by the S3 file system and the rest catalog. -if(PAIMON_ENABLE_S3 OR PAIMON_ENABLE_REST) +# libcurl supports the REST and S3 HTTP clients, the OSS SDK transport, and object store timestamps. +if(PAIMON_ENABLE_OSS + OR PAIMON_ENABLE_S3 + OR PAIMON_ENABLE_REST) find_package(CURL REQUIRED) endif() if(PAIMON_ENABLE_AVRO) @@ -85,6 +88,9 @@ endif() if(PAIMON_ENABLE_JINDO) add_definitions(-DPAIMON_ENABLE_JINDO) endif() +if(PAIMON_ENABLE_OSS) + add_definitions(-DPAIMON_ENABLE_OSS) +endif() if(PAIMON_ENABLE_S3) add_definitions(-DPAIMON_ENABLE_S3) endif() @@ -452,6 +458,13 @@ if(PAIMON_BUILD_TESTS) paimon_jindo_file_system_shared) list(APPEND TEST_STATIC_LINK_LIBS ${TEST_PLUGIN_LINK_LIBS}) endif() + if(PAIMON_ENABLE_OSS) + paimon_link_libraries_whole_archive(PAIMON_OSS_FILE_SYSTEM_STATIC_LINK_LIBS + paimon_oss_file_system_static) + paimon_link_libraries_no_as_needed(TEST_PLUGIN_LINK_LIBS + paimon_oss_file_system_shared) + list(APPEND TEST_STATIC_LINK_LIBS ${TEST_PLUGIN_LINK_LIBS}) + endif() if(PAIMON_ENABLE_S3) paimon_link_libraries_whole_archive(PAIMON_S3_FILE_SYSTEM_STATIC_LINK_LIBS paimon_s3_file_system_static) @@ -511,6 +524,7 @@ add_subdirectory(src/paimon/fs/local) if(PAIMON_ENABLE_JINDO) add_subdirectory(src/paimon/fs/jindo) endif() +add_subdirectory(src/paimon/fs/oss) add_subdirectory(src/paimon/fs/s3) add_subdirectory(src/paimon/format/blob) add_subdirectory(src/paimon/format/orc) diff --git a/README.md b/README.md index 00a391c0..da928e71 100644 --- a/README.md +++ b/README.md @@ -35,7 +35,7 @@ Paimon C++ currently provides: - **Scan**: batch and stream scan for append tables and primary key tables without changelog. - **Read**: append table read, primary key table read with deletion vector, and primary key table merge-on-read. - **Arrow integration**: batch read and write interfaces based on the [Arrow Columnar In-Memory Format](https://arrow.apache.org). -- **File systems**: file system abstraction with built-in local and Jindo file system support. +- **File systems**: file system abstraction with built-in local, Jindo, OSS, and S3 file system support. - **File formats**: file format abstraction with built-in ORC, Parquet, and Avro support. - **Runtime utilities**: memory pool and thread pool abstractions with default implementations. - **AI-Oriented Features**: supports RowTracking and DataEvolution mode and provides Global Index diff --git a/ci/scripts/build_paimon.sh b/ci/scripts/build_paimon.sh index 50983d4d..3826af1e 100755 --- a/ci/scripts/build_paimon.sh +++ b/ci/scripts/build_paimon.sh @@ -148,6 +148,7 @@ CMAKE_ARGS=( "-DCMAKE_BUILD_TYPE=${build_type}" "-DPAIMON_BUILD_TESTS=ON" "-DPAIMON_ENABLE_JINDO=ON" + "-DPAIMON_ENABLE_OSS=ON" "-DPAIMON_ENABLE_S3=ON" "-DPAIMON_ENABLE_LUMINA=${ENABLE_LUMINA}" "-DPAIMON_ENABLE_LUCENE=ON" diff --git a/cmake_modules/DefineOptions.cmake b/cmake_modules/DefineOptions.cmake index 21f65315..9f099208 100644 --- a/cmake_modules/DefineOptions.cmake +++ b/cmake_modules/DefineOptions.cmake @@ -219,6 +219,11 @@ if("${CMAKE_SOURCE_DIR}" STREQUAL "${CMAKE_CURRENT_SOURCE_DIR}") "" AUTO BUNDLED) + define_option_string(OSS_SDK_V2_SOURCE + "Dependency source for OSS SDK v2; SYSTEM is unsupported" + "" + AUTO + BUNDLED) define_option_string(fmt_SOURCE "Dependency source for fmt" "" diff --git a/cmake_modules/ThirdpartyToolchain.cmake b/cmake_modules/ThirdpartyToolchain.cmake index 77651910..75fe94c5 100644 --- a/cmake_modules/ThirdpartyToolchain.cmake +++ b/cmake_modules/ThirdpartyToolchain.cmake @@ -308,6 +308,16 @@ else() endif() endif() +if(DEFINED ENV{PAIMON_OSS_SDK_V2_URL}) + set(OSS_SDK_V2_SOURCE_URL "$ENV{PAIMON_OSS_SDK_V2_URL}") +elseif(EXISTS "${THIRDPARTY_DIR}/${PAIMON_OSS_SDK_V2_PKG_NAME}") + set_urls(OSS_SDK_V2_SOURCE_URL "${THIRDPARTY_DIR}/${PAIMON_OSS_SDK_V2_PKG_NAME}") +else() + set_urls(OSS_SDK_V2_SOURCE_URL + "${THIRDPARTY_MIRROR_URL}https://github.com/aliyun/alibabacloud-oss-cpp-sdk-v2/archive/refs/tags/${PAIMON_OSS_SDK_V2_BUILD_VERSION}.tar.gz" + ) +endif() + if(DEFINED ENV{PAIMON_LUMINA_URL}) set(LUMINA_SOURCE_URL "$ENV{PAIMON_LUMINA_URL}") elseif(EXISTS "${THIRDPARTY_DIR}/${PAIMON_LUMINA_PKG_NAME}") @@ -492,6 +502,25 @@ function(paimon_enforce_patched_dependency_policy) PARENT_SCOPE) endif() endif() + + if(PAIMON_ENABLE_OSS) + paimon_set_dependency_source_default( + OSS_SDK_V2 BUNDLED "OSS SDK v2 is only supported as a bundled dependency") + paimon_get_dependency_source(OSS_SDK_V2 _oss_sdk_v2_source) + if(_oss_sdk_v2_source STREQUAL "SYSTEM") + message(FATAL_ERROR "OSS_SDK_V2_SOURCE=SYSTEM is not supported. " + "Use OSS_SDK_V2_SOURCE=BUNDLED.") + elseif(_oss_sdk_v2_source STREQUAL "AUTO") + message(STATUS "Forcing OSS_SDK_V2_SOURCE to BUNDLED because paimon-cpp " + "only supports the bundled OSS SDK v2") + set(OSS_SDK_V2_SOURCE + "BUNDLED" + CACHE STRING "Dependency source for OSS SDK v2" FORCE) + set(OSS_SDK_V2_SOURCE + "BUNDLED" + PARENT_SCOPE) + endif() + endif() endfunction() function(paimon_apply_dependency_source_defaults) @@ -597,6 +626,8 @@ function(paimon_get_dependency_compat_target DEPENDENCY_NAME OUT_VAR) set(_target tbb) elseif("${DEPENDENCY_NAME}" STREQUAL "Avro") set(_target avro) + elseif("${DEPENDENCY_NAME}" STREQUAL "OSS_SDK_V2") + set(_target alibabacloud_oss_v2::oss) else() set(_target "${DEPENDENCY_NAME}") endif() @@ -671,6 +702,8 @@ macro(paimon_build_dependency DEPENDENCY_NAME) build_glog() elseif("${DEPENDENCY_NAME}" STREQUAL "Avro") build_avro() + elseif("${DEPENDENCY_NAME}" STREQUAL "OSS_SDK_V2") + build_oss_sdk_v2() elseif("${DEPENDENCY_NAME}" STREQUAL "GTest") build_gtest() elseif("${DEPENDENCY_NAME}" STREQUAL "benchmark") @@ -1359,6 +1392,89 @@ macro(build_jindosdk_nextarch) add_dependencies(jindosdk::nextarch jindosdk-nextarch_ep) endmacro() +macro(build_oss_sdk_v2) + message(STATUS "Building Alibaba Cloud OSS C++ SDK v2 from source") + find_package(CURL REQUIRED) + find_package(Threads REQUIRED) + + set(OSS_SDK_V2_PREFIX "${CMAKE_CURRENT_BINARY_DIR}/oss_sdk_v2_ep-install") + set(OSS_SDK_V2_INCLUDE_DIR "${OSS_SDK_V2_PREFIX}/include") + set(OSS_SDK_V2_INSTALL_LIBDIR "${CMAKE_INSTALL_LIBDIR}") + set(OSS_SDK_V2_LIB_DIR "${OSS_SDK_V2_PREFIX}/${OSS_SDK_V2_INSTALL_LIBDIR}") + set(OSS_SDK_V2_STATIC_LIB + "${OSS_SDK_V2_LIB_DIR}/${CMAKE_STATIC_LIBRARY_PREFIX}alibabacloud-oss-cpp-sdk-v2${CMAKE_STATIC_LIBRARY_SUFFIX}" + ) + + set(OSS_SDK_V2_CMAKE_ARGS + ${EP_COMMON_CMAKE_ARGS} + "-DCMAKE_INSTALL_PREFIX=${OSS_SDK_V2_PREFIX}" + "-DCMAKE_INSTALL_LIBDIR=${OSS_SDK_V2_INSTALL_LIBDIR}" + -DCMAKE_PARENT_CXX_STANDARD=17 + -DBUILD_SHARED_LIBS=OFF + -DBUILD_TESTS=OFF + -DBUILD_SAMPLES=OFF + -DENABLE_RTTI=OFF + -DUSE_CURL_TRANSPORT=ON + -DUSE_SYSTEM_CURL=ON + -DUSE_SYSTEM_OPENSSL=OFF + -DUSE_SYSTEM_MBEDTLS=OFF + -DUSE_SYSTEM_TINYXML2=OFF + -DUSE_STD_EXPECTED=OFF + -DENABLE_ENCRYPTION=OFF) + set(OSS_SDK_V2_CURL_INCLUDE_DIR "${CURL_INCLUDE_DIR}") + if(NOT OSS_SDK_V2_CURL_INCLUDE_DIR AND CURL_INCLUDE_DIRS) + list(GET CURL_INCLUDE_DIRS 0 OSS_SDK_V2_CURL_INCLUDE_DIR) + endif() + set(OSS_SDK_V2_CURL_LIBRARY "${CURL_LIBRARY_RELEASE}") + if(NOT OSS_SDK_V2_CURL_LIBRARY) + set(OSS_SDK_V2_CURL_LIBRARY "${CURL_LIBRARY}") + endif() + if(TARGET CURL::libcurl) + if(NOT OSS_SDK_V2_CURL_INCLUDE_DIR) + get_target_property(OSS_SDK_V2_CURL_INCLUDE_DIR CURL::libcurl + INTERFACE_INCLUDE_DIRECTORIES) + endif() + if(NOT OSS_SDK_V2_CURL_LIBRARY) + foreach(CURL_CONFIG RELEASE RELWITHDEBINFO DEBUG NOCONFIG) + get_target_property(OSS_SDK_V2_CURL_LIBRARY CURL::libcurl + "IMPORTED_LOCATION_${CURL_CONFIG}") + if(OSS_SDK_V2_CURL_LIBRARY) + break() + endif() + endforeach() + endif() + if(NOT OSS_SDK_V2_CURL_LIBRARY) + get_target_property(OSS_SDK_V2_CURL_LIBRARY CURL::libcurl IMPORTED_LOCATION) + endif() + endif() + if(OSS_SDK_V2_CURL_INCLUDE_DIR AND OSS_SDK_V2_CURL_LIBRARY) + list(APPEND + OSS_SDK_V2_CMAKE_ARGS + "-DCURL_INCLUDE_DIR=${OSS_SDK_V2_CURL_INCLUDE_DIR}" + "-DCURL_LIBRARY=${OSS_SDK_V2_CURL_LIBRARY}" + "-DCURL_LIBRARY_RELEASE=${OSS_SDK_V2_CURL_LIBRARY}") + endif() + + externalproject_add(oss_sdk_v2_ep + ${EP_COMMON_OPTIONS} + URL ${OSS_SDK_V2_SOURCE_URL} + URL_HASH "SHA256=${PAIMON_OSS_SDK_V2_BUILD_SHA256_CHECKSUM}" + CMAKE_ARGS ${OSS_SDK_V2_CMAKE_ARGS} ${THIRDPARTY_LOG_OPTIONS} + BUILD_BYPRODUCTS "${OSS_SDK_V2_STATIC_LIB}") + + file(MAKE_DIRECTORY "${OSS_SDK_V2_INCLUDE_DIR}") + file(MAKE_DIRECTORY "${OSS_SDK_V2_LIB_DIR}") + + add_library(alibabacloud_oss_v2::oss STATIC IMPORTED) + set_target_properties(alibabacloud_oss_v2::oss + PROPERTIES IMPORTED_LOCATION "${OSS_SDK_V2_STATIC_LIB}" + INTERFACE_INCLUDE_DIRECTORIES + "${OSS_SDK_V2_INCLUDE_DIR}") + target_link_libraries(alibabacloud_oss_v2::oss INTERFACE CURL::libcurl + Threads::Threads) + add_dependencies(alibabacloud_oss_v2::oss oss_sdk_v2_ep) +endmacro() + macro(build_protobuf) message(STATUS "Building protobuf from source") set(PROTOBUF_PREFIX "${CMAKE_CURRENT_BINARY_DIR}/protobuf_ep-install") @@ -1967,6 +2083,9 @@ if(PAIMON_ENABLE_JINDO) build_jindosdk_c() build_jindosdk_nextarch() endif() +if(PAIMON_ENABLE_OSS) + resolve_dependency(OSS_SDK_V2) +endif() if(PAIMON_ENABLE_S3) include(BuildAwsAuth) build_aws_auth() diff --git a/docs/source/build_system.rst b/docs/source/build_system.rst index a2a1a9e7..40964f6f 100644 --- a/docs/source/build_system.rst +++ b/docs/source/build_system.rst @@ -106,6 +106,8 @@ Paimon provides a set of built-in optional plugins that you can link to as neede - ``Paimon::paimon_local_file_system_shared`` - ``Paimon::paimon_jindo_file_system_shared`` + - ``Paimon::paimon_oss_file_system_shared`` + - ``Paimon::paimon_s3_file_system_shared`` - Index plugins: diff --git a/docs/source/building.rst b/docs/source/building.rst index 466d461b..cfc64652 100644 --- a/docs/source/building.rst +++ b/docs/source/building.rst @@ -178,6 +178,8 @@ 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 +* ``-DPAIMON_ENABLE_OSS=ON``: Support for Alibaba Cloud OSS through OSS SDK V2 +* ``-DPAIMON_ENABLE_S3=ON``: Support for Amazon S3-compatible filesystems * ``-DPAIMON_ENABLE_LUMINA=ON``: Support for the Lumina vector index. Requires Linux ``x86_64``; see :ref:`cpp-building-platforms`. * ``-DPAIMON_ENABLE_LUCENE=ON``: Support for Lucene full-text search indexes diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt index a9810424..2ee0a336 100644 --- a/src/paimon/CMakeLists.txt +++ b/src/paimon/CMakeLists.txt @@ -196,7 +196,7 @@ if(PAIMON_ENABLE_S3 OR PAIMON_ENABLE_REST) list(APPEND PAIMON_COMMON_SRCS common/utils/http_client.cpp) set(PAIMON_CURL_LINK_LIBS CURL::libcurl) endif() -if(PAIMON_ENABLE_S3) +if(PAIMON_ENABLE_OSS OR PAIMON_ENABLE_S3) list(APPEND PAIMON_COMMON_SRCS common/fs/object_store_file_system.cpp) endif() @@ -938,7 +938,7 @@ if(PAIMON_BUILD_TESTS) endif() set(PAIMON_OBJECT_STORE_FS_TEST_SOURCES) - if(PAIMON_ENABLE_S3) + if(PAIMON_ENABLE_OSS OR PAIMON_ENABLE_S3) list(APPEND PAIMON_OBJECT_STORE_FS_TEST_SOURCES common/fs/object_store_file_system_test.cpp) endif() diff --git a/src/paimon/common/fs/object_store_file_system.cpp b/src/paimon/common/fs/object_store_file_system.cpp index ae44835e..26cbe2aa 100644 --- a/src/paimon/common/fs/object_store_file_system.cpp +++ b/src/paimon/common/fs/object_store_file_system.cpp @@ -30,6 +30,7 @@ #include "paimon/common/utils/path_util.h" namespace paimon { + namespace { constexpr int64_t kMaxReadAheadMemory = 64LL * 1024LL * 1024LL; diff --git a/src/paimon/common/fs/object_store_file_system.h b/src/paimon/common/fs/object_store_file_system.h index a1313b85..a40ef15c 100644 --- a/src/paimon/common/fs/object_store_file_system.h +++ b/src/paimon/common/fs/object_store_file_system.h @@ -19,7 +19,12 @@ #pragma once +#include + +#include +#include #include +#include #include #include #include @@ -31,6 +36,71 @@ namespace paimon { +class ObjectStoreFileSystemUtils { + public: + static inline int64_t ParseModificationTime(const std::string& value) { + std::tm parsed_time{}; + const char* current = strptime(value.c_str(), "%Y-%m-%dT%H:%M:%S", &parsed_time); + if (current != nullptr) { + int32_t milliseconds = 0; + int32_t fraction_digits = 0; + if (*current == '.') { + ++current; + const char* fraction_begin = current; + while (std::isdigit(static_cast(*current))) { + if (fraction_digits < 3) { + milliseconds = milliseconds * 10 + (*current - '0'); + } + ++fraction_digits; + ++current; + } + if (current == fraction_begin) { + current = nullptr; + } + while (fraction_digits < 3) { + milliseconds *= 10; + ++fraction_digits; + } + } + + int32_t timezone_offset_seconds = 0; + bool valid_timezone = false; + if (current != nullptr && *current == 'Z' && current[1] == '\0') { + valid_timezone = true; + } else if (current != nullptr && (*current == '+' || *current == '-') && + std::isdigit(static_cast(current[1])) && + std::isdigit(static_cast(current[2])) && current[3] == ':' && + std::isdigit(static_cast(current[4])) && + std::isdigit(static_cast(current[5])) && current[6] == '\0') { + int32_t hours = (current[1] - '0') * 10 + current[2] - '0'; + int32_t minutes = (current[4] - '0') * 10 + current[5] - '0'; + if (hours <= 23 && minutes <= 59) { + timezone_offset_seconds = (hours * 60 + minutes) * 60; + if (*current == '-') { + timezone_offset_seconds = -timezone_offset_seconds; + } + valid_timezone = true; + } + } + + if (valid_timezone) { + errno = 0; + time_t seconds = timegm(&parsed_time); + if (seconds != static_cast(-1) || errno != EOVERFLOW) { + int64_t utc_seconds = static_cast(seconds) - timezone_offset_seconds; + return utc_seconds * 1000 + milliseconds; + } + } + } + + time_t seconds = curl_getdate(value.c_str(), nullptr); + if (seconds == static_cast(-1)) { + return FileStatus::kUnknownModificationTime; + } + return static_cast(seconds) * 1000; + } +}; + struct ObjectStorePath { std::string bucket; std::string key; @@ -39,7 +109,7 @@ struct ObjectStorePath { struct ObjectMetadata { std::string key; int64_t size = 0; - int64_t modification_time = 0; + int64_t modification_time = FileStatus::kUnknownModificationTime; }; struct ListObjectsResult { diff --git a/src/paimon/common/fs/object_store_file_system_test.cpp b/src/paimon/common/fs/object_store_file_system_test.cpp index 824060f8..6b8b5650 100644 --- a/src/paimon/common/fs/object_store_file_system_test.cpp +++ b/src/paimon/common/fs/object_store_file_system_test.cpp @@ -33,6 +33,17 @@ namespace paimon::test { namespace { +TEST(ObjectStoreFileSystemTest, TestParseModificationTime) { + ASSERT_EQ(1704067200000, + ObjectStoreFileSystemUtils::ParseModificationTime("2024-01-01T00:00:00.000Z")); + ASSERT_EQ(1704067200123, + ObjectStoreFileSystemUtils::ParseModificationTime("2024-01-01T08:00:00.123+08:00")); + ASSERT_EQ(1704067200000, + ObjectStoreFileSystemUtils::ParseModificationTime("Mon, 01 Jan 2024 00:00:00 GMT")); + ASSERT_EQ(FileStatus::kUnknownModificationTime, + ObjectStoreFileSystemUtils::ParseModificationTime("not-a-timestamp")); +} + using Range = std::pair; class MockObjectStoreClient : public ObjectStoreClient { diff --git a/src/paimon/fs/oss/CMakeLists.txt b/src/paimon/fs/oss/CMakeLists.txt new file mode 100644 index 00000000..fb0ea674 --- /dev/null +++ b/src/paimon/fs/oss/CMakeLists.txt @@ -0,0 +1,50 @@ +# 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_OSS) + add_paimon_lib(paimon_oss_file_system + SOURCES + oss_file_system.cpp + oss_file_system_factory.cpp + EXTRA_INCLUDES + ${OSS_SDK_V2_INCLUDE_DIR} + DEPENDENCIES + paimon_shared + CURL::libcurl + STATIC_LINK_LIBS + alibabacloud_oss_v2::oss + fmt + SHARED_LINK_LIBS + paimon_shared + SHARED_LINK_FLAGS + ${PAIMON_VERSION_SCRIPT_FLAGS}) + + add_dependencies(paimon_oss_file_system_objlib oss_sdk_v2_ep) + + if(PAIMON_BUILD_TESTS) + add_paimon_test(oss_file_system_test + SOURCES + oss_file_system_test.cpp + EXTRA_INCLUDES + ${OSS_SDK_V2_INCLUDE_DIR} + STATIC_LINK_LIBS + paimon_shared + test_utils_static + ${PAIMON_OSS_FILE_SYSTEM_STATIC_LINK_LIBS} + ${GTEST_LINK_TOOLCHAIN}) + endif() +endif() diff --git a/src/paimon/fs/oss/oss_file_system.cpp b/src/paimon/fs/oss/oss_file_system.cpp new file mode 100644 index 00000000..bc283cc8 --- /dev/null +++ b/src/paimon/fs/oss/oss_file_system.cpp @@ -0,0 +1,192 @@ +/* + * 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/oss/oss_file_system.h" + +#include +#include +#include +#include + +#include "alibabacloud/oss2/OSSClient.h" +#include "alibabacloud/oss2/Operation.h" +#include "alibabacloud/oss2/Types.h" +#include "alibabacloud/oss2/io/ByteWriter.h" +#include "alibabacloud/oss2/models/BucketBasic.h" +#include "alibabacloud/oss2/models/ObjectBasic.h" +#include "fmt/format.h" +#include "paimon/executor.h" + +namespace paimon::oss { +namespace { + +namespace oss2 = alibabacloud::oss2; + +constexpr int32_t kMaxKeysPerRequest = 1000; + +bool IsNotFoundError(const oss2::OperationError& error) { + return error.getStatusCode() == 404 || error.getCode() == "NoSuchKey" || + error.getCode() == "NoSuchBucket" || error.getCode() == "NotFound"; +} + +Status ToPaimonStatus(const oss2::OperationError& error, const std::string& operation, + const ObjectStorePath& path) { + std::string message = fmt::format("OSS {} 'oss://{}/{}' failed: code={}, status={}, message={}", + operation, path.bucket, path.key, error.getCode(), + error.getStatusCode(), error.getMessage()); + if (!error.getRequestId().empty()) { + message += fmt::format(", request_id={}", error.getRequestId()); + } + if (IsNotFoundError(error)) { + return Status::NotExist(message); + } + if (error.getCode() == "RequestCanceled") { + return Status::Cancelled(message); + } + return Status::IOError(message); +} + +class OssObjectStoreClient : public ObjectStoreClient, + public std::enable_shared_from_this { + public: + OssObjectStoreClient(std::string bucket, std::shared_ptr client, + std::unique_ptr executor) + : bucket_(std::move(bucket)), client_(std::move(client)), executor_(std::move(executor)) {} + + Result HeadObject(const ObjectStorePath& path) const override { + PAIMON_RETURN_NOT_OK(ValidateBucket(path)); + oss2::models::HeadObjectRequest request; + request.setBucket(path.bucket).setKey(path.key); + oss2::HeadObjectOutcome outcome = client_->headObject(request); + if (!outcome.has_value()) { + return ToPaimonStatus(outcome.error(), "HeadObject", path); + } + const oss2::models::HeadObjectResult& result = outcome.value(); + if (result.getContentLength() < 0) { + return Status::IOError("OSS HeadObject response is missing Content-Length"); + } + return ObjectMetadata{ + path.key, result.getContentLength(), + ObjectStoreFileSystemUtils::ParseModificationTime(result.getLastModified())}; + } + + Result ListObjects(const ObjectStorePath& path, + const std::string& continuation_token, + int32_t max_keys) const override { + PAIMON_RETURN_NOT_OK(ValidateBucket(path)); + oss2::models::ListObjectsV2Request request; + request.setBucket(path.bucket).setPrefix(path.key).setDelimiter("/"); + if (!continuation_token.empty()) { + request.setContinuationToken(continuation_token); + } + if (max_keys > 0) { + // OSS limits ListObjectsV2 requests to 1000 keys. + request.setMaxKeys(std::min(max_keys, kMaxKeysPerRequest)); + } + oss2::ListObjectsV2Outcome outcome = client_->listObjectsV2(request); + if (!outcome.has_value()) { + return ToPaimonStatus(outcome.error(), "ListObjectsV2", path); + } + const oss2::models::ListObjectsV2Result& value = outcome.value(); + ListObjectsResult result; + result.objects.reserve(value.getContents().size()); + for (const oss2::models::ObjectSummary& object : value.getContents()) { + result.objects.push_back(ObjectMetadata{ + object.key, object.size, + ObjectStoreFileSystemUtils::ParseModificationTime(object.lastModified)}); + } + result.common_prefixes.reserve(value.getCommonPrefixes().size()); + for (const oss2::models::CommonPrefix& prefix : value.getCommonPrefixes()) { + result.common_prefixes.push_back(prefix.prefix); + } + result.is_truncated = value.getIsTruncated(); + result.continuation_token = value.getNextContinuationToken(); + return result; + } + + Result GetObjectRange(const ObjectStorePath& path, int64_t offset, int64_t size, + char* buffer) const override { + PAIMON_RETURN_NOT_OK(ValidateBucket(path)); + if (size == 0) { + return 0; + } + auto writer = std::make_shared>(); + oss2::SinkFactory sink; + sink.isOneShot = false; + sink.supplier = [buffer, size, writer](int64_t, const oss2::HeaderCollection&) { + auto memory_writer = std::make_shared( + reinterpret_cast(buffer), static_cast(size)); + *writer = memory_writer; + return memory_writer; + }; + oss2::models::GetObjectRequest request; + request.setBucket(path.bucket) + .setKey(path.key) + .setRange(fmt::format("bytes={}-{}", offset, offset + size - 1)) + .setRangeBehavior("standard") + .setSinkFactory(std::move(sink)); + oss2::GetObjectOutcome outcome = client_->getObject(request); + if (!outcome.has_value()) { + return ToPaimonStatus(outcome.error(), "GetObject", path); + } + int64_t written = + *writer ? static_cast((*writer)->written()) : static_cast(0); + if (written != size) { + return Status::IOError( + fmt::format("OSS GetObject read {} bytes for oss://{}/{}, expected {}", written, + path.bucket, path.key, size)); + } + return written; + } + + void GetObjectRangeAsync(const ObjectStorePath& path, int64_t offset, int64_t size, + char* buffer, std::function&& callback) const override { + std::shared_ptr self = shared_from_this(); + executor_->Add([self = std::move(self), path, offset, size, buffer, + callback = std::move(callback)]() mutable { + Result result = self->GetObjectRange(path, offset, size, buffer); + callback(result.ok() ? Status::OK() : result.status()); + }); + } + + private: + Status ValidateBucket(const ObjectStorePath& path) const { + if (path.bucket != bucket_) { + return Status::Invalid( + fmt::format("OSS file system for bucket '{}' cannot access " + "'oss://{}/{}'", + bucket_, path.bucket, path.key)); + } + return Status::OK(); + } + + std::string bucket_; + std::shared_ptr client_; + std::unique_ptr executor_; +}; + +} // namespace + +OssFileSystem::OssFileSystem(std::string bucket, std::shared_ptr client, + std::unique_ptr executor) + : ObjectStoreFileSystem("oss", std::make_shared( + std::move(bucket), std::move(client), std::move(executor))) { +} + +} // namespace paimon::oss diff --git a/src/paimon/fs/oss/oss_file_system.h b/src/paimon/fs/oss/oss_file_system.h new file mode 100644 index 00000000..0749718e --- /dev/null +++ b/src/paimon/fs/oss/oss_file_system.h @@ -0,0 +1,51 @@ +/* + * 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 "paimon/common/fs/object_store_file_system.h" +#include "paimon/executor.h" + +namespace alibabacloud::oss2 { +class OSSClient; +} + +namespace paimon::oss { + +inline constexpr char kOssAccessKeyIdOption[] = "fs.oss.accessKeyId"; +inline constexpr char kOssAccessKeySecretOption[] = "fs.oss.accessKeySecret"; +inline constexpr char kOssEndpointOption[] = "fs.oss.endpoint"; +inline constexpr char kOssRegionOption[] = "fs.oss.region"; +inline constexpr char kOssSignatureVersionOption[] = "fs.oss.signatureVersion"; +inline constexpr char kOssSecurityTokenOption[] = "fs.oss.securityToken"; +inline constexpr char kOssSessionTokenOption[] = "fs.oss.sessionToken"; +inline constexpr char kOssUsePathStyleOption[] = "fs.oss.usePathStyle"; +inline constexpr char kOssExecutorThreadCountOption[] = "fs.oss.executor.thread-count"; + +class OssFileSystem : public ObjectStoreFileSystem { + public: + OssFileSystem(std::string bucket, std::shared_ptr client, + std::unique_ptr executor); + ~OssFileSystem() override = default; +}; + +} // namespace paimon::oss diff --git a/src/paimon/fs/oss/oss_file_system_factory.cpp b/src/paimon/fs/oss/oss_file_system_factory.cpp new file mode 100644 index 00000000..327586d0 --- /dev/null +++ b/src/paimon/fs/oss/oss_file_system_factory.cpp @@ -0,0 +1,234 @@ +/* + * 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/oss/oss_file_system_factory.h" + +#include +#include +#include +#include +#include +#include + +#include "alibabacloud/oss2/ClientConfiguration.h" +#include "alibabacloud/oss2/OSSClient.h" +#include "alibabacloud/oss2/credentials/CredentialsProvider.h" +#include "fmt/format.h" +#include "paimon/common/utils/path_util.h" +#include "paimon/common/utils/string_utils.h" +#include "paimon/factories/factory.h" +#include "paimon/fs/oss/oss_file_system.h" + +namespace paimon::oss { +namespace { + +namespace oss2 = alibabacloud::oss2; + +constexpr std::string_view kOssOptionPrefix = "fs.oss."; +constexpr char kEndpointPrefix[] = "oss-"; +constexpr char kEndpointSuffix[] = ".aliyuncs.com"; +constexpr char kDualStackEndpointSuffix[] = ".oss.aliyuncs.com"; +constexpr char kInternalEndpointSuffix[] = "-internal"; + +bool IsValidRegion(const std::string& region) { + size_t first_separator = region.find('-'); + if (first_separator < 2 || first_separator > 3 || first_separator + 1 == region.size()) { + return false; + } + for (size_t i = 0; i < first_separator; ++i) { + if (!std::islower(static_cast(region[i]))) { + return false; + } + } + bool previous_was_separator = true; + for (size_t i = first_separator + 1; i < region.size(); ++i) { + char value = region[i]; + if (value == '-') { + if (previous_was_separator || i + 1 == region.size()) { + return false; + } + previous_was_separator = true; + } else if (std::islower(static_cast(value)) || + std::isdigit(static_cast(value))) { + previous_was_separator = false; + } else { + return false; + } + } + return !StringUtils::EndsWith(region, "-dualstack") && !StringUtils::EndsWith(region, "-pub"); +} + +std::string GetBucketOptionKey(const std::string& bucket, std::string_view option) { + return fmt::format("fs.oss.bucket.{}.{}", bucket, option.substr(kOssOptionPrefix.size())); +} + +const std::string* FindOption(const std::map& options, + const std::string& bucket, std::string_view option, + std::string* option_key) { + std::string bucket_option_key = GetBucketOptionKey(bucket, option); + auto bucket_option = options.find(bucket_option_key); + if (bucket_option != options.end()) { + *option_key = std::move(bucket_option_key); + return &bucket_option->second; + } + option_key->assign(option); + auto global_option = options.find(*option_key); + return global_option == options.end() ? nullptr : &global_option->second; +} + +std::string GetOption(const std::map& options, const std::string& bucket, + std::string_view option) { + std::string option_key; + const std::string* value = FindOption(options, bucket, option, &option_key); + return value == nullptr ? "" : *value; +} + +Result GetRequiredOption(const std::map& options, + const std::string& bucket, std::string_view option) { + std::string option_key; + const std::string* value = FindOption(options, bucket, option, &option_key); + if (value == nullptr || value->empty()) { + return Status::Invalid(fmt::format("OSS option '{}' must not be empty", option_key)); + } + return *value; +} + +Result> CreateExecutor(const std::map& options, + const std::string& bucket) { + std::string option_key; + const std::string* value = + FindOption(options, bucket, kOssExecutorThreadCountOption, &option_key); + if (value == nullptr) { + return CreateDefaultExecutor(); + } + std::optional thread_count = StringUtils::StringToValue(*value); + if (!thread_count.has_value() || *thread_count == 0) { + return Status::Invalid(fmt::format( + "OSS executor thread count for option '{}' must be greater than 0", option_key)); + } + return CreateDefaultExecutor(*thread_count); +} + +std::string NormalizeEndpoint(std::string endpoint) { + if (!endpoint.empty() && endpoint.find("://") == std::string::npos) { + endpoint = "https://" + endpoint; + } + return endpoint; +} + +std::string InferRegion(std::string endpoint) { + size_t scheme = endpoint.find("://"); + if (scheme != std::string::npos) { + endpoint.erase(0, scheme + 3); + } + size_t slash = endpoint.find('/'); + if (slash != std::string::npos) { + endpoint.erase(slash); + } + size_t port_separator = endpoint.rfind(':'); + if (port_separator != std::string::npos && endpoint.find(':') == port_separator) { + std::optional port = + StringUtils::StringToValue(endpoint.substr(port_separator + 1)); + if (port.has_value()) { + endpoint.erase(port_separator); + } + } + + std::string region; + if (StringUtils::StartsWith(endpoint, kEndpointPrefix) && + StringUtils::EndsWith(endpoint, kEndpointSuffix)) { + region = endpoint.substr( + sizeof(kEndpointPrefix) - 1, + endpoint.size() - (sizeof(kEndpointPrefix) - 1) - (sizeof(kEndpointSuffix) - 1)); + if (StringUtils::EndsWith(region, kInternalEndpointSuffix)) { + region.erase(region.size() - (sizeof(kInternalEndpointSuffix) - 1)); + } + } else if (StringUtils::EndsWith(endpoint, kDualStackEndpointSuffix)) { + region = endpoint.substr(0, endpoint.size() - (sizeof(kDualStackEndpointSuffix) - 1)); + } + return IsValidRegion(region) ? region : ""; +} + +} // namespace + +const char OssFileSystemFactory::IDENTIFIER[] = "oss"; + +Result> OssFileSystemFactory::Create( + const std::string& path, const std::map& options) const { + PAIMON_ASSIGN_OR_RAISE(Path parsed_path, PathUtil::ToPath(path)); + if (parsed_path.scheme != "oss" || parsed_path.authority.empty()) { + return Status::Invalid(fmt::format("invalid OSS path '{}'", path)); + } + const std::string& bucket = parsed_path.authority; + PAIMON_ASSIGN_OR_RAISE(std::string access_key_id, + GetRequiredOption(options, bucket, kOssAccessKeyIdOption)); + PAIMON_ASSIGN_OR_RAISE(std::string access_key_secret, + GetRequiredOption(options, bucket, kOssAccessKeySecretOption)); + std::string endpoint = GetOption(options, bucket, kOssEndpointOption); + std::string region = GetOption(options, bucket, kOssRegionOption); + if (region.empty()) { + region = InferRegion(endpoint); + } + if (endpoint.empty() && region.empty()) { + return Status::Invalid("OSS endpoint or region must be configured"); + } + std::string signature_version = GetOption(options, bucket, kOssSignatureVersionOption); + if (!signature_version.empty() && signature_version != "v1" && signature_version != "v4") { + return Status::Invalid( + fmt::format("invalid OSS signature version '{}'", signature_version)); + } + if (region.empty() && signature_version != "v1") { + return Status::Invalid( + "OSS region must be configured when the endpoint does not identify a region"); + } + std::string security_token = GetOption(options, bucket, kOssSecurityTokenOption); + if (security_token.empty()) { + security_token = GetOption(options, bucket, kOssSessionTokenOption); + } + + oss2::ClientConfiguration config = oss2::ClientConfiguration::loadDefault(); + if (!endpoint.empty()) { + config.endpoint = NormalizeEndpoint(endpoint); + } + if (!region.empty()) { + config.region = region; + } + if (!signature_version.empty()) { + config.signatureVersion = signature_version; + } + config.userAgent = "paimon-cpp"; + config.credentialsProvider = std::make_shared( + std::move(access_key_id), std::move(access_key_secret), std::move(security_token)); + std::string path_style = GetOption(options, bucket, kOssUsePathStyleOption); + if (!path_style.empty()) { + std::optional value = StringUtils::StringToValue(path_style); + if (!value.has_value()) { + return Status::Invalid(fmt::format("invalid boolean value '{}' for OSS option '{}'", + path_style, kOssUsePathStyleOption)); + } + config.usePathStyle = *value; + } + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr executor, CreateExecutor(options, bucket)); + return std::make_unique(bucket, std::make_shared(config), + std::move(executor)); +} + +REGISTER_PAIMON_FACTORY(OssFileSystemFactory); + +} // namespace paimon::oss diff --git a/src/paimon/fs/oss/oss_file_system_factory.h b/src/paimon/fs/oss/oss_file_system_factory.h new file mode 100644 index 00000000..fe92715e --- /dev/null +++ b/src/paimon/fs/oss/oss_file_system_factory.h @@ -0,0 +1,38 @@ +/* + * 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 "paimon/fs/file_system_factory.h" + +namespace paimon::oss { + +class OssFileSystemFactory : public FileSystemFactory { + public: + static const char IDENTIFIER[]; + + const char* Identifier() const override { + return IDENTIFIER; + } + + Result> Create( + const std::string& path, const std::map& options) const override; +}; + +} // namespace paimon::oss diff --git a/src/paimon/fs/oss/oss_file_system_test.cpp b/src/paimon/fs/oss/oss_file_system_test.cpp new file mode 100644 index 00000000..fbd847f8 --- /dev/null +++ b/src/paimon/fs/oss/oss_file_system_test.cpp @@ -0,0 +1,327 @@ +/* + * 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/oss/oss_file_system.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "alibabacloud/oss2/ClientConfiguration.h" +#include "alibabacloud/oss2/OSSClient.h" +#include "alibabacloud/oss2/credentials/CredentialsProvider.h" +#include "alibabacloud/oss2/io/ByteWriter.h" +#include "alibabacloud/oss2/transport/HttpTransport.h" +#include "gtest/gtest.h" +#include "paimon/fs/oss/oss_file_system_factory.h" +#include "paimon/testing/utils/testharness.h" + +namespace paimon::oss { +namespace { + +namespace oss2 = alibabacloud::oss2; + +class MockHttpTransport : public oss2::HttpTransport { + public: + oss2::ResponseResult send(std::unique_ptr& request, + const oss2::RequestOptions& options) override { + requests_.emplace_back(std::make_unique(*request)); + if (responses_.empty()) { + return oss2::TransportError{std::make_error_code(std::errc::no_message_available), "", + ""}; + } + std::unique_ptr response = std::move(responses_.front()); + responses_.erase(responses_.begin()); + if (response->statusCode / 100 == 2 && options.sinkFactory.has_value() && + response->body != nullptr) { + int64_t content_length = -1; + auto content_length_header = response->headers.find("Content-Length"); + if (content_length_header != response->headers.end()) { + content_length = std::stoll(content_length_header->second); + } + std::shared_ptr sink = + options.sinkFactory.value()(content_length, response->headers); + std::ostringstream body; + body << response->body->rdbuf(); + const std::string data = body.str(); + sink->write(reinterpret_cast(data.data()), data.size()); + response->body.reset(); + } + return response; + } + + std::string getName() const override { + return "MockHttpTransport"; + } + + void AddResponse(int status_code, oss2::HeaderCollection headers, std::string body = "") { + std::shared_ptr response_body; + if (!body.empty()) { + response_body = std::make_shared(std::move(body)); + } + responses_.emplace_back(std::make_unique(oss2::ResponseMessage{ + status_code, "", std::move(headers), std::move(response_body), nullptr})); + } + + std::vector> responses_; + std::vector> requests_; +}; + +std::unique_ptr CreateFileSystem( + const std::shared_ptr& transport) { + oss2::ClientConfiguration config = oss2::ClientConfiguration::loadDefault(); + config.region = "cn-hangzhou"; + config.credentialsProvider = + std::make_shared("access-key", "secret-key"); + config.httpTransport = transport; + return std::make_unique("bucket", std::make_shared(config), + CreateDefaultExecutor()); +} + +} // namespace + +TEST(OssFileSystemFactoryTest, TestOptionValidation) { + OssFileSystemFactory factory; + std::map options; + ASSERT_NOK(factory.Create("s3://bucket/key", options)); + ASSERT_NOK(factory.Create("oss://bucket/key", options)); + + options[kOssAccessKeyIdOption] = "access-key"; + options[kOssAccessKeySecretOption] = "secret-key"; + options[kOssEndpointOption] = "oss-cn-hangzhou.aliyuncs.com"; + options[kOssUsePathStyleOption] = "treu"; + ASSERT_NOK(factory.Create("oss://bucket/key", options)); + + options[kOssUsePathStyleOption] = "false"; + options[kOssSignatureVersionOption] = "v2"; + ASSERT_NOK(factory.Create("oss://bucket/key", options)); + + options[kOssSignatureVersionOption] = "v4"; + options[kOssExecutorThreadCountOption] = "0"; + ASSERT_NOK(factory.Create("oss://bucket/key", options)); + + options[kOssExecutorThreadCountOption] = "4"; + ASSERT_OK(factory.Create("oss://bucket/key", options)); + + options[kOssEndpointOption] = ""; + options[kOssRegionOption] = "cn-hangzhou"; + ASSERT_OK(factory.Create("oss://bucket/key", options)); +} + +TEST(OssFileSystemFactoryTest, TestBucketOptionsOverrideGlobalOptions) { + OssFileSystemFactory factory; + std::map options = { + {kOssAccessKeyIdOption, ""}, + {kOssAccessKeySecretOption, ""}, + {kOssEndpointOption, ""}, + {"fs.oss.bucket.bucket.accessKeyId", "access-key"}, + {"fs.oss.bucket.bucket.accessKeySecret", "secret-key"}, + {"fs.oss.bucket.bucket.endpoint", "oss-cn-hangzhou.aliyuncs.com"}, + }; + ASSERT_OK(factory.Create("oss://bucket/key", options)); +} + +TEST(OssFileSystemFactoryTest, TestBucketOptionErrorReportsBucketKey) { + OssFileSystemFactory factory; + std::map options = { + {kOssAccessKeyIdOption, "access-key"}, + {kOssAccessKeySecretOption, "secret-key"}, + {kOssEndpointOption, "oss-cn-hangzhou.aliyuncs.com"}, + {"fs.oss.bucket.bucket.accessKeyId", ""}, + }; + ASSERT_NOK_WITH_MSG(factory.Create("oss://bucket/key", options), + "fs.oss.bucket.bucket.accessKeyId"); +} + +TEST(OssFileSystemFactoryTest, TestEndpointRegionValidation) { + OssFileSystemFactory factory; + std::map options = { + {kOssAccessKeyIdOption, "access-key"}, + {kOssAccessKeySecretOption, "secret-key"}, + }; + + options[kOssEndpointOption] = "oss-cn-hangzhou.aliyuncs.com"; + ASSERT_OK(factory.Create("oss://bucket/key", options)); + + options[kOssEndpointOption] = "oss-cn-hangzhou-internal.aliyuncs.com"; + ASSERT_OK(factory.Create("oss://bucket/key", options)); + + options[kOssEndpointOption] = "oss-ap-southeast-1.aliyuncs.com:443"; + ASSERT_OK(factory.Create("oss://bucket/key", options)); + + options[kOssEndpointOption] = "cn-hangzhou.oss.aliyuncs.com"; + ASSERT_OK(factory.Create("oss://bucket/key", options)); + + options[kOssEndpointOption] = "oss-accelerate.aliyuncs.com"; + ASSERT_NOK_WITH_MSG(factory.Create("oss://bucket/key", options), "OSS region must be"); + + options[kOssEndpointOption] = "oss-accelerate-overseas.aliyuncs.com"; + ASSERT_NOK_WITH_MSG(factory.Create("oss://bucket/key", options), "OSS region must be"); + + options[kOssEndpointOption] = "oss.example.com"; + ASSERT_NOK_WITH_MSG(factory.Create("oss://bucket/key", options), "OSS region must be"); + + options[kOssRegionOption] = "cn-hangzhou"; + ASSERT_OK(factory.Create("oss://bucket/key", options)); + + options.erase(kOssRegionOption); + options[kOssSignatureVersionOption] = "v1"; + ASSERT_OK(factory.Create("oss://bucket/key", options)); +} + +TEST(OssFileSystemTest, TestHeadObjectParsesMetadata) { + std::shared_ptr transport = std::make_shared(); + transport->AddResponse(200, {{"Content-Length", "3"}, + {"Last-Modified", "not-a-timestamp"}, + {"x-oss-request-id", "request-id"}}); + std::unique_ptr file_system = CreateFileSystem(transport); + + ASSERT_OK_AND_ASSIGN(FileStatus status, file_system->GetFileStatus("oss://bucket/key")); + ASSERT_EQ(3, status.GetLen()); + ASSERT_EQ(FileStatus::kUnknownModificationTime, status.GetModificationTime()); + ASSERT_EQ(1U, transport->requests_.size()); + ASSERT_EQ("HEAD", transport->requests_[0]->method); +} + +TEST(OssFileSystemTest, TestHeadObjectNotFound) { + std::shared_ptr transport = std::make_shared(); + transport->AddResponse(404, {{"x-oss-request-id", "request-id"}}, + "NoSuchKeymissing"); + transport->AddResponse(200, {}, + "false"); + std::unique_ptr file_system = CreateFileSystem(transport); + + Result status = file_system->GetFileStatus("oss://bucket/missing"); + ASSERT_TRUE(status.status().IsNotExist()) << status.status().ToString(); + ASSERT_NOK_WITH_MSG(status, "does not exist"); +} + +TEST(OssFileSystemTest, TestHeadObjectErrorMapping) { + std::shared_ptr transport = std::make_shared(); + transport->AddResponse(403, {{"x-oss-request-id", "request-id"}}, + "AccessDenieddenied"); + std::unique_ptr file_system = CreateFileSystem(transport); + + Result status = file_system->GetFileStatus("oss://bucket/key"); + ASSERT_TRUE(status.status().IsIOError()) << status.status().ToString(); + ASSERT_NOK_WITH_MSG(status, "code=AccessDenied, status=403"); +} + +TEST(OssFileSystemTest, TestListObjects) { + std::shared_ptr transport = std::make_shared(); + transport->AddResponse(404, {{"x-oss-request-id", "request-id"}}, + "NoSuchKeymissing"); + transport->AddResponse(200, {{"x-oss-request-id", "request-id"}}, R"( + + false + + prefix/file + 3 + 2024-01-01T00:00:00.000Z + + prefix/sub/ +)"); + std::unique_ptr file_system = CreateFileSystem(transport); + std::vector statuses; + + ASSERT_OK(file_system->ListFileStatus("oss://bucket/prefix", &statuses)); + ASSERT_EQ(2U, statuses.size()); + ASSERT_EQ(1704067200000, statuses[0].GetModificationTime()); + ASSERT_EQ(2U, transport->requests_.size()); + ASSERT_EQ("HEAD", transport->requests_[0]->method); + ASSERT_EQ("GET", transport->requests_[1]->method); + ASSERT_NE(std::string::npos, transport->requests_[1]->uri.find("list-type=2")); +} + +TEST(OssFileSystemTest, TestListObjectsErrorMapping) { + std::shared_ptr transport = std::make_shared(); + transport->AddResponse(403, {{"x-oss-request-id", "request-id"}}, + "AccessDenieddenied"); + std::unique_ptr file_system = CreateFileSystem(transport); + std::vector statuses; + + Status status = file_system->ListDir("oss://bucket/prefix/", &statuses); + ASSERT_TRUE(status.IsIOError()) << status.ToString(); + ASSERT_NOK_WITH_MSG(status, "code=AccessDenied, status=403"); +} + +TEST(OssFileSystemTest, TestGetObjectRangeAndShortRead) { + std::shared_ptr transport = std::make_shared(); + transport->AddResponse(206, {{"Content-Length", "3"}}, "abc"); + std::unique_ptr file_system = CreateFileSystem(transport); + FileStatus file_status("oss://bucket/key", 3); + ASSERT_OK_AND_ASSIGN(std::unique_ptr stream, file_system->Open(file_status)); + char buffer[3]; + + ASSERT_OK_AND_ASSIGN(int64_t bytes_read, stream->Read(buffer, 3, 0)); + ASSERT_EQ(3, bytes_read); + ASSERT_EQ("abc", std::string(buffer, sizeof(buffer))); + ASSERT_EQ(1U, transport->requests_.size()); + ASSERT_EQ("bytes=0-2", transport->requests_[0]->headers.at("range")); + + std::shared_ptr short_transport = std::make_shared(); + short_transport->AddResponse(206, {{"Content-Length", "2"}}, "ab"); + std::unique_ptr short_file_system = CreateFileSystem(short_transport); + ASSERT_OK_AND_ASSIGN(std::unique_ptr short_stream, + short_file_system->Open(file_status)); + ASSERT_NOK_WITH_MSG(short_stream->Read(buffer, 3, 0), "expected 3"); +} + +TEST(OssFileSystemTest, TestGetObjectRangeErrorMapping) { + std::shared_ptr transport = std::make_shared(); + transport->AddResponse(403, {{"x-oss-request-id", "request-id"}}, + "AccessDenieddenied"); + std::unique_ptr file_system = CreateFileSystem(transport); + FileStatus file_status("oss://bucket/key", 3); + ASSERT_OK_AND_ASSIGN(std::unique_ptr stream, file_system->Open(file_status)); + char buffer[3]; + + Result result = stream->Read(buffer, 3, 0); + ASSERT_TRUE(result.status().IsIOError()) << result.status().ToString(); + ASSERT_NOK_WITH_MSG(result, "code=AccessDenied, status=403"); +} + +TEST(OssFileSystemTest, TestGetObjectRangeAsync) { + std::shared_ptr transport = std::make_shared(); + transport->AddResponse(206, {{"Content-Length", "3"}}, "abc"); + std::unique_ptr file_system = CreateFileSystem(transport); + FileStatus file_status("oss://bucket/key", 3); + ASSERT_OK_AND_ASSIGN(std::unique_ptr stream, file_system->Open(file_status)); + char buffer[3]; + std::promise promise; + std::future future = promise.get_future(); + + stream->ReadAsync(buffer, 3, 0, + [&promise](Status status) { promise.set_value(std::move(status)); }); + + ASSERT_EQ(std::future_status::ready, future.wait_for(std::chrono::seconds(5))); + ASSERT_OK(future.get()); + ASSERT_EQ("abc", std::string(buffer, sizeof(buffer))); + ASSERT_EQ(1U, transport->requests_.size()); + ASSERT_EQ("bytes=0-2", transport->requests_[0]->headers.at("range")); +} + +} // namespace paimon::oss diff --git a/src/paimon/fs/s3/s3_file_system.cpp b/src/paimon/fs/s3/s3_file_system.cpp index 49668b70..82b05620 100644 --- a/src/paimon/fs/s3/s3_file_system.cpp +++ b/src/paimon/fs/s3/s3_file_system.cpp @@ -71,11 +71,6 @@ Result ParseNonNegativeInt64(const std::string& value, const std::strin return *result; } -int64_t ParseModificationTime(const std::string& value) { - time_t seconds = curl_getdate(value.c_str(), nullptr); - return seconds == static_cast(-1) ? 0 : static_cast(seconds) * 1000; -} - std::string XmlUnescape(const std::string& value) { const std::pair entities[] = { {"&", "&"}, {"<", "<"}, {">", ">"}, {""", "\""}, {"'", "'"}}; @@ -662,10 +657,10 @@ class S3ObjectStoreClient : public ObjectStoreClient, if (length == response.headers.end()) { return Status::IOError("HeadObject response is missing Content-Length"); } - int64_t modification_time = 0; + int64_t modification_time = FileStatus::kUnknownModificationTime; auto modified = response.headers.find("last-modified"); if (modified != response.headers.end()) { - modification_time = ParseModificationTime(modified->second); + modification_time = ObjectStoreFileSystemUtils::ParseModificationTime(modified->second); } PAIMON_ASSIGN_OR_RAISE(int64_t object_size, ParseNonNegativeInt64(length->second, "Content-Length")); @@ -712,10 +707,10 @@ class S3ObjectStoreClient : public ObjectStoreClient, PAIMON_ASSIGN_OR_RAISE(std::string decoded_key, PercentDecode(*key, "Key")); PAIMON_ASSIGN_OR_RAISE(int64_t object_size, ParseNonNegativeInt64(*size, "ListObjectsV2 Size")); - int64_t modified = 0; + int64_t modified = FileStatus::kUnknownModificationTime; auto last_modified = TagValue(block, "LastModified"); if (last_modified) { - modified = ParseModificationTime(*last_modified); + modified = ObjectStoreFileSystemUtils::ParseModificationTime(*last_modified); } result.objects.push_back(ObjectMetadata{decoded_key, object_size, modified}); } diff --git a/src/paimon/fs/s3/s3_file_system_test.cpp b/src/paimon/fs/s3/s3_file_system_test.cpp index 09b1bb4b..3b3d1dfd 100644 --- a/src/paimon/fs/s3/s3_file_system_test.cpp +++ b/src/paimon/fs/s3/s3_file_system_test.cpp @@ -292,13 +292,13 @@ TEST(S3ObjectStoreClientTest, TestInvalidModificationTime) { ASSERT_OK_AND_ASSIGN(std::shared_ptr client, MakeS3ObjectStoreClient(StaticOptions(), http)); ASSERT_OK_AND_ASSIGN(auto metadata, client->HeadObject({"bucket", "file"})); - ASSERT_EQ(metadata.modification_time, 0); + ASSERT_EQ(metadata.modification_time, FileStatus::kUnknownModificationTime); http->body_ = "falsefile" "invalid12"; ASSERT_OK_AND_ASSIGN(auto result, client->ListObjects({"bucket", ""}, "", 0)); - ASSERT_EQ(result.objects[0].modification_time, 0); + ASSERT_EQ(result.objects[0].modification_time, FileStatus::kUnknownModificationTime); } TEST(S3ObjectStoreClientTest, TestRegionFromEnvironment) { @@ -428,7 +428,9 @@ TEST(S3ObjectStoreClientTest, TestRangeAndListObjects) { ASSERT_TRUE(result.is_truncated); ASSERT_EQ(result.continuation_token, "next token"); ASSERT_EQ(result.objects[0].key, "dir/a&b"); + ASSERT_EQ(result.objects[0].modification_time, 1767225600000); ASSERT_EQ(result.objects[1].key, "dir/a<b"); + ASSERT_EQ(result.objects[1].modification_time, FileStatus::kUnknownModificationTime); ASSERT_EQ(result.common_prefixes[0], "dir/sub/"); ASSERT_NE(http->request_.url.find("amazonaws.com/?list-type=2"), std::string::npos); ASSERT_NE(http->request_.url.find("encoding-type=url"), std::string::npos); diff --git a/third_party/versions.txt b/third_party/versions.txt index 2e20dc72..e1a9aff5 100644 --- a/third_party/versions.txt +++ b/third_party/versions.txt @@ -97,6 +97,10 @@ PAIMON_AWS_S2N_BUILD_VERSION=v1.7.4 PAIMON_AWS_S2N_BUILD_SHA256_CHECKSUM=af5ce0783fd9e05ed1899fda0c76e02fa5dd92128018e5bfe71634312d2ce8e7 PAIMON_AWS_S2N_PKG_NAME=s2n-${PAIMON_AWS_S2N_BUILD_VERSION}.zip +PAIMON_OSS_SDK_V2_BUILD_VERSION=0.1.2 +PAIMON_OSS_SDK_V2_BUILD_SHA256_CHECKSUM=6a6c00692c8fe8461594a359de8ba97f4468d90f92ffb6bcb92b82a9d1b9311b +PAIMON_OSS_SDK_V2_PKG_NAME=alibabacloud-oss-cpp-sdk-v2-${PAIMON_OSS_SDK_V2_BUILD_VERSION}.tar.gz + PAIMON_FMT_BUILD_VERSION=11.2.0 PAIMON_FMT_BUILD_SHA256_CHECKSUM=bc23066d87ab3168f27cef3e97d545fa63314f5c79df5ea444d41d56f962c6af PAIMON_FMT_PKG_NAME=fmt-${PAIMON_FMT_BUILD_VERSION}.tar.gz @@ -158,6 +162,7 @@ DEPENDENCIES=( "PAIMON_LZ4_URL ${PAIMON_LZ4_PKG_NAME} ${THIRDPARTY_MIRROR_URL}https://github.com/lz4/lz4/archive/${PAIMON_LZ4_BUILD_VERSION}.tar.gz" "PAIMON_PROTOBUF_URL ${PAIMON_PROTOBUF_PKG_NAME} ${THIRDPARTY_MIRROR_URL}https://github.com/protocolbuffers/protobuf/releases/download/v${PAIMON_PROTOBUF_BUILD_VERSION}/protobuf-all-${PAIMON_PROTOBUF_BUILD_VERSION}.tar.gz" "PAIMON_TBB_URL ${PAIMON_TBB_PKG_NAME} ${THIRDPARTY_MIRROR_URL}https://github.com/uxlfoundation/oneTBB/archive/refs/tags/${PAIMON_TBB_BUILD_VERSION}.tar.gz" + "PAIMON_OSS_SDK_V2_URL ${PAIMON_OSS_SDK_V2_PKG_NAME} ${THIRDPARTY_MIRROR_URL}https://github.com/aliyun/alibabacloud-oss-cpp-sdk-v2/archive/refs/tags/${PAIMON_OSS_SDK_V2_BUILD_VERSION}.tar.gz" "PAIMON_ORC_URL ${PAIMON_ORC_PKG_NAME} ${THIRDPARTY_MIRROR_URL}https://github.com/apache/orc/archive/refs/tags/${PAIMON_ORC_BUILD_VERSION}.tar.gz" "PAIMON_GTEST_URL ${PAIMON_GTEST_PKG_NAME} ${THIRDPARTY_MIRROR_URL}https://github.com/google/googletest/archive/release-${PAIMON_GTEST_BUILD_VERSION}.tar.gz" "PAIMON_BENCHMARK_URL ${PAIMON_BENCHMARK_PKG_NAME} ${THIRDPARTY_MIRROR_URL}https://github.com/google/benchmark/archive/refs/tags/v${PAIMON_BENCHMARK_BUILD_VERSION}.tar.gz"