This is an automated email from the ASF dual-hosted git repository.
lxy-9602 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new 19cba8d1 feat(fs): add OSS file system (#205)
19cba8d1 is described below
commit 19cba8d1212bcb6f5b5d9cd605fe893c42e8a6a6
Author: Mr Dk. <[email protected]>
AuthorDate: Sat Aug 29 15:41:15 2026 +0800
feat(fs): add OSS file system (#205)
---
CMakeLists.txt | 18 +-
README.md | 2 +-
ci/scripts/build_paimon.sh | 1 +
cmake_modules/DefineOptions.cmake | 5 +
cmake_modules/ThirdpartyToolchain.cmake | 119 ++++++++
docs/source/build_system.rst | 2 +
docs/source/building.rst | 2 +
src/paimon/CMakeLists.txt | 4 +-
src/paimon/common/fs/object_store_file_system.cpp | 1 +
src/paimon/common/fs/object_store_file_system.h | 72 ++++-
.../common/fs/object_store_file_system_test.cpp | 11 +
src/paimon/fs/oss/CMakeLists.txt | 50 ++++
src/paimon/fs/oss/oss_file_system.cpp | 192 ++++++++++++
src/paimon/fs/oss/oss_file_system.h | 51 ++++
src/paimon/fs/oss/oss_file_system_factory.cpp | 234 +++++++++++++++
src/paimon/fs/oss/oss_file_system_factory.h | 38 +++
src/paimon/fs/oss/oss_file_system_test.cpp | 327 +++++++++++++++++++++
src/paimon/fs/s3/s3_file_system.cpp | 13 +-
src/paimon/fs/s3/s3_file_system_test.cpp | 6 +-
third_party/versions.txt | 5 +
20 files changed, 1136 insertions(+), 17 deletions(-)
diff --git a/CMakeLists.txt b/CMakeLists.txt
index 46a55426..2fe2df8e 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -62,6 +62,7 @@ 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_MOSAIC "Whether to enable mosaic file format (Rust FFI)"
OFF)
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)
@@ -77,8 +78,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_REST)
@@ -93,6 +96,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()
@@ -473,6 +479,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)
@@ -532,6 +545,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 58ed9b0a..cea59da8 100755
--- a/ci/scripts/build_paimon.sh
+++ b/ci/scripts/build_paimon.sh
@@ -149,6 +149,7 @@ CMAKE_ARGS=(
"-DPAIMON_BUILD_TESTS=ON"
"-DPAIMON_ENABLE_MOSAIC=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 a35ecd49..0c64819c 100644
--- a/cmake_modules/ThirdpartyToolchain.cmake
+++ b/cmake_modules/ThirdpartyToolchain.cmake
@@ -320,6 +320,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}")
@@ -504,6 +514,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)
@@ -609,6 +638,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()
@@ -683,6 +714,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")
@@ -1415,6 +1448,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")
@@ -2026,6 +2142,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 32c7e67c..a057ff53 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 3de2b667..0ec23a0a 100644
--- a/src/paimon/CMakeLists.txt
+++ b/src/paimon/CMakeLists.txt
@@ -201,7 +201,7 @@ set(PAIMON_REST_LINK_LIBS)
if(PAIMON_ENABLE_REST)
set(PAIMON_REST_LINK_LIBS OpenSSL::Crypto)
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()
@@ -948,7 +948,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 <curl/curl.h>
+
+#include <cctype>
+#include <cerrno>
#include <cstdint>
+#include <ctime>
#include <functional>
#include <memory>
#include <mutex>
@@ -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<unsigned char>(*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<unsigned char>(current[1])) &&
+ std::isdigit(static_cast<unsigned char>(current[2])) &&
current[3] == ':' &&
+ std::isdigit(static_cast<unsigned char>(current[4])) &&
+ std::isdigit(static_cast<unsigned char>(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<time_t>(-1) || errno != EOVERFLOW) {
+ int64_t utc_seconds = static_cast<int64_t>(seconds) -
timezone_offset_seconds;
+ return utc_seconds * 1000 + milliseconds;
+ }
+ }
+ }
+
+ time_t seconds = curl_getdate(value.c_str(), nullptr);
+ if (seconds == static_cast<time_t>(-1)) {
+ return FileStatus::kUnknownModificationTime;
+ }
+ return static_cast<int64_t>(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<int64_t, int64_t>;
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 <algorithm>
+#include <cstdint>
+#include <memory>
+#include <utility>
+
+#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<OssObjectStoreClient> {
+ public:
+ OssObjectStoreClient(std::string bucket, std::shared_ptr<oss2::OSSClient>
client,
+ std::unique_ptr<Executor> executor)
+ : bucket_(std::move(bucket)), client_(std::move(client)),
executor_(std::move(executor)) {}
+
+ Result<ObjectMetadata> 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<ListObjectsResult> 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<int64_t> 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<std::shared_ptr<oss2::MemoryWriter>>();
+ oss2::SinkFactory sink;
+ sink.isOneShot = false;
+ sink.supplier = [buffer, size, writer](int64_t, const
oss2::HeaderCollection&) {
+ auto memory_writer = std::make_shared<oss2::MemoryWriter>(
+ reinterpret_cast<uint8_t*>(buffer), static_cast<size_t>(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<int64_t>((*writer)->written()) :
static_cast<int64_t>(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<void(Status)>&&
callback) const override {
+ std::shared_ptr<const OssObjectStoreClient> self = shared_from_this();
+ executor_->Add([self = std::move(self), path, offset, size, buffer,
+ callback = std::move(callback)]() mutable {
+ Result<int64_t> 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<oss2::OSSClient> client_;
+ std::unique_ptr<Executor> executor_;
+};
+
+} // namespace
+
+OssFileSystem::OssFileSystem(std::string bucket,
std::shared_ptr<oss2::OSSClient> client,
+ std::unique_ptr<Executor> executor)
+ : ObjectStoreFileSystem("oss", std::make_shared<OssObjectStoreClient>(
+ 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 <memory>
+#include <string>
+
+#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<alibabacloud::oss2::OSSClient> client,
+ std::unique_ptr<Executor> 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 <cctype>
+#include <memory>
+#include <optional>
+#include <string>
+#include <string_view>
+#include <utility>
+
+#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<unsigned char>(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<unsigned char>(value)) ||
+ std::isdigit(static_cast<unsigned char>(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<std::string, std::string>&
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<std::string, std::string>& 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<std::string> GetRequiredOption(const std::map<std::string,
std::string>& 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<std::unique_ptr<Executor>> CreateExecutor(const std::map<std::string,
std::string>& 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<uint32_t> thread_count =
StringUtils::StringToValue<uint32_t>(*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<uint16_t> port =
+
StringUtils::StringToValue<uint16_t>(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<std::unique_ptr<FileSystem>> OssFileSystemFactory::Create(
+ const std::string& path, const std::map<std::string, std::string>&
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<oss2::StaticCredentialsProvider>(
+ 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<bool> value =
StringUtils::StringToValue<bool>(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> executor,
CreateExecutor(options, bucket));
+ return std::make_unique<OssFileSystem>(bucket,
std::make_shared<oss2::OSSClient>(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<std::unique_ptr<FileSystem>> Create(
+ const std::string& path, const std::map<std::string, std::string>&
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 <chrono>
+#include <cstdint>
+#include <future>
+#include <map>
+#include <memory>
+#include <sstream>
+#include <string>
+#include <system_error>
+#include <utility>
+#include <vector>
+
+#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<oss2::RequestMessage>& request,
+ const oss2::RequestOptions& options) override {
+
requests_.emplace_back(std::make_unique<oss2::RequestMessage>(*request));
+ if (responses_.empty()) {
+ return
oss2::TransportError{std::make_error_code(std::errc::no_message_available), "",
+ ""};
+ }
+ std::unique_ptr<oss2::ResponseMessage> 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<oss2::ByteWriter> 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<const uint8_t*>(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<std::iostream> response_body;
+ if (!body.empty()) {
+ response_body =
std::make_shared<std::stringstream>(std::move(body));
+ }
+
responses_.emplace_back(std::make_unique<oss2::ResponseMessage>(oss2::ResponseMessage{
+ status_code, "", std::move(headers), std::move(response_body),
nullptr}));
+ }
+
+ std::vector<std::unique_ptr<oss2::ResponseMessage>> responses_;
+ std::vector<std::unique_ptr<oss2::RequestMessage>> requests_;
+};
+
+std::unique_ptr<OssFileSystem> CreateFileSystem(
+ const std::shared_ptr<MockHttpTransport>& transport) {
+ oss2::ClientConfiguration config =
oss2::ClientConfiguration::loadDefault();
+ config.region = "cn-hangzhou";
+ config.credentialsProvider =
+ std::make_shared<oss2::StaticCredentialsProvider>("access-key",
"secret-key");
+ config.httpTransport = transport;
+ return std::make_unique<OssFileSystem>("bucket",
std::make_shared<oss2::OSSClient>(config),
+ CreateDefaultExecutor());
+}
+
+} // namespace
+
+TEST(OssFileSystemFactoryTest, TestOptionValidation) {
+ OssFileSystemFactory factory;
+ std::map<std::string, std::string> 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<std::string, std::string> 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<std::string, std::string> 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<std::string, std::string> 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<MockHttpTransport> transport =
std::make_shared<MockHttpTransport>();
+ transport->AddResponse(200, {{"Content-Length", "3"},
+ {"Last-Modified", "not-a-timestamp"},
+ {"x-oss-request-id", "request-id"}});
+ std::unique_ptr<OssFileSystem> 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<MockHttpTransport> transport =
std::make_shared<MockHttpTransport>();
+ transport->AddResponse(404, {{"x-oss-request-id", "request-id"}},
+
"<Error><Code>NoSuchKey</Code><Message>missing</Message></Error>");
+ transport->AddResponse(200, {},
+
"<ListBucketResult><IsTruncated>false</IsTruncated></ListBucketResult>");
+ std::unique_ptr<OssFileSystem> file_system = CreateFileSystem(transport);
+
+ Result<FileStatus> 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<MockHttpTransport> transport =
std::make_shared<MockHttpTransport>();
+ transport->AddResponse(403, {{"x-oss-request-id", "request-id"}},
+
"<Error><Code>AccessDenied</Code><Message>denied</Message></Error>");
+ std::unique_ptr<OssFileSystem> file_system = CreateFileSystem(transport);
+
+ Result<FileStatus> 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<MockHttpTransport> transport =
std::make_shared<MockHttpTransport>();
+ transport->AddResponse(404, {{"x-oss-request-id", "request-id"}},
+
"<Error><Code>NoSuchKey</Code><Message>missing</Message></Error>");
+ transport->AddResponse(200, {{"x-oss-request-id", "request-id"}}, R"(
+<ListBucketResult>
+ <IsTruncated>false</IsTruncated>
+ <Contents>
+ <Key>prefix/file</Key>
+ <Size>3</Size>
+ <LastModified>2024-01-01T00:00:00.000Z</LastModified>
+ </Contents>
+ <CommonPrefixes><Prefix>prefix/sub/</Prefix></CommonPrefixes>
+</ListBucketResult>)");
+ std::unique_ptr<OssFileSystem> file_system = CreateFileSystem(transport);
+ std::vector<FileStatus> 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<MockHttpTransport> transport =
std::make_shared<MockHttpTransport>();
+ transport->AddResponse(403, {{"x-oss-request-id", "request-id"}},
+
"<Error><Code>AccessDenied</Code><Message>denied</Message></Error>");
+ std::unique_ptr<OssFileSystem> file_system = CreateFileSystem(transport);
+ std::vector<BasicFileStatus> 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<MockHttpTransport> transport =
std::make_shared<MockHttpTransport>();
+ transport->AddResponse(206, {{"Content-Length", "3"}}, "abc");
+ std::unique_ptr<OssFileSystem> file_system = CreateFileSystem(transport);
+ FileStatus file_status("oss://bucket/key", 3);
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<InputStream> 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<MockHttpTransport> short_transport =
std::make_shared<MockHttpTransport>();
+ short_transport->AddResponse(206, {{"Content-Length", "2"}}, "ab");
+ std::unique_ptr<OssFileSystem> short_file_system =
CreateFileSystem(short_transport);
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<InputStream> 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<MockHttpTransport> transport =
std::make_shared<MockHttpTransport>();
+ transport->AddResponse(403, {{"x-oss-request-id", "request-id"}},
+
"<Error><Code>AccessDenied</Code><Message>denied</Message></Error>");
+ std::unique_ptr<OssFileSystem> file_system = CreateFileSystem(transport);
+ FileStatus file_status("oss://bucket/key", 3);
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<InputStream> stream,
file_system->Open(file_status));
+ char buffer[3];
+
+ Result<int64_t> 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<MockHttpTransport> transport =
std::make_shared<MockHttpTransport>();
+ transport->AddResponse(206, {{"Content-Length", "3"}}, "abc");
+ std::unique_ptr<OssFileSystem> file_system = CreateFileSystem(transport);
+ FileStatus file_status("oss://bucket/key", 3);
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<InputStream> stream,
file_system->Open(file_status));
+ char buffer[3];
+ std::promise<Status> promise;
+ std::future<Status> 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 e31cc9ca..b9b98f1e 100644
--- a/src/paimon/fs/s3/s3_file_system.cpp
+++ b/src/paimon/fs/s3/s3_file_system.cpp
@@ -71,11 +71,6 @@ Result<int64_t> 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<time_t>(-1) ? 0 :
static_cast<int64_t>(seconds) * 1000;
-}
-
std::string XmlUnescape(const std::string& value) {
const std::pair<const char*, const char*> 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 063a7dcb..822e1aed 100644
--- a/src/paimon/fs/s3/s3_file_system_test.cpp
+++ b/src/paimon/fs/s3/s3_file_system_test.cpp
@@ -302,13 +302,13 @@ TEST(S3ObjectStoreClientTest,
TestInvalidModificationTime) {
ASSERT_OK_AND_ASSIGN(std::shared_ptr<ObjectStoreClient> 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_ =
"<ListBucketResult><IsTruncated>false</IsTruncated><Contents><Key>file</Key>"
"<LastModified>invalid</LastModified><Size>12</Size></Contents></ListBucketResult>";
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) {
@@ -438,7 +438,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 ffb1bbf3..1afdf7a1 100644
--- a/third_party/versions.txt
+++ b/third_party/versions.txt
@@ -101,6 +101,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
@@ -162,6 +166,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";
